From ab297309a2913bcf1794c3ba843a7b7e90f9a765 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Wed, 12 Aug 2026 18:54:57 +0800 Subject: [PATCH] feat: support deletion vector merge-on-read --- .../datafusion/tests/pk_tables.rs | 69 +++- crates/paimon/src/spec/schema.rs | 41 +++ crates/paimon/src/table/kv_file_reader.rs | 348 +++++++++++++++++- crates/paimon/src/table/table_read.rs | 20 +- crates/paimon/src/table/table_scan.rs | 98 ++++- docs/src/sql.md | 9 + 6 files changed, 546 insertions(+), 39 deletions(-) diff --git a/crates/integrations/datafusion/tests/pk_tables.rs b/crates/integrations/datafusion/tests/pk_tables.rs index 0f44c8197..83a24d24f 100644 --- a/crates/integrations/datafusion/tests/pk_tables.rs +++ b/crates/integrations/datafusion/tests/pk_tables.rs @@ -2110,11 +2110,8 @@ async fn test_pk_partitioned_fixed_bucket_predicate_query() { // ======================= DV + Deduplicate Regression ======================= -/// Regression: DV-enabled Deduplicate PK table must not error on read. -/// Before the fix, removing the DV guard caused level-0 files to reach -/// KeyValueFileReader which rejects deletion-vector files with a hard error. -/// With the guard restored, level-0 files are skipped in scan (DV mode relies -/// on compaction to produce higher-level files). +/// DV-enabled Deduplicate PK tables keep the existing compacted-only behavior +/// unless merge-on-read is explicitly enabled. #[tokio::test] async fn test_pk_dv_deduplicate_read_no_error() { let (_tmp, sql_context) = setup_sql_context().await; @@ -2146,11 +2143,8 @@ async fn test_pk_dv_deduplicate_read_no_error() { .await .unwrap(); - // Read must not error. DV mode skips level-0 files, so only compacted - // (level > 0) files are visible. Without compaction, all files are level-0 - // and get skipped — count may be 0, but the read must succeed without error. - // Before the fix, this would hard-fail with "KeyValueFileReader does not - // support deletion vectors". + // The default remains compacted-only: level-0 files are skipped, so the + // read may be empty before compaction but must remain valid. let result = sql_context .sql("SELECT * FROM paimon.test_db.t_dv_dedup") .await @@ -2164,6 +2158,61 @@ async fn test_pk_dv_deduplicate_read_no_error() { ); } +/// Merge-on-read makes level-0 files visible and resolves overlapping primary +/// keys before applying the residual predicate. +#[tokio::test] +async fn test_pk_dv_merge_on_read_reads_level_zero() { + let (_tmp, sql_context) = setup_sql_context().await; + + sql_context + .sql( + "CREATE TABLE paimon.test_db.t_dv_mor ( + id INT NOT NULL, value INT, + PRIMARY KEY (id) + ) WITH ( + 'bucket' = '1', + 'deletion-vectors.enabled' = 'true', + 'deletion-vectors.merge-on-read' = 'true', + 'source.split.target-size' = '1b', + 'source.split.open-file-cost' = '1b' + )", + ) + .await + .unwrap(); + + sql_context + .sql("INSERT INTO paimon.test_db.t_dv_mor VALUES (1, 10), (2, 20)") + .await + .unwrap() + .collect() + .await + .unwrap(); + sql_context + .sql("INSERT INTO paimon.test_db.t_dv_mor VALUES (2, 200), (3, 30)") + .await + .unwrap() + .collect() + .await + .unwrap(); + + let rows = collect_id_value( + &sql_context, + "SELECT id, value FROM paimon.test_db.t_dv_mor ORDER BY id", + ) + .await; + assert_eq!(rows, vec![(1, 10), (2, 200), (3, 30)]); + + let stale_rows = collect_id_value( + &sql_context, + "SELECT id, value FROM paimon.test_db.t_dv_mor WHERE value = 20", + ) + .await; + assert!( + stale_rows.is_empty(), + "the residual predicate must run after primary-key merging" + ); +} + // ======================= Cross-Split Merge Correctness ======================= /// Regression: a 1-byte split target forces every data file into its own diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs index dd45712ef..c763f0ba0 100644 --- a/crates/paimon/src/spec/schema.rs +++ b/crates/paimon/src/spec/schema.rs @@ -3048,6 +3048,47 @@ mod tests { } } + #[test] + fn test_deletion_vector_merge_on_read_is_ignored_without_deletion_vectors() { + let schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("value", DataType::Int(IntType::new())) + .primary_key(["id"]) + .option("deletion-vectors.merge-on-read", "true") + .build() + .unwrap(); + assert_eq!( + schema + .options() + .get("deletion-vectors.merge-on-read") + .map(String::as_str), + Some("true") + ); + + let table_schema = TableSchema::new( + 0, + &Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("value", DataType::Int(IntType::new())) + .primary_key(["id"]) + .build() + .unwrap(), + ); + let changed = table_schema + .apply_changes(vec![crate::spec::SchemaChange::set_option( + "deletion-vectors.merge-on-read".to_string(), + "true".to_string(), + )]) + .unwrap(); + assert_eq!( + changed + .options() + .get("deletion-vectors.merge-on-read") + .map(String::as_str), + Some("true") + ); + } + #[test] fn test_deletion_vector_schema_validation_rejects_incompatible_changelog_producers() { for (producer, expected_message) in [ diff --git a/crates/paimon/src/table/kv_file_reader.rs b/crates/paimon/src/table/kv_file_reader.rs index ee0c3ecc4..d922c267a 100644 --- a/crates/paimon/src/table/kv_file_reader.rs +++ b/crates/paimon/src/table/kv_file_reader.rs @@ -31,6 +31,7 @@ use super::sort_merge::{ SortMergeReaderBuilder, }; use crate::arrow::{build_target_arrow_schema, ParquetReadBudget}; +use crate::deletion_vector::DeletionVectorFactory; use crate::io::FileIO; use crate::spec::{ BigIntType, DataField, DataFileMeta, DataType as PaimonDataType, MergeEngine, @@ -39,7 +40,7 @@ use crate::spec::{ }; use crate::table::schema_manager::SchemaManager; use crate::table::ArrowRecordBatchStream; -use crate::{DataSplit, Error}; +use crate::{DataSplit, DeletionFile, Error}; use arrow_array::{RecordBatch, RecordBatchOptions}; use async_stream::try_stream; @@ -529,15 +530,29 @@ impl KeyValueFileReader { Ok(try_stream! { for split_group in &split_groups { - // DV mode should not reach KeyValueFileReader. + // A deletion-vector merge-on-read split can mix compacted + // sources carrying DVs with uncompacted level-0 files. Keep + // only the small per-file metadata here; load each bitmap when + // its sorted run reaches that physical file. + let mut deletion_files_by_split = + HashMap::>>::new(); for split in split_group { - if split - .data_deletion_files() - .is_some_and(|files| files.iter().any(Option::is_some)) - { - Err(Error::Unsupported { - message: "KeyValueFileReader does not support deletion vectors".to_string(), - })?; + let Some(deletion_files) = split.data_deletion_files() else { + continue; + }; + let by_name = split + .data_files() + .iter() + .zip(deletion_files.iter()) + .filter_map(|(data_file, deletion_file)| { + deletion_file + .as_ref() + .map(|file| (data_file.file_name.clone(), file.clone())) + }) + .collect::>(); + if !by_name.is_empty() { + deletion_files_by_split + .insert(Arc::as_ptr(split) as usize, Arc::new(by_name)); } } for merge_group in plan_merge_groups( @@ -570,6 +585,8 @@ impl KeyValueFileReader { .with_batch_size(Some(read_batch_size)) .with_parquet_read_budget(group_parquet_read_budget.clone()); let run_schema_manager = schema_manager.clone(); + let run_file_io = file_io.clone(); + let deletion_files_by_split = deletion_files_by_split.clone(); let run_stream: ArrowRecordBatchStream = Box::pin(try_stream! { for MergeFile { split, file: file_meta } in files { let data_fields: Option> = @@ -577,14 +594,24 @@ impl KeyValueFileReader { let data_schema = run_schema_manager.schema(file_meta.schema_id).await?; Some(data_schema.fields().to_vec()) - } else { - None - }; + } else { + None + }; + let deletion_file = deletion_files_by_split + .get(&(Arc::as_ptr(&split) as usize)) + .and_then(|files| files.get(&file_meta.file_name)) + .cloned(); + let deletion_vector = match deletion_file { + Some(file) => Some(Arc::new( + DeletionVectorFactory::read(&run_file_io, &file).await?, + )), + None => None, + }; let mut file_stream = reader.read_single_file_stream( split.as_ref(), file_meta, data_fields, - None, + deletion_vector, split.row_ranges().map(|ranges| ranges.to_vec()), )?; while let Some(batch) = file_stream.next().await { @@ -693,20 +720,24 @@ impl KeyValueFileReader { mod tests { use super::*; use crate::catalog::Identifier; + use crate::deletion_vector::DeletionVector; use crate::io::FileIOBuilder; use crate::spec::{ stats::BinaryTableStats, BinaryRow, DataFileMeta, DataType, Datum, IntType, PredicateBuilder, Schema, TableSchema, VarCharType, }; - use crate::table::source::DataSplitBuilder; + use crate::table::source::{DataSplitBuilder, DeletionFile}; use crate::table::table_commit::TableCommit; use crate::table::{Table, TableWrite}; use arrow_array::{Array, Int32Array, Int64Array, Int8Array, StringArray}; use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema as ArrowSchema}; + use bytes::Bytes; use futures::TryStreamExt; use parquet::arrow::AsyncArrowWriter; use parquet::file::metadata::ParquetMetaDataReader; use parquet::file::properties::WriterProperties; + use roaring::RoaringBitmap; + use std::collections::HashMap; use std::sync::Arc; #[tokio::test] @@ -891,6 +922,35 @@ mod tests { .unwrap(); } + async fn write_deletion_file( + file_io: &FileIO, + table_path: &str, + deleted_rows: &[u32], + ) -> DeletionFile { + let path = format!("{table_path}/index/dv"); + file_io + .mkdirs(&format!("{table_path}/index/")) + .await + .unwrap(); + let bitmap = deleted_rows.iter().copied().collect::(); + 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, + 0, + i64::from(bitmap_length), + Some(deleted_rows.len() as i64), + ) + } + async fn read_rows( table: &Table, projection: Option<&[&str]>, @@ -946,6 +1006,225 @@ mod tests { } } + /// Java-compatible DV merge-on-read is a batch visibility override: the + /// default still hides uncompacted level-0 files, while a dynamic override + /// includes them and merges overlapping key versions. A tiny split target + /// makes this also catch planners that incorrectly separate overlapping + /// files into independent raw splits. + #[tokio::test] + async fn dv_merge_on_read_exposes_and_merges_level_zero_files() { + let file_io = test_file_io(); + let table_path = "memory:/dv_merge_on_read_level_zero"; + setup_dirs(&file_io, table_path).await; + let table = pk_table( + &file_io, + table_path, + &[ + ("deletion-vectors.enabled", "true"), + ("source.split.target-size", "1b"), + ("source.split.open-file-cost", "1b"), + ], + ); + + write_commit(&table, &int_batch(vec![1, 2], vec![Some(10), Some(20)])).await; + write_commit(&table, &int_batch(vec![1, 3], vec![Some(11), Some(30)])).await; + + let hidden_plan = table.new_read_builder().new_scan().plan().await.unwrap(); + assert!( + hidden_plan.splits().is_empty(), + "DV batch reads must keep hiding level-0 files by default" + ); + + let merge_on_read = table.copy_with_options(HashMap::from([( + "deletion-vectors.merge-on-read".to_string(), + "true".to_string(), + )])); + let read_builder = merge_on_read.new_read_builder(); + let plan = read_builder.new_scan().plan().await.unwrap(); + assert_eq!( + plan.splits().len(), + 1, + "overlapping versions must share one split" + ); + assert_eq!(plan.splits()[0].data_files().len(), 2); + assert!(!plan.splits()[0].raw_convertible()); + + let batches = read_builder + .new_read() + .unwrap() + .to_arrow(plan.splits()) + .unwrap() + .try_collect::>() + .await + .unwrap(); + assert_eq!(int_column(&batches, "id"), vec![1, 2, 3]); + assert_eq!(int_column(&batches, "value"), vec![11, 20, 30]); + + let stale_filter = PredicateBuilder::new(merge_on_read.schema().fields()) + .equal("value", Datum::Int(10)) + .unwrap(); + let stale_batches = read_rows(&merge_on_read, None, Some(stale_filter)).await; + assert_eq!( + stale_batches + .iter() + .map(RecordBatch::num_rows) + .sum::(), + 0, + "a predicate matching only the superseded L0 value must not resurrect it" + ); + } + + #[tokio::test] + async fn dynamic_dv_merge_on_read_is_ignored_without_deletion_vectors() { + let file_io = test_file_io(); + let table_path = "memory:/ignored_dynamic_dv_merge_on_read"; + setup_dirs(&file_io, table_path).await; + let table = pk_table(&file_io, table_path, &[]); + write_commit(&table, &int_batch(vec![1, 2], vec![Some(10), Some(20)])).await; + write_commit(&table, &int_batch(vec![1, 3], vec![Some(11), Some(30)])).await; + + let merge_on_read = table.copy_with_options(HashMap::from([( + "deletion-vectors.merge-on-read".to_string(), + "true".to_string(), + )])); + let batches = read_rows(&merge_on_read, None, None).await; + assert_eq!(int_column(&batches, "id"), vec![1, 2, 3]); + assert_eq!(int_column(&batches, "value"), vec![11, 20, 30]); + } + + /// A MOR split can contain both uncompacted records and compacted source + /// files with deletion vectors. DV filtering must happen per physical file + /// before the surviving records enter the key merge; otherwise a deleted + /// key with no replacement is resurrected. + #[tokio::test] + async fn dv_merge_on_read_applies_deletion_vectors_before_key_merge() { + let file_io = test_file_io(); + let table_path = "memory:/dv_merge_on_read_with_dv"; + setup_dirs(&file_io, table_path).await; + let table = pk_table( + &file_io, + table_path, + &[ + ("deletion-vectors.enabled", "true"), + ("deletion-vectors.merge-on-read", "true"), + ], + ); + + write_commit(&table, &int_batch(vec![1, 2], vec![Some(10), Some(20)])).await; + write_commit(&table, &int_batch(vec![1, 3], vec![Some(11), Some(30)])).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::>(); + files.sort_by_key(|file| file.min_sequence_number); + assert_eq!(files.len(), 2); + files[0].level = 1; + + let deletion_file = write_deletion_file(&file_io, table_path, &[1]).await; + let split = 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) + .with_data_files(files) + .with_data_deletion_files(vec![Some(deletion_file), None]) + .with_raw_convertible(false) + .build() + .unwrap(); + + let batches = table + .new_read_builder() + .new_read() + .unwrap() + .to_arrow(&[split]) + .unwrap() + .try_collect::>() + .await + .unwrap(); + assert_eq!(int_column(&batches, "id"), vec![1, 3]); + assert_eq!(int_column(&batches, "value"), vec![11, 30]); + } + + /// Disjoint files in one split are consumed as sequential merge groups. A + /// DV attached to a late file must not be read before an earlier group can + /// emit its output. + #[tokio::test] + async fn dv_merge_on_read_loads_deletion_vectors_per_file() { + let file_io = test_file_io(); + let table_path = "memory:/dv_merge_on_read_lazy_dv"; + setup_dirs(&file_io, table_path).await; + let table = pk_table( + &file_io, + table_path, + &[ + ("deletion-vectors.enabled", "true"), + ("deletion-vectors.merge-on-read", "true"), + ], + ); + + let mut files = Vec::new(); + for i in 0..10 { + files.push( + write_multi_row_group_kv_file( + &file_io, + table_path, + &format!("part-{i}.parquet"), + i * 1_000, + i64::from(i), + i, + ) + .await, + ); + } + let mut deletion_files = vec![None; files.len()]; + deletion_files[9] = Some(DeletionFile::new( + format!("{table_path}/index/not-yet-read.dv"), + 0, + 1, + Some(1), + )); + let split = DataSplitBuilder::new() + .with_snapshot(1) + .with_partition(BinaryRow::new(0)) + .with_bucket(0) + .with_bucket_path(format!("{table_path}/bucket-0")) + .with_total_buckets(1) + .with_data_files(files) + .with_data_deletion_files(deletion_files) + .with_raw_convertible(false) + .build() + .unwrap(); + + let mut stream = table + .new_read_builder() + .new_read() + .unwrap() + .to_arrow(&[split]) + .unwrap(); + let first = stream + .next() + .await + .expect("the first output batch") + .expect("a late deletion vector must be loaded lazily"); + assert_eq!(first.num_rows(), 128); + + let err = stream.try_collect::>().await.unwrap_err(); + assert!( + err.to_string().contains("not-yet-read.dv"), + "the late file must still surface its missing DV when reached: {err:?}" + ); + } + fn int_key(value: i32) -> Vec { let mut builder = crate::spec::BinaryRowBuilder::new(1); builder.write_int(0, value); @@ -1171,6 +1450,47 @@ mod tests { ); } + #[tokio::test] + async fn dv_mor_table_read_bounds_merge_fan_in() { + let file_io = test_file_io(); + let table_path = "memory:/dv_mor_merge_fan_in_limit"; + let table = pk_table( + &file_io, + table_path, + &[ + ("deletion-vectors.enabled", "true"), + ("deletion-vectors.merge-on-read", "true"), + ], + ); + let split = DataSplitBuilder::new() + .with_snapshot(1) + .with_partition(BinaryRow::new(0)) + .with_bucket(0) + .with_bucket_path(format!("{table_path}/bucket-0")) + .with_total_buckets(1) + .with_data_files( + (0..257) + .map(|i| dummy_data_file(format!("file-{i}.parquet"))) + .collect(), + ) + .build() + .unwrap(); + + let err = table + .new_read_builder() + .new_read() + .unwrap() + .to_arrow(&[split]) + .unwrap() + .try_collect::>() + .await + .unwrap_err(); + assert!( + matches!(err, Error::Unsupported { message } if message.contains("sorted-run input streams")), + "DV merge-on-read must reject unbounded production merge fan-in" + ); + } + #[test] fn sorted_run_planning_limits_each_section_to_overlap_depth() { let file = |name: &str, min: i32, max: i32| { diff --git a/crates/paimon/src/table/table_read.rs b/crates/paimon/src/table/table_read.rs index c96ffab7f..b36af8411 100644 --- a/crates/paimon/src/table/table_read.rs +++ b/crates/paimon/src/table/table_read.rs @@ -41,6 +41,8 @@ use futures::{stream, StreamExt}; use std::cmp::Ordering; use std::sync::Arc; +const MAX_MERGE_INPUT_STREAMS: usize = 256; + /// Table read: reads data from splits (e.g. produced by [TableScan::plan]). /// /// Reference: [pypaimon.read.table_read.TableRead](https://github.com/apache/paimon/blob/master/paimon-python/pypaimon/read/table_read.py) @@ -684,7 +686,7 @@ impl<'a> PaimonTableRead<'a> { .collect(), read_batch_size: core_options.read_batch_size()?, merge_splits: true, - max_merge_input_streams: Some(256), + max_merge_input_streams: Some(MAX_MERGE_INPUT_STREAMS), // Diff primes the before and after streams in sequence. Keeping // a row-group permit across yielded batches can otherwise let // the first side block the second side indefinitely. @@ -772,9 +774,10 @@ impl<'a> PaimonTableRead<'a> { 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. + // 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. let mut kv_splits = Vec::new(); let mut raw_splits = Vec::new(); for split in data_splits { @@ -824,7 +827,9 @@ impl<'a> PaimonTableRead<'a> { .collect(), read_batch_size: core_options.read_batch_size()?, merge_splits: false, - max_merge_input_streams: None, + max_merge_input_streams: (core_options.deletion_vectors_enabled() + && core_options.deletion_vectors_merge_on_read()) + .then_some(MAX_MERGE_INPUT_STREAMS), parquet_read_budget: Some(self.parquet_read_budget()?), }, ); @@ -1440,8 +1445,9 @@ 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 keep the plain level-0 dispatch: stale versions are -/// masked by DVs and KeyValueFileReader does not support DVs. +/// 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. fn pk_split_needs_merge(split: &DataSplit, dv_enabled: bool) -> bool { if dv_enabled { return split.data_files().iter().any(|f| f.level == 0); diff --git a/crates/paimon/src/table/table_scan.rs b/crates/paimon/src/table/table_scan.rs index a025fe601..e4cd60eff 100644 --- a/crates/paimon/src/table/table_scan.rs +++ b/crates/paimon/src/table/table_scan.rs @@ -646,6 +646,7 @@ fn should_skip_level_zero_for_scan( scan_all_files: bool, has_primary_keys: bool, deletion_vectors_enabled: bool, + deletion_vectors_merge_on_read: bool, merge_engine: crate::Result, ) -> bool { if scan_all_files { @@ -655,7 +656,8 @@ fn should_skip_level_zero_for_scan( return false; } - deletion_vectors_enabled || merge_engine.is_ok_and(|e| e == crate::spec::MergeEngine::FirstRow) + (deletion_vectors_enabled && !deletion_vectors_merge_on_read) + || merge_engine.is_ok_and(|e| e == crate::spec::MergeEngine::FirstRow) } fn is_system_field_id(field_id: i32) -> bool { @@ -1180,6 +1182,7 @@ impl<'a> PaimonTableScan<'a> { self.scan_all_files, has_primary_keys, deletion_vectors_enabled, + core_options.deletion_vectors_merge_on_read(), core_options.merge_engine(), ); @@ -1382,8 +1385,10 @@ impl<'a> PaimonTableScan<'a> { /// `KeyValueFileReader`. /// /// Exempt (full predicates kept): - /// - Deletion-vector tables: they read raw with per-row masks, stats are - /// a superset of live rows, full pruning stays safe. + /// - Deletion-vector tables without merge-on-read: they read raw with + /// per-row masks, stats are a superset of live rows, full pruning stays + /// safe. With merge-on-read enabled, visible L0 versions require the + /// same key-only pruning rule as an ordinary PK merge read. /// - `merge-engine=first-row`: planned with `skip_level_zero` and read /// via `DataFileReader` (see `TableRead::to_arrow`), no merge on the /// read path — pruning a file drops exactly the rows the raw path's @@ -1393,13 +1398,17 @@ impl<'a> PaimonTableScan<'a> { let has_primary_keys = !self.table.schema().primary_keys().is_empty(); let core_options = CoreOptions::new(self.table.schema().options()); let deletion_vectors_enabled = core_options.deletion_vectors_enabled(); + let deletion_vectors_merge_on_read = core_options.deletion_vectors_merge_on_read(); // An unknown merge engine stays conservative (key-only pruning); the // read side fails on it anyway before returning rows. let first_row = matches!( core_options.merge_engine(), Ok(crate::spec::MergeEngine::FirstRow) ); - if has_primary_keys && !deletion_vectors_enabled && !first_row { + if has_primary_keys + && (!deletion_vectors_enabled || deletion_vectors_merge_on_read) + && !first_row + { retain_primary_key_conjuncts( &self.data_predicates, self.table.schema().fields(), @@ -1873,10 +1882,12 @@ impl<'a> PaimonTableScan<'a> { // sort-merge reader sees every version of a key. The comparator decodes // the trimmed-PK min/max keys written by the kv writer. // - // Deletion-vector and first-row tables read without merging (stale rows - // are masked by DVs / level-0 is skipped), so they keep plain size-based - // packing like Java's MergeTreeSplitGenerator fast path. - let read_merges_overlapping_keys = !core_options.deletion_vectors_enabled() + // Deletion-vector tables without merge-on-read and first-row tables read + // without merging (stale rows are masked by DVs / level-0 is skipped), + // so they keep plain size-based packing. DV merge-on-read includes L0 + // files and must preserve overlapping key ranges just like ordinary MOR. + let read_merges_overlapping_keys = (!core_options.deletion_vectors_enabled() + || core_options.deletion_vectors_merge_on_read()) && !matches!( core_options.merge_engine(), Ok(crate::spec::MergeEngine::FirstRow) @@ -2799,6 +2810,7 @@ mod tests { false, true, false, + false, Ok(crate::spec::MergeEngine::FirstRow), )); } @@ -2809,10 +2821,29 @@ mod tests { true, true, false, + false, Ok(crate::spec::MergeEngine::FirstRow), )); } + #[test] + fn test_dv_merge_on_read_controls_batch_level_zero_visibility() { + assert!(should_skip_level_zero_for_scan( + false, + true, + true, + false, + Ok(crate::spec::MergeEngine::Deduplicate), + )); + assert!(!should_skip_level_zero_for_scan( + false, + true, + true, + true, + Ok(crate::spec::MergeEngine::Deduplicate), + )); + } + #[test] fn test_partition_filter_decode_failure_fails_open() { let fields = partition_string_field(); @@ -3592,6 +3623,57 @@ mod tests { ); } + /// Enabling DV merge-on-read puts the table back on a key-merge path for + /// visible level-0 files. Non-key stats pruning is therefore unsafe for the + /// same reason as ordinary MOR: pruning the newest version can resurrect an + /// older matching value. + #[tokio::test] + async fn test_dv_merge_on_read_stats_pruning_ignores_non_key_conjuncts() { + let table_path = "memory:/test_dv_mor_stats_gate"; + let table = pk_stats_gate_table(table_path).copy_with_options(HashMap::from([ + ("deletion-vectors.enabled".to_string(), "true".to_string()), + ( + "deletion-vectors.merge-on-read".to_string(), + "true".to_string(), + ), + ])); + setup_scan_trace_dirs(&table).await; + + let mut old = pk_stats_file("old-version.parquet", (1, 5), (100, 200)); + old.level = 1; + let mut new = pk_stats_file("new-version.parquet", (1, 5), (10, 60)); + new.level = 1; + TableCommit::new(table.clone(), "dv-mor-gate-test".to_string()) + .commit(vec![CommitMessage::new( + BinaryRowBuilder::new(0).build_serialized(), + 0, + vec![old, new], + )]) + .await + .unwrap(); + + let fields = vec![ + DataField::new(0, "id".to_string(), DataType::Int(IntType::new())), + DataField::new(1, "value".to_string(), DataType::Int(IntType::new())), + ]; + let value_filter = PredicateBuilder::new(&fields) + .greater_than("value", Datum::Int(90)) + .unwrap(); + let mut reader = table.new_read_builder(); + reader.with_filter(value_filter); + let (plan, trace) = reader.new_scan().plan_with_trace().await.unwrap(); + + assert_eq!(trace.manifest_entries_pruned_by_data_stats, 0); + assert_eq!( + plan.splits() + .iter() + .map(|split| split.data_files().len()) + .sum::(), + 2, + "both key versions must reach the merge path" + ); + } + /// `merge-engine=first-row` PK tables read raw (no merge on the read /// path: planned with `skip_level_zero`, read via `DataFileReader`), so /// pruning a file by a non-key conjunct cannot resurrect anything — it diff --git a/docs/src/sql.md b/docs/src/sql.md index 34aa36a49..6eea6d228 100644 --- a/docs/src/sql.md +++ b/docs/src/sql.md @@ -1982,6 +1982,14 @@ Set via `WITH ('key' = 'value')` at table creation time, or dynamically via `SET | `'merge-engine' = 'partial-update'` | Basic partial-update engine for PK tables | | `'merge-engine' = 'aggregation'` | Basic aggregation engine for PK tables | +For deletion-vector-enabled primary-key tables using the default `deduplicate` +engine, batch scans hide uncompacted level-0 files by default. Set +`'deletion-vectors.merge-on-read' = 'true'` to include those files and merge +their key versions on read. Existing deletion vectors are applied before the +key merge. This option affects batch snapshot reads only; it does not change +streaming or changelog behavior. It takes effect only when +`'deletion-vectors.enabled' = 'true'`; otherwise it is ignored. + Rust supports the basic partial-update engine with latest-non-null semantics. Set either `'ignore-delete' = 'true'` or `'partial-update.ignore-delete' = 'true'` to ignore `DELETE` and @@ -2086,6 +2094,7 @@ the normal physical format without wrapping the writer. | `'data-evolution.enabled' = 'true'` | Enable data evolution (partial-column writes, row-level UPDATE/MERGE/DELETE) | | `'global-index.enabled' = 'true'` | Enable global index metadata and reads | | `'deletion-vectors.enabled' = 'true'` | Enable deletion vectors | +| `'deletion-vectors.merge-on-read' = 'true'` | Include and key-merge uncompacted level-0 files in DV-enabled deduplicate batch reads | | `'changelog-producer' = 'input'` | Changelog producer; primary-key tables support reads and writes in this mode | Cross-partition updates are not configured by an option: a primary-key table is