add file size metadata to integration test (delta-io#1951)

## What changes are proposed in this pull request?
- Part of delta-io#1010 
- There are several test in `kernel/test/read.rs` and
`kernel/test/log_segment.rs` that read the Parquet file, however it is
using file_meta's constant value. In the next PR, when we used the
file_meta's size in the read the Parquet file, those tests will fail
since there is a mismatch in the correct parquet size and the
file_meta's size. In this PR, we will integrate the correct file_meta's
size to those affected test.
- Change included:
- Create a new mock Enum for add/remove commit with size called
`AddWithSize`, `RemoveWithSize`.
- Change the affected test in the `kernel/test/read.rs` to use the
`AddWithSize`, `RemoveWithSize`.
- Update the log_segment's test to write sidecar's file size in the
checkpoint.

## How was this change tested?

`cargo test --package delta_kernel`

Author: thinh2

kernel/src/log_segment/tests.rs

@@ -159,15 +159,22 @@ pub(crate) async fn add_checkpoint_to_store(
     write_parquet_to_store(store, path, data).await
 }
 
-/// Writes all actions to a _delta_log/_sidecars file in the store.
+/// Writes all actions to a _delta_log/_sidecars file in the store and return the [`FileMeta`].
 /// This function formats the provided filename into the _sidecars subdirectory.
 async fn add_sidecar_to_store(
     store: &Arc<InMemory>,
     data: Box<dyn EngineData>,
     filename: &str,
-) -> DeltaResult<()> {
+) -> DeltaResult<FileMeta> {
     let path = format!("_delta_log/_sidecars/{filename}");
-    write_parquet_to_store(store, path, data).await
+    write_parquet_to_store(store, path.clone(), data).await?;
+    let size = get_file_size(store, &path).await;
+    let location = Url::parse(&format!("memory:///{path}")).expect("valid url");
+    Ok(FileMeta {
+        location,
+        last_modified: 0,
+        size,
+    })
 }
 
 /// Writes all actions to a _delta_log json checkpoint file in the store.
@@ -989,21 +996,27 @@ async fn test_checkpoint_batch_with_sidecars_returns_sidecar_batches() -> DeltaR
     let engine = DefaultEngineBuilder::new(store.clone()).build();
     let read_schema = get_all_actions_schema().project(&[ADD_NAME, REMOVE_NAME, SIDECAR_NAME])?;
 
-    add_sidecar_to_store(
+    let sidecar1_size = add_sidecar_to_store(
         &store,
         add_batch_simple(read_schema.clone()),
         "sidecarfile1.parquet",
     )
-    .await?;
-    add_sidecar_to_store(
+    .await?
+    .size;
+
+    let sidecar2_size = add_sidecar_to_store(
         &store,
         add_batch_with_remove(read_schema.clone()),
         "sidecarfile2.parquet",
     )
-    .await?;
+    .await?
+    .size;
 
-    let checkpoint_batch = sidecar_batch_with_given_paths(
-        vec!["sidecarfile1.parquet", "sidecarfile2.parquet"],
+    let checkpoint_batch = sidecar_batch_with_given_paths_and_sizes(
+        vec![
+            ("sidecarfile1.parquet", sidecar1_size),
+            ("sidecarfile2.parquet", sidecar2_size),
+        ],
         read_schema.clone(),
     );
 
@@ -1058,16 +1071,19 @@ async fn test_reading_sidecar_files_with_predicate() -> DeltaResult<()> {
     let engine = DefaultEngineBuilder::new(store.clone()).build();
     let read_schema = get_all_actions_schema().project(&[ADD_NAME, REMOVE_NAME, SIDECAR_NAME])?;
 
-    let checkpoint_batch =
-        sidecar_batch_with_given_paths(vec!["sidecarfile1.parquet"], read_schema.clone());
-
     // Add a sidecar file with only add actions
-    add_sidecar_to_store(
+    let sidecar_size = add_sidecar_to_store(
         &store,
         add_batch_simple(read_schema.clone()),
         "sidecarfile1.parquet",
     )
-    .await?;
+    .await?
+    .size;
+
+    let checkpoint_batch = sidecar_batch_with_given_paths_and_sizes(
+        vec![("sidecarfile1.parquet", sidecar_size)],
+        read_schema.clone(),
+    );
 
     // Filter out sidecar files that do not contain remove actions
     let remove_predicate: LazyLock<Option<PredicateRef>> = LazyLock::new(|| {
@@ -1344,22 +1360,21 @@ async fn test_create_checkpoint_stream_reads_checkpoint_file_and_returns_sidecar
     let engine = DefaultEngineBuilder::new(store.clone()).build();
 
     // Write sidecars first so we can get their actual sizes
-    add_sidecar_to_store(
+    let sidecar1_size = add_sidecar_to_store(
         &store,
         add_batch_simple(get_commit_schema().project(&[ADD_NAME, REMOVE_NAME])?),
         "sidecarfile1.parquet",
     )
-    .await?;
-    add_sidecar_to_store(
+    .await?
+    .size;
+
+    let sidecar2_size = add_sidecar_to_store(
         &store,
         add_batch_with_remove(get_commit_schema().project(&[ADD_NAME, REMOVE_NAME])?),
         "sidecarfile2.parquet",
     )
-    .await?;
-
-    // Get actual sidecar sizes for correct FileMeta creation
-    let sidecar1_size = get_file_size(&store, "_delta_log/_sidecars/sidecarfile1.parquet").await;
-    let sidecar2_size = get_file_size(&store, "_delta_log/_sidecars/sidecarfile2.parquet").await;
+    .await?
+    .size;
 
     // Now create checkpoint with correct sidecar sizes
     add_checkpoint_to_store(

kernel/tests/read.rs

@@ -52,13 +52,15 @@ fn make_top_level_fields_nullable(batch: &RecordBatch) -> RecordBatch {
 async fn single_commit_two_add_files() -> Result<(), Box<dyn std::error::Error>> {
     let batch = generate_simple_batch()?;
     let storage = Arc::new(InMemory::new());
+    let parquet_bytes = record_batch_to_bytes(&batch);
+    let file_size = parquet_bytes.len() as u64;
     add_commit(
         storage.as_ref(),
         0,
         actions_to_string(vec![
             TestAction::Metadata,
-            TestAction::Add(PARQUET_FILE1.to_string()),
-            TestAction::Add(PARQUET_FILE2.to_string()),
+            TestAction::AddWithSize(PARQUET_FILE1.to_string(), file_size),
+            TestAction::AddWithSize(PARQUET_FILE2.to_string(), file_size),
         ]),
     )
     .await?;
@@ -99,19 +101,24 @@ async fn single_commit_two_add_files() -> Result<(), Box<dyn std::error::Error>>
 async fn two_commits() -> Result<(), Box<dyn std::error::Error>> {
     let batch = generate_simple_batch()?;
     let storage = Arc::new(InMemory::new());
+    let parquet_bytes = record_batch_to_bytes(&batch);
+    let file_size = parquet_bytes.len() as u64;
     add_commit(
         storage.as_ref(),
         0,
         actions_to_string(vec![
             TestAction::Metadata,
-            TestAction::Add(PARQUET_FILE1.to_string()),
+            TestAction::AddWithSize(PARQUET_FILE1.to_string(), file_size),
         ]),
     )
     .await?;
     add_commit(
         storage.as_ref(),
         1,
-        actions_to_string(vec![TestAction::Add(PARQUET_FILE2.to_string())]),
+        actions_to_string(vec![TestAction::AddWithSize(
+            PARQUET_FILE2.to_string(),
+            file_size,
+        )]),
     )
     .await?;
     storage
@@ -152,25 +159,33 @@ async fn two_commits() -> Result<(), Box<dyn std::error::Error>> {
 async fn remove_action() -> Result<(), Box<dyn std::error::Error>> {
     let batch = generate_simple_batch()?;
     let storage = Arc::new(InMemory::new());
+    let parquet_bytes = record_batch_to_bytes(&batch);
+    let file_size = parquet_bytes.len() as u64;
     add_commit(
         storage.as_ref(),
         0,
         actions_to_string(vec![
             TestAction::Metadata,
-            TestAction::Add(PARQUET_FILE1.to_string()),
+            TestAction::AddWithSize(PARQUET_FILE1.to_string(), file_size),
         ]),
     )
     .await?;
     add_commit(
         storage.as_ref(),
         1,
-        actions_to_string(vec![TestAction::Add(PARQUET_FILE2.to_string())]),
+        actions_to_string(vec![TestAction::AddWithSize(
+            PARQUET_FILE2.to_string(),
+            file_size,
+        )]),
     )
     .await?;
     add_commit(
         storage.as_ref(),
         2,
-        actions_to_string(vec![TestAction::Remove(PARQUET_FILE2.to_string())]),
+        actions_to_string(vec![TestAction::RemoveWithSize(
+            PARQUET_FILE2.to_string(),
+            file_size,
+        )]),
     )
     .await?;
     storage
@@ -209,6 +224,8 @@ async fn stats() -> Result<(), Box<dyn std::error::Error>> {
                 TestAction::Add(path) => format!(r#"{{"{action}":{{"path":"{path}","partitionValues":{{}},"size":262,"modificationTime":1587968586000,"dataChange":true, "stats":"{{\"numRecords\":2,\"nullCount\":{{\"id\":0}},\"minValues\":{{\"id\": 5}},\"maxValues\":{{\"id\":7}}}}"}}}}"#, action = "add", path = path),
                 TestAction::Remove(path) => format!(r#"{{"{action}":{{"path":"{path}","partitionValues":{{}},"size":262,"modificationTime":1587968586000,"dataChange":true}}}}"#, action = "remove", path = path),
                 TestAction::Metadata => METADATA.into(),
+                TestAction::AddWithSize(path, size) => format!(r#"{{"{action}":{{"path":"{path}","partitionValues":{{}},"size":{size},"modificationTime":1587968586000,"dataChange":true, "stats":"{{\"numRecords\":2,\"nullCount\":{{\"id\":0}},\"minValues\":{{\"id\": 5}},\"maxValues\":{{\"id\":7}}}}"}}}}"#, action = "add", path = path),
+                TestAction::RemoveWithSize(path, size) => format!(r#"{{"{action}":{{"path":"{path}","partitionValues":{{}},"size":{size},"modificationTime":1587968586000,"dataChange":true}}}}"#, action = "remove", path = path),
             })
             .fold(String::new(), |a, b| a + &b + "\n")
     }
@@ -218,22 +235,27 @@ async fn stats() -> Result<(), Box<dyn std::error::Error>> {
         ("id", vec![5, 7].into_array()),
         ("val", vec!["e", "g"].into_array()),
     ])?);
+    let file_size1 = record_batch_to_bytes(&batch1).len() as u64;
+    let file_size2 = record_batch_to_bytes(&batch2).len() as u64;
     let storage = Arc::new(InMemory::new());
     // valid commit with min/max (0, 2)
     add_commit(
         storage.as_ref(),
         0,
         actions_to_string(vec![
             TestAction::Metadata,
-            TestAction::Add(PARQUET_FILE1.to_string()),
+            TestAction::AddWithSize(PARQUET_FILE1.to_string(), file_size1),
         ]),
     )
     .await?;
     // storage.add_commit(1, &format!("{}\n", r#"{{"add":{{"path":"doesnotexist","partitionValues":{{}},"size":262,"modificationTime":1587968586000,"dataChange":true, "stats":"{{\"numRecords\":2,\"nullCount\":{{\"id\":0}},\"minValues\":{{\"id\": 0}},\"maxValues\":{{\"id\":2}}}}"}}}}"#));
     add_commit(
         storage.as_ref(),
         1,
-        generate_commit2(vec![TestAction::Add(PARQUET_FILE2.to_string())]),
+        generate_commit2(vec![TestAction::AddWithSize(
+            PARQUET_FILE2.to_string(),
+            file_size2,
+        )]),
     )
     .await?;
 
@@ -1335,15 +1357,18 @@ async fn test_row_index_metadata_column() -> Result<(), Box<dyn std::error::Erro
         ("value", vec!["p", "q", "r", "s"].into_array()),
     ])?;
 
+    let file_size1 = record_batch_to_bytes(&batch1).len() as u64;
+    let file_size2 = record_batch_to_bytes(&batch2).len() as u64;
+    let file_size3 = record_batch_to_bytes(&batch3).len() as u64;
     let storage = Arc::new(InMemory::new());
     add_commit(
         storage.as_ref(),
         0,
         actions_to_string(vec![
             TestAction::Metadata,
-            TestAction::Add(PARQUET_FILE1.to_string()),
-            TestAction::Add(PARQUET_FILE2.to_string()),
-            TestAction::Add(PARQUET_FILE3.to_string()),
+            TestAction::AddWithSize(PARQUET_FILE1.to_string(), file_size1),
+            TestAction::AddWithSize(PARQUET_FILE2.to_string(), file_size2),
+            TestAction::AddWithSize(PARQUET_FILE3.to_string(), file_size3),
         ]),
     )
     .await?;
@@ -1430,14 +1455,16 @@ async fn test_file_path_metadata_column() -> Result<(), Box<dyn std::error::Erro
         ("value", vec!["x", "y"].into_array()),
     ])?;
 
+    let file_size1 = record_batch_to_bytes(&batch1).len() as u64;
+    let file_size2 = record_batch_to_bytes(&batch2).len() as u64;
     let storage = Arc::new(InMemory::new());
     add_commit(
         storage.as_ref(),
         0,
         actions_to_string(vec![
             TestAction::Metadata,
-            TestAction::Add(PARQUET_FILE1.to_string()),
-            TestAction::Add(PARQUET_FILE2.to_string()),
+            TestAction::AddWithSize(PARQUET_FILE1.to_string(), file_size1),
+            TestAction::AddWithSize(PARQUET_FILE2.to_string(), file_size2),
         ]),
     )
     .await?;

test-utils/src/lib.rs

@@ -104,6 +104,10 @@ pub enum TestAction {
     Add(String),
     Remove(String),
     Metadata,
+    // TODO: This is a temporary fix to make the test compatible with the file size requirement.
+    // In the future, we can create an AddCommit/RemoveCommit struct type with DefaultAddCommit/DefaultRemoveCommit value to store the commit info in the enum for Add/Remove.
+    AddWithSize(String, u64),
+    RemoveWithSize(String, u64),
 }
 
 // TODO: We need a better way to mock tables :)
@@ -125,6 +129,8 @@ pub fn actions_to_string_with_metadata(actions: Vec<TestAction>, metadata: &str)
             TestAction::Add(path) => format!(r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":262,"modificationTime":1587968586000,"dataChange":true, "stats":"{{\"numRecords\":2,\"nullCount\":{{\"id\":0}},\"minValues\":{{\"id\": 1}},\"maxValues\":{{\"id\":3}}}}"}}}}"#),
             TestAction::Remove(path) => format!(r#"{{"remove":{{"path":"{path}","partitionValues":{{}},"size":262,"modificationTime":1587968586000,"dataChange":true}}}}"#),
             TestAction::Metadata => metadata.into(),
+            TestAction::AddWithSize(path, file_size) => format!(r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":{file_size},"modificationTime":1587968586000,"dataChange":true, "stats":"{{\"numRecords\":2,\"nullCount\":{{\"id\":0}},\"minValues\":{{\"id\": 1}},\"maxValues\":{{\"id\":3}}}}"}}}}"#),
+            TestAction::RemoveWithSize(path, file_size) => format!(r#"{{"remove":{{"path":"{path}","partitionValues":{{}},"size":{file_size},"modificationTime":1587968586000,"dataChange":true}}}}"#),
         })
         .join("\n")
 }