ryukobayashi opened a new pull request, #6586:
URL: https://github.com/apache/hive/pull/6586
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines:
https://cwiki.apache.org/confluence/display/Hive/HowToContribute
2. Ensure that you have created an issue on the Hive project JIRA:
https://issues.apache.org/jira/projects/HIVE/summary
3. Ensure you have added or run the appropriate tests for your PR:
4. If the PR is unfinished, add '[WIP]' in your PR title, e.g.,
'[WIP]HIVE-XXXXX: Your PR title ...'.
5. Be sure to keep the PR description updated to reflect all changes.
6. Please write your PR title to summarize what this PR proposes.
7. If possible, provide a concise example to reproduce the issue for a
faster review.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section
is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster
reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class
hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other
DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
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 `ReduceSinkOperator`s 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 `ReduceSinkOperator`s,
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?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
- (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?
<!--
Note that it means *any* user-facing change including all aspects such as
the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes
- provide the console output, description, screenshot and/or a reproducable
example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to
the released Hive versions or within the unreleased branches such as master.
If no, write 'No'.
-->
No
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some
test cases that check the changes thoroughly including negative and positive
cases if possible.
If it was tested in a way different from regular unit tests, please clarify
how you tested step by step, ideally copy and paste-able, so that other
reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why
it was difficult to add.
-->
- 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 `ReduceSinkOperator`s
(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 `ReduceWork`s 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.
--
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]
