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.