[SPARK-60101][SQL] Support native data sources implemented in Rust or C++ with the FFM API

[HyukjinKwon]

What changes were proposed in this pull request?

This is the FFM version of #59313, which it replaces, following the review of #59313: the same feature, with the binary interface as a plain C ABI called through the Foreign Function and Memory (FFM) API instead of JNI. It includes all the changes of #59313, up to e3ff80c, changes how Spark calls the libraries, and, following the review of this PR, gives the interface a shape that can grow without new exported functions: a function table, structs that Spark passes to the library, and capabilities for each data source (see Shape of the interface below). 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)
What a library exports	Java_org_apache_spark_sql_datasource_NativeBridge_createDataSource, ... (29 functions)	spark_ds_init, which returns a table of function pointers, SparkDsLibrary
What a library needs	jni.h, or the jni crate	nothing but the header, or #[repr(C)] declarations of its structs
Errors	throw a Java exception (ThrowNew)	return non-zero, with the message from the last_error function of the table
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 has its own table (SymbolLookup.libraryLookup finds spark_ds_init in each)
Supported operations	the JNI functions that the library implements	the capabilities that create returns for each data source
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 last_error, reported as NATIVE_DATA_SOURCE_ERROR), the JSON format of the pushed predicates, which Spark evaluates again after reading, so that a library only uses them to skip data and must never skip a row for which they are true under the semantics of Spark SQL (the jar of Spark SQL contains test cases for libraries, org/apache/spark/sql/datasource/predicate-cases.json), 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.

Shape of the interface, following @tdcmeehan's review, so that version 1 can grow without new exported functions or signatures:

• spark_ds_init(const SparkDsHost* host, const SparkDsLibrary** library) is the only function that a library exports. Spark calls it once in each JVM, when it loads the library, and it returns the table of the functions of the library, SparkDsLibrary. A library sets the functions of the operations it supports, and leaves the others NULL.
• The table, and the structs that Spark passes, start with their size. Within an ABI version, fields are only appended: Spark only reads the functions that fit in the size of a table, so a library built against an earlier header keeps working with the functions added later as NULL, and a library only reads the fields that fit in the size of a struct, so that it also runs on earlier releases of Spark. host->abi_version is the newest version that Spark supports, the table declares the version that it implements, and Spark rejects a table of a version that it does not support (UNSUPPORTED_ABI_VERSION).
• From Spark to the library: SparkDsHost, valid until the JVM exits, with the version of Spark and a log function that writes to the log of the driver or the executor, with the path of the library; SparkDsCreateInfo, passed to create, with the name of the data source and the options of the user, so that later releases can pass more, such as references to credentials instead of the credentials themselves in the options; SparkDsTaskContext, passed to read and create_data_writer and valid until the stream is released or the data writer is closed, with the IDs of the stage, partition, attempt and task attempt, and is_interrupted, which tells a read or a write that its task was killed, since Spark cannot interrupt native code; and SparkDsWriteInfo, passed to create_writer, which replaces spark_ds_create_writer and spark_ds_create_stream_writer, with the mode of the write, SPARK_DS_WRITE_APPEND or SPARK_DS_WRITE_TRUNCATE, and the ID of the query of a streaming write.
• Capabilities: create returns the capabilities of each data source, which may depend on its name and options: SPARK_DS_BATCH_READ, SPARK_DS_MICRO_BATCH_READ, SPARK_DS_BATCH_WRITE, SPARK_DS_STREAMING_WRITE and SPARK_DS_TRUNCATE. Spark reports an operation that a data source does not declare as unsupported, without calling the library, and only passes SPARK_DS_WRITE_TRUNCATE to a data source with SPARK_DS_TRUNCATE. The Data Source V2 table of a data source reports BATCH_WRITE and TRUNCATE as the data source declares them with the options of the table, so Data Source V2 rejects a batch write or overwrite that it does not support while it analyzes the query (UNSUPPORTED_FEATURE.TABLE_OPERATION), before it computes the data to write. The table always reports the reads and the streaming writes, which Data Source V2 would otherwise hand to Data Source V1, whose errors do not say why: they fail with DATA_SOURCE_*_NOT_SUPPORTED when Spark plans them. Each operation checks the capabilities again with its own options, which can override the options of a table in a catalog. If a declared capability needs a function that the table leaves NULL, the operation fails on the driver with NATIVE_DATA_SOURCE_ERROR and the name of the function, instead of being silently unsupported.
• Predicates: push_predicates sets, for each predicate, how the reader uses it: SPARK_DS_PREDICATE_UNUSED, SPARK_DS_PREDICATE_PARTIAL (only to skip data) or SPARK_DS_PREDICATE_COMPLETE. Spark shows the predicates that the reader uses as pushed in the plan, and still evaluates all of them after reading.
• Handles: each handle is released by the close function of its kind, close, close_reader, close_stream_reader, close_writer or close_data_writer, and by no other function. commit_data_writer no longer releases the data writer: Spark calls close_data_writer after the commit or the abort.
• Threads: the callbacks of the stream that read initializes can be called from any thread, but never concurrently, like the functions of SparkDsHost and SparkDsTaskContext, so that an engine that reads the partitions natively can call them from its own threads.
• Types: the header lists the Spark types of version 1 with their Arrow types: boolean, the integer types, float, double, decimal as decimal128, string, binary, date, timestamp, timestamp_ntz, array, map and struct. Spark fails an operation whose schema has another type, such as an interval, a variant or a time, with NATIVE_DATA_SOURCE_ERROR before it passes the schema to the library, so that a later version can add types for the data sources that declare a capability for them.

Capability	Required functions	Optional functions
All	create, close	last_error, schema
SPARK_DS_BATCH_READ	create_reader, partitions, read, close_reader	push_predicates, push_limit, prune_columns, serialize_reader
SPARK_DS_MICRO_BATCH_READ	create_stream_reader, initial_offset, latest_offset, stream_partitions, read, close_stream_reader	serialize_stream_reader, commit_offset
SPARK_DS_BATCH_WRITE, SPARK_DS_STREAMING_WRITE	create_writer, commit, abort, close_writer, create_data_writer, write, commit_data_writer, abort_data_writer, close_data_writer	serialize_writer

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. Spark calls spark_ds_init once for each library in a JVM, and binds each function of the table that is not NULL, within the size of the table, 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. The memory that Spark passes to a function is allocated with Platform, at the offsets of the structs of the header, and freed after the call, except SparkDsHost, kept until the JVM exits, and SparkDsTaskContext, freed once the stream is released or the data writer is closed. The upcall stubs, SparkDsSink, which copies the byte arrays that the library returns, SparkDsTaskContext.is_interrupted, and SparkDsHost.log, one for each library, never throw, since an exception thrown by an upcall would crash the JVM. Spark already runs with --enable-native-access=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, except for the capabilities, the task contexts, the write modes, the uses of the predicates and the planning of batch scans (see Other changes): the Scala classes catch NativeLibrary.MissingFunctionException and NativeLibrary.LoadException instead of UnsatisfiedLinkError.

