Merge pull request #4170 from BohuTANG/dev-copy-external-3586

COPY-3586: Support COPY from external location(S3) [PATCH-1]

cli/src/cmds/loads/load.rs

@@ -211,7 +211,7 @@ impl LoadCommand {
                 .parse()
                 .unwrap(),
         );
-        headers.insert("csv_header", skip_header.to_string().parse().unwrap());
+        headers.insert("skip_header", skip_header.to_string().parse().unwrap());
         let progress = execute_load(&cli, &url, headers, data).await?;
 
         let elapsed = start.elapsed();

common/exception/src/exception_code.rs

@@ -155,6 +155,7 @@ build_exceptions! {
     UserAlreadyExists(2202),
     IllegalUserInfoFormat(2203),
     UnknownRole(2204),
+    IllegalUserSettingFormat(2205),
 
     // Meta api error codes.
     DatabaseAlreadyExists(2301),
@@ -169,6 +170,7 @@ build_exceptions! {
     // Stage error codes.
     UnknownStage(2501),
     StageAlreadyExists(2502),
+    IllegalUserStageFormat(2503),
 
     // User defined function error codes.
     IllegalUDFFormat(2601),
@@ -190,6 +192,8 @@ build_exceptions! {
     DalTransportError(3003),
     DalPathNotFound(3004),
     SerdeError(3005),
+    DalS3Error(3006),
+    DalStatError(3007),
 }
 
 // Cache errors [4001, 5000].

common/io/tests/it/options_deserializer.rs

@@ -60,8 +60,8 @@ struct FileFormat {
     pub record_delimiter: String,
     #[serde(default = "default_field_delimiter")]
     pub field_delimiter: String,
-    #[serde(default = "default_csv_header")]
-    pub csv_header: bool,
+    #[serde(default = "default_skip_header")]
+    pub skip_header: i32,
     #[serde(default = "default_compression")]
     pub compression: Compression,
 }
@@ -74,8 +74,8 @@ fn default_field_delimiter() -> String {
     ",".to_string()
 }
 
-fn default_csv_header() -> bool {
-    false
+fn default_skip_header() -> i32 {
+    0
 }
 
 fn default_compression() -> Compression {
@@ -87,15 +87,15 @@ fn test_options_de() {
     let mut values = HashMap::new();
     values.insert("Format".to_string(), "Csv".to_string());
     values.insert("Field_delimiter".to_string(), "/".to_string());
-    values.insert("csv_header".to_string(), "1".to_string());
+    values.insert("skip_header".to_string(), "1".to_string());
     values.insert("compression".to_string(), "GZIP".to_string());
 
     let fmt = FileFormat::deserialize(OptionsDeserializer::new(&values)).unwrap();
     assert_eq!(fmt, FileFormat {
         format: Format::Csv,
         record_delimiter: "\n".to_string(),
         field_delimiter: "/".to_string(),
-        csv_header: true,
+        skip_header: 1,
         compression: Compression::Gzip
     });
 
@@ -104,7 +104,7 @@ fn test_options_de() {
         format: Format::Csv,
         record_delimiter: "\n".to_string(),
         field_delimiter: ",".to_string(),
-        csv_header: false,
+        skip_header: 0,
         compression: Compression::None
     });
 

common/management/tests/it/stage.rs

@@ -19,10 +19,7 @@ use common_exception::Result;
 use common_management::*;
 use common_meta_api::KVApi;
 use common_meta_embedded::MetaEmbedded;
-use common_meta_types::Credentials;
-use common_meta_types::FileFormat;
 use common_meta_types::SeqV;
-use common_meta_types::StageParams;
 use common_meta_types::UserStageInfo;
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
@@ -109,12 +106,7 @@ async fn test_unknown_stage_drop_stage() -> Result<()> {
 fn create_test_stage_info() -> UserStageInfo {
     UserStageInfo {
         stage_name: "mystage".to_string(),
-        stage_params: StageParams::new("test", Credentials {
-            access_key_id: String::from("test"),
-            secret_access_key: String::from("test"),
-        }),
-        file_format: FileFormat::default(),
-        comments: "".to_string(),
+        ..Default::default()
     }
 }
 

common/meta/types/src/user_setting.rs

@@ -52,7 +52,7 @@ impl TryFrom<Vec<u8>> for UserSetting {
     fn try_from(value: Vec<u8>) -> Result<Self> {
         match serde_json::from_slice(&value) {
             Ok(info) => Ok(info),
-            Err(serialize_error) => Err(ErrorCode::IllegalUserInfoFormat(format!(
+            Err(serialize_error) => Err(ErrorCode::IllegalUserSettingFormat(format!(
                 "Cannot deserialize setting from bytes. cause {}",
                 serialize_error
             ))),