HyukjinKwon opened a new pull request, #59424: URL: https://github.com/apache/spark/pull/59424
### What changes were proposed in this pull request? This is the FFM version of #59313, as a draft for comparison: the same feature, with the binary interface as a plain C ABI called through the [Foreign Function and Memory (FFM) API](https://openjdk.org/jeps/454) instead of JNI. It includes all the changes of #59313, up to ff7b1ae, and only changes how Spark calls the libraries. Native data sources then require Java 22 or later, where the FFM API is final; with Java 17 and 21, Spark compiles and runs as before, and a query that uses a native data source fails with `NATIVE_DATA_SOURCE_UNSUPPORTED_JAVA_VERSION`. Like #59313, this PR adds native data sources: data sources implemented in native code, such as Rust or C++, without any JVM code. A native data source is a shared library that implements a binary interface. Spark finds it automatically when it is installed like other native libraries, or in a package file added to a session, and loads it when a query uses it. Spark adapts it to Data Source V2, with batch and micro-batch streaming reads, batch and streaming writes, and predicate, column and limit pushdown. Data is exchanged only as columnar batches, through the Arrow C data interface. **Differences from #59313**: | | #59313 (JNI) | This PR (FFM) | |---|---|---| | Interface | `org.apache.spark.sql.datasource.NativeBridge`, a Java class whose `native` methods the library implements | `spark_datasource.h`, a C header (`sql/core/src/main/resources/org/apache/spark/sql/datasource/`, also in the `spark-sql` JAR) | | Function names | `Java_org_apache_spark_sql_datasource_NativeBridge_createDataSource`, ... | `spark_ds_create`, ... | | What a library needs | `jni.h`, or the `jni` crate | nothing but the header, or `extern "C"` declarations | | Errors | throw a Java exception (`ThrowNew`) | return non-zero, with the message from `spark_ds_last_error` | | Returned byte arrays (partitions, state, offsets, commit messages) | Java `byte[]`, `byte[][]`, `String` built by the library | passed to a `SparkDsSink` callback, which copies them | | Side-by-side libraries | a copy of `NativeBridge` per library, in its own class loader | each library looked up separately (`SymbolLookup.libraryLookup`) | | Java versions | 17, 21, 25 | 22 and later (25 among the supported LTS versions); a clear error otherwise | **Public API**: a single C header, `spark_datasource.h`, whose comments are the specification of the interface, with the same content as the Javadoc of `NativeBridge` in #59313: the lifecycle, the handles, the ownership of the Arrow structs and of the memory that Spark passes, errors (non-zero return values, with the message of `spark_ds_last_error`, reported as `NATIVE_DATA_SOURCE_ERROR`), the JSON format of the pushed predicates, the Spark SQL semantics that a predicate evaluated by the library must follow, the threads, and the compatibility rules. Version 1 of the interface, `SPARK_DS_ABI_VERSION`. The `org.apache.spark.sql.datasource.NativeBridge` class of #59313 is removed: there is no JVM API for native data sources. | Operation | Required functions | Optional functions | |-----------|--------------------|--------------------| | All | `spark_ds_abi_version`, `spark_ds_create`, `spark_ds_close` | `spark_ds_last_error`, `spark_ds_schema` | | Batch read | `spark_ds_create_reader`, `spark_ds_partitions`, `spark_ds_read`, `spark_ds_close_reader` | `spark_ds_push_predicates`, `spark_ds_push_limit`, `spark_ds_prune_columns`, `spark_ds_serialize_reader` | | Streaming read | `spark_ds_create_stream_reader`, `spark_ds_initial_offset`, `spark_ds_latest_offset`, `spark_ds_stream_partitions`, `spark_ds_read`, `spark_ds_close_stream_reader` | `spark_ds_serialize_stream_reader`, `spark_ds_commit_offset` | | Batch write | `spark_ds_create_writer`, `spark_ds_commit`, `spark_ds_abort`, `spark_ds_close_writer`, `spark_ds_create_data_writer`, `spark_ds_write`, `spark_ds_commit_data_writer`, `spark_ds_abort_data_writer` | `spark_ds_serialize_writer` | | Streaming write | `spark_ds_create_stream_writer`, `spark_ds_commit`, `spark_ds_abort`, `spark_ds_close_writer`, `spark_ds_create_data_writer`, `spark_ds_write`, `spark_ds_commit_data_writer`, `spark_ds_abort_data_writer` | `spark_ds_serialize_writer` | A library only exports the functions of the operations it supports. Spark reports a missing required function as `DATA_SOURCE_*_NOT_SUPPORTED`. **How Spark calls the libraries** (`execution.datasources.v2.ffi.NativeLibrary`): Spark still compiles for Java 17, where the FFM API does not exist, so `NativeLibrary` reaches `java.lang.foreign` only through method handles, looked up once when the first library is loaded, and only on Java 22 or later. Each function of a library is bound once to a downcall handle adapted to take and return primitive values only, with pointers as `long` addresses, so each call is a plain `invokeExact` without reflection, which the JIT compiles like a direct FFM call; a call to `spark_ds_partitions`, including the upcalls to the sink and the allocation of its arguments, takes about 175 ns on JDK 26. The memory that Spark passes to a function is allocated with `Platform` and freed after the call. A single upcall stub, `SparkDsSink`, copies the byte arrays that the library returns, and never throws, since an exception thrown by an upcall would crash the JVM. Spark already runs with `--enable-native-a ccess=ALL-UNNAMED`, which the restricted methods of the FFM API need to avoid warnings; a library that cannot be loaded, including when native access is denied, is reported as `INVALID_NATIVE_DATA_SOURCE_LIBRARY.CANNOT_LOAD` or `INVALID_NATIVE_DATA_SOURCE_PACKAGE.INVALID_LIBRARY`, as before. Everything above `NativeLibrary` is the same as in #59313: the Scala classes keep calling it with the same methods, and only catch `NativeLibrary.MissingFunctionException` and `NativeLibrary.LoadException` instead of `UnsatisfiedLinkError`. Spark still uses `arrow-c-data` to import the Arrow C structs, and `arrow-c-data` calls their `get_next` and `release` callbacks through the JNI library that it bundles, as in #59313. With FFM, Spark could call those callbacks itself and drop that library, which is left for later. **Finding native data sources**: when no Java or Python data source has the requested name, Spark looks for a native data source with that name, in two places: - **Installed libraries, automatically**: like the Python data sources installed in the Python path, a library installed like other native libraries is found without any configuration, by its file name. The library of the data source `rust_range` is `spark_datasource_rust_range`: `libspark_datasource_rust_range.so` on Linux, `libspark_datasource_rust_range.dylib` on macOS and `spark_datasource_rust_range.dll` on Windows. Spark checks whether the file exists in the native library path, which consists of: 1. the directories of `java.library.path`, where the JVM itself finds native libraries. They include `LD_LIBRARY_PATH` on Linux, `DYLD_LIBRARY_PATH` on macOS and `PATH` on Windows, and so `spark.driver.extraLibraryPath` and `spark.executor.extraLibraryPath`; 2. the `lib` directory of each installation prefix in `PATH`: `<prefix>/lib` for each `<prefix>/bin`. So libraries are found where native libraries are usually installed: `/usr/local/lib` (`make install`, CMake, cargo-c, Homebrew on Intel macs), `/opt/homebrew/lib`, `$CONDA_PREFIX/lib` and `$VIRTUAL_ENV/lib` in an activated environment, and `/usr/lib`. A library that implements several data sources is installed under each of their names, for example with symbolic links. The executors load the library from the same path as the driver, or else find it in their own native library path, and only load the same build as the driver, which they check with the SHA-256 checksum of the file. - **Packages of a session**: a native data source package is a zip file with the extension `.sparkpkg`, with a manifest, `spark-native-datasource.json`, that lists the ABI version, the names of the data sources, and the library for each platform such as `linux-x86_64` or `osx-aarch64`. A session adds packages with `spark.addArtifact` (Scala and Python, classic and Spark Connect: `.sparkpkg` files are added as file artifacts), or with `spark.sql.dataSource.native.paths`. They take precedence over the installed libraries, and the executors get them through the artifacts of the session, except in local mode, where they read them in place: a configured package is copied there under a name with its checksum, so it can be replaced in place. Spark ignores, and logs, a package that it cannot read or whose ABI version it does not support, unless its manifest lists the requested data source and no other package or installed library provides it. **Short names**: no JVM class is needed for each data source. After the Java data sources (`DataSourceRegister`) and the Python ones, `DataSource.lookupDataSource` looks the name up in the packages of the session and in the native library path, case-insensitively, and resolves it to a single provider class, `NativeDataSourceV2`. The caller then gives it the requested name, the way `PythonDataSourceV2` already works: a small `NamedTableProvider` trait generalizes how `PythonDataSourceV2` gets its short name, so the native provider shares the same call sites. The provider reuses the result of the lookup that found its class, instead of looking the name up again. A Python data source can still be registered with the name of a native data source, and takes precedence over it. Lookups only read manifests and check whether files exist; Spark loads a library when a query uses it. Native data sources work with `spark.read`/`write`, `readStream`/`writeStream`, and SQL tables (`CREATE TABLE ... USING rust_range OPTIONS (...)`, `SELECT` and `INSERT INTO`). **Other changes**, besides those of #59313: - Error condition: `NATIVE_DATA_SOURCE_UNSUPPORTED_JAVA_VERSION` (`0A000`), raised when a query uses a native data source on Java 21 or earlier. The lookups still find the data source, so the error says why it cannot be used rather than that it does not exist. - The test library `test_native_datasource.cc` implements the C interface, and the tests compile it with `spark_datasource.h` instead of the JNI headers. - The `CheckConnectJvmClientCompatibility` exclusion of `org.apache.spark.sql.datasource`, added by #59313 for `NativeBridge`, is not needed. The other changes are those of #59313: the configurations `spark.sql.dataSource.native.enabled` and `spark.sql.dataSource.native.paths`, the error conditions, the `arrow-c-data` dependency, the "Native Data Sources" documentation page (with the examples below), and: - Existing classes: - `DataSource.lookupDataSource` looks up the native data sources after the Python ones, and `DataSourceRegistration` skips them when it registers a Python data source. - `ArtifactManager` lists the native data source packages of a session, and forgets the cached packages read from the artifacts of a session when it cleans them up. - The classic `DataFrameWriter` builds its commands, and `DataStreamWriter` resolves its sink, with the session of the Dataset active, as the analysis of a query does: the lookups of the Python and native data sources use the active session. - PySpark `addArtifacts` adds `.sparkpkg` files to the session whatever the flags: in classic mode through the JVM session, and with Spark Connect as file artifacts. ### Examples Both examples implement a read-only data source that returns the numbers `[0, end)` as a column `id`, in two partitions, with the option `end` (10 by default). They only export the functions needed for batch reads, plus `spark_ds_last_error`. <details> <summary><b>Rust</b> (with arrow-rs only)</summary> `Cargo.toml`: ```toml [package] name = "rust_range" version = "0.1.0" edition = "2021" [lib] name = "spark_datasource_rust_range" crate-type = ["cdylib"] [dependencies] arrow-array = { version = "58", features = ["ffi"] } arrow-schema = { version = "58", features = ["ffi"] } ``` `src/lib.rs`: ```rust //! A native data source for Apache Spark, written in Rust. It reads the numbers [0, end) as a //! column `id`, in two partitions. The option `end` defaults to 10. use std::cell::RefCell; use std::ffi::{c_char, c_int, c_void, CStr, CString}; use std::panic::{catch_unwind, AssertUnwindSafe}; use std::sync::Arc; use arrow_array::ffi_stream::FFI_ArrowArrayStream; use arrow_array::{Int64Array, RecordBatch, RecordBatchIterator}; use arrow_schema::ffi::FFI_ArrowSchema; use arrow_schema::{DataType, Field, Schema, SchemaRef}; type Result<T> = std::result::Result<T, Box<dyn std::error::Error>>; /// `SparkDsSink`: receives a byte array returned to Spark, and copies it. type Sink = extern "C" fn(sink: *mut c_void, data: *const u8, length: i64); struct RangeSource { end: i64, } struct RangeReader { end: i64, } thread_local! { static LAST_ERROR: RefCell<CString> = RefCell::new(CString::default()); } fn schema() -> SchemaRef { Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)])) } /// Runs `f`, and reports its error or panic to Spark: neither may cross into the JVM. fn call(f: impl FnOnce() -> Result<()>) -> c_int { let message = match catch_unwind(AssertUnwindSafe(f)) { Ok(Ok(())) => return 0, Ok(Err(e)) => e.to_string(), Err(_) => "the data source panicked".to_string(), }; LAST_ERROR.with(|error| *error.borrow_mut() = CString::new(message).unwrap_or_default()); 1 } #[no_mangle] pub extern "C" fn spark_ds_last_error() -> *const c_char { LAST_ERROR.with(|error| error.borrow().as_ptr()) } #[no_mangle] pub extern "C" fn spark_ds_abi_version() -> i32 { 1 } #[no_mangle] pub unsafe extern "C" fn spark_ds_create( _name: *const c_char, option_keys: *const *const c_char, option_values: *const *const c_char, num_options: i64, source: *mut *mut c_void, ) -> c_int { call(|| { let mut range = RangeSource { end: 10 }; for i in 0..num_options as usize { if CStr::from_ptr(*option_keys.add(i)).to_str()? == "end" { range.end = CStr::from_ptr(*option_values.add(i)).to_str()?.parse()?; } } *source = Box::into_raw(Box::new(range)) as *mut c_void; Ok(()) }) } #[no_mangle] pub unsafe extern "C" fn spark_ds_schema(_source: *mut c_void, out: *mut FFI_ArrowSchema) -> c_int { call(|| { std::ptr::write(out, FFI_ArrowSchema::try_from(schema().as_ref())?); Ok(()) }) } #[no_mangle] pub unsafe extern "C" fn spark_ds_create_reader( source: *mut c_void, _schema: *mut FFI_ArrowSchema, reader: *mut *mut c_void, ) -> c_int { call(|| { let end = (*(source as *const RangeSource)).end; *reader = Box::into_raw(Box::new(RangeReader { end })) as *mut c_void; Ok(()) }) } #[no_mangle] pub unsafe extern "C" fn spark_ds_close(source: *mut c_void) { drop(Box::from_raw(source as *mut RangeSource)); } #[no_mangle] pub unsafe extern "C" fn spark_ds_partitions( reader: *mut c_void, add_partition: Sink, sink: *mut c_void, ) -> c_int { call(|| { let end = (*(reader as *const RangeReader)).end; for (start, end) in [(0, end / 2), (end / 2, end)] { let bytes = [start.to_le_bytes(), end.to_le_bytes()].concat(); add_partition(sink, bytes.as_ptr(), bytes.len() as i64); } Ok(()) }) } #[no_mangle] pub unsafe extern "C" fn spark_ds_close_reader(reader: *mut c_void) { drop(Box::from_raw(reader as *mut RangeReader)); } #[no_mangle] pub unsafe extern "C" fn spark_ds_read( _reader_state: *const u8, _reader_state_length: i64, partition: *const u8, partition_length: i64, out: *mut FFI_ArrowArrayStream, ) -> c_int { call(|| { let bytes = std::slice::from_raw_parts(partition, partition_length as usize); let start = i64::from_le_bytes(bytes[0..8].try_into()?); let end = i64::from_le_bytes(bytes[8..16].try_into()?); let ids = Int64Array::from_iter_values(start..end); let batch = RecordBatch::try_new(schema(), vec![Arc::new(ids)])?; let batches = RecordBatchIterator::new(vec![Ok(batch)], schema()); std::ptr::write(out, FFI_ArrowArrayStream::new(Box::new(batches))); Ok(()) }) } ``` Build and install, here on macOS: ```bash cargo build --release cp target/release/libspark_datasource_rust_range.dylib /usr/local/lib/ ``` Or package, to add it to a session with `spark.addArtifact`: ```bash mkdir -p package/osx-aarch64 cp target/release/libspark_datasource_rust_range.dylib package/osx-aarch64/ cat > package/spark-native-datasource.json <<EOF { "abiVersion": 1, "dataSources": ["rust_range"], "libraries": {"osx-aarch64": "osx-aarch64/libspark_datasource_rust_range.dylib"} } EOF (cd package && zip -r ../rust_range.sparkpkg .) ``` </details> <details> <summary><b>C++</b> (with Arrow C++)</summary> `cpp_range.cc`: ```cpp // A native data source for Apache Spark, written in C++ with Arrow C++. It reads the numbers // [0, end) as a column `id`, in two partitions. The option `end` defaults to 10. #include <cstdint> #include <cstring> #include <memory> #include <stdexcept> #include <string> #include <arrow/api.h> #include <arrow/c/bridge.h> #include "spark_datasource.h" namespace { struct RangeSource { int64_t end = 10; }; struct RangeReader { int64_t end; }; thread_local std::string last_error; std::shared_ptr<arrow::Schema> Schema() { return arrow::schema({arrow::field("id", arrow::int64(), /*nullable=*/false)}); } // Runs `f`, and reports its error to Spark: no C++ exception may cross into the JVM. template <typename F> int Call(F&& f) { try { f(); return 0; } catch (const std::exception& e) { last_error = e.what(); return 1; } catch (...) { last_error = "unknown error"; return 1; } } void Check(const arrow::Status& status) { if (!status.ok()) throw std::runtime_error(status.ToString()); } } // namespace extern "C" { const char* spark_ds_last_error(void) { return last_error.c_str(); } int32_t spark_ds_abi_version(void) { return SPARK_DS_ABI_VERSION; } int spark_ds_create(const char*, const char* const* option_keys, const char* const* option_values, int64_t num_options, void** source) { return Call([&] { auto range = std::make_unique<RangeSource>(); for (int64_t i = 0; i < num_options; i++) { if (std::string(option_keys[i]) == "end") range->end = std::stoll(option_values[i]); } *source = range.release(); }); } int spark_ds_schema(void*, struct ArrowSchema* out) { return Call([&] { Check(arrow::ExportSchema(*Schema(), out)); }); } int spark_ds_create_reader(void* source, struct ArrowSchema*, void** reader) { return Call([&] { *reader = new RangeReader{static_cast<RangeSource*>(source)->end}; }); } void spark_ds_close(void* source) { delete static_cast<RangeSource*>(source); } int spark_ds_partitions(void* reader, SparkDsSink add_partition, void* sink) { return Call([&] { int64_t end = static_cast<RangeReader*>(reader)->end; int64_t bounds[2][2] = {{0, end / 2}, {end / 2, end}}; for (const auto& partition : bounds) { add_partition(sink, reinterpret_cast<const uint8_t*>(partition), sizeof(partition)); } }); } void spark_ds_close_reader(void* reader) { delete static_cast<RangeReader*>(reader); } int spark_ds_read(const uint8_t*, int64_t, const uint8_t* partition, int64_t, struct ArrowArrayStream* out) { return Call([&] { int64_t bounds[2]; std::memcpy(bounds, partition, sizeof(bounds)); arrow::Int64Builder ids; for (int64_t id = bounds[0]; id < bounds[1]; id++) Check(ids.Append(id)); std::shared_ptr<arrow::Array> array; Check(ids.Finish(&array)); auto batch = arrow::RecordBatch::Make(Schema(), array->length(), {array}); auto batches = arrow::RecordBatchReader::Make({batch}, Schema()).ValueOrDie(); Check(arrow::ExportRecordBatchReader(batches, out)); }); } } // extern "C" ``` Get `spark_datasource.h`, which is in the `spark-sql` JAR, then build (Arrow C++ 24 requires C++20) and install, or package as above: ```bash unzip -j "$SPARK_HOME"/jars/spark-sql_*.jar org/apache/spark/sql/datasource/spark_datasource.h c++ -std=c++20 -O2 -shared -fPIC -I. -I"$ARROW_HOME/include" -L"$ARROW_HOME/lib" -larrow \ -o libspark_datasource_cpp_range.dylib cpp_range.cc cp libspark_datasource_cpp_range.dylib /usr/local/lib/ ``` </details> Usage is the same as in #59313, on Java 22 or later, in classic PySpark or with Spark Connect. Once installed, the data sources are found automatically, without `spark.addArtifact` or any configuration: ```python >>> spark.read.format("rust_range").option("end", 5).load().show() +---+ | id| +---+ | 0| | 1| | 2| | 3| | 4| +---+ >>> spark.read.format("cpp_range").option("end", 5).load().filter("id >= 3").collect() [Row(id=3), Row(id=4)] >>> spark.sql("CREATE TABLE t USING rust_range OPTIONS (end 3)") >>> spark.sql("SELECT * FROM t").collect() [Row(id=0), Row(id=1), Row(id=2)] ``` Or added to a session as a package: ```python >>> spark.addArtifact("rust_range/rust_range.sparkpkg") >>> spark.read.format("rust_range").option("end", 5).load().count() 5 ``` ### 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 change, a native data source implements a stable binary interface once, is installed like any other native library or distributed as a single file, and works with every Spark API, including Spark Connect clients in any language, without any JVM code. With FFM, the interface is also a plain C ABI, so a library needs no JVM headers or bindings, and other hosts than Spark could load the same libraries. ### Does this PR introduce _any_ user-facing change? Yes. It adds native data sources, with the `spark_datasource.h` interface, the configurations and the error conditions above, on Java 22 or later. Spark finds the native data source libraries installed in the native library path, and `spark.addArtifact` now accepts native data source packages (`.sparkpkg` files). Also, in classic mode, writing with a session that is not the active one now finds the data sources of that session, such as its Python data sources, because `DataFrameWriter` and `DataStreamWriter` resolve the data source with that session active. ### How was this patch tested? All the tests of #59313, ported to the C interface: - `NativeDataSourceSuite`, with the dependency-free C++ test library (`sql/core/src/test/resources/native-datasource/test_native_datasource.cc`), now implementing `spark_datasource.h`, which the tests compile with the C++ compiler of the machine against the header from the classpath. It covers inferred and user-specified schemas, partitions and batches, the supported types, predicate, column and limit pushdown, batch writes (append and overwrite), aborted writes, streaming reads and writes over several micro-batches with offset commits, SQL tables with `OPTIONS` (`SELECT` and `INSERT INTO`), 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, invalid packages, installed libraries and invalid installed libraries, plus the unit tests for the manifest, the platform names, the native library path and the JSON encoding of pred icates. - Tests added during the review, in `NativeDataSourceSuite` unless noted: - Packages: invalid and unsupported packages are ignored unless no other package or installed library provides the data source; a configured package replaced in place; the copies distributed to the executors are named by checksum and are not packages of the session; the cached packages of a session are forgotten when its artifacts are cleaned up; configured directories that cannot be read; distributing a package without an active session. - Lookups: a single lookup for each resolution, which the next lookup on the thread clears; a Python data source registered with the name of an installed native one; batch and streaming writes with a session that is not the active one. - Writes: a failed `commitDataWriter` aborts the data writer; a failed `commit` aborts the write, then closes the writer; the test library records when a writer is closed, which checks that a batch write closes its writer, and that each micro-batch of a streaming write closes the writer of the previous one; `ColumnarDataSourceSuite` checks that each micro-batch creates its stream writer with the ID of the query, and that batches end at `spark.sql.execution.arrow.maxBytesPerBatch`. - Cluster mode: new `NativeDataSourceClusterSuite` runs in `local-cluster[2,1,1024]`, where the executors are separate JVMs: batch reads, writes and streaming with a package added with `spark.addArtifact`, and a configured package that the executors can only load from its distributed copy (named by checksum, under the UUID of the session), which they check by its checksum. - Installed libraries: an executor does not load a library with another checksum than on the driver, at the driver's path or in its native library path. - Predicates: only the predicates listed in `spark_datasource.h` are pushed down, not arithmetic or other functions. - `spark.addArtifact` in classic PySpark accepts `file://` URIs and paths relative to the working directory of Python. - Reads: predicates on columns with collated strings are not pushed down, read schemas with collations, and the scan described by the name of the data source; sliced arrays, arrays shorter than their parent (a column, a struct field or the elements of a map), `decimal256` and list views are rejected with the name of the column, maps whose entry fields are named `keys` and `values` (as arrow-rs names them) are read, and `limit(...).offset(...)` asks the library for the limit once. - Writes whose upstream stages fail with adaptive execution create no writer (also in `ColumnarDataSourceSuite`). - `NativeDataSourceConnectSuite` (Spark Connect), `NativeDataSourceClusterSuite` (`local-cluster`, executors in separate JVMs) and `ColumnarDataSourceSuite`. New tests: - `native data sources require Java 22 or later`: on Java 21 or earlier, a query that uses an installed native data source fails with `NATIVE_DATA_SOURCE_UNSUPPORTED_JAVA_VERSION`, before Spark loads the library. It runs only on Java 21 or earlier, so it runs in the default CI on Java 17. - `report the errors of native code`: a library without `spark_ds_last_error` reports `spark_ds_create failed with the error code 1.` The tests that load a library are skipped before Java 22, like they are skipped without a C++ compiler. So the default CI, on Java 17, only runs the tests that do not load a library, and the scheduled Java 25 builds run them all. Locally on macOS (osx-aarch64): - Java 26 (Homebrew OpenJDK 26.0.1; Spark built with Java 17): `NativeDataSourceSuite`, `NativeDataSourceClusterSuite` (with the executors on Java 26 too), `ColumnarDataSourceSuite` and `NativeDataSourceConnectSuite` pass. With the executors of `local-cluster` on Java 17, the tasks fail with `NATIVE_DATA_SOURCE_UNSUPPORTED_JAVA_VERSION`, as expected. - Java 17: `NativeDataSourceSuite` passes, with the tests that load a library skipped, and the Java version test run. - The Rust and C++ examples above were built with the commands above (Rust with arrow-rs 58, C++ with Arrow C++ 24 and `-std=c++20`), installed in a directory of `java.library.path`, and read by Spark on Java 26: `spark.read.format("rust_range").option("end", 100).load()` and the same with `cpp_range` return the 100 rows in two partitions, and `option("end", "x")` fails with `NATIVE_DATA_SOURCE_ERROR` and the message of the library (`invalid digit found in string` and `stoll: no conversion`). - The documented steps of the C++ example also work from a packaged JAR: `unzip -j` of `spark_datasource.h` from the `spark-sql` JAR built by `sbt sql/package`, then the documented `c++` command with `-I.`. - The C++ test library builds without warnings with Apple clang (`-Wall -Wextra`), with and without `READ_ONLY`. - `SparkThrowableSuite`, for the new error condition. - `checkstyle` and `scalastyle` of the changed modules. ### 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]