Arrow data: Spark still uses arrow-c-data, as in #59313, to export and import the schemas and the arrays as Arrow C structs, without copying their buffers. arrow-c-data calls the callbacks of the C structs through the JNI library that it bundles, for Linux and macOS on x86_64 and aarch64 and Windows on x86_64: the get_schema, get_next, get_last_error and release callbacks of the streams that the libraries export, the release callbacks of the schemas and arrays that Spark imports, and the release callbacks of the schemas and arrays that Spark exports, which are functions of that JNI library. So Spark calls the libraries with FFM, and the Arrow callbacks go through JNI, inside arrow-c-data, once for each batch or schema rather than for each row. Library authors only see the C interface either way. An FFM implementation of the Arrow C data interface, for example in Arrow Java, could replace this JNI library 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 (linux is Linux with glibc, and Linux with musl is linux-<arch>-musl). 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 package that a session added last over its other packages and the configured ones, as a Python data source registered again replaces the previous one: a session cannot remove an artifact, so it upgrades a package by adding the new version under another file name. Two configured packages that provide the same data source fail with NATIVE_DATA_SOURCE_PACKAGE_CONFLICT. 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 conditions: NATIVE_DATA_SOURCE_UNSUPPORTED_JAVA_VERSION (0A000), raised when a query uses a native data source on Java 21 or earlier, and NATIVE_DATA_SOURCE_UNSUPPORTED_PLATFORM (0A000), raised where the JNI library of the C data interface of Arrow Java does not load: Arrow Java ships it for Linux with glibc and macOS on x86_64 and aarch64, and for Windows on x86_64. The lookups still find the data source, so the errors say why it cannot be used rather than that it does not exist. DATA_SOURCE_CONTINUOUS_WRITE_NOT_SUPPORTED (0A000), raised when a continuous query writes to a native data source, before Spark creates a writer, like DATA_SOURCE_CONTINUOUS_SCAN_NOT_SUPPORTED for reads.
• A library whose read returns success without initializing the stream is reported as an error: arrow-c-data calls the callbacks of a stream without checking them, and calling NULL would crash the JVM.
• The test library test_native_datasource.cc implements the function table, and the tests compile it with spark_datasource.h instead of the JNI headers. Compiled with defines, it also provides the libraries that the tests need: a read-only library, a table that ends before read, a table of version 2, a failing spark_ds_init, no table, and no spark_ds_init.
• The CheckConnectJvmClientCompatibility exclusion of org.apache.spark.sql.datasource, added by [SPARK-60101][SQL] Support native data sources implemented in Rust or C++ #59313 for NativeBridge, is not needed.
• CI: the Java 25 build with Maven job of a pull request also runs NativeDataSourceSuite, NativeDataSourceClusterSuite and NativeDataSourceConnectSuite, with the C++ compiler of the runner, since the other jobs, on Java 17, skip the tests that load a library.
• Checks of the data: before importing a batch, Spark also checks that the batch and each array have as many children as their type has fields, with non-NULL pointers, and after importing it, that the data agrees with the nullability of the schema to read, at every level. Option keys and values, and column names, that contain U+0000 fail with NATIVE_DATA_SOURCE_ERROR.
• Predicates: a float literal is written as the double of the same value, so that a reader that parses JSON numbers as doubles gets it exactly, a string literal that is not valid UTF-8 is not pushed, and a reader may only take a part of a predicate as true when every function above it is AND or OR.
• Packages: the abiVersion of a manifest is the oldest version of the interface that the library implements. Spark uses a package whose version is at most the newest that it supports, and ignores the fields of a manifest that it does not know.
• Arrow UDFs: Python UDFs over the batches of a native data source need SPARK-60294 ([SPARK-60294][SQL] Disable Arrow-backed columnar input to Python UDFs by default #59551), which disables the Arrow columnar input of Python UDFs (SPARK-56350) by default: it kept references to the pass-through columns of a batch after the source refilled or released their vectors, as the reader of a native data source does. The reader wraps its vectors in ReusedArrowColumnVector, whose closeIfFreeable does nothing, which the columnar input path can respect once it is enabled again with a marker for the vectors that their producer reuses.
• Batch scans: the Data Source V2 adapter, execution.datasources.v2.columnar, plans a batch scan with a driver-only ColumnarBatchPlanner, which gets the pushdown, and whose plan returns the partitions with a serializable reader factory. For a native data source, the planner holds the reader of the library, and the reader factory only has the location of the library, the name of the data source, the read schema and the state that serialize_reader returned. A scan builder builds its scan once, and the scan takes over the planner, or creates one if the operations were not pushed down, so a streaming scan never creates one. A batch is still planned once, when Spark first asks for its partitions, and a planning failure is kept and thrown again. The reader is released once the scan is planned, or by a Cleaner if the scan is never planned, as before.
• commit_data_writer returns its message through the sink. Spark never aborts a data writer once the library returns 0, even if it then rejects the message, and it always calls close_data_writer afterwards, which is the only function that releases a data writer.
• Publishing the header: dev/make-distribution.sh copies spark_datasource.h to the include directory of a binary distribution, as a JDK ships include/jni.h, and the spark-sql JAR contains it too. The header only depends on the C standard library, so a library can also copy it into its source tree and build without Spark, as for the Arrow C data interface and ADBC's adbc.h; the docs say so, with the URL of the header at the tag of a release.

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:

• From the latest review round of [SPARK-60101][SQL] Support native data sources implemented in Rust or C++ #59313 (aa0f3ba, e3ff80c), in the terms of the C interface:
  • Spark evaluates all the pushed predicates again after reading, and shows the predicates that a library uses in the plan. spark_datasource.h states the rule for skipping data (NaN, -0.0, nulls, the UTF-8 order of strings, parts of predicates), and the jar contains predicate-cases.json, generated by a test from the evaluation of Spark. The JSON of the predicates is ASCII.
  • create_writer gets the ID of the streaming query, SparkDsWriteInfo.query_id, which stays the same when the query restarts from its checkpoint.
  • Offsets must be compact JSON on a single line, as the offset log stores them.
  • An installed library replaced in place fails with INVALID_NATIVE_DATA_SOURCE_LIBRARY.REPLACED, and the caches identify files by their file key too.
  • Concurrent distributions of a configured package in a session are serialized, a scan planned again after a failure fails with the same error, the unsupported Arrow types and the link errors of Arrow's own JNI library are reported as NATIVE_DATA_SOURCE_ERROR, the column vectors of a read are not freeable between batches (which broke struct columns), packages of an unsupported version are logged, PySpark addArtifact only passes file: URIs as they are, and the docs describe the lifecycle of writers and the memory of native data sources.
  • A NULL handle was already reported as an error here, by NativeLibrary, and a missing offset as well, since the library returns it through the sink.
• 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 declare the capability SPARK_DS_BATCH_READ, and only implement the functions needed for batch reads, plus last_error, leaving the others NULL in their table.

Rust (with arrow-rs only)

Cargo.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:

//! 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 = unsafe extern "C" fn(sink: *mut c_void, data: *const u8, length: i64);

/// A function of `SparkDsLibrary` that this data source does not implement, left NULL.
type Unused = Option<unsafe extern "C" fn()>;

/// `SparkDsCreateInfo` of `spark_datasource.h`: the name and the options of a data source.
#[repr(C)]
pub struct SparkDsCreateInfo {
    size: i64,
    name: *const c_char,
    option_keys: *const *const c_char,
    option_values: *const *const c_char,
    num_options: i64,
}

/// `SparkDsLibrary` of `spark_datasource.h`, up to `read`, the last function that this data
/// source implements: Spark only reads the functions that fit in `size`.
#[repr(C)]
pub struct SparkDsLibrary {
    size: i64,
    abi_version: i32,
    last_error: unsafe extern "C" fn() -> *const c_char,
    create: unsafe extern "C" fn(
        info: *const SparkDsCreateInfo,
        capabilities: *mut u64,
        source: *mut *mut c_void,
    ) -> c_int,
    schema: unsafe extern "C" fn(source: *mut c_void, out: *mut FFI_ArrowSchema) -> c_int,
    create_reader: unsafe extern "C" fn(
        source: *mut c_void,
        schema: *mut FFI_ArrowSchema,
        reader: *mut *mut c_void,
    ) -> c_int,
    create_stream_reader: Unused,
    create_writer: Unused,
    close: unsafe extern "C" fn(source: *mut c_void),
    push_predicates: Unused,
    push_limit: Unused,
    prune_columns: Unused,
    partitions: unsafe extern "C" fn(reader: *mut c_void, add: Sink, sink: *mut c_void) -> c_int,
    serialize_reader: Unused,
    close_reader: unsafe extern "C" fn(reader: *mut c_void),
    initial_offset: Unused,
    latest_offset: Unused,
    stream_partitions: Unused,
    serialize_stream_reader: Unused,
    commit_offset: Unused,
    close_stream_reader: Unused,
    read: unsafe extern "C" fn(
        task: *const c_void,
        reader_state: *const u8,
        reader_state_length: i64,
        partition: *const u8,
        partition_length: i64,
        out: *mut FFI_ArrowArrayStream,
    ) -> c_int,
}

const SPARK_DS_BATCH_READ: u64 = 1;

static LIBRARY: SparkDsLibrary = SparkDsLibrary {
    size: std::mem::size_of::<SparkDsLibrary>() as i64,
    abi_version: 1,
    last_error,
    create,
    schema,
    create_reader,
    create_stream_reader: None,
    create_writer: None,
    close,
    push_predicates: None,
    push_limit: None,
    prune_columns: None,
    partitions,
    serialize_reader: None,
    close_reader,
    initial_offset: None,
    latest_offset: None,
    stream_partitions: None,
    serialize_stream_reader: None,
    commit_offset: None,
    close_stream_reader: None,
    read,
};

/// The function that Spark calls when it loads the library.
#[no_mangle]
pub unsafe extern "C" fn spark_ds_init(
    _host: *const c_void,
    library: *mut *const SparkDsLibrary,
) -> c_int {
    *library = &LIBRARY;
    0
}

struct RangeSource {
    end: i64,
}

struct RangeReader {
    end: i64,
}

thread_local! {
    static LAST_ERROR: RefCell<CString> = RefCell::new(CString::default());
}

fn arrow_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
}

unsafe extern "C" fn last_error() -> *const c_char {
    LAST_ERROR.with(|error| error.borrow().as_ptr())
}

unsafe extern "C" fn create(
    info: *const SparkDsCreateInfo,
    capabilities: *mut u64,
    source: *mut *mut c_void,
) -> c_int {
    call(|| {
        let info = &*info;
        let mut range = RangeSource { end: 10 };
        for i in 0..info.num_options as usize {
            if CStr::from_ptr(*info.option_keys.add(i)).to_str()? == "end" {
                range.end = CStr::from_ptr(*info.option_values.add(i)).to_str()?.parse()?;
            }
        }
        *capabilities = SPARK_DS_BATCH_READ;
        *source = Box::into_raw(Box::new(range)) as *mut c_void;
        Ok(())
    })
}

unsafe extern "C" fn schema(_source: *mut c_void, out: *mut FFI_ArrowSchema) -> c_int {
    call(|| {
        std::ptr::write(out, FFI_ArrowSchema::try_from(arrow_schema().as_ref())?);
        Ok(())
    })
}

unsafe extern "C" fn 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(())
    })
}

unsafe extern "C" fn close(source: *mut c_void) {
    drop(Box::from_raw(source as *mut RangeSource));
}

unsafe extern "C" fn partitions(reader: *mut c_void, add: 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(sink, bytes.as_ptr(), bytes.len() as i64);
        }
        Ok(())
    })
}

unsafe extern "C" fn close_reader(reader: *mut c_void) {
    drop(Box::from_raw(reader as *mut RangeReader));
}

unsafe extern "C" fn read(
    _task: *const c_void,
    _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(arrow_schema(), vec![Arc::new(ids)])?;
        let batches = RecordBatchIterator::new(vec![Ok(batch)], arrow_schema());
        std::ptr::write(out, FFI_ArrowArrayStream::new(Box::new(batches)));
        Ok(())
    })
}

Build and install, here on macOS:

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:

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 .)

C++ (with Arrow C++)

cpp_range.cc:

// 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());
}

const char* LastError() { return last_error.c_str(); }

int Create(const SparkDsCreateInfo* info, uint64_t* capabilities, void** source) {
  return Call([&] {
    auto range = std::make_unique<RangeSource>();
    for (int64_t i = 0; i < info->num_options; i++) {
      if (std::string(info->option_keys[i]) == "end") {
        range->end = std::stoll(info->option_values[i]);
      }
    }
    *capabilities = SPARK_DS_BATCH_READ;
    *source = range.release();
  });
}

