Skip to content

[core] Preserve append writer configuration per bucket across prepares - #1188

Open
JingsongLi wants to merge 1 commit into
apache:mainfrom
JingsongLi:codex/execution-protocol-unification
Open

JingsongLi wants to merge 1 commit into
apache:mainfrom
JingsongLi:codex/execution-protocol-unification

Conversation

@JingsongLi

@JingsongLi JingsongLi commented Oct 21, 2026 •

Copy link
Copy Markdown
Contributor

Purpose

Align append writer reconfiguration with current Java BaseAppendFileStoreWrite: settings configure future partition/bucket writers, while existing writers retain their write columns and Blob consumer across prepares.

Changes

  • Capture append schemas, fields and Blob handlers per partition/bucket, including after physical files are prepared.
  • Accept full-table or matching captured inputs and preflight every destination before staging a batch. Preserve named Arrow column semantics; incompatible equal-width subsets are rejected instead of being reinterpreted positionally.
  • Keep row-kind filtering before routing and payload conversion. Handle historical types with routing/event fields deterministically.
  • Add synchronous, read-only TableWrite.write_fields_for_row and private Python _row_write_columns transport helpers. Python row conversion can encode the actual destination's columns without duplicating bucket state or touching unused Blob payloads.
  • Traverse replaced Python Blob callbacks retained by Core, allowing writer/callback cycles to be collected. Preserve prepared files and files that may have escaped to a captured consumer on uncertain failures.

URI reader factory configuration retains its existing pre-write restriction.

Verification

  • 27 focused Core integration tests: partial writes, filtered inputs and precomputed buckets, including commit/readback across reconfiguration and multiple prepares.
  • Core failure-cleanup unit test (including replaced consumers).
  • Embedded Python callback GC/query unit test for both batch and stream writers.
  • Rust 1.98 Core and Python-binding Clippy with -D warnings; cargo fmt --check; git diff --check.

No Python extension was rebuilt or installed for these tests. The companion PyPaimon PR apache/paimon#10717 adopts the private row transport helper and unifies execution call boundaries.

@JeonDaehong JeonDaehong left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Checked the "existing writers retain their settings" part against Java (paimon-bundle 2.1.0). It matches while the writer is alive. After a bucket sits idle for one committed round, Java drops the writer and the next write uses the current settings, while here the old settings stay. Details inline.

.cloned()
.unwrap_or_else(|| self.append_write_config());
let writer = self.create_append_writer(partition_path, bucket, &config)?;
self.append_write_configs.insert(key, config);

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nothing removes an entry from append_write_configs, so a partition/bucket keeps its first settings for the whole life of the TableWrite. Java only keeps them while the writer is alive. AbstractFileStoreWrite.prepareCommit drops a writer that wasn't modified since the last committed identifier, and the next write to that bucket creates a writer with the current writeType / BlobConsumer.

I checked with paimon-bundle 2.1.0, DE append table (pt, id, a, b) partitioned by pt:

  1. Write (1, 1, 10, 100), then prepareCommit + commit.
  2. Write (9, 9, 9, 9) to another partition, then prepareCommit + commit. This is the round where pt=1 is idle.
  3. withWriteType(pt, id, a), write (1, 2, 20), commit.

Java reads back pt=1 id=2 a=20 b=null. With no idle round in between, Java fails in step 3 (ArrayIndexOutOfBoundsException), which matches what this PR keeps.

The same sequence here (with_write_type([pt, id]), write pt=1, commit, write pt=9, commit, with_write_type([pt, omitted]), write pt=1) fails with field name mismatch at index 1: expected 'id', actual 'omitted', after 0, 1 or 2 idle rounds.

Should a config be dropped once its bucket had no writes in a committed round, like the Java writer? That would also bound this map for a long-running writer that touches many partitions.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants