feat(ir): add logical sub-DAG sharing and export without execution timing

[zzylol]

Problem: the unified operator IR has no canonicalization, no sub-DAG sharing, and no logical export without execution timing

#509 §Stages says logical planning outputs CandidateLogicalASAPDAGs: what to compute, with no materialization or execution placement ("Materialization and execution placement are decided in later stages", §1. Logical ASAP-aware optimization). #509 §Pass 2: ASAP-aware common-subexpression elimination starts from traditional CSE:

Sub-DAG sharing means several consumers reference one operator and its upstream dependencies. Traditional CSE provides common sub-DAG sharing for eligible, structurally identical computations.

Its first rule is the identical-expression rule: "The input and computation semantics are identical" → "One common computation node serving multiple consumers."

#511 adds two constraints on the same graph:

• §2.3 Timing follows the planning-stage design (operator-sharing.md): "Logical nodes may retain None" for OperatorNode.timing; timing follows the physical materialization choice.
• §4 Acceptance (decoupling_op_and_expr.md): "Operator dependencies inside scalar conversions/subqueries remain visible and shared; scalar DAGs remain owned."

Before this PR. Canonicalization, CSE and DAG export exist only for the old split IRs:

Function on the base branch	Works on
pre_asap::canonicalize::canonicalize(QueryExpr) -> QueryExpr	pre-ASAP QueryExpr
pre_asap::cse::share_common_sub_dags(Vec<(Id, QueryExpr)>)	pre-ASAP QueryExpr
post_asap::cse	post-ASAP SummaryNode
dag_export::export / export_post_asap	QueryExpr (+ post-ASAP substitutions)

The unified ir::OperatorNode DAG from #511 (with #567 coverage and #560 SummaryMerge) has none of these. So:

1. Two equivalent spellings stay different shapes. Example: Filter { EXISTS (s) } and Join { Semi, true, left, s } mean the same thing but cannot be matched structurally.

2. Two identical sub-DAGs stay two Rcs. Example: one query with the same grouped p50 on both sides of a comparison:

   BinaryOp(==)
   ├── Aggregate(Quantile q=0.5 col=2, by [1]) ── Scan m     (allocation A)
   └── Aggregate(Quantile q=0.5 col=2, by [1]) ── Scan m     (allocation B)
   

   Nothing collapses A and B into one node with two consumers.

3. There is no flat, serializable logical DAG for the unified IR. The earlier feat(ir): add logical sub-DAG sharing and export without execution timing #537 proposal tied export to mandatory timing, so an untimed logical plan (timing = None, which docs: define unified operators and SQL/PromQL scalar boundaries #511 §2.3 allows) could not be exported.

Scope of this PR. It implements the traditional-CSE base of #509 Pass 2 (the identical-expression rule, as exact structural equality) and a transport for the stage-1 output that carries no execution timing. It also adds the canonicalization pass CSE needs as input. It does not implement the summary-capability rule or the window-composition rule of #509 Pass 2, Pass 1 candidate generation, or any physical-stage decision (materialization, timing assignment, phase splitting).

Proposed method

Three steps, all in the logical stage on bound OperatorNode DAGs. Each works on the unified IR in crates/types/src/ir/.

1. Canonicalize (ir::canonicalize). Bottom-up over operator inputs, memoized by Rc pointer. At each node it applies local rules until none matches:

• Heavy-hitter promotion. Limit { n: k, offset: 0, Sort { one descending column key, [passthrough Project] Aggregate(Reduce(by), [Count | Sum]) } } becomes Aggregate(Reduce(partition), [TopK { k }]) over the unchanged inner aggregate. It is skipped for ascending sorts, offsets, ranking by a group key, a Limit partition that differs from the Sort partition, and Sum over a derived child Aggregate (rate/increase).
• Subquery lowering. In a Filter, one EXISTS (s) conjunct becomes Join { Semi, true }, NOT EXISTS (s) becomes Join { Anti, true }, and a positive x IN (s) becomes Join { Semi, x = Column(left_width) }. Remaining conjuncts stay in an outer Filter. One conjunct per round.
• Left alone on purpose: NOT IN, scalar subqueries (a cross join would lose their zero-row NULL and multi-row error semantics), and ROW_NUMBER() filters (removing the window column would change the visible schema).

