adriangb opened a new pull request, #26824: URL: https://github.com/apache/datafusion/pull/26824
## Which issue does this PR close? - Precursor for #26555 and #26807. Related: #24331, #24631, #24628. ## Rationale for this change The `DataSource`, `FileSource` and `DataSink` proto hooks (added in 55) return a full `PhysicalPlanNode`, which is the Exec wrapper. Decode in #26555 / #26807 works at the source level. So: | Problem | Result | |---|---| | A source must build its own wrapper | Public helpers (`extension_data_source_node`); a source can return any plan variant | | Built-in sources use their own plan variants | Two ways to serialize a source ([thread](https://github.com/apache/datafusion/pull/26555#discussion_r4336903193)) | | No file-source level | A custom `FileSource` needs a marker `DataSource` | | No source level on the wire | An unregistered reader gives an unclear error, or decodes the wrong source | This PR adds a proto layer that mirrors the traits. Then #26555 / #26807 only add a decoder for the `extension` arms. ```text before after ────── ───── PhysicalPlanNode PhysicalPlanNode ├─ parquet_scan = 1 (deprecated) ├─ data_source = 42: DataSourceExecNode ├─ csv_scan = 2 (deprecated) │ └─ DataSourceNode (oneof) ├─ avro_scan = 17 (deprecated) │ ├─ file: FileScanNode ├─ json_scan = 31 (deprecated) │ │ ├─ base: FileScanExecConf ├─ memory_scan = 35 (deprecated) │ │ └─ FileSourceNode (oneof) ├─ arrow_scan = 38 (deprecated) │ │ ├─ parquet/csv/json/avro/arrow ├─ json_sink = 24 (deprecated) │ │ └─ extension {name, payload} ├─ csv_sink = 28 (deprecated) │ ├─ memory └─ parquet_sink = 29 (deprecated) │ └─ extension {name, payload} └─ data_sink = 43: DataSinkExecNode {input, sort_order} └─ DataSinkNode (oneof) ├─ json/csv/parquet └─ extension {name, payload} ``` **Timing:** land before 56.0.0 (#24461), together with #26555 / #26807. Then the 55 hooks change only once, through deprecation. ## What changes are included in this PR? | Area | Change | |---|---| | Proto | New messages (see above). The new format messages keep the field tags of the legacy scan nodes, so the legacy → new translation only moves fields. | | Encode | Built-ins always write the new form. | | Decode | `datafusion/proto/src/physical_plan/legacy.rs` translates the 9 legacy variants into the new form. Then only the new path decodes. | | New hooks (default `Ok(None)`) | `DataSource::try_to_proto_source → DataSourceNode`<br>`FileSource::try_to_proto_file_source → FileSourceNode`<br>`DataSink::try_to_proto_sink → DataSinkNode` | | Deprecated (not removed) | The 55 `try_to_proto` hooks and the old public decode functions. Their defaults wrap the new hooks. | | Encode order | new hook → deprecated hook → `PhysicalExtensionCodec` | | `extension` arms | Decode returns `No decoder for extension DataSource '{name}'`. The registry is in #26555 / #26807. | Suggested shape for #26555 after this PR: ```rust pub trait ExtensionDataSourceFromProto: DataSource + Sized { const NAME: &'static str; fn try_from_proto(payload: &[u8], ctx: &ExecutionPlanDecodeCtx<'_>) -> Result<Arc<dyn DataSource>>; fn register_proto(registry: &mut ProtoDecoderRegistry) -> Result<()> { /* provided */ } } ``` ## What is the testing strategy for this PR? `datafusion/proto/tests/cases/plans/legacy.rs`: | Test | Checks | |---|---| | `built_ins_roundtrip_in_layered_form` | Each built-in source/sink round-trips in the new form | | `legacy_variants_decode_to_same_plan`, `legacy_variants_reencode_byte_identical` | Each legacy variant decodes, then re-encodes byte-identical to the new form | | `legacy_scans_nested_in_other_plans` | Legacy nodes under Projection / Repartition / Union | | `truncated_or_garbage_bytes_error_without_panicking` | Each truncated prefix gives an error, not a panic | | `missing_fields_give_clear_errors`, `extension_*_without_decoder_error` | Error text | | `deprecated_*_hook_forwarding_to_migrated_built_in_still_encodes`, `source_without_hooks_falls_back_to_codec` | Backward compatibility of the deprecated hooks | cargo-mutants (`legacy.rs` and the `try_to_proto*` hooks and wrapper helpers in `source.rs`, `sink.rs`, `file.rs`): 17 caught, 0 missed, 4 unviable (`legacy.rs`: `PhysicalPlanType` has no `Default`). ## Are there any user-facing changes? - **Deprecations:** the 55 `try_to_proto` hooks on `DataSource`, `FileSource` and `DataSink`. Existing implementations still work. Migration notes are in `upgrading/56.0.0.md`. - **Wire:** a 56 writer produces plans that a 55 reader cannot decode. A 56 reader decodes 55 plans. ([Plans are not guaranteed compatible across versions](https://github.com/apache/datafusion/blob/main/datafusion/proto/src/lib.rs#L47-L51).) ## Open points - `#[deprecated]` on a trait method gives no warning to implementers who override it. - The generated code was produced with `protox` because `protoc` was not available. Run `datafusion/proto-models/regen.sh` to confirm. - `legacy.rs` clones each legacy subtree. This is acceptable for a compatibility path. 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- 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]
