# ADR-007: Native Connector Architecture, UDF Exposure, and Version Matrix

## Status and scope

- **A. Native Spark connector — DONE.** Catalog/DDL/read/write/row-level ops are
  native; MERGE matrix verified on Spark 3.5.9 + 4.2.0 with codegen ON. Row-level
  ops shipped as `SupportsDelta` with **primary-key row identity** (engine
  position deletes are a placeholder, so `deleteRows`+`appendBatch` replaced the
  `mergeRows` sketch). `iceberg-spark-runtime` is gone from Spark 4.x builds and
  remains `provided` in the 3.5 profile only (native DSv2 `CALL` landed in 4.0).
- **B1. Scalar catalog functions — DONE.** `FunctionCatalog` on
  `BenoStreamCatalog` exposes `benostream.system.{cosine_distance, l2_distance,
  dot_product, vector_distance, sparse_dot_product, hybrid_score,
  reciprocal_rank_fusion}` — verified on 3.5.9 and 4.2.0, no registration, works
  from PySpark. `ScalarFunction` implementations must override
  `produceResult(InternalRow)` (required by Spark 4.x, accepted by 3.4+).
- **B2 — all four items closed (2026-11-10, v0.15.1).**
  (1) pass-through `option("query", sql)` reads through the engine DataFusion
  session (`BenoStreamQueryTable`).
  (2) `BenoStreamScanBuilder` now implements `SupportsPushDownTopN`, so
  `ORDER BY <distance> LIMIT k` written in Spark SQL is rendered back into the
  scan SQL and reaches the engine's KNN rule; a pushed TopN forces the single
  SQL scan because a per-file fan-out cannot produce a global top-k.
  (3) all 21 graph aggregates are reachable through the pass-through reader.
  (4) `getTableFunctions()` registers `vector_search`, `keyword_search`,
  `hybrid_search`, `graph_neighbors`, `graph_shortest_path`, `graph_subgraph`,
  `list_indexes`, `regional_drift`, `graph_rag_search`.
- **Two scope corrections, recorded so this does not reopen.** (a) A Spark
  `bsdb_query(sql)` table function is **not implementable**: Spark has no public
  DSv2 `TableFunction` interface (its `TableFunctionRegistry`/`SQLTableFunction`
  are catalyst-internal, for `CREATE FUNCTION … RETURNS TABLE` with a declared
  output), and `BoundFunction.resultType()` is fixed at bind time, so a function
  could not derive a schema from the SQL string. `option("query", …)` is the
  supported mechanism and already covers the whole DataFusion surface.
  (b) Trino cannot host a generic pass-through either (fixed return schema), so
  of the 21 graph aggregates it exposes the four that have table-function
  shapes (neighbors, shortest path, subgraph, regional drift). Exposing the
  remaining 17 requires one fixed-schema TVF per algorithm, a deliberate
  follow-up rather than a gap in this plan.
- **B2. Original scope from this milestone**: (1) pass-through `option("query", sql)`
  reader (full DataFusion surface incl. pgvector operators + KNN optimizer);
  (2) index-accelerated KNN for Spark 4.x via `SupportsTopNPushDown` rendering
  `ORDER BY <distance fn> LIMIT k` back into engine SQL; (3) **all 21 graph
  UDAFs** (`src/core/sql/graph_udf/mod.rs::all_graph_aggregates`: pagerank,
  personalized_pagerank, connected_components, scc, louvain, leiden, modularity,
  betweenness/closeness/degree centrality, jaccard, triangle_count,
  shortest_path, all_shortest_paths, label_propagation, subgraph, connecting_paths,
  graph_neighbors, drift_search, regional_drift, preferential_attachment) +
  keyword/hybrid search — these are set-returning aggregates, so they surface
  through the pass-through reader (Spark has no public set-returning catalog
  TVF API in 3.5 or 4.x); (4) Trino `ConnectorTableFunction` set.
- **Dialect decision**: pgvector operators (`<->`) are engine/pass-through only.
  Spark SQL exposes snake_case function names matching Trino/DataFusion.

## Problem statement

Three interlocking requirements emerged while testing the Spark connector:

1. **Remove the Iceberg Java dependency.** The Spark connector currently wraps
   `IcebergSource`/`SparkTable`/`SparkCatalog`. This (a) couples the connector to
   a per-Spark-minor `iceberg-spark-runtime` jar, and (b) is a dead end: the
   row-level writer (`BenoStreamPositionDeltaWriter`) has **stubbed
   `insert()`/`update()`**, so MERGE INSERT/UPDATE silently drops rows. Reads
   delegate to Iceberg and use JNI only as a discarded hint.
2. **Spark version matrix.** Environment has Spark 3.5.9 (Scala 2.12) and 4.2.0
   (Scala 2.13). Support full 3.5 and 4 series.
3. **Expose the engine's UDFs** in both Trino and Spark — at minimum the vector
   distance UDFs and the graph UDAFs (and ideally the full DataFusion surface).

## Key facts

- The native engine already reads/writes Iceberg-format tables itself (Rust),
  with no Java. The **Trino connector proves the native model**: it reads via
  JNI sessions + Arrow C Data, writes via `appendBatch`/`deleteRows`/`mergeRows`,
  and runs **arbitrary DataFusion SQL** via `Table::sql()` (JNI `openQuery`).
- The engine's DataFusion session registers: `all_vector_udfs()` (dist_* +
  `<->`/`<=>` operators + KNN optimizer rule), `all_json_udfs()`, BM25 pushdown,
  and `all_graph_aggregates()` (~18 graph UDAFs).
