timsaucer commented on PR #1679:
URL:
https://github.com/apache/datafusion-python/pull/1679#issuecomment-5544543804
Thanks for chasing this down — the `write_*` failures are real, but I think
the fix is much smaller than a new planner API, and it is not in this PR.
**Where it actually breaks.** `FFI_LogicalExtensionCodec` has no vtable
entries for file formats at all. Both hooks are hardcoded stubs in
`datafusion-ffi` (`src/proto/logical_extension_codec.rs`,
`try_decode_file_format` / `try_encode_file_format` → `not_impl_err!("FFI does
not support ...")`). So *any* codec that crosses the FFI boundary loses
file-format support, and `df.write_csv/write_parquet/write_json` build a
`LogicalPlan::Copy` that carries a `FileFormatFactory`. That is the whole
failure.
**This PR already survives it on the datafusion-python side.**
`PythonLogicalCodec::try_{encode,decode}_file_format` go through
`chain_encode`/`chain_decode`, which collect a chained codec's error and fall
through to the terminal `DefaultLogicalExtensionCodec`. Built-in formats encode
and decode fine there. The direction that fails is the mirror image: Ballista
holds an `FFI_LogicalExtensionCodec` wrapping the Python session's codec and
calls it directly, with no Default fallback behind it.
**Two fixes, both much cheaper than a new API:**
1. *Ballista side, today, no upstream change.* Compose the FFI codec you get
from Python with a `DefaultLogicalExtensionCodec` fallback on the two
file-format hooks — the same shape `PythonLogicalCodec` uses here. One file,
unblocks DF 55.
2. *Upstream `datafusion-ffi`, ~8 lines.* Make those two hooks delegate to a
local `DefaultLogicalExtensionCodec` instead of returning `not_impl_err!`. No
new vtable entries, so no ABI break, so it can ship in a 55.x patch. It works
because built-in file formats are fully self-describing protobuf and nothing
has to cross the boundary:
`DefaultLogicalExtensionCodec::try_encode_file_format` in `datafusion-proto`
covers csv/json/arrow/avro/parquet by downcasting the factory locally. What is
left unsupported is a *custom* `FileFormatFactory` owned by the foreign
library, and I agree that one needs a real `FFI_FileFormatFactory` and will not
make the 55 window — but it is also not what is breaking `write_*`.
**On `CallbackPlanner`.** I do not think the concept holds up, and I want to
be concrete about why rather than just deferring it.
A `QueryPlanner` returning a `CallbackExec` leaf replaces the *entire* plan
with one opaque node: no children, no partitioning DataFusion can see, no
pushdown, no repartition, no limit, no statistics. That is not planning, it is
interception at the root. It is coherent for exactly one architecture — thin
client, full delegation — which is what Ballista's `DistributedExec` already
is; the local plan is degenerate because none of the work is local. But it has
no answer for a hybrid plan, a local table joined against a remote one, because
there is nothing left to split on. All or nothing.
It also buys very little over what already ships:
```python
plan_bytes = df.logical_plan().to_bytes(ctx)
batches = my_grpc_client.execute(plan_bytes) # your gRPC logic, in Python
result = ctx.from_arrow(batches) # any __arrow_c_stream__
object
```
All three exist today, no new API. The only thing `CallbackPlanner` adds on
top is transparency — the user keeps writing `ctx.sql(...).collect()` and never
sees the interception. That is real, but it is a thin return for a permanent
public API, and a second planner mechanism sitting next to the FFI query
planner means two ways to do one thing.
If the goal is *partial* delegation rather than whole-plan delegation, the
design that answers it already exists: a table provider exported from Python
(`__datafusion_table_provider__`, see
`docs/source/user-guide/io/table_provider.md`). Remote data appears as a table,
DataFusion plans around it, filter and projection pushdown work, and local and
remote sources mix in one query. That composes; a root-level callback does not.
The part that makes me most hesitant is the stated rationale: *"As
`CallbackPlanner` have same library marker as df python, `FFILogicalCodec` will
not be triggered."* That shapes a public API around which cdylib the code is
compiled into, in order to route around an eight-line gap in `datafusion-ffi`.
Fixing the gap is the smaller and more durable change.
**One genuinely open question, independent of all of the above** — and I
think it is the same "chaining of codecs" point you flagged. Once Ballista's
codec is installed on the Python session it becomes a chain entry, and chain
entries write a framed payload (`DFPYCHN` + codec id) rather than bare bytes. A
plain Rust `BallistaCodec` in the scheduler will not strip that envelope. So
how does the scheduler decode — does it link `PythonLogicalCodec`, or does
Ballista's codec stay off the chain? I am happy to add whatever hook makes the
first option workable; I would rather solve that than work around it.
Proposal: keep #1679 as is, since the file-format gap predates it and is
orthogonal, and I will open the upstream issue for (2). If you still want the
callback route after the above, let us give it its own issue so the design can
be argued on its own terms.
--
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]
