fix: cherry-pick #27 (#41)

Author: chenzl25

crates/iceberg/src/arrow/delete_file_manager.rs

@@ -57,6 +57,7 @@ impl CachingDeleteFileManager {
         }
     }
 
+    #[allow(dead_code)]
     pub(crate) async fn load_deletes(
         &self,
         delete_file_entries: Vec<FileScanTaskDeleteFile>,

crates/iceberg/src/arrow/reader.rs

@@ -39,8 +39,8 @@ use parquet::arrow::{ParquetRecordBatchStreamBuilder, ProjectionMask, PARQUET_FI
 use parquet::file::metadata::{ParquetMetaData, ParquetMetaDataReader, RowGroupMetaData};
 use parquet::schema::types::{SchemaDescriptor, Type as ParquetType};
 
+use super::record_batch_transformer::RecordBatchTransformer;
 use crate::arrow::delete_file_manager::CachingDeleteFileManager;
-use crate::arrow::record_batch_transformer::RecordBatchTransformer;
 use crate::arrow::{arrow_schema_to_schema, get_arrow_datum};
 use crate::delete_vector::DeleteVector;
 use crate::error::Result;
@@ -50,7 +50,7 @@ use crate::expr::visitors::row_group_metrics_evaluator::RowGroupMetricsEvaluator
 use crate::expr::{BoundPredicate, BoundReference};
 use crate::io::{FileIO, FileMetadata, FileRead};
 use crate::scan::{ArrowRecordBatchStream, FileScanTask, FileScanTaskStream};
-use crate::spec::{Datum, NestedField, PrimitiveType, Schema, Type};
+use crate::spec::{DataContentType, Datum, NestedField, PrimitiveType, Schema, Type};
 use crate::utils::available_parallelism;
 use crate::{Error, ErrorKind};
 
@@ -176,14 +176,12 @@ impl ArrowReader {
             (row_selection_enabled && task.predicate.is_some()) || !task.deletes.is_empty();
 
         // concurrently retrieve delete files and create RecordBatchStreamBuilder
-        let (_, mut record_batch_stream_builder) = try_join!(
-            delete_file_manager.load_deletes(task.deletes.clone()),
-            Self::create_parquet_record_batch_stream_builder(
-                &task.data_file_path,
-                file_io.clone(),
-                should_load_page_index,
-            )
-        )?;
+        let mut record_batch_stream_builder = Self::create_parquet_record_batch_stream_builder(
+            &task.data_file_path,
+            file_io.clone(),
+            should_load_page_index,
+        )
+        .await?;
 
         // Create a projection mask for the batch stream to select which columns in the
         // Parquet file that we want in the response
@@ -305,13 +303,16 @@ impl ArrowReader {
 
         // Build the batch stream and send all the RecordBatches that it generates
         // to the requester.
-        let record_batch_stream =
-            record_batch_stream_builder
-                .build()?
-                .map(move |batch| match batch {
+        let record_batch_stream = record_batch_stream_builder.build()?.map(move |batch| {
+            if matches!(task.data_file_content, DataContentType::PositionDeletes) {
+                Ok(batch?)
+            } else {
+                match batch {
                     Ok(batch) => record_batch_transformer.process_record_batch(batch),
                     Err(err) => Err(err.into()),
-                });
+                }
+            }
+        });
 
         Ok(Box::pin(record_batch_stream) as ArrowRecordBatchStream)
     }
@@ -1520,6 +1521,8 @@ message schema {
                 project_field_ids: vec![1],
                 predicate: Some(predicate.bind(schema, true).unwrap()),
                 deletes: vec![],
+                sequence_number: 0,
+                equality_ids: vec![],
             })]
             .into_iter(),
         )) as FileScanTaskStream;

crates/iceberg/src/delete_file_index.rs

@@ -26,7 +26,7 @@ use futures::channel::mpsc::{channel, Sender};
 use futures::StreamExt;
 
 use crate::runtime::spawn;
-use crate::scan::{DeleteFileContext, FileScanTaskDeleteFile};
+use crate::scan::{DeleteFileContext, FileScanTask};
 use crate::spec::{DataContentType, DataFile, Struct};
 use crate::{Error, ErrorKind, Result};
 
