Spark DataSource V2 architecture — catalogs, capabilities, pushdown negotiation, commit protocol

Why architecture matters here

The architecture matters because the difference between pushdown working and not working is not a percentage — it is orders of magnitude, and it is invisible. A filter evaluated at the source touches the rows that match. The same filter evaluated in Spark means every row crossed the network, was deserialised, was materialised in executor memory, and was then discarded. The query returns identical results either way. The only difference is that one version reads a gigabyte and the other reads a terabyte, and nothing in the output tells you which one you got.

It matters because the V1 model made this failure structurally undetectable, and V2's central design decision is aimed squarely at that. When the API has no channel for 'I could not handle this filter', a source that ignores pushdown and a source that implements it perfectly are indistinguishable to the optimiser. Both return correct rows. V2's requirement that pushFilters return the filters it did not handle converts a silent correctness-preserving performance catastrophe into a value the planner can act on and a human can read in EXPLAIN.

It matters because catalogs turned data sources from paths into first-class objects with lifecycles. V1 could read from and write to a location. It could not create a table, evolve its schema, alter its properties, or drop it, because the API had no concept of a table as a managed entity — only of bytes at a location. The catalog interfaces are what let a connector support DDL, which is what lets a lakehouse table format behave like a database from inside Spark SQL rather than like a directory you agreed to interpret consistently.

And it matters because it unified batch and streaming behind one set of interfaces, which sounds like an implementation convenience and is actually a correctness property. Under V1, a source with separate batch and streaming implementations had two code paths that could — and routinely did — disagree about schema, about type mapping, about how nulls or timestamps were handled. In V2 the Table is one object that declares which capabilities it supports; the batch and streaming scans share the schema and the type handling by construction, and the class of bug where a table means one thing in a batch job and another in a streaming job simply stops existing.

The architecture: every piece explained

The TableCatalog is the entry point and the thing V1 had no equivalent of. It maps a namespace and identifier to a Table, and it is what makes spark.sql("SELECT * FROM my_catalog.db.tbl") resolve. Beyond loading, it carries the DDL surface: create, alter, drop, list. A connector that implements it gets to be a catalog in Spark SQL rather than a format you point a path at, and that is the difference between a table users can manage and a location they must remember the conventions for.

The Table is the central abstraction and it is deliberately thin. It has a name, a schema, a partitioning, and properties — and a capabilities() set. That set is the honest declaration of what this table can do: BATCH_READ, BATCH_WRITE, MICRO_BATCH_READ, STREAMING_WRITE, TRUNCATE, OVERWRITE_BY_FILTER, and so on. Spark consults it before planning, so an unsupported operation fails at analysis with a clear message rather than at runtime with a confusing one.

The ScanBuilder is where the negotiation happens and it is the most important object in the API. Spark obtains one, then interrogates it by checking which mixins it implements. SupportsPushDownFilters offers filters and — the load-bearing detail — its pushFilters returns the subset it could not handle, which Spark then plans to evaluate itself. SupportsPushDownRequiredColumns prunes columns. Later mixins push aggregates, limits, and joins. Each is optional; a source implementing none is still correct, just slow. That is the design: capability is opt-in, correctness is not negotiable.

SupportsReportStatistics is the quiet one that changes plans. If the source can estimate row count and size — even roughly — after pushdown has been applied, that estimate flows into the cost-based optimiser and decides things like whether a join is a broadcast. A source that reports nothing gets default estimates, which are usually wrong, which is how you get a shuffle join where a broadcast would have finished in seconds. Reporting bad statistics, however, is worse than reporting none: it produces confidently wrong plans.

On the execution side, build() yields a Scan, which for batch yields a Batch, which yields InputPartition objects via planInputPartitions() — this is where parallelism is decided, on the driver — and a PartitionReaderFactory, which is serialised to executors and creates the reader that actually pulls rows. The factory can advertise columnar support, letting Spark skip row-by-row deserialisation entirely. The write side mirrors this: WriteBuilder to Write to BatchWrite, with DataWriterFactory on executors and a driver-side commit/abort that receives every task's commit message — the two-phase protocol that makes atomic writes possible.

End-to-end flow

A query arrives: SELECT name FROM cat.db.users WHERE age > 30. The analyser resolves cat to a TableCatalog implementation, calls loadTable with the remaining identifier, and gets a Table. It reads the schema to resolve name and age and checks capabilities() contains BATCH_READ. If it does not, the query fails right here, at analysis, with a message naming the missing capability — before any executor has been asked to do anything.

The optimiser calls newScanBuilder and starts the negotiation. It checks for SupportsPushDownFilters, and if present offers [age > 30]. The source examines it. Suppose it is a JDBC source and this translates cleanly to SQL: it accepts, and returns an empty array of unhandled filters. Spark takes that literally and removes the Filter node from the plan entirely. Had the source returned the filter back, Spark would have kept the node and evaluated it after reading — correct either way, wildly different in cost, and the return value is the only thing that decided which.

