[jianguotian]

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.

Performance validation

Local A/B testing measured only the overhead of the new validation: the same direct/native bundle-write path with all checks disabled versus enabled.

• Dimensions: Arrow, Lance, Vortex, and Mosaic; 32- and 128-column mixed-type schemas (numeric, decimal, string, binary, temporal, and array); 128 and 1,024 rows per bundle.
• Method: 2 warmups and 5 alternating paired runs in independent JVMs. Every sample remained on the direct/native path with no row fallback.
• Checks: non-empty Arrow vectors; identity field mapping; matching Paimon logical layout and Arrow schema; and allocator-root compatibility where zero-copy sharing requires it (Arrow, Lance, and Mosaic).

Wall-time change with all checks enabled:

Format	32 columns 128 rows/bundle	32 columns 1,024 rows/bundle	128 columns 128 rows/bundle	128 columns 1,024 rows/bundle
Lance	+4.28%	+4.55%	+0.34%	-0.01%
Vortex	+1.07%	+0.26%	+1.54%	+0.21%
Mosaic	+4.50%	-2.24%	+1.04%	+0.81%

Across the 16 scenarios, the median overhead was +1.30%; 14 scenarios were within 5%, with a maximum of +6.84%. Small negative deltas are treated as run-to-run variance, not as a performance improvement. Files produced with checks enabled and disabled matched in both values and row order when read back.

[JingsongLi]

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.

[JingsongLi]

We should remove supportsRowEquivalentBundleWrite too.

[jianguotian]

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.

[JingsongLi]

+1