spark-adbc

A Spark DataSource V2 connector for Apache Arrow ADBC (Arrow Database Connectivity). It enables Spark to read from ADBC-compatible databases using Arrow's columnar format, avoiding the row-by-row serialization overhead of traditional JDBC.

Features

• Columnar data transfer — reads data as Arrow columnar batches, enabling zero-copy integration with Spark's columnar execution engine
• Read support — read via dbtable or custom query
• Column pruning — only fetches the columns Spark actually needs
• Filter pushdown — pushes filter expressions down to the database (equality, comparison, null checks, IN, string LIKE patterns, AND/OR/NOT)
• Aggregate pushdown — pushes COUNT, SUM, AVG, MIN, MAX (including DISTINCT variants) and GROUP BY down to the database
• Limit pushdown — pushes LIMIT down to the database
• Top-N pushdown — pushes ORDER BY ... LIMIT N down to the database
• SQL dialect support — generates dialect-specific SQL for PostgreSQL, SQLite, and MSSQL (auto-detected from jni.driver or set explicitly via dialect option)
• Partitioned reads — split reads across multiple Spark partitions by a numeric column for parallel data ingestion
• Driver-driven partitioning — lets drivers that implement ADBC executePartitioned (e.g. Flight SQL backends) split query results across Spark partitions
• Works with any ADBC driver — PostgreSQL, SQLite, MSSQL, DuckDB, Flight SQL, Snowflake, etc.

Usage

// Read from a table
val df = spark.read
  .format("com.tokoko.spark.adbc")
  .option("driver", "org.apache.arrow.adbc.driver.jni.JniDriverFactory")
  .option("jni.driver", "postgresql")
  .option("uri", "postgresql://user:pass@localhost:5432/mydb")
  .option("dbtable", "my_table")
  .load()

// Read from a query
val df = spark.read
  .format("com.tokoko.spark.adbc")
  .option("driver", "org.apache.arrow.adbc.driver.jni.JniDriverFactory")
  .option("jni.driver", "postgresql")
  .option("uri", "postgresql://user:pass@localhost:5432/mydb")
  .option("query", "SELECT id, name FROM my_table WHERE active = true")
  .load()

// Partitioned read (parallel ingestion across 8 tasks)
val df = spark.read
  .format("com.tokoko.spark.adbc")
  .option("driver", "org.apache.arrow.adbc.driver.jni.JniDriverFactory")
  .option("jni.driver", "postgresql")
  .option("uri", "postgresql://user:pass@localhost:5432/mydb")
  .option("dbtable", "my_table")
  .option("partitionColumn", "id")
  .option("lowerBound", "0")
  .option("upperBound", "1000000")
  .option("numPartitions", "8")
  .load()

lowerBound and upperBound are used to compute the stride for range splitting — they do not filter data. Rows outside the specified range are still read (they land in the first or last partition).

Driver-driven partitioning

The driverPartitioning option controls whether each query is run with ADBC executePartitioned, so the driver splits the result into partition descriptors, and each descriptor becomes its own Spark partition:

• auto (default without range options): use executePartitioned when the driver implements it, otherwise run a plain query
• required: fail if the driver doesn't implement executePartitioned
• none (default with range options): always run plain queries

It works independently of range partitioning. If you set both, each of the numPartitions range queries is split again by the driver. In that case all range queries run on the database while Spark plans the job (not as tasks start), and the server keeps every result until executors read it.

With driver partitioning, aggregates are fully pushed down. Limit and Top-N are pushed down too, and Spark re-applies them because the order across partitions isn't guaranteed. Partition descriptors are created once per planned query and re-read on task retries, so the driver's descriptors must be readable more than once.

The JNI driver (JniDriverFactory) doesn't implement executePartitioned as of ADBC 0.23, so with it auto always falls back to plain queries.

Benchmarks

Local benchmarks comparing spark-adbc against Spark's built-in JDBC connector, reading 20 million rows from PostgreSQL 16 (single-node, local[*] mode).

Benchmark	ADBC	JDBC	Speedup
Full table scan	73.9s	135.8s	1.84x
Full table scan — columnar*	38.3s	142.6s	3.72x
Column projection (2 of 6 cols)	34.2s	64.3s	1.88x
Filtered read (category = 'cat_5')	5.6s	6.2s	1.11x
Aggregation (GROUP BY with AVG, SUM, COUNT)	4.1s	49.2s	11.91x
Join + filter + aggregate (with Comet)	15.8s	43.7s	2.76x

*The columnar benchmark measures Arrow batch ingestion without Spark's columnar-to-row conversion. This represents a theoretical upper bound — Spark currently converts columnar batches back to rows for most operations, so real-world queries go through the standard (non-columnar) read path.

Apache DataFusion Comet

Apache DataFusion Comet can operate directly on the Arrow columnar batches returned by spark-adbc, avoiding Spark's columnar-to-row conversion for downstream operations like joins, filters, and aggregations. To enable this, configure Comet's experimental sparkToColumnar feature to accept BatchScan as a columnar source:

val spark = SparkSession.builder()
  .config("spark.sql.extensions", "org.apache.comet.CometSparkSessionExtensions")
  .config("spark.comet.enabled", "true")
  .config("spark.comet.exec.enabled", "true")
  .config("spark.comet.exec.all.enabled", "true")
  .config("spark.comet.sparkToColumnar.enabled", "true")
  .config("spark.comet.sparkToColumnar.supportedOperatorList",
    "Range,InMemoryTableScan,RDDScan,BatchScan")
  .getOrCreate()

This replaces Spark's ColumnarToRow with Comet's CometSparkColumnarToColumnar bridge, keeping the entire pipeline columnar through DataFusion's native execution engine.

Note: CometSparkColumnarToColumnar is not zero-copy — it reads values element-by-element from Spark's ColumnVector API and writes them into new Arrow vectors for Comet's native format, even though the source data is already Arrow-backed. This adds conversion overhead at the scan boundary, but all subsequent operations (joins, filters, aggregations) run natively in DataFusion without further conversion.