Data Source V2

• Overview
• Entry Points
  • TableProvider
  • CatalogPlugin
• Catalog Interfaces
  • TableCatalog
  • StagingTableCatalog
  • SupportsNamespaces
  • ViewCatalog
  • FunctionCatalog
  • ProcedureCatalog
• Table
  • Read and Write Mix-ins
• View
• Read Path
  • ScanBuilder
  • Scan
  • Batch
  • InputPartition and PartitionReader
  • Scan Mix-ins
• Write Path
  • WriteBuilder
  • Write
  • BatchWrite
  • Distribution and Ordering Requirements
• Row-Level DML
  • Filter-Based Delete
  • Row-Level Operations
• Expressions
• Streaming
• Building a Connector
  • Step 1: A Minimal Batch Read
  • Step 2: Projection and Filter Pushdown (Optional)
  • Step 3: Batch Writes
  • Step 4: Catalog Integration (Optional)
• Further Reading

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 so that DataFrameReader.load and DataFrameWriter.save resolve the user-supplied options to an identifier in a catalog (the session catalog unless the options name another), which enables existence checks and create-on-save for those calls.

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:

Spark calls createTable(ident, tableInfo) (since 4.1.0) for CREATE TABLE, REPLACE TABLE, and CTAS, so a new catalog overrides that form. Its default forwards to the deprecated createTable(ident, columns, partitions, properties) and then createTable(ident, schema, partitions, properties) (deprecated since 4.1.0 and 3.4.0), which keeps older catalogs working but drops the table’s constraints. The default rejects a requested write distribution and ordering. A catalog that reports SUPPORTS_CREATE_TABLE_WITH_WRITE_DISTRIBUTION_AND_ORDERING overrides createTable(ident, tableInfo), and for a staging catalog also the three stage* forms below. A StagingTableCatalog is reached through its own methods instead, except for a plain CREATE TABLE, which still arrives at createTable.

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. Spark reaches these paths through stageCreate(ident, tableInfo), stageReplace(ident, tableInfo), and stageCreateOrReplace(ident, tableInfo) rather than createTable, so a staging catalog overrides those three in addition to the TableCatalog methods. Their defaults forward to the deprecated forms, which throw, and reject a requested write distribution and ordering.

SupportsNamespaces

SupportsNamespaces adds namespace (database/schema) management:

ViewCatalog

ViewCatalog extends CatalogPlugin and provides methods for view lifecycle management. View DDL (CREATE/REPLACE/ALTER/DROP/SHOW VIEWS) and reading from a view (SELECT) are resolved and executed against the catalog. Connectors that expose both tables and views in a single shared identifier namespace implement RelationCatalog (since 4.2.0), which extends both TableCatalog and ViewCatalog and adds loadRelation(ident) so a lookup that may be either kind resolves in one round trip. A catalog that implements TableCatalog and ViewCatalog directly, without RelationCatalog, is rejected when it is initialized. A view’s metadata is represented by View.

Note: The built-in session catalog (V2SessionCatalog) is not a ViewCatalog; views in the session catalog continue to use the v1 (Hive-compatible) code path. DSV2 view support applies to custom catalogs registered via spark.sql.catalog.<name>.

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:

View

View (since 4.2.0) is the metadata abstraction for a view — a named SQL query managed through a ViewCatalog. Unlike a Table, a view has no data of its own: it stores a SQL query along with the context needed to resolve and expand that query consistently whenever the view is referenced.

A View provides:

When a view is read (SELECT), Spark resolves and expands queryText() using the captured catalog, namespace, and SQL configs, so the view behaves consistently regardless of the session state at query time.

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 (SupportsPushDownTableSample)
2. Filter (SupportsPushDownV2Filters / SupportsPushDownFilters)
3. Join (SupportsPushDownJoin, since 4.1.0)
4. Aggregate (SupportsPushDownAggregates)
5. Variant extraction (SupportsPushDownVariantExtractions, since 4.1.0)
6. Limit, top-N and offset (SupportsPushDownLimit, SupportsPushDownTopN, SupportsPushDownOffset)
7. Column pruning (SupportsPushDownRequiredColumns)

