Pain Sun created IMPALA-15592:
---------------------------------
Summary: Paimon scan ranges are not balanced: most splits are
scheduled on a single executor
Key: IMPALA-15592
URL: https://issues.apache.org/jira/browse/IMPALA-15592
Project: IMPALA
Issue Type: Bug
Components: Backend, Frontend
Affects Versions: Impala 5.0.0
Reporter: Pain Sun
When a Paimon table is queried, almost all Paimon splits are assigned to one
executor.
The other executors get about one split each. The query runs at the speed of
that
single host.
h3. Reproduction
Any Paimon table that has many more splits than there are executors. We used a
primary-key table with 500 buckets and about 6.26 billion rows (hadoop catalog
on HDFS):
{code:sql}
SET MT_DOP=8;
SELECT count(*) FROM test_db.paimon_pk_table;
{code}
h3. Actual result
Duration 35.5s. Rows read per PAIMON_SCAN_NODE instance:
{noformat}
host-A 8 instances 713M .. 836M rows each 31.3s .. 34.6s each (6.18B
rows in total = 98.8%)
31 other hosts 1 instance each 0.1M .. 5.3M rows 0.01s .. 0.18s
{noformat}
All other operators take under 0.6s. The wall-clock time is the scan time on
host-A.
h3. Expected result
Splits are spread evenly across all executors, as already promised by the
comment in
PaimonScanNode.computeNumNodes(): "scan ranges are scheduled round-robin, since
they
have no location information".
h3. Analysis
1. PaimonScanNode.computeScanRangeLocations() gives every split the same dummy
location
"localhost:12345" and stores the split in TScanRange.file_metadata. No
length is set.
So all Paimon ranges become remote ranges.
2. In Scheduler::AssignmentCtx::RecordScanRangeAssignment(), only
hdfs_file_split
(real length) and kudu_scan_token (constant 1000) add to an executor's
assigned
bytes. A Paimon range counts as 0 bytes.
3. Scheduler::AssignmentCtx::SelectRemoteExecutor() gives one range to each
unused
executor first, and then always takes assignment_heap_.top(). Every executor
stays
at 0 assigned bytes, so the heap top never changes, and all remaining ranges
go to
that one executor.
h3. Suggested fix
Minimal fix in scheduler.cc, the same idea as the existing Kudu hack:
{code:cpp}
} else if (scan_range_locations.scan_range.__isset.kudu_scan_token) {
scan_range_length = 1000;
} else if (scan_range_locations.scan_range.__isset.file_metadata) {
// Paimon splits have no length; use a nominal size so they are balanced.
scan_range_length = 1000;
}
{code}
A better long-term fix: send the real split size (the sum of DataSplit data
file sizes)
and the real HDFS block locations, so that balancing and locality both work.
h3. Workaround
None found. NUM_REMOTE_EXECUTOR_CANDIDATES only applies to hdfs_file_split
ranges.
Raising MT_DOP only adds threads on the overloaded host.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
