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 5ba48b518f49fde8c0974bab5da85748ca061030 Author: Laszlo Bodor <[email protected]> AuthorDate: Thu Sep 17 09:24:52 2026 +0200 Address Copilot review comments on PR #6793 Three Critical findings on the native Parquet cache path: - projectedLeaves used fileSchema.getFieldIndex to pick column chunks, but BlockMetaData.getColumns() is in flat leaf order. A file schema like 'group nested {a,b}, x' projecting 'x' would hit nested.b instead of x. Map each requested top-level primitive to its single-segment leaf in fileSchema.getColumns() instead. Add a regression test. - putColumn set part.owned only after processing all missing ranges for the column, so a mid-run throw from putFileData (e.g. its length-mismatch guard) left already-inserted cache buffers looking like raw allocations and finishFetch would allocator.deallocate cache-owned memory. Insert one range at a time and flip ownership per part. - ParquetEncodedDataReader.loadFooter builds requestedSchema without the row-lineage columns the fallback reader adds via RowLineageUtils.getRequestedSchemaWithRowLineageColumns, so with row lineage on the native path would silently emit nulls for ROW__LINEAGE__ID / LAST__UPDATED__SEQUENCE__NUMBER. Detect that case in the producer and fall back, matching the nested-projection fallback. Co-Authored-By: Claude Code <[email protected]> --- .../io/decode/ParquetColumnVectorProducer.java | 5 ++ .../llap/io/encoded/ParquetEncodedDataReader.java | 81 ++++++++++++++++------ .../io/encoded/TestParquetEncodedDataReader.java | 37 ++++++++++ 3 files changed, 101 insertions(+), 22 deletions(-) diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/ParquetColumnVectorProducer.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/ParquetColumnVectorProducer.java index 9e159b166c7..b55b93e86fb 100644 --- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/ParquetColumnVectorProducer.java +++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/ParquetColumnVectorProducer.java @@ -73,6 +73,11 @@ public ReadPipeline createReadPipeline(Consumer<ColumnVectorBatch> consumer, Fil + "nested types in the projection", split.getPath()); return null; } + if (reader.needsRowLineage()) { + LlapIoImpl.LOG.info("Parquet native cache: falling back to normal reader for {} due to " + + "row-lineage virtual columns", split.getPath()); + return null; + } edc.init(reader, reader); return edc; } catch (IOException e) { diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/ParquetEncodedDataReader.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/ParquetEncodedDataReader.java index ef66853bbf7..d03a0f75a3e 100644 --- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/ParquetEncodedDataReader.java +++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/ParquetEncodedDataReader.java @@ -66,6 +66,10 @@ import org.apache.hadoop.hive.ql.io.orc.encoded.StoppableAllocator; import org.apache.hadoop.hive.ql.io.parquet.ParquetRecordReaderBase; import org.apache.hadoop.hive.ql.io.parquet.vector.VectorizedParquetRecordReader; +import org.apache.hadoop.hive.ql.exec.Utilities; +import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatchCtx; +import org.apache.hadoop.hive.ql.metadata.RowLineageUtils; +import org.apache.hadoop.hive.ql.metadata.VirtualColumn; import org.apache.hadoop.mapred.FileSplit; import org.apache.hadoop.mapred.JobConf; import org.apache.hadoop.security.UserGroupInformation; @@ -81,6 +85,7 @@ import org.apache.parquet.hadoop.util.HadoopStreams; import org.apache.parquet.io.InputFile; import org.apache.parquet.io.SeekableInputStream; +import org.apache.parquet.column.ColumnDescriptor; import org.apache.parquet.schema.MessageType; import org.apache.parquet.schema.Type; import org.apache.tez.common.CallableWithNdc; @@ -274,15 +279,48 @@ public boolean projectsNestedTypes() { return false; } - /** File-schema positions of the requested fields; column chunks follow the schema order. */ - private static int[] projectedLeaves(MessageType requestedSchema, MessageType fileSchema) { - List<Integer> leaves = new ArrayList<>(); + /** + * Whether the query needs row-lineage virtual columns ({@code ROW__LINEAGE__ID} / + * {@code LAST__UPDATED__SEQUENCE__NUMBER}) that the fallback reader augments the requested schema + * with via {@link RowLineageUtils#getRequestedSchemaWithRowLineageColumns}. The native pipeline + * doesn't propagate that augmentation through {@code Includes} yet, so we defer to the fallback + * reader instead of quietly emitting nulls in the lineage slots. + */ + public boolean needsRowLineage() { + VectorizedRowBatchCtx rbCtx = Utilities.getVectorizedRowBatchCtx(jobConf); + if (rbCtx == null) { + return false; + } + MessageType fileSchema = footer.getFileMetaData().getSchema(); + return RowLineageUtils.isRowLineageColumnPresent(rbCtx, fileSchema, VirtualColumn.ROW_LINEAGE_ID) + || RowLineageUtils.isRowLineageColumnPresent(rbCtx, fileSchema, + VirtualColumn.LAST_UPDATED_SEQUENCE_NUMBER); + } + + /** + * Leaf-column positions of the requested fields; {@code BlockMetaData.getColumns()} is a flat list + * of leaves in file-schema order, so a top-level primitive at ordinal {@code k} in the file schema + * may sit at a very different leaf index (e.g. a nested group of two leaves before it shifts it + * from {@code 1} to {@code 2}). We only reach this method when every requested field is a + * top-level primitive ({@link #projectsNestedTypes()} would have forced a fallback otherwise), so + * each requested field maps to exactly one leaf whose path is a single segment. + */ + static int[] projectedLeaves(MessageType requestedSchema, MessageType fileSchema) { + List<ColumnDescriptor> allLeaves = fileSchema.getColumns(); + List<Integer> selected = new ArrayList<>(); for (Type field : requestedSchema.getFields()) { - if (fileSchema.containsField(field.getName())) { - leaves.add(fileSchema.getFieldIndex(field.getName())); + if (!fileSchema.containsField(field.getName())) { + continue; + } + for (int i = 0; i < allLeaves.size(); ++i) { + String[] path = allLeaves.get(i).getPath(); + if (path.length == 1 && path[0].equals(field.getName())) { + selected.add(i); + break; + } } } - return leaves.stream().mapToInt(Integer::intValue).toArray(); + return selected.stream().mapToInt(Integer::intValue).toArray(); } private static long bytes(int[] projected, BlockMetaData block) { @@ -525,25 +563,24 @@ private void putColumn(Allocator allocator, ColumnPlan column) { if (fileKey == null) { for (Part part : parts) { bufferManager.incRefBuffer(part.buffer); + part.owned = true; } - } else { - MemoryBuffer[] fresh = new MemoryBuffer[run.count]; - MemoryBuffer[] cached = new MemoryBuffer[run.count]; - DiskRange[] ranges = new DiskRange[run.count]; - for (int i = 0; i < run.count; ++i) { - fresh[i] = cached[i] = parts.get(i).buffer; - ranges[i] = parts.get(i).range; - } - lowLevelCache.putFileData(fileKey, ranges, cached, 0, Priority.NORMAL, counters, cacheTag); - for (int i = 0; i < run.count; ++i) { - if (cached[i] != fresh[i]) { - // The cache kept its own buffer (locked for us) and unlocked ours without freeing it. - allocator.deallocate(fresh[i]); - parts.get(i).buffer = cached[i]; - } - } + continue; } + // One range at a time so a mid-run throw from putFileData (e.g. the length-mismatch guard + // in LowLevelCacheImpl) leaves ownership clean: parts already handled are marked owned so + // cleanup calls decRefBuffer on cache-owned memory, while the current and later ones stay + // raw allocations that cleanup can safely deallocate. for (Part part : parts) { + MemoryBuffer fresh = part.buffer; + MemoryBuffer[] pair = new MemoryBuffer[] { fresh }; + DiskRange[] range = new DiskRange[] { part.range }; + lowLevelCache.putFileData(fileKey, range, pair, 0, Priority.NORMAL, counters, cacheTag); + if (pair[0] != fresh) { + // The cache kept its own buffer (locked for us) and unlocked ours without freeing it. + allocator.deallocate(fresh); + part.buffer = pair[0]; + } part.owned = true; } } diff --git a/llap-server/src/test/org/apache/hadoop/hive/llap/io/encoded/TestParquetEncodedDataReader.java b/llap-server/src/test/org/apache/hadoop/hive/llap/io/encoded/TestParquetEncodedDataReader.java index 8a8a4ac1bf6..d3f996d3fdc 100644 --- a/llap-server/src/test/org/apache/hadoop/hive/llap/io/encoded/TestParquetEncodedDataReader.java +++ b/llap-server/src/test/org/apache/hadoop/hive/llap/io/encoded/TestParquetEncodedDataReader.java @@ -605,6 +605,43 @@ public void testNoFileKeyReadsUncached() throws Exception { } } + @Test + public void testProjectedLeavesSkipsPrecedingNestedGroup() { + // File schema: struct nested { a, b }, x + // Requested: x (a top-level primitive) + // Column chunks are in leaf order — nested.a, nested.b, x — so a top-level ordinal (1) would + // pick nested.b, whereas the leaf-ordinal (2) picks x. Regression for the getFieldIndex misuse. + MessageType fileSchema = Types.buildMessage() + .requiredGroup() + .required(PrimitiveTypeName.INT32).named("a") + .required(PrimitiveTypeName.INT32).named("b") + .named("nested") + .required(PrimitiveTypeName.INT32).named("x") + .named("file_schema"); + MessageType requested = Types.buildMessage() + .required(PrimitiveTypeName.INT32).named("x") + .named("requested"); + + int[] leaves = ParquetEncodedDataReader.projectedLeaves(requested, fileSchema); + + assertArrayEquals(new int[] { 2 }, leaves); + } + + @Test + public void testProjectedLeavesMissingFieldSkipped() { + MessageType fileSchema = Types.buildMessage() + .required(PrimitiveTypeName.INT32).named("a") + .required(PrimitiveTypeName.INT32).named("b") + .named("file_schema"); + MessageType requested = Types.buildMessage() + .required(PrimitiveTypeName.INT32).named("a") + .required(PrimitiveTypeName.INT32).named("missing") + .named("requested"); + + assertArrayEquals(new int[] { 0 }, + ParquetEncodedDataReader.projectedLeaves(requested, fileSchema)); + } + // ---- fixture ---- private static void writeFile(Path path, Configuration conf) throws IOException {
