Data Source V2

• Table of contents {:toc}

Overview

Data Source V2 (DSV2) is Spark's extensible API for integrating external data systems. It is a set of Java interfaces in the org.apache.spark.sql.connector package that allow connectors to plug into Spark's query planning and execution. Connectors opt in to features and optimizations — such as filter pushdown, columnar reads, or catalog support — by implementing specific mix-in interfaces, so a minimal connector can add capabilities incrementally. Notable users include:

• JDBC data source (Spark built-in)
• Apache Iceberg
• Delta Lake
• Lance

Compared to the earlier Data Source V1 API, DSV2 offers:

Entry Points

There are two ways to plug a data source into Spark:

TableProvider

TableProvider returns a Table given a set of options. Implementations must have a public no-arg constructor.

The key methods are:

A TableProvider can also implement SupportsCatalogOptions to participate in DDL such as CREATE TABLE by extracting a catalog identifier from the user-supplied options. This bridges option-based sources to the session catalog, which is how built-in file data sources (Parquet, ORC, etc.) support table creation.

CatalogPlugin

CatalogPlugin is a marker interface for catalog implementations. After instantiation, Spark calls initialize(name, options) with the catalog name and all configuration properties that share the prefix spark.sql.catalog.<name>..

A catalog adds capabilities by mixing in additional catalog interfaces:

CatalogExtension is a special variant that wraps Spark's built-in session catalog and can be used to add custom behavior while delegating to the default implementation.

Catalog Interfaces

The interfaces below are mix-ins that a CatalogPlugin implements to expose specific categories of metadata operations through Spark SQL.

TableCatalog

TableCatalog extends CatalogPlugin and provides methods for Table lifecycle management:

StagingTableCatalog

StagingTableCatalog extends TableCatalog and enables atomic CREATE TABLE AS SELECT and REPLACE TABLE AS SELECT. It returns a StagedTable whose changes only become visible when commitStagedChanges() is called. If the write fails, the catalog remains unchanged.

SupportsNamespaces

SupportsNamespaces adds namespace (database/schema) management:

ViewCatalog

Note: ViewCatalog is a work in progress. The interface is defined but is not yet integrated into Spark's query resolution or planning.

ViewCatalog extends CatalogPlugin and provides methods for view lifecycle management:

FunctionCatalog

FunctionCatalog adds user-defined function management:

ProcedureCatalog

ProcedureCatalog adds stored-procedure support:

Procedures are invoked via CALL catalog.procedure(args).

Table

Table is the central abstraction representing a logical dataset — for example, a directory of Parquet files, a Kafka topic, or a table managed by an external metastore. A Table gains read and write abilities through mix-in interfaces.

A Table provides:

Read and Write Mix-ins

A Table gains read and write abilities by implementing mix-in interfaces:

• SupportsRead adds newScanBuilder(options), which returns a ScanBuilder for batch or streaming reads. See Read Path.
• SupportsWrite adds newWriteBuilder(info), which returns a WriteBuilder for batch or streaming writes. See Write Path.
• SupportsRowLevelOperations enables DELETE, UPDATE, and MERGE INTO through a read-and-rewrite cycle. See Row-Level DML.

Additional mix-ins enable further capabilities:

Read Path

The read path follows a builder pattern that separates logical planning from physical execution:

SupportsRead.newScanBuilder(options)
  └─▸ ScanBuilder          (logical: pushdown negotiation)
        └─▸ Scan            (logical: read schema, description)
              └─▸ Batch     (physical: partitions + reader factory)
                    ├─ InputPartition[]
                    └─ PartitionReaderFactory
                         └─▸ PartitionReader   (per-task I/O)

ScanBuilder

ScanBuilder is the starting point for configuring a read. Spark calls build() to obtain the final Scan.

Before calling build(), Spark negotiates operator pushdown with the scan builder by checking for mix-in interfaces. The pushdown order is:

1. Sample (SupportsPushDownSample)
2. Filter (SupportsPushDownFilters / SupportsPushDownV2Filters)
3. Aggregate (SupportsPushDownAggregates)
4. Limit / Top-N (SupportsPushDownLimit / SupportsPushDownTopN)
5. Offset (SupportsPushDownOffset)
6. Column pruning (SupportsPushDownRequiredColumns)

Each mix-in interface has a method that Spark calls with the relevant operators. The implementation returns the operators it can handle, and Spark applies the remaining operators itself.

Scan

Scan is a logical representation of a configured data source read. It provides:

