This is an automated email from the ASF dual-hosted git repository.
kasakrisz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new 2af70f7fce1 HIVE-30080: Queries can get stuck compiling during
SharedWorkOptimizer (#6806)
2af70f7fce1 is described below
commit 2af70f7fce1957a3fb33d9064d8b4bab12610416
Author: Krisztian Kasa <[email protected]>
AuthorDate: Fri Sep 25 09:46:39 2026 +0200
HIVE-30080: Queries can get stuck compiling during SharedWorkOptimizer
(#6806)
---
.../hive/ql/optimizer/SharedWorkOptimizer.java | 68 +++---
.../hive/ql/optimizer/TestSharedWorkOptimizer.java | 71 ++++++
.../clientpositive/sharedwork_dpp_no_map_join.q | 48 ++++
.../llap/sharedwork_dpp_no_map_join.q.out | 265 +++++++++++++++++++++
4 files changed, 420 insertions(+), 32 deletions(-)
diff --git
a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SharedWorkOptimizer.java
b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SharedWorkOptimizer.java
index 068f81c9870..bbfbc9e2b69 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SharedWorkOptimizer.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SharedWorkOptimizer.java
@@ -2068,44 +2068,48 @@ private static Set<Operator<?>>
findAscendantOperators(SharedWorkOptimizerCache
return visited;
}
- private static Set<Operator<?>> findDescendantWorkOperators(ParseContext
pctx,
+ @VisibleForTesting
+ static Set<Operator<?>> findDescendantWorkOperators(ParseContext pctx,
SharedWorkOptimizerCache optimizerCache, Operator<?> start,
Set<Operator<?>> excludeOps) {
- // Find operators in work
- Set<Operator<?>> workOps = findWorkOperators(optimizerCache, start);
- // Gather output works operators
- Set<Operator<?>> result = new HashSet<Operator<?>>();
- Set<Operator<?>> set;
- while (!workOps.isEmpty()) {
- set = new HashSet<Operator<?>>();
- for (Operator<?> op : workOps) {
- if (excludeOps.contains(op)) {
- continue;
- }
- if (op instanceof ReduceSinkOperator) {
- if (op.getChildOperators() != null) {
- // All children of RS are descendants
- for (Operator<?> child : op.getChildOperators()) {
- set.addAll(findWorkOperators(optimizerCache, child));
- }
- }
- // Semijoin DPP work is considered a descendant because work needs
- // to finish for it to execute
- SemiJoinBranchInfo sjbi = pctx.getRsToSemiJoinBranchInfo().get(op);
- if (sjbi != null) {
- set.addAll(findWorkOperators(optimizerCache, sjbi.getTsOp()));
+
+ // Gather input operators
+ Set<Operator<?>> startWorkOperators = findWorkOperators(optimizerCache,
start);
+ Set<Operator<?>> visited = new HashSet<>();
+ Queue<Operator<?>> remaining = new LinkedList<>(startWorkOperators);
+ while (!remaining.isEmpty()) {
+ Operator<?> op = remaining.poll();
+
+ if (excludeOps.contains(op)) {
+ continue;
+ }
+
+ if (!visited.add(op)) {
+ continue;
+ }
+
+ if (op instanceof ReduceSinkOperator) {
+ if (op.getChildOperators() != null) {
+ // All children of RS are descendants
+ for (Operator<?> child : op.getChildOperators()) {
+ remaining.addAll(findWorkOperators(optimizerCache, child));
}
- } else if(op.getConf() instanceof DynamicPruningEventDesc) {
- // DPP work is considered a descendant because work needs
- // to finish for it to execute
- set.addAll(findWorkOperators(
- optimizerCache, ((DynamicPruningEventDesc)
op.getConf()).getTableScan()));
}
+ // Semijoin DPP work is considered a descendant because work needs
+ // to finish for it to execute
+ SemiJoinBranchInfo sjbi = pctx.getRsToSemiJoinBranchInfo().get(op);
+ if (sjbi != null) {
+ remaining.addAll(findWorkOperators(optimizerCache, sjbi.getTsOp()));
+ }
+ } else if (op.getConf() instanceof DynamicPruningEventDesc) {
+ // DPP work is considered a descendant because work needs
+ // to finish for it to execute
+ remaining.addAll(findWorkOperators(optimizerCache,
((DynamicPruningEventDesc) op.getConf()).getTableScan()));
}
- workOps = set;
- result.addAll(set);
}
- return result;
+
+ visited.removeAll(startWorkOperators);
+ return visited;
}
// Stores result in cache
diff --git
a/ql/src/test/org/apache/hadoop/hive/ql/optimizer/TestSharedWorkOptimizer.java
b/ql/src/test/org/apache/hadoop/hive/ql/optimizer/TestSharedWorkOptimizer.java
index c352f2273b3..90c5e3e1496 100644
---
a/ql/src/test/org/apache/hadoop/hive/ql/optimizer/TestSharedWorkOptimizer.java
+++
b/ql/src/test/org/apache/hadoop/hive/ql/optimizer/TestSharedWorkOptimizer.java
@@ -24,12 +24,16 @@
import org.apache.hadoop.hive.ql.exec.MapJoinOperator;
import org.apache.hadoop.hive.ql.exec.Operator;
import org.apache.hadoop.hive.ql.exec.OperatorFactory;
+import org.apache.hadoop.hive.ql.exec.ReduceSinkOperator;
import org.apache.hadoop.hive.ql.exec.TableScanOperator;
import org.apache.hadoop.hive.ql.metadata.Table;
import
org.apache.hadoop.hive.ql.optimizer.SharedWorkOptimizer.SharedWorkOptimizerCache;
import org.apache.hadoop.hive.ql.optimizer.SharedWorkOptimizer.TSComparator;
import org.apache.hadoop.hive.ql.parse.JoinType;
+import org.apache.hadoop.hive.ql.parse.ParseContext;
import org.apache.hadoop.hive.ql.parse.SemanticException;
+import org.apache.hadoop.hive.ql.parse.SemiJoinBranchInfo;
+import org.apache.hadoop.hive.ql.plan.DynamicPruningEventDesc;
import org.apache.hadoop.hive.ql.plan.ExprNodeColumnDesc;
import org.apache.hadoop.hive.ql.plan.ExprNodeConstantDesc;
import org.apache.hadoop.hive.ql.plan.ExprNodeDesc;
@@ -49,9 +53,11 @@
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.EnumSet;
import java.util.List;
import java.util.Properties;
+import java.util.Set;
import static
org.apache.hadoop.hive.ql.plan.ReduceSinkDesc.ReducerTraits.AUTOPARALLEL;
import static
org.apache.hadoop.hive.ql.plan.ReduceSinkDesc.ReducerTraits.FIXED;
@@ -278,6 +284,71 @@ public void testSharedWorkOptimizerCache() {
}
+ @Test
+ public void testFindDescendantWorkOperatorsTerminatesOnSemiJoinCycle() {
+ // Map 1
+ TableScanOperator ts1 = getTsOp();
+ Operator<?> rs1 = OperatorFactory.getAndMakeChild(getReduceSinkDesc(),
ts1);
+
+ // Reducer 2 <- Map1
+ Operator<?> fil = OperatorFactory.getAndMakeChild(getFilterDesc(0), rs1);
+ Operator<?> rs2 = OperatorFactory.getAndMakeChild(getReduceSinkDesc(),
fil);
+
+ // Add semi join edge
+ ParseContext pctx = new ParseContext();
+ pctx.getRsToSemiJoinBranchInfo().put((ReduceSinkOperator) rs2, new
SemiJoinBranchInfo(ts1));
+
+ Set<Operator<?>> descendants =
SharedWorkOptimizer.findDescendantWorkOperators(
+ pctx, new SharedWorkOptimizerCache(), ts1, Collections.emptySet());
+
+ assertEquals(2, descendants.size());
+ assertTrue(descendants.contains(fil));
+ assertTrue(descendants.contains(rs2));
+ }
+
+ @Test
+ public void testFindDescendantWorkOperatorsTerminatesOnDppCycle() {
+ // Map 1
+ TableScanOperator ts1 = getTsOp();
+ Operator<?> rs1 = OperatorFactory.getAndMakeChild(getReduceSinkDesc(),
ts1);
+
+ // Reducer 2 <- Map 1
+ Operator<?> fil = OperatorFactory.getAndMakeChild(getFilterDesc(0), rs1);
+ DynamicPruningEventDesc dppDesc = new DynamicPruningEventDesc();
+ dppDesc.setTableScan(ts1);
+ Operator<?> eventOp = OperatorFactory.getAndMakeChild(dppDesc, fil);
+
+ Set<Operator<?>> descendants =
SharedWorkOptimizer.findDescendantWorkOperators(
+ new ParseContext(), new SharedWorkOptimizerCache(), ts1,
Collections.emptySet());
+
+ assertEquals(2, descendants.size());
+ assertTrue(descendants.contains(fil));
+ assertTrue(descendants.contains(eventOp));
+ }
+
+ @Test
+ public void testFindDescendantWorkOperatorsDoesNotReturnExcludedOperators() {
+ // Map 1
+ TableScanOperator ts1 = getTsOp();
+ Operator<?> rs1 = OperatorFactory.getAndMakeChild(getReduceSinkDesc(),
ts1);
+
+ // Reducer 2 <- Map 1
+ Operator<?> fil = OperatorFactory.getAndMakeChild(getFilterDesc(0), rs1);
+ Operator<?> exclFil = OperatorFactory.getAndMakeChild(getFilterDesc(0),
fil);
+ Operator<?> rs2 = OperatorFactory.getAndMakeChild(getReduceSinkDesc(),
fil);
+
+ Set<Operator<?>> descendants =
SharedWorkOptimizer.findDescendantWorkOperators(
+ new ParseContext(), new SharedWorkOptimizerCache(), ts1,
Set.of(exclFil));
+
+ assertEquals(2, descendants.size());
+ assertTrue(descendants.contains(fil));
+ assertTrue(descendants.contains(rs2));
+ }
+
+ private FilterDesc getFilterDesc(int constVal) {
+ return new FilterDesc(new ExprNodeConstantDesc(constVal), true);
+ }
+
private ReduceSinkDesc getReduceSinkDesc() {
TableDesc dummyKeySerializeInfo = new TableDesc();
dummyKeySerializeInfo.setProperties(new Properties());
diff --git a/ql/src/test/queries/clientpositive/sharedwork_dpp_no_map_join.q
b/ql/src/test/queries/clientpositive/sharedwork_dpp_no_map_join.q
new file mode 100644
index 00000000000..4bf298628b8
--- /dev/null
+++ b/ql/src/test/queries/clientpositive/sharedwork_dpp_no_map_join.q
@@ -0,0 +1,48 @@
+set hive.auto.convert.join=false;
+
+CREATE TABLE swo_hang (
+ a_c string,
+ b_c int
+)
+PARTITIONED BY (
+ day int
+)
+STORED AS ORC;
+
+INSERT INTO swo_hang VALUES ('userid1', 1, 2);
+
+explain SELECT
+ u.day
+FROM (
+ SELECT
+ v.day,
+ v.a_c
+ FROM (
+ SELECT day,
+ a_c,
+ MIN(b_c)
+ FROM swo_hang
+ WHERE day >=1
+ AND a_c IS NOT NULL
+ GROUP BY day, a_c
+ ) v
+ JOIN (
+ SELECT day,
+ a_c,
+ MIN(b_c)
+ FROM swo_hang
+ WHERE day >=1
+ AND a_c IS NOT NULL
+ GROUP BY day, a_c
+ ) a
+ ON v.a_c = a.a_c
+ AND v.day = a.day
+ LEFT JOIN (
+ SELECT a_c
+ FROM swo_hang
+ WHERE day >=1
+ AND a_c IS NOT NULL
+ ) p
+ ON v.a_c = p.a_c
+ WHERE p.a_c IS NULL
+) u;
diff --git
a/ql/src/test/results/clientpositive/llap/sharedwork_dpp_no_map_join.q.out
b/ql/src/test/results/clientpositive/llap/sharedwork_dpp_no_map_join.q.out
new file mode 100644
index 00000000000..8f4ac236d81
--- /dev/null
+++ b/ql/src/test/results/clientpositive/llap/sharedwork_dpp_no_map_join.q.out
@@ -0,0 +1,265 @@
+PREHOOK: query: CREATE TABLE swo_hang (
+ a_c string,
+ b_c int
+)
+PARTITIONED BY (
+ day int
+)
+STORED AS ORC
+PREHOOK: type: CREATETABLE
+PREHOOK: Output: database:default
+PREHOOK: Output: default@swo_hang
+POSTHOOK: query: CREATE TABLE swo_hang (
+ a_c string,
+ b_c int
+)
+PARTITIONED BY (
+ day int
+)
+STORED AS ORC
+POSTHOOK: type: CREATETABLE
+POSTHOOK: Output: database:default
+POSTHOOK: Output: default@swo_hang
+PREHOOK: query: INSERT INTO swo_hang VALUES ('userid1', 1, 2)
+PREHOOK: type: QUERY
+PREHOOK: Input: _dummy_database@_dummy_table
+PREHOOK: Output: default@swo_hang
+POSTHOOK: query: INSERT INTO swo_hang VALUES ('userid1', 1, 2)
+POSTHOOK: type: QUERY
+POSTHOOK: Input: _dummy_database@_dummy_table
+POSTHOOK: Output: default@swo_hang
+POSTHOOK: Output: default@swo_hang@day=2
+POSTHOOK: Lineage: swo_hang PARTITION(day=2).a_c SCRIPT []
+POSTHOOK: Lineage: swo_hang PARTITION(day=2).b_c SCRIPT []
+PREHOOK: query: explain SELECT
+ u.day
+FROM (
+ SELECT
+ v.day,
+ v.a_c
+ FROM (
+ SELECT day,
+ a_c,
+ MIN(b_c)
+ FROM swo_hang
+ WHERE day >=1
+ AND a_c IS NOT NULL
+ GROUP BY day, a_c
+ ) v
+ JOIN (
+ SELECT day,
+ a_c,
+ MIN(b_c)
+ FROM swo_hang
+ WHERE day >=1
+ AND a_c IS NOT NULL
+ GROUP BY day, a_c
+ ) a
+ ON v.a_c = a.a_c
+ AND v.day = a.day
+ LEFT JOIN (
+ SELECT a_c
+ FROM swo_hang
+ WHERE day >=1
+ AND a_c IS NOT NULL
+ ) p
+ ON v.a_c = p.a_c
+ WHERE p.a_c IS NULL
+) u
+PREHOOK: type: QUERY
+PREHOOK: Input: default@swo_hang
+PREHOOK: Input: default@swo_hang@day=2
+#### A masked pattern was here ####
+POSTHOOK: query: explain SELECT
+ u.day
+FROM (
+ SELECT
+ v.day,
+ v.a_c
+ FROM (
+ SELECT day,
+ a_c,
+ MIN(b_c)
+ FROM swo_hang
+ WHERE day >=1
+ AND a_c IS NOT NULL
+ GROUP BY day, a_c
+ ) v
+ JOIN (
+ SELECT day,
+ a_c,
+ MIN(b_c)
+ FROM swo_hang
+ WHERE day >=1
+ AND a_c IS NOT NULL
+ GROUP BY day, a_c
+ ) a
+ ON v.a_c = a.a_c
+ AND v.day = a.day
+ LEFT JOIN (
+ SELECT a_c
+ FROM swo_hang
+ WHERE day >=1
+ AND a_c IS NOT NULL
+ ) p
+ ON v.a_c = p.a_c
+ WHERE p.a_c IS NULL
+) u
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@swo_hang
+POSTHOOK: Input: default@swo_hang@day=2
+#### A masked pattern was here ####
+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:
+ Reducer 2 <- Map 1 (SIMPLE_EDGE), Map 4 (SIMPLE_EDGE)
+ Reducer 3 <- Map 1 (SIMPLE_EDGE), Reducer 2 (SIMPLE_EDGE)
+#### A masked pattern was here ####
+ Vertices:
+ Map 1
+ Map Operator Tree:
+ TableScan
+ alias: swo_hang
+ filterExpr: a_c is not null (type: boolean)
+ Statistics: Num rows: 1 Data size: 95 Basic stats: COMPLETE
Column stats: COMPLETE
+ Filter Operator
+ predicate: a_c is not null (type: boolean)
+ Statistics: Num rows: 1 Data size: 95 Basic stats:
COMPLETE Column stats: COMPLETE
+ Group By Operator
+ keys: a_c (type: string), day (type: int)
+ minReductionHashAggr: 0.4
+ mode: hash
+ outputColumnNames: _col0, _col1
+ Statistics: Num rows: 1 Data size: 95 Basic stats:
COMPLETE Column stats: COMPLETE
+ Reduce Output Operator
+ key expressions: _col0 (type: string), _col1 (type:
int)
+ null sort order: zz
+ sort order: ++
+ Map-reduce partition columns: _col0 (type: string),
_col1 (type: int)
+ Statistics: Num rows: 1 Data size: 95 Basic stats:
COMPLETE Column stats: COMPLETE
+ Select Operator
+ expressions: a_c (type: string)
+ outputColumnNames: _col0
+ Statistics: Num rows: 1 Data size: 91 Basic stats:
COMPLETE Column stats: COMPLETE
+ Group By Operator
+ keys: _col0 (type: string)
+ minReductionHashAggr: 0.4
+ mode: hash
+ outputColumnNames: _col0
+ Statistics: Num rows: 1 Data size: 91 Basic stats:
COMPLETE Column stats: COMPLETE
+ Reduce Output Operator
+ key expressions: _col0 (type: string)
+ null sort order: z
+ sort order: +
+ Map-reduce partition columns: _col0 (type: string)
+ Statistics: Num rows: 1 Data size: 91 Basic stats:
COMPLETE Column stats: COMPLETE
+ Execution mode: vectorized, llap
+ LLAP IO: all inputs
+ Map 4
+ Map Operator Tree:
+ TableScan
+ alias: swo_hang
+ filterExpr: a_c is not null (type: boolean)
+ Statistics: Num rows: 1 Data size: 95 Basic stats: COMPLETE
Column stats: COMPLETE
+ Filter Operator
+ predicate: a_c is not null (type: boolean)
+ Statistics: Num rows: 1 Data size: 95 Basic stats:
COMPLETE Column stats: COMPLETE
+ Group By Operator
+ keys: a_c (type: string), day (type: int)
+ minReductionHashAggr: 0.4
+ mode: hash
+ outputColumnNames: _col0, _col1
+ Statistics: Num rows: 1 Data size: 95 Basic stats:
COMPLETE Column stats: COMPLETE
+ Reduce Output Operator
+ key expressions: _col0 (type: string), _col1 (type:
int)
+ null sort order: zz
+ sort order: ++
+ Map-reduce partition columns: _col0 (type: string),
_col1 (type: int)
+ Statistics: Num rows: 1 Data size: 95 Basic stats:
COMPLETE Column stats: COMPLETE
+ Execution mode: vectorized, llap
+ LLAP IO: all inputs
+ Reducer 2
+ Reduce Operator Tree:
+ Group By Operator
+ keys: KEY._col0 (type: string), KEY._col1 (type: int)
+ mode: mergepartial
+ outputColumnNames: _col0, _col1
+ Statistics: Num rows: 1 Data size: 95 Basic stats: COMPLETE
Column stats: COMPLETE
+ Select Operator
+ expressions: _col1 (type: int), _col0 (type: string)
+ outputColumnNames: _col0, _col1
+ Statistics: Num rows: 1 Data size: 95 Basic stats: COMPLETE
Column stats: COMPLETE
+ Dummy Store
+ Execution mode: llap
+ Reduce Operator Tree:
+ Group By Operator
+ keys: KEY._col0 (type: string), KEY._col1 (type: int)
+ mode: mergepartial
+ outputColumnNames: _col0, _col1
+ Statistics: Num rows: 1 Data size: 95 Basic stats: COMPLETE
Column stats: COMPLETE
+ Select Operator
+ expressions: _col1 (type: int), _col0 (type: string)
+ outputColumnNames: _col0, _col1
+ Statistics: Num rows: 1 Data size: 95 Basic stats: COMPLETE
Column stats: COMPLETE
+ Select Operator
+ expressions: _col0 (type: int)
+ outputColumnNames: _col0
+ Statistics: Num rows: 1 Data size: 4 Basic stats: COMPLETE
Column stats: COMPLETE
+ Group By Operator
+ keys: _col0 (type: int)
+ minReductionHashAggr: 0.4
+ mode: hash
+ outputColumnNames: _col0
+ Statistics: Num rows: 1 Data size: 4 Basic stats:
COMPLETE Column stats: COMPLETE
+ Dynamic Partitioning Event Operator
+ Target column: day (int)
+ Target Input: swo_hang
+ Partition key expr: day
+ Statistics: Num rows: 1 Data size: 4 Basic stats:
COMPLETE Column stats: COMPLETE
+ Target Vertex: Map 4
+ Merge Join Operator
+ condition map:
+ Inner Join 0 to 1
+ keys:
+ 0 _col1 (type: string), _col0 (type: int)
+ 1 _col1 (type: string), _col0 (type: int)
+ outputColumnNames: _col0, _col1
+ Statistics: Num rows: 1 Data size: 95 Basic stats:
COMPLETE Column stats: COMPLETE
+ Reduce Output Operator
+ key expressions: _col1 (type: string)
+ null sort order: z
+ sort order: +
+ Map-reduce partition columns: _col1 (type: string)
+ Statistics: Num rows: 1 Data size: 95 Basic stats:
COMPLETE Column stats: COMPLETE
+ value expressions: _col0 (type: int)
+ Reducer 3
+ Execution mode: llap
+ Reduce Operator Tree:
+ Merge Join Operator
+ condition map:
+ Anti Join 0 to 1
+ keys:
+ 0 _col1 (type: string)
+ 1 _col0 (type: string)
+ outputColumnNames: _col0
+ Statistics: Num rows: 1 Data size: 4 Basic stats: COMPLETE
Column stats: COMPLETE
+ File Output Operator
+ compressed: false
+ Statistics: Num rows: 1 Data size: 4 Basic stats: COMPLETE
Column stats: COMPLETE
+ 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
+
+ Stage: Stage-0
+ Fetch Operator
+ limit: -1
+ Processor Tree:
+ ListSink
+