Untouched sub-DAGs keep their pointer identity, a rewritten shared sub-DAG stays shared, and the pass is idempotent. A Concat whose first branch changes schema drops its discriminator_unique_key.

2. Share common sub-DAGs (ir::cse). Hash-consing over all roots of a workload batch, bottom-up:

• "Children" are operator inputs and operator nodes read by scalar expressions (PromqlScalarFromVector, ScalarSubquery, Exists, InSubquery). So scalar(v) in two queries can share v.
• structural_hash only picks a bucket. The decision is typed equality (same_node): children by pointer (or memoized value comparison), then result_kind, schema, timing, coverage, guarantee, and the operator's own fields. Guarantee and own fields must also serialize identically, so 0.0 vs -0.0 and NaN vs NaN are not shared.
• Legality gate: a non-ASAP node is reused only if schema.has_unique_key(). ASAP nodes have no such gate.
• coverage is part of both hash and equality, so two summary states over different observations are never shared.
• Each root's value is unchanged (PartialEq-equal to its input); only its Rc structure may now alias.

The pass does not decide whether sharing is cheaper. That is selection's job (#509 stage 3).

3. Export a logical DAG with no execution timing assigned (ir::export). compile_logical_asap_query first runs QueryRoot::validate_structure(). Then it visits the DAG in post-order and gives each distinct Rc one LogicalASAPNodeId (0, 1, 2, … in visit order). For each node it emits:

• the operator payload with children removed (children become edges) and scalar expressions mirrored as WireScalarExpr, with operator references replaced by IDs;
• result_kind, output_schema, guarantee, coverage;
• one edge per operator input (Input / Left / Right) and one ScalarRef edge per operator node read by its scalar expressions (and by a SummaryAgg.filter).

timing is not exported. A standalone scalar root (QueryRoot::Scalar) stays a scalar root; it gets no synthetic operator node.

LogicalASAPDAGDocument::validate() checks a received DAG: version, unique IDs, edge endpoints, edge schema = producer schema, summary family/grouping in the schema, coverage rules, no cycles, every node reachable from the root.

Key code interfaces

crates/types/src/ir/canonicalize.rs

pub fn canonicalize(root: Rc<OperatorNode>) -> Result<Rc<OperatorNode>, SchemaDerivationError>;

crates/types/src/ir/cse.rs

pub type HashCache = HashMap<*const OperatorNode, u64>;

pub fn structural_hash(node: &OperatorNode, cache: &mut HashCache) -> u64;
pub fn share_common_sub_dags<Id>(
    roots: Vec<(Id, Rc<OperatorNode>)>,
) -> Vec<(Id, Rc<OperatorNode>)>;
pub fn dag_node_count(root: &Rc<OperatorNode>) -> usize;

crates/types/src/ir/export.rs

pub const LOGICAL_ASAP_DAG_WIRE_VERSION: u32 = 1;

pub struct LogicalASAPDAGDocument {
    pub schema_version: u32,
    pub dag: LogicalASAPDAG,
}

pub struct LogicalASAPDAG {
    pub nodes: Vec<LogicalASAPDAGNode>,
    pub edges: Vec<LogicalASAPDAGEdge>,
    pub root: LogicalASAPQueryRoot,
}

pub struct LogicalASAPDAGNode {
    pub id: LogicalASAPNodeId,
    pub payload: LogicalASAPOperatorPayload,
    pub result_kind: OperatorResultKind,
    pub output_schema: Schema,
    pub guarantee: Option<ResultGuarantee>,
    #[serde(default)]
    pub coverage: Option<SummaryCoverage>,
}

