Zoltán Borók-Nagy created IMPALA-15656:
------------------------------------------
Summary: CTE producer hangs in LocalExchanger::Release() when a
consumer fails before opening the exchanger
Key: IMPALA-15656
URL: https://issues.apache.org/jira/browse/IMPALA-15656
Project: IMPALA
Issue Type: Bug
Components: Backend
Reporter: Zoltán Borók-Nagy
LocalExchanger::Release(), called from CTEProducerNode::Close(), waits until
every consumer has called CloseConsumer() (local-exchanger.cc:121 on master
1a26e1f778):
{code}
all_consumers_done_cv_.wait(l, [&]() {
return consumers_done_ == consumer_count_;
});
{code}
CTEConsumerNode looks up its exchanger only at the end of Open(), after
ExecNode::Open(), ClaimBufferReservation() and the expr opens. Close() calls
CloseConsumer() only if that lookup happened. So a consumer is never counted as
closed when:
* an earlier step of its Open() fails, e.g. the reservation claim under memory
pressure, or a cancellation;
* its fragment instance fails or is cancelled before the node is opened, e.g. a
join whose build side fails before the probe side is opened;
* its fragment instance never starts.
The producer's fragment instance then blocks in Release() forever. The query is
never torn down, and all row batches in the exchanger stay charged to it.
*Seen with* TPC-DS q14_1 at SF3000 on 8 executors (PLANNER=CALCITE,
CTE_THRESHOLD=1, MT_DOP=8, mem_limit=40g). The query fails with "Memory limit
exceeded" (the CTE buffer cannot spill, IMPALA-15313). Afterwards every
executor keeps the query's memory (about 23 GB on the coordinator). In one run
all 91 queries that followed failed with "Failed to get minimum memory
reservation" until the impalads were restarted. A stack dump of the coordinator
60 s after the failure shows 16 fragment instance threads of the failed query
in:
{noformat}
std::condition_variable::wait<LocalExchanger::Release()::<lambda()>>
impala::LocalExchanger::Release() local-exchanger.cc:121
impala::CTEProducerNode::Close() cte-producer-node.cc:135
impala::FragmentInstanceState::Close() fragment-instance-state.cc:572
{noformat}
No consumer threads of that query were alive.
*Repro idea (not tried):* run a query whose CTE is materialized
(PLANNER=CALCITE, CTE_THRESHOLD=1) with a DEBUG_ACTION that fails one
consumer's fragment before the CTE consumer is opened. The query should fail,
and /memz should keep listing it.
*Fix (prototyped):* once the query is cancelled, Release() waits only for the
consumers that called Open(). After the release, LocalExchanger::Open() returns
-1 and a late consumer runs without an exchanger; that is safe because the
query is already cancelled. With this change q14_1's memory was released 5-11 s
after the failure, and the queries after it ran normally.
An alternative is to register consumers in Prepare() and always release the
consumer slot in Close().
Prototype: https://gerrit.cloudera.org/#/c/25095/
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
