Apache Paimon

Status: experimental. The optional streamfusion-paimon module accelerates streaming SELECT reads and INSERT INTO jobs on Paimon append-only tables and primary-key tables with fixed, dynamic, or postpone buckets on the published Paimon 2.0.0 Flink 2.2 connector. The separate Paimon 1.0.0 adapter targets Flink 1.18. Paimon keeps every table-level responsibility: schema and catalog, bucket assignment rules, sequence numbering rules, file rolling, statistics, manifests, snapshots, commits, and compaction. StreamFusion replaces the per-row shuffle in front of the writers, append buffering and spilling, the Parquet/ORC encoding of each data file, and, for primary-key tables, the sort and merge that turns a bucket's changelog into a level-0 file. Postpone staging retains every accepted change for Paimon's separate compactor.

Both file formats share the same Java lifecycle, native routing and merge paths. ORC's physical encoding options and UTC timestamp restrictions are detailed on the ORC page.

Timestamp bundles use lossless millisecond/fraction columns, including years 0001–9999. Sink writers support TIMESTAMP and TIMESTAMP_LTZ through precision 9, including nested values. Parquet follows released Paimon's per-field encoding: INT64 milliseconds for precision 0–3, INT64 microseconds for 4–6, and INT96 for 7–9. Mixed-precision schemas retain each field's physical type and field ID. ORC preserves the nanosecond remainder and the stock writer's historical calendar and LTZ rules; its existing UTC-JVM requirement still applies. Native source and snapshot-reader admission remains limited to precision 6.

Streaming source

Streaming Parquet and ORC reads retain the released Java Paimon client for table/catalog resolution, snapshot and manifest discovery, split assignment, filesystem credentials, and enumerator checkpoint serialization. A task-side reader emits Arrow batches into the native pipeline:

Read phase	Implementation
Append snapshot and subsequent committed data	Native Parquet/ORC decoding into Arrow
Primary-key initial snapshot	Native Parquet/ORC for raw-convertible files; released paimon-rust merge for admitted deduplication/first-row/partial-update splits; Java merge-to-Arrow for other splits
Primary-key changelog tailing	Native Parquet/ORC decoding of value columns and the stored row-kind byte

Native readers use Flink's single-thread fetcher lifecycle and its configured bounded handover queue. The Flink 1.18 and 2.2 adapters obtain the running fetcher from one iterator, checking hasNext() before next(). The released manager's separate Map.isEmpty() check can race with the I/O thread retiring the last fetcher and fail a new split assignment with NoSuchElementException. An empty iterator instead lets Flink create a new fetcher normally; split ordering, checkpoint offsets and shutdown remain managed by the released reader. PaimonFetcherLifecycleTest reproduces that retirement deterministically against both the stock manager and the native adapter. The original lookup/changelog SQL tests still compare their results without retries or relaxed assertions.

Primary-key tailing requires changelog-producer = input, lookup, or full-compaction, and Parquet or ORC changelog files. The producer has already generated the changes; the source preserves +I, -U, +U, and -D without another merge. Default startup reads the current snapshot before following commits. scan.mode = latest starts with new commits only. Remaining snapshot merge coverage is tracked in #53. The snapshot-merge design records the Java planning and Arrow batch interface.

Both source adapters use the canonical crates.io paimon = 0.4.0 public reader. Java passes its already resolved schema, selected file metadata and projection to Table::from_resolved_schema, DataSplit and ReadBuilder; Rust performs no catalog or manifest discovery. The small descriptor bridges different Java split serialization versions; it does not implement a second schema or metadata planner. Historical schemas and deletion vectors still retain Java.

FileIOBuilder::with_provider connects the reader to Paimon's existing seekable Java FileIO through a reusable 64 KiB transfer buffer. The provider binds each original URI directly to Java, using an opaque OpenDAL key so URI normalization cannot change the configured filesystem path. No Rust storage client or separate credentials are configured. paimon-rust delegates projected file decoding to parquet-rs or orc-rust and returns Arrow batches of up to 4,096 rows. It does not load an entire file into Java memory. Arrow C Data exports transfer ownership to the source batch; closing the reader releases the native decoder and Java input stream. Memory includes compressed column chunks, decoder working buffers, and the source's bounded fetch queue; the batch row count is not a hard byte budget.

Top-level projection, reordered columns, nested values, and supported periodic source watermarks are preserved. Query predicates remain Flink residual filters; this first source does not push query predicates into its Java scan or native decoder. Checkpoints save the number of emitted logical rows per split with Paimon's existing serializer. Recovery skips precisely that many rows, including positions within a new batch size. Prefetched rows do not advance the checkpoint. source.operator-uid.suffix uses Paimon's released UID generation, preserving explicit source identity when restoring savepoints after job graph changes.

The reader deliberately retains Java decoding and row-to-Arrow conversion for splits requiring schema evolution, deletion-vector selection, ORC schemas containing CHAR (including nested CHAR), unsafe/unknown ORC timestamp bounds, or files without an installed native codec, and for specialized split representations. External paths must end in the metadata's physical format suffix; query/fragment suffixes or a different extension retain Java because released Rust dispatches by that suffix. The adapter never rewrites such storage paths. This preserves those files' Java read semantics within an admitted source. Primary-key snapshot splits needing unsupported merge semantics retain Java, including its merge engine and delete handling.

Native initial snapshot merge

The released paimon-rust reader owns interval partitioning, sorted runs, decoding and merge functions. StreamFusion no longer carries a copied loser tree, cursor merger or partial-update reducer. Java's IntervalPartition remains a conservative admission check for fan-in and footer memory; it does not drive the Rust merge loop.

AuditLogRead exposes the winning stored row kind for materialized snapshots. A columnar Arrow selection drops winning retracts, then translates the small row-kind column into Flink's byte sidecar. This preserves Java's INSERT/UPDATE_AFTER distinction, which ordinary to_arrow_with_row_kind intentionally normalizes to INSERT for snapshots. Streaming splits use that ordinary method to preserve all four physical change kinds without merging. Payload values stay in Arrow throughout. Snapshot selection, comparison and field reduction remain upstream.

The native merger admits fixed and dynamic hash buckets with deduplicate, first-row, or basic partial-update, the default loser-tree sort engine, and nonempty stored scalar keys: BOOLEAN, TINYINT through BIGINT, FLOAT/DOUBLE, DECIMAL, CHAR/VARCHAR, BINARY/VARBINARY, DATE, and TIMESTAMP/TIMESTAMP_LTZ through precision 6. Composite keys use those same types after Java removes table partition columns from the stored key. Floating comparisons canonicalize NaN payloads and distinguish signed zeros like Java; this source coverage does not broaden the sink's key whitelist. Timestamps above precision 6 retain the stock source. Values use the supported source types. Files must use the current schema and Parquet/ORC without deletion vectors.

Deduplication compares optional sequence.field columns in the configured sequence.field.sort-order, with nulls first in either direction, then stored sequence numbers. Sequence columns use the comparable types above and are decoded even when projected away. First-row selects the earliest stored sequence. Both engines honor ignore-delete; otherwise first-row requires every file to report zero retracts, leaving Java to handle unsupported input. Potential stored-sequence ties retain Java: file sequence ranges within a section must be non-overlapping, and a file must have at least as many sequence positions as physical rows. This conservative metadata gate avoids the released readers' different equal-sequence traversal without implementing a custom tie breaker. It can decline files whose actual keys never tie.

Basic partial-update selects the latest non-null cell for each column. It supports ignore-delete and partial-update.remove-record-on-delete, including stored historical retracts and later reinsertion for the remove-record policy. Partial updates with ignore-delete=true retain Java when any file has stored or unknown retracts: released Java can initialize cells from the first retract, whereas the Rust reader skips it. Ordinary add-only partial-update files remain native. Without either delete policy, files must have a known zero delete count. Multi-record partial reductions emit inserts or remove the key; singleton groups preserve their original row kind, matching Java's reducer wrapper. Nested values are selected as whole cells, preserving their internal nulls. The upstream reader returns Arrow batches without Java payload rows; its internal allocation strategy remains upstream. Sequence groups and configured field aggregates retain Java.

ORC timestamps need known top-level bounds within ±9,000,000,000,000 epoch milliseconds, or a known all-null column. Nested timestamp columns and missing/out-of-range bounds retain Java: the released orc-rust 0.8 decoder first materializes nanoseconds, which cannot represent Paimon's full year range. The bound is deliberately inside that nanosecond range. This gate also applies to raw append/changelog files, preventing a decode failure after entering the native path. ORC retains Java for fractional timestamp key bounds that can lose sorted order through its last-negative-second alias, and for non-leading fractional timestamp keys; see the ORC timestamp restrictions. These snapshot combinations retain Java:

• Aggregation; partial updates with sequence groups, removal by sequence group, or configured field aggregates; postpone and cross-partition key-dynamic buckets; unsupported key/sequence types; and specialized split representations.
• First-row files with nonzero or unknown delete counts when ignore-delete is false, and basic partial-update files with those counts when neither supported delete policy is enabled; stored/unknown retracts also retain Java for partial updates with ignore-delete=true.
• Potential sequence ties; ORC schemas containing CHAR because released padding differs from Java.
• Sections exceeding the smaller of 256 runs and sort-spill-threshold, or the encoded row-group admission budget below.

For raw-convertible primary-key snapshot splits with known delete counts and no deletion vectors, the source follows Java's raw reader: native decoded values emit insert rows. This path does not need a merge. Selection merges drop winning retracts and retain the winning add kind; partial-update reductions follow the row-kind rules above. Changelog tailing continues to preserve all stored changes.

sort-spill-buffer-size supplies the snapshot merge read resource budget. Raw append/changelog decoding keeps the existing unbounded resource context rather than acquiring an unrelated merge limit; batch rows and the handover queue still bound batching, not total bytes. Before emission, the existing footer check requires the sum of each run's largest compressed-plus-uncompressed row group (Parquet), or conservative stripe/dictionary estimate (ORC), to fit half that budget. Larger splits retain Java's spill-capable reader. paimon-rust's ResourceContext charges projected Parquet row groups; it does not bound all merge state, ORC allocations or output. peakRetainedBytes() reports that upstream reservation, not the previous copied merger's total retained-buffer estimate.

There is no native merge spill or strict total-process/Flink managed-memory reservation. Decoder and output allocations, wide records, Java stream buffers and the handover queue add memory. A resource or storage failure after emission fails the attempt; it never silently switches readers in the middle of a split. Closing the reader releases every Java stream; already exported Arrow batches remain valid until their consumer releases them. JNI preserves the original Java storage exception, and interrupted reads fail at batch and I/O boundaries.

Recovery still checkpoints emitted logical rows after merging and delete removal. Replaying the same split and skipping that prefix supports changes in batch size and Java/native restoration in either direction. It rereads the prefix rather than persisting native pointers or per-file version offsets.

PaimonSnapshotMergeTest checks stock- and native-written overlapping snapshots, nested values, projections omitting keys, partitioned composite keys, Java/native restoration and admission fallbacks. PaimonSnapshotKeyTypesTest covers admitted key types across Parquet and ORC with stock/native writers (floating keys use stock writers), signed integer extremes, all three Parquet decimal physical encodings, Unicode and binary prefix ordering, pre-epoch dates/timestamps, timestamp precisions 0–6, projections and restoration in both Java/native directions. Nanosecond keys exercise Java fallback even when projected away. PaimonSnapshotModesTest compares fixed/dynamic buckets, first-row, stored retracts with ignore-delete, ascending/descending compound sequences, nulls, floating edge cases and scalar sequence types against Java, including projections and restoration in both directions. PaimonSnapshotMergeOracleTest writes files with the released Java writer and compares projected reads across merge/delete policies, ascending/descending sequences, missing keys, 1–16 runs and batch sizes. Tied files, including repeated keys within a run, explicitly require Java fallback. PaimonSnapshotPartialUpdateTest checks actual Parquet/ORC sequence ties and all supported value types, fixed/dynamic buckets, projection and cross-reader restoration. PaimonSourceSqlTest follows snapshots into subsequent commits; PaimonSourceRecoveryTest restores before and after emission. PaimonRustReaderTest checks original storage exceptions, stream cleanup, original URI handling and Arrow batch ownership after reader close. The SQL harness requires a marker proving that the native snapshot reader emitted a batch. Upstream SQL assertions remain unchanged.

The previous copied-merger implementation passed 653 regression cases and eight unchanged upstream continuous-reader, partial-update and read/write SQL cases with required native markers. The complete Paimon 2.0.0 SQL harness previously passed all 265 result, plan and recovery cases with the complete-plan native hook, including the unchanged source-reuse assertion. A focused subset is shown below. Include the continuous-reader cases when checking the harness's required native snapshot-emission marker; the other suites can finish without needing that path.

FLINK_SUITE_TEST='org.apache.paimon.flink.ReadWriteTableITCase,org.apache.paimon.flink.PrimaryKeyFileStoreTableITCase,org.apache.paimon.flink.FlinkJobRecoveryITCase,org.apache.paimon.flink.ContinuousFileStoreITCase#testWithPrimaryKey+testProjectionWithPrimaryKey' \
  bin/flink-suite.sh paimon

Snapshot catch-up diagnostic

