diff --git a/opsqueue/migrations/20260710113337_drop_chunks_metadata.sql b/opsqueue/migrations/20260710113337_drop_chunks_metadata.sql new file mode 100644 index 0000000..658a440 --- /dev/null +++ b/opsqueue/migrations/20260710113337_drop_chunks_metadata.sql @@ -0,0 +1 @@ +DROP TABLE chunks_metadata; diff --git a/opsqueue/src/common/chunk.rs b/opsqueue/src/common/chunk.rs index 1106f4e..b5b8441 100644 --- a/opsqueue/src/common/chunk.rs +++ b/opsqueue/src/common/chunk.rs @@ -219,7 +219,6 @@ impl Chunk { pub mod db { use super::*; use crate::common::errors::{ChunkNotFound, DatabaseError, SubmissionNotFound, E}; - use crate::common::StrategicMetadataMap; use crate::db::{Connection, True, WriterConnection}; use axum_prometheus::metrics::{counter, gauge}; use sqlx::{query, query_as}; @@ -277,33 +276,6 @@ pub mod db { Ok(()) } - #[tracing::instrument(skip(conn))] - pub async fn insert_chunk_metadata( - chunk: Chunk, - metadata_key: &[u8], - metadata_value: &[u8], - mut conn: impl WriterConnection, - ) -> sqlx::Result<()> { - query!( - " - INSERT INTO chunks_metadata - ( submission_id - , chunk_index - , metadata_key - , metadata_value - ) - VALUES ($1, $2, $3, $4) - ", - chunk.submission_id, - chunk.chunk_index, - metadata_key, - metadata_value, - ) - .execute(conn.get_inner()) - .await?; - Ok(()) - } - #[tracing::instrument(skip(conn))] pub async fn complete_chunk( chunk_id: ChunkId, @@ -530,33 +502,6 @@ pub mod db { .await } - /// Retrieves the earlier stored strategic metadata. - /// - /// Primarily for testing and introspection. - /// - /// Be aware that the strategic metadata for individual chunks - /// is cleaned up once the chunk is marked as completed or failed. - /// (At that time it is still available on the submission level). - pub async fn get_chunk_strategic_metadata( - full_chunk_id: ChunkId, - mut conn: impl Connection, - ) -> Result { - use futures::{future, TryStreamExt}; - let metadata = query!( - r#" - SELECT metadata_key, metadata_value FROM chunks_metadata - WHERE submission_id = $1 AND chunk_index = $2 - "#, - full_chunk_id.submission_id, - full_chunk_id.chunk_index, - ) - .fetch(conn.get_inner()) - .and_then(|row| future::ok((row.metadata_key, row.metadata_value))) - .try_collect() - .await?; - Ok(metadata) - } - #[tracing::instrument(skip(chunks, conn))] pub async fn insert_many_chunks( chunks: &[Chunk], @@ -584,40 +529,6 @@ pub mod db { Ok(()) } - pub async fn insert_many_chunks_metadata( - chunks: &[Chunk], - metadata: &StrategicMetadataMap, - mut conn: impl WriterConnection, - ) -> sqlx::Result<()> { - use itertools::Itertools; - const ROWS_PER_QUERY: usize = 1000; - - let mut iter = chunks.iter().cartesian_product(metadata).peekable(); - while iter.peek().is_some() { - let query_rows = iter.by_ref().take(ROWS_PER_QUERY); - - let mut query_builder: QueryBuilder = QueryBuilder::new( - " - INSERT INTO chunks_metadata - ( submission_id - , chunk_index - , metadata_key - , metadata_value - ) - ", - ); - query_builder.push_values(query_rows, |mut b, (chunk, metadata)| { - b.push_bind(chunk.submission_id) - .push_bind(chunk.chunk_index) - .push_bind(metadata.0) - .push_bind(metadata.1); - }); - let query = query_builder.build(); - query.execute(conn.get_inner()).await?; - } - Ok(()) - } - #[tracing::instrument(skip(conn))] pub async fn skip_remaining_chunks( submission_id: SubmissionId, diff --git a/opsqueue/src/common/submission.rs b/opsqueue/src/common/submission.rs index e5998f6..5355f64 100644 --- a/opsqueue/src/common/submission.rs +++ b/opsqueue/src/common/submission.rs @@ -392,12 +392,6 @@ pub mod db { ) .await?; super::chunk::db::insert_many_chunks(&chunks, &mut tx).await?; - super::chunk::db::insert_many_chunks_metadata( - &chunks, - &submission.strategic_metadata, - &mut tx, - ) - .await?; Ok(()) } .boxed() @@ -1255,24 +1249,7 @@ pub mod test { let fetched_metadata = get_submission_strategic_metadata(submission_id, &mut conn) .await .unwrap(); - assert!(fetched_metadata == strategic_metadata); - - let res = sqlx::query!("SELECT * FROM chunks_metadata;") - .fetch_all(conn.get_inner()) - .await - .unwrap(); - dbg!(res); - - let chunk_id = (submission_id, ChunkIndex::zero()).into(); - let chunk_fetched_metadata = chunk::db::get_chunk_strategic_metadata(chunk_id, &mut conn) - .await - .unwrap(); - - dbg!(&strategic_metadata); - dbg!(&fetched_metadata); - dbg!(&chunk_fetched_metadata); - - assert_eq!(chunk_fetched_metadata, strategic_metadata); + assert_eq!(fetched_metadata, strategic_metadata); } #[sqlx::test]