Design: std::influxdb — querying InfluxDB 3 as a DataFrame source

Status: spike complete (2026-09-22), ready for implementation

Goal: add a new stdlib source function alongside std::df::fromcsv — a
std::influxdb module whose query function returns a DataFrame<'a>
built by querying an InfluxDB 3 database over its Arrow Flight API.

0. Design constraint (per Nathaniel, 2026-09-22)

The host does not accept an arbitrary query string. The query is always
of the form SELECT <cols> FROM <table>, where the selected columns are
known statically: they are the field labels of the DataFrame<'a> row type
at the call site (the same type-spec mechanism fromcsv uses for
read-time validation). This removes SQL injection surface, makes the result
schema statically determined, and lets the compiler's type-checker (not the
server) reject unknown columns for the caller.

1. How fromcsv works today (the template)

A stdlib module is three layers:

1. Guest wrapper — stdlib/src/df.blr. Declares resource linear Frame, the extern fn fromcsv_private(path: string, row: reflect::TypeSpec) -> Frame; import, and
   pub fn fromcsv(path: string) -> DataFrame<'a> { to_dataframe(fromcsv_private(path, type)) }.
   The type in value position closes the caller's inferred row type into a
   type-spec node list passed to the host.
2. WIT boundary — stdlib/wit/df.wit. Package blr:df, interface
   df-interface (resources frame/cached-frame/table, plus
   fromcsv-private: func(path: string, row: list<type-spec-node>) -> frame),
   and world df which imports and re-exports the interface.
3. Host implementation — stdlib/src/df.rs.
   wasmtime::component::bindgen!({ world: "df", with: { frame: DataFrame, … } })
   maps the frame resource to a DataFusion DataFrame held in a
   ResourceTable owned by df::Host. fromcsv_private reads via a DataFusion
   SessionContext, validates the schema against the row type-spec at read
   time (validate_csv_schema), and pushes the frame. lang/src/exec.rs
   links the world: Df::add_to_linker::<_, HasSelf<_>>(linker, |state| &mut state.df).

influxdb::query follows the same shape; the file read is replaced by a
Flight DoGet, and the type-spec is used to build the query, not just
validate it.

2. Spike results (verified 2026-09-22)

Environment: quay.io/influxdb/influxdb3-core:latest
(= 3.12.0-nightly, serve --node-id spike --object-store memory),
client = scratch bin on arrow-flight 57 + tonic 0.14
(kept at /tmp/spike while the machine is up; the essential code is in
§4.3). Server source checked against main and tag v3.11.4
(core/service_grpc_flight/, influxdb3/src/commands/serve/).

2.1 Endpoint & transport

• The public Arrow Flight service is multiplexed onto the HTTP listener.
  v3.11.4 source: single combined "HTTP/gRPC service",
  DEFAULT_HTTP_BIND_ADDR = "0.0.0.0:8181" (Core). There is no separate
  gRPC bind flag anymore. Earlier 3.x docs mention gRPC port 9092; in the
  current nightly 9092 is an internal gRPC port that rejects public Flight
  requests ("operation was cancelled"). The user-supplied endpoint is the
  HTTP address (e.g. localhost:8181), which covers both port layouts.
• Plaintext gRPC (no TLS in the spike; TLS = later).
• Auth: gRPC metadata authorization: Bearer <token>
  (FlightClient::add_header). Wrong/missing token → clean
  Unauthenticated status: "The request does not have valid
  authentication credentials".

2.2 Request format — no GetFlightInfo, direct DoGet with a ticket

Server code (core/service_grpc_flight/src/request.rs, identical on
main and v3.11.4) documents the "native IOx API": encode the query as
a Ticket, call DoGet(Ticket) directly, receive a FlightData stream.
GetFlightInfo is only part of the separate FlightSQL flow.

The ticket may be encoded three ways; the server tries them in order:

1. influxdata.iox.querier.v1.ReadInfo protobuf wrapped in a protobuf Any
   (type.googleapis.com/influxdata.iox.querier.v1.ReadInfo),
2. bare ReadInfo protobuf,
3. JSON (the encoding the official Go client uses):

{ "database": "spike", "sql_query": "SELECT host, usage FROM cpu", "query_type": "sql" }

We use the JSON ticket — no vendored proto, no prost codegen, and the
format is documented server-side as a client contract. ReadInfo
protobuf remains the upgrade path if JSON is ever dropped.

2.3 SQL, not InfluxQL

• query_type: "sql" returns exactly the selected columns.
• query_type: "influxql" adds a spurious iox::measurement column →
  breaks the "columns == declared row fields" invariant. SQL is the wire
  language.
• Identifier quoting: "host"/"cpu" double-quotes work; unquoted works
  for simple names. blr labels are snake_case, so the host emits labels
  unquoted and the user-supplied table name as-is (quoted only if it is not
  a simple [a-z][a-z0-9_]* identifier).
• Missing column → InvalidArgument: "No field named nope. Valid fields
  are cpu.active, cpu.host, …"; missing table → "table
  'public.iox.my table' not found". Both surface as tonic Status — ideal
  for panic messages. (The compiler already rejects unknown labels for
  the caller; these catch table/renamed-column mismatches at runtime.)

