[EPIC/reference] Port ExecutionPlan serialization to try_to_proto / try_from_proto hooks (full)#23496
Draft
adriangb wants to merge 13 commits into
Draft
[EPIC/reference] Port ExecutionPlan serialization to try_to_proto / try_from_proto hooks (full)#23496adriangb wants to merge 13 commits into
adriangb wants to merge 13 commits into
Conversation
…nExec reference (apache#22419) General part: ExecutionPlanEncodeCtx/DecodeCtx + internal dispatch traits in datafusion-physical-plan (mirrors the PhysicalExpr pattern), the try_to_proto trait hook, expect_plan_variant! macro, and the ConverterPlanEncoder/Decoder adapters wiring the hook into the central dispatch in datafusion-proto. ProjectionExec migrated as the reference: encode downcast arm deleted, decode arm reduced to a one-liner delegating to ProjectionExec::try_from_proto. All 189 proto_integration roundtrip tests pass (incl. TPC-H). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SfBPC2yYCwHUn4V8pQCVj3
apache#22419) Fleet-migrated Filter, Empty, CoalesceBatches, Union/Interleave, Sort, Repartition, and HashJoin onto try_to_proto/try_from_proto. Central encode downcast arms deleted; decode arms reduced to one-liners. All 190 proto_integration tests pass. Findings proven by the prototype: - Partitioning and PhysicalSortExpr wrapping (serialize_partitioning / serialize_physical_sort_exprs) inline cleanly from datafusion-proto-models; no datafusion-proto-only helper needed. - Schema and ScalarValue conversions come from datafusion-proto-common (already a physical-plan dep under the proto feature). - Enum conversions (JoinType/NullEquality/PartitionMode/JoinSide) must be inlined as by-NAME matches; proto vs common variant ordering differs. - The hook must follow downcast_delegate() before dispatch, else wrapper plans serialize wrong (caught by serialize_uses_downcast_delegate). - No migrated plan needed the PhysicalExtensionCodec or function registry. Added ExecutionPlanDecodeCtx::decode_children for the N-inputs shape. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SfBPC2yYCwHUn4V8pQCVj3
…roto ctx (apache#22419) Adds encode_udf/udaf/udwf + decode_udf/udaf/udwf to ExecutionPlanEncodeCtx/ DecodeCtx (and the internal dispatch traits). These take datafusion-expr types + Vec<u8> and never name a proto type; the datafusion-proto adapter backs them over PhysicalExtensionCodec + the registry, owning the payload/registry lookup-order policy in one place. Wire-identical to the existing fun_definition semantics ((!buf.is_empty()).then_some(buf)). This lets function-carrying plans (Aggregate, window, ScalarSubquery) ride the polymorphic hook instead of staying as typed arms — possible because physical-plan sits above datafusion-expr (unlike the expr-side ctx). apache#23421's ScalarUdfCodec stays closed; ScalarFunctionExpr stays special-cased on the expr side where the crate cycle forces it. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SfBPC2yYCwHUn4V8pQCVj3
…ggregateExec (apache#22419) Bucket A: CrossJoin, NestedLoopJoin, SortMergeJoin, SymmetricHashJoin, CoalescePartitions, Cooperative, Buffer, GlobalLimit, LocalLimit, PlaceholderRow, Unnest, Explain, Analyze. Bucket B (function-carrying, proves the ctx function-codec design): AggregateExec — UDAF round-trips purely through ctx.encode_udaf/decode_udaf with zero proto/codec/registry types crossing the boundary. All join enum conversions inlined by-name (StreamPartitionMode, JoinType, NullEquality, JoinSide). StringifiedPlan and UnnestOptions conversions (which lived only in datafusion-proto) inlined from proto-common/models. All 189 roundtrip tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SfBPC2yYCwHUn4V8pQCVj3
…o hook (apache#22419) Both share the Window proto variant; a single WindowAggExec::try_from_proto dispatches on input_order_mode. UDWF via ctx.encode_udwf/decode_udwf and UDAF-as-window via ctx.encode_udaf/decode_udaf — no raw codec. Window frame + InputOrderMode enums inlined by-name. 189 roundtrip tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SfBPC2yYCwHUn4V8pQCVj3
…ache#22419 stack C) Adds a proto feature + try_to_proto hook to the DataSource and FileSource traits (datafusion-datasource), so DataSourceExec rides the ExecutionPlan hook and delegates polymorphically: DataSourceExec -> DataSource -> FileSource. No downcast in the encode path for migrated sources. Ports the shared FileScanConfig <-> proto conversion into datafusion-datasource (FileScanConfig::{to_proto_conf, from_proto_conf, parse_table_schema_from_proto}); orderings/partition exprs go through the ctx, schema/stats/scalars via proto-common. Needs NO PhysicalExtensionCodec. CsvSource migrated as proof: its central encode branch is deleted and the CsvScan decode arm repointed to CsvSource::try_from_proto. New roundtrip_csv_scan test proves the hook chain end-to-end. All 191 proto_integration tests pass. Foundation confirms all 4 remaining formats (json/arrow/avro/parquet) are clean (parquet encryption is a non-issue: crypto isn't in the proto today). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SfBPC2yYCwHUn4V8pQCVj3
… hook (apache#22419) Completes the file-scan family: all 5 FileSources now serialize via FileSource::try_to_proto and decode via <Source>::try_from_proto. The central scan encode branches in try_from_data_source_exec are deleted (only MemorySourceConfig remains) and the 5 scan decode arms repointed. Parquet carries its predicate via ctx.encode_expr and options via proto-common; none of the formats need the raw codec. New roundtrip_json_scan / roundtrip_avro_scan tests (+ existing csv/arrow/parquet scan tests) exercise every FileSource hook end-to-end. 192 tests pass (193 with avro). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SfBPC2yYCwHUn4V8pQCVj3
…hook (apache#22419) AsyncFuncExec: clean single-child + N-exprs, no codec. ScalarSubqueryExec needed one small typed ctx addition: decode_child_with_scalar_subquery_results (dispatch method + ctx wrapper + one-line datafusion-proto adapter that re-enters with a scalar-subquery-results scope). The scope handle is a datafusion-expr type, so no proto type leaks and no codec is exposed — same precedent as the UDF ctx methods. The existing roundtrip_scalar_subquery_exec test exercises the scoped decode. 192 roundtrip tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SfBPC2yYCwHUn4V8pQCVj3
…ok (apache#22419) MemorySourceConfig rides DataSource::try_to_proto (in datafusion-datasource, where the type actually lives). Record-batch IPC serde is pure Arrow, inlined locally like CsvSource's byte helpers — no codec. With every DataSource migrated (5 file formats + memory), try_from_data_source_exec and the DataSourceExec encode downcast arm are deleted entirely: DataSourceExec now serializes purely via the ExecutionPlan hook, delegating through DataSource -> FileSource with zero downcasts in the central encode path. roundtrip_memory_source + the 5 scan tests stay green (192 total). GenerateSeries/LazyMemoryExec remain typed arms (their generator types live in datafusion-functions-table, above physical-plan — the hook can't name them without a crate cycle). Extension stays as the terminal codec.try_encode fallback. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SfBPC2yYCwHUn4V8pQCVj3
…o the proto hook (apache#22419) Mirrors the scan side: DataSink::try_to_proto hook on the trait, DataSinkExec rides the ExecutionPlan hook and delegates to its sink (encoding input + sort_order via the ctx, since a sink carries no child exprs). FileSinkConfig serde ported into datafusion-datasource (file_sink_config/proto.rs), reusing the scan side's PartitionedFile helpers; no codec needed. CsvSink/JsonSink/ParquetSink each implement DataSink::try_to_proto + an inherent try_from_proto. The central try_from_data_sink_exec, the DataSinkExec encode arm, the try_into_*_sink_physical_plan decoders, and the colliding TryFromProto<&protobuf::*Sink> for *Sink decode impls are all deleted; the 3 sink decode arms repoint to <Fmt>Sink::try_from_proto. roundtrip_{csv,json,parquet}_sink stay green (193 tests). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SfBPC2yYCwHUn4V8pQCVj3
…e#22419) Same shape as SortExec (input + ExprType::Sort-wrapped orderings + fetch sentinel). No codec. 193 roundtrip tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SfBPC2yYCwHUn4V8pQCVj3
…wncast chain (apache#22419) With every plan/source/sink migrated to the try_to_proto/try_from_proto hooks, delete the 58 now-dead per-plan try_from_*_exec / try_into_*_physical_plan methods from PhysicalPlanNodeExt and try_from_data_sink_exec. The central encode path is now a single hook call + the extension codec fallback; decode is a one-liner dispatch table. Only GenerateSeries/LazyMemoryExec (crate cycle) and Extension (third-party fallback) remain as typed arms. This removes released-in-54.0.0 pub API (the per-plan PhysicalPlanNodeExt methods) and the TryFromProto<&protobuf::*Sink> impls — a deliberate, documented api-change (coherent redesign; these were pub-for-proto scaffolding from apache#21929). mod.rs shrinks by ~1900 lines. 193 tests pass; clippy -D warnings + fmt clean. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SfBPC2yYCwHUn4V8pQCVj3
github-actions
Bot
added
proto
This was referenced Jul 12, 2026
This was referenced Jul 12, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Rationale for this change
This draft shows the complete end state of #23494 in one place: every built-in
ExecutionPlan, all fiveFileSourcescans, all three file sinks, andMemorySourceConfigmoved onto thetry_to_proto/try_from_protohooks, with the centraldowncast_refchain deleted. It is a reference for the incremental PRs — those land one plan/family at a time off #23495; this is not meant to be merged directly.What changes are included in this PR?
SortPreservingMergeExec), limits, repartition, both coalesces, unnest, explain, analyze, cooperative, buffer, placeholder, empty, coalesce-batches,AggregateExec(UDAF), both window execs (UDWF),AsyncFuncExec,ScalarSubqueryExec.try_to_protohook on theDataSource/FileSourcetraits + theFileScanConfigserde ported intodatafusion-datasource; csv/json/arrow/avro/parquet sources +MemorySourceConfig.DataSourceExecnow serializes purely by delegatingDataSource → FileSource— no downcast in the central encode path.DataSinkhook +FileSinkConfigserde port + the three file sinks.PhysicalPlanNodeExtmethods and thetry_from_data_source_exec/try_from_data_sink_execdispatchers are deleted;mod.rsshrinks by ~1900 lines.Left as typed dispatch (by design):
GenerateSeries/LazyMemoryExec(their generator types live above physical-plan — a hook can't name them without a crate cycle) andExtension(the terminal third-party fallback).Notable findings baked in: enum conversions are inlined as by-name matches (proto vs.
datafusion_commonnumbering differs); the encode dispatch resolvesdowncast_delegate()before the hook (a wrapper-plan bug this caught);Partitioning/PhysicalSortExpr/Schema/ScalarValueall serialize fromdatafusion-proto-models/-commonwith nodatafusion-proto-only helper; no migrated plan needs the rawPhysicalExtensionCodec.Are these changes tested?
Yes — the full
proto_integrationroundtrip suite passes (193 tests incl. TPC-H, withavro/parquet), plus new scan/sink roundtrip tests (roundtrip_{csv,json,avro}_scan) that exercise the new hook paths.clippy -D warnings+fmtclean; the workspace builds.Are there any user-facing changes?
Yes — a documented breaking
api change: the per-planPhysicalPlanNodeExtserialization methods and theTryFromProto<&protobuf::{Json,Csv,Parquet}Sink>impls are removed (pub-for-proto scaffolding from #21929, released in 54.0.0). The load-bearing public API (AsExecutionPlan,PhysicalExtensionCodec,PhysicalProtoConverterExtension) is unchanged, and the wire format is byte-for-byte identical. An Upgrade Guide note will accompany the incremental removals.