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 9a2de856 feat(table): read materialized deletion-vector PK splits
(#580)
9a2de856 is described below
commit 9a2de8562452744ce1efbf2227d4b51920690933
Author: shyjsarah <[email protected]>
AuthorDate: Wed Jul 22 11:18:45 2026 +0800
feat(table): read materialized deletion-vector PK splits (#580)
---
crates/paimon/src/table/read_builder.rs | 135 ++++++++++++++++++++++++++------
crates/paimon/src/table/source.rs | 44 +++++++++++
crates/paimon/src/table/table_read.rs | 57 ++++++++++----
docs/src/sql.md | 14 +++-
4 files changed, 207 insertions(+), 43 deletions(-)
diff --git a/crates/paimon/src/table/read_builder.rs
b/crates/paimon/src/table/read_builder.rs
index d959b0bb..c75ee528 100644
--- a/crates/paimon/src/table/read_builder.rs
+++ b/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,7 +723,7 @@ 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,
@@ -703,7 +731,7 @@ mod tests {
.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(),
@@ -711,12 +739,54 @@ mod tests {
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")
+ ));
}
}
diff --git a/crates/paimon/src/table/source.rs
b/crates/paimon/src/table/source.rs
index e852630f..a8d6b785 100644
--- a/crates/paimon/src/table/source.rs
+++ b/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);
diff --git a/crates/paimon/src/table/table_read.rs
b/crates/paimon/src/table/table_read.rs
index 2a1f70cd..6c57f6b6 100644
--- a/crates/paimon/src/table/table_read.rs
+++ b/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 {
diff --git a/docs/src/sql.md b/docs/src/sql.md
index ac3d0ee9..ac810727 100644
--- a/docs/src/sql.md
+++ b/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
