fix: deepsource

Author: parmesant

src/handlers/http/query.rs

@@ -117,7 +117,7 @@ pub async fn get_records_and_fields(
     Ok((Some(records), Some(fields)))
 }
 
-#[tracing::instrument(name = "query", skip(req, query_request), fields(otel.kind = "server", query.sql = %query_request.query, query.streaming = query_request.streaming))]
+#[tracing::instrument(name = "query", skip(req, query_request), fields(otel.kind = "server", query.sql = %query_request.query, query.streaming = query_request.streaming, query.start_time = query_request.start_time, query.end_time = query_request.end_time))]
 pub async fn query(req: HttpRequest, query_request: Query) -> Result<HttpResponse, QueryError> {
     let mut session_state = QUERY_SESSION.get_ctx().state();
     let time_range =
@@ -497,8 +497,18 @@ pub async fn create_streams_for_distributed(
     }
 
     while let Some(result) = join_set.join_next().await {
-        if let Err(join_error) = result {
-            warn!("Task join error: {}", join_error);
+        match result {
+            Err(join_error) => {
+                warn!("Task join error: {}", join_error);
+            }
+            Ok((stream_name, Err(e))) => {
+                return Err(QueryError::Anyhow(anyhow::anyhow!(
+                    "Failed to create stream '{}': {}",
+                    stream_name,
+                    e
+                )));
+            }
+            Ok((_, Ok(_))) => {}
         }
     }
 

src/query/mod.rs

@@ -78,7 +78,8 @@ use crate::utils::time::TimeRange;
 // pub static QUERY_SESSION: Lazy<SessionContext> =
 //     Lazy::new(|| Query::create_session_context(PARSEABLE.storage()));
 
-pub type RB = Either<
+/// Takes care of both streaming and non-streaming query flows
+pub type QueryResult = Either<
     Vec<RecordBatch>,
     Pin<
         Box<
@@ -159,7 +160,7 @@ pub async fn execute(
     query: Query,
     is_streaming: bool,
     tenant_id: &Option<String>,
-) -> Result<(RB, Vec<String>), ExecuteError> {
+) -> Result<(QueryResult, Vec<String>), ExecuteError> {
     let id = tenant_id.clone();
 
     // W3C TraceContext propagation across QUERY_RUNTIME (separate OS-thread runtime).
@@ -293,7 +294,7 @@ impl Query {
         &self,
         is_streaming: bool,
         tenant_id: &Option<String>,
-    ) -> Result<(RB, Vec<String>), ExecuteError> {
+    ) -> Result<(QueryResult, Vec<String>), ExecuteError> {
         let df = QUERY_SESSION
             .get_ctx()
             .execute_logical_plan(self.final_logical_plan(tenant_id))

src/telemetry.rs

@@ -25,6 +25,9 @@ use opentelemetry_sdk::{
     trace::{BatchSpanProcessor, SdkTracerProvider},
 };
 
+const EXPORTER_OTLP_ENDPOINT: &str = "OTEL_EXPORTER_OTLP_ENDPOINT";
+const EXPORTER_OTLP_PROTOCOL: &str = "OTEL_EXPORTER_OTLP_PROTOCOL";
+
 /// Initialise an OTLP tracer provider.
 ///
 /// **Required env var:**
@@ -48,10 +51,10 @@ use opentelemetry_sdk::{
 pub fn init_otel_tracer() -> Option<SdkTracerProvider> {
     // Only used to decide whether OTEL is enabled; the SDK reads it again
     // from env to build the exporter (which also appends /v1/traces for HTTP).
-    std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok()?;
+    std::env::var(EXPORTER_OTLP_ENDPOINT).ok()?;
 
     let protocol =
-        std::env::var("OTEL_EXPORTER_OTLP_PROTOCOL").unwrap_or_else(|_| "http/json".to_string());
+        std::env::var(EXPORTER_OTLP_PROTOCOL).unwrap_or_else(|_| "http/json".to_string());
 
     // Build the exporter using the SDK's env-var-aware builders.
     // We intentionally do NOT call .with_endpoint() / .with_headers() /
@@ -69,10 +72,18 @@ pub fn init_otel_tracer() -> Option<SdkTracerProvider> {
         // ── HTTP/JSON (default) ──────────────────────────────────────────────
         // Default when OTEL_EXPORTER_OTLP_PROTOCOL is unset.
         // Required for Parseable OSS — it only accepts application/json.
-        _ => SpanExporter::builder()
+        "http/json" => SpanExporter::builder()
             .with_http()
             .with_protocol(Protocol::HttpJson)
             .build(),
+        other => {
+            tracing::warn!(
+                "Unknown OTEL_EXPORTER_OTLP_PROTOCOL value '{}'; disabling OTEL tracing. \
+                 Supported values: grpc, http/protobuf, http/json",
+                other
+            );
+            return None;
+        }
     };
 
     let exporter = exporter