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)

---

Building the Connector

The connector source resides in spark-benostreamdb. Use Maven to build the profile corresponding to your Spark version:

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:

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)

---

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:

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

-- 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

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<indices: List<Int64>, values: List<Float32>>:

-- 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:

spark.read.format("benostream")
  .option("path", "s3://my-lakehouse/tables/edges")
  .option("query", "SELECT * FROM graph_neighbors('t', '101', 2, 'auto')")
  .load()

-- 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:

-- 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:

// 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.)

-- Add an index to a column: algorithm is one of
--   hnsw | hnsw_pq | hnsw_tq4 | hnsw_tq8 | 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');

-- List all configured and built indexes (JSON)
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 (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');

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

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.

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.