diff --git a/itests/src/test/resources/testconfiguration.properties b/itests/src/test/resources/testconfiguration.properties index bfefc3b5f023..2a145f24b184 100644 --- a/itests/src/test/resources/testconfiguration.properties +++ b/itests/src/test/resources/testconfiguration.properties @@ -34,6 +34,7 @@ minitez.query.files=\ orc_vectorization_ppd.q,\ partition_default_name_change_numeric.q,\ tez_complextype_with_null.q,\ + tez_dummystore_reduce_side_reuse.q,\ tez_tag.q,\ tez_union_udtf.q,\ tez_union_with_udf.q,\ diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/TezDummyStoreOperator.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/TezDummyStoreOperator.java index 08c1cc408c37..448bc6b27a41 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/TezDummyStoreOperator.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/TezDummyStoreOperator.java @@ -18,8 +18,11 @@ package org.apache.hadoop.hive.ql.exec; +import java.util.List; + import org.apache.hadoop.hive.ql.CompilationOpContext; import org.apache.hadoop.hive.ql.metadata.HiveException; +import org.apache.hadoop.hive.ql.plan.OperatorDesc; /** * A dummy store operator same as the dummy store operator but for tez. This is required so that we @@ -59,6 +62,16 @@ public void setFetchDone(boolean fetchDone) { @Override public void closeOp(boolean abort) throws HiveException { + List> children = getChildOperators(); + if (children != null && !children.isEmpty()) { + for (Operator child : children) { + List> parents = child.getParentOperators(); + if (parents != null) { + parents.remove(this); + } + } + children.clear(); + } super.closeOp(abort); fetchDone = false; } diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java index 96714f29dc6f..8406980cef2e 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java @@ -358,13 +358,16 @@ void close() { if (abort) { setAborted(true); } - reducer.close(abort); + // Close Tree1 (DummyStore chain) before Tree2 (MergeJoin chain). + // DummyStore.closeOp() removes itself from MergeJoin.parentOperators, so that when + // Tree2 closes, MergeJoin.allInitializedParentsAreClosed() returns true and + // checkAndGenObject() fires to emit the last join group. if (mergeWorkList != null) { for (BaseWork redWork : mergeWorkList) { ((ReduceWork) redWork).getReducer().close(abort); } } - + reducer.close(abort); // Need to close the dummyOps as well. The operator pipeline // is not considered "closed/done" unless all operators are // done. For broadcast joins that includes the dummy parents. diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezUtils.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezUtils.java index df297cd31775..7fbfef464172 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezUtils.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezUtils.java @@ -31,6 +31,7 @@ import org.apache.hadoop.hive.ql.ddl.DDLUtils; import org.apache.hadoop.hive.ql.exec.AbstractFileMergeOperator; import org.apache.hadoop.hive.ql.exec.AppMasterEventOperator; +import org.apache.hadoop.hive.ql.exec.CommonMergeJoinOperator; import org.apache.hadoop.hive.ql.exec.FetchTask; import org.apache.hadoop.hive.ql.exec.FileSinkOperator; import org.apache.hadoop.hive.ql.exec.FilterOperator; @@ -110,6 +111,14 @@ public static ReduceWork createReduceWork( assert context.parentOfRoot instanceof ReduceSinkOperator; ReduceSinkOperator reduceSink = (ReduceSinkOperator) context.parentOfRoot; + // If this RS feeds into a CommonMergeJoinOp (possibly via intermediate ops like GroupBy + // or DummyStore), normalize all RS inputs to the same numReducers so that every mapper + // writes to the same number of partitions and join keys land in the same task. Also + // disable Tez's dynamic auto-parallelism for these RS: ShuffleVertexManager can still + // shrink the actual task count at runtime independently of our numReducers value, which + // would silently break the same-key-same-partition guarantee this normalization relies on. + boolean isMergeJoinInput = normalizeMergeJoinReducers(reduceSink); + reduceWork.setNumReduceTasks(reduceSink.getConf().getNumReducers()); reduceWork.setSlowStart(reduceSink.getConf().isSlowStart()); float minSrcFraction = context.conf.getFloat( @@ -127,7 +136,8 @@ public static ReduceWork createReduceWork( reduceSink.getConf().getReducerTraits().remove(AUTOPARALLEL); } - if (isAutoReduceParallelism && reduceSink.getConf().getReducerTraits().contains(AUTOPARALLEL)) { + if (isAutoReduceParallelism && !isMergeJoinInput + && reduceSink.getConf().getReducerTraits().contains(AUTOPARALLEL)) { // configured limit for reducers final int maxReducers = context.conf.getIntVar(HiveConf.ConfVars.MAX_REDUCERS); @@ -187,6 +197,87 @@ public static ReduceWork createReduceWork( return reduceWork; } + /** + * If the given RS leads (through child ops) to a CommonMergeJoinOperator, normalizes all RS + * ancestors of that MergeJoinOp's parents to use the same maximum numReducers. This is a + * last-resort fix: even if earlier optimizer passes failed to align numReducers, we correct + * it at edge-creation time so each mapper writes the same partition count. + */ + private static boolean normalizeMergeJoinReducers(ReduceSinkOperator reduceSink) { + CommonMergeJoinOperator mergeJoin = findDownstreamMergeJoin(reduceSink, 8); + if (mergeJoin == null) { + return false; + } + List allRS = new ArrayList<>(); + for (Operator parent : mergeJoin.getParentOperators()) { + ReduceSinkOperator rsAnc = findUpstreamRS(parent, 8); + if (rsAnc != null) { + allRS.add(rsAnc); + } + } + if (allRS.isEmpty()) { + return true; + } + int maxNR = 0; + for (ReduceSinkOperator rs : allRS) { + if (rs.getConf().getNumReducers() > maxNR) { + maxNR = rs.getConf().getNumReducers(); + } + } + if (maxNR <= 0) { + return true; + } + for (ReduceSinkOperator rs : allRS) { + int orig = rs.getConf().getNumReducers(); + boolean hasAutoParallel = rs.getConf().getReducerTraits().contains(AUTOPARALLEL); + boolean hasUniform = rs.getConf().getReducerTraits().contains(UNIFORM); + if (orig != maxNR || hasAutoParallel || hasUniform) { + rs.getConf().setNumReducers(maxNR); + // ReduceSinkDesc.setReducerTraits() is ADDITIVE (this.reduceTraits.addAll(traits)) in + // every branch except when the passed set contains FIXED, in which case it explicitly + // removes AUTOPARALLEL/UNIFORM first. Passing a traits set with those already removed + // is therefore a no-op against the existing (unremoved) traits - FIXED must be passed + // to actually clear them. + rs.getConf().setReducerTraits(EnumSet.of(ReduceSinkDesc.ReducerTraits.FIXED)); + } + } + return true; + } + + private static CommonMergeJoinOperator findDownstreamMergeJoin( + ReduceSinkOperator rs, int maxDepth) { + Operator curr = rs; + for (int d = 0; d < maxDepth; d++) { + List> children = curr.getChildOperators(); + if (children == null || children.isEmpty()) { + break; + } + curr = children.get(0); + if (curr instanceof CommonMergeJoinOperator) { + return (CommonMergeJoinOperator) curr; + } + } + return null; + } + + private static ReduceSinkOperator findUpstreamRS(Operator op, int maxDepth) { + if (op instanceof ReduceSinkOperator) { + return (ReduceSinkOperator) op; + } + Operator curr = op; + for (int d = 0; d < maxDepth; d++) { + List> parents = curr.getParentOperators(); + if (parents == null || parents.isEmpty()) { + break; + } + curr = parents.get(0); + if (curr instanceof ReduceSinkOperator) { + return (ReduceSinkOperator) curr; + } + } + return null; + } + /** * Checks if there is a Bucket Map Join (BMJ) following a hierarchy: * ReduceSinkOperator (RS) -> ReduceSinkOperator (RS) -> MapJoinOperator (MJ). diff --git a/ql/src/test/org/apache/hadoop/hive/ql/exec/TestTezDummyStoreOperator.java b/ql/src/test/org/apache/hadoop/hive/ql/exec/TestTezDummyStoreOperator.java new file mode 100644 index 000000000000..db85920e4682 --- /dev/null +++ b/ql/src/test/org/apache/hadoop/hive/ql/exec/TestTezDummyStoreOperator.java @@ -0,0 +1,140 @@ +/* + * 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.exec; + +import java.util.ArrayList; +import java.util.List; + +import org.apache.hadoop.hive.ql.metadata.HiveException; +import org.apache.hadoop.hive.ql.plan.OperatorDesc; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +/** + * Unit tests for {@link TezDummyStoreOperator#closeOp(boolean)}. + * + * When a Tez container is reused for a subsequent task of the same reducer vertex, Hive's + * per-query ObjectCache (keyed by vertex name) can hand the new task the same already- + * initialized DummyStore/MergeJoin operator instances the previous task used. If closeOp() + * leaves stale wiring behind (this operator still listed as a parent of its former child, + * and/or that child still referenced from childOperators), the reused instances carry over + * incorrect state into the next task's execution, causing + * ReduceRecordProcessor.getJoinParentOp() to walk into the wrong operator and throw + * IllegalStateException("Was expecting dummy store operator but found: ..."). + */ +public class TestTezDummyStoreOperator { + + private TezDummyStoreOperator dummyStore; + + @Before + public void setUp() { + dummyStore = new TezDummyStoreOperator(); + } + + @Test + public void testCloseOpRemovesSelfFromChildParents() throws HiveException { + CommonMergeJoinOperator mergeJoin = new CommonMergeJoinOperator(); + + wireParentChild(dummyStore, mergeJoin); + + Assert.assertTrue("precondition: mergeJoin's parents must contain dummyStore", + mergeJoin.getParentOperators().contains(dummyStore)); + Assert.assertTrue("precondition: dummyStore's children must contain mergeJoin", + dummyStore.getChildOperators().contains(mergeJoin)); + + dummyStore.closeOp(false); + + Assert.assertFalse("dummyStore must be removed from its former child's parentOperators", + mergeJoin.getParentOperators().contains(dummyStore)); + Assert.assertTrue("dummyStore's own childOperators must be cleared", + dummyStore.getChildOperators().isEmpty()); + } + + @Test + public void testCloseOpRemovesSelfFromAllChildrenWhenMultiple() throws HiveException { + CommonMergeJoinOperator child1 = new CommonMergeJoinOperator(); + CommonMergeJoinOperator child2 = new CommonMergeJoinOperator(); + + List> children = new ArrayList<>(); + children.add(child1); + children.add(child2); + dummyStore.setChildOperators(children); + + List> parentsOfChild1 = new ArrayList<>(); + parentsOfChild1.add(dummyStore); + child1.setParentOperators(parentsOfChild1); + + List> parentsOfChild2 = new ArrayList<>(); + parentsOfChild2.add(dummyStore); + child2.setParentOperators(parentsOfChild2); + + dummyStore.closeOp(false); + + Assert.assertFalse(child1.getParentOperators().contains(dummyStore)); + Assert.assertFalse(child2.getParentOperators().contains(dummyStore)); + Assert.assertTrue(dummyStore.getChildOperators().isEmpty()); + } + + @Test + public void testCloseOpWithNoChildrenDoesNotThrow() throws HiveException { + dummyStore.setChildOperators(new ArrayList<>()); + dummyStore.closeOp(false); + Assert.assertTrue(dummyStore.getChildOperators().isEmpty()); + } + + @Test + public void testCloseOpWithNullChildrenDoesNotThrow() throws HiveException { + dummyStore.setChildOperators(null); + // setChildOperators(null) replaces null with an empty list (see Operator#setChildOperators). + dummyStore.closeOp(false); + Assert.assertTrue(dummyStore.getChildOperators().isEmpty()); + } + + @Test + public void testCloseOpDoesNotAffectUnrelatedParentReferences() throws HiveException { + CommonMergeJoinOperator mergeJoin = new CommonMergeJoinOperator(); + TezDummyStoreOperator otherDummyStore = new TezDummyStoreOperator(); + + List> parents = new ArrayList<>(); + parents.add(otherDummyStore); + parents.add(dummyStore); + mergeJoin.setParentOperators(parents); + + List> children = new ArrayList<>(); + children.add(mergeJoin); + dummyStore.setChildOperators(children); + + dummyStore.closeOp(false); + + Assert.assertFalse(mergeJoin.getParentOperators().contains(dummyStore)); + Assert.assertTrue("unrelated parent reference must be left untouched", + mergeJoin.getParentOperators().contains(otherDummyStore)); + } + + private static void wireParentChild( + Operator parent, Operator child) { + List> children = new ArrayList<>(); + children.add(child); + parent.setChildOperators(children); + + List> parents = new ArrayList<>(); + parents.add(parent); + child.setParentOperators(parents); + } +} diff --git a/ql/src/test/org/apache/hadoop/hive/ql/parse/TestGenTezUtilsMergeJoinNumReducers.java b/ql/src/test/org/apache/hadoop/hive/ql/parse/TestGenTezUtilsMergeJoinNumReducers.java new file mode 100644 index 000000000000..e01e4b09f4ba --- /dev/null +++ b/ql/src/test/org/apache/hadoop/hive/ql/parse/TestGenTezUtilsMergeJoinNumReducers.java @@ -0,0 +1,147 @@ +/* + * 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.parse; + +import static org.junit.Assert.assertEquals; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.Properties; + +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.conf.HiveConfForTest; +import org.apache.hadoop.hive.ql.CompilationOpContext; +import org.apache.hadoop.hive.ql.Context; +import org.apache.hadoop.hive.ql.exec.CommonMergeJoinOperator; +import org.apache.hadoop.hive.ql.exec.GroupByOperator; +import org.apache.hadoop.hive.ql.exec.Operator; +import org.apache.hadoop.hive.ql.exec.ReduceSinkOperator; +import org.apache.hadoop.hive.ql.exec.Task; +import org.apache.hadoop.hive.ql.exec.TezDummyStoreOperator; +import org.apache.hadoop.hive.ql.plan.GroupByDesc; +import org.apache.hadoop.hive.ql.plan.MapWork; +import org.apache.hadoop.hive.ql.plan.OperatorDesc; +import org.apache.hadoop.hive.ql.plan.ReduceSinkDesc; +import org.apache.hadoop.hive.ql.plan.ReduceWork; +import org.apache.hadoop.hive.ql.plan.TableDesc; +import org.apache.hadoop.hive.ql.plan.TezWork; +import org.apache.hadoop.hive.ql.session.SessionState; +import org.junit.Before; +import org.junit.Test; + +/** + * Reproduces a reduce-side merge join bug: {@link GenTezUtils#createReduceWork} sets each + * {@link ReduceWork}'s numReduceTasks purely from that branch's own ReduceSinkOperator, with no + * cross-check against sibling ReduceSinkOperators that feed the same downstream + * {@link CommonMergeJoinOperator}. When two such siblings end up with different numReducers + * (e.g. because Tez auto-parallelism at runtime shrinks one side's task count but not the + * other's), rows with the same join key can land in different reduce tasks on each side, + * causing the merge join to silently miss matches. + */ +public class TestGenTezUtilsMergeJoinNumReducers { + + private static final int BIG_TABLE_NUM_REDUCERS = 5; + private static final int SMALL_TABLE_NUM_REDUCERS = 20; + + private GenTezProcContext ctx; + private TezWork tezWork; + + private ReduceSinkOperator rsBigTable; + private GroupByOperator groupBy; + + private ReduceSinkOperator rsSmallTable; + private TezDummyStoreOperator dummyStore; + + @Before + public void setUp() throws Exception { + HiveConf conf = new HiveConfForTest(getClass()); + SessionState.start(conf); + + ParseContext pctx = new ParseContext(); + pctx.setContext(new Context(conf)); + + ctx = new GenTezProcContext(conf, pctx, Collections.EMPTY_LIST, + new ArrayList>(), Collections.EMPTY_SET, Collections.EMPTY_SET); + + tezWork = new TezWork("test"); + + CompilationOpContext cCtx = new CompilationOpContext(); + + rsBigTable = newReduceSink(cCtx, BIG_TABLE_NUM_REDUCERS); + groupBy = new GroupByOperator(cCtx); + groupBy.setConf(new GroupByDesc()); + groupBy.getConf().setKeys(new ArrayList<>()); + wireParentChild(rsBigTable, groupBy); + + rsSmallTable = newReduceSink(cCtx, SMALL_TABLE_NUM_REDUCERS); + dummyStore = new TezDummyStoreOperator(cCtx); + wireParentChild(rsSmallTable, dummyStore); + + CommonMergeJoinOperator mergeJoin = new CommonMergeJoinOperator(cCtx); + groupBy.setChildOperators(newList(mergeJoin)); + dummyStore.setChildOperators(newList(mergeJoin)); + mergeJoin.setParentOperators(newList(groupBy, dummyStore)); + } + + /** + * Both ReduceSinkOperators feed the same CommonMergeJoinOperator (one via a GroupBy, one via + * a DummyStore), but start out with different numReducers. Today, createReduceWork does not + * reconcile them: this assertion fails against unfixed code (5 != 20) and should pass once + * GenTezUtils normalizes numReducers across merge-join siblings. + */ + @Test + public void testSiblingReduceSinksFeedingMergeJoinGetSameNumReducers() throws Exception { + ReduceWork rwBigTable = createReduceWorkFor(rsBigTable, groupBy); + ReduceWork rwSmallTable = createReduceWorkFor(rsSmallTable, dummyStore); + + assertEquals("both sides of the merge join must partition into the same number of tasks", + rwBigTable.getNumReduceTasks(), rwSmallTable.getNumReduceTasks()); + } + + private ReduceWork createReduceWorkFor(ReduceSinkOperator rs, Operator root) throws Exception { + MapWork preceedingWork = new MapWork("Map for " + rs); + tezWork.add(preceedingWork); + ctx.preceedingWork = preceedingWork; + ctx.parentOfRoot = rs; + return GenTezUtils.createReduceWork(ctx, root, tezWork); + } + + private static ReduceSinkOperator newReduceSink(CompilationOpContext cCtx, int numReducers) { + ReduceSinkOperator rs = new ReduceSinkOperator(cCtx); + rs.setConf(new ReduceSinkDesc()); + TableDesc tableDesc = new TableDesc(); + tableDesc.setProperties(new Properties()); + rs.getConf().setKeySerializeInfo(tableDesc); + rs.getConf().setValueSerializeInfo(tableDesc); + rs.getConf().setNumReducers(numReducers); + return rs; + } + + private static void wireParentChild( + Operator parent, Operator child) { + parent.setChildOperators(newList(child)); + child.setParentOperators(newList(parent)); + } + + @SafeVarargs + private static ArrayList newList(T... items) { + ArrayList list = new ArrayList<>(); + Collections.addAll(list, items); + return list; + } +} diff --git a/ql/src/test/queries/clientpositive/tez_dummystore_reduce_side_reuse.q b/ql/src/test/queries/clientpositive/tez_dummystore_reduce_side_reuse.q new file mode 100644 index 000000000000..ce4f0beab157 --- /dev/null +++ b/ql/src/test/queries/clientpositive/tez_dummystore_reduce_side_reuse.q @@ -0,0 +1,62 @@ +set hive.mapred.mode=nonstrict; +set hive.auto.convert.join=false; +set hive.auto.convert.sortmerge.join=true; +set hive.auto.convert.sortmerge.join.reduce.side=true; +set hive.auto.convert.sortmerge.join.to.mapjoin=false; +set hive.optimize.bucketmapjoin=true; +-- Force multiple reduce tasks for the SAME vertex, and inflate each task's requested +-- container memory so only a couple of containers fit in the (small, test-sized) +-- NodeManager at once. This forces the Tez AM to sequentially REUSE the same container +-- for multiple tasks of the Reducer vertex within this single query - the ObjectCache +-- entry (keyed by vertex name, scoped per-query) then hands the second task the SAME +-- already-initialized operator instances the first task used, exercising +-- TezDummyStoreOperator's wiring cleanup on closeOp(). +set mapred.reduce.tasks=8; +set hive.exec.reducers.max=8; +set hive.tez.container.size=2048; +set hive.tez.java.opts=-Xmx1600m; + +CREATE TABLE dummystore_src (id int, val string, sort_col int); + +INSERT INTO dummystore_src VALUES + (1,'A',3),(1,'A',1),(1,'A',2), + (2,'B',1),(2,'B',3),(2,'B',2), + (3,'C',2),(3,'C',1), + (4,'D',1),(4,'D',3),(4,'D',2), + (5,'E',1),(5,'E',3),(5,'E',2), + (6,'F',2),(6,'F',1), + (7,'G',1),(7,'G',3),(7,'G',2), + (8,'H',1),(8,'H',3),(8,'H',2); + +-- Reduce-side merge join with PTF (ROW_NUMBER) + LEFT OUTER JOIN: one side aggregates via +-- GROUP BY, the other picks the "latest" row per id via ROW_NUMBER. Both sides shuffle into +-- a Reducer vertex containing a DummyStore + CommonMergeJoinOperator pipeline. With 8 +-- reducer tasks contending for a couple of containers, some containers must sequentially +-- process more than one task of this vertex, exercising TezDummyStoreOperator's wiring +-- cleanup on closeOp() and ReduceRecordProcessor's close ordering for the last group in +-- each task's merge-scan order. +EXPLAIN +SELECT b.id, b.max_val, e.enrch_val +FROM + (SELECT id, max(val) AS max_val FROM dummystore_src GROUP BY id) b + LEFT JOIN + (SELECT id, val AS enrch_val + FROM (SELECT id, val, sort_col, + ROW_NUMBER() OVER (PARTITION BY id ORDER BY sort_col DESC) AS rn + FROM dummystore_src) t + WHERE rn = 1) e + ON b.id = e.id; + +-- SORT_QUERY_RESULTS +SELECT b.id, b.max_val, e.enrch_val +FROM + (SELECT id, max(val) AS max_val FROM dummystore_src GROUP BY id) b + LEFT JOIN + (SELECT id, val AS enrch_val + FROM (SELECT id, val, sort_col, + ROW_NUMBER() OVER (PARTITION BY id ORDER BY sort_col DESC) AS rn + FROM dummystore_src) t + WHERE rn = 1) e + ON b.id = e.id; + +DROP TABLE dummystore_src; diff --git a/ql/src/test/results/clientpositive/tez/tez_dummystore_reduce_side_reuse.q.out b/ql/src/test/results/clientpositive/tez/tez_dummystore_reduce_side_reuse.q.out new file mode 100644 index 000000000000..5a1ff43c79ba --- /dev/null +++ b/ql/src/test/results/clientpositive/tez/tez_dummystore_reduce_side_reuse.q.out @@ -0,0 +1,150 @@ +PREHOOK: query: CREATE TABLE dummystore_src (id int, val string, sort_col int) +PREHOOK: type: CREATETABLE +PREHOOK: Output: database:default +PREHOOK: Output: default@dummystore_src +POSTHOOK: query: CREATE TABLE dummystore_src (id int, val string, sort_col int) +POSTHOOK: type: CREATETABLE +POSTHOOK: Output: database:default +POSTHOOK: Output: default@dummystore_src +PREHOOK: query: INSERT INTO dummystore_src VALUES + (1,'A',3),(1,'A',1),(1,'A',2), + (2,'B',1),(2,'B',3),(2,'B',2), + (3,'C',2),(3,'C',1), + (4,'D',1),(4,'D',3),(4,'D',2), + (5,'E',1),(5,'E',3),(5,'E',2), + (6,'F',2),(6,'F',1), + (7,'G',1),(7,'G',3),(7,'G',2), + (8,'H',1),(8,'H',3),(8,'H',2) +PREHOOK: type: QUERY +PREHOOK: Input: _dummy_database@_dummy_table +PREHOOK: Output: default@dummystore_src +POSTHOOK: query: INSERT INTO dummystore_src VALUES + (1,'A',3),(1,'A',1),(1,'A',2), + (2,'B',1),(2,'B',3),(2,'B',2), + (3,'C',2),(3,'C',1), + (4,'D',1),(4,'D',3),(4,'D',2), + (5,'E',1),(5,'E',3),(5,'E',2), + (6,'F',2),(6,'F',1), + (7,'G',1),(7,'G',3),(7,'G',2), + (8,'H',1),(8,'H',3),(8,'H',2) +POSTHOOK: type: QUERY +POSTHOOK: Input: _dummy_database@_dummy_table +POSTHOOK: Output: default@dummystore_src +POSTHOOK: Lineage: dummystore_src.id SCRIPT [] +POSTHOOK: Lineage: dummystore_src.sort_col SCRIPT [] +POSTHOOK: Lineage: dummystore_src.val SCRIPT [] +PREHOOK: query: EXPLAIN +SELECT b.id, b.max_val, e.enrch_val +FROM + (SELECT id, max(val) AS max_val FROM dummystore_src GROUP BY id) b + LEFT JOIN + (SELECT id, val AS enrch_val + FROM (SELECT id, val, sort_col, + ROW_NUMBER() OVER (PARTITION BY id ORDER BY sort_col DESC) AS rn + FROM dummystore_src) t + WHERE rn = 1) e + ON b.id = e.id +PREHOOK: type: QUERY +PREHOOK: Input: default@dummystore_src +PREHOOK: Output: hdfs://### HDFS PATH ### +POSTHOOK: query: EXPLAIN +SELECT b.id, b.max_val, e.enrch_val +FROM + (SELECT id, max(val) AS max_val FROM dummystore_src GROUP BY id) b + LEFT JOIN + (SELECT id, val AS enrch_val + FROM (SELECT id, val, sort_col, + ROW_NUMBER() OVER (PARTITION BY id ORDER BY sort_col DESC) AS rn + FROM dummystore_src) t + WHERE rn = 1) e + ON b.id = e.id +POSTHOOK: type: QUERY +POSTHOOK: Input: default@dummystore_src +POSTHOOK: Output: hdfs://### HDFS PATH ### +Plan optimized by CBO. + +Vertex dependency in root stage +Reducer 2 <- Map 1 (SIMPLE_EDGE), Map 3 (SIMPLE_EDGE) + +Stage-0 + Fetch Operator + limit:-1 + Stage-1 + Reducer 2 + File Output Operator [FS_19] + Select Operator [SEL_18] (rows=11 width=273) + Output:["_col0","_col1","_col2"] + Merge Join Operator [MERGEJOIN_23] (rows=11 width=273) + Conds:GBY_4._col0=DUMMY_STORE_24._col0(Left Outer),Output:["_col0","_col1","_col3"] + <-Dummy Store [DUMMY_STORE_24] + Select Operator [SEL_11] (rows=11 width=89) + Output:["_col0","_col1"] + Filter Operator [FIL_21] (rows=11 width=93) + predicate:(ROW_NUMBER_window_0 = 1) + PTF Operator [PTF_10] (rows=22 width=93) + Function definitions:[{},{"name:":"windowingtablefunction","order by:":"_col2 DESC NULLS FIRST","partition by:":"_col0"}] + Select Operator [SEL_9] (rows=22 width=93) + Output:["_col0","_col1","_col2"] + <-Group By Operator [GBY_4] (rows=8 width=188) + Output:["_col0","_col1"],aggregations:["max(VALUE._col0)"],keys:KEY._col0 + <-Map 1 [SIMPLE_EDGE] vectorized + SHUFFLE [RS_27] + PartitionCols:_col0 + Group By Operator [GBY_26] (rows=8 width=188) + Output:["_col0","_col1"],aggregations:["max(val)"],keys:id + Select Operator [SEL_25] (rows=22 width=89) + Output:["id","val"] + TableScan [TS_0] (rows=22 width=89) + default@dummystore_src,dummystore_src,Tbl:COMPLETE,Col:COMPLETE,Output:["id","val"] + <-Map 3 [SIMPLE_EDGE] vectorized + SHUFFLE [RS_29] + PartitionCols:id + Filter Operator [FIL_28] (rows=22 width=93) + predicate:id is not null + TableScan [TS_6] (rows=22 width=93) + default@dummystore_src,dummystore_src,Tbl:COMPLETE,Col:COMPLETE,Output:["id","val","sort_col"] + +PREHOOK: query: SELECT b.id, b.max_val, e.enrch_val +FROM + (SELECT id, max(val) AS max_val FROM dummystore_src GROUP BY id) b + LEFT JOIN + (SELECT id, val AS enrch_val + FROM (SELECT id, val, sort_col, + ROW_NUMBER() OVER (PARTITION BY id ORDER BY sort_col DESC) AS rn + FROM dummystore_src) t + WHERE rn = 1) e + ON b.id = e.id +PREHOOK: type: QUERY +PREHOOK: Input: default@dummystore_src +PREHOOK: Output: hdfs://### HDFS PATH ### +POSTHOOK: query: SELECT b.id, b.max_val, e.enrch_val +FROM + (SELECT id, max(val) AS max_val FROM dummystore_src GROUP BY id) b + LEFT JOIN + (SELECT id, val AS enrch_val + FROM (SELECT id, val, sort_col, + ROW_NUMBER() OVER (PARTITION BY id ORDER BY sort_col DESC) AS rn + FROM dummystore_src) t + WHERE rn = 1) e + ON b.id = e.id +POSTHOOK: type: QUERY +POSTHOOK: Input: default@dummystore_src +POSTHOOK: Output: hdfs://### HDFS PATH ### +1 A A +2 B B +3 C C +4 D D +5 E E +6 F F +7 G G +8 H H +PREHOOK: query: DROP TABLE dummystore_src +PREHOOK: type: DROPTABLE +PREHOOK: Input: default@dummystore_src +PREHOOK: Output: database:default +PREHOOK: Output: default@dummystore_src +POSTHOOK: query: DROP TABLE dummystore_src +POSTHOOK: type: DROPTABLE +POSTHOOK: Input: default@dummystore_src +POSTHOOK: Output: database:default +POSTHOOK: Output: default@dummystore_src