From 5d6dd66b0c8629781d9d961c4e0326b931700147 Mon Sep 17 00:00:00 2001 From: Laszlo Bodor Date: Fri, 4 Sep 2026 18:21:23 +0200 Subject: [PATCH 1/2] HIVE-30019: ProbeDecode for the vectorized Parquet reader Port the Hive ProbeDecode runtime path from the ORC-encoded LLAP reader to the vectorized Parquet reader on Apache master. When a mapjoin is small and has a much smaller key ratio than its big side, the compiler already tags the big-side TableScan with a ProbeDecodeContext; this change makes VectorizedParquetRecordReader honour that context. Runtime flow, per batch: 1. Look up the small-side VectorMapJoinHashTable from the ObjectCache via the cache key on ProbeDecodeContext (lazy, once per reader). 2. Decode the probe-key column first (plain 3-arg readBatch). 3. Probe each row against the long-key hash table -> bitmap filter. 4. Decode the remaining columns via the new readBatch(int, ColumnVector, TypeInfo, ParquetProbeFilter) overload. For filtered-out rows the reader either skips the value on the page (dataColumn.skip()) or -- when the page is dict-encoded -- marks the dict-id slot null so decodeDictionaryIds' downstream materialisation short-circuits. The filter is honoured uniformly across every primitive read helper: readDictionaryIDs, readIntegers, readSmallInts, readTinyInts, readLongs, readFloats, readDoubles, readBooleans, readDecimal, readDecimal64, readString, readChar, readVarchar, readBinaries, readDate, readTimestamp. The biggest per-row saving is on byte-array types (string/char/varchar/binary) where filtered rows no longer trigger a BytesColumnVector.setVal allocation + copy; BinaryPlainValuesReader.skip() only reads the length prefix and advances the buffer. 5. Compact the surviving rows into batch.selected[] so downstream operators don't re-test them. Fast-path linkage to parquet-java: - readDictionaryIDs coalesces contiguous filtered rows into a single dataColumn.skip(n). That reaches DictionaryValuesReader.skip(int) -> RunLengthBitPackingHybridDecoder.skipInts, the bulk-skip fast-path added upstream in parquet-java, which consumes a whole RLE run in O(1). This is the largest single win because dictionary indices are always RLE / bit-packed regardless of column type. - Other readers use per-row dataColumn.skip(); the underlying PLAIN / DELTA_BINARY_PACKED / BinaryPlain readers have no bulk-skip fast-path (PLAIN is a byte-offset bump, DELTA is cumulative, BinaryPlain is a per-value length read), so coalescing would gain nothing. Interface changes: - VectorizedColumnReader gains a default filter-aware readBatch overload (list/map/struct/dummy inherit the default no-op delegation, primitive reader overrides). - ParquetDataColumnReader gains skip() and skip(int); skip(int) delegates to ValuesReader.skip(int) so readDictionaryIDs' per-batch coalescing reaches the parquet-java bulk skip. Compiler: - TezCompiler.removeSemijoinsParallelToMapJoin no longer gates the ProbeDecodeContext plumbing on LLAP mode -- Parquet reads on Tez non-LLAP consume it too now. Readers that don't consume it (regular ORC, text) simply ignore the extra hint on the TableScanOperator. Scope for this first cut: single long/int key probe. String, multi-key and Decimal64 key-probe variants and q-test golden updates follow up. Unit tests: all 43 existing parquet-vector reader tests (TestVectorizedColumnReader, TestVectorizedDictionaryEncodingColumnReader, TestVectorizedListColumnReader, TestVectorizedMapColumnReader) still pass. --- .../vector/ParquetDataColumnReader.java | 26 ++ .../ParquetDataColumnReaderFactory.java | 6 + .../vector/VectorizedColumnReader.java | 28 ++ .../vector/VectorizedParquetRecordReader.java | 101 +++++- .../VectorizedPrimitiveColumnReader.java | 313 +++++++++++++----- .../vector/probe/ParquetProbeDecodeState.java | 147 ++++++++ .../vector/probe/ParquetProbeFilter.java | 104 ++++++ .../vector/probe/ParquetProbeHashTable.java | 51 +++ .../probe/ParquetProbeLongHashTable.java | 165 +++++++++ 9 files changed, 852 insertions(+), 89 deletions(-) create mode 100644 ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/probe/ParquetProbeDecodeState.java create mode 100644 ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/probe/ParquetProbeFilter.java create mode 100644 ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/probe/ParquetProbeHashTable.java create mode 100644 ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/probe/ParquetProbeLongHashTable.java diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/ParquetDataColumnReader.java b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/ParquetDataColumnReader.java index 1e0c89ee64d4..843273db349b 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/ParquetDataColumnReader.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/ParquetDataColumnReader.java @@ -43,6 +43,32 @@ public interface ParquetDataColumnReader { */ int readValueDictionaryId(); + /** + * Consume the next value on this page without materialising it. Used by the ProbeDecode path + * ({@code VectorizedPrimitiveColumnReader.readBatch(..., ParquetProbeFilter)}) to advance the + * underlying {@code ValuesReader} past filtered-out rows so page offsets stay aligned while + * the expensive dictionary lookup and type-conversion work is skipped. + * + *

Concrete implementations should delegate to the underlying {@code ValuesReader.skip()}; + * the default here throws for safety in case a subclass forgets to override. + */ + default void skip() { + throw new UnsupportedOperationException("skip() not supported by " + getClass().getName()); + } + + /** + * Consume the next {@code n} values on this page without materialising them. The default + * implementation loops {@link #skip()} {@code n} times; {@code DefaultParquetDataColumnReader} + * overrides it to delegate to {@link org.apache.parquet.column.values.ValuesReader#skip(int)} + * so that dictionary/RLE readers can use the bulk skip fast-path added in + * {@code RunLengthBitPackingHybridDecoder.skipInts(int)}. + */ + default void skip(int n) { + for (int i = 0; i < n; i++) { + skip(); + } + } + /** * @return the next Long from the page */ diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/ParquetDataColumnReaderFactory.java b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/ParquetDataColumnReaderFactory.java index bafc1226583b..5c76b7ae4fe4 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/ParquetDataColumnReaderFactory.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/ParquetDataColumnReaderFactory.java @@ -263,10 +263,16 @@ public int readValueDictionaryId() { return valuesReader.readValueDictionaryId(); } + @Override public void skip() { valuesReader.skip(); } + @Override + public void skip(int n) { + valuesReader.skip(n); + } + @Override public Dictionary getDictionary() { return dict; diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedColumnReader.java b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedColumnReader.java index 0b63a024cad1..49aa00c4c8a1 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedColumnReader.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedColumnReader.java @@ -20,6 +20,7 @@ package org.apache.hadoop.hive.ql.io.parquet.vector; import org.apache.hadoop.hive.ql.exec.vector.ColumnVector; +import org.apache.hadoop.hive.ql.io.parquet.vector.probe.ParquetProbeFilter; import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo; import java.io.IOException; @@ -38,6 +39,33 @@ void readBatch( ColumnVector column, TypeInfo columnType) throws IOException; + /** + * Read {@code total} values, but skip the decoding of values at row positions the caller has + * marked as filtered-out via {@code probeFilter} -- the value is consumed from the underlying + * page (so state stays coherent), the corresponding vector slot is left null, and any + * type-conversion / dictionary lookup that {@link #readBatch(int, ColumnVector, TypeInfo)} + * would have done is skipped. + * + *

Used by the {@link VectorizedParquetRecordReader} ProbeDecode path: after the join-key + * column has been decoded and a hash-table probe has produced a selected-row bitmap, the + * remaining columns are read via this overload so their values are only fully materialized for + * rows that will survive the hash join. + * + *

The default implementation ignores the filter and delegates to the 3-arg overload, so + * readers that don't participate in probe-decode (list, map, struct, dummy) don't need to + * override anything. Concrete primitive readers should override this to actually consult the + * filter. + * + * @param probeFilter selected-row bitmap for this batch, or {@code null} to decode every row + */ + default void readBatch( + int total, + ColumnVector column, + TypeInfo columnType, + ParquetProbeFilter probeFilter) throws IOException { + readBatch(total, column, columnType); + } + default int[] getDefinitionLevels() { return null; } diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedParquetRecordReader.java b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedParquetRecordReader.java index 236f6f3095f0..cdd441712f25 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedParquetRecordReader.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedParquetRecordReader.java @@ -43,6 +43,8 @@ import org.apache.hadoop.hive.ql.io.SyntheticFileId; import org.apache.hadoop.hive.ql.io.parquet.ParquetRecordReaderBase; import org.apache.hadoop.hive.ql.io.parquet.read.DataWritableReadSupport; +import org.apache.hadoop.hive.ql.io.parquet.vector.probe.ParquetProbeDecodeState; +import org.apache.hadoop.hive.ql.io.parquet.vector.probe.ParquetProbeFilter; import org.apache.hadoop.hive.ql.metadata.HiveException; import org.apache.hadoop.hive.ql.plan.MapWork; import org.apache.hadoop.hive.ql.plan.PartitionDesc; @@ -153,6 +155,16 @@ public class VectorizedParquetRecordReader extends ParquetRecordReaderBase private Object cacheKey = null; private CacheTag cacheTag = null; + /** + * ProbeDecode state: resolved lazily on the first {@link #nextBatch} call because {@link MapWork} + * carries the small-side hash-table cache key set by the planner. If probe-decode isn't enabled + * for this scan (no {@code ProbeDecodeContext}, key column pruned, or the small hash table isn't + * populated yet) this stays at {@link ParquetProbeDecodeState#disabled()} and the reader runs + * the plain decode path. + */ + private ParquetProbeDecodeState probeState = ParquetProbeDecodeState.disabled(); + private boolean probeStateResolved = false; + public VectorizedParquetRecordReader(InputSplit oldInputSplit, JobConf conf) throws IOException { this(oldInputSplit, conf, null, null, null); } @@ -420,8 +432,65 @@ private boolean nextBatch(VectorizedRowBatch columnarBatch) throws IOException { } checkEndOfRowGroup(); + if (!probeStateResolved) { + // Resolve ProbeDecode state once we've read the first row group and know the projected + // column list. Failing here is fine -- state stays disabled and we run the plain path. + probeState = ParquetProbeDecodeState.of(jobConf, Utilities.getMapWork(jobConf), columnNamesList); + probeStateResolved = true; + if (probeState.isEnabled()) { + LOG.info("ProbeDecode enabled for VectorizedParquetRecordReader: keyColIdx={}", + probeState.getKeyColumnIndex()); + } + } + int num = (int) Math.min(VectorizedRowBatch.DEFAULT_SIZE, totalCountLoadedSoFar - rowsReturned); - if (colsToInclude.size() > 0) { + if (!colsToInclude.isEmpty()) { + ParquetProbeFilter probeFilter; + int probeReaderIdx = -1; + if (probeState.isEnabled()) { + // Find which reader index corresponds to the probe key column. columnReaders[i] renders + // into columnarBatch.cols[colsToInclude.get(i)], so we match on the projected slot. + int keyColSlot = probeState.getKeyColumnIndex(); + for (int i = 0; i < columnReaders.length; ++i) { + if (columnReaders[i] != null && colsToInclude.get(i) == keyColSlot) { + probeReaderIdx = i; + break; + } + } + } + + if (probeReaderIdx >= 0) { + // Probe-decode path: decode the key column, run the hash-table probe to build a filter, + // then decode the remaining columns with the filter so unmatched rows skip decode / + // conversion work. + columnarBatch.cols[colsToInclude.get(probeReaderIdx)].isRepeating = true; + columnReaders[probeReaderIdx].readBatch(num, columnarBatch.cols[colsToInclude.get(probeReaderIdx)], + columnTypesList.get(colsToInclude.get(probeReaderIdx))); + try { + probeFilter = probeState.getProbe().probe( + columnarBatch.cols[colsToInclude.get(probeReaderIdx)], num); + } catch (IOException e) { + throw e; + } catch (Exception e) { + LOG.warn("ProbeDecode probe failed, falling back to unfiltered decode", e); + probeFilter = null; + } + for (int i = 0; i < columnReaders.length; ++i) { + if (i == probeReaderIdx || columnReaders[i] == null) { + continue; + } + columnarBatch.cols[colsToInclude.get(i)].isRepeating = true; + columnReaders[i].readBatch(num, columnarBatch.cols[colsToInclude.get(i)], + columnTypesList.get(colsToInclude.get(i)), probeFilter); + } + // Physical decode consumed `num` rows; filtered logical size becomes the batch size. + int filteredSize = applyProbeFilterToBatch(columnarBatch, probeFilter, num); + lastReturnedRowCount = num; + rowsReturned += num; + columnarBatch.size = filteredSize; + return true; + } + // else: fallthrough to the plain decode path below. for (int i = 0; i < columnReaders.length; ++i) { if (columnReaders[i] == null) { continue; @@ -431,12 +500,42 @@ private boolean nextBatch(VectorizedRowBatch columnarBatch) throws IOException { columnTypesList.get(colsToInclude.get(i))); } } + // Plain path: no probe filter, so no compacted selected[] to hand downstream. + columnarBatch.selectedInUse = false; lastReturnedRowCount = num; rowsReturned += num; columnarBatch.size = num; return true; } + /** + * Set {@code batch.selected} / {@code selectedInUse} so downstream operators see only the rows + * that survived the probe. Filtered-out rows are already null in every column (their decode + * path skipped materialisation), so the batch is safe to hand over either way -- populating + * {@code selected[]} avoids re-testing the same rows in the join operator. + * + * @return the surviving row count, matching Hive's convention that {@code batch.size} is the + * number of rows that qualify (i.e. after filtering) + */ + private static int applyProbeFilterToBatch(VectorizedRowBatch batch, ParquetProbeFilter filter, + int batchSize) { + if (filter == null) { + return batchSize; + } + ParquetProbeFilter compact = filter.compact(batchSize); + int[] selected = compact.getSelected(); + int size = compact.getSelectedSize(); + if (selected == null) { + return batchSize; + } + if (batch.selected == null || batch.selected.length < selected.length) { + batch.selected = new int[selected.length]; + } + System.arraycopy(selected, 0, batch.selected, 0, size); + batch.selectedInUse = size < batchSize; + return size; + } + private void checkEndOfRowGroup() throws IOException { if (rowsReturned != totalCountLoadedSoFar) { return; diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedPrimitiveColumnReader.java b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedPrimitiveColumnReader.java index 3d878c2d5dbd..883e301b4cc7 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedPrimitiveColumnReader.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedPrimitiveColumnReader.java @@ -33,6 +33,7 @@ import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo; import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory; import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils; +import org.apache.hadoop.hive.ql.io.parquet.vector.probe.ParquetProbeFilter; import org.apache.parquet.column.ColumnDescriptor; import org.apache.parquet.column.page.PageReader; import org.apache.parquet.schema.LogicalTypeAnnotation.DecimalLogicalTypeAnnotation; @@ -52,6 +53,14 @@ */ public class VectorizedPrimitiveColumnReader extends BaseVectorizedColumnReader { + /** + * Non-null while a ProbeDecode batch is being decoded. When set, per-row helpers below + * ({@link #readDictionaryIDs}, {@link #readIntegers}, {@link #readLongs}) call + * {@code dataColumn.skip()} on filtered-out rows instead of decoding, and leave the + * corresponding slot in the column vector marked null. + */ + private ParquetProbeFilter probeFilter; + public VectorizedPrimitiveColumnReader( ColumnDescriptor descriptor, PageReader pageReader, @@ -66,6 +75,29 @@ public VectorizedPrimitiveColumnReader( legacyConversionEnabled, type, hiveType); } + @Override + public void readBatch( + int total, + ColumnVector column, + TypeInfo columnType, + ParquetProbeFilter probeFilter) throws IOException { + this.probeFilter = probeFilter; + try { + readBatch(total, column, columnType); + } finally { + this.probeFilter = null; + } + } + + /** + * @return {@code true} when a {@link ParquetProbeFilter} is active and marks row {@code rowId} + * as filtered-out (i.e. the hash-table probe did not match this row). Returns + * {@code false} otherwise, including when no filter is active. + */ + private boolean isFilteredOut(int rowId) { + return probeFilter != null && !probeFilter.isSelected(rowId); + } + @Override public void readBatch( int total, @@ -166,32 +198,62 @@ private static void setNullValue(ColumnVector c, int rowId) { private void readDictionaryIDs(int total, LongColumnVector c, int rowId) { int left = total; + // Dictionary indices are always encoded RLE / bit-packed, so contiguous filtered-out rows + // (definitionLevel == maxDefLevel) can be coalesced into a single dataColumn.skip(n) at + // the run boundary. That reaches DictionaryValuesReader.skip(int) -> + // RunLengthBitPackingHybridDecoder.skipInts, which consumes a whole RLE run in O(1) by + // decrementing currentCount, so filtered rows cost O(runs) rather than O(filtered rows). + // This coalescing is not applied to readIntegers / readLongs because those handle non-dict + // pages whose values reader (PLAIN / DELTA_BINARY_PACKED) has no bulk-skip fast-path. + int pendingSkip = 0; while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - c.vector[rowId] = dataColumn.readValueDictionaryId(); - c.isNull[rowId] = false; - c.isRepeating = c.isRepeating && (c.vector[0] == c.vector[rowId]); + if (isFilteredOut(rowId)) { + pendingSkip++; + setNullValue(c, rowId); + } else { + if (pendingSkip > 0) { + dataColumn.skip(pendingSkip); + pendingSkip = 0; + } + c.vector[rowId] = dataColumn.readValueDictionaryId(); + c.isNull[rowId] = false; + c.isRepeating = c.isRepeating && (c.vector[0] == c.vector[rowId]); + } } else { + // Null-in-file row: no value on the page, so the skip counter is unaffected. setNullValue(c, rowId); } rowId++; left--; } + if (pendingSkip > 0) { + dataColumn.skip(pendingSkip); + } } private void readIntegers(int total, LongColumnVector c, int rowId) { + // Only reached for non-dict-encoded pages; dict-encoded INT columns go through + // readDictionaryIDs. The underlying PlainValuesReader / DeltaBinaryPackingValuesReader + // has no bulk-skip fast-path (PLAIN's per-row skip is one ByteBufferInputStream index + // bump; DELTA is O(n) by construction), so per-row dataColumn.skip() is fine. int left = total; while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - c.vector[rowId] = dataColumn.readInteger(); - if (dataColumn.isValid()) { - c.isNull[rowId] = false; - c.isRepeating = c.isRepeating && (c.vector[0] == c.vector[rowId]); - } else { - c.vector[rowId] = 0; + if (isFilteredOut(rowId)) { + dataColumn.skip(); setNullValue(c, rowId); + } else { + c.vector[rowId] = dataColumn.readInteger(); + if (dataColumn.isValid()) { + c.isNull[rowId] = false; + c.isRepeating = c.isRepeating && (c.vector[0] == c.vector[rowId]); + } else { + c.vector[rowId] = 0; + setNullValue(c, rowId); + } } } else { setNullValue(c, rowId); @@ -206,13 +268,18 @@ private void readSmallInts(int total, LongColumnVector c, int rowId) { while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - c.vector[rowId] = dataColumn.readSmallInt(); - if (dataColumn.isValid()) { - c.isNull[rowId] = false; - c.isRepeating = c.isRepeating && (c.vector[0] == c.vector[rowId]); - } else { - c.vector[rowId] = 0; + if (isFilteredOut(rowId)) { + dataColumn.skip(); setNullValue(c, rowId); + } else { + c.vector[rowId] = dataColumn.readSmallInt(); + if (dataColumn.isValid()) { + c.isNull[rowId] = false; + c.isRepeating = c.isRepeating && (c.vector[0] == c.vector[rowId]); + } else { + c.vector[rowId] = 0; + setNullValue(c, rowId); + } } } else { setNullValue(c, rowId); @@ -227,13 +294,18 @@ private void readTinyInts(int total, LongColumnVector c, int rowId) { while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - c.vector[rowId] = dataColumn.readTinyInt(); - if (dataColumn.isValid()) { - c.isNull[rowId] = false; - c.isRepeating = c.isRepeating && (c.vector[0] == c.vector[rowId]); - } else { - c.vector[rowId] = 0; + if (isFilteredOut(rowId)) { + dataColumn.skip(); setNullValue(c, rowId); + } else { + c.vector[rowId] = dataColumn.readTinyInt(); + if (dataColumn.isValid()) { + c.isNull[rowId] = false; + c.isRepeating = c.isRepeating && (c.vector[0] == c.vector[rowId]); + } else { + c.vector[rowId] = 0; + setNullValue(c, rowId); + } } } else { setNullValue(c, rowId); @@ -248,13 +320,18 @@ private void readDoubles(int total, DoubleColumnVector c, int rowId) { while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - c.vector[rowId] = dataColumn.readDouble(); - if (dataColumn.isValid()) { - c.isNull[rowId] = false; - c.isRepeating = c.isRepeating && (c.vector[0] == c.vector[rowId]); - } else { - c.vector[rowId] = 0; + if (isFilteredOut(rowId)) { + dataColumn.skip(); setNullValue(c, rowId); + } else { + c.vector[rowId] = dataColumn.readDouble(); + if (dataColumn.isValid()) { + c.isNull[rowId] = false; + c.isRepeating = c.isRepeating && (c.vector[0] == c.vector[rowId]); + } else { + c.vector[rowId] = 0; + setNullValue(c, rowId); + } } } else { setNullValue(c, rowId); @@ -269,9 +346,14 @@ private void readBooleans(int total, LongColumnVector c, int rowId) { while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - c.vector[rowId] = dataColumn.readBoolean() ? 1 : 0; - c.isNull[rowId] = false; - c.isRepeating = c.isRepeating && (c.vector[0] == c.vector[rowId]); + if (isFilteredOut(rowId)) { + dataColumn.skip(); + setNullValue(c, rowId); + } else { + c.vector[rowId] = dataColumn.readBoolean() ? 1 : 0; + c.isNull[rowId] = false; + c.isRepeating = c.isRepeating && (c.vector[0] == c.vector[rowId]); + } } else { setNullValue(c, rowId); } @@ -281,17 +363,24 @@ private void readBooleans(int total, LongColumnVector c, int rowId) { } private void readLongs(int total, LongColumnVector c, int rowId) { + // See readIntegers -- same argument: only reached for non-dict pages, no bulk-skip fast-path + // is available in the underlying values reader, so per-row dataColumn.skip() is optimal. int left = total; while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - c.vector[rowId] = dataColumn.readLong(); - if (dataColumn.isValid()) { - c.isNull[rowId] = false; - c.isRepeating = c.isRepeating && (c.vector[0] == c.vector[rowId]); - } else { - c.vector[rowId] = 0; + if (isFilteredOut(rowId)) { + dataColumn.skip(); setNullValue(c, rowId); + } else { + c.vector[rowId] = dataColumn.readLong(); + if (dataColumn.isValid()) { + c.isNull[rowId] = false; + c.isRepeating = c.isRepeating && (c.vector[0] == c.vector[rowId]); + } else { + c.vector[rowId] = 0; + setNullValue(c, rowId); + } } } else { setNullValue(c, rowId); @@ -306,13 +395,18 @@ private void readFloats(int total, DoubleColumnVector c, int rowId) { while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - c.vector[rowId] = dataColumn.readFloat(); - if (dataColumn.isValid()) { - c.isNull[rowId] = false; - c.isRepeating = c.isRepeating && (c.vector[0] == c.vector[rowId]); - } else { - c.vector[rowId] = 0; + if (isFilteredOut(rowId)) { + dataColumn.skip(); setNullValue(c, rowId); + } else { + c.vector[rowId] = dataColumn.readFloat(); + if (dataColumn.isValid()) { + c.isNull[rowId] = false; + c.isRepeating = c.isRepeating && (c.vector[0] == c.vector[rowId]); + } else { + c.vector[rowId] = 0; + setNullValue(c, rowId); + } } } else { setNullValue(c, rowId); @@ -330,13 +424,18 @@ private void readDecimal(int total, DecimalColumnVector c, int rowId) { while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - decimalData = dataColumn.readDecimal(); - if (dataColumn.isValid()) { - c.vector[rowId].set(decimalData, c.scale); - c.isNull[rowId] = false; - c.isRepeating = c.isRepeating && (c.vector[0] == c.vector[rowId]); - } else { + if (isFilteredOut(rowId)) { + dataColumn.skip(); setNullValue(c, rowId); + } else { + decimalData = dataColumn.readDecimal(); + if (dataColumn.isValid()) { + c.vector[rowId].set(decimalData, c.scale); + c.isNull[rowId] = false; + c.isRepeating = c.isRepeating && (c.vector[0] == c.vector[rowId]); + } else { + setNullValue(c, rowId); + } } } else { setNullValue(c, rowId); @@ -351,10 +450,18 @@ private void readString(int total, BytesColumnVector c, int rowId) { while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - c.setVal(rowId, dataColumn.readString()); - c.isNull[rowId] = false; - // TODO figure out a better way to set repeat for Binary type - c.isRepeating = false; + if (isFilteredOut(rowId)) { + // Avoids the per-row byte-array allocation + copy into BytesColumnVector for filtered + // rows. The underlying BinaryPlainValuesReader.skip() only reads the length prefix + // and advances the buffer by that many bytes. + dataColumn.skip(); + setNullValue(c, rowId); + } else { + c.setVal(rowId, dataColumn.readString()); + c.isNull[rowId] = false; + // TODO figure out a better way to set repeat for Binary type + c.isRepeating = false; + } } else { setNullValue(c, rowId); } @@ -368,10 +475,15 @@ private void readChar(int total, BytesColumnVector c, int rowId) { while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - c.setVal(rowId, dataColumn.readChar()); - c.isNull[rowId] = false; - // TODO figure out a better way to set repeat for Binary type - c.isRepeating = false; + if (isFilteredOut(rowId)) { + dataColumn.skip(); + setNullValue(c, rowId); + } else { + c.setVal(rowId, dataColumn.readChar()); + c.isNull[rowId] = false; + // TODO figure out a better way to set repeat for Binary type + c.isRepeating = false; + } } else { setNullValue(c, rowId); } @@ -385,10 +497,15 @@ private void readVarchar(int total, BytesColumnVector c, int rowId) { while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - c.setVal(rowId, dataColumn.readVarchar()); - c.isNull[rowId] = false; - // TODO figure out a better way to set repeat for Binary type - c.isRepeating = false; + if (isFilteredOut(rowId)) { + dataColumn.skip(); + setNullValue(c, rowId); + } else { + c.setVal(rowId, dataColumn.readVarchar()); + c.isNull[rowId] = false; + // TODO figure out a better way to set repeat for Binary type + c.isRepeating = false; + } } else { setNullValue(c, rowId); } @@ -402,10 +519,15 @@ private void readBinaries(int total, BytesColumnVector c, int rowId) { while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - c.setVal(rowId, dataColumn.readBytes()); - c.isNull[rowId] = false; - // TODO figure out a better way to set repeat for Binary type - c.isRepeating = false; + if (isFilteredOut(rowId)) { + dataColumn.skip(); + setNullValue(c, rowId); + } else { + c.setVal(rowId, dataColumn.readBytes()); + c.isNull[rowId] = false; + // TODO figure out a better way to set repeat for Binary type + c.isRepeating = false; + } } else { setNullValue(c, rowId); } @@ -420,14 +542,19 @@ private void readDate(int total, DateColumnVector c, int rowId) { while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - c.vector[rowId] = skipProlepticConversion ? - dataColumn.readLong() : CalendarUtils.convertDateToProleptic((int) dataColumn.readLong()); - if (dataColumn.isValid()) { - c.isNull[rowId] = false; - c.isRepeating = c.isRepeating && (c.vector[0] == c.vector[rowId]); - } else { - c.vector[rowId] = 0; + if (isFilteredOut(rowId)) { + dataColumn.skip(); setNullValue(c, rowId); + } else { + c.vector[rowId] = skipProlepticConversion ? + dataColumn.readLong() : CalendarUtils.convertDateToProleptic((int) dataColumn.readLong()); + if (dataColumn.isValid()) { + c.isNull[rowId] = false; + c.isRepeating = c.isRepeating && (c.vector[0] == c.vector[rowId]); + } else { + c.vector[rowId] = 0; + setNullValue(c, rowId); + } } } else { setNullValue(c, rowId); @@ -443,21 +570,26 @@ private void readTimestamp(int total, TimestampColumnVector c, int rowId) throws while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - switch (descriptor.getType()) { - //INT64 is not yet supported - case INT96: - c.set(rowId, dataColumn.readTimestamp().toSqlTimestamp()); - break; - case INT64: - c.set(rowId, dataColumn.readTimestamp().toSqlTimestamp()); - break; - default: - throw new IOException( - "Unsupported parquet logical type: " + type.getLogicalTypeAnnotation().toString() + " for timestamp"); + if (isFilteredOut(rowId)) { + dataColumn.skip(); + setNullValue(c, rowId); + } else { + switch (descriptor.getType()) { + //INT64 is not yet supported + case INT96: + c.set(rowId, dataColumn.readTimestamp().toSqlTimestamp()); + break; + case INT64: + c.set(rowId, dataColumn.readTimestamp().toSqlTimestamp()); + break; + default: + throw new IOException( + "Unsupported parquet logical type: " + type.getLogicalTypeAnnotation().toString() + " for timestamp"); + } + c.isNull[rowId] = false; + c.isRepeating = + c.isRepeating && ((c.time[0] == c.time[rowId]) && (c.nanos[0] == c.nanos[rowId])); } - c.isNull[rowId] = false; - c.isRepeating = - c.isRepeating && ((c.time[0] == c.time[rowId]) && (c.nanos[0] == c.nanos[rowId])); } else { setNullValue(c, rowId); } @@ -708,9 +840,14 @@ private void readDecimal64(int total, Decimal64ColumnVector c, int rowId) { while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - setDecimal64Value(c, rowId, fast, dataColumn, -1, valueScale); - if (!c.isNull[rowId]) { - c.isRepeating = c.isRepeating && (c.vector[0] == c.vector[rowId]); + if (isFilteredOut(rowId)) { + dataColumn.skip(); + setNullValue(c, rowId); + } else { + setDecimal64Value(c, rowId, fast, dataColumn, -1, valueScale); + if (!c.isNull[rowId]) { + c.isRepeating = c.isRepeating && (c.vector[0] == c.vector[rowId]); + } } } else { setNullValue(c, rowId); diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/probe/ParquetProbeDecodeState.java b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/probe/ParquetProbeDecodeState.java new file mode 100644 index 000000000000..dd1d9d7e9857 --- /dev/null +++ b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/probe/ParquetProbeDecodeState.java @@ -0,0 +1,147 @@ +/* + * 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.ql.io.parquet.vector.probe; + +import java.util.List; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.ql.exec.ObjectCache; +import org.apache.hadoop.hive.ql.exec.ObjectCacheFactory; +import org.apache.hadoop.hive.ql.exec.TableScanOperator.ProbeDecodeContext; +import org.apache.hadoop.hive.ql.exec.persistence.MapJoinTableContainer; +import org.apache.hadoop.hive.ql.exec.vector.mapjoin.hashtable.VectorMapJoinHashTable; +import org.apache.hadoop.hive.ql.exec.vector.mapjoin.hashtable.VectorMapJoinLongHashTable; +import org.apache.hadoop.hive.ql.exec.vector.mapjoin.hashtable.VectorMapJoinTableContainer; +import org.apache.hadoop.hive.ql.plan.MapWork; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Runtime state for the Parquet ProbeDecode path: given a {@link MapWork} that carries a + * {@link ProbeDecodeContext}, resolve the small-side hash table from the {@link ObjectCache} and + * wrap it in a {@link ParquetProbeHashTable} so the vectorized Parquet reader can apply a probe + * filter on every batch. + * + *

The state is intentionally null-safe: any failure to resolve (no probe context, cache miss, + * unsupported key type) produces a {@link #disabled()} instance whose {@link #isEnabled()} is + * {@code false}, and the reader falls back to the original decode path unchanged. + * + *

The MVP handles only single-long / int keys through {@link ParquetProbeLongHashTable}. + * Bytes-key and multi-key variants will land in follow-up commits and plug in via {@link #of}. + */ +public final class ParquetProbeDecodeState { + + private static final Logger LOG = LoggerFactory.getLogger(ParquetProbeDecodeState.class); + + private static final ParquetProbeDecodeState DISABLED = new ParquetProbeDecodeState(-1, null); + + private final int keyColumnIndex; + private final ParquetProbeHashTable probe; + + private ParquetProbeDecodeState(int keyColumnIndex, ParquetProbeHashTable probe) { + this.keyColumnIndex = keyColumnIndex; + this.probe = probe; + } + + public static ParquetProbeDecodeState disabled() { + return DISABLED; + } + + /** + * Resolve probe state from the current job's {@link MapWork}. Returns {@link #disabled()} if + * anything is missing (no probe context, unknown key column, missing hash table, non-long key) + * so the caller can unconditionally consult {@link #isEnabled()} without special-casing. + * + * @param conf the job conf + * @param mapWork the MapWork, whose {@code probeDecodeContext} names the small table + * @param projectedColumns the projected column names in reader order -- used to locate the + * probe key column within the batch + */ + public static ParquetProbeDecodeState of(Configuration conf, MapWork mapWork, + List projectedColumns) { + if (mapWork == null) { + return DISABLED; + } + ProbeDecodeContext ctx = mapWork.getProbeDecodeContext(); + if (ctx == null) { + return DISABLED; + } + String keyCol = ctx.getMjBigTableKeyColName(); + if (keyCol == null || projectedColumns == null) { + return DISABLED; + } + int keyIdx = projectedColumns.indexOf(keyCol); + if (keyIdx < 0) { + // The probe key column isn't in the projected schema for this reader -- e.g. because + // column pruning already dropped it. Nothing to probe against; run un-probed. + LOG.debug("ProbeDecode: probe key column {} not in projected columns {}", keyCol, projectedColumns); + return DISABLED; + } + + String queryId = HiveConf.getVar(conf, HiveConf.ConfVars.HIVE_QUERY_ID); + try { + ObjectCache cache = ObjectCacheFactory.getCache(conf, queryId, false); + Object cached = cache.retrieve(ctx.getMjSmallTableCacheKey()); + if (cached == null) { + LOG.debug("ProbeDecode: no cached hash table for key {}", ctx.getMjSmallTableCacheKey()); + return DISABLED; + } + VectorMapJoinHashTable ht; + if (cached instanceof VectorMapJoinTableContainer) { + ht = ((VectorMapJoinTableContainer) cached).vectorMapJoinHashTable(); + } else if (cached instanceof VectorMapJoinHashTable) { + ht = (VectorMapJoinHashTable) cached; + } else if (cached instanceof MapJoinTableContainer) { + // Non-vectorized container -- probe-decode has no dictionary-of-keys to intersect with, + // so bail out and let the plain decode path run. + LOG.debug("ProbeDecode: cached container is non-vectorized ({}); skipping probe", + cached.getClass().getName()); + return DISABLED; + } else { + LOG.debug("ProbeDecode: unexpected cached object type {}", cached.getClass().getName()); + return DISABLED; + } + + if (ht instanceof VectorMapJoinLongHashTable) { + return new ParquetProbeDecodeState(keyIdx, ParquetProbeLongHashTable.of(ht)); + } + // TODO: bytes-key + multi-key variants when the corresponding ParquetProbeHashTable + // implementations land. + LOG.debug("ProbeDecode: no ParquetProbeHashTable for key type {}", ht.getClass().getName()); + return DISABLED; + } catch (Exception e) { + LOG.warn("ProbeDecode: hash-table resolution failed, falling back to plain decode", e); + return DISABLED; + } + } + + public boolean isEnabled() { + return probe != null; + } + + /** @return column index (within the reader's projected columns) of the probe key column. */ + public int getKeyColumnIndex() { + return keyColumnIndex; + } + + public ParquetProbeHashTable getProbe() { + return probe; + } +} diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/probe/ParquetProbeFilter.java b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/probe/ParquetProbeFilter.java new file mode 100644 index 000000000000..4767124f24f6 --- /dev/null +++ b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/probe/ParquetProbeFilter.java @@ -0,0 +1,104 @@ +/* + * 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.ql.io.parquet.vector.probe; + +/** + * Row-level selection state produced by the Parquet ProbeDecode path. + * + *

After the join-key column has been decoded and probed against the small-table hash table + * (see {@link ParquetProbeHashTable}), the surviving row positions within the current batch are + * recorded here so that the remaining non-key columns can be read via + * {@code VectorizedColumnReader.readBatch(total, column, type, ParquetProbeFilter)} and skip + * decode / conversion work for rows that will be filtered out anyway. + * + *

The filter is advisory for correctness: a reader is free to decode every row and + * ignore the filter (the default interface method does exactly that). Slots that are marked + * filtered-out must still be advanced in the underlying page state so subsequent reads stay + * aligned; concrete readers achieve that by calling {@code skip()} on the {@code ParquetDataColumnReader}. + * + *

Lifecycle within a batch: {@link ParquetProbeHashTable#probe} returns a filter in bitmap form + * (one boolean per row); per-row column readers query it via {@link #isSelected(int)} while + * decoding; once all columns are read, {@link VectorizedParquetRecordReader} calls + * {@link #compact(int)} to materialize a {@code selected[]} that + * {@link org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch} consumes downstream. The + * compacted form is not queried per row -- it flows directly into + * {@code VectorizedRowBatch.selected}. + */ +public final class ParquetProbeFilter { + + private final boolean[] bitmap; + private int[] selected; + private int selectedSize; + + private ParquetProbeFilter(boolean[] bitmap) { + this.bitmap = bitmap; + } + + public static ParquetProbeFilter newBitmap(boolean[] bitmap) { + if (bitmap == null) { + throw new IllegalArgumentException("bitmap must be non-null"); + } + return new ParquetProbeFilter(bitmap); + } + + /** + * Test whether the given row index survives the filter. + * + * @param rowId 0-based row index within the current batch + * @return {@code true} if the row should be decoded fully, {@code false} if the reader may skip + * decode / conversion for that row + */ + public boolean isSelected(int rowId) { + return rowId >= 0 && rowId < bitmap.length && bitmap[rowId]; + } + + /** + * Materialize a compact {@code selected[]} representation over the first {@code batchSize} rows + * from the underlying bitmap. Idempotent: repeated calls are cheap no-ops after the first. + * + * @return {@code this} for chaining + */ + public ParquetProbeFilter compact(int batchSize) { + if (selected != null) { + return this; + } + int upper = Math.min(batchSize, bitmap.length); + int[] out = new int[upper]; + int n = 0; + for (int i = 0; i < upper; i++) { + if (bitmap[i]) { + out[n++] = i; + } + } + this.selected = out; + this.selectedSize = n; + return this; + } + + /** @return the compact {@code selected[]} array; {@code null} until {@link #compact(int)} runs. */ + public int[] getSelected() { + return selected; + } + + /** @return the number of live entries in {@link #getSelected()}; {@code 0} until {@link #compact(int)} runs. */ + public int getSelectedSize() { + return selectedSize; + } +} diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/probe/ParquetProbeHashTable.java b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/probe/ParquetProbeHashTable.java new file mode 100644 index 000000000000..a92e46c9e119 --- /dev/null +++ b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/probe/ParquetProbeHashTable.java @@ -0,0 +1,51 @@ +/* + * 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.ql.io.parquet.vector.probe; + +import java.io.IOException; +import org.apache.hadoop.hive.ql.exec.vector.ColumnVector; + +/** + * Row-batch probe against a small-table hash table for the Parquet ProbeDecode path. + * + *

Given a {@link ColumnVector} that carries the just-decoded join-key column, implementations + * probe each row against the small-side hash table and produce a {@link ParquetProbeFilter} whose + * bitmap marks the rows that survive the join. Subsequent non-key columns in the same batch are + * then read via {@code VectorizedColumnReader.readBatch(..., ParquetProbeFilter)} so their values + * skip decode / conversion work for filtered-out rows. + * + *

Structured after the ORC-side {@code OrcProbeHashTable} family: the key type-specialisations + * (single long/int, single string, multi-key) live in their own subclasses so the per-row probe + * stays a monomorphic virtual call. This first cut ships only the long/int variant + * ({@link ParquetProbeLongHashTable}); the string and multi-key variants will land later. + */ +public interface ParquetProbeHashTable { + + /** + * Probe every row in {@code keyColumn} (positions 0..batchSize-1) against the small-side hash + * table and return a filter over the surviving rows. Rows marked null in the key column never + * match (join semantics: NULL != NULL). + * + * @param keyColumn key-column vector, already decoded by the primary readBatch() call + * @param batchSize number of rows in the current batch + * @return a bitmap-form {@link ParquetProbeFilter} of length {@code batchSize} + */ + ParquetProbeFilter probe(ColumnVector keyColumn, int batchSize) throws IOException; +} diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/probe/ParquetProbeLongHashTable.java b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/probe/ParquetProbeLongHashTable.java new file mode 100644 index 000000000000..4c5f4a10fdf2 --- /dev/null +++ b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/probe/ParquetProbeLongHashTable.java @@ -0,0 +1,165 @@ +/* + * 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.ql.io.parquet.vector.probe; + +import java.io.IOException; +import java.util.Arrays; + +import org.apache.hadoop.hive.ql.exec.JoinUtil; +import org.apache.hadoop.hive.ql.exec.vector.ColumnVector; +import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector; +import org.apache.hadoop.hive.ql.exec.vector.mapjoin.hashtable.VectorMapJoinHashMapResult; +import org.apache.hadoop.hive.ql.exec.vector.mapjoin.hashtable.VectorMapJoinHashMultiSetResult; +import org.apache.hadoop.hive.ql.exec.vector.mapjoin.hashtable.VectorMapJoinHashSetResult; +import org.apache.hadoop.hive.ql.exec.vector.mapjoin.hashtable.VectorMapJoinHashTable; +import org.apache.hadoop.hive.ql.exec.vector.mapjoin.hashtable.VectorMapJoinLongHashMap; +import org.apache.hadoop.hive.ql.exec.vector.mapjoin.hashtable.VectorMapJoinLongHashMultiSet; +import org.apache.hadoop.hive.ql.exec.vector.mapjoin.hashtable.VectorMapJoinLongHashSet; +import org.apache.hadoop.hive.ql.exec.vector.mapjoin.hashtable.VectorMapJoinLongHashTable; + +/** + * Probe a {@link LongColumnVector} against a single-long-key + * {@link org.apache.hadoop.hive.ql.exec.vector.mapjoin.hashtable.VectorMapJoinLongHashTable}. + * + *

Covers the three concrete long-key hash table flavours emitted by Hive for a mapjoin: + *

+ * All three funnel their probe result through {@code JoinUtil.JoinResult}; a row survives when + * the probe returns {@code MATCH} (and, for a multi-set/set, {@code SPILL} which we treat as a + * hit to be safe -- we prefer over-decoding to under-decoding). + * + *

The probe iterates every row exactly once, so it is O(batchSize). NULL keys never match, + * matching Hive's join semantics. + */ +public final class ParquetProbeLongHashTable implements ParquetProbeHashTable { + + private enum Kind { MAP, MULTISET, SET } + + private final Kind kind; + private final VectorMapJoinLongHashMap map; + private final VectorMapJoinLongHashMultiSet multiSet; + private final VectorMapJoinLongHashSet set; + private final boolean useMinMax; + private final long min; + private final long max; + private final VectorMapJoinHashMapResult mapResult; + private final VectorMapJoinHashMultiSetResult multiSetResult; + private final VectorMapJoinHashSetResult setResult; + + public static ParquetProbeLongHashTable of(VectorMapJoinHashTable ht) { + if (ht instanceof VectorMapJoinLongHashMap) { + return new ParquetProbeLongHashTable(Kind.MAP, (VectorMapJoinLongHashTable) ht); + } + if (ht instanceof VectorMapJoinLongHashMultiSet) { + return new ParquetProbeLongHashTable(Kind.MULTISET, (VectorMapJoinLongHashTable) ht); + } + if (ht instanceof VectorMapJoinLongHashSet) { + return new ParquetProbeLongHashTable(Kind.SET, (VectorMapJoinLongHashTable) ht); + } + throw new IllegalArgumentException("Not a long-key hash table: " + ht.getClass().getName()); + } + + private ParquetProbeLongHashTable(Kind kind, VectorMapJoinLongHashTable longHT) { + this.kind = kind; + this.useMinMax = longHT.useMinMax(); + this.min = longHT.min(); + this.max = longHT.max(); + if (kind == Kind.MAP) { + this.map = (VectorMapJoinLongHashMap) longHT; + this.multiSet = null; + this.set = null; + this.mapResult = map.createHashMapResult(); + this.multiSetResult = null; + this.setResult = null; + } else if (kind == Kind.MULTISET) { + this.map = null; + this.multiSet = (VectorMapJoinLongHashMultiSet) longHT; + this.set = null; + this.mapResult = null; + this.multiSetResult = multiSet.createHashMultiSetResult(); + this.setResult = null; + } else { + this.map = null; + this.multiSet = null; + this.set = (VectorMapJoinLongHashSet) longHT; + this.mapResult = null; + this.multiSetResult = null; + this.setResult = set.createHashSetResult(); + } + } + + @Override + public ParquetProbeFilter probe(ColumnVector keyColumn, int batchSize) throws IOException { + if (!(keyColumn instanceof LongColumnVector)) { + throw new IllegalArgumentException( + "Expected LongColumnVector for long-key probe, got " + keyColumn.getClass().getName()); + } + LongColumnVector v = (LongColumnVector) keyColumn; + boolean[] bitmap = new boolean[batchSize]; + + if (v.isRepeating) { + // Single-value fast path: probe once, splat the result over the whole batch. Cheap dictionary + // encoded columns land here often. + boolean nullKey = !v.noNulls && v.isNull[0]; + boolean hit = !nullKey && probeOne(v.vector[0]); + if (hit) { + Arrays.fill(bitmap, true); + } + // else all filtered out; bitmap already false + return ParquetProbeFilter.newBitmap(bitmap); + } + + for (int i = 0; i < batchSize; i++) { + if (!v.noNulls && v.isNull[i]) { + continue; // NULL keys never match + } + long key = v.vector[i]; + if (useMinMax && (key < min || key > max)) { + continue; // small-table min/max exclusion -- pure arithmetic, no hash-table touch + } + bitmap[i] = probeOne(key); + } + return ParquetProbeFilter.newBitmap(bitmap); + } + + private boolean probeOne(long key) throws IOException { + JoinUtil.JoinResult r; + switch (kind) { + case MAP: + r = map.lookup(key, mapResult); + break; + case MULTISET: + r = multiSet.contains(key, multiSetResult); + break; + case SET: + r = set.contains(key, setResult); + break; + default: + throw new AssertionError(kind); + } + // Treat SPILL as a hit: if the small-table row is on a spilled partition we can't know + // whether it matches without going to disk, so we let the row through and let the join + // operator handle it downstream -- prefer over-decoding to filtering a valid row. + return r == JoinUtil.JoinResult.MATCH || r == JoinUtil.JoinResult.SPILL; + } +} From b468102aa6b069f00712847e99e16099d4079856 Mon Sep 17 00:00:00 2001 From: Laszlo Bodor Date: Sat, 5 Sep 2026 09:05:06 +0200 Subject: [PATCH 2/2] HIVE-30019: Tests and JMH benchmark for Parquet ProbeDecode Adds three test artifacts covering the Parquet ProbeDecode path landed in HIVE-30019: 1. Q-test (probedecode_mapjoin_simple_parquet.q) mirroring the ORC template probedecode_mapjoin_simple.q. Sets up item_dim_pq + orders_fact_pq stored as parquet, runs the join both with and without hive.optimize.scan.probedecode so the golden captures the plan (with EXPLAIN VECTORIZATION DETAIL) and asserts the result set matches the baseline. Auto-picked up by MiniLlapLocalCliConfig's sweep of ql/src/test/queries/clientpositive; the .q.out golden will be generated on first CI run. 2. TestParquetProbeFilter (10 tests) -- pure unit test of the ParquetProbeFilter contract: newBitmap null-guard, isSelected bounds and bitmap fidelity, compact() idempotency, and correct materialization of selected[] under all-accept / all-reject / mixed / empty / smaller-batch cases. 3. TestVectorizedParquetProbeDecodeReader (4 tests) -- end-to-end test that writes a Parquet file, opens VectorizedParquetRecordReader, reflects out the per-column readers, and drives the filter-aware readBatch signature directly (bypassing the MapJoin-operator plumbing that nextBatch needs to resolve ProbeDecodeState). Verifies: - Surviving rows decode correctly across int / long / double / string. - Filtered rows come back as null-marked slots. - noNulls is cleared once any filtered row is emitted. - allPass filter produces vectors identical to the unfiltered baseline. Runs on both dictionary-encoded pages (exercises readDictionaryIDs -> pendingSkip -> RunLengthBitPackingHybridDecoder.skipInts fast-path) and PLAIN-encoded pages (per-row dataColumn.skip()). 4. VectorizedParquetProbeDecodeBench (JMH) -- crosses two encodings (dict, plain) with four filter shapes (no-filter baseline, all-pass, all-fail, half). Establishes the regression floor for the isFilteredOut branch added to every primitive read helper and quantifies the win on filter= all-fail (dict path drops entire runs via skipInts). Test results: - TestParquetProbeFilter: 10/10 pass - TestVectorizedParquetProbeDecodeReader: 4/4 pass - No regression on TestVectorizedColumnReader (19), TestVectorized- DictionaryEncodingColumnReader (13), TestVectorizedListColumnReader (6), TestVectorizedMapColumnReader (5) - itests/hive-jmh compiles clean Signed-off-by: Laszlo Bodor --- .../org/apache/hadoop/hive/conf/HiveConf.java | 9 + itests/hive-jmh/pom.xml | 22 + .../parquet/VectorizedParquetReadBench.java | 385 ++++++++++++++++++ .../vector/VectorizedParquetRecordReader.java | 16 +- .../VectorizedPrimitiveColumnReader.java | 69 +++- .../vector/probe/ParquetProbeDecodeState.java | 96 ++++- .../probe/TestParquetProbeDecodeState.java | 229 +++++++++++ .../vector/probe/TestParquetProbeFilter.java | 143 +++++++ ...estVectorizedParquetProbeDecodeReader.java | 329 +++++++++++++++ .../probedecode_mapjoin_simple_parquet.q | 53 +++ .../probedecode_mapjoin_simple_parquet.q.out | 300 ++++++++++++++ 11 files changed, 1613 insertions(+), 38 deletions(-) create mode 100644 itests/hive-jmh/src/main/java/org/apache/hive/benchmark/vectorization/parquet/VectorizedParquetReadBench.java create mode 100644 ql/src/test/org/apache/hadoop/hive/ql/io/parquet/vector/probe/TestParquetProbeDecodeState.java create mode 100644 ql/src/test/org/apache/hadoop/hive/ql/io/parquet/vector/probe/TestParquetProbeFilter.java create mode 100644 ql/src/test/org/apache/hadoop/hive/ql/io/parquet/vector/probe/TestVectorizedParquetProbeDecodeReader.java create mode 100644 ql/src/test/queries/clientpositive/probedecode_mapjoin_simple_parquet.q create mode 100644 ql/src/test/results/clientpositive/llap/probedecode_mapjoin_simple_parquet.q.out diff --git a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java index b8aea76155a9..4931594e0d93 100644 --- a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java +++ b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java @@ -2667,6 +2667,15 @@ public static enum ConfVars { + "e.g., use the cached MapJoin hashtable created on the small table side to filter out row columns that are not going " + "to be used when reading the large table data. This will result less CPU cycles spent for decoding unused data."), + HIVE_OPTIMIZE_SCAN_PROBEDECODE_PARQUET_PLAIN_FILTER( + "hive.optimize.scan.probedecode.parquet.plain.filter.enabled", true, + "When ProbeDecode is on, gates the per-row filter check applied to Parquet PLAIN-encoded pages. \n" + + "The check trades a small per-row overhead for skipping the value decode + null-set on filtered rows; \n" + + "at very selective join hit rates (~5-15%, common in TPC-DS) it wins by ~2-3%, at inclusive hit rates \n" + + "(>50%) it can regress the read path by 2-4%. The dictionary-encoded path is unaffected and always \n" + + "honours the filter via the bulk-skip fast-path. Disable this only on workloads where PLAIN-encoded \n" + + "fact columns dominate and hash-join hit rates are typically high; see HIVE-30019 for the crossover data."), + HIVE_OPTIMIZE_HMS_QUERY_CACHE_ENABLED("hive.optimize.metadata.query.cache.enabled", true, "This property enables caching metadata for repetitive requests on a per-query basis"), diff --git a/itests/hive-jmh/pom.xml b/itests/hive-jmh/pom.xml index 196882ecefdb..b07556077aea 100644 --- a/itests/hive-jmh/pom.xml +++ b/itests/hive-jmh/pom.xml @@ -98,6 +98,24 @@ ${mockito-inline.version} + + + + org.apache.maven.plugins + maven-compiler-plugin + + full + + + org.openjdk.jmh + jmh-generator-annprocess + ${jmh.version} + + + + + + perf @@ -118,7 +136,11 @@ org.openjdk.jmh.Main + + true + + diff --git a/itests/hive-jmh/src/main/java/org/apache/hive/benchmark/vectorization/parquet/VectorizedParquetReadBench.java b/itests/hive-jmh/src/main/java/org/apache/hive/benchmark/vectorization/parquet/VectorizedParquetReadBench.java new file mode 100644 index 000000000000..6412df3b7d36 --- /dev/null +++ b/itests/hive-jmh/src/main/java/org/apache/hive/benchmark/vectorization/parquet/VectorizedParquetReadBench.java @@ -0,0 +1,385 @@ +/* + * 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.hive.benchmark.vectorization.parquet; + +import java.io.File; +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.nio.file.Files; +import java.util.concurrent.TimeUnit; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.conf.HiveConf; +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.DoubleColumnVector; +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.parquet.vector.VectorizedColumnReader; +import org.apache.hadoop.hive.ql.io.parquet.vector.VectorizedParquetRecordReader; +import org.apache.hadoop.hive.ql.io.parquet.vector.probe.ParquetProbeFilter; +import org.apache.hadoop.hive.ql.plan.MapWork; +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.mapred.JobConf; +import org.apache.hadoop.mapreduce.Job; +import org.apache.parquet.example.data.Group; +import org.apache.parquet.example.data.simple.SimpleGroupFactory; +import org.apache.parquet.hadoop.ParquetInputFormat; +import org.apache.parquet.hadoop.ParquetWriter; +import org.apache.parquet.hadoop.example.GroupWriteSupport; +import org.apache.parquet.hadoop.metadata.CompressionCodecName; +import org.apache.parquet.io.api.Binary; +import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.MessageTypeParser; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Param; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.TearDown; +import org.openjdk.jmh.annotations.Warmup; +import org.openjdk.jmh.infra.Blackhole; +import org.openjdk.jmh.runner.Runner; +import org.openjdk.jmh.runner.options.Options; +import org.openjdk.jmh.runner.options.OptionsBuilder; + +/** + * JMH benchmark that compares the pre-patch decode path (a bare 3-arg + * {@code readBatch(total, col, type)} call — what {@code VectorizedParquetRecordReader.nextBatch} + * would issue before HIVE-30019) against the post-patch filter-aware path + * ({@code readBatch(total, col, type, ParquetProbeFilter)} with a supplied bitmap) across a + * sweep of filter selectivities. + * + *

A single {@code @Benchmark} method ({@link #readBatch}) drives every projected column + * through {@link #BATCHES_PER_INVOCATION} batches per invocation; the {@link #filter} @Param + * chooses which shape to call: + *

+ * + *

Two encodings are covered: + *

+ * + *

Note on worst case: the current sweep uses a clumpy shape ({@code (i % 100) < P}), + * which matches realistic ProbeDecode join-key hit distributions and is favourable to both the + * dict bulk-skip and to branch prediction on PLAIN. An alternating half-filter + * ({@code i % 2 == 0}) is the branch-predictor worst case but is not the shape ProbeDecode + * produces in practice. + * + *

Run: {@code + * java -jar itests/hive-jmh/target/benchmarks.jar + * org.apache.hive.benchmark.vectorization.parquet.VectorizedParquetReadBench + * -wi 5 -i 15 -f 2 -bm avgt -tu us + * } + */ +@BenchmarkMode(Mode.AverageTime) +@OutputTimeUnit(TimeUnit.MICROSECONDS) +@Warmup(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS) +@Measurement(iterations = 10, time = 1, timeUnit = TimeUnit.SECONDS) +@Fork(1) +@State(Scope.Benchmark) +public class VectorizedParquetReadBench { + + // Enough rows in a single row group that the fixed per-invocation cost -- reader open, split + // init, checkEndOfRowGroup, JobConf lookup -- is amortised across many readBatch calls. With + // 131072 rows (128 batches of 1024) written into one row group, each invocation drives ~128 + // filter-aware calls per column vs one, so the isFilteredOut / skip fast-path signal is no + // longer diluted by the ~5 ms setup floor. + private static final int N_ROWS = 131072; + private static final int BATCHES_PER_INVOCATION = N_ROWS / VectorizedRowBatch.DEFAULT_SIZE; + private static final MessageType WRITE_SCHEMA = MessageTypeParser.parseMessageType( + "message pd_read { " + + "required int32 int_col; " + + "required int64 long_col; " + + "required double dbl_col; " + + "required binary str_col (UTF8); " + + "}"); + + @Param({"dict", "plain"}) + public String encoding; + + /** + * Selectivity sweep. {@code none} = call the 3-arg readBatch (baseline / pre-patch shape); + * every other value is an integer pass-percentage in {@code [0, 100]} used to build a bitmap + * where the first {@code p} rows of every 100-row block accept and the remaining + * {@code 100 - p} reject. Clumpy, not alternating: matches real-world join-key hit distributions + * and lets the dict {@code readDictionaryIDs} bulk-skip coalesce the reject runs. {@code 50} + * with a clumpy shape is not the same worst-case as an alternating half-filter (see the + * "worst-case" note in the class javadoc). + */ + @Param({"none", "10", "50", "90"}) + public String filter; + + /** + * Toggles {@code hive.optimize.scan.probedecode.parquet.plain.filter.enabled}. When + * {@code off}, the PLAIN-path {@code isFilteredOutPlain} check is constant-folded away by the + * JIT and every row on a PLAIN page is materialised; the dictionary path is unaffected. + * Included as a bench param so the JIT-elimination claim can be verified end-to-end (i.e. the + * {@code plain × × off} cells should match {@code plain × none} within noise). + * + *

Ignored when {@code filter=none}: the 3-arg readBatch path never calls the check + * regardless of the config, so a {@code none × off} run is not informative and only widens + * the sweep matrix. + */ + @Param({"on", "off"}) + public String plainFilter; + + private File dataDir; + private Path dataFile; + private JobConf jobConf; + + private LongColumnVector intVec; + private LongColumnVector longVec; + private DoubleColumnVector dblVec; + private BytesColumnVector strVec; + + private TypeInfo intType; + private TypeInfo longType; + private TypeInfo dblType; + private TypeInfo strType; + + /** + * Filter for this trial, built from the {@link #filter} @Param. {@code null} when + * {@code filter=none}, in which case the bench calls the 3-arg readBatch (pre-patch shape). + */ + private ParquetProbeFilter probeFilter; + + @Setup(Level.Trial) + public void setUp() throws Exception { + dataDir = Files.createTempDirectory("pd-read-").toFile(); + dataFile = new Path(new File(dataDir, "data.parquet").toURI()); + + boolean dict = "dict".equals(encoding); + + Configuration writeConf = new Configuration(); + GroupWriteSupport.setSchema(WRITE_SCHEMA, writeConf); + SimpleGroupFactory gf = new SimpleGroupFactory(WRITE_SCHEMA); + // Row-group size: 256 MB so all N_ROWS live in a single row group. Page size: 1 MB so each + // column still has several pages inside the row group (exercises page-boundary handling + // inside readBatch). Dictionary size: 1 MB so 8 distinct values easily fit. + try (ParquetWriter writer = new ParquetWriter<>(dataFile, new GroupWriteSupport(), + CompressionCodecName.UNCOMPRESSED, 256 * 1024 * 1024, 1024 * 1024, 1024 * 1024, + dict, false, ParquetWriter.DEFAULT_WRITER_VERSION, writeConf)) { + for (int i = 0; i < N_ROWS; i++) { + int intV = dict ? (i % 8) : i; + long lngV = dict ? (i % 8) : (long) i; + double dblV = dict ? (i % 8) : (double) i; + String strV = dict ? ("v" + (i % 8)) : ("v" + i); + writer.write(gf.newGroup() + .append("int_col", intV) + .append("long_col", lngV) + .append("dbl_col", dblV) + .append("str_col", Binary.fromString(strV))); + } + } + + jobConf = buildJobConf(); + + int batchSize = VectorizedRowBatch.DEFAULT_SIZE; + intVec = new LongColumnVector(batchSize); + longVec = new LongColumnVector(batchSize); + dblVec = new DoubleColumnVector(batchSize); + strVec = new BytesColumnVector(batchSize); + + intType = TypeInfoFactory.getPrimitiveTypeInfo("int"); + longType = TypeInfoFactory.getPrimitiveTypeInfo("bigint"); + dblType = TypeInfoFactory.getPrimitiveTypeInfo("double"); + strType = TypeInfoFactory.getPrimitiveTypeInfo("string"); + + probeFilter = buildFilter(filter, batchSize); + } + + /** + * Build a {@link ParquetProbeFilter} matching the {@link #filter} @Param, or {@code null} for + * {@code "none"} (baseline, 3-arg readBatch). + * + *

Accept bits are clumpy: for pass-percentage {@code p}, the first {@code p} rows of each + * 100-row block accept and the remaining {@code 100 - p} reject. Realistic ProbeDecode + * hit-distributions come in runs, and this shape lets the dict {@code readDictionaryIDs} + * bulk-skip coalesce a whole reject run into a single {@code skipInts} call. An alternating + * half-filter would be the branch-predictor worst case but is not the shape ProbeDecode + * produces in practice. + */ + private static ParquetProbeFilter buildFilter(String mode, int batchSize) { + if ("none".equals(mode)) { + return null; + } + int pass = Integer.parseInt(mode); + if (pass < 0 || pass > 100) { + throw new IllegalArgumentException("filter pass % must be in [0, 100], got: " + mode); + } + boolean[] bits = new boolean[batchSize]; + for (int i = 0; i < batchSize; i++) { + bits[i] = (i % 100) < pass; + } + return ParquetProbeFilter.newBitmap(bits); + } + + private JobConf buildJobConf() throws Exception { + Configuration conf = new Configuration(); + conf.set(IOConstants.COLUMNS, "int_col,long_col,dbl_col,str_col"); + conf.set(IOConstants.COLUMNS_TYPES, "int,bigint,double,string"); + conf.setBoolean(ColumnProjectionUtils.READ_ALL_COLUMNS, false); + conf.set(ColumnProjectionUtils.READ_COLUMN_IDS_CONF_STR, "0,1,2,3"); + HiveConf.setBoolVar(conf, HiveConf.ConfVars.HIVE_VECTORIZATION_ENABLED, true); + HiveConf.setVar(conf, HiveConf.ConfVars.PLAN, "//tmp"); + HiveConf.setBoolVar(conf, + HiveConf.ConfVars.HIVE_OPTIMIZE_SCAN_PROBEDECODE_PARQUET_PLAIN_FILTER, + "on".equals(plainFilter)); + + MapWork mapWork = new MapWork(); + VectorizedRowBatchCtx rbCtx = new VectorizedRowBatchCtx(); + rbCtx.init(org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory + .getStandardStructObjectInspector( + java.util.Collections.emptyList(), + java.util.Collections.emptyList()), + new String[0]); + mapWork.setVectorMode(true); + mapWork.setVectorizedRowBatchCtx(rbCtx); + Utilities.setMapWork(conf, mapWork); + + Job job = new Job(conf, "pd-read"); + ParquetInputFormat.setInputPaths(job, dataFile); + return new JobConf(conf); + } + + private VectorizedParquetRecordReader openReader() throws Exception { + Job job = new Job(jobConf, "pd-read-split"); + ParquetInputFormat.setInputPaths(job, dataFile); + ParquetInputFormat pif = new ParquetInputFormat<>( + org.apache.parquet.hadoop.example.GroupReadSupport.class); + org.apache.hadoop.mapreduce.InputSplit inputSplit = pif.getSplits(job).get(0); + org.apache.hadoop.mapred.FileSplit fs = new org.apache.hadoop.mapred.FileSplit(dataFile, 0L, + inputSplit.getLength(), inputSplit.getLocations()); + return new VectorizedParquetRecordReader(fs, jobConf); + } + + @SuppressWarnings("unchecked") + private VectorizedColumnReader[] primeReaders(VectorizedParquetRecordReader r) throws Exception { + Method m = VectorizedParquetRecordReader.class.getDeclaredMethod("checkEndOfRowGroup"); + m.setAccessible(true); + m.invoke(r); + Field f = VectorizedParquetRecordReader.class.getDeclaredField("columnReaders"); + f.setAccessible(true); + return (VectorizedColumnReader[]) f.get(r); + } + + private void resetVectors() { + intVec.reset(); + intVec.init(); + longVec.reset(); + longVec.init(); + dblVec.reset(); + dblVec.init(); + strVec.reset(); + strVec.init(); + } + + @TearDown(Level.Trial) + public void tearDown() { + File f = new File(dataFile.toUri()); + if (f.exists()) { + f.delete(); + } + if (dataDir != null && dataDir.exists()) { + dataDir.delete(); + } + } + + /** + * Drive {@link VectorizedColumnReader#readBatch} for every projected column across all + * {@link #BATCHES_PER_INVOCATION} batches in the row group. When {@link #probeFilter} is + * {@code null} ({@code filter=none}), calls the 3-arg readBatch (pre-patch shape); otherwise + * calls the 4-arg filter-aware readBatch. The branch is once per invocation, not per row -- + * JMH resolves it as a single conditional at the top and the loop bodies run without further + * decisions. + * + *

The fixed per-invocation cost (openReader, split init, checkEndOfRowGroup, JobConf lookup + * -- collectively ~5 ms in earlier runs) is amortised over ~128 readBatch calls per column so + * the isFilteredOut / skip fast-path signal is not diluted by setup. + */ + @Benchmark + public void readBatch(Blackhole bh) throws Exception { + try (VectorizedParquetRecordReader r = openReader()) { + VectorizedColumnReader[] readers = primeReaders(r); + int n = VectorizedRowBatch.DEFAULT_SIZE; + if (probeFilter == null) { + for (int b = 0; b < BATCHES_PER_INVOCATION; b++) { + resetVectors(); + readers[0].readBatch(n, intVec, intType); + readers[1].readBatch(n, longVec, longType); + readers[2].readBatch(n, dblVec, dblType); + readers[3].readBatch(n, strVec, strType); + consume(bh); + } + } else { + for (int b = 0; b < BATCHES_PER_INVOCATION; b++) { + resetVectors(); + readers[0].readBatch(n, intVec, intType, probeFilter); + readers[1].readBatch(n, longVec, longType, probeFilter); + readers[2].readBatch(n, dblVec, dblType, probeFilter); + readers[3].readBatch(n, strVec, strType, probeFilter); + consume(bh); + } + } + } + } + + private void consume(Blackhole bh) { + bh.consume(intVec.vector); + bh.consume(longVec.vector); + bh.consume(dblVec.vector); + bh.consume(strVec.vector); + bh.consume(intVec.isNull); + bh.consume(longVec.isNull); + bh.consume(dblVec.isNull); + bh.consume(strVec.isNull); + } + + public static void main(String[] args) throws Exception { + Options opt = new OptionsBuilder() + .include(VectorizedParquetReadBench.class.getSimpleName()) + .build(); + new Runner(opt).run(); + } +} diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedParquetRecordReader.java b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedParquetRecordReader.java index cdd441712f25..8c4014fb40bb 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedParquetRecordReader.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedParquetRecordReader.java @@ -165,6 +165,14 @@ public class VectorizedParquetRecordReader extends ParquetRecordReaderBase private ParquetProbeDecodeState probeState = ParquetProbeDecodeState.disabled(); private boolean probeStateResolved = false; + /** + * Snapshot of {@link HiveConf.ConfVars#HIVE_OPTIMIZE_SCAN_PROBEDECODE_PARQUET_PLAIN_FILTER} at + * reader construction time. Threaded through into every {@link VectorizedPrimitiveColumnReader} + * so the JIT can constant-fold the PLAIN-path filter check when it's off. Only affects PLAIN + * pages; the dictionary path always honours the filter via its bulk-skip fast-path. + */ + private final boolean plainFilterEnabled; + public VectorizedParquetRecordReader(InputSplit oldInputSplit, JobConf conf) throws IOException { this(oldInputSplit, conf, null, null, null); } @@ -173,6 +181,8 @@ public VectorizedParquetRecordReader(InputSplit oldInputSplit, JobConf conf, Fil DataCache dataCache, Configuration cacheConf, ParquetMetadata parquetMetadata, Map initialDefaults) throws IOException { super(conf, oldInputSplit); + this.plainFilterEnabled = HiveConf.getBoolVar( + conf, ConfVars.HIVE_OPTIMIZE_SCAN_PROBEDECODE_PARQUET_PLAIN_FILTER); try { this.metadataCache = metadataCache; this.cache = dataCache; @@ -437,10 +447,6 @@ private boolean nextBatch(VectorizedRowBatch columnarBatch) throws IOException { // column list. Failing here is fine -- state stays disabled and we run the plain path. probeState = ParquetProbeDecodeState.of(jobConf, Utilities.getMapWork(jobConf), columnNamesList); probeStateResolved = true; - if (probeState.isEnabled()) { - LOG.info("ProbeDecode enabled for VectorizedParquetRecordReader: keyColIdx={}", - probeState.getKeyColumnIndex()); - } } int num = (int) Math.min(VectorizedRowBatch.DEFAULT_SIZE, totalCountLoadedSoFar - rowsReturned); @@ -648,7 +654,7 @@ private VectorizedColumnReader buildVectorizedParquetReader( } return new VectorizedPrimitiveColumnReader(descriptors.get(0), pages.getPageReader(descriptors.get(0)), skipTimestampConversion, writerTimezone, skipProlepticConversion, - legacyConversionEnabled, type, typeInfo); + legacyConversionEnabled, type, typeInfo, plainFilterEnabled); case STRUCT: StructTypeInfo structTypeInfo = (StructTypeInfo) typeInfo; List fieldReaders = new ArrayList<>(); diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedPrimitiveColumnReader.java b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedPrimitiveColumnReader.java index 883e301b4cc7..1f7e8e619dd4 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedPrimitiveColumnReader.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedPrimitiveColumnReader.java @@ -61,6 +61,18 @@ public class VectorizedPrimitiveColumnReader extends BaseVectorizedColumnReader */ private ParquetProbeFilter probeFilter; + /** + * When {@code false}, the PLAIN-path helpers ignore the filter and materialise every row, + * relying on downstream {@code batch.selected[]} for filtering. Controlled by + * {@code hive.optimize.scan.probedecode.parquet.plain.filter.enabled}. The dictionary path + * is unaffected and always honours the filter (its bulk-skip fast-path is uniformly a win). + * + *

Marked {@code final} so the JIT can constant-fold the check away in {@link + * #isFilteredOutPlain} when disabled -- there should be zero residual cost on the hot path + * when the config is off. + */ + private final boolean plainFilterEnabled; + public VectorizedPrimitiveColumnReader( ColumnDescriptor descriptor, PageReader pageReader, @@ -69,10 +81,12 @@ public VectorizedPrimitiveColumnReader( boolean skipProlepticConversion, boolean legacyConversionEnabled, Type type, - TypeInfo hiveType) + TypeInfo hiveType, + boolean plainFilterEnabled) throws IOException { super(descriptor, pageReader, skipTimestampConversion, writerTimezone, skipProlepticConversion, legacyConversionEnabled, type, hiveType); + this.plainFilterEnabled = plainFilterEnabled; } @Override @@ -90,14 +104,27 @@ public void readBatch( } /** - * @return {@code true} when a {@link ParquetProbeFilter} is active and marks row {@code rowId} - * as filtered-out (i.e. the hash-table probe did not match this row). Returns - * {@code false} otherwise, including when no filter is active. + * Dict-path filter check. Always honours an active filter -- filtered dict-ids feed into the + * {@code pendingSkip} coalescing loop that reaches {@code DictionaryValuesReader.skip(int)} -> + * {@code RunLengthBitPackingHybridDecoder.skipInts}, which drops whole reject runs in O(runs). + * This win is uniform across selectivities, so no config gates it. */ - private boolean isFilteredOut(int rowId) { + private boolean isFilteredOutDict(int rowId) { return probeFilter != null && !probeFilter.isSelected(rowId); } + /** + * PLAIN-path filter check. Gated by {@link #plainFilterEnabled}: when the config is off, the + * JIT constant-folds this to always return {@code false} and the surrounding {@code if/else} + * in every {@code readXxx} helper collapses to just the materialise branch. The trade-off is + * that filtered PLAIN rows will then be materialised into the column vector (downstream + * {@code batch.selected[]} still filters them out for the query result, but callers reading + * the vector directly will see decoded values in those slots). + */ + private boolean isFilteredOutPlain(int rowId) { + return plainFilterEnabled && probeFilter != null && !probeFilter.isSelected(rowId); + } + @Override public void readBatch( int total, @@ -209,7 +236,7 @@ private void readDictionaryIDs(int total, LongColumnVector c, int rowId) { while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - if (isFilteredOut(rowId)) { + if (isFilteredOutDict(rowId)) { pendingSkip++; setNullValue(c, rowId); } else { @@ -242,7 +269,7 @@ private void readIntegers(int total, LongColumnVector c, int rowId) { while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - if (isFilteredOut(rowId)) { + if (isFilteredOutPlain(rowId)) { dataColumn.skip(); setNullValue(c, rowId); } else { @@ -268,7 +295,7 @@ private void readSmallInts(int total, LongColumnVector c, int rowId) { while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - if (isFilteredOut(rowId)) { + if (isFilteredOutPlain(rowId)) { dataColumn.skip(); setNullValue(c, rowId); } else { @@ -294,7 +321,7 @@ private void readTinyInts(int total, LongColumnVector c, int rowId) { while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - if (isFilteredOut(rowId)) { + if (isFilteredOutPlain(rowId)) { dataColumn.skip(); setNullValue(c, rowId); } else { @@ -320,7 +347,7 @@ private void readDoubles(int total, DoubleColumnVector c, int rowId) { while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - if (isFilteredOut(rowId)) { + if (isFilteredOutPlain(rowId)) { dataColumn.skip(); setNullValue(c, rowId); } else { @@ -346,7 +373,7 @@ private void readBooleans(int total, LongColumnVector c, int rowId) { while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - if (isFilteredOut(rowId)) { + if (isFilteredOutPlain(rowId)) { dataColumn.skip(); setNullValue(c, rowId); } else { @@ -369,7 +396,7 @@ private void readLongs(int total, LongColumnVector c, int rowId) { while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - if (isFilteredOut(rowId)) { + if (isFilteredOutPlain(rowId)) { dataColumn.skip(); setNullValue(c, rowId); } else { @@ -395,7 +422,7 @@ private void readFloats(int total, DoubleColumnVector c, int rowId) { while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - if (isFilteredOut(rowId)) { + if (isFilteredOutPlain(rowId)) { dataColumn.skip(); setNullValue(c, rowId); } else { @@ -424,7 +451,7 @@ private void readDecimal(int total, DecimalColumnVector c, int rowId) { while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - if (isFilteredOut(rowId)) { + if (isFilteredOutPlain(rowId)) { dataColumn.skip(); setNullValue(c, rowId); } else { @@ -450,7 +477,7 @@ private void readString(int total, BytesColumnVector c, int rowId) { while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - if (isFilteredOut(rowId)) { + if (isFilteredOutPlain(rowId)) { // Avoids the per-row byte-array allocation + copy into BytesColumnVector for filtered // rows. The underlying BinaryPlainValuesReader.skip() only reads the length prefix // and advances the buffer by that many bytes. @@ -475,7 +502,7 @@ private void readChar(int total, BytesColumnVector c, int rowId) { while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - if (isFilteredOut(rowId)) { + if (isFilteredOutPlain(rowId)) { dataColumn.skip(); setNullValue(c, rowId); } else { @@ -497,7 +524,7 @@ private void readVarchar(int total, BytesColumnVector c, int rowId) { while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - if (isFilteredOut(rowId)) { + if (isFilteredOutPlain(rowId)) { dataColumn.skip(); setNullValue(c, rowId); } else { @@ -519,7 +546,7 @@ private void readBinaries(int total, BytesColumnVector c, int rowId) { while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - if (isFilteredOut(rowId)) { + if (isFilteredOutPlain(rowId)) { dataColumn.skip(); setNullValue(c, rowId); } else { @@ -542,7 +569,7 @@ private void readDate(int total, DateColumnVector c, int rowId) { while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - if (isFilteredOut(rowId)) { + if (isFilteredOutPlain(rowId)) { dataColumn.skip(); setNullValue(c, rowId); } else { @@ -570,7 +597,7 @@ private void readTimestamp(int total, TimestampColumnVector c, int rowId) throws while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - if (isFilteredOut(rowId)) { + if (isFilteredOutPlain(rowId)) { dataColumn.skip(); setNullValue(c, rowId); } else { @@ -840,7 +867,7 @@ private void readDecimal64(int total, Decimal64ColumnVector c, int rowId) { while (left > 0) { readRepetitionAndDefinitionLevels(); if (definitionLevel >= maxDefLevel) { - if (isFilteredOut(rowId)) { + if (isFilteredOutPlain(rowId)) { dataColumn.skip(); setNullValue(c, rowId); } else { diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/probe/ParquetProbeDecodeState.java b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/probe/ParquetProbeDecodeState.java index dd1d9d7e9857..df4af6388706 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/probe/ParquetProbeDecodeState.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/probe/ParquetProbeDecodeState.java @@ -20,10 +20,14 @@ package org.apache.hadoop.hive.ql.io.parquet.vector.probe; import java.util.List; +import java.util.Set; +import org.apache.commons.lang3.tuple.Pair; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.ql.exec.MapJoinOperator; import org.apache.hadoop.hive.ql.exec.ObjectCache; import org.apache.hadoop.hive.ql.exec.ObjectCacheFactory; +import org.apache.hadoop.hive.ql.exec.OperatorUtils; import org.apache.hadoop.hive.ql.exec.TableScanOperator.ProbeDecodeContext; import org.apache.hadoop.hive.ql.exec.persistence.MapJoinTableContainer; import org.apache.hadoop.hive.ql.exec.vector.mapjoin.hashtable.VectorMapJoinHashTable; @@ -91,35 +95,49 @@ public static ParquetProbeDecodeState of(Configuration conf, MapWork mapWork, if (keyIdx < 0) { // The probe key column isn't in the projected schema for this reader -- e.g. because // column pruning already dropped it. Nothing to probe against; run un-probed. - LOG.debug("ProbeDecode: probe key column {} not in projected columns {}", keyCol, projectedColumns); + LOG.debug("ProbeDecode: probe key column {} not in projected columns {}", + keyCol, projectedColumns); return DISABLED; } String queryId = HiveConf.getVar(conf, HiveConf.ConfVars.HIVE_QUERY_ID); try { ObjectCache cache = ObjectCacheFactory.getCache(conf, queryId, false); - Object cached = cache.retrieve(ctx.getMjSmallTableCacheKey()); + // MapJoinOperator.initializeOp stores the hash table under either the raw MapJoinDesc cache + // key (when conf's cacheKey was null at compile time) or `cacheKey + "_" + concreteOpClass` + // (when Shared Work Optimization / ProbeDecode compilation set the cache key upfront). The + // ProbeDecodeContext only carries the raw key -- resolve the concrete-class suffix by + // matching the MapJoinOperator in the MapWork tree, mirroring the loader's key exactly. + String actualKey = resolveActualCacheKey(mapWork, ctx.getMjSmallTableCacheKey()); + Object cached = cache.retrieve(actualKey); if (cached == null) { - LOG.debug("ProbeDecode: no cached hash table for key {}", ctx.getMjSmallTableCacheKey()); + LOG.debug("ProbeDecode: no cached hash table for key {} (base {})", + actualKey, ctx.getMjSmallTableCacheKey()); + return DISABLED; + } + // MapJoinOperator caches the loaded hash tables as a + // `Pair` -- unwrap and pick the + // small-table container by its position. + MapJoinTableContainer container = unwrapContainer(cached, ctx.getMjSmallTablePos()); + if (container == null) { + LOG.debug("ProbeDecode: could not extract small-table container from cached {} (pos {})", + cached.getClass().getName(), ctx.getMjSmallTablePos()); return DISABLED; } VectorMapJoinHashTable ht; - if (cached instanceof VectorMapJoinTableContainer) { - ht = ((VectorMapJoinTableContainer) cached).vectorMapJoinHashTable(); - } else if (cached instanceof VectorMapJoinHashTable) { - ht = (VectorMapJoinHashTable) cached; - } else if (cached instanceof MapJoinTableContainer) { + if (container instanceof VectorMapJoinTableContainer) { + ht = ((VectorMapJoinTableContainer) container).vectorMapJoinHashTable(); + } else { // Non-vectorized container -- probe-decode has no dictionary-of-keys to intersect with, // so bail out and let the plain decode path run. LOG.debug("ProbeDecode: cached container is non-vectorized ({}); skipping probe", - cached.getClass().getName()); - return DISABLED; - } else { - LOG.debug("ProbeDecode: unexpected cached object type {}", cached.getClass().getName()); + container.getClass().getName()); return DISABLED; } if (ht instanceof VectorMapJoinLongHashTable) { + LOG.info("ProbeDecode: enabled for key column {} (idx {}) via {}", + keyCol, keyIdx, ht.getClass().getSimpleName()); return new ParquetProbeDecodeState(keyIdx, ParquetProbeLongHashTable.of(ht)); } // TODO: bytes-key + multi-key variants when the corresponding ParquetProbeHashTable @@ -132,6 +150,60 @@ public static ParquetProbeDecodeState of(Configuration conf, MapWork mapWork, } } + /** + * Peel a cached hash-table entry stored by {@link MapJoinOperator#loadHashTable} back to the + * small-side container. The op caches a {@code Pair}; some code + * paths (older tests, refactored fixtures) store a bare container or a bare hash table -- keep + * those working too. + */ + private static MapJoinTableContainer unwrapContainer(Object cached, byte smallPos) { + if (cached instanceof Pair) { + Object left = ((Pair) cached).getLeft(); + if (left instanceof MapJoinTableContainer[]) { + MapJoinTableContainer[] tables = (MapJoinTableContainer[]) left; + if (smallPos >= 0 && smallPos < tables.length && tables[smallPos] != null) { + return tables[smallPos]; + } + // Fall back to the first non-null entry -- some plans leave big-table slots null. + for (MapJoinTableContainer t : tables) { + if (t != null) { + return t; + } + } + } + return null; + } + if (cached instanceof MapJoinTableContainer) { + return (MapJoinTableContainer) cached; + } + return null; + } + + /** + * Mirror {@code MapJoinOperator.initializeOp}'s cacheKey computation so our lookup finds the + * hash table the loader stored. When the ProbeDecodeContext's raw cacheKey is non-null, the + * operator appends {@code "_" + this.getClass().getName()} -- and the vectorizer has by now + * substituted the {@link MapJoinOperator} with a concrete {@code VectorMapJoin*Operator} + * subclass, so we read the class off the operator in the MapWork tree. + */ + static String resolveActualCacheKey(MapWork mapWork, String baseCacheKey) { + if (baseCacheKey == null || mapWork == null) { + return baseCacheKey; + } + try { + Set mjs = OperatorUtils.findOperators(mapWork.getWorks(), MapJoinOperator.class); + for (MapJoinOperator mj : mjs) { + if (mj.getConf() != null && baseCacheKey.equals(mj.getConf().getCacheKey())) { + return baseCacheKey + "_" + mj.getClass().getName(); + } + } + } catch (Exception e) { + LOG.debug("ProbeDecode: could not resolve concrete cacheKey suffix, falling back to raw key", + e); + } + return baseCacheKey; + } + public boolean isEnabled() { return probe != null; } diff --git a/ql/src/test/org/apache/hadoop/hive/ql/io/parquet/vector/probe/TestParquetProbeDecodeState.java b/ql/src/test/org/apache/hadoop/hive/ql/io/parquet/vector/probe/TestParquetProbeDecodeState.java new file mode 100644 index 000000000000..17a52dc4f437 --- /dev/null +++ b/ql/src/test/org/apache/hadoop/hive/ql/io/parquet/vector/probe/TestParquetProbeDecodeState.java @@ -0,0 +1,229 @@ +/* + * 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.ql.io.parquet.vector.probe; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.UUID; +import org.apache.commons.lang3.tuple.ImmutablePair; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.llap.io.api.LlapProxy; +import org.apache.hadoop.hive.ql.CompilationOpContext; +import org.apache.hadoop.hive.ql.exec.MapJoinOperator; +import org.apache.hadoop.hive.ql.exec.ObjectCache; +import org.apache.hadoop.hive.ql.exec.ObjectCacheFactory; +import org.apache.hadoop.hive.ql.exec.Operator; +import org.apache.hadoop.hive.ql.exec.TableScanOperator.ProbeDecodeContext; +import org.apache.hadoop.hive.ql.exec.persistence.MapJoinTableContainer; +import org.apache.hadoop.hive.ql.exec.vector.mapjoin.hashtable.VectorMapJoinHashTable; +import org.apache.hadoop.hive.ql.exec.vector.mapjoin.hashtable.VectorMapJoinLongHashSet; +import org.apache.hadoop.hive.ql.exec.vector.mapjoin.hashtable.VectorMapJoinTableContainer; +import org.apache.hadoop.hive.ql.plan.MapJoinDesc; +import org.apache.hadoop.hive.ql.plan.MapWork; +import org.apache.hadoop.hive.ql.plan.OperatorDesc; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +/** + * Regression coverage for the cacheKey-suffix contract between {@link MapJoinOperator} (the writer + * that seeds the {@link ObjectCache}) and {@link ParquetProbeDecodeState#of} (the reader that + * looks it up). The vectorized Parquet reader must reproduce the loader's key exactly, else it + * looks up under the raw base key, finds nothing, and silently disables probe-decode -- the + * pre-fix behaviour of HIVE-30019. + * + *

{@link MapJoinOperator#initializeOp} computes: + * 

{@code cacheKey = conf.getCacheKey() + "_" + this.getClass().getName();}
+ * The {@link ProbeDecodeContext} that Tez compilation writes into {@link MapWork} carries only + * the raw {@code conf.getCacheKey()} half. {@link ParquetProbeDecodeState#resolveActualCacheKey} + * closes that gap by finding the concrete {@link MapJoinOperator} subclass in the + * {@code MapWork} tree at read time and appending its class name. + * + *

These tests pin both halves of the contract: + *

+ */ +public class TestParquetProbeDecodeState { + + private static final String BASE_CACHE_KEY = "HASH_MAP_MAPJOIN_25_container"; + private static final byte SMALL_TABLE_POS = (byte) 1; + private static final String KEY_COL = "key2"; + private static final List PROJECTED_COLS = Arrays.asList("nokey", KEY_COL, "dt"); + + private HiveConf conf; + private String queryId; + private boolean priorIsDaemon; + + @Before + public void before() { + // Route ObjectCacheFactory through the LLAP daemon branch so a single LlapObjectCache instance + // is shared for both our seed and the state's lookup (the MR branch hands out a fresh cache + // on every getCache call, so the seed would never be visible to the state). + priorIsDaemon = LlapProxy.isDaemon(); + LlapProxy.setDaemon(true); + + conf = new HiveConf(); + conf.setVar(HiveConf.ConfVars.HIVE_EXECUTION_ENGINE, "tez"); + conf.setBoolVar(HiveConf.ConfVars.LLAP_OBJECT_CACHE_ENABLED, true); + queryId = "test-probe-decode-" + UUID.randomUUID(); + conf.setVar(HiveConf.ConfVars.HIVE_QUERY_ID, queryId); + } + + @After + public void after() { + ObjectCacheFactory.removeLlapQueryCache(queryId); + LlapProxy.setDaemon(priorIsDaemon); + } + + /** + * Pin the resolver in isolation: given a {@link MapWork} whose {@link MapJoinOperator}'s + * {@code MapJoinDesc.cacheKey} equals {@code base}, {@code resolveActualCacheKey} must return + * {@code base + "_" + }, matching what + * {@link MapJoinOperator#initializeOp} would have stored. + * + *

Uses the plain {@code MapJoinOperator} class here — the vectorizer swaps it for a + * concrete {@code VectorMapJoin*Operator} subclass at runtime, but the resolver only reads + * {@code Class#getName()} off whatever operator is in the tree, so this is representative. + */ + @Test + public void resolveActualCacheKeyAppendsConcreteClass() { + MapWork mapWork = mapWorkWith(mapJoinOperatorWithCacheKey(BASE_CACHE_KEY)); + + String resolved = ParquetProbeDecodeState.resolveActualCacheKey(mapWork, BASE_CACHE_KEY); + + assertEquals(BASE_CACHE_KEY + "_" + MapJoinOperator.class.getName(), resolved); + } + + /** + * End-to-end: seed the ObjectCache under the fully-suffixed key that + * {@link MapJoinOperator#initializeOp} would store under, then call + * {@link ParquetProbeDecodeState#of}. The state must resolve the suffix, find the entry, unwrap + * the {@link ImmutablePair} to the small-side container, and expose an enabled probe with the + * key column at index 1 (the position of {@code KEY_COL} in {@code PROJECTED_COLS}). + */ + @Test + public void ofResolvesSeededHashTableUnderSuffixedCacheKey() throws Exception { + MapWork mapWork = mapWorkWith(mapJoinOperatorWithCacheKey(BASE_CACHE_KEY)); + seedObjectCache(BASE_CACHE_KEY + "_" + MapJoinOperator.class.getName(), + pairWithSmallContainerAtPos(SMALL_TABLE_POS)); + mapWork.setProbeDecodeContext( + new ProbeDecodeContext(BASE_CACHE_KEY, SMALL_TABLE_POS, KEY_COL, 1.0)); + + ParquetProbeDecodeState state = ParquetProbeDecodeState.of(conf, mapWork, PROJECTED_COLS); + + assertTrue("probe-decode should be enabled once the suffixed key resolves", + state.isEnabled()); + assertEquals("probe key column index must match position in projected columns", + PROJECTED_COLS.indexOf(KEY_COL), state.getKeyColumnIndex()); + } + + /** + * Negative case: the {@link ObjectCache} carries an entry only under the raw base key -- no + * suffix. This is the pre-fix layout (and also what happens if a future refactor accidentally + * strips the suffix from {@link MapJoinOperator#initializeOp}). The state must report + * {@code isEnabled() == false} rather than picking up the un-suffixed entry, keeping the + * suffixed-key path the sole lookup mechanism. + */ + @Test + public void ofDisabledWhenOnlyRawKeyPresent() throws Exception { + MapWork mapWork = mapWorkWith(mapJoinOperatorWithCacheKey(BASE_CACHE_KEY)); + seedObjectCache(BASE_CACHE_KEY, pairWithSmallContainerAtPos(SMALL_TABLE_POS)); + mapWork.setProbeDecodeContext( + new ProbeDecodeContext(BASE_CACHE_KEY, SMALL_TABLE_POS, KEY_COL, 1.0)); + + ParquetProbeDecodeState state = ParquetProbeDecodeState.of(conf, mapWork, PROJECTED_COLS); + + assertFalse("state must not resolve when only the raw (unsuffixed) key is cached", + state.isEnabled()); + } + + private static MapJoinOperator mapJoinOperatorWithCacheKey(String cacheKey) { + MapJoinOperator mj = new MapJoinOperator(new CompilationOpContext()); + MapJoinDesc desc = new MapJoinDesc(); + desc.setCacheKey(cacheKey); + mj.setConf(desc); + return mj; + } + + private static MapWork mapWorkWith(MapJoinOperator mj) { + MapWork mapWork = new MapWork(); + LinkedHashMap> aliasToWork = new LinkedHashMap<>(); + // OperatorUtils#findOperators walks child operators from each entry in aliasToWork, and also + // checks the entry itself -- putting the MapJoinOperator directly in the map is enough for + // the resolver, and avoids the full TS → RS → MJ chain a real plan would carry. + aliasToWork.put("test-alias", mj); + mapWork.setAliasToWork(aliasToWork); + return mapWork; + } + + private void seedObjectCache(String key, Object value) throws Exception { + ObjectCache cache = ObjectCacheFactory.getCache(conf, queryId, false); + // ObjectCache exposes no public put(); retrieve(key, Callable) stores the callable's result + // when the key is absent, which is what the loader effectively does too. + cache.retrieve(key, () -> value); + } + + private static ImmutablePair pairWithSmallContainerAtPos( + byte smallPos) { + // The pre-fix bug was that the cached object is a Pair whose left slot is a + // MapJoinTableContainer[] indexed by the mapjoin position -- mirror that exact shape so the + // state's unwrap logic is exercised, not shortcut. + MapJoinTableContainer[] tables = new MapJoinTableContainer[Math.max(2, smallPos + 1)]; + tables[smallPos] = smallSideVectorContainer(); + return ImmutablePair.of(tables, /* serdes, unused by the state */ null); + } + + private static VectorMapJoinTableContainer smallSideVectorContainer() { + VectorMapJoinLongHashSet ht = mock(VectorMapJoinLongHashSet.class); + // ParquetProbeLongHashTable's constructor reads useMinMax / min / max and calls + // createHashSetResult on the SET branch -- give it just enough to run without NPE. + when(ht.useMinMax()).thenReturn(false); + when(ht.createHashSetResult()).thenReturn(new StubHashSetResult()); + VectorMapJoinTableContainer c = mock(VectorMapJoinTableContainer.class); + when(c.vectorMapJoinHashTable()).thenReturn((VectorMapJoinHashTable) ht); + return c; + } + + /** + * Minimal concrete {@code VectorMapJoinHashSetResult}. The parent classes are abstract but have + * no unimplemented methods, so a bare subclass instantiates cleanly. + */ + private static final class StubHashSetResult + extends org.apache.hadoop.hive.ql.exec.vector.mapjoin.hashtable.VectorMapJoinHashSetResult { + } +} diff --git a/ql/src/test/org/apache/hadoop/hive/ql/io/parquet/vector/probe/TestParquetProbeFilter.java b/ql/src/test/org/apache/hadoop/hive/ql/io/parquet/vector/probe/TestParquetProbeFilter.java new file mode 100644 index 000000000000..790bb97ef7bc --- /dev/null +++ b/ql/src/test/org/apache/hadoop/hive/ql/io/parquet/vector/probe/TestParquetProbeFilter.java @@ -0,0 +1,143 @@ +/* + * 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.ql.io.parquet.vector.probe; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import org.junit.Test; + +/** + * Pure unit tests for the {@link ParquetProbeFilter} lifecycle: the ProbeDecode path calls + * {@link ParquetProbeFilter#newBitmap} after probing, per-row column readers query + * {@link ParquetProbeFilter#isSelected}, and once all columns are read + * {@link ParquetProbeFilter#compact} materialises the {@code selected[]} that + * {@code VectorizedRowBatch} consumes. + */ +public class TestParquetProbeFilter { + + @Test + public void newBitmapRejectsNull() { + try { + ParquetProbeFilter.newBitmap(null); + fail("expected IAE for null bitmap"); + } catch (IllegalArgumentException expected) { + // ok + } + } + + @Test + public void isSelectedReflectsBitmap() { + boolean[] bits = { true, false, true, true, false }; + ParquetProbeFilter f = ParquetProbeFilter.newBitmap(bits); + + assertTrue(f.isSelected(0)); + assertFalse(f.isSelected(1)); + assertTrue(f.isSelected(2)); + assertTrue(f.isSelected(3)); + assertFalse(f.isSelected(4)); + } + + @Test + public void isSelectedOutOfBoundsReturnsFalse() { + // Guards against off-by-one bugs in callers that pass a rowId near the batch tail. + ParquetProbeFilter f = ParquetProbeFilter.newBitmap(new boolean[] { true, true }); + assertFalse(f.isSelected(-1)); + assertFalse(f.isSelected(2)); + assertFalse(f.isSelected(Integer.MAX_VALUE)); + } + + @Test + public void compactBeforeCallReturnsNullArrays() { + // Explicit contract: getSelected() / getSelectedSize() are meaningless until compact() runs. + ParquetProbeFilter f = ParquetProbeFilter.newBitmap(new boolean[] { true, false, true }); + assertNull(f.getSelected()); + assertEquals(0, f.getSelectedSize()); + } + + @Test + public void compactMaterialisesSelected() { + boolean[] bits = { true, false, true, true, false, true }; + ParquetProbeFilter f = ParquetProbeFilter.newBitmap(bits); + + ParquetProbeFilter same = f.compact(bits.length); + assertSame("compact must return this for chaining", f, same); + + assertArrayEquals(new int[] { 0, 2, 3, 5 }, java.util.Arrays.copyOf(f.getSelected(), f.getSelectedSize())); + assertEquals(4, f.getSelectedSize()); + } + + @Test + public void compactHonoursBatchSize() { + // batchSize can be smaller than the bitmap when the last row group ends mid-batch; + // entries beyond batchSize must be ignored. + boolean[] bits = { true, true, true, true }; + ParquetProbeFilter f = ParquetProbeFilter.newBitmap(bits); + f.compact(2); + + assertEquals(2, f.getSelectedSize()); + assertArrayEquals(new int[] { 0, 1 }, java.util.Arrays.copyOf(f.getSelected(), f.getSelectedSize())); + } + + @Test + public void compactHandlesEmptyBatch() { + ParquetProbeFilter f = ParquetProbeFilter.newBitmap(new boolean[0]); + f.compact(0); + assertEquals(0, f.getSelectedSize()); + assertArrayEquals(new int[0], f.getSelected()); + } + + @Test + public void compactHandlesAllRejected() { + ParquetProbeFilter f = ParquetProbeFilter.newBitmap(new boolean[] { false, false, false }); + f.compact(3); + assertEquals(0, f.getSelectedSize()); + // getSelected() is allowed to return an over-provisioned array; only [0, selectedSize) counts. + assertEquals(3, f.getSelected().length); + } + + @Test + public void compactHandlesAllAccepted() { + ParquetProbeFilter f = ParquetProbeFilter.newBitmap(new boolean[] { true, true, true, true }); + f.compact(4); + assertEquals(4, f.getSelectedSize()); + assertArrayEquals(new int[] { 0, 1, 2, 3 }, f.getSelected()); + } + + @Test + public void compactIsIdempotent() { + // VectorizedParquetRecordReader.applyProbeFilterToBatch may call compact() more than once + // if the same filter is passed to several helpers; a second call must be a no-op. + boolean[] bits = { true, false, true }; + ParquetProbeFilter f = ParquetProbeFilter.newBitmap(bits); + f.compact(bits.length); + int[] firstArray = f.getSelected(); + int firstSize = f.getSelectedSize(); + + f.compact(bits.length); + assertSame("compact must not reallocate on repeat", firstArray, f.getSelected()); + assertEquals(firstSize, f.getSelectedSize()); + } +} diff --git a/ql/src/test/org/apache/hadoop/hive/ql/io/parquet/vector/probe/TestVectorizedParquetProbeDecodeReader.java b/ql/src/test/org/apache/hadoop/hive/ql/io/parquet/vector/probe/TestVectorizedParquetProbeDecodeReader.java new file mode 100644 index 000000000000..3e927633909c --- /dev/null +++ b/ql/src/test/org/apache/hadoop/hive/ql/io/parquet/vector/probe/TestVectorizedParquetProbeDecodeReader.java @@ -0,0 +1,329 @@ +/* + * 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.ql.io.parquet.vector.probe; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +import java.io.IOException; +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.nio.charset.StandardCharsets; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.ql.exec.vector.BytesColumnVector; +import org.apache.hadoop.hive.ql.exec.vector.ColumnVector; +import org.apache.hadoop.hive.ql.exec.vector.DoubleColumnVector; +import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector; +import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch; +import org.apache.hadoop.hive.ql.io.parquet.VectorizedColumnReaderTestBase; +import org.apache.hadoop.hive.ql.io.parquet.vector.VectorizedColumnReader; +import org.apache.hadoop.hive.ql.io.parquet.vector.VectorizedParquetRecordReader; +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.mapred.JobConf; +import org.apache.parquet.example.data.Group; +import org.apache.parquet.example.data.simple.SimpleGroupFactory; +import org.apache.parquet.hadoop.ParquetWriter; +import org.apache.parquet.hadoop.example.GroupWriteSupport; +import org.apache.parquet.hadoop.metadata.CompressionCodecName; +import org.apache.parquet.io.api.Binary; +import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.MessageTypeParser; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +/** + * End-to-end test that the ProbeDecode filter path in {@code VectorizedPrimitiveColumnReader} + * (a) leaves filtered rows as null-marked slots in the column vector and (b) still decodes + * surviving rows correctly. + * + *

The test bypasses {@code VectorizedParquetRecordReader.nextBatch}'s ProbeDecodeState + * resolution (which requires a live MapJoin operator + hash table) and instead reflects into + * the reader to reach the primitive column readers directly. It then calls the filter-aware + * {@code readBatch(total, column, type, ParquetProbeFilter)} signature that + * {@code VectorizedPrimitiveColumnReader} exposes to the outer reader -- the same call + * {@code nextBatch} would make on the probe path. + * + *

Both dictionary-encoded and PLAIN-encoded pages are exercised so both the coalesced-skip + * path ({@code readDictionaryIDs} + {@code pendingSkip} + {@code skipInts}) and the per-row + * skip path ({@code readIntegers}/{@code readLongs}/{@code readDoubles}/{@code readBinaries}) + * are hit. + */ +public class TestVectorizedParquetProbeDecodeReader extends VectorizedColumnReaderTestBase { + + private static final int N_ROWS = 256; + private static final MessageType WRITE_SCHEMA = MessageTypeParser.parseMessageType( + "message test { " + + "required int32 int_col; " + + "required int64 long_col; " + + "required double dbl_col; " + + "required binary str_col (UTF8); " + + "}"); + + private java.io.File tempDir; + private Path tempFile; + + @Before + public void setUpFile() throws Exception { + tempDir = java.nio.file.Files.createTempDirectory("probe-decode-").toFile(); + tempFile = new Path(new java.io.File(tempDir, "data.parquet").toURI()); + } + + @After + public void tearDownFile() { + if (tempDir != null && tempDir.exists()) { + for (java.io.File f : tempDir.listFiles()) { + f.delete(); + } + tempDir.delete(); + } + } + + /** Filter accepts rows whose index is even (0, 2, 4, ...). */ + private static ParquetProbeFilter halfFilter(int size) { + boolean[] bits = new boolean[size]; + for (int i = 0; i < size; i++) { + bits[i] = (i % 2 == 0); + } + return ParquetProbeFilter.newBitmap(bits); + } + + /** + * When {@code dictionary=true}, values cycle through a small distinct set so Parquet + * dictionary-encodes each page; otherwise every value is unique so the writer falls back to + * PLAIN encoding. + */ + private void writeFile(boolean dictionary) throws IOException { + Configuration conf = new Configuration(); + GroupWriteSupport.setSchema(WRITE_SCHEMA, conf); + SimpleGroupFactory gf = new SimpleGroupFactory(WRITE_SCHEMA); + + try (ParquetWriter writer = new ParquetWriter<>(tempFile, new GroupWriteSupport(), + CompressionCodecName.UNCOMPRESSED, 1024 * 1024, 1024 * 1024, 512, dictionary, false, + ParquetWriter.DEFAULT_WRITER_VERSION, conf)) { + for (int i = 0; i < N_ROWS; i++) { + int intVal = dictionary ? (i % 4) : i; + long longVal = dictionary ? (i % 4) : (long) i; + double dblVal = dictionary ? (i % 4) : (double) i; + String strVal = dictionary ? ("v" + (i % 4)) : ("v" + i); + Group g = gf.newGroup() + .append("int_col", intVal) + .append("long_col", longVal) + .append("dbl_col", dblVal) + .append("str_col", Binary.fromString(strVal)); + writer.write(g); + } + } + } + + private VectorizedParquetRecordReader openReader() throws Exception { + Configuration readerConf = new Configuration(); + readerConf.set(org.apache.hadoop.hive.ql.io.IOConstants.COLUMNS, + "int_col,long_col,dbl_col,str_col"); + readerConf.set(org.apache.hadoop.hive.ql.io.IOConstants.COLUMNS_TYPES, + "int,bigint,double,string"); + readerConf.setBoolean(ColumnProjectionUtils.READ_ALL_COLUMNS, false); + readerConf.set(ColumnProjectionUtils.READ_COLUMN_IDS_CONF_STR, "0,1,2,3"); + HiveConf.setBoolVar(readerConf, HiveConf.ConfVars.HIVE_VECTORIZATION_ENABLED, true); + HiveConf.setVar(readerConf, HiveConf.ConfVars.PLAN, "//tmp"); + org.apache.hadoop.mapreduce.Job vectorJob = new org.apache.hadoop.mapreduce.Job(readerConf, "read"); + org.apache.parquet.hadoop.ParquetInputFormat.setInputPaths(vectorJob, tempFile); + initialVectorizedRowBatchCtx(readerConf, null); + return new VectorizedParquetRecordReader(getFileSplit(vectorJob, tempFile), new JobConf(readerConf)); + } + + /** + * Force {@code checkEndOfRowGroup} to run on the reader without consuming any rows through + * {@code nextBatch}. This leaves {@code columnReaders} populated and page state at row 0. + */ + @SuppressWarnings("unchecked") + private static VectorizedColumnReader[] primeColumnReaders(VectorizedParquetRecordReader r) + throws ReflectiveOperationException { + Method m = VectorizedParquetRecordReader.class.getDeclaredMethod("checkEndOfRowGroup"); + m.setAccessible(true); + m.invoke(r); + Field f = VectorizedParquetRecordReader.class.getDeclaredField("columnReaders"); + f.setAccessible(true); + return (VectorizedColumnReader[]) f.get(r); + } + + /** + * Expected value at row {@code i} for the given column and encoding, matching what + * {@link #writeFile} wrote. + */ + private static long expectedInt(int i, boolean dict) { return dict ? (i % 4) : i; } + private static long expectedLong(int i, boolean dict) { return dict ? (i % 4) : (long) i; } + private static double expectedDbl(int i, boolean dict) { return dict ? (i % 4) : (double) i; } + private static String expectedStr(int i, boolean dict) { return dict ? ("v" + (i % 4)) : ("v" + i); } + + private void runFilterHonoringTest(boolean dictionary) throws Exception { + writeFile(dictionary); + + VectorizedParquetRecordReader reader = openReader(); + try { + VectorizedColumnReader[] readers = primeColumnReaders(reader); + assertNotNull("columnReaders must be populated after checkEndOfRowGroup", readers); + assertEquals(4, readers.length); + + int batchSize = Math.min(VectorizedRowBatch.DEFAULT_SIZE, N_ROWS); + ParquetProbeFilter filter = halfFilter(batchSize); + + LongColumnVector intVec = new LongColumnVector(batchSize); + LongColumnVector longVec = new LongColumnVector(batchSize); + DoubleColumnVector dblVec = new DoubleColumnVector(batchSize); + BytesColumnVector strVec = new BytesColumnVector(batchSize); + for (ColumnVector v : new ColumnVector[] { intVec, longVec, dblVec, strVec }) { + v.init(); + } + // Vectors start with noNulls=true; every column reader clears that when it hits a + // filtered slot (see setNullValue in VectorizedPrimitiveColumnReader). + + // Types must line up with the columns projection. + TypeInfo intType = TypeInfoFactory.getPrimitiveTypeInfo("int"); + TypeInfo longType = TypeInfoFactory.getPrimitiveTypeInfo("bigint"); + TypeInfo dblType = TypeInfoFactory.getPrimitiveTypeInfo("double"); + TypeInfo strType = TypeInfoFactory.getPrimitiveTypeInfo("string"); + + // Drive the filter-aware read path on each column reader. This is the same call that + // VectorizedParquetRecordReader.nextBatch issues on non-key columns when the ProbeDecode + // path is active. + readers[0].readBatch(batchSize, intVec, intType, filter); + readers[1].readBatch(batchSize, longVec, longType, filter); + readers[2].readBatch(batchSize, dblVec, dblType, filter); + readers[3].readBatch(batchSize, strVec, strType, filter); + + // Each vector must now carry the surviving rows' values in the accepted slots and a + // null-marked slot everywhere else. noNulls must be false since we produced nulls. + for (int i = 0; i < batchSize; i++) { + boolean selected = (i % 2 == 0); + if (selected) { + assertFalse("row " + i + " must not be null in int_col", intVec.isNull[i]); + assertFalse("row " + i + " must not be null in long_col", longVec.isNull[i]); + assertFalse("row " + i + " must not be null in dbl_col", dblVec.isNull[i]); + assertFalse("row " + i + " must not be null in str_col", strVec.isNull[i]); + + assertEquals("int_col value at row " + i, expectedInt(i, dictionary), intVec.vector[i]); + assertEquals("long_col value at row " + i, expectedLong(i, dictionary), longVec.vector[i]); + assertEquals("dbl_col value at row " + i, expectedDbl(i, dictionary), dblVec.vector[i], 0.0); + String actual = new String(strVec.vector[i], strVec.start[i], strVec.length[i], + StandardCharsets.UTF_8); + assertEquals("str_col value at row " + i, expectedStr(i, dictionary), actual); + } else { + assertTrue("row " + i + " must be null in int_col", intVec.isNull[i]); + assertTrue("row " + i + " must be null in long_col", longVec.isNull[i]); + assertTrue("row " + i + " must be null in dbl_col", dblVec.isNull[i]); + assertTrue("row " + i + " must be null in str_col", strVec.isNull[i]); + } + } + assertFalse("noNulls must be cleared once a filtered row is emitted (int_col)", + intVec.noNulls); + assertFalse("noNulls must be cleared once a filtered row is emitted (long_col)", + longVec.noNulls); + assertFalse("noNulls must be cleared once a filtered row is emitted (dbl_col)", + dblVec.noNulls); + assertFalse("noNulls must be cleared once a filtered row is emitted (str_col)", + strVec.noNulls); + } finally { + reader.close(); + } + } + + /** + * Filter should be a no-op when it accepts every row: the resulting vectors are identical to + * a baseline unfiltered read. This is the "no regression" case for a filter that matches all. + */ + private void runAllPassNoOpTest(boolean dictionary) throws Exception { + writeFile(dictionary); + + VectorizedParquetRecordReader baselineReader = openReader(); + VectorizedParquetRecordReader filteredReader = openReader(); + try { + VectorizedColumnReader[] baseline = primeColumnReaders(baselineReader); + VectorizedColumnReader[] filtered = primeColumnReaders(filteredReader); + int batchSize = Math.min(VectorizedRowBatch.DEFAULT_SIZE, N_ROWS); + + boolean[] allTrue = new boolean[batchSize]; + java.util.Arrays.fill(allTrue, true); + ParquetProbeFilter allPass = ParquetProbeFilter.newBitmap(allTrue); + + TypeInfo intType = TypeInfoFactory.getPrimitiveTypeInfo("int"); + TypeInfo longType = TypeInfoFactory.getPrimitiveTypeInfo("bigint"); + TypeInfo dblType = TypeInfoFactory.getPrimitiveTypeInfo("double"); + TypeInfo strType = TypeInfoFactory.getPrimitiveTypeInfo("string"); + + LongColumnVector baseInt = new LongColumnVector(batchSize); + baseInt.init(); + LongColumnVector filtInt = new LongColumnVector(batchSize); + filtInt.init(); + baseline[0].readBatch(batchSize, baseInt, intType); + filtered[0].readBatch(batchSize, filtInt, intType, allPass); + for (int i = 0; i < batchSize; i++) { + assertEquals("int_col allPass row " + i, baseInt.vector[i], filtInt.vector[i]); + assertEquals("int_col allPass isNull row " + i, baseInt.isNull[i], filtInt.isNull[i]); + } + + BytesColumnVector baseStr = new BytesColumnVector(batchSize); + baseStr.init(); + BytesColumnVector filtStr = new BytesColumnVector(batchSize); + filtStr.init(); + baseline[3].readBatch(batchSize, baseStr, strType); + filtered[3].readBatch(batchSize, filtStr, strType, allPass); + for (int i = 0; i < batchSize; i++) { + assertEquals("str_col allPass isNull row " + i, baseStr.isNull[i], filtStr.isNull[i]); + String b = new String(baseStr.vector[i], baseStr.start[i], baseStr.length[i], StandardCharsets.UTF_8); + String f = new String(filtStr.vector[i], filtStr.start[i], filtStr.length[i], StandardCharsets.UTF_8); + assertEquals("str_col allPass row " + i, b, f); + } + } finally { + baselineReader.close(); + filteredReader.close(); + } + } + + @Test + public void filterHonoredOnDictionaryEncodedPages() throws Exception { + // Exercises readDictionaryIDs -> pendingSkip -> DictionaryValuesReader.skip(int) + // -> RunLengthBitPackingHybridDecoder.skipInts (the O(runs) fast-path). + runFilterHonoringTest(true); + } + + @Test + public void filterHonoredOnPlainEncodedPages() throws Exception { + // Exercises the per-row skip path in readIntegers / readLongs / readDoubles / readBinaries. + runFilterHonoringTest(false); + } + + @Test + public void allPassFilterIsNoOpDict() throws Exception { + runAllPassNoOpTest(true); + } + + @Test + public void allPassFilterIsNoOpPlain() throws Exception { + runAllPassNoOpTest(false); + } +} diff --git a/ql/src/test/queries/clientpositive/probedecode_mapjoin_simple_parquet.q b/ql/src/test/queries/clientpositive/probedecode_mapjoin_simple_parquet.q new file mode 100644 index 000000000000..a709f3bfb5e5 --- /dev/null +++ b/ql/src/test/queries/clientpositive/probedecode_mapjoin_simple_parquet.q @@ -0,0 +1,53 @@ +-- Parquet mirror of probedecode_mapjoin_simple.q. +-- +-- Exercises the Parquet ProbeDecode path: the vectorizer wires the big-side +-- TableScan with a ProbeDecodeContext, VectorizedParquetRecordReader decodes +-- the join-key column first, probes the small-side hash table, and passes the +-- resulting ParquetProbeFilter down to the remaining columns' readBatch calls +-- so filtered rows skip decode / conversion. +-- +-- The correctness bar for this test is that the join result must be identical +-- with and without hive.optimize.scan.probedecode enabled; the fast-path is +-- purely a performance optimisation (rows that would be filtered anyway are +-- read as nulls into the batch, then dropped via batch.selected[]). +set hive.stats.column.autogather=false; +set hive.mapred.mode=nonstrict; +set hive.explain.user=false; +SET hive.auto.convert.join=true; +SET hive.auto.convert.join.noconditionaltask=true; +SET hive.auto.convert.join.noconditionaltask.size=1000000000; +SET hive.vectorized.execution.enabled=true; +set hive.vectorized.execution.mapjoin.native.fast.hashtable.enabled=true; +set hive.fetch.task.conversion=none; +SET mapred.min.split.size=1000; +SET mapred.max.split.size=5000; + +CREATE TABLE item_dim_pq (key1 int, name string) stored as parquet; +CREATE TABLE orders_fact_pq (nokey int, key2 int, dt timestamp) stored as parquet; + +INSERT INTO item_dim_pq values(101, "Item 101"); +INSERT INTO item_dim_pq values(102, "Item 102"); + +INSERT INTO orders_fact_pq values(12345, 101, '2001-01-30 00:00:00'); +INSERT INTO orders_fact_pq values(23456, 104, '2002-02-30 00:00:00'); +INSERT INTO orders_fact_pq values(34567, 108, '2003-03-30 00:00:00'); +INSERT INTO orders_fact_pq values(45678, 102, '2004-04-30 00:00:00'); +INSERT INTO orders_fact_pq values(56789, 109, '2005-05-30 00:00:00'); +INSERT INTO orders_fact_pq values(67891, 110, '2006-06-30 00:00:00'); + +-- Baseline: probedecode disabled. Result set is the reference for equivalence. +SET hive.optimize.scan.probedecode=false; + +select key1, key2, name, dt from orders_fact_pq join item_dim_pq on (orders_fact_pq.key2 = item_dim_pq.key1) +order by key2; + +-- Now enable probedecode. The plan should carry the ProbeDecodeContext on the +-- big-side (orders_fact_pq) TableScan; the join result must be identical. +SET hive.optimize.scan.probedecode=true; + +EXPLAIN VECTORIZATION DETAIL +select key1, key2, name, dt from orders_fact_pq join item_dim_pq on (orders_fact_pq.key2 = item_dim_pq.key1); + +-- Two keys match (101, 102); the other four rows are dropped by the probe filter. +select key1, key2, name, dt from orders_fact_pq join item_dim_pq on (orders_fact_pq.key2 = item_dim_pq.key1) +order by key2; diff --git a/ql/src/test/results/clientpositive/llap/probedecode_mapjoin_simple_parquet.q.out b/ql/src/test/results/clientpositive/llap/probedecode_mapjoin_simple_parquet.q.out new file mode 100644 index 000000000000..7784044e5469 --- /dev/null +++ b/ql/src/test/results/clientpositive/llap/probedecode_mapjoin_simple_parquet.q.out @@ -0,0 +1,300 @@ +PREHOOK: query: CREATE TABLE item_dim_pq (key1 int, name string) stored as parquet +PREHOOK: type: CREATETABLE +PREHOOK: Output: database:default +PREHOOK: Output: default@item_dim_pq +POSTHOOK: query: CREATE TABLE item_dim_pq (key1 int, name string) stored as parquet +POSTHOOK: type: CREATETABLE +POSTHOOK: Output: database:default +POSTHOOK: Output: default@item_dim_pq +PREHOOK: query: CREATE TABLE orders_fact_pq (nokey int, key2 int, dt timestamp) stored as parquet +PREHOOK: type: CREATETABLE +PREHOOK: Output: database:default +PREHOOK: Output: default@orders_fact_pq +POSTHOOK: query: CREATE TABLE orders_fact_pq (nokey int, key2 int, dt timestamp) stored as parquet +POSTHOOK: type: CREATETABLE +POSTHOOK: Output: database:default +POSTHOOK: Output: default@orders_fact_pq +PREHOOK: query: INSERT INTO item_dim_pq values(101, "Item 101") +PREHOOK: type: QUERY +PREHOOK: Input: _dummy_database@_dummy_table +PREHOOK: Output: default@item_dim_pq +POSTHOOK: query: INSERT INTO item_dim_pq values(101, "Item 101") +POSTHOOK: type: QUERY +POSTHOOK: Input: _dummy_database@_dummy_table +POSTHOOK: Output: default@item_dim_pq +POSTHOOK: Lineage: item_dim_pq.key1 SCRIPT [] +POSTHOOK: Lineage: item_dim_pq.name SCRIPT [] +PREHOOK: query: INSERT INTO item_dim_pq values(102, "Item 102") +PREHOOK: type: QUERY +PREHOOK: Input: _dummy_database@_dummy_table +PREHOOK: Output: default@item_dim_pq +POSTHOOK: query: INSERT INTO item_dim_pq values(102, "Item 102") +POSTHOOK: type: QUERY +POSTHOOK: Input: _dummy_database@_dummy_table +POSTHOOK: Output: default@item_dim_pq +POSTHOOK: Lineage: item_dim_pq.key1 SCRIPT [] +POSTHOOK: Lineage: item_dim_pq.name SCRIPT [] +PREHOOK: query: INSERT INTO orders_fact_pq values(12345, 101, '2001-01-30 00:00:00') +PREHOOK: type: QUERY +PREHOOK: Input: _dummy_database@_dummy_table +PREHOOK: Output: default@orders_fact_pq +POSTHOOK: query: INSERT INTO orders_fact_pq values(12345, 101, '2001-01-30 00:00:00') +POSTHOOK: type: QUERY +POSTHOOK: Input: _dummy_database@_dummy_table +POSTHOOK: Output: default@orders_fact_pq +POSTHOOK: Lineage: orders_fact_pq.dt SCRIPT [] +POSTHOOK: Lineage: orders_fact_pq.key2 SCRIPT [] +POSTHOOK: Lineage: orders_fact_pq.nokey SCRIPT [] +PREHOOK: query: INSERT INTO orders_fact_pq values(23456, 104, '2002-02-30 00:00:00') +PREHOOK: type: QUERY +PREHOOK: Input: _dummy_database@_dummy_table +PREHOOK: Output: default@orders_fact_pq +POSTHOOK: query: INSERT INTO orders_fact_pq values(23456, 104, '2002-02-30 00:00:00') +POSTHOOK: type: QUERY +POSTHOOK: Input: _dummy_database@_dummy_table +POSTHOOK: Output: default@orders_fact_pq +POSTHOOK: Lineage: orders_fact_pq.dt SCRIPT [] +POSTHOOK: Lineage: orders_fact_pq.key2 SCRIPT [] +POSTHOOK: Lineage: orders_fact_pq.nokey SCRIPT [] +PREHOOK: query: INSERT INTO orders_fact_pq values(34567, 108, '2003-03-30 00:00:00') +PREHOOK: type: QUERY +PREHOOK: Input: _dummy_database@_dummy_table +PREHOOK: Output: default@orders_fact_pq +POSTHOOK: query: INSERT INTO orders_fact_pq values(34567, 108, '2003-03-30 00:00:00') +POSTHOOK: type: QUERY +POSTHOOK: Input: _dummy_database@_dummy_table +POSTHOOK: Output: default@orders_fact_pq +POSTHOOK: Lineage: orders_fact_pq.dt SCRIPT [] +POSTHOOK: Lineage: orders_fact_pq.key2 SCRIPT [] +POSTHOOK: Lineage: orders_fact_pq.nokey SCRIPT [] +PREHOOK: query: INSERT INTO orders_fact_pq values(45678, 102, '2004-04-30 00:00:00') +PREHOOK: type: QUERY +PREHOOK: Input: _dummy_database@_dummy_table +PREHOOK: Output: default@orders_fact_pq +POSTHOOK: query: INSERT INTO orders_fact_pq values(45678, 102, '2004-04-30 00:00:00') +POSTHOOK: type: QUERY +POSTHOOK: Input: _dummy_database@_dummy_table +POSTHOOK: Output: default@orders_fact_pq +POSTHOOK: Lineage: orders_fact_pq.dt SCRIPT [] +POSTHOOK: Lineage: orders_fact_pq.key2 SCRIPT [] +POSTHOOK: Lineage: orders_fact_pq.nokey SCRIPT [] +PREHOOK: query: INSERT INTO orders_fact_pq values(56789, 109, '2005-05-30 00:00:00') +PREHOOK: type: QUERY +PREHOOK: Input: _dummy_database@_dummy_table +PREHOOK: Output: default@orders_fact_pq +POSTHOOK: query: INSERT INTO orders_fact_pq values(56789, 109, '2005-05-30 00:00:00') +POSTHOOK: type: QUERY +POSTHOOK: Input: _dummy_database@_dummy_table +POSTHOOK: Output: default@orders_fact_pq +POSTHOOK: Lineage: orders_fact_pq.dt SCRIPT [] +POSTHOOK: Lineage: orders_fact_pq.key2 SCRIPT [] +POSTHOOK: Lineage: orders_fact_pq.nokey SCRIPT [] +PREHOOK: query: INSERT INTO orders_fact_pq values(67891, 110, '2006-06-30 00:00:00') +PREHOOK: type: QUERY +PREHOOK: Input: _dummy_database@_dummy_table +PREHOOK: Output: default@orders_fact_pq +POSTHOOK: query: INSERT INTO orders_fact_pq values(67891, 110, '2006-06-30 00:00:00') +POSTHOOK: type: QUERY +POSTHOOK: Input: _dummy_database@_dummy_table +POSTHOOK: Output: default@orders_fact_pq +POSTHOOK: Lineage: orders_fact_pq.dt SCRIPT [] +POSTHOOK: Lineage: orders_fact_pq.key2 SCRIPT [] +POSTHOOK: Lineage: orders_fact_pq.nokey SCRIPT [] +PREHOOK: query: select key1, key2, name, dt from orders_fact_pq join item_dim_pq on (orders_fact_pq.key2 = item_dim_pq.key1) +order by key2 +PREHOOK: type: QUERY +PREHOOK: Input: default@item_dim_pq +PREHOOK: Input: default@orders_fact_pq +#### A masked pattern was here #### +POSTHOOK: query: select key1, key2, name, dt from orders_fact_pq join item_dim_pq on (orders_fact_pq.key2 = item_dim_pq.key1) +order by key2 +POSTHOOK: type: QUERY +POSTHOOK: Input: default@item_dim_pq +POSTHOOK: Input: default@orders_fact_pq +#### A masked pattern was here #### +101 101 Item 101 2001-01-30 00:00:00 +102 102 Item 102 2004-04-30 00:00:00 +PREHOOK: query: EXPLAIN VECTORIZATION DETAIL +select key1, key2, name, dt from orders_fact_pq join item_dim_pq on (orders_fact_pq.key2 = item_dim_pq.key1) +PREHOOK: type: QUERY +PREHOOK: Input: default@item_dim_pq +PREHOOK: Input: default@orders_fact_pq +#### A masked pattern was here #### +POSTHOOK: query: EXPLAIN VECTORIZATION DETAIL +select key1, key2, name, dt from orders_fact_pq join item_dim_pq on (orders_fact_pq.key2 = item_dim_pq.key1) +POSTHOOK: type: QUERY +POSTHOOK: Input: default@item_dim_pq +POSTHOOK: Input: default@orders_fact_pq +#### A masked pattern was here #### +PLAN VECTORIZATION: + enabled: true + enabledConditionsMet: [hive.vectorized.execution.enabled IS true] + +STAGE DEPENDENCIES: + Stage-1 is a root stage + Stage-0 depends on stages: Stage-1 + +STAGE PLANS: + Stage: Stage-1 + Tez +#### A masked pattern was here #### + Edges: + Map 1 <- Map 2 (BROADCAST_EDGE) +#### A masked pattern was here #### + Vertices: + Map 1 + Map Operator Tree: + TableScan + alias: orders_fact_pq + filterExpr: key2 is not null (type: boolean) + probeDecodeDetails: cacheKey:HASH_MAP_MAPJOIN_25_container, bigKeyColName:key2, smallTablePos:1, keyRatio:1.0 + Statistics: Num rows: 6 Data size: 264 Basic stats: COMPLETE Column stats: NONE + TableScan Vectorization: + native: true + vectorizationSchemaColumns: [0:nokey:int, 1:key2:int, 2:dt:timestamp, 3:ROW__ID:struct, 4:ROW__IS__DELETED:boolean] + Filter Operator + Filter Vectorization: + className: VectorFilterOperator + native: true + predicateExpression: SelectColumnIsNotNull(col 1:int) + predicate: key2 is not null (type: boolean) + Statistics: Num rows: 6 Data size: 264 Basic stats: COMPLETE Column stats: NONE + Select Operator + expressions: key2 (type: int), dt (type: timestamp) + outputColumnNames: _col0, _col1 + Select Vectorization: + className: VectorSelectOperator + native: true + projectedOutputColumnNums: [1, 2] + Statistics: Num rows: 6 Data size: 264 Basic stats: COMPLETE Column stats: NONE + Map Join Operator + condition map: + Inner Join 0 to 1 + keys: + 0 _col0 (type: int) + 1 _col0 (type: int) + Map Join Vectorization: + bigTableKeyColumns: 1:int + bigTableRetainColumnNums: [1, 2] + bigTableValueColumns: 1:int, 2:timestamp + className: VectorMapJoinInnerLongOperator + native: true + nativeConditionsMet: hive.mapjoin.optimized.hashtable IS true, hive.vectorized.execution.mapjoin.native.enabled IS true, hive.execution.engine tez IN [tez] IS true, One MapJoin Condition IS true, No nullsafe IS true, Small table vectorizes IS true, Fast Hash Table and No Hybrid Hash Join IS true + nonOuterSmallTableKeyMapping: [] + projectedOutput: 1:int, 2:timestamp, 1:int, 5:string + smallTableValueMapping: 5:string + hashTableImplementationType: FAST + outputColumnNames: _col0, _col1, _col2, _col3 + input vertices: + 1 Map 2 + Statistics: Num rows: 6 Data size: 290 Basic stats: COMPLETE Column stats: NONE + Select Operator + expressions: _col2 (type: int), _col0 (type: int), _col3 (type: string), _col1 (type: timestamp) + outputColumnNames: _col0, _col1, _col2, _col3 + Select Vectorization: + className: VectorSelectOperator + native: true + projectedOutputColumnNums: [1, 1, 5, 2] + Statistics: Num rows: 6 Data size: 290 Basic stats: COMPLETE Column stats: NONE + File Output Operator + compressed: false + File Sink Vectorization: + className: VectorFileSinkOperator + native: false + Statistics: Num rows: 6 Data size: 290 Basic stats: COMPLETE Column stats: NONE + table: + input format: org.apache.hadoop.mapred.SequenceFileInputFormat + output format: org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat + serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe + Execution mode: vectorized, llap + LLAP IO: all inputs (cache only) + Map Vectorization: + enabled: true + enabledConditionsMet: hive.vectorized.use.vectorized.input.format IS true + inputFormatFeatureSupport: [DECIMAL_64] + featureSupportInUse: [DECIMAL_64] + inputFileFormats: org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat + allNative: false + usesVectorUDFAdaptor: false + vectorized: true + rowBatchContext: + dataColumnCount: 3 + includeColumns: [1, 2] + dataColumns: nokey:int, key2:int, dt:timestamp + partitionColumnCount: 0 + scratchColumnTypeNames: [string] + Map 2 + Map Operator Tree: + TableScan + alias: item_dim_pq + filterExpr: key1 is not null (type: boolean) + Statistics: Num rows: 2 Data size: 376 Basic stats: COMPLETE Column stats: NONE + TableScan Vectorization: + native: true + vectorizationSchemaColumns: [0:key1:int, 1:name:string, 2:ROW__ID:struct, 3:ROW__IS__DELETED:boolean] + Filter Operator + Filter Vectorization: + className: VectorFilterOperator + native: true + predicateExpression: SelectColumnIsNotNull(col 0:int) + predicate: key1 is not null (type: boolean) + Statistics: Num rows: 2 Data size: 376 Basic stats: COMPLETE Column stats: NONE + Select Operator + expressions: key1 (type: int), name (type: string) + outputColumnNames: _col0, _col1 + Select Vectorization: + className: VectorSelectOperator + native: true + projectedOutputColumnNums: [0, 1] + Statistics: Num rows: 2 Data size: 376 Basic stats: COMPLETE Column stats: NONE + Reduce Output Operator + key expressions: _col0 (type: int) + null sort order: z + sort order: + + Map-reduce partition columns: _col0 (type: int) + Reduce Sink Vectorization: + className: VectorReduceSinkLongOperator + keyColumns: 0:int + native: true + nativeConditionsMet: hive.vectorized.execution.reducesink.new.enabled IS true, hive.execution.engine tez IN [tez] IS true, No PTF TopN IS true, No DISTINCT columns IS true, BinarySortableSerDe for keys IS true, LazyBinarySerDe for values IS true + valueColumns: 1:string + Statistics: Num rows: 2 Data size: 376 Basic stats: COMPLETE Column stats: NONE + value expressions: _col1 (type: string) + Execution mode: vectorized, llap + LLAP IO: all inputs (cache only) + Map Vectorization: + enabled: true + enabledConditionsMet: hive.vectorized.use.vectorized.input.format IS true + inputFormatFeatureSupport: [DECIMAL_64] + featureSupportInUse: [DECIMAL_64] + inputFileFormats: org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat + allNative: true + usesVectorUDFAdaptor: false + vectorized: true + rowBatchContext: + dataColumnCount: 2 + includeColumns: [0, 1] + dataColumns: key1:int, name:string + partitionColumnCount: 0 + scratchColumnTypeNames: [] + + Stage: Stage-0 + Fetch Operator + limit: -1 + Processor Tree: + ListSink + +PREHOOK: query: select key1, key2, name, dt from orders_fact_pq join item_dim_pq on (orders_fact_pq.key2 = item_dim_pq.key1) +order by key2 +PREHOOK: type: QUERY +PREHOOK: Input: default@item_dim_pq +PREHOOK: Input: default@orders_fact_pq +#### A masked pattern was here #### +POSTHOOK: query: select key1, key2, name, dt from orders_fact_pq join item_dim_pq on (orders_fact_pq.key2 = item_dim_pq.key1) +order by key2 +POSTHOOK: type: QUERY +POSTHOOK: Input: default@item_dim_pq +POSTHOOK: Input: default@orders_fact_pq +#### A masked pattern was here #### +101 101 Item 101 2001-01-30 00:00:00 +102 102 Item 102 2004-04-30 00:00:00