Merge branch 'main' into partition-spec-support

Author: mbutrovich

Cargo.lock

@@ -71,6 +71,7 @@ enum-ordinalize = "4.3.0"
 env_logger = "0.11.8"
 expect-test = "1"
 faststr = "0.2.31"
+flate2 = "1.1.5"
 fnv = "1.0.7"
 fs-err = "3.1.0"
 futures = "0.3"

Cargo.toml

@@ -63,6 +63,7 @@ bytes = { workspace = true }
 chrono = { workspace = true }
 derive_builder = { workspace = true }
 expect-test = { workspace = true }
+flate2 = { workspace = true }
 fnv = { workspace = true }
 futures = { workspace = true }
 itertools = { workspace = true }

crates/iceberg/Cargo.toml

@@ -788,4 +788,117 @@ mod tests {
         assert_eq!(data_col.value(1), "d");
         assert_eq!(data_col.value(2), "g");
     }
+
+    /// Test loading a FileScanTask with BOTH positional and equality deletes.
+    /// Verifies the fix for the inverted condition that caused "Missing predicate for equality delete file" errors.
+    #[tokio::test]
+    async fn test_load_deletes_with_mixed_types() {
+        use crate::scan::FileScanTask;
+        use crate::spec::{DataFileFormat, Schema};
+
+        let tmp_dir = TempDir::new().unwrap();
+        let table_location = tmp_dir.path();
+        let file_io = FileIO::from_path(table_location.as_os_str().to_str().unwrap())
+            .unwrap()
+            .build()
+            .unwrap();
+
+        // Create the data file schema
+        let data_file_schema = Arc::new(
+            Schema::builder()
+                .with_fields(vec![
+                    crate::spec::NestedField::optional(
+                        2,
+                        "y",
+                        crate::spec::Type::Primitive(crate::spec::PrimitiveType::Long),
+                    )
+                    .into(),
+                    crate::spec::NestedField::optional(
+                        3,
+                        "z",
+                        crate::spec::Type::Primitive(crate::spec::PrimitiveType::Long),
+                    )
+                    .into(),
+                ])
+                .build()
+                .unwrap(),
+        );
+
+        // Write positional delete file
+        let positional_delete_schema = crate::arrow::delete_filter::tests::create_pos_del_schema();
+        let file_path_values =
+            vec![format!("{}/data-1.parquet", table_location.to_str().unwrap()); 4];
+        let file_path_col = Arc::new(StringArray::from_iter_values(&file_path_values));
+        let pos_col = Arc::new(Int64Array::from_iter_values(vec![0i64, 1, 2, 3]));
+
+        let positional_deletes_to_write =
+            RecordBatch::try_new(positional_delete_schema.clone(), vec![
+                file_path_col,
+                pos_col,
+            ])
+            .unwrap();
+
+        let props = WriterProperties::builder()
+            .set_compression(Compression::SNAPPY)
+            .build();
+
+        let pos_del_path = format!("{}/pos-del-mixed.parquet", table_location.to_str().unwrap());
+        let file = File::create(&pos_del_path).unwrap();
+        let mut writer = ArrowWriter::try_new(
+            file,
+            positional_deletes_to_write.schema(),
+            Some(props.clone()),
+        )
+        .unwrap();
+        writer.write(&positional_deletes_to_write).unwrap();
+        writer.close().unwrap();
+
+        // Write equality delete file
+        let eq_delete_path = setup_write_equality_delete_file_1(table_location.to_str().unwrap());
+
+        // Create FileScanTask with BOTH positional and equality deletes
+        let pos_del = FileScanTaskDeleteFile {
+            file_path: pos_del_path,
+            file_type: DataContentType::PositionDeletes,
+            partition_spec_id: 0,
+            equality_ids: None,
+        };
+
+        let eq_del = FileScanTaskDeleteFile {
+            file_path: eq_delete_path.clone(),
+            file_type: DataContentType::EqualityDeletes,
+            partition_spec_id: 0,
+            equality_ids: Some(vec![2, 3]), // Only use field IDs that exist in both schemas
+        };
+
+        let file_scan_task = FileScanTask {
+            start: 0,
+            length: 0,
+            record_count: None,
+            data_file_path: format!("{}/data-1.parquet", table_location.to_str().unwrap()),
+            data_file_format: DataFileFormat::Parquet,
+            schema: data_file_schema.clone(),
+            project_field_ids: vec![2, 3],
+            predicate: None,
+            deletes: vec![pos_del, eq_del],
+        };
+
+        // Load the deletes - should handle both types without error
+        let delete_file_loader = CachingDeleteFileLoader::new(file_io.clone(), 10);
+        let delete_filter = delete_file_loader
+            .load_deletes(&file_scan_task.deletes, file_scan_task.schema_ref())
+            .await
+            .unwrap()
+            .unwrap();
+
+        // Verify both delete types can be processed together
+        let result = delete_filter
+            .build_equality_delete_predicate(&file_scan_task)
+            .await;
+        assert!(
+            result.is_ok(),
+            "Failed to build equality delete predicate: {:?}",
+            result.err()
+        );
+    }
 }

crates/iceberg/src/arrow/caching_delete_file_loader.rs

