[
https://issues.apache.org/jira/browse/HIVE-29702?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Work on HIVE-29702 started by Ryu Kobayashi.
--------------------------------------------
> Reduce-side merge join can produce NULL values or fail with
> IllegalStateException under Tez container reuse
> -----------------------------------------------------------------------------------------------------------
>
> Key: HIVE-29702
> URL: https://issues.apache.org/jira/browse/HIVE-29702
> Project: Hive
> Issue Type: Bug
> Reporter: Ryu Kobayashi
> Assignee: Ryu Kobayashi
> Priority: Major
> Labels: pull-request-available
>
> h2. Summary
> Reduce-side merge join (a DummyStore + CommonMergeJoinOperator pipeline used
> for SMB / merge-join fallback) can misbehave when Tez reuses a container
> across multiple tasks of the same reducer vertex. Depending on timing, this
> manifests as either:
> (a) a task failure:
> {code}
> java.lang.IllegalStateException: Was expecting dummy store operator but
> found: FS[...]
> {code}
> (b) or, if no exception happens to be thrown, silent NULL values in the
> columns produced by the join, with no error reported to the user.
> h2. Root cause 1: stale operator-tree wiring on container reuse (crash)
> Hive's per-query {{ObjectCache}} (keyed by vertex name) can hand a task the
> same already-initialized {{DummyStoreOperator}} / {{CommonMergeJoinOperator}}
> instances that a previous task (sharing the same JVM/container) already used.
> {{TezDummyStoreOperator#closeOp}} does not remove itself from its child's
> {{parentOperators}}, nor clear its own {{childOperators}}, when closing. When
> the cached operator instances are reused by a subsequent task, the
> DummyStoreOperator still has a stale child reference left over from the
> previous task's execution. This causes
> {{ReduceRecordProcessor#getJoinParentOp}} to walk past the DummyStoreOperator
> into the wrong subtree, eventually reaching a non-DummyStoreOperator leaf and
> throwing {{IllegalStateException}}.
> Additionally, {{ReduceRecordProcessor#close()}} closes the main {{reducer}}
> before the {{mergeWorkList}} (DummyStore chain). Since
> {{DummyStoreOperator#closeOp}} is what removes the stale parent reference
> from the MergeJoin side, closing the main reducer first means
> {{CommonMergeJoinOperator#allInitializedParentsAreClosed()}} cannot detect
> all parents as closed in the correct order to flush the final join group.
> h2. Root cause 2: reducer count / partitioning mismatch (NULL values)
> {{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
> and trait independently: if a RS already has {{numReducers}} set (e.g.
> derived from a bucketed table's bucket count), it gets the {{FIXED}} trait
> and keeps that value; otherwise it computes its own byte-size-based estimate
> and gets the {{AUTOPARALLEL}} trait (plus {{UNIFORM}} when its key columns
> match its partition columns). Two sibling RS operators feeding the same merge
> join can therefore end up with independently-derived {{numReducers}}/traits
> with no reconciliation between them.
> Since the {{UNIFORM}} trait also affects which partitioning/hash strategy a
> {{ReduceWork}} uses ({{reduceWork.setUniformDistribution(...)}}), a
> divergence between siblings can cause rows with the same join key to be
> routed to different partitions on each side of the join even without
> necessarily differing reducer counts, silently producing incomplete join
> results.
> Confirmed via a unit test that 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}}: the two resulting
> {{ReduceWork}}s keep their original, mismatched {{numReduceTasks}} with no
> normalization.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
