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"