int GetSchema(void*, struct ArrowSchema* out) {
  return Call([&] { Check(arrow::ExportSchema(*Schema(), out)); });
}

int CreateReader(void* source, struct ArrowSchema*, void** reader) {
  return Call([&] { *reader = new RangeReader{static_cast<RangeSource*>(source)->end}; });
}

void Close(void* source) { delete static_cast<RangeSource*>(source); }

int 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 CloseReader(void* reader) { delete static_cast<RangeReader*>(reader); }

int Read(const SparkDsTaskContext*, 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));
  });
}

// The functions of the data source. The ones that it does not implement are NULL.
const SparkDsLibrary kLibrary = {
    .size = sizeof(SparkDsLibrary),
    .abi_version = SPARK_DS_ABI_VERSION,
    .last_error = LastError,
    .create = Create,
    .schema = GetSchema,
    .create_reader = CreateReader,
    .close = Close,
    .partitions = Partitions,
    .close_reader = CloseReader,
    .read = Read,
};

}  // namespace

extern "C" SPARK_DS_EXPORT int spark_ds_init(const SparkDsHost*, const SparkDsLibrary** library) {
  *library = &kLibrary;
  return 0;
}

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:

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/

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:

>>> 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:

>>> spark.addArtifact("rust_range/rust_range.sparkpkg")
>>> spark.read.format("rust_range").option("end", 5).load().count()
5

Why are the changes needed?

Some data systems only have a client in native code, such as an in-house storage system written in C++ or the C SDK of a vendor, and Spark only needs to read and write them. Using them from Spark today requires writing a JVM Data Source V2 connector with its own JNI glue. With this change, such a source implements a stable binary interface once, in Rust, C++ or any language that can export C functions, is installed like any other native library or distributed as a single file, keeps working with the later releases of Spark that support its version of the interface, and works with every Spark API, including Spark Connect clients in any language, without any JVM code.

A source that needs what version 1 does not cover yet, such as row-level operations, or extensions outside of Data Source V2, such as table functions and SQL commands, is better served by a JVM connector, which can still read and write through a native library with the Arrow C stream interface, as lance-spark does with Lance. The docs say so in a new "When to Use a Native Data Source" section.

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 predicates.
• 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).
  • From the latest review round of [SPARK-60101][SQL] Support native data sources implemented in Rust or C++ #59313: Spark evaluates the predicates again when a library claims them all; the answers map back to the predicates that Spark passed; no limit is pushed with a filter; the columns of the predicates are read; the JSON is ASCII; the predicate test cases (golden file); a struct column read in several batches through whole-stage codegen and a nested field; large lists and dictionary encoding rejected with the name of the column; a reader planned again after a failure; offsets that are missing, multi-line or not JSON; the query ID across a restart; a failed micro-batch aborted with its epoch ID; stop() closing the stream reader; NULL handles from each of the five create functions; concurrent distributions of a configured package; a valid package with another checksum; the logging of packages of an unsupported version; an installed library replaced in place; link errors; and addArtifact with a file name that contains a colon.
• 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 last_error reports create failed with the error code 1., and a library whose read returns success without initializing the stream fails with The library did not initialize the stream, or a callback of the stream is NULL.

New tests for the shape of the interface, in NativeDataSourceSuite:

• the bindings match the layout of spark_datasource.h: compiles a program that prints the size of each struct of the header and the offset of each field, and compares them with the offsets that NativeLibrary uses.
• operations that a data source does not support: a read-only library, whose data sources only declare SPARK_DS_BATCH_READ, fails batch writes while Spark analyzes the query, with UNSUPPORTED_FEATURE.TABLE_OPERATION, before a shuffle stage of the data to write fails, and streaming reads and streaming writes with DATA_SOURCE_*_NOT_SUPPORTED; a data source whose library implements batch reads but that only declares streaming reads fails batch reads; a declared capability with a NULL function fails with the name of the function; an overwrite without SPARK_DS_TRUNCATE fails, in batch while Spark analyzes the query, and in streaming with the output mode complete; and a write to a table in a catalog whose options remove a capability of the table fails when Spark plans it, with DATA_SOURCE_BATCH_WRITE_NOT_SUPPORTED, or NATIVE_DATA_SOURCE_ERROR for an overwrite.
• ColumnarDataSourceSuite: the table reports the capabilities of the data source, where a batch write or overwrite that the data source does not support fails while Spark analyzes it, without a writer, and build and plan a batch scan once, where a scan builder builds one scan, whose batch is planned once, and a failed planning is not retried but fails again with the same error.
• reject the data that has nulls where the schema to read has none; reject the types and the strings that a library cannot get, with an interval in a read, an array of variants in a write, which fails before create_writer, and an option with U+0000; reject the Arrow data that Spark cannot read, with batches without their columns; parse manifests, with an abiVersion that is not positive; encode predicates as JSON, with float literals and strings that are not valid UTF-8; and predicate cases, with a float column and a predicate under =.
• From the review of gerashegalov: continuous processing fails with a native data source, for writes and reads; the package that a session added last replaces the other packages; comparisons and IN on arrays and structs are not pushed, while null checks and the comparisons of struct fields of simple types are; fail clearly where the C data interface of Arrow Java does not load; reject the Arrow data that Spark cannot read, with batches whose struct marks rows as null, counted or not, and arrays with a null_count of -1 but no validity buffer, which Arrow Java cannot import; predicate cases, with a binary column, which compares as unsigned bytes; and platform names, with musl.
• Arrow UDFs on the columns of a native data source: a pandas UDF over one batch, two and 2,000, which checks the pass-through columns of each row against the result of the UDF. Locally, it fails without [SPARK-60294][SQL] Disable Arrow-backed columnar input to Python UDFs by default #59551 and passes with it. The CI of pull requests skips it, since the job that runs the native data source suites has no pandas.
• read only the functions that fit in the table of a library: a table that ends before read, like the table of a library built against an earlier header, makes read unsupported rather than read past its end.
• pass the version of Spark, and log with the log of Spark: the library checks SparkDsHost, and its messages reach the log of Spark at their level, with the path of the library.
• pass the task to reads and data writers: each read and each data writer gets the IDs of its task.
• tell a read that its task was killed: a read that waits until is_interrupted returns true ends once its job is canceled.
• reject the libraries that spark_ds_init does not initialize: a failing spark_ds_init, which logs why once, a library that returns no table, and one that does not export spark_ds_init, with the same error when they are used again, since Spark initializes a library only once.
• push down predicates checks the answers of push_predicates, including SPARK_DS_PREDICATE_PARTIAL and a value that Spark does not know, and invalid installed libraries and invalid packages check a table of version 2, which Spark rejects with UNSUPPORTED_ABI_VERSION.

The tests that load a library are skipped before Java 22, like they are skipped without a C++ compiler. So the jobs on Java 17 only run the tests that do not load a library, and the Java 25 build with Maven job of a pull request runs the native data source suites, as do the scheduled Java 25 builds. do not abort a data writer that committed with an invalid message checks that a data writer released by its commit is not aborted. 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; both export only spark_ds_init), 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 work with the header in the include directory of a distribution: the header copied as dev/make-distribution.sh copies it, then the documented c++ command with -I"$SPARK_HOME/include".
• The C++ test library builds without warnings with Apple clang (-Wall -Wextra), in all its variants.
• SparkThrowableSuite, for the new error conditions.
• 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.

[LuciferYang]

Thanks for the FFM version. I compared the 30 bindings in NativeLibrary with spark_datasource.h, the test library and the Rust and C++ examples in the docs, and they match. The in-place replacement of an installed library, from my last review on #59313, now fails with INVALID_NATIVE_DATA_SOURCE_LIBRARY.REPLACED. I left one inline comment, on a failure path introduced by returning the commit message through the sink.

On testing: the tests that load a library need Java 22 or later, so on the Java 17 CI of a PR they are all canceled by assume, and the JUnit reports list them as passed. No FFM call in NativeLibrary runs on a PR, and a regression in the bindings would only show up in the scheduled Java 25 build after the merge. The Java 25 build with Maven job already builds Spark on Java 25, and it ran on this PR. Could it also run NativeDataSourceSuite, NativeDataSourceClusterSuite and NativeDataSourceConnectSuite? On the PR head, with the test JVM on JDK 26, NativeDataSourceSuite and NativeDataSourceClusterSuite pass, except the Java version test, which is canceled as expected. The GitHub runner has a C++ compiler.

[HyukjinKwon]

@LuciferYang Thanks for checking the bindings against the header and the examples. On testing, good point: in e5352e1, the Java 25 build with Maven job of a pull request also runs NativeDataSourceSuite, NativeDataSourceClusterSuite and NativeDataSourceConnectSuite after the build, with the C++ compiler of the runner. On this commit, they ran on Linux with Java 25: 55 tests passed in NativeDataSourceSuite and NativeDataSourceClusterSuite, with only the Java version test canceled, as expected, and 2 in NativeDataSourceConnectSuite. So the FFM calls now run on every pull request. I also addressed the inline comment, with a test.

[tdcmeehan]

Thanks for moving to the C header, and for documenting the lifecycle so carefully. I have one concern, about how the header grows. I think it's worth settling before 4.4.0: version 1 is what library binaries get built against, and this is the one change that can't come later as an optional function.

The problem: the header grows by one function per feature.

