[
https://issues.apache.org/jira/browse/PHOENIX-8006?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Tanuj Khurana updated PHOENIX-8006:
-----------------------------------
Description:
For an ungrouped aggregate query (e.g. SELECT COUNT(*) FROM t WHERE ...)
executed with scanner lease renewal enabled (the default,
phoenix.scanner.lease.renew.enabled=true), the per-region scan results are
drained serially on the client caller thread whenever the server cannot
complete a region's aggregation within a single page-time budget. Instead of
the regions being scanned concurrently, the query degrades to one client RPC
per scanned page, summed sequentially across all regions. On large tables this
turns a query that should complete in tens of seconds into one that can take
many minutes.
All ungrouped aggregate queries (no GROUP BY) — COUNT/SUM/MIN/MAX/AVG,
COUNT(DISTINCT ...) are affected.
Mechanism
# For an ungrouped aggregate under lease renewal,
AggregatePlan.wrapParallelIteratorFactory() selects NOOP_FACTORY, so the
parallel-scan worker threads only peek() (open the scanner and fetch one page)
— they do not drain. The 73/N per-region chunk iterators are then combined by
UngroupedAggregatingResultIterator(ConcatResultIterator(...)), which is drained
on the caller thread.
# When a region cannot finish its aggregation within one page budget,
UngroupedAggregateRegionScanner emits a real partial aggregate keyed at the
last-scanned row key and sets returnImmediately so the RegionServer returns the
RPC to the client (releasing the handler).
# On the client, ScanningResultIterator skips only dummy results; a non-dummy
partial exits the skip loop immediately, so peek() returns after a single page
rather than driving the region to completion. Every subsequent page for that
region must be fetched by another next() call from the caller-thread
ConcatResultIterator — a serial continuation.
# Net effect: after an initial one-page-deep parallel peek() across regions,
all remaining pages of all regions are fetched one at a time on a single client
thread. NUM_PARALLEL_SCANS stays at the plan-time chunk count while
COUNT_RPC_CALLS grows to roughly one RPC per scanned row plus one scanner-open
per region.
Repro
{code:java}
# Turn trace logging phoenix-core/src/test/resources/log4j2-test.properties
logger.rpc.name = org.apache.hadoop.hbase.ipc.AbstractRpcClient
logger.rpc.level = TRACE {code}
Test in ServerPagingIT
[^ungrouped-aggregate-serial-drain-repro.patch]
Serial Execution output
[^ungrouped-aggregate-serial-drain-rpc-trace-evidence.txt]
was:
For an ungrouped aggregate query (e.g. SELECT COUNT(*) FROM t WHERE ...)
executed with scanner lease renewal enabled (the default,
phoenix.scanner.lease.renew.enabled=true), the per-region scan results are
drained serially on the client caller thread whenever the server cannot
complete a region's aggregation within a single page-time budget. Instead of
the regions being scanned concurrently, the query degrades to one client RPC
per scanned page, summed sequentially across all regions. On large tables this
turns a query that should complete in tens of seconds into one that can take
many minutes.
All ungrouped aggregate queries (no GROUP BY) — COUNT/SUM/MIN/MAX/AVG,
COUNT(DISTINCT ...) are affected.
Mechanism
# For an ungrouped aggregate under lease renewal,
AggregatePlan.wrapParallelIteratorFactory() selects NOOP_FACTORY, so the
parallel-scan worker threads only peek() (open the scanner and fetch one page)
— they do not drain. The 73/N per-region chunk iterators are then combined by
UngroupedAggregatingResultIterator(ConcatResultIterator(...)), which is drained
on the caller thread.
# When a region cannot finish its aggregation within one page budget,
UngroupedAggregateRegionScanner emits a real partial aggregate keyed at the
last-scanned row key and, since PHOENIX-7701, sets returnImmediately so the
RegionServer returns the RPC to the client (releasing the handler).
# On the client, ScanningResultIterator skips only dummy results; a non-dummy
partial exits the skip loop immediately, so peek() returns after a single page
rather than driving the region to completion. Every subsequent page for that
region must be fetched by another next() call from the caller-thread
ConcatResultIterator — a serial continuation.
# Net effect: after an initial one-page-deep parallel peek() across regions,
all remaining pages of all regions are fetched one at a time on a single client
thread. NUM_PARALLEL_SCANS stays at the plan-time chunk count while
COUNT_RPC_CALLS grows to roughly one RPC per scanned row plus one scanner-open
per region.
Repro
{code:java}
# Turn trace logging phoenix-core/src/test/resources/log4j2-test.properties
logger.rpc.name = org.apache.hadoop.hbase.ipc.AbstractRpcClient
logger.rpc.level = TRACE {code}
Test in ServerPagingIT
[^ungrouped-aggregate-serial-drain-repro.patch]
Serial Execution output
[^ungrouped-aggregate-serial-drain-rpc-trace-evidence.txt]
> Ungrouped aggregate queries serialize on the client when server paging
> returns partial results
> -----------------------------------------------------------------------------------------------
>
> Key: PHOENIX-8006
> URL: https://issues.apache.org/jira/browse/PHOENIX-8006
> Project: Phoenix
> Issue Type: Bug
> Affects Versions: 5.2.2, 5.3.1, 5.3.2
> Reporter: Tanuj Khurana
> Assignee: Tanuj Khurana
> Priority: Major
> Attachments: ungrouped-aggregate-serial-drain-repro.patch,
> ungrouped-aggregate-serial-drain-rpc-trace-evidence.txt
>
>
> For an ungrouped aggregate query (e.g. SELECT COUNT(*) FROM t WHERE ...)
> executed with scanner lease renewal enabled (the default,
> phoenix.scanner.lease.renew.enabled=true), the per-region scan results are
> drained serially on the client caller thread whenever the server cannot
> complete a region's aggregation within a single page-time budget. Instead of
> the regions being scanned concurrently, the query degrades to one client RPC
> per scanned page, summed sequentially across all regions. On large tables
> this turns a query that should complete in tens of seconds into one that can
> take many minutes.
> All ungrouped aggregate queries (no GROUP BY) — COUNT/SUM/MIN/MAX/AVG,
> COUNT(DISTINCT ...) are affected.
>
> Mechanism
> # For an ungrouped aggregate under lease renewal,
> AggregatePlan.wrapParallelIteratorFactory() selects NOOP_FACTORY, so the
> parallel-scan worker threads only peek() (open the scanner and fetch one
> page) — they do not drain. The 73/N per-region chunk iterators are then
> combined by UngroupedAggregatingResultIterator(ConcatResultIterator(...)),
> which is drained on the caller thread.
> # When a region cannot finish its aggregation within one page budget,
> UngroupedAggregateRegionScanner emits a real partial aggregate keyed at the
> last-scanned row key and sets returnImmediately so the RegionServer returns
> the RPC to the client (releasing the handler).
> # On the client, ScanningResultIterator skips only dummy results; a
> non-dummy partial exits the skip loop immediately, so peek() returns after a
> single page rather than driving the region to completion. Every subsequent
> page for that region must be fetched by another next() call from the
> caller-thread ConcatResultIterator — a serial continuation.
> # Net effect: after an initial one-page-deep parallel peek() across regions,
> all remaining pages of all regions are fetched one at a time on a single
> client thread. NUM_PARALLEL_SCANS stays at the plan-time chunk count while
> COUNT_RPC_CALLS grows to roughly one RPC per scanned row plus one
> scanner-open per region.
> Repro
> {code:java}
> # Turn trace logging phoenix-core/src/test/resources/log4j2-test.properties
> logger.rpc.name = org.apache.hadoop.hbase.ipc.AbstractRpcClient
> logger.rpc.level = TRACE {code}
>
> Test in ServerPagingIT
> [^ungrouped-aggregate-serial-drain-repro.patch]
> Serial Execution output
> [^ungrouped-aggregate-serial-drain-rpc-trace-evidence.txt]
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
