fix(query): fix copy into table with disable distributed copy into (#…

…16529)

* fix(query): fix copy into table with diable distributed copy into

* fix(query): fix copy into table with diable distributed copy into

src/query/sql/src/executor/physical_plan.rs

@@ -591,7 +591,6 @@ impl PhysicalPlan {
             | PhysicalPlan::CacheScan(_)
             | PhysicalPlan::ExchangeSource(_)
             | PhysicalPlan::CompactSource(_)
-            | PhysicalPlan::CopyIntoTable(_)
             | PhysicalPlan::ReplaceAsyncSourcer(_)
             | PhysicalPlan::Recluster(_)
             | PhysicalPlan::RecursiveCteScan(_) => Box::new(std::iter::empty()),
@@ -652,6 +651,10 @@ impl PhysicalPlan {
             PhysicalPlan::ChunkAppendData(plan) => Box::new(std::iter::once(plan.input.as_ref())),
             PhysicalPlan::ChunkMerge(plan) => Box::new(std::iter::once(plan.input.as_ref())),
             PhysicalPlan::ChunkCommitInsert(plan) => Box::new(std::iter::once(plan.input.as_ref())),
+            PhysicalPlan::CopyIntoTable(v) => match &v.source {
+                CopyIntoTableSource::Query(v) => Box::new(std::iter::once(v.as_ref())),
+                CopyIntoTableSource::Stage(v) => Box::new(std::iter::once(v.as_ref())),
+            },
         }
     }
 

tests/sqllogictests/suites/base/03_common/03_0046_copy_into_from_stage.test

@@ -0,0 +1,26 @@
+statement ok
+DROP STAGE IF EXISTS stage_t4;
+
+statement ok
+DROP TABLE IF EXISTS t4;
+
+statement ok
+CREATE STAGE stage_t4;
+
+statement ok
+CREATE TABLE t4(str string);
+
+statement ok
+set enable_distributed_copy_into = 0;
+
+statement ok
+copy into @stage_t4 from (SELECT to_string(number) as str from numbers(10));
+
+statement ok
+COPY INTO t4 from @stage_t4 pattern='.*' FILE_FORMAT = (TYPE = 'parquet') PURGE=true FORCE=true max_files=10000;
+
+statement ok
+DROP STAGE IF EXISTS stage_t4;
+
+statement ok
+DROP TABLE IF EXISTS t4;