HIVE-30059: LLAP IO Cache: native parquet data cache

[abstractdog]

What changes were proposed in this pull request?

Native LLAP encoded-data path for Parquet, mirroring the existing ORC pipeline:

• ParquetColumnVectorProducer → ParquetEncodedDataReader → ParquetEncodedDataConsumer, feeding VectorizedColumnReaders through a shared ParquetRowGroupDecoder (extracted from VectorizedParquetRecordReader).
• Column chunks are held natively as MemoryBuffer[] in ParquetEncodedColumnBatch; ParquetCachedPageReadStore parses pages in place over those cached buffers with lazy per-page decompression — no ParquetFileReader, no copy from cache into a Parquet read buffer.
• Enabled by default via hive.llap.io.parquet.native.enable=true.

Design doc: HIVE-30059-design-v3.pdf.

Why are the changes needed?

Parquet under LLAP already caches byte-ranges via LlapCacheAwareFs, but reads still go through a stock ParquetFileReader over a virtual llapcache:// URL: cached bytes get copied into the reader's own buffers, page parsing and decompression run per-read, and the path isn't wired into LLAP's encoded-column pipeline or fragment counters. This PR brings Parquet to parity with ORC's native LLAP path: zero-copy page parsing over cached buffers, lazy decompression, full LLAP IO counter integration (METADATA_CACHE_HIT/MISS, SELECTED_ROWGROUPS, HDFS/IO time), and the shared ColumnVectorProducer/EncodedDataConsumer execution model.

Does this PR introduce any user-facing change?

Better Parquet read performance under LLAP (zero-copy page parsing over cached buffers, lazy decompression). Query summaries now report DATA_HIT / METADATA_CACHE_HIT (and their _MISS counterparts) for Parquet, so cache effectiveness is visible the same way it is for ORC. Opt-out via hive.llap.io.parquet.native.enable=false.

How was this patch tested?

• TestParquetEncodedDataReader (27 tests) — plan/fetch/assemble, cache hit/miss, footer caching, encryption fallback, schema evolution.
• TestParquetCacheLayout, TestParquetRangeBuffers (8 tests) — page-range math and buffer reuse.
• Existing TestVectorizedParquetRecordReader unchanged (shared ParquetRowGroupDecoder).
• Manual verification through miniHS2 with LLAP.

[Copilot]

🟡 Changes recommended

Unresolved critical and moderate findings affect schema projection, cache ownership, row lineage, configuration, and fallback behavior.

Get a fresh assessment by requesting another Copilot review.

Pull request overview

Adds opt-in native Parquet LLAP caching with vectored reads, cached-page decoding, fallback integration, and supporting tests.

Changes:

• Adds native Parquet LLAP configuration, APIs, and dispatch.
• Implements cache layouts, range buffers, vectored reads, and page decoding.
• Refactors shared Parquet reader logic and expands test coverage.

File summaries

File	Reviewed changes and final comments
ql/src/test/org/apache/hadoop/hive/llap/TestParquetRangeBuffers.java	Tests range-buffer reuse and slicing; no final comments.
ql/src/test/org/apache/hadoop/hive/llap/TestParquetCacheLayout.java	Tests cache layout behavior; no final comments.
ql/src/java/org/apache/hadoop/hive/ql/plan/MapWork.java	Enables native Parquet LLAP eligibility; no final comments.
ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedParquetRecordReader.java	Critical, 1 vote: Native reads omit row-lineage columns, producing null/default lineage values.
ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/ParquetRowGroupDecoder.java	Shares row-group decoder construction; no final comments.
ql/src/java/org/apache/hadoop/hive/ql/io/parquet/ParquetRecordReaderBase.java	Reviewed as part of Parquet reader integration; no final comments.
ql/src/java/org/apache/hadoop/hive/ql/io/HiveInputFormat.java	Gates native Parquet wrapping; no final comments.
ql/src/java/org/apache/hadoop/hive/llap/ParquetRangeBuffers.java	Manages vectored-read buffers; no final comments.
ql/src/java/org/apache/hadoop/hive/llap/ParquetCacheLayout.java	Moderate, 1 vote: Native layout can create fallback-incompatible cache gaps and trigger an assertion instead of refetching.
ql/src/java/org/apache/hadoop/hive/llap/LlapCacheAwareFs.java	Supports cache-aware fallback reads; no final comments.
llap-server/src/test/org/apache/hadoop/hive/llap/io/encoded/TestParquetEncodedDataReader.java	Tests native reads and caching; no final comments.
llap-server/src/test/org/apache/hadoop/hive/llap/io/api/impl/TestLlapRecordReader.java	Tests LLAP reader behavior; no final comments.
llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/ParquetEncodedDataReader.java	Critical, 2 votes: Top-level field indexes are used against flattened leaf columns, misselecting nested projections. Critical, 1 vote: Partial cache insertion failure can cause cache-owned buffers to be deallocated. Nit, 1 vote: count(*) fetches and caches unnecessary column chunks.
llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/ParquetEncodedColumnBatch.java	Carries cached row-group data; no final comments.
llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/ParquetEncodedDataConsumer.java	Moderate, 1 vote: Decode exceptions are swallowed, allowing later I/O and resource use after failure.
llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/ParquetColumnVectorProducer.java	Creates native Parquet pipelines; no final comments.
llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/ParquetCachedPageReadStore.java	Reads cached Parquet pages; no final comments.
llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapRecordReader.java	Supports producer fallback; no final comments.
llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java	Moderate, 1 vote: Discards supplied fileKey and tag. Moderate, 1 vote: Can create the native producer without a metadata cache in memory mode none. Moderate, 1 vote: Does not honor the per-query native Parquet disable flag.
llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapInputFormat.java	Integrates LLAP input handling; no final comments.
llap-client/src/java/org/apache/hadoop/hive/llap/io/api/LlapIo.java	Extends the Parquet LLAP API; no final comments.
iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vector/HiveVectorizedReader.java	Updates footer-cache API usage; no final comments.
iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java	Reviewed for storage-handler integration; no final comments.
common/src/java/org/apache/hadoop/hive/conf/HiveConf.java	Adds native Parquet configuration; no final comments.

