fix: panic in relation not found in backfill order (#23573)

Signed-off-by: lyang24 <[email protected]>

Author: lyang24

e2e_test/backfill/test_backfill_order_validation.slt

@@ -0,0 +1,99 @@
+statement ok
+SET RW_IMPLICIT_FLUSH TO TRUE;
+
+statement ok
+drop table if exists car_sales cascade;
+
+statement ok
+drop table if exists car_info cascade;
+
+statement ok
+drop table if exists car_regions cascade;
+
+statement ok
+drop table if exists t cascade;
+
+statement ok
+create table car_sales(id int, car_id int, region_id int, price int);
+
+statement ok
+create table car_info(id int, name varchar);
+
+statement ok
+create table car_regions(id int, region varchar);
+
+# Create table t that is NOT used in the query
+statement ok
+create table t(a int, b int, c int);
+
+# Test 1: Should fail because 't' is specified in backfill_order but not used in query
+statement error Table or source 't' specified in backfill_order is not used in the query
+create materialized view m1
+    with (backfill_order = FIXED(car_regions -> car_sales, t -> car_sales))
+    as
+      with price_ranges as (
+        select
+          car_info.name as name,
+          car_sales.price as price,
+          round(log10(1 + car_sales.price)::numeric, 1) as price_range
+        from car_sales join car_info
+          on car_sales.car_id = car_info.id
+          join car_regions
+            on car_sales.region_id = car_regions.id
+      )
+      select
+        name,
+        price_range,
+        count(*) as sales_count,
+        sum(price) as sales_volume,
+        avg(price) as sales_avg,
+        min(price) as sales_min,
+        max(price) as sales_max,
+        approx_percentile(0.5) WITHIN GROUP (ORDER BY price) as sales_est_median,
+        approx_percentile(0.01) WITHIN GROUP (ORDER BY price) as sales_est_bottom_1_percent,
+        approx_percentile(0.99) WITHIN GROUP (ORDER BY price) as sales_est_top_1_percent
+      FROM
+        price_ranges
+GROUP BY name, price_range;
+
+# Test 2: Should also fail when 't' is on the right side of the arrow
+statement error Table or source 't' specified in backfill_order is not used in the query
+create materialized view m2
+    with (backfill_order = FIXED(car_sales -> t))
+    as
+      select
+        car_info.name as name,
+        car_sales.price as price
+      from car_sales join car_info
+        on car_sales.car_id = car_info.id
+        join car_regions
+          on car_sales.region_id = car_regions.id;
+
+# Test 3: Should succeed when only actual tables from the query are specified
+statement ok
+create materialized view m3
+    with (backfill_order = FIXED(car_regions -> car_sales))
+    as
+      select
+        car_info.name as name,
+        car_sales.price as price
+      from car_sales join car_info
+        on car_sales.car_id = car_info.id
+        join car_regions
+          on car_sales.region_id = car_regions.id;
+
+# Cleanup
+statement ok
+drop materialized view m3;
+
+statement ok
+drop table car_sales;
+
+statement ok
+drop table car_info;
+
+statement ok
+drop table car_regions;
+
+statement ok
+drop table t;

src/frontend/src/optimizer/backfill_order_strategy.rs

@@ -232,7 +232,7 @@ pub mod auto {
 }
 
 mod fixed {
-    use std::collections::HashMap;
+    use std::collections::{HashMap, HashSet};
 
     use risingwave_common::bail;
     use risingwave_common::catalog::ObjectId;
@@ -243,16 +243,65 @@ mod fixed {
     use crate::optimizer::backfill_order_strategy::common::{
         bind_backfill_relation_id_by_name, has_cycle,
     };
+    use crate::optimizer::plan_node::{StreamPlanNodeType, StreamPlanRef};
     use crate::session::SessionImpl;
 
+    /// Collect all relation IDs (tables and sources) that are scanned in the plan.
+    fn collect_scanned_relation_ids(plan: StreamPlanRef) -> HashSet<ObjectId> {
+        let mut relation_ids = HashSet::new();
+
+        fn visit(plan: StreamPlanRef, relation_ids: &mut HashSet<ObjectId>) {
+            match plan.node_type() {
+                StreamPlanNodeType::StreamTableScan => {
+                    let table_scan = plan.as_stream_table_scan().expect("table scan");
+                    let relation_id = table_scan.core().table_catalog.id().into();
+                    relation_ids.insert(relation_id);
+                }
+                StreamPlanNodeType::StreamSourceScan => {
+                    let source_scan = plan.as_stream_source_scan().expect("source scan");
+                    let relation_id = source_scan.source_catalog().id;
+                    relation_ids.insert(relation_id);
+                }
+                _ => {}
+            }
+
+            // Recursively visit all inputs
+            for child in plan.inputs() {
+                visit(child, relation_ids);
+            }
+        }
+
+        visit(plan, &mut relation_ids);
+        relation_ids
+    }
+
     pub(super) fn plan_fixed_strategy(
         session: &SessionImpl,
         orders: Vec<(ObjectName, ObjectName)>,
+        plan: StreamPlanRef,
     ) -> Result<HashMap<ObjectId, Uint32Vector>> {
+        // Collect all scanned relation IDs from the plan
+        let scanned_relation_ids = collect_scanned_relation_ids(plan);
+
         let mut order: HashMap<ObjectId, Uint32Vector> = HashMap::new();
         for (start_name, end_name) in orders {
-            let start_relation_id = bind_backfill_relation_id_by_name(session, start_name)?;
-            let end_relation_id = bind_backfill_relation_id_by_name(session, end_name)?;
+            let start_relation_id = bind_backfill_relation_id_by_name(session, start_name.clone())?;
+            let end_relation_id = bind_backfill_relation_id_by_name(session, end_name.clone())?;
+
+            // Validate that both relations are present in the query plan
+            if !scanned_relation_ids.contains(&start_relation_id) {
+                bail!(
+                    "Table or source '{}' specified in backfill_order is not used in the query",
+                    start_name
+                );
+            }
+            if !scanned_relation_ids.contains(&end_relation_id) {
+                bail!(
+                    "Table or source '{}' specified in backfill_order is not used in the query",
+                    end_name
+                );
+            }
+
             order
                 .entry(start_relation_id)
                 .or_default()
@@ -423,7 +472,7 @@ pub fn plan_backfill_order(
     let order = match backfill_order_strategy {
         BackfillOrderStrategy::Default | BackfillOrderStrategy::None => Default::default(),
         BackfillOrderStrategy::Auto => plan_auto_strategy(session, plan),
-        BackfillOrderStrategy::Fixed(orders) => plan_fixed_strategy(session, orders)?,
+        BackfillOrderStrategy::Fixed(orders) => plan_fixed_strategy(session, orders, plan)?,
     };
     Ok(BackfillOrder { order })
 }