PaimonSnapshotMergeBenchmark writes 65,536 keys over one, four and eight commits for INT, and four commits for other scalar keys and first-row, partial-update, sequence and dynamic-bucket cases, including updates, deletes, strings and nested arrays, before timing. It compares the existing Java merge-to-Arrow path with native snapshot reading, including Java run planning and footer checks, file open/close, byte reads, decoding, merging, final Arrow import and an integer payload checksum. Keys span negative and positive values; a separate integer payload identifies both key and commit so the checksum checks version selection and uses the same work across key types. Both paths produce Arrow. One warmup and three measured runs alternate engine order and report best times. Partial-update cases alternate null string and array columns so output combines different commits, with default, ignore-delete and whole-record-delete policies. The sequence case orders by this payload so earlier commits can win; first-row ignores input retracts, and the dynamic case uses two explicitly assigned buckets. Counts and checksums must match, and every native run must exercise the merger.

The following historical results describe the previous copied merger. The released-reader comparison below supersedes them for the current implementation.

Parquet case	Commits	Java merge to Arrow	Native merge to Arrow	Throughput ratio
INT	1	0.042 s	0.018 s	2.40×
INT	4	0.072 s	0.041 s	1.75×
INT	8	0.122 s	0.078 s	1.57×
DECIMAL(38,2)	4	0.088 s	0.044 s	1.99×
TIMESTAMP(6)	4	0.065 s	0.042 s	1.55×
VARBINARY(4)	4	0.079 s	0.046 s	1.72×
DATE	4	0.070 s	0.039 s	1.77×
FLOAT	4	0.067 s	0.040 s	1.69×
DOUBLE	4	0.067 s	0.043 s	1.58×
First-row + ignore-delete (INT)	4	0.074 s	0.038 s	1.96×
User sequence (INT)	4	0.082 s	0.036 s	2.24×
Dynamic buckets (INT)	4	0.072 s	0.037 s	1.97×
Partial update (INT)	4	0.071 s	0.040 s	1.78×
Partial update + remove-record-on-delete (INT)	4	0.069 s	0.038 s	1.82×
Partial update + ignore-delete (INT)	4	0.069 s	0.039 s	1.79×

Four-commit cases with ORC and a 256 mb admission budget:

ORC case	Java merge to Arrow	Native merge to Arrow	Throughput ratio
FLOAT	0.064 s	0.038 s	1.68×
DOUBLE	0.062 s	0.036 s	1.72×
First-row + ignore-delete (INT)	0.066 s	0.032 s	2.06×
User sequence (INT)	0.066 s	0.032 s	2.03×
Dynamic buckets (INT)	0.059 s	0.027 s	2.18×
Partial update (INT)	0.063 s	0.035 s	1.79×
Partial update + remove-record-on-delete (INT)	0.061 s	0.033 s	1.83×
Partial update + ignore-delete (INT)	0.064 s	0.034 s	1.89×

These release measurements describe local snapshot catch-up, not whole-job or remote-storage performance. Run with:

SF_PAIMON_SNAPSHOT_BENCHMARK=true mvn test -Pbench,paimon \
  -pl :streamfusion-paimon -am -Dtest=PaimonSnapshotMergeBenchmark \
  -Dsurefire.failIfNoSpecifiedTests=false

SF_PAIMON_SNAPSHOT_ROWS changes the key count; SF_PAIMON_SNAPSHOT_CASES selects comma-separated case names (for example partial-update,partial-delete,partial-ignore for the new partial-update cases). SF_PAIMON_FILE_FORMAT=orc selects ORC. SF_PAIMON_SNAPSHOT_BUDGET sets sort-spill-buffer-size for both readers (default 64 mb). The 65,536-key multi-run ORC fixture exceeds the default footer admission budget; the ORC measurements use 256 mb. Native defaults and fallback gates are unchanged; the eight-run ORC control still falls back at 256 mb. A benchmark fails if its native run falls back to Java.

Released-reader migration diagnostic

The 2026-11-03 comparison uses an Apple M4 (16 GiB), JDK 17, Rust 1.95.0 release with mimalloc, one test JVM, the default 8 GiB heap and visible CPU count, and the existing production byte-range JNI/FileIO and Arrow import. The previous implementation is canonical main 5d628b6a; the candidate uses canonical paimon-rust 0.4.0. Both run the same fixture/harness, writer settings, values, projection and resources, outside concurrent builds/tests. Files are written before timing. One warmup and three measured observations rotate reader order. Cells are median [minimum, maximum] in milliseconds; raw observations include warmup, row counts, checksums and native merge evidence. Final raw controls use separate benchmark-only JVMs (final-isolated in the CSV); initial combined runs and the intermediate validation run are retained under their own phases. Raw reads keep an unbounded resource context, as before this migration. Snapshot budget settings are unchanged. Every required native measurement performed native work.

File reads use 262,144 rows and four projected columns, including a nested array. Stock rows checksum the id directly; Java-to-Arrow additionally pays the production fallback's copy/conversion. The latter matches the native pipeline's Arrow output boundary. These are local diagnostics, not whole-job or remote-storage gains.

Format / read	Stock Java rows	Java to Arrow	Previous native	Released reader
parquet / append	24.49 [21.97, 26.48]	75.48 [63.04, 76.51]	18.03 [17.28, 20.16]	17.59 [15.98, 18.66]
parquet / changelog	23.82 [19.91, 28.08]	64.31 [60.84, 83.75]	12.01 [11.95, 12.38]	10.75 [10.66, 11.14]
orc / append	17.75 [17.21, 20.39]	55.99 [55.79, 56.09]	20.12 [18.57, 21.36]	21.88 [20.60, 22.77]
orc / changelog	17.96 [13.13, 21.26]	66.13 [57.11, 82.37]	10.81 [10.73, 11.00]	12.82 [12.19, 13.15]

Snapshots use 65,536 keys, strings and nested arrays. Both paths output Arrow and checksum the same integer payload. Budgets are 64 MiB for Parquet and 256 MiB for ORC, unchanged between implementations. Java controls below are the candidate run's stock merge plus production conversion to Arrow.

Format / merge / commits	Java to Arrow	Previous native	Released reader
parquet / int / 1	25.09 [23.29, 28.34]	11.00 [9.01, 12.66]	14.13 [12.57, 14.66]
parquet / int / 4	46.61 [46.55, 48.25]	25.81 [25.29, 26.20]	24.84 [24.32, 24.92]
parquet / int / 8	79.23 [78.70, 81.12]	49.24 [48.97, 49.77]	44.48 [44.39, 45.29]
parquet / first-row / 4	45.49 [45.47, 50.81]	23.84 [23.67, 24.27]	21.92 [21.64, 22.96]
parquet / sequence / 4	51.23 [50.87, 55.11]	23.99 [23.91, 24.63]	24.84 [24.05, 25.75]
parquet / partial-update / 4	49.40 [47.66, 54.55]	25.42 [24.36, 25.71]	41.00 [40.72, 41.46]
parquet / partial-delete / 4	47.10 [45.77, 49.61]	24.05 [23.98, 24.43]	39.55 [38.75, 40.11]
parquet / partial-ignore / 4	47.29 [46.44, 47.60]	24.37 [24.21, 24.67]	40.33 [39.81, 41.69]
orc / first-row / 4	41.13 [38.79, 44.62]	22.78 [22.59, 22.88]	25.80 [24.55, 27.08]
orc / sequence / 4	40.51 [40.13, 43.08]	20.92 [20.31, 21.05]	25.90 [24.98, 25.94]
orc / partial-update / 4	42.45 [41.13, 44.22]	20.13 [19.86, 20.17]	41.04 [40.87, 41.75]
orc / partial-delete / 4	41.45 [40.93, 41.69]	19.78 [19.59, 20.01]	39.28 [39.09, 40.71]
orc / partial-ignore / 4	41.92 [41.65, 42.12]	19.82 [19.73, 20.37]	39.92 [39.21, 40.72]

Public reader reuse keeps multi-run deduplication/first-row competitive. Raw-read medians vary against the previous implementation; the one-commit Parquet snapshot and ORC merges also regress. Partial updates regress substantially against our removed custom gatherer: upstream materializes per-key reductions before assembling batches. Their measured benefit over Java-to-Arrow is about 1.2× for Parquet and only 1.03–1.06× for ORC. ORC append does not beat the Java row-only checksum; it does beat Java-to-Arrow. This change prioritizes released reader reuse and deleting local algorithms; it is not a blanket throughput improvement. Writer behavior and writer benchmarks are unchanged. The old in-memory Criterion merger probe was removed with its implementation; use the production file-reader diagnostics above for this boundary.

The existing benchmark commands reproduce this comparison. Each now logs individual *_SAMPLE observations; iteration 0 is warmup. Set SF_PAIMON_SNAPSHOT_CASES to int,first-row,sequence,partial-update,partial-delete,partial-ignore for Parquet, omit int for ORC, and set SF_PAIMON_SNAPSHOT_BUDGET='256 mb' for ORC. Keep the same harness on the previous revision. Explicit native assertions fail if a selected fixture falls back.

At-least-once consumer retention

Paimon 2.0 streaming sources admit consumer-id with explicit consumer.mode=at-least-once and a configured consumer.expiration-time. The existing source type, format, projection and watermark gates still apply. Other explicit consumer.* options retain fallback until verified.

Paimon's released continuous enumerator owns the durable consumer cursor and updates it only when Flink completes a checkpoint. Native readers report completed split snapshots through the same progress events as stock readers. An uncompleted newer checkpoint cannot advance an older completed cursor. Restored readers retain their snapshot until progress is reported; snapshot expiration remains owned by Paimon and respects that cursor.

A fresh job starts at the persisted consumer snapshot. This at-least-once cursor can replay a partially consumed snapshot; regular Flink checkpoint recovery also restores each split's emitted-row offset. Coverage compares stock/native cursor restart, nullable append and primary-key rows, capture versus completion, enumerator serialization/restore, expiration protection/release, and recovery inside a native batch.

Default or explicit exactly-once consumer mode remains on stock Paimon: its released source builder selects dedicated split generation, which is a different runtime protocol. Missing expiration configuration retains Paimon's validation error. Broader consumer modes remain tracked in #27.

Consumer-retention performance

A released Flink 2.2.1 / Paimon 2.0.0 comparison on 2026-09-28 used JDK 17, release Rust with mimalloc, parallelism 1, a 2 GiB JVM heap, 100 ms checkpoints, 1,024-row native batches and Paimon's default Zstandard compression. Each mode had two warmups and five measured jobs, rotating execution order. Input contains unique BIGINT IDs, nullable STRING payloads (4,096 distinct non-null values, roughly 60 bytes each) and nullable BIGINT amounts. Append tables use bucket -1; primary-key tables use bucket 1 and input changelogs. The bounded streaming scan reads one data snapshot and stops at the following watermark sentinel.

Timing includes planning, job startup, file reads, JNI, Arrow-to-RowData conversion and job completion. Fixture creation and independent plan/decode/result checks are outside timing. Append jobs use a SQL blackhole sink. Primary-key jobs use toChangelogStream and a rowwise DataStream discard sink, retaining the external Row conversion and all changelog kinds. A primary-key SQL blackhole plan inserts DropUpdateBefore, which currently triggers the shared whole-island fallback; these primary-key timings do not establish acceleration for that SQL sink.

“Previous source” enables StreamFusion but disables its Paimon source substitution, reproducing the earlier host source route on base 665dc10d; it is an ablation, not a separately rebuilt historical release. The harness verifies native source substitution in every native job and native file decoding independently.

Median seconds (minimum–maximum):

Format / table	Rows	Stock Flink	Native	Previous source
Parquet / append	200,000	0.183 (0.174–0.187)	0.104 (0.101–0.171)	0.125 (0.116–0.191)
Parquet / append	2,000,000	0.415 (0.398–0.420)	0.279 (0.262–0.325)	0.411 (0.400–0.428)
Parquet / primary key	200,000	0.200 (0.197–0.205)	0.167 (0.161–0.171)	0.201 (0.196–0.278)
Parquet / primary key	2,000,000	1.162 (1.153–1.166)	0.793 (0.787–0.804)	1.161 (1.146–1.168)
ORC / append	200,000	0.168 (0.160–0.171)	0.104 (0.098–0.164)	0.121 (0.110–0.170)
ORC / append	2,000,000	0.381 (0.370–0.386)	0.268 (0.263–0.276)	0.383 (0.370–0.385)
ORC / primary key	200,000	0.187 (0.186–0.201)	0.164 (0.159–0.169)	0.191 (0.188–0.195)
ORC / primary key	2,000,000	1.096 (1.082–1.354)	0.798 (0.771–0.898)	1.072 (1.060–1.085)

The two-million-row cases reduce median time by 27–33% versus stock and 26–32% versus the previous route. These are bounded jobs, not sustained streaming throughput measurements; especially the 200,000-row results are sensitive to startup and JVM warmup. The primary-key fixture contains inserts, while updating changelog and checkpoint correctness are covered separately by the source tests.

All individual trials include an earlier Parquet pilot marked pilot_per_job_explain. That pilot explained the native plan immediately before each native job, creating asymmetric planner warmup. Its 200,000-row native median was 0.162 s versus 0.172 s stock and 0.115 s previous source (40% slower than the latter). The final harness validates the plan in a separate environment before all modes and retains this unfavorable pilot rather than mixing the two methods.

Reproduce with SF_BENCHMARK=true SF_PAIMON_FILE_FORMAT=parquet mvn -pl streamfusion-paimon -am test -Ppaimon,bench -Dtest=PaimonConsumerBenchmark -Dsurefire.failIfNoSpecifiedTests=false -Dconsumer.rows=2000000 -Dconsumer.primaryKey=false -Dsf.extraJvmArgs=-Xmx2g using JDK 17. Repeat with orc, consumer.primaryKey=true and consumer.rows=200000 for the other cases.

Source admission and fallbacks

Only streaming data-table reads enter this path. It uses the sink's supported value types below, except that timestamps remain limited to precision 6, including inside nested values. These configurations retain the stock source at planning time:

