[dragonknightk]

Purpose

Linked issue: close #1230

INSERT INTO a Paimon table through DataFusion encodes every data file in one TableWrite, so Parquet encoding uses one core no matter how many input partitions DataFusion runs. This change lets PaimonDataSink deal input batches to several TableWrites when a bucket-unaware append-only table sets sink.parallelism, and commits their output as one snapshot.

DataSinkExec still requires a single input partition. DataFusion's own DataSink docs leave splitting the input to the sink, and the file sinks do the same inside write_all, so the change stays inside PaimonDataSink and adds no execution plan node. Upstream input partitions already run in parallel under the coalesce. Inside one TableWrite the (partition, bucket) groups of a batch are written in turn and only rolled files close in background tasks, so encoding was the serial part.

The writer count is min(sink.parallelism, session target_partitions). The cap keeps the count within the parallelism the session already grants to its operators. Tables shared with Flink often carry sink.parallelism for Flink jobs; such a table, when it is bucket-unaware append-only, now also writes in parallel from DataFusion, with at most target_partitions writers whatever the Flink value is. The option is validated for every table before the sink decides whether it applies, with the forms Java Integer.parseInt accepts, so an invalid value fails on any table.

How the writers become one commit: all TableWrites come from one WriteBuilder, so they share its commit user and its DataFusion memory reservation. Each writer runs in a SpawnedTask that drains a one-slot channel and then calls prepare_commit. After the input ends, the sink joins every task, concatenates their CommitMessages and passes them to a single TableCommit::commit. Data file names carry a UUID made per file writer, so messages from different writers for the same partition never collide, and the commit produces one snapshot exactly as it does for one writer with several partitions. If any writer or the input fails, the statement returns the error before the commit, so no snapshot is written; files that other writers already closed stay on storage uncommitted, as with a failed single-writer INSERT.

Only bucket-unaware (bucket = -1) append-only tables use several writers, because there any writer may own any file of a partition. Fixed-bucket append tables, primary-key tables, Format Tables and INSERT OVERWRITE keep the single writer.

Differences from Java (Flink):

• sink.parallelism is a Flink connector option, and Java's FlinkSinkBuilder also reads it from the table options; that is why this change reuses the name. Without it Flink derives the writer count from the job parallelism; here the absence of the option keeps one writer, the current behavior. Flink does not cap the value at the input parallelism.
• Flink also honors the option for fixed-bucket and primary-key tables by shuffling rows by bucket, and for overwrite; this change does not.
• Flink rebalances records; this sink deals whole RecordBatches.
• partition.sink-strategy is not implemented. Each writer writes every partition it receives rows for, so on a partitioned table the file count multiplies by the writer count. Java's default strategy NONE behaves the same way; its HASH strategy, which routes a partition to fewer writers, has no counterpart here.

Brief change log

• PaimonDataSink validates sink.parallelism for every table, and for bucket-unaware append-only INSERT INTO uses min(sink.parallelism, target_partitions) TableWrites: input batches are dealt round-robin to spawned writers and their combined messages are committed once.
• The single-writer path is unchanged apart from moving into a branch.
• docs/src/sql.md documents the option, the cap, the file count on partitioned tables and the effect on tables configured for Flink.

Tests

New tests in paimon-datafusion (table/mod.rs) insert 8 in-memory batches of 2 rows into a memory-backed table with target_partitions = 8 unless stated, and read the table back:

• sink.parallelism = 4: all 16 rows, one snapshot, and four data files of 4 rows each (8 batches dealt round-robin to 4 writers). The previous sink leaves one 16-row file and fails this test.
• Partitioned by a column with 2 values, every batch holding one row of each: 16 rows, one snapshot, and 4 writers x 2 partitions = 8 files.
• sink.parallelism = 4 with target_partitions = 2: two 8-row files.
• Without the option: one 16-row file, the existing behavior.
• Table-driven, each with sink.parallelism = 4: a fixed-bucket table, a primary-key table and INSERT OVERWRITE each leave one 16-row file. Removing the primary-key guard or the overwrite guard from the sink makes this test fail with four 4-row files.
• Table-driven invalid values: "0", "-1" and " 4" on an append table and "0" on a primary-key table fail with a configuration error.

The tests assert the writer count through the file layout. They do not assert that the writers run concurrently, and they do not measure CPU utilization.

API and Format

No public API or file format change. sink.parallelism is a new table option read by the DataFusion integration. Tables that already carry the option, for example from Flink, now get parallel INSERT INTO when they are bucket-unaware append-only tables, and fail INSERT if the value is not a positive integer.

Documentation

docs/src/sql.md, INSERT INTO section.

AI Disclosure

Human-led contribution. Grok Bot, working for dragonknightk, assisted with this change. All submitted content was reviewed by a human before submission.

Assumptions and open points:

• Reusing the Flink key sink.parallelism instead of a DataFusion session option is a naming choice that reviewers may want to revisit.
• The dispatcher waits on the next writer's one-slot queue, so a slow writer holds back the others; there is no work stealing between writers.
• All writers draw on one memory reservation, so a high writer count can reach the DataFusion memory limit sooner than one writer would.

(๑•̀ㅂ•́)و✧