feat: streaming response (#1317)

for queries that are expected to return lot of data, we are adding 
the streaming response mechanism. It uses datafusion's 
execute_stream function that sends streaming response
use query param streaming=true to get the streaming response
defaulted to false, for prism or other clients to work as usual

query response (with streaming=true) sends multiple batches 
one after the other split by new line character

Author: nikhilsinhaparseable

Cargo.lock

@@ -121,6 +121,7 @@ static-files = "0.2"
 thiserror = "2.0"
 ulid = { version = "1.0", features = ["serde"] }
 xxhash-rust = { version = "0.8", features = ["xxh3"] }
+futures-core = "0.3.31"
 
 [build-dependencies]
 cargo_toml = "0.21"

Cargo.toml

@@ -16,7 +16,7 @@
  *
  */
 
-use std::{io::ErrorKind, sync::Arc};
+use std::sync::Arc;
 
 use chrono::{DateTime, Local, NaiveTime, Utc};
 use column::Column;
@@ -259,10 +259,7 @@ async fn create_manifest(
         .date_naive()
         .and_time(
             NaiveTime::from_num_seconds_from_midnight_opt(23 * 3600 + 59 * 60 + 59, 999_999_999)
-                .ok_or(IOError::new(
-                    ErrorKind::Other,
-                    "Failed to create upper bound for manifest",
-                ))?,
+                .ok_or(IOError::other("Failed to create upper bound for manifest"))?,
         )
         .and_utc();
 

src/catalog/mod.rs

@@ -126,9 +126,11 @@ impl FlightService for AirServiceImpl {
     }
 
     async fn do_get(&self, req: Request<Ticket>) -> Result<Response<Self::DoGetStream>, Status> {
-        let key = extract_session_key(req.metadata())?;
+        let key = extract_session_key(req.metadata())
+            .map_err(|e| Status::unauthenticated(e.to_string()))?;
 
-        let ticket = get_query_from_ticket(&req)?;
+        let ticket =
+            get_query_from_ticket(&req).map_err(|e| Status::invalid_argument(e.to_string()))?;
 
         info!("query requested to airplane: {:?}", ticket);
 
@@ -217,10 +219,19 @@ impl FlightService for AirServiceImpl {
         })?;
         let time = Instant::now();
 
-        let (records, _) = execute(query, &stream_name)
+        let (records, _) = execute(query, &stream_name, false)
             .await
             .map_err(|err| Status::internal(err.to_string()))?;
 
+        let records = match records {
+            actix_web::Either::Left(rbs) => rbs,
+            actix_web::Either::Right(_) => {
+                return Err(Status::failed_precondition(
+                    "Expected batch results, got stream",
+                ))
+            }
+        };
+
         /*
         * INFO: No returning the schema with the data.
         * kept it in case it needs to be sent in the future.
@@ -246,7 +257,7 @@ impl FlightService for AirServiceImpl {
             .observe(time);
 
         // Airplane takes off 🛫
-        out
+        out.map_err(|e| *e)
     }
 
     async fn do_put(