• Primary-key tables without a changelog producer, table/changelog formats without an installed native Parquet or ORC module, thin data files, data evolution, row tracking, chain tables, query authorization, and exposing internal key-value sequence numbers.
• Nested subfield pruning, zero-column projections, metadata columns, pushed limits/aggregates, and source abilities other than top-level projection, residual filters, and supported watermarks.
• Consumer retention outside the verified at-least-once configuration above, dedicated split generation, checkpoint/snapshot alignment, and postpone.merge-on-read.
• Source watermarks outside the shared periodic constant-interval expression contract, including on-event emission and watermark alignment. Non-negative day-time and YEAR/MONTH/YEAR TO MONTH delays are supported, including chained subtractions; calendar intervals use Flink's month-end and leap-year arithmetic. Serialized expression plans are evaluated by each reader through the shared watermark evaluator, with native handles closed when the reader closes.
• Unverified scan.*, streaming-read-*, log.*, and custom parquet.*/orc.* settings. ORC also admits the boolean orc.timestamp-ltz.legacy.type; timestamp schemas require a UTC JVM timezone. The admitted scan settings are scan.mode, scan.snapshot-id, scan.timestamp-millis, scan.timestamp, scan.tag-name, scan.watermark, scan.bounded.watermark, scan.parallelism, scan.infer-parallelism, scan.infer-parallelism.max, scan.remove-normalize, and supported watermark idle/emit settings.

streamfusion.operator.paimonSource.enabled=false disables the substitution. The selected Parquet or ORC module must be installed alongside the Paimon module, as for the sink. Native format-factory discovery is not required for this reader: it calls the decoder directly on Java-planned files.

Repeated compatible native scans share one reader, including scans with different top-level projections and scans feeding different sinks in a statement set. The deployed streaming planner lets Flink union their projected columns before native substitution, then places the reader under an explicit Arrow share operator. Every branch takes a retained buffer view, preserving row kinds and the normal source watermark/checkpoint flow. Parquet and ORC use the same sharing path. When a watermarked split finishes, the reader flushes its final maximum watermark candidate through Flink's split output before releasing it. Candidates are calculated per row before taking the maximum, including for calendar delays, and carried separately from event timestamps. Otherwise a tiny native file can finish before a periodic tick and lose its final watermark. Active files retain periodic emission, and Flink still combines concurrent splits and handles idleness. Filters, limits, startup hints, table options, schemas and other scan semantics must agree; different scans remain independent. A branch that falls back to Flink is excluded from the native consumer count. Disabling streamfusion.plan.shareSources, table.optimizer.reuse-source-enabled or table.optimizer.reuse-sub-plan-enabled restores independent native readers.

PaimonSourceSharingTest checks three-sink projection union, identical projections, repeated compilation, disabled sharing, distinct scans, mixed native/Flink branches, and snapshot-to-tail results including deletes and nulls for both formats, with chained and network edges. The admitted ABS projection participates in three-consumer sharing; a STRING-to-BOOLEAN TRY_CAST projection retains a separate host branch and two native consumers. Both shapes execute snapshot/tail comparisons against Flink. The suite also checks window closure on every shared branch after a new commit. Paimon's unchanged ContinuousFileStoreITCase.testSourceReuseWithScanPushDown passes its Reused assertion and its filter/limit separation assertions. Cross-sink sharing requires the deployed planner hook; installing a program into an already constructed stock planner still optimizes each root separately. Other source gaps remain in #27.

PaimonSourceReadTest checks append/primary-key twin readers, row kinds, nested values, partition values, historical schema mapping, projections, latest startup and within-batch resume. PaimonSourceRecoveryTest checkpoints the asynchronous source reader after emission and restores with Paimon's serialized split state. PaimonSourceSqlTest follows an initial snapshot and a later commit through streaming SQL, including source watermarks; PaimonSourceAdmissionTest checks fallback and watermark admission.

Source file-reader diagnostic

The opt-in PaimonPartitionPruningBenchmark isolates partition predicate handoff before SQL planner wiring. It reads the same committed append table with 32 non-null string partitions, 131,072 rows, and 64/256-byte payloads through the current native full scan, a partition-pruned released ReadBuilder feeding the production native split reader, and the released pruned Java reader. All paths retain a residual pt = 'p00' check and verify identical selected-row counts and id checksums. The timer includes scan planning, file reads, decode, and the residual/checksum; table creation and commit are outside it. Reports distinguish planned file counts/bytes, physically decoded rows, and native file-read witnesses; planned bytes are not measured I/O bytes.

Run with SF_PAIMON_PARTITION_PRUNING_BENCHMARK=true, -Pbench, and -Dtest=PaimonPartitionPruningBenchmark in the Paimon module. Optional SF_PAIMON_PARTITION_ROWS and SF_PAIMON_PARTITION_REPEATS select size and repetitions; each mode warms up once and retains every measured observation. This fixture measures the available split-pruning boundary, not current SQL source pushdown or an end-to-end job speedup. The companion PaimonPartitionSqlBenchmark supplies that SQL job boundary: set SF_PAIMON_PARTITION_SQL_BENCHMARK=true and select its test with -Pbench,paimon. It compares released stock and current native source jobs on one immutable table for pt = 'p00', pt = 'p31', and an unfiltered query, and asserts the source transformation plus every collected row and checksum. The two selective cases exercise opposite partition-name ends, without assuming Paimon schedules partitions lexically. The fixture logs the released scan's planned partition order and each query's partition tag and native pruning-plan witness. Set SF_PAIMON_PARTITION_REQUIRE_PRUNING=true for candidate runs to require that witness; leave it unset when running the same fixture against the previous source implementation. Its timer includes execution startup, collection, and cancellation, with Arrow-to-row conversion retained. Planning/setup and result validation are outside timing. Defaults are 32,768 rows and three measured runs after one warmup; use the same row/repetition variables for before/after runs. These short jobs can be startup dominated and must not be described as sustained decode throughput. Each streaming query collects its expected initial rows and then cancels the live iterator. A selective query can finish before unrelated partitions are decoded; its timer measures the latency to that result prefix, not a complete unfiltered snapshot scan. Only the unfiltered case collects every snapshot row. A fast early partition can therefore hide an otherwise expensive full-scan source, which is why both selective partition names are kept. The released FIFO split assigner polls its supplied order, while pre-assignment can group splits; partition-name ordering is not a source API guarantee. Do not reorder splits or add artificial delays to make the pruning comparison favorable.

Final paired 262,144-row, five-observation SQL runs prove selective native planner hints and exact results, but provide no meaningful end-to-end gain: p00 native medians 247.428→243.529 ms, p31 243.779→239.002 ms, and unfiltered 897.027→901.372 ms. Released plan order puts p00/p31 at positions 18/29, not schedule endpoints; stock startup and unfiltered outliers vary materially. The full-drain fixture independently verifies 32→1 files and 131,072→4,096 decoded rows for partition pruning. The optimization ledger retains all ranges/IQRs and links to raw micro and SQL observations; these reductions must not be presented as a SQL throughput improvement.

The opt-in PaimonSourceBenchmark compares three paths on 262,144 rows and four projected columns, including a nested array: released Java reading to rows, Java reading followed by conversion to Arrow, and native Parquet/ORC reading to Arrow. Files are generated before timing. Each read includes file open/close, decode, and an id-column checksum; native reads also include the Java FileIO/JNI transfer and Arrow import. The Java-to-Arrow baseline uses the existing fallback's row ownership copy and converter, with the same 4,096-row batch size and changelog sidecar as the native reader. Both Arrow paths materialize every projected column and checksum the resulting id vector. The Java row scan checksums the id directly and does not construct Arrow output. It verifies row counts and checksums across all three paths. One warmup and three measured runs rotate execution order and report each path's best time. speedup compares native with Java rows; arrow_speedup compares native with Java-to-Arrow. Both Arrow endpoints in this older diagnostic are Java Arrow, including a native-to-Java import; they do not directly measure feeding our Rust operators. The ORC reader comparison ends the production Rust reader and Java baseline at an arrow-rs RecordBatch consumed in Rust. This is a local file-reader diagnostic, not an end-to-end Flink or remote-storage benchmark.

Historical Apple Silicon measurements for the previous file reader, with release libraries and mimalloc (the current released-reader comparison is above):

Format	Path	Java rows	Java to Arrow	Native to Arrow	Native / Java-to-Arrow throughput
Parquet	Append files	0.032 s	0.130 s	0.026 s	5.00×
Parquet	Primary-key changelog files	0.035 s	0.124 s	0.019 s	6.70×
ORC	Append files	0.027 s	0.104 s	0.057 s	1.82×
ORC	Primary-key changelog files	0.020 s	0.095 s	0.047 s	2.01×

ORC remains slower than Java when the consumer only needs the row checksum. At the Arrow output boundary both native formats beat the current Java-to-Arrow path in this diagnostic. This does not isolate codec speed from representation conversion or establish whole-job throughput.

Run with release native libraries:

SF_PAIMON_SOURCE_BENCHMARK=true mvn test -Pbench,paimon \
  -pl :streamfusion-paimon -am -Dtest=PaimonSourceBenchmark \
  -Dsurefire.failIfNoSpecifiedTests=false

SF_PAIMON_SOURCE_ROWS changes the input count; SF_PAIMON_FILE_FORMAT=orc selects ORC.

Streaming sink

A sink's Arrow batches are split natively into one sub-batch per (partition, bucket) pair. Rust computes the partition BinaryRow and Paimon's default bucket hash column-wise (Paimon's row layout and Murmur hash are Flink's, so the native key encoder already produces both), and the routed batches are shuffled while still Arrow with Paimon's own channel formula. Each batch then enters Paimon's bundle write entry point for its bucket and reaches a StreamFusion FileFormatFactory registered under the selected parquet or orc identifier, whose writer encodes the whole batch with parquet-rs or Paimon's Java vectorized ORC writer over Paimon's output stream. Paimon reads statistics from the resulting footer. Native Parquet writers additionally collect top-level floating-point bounds with Paimon's collector and pass them through its writer-metadata API: parquet-rs excludes NaN from footer bounds, whereas Paimon's manifest bounds include it. No Java rows are materialized for this column-wise collection. Fixed-length BINARY(n) uses Paimon's BYTE_ARRAY encoding, including when nested; Arrow's fixed-size buffers are converted at the writer boundary.

Supported:

• Bucket-unaware tables (bucket = -1), partitioned or not, with partition.sink-strategy none or hash, including Paimon's in-job compaction coordinator and workers.
• Fixed-bucket append tables (bucket > 0 with bucket-key), partitioned or not, with the sink.parallelism and small-bucket-count parallelism rules of the stock sink.
• file.format = orc with NONE/ZLIB/SNAPPY/LZ4/ZSTD compression, Paimon block/level mappings, and the verified ORC writer settings.
• file.format = parquet with file.compression none/snappy/gzip/zstd (and file.compression.zstd-level), file.block-size, and the parquet.* writer keys the stock writer honours: page and dictionary-page size, dictionary encoding, writer version.
• Column types BOOLEAN, TINYINT..BIGINT, FLOAT, DOUBLE, DECIMAL, CHAR/VARCHAR, BINARY/VARBINARY, DATE, TIMESTAMP/TIMESTAMP_LTZ up to precision 9, and ARRAY, MAP, ROW of those, recursively, with Paimon's field ids on every column.
• Hint options (/*+ OPTIONS(...) */) and the paimon.<catalog>.<db>.<table>.<option> dynamic options from the job configuration, resolved the way Paimon's own factory resolves them. Statement hints are read explicitly from the physical sink, including on Flink 1.18 where the resolved catalog table retains its original options. They affect both admission and the writer destination: a branch hint must leave the main branch unchanged. Table-scoped job configuration overrides the hinted options, matching Paimon's factory precedence.
• sink.writer-refresh-detectors: the writer re-reads the refreshed option groups (external data paths) after each checkpoint's commit preparation, exactly when the stock operator does.
• Any insert-only query shape: columns bind to the table by position as in Flink's own sink, so aliased projections (SELECT a AS x ...) and casts the planner inserts are written under the table's names and nullability, and an insert-only stream coming out of a changelog-capable operator (a join, an aggregate) is accepted with its hidden row-kind column dropped.

Append buffering and local spill

Each active table-partition/bucket pair needs a file writer. To limit encoder memory when a task touches many destinations, Paimon switches append writers to buffering when opening a writer would exceed write-max-writers-to-spill (default 10). StreamFusion uses the same threshold, finishes the existing native files, and retains subsequent Arrow batches in Rust. The mode stays enabled for the rest of the writer's life, including after checkpoints and writer-option refreshes.

When retained Arrow memory exceeds write-buffer-size, the largest bucket spills to an Arrow IPC stream in Flink's task-local spilling directories. Spills support spill-compression = lz4, lzo, and zstd (the default), including spill-compression.zstd-level (default 1). LZO uses Paimon's Java compressor on 64 KiB IPC byte blocks and a native decoder, with reusable codec buffers; the data stays columnar and a whole spill need not fit in memory. These settings control temporary spill files; file.compression controls final data files. At a checkpoint the writer drains each bucket's spills and then its in-memory batches in arrival order, through the native Parquet encoder. Only one bucket's encoder is open during this drain. Paimon still owns file metadata, compaction, checkpoint state, and commits. Local spills are temporary: drain and cancellation delete them; recovery uses checkpointed Paimon files and source replay.

