[zipdoki]

Summary

A multi-hop query returns one frame per step and nothing more. Joining two hops, ranking their rows or projecting a subset all happens on the caller. This lets a query name SQL that runs over the frames it just fetched, in process, before the response goes out.

{
  "fetch": [
    { "type": "SCAN", "name": "follows", "database": "graph", "table": "follows", "index": "created_at", "source": { "type": "VALUE", "value": "user1" } },
    { "type": "TOPK", "name": "ranked", "database": "graph", "table": "orders", "topk": "top_product_groups_1y", "entity": { "type": "REF", "ref": "follows", "field": "tgt" }, "limit": 100 }
  ],
  "transform": [
    {
      "type": "SQL",
      "name": "top_items",
      "sql": "SELECT ranked.tgt, SUM(ranked.metric) AS metric FROM ranked GROUP BY ranked.tgt ORDER BY metric DESC LIMIT ?",
      "arguments": ["10"]
    }
  ]
}

Each step's frame is a table named after the step, so SQL addresses ranked.metric directly. Transforms run in order and each one's output joins the context under its name, so a later transform can read an earlier one. They see the whole context including include = false steps — a step can exist only to be subtracted from a later one —
and what the caller gets back is filtered afterwards.

Calcite is the engine, over a JDBC connection whose root schema holds one slot per frame. Most of the design here is about not paying its planning cost per request.

SqlTransform caches a PreparedTransform per (sql, input columns). Calcite's driver keeps no plan between statements, so without this every request re-parses, re-plans and re-compiles the same SQL. Each prepared transform owns a pool of sessions rather than one shared session, because a PreparedStatement takes a single thread.
Sessions open at prepare time, not on the request path — opening one touches disk for the driver and the generated class. prepare pays that cost and prepared answers only if it was already paid, so a caller can keep it off a thread that must not block.

The pool is capped per transform, so a burst of concurrent requests on the same SQL queues instead of scaling out. I'd rather queue than open connections on the request path. If that queue shows up in latency, the cap is the knob.

Two smaller decisions. The connection sets lex=JAVA so identifiers stay as written and hop1.paidAt is not folded to PAIDAT, and fun=all so the dialect operator tables are available — the standard one has no IFNULL. And calcite.bindable.cache.maxSize=1000 is set in three places: the Gradle test JVM, the Jib image, and as a
default in the pool. That cache is process-wide and its default is small enough that a handful of distinct transforms evict each other.

Nothing here bounds what a transform can ask for — no row cap on the result, no timeout, no check on the SQL beyond what Calcite refuses to plan. That needs to exist before this is reachable by an untrusted caller, and I'll do it in a follow-up.

The two probes are @Disabled, and that is deliberate. TransformSchedulerProbe installs BlockHound for the whole JVM, and CalciteLatencyProbe measures Calcite's planning cost rather than anything in this repo. They document how the numbers above were arrived at; run them by hand from the IDE when the numbers matter.

Stacked on #498 — that needs to land first.

Part of #490.

Test plan

• ./gradlew :engine:test --tests '*SqlTransformTest*' — plan cache, session pool, prepared-vs-prepare, argument binding, closing
• ./gradlew :engine:test --tests '*DataFrameSqlTransformTest*' — SQL over frames: projection, join across steps, aggregation, ordering
• ./gradlew :engine:test --tests '*DataFrameShowTest*' — the frame formatter the tests read failures through
• ./gradlew spotlessCheck build — formatting and full build

AI Assistance

• This PR was written largely with AI assistance.
  • Tool / model: Claude Code (Opus 5)

[zipdoki]

Optimistic Merge