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 d8c7945d [core] Align dynamic PK events and first-row scans with Java
(#845)
d8c7945d is described below
commit d8c7945d50a17b6a6ec0b7b1311e3399c9e30893
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Sep 15 22:40:11 2026 +0800
[core] Align dynamic PK events and first-row scans with Java (#845)
---
crates/paimon/src/arrow/mod.rs | 1 +
crates/paimon/src/arrow/partition.rs | 200 +++++++++++++++++++++
crates/paimon/src/table/format_table_read.rs | 126 +-------------
crates/paimon/src/table/kv_file_reader.rs | 12 +-
crates/paimon/src/table/sort_merge.rs | 106 ++++++++++++
crates/paimon/src/table/table_read.rs | 9 +-
crates/paimon/src/table/table_scan.rs | 102 ++++++-----
crates/paimon/src/table/table_write.rs | 35 +++-
crates/paimon/tests/dynamic_bucket_scan_test.rs | 220 ++++++++++++++++++++++++
crates/paimon/tests/first_row_scan_test.rs | 162 +++++++++++++++++
10 files changed, 794 insertions(+), 179 deletions(-)
diff --git a/crates/paimon/src/arrow/mod.rs b/crates/paimon/src/arrow/mod.rs
index 0995bd45..e30d83e5 100644
--- a/crates/paimon/src/arrow/mod.rs
+++ b/crates/paimon/src/arrow/mod.rs
@@ -19,6 +19,7 @@ pub(crate) mod filtering;
pub(crate) mod format;
pub(crate) mod nested_evolution;
mod parquet_read_budget;
+pub(crate) mod partition;
pub(crate) mod residual;
mod row_filter;
pub(crate) mod schema_evolution;
diff --git a/crates/paimon/src/arrow/partition.rs
b/crates/paimon/src/arrow/partition.rs
new file mode 100644
index 00000000..5e426cab
--- /dev/null
+++ b/crates/paimon/src/arrow/partition.rs
@@ -0,0 +1,200 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Expand a serialized partition field into an Arrow column.
+
+use crate::arrow::paimon_type_to_arrow;
+use crate::spec::{extract_datum, BinaryRow, DataType, Datum};
+use crate::Error;
+use arrow_array::{
+ new_null_array, ArrayRef, BinaryArray, BooleanArray, Date32Array,
Decimal128Array,
+ Float32Array, Float64Array, Int16Array, Int32Array, Int64Array, Int8Array,
StringArray,
+ Time32MillisecondArray, TimestampMicrosecondArray,
TimestampMillisecondArray,
+ TimestampNanosecondArray,
+};
+use std::sync::Arc;
+
+pub(crate) fn partition_array(
+ partition: &BinaryRow,
+ partition_index: usize,
+ data_type: &DataType,
+ num_rows: usize,
+) -> crate::Result<ArrayRef> {
+ let arrow_type = paimon_type_to_arrow(data_type)?;
+ if partition.arity() <= partition_index as i32 ||
partition.is_null_at(partition_index) {
+ return Ok(new_null_array(&arrow_type, num_rows));
+ }
+
+ let datum = extract_datum(partition, partition_index, data_type)?;
+ let Some(datum) = datum else {
+ return Ok(new_null_array(&arrow_type, num_rows));
+ };
+
+ Ok(match (datum, data_type) {
+ (Datum::Bool(value), DataType::Boolean(_)) => {
+ Arc::new(BooleanArray::from(vec![Some(value); num_rows]))
+ }
+ (Datum::TinyInt(value), DataType::TinyInt(_)) => {
+ Arc::new(Int8Array::from(vec![Some(value); num_rows]))
+ }
+ (Datum::SmallInt(value), DataType::SmallInt(_)) => {
+ Arc::new(Int16Array::from(vec![Some(value); num_rows]))
+ }
+ (Datum::Int(value), DataType::Int(_)) => {
+ Arc::new(Int32Array::from(vec![Some(value); num_rows]))
+ }
+ (Datum::Long(value), DataType::BigInt(_)) => {
+ Arc::new(Int64Array::from(vec![Some(value); num_rows]))
+ }
+ (Datum::Float(value), DataType::Float(_)) => {
+ Arc::new(Float32Array::from(vec![Some(value); num_rows]))
+ }
+ (Datum::Double(value), DataType::Double(_)) => {
+ Arc::new(Float64Array::from(vec![Some(value); num_rows]))
+ }
+ (Datum::String(value), DataType::Char(_) | DataType::VarChar(_)) => {
+ let values = std::iter::repeat_with(|| Some(value.as_str()))
+ .take(num_rows)
+ .collect::<Vec<_>>();
+ Arc::new(StringArray::from(values))
+ }
+ (Datum::Bytes(value), DataType::Binary(_) | DataType::VarBinary(_)) =>
{
+ let values = std::iter::repeat_with(|| Some(value.as_slice()))
+ .take(num_rows)
+ .collect::<Vec<_>>();
+ Arc::new(BinaryArray::from(values))
+ }
+ (
+ Datum::Decimal {
+ unscaled, scale, ..
+ },
+ DataType::Decimal(decimal),
+ ) => Arc::new(
+ Decimal128Array::from(vec![Some(unscaled); num_rows])
+ .with_precision_and_scale(decimal.precision() as u8, scale as
i8)
+ .map_err(|error| Error::DataInvalid {
+ message: format!("Invalid decimal partition: {error}"),
+ source: Some(Box::new(error)),
+ })?,
+ ),
+ (Datum::Date(value), DataType::Date(_)) => {
+ Arc::new(Date32Array::from(vec![Some(value); num_rows]))
+ }
+ (Datum::Time(value), DataType::Time(_)) => {
+ Arc::new(Time32MillisecondArray::from(vec![Some(value); num_rows]))
+ }
+ (Datum::Timestamp { millis, nanos }, DataType::Timestamp(ts)) => {
+ timestamp_array(millis, nanos, ts.precision(), None, num_rows)?
+ }
+ (Datum::LocalZonedTimestamp { millis, nanos },
DataType::LocalZonedTimestamp(ts)) => {
+ timestamp_array(millis, nanos, ts.precision(), Some("UTC"),
num_rows)?
+ }
+ (_, other) => {
+ return Err(Error::Unsupported {
+ message: format!(
+ "Partition column type '{other:?}' is not supported by the
Rust reader yet"
+ ),
+ });
+ }
+ })
+}
+
+fn timestamp_array(
+ millis: i64,
+ nanos: i32,
+ precision: u32,
+ timezone: Option<&'static str>,
+ num_rows: usize,
+) -> crate::Result<ArrayRef> {
+ let array: ArrayRef = match precision {
+ 0..=3 => {
+ let array = TimestampMillisecondArray::from(vec![Some(millis);
num_rows]);
+ match timezone {
+ Some(tz) => Arc::new(array.with_timezone(tz)),
+ None => Arc::new(array),
+ }
+ }
+ 4..=6 => {
+ let value = millis * 1_000 + (nanos as i64) / 1_000;
+ let array = TimestampMicrosecondArray::from(vec![Some(value);
num_rows]);
+ match timezone {
+ Some(tz) => Arc::new(array.with_timezone(tz)),
+ None => Arc::new(array),
+ }
+ }
+ 7..=9 => {
+ let value = millis * 1_000_000 + (nanos as i64);
+ let array = TimestampNanosecondArray::from(vec![Some(value);
num_rows]);
+ match timezone {
+ Some(tz) => Arc::new(array.with_timezone(tz)),
+ None => Arc::new(array),
+ }
+ }
+ _ => {
+ return Err(Error::Unsupported {
+ message: format!("Unsupported timestamp precision for
partition: {precision}"),
+ });
+ }
+ };
+ Ok(array)
+}
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::spec::{
+ extract_datum_from_array, DecimalType, IntType,
LocalZonedTimestampType, VarCharType,
+ };
+
+ #[test]
+ fn partition_columns_preserve_nulls_decimals_and_timestamp_precision() {
+ let values = [
+ (None, DataType::VarChar(VarCharType::string_type())),
+ (Some(Datum::Int(-7)), DataType::Int(IntType::new())),
+ (
+ Some(Datum::Decimal {
+ unscaled: -12345678901234567890,
+ precision: 20,
+ scale: 3,
+ }),
+ DataType::Decimal(DecimalType::new(20, 3).unwrap()),
+ ),
+ (
+ Some(Datum::LocalZonedTimestamp {
+ millis: 1234,
+ nanos: 567890,
+ }),
+
DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(9).unwrap()),
+ ),
+ ];
+ let datums: Vec<_> = values
+ .iter()
+ .map(|(value, ty)| (value.as_ref(), ty))
+ .collect();
+ let partition = BinaryRow::from_datums(&datums);
+ for (index, (value, ty)) in values.iter().enumerate() {
+ let column = partition_array(&partition, index, ty, 3).unwrap();
+ assert_eq!(column.len(), 3);
+ assert_eq!(column.data_type(), &paimon_type_to_arrow(ty).unwrap());
+ for row in 0..3 {
+ assert_eq!(
+ extract_datum_from_array(&column, row, index, ty).unwrap(),
+ *value
+ );
+ }
+ }
+ }
+}
diff --git a/crates/paimon/src/table/format_table_read.rs
b/crates/paimon/src/table/format_table_read.rs
index 9f0814d6..cb6e232f 100644
--- a/crates/paimon/src/table/format_table_read.rs
+++ b/crates/paimon/src/table/format_table_read.rs
@@ -21,15 +21,11 @@ use super::data_file_reader::DataFileReader;
use super::read_builder::split_scan_predicates;
use super::table_read::configured_parquet_read_budget;
use super::{ArrowRecordBatchStream, Table};
-use crate::arrow::{build_target_arrow_schema, paimon_type_to_arrow,
ParquetReadBudget};
-use crate::spec::{extract_datum, BinaryRow, DataField, DataType, Datum,
Predicate};
+use crate::arrow::partition::partition_array;
+use crate::arrow::{build_target_arrow_schema, ParquetReadBudget};
+use crate::spec::{DataField, Predicate};
use crate::{DataSplit, Error};
-use arrow_array::{
- new_null_array, ArrayRef, BinaryArray, BooleanArray, Date32Array,
Float32Array, Float64Array,
- Int16Array, Int32Array, Int64Array, Int8Array, RecordBatch,
RecordBatchOptions, StringArray,
- Time32MillisecondArray, TimestampMicrosecondArray,
TimestampMillisecondArray,
- TimestampNanosecondArray,
-};
+use arrow_array::{RecordBatch, RecordBatchOptions};
use async_stream::try_stream;
use futures::StreamExt;
use std::sync::Arc;
@@ -278,120 +274,6 @@ fn project_format_batch(
})
}
-fn partition_array(
- partition: &BinaryRow,
- partition_index: usize,
- data_type: &DataType,
- num_rows: usize,
-) -> crate::Result<ArrayRef> {
- let arrow_type = paimon_type_to_arrow(data_type)?;
- if partition.arity() <= partition_index as i32 ||
partition.is_null_at(partition_index) {
- return Ok(new_null_array(&arrow_type, num_rows));
- }
-
- let datum = extract_datum(partition, partition_index, data_type)?;
- let Some(datum) = datum else {
- return Ok(new_null_array(&arrow_type, num_rows));
- };
-
- Ok(match (datum, data_type) {
- (Datum::Bool(value), DataType::Boolean(_)) => {
- Arc::new(BooleanArray::from(vec![Some(value); num_rows]))
- }
- (Datum::TinyInt(value), DataType::TinyInt(_)) => {
- Arc::new(Int8Array::from(vec![Some(value); num_rows]))
- }
- (Datum::SmallInt(value), DataType::SmallInt(_)) => {
- Arc::new(Int16Array::from(vec![Some(value); num_rows]))
- }
- (Datum::Int(value), DataType::Int(_)) => {
- Arc::new(Int32Array::from(vec![Some(value); num_rows]))
- }
- (Datum::Long(value), DataType::BigInt(_)) => {
- Arc::new(Int64Array::from(vec![Some(value); num_rows]))
- }
- (Datum::Float(value), DataType::Float(_)) => {
- Arc::new(Float32Array::from(vec![Some(value); num_rows]))
- }
- (Datum::Double(value), DataType::Double(_)) => {
- Arc::new(Float64Array::from(vec![Some(value); num_rows]))
- }
- (Datum::String(value), DataType::Char(_) | DataType::VarChar(_)) => {
- let values = std::iter::repeat_with(|| Some(value.as_str()))
- .take(num_rows)
- .collect::<Vec<_>>();
- Arc::new(StringArray::from(values))
- }
- (Datum::Bytes(value), DataType::Binary(_) | DataType::VarBinary(_)) =>
{
- let values = std::iter::repeat_with(|| Some(value.as_slice()))
- .take(num_rows)
- .collect::<Vec<_>>();
- Arc::new(BinaryArray::from(values))
- }
- (Datum::Date(value), DataType::Date(_)) => {
- Arc::new(Date32Array::from(vec![Some(value); num_rows]))
- }
- (Datum::Time(value), DataType::Time(_)) => {
- Arc::new(Time32MillisecondArray::from(vec![Some(value); num_rows]))
- }
- (Datum::Timestamp { millis, nanos }, DataType::Timestamp(ts)) => {
- timestamp_array(millis, nanos, ts.precision(), None, num_rows)?
- }
- (Datum::LocalZonedTimestamp { millis, nanos },
DataType::LocalZonedTimestamp(ts)) => {
- timestamp_array(millis, nanos, ts.precision(), Some("UTC"),
num_rows)?
- }
- (_, other) => {
- return Err(Error::Unsupported {
- message: format!(
- "Format table partition column type '{other:?}' is not
supported by the Rust reader yet"
- ),
- });
- }
- })
-}
-
-fn timestamp_array(
- millis: i64,
- nanos: i32,
- precision: u32,
- timezone: Option<&'static str>,
- num_rows: usize,
-) -> crate::Result<ArrayRef> {
- let array: ArrayRef = match precision {
- 0..=3 => {
- let array = TimestampMillisecondArray::from(vec![Some(millis);
num_rows]);
- match timezone {
- Some(tz) => Arc::new(array.with_timezone(tz)),
- None => Arc::new(array),
- }
- }
- 4..=6 => {
- let value = millis * 1_000 + (nanos as i64) / 1_000;
- let array = TimestampMicrosecondArray::from(vec![Some(value);
num_rows]);
- match timezone {
- Some(tz) => Arc::new(array.with_timezone(tz)),
- None => Arc::new(array),
- }
- }
- 7..=9 => {
- let value = millis * 1_000_000 + (nanos as i64);
- let array = TimestampNanosecondArray::from(vec![Some(value);
num_rows]);
- match timezone {
- Some(tz) => Arc::new(array.with_timezone(tz)),
- None => Arc::new(array),
- }
- }
- _ => {
- return Err(Error::Unsupported {
- message: format!(
- "Unsupported timestamp precision for format table
partition: {precision}"
- ),
- });
- }
- };
- Ok(array)
-}
-
fn apply_limit(batch: RecordBatch, remaining: &mut Option<usize>) ->
Option<RecordBatch> {
let Some(value) = remaining else {
return Some(batch);
diff --git a/crates/paimon/src/table/kv_file_reader.rs
b/crates/paimon/src/table/kv_file_reader.rs
index f3adf84f..8bc665e1 100644
--- a/crates/paimon/src/table/kv_file_reader.rs
+++ b/crates/paimon/src/table/kv_file_reader.rs
@@ -27,14 +27,14 @@
use super::data_file_reader::DataFileReader;
use super::sort_merge::{
- AggregateMergeFunction, DeduplicateMergeFunction, MergeFunction,
PartialUpdateMergeFunction,
- SortMergeReaderBuilder,
+ AggregateMergeFunction, DeduplicateMergeFunction, FirstRowMergeFunction,
MergeFunction,
+ PartialUpdateMergeFunction, 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,
+ BigIntType, CoreOptions, DataField, DataFileMeta, DataType as
PaimonDataType, MergeEngine,
PartialUpdateConfig, Predicate, TinyIntType, SEQUENCE_NUMBER_FIELD_ID,
SEQUENCE_NUMBER_FIELD_NAME, VALUE_KIND_FIELD_ID, VALUE_KIND_FIELD_NAME,
};
@@ -297,9 +297,9 @@ impl KeyValueFileReader {
&config.primary_keys,
)?))
}
- MergeEngine::FirstRow => Err(Error::Unsupported {
- message: "KeyValueFileReader does not support
merge-engine=first-row; first-row reads should use the non-KV path".to_string(),
- }),
+ MergeEngine::FirstRow => Ok(Box::new(FirstRowMergeFunction {
+ ignore_delete:
CoreOptions::new(&config.table_options).ignore_delete(),
+ })),
MergeEngine::Aggregation =>
Ok(Box::new(AggregateMergeFunction::new(
&config.table_options,
&config.table_name,
diff --git a/crates/paimon/src/table/sort_merge.rs
b/crates/paimon/src/table/sort_merge.rs
index e63be64a..1e88005b 100644
--- a/crates/paimon/src/table/sort_merge.rs
+++ b/crates/paimon/src/table/sort_merge.rs
@@ -182,6 +182,44 @@ impl MergeFunction for DeduplicateMergeFunction {
}
}
+/// First-row merge: keep the earliest sequence, retaining the first input on
ties.
+/// Java rejects retracts even when an earlier add has already been selected.
+pub(crate) struct FirstRowMergeFunction {
+ pub ignore_delete: bool,
+}
+
+impl MergeFunction for FirstRowMergeFunction {
+ fn merge(
+ &self,
+ rows: &[MergeRow],
+ _batch_buffer: &[BufferedBatch],
+ _source_output_col_indices: &[usize],
+ _output_schema: &SchemaRef,
+ ) -> crate::Result<MergeResult> {
+ let mut first: Option<&MergeRow> = None;
+ for row in rows {
+ if !RowKind::from_value(row.value_kind)?.is_add() {
+ if self.ignore_delete {
+ continue;
+ }
+ return Err(Error::Unsupported {
+ message: "merge-engine=first-row does not support DELETE
or UPDATE_BEFORE rows; set ignore-delete=true to ignore them".to_string(),
+ });
+ }
+ if first.is_none_or(|best| row.sequence_number <
best.sequence_number) {
+ first = Some(row);
+ }
+ }
+ Ok(match first {
+ Some(row) => MergeResult::SourceRow {
+ batch_idx: row.batch_idx,
+ row_idx: row.row_idx,
+ },
+ None => MergeResult::Omit,
+ })
+ }
+}
+
/// Partial-update merge: for each non-key column, keep the latest non-null
/// value or apply its configured field aggregator.
///
@@ -1305,6 +1343,74 @@ mod tests {
use std::collections::HashMap;
use std::sync::Arc;
+ #[test]
+ fn first_row_merge_keeps_earliest_sequence_and_first_tie() {
+ let rows: Vec<_> = [30, 10, 10, 20]
+ .into_iter()
+ .enumerate()
+ .map(|(row_idx, sequence_number)| MergeRow {
+ batch_idx: 0,
+ row_idx,
+ sequence_number,
+ value_kind: 0,
+ user_sequences: vec![],
+ })
+ .collect();
+ let result = FirstRowMergeFunction {
+ ignore_delete: false,
+ }
+ .merge(&rows, &[], &[], &make_output_schema())
+ .unwrap();
+ assert!(matches!(
+ result,
+ MergeResult::SourceRow {
+ batch_idx: 0,
+ row_idx: 1
+ }
+ ));
+ }
+
+ #[test]
+ fn first_row_merge_validates_every_retract_and_honors_ignore_delete() {
+ for kind in [1, 3] {
+ let mut rows = vec![
+ MergeRow {
+ batch_idx: 0,
+ row_idx: 0,
+ sequence_number: 1,
+ value_kind: 0,
+ user_sequences: vec![],
+ },
+ MergeRow {
+ batch_idx: 0,
+ row_idx: 1,
+ sequence_number: 2,
+ value_kind: kind,
+ user_sequences: vec![],
+ },
+ ];
+ let merge = FirstRowMergeFunction {
+ ignore_delete: false,
+ };
+ assert!(matches!(
+ merge.merge(&rows, &[], &[], &make_output_schema()),
+ Err(Error::Unsupported { .. })
+ ));
+ let merge = FirstRowMergeFunction {
+ ignore_delete: true,
+ };
+ assert!(matches!(
+ merge.merge(&rows, &[], &[], &make_output_schema()).unwrap(),
+ MergeResult::SourceRow { row_idx: 0, .. }
+ ));
+ rows.remove(0);
+ assert!(matches!(
+ merge.merge(&rows, &[], &[], &make_output_schema()).unwrap(),
+ MergeResult::Omit
+ ));
+ }
+ }
+
fn make_schema() -> SchemaRef {
Arc::new(Schema::new(vec![
Field::new("pk", DataType::Int32, false),
diff --git a/crates/paimon/src/table/table_read.rs
b/crates/paimon/src/table/table_read.rs
index 7f5f70f1..b684d411 100644
--- a/crates/paimon/src/table/table_read.rs
+++ b/crates/paimon/src/table/table_read.rs
@@ -727,12 +727,15 @@ impl<'a> PaimonTableRead<'a> {
let merge_engine = core_options.merge_engine()?;
// Route supported PK merge engines through the split-aware reader.
- // Deduplicate may mix raw and KV splits. Partial-update and
aggregation
+ // Deduplicate and first-row 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,
- MergeEngine::Deduplicate | MergeEngine::PartialUpdate |
MergeEngine::Aggregation
+ MergeEngine::Deduplicate
+ | MergeEngine::FirstRow
+ | MergeEngine::PartialUpdate
+ | MergeEngine::Aggregation
)
{
return self.read_pk(data_splits, &core_options);
@@ -745,7 +748,7 @@ impl<'a> PaimonTableRead<'a> {
}
}
- /// Read PK table. For `Deduplicate`, splits marked raw convertible by scan
+ /// 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
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index 42e008d9..dda84796 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -724,8 +724,10 @@ fn should_skip_level_zero_for_scan(
return false;
}
- (deletion_vectors_enabled && !deletion_vectors_merge_on_read)
- || merge_engine.is_ok_and(|e| e == crate::spec::MergeEngine::FirstRow)
+ if deletion_vectors_enabled {
+ return !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 {
@@ -1301,7 +1303,6 @@ impl<'a> PaimonTableScan<'a> {
let data_evolution_enabled = core_options.data_evolution_enabled();
let has_primary_keys = !self.table.schema().primary_keys().is_empty();
- let deletion_vectors_enabled = core_options.deletion_vectors_enabled();
// Skip level-0 files for PK tables when:
// - DV mode: level-0 files are unmerged, DV handles dedup at higher
levels
@@ -1312,13 +1313,7 @@ impl<'a> PaimonTableScan<'a> {
//
// Non-read paths (overwrite, truncate, writer restore) set
scan_all_files=true
// to see all files including level-0, matching Java's CommitScanner
behavior.
- let skip_level_zero = should_skip_level_zero_for_scan(
- self.scan_all_files || self.is_streaming(),
- has_primary_keys,
- deletion_vectors_enabled,
- core_options.deletion_vectors_merge_on_read(),
- core_options.merge_engine(),
- );
+ let skip_level_zero = self.skip_level_zero();
let partition_fields = self.table.schema().partition_fields();
@@ -1535,31 +1530,12 @@ impl<'a> PaimonTableScan<'a> {
/// are still enforced exactly by the post-merge residual filter in
/// `KeyValueFileReader`.
///
- /// Exempt (full predicates kept):
- /// - 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
- /// exact residual filter would drop anyway. If first-row ever gains a
- /// merge read path, this exemption must be revisited.
+ /// Full predicates are safe for materialized first-row / deletion-vector
+ /// files. First-row all-files scans and incremental scans retain every
+ /// version before the reader applies its residual filter.
fn stats_pruning_predicates(&self) -> Vec<Predicate> {
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
- && (self.is_streaming()
- || ((!deletion_vectors_enabled ||
deletion_vectors_merge_on_read) && !first_row))
- {
+ if has_primary_keys && self.requires_key_merge() {
retain_primary_key_conjuncts(
&self.data_predicates,
self.table.schema().fields(),
@@ -1570,6 +1546,28 @@ impl<'a> PaimonTableScan<'a> {
}
}
+ fn requires_key_merge(&self) -> bool {
+ let options = self.table.schema().core_options();
+ self.is_streaming()
+ || match options.merge_engine() {
+ Ok(crate::spec::MergeEngine::FirstRow) =>
!self.skip_level_zero(),
+ _ => {
+ !options.deletion_vectors_enabled() ||
options.deletion_vectors_merge_on_read()
+ }
+ }
+ }
+
+ fn skip_level_zero(&self) -> bool {
+ let options = self.table.schema().core_options();
+ should_skip_level_zero_for_scan(
+ self.scan_all_files || self.is_streaming(),
+ !self.table.schema().primary_keys().is_empty(),
+ options.deletion_vectors_enabled(),
+ options.deletion_vectors_merge_on_read(),
+ options.merge_engine(),
+ )
+ }
+
/// Project file-safe predicates onto trimmed primary-key columns while
/// preserving table-schema field indices.
fn key_stats_predicates(&self, predicates: &[Predicate]) -> Vec<Predicate>
{
@@ -2096,17 +2094,10 @@ 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 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 use_key_interval_packing = self.is_streaming()
- || (!core_options.deletion_vectors_enabled()
- || core_options.deletion_vectors_merge_on_read())
- && !matches!(
- core_options.merge_engine(),
- Ok(crate::spec::MergeEngine::FirstRow)
- );
+ // Materialized first-row / DV data keeps size-based packing. First-row
+ // all-files scans and incremental scans retain L0 and must keep
+ // overlapping versions together.
+ let use_key_interval_packing = self.requires_key_merge();
let pk_comparator = if use_key_interval_packing {
KeyComparator::from_table_schema(self.table.schema())
} else {
@@ -2275,9 +2266,7 @@ impl<'a> PaimonTableScan<'a> {
// Java MergeTreeSplitGenerator#splitForBatch). Only engines
// whose writer deduplicates at flush guarantee a file never
// holds two rows of one key, so only they may mark groups raw
- // convertible; see merge_tree_split_for_batch. (First-row
- // tables do not take this path today, but its writer dedups
- // too, so keep the gate accurate.)
+ // convertible; see merge_tree_split_for_batch.
let file_keys_unique = matches!(
core_options.merge_engine(),
Ok(crate::spec::MergeEngine::Deduplicate)
@@ -3147,6 +3136,25 @@ mod tests {
));
}
+ #[test]
+ fn test_first_row_dv_merge_on_read_keeps_level_zero() {
+ // Java permits first-row DVs in pk-clustering-override tables.
+ assert!(!should_skip_level_zero_for_scan(
+ false,
+ true,
+ true,
+ true,
+ Ok(crate::spec::MergeEngine::FirstRow)
+ ));
+ assert!(should_skip_level_zero_for_scan(
+ false,
+ true,
+ true,
+ false,
+ Ok(crate::spec::MergeEngine::FirstRow)
+ ));
+ }
+
#[test]
fn test_scan_all_files_disables_first_row_level_zero_skip() {
assert!(!should_skip_level_zero_for_scan(
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index be343117..1a9c9a70 100644
--- a/crates/paimon/src/table/table_write.rs
+++ b/crates/paimon/src/table/table_write.rs
@@ -21,6 +21,7 @@
//! and [pypaimon
FileStoreWrite](https://github.com/apache/paimon/blob/master/paimon-python/pypaimon/write/file_store_write.py)
use crate::arrow::build_target_arrow_schema;
+use crate::arrow::partition::partition_array;
use crate::spec::PartitionComputer;
use crate::spec::{
first_row_supports_changelog_producer, BinaryRow, ChangelogProducer,
CoreOptions, DataField,
@@ -684,7 +685,39 @@ impl TableWrite {
}
for (key, row_indices) in delete_groups {
let sub_batch = take_rows(batch, &row_indices)?;
- let delete_batch = Self::add_value_kind_column(&sub_batch, 1)?;
+ // Java DeleteExistingProcessor emits the incoming values with
+ // the old partition and DELETE kind. Routing the file alone
+ // leaves incorrect physical values and partition statistics.
+ let partition = BinaryRow::from_serialized_bytes(&key.0)?;
+ let mut columns = sub_batch.columns().to_vec();
+ for (partition_index, field) in
+ self.table.schema().partition_fields().iter().enumerate()
+ {
+ let column_index =
+
sub_batch.schema().index_of(field.name()).map_err(|error| {
+ crate::Error::DataInvalid {
+ message: format!(
+ "Missing partition field '{}': {error}",
+ field.name()
+ ),
+ source: Some(Box::new(error)),
+ }
+ })?;
+ columns[column_index] = partition_array(
+ &partition,
+ partition_index,
+ field.data_type(),
+ sub_batch.num_rows(),
+ )?;
+ }
+ let sub_batch =
+ RecordBatch::try_new(sub_batch.schema(),
columns).map_err(|error| {
+ crate::Error::DataInvalid {
+ message: format!("Failed to restore old partition
for delete: {error}"),
+ source: Some(Box::new(error)),
+ }
+ })?;
+ let delete_batch = Self::add_value_kind_column(&sub_batch, 3)?;
result.push((key, delete_batch));
}
}
diff --git a/crates/paimon/tests/dynamic_bucket_scan_test.rs
b/crates/paimon/tests/dynamic_bucket_scan_test.rs
new file mode 100644
index 00000000..15123330
--- /dev/null
+++ b/crates/paimon/tests/dynamic_bucket_scan_test.rs
@@ -0,0 +1,220 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+mod common;
+
+use arrow_array::{Int32Array, Int8Array, RecordBatch, StringArray};
+use common::incremental_helpers::{
+ make_partitioned_batch, memory_table, persist_table_schema, setup_dirs,
write_batch,
+};
+use futures::TryStreamExt;
+use paimon::spec::{
+ DataField, DataType, Datum, IntType, PredicateBuilder, Schema,
TableSchema, TinyIntType,
+ VarCharType, VALUE_KIND_FIELD_ID, VALUE_KIND_FIELD_NAME,
+};
+use paimon::table::{IncrementalScanMode, ReadBuilder};
+
+async fn events(
+ builder: &ReadBuilder<'_>,
+ splits: &[paimon::DataSplit],
+) -> Vec<(String, i32, i32, i8)> {
+ let batches: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(splits)
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let mut events: Vec<_> = batches
+ .iter()
+ .flat_map(|batch| {
+ let pt = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ let id = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let value = batch
+ .column(2)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let kind = batch
+ .column(3)
+ .as_any()
+ .downcast_ref::<Int8Array>()
+ .unwrap();
+ (0..batch.num_rows())
+ .map(|i| {
+ (
+ pt.value(i).to_string(),
+ id.value(i),
+ value.value(i),
+ kind.value(i),
+ )
+ })
+ .collect::<Vec<_>>()
+ })
+ .collect();
+ events.sort_unstable();
+ events
+}
+
+#[tokio::test]
+async fn dynamic_and_cross_partition_scans_preserve_migration_events() {
+ for cross_partition in [false, true] {
+ for engine in ["deduplicate", "first-row"] {
+ let schema = Schema::builder()
+ .column("pt", DataType::VarChar(VarCharType::string_type()))
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .primary_key(if cross_partition {
+ vec!["id"]
+ } else {
+ vec!["id", "pt"]
+ })
+ .partition_keys(["pt"])
+ .option("bucket", "-1")
+ .option("merge-engine", engine)
+ .option("dynamic-bucket.target-row-num", "1")
+ .option("source.split.target-size", "1b")
+ .option("source.split.open-file-cost", "1b")
+ .build()
+ .unwrap();
+ let path =
format!("memory:/bucket_scan/{cross_partition}/{engine}");
+ let (io, table) = memory_table(&path, TableSchema::new(0,
&schema));
+ setup_dirs(&io, &path).await;
+ persist_table_schema(&io, &path, table.schema()).await;
+ write_batch(
+ &table,
+ &make_partitioned_batch(vec!["a", "a"], vec![1, 2], vec![10,
20]),
+ )
+ .await;
+ // Reopening the writer must restore the key-to-partition/bucket
index.
+ write_batch(
+ &table,
+ &make_partitioned_batch(vec!["b", "a"], vec![1, 3], vec![99,
30]),
+ )
+ .await;
+ let mut expected = vec![
+ ("a".into(), 1, 10, 0),
+ ("a".into(), 2, 20, 0),
+ ("a".into(), 3, 30, 0),
+ ];
+ if !(cross_partition && engine == "first-row") {
+ expected.push(("b".into(), 1, 99, 0));
+ }
+ if cross_partition && engine == "deduplicate" {
+ // Java DeleteExistingProcessor replaces the partition fields
+ // of the incoming row and emits DELETE, not UPDATE_BEFORE.
+ expected.push(("a".into(), 1, 99, 3));
+ }
+ expected.sort_unstable();
+ for partition in [None, Some("a"), Some("b")] {
+ let mut builder = table.new_read_builder();
+ let mut fields = table.schema().fields().to_vec();
+ fields.push(DataField::new(
+ VALUE_KIND_FIELD_ID,
+ VALUE_KIND_FIELD_NAME.to_string(),
+ DataType::TinyInt(TinyIntType::new()),
+ ));
+ builder.with_read_type(fields);
+ if let Some(partition) = partition {
+ builder.with_filter(
+ PredicateBuilder::new(table.schema().fields())
+ .equal("pt", Datum::String(partition.into()))
+ .unwrap(),
+ );
+ }
+ let plan = builder
+ .new_incremental_scan(IncrementalScanMode::Delta, 0, 2)
+ .plan_combined_delta()
+ .await
+ .unwrap();
+ assert!(plan.splits().iter().all(|s| s.is_streaming()));
+ let expected: Vec<_> = expected
+ .iter()
+ .filter(|event| partition.is_none_or(|p| event.0 == p))
+ .cloned()
+ .collect();
+ assert_eq!(
+ events(&builder, plan.splits()).await,
+ expected,
+ "cross_partition={cross_partition}, engine={engine},
partition={partition:?}"
+ );
+ }
+ let builder = table.new_read_builder();
+ let plan = builder
+ .new_scan()
+ .with_scan_all_files()
+ .plan()
+ .await
+ .unwrap();
+ assert!(plan.splits().iter().any(|s| s.bucket() > 0));
+ let batches: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let mut actual: Vec<_> = batches
+ .iter()
+ .flat_map(|batch| {
+ let pt = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ let id = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let value = batch
+ .column(2)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ (0..batch.num_rows())
+ .map(|i| (pt.value(i).to_string(), id.value(i),
value.value(i)))
+ .collect::<Vec<_>>()
+ })
+ .collect();
+ actual.sort_unstable();
+ let mut current: Vec<_> = expected
+ .iter()
+ .filter(|event| {
+ event.3 == 0
+ && !(cross_partition
+ && engine == "deduplicate"
+ && event.0 == "a"
+ && event.1 == 1)
+ })
+ .map(|event| (event.0.clone(), event.1, event.2))
+ .collect();
+ current.sort_unstable();
+ assert_eq!(actual, current);
+ }
+ }
+}
diff --git a/crates/paimon/tests/first_row_scan_test.rs
b/crates/paimon/tests/first_row_scan_test.rs
new file mode 100644
index 00000000..c9f4a158
--- /dev/null
+++ b/crates/paimon/tests/first_row_scan_test.rs
@@ -0,0 +1,162 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+mod common;
+
+use arrow_array::{Int32Array, RecordBatch};
+use common::incremental_helpers::{
+ make_batch, memory_table, persist_table_schema, pk_schema, setup_dirs,
write_batch,
+};
+use futures::TryStreamExt;
+use paimon::spec::{Datum, PredicateBuilder};
+use paimon::table::{IncrementalScanMode, Plan, ReadBuilder, Table};
+
+async fn table_with_versions(path: &str, compacted: bool) -> Table {
+ let (io, table) = memory_table(
+ path,
+ pk_schema(&[
+ ("merge-engine", "first-row"),
+ ("source.split.target-size", "1b"),
+ ("source.split.open-file-cost", "1b"),
+ ]),
+ );
+ setup_dirs(&io, path).await;
+ persist_table_schema(&io, path, table.schema()).await;
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer
+ .write_arrow_batch(&make_batch(vec![1, 2], vec![10, 20]))
+ .await
+ .unwrap();
+ let mut messages = writer.prepare_commit().await.unwrap();
+ if compacted {
+ // A unique, sorted run needs only a level upgrade during compaction.
+ for message in &mut messages {
+ for file in &mut message.new_files {
+ file.level = 1;
+ }
+ }
+ }
+ builder.new_commit().commit(messages).await.unwrap();
+ write_batch(&table, &make_batch(vec![1, 3], vec![99, 30])).await;
+ table
+}
+
+async fn rows(builder: &ReadBuilder<'_>, plan: &Plan) -> Vec<(i32, i32)> {
+ let batches: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let mut rows: Vec<_> = batches
+ .iter()
+ .flat_map(|batch| {
+ let ids = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let values = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ (0..batch.num_rows())
+ .map(|i| (ids.value(i), values.value(i)))
+ .collect::<Vec<_>>()
+ })
+ .collect();
+ rows.sort_unstable();
+ rows
+}
+
+#[tokio::test]
+async fn first_row_batch_reads_compacted_rows_and_skips_new_level_zero() {
+ let table = table_with_versions("memory:/first_row/compacted", true).await;
+ let builder = table.new_read_builder();
+ let plan = builder.new_scan().plan().await.unwrap();
+ assert_eq!(plan.snapshot_id(), Some(2));
+ assert!(plan.splits().iter().all(
+ |split| split.raw_convertible() &&
split.data_files().iter().all(|file| file.level > 0)
+ ));
+ assert_eq!(rows(&builder, &plan).await, vec![(1, 10), (2, 20)]);
+}
+
+#[tokio::test]
+async fn first_row_all_files_keeps_overlapping_versions_in_one_split() {
+ for compacted in [false, true] {
+ let path = format!("memory:/first_row/all_files/{compacted}");
+ let table = table_with_versions(&path, compacted).await;
+ let builder = table.new_read_builder();
+ let plan = builder
+ .new_scan()
+ .with_scan_all_files()
+ .plan()
+ .await
+ .unwrap();
+ // Both ranges overlap at key 1, even with a one-byte target split
size.
+ assert_eq!(plan.splits().len(), 1);
+ assert!(!plan.splits()[0].raw_convertible());
+ assert_eq!(plan.splits()[0].data_files().len(), 2);
+ assert_eq!(rows(&builder, &plan).await, vec![(1, 10), (2, 20), (3,
30)]);
+ }
+}
+
+#[tokio::test]
+async fn first_row_value_filter_runs_after_merging_all_versions() {
+ let table = table_with_versions("memory:/first_row/value_filter",
false).await;
+ for (value, expected) in [(10, vec![(1, 10)]), (99, vec![])] {
+ let predicate = PredicateBuilder::new(table.schema().fields())
+ .equal("value", Datum::Int(value))
+ .unwrap();
+ let mut builder = table.new_read_builder();
+ builder.with_filter(predicate);
+ let plan = builder
+ .new_scan()
+ .with_scan_all_files()
+ .plan()
+ .await
+ .unwrap();
+ assert_eq!(
+ plan.splits()
+ .iter()
+ .map(|s| s.data_files().len())
+ .sum::<usize>(),
+ 2
+ );
+ assert_eq!(rows(&builder, &plan).await, expected);
+ }
+}
+
+#[tokio::test]
+async fn first_row_incremental_preserves_events_instead_of_merging() {
+ let table = table_with_versions("memory:/first_row/incremental",
false).await;
+ let builder = table.new_read_builder();
+ let plan = builder
+ .new_incremental_scan(IncrementalScanMode::Delta, 0, 2)
+ .plan_combined_delta()
+ .await
+ .unwrap();
+ assert!(plan.splits().iter().all(|split| split.is_streaming()));
+ assert_eq!(
+ rows(&builder, &plan).await,
+ vec![(1, 10), (1, 99), (2, 20), (3, 30)]
+ );
+}