1. Every feature is a new function. The Compatibility section adds features as optional functions, and the Limitations section lists six to add that way: statistics, custom metrics, admission control, aggregate/sort/partitioning pushdown, cancellation and memory accounting. The predicate declaration and three-state answer from [SPARK-60101][SQL] Support native data sources implemented in Rust or C++ #59313 are planned the same way. DSv2 itself keeps growing: master has 37 Supports* interfaces, 7 of them added since July 2025.
2. Each function is a separate contract that every host and SDK has to reimplement. Each has its own argument encoding, ownership, threading and ordering rules, written in prose. There are five handle types, released by six different functions. That's where bugs land: @LuciferYang's catch in this PR, a released data writer passed to spark_ds_abort_data_writer, came from one of those rules, in Spark's own binding.
3. Signatures are frozen, so a change is a second function. For example, giving spark_ds_read a task context for memory accounting or cancellation means a second read, and a library that also runs on older releases exports both.
4. Support is inferred from symbol names. It's per library, not per data source or options, and a library can't declare which predicates it handles. A missing export, for example a misspelled name, isn't an error: the operation becomes unsupported, or, for the pushdown and serialize functions, silently does nothing.
5. Nothing flows from Spark to the library. The library reports its version, but Spark never tells it Spark's version or features, and there's no callback table. Cancellation, memory accounting and logging all need one.

Proposal: freeze the C core, and move everything that grows into versioned messages.

The header is already halfway there: 14 of its 30 functions pass bytes or JSON (partitions, state, commit messages, offsets, predicates). This finishes the move, with one schema. A sketch, with placeholder names:

int32_t     spark_ds_abi_version(void);
const char* spark_ds_last_error(void);

/* Opens a data source, reader, writer or data writer. The request says which, with what it
 * needs, plus Spark's protocol version and features. The reply has the library's version and
 * what this object supports. */
int  spark_ds_open(const SparkDsHost* host, const uint8_t* request, int64_t length,
                   SparkDsSink reply, void* sink, void** handle);

/* Every control operation: schema, pushdown, partitions, offsets, commit, abort. */
int  spark_ds_call(void* handle, const uint8_t* request, int64_t length,
                   SparkDsSink reply, void* sink);

/* Every data operation, as Arrow C streams: a read fills `out`, a write consumes `in`. */
int  spark_ds_stream(void* handle, const uint8_t* request, int64_t length,
                     struct ArrowArrayStream* in, struct ArrowArrayStream* out);

/* Releases any handle. Nothing else does. */
void spark_ds_close(void* handle);

• SparkDsHost holds Spark's callbacks, such as memory reservation, cancellation and logging. It starts with its size, and fields are only appended.
• Requests and replies are protobuf messages, in a .proto next to the header. Old code ignores new fields, and field numbers are never reused.
• Spark only sends what the library declared in its open reply, so a new request never reaches an old library.
• Data stays as it is: zero-copy Arrow C streams.

How the planned additions would land:

Addition	As optional functions	As messages
Declared predicates, three-state answer	new pushdown functions	a list in the open reply, an enum in the pushdown reply
Memory accounting, cancellation	a setup function each, or a second read	fields appended to SparkDsHost
Statistics, partitioning, sort order	a function each	fields in the planning reply
Aggregate and top-N pushdown	a function each	fields in the pushdown request
Admission control	a second latest_offset	a field in the offset request
Overwrite by filter	a new function, since bool overwrite can't carry a filter	a write mode and a predicate field
Substrait, as @lidavidm suggested	a new pushdown function	another encoding of the predicate field

What it buys:

• A feature is new fields, with generated code in every language. No new C function, binding or ownership rule.
• One ownership rule: request bytes live for the call, the sink copies replies, close releases handles, and Arrow structs follow Arrow's rules.
• Tooling checks compatibility. Spark's CI already runs buf breaking on the Spark Connect protos, and recorded messages from each release can be replayed against current libraries.
• A host that runs a library in another process, for example to isolate it on a shared server, forwards the control messages unchanged.
• If Spark ever wants native data sources on Java 17 and 21, a JNI bridge would be six functions that never change.

What it costs:

• Library authors need a protobuf runtime. SDKs can vendor it and link it statically, with hidden symbols, since several libraries can share a process. Spark already uses protobuf for Spark Connect and the UDF worker protocol.
• The .proto and its generated types check the contract, not the C compiler.
• Control calls pay for encoding. They're rare, and data stays zero-copy.
• Semantics still need specifying. The NaN, null and partial-predicate rules you wrote carry over unchanged.

Precedents:

• Arrow splits it the same way. The C Data Interface is a few frozen structs. The IPC format is versioned messages: since 1.0, which guarantees forward and backward compatibility, it has been extended five times, and its Feature enum is "A means of negotiating between a client and server what features a stream is allowed to use".
• Comet shows both sides. Its plan, with 24 operator, 66 expression and 27 aggregate kinds, crosses one JNI function, createPlan, as protobuf, and a contrib envelope means "core's proto never has to change when a contrib is added". Its other JNI functions went from 9 to 19 in the past year, and createPlan gained two parameters. That's fine for Comet, whose JVM and native halves ship in one jar; under a public ABI, each would be a second function or a break. Gluten also passes plans as Substrait bytes, and asks the native side whether it supports a plan.
• Spark's UDF worker protocol (SPARK-55278) is protobuf, with Init.protocol_version and WorkerCapabilities that Spark checks during planning.
• protoc plugins: the compiler sends its version in the request, and the plugin answers with supported_features.
• OpenSSL 3 providers run in process, with dispatch tables and typed key/value parameters. Result: "A FIPS provider built from any validated version may be used together with an OpenSSL library built from any supported release from OpenSSL 3.0 onwards" (README-FIPS.md).
• ADBC is a wide C API that works, and shows what it takes. Drivers fill a function table through a versioned init function, and a driver manager handles the versions. The table has grown once in four years: revision 1.1.0 (2023) took it from 26 to 55 functions, 21 of them typed option getters and setters, and needed a handshake in which a 1.0.0 driver must fail and be retried. Since then it has added option keys and info codes, but no functions. DSv2, which this header mirrors, added 7 optional interfaces in the last 15 months.
• The other way: DataFusion FFI mirrors DataFusion's traits as C structs, and recommends "the same version of DataFusion by both the producer and consumer", though it's not strictly required.

Could version 1 take this shape? I'm happy to write the .proto for the current feature set, and the replay test.

[HyukjinKwon]

@tdcmeehan Thank you for the detailed proposal, and for offering to write the .proto. I agree that version 1 is what libraries get built against, so its shape has to be settled now. In d2df236, I kept typed C functions, but behind a single entry point and size-prefixed structs, the way ADBC's function table and JNI's JNIEnv grow, and moved what grows into fields and capabilities. On your five points:

1. Every feature is a new function. Features that are parameters now go into structs that grow: memory accounting, for example, would be functions appended to SparkDsTaskContext, and overwrite by filter a write mode with a field appended to SparkDsWriteInfo. A new operation, such as statistics, is still a new function, but a slot appended to the function table rather than a new exported symbol, and Spark only calls it for a data source that declares it.
2. Each function is a separate contract. Handles now follow one rule: each handle is released by the close function of its kind, and by nothing else. commit_data_writer no longer releases the data writer, which removes the class of bug that @LuciferYang found: Spark calls the new close_data_writer after the commit or the abort. What Spark passes is valid only during the call, except SparkDsHost, until the JVM exits, and SparkDsTaskContext, until the stream is released or the data writer is closed.
3. Signatures are frozen. read and create_data_writer get a const SparkDsTaskContext*, and create_writer, which replaces create_writer and create_stream_writer, gets a const SparkDsWriteInfo* with the write mode and the query ID. Both start with their size and only get fields appended, so a task context for memory accounting or cancellation does not need a second read.
4. Support is inferred from symbol names. A library exports only spark_ds_init, which returns its table, SparkDsLibrary, so in C and C++ the compiler checks the names and types of the functions. create returns the capabilities of each data source, which can depend on its options. Spark reports an operation that a data source does not declare as unsupported without calling the library, and if a declared capability needs a function that the table leaves NULL, it fails on the driver with the name of the function. push_predicates now answers, for each predicate, SPARK_DS_PREDICATE_UNUSED, PARTIAL (only to skip data) or COMPLETE. Spark shows the used ones as pushed and still evaluates all of them, so trusting COMPLETE can come later as an opt-in capability.
5. Nothing flows from Spark to the library. spark_ds_init gets SparkDsHost: the newest ABI version that Spark supports, the version of Spark, and a log function that writes to the log of the driver or the executor. SparkDsTaskContext has the IDs of the stage, partition, attempt and task attempt, and is_interrupted, so a read or a write that blocks can stop once its task is killed. The library returns a table of the version that it implements, so a new version does not need ADBC's fail-and-retry handshake.

Your table, in these terms:

Addition	Now
Declared predicates, three-state answer	in version 1: the answers of push_predicates
Cancellation	in version 1: SparkDsTaskContext.is_interrupted
Memory accounting	functions appended to SparkDsTaskContext
Statistics, partitioning, sort order	a function appended to the table, with a capability
Aggregate and top-N pushdown	a function each appended to the table, with a capability
Admission control	a function appended to the table, with a capability
Overwrite by filter	a write mode, and a predicate field appended to SparkDsWriteInfo, with a capability
Substrait	a pushdown function appended to the table, with a capability

Why I would keep typed functions rather than messages:

• The header only depends on the C standard library, like the Arrow C data interface that it builds on. Messages would add a protobuf runtime and generated code to every library, which weighs most on C and C++ libraries.
• The compilers check the table. With messages, the encoding is checked, but which fields a request carries, and who owns what, would still be prose.
• Control calls are rare, and the data is Arrow C structs either way, so neither design is faster.

Two costs that messages would avoid remain: a new operation is still a new function to specify, and a host that runs a library in another process would marshal each function instead of forwarding bytes.