pub struct LogicalASAPDAGEdge {
    pub producer: LogicalASAPNodeId,
    pub consumer: LogicalASAPNodeId,
    pub role: EdgeRole,
    pub intermediate_schema: Schema,
    pub grouping: GroupingEdgeCompatibility,
}

pub enum LogicalASAPQueryRoot {
    Operator(LogicalASAPNodeId),
    Scalar(WireScalarExpr),
}

pub enum LogicalASAPDAGValidationError {
    UnsupportedVersion(u32),
    DuplicateNode(LogicalASAPNodeId),
    MissingNode(LogicalASAPNodeId),
    EdgeSchemaMismatch(LogicalASAPNodeId),
    SummarySchemaMismatch(LogicalASAPNodeId),
    InvalidCoverage(LogicalASAPNodeId),
    Cycle,
    UnreachableNode(LogicalASAPNodeId),
}

impl LogicalASAPDAGDocument {
    pub fn new(dag: LogicalASAPDAG) -> Self;
    pub fn validate(&self) -> Result<(), LogicalASAPDAGValidationError>;
}
impl LogicalASAPDAG {
    pub fn validate(&self) -> Result<(), LogicalASAPDAGValidationError>;
}
impl LogicalASAPQueryRoot {
    pub fn operator_refs(&self) -> Vec<LogicalASAPNodeId>;
}

pub fn compile_logical_asap_dag(root: &Rc<OperatorNode>)
    -> Result<LogicalASAPDAG, SchemaDerivationError>;
pub fn compile_logical_asap_query(root: &QueryRoot)
    -> Result<LogicalASAPDAG, SchemaDerivationError>;
pub fn compile_logical_asap_dag_with_node_ids(root: &Rc<OperatorNode>)
    -> Result<LogicalASAPDAGCompilation, SchemaDerivationError>;
pub fn compile_logical_asap_query_with_node_ids(root: &QueryRoot)
    -> Result<LogicalASAPDAGCompilation, SchemaDerivationError>;

pub struct LogicalASAPDAGCompilation {
    pub dag: LogicalASAPDAG,
    pub node_ids: LogicalASAPNodeIdentityMap,
}
impl LogicalASAPNodeIdentityMap {
    pub fn node_id(&self, node: &Rc<OperatorNode>) -> Option<LogicalASAPNodeId>;
    pub fn operator_node(&self, id: LogicalASAPNodeId) -> Option<&Rc<OperatorNode>>;
}

crates/types/src/ir/wire.rs (private module; these are re-exported from ir::export)

pub struct LogicalASAPNodeId(pub u32);

pub enum EdgeRole { Input, Left, Right, ScalarRef }

pub enum GroupingEdgeCompatibility {
    Identical,
    ConsumerCoarsensProducer,
    Incompatible,
    NotApplicable,
}

pub enum LogicalASAPOperatorPayload {
    Relational { operator: NonASAPOpKind },
    SummaryAgg {
        family: FieldDataType,
        input: SummaryUpdate,
        reduction: Reduction,
        grouping: GroupingStrategy,
        #[serde(default)]
        filter: Option<WirePredicate>,
    },
    SummaryEstimate { query: SketchStatistic },
    FinalizeExactAccumulator,
    MaintainPopulation { population: MaintainedPopulation<OperatorNode> },
    EvaluatePopulation { evaluation: PopulationStatistic },
    SummaryMerge,
    SummarySubtract,
    SummaryDelete { key: ColumnId },
    SummaryJoin { key: ColumnId, family: FieldDataType },
    Extension { name: String },
}

