fix: check for table scan before expanding (#2491)

* fix: check for table scan before expanding

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* change assert_ok to unwrap

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* fix clippy warning

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* update sqlness result

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* don't skip dml

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* uncomment ignored tests

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

---------

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

src/query/src/dist_plan/analyzer.rs

@@ -14,7 +14,6 @@
 
 use std::sync::Arc;
 
-use common_telemetry::info;
 use datafusion::datasource::DefaultTableSource;
 use datafusion::error::Result as DfResult;
 use datafusion_common::config::ConfigOptions;
@@ -46,7 +45,9 @@ impl AnalyzerRule for DistPlannerAnalyzer {
     ) -> datafusion_common::Result<LogicalPlan> {
         let plan = plan.transform(&Self::inspect_plan_with_subquery)?;
         let mut rewriter = PlanRewriter::default();
-        plan.rewrite(&mut rewriter)
+        let result = plan.rewrite(&mut rewriter)?;
+
+        Ok(result)
     }
 }
 
@@ -138,10 +139,6 @@ impl PlanRewriter {
     /// Return true if should stop and expand. The input plan is the parent node of current node
     fn should_expand(&mut self, plan: &LogicalPlan) -> bool {
         if DFLogicalSubstraitConvertor.encode(plan).is_err() {
-            info!(
-                "substrait error: {:?}",
-                DFLogicalSubstraitConvertor.encode(plan)
-            );
             return true;
         }
 
@@ -251,6 +248,13 @@ impl TreeNodeRewriter for PlanRewriter {
             return Ok(node);
         }
 
+        // only expand when the leaf is table scan
+        if node.inputs().is_empty() && !matches!(node, LogicalPlan::TableScan(_)) {
+            self.set_expanded();
+            self.pop_stack();
+            return Ok(node);
+        }
+
         self.maybe_set_partitions(&node);
 
         let Some(parent) = self.get_parent() else {