feat(table): read materialized deletion-vector PK splits

crates/paimon/src/table/read_builder.rs

@@ -639,14 +639,17 @@ mod tests {
     }
 
     use crate::catalog::Identifier;
+    use crate::deletion_vector::DeletionVector;
     use crate::io::FileIOBuilder;
     use crate::spec::{
         BinaryRow, DataField, DataType, IntType, Predicate, PredicateBuilder, Schema, TableSchema,
         VarCharType,
     };
-    use crate::table::{query_auth_table, DataSplitBuilder, Table};
+    use crate::table::{query_auth_table, DataSplitBuilder, DeletionFile, Table};
     use arrow_array::{Int32Array, RecordBatch};
+    use bytes::Bytes;
     use futures::TryStreamExt;
+    use roaring::RoaringBitmap;
     use std::collections::{HashMap, HashSet};
     use std::fs;
     use std::sync::atomic::{AtomicUsize, Ordering};
@@ -675,6 +678,31 @@ mod tests {
             .collect()
     }
 
+    async fn write_test_deletion_file(
+        file_io: &crate::io::FileIO,
+        path: &str,
+        deleted_rows: &[u32],
+    ) -> DeletionFile {
+        let bitmap = deleted_rows.iter().copied().collect::<RoaringBitmap>();
+        let bytes = DeletionVector::from_bitmap(bitmap)
+            .serialize_to_bytes()
+            .unwrap();
+        let bitmap_length = i32::from_be_bytes(bytes[0..4].try_into().unwrap());
+        file_io
+            .new_output(path)
+            .unwrap()
+            .write(Bytes::from(bytes))
+            .await
+            .unwrap();
+
+        DeletionFile::new(
+            path.to_string(),
+            0,
+            bitmap_length as i64,
+            Some(deleted_rows.len() as i64),
+        )
+    }
+
     fn simple_table() -> Table {
         let file_io = FileIOBuilder::new("file").build().unwrap();
         let table_schema = TableSchema::new(
@@ -695,28 +723,70 @@ mod tests {
         )
     }
 
-    fn partial_update_dv_pk_table() -> Table {
+    fn dv_pk_table(table_path: &str, merge_engine: &str) -> Table {
         let file_io = FileIOBuilder::new("file").build().unwrap();
         let table_schema = TableSchema::new(
             0,
             &Schema::builder()
                 .column("id", DataType::Int(IntType::new()))
                 .column("value", DataType::Int(IntType::new()))
                 .primary_key(["id"])
-                .option("merge-engine", "partial-update")
+                .option("merge-engine", merge_engine)
                 .option("deletion-vectors.enabled", "true")
                 .build()
                 .unwrap(),
         );
         Table::new(
             file_io,
             Identifier::new("default", "partial_update_dv_t"),
-            "/tmp/test-partial-update-dv-read-builder".to_string(),
+            table_path.to_string(),
             table_schema,
             None,
         )
     }
 
+    async fn read_compacted_dv_table(merge_engine: &str) -> Vec<RecordBatch> {
+        let tempdir = tempdir().unwrap();
+        let table_path = local_file_path(tempdir.path());
+        let bucket_dir = tempdir.path().join("bucket-0");
+        let index_dir = tempdir.path().join("index");
+        fs::create_dir_all(&bucket_dir).unwrap();
+        fs::create_dir_all(&index_dir).unwrap();
+
+        let data_path = bucket_dir.join("data.parquet");
+        write_int_parquet_file(
+            &data_path,
+            vec![("id", vec![1, 2, 3]), ("value", vec![10, 20, 30])],
+            None,
+        );
+        let file_size = fs::metadata(&data_path).unwrap().len() as i64;
+        let file_io = FileIOBuilder::new("file").build().unwrap();
+        let deletion_file =
+            write_test_deletion_file(&file_io, &local_file_path(&index_dir.join("dv")), &[1]).await;
+
+        let table = dv_pk_table(&table_path, merge_engine);
+        let mut data_file =
+            test_data_file::<crate::spec::DataFileMeta>("data.parquet", 3, file_size);
+        data_file.delete_row_count = Some(0);
+        let split = DataSplitBuilder::new()
+            .with_snapshot(1)
+            .with_partition(BinaryRow::new(0))
+            .with_bucket(0)
+            .with_bucket_path(local_file_path(&bucket_dir))
+            .with_total_buckets(1)
+            .with_data_files(vec![data_file])
+            .with_data_deletion_files(vec![Some(deletion_file)])
+            .build()
+            .unwrap();
+
+        TableRead::new(&table, table.schema().fields().to_vec(), Vec::new())
+            .to_arrow(&[split])
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap()
+    }
+
     #[test]
     fn test_read_fails_closed_when_query_auth_enabled() {
         let table = query_auth_table();
@@ -1540,33 +1610,50 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn test_direct_table_read_rejects_partial_update_with_deletion_vectors() {
-        let table = partial_update_dv_pk_table();
+    async fn test_direct_table_read_reads_compacted_partial_update_with_deletion_vectors() {
+        let batches = read_compacted_dv_table("partial-update").await;
+
+        assert_eq!(collect_int_column(&batches, "id"), vec![1, 3]);
+        assert_eq!(collect_int_column(&batches, "value"), vec![10, 30]);
+    }
+
+    #[tokio::test]
+    async fn test_direct_table_read_reads_compacted_aggregation_with_deletion_vectors() {
+        let batches = read_compacted_dv_table("aggregation").await;
+
+        assert_eq!(collect_int_column(&batches, "id"), vec![1, 3]);
+        assert_eq!(collect_int_column(&batches, "value"), vec![10, 30]);
+    }
+
+    #[test]
+    fn test_direct_table_read_rejects_partial_update_dv_merge_on_read() {
+        let table = dv_pk_table(
+            "/tmp/test-partial-update-dv-merge-on-read",
+            "partial-update",
+        )
+        .copy_with_options(HashMap::from([(
+            "deletion-vectors.merge-on-read".to_string(),
+            "true".to_string(),
+        )]));
+        let mut data_file = test_data_file::<crate::spec::DataFileMeta>("data.parquet", 1, 0);
+        data_file.delete_row_count = Some(0);
         let split = DataSplitBuilder::new()
             .with_snapshot(1)
             .with_partition(BinaryRow::new(0))
             .with_bucket(0)
-            .with_bucket_path("/tmp/test-partial-update-dv-read-builder/bucket-0".to_string())
+            .with_bucket_path("/tmp/test-partial-update-dv-merge-on-read/bucket-0".to_string())
             .with_total_buckets(1)
-            .with_data_files(vec![test_data_file("data.parquet", 1, 0)])
-            .with_data_deletion_files(vec![Some(crate::table::source::DeletionFile::new(
-                "/tmp/test-partial-update-dv-read-builder/index/dv".to_string(),
-                0,
-                0,
-                None,
-            ))])
+            .with_data_files(vec![data_file])
             .build()
             .unwrap();
-        let err = TableRead::new(&table, table.schema().fields().to_vec(), Vec::new())
-            .to_arrow(&[split])
-            .unwrap()
-            .try_collect::<Vec<_>>()
-            .await
-            .unwrap_err();
 
-        assert!(
-            matches!(err, crate::Error::Unsupported { ref message } if message.contains("deletion vectors")),
-            "expected partial-update+DV read to fail fast with Unsupported, got {err:?}"
-        );
+        let result =
+            TableRead::new(&table, table.schema().fields().to_vec(), Vec::new()).to_arrow(&[split]);
+
+        assert!(matches!(
+            result,
+            Err(crate::Error::Unsupported { ref message })
+                if message.contains("merge-on-read")
+        ));
     }
 }

crates/paimon/src/table/source.rs

@@ -525,6 +525,20 @@ impl DataSplit {
         self.raw_convertible
     }
 
+    /// Whether this primary-key split is safe to read raw when deletion
+    /// vectors provide row-level filtering.
+    ///
+    /// This is deliberately stronger than [`Self::raw_convertible`]: direct
+    /// or deserialized splits may bypass scan planning, so every file must
+    /// also be compacted and known not to contain retract rows.
+    pub(crate) fn is_fully_materialized_pk_dv(&self) -> bool {
+        self.raw_convertible
+            && self
+                .data_files
+                .iter()
+                .all(|file| file.level != 0 && file.delete_row_count == Some(0))
+    }
+
     /// Returns the deletion file for the data file at the given index, if any. `None` at that index means no deletion file.
     pub fn deletion_file_for_data_file_index(&self, index: usize) -> Option<&DeletionFile> {
         self.data_deletion_files
@@ -1318,6 +1332,36 @@ mod tests {
         assert_eq!(s.merged_row_count(), Some(15));
     }
 
+    #[test]
+    fn test_fully_materialized_pk_dv_requires_compacted_files() {
+        let mut level_zero = file("a", 10, None);
+        level_zero.level = 0;
+        level_zero.delete_row_count = Some(0);
+
+        assert!(!split(vec![level_zero], true).is_fully_materialized_pk_dv());
+    }
+
+    #[test]
+    fn test_fully_materialized_pk_dv_requires_raw_convertible() {
+        let mut compacted = file("a", 10, None);
+        compacted.delete_row_count = Some(0);
+
+        assert!(!split(vec![compacted], false).is_fully_materialized_pk_dv());
+    }
+
+    #[test]
+    fn test_fully_materialized_pk_dv_requires_known_zero_delete_rows() {
+        let unknown = file("unknown", 10, None);
+        let mut retracts = file("retracts", 10, None);
+        retracts.delete_row_count = Some(1);
+        let mut materialized = file("materialized", 10, None);
+        materialized.delete_row_count = Some(0);
+
+        assert!(!split(vec![unknown], true).is_fully_materialized_pk_dv());
+        assert!(!split(vec![retracts], true).is_fully_materialized_pk_dv());
+        assert!(split(vec![materialized], true).is_fully_materialized_pk_dv());
+    }
+
     #[test]
     fn test_data_file_path_prefers_external_path() {
         let mut f = file("data-0.parquet", 10, None);

crates/paimon/src/table/table_read.rs

@@ -369,13 +369,9 @@ impl<'a> PaimonTableRead<'a> {
         core_options.ensure_read_authorized()?;
         let merge_engine = core_options.merge_engine()?;
 
-        // PK table with Deduplicate engine: splits that may hold multiple
-        // versions of a key need KeyValueFileReader for sort-merge dedup;
-        // splits marked raw convertible by scan planning — and all compacted
-        // files of deletion-vector tables, where DVs mask stale versions —
-        // use the faster DataFileReader.
-        // PartialUpdate / Aggregation always go through KeyValueFileReader so
-        // that per-key materialization can run on the read side.
+        // Route supported PK merge engines through the split-aware reader.
+        // Deduplicate may mix raw and KV splits. Partial-update and aggregation
+        // use KV reads normally, but fully materialized DV plans can read raw.
         if has_primary_keys
             && matches!(
                 merge_engine,
@@ -395,28 +391,57 @@ impl<'a> PaimonTableRead<'a> {
     /// Read PK table. For `Deduplicate`, splits marked raw convertible by scan
     /// planning (mirrors Java `DataSplit#convertToRawFiles`) use the faster
     /// DataFileReader; the rest go through KeyValueFileReader for sort-merge
-    /// dedup. Deletion-vector tables are exempt: their stale versions are
-    /// masked by DVs, and KeyValueFileReader does not support DVs, so they keep
-    /// the plain level-0 dispatch. `PartialUpdate` and `Aggregation` always go
-    /// through KeyValueFileReader because their merge semantics require per-key
-    /// materialization even for compacted runs.
+    /// dedup. A fully materialized deletion-vector plan for `PartialUpdate` or
+    /// `Aggregation` can also be read raw because DVs already mask stale rows.
+    /// Plans that still need any per-key merge fail closed because mixing raw
+    /// and merged outputs would produce incorrect results.
     fn read_pk(
         &self,
         data_splits: &[DataSplit],
         core_options: &CoreOptions,
     ) -> crate::Result<ArrowRecordBatchStream> {
+        let merge_engine = core_options.merge_engine()?;
+        let dv_enabled = core_options.deletion_vectors_enabled();
         if matches!(
-            core_options.merge_engine()?,
+            merge_engine,
             MergeEngine::PartialUpdate | MergeEngine::Aggregation
-        ) {
+        ) && !dv_enabled
+        {
             return self.read_kv(data_splits, core_options);
         }
 
+        if matches!(
+            merge_engine,
+            MergeEngine::PartialUpdate | MergeEngine::Aggregation
+        ) {
+            let merge_engine_name = match merge_engine {
+                MergeEngine::PartialUpdate => "partial-update",
+                MergeEngine::Aggregation => "aggregation",
+                _ => unreachable!("guarded by partial-update/aggregation match"),
+            };
+            if core_options.deletion_vectors_merge_on_read() {
+                return Err(crate::Error::Unsupported {
+                    message: format!(
+                        "merge-engine={merge_engine_name} with deletion-vectors.merge-on-read=true is not supported"
+                    ),
+                });
+            }
+            if !data_splits
+                .iter()
+                .all(DataSplit::is_fully_materialized_pk_dv)
+            {
+                return Err(crate::Error::Unsupported {
+                    message: format!(
+                        "merge-engine={merge_engine_name} with deletion vectors can only read fully materialized compacted splits"
+                    ),
+                });
+            }
+            return self.read_raw(data_splits);
+        }
+
         // Deletion-vector tables read raw by design: stale versions of a key
         // are masked by DVs, not merged, and KeyValueFileReader does not
         // support DVs. Keep the plain level-0 dispatch for them.
-        let dv_enabled = core_options.deletion_vectors_enabled();
-
         let mut kv_splits = Vec::new();
         let mut raw_splits = Vec::new();
         for split in data_splits {

docs/src/sql.md

@@ -1889,6 +1889,13 @@ an existing partial-update table, `ignore-delete` cannot be changed back to
 `false`. Advanced partial-update features such as sequence groups, partial
 aggregation, and remove-record-on-delete are not supported.
 
+Rust can read fully materialized compacted files from deletion-vector-enabled
+partial-update and aggregation tables. Every split must be raw-convertible,
+every file must have a known zero delete-row count, and
+`deletion-vectors.merge-on-read` must remain disabled. Writing these table
+combinations and reading plans that still require per-key merging are not
+supported.
+
 Rust currently supports `merge-engine=aggregation` in basic mode only. It works
 with fixed buckets and ordinary dynamic buckets (`'bucket' = '-1'`) when the
 primary key includes all partition columns. It supports per-field aggregate
@@ -1900,9 +1907,10 @@ Sequence fields are always merged with `last_value`. Defining
 validation.
 
 This is not full Java feature parity. Aggregation tables do not support retract
-rows (`DELETE` / `UPDATE_BEFORE`), deletion vectors, cross-partition dynamic
-bucket writes, or advanced aggregation options such as `ignore-retract`,
-`distinct`, `nested-key`, `count-limit`, and sequence groups.
+rows (`DELETE` / `UPDATE_BEFORE`), deletion-vector writes or merge-on-read
+plans, cross-partition dynamic bucket writes, or advanced aggregation options
+such as `ignore-retract`, `distinct`, `nested-key`, `count-limit`, and sequence
+groups.
 
 ### Global Index Options