[qzyu999]

Summary

Adds ArrowParquetWriter to the parquet-arrow module — a page-level writer that accepts Arrow VectorSchemaRoot batches and produces valid Parquet files without per-row object construction.

Closes #3733. Related: #2264, #3353.

Motivation

Multiple downstream projects (Iceberg #17748, Fluss #4047, Paimon) work with Arrow-columnar data internally but must materialize row objects to write Parquet via ParquetWriter<T>.write(T). This PR provides a direct Arrow-to-Parquet path, eliminating the row-object API mismatch. Arrow C++/Python have had this since write_table().

Design

Bypasses RecordConsumer/ColumnWriter entirely. Writes assembled pages directly to PageWriter with per-column strategy selection:

• ZeroCopyPlainWriter (required fixed-width): wraps Arrow data buffer as page BytesInput. O(1) level encoding. Data bytes are not transformed — Arrow's little-endian layout IS Parquet PLAIN encoding.
• NullablePlainWriter (optional fixed-width): single-pass compaction of non-null values with inline statistics and null counting.
• VarWidthPlainWriter (STRING/BINARY): single-pass offset-to-length-prefix rewrite.
• BooleanPlainWriter (BOOLEAN): bit-pack compaction with null handling.

Does not extend ParquetWriter<T> because write(T) increments an internal record count by 1 per call, incompatible with batch semantics.

Key properties

• V1 page format: length-prefixed RLE-encoded RL/DL sections + PLAIN values
• Buffer safety: ConcatenatingByteBufferCollector.collect() copies page bytes during writePage() — Arrow buffers can be freed after writeBatch() returns
• Thread-safe: StatsResult value object, no shared mutable state
• NaN tracking: float/double statistics exclude NaN from min/max, count NaN separately
• Page chunking: large batches split into ~1MB pages for column-index effectiveness
• Row group flush: based on actual PageWriter.getMemSize() sum, not heuristic
• Null-safe: defensive validity buffer null-checks on all nullable writers

Types supported

INT32, INT64, FLOAT, DOUBLE, BOOLEAN, BINARY (string), FIXED_LEN_BYTE_ARRAY — nullable and required.

Not included (follow-up PRs)

• Dictionary encoding: requires two-pass algorithm (build dictionary, then write RLE indices). Throws UnsupportedOperationException.
• Nested types (List, Map, Struct): requires recursive RL/DL level computation. Throws UnsupportedOperationException.

Relationship to #3530 (Performance Improvements series)

This PR is complementary to the encoding-level optimizations in #3530. Specifically:

• GH-3530: Optimize PLAIN encoding and decoding with direct ByteBuffer I/O #3565 (PLAIN encoding, merged): Made PlainValuesWriter.writeInteger() use direct ByteBuffer.putInt(). Our zero-copy path bypasses PlainValuesWriter entirely for required columns, but nullable/varwidth paths would benefit from GH-3530: Optimize PLAIN encoding and decoding with direct ByteBuffer I/O #3565's faster fallback when dict encoding is eventually added.
• GH-3530: Optimize RLE hybrid encoder/decoder scalar hot-path performance #3568 (RLE codec, open): Our LevelEncoder.encodeFromValidityBitmap() uses RunLengthBitPackingHybridEncoder for nullable columns. GH-3530: Optimize RLE hybrid encoder/decoder scalar hot-path performance #3568's optimizations would make this faster.
• "level write batching" (par9, planned): Would add writeIntegers(int[], offset, count) batch methods to ValuesWriter. When this lands, a future DictionaryWriter strategy could use batch index encoding rather than per-value.

Statistics: irreducible O(N) scan

The stats scan (min/max/NaN count) is the minimum work required to produce a valid Parquet file with predicate pushdown support. For the zero-copy path, it is the ONLY per-value work performed. Future optimizations possible:

• Java Vector API (SIMD) for bulk min/max computation (aligned with Apache Parquet Java Performance Improvements #3530's use of Vector API)
• Pre-computed stats passthrough (if source already has bounds)
• Optional stats disable for latency-sensitive paths

Tests

10 round-trip tests (write via ArrowParquetWriter, read via standard ParquetReader):

• Required INT32/INT64/DOUBLE (ZeroCopyPlain path)
• Nullable INT32 with nulls
• Float/Double with NaN preservation
• Strings with nulls and empty values
• Booleans with nulls
• Mixed schema (int + string + bool + double)
• Multiple batches
• 500K-row large batch with page splitting
• Footer row count verification

Dependencies added

• parquet-hadoop (compile) — ParquetFileWriter, ColumnChunkPageWriteStore
• arrow-memory-netty (test) — Arrow allocator runtime
• hadoop-common + hadoop-mapreduce-client-core (test) — for ParquetReader in round-trip tests
• parquet-hadoop test-jar (test) — GroupReadSupport

Discussion point

Adding parquet-hadoop as a compile dependency to parquet-arrow increases the module's dependency footprint. ParquetFileWriter and ColumnChunkPageWriteStore exist only in parquet-hadoop — no alternative implementations. Every Java Parquet writer (Iceberg, Spark, Flink) depends on parquet-hadoop. Alternative: create a new parquet-arrow-hadoop module. Open to guidance.

How to run tests

mvn test -pl parquet-arrow \
  -Dsurefire.argLine="--add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/sun.nio.ch=ALL-UNNAMED"

[github-actions]

This pull request has been automatically marked as stale because it has had no activity for at least 2 months. If you are still working on this change or plan to move it forward, please leave a comment or push a new commit so we know to keep it open. Otherwise, this PR will be closed automatically in about one month. Thank you for your contribution to Apache Parquet!

[qzyu999]

bump

[mikepapadim]

I ran this PR as a writer for Apache Fluss lake tiering, which is one of the downstream users named in the description. I used head 1856d5b, built against parquet 1.18.2 on JDK 21, and the 10 tests pass. The data was a TPC-H lineitem-like table: 19 nullable columns (INT32, INT64, DOUBLE, DATE, STRING, TIMESTAMP) in 16384-row batches, written with ZSTD. For comparison I wrote the same rows with ExampleParquetWriter.

Page sizing. With dictionary off on both sides, this PR's file is 26.8 MB against 21.7 MB from ExampleParquetWriter (128 MB of Arrow input). The gap is in columns where encoding doesn't matter: a date column is 1.94 vs 0.70 MB, and a constant INT32 column is 30.7 vs 2.7 KB. The cause is in writeBatch:

int typeLen = col.getPrimitiveType().getTypeLength();
estimatedRowBytes += typeLen > 0 ? typeLen : VAR_WIDTH_ESTIMATE_BYTES;
...
int rowsPerPage = Math.max(1, TARGET_PAGE_SIZE_BYTES / Math.max(estimatedRowBytes, 1));

getTypeLength() is 0 for every type except FIXED_LEN_BYTE_ARRAY, so each column is estimated at 32 bytes. Because the estimate is summed over all columns, the 1 MB target is spread across the whole row. With 19 columns that gives 1724 rows a page: about 7 KB INT32 pages and 590 pages a column chunk, where ExampleParquetWriter writes 49. Each page is compressed on its own, so ZSTD gets very little context.

I tried sizing by the widest column instead, using primitive widths (BOOLEAN 1, INT32/FLOAT 4, INT64/DOUBLE 8, INT96 12, FLBA its length, binary 32), and taking the max instead of the sum:

Arrow input	as is	per-column page sizing
128 MB	700 ms, 26.8 MB	457 ms, 21.8 MB
512 MB	2568 ms, 107.5 MB	1733 ms, 87.5 MB

With the change, the output matches ExampleParquetWriter with dictionary off (21.79 vs 21.74 MB), and the write is about a third faster. Pages are still capped at one batch, because pages don't span writeBatch calls. A real fix should probably also estimate each column's bytes separately (or cut pages per column by bytes), so that wide string columns and narrow INT32 columns both get sensible pages. I have this as a small commit with two tests: testPageSizeDoesNotShrinkWithColumnCount, plus an assertion on the page count in testLargeBatchProducesMultiplePages, whose comment expects about 2 pages but gets 16 today. It's up as a PR against your branch: qzyu999#1

Dictionary. After that change, the remaining gap to a dictionary-encoded file (16.9 MB from ExampleParquetWriter) comes from low-cardinality strings and repeated doubles. For example, a 4-value string column is 798 KB plain vs 30 KB with dictionary. So dictionary encoding, which is already listed under follow-ups, matters for size.

For reference, on the same machine (i9-13900K), the row-object path through Iceberg's GenericRecord writer takes 7.3 s for the 512 MB.

JDK note. On JDK 17+, the writer needs --add-opens=java.base/java.nio=ALL-UNNAMED, because ArrowBuf.nioBuffer() goes through MemoryUtil.directBuffer. It might be worth documenting that next to the writer.

Formatting. Running mvn spotless:apply in parquet-arrow reformats 8 of the new main files and the test. CI hasn't run on this PR yet, so that will probably be the first failure once the workflows are approved.

[qzyu999]

Thanks for taking a look and fixing the bug @mikepapadim, merged