HIVE-24231: Enhance shared work optimizer to merge scans with filters on both sides

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

@@ -2592,6 +2592,9 @@ public static enum ConfVars {
         "When shared work optimizer is enabled, whether we should reuse the cache for the broadcast side\n" +
         "of mapjoin operators that share same broadcast input. Requires hive.optimize.shared.work\n" +
         "to be set to true. Tez only."),
+    HIVE_SHARED_WORK_DPPUNION_OPTIMIZATION("hive.optimize.shared.work.dppunion", true,
+        "Enables dppops unioning. This optimization will enable to merge multiple tablescans with different "
+            + "dynamic filters into a single one (with a more complex filter)"),
     HIVE_COMBINE_EQUIVALENT_WORK_OPTIMIZATION("hive.combine.equivalent.work.optimization", true, "Whether to " +
             "combine equivalent work objects during physical optimization.\n This optimization looks for equivalent " +
             "work objects and combines them if they meet certain preconditions. Spark only."),

ql/src/java/org/apache/hadoop/hive/ql/parse/SemiJoinBranchInfo.java

@@ -60,4 +60,8 @@ public void setShouldRemove(boolean shouldRemove) {
       this.shouldRemove = shouldRemove;
     }
   }
+
+  public void setTableScan(TableScanOperator newTs) {
+    ts = newTs;
+  }
 }

ql/src/java/org/apache/hadoop/hive/ql/plan/ExprNodeDescUtils.java

@@ -33,6 +33,7 @@
 import org.apache.hadoop.hive.ql.exec.ReduceSinkOperator;
 import org.apache.hadoop.hive.ql.exec.RowSchema;
 import org.apache.hadoop.hive.ql.exec.UDF;
+import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
 import org.apache.hadoop.hive.ql.exec.Utilities;
 import org.apache.hadoop.hive.ql.optimizer.ConstantPropagateProcFactory;
 import org.apache.hadoop.hive.ql.parse.SemanticException;
@@ -42,12 +43,12 @@
 import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPNotEqual;
 import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPNotNull;
 import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPNull;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPOr;
 import org.apache.hadoop.hive.ql.udf.generic.GenericUDFStruct;
 import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
 import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorUtils;
 import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
 import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector.PrimitiveCategory;
-import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorUtils;
 import org.apache.hadoop.hive.serde2.typeinfo.HiveDecimalUtils;
 import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;
 import org.apache.hadoop.hive.serde2.typeinfo.StructTypeInfo;
@@ -1060,4 +1061,79 @@ public static boolean isStructUDF(ExprNodeDesc columnDesc) {
     return false;
   }
 
+  public static ExprNodeDesc conjunction(List<ExprNodeDesc> semijoinExprNodes) throws UDFArgumentException {
+    if (semijoinExprNodes.isEmpty()) {
+      return null;
+    }
+    if (semijoinExprNodes.size() > 1) {
+      return ExprNodeGenericFuncDesc.newInstance(new GenericUDFOPAnd(), semijoinExprNodes);
+    } else {
+      return semijoinExprNodes.get(0);
+    }
+  }
+
+  public static ExprNodeDesc conjunction(List<ExprNodeDesc> semijoinExprNodes, ExprNodeDesc exprNode)
+      throws UDFArgumentException {
+    if (semijoinExprNodes != null && !semijoinExprNodes.isEmpty()) {
+      if (exprNode != null) {
+        semijoinExprNodes.add(0, exprNode);
+      }
+      if (semijoinExprNodes.size() > 1) {
+        exprNode = ExprNodeGenericFuncDesc.newInstance(new GenericUDFOPAnd(), semijoinExprNodes);
+      } else {
+        exprNode = semijoinExprNodes.get(0);
+      }
+    }
+    return exprNode;
+  }
+
+  public static ExprNodeDesc disjunction(ExprNodeDesc e1, ExprNodeDesc e2) throws UDFArgumentException {
+    if (e1 == null) {
+      return e2;
+    }
+    if (e2 == null) {
+      return e1;
+    }
+    if (e1.isSame(e2)) {
+      return e1;
+    }
+    List<ExprNodeDesc> operands = new ArrayList<ExprNodeDesc>();
+    disjunctiveDecomposition(e1, operands);
+    disjunctiveDecomposition(e2, operands);
+    return disjunction(operands);
+  }
+
+  public static ExprNodeDesc disjunction(List<ExprNodeDesc> operands) throws UDFArgumentException {
+    if (operands.size() == 0) {
+      return null;
+    }
+    if (operands.size() == 1) {
+      return operands.get(0);
+    }
+    return ExprNodeGenericFuncDesc.newInstance(new GenericUDFOPOr(), operands);
+  }
+
+  public static void disjunctiveDecomposition(ExprNodeDesc expr, List<ExprNodeDesc> operands) {
+    if (isOr(expr)) {
+      for (ExprNodeDesc c : expr.getChildren()) {
+        disjunctiveDecomposition(c, operands);
+      }
+    } else {
+      for (ExprNodeDesc o : operands) {
+        if (o.isSame(expr)) {
+          return;
+        }
+      }
+      operands.add(expr);
+    }
+  }
+
+  public static boolean isOr(ExprNodeDesc expr) {
+    if (expr instanceof ExprNodeGenericFuncDesc) {
+      ExprNodeGenericFuncDesc exprNodeGenericFuncDesc = (ExprNodeGenericFuncDesc) expr;
+      return (exprNodeGenericFuncDesc.getGenericUDF() instanceof GenericUDFOPOr);
+    }
+    return false;
+  }
+
 }

