feat: flightsql initial support

crates/glaredb/src/local.rs

@@ -207,7 +207,7 @@ impl LocalSession {
         let statements = self.sess.parse_query(text)?;
         for stmt in statements {
             self.sess
-                .prepare_statement(UNNAMED, Some(stmt), Vec::new())
+                .prepare_statement(UNNAMED, stmt, Vec::new())
                 .await?;
             let prepared = self.sess.get_prepared_statement(&UNNAMED)?;
             let num_fields = prepared.output_fields().map(|f| f.len()).unwrap_or(0);

crates/glaredb/src/server.rs

@@ -4,6 +4,7 @@ use pgsrv::auth::LocalAuthenticator;
 use pgsrv::handler::{ProtocolHandler, ProtocolHandlerConfig};
 use protogen::gen::rpcsrv::service::execution_service_server::ExecutionServiceServer;
 use protogen::gen::rpcsrv::simple::simple_service_server::SimpleServiceServer;
+use rpcsrv::flight_handler::{FlightServiceServer, FlightSessionHandler};
 use rpcsrv::handler::{RpcHandler, SimpleHandler};
 use sqlexec::engine::{Engine, EngineStorageConfig};
 use std::collections::HashMap;
@@ -200,10 +201,13 @@ impl ComputeServer {
                 self.disable_rpc_auth,
                 self.integration_testing,
             );
+            let flight_handler = FlightSessionHandler::try_new(&self.engine).await?;
+
             tokio::spawn(async move {
                 let mut server = Server::builder()
                     .trace_fn(|_| debug_span!("rpc_service_request"))
-                    .add_service(ExecutionServiceServer::new(handler));
+                    .add_service(ExecutionServiceServer::new(handler))
+                    .add_service(FlightServiceServer::new(flight_handler));
 
                 // Add in the simple interface if requested.
                 if self.enable_simple_query_rpc {

crates/pgsrv/src/handler.rs

@@ -461,10 +461,7 @@ where
             const UNNAMED: String = String::new();
 
             // Parse...
-            if let Err(e) = session
-                .prepare_statement(UNNAMED, Some(stmt), Vec::new())
-                .await
-            {
+            if let Err(e) = session.prepare_statement(UNNAMED, stmt, Vec::new()).await {
                 self.send_error(e.into()).await?;
                 return self.ready_for_query().await;
             };

crates/rpcsrv/Cargo.toml

@@ -27,3 +27,4 @@ tonic = { workspace = true }
 bytes = "1.4"
 futures = "0.3.29"
 dashmap = "5.5.0"
+arrow-flight = { version = "47.0.0", features = ["flight-sql-experimental"] }

crates/rpcsrv/src/errors.rs

@@ -1,3 +1,5 @@
+use datafusion::arrow::error::ArrowError;
+
 #[derive(Debug, thiserror::Error)]
 pub enum RpcsrvError {
     #[error("Invalid {0} id: {1}")]
@@ -53,6 +55,11 @@ pub enum RpcsrvError {
 
     #[error("{0}")]
     Internal(String),
+
+    #[error("{0}")]
+    ParseError(String),
+    #[error("{0}")]
+    ArrowError(ArrowError),
 }
 
 pub type Result<T, E = RpcsrvError> = std::result::Result<T, E>;