This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 1aa6c493 Support PK partial-update and aggregation reads with deletion
vectors (#905)
1aa6c493 is described below
commit 1aa6c493b9a2075b4df05c801366a200c56c1ffa
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Sep 21 19:06:12 2026 +0800
Support PK partial-update and aggregation reads with deletion vectors (#905)
---
crates/paimon/src/table/kv_file_reader.rs | 164 ++++++++++++++++++++++++++++++
crates/paimon/src/table/read_builder.rs | 51 ++++------
crates/paimon/src/table/table_read.rs | 63 ++++--------
3 files changed, 202 insertions(+), 76 deletions(-)
diff --git a/crates/paimon/src/table/kv_file_reader.rs
b/crates/paimon/src/table/kv_file_reader.rs
index 9b2a868f..cfc265bc 100644
--- a/crates/paimon/src/table/kv_file_reader.rs
+++ b/crates/paimon/src/table/kv_file_reader.rs
@@ -1077,6 +1077,170 @@ mod tests {
);
}
+ #[tokio::test]
+ async fn dv_merge_on_read_supports_partial_update_and_aggregation() {
+ for (engine, options, second_value, expected) in [
+ ("partial-update", vec![], None, 10),
+ (
+ "aggregation",
+ vec![("fields.value.aggregate-function", "sum")],
+ Some(5),
+ 15,
+ ),
+ ] {
+ let file_io = test_file_io();
+ let table_path = format!("memory:/dv_merge_on_read_{engine}");
+ setup_dirs(&file_io, &table_path).await;
+ let mut table_options = vec![
+ ("merge-engine", engine),
+ ("source.split.target-size", "1b"),
+ ("source.split.open-file-cost", "1b"),
+ ];
+ table_options.extend(options);
+ let table = pk_table(&file_io, &table_path, &table_options);
+
+ write_commit(&table, &int_batch(vec![1, 2], vec![Some(10),
Some(20)])).await;
+ write_commit(&table, &int_batch(vec![1, 3], vec![second_value,
Some(30)])).await;
+
+ // The Rust writer does not yet support DV compaction for these
+ // merge engines. Read the real L0 files through a DV-enabled view
+ // of the same table, as a Java-written table would be read.
+ let read_table = table.copy_with_options(HashMap::from([
+ ("deletion-vectors.enabled".to_string(), "true".to_string()),
+ (
+ "deletion-vectors.merge-on-read".to_string(),
+ "true".to_string(),
+ ),
+ ]));
+
+ let plan = read_table
+ .new_read_builder()
+ .new_scan()
+ .plan()
+ .await
+ .unwrap();
+ assert_eq!(plan.splits().len(), 1, "overlapping L0 files must
merge");
+ assert!(!plan.splits()[0].raw_convertible());
+ let batches = read_rows(&read_table, None, None).await;
+ assert_eq!(int_column(&batches, "id"), vec![1, 2, 3]);
+ assert_eq!(int_column(&batches, "value"), vec![expected, 20, 30]);
+
+ let fields = read_table.schema().fields().to_vec();
+ let merged_filter = PredicateBuilder::new(&fields)
+ .equal("value", Datum::Int(expected))
+ .unwrap();
+ let filtered = read_rows(&read_table, Some(&["id"]),
Some(merged_filter)).await;
+ assert_eq!(int_column(&filtered, "id"), vec![1]);
+ if engine == "aggregation" {
+ let input_filter = PredicateBuilder::new(&fields)
+ .equal("value", Datum::Int(10))
+ .unwrap();
+ let stale = read_rows(&read_table, None,
Some(input_filter)).await;
+ assert_eq!(
+ stale.iter().map(RecordBatch::num_rows).sum::<usize>(),
+ 0,
+ "an unmerged input value must not leak"
+ );
+ }
+ }
+ }
+
+ /// A compacted file's DV must be applied before its surviving values are
+ /// merged with an L0 file, while a materialized compacted-only split stays
+ /// on the raw path. This exercises both routes for each merge function.
+ #[tokio::test]
+ async fn partial_update_and_aggregation_apply_dv_before_merge() {
+ for (engine, options, second_value, expected) in [
+ ("partial-update", vec![], None, 10),
+ (
+ "aggregation",
+ vec![("fields.value.aggregate-function", "sum")],
+ Some(5),
+ 15,
+ ),
+ ] {
+ let file_io = test_file_io();
+ let table_path = format!("memory:/dv_compacted_and_l0_{engine}");
+ setup_dirs(&file_io, &table_path).await;
+ let mut table_options = vec![("merge-engine", engine)];
+ table_options.extend(options);
+ let table = pk_table(&file_io, &table_path, &table_options);
+ write_commit(&table, &int_batch(vec![1, 2], vec![Some(10),
Some(20)])).await;
+ write_commit(&table, &int_batch(vec![1, 3], vec![second_value,
Some(30)])).await;
+ write_commit(&table, &int_batch(vec![4], vec![Some(40)])).await;
+
+ let all_files = table
+ .new_read_builder()
+ .new_scan()
+ .with_scan_all_files()
+ .plan()
+ .await
+ .unwrap();
+ let mut files = all_files
+ .splits()
+ .iter()
+ .flat_map(|split| split.data_files().iter().cloned())
+ .collect::<Vec<_>>();
+ files.sort_by_key(|file| file.min_sequence_number);
+ assert_eq!(files.len(), 3);
+ files[0].level = 1;
+ files[2].level = 1;
+ let deletion_file = write_deletion_file(&file_io, &table_path,
&[1]).await;
+ let split_builder = || {
+ DataSplitBuilder::new()
+ .with_snapshot(2)
+ .with_partition(BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path(format!("{table_path}/bucket-0"))
+ .with_total_buckets(1)
+ };
+ let read_table = table.copy_with_options(HashMap::from([(
+ "deletion-vectors.enabled".to_string(),
+ "true".to_string(),
+ )]));
+ let read = read_table.new_read_builder().new_read().unwrap();
+
+ let raw_split = split_builder()
+ .with_data_files(vec![files[0].clone()])
+ .with_data_deletion_files(vec![Some(deletion_file.clone())])
+ .with_raw_convertible(true)
+ .build()
+ .unwrap();
+ let raw_batches = read
+ .to_arrow(&[raw_split])
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_eq!(int_column(&raw_batches, "id"), vec![1]);
+ assert_eq!(int_column(&raw_batches, "value"), vec![10]);
+
+ let merge_split = split_builder()
+ .with_data_files(files[..2].to_vec())
+ .with_data_deletion_files(vec![Some(deletion_file), None])
+ .with_raw_convertible(false)
+ .build()
+ .unwrap();
+ let disjoint_raw_split = split_builder()
+ .with_data_files(vec![files[2].clone()])
+ .with_raw_convertible(true)
+ .build()
+ .unwrap();
+ let merged_batches = read
+ .to_arrow(&[merge_split, disjoint_raw_split])
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ let mut rows = int_column(&merged_batches, "id")
+ .into_iter()
+ .zip(int_column(&merged_batches, "value"))
+ .collect::<Vec<_>>();
+ rows.sort_unstable_by_key(|row| row.0);
+ assert_eq!(rows, vec![(1, expected), (3, 30), (4, 40)]);
+ }
+ }
+
#[tokio::test]
async fn dynamic_dv_merge_on_read_is_ignored_without_deletion_vectors() {
let file_io = test_file_io();
diff --git a/crates/paimon/src/table/read_builder.rs
b/crates/paimon/src/table/read_builder.rs
index c25f789b..063b898c 100644
--- a/crates/paimon/src/table/read_builder.rs
+++ b/crates/paimon/src/table/read_builder.rs
@@ -925,7 +925,7 @@ mod tests {
)
}
- async fn read_compacted_dv_table(merge_engine: &str) -> Vec<RecordBatch> {
+ async fn read_compacted_dv_table(merge_engine: &str, merge_on_read: bool)
-> Vec<RecordBatch> {
let tempdir = tempdir().unwrap();
let table_path = local_file_path(tempdir.path());
let bucket_dir = tempdir.path().join("bucket-0");
@@ -945,6 +945,14 @@ mod tests {
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 table = if merge_on_read {
+ table.copy_with_options(HashMap::from([(
+ "deletion-vectors.merge-on-read".to_string(),
+ "true".to_string(),
+ )]))
+ } else {
+ table
+ };
let mut data_file =
test_data_file::<crate::spec::DataFileMeta>("data.parquet", 3,
file_size);
data_file.delete_row_count = Some(0);
@@ -2099,7 +2107,7 @@ mod tests {
#[tokio::test]
async fn
test_direct_table_read_reads_compacted_partial_update_with_deletion_vectors() {
- let batches = read_compacted_dv_table("partial-update").await;
+ let batches = read_compacted_dv_table("partial-update", false).await;
assert_eq!(collect_int_column(&batches, "id"), vec![1, 3]);
assert_eq!(collect_int_column(&batches, "value"), vec![10, 30]);
@@ -2107,41 +2115,18 @@ mod tests {
#[tokio::test]
async fn
test_direct_table_read_reads_compacted_aggregation_with_deletion_vectors() {
- let batches = read_compacted_dv_table("aggregation").await;
+ let batches = read_compacted_dv_table("aggregation", false).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-merge-on-read/bucket-0".to_string())
- .with_total_buckets(1)
- .with_data_files(vec![data_file])
- .build()
- .unwrap();
-
- 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")
- ));
+ #[tokio::test]
+ async fn
test_direct_table_read_supports_partial_update_and_aggregation_dv_merge_on_read()
{
+ for engine in ["partial-update", "aggregation"] {
+ let batches = read_compacted_dv_table(engine, true).await;
+ assert_eq!(collect_int_column(&batches, "id"), vec![1, 3]);
+ assert_eq!(collect_int_column(&batches, "value"), vec![10, 30]);
+ }
}
}
diff --git a/crates/paimon/src/table/table_read.rs
b/crates/paimon/src/table/table_read.rs
index ed1838d1..dde0f512 100644
--- a/crates/paimon/src/table/table_read.rs
+++ b/crates/paimon/src/table/table_read.rs
@@ -871,10 +871,9 @@ impl<'a> PaimonTableRead<'a> {
/// Read PK table. For `Deduplicate` and `FirstRow`, raw-convertible
splits from scan
/// planning (mirrors Java `DataSplit#convertToRawFiles`) use the faster
/// DataFileReader; the rest go through KeyValueFileReader for sort-merge
- /// 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.
+ /// dedup. Deletion-vector splits for any merge engine are read raw only
+ /// when their rows are fully materialized; otherwise the per-file DVs are
+ /// applied before the key merge.
fn read_pk(
&self,
data_splits: &[DataSplit],
@@ -903,39 +902,9 @@ impl<'a> PaimonTableRead<'a> {
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);
- }
-
- // Compacted deletion-vector splits read raw: their stale versions are
- // masked directly by DVs. A split containing level-0 data goes through
- // the key merge; KeyValueFileReader applies any attached per-file DVs
- // before merging the uncompacted versions.
+ // Fully materialized deletion-vector splits read raw: their stale
+ // versions are masked directly by DVs. Other splits go through the key
+ // merge; KeyValueFileReader applies any attached per-file DVs first.
let mut kv_splits = Vec::new();
let mut raw_splits = Vec::new();
for split in data_splits {
@@ -1643,12 +1612,12 @@ fn scalar_compare(
/// planning treats the missing stat as "no deletes" for compatibility, so the
/// read side must fall back to the merge reader, which drops them.
///
-/// Deletion-vector tables merge only splits containing level-0 files. Fully
-/// compacted splits stay on the raw path, while the merge reader applies any
-/// attached DVs before reconciling uncompacted key versions.
+/// Deletion-vector tables also merge any split that is not fully materialized,
+/// including level-0 data and legacy or retract-containing compacted files.
+/// The merge reader applies attached DVs before reconciling key versions.
fn pk_split_needs_merge(split: &DataSplit, dv_enabled: bool) -> bool {
if dv_enabled {
- return split.data_files().iter().any(|f| f.level == 0);
+ return !split.is_fully_materialized_pk_dv();
}
!split.raw_convertible()
|| split
@@ -1883,10 +1852,18 @@ mod tests {
let legacy = split(vec![file("a", 5, None)], true);
assert!(pk_split_needs_merge(&legacy, false));
- // Deletion-vector tables dispatch on level 0 only.
+ // DV reads can only bypass the merge when the split is known to hold
+ // fully materialized rows. Level, raw-convertibility and retract-row
+ // metadata all matter, including for caller-constructed splits.
let dv_l0 = split(vec![file("a", 0, None)], false);
assert!(pk_split_needs_merge(&dv_l0, true));
- let dv_compacted = split(vec![file("a", 5, None)], false);
+ let dv_non_raw = split(vec![file("a", 5, Some(0))], false);
+ assert!(pk_split_needs_merge(&dv_non_raw, true));
+ let dv_legacy = split(vec![file("a", 5, None)], true);
+ assert!(pk_split_needs_merge(&dv_legacy, true));
+ let dv_retracts = split(vec![file("a", 5, Some(1))], true);
+ assert!(pk_split_needs_merge(&dv_retracts, true));
+ let dv_compacted = split(vec![file("a", 5, Some(0))], true);
assert!(!pk_split_needs_merge(&dv_compacted, true));
}
