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
