feat: Use object_store to write files (#492)

Also introduces paged file output.

This does a hacky CSV write by buffering a batch at a time.

This is related to #486.
This is part of #465.
This is part of #466.

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

@@ -207,7 +207,9 @@ async fn test_sum_i64_final_at_time() {
 #[tokio::test]
 async fn test_sum_i64_all_filtered_final_at_time() {
     let datetime = NaiveDateTime::new(date_for_test(1970, 12, 20), time_for_test(0, 39, 58));
-    insta::assert_snapshot!(QueryFixture::new("{ sum_field: sum(Numbers.m) }").with_final_results_at_time(datetime).run_to_csv(&i64_data_fixture().await).await.unwrap(), @"");
+    insta::assert_snapshot!(QueryFixture::new("{ sum_field: sum(Numbers.m) }").with_final_results_at_time(datetime).run_to_csv(&i64_data_fixture().await).await.unwrap(), @r###"
+    _time,_subsort,_key_hash,_key,sum_field
+    "###);
 }
 
 async fn shift_data_fixture_at_time() -> DataFixture {

crates/sparrow-runtime/src/execute.rs

@@ -157,7 +157,7 @@ pub async fn execute(
         None
     };
 
-    let object_stores = ObjectStoreRegistry::default();
+    let object_stores = Arc::new(ObjectStoreRegistry::default());
 
     let primary_grouping_key_type = plan
         .primary_grouping_key_type
@@ -300,7 +300,7 @@ pub async fn materialize(
         .into_report()
         .change_context(Error::internal_msg("get primary grouping ID"))?;
 
-    let object_stores = ObjectStoreRegistry::default();
+    let object_stores = Arc::new(ObjectStoreRegistry::default());
     key_hash_inverse
         .add_from_data_context(&data_context, primary_group_id, &object_stores)
         .await

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

@@ -81,7 +81,7 @@ use crate::Batch;
 pub(crate) struct OperationContext {
     pub plan: ComputePlan,
     pub plan_hash: PlanHash,
-    pub object_stores: ObjectStoreRegistry,
+    pub object_stores: Arc<ObjectStoreRegistry>,
     pub data_context: DataContext,
     pub compute_store: Option<Arc<ComputeStore>>,
     /// The key hash inverse to produce the output results, if one exists.

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

@@ -449,7 +449,7 @@ mod tests {
                 ..ComputePlan::default()
             },
             plan_hash: PlanHash::default(),
-            object_stores: ObjectStoreRegistry::default(),
+            object_stores: Arc::new(ObjectStoreRegistry::default()),
             data_context,
             compute_store: None,
             key_hash_inverse,

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

@@ -182,7 +182,7 @@ pub(super) async fn run_operation(
             ..ComputePlan::default()
         },
         plan_hash: PlanHash::default(),
-        object_stores: ObjectStoreRegistry::default(),
+        object_stores: Arc::new(ObjectStoreRegistry::default()),
         data_context: DataContext::default(),
         compute_store: None,
         key_hash_inverse,
@@ -240,7 +240,7 @@ pub(super) async fn run_operation_json(
             ..ComputePlan::default()
         },
         plan_hash: PlanHash::default(),
-        object_stores: ObjectStoreRegistry::default(),
+        object_stores: Arc::new(ObjectStoreRegistry::default()),
         data_context: DataContext::default(),
         compute_store: None,
         key_hash_inverse,

crates/sparrow-runtime/src/execute/output.rs

@@ -18,9 +18,7 @@ use crate::execute::operation::OperationContext;
 use crate::execute::progress_reporter::ProgressUpdate;
 use crate::Batch;
 
-mod csv;
 mod object_store;
-mod parquet;
 mod redis;
 
 pub mod pulsar;
@@ -49,7 +47,7 @@ pub(super) fn write(
     batches: BoxStream<'static, Batch>,
     progress_updates_tx: tokio::sync::mpsc::Sender<ProgressUpdate>,
     destination: v1alpha::Destination,
-) -> error_stack::Result<impl Future<Output = Result<(), Error>>, Error> {
+) -> error_stack::Result<impl Future<Output = Result<(), Error>> + 'static, Error> {
     let sink_schema = determine_output_schema(context)?;
 
     // Clone things that need to move into the async stream.
@@ -89,15 +87,17 @@ pub(super) fn write(
         .destination
         .ok_or(Error::UnspecifiedDestination)?;
     match destination {
-        Destination::ObjectStore(store) => {
-            Ok(
-                object_store::write(store, sink_schema, progress_updates_tx, batches)
-                    .change_context(Error::WritingToDestination {
-                        dest_name: "object_store".to_owned(),
-                    })
-                    .boxed(),
-            )
-        }
+        Destination::ObjectStore(destination) => Ok(object_store::write(
+            context.object_stores.clone(),
+            destination,
+            sink_schema,
+            progress_updates_tx,
+            batches,
+        )
+        .change_context(Error::WritingToDestination {
+            dest_name: "object_store".to_owned(),
+        })
+        .boxed()),
         Destination::Redis(redis) => {
             Ok(
                 redis::write(redis, sink_schema, progress_updates_tx, batches)