Fix ordering propagation around table and window functions

[JackieTien97]

Description

Queries that order TVF or window output can lose required sorting, fail when aggregating a non-prefix subset of TVF partition keys, or add blocking sorts to already ordered device/time streams. Derive output order from each operator's actual guarantees and retain only proven prefixes. Preserve native scan/merge order for TUMBLE, HOP, CUMULATE, SESSION, VARIATION, and CAPACITY; unknown TVFs retain only a proven partition prefix.

Keep safe sort elimination and streaming sort for common device/time, window, LIMIT, and TopK queries. Do not sort all TopK input rows by ranking fields. Preserve all regional inputs when Group has already been eliminated, and distinguish per-partition rank order from global order. Constant FILL retains order only for unaffected keys. PREVIOUS/NEXT/LINEAR preserve prefixes of FILL_GROUP keys and proven non-null columns; an ascending NULLS LAST time key immediately after all group keys is also preserved. A filled time key may merge peers, so secondary ordering is retained only when separately proven. This also removes redundant sorting after grouped date_bin/date_bin_gapfill aggregation. Sort-elimination restrictions cannot be cleared by another node or sibling.

The change also protects non-TVF queries whose requested streaming prefix is not supplied by a window's actual partitioning, DIFF, or TopK output, and corrects DESC NULL placement in native device ordering. Integration-test tie ordering is explicit, and timestamp/window-index assertions use typed comparisons.

Runtime verification with small TsBlocks also exposed three execution issues: stream sort flushed incomplete groups when a previous group left buffered output; TopK did not propagate merge-input readiness; and TopK compaction changed row positions without copying the underlying block. Fix these without replacing streaming sort with a blocking sort, and add operator regressions for progress and exact retained values.

Grouped NEXT/LINEAR lookahead stops at cached group boundaries. LINEAR resets state before preparing values for a new group, preserving valid lookahead across retries and preventing values from leaking between groups.

Validation

• 80 before/after distributed-plan probes: 30 affected grouped-fill queries lose an unnecessary blocking sort; the other 50 plans are unchanged. Tests assert sort keys, directions, required prefixes, and device order.
• Clean planner/analyzer Maven tests: 290 run, 0 failures, 0 errors, 3 skipped. Two pre-existing CTE PowerMock/JDK initialization failures remain excluded and reproduce with baseline planner classes.
• Final targeted planner/fill-operator Maven tests: 54 passed, including cross-block group boundaries, NULL helpers, and retained linear-fill lookahead.
• Java 17 integration verification covers 91 cases per deployment. The 79 existing window/TVF cases passed under both 1C1D and 1C3D. After matching the existing GapFill IT sort-buffer setting (128 KB) to avoid concurrent admission failures, all 12 cases in the modified ordering IT passed again under both deployments.
• Full-reactor English and Chinese test-compilation passed. Spotless, Checkstyle, and RAT checks passed; pre-existing untracked diagrams were locally excluded from RAT.

This PR has:

• been self-reviewed.
• added comments explaining ordering guarantees and optimization boundaries.
• added or updated unit tests for the changed code paths.
• added integration tests.
• been tested in 1C1D and 1C3D IoTDB clusters.

Key changed classes and packages

TableDistributedPlanGenerator, SortElimination, TableFunctionOrdering, TransformAggregationToStreamable, TransformSortToStreamSort, TableBuiltinTableFunction, and org.apache.iotdb.calc.execution.operator.

[JackieTien97]

Perf regression: ORDER BY <group keys>, time over FILL METHOD PREVIOUS | NEXT | LINEAR ... FILL_GROUP now gets an extra blocking sort above the fill

The non-constant branch of visitFill now drops the fill's ordering unconditionally (this line). With FILL_GROUP, QueryPlanner.fillGroup already sorts the fill input by (group keys, TIME_COLUMN), so on master an outer ORDER BY <group keys>, time is eliminated. With this PR it becomes a full SortNode in the root fragment, which buffers the whole result (and may spill) before emitting anything, for the common "fill per device, return per device" query:

SELECT time, tag1, tag2, tag3, s1 FROM table1
FILL METHOD PREVIOUS TIME_COLUMN 1 FILL_GROUP 2,3,4
ORDER BY tag1, tag2, tag3, time

master (and with the change below)        this PR (20d7b00)
Output                                    Output
  PreviousFill                              Sort [tag1, tag2, tag3, time]   <-- new
    MergeSort [tag1, tag2, tag3, time]        PreviousFill
      DeviceTableScan x 3 regions               MergeSort [tag1, tag2, tag3, time]
                                                  DeviceTableScan x 3 regions

