faucet-source-databricks

Databricks SQL query source for the faucet-stream ecosystem. Runs a SQL statement against a Databricks SQL Warehouse via the Statement Execution API (plain REST — no JDBC/ODBC driver, no Python) and streams the result rows as typed JSON objects.

This is the query-results read path for Databricks — joins, aggregates, filtered extracts. For full-table lakehouse scans and the write path, use the Delta Lake connectors (faucet-source-delta / faucet-sink-delta). To load into a Databricks SQL warehouse (Unity Catalog permissions, COPY INTO, MERGE, exactly-once), use faucet-sink-databricks.

Highlights

• Async statement lifecycle — submit → poll until terminal → stream result chunks (EXTERNAL_LINKS + JSON_ARRAY: each chunk is downloaded from its presigned URL without the workspace token), following the chunk links, so results of any size stream — INLINE results are capped at 25 MiB by Databricks.
• Type-aware decode from the response manifest column schema (every JSON_ARRAY cell is a string; decoded per type_name to typed JSON, with DECIMAL/large LONG preserved losslessly as strings). A result with two columns of the same name (a join's id) fails with an error asking you to alias them, rather than one silently replacing the other.
• Incremental replication — reads only rows not yet delivered, in cursor order, without skipping rows that share a cursor value; a crash replays at most one page (see Incremental replication).
• Bearer auth (PAT / OAuth M2M), inline or via the shared auth: catalog. The auth and Statement Execution API client live in faucet-common-databricks (shared with the sink). This crate keeps its own DatabricksAuth (same wire shape, converted with From) so its public API is unchanged.
• Warm-up tolerant — a submit refused with 429/503 (a serverless warehouse starting) is retried with backoff honouring Retry-After, polls retry transient 5xx, and a 401 makes a shared provider mint a fresh token once.
• Arrow-native fetch — behind the arrow feature, arrow_native: true fetches results as EXTERNAL_LINKS + ARROW_STREAM and decodes each chunk as an Arrow IPC stream, enabling the columnar fast path and skipping the per-cell JSON decode on the row path. See Arrow columnar (Parquet) mode.

Configuration

pipeline:
  source:
    type: databricks
    config:
      workspace_url: https://dbc-xxxx.cloud.databricks.com
      warehouse_id: 0123456789abcdef
      sql: SELECT id, ts FROM events
      auth: { type: pat, config: { token: "${env:DATABRICKS_TOKEN}" } }
      replication: { type: incremental, column: ts, initial_value: "2026-01-01" }

Incremental replication

replication: { type: incremental, column, initial_value } wraps the SQL as SELECT * FROM (<sql>) AS q WHERE q.`<column>` >= :_faucet_bookmark ORDER BY q.`<column>` , so the warehouse filters and orders by the cursor; initial_value is the inclusive lower bound for the first run. To apply the cursor further inside the SQL, write ${bookmark} there with >= (it binds the same named parameter); a > ${bookmark} is refused at load time because it would skip rows that share the bookmark value.

Rows that share a cursor value are never skipped. The bookmark stores the last cursor value and a 128-bit fingerprint of every row already emitted at it — {"value": "...", "boundary": ["<fingerprint>", ...]}. The next read re-reads that value with >= and drops only the rows it already emitted, so a row committed later with the last updated_at is delivered. Above 10,000 rows at one value the set is dropped and the value is re-read in full: duplicates, never loss. Every page carries the bookmark after its last row, so a crash replays at most one page. A bare bookmark stored by an earlier release reads as that value with an empty boundary (its rows are re-delivered once). Rows whose cursor is NULL are never read.

Arrow columnar (Parquet) mode

Behind the crate-local arrow Cargo feature, setting arrow_native: true submits the statement with EXTERNAL_LINKS disposition and ARROW_STREAM format; each result chunk's presigned link is fetched and decoded as an Arrow IPC stream rather than a JSON_ARRAY. This has two effects:

• the opt-in columnar fast path (RFC 0002 / #375) — when the sink is also Arrow-native (the Parquet or Delta Lake sink) and no Value-shaped transform is configured, records move end-to-end as Arrow RecordBatches with no serde_json::Value materialization; and
• a faster row path — even into a JSON sink, decoding Arrow batches skips the per-cell string→JSON decode the default JSON_ARRAY path pays.

arrow_native: true requires the arrow feature and only works with replication: full — config validation rejects arrow_native combined with incremental replication, because the columnar path does not track the per-row cursor bookmark. The default (arrow_native: false) keeps the JSON_ARRAY row path.

# databricks(arrow) → parquet — runs Arrow end-to-end
pipeline:
  source:
    type: databricks
    config:
      workspace_url: https://dbc-xxxx.cloud.databricks.com
      warehouse_id: 0123456789abcdef
      sql: SELECT id, ts, payload FROM events
      auth: { type: pat, config: { token: "${env:DATABRICKS_TOKEN}" } }
      arrow_native: true   # requires the `arrow` feature; replication must be `full`
  sink:
    type: file
    config:
      path: ./out/
      format: parquet

Enable it with cargo add faucet-source-databricks --features arrow (library) or cargo install faucet-cli --features "source-databricks,arrow" (CLI).

Out of scope (v1)

discover() via INFORMATION_SCHEMA and reading from Unity Catalog Volumes. To write to a SQL warehouse, use the faucet-sink-databricks sink.

License: MIT OR Apache-2.0.

Capabilities

Declares columnar (Arrow) output only when the crate is built with the arrow feature and the config sets arrow_native: true; otherwise it declares nothing. faucet list --json prints these declared capabilities.

Timeouts, cancellation and retries

Every request — statement submit and polls, result chunks, presigned external links — runs through the pipeline's CallContext (#880): the run's cancel token stops an in-flight request or poll (an abandoned statement is cancelled on the warehouse), any caller deadline ends it, and each is metered as a round trip (submit, poll, chunk). The HTTP client's connect / idle-read timeouts, statement_timeout_secs and the 429 / 503 retry are unchanged.