write-buffer-spill.max-disk-size is a soft per-bucket limit, checked before another spill as in Paimon. A full disk allowance flushes the bucket to final data files. Memory pressure is checked after accepting a batch, so the budget can temporarily be exceeded by one incoming batch and encoding/IPC working memory. Automatic writer-count spilling enables disk spill even when write-buffer-spillable = false, as Paimon's automatic transition does.

sink.use-managed-memory-allocator defaults to false, giving the sink an independent buffer budget controlled by write-buffer-size. Enabling it obtains the buffer budget from Flink's managed memory pool. Retained native Arrow buffers reserve their byte counts through Flink's MemoryManager, use the operator's assigned managed-memory share, and spill the largest bucket if the share or the available reservation is exhausted. Checkpoint drains and close return the reservations. This accounts for retained buffers; incoming batches, sort/encode scratch space, and file-encoder allocations remain transient off-heap allocations. Paimon's Java compactor retains its own managed-memory allocator. Both append and primary-key sinks support this setting.

Paimon rereads and rewrites unfinished files at the transition; StreamFusion retains those valid files. This avoids a row conversion and repeated encoding, but file boundaries, per-file statistics, and absolute sequence numbers can differ from a stock run after the transition. Rows, destination buckets, schema, codecs, and sequence-number progression within each bucket writer are preserved. The focused NativeAppendSinkWriteTest checks multiple checkpoints, writer refresh/reopen, disk limits, native footers, and spill cleanup; PaimonSinkParityTest covers SQL, coordinator commits, and checkpoint failure/recovery. Its many-writer transition case compares complete row results and destination buckets, verifies native encoding and continuous sequence ranges, and permits different file boundaries. Cases without that transition also compare per-file statistics and footer layout against the stock twin. The release spill diagnostic measured 1.12× throughput against the previous native path that reverted to Java spilling and encoding.

Primary-key tables

