snuyanzin commented on code in PR #28858:
URL: https://github.com/apache/flink/pull/28858#discussion_r3747387479
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/over/RangeBoundComparatorCodeGenerator.scala:
##########
@@ -135,16 +135,25 @@ 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)) {
+ (bound, 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)) {
+ // TypeFamily covers all TIMESTAMP variants; .getMillisecond() yields
epoch millis for ExprCodeGenerator
+ (s"$inputValue.getMillisecond()", s"$currentValue.getMillisecond()")
Review Comment:
for instance
```java
static final TableTestProgram OVER_AGGREGATE_RANGE_TIMESTAMP_SUBMILLI =
TableTestProgram.of(
"over-aggregate-range-timestamp-submilli",
"RANGE OVER with sub-millisecond TIMESTAMP(6)
ORDER BY")
.setupTableSource(
SourceTestStep.newBuilder("ts6_source")
.addSchema("ts TIMESTAMP(6)", "val INT")
.producedValues(
Row.of(LocalDateTime.of(2021, 1,
1, 12, 0, 0, 0), 1),
Row.of(LocalDateTime.of(2021, 1,
1, 12, 0, 10, 600_000), 2))
.build())
.setupTableSink(
SinkTestStep.newBuilder("ts6_sink")
.addSchema("val INT", "cnt BIGINT")
.consumedValues(Row.of(1, 1L), Row.of(2,
1L)) // CORRECT semantics
.build())
.runSql(
"INSERT INTO ts6_sink SELECT val, COUNT(val)
OVER (ORDER BY ts"
+ " RANGE BETWEEN INTERVAL '10' SECOND
PRECEDING AND CURRENT ROW)"
+ " FROM ts6_source")
.build();
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
