refactor(pass2): shared segments check tiling; their coverage is derived by tumbling_state

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W7qG9aFyPij5uWsyAJCxDW

Author: zzylol

crates/integration-tests/tests/planner_layering_common/mod.rs

@@ -577,13 +577,10 @@ pub fn query_option(
 
 /// Whether `dag` shares a summary build across queries: Pass 2's
 /// shared-segment rule (Q60), whose segments several queries merge.
-pub fn shares_segments(dag: &impl ExportedDag, query_roots: &[LogicalASAPNodeId]) -> bool {
-    cross_query_nodes(dag, query_roots).into_iter().any(|id| {
-        matches!(
-            dag.payload(id),
-            LogicalASAPOperatorPayload::SummaryAgg { .. }
-        )
-    })
+pub fn shares_segments(dag: &impl ExportedDag, query_roots: &[PhysicalASAPNodeId]) -> bool {
+    cross_query_nodes(dag, query_roots)
+        .into_iter()
+        .any(|id| matches!(dag.payload(id), Operator::ASAP(ASAPOp::SummaryAgg { .. })))
 }
 
 /// Nodes reachable from more than one query root.

crates/integration-tests/tests/planner_layering_example4.rs

@@ -100,7 +100,7 @@ fn eh_options(run: &Run) -> BTreeMap<Materialization, (String, PhysicalASAPNodeI
 
 /// The physical candidates of the shared-segment candidate (Q60), keyed by
 /// the materialization of its segments, with the newest segment's id.
-fn segment_options(run: &Run) -> BTreeMap<Materialization, (String, LogicalASAPNodeId)> {
+fn segment_options(run: &Run) -> BTreeMap<Materialization, (String, PhysicalASAPNodeId)> {
     let logical = run
         .logical
         .iter()
@@ -224,7 +224,7 @@ fn stage2_a_segments_are_built_once_for_all_consumers() {
                 .dag
                 .nodes
                 .iter()
-                .filter(|n| matches!(n.payload, LogicalASAPOperatorPayload::SummaryMerge))
+                .filter(|n| matches!(n.payload, Operator::ASAP(ASAPOp::SummaryMerge { .. })))
                 .count();
             assert_eq!(merges, 5, "{id}");
             if let Some(cost) = run.selection.costs.get(id) {

crates/logical-optimizer/src/pass2/window_composition.rs

@@ -359,7 +359,7 @@ pub fn share_window_segments<Id: Clone>(
         let end = windows.iter().map(|w| bounds(w).1).max().unwrap_or(0);
         if width == 0
             || (end - start) / width > MAX_PANES
-            || !windows.iter().all(|w| panes_cover_window(w, width))
+            || !windows.iter().all(|w| panes_tile_window(w, width))
         {
             continue;
         }