Hi all,
A short progress update on the nested-TVF parallelization we discussed earlier.
The PR has been created and related UT & IT were added.
Change
------
The fix is small: ParallelizeGrouping#checkPrefixMatch now compares the
partition prefix as a column *set* instead of position by position. Partition
columns identify a partition by their values, so any writing order of the same
set still clusters rows into the same partitions.
The ORDER BY part needs no check there: a group node is executed as a streaming
sort, and it re-establishes the within-partition order itself from the full
ordering scheme; the input only has to be clustered. Distributed planning needs
no change at all — once the outer group survives, the existing split logic
schedules both processor levels into the same per-region fragment. As a
consequence, even when the two levels declare different ORDER BY clauses, both
levels now stay parallel and the outer group sorts within each partition
locally.
Distributed planning still adjusts the partition order when it can
------------------------------------------------------------------
This PR reuse the existing mechanism that rewrites ordering
during distributed planning; we keep relying on it and it stays in effect.
For a set-semantic processor (and for window / pattern-recognition / row-number
nodes), the distributed planner takes the ordering requested by the ancestor —
the top-level ORDER BY if there is one, otherwise the outer function's own
declared order — and, when the request starts with the same set of partition
columns, rewrites the group's partition prefix to that order. What that
actually changes is the device comparator and the scan order, i.e. the order in
which partitions are emitted, not the order of rows inside a partition. Two
invariants hold:
- the prefix region always remains "all partition columns"; columns the
request does not mention are appended after the requested ones, and the
ORDER BY suffix always stays behind the prefix;
- the suffix keeps its original relative order — it is never permuted. If the
request interleaves a non-partition column into the prefix, or needs a
different suffix order, the rewrite is abandoned and a real sort is kept.
So partition order is adjusted only where it is not observable from SQL
(clustering is defined by the column set), and whenever a specific row order is
required we pay for a real sort instead of pretending. When the scan can prove
the rewritten order the group is eliminated; otherwise it remains as a sort.
Thanks,
Weihao Li
At 2026-10-30 19:12:55, "Weihao Li" <[email protected]> wrote:
>Hi all,
>
>
>I'd like to share an optimization we are working on for the table model:
>running
>nested set-semantic table functions (TVFs) in parallel when they declare the
>same
>PARTITION BY columns. Feedback on the scope and on the open questions at the
>end
>would be very welcome.
>
>
>Background
>----------
>A set-semantic table function processes each partition independently, in the
>order
>given by its ORDER BY clause. In the planner this is represented as a GroupNode
>under the TableFunctionProcessorNode: the group's ordering scheme is
>(PARTITION BY columns, ORDER BY columns), and the number of partition keys is
>the
>size of the PARTITION BY list.
>
>
>ParallelizeGrouping keeps the GroupNode only when the input is provably
>clustered
>by the partition keys (for a device table scan this requires the partition
>keys to
>cover all tag columns, i.e. each partition is region-local); otherwise the
>group
>degrades to a SortNode. When the GroupNode survives, the distributed plan
>generator
>splits the TableFunctionProcessorNode per region instead of merging all regions
>behind a Collect/MergeSort, so the function runs on every data region in
>parallel.
>
>
>The nested case
>---------------
>When TVF B takes the output of TVF A as its table argument and both declare the
>same PARTITION BY, A's output is already clustered by those keys. B's grouping
>is
>therefore implied and its GroupNode can be removed, letting both processors be
>scheduled into the same per-region fragment: both levels run in parallel,
>pipelined,
>with no materialization boundary between them.
>
>
>This already worked when the two PARTITION BY lists were written in the same
>column order. When the same column set appeared in a different order, e.g.
>
>
> SELECT * FROM SESSION(DATA => (
> SELECT * FROM SESSION(DATA => table1
> PARTITION BY (tag1, tag2, tag3) ORDER BY time, GAP => 1m))
> PARTITION BY (tag3, tag2, tag1) ORDER BY time, GAP => 2m)
>
>
>the prefix check failed, the outer GroupNode was rewritten to a SortNode, and
>the
>distributed plan degraded to a single outer processor consuming a global
>MergeSort
>over all regions, while the inner level stayed parallel. Simplified plan
>before:
>
>
> Collect
> +- Exchange
> | +- TFP[outer session, PARTITION BY tag3,tag2,tag1] <- single instance
> | +- MergeSort(tag3, tag2, tag1, time)
> | +- Exchange <- TFP[inner session, ...] <- Scan (region 1)
> | +- Exchange <- TFP[inner session, ...] <- Scan (region 2)
> +- ...
>
>
>Proposal
>--------
>Treat PARTITION BY as a set in this check. If the outer group's partition-key
>set
>equals the nested processor's partition-key set but the orders differ, align
>the
>group's partition prefix with the child's actual order. This is safe because:
>
>
> - The set of columns, not their order, identifies a partition.
> - SQL does not constrain the order between partitions; only the
> within-partition
> ORDER BY is observable, and it is preserved unchanged as the ordering
> suffix.
> - Ordering metadata published above the processor is derived from the child's
> actual ordering, so no ordering is claimed that is not produced.
> - If an ancestor requires a specific inter-partition order, the rewrite is
> not
> applied and the existing sort/merge fallback handles it.
> - In distributed planning we never rewrite the partition prefix of a
> GroupNode
> fed by a TVF processor (that would silently assume a sort that does not
> happen); the order actually produced by the input is kept.
>
>
>With this, the plan above becomes (simplified):
>
>
> Collect
> +- Exchange
> | +- TFP[outer session, PARTITION BY tag3,tag2,tag1]
> | +- TFP[inner session, PARTITION BY tag1,tag2,tag3]
> | +- Scan (region 1) <- both levels in the same fragment
> +- ... <- likewise for every region
>
>
>Scope and limitations
>---------------------
> - Parallelism still requires region-local partitions (partition keys cover
> all
> tag columns); otherwise the correct single-instance + sort/merge plan is
> kept.
> - Functions that cannot be split across regions (e.g. FFT) and row-semantics
> arguments keep their current behavior.
> - The alignment currently applies to direct nesting. A Project between the
> two
> processors prevents it, which is a possible follow-up.
> - Queries whose ancestors demand a specific global order still need a global
> sort; the optimization targets the common case where the order between
> partitions is unconstrained.
>
>
>Thanks,
>Weihao
