nordisk/skade

skade-flight/tests/flight_sql_roundtrip.rs

//! End to end over real TCP with the stock `FlightSqlServiceClient` — the
//! same handshake, introspection and query shape Power BI's connector and
//! the Flight SQL ODBC driver drive.

use std::sync::Arc;
use std::time::Duration;

use arrow_array::{Float64Array, Int64Array, RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema};
use skade_flight::arrow_flight::sql::client::FlightSqlServiceClient;
use skade_flight::arrow_flight::sql::{CommandGetDbSchemas, CommandGetTables, CommandGetXdbcTypeInfo};
use skade_flight::{Auth, Engine};
use futures::TryStreamExt;
use tonic::transport::Channel;

fn people() -> (Schema, RecordBatch) {
    let schema = Schema::new(vec![
        Field::new("id", DataType::Int64, false),
        Field::new("name", DataType::Utf8, false),
        Field::new("score", DataType::Float64, true),
    ]);
    let batch = RecordBatch::try_new(
        Arc::new(schema.clone()),
        vec![
            Arc::new(Int64Array::from(vec![1, 2, 3, 4, 5])),
            Arc::new(StringArray::from(vec!["ada", "bob", "cyd", "dee", "eve"])),
            Arc::new(Float64Array::from(vec![Some(9.5), Some(3.0), None, Some(7.25), Some(1.0)])),
        ],
    )
    .unwrap();
    (schema, batch)
}

fn cities() -> (Schema, RecordBatch) {
    let schema = Schema::new(vec![
        Field::new("person_id", DataType::Int64, false),
        Field::new("city", DataType::Utf8, false),
    ]);
    let batch = RecordBatch::try_new(
        Arc::new(schema.clone()),
        vec![
            Arc::new(Int64Array::from(vec![1, 2, 4])),
            Arc::new(StringArray::from(vec!["Umeå", "Luleå", "Umeå"])),
        ],
    )
    .unwrap();
    (schema, batch)
}

struct Live {
    _dir: tempfile::TempDir,
    engine: Arc<Engine>,
    client: FlightSqlServiceClient<Channel>,
    _server: tokio::task::JoinHandle<()>,
}

async fn bring_up(auth: Auth) -> Live {
    bring_up_with(auth, Duration::from_millis(50)).await
}

async fn bring_up_with(auth: Auth, refresh_after: Duration) -> Live {
    let dir = tempfile::tempdir().unwrap();
    let wh = skade::Warehouse::open(dir.path().join("wh")).await.unwrap();
    let (s, b) = people();
    wh.create_table("people", &s).await.unwrap().append(&[b]).await.unwrap();
    let (s, b) = cities();
    wh.create_table("cities", &s).await.unwrap().append(&[b]).await.unwrap();

    let engine = Arc::new(Engine::with_options(wh, refresh_after, 2).await.unwrap());
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let auth = Arc::new(auth);
    let server = {
        let engine = engine.clone();
        tokio::spawn(async move {
            skade_flight::serve_on(engine, auth, listener, Default::default(), futures::future::pending()).await.unwrap();
        })
    };
    let channel = Channel::from_shared(format!("http://{addr}")).unwrap().connect().await.unwrap();
    Live {
        _dir: dir,
        engine,
        client: FlightSqlServiceClient::new(channel),
        _server: server,
    }
}

async fn fetch(client: &mut FlightSqlServiceClient<Channel>, sql: &str) -> Vec<RecordBatch> {
    let info = client.execute(sql.to_string(), None).await.unwrap();
    let ticket = info.endpoint[0].ticket.clone().unwrap();
    client.do_get(ticket).await.unwrap().try_collect().await.unwrap()
}

/// `fetch`, but the failure is the answer: describe (GetFlightInfo) then
/// execute (DoGet), returning whichever RPC refused.
async fn try_fetch(client: &mut FlightSqlServiceClient<Channel>, sql: &str) -> Result<Vec<RecordBatch>, String> {
    let info = client.execute(sql.to_string(), None).await.map_err(|e| e.to_string())?;
    let ticket = info.endpoint[0].ticket.clone().unwrap();
    client.do_get(ticket).await.map_err(|e| e.to_string())?.try_collect().await.map_err(|e| e.to_string())
}

