JingsongLi commented on code in PR #659:
URL: https://github.com/apache/paimon-rust/pull/659#discussion_r3719132961
##########
crates/paimon/src/table/table_commit.rs:
##########
@@ -1853,6 +1857,7 @@ impl TableCommit {
check_from_snapshot: Option<i64>,
) -> Result<()> {
self.check_delete_entries_against_base(base_entries, delta_entries)?;
+ self.check_total_bucket_conflicts(base_entries, delta_entries)?;
Review Comment:
This bucket-count check runs unconditionally before considering
`commit_kind`, and it compares raw base and delta entries. An overwrite
contains DELETE entries for the old layout followed by ADD entries for the
replacement layout, so a valid rescale such as 1 -> 3 buckets is rejected as a
conflict. The upstream Java implementation skips the old-layout consistency
check for `OVERWRITE`, and PyPaimon has
`test_postpone_overwrite_allows_bucket_rescale`. Please continue checking that
all new ADD entries in the delta agree, but compare them with the base layout
only for non-overwrite commits (or apply ADD/DELETE changes before checking the
final active entries).
##########
crates/paimon/src/table/postpone_batch_table_write.rs:
##########
@@ -0,0 +1,370 @@
+// 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.
+
+//! One-shot fixed-bucket planning for batch writes to postpone tables.
+//!
+//! This mirrors pypaimon's `PostponeFixedBucketBatchTableWrite`: partitions
+//! with an existing real-bucket count stream directly to their writers, while
+//! new partitions are buffered until `prepare_commit` can infer one bucket
+//! count from the complete batch.
+
+use crate::spec::{
+ batch_to_serialized_bytes, BucketFunctionType, CoreOptions, DataField,
EMPTY_SERIALIZED_ROW,
+ POSTPONE_BUCKET,
+};
+use crate::table::bucket_function::{batch_bucket_ids,
validate_bucket_function};
+use crate::table::postpone_bucket::binary_row_batch_size;
+use crate::table::{SnapshotManager, Table, TableScan};
+use crate::Result;
+use arrow_array::{RecordBatch, UInt32Array};
+use std::collections::HashMap;
+
+pub(super) struct PostponeBucketBatch {
+ pub(super) partition: Vec<u8>,
+ pub(super) bucket: i32,
+ pub(super) batch: RecordBatch,
+}
+
+/// Planning state for a single fixed-bucket batch write to a postpone table.
+pub(super) struct PostponeFixedBucketWriter {
+ partition_field_indices: Vec<usize>,
+ bucket_key_indices: Vec<usize>,
+ bucket_function_type: BucketFunctionType,
+ max_parallelism: i32,
+ target_rows_per_bucket: Option<i64>,
+ target_size_per_bucket: i64,
+ metadata_loaded: bool,
+ known_bucket_counts: HashMap<Vec<u8>, i32>,
+ postpone_row_counts: HashMap<Vec<u8>, i64>,
+ buffered_batches: HashMap<Vec<u8>, Vec<RecordBatch>>,
+ /// Bucket counts used by this prepare-commit round.
+ bucket_counts: HashMap<Vec<u8>, i32>,
+ prepare_started: bool,
+}
+
+impl PostponeFixedBucketWriter {
+ pub(super) fn new(
+ table: &Table,
+ partition_field_indices: Vec<usize>,
+ bucket_key_indices: Vec<usize>,
+ bucket_function_type: BucketFunctionType,
+ ) -> Result<Self> {
+ let schema = table.schema();
+ let options = CoreOptions::new(schema.options());
+ let total_buckets = options.bucket();
+ if total_buckets != POSTPONE_BUCKET ||
schema.primary_keys().is_empty() {
+ return Err(crate::Error::Unsupported {
+ message: format!(
+ "Postpone fixed-bucket writes require a primary-key table
with bucket=-2, but table '{}' has bucket={total_buckets}",
+ table.identifier().full_name()
+ ),
+ });
+ }
+ if options.deletion_vectors_enabled() {
+ return Err(crate::Error::Unsupported {
+ message: format!(
+ "Table '{}' cannot use postpone fixed-bucket writes with
deletion-vectors.enabled=true because deletion-vector scans skip the level-0
files produced by batch writers; use the normal postpone writer or disable
deletion vectors",
+ table.identifier().full_name()
+ ),
+ });
+ }
+
+ let bucket_key_fields: Vec<DataField> = bucket_key_indices
+ .iter()
+ .map(|&index| schema.fields()[index].clone())
+ .collect();
+ if !bucket_key_fields.is_empty() {
+ validate_bucket_function(bucket_function_type,
&bucket_key_fields)?;
+ }
+
+ Ok(Self {
+ partition_field_indices,
+ bucket_key_indices,
+ bucket_function_type,
+ max_parallelism:
options.postpone_batch_write_fixed_bucket_max_parallelism()?,
+ target_rows_per_bucket:
options.postpone_target_row_num_per_bucket()?,
+ target_size_per_bucket: options.postpone_target_size_per_bucket()?,
+ metadata_loaded: false,
+ known_bucket_counts: HashMap::new(),
+ postpone_row_counts: HashMap::new(),
+ buffered_batches: HashMap::new(),
+ bucket_counts: HashMap::new(),
+ prepare_started: false,
+ })
+ }
+
+ pub(super) fn ensure_writable(&self) -> Result<()> {
+ if self.prepare_started {
+ return Err(Self::one_shot_error());
+ }
+ Ok(())
+ }
+
+ pub(super) fn start_prepare(&mut self) -> Result<()> {
+ self.ensure_writable()?;
+ // A failed prepare may already have consumed buffered batches or
+ // closed file writers, so the same writer cannot be retried safely.
+ self.prepare_started = true;
+ Ok(())
+ }
+
+ pub(super) async fn write_batch(
+ &mut self,
+ table: &Table,
+ batch: &RecordBatch,
+ ) -> Result<Vec<PostponeBucketBatch>> {
+ self.ensure_metadata_loaded(table).await?;
+
+ let partitions = if self.partition_field_indices.is_empty() {
+ vec![EMPTY_SERIALIZED_ROW.clone(); batch.num_rows()]
+ } else {
+ batch_to_serialized_bytes(
+ batch,
+ &self.partition_field_indices,
+ table.schema().fields(),
+ )?
+ };
+
+ let mut groups: HashMap<Vec<u8>, Vec<usize>> = HashMap::new();
+ for (row, partition) in partitions.into_iter().enumerate() {
+ groups.entry(partition).or_default().push(row);
+ }
+
+ let mut output = Vec::new();
+ for (partition, rows) in groups {
+ let sub_batch = take_rows(batch, &rows)?;
+ if let Some(total_buckets) =
self.known_bucket_counts.get(&partition).copied() {
+ self.bucket_counts.insert(partition.clone(), total_buckets);
+ output.extend(self.route_batch(table, partition, sub_batch,
total_buckets)?);
+ } else {
+ self.buffered_batches
+ .entry(partition)
+ .or_default()
+ .push(sub_batch);
+ }
+ }
+ Ok(output)
+ }
+
+ pub(super) async fn prepare_batch(
+ &mut self,
+ table: &Table,
+ is_overwrite: bool,
+ ) -> Result<Vec<PostponeBucketBatch>> {
+ if self.buffered_batches.is_empty() {
+ return Ok(Vec::new());
+ }
+
+ let buffered_batches = std::mem::take(&mut self.buffered_batches);
+ let mut output = Vec::new();
+ for (partition, batches) in buffered_batches {
+ let input_rows = batches.iter().fold(0_i64, |rows, batch| {
+ rows.saturating_add(batch.num_rows() as i64)
+ });
+ // Match pypaimon: row-count planning does not inspect row sizes.
+ // Size planning ignores the trailing internal `_VALUE_KIND` field
+ // appended by TableWrite after row-kind generation.
+ let input_size = if self.target_rows_per_bucket.is_none() {
+ batches.iter().try_fold(0_i64, |size, batch| {
+ Ok::<_, crate::Error>(
+ size.saturating_add(binary_row_batch_size(batch,
table.schema().fields())?),
+ )
+ })?
+ } else {
+ 0
+ };
+ let postpone_rows = if is_overwrite {
+ 0
+ } else {
+ self.postpone_row_counts
+ .get(&partition)
+ .copied()
+ .unwrap_or(0)
+ };
+ let total_buckets = infer_bucket_count(
Review Comment:
Each `TableWrite` derives `total_buckets` from only its local buffered rows.
However, the C API supports merging messages from multiple fixed-bucket writers
that share a `commit_user`, and there is no way to provide those writers with
one precomputed bucket plan. Two workers writing the same new partition can
therefore infer different counts and make the whole commit fail; even when they
infer the same count, the plan is based on shard-local rather than global batch
statistics and can under-bucket the partition. PyPaimon avoids this by
aggregating partition statistics on the driver and injecting the same
`PostponeBucketPlan` into every worker. Please expose and validate a
precomputed `partition -> total_buckets` plan in the Rust builder/writer and C
API, or explicitly reject this multi-writer mode.
##########
crates/paimon/src/table/postpone_batch_table_write.rs:
##########
@@ -0,0 +1,370 @@
+// 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.
+
+//! One-shot fixed-bucket planning for batch writes to postpone tables.
+//!
+//! This mirrors pypaimon's `PostponeFixedBucketBatchTableWrite`: partitions
+//! with an existing real-bucket count stream directly to their writers, while
+//! new partitions are buffered until `prepare_commit` can infer one bucket
+//! count from the complete batch.
+
+use crate::spec::{
+ batch_to_serialized_bytes, BucketFunctionType, CoreOptions, DataField,
EMPTY_SERIALIZED_ROW,
+ POSTPONE_BUCKET,
+};
+use crate::table::bucket_function::{batch_bucket_ids,
validate_bucket_function};
+use crate::table::postpone_bucket::binary_row_batch_size;
+use crate::table::{SnapshotManager, Table, TableScan};
+use crate::Result;
+use arrow_array::{RecordBatch, UInt32Array};
+use std::collections::HashMap;
+
+pub(super) struct PostponeBucketBatch {
+ pub(super) partition: Vec<u8>,
+ pub(super) bucket: i32,
+ pub(super) batch: RecordBatch,
+}
+
+/// Planning state for a single fixed-bucket batch write to a postpone table.
+pub(super) struct PostponeFixedBucketWriter {
+ partition_field_indices: Vec<usize>,
+ bucket_key_indices: Vec<usize>,
+ bucket_function_type: BucketFunctionType,
+ max_parallelism: i32,
+ target_rows_per_bucket: Option<i64>,
+ target_size_per_bucket: i64,
+ metadata_loaded: bool,
+ known_bucket_counts: HashMap<Vec<u8>, i32>,
+ postpone_row_counts: HashMap<Vec<u8>, i64>,
+ buffered_batches: HashMap<Vec<u8>, Vec<RecordBatch>>,
+ /// Bucket counts used by this prepare-commit round.
+ bucket_counts: HashMap<Vec<u8>, i32>,
+ prepare_started: bool,
+}
+
+impl PostponeFixedBucketWriter {
+ pub(super) fn new(
+ table: &Table,
+ partition_field_indices: Vec<usize>,
+ bucket_key_indices: Vec<usize>,
+ bucket_function_type: BucketFunctionType,
+ ) -> Result<Self> {
+ let schema = table.schema();
+ let options = CoreOptions::new(schema.options());
+ let total_buckets = options.bucket();
+ if total_buckets != POSTPONE_BUCKET ||
schema.primary_keys().is_empty() {
+ return Err(crate::Error::Unsupported {
+ message: format!(
+ "Postpone fixed-bucket writes require a primary-key table
with bucket=-2, but table '{}' has bucket={total_buckets}",
+ table.identifier().full_name()
+ ),
+ });
+ }
+ if options.deletion_vectors_enabled() {
+ return Err(crate::Error::Unsupported {
+ message: format!(
+ "Table '{}' cannot use postpone fixed-bucket writes with
deletion-vectors.enabled=true because deletion-vector scans skip the level-0
files produced by batch writers; use the normal postpone writer or disable
deletion vectors",
+ table.identifier().full_name()
+ ),
+ });
+ }
+
+ let bucket_key_fields: Vec<DataField> = bucket_key_indices
+ .iter()
+ .map(|&index| schema.fields()[index].clone())
+ .collect();
+ if !bucket_key_fields.is_empty() {
+ validate_bucket_function(bucket_function_type,
&bucket_key_fields)?;
+ }
+
+ Ok(Self {
+ partition_field_indices,
+ bucket_key_indices,
+ bucket_function_type,
+ max_parallelism:
options.postpone_batch_write_fixed_bucket_max_parallelism()?,
+ target_rows_per_bucket:
options.postpone_target_row_num_per_bucket()?,
+ target_size_per_bucket: options.postpone_target_size_per_bucket()?,
Review Comment:
`postpone.target-size-per-bucket` is parsed and validated even when
`postpone.target-row-num-per-bucket` is configured. The option contract says
that the size target is ignored in that case, and the PyPaimon planner only
reads it in the `row target is None` branch. With a valid row target plus an
invalid or zero size target, Rust currently rejects writer creation even though
the size value is unused. Please parse and validate the size target only when
no row-count target is present.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
