HIVE-30019: ProbeDecode for the vectorized Parquet reader

common/src/java/org/apache/hadoop/hive/conf/HiveConf.java

@@ -2667,6 +2667,15 @@
             + "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"),
 

itests/hive-jmh/pom.xml

@@ -98,6 +98,24 @@
       <version>${mockito-inline.version}</version>
     </dependency>
   </dependencies>
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-compiler-plugin</artifactId>
+        <configuration>
+          <proc>full</proc>
+          <annotationProcessorPaths>
+            <path>
+              <groupId>org.openjdk.jmh</groupId>
+              <artifactId>jmh-generator-annprocess</artifactId>
+              <version>${jmh.version}</version>
+            </path>
+          </annotationProcessorPaths>
+        </configuration>
+      </plugin>
+    </plugins>
+  </build>
   <profiles>
     <profile>
       <id>perf</id>
@@ -118,7 +136,11 @@
                     <transformer implementation="com.github.edwgiz.mavenShadePlugin.log4j2CacheTransformer.PluginsCacheFileTransformer"/>
                     <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                       <mainClass>org.openjdk.jmh.Main</mainClass>
+                      <manifestEntries>
+                        <Multi-Release>true</Multi-Release>
+                      </manifestEntries>
                     </transformer>
+                    <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
                   </transformers>
                   <filters>
                     <filter>

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.
+   *
+   * <p>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
    */

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;

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 @@
     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.
+   *
+   * <p>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.
+   *
+   * <p>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;
   }

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<String, Object> 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;
@@ -420,8 +442,61 @@ 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;
+    }
+
     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 +506,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;
@@ -549,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<VectorizedColumnReader> fieldReaders = new ArrayList<>();