Copilot commented on code in PR #6646:
URL: https://github.com/apache/hive/pull/6646#discussion_r4296455244
##########
ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezUtils.java:
##########
@@ -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.
+ *
+ * <p>Two guards keep this from touching cases it shouldn't:
+ * <ul>
+ * <li>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.</li>
+ * <li>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.</li>
+ * </ul>
+ */
+ 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<ReduceSinkOperator> 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<Operator<?>> 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);
Review Comment:
Stop the downstream scan at another `ReduceSinkOperator`. For `RS0 → GroupBy
→ RS1 → GroupBy → merge join`, this walk identifies RS0 as a merge-join input
even though RS1 introduces another shuffle boundary. Normalization collects RS1
and its siblings but returns true for RS0, so `createReduceWork` disables
adaptive parallelism for the unrelated upstream stage. That stage can retain an
overestimated reducer count instead of allowing Tez to shrink it. Restrict the
search to the current reducer pipeline.
##########
ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezUtils.java:
##########
@@ -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.
+ *
+ * <p>Two guards keep this from touching cases it shouldn't:
+ * <ul>
+ * <li>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.</li>
+ * <li>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.</li>
+ * </ul>
+ */
+ 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<ReduceSinkOperator> allRS = new ArrayList<>();
+ for (Operator<?> parent : mergeJoin.getParentOperators()) {
+ ReduceSinkOperator rsAnc = findUpstreamRS(parent, 8);
+ if (rsAnc != null) {
+ allRS.add(rsAnc);
+ }
Review Comment:
If a sibling's ReduceSink is more than eight parent links away,
`findUpstreamRS` returns null and silently excludes that input. Normalization
then clears `UNIFORM` only on the discovered inputs. For partitioned keys,
`ReduceSinkOperator` selects `computeHashCode` without this trait and
`computeMurmurHash` with it, so previously compatible inputs can use different
hashes and silently lose join matches—even if their reducer counts were equal.
Leave all siblings untouched when any lineage cannot be resolved, and add a
regression test with a longer sibling branch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
