[Wei-hao-Li]

Experiment

Machine and database conf:
16 cores cpu, 32GB memory, 1C1D, disable compaction

Benchmark conf:
IoTDB_DIALECT_MODE=table
OPERATION_PROPORTION=1:0:0:0:0:0:0:0:0:0:0:0
DEVICE_NUMBER=1000
SENSOR_NUMBER=10
BATCH_SIZE_PER_WRITE=1000
LOOP=100

SQL:
Three-level nested TVF query

SELECT count(*)
FROM SESSION(
  DATA => (
    SELECT time, tag_0, tag_1, tag_2
    FROM SESSION(
      DATA => (
        SELECT time, tag_0, tag_1, tag_2
        FROM SESSION(
          DATA => table_0
            PARTITION BY (tag_0, tag_1, tag_2) ORDER BY time, GAP => 1m
        )
      )
      PARTITION BY (tag_1, device_0, tag_2) ORDER BY time, GAP => 2m
    )
  )
  PARTITION BY (tag_1, tag_2, device_0) ORDER BY time, GAP => 3m
);

Result:

Region num	quey costs-before（ms）	quey costs-after (ms)	speedup
1	856	903	0.95x
2	558	461	1.21x
4	584	309	1.89x
6	600	244	2.14x
8	1290	943	2.46x

[JackieTien97]

I don't think this is ready to merge yet. The intended case works: when the outer PARTITION BY lists exactly the inner columns in a different order, results match the base and the plan stays parallel. But checkPrefixMatch now also accepts an outer PARTITION BY that is a reordered strict subset of the child's keys, and those queries return wrong results. The method has 8 call sites, so this also affects plain window functions, not just set-semantic TVFs.

Results from a 1C1D cluster with 4 DataRegions. "Base" is this PR with ParallelizeGrouping.java reverted to its parent version (e92ecf1):

Query	Base	This PR	With the suggested fix
SESSION (b, a) over SESSION (a, b, c)	1 window, 40 rows	4 windows, 10 rows each	1 window, 40 rows
count(x) OVER (PARTITION BY b, a) over a window on (a, b, c)	40	10	40
SESSION (b, a) over GROUP BY a, b, c	1 window	4 windows	1 window
SESSION (s1, s2) over hash GROUP BY s2, s1, s3	1 window per key	6 windows per key	1 window per key
SESSION (c, b, a) over SESSION (a, b, c) (the intended case)	40 windows	40 windows	40 windows

The suggested fix is in the inline comment on checkPrefixMatch. With it, the new TableFunctionTest cases still pass and the relational plan tests show no failures (527 run, 4 skipped). Plans with an outer row_number() ... WHERE rn <= 2 (TopKRanking) or an outer MATCH_RECOGNIZE are split per region in the same way, but I only checked those plans and did not run them. I did not run 1C3D.

Existing bug, not introduced here: the same-order strict prefix is already wrong on the base. With outer (a, b) over inner (a, b, c), SESSION gives 4 windows and the window count gives 10. GROUP BY s1, s2, s3 with PARTITION BY (s1, s2) gives 6 windows per key. The class Javadoc also lists GroupNode[PK={tag1}, OK={}] -> TableFunctionNode[partition={tag1,tag2}] as valid. Requiring the exact key set everywhere fixed these results in my experiment, but it fails WindowFunctionOptimizationTest.testSwapWindowFunctions, because a strict prefix is safe when the inner input is a single sorted stream. So that fix needs more thought and can be handled separately.

Could you also add a PR description and a more descriptive commit message than fix? The title has a typo too (conetent).

[JackieTien97]

This accepts more than "the same PARTITION BY columns in a different order". Only the first partitionKeyCount child keys are compared, so outer (b, a) over inner (a, b, c) also becomes ENABLE_PARALLEL.

That outer partition can span DataRegions. TableDistributedPlanGenerator.visitGroup still returns one node per child, and canSplitTableFunctionProcessor then splits the outer function per region, so each region computes its own copy of the partition. visitWindowFunction, visitRowNumber, visitTopKRanking and visitPatternRecognition split in the same way. The base commit plans this query with a single outer function above a MergeSort.

Repro with 4 DataRegions and 40 devices (a = 'A', b = 'B', c = 'c00'..'c39', one row each at time 1..40):

CREATE TABLE t(a STRING TAG, b STRING TAG, c STRING TAG, s1 DOUBLE FIELD);