pub enum WireScalarExpr {
    // Same variants as ScalarExpr; the four operator references become IDs:
    PromqlScalarFromVector(LogicalASAPNodeId),
    ScalarSubquery(LogicalASAPNodeId),
    Exists { subquery: LogicalASAPNodeId, negated: bool },
    InSubquery { expr: Box<WireScalarExpr>, subquery: LogicalASAPNodeId, negated: bool },
    // ... Column, Literal, Negative, Compare, BoolAnd, BoolOr, Not, IsNull,
    //     IsNotNull, Cast, InList, FunctionCall, Arithmetic, Case,
    //     CurrentTimestamp, EvalTimestamp
}
impl WireScalarExpr {
    pub fn operator_refs(&self) -> Vec<LogicalASAPNodeId>;
    pub fn from_expr(expr: &ScalarExpr,
                     id_of: &mut impl FnMut(&Rc<OperatorNode>) -> LogicalASAPNodeId) -> Self;
}

Usage:

let roots = share_common_sub_dags(vec![("q1", canonicalize(q1)?), ("q2", canonicalize(q2)?)]);
let compiled = compile_logical_asap_dag_with_node_ids(&roots[0].1)?;
let doc = LogicalASAPDAGDocument::new(compiled.dag);
doc.validate()?;
let json = serde_json::to_string(&doc)?;

Also added, listed by name only: NonASAPOpKind (the child-free, WireScalarExpr-based mirror of every NonASAPOp variant; kind fields renamed join_kind, set_kind, range_kind, sample_kind), WirePredicate, WireProjectItem, WireSortKey, NonASAPOpKind::from_op. Doc: docs/develop_docs/logical-asap-dag.md.

Fields

Canonicalize / CSE

Name	Type	Meaning
canonicalize(root)	Rc<OperatorNode> → Result<Rc<OperatorNode>, SchemaDerivationError>	Canonical form of root. Returns the same Rc if nothing changes. Errors only when a rebuilt node fails schema derivation.
share_common_sub_dags(roots)	Vec<(Id, Rc<OperatorNode>)>	One workload batch. Id is any caller key (query name, index); it is returned unchanged, in the same order. A batch of one still deduplicates repeats inside that query. Input should be canonicalized.
structural_hash(node, cache)	&OperatorNode, &mut HashCache → u64	SipHash of the node's own fields (operator with children blanked, result_kind, schema, guarantee, timing, coverage) plus each child's hash. Bucket key only, never the sharing decision.
HashCache	HashMap<*const OperatorNode, u64>	Memo of child hashes by pointer. The caller must keep every cached node alive while the cache is in use, or a reused address aliases a stale entry.
dag_node_count(root)	&Rc<OperatorNode> → usize	Number of unique nodes reachable from root by pointer (operator inputs and scalar-referenced nodes). Measures sharing.

LogicalASAPDAGDocument

Field	Type	Meaning
schema_version	u32	Must equal LOGICAL_ASAP_DAG_WIRE_VERSION (1). Set by new. Version 1 is separate from the older phase-assigned format.
dag	LogicalASAPDAG	The graph.

LogicalASAPDAG

Field	Type	Meaning
nodes	Vec<LogicalASAPDAGNode>	One entry per distinct operator. The compiler pushes them in post-order, so nodes[i].id == LogicalASAPNodeId(i); validate only requires unique IDs.
edges	Vec<LogicalASAPDAGEdge>	All producer → consumer dependencies.
root	LogicalASAPQueryRoot	The query result.

LogicalASAPDAGNode

Field	Type	Meaning
id	LogicalASAPNodeId	Local to this export. Not stable across exports.
payload	LogicalASAPOperatorPayload	The operation without its children.
result_kind	OperatorResultKind	Relation, InstantVector, RangeVector or State (unfinalized summary/accumulator state). Copied from the node.
output_schema	Schema	The node's schema (#535).
guarantee	Option<ResultGuarantee>	Copied accuracy guarantee. None means not assessed, not exact.
coverage	Option<SummaryCoverage>	#567 observation coverage, copied. Some only on State nodes. Required on every SummaryAgg. On a SummaryMerge it must equal SummaryCoverage::merge_disjoint of its input nodes' coverage. Missing in JSON reads as None.

LogicalASAPDAGEdge