- Spark JNI natives already exist (`listDataFiles`, `PartitionReader
  openSession/readBatch`); Spark-named mirrors for the Trino surface were just
  added in `src/core/ffi.rs` (getTableSchema/appendBatch/mergeRows/deleteRows/
  createTable/openQuery/readQueryBatch/closeQuery/list*/dropTable/closeSession).
- **Trino has no generic pass-through query mechanism** and connectors cannot
  register arbitrary scalar UDFs; it *does* support connector **table
  functions** (`ConnectorTableFunction`, SPI 482+).
- **Spark 3.4+/4.x support DSv2 `FunctionCatalog` scalar functions**
  (`ScalarFunction`/`UnboundFunction`); set-returning catalog TVFs are **not**
  public API in either line (4.x `TableFunction` is row-level LATERAL only).
  Relation options (`option("query", ...)`) work everywhere.

## Recommended architecture

### A. Native Spark connector (Iceberg-free)

Rewrite the Spark connector to mirror the Trino connector's native model:

- `BenoStreamTable`: holds `tableUri` + `StructType` (from `getTableSchema`);
  implements `SupportsRead`, `SupportsWrite`, `SupportsRowLevelOperations`.
- **Read**: `BenoStreamScanBuilder` translates required columns + Spark filters →
  a DataFusion SQL string; `BenoStreamScan` → one `InputPartition`;
  `BenoStreamPartitionReader` calls `openQuery`/`readQueryBatch` and decodes
  Arrow→catalyst (new `BenoStreamArrowUtils`, already scaffolded).
- **Write**: `BenoStreamWriteBuilder` → `BatchWrite` → `DataWriter` buffers rows,
  converts to Arrow, calls `appendBatch` (commit already done engine-side per batch).
- **Row-level ops**: DELETE → `deleteRows(filter)`; UPDATE/MERGE → `mergeRows`
  (engine upsert-by-PK). Replaces the stubbed DeltaWrite.
- `BenoStreamCatalog` implements `TableCatalog`/`SupportsNamespaces` via
  `listSchemas`/`listTables`/`getTableSchema`/`createTable`/`dropTable` (no
  `SparkCatalog`). `DefaultSource` returns the native table.
- Drop `iceberg-spark-runtime` from the pom; remove all `org.apache.iceberg.*`
  imports from the connector.

### B. UDF exposure

Two complementary mechanisms:

1. **Pass-through native SQL (full DataFusion surface) — Spark first.**
   `spark.read.format("benostream").option("query", "SELECT id, cosine_distance(...)
   FROM t ORDER BY embedding <-> [..] LIMIT 10").load()` runs engine SQL and
   returns rows. This gives *all* UDFs + operators + KNN/graph optimizer rules
   for free.
2. **Typed table functions (both engines) — the named UDFs.** Define fixed-schema
   table functions so users get IntelliSense-style calls without raw SQL:
   - `vector_search(table, column, query_vector, k, metric)` → `(row_id, score)`
   - `graph_neighbors(table, node, hops, graph_column)` → `(node, hop, ...)`
   - `drift_search` / `regional_drift`, `keyword_search` (BM25), `hybrid_search`
   - Trino: `ConnectorTableFunction` registered on the connector.
     Spark 4: DSv2 `TableFunction`; Spark 3.5: DataFrame API + registered scalar
     UDFs (`BenoStreamFunctions`) as the fallback.

Trino cannot do generic pass-through, so its UDF story is the typed table
functions (+ scalar distance operators where feasible). Spark gets both.

### C. Version matrix (answer to the question)

Once Iceberg is gone, the only real axis is **Scala**:
- **Spark 3.5.x** (incl. 3.5.9): one Scala 2.12 artifact (`spark-3.5` profile).
  Patch releases are binary-compatible; no per-patch builds.
- **Spark 4.0 / 4.1 / 4.2**: one Scala 2.13 artifact covers all of them (DSv2
  APIs are stable across 4.x; we no longer touch version-specific Iceberg).
  Add the `spark-4.2` profile (done) for compile/CI targeting.
- **Trino**: two SPI builds, band `d` (482-483) and band `e` (484+); see
  `docs/integrations/TRINO_VERSION_BANDS.md`.
Net: 2 Spark artifacts (2.12, 2.13) + 1 Trino, not per-minor.

## Sequenced milestones

1. [Native read] Scan builder + partition reader + Arrow→catalyst; verify reads
   work with **no Iceberg on the classpath**.
2. [Native write] DataWriter → Arrow → appendBatch; verify INSERT/CTAS/overwrite.
3. [Native row-level] DELETE→deleteRows; UPDATE/MERGE→mergeRows; verify with
   value assertions (the thing the Iceberg wrapper silently broke).
4. [Native catalog/DDL] BenoStreamCatalog + DefaultSource; drop Iceberg dep.
5. [Spark UDFs] Pass-through `query` option + Spark 4 `bsdb_query` table function
   + typed table functions (vector_search, graph_neighbors, drift, keyword, hybrid).
6. [Trino UDFs] `ConnectorTableFunction` set (vector_search, graph_neighbors, …).
7. [Matrix/CI] Build 2.12 + 2.13 artifacts; add spark-4.2 to build matrix; docs.
8. [Regression] Re-run full surface suites for both connectors.

## Risks / open questions

- Native write must preserve Iceberg-format correctness (snapshots/manifests) —
  the engine already does this (Trino writes prove it), but Spark's partitioning/
  bucketing semantics need parity work.
- Trino table functions have fixed return schemas; vector/graph outputs must be
  modeled accordingly (row_id + score / node columns).
- `libbenostreamdb.so` must be on `java.library.path` on every Spark executor /
  Trino worker (already true for Trino).
- Scope is large; ordering below is a recommendation, not a commitment.