fn rows(batches: &[RecordBatch]) -> usize {
    batches.iter().map(|b| b.num_rows()).sum()
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn power_bi_shape_over_open_server() {
    let mut live = bring_up(Auth::open()).await;
    let c = &mut live.client;

    // Introspection.
    let info = c.get_db_schemas(CommandGetDbSchemas::default()).await.unwrap();
    let b = c.do_get(info.endpoint[0].ticket.clone().unwrap()).await.unwrap().try_collect::<Vec<_>>().await.unwrap();
    assert_eq!(rows(&b), 1, "one namespace (main)");

    let info = c.get_tables(CommandGetTables { include_schema: true, ..Default::default() }).await.unwrap();
    let b = c.do_get(info.endpoint[0].ticket.clone().unwrap()).await.unwrap().try_collect::<Vec<_>>().await.unwrap();
    assert_eq!(rows(&b), 2, "cities + people");
    let names = b[0].column_by_name("table_name").unwrap();
    let names = names.as_any().downcast_ref::<StringArray>().unwrap();
    assert_eq!(names.value(0), "cities");
    assert_eq!(names.value(1), "people");

    let info = c.get_sql_info(vec![]).await.unwrap();
    let b = c.do_get(info.endpoint[0].ticket.clone().unwrap()).await.unwrap().try_collect::<Vec<_>>().await.unwrap();
    assert!(rows(&b) > 60, "sql_info rows {}", rows(&b));

    let info = c.get_xdbc_type_info(CommandGetXdbcTypeInfo::default()).await.unwrap();
    let b = c.do_get(info.endpoint[0].ticket.clone().unwrap()).await.unwrap().try_collect::<Vec<_>>().await.unwrap();
    assert_eq!(rows(&b), 13);

    // Power BI asks per type code: a known code gives its row, ODBC's
    // SQL_ALL_TYPES (0) and a code we do not carry give the whole table —
    // never an empty stream, which the ADBC layer turns into a null and a crash.
    for (dt, want) in [(93, 1), (3, 1), (0, 13), (9, 13), (12345, 13)] {
        let info = c.get_xdbc_type_info(CommandGetXdbcTypeInfo { data_type: Some(dt) }).await.unwrap();
        let b = c.do_get(info.endpoint[0].ticket.clone().unwrap()).await.unwrap().try_collect::<Vec<_>>().await.unwrap();
        assert_eq!(rows(&b), want, "GetXdbcTypeInfo(data_type={dt})");
    }

    // The driver's first query.
    let b = fetch(c, r#"SELECT * FROM "people""#).await;
    assert_eq!(rows(&b), 5);
    assert_eq!(b[0].schema().fields().len(), 3);

    // Folded: projection + filter + order + limit.
    let b = fetch(c, "SELECT name, score FROM people WHERE score > 2 ORDER BY score DESC LIMIT 2").await;
    assert_eq!(rows(&b), 2);
    let n = b[0].column(0).as_any().downcast_ref::<StringArray>().unwrap();
    assert_eq!(n.value(0), "ada");
    assert_eq!(n.value(1), "dee");

    // Aggregate.
    let b = fetch(c, "SELECT count(*) AS n, avg(score) AS m FROM people").await;
    let n = b[0].column(0).as_any().downcast_ref::<Int64Array>().unwrap();
    assert_eq!(n.value(0), 5);

    // Join across two Iceberg tables, qualified and bare names both resolve.
    let b = fetch(
        c,
        "SELECT p.name, c.city FROM skade.main.people p JOIN cities c ON c.person_id = p.id ORDER BY p.name",
    )
    .await;
    assert_eq!(rows(&b), 3);

    // Prepared statement path (what ADBC drivers use).
    let mut ps = c.prepare("SELECT id FROM people WHERE id <= 2".to_string(), None).await.unwrap();
    assert_eq!(ps.dataset_schema().unwrap().fields().len(), 1);
    let info = ps.execute().await.unwrap();
    let b = c.do_get(info.endpoint[0].ticket.clone().unwrap()).await.unwrap().try_collect::<Vec<_>>().await.unwrap();
    assert_eq!(rows(&b), 2);
    ps.close().await.unwrap();

    // Bad SQL is the caller's fault and says so.
    let err = c.execute("SELEKT 1".to_string(), None).await.unwrap_err();
    assert!(err.to_string().contains("SELEKT") || err.to_string().to_lowercase().contains("sql"), "{err}");

    // Read-only: an update is refused by name.
    let err = c.execute_update("DELETE FROM people".to_string(), None).await.unwrap_err();
    assert!(err.to_string().contains("read-only"), "{err}");

    // A commit after the session was built becomes visible after the refresh window.
    // (Through the engine's own warehouse: redb is single-writer, the ONE live
    // catalog — a second `Warehouse::open` on the same dir is refused by lock.)
    let (s, b) = people();
    live.engine.warehouse().create_table("late", &s).await.unwrap().append(&[b]).await.unwrap();
    tokio::time::sleep(Duration::from_millis(80)).await;
    let b = fetch(c, "SELECT count(*) FROM late").await;
    let n = b[0].column(0).as_any().downcast_ref::<Int64Array>().unwrap();
    assert_eq!(n.value(0), 5);
    assert_eq!(live.engine.tables().await.unwrap().len(), 3);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn basic_handshake_issues_bearer_and_gates_calls() {
    let mut live = bring_up(Auth::bearer("t0k3n").with_basic("pbi", "hunter2")).await;
    let c = &mut live.client;

    // No credentials: refused.
    let err = c.execute("SELECT 1".to_string(), None).await.unwrap_err();
    assert!(err.to_string().contains("authorization"), "{err}");

    // Wrong password: refused at the handshake.
    assert!(c.handshake("pbi", "nope").await.is_err());

    // Right Basic → bearer, and the client replays it. The bearer arrives in the
    // `authorization` RESPONSE HEADER (what every client authenticates with); the
    // handshake deliberately sends no HandshakeResponse message (see
    // `do_handshake`), so the payload arrow-rs returns is empty. Assert the gate:
    // the next call is accepted, and the header carried the real token.
    c.handshake("pbi", "hunter2").await.unwrap();
    assert_eq!(c.token().map(String::as_str), Some("t0k3n"));
    let b = fetch(c, "SELECT count(*) FROM people").await;
    assert_eq!(rows(&b), 1);
}

/// The navigator's key questions: identifier fields answer as the primary
/// key, foreign keys are an EMPTY set with the spec schema — never
/// `UNIMPLEMENTED`.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn keys_metadata_answers_navigator() {
    use skade_flight::arrow_flight::sql::{CommandGetExportedKeys, CommandGetImportedKeys, CommandGetPrimaryKeys};
    let mut live = bring_up(Auth::open()).await;
    // A table WITH identifier fields.
    let (s, b) = people();
    live.engine
        .warehouse()
        .create_table_with_identity("keyed", &s, &["id"])
        .await
        .unwrap()
        .append(&[b])
        .await
        .unwrap();
    live.engine.refresh().await.unwrap();
    let c = &mut live.client;

    let pk = |table: &str| CommandGetPrimaryKeys { catalog: None, db_schema: None, table: table.to_string() };
    let info = c.get_primary_keys(pk("keyed")).await.unwrap();
    let b: Vec<RecordBatch> =
        c.do_get(info.endpoint[0].ticket.clone().unwrap()).await.unwrap().try_collect().await.unwrap();
    assert_eq!(rows(&b), 1);
    let col = b[0].column_by_name("column_name").unwrap();
    assert_eq!(col.as_any().downcast_ref::<StringArray>().unwrap().value(0), "id");

    // No identifier fields → empty, same schema. Unknown table → empty too.
    // (A zero-row result is a schema message and no data batch, so the
    // schema is read off the stream, exactly as a driver does.)
    for t in ["people", "nope"] {
        let info = c.get_primary_keys(pk(t)).await.unwrap();
        let (n, fields) = drain(c, info).await;
        assert_eq!(n, 0, "{t}");
        assert_eq!(fields, 6, "{t}");
    }

    let info = c
        .get_exported_keys(CommandGetExportedKeys { catalog: None, db_schema: None, table: "keyed".into() })
        .await
        .unwrap();
    let (n, fields) = drain(c, info).await;
    assert_eq!((n, fields), (0, 13));
    let info = c
        .get_imported_keys(CommandGetImportedKeys { catalog: None, db_schema: None, table: "keyed".into() })
        .await
        .unwrap();
    let (n, _) = drain(c, info).await;
    assert_eq!(n, 0);
}

/// TLS: a self-signed server cert, the client trusting exactly that cert,
/// Basic handshake over the encrypted channel. What the ODBC driver does by
/// default (`useEncryption=true`).
#[cfg(feature = "tls")]
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn tls_roundtrip_with_self_signed_cert() {
    use tonic::transport::{Certificate, ClientTlsConfig};
    let ck = rcgen::generate_simple_self_signed(vec!["localhost".to_string()]).unwrap();
    let cert_pem = ck.cert.pem();
    let key_pem = ck.signing_key.serialize_pem();

    let dir = tempfile::tempdir().unwrap();
    let wh = skade::Warehouse::open(dir.path().join("wh")).await.unwrap();
    let (s, b) = people();
    wh.create_table("people", &s).await.unwrap().append(&[b]).await.unwrap();
    let engine = Arc::new(Engine::new(wh).await.unwrap());
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let port = listener.local_addr().unwrap().port();
    let auth = Arc::new(Auth::bearer("t0k3n").with_basic("pbi", "hunter2"));
    let opts = skade_flight::ServeOptions {
        tls: Some(skade_flight::server::TlsOptions {
            cert_pem: cert_pem.clone().into_bytes(),
            key_pem: key_pem.into_bytes(),
            client_ca_pem: None,
        }),
    };
    let _server = tokio::spawn(skade_flight::serve_on(engine, auth, listener, opts, futures::future::pending()));

    let tls = ClientTlsConfig::new().ca_certificate(Certificate::from_pem(cert_pem)).domain_name("localhost");
    let channel = Channel::from_shared(format!("https://localhost:{port}"))
        .unwrap()
        .tls_config(tls)
        .unwrap()
        .connect()
        .await
        .unwrap();
    let mut c = FlightSqlServiceClient::new(channel);
    c.handshake("pbi", "hunter2").await.unwrap();
    assert_eq!(c.token().map(String::as_str), Some("t0k3n"));
    let b = fetch(&mut c, "SELECT count(*) FROM people").await;
    assert_eq!(rows(&b), 1);

    // Plaintext against the TLS port must fail, not hang.
    let plain = Channel::from_shared(format!("http://127.0.0.1:{port}")).unwrap().connect().await;
    match plain {
        Ok(ch) => {
            let mut c = FlightSqlServiceClient::new(ch);
            assert!(c.handshake("pbi", "hunter2").await.is_err());
        }
        Err(_) => {}
    }
}

/// Drain a FlightInfo's first endpoint: `(rows, schema field count)`, the
/// schema taken from the stream itself so a zero-row result still has one.
async fn drain(c: &mut FlightSqlServiceClient<Channel>, info: skade_flight::arrow_flight::FlightInfo) -> (usize, usize) {
    let mut stream = c.do_get(info.endpoint[0].ticket.clone().unwrap()).await.unwrap();
    let mut n = 0;
    while let Some(b) = stream.try_next().await.unwrap() {
        n += b.num_rows();
    }
    let fields = stream.schema().map(|s| s.fields().len()).unwrap_or(0);
    (n, fields)
}

/// **TPC-H Q15's shape — `CREATE VIEW` / query the view / `DROP VIEW` — runs,
/// and the view never reaches the warehouse.**
///
/// MEASURED (t14s, 2026-09-10, the 22-query Flight SQL run): 21/22, Q15 refused
/// because its spec form creates a view. "Read-only" and "cannot hold a
/// session-scoped view" are not the same fact: the warehouse is never written
/// by a view, so refusing it was the engine confusing its own session with the
/// catalog. RED direction: strip the view overlay and the first `execute` fails
/// with iceberg-datafusion's "register table" refusal.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn session_views_serve_q15s_shape_and_never_touch_the_catalog() {
    let mut live = bring_up(Auth::open()).await;
    let c = &mut live.client;

    // 1. CREATE VIEW over an aggregate — Q15's `revenue0`. Goes through the
    //    statement door exactly as an ODBC SQLExecDirect sends it.
    let b = fetch(c, "CREATE VIEW top_scores AS SELECT id AS person, score * 2 AS doubled FROM people WHERE score IS NOT NULL").await;
    assert_eq!(rows(&b), 0, "DDL streams no rows");

    // 2. Query it, unqualified, joined with a real table, with a scalar
    //    subquery over the same view (Q15's `= (SELECT max(...) FROM revenue0)`).
    let b = fetch(
        c,
        "SELECT p.name, v.doubled FROM top_scores v JOIN people p ON p.id = v.person \
         WHERE v.doubled = (SELECT max(doubled) FROM top_scores)",
    )
    .await;
    assert_eq!(rows(&b), 1);
    let n = b[0].column(0).as_any().downcast_ref::<StringArray>().unwrap();
    assert_eq!(n.value(0), "ada");

    // 3. It is a SESSION view: not in the warehouse, not advertised as a table.
    let wh_tables = live.engine.warehouse().table_idents().await.unwrap();
    assert!(!wh_tables.iter().any(|t| t.name() == "top_scores"), "the catalog must not hold it: {wh_tables:?}");
    assert_eq!(live.engine.tables().await.unwrap().len(), 2, "GetTables advertises warehouse tables only");

    // 4. It survives the engine's session rebuild (the 50 ms refresh window
    //    here, 5 s in production) — a client's view must not vanish mid-report.
    tokio::time::sleep(Duration::from_millis(80)).await;
    live.engine.refresh().await.unwrap();
    let b = fetch(c, "SELECT count(*) FROM top_scores").await;
    let n = b[0].column(0).as_any().downcast_ref::<Int64Array>().unwrap();
    assert_eq!(n.value(0), 4);

    // 5. CREATE without OR REPLACE on an existing name is refused — on the
    //    EXECUTION RPC (DoGet): GetFlightInfo only describes, so a driver that
    //    prepares, describes and executes applies a DDL exactly once. OR
    //    REPLACE swaps it and the plan cache forgets the old shape.
    let err = try_fetch(c, "CREATE VIEW top_scores AS SELECT 1 AS x").await.unwrap_err();
    assert!(err.contains("already exists"), "{err}");
    let _ = fetch(c, "CREATE OR REPLACE VIEW top_scores AS SELECT id AS person FROM people").await;
    let b = fetch(c, "SELECT * FROM top_scores").await;
    assert_eq!(b[0].schema().fields().len(), 1, "the replaced shape is what plans now");

    // 6. A view may not shadow a warehouse table.
    let err = try_fetch(c, "CREATE VIEW people AS SELECT 1 AS x").await.unwrap_err();
    assert!(err.contains("table in the warehouse"), "{err}");

    // 7. DROP VIEW, then the name is gone; IF EXISTS is quiet, plain is not.
    let b = fetch(c, "DROP VIEW top_scores").await;
    assert_eq!(rows(&b), 0);
    assert!(try_fetch(c, "SELECT * FROM top_scores").await.is_err());
    let _ = fetch(c, "DROP VIEW IF EXISTS top_scores").await;
    let err = try_fetch(c, "DROP VIEW top_scores").await.unwrap_err();
    assert!(err.contains("no such view"), "{err}");

    // 8. The same DDL through the UPDATE door (a driver that classifies DDL as
    //    an update) is accepted, and reports 0 rows affected.
    let n = c.execute_update("CREATE VIEW via_update AS SELECT id FROM people".to_string(), None).await.unwrap();
    assert_eq!(n, 0);
    let b = fetch(c, "SELECT count(*) FROM via_update").await;
    assert_eq!(b[0].column(0).as_any().downcast_ref::<Int64Array>().unwrap().value(0), 5);

    // 9. Real writes are still refused — and the refusal now says what IS
    //    allowed, so a Q15 author reads the rewrite instead of "forbidden".
    for sql in ["DELETE FROM people", "CREATE TABLE t (x INT)", "INSERT INTO people VALUES (9, 'z', 1.0)"] {
        let err = c.execute_update(sql.to_string(), None).await.unwrap_err();
        let m = err.to_string();
        assert!(m.contains("read-only"), "{m}");
        assert!(m.contains("WITH") && m.contains("CREATE") && m.contains("VIEW"), "the refusal must name the CTE rewrite and session views: {m}");
    }
    // A real DDL through the statement door is refused already at describe
    // time — there is nothing to describe about a statement that will not run.
    let err = c.execute("CREATE TABLE t AS SELECT 1".to_string(), None).await.unwrap_err();
    assert!(err.to_string().contains("read-only"), "{err}");
    match &err {
        skade_flight::arrow_flight::error::FlightError::Tonic(st) => {
            assert_eq!(st.code(), tonic::Code::InvalidArgument, "the statement is out of scope, not the server: {err}")
        }
        other => panic!("expected a status, got {other}"),
    }
}

// ── plan once, signed handles, bound parameters ──────────────────────────

/// **One logical and one physical plan per distinct statement, however many
/// RPCs carry it.** `GetFlightInfo` plans; `DoGet` executes what it planned;
/// a second `GetFlightInfo` + `DoGet` of the same SQL plans nothing.
///
/// Before the ticket carried a signed handle to the cached plan, `DoGet`
/// rebuilt the physical plan on every execute (`DataFrame::execute_stream`
/// runs the optimizer + physical planner each call): two executes of one
/// ticket were two physical plans. RED direction: make `Engine::physical`
/// skip its `OnceCell` and the second assertion reads physical = 2.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_statement_is_planned_once_across_every_rpc_that_carries_it() {
    // A long refresh window: a rebuild drops every plan (they point at the old
    // snapshot), which is right and is not what this measures.
    let mut live = bring_up_with(Auth::open(), Duration::from_secs(600)).await;
    let sql = "SELECT name, score FROM people WHERE score > 2 ORDER BY score DESC";
    let before = live.engine.plan_counts();

    let info = live.client.execute(sql.to_string(), None).await.unwrap();
    let after_info = live.engine.plan_counts();
    assert_eq!(after_info.logical - before.logical, 1, "GetFlightInfo parses + resolves once");
    assert_eq!(after_info.physical - before.physical, 1, "GetFlightInfo optimizes + plans physically once");

    let ticket = info.endpoint[0].ticket.clone().unwrap();
    let b: Vec<RecordBatch> = live.client.do_get(ticket.clone()).await.unwrap().try_collect().await.unwrap();
    assert_eq!(rows(&b), 3);
    let b: Vec<RecordBatch> = live.client.do_get(ticket).await.unwrap().try_collect().await.unwrap();
    assert_eq!(rows(&b), 3, "the same ticket executes again");
    let info2 = live.client.execute(sql.to_string(), None).await.unwrap();
    let (n, _) = drain(&mut live.client, info2).await;
    assert_eq!(n, 3);

    let after_all = live.engine.plan_counts();
    assert_eq!(after_all, after_info, "two DoGets and a second GetFlightInfo planned NOTHING: {after_all:?}");
}

