This is an automated email from the ASF dual-hosted git repository.
szehon-ho pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 326704c48af5 [SPARK-60437][DOC] Fix stale API references in the Data
Source V2 page and add a connector walkthrough
326704c48af5 is described below
commit 326704c48af5fcd054aa7f05fc24fa4e9c370ca6
Author: lnc <[email protected]>
AuthorDate: Tue Oct 27 17:10:05 2026 -0700
[SPARK-60437][DOC] Fix stale API references in the Data Source V2 page and
add a connector walkthrough
### What changes were proposed in this pull request?
Two things to `docs/sql-v2-data-sources.md`, the page on writing a Data
Source V2 connector.
**Fixes three statements that no longer match the API**
- The `TableCatalog` method table lists `createTable(ident, columns,
partitions, properties)`, deprecated since 4.1.0. The current contract is
`createTable(Identifier, TableInfo)`, which is the overload the create, replace
and CTAS execs call.
- The `ViewCatalog` section links `TableViewCatalog`, which does not exist.
The combined interface is `RelationCatalog` (4.2.0), and `Catalogs.load`
rejects a plugin that implements `TableCatalog` and `ViewCatalog` without it.
- The `ScanBuilder` section says each pushdown mix-in "returns the
operators it can handle". `pushPredicates` / `pushFilters` return the
predicates Spark must still evaluate after the scan, and `pushedPredicates()` /
`pushedFilters()` report what was accepted; the limit, top-N, offset and
aggregate mix-ins return a boolean. `SupportsPushDownV2Filters` is listed first
and `SupportsPushDownFilters` described as the pre-3.3 form.
**Adds a "Building a Connector" walkthrough**
The page describes the interfaces one by one and ends there. The new
section puts them together in the order an author needs them, in four steps: a
minimal batch read (provider, table, scan, partitions, reader, and the internal
value types an `InternalRow` holds), optional projection and filter pushdown
with the residual contract, batch writes (capabilities, builder, task writers,
driver commit, and why the staging location has to be reachable from both
executors and driver), and opti [...]
### Why are the changes needed?
The three statements send a connector author down the wrong path (a
deprecated overload, a class that does not exist, pushdown described
backwards). Beyond that, the runtime contracts an author hits first are not
written down anywhere: which `TableCapability` unlocks which operation and that
Spark checks it before calling any builder, `InternalRow` wanting `UTF8String`
rather than `String`, what the residual from `pushPredicates` means, and how
commit messages reach the driver.
### Does this PR introduce _any_ user-facing change?
Docs only.
### How was this patch tested?
The three test connectors the excerpts come from were compiled against the
Spark 4.3.0 distribution on JDK 17 and run under `local-cluster[2,1,1024]`, so
the read, pushdown and write steps were exercised on executors in separate JVMs
from the driver: the pushdown step was checked by inspecting the planned
`Batch` for a few filter and select combinations, and the write step by
confirming that tasks stage under `<path>/_temporary/<queryId>` and the driver
renames from there on commit. D [...]
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code. Examples were compiled and run by hand as
described above.
Closes #59683 from lisancao/SPARK-60437-dsv2-docs.
Authored-by: lnc <[email protected]>
Signed-off-by: Szehon Ho <[email protected]>
(cherry picked from commit e2acd3e6f67bb4c92eee5f532fa17caf0f167028)
Signed-off-by: Szehon Ho <[email protected]>
---
docs/sql-v2-data-sources.md | 325 ++++++++++++++++++++++++++++++++++++++++----
1 file changed, 295 insertions(+), 30 deletions(-)
diff --git a/docs/sql-v2-data-sources.md b/docs/sql-v2-data-sources.md
index 4b4d0f35cb4e..b14b86f81377 100644
--- a/docs/sql-v2-data-sources.md
+++ b/docs/sql-v2-data-sources.md
@@ -55,7 +55,7 @@ There are two ways to plug a data source into Spark:
| Entry Point | Use Case |
|-------------|----------|
-| [`TableProvider`](#tableprovider) | The simpler entry point, for sources
that identify a table by **options** (e.g. a file path or Kafka topic) rather
than through a catalog. Can implement `SupportsCatalogOptions` to also
participate in DDL via the session catalog. |
+| [`TableProvider`](#tableprovider) | The simpler entry point, for sources
that identify a table by **options** (e.g. a file path or Kafka topic) rather
than through a catalog. Can implement `SupportsCatalogOptions` to map reader
and writer options to a table in a catalog. |
| [`CatalogPlugin`](#catalogplugin) | Typically used by external data sources
that manage their own **catalog** of namespaces, tables, and optionally views
and functions (e.g. Iceberg, Delta Lake). Registered via
`spark.sql.catalog.<name>=com.example.MyCatalog`. |
@@ -74,9 +74,9 @@ The key methods are:
A `TableProvider` can also implement
[`SupportsCatalogOptions`](api/java/org/apache/spark/sql/connector/catalog/SupportsCatalogOptions.html)
-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.
+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
@@ -95,6 +95,7 @@ A catalog adds capabilities by mixing in additional
| `StagingTableCatalog` | Atomic create-table-as-select /
replace-table-as-select |
| `SupportsNamespaces` | Create, alter, drop, and list namespaces |
| `ViewCatalog` | List, load, create, replace, rename, and drop views |
+| `RelationCatalog` | Tables and views that share one identifier namespace
(required when a catalog exposes both) |
| `FunctionCatalog` | List and load functions |
| `ProcedureCatalog` | Load and list stored procedures |
@@ -116,16 +117,28 @@ categories of metadata operations through Spark SQL.
|--------|-------------|
| `listTables(namespace)` | List tables in a namespace |
| `loadTable(ident)` | Load a table by identifier |
-| `createTable(ident, columns, partitions, properties)` | Create a new table |
+| `createTable(ident, tableInfo)` | Create a new table from a
[`TableInfo`](api/java/org/apache/spark/sql/connector/catalog/TableInfo.html)
carrying columns, partitioning, properties, and constraints |
| `alterTable(ident, changes...)` | Alter a table's schema, properties, or
constraints |
| `dropTable(ident)` | Drop a table |
+| `renameTable(oldIdent, newIdent)` | Rename a table |
+
+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. A
+[`StagingTableCatalog`](#stagingtablecatalog) is reached through its own
methods instead.
### StagingTableCatalog
[`StagingTableCatalog`](api/java/org/apache/spark/sql/connector/catalog/StagingTableCatalog.html)
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.
+`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 and throw.
### SupportsNamespaces
@@ -145,20 +158,22 @@ adds namespace (database/schema) management:
`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 namespace implement
-[`TableViewCatalog`](api/java/org/apache/spark/sql/connector/catalog/TableViewCatalog.html),
-which extends both `TableCatalog` and `ViewCatalog`. A view's metadata is
represented by
-[`ViewInfo`](#view).
+tables and views in a single shared identifier namespace implement
+[`RelationCatalog`](api/java/org/apache/spark/sql/connector/catalog/RelationCatalog.html)
+(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`](#view).
| Method | Description |
|--------|-------------|
| `listViews(namespace)` | List views in a namespace |
-| `loadView(ident)` | Load a view's `ViewInfo` by identifier |
+| `loadView(ident)` | Load a view's `View` by identifier |
| `viewExists(ident)` | Test whether a view exists (has a default
implementation) |
| `invalidateView(ident)` | Invalidate cached metadata for a view (has a
default implementation) |
| `createView(ident, info)` | Create a new view |
| `replaceView(ident, info)` | Atomically replace an existing view (`ALTER
VIEW ... AS`) |
-| `createOrReplaceView(ident, info)` | Create a view, or atomically replace it
if it exists (`CREATE OR REPLACE VIEW`; has a default implementation) |
+| `createOrReplaceView(ident, info)` | Create a view, or replace it if it
exists (`CREATE OR REPLACE VIEW`); the default calls `replaceView` and falls
back to `createView`, which is not atomic, so override it for an atomic swap |
| `dropView(ident)` | Drop a view |
| `renameView(oldIdent, newIdent)` | Rename a view |
@@ -228,12 +243,12 @@ Additional mix-ins enable further capabilities:
## View
-[`ViewInfo`](api/java/org/apache/spark/sql/connector/catalog/ViewInfo.html) is
the metadata
+[`View`](api/java/org/apache/spark/sql/connector/catalog/View.html) (since
4.2.0) is the metadata
abstraction for a view — a named SQL query managed through a
[`ViewCatalog`](#viewcatalog).
Unlike a [`Table`](#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 `ViewInfo` provides:
+A `View` provides:
| Method | Description |
|--------|-------------|
@@ -273,16 +288,44 @@ 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.
+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`](api/java/org/apache/spark/sql/connector/read/SupportsPushDownV2Filters.html):
+ `pushPredicates(predicates)` receives the candidate
[`Predicate`](#expressions) 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`](api/java/org/apache/spark/sql/connector/read/SupportsPushDownFilters.html)
+ 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`](api/java/org/apache/spark/sql/connector/read/SupportsPushDownRequiredColumns.html):
+ `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`](api/java/org/apache/spark/sql/connector/read/SupportsPushDownLimit.html)`.pushLimit`,
+
[`SupportsPushDownTopN`](api/java/org/apache/spark/sql/connector/read/SupportsPushDownTopN.html)`.pushTopN`,
+
[`SupportsPushDownOffset`](api/java/org/apache/spark/sql/connector/read/SupportsPushDownOffset.html)`.pushOffset`
+ and
+
[`SupportsPushDownAggregates`](api/java/org/apache/spark/sql/connector/read/SupportsPushDownAggregates.html)`.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
@@ -378,10 +421,12 @@ 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).
+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.
@@ -408,7 +453,7 @@ predicate are removed without rewriting individual records.
| Method | Description |
|--------|-------------|
-| `canDeleteWhere(predicates)` | Returns whether the data source can
efficiently delete rows matching the given predicates. If this returns `false`,
Spark falls back to row-level rewriting (see below) |
+| `canDeleteWhere(predicates)` | Returns whether the data source can
efficiently delete rows matching the given predicates. If this returns `false`,
Spark falls back to row-level rewriting ([Row-Level
Operations](#row-level-operations)) when the table also implements
`SupportsRowLevelOperations`; otherwise the `DELETE` fails |
| `deleteWhere(predicates)` | Deletes all rows matching the predicates |
This approach is well-suited for data sources that can drop entire partitions
or files without
@@ -444,7 +489,8 @@ Data sources fall into two categories:
The `org.apache.spark.sql.connector.expressions` package provides a neutral
expression
representation used across the DSV2 API:
-- **Transforms** (`IdentityTransform`, `BucketTransform`, `YearsTransform`,
etc.)
+- **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.
@@ -466,6 +512,225 @@ The same `Table` and `Scan` abstractions support
streaming queries. A `Table` th
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`](https://github.com/apache/spark/blob/master/sql/core/src/test/java/test/org/apache/spark/sql/connector/JavaSimpleDataSourceV2.java)
+for a minimal read,
+[`JavaAdvancedDataSourceV2WithV2Filter`](https://github.com/apache/spark/blob/master/sql/core/src/test/java/test/org/apache/spark/sql/connector/JavaAdvancedDataSourceV2WithV2Filter.java)
+for pushdown, and
+[`JavaSimpleWritableDataSource`](https://github.com/apache/spark/blob/master/sql/core/src/test/java/test/org/apache/spark/sql/connector/JavaSimpleWritableDataSource.java)
+for writes, exercised by
+[`DataSourceV2Suite`](https://github.com/apache/spark/blob/master/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala).
+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:
+
+{% highlight java %}
+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();
+ }
+ };
+ }
+}
+{% endhighlight %}
+
+`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:
+
+{% highlight java %}
+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() {
+ }
+};
+{% endhighlight %}
+
+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](#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:
+
+{% highlight java %}
+@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;
+}
+{% endhighlight %}
+
+`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`](https://github.com/apache/spark/blob/master/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala)
+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`](api/java/org/apache/spark/sql/connector/catalog/TableCapability.html)
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](#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`](https://github.com/apache/spark/blob/master/sql/core/src/test/java/test/org/apache/spark/sql/connector/JavaSimpleWritableDataSource.java)),
+and commits by renaming the staged files into the table directory one at a time
+([`MyBatchWrite.commit`](https://github.com/apache/spark/blob/master/sql/core/src/test/java/test/org/apache/spark/sql/connector/JavaSimpleWritableDataSource.java)).
+`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`](api/java/org/apache/spark/sql/connector/write/WriterCommitMessage.html)
+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`](#tablecatalog), whose
`loadTable` returns
+the same `Table` class the `TableProvider` builds, and register it as under
+[CatalogPlugin](#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)`](api/java/org/apache/spark/sql/connector/catalog/Column.html).
+`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)](api/java/org/apache/spark/sql/connector/package-summary.html)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
