Merge pull request #10630 from sundy-li/copy-transform

Author: sundy-li

src/meta/app/src/principal/user_stage.rs

@@ -106,6 +106,12 @@ pub enum StageFileCompression {
     None,
 }
 
+impl StageFileFormatType {
+    pub fn has_inner_schema(&self) -> bool {
+        matches!(self, StageFileFormatType::Parquet)
+    }
+}
+
 impl Default for StageFileCompression {
     fn default() -> Self {
         Self::None

src/query/service/src/interpreters/interpreter_copy.rs

@@ -24,6 +24,7 @@ use common_exception::ErrorCode;
 use common_exception::Result;
 use common_expression::infer_table_schema;
 use common_expression::DataField;
+use common_expression::DataSchema;
 use common_expression::DataSchemaRef;
 use common_expression::DataSchemaRefExt;
 use common_meta_app::principal::StageInfo;
@@ -41,6 +42,7 @@ use tracing::info;
 use crate::interpreters::common::append2table;
 use crate::interpreters::Interpreter;
 use crate::interpreters::SelectInterpreter;
+use crate::pipelines::processors::transforms::TransformRuntimeCastSchema;
 use crate::pipelines::processors::TransformCastSchema;
 use crate::pipelines::processors::TransformLimit;
 use crate::pipelines::PipelineBuildResult;
@@ -334,6 +336,26 @@ impl CopyInterpreter {
             )?;
         }
 
+        if stage_table_info
+            .stage_info
+            .file_format_options
+            .format
+            .has_inner_schema()
+        {
+            let dst_schema: Arc<DataSchema> = Arc::new(to_table.schema().into());
+            let func_ctx = self.ctx.get_function_context()?;
+            build_res.main_pipeline.add_transform(
+                |transform_input_port, transform_output_port| {
+                    TransformRuntimeCastSchema::try_create(
+                        transform_input_port,
+                        transform_output_port,
+                        dst_schema.clone(),
+                        func_ctx,
+                    )
+                },
+            )?;
+        }
+
         // Build append data pipeline.
         to_table.append_data(
             ctx.clone(),

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

@@ -378,20 +378,23 @@ impl Interpreter for InsertInterpreter {
                     .format
                     .exec_stream(input_context.clone(), &mut build_res.main_pipeline)?;
 
-                if Ok(StageFileFormatType::Parquet) == StageFileFormatType::from_str(format) {
-                    let dest_schema = plan.schema();
-                    let func_ctx = self.ctx.get_function_context()?;
-
-                    build_res.main_pipeline.add_transform(
-                        |transform_input_port, transform_output_port| {
-                            TransformRuntimeCastSchema::try_create(
-                                transform_input_port,
-                                transform_output_port,
-                                dest_schema.clone(),
-                                func_ctx,
-                            )
-                        },
-                    )?;
+                match StageFileFormatType::from_str(format) {
+                    Ok(f) if f.has_inner_schema() => {
+                        let dest_schema = plan.schema();
+                        let func_ctx = self.ctx.get_function_context()?;
+
+                        build_res.main_pipeline.add_transform(
+                            |transform_input_port, transform_output_port| {
+                                TransformRuntimeCastSchema::try_create(
+                                    transform_input_port,
+                                    transform_output_port,
+                                    dest_schema.clone(),
+                                    func_ctx,
+                                )
+                            },
+                        )?;
+                    }
+                    _ => {}
                 }
             }
             InsertInputSource::StreamingWithFileFormat(format_options, _, input_context) => {
@@ -400,7 +403,7 @@ impl Interpreter for InsertInterpreter {
                     .format
                     .exec_stream(input_context.clone(), &mut build_res.main_pipeline)?;
 
-                if StageFileFormatType::Parquet == format_options.format {
+                if format_options.format.has_inner_schema() {
                     let dest_schema = plan.schema();
                     let func_ctx = self.ctx.get_function_context()?;
 

tests/suites/1_stateful/00_copy/00_0002_copy_from_fs_location.result

@@ -6,3 +6,4 @@
 199	2020.0	769
 199	2020.0	769
 199	2020.0	769
+199	2020.0000	769.00

tests/suites/1_stateful/00_copy/00_0002_copy_from_fs_location.sh

@@ -39,5 +39,13 @@ for i in "${copy_from_location_cases[@]}"; do
   echo "truncate table ontime200" | $MYSQL_CLIENT_CONNECT
 done
 
+
+## test copy from parquet with auto cast schema
+echo "create table test_decimal(Year Decimal(15,2), DayOfWeek Decimal(15,2),  OriginCityMarketID Decimal(15,2) )" | $MYSQL_CLIENT_CONNECT
+echo "copy into test_decimal from 'fs://${DATADIR}/' PATTERN = 'ontime.*parquet' FILE_FORMAT = (type = PARQUET)" | $MYSQL_CLIENT_CONNECT
+echo "select count(1), avg(Year), sum(DayOfWeek)  from test_decimal" | $MYSQL_CLIENT_CONNECT
+
+
 ## Drop table
 echo "drop table if exists ontime200;" | $MYSQL_CLIENT_CONNECT
+echo "drop table if exists test_decimal;" | $MYSQL_CLIENT_CONNECT