From 85bf856019b43488ab68a904f1a45dc2d586323a Mon Sep 17 00:00:00 2001 From: Denys Kuzmenko Date: Sat, 3 Oct 2026 18:52:36 +0300 Subject: [PATCH 1/4] HIVE-30102: Iceberg: LLAP cannot read an ORC file that has positional deletes - LLAP reports the file row of the first row of each batch: ORC files with position deletes or virtual columns read through LLAP IO - The non-LLAP vectorized ORC reader forces orc.sarg.to.filter off: the row-level filter drops rows and whole batches without reporting their positions; the SARG still prunes stripes and row groups (HIVE-30110) - Impact: correctness/routing; ORC files with deletes now read through LLAP IO (not timed) --- .../mr/hive/vector/HiveVectorizedReader.java | 9 +- .../vector/TestHiveIcebergVectorization.java | 25 + .../llap/io/api/impl/ColumnVectorBatch.java | 4 +- .../llap/io/api/impl/LlapRecordReader.java | 12 +- .../decode/GenericColumnVectorProducer.java | 12 + .../io/decode/OrcEncodedDataConsumer.java | 10 + .../io/metadata/ConsumerFileMetadata.java | 4 + ...estOrcEncodedDataConsumerRowPositions.java | 509 ++++++++++++++++++ 8 files changed, 579 insertions(+), 6 deletions(-) create mode 100644 llap-server/src/test/org/apache/hadoop/hive/llap/io/decode/TestOrcEncodedDataConsumerRowPositions.java diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vector/HiveVectorizedReader.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vector/HiveVectorizedReader.java index 52140dfb88b7..9ae1bc2b1f30 100644 --- a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vector/HiveVectorizedReader.java +++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vector/HiveVectorizedReader.java @@ -61,7 +61,6 @@ import org.apache.iceberg.io.CloseableIterable; import org.apache.iceberg.io.CloseableIterator; import org.apache.iceberg.io.FileIO; -import org.apache.iceberg.mr.InputFormatConfig; import org.apache.iceberg.mr.hive.HiveIcebergInputFormat; import org.apache.iceberg.mr.mapred.MapredIcebergInputFormat; import org.apache.iceberg.orc.VectorizedReadUtils; @@ -220,6 +219,8 @@ private static RecordReader orcRecordReader(Jo // Need to turn positional schema evolution off since we use column name based schema evolution for projection // and Iceberg will make a mapping between the file schema and the current reading schema. job.setBoolean(OrcConf.FORCE_POSITIONAL_EVOLUTION.getHiveConfName(), false); + // Row positions are derived from the batch index, so the reader must not filter rows within a batch. + job.setBoolean(OrcConf.ALLOW_SARG_TO_FILTER.getHiveConfName(), false); // Metadata information has to be passed along in the OrcSplit. Without specifying this, the vectorized // reader will assume that the ORC file ends at the task's start + length, and might fail reading the tail.. @@ -229,10 +230,10 @@ private static RecordReader orcRecordReader(Jo VectorizedReadUtils.handleIcebergProjection(task, job, VectorizedReadUtils.deserializeToShadedOrcTail(serializedOrcTail).getSchema(), residual); - // If LLAP enabled, try to retrieve an LLAP record reader - this might yield to null in some special cases - // TODO: add support for reading files with positional deletes with LLAP (LLAP would need to provide file row num) + // If LLAP enabled, try to retrieve an LLAP record reader - this might yield to null in some special cases. + // Position deletes and the virtual columns are served from the row positions the reader reports. if (HiveConf.getBoolVar(job, HiveConf.ConfVars.LLAP_IO_ENABLED, LlapProxy.isDaemon()) && - LlapProxy.getIo() != null && task.deletes().isEmpty() && !InputFormatConfig.fetchVirtualColumns(job)) { + LlapProxy.getIo() != null) { recordReader = LlapProxy.getIo().orc().vectorizedReaderForPath(fileId, path, null, readColumnIds, job, start, length, reporter); } diff --git a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/vector/TestHiveIcebergVectorization.java b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/vector/TestHiveIcebergVectorization.java index 28487b715ff4..e57ab1823670 100644 --- a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/vector/TestHiveIcebergVectorization.java +++ b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/vector/TestHiveIcebergVectorization.java @@ -219,6 +219,31 @@ private void testVectorizedReadWithDeleteFilter(Map props) { validation.apply(1501); } + /** + * ORC can filter the rows of a batch by the pushed down predicate; row positions and deletes must still apply to + * the right rows. + */ + @Test + public void testHiveDeleteFilterWithOrcRowFiltering() { + Assume.assumeTrue(fileFormat == FileFormat.ORC); + + Schema schema = new Schema(optional(1, "customer_id", Types.LongType.get())); + List records = TestHelper.generateRandomRecords(schema, 30000, 0L); + for (int i = 0; i < records.size(); ++i) { + records.get(i).setField("customer_id", (long) i); + } + testTables.createTable(shell, "vectordelete", schema, PartitionSpec.unpartitioned(), fileFormat, records, 2); + shell.executeStatement("DELETE FROM vectordelete WHERE customer_id = 25000"); + + shell.setHiveSessionValue("orc.sarg.to.filter", true); + shell.setHiveSessionValue("orc.filter.use.selected", true); + List result = shell.executeStatement( + "select customer_id, ROW__POSITION from vectordelete where customer_id in (25000, 25001)"); + + Assert.assertEquals(1, result.size()); + Assert.assertArrayEquals(new Object[] { 25001L, 25001L }, result.get(0)); + } + @Test public void testHiveDeleteFilterWithFilteredParquetBlock() { Assume.assumeTrue(fileFormat == FileFormat.PARQUET); diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/ColumnVectorBatch.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/ColumnVectorBatch.java index df9bcf2d7e1e..a675e37cf01b 100644 --- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/ColumnVectorBatch.java +++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/ColumnVectorBatch.java @@ -34,6 +34,8 @@ public class ColumnVectorBatch { public MutableFilterContext filterContext; public ColumnVector[] cols; public int size; + /** File row position of the first row; -1 when the producer does not track positions. */ + public long startRowInFile = -1; public ColumnVectorBatch(int columnCount) { this(columnCount, VectorizedRowBatch.DEFAULT_SIZE); @@ -101,4 +103,4 @@ public String toString() { return b.toString(); } -} \ No newline at end of file +} 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 d74efcd2d3be..6db6cc854495 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 @@ -53,6 +53,7 @@ import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatchCtx; import org.apache.hadoop.hive.ql.io.AcidUtils; import org.apache.hadoop.hive.ql.io.BucketIdentifier; +import org.apache.hadoop.hive.ql.io.RowPositionAwareVectorizedRecordReader; import org.apache.hadoop.hive.ql.io.orc.OrcInputFormat; import org.apache.hadoop.hive.ql.io.orc.OrcRecordUpdater; import org.apache.hadoop.hive.ql.io.orc.OrcSplit; @@ -85,7 +86,8 @@ import static org.apache.hadoop.hive.llap.LlapHiveUtils.throwIfCacheOnlyRead; -class LlapRecordReader implements RecordReader, Consumer { +class LlapRecordReader implements RecordReader, Consumer, + RowPositionAwareVectorizedRecordReader { private static final Logger LOG = LoggerFactory.getLogger(LlapRecordReader.class); private static final Object DONE_OBJECT = new Object(); @@ -581,6 +583,14 @@ ColumnVectorBatch nextCvb() throws InterruptedException, IOException { return lastCvb; } + @Override + public long getRowNumber() { + if (lastCvb.startRowInFile < 0) { + throw new IllegalStateException("Row positions are not provided by " + rp.getClass().getSimpleName()); + } + return lastCvb.startRowInFile; + } + @Override public NullWritable createKey() { return NullWritable.get(); diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/GenericColumnVectorProducer.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/GenericColumnVectorProducer.java index 49426eb73636..562cd973f141 100644 --- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/GenericColumnVectorProducer.java +++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/GenericColumnVectorProducer.java @@ -21,6 +21,7 @@ import java.io.IOException; import java.util.ArrayList; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.TimeZone; @@ -59,6 +60,7 @@ import org.apache.orc.OrcProto.RowIndex; import org.apache.orc.OrcProto.RowIndexEntry; import org.apache.orc.OrcProto.Type; +import org.apache.orc.StripeInformation; import org.apache.orc.TypeDescription; public class GenericColumnVectorProducer implements ColumnVectorProducer { @@ -304,5 +306,15 @@ public OrcFile.Version getFileVersion() { public CalendarKind getCalendar() { return CalendarKind.JULIAN_GREGORIAN; } + + @Override + public List getStripes() { + return Collections.emptyList(); + } + + @Override + public int getRowIndexStride() { + return 0; + } } } diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcEncodedDataConsumer.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcEncodedDataConsumer.java index 36e1ab3c54ae..d00a7d93b2e0 100644 --- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcEncodedDataConsumer.java +++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcEncodedDataConsumer.java @@ -70,6 +70,7 @@ public class OrcEncodedDataConsumer private TypeReader[] columnReaders; private int previousStripeIndex = -1; private ConsumerFileMetadata fileMetadata; // We assume one request is only for one file. + private long[] stripeStartRows; // file row of each stripe's first row; null without file positions private CompressionCodec codec; private List stripes; private SchemaEvolution evolution; @@ -95,6 +96,12 @@ public void setUseDecimal64ColumnVectors(final boolean useDecimal64ColumnVectors public void setFileMetadata(ConsumerFileMetadata f) { assert fileMetadata == null; fileMetadata = f; + if (!f.getStripes().isEmpty()) { + stripeStartRows = new long[f.getStripes().size()]; + for (int i = 1; i < stripeStartRows.length; ++i) { + stripeStartRows[i] = stripeStartRows[i - 1] + f.getStripes().get(i - 1).getNumberOfRows(); + } + } stripes = new ArrayList<>(f.getStripeCount()); codec = WriterImpl.createCodec(fileMetadata.getCompressionKind()); } @@ -137,6 +144,8 @@ protected void decodeBatch(OrcEncodedColumnBatch batch, } int maxBatchesRG = (int) ((nonNullRowCount / VectorizedRowBatch.DEFAULT_SIZE) + 1); int batchSize = VectorizedRowBatch.DEFAULT_SIZE; + long rgStartRow = stripeStartRows == null ? -1 : stripeStartRows[currentStripeIndex] + + ((rgIdx == OrcEncodedColumnBatch.ALL_RGS || noIndex) ? 0 : (long) rgIdx * fileMetadata.getRowIndexStride()); TypeDescription fileSchema = fileMetadata.getSchema(); if (columnReaders == null || !sameStripe || noIndex) { @@ -155,6 +164,7 @@ protected void decodeBatch(OrcEncodedColumnBatch batch, ColumnVectorBatch cvb = takeBatch(batchSize); // assert cvb.cols.length == batch.getColumnIxs().length; // Must be constant per split. + cvb.startRowInFile = rgStartRow < 0 ? -1 : rgStartRow + (long) i * VectorizedRowBatch.DEFAULT_SIZE; for (int idx = 0; idx < columnReaders.length; ++idx) { /* * Currently, ORC's TreeReaderFactory class does this: diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/ConsumerFileMetadata.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/ConsumerFileMetadata.java index 928b48336753..59a3ccf13e36 100644 --- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/ConsumerFileMetadata.java +++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/ConsumerFileMetadata.java @@ -25,6 +25,7 @@ import org.apache.orc.OrcFile; import org.apache.orc.OrcProto.CalendarKind; import org.apache.orc.OrcProto.Type; +import org.apache.orc.StripeInformation; import org.apache.orc.TypeDescription; public interface ConsumerFileMetadata { @@ -34,4 +35,7 @@ public interface ConsumerFileMetadata { TypeDescription getSchema() throws FileFormatException; OrcFile.Version getFileVersion(); CalendarKind getCalendar(); + /** Every stripe of the file in order, or empty when the data has no file positions (re-encoded input). */ + List getStripes(); + int getRowIndexStride(); } diff --git a/llap-server/src/test/org/apache/hadoop/hive/llap/io/decode/TestOrcEncodedDataConsumerRowPositions.java b/llap-server/src/test/org/apache/hadoop/hive/llap/io/decode/TestOrcEncodedDataConsumerRowPositions.java new file mode 100644 index 000000000000..45cf1385db6a --- /dev/null +++ b/llap-server/src/test/org/apache/hadoop/hive/llap/io/decode/TestOrcEncodedDataConsumerRowPositions.java @@ -0,0 +1,509 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.hadoop.hive.llap.io.decode; + +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.apache.hadoop.hive.metastore.api.hive_metastoreConstants.META_TABLE_NAME; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.nio.file.Files; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Properties; + +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.conf.HiveConf.ConfVars; +import org.apache.hadoop.hive.llap.counters.QueryFragmentCounters; +import org.apache.hadoop.hive.llap.io.api.LlapProxy; +import org.apache.hadoop.hive.llap.io.api.impl.ColumnVectorBatch; +import org.apache.hadoop.hive.llap.io.decode.ColumnVectorProducer.Includes; +import org.apache.hadoop.hive.llap.io.metadata.ConsumerFileMetadata; +import org.apache.hadoop.hive.llap.io.metadata.ConsumerStripeMetadata; +import org.apache.hadoop.hive.llap.metrics.LlapDaemonIOMetrics; +import org.apache.hadoop.hive.ql.exec.Utilities; +import org.apache.hadoop.hive.ql.exec.vector.BytesColumnVector; +import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector; +import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch; +import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatchCtx; +import org.apache.hadoop.hive.ql.io.IOConstants; +import org.apache.hadoop.hive.ql.io.RowPositionAwareVectorizedRecordReader; +import org.apache.hadoop.hive.ql.io.orc.OrcInputFormat; +import org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat; +import org.apache.hadoop.hive.ql.io.orc.encoded.Consumer; +import org.apache.hadoop.hive.ql.io.orc.encoded.IoTrace; +import org.apache.hadoop.hive.ql.io.orc.encoded.Reader.OrcEncodedColumnBatch; +import org.apache.hadoop.hive.ql.plan.MapWork; +import org.apache.hadoop.hive.ql.plan.PartitionDesc; +import org.apache.hadoop.hive.ql.plan.TableDesc; +import org.apache.hadoop.hive.serde2.ColumnProjectionUtils; +import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo; +import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory; +import org.apache.hadoop.io.NullWritable; +import org.apache.hadoop.mapred.JobConf; +import org.apache.hadoop.mapred.RecordReader; +import org.apache.hadoop.mapred.Reporter; +import org.apache.orc.CompressionKind; +import org.apache.orc.OrcFile; +import org.apache.orc.OrcProto; +import org.apache.orc.OrcProto.CalendarKind; +import org.apache.orc.OrcProto.ColumnEncoding; +import org.apache.orc.OrcProto.RowIndex; +import org.apache.orc.OrcProto.RowIndexEntry; +import org.apache.orc.StripeInformation; +import org.apache.orc.TypeDescription; +import org.apache.orc.Writer; +import org.apache.tez.common.counters.TezCounters; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; + +/** + * Absolute file row positions reported by {@link OrcEncodedDataConsumer} on every emitted batch. + * The indexed and index-disabled files are read end to end through LLAP IO; the ALL_RGS and the + * position-less (SerDe-like) metadata cases are driven straight into decodeBatch. + */ +public class TestOrcEncodedDataConsumerRowPositions { + + private static final TypeDescription SCHEMA = TypeDescription.fromString("struct"); + private static final int STRIDE = 2500; + // Neither a multiple of the row index stride nor of VectorizedRowBatch.DEFAULT_SIZE. + private static final int[] STRIPE_ROWS = {8300, 6100, 3700}; + private static final int[] NO_INDEX_STRIPE_ROWS = {5300, 2700}; + + private static final LlapDaemonIOMetrics IO_METRICS = + LlapDaemonIOMetrics.create("TestOrcEncodedDataConsumerRowPositions-io", "1", null); + + private static HiveConf daemonConf; + private static Path indexedFile; + private static Path noIndexFile; + + @BeforeClass + public static void setUpClass() throws Exception { + daemonConf = new HiveConf(); + HiveConf.setVar(daemonConf, ConfVars.LLAP_IO_MEMORY_MODE, "none"); + + Path tmpDir = new Path(Files.createTempDirectory("llap-orc-row-positions").toString()); + indexedFile = new Path(tmpDir, "indexed.orc"); + noIndexFile = new Path(tmpDir, "no-index.orc"); + writeFile(indexedFile, STRIDE, STRIPE_ROWS); + writeFile(noIndexFile, 0, NO_INDEX_STRIPE_ROWS); + assertLayout(indexedFile, STRIDE, STRIPE_ROWS); + assertLayout(noIndexFile, 0, NO_INDEX_STRIPE_ROWS); + + LlapProxy.setDaemon(true); + LlapProxy.initializeLlapIo(daemonConf); + } + + @AfterClass + public static void tearDownClass() { + LlapProxy.close(); + } + + @Test + public void testRowGroupPositionsAcrossStripes() throws Exception { + List batches = readThroughLlap(indexedFile); + + assertEquals(expectedBatches(STRIDE, STRIPE_ROWS), render(batches)); + // First RG of stripe 0, a middle RG of stripe 0, first RG of stripe 1, last RG of the last stripe. + for (long rgStart : new long[] {0, 5000, 8300, 16900}) { + assertEquals("no batch starts at " + rgStart, 1, countStartingAt(batches, rgStart)); + assertGroundTruth(rgStart); + } + for (long[] batch : batches) { + assertEquals("id of the first row of the batch at " + batch[0], batch[0], batch[2]); + } + } + + @Test + public void testBatchesAdvanceWithinRowGroup() throws Exception { + List batches = readThroughLlap(indexedFile); + + // The row group at file row 8300 holds 2500 rows: 1024 + 1024 + 452. + List rg = batches.subList(indexOfStart(batches, 8300), indexOfStart(batches, 8300) + 3); + assertEquals(Arrays.asList("8300:1024", "9324:1024", "10348:452"), render(rg)); + assertEquals(8300 + STRIDE, rg.get(2)[0] + rg.get(2)[1]); + } + + @Test + public void testIndexDisabledFileStartsAtStripeBoundary() throws Exception { + List batches = readThroughLlap(noIndexFile); + + assertEquals(expectedBatches(Integer.MAX_VALUE, NO_INDEX_STRIPE_ROWS), render(batches)); + assertEquals(1, countStartingAt(batches, NO_INDEX_STRIPE_ROWS[0])); + for (long[] batch : batches) { + assertEquals("id of the first row of the batch at " + batch[0], batch[0], batch[2]); + } + } + + @Test + public void testAllRowGroupsBatchStartsAtStripeBoundary() throws Exception { + List starts = decodeStripe(fileMetadata(STRIPE_ROWS, STRIDE), 1, OrcEncodedColumnBatch.ALL_RGS, true); + + assertEquals(startsWithin(STRIPE_ROWS[0], STRIPE_ROWS[1]), starts); + } + + @Test + public void testIndexDisabledStripeStartsAtStripeBoundary() throws Exception { + List starts = decodeStripe(fileMetadata(STRIPE_ROWS, STRIDE), 1, 0, false); + + assertEquals(startsWithin(STRIPE_ROWS[0], STRIPE_ROWS[1]), starts); + } + + @Test + public void testNoStripesReportsNoPosition() throws Exception { + ConsumerFileMetadata noPositions = fileMetadata(new int[0], 0); + assertTrue(noPositions.getStripes().isEmpty()); + + List starts = decodeStripe(noPositions, 0, OrcEncodedColumnBatch.ALL_RGS, true); + + assertEquals(Collections.nCopies(startsWithin(0, STRIPE_ROWS[0]).size(), -1L), starts); + } + + // ---- LLAP IO end to end ---- + + /** One {startRowInFile, size, id of the first row} per batch, in the order LLAP returned them. */ + private static List readThroughLlap(Path path) throws Exception { + RecordReader reader = llapReader(path); + List batches = new ArrayList<>(); + try { + VectorizedRowBatch vrb = reader.createValue(); + while (reader.next(NullWritable.get(), vrb)) { + long position = ((RowPositionAwareVectorizedRecordReader) reader).getRowNumber(); + batches.add(new long[] {position, vrb.size, ((LongColumnVector) vrb.cols[0]).vector[0]}); + } + } finally { + reader.close(); + } + return batches; + } + + private static RecordReader llapReader(Path path) throws IOException { + JobConf job = jobConf(path); + long fileLen = path.getFileSystem(job).getFileStatus(path).getLen(); + RecordReader reader = LlapProxy.getIo().orc().vectorizedReaderForPath( + null, path, null, Arrays.asList(0, 1), job, 0, fileLen, Reporter.NULL); + assertNotNull("LLAP should handle this ORC read", reader); + return reader; + } + + private static void assertGroundTruth(long fileRow) throws IOException { + try (org.apache.orc.Reader reader = OrcFile.createReader(indexedFile, OrcFile.readerOptions(daemonConf)); + org.apache.orc.RecordReader rows = reader.rows()) { + VectorizedRowBatch batch = SCHEMA.createRowBatch(1); + rows.seekToRow(fileRow); + assertTrue("file row " + fileRow + " is past the end", rows.nextBatch(batch)); + assertEquals(fileRow, ((LongColumnVector) batch.cols[0]).vector[0]); + assertEquals("n" + fileRow, ((BytesColumnVector) batch.cols[1]).toString(0)); + } + } + + // ---- decodeBatch with faked metadata ---- + + /** Row positions of the batches decodeBatch emits for one stripe of a zero-column read. */ + private static List decodeStripe(ConsumerFileMetadata fileMetadata, int stripeIx, int rgIx, + boolean withRowIndex) throws Exception { + QueryFragmentCounters counters = new QueryFragmentCounters(daemonConf, new TezCounters()); + List starts = new ArrayList<>(); + OrcEncodedDataConsumer consumer = new OrcEncodedDataConsumer(null, includes(), counters, IO_METRICS); + consumer.init(null, null, new IoTrace(0, false)); + consumer.setFileMetadata(fileMetadata); + consumer.setStripeMetadata(new FakeStripeMetadata(stripeIx, STRIPE_ROWS[stripeIx], withRowIndex)); + + OrcEncodedColumnBatch batch = new OrcEncodedColumnBatch(); + batch.init(null, stripeIx, rgIx, 0); + consumer.decodeBatch(batch, new Consumer() { + @Override public void consumeData(ColumnVectorBatch cvb) { + starts.add(cvb.startRowInFile); + } + + @Override public void setDone() { + } + + @Override public void setError(Throwable t) { + throw new AssertionError(t); + } + }); + return starts; + } + + private static ConsumerFileMetadata fileMetadata(int[] stripeRows, int stride) { + List stripes = new ArrayList<>(); + for (int rows : stripeRows) { + StripeInformation stripe = mock(StripeInformation.class); + when(stripe.getNumberOfRows()).thenReturn((long) rows); + stripes.add(stripe); + } + return new ConsumerFileMetadata() { + @Override public int getStripeCount() { + return stripes.size(); + } + + @Override public CompressionKind getCompressionKind() { + return CompressionKind.NONE; + } + + @Override public List getTypes() { + return Collections.emptyList(); + } + + @Override public TypeDescription getSchema() { + return SCHEMA; + } + + @Override public OrcFile.Version getFileVersion() { + return OrcFile.Version.V_0_12; + } + + @Override public CalendarKind getCalendar() { + return CalendarKind.PROLEPTIC_GREGORIAN; + } + + @Override public List getStripes() { + return stripes; + } + + @Override public int getRowIndexStride() { + return stride; + } + }; + } + + /** Reports a row count only; a null row index entry is what an index-disabled stripe returns. */ + private static final class FakeStripeMetadata implements ConsumerStripeMetadata { + private final int stripeIx; + private final long rowCount; + private final boolean withRowIndex; + + FakeStripeMetadata(int stripeIx, long rowCount, boolean withRowIndex) { + this.stripeIx = stripeIx; + this.rowCount = rowCount; + this.withRowIndex = withRowIndex; + } + + @Override public int getStripeIx() { + return stripeIx; + } + + @Override public long getRowCount() { + return rowCount; + } + + @Override public List getEncodings() { + return Collections.singletonList(ColumnEncoding.newBuilder() + .setKind(ColumnEncoding.Kind.DIRECT).build()); + } + + @Override public String getWriterTimezone() { + return "UTC"; + } + + @Override public RowIndexEntry getRowIndexEntry(int colIx, int rgIx) { + return null; + } + + @Override public RowIndex[] getRowIndexes() { + return new RowIndex[0]; + } + + @Override public boolean supportsRowIndexes() { + return withRowIndex; + } + } + + private static Includes includes() { + return new Includes() { + @Override public List getPhysicalColumnIds() { + return Collections.emptyList(); + } + + @Override public List getReaderLogicalColumnIds() { + return Collections.emptyList(); + } + + @Override public List getLogicalOrderedColumnIds() { + return Collections.emptyList(); + } + + @Override public boolean[] generateFileIncludes(TypeDescription fileSchema) { + throw new UnsupportedOperationException(); + } + + @Override public TypeDescription[] getBatchReaderTypes(TypeDescription fileSchema) { + return new TypeDescription[0]; + } + + @Override public String[] getOriginalColumnNames(TypeDescription fileSchema) { + throw new UnsupportedOperationException(); + } + + @Override public String getQueryId() { + return "test-query"; + } + + @Override public boolean isProbeDecodeEnabled() { + return false; + } + + @Override public byte getProbeMjSmallTablePos() { + return -1; + } + + @Override public String getProbeCacheKey() { + return null; + } + + @Override public String getProbeColName() { + return null; + } + + @Override public int getProbeColIdx() { + return -1; + } + }; + } + + // ---- fixture ---- + + private static void writeFile(Path path, int stride, int[] stripeRows) throws IOException { + try (Writer writer = OrcFile.createWriter(path, OrcFile.writerOptions(daemonConf) + .setSchema(SCHEMA).rowIndexStride(stride).compress(CompressionKind.NONE))) { + VectorizedRowBatch batch = SCHEMA.createRowBatch(); + LongColumnVector ids = (LongColumnVector) batch.cols[0]; + BytesColumnVector names = (BytesColumnVector) batch.cols[1]; + long row = 0; + for (int s = 0; s < stripeRows.length; ++s) { + for (int left = stripeRows[s]; left > 0; ) { + int size = Math.min(left, batch.getMaxSize()); + batch.size = size; + for (int i = 0; i < size; ++i, ++row) { + ids.vector[i] = row; + names.setVal(i, ("n" + row).getBytes(UTF_8)); + } + writer.addRowBatch(batch); + batch.reset(); + left -= size; + } + if (s < stripeRows.length - 1) { + writer.writeIntermediateFooter(); + } + } + } + } + + private static void assertLayout(Path path, int stride, int[] stripeRows) throws IOException { + try (org.apache.orc.Reader reader = OrcFile.createReader(path, OrcFile.readerOptions(daemonConf))) { + assertEquals(stride, reader.getRowIndexStride()); + List stripes = reader.getStripes(); + assertEquals(stripeRows.length, stripes.size()); + for (int i = 0; i < stripeRows.length; ++i) { + assertEquals("stripe " + i, stripeRows[i], stripes.get(i).getNumberOfRows()); + if (stride > 0) { + assertTrue("stripe " + i + " is a multiple of the stride", stripeRows[i] % stride != 0); + } + assertTrue("stripe " + i + " is a multiple of the batch size", + stripeRows[i] % VectorizedRowBatch.DEFAULT_SIZE != 0); + } + } + } + + private static JobConf jobConf(Path path) { + JobConf job = new JobConf(daemonConf); + HiveConf.setBoolVar(job, ConfVars.HIVE_VECTORIZATION_ENABLED, true); + HiveConf.setVar(job, ConfVars.PLAN, "//tmp"); + job.set(IOConstants.COLUMNS, "id,name"); + job.set(IOConstants.COLUMNS_TYPES, "bigint,string"); + job.set(ColumnProjectionUtils.ORC_SCHEMA_STRING, SCHEMA.toString()); + + Properties tblProps = new Properties(); + tblProps.setProperty(META_TABLE_NAME, "default.test_orc_positions"); + TableDesc tableDesc = new TableDesc(OrcInputFormat.class, OrcOutputFormat.class, tblProps); + + MapWork mapWork = new MapWork(); + mapWork.setVectorMode(true); + mapWork.setVectorizedRowBatchCtx(new VectorizedRowBatchCtx( + new String[] {"id", "name"}, + new TypeInfo[] {TypeInfoFactory.longTypeInfo, TypeInfoFactory.stringTypeInfo}, + null, null, 0, 0, null, new String[0], null)); + PartitionDesc partitionDesc = new PartitionDesc(); + partitionDesc.setTableDesc(tableDesc); + mapWork.addPathToPartitionInfo(path.getParent(), partitionDesc); + Utilities.setMapWork(job, mapWork); + return job; + } + + // ---- expectations ---- + + /** "startRowInFile:size" of every batch of a file whose stripes are read whole. */ + private static List expectedBatches(int stride, int[] stripeRows) { + List expected = new ArrayList<>(); + long base = 0; + for (int rows : stripeRows) { + for (int rgStart = 0; rgStart < rows; rgStart += stride) { + int rgRows = Math.min(stride, rows - rgStart); + for (int off = 0; off < rgRows; off += VectorizedRowBatch.DEFAULT_SIZE) { + expected.add((base + rgStart + off) + ":" + Math.min(VectorizedRowBatch.DEFAULT_SIZE, rgRows - off)); + } + } + base += rows; + } + return expected; + } + + private static List startsWithin(long stripeStart, int stripeRows) { + List starts = new ArrayList<>(); + for (int off = 0; off < stripeRows; off += VectorizedRowBatch.DEFAULT_SIZE) { + starts.add(stripeStart + off); + } + return starts; + } + + private static List render(List batches) { + List rendered = new ArrayList<>(); + for (long[] batch : batches) { + rendered.add(batch[0] + ":" + batch[1]); + } + return rendered; + } + + private static int countStartingAt(List batches, long start) { + int count = 0; + for (long[] batch : batches) { + if (batch[0] == start) { + ++count; + } + } + return count; + } + + private static int indexOfStart(List batches, long start) { + for (int i = 0; i < batches.size(); ++i) { + if (batches.get(i)[0] == start) { + return i; + } + } + throw new AssertionError("no batch starts at " + start); + } +} From fda86e88994cc3e63df0481ee986c024f2f9f18b Mon Sep 17 00:00:00 2001 From: Denys Kuzmenko Date: Thu, 29 Oct 2026 00:06:08 +0200 Subject: [PATCH 2/4] HIVE-30102: Review comments - llap_iceberg_read_orc_deletes.q: V2 position deletes and V3 deletion vectors on an ORC file with three row groups, read through LLAP IO (LLAP IO counters in the golden) - The row-mode ORC reader forces orc.sarg.to.filter off too: Iceberg's OrcIterable takes the batch's file position before ORC can skip filtered batches - LlapRecordReader.getRowNumber returns Long.MIN_VALUE (unknown) instead of throwing - OrcEncodedDataConsumer: drop the redundant noIndex check - TestOrcEncodedDataConsumerRowPositions: drop redundant tests and asserts; a read without columns replaces the faked ALL_RGS case - HiveVectorizedReader comments --- .../mr/hive/vector/HiveVectorizedReader.java | 5 +- .../mr/mapreduce/IcebergRecordReader.java | 3 + .../iceberg/mr/hive/TestHiveIcebergCRUD.java | 26 ++++ .../positive/llap_iceberg_read_orc_deletes.q | 26 ++++ .../llap/llap_iceberg_read_orc_deletes.q.out | 138 +++++++++++++++++ .../resources/testconfiguration.properties | 2 + .../llap/io/api/impl/ColumnVectorBatch.java | 4 +- .../llap/io/api/impl/LlapRecordReader.java | 4 +- .../io/decode/OrcEncodedDataConsumer.java | 7 +- ...estOrcEncodedDataConsumerRowPositions.java | 144 ++++-------------- 10 files changed, 237 insertions(+), 122 deletions(-) create mode 100644 iceberg/iceberg-handler/src/test/queries/positive/llap_iceberg_read_orc_deletes.q create mode 100644 iceberg/iceberg-handler/src/test/results/positive/llap/llap_iceberg_read_orc_deletes.q.out diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vector/HiveVectorizedReader.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vector/HiveVectorizedReader.java index 9ae1bc2b1f30..99a609a36a46 100644 --- a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vector/HiveVectorizedReader.java +++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vector/HiveVectorizedReader.java @@ -219,7 +219,7 @@ private static RecordReader orcRecordReader(Jo // Need to turn positional schema evolution off since we use column name based schema evolution for projection // and Iceberg will make a mapping between the file schema and the current reading schema. job.setBoolean(OrcConf.FORCE_POSITIONAL_EVOLUTION.getHiveConfName(), false); - // Row positions are derived from the batch index, so the reader must not filter rows within a batch. + // Row positions are derived from the batch index, and ORC's row-level filter can drop rows and whole batches. job.setBoolean(OrcConf.ALLOW_SARG_TO_FILTER.getHiveConfName(), false); // Metadata information has to be passed along in the OrcSplit. Without specifying this, the vectorized @@ -230,8 +230,7 @@ private static RecordReader orcRecordReader(Jo VectorizedReadUtils.handleIcebergProjection(task, job, VectorizedReadUtils.deserializeToShadedOrcTail(serializedOrcTail).getSchema(), residual); - // If LLAP enabled, try to retrieve an LLAP record reader - this might yield to null in some special cases. - // Position deletes and the virtual columns are served from the row positions the reader reports. + // If LLAP enabled, try to retrieve an LLAP record reader - this might yield to null in some special cases if (HiveConf.getBoolVar(job, HiveConf.ConfVars.LLAP_IO_ENABLED, LlapProxy.isDaemon()) && LlapProxy.getIo() != null) { recordReader = LlapProxy.getIo().orc().vectorizedReaderForPath(fileId, path, null, readColumnIds, diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergRecordReader.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergRecordReader.java index 2ea261f20141..a19a1fba2992 100644 --- a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergRecordReader.java +++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergRecordReader.java @@ -34,6 +34,7 @@ import org.apache.hadoop.hive.ql.io.IOContextAwareRecordReader; import org.apache.hadoop.mapreduce.InputSplit; import org.apache.hadoop.mapreduce.TaskAttemptContext; +import org.apache.hive.iceberg.org.apache.orc.OrcConf; import org.apache.iceberg.DataFile; import org.apache.iceberg.DataTask; import org.apache.iceberg.FileFormat; @@ -264,6 +265,8 @@ private CloseableIterable newOrcIterable( .filter(residual) .caseSensitive(isCaseSensitive()) .split(task.start(), task.length()); + // Row positions are derived from the batch index, and ORC's row-level filter can drop rows and whole batches. + orcReadBuilder.config(OrcConf.ALLOW_SARG_TO_FILTER.getHiveConfName(), "false"); if (getNameMapping() != null) { orcReadBuilder.withNameMapping(NameMappingParser.fromJson(getNameMapping())); diff --git a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergCRUD.java b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergCRUD.java index e3236fe2c3e5..ee4d285a9ab8 100644 --- a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergCRUD.java +++ b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergCRUD.java @@ -215,6 +215,32 @@ public void testReadAndWriteFormatV2Partitioned_PosDelete() throws IOException { Assert.assertArrayEquals(new Object[] {2L, "Trudy", "Pink"}, objects.get(3)); } + /** + * ORC can filter rows by the pushed down predicate; in row mode, row positions and deletes must still apply to + * the right rows. + */ + @Test + public void testPositionDeleteWithOrcRowFiltering() { + Assume.assumeTrue(fileFormat == FileFormat.ORC && formatVersion >= 2 && !isVectorized); + + Schema schema = new Schema(optional(1, "customer_id", Types.LongType.get())); + List records = TestHelper.generateRandomRecords(schema, 30000, 0L); + for (int i = 0; i < records.size(); ++i) { + records.get(i).setField("customer_id", (long) i); + } + testTables.createTable(shell, "customers", schema, PartitionSpec.unpartitioned(), fileFormat, records, + formatVersion); + shell.executeStatement("DELETE FROM customers WHERE customer_id = 25000"); + + shell.setHiveSessionValue("orc.sarg.to.filter", true); + shell.setHiveSessionValue("orc.filter.use.selected", true); + List result = shell.executeStatement( + "SELECT customer_id, ROW__POSITION FROM customers WHERE customer_id IN (25000, 25001)"); + + Assert.assertEquals(1, result.size()); + Assert.assertArrayEquals(new Object[] {25001L, 25001L}, result.get(0)); + } + /** * Tests reading/consuming legacy Iceberg V2 tables containing position delete files with embedded * row data. Iceberg 1.12.0 removed writer support for row data in position deletes, but Hive diff --git a/iceberg/iceberg-handler/src/test/queries/positive/llap_iceberg_read_orc_deletes.q b/iceberg/iceberg-handler/src/test/queries/positive/llap_iceberg_read_orc_deletes.q new file mode 100644 index 000000000000..bc37f0a23c4c --- /dev/null +++ b/iceberg/iceberg-handler/src/test/queries/positive/llap_iceberg_read_orc_deletes.q @@ -0,0 +1,26 @@ +-- ORC files with position deletes are read through LLAP IO (see the LLAP IO counters): the deletes and ROW__POSITION +-- apply to the right rows in a later row group and across a batch boundary (the third row group starts at file row +-- 20000, its sixth batch at 25120) +set hive.llap.io.enabled=true; +set hive.vectorized.execution.enabled=true; +set hive.fetch.task.conversion=none; +set hive.optimize.index.filter=true; + +-- V2: position delete files +CREATE EXTERNAL TABLE ice_orc_v2 (id INT) STORED BY ICEBERG STORED AS ORC +TBLPROPERTIES ('format-version'='2'); +INSERT INTO ice_orc_v2 SELECT pos FROM (SELECT posexplode(split(space(29999), ' ')) AS (pos, val)) t; +DELETE FROM ice_orc_v2 WHERE id IN (25119, 25121); + +-- V3: deletion vectors +CREATE EXTERNAL TABLE ice_orc_v3 (id INT) STORED BY ICEBERG STORED AS ORC +TBLPROPERTIES ('format-version'='3'); +INSERT INTO ice_orc_v3 SELECT pos FROM (SELECT posexplode(split(space(29999), ' ')) AS (pos, val)) t; +DELETE FROM ice_orc_v3 WHERE id IN (25119, 25121); + +SET hive.exec.post.hooks=org.apache.hadoop.hive.ql.hooks.PostExecTezSummaryPrinter; +SELECT id, ROW__POSITION FROM ice_orc_v2 WHERE id BETWEEN 25117 AND 25123 ORDER BY id; +SELECT id, ROW__POSITION FROM ice_orc_v3 WHERE id BETWEEN 25117 AND 25123 ORDER BY id; + +DROP TABLE ice_orc_v2; +DROP TABLE ice_orc_v3; diff --git a/iceberg/iceberg-handler/src/test/results/positive/llap/llap_iceberg_read_orc_deletes.q.out b/iceberg/iceberg-handler/src/test/results/positive/llap/llap_iceberg_read_orc_deletes.q.out new file mode 100644 index 000000000000..d49ae17cb097 --- /dev/null +++ b/iceberg/iceberg-handler/src/test/results/positive/llap/llap_iceberg_read_orc_deletes.q.out @@ -0,0 +1,138 @@ +PREHOOK: query: CREATE EXTERNAL TABLE ice_orc_v2 (id INT) STORED BY ICEBERG STORED AS ORC +TBLPROPERTIES ('format-version'='2') +PREHOOK: type: CREATETABLE +PREHOOK: Output: database:default +PREHOOK: Output: default@ice_orc_v2 +POSTHOOK: query: CREATE EXTERNAL TABLE ice_orc_v2 (id INT) STORED BY ICEBERG STORED AS ORC +TBLPROPERTIES ('format-version'='2') +POSTHOOK: type: CREATETABLE +POSTHOOK: Output: database:default +POSTHOOK: Output: default@ice_orc_v2 +PREHOOK: query: INSERT INTO ice_orc_v2 SELECT pos FROM (SELECT posexplode(split(space(29999), ' ')) AS (pos, val)) t +PREHOOK: type: QUERY +PREHOOK: Input: _dummy_database@_dummy_table +PREHOOK: Output: default@ice_orc_v2 +POSTHOOK: query: INSERT INTO ice_orc_v2 SELECT pos FROM (SELECT posexplode(split(space(29999), ' ')) AS (pos, val)) t +POSTHOOK: type: QUERY +POSTHOOK: Input: _dummy_database@_dummy_table +POSTHOOK: Output: default@ice_orc_v2 +PREHOOK: query: DELETE FROM ice_orc_v2 WHERE id IN (25119, 25121) +PREHOOK: type: QUERY +PREHOOK: Input: default@ice_orc_v2 +PREHOOK: Output: default@ice_orc_v2 +POSTHOOK: query: DELETE FROM ice_orc_v2 WHERE id IN (25119, 25121) +POSTHOOK: type: QUERY +POSTHOOK: Input: default@ice_orc_v2 +POSTHOOK: Output: default@ice_orc_v2 +PREHOOK: query: CREATE EXTERNAL TABLE ice_orc_v3 (id INT) STORED BY ICEBERG STORED AS ORC +TBLPROPERTIES ('format-version'='3') +PREHOOK: type: CREATETABLE +PREHOOK: Output: database:default +PREHOOK: Output: default@ice_orc_v3 +POSTHOOK: query: CREATE EXTERNAL TABLE ice_orc_v3 (id INT) STORED BY ICEBERG STORED AS ORC +TBLPROPERTIES ('format-version'='3') +POSTHOOK: type: CREATETABLE +POSTHOOK: Output: database:default +POSTHOOK: Output: default@ice_orc_v3 +PREHOOK: query: INSERT INTO ice_orc_v3 SELECT pos FROM (SELECT posexplode(split(space(29999), ' ')) AS (pos, val)) t +PREHOOK: type: QUERY +PREHOOK: Input: _dummy_database@_dummy_table +PREHOOK: Output: default@ice_orc_v3 +POSTHOOK: query: INSERT INTO ice_orc_v3 SELECT pos FROM (SELECT posexplode(split(space(29999), ' ')) AS (pos, val)) t +POSTHOOK: type: QUERY +POSTHOOK: Input: _dummy_database@_dummy_table +POSTHOOK: Output: default@ice_orc_v3 +PREHOOK: query: DELETE FROM ice_orc_v3 WHERE id IN (25119, 25121) +PREHOOK: type: QUERY +PREHOOK: Input: default@ice_orc_v3 +PREHOOK: Output: default@ice_orc_v3 +POSTHOOK: query: DELETE FROM ice_orc_v3 WHERE id IN (25119, 25121) +POSTHOOK: type: QUERY +POSTHOOK: Input: default@ice_orc_v3 +POSTHOOK: Output: default@ice_orc_v3 +PREHOOK: query: SELECT id, ROW__POSITION FROM ice_orc_v2 WHERE id BETWEEN 25117 AND 25123 ORDER BY id +PREHOOK: type: QUERY +PREHOOK: Input: default@ice_orc_v2 +#### A masked pattern was here #### +Stage-1 FILE SYSTEM COUNTERS: +Stage-1 HIVE COUNTERS: + CREATED_FILES: 1 + DESERIALIZE_ERRORS: 0 + RECORDS_IN_Map_1: 9998 + RECORDS_OUT_0: 5 + RECORDS_OUT_INTERMEDIATE_Map_1: 5 + RECORDS_OUT_INTERMEDIATE_Reducer_2: 0 + RECORDS_OUT_OPERATOR_FIL_7: 5 + RECORDS_OUT_OPERATOR_FS_11: 5 + RECORDS_OUT_OPERATOR_MAP_0: 0 + RECORDS_OUT_OPERATOR_RS_9: 5 + RECORDS_OUT_OPERATOR_SEL_10: 5 + RECORDS_OUT_OPERATOR_SEL_8: 5 + RECORDS_OUT_OPERATOR_TS_0: 9998 + TOTAL_TABLE_ROWS_WRITTEN: 0 +Stage-1 LLAP IO COUNTERS: + CACHE_HIT_BYTES: 229 + CACHE_MISS_BYTES: 0 + METADATA_CACHE_HIT: 2 + NUM_DECODED_BATCHES: 1 + NUM_VECTOR_BATCHES: 10 + ROWS_EMITTED: 10000 + SELECTED_ROWGROUPS: 1 +Stage-1 INPUT COUNTERS: + GROUPED_INPUT_SPLITS_Map_1: 1 + INPUT_DIRECTORIES_Map_1: 1 + INPUT_FILES_Map_1: 1 + RAW_INPUT_SPLITS_Map_1: 1 +25117 25117 +25118 25118 +25120 25120 +25122 25122 +25123 25123 +PREHOOK: query: SELECT id, ROW__POSITION FROM ice_orc_v3 WHERE id BETWEEN 25117 AND 25123 ORDER BY id +PREHOOK: type: QUERY +PREHOOK: Input: default@ice_orc_v3 +#### A masked pattern was here #### +Stage-1 FILE SYSTEM COUNTERS: +Stage-1 HIVE COUNTERS: + CREATED_FILES: 1 + DESERIALIZE_ERRORS: 0 + RECORDS_IN_Map_1: 9998 + RECORDS_OUT_0: 5 + RECORDS_OUT_INTERMEDIATE_Map_1: 5 + RECORDS_OUT_INTERMEDIATE_Reducer_2: 0 + RECORDS_OUT_OPERATOR_FIL_7: 5 + RECORDS_OUT_OPERATOR_FS_11: 5 + RECORDS_OUT_OPERATOR_MAP_0: 0 + RECORDS_OUT_OPERATOR_RS_9: 5 + RECORDS_OUT_OPERATOR_SEL_10: 5 + RECORDS_OUT_OPERATOR_SEL_8: 5 + RECORDS_OUT_OPERATOR_TS_0: 9998 + TOTAL_TABLE_ROWS_WRITTEN: 0 +Stage-1 LLAP IO COUNTERS: + CACHE_HIT_BYTES: 229 + CACHE_MISS_BYTES: 0 + METADATA_CACHE_HIT: 2 + NUM_DECODED_BATCHES: 1 + NUM_VECTOR_BATCHES: 10 + ROWS_EMITTED: 10000 + SELECTED_ROWGROUPS: 1 +Stage-1 INPUT COUNTERS: + GROUPED_INPUT_SPLITS_Map_1: 1 + INPUT_DIRECTORIES_Map_1: 1 + INPUT_FILES_Map_1: 1 + RAW_INPUT_SPLITS_Map_1: 1 +25117 25117 +25118 25118 +25120 25120 +25122 25122 +25123 25123 +PREHOOK: query: DROP TABLE ice_orc_v2 +PREHOOK: type: DROPTABLE +PREHOOK: Input: default@ice_orc_v2 +PREHOOK: Output: database:default +PREHOOK: Output: default@ice_orc_v2 +PREHOOK: query: DROP TABLE ice_orc_v3 +PREHOOK: type: DROPTABLE +PREHOOK: Input: default@ice_orc_v3 +PREHOOK: Output: database:default +PREHOOK: Output: default@ice_orc_v3 diff --git a/itests/src/test/resources/testconfiguration.properties b/itests/src/test/resources/testconfiguration.properties index e375f4f16e78..8de15596c5d6 100644 --- a/itests/src/test/resources/testconfiguration.properties +++ b/itests/src/test/resources/testconfiguration.properties @@ -423,6 +423,7 @@ iceberg.llap.query.files=\ iceberg_merge_files.q,\ iceberg_row_mode_virtual_columns.q,\ llap_iceberg_read_orc.q,\ + llap_iceberg_read_orc_deletes.q,\ llap_iceberg_read_parquet.q,\ puffin_col_stats_with_time_travel.q,\ vectorized_iceberg_read_mixed.q,\ @@ -483,6 +484,7 @@ iceberg.llap.only.query.files=\ iceberg_merge_files.q,\ iceberg_row_mode_virtual_columns.q,\ llap_iceberg_read_orc.q,\ + llap_iceberg_read_orc_deletes.q,\ llap_iceberg_read_parquet.q,\ puffin_col_stats_with_time_travel.q diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/ColumnVectorBatch.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/ColumnVectorBatch.java index a675e37cf01b..8360ed452991 100644 --- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/ColumnVectorBatch.java +++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/ColumnVectorBatch.java @@ -34,8 +34,8 @@ public class ColumnVectorBatch { public MutableFilterContext filterContext; public ColumnVector[] cols; public int size; - /** File row position of the first row; -1 when the producer does not track positions. */ - public long startRowInFile = -1; + /** File row position of the first row; Long.MIN_VALUE when the producer does not track positions. */ + public long startRowInFile = Long.MIN_VALUE; public ColumnVectorBatch(int columnCount) { this(columnCount, VectorizedRowBatch.DEFAULT_SIZE); 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 6db6cc854495..bd5b0130dd1a 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 @@ -583,11 +583,9 @@ ColumnVectorBatch nextCvb() throws InterruptedException, IOException { return lastCvb; } + /** Long.MIN_VALUE (unknown) when the producer does not track row positions. */ @Override public long getRowNumber() { - if (lastCvb.startRowInFile < 0) { - throw new IllegalStateException("Row positions are not provided by " + rp.getClass().getSimpleName()); - } return lastCvb.startRowInFile; } diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcEncodedDataConsumer.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcEncodedDataConsumer.java index d00a7d93b2e0..0702adf3207f 100644 --- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcEncodedDataConsumer.java +++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcEncodedDataConsumer.java @@ -144,8 +144,8 @@ protected void decodeBatch(OrcEncodedColumnBatch batch, } int maxBatchesRG = (int) ((nonNullRowCount / VectorizedRowBatch.DEFAULT_SIZE) + 1); int batchSize = VectorizedRowBatch.DEFAULT_SIZE; - long rgStartRow = stripeStartRows == null ? -1 : stripeStartRows[currentStripeIndex] - + ((rgIdx == OrcEncodedColumnBatch.ALL_RGS || noIndex) ? 0 : (long) rgIdx * fileMetadata.getRowIndexStride()); + long rgStartRow = stripeStartRows == null ? Long.MIN_VALUE : stripeStartRows[currentStripeIndex] + + (rgIdx == OrcEncodedColumnBatch.ALL_RGS ? 0 : (long) rgIdx * fileMetadata.getRowIndexStride()); TypeDescription fileSchema = fileMetadata.getSchema(); if (columnReaders == null || !sameStripe || noIndex) { @@ -164,7 +164,8 @@ protected void decodeBatch(OrcEncodedColumnBatch batch, ColumnVectorBatch cvb = takeBatch(batchSize); // assert cvb.cols.length == batch.getColumnIxs().length; // Must be constant per split. - cvb.startRowInFile = rgStartRow < 0 ? -1 : rgStartRow + (long) i * VectorizedRowBatch.DEFAULT_SIZE; + cvb.startRowInFile = stripeStartRows == null + ? Long.MIN_VALUE : rgStartRow + (long) i * VectorizedRowBatch.DEFAULT_SIZE; for (int idx = 0; idx < columnReaders.length; ++idx) { /* * Currently, ORC's TreeReaderFactory class does this: diff --git a/llap-server/src/test/org/apache/hadoop/hive/llap/io/decode/TestOrcEncodedDataConsumerRowPositions.java b/llap-server/src/test/org/apache/hadoop/hive/llap/io/decode/TestOrcEncodedDataConsumerRowPositions.java index 45cf1385db6a..fca0a3c7d301 100644 --- a/llap-server/src/test/org/apache/hadoop/hive/llap/io/decode/TestOrcEncodedDataConsumerRowPositions.java +++ b/llap-server/src/test/org/apache/hadoop/hive/llap/io/decode/TestOrcEncodedDataConsumerRowPositions.java @@ -21,9 +21,6 @@ import static org.apache.hadoop.hive.metastore.api.hive_metastoreConstants.META_TABLE_NAME; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertTrue; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.when; import java.io.IOException; import java.nio.file.Files; @@ -82,8 +79,8 @@ /** * Absolute file row positions reported by {@link OrcEncodedDataConsumer} on every emitted batch. - * The indexed and index-disabled files are read end to end through LLAP IO; the ALL_RGS and the - * position-less (SerDe-like) metadata cases are driven straight into decodeBatch. + * ORC files are read end to end through LLAP IO; the position-less (SerDe-like) metadata case is + * driven straight into decodeBatch. */ public class TestOrcEncodedDataConsumerRowPositions { @@ -92,6 +89,7 @@ public class TestOrcEncodedDataConsumerRowPositions { // Neither a multiple of the row index stride nor of VectorizedRowBatch.DEFAULT_SIZE. private static final int[] STRIPE_ROWS = {8300, 6100, 3700}; private static final int[] NO_INDEX_STRIPE_ROWS = {5300, 2700}; + private static final List ALL_COLUMNS = Arrays.asList(0, 1); private static final LlapDaemonIOMetrics IO_METRICS = LlapDaemonIOMetrics.create("TestOrcEncodedDataConsumerRowPositions-io", "1", null); @@ -124,69 +122,40 @@ public static void tearDownClass() { @Test public void testRowGroupPositionsAcrossStripes() throws Exception { - List batches = readThroughLlap(indexedFile); + List batches = readThroughLlap(indexedFile, ALL_COLUMNS); assertEquals(expectedBatches(STRIDE, STRIPE_ROWS), render(batches)); - // First RG of stripe 0, a middle RG of stripe 0, first RG of stripe 1, last RG of the last stripe. - for (long rgStart : new long[] {0, 5000, 8300, 16900}) { - assertEquals("no batch starts at " + rgStart, 1, countStartingAt(batches, rgStart)); - assertGroundTruth(rgStart); - } - for (long[] batch : batches) { - assertEquals("id of the first row of the batch at " + batch[0], batch[0], batch[2]); - } - } - - @Test - public void testBatchesAdvanceWithinRowGroup() throws Exception { - List batches = readThroughLlap(indexedFile); - - // The row group at file row 8300 holds 2500 rows: 1024 + 1024 + 452. - List rg = batches.subList(indexOfStart(batches, 8300), indexOfStart(batches, 8300) + 3); - assertEquals(Arrays.asList("8300:1024", "9324:1024", "10348:452"), render(rg)); - assertEquals(8300 + STRIDE, rg.get(2)[0] + rg.get(2)[1]); + assertFirstRowIds(batches); } @Test public void testIndexDisabledFileStartsAtStripeBoundary() throws Exception { - List batches = readThroughLlap(noIndexFile); + List batches = readThroughLlap(noIndexFile, ALL_COLUMNS); assertEquals(expectedBatches(Integer.MAX_VALUE, NO_INDEX_STRIPE_ROWS), render(batches)); - assertEquals(1, countStartingAt(batches, NO_INDEX_STRIPE_ROWS[0])); - for (long[] batch : batches) { - assertEquals("id of the first row of the batch at " + batch[0], batch[0], batch[2]); - } + assertFirstRowIds(batches); } + /** Without columns, the reader decodes each stripe as one ALL_RGS batch. */ @Test - public void testAllRowGroupsBatchStartsAtStripeBoundary() throws Exception { - List starts = decodeStripe(fileMetadata(STRIPE_ROWS, STRIDE), 1, OrcEncodedColumnBatch.ALL_RGS, true); + public void testNoColumnReadStartsAtStripeBoundary() throws Exception { + List batches = readThroughLlap(indexedFile, Collections.emptyList()); - assertEquals(startsWithin(STRIPE_ROWS[0], STRIPE_ROWS[1]), starts); - } - - @Test - public void testIndexDisabledStripeStartsAtStripeBoundary() throws Exception { - List starts = decodeStripe(fileMetadata(STRIPE_ROWS, STRIDE), 1, 0, false); - - assertEquals(startsWithin(STRIPE_ROWS[0], STRIPE_ROWS[1]), starts); + assertEquals(expectedBatches(Integer.MAX_VALUE, STRIPE_ROWS), render(batches)); } @Test public void testNoStripesReportsNoPosition() throws Exception { - ConsumerFileMetadata noPositions = fileMetadata(new int[0], 0); - assertTrue(noPositions.getStripes().isEmpty()); - - List starts = decodeStripe(noPositions, 0, OrcEncodedColumnBatch.ALL_RGS, true); + List starts = decodeStripe(noStripesMetadata(), 2 * VectorizedRowBatch.DEFAULT_SIZE + 1); - assertEquals(Collections.nCopies(startsWithin(0, STRIPE_ROWS[0]).size(), -1L), starts); + assertEquals(Collections.nCopies(3, Long.MIN_VALUE), starts); } // ---- LLAP IO end to end ---- /** One {startRowInFile, size, id of the first row} per batch, in the order LLAP returned them. */ - private static List readThroughLlap(Path path) throws Exception { - RecordReader reader = llapReader(path); + private static List readThroughLlap(Path path, List columnIds) throws Exception { + RecordReader reader = llapReader(path, columnIds); List batches = new ArrayList<>(); try { VectorizedRowBatch vrb = reader.createValue(); @@ -200,40 +169,35 @@ private static List readThroughLlap(Path path) throws Exception { return batches; } - private static RecordReader llapReader(Path path) throws IOException { + private static RecordReader llapReader(Path path, List columnIds) + throws IOException { JobConf job = jobConf(path); long fileLen = path.getFileSystem(job).getFileStatus(path).getLen(); RecordReader reader = LlapProxy.getIo().orc().vectorizedReaderForPath( - null, path, null, Arrays.asList(0, 1), job, 0, fileLen, Reporter.NULL); + null, path, null, columnIds, job, 0, fileLen, Reporter.NULL); assertNotNull("LLAP should handle this ORC read", reader); return reader; } - private static void assertGroundTruth(long fileRow) throws IOException { - try (org.apache.orc.Reader reader = OrcFile.createReader(indexedFile, OrcFile.readerOptions(daemonConf)); - org.apache.orc.RecordReader rows = reader.rows()) { - VectorizedRowBatch batch = SCHEMA.createRowBatch(1); - rows.seekToRow(fileRow); - assertTrue("file row " + fileRow + " is past the end", rows.nextBatch(batch)); - assertEquals(fileRow, ((LongColumnVector) batch.cols[0]).vector[0]); - assertEquals("n" + fileRow, ((BytesColumnVector) batch.cols[1]).toString(0)); + private static void assertFirstRowIds(List batches) { + for (long[] batch : batches) { + assertEquals("id of the first row of the batch at " + batch[0], batch[0], batch[2]); } } // ---- decodeBatch with faked metadata ---- - /** Row positions of the batches decodeBatch emits for one stripe of a zero-column read. */ - private static List decodeStripe(ConsumerFileMetadata fileMetadata, int stripeIx, int rgIx, - boolean withRowIndex) throws Exception { + /** Row positions of the batches decodeBatch emits for the first stripe of a zero-column read. */ + private static List decodeStripe(ConsumerFileMetadata fileMetadata, int stripeRows) throws Exception { QueryFragmentCounters counters = new QueryFragmentCounters(daemonConf, new TezCounters()); List starts = new ArrayList<>(); OrcEncodedDataConsumer consumer = new OrcEncodedDataConsumer(null, includes(), counters, IO_METRICS); consumer.init(null, null, new IoTrace(0, false)); consumer.setFileMetadata(fileMetadata); - consumer.setStripeMetadata(new FakeStripeMetadata(stripeIx, STRIPE_ROWS[stripeIx], withRowIndex)); + consumer.setStripeMetadata(new FakeStripeMetadata(stripeRows)); OrcEncodedColumnBatch batch = new OrcEncodedColumnBatch(); - batch.init(null, stripeIx, rgIx, 0); + batch.init(null, 0, OrcEncodedColumnBatch.ALL_RGS, 0); consumer.decodeBatch(batch, new Consumer() { @Override public void consumeData(ColumnVectorBatch cvb) { starts.add(cvb.startRowInFile); @@ -249,16 +213,10 @@ private static List decodeStripe(ConsumerFileMetadata fileMetadata, int st return starts; } - private static ConsumerFileMetadata fileMetadata(int[] stripeRows, int stride) { - List stripes = new ArrayList<>(); - for (int rows : stripeRows) { - StripeInformation stripe = mock(StripeInformation.class); - when(stripe.getNumberOfRows()).thenReturn((long) rows); - stripes.add(stripe); - } + private static ConsumerFileMetadata noStripesMetadata() { return new ConsumerFileMetadata() { @Override public int getStripeCount() { - return stripes.size(); + return 0; } @Override public CompressionKind getCompressionKind() { @@ -282,29 +240,25 @@ private static ConsumerFileMetadata fileMetadata(int[] stripeRows, int stride) { } @Override public List getStripes() { - return stripes; + return Collections.emptyList(); } @Override public int getRowIndexStride() { - return stride; + return 0; } }; } - /** Reports a row count only; a null row index entry is what an index-disabled stripe returns. */ + /** Reports a row count only. */ private static final class FakeStripeMetadata implements ConsumerStripeMetadata { - private final int stripeIx; private final long rowCount; - private final boolean withRowIndex; - FakeStripeMetadata(int stripeIx, long rowCount, boolean withRowIndex) { - this.stripeIx = stripeIx; + FakeStripeMetadata(long rowCount) { this.rowCount = rowCount; - this.withRowIndex = withRowIndex; } @Override public int getStripeIx() { - return stripeIx; + return 0; } @Override public long getRowCount() { @@ -329,7 +283,7 @@ private static final class FakeStripeMetadata implements ConsumerStripeMetadata } @Override public boolean supportsRowIndexes() { - return withRowIndex; + return false; } } @@ -420,11 +374,6 @@ private static void assertLayout(Path path, int stride, int[] stripeRows) throws assertEquals(stripeRows.length, stripes.size()); for (int i = 0; i < stripeRows.length; ++i) { assertEquals("stripe " + i, stripeRows[i], stripes.get(i).getNumberOfRows()); - if (stride > 0) { - assertTrue("stripe " + i + " is a multiple of the stride", stripeRows[i] % stride != 0); - } - assertTrue("stripe " + i + " is a multiple of the batch size", - stripeRows[i] % VectorizedRowBatch.DEFAULT_SIZE != 0); } } } @@ -472,14 +421,6 @@ private static List expectedBatches(int stride, int[] stripeRows) { return expected; } - private static List startsWithin(long stripeStart, int stripeRows) { - List starts = new ArrayList<>(); - for (int off = 0; off < stripeRows; off += VectorizedRowBatch.DEFAULT_SIZE) { - starts.add(stripeStart + off); - } - return starts; - } - private static List render(List batches) { List rendered = new ArrayList<>(); for (long[] batch : batches) { @@ -487,23 +428,4 @@ private static List render(List batches) { } return rendered; } - - private static int countStartingAt(List batches, long start) { - int count = 0; - for (long[] batch : batches) { - if (batch[0] == start) { - ++count; - } - } - return count; - } - - private static int indexOfStart(List batches, long start) { - for (int i = 0; i < batches.size(); ++i) { - if (batches.get(i)[0] == start) { - return i; - } - } - throw new AssertionError("no batch starts at " + start); - } } From 51413af1ea789d113e25b5b1aab834a118c1aafe Mon Sep 17 00:00:00 2001 From: Denys Kuzmenko Date: Thu, 29 Oct 2026 12:33:41 +0200 Subject: [PATCH 3/4] HIVE-30102: Address Sonar issues - ColumnVectorBatch: startRowInFile is private, with accessors - LlapRecordReader is final - OrcEncodedDataConsumer: row group start row without a nested ternary - TestOrcEncodedDataConsumerRowPositions: Hive's ASF header; SerDeStripeMetadata and Mockito mocks instead of hand-written fakes; one row batch per stripe in the fixture - IcebergRecordReader: sarg-to-filter set in the ORC read builder chain, next to the filter - IDE inspection cleanups in the touched classes --- .../mr/mapreduce/IcebergRecordReader.java | 4 +- .../llap/io/api/impl/ColumnVectorBatch.java | 20 +- .../llap/io/api/impl/LlapRecordReader.java | 76 +++--- .../io/decode/OrcEncodedDataConsumer.java | 38 +-- ...estOrcEncodedDataConsumerRowPositions.java | 250 +++++------------- 5 files changed, 132 insertions(+), 256 deletions(-) diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergRecordReader.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergRecordReader.java index a19a1fba2992..52faaf202305 100644 --- a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergRecordReader.java +++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergRecordReader.java @@ -263,10 +263,10 @@ private CloseableIterable newOrcIterable( ORC.ReadBuilder orcReadBuilder = ORC.read(inputFile) .project(readSchemaWithoutConstantAndMetadataFields) .filter(residual) + // Row positions are derived from the batch index, and ORC's row-level filter can drop rows and whole batches. + .config(OrcConf.ALLOW_SARG_TO_FILTER.getHiveConfName(), "false") .caseSensitive(isCaseSensitive()) .split(task.start(), task.length()); - // Row positions are derived from the batch index, and ORC's row-level filter can drop rows and whole batches. - orcReadBuilder.config(OrcConf.ALLOW_SARG_TO_FILTER.getHiveConfName(), "false"); if (getNameMapping() != null) { orcReadBuilder.withNameMapping(NameMappingParser.fromJson(getNameMapping())); diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/ColumnVectorBatch.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/ColumnVectorBatch.java index 8360ed452991..fe1f041effec 100644 --- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/ColumnVectorBatch.java +++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/ColumnVectorBatch.java @@ -35,7 +35,7 @@ public class ColumnVectorBatch { public ColumnVector[] cols; public int size; /** File row position of the first row; Long.MIN_VALUE when the producer does not track positions. */ - public long startRowInFile = Long.MIN_VALUE; + private long startRowInFile = Long.MIN_VALUE; public ColumnVectorBatch(int columnCount) { this(columnCount, VectorizedRowBatch.DEFAULT_SIZE); @@ -53,6 +53,13 @@ public void swapColumnVector(int ix, ColumnVector[] other, int otherIx) { cols[ix] = old; } + public long getStartRowInFile() { + return startRowInFile; + } + + public void setStartRowInFile(long startRowInFile) { + this.startRowInFile = startRowInFile; + } @Override public String toString() { @@ -78,7 +85,6 @@ public String toString() { } b.append('\n'); - for (int i = 0; i < size; i++) { b.append('['); for (int k = 0; k < cols.length; k++) { @@ -87,12 +93,10 @@ public String toString() { b.append(", "); } if (cv == null) continue; - if (cv != null) { - try { - cv.stringifyValue(b, i); - } catch (Exception ex) { - b.append("invalid"); - } + try { + cv.stringifyValue(b, i); + } catch (Exception ex) { + b.append("invalid"); } } b.append(']'); 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 bd5b0130dd1a..80713688cd9f 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 @@ -86,7 +86,7 @@ import static org.apache.hadoop.hive.llap.LlapHiveUtils.throwIfCacheOnlyRead; -class LlapRecordReader implements RecordReader, Consumer, +final class LlapRecordReader implements RecordReader, Consumer, RowPositionAwareVectorizedRecordReader { private static final Logger LOG = LoggerFactory.getLogger(LlapRecordReader.class); @@ -98,7 +98,6 @@ class LlapRecordReader implements RecordReader private final SearchArgument sarg; private final VectorizedRowBatchCtx rbCtx; private final boolean isVectorized; - private final boolean probeDecodeEnabled; private VectorizedOrcAcidRowBatchReader acidReader; private Object[] partitionValues; private VectorizedRowBatch acidInputVrb; @@ -224,8 +223,8 @@ private LlapRecordReader(MapWork mapWork, JobConf job, FileSplit split, this.includes = new IncludesImpl(tableIncludedCols, isAcidFormat, rbCtx, schema, job, isAcidScan && acidReader.includeAcidColumns()); - this.probeDecodeEnabled = HiveConf.getBoolVar(jobConf, ConfVars.HIVE_OPTIMIZE_SCAN_PROBEDECODE); - if (this.probeDecodeEnabled) { + boolean probeDecodeEnabled = HiveConf.getBoolVar(jobConf, ConfVars.HIVE_OPTIMIZE_SCAN_PROBEDECODE); + if (probeDecodeEnabled) { includes.setProbeDecodeContext(mapWork.getProbeDecodeContext()); LOG.info("LlapRecordReader ProbeDecode is enabled"); } @@ -303,34 +302,33 @@ static int determineQueueLimit(long maxBufferedSize, } else { PrimitiveTypeInfo pti = (PrimitiveTypeInfo) ti; switch (pti.getPrimitiveCategory()) { - case BINARY: - case CHAR: - case VARCHAR: - case STRING: - colWeight = COL_WEIGHT_STRING; - break; + case BINARY: + case CHAR: + case VARCHAR: + case STRING: + colWeight = COL_WEIGHT_STRING; + break; //Timestamp column vector uses an int and long arrays - case TIMESTAMP: - case INTERVAL_DAY_TIME: - colWeight = 2; - break; - case DECIMAL: - boolean useDecimal64 = false; - if (ti instanceof DecimalTypeInfo) { - DecimalTypeInfo dti = (DecimalTypeInfo) ti; - if (dti.getPrecision() <= TypeDescription.MAX_DECIMAL64_PRECISION && decimal64Support) { - useDecimal64 = true; + case TIMESTAMP: + case INTERVAL_DAY_TIME: + colWeight = 2; + break; + case DECIMAL: + boolean useDecimal64 = false; + if (ti instanceof DecimalTypeInfo dti) { + if (dti.getPrecision() <= TypeDescription.MAX_DECIMAL64_PRECISION && decimal64Support) { + useDecimal64 = true; + } } - } - // decimal_64 column vectors gets the same weight as long column vectors - if (useDecimal64) { + // decimal_64 column vectors gets the same weight as long column vectors + if (useDecimal64) { + colWeight = 1; + } else { + colWeight = COL_WEIGHT_HIVEDECIMAL; + } + break; + default: colWeight = 1; - } else { - colWeight = COL_WEIGHT_HIVEDECIMAL; - } - break; - default: - colWeight = 1; } } totalWeight += colWeight * 8 * scale; @@ -502,7 +500,7 @@ private AcidWrapper(VectorizedRowBatch acidVrb) { } @Override - public boolean next(NullWritable key, VectorizedRowBatch value) throws IOException { + public boolean next(NullWritable key, VectorizedRowBatch value) { return true; } @@ -517,16 +515,16 @@ public VectorizedRowBatch createValue() { } @Override - public long getPos() throws IOException { + public long getPos() { return 0; } @Override - public void close() throws IOException { + public void close() { } @Override - public float getProgress() throws IOException { + public float getProgress() { return 0; } } @@ -586,7 +584,7 @@ ColumnVectorBatch nextCvb() throws InterruptedException, IOException { /** Long.MIN_VALUE (unknown) when the producer does not track row positions. */ @Override public long getRowNumber() { - return lastCvb.startRowInFile; + return lastCvb.getStartRowInFile(); } @Override @@ -660,7 +658,11 @@ public void setError(Throwable t) throws InterruptedException { private void enqueueInternal(Object o) throws InterruptedException { // We need to loop here to handle the case where consumer goes away. - do {} while (!isClosed && !isInterrupted && !queue.offer(o, 100, TimeUnit.MILLISECONDS)); + while (!isClosed && !isInterrupted) { + if (queue.offer(o, 100, TimeUnit.MILLISECONDS)) { + return; + } + } } @Override @@ -673,10 +675,6 @@ void setPartitionValues(Object[] partitionValues) { this.partitionValues = partitionValues; } - ReadPipeline getReadPipeline() { - return rp; - } - /** This class encapsulates include-related logic for LLAP readers. It is not actually specific * to LLAP IO but in LLAP IO in particular, I want to encapsulate all this mess for now until * we have smth better like Schema Evolution v2. This can also hypothetically encapsulate diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcEncodedDataConsumer.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcEncodedDataConsumer.java index 0702adf3207f..33986ffc248d 100644 --- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcEncodedDataConsumer.java +++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcEncodedDataConsumer.java @@ -70,7 +70,7 @@ public class OrcEncodedDataConsumer private TypeReader[] columnReaders; private int previousStripeIndex = -1; private ConsumerFileMetadata fileMetadata; // We assume one request is only for one file. - private long[] stripeStartRows; // file row of each stripe's first row; null without file positions + private long[] stripeStartRows; // row number in the file where each stripe starts; null without file positions private CompressionCodec codec; private List stripes; private SchemaEvolution evolution; @@ -144,8 +144,13 @@ protected void decodeBatch(OrcEncodedColumnBatch batch, } int maxBatchesRG = (int) ((nonNullRowCount / VectorizedRowBatch.DEFAULT_SIZE) + 1); int batchSize = VectorizedRowBatch.DEFAULT_SIZE; - long rgStartRow = stripeStartRows == null ? Long.MIN_VALUE : stripeStartRows[currentStripeIndex] - + (rgIdx == OrcEncodedColumnBatch.ALL_RGS ? 0 : (long) rgIdx * fileMetadata.getRowIndexStride()); + long rgStartRow = Long.MIN_VALUE; + if (stripeStartRows != null) { + rgStartRow = stripeStartRows[currentStripeIndex]; + if (rgIdx != OrcEncodedColumnBatch.ALL_RGS) { + rgStartRow += (long) rgIdx * fileMetadata.getRowIndexStride(); + } + } TypeDescription fileSchema = fileMetadata.getSchema(); if (columnReaders == null || !sameStripe || noIndex) { @@ -164,8 +169,8 @@ protected void decodeBatch(OrcEncodedColumnBatch batch, ColumnVectorBatch cvb = takeBatch(batchSize); // assert cvb.cols.length == batch.getColumnIxs().length; // Must be constant per split. - cvb.startRowInFile = stripeStartRows == null - ? Long.MIN_VALUE : rgStartRow + (long) i * VectorizedRowBatch.DEFAULT_SIZE; + cvb.setStartRowInFile(stripeStartRows == null ? + Long.MIN_VALUE : rgStartRow + (long) i * VectorizedRowBatch.DEFAULT_SIZE); for (int idx = 0; idx < columnReaders.length; ++idx) { /* * Currently, ORC's TreeReaderFactory class does this: @@ -313,11 +318,13 @@ private ColumnVector createColumn(TypeDescription type, int batchSize, final boo private void positionInStreams(TypeReader[] columnReaders, OrcBatchKey batchKey, ConsumerStripeMetadata stripeMetadata) throws IOException { PositionProvider[] pps = createPositionProviders(columnReaders, batchKey, stripeMetadata); - if (pps == null) return; - for (int i = 0; i < columnReaders.length; i++) { - if (columnReaders[i] == null) continue; + if (pps == null) { + return; + } + for (TypeReader columnReader : columnReaders) { + if (columnReader == null) continue; // TODO: we could/should trace seek destinations; pps needs a "peek" method - columnReaders[i].seek(pps, TypeReader.ReadPhase.ALL); + columnReader.seek(pps, TypeReader.ReadPhase.ALL); } } @@ -329,9 +336,10 @@ private void repositionInStreams(TypeReader[] columnReaders, if (LlapIoImpl.ORC_LOGGER.isTraceEnabled()) { LlapIoImpl.ORC_LOGGER.trace("Created pps {}", Arrays.toString(pps)); } - if (pps == null) return; - for (int i = 0; i < columnReaders.length; i++) { - TypeReader reader = columnReaders[i]; + if (pps == null) { + return; + } + for (TypeReader reader : columnReaders) { if (reader == null) continue; // Note: we assume this never happens for SerDe reader - the batch would never have vectors. // That is always true now; but it wasn't some day, the below would throw in getColumnData. @@ -340,7 +348,7 @@ private void repositionInStreams(TypeReader[] columnReaders, // SettableTreeReader so that we can avoid this check. if (reader instanceof EncodedTreeReaderFactory.TimestampStreamReader && !sameStripe) { ((EncodedTreeReaderFactory.TimestampStreamReader) reader) - .updateTimezone(stripeMetadata.getWriterTimezone()); + .updateTimezone(stripeMetadata.getWriterTimezone()); } reader.seek(pps, TypeReader.ReadPhase.ALL); } @@ -370,9 +378,7 @@ private PositionProvider[] createPositionProviders( if (!stripeMetadata.supportsRowIndexes()) { PositionProvider singleRgPp = new IndexlessPositionProvider(); pps = new PositionProvider[stripeMetadata.getEncodings().size()]; - for (int i = 0; i < pps.length; ++i) { - pps[i] = singleRgPp; - } + Arrays.fill(pps, singleRgPp); } else { int rowGroupIndex = batchKey.rgIx; if (rowGroupIndex == OrcEncodedColumnBatch.ALL_RGS) { diff --git a/llap-server/src/test/org/apache/hadoop/hive/llap/io/decode/TestOrcEncodedDataConsumerRowPositions.java b/llap-server/src/test/org/apache/hadoop/hive/llap/io/decode/TestOrcEncodedDataConsumerRowPositions.java index fca0a3c7d301..45c8ad6d33ff 100644 --- a/llap-server/src/test/org/apache/hadoop/hive/llap/io/decode/TestOrcEncodedDataConsumerRowPositions.java +++ b/llap-server/src/test/org/apache/hadoop/hive/llap/io/decode/TestOrcEncodedDataConsumerRowPositions.java @@ -9,11 +9,12 @@ * * http://www.apache.org/licenses/LICENSE-2.0 * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. */ package org.apache.hadoop.hive.llap.io.decode; @@ -21,6 +22,9 @@ import static org.apache.hadoop.hive.metastore.api.hive_metastoreConstants.META_TABLE_NAME; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; import java.io.IOException; import java.nio.file.Files; @@ -37,8 +41,8 @@ import org.apache.hadoop.hive.llap.io.api.LlapProxy; import org.apache.hadoop.hive.llap.io.api.impl.ColumnVectorBatch; import org.apache.hadoop.hive.llap.io.decode.ColumnVectorProducer.Includes; +import org.apache.hadoop.hive.llap.io.decode.GenericColumnVectorProducer.SerDeStripeMetadata; import org.apache.hadoop.hive.llap.io.metadata.ConsumerFileMetadata; -import org.apache.hadoop.hive.llap.io.metadata.ConsumerStripeMetadata; import org.apache.hadoop.hive.llap.metrics.LlapDaemonIOMetrics; import org.apache.hadoop.hive.ql.exec.Utilities; import org.apache.hadoop.hive.ql.exec.vector.BytesColumnVector; @@ -64,11 +68,7 @@ import org.apache.hadoop.mapred.Reporter; import org.apache.orc.CompressionKind; import org.apache.orc.OrcFile; -import org.apache.orc.OrcProto; -import org.apache.orc.OrcProto.CalendarKind; import org.apache.orc.OrcProto.ColumnEncoding; -import org.apache.orc.OrcProto.RowIndex; -import org.apache.orc.OrcProto.RowIndexEntry; import org.apache.orc.StripeInformation; import org.apache.orc.TypeDescription; import org.apache.orc.Writer; @@ -79,8 +79,8 @@ /** * Absolute file row positions reported by {@link OrcEncodedDataConsumer} on every emitted batch. - * ORC files are read end to end through LLAP IO; the position-less (SerDe-like) metadata case is - * driven straight into decodeBatch. + * ORC files are read end to end through LLAP IO; the position-less SerDe case is driven straight + * into decodeBatch. */ public class TestOrcEncodedDataConsumerRowPositions { @@ -122,46 +122,44 @@ public static void tearDownClass() { @Test public void testRowGroupPositionsAcrossStripes() throws Exception { - List batches = readThroughLlap(indexedFile, ALL_COLUMNS); - - assertEquals(expectedBatches(STRIDE, STRIPE_ROWS), render(batches)); - assertFirstRowIds(batches); + assertEquals(expectedBatches(STRIDE, STRIPE_ROWS), readThroughLlap(indexedFile, ALL_COLUMNS)); } @Test public void testIndexDisabledFileStartsAtStripeBoundary() throws Exception { - List batches = readThroughLlap(noIndexFile, ALL_COLUMNS); - - assertEquals(expectedBatches(Integer.MAX_VALUE, NO_INDEX_STRIPE_ROWS), render(batches)); - assertFirstRowIds(batches); + assertEquals(expectedBatches(Integer.MAX_VALUE, NO_INDEX_STRIPE_ROWS), readThroughLlap(noIndexFile, ALL_COLUMNS)); } /** Without columns, the reader decodes each stripe as one ALL_RGS batch. */ @Test public void testNoColumnReadStartsAtStripeBoundary() throws Exception { - List batches = readThroughLlap(indexedFile, Collections.emptyList()); - - assertEquals(expectedBatches(Integer.MAX_VALUE, STRIPE_ROWS), render(batches)); + assertEquals(expectedBatches(Integer.MAX_VALUE, STRIPE_ROWS), + readThroughLlap(indexedFile, Collections.emptyList())); } @Test public void testNoStripesReportsNoPosition() throws Exception { - List starts = decodeStripe(noStripesMetadata(), 2 * VectorizedRowBatch.DEFAULT_SIZE + 1); - - assertEquals(Collections.nCopies(3, Long.MIN_VALUE), starts); + assertEquals(Collections.nCopies(3, Long.MIN_VALUE), decodeStripe(2 * VectorizedRowBatch.DEFAULT_SIZE + 1)); } // ---- LLAP IO end to end ---- - /** One {startRowInFile, size, id of the first row} per batch, in the order LLAP returned them. */ - private static List readThroughLlap(Path path, List columnIds) throws Exception { + /** + * "startRowInFile:size" per batch, in the order LLAP returned them. When columns are read, the first row of each + * batch must be the row at the reported position. + */ + private static List readThroughLlap(Path path, List columnIds) throws Exception { RecordReader reader = llapReader(path, columnIds); - List batches = new ArrayList<>(); + List batches = new ArrayList<>(); try { VectorizedRowBatch vrb = reader.createValue(); while (reader.next(NullWritable.get(), vrb)) { long position = ((RowPositionAwareVectorizedRecordReader) reader).getRowNumber(); - batches.add(new long[] {position, vrb.size, ((LongColumnVector) vrb.cols[0]).vector[0]}); + if (!columnIds.isEmpty()) { + assertEquals("id of the first row of the batch at " + position, + position, ((LongColumnVector) vrb.cols[0]).vector[0]); + } + batches.add(position + ":" + vrb.size); } } finally { reader.close(); @@ -179,187 +177,65 @@ private static RecordReader llapReader(Path pa return reader; } - private static void assertFirstRowIds(List batches) { - for (long[] batch : batches) { - assertEquals("id of the first row of the batch at " + batch[0], batch[0], batch[2]); - } - } + // ---- decodeBatch without file positions ---- - // ---- decodeBatch with faked metadata ---- + /** Row positions of the batches decodeBatch emits for a zero-column read of one SerDe stripe. */ + private static List decodeStripe(int stripeRows) throws Exception { + // No stripes, as for SerDe-encoded input. + ConsumerFileMetadata fileMetadata = mock(ConsumerFileMetadata.class); + when(fileMetadata.getCompressionKind()).thenReturn(CompressionKind.NONE); + SerDeStripeMetadata stripeMetadata = new SerDeStripeMetadata(0); + stripeMetadata.setRowCount(stripeRows); + stripeMetadata.setEncodings(Collections.singletonList( + ColumnEncoding.newBuilder().setKind(ColumnEncoding.Kind.DIRECT).build())); + Includes includes = mock(Includes.class); + when(includes.getBatchReaderTypes(any())).thenReturn(new TypeDescription[0]); - /** Row positions of the batches decodeBatch emits for the first stripe of a zero-column read. */ - private static List decodeStripe(ConsumerFileMetadata fileMetadata, int stripeRows) throws Exception { QueryFragmentCounters counters = new QueryFragmentCounters(daemonConf, new TezCounters()); - List starts = new ArrayList<>(); - OrcEncodedDataConsumer consumer = new OrcEncodedDataConsumer(null, includes(), counters, IO_METRICS); + OrcEncodedDataConsumer consumer = new OrcEncodedDataConsumer(null, includes, counters, IO_METRICS); consumer.init(null, null, new IoTrace(0, false)); consumer.setFileMetadata(fileMetadata); - consumer.setStripeMetadata(new FakeStripeMetadata(stripeRows)); + consumer.setStripeMetadata(stripeMetadata); + List starts = new ArrayList<>(); OrcEncodedColumnBatch batch = new OrcEncodedColumnBatch(); batch.init(null, 0, OrcEncodedColumnBatch.ALL_RGS, 0); - consumer.decodeBatch(batch, new Consumer() { - @Override public void consumeData(ColumnVectorBatch cvb) { - starts.add(cvb.startRowInFile); + consumer.decodeBatch(batch, new Consumer<>() { + @Override + public void consumeData(ColumnVectorBatch cvb) { + starts.add(cvb.getStartRowInFile()); } - @Override public void setDone() { + @Override + public void setDone() { + // decodeBatch does not call it } - @Override public void setError(Throwable t) { + @Override + public void setError(Throwable t) { throw new AssertionError(t); } }); return starts; } - private static ConsumerFileMetadata noStripesMetadata() { - return new ConsumerFileMetadata() { - @Override public int getStripeCount() { - return 0; - } - - @Override public CompressionKind getCompressionKind() { - return CompressionKind.NONE; - } - - @Override public List getTypes() { - return Collections.emptyList(); - } - - @Override public TypeDescription getSchema() { - return SCHEMA; - } - - @Override public OrcFile.Version getFileVersion() { - return OrcFile.Version.V_0_12; - } - - @Override public CalendarKind getCalendar() { - return CalendarKind.PROLEPTIC_GREGORIAN; - } - - @Override public List getStripes() { - return Collections.emptyList(); - } - - @Override public int getRowIndexStride() { - return 0; - } - }; - } - - /** Reports a row count only. */ - private static final class FakeStripeMetadata implements ConsumerStripeMetadata { - private final long rowCount; - - FakeStripeMetadata(long rowCount) { - this.rowCount = rowCount; - } - - @Override public int getStripeIx() { - return 0; - } - - @Override public long getRowCount() { - return rowCount; - } - - @Override public List getEncodings() { - return Collections.singletonList(ColumnEncoding.newBuilder() - .setKind(ColumnEncoding.Kind.DIRECT).build()); - } - - @Override public String getWriterTimezone() { - return "UTC"; - } - - @Override public RowIndexEntry getRowIndexEntry(int colIx, int rgIx) { - return null; - } - - @Override public RowIndex[] getRowIndexes() { - return new RowIndex[0]; - } - - @Override public boolean supportsRowIndexes() { - return false; - } - } - - private static Includes includes() { - return new Includes() { - @Override public List getPhysicalColumnIds() { - return Collections.emptyList(); - } - - @Override public List getReaderLogicalColumnIds() { - return Collections.emptyList(); - } - - @Override public List getLogicalOrderedColumnIds() { - return Collections.emptyList(); - } - - @Override public boolean[] generateFileIncludes(TypeDescription fileSchema) { - throw new UnsupportedOperationException(); - } - - @Override public TypeDescription[] getBatchReaderTypes(TypeDescription fileSchema) { - return new TypeDescription[0]; - } - - @Override public String[] getOriginalColumnNames(TypeDescription fileSchema) { - throw new UnsupportedOperationException(); - } - - @Override public String getQueryId() { - return "test-query"; - } - - @Override public boolean isProbeDecodeEnabled() { - return false; - } - - @Override public byte getProbeMjSmallTablePos() { - return -1; - } - - @Override public String getProbeCacheKey() { - return null; - } - - @Override public String getProbeColName() { - return null; - } - - @Override public int getProbeColIdx() { - return -1; - } - }; - } - // ---- fixture ---- private static void writeFile(Path path, int stride, int[] stripeRows) throws IOException { try (Writer writer = OrcFile.createWriter(path, OrcFile.writerOptions(daemonConf) .setSchema(SCHEMA).rowIndexStride(stride).compress(CompressionKind.NONE))) { - VectorizedRowBatch batch = SCHEMA.createRowBatch(); + VectorizedRowBatch batch = SCHEMA.createRowBatch(Arrays.stream(stripeRows).max().getAsInt()); LongColumnVector ids = (LongColumnVector) batch.cols[0]; BytesColumnVector names = (BytesColumnVector) batch.cols[1]; long row = 0; for (int s = 0; s < stripeRows.length; ++s) { - for (int left = stripeRows[s]; left > 0; ) { - int size = Math.min(left, batch.getMaxSize()); - batch.size = size; - for (int i = 0; i < size; ++i, ++row) { - ids.vector[i] = row; - names.setVal(i, ("n" + row).getBytes(UTF_8)); - } - writer.addRowBatch(batch); - batch.reset(); - left -= size; + batch.reset(); + batch.size = stripeRows[s]; + for (int i = 0; i < batch.size; ++i, ++row) { + ids.vector[i] = row; + names.setVal(i, ("n" + row).getBytes(UTF_8)); } + writer.addRowBatch(batch); if (s < stripeRows.length - 1) { writer.writeIntermediateFooter(); } @@ -420,12 +296,4 @@ private static List expectedBatches(int stride, int[] stripeRows) { } return expected; } - - private static List render(List batches) { - List rendered = new ArrayList<>(); - for (long[] batch : batches) { - rendered.add(batch[0] + ":" + batch[1]); - } - return rendered; - } } From 2f0cd0983482d41aa8d5c29c0ac9d41b8fc5cc6e Mon Sep 17 00:00:00 2001 From: Denys Kuzmenko Date: Fri, 30 Oct 2026 12:57:36 +0200 Subject: [PATCH 4/4] HIVE-30102: Address Sonar issues - LlapRecordReader: merge the nested decimal64 if - OrcEncodedDataConsumer: braces on the null-reader continue --- .../hadoop/hive/llap/io/api/impl/LlapRecordReader.java | 7 +++---- .../hadoop/hive/llap/io/decode/OrcEncodedDataConsumer.java | 4 +++- 2 files changed, 6 insertions(+), 5 deletions(-) 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 80713688cd9f..7a2b26b00e6b 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 @@ -315,10 +315,9 @@ static int determineQueueLimit(long maxBufferedSize, break; case DECIMAL: boolean useDecimal64 = false; - if (ti instanceof DecimalTypeInfo dti) { - if (dti.getPrecision() <= TypeDescription.MAX_DECIMAL64_PRECISION && decimal64Support) { - useDecimal64 = true; - } + if (ti instanceof DecimalTypeInfo dti && + dti.getPrecision() <= TypeDescription.MAX_DECIMAL64_PRECISION && decimal64Support) { + useDecimal64 = true; } // decimal_64 column vectors gets the same weight as long column vectors if (useDecimal64) { diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcEncodedDataConsumer.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcEncodedDataConsumer.java index 33986ffc248d..78e954fb1e48 100644 --- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcEncodedDataConsumer.java +++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/OrcEncodedDataConsumer.java @@ -322,7 +322,9 @@ private void positionInStreams(TypeReader[] columnReaders, return; } for (TypeReader columnReader : columnReaders) { - if (columnReader == null) continue; + if (columnReader == null) { + continue; + } // TODO: we could/should trace seek destinations; pps needs a "peek" method columnReader.seek(pps, TypeReader.ReadPhase.ALL); }