You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by ab...@apache.org on 2026/09/17 07:25:22 UTC
(hive) 06/06: Address Copilot review comments on PR #6793
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 <bo...@gmail.com>
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 <no...@anthropic.com>
---
.../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 {