gengliangwang opened a new pull request, #59726: URL: https://github.com/apache/spark/pull/59726
### What changes were proposed in this pull request? Add an experimental C++ SDK for batch and micro-batch streaming reads, and batch and streaming writes, over the native data source C ABI in #59424, pinned to `74aa0dffc53463034fa6549796c4dd599e9705de`. It is the C++ counterpart of the Rust SDK in #59520, with the same examples: a Range source, which also supports streaming reads, and a write-only JSONL sink, which also supports exactly-once streaming writes. This draft depends on #59424; the framework changes are not included here. The SDK is header-only C++20 on [Arrow C++](https://arrow.apache.org/docs/cpp/) and [nlohmann/json](https://github.com/nlohmann/json). Each data source library compiles it, so no C++ type crosses the boundary of a library: Spark sees only the function table of `spark_datasource.h`, an unmodified copy of the pinned header. The API follows the Rust SDK, with its traits as abstract class templates, its associated types as template parameters and type aliases, and its default methods as virtual methods with defaults: - `DataSource` handles construction, with `static arrow::Result<S> Make(const CreateInfo&)`, schema discovery, optional per-instance capability restrictions, and an optional per-JVM `Init`. Its type aliases `ScanBuilder`, `MicroBatchStream`, `BatchWrite`, and `StreamingWrite` are `Unsupported` by default; a source redeclares those of the operations that it implements, each with its `New*` method. - `ScanBuilder<Factory>` negotiates predicates, limit, and projection, and its `Plan()` returns a `ScanPlan<Factory>` of typed partitions and a serializable `PartitionReaderFactory<Partition>`. The factory opens a reader, any `arrow::RecordBatchReader`, from a `ReaderContext`, which carries the negotiated schema and the task. - `MicroBatchStream<Offset, Factory>` returns typed offsets and plans each micro-batch between two of them as a `ScanPlan`, whose factory the executors use as that of a batch scan. - `BatchWrite<Factory>` and `StreamingWrite<Factory>` coordinate writes on the driver. `CreateWriterFactory()` returns a serializable `DataWriterFactory<Message>`, which opens a `DataWriter<Message>` for each task attempt from a `WriterContext`. Driver commit receives the winning message of each partition at its position; abort receives `std::vector<std::optional<Message>>`, with nothing for a message that is missing or cannot be decoded. A streaming write gets the epoch ID in `Commit` and `Abort`, and its task writers in `WriterContext::epoch_id`. ```cpp class RangeSource : public spark::native::DataSource { public: using ScanBuilder = RangeScan; using MicroBatchStream = RangeStream; static arrow::Result<RangeSource> Make(const spark::native::CreateInfo& info); arrow::Result<std::shared_ptr<arrow::Schema>> InferSchema() const override; arrow::Result<RangeScan> NewScan(const spark::native::ScanInfo& info) const; arrow::Result<RangeStream> NewMicroBatchStream( const spark::native::MicroBatchStreamInfo& info) const; }; SPARK_EXPORT_DATA_SOURCE(RangeSource); ``` The type aliases determine the native callbacks and default capabilities, so the macro takes only the source type, and it checks the declarations when the library compiles: each operation must derive from its class template and have its `New*` method, and the serialized types must be serializable, or the compiler reports the requirement as the first error. An `Unsupported` operation has null callbacks and no capability. A write opts into truncation with `static constexpr bool kSupportsTruncate = true;`. `capabilities()` defaults to `Capabilities::All()` and can only narrow the implemented operations; as in the Rust SDK, the one `TRUNCATE` flag of the ABI is declared only while a write that supports truncation remains. Method inputs are structs with constructors (`CreateInfo`, `ScanInfo`, `MicroBatchStreamInfo`, `WriteInfo`, `StreamingWriteInfo`, `HostInfo`, `ReaderContext`, and `WriterContext`), so new ABI fields can reach connectors without breaking them. The SDK owns native handles, serialization, Arrow import and export, schema and nullability validation, and the conversion of errors and exceptions. Methods return `arrow::Status` or `arrow::Result`, whose message Spark reports, with the detail of an Arrow error; an exception that connector code throws is reported as the error of the call, so no exception crosses the ABI. Factories, partitions, offsets, and messages are serialized with nlohmann/json, for example with `NLOHMANN_DEFINE_TYPE_NON_INTRUSIVE`; the SDK wraps a factory with the negotiated schema, as Arrow IPC, and the kind of operation, batch or streaming, so that the shared `read` and `create_data_writer` entry points restore the right factory. Offsets are compact single-line JSON without an envelope, and nlohmann/json orders object keys, so a position serializes deterministically. Reads retain the framework's planner/reader separation: driver planning and failures are cached, a builder that fails negotiation is destroyed so that it is never reused, and executor reads restore only serialized values. The schema of a reader is the schema of its stream, so it can use any supported encoding, such as `large_utf8` for a string; the SDK checks it against the negotiated schema before exporting the stream, and checks each batch: its schema, `RecordBatch::Validate()`, since the C data interface does not carry buffer sizes for Spark to check, zero offsets at every level, and nullability, including nested fields. The SDK parses each pushed predicate on its own, so one that it cannot parse stays unsupported, `Predicate::literal_value()` decodes literals by their Spark SQL type, and a limit is declined after a residual predicate. Writes follow the lifecycle of the Rust SDK: factory creation and serialization are cached, failed task or driver commits remain abortable, succ essful task messages and terminal failures are cached to prevent repeated side effects, and batch writes reject epoch IDs other than `-1`. A failed stream call stops the query, so the SDK returns that failure from every later call on the stream. `spark::native::Log` writes to the driver or executor log through Spark at `kInfo` and above by default, and `TaskContext` reads Spark's live cancellation, rejects a context shorter than version 1 of the ABI, and has `TaskContext::ForTesting()` for tests outside Spark. The standalone CMake project builds the SDK, its tests, and the examples, and does not change Maven/SBT or require C++ to build Spark. `spark_add_native_data_source(<name> <sources>)` builds `libspark_datasource_<name>.so` with hidden visibility and a linker version script, so that a library exports only `spark_ds_init`: hidden visibility alone left about 115 template instantiations of Arrow and of the standard library exported, which other libraries of the JVM process could interpose. The Range and sink examples behave as those of #59520; the Range source keeps its selection within the range with 64-bit arithmetic, so its bounds never overflow at the limits of `int64_t`. The C client of the Range library is the C test of #59520, with only the name of the source changed. ### Why are the changes needed? C++ is the other target language of #59424, which supports native data sources implemented in Rust or C++. C++ connectors should implement planning, reading, and transactional write coordination without manually managing C function tables, opaque handles, serialized messages, Arrow ownership, or task-context lifetimes, and without letting an exception cross the ABI, which would crash the JVM. A C SDK would add little over `spark_datasource.h` itself, since C has no RAII, templates, or exceptions to make these safer. Arrow C++ is the Arrow implementation that native C++ engines and formats such as Parquet already use, so a connector can return any `arrow::RecordBatchReader` and receive `arrow::RecordBatch` inputs directly. Keeping the API shape of the Rust SDK, with the same lifecycle guarantees, lets the two SDKs be reviewed and evolved together. ### Does this PR introduce _any_ user-facing change? Yes. It adds an experimental, unpublished C++ SDK with batch and micro-batch streaming read APIs, and batch and streaming write APIs that append or truncate, plus runnable `cpp_sdk_range` and `cpp_sdk_sink` examples. This is a prototype API, not a published compatibility commitment. Running the examples in Spark requires a distribution containing #59424 and Java 22 or later. Existing Spark behavior is unchanged. Continuous processing and admission control for streaming reads are not implemented: version 1 of the ABI does not provide them. ### How was this patch tested? From the repository root, with GCC 10.5, CMake 3.16, Arrow C++ 26 (the libraries and headers of the pyarrow 26 wheel), and nlohmann/json 3.11.3 on Linux x86-64: ```sh cmake -S sql/native/cpp -B sql/native/cpp/build -DCMAKE_BUILD_TYPE=Release cmake --build sql/native/cpp/build (cd sql/native/cpp/build && ctest --output-on-failure) ``` All 8 test executables passed, in Release, in Debug with `-Werror`, and with AddressSanitizer and UndefinedBehaviorSanitizer, which reported nothing: 84 C++ test cases and the C test. The tests port those of the Rust SDK and its examples: - `sdk_test` (39): error and exception conversion and thread-local errors, sized ABI structures, task contexts and live cancellation, capabilities, predicates and typed literals, base64, serialization and its version, single-line offsets, and the exported Arrow stream: early release, empty batches, alternative encodings, errors and exceptions, schema mismatch, unsupported nested types, offsets, short columns, nested nullability, and batches without columns. - `library_test` (1), `read_test` (12), `batch_write_test` (8), and `streaming_test` (4) each export a data source and call its actual function table through `spark_ds_init`. They cover initialization and its failure, host logging and its levels, the source name, narrowed, unimplemented, and failing capability declarations, cached plans and planning failures, discarded builders after failed negotiation, predicates that do not parse, limits after residual filters, independent planners, executor retries from serialized values only, the rejection of a reader that does not return the negotiated schema, factory and message round trips, retained Arrow buffers, live cancellation, failed writes and commits, factory creation and serialization failures, missing and undecodable messages, repeated callbacks, epoch validation, the state of each planned micro-batch, reads that restore batch or micro-batch factories of different types, exceptions that fail a stream, and the query and epoch IDs o f streaming writes. - `range_source_test` (9) and `json_sink_test` (11) port the tests of the examples, and `range_c_abi_smoke` runs the C client against the Range library. Both libraries export only `spark_ds_init` (`nm -D --defined-only`). The code is formatted with clang-format and the new `.clang-format`, and passes the line-length and ASCII checks. Both Spark smoke scripts passed with `spark-submit --master 'local[2]'`, with Debug and Release builds of the libraries packaged by `package.py`, on Spark built with Maven and Java 25 from the pinned framework commit. The Range script checked reading and pushdowns, early rejection of unsupported writes, and a streaming read with `Trigger.AvailableNow` into a memory sink, which returned all 31 rows of its range. The sink script checked append, overwrite, failed-overwrite cleanup, analysis-time rejection with writes disabled, and a streaming write that ran three times from one checkpoint: the first run committed epoch 0, a restart without new input left the manifest unchanged, and a restart with new input committed epoch 1. The only errors in either log come from the sink's intentional input failure. **Not run:** macOS, whose exported-symbol linker option is untested, and Windows; Clang; Arrow versions other than 26; Java versions other than 25; and cluster deployments, whose executors load the library and Arrow C++ in their own JVMs. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Isaac (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]
