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 × 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 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. 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/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..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
@@ -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,24 @@ 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;
+
+ /**
+ * 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);
}
@@ -161,6 +181,8 @@ public VectorizedParquetRecordReader(InputSplit oldInputSplit, JobConf conf, Fil
DataCache dataCache, Configuration cacheConf, ParquetMetadata parquetMetadata,
Map 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,
@@ -60,10 +81,48 @@ 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
+ 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;
+ }
+ }
+
+ /**
+ * 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 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
@@ -166,32 +225,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 (isFilteredOutDict(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 (isFilteredOutPlain(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 +295,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 (isFilteredOutPlain(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 +321,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 (isFilteredOutPlain(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 +347,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 (isFilteredOutPlain(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 +373,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 (isFilteredOutPlain(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 +390,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 (isFilteredOutPlain(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 +422,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 (isFilteredOutPlain(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 +451,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 (isFilteredOutPlain(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 +477,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 (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.
+ 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 +502,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 (isFilteredOutPlain(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 +524,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 (isFilteredOutPlain(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 +546,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 (isFilteredOutPlain(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 +569,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 (isFilteredOutPlain(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 +597,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 (isFilteredOutPlain(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 +867,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 (isFilteredOutPlain(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..df4af6388706
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/probe/ParquetProbeDecodeState.java
@@ -0,0 +1,219 @@
+/*
+ * 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 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;
+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 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:
+ * 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;
+ }
+}
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:
+ * These tests pin both halves of the contract:
+ * 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 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
+ *
+ * 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).
+ *
+ * {@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.
+ *
+ *
+ *
+ */
+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