Field	Type	Meaning
producer	LogicalASAPNodeId	Node whose output is read.
consumer	LogicalASAPNodeId	Node that reads it.
role	EdgeRole	Input: single or n-ary input (child, Concat/SummaryMerge children, summary_input). Left/Right: two-sided operators (Join, SetOp, BinaryOp, SummarySubtract, SummaryJoin outer/inner). ScalarRef: the consumer reads the producer inside a scalar expression (scalar(v), scalar subquery, EXISTS, IN, or a SummaryAgg filter).
intermediate_schema	Schema	The producer's schema. validate requires equality with the producer's output_schema.
grouping	GroupingEdgeCompatibility	Only computed for SummaryAgg → SummaryAgg edges, from their reductions. Identical: equal reductions. ConsumerCoarsensProducer: PerEntity → Reduce(_), or Reduce(p) → Reduce(c) with both not without and every key of c in p. Incompatible: any other pair of SummaryAggs. NotApplicable: every other edge.

LogicalASAPQueryRoot

Variant	Meaning
Operator(LogicalASAPNodeId)	The result is that node's output.
Scalar(WireScalarExpr)	A standalone scalar root. A constant (42.0) produces zero nodes; a scalar subquery produces the referenced producer DAG.
operator_refs()	The root's node IDs: one for Operator, all referenced IDs for Scalar. Reachability starts here.

LogicalASAPOperatorPayload (each ASAP variant is the matching ASAPOp without children)

Variant / field	Meaning
Relational { operator: NonASAPOpKind }	Any ordinary operator.
SummaryAgg.family	FieldDataType: committed summary family and parameters (e.g. Sketch(KLL{k=200}, …)). validate requires a schema field with this type.
SummaryAgg.input	SummaryUpdate: what each row contributes.
SummaryAgg.reduction	Reduction: grouping (PerEntity or Reduce(keys)).
SummaryAgg.grouping	GroupingStrategy. validate requires every Sketch field in the schema to carry this strategy.
SummaryAgg.filter	Option<WirePredicate>: pre-aggregation filter; its operator references become ScalarRef edges.
SummaryEstimate.query	SketchStatistic read from the state (e.g. Quantile { q }).
FinalizeExactAccumulator	Exact accumulator state → plain value.
MaintainPopulation.population	MaintainedPopulation<OperatorNode>, cloned as is.
EvaluatePopulation.evaluation	PopulationStatistic.
SummaryMerge	#560 merge; inputs are its Input edges.
SummarySubtract	Reserved operator; exported if present.
SummaryDelete.key, SummaryJoin.key	ColumnId of the key column.
SummaryJoin.family	FieldDataType of the joined summary.
Extension.name	Extension operator name.

LogicalASAPDAGValidationError

Variant	Raised when
UnsupportedVersion(v)	schema_version != 1 (document only).
DuplicateNode(id)	Two nodes share an ID.
MissingNode(id)	A root reference or edge endpoint has no node.
EdgeSchemaMismatch(producer)	intermediate_schema differs from the producer's output_schema.
SummarySchemaMismatch(id)	A SummaryAgg schema lacks a field of family, or has a Sketch field with a different grouping strategy.
InvalidCoverage(id)	Coverage on a non-State node; coverage fails SummaryCoverage::validate; SummaryAgg without coverage; a SummaryMerge input without coverage, an input union that merge_disjoint rejects, or a stored merge coverage that differs from the union.
Cycle	DFS from the roots finds a back edge.
UnreachableNode(id)	A node not reachable from the root.

Compilation and identity

Name	Meaning
compile_logical_asap_dag(root)	Same as compile_logical_asap_query(&QueryRoot::Operator(root)), without the map.
compile_logical_asap_query(root)	Runs root.validate_structure() (returns its SchemaDerivationError), then exports. Does not call LogicalASAPDAG::validate.
*_with_node_ids	Also return the identity map.
LogicalASAPDAGCompilation.dag	The exported DAG.
LogicalASAPDAGCompilation.node_ids	LogicalASAPNodeIdentityMap: in-memory only, not serialized.
node_id(&rc)	ID of that exact Rc (pointer equality), or None.
operator_node(id)	The Rc<OperatorNode> exported as id, or None.

