tdcmeehan commented on PR #59424:
URL: https://github.com/apache/spark/pull/59424#issuecomment-6173067539
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 #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:
```c
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`](https://github.com/apache/spark/blob/4ccf07b6d91e2b126b7001741778ee28f22c6609/.github/workflows/build_and_test.yml#L834-L838)
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](https://github.com/apache/arrow/blob/910292405bb82ccbf057b9801f5e4b922ea23788/format/Schema.fbs#L19-L26),
and its [`Feature`
enum](https://github.com/apache/arrow/blob/910292405bb82ccbf057b9801f5e4b922ea23788/format/Schema.fbs#L54-L81)
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`](https://github.com/apache/datafusion-comet/blob/7d8d9b1544b734c85a93cca2cc9d5380ad70accd/spark/src/main/scala/org/apache/comet/Native.scala#L61-L79),
as protobuf, and a [contrib
envelope](https://github.com/apache/datafusion-comet/blob/7d8d9b1544b734c85a93cca2cc9d5380ad70accd/native/proto/src/proto/operator.proto#L49-L81)
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](https://github.com/apache/gluten/blob/91ae3b22fcb1dbff6937ae517fc0d816c23f4501/gluten-arrow/src/main/java/org/apache/gluten/vectorized/PlanEvaluatorJniWrapper.j
ava#L46-L82) whether it supports a plan.
- **Spark's UDF worker protocol** (SPARK-55278) is protobuf, with
[`Init.protocol_version`](https://github.com/apache/spark/blob/4ccf07b6d91e2b126b7001741778ee28f22c6609/udf/worker/proto/src/main/protobuf/org/apache/spark/udf/worker/udf_message.proto#L162-L167)
and
[`WorkerCapabilities`](https://github.com/apache/spark/blob/4ccf07b6d91e2b126b7001741778ee28f22c6609/udf/worker/proto/src/main/protobuf/org/apache/spark/udf/worker/worker_spec.proto#L113-L135)
that Spark checks during planning.
- **protoc plugins:** the compiler sends its version in the request, and the
plugin answers with
[`supported_features`](https://github.com/protocolbuffers/protobuf/blob/cd68a2eab18e9646bf746182fa3b858c6d9caa7e/src/google/protobuf/compiler/plugin.proto#L79-L115).
- **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](https://github.com/openssl/openssl/blob/ed57129cd43b77e077c68fd5f7ea951976766f38/README-FIPS.md?plain=1#L36-L41)).
- **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](https://github.com/apache/arrow-adbc/blob/fe88b72a9917763a1ff2eb694341b5b6da8aa6f4/c/include/arrow-adbc/adbc.h#L1055-L1074)
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](https://github.com/apache/datafusion/blob/0d79252f61bd8777990b2749c216ffd07079c30d/datafusion/ffi/README.md?plain=1#L38-L40)
"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.
--
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]
