[Wei-hao-Li]

fix #19207

[codecov]

Codecov Report

❌ Patch coverage is 90.00000% with 3 lines in your changes missing coverage. Please review.
✅ Project coverage is 48.94%. Comparing base (54bff3d) to head (1d47470).

Files with missing lines	Patch %	Lines
...nal/planner/optimizations/ParallelizeGrouping.java	90.00%	3 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff            @@
##             master   #19239   +/-   ##
=========================================
  Coverage     48.94%   48.94%           
  Complexity      689      689           
=========================================
  Files          5589     5589           
  Lines        413857   413877   +20     
  Branches      54594    54602    +8     
=========================================
+ Hits         202544   202563   +19     
- Misses       211313   211314    +1     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[sonarqubecloud]

Quality Gate failed

Failed conditions
C Reliability Rating on New Code (required ≥ A)

See analysis details on SonarQube Cloud

Catch issues before they fail your Quality Gate with our IDE extension SonarQube for IDE

[JackieTien97]

Thanks for the fix. I reproduced Q1-Q7 from #19207 and this PR fixes them. With only ParallelizeGrouping.java reverted to the merge base, exactly the 4 new planner tests and the 6 new IoTDBWindowTVFRegionsIT cases fail; with the PR they all pass.

How I checked: PlanTester (logical and distributed plans), plus a 1C1D cluster with 4 DataRegions. I compared the PR head against the same tree with only ParallelizeGrouping.java swapped back to the base version.

Two things I'd like to see addressed:

1. The same wrong-result class is still reachable through RowNumber / TopKRanking (inline on L220). The new model assumes that a group which degrades to a SortNode yields one globally ordered stream. But TableDistributedPlanGenerator drops that SortNode under RowNumberNode and TopKRankingNode. On the PR's own t2, an outer PARTITION BY s1 over row_number() OVER (PARTITION BY s1, s3) returns 18 rows instead of 3. The merge base gives the same result, but this is exactly the shape in the PR title.
2. Safe cases now lose parallelism or gain a full sort (inline on L370, L355 and L141). The exact-key-set rule also turns the cases #19207 lists as safe into sorts. For example, a window PARTITION BY all TAG columns over GROUP BY all TAG columns + date_bin(...) no longer runs per region.

Smaller items: the class Javadoc has a stale example (L75), one test comment's reasoning doesn't hold (TableFunctionTest L376), and tests are missing both for the remaining wrong-result shape and for the cases that should stay parallel (IoTDBWindowTVFRegionsIT L178).

[JackieTien97]

This branch tells the ancestor that the degraded group is "one stream ordered by the full scheme". That stops being true in distribution planning when the consumer is a RowNumberNode or TopKRankingNode: TableDistributedPlanGenerator.visitRowNumber and visitTopKRanking drop a SortNode child with node.setChild(((SortNode) node.getChild()).getChild()) (around L3759 and L3825). RowNumber then emits rows in input order, and TopKRanking emits groups in the order it first sees them. An outer strict prefix therefore still runs as a GroupNode (TableStreamSortOperator) over unclustered input.

Repro on the t2 from IoTDBWindowTVFRegionsIT (one device, 6000 rows, 4 DataRegions), PR head:

SELECT s1, cnt, count(*) AS n, max(rn) AS mrn FROM (
  SELECT s1, rn, count(*) OVER (PARTITION BY s1) AS cnt FROM (
    SELECT s1, s3, row_number() OVER (PARTITION BY s1, s3) AS rn FROM t2))
GROUP BY s1, cnt ORDER BY s1, cnt

Expected 3 rows (0,1980,1980,1, 1,2013,2013,1, 2,2007,2007,1). Actual: 18 rows. Each s1 comes back as 6 separate partitions of 265-384 rows. The merge base returns the same 18 rows. The plan is WindowFunction(s1) > GroupNode[s1] > Project > RowNumber(s1, s3) > DeviceTableScan; the Sort below RowNumber is gone.

The TopKRanking form fails the same way. I gave t2 an extra INT32 field s2, filled with a second pseudo-random value in 0..4. On both the PR and the merge base, this returns 20 rows instead of 3:

SELECT s1, cnt, count(*) AS n, count(s2) AS c2, count(s3) AS c3 FROM (
  SELECT s1, s2, s3, count(*) OVER (PARTITION BY s1) AS cnt FROM (
    SELECT s1, s2, s3 FROM (
      SELECT s1, s2, s3, row_number() OVER (PARTITION BY s1, s3 ORDER BY s2) AS rn FROM t2)
    WHERE rn <= 1))
GROUP BY s1, cnt ORDER BY s1, cnt

The plan is WindowFunction(s1) > GroupNode[s1] > Project > TopKRanking(s1, s3) > DeviceTableScan. When the input spans several regions, a Collect ends up under the ranking node and groupingInvalidatedByCollect happens to catch both shapes. When the input comes from a single region (here: one device), nothing does. (Variants that order by time keep a pruning Project between the Filter and the Window. They stay a Window over a kept Sort, and are correct.)

Suggestion: treat RowNumberNode and TopKRankingNode like table functions and pattern recognition. Add visitRowNumber / visitTopKRanking that call checkPartitionKeysMatch(context, node.getPartitionBy()), so the ancestor's context never reaches a SortNode that will be dropped. I also checked one case that is correct today: outer keys that extend into TopKRanking's ORDER BY (PARTITION BY (s1, s2) over PARTITION BY s1 ORDER BY s2), because groups are drained in rank order. That case could stay allowed.

