Skip to content

[mosaic] Support Arrow bundle writes - #9100

Open
jianguotian wants to merge 5 commits into
apache:masterfrom
jianguotian:feat/direct-bundle-write-20260807
Open

[mosaic] Support Arrow bundle writes#9100
jianguotian wants to merge 5 commits into
apache:masterfrom
jianguotian:feat/direct-bundle-write-20260807

Conversation

@jianguotian

@jianguotian jianguotian commented Aug 7, 2026

Copy link
Copy Markdown
Contributor

Purpose

Support writing eligible Paimon BundleRecords as Arrow batches in Mosaic, avoiding row-by-row materialization when the format writer can preserve the same values, order, accounting, and buffer lifetime as normal row writes.

Mosaic currently requires the input vectors and writer to share one Arrow allocator root. apache/paimon-mosaic#73 adds safe Arrow batch writes across allocator roots. This PR remains correct without that change: it keeps the same-root check and falls back to the existing row path when direct batch consumption is not compatible.

Changes

  • Add an explicit BundleFormatWriter.supportsRowEquivalentBundleWrite() opt-in contract; writeBundle() remains implemented by each format writer.
  • Forward bundles from RowDataFileWriter only when the underlying writer opts in and no auxiliary writer, row tracking, or per-record statistics require row iteration.
  • Let Mosaic flush buffered rows and write schema-compatible, same-root ArrowBundleRecords as one batch; otherwise iterate the bundle through the existing row writer.
  • Preserve record-count and sequence-number accounting for forwarded bundles.
  • Compare Arrow schemas by field name and logical layout while ignoring non-writing metadata such as field descriptions.
  • Enforce variant.shredding.maxInferBufferRow while consuming bundles so a large first bundle cannot bypass the configured inference limit.
  • Keep Arrow, Lance, and Vortex on their existing row-write behavior.

Tests

  • 80 focused tests passed on JDK 11 for bundle dispatch, schema compatibility, row ordering, accounting, shredding limits, Mosaic batch writes, and row fallback.
  • The same 80 focused tests passed after merging with Apache master at 30bfd4a60.
  • JDK 8 affected-module package builds passed for Common, Arrow, Core, Lance, Vortex, and Mosaic.
  • git diff --check passed, and the pushed PR merge tree matches the locally validated integration tree.

@JingsongLi

Copy link
Copy Markdown
Contributor

Thanks for working on this. I wonder whether BundleRecords.isDirectWriteBundle() is needed at all.

As I understand it, this flag does not actually guarantee a zero-copy or native write. It only controls whether RowDataFileWriter may bypass its per-record path and forward the bundle to SingleFileWriter. Once the per-record consumers have been ruled out (auxiliary writers, row tracking, and per-record statistics), it should be safe to forward any BundleRecords:

long rowCount = bundle.rowCount();
if (auxiliaryFileWriters.isEmpty()
        && sequenceNumberTracker.supportsRowCountUpdate()
        && !requiresPerRecordStats()) {
    super.writeBundle(bundle);
    sequenceNumberTracker.updateByRowCount(rowCount);
    return;
}

for (InternalRow row : bundle) {
    write(row);
}

The existing abstractions already provide the remaining dispatch:

  • SingleFileWriter.writeBundle falls back to row iteration when the format writer is not a BundleFormatWriter.
  • A BundleFormatWriter can inspect the bundle type and decide whether to consume it natively, convert or copy it, or fall back to addElement row by row.
  • If the underlying native write is asynchronous, the format writer can copy the borrowed data or acquire independent ownership before returning.

In other words, whether to copy or use a native batch path looks like a format-writer implementation detail. The extra boolean duplicates that dispatch and is slightly misleading: true still does not mean that the write is direct or zero-copy.

Could we remove isDirectWriteBundle() and instead strengthen the contract of BundleFormatWriter.writeBundle() to require semantic equivalence with row-by-row writes and prohibit retaining borrowed buffers after the method returns unless they have been copied or independently retained? This would keep BundleRecords as a data abstraction and leave the write strategy entirely to the format writer.

mingfeng and others added 4 commits August 8, 2026 02:48
- Remove the producer-side direct-write flag.
- Require row-equivalent writes and safe borrowed-buffer handling.
- Add schema and allocator fallback checks.
- Fix row-count and shredding consistency.
- Add regression tests for Arrow, Lance, Mosaic, and Vortex.
…saic writers

Preserve existing Arrow, Lance, and Vortex writer behavior by keeping them on the row path unless a writer explicitly supports row-equivalent bundle writes.

Enforce the shredding inference bound and keep Mosaic native writes behind schema and allocator-root compatibility checks with safe row fallback.
@jianguotian jianguotian changed the title [core] Add direct write path for BundleRecords [arrow][core][mosaic] Enable safe columnar bundle writes for Mosaic Aug 9, 2026
@jianguotian jianguotian changed the title [arrow][core][mosaic] Enable safe columnar bundle writes for Mosaic [mosaic] Support Arrow bundle writes Aug 9, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants