fix

Author: andylokandy

src/query/expression/src/types/nullable.rs

@@ -355,6 +355,7 @@ impl NullableColumnBuilder<AnyType> {
         if self.validity.pop()? {
             Some(Some(self.builder.pop().unwrap()))
         } else {
+            self.builder.pop().unwrap();
             Some(None)
         }
     }

src/query/expression/src/values.rs

@@ -1735,9 +1735,15 @@ impl ColumnBuilder {
             ColumnBuilder::String(_) => DataType::String,
             ColumnBuilder::Timestamp(_) => DataType::Timestamp,
             ColumnBuilder::Date(_) => DataType::Date,
-            ColumnBuilder::Array(_) => DataType::Array(Box::new(self.data_type())),
-            ColumnBuilder::Map(_) => DataType::Map(Box::new(self.data_type())),
-            ColumnBuilder::Nullable(c) => DataType::Nullable(Box::new(c.builder.data_type())),
+            ColumnBuilder::Array(col) => {
+                let inner = col.builder.data_type();
+                DataType::Array(Box::new(inner))
+            }
+            ColumnBuilder::Map(col) => {
+                let inner = col.builder.data_type();
+                DataType::Map(Box::new(inner))
+            }
+            ColumnBuilder::Nullable(col) => DataType::Nullable(Box::new(col.builder.data_type())),
             ColumnBuilder::Tuple(fields) => {
                 DataType::Tuple(fields.iter().map(|f| f.data_type()).collect::<Vec<_>>())
             }

src/query/formats/src/field_decoder/fast_values.rs

@@ -168,7 +168,6 @@ impl FastFieldDecoderValues {
             column.push_null();
         } else if reader.ignore_bytes(b"NULL") || reader.ignore_bytes(b"null") {
             column.push_null();
-            return Ok(());
         } else {
             self.read_field(&mut column.builder, reader, positions)?;
             column.validity.push(true);

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

@@ -45,7 +45,7 @@ use common_expression::DataField;
 use common_expression::DataSchema;
 use common_expression::DataSchemaRef;
 use common_expression::Expr;
-use common_expression::Scalar as DataScalar;
+use common_expression::Scalar;
 use common_expression::Value;
 use common_formats::FastFieldDecoderValues;
 use common_io::cursor_ext::ReadBytesExt;
@@ -153,7 +153,7 @@ impl InsertInterpreter {
         }
     }
 
-    async fn prepared_values(&self, values_str: &str) -> Result<(DataSchemaRef, Vec<DataScalar>)> {
+    async fn prepared_values(&self, values_str: &str) -> Result<(DataSchemaRef, Vec<Scalar>)> {
         let settings = self.ctx.get_settings();
         let sql_dialect = settings.get_sql_dialect()?;
         let tokens = tokenize_sql(values_str)?;
@@ -588,17 +588,16 @@ impl ValueSource {
             .map(|f| ColumnBuilder::with_capacity(f.data_type(), estimated_rows))
             .collect::<Vec<_>>();
 
-        let mut rows = 0;
         let format = self.ctx.get_format_settings()?;
         let field_decoder = FastFieldDecoderValues::create_for_insert(format);
 
-        loop {
+        for row in 0.. {
             let _ = reader.ignore_white_spaces();
             if reader.eof() {
                 break;
             }
             // Not the first row
-            if rows != 0 {
+            if row != 0 {
                 reader.must_ignore_byte(b',')?;
             }
 
@@ -611,11 +610,6 @@ impl ValueSource {
                 self.metadata.clone(),
             )
             .await?;
-            rows += 1;
-        }
-
-        if rows == 0 {
-            return Ok(DataBlock::empty_with_schema(self.schema.clone()));
         }
 
         let columns = columns
@@ -702,8 +696,8 @@ impl ValueSource {
                 )
                 .await?;
 
-                for (append_idx, col) in columns.iter_mut().enumerate().take(col_size) {
-                    col.push(values[append_idx].as_ref());
+                for (col, scalar) in columns.iter_mut().zip(values) {
+                    col.push(scalar.as_ref());
                 }
                 reader.set_position(end_pos_of_row);
                 return Ok(());
@@ -798,13 +792,13 @@ async fn fill_default_value(
         if field.data_type().is_nullable() {
             let expr = Expr::Constant {
                 span: None,
-                scalar: DataScalar::Null,
+                scalar: Scalar::Null,
                 data_type: field.data_type().clone(),
             };
             map_exprs.push(expr);
         } else {
             let data_type = field.data_type().clone();
-            let default_value = DataScalar::default_value(&data_type);
+            let default_value = Scalar::default_value(&data_type);
             let expr = Expr::Constant {
                 span: None,
                 scalar: default_value,
@@ -823,7 +817,7 @@ async fn exprs_to_scalar(
     name_resolution_ctx: &NameResolutionContext,
     bind_context: &BindContext,
     metadata: MetadataRef,
-) -> Result<Vec<DataScalar>> {
+) -> Result<Vec<Scalar>> {
     let schema_fields_len = schema.fields().len();
     if exprs.len() != schema_fields_len {
         return Err(ErrorCode::TableSchemaMismatch(format!(
@@ -870,7 +864,7 @@ async fn exprs_to_scalar(
     let one_row_chunk = DataBlock::new(
         vec![BlockEntry {
             data_type: DataType::Number(NumberDataType::UInt8),
-            value: Value::Scalar(DataScalar::Number(NumberScalar::UInt8(1))),
+            value: Value::Scalar(Scalar::Number(NumberScalar::UInt8(1))),
         }],
         1,
     );
@@ -880,13 +874,13 @@ async fn exprs_to_scalar(
         ctx: func_ctx,
     };
     let res = expression_transform.transform(one_row_chunk)?;
-    let data_scalars: Vec<DataScalar> = res
+    let scalars: Vec<Scalar> = res
         .columns()
         .iter()
         .skip(1)
         .map(|col| unsafe { col.value.as_ref().index_unchecked(0).to_owned() })
         .collect();
-    Ok(data_scalars)
+    Ok(scalars)
 }
 
 // TODO:(everpcpc) tmp copy from src/query/sql/src/planner/binder/copy.rs