HyukjinKwon opened a new pull request, #59313:
URL: https://github.com/apache/spark/pull/59313
### What changes were proposed in this pull request?
This PR adds the columnar data source API: a simpler way to plug a data
source into Spark than implementing the Data Source V2 interfaces directly. It
follows the Python Data Source API, but data is exchanged only as columnar
batches. A data source can be implemented on the JVM, or in native code such as
Rust or C++ without any JVM code.
**Public API** (`org.apache.spark.sql.datasource`, Java, `@Evolving`, since
4.4.0):
| Class | Description |
|-------|-------------|
| `DataSource` | `schema`, `reader`, `streamReader`, `writer`,
`streamWriter`. Only the supported operations need to be implemented. |
| `DataSourceReader` | `pushPredicates` (V2 predicates), `pushLimit`,
`pruneColumns`, `partitions`, `read` returning
`PartitionReader<ColumnarBatch>`. |
| `DataSourceStreamReader` | `initialOffset`, `latestOffset`,
`partitions(start, end)`, `read`, `commit`, `stop`, with JSON offsets. |
| `DataSourceWriter`, `DataSourceStreamWriter` | `createWriter` returning
`DataWriter<ColumnarBatch>`, `commit`, `abort`. The batches are backed by Arrow
(`ArrowColumnVector`). |
| `DataSourceProvider` | The entry point of a JVM data source: a
`TableProvider` and `DataSourceRegister` that creates the `DataSource`. |
| `NativeBridge` | The binary interface of native data sources. A shared
library implements its `native` methods as JNI functions, and exchanges schemas
and data through the Arrow C data interface. Spark loads each library into its
own copy of the class, with a separate class loader, so any number of libraries
exporting the same functions can be loaded side by side. Spark itself ships no
native code. |
**Native data sources**:
- A library is distributed as a native data source package: a zip file with
the extension `.sparkpkg`, containing a manifest
(`spark-native-datasource.json`: ABI version, data source names, and the
library for each platform such as `linux-x86_64` or `osx-aarch64`).
- Packages are added to a session with `spark.addArtifact` (Scala and
Python, classic and Spark Connect: `.sparkpkg` files are added as file
artifacts), or found under `spark.sql.dataSource.native.paths`, in which case
Spark adds them to the session's artifacts. Executors get the packages through
the artifacts of the session.
- `DataSource.lookupDataSource` falls back to native data sources after Java
and Python ones. A small `NamedTableProvider` trait generalizes how
`PythonDataSourceV2` gets its short name, so that the native provider shares
the same three call sites.
- A library only exports the functions of the operations it supports:
missing functions are reported as `DATA_SOURCE_*_NOT_SUPPORTED`, except for the
optional pushdown and serialization functions.
**Internals**:
- `execution.datasources.v2.columnar`: adapts the API to Data Source V2, for
batch and micro-batch reads, batch and streaming writes (append,
overwrite/complete), and predicate, column (top-level) and limit pushdown.
Writes convert rows to Arrow batches of at most
`spark.sql.execution.arrow.maxRecordsPerBatch` rows.
- `execution.datasources.v2.ffi`: package parsing and discovery, library
loading, the implementation of the API on top of `NativeBridge` with
`arrow-c-data`, and the JSON encoding of pushed predicates.
**Other changes**:
- Configurations: `spark.sql.dataSource.native.enabled` (static, default
`true`) and `spark.sql.dataSource.native.paths`.
- Error conditions: `NATIVE_DATA_SOURCE_ERROR`,
`INVALID_NATIVE_DATA_SOURCE_PACKAGE` (`INVALID_MANIFEST`, `INVALID_LIBRARY`,
`UNSUPPORTED_ABI_VERSION`, `UNSUPPORTED_PLATFORM`),
`NATIVE_DATA_SOURCE_PACKAGE_CONFLICT` and
`NATIVE_DATA_SOURCE_PACKAGE_NOT_FOUND`.
- New dependency: `org.apache.arrow:arrow-c-data`, with the same version as
`arrow-vector`. It bundles its own JNI library for Linux, macOS and Windows.
- Documentation: a new page, "Columnar Data Source API", with complete
examples in Scala, Rust (`jni` and arrow-rs) and C++ (Arrow C++), linked from
the Data Source V2 page.
For example, with the Rust example of the documentation:
```python
spark.addArtifact("/path/to/rust_range.sparkpkg")
spark.read.format("rust_range").option("end", 100).load().show()
```
### Why are the changes needed?
Many data systems have their best client libraries in native code, such as
arrow-rs, DataFusion, delta-rs, iceberg-rust and Lance in Rust, or C++
libraries. Using them from Spark today requires writing a JVM Data Source V2
connector with its own JNI glue. With this API, a native data source implements
a stable binary interface once, is distributed as a single file, and works with
every Spark API, including Spark Connect clients in any language, without any
JVM code. The data is exchanged through the Arrow C data interface without
copying when reading.
The same API also gives Java and Scala developers a much simpler alternative
to Data Source V2, comparable to the Python Data Source API.
### Does this PR introduce _any_ user-facing change?
Yes. It adds the APIs, configurations and error conditions above, and
`spark.addArtifact` now accepts native data source packages (`.sparkpkg` files).
### How was this patch tested?
- New `ColumnarDataSourceSuite`, with JVM data sources: batch reads,
predicate, column and limit pushdown, batch writes in Arrow batches (append and
overwrite), aborted writes, streaming reads and writes, and the errors of
unsupported operations.
- New `NativeDataSourceSuite`, with a dependency-free C++ test library
(`sql/core/src/test/resources/native-datasource/test_native_datasource.cc`)
compiled by the suite with the C++ compiler of the machine; the tests that load
it are skipped if there is no compiler. It covers inferred and user-specified
schemas, partitions and batches, all the types of the test library, predicate,
column and limit pushdown, batch writes, aborted writes, streaming reads and
writes over several micro-batches with offset commits, errors reported by
native code, missing functions, several libraries exporting the same functions
side by side, `spark.addArtifact`, packages in directories, package conflicts,
the precedence of Java data sources, and invalid packages (manifest, ABI
version, platform and library). It also has unit tests for the manifest, the
platform names and the JSON encoding of predicates.
- New PySpark tests for adding `.sparkpkg` artifacts in classic mode and
with Spark Connect.
- The Rust and C++ examples of the documentation were built and run with
Spark, to check that they work as documented.
- Existing suites: `ResolvedDataSourceSuite`, `ArtifactManagerSuite`,
`PythonDataSourceSuite`, `SparkThrowableSuite` and
`SparkConfigBindingPolicySuite`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5.5)
This pull request and its description were written by Isaac.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
