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 dc4f7f3700 [CALCITE-7683] SessionizationEnumerator produces wrong
results for SESSION table function
dc4f7f3700 is described below
commit dc4f7f37005d7b267fb578bf8e518160bc21aada
Author: Mihai Budiu <[email protected]>
AuthorDate: Fri Jul 31 10:51:16 2026 -0700
[CALCITE-7683] SessionizationEnumerator produces wrong results for SESSION
table function
Signed-off-by: Mihai Budiu <[email protected]>
---
.../calcite/adapter/enumerable/EnumUtils.java | 34 ++++++++++------
.../org/apache/calcite/runtime/SortedMultiMap.java | 4 ++
core/src/test/resources/sql/stream.iq | 47 ++++++++++++++++++++++
3 files changed, 73 insertions(+), 12 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 d1c395c226..9cd7ba802b 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
@@ -55,7 +55,6 @@
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.rex.RexProgramBuilder;
import org.apache.calcite.runtime.PairList;
-import org.apache.calcite.runtime.SortedMultiMap;
import org.apache.calcite.runtime.SqlFunctions;
import org.apache.calcite.runtime.Utilities;
import org.apache.calcite.sql.SqlCollation;
@@ -87,7 +86,9 @@
import java.util.List;
import java.util.Locale;
import java.util.Map;
+import java.util.NavigableMap;
import java.util.TimeZone;
+import java.util.TreeMap;
import java.util.function.Function;
import static
org.apache.calcite.config.CalciteSystemProperty.JOIN_SELECTOR_COMPACT_CODE_THRESHOLD;
@@ -1163,15 +1164,22 @@ private static class SessionizationEnumerator
implements Enumerator<@Nullable Ob
}
@Override public @Nullable Object[] current() {
- if (!initialized) {
- initialize();
- initialized = true;
- }
return list.removeFirst();
}
@Override public boolean moveNext() {
- return initialized ? !list.isEmpty() : inputEnumerator.moveNext();
+ // Sessionization needs to see all the input, so the first call consumes
+ // it entirely. Initializing here rather than in current() lets this
+ // method report that there is nothing to return, which happens when
+ // every row was discarded for having a NULL timestamp.
+ if (!initialized) {
+ initialized = true;
+ if (!inputEnumerator.moveNext()) {
+ return false;
+ }
+ initialize();
+ }
+ return !list.isEmpty();
}
@Override public void reset() {
@@ -1196,24 +1204,26 @@ private void initialize() {
elements.add(inputEnumerator.current());
}
- Map<@Nullable Object, SortedMultiMap<Pair<Long, Long>, @Nullable
Object[]>> sessionKeyMap =
- new HashMap<>();
+ // The windows of each key are kept sorted by start time; the merge
+ // below only compares a window with the one that precedes it.
+ Map<@Nullable Object, NavigableMap<Pair<Long, Long>, List<@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(key, k -> new SortedMultiMap<>());
Object watermark =
requireNonNull(element[indexOfWatermarkedColumn],
"element[indexOfWatermarkedColumn]");
+ NavigableMap<Pair<Long, Long>, List<@Nullable Object[]>> session =
+ sessionKeyMap.computeIfAbsent(key, k -> new TreeMap<>());
Pair<Long, Long> initWindow =
computeInitWindow(SqlFunctions.toLong(watermark), gap);
- session.putMulti(initWindow, element);
+ session.computeIfAbsent(initWindow, k -> new
ArrayList<>()).add(element);
}
// merge per key session windows if there is any overlap between windows.
- for (Map.Entry<@Nullable Object, SortedMultiMap<Pair<Long, Long>,
@Nullable Object[]>>
+ for (Map.Entry<@Nullable Object, NavigableMap<Pair<Long, Long>,
List<@Nullable Object[]>>>
perKeyEntry : sessionKeyMap.entrySet()) {
Map<Pair<Long, Long>, List<@Nullable Object[]>> finalWindowElementsMap
= new HashMap<>();
Pair<Long, Long> currentWindow = null;
diff --git a/core/src/main/java/org/apache/calcite/runtime/SortedMultiMap.java
b/core/src/main/java/org/apache/calcite/runtime/SortedMultiMap.java
index 7b2448c5c9..958d3849f9 100644
--- a/core/src/main/java/org/apache/calcite/runtime/SortedMultiMap.java
+++ b/core/src/main/java/org/apache/calcite/runtime/SortedMultiMap.java
@@ -28,6 +28,10 @@
* Map that allows you to partition values into lists according to a common
* key, and then convert those lists into an iterator of sorted arrays.
*
+ * <p>Only the values are sorted, by {@link #arrays(Comparator)}; the keys are
+ * not, because this map extends {@link HashMap}. Use a {@link
java.util.TreeMap}
+ * if you need to visit the keys in order.
+ *
* @param <K> Key type
* @param <V> Value type
*/
diff --git a/core/src/test/resources/sql/stream.iq
b/core/src/test/resources/sql/stream.iq
index b9df0692cc..4328a63cca 100644
--- a/core/src/test/resources/sql/stream.iq
+++ b/core/src/test/resources/sql/stream.iq
@@ -327,3 +327,50 @@ SELECT * FROM TABLE(
(5 rows)
!ok
+
+# Test case for [CALCITE-7683] SessionizationEnumerator produces wrong results
+# for SESSION table function.
+# Here 10:05 and 10:30 are 25 minutes apart and 10:40 and 11:30 are 50 minutes
+# apart, both more than the 15 minute gap, so the rows form three sessions.
+SELECT * FROM TABLE(
+ SESSION(
+ (SELECT * FROM (VALUES
+ (TIMESTAMP '2020-01-01 10:00:00', 'a'),
+ (TIMESTAMP '2020-01-01 10:05:00', 'a'),
+ (TIMESTAMP '2020-01-01 10:30:00', 'a'),
+ (TIMESTAMP '2020-01-01 10:40:00', 'a'),
+ (TIMESTAMP '2020-01-01 11:30:00', 'a')) AS T(TS, UID)),
+ DESCRIPTOR(TS), DESCRIPTOR(UID), INTERVAL '15' MINUTE))
+ORDER BY TS;
++---------------------+-----+---------------------+---------------------+
+| TS | UID | window_start | window_end |
++---------------------+-----+---------------------+---------------------+
+| 2020-01-01 10:00:00 | a | 2020-01-01 10:00:00 | 2020-01-01 10:20:00 |
+| 2020-01-01 10:05:00 | a | 2020-01-01 10:00:00 | 2020-01-01 10:20:00 |
+| 2020-01-01 10:30:00 | a | 2020-01-01 10:30:00 | 2020-01-01 10:55:00 |
+| 2020-01-01 10:40:00 | a | 2020-01-01 10:30:00 | 2020-01-01 10:55:00 |
+| 2020-01-01 11:30:00 | a | 2020-01-01 11:30:00 | 2020-01-01 11:45:00 |
++---------------------+-----+---------------------+---------------------+
+(5 rows)
+
+!ok
+
+# Test case for [CALCITE-7683] SessionizationEnumerator produces wrong results
+# for SESSION table function.
+SELECT * FROM TABLE(
+ SESSION(
+ (SELECT * FROM (VALUES
+ (TIMESTAMP '2020-01-01 10:00:00', 'a'),
+ (TIMESTAMP '2020-01-01 10:05:00', 'a'),
+ (TIMESTAMP '2020-01-01 10:30:00', 'a')) AS T(TS, UID)),
+ DESCRIPTOR(TS), DESCRIPTOR(UID), INTERVAL '15' MINUTE));
++---------------------+-----+---------------------+---------------------+
+| TS | UID | window_start | window_end |
++---------------------+-----+---------------------+---------------------+
+| 2020-01-01 10:00:00 | a | 2020-01-01 10:00:00 | 2020-01-01 10:20:00 |
+| 2020-01-01 10:05:00 | a | 2020-01-01 10:00:00 | 2020-01-01 10:20:00 |
+| 2020-01-01 10:30:00 | a | 2020-01-01 10:30:00 | 2020-01-01 10:45:00 |
++---------------------+-----+---------------------+---------------------+
+(3 rows)
+
+!ok
