feat: aes 291 continuous queries in olap aggregator

[samika98]

The OLAP aggregator will continuously run the data event and time event queries and store the results into Basin. This code here adds logic to run the queries continuously.

We opted not to utilize Data Fusion's streaming capabilities, as they lack the features and flexibility required for our use case. Data Fusion's streaming functionality is relatively limited in scope.

[linear]

AES-291 Continuous Queries in OLAP Aggregator

[nathanielc]

Why is last_processed_index inside a Cell? Why not accept a &mut self in the process_batch method?

[samika98]

We do not want ctx to be mutable on the struct. Taking &self can make the API more flexible. I think interior mutability can make the API more flexible. Example :

    let mut processor = ContinuousStreamProcessor::new();
    
    // This works fine
    processor.process_batch(&[Data::new()]);
    
    // But what if we want to do this?
    let processors = vec![&processor, &processor];
    
    for p in &processors {
        p.process_batch(&[Data::new()]); // This won't compile
    }```
    Don't feel too strongly either ways.

[nathanielc]

process_batch only ever returns true. Should we simplify this logic such that it returns Result<()>?

[nathanielc]

nit, since the name df_clone doesn't convey any meaning I'd say this should just be process_feed_batch(self.ctx.clone(), df.clone()).await?;

[nathanielc]

A comment here that we need a cache because we do two passes over the data would be good. First pass is the normal process_feed_batch pass while the second pass is to get the highest index

[nathanielc]

Any luck on getting a test working with the custom context?