2.4 Result stream

• DoGet yields a multi-batch stream (5 rows arrived as 3+1+1 batches
  in the spike) → host must concat_batches (same as collect_private).
• Columns come back sorted alphabetically by name — identical to blr's
  canonical (sorted-by-label) row order. If the host emits the SELECT list
  in sorted-label order, input, output, and blr row order all agree.
• Column types (from the spike's cpu table — LP-written tags + fields):
  • tags → Utf8, nullable
  • fields → Int64 / Float64 / Boolean, nullable
  • time → Timestamp(Nanosecond, None), non-nullable
  • no dictionary encoding (tags are plain Utf8)
  • every field carries iox::column::type metadata
    (…::tag, …::timestamp, …::field::{integer,float,boolean}) —
    a server-authoritative column-kind hint the host may use for
    diagnostics.
• Missing field values arrive as nulls (a row written without usage
  → usage = NULL). blr has no null/option today and collect panics on
  nulls; v1 panics with a clear message (see §4.3 step 5).
• An empty result is a zero-batch stream with no schema at all. The
  host synthesizes the frame schema from the type-spec instead (§4.3 step
  6).

3. Proposed user-facing API

import std::influxdb

pub fn main() -> df.DataFrame<'a> {
    influxdb::query("localhost:8181", "my-token", "spike", "cpu")
}

• influxdb::query(endpoint: string, token: string, database: string, table: string) -> DataFrame<'a> — linear DataFrame, behaves exactly
  like a fromcsv result (filter/map/join/aggregate/collect all work).
• The selected columns are the row fields of DataFrame<'a> as
  type-checked at the call site: the type-spec's row labels become the
  SELECT list, so DataFrame<'a> for a 3-field row issues
  SELECT a, b, c FROM <table>.
  An open row (DataFrame<{'a .. 'b}>-style, RowOpen type-spec —
  "nothing selected") issues SELECT * FROM "table".
