fix(query): fix partial insert select

src/query/expression/src/kernels/sort.rs

@@ -15,6 +15,7 @@
 use std::iter::once;
 use std::sync::Arc;
 
+use arrow_array::Decimal256Array;
 use common_arrow::arrow::array::ord as arrow_ord;
 use common_arrow::arrow::array::ord::DynComparator;
 use common_arrow::arrow::array::Array;
@@ -201,6 +202,21 @@ fn compare_variant(left: &dyn Array, right: &dyn Array) -> ArrowResult<DynCompar
     }))
 }
 
+fn compare_decimal256(left: &dyn Array, right: &dyn Array) -> ArrowResult<DynComparator> {
+    let left = left
+        .as_any()
+        .downcast_ref::<Decimal256Array>()
+        .unwrap()
+        .clone();
+    let right = right
+        .as_any()
+        .downcast_ref::<Decimal256Array>()
+        .unwrap()
+        .clone();
+
+    Ok(Box::new(move |i, j| left.value(i).cmp(&right.value(j))))
+}
+
 fn build_compare(left: &dyn Array, right: &dyn Array) -> ArrowResult<DynComparator> {
     match left.data_type() {
         ArrowType::Extension(name, _, _) => {
@@ -213,6 +229,7 @@ fn build_compare(left: &dyn Array, right: &dyn Array) -> ArrowResult<DynComparat
                 )))
             }
         }
+        ArrowType::Decimal256(_, _) => compare_decimal256(left, right),
         _ => arrow_ord::build_compare(left, right),
     }
 }

src/query/service/src/interpreters/interpreter_insert.rs

@@ -116,10 +116,12 @@ impl InsertInterpreter {
         let select_schema = plan.schema();
 
         // validate schema
-        if select_schema.fields().len() < output_schema.fields().len() {
-            return Err(ErrorCode::BadArguments(
-                "Fields in select statement is less than expected",
-            ));
+        if select_schema.fields().len() != output_schema.fields().len() {
+            return Err(ErrorCode::BadArguments(format!(
+                "Fields in select statement is not equal with expected, select fields: {}, insert fields: {}",
+                select_schema.fields().len(),
+                output_schema.fields().len(),
+            )));
         }
 
         // check if cast needed
@@ -457,7 +459,6 @@ impl Interpreter for InsertInterpreter {
                 };
 
                 let catalog = self.plan.catalog.clone();
-
                 let insert_select_plan = match select_plan {
                     PhysicalPlan::Exchange(ref mut exchange) => {
                         // insert can be dispatched to different nodes

src/query/sql/src/planner/planner.rs

@@ -33,6 +33,8 @@ use super::semantic::DistinctToGroupBy;
 use crate::optimizer::optimize;
 use crate::optimizer::OptimizerConfig;
 use crate::optimizer::OptimizerContext;
+use crate::plans::Insert;
+use crate::plans::InsertInputSource;
 use crate::plans::Plan;
 use crate::Binder;
 use crate::Metadata;
@@ -117,7 +119,21 @@ impl Planner {
             }
             .await;
 
-            if res.is_err() && matches!(tokenizer.peek(), Some(Ok(_))) {
+            let mut maybe_partial_insert = false;
+            if is_insert_stmt && matches!(tokenizer.peek(), Some(Ok(_))) {
+                if let Ok((
+                    Plan::Insert(box Insert {
+                        source: InsertInputSource::SelectPlan(_),
+                        ..
+                    }),
+                    _,
+                )) = &res
+                {
+                    maybe_partial_insert = true;
+                }
+            }
+
+            if maybe_partial_insert || (res.is_err() && matches!(tokenizer.peek(), Some(Ok(_)))) {
                 // Remove the previous EOI.
                 tokens.pop();
                 // Tokenize more and try again.

tests/sqllogictests/suites/base/03_common/03_0014_insert_into_select

@@ -85,11 +85,35 @@ SELECT * FROM aggregate_table ORDER BY b
 7
 9
 
+statement ok
+create table test_insert(a int)
+
+statement ok
+insert into test_insert select 1 last from numbers(10) t where t.number = 1 and t.number = 2
+
+query I
+select * from test_insert
+----
+
+statement error 1006
+insert into test_insert select 1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31,32,33,34,35,36,37,38,39,40,41,42,43,44,45,46,47,48,49,50,51,52,53,54,55,56 last from numbers(10) t where t.number = 1
+
+statement ok
+insert into test_insert select unnest([1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31,32,33,34,35,36,37,38,39,40,41,42,43,44,45,46,47,48,49,50,51,52,53,54,55,56])
+
+query I
+select count(1) from test_insert
+----
+56
+
 statement ok
 DROP TABLE base_table
 
 statement ok
 DROP TABLE aggregate_table
 
+statement ok
+DROP TABLE test_insert
+
 statement ok
 DROP DATABASE db1