Tests: the bindings match the layout of spark_datasource.h compiles the header and compares every offset and size with the bindings, and a library whose table ends before read checks that Spark does not read past the size of a table. Others cover the capabilities, the host and its logging, the task context, a read that only ends once is_interrupted is true, and a failing spark_ds_init. Once 4.4.0 is released, the C counterpart of your replay test would be to keep building the test library against the released header. I also updated the PR description.

Does this address your concern? If you still think the control calls should be messages, could you point to the addition that you think does not fit a table?

[LuciferYang]

Thanks for fixing the data writer so quickly, with a test, and for running the native suites on Java 25 in the CI of pull requests. The function table also makes the header easier to build against than JNI: the compiler checks every function a library provides.

Speaking as one of the Lance maintainers, I'd like to share how this looks from lance-spark, since Lance is one of the motivating examples in the description, and hear your thoughts.

For reads and writes, lance-spark already goes through the Rust core of Lance: the Lance Java bindings fill an ArrowArrayStream at an address that the JVM allocates (LanceFragmentScanner.exportArrowStream), which is the same exchange as the read function of SparkDsLibrary. So a native data source would not remove a layer or a copy for Lance.

Most of what users get from lance-spark is beyond reads and writes: the VECTOR_SEARCH, SEARCH and HYBRID_SEARCH table functions, CREATE INDEX and the other index commands, UPDATE, DELETE and MERGE INTO through _rowid and _rowaddr, fragment-aware joins, count(*) and top-N pushdown, and branches and tags. Version 1 of the interface covers reads and writes with filter, column and limit pushdown, so a native Lance data source would give users less than lance-spark does today.

This is also where our cost of supporting several Spark releases is. lance-spark has a module for each Spark version from 3.4 to 4.2, but the read and write code is shared. What differs between versions is mainly the SQL extensions (parser and optimizer rules) and the row-level operations. For example, Spark 4.2 plans an insert-only MERGE INTO as InsertOnlyMerge instead of AppendData, and we needed a new rule so that blob columns are still copied. A stable ABI would not help with that part.

Where native data sources look like a clear win to me is a source that only has a native client and only needs reads and writes, such as an in-house storage system in C++ or a vendor's C SDK. Could the description and the docs lead with that kind of source, and with a concrete one if this work started from one? Your reply to @tdcmeehan lists what could be appended to the table later, such as statistics, aggregate and top-N pushdown, and overwrite by filter. For a source like Lance, table functions and row-level operations matter as much. Do you see them in scope for the table later, or out of scope on purpose? Saying so in the docs would help library authors tell early whether it fits their source.

None of this blocks the implementation; it is only about how the feature is positioned.

[HyukjinKwon]

@LuciferYang Thank you, the comparison with lance-spark is very helpful. I agree with the positioning. In 51af65c, the docs have a new "When to Use a Native Data Source" section, which leads with a source whose client exists only in native code, such as an in-house storage system in C++ or the C SDK of a vendor, and that Spark needs to read and write. The PR description leads with it too, and no longer lists Lance among the motivating examples. This work did not start from one specific source: the goal was to let a data source be written only in native code, which is why the examples in the docs read a range of numbers.

On scope, the line I would draw is Data Source V2 itself:

• Features of Data Source V2 that version 1 does not cover are in scope for later, as functions and capabilities appended to the table: aggregate pushdown, including count(*), and top-N pushdown; statistics; reporting the partitioning and the sort order of the data, which storage-partitioned joins use and which is the closest to fragment-aware joins; overwrite by filter; and row-level operations with metadata columns, as SupportsRowLevelOperations and SupportsDelta do with a row ID. The Limitations section now lists top-N pushdown, row-level operations and metadata columns too.
• Extensions outside of Data Source V2, such as table functions like VECTOR_SEARCH, SQL commands like CREATE INDEX, and parser and optimizer rules, are out of the scope of this interface on purpose: they plug into Spark's extension points, such as SparkSessionExtensions, rather than into Data Source V2. A source that needs them is better served by a JVM connector like lance-spark, which can keep sharing the native read and write path through the Arrow C stream interface, as it does now. The new section of the docs says so.

[zhouyuan]

Moving my comment from #59313 here, updated for the C header.

Thanks for this. The plain C ABI also makes native data sources a good fit for native execution engines running inside Spark, such as Apache Gluten (Velox) and Comet (DataFusion): as the description says, other hosts can load the same libraries. The library reads the data, and the engine runs the rest of the query natively. What's missing is on the Spark side.

Today, a query on a native data source with Gluten enabled looks like this:

read -> ArrowArrayStream -> Arrow Java -> ArrowColumnVector
     -> ColumnarToRow -> RowToVeloxColumnar -> Velox operators

That's two extra copies, so turning on a native engine can make these queries slower. With a small hook, the data could stay columnar in native memory from the source to the shuffle:

read -> ArrowArrayStream -> Velox (import) -> filter / agg / join / shuffle

A source library written once would then run on vanilla Spark and get native downstream execution under Gluten or Comet, without per-source connector work in each engine.

Two small requests:

1. A developer API to open a partition as an Arrow C stream. The library location, reader state and partitions are private to execution.datasources.v2.ffi, so an external engine can't call read for a planned scan today. Something like:

   @DeveloperApi
   trait SupportsArrowStreamRead { this: PartitionReaderFactory =>
     /** Initializes the ArrowArrayStream at `streamAddress` with the data of `partition`.
      *  The caller owns the stream and validates the arrays it reads. */
     def openArrowStream(partition: InputPartition, streamAddress: Long): Unit
   }

   ColumnarPartitionReaderFactory would implement it by calling read with Spark's own SparkDsTaskContext, which is freed when the caller releases the stream. Library loading, the checksum checks and spark_ds_init stay in Spark, and the vanilla path stays exactly as it is.

2. A threading contract for the stream in the header. The doc of read says Spark calls the stream "from the thread of the task that reads the partition", so library authors may rely on it. A native engine may call get_next from its own driver threads. Could version 1 guarantee instead that the stream may be called from any thread, though never concurrently? The functions of SparkDsHost and SparkDsTaskContext can already be called from any thread, so this would only extend that to the stream. It's free now, but hard to add once libraries depend on the current wording.

Optional, could be a follow-up: a columnar write entry point (the internal DataWriter[ColumnarBatch] that takes Arrow batches), so a native engine can write to a native sink without going through rows.

If this direction sounds reasonable, I'm happy to help review.

[gengliangwang]

I opened a companion Rust SDK prototype in #59520 for this PR's C ABI, pinned to 51af65c. It separates driver-side BatchSource / ScanBuilder from a serializable BatchReaderFactory, caches immutable plans after pushdown negotiation, and exposes a lifetime-bound task context with cooperative cancellation. It advertises only SPARK_DS_BATCH_READ, leaving unsupported function slots null, without changing the native ABI.

The SDK handles the function table, plan serialization, provider errors and unwinding panics, and Arrow ownership. It includes a Range example, packaging instructions, and passing standalone Rust/C tests. End-to-end Spark validation is still pending.

Feedback on the connector API and lifecycle is welcome!

[gengliangwang]

Following up on my two DSv2 adapter suggestions on #59313, after checking this PR at 51af65c. The planner/reader separation still applies; the capability suggestion is partly addressed and can now be narrowed to the DSv2 table layer.

1. Separate driver-side planning from executor-side reading.

   ColumnarReader still combines pushdown negotiation and partition planning with executor read(partition). NativeDataSourceReader still changes roles through a transient native handle and mutable readerState, and ColumnarScanBuilder.build() passes closures capturing the mutable builder.

   Could we keep ColumnarTable -> ColumnarScanBuilder -> ColumnarScan, but introduce a driver-only BatchReadPlanner whose plan() returns PlannedBatch(partitions, readerFactory)? The serializable executor factory would contain only the library location, source name, read schema, and serialized native state. build() would establish stable scan metadata and explicit planner ownership, with defined behavior for repeated calls.

   This can preserve the C ABI and lazy planning point: call partitions, then serialize_reader, and close the native reader in finally. Keep the existing caching of successful plans and planning failures, lazy batch-planner creation for streaming-only sources, and the Cleaner fallback for scans that never reach planning. The benefit is clearer ownership and a smaller executor object, without changing when native partitions are planned.

2. Expose source-specific capabilities through Table.capabilities().

   The new capability flags returned by native create and the checks in NativeDataSource.create address discovery and reject unsupported operations before invoking native reader/writer creation. However, ColumnarTable.capabilities() still advertises batch/streaming reads and writes plus truncate for every source, and NativeDataSourceV2.getTable() constructs that table without a capability override.

   Could we propagate the existing source-specific capability information into the DSv2 table as well, so a batch-only source is represented as such during DSv2 capability checks? The discovery mechanism now exists, so there is no need to add another native ABI query. Since capabilities may depend on options and query options can override table options, the table-level declaration needs defined, stable semantics; the operation-specific checks should remain.

These are internal adapter suggestions. The Rust-side planner/factory split in #59520 complements them, but does not change these Scala classes.

[HyukjinKwon]

@gengliangwang Thank you for both suggestions, and for the Rust SDK prototype in #59520.

1. Planning on the driver, reading on the executors: done in 534dfca, as you suggested. ColumnarReader is replaced by a driver-only ColumnarBatchPlanner, which gets the pushdown, and whose plan() returns ColumnarPlannedBatch(partitions, readerFactory). For a native data source, NativeBatchPlanner holds the native reader: plan() calls partitions, then serialize_reader, and closes the reader in finally. The serializable NativePartitionReaderFactory only has the location of the library, the name of the data source, the read schema and the state of the reader. ColumnarScanBuilder.build() builds its scan once, and the scan takes over the planner that the operations were pushed down to, with no closure over the builder. A scan whose operations were not pushed down only creates a planner if its batch is planned, so a streaming scan never creates one. ColumnarBatchScan plans once, when Spark first asks for the partitions or the reader factory, and keeps the result or the failure. The Cleaner still releases the reader of a scan that is never planned. Tests: build and plan a batch scan once in ColumnarDataSourceSuite, and a scan whose planning failed fails again with the same error in NativeDataSourceSuite, now through ColumnarBatchScan.

