This is an automated email from the ASF dual-hosted git repository.
mihaibudiu pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/calcite.git
The following commit(s) were added to refs/heads/main by this push:
new ed12f36f19 [CALCITE-7684] HOPPING and TUMBLING window queries crash at
runtime for NULL timestamps
ed12f36f19 is described below
commit ed12f36f19f4323a1bd02443d3e36ed076cbd425
Author: Mihai Budiu <[email protected]>
AuthorDate: Fri Jul 31 13:27:11 2026 -0700
[CALCITE-7684] HOPPING and TUMBLING window queries crash at runtime for
NULL timestamps
Signed-off-by: Mihai Budiu <[email protected]>
---
.../calcite/adapter/enumerable/EnumUtils.java | 66 +++++++++++++++------
.../calcite/adapter/enumerable/RexImpTable.java | 12 +---
core/src/test/resources/sql/stream.iq | 69 ++++++++++++++++++++++
3 files changed, 121 insertions(+), 26 deletions(-)
diff --git
a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumUtils.java
b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumUtils.java
index 80c46ec3f9..d3da43466d 100644
--- a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumUtils.java
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumUtils.java
@@ -1036,7 +1036,7 @@ static Expression generatePredicate(
static Expression tumblingWindowSelector(
PhysType inputPhysType,
PhysType outputPhysType,
- Expression wmColExpr,
+ int wmColIndex,
Expression windowSizeExpr,
Expression offsetExpr) {
// Generate all fields.
@@ -1053,6 +1053,9 @@ static Expression tumblingWindowSelector(
outputPhysType.getJavaFieldType(expressions.size()));
expressions.add(expression);
}
+ final Expression wmColExpr =
+ inputPhysType.fieldReference(parameter, wmColIndex,
+ outputPhysType.getJavaFieldType(fieldCount));
final Expression wmColExprToLong = EnumUtils.convert(wmColExpr,
long.class);
// Find the fixed window for a timestamp given a window size and an
offset, and return the
@@ -1074,8 +1077,20 @@ static Expression tumblingWindowSelector(
expressions.add(windowEndExpr);
- return Expressions.lambda(Function1.class,
- outputPhysType.record(expressions), parameter);
+ Expression body = outputPhysType.record(expressions);
+ if (inputPhysType.getRowType().getFieldList().get(wmColIndex).getType()
+ .isNullable()) {
+ // A row whose timestamp is NULL belongs to no window, since window_start
+ // and window_end are declared NOT NULL. Return null for such a row
+ body =
+ Expressions.condition(
+ Expressions.equal(
+ inputPhysType.fieldReference(parameter, wmColIndex),
+ Expressions.constant(null)),
+ Expressions.constant(null, body.getType()),
+ body);
+ }
+ return Expressions.lambda(Function1.class, body, parameter);
}
/**
@@ -1307,13 +1322,22 @@ private static class HopEnumerator implements
Enumerator<@Nullable Object[]> {
}
@Override public @Nullable Object[] current() {
- if (!list.isEmpty()) {
- return takeOne();
- } else {
+ return takeOne();
+ }
+
+ @Override public boolean moveNext() {
+ // Expand input rows until one of them yields a window. A row whose
+ // timestamp is NULL belongs to no window, and window_start and
+ // window_end are declared NOT NULL, so such a row is discarded.
+ while (list.isEmpty()) {
+ if (!inputEnumerator.moveNext()) {
+ return false;
+ }
@Nullable Object[] current = inputEnumerator.current();
- Object watermark =
- requireNonNull(current[indexOfWatermarkedColumn],
- "element[indexOfWatermarkedColumn]");
+ Object watermark = current[indexOfWatermarkedColumn];
+ if (watermark == null) {
+ continue;
+ }
PairList<Long, Long> windows =
hopWindows(SqlFunctions.toLong(watermark), emitFrequency,
windowSize, offset);
@@ -1324,12 +1348,8 @@ private static class HopEnumerator implements
Enumerator<@Nullable Object[]> {
curWithWindow[current.length + 1] = right;
list.offer(curWithWindow);
});
- return takeOne();
}
- }
-
- @Override public boolean moveNext() {
- return !list.isEmpty() || inputEnumerator.moveNext();
+ return true;
}
@Override public void reset() {
@@ -1367,17 +1387,29 @@ public static <TSource, TResult> Enumerable<TResult>
tumbling(
Function1<TSource, TResult> outSelector) {
return new AbstractEnumerable<TResult>() {
// Applies tumbling on each element from the input enumerator and
produces
- // exactly one element for each input element.
+ // at most one element for each input element.
@Override public Enumerator<TResult> enumerator() {
return new Enumerator<TResult>() {
final Enumerator<TSource> inputs = inputEnumerable.enumerator();
+ @Nullable TResult current;
@Override public TResult current() {
- return outSelector.apply(inputs.current());
+ return requireNonNull(current, "current");
}
@Override public boolean moveNext() {
- return inputs.moveNext();
+ // The selector returns null for a row whose timestamp is NULL:
+ // such a row belongs to no window, and window_start and window_end
+ // are declared NOT NULL, so the row is discarded.
+ while (inputs.moveNext()) {
+ TResult result = outSelector.apply(inputs.current());
+ if (result != null) {
+ current = result;
+ return true;
+ }
+ }
+ current = null;
+ return false;
}
@Override public void reset() {
diff --git
a/core/src/main/java/org/apache/calcite/adapter/enumerable/RexImpTable.java
b/core/src/main/java/org/apache/calcite/adapter/enumerable/RexImpTable.java
index 2ad2be2893..2ca2528d5c 100644
--- a/core/src/main/java/org/apache/calcite/adapter/enumerable/RexImpTable.java
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/RexImpTable.java
@@ -4970,14 +4970,8 @@ private static class TumbleImplementor implements
TableFunctionCallImplementor {
// represents the input, see
StandardConvertletTable#convertWindowFunction.
Expression intervalExpression =
translator.translate(call.getOperands().get(1));
RexCall descriptor = (RexCall) call.getOperands().get(0);
- final ParameterExpression parameter =
- Expressions.parameter(Primitive.box(inputPhysType.getJavaRowType()),
- "_input");
- Expression wmColExpr =
- inputPhysType.fieldReference(parameter,
- ((RexInputRef) descriptor.getOperands().get(0)).getIndex(),
- outputPhysType.getJavaFieldType(
- inputPhysType.getRowType().getFieldCount()));
+ final int wmColIndex =
+ ((RexInputRef) descriptor.getOperands().get(0)).getIndex();
// handle the optional offset parameter. Use 0 for the default value
when offset
// parameter is not set.
@@ -4991,7 +4985,7 @@ private static class TumbleImplementor implements
TableFunctionCallImplementor {
EnumUtils.tumblingWindowSelector(
inputPhysType,
outputPhysType,
- wmColExpr,
+ wmColIndex,
intervalExpression,
offsetExpr));
}
diff --git a/core/src/test/resources/sql/stream.iq
b/core/src/test/resources/sql/stream.iq
index f20a7fe640..3941143762 100644
--- a/core/src/test/resources/sql/stream.iq
+++ b/core/src/test/resources/sql/stream.iq
@@ -95,6 +95,40 @@ SELECT * FROM TABLE(TUMBLE((SELECT * FROM ORDERS),
DESCRIPTOR(ROWTIME), INTERVAL
!ok
+# Test case for [CALCITE-7684] HOPPING and TUMBLING window queries crash at
+# runtime for NULL timestamps.
+# A row whose timestamp is NULL belongs to no window; window_start and
+# window_end are declared NOT NULL, so such a row is discarded.
+SELECT * FROM TABLE(
+ TUMBLE(
+ (SELECT * FROM (VALUES
+ (TIMESTAMP '2020-01-01 10:00:00', 'a'),
+ (CAST(NULL AS TIMESTAMP), 'b')) AS T(TS, UID)),
+ DESCRIPTOR(TS), INTERVAL '1' HOUR));
++---------------------+-----+---------------------+---------------------+
+| TS | UID | window_start | window_end |
++---------------------+-----+---------------------+---------------------+
+| 2020-01-01 10:00:00 | a | 2020-01-01 10:00:00 | 2020-01-01 11:00:00 |
++---------------------+-----+---------------------+---------------------+
+(1 row)
+
+!ok
+
+# As above, but every row is discarded, so the result is empty.
+SELECT * FROM TABLE(
+ TUMBLE(
+ (SELECT * FROM (VALUES
+ (CAST(NULL AS TIMESTAMP), 'a'),
+ (CAST(NULL AS TIMESTAMP), 'b')) AS T(TS, UID)),
+ DESCRIPTOR(TS), INTERVAL '1' HOUR));
++----+-----+--------------+------------+
+| TS | UID | window_start | window_end |
++----+-----+--------------+------------+
++----+-----+--------------+------------+
+(0 rows)
+
+!ok
+
SELECT * FROM TABLE(HOP(TABLE ORDERS, DESCRIPTOR(ROWTIME), INTERVAL '5'
MINUTE, INTERVAL '10' MINUTE));
+---------------------+----+---------+-------+---------------------+---------------------+
| ROWTIME | ID | PRODUCT | UNITS | window_start |
window_end |
@@ -176,6 +210,41 @@ SELECT * FROM TABLE(HOP((SELECT * FROM ORDERS),
DESCRIPTOR(ROWTIME), INTERVAL '5
!ok
+# Test case for [CALCITE-7684] HOPPING and TUMBLING window queries crash at
+# runtime for NULL timestamps.
+# A row whose timestamp is NULL belongs to no window, not even to the first
+# of the windows that HOP would otherwise produce for it.
+SELECT * FROM TABLE(
+ HOP(
+ (SELECT * FROM (VALUES
+ (TIMESTAMP '2020-01-01 10:00:00', 'a'),
+ (CAST(NULL AS TIMESTAMP), 'b')) AS T(TS, UID)),
+ DESCRIPTOR(TS), INTERVAL '30' MINUTE, INTERVAL '1' HOUR));
++---------------------+-----+---------------------+---------------------+
+| TS | UID | window_start | window_end |
++---------------------+-----+---------------------+---------------------+
+| 2020-01-01 10:00:00 | a | 2020-01-01 09:30:00 | 2020-01-01 10:30:00 |
+| 2020-01-01 10:00:00 | a | 2020-01-01 10:00:00 | 2020-01-01 11:00:00 |
++---------------------+-----+---------------------+---------------------+
+(2 rows)
+
+!ok
+
+# As above, but every row is discarded, so the result is empty.
+SELECT * FROM TABLE(
+ HOP(
+ (SELECT * FROM (VALUES
+ (CAST(NULL AS TIMESTAMP), 'a'),
+ (CAST(NULL AS TIMESTAMP), 'b')) AS T(TS, UID)),
+ DESCRIPTOR(TS), INTERVAL '30' MINUTE, INTERVAL '1' HOUR));
++----+-----+--------------+------------+
+| TS | UID | window_start | window_end |
++----+-----+--------------+------------+
++----+-----+--------------+------------+
+(0 rows)
+
+!ok
+
SELECT * FROM TABLE(SESSION(TABLE ORDERS, DESCRIPTOR(ROWTIME),
DESCRIPTOR(PRODUCT), INTERVAL '20' MINUTE));
+---------------------+----+---------+-------+---------------------+---------------------+
| ROWTIME | ID | PRODUCT | UNITS | window_start |
window_end |
