ryukobayashi opened a new pull request, #6646:
URL: https://github.com/apache/hive/pull/6646
### 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 `ReduceSinkOperator`s 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 h
old, 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 `ReduceSinkOperator`s 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 `ReduceSinkOperator`s, 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.
--
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]
