Skip to content

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

Merged
JingsongLi merged 5 commits into
apache:mainfrom
JingsongLi:codex/execution-protocol-unification
Oct 22, 2026
Merged

JingsongLi merged 5 commits into
apache:mainfrom
JingsongLi:codex/execution-protocol-unification

Conversation

@JingsongLi

@JingsongLi JingsongLi commented Oct 21, 2026 •

Copy link
Copy Markdown
Contributor

Purpose

Keep one declared append write type throughout a TableWrite lifetime and preserve Java's Blob consumer lifecycle. Current Java callers configure their row type before writing; bucket writers capture that schema and consume rows positionally.

Changes

  • Configure with_write_type before the first effective write. Repeating the same ordered selection is a no-op; changing columns or order after writing is rejected, including after prepare or idle cleanup. Empty and ignored input does not lock the configuration.
  • Remove _row_write_columns, its Python stubs and Core resolver, along with historical schema lookup and per-bucket write-column caches. Native rows encode the selected columns and required routing/event fields in one batch.
  • Accept full-table Arrow input or the current selected layout with routing fields. Preserve selected column order and validate input before staging; omitted payloads, including live Blobs, remain untouched.
  • Capture Blob handlers per partition/bucket across consecutive active prepares. Retire idle unaware writers; fixed buckets require a strictly newer committed identifier from the same user. Filesystem lookup stops at expired snapshot gaps and propagates other errors before draining pending output.
  • Keep the ordinary Core prepare_commit() primitive unchanged. It conservatively retains fixed-bucket context without a checkpoint identifier; bindings forward their existing batch/stream identifiers through the internal bridge.
  • Release retired Blob callbacks promptly, preserve callbacks captured by active buckets, and expose real Python references to cycle collection. Preserve prepared files and consumer-visible Blob files on errors and close.
  • Keep URI reader factories configurable only before writing.
  • Merge current Rust main, including [core] Support scheduled execution of snapshot-scoped search plans #1191. Preserve all search execution bridges alongside the writer identifier bridge and keep one internal module declaration.

Verification

  • 85 Core writer unit tests, including eight lifecycle regressions.
  • Five Blob consumer and cleanup unit tests, including the previously failing Windows case.
  • 40 append, filtering and precomputed-bucket integration tests, including fixed layout, selected transport, empty/ignored input and readback.
  • All 119 embedded Rust Python binding unit tests after merging current main, including callback retirement, GC and prepared-file preservation.
  • Rust 1.98 strict workspace Clippy with fulltext,vortex; formatting and diff checks.

No Native extension was rebuilt or installed into the Python test environments. Matching PyPaimon adapter, Pure Python lifecycle, regression tests and documentation changes are in apache/paimon#10724.

@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.

@leaves12138 leaves12138 left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

LGTM

@JingsongLi
JingsongLi merged commit 77b1d49 into apache:main Oct 22, 2026
16 checks passed
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.

3 participants