Spark offers aggregate, limit, top-N and offset pushdown only when no filter is left to evaluate above the scan, so a filter returned by pushPredicates blocks them. For the operators that can be partly handled, the return value tells Spark what it must still evaluate itself:

• SupportsPushDownV2Filters: pushPredicates(predicates) receives the candidate Predicate instances and returns the ones that still need to be evaluated after the scan. pushedPredicates() returns the predicates the source accepted. A predicate may appear in both: a source that can only skip whole files or row groups on a predicate reports it as pushed and also returns it, so Spark re-checks it row by row. Spark evaluates only what pushPredicates returns (plus anything it could not translate), so a predicate that is neither returned nor applied by the reader is silently dropped from the query.
• SupportsPushDownFilters is the pre-3.3 form of the same contract over the V1 org.apache.spark.sql.sources.Filter type: pushFilters(filters) returns the post-scan filters and pushedFilters() reports the accepted ones. New connectors should implement SupportsPushDownV2Filters; Spark checks for SupportsPushDownFilters first, so a ScanBuilder should implement one or the other.
• SupportsPushDownRequiredColumns: pruneColumns(requiredSchema) passes the narrowest schema the query needs. Pruning may be partial (for example top-level columns only), and Scan.readSchema() must report whatever pruning was actually applied.
• SupportsPushDownLimit.pushLimit, SupportsPushDownTopN.pushTopN, SupportsPushDownOffset.pushOffset and SupportsPushDownAggregates.pushAggregation return true when the source accepts the operator. For limit and top-N, isPartiallyPushed() (default true) tells Spark whether it must keep the operator above the scan. For aggregation, supportCompletePushDown(aggregation) tells Spark whether the source handles the aggregation completely.

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. For a SupportsStorageFilters scan that Spark offered a storage filter, createReader(partition, request) or createColumnarReader(partition, request) creates the reader instead. The StorageFilterRequest carries the filter, and whether the reader appends a boolean column that is true only on the rows the filter was evaluated on and kept. When the filter throws StorageFilterNotAppliedException, the reader reads the rest of the unit without it. Scala does not check exceptions, so a reader written in Scala must catch it all the same.

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 task attempt. When the attempt has written its rows and the commit coordinator allows it, Spark calls commit(), which returns a WriterCommitMessage; if the attempt fails, Spark calls abort(). Failed tasks are retried.
3. On the driver, onDataWriterCommit(message) runs as each task’s message arrives. If every task commits, Spark calls commit(messages[]); if the job fails after task retries, or onDataWriterCommit or commit throws, it calls abort(messages[]).

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 (Transform, built with factory methods on Expressions such as identity, bucket and years) 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.

Building a Connector

Spark’s own test connectors make a working example of each step: JavaSimpleDataSourceV2 for a minimal read, JavaAdvancedDataSourceV2WithV2Filter for pushdown, and JavaSimpleWritableDataSource for writes, exercised by DataSourceV2Suite. The three share a few test helpers. TestingV2Source is a test trait over TableProvider: it implements inferSchema (two int columns, i and j) and forwards getTable(schema, partitioning, properties) to a one-argument getTable(options). The excerpts override the one-argument form; a connector implements the three-argument form. JavaSimpleBatchTable is a Table that declares BATCH_READ and returns that schema (through the deprecated schema(); a new Table implements columns()); JavaRangeInputPartition carries a start and end; and JavaSimpleScanBuilder is a ScanBuilder that is also its own Scan and Batch.

Step 1: A Minimal Batch Read

The smallest connector is a TableProvider that returns a Table, a Table that declares BATCH_READ and returns a ScanBuilder, and a ScanBuilder whose Batch plans partitions and creates readers. JavaSimpleDataSourceV2 is exactly that:

public class JavaSimpleDataSourceV2 implements TestingV2Source {

  static class MyScanBuilder extends JavaSimpleScanBuilder {
    @Override
    public InputPartition[] planInputPartitions() {
      InputPartition[] partitions = new InputPartition[2];
      partitions[0] = new JavaRangeInputPartition(0, 5);
      partitions[1] = new JavaRangeInputPartition(5, 10);
      return partitions;
    }
  }

