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 fc95bb0e68 [CALCITE-7682] SESSION table function without the optional
key descriptor fails at runtime
fc95bb0e68 is described below
commit fc95bb0e68c838da4232aabf47ee5037cf73b778
Author: Mihai Budiu <[email protected]>
AuthorDate: Fri Jul 31 13:43:29 2026 -0700
[CALCITE-7682] SESSION table function without the optional key descriptor
fails at runtime
Signed-off-by: Mihai Budiu <[email protected]>
---
.../calcite/adapter/enumerable/EnumUtils.java | 12 ++++++--
.../calcite/adapter/enumerable/RexImpTable.java | 16 +++++++---
core/src/test/resources/sql/stream.iq | 36 ++++++++++++++++++++++
3 files changed, 57 insertions(+), 7 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 40a824536b..d1c395c226 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
@@ -1100,6 +1100,9 @@ static Expression tumblingWindowSelector(
* Creates enumerable implementation that applies sessionization to elements
from the input
* enumerator based on a specified key. Elements are windowed into sessions
separated by
* periods with no input for at least the duration specified by gap
parameter.
+ *
+ * <p>The key is optional: pass -1 for {@code indexOfKeyColumn} to place
every
+ * element on a single session timeline.
*/
public static Enumerable<@Nullable Object[]> sessionize(
Enumerator<@Nullable Object[]> inputEnumerator,
@@ -1144,7 +1147,9 @@ private static class SessionizationEnumerator implements
Enumerator<@Nullable Ob
*
* @param inputEnumerator the enumerator to provide an array of objects as
input
* @param indexOfWatermarkedColumn the index of timestamp column upon
which a watermark is built
- * @param indexOfKeyColumn the index of column that acts as grouping key
+ * @param indexOfKeyColumn the index of column that acts as grouping key,
+ * or -1 if there is no key and all rows belong to
+ * a single session timeline
* @param gap gap parameter
*/
SessionizationEnumerator(Enumerator<@Nullable Object[]> inputEnumerator,
@@ -1194,8 +1199,11 @@ private void initialize() {
Map<@Nullable Object, SortedMultiMap<Pair<Long, Long>, @Nullable
Object[]>> sessionKeyMap =
new HashMap<>();
for (@Nullable Object[] element : elements) {
+ // A key column index of -1 means that there is no key; every element
+ // then maps to the same (null) key, forming one session timeline.
+ Object key = indexOfKeyColumn < 0 ? null : element[indexOfKeyColumn];
SortedMultiMap<Pair<Long, Long>, @Nullable Object[]> session =
- sessionKeyMap.computeIfAbsent(element[indexOfKeyColumn], k -> new
SortedMultiMap<>());
+ sessionKeyMap.computeIfAbsent(key, k -> new SortedMultiMap<>());
Object watermark =
requireNonNull(element[indexOfWatermarkedColumn],
"element[indexOfWatermarkedColumn]");
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 2ca2528d5c..7a5ea10f14 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
@@ -5113,14 +5113,20 @@ private static Method getMethod(Type
comparisonReturnType, SqlKind kind) {
private static class SessionImplementor implements
TableFunctionCallImplementor {
@Override public Expression implement(RexToLixTranslator translator,
Expression inputEnumerable, RexCall call, PhysType inputPhysType,
PhysType outputPhysType) {
- RexCall timestampDescriptor = (RexCall) call.getOperands().get(0);
- RexCall keyDescriptor = (RexCall) call.getOperands().get(1);
- Expression gapInterval = translator.translate(call.getOperands().get(2));
+ final List<RexNode> operands = call.getOperands();
+ RexCall timestampDescriptor = (RexCall) operands.get(0);
+ // The gap is always the last operand; the key descriptor between them is
+ // optional. Without a key every row belongs to a single session
+ // timeline, which a key column index of -1 denotes.
+ Expression gapInterval = translator.translate(Util.last(operands));
+ final int keyColIndex =
+ operands.size() > 2 && operands.get(1).getKind() ==
SqlKind.DESCRIPTOR
+ ? ((RexInputRef) ((RexCall)
operands.get(1)).getOperands().get(0)).getIndex()
+ : -1;
Expression wmColIndexExpr =
Expressions.constant(((RexInputRef)
timestampDescriptor.getOperands().get(0)).getIndex());
- Expression keyColIndexExpr =
- Expressions.constant(((RexInputRef)
keyDescriptor.getOperands().get(0)).getIndex());
+ Expression keyColIndexExpr = Expressions.constant(keyColIndex);
return Expressions.call(BuiltInMethod.SESSIONIZATION.method,
Expressions.list(
diff --git a/core/src/test/resources/sql/stream.iq
b/core/src/test/resources/sql/stream.iq
index 3941143762..b9df0692cc 100644
--- a/core/src/test/resources/sql/stream.iq
+++ b/core/src/test/resources/sql/stream.iq
@@ -291,3 +291,39 @@ SELECT * FROM TABLE(SESSION((SELECT * FROM ORDERS),
DESCRIPTOR(ROWTIME), DESCRIP
(5 rows)
!ok
+
+# Test case for [CALCITE-7682] SESSION table function without the optional key
+# descriptor fails at runtime.
+SELECT * FROM TABLE(SESSION(TABLE ORDERS, DESCRIPTOR(ROWTIME), INTERVAL '1'
HOUR));
++---------------------+----+---------+-------+---------------------+---------------------+
+| ROWTIME | ID | PRODUCT | UNITS | window_start |
window_end |
++---------------------+----+---------+-------+---------------------+---------------------+
+| 2015-02-15 10:15:00 | 1 | paint | 10 | 2015-02-15 10:15:00 |
2015-02-15 12:10:00 |
+| 2015-02-15 10:24:15 | 2 | paper | 5 | 2015-02-15 10:15:00 |
2015-02-15 12:10:00 |
+| 2015-02-15 10:24:45 | 3 | brush | 12 | 2015-02-15 10:15:00 |
2015-02-15 12:10:00 |
+| 2015-02-15 10:58:00 | 4 | paint | 3 | 2015-02-15 10:15:00 |
2015-02-15 12:10:00 |
+| 2015-02-15 11:10:00 | 5 | paint | 3 | 2015-02-15 10:15:00 |
2015-02-15 12:10:00 |
++---------------------+----+---------+-------+---------------------+---------------------+
+(5 rows)
+
+!ok
+
+# As above, but with named parameters, so the KEY parameter is omitted rather
+# than simply absent.
+SELECT * FROM TABLE(
+ SESSION(
+ DATA => TABLE ORDERS,
+ TIMECOL => DESCRIPTOR(ROWTIME),
+ SIZE => INTERVAL '1' HOUR));
++---------------------+----+---------+-------+---------------------+---------------------+
+| ROWTIME | ID | PRODUCT | UNITS | window_start |
window_end |
++---------------------+----+---------+-------+---------------------+---------------------+
+| 2015-02-15 10:15:00 | 1 | paint | 10 | 2015-02-15 10:15:00 |
2015-02-15 12:10:00 |
+| 2015-02-15 10:24:15 | 2 | paper | 5 | 2015-02-15 10:15:00 |
2015-02-15 12:10:00 |
+| 2015-02-15 10:24:45 | 3 | brush | 12 | 2015-02-15 10:15:00 |
2015-02-15 12:10:00 |
+| 2015-02-15 10:58:00 | 4 | paint | 3 | 2015-02-15 10:15:00 |
2015-02-15 12:10:00 |
+| 2015-02-15 11:10:00 | 5 | paint | 3 | 2015-02-15 10:15:00 |
2015-02-15 12:10:00 |
++---------------------+----+---------+-------+---------------------+---------------------+
+(5 rows)
+
+!ok