Same pattern for these shapes (distributed plans from PlanTester, 3 regions):

Shape	master	this PR	with the change below
LINEAR TIME_COLUMN 1 FILL_GROUP 2,3,4 or NEXT TIME_COLUMN 1 FILL_GROUP 2,3,4, ORDER BY tag1, tag2, tag3, time	no sort above the fill	+ full Sort	as master
PREVIOUS TIME_BOUND 1h TIME_COLUMN 1 FILL_GROUP 2,3,4, same ORDER BY	no sort above the fill	+ full Sort	as master
PREVIOUS TIME_COLUMN 1 FILL_GROUP 2,3,4 ORDER BY tag1, tag2, tag3	no sort above the fill	+ full Sort	as master
SELECT time, tag1, s1 ... FILL_GROUP 2 ORDER BY tag1, time	no sort above the fill	+ full Sort	as master
PREVIOUS ... FILL_GROUP 2,3,4 over HOP(DATA => table1, ...)	no sort above the fill	+ full Sort	as master
PREVIOUS ... FILL_GROUP 2,3,4 with lag(s1) OVER (PARTITION BY tag1, tag2, tag3 ORDER BY time) in the select list	StreamSort below the fill	+ full Sort above the fill	no sort at all
date_bin_gapfill(...) or date_bin(...) aggregation + FILL ... FILL_GROUP 2,3,4 ORDER BY tag1, tag2, tag3, t	1 Sort (below the fill)	2 Sorts	still 2 Sorts (see the end)

Dropping the ordering is right for sort keys the fill can rewrite, and it fixes a real master bug: master eliminates the outer sort in SELECT * FROM (SELECT * FROM table1 ORDER BY s1, s2 LIMIT 10) FILL METHOD PREVIOUS ORDER BY s1, s2 (also with NEXT / LINEAR), although PREVIOUS turns (5, 1), (5, 9), (NULL, 2) into (5, 1), (5, 9), (5, 2). But part of the child ordering provably survives a PREVIOUS / NEXT / LINEAR fill:

• the fill only writes NULL positions and emits rows in input order;
• it resets at every FILL_GROUP boundary, and all rows of a group share their group-key values, so a group key is never rewritten (a NULL key has no non-NULL value in its group to copy);
• non-NULL columns (native time, tags / attributes that are non-null for every device) are never rewritten; getNonNullSymbols already proves this for the constant fill.

So the fill can keep the prefix of its child's ordering that consists of FILL_GROUP keys and provably non-null columns. As in the constant-fill branch, getNonNullSymbols (which may read the coordinator's spilled device entries) only runs when an ancestor requests an ordering; the group keys alone need no metadata.

--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
@@ -52,11 +52,14 @@
 import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.GroupNode;
 import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.JoinNode;
 import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.LimitNode;
+import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.LinearFillNode;
 import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.MarkDistinctNode;
 import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.MergeSortNode;
+import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.NextFillNode;
 import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.OffsetNode;
 import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.OutputNode;
 import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.PatternRecognitionNode;
+import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.PreviousFillNode;
 import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.ProjectNode;
 import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.RowNumberNode;
 import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.SemiJoinNode;
@@ -323,9 +326,24 @@
   @Override
   public List<PlanNode> visitFill(FillNode node, PlanContext context) {
     if (!(node instanceof ValueFillNode)) {
+      boolean hasExpectedOutputOrdering = context.hasSortProperty;
       context.clearExpectedOrderingScheme();
+      // PREVIOUS/NEXT/LINEAR fill only replaces NULLs, keeps the row order and resets at every
+      // FILL_GROUP boundary. Group keys are constant within a group and non-null columns are never
+      // rewritten, so a child ordering prefix made of such columns still holds after the fill.
+      Set<Symbol> unchanged = new HashSet<>(getFillGroupingKeys(node));
+      if (hasExpectedOutputOrdering) {
+        unchanged.addAll(getNonNullSymbols(node.getChild()));
+      }
       List<PlanNode> result = dealWithPlainSingleChildNode(node, context);
-      nodeOrderingMap.remove(node.getPlanNodeId());
+      OrderingScheme ordering =
+          TableFunctionOrdering.retainOrderingPrefix(
+              nodeOrderingMap.get(node.getPlanNodeId()), unchanged);
+      if (ordering == null) {
+        nodeOrderingMap.remove(node.getPlanNodeId());
+      } else {
+        nodeOrderingMap.put(node.getPlanNodeId(), ordering);
+      }
       return result;
     }
     // Inspect metadata before distribution consumes a spilled device data set. No data rows need
@@ -354,6 +372,18 @@
     return Collections.singletonList(node);
   }
 