@@ -118,10 +118,11 @@ impl PopulatedDeleteFileIndex {
             // The spec states that "Equality delete files stored with an unpartitioned spec are applied as global deletes".
             if partition.fields().is_empty() {
                 // TODO: confirm we're good to skip here if we encounter a pos del
-                if arc_ctx.manifest_entry.content_type() != DataContentType::PositionDeletes {
-                    global_deletes.push(arc_ctx);
-                    return;
-                }
+                // FIXME(Dylan): allow putting position delete to global deletes.
+                // if arc_ctx.manifest_entry.content_type() != DataContentType::PositionDeletes {
+                global_deletes.push(arc_ctx);
+                return;
+                // }
             }
 
             let destination_map = match arc_ctx.manifest_entry.content_type() {
@@ -150,7 +151,7 @@ impl PopulatedDeleteFileIndex {
         &self,
         data_file: &DataFile,
         seq_num: Option<i64>,
-    ) -> Vec<FileScanTaskDeleteFile> {
+    ) -> Vec<FileScanTask> {
         let mut results = vec![];
 
         self.global_deletes
@@ -203,7 +204,7 @@ pub(crate) struct DeletesForDataFile<'a> {
 }
 
 impl Future for DeletesForDataFile<'_> {
-    type Output = Result<Vec<FileScanTaskDeleteFile>>;
+    type Output = Result<Vec<FileScanTask>>;
 
     fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
         match self.state.try_read() {

crates/iceberg/src/scan/context.rs

@@ -140,6 +140,8 @@ impl ManifestEntryContext {
                 .map(|x| x.as_ref().snapshot_bound_predicate.clone()),
 
             deletes,
+            sequence_number: self.manifest_entry.sequence_number().unwrap_or(0),
+            equality_ids: self.manifest_entry.data_file().equality_ids().to_vec(),
         })
     }
 }

crates/iceberg/src/scan/mod.rs

@@ -605,6 +605,8 @@ impl TableScan {
             .send(DeleteFileContext {
                 manifest_entry: manifest_entry_context.manifest_entry.clone(),
                 partition_spec_id: manifest_entry_context.partition_spec_id,
+                snapshot_schema: manifest_entry_context.snapshot_schema.clone(),
+                field_ids: manifest_entry_context.field_ids.clone(),
             })
             .await?;
 
@@ -1791,6 +1793,8 @@ pub mod tests {
             record_count: Some(100),
             data_file_format: DataFileFormat::Parquet,
             deletes: vec![],
+            sequence_number: 0,
+            equality_ids: vec![],
         };
         test_fn(task);
 
@@ -1806,6 +1810,8 @@ pub mod tests {
             record_count: None,
             data_file_format: DataFileFormat::Avro,
             deletes: vec![],
+            sequence_number: 0,
+            equality_ids: vec![],
         };
         test_fn(task);
     }

crates/iceberg/src/scan/task.rs

@@ -15,6 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use std::sync::Arc;
+
 use futures::stream::BoxStream;
 use serde::{Deserialize, Serialize};
 
