[CALCITE-7683] SessionizationEnumerator produces wrong results for SESSION table function - #5137
[CALCITE-7683] SessionizationEnumerator produces wrong results for SESSION table function#5137mihaibudiu wants to merge 1 commit into
Conversation
|
@amaliujia seems to have authored this code, so I am requesting his review. |
…SSION table function Signed-off-by: Mihai Budiu <mbudiu@feldera.com>
|
|
Ah It's been a while to go back to this code and thanks for the finding. I will take a look. |
| "element[indexOfWatermarkedColumn]"); | ||
| Object watermark = element[indexOfWatermarkedColumn]; | ||
| if (watermark == null) { | ||
| // A row without a timestamp belongs to no session, and window_start |
There was a problem hiding this comment.
This is changing behavior:
The previous we expect all the rows should have timestamp otherwise will fail by requireNonNull.
Can you remind me when a row without timestamp could hit here?
There was a problem hiding this comment.
There is a test: just insert a row with null timestamp in the input table.
The validator does not reject that.
| // 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<>(); |
There was a problem hiding this comment.
I am a bit confused: how do we achieve sorted timestamp here?
There was a problem hiding this comment.
SortedMultiMap sorts values (previous implementation), but NavigableMap is a subclass of SortedMap which sorts keys. Keys are Pair objects, which are Comparable.
2 participants
Jira Link
CALCITE-7683
Changes Proposed
Computing session windows needs to sort by timestamp, but the data structure that was being used
SortedMultiMapwas not doing this.