Then column pruning: SupportsPushDownRequiredColumns gets told the query needs only name. Note the order dependency here, which trips people up — the required columns must include anything needed by filters the source did not handle, because Spark still has to evaluate those, and it cannot evaluate a filter on a column it asked the source not to return. The API's ordering rules exist because these two pushdowns interact.

With the negotiation settled, Spark asks for statistics if the source offers them and finalises the plan — join strategies, shuffle partitions, everything downstream keyed off that estimate. Then build(), toBatch(), planInputPartitions(). The source decides how to split the work: JDBC might partition by a numeric column's ranges, a file source by file or by block. This is a driver-side decision and it sets the query's parallelism ceiling — return one partition and the whole cluster reads through one core, no matter how many executors you provisioned.

It is worth pausing on what has happened by this point, because it is the whole argument for the API's shape. Before a single byte has been read, two independent planners have agreed a division of labour: the source has said what it will do, Spark has planned around the gaps, and both halves of that agreement are written into the physical plan where a human can read them. Under V1 this conversation did not exist — Spark assumed and the source complied or quietly did not. Everything that makes V2 verbose is in service of making that negotiation explicit, which is also what makes it debuggable.

The PartitionReaderFactory is serialised and shipped to executors — so everything it closes over must be serialisable, which is the most common trap in connector development, since a live connection is not. Each task deserialises the factory, creates a reader for its partition, opens its own connection, and iterates. Rows flow into Spark's engine, and if the factory advertised columnar support they arrive as batches that skip per-row deserialisation.

The write path inverts it and adds a commit protocol worth understanding. Each task writes to a staging location and returns a WriterCommitMessage to the driver — a small object describing what it produced. The driver collects all of them and calls commit(messages) once, and it is that call that makes the write visible: publishing files, updating a manifest, whatever the format requires. If any task fails, the driver calls abort instead and the staged work is discarded. This is why V2 connectors can offer atomicity that V1's write-and-hope model could not — the visibility decision is a single driver-side action taken after every task has succeeded, and speculative execution or task retries can produce duplicate staged output without ever producing duplicate committed output.

Failure modes and mitigations

• Claiming a filter you did not apply. Returning an empty unhandled array while ignoring the filter is a correctness bug, not a performance one: Spark drops the Filter node and trusts you. The query returns rows that should have been excluded. This is the most dangerous mistake the API permits.
• Non-serialisable reader factories. The factory crosses to executors. Closing over a connection, a session, or a config object with a socket in it produces a NotSerializableException at submit time — or worse, works locally and fails only on a real cluster.
• Bad statistics. A wrong row-count estimate is worse than none, because the optimiser trusts it: an underestimate triggers a broadcast of something that will not fit and the driver dies with an OOM that names nothing about your connector.
• Single-partition scans. A planInputPartitions returning one partition caps the query at one core. It is correct, it is fast on the test dataset, and it does not scale at all. This is the classic JDBC connector mistake.
• Skewed partitions. Splitting by a column with an uneven distribution gives one task most of the data. The stage's duration becomes that task's duration, and adding executors changes nothing.
• Non-idempotent commit. The driver can call commit after a partial failure and recovery; task output can be duplicated by speculation. A commit that appends without deduplicating turns a retry into duplicate data.
• Column pruning that drops filter inputs. Pruning away a column that an unhandled filter still needs produces an analysis error or, in a buggy source, wrong results. The pushdown mixins interact and the ordering contract is not optional.

Operational playbook

• Read the physical plan on every change. EXPLAIN FORMATTED shows exactly which filters reached the source and which columns were requested. This is the connector developer's primary instrument, and it is the only way to see a pushdown that silently stopped working.
• Test that pushdown actually happened, not just that results are right. Correct results prove nothing about pushdown — that is the whole problem. Assert on the plan string in a test, so a regression fails the build instead of quietly costing money.
• Instrument bytes read at the source. The source's own metric for how much it scanned is the ground truth. Compare it against the result size; a large ratio is a pushdown that is not working regardless of what the plan claims.
• Start by implementing nothing optional. A correct scan with no pushdown mixins is a working connector. Add capabilities one at a time, each with a plan-assertion test. Claiming capabilities you have not verified is how correctness bugs get in.
• Make commit idempotent and prove it. Kill the driver between task completion and commit in a test and confirm the retry produces one copy of the data. This is the property users assume and rarely check.
• Watch task duration distribution, not the average. Partition skew shows up as a long tail in the stage's task timeline. The average hides it completely; the max is the number that determines your stage duration.
• Report statistics only when you can compute them honestly. Returning nothing is a safe default that gets you conservative plans. Returning a guess gets you confident wrong ones.