feat(scan): support deletion vector cardinality (#200)

Author: luoyuxia

crates/paimon/src/spec/index_file_meta.rs

@@ -20,6 +20,13 @@ use std::fmt::{Display, Formatter};
 
 use indexmap::IndexMap;
 
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct DeletionVectorMeta {
+    pub offset: i32,
+    pub length: i32,
+    pub cardinality: Option<i64>,
+}
+
 /// Metadata of index file.
 ///
 /// Impl Reference: <https://github.com/apache/paimon/blob/release-0.8.2/paimon-core/src/main/java/org/apache/paimon/index/IndexFileMeta.java>
@@ -44,7 +51,7 @@ pub struct IndexFileMeta {
         rename = "_DELETIONS_VECTORS_RANGES",
         alias = "_DELETION_VECTORS_RANGES"
     )]
-    pub deletion_vectors_ranges: Option<IndexMap<String, (i32, i32)>>,
+    pub deletion_vectors_ranges: Option<IndexMap<String, DeletionVectorMeta>>,
 }
 
 impl Display for IndexFileMeta {
@@ -65,42 +72,56 @@ mod map_serde {
     use indexmap::IndexMap;
     use serde::{Deserialize, Deserializer, Serialize, Serializer};
 
+    use super::DeletionVectorMeta;
+
     #[derive(Deserialize, Serialize)]
     struct Temp {
         f0: String,
         f1: i32,
         f2: i32,
+        #[serde(default, rename = "_CARDINALITY")]
+        cardinality: Option<i64>,
     }
 
     pub fn serialize<S>(
-        date: &Option<IndexMap<String, (i32, i32)>>,
+        data: &Option<IndexMap<String, DeletionVectorMeta>>,
         s: S,
     ) -> Result<S::Ok, S::Error>
     where
         S: Serializer,
     {
-        match *date {
+        match data {
             None => s.serialize_none(),
-            Some(ref d) => s.collect_seq(d.iter().map(|(s, (i1, i2))| Temp {
-                f0: s.into(),
-                f1: *i1,
-                f2: *i2,
+            Some(d) => s.collect_seq(d.iter().map(|(path, meta)| Temp {
+                f0: path.clone(),
+                f1: meta.offset,
+                f2: meta.length,
+                cardinality: meta.cardinality,
             })),
         }
     }
 
     #[allow(clippy::type_complexity)]
     pub fn deserialize<'de, D>(
         deserializer: D,
-    ) -> Result<Option<IndexMap<String, (i32, i32)>>, D::Error>
+    ) -> Result<Option<IndexMap<String, DeletionVectorMeta>>, D::Error>
     where
         D: Deserializer<'de>,
     {
         match Option::deserialize(deserializer)? {
             None => Ok(None),
             Some::<Vec<Temp>>(s) => Ok(Some(
                 s.into_iter()
-                    .map(|t| (t.f0, (t.f1, t.f2)))
+                    .map(|t| {
+                        (
+                            t.f0,
+                            DeletionVectorMeta {
+                                offset: t.f1,
+                                length: t.f2,
+                                cardinality: t.cardinality,
+                            },
+                        )
+                    })
                     .collect::<IndexMap<_, _>>(),
             )),
         }

crates/paimon/src/spec/index_manifest.rs

@@ -90,54 +90,38 @@ mod tests {
     use indexmap::IndexMap;
 
     use super::*;
+    use crate::spec::DeletionVectorMeta;
 
     #[test]
     fn test_read_index_manifest_file() {
         let workdir =
             std::env::current_dir().unwrap_or_else(|err| panic!("current_dir must exist: {err}"));
         let path = workdir
-            .join("tests/fixtures/manifest/index-manifest-85cc6729-f5af-431a-a1c3-ef45319328fb-0");
+            .join("tests/fixtures/manifest/index-manifest-7e816ed9-9f3b-4786-9985-8937d4e07b6e-0");
         let source = std::fs::read(path.to_str().unwrap()).unwrap();
-        let mut reader =
-            serde_avro_fast::object_container_file_encoding::Reader::from_slice(source.as_slice())
-                .unwrap();
-        let res: Vec<_> = reader
-            .deserialize::<IndexManifestEntry>()
-            .collect::<Result<_, _>>()
-            .unwrap();
+        let res = IndexManifest::read_from_bytes(&source).unwrap();
         assert_eq!(
             res,
-            vec![
-                IndexManifestEntry {
-                    version: 1,
-                    kind: FileKind::Add,
-                    partition: vec![0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
-                    bucket: 0,
-                    index_file: IndexFileMeta {
-                        index_type: "HASH".into(),
-                        file_name: "index-a984b43a-c3fb-40b4-ad29-536343c239a6-0".into(),
-                        file_size: 16,
-                        row_count: 4,
-                        deletion_vectors_ranges: None,
-                    }
-                },
-                IndexManifestEntry {
-                    version: 1,
-                    kind: FileKind::Add,
-                    partition: vec![0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
-                    bucket: 0,
-                    index_file: IndexFileMeta {
-                        index_type: "DELETION_VECTORS".into(),
-                        file_name: "index-3f0986c5-4398-449b-be82-95f019d7a748-0".into(),
-                        file_size: 33,
-                        row_count: 1,
-                        deletion_vectors_ranges: Some(IndexMap::from([(
-                            "data-9b76122c-6bb5-4952-a946-b5bce29694a1-0.orc".into(),
-                            (1, 24)
-                        )])),
-                    }
+            vec![IndexManifestEntry {
+                version: 1,
+                kind: FileKind::Add,
+                partition: vec![0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
+                bucket: 0,
+                index_file: IndexFileMeta {
+                    index_type: "DELETION_VECTORS".into(),
+                    file_name: "index-4326356b-aad7-4fd8-9d88-2bb6993c8ce9-0".into(),
+                    file_size: 35,
+                    row_count: 1,
+                    deletion_vectors_ranges: Some(IndexMap::from([(
+                        "data-a989fc44-a361-42c2-801f-e50baba95a92-0.parquet".into(),
+                        DeletionVectorMeta {
+                            offset: 1,
+                            length: 26,
+                            cardinality: Some(3),
+                        }
+                    )])),
                 }
-            ]
+            }]
         );
     }
 
@@ -153,7 +137,14 @@ mod tests {
                 file_name: "test1".into(),
                 file_size: 33,
                 row_count: 1,
-                deletion_vectors_ranges: Some(IndexMap::from([("test1".into(), (1, 24))])),
+                deletion_vectors_ranges: Some(IndexMap::from([(
+                    "test1".into(),
+                    DeletionVectorMeta {
+                        offset: 1,
+                        length: 24,
+                        cardinality: Some(7),
+                    },
+                )])),
             },
         };
 
@@ -180,7 +171,8 @@ mod tests {
                             "fields": [
                                 {"name": "f0", "type": "string"}, 
                                 {"name": "f1", "type": "int"}, 
-                                {"name": "f2", "type": "int"}
+                                {"name": "f2", "type": "int"},
+                                {"name": "_CARDINALITY", "type": ["null", "long"], "default": null}
                             ]
                         }]
                     }]

crates/paimon/src/table/source.rs

@@ -168,18 +168,6 @@ impl DataSplit {
         format!("{}/{}", base, file.file_name)
     }
 
-    /// Iterate over each data file in this split, yielding `(path, &DataFileMeta)`.
-    /// Use this to read each data file one by one (e.g. in ArrowReader).
-    pub fn data_file_entries(&self) -> impl Iterator<Item = (String, &DataFileMeta)> + '_ {
-        let base = self.bucket_path.trim_end_matches('/');
-        // todo: consider partition table
-        // todo: consider external path
-        self.data_files.iter().map(move |file| {
-            let path = format!("{}/{}", base, file.file_name);
-            (path, file)
-        })
-    }
-
     /// Total row count of all data files in this split.
     pub fn row_count(&self) -> i64 {
         self.data_files.iter().map(|f| f.row_count).sum()

crates/paimon/src/table/table_scan.rs

@@ -113,15 +113,14 @@ fn build_deletion_files_map(
         let key = PartitionBucket::new(entry.partition.clone(), entry.bucket);
         let dv_path = format!("{}/{}", index_path_prefix, entry.index_file.file_name);
         let per_bucket = map.entry(key).or_default();
-        for (data_file_name, (offset, length)) in ranges {
+        for (data_file_name, meta) in ranges {
             per_bucket.insert(
                 data_file_name.clone(),
                 DeletionFile::new(
                     dv_path.clone(),
-                    *offset as i64,
-                    *length as i64,
-                    // todo: consider cardinality
-                    None,
+                    meta.offset as i64,
+                    meta.length as i64,
+                    meta.cardinality,
                 ),
             );
         }
@@ -513,9 +512,11 @@ impl<'a> TableScan<'a> {
 mod tests {
     use super::{group_by_overlapping_row_id, partition_matches_predicate};
     use crate::spec::{
-        stats::BinaryTableStats, ArrayType, DataField, DataFileMeta, DataType, Datum, IntType,
-        Predicate, PredicateBuilder, PredicateOperator, VarCharType,
+        stats::BinaryTableStats, ArrayType, DataField, DataFileMeta, DataType, Datum,
+        DeletionVectorMeta, FileKind, IndexFileMeta, IndexManifestEntry, IntType, Predicate,
+        PredicateBuilder, PredicateOperator, VarCharType,
     };
+    use crate::table::source::DeletionFile;
     use crate::Error;
     use chrono::{DateTime, Utc};
 
@@ -718,4 +719,42 @@ mod tests {
         // Sorted by descending max_sequence_number: b(3), c(2), a(1)
         assert_eq!(file_names(&groups), vec![vec!["b", "c", "a"]]);
     }
+
+    #[test]
+    fn test_build_deletion_files_map_preserves_cardinality() {
+        let entries = vec![IndexManifestEntry {
+            version: 1,
+            kind: FileKind::Add,
+            partition: vec![1, 2, 3],
+            bucket: 7,
+            index_file: IndexFileMeta {
+                index_type: "DELETION_VECTORS".into(),
+                file_name: "index-file".into(),
+                file_size: 128,
+                row_count: 1,
+                deletion_vectors_ranges: Some(indexmap::IndexMap::from([(
+                    "data-file.parquet".into(),
+                    DeletionVectorMeta {
+                        offset: 11,
+                        length: 22,
+                        cardinality: Some(33),
+                    },
+                )])),
+            },
+        }];
+
+        let map = super::build_deletion_files_map(&entries, "file:/tmp/table");
+
+        let by_bucket = map
+            .get(&super::PartitionBucket::new(vec![1, 2, 3], 7))
+            .expect("partition bucket should exist");
+        let deletion_file = by_bucket
+            .get("data-file.parquet")
+            .expect("deletion file should exist");
+
+        assert_eq!(
+            deletion_file,
+            &DeletionFile::new("file:/tmp/table/index/index-file".into(), 11, 22, Some(33))
+        );
+    }
 }