timsaucer opened a new issue, #1719: URL: https://github.com/apache/datafusion-python/issues/1719
`with_extensions` (#1679) and composable codecs (#1678) gave extension libraries a way to ship codecs and a query planner as one atomic bundle. What the repository still has no example of is the thing that machinery exists for: several independently-compiled libraries cooperating inside one distributed query, with plans actually leaving the process. This issue tracks building that example, and records the gaps found while validating that it is possible. Several of them are load-bearing, and one is an upstream blocker that the current examples only avoid by accident. ## The example Three extension libraries under `examples/distributed/`, plus a toy distributed engine that runs real OS-process workers: - **A UDF library** — scalar UDF, UDAF and UDWF resolved by name, with a logical codec whose payload is empty. Deliberately does **not** implement `__datafusion_session_components__`, so it is installed the old way with `register_udf` plus `with_logical_extension_codec`. This is the mixed-workflow case, and it is currently the honest one: `SessionExtensionComponents` has no UDF field yet. - **A storage library** — a Parquet-directory table provider reporting one output partition per file, with a physical codec that emits genuinely portable bytes rather than a process-local token. There is no such codec anywhere in the repository today (see G9). - **An engine library** — stage-splitting query planner, its own `StageExec` and `ShuffleReadExec` nodes, both codecs, and a mixed Rust/Python maturin package carrying the driver, the worker entry point, and the shared session factory. Workers are separate processes spawned through `sys.executable`; results come back as Arrow IPC files, one per `(stage, partition)`. Queries run against TPC-H `lineitem` at SF 1, which CI already generates. One query uses an inline Python UDF defined in the driver's `__main__`, paired with the case that breaks — the same UDF defined in an importable module the worker cannot import. The existing `datafusion-ffi-query-planner-example` is retired into the engine library; `datafusion-ffi-example` stays as the protocol-conformance test bed. ## Gaps found Every item below was reproduced against a built extension, not inferred. ### Blocked upstream **G1 — a stock node that crosses FFI cannot be serialized on the query-planner return path.** `FFI_QueryPlanner` returns proto bytes rather than a plan handle, so every query serializes the physical plan. The plan contains a `CooperativeExec` inserted by the always-on `EnsureCooperative` rule, which runs on the *host* during a foreign planner's `create_physical_plan` and so arrives inside the library as an opaque `ForeignExecutionPlan`. That type has no reachable `try_to_proto`, so the native encoder never runs and a perfectly serializable node becomes unserializable purely by having crossed the boundary. **G2 — the greedy codec claim in `datafusion-ffi-example` is a symptom of G1, not sloppiness.** Its physical codec claims `node.is::<ForeignExecutionPlan>()`, which takes every other library's nodes, and `extension-guide/checklist.md` tells authors never to do this. It is nevertheless load-bearing: narrowing it to `DataSourceExec` alone makes **31 of the 51 tests** in `datafusion-ffi-query-planner-example` fail, every one on the `CooperativeExec` node from G1. The arm has to stay until G1 is fixed. A planner that controls its own physical optimizer rules never sees a foreign node and needs no such arm — which is how the new engine library will avoid it. **G7 — a codec reached over FFI gets a `TaskContext` with no object stores and no catalog.** `FFI_TaskContext` is rebuilt with `RuntimeEnv::default()`, so decode-time object-store resolution — which `ParquetSource::try_from_proto` performs — can only ever see `file://`, and no codec can resolve a table by name at decode time. ### datafusion-python **G3 — no way to read a plan's partitioning scheme.** Only `partition_count` was exposed, so a driver could not distinguish hash-distributed output from merely counted output, nor read the hash keys. Fixed by `ExecutionPlan.output_partitioning`. **G4 — `SessionContext.execute` did not bounds-check the partition index.** The plan's leaves index their partition vector directly, so an out-of-range index panicked and surfaced as `index out of bounds: the len is 2 but the index is 5`, naming neither the plan nor the index. Fixed. **G5 — `SessionConfig.set` raised `PanicException`.** It routed through `SessionConfig::set_str`, which unwraps, so an unknown namespace aborted rather than raised — and `PanicException` derives from `BaseException`, escaping `except Exception`. Fixed on the Python side; the upstream `unwrap` remains. **G6 — there is no session-config snapshot or restore, and nothing documents what a worker has to match.** `SessionConfig` and `RuntimeEnvBuilder` are write-only from Python. `information_schema.df_settings` is readable but not replayable: it lists `datafusion.runtime.*` keys that have no `ConfigOptions` namespace, so a naive replay loop hits G5 on its first row. Worker parity has to be hand-maintained, and the checklist for doing so does not exist yet. **G10 — the example planner emits an invalid plan on multi-partition input.** `DistributedQueryPlanner` wraps `GlobalLimitExec` *after* optimization, so nothing inserts the coalesce it requires: `Assertion failed: self.input.output_partitioning().partition_count() == 1 (left: 2, right: 1)`. Latent because every test for it uses a single-partition input. ### Documentation **G8 — two `ExecutionPlan` docstrings claimed memory-backed tables cannot be serialized.** True of `LogicalPlan`, whose `try_encode_table_provider` has no arm for one; false of the physical layer, which inlines the batches. Verified by decoding on a context sharing nothing with the encoder and executing. Fixed. **G9 — `extension_codec_durable_metadata` has no reference implementation.** The guide tells authors to encode durable metadata and states that the in-repo examples deliberately do not. Nothing in the repository shows one that does. The new storage library becomes that reference. ## Validated, for the record The design depends on these, so each was checked rather than assumed: Parquet and memory-backed physical plans round-trip through `to_bytes`/`from_bytes` on a fresh context with no codecs installed; `partition_count` survives the FFI boundary; a `__main__`-defined Python UDF ships by value to a genuinely unrelated process; codec-id dispatch is order-independent and names the missing id on failure; and a query planner that wraps the foreign session with its own physical optimizer rule list produces a plan free of `ForeignExecutionPlan`, which another process then decodes and executes. ## PRs - [ ] Report physical partitioning, and stop two panics escaping as panics — G3, G4, G5, G8, and the comment for G2 - [ ] The multi-library distributed example itself — G6, G9, and the worker-parity documentation - [ ] Upstream issues against `apache/datafusion` for G1, G5's `unwrap`, and G7 -- 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]