A fixed- or dynamic-bucket primary-key table takes the changelog Flink infers for it (+I/+U/-D; Paimon's sink declares that it needs no UPDATE_BEFORE). The routed batches keep their row kinds and are held per bucket in a native buffer. At a checkpoint, or once a task's buffers exceed write-buffer-size (largest bucket first, as Paimon's memory pool spills), a bucket's rows are sorted by key, optional user sequence, and arrival, reduced by the table's merge engine, and written straight into level-0 data files in Paimon's key-value layout with the native Parquet encoder, rolled at target-file-size. Every file carries the metadata Paimon's own writer records: key bounds, key and value statistics from the format, sequence range, delete count, level 0. Sequence numbers continue from the bucket's committed files exactly as a restored Paimon writer's do, so a native run numbers its rows like a stock run. The native writer also checkpoints each bucket's next sequence number in Paimon's partitioned writer state. Coordinated recovery uses the larger of this saved position and the restored files' maximum, so a coordinated manifest cache from an earlier snapshot cannot make replayed updates older than files already written by the recovered task. Idle checkpoints retain the saved positions; Paimon's state filter redistributes them with the buckets when sink parallelism changes.

Compaction stays Paimon's, in the same job: the new files are handed to Paimon's merge-tree writer for the bucket before it prepares each checkpoint's commit, through the entry Paimon's dedicated compaction operator uses for files written elsewhere, so the writer compacts them with the table's own strategy (num-sorted-run.compaction-trigger, compaction.*, commit.force-compact) and the rewrites run through Paimon's stock Parquet writer as they do today. Paimon's writer sees no rows, only files; when it is idle across checkpoints Paimon closes it and the next hand-off recreates it with a scan of the bucket's committed files, the same cost Paimon's dedicated compactor pays. With write-only = true the hand-off is inert and a dedicated compaction job (CALL sys.compact or a compaction action) picks the level-0 files up unchanged.

Supported: bucket >= 1 with the default or an explicit bucket-key, or dynamic bucket = -1, merge-engine = deduplicate/first-row/partial-update/aggregation, changelog-producer none/input/lookup/full-compaction, ignore-delete, ignore-update-before, write-only, file.compression* and the selected format's writer keys as for append tables, and key columns of type BOOLEAN, TINYINT..BIGINT, DECIMAL, CHAR/VARCHAR, BINARY/VARBINARY, DATE, TIMESTAMP, and TIMESTAMP_LTZ (the native sort orders keys by their Arrow byte encoding, which agrees with Paimon's key comparator for exactly these types). An insert-only stream into a primary-key table is taken as all inserts.

Primary-key writer options

data-file.thin-mode = true writes only sequence number, row kind, and table values, omitting duplicate _KEY_* columns from both data and input-changelog files. The encoder borrows the existing value vectors; key bounds still come from the sorted keys, and key statistics map to the corresponding value fields with Paimon's key-statistics policy. Java compaction reads and rewrites these files normally. Native source admission for thin files remains a separate source limitation.

data-file.external-paths and data-file.external-paths.strategy use Paimon's path factory and persist the full external path in each file's metadata. Writer option refresh takes effect after checkpoints. write.sequence-number-init-mode = snapshot starts from the larger of the restored bucket maximum and the snapshot's generated-sequence property. Write-only mode skips the bucket scan when that property exists, and scans older snapshots that lack it, matching Paimon.

sink.key-only-deletes.enabled retains Paimon's changelog negotiation: supported upsert inputs may supply keys with null value fields on deletes. Paimon ignores this negotiation flag for input changelogs, aggregation, and partial updates with aggregates. precommit-compact retains the stock changelog compaction coordinator, workers, and creation-time sort before commit. Its buffer and thread settings are parsed by those Java operators.

local-merge-buffer-size combines updates by the full table primary key, including partition columns, before the bucket shuffle. With changelog-producer=none, deduplicate and first-row retain Arrow batches in the existing native sink merger, compare keys and user/arrival sequences, and gather surviving values directly into Arrow columns. The outgoing vectors transfer ownership without a Java row conversion. The configured size bounds retained input at batch boundaries; the batch that reaches the limit triggers a flush, so buffering can exceed the target by one input batch. Arrow storage can produce different intermediate flush groups from Java's row buffer.

Field merges (partial-update, aggregation) and every changelog-producing mode retain Paimon's Java local merger and its exact buffer grouping. Those inputs are exposed as Arrow row views, copied into Paimon's row buffer, and merged results are written into new Arrow batches. This pays a row conversion cost to preserve aggregation/retraction results and intermediate input changelogs. Both paths keep row kinds, hold watermarks behind pending rows, flush before checkpoint barriers and end of input, and release uncommitted buffers on cancellation. Batch clustering remains on stock Paimon. PaimonLocalMergeTest, PaimonLocalMergeBoundaryTest, and PaimonLocalMergeNativeTest cover SQL writes and reopened jobs, released Java buffer boundaries, and Arrow ownership/checkpoint restore.

Merge engines and input ordering

• deduplicate keeps the last row in merge order; first-row keeps the first. Paimon's normal first-row restrictions still apply, including its lookup producer and delete handling.
• For deduplicate, sequence.field compares one or more value columns before the arrival number, including ascending or descending sequence.field.sort-order. Nulls sort first in both directions; equal user sequences are resolved by arrival. The stored _SEQUENCE_NUMBER continues to count accepted arrivals. Sequence columns admit the primary-key types above plus FLOAT and DOUBLE. Floating comparisons match Java's ordering of signed zero, infinities, and canonicalized NaNs.
• rowkind.field reads +I, -U, +U, or -D from a string column. Filtering of ignored deletes and update-before rows happens after this override and before assigning sequence numbers. Null or invalid kind strings fail as in Paimon.
• partial-update merges non-null values. Sequence groups (fields.<sequence columns>.sequence-group) can replace protected values with null, retract only their columns, and use the supported field aggregates. Both partial-update.remove-record-on-delete and partial-update.remove-record-on-sequence-group follow the released writer's behavior.
• aggregation reduces each field with its configured aggregate. Supported functions are sum and product on integer, floating-point and decimal columns; min/max on the comparable types above, including floats; bool_and/bool_or; string listagg with fields.<column>.list-agg-delimiter and fields.<column>.distinct; and first_value, last_value, first_non_null_value (also first_not_null_value), and last_non_null_value on all supported value types, including nested columns. Paimon's default aggregate selection, fields.<column>.ignore-retract, and aggregation.remove-record-on-delete are preserved.
• Specialized collect, merge_map, merge_map_with_keytime, nested_update, nested_partial_update, hll_sketch, theta_sketch, rbm32, and rbm64 use the released Java field aggregators. Rust retains key grouping and ordinary field reductions, and passes each specialized column's operations to Java in one C Data call per flush. This preserves Paimon's collection ordering, nested-field options, and serialized sketch formats. It adds no dependency on paimon-rust and does not claim these Java field computations are native. Callback failures retain the original Java exception across Arrow cleanup, following Comet's retained-throwable pattern. Unsupported retractions and malformed sketches fail as in Paimon, even when a later delete would discard the intermediate result.
• Column defaults are parsed by released Java Paimon and fill null values in Arrow before merging. Defaults on primary-key, partition or bucket-key columns fall back because they affect routing.

As in Paimon's level-0 writer, a key with just one buffered row passes through without invoking the merge function. This preserves its original row kind and values. Multiple-row partial and aggregate reductions produce the same insert/delete kind as Paimon. Compaction and subsequent reads apply Paimon's merge functions to these files.

PaimonMergeEngineTest compares native and stock writers over checkpoints and restarts, including file sequence ranges, delete counts, nested values, defaults, input changelogs and compaction. PaimonSinkParityTest also checks streaming SQL admission and results for each feature. PaimonMergeAdditionalTypesTest covers floating ordering and distinct concatenation; PaimonSpecializedAggregateTest compares collections, nested values, maps, and serialized sketches across checkpoints, compaction, and reopened writers on both file formats. PaimonFieldAggregatorTest checks released exceptions and Arrow cleanup on failure. A focused run of Paimon's unchanged SQL tests also covers collection retractions with lookup and full-compaction changelogs, map retractions, and the stock streaming-read restrictions for product and listagg tables without a changelog producer.

Dynamic buckets

For primary-key tables with bucket = -1, the first native shuffle computes Paimon's assigner channel from the partition and trimmed primary-key hashes. Paimon's released HashBucketAssigner assigns bucket IDs from those hashes, and a second native gather and Arrow shuffle sends each partition/bucket batch to its writer. dynamic-bucket.target-row-num, dynamic-bucket.assigner-parallelism, dynamic-bucket.initial-buckets, and dynamic-bucket.max-buckets retain Paimon's behavior.

Each native flush also supplies its distinct keys to Paimon's DynamicBucketIndexMaintainer. The hash-index files are committed with the data files, so updates after restart or rescaling return to the persisted buckets. Assigner commit-user state and checkpoint cleanup follow Paimon's Flink operator. The changelog producers, deletion vectors, and writer options supported for fixed buckets also apply here. Cross-partition primary keys (KEY_DYNAMIC) still fall back.

Postpone buckets

Streaming bucket = -2 tables write every accepted change in arrival order, without sorting or deduplication, with Paimon's unknown sequence number (-1). The shuffle follows Paimon's partition/primary-key channel formula, or its partition-only routing with partition.sink-strategy = hash. Files retain Paimon's commit-user and writer-ID prefix so its compactor can replay each writer's input in order. ignore-delete still filters retracts. Creation times increase by at least one millisecond per partition/writer, even when files roll within one clock tick. Recovery continues after that writer's committed staging files, so neither manifest scan order nor a stalled clock can reorder its changes.

Native postpone files use the configured Parquet or ORC format. Stock Paimon 2.0.0 normally chooses Avro internally for these staging files; Paimon's released readers and compactor accept the configured native formats too. Native staging therefore differs in physical encoding while preserving the records, row kinds, replay order, and commit protocol. It uses the selected native format's option whitelist.

Postpone ingestion does not produce the final merged table or changelog. Paimon's dedicated compaction job (CALL sys.compact in batch mode) assigns real buckets and applies the configured merge, changelog, and deletion-vector behavior. Default reads omit uncompacted postpone files; Paimon's batch postpone.merge-on-read reader can include them. The native writer preserves the stock stateless writer and restore-only committer lifecycle. Batch writes and primary keys that omit partition columns remain outside the native whitelist.

Changelog producers and deletion vectors

For fixed and dynamic buckets:

• Input: the same native sort returns every retained input row before deduplication, ordered by primary key, user sequence when configured, and then arrival sequence. These rows, including UPDATE_BEFORE and deletes, are encoded into separate Parquet changelog files and committed through Paimon's changelog manifests. They are not data-file extras, and their rolling boundaries are independent of the merged data files. ignore-delete removes retracts before both outputs and before numbering. changelog-file.format matching the table format (parquet or orc), changelog-file.compression, and changelog-file.stats-mode follow Paimon's writer settings; compression has the same native whitelist as data files.
• Lookup and force-lookup: native level-0 files enter Paimon's selected lookup writer. Paimon manages previous values and compaction-produced changelogs, and applies lookup-wait and its lookup compaction strategy. Its active-bucket checkpoint state is preserved, so recovery resumes unfinished lookup even if no more input arrives.
• Full compaction: Paimon's own global full-compaction writer tracks modified buckets and schedules full compaction using full-compaction.delta-commits or changelog-producer.compaction-interval, including idle checkpoints and restored buckets. These scheduling options also work with changelog-producer = none or input. Ordinary table compaction can still run between scheduled full compactions, as in stock Paimon. Native file hand-off also marks the bucket's data checkpoint before Paimon checks idle cleanup. Otherwise its own increment can look empty before the native files are attached, allowing the writer to close and scheduled compaction to reopen an older snapshot. The released Paimon 1.0/2.0 writer-container hook updates only that bucket's modification identifier; unrelated committed idle buckets still close. Its field layout and accessibility are checked at planning time, with stock-sink fallback if the hook is unavailable.
• Deletion vectors: for deduplicate tables with none, input, or lookup, Paimon's lookup compactor produces the deletion vectors and its index metadata is retained in the checkpoint commit. Read visibility follows Paimon's options: uncompacted level-0 files are hidden by default in deletion-vector tables; deletion-vectors.merge-on-read includes them.

write-only retains Paimon's normal behavior: input changelog is written immediately, while lookup, full compaction, and deletion-vector maintenance are left to a separate compaction job. The native sink uses the released Java connector for these lifecycles; see the writer boundary and paimon-rust assessment.

Parity

Twin-table tests compare logical rows, key/value statistics, sequence and row-kind metadata, and footer schemas against the stock writer in PaimonSinkParityTest, NativePaimonParquetWriterTest, NativePaimonOrcTest, NativePaimonKeyValueFileWriterTest, NativeKeyValueSinkWriteTest, and PaimonChangelogSinkWriteTest. Compressed sizes can differ; size-based rolling and compaction still follow Paimon's decisions on the actual files. Postpone staging, append-buffer transitions and random PARTITION_DYNAMIC routing have the additional physical-file differences described on this page. bin/flink-suite.sh paimon runs Paimon's own unchanged append-table SQL integration tests with the native sink installed (see the upstream suite). Parquet's physical floating-point footer bounds can differ for NaN and signed zero; Paimon manifest statistics use the collector described above. PaimonValueTypesTest checks 40 scalar edge-value cases across both formats, including native snapshot reads, NaN/infinities, signed zero, integer extremes, precision-38 decimals, Unicode, binary and dates.

PaimonTimestampPrecisionTest compares mixed precision 0/3/6/7/8/9, local and instant timestamps, nested rows/arrays/maps, NULLs, pre-epoch fractions and years 0001/9999 against released Paimon. It checks committed native file markers, schemas/field IDs, manifest statistics, repeated keys, nanosecond key/sequence differences, all four merge engines, writer restart and stock compaction of native files. Both ORC LTZ representations are covered. PaimonTimestampSqlTest runs partitioned append and primary-key SQL inserts on both supported Flink/Paimon lines, verifies actual native-written files and reads them with the released Paimon reader. These sink tests do not broaden native high-precision source coverage.

The released Paimon 1.0 Parquet reader rejects negative sub-millisecond remainders in pre-epoch INT96 values, including files its own writer produces. The native sink preserves those exact bytes and the released read failure; it does not normalize the file to repair that upstream behavior. LegacyPaimonInt96Test checks precision 7/8/9 and LTZ bytes against the stock writer and asserts the same reader exception. Exact successful pre-epoch fractional readback is verified on Paimon 2.0 and on ORC; the shared 1.0/2.0 Parquet SQL success test uses whole-millisecond fractions before 1970 and nanosecond fractions from 1970 onward.

High-precision sink measurements

PaimonTimestampBenchmark compares the released Paimon primary-key sink with the new sink path. Before this support, these types fell back to the released sink, so stock is also the previous StreamFusion baseline. Timing includes row routing, RowData-to-Arrow conversion, JNI, merge, Parquet/ORC encoding and snapshot commit. Table/writer setup, prepared logical input and readback are outside timing; this is a sink-boundary measurement, not a complete SQL job benchmark. Every native sample verifies committed file markers and exact released-reader parity.

Measured on 2026-10-25 on an Apple M4 Pro (24 GiB RAM), release/mimalloc, Flink 2.2.1, Paimon 2.0.0, JDK 17 (-Xmx2g), one writer, two buckets, 16,384 integer keys, three nullable timestamps (TIMESTAMP(7), TIMESTAMP(9), TIMESTAMP_LTZ(8)), batch size 4,096, two warmup pairs and five alternating measurement pairs per configuration:

Format	Input Rows	Stock Median [Min, Max], s	Native Median [Min, Max], s	Speedup
Parquet	32,768	0.034769 [0.031525, 0.046823]	0.020919 [0.018542, 0.025237]	1.66x
ORC	32,768	0.031016 [0.029984, 0.032741]	0.019716 [0.018809, 0.022578]	1.57x
Parquet	131,072	0.063403 [0.061267, 0.078037]	0.039551 [0.039228, 0.041066]	1.60x
ORC	131,072	0.061619 [0.060829, 0.063555]	0.038189 [0.037654, 0.039106]	1.61x

Reproduce with SF_PAIMON_TIMESTAMP_BENCHMARK=true mvn -Ppaimon,bench -pl streamfusion-paimon -am test -Dtest=PaimonTimestampBenchmark -Dsurefire.failIfNoSpecifiedTests=false -Dsf.extraJvmArgs=-Xmx2g. timestamp.warmup and timestamp.samples control repetitions.

The regular CI Paimon job runs SQL parity for dynamic and postpone buckets, including reopened jobs, assigner/writer rescaling, and Paimon's SQL compaction of native postpone files. A recovery harness discards an uncommitted dynamic checkpoint, restores the assigner state, replays the input, and compares the resulting hash-to-bucket index and changelog with stock Paimon. A larger SQL fixture verifies postpone replay across rolled files. These are ordinary correctness tests; the performance diagnostics below are opt-in. That rolled-file fixture uses stock fixed-bucket ingestion as its arrival-order oracle: released Java postpone writers can give two files the same millisecond creation time and replay them out of order. Native postpone files use strictly increasing creation times as described above.

Native file hand-off and checkpoint visibility

The post-merge Paimon job on d0c47748 timed out waiting for a full-compaction changelog. Stock replay of its retained warehouse reads all 256 final values in batch, but only 236 terminal keys from the stored changelog. Stock replay of the second PR #308 attempt also reads all 256 final values in batch while its changelog retains older values.

A deterministic stock/native regression delays publication of checkpoint 4 until after scheduled full-compaction checkpoint 6. Stock retains the pending update, while the original native implementation emits only the original inserts. The native writer had been closed during commit preparation: Paimon's own commit increment appeared empty before the native data files were added. Reopening from the earlier committed snapshot lost their compaction input. Marking the native checkpoint in its bucket's writer container preserves that compaction input until the data commit becomes visible. Regression coverage exercises default and explicitly installed Paimon 2.0 restore providers, delayed commits under Paimon 1.0, and cleanup of an unrelated committed bucket while another bucket still has pending data. No task-wide retention barrier is used. This race is distinct from the released split-commit recovery defect below; neither timeout extension nor retry changes either contract.

Released Paimon 2.0 split-commit recovery diagnostic

Released Paimon 2.0.0 can lose the final full-compaction changelog when filesystem access fails between its separate APPEND and COMPACT snapshot commits. A deterministic stock-only test writes 10, then 20, and injects one manifest-read failure after the second APPEND. Recovery through filterAndCommit retains the newest data but exposes only +I(1,10) in the stream; the no-failure control also exposes -U(1,10) and +U(1,20). This test uses released Java Paimon directly, without the StreamFusion planner or native writer.

Run the diagnostic with JDK 17. The recovery case is expected to fail on 2.0.0; the control passes. The diagnostic is opt-in because it asserts the missing upstream contract:

mvn -pl streamfusion-paimon -am -Ppaimon test \
  -Dtest=PaimonFullCompactionCommitRecoveryTest \
  -Dsurefire.failIfNoSpecifiedTests=false -Dsf.paimon.reproduceSplitCommit=true

The upstream randomized streaming full-compaction test timed out in both attempts of CI run 36985518414. In the second attempt the producer recovered from injected I/O failures and finished about 50 seconds before the timeout, but committed no snapshots after the second recovery. That evidence is consistent with this split-commit failure; it does not prove an identical cause or rule out a native-path defect. The randomized test remains enabled with its original assertions and deadline. A separate stock/native test passes when restarting immediately before or after scheduled full compaction and continuing only idle checkpoints through checkpoint 12; it does not inject the split-commit failure. No connector admission or production behavior changes as a result of these diagnostics.

Dynamic partition routing and clustering options

Partitioned bucket-unaware append tables support partition.sink-strategy = PARTITION_DYNAMIC. Paimon's Java statistics operator reports partition row counts at checkpoints; its existing coordinator aggregates and broadcasts them, and its weighted channel selector distributes each partition across writers. Before statistics arrive, a partition uses up to four writers. Recovery starts collecting fresh statistics, just as in stock Paimon. Only the partition keys enter this Java logic; payloads stay in Arrow batches, split natively by the selected channel. The writer still receives Paimon's unaware bucket 0, independently of its shuffle channel.

The choices are random in stock Paimon, so individual file contents, sizes, and statistics may differ between equivalent executions. Table rows, partition keys, bucket identity, footer schema, and codecs retain parity. Tests cover statistics replacement, null partitions, row order within each channel, different writer parallelisms, watermarks, and recovery after a completed checkpoint. Other bucket modes and unpartitioned tables retain their normal routing, as in stock Paimon.

sink.clustering.* (including the clustering.columns and clustering.strategy names) does not force fallback. Paimon 2.0.0 skips range clustering in STREAMING execution mode, even for a bounded source, so these options do not insert a shuffle or sort. Typed option values are still parsed as in Paimon, so malformed booleans and sample factors remain errors. Incremental-clustering table settings retain Paimon's Java write and compaction behavior. StreamFusion accelerates only the streaming planner; batch range clustering and separate clustering jobs remain stock Paimon.

Writer and commit coordinators

Both coordinator options retain Paimon 2.0.0's Java implementations while the data writer consumes Arrow batches. No coordinator logic is implemented in Rust.

sink.writer-coordinator.enabled = true selects Paimon's coordinated restore factory for fixed and dynamic buckets. The JobManager serves restored files through Paimon's paged requests and metadata cache. The native primary-key writer uses this same restore service to continue sequence numbers; Paimon's selected writer retains compaction, changelog, and deletion-vector restoration. As in stock Paimon, unaware append and postpone writers do not use the writer restore coordinator.

sink.coordinator-commit.enabled = true selects Paimon's committing writer coordinator for bucket-unaware append tables. Its existing writer lifecycle owns pending committables, checkpoint events, watermark and idle status, replay, and the deliberate restart after recovering an uncommitted snapshot. The native subclass replaces only record ingestion with bundle ingestion. Paimon's sink builder removes the downstream global committer and retains its configuration checks: streaming checkpointing must be enabled, write-only = true, precommit-compact = false, sink.savepoint.auto-tag = false, and at most one checkpoint may run concurrently. This does not add bounded end-input commit support beyond the released connector. The option is ignored by Paimon's fixed-bucket and primary-key sinks, and StreamFusion preserves that behavior.

SQL parity tests reopen fixed and dynamic primary-key tables with paged coordinated restoration, including input/lookup/full-compaction changelogs and deletion vectors, and compare append-file metadata and sequence numbers across jobs. SQL failover tests replay append and fixed/dynamic primary-key writes after a completed checkpoint. The commit recovery harness compares stock and native rows and snapshot histories after a lost checkpoint and replay, including an already-committed snapshot whose acknowledgement was lost. The upstream SQL suite includes Paimon's unchanged CoordinatorCommitITCase for topology, commit metrics, committed rows, and idle-watermark parity.

Sink falls back to stock Paimon on

Each of these declines at planning time with a reason visible in NativePlanner.explain:

• Cross-partition dynamic keys (KEY_DYNAMIC) and postpone primary keys that omit partition columns.
• A changelog (retracting or updating) input into an append table, a sink Flink plans with a SinkUpsertMaterializer (table.exec.sink.upsert-materialize; Paimon itself refuses that operator), INSERT OVERWRITE, and batch-mode inserts (the substitution only exists in the streaming planner).
• Primary-key tables with field aggregate/type combinations outside the merge whitelist above; sequence fields or sequence groups outside the comparable-type whitelist; sequence.field combined with partial-update or aggregation; defaults on routing columns; input changelog with changelog-file.format differing from the table format or unsupported changelog compression; primary-key vector, full-text, BTree, or bitmap indexes; or a FLOAT/DOUBLE key column.
• file.format other than installed parquet or orc, file.format.per.level, write-buffer-for-append = true, file indexes (file-index.*), row-tracking.enabled, data-evolution.enabled, BLOB columns.
• Append tables with spill-compression other than lz4/lzo/zstd. Released Paimon 2.0.0 fails when actually spilling with none; it remains on the stock path.
• A nullable query field assigned to a NOT NULL target with table.exec.sink.not-null-enforcer=DROP, or a bounded CHAR/VARCHAR or BINARY/VARBINARY target while table.exec.sink.type-length-enforcer is enabled. The stock sink path preserves Flink's configured fail/drop and trim/pad/error behavior. In both the 2.0 and released 1.0 adapters, the default ERROR mode stays columnar: the shared columnar constraint enforcer runs before local merging and bucket routing, preserves Flink's first offending row/column, error message and valid-prefix behavior, and rejects null values without changing primary-key or changelog admission. Non-null batches pass unchanged without materializing rows, taking a consumer share, or copying their buffers. Other constraints and connector recovery/commit behavior are unchanged.
• TIME, VARIANT, vector, and geospatial types.
• parquet.* keys the native writer cannot honour: bloom filters, page validation, custom padding, page row-count limits, statistics/column-index truncation, page-size row checks, multithreaded zstd, and any per-column (parquet.*#column) or unrecognised writer key.
• The selected format identifier resolving to Paimon's own factory (see deployment below).

Compaction rewrites (in-job or from a dedicated compaction job) still use Paimon's stock Parquet writer. Append buffer spilling retains native encoding as described above. Primary-key buckets use their existing native buffers, which flush by size into level-0 files.

Benchmark

The fresh Kafka→Paimon streaming RocksDB/off suite uses 2M unchanged events, four Kafka partitions, four output buckets, Flink 2.2.1, 512 MiB managed memory across four slots, and one warmup plus three alternating measured pairs. All 23 queries pass committed-table validation. Repeated-median geomeans are 1.43× overall, 1.24× append and 1.97× primary-key; q5/q20/q23 remain slower. The corresponding Iceberg run is 1.44× overall, with different append/update results and wide ranges retained. bin/bench-paimon.sh reproduces the release/mimalloc run. The report includes every trial, full-value fingerprints, plans and Rust Parquet file witnesses. The older memory-state figures below use a different sampling method and host.

The measurements below use Parquet files. ORC diagnostics report that format's reader and writer results separately.

On the 2M-event, four-partition Kafka JSON Nexmark sink diagnostic (memory state, mini-batching off, one warmup, best of three), the 16 append-only queries completed on both engines and StreamFusion's suite geomean was 1.47× the stock published-Paimon path for bucket-unaware tables with in-job compaction and 1.47× for four fixed buckets, from 1.08× on a join that emits a few hundred rows to 2.04× on the query that writes 5.5 M joined rows. Row counts read back through Paimon's snapshots agree on every query except the processing-time window q12, whose output is non-deterministic by construction.

The seven updating queries (q4, q9, q15–q19) ran against deduplicate primary-key tables with four fixed buckets and Paimon's default in-job compaction on the same diagnostic: StreamFusion's suite geomean was 1.64× the stock path, from 1.18× on q9 (updates concentrated on 120 K keys) to 2.38× on q16 (eight keys rewritten a million times), and the merged row counts read back through Paimon agree on every query. Stock Paimon pays for its row-at-a-time sort buffer; the native sink merges each bucket's routed Arrow batches by key in Rust and writes the level-0 files directly. See Benchmarks for the method and reproduction command.

Changelog writer diagnostic

The opt-in writer diagnostic exercises input, lookup, full-compaction, and deletion-vector writes against stock Paimon, with one warmup and best of three measured runs. It starts with the same row fixture, includes routing and the native path's RowData-to-Arrow conversion, and measures writing plus commit preparation and commit. Each run checks the merged table contents. This is a writer diagnostic, not an end-to-end Nexmark result; the Nexmark harness is unchanged. It measures one checkpoint into an empty table. It does not measure ongoing lookup against older files or sparse-update deletion-vector maintenance; those are covered by the recovery parity tests.

SF_PAIMON_CHANGELOG_BENCHMARK=true mvn test -Ppaimon,bench \
  -pl :streamfusion-paimon -am -Dtest=PaimonChangelogSinkBenchmark \
  -Dsurefire.failIfNoSpecifiedTests=false

The default input is 200,000 changes over 20,000 keys, four buckets, and batches of 4,096 rows. SF_PAIMON_CHANGELOG_ROWS changes the input size while retaining the ten-to-one change/key ratio.

Local release measurements with Paimon 2.0.0: the 1M-change run is from 2026-09-11; the 200K-change run was repeated on 2026-09-13 with the shared file writer and floating statistics.

Mode	Stock, 1M changes	Native, 1M changes	Speedup, 1M	Speedup, 200K
Input changelog	3.716 s	2.319 s	1.60×	1.92×
Lookup changelog	2.387 s	1.856 s	1.29×	1.28×
Full-compaction changelog	1.276 s	1.343 s	0.95×	0.93×
Deletion vectors enabled	1.498 s	1.122 s	1.34×	1.34×

Full compaction is slower in this row-fed diagnostic. It retains Paimon's rowwise compaction and adds the native path's conversion and file hand-off costs. Its admission adds coverage for columnar pipelines; these measurements do not establish a throughput improvement for that mode.

Merge-engine writer diagnostic

PaimonMergeBenchmark measures row routing, the RowData-to-Arrow conversion, merging, level-0 file encoding and commit against the released stock writer. It uses 131,072 rows, 16,384 keys, two buckets and nine columns including nulls and nested arrays. write-only isolates ingestion from compaction. Writer setup, close and result verification are outside the timer. After one warmup per mode, three measured runs alternate engine order and report the best of each.

On the same local machine with release native libraries and Parquet:

Mode	Stock	Native	Throughput ratio
First row	0.116 s	0.093 s	1.24×
Partial update with a sequence group	0.145 s	0.120 s	1.21×
Aggregation with sum	0.127 s	0.106 s	1.20×
Deduplicate with two sequence fields	0.108 s	0.085 s	1.27×

With SF_PAIMON_MERGE_THIN=true on the same 131,072-row fixture, stock/native seconds were 0.109/0.092 for first row, 0.140/0.118 for partial update, 0.124/0.106 for aggregation, and 0.104/0.087 for sequence-based deduplication: 1.17–1.20× stock throughput with matching rows.

With SF_PAIMON_MERGE_MODES=collect, distinct array collection uses the released Java field kernel inside the native writer. The final release fixture measured 0.212 s stock / 0.179 s native (1.18×) with Parquet and 0.190 / 0.187 s (1.02×) with ORC, which is roughly tied. Both include the ingress conversion and column callback and verify identical array contents and ordering. These measure the whole writer path, not a faster implementation of Java's collection function or a benchmark of every specialized aggregate. The environment variable accepts a comma-separated subset of the modes above plus collect.

Every run verifies the final table contents against its stock twin. This is a writer diagnostic, not an end-to-end Flink job benchmark. Run it with:

SF_PAIMON_MERGE_BENCHMARK=true mvn test -Pbench,paimon \
  -pl :streamfusion-paimon -am -Dtest=PaimonMergeBenchmark \
  -Dsurefire.failIfNoSpecifiedTests=false

SF_PAIMON_MERGE_ROWS changes the input count; SF_PAIMON_FILE_FORMAT=orc selects ORC.

Dynamic and postpone SQL diagnostic

The opt-in bucket-mode diagnostic measures streaming SQL ingestion from the same row fixture, including table creation, planning, job startup, RowData-to-Arrow conversion, shuffles, and commit. It uses one source task, two sink writers, ten changes per key, one warmup, and the best of three measured runs, alternating engine order. Dynamic buckets target 5,000 rows per bucket. Each run compares the resulting table contents. For postpone tables, SQL compaction makes the rows visible and validates replay outside the ingestion timer; these numbers do not measure the complete ingestion-and-compaction lifecycle. Stock postpone staging uses Avro, while native staging uses Parquet, as described above.

SF_PAIMON_BUCKET_BENCHMARK=true SF_PAIMON_BUCKET_ROWS=1000000 \
  mvn test -Ppaimon,bench -pl :streamfusion-paimon -am \
  -Dtest=PaimonBucketModeBenchmark -Dsurefire.failIfNoSpecifiedTests=false

Local release measurements on 2026-09-11 with Paimon 2.0.0:

Mode	Stock, 1M changes	Native, 1M changes	Speedup, 1M	Speedup, 200K
Dynamic buckets	3.025 s	3.139 s	0.96×	1.05×
Postpone buckets	2.618 s	3.229 s	0.81×	0.89×

These timings precede the final postpone creation-time ordering fix; its metadata counter and restoration scan were validated separately and have not been rebenchmarked.

These row-fed results do not establish a throughput improvement. Dynamic assignment still calls Paimon's Java index per key, and postpone staging pays for conversion and Parquet encoding without the reduction in output rows that native deduplication normally provides. Admission adds coverage for existing columnar pipelines and a boundary for future optimization; faster ingestion from an Arrow source has not been measured. The default fixture size is 200,000 changes. The Nexmark harness is unchanged.

Local merge diagnostic

PaimonLocalMergeBenchmark compares local merge with identical Arrow input and output boundaries, including Java's row-buffer conversion on its path. A release run over 131,072 rows, nine columns (including an array), 16,384 keys and a 4 MiB local buffer measured 0.107 s Java / 0.022 s native for deduplication (4.78×) and 0.050 s / 0.017 s for first-row (2.93×). Input Arrow construction and output verification are outside this operator timer; both paths verify the selected values.

The same benchmark separately runs whole streaming SQL jobs over 131,072 eleven-column changelog rows with two sink writers. This includes SQL planning, job startup, row-to-Arrow ingress, local merge, shuffle, file encoding and commit. Stock/native times were 0.712 / 0.594 s with Parquet (1.20×) and 0.762 / 0.649 s with ORC (1.17×), with matching final tables. Both measurements use one warmup, three measured iterations and alternating engine order, reporting the best time. These are local fixtures, not distributed throughput claims. Run with:

SF_PAIMON_LOCAL_MERGE_BENCHMARK=true mvn -Pbench,paimon \
  -pl :streamfusion-paimon -am -Dtest=PaimonLocalMergeBenchmark \
  -Dsurefire.failIfNoSpecifiedTests=false test

Set SF_PAIMON_FILE_FORMAT=orc for the ORC SQL run, and SF_PAIMON_LOCAL_MERGE_ROWS to change row count. The Arrow-only comparison performs no file IO and does not depend on the file format.

Cross-format validation

SF_PAIMON_FILE_FORMAT=orc runs the shared Paimon table, SQL, spill, merge, bucket and recovery fixtures with ORC defaults. Explicit format-specific tests keep their own format. CI runs the suite for both defaults, with UTC test JVMs; timezone-fallback tests set and restore other zones. The source/key tests also parameterize the formats directly, and native append files are checked for their encoder's footer writer ID. The ORC-default run passes 436 Paimon cases, alongside the Parquet regression suite. Shared filesystem tests pass 52 cases covering both codecs, including failed-footer cleanup. The option tests inspect actual bloom-filter streams, index strides, compression blocks and file versions using Paimon's released reader.

SF_PAIMON_FILE_FORMAT=orc mvn test -Ppaimon -pl :streamfusion-paimon -am \
  '-Dtest=tech.streamfusion.paimon.*Test' -Dsurefire.failIfNoSpecifiedTests=false

The merge diagnostic also accepts SF_PAIMON_MERGE_THIN=true to measure thin data files against stock Paimon with the same setting, including ingress conversion, routing, merge, encoding, and commit. Use the release bench profile for timing.

The source and changelog diagnostics accept the same environment variable with -Pbench,paimon.

Deployment

Install the published paimon-flink-2.2-2.0.0.jar, the selected streamfusion-parquet or streamfusion-orc module, and streamfusion-paimon in Flink's lib/. Paimon resolves a file format by taking the first FileFormatFactory on the classpath that claims the identifier, and Flink adds lib/ JARs in sorted name order, so the StreamFusion JAR must sort before paimon-flink-*: name it 01-streamfusion-paimon.jar (the same convention as 00-streamfusion-loader.jar). The planner checks at planning time that the selected format resolves to the StreamFusion factory and declines the sink with an explicit reason otherwise; it never enters the native topology only to discover the stock format at runtime. Removing the StreamFusion Paimon JAR restores the stock connector entirely.

Batch inserts, compaction jobs, and any other row-fed path see the same factory: for rows it delegates each file to Paimon's own writer, so a mixed deployment stays byte-compatible with stock Paimon.

Outlook

The sequence/merge combinations, complex sequence types, and routing-column defaults listed above deliberately retain the stock sink. Their acceleration would require further ordering and routing work beyond the small ports and released-kernel reuse chosen for this implementation (decision).

A Paimon bundle entry for the merge-tree writer would remove the compaction hand-off and the idle-writer rescan. Released Paimon 2.0.0 walks a bundle row by row before the format writer; a Paimon release that passes bundles through takes the same writer's direct path with no change here (#39). The jar-ordering requirement goes away once Paimon's format discovery gains a priority, which #38 proposes upstream. Remaining streaming source coverage is issue #27.

Build with the paimon Maven profile. The module has no snapshot, local-Maven, path, or forked Paimon dependency.

Flink 1.18 nested fallback batches

The experimental 1.18 Java reader copies each fallback row into owned binary storage before collecting an Arrow batch. Flink 1.18's serializer can otherwise reuse a nested custom-array buffer across rows, corrupting values when restoring a split between Java and native readers. Snapshot parity checks cover nested arrays, projections, and restore offsets on both lines. Compaction parity fixtures pass all eight positional procedure arguments on both lines, using empty strings for optional filters and the explicit full batch-compaction strategy. This avoids depending on named/optional argument annotations and null argument conversion that Flink 1.18 cannot interpret. Batch mode is explicit in the table configuration because 1.18 constructs the procedure’s execution environment from that configuration.

Paimon 1.0.0 on Flink 1.18

The paimon1 profile builds a separate streamfusion-paimon1-flink1.18 adapter against the released Paimon 1.0.0 connector. Build it together with flink-1.18; the build rejects other Flink lines. Install this adapter instead of the Paimon 2.0 adapter, alongside paimon-flink-1.18-1.0.0.jar and the selected StreamFusion Parquet/ORC modules. Keep the same 01-streamfusion-paimon.jar classpath ordering. The two adapters must not be installed together.

The adapter shares the released paimon-rust source reader, Java FileIO bridge, native writer codecs and Arrow batch routing with the 2.0 integration. Its build stages shared Java sources and overlays the classes whose released APIs changed. The artifacts deliberately use the same package names: choose one at deployment rather than mixing incompatible Paimon APIs in one classloader. Java uses the 1.0 append topology, writer memory pools, file metadata, hash indexes, commit messages, and checkpoint lifecycle. Streaming splits containing before-files retain Java's reverse-row-kind reader. Postpone buckets and newer coordinator/partition-routing modes do not exist in this release. External log systems retain the stock source and sink. Numeric aggregation preserves 1.0's wrapping integer arithmetic and empty-accumulator product retractions; the 2.0 adapter keeps its checked arithmetic. ORC timestamp writes reproduce 1.0's historical java.sql.Timestamp calendar conversion for both TIMESTAMP and TIMESTAMP_LTZ, while retaining timestamp-instant physical storage for LTZ. Field defaults retain the stock source because Paimon 1.0 applies them on reads, after merging and schema evolution. Paimon 1.0 has no writer-metadata handoff; native Parquet floating bounds are omitted conservatively so NaNs cannot cause files to be incorrectly pruned.

The profile is opt-in and pins 1.0.0, not the broader 1.x release family. The source and sink admission rules above apply, with the release differences listed here. Batch SQL keeps Paimon's planner and writer. Unsupported combinations keep the stock connector; passing upstream tests does not imply every query is native.

Reproducing compatibility checks

Use JDK 17 and the released Flink 1.18.1 dependencies selected by flink-1.18:

mvn -Pflink-1.18,paimon1 -pl streamfusion-paimon1 -am \
  -Dsf.testForks=1 '-Dtest=LegacyPaimon*Test,PaimonSourceSqlTest,PaimonSplitWatermarksTest' \
  -Dsurefire.failIfNoSpecifiedTests=false \
  '-Dsf.extraJvmArgs=-Xmx2g -XX:ActiveProcessorCount=2 --add-opens=java.base/java.math=ALL-UNNAMED --add-opens=java.base/java.time=ALL-UNNAMED --add-opens=java.base/java.lang.invoke=ALL-UNNAMED' test
bin/paimon1-suite.sh -Pbench

The upstream runner loads the published paimon-flink-common:1.0.0 test JAR, plus two unchanged Flink-1.18-specific tests extracted from the checksum-pinned release-1.0.0 source archive. It uses the existing planner agent to install StreamFusion; upstream SQL, expected rows, assertions, and timeouts stay unchanged. All production Paimon code comes from canonical released Maven artifacts. The release build keeps native execution within upstream's short SQL deadlines. The runner records per-test SQL/planner inventory, Flink diagnostics, and Surefire reports under streamfusion-paimon1/target. SF_PAIMON1_STOCK=true reruns the same tests without the planner agent; SF_PAIMON1_TESTS selects a narrower class/method set using Surefire syntax. The dedicated compatibility workflow also tracks shared Arrow, codec, and upstream-agent changes and retains the test evidence.

CI runs the source/sink parity reactor with -Pflink-1.18,paimon1,bench, building optimized native libraries before testing. The following upstream invocation uses bin/paimon1-suite.sh -Pbench -Dnative.build.skip=true to reuse those libraries from the same checkout and job. This removes the previous debug-then-release build cycle. Do not pass -Dnative.build.skip=true on a fresh checkout: the full release reactor must already have completed successfully. Local commands above retain their default independent builds.

SF_PAIMON1_TEST_FORKS selects a positive integer number of upstream test JVMs and defaults to one locally and in CI, retaining the 2 GiB heap and two visible CPUs. Surefire assigns mvn.forkNumber separately to each JVM; inventory, native evidence, and diagnostic paths include fork-<number> so the processes cannot overwrite each other's files. Tests inside each JVM remain serial, as required by the planner inventory. The selected classes, nested aggregation tests, native-sink witness, and upstream assertions remain unchanged. This is compatibility testing; throughput benchmarks still require one uncontended JVM.

The pre-change CI run took 110.5 minutes, including approximately 31 minutes of debug compilation and 49 minutes of release compilation. A two-fork trial completed in approximately 67 minutes but failed PartialUpdateITCase.testForeignKeyJoin with an upstream wait timeout (241 of 242 tests passed). CI therefore keeps one fork to avoid the additional contention while retaining release-build reuse. The failed trial is not a validated speedup; a passing hosted comparison is still required.

The selected upstream coverage includes append and primary-key tables, dynamic buckets, partial updates, first-row and field aggregation, input/lookup/full compaction changelogs, schema evolution and filter pushdown, composite partitioned keys, and savepoint/job recovery. It is a targeted SQL suite, not all Paimon tests. Dedicated stock/native twins additionally check Parquet and ORC across repeated commits, managed writer memory, read/restore offsets, before-file retractions, field defaults after merging, floating-point edge values, and historical timestamps.

Known Paimon 1.0 recovery limitation

PrimaryKeyFileStoreTableITCase.testNoChangelogProducerStreamingRandom can return stale rows when deletion vectors and injected filesystem failures coincide. This is reproducible using released Paimon alone, without the StreamFusion planner or native operators:

1. The final commit writes an APPEND snapshot containing the newest level-0 files.
2. Filesystem access fails before the separate COMPACT snapshot is committed.
3. Recovery filters out that commit identifier because the APPEND already exists, dropping its pending compaction changes.
4. The default deletion-vector batch scan excludes level-0 files, so the older value remains visible.

A deterministic two-row reproducer writes 10, then 20, and injects one manifest-read failure between the final APPEND and COMPACT. Recovery completes but returns 10; the control without failure returns 20. This also explains a captured upstream failure whose correct final row was present in a live level-0 Parquet file while a fresh stock reader returned an older level-5 row. The upstream randomized test remains enabled and unchanged; longer deadlines do not resolve this durable snapshot state. A passing randomized run does not establish absence of this bug. StreamFusion remains on released Paimon 1.0.0; fixing the split-commit recovery requires a verified upstream release, not a planner workaround or a forked dependency.

An implementation is now submitted for canonical upstream review in apache/paimon#10568. It preserves separate APPEND and COMPACT snapshots and resumes required compaction when only APPEND is durable. This retains incremental delta-query results and the existing file-retention protocol. It also handles competing compaction refresh, checkpointed pending input, and reader cleanup exposed by failure injection. The revised local normal Maven core selection passed 394 tests, with one opt-in performance test skipped, including 45 recovery cases and unchanged incremental scanners. The unchanged global compaction-writer check also passed on both upstream Flink profiles. This is not a released fix or a claim that the complete upstream CI matrix has passed. StreamFusion continues to use its canonical released dependencies. Issue #318 is closed as not planned: this is an upstream recovery defect, and no StreamFusion-specific workaround will be added. A verified canonical release can prompt reassessment; this closure does not mean the defect is fixed.

Released 1.0 code confirms that the normal Flink sink uses the same recovery commit filter: it recognizes a checkpoint identifier from the already published APPEND snapshot and can omit its unfinished COMPACT half. The full-compaction streaming scanner reads COMPACT snapshots, which is why moving changelog files into an APPEND message is not a transparent workaround. Pure released-SDK probes of message rewriting found that APPEND-only commits omit streaming changelogs and competing compactions can leave duplicate active rows. COMPACT-only commits passed the single-writer data/changelog/lost-ack cases but rejected a competing writer's stale file deletion. Recovery and concurrent-writer protocols still need implementation; neither message rewrite is used in StreamFusion. Stock Flink lookup recovery also has safeguards introduced in apache/paimon#3512 and extended in apache/paimon#5796. Sharing the core protocol does not establish that every stock Flink configuration reproduces the failure.

The same incomplete commit also has a deterministic full-compaction and lookup changelog reproducer, without deletion vectors: commit 10, consume the initial stream snapshot, then write 20 and inject a manifest-read failure after APPEND and before COMPACT. After recovery a fresh batch reader returns 20, but the incremental stream scan returns no final changelog. The no-failure control returns the update. This uses released Paimon writer/commit/read APIs without a Flink job or native operators. Standalone runs excluding StreamFusion production classes reproduce the result on both JDK 11 and 17. The explicit recovery assertion fails on 1.0.0; it is an opt-in diagnostic, not a passing recovery test:

mvn -Pflink-1.18,paimon1 -pl streamfusion-paimon1 -am \
  -Dtest=PaimonFullCompactionRecoveryReproducerTest \
  -Dsurefire.failIfNoSpecifiedTests=false \
  -Dsf.paimon1.reproduceFullCompactionCommit=true test

The diagnostic now covers both full-compaction and lookup with deletion vectors disabled and lookup-wait=false. Both no-failure controls pass; both recovered commits fail the required final-changelog assertion (batch=[20], changelog=[], latest snapshot APPEND). Standalone runs excluding StreamFusion production classes reproduce both controls and both failures on JDK 11 and 17. These are failing opt-in diagnostics of the released dependency, not passing StreamFusion recovery coverage.

#318 records the upstream defect and the decision against a StreamFusion-specific workaround. In run 38020828273, 241 of 242 upstream cases passed; the unchanged full-compaction streaming test timed out after injected-failure recovery. The writer finished while its consumer kept waiting. That invocation did not retain the warehouse, so the standalone defect does not prove its exact cause. The Paimon 1.0 workflow now enables the existing failed-warehouse capture and uploads snapshots, manifests and data files with its diagnostics. Upstream assertions, randomized inputs and deadlines remain authoritative; the capture is diagnostic rather than an atomic backup. Compare stock/native replay over that retained state before claiming the timeout resolved.

PR #333's run 38231158108 retains enough evidence to establish the same failure boundary for the unchanged testLookupChangelogProducerStreamingRandom invocation. Of 242 upstream tests, 241 pass and this case times out. Two producers write 512 terminal keys in total; both eventually finish. The second producer's final commit persists snapshot 16, an APPEND at Long.MAX_VALUE, then fails with a file-deletion compaction conflict. After checkpoint recovery the job finishes without a COMPACT snapshot for that identifier. Only the first producer's 256 terminal keys are in committed changelog.

A standalone released-stock replay of the retained table, with every StreamFusion production class and JAR excluded, reads all 512 terminal values in batch but only 256 terminal keys in the changelog. Exact stock/native split replay agrees on all 73 splits and 2,290 rows, including RowKinds and every field; the native replay reads 83 files through its native decoder. Replay begins at snapshot 1 and is finite; it verifies captured durable state and reader parity, without reconstructing live scheduling or claiming that every concurrent native write schedule matches stock. The timeout reflects missing committed changelog, not a slow consumer or the separate fetcher-retirement exception fixed in #334. The captured evidence record links the original artifact and records these boundaries. The upstream recovery defect remains unresolved; #318 is closed as not planned for StreamFusion. The original randomized test, SQL, assertions and deadline remain enabled.

Run the deletion-vector opt-in diagnostic with JDK 17. The recovery case is expected to fail on 1.0.0 (expected: 20 but was: 10); the control passes. It is separate from the regular test gate:

SF_PAIMON1_STOCK=true SF_PAIMON1_TESTS=PaimonSplitCommitRecoveryReproducerTest \
  bin/paimon1-suite.sh -Dsf.paimon1.reproduceSplitCommit=true

Recorded validation

Validation on 2026-09-28 passed all 242 selected upstream cases in one release run, with no failures, errors, or skips, plus 89 dedicated 1.0 cases and 97 core SQL/boundary checks. The final upstream inventory contains native-source job graphs in 37 invocations and native-writer graphs in 28; runtime logs confirm native bundle and level-0 writes. Other cases exercise stock batch execution or supported fallbacks. Dedicated read counters additionally verify native file reads. All 83 selected Paimon 2.0 regressions also pass on the merged code. The Rust arithmetic regression distinguishes 1.0 wrapping from 2.0 overflow checks; release packaging verifies the native payload and provided 1.0.0 dependency.

Release benchmark

LegacyPaimonBenchmark compares stock Paimon, the previous StreamFusion fallback, and the new adapter on identical data. Local measurements on 2026-09-28 include the shared boundary and aggregate changes through main 0b7243c3, and used JDK 17, Flink 1.18.1, Paimon 1.0.0, release Rust with mimalloc, one task, two JVM CPUs, a 1.5 GiB test heap, 4,096-row Arrow batches, and a 32 MiB writer buffer. The bench profile disables local zero-copy exchange, retaining serialization costs. Each configuration has one warmup and three measurements with rotating engine order. Times are median [minimum, maximum], not best-of-run throughput claims.

The row-fed sink query writes (BIGINT id, BIGINT v, STRING label) using INSERT INTO t SELECT id, v + 1, label FROM input_rows. Labels have 100 distinct values. Primary-key tables have one bucket and 10% distinct keys; append tables use unaware buckets. Both use write-only=true. Timers include SQL execution, startup, input generation, conversions, JNI, routing, merge, encoding and commit; plan inspection and final stock-reader verification are outside the timer. Compaction and distributed checkpoint throughput are not measured. The previous fallback keeps native computation but disables the native sink. All runs check row counts and value checksums, and assert the intended physical operators.

Format	Input rows	Sink	Stock ms	Previous fallback ms	Native ms
Parquet	200K	Append	191.07 [180.38, 295.06]	220.07 [211.91, 274.07]	148.84 [135.13, 158.87]
Parquet	200K	Primary key	221.96 [206.20, 224.80]	231.49 [217.94, 250.73]	184.31 [182.43, 186.62]
Parquet	1M	Append	703.70 [694.16, 706.15]	730.81 [724.68, 794.95]	462.32 [448.61, 465.81]
Parquet	1M	Primary key	867.37 [825.96, 874.78]	907.30 [875.65, 1017.15]	585.28 [584.12, 652.56]
ORC	200K	Append	194.52 [181.04, 207.76]	241.62 [203.64, 308.95]	160.28 [154.80, 162.94]
ORC	200K	Primary key	211.62 [210.50, 218.51]	225.05 [222.79, 240.72]	195.06 [182.23, 196.06]
ORC	1M	Append	688.03 [626.36, 706.33]	668.46 [656.55, 760.22]	473.58 [456.77, 489.15]
ORC	1M	Primary key	887.53 [883.98, 890.52]	934.06 [890.45, 962.84]	599.32 [583.64, 646.66]

At 1M input rows the native sinks improve over stock by 1.45–1.52× and over the previous fallback by 1.41–1.58×. At 200K rows, the improvements over stock range from 1.08× to 1.28×. These small jobs are startup-sensitive; the ranges matter, and these results do not establish a win for every job size or configuration.

The source SQL query is SELECT COUNT(*), SUM(v), SUM(CHAR_LENGTH(label)) FROM t over the same committed append files for all engines. Its timer includes planning, startup, decode, conversions, native computation where enabled, and collecting the complete initial-snapshot aggregate. Job cancellation follows the timer. All three aggregates must match. This uses one-phase aggregation with a 4,096-row mini-batch and 10 ms allowed latency. The previous fallback disables the native source but keeps native computation. Plan assertions require the native source and aggregate on the new path.

Format, 1M rows	Stock ms	Previous fallback ms	Native ms
Parquet	588.63 [499.18, 588.88]	548.02 [544.83, 556.08]	372.40 [226.00, 372.73]
ORC	585.81 [551.87, 737.50]	549.34 [399.10, 552.98]	371.40 [225.39, 374.94]

This complete SQL source pipeline improves by 1.58× for both formats against stock, and 1.47–1.48× against the previous fallback. These are local snapshot-to-result measurements, not sustained tailing throughput. Startup and result collection contribute substantially to the subsecond times and their range.

A separate reader diagnostic selects all three columns but sums only v, without a SQL job. Stock consumes Paimon rows directly, the previous boundary copies stock rows into owned storage and transposes them to Arrow, and native consumes Arrow. The same native-written files are used for each engine. Primary-key reads return 10% of the input count after deduplication. This retains native decode/JNI costs but does not represent equal work for unused columns: stock row access can leave string values unmaterialized while the native reader constructs Arrow arrays.

Format	Input rows	Read	Stock ms	Previous boundary ms	Native ms
Parquet	200K	Append	8.57 [8.35, 10.04]	31.66 [29.61, 44.29]	10.40 [8.63, 11.07]
Parquet	200K	Primary key	1.74 [1.67, 2.01]	4.25 [4.11, 8.03]	1.38 [1.19, 1.57]
Parquet	1M	Append	29.61 [26.28, 31.56]	127.97 [117.22, 178.71]	29.68 [29.53, 31.33]
Parquet	1M	Primary key	7.70 [7.49, 8.75]	20.31 [19.99, 24.97]	5.50 [4.00, 5.60]
ORC	200K	Append	3.73 [3.54, 4.02]	37.92 [35.28, 43.38]	4.60 [4.57, 4.89]
ORC	200K	Primary key	1.29 [1.03, 1.36]	4.37 [4.36, 4.45]	1.12 [0.89, 1.18]
ORC	1M	Append	15.07 [11.16, 16.93]	122.57 [109.31, 152.25]	18.05 [17.12, 18.58]
ORC	1M	Primary key	4.40 [4.22, 4.44]	17.83 [17.38, 18.05]	4.01 [2.58, 4.29]

The full-width ORC append reader is 20–23% slower than direct stock row consumption; Parquet's 200K append median is 21% slower, while its 1M median is approximately equal. These unfavorable results are retained. The earlier adapter revision f94255a1 measured a 29–31% raw ORC deficit and a 2% small ORC primary-key sink deficit; the tables above supersede those measurements after integrating main's boundary work. A diagnostic at the earlier revision using the existing projection pushdown to read only v measured ORC 1M append at 8.18 [5.02, 8.37] ms stock, 40.64 [34.03, 52.17] ms previous boundary, and 7.34 [6.79, 9.13] ms native. The overlapping ranges do not establish a reliable raw-reader win. For 100K deduplicated keys, the same projection measured 3.40 [3.30, 3.51] / 8.94 [8.59, 9.11] / 1.03 [0.98, 1.04] ms. The adapter's source benefit is demonstrated in the complete SQL pipeline above; it should not be interpreted as universal superiority of Arrow over row access.

Reproduce each format and size separately, without concurrent test/build jobs:

SF_PAIMON1_BENCHMARK=true SF_PAIMON1_BENCH_ROWS=1000000 \
SF_PAIMON_FILE_FORMAT=parquet SF_PAIMON1_SQL_SOURCE_BENCHMARK=true \
  mvn -Pflink-1.18,paimon1,bench -pl streamfusion-paimon1 -am \
  -Dsf.testForks=1 -Dtest=LegacyPaimonBenchmark \
  -Dsurefire.failIfNoSpecifiedTests=false \
  '-Dsf.extraJvmArgs=-Xmx1536m -XX:ActiveProcessorCount=2' test

Use SF_PAIMON_FILE_FORMAT=orc and SF_PAIMON1_BENCH_ROWS=200000 for the other configurations. SF_PAIMON1_SOURCE_PROJECT_VALUE=true selects the additional projection diagnostic. The default reader diagnostic keeps all columns.

Focused streaming full-compaction diagnostic

A focused reproduction on 2026-10-03 ran the exact released Paimon 1.0.0 PrimaryKeyFileStoreTableITCase.testFullCompactionChangelogProducerStreamingRandom test with Flink 1.18 and Java 17, stock first and StreamFusion second. Both ran one test with zero failures, errors, or skips. Native bundle and level-0 file writes were observed in the accelerated run. Retained results record test durations; these correctness-test durations are not a performance benchmark.

The previously observed 180-second upstream CI timeout did not reproduce in this isolated pair. It remains unresolved: the test uses randomized parameters and checkpoint/restart behavior, and the CI failure occurred within a larger suite. One passing pair neither proves the timeout fixed nor replaces the full upstream parity requirement. Both runs' Surefire reports and diagnostics were retained separately for investigation.

The follow-up full PrimaryKeyFileStoreTableITCase reproduction also passed all 17 tests under stock and native execution (zero failures, errors, or skips). All 34 per-test results are retained, including full-compaction streaming at 18.631 seconds stock and 18.799 seconds native. Native bundle and level-0 writes were observed, and all 10 retained release-library hashes remained unchanged. These are randomized correctness tests, not controlled performance measurements. The timeout remains unresolved: it reproduced in the broader upstream CI suite but not in this isolated test or class.

The subsequent complete default released upstream suite passed all 242 tests with zero failures, errors, or skips under native execution. It used Java 17, Flink 1.18, Paimon 1.0.0, one test fork, two active processors and a 2 GiB test heap; SF_PAIMON1_TESTS was unset. The command was bin/paimon1-suite.sh -Pbench -Dnative.build.skip=true, reusing the verified release libraries. All ten library hashes remained unchanged. Native bundle, level-0 file and merged snapshot witnesses were present. The complete 242 per-test outcomes include both former CI timeout cases passing; the archived Surefire reports and upstream diagnostics retain the broader suite context. This single randomized suite draw does not identify the CI timeout's cause or establish a fix. Its test durations are correctness diagnostics, not performance comparisons.

The aggregate parity workflow for evidence commit f4245e54 subsequently failed the same full-compaction streaming case with an assertion mismatch rather than a timeout: expected 10000|10000.str, observed 477|3948.str after 28.892 seconds. The released upstream suite reported 242 tests, one failure, zero errors and zero skips; the primary-key class reported 17 tests and one failure. The separate Flink 1.18 Paimon check passed, so that green check must not be used to claim aggregate parity. The failing workflow is run 37175053239; its exact Surefire reports and upstream diagnostics were retained from artifact 11294760131. The cause remains under investigation. This evidence contradicts any claim that the earlier passing local suite establishes complete streaming changelog correctness.

The failing invocation ran two concurrent writers. Its runtime diagnostics show Paimon 1.0.0 rejecting a compaction commit with File deletion conflicts detected! Give up committing. at 04:57:53.725 UTC, followed by the affected writer restoring checkpoint 6 at 04:57:54.751 UTC. The exception originates in the released Paimon commit implementation and reports different base and current commit users. These logs establish a commit conflict and recovery, but do not establish that recovery caused the incorrect final value, or that this is the same failure as the stock-only split-commit reproduction above. The retained artifact has no table snapshot or manifest files with which to prove that sequence.

For a focused native upstream-suite reproduction, opt in to retaining a failed fixture’s local warehouse before teardown with -Dstreamfusion.flink-suite.retain-failed-warehouse=true. The test agent copies the released fixture’s path directory under its configured diagnostics directory as failed-warehouses/<test>-<unique-id>, including table snapshots, manifests, and data files. Capture is disabled by default and occurs after a test failure; capture errors are reported without replacing the original failure. A copy made while background jobs are still active is diagnostic evidence, not an atomic checkpoint or a guaranteed consistent table backup.

Java 17 validation of the retention hook passes all seven selected tests with zero failures, errors or skips: four warehouse-capture cases and three existing test-watch diagnostics cases. These validate the diagnostic hook, not recovery correctness in the failing released upstream test.

A focused local run with warehouse capture enabled also passes the released full-compaction streaming test on Flink 1.18.1/Paimon 1.0.0: one test, zero failures, errors or skips, 34.042 seconds. The Surefire report confirms the capture property, native bundle and level-0 write witnesses are present, and all ten release library hashes match the earlier full-suite reproduction before and after this run. Reports, inventory and runtime diagnostics are retained. No failure warehouse is generated for a passing test; this randomized draw does not explain the CI assertion mismatch or establish a recovery fix.

A subsequent PR-head CI run also fails the released Flink 2.2 Paimon full-compaction changelog test. On 505e94d5, job 111379754377 reports testFullCompactionChangelogProducerStreamingRandom expected 10000|10000.str but observed 289|9496.str after 41.774 seconds. This is an actual value mismatch, not a timeout. The preceding 260-test shard passes; the failing focused shard has one failure and subsequent shards pass. These passes do not resolve the mismatch or establish a native cause. Exact job logs are retained; artifact/runtime and retained-warehouse evidence must be examined before attributing the failure or accepting connector parity.

The upstream Paimon suite now forwards the opt-in FLINK_SUITE_RETAIN_FAILED_WAREHOUSE=true to its existing failure-capture hook. Canonical upstream CI enables it on both released Flink lines; local suite runs retain the previous default of false. The existing diagnostics artifact path includes captured warehouses. Capture occurs after a test fails, before fixture cleanup, and capture errors preserve the original failure. It does not alter upstream queries or assertions, repair the value mismatch, or provide an atomic snapshot while background jobs are active. This change makes future failed table files available for diagnosis; the prior artifact has no such files.

Partition pruning for native streaming scans

The planner passes retained VARCHAR partition-column equality predicates to Paimon's released ReadBuilder.withFilter API. Conjunctions may contribute supported partition equalities while other terms remain residuals. Disjunctions, null comparisons, casts, non-partition fields, and other types do not contribute pruning hints. The original Flink filter remains in the plan; source filters consumed by the stock connector require stock-source fallback.

Predicate fields are mapped through the ordered source-ability schemas back to the original full-table column ordinals, independently of the native output projection. Paimon's existing snapshot and continuous split planning applies the partition hint; native file decoding and checkpoint split offsets keep their existing boundaries. The hint is retained in physical-plan copies and shared-source execution nodes. Unfiltered scans use the existing read builder.

The retained partition-equality handoff is admitted for Paimon 2.0. The legacy Paimon 1.0 adapter keeps its existing unfiltered read-builder path and Flink residual filtering; its shared physical source constructor passes no pruning predicate. Shared SQL parity tests verify both versions' exact snapshot/live results and require the version-specific presence or absence of the hint.

Stock replay identifies premature terminal-event completion

The required Flink 2.2/Paimon 2.0 suite on PR #300 captured a full-compaction failure with expected 10000|10000.str but observed 528|5200.str. Artifact 11416621697 retains the failed warehouse, reports, and runtime diagnostics. The table has pt STRING, k INT, v1 BIGINT, v2 STRING, partition pt, primary key (pt,k), four buckets, full-compaction.delta-commits=3, and a 5 MiB local merge buffer. Two producers imply 512 final keys.

A released-stock bounded read of that same captured table passes the original exact result checker for all 512 final rows. Finite stock changelog replay reproduces the identical CI assertion at the upstream stop condition: 512 terminal events contain only 488 distinct terminal keys, leaving 24 incorrect current values. Snapshot 30 includes 12 terminal no-op update-before/update-after pairs; these add 24 counted events without completing another key. The released test counts every RowKind whose v1 >= 10000, including these retractions.

Continuing the stock replay consumes 3,618 rows and 536 terminal events, reaches all 512 final keys, and passes every original RowKind/value transition and final-state assertion. Snapshot 32 contains the final compaction; snapshot 33 is the latest append. Both a temporary guard and the actual suite-agent completion helper, replayed against every stock row, track currently live distinct terminal keys. They complete at snapshot 32 after 536 terminal events and pass the same original checker. This identifies premature event-count termination for this captured failure; it does not justify suppressing no-op records in the native reader or changing the released source. Queries, per-row checks, final assertions, and timeout remain authoritative.

The portable proof artifact records schema, configuration, snapshot IDs, all counters, replay boundary and limitations; its sanitized stock trace includes the cutoff assertion and representative duplicate terminal images. The finite replay starts at snapshot 1 and does not reconstruct live scheduling. Warehouse capture is diagnostic evidence, not an atomic backup. This proof does not establish the cause of every separate legacy timeout.

Nested ROW execution metadata

Internal nested RowKind children remain outside persisted SQL schemas. File schemas and reader projections contain only declared fields; Arrow file writes retain the original value buffers while excluding these execution sidecars. Nested ROWs loaded from files have INSERT kind, matching Flink. Top-level changelog columns retain their existing contract. See nested Java scalar results.