From c48317e6b4e184e3c12358334170c110ef17a4b9 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Tue, 15 Sep 2026 22:08:56 +0800 Subject: [PATCH] [core] Align dynamic PK events and first-row scans with Java --- 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 ++- .../paimon/tests/dynamic_bucket_scan_test.rs | 220 ++++++++++++++++++ crates/paimon/tests/first_row_scan_test.rs | 162 +++++++++++++ 10 files changed, 794 insertions(+), 179 deletions(-) create mode 100644 crates/paimon/src/arrow/partition.rs create mode 100644 crates/paimon/tests/dynamic_bucket_scan_test.rs create mode 100644 crates/paimon/tests/first_row_scan_test.rs diff --git a/crates/paimon/src/arrow/mod.rs b/crates/paimon/src/arrow/mod.rs index 0995bd454..e30d83e59 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 000000000..5e426cab6 --- /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 { + 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::>(); + 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::>(); + 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 { + 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 9f0814d6f..cb6e232ff 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 { - 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::>(); - 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::>(); - 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 { - 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) -> Option { 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 f3adf84f2..8bc665e19 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 e63be64ac..1e88005bf 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 { + 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 7f5f70f19..b684d4111 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 42e008d9e..dda84796c 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 { 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 { @@ -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 be3431174..1a9c9a708 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 000000000..151233302 --- /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 = 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::() + .unwrap(); + let id = batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + let value = batch + .column(2) + .as_any() + .downcast_ref::() + .unwrap(); + let kind = batch + .column(3) + .as_any() + .downcast_ref::() + .unwrap(); + (0..batch.num_rows()) + .map(|i| { + ( + pt.value(i).to_string(), + id.value(i), + value.value(i), + kind.value(i), + ) + }) + .collect::>() + }) + .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 = 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::() + .unwrap(); + let id = batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + let value = batch + .column(2) + .as_any() + .downcast_ref::() + .unwrap(); + (0..batch.num_rows()) + .map(|i| (pt.value(i).to_string(), id.value(i), value.value(i))) + .collect::>() + }) + .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 000000000..c9f4a158a --- /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 = 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::() + .unwrap(); + let values = batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + (0..batch.num_rows()) + .map(|i| (ids.value(i), values.value(i))) + .collect::>() + }) + .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::(), + 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)] + ); +}