Implementations must override the method corresponding to the TableCapability declared by their Table.

Batch

Batch is the physical representation of a batch read. It has two methods:

InputPartition and PartitionReader

InputPartition is a serializable handle representing a data split. It may optionally declare preferredLocations() for data locality.

PartitionReaderFactory is serialized to executors and creates a PartitionReader for each input partition. The reader iterates over rows (or ColumnarBatch instances for columnar sources) and is closed after consumption.

Scan Mix-ins

Scan implementations can opt in to additional optimizations by implementing mix-in interfaces:

Write Path

The write path mirrors the read path with a builder pattern that separates logical configuration from physical execution:

SupportsWrite.newWriteBuilder(info)
  └─▸ WriteBuilder           (logical: mode selection)
        └─▸ Write             (logical: description, metrics)
              └─▸ BatchWrite  (physical: commit protocol)
                    ├─ DataWriterFactory
                    │    └─▸ DataWriter      (per-task I/O)
                    ├─ commit(messages[])
                    └─ abort(messages[])

WriteBuilder

WriteBuilder is the starting point for configuring a write. Calling build() returns a logical Write object.

Write modes are configured by mixing in additional interfaces on the builder:

Write

Write is a logical representation of a configured write. Similar to Scan, it bridges to the physical layer:

BatchWrite

BatchWrite defines the two-phase commit protocol for batch writes:

1. createBatchWriterFactory(info) — creates a DataWriterFactory that is serialized and sent to executors.
2. On each executor, the factory creates a DataWriter per partition. If all rows are written successfully, the writer calls commit(); otherwise it calls abort().
3. After all tasks complete, the driver calls either commit(messages[]) (if all tasks succeeded) or abort(messages[]) (if any task failed).

Data written by individual tasks should not be visible to readers until the driver-level commit succeeds.

Distribution and Ordering Requirements

A Write implementation can also implement RequiresDistributionAndOrdering to tell Spark how input data must be distributed and sorted before writing. Spark will insert shuffle and sort nodes as needed to satisfy these requirements.

Row-Level DML

DSV2 provides interfaces for connectors to support DELETE, UPDATE, and MERGE INTO statements.

Filter-Based Delete

The simplest form of DML is a filter-based delete, where entire groups of rows matching a predicate are removed without rewriting individual records.

SupportsDeleteV2 (and the older SupportsDelete) is a Table mix-in for this use case. Its key methods are:

This approach is well-suited for data sources that can drop entire partitions or files without rewriting data.

Row-Level Operations

For more complex DML — such as UPDATE, MERGE INTO, or DELETE operations that cannot be handled by a simple filter — connectors implement SupportsRowLevelOperations. This interface returns a RowLevelOperation, which coordinates a read-and-rewrite cycle:

SupportsRowLevelOperations.newRowLevelOperationBuilder(info)
  └─▸ RowLevelOperation
        ├─ newScanBuilder(options)   → reads affected rows
        └─ newWriteBuilder(info)     → writes rewritten data

Data sources fall into two categories:

• Delta-based sources (sometimes called merge-on-read) can handle a stream of individual row changes (inserts, updates, deletes). The scan only needs to produce the rows that are changing.
• Group-based sources (sometimes called copy-on-write) replace entire groups of rows (e.g. files or partitions). The scan must return all rows in each affected group, including unchanged rows, so that the data source can rewrite the group with the modifications applied.

Expressions

The org.apache.spark.sql.connector.expressions package provides a neutral expression representation used across the DSV2 API:

• Transforms (IdentityTransform, BucketTransform, YearsTransform, etc.) describe table partitioning schemes.
• Sort orders (SortOrder) describe ordering requirements.
• Filter predicates (Predicate and subclasses) represent pushed-down filter conditions.
• Aggregates represent pushed-down aggregate functions.

These expression types are independent of Spark's internal Catalyst expressions, allowing connectors to avoid a dependency on Spark internals.

Streaming

The same Table and Scan abstractions support streaming queries. A Table that declares MICRO_BATCH_READ or CONTINUOUS_READ capabilities provides streaming reads through:

A Table that declares STREAMING_WRITE supports streaming writes through Write.toStreaming(), which returns a StreamingWrite.

Further Reading

• API documentation (Javadoc) for the full interface reference.
• Data Sources for the user-facing guide to built-in data sources (DSV1).
• Python Data Source API for writing lightweight connectors entirely in Python.
• Storage Partition Join for how DSV2 partitioning reporting enables join optimizations.