ql/src/test/queries/clientpositive/sharedwork_semi.q

@@ -0,0 +1,139 @@
+set hive.explain.user=true;
+set hive.optimize.index.filter=true;
+set hive.auto.convert.join=true;
+set hive.vectorized.execution.enabled=true;
+
+drop table if exists x1_store_sales;
+drop table if exists x1_date_dim;
+
+create table x1_store_sales 
+(
+	ss_sold_date_sk int,
+	ss_item_sk	int
+)
+stored as orc;
+
+create table x1_date_dim
+(
+	d_date_sk	int,
+	d_month_seq	int,
+	d_year		int,
+	d_moy		int
+)
+stored as orc;
+
+insert into x1_date_dim values	(1,1,2000,1),
+				(2,2,2001,2),
+				(3,2,2001,3),
+				(4,2,2001,4),
+				(5,2,2001,5),
+				(6,2,2001,6),
+				(7,2,2001,7),
+				(8,2,2001,8);
+
+insert into x1_store_sales values (1,1),(3,3),(4,4),(5,5),(6,6),(7,7),(8,8),(9,9),(10,10),(11,11);
+
+alter table x1_store_sales update statistics set(
+'numRows'='123456',
+'rawDataSize'='1234567');
+
+alter table x1_date_dim update statistics set(
+'numRows'='28',
+'rawDataSize'='81449');
+
+
+set hive.auto.convert.join.noconditionaltask.size=1;
+set hive.tez.dynamic.partition.pruning=true;
+set hive.tez.dynamic.semijoin.reduction=true;
+set hive.optimize.index.filter=true;
+set hive.tez.bigtable.minsize.semijoin.reduction=1;
+set hive.tez.min.bloom.filter.entries=1;
+set hive.tez.bloom.filter.factor=1.0f;
+set hive.explain.user=false;
+
+set hive.optimize.shared.work.dppunion=false;
+
+select 'expected to see a plan in which ts scan could be shared by combining semijoin conditions';
+-- note: this plan should involve a semijoin reduction
+explain 
+select   sum(s.ss_item_sk)
+ from
+     x1_store_sales s
+     ,x1_date_dim d
+ where
+        1=1
+        and s.ss_sold_date_sk=d.d_date_sk
+	and d.d_moy=3
+union
+select   sum(s.ss_item_sk)
+ from
+     x1_store_sales s
+     ,x1_date_dim d
+ where
+        1=1
+        and s.ss_sold_date_sk=d.d_date_sk
+	and d.d_moy=5
+;
+
+select   sum(s.ss_item_sk)
+ from
+     x1_store_sales s
+     ,x1_date_dim d
+ where
+        1=1
+        and s.ss_sold_date_sk=d.d_date_sk
+	and d.d_moy=3
+union
+select   sum(s.ss_item_sk)
+ from
+     x1_store_sales s
+     ,x1_date_dim d
+ where
+        1=1
+        and s.ss_sold_date_sk=d.d_date_sk
+	and d.d_moy=5
+;
+
+
+set hive.optimize.shared.work.dppunion=true;
+
+select 'expected to see a plan in which x1_store_sales(s) is only scanned once';
+explain 
+select   sum(s.ss_item_sk)
+ from
+     x1_store_sales s
+     ,x1_date_dim d
+ where
+        1=1
+        and s.ss_sold_date_sk=d.d_date_sk
+	and d.d_moy=3
+union
+select   sum(s.ss_item_sk)
+ from
+     x1_store_sales s
+     ,x1_date_dim d
+ where
+        1=1
+        and s.ss_sold_date_sk=d.d_date_sk
+	and d.d_moy=5
+;
+
+select   sum(s.ss_item_sk)
+ from
+     x1_store_sales s
+     ,x1_date_dim d
+ where
+        1=1
+        and s.ss_sold_date_sk=d.d_date_sk
+	and d.d_moy=3
+union
+select   sum(s.ss_item_sk)
+ from
+     x1_store_sales s
+     ,x1_date_dim d
+ where
+        1=1
+        and s.ss_sold_date_sk=d.d_date_sk
+	and d.d_moy=5
+;
+