Skip to content

Commit 97dbf85

Browse files
timsaucerclaude
andcommitted
feat: ADBC-backed Spark DataSource for DataFusion table providers
Add a Spark DataSourceV2 connector that reads from a DataFusion TableProvider through a standard ADBC driver. Spark talks to the upstream arrow-adbc Java driver manager (adbc-core + adbc-driver-jni), which loads a native DataFusion ADBC cdylib and returns arrow-java ArrowReaders consumed zero-copy as ArrowColumnVectors on the cluster-provided Arrow. Spark connector (spark/): - AdbcDatafusionTableProvider: registers the `adbc-datafusion` format; schema probed once on the driver via AdbcConnection.getTableSchema. - Scan with projection/filter/limit pushed into the scan as a Substrait plan, with a SQL fallback path. - Multi-partition reads via executePartitioned/readPartition; the target_partitions option tunes scan parallelism. - Per-executor AdbcConnectionPool caches the AdbcDatabase per driver + options key, with a per-task connection (level 1) as the safe default; connection sharing (level 2) is gated and off by default. Example driver (examples/adbc-datafusion-driver/): a DataFusion ADBC cdylib exercising the full stack, with a PySpark end-to-end script and partitioning tests. Docs: ADBC-to-Spark partitioning guidance and connector notes. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1 parent ab57cb9 commit 97dbf85

30 files changed

Lines changed: 8985 additions & 0 deletions

‎datafusion-java-connector-items.md‎