SELECT window_start, window_end, count(*) AS cnt FROM SESSION(DATA => (
    SELECT window_start AS inner_start, window_end AS inner_end, time, a, b, c, s1
    FROM SESSION(DATA => t PARTITION BY (a, b, c) ORDER BY time, GAP => 1ms)
) PARTITION BY (b, a) ORDER BY time, GAP => 100ms)
GROUP BY window_start, window_end ORDER BY window_start;

Expected, and what the base returns: one window [1ms, 40ms] with cnt = 40. This PR returns 4 windows with cnt = 10 each.

I'd apply the set comparison only when the child exposes exactly the partition keys, and keep the positional rule otherwise. The Javadoc on lines 125-128 needs the same restriction.

Suggested change

	context.canOptimized =
	ImmutableSet.copyOf(requiredPrefix).equals(ImmutableSet.copyOf(actualPrefix))
	? ENABLE_PARALLEL
	: TO_SORT;
	boolean sameKeySet =
	partitionKeyCount == childOrder.size()
	&& ImmutableSet.copyOf(requiredPrefix).equals(ImmutableSet.copyOf(actualPrefix));
	context.canOptimized =
	sameKeySet || requiredPrefix.equals(actualPrefix) ? ENABLE_PARALLEL : TO_SORT;

[JackieTien97]

childOrder is not always a sort order. visitAggregation passes node.getGroupingKeys() here, and a hash aggregation's output is not clustered by a prefix of those keys. With this change, GROUP BY s2, s1, s3 with PARTITION BY (s1, s2) keeps the GroupNode, where the base commit used a Sort. TableStreamSortOperator then flushes non-contiguous input in batches, so one partition is cut into several runs.

Repro with one device and 6000 rows, where (s1, s2) cycles through (0, 0), (1, 1), (2, 2) and s3 is unique:

SELECT s1, s2, window_start, window_end, count(*) AS cnt FROM SESSION(DATA => (
    SELECT s1, s2, s3, max(time) AS time FROM t2 WHERE d = 'd1' GROUP BY s2, s1, s3
) PARTITION BY (s1, s2) ORDER BY time, GAP => 100000ms)
GROUP BY s1, s2, window_start, window_end ORDER BY s1, s2, window_start;

The base returns 1 window per key (3 rows). This PR returns 6 windows per key (18 rows). With several regions, groupingInvalidatedByCollect falls back to a Sort, so this only happens when there is no Collect under the aggregation. The suggestion on lines 145-148 also fixes this case.

[JackieTien97]

All the new cases use the complete key set with SESSION, so none of them catches the strict-subset regression: the relational plan tests all pass with it.

Could you add a negative plan test? PARTITION BY (tag2, tag1) over (tag1, tag2, tag3) should keep the outer function in the root fragment (Output > TableFunctionProcessor > MergeSort). Since checkPrefixMatch is shared, a window-function case would help too. An IT with a fixed region count (setDataRegionGroupExtensionPolicy("CUSTOM") and setDefaultDataRegionGroupNumPerDatabase(4)) that checks window counts would cover the results.

[JackieTien97]

The comment says all three groups are executed in parallel per region, but this test only asserts the logical plan. If the distributed plan went back to a single stream at the root, it would still pass. Could you add getFragmentPlan assertions?

[JackieTien97]

Nit: the MergeSort in fragment 3 comes from the cross-region device branch of constructDeviceTableScanByTags (the mock's DEVICE_5 and DEVICE_6 are in two regions), not from a region having several devices. The fragment for region 11 scans two devices without a MergeSort.

Suggested change

	// Both levels are split per region; only regions with several devices need a merged stream.
	// Both levels are split per region; only cross-region devices need a merged stream.

[JackieTien97]

Nit: logicalQueryPlan is unused here and on line 362. These two tests only assert fragments, so calling planTester.createPlan(...) without the assignment is enough.

[JackieTien97]

Nit: table5 has only two tags, so there are only two possible orders, and the innermost and outermost levels both use (factory_id, device_id).

Suggested change

	// Three levels list the same PARTITION BY columns in three different orders.
	// Adjacent levels list the same PARTITION BY columns in different orders.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 48.81%. Comparing base (e92ecf1) to head (94d7015).
⚠️ Report is 48 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #19155      +/-   ##
============================================
+ Coverage     48.44%   48.81%   +0.37%     
- Complexity      635      689      +54     
============================================
  Files          5572     5588      +16     
  Lines        411975   413245    +1270     
  Branches      54302    54487     +185     
============================================
+ Hits         199576   201723    +2147     
+ Misses       212399   211522     -877     

☔ 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
B 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