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. TezDummyStoreOperator#closeOp did not remove itself from its child's parentOperators, nor clear its own childOperators, when closing. When Hive's per-query ObjectCache hands a subsequent task the same DummyStoreOperator/CommonMergeJoinOperator instances a previous task (sharing the same JVM/container) already used, the stale child reference causes ReduceRecordProcessor#getJoinParentOp to walk into the wrong operator subtree and throw IllegalStateException: Was expecting dummy store operator but found: ....

   ReduceRecordProcessor#close() also closed the main reducer before the mergeWorkList (DummyStore chain). Since DummyStoreOperator#closeOp is what removes the stale parent reference from the CommonMergeJoinOperator side, closing the main reducer first could prevent CommonMergeJoinOperator#allInitializedParentsAreClosed() from detecting all parents as closed in the correct order to flush the final 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. SetReducerParallelism assigns each ReduceSinkOperator's reducer count/traits independently, so two siblings feeding the same merge join can end up with different numReducers (and different partitioning behavior, since the UNIFORM trait also affects ReduceWork#setUniformDistribution), with no reconciliation between them.

A new normalizeMergeJoinReducers helper walks from a ReduceSinkOperator downstream to find a CommonMergeJoinOperator, then walks upstream from each of the merge join's parents to find their originating ReduceSinkOperators, and normalizes all of them to the same (maximum) numReducers, clearing AUTOPARALLEL/UNIFORM traits so Tez's dynamic 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 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 TestTezDummyStoreOperator (unit test) for (1): verifies closeOp correctly removes stale parent/child references. Fails against the pre-fix code (3/5 assertions), passes after the fix.
• 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 (unit test) for (2): drives GenTezUtils#createReduceWork directly for two synthetic ReduceSinkOperators (one feeding the merge join through a GroupByOperator, the other through a DummyStoreOperator) with different numReducers. Fails against the pre-fix code (the two resulting ReduceWorks keep their original, mismatched numReduceTasks), passes after the fix.
• No .q test was added for (2): the mismatch only manifests through Tez's runtime dynamic-parallelism shrinking (ShuffleVertexManager) at realistic data volumes, which could not be reproduced deterministically at qtest scale without either breaking the reduce-side-merge-join plan shape or triggering unrelated flush-ordering issues in synthetic query shapes tried. The unit test above exercises the exact defect directly and deterministically instead.

[sonarqubecloud]

Quality Gate passed

Issues
5 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