[
https://issues.apache.org/jira/browse/HIVE-29702?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ryu Kobayashi updated HIVE-29702:
---------------------------------
Description:
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.
was:
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)
(to be filled in once the corresponding fix is finalized)
> 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
>
> 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)
