improve with reviewers suggestion

Author: ariesdevil

src/meta/types/src/user_stage.rs

@@ -184,16 +184,7 @@ impl FromStr for StageFileFormatType {
 
 impl ToString for StageFileFormatType {
     fn to_string(&self) -> String {
-        match *self {
-            StageFileFormatType::Csv => "CSV".to_string(),
-            StageFileFormatType::Tsv => "Tsv".to_string(),
-            StageFileFormatType::Json => "Json".to_string(),
-            StageFileFormatType::NdJson => "NdJson".to_string(),
-            StageFileFormatType::Avro => "Avro".to_string(),
-            StageFileFormatType::Orc => "Orc".to_string(),
-            StageFileFormatType::Parquet => "Parquet".to_string(),
-            StageFileFormatType::Xml => "Xml".to_string(),
-        }
+        format!("{:?}", *self)
     }
 }
 

src/query/pipeline/sources/src/processors/sources/input_formats/input_context.rs

@@ -29,6 +29,7 @@ use common_formats::ClickhouseFormatType;
 use common_formats::FileFormatOptionsExt;
 use common_formats::FileFormatTypeExt;
 use common_formats::RecordDelimiter;
+use common_meta_types::FileFormatOptions;
 use common_meta_types::StageFileCompression;
 use common_meta_types::StageFileFormatType;
 use common_meta_types::UserStageInfo;
@@ -253,6 +254,48 @@ impl InputContext {
         })
     }
 
+    pub async fn try_create_from_insert_v2(
+        stream_receiver: Receiver<Result<StreamingReadBatch>>,
+        settings: Arc<Settings>,
+        file_format_options: FileFormatOptions,
+        schema: DataSchemaRef,
+        scan_progress: Arc<Progress>,
+        is_multi_part: bool,
+        block_compact_thresholds: BlockCompactThresholds,
+    ) -> Result<Self> {
+        let read_batch_size = settings.get_input_read_buffer_size()? as usize;
+        let format_typ = file_format_options.format.clone();
+        let file_format_options =
+            StageFileFormatType::get_ext_from_stage(file_format_options, &settings)?;
+        let file_format_options = format_typ.final_file_format_options(&file_format_options)?;
+        let format = Self::get_input_format(&format_typ)?;
+        let field_delimiter = file_format_options.get_field_delimiter();
+        let record_delimiter = file_format_options.get_record_delimiter()?;
+        let rows_to_skip = file_format_options.stage.skip_header as usize;
+        let compression = file_format_options.stage.compression;
+
+        let plan = StreamPlan {
+            is_multi_part,
+            compression,
+        };
+
+        Ok(InputContext {
+            format,
+            schema,
+            settings,
+            record_delimiter,
+            read_batch_size,
+            rows_to_skip,
+            field_delimiter,
+            scan_progress,
+            source: InputSource::Stream(Mutex::new(Some(stream_receiver))),
+            plan: InputPlan::StreamingLoad(plan),
+            splits: vec![],
+            block_compact_thresholds,
+            format_options: file_format_options,
+        })
+    }
+
     pub fn num_prefetch_splits(&self) -> Result<usize> {
         Ok(self.settings.get_max_threads()? as usize)
     }

src/query/service/src/servers/http/clickhouse_handler.rs

@@ -320,10 +320,10 @@ pub async fn clickhouse_handler_post(
                 .map_err(InternalServerError)?;
 
             let input_context = Arc::new(
-                InputContext::try_create_from_insert(
-                    option_settings.format.to_string().as_str(),
+                InputContext::try_create_from_insert_v2(
                     rx,
                     ctx.get_settings(),
+                    option_settings.clone(),
                     schema,
                     ctx.get_scan_progress(),
                     false,