This is an automated email from the ASF dual-hosted git repository. asf-gitbox-commits pushed a commit to branch parquet-io-cache in repository https://gitbox.apache.org/repos/asf/hive.git
commit ea80ce8fd6f78b2bb054c04ee7bbe46a131530c6 Author: Laszlo Bodor <[email protected]> AuthorDate: Thu Sep 17 09:04:08 2026 +0200 Address Denys's PR review comments - HiveIcebergStorageHandler: import ParquetRowGroupDecoder so the javadoc {@link} can use the short name. - LlapIoImpl: rename local bufferManagerOrc -> bufferManagerData; the buffer manager is now shared with the native Parquet producer, so the Orc suffix is stale. - LlapRecordReader: rename checkOrcSchemaEvolution -> checkSchemaEvolution since the method already tolerates producers (like native Parquet) that do not expose an ORC-style SchemaEvolution. - ParquetCachedPageReadStore.chunkBuffers: split the two-per-line locals, and rename from/to -> sliceStart/sliceEnd (with chunkStart/chunkEnd and bufferStart/bufferEnd for the surrounding bounds). Co-Authored-By: Claude Code <[email protected]> --- .../apache/iceberg/mr/hive/HiveIcebergStorageHandler.java | 3 ++- .../org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java | 10 +++++----- .../hadoop/hive/llap/io/api/impl/LlapRecordReader.java | 9 +++++---- .../hive/llap/io/decode/ParquetCachedPageReadStore.java | 13 ++++++++----- 4 files changed, 20 insertions(+), 15 deletions(-) diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java index 0a540f746da..729fb0e268e 100644 --- a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java +++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java @@ -84,6 +84,7 @@ import org.apache.hadoop.hive.ql.exec.Utilities; import org.apache.hadoop.hive.ql.hooks.WriteEntity; import org.apache.hadoop.hive.ql.io.StorageFormatDescriptor; +import org.apache.hadoop.hive.ql.io.parquet.vector.ParquetRowGroupDecoder; import org.apache.hadoop.hive.ql.io.sarg.ConvertAstToSearchArg; import org.apache.hadoop.hive.ql.io.sarg.SearchArgument; import org.apache.hadoop.hive.ql.metadata.DefaultStorageHandler; @@ -1889,7 +1890,7 @@ private static boolean hasOrcTimeInSchema(Properties tableProps, Schema tableSch /** * Vectorized reads of parquet files from columns with list or map type is only supported if the nested types are of * primitive type category - * check {@link org.apache.hadoop.hive.ql.io.parquet.vector.ParquetRowGroupDecoder#checkListColumnSupport} for + * check {@link ParquetRowGroupDecoder#checkListColumnSupport} for * details on nested types under lists * @param tableProps iceberg table properties * @param tableSchema iceberg table schema diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java index adae60713f3..d95b43d7e2f 100644 --- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java +++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java @@ -178,7 +178,7 @@ private LlapIoImpl(Configuration conf) throws IOException { MetadataCache metadataCache = null; SerDeLowLevelCacheImpl serdeCache = null; // TODO: extract interface when needed - BufferUsageManager bufferManagerOrc = null, bufferManagerGeneric = null; + BufferUsageManager bufferManagerData = null, bufferManagerGeneric = null; boolean isEncodeEnabled = useLowLevelCache && HiveConf.getBoolVar(conf, ConfVars.LLAP_IO_ENCODE_ENABLED); if (useLowLevelCache) { @@ -226,7 +226,7 @@ private LlapIoImpl(Configuration conf) throws IOException { cachePolicyWrapper.setEvictionListener(e); cacheImpl.startThreads(); // Start the cache threads. - bufferManager = bufferManagerOrc = cacheImpl; // Cache also serves as buffer manager. + bufferManager = bufferManagerData = cacheImpl; // Cache also serves as buffer manager. bufferManagerGeneric = serdeCache; if (trackUsage) { debugDumpComponents.add(cachePolicyWrapper); // Cache contents tracker. @@ -245,7 +245,7 @@ private LlapIoImpl(Configuration conf) throws IOException { this.allocator = new SimpleAllocator(conf); fileMetadataCache = null; SimpleBufferManager sbm = new SimpleBufferManager(allocator, cacheMetrics); - bufferManager = bufferManagerOrc = bufferManagerGeneric = sbm; + bufferManager = bufferManagerData = bufferManagerGeneric = sbm; dataCache = sbm; this.memoryManager = null; debugDumpComponents.add(new LlapIoDebugDump() { @@ -276,12 +276,12 @@ public void debugDumpShort(StringBuilder sb) { // TODO: this should depends on input format and be in a map, or something. this.orcCvp = new OrcColumnVectorProducer( - metadataCache, dataCache, pathCache, bufferManagerOrc, conf, cacheMetrics, ioMetrics, tracePool); + metadataCache, dataCache, pathCache, bufferManagerData, conf, cacheMetrics, ioMetrics, tracePool); this.genericCvp = isEncodeEnabled ? new GenericColumnVectorProducer( serdeCache, bufferManagerGeneric, conf, cacheMetrics, ioMetrics, tracePool, encodeExecutor) : null; // Native Parquet IO is gated per query by the job conf at the dispatch sites. this.parquetCvp = dataCache != null - ? new ParquetColumnVectorProducer(dataCache, bufferManagerOrc, conf, cacheMetrics, ioMetrics) + ? new ParquetColumnVectorProducer(dataCache, bufferManagerData, conf, cacheMetrics, ioMetrics) : null; LOG.info("LLAP IO initialized"); diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapRecordReader.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapRecordReader.java index cbf9a7fad1c..01fa9cd4705 100644 --- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapRecordReader.java +++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapRecordReader.java @@ -138,7 +138,7 @@ public static LlapRecordReader create(JobConf job, FileSplit split, if (rr.rp == null) { return null; // The producer declined the split; the caller uses the source reader. } - if (!rr.checkOrcSchemaEvolution()) { + if (!rr.checkSchemaEvolution()) { rr.close(); throwIfCacheOnlyRead(HiveConf.getBoolVar(job, ConfVars.LLAP_IO_CACHE_ONLY)); return null; @@ -351,11 +351,12 @@ public void start() { executor.submit(rp.getReadCallable()); } - private boolean checkOrcSchemaEvolution() { + private boolean checkSchemaEvolution() { SchemaEvolution evolution = rp.getSchemaEvolution(); if (evolution == null) { - // No ORC-style schema evolution to validate (e.g. native parquet path); - // parquet handles its own column resolution. Nothing to check here. + // Only the ORC pipeline hangs an ORC SchemaEvolution off the ReadPipeline; other formats + // (native Parquet today) handle column resolution themselves, so there is nothing to + // validate here. return true; } diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/ParquetCachedPageReadStore.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/ParquetCachedPageReadStore.java index 09036b065b4..a35c770d09c 100644 --- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/ParquetCachedPageReadStore.java +++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/ParquetCachedPageReadStore.java @@ -75,14 +75,17 @@ class ParquetCachedPageReadStore implements PageReadStore { /** Slices of the cached buffers covering exactly the chunk's byte region, in file order. */ private static List<ByteBuffer> chunkBuffers(ParquetEncodedColumnBatch batch, int pc) { - long start = batch.chunks[pc].getStartingPos(), end = start + batch.chunks[pc].getTotalSize(); + long chunkStart = batch.chunks[pc].getStartingPos(); + long chunkEnd = chunkStart + batch.chunks[pc].getTotalSize(); List<ByteBuffer> slices = new ArrayList<>(batch.columnBuffers[pc].length); for (int i = 0; i < batch.columnBuffers[pc].length; ++i) { - long offset = batch.bufferOffsets[pc][i]; - long from = Math.max(start, offset), to = Math.min(end, offset + batch.bufferLengths[pc][i]); + long bufferStart = batch.bufferOffsets[pc][i]; + long bufferEnd = bufferStart + batch.bufferLengths[pc][i]; + long sliceStart = Math.max(chunkStart, bufferStart); + long sliceEnd = Math.min(chunkEnd, bufferEnd); ByteBuffer bb = batch.columnBuffers[pc][i].getByteBufferDup(); - bb.position(bb.position() + (int) (from - offset)); - bb.limit(bb.position() + (int) (to - from)); + bb.position(bb.position() + (int) (sliceStart - bufferStart)); + bb.limit(bb.position() + (int) (sliceEnd - sliceStart)); slices.add(bb.slice()); } return slices;
