The native Parquet reader (VectorParquetScanExec)

This page is the design note for VecRuntime's own Parquet scan, #559. It covers what the scan does today, how it is put together, what it does not decode yet, and the work that is left.

In short (2026-10-18): the scan reads every Parquet type Spark 4.1 reads and every encoding parquet-java's v1 and v2 writers produce, natively: flat types, unsigned integers, integers read as wide decimals, ANSI intervals, Spark 4.1's time (any INT64 unit) and Spark 4's variant, shredded or not, with variant extraction pushed into the scan (#787, #812). Structs, lists and maps nested in one another at any depth are read natively too (#568), and so are INT96 timestamps and files in any calendar rebase mode, files older than the table (missing columns, existence defaults, widened types), collated strings, bucketed tables and the _metadata columns. Over the whole apache/parquet-testing and Hardwood corpora (303 files, #808), every file gives Spark's rows or Spark's refusal, and all but two are read natively. What still goes to Spark's reader is listed, with the reason, under Current limits. On the 1 TB TPC-DS run of 2026-10-02 it took VecRuntime to 1,542 s for the 103 queries, against 3,099 s for Spark, 1,999 s for Comet and 1,954 s for Comet's reader under our operators.

The scan is on by default since 0.0.6 (spark.vecruntime.scan.nativeParquet.enabled), except in a session where Comet's scan is active: Comet's plugin or extension registered, with spark.comet.enabled and spark.comet.scan.enabled not false. There Comet's reader stays the scan unless the key is set. Setting the key to false brings back Spark's vectorized reader under our operators. Its configuration keys are in the Configuration reference, and the planner's fallback reasons are on the Operators page.

# Why the scan exists

Spark's vectorized Parquet reader decodes into on-heap ColumnVectors. VecRuntime's operators then copy those batches into native Arrow memory, once per operator chain. Comet's native scan avoids that copy, but it brings its own operators and memory model.

VectorParquetScanExec decodes Parquet pages straight into the Arrow buffers that VecRuntime's operators read, so there is no conversion between the scan and the first operator. It reuses everything Spark already computed for the scan:

• the selected partitions, including dynamic partition pruning;
• the file splits and bucketing;
• the pushed data filters;
• the required schema and the partition schema.

# Data path

FileSourceScanExec (planned by Spark)
  └─ VectorParquetScanExec           (replaces it when the plan is supported; VectorParquetScanPlanner)
       └─ VectorParquetRDD            (one partition per Spark FilePartition)
            └─ VectorParquetPartitionReader   (task thread)
                 ├─ prepare(file)     open: getFileStatus, footer, clip to the required schema,
                 │                    encoding check, ParquetFileReader.open, first filtered row group
                 ├─ install(prepared) column readers, partition values, or the per-file fallback
                 └─ next()            NativeParquetColumnReader per column → ColumnChunkDecoder (kernels)
                                      → batch-owned Arrow vectors + constant partition columns