Wire types

Name	Meaning
LogicalASAPNodeId(pub u32)	Node ID; serialized as a bare number.
WireScalarExpr	ScalarExpr with each operator reference (PromqlScalarFromVector, ScalarSubquery, Exists.subquery, InSubquery.subquery) replaced by the producer's ID. Every other variant keeps its fields, including ExprSemantics.
WireScalarExpr::operator_refs()	All IDs referenced anywhere in the tree.
WireScalarExpr::from_expr(expr, id_of)	Mirrors expr, resolving each referenced node through id_of.

Examples

Example 1: summary merge export (crates/types/tests/logical_export.rs, merged_summary_preserves_typed_state)

Input: one Values node, one KLL (k = 200) SummaryAgg over it, cloned twice with #567 coverage time_ms = 0..1 and 1..2 (same source, input, reduction), and a SummaryMerge of the two. No node has timing.

SummaryMerge                          coverage = merge_disjoint([0,1), [1,2))
├── SummaryAgg(KLL k=200)             coverage time [0,1)
│   └── Values ─┐
└── SummaryAgg(KLL k=200)             coverage time [1,2)
    └── (same Values Rc)

compile_logical_asap_dag(&root) produces (post-order IDs):

id	payload	result_kind	coverage
0	Relational { Values }	Relation	None
1	SummaryAgg { family: KLL{k=200}, … }	State	time [0,1)
2	SummaryAgg { … }	State	time [1,2)
3	SummaryMerge	State	root.coverage

Edges: 0→1, 0→2, 1→3, 2→3, all role: Input, grouping: NotApplicable. The test checks 4 nodes, dag.validate() passes, and the merge node keeps result_kind, output_schema and coverage. Then it sets coverage = None on one SummaryAgg and validate() returns InvalidCoverage(<that id>).

Example 2: CSE on the identical-expression rule (crates/types/src/ir/cse.rs tests)

The BinaryOp from the Problem section, after share_common_sub_dags(vec![("q", root)]): lhs and rhs are the same Rc (single_query_shares_its_own_repeated_sub_dag). The grouped aggregate has unique key by [1], so it is legal to share.

Case (test)	Shared?	Why
Same grouped Quantile q=0.5 col=2 in two roots (median_and_explicit_half_percentile_merge)	yes	equal, unique key
Same grouped quantile on col=2 vs col=3 (distinct_column_quantiles_do_not_merge)	no	own fields differ
Identical global (ungrouped) quantile (no_unique_keys_means_no_merge_even_when_structurally_identical)	no	non-ASAP, no unique key
Identical without(..) aggregate (group_keys_gate_still_prevented_when_partition_by_without_used)	no	no unique key
Identical Dedup { cols: [1] } (dedup_gates_sharing_the_same_as_aggregate)	yes	Dedup adds a unique key
vector(scalar(sum by (service)(up))) twice (scalar_referenced_vector_is_shared_across_queries)	inner vector yes, bridge no	vector is a scalar-referenced child with a unique key; the bridge has none
Two DDSketch SummaryAgg, alpha=0.01, same guarantee (asap_nodes_share_without_a_unique_key)	yes	ASAP: no unique-key gate
DDSketch alpha=0.01 vs 0.001; exact guarantee vs None (asap_nodes_with_distinct_parameters_or_guarantees_are_not_shared)	no	parameters / guarantee differ
SummaryEstimate p95 and p99 over equal producers (evaluations_share_their_producer_but_not_each_other)	producer yes, estimates no	estimates differ in q
Literal 0.0 vs -0.0, +inf vs -inf, NaN vs NaN (signed_zero_and_nonfinite_values_remain_distinct)	no	typed + serialized equality; +inf vs +inf is shared
Equal summary states with different coverage	no	coverage is in hash and equality (no dedicated test)