• token is an empty string for unauthenticated servers (--without-auth
  sentinel convention, like limit's fetch == 0).
• Column type compatibility (int/float mutual, timestamp-as-int, …) is
  still validated against the returned schema at query time, reusing
  fromcsv's validator; extra server columns are impossible by
  construction for a closed row (for SELECT *, unselected columns are
  ignored, exactly like fromcsv's extra-file-columns rule).
• Row order is unspecified (InfluxDB makes no ordering promise without
  ORDER BY); users df::sort as needed — same as a CSV with no sort.

Deliberately deferred: Config record (TLS, timeouts), WHERE/filter
pushdown, InfluxQL.

4. Architecture

4.1 WIT — new stdlib/wit/influxdb.wit

package blr:influxdb {
    interface influxdb-interface {
        use blr:df/df-interface.{frame};
        use blr:reflect/reflect-interface.{type-spec-node};

        // The query read: the row type-spec both selects the columns of
        // the generated `SELECT … FROM` and is validated against the
        // result schema, like fromcsv-private validates the file schema.
        query-private: func(
            endpoint: string,
            token: string,
            database: string,
            table: string,
            row: list<type-spec-node>,
        ) -> frame;
    }
}

world influxdb {
    import blr:df/df-interface;
    import blr:reflect/reflect-interface;
    import blr:influxdb/influxdb-interface;
    export blr:influxdb/influxdb-interface;
}

wasmtime::component::bindgen! picks it up automatically from wit/.

4.2 Guest — new stdlib/src/influxdb.blr

import std::df
import std::reflect

extern fn query_private(endpoint: string, token: string, database: string, table: string, row: reflect::TypeSpec) -> df.Frame;

pub fn query(endpoint: string, token: string, database: string, table: string) -> df.DataFrame<'a> {
    to_dataframe(query_private(endpoint, token, database, table, type))
}

DataFrame/to_dataframe are compiler builtins bridging the typed
DataFrame<'a> and the Frame resource (lang/src/runtime/df.rs), so the
wrapper is a thin cast. The new embedded stdlib file makes
import std::influxdb resolve automatically.

4.3 Host — new stdlib/src/influxdb.rs

wasmtime::component::bindgen!({
    world: "influxdb",
    imports: { default: async },
    with: { "blr:df/df-interface.frame": DataFrame },
});

Shared frame table. Frame reps must resolve in one ResourceTable:
frames produced here are consumed by df::filter etc. via
df::Host::frames. So the influxdb interface's Host trait is
implemented on the existing df::Host type (impl in influxdb.rs),
and lang/src/exec.rs just links the world against the same state field:

blr_stdlib::influxdb::Influxdb::add_to_linker::<_, HasSelf<_>>(linker, |state| &mut state.df)
    .map_err(|err| anyhow::anyhow!("failed to link std::influxdb: {err}"))?;

query_private(endpoint, token, database, table, row) pipeline
(current host style: unwrap()/descriptive panic!):

1. Build the SQL from the type-spec: Row(fields) →
   SELECT <labels, sorted> FROM <table> (labels unquoted; table quoted
   only when not [a-z][a-z0-9_]*); RowOpen → SELECT * FROM <table>.
2. Connect: tonic::transport::Endpoint::from_shared("http://{endpoint}")
   → connect() → arrow_flight::client::FlightClient::new(channel)
   (synchronous constructor in 57.3.1; if endpoint has no scheme,
   prepend http://). If token non-empty,
   client.add_header("authorization", "Bearer {token}").
3. Query: JSON ticket {"database", "sql_query": sql, "query_type": "sql"} → client.do_get(Ticket::new(bytes)).await. Map tonic Status
   into the panic message (auth errors, "No field named …", "table … not
   found" are already descriptive).
4. Collect: drain FlightRecordBatchStream (a futures Stream of
   Result<RecordBatch>) into Vec<RecordBatch>; concat_batches the
   non-empty case (empty case handled in step 6).
5. Normalize (before MemTable):
   • Timestamp(_, _) → cast to Int64 epoch nanoseconds (blr has no
     time scalar; lossless, arithmetic/sort/collect all work). If the
     declared type for that column is string, cast to RFC 3339 Utf8
     instead (arrow cast Timestamp(Nanosecond) → Utf8 emits ISO-8601).
   • Dictionary(_, Utf8) → cast to Utf8 (defensive; the 3.x spike saw
     plain Utf8 tags, but 2.x-interop data paths may still use dicts).
   • Null check: any column with null_count() > 0 →
     panic!("influxdb::query: column {name} contains nulls") (matches
     collect's current behavior; null/option support is a language
     follow-up).
6. Empty result (zero batches): synthesize the schema from the
   type-spec — one field per declared label with the declared arrow type
   (Bool→Boolean, Int→Int64, Float→Float64, String→Utf8,
   Any/unknown→Null, mirroring fromcsv's empty-file Null columns) — and
   build a 0-row RecordBatch.
7. Frame: MemTable::try_new(schema, vec![batch]) →
   SessionContext::read_table(Arc::new(table)) (same construction as the
   df unit-test helper).
8. Validate: refactored shared validate_schema(source, schema, row)
   (extracted from validate_csv_schema; source =
   `influxdb::query` on `{endpoint}`) checks each declared label
   exists and is type-compatible — now mostly a defense-in-depth and
   diagnostics path, since the server already failed the query on unknown
   columns. Rename returned columns to the exact declared labels if the
   server folded identifier case (compare case-insensitively).
9. self.frames.push(df).

4.4 Crate wiring

• stdlib/Cargo.toml:
  • arrow-flight = "57" — resolves to 57.3.1 in the spike; the workspace
    must end up with one arrow copy (57.x line, compatible with the
    DataFusion 51 requirement), so RecordBatch/Schema are shared with
    DataFusion's. Consider a lockfile-assertion test (cf. the stdlib
    embedding drift test in lang/src/stdlib.rs).
  • tonic = "0.14" (matching arrow-flight 57's tonic) for Endpoint/
    Channel.
  • No prost, no vendored protos (JSON ticket).
• stdlib/src/lib.rs: pub mod influxdb;
• lang/src/exec.rs: link the world (§4.3). No new State field.

5. Phased plan

Phase 0 — Spike: DONE (results in §2; code pattern in §4.3).

Phase 1 — Boundary. stdlib/wit/influxdb.wit, stdlib/src/influxdb.blr,
pub mod influxdb;, and a stub query_private (returns an empty
synthesized frame from the type-spec, no network) + Influxdb::add_to_linker.
Proves end-to-end: import std::influxdb, type-spec lowering with the new
database/table strings, linearity of the result, and that stub frames
flow through df::filter/collect.

Phase 2 — Host implementation. SQL builder from the type-spec,
tonic/FlightClient, JSON ticket, batch concat, normalization (timestamp,
dictionary, null panic), empty-result synthesis, shared
validate_schema refactor, exec.rs link.

Phase 3 — Tests.

• Unit (stdlib, offline): SQL-builder cases (closed row sorted labels, open
  row → SELECT *, table quoting); batch→frame normalization on canned
  RecordBatches (timestamp→int, dict→utf8, multi-batch concat, null
  panic); empty-result synthesis; validate_schema reuse cases.
• Integration (opt-in, network): gated on INFLUXDB_ENDPOINT/
  INFLUXDB_TOKEN env vars, #[ignore]d when unset; Justfile task
  (just influxdb-test) starting
  quay.io/influxdb/influxdb3-core with
  serve --node-id spike --object-store memory, creating a token
  (influxdb3 create token --admin --host http://localhost:8181), writing
  test data via POST /api/v3/write_lp?db=… (line protocol), and running a
  query → filter → collect program.

Phase 4 — Docs & follow-ups. Knowledge-vault note (this doc), mdbook
page if stdlib docs get one. Follow-ups: TLS endpoint, Config record
(timeouts), InfluxDB 2.x backend (different protocol: Flux over gRPC on
9090 — the query_private shape is unchanged, only request/decode
differ), null/option support, WHERE pushdown, timestamp-first-class type.

6. Open questions (post-spike)