@@ -68,10 +68,12 @@ impl DeleteFilter {
     pub(crate) fn try_start_eq_del_load(&self, file_path: &str) -> Option<Arc<Notify>> {
         let mut state = self.state.write().unwrap();
 
-        if !state.equality_deletes.contains_key(file_path) {
+        // Skip if already loaded/loading - another task owns it
+        if state.equality_deletes.contains_key(file_path) {
             return None;
         }
 
+        // Mark as loading to prevent duplicate work
         let notifier = Arc::new(Notify::new());
         state
             .equality_deletes

crates/iceberg/src/arrow/delete_filter.rs

@@ -38,19 +38,13 @@ pub const PROJECTED_PARTITION_VALUE_COLUMN: &str = "_partition";
 /// The splitter supports two modes for obtaining partition values:
 /// - **Computed mode** (`calculator` is `Some`): Computes partition values from source columns using transforms
 /// - **Pre-computed mode** (`calculator` is `None`): Expects a `_partition` column in the input batch
-// # TODO
-// Remove this after partition writer supported.
-#[allow(dead_code)]
 pub struct RecordBatchPartitionSplitter {
     schema: SchemaRef,
     partition_spec: PartitionSpecRef,
     calculator: Option<PartitionValueCalculator>,
     partition_type: StructType,
 }
 
-// # TODO
-// Remove this after partition writer supported.
-#[allow(dead_code)]
 impl RecordBatchPartitionSplitter {
     /// Create a new RecordBatchPartitionSplitter.
     ///

crates/iceberg/src/arrow/record_batch_partition_splitter.rs

@@ -22,10 +22,12 @@ use std::cmp::Ordering;
 use std::collections::HashMap;
 use std::fmt::{Display, Formatter};
 use std::hash::Hash;
+use std::io::Read as _;
 use std::sync::Arc;
 
 use _serde::TableMetadataEnum;
 use chrono::{DateTime, Utc};
+use flate2::read::GzDecoder;
 use serde::{Deserialize, Serialize};
 use serde_repr::{Deserialize_repr, Serialize_repr};
 use uuid::Uuid;
@@ -426,9 +428,30 @@ impl TableMetadata {
         file_io: &FileIO,
         metadata_location: impl AsRef<str>,
     ) -> Result<TableMetadata> {
+        let metadata_location = metadata_location.as_ref();
         let input_file = file_io.new_input(metadata_location)?;
         let metadata_content = input_file.read().await?;
-        let metadata = serde_json::from_slice::<TableMetadata>(&metadata_content)?;
+
+        // Check if the file is compressed by looking for the gzip "magic number".
+        let metadata = if metadata_content.len() > 2
+            && metadata_content[0] == 0x1F
+            && metadata_content[1] == 0x8B
+        {
+            let mut decoder = GzDecoder::new(metadata_content.as_ref());
+            let mut decompressed_data = Vec::new();
+            decoder.read_to_end(&mut decompressed_data).map_err(|e| {
+                Error::new(
+                    ErrorKind::DataInvalid,
+                    "Trying to read compressed metadata file",
+                )
+                .with_context("file_path", metadata_location)
+                .with_source(e)
+            })?;
+            serde_json::from_slice(&decompressed_data)?
+        } else {
+            serde_json::from_slice(&metadata_content)?
+        };
+
         Ok(metadata)
     }
 
@@ -1516,6 +1539,7 @@ impl SnapshotLog {
 mod tests {
     use std::collections::HashMap;
     use std::fs;
+    use std::io::Write as _;
     use std::sync::Arc;
 
     use anyhow::Result;
@@ -3524,6 +3548,30 @@ mod tests {
         assert_eq!(read_metadata, original_metadata);
     }
 
+    #[tokio::test]
+    async fn test_table_metadata_read_compressed() {
+        let temp_dir = TempDir::new().unwrap();
+        let metadata_location = temp_dir.path().join("v1.gz.metadata.json");
+
+        let original_metadata: TableMetadata = get_test_table_metadata("TableMetadataV2Valid.json");
+        let json = serde_json::to_string(&original_metadata).unwrap();
+
+        let mut encoder = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
+        encoder.write_all(json.as_bytes()).unwrap();
+        std::fs::write(&metadata_location, encoder.finish().unwrap())
+            .expect("failed to write metadata");
+
+        // Read the metadata back
+        let file_io = FileIOBuilder::new_fs_io().build().unwrap();
+        let metadata_location = metadata_location.to_str().unwrap();
+        let read_metadata = TableMetadata::read_from(&file_io, metadata_location)
+            .await
+            .unwrap();
+
+        // Verify the metadata matches
+        assert_eq!(read_metadata, original_metadata);
+    }
+
     #[tokio::test]
     async fn test_table_metadata_read_nonexistent_file() {
         // Create a FileIO instance

crates/iceberg/src/spec/table_metadata.rs

@@ -297,9 +297,7 @@ impl StaticTable {
         table_ident: TableIdent,
         file_io: FileIO,
     ) -> Result<Self> {
-        let metadata_file = file_io.new_input(metadata_location)?;
-        let metadata_file_content = metadata_file.read().await?;
-        let metadata = serde_json::from_slice::<TableMetadata>(&metadata_file_content)?;
+        let metadata = TableMetadata::read_from(&file_io, metadata_location).await?;
 
         let table = Table::builder()
             .metadata(metadata)

crates/iceberg/src/table.rs

@@ -23,6 +23,7 @@
 
 pub mod clustered_writer;
 pub mod fanout_writer;
+pub mod unpartitioned_writer;
 
 use crate::Result;
 use crate::spec::PartitionKey;