[Refactor] Input slots take their schema from the physical DAG node; remove InputContract #965

Example

A compile check cuts a plan at its precompute frontier and at its raw Scan leaves, then compiles each side without a reader. Every boundary is built the same way: copy the cut node's schema out of the PhysicalASAPDAG and wrap it in an InputContract.

// crates/executor/src/physical_planner/candidates.rs:406
InputContract::bounded(Arc::new(state.output_schema.clone()))
// crates/executor/src/physical_planner/mod.rs:131 (scan_contract)
InputContract::bounded(Arc::new(schema.clone()))

InputContract is

pub struct InputContract {
    pub schema: SchemaRef,          // Arc<planner_types::ir::schema::Schema>
    pub properties: PlanProperties, // { boundedness, emission }
}

and SchemaRef is Arc<Schema> over the same planner_types::ir::schema::Schema that the logical DAG's OperatorNode.schema and the physical DAG's PhysicalASAPDAGNode.output_schema hold (crates/executor/src/values/mod.rs:16). So the contract at a boundary repeats information the DAG already has, and the two extra properties carry a constant:

Today a caller of compile(dag, inputs, roots) passes BTreeMap<NodeId, InputContract> and must rebuild each schema from the DAG it also passes. It should pass only which nodes are boundaries.

Concrete maintenance problem

• The schema of an input slot has two sources: the DAG node and the contract the caller builds. Nothing checks that they agree. compile trusts the contract.
• PlanProperties in the contract suggests that boundaries can be unbounded or have a required emission mode. No caller uses either. The only live rule is "every bound input is bounded".
• Thirteen non-test call sites (candidates.rs, mod.rs, precompute.rs, promql_rows.rs, promql_values.rs, planner/src/installable.rs) repeat InputContract::bounded(Arc::new(<node>.output_schema.clone())).

Behavior that must remain unchanged

• Compilation still opens no readers.
• instantiate still rejects a missing binding, an extra binding, a schema mismatch, and a source that does not declare Bounded.
• cut, compose, cut_candidate and frontier_from_timing produce the same compiled DAGs.

Before and after structure

• Before: compile(dag, inputs: BTreeMap<NodeId, InputContract>, roots). An input node stores Node::Input(InputContract).
• After:
  • compile(dag, boundaries: &BTreeSet<NodeId>, roots) reads each boundary's schema from dag's node output_schema. Raw Scan leaves keep the predicate check that scan_contract does today.
  • An input node stores Node::Input(SchemaRef). The compiled DAG does not keep the PhysicalASAPDAG, so the schema stays there.
  • instantiate requires every source to declare Bounded as a fixed rule.
  • InputContract, PlanProperties on inputs, InputContract::from_source and scan_contract are removed. input_contracts() and output_contract() become input_schemas() and output_schema(id).
  • compose and from_operators take BTreeMap<NodeId, SchemaRef>, since PromQL value-function fragments have no DAG node to read a schema from.
  • CompiledPhysicalDAG drops Serialize/Deserialize (and UncheckedDAG). No code outside the executor stores one; the persisted handoff is InstallablePlan. After [Refactor] TableProvider (RawSource renamed) with a ScanRequest, and one BindingDataSources for table slots and materialized-view slots #966 every Scan leaf is always a table slot, so boundaries then holds only the materialized views of the cut.

Why the proposed structure is minimally complex

It removes one public type and two fields that never vary. No new type is added. The schema of a slot has one source: the DAG node.

Existing unit and end-to-end protection

• crates/executor/tests/precompute_candidates.rs (cut_candidate, frontiers), deployment_computation.rs, current_series_heap.rs, hydra_cms_execution.rs.
• The planner compile check, PlannerExecutorCompileCheck (crates/planner/src/installable.rs).
• Add a test: binding a source whose schema differs from the boundary node's output_schema fails in instantiate.
• Add a test: a RawSource with the default Boundedness::Unknown is still rejected.

Compatibility or migration impact

This is a public API change in physical_planner: the compile, compile_before, compose and from_operators signatures change, and InputContract is removed. The backend main does not reference InputContract. Only older backend branches do. CompiledPhysicalDAG loses its serde form, which is safe because no stored compiled DAG exists outside the Planner.

Parent: #968 (PR 2 of 6). Related: the design doc docs/design_docs/physical-planning-and-deployment.md §4 (updated by PR 1 of #968).