Review details

Suppressed comments (6)

llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java:503

• This constructs a plain FileSplit and silently discards the fileKey and tag supplied by the public API. The native reader then derives a different key from the job configuration/filesystem in loadFooter() and computes its own tag, so callers that provide a stable file identity cannot read or populate the intended data-cache entry or preserve the administrative cache tag. Propagate these arguments through the native reader, or remove them from the API instead of ignoring them.

    FileSplit split = new FileSplit(path, offset, length, (String[]) null);
    try {
      LlapRecordReader rr = LlapRecordReader.create(conf, split, tableIncludedCols, HiveStringUtils.getHostname(),

llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java:285

• dataCache is non-null in LLAP_IO_MEMORY_MODE=none because LlapIoImpl assigns a SimpleBufferManager to it while leaving fileMetadataCache null. With the native flag enabled, this condition still creates parquetCvp; ParquetEncodedDataReader.loadFooter then calls getParquetFooterBuffersFromCache, whose first operation requires a non-null metadata cache, so the LLAP input format fails instead of falling back to the normal Parquet reader. Gate this producer on the metadata cache as well (or make the native reader bypass the footer cache when it is unavailable).

    this.parquetCvp = dataCache != null
        ? new ParquetColumnVectorProducer(dataCache, bufferManagerOrc, conf, cacheMetrics, ioMetrics)
        : null;

llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java:500

• The API contract says this method returns null when native Parquet IO is disabled, but parquetCvp is constructed independently of the per-query flag and this check only tests for null. A caller can therefore set hive.llap.io.parquet.native.enabled=false and still start the native reader, bypassing the configuration gate used by HiveInputFormat. Include ConfVars.LLAP_IO_PARQUET_NATIVE_ENABLED in this guard.

    if (parquetCvp == null) {
      return null;
    }

llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/ParquetEncodedDataConsumer.java:195

• These exceptions are reported but swallowed, so EncodedDataConsumer.consumeData returns normally and ParquetEncodedDataReader.performDataRead continues fetching and decoding later row groups before calling setDone. A corrupt page in an early group can therefore keep doing I/O and holding cache resources after the downstream reader has failed. Propagate the decode failure after recording it (or let consumeData perform the error notification) so the reader's abort path stops the split.

    } catch (IOException | RuntimeException e) {
      // parquet-mr reports decode failures as runtime ParquetDecodingException.
      LlapIoImpl.LOG.error("Parquet decodeBatch failed for rowGroup " + batch.rowGroupIx + " of " + path, e);
      downstreamConsumer.setError(e);
    } finally {

llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/ParquetEncodedDataReader.java:197

• For the count(*) projection, colsToInclude can be empty while requestedSchema still contains the table fields; the shared decoder explicitly documents that all readers are null in this case. This code nevertheless derives every requested field as a projected leaf, so the native path fetches and caches every column chunk even though it only needs each row group's row count. Short-circuit the projected-leaf list when includes.getPhysicalColumnIds() is empty.

    int[] projected = projectedLeaves(requestedSchema, fileSchema);

ql/src/java/org/apache/hadoop/hive/llap/ParquetCacheLayout.java:88

• This native layout is incompatible with the existing Parquet fallback cache writer. LlapCacheAwareFs.getAndValidateMissingChunks requires every missing gap to start and end on maxAlloc boundaries, while this method can split a 5 MiB chunk into 4 MiB + 1 MiB. If the 1 MiB native buffer is evicted and a later nested-type/fallback reader requests the chunk, the fallback sees a gap starting at chunkStart + 4 MiB and throws an AssertionError instead of refetching it. Both paths need to share the same layout, or the fallback must accept arbitrary cache gaps.

  public int[] bufferSizes(long length) {
    int count = 0;
    for (long left = length; left > 0; ++count) {
      left -= left < minBuffer ? left : Math.min(maxBuffer, Long.highestOneBit(left));
    }

• Files reviewed: 24/24 changed files
• Comments generated: 3
• Review effort level: Lite

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.