Repository navigation
Push expressions of String columns below window functions in the query plan - #126558
yakov-olkhovskiy wants to merge 3 commits into
Conversation
…y plan
With the analyzer, the projection is computed after the window functions, so
the `Window` step buffers the columns it reads. For
`SELECT toUInt64OrZero(splitByString('@', s)[2]), count() OVER () FROM t` the
window step buffers the `String` column `s` instead of a `UInt64`, which
increased the peak memory usage about ten times compared with the old query
analysis, which computed such expressions before the window functions.
Add the query plan optimization `pushDownExpressionsBelowWindow` (setting
`query_plan_push_down_expressions_below_window`, enabled by default). It moves
the largest expressions of the `Expression` step above `Window` steps that read
a single `String` / `FixedString` column and have a fixed-size result not larger
than the smallest value of the column below the `Window` steps and the
`Sorting` steps for them. Like `pushDownVolumeReducingFunctions`, it is applied
only if nothing else reads the column after the window functions and the column
does not share memory with another column. Expressions that are executed lazily,
stateful, block-dependent or with side effects are not moved, nor are expressions
below a `LIMIT`, which may need them only for some rows. It is a second pass
optimization because it needs the steps above, and so that `liftUpFunctions`
does not move the expressions back above the window sorting.
Co-authored-by: yakov <yakov@clickhouse.com>
|
|
|
Workflow [PR], commit [9fc06eb] AI ReviewSummaryThe PR pushes small fixed-size expressions below window functions to reduce buffering. Three regressions were reproduced with the PR-head fast-test binary: two valid queries begin throwing, and a query over a Findings❌ Blockers
Missing context / blind spots
Final Verdict❌ Block — handle early-terminating |
- `typeid_cast` does not match the classes derived from `ISourceStep`, so no read
step was accepted as a source of separate columns. Use `dynamic_cast`.
- The outer `WHERE` of the original query is merged with the projection into the
`Filter` step above the `Window` step. A `Filter` keeps the same rows on
both sides of the window step, so its expressions are moved too.
- When an intermediate result is read by something else (`v` in `v > 0`), the
largest expressions can not replace the column, so the lowest ones are tried.
- Two columns can share memory only through functions that may return an argument
as is: with the same type up to `Nullable` / `LowCardinality`, or of a
composite argument. Follow only those instead of all the arguments, so that
`concat('p@', toString(number))` is not considered to share memory with
`number % 3`.
- The `Sorting` steps for windows have `PARTITION BY` descriptions, and a
`Window` step only adds columns, so both are passed through.
Co-authored-by: yakov <yakov@clickhouse.com>
The test checks that the window step does not buffer the column read only by the expression (the query fails with `MEMORY_LIMIT_EXCEEDED` with the optimization disabled), that the expression is computed below the `Window` and `Sorting` steps in the plan, also on the initiator of a distributed query, and that it stays after the window functions when the column is still read, the results are larger, the column may share memory with another one, or with a fixed-size grouping key, `LIMIT`, `QUALIFY`, lazily executed arguments, block-dependent functions, side effects and `arrayJoin`. Co-authored-by: yakov <yakov@clickhouse.com>
| if (typeid_cast<const LimitStep *>(step) || typeid_cast<const LimitRangeStep *>(step)) | ||
| return true; | ||
|
|
||
| if (const auto * sorting = typeid_cast<const SortingStep *>(step); sorting && sorting->getLimit()) | ||
| return true; |
There was a problem hiding this comment.
[P1] Treat early-terminating LIMIT BY as a projection barrier
LimitByStep can also stop reading its input: when all grouping keys are constant, LimitByTransform cancels after satisfying the group limit. This guard misses that case, so pushing a throwing expression below an unbounded window evaluates rows that the original projection never reads. With the PR's fast-test binary, this returns 1, 10000 with the new optimization disabled, but throws CANNOT_PARSE_TEXT with it enabled:
SELECT toUInt64(s), count() OVER ()
FROM
(
SELECT if(number < 1000, '1', 'x') AS s
FROM numbers(10000)
)
LIMIT 1 BY 'group'
SETTINGS max_block_size = 10, max_threads = 1;Include early-terminating LimitByStep ancestors in this check and add this case to the regression coverage.
| /// Merging into a `Filter` below would compute the expressions for the rows it removes. | ||
| if (typeid_cast<const ExpressionStep *>(bottom_node->step.get())) | ||
| tryMergeExpressions(&pushed_node, nodes, settings); |
There was a problem hiding this comment.
[P1] Keep the pushed expression after an arrayJoin in the lower DAG
The separate pushed step is safe here, but merging it into an expression containing arrayJoin can move it before rows are eliminated. ExpressionActions::linearizeActions delays array joins until other ready actions have executed, and tryMergeExpressions only guards stateful/non-deterministic functions. A deterministic conversion that can throw is therefore evaluated on an input row whose empty array previously removed it. This reproduces on the PR's fast-test binary:
SELECT toUInt64(s), count() OVER ()
FROM
(
SELECT if(number = 0, 'x', '1') AS s,
arrayJoin(if(number = 0, [], [1])) AS a
FROM numbers(2)
)
SETTINGS query_plan_lower_array_join_function = 0,
max_threads = 1;With query_plan_push_down_expressions_below_window = 0 it returns 1, 1; with the default enabled it throws CANNOT_PARSE_TEXT for 'x'. Skip this merge when the lower expression contains an array join, keeping the pushed step after it, and cover this supported setting combination.
| /// `typeid_cast` does not match derived classes. | ||
| if (dynamic_cast<const ISourceStep *>(step) || typeid_cast<const AggregatingStep *>(step) || typeid_cast<const MergingAggregatedStep *>(step)) | ||
| return true; |
There was a problem hiding this comment.
[P2] Account for source columns whose storage remains owned elsewhere
An ISourceStep does not guarantee that removing its output releases the column's payload. An uncompressed Memory table returns columns from its retained storage snapshot, so the original window can buffer those pointers while computing the hash one output block at a time. Pushing the hash down retains an additional result column for the entire window without releasing the table's strings.
On the PR's fast-test binary, the final query below peaks at 16.50 MiB with the optimization disabled and 20.09 MiB enabled; setting max_memory_usage = '19M' makes only the enabled version fail with MEMORY_LIMIT_EXCEEDED:
CREATE TABLE m (s String) ENGINE = Memory;
INSERT INTO m SELECT toString(number) FROM numbers(1000000);
SELECT cityHash64(s), count() OVER () FROM m FORMAT Null
SETTINGS max_block_size = 8192, max_threads = 1,
max_memory_usage = '19M';EXPLAIN confirms the hash moves below Window. Conservatively exclude borrowed source columns such as these unless releasing their payload can be established, and add a memory regression test for this read path.
closes https://github.com/ClickHouse/clickhouse-private/issues/78134
Alternative to #126062: the same fix as a query plan optimization instead of a change of the planner.
With the analyzer, the projection is computed after the window functions, so the
Windowstep buffers the columns it reads. ForSELECT toUInt64OrZero(splitByString('@', c2)[2]) AS v, count() OVER () AS n FROM t1, the window step buffers theStringcolumnc2instead of aUInt64. The old query analysis computed such expressions before the window functions, and since 26.9 it cannot be selected.Peak memory with
clickhouse local --max_threads 1, on 10 million rows of 160-byte strings:SELECT count() FROM (SELECT DISTINCT toUInt64OrZero(splitByString('@', c2)[2]) AS v, count() OVER () AS n FROM ...) WHERE v > 0(the original report)... count() OVER (PARTITION BY k) ...(no external sorting)This adds the optimization
pushDownExpressionsBelowWindow(settingquery_plan_push_down_expressions_below_window, enabled by default). It generalizes the existingpushDownVolumeReducingFunctions, which moveslength,lengthUTF8,emptyandnotEmptybelowSortingandFiltersteps, to the window case.It looks at the
ExpressionorFilterstep directly above aWindowstep. It moves the expressions that read a singleString/FixedStringcolumn and whose result type has a maximum size not larger than the smallest value of the column (the 8-byte offset of aString,NforFixedString(N)). They go below the chain ofWindowsteps and theSortingsteps for them, and are merged into theExpressionstep there. The plan rewrite is the same as inpushDownVolumeReducingFunctions:ActionsDAG::splitof the parent, and a new step whose first part is computed below. AFilterkeeps the same rows on both sides of the window functions, so the outerWHERE v > 0of the original query, merged with the projection, does not prevent it.The expressions are moved only if this cannot make the buffered rows larger or change the results:
PARTITION BY/ORDER BYcolumns of the windows and sortings. Several expressions of one column are moved together, only if the sum of their result sizes fits. The largest expressions are tried first (toUInt8(length(s))), then the lowest, when an intermediate result is read by something else (vinv > 0).Nullable/LowCardinality, or a composite one, like with an identityCAST,toNullable,assumeNotNull,tupleElementor an alias. Following such functions through the steps below (Expression,Filter,Window,Union, header-preserving steps) to reads or aggregation, the column must have no common node with another column.runningDifference), non-deterministic in the scope of the query (blockSize,rand), or has side effects (sleepEachRow). The window and sorting steps change the order of rows and their blocks.if,multiIf,and,or). Otherwise it would be executed for all rows, and for example throw.LIMIT,LIMIT AFTER/UNTILorORDER BY ... LIMITabove the window functions, unless there is an aggregation in between. With aLIMIT, the projection may be needed only for some rows. This includes aLIMITof an outer query, because the whole plan is visible.QUALIFY, the projection is a separate step above theFilterand is not changed, and anarrayJoinin the parent prevents the rewrite.It is a second pass optimization:
LIMIT.liftUpFunctions(tryExecuteFunctionsAfterSorting), which moves expressions aboveSortingsteps, runs only in the first pass. So the expressions can also be moved below theSortingsteps for the windows without being moved back.optimizeReadInOrderandoptimizeWindowPerPartition, and the moved expressions are merged into theExpressionstep below the window sorting, so those see the same shape of the plan.ReadFromRemoteheader is still theStringcolumn. It is not applied withmake_distributed_plan.Testing: the new test
05407_window_functions_push_down_expressions_below_window:MEMORY_LIMIT_EXCEEDEDwith the setting disabled) and the results for all the cases above.All 307 stateless tests that use window functions or
QUALIFYpass locally with the test cluster configs, and so do the tests of settings documentation and history.Changelog category (leave one):
Changelog entry (a user-readable short description of the changes that goes into CHANGELOG.md):
Reduce the memory usage of queries with window functions: an expression of a
StringorFixedStringcolumn with a small fixed-size result, such astoUInt64OrZero(splitByString('@', s)[2])orlength(s), is now computed before the window functions if nothing else reads the column after them, so the window functions buffer the result instead of the column. Controlled by the new settingquery_plan_push_down_expressions_below_window.Workflow [PR]
Sync PR [sync-upstream/pr/126558]