@Override
  public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx, Object... nodeOutputs)
      throws SemanticException {
    ParseContext parseContext;
    if (procCtx instanceof OptimizeTezProcContext) {
      parseContext = ((OptimizeTezProcContext) procCtx).parseContext;
    } else {
      throw new IllegalArgumentException("expected parseContext " +
          "OptimizeTezProcContext, but found " +
          procCtx.getClass().getName());
    }

    FilterOperator filter = (FilterOperator) nd;
    FilterDesc desc = filter.getConf();

    if (!parseContext.getConf().getBoolVar(ConfVars.TEZ_DYNAMIC_PARTITION_PRUNING)) {
      // nothing to do when the optimization is off
      return null;
    }

    TableScanOperator ts = null;

    if (filter.getParentOperators().size() == 1
        && filter.getParentOperators().get(0) instanceof TableScanOperator) {
      ts = (TableScanOperator) filter.getParentOperators().get(0);
    }

    if (LOG.isDebugEnabled()) {
      LOG.debug("Parent: " + filter.getParentOperators().get(0));
      LOG.debug("Filter: " + desc.getPredicateString());
      LOG.debug("TableScan: " + ts);
    }

    DynamicPartitionPrunerContext removerContext = new DynamicPartitionPrunerContext();

    // collect the dynamic pruning conditions
    removerContext.dynLists.clear();
    GenTezUtils.collectDynamicPruningConditions(desc.getPredicate(), removerContext);

    if (ts == null) {
      // Replace the synthetic predicate with true and bail out
      for (DynamicListContext ctx : removerContext) {
        ExprNodeDesc constNode =
                new ExprNodeConstantDesc(ctx.parent.getTypeInfo(), true);
        replaceExprNode(ctx, desc, constNode);
      }
      return false;
    }

    boolean semiJoin = parseContext.getConf().getBoolVar(ConfVars.TEZ_DYNAMIC_SEMIJOIN_REDUCTION);

    List<ExprNodeDesc> newBetweenNodes = new ArrayList<>();
    List<ExprNodeDesc> newBloomFilterNodes = new ArrayList<>();
    for (DynamicListContext ctx : removerContext) {
      if (ctx.desc.getTypeInfo().getCategory()  != ObjectInspector.Category.PRIMITIVE) {
        // DPP is not supported for complex types.
        // https://issues.apache.org/jira/browse/HIVE-24988
        continue;
      }
      String column = ExprNodeDescUtils.extractColName(ctx.parent);
      boolean semiJoinAttempted = false;

      ExprNodeDesc constNode =
          new ExprNodeConstantDesc(ctx.parent.getTypeInfo(), true);
      if (column != null) {
        // Need unique IDs to refer to each min/max key value in the DynamicValueRegistry
        String keyBaseAlias = "";

        Table table = ts.getConf().getTableMetadata();

        boolean nonEquiJoin = isNonEquiJoin(ctx.parent);
        if (table != null && table.isPartitionKey(column) && !nonEquiJoin) {
          String columnType = table.getPartColByName(column).getType();
          String alias = ts.getConf().getAlias();
          PrunedPartitionList plist = parseContext.getPrunedPartitions(alias, ts);
          if (LOG.isDebugEnabled()) {
            LOG.debug("alias: " + alias);
            LOG.debug("pruned partition list: ");
            if (plist != null) {
              for (Partition p : plist.getPartitions()) {
                LOG.debug(p.getCompleteName());
              }
            }
          }
          // If partKey is a constant, we can check whether the partitions
          // have been already filtered
          if (plist == null || !plist.getPartitions().isEmpty()) {
            LOG.info("Dynamic partitioning: " + table.getCompleteName() + "." + column);
            generateEventOperatorPlan(ctx, parseContext, ts, column, columnType, null);
          } else {
            // all partitions have been statically removed
            LOG.debug("No partition pruning necessary.");
          }
        } else if (table.isNonNative() &&
          table.getStorageHandler().addDynamicSplitPruningEdge(table, ctx.parent)) {
          generateEventOperatorPlan(ctx, parseContext, ts, column,
            table.getCols().stream().filter(e -> e.getName().equals(column)).
          map(e -> e.getType()).findFirst().get(), ctx.parent);
        } else { // semijoin
          LOG.debug("Column " + column + " is not a partition column");
          if (semiJoin && !disableSemiJoinOptDueToExternalTable(parseContext.getConf(), ts, ctx)
                  && ts.getConf().getFilterExpr() != null && !nonEquiJoin) {
            LOG.debug("Initiate semijoin reduction for " + column + " ("
                + ts.getConf().getFilterExpr().getExprString());

            StringBuilder internalColNameBuilder = new StringBuilder();
            StringBuilder colNameBuilder = new StringBuilder();

            // Apply best effort to fetch the correct table alias. If not
            // found, fallback to old logic.
            StringBuilder tabAliasBuilder = new StringBuilder();
            if (getColumnInfo(ctx, internalColNameBuilder, colNameBuilder, tabAliasBuilder)) {
              String colName = colNameBuilder.toString();
              String tableAlias;
              if (tabAliasBuilder.length() > 0) {
                tableAlias = tabAliasBuilder.toString();
              } else {
                //falling back
                Operator<?> op = ctx.generator;

                while (!(op == null || op instanceof TableScanOperator)) {
                  op = op.getParentOperators().get(0);
                }
                tableAlias = (op == null ? "" : ((TableScanOperator) op).
                        getConf().getAlias());
              }

              // Use the tableAlias to generate keyBaseAlias
              keyBaseAlias = ctx.generator.getOperatorId() + "_" + tableAlias
                      + "_" + colName;
              Map<String, List<SemiJoinHint>> hints = parseContext.getSemiJoinHints();
              if (hints != null) {
                // Create semijoin optimizations ONLY for hinted columns
                semiJoinAttempted = processSemiJoinHints(
                        parseContext, ctx, hints, tableAlias,
                        internalColNameBuilder.toString(), colName, ts,
                        keyBaseAlias);
              } else {
                // fallback to regular logic
                semiJoinAttempted = generateSemiJoinOperatorPlan(
                        ctx, parseContext, ts, keyBaseAlias,
                        internalColNameBuilder.toString(), colName, null);
              }
            }
          }
        }

        // If semijoin is attempted then replace the condition with a min-max filter
        // and bloom filter else,
        // we always remove the condition by replacing it with "true"
        if (semiJoinAttempted) {
          List<ExprNodeDesc> betweenArgs = new ArrayList<ExprNodeDesc>();
          betweenArgs.add(new ExprNodeConstantDesc(Boolean.FALSE)); // Do not invert between result
          // add column expression here
          betweenArgs.add(ctx.parent.getChildren().get(0));
          betweenArgs.add(new ExprNodeDynamicValueDesc(new DynamicValue(keyBaseAlias + "_min", ctx.desc.getTypeInfo())));
          betweenArgs.add(new ExprNodeDynamicValueDesc(new DynamicValue(keyBaseAlias + "_max", ctx.desc.getTypeInfo())));
          ExprNodeDesc betweenNode = ExprNodeGenericFuncDesc.newInstance(
                  FunctionRegistry.getFunctionInfo("between").getGenericUDF(), betweenArgs);
          // add column expression for bloom filter
          List<ExprNodeDesc> bloomFilterArgs = new ArrayList<ExprNodeDesc>();
          bloomFilterArgs.add(ctx.parent.getChildren().get(0));
          bloomFilterArgs.add(new ExprNodeDynamicValueDesc(
                  new DynamicValue(keyBaseAlias + "_bloom_filter",
                          TypeInfoFactory.binaryTypeInfo)));
          ExprNodeDesc bloomFilterNode = ExprNodeGenericFuncDesc.newInstance(
                  FunctionRegistry.getFunctionInfo("in_bloom_filter").
                          getGenericUDF(), bloomFilterArgs);
          newBetweenNodes.add(betweenNode);
          newBloomFilterNodes.add(bloomFilterNode);
        }
      }
      replaceExprNode(ctx, desc, constNode);
    }

    if (!newBetweenNodes.isEmpty()) {
      // We need to add the new nodes: first the between nodes, then the bloom filters
      if (FunctionRegistry.isOpAnd(desc.getPredicate())) { // AND
        desc.getPredicate().getChildren().addAll(newBetweenNodes);
        desc.getPredicate().getChildren().addAll(newBloomFilterNodes);
      } else {
        List<ExprNodeDesc> andArgs = new ArrayList<>();
        andArgs.add(desc.getPredicate());
        andArgs.addAll(newBetweenNodes);
        andArgs.addAll(newBloomFilterNodes);
        ExprNodeGenericFuncDesc andExpr = ExprNodeGenericFuncDesc.newInstance(
            FunctionRegistry.getFunctionInfo("and").getGenericUDF(), andArgs);
        // Also pass in filter as tableScan filterExpr
        ts.getConf().setFilterExpr(andExpr);
        desc.setPredicate(andExpr);
      }
    }

    // if we pushed the predicate into the table scan we need to remove the
    // synthetic conditions there.
    cleanTableScanFilters(ts);

    return false;
  }
This optimization looks for expressions of the kind "x IN (RS[n])". If such
an expression made it to a table scan operator and x is a partition column we
can use an existing join to dynamically prune partitions. This class sets up
the infrastructure for that.
ql/src/java/org/apache/hadoop/hive/ql/optimizer/DynamicPartitionPruningOptimization.java