2. Capabilities in Table.capabilities(): done for the writes. ColumnarTable.capabilities() now asks the data source, once per table, with the options of the table: for a native data source, Spark creates a data source with them and reads its capabilities. The table reports BATCH_WRITE and TRUNCATE as the data source declares them, so Data Source V2 rejects a batch write or overwrite that the source does not support while it analyzes the query, with UNSUPPORTED_FEATURE.TABLE_OPERATION, before it computes the data to write. Before, with adaptive execution, the shuffle stages of the data to write ran first.

   The table always reports BATCH_READ, MICRO_BATCH_READ and STREAMING_WRITE. When a TableProvider's table lacks one of them, Data Source V2 does not reject the query: it falls back to Data Source V1, in DataFrameReader.load, DataStreamReader.load and DataStreamWriter.start. The query then fails with <format> is not a valid Spark SQL Data Source. or Data source <format> does not support streamed reading. (legacy errors), which do not say that the data source does not declare the operation. So those operations fail when Spark plans them instead, still on the driver and before any task, with DATA_SOURCE_BATCH_SCAN_NOT_SUPPORTED, DATA_SOURCE_MICRO_BATCH_SCAN_NOT_SUPPORTED or DATA_SOURCE_STREAMING_WRITE_NOT_SUPPORTED.

   On the semantics: the table has the options of the table, which for a table in a catalog are those of CREATE TABLE, and the options of a write can override them. So, as you said, the checks of each operation remain, with its own options. The header and the docs say so.

The test operations that a data source does not support covers both: a batch write whose data would fail in a shuffle stage now fails while the query is analyzed, and a write to a table in a catalog whose options remove a capability of the table fails when it is planned.

[HyukjinKwon]

@zhouyuan Thank you, running a native data source under Gluten or Comet without copies is a good use of the C ABI.

2. Threading contract: agreed, and done in 534dfca. The header now says that the callbacks of the stream that read initializes can be called from any thread, but never concurrently, like the functions of SparkDsHost and SparkDsTaskContext, and the docs say so too. Spark itself still calls them from the thread of the task.

3. A developer API to open a partition as an Arrow C stream: this fits version 1 without changing the C ABI, since read already initializes a stream at an address that the caller allocates. I would rather add it in a follow-up, as a @DeveloperApi with its own review, than make it public in this PR, because a few things need deciding first: which object an engine gets it from; who owns the SparkDsTaskContext, and when it is freed; and what the engine must check before importing the arrays, which Spark checks today (sliced arrays, lengths shorter than the parent, the types it cannot read). After the planner split in 534dfca, from @gengliangwang's review, the reader factory of a batch scan only holds the location of the library, the name of the data source, the read schema and the state of the reader, which is what your SupportsArrowStreamRead needs.

Columnar write entry point: also a follow-up. The data writers of the adapter already take Arrow-backed ColumnarBatches, and an engine could pass an Arrow C stream to write directly in the same way.

I'd be glad to have your review on both follow-ups.

[HyukjinKwon]

@dongjoon-hyun @sarutak @gerashegalov You approved #59313, which this PR replaces, so the approvals there do not carry over here. Could you take another look? Since #59313, which this PR includes up to e3ff80c, the changes are:

• The interface is a C header, spark_datasource.h, called through the FFM API on Java 22 or later instead of JNI (b9ad01a), and shipped in the include directory of a distribution (4f032d5).
• Following the review here, a library exports only spark_ds_init, which returns a function table, with size-prefixed structs from Spark and the capabilities of each data source (d2df236, b36ec5e).
• A data writer that committed is no longer aborted, and the native data source suites run in the Java 25 build with Maven job of pull requests (e5352e1). The docs say which sources native data sources fit (51af65c).
• Batch scans are planned on the driver, with a serializable reader factory for the executors, and the Data Source V2 table reports the batch writes that a data source supports, so an unsupported write fails while the query is analyzed (534dfca).

The "Differences from #59313" table at the top of the description summarizes the interface. CI is green on 534dfca.

[FelixYBW]

@zhouyuan

read -> ArrowArrayStream -> Arrow Java -> ArrowColumnVector
     -> ColumnarToRow -> RowToVeloxColumnar -> Velox operators

This is what @zhztheplayer's proposal SPARK-57468 tries to solve exactly. We still need couple of PRs in Spark + Gluten to achieve this. I implemented a prototype here

apache/gluten#13140 (comment)

[zhztheplayer]

@HyukjinKwon Thanks for working on this. Glad to see the community getting more active on native support.

Worth to mention or not, but I found that this PR combines 2 responsibilities:

1. Native library infrastructure: package discovery, platform selection, distribution, checksum verification, extraction, and loading.
2. Data source integration: provider names, capabilities, schema, pushdown, and reader/writer lifecycle.

Do you think it's possible to make the first part (native library infrastructure) general-purpose? Perhaps also keep the .sparkpkg infrastructure as a separate feature so that third-party plugins like Gluten can directly leverage it, regardless of whether they use FFM or JNI to implement the bindings.

Also, building on this PR’s columnar table and writer support, do you think refactoring V2TableWriteExec to accept columnar input directly would be a desirable follow-up? That would let upstream columnar operators feed batches into the native writer without going through rows and rebuilding Arrow batches. It’s a long-standing need for Gluten as well.

[zhztheplayer]

I felt this may cause bug since reusing a fixed VectorSchemaRoot can cause data corruption when the output of NativeDataSource is accumulated by the downstream operators (example), since the new data will replace the content of the old vectors in early ColumnarBatch instances.

It might be worth to try the root-free Data.importVectorSchemaRoot APIs that creates fresh roots internally?

[HyukjinKwon]

Thanks, this is the bug that @FelixYBW reported. A fresh root for each batch, as Data.importVectorSchemaRoot creates, would not fix it alone: the reader would still have to release each root at some point, and nothing tells it when a consumer is done with a batch. ColumnarBatch lets a reader reuse its vectors, and @sarutak showed that ArrowCachedBatchSerializer, which creates a root for each batch and closes the previous one in next(), fails in the same way. So the fix is in the consumer, which kept references to the pass-through columns: SPARK-60294, #59551, copies them. This PR keeps one root, and adds a test of Arrow UDFs over a native data source in 74aa0df.

[zhztheplayer]

This is what @zhztheplayer's proposal SPARK-57468 tries to solve exactly.

Yes, I am working on the PR #59154 as a fix. It will help Spark wire legacy columnar operators and Arrow-based operators (like what this PR is adding) correctly.

[sarutak]

Thanks for the FFM version, @HyukjinKwon. I reviewed 534dfca by building it with Java 22 and running two libraries against it: the test library of this PR, and a small probe library written against spark_datasource.h that logs what Spark passes to it and returns some malformed data on purpose. Besides the item below, I left inline comments on:

• Data that Spark accepts but reads wrongly, without an error: a batch with n_children = 0, and data with nulls in a column that schema declares not nullable.
• The predicate contract: how float and string literals are encoded, and the rule for using a part of a predicate.
• Two version 1 decisions that are hard to change after 4.4.0: the abiVersion of the manifest, and the set of Arrow types that reach a library.

Arrow UDFs over a native data source

I reproduced @FelixYBW's report on 534dfca with the test library (native_range, one partition). With end=200, batch_size=100, the query fails with IndexOutOfBoundsException: index: 0, length: 1 (expected: range(0, 0)), and with end=2000000, batch_size=1000, the check fails with MISMATCH id 111000 uid 0. With spark.sql.execution.arrow.pythonUDF.columnarInput.enabled=false, the results are correct.

The root cause is not in this PR but in the existing columnar input path of ArrowEvalPythonExec; this PR adds a source that runs into it with the default configuration. The Arrow columnar path of ColumnarArrowEvalPythonEvaluatorFactory (SPARK-56350, in 4.2.0, enabled by default) keeps the pass-through ColumnVectors of each input batch in passThruQueue until the Python result of that batch comes back, based on the assumption in its comment that "Arrow vectors are independently allocated per batch". A columnar PartitionReader does not promise that. ColumnarDataSource.scala:100 in this PR says that the batches "only need to be valid until the next one", and NativePartitionReader imports every batch into the same VectorSchemaRoot (NativeDataSource.scala:633) and closes it as soon as DataSourceRDD sees the end of the partition.

The same failure happens without a native data source. With spark.sql.cache.serializer=org.apache.spark.sql.execution.columnar.ArrowCachedBatchSerializer, which closes the previous root in next(), spark.range(0, 2000000, 1, 1).cache() with spark.sql.inMemoryColumnarStorage.batchSize=1000 and the same UDF fails with the same IndexOutOfBoundsException on the same build. (200 rows in batches of 100 worked there; I did not look into how far the input runs ahead in that case.)

So I think the fix belongs in the columnar input path, in a separate JIRA (I could not find an existing one): it must not keep references to an input batch after the producer has moved on, for example by transferring or copying the pass-through vectors, or by falling back to the row queue path. cc @viirya, since this is the path of SPARK-56350. As far as I can see, #59154 (SPARK-57468), which @zhztheplayer mentioned above, is about the conventions of the batches and the transitions between operators, and does not change this path, so the lifetime of the input batches would still need its own fix.

For this PR, two questions:

• Should it wait for that fix, or send the batches of a native data source to the row queue path until then? With the default configuration, any scalar Arrow or pandas UDF on a column of a native data source runs into this today.
• Could it add a test that runs an Arrow UDF over a native data source, with one batch and with several, and checks the pass-through columns against the UDF results? No test of this PR goes through ArrowEvalPythonExec, which is why this was not caught.

