This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch release-1.20
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.20 by this push:
new ea37edb710f [FLINK-25802][FLINK-30499][table] Fix TIMESTAMP codegen
for `RANGE OVER` window bounds
ea37edb710f is described below
commit ea37edb710f8069ed20f47bc9a979f14df7e44c7
Author: Aleksandr Iushmanov <[email protected]>
AuthorDate: Mon Aug 24 16:26:24 2026 +0100
[FLINK-25802][FLINK-30499][table] Fix TIMESTAMP codegen for `RANGE OVER`
window bounds
Generated-by: Claude Code
---
.../over/RangeBoundComparatorCodeGenerator.scala | 52 +++++++++++++++++-----
.../runtime/batch/sql/OverAggregateITCase.scala | 47 +++++++++++++++++++
2 files changed, 88 insertions(+), 11 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/over/RangeBoundComparatorCodeGenerator.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/over/RangeBoundComparatorCodeGenerator.scala
index c1f2965c80f..1fe59ace8bc 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/over/RangeBoundComparatorCodeGenerator.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/over/RangeBoundComparatorCodeGenerator.scala
@@ -135,17 +135,41 @@ 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.
- (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)
+ val isTimestamp = keyType.getTypeRoot match {
+ case LogicalTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE |
+ LogicalTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE =>
+ true
+ case _ => false
}
+ val (realBoundValue, realKeyType) = if (isTimestamp) {
+ // 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
+ 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.
+ (bound.asInstanceOf[Long] / DateTimeUtils.MILLIS_PER_DAY, new
IntType())
+ case LogicalTypeRoot.TIME_WITHOUT_TIME_ZONE => (bound, new IntType())
+ case _ => (bound, keyType)
+ }
+
+ // Epoch microseconds: millis * 1000 + nanoOfMillis / 1000 preserves
TIMESTAMP(6) precision
+ val (realInputValue, realCurrentValue) =
+ if (isTimestamp) {
+ (
+ 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 +181,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 +195,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}
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/OverAggregateITCase.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/OverAggregateITCase.scala
index c50b44d70eb..9c76417b6c1 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/OverAggregateITCase.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/OverAggregateITCase.scala
@@ -18,6 +18,7 @@
package org.apache.flink.table.planner.runtime.batch.sql
import org.apache.flink.api.common.typeinfo.BasicTypeInfo._
+import org.apache.flink.api.common.typeinfo.LocalTimeTypeInfo.{LOCAL_DATE_TIME
=> LOCAL_DATE_TIME_TYPE_INFO}
import org.apache.flink.api.java.tuple.{Tuple1 => JTuple1}
import org.apache.flink.api.java.typeutils.RowTypeInfo
import org.apache.flink.table.api.DataTypes
@@ -35,6 +36,7 @@ import org.apache.flink.types.Row
import org.junit.jupiter.api.{BeforeEach, Test}
import java.lang.{Iterable => JIterable, Long => JLong}
+import java.time.Instant
import java.util.{Collections, Optional}
import scala.util.Random
@@ -2361,6 +2363,51 @@ class OverAggregateITCase extends BatchTestBase {
)
}
+ // Regression test for FLINK-25802 / FLINK-30499: RANGE OVER with TIMESTAMP
ORDER BY.
+ // Row 3 is 10s+1ms after row 1, so row 1 falls just outside row 3's
10-second window.
+ @Test
+ def testRangeOverWithTimestamp(): Unit = {
+ val data = Seq(
+ row(localDateTime("2021-01-01 00:00:00.000000"), 1),
+ row(localDateTime("2021-01-01 00:00:05.000000"), 2),
+ row(localDateTime("2021-01-01 00:00:10.001"), 3)
+ )
+ registerCollection(
+ "TimestampRangeTable",
+ data,
+ new RowTypeInfo(LOCAL_DATE_TIME_TYPE_INFO, INT_TYPE_INFO),
+ "ts, val",
+ Array(false, false))
+ checkResult(
+ "SELECT val, COUNT(val) OVER (ORDER BY ts RANGE BETWEEN INTERVAL '10'
SECOND" +
+ " PRECEDING AND CURRENT ROW) FROM TimestampRangeTable",
+ Seq(row(1, 1L), row(2, 2L), row(3, 2L))
+ )
+ }
+
+ @Test
+ def testRangeOverWithTimestampLtz(): Unit = {
+ val epoch = localDateTime("2021-01-01 00:00:00.000000")
+ .atZone(java.time.ZoneOffset.UTC)
+ .toInstant
+ val data = Seq(
+ row(epoch, 1),
+ row(epoch.plusSeconds(5), 2),
+ row(epoch.plusSeconds(10).plusNanos(1000000), 3)
+ )
+ registerCollection(
+ "TimestampLtzRangeTable",
+ data,
+ new RowTypeInfo(INSTANT_TYPE_INFO, INT_TYPE_INFO),
+ "ts, val",
+ Array(false, false))
+ checkResult(
+ "SELECT val, COUNT(val) OVER (ORDER BY ts RANGE BETWEEN INTERVAL '10'
SECOND" +
+ " PRECEDING AND CURRENT ROW) FROM TimestampLtzRangeTable",
+ Seq(row(1, 1L), row(2, 2L), row(3, 2L))
+ )
+ }
+
@Test
def testOverWindowBasedStringOrderBy(): Unit = {
checkResult(
