Skip to content

[core][format] Support Arrow bundle writes - #9100

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

[core][format] Support Arrow bundle writes#9100
jianguotian wants to merge 7 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 without row-by-row materialization when format-specific batch handling is compatible.

Bundles that require per-row processing or are incompatible with the target format writer continue to use the existing row path.

Changes

  • Forward bundles from RowDataFileWriter when no auxiliary file writer, row-level sequence tracking, or per-record statistics require row iteration.
  • Let each BundleFormatWriter consume compatible bundles natively, copy them, or fall back to rows while preserving row-equivalent values, ordering, accounting, and borrowed-buffer lifetime.
  • Validate Arrow schema, logical layout, field mapping, and allocator compatibility before batch writes.
  • Flush buffered rows before compatible bundle writes to preserve write order.
  • Write compatible Arrow bundles directly and release unconsumed Arrow C Data resources on failures.
  • Write compatible same-root Lance bundles directly and fall back to rows for incompatible bundles.
  • Copy compatible Vortex bundles into writer-owned memory before native writes.
  • Write compatible same-root Mosaic bundles directly and fall back to rows for incompatible bundles.
  • Preserve record-count and sequence-number accounting, statistics behavior, and variant.shredding.maxInferBufferRow.

Tests

  • Cover bundle forwarding and row fallback with auxiliary writers, row tracking, and per-record statistics.
  • Cover schema reorder, missing columns, non-identity mappings, row/bundle ordering, and record counting.
  • Cover allocator compatibility, borrowed-buffer lifetime, native failure cleanup, and Vortex asynchronous writes.
  • Cover shredding inference limits for bundled records.

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

Copy link
Copy Markdown
Contributor

We should remove supportsRowEquivalentBundleWrite too.

@jianguotian jianguotian changed the title [mosaic] Support Arrow bundle writes [core][format] Support Arrow bundle writes Aug 10, 2026
@jianguotian

Copy link
Copy Markdown
Contributor Author

undleRecords.isDirectWriteBundle()

Both removed.
Since these writeBundle methods can now be reached through the general RowDataFileWriter path, I also hardened the Arrow, Lance, and Vortex implementations for schema/layout compatibility, pending-row ordering, allocator and buffer lifetime, and failure cleanup. Incompatible bundles fall back to row writes.

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