  @Override
  public Table getTable(CaseInsensitiveStringMap options) {
    return new JavaSimpleBatchTable() {
      @Override
      public ScanBuilder newScanBuilder(CaseInsensitiveStringMap options) {
        return new MyScanBuilder();
      }
    };
  }
}

planInputPartitions() runs once on the driver and returns one InputPartition per task. Each partition is serialized to an executor, so it should hold only what identifies the split (here a range of integers; for a file source, a path and a byte range). The PartitionReaderFactory is serialized the same way and creates a PartitionReader on the executor. The reader is a cursor: next() advances, get() returns the current row, close() releases any handle. The one in JavaSimpleReaderFactory counts through the partition’s range:

return new PartitionReader<InternalRow>() {
  private int current = p.start - 1;

  @Override
  public boolean next() {
    current += 1;
    return current < p.end;
  }

  @Override
  public InternalRow get() {
    return new GenericInternalRow(new Object[] {current, -current});
  }

  @Override
  public void close() {
  }
};

A PartitionReader<InternalRow> produces InternalRow, which holds each column in Spark’s internal representation rather than the external one a Row carries: a boxed int is fine for an IntegerType column, but a StringType column holds UTF8String (not java.lang.String), DecimalType holds Decimal, DateType holds an int of days, TimestampType a long of microseconds, BinaryType a byte[], and nested types hold ArrayData, MapData and InternalRow. InternalRow and GenericInternalRow live in Spark’s internal catalyst package and have no API docs, but they are the row types the connector API uses. The accessor names on InternalRow (getUTF8String, getDecimal, getInt, getLong, getBinary, getArray, getMap, getStruct) are the reference for each type. GenericInternalRow takes Object[], so a value of the wrong representation compiles and fails on the executor with a ClassCastException.

With this much, spark.read.format("...JavaSimpleDataSourceV2").load() returns ten rows, and the select and filter in the “simplest implementation” test work too: Spark applies the projection and the filter itself, above the scan.

Step 2: Projection and Filter Pushdown (Optional)

A connector that can skip columns or rows adds the pushdown mix-ins to its ScanBuilder. The list, the order Spark applies them in, and the residual-filter contract are under ScanBuilder. JavaAdvancedDataSourceV2WithV2Filter accepts only i > <int literal> and returns everything else for Spark to evaluate. Unlike the test, which asserts the shapes it expects, the following version checks them, because Spark can also push i > j as a > predicate; it also matches on the public NamedReference rather than the test’s FieldReference. It accepts a threshold only from -1 up, because the planner starts the first partition at the threshold plus one and the source’s data is 0 to 9; a lower threshold stays with Spark. The planner enforces one lower bound, so the example accepts the first matching predicate and returns any others to Spark. Replacing predicates on each call is right while supportsIterativePushdown() returns false (the default); an iterative source accumulates them instead:

@Override
public Predicate[] pushPredicates(Predicate[] predicates) {
  List<Predicate> supported = new ArrayList<>();
  List<Predicate> unsupported = new ArrayList<>();
  for (Predicate p : predicates) {
    if (supported.isEmpty() && isIGreaterThanInt(p)) {
      supported.add(p);
    } else {
      unsupported.add(p);
    }
  }
  this.predicates = supported.toArray(new Predicate[0]);
  return unsupported.toArray(new Predicate[0]);
}

private static boolean isIGreaterThanInt(Predicate p) {
  return p.name().equals(">")
      && p.children()[0] instanceof NamedReference column
      && column.fieldNames().length == 1
      && column.fieldNames()[0].equals("i")
      && p.children()[1] instanceof Literal<?> value
      && value.value() instanceof Integer bound
      && bound >= -1;
}

@Override
public Predicate[] pushedPredicates() {
  return predicates;
}

@Override
public void pruneColumns(StructType requiredSchema) {
  this.requiredSchema = requiredSchema;
}

