Merge pull request #105 from kaskada-ai/engine/fix-shift-to-bounds

bug: fix off by one error in shift to

crates/sparrow-main/tests/e2e/notebooks.rs

@@ -1,4 +1,5 @@
 //! e2e tests based on the queries used in notebooks.
 mod documentation_code_tests;
 mod event_data_tests;
+mod gaming_tests;
 mod sample_tests;

crates/sparrow-main/tests/e2e/notebooks/gaming_tests.rs

@@ -0,0 +1,58 @@
+use indoc::indoc;
+use sparrow_api::kaskada::v1alpha::TableConfig;
+use uuid::Uuid;
+
+use crate::{DataFixture, QueryFixture};
+
+const GAMING_EVENTS_CSV: &str = indoc! {"
+event_at,entity_id,duration,won
+2022-01-01 02:30:00+00:00,Alice,10,true
+2022-01-01 02:35:00+00:00,Bob,3,false
+2022-01-01 03:46:00+00:00,Bob,8,false
+2022-01-01 03:58:00+00:00,Bob,23,true
+2022-01-01 04:25:00+00:00,Bob,8,true
+2022-01-01 05:05:00+00:00,Alice,53,true
+2022-01-01 05:36:00+00:00,Alice,2,false
+2022-01-01 07:22:00+00:00,Bob,7,false
+2022-01-01 08:35:00+00:00,Alice,5,false
+2022-01-01 10:01:00+00:00,Alice,43,true
+"};
+
+/// Create a fixture with the sample events csv.
+fn gaming_data_fixture() -> DataFixture {
+    DataFixture::new()
+        .with_table_from_csv(
+            TableConfig::new(
+                "GamePlay",
+                &Uuid::new_v4(),
+                "event_at",
+                None,
+                "entity_id",
+                "User",
+            ),
+            GAMING_EVENTS_CSV,
+        )
+        .unwrap()
+}
+
+const GAMING_EVENTS: &str = indoc! {"
+    let GameDefeat = GamePlay | when(not(GamePlay.won))
+    
+    let features = {
+        loss_duration: sum(GameDefeat.duration) }
+    
+    let is_prediction_time = not(GamePlay.won) and (count(GameDefeat, window=since(GamePlay.won)) == 2)
+    
+    let example = features | when(is_prediction_time)| shift_by(seconds(60*10))
+    in example
+"};
+
+#[tokio::test]
+async fn test_gaming_events_to_csv() {
+    insta::assert_snapshot!(QueryFixture::new(GAMING_EVENTS).with_dump_dot("gaming").run_to_csv(&gaming_data_fixture()).await.unwrap(),
+    @r###"
+    _time,_subsort,_key_hash,_key,loss_duration
+    2022-01-01T03:56:00.000000000,0,17054345325612802246,Bob,11
+    2022-01-01T08:45:00.000000000,0,5902814233694669492,Alice,7
+    "###);
+}

crates/sparrow-runtime/src/execute/operation/shift_to.rs

@@ -525,7 +525,14 @@ impl ShiftToColumnOperation {
             lower_bound.subsort = 0;
             let mut upper_bound = prefix.upper_bound;
             upper_bound.time = cmp::min(incoming_upper_bound_time, upper_bound.time);
-            upper_bound.subsort = prefix.time.len() as u64; // u64::MAX;
+
+            let upper_subsort = if prefix.time.len() == 0 {
+                0
+            } else {
+                // 0-indexed, so sub 1
+                prefix.time.len() as u64 - 1
+            };
+            upper_bound.subsort = upper_subsort;
 
             let prefix = InputBatch {
                 time: prefix.time,