Phase 2: DataFusion query engine over Parquet

- queryd: SessionContext with custom URL scheme to avoid path doubling with LocalFileSystem
- queryd: ListingTable registration from catalog ObjectRefs with schema inference
- queryd: POST /query/sql returns JSON {columns, rows, row_count}
- queryd→catalogd wiring: reads all datasets, registers as named tables
- gateway: wires QueryEngine with shared store + registry
- e2e verified: SELECT *, WHERE/ORDER BY, COUNT/AVG all correct

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

crates/queryd/src/service.rs

@@ -1,9 +1,89 @@
-use axum::{Router, routing::get};
+use arrow::array::RecordBatch;
+use arrow::json::writer::{JsonArray, Writer as JsonWriter};
+use axum::{
+    Json, Router,
+    extract::State,
+    http::StatusCode,
+    response::IntoResponse,
+    routing::{get, post},
+};
+use serde::{Deserialize, Serialize};
 
-pub fn router() -> Router {
-    Router::new().route("/health", get(health))
+use crate::context::QueryEngine;
+
+pub fn router(engine: QueryEngine) -> Router {
+    Router::new()
+        .route("/health", get(health))
+        .route("/sql", post(execute_query))
+        .with_state(engine)
 }
 
 async fn health() -> &'static str {
     "queryd ok"
 }
+
+#[derive(Deserialize)]
+struct QueryRequest {
+    sql: String,
+}
+
+#[derive(Serialize)]
+struct QueryResponse {
+    columns: Vec<ColumnInfo>,
+    rows: serde_json::Value,
+    row_count: usize,
+}
+
+#[derive(Serialize)]
+struct ColumnInfo {
+    name: String,
+    data_type: String,
+}
+
+fn batches_to_json(batches: &[RecordBatch]) -> Result<serde_json::Value, String> {
+    let mut buf = Vec::new();
+    let mut writer = JsonWriter::<_, JsonArray>::new(&mut buf);
+    for batch in batches {
+        writer.write(batch).map_err(|e| format!("JSON write error: {e}"))?;
+    }
+    writer.finish().map_err(|e| format!("JSON finish error: {e}"))?;
+    drop(writer);
+    serde_json::from_slice(&buf).map_err(|e| format!("JSON parse error: {e}"))
+}
+
+async fn execute_query(
+    State(engine): State<QueryEngine>,
+    Json(req): Json<QueryRequest>,
+) -> impl IntoResponse {
+    tracing::info!("executing query: {}", req.sql);
+
+    match engine.query(&req.sql).await {
+        Ok(batches) => {
+            if batches.is_empty() {
+                return Ok(Json(QueryResponse {
+                    columns: vec![],
+                    rows: serde_json::Value::Array(vec![]),
+                    row_count: 0,
+                }));
+            }
+
+            let schema = batches[0].schema();
+            let columns: Vec<ColumnInfo> = schema.fields().iter().map(|f| ColumnInfo {
+                name: f.name().clone(),
+                data_type: f.data_type().to_string(),
+            }).collect();
+
+            let rows = batches_to_json(&batches)
+                .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e))?;
+
+            let row_count = rows.as_array().map(|a| a.len()).unwrap_or(0);
+
+            Ok(Json(QueryResponse {
+                columns,
+                rows,
+                row_count,
+            }))
+        }
+        Err(e) => Err((StatusCode::BAD_REQUEST, e)),
+    }
+}