HyukjinKwon opened a new pull request, #59498: URL: https://github.com/apache/spark/pull/59498
### What changes were proposed in this pull request? This is a draft for comparison with #59424, following [@tdcmeehan's proposal](https://github.com/apache/spark/pull/59424#issuecomment-6173067539) on how the interface of native data sources grows. It is #59424 up to 51af65c, with its C interface, a table of typed functions, replaced by a frozen C core and versioned Protocol Buffers messages. Everything else is the same as in #59424: finding the libraries, the packages, the Data Source V2 adapter, the Arrow data path, the checks of the data, the docs and the tests. **The C core**: `spark_datasource.h` declares six functions, which never change: ```c int32_t spark_ds_abi_version(void); const char* spark_ds_last_error(void); int spark_ds_open(const SparkDsHost* host, const uint8_t* request, int64_t length, SparkDsSink reply, void* sink, void** handle); int spark_ds_call(void* handle, const uint8_t* request, int64_t length, SparkDsSink reply, void* sink); int spark_ds_stream(void* handle, const uint8_t* request, int64_t length, struct ArrowArrayStream* in, struct ArrowArrayStream* out); void spark_ds_close(void* handle); ``` and `SparkDsHost`, which starts with its size and only gets fields appended, with the callbacks of Spark: `log`, which writes to the log of the driver or the executor, and `is_interrupted(task_attempt_id)`, which tells a read or a write that its task was killed. **The messages**: everything else is in `spark_datasource.proto` (`sql/core/src/main/protobuf/org/apache/spark/sql/datasource/`, also copied to the `include` directory of a distribution): - `OpenRequest` opens a data source, a batch reader, a stream reader, a writer, a partition reader or a data writer, with everything that it needs: the name and the options of the data source, the schema, the write mode and the ID of a streaming query, the `TaskContext`, or the serialized state and partition. It also carries the version of Spark and of the messages. The `OpenReply` of a data source declares its capabilities, and Spark reports the operations that a data source does not declare as unsupported, without opening a reader or a writer. - `CallRequest` and `CallReply` carry every control operation: `InferSchema`, `PushPredicates` (with the answer of the reader for each predicate, `PredicateUse`), `PushLimit`, `PruneColumns`, `PlanPartitions`, `Serialize`, `InitialOffset`, `LatestOffset`, `PlanStreamPartitions`, `CommitOffset`, `Commit`, `Abort`, `CommitDataWriter` and `AbortDataWriter`. An empty reply is valid for every request: a reader without pushdown replies to `PushPredicates` with nothing. - `StreamRequest` exchanges the data, as Arrow C streams, without copying: `Read` initializes the output stream of a partition reader, and `Write` gives a data writer the data of its task as an input stream, which the library reads during the call, on the thread of the task. Spark exports the rows of the task as one stream, from `DataWriter.writeAll`, in batches that end at `spark.sql.execution.arrow.maxRecordsPerBatch` rows or `spark.sql.execution.arrow.maxBytesPerBatch` bytes. - Schemas are Spark SQL types, as in Spark Connect, so that every control message can be forwarded unchanged to a library that runs in another process, for example to isolate it, while the data is converted between Arrow C streams and Arrow IPC. - The messages use `optimize_for = LITE_RUNTIME`. With the full C++ runtime, the generated code registers `spark_datasource.proto` in a pool of the process: the C++ example of the docs, linked to the libprotobuf of Homebrew, aborted the JVM with `File already exists in database: spark_datasource.proto` when Spark loaded a second copy of it, the library of a package after the installed one. Lite messages register nothing. Libraries built against different versions of Protocol Buffers still conflict if they link it dynamically, so the docs ask for static linking, with hidden symbols. **Compatibility**: the six functions and the fields of `SparkDsHost` never change. The messages only grow: a later release may add fields, messages and values of enums, but never changes or reuses the number of a field, both sides ignore the fields that they do not know, and Spark only sends a new kind of request to an object whose `OpenReply` declares that it supports it. `SPARK_DS_ABI_VERSION` stays 1. **How Spark calls the libraries**: `NativeLibrary.java` binds the six functions by name with the FFM API, as #59424 does, checks the ABI version before it binds the others, allocates one `SparkDsHost` for each library, and never throws from the upcalls (`SparkDsSink`, `log` and `is_interrupted`). The tasks that use a library are registered by their attempt ID for `is_interrupted`. `NativeDataSource.scala` builds the requests and reads the replies, and `NativeSchemas` converts the schemas. The columnar adapter gets `ArrowStreamDataWriter` and `ArrowRowReader`, so that a native data writer gets the input of a task as one Arrow stream; an error of the rows, which the stream only reports as a message, is rethrown as it is. | | #59424 (function table) | This PR (messages) | |---|---|---| | What a library exports | `spark_ds_init`, which returns a table of 29 functions | 6 functions, which never change | | What grows | functions appended to the table, fields appended to its size-prefixed structs | fields and messages of `spark_datasource.proto` | | What checks it | the C, C++ or Rust compiler, for each function | the generated code of the messages; the compiler checks the 6 functions | | What a library needs | the header | the header, the `.proto`, and a Protocol Buffers compiler and runtime, such as prost or protobuf C++ | | Libraries in one process | share nothing but the header | each has its own Protocol Buffers runtime: lite messages, linked statically with hidden symbols for C++ | | A host in another process | maps each function to a message of its own | forwards the control messages unchanged | | The binding of Spark | a method handle for each function, and the structs | 6 method handles; the requests and replies in Scala | | The C++ test library, without dependencies | fills the table | a minimal Protocol Buffers codec written by hand, of about 200 lines | ### Why are the changes needed? To compare the two shapes of version 1 of the interface, which library binaries get built against, before Spark 4.4.0: see the discussion in #59424. A library that runs in another process, as the UDF worker protocol (SPARK-55278) does for UDFs, could get the same control messages, while #59424 would need a message for each of its functions. ### Does this PR introduce _any_ user-facing change? The same as #59424, with this interface: native data sources implement `spark_datasource.h` and `spark_datasource.proto`, which a Spark distribution contains in its `include` directory. ### How was this patch tested? All the tests of #59424, adapted to the messages, with the C++ test library implementing them with its own minimal Protocol Buffers codec, so that it still needs nothing but a C++ compiler. New tests: - `ignore the fields of messages that they do not know`: Spark ignores a field of a later version in every reply of a library, and a library ignores a field of a later version in a request. - `write the data of a task in one stream`: a data writer gets the 10 rows of its task in one stream of 4 batches, with `spark.sql.execution.arrow.maxRecordsPerBatch` at 3. - `report the errors of the data that a library writes`: a UDF that fails while the library reads the rows fails the write with the error of the UDF, not with the error that the library returns, and the write is aborted. - `reject the libraries that do not export the functions of the interface`: a library without `spark_ds_open`. Locally on macOS (osx-aarch64), with Java 26 (Spark built with Java 17): `NativeDataSourceSuite`, `NativeDataSourceClusterSuite` (executors on Java 26 too), `ColumnarDataSourceSuite` and `NativeDataSourceConnectSuite` pass, and with Java 17, `NativeDataSourceSuite` passes with the tests that load a library skipped. The Rust example of the docs (arrow-rs 58, prost 0.14) and the C++ example (Arrow C++ 24, protobuf-lite 35) were built with the documented commands and read by Spark on Java 26: installed and in a package, with errors of the library reported as `NATIVE_DATA_SOURCE_ERROR`. The C++ test library builds without warnings with Apple clang (`-Wall -Wextra`) in all its variants. `checkstyle` and `scalastyle` of the changed modules pass. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5.5) This pull request and its description were written by Isaac. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
