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 bede6650 Align postpone retract validation and managed Blob writes
with Java (#1020)
bede6650 is described below
commit bede665043268e88aa25a0c1e4c56e5e728ae104
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Oct 4 11:32:15 2026 +0800
Align postpone retract validation and managed Blob writes with Java (#1020)
---
crates/paimon/src/table/kv_file_writer.rs | 8 +-
.../paimon/src/table/managed_blob_table_tests.rs | 238 +++++++++++++++++++++
crates/paimon/src/table/managed_blob_writer.rs | 29 ++-
crates/paimon/src/table/mod.rs | 1 +
crates/paimon/src/table/postpone_file_writer.rs | 65 +++++-
crates/paimon/src/table/postpone_retract.rs | 141 ++++++++++++
crates/paimon/src/table/read_builder.rs | 9 +-
crates/paimon/src/table/sort_merge.rs | 18 +-
crates/paimon/src/table/table_write.rs | 15 +-
crates/paimon/tests/postpone_native_write_test.rs | 160 ++++++++++++++
10 files changed, 655 insertions(+), 29 deletions(-)
diff --git a/crates/paimon/src/table/kv_file_writer.rs
b/crates/paimon/src/table/kv_file_writer.rs
index 4b47263e..a14ff754 100644
--- a/crates/paimon/src/table/kv_file_writer.rs
+++ b/crates/paimon/src/table/kv_file_writer.rs
@@ -169,13 +169,19 @@ impl KeyValueFileWriter {
}
}
- let managed_blob_writer = ManagedBlobWriteState::new(&file_io,
&config)?;
let paths = super::data_file_path_factory::DataFilePathFactory::new(
&config.table_location,
&config.partition_path,
config.bucket,
&config.table_options,
)?;
+ let managed_blob_writer = ManagedBlobWriteState::new(
+ &file_io,
+ paths.bucket_path(),
+ &config.data_file_prefix,
+ &config.value_fields,
+ &core_options,
+ )?;
Ok(Self {
paths,
diff --git a/crates/paimon/src/table/managed_blob_table_tests.rs
b/crates/paimon/src/table/managed_blob_table_tests.rs
index 7a9cd190..e9d7cd92 100644
--- a/crates/paimon/src/table/managed_blob_table_tests.rs
+++ b/crates/paimon/src/table/managed_blob_table_tests.rs
@@ -1119,3 +1119,241 @@ fn
blobref_serialization_matches_java_modified_utf_and_checksum() {
.windows(6)
.any(|window| window == [0xed, 0xa0, 0xbd, 0xed, 0xb8, 0x80]));
}
+
+async fn read_pending_values(table: &Table, file: &crate::spec::DataFileMeta)
-> RecordBatch {
+ use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
+ let path = file
+ .external_path
+ .clone()
+ .unwrap_or_else(|| format!("{}/bucket-postpone/{}", table.location(),
file.file_name));
+ let bytes = table
+ .file_io()
+ .new_input(&path)
+ .unwrap()
+ .read()
+ .await
+ .unwrap();
+ let batches = ParquetRecordBatchReaderBuilder::try_new(bytes)
+ .unwrap()
+ .build()
+ .unwrap()
+ .map(|b| b.unwrap())
+ .collect::<Vec<_>>();
+ let batch = arrow_select::concat::concat_batches(&batches[0].schema(),
&batches).unwrap();
+ let values = batch
+ .project(&(2..batch.num_columns()).collect::<Vec<_>>())
+ .unwrap();
+ let stream = Box::pin(futures::stream::iter(vec![Ok(values)]));
+ let resolved: Vec<_> =
super::managed_blob_reader::resolve_primary_key_blob_stream(
+ stream,
+ table.schema().fields(),
+ &table.schema().core_options(),
+ table.file_io().clone(),
+ 2,
+ )
+ .try_collect()
+ .await
+ .unwrap();
+ resolved.into_iter().next().unwrap()
+}
+
+#[tokio::test]
+async fn
postpone_scalar_blobs_keep_events_and_reference_packs_for_each_rolled_file() {
+ let io = test_file_io();
+ let path = "memory:/postpone_managed_blob_roll";
+ setup_dirs(&io, path).await;
+ let table = scalar_table(
+ &io,
+ path,
+ &[
+ ("bucket", "-2"),
+ ("target-file-row-num", "2"),
+ ("blob.target-file-size", "1 b"),
+ ],
+ );
+ let mut writer = TableWrite::new(&table, "postpone-blob".into()).unwrap();
+ for batch in [
+ scalar_batch_with_kinds(&[(3, Some(b"one"), 0), (1, Some(b"ignored"),
3)]),
+ scalar_batch_with_kinds(&[(3, Some(b"two"), 2), (2, None, 0)]),
+ ] {
+ writer.write_arrow_batch(&batch).await.unwrap();
+ }
+ let messages = writer.prepare_commit().await.unwrap();
+ let files = &messages[0].new_files;
+ assert_eq!(files.len(), 2);
+ for (index, file) in files.iter().enumerate() {
+ assert_eq!(file.row_count, 2);
+ assert_eq!(file.delete_row_count, Some(if index == 0 { 1 } else { 0
}));
+ assert_eq!(file.extra_files.len(), 1);
+ let reference = io
+ .new_input(&format!("{path}/bucket-postpone/{}",
file.extra_files[0]))
+ .unwrap()
+ .read()
+ .await
+ .unwrap();
+ assert_eq!(&reference[0..5], &[0x50, 0x42, 0x4c, 0x52, 1]);
+ assert_eq!(i32::from_be_bytes(reference[5..9].try_into().unwrap()), 1);
+ let resolved = read_pending_values(&table, file).await;
+ let payloads = resolved
+ .column(1)
+ .as_any()
+ .downcast_ref::<LargeBinaryArray>()
+ .unwrap();
+ assert_eq!(payloads.value(0), if index == 0 { b"one" } else { b"two"
});
+ assert!(payloads.is_null(1));
+ }
+ writer.close().await;
+ // Preparing transfers ownership; closing must preserve both Parquet and
packs.
+ for file in files {
+ read_pending_values(&table, file).await;
+ }
+}
+
+#[tokio::test]
+async fn postpone_nested_blobs_preserve_nulls_and_duplicate_keys() {
+ let io = test_file_io();
+ let path = "memory:/postpone_nested_blob";
+ setup_dirs(&io, path).await;
+ let table = nested_table(&io,
path).copy_with_options(std::collections::HashMap::from([(
+ "bucket".into(),
+ "-2".into(),
+ )]));
+ let batch = nested_batch(&table);
+ let mut writer = TableWrite::new(&table,
"postpone-nested".into()).unwrap();
+ writer.write_arrow_batch(&batch).await.unwrap();
+ writer.write_arrow_batch(&batch).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ let file = &messages[0].new_files[0];
+ assert_eq!(file.row_count, (2 * batch.num_rows()) as i64);
+ let actual = read_pending_values(&table, file).await;
+ let expected =
+ arrow_select::concat::concat_batches(&batch.schema(), &[batch.clone(),
batch]).unwrap();
+ assert_eq!(actual, expected);
+}
+
+#[tokio::test]
+async fn postpone_blob_failure_aborts_all_unprepared_files_and_packs() {
+ let io = test_file_io();
+ let path = "memory:/postpone_blob_failed";
+ setup_dirs(&io, path).await;
+ let table = scalar_table(
+ &io,
+ path,
+ &[
+ ("bucket", "-2"),
+ ("merge-engine", "first-row"),
+ ("target-file-row-num", "1"),
+ ],
+ );
+ let mut writer = TableWrite::new(&table, "postpone-fail".into()).unwrap();
+ writer
+ .write_arrow_batch(&scalar_batch_with_kinds(&[(1, Some(b"one"), 0)]))
+ .await
+ .unwrap();
+ assert!(writer
+ .write_arrow_batch(&scalar_batch_with_kinds(&[(1, Some(b"two"), 3)]))
+ .await
+ .is_err());
+ assert!(writer.prepare_commit().await.is_err());
+ assert!(io
+ .list_status(&format!("{path}/bucket-postpone/"))
+ .await
+ .unwrap()
+ .is_empty());
+}
+
+#[tokio::test]
+async fn primary_key_blob_payload_limit_keeps_later_buckets() {
+ let file_io = test_file_io();
+ let path = "memory:/managed_blob_payload_limit";
+ setup_dirs(&file_io, path).await;
+ let table = scalar_table(&file_io, path, &[("bucket", "8")]);
+ let payloads: Vec<Vec<u8>> = (0..32)
+ .map(|id| format!("payload-{id}").into_bytes())
+ .collect();
+ let rows: Vec<_> = payloads
+ .iter()
+ .enumerate()
+ .map(|(id, payload)| (id as i32, Some(payload.as_slice())))
+ .collect();
+ let mut writer = TableWrite::new(&table,
"limit-test".to_string()).unwrap();
+ writer
+ .write_arrow_batch(&scalar_batch(&rows))
+ .await
+ .unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ assert!(messages.len() > 1);
+ TableCommit::new(table.clone(), "limit-test".to_string())
+ .commit(messages)
+ .await
+ .unwrap();
+ for (id, payload) in payloads.into_iter().enumerate() {
+ let predicate = PredicateBuilder::new(table.schema().fields())
+ .equal("payload", Datum::Bytes(payload))
+ .unwrap();
+ let mut builder = table.new_read_builder();
+ builder.with_projection(&["id"]).unwrap();
+ builder.with_filter(predicate).with_limit(1);
+ let plan = builder.new_scan().plan().await.unwrap();
+ // Each bucket's raw row count includes non-matching payloads.
+ assert!(plan.splits().len() > 1);
+ let batches: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(),
1);
+ assert_eq!(
+ batches
+ .iter()
+ .find(|batch| batch.num_rows() > 0)
+ .unwrap()
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .value(0),
+ id as i32
+ );
+ }
+}
+
+#[tokio::test]
+async fn postpone_blob_failed_checkpoint_preserves_prepared_packs() {
+ let file_io = test_file_io();
+ let path = "memory:/postpone_blob_checkpoint_ownership";
+ setup_dirs(&file_io, path).await;
+ let table = scalar_table(
+ &file_io,
+ path,
+ &[("bucket", "-2"), ("merge-engine", "first-row")],
+ );
+ let mut writer = TableWrite::new(&table,
"ownership-test".to_string()).unwrap();
+ writer
+ .write_arrow_batch(&scalar_batch(&[(1, Some(b"prepared"))]))
+ .await
+ .unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ writer
+ .write_arrow_batch(&scalar_batch(&[(2, Some(b"unprepared"))]))
+ .await
+ .unwrap();
+ assert!(writer
+ .write_arrow_batch(&scalar_batch_with_kinds(&[(2, Some(b"delete"),
3)]))
+ .await
+ .is_err());
+ let values = read_pending_values(&table, &messages[0].new_files[0]).await;
+ let payload = values
+ .column(1)
+ .as_any()
+ .downcast_ref::<LargeBinaryArray>()
+ .unwrap();
+ assert_eq!(payload.value(0), b"prepared");
+ let bucket = bucket_path_under(path, "", -2);
+ let paths = file_io.list_status(&format!("{bucket}/")).await.unwrap();
+ // Only the prepared data file, reference sidecar and pack survive.
+ assert_eq!(paths.len(), 3);
+}
diff --git a/crates/paimon/src/table/managed_blob_writer.rs
b/crates/paimon/src/table/managed_blob_writer.rs
index d53be232..5a1e8d6a 100644
--- a/crates/paimon/src/table/managed_blob_writer.rs
+++ b/crates/paimon/src/table/managed_blob_writer.rs
@@ -23,11 +23,10 @@
//! Parquet. A separate `.blobref` sidecar records the packs retained by each
//! physical data file.
-use super::kv_file_writer::KeyValueWriteConfig;
use crate::arrow::format::blob::BlobFormatWriter;
use crate::arrow::format::FormatFileWriter;
use crate::io::FileIO;
-use crate::spec::{bucket_path_under, BlobDescriptor, CoreOptions, DataField,
DataType, RowKind};
+use crate::spec::{BlobDescriptor, CoreOptions, DataField, DataType, RowKind};
use crate::Result;
use arrow_array::builder::LargeBinaryBuilder;
use arrow_array::{
@@ -98,19 +97,19 @@ pub(crate) struct ManagedBlobWriteState {
}
impl ManagedBlobWriteState {
- pub(crate) fn new(file_io: &FileIO, config: &KeyValueWriteConfig) ->
Result<Self> {
- let options = CoreOptions::new(&config.table_options);
+ pub(crate) fn new(
+ file_io: &FileIO,
+ bucket_dir: &str,
+ file_prefix: &str,
+ value_fields: &[DataField],
+ options: &CoreOptions<'_>,
+ ) -> Result<Self> {
let writer = ManagedBlobWriter::new(
file_io.clone(),
- &crate::spec::data_file_path(
- &config.table_location,
- options.data_file_path_directory(),
- ),
- &config.partition_path,
- config.bucket,
- &config.data_file_prefix,
+ bucket_dir.to_string(),
+ file_prefix,
options.blob_target_file_size(),
- managed_blob_fields(&config.value_fields, &options),
+ managed_blob_fields(value_fields, options),
)?;
Ok(Self {
writer: writer.map(|writer|
tokio::sync::Mutex::new(Box::new(writer))),
@@ -145,9 +144,7 @@ impl ManagedBlobWriteState {
impl ManagedBlobWriter {
pub(crate) fn new(
file_io: FileIO,
- table_location: &str,
- partition_path: &str,
- bucket: i32,
+ bucket_dir: String,
file_prefix: &str,
target_file_size: i64,
fields: Vec<(usize, ManagedBlobKind)>,
@@ -164,7 +161,7 @@ impl ManagedBlobWriter {
})?;
Ok(Some(Self {
file_io,
- bucket_dir: bucket_path_under(table_location, partition_path,
bucket),
+ bucket_dir,
file_prefix: file_prefix.to_string(),
target_file_size,
fields: fields
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index f3e02e50..ad5fb854 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -111,6 +111,7 @@ mod postpone_file_writer;
mod postpone_fixed_bucket_router;
mod postpone_fixed_bucket_write;
mod postpone_fixed_bucket_write_builder;
+mod postpone_retract;
mod prepared_files;
mod query_auth;
mod read_builder;
diff --git a/crates/paimon/src/table/postpone_file_writer.rs
b/crates/paimon/src/table/postpone_file_writer.rs
index e89235cb..89f3257b 100644
--- a/crates/paimon/src/table/postpone_file_writer.rs
+++ b/crates/paimon/src/table/postpone_file_writer.rs
@@ -22,15 +22,18 @@
//!
//! Uses a special file naming prefix: `data--u-{commitUser}-s-{writeId}-w-`.
//!
-//! Reference:
[PostponeBucketWriter](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/table/sink/PostponeBucketWriter.java)
+//! Reference:
[PostponeBucketWriter](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketWriter.java)
use super::data_file_path_factory::{DataFilePath, DataFilePathFactory};
+use super::managed_blob_reference::{ManagedBlobReferences,
REFERENCE_FILE_SUFFIX};
+use super::managed_blob_writer::ManagedBlobWriteState;
+use super::postpone_retract::PostponeRetractValidator;
use crate::arrow::format::{create_format_writer, with_write_resources,
FormatFileWriter};
use crate::io::FileIO;
use crate::resource::ResourceContext;
use crate::spec::stats::BinaryTableStats;
use crate::spec::{
- data_file_to_file_index_file_name, BinaryRow, DataField, DataFileMeta,
RowKind,
+ data_file_to_file_index_file_name, BinaryRow, CoreOptions, DataField,
DataFileMeta, RowKind,
EMPTY_SERIALIZED_ROW, VALUE_KIND_FIELD_NAME,
};
use crate::table::data_file_index_writer::{DataFileIndexWriter,
FileIndexOptions};
@@ -43,6 +46,8 @@ use tokio::task::JoinSet;
/// Configuration for [`PostponeFileWriter`].
pub(crate) struct PostponeWriteConfig {
+ pub table_name: String,
+ pub primary_keys: Vec<String>,
pub table_options: std::collections::HashMap<String, String>,
pub table_location: String,
pub partition_path: String,
@@ -73,6 +78,9 @@ pub(crate) struct PostponeFileWriter {
next_sequence_number: i64,
current_writer: Option<Box<dyn FormatFileWriter>>,
current_index: Option<DataFileIndexWriter>,
+ current_blob_references: Option<ManagedBlobReferences>,
+ managed_blob_writer: ManagedBlobWriteState,
+ retract_validator: PostponeRetractValidator,
current_file_name: Option<String>,
current_file_path: Option<DataFilePath>,
current_stats: PostponeFileStats,
@@ -93,6 +101,14 @@ impl PostponeFileWriter {
config.bucket,
&config.table_options,
)?;
+ let managed_blob_writer = ManagedBlobWriteState::new(
+ &file_io,
+ paths.bucket_path(),
+ &config.data_file_prefix,
+ &config.value_fields,
+ &CoreOptions::new(&config.table_options),
+ )?;
+ let retract_validator = PostponeRetractValidator::new(&config)?;
Ok(Self {
paths,
file_io,
@@ -100,6 +116,9 @@ impl PostponeFileWriter {
next_sequence_number: 0,
current_writer: None,
current_index: None,
+ current_blob_references: None,
+ managed_blob_writer,
+ retract_validator,
current_file_name: None,
current_file_path: None,
current_stats: PostponeFileStats::default(),
@@ -118,7 +137,7 @@ impl PostponeFileWriter {
pub(crate) async fn write(&mut self, batch: &RecordBatch) -> Result<()> {
let result = self.write_batch(batch).await;
- if result.is_err() && self.config.file_index_options.is_some() {
+ if result.is_err() {
self.abort().await;
}
result
@@ -129,6 +148,8 @@ impl PostponeFileWriter {
return Ok(());
}
+ let batch = self.managed_blob_writer.externalize(batch.clone()).await?;
+ self.retract_validator.validate(&batch)?;
if self.current_writer.is_none() {
self.open_new_file(batch.schema()).await?;
}
@@ -168,6 +189,7 @@ impl PostponeFileWriter {
}
})?;
+ ManagedBlobReferences::collect(&mut self.current_blob_references,
&physical_batch)?;
self.current_writer
.as_mut()
.unwrap()
@@ -188,7 +210,7 @@ impl PostponeFileWriter {
}
self.next_sequence_number = end_seq + 1;
self.current_stats
- .add_batch(batch, &self.config, start_seq, end_seq)?;
+ .add_batch(&batch, &self.config, start_seq, end_seq)?;
// Roll to a new file if target size is reached — close in background
if self.current_writer.as_ref().unwrap().num_bytes() as i64 >=
self.config.target_file_size
@@ -212,6 +234,7 @@ impl PostponeFileWriter {
let _ = writer.close().await;
}
self.current_index = None;
+ self.current_blob_references = None;
self.current_file_name = None;
self.current_file_path = None;
while self.in_flight_closes.join_next().await.is_some() {}
@@ -219,11 +242,12 @@ impl PostponeFileWriter {
let _ = self.file_io.delete_file(&path).await;
}
self.written_files.clear();
+ self.managed_blob_writer.abort().await;
}
pub(crate) async fn prepare_commit(&mut self) -> Result<Vec<DataFileMeta>>
{
let result = self.finish().await;
- if result.is_err() && self.config.file_index_options.is_some() {
+ if result.is_err() {
self.abort().await;
}
result
@@ -242,6 +266,7 @@ impl PostponeFileWriter {
// manifest order for ties. Background closes must not reorder
arrivals.
self.written_files
.sort_by_key(|file| file.min_sequence_number);
+ self.managed_blob_writer.prepare_commit().await?;
self.created_paths.clear();
Ok(std::mem::take(&mut self.written_files))
}
@@ -254,6 +279,7 @@ impl PostponeFileWriter {
};
let file_name = self.current_file_name.take().unwrap();
let index = self.current_index.take();
+ let blob_references = self.current_blob_references.take();
let file_io = self.file_io.clone();
let location = self.current_file_path.take().unwrap();
let bucket_dir = location.parent().to_string();
@@ -271,6 +297,14 @@ impl PostponeFileWriter {
self.in_flight_closes.spawn(async move {
let file_size = writer.close().await?.file_size as i64;
let mut meta = build_meta(file_name, file_size, stats, schema_id,
creation_time);
+ ManagedBlobReferences::finish(
+ blob_references,
+ &file_io,
+ &location.path,
+ &bucket_dir,
+ &mut meta,
+ )
+ .await?;
meta.external_path = location.external_path;
write_index(index, threshold, &file_io, &bucket_dir, &mut
meta).await?;
Ok(meta)
@@ -303,6 +337,17 @@ impl PostponeFileWriter {
data_file_to_file_index_file_name(&file_name)
));
}
+ let blob_references = ManagedBlobReferences::new(
+ &self.config.value_fields,
+ &CoreOptions::new(&self.config.table_options),
+ &physical_schema,
+ self.managed_blob_writer.enabled(),
+ false,
+ )?;
+ if blob_references.is_some() {
+ self.created_paths
+ .push(format!("{file_path}{REFERENCE_FILE_SUFFIX}"));
+ }
let output = self.file_io.new_output(&file_path)?;
let writer = create_format_writer(
&output,
@@ -316,6 +361,7 @@ impl PostponeFileWriter {
.await?;
self.current_writer = Some(with_write_resources(writer,
self.resources.as_ref()));
self.current_index = index;
+ self.current_blob_references = blob_references;
self.current_file_name = Some(file_name);
self.current_file_path = Some(location);
self.current_stats = PostponeFileStats::default();
@@ -330,6 +376,7 @@ impl PostponeFileWriter {
};
let file_name = self.current_file_name.take().unwrap();
let index = self.current_index.take();
+ let blob_references = self.current_blob_references.take();
let stats = std::mem::take(&mut self.current_stats);
let file_size = writer.close().await?.file_size as i64;
@@ -347,6 +394,14 @@ impl PostponeFileWriter {
.file_index_options
.as_ref()
.map(|options| options.in_manifest_threshold);
+ ManagedBlobReferences::finish(
+ blob_references,
+ &self.file_io,
+ &location.path,
+ &bucket_dir,
+ &mut meta,
+ )
+ .await?;
meta.external_path = location.external_path;
write_index(index, threshold, &self.file_io, &bucket_dir, &mut
meta).await?;
self.written_files.push(meta);
diff --git a/crates/paimon/src/table/postpone_retract.rs
b/crates/paimon/src/table/postpone_retract.rs
new file mode 100644
index 00000000..8944902d
--- /dev/null
+++ b/crates/paimon/src/table/postpone_retract.rs
@@ -0,0 +1,141 @@
+// 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.
+
+//! Validate the first retract using the unwrapped merge function, as Java's
+//! PostponeBucketWriter does. Pending rows themselves are never merged.
+
+use super::postpone_file_writer::PostponeWriteConfig;
+use super::sort_merge::{
+ AggregateMergeFunction, BufferedBatch, FirstRowMergeFunction,
MergeFunction, MergeRow,
+ PartialUpdateMergeFunction,
+};
+use crate::spec::{
+ AggregationConfig, CoreOptions, MergeEngine, PartialUpdateConfig, RowKind,
+ VALUE_KIND_FIELD_NAME,
+};
+use crate::Result;
+use arrow_array::{Array, Int8Array, RecordBatch};
+
+enum RetractMerge {
+ Deduplicate,
+ FirstRow(FirstRowMergeFunction),
+ PartialUpdate(PartialUpdateMergeFunction),
+ Aggregate(AggregateMergeFunction),
+}
+
+pub(super) struct PostponeRetractValidator {
+ merge: Box<RetractMerge>,
+ validated: bool,
+ value_indices: Vec<usize>,
+}
+
+impl PostponeRetractValidator {
+ pub(super) fn new(config: &PostponeWriteConfig) -> Result<Self> {
+ let options = CoreOptions::new(&config.table_options);
+ let merge = match options.merge_engine()? {
+ MergeEngine::Deduplicate => RetractMerge::Deduplicate,
+ MergeEngine::FirstRow =>
RetractMerge::FirstRow(FirstRowMergeFunction {
+ ignore_delete: options.ignore_delete(),
+ }),
+ MergeEngine::PartialUpdate => {
+ PartialUpdateConfig::new(&config.table_options)
+ .validate_write_mode(true, &config.table_name)?;
+
RetractMerge::PartialUpdate(PartialUpdateMergeFunction::new_with_schema(
+ &config.table_options,
+ &config.table_name,
+ &config.value_fields,
+ &config.value_fields,
+ &config.primary_keys,
+ )?)
+ }
+ MergeEngine::Aggregation => {
+ AggregationConfig::new(&config.table_options)
+ .validate_runtime_mode(true, &config.table_name)?;
+ let sequences = options
+ .sequence_fields()
+ .iter()
+ .map(|s| s.to_string())
+ .collect::<Vec<_>>();
+ RetractMerge::Aggregate(AggregateMergeFunction::new(
+ &config.table_options,
+ &config.table_name,
+ &config.value_fields,
+ &config.primary_keys,
+ &sequences,
+ )?)
+ }
+ };
+ Ok(Self {
+ merge: Box::new(merge),
+ validated: false,
+ value_indices: (0..config.value_fields.len()).collect(),
+ })
+ }
+
+ pub(super) fn validate(&mut self, batch: &RecordBatch) -> Result<()> {
+ if self.validated {
+ return Ok(());
+ }
+ let Some(kinds) = batch.column_by_name(VALUE_KIND_FIELD_NAME) else {
+ return Ok(());
+ };
+ let kinds = kinds
+ .as_any()
+ .downcast_ref::<Int8Array>()
+ .expect("validated row kinds");
+ for index in 0..batch.num_rows() {
+ if kinds.is_null(index) {
+ continue;
+ }
+ let kind = RowKind::from_value(kinds.value(index))?;
+ if !kind.is_retract() {
+ continue;
+ }
+ let row = MergeRow {
+ batch_idx: 0,
+ row_idx: index,
+ sequence_number: 0,
+ value_kind: kind.to_value(),
+ user_sequence: None,
+ };
+ let values =
+ batch
+ .project(&self.value_indices)
+ .map_err(|error| crate::Error::DataInvalid {
+ message: format!("Cannot project postpone values:
{error}"),
+ source: Some(Box::new(error)),
+ })?;
+ let schema = values.schema();
+ let buffer = [BufferedBatch::Source(values)];
+ match self.merge.as_ref() {
+ RetractMerge::Deduplicate => {}
+ RetractMerge::FirstRow(merge) => {
+ merge.merge(&[row], &buffer, &self.value_indices,
&schema)?;
+ }
+ RetractMerge::PartialUpdate(merge) => {
+ merge.merge_unreduced(&[row], &buffer,
&self.value_indices, &schema)?;
+ }
+ RetractMerge::Aggregate(merge) => {
+ merge.merge_ordered(&[&row], &buffer, &self.value_indices,
&schema)?;
+ }
+ }
+ self.validated = true;
+ break;
+ }
+ Ok(())
+ }
+}
diff --git a/crates/paimon/src/table/read_builder.rs
b/crates/paimon/src/table/read_builder.rs
index ee54d87a..5b8295e0 100644
--- a/crates/paimon/src/table/read_builder.rs
+++ b/crates/paimon/src/table/read_builder.rs
@@ -532,12 +532,19 @@ impl<'a> PaimonReadBuilder<'a> {
let read_type = self.resolve_read_type().unwrap_or(None);
let scan_predicates =
super::managed_blob_reader::scan_predicates(self.table,
&self.filter.data_predicates);
+ // A payload predicate is evaluated after descriptor resolution. A
+ // scan limit must not count rows which may fail that residual filter.
+ let scan_limit = if scan_predicates.len() ==
self.filter.data_predicates.len() {
+ self.limit
+ } else {
+ None
+ };
TableScan::new(
self.table,
partition_filter,
scan_predicates,
self.filter.bucket_predicate.clone(),
- self.limit,
+ scan_limit,
self.effective_row_ranges(),
)
.with_projected_read_field_ids(projected_read_field_ids_with_predicates(
diff --git a/crates/paimon/src/table/sort_merge.rs
b/crates/paimon/src/table/sort_merge.rs
index 3c130745..2641d68c 100644
--- a/crates/paimon/src/table/sort_merge.rs
+++ b/crates/paimon/src/table/sort_merge.rs
@@ -458,6 +458,21 @@ impl MergeFunction for PartialUpdateMergeFunction {
batch_buffer: &[BufferedBatch],
source_output_col_indices: &[usize],
output_schema: &SchemaRef,
+ ) -> crate::Result<MergeResult> {
+ if let Some(result) = singleton_reducer_result(rows)? {
+ return Ok(result);
+ }
+ self.merge_unreduced(rows, batch_buffer, source_output_col_indices,
output_schema)
+ }
+}
+
+impl PartialUpdateMergeFunction {
+ pub(super) fn merge_unreduced(
+ &self,
+ rows: &[MergeRow],
+ batch_buffer: &[BufferedBatch],
+ source_output_col_indices: &[usize],
+ output_schema: &SchemaRef,
) -> crate::Result<MergeResult> {
if rows.is_empty() {
return Err(Error::UnexpectedError {
@@ -465,9 +480,6 @@ impl MergeFunction for PartialUpdateMergeFunction {
source: None,
});
}
- if let Some(result) = singleton_reducer_result(rows)? {
- return Ok(result);
- }
let mut ordered_row_indices: Vec<usize> = (0..rows.len()).collect();
ordered_row_indices.sort_by(|&lhs_idx, &rhs_idx| {
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index 814b5c46..afaa3daf 100644
--- a/crates/paimon/src/table/table_write.rs
+++ b/crates/paimon/src/table/table_write.rs
@@ -91,7 +91,7 @@ impl FileWriter {
}
}
- async fn prepare_commit(mut self) -> Result<PreparedFiles> {
+ async fn prepare_commit(mut self) -> (Result<PreparedFiles>, Option<Self>)
{
let result = match &mut self {
FileWriter::Append(w) =>
w.prepare_commit().await.map(PreparedFiles::data),
FileWriter::AppendDedicated(w) =>
w.prepare_commit().await.map(PreparedFiles::data),
@@ -101,7 +101,10 @@ impl FileWriter {
if result.is_err() {
self.abort().await;
}
- result
+ // A pending-bucket writer keeps its first-retract validation and
+ // arrival sequence across stream checkpoints, as Java does.
+ let retain = result.is_ok() && matches!(&self,
FileWriter::Postpone(_));
+ (result, retain.then_some(self))
}
async fn abort(&mut self) {
@@ -1081,7 +1084,11 @@ impl TableWrite {
let mut messages = Vec::new();
let mut error = None;
- for (partition_bytes, bucket, result) in results {
+ for (partition_bytes, bucket, (result, retained)) in results {
+ if let Some(writer) = retained {
+ self.partition_writers
+ .insert((partition_bytes.clone(), bucket), writer);
+ }
match result {
Ok(files) if !files.data_files.is_empty() ||
!files.changelog_files.is_empty() => {
let mut message = CommitMessage::new(partition_bytes,
bucket, files.data_files);
@@ -1222,6 +1229,8 @@ impl TableWrite {
PostponeFileWriter::new(
self.table.file_io().clone(),
PostponeWriteConfig {
+ table_name: self.table.identifier().full_name(),
+ primary_keys: self.table.schema().primary_keys().to_vec(),
table_location: self.table.location().to_string(),
table_options: self.table.schema().options().clone(),
partition_path,
diff --git a/crates/paimon/tests/postpone_native_write_test.rs
b/crates/paimon/tests/postpone_native_write_test.rs
index 1063e104..3e4d7b53 100644
--- a/crates/paimon/tests/postpone_native_write_test.rs
+++ b/crates/paimon/tests/postpone_native_write_test.rs
@@ -135,3 +135,163 @@ async fn
postpone_delete_counts_reset_after_prepare_commit() {
assert_eq!(messages[0].new_files[0].delete_row_count, Some(expected));
}
}
+
+// Java PostponeBucketWriter validates a retract with the actual merge
function,
+// not ReducerMergeFunctionWrapper's singleton shortcut.
+#[tokio::test]
+async fn postpone_retract_validation_matches_all_merge_engines() {
+ type Options<'a> = &'a [(&'a str, &'a str)];
+ let cases: &[(&str, Options<'_>, i8, bool)] = &[
+ ("deduplicate", &[], 3, true),
+ ("first-row", &[], 3, false),
+ ("first-row", &[], 1, false),
+ ("first-row", &[("ignore-delete", "true")], 3, true),
+ ("partial-update", &[], 3, false),
+ (
+ "partial-update",
+ &[("partial-update.remove-record-on-delete", "true")],
+ 3,
+ true,
+ ),
+ (
+ "partial-update",
+ &[("partial-update.remove-record-on-delete", "true")],
+ 1,
+ true,
+ ),
+ (
+ "aggregation",
+ &[("fields.value.aggregate-function", "sum")],
+ 3,
+ true,
+ ),
+ (
+ "aggregation",
+ &[("fields.value.aggregate-function", "min")],
+ 3,
+ false,
+ ),
+ (
+ "aggregation",
+ &[
+ ("fields.value.aggregate-function", "min"),
+ ("fields.value.ignore-retract", "true"),
+ ],
+ 3,
+ true,
+ ),
+ (
+ "aggregation",
+ &[
+ ("fields.value.aggregate-function", "min"),
+ ("aggregation.remove-record-on-delete", "true"),
+ ],
+ 3,
+ true,
+ ),
+ ];
+ for (index, &(engine, extra, kind, accepted)) in cases.iter().enumerate() {
+ let mut options = vec![
+ ("bucket", "-2"),
+ ("merge-engine", engine),
+ ("target-file-row-num", "1"),
+ ];
+ options.extend_from_slice(extra);
+ let (io, table) = memory_table(
+ &format!("memory:/postpone_retract_{index}"),
+ pk_schema(&options),
+ );
+ setup_dirs(&io, table.location()).await;
+ let mut writer = table.new_write_builder().new_write().unwrap();
+ writer
+ .write_arrow_batch(&make_batch_with_kinds(vec![2], vec![10],
vec![0]))
+ .await
+ .unwrap();
+ let result = writer
+ .write_arrow_batch(&make_batch_with_kinds(vec![1], vec![20],
vec![kind]))
+ .await;
+ assert_eq!(
+ result.is_ok(),
+ accepted,
+ "{engine}, {extra:?}, {kind}: {result:?}"
+ );
+ if accepted {
+ let messages = writer.prepare_commit().await.unwrap();
+ let expected = if extra.contains(&("ignore-delete", "true")) {
+ 1
+ } else {
+ 2
+ };
+ assert_eq!(
+ messages[0]
+ .new_files
+ .iter()
+ .map(|f| f.row_count)
+ .sum::<i64>(),
+ expected
+ );
+ } else {
+ assert!(writer.prepare_commit().await.is_err());
+ assert!(writer
+ .write_arrow_batch(&make_batch_with_kinds(vec![3], vec![30],
vec![0]))
+ .await
+ .is_err());
+ let files = io
+ .list_status(&format!("{}/bucket-postpone/", table.location()))
+ .await
+ .unwrap();
+ assert!(files.is_empty(), "failed writer left files: {files:?}");
+ }
+ }
+}
+
+#[tokio::test]
+async fn postpone_validates_only_the_first_retract_even_across_checkpoints() {
+ let (io, table) = memory_table(
+ "memory:/postpone_retract_once",
+ pk_schema(&[
+ ("bucket", "-2"),
+ ("merge-engine", "aggregation"),
+ ("fields.value.aggregate-function", "min"),
+ ("aggregation.remove-record-on-delete", "true"),
+ ]),
+ );
+ setup_dirs(&io, table.location()).await;
+ let mut writer = table.new_write_builder().new_write().unwrap();
+ // DELETE validates successfully via remove-record-on-delete. A subsequent
+ // UPDATE_BEFORE must be retained without validating min's retract again.
+ for kind in [3, 1] {
+ writer
+ .write_arrow_batch(&make_batch_with_kinds(vec![1], vec![20],
vec![kind]))
+ .await
+ .unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ assert_eq!(messages[0].new_files[0].delete_row_count, Some(1));
+ }
+}
+
+#[tokio::test]
+async fn postpone_preserves_inserts_without_merging_for_all_merge_engines() {
+ for engine in ["deduplicate", "first-row", "partial-update",
"aggregation"] {
+ let mut options = vec![("bucket", "-2"), ("merge-engine", engine)];
+ if engine == "aggregation" {
+ options.push(("fields.value.aggregate-function", "sum"));
+ }
+ let (io, table) = memory_table(
+ &format!("memory:/postpone_inserts_{engine}"),
+ pk_schema(&options),
+ );
+ setup_dirs(&io, table.location()).await;
+ let mut writer = table.new_write_builder().new_write().unwrap();
+ writer
+ .write_arrow_batch(&make_batch_with_kinds(
+ vec![3, 1, 3],
+ vec![10, 20, 30],
+ vec![0, 0, 2],
+ ))
+ .await
+ .unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ assert_eq!(messages[0].new_files[0].row_count, 3, "{engine}");
+ }
+}