+  private static List<Symbol> getFillGroupingKeys(FillNode node) {
+    Optional<List<Symbol>> groupingKeys = Optional.empty();
+    if (node instanceof PreviousFillNode) {
+      groupingKeys = ((PreviousFillNode) node).getGroupingKeys();
+    } else if (node instanceof NextFillNode) {
+      groupingKeys = ((NextFillNode) node).getGroupingKeys();
+    } else if (node instanceof LinearFillNode) {
+      groupingKeys = ((LinearFillNode) node).getGroupingKeys();
+    }
+    return groupingKeys.orElse(Collections.emptyList());
+  }
+
   private Set<Symbol> getNonNullSymbols(PlanNode node) {
     Set<Symbol> cached = nonNullSymbols.get(node.getPlanNodeId());
     if (cached != null) {

Regression tests (OrderingPropertyPropagationTest)

--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/OrderingPropertyPropagationTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/OrderingPropertyPropagationTest.java
@@ -23,6 +23,7 @@
 import org.apache.iotdb.commons.queryengine.plan.relational.planner.Symbol;
 import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.AggregationNode;
 import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.CollectNode;
+import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.FillNode;
 import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.PatternRecognitionNode;
 import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.RowNumberNode;
 import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.SortNode;
@@ -296,6 +297,48 @@
   }
 
   @Test
+  public void testFillGroupKeepsGroupKeyAndTimeOrdering() {
+    String window = ", lag(s1) OVER (PARTITION BY " + TAGS + " ORDER BY time) AS lg";
+    for (String extraColumn : List.of("", window)) {
+      for (String method :
+          List.of(
+              "PREVIOUS TIME_COLUMN 1",
+              "PREVIOUS TIME_BOUND 1h TIME_COLUMN 1",
+              "NEXT TIME_COLUMN 1",
+              "LINEAR TIME_COLUMN 1")) {
+        for (String[] group : new String[][] {{"2,3,4", TAGS}, {"2", "tag1"}}) {
+          String sql =
+              "SELECT time, "
+                  + TAGS
+                  + ", s1"
+                  + extraColumn
+                  + " FROM table1 FILL METHOD "
+                  + method
+                  + " FILL_GROUP "
+                  + group[0]
+                  + " ORDER BY "
+                  + group[1]
+                  + ", time";
+          PlanTester tester = plan(sql);
+          assertFalse(sql, findNodes(tester, FillNode.class).isEmpty());
+          // FILL_GROUP already orders the input by its keys and time, and the fill changes neither.
+          assertTrue(sql, sortsAbove(tester, FillNode.class).isEmpty());
+        }
+      }
+    }
+  }
+
+  @Test
+  public void testFillKeepsSortOnFilledNullableKeys() {
+    // PREVIOUS can turn (5, 1), (5, 9), (NULL, 2) into (5, 1), (5, 9), (5, 2), so an ordering on
+    // nullable columns that are not FILL_GROUP keys must not survive the fill.
+    String sql =
+        "SELECT * FROM (SELECT * FROM table1 ORDER BY s1, s2 LIMIT 10)"
+            + " FILL METHOD PREVIOUS ORDER BY s1, s2";
+    assertFalse(sql, sortsAbove(plan(sql), FillNode.class).isEmpty());
+  }
+
+  @Test
   public void testUnorderedQueriesKeepUnorderedCollection() {
     for (String sql :
         List.of(

Checked locally on top of 20d7b00:

• testFillGroupKeepsGroupKeyAndTimeOrdering fails on 20d7b00 (Sort above the fill) and passes with the change. testFillKeepsSortOnFilledNullableKeys passes on both, so the master bug above stays fixed.
• Of 149 distributed-plan probes (TVF / window / TopK / fill / aggregation / LIMIT / JOIN / UNION / single-device shapes), only the 8 FILL_GROUP shapes above that the change fixes differ from 20d7b00; the other 141 plans are identical.
• All unit tests under org/apache/iotdb/db/queryengine/plan/relational pass (493 run) except CteMaterializerTest / CteSubqueryTest, which error locally on JDK 21 with or without the change.

Not covered: gapfill + FILL_GROUP and date_bin aggregation + FILL_GROUP still sort twice. There the bucket column comes out of an aggregation, which getNonNullSymbols does not model, and AbstractGapFillOperator explicitly passes NULL buckets through, so I did not assume it is non-NULL. Covering these needs a separate argument about how NULLs in the key right after the group keys get filled; I'd leave that for a follow-up.