/// **A ticket is a capability, not a string.** Tampered → UNAUTHENTICATED;
/// minted under another key → UNAUTHENTICATED; the old raw-SQL shape →
/// UNAUTHENTICATED (never executed); issued to another principal →
/// UNAUTHENTICATED. RED direction: put the SQL back in the ticket and every
/// arm here executes the query instead.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_ticket_is_signed_bound_to_its_caller_and_never_raw_sql() {
    use skade_flight::arrow_flight::Ticket;
    use skade_flight::arrow_flight::sql::{Any, ProstMessageExt, TicketStatementQuery};
    use skade_flight::ticket::{Signer, TICKET_TTL};
    use skade_flight::Statement;
    use prost::Message;

    let mut live = bring_up(Auth::bearer("t0k3n").with_basic("pbi", "hunter2")).await;
    live.client.handshake("pbi", "hunter2").await.unwrap();
    let sql = "SELECT count(*) FROM people";
    let info = live.client.execute(sql.to_string(), None).await.unwrap();
    let ticket = info.endpoint[0].ticket.clone().unwrap();

    // Opaque: the SQL is not in the ticket in the clear.
    let any = Any::decode(ticket.ticket.clone()).unwrap();
    let tsq: TicketStatementQuery = any.unpack().unwrap().unwrap();
    assert!(
        !String::from_utf8_lossy(&tsq.statement_handle).contains("count(*)"),
        "the handle is an envelope, not the statement"
    );
    // and the untampered one works, twice.
    for _ in 0..2 {
        let b: Vec<RecordBatch> = live.client.do_get(ticket.clone()).await.unwrap().try_collect().await.unwrap();
        assert_eq!(rows(&b), 1);
    }

    let unauth = |err: skade_flight::arrow_flight::error::FlightError| {
        let s = err.to_string();
        assert!(s.contains("Unauthenticated") || s.contains("statement handle refused"), "{s}");
    };

    // Tampered: one byte inside the handle.
    let mut handle = tsq.statement_handle.to_vec();
    let mid = handle.len() / 2;
    handle[mid] ^= 0x01;
    let bad = Ticket::new(TicketStatementQuery { statement_handle: handle.into() }.as_any().encode_to_vec());
    unauth(live.client.do_get(bad).await.unwrap_err());

    // Minted under another key.
    let foreign = Signer::from_secret("not-this-server").issue(Statement::sql(sql), "bearer", TICKET_TTL).unwrap();
    let bad = Ticket::new(TicketStatementQuery { statement_handle: foreign }.as_any().encode_to_vec());
    unauth(live.client.do_get(bad).await.unwrap_err());

    // The old shape: raw SQL as the handle.
    let raw = Ticket::new(TicketStatementQuery { statement_handle: sql.as_bytes().to_vec().into() }.as_any().encode_to_vec());
    unauth(live.client.do_get(raw).await.unwrap_err());

    // Same key, another principal: the server's own signer, a handle issued to
    // someone the caller is not.
    let other = Signer::from_secret("t0k3n").issue(Statement::sql(sql), "mallory", TICKET_TTL).unwrap();
    let bad = Ticket::new(TicketStatementQuery { statement_handle: other }.as_any().encode_to_vec());
    unauth(live.client.do_get(bad).await.unwrap_err());
}

