# Apache Spark Connector The BenoStreamDB Spark connector is a **fully native** Apache Spark DataSource V2 connector: reads, writes, DDL, catalog metadata, and row-level operations (DELETE / UPDATE / MERGE) all execute inside the BenoStreamDB Rust engine via JNI and the Arrow C Data Interface. It provides universal vector search, secondary index pushdown, stored procedures, and streaming capabilities. There is **no Iceberg runtime dependency** for Spark 4.x builds. (Spark 3.5 predates the native DSv2 `ProcedureCatalog`/`CALL` API, so the optional `spark-3.5` build links `iceberg-spark-runtime` in `provided` scope solely to expose stored procedures; tables, reads, writes, and row-level ops on 3.5 are native too.) --- ## Requirements * **Apache Spark**: 3.5.x, 4.0.x, 4.1.x, or 4.2.x * **Java**: 17 (required for Spark 3.5; supported on 4.x) * **Scala**: 2.12 (Spark 3.5) / 2.13 (Spark 4.x) — one artifact per Scala binary covers every Spark minor in that line * **BenoStreamDB Native Library**: `libbenostreamdb.so` (bundled in the JAR resources, or on `java.library.path`) --- > **Running the connector's own tests:** `mvn test` loads the JNI library, so > build it first (`cargo build --features java` — the surefire `argLine` points > `java.library.path` at `../target/debug`) and use **JDK 17**. JDK 23+ removed > `Subject.getSubject`, which Spark still calls, so the Spark suites fail on a > newer JDK for reasons unrelated to this connector. ## Building the Connector The connector source resides in `spark-benostreamdb`. Use Maven to build the profile corresponding to your Spark version: ```bash cd spark-benostreamdb # Default (Spark 4.0) mvn clean package -Pspark-4.0 # Spark 3.5 (Scala 2.12) mvn clean package -Pspark-3.5 # Spark 4.1 mvn clean package -Pspark-4.1 # Spark 4.2 mvn clean package -Pspark-4.2 ``` --- ## Configuration & Catalog Setup Configure SparkSession with the native catalog. On Spark 4.x use `BenoStreamProcedureCatalog` (the native table catalog **plus** `system.*` stored procedures); on Spark 3.5 use `BenoStreamCatalog` for tables/DDL: ```scala val spark = SparkSession.builder() .appName("BenoStreamSparkApp") // Spark 4.x: tables + namespaces + CALL in one catalog .config("spark.sql.catalog.benostream", "com.benostreamdb.spark.BenoStreamProcedureCatalog") .config("spark.sql.catalog.benostream.warehouse", "s3://my-lakehouse/warehouse") .getOrCreate() // Register BenoStreamDB vector search and similarity UDFs import com.benostreamdb.spark.functions.BenoStreamFunctions BenoStreamFunctions.register(spark) ``` Writes are chunked by `spark.benostream.write.batchSize` (default `10000`; also settable per write option or table property). See [Row-Level Operations](#5-row-level-operations-delete--update--merge) for how it bounds MERGE memory. Reads fan out by file when `spark.benostream.split.maxSize` is greater than zero (default `128 MB`). The scan then plans one partition per Parquet file instead of one table-wide query, provided Spark pushed no filter into the scan. Setting it to `0` forces the single-partition SQL scan. Splits are file-granular, so the configured value acts as the on/off threshold rather than a byte-range size. Tables with merge-on-read delete files always use the single SQL scan: a raw per-file read would resurrect deleted rows. --- ## Core Capabilities ### 1. Vector Search & Similarity Functions BenoStreamDB provides native SQL functions and DataFrame extensions for high-dimensional vector search. #### Catalog Functions (no registration required) The functions below are exposed through the Spark DSv2 `FunctionCatalog` and resolve with their fully-qualified names from Spark SQL, DataFrames, and PySpark on both Spark 3.5 and 4.x — no `register()` call needed: ```sql SHOW FUNCTIONS IN benostream.system; SELECT id, title, benostream.system.cosine_distance(embedding, array(0.12, 0.45, -0.23)) AS dist FROM benostream.default.documents ORDER BY dist ASC LIMIT 10; ``` Registering the session-level aliases (`BenoStreamFunctions.register(spark)`, shown below) lets you use the unqualified names. Note that pgvector operators (`<->`) are intentionally **not** part of Spark SQL — Spark's grammar rejects them and they would be inconsistent with other catalogs in the same session; use these function names instead (identical to the Trino/DataFusion spellings). #### Spark SQL Functions ```sql -- Cosine Distance (0.0 = identical, 1.0 = orthogonal) SELECT id, title, cosine_distance(embedding, array(0.12, 0.45, -0.23)) as dist FROM spark_catalog.default.documents ORDER BY dist ASC LIMIT 10; -- Euclidean (L2) Distance SELECT id, l2_distance(embedding, array(0.12, 0.45, -0.23)) as dist FROM spark_catalog.default.documents; -- Dot Product (Higher is more similar) SELECT id, dot_product(embedding, array(0.12, 0.45, -0.23)) as score FROM spark_catalog.default.documents; -- Generic Vector Distance SELECT id, vector_distance(embedding, array(0.12, 0.45, -0.23), 'cosine') as dist FROM spark_catalog.default.documents; ``` #### Fluent DataFrame API ```scala import com.benostreamdb.spark.implicits._ val query = Array(0.12, 0.45, -0.23) // Retrieve Top-K nearest neighbors val top10 = df.vectorSearch( vectorCol = "embedding", query = query, k = 10, metric = "cosine" ) // Enrich DataFrame with distance column val withDist = df.withVectorDistance( outputCol = "_distance", vectorCol = "embedding", query = query, metric = "cosine" ) ``` --- ### 2. Sparse Vectors & Hybrid Search (SPLADE / BM25) BenoStreamDB supports learned sparse vectors (such as SPLADE or BM25 term weights) represented as Arrow `Struct, values: List>`: ```sql -- Compute sparse dot product across token weights SELECT id, sparse_dot_product(query_indices, query_values, doc_indices, doc_values) as score FROM spark_catalog.default.search_corpus; -- Score Blending: Combine dense embedding similarity with sparse lexical score SELECT id, hybrid_score(dense_score, sparse_score, 0.7) as final_score FROM ranked_candidates; -- Reciprocal Rank Fusion (RRF) across multi-retrieval candidate lists SELECT id, reciprocal_rank_fusion(array(dense_rank, sparse_rank), 60) as rrf_score FROM candidates; ``` --- ### 2b. Graph Traversal (table functions) Graph walks are available directly in `FROM` clauses via the engine's DataFusion table functions, reached through the pass-through reader (`option("query", ...)`). Each follows the same in-memory / out-of-core `GraphMode` pattern as the graph UDAFs (`auto` | `in_memory` | `out_of_core` | `cached`). The pass-through reader registers the table as `t`: ```scala spark.read.format("benostream") .option("path", "s3://my-lakehouse/tables/edges") .option("query", "SELECT * FROM graph_neighbors('t', '101', 2, 'auto')") .load() ``` ```sql -- Nodes within 2 hops of node 101 (node, hop, seed) SELECT * FROM graph_neighbors('edges', '101', 2, 'auto'); -- Shortest path between two nodes (node, hop) SELECT * FROM graph_shortest_path('edges', 101, 205, 'auto'); -- Every shortest path (path) SELECT * FROM graph_all_shortest_paths('edges', 101, 205, 'auto'); -- Induced subgraph edges within 2 hops (source, target) SELECT * FROM graph_subgraph('edges', '101,102', 2, 'auto'); -- Union of pairwise shortest paths between seeds (source, target) SELECT * FROM graph_connecting_paths('edges', '101,205,309', 'auto'); ``` Endpoint columns are auto-detected (`source`/`src`/`src_id`/`from`/`u` and `target`/`dst`/`dst_id`/`to`/`v`); pass trailing `source`, `target` string arguments to name them explicitly. Declaring an edge table (`SET TBLPROPERTIES ('table_type'='edge', 'src_col'=…, 'dst_col'=…)`) automatically configures the forward (source) and reverse (target) CSR graph indexes. --- ### 2c. Live Subscriptions (change feed) `subscribe_events` is a DataFusion table function, so it is reachable from Spark through the pass-through reader. It drains the next committed-change events for a table (a bounded live tail) and returns one row per event: ```sql -- event_type ('batch' | 'commit'), rows SELECT * FROM subscribe_events('edges', 'weight > 0.5', 100, 1000); ``` Arguments: `table`, optional `filter` (SQL predicate), optional `max_events` (default 100), optional `timeout_ms` (default 1000). The change feed is **in-process**: a Spark job only observes commits made by writers in the same JVM. For cross-process streaming use the Flight SQL subscription ticket. --- ### 3. Secondary & Bitmap Index Filter Pushdown `BenoStreamScanBuilder` implements `SupportsPushDownFilters` and `SupportsPushDownRequiredColumns`. Supported predicates (`=`, `<`, `<=`, `>`, `>=`, `IN (...)`, `IS [NOT] NULL`, `LIKE 'prefix%'`, `AND`/`OR`/`NOT`) are translated to SQL and pushed into the engine's DataFusion planner through `openQuery`, which applies the scalar/Bitmap, Bloom, and lexical indexes during planning — pruning data files and row splits before any Parquet is decoded by Spark executors: ```scala // Prunes unindexed Parquet splits using BenoStreamDB bitmap indexes val activeUsers = spark.read .format("benostream") .load("s3://my-lakehouse/tables/users") .filter("user_id IN (1001, 1002, 1003)") ``` --- ### 4. Index Lifecycle & Catalog Procedures On Spark 4.x, manage BenoStreamDB indexes and primary keys directly through native Spark stored procedures under `benostream.system` (the catalog above implements the DSv2 `ProcedureCatalog` API). Every procedure routes straight into the engine over JNI, so it applies to native BenoStreamDB tables as well as externally mapped ones. (`list_indexes`, `optimize`, and `compact` are aliases.) ```sql -- Add an index to a column: algorithm is one of -- hnsw | hnsw_pq | hnsw_tq4 | hnsw_tq8 | hnsw_sq8 | bm25 | bloom | bitmap | -- composite_bitmap | csr_graph | json_path -- (defaults to full-precision hnsw; hnsw_tq8 and hnsw_tq4 are opt-in) CALL benostream.system.add_index('spark_catalog.default.users', 'embedding', 'hnsw'); -- Build offline index files for all segments CALL benostream.system.build_index('spark_catalog.default.users'); -- Rebuild a specific index column (drops and re-creates it) CALL benostream.system.rebuild_index('spark_catalog.default.users', 'embedding'); -- Rebuild table-level Voronoi cluster index into rolling <= 512 MiB Puffin shards CALL benostream.system.rebuild_cluster_index('spark_catalog.default.users', 'embedding', 1024, 16, 200, 512); -- Pre-warm Voronoi centroid routes and active shards into RAM cache CALL benostream.system.warmup('spark_catalog.default.users', 500, 8.0); -- List all configured and built indexes: one row per index with -- (column_name, index_category, algorithm, index_id, format_version) CALL benostream.system.show_indexes('spark_catalog.default.users'); -- Drop an index CALL benostream.system.drop_index('spark_catalog.default.users', 'embedding', 'hnsw'); -- Compact segments and consolidate index metadata CALL benostream.system.compact('spark_catalog.default.users'); -- Set / drop the primary key used for Merge-on-Read row identity CALL benostream.system.set_primary_key('spark_catalog.default.users', 'id'); CALL benostream.system.drop_primary_key('spark_catalog.default.users', 'id'); -- Regional DRIFT graph search: returns a node_id column (top-k node ids) CALL benostream.system.regional_drift_search( 'spark_catalog.default.edges', 'vector query text', array(101L, 205L), 5, 2, 2, 3, 'auto'); ``` `show_indexes` and `regional_drift_search` return result sets on Spark 4.x as well as 3.5 (Spark 4's `BoundProcedure.call` hands back a `Scan`, which the connector fills from the already-computed JNI result). **Table argument resolution.** Every procedure resolves its `table` argument through `BenoStreamTableResolver`, so namespaces are preserved instead of being dropped: | Argument | Resolved URI | |----------|--------------| | `users` | `/default/users` | | `sales.users` | `/sales/users` | | `benostream.sales.users` | `/sales/users` | | `s3://bucket/path/users` | unchanged (explicit URI) | The warehouse comes from `spark.sql.catalog..warehouse` (the `benostream` catalog is preferred when several are configured), falling back to `spark.benostream.warehouse`. Previously `table.split(".").last` discarded the namespace, so `benostream.default.users` and `users` were indistinguishable. On Spark 3.5 the same procedures are reachable through the Iceberg `ProcedureCatalog` shim. On Spark 4.2+ the catalog also implements `listProcedures`, so `SHOW PROCEDURES IN benostream.system` lists them. --- ### 5. Row-Level Operations: DELETE / UPDATE / MERGE The connector implements `SupportsRowLevelOperations` natively (`BenoStreamRowLevelOperation` extends `SupportsDelta`). Row identity is the table's **primary key** columns (`rowId()` / `requiredMetadataAttributes()`), which are ordinary table columns — so Spark resolves them against the target relation without metadata-column plumbing. Execution maps directly onto the engine's primitives: * matched `DELETE` / `UPDATE` (old image) → predicate deletes via JNI (`deleteRows`) * `INSERT` / updated new rows → Arrow record batches appended via JNI (`appendBatch`) * plain `DELETE FROM ... WHERE ...` (no subquery) → pushed down as a single `deleteRows` predicate through `SupportsDelete` ```sql MERGE INTO benostream.default.users target USING new_updates source ON target.id = source.id WHEN MATCHED AND source.age < 0 THEN DELETE WHEN MATCHED THEN UPDATE SET age = source.age WHEN NOT MATCHED THEN INSERT (id, age) VALUES (source.id, source.age); UPDATE benostream.default.users SET age = 31 WHERE id = 7; DELETE FROM benostream.default.users WHERE age < 0; ``` The full MERGE matrix (UPDATE / DELETE / INSERT clauses, plus standalone UPDATE/DELETE) is verified end-to-end on Spark 3.5.9 and 4.2.0 with whole-stage codegen enabled. Declare the primary key with `CALL benostream.system.set_primary_key('db.tbl', 'id')` (or the `primary_key` table property); when unset, the first column is used. A table with **no** key and no columns cannot service `UPDATE`/`DELETE`; the connector records no predicates for those rows instead of failing the task. **Exactly-once writes (file protocol, Iceberg's shape).** Spark calls `DataWriter.commit()` inside the task and `BatchWrite.commit(messages)` once on the driver. Only the driver call is a valid commit point: a task can succeed and then be retried, and Spark replays the messages of successful attempts only. So the write is split the way Iceberg's driver splits it: 1. **Task** — writes its rows as *data files* (with their indexes, built in the task) and returns the manifest entries as JSON: `BenoStreamFileCommitMessage` → `stageDataFiles` → `(Table::stage_data_files_async)` in `src/core/table/stage.rs`. The task never commits; the files are invisible. 2. **Driver** — commits only that metadata, in one snapshot: `commitStagedEntries` → `Table::commit_staged_entries_async`, which is a single manifest commit plus the snapshot publish. No row data crosses the driver, so the commit is O(files) instead of O(bytes), and a 20-partition `INSERT` is one snapshot. 3. **Retries** — a retried task attempt writes new files at new paths; only the entries the driver accepted are committed, and the rest are orphans for `remove_orphan_files`. A retried `commit` is a no-op: tokens already committed are recorded and skipped. `format_version = 3` tables fall back to the buffered path (`writeMode` returns `buffered`): `_row_id` is assigned from the manifest head, so two tasks staging concurrently would hand out overlapping ids. On those tables the task encodes rows as Arrow IPC (`BenoStreamCommitMessage`), the driver stages them in the engine and commits once — still exactly-once and still one snapshot, but the rows cross the driver. MERGE uses the same driver commit point: every rendered delete predicate is applied as one OR-combined `deleteRows` commit, then the staged inserts, in that order — an UPDATE reuses the key, so inserting first would let the delete reap the new row. Conformance with Iceberg's Spark driver (`iceberg-spark-runtime`), verified against the 1.11.0 jar: | Contract | Iceberg | This connector | | --- | --- | --- | | `BatchWrite.createBatchWriterFactory` | `SparkWrite$BaseBatchWrite` | `BenoStreamBatchWrite` | | `DataWriter` writes into the engine | no — writes Parquet data files | no — writes Parquet data files (`stage_data_files`) | | `WriterCommitMessage` payload | `SparkWrite$TaskCommit` → `DataFile[] files()` | `BenoStreamFileCommitMessage` → manifest entries (JSON) | | `BatchWrite.commit(messages)` | `SparkWrite$BatchAppend`: one metadata snapshot | one metadata snapshot (`commit_staged_entries`) | | `BatchWrite.abort(messages)` | staged files become orphans | staged files become orphans (`remove_orphan_files`) | | Row-level (`SupportsDelta`) | `SparkPositionDeltaWrite$DeltaTaskCommit` → `dataFiles()`, `deleteFiles()`, … | `BenoStreamDeltaCommitMessage` → predicates + encoded rows (buffered) | | Retry safety | unique file names per attempt | unique file names per attempt; commit token is idempotent | Index building moved into the task as a consequence: the driver commits only metadata, so there is no post-commit hook that could attach index files later. The staged file's `index_files` are populated before the entry is returned, so the snapshot is complete the moment it is published. **Bounded task memory.** Each task buffers its own rows and encodes them in `spark.benostream.write.batchSize` (default `10000`) row chunks: * Deletes accumulate rendered key literals, not one growing SQL string. They are applied as `pk IN (...)` predicates of 1,000 keys (single-column key) or `(a = .. AND b = ..) OR ...` chunks of 500 (compound key); rows containing a `NULL` key get their own `pk IS NULL`-style predicate. This keeps a 10k-row `DELETE` clear of the engine's AST recursion limit and of task OOM. * Inserts are encoded in chunks of `write.batchSize` rows. Configure it per write (`writeTo(...).option("write.batchSize", "5000")`), per table (`TBLPROPERTIES ('write.batchSize' = '5000')`), or session-wide via `spark.benostream.write.batchSize`. > Note: the engine does not currently honor Iceberg position-delete files (`commitPositionDeletes` is a placeholder), so deletes are applied as predicate removals at commit time rather than as delete-file overlays. --- ### 6. Dynamic GPU Task Binding For distributed GPU clusters (NVIDIA CUDA, AMD ROCm, Apple Metal, Intel Arc/XPU), `GpuContextResolver` automatically resolves and binds the native thread GPU context per task: 1. Dynamic discovery from Spark `TaskContext.get().resources().get("gpu")`. 2. `CUDA_VISIBLE_DEVICES` environment variable mapping. 3. Fallback to `spark.benostream.gpu.device` or table properties.