From f831c174a72dc5623cdff6721aa736fbd44320b4 Mon Sep 17 00:00:00 2001 From: Ryu Kobayashi Date: Wed, 29 Jul 2026 17:01:31 +0900 Subject: [PATCH 1/5] HIVE-29702: Fix stale DummyStore wiring causing reduce-side merge join failures on Tez container reuse --- .../resources/testconfiguration.properties | 1 + .../ql/exec/tez/ReduceRecordProcessor.java | 34 ++-- .../tez_dummystore_reduce_side_reuse.q | 62 ++++++++ .../tez_dummystore_reduce_side_reuse.q.out | 150 ++++++++++++++++++ 4 files changed, 238 insertions(+), 9 deletions(-) create mode 100644 ql/src/test/queries/clientpositive/tez_dummystore_reduce_side_reuse.q create mode 100644 ql/src/test/results/clientpositive/tez/tez_dummystore_reduce_side_reuse.q.out diff --git a/itests/src/test/resources/testconfiguration.properties b/itests/src/test/resources/testconfiguration.properties index efdbebcf9b54..8df89c6bb690 100644 --- a/itests/src/test/resources/testconfiguration.properties +++ b/itests/src/test/resources/testconfiguration.properties @@ -35,6 +35,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/tez/ReduceRecordProcessor.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java index 96714f29dc6f..6bf9adf5aedd 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 @@ -324,16 +324,29 @@ public void abort() { */ private List getShuffleInputs(Map inputs) throws Exception { // the reduce plan inputs have tags, add all inputs that have tags - Map tagToinput = reduceWork.getTagToInput(); ArrayList shuffleInputs = new ArrayList(); - for (String inpStr : tagToinput.values()) { + startTaggedInputs(reduceWork, inputs, shuffleInputs); + // Merge-join siblings (e.g. the DummyStore side of a reduce-side merge join) are separate + // ReduceWorks with their own tagged inputs. init() later reads from all of them via + // tagToReducerMap, so they must be started here as well or getReader() throws + // "Must start input before invoking this method". + if (mergeWorkList != null) { + for (BaseWork mergeWork : mergeWorkList) { + startTaggedInputs((ReduceWork) mergeWork, inputs, shuffleInputs); + } + } + return shuffleInputs; + } + + private void startTaggedInputs(ReduceWork redWork, Map inputs, + List shuffleInputs) throws Exception { + for (String inpStr : redWork.getTagToInput().values()) { if (inputs.get(inpStr) == null) { throw new AssertionError("Cound not find input: " + inpStr); } inputs.get(inpStr).start(); shuffleInputs.add(inputs.get(inpStr)); } - return shuffleInputs; } @Override @@ -391,12 +404,15 @@ void close() { private DummyStoreOperator getJoinParentOp(Operator mergeReduceOp) { for (Operator childOp : mergeReduceOp.getChildOperators()) { - if ((childOp.getChildOperators() == null) || (childOp.getChildOperators().isEmpty())) { - if (childOp instanceof DummyStoreOperator) { - return (DummyStoreOperator) childOp; - } else { - throw new IllegalStateException("Was expecting dummy store operator but found: " + childOp); - } + // Check the type first: on a reused (cached) operator tree, a prior attempt's + // CommonMergeJoinOperator.initializeLocalWork() may have already appended itself to this + // DummyStoreOperator's children, so childOp.getChildOperators() is no longer empty. That + // must not be mistaken for "not a leaf, keep recursing" - it is still the DummyStore we + // want, just with wiring left over from the previous attempt. + if (childOp instanceof DummyStoreOperator) { + return (DummyStoreOperator) childOp; + } else if ((childOp.getChildOperators() == null) || (childOp.getChildOperators().isEmpty())) { + throw new IllegalStateException("Was expecting dummy store operator but found: " + childOp); } else { return getJoinParentOp(childOp); } 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 From df3c31011ea8e384cc4ff072dcc85b14c062d53a Mon Sep 17 00:00:00 2001 From: Ryu Kobayashi Date: Wed, 29 Jul 2026 17:02:23 +0900 Subject: [PATCH 2/5] HIVE-29702: Normalize reducer count across ReduceSinkOperators feeding the same merge join --- .../hadoop/hive/ql/parse/GenTezUtils.java | 145 +++++++++++- .../TestGenTezUtilsMergeJoinNumReducers.java | 222 ++++++++++++++++++ .../clientpositive/llap/sharedwork.q.out | 4 +- 3 files changed, 368 insertions(+), 3 deletions(-) create mode 100644 ql/src/test/org/apache/hadoop/hive/ql/parse/TestGenTezUtilsMergeJoinNumReducers.java 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..8ff1d951a70a 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; @@ -42,6 +43,7 @@ import org.apache.hadoop.hive.ql.exec.ReduceSinkOperator; import org.apache.hadoop.hive.ql.exec.SerializationUtilities; import org.apache.hadoop.hive.ql.exec.TableScanOperator; +import org.apache.hadoop.hive.ql.exec.TezDummyStoreOperator; import org.apache.hadoop.hive.ql.exec.UnionOperator; import org.apache.hadoop.hive.ql.exec.Utilities; import org.apache.hadoop.hive.ql.hooks.WriteEntity; @@ -110,6 +112,15 @@ public static ReduceWork createReduceWork( assert context.parentOfRoot instanceof ReduceSinkOperator; ReduceSinkOperator reduceSink = (ReduceSinkOperator) context.parentOfRoot; + // If this RS feeds into a CommonMergeJoinOp together with a TezDummyStoreOperator sibling + // (possibly via intermediate ops like GroupBy), 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 +138,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 +199,137 @@ public static ReduceWork createReduceWork( return reduceWork; } + /** + * If the given RS leads (through child ops) to a CommonMergeJoinOperator that also has a + * {@link TezDummyStoreOperator} sibling parent, 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. + * + *

Two guards keep this from touching cases it shouldn't: + *

    + *
  • The DummyStore check: it marks the merge join's sibling as living in its own, + * separately auto-parallelized Tez vertex (see ReduceRecordProcessor's mergeWorkList). A + * merge join whose parents are just multiple tags of this same ReduceWork (no DummyStore + * involved) already shares this vertex's single parallelism setting, so there is no + * independent-vertex mismatch to guard against.
    • + *
  • The AUTOPARALLEL/UNIFORM check: genuine bucketed SMB joins (e.g. a 2-bucket table + * joined with a 3-bucket table) legitimately have different, FIXED numReducers per side, + * each anchored to that side's own bucket layout on disk. Only AUTOPARALLEL/UNIFORM traits + * mean a side's numReducers is a mutable suggestion that ShuffleVertexManager can + * independently shrink at runtime; normalizing FIXED siblings would break the + * bucket-to-reducer correspondence those joins rely on.
    • + *
+ */ + private static boolean normalizeMergeJoinReducers(ReduceSinkOperator reduceSink) { + CommonMergeJoinOperator mergeJoin = findDownstreamMergeJoin(reduceSink, 8); + if (mergeJoin == null) { + return false; + } + boolean hasDummyStoreSibling = false; + for (Operator parent : mergeJoin.getParentOperators()) { + if (parent instanceof TezDummyStoreOperator) { + hasDummyStoreSibling = true; + break; + } + } + if (!hasDummyStoreSibling) { + 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; + } + boolean anySiblingCanDrift = false; + for (ReduceSinkOperator rs : allRS) { + if (rs.getConf().getReducerTraits().contains(AUTOPARALLEL) + || rs.getConf().getReducerTraits().contains(UNIFORM)) { + anySiblingCanDrift = true; + break; + } + } + if (!anySiblingCanDrift) { + 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; + } + // A branch (e.g. shared work reused across multiple consumers, as in sharedwork.q) means + // we can no longer be sure which downstream path is the one relevant to this RS - bail + // rather than guess, since guessing wrong can misidentify an unrelated merge join and + // normalize reducers that have nothing to do with it. + if (children.size() > 1) { + return null; + } + 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; + } + // Same reasoning as findDownstreamMergeJoin: a branch point (e.g. a join or union + // upstream) means we cannot reliably pick the one lineage that actually feeds this + // merge join, so bail instead of guessing. + if (parents.size() > 1) { + return null; + } + 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/parse/TestGenTezUtilsMergeJoinNumReducers.java b/ql/src/test/org/apache/hadoop/hive/ql/parse/TestGenTezUtilsMergeJoinNumReducers.java new file mode 100644 index 000000000000..97fc92a7aefc --- /dev/null +++ b/ql/src/test/org/apache/hadoop/hive/ql/parse/TestGenTezUtilsMergeJoinNumReducers.java @@ -0,0 +1,222 @@ +/* + * 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 static org.junit.Assert.assertTrue; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.EnumSet; +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); + // AUTOPARALLEL is what makes the mismatch a real bug: it tells ShuffleVertexManager this + // side's task count is free to shrink at runtime, independently of the other side's. + rsSmallTable.getConf().setReducerTraits(EnumSet.of(ReduceSinkDesc.ReducerTraits.AUTOPARALLEL)); + 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, and the DummyStore side carries + * AUTOPARALLEL - i.e. its task count is free to drift at runtime. 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()); + } + + /** + * When a CommonMergeJoinOperator's parents are plain ReduceSinkOperators with no + * TezDummyStoreOperator sibling, both sides already run as tags of the same single Tez + * vertex and share its one parallelism setting. Normalization must be a no-op here: it must + * not touch numReducers or strip the AUTOPARALLEL trait, since there is no independent-vertex + * mismatch to protect against. + */ + @Test + public void testMergeJoinWithoutDummyStoreSiblingIsNotNormalized() throws Exception { + CompilationOpContext cCtx = new CompilationOpContext(); + ReduceSinkOperator rsLeft = newReduceSink(cCtx, BIG_TABLE_NUM_REDUCERS); + rsLeft.getConf().setReducerTraits(EnumSet.of(ReduceSinkDesc.ReducerTraits.AUTOPARALLEL)); + ReduceSinkOperator rsRight = newReduceSink(cCtx, SMALL_TABLE_NUM_REDUCERS); + rsRight.getConf().setReducerTraits(EnumSet.of(ReduceSinkDesc.ReducerTraits.AUTOPARALLEL)); + + CommonMergeJoinOperator mergeJoin = new CommonMergeJoinOperator(cCtx); + rsLeft.setChildOperators(newList(mergeJoin)); + rsRight.setChildOperators(newList(mergeJoin)); + mergeJoin.setParentOperators(newList(rsLeft, rsRight)); + + ReduceWork rwLeft = createReduceWorkFor(rsLeft, mergeJoin); + ReduceWork rwRight = createReduceWorkFor(rsRight, mergeJoin); + + assertEquals("numReducers must be untouched when there is no DummyStore sibling", + (long) BIG_TABLE_NUM_REDUCERS, (long) rwLeft.getNumReduceTasks()); + assertEquals("numReducers must be untouched when there is no DummyStore sibling", + (long) SMALL_TABLE_NUM_REDUCERS, (long) rwRight.getNumReduceTasks()); + assertTrue("AUTOPARALLEL must survive when there is no DummyStore sibling to protect against", + rsLeft.getConf().getReducerTraits().contains(ReduceSinkDesc.ReducerTraits.AUTOPARALLEL)); + assertTrue("AUTOPARALLEL must survive when there is no DummyStore sibling to protect against", + rsRight.getConf().getReducerTraits().contains(ReduceSinkDesc.ReducerTraits.AUTOPARALLEL)); + } + + /** + * A DummyStore sibling alone isn't sufficient to normalize: genuine bucketed SMB joins (e.g. + * a 2-bucket table joined with a 3-bucket table) legitimately have different, FIXED + * numReducers per side, each anchored to that side's own on-disk bucket layout. Neither side + * carries AUTOPARALLEL/UNIFORM, so there is no runtime-drift risk - forcing them to match + * would break the bucket-to-reducer correspondence the join relies on. + */ + @Test + public void testMergeJoinWithFixedDifferentBucketCountsIsNotNormalized() throws Exception { + CompilationOpContext cCtx = new CompilationOpContext(); + ReduceSinkOperator rsTwoBuckets = newReduceSink(cCtx, 2); + rsTwoBuckets.getConf().setReducerTraits(EnumSet.of(ReduceSinkDesc.ReducerTraits.FIXED)); + GroupByOperator bigTableSide = new GroupByOperator(cCtx); + bigTableSide.setConf(new GroupByDesc()); + bigTableSide.getConf().setKeys(new ArrayList<>()); + wireParentChild(rsTwoBuckets, bigTableSide); + + ReduceSinkOperator rsThreeBuckets = newReduceSink(cCtx, 3); + rsThreeBuckets.getConf().setReducerTraits(EnumSet.of(ReduceSinkDesc.ReducerTraits.FIXED)); + TezDummyStoreOperator smallTableSide = new TezDummyStoreOperator(cCtx); + wireParentChild(rsThreeBuckets, smallTableSide); + + CommonMergeJoinOperator mergeJoin = new CommonMergeJoinOperator(cCtx); + bigTableSide.setChildOperators(newList(mergeJoin)); + smallTableSide.setChildOperators(newList(mergeJoin)); + mergeJoin.setParentOperators(newList(bigTableSide, smallTableSide)); + + ReduceWork rwTwoBuckets = createReduceWorkFor(rsTwoBuckets, bigTableSide); + ReduceWork rwThreeBuckets = createReduceWorkFor(rsThreeBuckets, smallTableSide); + + assertEquals("a FIXED bucket count must not be overwritten by a sibling's count", + 2L, (long) rwTwoBuckets.getNumReduceTasks()); + assertEquals("a FIXED bucket count must not be overwritten by a sibling's count", + 3L, (long) rwThreeBuckets.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/results/clientpositive/llap/sharedwork.q.out b/ql/src/test/results/clientpositive/llap/sharedwork.q.out index e94f9826ed78..f01d6593a362 100644 --- a/ql/src/test/results/clientpositive/llap/sharedwork.q.out +++ b/ql/src/test/results/clientpositive/llap/sharedwork.q.out @@ -655,7 +655,7 @@ STAGE PLANS: Map-reduce partition columns: _col0 (type: int) Statistics: Num rows: 21 Data size: 84 Basic stats: COMPLETE Column stats: COMPLETE tag: -1 - auto parallelism: true + auto parallelism: false Reduce Output Operator bucketingVersion: 2 key expressions: _col0 (type: int) @@ -775,7 +775,7 @@ STAGE PLANS: Statistics: Num rows: 21 Data size: 420 Basic stats: COMPLETE Column stats: COMPLETE tag: -1 value expressions: _col1 (type: bigint), _col2 (type: bigint) - auto parallelism: true + auto parallelism: false Execution mode: vectorized, llap LLAP IO: all inputs Path -> Alias: From 8ee2dd37068b8e3e860b38da6b6991943aec9c01 Mon Sep 17 00:00:00 2001 From: Ryu Kobayashi Date: Thu, 30 Jul 2026 12:11:12 +0900 Subject: [PATCH 3/5] HIVE-29702: Trigger CI rebuild From 3494b250bc99219ccde56bb215e96b29b4beb3e8 Mon Sep 17 00:00:00 2001 From: Ryu Kobayashi Date: Thu, 30 Jul 2026 15:19:27 +0900 Subject: [PATCH 4/5] HIVE-29702: Trigger CI rebuild From 499749e1f5f3f30a778b46ae6299ac50dc390baa Mon Sep 17 00:00:00 2001 From: Ryu Kobayashi Date: Thu, 30 Jul 2026 15:21:10 +0900 Subject: [PATCH 5/5] HIVE-29702: Trigger CI rebuild