HIVE-29702: Reduce-side merge join can produce NULL values or fail with IllegalStateException under Tez container reuse

[ryukobayashi]

What changes were proposed in this pull request?

This PR fixes two related defects in Hive's reduce-side merge join (SMB / DummyStore + CommonMergeJoinOperator pipeline) on Tez:

1. ReduceRecordProcessor had two bugs that surface specifically when Tez reuses a container across multiple tasks of the same reducer vertex (Hive's per-query ObjectCache, keyed by vertex name, hands the new task the same already-initialized operator instances the previous task used):

   • getShuffleInputs() only called start() on the main ReduceWork's tagged inputs. The mergeWorkList siblings (e.g. the DummyStoreOperator side of the merge join) have their own tagged inputs that init() also reads from via tagToReducerMap, but those were never started, so getReader() throws "Must start input before invoking this method".
   • getJoinParentOp() located the DummyStoreOperator by treating "childOperators is empty" as the leaf marker. On a reused operator tree, a prior attempt's CommonMergeJoinOperator#initializeLocalWork may have already appended itself to that DummyStoreOperator's children, so the leaf check no longer holds and the walk recurses into the wrong subtree, throwing IllegalStateException: Was expecting dummy store operator but found: ....
   • close() closed the main reducer before the mergeWorkList (DummyStore chain). The DummyStoreOperator has no rows of its own to close out, so closing its side first is what lets Operator#allInitializedParentsAreClosed() see it as done, and the main reducer's close() — now closing last — is what finally lets CommonMergeJoinOperator emit its last join group.

2. GenTezUtils#createReduceWork sets each ReduceWork's numReduceTasks purely from that branch's own ReduceSinkOperator, with no cross-check against sibling ReduceSinkOperators that feed the same downstream CommonMergeJoinOperator through a separate, independently auto-parallelized Tez vertex (the DummyStoreOperator/mergeWorkList case above). SetReducerParallelism assigns each ReduceSinkOperator's reducer count/traits independently, so two such siblings can end up with different numReducers, and Tez's dynamic auto-parallelism (ShuffleVertexManager) can shrink one side's task count at runtime independently of the other's.

   A new normalizeMergeJoinReducers helper walks from a ReduceSinkOperator downstream to find a CommonMergeJoinOperator, and only acts when: (a) one of the merge join's parents is a TezDummyStoreOperator — this is what marks the sibling as living in its own, separately auto-parallelized vertex; a merge join whose parents are just multiple tags of the same ReduceWork already shares that single vertex's parallelism, so there's nothing to reconcile — and (b) at least one sibling carries the AUTOPARALLEL/UNIFORM trait — i.e. its numReducers is a mutable suggestion Tez can shrink at runtime. 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; without guard (b) this normalization would overwrite one side's bucket count with the other's and break the bucket-to-reducer correspondence those joins rely on. When both guards hold, all siblings are normalized to the same (maximum) numReducers with FIXED traits, so Tez's auto-parallelism cannot independently shrink one side's task count at runtime.

Why are the changes needed?

• (1) causes a task failure (IllegalStateException) when Tez reuses a container across multiple tasks of the same reducer vertex.
• (2) causes rows with the same join key to be routed to different partitions on each side of the join (since the two sides no longer agree on the same partition count/hash strategy), silently producing incomplete join results (e.g. missing matches or NULLs in a LEFT JOIN) with no error reported to the user.

Does this PR introduce any user-facing change?

No

How was this patch tested?

• Added tez_dummystore_reduce_side_reuse.q (TestMiniTezCliDriver) for (1): forces Tez to reuse a container across multiple tasks of the same reducer vertex. Reproduces the exact IllegalStateException against the pre-fix code; passes cleanly after the fix.
• Added TestGenTezUtilsMergeJoinNumReducers (3 unit tests) for (2):
  • Two synthetic ReduceSinkOperators feeding the same merge join through a GroupByOperator and a TezDummyStoreOperator respectively, with different numReducers and AUTOPARALLEL on the DummyStore side: fails against the pre-fix code (mismatched numReduceTasks), passes after the fix.
  • The same shape without a TezDummyStoreOperator sibling (both sides plain ReduceSinkOperators, i.e. tags of the same vertex): asserts normalization is a no-op, since there is no independent-vertex mismatch to guard against.
  • The same shape with a TezDummyStoreOperator sibling but both sides FIXED-trait with different numReducers (mirroring a 2-bucket/3-bucket SMB join): asserts normalization is a no-op, since forcing them to match would break the bucket-to-reducer correspondence the join relies on.
• Manually verified against the existing qtest suite (e.g. sample8, bucketmapjoin1, pcr, join9, smb_join_with_different_bucket_size, auto_sortmerge_join_9) to confirm the guards in (2) leave unrelated merge-join plans untouched.

[sonarqubecloud]

Quality Gate passed

Issues
7 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud