[Bug] Table model: wrong results when an outer PARTITION BY is a prefix of the inner PARTITION BY / GROUP BY keys #19207

Search before asking

• I searched in the issues and found nothing similar.

Version

• Apache IoTDB 2.0.11-SNAPSHOT, master at commit a4568b6e90c984ecbf7531679473eaf6ae983b7e.
• 1 ConfigNode + 1 DataNode, started from the binary distribution built from that commit. macOS 27.0.1, OpenJDK 21.0.2, 10 cores.
• I have not tested released versions. The check involved was added together with GroupNode in 7c30113 (April 2025).

Describe the bug and provide the minimal reproduce step

In the table model, a query silently returns wrong results when an outer operation partitions by columns that are a leading prefix, in the same order, of the keys its input is already partitioned or grouped by. For example, the outer operation has PARTITION BY (a, b) and its input comes from PARTITION BY (a, b, c) or GROUP BY a, b, c. The outer operation can be a set-semantic table function, a window function, a row_number() top-N filter or MATCH_RECOGNIZE.

ParallelizeGrouping treats such input as already clustered by (a, b) and keeps the GroupNode instead of turning it into a full sort. That is only true when the input is a single stream sorted by (a, b, c). It goes wrong in two situations:

1. The database has more than one DataRegion. The outer operation is planned once per region, so every region computes its own copy of each outer partition.
2. The input is a hash aggregation. Its output is not clustered by the prefix, so a partition is cut into pieces and rows with different keys are processed as one partition.

If the same outer keys are written in a different order, for example PARTITION BY (b, a), the result is correct, because that case falls back to a full sort (#19155).

Running the SQL. The CLI executes its input line by line, so each statement below is on one line. I used ./sbin/start-cli.sh -sql_dialect table -disableISO8601, which prints timestamps as numbers.

Case 1: more than one DataRegion

To get exactly the numbers shown below, set this in conf/iotdb-system.properties before the first start. It gives the database 4 DataRegions:

data_region_group_extension_policy=CUSTOM
default_data_region_group_num_per_database=4

The default configuration reproduces it too. On my machine the database then got 5 DataRegions (check with SHOW REGIONS), and Q1 returned 5 windows of 8 rows each.

Setup: 40 devices that share a and b, one row each at time 1 to 40.

CREATE DATABASE repro;
USE repro;
CREATE TABLE t(a STRING TAG, b STRING TAG, c STRING TAG, s1 DOUBLE FIELD);
INSERT INTO t(time, a, b, c, s1) VALUES (1, 'A', 'B', 'c00', 1.0), (2, 'A', 'B', 'c01', 1.0), (3, 'A', 'B', 'c02', 1.0), (4, 'A', 'B', 'c03', 1.0), (5, 'A', 'B', 'c04', 1.0), (6, 'A', 'B', 'c05', 1.0), (7, 'A', 'B', 'c06', 1.0), (8, 'A', 'B', 'c07', 1.0), (9, 'A', 'B', 'c08', 1.0), (10, 'A', 'B', 'c09', 1.0), (11, 'A', 'B', 'c10', 1.0), (12, 'A', 'B', 'c11', 1.0), (13, 'A', 'B', 'c12', 1.0), (14, 'A', 'B', 'c13', 1.0), (15, 'A', 'B', 'c14', 1.0), (16, 'A', 'B', 'c15', 1.0), (17, 'A', 'B', 'c16', 1.0), (18, 'A', 'B', 'c17', 1.0), (19, 'A', 'B', 'c18', 1.0), (20, 'A', 'B', 'c19', 1.0), (21, 'A', 'B', 'c20', 1.0), (22, 'A', 'B', 'c21', 1.0), (23, 'A', 'B', 'c22', 1.0), (24, 'A', 'B', 'c23', 1.0), (25, 'A', 'B', 'c24', 1.0), (26, 'A', 'B', 'c25', 1.0), (27, 'A', 'B', 'c26', 1.0), (28, 'A', 'B', 'c27', 1.0), (29, 'A', 'B', 'c28', 1.0), (30, 'A', 'B', 'c29', 1.0), (31, 'A', 'B', 'c30', 1.0), (32, 'A', 'B', 'c31', 1.0), (33, 'A', 'B', 'c32', 1.0), (34, 'A', 'B', 'c33', 1.0), (35, 'A', 'B', 'c34', 1.0), (36, 'A', 'B', 'c35', 1.0), (37, 'A', 'B', 'c36', 1.0), (38, 'A', 'B', 'c37', 1.0), (39, 'A', 'B', 'c38', 1.0), (40, 'A', 'B', 'c39', 1.0);

Q1. SESSION partitioned by (a, b) over SESSION partitioned by (a, b, c):

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

Q2. Window function over (a, b) on top of a window function over (a, b, c):

SELECT cnt, count(*) AS n FROM (SELECT count(inner_sum) OVER (PARTITION BY a, b) AS cnt FROM (SELECT a, b, c, sum(s1) OVER (PARTITION BY a, b, c ORDER BY time) AS inner_sum FROM t)) GROUP BY cnt ORDER BY cnt;

Q3. Top 2 rows per (a, b) with row_number(), on top of row_number() over (a, b, c):

SELECT time, c, rn2 FROM (SELECT *, row_number() OVER (PARTITION BY a, b ORDER BY time) AS rn2 FROM (SELECT time, a, b, c, s1, row_number() OVER (PARTITION BY a, b, c ORDER BY time) AS rn1 FROM t)) WHERE rn2 <= 2 ORDER BY time;

Q4. MATCH_RECOGNIZE partitioned by (a, b) over SESSION partitioned by (a, b, c):

SELECT * FROM (SELECT time, a, b, c, s1 FROM SESSION(DATA => t PARTITION BY (a, b, c) ORDER BY time, GAP => 1ms)) MATCH_RECOGNIZE (PARTITION BY a, b ORDER BY time MEASURES RPR_FIRST(A.time) AS first_t, RPR_LAST(A.time) AS last_t, COUNT(A.s1) AS cnt ONE ROW PER MATCH PATTERN (A+) DEFINE A AS true) ORDER BY first_t;

Q5. SESSION partitioned by (a, b) over GROUP BY a, b, c:

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

Case 2: hash aggregation

One device with 6000 rows. s1 takes the values 0, 1 and 2 in a pseudo-random sequence (1980, 2013 and 2007 rows), and s3 is unique. This case does not depend on the number of DataRegions.

CREATE TABLE repro.t2(d STRING TAG, s1 INT32 FIELD, s3 INT32 FIELD);

Save this as gen_t2_data.py:

for start in range(0, 6000, 1000):
    rows = []
    for row in range(start, start + 1000):
        s1 = (((row * 0x9E3779B1) & 0xFFFFFFFF) >> 20) % 3  # pseudo-random 0/1/2
        rows.append(f"({row + 1}, 'd1', {s1}, {row})")
    print("INSERT INTO repro.t2(time, d, s1, s3) VALUES " + ", ".join(rows) + ";")

Then generate and load the data:

python3 gen_t2_data.py > t2_data.sql
./sbin/start-cli.sh -sql_dialect table < t2_data.sql

Q6. SESSION partitioned by (s1) over GROUP BY s1, s3:

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

Q7. Window function over (s1) on top of GROUP BY s1, s3:

SELECT s1, cnt, count(*) AS n FROM (SELECT s1, count(*) OVER (PARTITION BY s1) AS cnt FROM (SELECT s1, s3 FROM t2 GROUP BY s1, s3)) GROUP BY s1, cnt ORDER BY s1, cnt;

What did you expect to see?

All 40 devices in t share (a, b), so there is exactly one outer partition:

In t2, every s1 value is one partition with one session window:

What did you see instead?

Case 1, with 4 DataRegions

Every region computes the outer operation on its own 10 devices.

Q1 returns 4 windows instead of 1:

+------------+----------+---+
|window_start|window_end|cnt|
+------------+----------+---+
|           1|        37| 10|
|           2|        38| 10|
|           3|        39| 10|
|           4|        40| 10|
+------------+----------+---+
Total line number = 4

Q2 counts 10 instead of 40:

+---+--+
|cnt| n|
+---+--+
| 10|40|
+---+--+
Total line number = 1

Q3 returns the top 2 of every region, 8 rows instead of 2:

+----+---+---+
|time|  c|rn2|
+----+---+---+
|   1|c00|  1|
|   2|c01|  1|
|   3|c02|  1|
|   4|c03|  1|
|   5|c04|  2|
|   6|c05|  2|
|   8|c07|  2|
|  12|c11|  2|
+----+---+---+
Total line number = 8

Q4 returns 4 matches instead of 1:

+-+-+-------+------+---+
|a|b|first_t|last_t|cnt|
+-+-+-------+------+---+
|A|B|      1|    37| 10|
|A|B|      2|    38| 10|
|A|B|      3|    39| 10|
|A|B|      4|    40| 10|
+-+-+-------+------+---+
Total line number = 4

Q5 returns the same 4 rows as Q1.

As a control, Q1 with the outer keys written as PARTITION BY (b, a) returns the expected single row (1, 40, 40). With a single DataRegion (the same settings with default_data_region_group_num_per_database=1), Q1 to Q5 are all correct.

Case 2

Q6 returns 6 windows instead of 3, and none of the counts is right:

+--+------------+----------+----+
|s1|window_start|window_end| cnt|
+--+------------+----------+----+
| 0|           1|      1998| 674|
| 0|        2000|      5997|3310|
| 1|           5|      1997| 654|
| 1|        5000|      5999| 348|
| 2|           2|      1999| 671|
| 2|        5002|      6000| 343|
+--+------------+----------+----+
Total line number = 6

Q7 returns 8 rows instead of 3. Rows of all three s1 values get the count 3310:

+--+----+----+
|s1| cnt|   n|
+--+----+----+
| 0| 674| 674|
| 0|3310|1306|
| 1| 348| 348|
| 1| 654| 654|
| 1|3310|1011|
| 2| 343| 343|
| 2| 671| 671|
| 2|3310| 993|
+--+----+----+
Total line number = 8

Reading the output of the SESSION call in Q6 row by row, without the outer GROUP BY, gives 15 runs of s1. The first three runs (674, 654 and 671 rows) are partitions with a single s1 value each. The next ten runs, 3310 rows that contain all three s1 values, share one window [2000, 5997], so they were processed as a single partition. The last two runs (348 and 343 rows) are separate partitions again.

The exact numbers depend on where batch boundaries fall. With only 600 rows in t2, both queries are correct.

Anything else?

Cause. ParallelizeGrouping.checkPrefixMatch keeps the GroupNode when its partition keys equal the first entries of the child's key list (requiredPrefix.equals(actualPrefix)). That is sound when the list is a sort order (the Sort, StreamSort and TopK call sites), because one sorted stream is clustered by every prefix. But five call sites pass partition or grouping keys instead: the inner GroupNode (L169), a table function's PARTITION BY (L233), MATCH_RECOGNIZE (L254) and the GROUP BY keys of an aggregation (L313, L319). Being partitioned by (a, b, c) says nothing about (a, b). The class Javadoc even lists GroupNode[PK={tag1}, OK={}] -> TableFunctionNode[partition={tag1,tag2}] as a valid case (L76).

Case 1. The inner partition keys contain every TAG column, so the inner side runs per region. TableDistributedPlanGenerator.visitGroup keeps one group per child stream, and the consumer above it is cloned per child with splitForEachChild. EXPLAIN of Q1 shows 8 TableFunctionProcessor nodes (outer and inner for each of the 4 regions) under one Collect.

Case 2. The GroupNode runs as TableStreamSortOperator, which assumes that rows with the same partition key are contiguous. It sorts and emits its buffer when the last row of an incoming block has a different key. The output of the hash aggregation has 4935 runs of s1 over its 6000 rows, so the operator emits five separately sorted batches, each containing the keys 0, 1 and 2. Downstream, PartitionRecognizer.findNextDifferentRowIndex makes the same assumption: when the last row of a block has the current partition key, the whole block is taken as that partition.

When it does not happen (all checked on the same commit):

• Case 1 with a single DataRegion.
• The prefix still contains every TAG column, for example outer (a, b) over inner (a, b, attr).
• The inner side is one sorted stream, for example a subquery with ORDER BY a, b, c, time LIMIT n.
• The partition keys are the TAG columns that the aggregation streams on, for example GROUP BY a, b, s1 with PARTITION BY (a, b).
• The inner side itself runs as a single instance, for example outer (a) over inner (a, b) on a table with three TAG columns.

About a fix. In an experiment, requiring the exact key set for partition and grouping keys fixed the SESSION, window-function and aggregation cases I tried, but it failed WindowFunctionOptimizationTest.testSwapWindowFunctions. In that test the inner GroupNode becomes a Sort, so its output is one sorted stream and the prefix is safe. The check probably needs to know whether the child ends up as one sorted stream, or as several streams or a hash table that are only clustered by the full key set.

Not tested: 1C3D, devices whose data spans several regions, and table functions other than SESSION.

Are you willing to submit a PR?

• I'm willing to submit a PR!