[sarutak]

checkArrays only checks the children that an array has, so a batch with n_children = 0 skips the checks of all its columns. Arrow Java does not catch it either: ArrayImporter.doImport only compares the number of children with the vector when there is at least one child. I tried it with a library that returns a batch of 4 rows with n_children = 0: the scan returns 4 rows in which every column is null, without an error. These are the same silent nulls as the ones that the length check of a8f178b prevents.

Could the checks of the batch and of each column also require n_children to be the number of fields of the type, and the child pointers to be non-NULL, before children reads them? With an n_children larger than the array of pointers, children would also read past that array before Arrow gets to see it.

[HyukjinKwon]

Thanks, fixed in 74aa0df. checkArrays now requires the batch, and each array at every level, to have as many children as its type has fields, with non-NULL pointers, before it reads them, so it does not read past the array of pointers either. The test library can now export batches without their columns (short=children), and reject the Arrow data that Spark cannot read checks that the read fails with The batch has 0 child arrays, but the schema has 7 columns.

[sarutak]

The nullability of the columns comes from schema of the library (NativeDataSource.schema), and this comparison ignores nullability, so nothing checks that the data agrees with the schema. I tried a library whose schema returns id as not nullable, while its stream declares id nullable and has nulls in 2 of 4 rows:

• collect() fails with java.lang.IllegalStateException: Value at index is null from BigIntVector.get, which does not say what is wrong.
• filter(isnull(col("id"))).count() returns 0 instead of 2, without an error, since the optimizer folds isnull of a non-nullable column to false.
• With -Darrow.enable_null_check_for_get=false, which @FelixYBW suggests on [SPARK-60101][SQL] Support native data sources implemented in Rust or C++ #59313 for performance, collect() no longer fails: it returns the values in the data buffer for the null rows (101 and 103 with my library), without an error.

Could Spark fail with NATIVE_DATA_SOURCE_ERROR when a column that the schema to read declares not nullable has a null_count other than 0 (it is in the array, so the check is cheap; a null_count of -1 needs a pass over the validity buffer)? Or make the inferred schema nullable. Either way, the header could say that the data must agree with the nullability of the schema.

[HyukjinKwon]

Thanks, fixed in 74aa0df. After importing a batch, Spark now checks that the data agrees with the nullability of the schema to read, at every level: a column, a struct field, an element or a map value that the schema declares not nullable must have no null where its parent is not null, and map keys none at all. It starts from the null counts of the vectors, so only a column that has nulls is scanned further. A null fails the read with NATIVE_DATA_SOURCE_ERROR, such as The column name has a null, but the schema to read declares it not nullable., and the header says that the data must agree with the nullability of the schema. I checked the data rather than making the inferred schema nullable, since a user can also declare a column NOT NULL in the schema to read, and the schema of a table in a catalog also applies to its writes. Test: reject the data that has nulls where the schema to read has none.

[sarutak]

Two literal encodings let a library that follows the header skip rows that match:

• A float literal is written as the shortest decimal of the float: col("f") == lit(0.1).cast("float") pushes {"type":"literal","dataType":"float","value":0.1}. A reader that parses JSON numbers as doubles, as most JSON parsers do, and widens the float column to compare it, finds 0.10000000149011612 = 0.1 false and skips the matching row. Could the header say that a literal must be read at the precision of its dataType, binary32 for float, and that the values of the integer types, dates and timestamps are exact 64-bit integers, which a double does not hold above 2^53? predicate-cases.json has no float literal, so it does not catch this.
• s.toString replaces invalid UTF-8 with U+FFFD: col("s") == expr("cast(X'FF' as string)") pushes "value":"\uFFFD". Spark compares the bytes, so a reader that compares EF BF BD with its statistics can skip a file whose strings contain the byte FF. Could a string literal that is not valid UTF-8 (!s.isValid) make the predicate not pushed, as for the other values that cannot be expressed?

[HyukjinKwon]

Thanks, both fixed in 74aa0df:

• A float literal is now written as the double of the same value, 0.10000000149011612 for 0.1f, which a reader that parses JSON numbers as doubles gets exactly, and which a reader that parses it as a float reads back as the same float. The header says to read a literal at the precision of its data type, and that the values of the integer types, dates and timestamps are exact 64-bit integers. predicate-cases.json has a float column now, with f = CAST(0.1 AS FLOAT) and f > CAST(0.1 AS FLOAT), and writes the floats of its rows the same way.
• A string literal that is not valid UTF-8 is no longer pushed, and the header says so.

[sarutak]

The rule for using a part of a predicate is only safe under AND and OR, but Spark also pushes predicates as children of = and <=>. (col("id") > 101) == (col("id") < 103) is pushed as {"type":"function","name":"=","children":[{..."name":">"...},{..."name":"<"...}]}, and (col("id") > 101).eqNullSafe(lit(True)) as <=> with a predicate child. Taking a part as true there can drop rows. For (id > 101) = (id < 101), the row with id = 101 matches (false = false), but a reader that takes id < 101 as true evaluates false = true and skips it.

Could the rule say that a reader may only take a part as true when every function between that part and the root is AND or OR, and must otherwise not use the predicate? predicate-cases.json could have one such case.

[HyukjinKwon]

Thanks, fixed in 74aa0df. The rule now says that a reader may only take a part as true when every function between that part and the root of the predicate is AND or OR, with your example, and predicate-cases.json has (id > 3) = (id < 3), which is true for the row whose id is 3.

[sarutak]

A package declares a single abiVersion, and Spark only uses a package whose abiVersion is the version of Spark, while the header lets a library implement several versions and choose one in spark_ds_init (spark_datasource.h L676). Once there is a version 2, a library that implements both cannot say so in its manifest. With "abiVersion": 2, Spark 4.4 reports UNSUPPORTED_ABI_VERSION without calling spark_ds_init, although the library would return a version 1 table. With 1, the field no longer means "the version that the library implements", as the docs say. Installed libraries have no manifest, so only packages have this problem.

The manifest of 4.4.0 is what all later packages have to work with, so could it be settled now? For example, the manifest could list the versions that the library implements, such as "abiVersions": [1], and Spark would use a package if it supports one of them. Or the docs could define abiVersion as the oldest version that the library implements, so that a later release can add a field for the others: parseManifest already ignores fields that it does not know, which the docs could also promise.

[HyukjinKwon]

Thanks, settled in 74aa0df with your second option: abiVersion is the oldest version that the library implements. Spark uses a package whose abiVersion is at most the newest version that it supports, and spark_ds_init returns a table of a version that both support. So a library that implements 1 and 2 declares 1, and a later Spark that supports only 2 and 3 still loads it, and gets a table of version 2. The docs promise that Spark ignores the fields of the manifest that it does not know, so a later release can add one, for example for the newest version. The manifest must also have a positive abiVersion, and names without U+0000.

[sarutak]

The Arrow types are defined "as in spark.sql.execution.arrow", which is internal and keeps growing, and NativeArrow.exportSchema has no allowlist, so types that the header does not describe reach the library. With a library that logs the schema that create_writer gets:

• a variant column arrives as a struct +s of two z fields, value and metadata, with the metadata variant=true on metadata;
• a time column arrives as ttn, with the metadata SPARK::time::precision=6;
• a day-time interval arrives as tDu.

The header does not tell a library what these mean, and once 4.4.0 is released, these encodings are part of version 1, also when ArrowUtils changes them for Python. Could the header list the Spark types of version 1 with their Arrow types, and could Spark reject the other types on the driver with NATIVE_DATA_SOURCE_ERROR before it passes a schema to the library, in create_reader and create_writer? A type could then be added later, with a capability.

[HyukjinKwon]

Thanks, done in 74aa0df. The header now lists the Spark types of version 1, with their Arrow types and formats: boolean, the integer types, float and double, decimal(p, s) as decimal128, string, binary, date, timestamp, timestamp_ntz, and array, map and struct. Spark fails an operation whose schema has another type with NATIVE_DATA_SOURCE_ERROR before it passes the schema to the library: the schema to read before create_reader and create_stream_reader, the data to write before create_writer, and a schema that schema returns. So variants, times and intervals do not reach a library, and a later version can add them for the data sources that declare a capability for them. Test: reject the types and the strings that a library cannot get, with a day-time interval in a read, and an array of variants in a write, which fails before create_writer.

[sarutak]

Nit: string copies value.getBytes(UTF_8), so a NUL in a string ends it early for the library: an option value "ab\u0000cd" reaches create as "ab", and "\u0000" as an empty string, without an error. Could Spark reject option keys and values that contain U+0000 with NATIVE_DATA_SOURCE_ERROR, or the header say what happens to them?

[HyukjinKwon]

Thanks, fixed in 74aa0df. create now fails with NATIVE_DATA_SOURCE_ERROR when an option key or value contains U+0000, and so does an operation whose schema has a column name with it. The data source names of a manifest cannot contain it either. The header says so.

[HyukjinKwon]

@sarutak Thank you for the thorough review, with a probe library. All the inline comments are addressed in 74aa0df, with replies there.

On Arrow UDFs: I agree that the root cause is in the columnar input path, and thanks for showing it with ArrowCachedBatchSerializer. I opened SPARK-60294 with #59551. It copies the pass-through columns of each batch into vectors of the task with VectorAppender, and releases them once the output batch that holds them is no longer used. Its test reproduces both failures without native data sources, with a test source that refills the same vectors for every batch. On your two questions:

