feat: Iceberg Table supports equality delete of merge on read when reading (#18015)

* feat: feat: Iceberg Table supports equality delete of merge on read when reading

* test: add iceberg driver for equality delete file

* test: add a test to insert new data after deletion

* chore: simplify code

Author: KKould

src/query/expression/src/function.rs

@@ -154,7 +154,7 @@ pub enum FunctionEval {
     },
 }
 
-#[derive(Clone)]
+#[derive(Clone, Debug)]
 pub struct FunctionContext {
     pub tz: TimeZone,
     pub now: Zoned,

src/query/storages/iceberg/src/partition.rs

@@ -13,9 +13,12 @@
 // limitations under the License.
 
 use databend_common_catalog::plan::PartInfo;
+use databend_common_storages_parquet::DeleteTask;
+use databend_common_storages_parquet::DeleteType;
 use databend_common_storages_parquet::ParquetFilePart;
 use databend_common_storages_parquet::ParquetPart;
 use databend_storages_common_stage::SingleFilePartition;
+use iceberg::spec::DataContentType;
 use iceberg::spec::DataFileFormat;
 
 pub(crate) fn convert_file_scan_task(task: iceberg::scan::FileScanTask) -> Box<dyn PartInfo> {
@@ -42,7 +45,18 @@ pub(crate) fn convert_file_scan_task(task: iceberg::scan::FileScanTask) -> Box<d
                     deletes: task
                         .deletes
                         .iter()
-                        .map(|file| file.file_path.clone())
+                        .filter_map(|file| {
+                            let ty = match file.file_type {
+                                DataContentType::Data => return None,
+                                DataContentType::PositionDeletes => DeleteType::Position,
+                                DataContentType::EqualityDeletes => DeleteType::Equality,
+                            };
+                            Some(DeleteTask {
+                                path: file.file_path.clone(),
+                                ty,
+                                equality_ids: file.equality_ids.clone(),
+                            })
+                        })
                         .collect(),
                 })
             } else {

src/query/storages/parquet/src/copy_into_table/reader.rs

@@ -54,7 +54,7 @@ impl RowGroupReaderForCopy {
             InMemoryRowGroup::new(&part.location, op.clone(), &part.meta, None, *read_settings);
         let mut _sorter = None;
         self.row_group_reader_builder
-            .fetch_and_build(row_group, None, &mut _sorter, None, batch_size)
+            .fetch_and_build(row_group, None, &mut _sorter, None, batch_size, None)
             .await
     }
 

src/query/storages/parquet/src/lib.rs

@@ -44,6 +44,8 @@ mod meta;
 mod schema;
 
 pub use copy_into_table::ParquetTableForCopy;
+pub use parquet_part::DeleteTask;
+pub use parquet_part::DeleteType;
 pub use parquet_part::ParquetFilePart;
 pub use parquet_part::ParquetPart;
 pub use parquet_reader::InmMemoryFile;

src/query/storages/parquet/src/parquet_part.rs

@@ -27,14 +27,28 @@ use databend_common_exception::Result;
 
 use crate::partition::ParquetRowGroupPart;
 
+#[derive(serde::Serialize, serde::Deserialize, PartialEq, Eq, Debug, Clone)]
+pub enum DeleteType {
+    Position,
+    Equality,
+}
+
+#[derive(serde::Serialize, serde::Deserialize, PartialEq, Eq, Debug, Clone)]
+pub struct DeleteTask {
+    pub path: String,
+    pub ty: DeleteType,
+    /// equality ids for equality deletes (empty for positional deletes)
+    pub equality_ids: Vec<i32>,
+}
+
 #[derive(serde::Serialize, serde::Deserialize, PartialEq, Debug, Clone)]
 pub enum ParquetPart {
     File(ParquetFilePart),
     SmallFiles(Vec<ParquetFilePart>),
     RowGroup(ParquetRowGroupPart),
     FileWithDeletes {
         inner: ParquetFilePart,
-        deletes: Vec<String>,
+        deletes: Vec<DeleteTask>,
     },
 }
 
@@ -114,7 +128,7 @@ impl PartInfo for ParquetPart {
                 let mut paths = Vec::with_capacity(deletes.len() + 1);
                 paths.push(&inner.file);
                 for delete in deletes {
-                    paths.push(delete);
+                    paths.push(&delete.path);
                 }
                 paths
             }

src/query/storages/parquet/src/parquet_reader/predicate.rs

@@ -39,6 +39,7 @@ use super::utils::transform_record_batch;
 use super::utils::FieldPaths;
 use crate::parquet_reader::utils::compute_output_field_paths;
 
+#[derive(Debug)]
 pub struct ParquetPredicate {
     func_ctx: FunctionContext,
 

src/query/storages/parquet/src/parquet_reader/read_policy/no_prefetch.rs

@@ -28,7 +28,10 @@ use parquet::schema::types::SchemaDescriptor;
 use super::policy::ReadPolicy;
 use super::policy::ReadPolicyBuilder;
 use super::policy::ReadPolicyImpl;
+use crate::parquet_reader::predicate::ParquetPredicate;
+use crate::parquet_reader::read_policy::utils::read_all;
 use crate::parquet_reader::row_group::InMemoryRowGroup;
+use crate::parquet_reader::utils::bitmap_to_boolean_array;
 use crate::parquet_reader::utils::transform_record_batch;
 use crate::parquet_reader::utils::FieldPaths;
 use crate::transformer::RecordBatchTransformer;
@@ -45,11 +48,47 @@ impl ReadPolicyBuilder for NoPretchPolicyBuilder {
     async fn fetch_and_build(
         &self,
         mut row_group: InMemoryRowGroup<'_>,
-        row_selection: Option<RowSelection>,
+        mut row_selection: Option<RowSelection>,
         _sorter: &mut Option<TopKSorter>,
         transformer: Option<RecordBatchTransformer>,
         batch_size: usize,
+        filter: Option<Arc<ParquetPredicate>>,
     ) -> Result<Option<ReadPolicyImpl>> {
+        if let Some(predicate) = filter {
+            row_group
+                .fetch(predicate.projection(), row_selection.as_ref())
+                .await?;
+
+            let num_rows = row_selection
+                .as_ref()
+                .map(|x| x.row_count())
+                .unwrap_or(row_group.row_count());
+
+            let block = read_all(
+                &DataSchema::from(predicate.schema()),
+                &row_group,
+                predicate.field_levels(),
+                row_selection.clone(),
+                predicate.field_paths(),
+                num_rows,
+            )?;
+            let filter = predicate.evaluate_block(&block)?;
+            if filter.null_count() == num_rows {
+                // All rows in current row group are filtered out.
+                return Ok(None);
+            }
+            let filter = bitmap_to_boolean_array(filter);
+            let sel = RowSelection::from_filters(&[filter]);
+            match row_selection.as_mut() {
+                Some(selection) => {
+                    *selection = selection.and_then(&sel);
+                }
+                None => {
+                    row_selection = Some(sel);
+                }
+            }
+        }
+
         row_group.fetch(&self.projection, None).await?;
         let reader = ParquetRecordBatchReader::try_new_with_row_groups(
             &self.field_levels,

src/query/storages/parquet/src/parquet_reader/read_policy/policy.rs

@@ -12,11 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::sync::Arc;
+
 use databend_common_exception::Result;
 use databend_common_expression::DataBlock;
 use databend_common_expression::TopKSorter;
 use parquet::arrow::arrow_reader::RowSelection;
 
+use crate::parquet_reader::predicate::ParquetPredicate;
 use crate::parquet_reader::row_group::InMemoryRowGroup;
 use crate::transformer::RecordBatchTransformer;
 
@@ -66,6 +69,7 @@ pub trait ReadPolicyBuilder: Send + Sync {
         _sorter: &mut Option<TopKSorter>,
         _transformer: Option<RecordBatchTransformer>,
         _batch_size: usize,
+        _filter: Option<Arc<ParquetPredicate>>,
     ) -> Result<Option<ReadPolicyImpl>> {
         unreachable!()
     }

src/query/storages/parquet/src/parquet_reader/read_policy/predicate_and_topk.rs

@@ -161,6 +161,7 @@ impl ReadPolicyBuilder for PredicateAndTopkPolicyBuilder {
         sorter: &mut Option<TopKSorter>,
         transformer: Option<RecordBatchTransformer>,
         batch_size: usize,
+        _filter: Option<Arc<ParquetPredicate>>,
     ) -> Result<Option<ReadPolicyImpl>> {
         let mut num_rows = selection
             .as_ref()

src/query/storages/parquet/src/parquet_reader/read_policy/topk_only.rs

@@ -35,6 +35,7 @@ use super::policy::ReadPolicyBuilder;
 use super::policy::ReadPolicyImpl;
 use super::utils::evaluate_topk;
 use super::utils::read_all;
+use crate::parquet_reader::predicate::ParquetPredicate;
 use crate::parquet_reader::row_group::InMemoryRowGroup;
 use crate::parquet_reader::topk::BuiltTopK;
 use crate::parquet_reader::topk::ParquetTopK;
@@ -136,6 +137,7 @@ impl ReadPolicyBuilder for TopkOnlyPolicyBuilder {
         sorter: &mut Option<TopKSorter>,
         transformer: Option<RecordBatchTransformer>,
         batch_size: usize,
+        _filter: Option<Arc<ParquetPredicate>>,
     ) -> Result<Option<ReadPolicyImpl>> {
         debug_assert!(sorter.is_some());
         let sorter = sorter.as_mut().unwrap();