delay auth check for binding parameterized queries (spiceai#5475)

Author: kczimm

crates/runtime/src/flight/do_put.rs

@@ -54,25 +54,6 @@ pub(crate) async fn handle(
     flight_svc: &Service,
     request: Request<Streaming<FlightData>>,
 ) -> Result<Response<<Service as FlightService>::DoPutStream>, Status> {
-    match RequestContext::current(crate::request::AsyncMarker::new().await).auth_principal() {
-        Some(principal) => {
-            if !principal
-                .groups()
-                .iter()
-                .any(|group| *group == "write" || *group == "read_write")
-            {
-                return Err(Status::permission_denied(
-                    "Write access denied. Verify that authentication key used has write access and try again.",
-                ));
-            }
-        }
-        None => {
-            return Err(Status::unauthenticated(
-                "Flight DoPut requires authentication.\nFor auth details, visit https://spiceai.org/docs/api/auth",
-            ));
-        }
-    }
-
     let mut streaming_flight = request.into_inner().peekable();
 
     // We need to peek at the stream in case we branch below to prepared statements
@@ -94,6 +75,25 @@ pub(crate) async fn handle(
         }
     }
 
+    match RequestContext::current(crate::request::AsyncMarker::new().await).auth_principal() {
+        Some(principal) => {
+            if !principal
+                .groups()
+                .iter()
+                .any(|group| *group == "write" || *group == "read_write")
+            {
+                return Err(Status::permission_denied(
+                    "Write access denied. Verify that authentication key used has write access and try again.",
+                ));
+            }
+        }
+        None => {
+            return Err(Status::unauthenticated(
+                "Flight DoPut requires authentication.\nFor auth details, visit https://spiceai.org/docs/api/auth",
+            ));
+        }
+    }
+
     // Since it is not a prepared statement we can take from the stream
     let Some(Ok(first_message)) = streaming_flight.next().await else {
         let _start = metrics::track_flight_request("do_put", None);