@@ -56,7 +58,11 @@ pub struct FileScanTask {
     pub predicate: Option<BoundPredicate>,
 
     /// The list of delete files that may need to be applied to this data file
-    pub deletes: Vec<FileScanTaskDeleteFile>,
+    pub deletes: Vec<FileScanTask>,
+    /// sequence number
+    pub sequence_number: i64,
+    /// equality ids
+    pub equality_ids: Vec<i32>,
 }
 
 impl FileScanTask {
@@ -90,6 +96,8 @@ impl FileScanTask {
 pub(crate) struct DeleteFileContext {
     pub(crate) manifest_entry: ManifestEntryRef,
     pub(crate) partition_spec_id: i32,
+    pub(crate) snapshot_schema: SchemaRef,
+    pub(crate) field_ids: Arc<Vec<i32>>,
 }
 
 impl From<&DeleteFileContext> for FileScanTaskDeleteFile {
@@ -102,6 +110,27 @@ impl From<&DeleteFileContext> for FileScanTaskDeleteFile {
     }
 }
 
+impl From<&DeleteFileContext> for FileScanTask {
+    fn from(ctx: &DeleteFileContext) -> Self {
+        FileScanTask {
+            start: 0,
+            length: ctx.manifest_entry.file_size_in_bytes(),
+            record_count: Some(ctx.manifest_entry.record_count()),
+
+            data_file_path: ctx.manifest_entry.file_path().to_string(),
+            data_file_content: ctx.manifest_entry.content_type(),
+            data_file_format: ctx.manifest_entry.file_format(),
+
+            schema: ctx.snapshot_schema.clone(),
+            project_field_ids: ctx.field_ids.to_vec(),
+            predicate: None,
+            deletes: vec![],
+            sequence_number: ctx.manifest_entry.sequence_number().unwrap_or(0),
+            equality_ids: ctx.manifest_entry.data_file().equality_ids().to_vec(),
+        }
+    }
+}
+
 /// A task to scan part of file.
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 pub struct FileScanTaskDeleteFile {

crates/iceberg/src/writer/function_writer/equality_delta_writer.rs

@@ -255,12 +255,14 @@ where
 
 #[cfg(test)]
 mod test {
+    use std::collections::HashMap;
     use std::sync::Arc;
 
     use arrow_array::{Int32Array, Int64Array, RecordBatch, StringArray};
     use arrow_schema::{DataType, Field, Schema as ArrowSchema};
     use arrow_select::concat::concat_batches;
     use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
+    use parquet::arrow::PARQUET_FIELD_ID_META_KEY;
     use parquet::file::properties::WriterProperties;
     use tempfile::TempDir;
 
@@ -318,7 +320,7 @@ mod test {
                 location_gen.clone(),
                 file_name_gen.clone(),
             );
-            DataFileWriterBuilder::new(pw.clone(), None, None)
+            DataFileWriterBuilder::new(pw.clone(), None, 0)
         };
         let position_delete_writer_builder = {
             let pw = ParquetWriterBuilder::new(
@@ -331,7 +333,7 @@ mod test {
             SortPositionDeleteWriterBuilder::new(pw.clone(), 100, None, None)
         };
         let equality_delete_writer_builder = {
-            let config = EqualityDeleteWriterConfig::new(vec![1, 2], schema, None, None)?;
+            let config = EqualityDeleteWriterConfig::new(vec![1, 2], schema, None, 0)?;
             let pw = ParquetWriterBuilder::new(
                 WriterProperties::builder().build(),
                 arrow_schema_to_schema(config.projected_arrow_schema_ref())
@@ -355,9 +357,18 @@ mod test {
 
         // write data
         let schema = Arc::new(ArrowSchema::new(vec![
-            Field::new("id", DataType::Int64, true),
-            Field::new("data", DataType::Utf8, true),
-            Field::new("op", DataType::Int32, false),
+            Field::new("id", DataType::Int64, true).with_metadata(HashMap::from([(
+                PARQUET_FIELD_ID_META_KEY.to_string(),
+                1.to_string(),
+            )])),
+            Field::new("data", DataType::Utf8, true).with_metadata(HashMap::from([(
+                PARQUET_FIELD_ID_META_KEY.to_string(),
+                2.to_string(),
+            )])),
+            Field::new("op", DataType::Int32, false).with_metadata(HashMap::from([(
+                PARQUET_FIELD_ID_META_KEY.to_string(),
+                3.to_string(),
+            )])),
         ]));
         {
             let id_array = Int64Array::from(vec![1, 2, 1, 3, 2, 3, 1]);
@@ -388,8 +399,14 @@ mod test {
         assert_eq!(data_files.len(), 3);
         // data file
         let data_schema = Arc::new(ArrowSchema::new(vec![
-            Field::new("id", DataType::Int64, true),
-            Field::new("data", DataType::Utf8, true),
+            Field::new("id", DataType::Int64, true).with_metadata(HashMap::from([(
+                PARQUET_FIELD_ID_META_KEY.to_string(),
+                1.to_string(),
+            )])),
+            Field::new("data", DataType::Utf8, true).with_metadata(HashMap::from([(
+                PARQUET_FIELD_ID_META_KEY.to_string(),
+                2.to_string(),
+            )])),
         ]));
         let data_file = data_files
             .iter()

crates/iceberg/src/writer/function_writer/mod.rs

@@ -17,5 +17,6 @@
 
 //! This module contains the functional writer.
 
+pub mod equality_delta_writer;
 pub mod fanout_partition_writer;
 pub mod precompute_partition_writer;