Xianqing He created IMPALA-15541:
------------------------------------
Summary: MT_DOP>0 HDFS scan hangs when a UNION child scan node is
never opened due to a LIMIT
Key: IMPALA-15541
URL: https://issues.apache.org/jira/browse/IMPALA-15541
Project: IMPALA
Issue Type: Bug
Components: Backend
Affects Versions: Impala 5.0.0, Impala 4.5.3
Reporter: Xianqing He
h3. Symptom
With MT_DOP > 0 and multiple fragment instances co-located on one backend, a
query with
UNION ALL whose scan node children are below GROUP BY ... LIMIT can hang
forever. One
fragment instance is blocked in ScanRangeSharedState::GetNextScanRange()
waiting for a
scan range submission that never happens:
{code:java}
#2 impala::ConditionVariable::Wait
#3 impala::ScanRangeSharedState::GetNextScanRange
(be/src/exec/hdfs-scan-node-base.cc)
#4 impala::HdfsScanNodeMt::GetNextScanRangeToRead
(be/src/exec/hdfs-scan-node-mt.cc)
#5 impala::HdfsScanNodeBase::StartNextScanRange
#6 impala::HdfsScanNodeMt::GetNext
{code}
h3. Root cause
* ScanRangeSharedState::remaining_scan_range_submissions_ is initialized to the
number
of fragment instances of the scan node. Each instance decrements it via the
scope-exit
trigger at the end of HdfsScanNodeBase::IssueInitialScanRanges().
* UnionNode::Open() opens its children lazily: it only opens the current child
(child(child_idx_)->Open()). If a LIMIT above the union stops pulling rows in
one
fragment instance before that instance opens the next union child, the child's
HdfsScanNodeMt is never opened. That instance therefore never calls
IssueInitialScanRanges() and never decrements the counter.
* Another fragment instance that did open the child (because it exhausted the
earlier
union children) drains the ranges it submitted itself and then blocks in
ScanRangeSharedState::GetNextScanRange() until the counter reaches 0. The
counter never
reaches 0, so the query hangs.
h3. Reproducer
Fragment instances must be co-located on one backend to share the scan range
queue, so
use a single-backend cluster.
{code:sql}
create table repro0 (k int) row format delimited fields terminated by ','
stored as textfile;
-- one INSERT writes one file; at least 10 distinct keys
insert into repro0 values
(0),(1),(2),(3),(4),(5),(6),(7),(8),(9),(10),(11),(12),(13),
(14),(15),(16),(17),(18),(19),(20),(21),(22),(23),(24);
create table repro1 (k int) row format delimited fields terminated by ','
stored as textfile;
-- two INSERTs write two files, which forces two co-located fragment instances
insert into repro1 values (1),(2),(3);
insert into repro1 values (4),(5),(6);
set mt_dop=2;
-- hangs: one instance reaches the aggregation limit from repro0 and never
opens the
-- repro1 scan; the other instance opens repro1 and waits forever for its
submission
select k from (select k from repro0 union all select k from repro1) t group by
1 limit 10;
{code}
Without the fix the query hangs (see stack above). With the fix it completes in
~0.1s.
h3. Fix
In HdfsScanNodeMt::Close(), if initial_ranges_issued_ is false, decrement the
shared
remaining_scan_range_submissions_ counter so that fragment instances waiting in
ScanRangeSharedState::GetNextScanRange() are released. This mirrors the
decrement that
IssueInitialScanRanges() performs for instances that do issue their ranges.
h3. Testing
Added the end-to-end regression test
tests/custom_cluster/test_mt_dop.py::TestMtDopFlags::test_mt_dop_union_limit_scan_range_submission,
which sets up the tables above on a single-backend cluster, runs the query (with
EXEC_TIME_LIMIT_S as a safety net) and verifies the result. Verified that the
test fails
(times out) without the fix and passes with it.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