pushedPredicates() reports what the source accepted. The Batch in the same class uses the accepted predicate to plan fewer partitions, and its reader fills only the columns in requiredSchema. The “advanced implementation with V2 Filter” test in DataSourceV2Suite checks the planned Batch for a few filter and select combinations. EXPLAIN does not show pushedPredicates(); it shows Scan.description() inside the BatchScan node, after the output columns. Its default is getClass().toString(), so override it to include the accepted predicates and the pruned columns.

Step 3: Batch Writes

A writable table declares BATCH_WRITE and implements SupportsWrite. For SaveMode.Overwrite it also declares TRUNCATE (or OVERWRITE_BY_FILTER) and implements the matching builder mix-in, such as SupportsTruncate; see TableCapability for the others. Spark checks the capability set during analysis, before calling any builder, so a missing value fails the query even when the mix-in is implemented. JavaSimpleWritableDataSource declares BATCH_READ, BATCH_WRITE and TRUNCATE. A TableProvider like this one, which does not implement SupportsCatalogOptions, supports only the append and overwrite save modes; the default, errorifexists, and ignore fail with UNSUPPORTED_DATA_SOURCE_SAVE_MODE. A provider that implements SupportsCatalogOptions takes its catalog and identifier from the write options, and Spark runs those two modes as CREATE TABLE ... AS SELECT in that catalog.

The write then follows the protocol under BatchWrite. A file-based connector typically has each DataWriter write its task’s output to a staging location, and BatchWrite.commit on the driver publish the staged files. The staging location must be reachable from both sides: executors write there and the driver renames or registers from there, usually on different machines. JavaSimpleWritableDataSource stages under <path>/_temporary/<queryId> on the table’s own file system, writes one file per task (JavaCSVDataWriterFactory.createWriter), and commits by renaming the staged files into the table directory one at a time (MyBatchWrite.commit). commit is a single driver-side call, but that alone does not make the publication atomic: a reader listing the directory between two renames sees a partial result. A connector that needs all-or-nothing visibility has to provide it itself, for example by writing a manifest or committing a transaction as the last step.

Two shortcuts in the test connector are not safe outside a test. Its commit publishes every file under the staging directory rather than the ones named in messages, so a partial file left by an executor lost mid-task would be published next to the retry’s file; a real connector commits only the files its messages name. And it truncates eagerly in Write.toBatch(), before any task runs, so a failed overwrite leaves the table empty; a real connector defers the delete to BatchWrite.commit.

A WriterCommitMessage is returned by DataWriter.commit() on an executor and delivered to BatchWrite.commit on the driver, where the whole array is held in memory. It should carry a reference to what the task wrote (a path, an object key, a transaction id), never the rows. This connector keeps its state in the staging directory and returns null. BatchWrite.abort(messages) removes the staged output; it receives null entries for tasks that never committed and must tolerate them. DataWriter.abort() on the executor side deletes that task’s own file.

DataWriter.write receives rows in the internal representation described in Step 1, so a StringType column is read back with record.getUTF8String(i).toString().

Step 4: Catalog Integration (Optional)

A TableProvider is reached by class name from spark.read.format(...). To reach the same tables through SQL identifiers and DDL (CREATE TABLE demo.db.t, INSERT INTO demo.db.t, SHOW TABLES IN demo.db), implement TableCatalog, whose loadTable returns the same Table class the TableProvider builds, and register it as under CatalogPlugin:

spark.sql.catalog.demo=com.example.DemoCatalog
spark.sql.catalog.demo.warehouse=/path/to/warehouse    # arrives as options.get("warehouse")

CREATE TABLE arrives as createTable(ident, tableInfo), and tableInfo.columns() is a Column[]; Table.columns() returns the same type, while the read and write paths speak StructType. Build the array with Column.create(name, dataType, nullable). spark.sql.defaultCatalog names the catalog that is current when a session has not set one, so after SET spark.sql.defaultCatalog=demo, db.t resolves to demo.db.t.

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.
• Native Data Sources for writing connectors entirely in native code, such as Rust or C++.
• Storage Partition Join for how DSV2 partitioning reporting enables join optimizations.