/// **A prepared statement advertises its placeholders and executes what was
/// bound.** `parameter_schema` is `$1: Int64` (inferred from `id <= $1`), a
/// bound 2 returns two rows, a re-bound 4 returns four — two plans, since the
/// optimizer folds the literal. RED direction: leave `parameter_schema` empty
/// and the first assertion fails; ignore the bound batch and both counts read 5.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_prepared_statement_binds_its_parameters() {
    let mut live = bring_up_with(Auth::open(), Duration::from_secs(600)).await;
    let c = &mut live.client;
    let mut ps = c.prepare("SELECT id, name FROM people WHERE id <= $1 ORDER BY id".to_string(), None).await.unwrap();
    let pschema = ps.parameter_schema().unwrap();
    assert_eq!(pschema.fields().len(), 1, "one placeholder: {pschema:?}");
    assert_eq!(pschema.field(0).name(), "$1");
    assert_eq!(pschema.field(0).data_type(), &DataType::Int64, "inferred from `id <= $1`");
    assert_eq!(ps.dataset_schema().unwrap().fields().len(), 2);

    // Unbound: refused by name, not executed as a full scan.
    let err = ps.execute().await.unwrap_err().to_string();
    assert!(err.contains("unbound parameter") && err.contains("$1"), "{err}");

    let bind = |v: i64| {
        RecordBatch::try_new(
            Arc::new(Schema::new(vec![Field::new("$1", DataType::Int64, true)])),
            vec![Arc::new(Int64Array::from(vec![v]))],
        )
        .unwrap()
    };
    let before = live.engine.plan_counts();
    ps.set_parameters(bind(2)).unwrap();
    let info = ps.execute().await.unwrap();
    let b: Vec<RecordBatch> = c.do_get(info.endpoint[0].ticket.clone().unwrap()).await.unwrap().try_collect().await.unwrap();
    assert_eq!(rows(&b), 2);
    let ids = b[0].column(0).as_any().downcast_ref::<Int64Array>().unwrap();
    assert_eq!(ids.values(), &[1, 2]);

    ps.set_parameters(bind(4)).unwrap();
    let info = ps.execute().await.unwrap();
    let b: Vec<RecordBatch> = c.do_get(info.endpoint[0].ticket.clone().unwrap()).await.unwrap().try_collect().await.unwrap();
    assert_eq!(rows(&b), 4);
    let after = live.engine.plan_counts();
    assert_eq!(after.physical - before.physical, 2, "two bindings, two plans, no more: {after:?}");
    ps.close().await.unwrap();
}