[JackieTien97]

Requiring the exact key set here (and in visitAggregation and the surviving-group branch) also rejects the strict prefixes that #19207 lists as safe: prefixes that still contain every TAG column. Such a partition cannot span devices, so it cannot span regions, and each region's stream is already ordered by device.

The most visible case is a per-device window over time-bucketed aggregates. With 12 devices over 4 DataRegions (TAG columns a, b):

SELECT a, b, h, m, lag(m) OVER (PARTITION BY a, b ORDER BY h) AS prev
FROM (SELECT a, b, date_bin(1h, time) AS h, avg(s1) AS m FROM tm GROUP BY a, b, date_bin(1h, time))

On the merge base, EXPLAIN shows one WindowFunction per region directly over its AggregationTableScan, with only a Collect on the coordinator. With this PR there is a single WindowFunction on the coordinator over a MergeSort, and the regions only scan. The results are the same, but this shape is very common: aggregate per device and time bucket, then apply a window, FORECAST or MATCH_RECOGNIZE. PlanTester shows the same loss for count(s1) OVER (PARTITION BY tag1, tag2, tag3) over max(s1) OVER (PARTITION BY tag1, tag2, tag3, attr1): the outer window moves from the region fragments to a single instance.

For AggregationTableScanNode, I think also accepting ImmutableSet.copyOf(requiredKeys).equals(ImmutableSet.copyOf(node.getPreGroupedSymbols())) would keep this case. Each region's output is ordered by the pre-grouped keys, which include every TAG whenever the scan runs per region. AggregationNode needs more care, because the distributed visitAggregation can still drop its pre-grouping. More generally, the decision needs the bottom-up property described in #19207. One sorted stream allows any prefix. Per-instance clustering allows the full key set, and also subsets that still contain every TAG when each instance is ordered by device.

[JackieTien97]

A completed aggregation emits each full grouping key exactly once, whether it hashed its input or not. So when the required keys are exactly the grouping keys, every partition is a single row and is trivially clustered. TableDistributedPlanGenerator.groupingInvalidatedByCollect already relies on this ("A completed aggregation emits each full grouping key once, even when it hashes its input.", around L2209). Only strict subsets need TO_SORT, and checkPartitionKeysMatch already rejects those.

As written, SESSION(DATA => (SELECT s1, s2, max(time) AS time FROM table1 GROUP BY s1, s2) PARTITION BY (s1, s2) ...) goes from TableFunctionProcessor > GroupNode[s1, s2] > Aggregation on the base to TableFunctionProcessor > Sort > Aggregation, which buffers and fully sorts the aggregation output. I'd drop the getPreGroupedSymbols().isEmpty() branch (here and in visitAggregationTableScan) and keep only checkPartitionKeysMatch. The logical preGroupedSymbols is not final anyway: the distributed visitAggregation can still clear it.

[JackieTien97]

This says a permutation of the same key set also matches, but sameKeySet below only accepts it when partitionKeyCount == childOrder.size(). The degraded-group branch now passes the full ordering (partition keys + ORDER BY keys), so a reordered key set over an inner window with an ORDER BY no longer matches.

Example: count(s1) OVER (PARTITION BY tag2, tag1) over sum(s1) OVER (PARTITION BY tag1, tag2 ORDER BY time) was WindowFunction > GroupNode[tag2, tag1] > MergeSort on the base. It is now WindowFunction > Sort > MergeSort, an extra full sort. In one sorted stream, any permutation of the first k sort keys is clustered. Comparing ImmutableSet.copyOf(requiredPrefix) with ImmutableSet.copyOf(childOrder.subList(0, k)) would be correct and would also match this Javadoc.

[JackieTien97]

After this change, GroupNode[PK={tag1,tag2}, OK={s1}] -> AggregationNode[group={tag1,tag2,s1}] is no longer parallelized. checkPartitionKeysMatch requires the exact grouping key set, so this shape now becomes a SortNode; in PlanTester, PARTITION BY tag1, tag2, tag3 over GROUP BY tag1, tag2, tag3, s1 plans as WindowFunction > Sort > Aggregation. Please update or remove this example so the list matches the new rules.

[JackieTien97]

The reasoning here doesn't hold. Every row of a completed aggregation has a distinct grouping key, so with the partition keys equal to the grouping keys each partition is one row. The input is then clustered whether or not the aggregation hashed it (see groupingInvalidatedByCollect in TableDistributedPlanGenerator). As written, this test locks in an extra full sort. If the hash branch is dropped, I'd change this test to assert that the group is kept. A strict-subset variant (PARTITION BY (s1) over GROUP BY s1, s2) can stay as the case that must sort.

[JackieTien97]

The new tests only check one direction: "must fall back to a sort". Two gaps:

• The shapes from the comment on ParallelizeGrouping L220 are not covered. An outer PARTITION BY s1 over row_number() OVER (PARTITION BY s1, s3), with or without ORDER BY ... WHERE rn <= 1, still returns wrong results on this t2.
• Nothing pins the safe cases that should keep their parallelism or their streaming group. One example would be a distributed-plan assertion for a window PARTITION BY all TAG columns over GROUP BY all TAG columns + date_bin.

Without these, the regressions described on L370, L355 and L141 pass CI unnoticed.