• This PR waits for [SPARK-60294][SQL] Disable Arrow-backed columnar input to Python UDFs by default #59551, rather than sending its batches to the row queue path: with the fix, the batches of a native data source take the Arrow columnar path like the batches of other sources, and the Arrow cache is fixed too.
• Arrow UDFs on the columns of a native data source runs a UDF over one batch, two and 2,000, and checks the pass-through columns of every row against the result of the UDF. Locally on Java 26, it fails without [SPARK-60294][SQL] Disable Arrow-backed columnar input to Python UDFs by default #59551 and passes with it. The CI of pull requests skips it, since the job that runs the native data source suites has no pandas.

@zhztheplayer Thanks for the review.

• Native library infrastructure: I agree that discovery, platform selection, distribution, checksums and loading are not specific to data sources. I would keep them internal in this PR, because a general facility needs its own API: what a plugin declares in a manifest, how it looks up its libraries, and how the libraries of the sessions are isolated. They are already in classes separate from the data source integration (NativeDataSourcePackage, NativeLibraries and NativeDataSourceRegistry, apart from NativeDataSource), so a follow-up can extract them without changing the package format, once there is a second user such as Gluten. I'd be glad to work on that with you.
• Columnar writes: yes, a desirable follow-up, which @zhouyuan asked for too. The writers of the adapter already take Arrow-backed ColumnarBatches, so V2TableWriteExec could pass the batches of a columnar child to them directly.
• SPARK-57468 and [SPARK-57468][SQL] Add SparkPlan.convention() API to support different columnar batch types #59154: as @sarutak said, they are about the conventions of the batches and the transitions between operators, so the lifetime of the input batches of Python UDFs needs its own fix, [SPARK-60294][SQL] Disable Arrow-backed columnar input to Python UDFs by default #59551.

[gerashegalov]

The docs say continuous streaming is not supported, but nothing stops it: this table always reports STREAMING_WRITE, and Spark accepts any such sink with Trigger.Continuous. I ran rate into native_sink with Trigger.Continuous("1 second") on Java 25, and it ran:

• One stream writer was created and used for every epoch.
• When the query stopped, the data writers of the last epoch were closed without a commit or an abort. Their files were left in _temporary/.

The header promises a library that a data writer is always committed or aborted before it is closed, so a library can't rely on that here.

Could ColumnarStreamingWrite.createStreamingWriterFactory fail when the write runs in continuous mode (for example, when the local property ContinuousExecution.EPOCH_COORDINATOR_ID_KEY is set), with a test?

[HyukjinKwon]

Thanks, fixed in 840971d. ColumnarStreamingWrite.createStreamingWriterFactory now fails with the new DATA_SOURCE_CONTINUOUS_WRITE_NOT_SUPPORTED when the local property ContinuousExecution.EPOCH_COORDINATOR_ID_KEY is set, which only continuous processing sets, right before it runs the write. So Spark creates no writer, and a library is never left with data writers that are closed without a commit or an abort. Continuous reads already fail when the query starts, since a native scan does not report CONTINUOUS_READ. The header now says that streaming writes are micro-batch only, and the docs list continuous processing among the limitations. Test: continuous processing fails with a native data source writes rate to native_sink with Trigger.Continuous, and checks that the library created no writer.

[gerashegalov]

The docs suggest putting the version in the file name of a package. But if a session adds rs-1.0.sparkpkg and then rs-1.1.sparkpkg, every later query on that source fails with NATIVE_DATA_SOURCE_PACKAGE_CONFLICT, which says to remove all but one. A session has no way to remove an artifact, and adding rs-1.0.sparkpkg again with new contents fails with ARTIFACT_ALREADY_EXISTS. So the session is stuck until it is closed. I checked this on Java 25.

The same conflict happens when an admin's package in spark.sql.dataSource.native.paths and a user's added package provide the same name.

Could Spark choose one, for example the package the session added last, then the configured ones? Or else the docs could say that an upgrade needs a new session, and stop suggesting versioned file names.

[HyukjinKwon]

Thanks, fixed in 840971d with the first option, which is what Spark does for Python data sources: one registered again replaces the previous one, and a statically registered one, with a warning. The package that a session added last now takes precedence over the packages it added before and over the configured ones, and Spark logs once that it replaces them. So a session upgrades rs-1.0.sparkpkg by adding rs-1.1.sparkpkg, and a package that a user adds replaces an admin's package in spark.sql.dataSource.native.paths for that session. Two configured packages that provide the same data source still fail with NATIVE_DATA_SOURCE_PACKAGE_CONFLICT, whose message now names the configuration: they are files on the driver that can be removed, and their file names do not tell which one is newer. The docs describe the order and how to upgrade a package. Test: the package that a session added last replaces the other packages.

[gerashegalov]

This passes a column of any type, so s1 = s2 on two struct columns, and a1 < a2 on two array columns, are pushed to the library as they are. I checked both with V2ExpressionBuilder and toJson.

The rules in the header only cover single values. For example, they say comparisons with null are null. But Spark treats nulls inside structs and arrays as equal to each other and smaller than other values, so array(1, null) = array(1, null) is true. A library that follows the header can skip those rows, and Spark can't get them back by checking the predicates again after the scan.

Could comparisons and IN only be pushed for columns of simple types (null checks are fine for any type)? Or else the header could describe the rules for nested values. It could also say that binary values compare as unsigned bytes.

[HyukjinKwon]

Thanks, fixed in 840971d. Comparisons and IN are now only pushed when none of their operands is an array, a map or a struct, so s1 = s2 and a1 < a2 are no longer pushed. IS NULL and IS NOT NULL still are, for any type, and so are the comparisons of a field of a struct that has a simple type, such as s.x = 1. The header says both, and that binary values compare as unsigned bytes. predicate-cases.json now has a binary column, with cases for which unsigned and signed comparisons differ: b > X'7F' matches X'80', X'FF' and X'7F00'. Test: comparisons and IN on arrays and structs are not pushed.

[gerashegalov]

The checks here look at the offset and the children of the batch, but not its null_count. As far as I can tell, Arrow Java ignores the validity of the top-level struct when it imports a batch. So if a library marks some rows of the batch itself as null, Spark reads them as ordinary rows built from the child values, with no error. Arrow C++ rejects such a batch on import.

Could this also require null_count == 0, or -1 with no validity buffer? It is a constant-time check.

[HyukjinKwon]

Thanks, fixed in 840971d. checkArrays now rejects a batch whose struct marks rows as null, with The struct of the batch marks 1 of its rows as null, but Spark only reads batches without null rows: mark the values of the columns as null instead.

Two details differ from the suggestion. I checked them with ImportRecordBatch of Arrow C++ 24, on a struct of 2 rows:

null_count	validity buffer	Arrow C++	Spark now
-1	no nulls	imported	read
-1	first row null	ArrowArray struct has non-zero null count, cannot be imported as RecordBatch	rejected
1	first row null	same	rejected
-1	none	imported	rejected

• When the library did not count the nulls, with a null_count of -1, Spark counts them in the validity buffer, as Arrow C++ does, rather than rejecting a batch that has none. This reads length / 8 bytes, which the buffer must hold anyway.
• A null_count of -1 without a validity buffer is not allowed by the C data interface: "The buffer pointers MAY be null only in two situations: 1. for the null bitmap buffer, if ArrowArray.null_count is 0". Arrow Java takes it as a mix of nulls and values, and fails to import it with Could not load buffers for field ...: Cannot invoke "org.apache.arrow.memory.ArrowBuf.getReferenceManager()". So Spark now rejects it, for every array and not only the batch, with The batch has a null_count of -1 but no validity buffer, which the C data interface only allows with a null_count of 0.

The header now says both. The test library has a new option null_rows, and reject the Arrow data that Spark cannot read checks each row of the table, and a column with a null_count of -1 but no validity buffer.

[gerashegalov]

stream.release() runs after root.close() and dictionaries.close() in the same block, while task.close() is in the finally. If either Arrow close throws, the library never gets release, but the task context is freed anyway. A library thread that still checks task->is_interrupted then reads freed memory. The header says the task stays valid until the stream is released.

Unlikely, but easy to rule out: could the stream be released in its own finally, before task.close()?

[HyukjinKwon]

Thanks, fixed in 840971d: the stream is now released in a finally of closing the root and the dictionaries, and the task is closed after it, so a failure there no longer skips the release. The data writer was already in that order: close_data_writer is the first thing that its close does, before the task is closed.

[gerashegalov]

Spark calls the library with FFM, but it still exchanges Arrow data through arrow-c-data, which loads Arrow's own JNI library, libarrow_cdata_jni. That library is only shipped for Linux and macOS on x86_64 and aarch64, and for Windows on x86_64. On ppc64le, s390x or Alpine (musl), a correctly installed native data source fails on its first read, with an Arrow error that doesn't mention native data sources.

Could the docs list the supported platforms, and Spark fail early with a clear error elsewhere? The platform names in the manifest also can't tell glibc from musl, which will be hard to change once 4.4.0 is out.

[HyukjinKwon]

Thanks, fixed in 840971d:

• Before it loads a library, on the driver and on the executors, Spark now checks that the JNI library of the C data interface of Arrow Java loads, and otherwise fails with the new NATIVE_DATA_SOURCE_UNSUPPORTED_PLATFORM. The error names the platform and keeps Arrow's error as its cause: an IllegalStateException when Arrow Java has no library for the architecture, as on ppc64le or s390x, or an UnsatisfiedLinkError when the library does not link, as with musl.
• The docs list the platforms for which Arrow Java 19 ships the library: Linux with glibc and macOS on x86_64 and aarch64, and Windows on x86_64.
• In the manifest, linux-<arch> now means Linux with glibc, and Linux with musl is linux-<arch>-musl, as the Rust targets for musl end with -musl. Spark detects musl from the C library mapped in its own process (/proc/self/maps), rather than from files that a glibc system may also have, so it never picks a glibc build on musl, and a package can add a musl build once native data sources run there.

Tests: fail clearly where the C data interface of Arrow Java does not load, and the musl cases of platform names.