Example 3: canonicalization (crates/types/src/ir/canonicalize.rs tests)

Input	Output
Limit 5 { Sort desc col 1 { Aggregate(by [1], [Count]) } }	Aggregate(by [], [TopK{k:5}]) { Aggregate(by [1], [Count]) } (promotes_count_ranked_limit_sort); also through a passthrough Project
Same with ascending sort, or with an offset	unchanged
Filter { EXISTS (s) }	Join { Semi, true, left, s } (exists_filter_becomes_semi_join)
Filter { NOT EXISTS (s) }	Join { Anti, true, left, s }
Filter { service IN (s) } over a 3-column left	Join { Semi, service = Column(3) }
Filter { value > 1 AND EXISTS (s) }	Filter { value > 1 } { Join { Semi } }
Filter { NOT IN (s) }	unchanged
Project [ScalarSubquery(s) AS a, ScalarSubquery(s) AS b]	unchanged; both still point to the same s (scalar_subqueries_remain_explicit_and_shared)
Filter { rn <= 5 } { ROW_NUMBER window }	unchanged

Every rewrite test also checks canonicalize(out) returns the same Rc.

Example 4: transport rejections (malformed_transport_is_rejected, standalone_scalar_roots_roundtrip_without_synthetic_operators)

Change to a valid document	Result
schema_version = 99	UnsupportedVersion(99)
node pushed twice	DuplicateNode
root set to LogicalASAPNodeId(9)	MissingNode
extra node with ID 1 not reachable from root	UnreachableNode
JSON node gets "output_state": {"timing": "ingestion_time"}	deserialization error (deny_unknown_fields)
QueryRoot::Scalar(literal 42.0)	valid, 0 nodes
QueryRoot::Scalar(ScalarSubquery(values))	valid, 1 node

scalar_dependencies_share_one_exported_producer: a Project whose child and ScalarSubquery are the same Values exports 2 nodes and 2 edges (one Input, one ScalarRef), and the JSON contains none of output_state, data_state, timing, retention, window.

Out of scope

• docs: propose workload-wide planning, summary sharing, and materialization #509 Pass 2 summary-capability and window-composition rules, and subsumption ("a stricter summary can answer a looser request"). CSE here is exact structural equality only.
• Whether sharing is cheaper (docs: propose workload-wide planning, summary sharing, and materialization #509 stage 3), and reuse across different evaluation times (docs: define unified operators and SQL/PromQL scalar boundaries #511 §3.5). CSE compares timing but has no evaluation-context field to compare.
• Timing assignment, phase validation and splitting, materialization, physical implementation, retention, runtime. OperatorNode.timing stays common IR metadata and is not exported.
• Migrating callers of pre_asap::canonicalize, pre_asap::cse, post_asap::cse and dag_export to the new modules. Downstream physical PRs keep their legacy base until that refactor.

Reviewer notes:

• compile_logical_asap_* does not run LogicalASAPDAG::validate. A SummaryAgg with coverage = None compiles but then fails validate() with InvalidCoverage.
• Some doc comments in canonicalize.rs still describe ROW_NUMBER rewriting and scalar-subquery cross-join lowering (apply_local_rules, filter). The code does neither, and the tests assert the shapes stay unchanged.

Stack and validation

Revised logical foundation 3/6 · Previous: #560 · Next: #539 · Tracker: #528

Order: #567 → #560 → #537 → #539 → #540 → #561.

• The untimed-export regression (logical_export_accepts_unassigned_timing) failed on the original timing requirement and passes now.
• Tests cover ordinary and scalar roots, shared scalar producer references, typed merge states with coverage, JSON round trips without physical metadata, and malformed transport.
• The complete first-five tip passes 1,961 workspace tests/doctests, formatting, and workspace/all-target/all-feature Clippy with warnings denied.

🤖 Generated with Claude Code