Opening a file. Each split opens its file once. The FileStatus comes from the split instead of a separate HEAD request (#559 slice 1). The footer is read in a single stream, and readNextFilteredRowGroup applies the pushed filters through parquet-java: row-group statistics, dictionary and bloom filters, and the column index for page skipping (#568), off by default since #873 (below): with it on, pages outside the filter's row ranges are dropped and each column reader decodes only the selected rows. Rows are not otherwise filtered inside the decoder. Each task prefers the same hosts as Spark's FileScanRDD: up to three holding the most of its bytes, with localhost dropped. S3A reports localhost for every block, so on S3 the tasks have no preference, as Spark's do; on HDFS they keep block locality. Comet's equivalent switch, spark.comet.parquet.rowFilterPushdown.enabled, defaults to off as well.

Decoding. NativeParquetColumnReader turns each DataPageV1 or DataPageV2 into a kernel page. Definition levels and values are written into one reusable buffer per reader. ColumnChunkDecoder then decodes, at a row offset into the output:

• the validity bitmap, built from the definition levels;
• fixed-width lanes, written straight into the Arrow data buffer;
• dictionary indices, resolved against a decoded dictionary held in a GC-managed arena, for every lane but strings;
• for a string column (#612, …dictionaryStrings, on by default), dictionary ids instead: the batch is a VectorDictionaryColumnVector over one Arrow dictionary per row group, shared by its batches and reference-counted, closed when the reader and the last batch let go. A data page that falls back from the dictionary (parquet-java does so mid-chunk once the dictionary outgrows its page limit) resolves the ids already in the batch into bytes, and the rest of the row group decodes flat.

Parquet-java's BytePacker is injected for the bit-unpacking. Vectors are pooled and reused from one row group to the next.

Per-file fallback. A file whose column-chunk metadata lists an encoding the decoder does not handle goes to SparkFallbackFileReader, which is Spark's own vectorized reader for that one file. The check runs when the file is opened, so results stay correct and nothing fails part-way through a decode.

# Read-ahead

At 1 TB, TPC-DS store_sales is about 14.6k files of about 7 MB, and most of them hold a single row group. When everything ran on the task thread, JFR showed task threads parked on S3 for at least 66% of their time, and executors used 2–3 of their 13 cores.

The read-ahead fixes that with two independent depths:

A few rules hold the design together:

• Ordering. Row-group reads form a chained CompletableFuture pipeline: each read starts after the previous one completes. A ParquetFileReader is therefore used by one thread at a time, and row groups arrive in file order.
• Decode on the task thread. Pages stay compressed until the task thread decompresses and decodes them, in order.
• Task context. Prefetch steps run with the task's TaskContext set. Files opened ahead and not used are closed at task end, including when a LIMIT stops the task early.
• Platform threads. The pool is cached daemon platform threads; the number alive is bounded by task slots × depth. Virtual threads deadlocked on JDK 25. A file open runs class initializers (Parquet's BloomFilterImpl, the S3 client), and a virtual thread blocked in a class initializer still pins its carrier. On q88, every carrier of one executor was pinned behind BloomFilterImpl.<clinit>, which was waiting for a log4j lock held by an unmounted virtual thread. VectorParquetPrefetchPoolSuite checks the pool's thread kind.
• Metrics. Task input metrics follow FileScanRDD: records are counted per batch, and bytes are the task thread's FileSystem read count plus the bytes each prefetch step read on its own thread.
• Memory. Per task, read-ahead adds up to N opened files and M row groups of compressed pages, capped in bytes by prefetchMaxBytes (default 256 MiB, #568; see the work list).

Decode-ahead (#606, off by default). With …nativeParquet.decodeAhead = K, the reader loop runs on a producer thread of the task's own, up to K batches ahead (DecodeAheadIterator). The task thread takes batches in order and runs the operators above the scan on them, while the producer decodes the next ones across row groups and files.

• Ownership. A batch is the consumer's until its next next(), when its vectors go back to their column reader's pool. The pool is guarded for the two threads; a vector released after its file closed is closed instead.
• Fallback files. A batch of Spark's fallback reader is recycled by that reader, so the producer waits until the consumer has moved past it.
• Threads. A fresh virtual thread per task by default, or a platform thread. The first batch is decoded on the task thread, so the decoder's classes are initialized before a virtual thread runs it; file opens stay on the platform prefetch pool. A fresh thread also keeps the task's FileSystem byte count right.
• Close. At task end, LIMIT or failure, the producer is stopped, the queue is drained, the producer is joined, and only then is the reader closed. A producer failure is rethrown on the task thread in order.
• Tests. DecodeAheadSuite checks results against Spark's reader for K = 1 and 4 on both thread kinds, across row groups and files; a LIMIT that ends the task early, with the allocator back at its baseline; a table mixing native and fallback files; and a producer failure.

Decoding the next row group ahead within a split. The two stages compose into exactly that: prefetchRowGroups reads the next row group's compressed pages ahead (chained, in order), and decodeAhead decompresses and decodes a row group's pages on the producer thread while the task thread consumes the batches of the row group before it. Both run on the bounded prefetch pool and both reserve against the same execution-memory accounting (ReadAheadMemory, #915) — compressed pages while a row group waits, decoded Arrow bytes while a batch is queued — given back when the row group or batch is used or dropped, so a task ended mid-split (a LIMIT, a kill) releases every reserved byte, held byte and Arrow vector through the close path. Row order is preserved end to end. The decode stays on one producer thread rather than a second column-reader set per in-flight row group: a row group's column readers are stateful and single-instance, decode is CPU-bound (a second decode thread per task would contend for the same cores, not overlap I/O the way prefetchRowGroups already does), and the existing producer already overlaps decode with the operators above the scan. NativeParquetDecodeAheadSuite drives multi-row-group splits with both stages on: rows equal the sequential path, order holds, a mid-split LIMIT releases all read-ahead memory and vectors, and a local timing confirms the ahead path is no slower.

Arena lifetime. A column reader exists per column per file, so it must not close a shared FFM arena. Closing a shared arena triggers a JVM-wide handshake, and on q88 those handshakes cost 8–9% of executor CPU; with GC-managed scratch they are 0.2–0.4%.

# How reads are issued (#834)

On the classic S3A stream each ParquetFileReader read is its own request, and at 1 TB the scan-bound queries were request-latency-bound, not byte-bound: store_sales is ~14.6k files of ~7 MB, mostly one row group, so a file costs a footer-length read, a footer read, the row-group reads and — with the column index on — a GET per selected page. Switching S3A off the Analytics Accelerator (AAL 1.3.1) to the classic stream took the 103-query suite from 846 s to 664 s but gave up three things AAL did; #834 does two of them inside our scan, scoped to what the reader already knows (the file size from the split, the footer, the exact column chunks), with the Accelerator's design as reference but none of its process-wide shared state. Both are off by default until a 1 TB A/B sets the thresholds; neither changes results (same rows, same checksums — NativeParquetRangeMergeSuite).

1. Small-object prefetch (…nativeParquet.smallFileThreshold, a byte size; 0 off). A file at or under the threshold is fetched whole in one ranged GET, on the prefetch thread, before the footer is read; ParquetFileReader is then handed a BufferBackedSeekableInputStream over that buffer, so the footer, every column chunk and the page indexes all come from memory with no further request. That turns the per-file footer-length + footer + row-group(s) (+ per-page, under the column index) reads into one. The buffer is heap byte[], bounded by the existing read-ahead budgets (prefetchExecutorMaxBytes). The threshold itself is the gate, as AAL prefetches a small object whole; size it, from AnyBlob's (PVLDB 2023) measured optimal request size, at the point where reading every byte beats several ranged reads (the ~7 MB store_sales files are the target).

2. Range merging (…nativeParquet.rangeMergeGap, a byte size; -1 off, 0 adjacent-only). When a file is not read whole, ParquetReadRangePlanner plans the exact byte ranges the reader needs — the projected column chunks of the row groups the split keeps, and with the column index on only the selected pages plus the offset/column index reads — and (a) coalesces ranges whose gap is at or under rangeMergeGap into one larger GET, and (b) splits a coalesced run longer than rangeTargetSize (× a 1.4 tolerance ratio, short tail folded back) into parts issued in parallel on the prefetch threads, so one long run is not a single giant stream that starves bandwidth (DuckDB 2.0's too-few-too-large result). The planned ranges are read into a BufferBackedSeekableInputStream and served to parquet-java; the original stream is kept as the fallback for any byte the plan did not cover, so a miss is correct, never wrong. It engages only when it reduces the request count. The planner is pure arithmetic over half-open ranges and is unit-tested on its own (ParquetReadRangePlannerSuite): the coalesce/split invariant is that every input byte lands in exactly one sorted, non-overlapping output range.

Both buffers live for the file's reader and are read on the prefetch thread, so they fit the existing single-owner-per-reader rule (no ParquetFileReader touched by two threads). Request-count reductions are proven with a counting FileSystem (NativeParquetRangeMergeSuite): a small file reads in one request (footer and chunks from the buffer), and with the column index on a selective filter over a small file is still one request rather than one GET per selected page. Not done here: the block cache (#834 part 3) — our scan reads each byte range once, so a general cache buys little; a footer cache is left for a measurement that shows reuse.

# Support matrix

What happens when a column is not supported: the scan keeps Spark's result in every case, through Spark's reader for one file or Spark's scan for the plan, as listed under Current limits below. input_file_name() under decode-ahead is native (#568). A bucketed scan is native (#568): it reports Spark's bucket partitioning and reads Spark's own bucketed file partitions (pruned and coalesced as Spark plans them), so a join of tables bucketed alike still needs no shuffle.

Conformance corpus (#808). ParquetCorpusConformanceSuite reads every file of apache/parquet-testing and of Hardwood's test corpus and pins how each one is read; see "Whole-corpus conformance" below.

# Current limits

Everything below returns Spark's own result: either Spark's reader decodes it, or Spark's scan stays in the plan, or the query fails with Spark's error. None of these limits is a wrong answer. The planner records a reason for each whole-plan fallback; the Operators page lists them.

On Spark's reader on purpose, because Spark's outcome there is an error, or a reading only Spark's reader defines:

• Encrypted Parquet (modular encryption) is not a native feature. Decryption is parquet-java's: with Spark's crypto factory and KMS settings configured, parquet-java decrypts the footer and pages and the scan decodes them; without the keys the read fails as Spark's does. The footer-and-columns-encrypted path is tested (NativeParquetPerFileSuite). KMS integrations, plaintext footers and per-column key mixes are not tested; they are Spark's and parquet-java's to support.
• A UTC-adjusted timestamp read as timestamp_ntz (Spark refuses it), INT32 TIME(MILLIS) (Spark refuses it), and a missing REQUIRED column (Spark fails the read).
• The field-id corners: a file with no field ids under a read schema with ids (Spark's error unless ignoreMissing), and an id match carrying a sibling's name (Spark reads it as the sibling).
• A legacy two-level list whose element is a repeated group holding a repeated field (Hardwood's list_of_lists_legacy_two_level_test). Spark's backward-compatibility rule reads it as array<int> over two repeated levels, returning [1, 2] where the file holds [[1, 2], [3]]. No writer produces that shape for that type, and the native readers do not follow it.
• A bare repeated group annotated VARIANT (Hardwood's repeated_variant_group_test), which Spark clips only against its own array<variant>.

Spark's reader for that one file (the rest of the scan stays native): a shredded variant below the top level, inside a struct, a list or a map.

Spark's scan for the whole plan:

• A variant extraction Spark pushed into the scan below the top level (spark.sql.variant.pushVariantIntoScan).
• On Spark 4.2, its geospatial types (geometry, geography). There is no decode path for them; the file's other columns are native.
• Any type Spark 4.1 cannot read itself (FLOAT16, the legacy INTERVAL converted type, 12-byte timestamps): Spark refuses the file before any scan runs. CalendarInterval never reaches the scan, because Spark cannot write it to Parquet. (A UUID column — FIXED_LEN_BYTE_ARRAY(16) with the UUID logical type — is read natively as its canonical-text string, #rtypes.)

In the operators above the scan:

• variant has no engine lane, by design. Spark defines no equality or order on a variant, so it can be no grouping, join or sort key, and every function over one interprets its binary encoding. Operators carry it; an operator that computes on one stays Spark's.
• time and the ANSI intervals ride the INT64 / INT32 lanes for carrying, comparison, null tests, grouping, sorting, joins, windows, count, min, max, first and last. Arithmetic, casts, functions, sum and avg over them stay Spark's. A row input holding a time keeps the operator above it Spark's, because Spark 4.1's RowToColumnConverter has no time converter.

Performance, not correctness:

• Late materialization (lateMaterialization, #611/#962) defaults to auto: the decode filter is attached at plan time only when the filter columns are a small fraction of the projected bytes, and the reader then decides per row group whether to late-materialize it from the row group's statistics, dictionary pages, bloom filters and the observed selectivity of earlier row groups (with hysteresis), late-materializing only when the estimated surviving fraction is low and decoding eagerly otherwise. off/on force it off/on. A global flip regressed at 1 TB (q24b +11.5%, #901), which auto avoids by keeping the eager path except where a filter is selective on a wide projection. Its two-pass mode (#812), which also skips the I/O of pages with no surviving row, stays behind its own twoPass flag; under auto it fires only for very-low-selectivity row groups.
• Column-index page skipping depends on the file's offset index, which parquet-java writes by default. A file without one is read over whole row groups, as Spark reads it.

# Codecs

Compression is not a decode concern of the scan's own. The scan installs ReusingCodecFactory, which decompresses zstd and snappy through one reused buffer per column chunk (#810) and hands every other codec to parquet-java's own CodecFactory; each codec's decompressor then lives as long as its file (#828, #829). So the codec is never a fall-back reason — VectorParquetScanExec.hasUnsupportedEncodingInFooter checks only encodings and physical types — and a file in any codec whose decompressor is on the classpath is read natively, byte for byte as Spark's own reader (which uses the same parquet-java decompressors). NativeParquetCodecSuite pins this for every such codec across v1/v2 pages, dictionary/plain pages and flat/nested columns. Only zstd and snappy have the per-page reuse; the others are no slower than in Spark's reader but do not get it. Dedicated gzip and LZ4_RAW paths are tracked in #830, to be measured first.

The codecs above are all decoded by aircompressor's pure-Java codecs, which ship on the classpath, so the shipped build reads every one of them. BROTLI and LZO are the exception: no Brotli or LZO codec is on the classpath (not in aircompressor, parquet-java or Spark — LZO is GPL and not shipped with Hadoop), so a file in either is read by neither Spark's reader nor this one. Because the codec is not a fall-back reason, such a file is not diverted to Spark's reader — the read reaches the decompressor and parquet-java raises BadConfigurationException (the missing BrotliCodec/LzoCodec class), exactly as Spark's own reader would raise it for the same file. Decoding Brotli natively would add a new runtime dependency (brotli4j or org.brotli:dec); it is left out until that dependency is accepted, and would then wire into ReusingCodecFactory beside zstd and snappy.

# Default measurement: column index vs late materialization (SF10, local)

A local SF10 A/B of the two page-skipping defaults (local[8], warm OS cache, so this isolates decode CPU, not the S3 I/O the 1 TB numbers above measure), median of 3 runs after 1 warm-up, each flag combination its own JVM, on selective TPC-DS queries:

Total task CPU (executorRunTimeMs) moves the same way: late materialization lowers it on every query here.

What this shows, and the recommendation — defaults are left unchanged here, this is input for the maintainer:

• Late materialization is the consistent win on these selective queries: faster on all seven (−4% to −38%, ~−24% median) with lower task CPU throughout, no regression. These are exactly the selective-filter shapes it targets. But its current false default rests on a 1 TB reading where q24b was 10% slower (the regression a warm local cache cannot reproduce), and on minScanBytes=1 GiB which an SF10 scan does not clear without the override used here. Recommend: keep the default false, but settle it with a 1 TB A/B over this query set — the local evidence is strong enough to justify that run, which the 1 TB guidance in docs/configuration.md has not yet covered for these queries.
• Column index helps most queries (−6% to −30%) but regressed q19 by +13% even on a warm local cache with no per-page S3 request to pay — consistent with the documented #873 reason (index/offset reads and per-page bookkeeping that only range-merging, #834, pays back). Recommend: keep the default false until #834 lands; the local data does not contradict the existing guidance.
• Two-pass is roughly even with plain late materialization here (better on q55, worse on q42/q52): its distinctive benefit is skipping the page I/O of non-surviving rows, which a warm local cache does not exercise, so this bench cannot credit it. Recommend: keep the default false; it needs the 1 TB I/O-bound measurement its config note already calls for.

The three suites that cover these paths — NativeParquetColumnIndexSuite, LateMaterializationSuite, TwoPassLateMaterializationSuite — pass with the flags on.

# Status at 1 TB TPC-DS

Full run, 2026-10-02. All 103 queries once, the four engines back to back in one session (8 × m5.4xlarge, one AZ, the launcher's defaults):

The scan-bound queries gain most: q88 35 s against Spark's 96 and Comet's 84, q9 31 against 73 and 51, q28 58 against 102 and 67. The per-query tables are in docs/results.md.

The scan-bound queries, flag off against flag on. These numbers come from 8 × m5.4xlarge in one AZ, with ACCP and the S3A read settings on. The flag-off and flag-on legs ran in one session; checksums equal the references.

The flag-on column comes from the read-ahead build with the arena fix. Comet's column comes from an earlier session that had the read-ahead but not the arena fix; that session's flag-on numbers were q28 46.6 s, q9 31.0 s, q44 14.1 s and q88 83.5 s, already ahead of Comet on the first three. The flag-on/off pair (round 14) and the Comet column (round 10) are from different sessions, so compare across that boundary with care.

Where executor CPU goes on q88 with the flag on (async-profiler):

# Work list

The items run roughly from most to least valuable.

# Encodings

These are the Parquet v2 writer encodings. A file that uses one the reader does not decode yet goes entirely to Spark's reader. Parquet-java and Spark 3.x/4.x write them when parquet.writer.version=v2, and other engines often do by default.

1. DELTA_BINARY_PACKED for INT32 and INT64: done (slice 2). DeltaBinaryPackedReader reads the block and miniblock headers, unpacks each miniblock through the injected BytePacker (widths up to 32; INT64 widths above 32 with a scalar long unpack), and adds the deltas in the column's width, wrapping as the writer does. It decodes one miniblock at a time, so a batch can stop inside one. ColumnChunkDecoder writes the values straight into the lane, including INT32 widened to INT64. The footer check admits the encoding only on INT32/INT64 columns. It is checked:

   • against a from-scratch encoder of the spec (DeltaBinaryPackedReaderTest, ColumnChunkDecoderTest);
   • against parquet-java's own writers (DeltaBinaryPackedCrossCheckSuite);
   • through Spark with parquet.writer.version=v2 and the dictionary off (VectorParquetScanSuite).

   DeltaBinaryPackedBenchmark (JMH, 20,000 values a page) decodes 2.0–4.0x as many pages per ms as Spark's VectorizedDeltaBinaryPackedReader:

   values	ours (ops/ms)	Spark (ops/ms)
   sorted INT32	35.2	17.7
   random INT32	24.4	10.3
   INT64 timestamps	32.7	14.9
   44-bit random INT64	29.1	7.2

   That is a kernel check, not a verdict; no TPC-DS file uses the encoding.

2. DELTA_LENGTH_BYTE_ARRAY for strings: done. At feedPage, DeltaBinaryPackedReader decodes the page's lengths. Its position() after the last value is where the bytes start; the spec pads the last miniblock, so this is exact. Batches then copy the bytes into Arrow offsets and data like PLAIN. The total length is checked against the page's extent. JMH: 1.46x the pages per ms of parquet-java's DeltaLengthByteArrayValuesReader (10.2 against 7.0, 20,000 short strings, 1024-row batches).

3. DELTA_BYTE_ARRAY for strings: done. This is what parquet-java writes for v2 strings without a dictionary. The page is a DELTA_BINARY_PACKED run of prefix lengths (bytes shared with the previous value), then a DELTA_LENGTH_BYTE_ARRAY of the suffixes. Each value needs the previous one, so feedPage rebuilds the page's values back to back in one sequential pass into a reused buffer, and the batches copy out of it like DELTA_LENGTH_BYTE_ARRAY. The prefix chain restarts on each page. A prefix longer than the previous value, a count mismatch between the two runs, or suffix bytes past the page end fail the page. JMH: 2.0x the pages per ms of parquet-java's DeltaByteArrayReader (2.54 against 1.26, 20,000 sorted URL-like keys).

4. BYTE_STREAM_SPLIT for INT32, INT64 and DOUBLE: done. Byte j of value i is at stream j, position i, and the stride is the value region's length over the width. The scan now passes each page's real extent to the decoder, not its reused buffer's. Each value's K bytes are gathered from the streams straight into the lane, including INT32 widened to INT64. FLOAT and fixed-length byte arrays have no lane. Spark's vectorized reader rejects this encoding on INT32/INT64, so those files read only with the flag on. JMH: 8.8x the pages per ms of parquet-java's ByteStreamSplitValuesReaderForDouble / ForLong (20.5 against 2.3). It is a transpose (Vector API or SWAR, by vector width); see the optimization pass below.

5. RLE for values: done, for BOOLEAN (the only physical type it applies to as a value encoding). The page holds a 4-byte little-endian length, then a width-1 RLE/bit-packed hybrid stream. RleBitPackingReader.readBits decodes straight into the batch's bitmap: an RLE run of 1s sets its bit range a word at a time, a run of 0s writes nothing, and a width-1 bit-packed run is already an LSB-first bitmap, so it moves 64 bits a step. See BOOLEAN below for the numbers.

Optimization pass on items 1–4. JMH, one 20,000-value page in 1024-row batches, ops/ms. Each before/after pair ran on one host: x86 is an AVX-512 host, Graviton4 is m8g.4xlarge (Neoverse V2) with the default SVE codegen. Graviton4 with NEON codegen (-XX:UseSVE=0) matched SVE within a few percent, except the strings: DELTA_BYTE_ARRAY 2.8 → 3.6, DELTA_LENGTH_BYTE_ARRAY 7.0 → 7.6.

The transpose and the prefix sum are switches, read once at startup, for A/B runs:

• -Dvecruntime.parquet.bssMode=auto|vector|swar|scalar, default auto. auto uses the Vector API for 4-byte values at every width, and for 8-byte values when a register holds at least four longs. At 128 bits (NEON, or SVE on Graviton4) widening a byte vector into two long lanes is not intrinsified, and the 8-byte vector transpose ran 0.45 against SWAR's 31. It is the same on x86 forced to -Dvecruntime.vectorBits=128.
• -Dvecruntime.parquet.deltaScan=scalar|vector, default scalar. The vector prefix sum (a log-step scan per register) gives the same wrapping sums. It measured the same as the serial add on x86 (43.5 against 42.4) and slower on Graviton4 NEON (28.9 against 37.0).

ByteStreamSplitKernelsTest compares every variant, and the dispatch, with its scalar reference at every count, offset and array-end position. It passes at 128, 256 and 512 bits (-Dvecruntime.vectorBits).

For each encoding:

• Add a ParquetPageDecoder.Encoding case.
• Add a ColumnChunkDecoder path that writes at a row offset.
• Admit the encoding in SupportedEncodings only once all of these pass:
  • cross-check tests against parquet-java's own readers, in the style of ParquetPageDecoderCrossCheckSuite;
  • a round trip through Spark with parquet.writer.version=v2;
  • a decode micro-benchmark in ParquetDecodeBenchmark.

# Types

Physical types are checked per file. The planner admits a column by its Spark type, but each file decides how the column is stored. A decimal(p <= 18) can be INT32, INT64, FIXED_LEN_BYTE_ARRAY or BYTE_ARRAY: Spark's legacy writer, Hive and Impala use FIXED_LEN_BYTE_ARRAY. At open, VectorParquetScanExec.physicalMatches checks every requested column's physical type against its lane, and a mismatch falls that file over to Spark's reader, like an unsupported encoding. Before this check (0.0.4–0.0.5), a FIXED_LEN_BYTE_ARRAY decimal was decoded as INT64 and returned wrong values. The apache/parquet-testing corpus found it.

Conformance corpus. ParquetTestingCorpusSuite reads 27 files from apache/parquet-testing. They are vendored under spark/src/test/resources/parquet-testing/, Apache-2.0, at the commit in SOURCE_SHA. The files were written by parquet-mr of several ages, parquet-cpp, arrow-rs and Impala, and cover v2 encodings, empty and null pages, checksums, a dictionary page at offset 0, and corrupt files from bad_data/. The suite reads each file column by column:

• a column the scan supports must be read natively, with no per-file fallback, and match Spark's row-based reader;
• any other column must return what Spark returns with the plugin off, rows or refusal;
• a corrupt file Spark refuses must be refused too.

The corpus approach follows Hardwood (hardwood-hq/hardwood), as does bounding every file-declared size before allocating: the DELTA_BINARY_PACKED block size, a BYTE_STREAM_SPLIT region that is not a whole number of values, and bit-packed runs past the stream end.

Whole-corpus conformance (#808). ParquetCorpusConformanceSuite reads every Parquet file of apache/parquet-testing (data/ and bad_data/, 80 files) and of Hardwood's core test corpus (spark/src/test/resources/hardwood/, 223 files, Apache-2.0, commit in its SOURCE_SHA): every physical and logical type, encodings, codecs, page and bloom-filter indexes, nested shapes, variants and malformed files. Each file is read with the plugin on and checked against Spark's outcome with it off (its vectorized reader, else its row-based reader, else the refusal): the same rows by value, or both refuse. How the plugin read it is pinned per file, and per column when the file is not native as a whole, in parquet-corpus-expected.tsv; a read that changes fails the suite until CORPUS_UPDATE_BASELINE=true regenerates the file, so a fallback is always a recorded choice.

Today: 240 files are read natively, 2 are empty, 59 are refused by Spark and by the plugin alike (malformed files; types Spark 4.1 does not read, such as FLOAT16, the legacy INTERVAL converted type, 12-byte timestamps and geospatial types; encrypted files read without keys; a few legacy list and map shapes Spark rejects), and 2 go to Spark's reader for the file:

• list_of_lists_legacy_two_level_test: a legacy two-level list, mylist (LIST) { repeated group bag { repeated int32 num } }. Spark's backward-compatibility rule (isElementType) does not take bag as the element, so Spark reads mylist as array<int> over two repeated levels (on this file it returns [1, 2], where Hardwood reads [{num: [1, 2]}, {num: [3]}]), a shape no Parquet writer produces for that type and the native readers do not follow. Kept on Spark's reader on purpose, for Spark's reading.
• repeated_variant_group_test: a bare repeated group annotated VARIANT, which Spark clips only against its own array<variant>.

On Spark 4.2 one file reads differently, recorded in parquet-corpus-expected-spark-4.2.tsv, which replaces the base's lines for it: Spark 4.2 reads the GEOMETRY column of geospatial_e2e_test (Spark 4.1 refuses the file), and the native scan has no decode path for Spark's geospatial types, so a plan reading that column keeps Spark's scan; its other columns are native.

The parquet-java reader tests themselves build their files in code; the cases they cover (encodings, page versions, dictionary fallbacks, nulls, nesting, schema evolution, field ids) are exercised by the suites listed under each feature below.

• BOOLEAN. Done: v1 PLAIN (bit-packed, LSB first) and v2 RLE, into a BOOL lane (Arrow's bit-packed vector). PLAIN with every row present copies up to 64 bits a step from any source bit offset to any destination bit offset. With nulls, the bits are scattered one per present row. flushBool masks stray bits past the batch's last row. A page that claims more values than it holds fails.

  • Checked against a scalar reference, at every batch size, with nulls and across pages (ColumnChunkDecoderTest).
  • Checked against readInt at random bit offsets (RleBitPackingReaderTest.readBitsMatchesReadIntAtWidthOne).
  • Cross-checked against parquet-java's BooleanPlainValuesWriter/Reader and RunLengthBitPackingHybridValuesWriter/Reader (DeltaBinaryPackedCrossCheckSuite).
  • Round-tripped through Spark with v1 and v2 pages (VectorParquetScanSuite).
  • Covered by the corpus file rle_boolean_encoding.parquet.

  JMH (V2EncodingsBenchmark, x86, one 20,000-value page in 1024-row batches):

  page	ours (ops/ms)	parquet-java (ops/ms)
  PLAIN, random values	837	10.7 (BooleanPlainValuesReader)
  RLE, runs of 1–64	212	11.2 (RunLengthBitPackingHybridValuesReader)

  RLE measured 60.7 before readBits, which expands each value to an int and repacks it.

• TINYINT / SMALLINT. Done. INT32 physical (INTEGER(8|16, true)), decoded by the INT32 path (every encoding it has) into the INT32 lane, with the declared type on output (VectorNarrowIntColumnVector). A writer may store any INT32 under the annotation. Spark's readers narrow with a (byte) / (short) cast, so fillFixed sign-extends the low 8 or 16 bits of each slot in place. Operators compute on the int, and a value outside the range must wrap there exactly as Spark wraps it. This is one pass over the batch's ints; there is no new decode kernel, so no new JMH.

  • Checked through Spark with dictionary, PLAIN and v2 DELTA_BINARY_PACKED pages, nulls, the extremes, filters, an aggregate, arithmetic and casts (VectorParquetScanSuite).
  • Checked on a file written by parquet-java's ExampleParquetWriter with out-of-range values, against Spark's parquet-java-based reader. Without the narrowing, that test's sum differs.
  • The corpus columns tinyint_col / smallint_col of the alltypes_* files are now read natively.

• FLOAT, BINARY and TIMESTAMP_NTZ. Done, as option A of the design note: decoded by the scan, no engine lane. Each is decoded on the lane of the same layout:

  • FLOAT's 4-byte IEEE bits on INT32: PLAIN, dictionary, BYTE_STREAM_SPLIT;
  • BINARY on the UTF8 path: PLAIN, dictionary, DLBA, DBA, and FIXED_LEN_BYTE_ARRAY read as binary;
  • TIMESTAMP_NTZ on INT64: micros, MILLIS scaled with Spark's overflow check, only from a column not adjusted to UTC; a UTC-adjusted column falls over to Spark's reader, which refuses it.

  They are emitted as Arrow Float4Vector / VarBinaryVector / TimeStampMicroVector, which Spark's ArrowColumnVector reads. FLOAT's 4-byte bits come out in a Float4Vector that operators widen onto the FLOAT64 lane it has had since #712. BINARY has had the UTF8 lane since #705, and TIMESTAMP_NTZ the INT64 lane: operators compare, hash, sort and carry them, and the expressions that read characters or the session time zone accept only string / timestamp, so over these they stay Spark's.

  Filters and projections carry them as foreign columns; an operator that computes on one stays Spark's, reading our batches as it reads Spark's scan. A table with such a column no longer keeps all its other columns on Spark's reader.

  • Tests:
    • DeltaBinaryPackedCrossCheckSuite: FLOAT PLAIN and BYTE_STREAM_SPLIT from parquet-java's writers, checked bit for bit against its readers (NaN payloads, -0.0, infinities).
    • VectorParquetScanSuite: round trips for all three (v1 dictionary, v1 PLAIN with MILLIS, v2), and filters, an aggregate and predicates on them. Also the UTC-adjusted TIMESTAMP_NTZ fallback.
    • The corpus columns float_col, date_string_col, string_col, f32, float_plain, float_byte_stream_split, flba5_plain and binary_field are now read natively.
  • JMH: no new decode kernel (the INT32, UTF8 and INT64 paths are the ones measured above), so there is no new benchmark.
  • Option B (real lanes, so operators compute on these types): done -- float rides the FLOAT64 lane (#712), binary the UTF8 lane (#705), and time and the ANSI intervals their INT64 / INT32 lanes (#812, below); variant stays without a lane by design (the types table above).
  • Time and ANSI intervals on lanes (#812). TypeMapping.isCarriedLong / isComputeRestricted: time and interval day to second are longs and interval year to month an int in every Spark vector and row, so the kernels carry, compare, hash, sort, group and join them as Spark does, and shuffle and spill buckets hash them as Spark's hash partitioning does (a long, an int). The expression compiler admits over them only references, comparisons, null tests and the conditionals that pick a value; the aggregates only count, min, max, first and last. Arithmetic (an interval's + fails on overflow in Spark; time takes + interval through TimeAddInterval), casts, functions, sum and avg stay Spark's. Spark 4.1's RowToColumnConverter has no time converter, so a row input holding a time keeps the operator above it Spark's. Tests: VectorTimeIntervalSuite (filters, projections, grouping with min / max / count, top-k, a join on a time and an interval, windows -- each asserting our operator -- and arithmetic, casts, functions, sum and avg with Spark's answer).

• TIME (#787): done. Spark 4.1's time is nanoseconds of the day; Spark writes it as INT64 TIME(MICROS). physicalMatches admits it (and, #812, any INT64, as Spark does), and the INT64 decode path (every encoding it has) fills the INT64 lane, scaled by 1000 with Math.multiplyExact as Spark's LongAsNanosUpdater (DateTimeUtils.microsToNanos) does, on the present rows only. The batch is an Arrow TimeNanoVector, which Spark's ArrowColumnVector reads (TimeNanoAccessor); no engine lane holds it, so operators carry it and one that computes on it stays Spark's. A TIME(NANOS) or un-annotated INT64 file is read the same way, as Spark's LongAsNanosUpdater reads any INT64 (#812); an INT32 TIME(MILLIS) file, which Spark refuses, goes to Spark's reader for Spark's error. Flat, in lists and structs, and as a partition column (a constant column).

  • Tests (NativeParquetTimeSuite), against Spark's reader: v1 dictionary, v1 PLAIN and v2 pages, two batch sizes, nulls, values across the whole day with fractional micros, flat, in an array and in a struct, filters and aggregates over it, a partition column, and the per-file unit match.
  • JMH: no new decode kernel (the INT64 path measured above, plus the scaling pass TIMESTAMP(MILLIS) already runs).

• VARIANT (#787, #750): done for an unshredded variant, flat, in lists and in structs. Spark writes a variant as a group of two required binaries, value and metadata. The readers decode the schema with every variant replaced by the struct <value: binary, metadata: binary> (NativeParquetSupport.variantAsStruct); clipping by name puts the file's group in that order whatever order it is stored in, and the struct, list and tree readers above read it like any struct of binaries, every binary encoding and the dictionary path included. The struct column the scan emits is Spark's variant column: ColumnVector.getVariant reads child 0 as the value and child 1 as the metadata. Spark's fallback reader keeps the requested schema.

  • Shredded variants (#812): done for a top-level variant. A typed_value beside the group's metadata and value makes the file's read schema take that variant as the group's own struct (Spark's SparkShreddingUtils.parquetTypeToSparkType), so the struct, list and tree readers decode every typed column and the residual binary natively; NativeParquetVariantColumn then rebuilds each row's variant with Spark's SparkShreddingUtils.assembleVariant, as Spark's assembleVariantBatch does, into the struct of its two binaries. A shredded variant below the top level (in a struct, a list or a map) keeps the file on Spark's reader.
  • Pushed extraction (#812): done for a top-level column. With spark.sql.variant.pushVariantIntoScan, Spark asks the scan for a struct of the extracted paths (VariantMetadata). The scan reads the variant's file group as its struct, shredded or not, and builds each row's struct with Spark's assembleVariantStruct and the paths' casts (getFieldsToExtract), as Spark's assembleVariantStructBatch does. The assembly allocates per row, as Spark's does. A pushed struct below the top level keeps Spark's scan for the plan.
  • Tests (NativeParquetVariantSuite), against Spark's reader: objects, arrays, scalars, JSON and SQL nulls; v1 dictionary, v1 PLAIN and v2 pages, two batch sizes; flat, in an array and in a struct; to_json, variant_get, schema_of_variant, is_variant_null and aggregates; the rows themselves compared with Spark's; a shredded file (decoded natively, #812), and pushed-down extraction over shredded and unshredded files (native, #812).
  • JMH: no new decode kernel; the binaries take the UTF8 path measured above.

• TIMESTAMP. Done for INT64 MICROS and MILLIS, into the INT64 lane of micros. physicalMatches admits the TIMESTAMP annotation with either unit and either isAdjustedToUTC, the same match as Spark's ParquetVectorUpdaterFactory. MILLIS is scaled by 1000 with Math.multiplyExact, as Spark's LongAsMicrosUpdater does, so an overflowing value fails the read instead of wrapping. Only present rows are scaled, because a null slot holds whatever the reused buffer held.

  • Rebase, per file (#568 item 6). At open, fallbackModes resolves the file's datetime and INT96 rebase specs with DataSourceUtils.datetimeRebaseSpec / int96RebaseSpec, from the writer version and legacy keys in the footer, or else the session configuration, and the INT96 time-zone conversion the way ParquetFileFormat does. Every column reader gets them as one ParquetRebase, which runs over each decoded date or timestamp batch as Spark's ParquetVectorUpdaterFactory runs per value: a value below the last Julian switch (RebaseDateTime.lastSwitchJulianDay / lastSwitchJulianTs) fails the read under EXCEPTION with DataSourceUtils.newRebaseExceptionInRead and is rebased from the hybrid calendar under LEGACY (in the file's zone); CORRECTED leaves it. Above the switch both rebases are the identity, so modern data costs one comparison per value and no write. Row-group pushdown builds ParquetFilters with the file's own rebase spec, as Spark does.
  • INT96 (Spark's default output timestamp type, and Hive's and Impala's): done. The 12-byte values (nanoseconds in the day, then the Julian day, little-endian) are staged on the fixed-length binary path, PLAIN and dictionary, and ColumnChunkDecoder.flushInt96 converts them to micros exactly as Spark's binaryToSQLTimestamp / fromJulianDay do; the file's INT96 rebase mode and, for a file not written by parquet-mr under int96TimestampConversion, the conversion from the session zone to UTC follow.
  • The fallback reader is opened as ParquetFileFormat opens Spark's vectorized reader: over the split's byte range, with the scan's Hadoop conf carrying setupHadoopConf's flags (INT96 as timestamp, binary as string, the session time zone, ...), and with the file's own rebase modes and INT96 conversion. The earlier initialize(path, columns) read the whole file on the first split, turned int96AsTimestamp off, and was hard-wired to CORRECTED. A LEGACY file would have been read without its rebase.
  • Checked through Spark with TIMESTAMP_MICROS and TIMESTAMP_MILLIS output, v1 and v2 pages, nulls, values before 1582 and after 2038, filters and an aggregate, all natively.
  • Checked across one table holding an INT96 file, a LEGACY-calendar file and a CORRECTED file, each decoded natively with Spark's values, under CORRECTED and EXCEPTION. NativeParquetRebaseSuite adds INT96 over v1 and v2 pages with and without dictionaries, inside a struct and a list, parquet-java's and parquet-testing's int96_from_spark, the time-zone conversion in two zones, LEGACY dates and timestamps on both sides of the switch with a pushed-down filter, and EXCEPTION on files without Spark's metadata: native when no value needs a rebase, Spark's own error when one does.
  • Checked on a parquet-java-written MILLIS file whose micros overflow. It fails with ArithmeticException under both readers.

• Wide decimals and decimals stored as bytes. Done. A decimal stored as FIXED_LEN_BYTE_ARRAY or BINARY is decoded by the binary page paths: PLAIN (fixed-length values have no length prefix), dictionary, DELTA_LENGTH_BYTE_ARRAY and DELTA_BYTE_ARRAY, which is what parquet-java's v2 writer uses for FLBA. Its big-endian two's-complement bytes are then converted at flush (ColumnChunkDecoder.flushDecimal):

  • decimal(p > 18) goes into the DECIMAL128 lane: two little-endian limbs, sign-extended, which is Arrow's Decimal128 layout. A 16-byte value is read as two big-endian longs. A value longer than 16 bytes must be a sign extension, or the read fails.
  • decimal(p <= 18) stored as bytes goes into the INT64 lane, exactly as Spark's binaryToUnscaledLong computes it. Spark's legacy writer, Hive and Impala store such decimals; since #592 those files had fallen over to Spark's reader.
  • Scale check. physicalMatches now also requires the file's decimal annotation to have the requested scale and at most the requested precision. Spark reads, for example, a decimal(9,2) file as decimal(12,4) by rescaling, and the lanes carry the stored unscaled values as they are, so such a file fell over to Spark's reader until #568 schema evolution rescaled it natively (below). Before this check, an INT64 (or narrow INT32) decimal read with another scale was decoded natively without the rescale: wrong values.
  • Tests.
    • ColumnChunkDecoderTest: every width from 1 to 16, FLBA and BINARY (with sign-extended values longer than 16 bytes), both lanes, against BigInteger and Spark's formula.
    • DeltaBinaryPackedCrossCheckSuite: parquet-java's FixedLenByteArrayPlainValuesWriter and DeltaByteArrayWriter, read back by its matching readers.
    • VectorParquetScanSuite: decimal(20|25|38) round trips (v1 dictionary, PLAIN, v2 DELTA_BYTE_ARRAY, legacy format) and the rescale fallback.
    • The corpus files fixed_length_decimal* and byte_array_decimal are now read natively.
  • JMH (V2EncodingsBenchmark flba-dec38, x86, one 20,000-value page): 13.5 pages per ms. parquet-java's FixedLenByteArrayPlainValuesReader.readBytes does 13.7, but it only slices the bytes and does no conversion. Before the big-endian long reads this path measured 3.4.

• Structs (#568): done, structs of structs included, over every leaf type and encoding above. A struct has no column chunk of its own; Parquet records its nulls in its leaves' definition levels. NativeParquetStructReader holds one reader per field (a leaf's NativeParquetColumnReader, or another struct reader), and each leaf decodes its rows straight into its own batch vector, as a flat column does. The struct is present on a row where its first leaf's level reaches the struct's own definition level (the optional groups from the root down to it), so ColumnChunkDecoder keeps that one leaf's batch levels (setRecordLevels) and the struct reader turns them into the struct's nulls in one pass. That is how Spark's vectorized reader computes them too. The batch is Spark's struct column over the fields' columns (VectorNestedColumnVector.struct), which our operators read through getStruct / getChild.

  • Per file, forEachLeaf walks the clipped group in the struct's field order and checks each leaf's physical type against its lane (physicalMatches). Fields are matched by position, as the clipped schema keeps the requested order (under field-id matching a field keeps the file's name). A shape it cannot follow (a repeated group where a struct is requested, say) falls the file over. A field the file does not hold is null (schema evolution, below). Dates and timestamps under a struct count for the calendar-rebase check, and an INT96 leaf falls the file over as at top level.
  • Strings under a struct are dictionary ids too (#612; since #568), at any struct depth, where the leaf allows it, as at the top level. A struct field is read zero-copy by the operators: ColumnVectorAdapters looks under the null-safe wrapper the nested vectors put on their leaves (VectorNestedColumnVector.unwrapLeaf) and reads the leaf's own buffers (an Arrow vector, a dictionary vector) instead of copying it through Spark's getters. The strings of lists and maps (elements, keys, values, and the fields of their structs, at any depth) are dictionary ids too, the same way.
  • Tests (NativeParquetStructSuite), each against Spark's row-based reader, with the native scan asserted to decode the row groups itself:
    • Spark-written files: every lane type as a field, nulls at the struct and at the leaf, a struct of structs, v1 and v2 pages, dictionary on and off. Nested schema pruning, filters and aggregates on fields, a shuffle.
    • parquet-java's ExampleParquetWriter: required groups, required leaves under optional groups, and an optional group under a required one, which Spark does not write. v1 and v2, dictionary on and off.
    • Hardwood's NestedSchemaTest struct-of-struct shape, and the apache/parquet-testing deep struct from arrow-rs (required groups of required leaves; its unsigned INT64 fields are a decimal on Spark's side, so selecting them keeps that file on Spark's reader).
    • A struct field the file lacks is null, natively since #568 schema evolution (below); an INT96 timestamp in a struct is decoded natively since #568 item 6. A struct holding a list or a map is read natively (below). Field names resolve case-insensitively.
  • JMH: no new decode kernel. The leaves are the flat decoders measured above, and the struct adds one pass over one leaf's levels per batch.

• Lists and maps (#568): done for a top-level array<flat> or map<flat, flat>, every flat type and encoding above. Each value slot of a list's leaf is an entry with a repetition and a definition level, and repetition level 0 starts a row.

  • Levels. On the leaf's path there is exactly one repeated node, in every list form Spark reads as an array of a flat type: the three-level LIST, Spark's legacy bag / array form, the two-level form, and a bare repeated leaf. Below the level of that node's parent, the row's list is null. Below the repeated node's own level it is empty. At or above it the entry is an element, and the element is null below the leaf's maximum level.
  • The leaf in repeated mode (NativeParquetColumnReader). It reads pages ahead of the decoder, into their own buffers, because a batch's rows may run past the page in hand. Each page's repetition and definition levels are kept, and the page past its repetition levels goes to the decoder (Page.v1At / v2At). entriesForRows(n) counts the entries of the next n rows.
  • Decoding. readElements walks those entries in runs: element runs decode back to back into the batch vector, and the entries of null and empty lists are skipped (ColumnChunkDecoder.skip, which consumes no value for them).
  • The batch. NativeParquetListReader builds the offsets and the null rows from the first leaf's levels. A map's key and value are two leaves under one key_value group, so each decodes its own elements. The batch is Spark's list or map column over the leaves' vectors (VectorNestedColumnVector.list / map).
  • Per file, forEachLeaf requires one leaf for a list, or two for a map (key first, under the same repeated group). Each leaf must be repeated exactly once and stored as its lane needs; anything else falls the file over.
  • Tests (NativeParquetListSuite), each against Spark's row-based reader, with the native scan asserted to decode the row groups itself:
    • Spark-written files, three-level and legacy form, v1 and v2 pages, dictionary on and off. Lists of every lane type with null lists, empty lists and null elements, a list longer than many pages, and maps with null values and empty maps. Queries cover size, element access, exists / filter / transform, map_keys / map_values, explode of a list and of a map, aggregates, and a shuffle.
    • parquet-java's ExampleParquetWriter, v1 and v2, dictionary on and off: the two-level list, required and optional three-level elements, a bare repeated double, and a map with optional values.
    • Hardwood's dictionary-encoded list spanning batches.
    • The apache/parquet-testing files. list_columns, null_list and the top-level repeated primitive of repeated_primitive_no_list are read natively. The nested maps, incorrect_map_schema and map_no_value answer what Spark answers; the nested lists and the legacy LIST<LIST<int32>> are read natively (lists of lists, below).
    • Fallbacks: none left for nesting itself; a nested column keeps Spark's scan only for an unreadable leaf type.
  • JMH: no new decode kernel. The leaf decoders are the ones measured above; the list adds one pass over a leaf's levels per batch, plus the page buffers.

• Lists of structs (#568): done for an array of structs of flat fields, in Spark's three-level form, the legacy two-level form (the repeated group is the struct) and parquet-java's. Every field is a leaf under the same struct group, repeated once, so all share one entry stream: the offsets and null lists come from the first field's levels as for a list of leaves, an element's struct is null where that field's level is below the struct group's, and each field decodes its own elements (readElements). The batch is Spark's list column over a struct column over the fields' vectors. Tests: Spark-written files (legacy and three-level form, v1 and v2 pages, dictionary on and off; nulls at the list, the struct and the field; transform, filter, exists, explode, aggregates) and parquet-java's three-level and two-level shapes, v1 and v2, each against Spark's row-based reader.

• Lists of lists (#568): done, any depth, over one flat leaf (NativeParquetNestedListReader). The leaf is repeated as deep as the arrays nest; the j-th repeated node on its path gives two thresholds, where the depth-j list is present (its parent's definition level) and where an entry is one of its elements (the node's own). An entry opens a depth-j list where its repetition level is at most j - 1 and the list exists there, and a depth-j element where its repetition level is at most j and its definition level reaches the element threshold; one pass over the leaf's levels builds every depth's offsets and nulls, and the leaf decodes its elements back to back. Spark's three-level LIST of LIST, the legacy two-level form and parquet-java's are all read. Tests: Spark-written lists two and three deep (legacy and three-level form, v1 and v2 pages, dictionary on and off; nulls and empty lists at every depth; flatten, nested exists, explode), parquet-java's three-level and two-level shapes, and parquet-testing's nested_lists and old_list_structure, which now read natively.

• Lists and maps inside structs (#568): done. A struct's field may be any list or map the list readers follow (lists and maps of flat types, lists of structs of flat types, lists of lists); each is read at its path under the struct. The struct's nulls come from its first field, whatever it is: a flat leaf's batch levels as before, or for a list or map the level of each row's first entry (RowLevels), since every entry of a row carries the struct's own definition. Tests: Spark-written structs whose first field is a list, with a leaf and a map after it, and a struct of a struct of a list of lists (legacy and three-level form, v1 and v2 pages, dictionary on and off; nulls at the struct, the list, the map and the leaves), and parquet-java's struct holding a three-level LIST, a legacy two-level list and a MAP, v1 and v2, each against Spark's row-based reader.

• Maps of structs (#568): done for a map whose value is a struct of flat fields, at the top or as a struct's field. The key and every value field are leaves under the same key_value group, repeated once, so they share one entry stream: the offsets and null maps come from the key's levels, as for a map of leaves; a value struct is null where its first field's level is below the value group's (the key is required, so its own levels cannot tell); and each leaf decodes its own elements. The batch is Spark's map column over the key's vector and a struct column over the value fields' vectors. Tests: Spark-written maps of structs, at the top and inside a struct (v1 and v2 pages, dictionary on and off; nulls at the map, the value struct and its fields; element access, map_keys, explode, aggregates), and parquet-java's MAP with an optional value group, v1 and v2, each against Spark's row-based reader.

• Any nesting (#568): done. A list or map whose nesting the readers above do not follow (a list of structs holding a list or a map, a map of structs holding a list, a list of lists of structs, a list of maps, a map of lists, a map of maps) is read by NativeParquetTreeReader. It builds a node tree from the clipped schema and the Spark type (Spark's three-level LIST, the legacy two-level forms by Spark's isElementType rule, bare repeated fields, MAP groups), and every node takes its structure from the levels of the first leaf below it, which encodes it as every other leaf there does: a node inside j lists exists at an entry with repetition level at most j whose definition level reaches the innermost enclosing list's element level; a list or map at depth j + 1 is null below its own level and has an element at an entry with repetition level at most j + 1 reaching its element level; a struct is null below its own level; and each leaf decodes its elements back to back. Tests: Spark-written lists of structs holding a list and a map, lists of lists of structs, maps of structs holding a list, maps of lists and lists of maps (legacy and three-level form, v1 and v2 pages, dictionary on and off; nulls and empty collections at every level), parquet-java's three-level list of structs holding a list and the legacy two-level group with a repeated field, and parquet-testing's nested_maps, each against Spark's row-based reader.

• Metadata columns (#568): done. The file-constant _metadata fields are constant columns after the partition columns, filled per file by FileFormat.updateMetadataInternalRow with the format's extractors, on native and fallback files alike (a fallback file's batch gets them appended, as FileScanRDD does). _metadata.row_index is NativeParquetRowIndexColumn: the footer's first row index of the group plus the position, or, in a row group filtered by the column index, plus the selected row's index in the group. The reader sets InputFileBlockHolder per file on the task thread for input_file_name(), input_file_block_start() and input_file_block_length(). Under decode-ahead the files are opened on the producer thread, whose holder is the task thread's own (it is inheritable), so the producer sets nothing: each batch carries its file, and DecodeAheadIterator sets the holder as the task thread takes a batch from another file. Before this, a query reading _metadata failed in the native scan and input_file_name() did not return Spark's value. Tests: NativeParquetMetadataSuite, against Spark's row-based reader, on Spark- and parquet-java-written files, partitioned tables, split files, a decode filter and an INT96 file next to native ones.

• Schema evolution (#568): done: missing columns and fields (at any depth, inside lists and maps too), type widening and field-id matching. A table read with a schema newer than some of its files:

  • Missing columns and struct fields are null in every row, as Spark's ParquetColumnVector makes them (setMissing). A top-level column the file lacks (flat or nested) is NativeParquetNullColumn. A struct field the file lacks, at any struct depth, is one too, and the struct's nulls come from its first field the file holds. missingReadable admits a missing node that is a top-level column or a field of a struct at any depth, optional, and (top level) without an existence default. Inside a list or a map, NativeParquetTreeReader takes the column: a missing field is a Missing node, an all-null Spark vector as long as its struct's instances, and the struct's structure comes from its first field the file holds. A top-level column added with a DEFAULT holds its existence default in every row of a file without it, a constant vector as Spark fills it (#812; a type with an engine lane, a string or a binary). A struct the file holds none of the requested fields of takes its nulls from the field Spark 4.1's clipping appends for them, the file's cheapest leaf, which the struct reader reads as its level source and does not emit (#812). A REQUIRED missing column stays on Spark's reader, which fails the read with its own error.

  • Type widening, as Spark's ParquetVectorUpdaterFactory widens. physicalMatches admits each pairing below, and NativeParquetColumnReader.widenLane converts each decoded batch in place: INT32 and FLOAT values are decoded sign-extended into INT64 slots (the kernel's existing INT32-to-INT64 path, every encoding).

    • INT32 and FLOAT as double.
    • DATE as timestamp_ntz: midnight UTC, after the file's date rebase (ParquetRebase.daysToMicros).
    • A decimal (p, s) read as (p', s') with s' >= s and p' - p >= s' - s: the unscaled values times 10^(s' - s), in the INT64 lane by a long multiply and in the DECIMAL128 lane through BigInteger (Spark rescales through BigDecimal per value).
    • An un-annotated or signed-integer INT32 as a decimal with at least 10 integer digits.
    • An INT32 or INT64 read as a decimal of more than 18 digits: each value is decoded into an 8-byte slot and sign-extended to a 16-byte DECIMAL128 slot in place (WIDEN_LONG_TO_DEC128; zero-extended for a UINT_64, WIDEN_ULONG_TO_DEC128), rescaled when the requested scale is larger, as Spark's IntegerToBinaryUpdater / LongToBinaryUpdater (and their ToDecimal forms) widen it. Spark reads an un-annotated integer as a decimal only when the requested precision leaves room for its integer digits (INT32 at least 10, INT64 at least 20). NativeParquetSchemaEvolutionSuite covers decimal(20,2) and decimal(38,0) from INT32 and decimal(30,2) / decimal(38,0) from INT64.

  • Field-id matching (spark.sql.parquet.fieldId.read.enabled). clipParquetSchema matches by id and keeps the requested order, under the file's names, with a generated name for an id the file lacks (a missing column: null). So every reader pairs the requested fields with the clipped ones by position, not by name: clippedName at the top level, and the struct, list and tree readers' fields and leaves. Row-group pushdown is built over the file's schema, as ParquetFileFormat builds it. Two cases go to Spark's reader for Spark's outcome:

    • a file with no ids at all when the read schema has some, unless fieldId.read.ignoreMissing is set (Spark's ParquetReadSupport fails it);
    • a field matched by id that carries a sibling's requested name (idNameClash). Spark's reader pairs the clipped fields with the read schema by name after the id match, so it reads such a field as the sibling: a type mismatch, or another column's values.

    A read error from Spark's reader inside the native scan carries the file, as FileScanRDD attaches it (FileDataSourceV2.attachFilePath).

  • Tests: NativeParquetFieldIdSuite, against Spark's vectorized reader, v1 and v2 pages × dictionary on and off. Columns and fields renamed and reordered at the top level, in a struct, in a list of structs and in a map's struct values, with ids the file lacks; a filter on id-matched columns; swapped names (Spark's outcome); a file with no ids (Spark's error, and nulls under ignoreMissing); and struct fields missing inside a list, a map and a list of lists of structs.

  • Tests: NativeParquetSchemaEvolutionSuite, against Spark's vectorized reader, v1 and v2 pages × dictionary on and off. Missing flat, struct, list, map and wide-decimal columns; missing struct fields at two depths, including the first field; widening flat, in a struct and in a list; legacy-format FLBA decimals; a LEGACY-calendar date read as timestamp_ntz; parquet-java INT32, INTEGER(16) and FLOAT files; and the fallbacks above. VectorParquetScanSuite's rescale test and NativeParquetStructSuite's missing-field test now assert native reads.

# Performance

• Nested readers, JMH (#568, NestedParquetReadBenchmark): one Spark-written 1M-row snappy file, one nested column at a time, read whole and folded into a checksum through Spark's ColumnVector getters (the way an operator above the scan reads it). Ours is compared with Spark's VectorizedParquetRecordReader; both agree on the checksum. x86, -wi 2 -i 3 -f 1, ms per read:

  column	reader	ours	Spark
  array<int>	list	79.2 ± 9.1	88.1 ± 4.4
  array<string>	list	114.9 ± 2.4	117.9 ± 3.1
  struct<int, string>	struct	62.1 ± 5.0	59.9 ± 4.1
  map<string, int>	list (map)	113.7 ± 5.4	98.5 ± 9.5
  array<array<int>>	nested list	200.8 ± 16.0	230.6 ± 8.8
  array<struct<int, array<string>>>	tree	250.5 ± 8.6	250.6 ± 2.5

  At parity or ahead everywhere but the map (15% behind at the time of this table; since #812 the map reads faster than Spark's reader, below). The folding through the getters is a large share of both columns. Run with benchmarks/scripts/run-parquet-e2e.sh's classpath and JVM options.

  #812, the getters. A profile of the map read put a quarter of its time in Arrow's bounds-checked isNull, reached once per element through Spark's ArrowColumnVector accessors, and getInt and getUTF8String check the bit again inside Arrow. VectorArrowColumnVector now reads the validity bit, INT32 and INT64 values and string offsets and bytes straight from the buffers' addresses, as Spark's own vectors read theirs, with no allocation; it reads each buffer's address on every call, so a vector refilled between batches is read where it lives now (VectorArrowColumnVectorSuite). Same run settings, -wi 3 -i 6, ms per read, ours before / after, then Spark in the same session as after:

  column	ours, main	ours, #812	Spark
  map<string, int>	108.4 ± 1.6	92.6 ± 1.1	111.6 ± 2.0
  array<string>	112.6 ± 5.1	101.6 ± 4.3	111.0 ± 3.7
  array<int>		66.5 ± 1.2	86.7 ± 1.5
  struct<int, string>		49.5 ± 0.5	56.3 ± 0.7

  The map now reads 17% faster than Spark's reader. Since every Arrow-backed column a Spark operator reads goes through this wrapper, the gain is not specific to nested columns.

  oursIds reads the same columns with strings as dictionary ids, as the scan does by default (#612; in structs, lists and maps since #568). Same run settings, ms per read, ours / oursIds: array<string> 115.7 / 103.6, struct<int, string> 62.3 / 57.3, map<string, int> 112.6 / 109.0 (± 38), array<struct<int, array<string>>> 246.0 / 238.6. The read itself gains up to 10%; an operator that groups, joins or compares the strings by id gains more, which this benchmark does not measure.

• Column-index page filtering (#568, #626): done, for every column the scan reads natively. useColumnIndexFilter is off by default (#873) and on with parquet.filter.columnindex.enabled=true (a session key, or spark.hadoop.parquet.filter.columnindex.enabled). On the classic S3A stream every selected page is a separate seek and GET, plus the column and offset index reads: at 1 TB TPC-DS on 9 x m5.4xlarge, q75, q95, q85, q80 and q44 took 100.2 s with it on and 73.2 s off (75.4 s before #748 added it), so it waits for page ranges to be merged into larger reads (#834). Row-group statistics, dictionary and bloom-filter pruning stay on. With it on: readNextFilteredRowGroup then keeps only the pages overlapping the filter's row ranges, and the store carries the selected row indexes. NativeParquetColumnReader turns them into ranges and follows them: it moves the decoder to each range's first row, skipping rows within a page and over the pages Parquet dropped (each page's first row index gives the gap), then decodes the range. So every column emits the same selected rows although their pages start at different rows. A list or map counts entries, not rows (#626). A page of a file with an offset index begins on a row boundary, as the format requires, so each repeated leaf keeps every pulled page's first entry and first row index: before a range it skips the entries of the rows ahead of it in the page in hand and jumps over the pages Parquet dropped, then counts and decodes the range's entries as before. The list, map, struct and tree readers take their structure from those entries unchanged. A leaf skips only at a batch's start, so in a file that requests a repeated leaf a batch ends where a range ends. Before #568 the column index was off: with it, the columns' pages misaligned.

  • Pages skipped (#626): numPagesSkipped counts the requested pages the column index dropped: per row group read, the pages in each requested column's offset index that overlap none of the selected ranges. It is computed on the thread that read the row group, from the offset indexes Parquet's filter had already read, so it adds no I/O.
  • Late materialization's I/O (#611): the decode filter does not change which pages are read. readNextFilteredRowGroup fetches every requested column's pages that overlap the column index's ranges before any row is decoded, so the non-filter columns of a row group are read in full over those ranges, even where the decode filter then drops every row. Skipping them would mean reading a row group in two passes: the filter columns first, then the others through readFilteredRowGroup(index, RowRanges) over the surviving rows. That stays open until the measurements say it pays.
  • Tests: NativeParquetColumnIndexSuite checks against Spark's own scan, over sorted data with small pages. It covers every flat type and a struct, and a list and a map of flat types, a list of structs, a list of lists, a struct holding a list and a map, a map of structs and a list of structs holding a list (#626). Each is run with v1 and v2 pages, dictionary on and off, one and several ranges, a filter no page matches, _metadata.row_index, late materialization, small batches and the switch off. The native scan must output under a quarter of the rows it reads with the column index off, and must count skipped pages.

• Make the sort compare robust against C2 uncommon-trap storms. When q67 runs after q88, q28, q9 and q44 in the same app, 0–3 of 8 executors run q67's sort stage at about 2.3× CPU per task for the rest of the app.

  • The deoptimization log shows hundreds of thousands of unstable_if traps with action none in StringCompareKernels.compareBytes at bytecode 33, where the 8-byte loop runs out. The profile never saw that exit taken, and once HotSpot stops recompiling, the trap stays.
  • Ruled out: a separate copy of the method for sorts, and -XX:-OptimizeUnstableIf.
  • Being validated: warming the method's branch profile at class initialization.

  This is not specific to the native scan. It also happens with the flag off, less often.

• Decode in parallel (#606): done, off by default. Decode-ahead (above) runs decoding on ONE producer thread per task. spark.vecruntime.scan.nativeParquet.parallelDecode = K (default 0, off) adds a second step: the row groups AFTER the one the task thread consumes are decoded by K HELPER threads, each on its own column-reader set and child allocator, and handed back in file order so the output is identical (ParallelDecodeIterator). It engages per file, only for a file with at least two row groups at or above …parallelDecode.minRowGroupBytes (default 32 MiB) and only on the plain whole-row-group path (no decode filter, no two-pass): a file outside that — the TPC-DS store_sales shape, ~7 MB single-row-group files — decodes single-producer as before. The lead reader keeps all of file iteration, read-ahead (which has already fetched a row group's compressed pages when the job is handed out), per-file fallback, metrics and order; a helper only decompresses and decodes a fetched row group into vectors it owns. Threads are virtual by default (…parallelDecode.threads), with the JDK 25 rules the decode-ahead producer follows (the first job of a task decodes on the task thread, so the decode classes are initialized before any helper runs; file opens stay on the platform prefetch pool). Off by default for the same reason decode-ahead is modest: a cluster running as many tasks as cores already keeps every core busy, so a second decode thread per task only contends there — the 1 TB TPC-DS A/B measured +1-5% for decode-ahead, and parallel decode adds more helpers to the same contention. The win is where cores sit idle: few large files with many large row groups, and stage tails. A local A/B on one large file with idle cores (the target shape): a scan-heavy aggregate over a 3M-row mixed-type file of two 64 MB row groups on one task ran 436 ms serial, 421 ms decode-ahead=4, and 333 ms / 337 ms at parallelDecode=2 / 4 (−24% / −23%); over five 16 MB row groups the gap narrows (297 ms serial, 259 ms decode-ahead, 262 ms parallelDecode=4) as the fixed per-row-group reader-set setup dilutes a smaller group's decode. The 1 TB cluster A/B over this query set is the next step before changing the default. Tests: ParallelDecodeSuite (rows equal the serial path for scans, filters, group-bys and nested / decimal / string columns at K = 1, 2, 4 on virtual and platform helpers; order preserved row by row; a LIMIT and a helper failure leave the allocator at baseline; a mixed native / fallback table; the minRowGroupBytes gate and a decode filter keep a file on the lead; the column index on), and IcebergNativeScanSuite (identical over positional, deletion-vector and equality deletes through the Iceberg native scan).

• Cap the read-ahead in bytes (#568): done. Besides the depths (prefetchFiles, prefetchRowGroups), spark.vecruntime.scan.nativeParquet.prefetchMaxBytes (default 256 MiB) caps the compressed bytes one task's read-ahead holds and prefetchExecutorMaxBytes (default 2 GiB) caps every task of the executor together; 0 lifts either. A file opened ahead counts at its split's length and a row group read ahead at its requested column chunks' size; each is given back when it is used or dropped. One file and one row group ahead are always allowed, so the overlap never stops. At the TPC-DS shapes (files of ~7 MB, depth 12) neither cap binds. Tests: DecodeAheadSuite (Spark's rows under each cap, the peak bounded by one item ahead, every byte given back).

• Read-ahead in the task's memory budget (#915): done. The compressed pages read ahead and the decoded batches queued by decode-ahead are held in the task's Spark execution memory, granted only from memory that is free now: the pool must have the bytes, and in the operators' memory mode the task must stay within its share (pool / spark.executor.cores). A request that would make Spark spill another consumer of the task (a sort's runs, a join's build) is refused instead, and the window narrows: one file and one row group ahead stay allowed, and decode-ahead waits for its queue to empty before queuing the next batch. So a wide scan stage with free memory reads deep and a narrow stage whose operators hold the pool reads shallow; the byte caps above remain the upper clamp. Compressed pages are charged to the on-heap pool (parquet-mr reads column chunks into heap buffers), decoded batches to the mode Arrow memory is budgeted in. Executor metrics: scanReadAheadHeldBytes, scanReadAheadPeakBytes, scanReadAheadRefusals. Tests: ReadAheadMemorySuite (a grant held and given back; refused without spilling another consumer; the task's share), DecodeAheadSuite (every byte given back).

• Row filtering during decode (late materialization, #611). Done for flat columns: with spark.vecruntime.scan.nativeParquet.lateMaterialization on, the scan decodes the decode filter's columns first, then the other columns only at the rows that survive (readBatchSelected; a string column skips the others in its pages, a fixed-width one decodes whole and carries the selection). Since #568 a struct does the same field by field: each flat string leaf decodes the survivors only, and the struct's nulls come from its level source as before. A list, a map or the tree reader still decodes the batch whole (their leaves count entries, not rows), and the batch's selection excludes the rest. Tests: LateMaterializationSuite, including a struct of strings and a struct inside it at five selectivities. Still to measure against Comet with rowFilterPushdown on.

  • Two passes, skipping I/O (#812). One pass saves decode CPU only: parquet-java fetches every requested column's pages before a row is decoded. With spark.vecruntime.scan.nativeParquet.lateMaterialization.twoPass on, a row group is read twice. First only the decode filter's columns (readRowGroup), decoded and filtered batch by batch over the whole row group, recording the surviving rows as runs. Then every requested column over those runs only: readFilteredRowGroup(i, ranges) fetches the pages that hold a surviving row, and the column readers align to the ranges as they do for column-index filtering (#748). The filter is evaluated again over those rows by the one-pass path, so pass 1 only chooses the row set. The runs become a RowRanges through a RunsOffsetIndex whose "pages" are the runs. Pass 1 does not use readFilteredRowGroup: parquet-java caches a row group's column-index store with the paths requested when it is first built, which would leave pass 2 without the other columns' offset indexes. Read in one pass: a row group with a requested column chunk that has no offset index, one where the filter keeps more than twoPass.maxFraction (0.5) of the rows, and, after two of those in a row, the rest of the file; a row group with no survivor is skipped. Row-group read-ahead is off under two-pass (it reads by index on the task thread), and so is reading the first row group when the file is opened ahead.
    • Tests (TwoPassLateMaterializationSuite): Spark's rows for eleven filters (none, a narrow run pushed and not pushed to Parquet, a quarter, scattered, runs, most, all, nulls, strings, dates) on a file sorted by the filter column and an unsorted one, with decode-ahead and row-group read-ahead; and a selective unpushed filter on the sorted file reads under half the bytes of one pass, with no allocator leak.
    • Local measurement, 8M rows sorted by id in 4 files, SELECT sum, sum, sum, max(string), max(string) WHERE <filter>, median of 5 ms, one pass / two passes: id * 2 in 0.01% of the range 529 / 157; 1% 465 / 154; 10% 497 / 301; scattered 1% (k * 2 < 20) 457 / 409. Not yet run on the cluster.

• TLS and HTTP client CPU (about 46% of q88's executor CPU). Options: fewer and larger range reads (#834 — small-file prefetch and range merging are now in the scan itself, behind off-by-default confs; see How reads are issued — since the benchmarks no longer read through the Analytics Accelerator, #835); HTTP/1.1 connection reuse settings; and checking that the CRT-based client really uses ACCP's AES-GCM. The 1 TB A/B that sets smallFileThreshold and rangeMergeGap (and that could turn the column index back on, #873) is the remaining step.

• q88 iteration variance (38–110 s within one leg). Next step is a per-task timeline across iterations to tell stragglers apart from S3 throttling or connection churn.

• Vectored reads. parquet.hadoop.vectored.io.enabled=false was 3–8% faster in one A/B. Re-measure it with read-ahead on before changing any default.

# Correctness and operations

• Run the golden suite with the flag on in CI, not only in local gates.
• Run the read-ahead tests under an executor-loss and task-retry stress test, to show that files opened ahead are always closed.
• Add metrics for the number of files opened ahead, the time task threads waited on read-ahead, and the number of fallback files.