Lines changed: 142 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,142 @@
1+
# Java connector work items (`datafusion-java`)
2+
3+
Handoff for the connector-side pieces of the ADBC partition fast-path plan. The
4+
**driver-side** work (`adbc-driver-datafusion`, Rust) is done — landed on the driver's
5+
`feat/execute-partitions` branch. This file covers only the items that live in the Spark
6+
connector repo (`datafusion-java`) and cannot land in the driver.
7+
8+
## What already landed upstream (driver, Rust)
9+
10+
Context for the connector work below — already merged on the driver's
11+
`feat/execute-partitions` branch:
12+
13+
- **Proto descriptors (always on).** `execute_partitions` always serializes the physical
14+
plan into each descriptor; `read_partition` deserializes it and executes the given
15+
partition index — no re-planning. A `PhysicalExtensionCodec` can be registered alongside
16+
`ContextInit` (`DataFusionDriver::new_with_context_init_and_codec`) to cover custom
17+
`ExecutionPlan` nodes; without one, the default codec covers built-in nodes. A node that
18+
neither codec can encode fails the query (no Substrait/SQL fallback path).
19+
- **Per-connection plan cache.** `read_partition` deserializes a given plan once per
20+
distinct plan-bytes and reuses it across partition indices (cap 16 plans, cleared on
21+
overflow). The win is realized **only if the connector shares one connection across an
22+
executor's task slots** — which is exactly Part 1 below.
23+
- **`partition_mode` option** (`datafusion.partition_mode` ∈ `auto`/`single`/`multi`,
24+
default `auto`). Controls shuffle handling: `auto`/`single` collapse a shuffling plan to
25+
one coalesced partition; `multi` keeps the natural partition count and warns. Opaque to
26+
the connector — it only changes how many descriptors come back.
27+
- **Descriptor format.** Opaque to the connector. Internally:
28+
`[u8 version=1][u32 LE index][proto plan bytes]`. No encoding discriminant — proto is the
29+
only encoding. The connector must keep treating it as an opaque `byte[]`; do not parse it.
30+
31+
---
32+
33+
## Part 1 — per-executor connection cache
34+
35+
Connector-only. Independently valuable; ships without any further driver dependency.
36+
**This is the prerequisite that makes the driver's per-connection plan cache pay off.**
37+
38+
### Problem
39+
40+
`AdbcColumnarPartitionReader` opens its own `AdbcDatabase` + `AdbcConnection` in every task.
41+
On one executor JVM, the 8 (or however many) task slots each repeat driver load, provider
42+
registration, and session setup against the same cdylib. The driver manager already de-dups
43+
the native library load; the expensive part — session construction and the `ContextInit`
44+
provider registration — is repeated per task.
45+
46+
### Approach
47+
48+
Cache the expensive ADBC objects per executor JVM, keyed by what fully determines them: the
49+
driver path/manifest plus the sorted database-option map (everything in
50+
`AdbcOptions.driverParameters()`). Two JVM-lifetime statics:
51+
52+
```
53+
static final ConcurrentHashMap<Key, AdbcDatabase> DATABASES
54+
static final ConcurrentHashMap<Key, AdbcConnection> CONNECTIONS // see thread-safety
55+
```
56+
57+
`Key` = (driver, immutable copy of the database-option map). `AdbcOptions` is already
58+
`Serializable` and rides to the executor inside `AdbcInputPartition`, so the key is
59+
reconstructable there with no new wire fields.
60+
61+
### Thread-safety — decide the sharing level
62+
63+
ADBC connections are not guaranteed thread-safe across concurrent statement execution; this
64+
varies by driver. Two levels:
65+
66+
1. **Safe default — cache the `AdbcDatabase`, open a `Connection` per task.** Shares the
67+
costly driver + database init; each task still gets its own short-lived connection +
68+
statement. Lowest risk.
69+
2. **Further — cache the `AdbcConnection` too**, with one `AdbcStatement` per task. Only if
70+
the DataFusion driver documents a connection as safe for concurrent statements.
71+
72+
**Driver fact (verified):** `ContextInit` runs **per database**
73+
(`new_database_with_database_opts` in the driver's `src/lib.rs`), not per connection. So
74+
**level 1 already removes the provider-registration cost** — caching the `AdbcDatabase` is
75+
sufficient for the Part 1 win. Level 2 only adds value for the driver's per-connection plan
76+
cache, which needs a *shared connection* to dedup plan deserialization across task
77+
slots. Pursue level 2 only if the driver's connection is confirmed safe for concurrent
78+
statements; otherwise stay at level 1 and accept per-task plan deserialization.
79+
80+
### Lifecycle
81+
82+
Spark executors are long-lived; JVM-lifetime statics would leak connections across many jobs.
83+
Mitigate:
84+
85+
- Register a shutdown/cleanup that closes cached handles. Spark `TaskCompletionListener` is
86+
per-task — wrong scope; prefer a JVM shutdown hook / `ShutdownHookManager` entry, or a small
87+
idle-eviction cache.
88+
- Bound the cache (size or idle TTL) so distinct queries/options do not accumulate handles
89+
without limit.
90+
- Close order: statements → connection → database → release the `RootAllocator`, matching the
91+
existing reader's cleanup ordering.
92+
93+
### Touch points
94+
95+
- `AdbcColumnarPartitionReader` (constructor): fetch-or-create the cached database/connection
96+
instead of opening unconditionally.
97+
- New small holder class (e.g. `AdbcConnectionPool`) for the statics, the `Key`, and cleanup.
98+
- No descriptor/wire changes; no driver changes.
99+
100+
### Tests
101+
102+
- Unit: two readers with identical options share one cached database/connection; differing
103+
options do not.
104+
- Concurrency: M threads build readers against one key concurrently → exactly one database
105+
created (and at level 2, one connection), no races.
106+
- E2E (`AdbcSourceTest`, driver-gated): existing multi-partition assertions still pass; add a
107+
counter/log assertion that provider registration ran once per executor, not once per task.
108+
109+
---
110+
111+
## Sequencing
112+
113+
1. **Part 1** — per-executor connection cache (level 1: cache `AdbcDatabase`). Ships now;
114+
removes per-task connection/provider-setup duplication and enables the driver's plan cache.
115+
116+
The descriptor-shrink follow-up (ship the plan once, descriptors reference it) was
117+
**abandoned**: it cannot stay within ADBC's self-contained-partition contract. The full plan
118+
rides each of the N descriptors by design; keep descriptors opaque and ship them whole.
119+
120+
## Risks
121+
122+
- **R1 — connection thread-safety. RESOLVED: level 2 is unsound; ship level 1.** Verified by
123+
reading the full call path. The arrow-adbc Rust FFI exporter (`adbc_ffi 0.23`
124+
`driver_exporter.rs`, which our cdylib links) reaches the inner connection on *every* C call
125+
via `connection_private_data`, which casts the shared `private_data` pointer to a
126+
`&mut ExportedConnection` with **no lock** (`connection_read_partition` → `try_connection(&mut
127+
self)` → `connection.read_partition(...)`). Even though `Connection::read_partition` is `&self`
128+
at the trait level, the wrapper manufactures a `&mut` to the same connection object on each
129+
call, so two task threads sharing one connection alias `&mut` → undefined behavior, regardless
130+
of the driver being internally `Sync`. No layer above serializes: the Java `JniConnection`
131+
holds no lock (bytecode-confirmed), the C driver manager is a thin dispatch, and ADBC's
132+
contract says connections are not thread-safe. Serializing calls in Java would defeat the
133+
parallelism that motivates sharing. **Decision:** default to level 1 (cache `AdbcDatabase`,
134+
per-task connection); keep level 2 behind `-Dadbc.pool.shareConnection=true` only for
135+
experiments / a future fixed exporter, with a loud one-time warning. The driver's
136+
per-connection plan cache then dedups only within a task (one partition per task = one
137+
deserialize anyway) — correct, just no cross-slot plan-cache win.
138+
- **R2 — cache lifecycle leaks.** JVM-lifetime statics on long-lived executors must be
139+
bounded/evicted and closed on shutdown, or they leak native handles.
140+
- **R4 — descriptor size ×N.** Proto descriptors carry the full plan per partition (no shrink
141+
is coming). The driver logs when a single proto descriptor exceeds 8 MiB; watch for it on
142+
plans that inline data or very large queries.

‎docs/adbc-spark-partitioning.md‎

Lines changed: 146 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,146 @@
1+
# Partitioning: DataFusion table providers → Spark tasks
2+
3+
How DataFusion partitions become Spark tasks through the `adbc-datafusion`
4+
connector, how to size them, and what a `TableProvider` author should do. Pairs
5+
with the connector docs under `examples/adbc-datafusion-driver/`
6+
(`PARTITIONS.md`, `UPSTREAM_PARTITIONS_PLAN.md`).
7+
8+
## The mapping is 1:1, and the provider owns it
9+
10+
```
11+
provider.scan() output partitions
12+
→ DataFusion physical plan partition count N
13+
→ executePartitioned() = N descriptors
14+
→ N Spark InputPartitions = N Spark tasks (scan stage)
15+
```
16+
17+
For a pushed scan (projection / filter / limit only, no DataFusion-side join or
18+
aggregation), the physical plan's output partitioning **is** the provider's
19+
`scan()` partitioning. So **the provider's `scan()` partition count is the Spark
20+
task count.** Spark never splits or merges a partition afterward; the cluster's
21+
core count only bounds how many of the `N` run concurrently (the rest run in
22+
waves). `df.rdd().getNumPartitions() == N`.
23+
24+
## The Spark-task sweet spot
25+
26+
- **Floor:** `N` ≥ total executor cores, or cores sit idle. ~2–4× cores gives
27+
slack for skew and stragglers.
28+
- **Per-partition size:** large enough to amortize per-task overhead. Canonical
29+
target ≈ **128–256 MB** of data (Spark's `spark.sql.files.maxPartitionBytes`
30+
default is 128 MB), or ≥ ~100 ms–seconds of work.
31+
- **Ceiling — and ADBC makes it lower.** Beyond Spark's generic per-task
32+
scheduling cost, this connector adds two per-partition costs: each ADBC
33+
partition descriptor carries the serialized query bytes (duplicated ×N on the
34+
driver), and **each task re-plans the whole query** in `read_partition`. So
35+
thousands of partitions is far costlier here than for a native Spark file
36+
source. Keep `N` in the hundreds, not tens of thousands; prefer bigger
37+
partitions over a huge count.
38+
39+
Rule of thumb: `N ≈ clamp(total_bytes / target_bytes, floor, ceiling)`, balanced
40+
by **bytes**, not by split count.
41+
42+
## The model mismatch you must account for
43+
44+
DataFusion and Spark size scans differently:
45+
46+
- **Spark native file source — byte-bound.** `N = Σ ceil(file_size / 128MB)`.
47+
Large data ⇒ many partitions, `N ≫ cores`, run in waves; each task ≈ 128 MB.
48+
- **DataFusion scan — parallelism-bound.** `ListingTable` packs files into
49+
**`target_partitions` groups** (≈ machine cores), with
50+
`repartition_file_min_size` (~10 MB) as a floor and optional single-file
51+
splitting to *reach* `target_partitions`. So DataFusion's default
52+
`N ≈ target_partitions`, **not** `bytes / 128MB`.
53+
54+
Consequence: if a provider mimics DataFusion's default (`N = target_partitions`)
55+
and the connector sets `target_partitions = defaultParallelism` (≈ cores), then
56+
**large data → ~cores giant tasks** (`total/cores` each) → memory pressure and
57+
poor sizing. For feeding Spark you usually want the **byte-bound** model — more
58+
partitions than cores for big data.
59+
60+
**So a custom provider feeding Spark should size by bytes, not blindly use
61+
`target_partitions` as the literal partition count.**
62+
63+
## Provider recipe
64+
65+
`scan(state, …)` receives the session. Read the connector-supplied hint, then
66+
decide by what the provider knows:
67+
68+
```text
69+
T = state.config().target_partitions() // connector hint ≈ desired parallelism
70+
// (set from SparkContext.defaultParallelism)
71+
target_bytes = ~128–256 MB // bias larger for ADBC to keep N down
72+
min_bytes = floor so partitions never get tiny // ~ tens of MB
73+
74+
if total_bytes known AND splittable (files / row groups / key ranges):
75+
N = clamp(ceil(total_bytes / target_bytes), 1, CEILING) // byte-bound
76+
bin-pack splits into N groups BALANCED BY BYTES // not by count
77+
do not split below min_bytes; merge small splits
78+
79+
elif split_count known but not bytes (shards / external partitions):
80+
N = min(split_count, CEILING) // natural partitioning
81+
if split_count >> T: coalesce shards into ~T balanced groups
82+
83+
else (size and splits unknown; opaque / streaming):
84+
N = T // fall back to the hint
85+
lean slightly high — stragglers hurt more than overhead
86+
```
87+
88+
- **Balance by bytes, not split count.** Spark can't rebalance a fat partition
89+
(each task is atomic). One 10 GB partition among 100 small ones stalls the
90+
whole stage. Bin-pack splits so partition bytes are even.
91+
- `T` (the connector's `target_partitions`) is the **hint / cap / fallback** —
92+
not the data-bound count. Use it when bytes are unknown, or as a parallelism
93+
ceiling. For known data, **bytes drive `N`**.
94+
95+
## Determinism caveat (specific to the ADBC partition path)
96+
97+
`execute_partitions` counts `N` on the **driver**; `read_partition` re-plans on
98+
the **executor**. Partition `index i` must mean the same slice both times, so
99+
**`scan()` must produce the identical `N` in both places.**
100+
101+
- If the provider keys `N` off `target_partitions`, the driver **pins it in the
102+
descriptor** and `read_partition` re-plans with it held fixed → determinism for
103+
free. (That is why the pin exists.)
104+
- If the provider sizes by **bytes** (ignoring `target_partitions`), the pin does
105+
not help — compute `N` from **stable** inputs (a cached file listing / a fixed
106+
stats snapshot). Otherwise stat drift between driver-plan and executor-read
107+
changes `N` and shifts partitions.
108+
109+
## Connector knobs
110+
111+
- **`target_partitions`** (implemented): defaults to
112+
`SparkContext.defaultParallelism` (total executor cores). The connector issues
113+
`SET datafusion.execution.target_partitions = N` on the planning session; the
114+
driver pins it into each descriptor, so executors re-plan identically. Reaches
115+
the provider via `state.config()`. Optionally pass `k×cores`.
116+
- **`batch_size`** (follow-up, connector-only): would `SET
117+
datafusion.execution.batch_size` on the executor connection in the reader;
118+
`read_partition` inherits it via `copied_config()`. Affects Arrow batch
119+
granularity, not partition count, so no pinning needed.
120+
- **`maxPartitionBytes`**: not a DataFusion session config and the connector
121+
cannot derive it (it does not know the data size). Byte-sizing lives in the
122+
provider's `scan()`; the connector can only forward it as a provider-specific
123+
option.
124+
- **Reported partitioning / storage-partitioned joins
125+
(`SupportsReportPartitioning`)**: not viable over ADBC today — the descriptor
126+
carries no distribution metadata, and DataFusion's hash partitioning does not
127+
match Spark's key-grouped/`bucket` model. See `UPSTREAM_PARTITIONS_PLAN.md`.
128+
129+
## Recommendation (summary)
130+
131+
- **Connector:** keep `target_partitions` defaulted from `defaultParallelism`; it
132+
is the parallelism hint that reaches `scan()` and is pinned for determinism.
133+
- **Provider (the real lever):**
134+
- bytes known + splittable → **byte-bound**: `N = ceil(total_bytes/target_bytes)`,
135+
bin-packed by bytes, merged above `min_bytes`, capped well below "thousands"
136+
(ADBC re-plans per partition); compute `N` from a stable stat snapshot.
137+
- only splits known → natural partitioning, coalesce toward `T` if huge.
138+
- nothing known → `N = T`, lean slightly high.
139+
- default `target_bytes ≈ 128–256 MB`; raise it (fewer, bigger partitions)
140+
before letting `N` reach the thousands.
141+
142+
One line: **the provider's `scan()` partition count is the Spark task count; size
143+
it by bytes toward ~128–256 MB balanced partitions, floored at executor cores and
144+
capped well below thousands because ADBC re-plans per partition — and make that
145+
count a deterministic function of a stat snapshot so the driver plan and executor
146+
reads agree.**

0 commit comments

Comments
 (0)