feat: make collect continuous by setting interpolation (#647)

crates/sparrow-compiler/src/plan/interpolations.rs

@@ -50,8 +50,10 @@ impl Interpolations {
                     // always be able to be present in subsequent instructions.
                     Interpolation::AsOf
                 }
+                // TODO: `collect` should be in it's own special grouping,
+                // or we should just start calling it an aggregation everywhere.
                 StepKind::Expression(Expression::Inst(InstKind::Simple(inst)))
-                    if inst.is_aggregation() =>
+                    if inst.is_aggregation() || inst.name() == "collect" =>
                 {
                     Interpolation::AsOf
                 }

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

@@ -473,6 +473,10 @@ async fn test_collect_boolean_since_hourly() {
 
 #[tokio::test]
 async fn test_collect_struct_since_hourly() {
+    // TODO: The results here are weird, because `collect` is latched. I don't think I'd expect
+    // the results we have here, but it's possible they're technically in line with what we expect
+    // given our continuity rules. We should revisit this.
+    // https://github.com/kaskada-ai/kaskada/issues/648
     insta::assert_snapshot!(QueryFixture::new("{ 
         b: Collect.b,
         f0: ({b: Collect.b} | collect(max=10, window=since(hourly())) | index(0)).b | when(is_valid($input)),
@@ -495,8 +499,8 @@ async fn test_collect_struct_since_hourly() {
     1996-12-21T00:43:57.000000000,9223372036854775808,2867199309159137213,B,false,false,,true,false,
     1996-12-21T00:44:57.000000000,9223372036854775808,2867199309159137213,B,true,false,,true,false,true
     1996-12-21T01:00:00.000000000,18446744073709551615,2867199309159137213,B,,false,,true,false,true
-    1996-12-21T01:44:57.000000000,9223372036854775808,2867199309159137213,B,true,true,,,,
-    1996-12-21T02:00:00.000000000,18446744073709551615,2867199309159137213,B,,true,,,,
+    1996-12-21T01:44:57.000000000,9223372036854775808,2867199309159137213,B,true,true,,true,false,true
+    1996-12-21T02:00:00.000000000,18446744073709551615,2867199309159137213,B,,true,,true,false,true
     1996-12-22T00:44:57.000000000,9223372036854775808,2521269998124177631,C,true,true,,,,
     1996-12-22T00:45:57.000000000,9223372036854775808,2521269998124177631,C,true,true,true,,,
     1996-12-22T00:46:57.000000000,9223372036854775808,2521269998124177631,C,true,true,true,true,,