gortiz opened a new pull request, #19905:
URL: https://github.com/apache/pinot/pull/19905
## Summary
`ANY_VALUE` in a group-by query evaluated a transform input once per row
instead of once per block. This change reads the input values once per block.
One block of N rows did O(N^2) work before this change, and does O(N) work
after it.
## Background
`AnyValueAggregationFunction.aggregateHelper()` reads a non-dictionary input
through `getDirectValue(bvs, i)`. That method called `bvs.get*ValuesSV()[i]`
for each row.
When the input is a transform, `TransformBlockValSet` does not cache the
result. Each `get*ValuesSV()` call evaluates the transform on the full block
again. In group-by mode the loop never stops early, so a block of N rows
evaluates the transform N times. Without group-by, the loop stops after the
first value, and the cost stays small.
```
GroupByExecutor.process(block) one call, no termination
check inside
└─ AnyValueAggregationFunction.aggregateGroupBySV()
└─ aggregateHelper(): for each row i (N rows)
└─ getDirectValue(bvs, i)
└─ TransformBlockValSet.getStringValuesSV()
└─ jsonExtractScalar on all N rows → N x N JSON parses
```
We saw this on a production cluster with this query shape on the leaf stage
of a multi-stage query:
```sql
SELECT series_id, <time bucket>,
LASTWITHTIME(value, ts, 'DOUBLE'),
ANY_VALUE(jsonExtractScalar(attributes,
'$[''cloud.availability_zone'']', 'STRING', 'unknown'))
FROM t
WHERE ...
GROUP BY series_id, <time bucket>
```
The `attributes` values are about 1.6 KB of JSON. A full block of 10,000
rows did 100 million JSON parses. Each leaf worker stayed in one `process()`
call for 15 to 40 minutes. The query timeout and the cancellation do not stop
the work, because the work does not reach a termination check. As a result, the
work continued for a long time after the query failed, and the CPU usage on the
servers increased by about 5 cores each.
## Change
- Replace `getDirectValue(BlockValSet, int)` with
`getDirectValueReader(BlockValSet)`. The new method reads the values array once
and returns an `IntFunction` over that array.
- `aggregateHelper()` gets the reader once per block, before the loop over
the rows.
## Tests
`AnyValueGroupByRecomputationTest` calls `aggregateGroupBySV()` on a block
of 1,000 rows with a non-dictionary STRING input. The test makes sure that
`getStringValuesSV()` is called once. Before the fix, the test failed with
1,000 calls.
A local benchmark (not part of this PR) ran the query shape above on one
segment, with one group per row and about 1.6 KB of JSON per row:
| Rows in the block | Before | After | Baseline: `DISTINCTCOUNT` on the same
transform |
|---|---|---|---|
| 1,000 | 1.9 s | 7 ms | 7 ms |
| 2,000 | 16.7 s | 8 ms | 7 ms |
| 4,000 | 31.6 s | 13 ms | 15 ms |
| 8,000 | 238 s | 23 ms | 26 ms |
The existing `AnyValueAggregationFunctionTest` tests (38) pass.
## Follow-up
This change does not add a termination check inside
`GroupByExecutor.process()`. A different aggregation with an expensive
per-block cost can still run past the query deadline. That is a separate change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
