[FLINK-25802][FLINK-30499][table-planner] Fix TIMESTAMP codegen for RANGE OVER window bounds

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/over/RangeBoundComparatorCodeGenerator.scala

@@ -23,7 +23,7 @@ import org.apache.flink.table.planner.codegen.{CodeGeneratorContext, CodeGenUtil
 import org.apache.flink.table.planner.codegen.CodeGenUtils.{newName, ROW_DATA}
 import org.apache.flink.table.planner.codegen.Indenter.toISC
 import org.apache.flink.table.runtime.generated.{GeneratedRecordComparator, RecordComparator}
-import org.apache.flink.table.types.logical.{BigIntType, IntType, LogicalType, LogicalTypeRoot, RowType}
+import org.apache.flink.table.types.logical.{BigIntType, IntType, LogicalType, LogicalTypeFamily, LogicalTypeRoot, RowType}
 
 import org.apache.calcite.avatica.util.DateTimeUtils
 import org.apache.calcite.rex.{RexInputRef, RexWindowBound}
@@ -135,16 +135,32 @@ class RangeBoundComparatorCodeGenerator(
       inputValue: String,
       currentValue: String,
       parentCtx: CodeGeneratorContext): String = {
-    val (realBoundValue, realKeyType) = keyType.getTypeRoot match {
-      case LogicalTypeRoot.DATE =>
-        // The constant about time is expressed based millisecond unit in calcite, but
-        // the field about date is expressed based day unit. So here should keep the same unit for
-        // comparator.
+    val (realBoundValue, realKeyType) =
+      if (keyType.is(LogicalTypeFamily.TIMESTAMP)) {
+        // Scale millis bound to microseconds to preserve sub-millisecond TIMESTAMP(n>3) precision
+        val microsBound = bound match {
+          case l: Long => l * 1000L
+          case bg: BigDecimal => bg.multiply(BigDecimal.valueOf(1000))
+        }
+        (microsBound, new BigIntType())
+      } else if (keyType.is(LogicalTypeFamily.TIME)) {
+        (bound, new IntType())
+      } else if (keyType.is(LogicalTypeRoot.DATE)) {
+        // Calcite bound is in millis; DATE field is in days
         (bound.asInstanceOf[Long] / DateTimeUtils.MILLIS_PER_DAY, new IntType())
-      case LogicalTypeRoot.TIME_WITHOUT_TIME_ZONE => (bound, new IntType())
-      case LogicalTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE => (bound, new BigIntType())
-      case _ => (bound, keyType)
-    }
+      } else {
+        (bound, keyType)
+      }
+
+    val (realInputValue, realCurrentValue) =
+      if (keyType.is(LogicalTypeFamily.TIMESTAMP)) {
+        // Epoch microseconds: millis * 1000 + nanoOfMillis / 1000 preserves TIMESTAMP(6) precision
+        (
+          s"($inputValue.getMillisecond() * 1000L + $inputValue.getNanoOfMillisecond() / 1000)",
+          s"($currentValue.getMillisecond() * 1000L + $currentValue.getNanoOfMillisecond() / 1000)")
+      } else {
+        (inputValue, currentValue)
+      }
 
     val typeFactory = relBuilder.getTypeFactory.asInstanceOf[FlinkTypeFactory]
     val relKeyType = typeFactory.createFieldTypeFromLogicalType(realKeyType)
@@ -157,7 +173,9 @@ class RangeBoundComparatorCodeGenerator(
     } else {
       relBuilder.call(MINUS, new RexInputRef(1, relKeyType), new RexInputRef(0, relKeyType))
     }
-    exprCodeGenerator.bindInput(realKeyType, inputValue).bindSecondInput(realKeyType, currentValue)
+    exprCodeGenerator
+      .bindInput(realKeyType, realInputValue)
+      .bindSecondInput(realKeyType, realCurrentValue)
     val literal = relBuilder.literal(realBoundValue)
 
     // In order to avoid the loss of precision in long cast to int.
@@ -169,8 +187,12 @@ class RangeBoundComparatorCodeGenerator(
 
     val comExpr = exprCodeGenerator.generateExpression(comCall)
 
+    val childMemberCode = ctx.reuseMemberCode()
+    if (childMemberCode.nonEmpty) {
+      parentCtx.addReusableMember(childMemberCode)
+    }
+
     j"""
-       ${ctx.reuseMemberCode()}
        ${ctx.reuseLocalVariableCode()}
        ${ctx.reuseInputUnboxingCode()}
        ${comExpr.code}

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/OverAggregateBatchRestoreTest.java

@@ -35,10 +35,11 @@ public OverAggregateBatchRestoreTest() {
     @Override
     public List<TableTestProgram> programs() {
         return Arrays.asList(
-                // These tests fail due to FLINK-25802
-                // OverAggregateTestPrograms.OVER_AGGREGATE_TIME_BOUNDED_PARTITIONED_ROWS,
-                // OverAggregateTestPrograms.OVER_AGGREGATE_TIME_BOUNDED_NON_PARTITIONED_ROWS
+                OverAggregateTestPrograms.OVER_AGGREGATE_TIME_BOUNDED_PARTITIONED_ROWS,
+                OverAggregateTestPrograms.OVER_AGGREGATE_TIME_BOUNDED_NON_PARTITIONED_ROWS,
                 OverAggregateTestPrograms.OVER_AGGREGATE_UNBOUNDED_PARTITIONED_ROWS,
-                OverAggregateTestPrograms.OVER_AGGREGATE_ROW_BOUNDED_PARTITIONED_PRECEDING_ROWS);
+                OverAggregateTestPrograms.OVER_AGGREGATE_ROW_BOUNDED_PARTITIONED_PRECEDING_ROWS,
+                OverAggregateTestPrograms.OVER_AGGREGATE_RANGE_TIMESTAMP,
+                OverAggregateTestPrograms.OVER_AGGREGATE_RANGE_TIMESTAMP_LTZ);
     }
 }

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/OverAggregateTestPrograms.java

@@ -25,6 +25,9 @@
 import org.apache.flink.table.test.program.TableTestProgram;
 import org.apache.flink.types.Row;
 
+import java.time.Instant;
+import java.time.LocalDateTime;
+
 import static org.apache.flink.table.api.config.TableConfigOptions.LOCAL_TIME_ZONE;
 
 /**
@@ -383,6 +386,66 @@ public class OverAggregateTestPrograms {
                             BEFORE_RESTORE_DATA_PRECEDING_ROWS_WITH_OUT_OF_ORDER_RECORDS,
                             AFTER_RESTORE_DATA_PRECEDING_ROWS_WITH_OUT_OF_ORDER_RECORDS);
 
+    /**
+     * Regression test for FLINK-25802: RANGE OVER with TIMESTAMP ORDER BY. Uses TIMESTAMP(6) with a
+     * sub-millisecond row to verify that the range comparator uses microsecond precision: row 3 is
+     * 10s+600µs after row 1, so row 1 falls just outside row 3's 10-second window (count=2, not 3).
+     */
+    public static final TableTestProgram OVER_AGGREGATE_RANGE_TIMESTAMP =
+            TableTestProgram.of(
+                            "over-aggregate-batch-range-timestamp",
+                            "RANGE OVER with TIMESTAMP(6) ORDER BY")
+                    .setupTableSource(
+                            SourceTestStep.newBuilder("timestamp_range_source")
+                                    .addSchema("ts TIMESTAMP(6)", "val INT")
+                                    .producedValues(
+                                            Row.of(LocalDateTime.of(2021, 1, 1, 0, 0, 0, 0), 1),
+                                            Row.of(LocalDateTime.of(2021, 1, 1, 0, 0, 5, 0), 2),
+                                            Row.of(
+                                                    LocalDateTime.of(2021, 1, 1, 0, 0, 10, 600_000),
+                                                    3))
+                                    .build())
+                    .setupTableSink(
+                            SinkTestStep.newBuilder("timestamp_range_sink")
+                                    .addSchema("val INT", "cnt BIGINT")
+                                    .consumedValues(Row.of(1, 1L), Row.of(2, 2L), Row.of(3, 2L))
+                                    .build())
+                    .runSql(
+                            "INSERT INTO timestamp_range_sink"
+                                    + " SELECT val, COUNT(val) OVER (ORDER BY ts"
+                                    + " RANGE BETWEEN INTERVAL '10' SECOND PRECEDING"
+                                    + " AND CURRENT ROW)"
+                                    + " FROM timestamp_range_source")
+                    .build();
+
+    /**
+     * Regression test for FLINK-30499: RANGE OVER with TIMESTAMP_LTZ(3) ORDER BY failed to compile.
+     */
+    public static final TableTestProgram OVER_AGGREGATE_RANGE_TIMESTAMP_LTZ =
+            TableTestProgram.of(
+                            "over-aggregate-batch-range-timestamp-ltz",
+                            "RANGE OVER with TIMESTAMP_LTZ(3) ORDER BY")
+                    .setupTableSource(
+                            SourceTestStep.newBuilder("timestamp_ltz_range_source")
+                                    .addSchema("ts TIMESTAMP_LTZ(3)", "val INT")
+                                    .producedValues(
+                                            Row.of(Instant.parse("2021-01-01T00:00:00Z"), 1),
+                                            Row.of(Instant.parse("2021-01-01T00:00:05Z"), 2),
+                                            Row.of(Instant.parse("2021-01-01T00:00:12Z"), 3))
+                                    .build())
+                    .setupTableSink(
+                            SinkTestStep.newBuilder("timestamp_ltz_range_sink")
+                                    .addSchema("val INT", "cnt BIGINT")
+                                    .consumedValues(Row.of(1, 1L), Row.of(2, 2L), Row.of(3, 2L))
+                                    .build())
+                    .runSql(
+                            "INSERT INTO timestamp_ltz_range_sink"
+                                    + " SELECT val, COUNT(val) OVER (ORDER BY ts"
+                                    + " RANGE BETWEEN INTERVAL '10' SECOND PRECEDING"
+                                    + " AND CURRENT ROW)"
+                                    + " FROM timestamp_ltz_range_source")
+                    .build();
+
     private static SourceTestStep getSourceTestStep(Row[] data, Row[] afterData) {
         return SourceTestStep.newBuilder("MyTable")
                 .addSchema(

flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-over-aggregate_1/over-aggregate-batch-range-timestamp-ltz/plan/over-aggregate-batch-range-timestamp-ltz.json

@@ -0,0 +1,214 @@
+{
+  "flinkVersion" : "2.2",
+  "nodes" : [ {
+    "id" : 25,
+    "type" : "batch-exec-table-source-scan_1",
+    "scanTableSource" : {
+      "table" : {
+        "identifier" : "`default_catalog`.`default_database`.`timestamp_ltz_range_source`",
+        "resolvedTable" : {
+          "schema" : {
+            "columns" : [ {
+              "name" : "ts",
+              "dataType" : "TIMESTAMP(3) WITH LOCAL TIME ZONE"
+            }, {
+              "name" : "val",
+              "dataType" : "INT"
+            } ]
+          }
+        }
+      }
+    },
+    "outputType" : "ROW<`ts` TIMESTAMP(3) WITH LOCAL TIME ZONE, `val` INT>",
+    "description" : "TableSourceScan(table=[[default_catalog, default_database, timestamp_ltz_range_source]], fields=[ts, val])",
+    "dynamicFilteringDataListenerID" : "eea9acbe-3cde-4c2e-8074-adb76b76e51c"
+  }, {
+    "id" : 26,
+    "type" : "batch-exec-exchange_1",
+    "inputProperties" : [ {
+      "requiredDistribution" : {
+        "type" : "SINGLETON"
+      },
+      "damBehavior" : "BLOCKING",
+      "priority" : 0
+    } ],
+    "outputType" : "ROW<`ts` TIMESTAMP(3) WITH LOCAL TIME ZONE, `val` INT>",
+    "description" : "Exchange(distribution=[single])",
+    "requiredExchangeMode" : "UNDEFINED"
+  }, {
+    "id" : 27,
+    "type" : "batch-exec-sort_1",
+    "configuration" : {
+      "table.exec.resource.sort.memory" : "128 mb",
+      "table.exec.sort.async-merge-enabled" : "true",
+      "table.exec.sort.max-num-file-handles" : "128",
+      "table.exec.spill-compression.block-size" : "64 kb",
+      "table.exec.spill-compression.enabled" : "true"
+    },
+    "sortSpec" : {
+      "fields" : [ {
+        "index" : 0,
+        "isAscending" : true,
+        "nullIsLast" : false
+      } ]
+    },
+    "inputProperties" : [ {
+      "requiredDistribution" : {
+        "type" : "UNKNOWN"
+      },
+      "damBehavior" : "END_INPUT",
+      "priority" : 0
+    } ],
+    "outputType" : "ROW<`ts` TIMESTAMP(3) WITH LOCAL TIME ZONE, `val` INT>",
+    "description" : "Sort(orderBy=[ts ASC])"
+  }, {
+    "id" : 28,
+    "type" : "batch-exec-over-aggregate_1",
+    "configuration" : {
+      "table.exec.resource.external-buffer-memory" : "10 mb"
+    },
+    "overSpec" : {
+      "partition" : {
+        "fields" : [ ]
+      },
+      "groups" : [ {
+        "orderBy" : {
+          "fields" : [ {
+            "index" : 0,
+            "isAscending" : true,
+            "nullIsLast" : false
+          } ]
+        },
+        "isRows" : false,
+        "lowerBound" : {
+          "kind" : "BOUNDED_WINDOW",
+          "isPreceding" : true,
+          "offset" : {
+            "kind" : "INPUT_REF",
+            "inputIndex" : 2,
+            "type" : "INTERVAL SECOND(6) NOT NULL"
+          }
+        },
+        "upperBound" : {
+          "kind" : "CURRENT_ROW"
+        },
+        "aggCalls" : [ {
+          "name" : "w0$o0",
+          "syntax" : "FUNCTION_STAR",
+          "internalName" : "$COUNT$1",
+          "argList" : [ 1 ],
+          "filterArg" : -1,
+          "distinct" : false,
+          "approximate" : false,
+          "ignoreNulls" : false,
+          "type" : "BIGINT NOT NULL"
+        } ]
+      } ],
+      "constants" : [ {
+        "kind" : "LITERAL",
+        "value" : "10000",
+        "type" : "INTERVAL SECOND(6) NOT NULL"
+      } ],
+      "originalInputFields" : 2
+    },
+    "inputProperties" : [ {
+      "requiredDistribution" : {
+        "type" : "SINGLETON"
+      },
+      "damBehavior" : "PIPELINED",
+      "priority" : 0
+    } ],
+    "outputType" : "ROW<`ts` TIMESTAMP(3) WITH LOCAL TIME ZONE, `val` INT, `w0$o0` BIGINT NOT NULL>",
+    "description" : "OverAggregate(orderBy=[ts ASC], window#0=[COUNT(val) AS w0$o0 RANGE BETWEEN 10000 PRECEDING AND CURRENT ROW], select=[ts, val, w0$o0])"
+  }, {
+    "id" : 29,
+    "type" : "batch-exec-calc_1",
+    "projection" : [ {
+      "kind" : "INPUT_REF",
+      "inputIndex" : 1,
+      "type" : "INT"
+    }, {
+      "kind" : "INPUT_REF",
+      "inputIndex" : 2,
+      "type" : "BIGINT NOT NULL"
+    } ],
+    "condition" : null,
+    "inputProperties" : [ {
+      "requiredDistribution" : {
+        "type" : "UNKNOWN"
+      },
+      "damBehavior" : "PIPELINED",
+      "priority" : 0
+    } ],
+    "outputType" : "ROW<`val` INT, `$1` BIGINT NOT NULL>",
+    "description" : "Calc(select=[val, w0$o0 AS $1])"
+  }, {
+    "id" : 30,
+    "type" : "batch-exec-sink_1",
+    "configuration" : {
+      "table.exec.sink.not-null-enforcer" : "ERROR",
+      "table.exec.sink.type-length-enforcer" : "IGNORE"
+    },
+    "dynamicTableSink" : {
+      "table" : {
+        "identifier" : "`default_catalog`.`default_database`.`timestamp_ltz_range_sink`",
+        "resolvedTable" : {
+          "schema" : {
+            "columns" : [ {
+              "name" : "val",
+              "dataType" : "INT"
+            }, {
+              "name" : "cnt",
+              "dataType" : "BIGINT"
+            } ]
+          }
+        }
+      }
+    },
+    "inputProperties" : [ {
+      "requiredDistribution" : {
+        "type" : "UNKNOWN"
+      },
+      "damBehavior" : "BLOCKING",
+      "priority" : 0
+    } ],
+    "outputType" : "ROW<`val` INT, `$1` BIGINT NOT NULL>",
+    "description" : "Sink(table=[default_catalog.default_database.timestamp_ltz_range_sink], fields=[val, $1])"
+  } ],
+  "edges" : [ {
+    "source" : 25,
+    "target" : 26,
+    "shuffle" : {
+      "type" : "FORWARD"
+    },
+    "shuffleMode" : "PIPELINED"
+  }, {
+    "source" : 26,
+    "target" : 27,
+    "shuffle" : {
+      "type" : "FORWARD"
+    },
+    "shuffleMode" : "PIPELINED"
+  }, {
+    "source" : 27,
+    "target" : 28,
+    "shuffle" : {
+      "type" : "FORWARD"
+    },
+    "shuffleMode" : "PIPELINED"
+  }, {
+    "source" : 28,
+    "target" : 29,
+    "shuffle" : {
+      "type" : "FORWARD"
+    },
+    "shuffleMode" : "PIPELINED"
+  }, {
+    "source" : 29,
+    "target" : 30,
+    "shuffle" : {
+      "type" : "FORWARD"
+    },
+    "shuffleMode" : "PIPELINED"
+  } ]
+}