Skip to content

[AURON #2636] Convert only the read columns before a native Paimon write - #2640

Open
COLDMANYU wants to merge 11 commits into
apache:masterfrom
COLDMANYU:paimon-write-narrow-convert
Open

COLDMANYU wants to merge 11 commits into
apache:masterfrom
COLDMANYU:paimon-write-narrow-convert

Conversation

@COLDMANYU

@COLDMANYU COLDMANYU commented Oct 29, 2026 •

Copy link
Copy Markdown
Contributor

Stacked on #2639 (native Paimon write), which is stacked on #2637. Please review only the commit [AURON #2636] Convert only the read columns before a native Paimon write; the last commit is the same fallback-reason change as in #2639.

Which issue does this PR close?

Part of #2636 (follow-up to #2639)

Rationale for this change

INSERT INTO wide_table SELECT ... usually pads most target columns with constants or NULLs.
When the source cannot be read natively, the padding Project runs in the JVM, and the
row→Arrow conversion in front of the native writer has to carry every padded column. In our
job that is 201 columns, while only 18 are actually read. This conversion took about 36% of
task time and cancelled the gain of the native writer.

What changes are included in this PR?

  • NativePaimonWriteExec: if its input is not native and is a ProjectExec
    (possibly wrapped by AQE / WholeStageCodegenExec / InputAdapter) that outputs more
    columns than it reads, convert the project's child to native and run the project natively.
    If the project cannot be converted, the whole input is converted as before.
  • New config spark.auron.paimon.write.narrowConvert, default true. Setting it to false
    restores the previous behaviour.

Are there any user-facing changes?

The new config above. Results are unchanged; only where the row→columnar conversion happens
moves.

How was this patch tested?

  • New test in AuronPaimonWriteSuite (11 tests in total with this PR): a padding projection over a non-native source. It
    asserts that the narrow path was taken (the converted schema has only the read columns) and
    compares the result with Paimon's writer.
  • Production run on the same input as the AIP table: 399 s / 26.55 task-hours → 264 s /
    16.54 task-hours. Output is identical: row count, all-column hash, key distinct count. In the
    event log the RDD chain is ConvertToNative → Project → PaimonSink.

Was this patch authored or co-authored using generative AI tooling?

  • Yes
  • No

Generated-by: Hermes Agent 0.22.2 (Claude Opus 5.5)

… 2.0

Paimon 1.3+ is binary incompatible with code compiled against 1.2
(DataFileMeta became an interface and DataFileMeta.forAppend gained
parameters), so a module compiled against 1.2 fails at planning time with
IncompatibleClassChangeError/NoSuchMethodError on Paimon 2.0. Such
LinkageErrors are not caught by the conversion fallback.

- add a paimon-2.0 profile compiling the module against Paimon 2.0.0, using
  the Scala-suffixed paimon-spark artifact introduced in 1.3
- support Paimon version specific sources (src/{main,test}/paimon-<ver>/scala)
  and include them in spotless/scalastyle and dev/reformat
- expose the profile through auron-build.sh and the Paimon CI matrix
…th Paimon 1.4+

apache/paimon#7098 (Paimon 1.4) changed the __paimon_file_path metadata
column from path.toUri.toString to path.toString. They differ for paths
with characters that need escaping (e.g. a '%' in a partition value), so
the native scan returned a different value than the Paimon reader on
Paimon 2.0. Follow the Paimon version found on the classpath.
Execute DataSourceV2 writes (spark.paimon.write.use-v2-write=true) into
Paimon bucket-unaware append-only parquet tables with a native sink.
Requires the paimon-2.0 profile; disabled by default
(spark.auron.enable.paimon.write).

- native PaimonSinkExec: one open file per partition, size based rolling at
  target-file-size, bounded open files / row group memory, Paimon field ids
  as parquet field ids
- data file naming, column statistics and commit stay in Paimon: paths from
  FileStorePathFactory, stats from Paimon's SimpleStatsExtractor, commit via
  Paimon's BatchWrite (overwrite semantics unchanged)
- unsupported tables (primary key, fixed/dynamic bucket, deletion vectors,
  row tracking, data evolution, clustering, file index, external paths,
  merge-schema, snapshot sequence init, non-parquet, nested/timestamp
  types, row level operations) fall back to the Paimon writer
…mon write

INSERT ... SELECT into a wide table typically pads most columns with
constants. When such a projection sits on top of a source that cannot be
read natively, converting the projection output to Arrow carries every
padded column (201 instead of 18 on a production job), which cancelled
the gain of the native writer. Convert the projection's child instead and
run the projection natively; fall back to converting the whole input when
the projection cannot be executed natively.

Controlled by spark.auron.paimon.write.narrowConvert (default true).
Unsupported Paimon writes were rejected in isSupported() with a WARN log, so the plan only showed 'not marked, default to NeverConvert'. Follow the Iceberg and Hudi scans instead: claim every Paimon V2 write, reject unsupported tables in convert() with an AssertionError, and let Auron record the reason in auron.never.convert.reason. The fallback tests now assert the recorded reason.
Unsupported Paimon writes were rejected in isSupported() with a WARN log, so the plan only showed 'not marked, default to NeverConvert'. Follow the Iceberg and Hudi scans instead: claim every Paimon V2 write, reject unsupported tables in convert() with an AssertionError, and let Auron record the reason in auron.never.convert.reason. The fallback tests now assert the recorded reason.
@COLDMANYU

This comment was marked as resolved.

@COLDMANYU

Copy link
Copy Markdown
Contributor Author

Update: this is done. I merged the latest master into #2637, #2639 and #2640, and adapted PaimonSinkExec to the DataFusion 55 ExecutionPlan API in cc233dc, following the same changes ParquetSinkExec got in #2656. CI is green on all three PRs now.

Sorry, I should have posted a follow-up here instead of just hiding the comment above.

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

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant