Serialize `LazyMemoryExec` / `GenerateSeries` through a generator-level proto hook

[adriangb]

Is your feature request related to a problem or challenge?

LazyMemoryExec is the last built-in ExecutionPlan that datafusion-proto serializes through a central downcast_ref chain. The EPIC #23494 lists it under "Intentionally left as typed dispatch", because an ExecutionPlan-keyed hook cannot reach the generator types. This issue is the follow-up design that section calls for.

Three things follow from the chain staying where it is.

1. datafusion-proto depends on datafusion-functions-table only for this. datafusion/proto/src/physical_plan/mod.rs:45 imports five items to do it:

use datafusion_functions_table::generate_series::{
    Empty, GenSeriesArgs, GenerateSeriesTable, GenericSeriesState, TimestampValue,
};

2. Encode is a downcast_ref ladder over generator types. try_from_lazy_memory_exec reads exec.generators(), takes the read lock, and tries Empty, then GenericSeriesState<i64>, then GenericSeriesState<TimestampValue>, then the date state. A generator it does not recognise returns Ok(None) and the plan silently falls through to the PhysicalExtensionCodec.

3. A new generator is silently unserializable. Adding a generator to generate_series.rs compiles, runs, and serializes to nothing, because nothing connects the two files. This is the failure mode #21835 described for PhysicalExpr and #23494 for plans.

Reproduction

LazyMemoryExec with more than one generator does not serialize at all, and the error blames DataFusion for it:

let table = GenerateSeriesTable::new(
    Arc::clone(&schema),
    GenSeriesArgs::Int64Args { start: 1, end: 3, step: 1, include_end: true, name: "generate_series" },
);
let one = table.as_generator(8192)?;
let two = table.as_generator(8192)?;

let plan: Arc<dyn ExecutionPlan> =
    Arc::new(LazyMemoryExec::try_new(schema, vec![one, two])?);

// One generator encodes. Two do not.
PhysicalPlanNode::try_from_physical_plan(plan, &DefaultPhysicalExtensionCodec {})?;

Internal error: Unsupported plan and extension codec failed with
[This feature is not implemented: PhysicalExtensionCodec is not provided]. Plan: LazyMemoryExec { … }.
This issue was likely caused by a bug in DataFusion's code. Please help us to resolve this by
filing a bug report in our issue tracker: https://github.com/apache/datafusion/issues

The cause is this line in try_from_lazy_memory_exec:

let [generator] = generators.as_slice() else {
    return Ok(None);
};

Ok(None) sends the plan on to codec.try_encode, which the default codec refuses, and the caller reports an Internal error that asks the user to file a bug. Nothing in LazyMemoryExec says it is unserializable with more than one generator.

Describe the solution you'd like

Give the generator the hook, not the plan. LazyBatchGenerator lives in datafusion-physical-plan, and the concrete generators live in datafusion-functions-table, so a hook on the generator trait puts the wire format in the same crate as the type that owns it:

• An encode hook on LazyBatchGenerator, defaulting to Ok(None), matching ExecutionPlan::try_to_proto.
• A decode path keyed by the generator, reached from the PhysicalPlanType::GenerateSeries arm.
• LazyMemoryExec::try_to_proto then encodes its schema and asks each generator for its payload, which also removes the one-generator limit above.

Both downcast_ref ladders and the datafusion-functions-table dependency then leave datafusion-proto.

The open design question is what the decode side keys on. The GenerateSeries wire variant is a single message with a oneof over the argument types, so it is not extensible by a third party today. Two options:

1. Keep GenerateSeries as the only generator variant and dispatch inside datafusion-functions-table. Smallest change, keeps the wire format byte-identical, and does not make generators extensible.
2. Give generators a name-keyed registry, the same shape as the extension-plan registry in Allow extension ExecutionPlans to decode via a per-type registry instead of PhysicalExtensionCodec #24625. Larger, and it makes third-party generators serializable, which nothing asks for yet.

Option 1 looks right until someone needs option 2.

Describe alternatives you've considered

• Leave it as is. It works and the wire format is stable. The cost is that datafusion-proto keeps a dependency and two downcast_ref ladders that exist for one plan, and a new generator keeps failing silently.
• Put the hook on LazyMemoryExec. This is what the EPIC rejected: datafusion-functions-table sits above datafusion-physical-plan, so LazyMemoryExec cannot name the generator types without a cycle.

Additional context

• EPIC: [EPIC] Port ExecutionPlan serialization to try_to_proto / try_from_proto hooks #23494 ("Intentionally left as typed dispatch").
• The PhysicalExpr equivalent: [EPIC] Port PhysicalExpr implementations to use try_from_proto / try_to_proto #22418.
• The extension-plan registry that option 2 would mirror: Allow extension ExecutionPlans to decode via a per-type registry instead of PhysicalExtensionCodec #24625 and feat: allow extension ExecutionPlans to decode via a per-type registry instead of PhysicalExtensionCodec #24631.
• No wire-format change is needed for option 1, so this can land on its own at any time.