Support copy from stage

.github/actions/test_stateful_standalone/action.yml

@@ -52,3 +52,15 @@ runs:
       shell: bash
       run: |
         bash ./scripts/ci/ci-run-stateful-tests-standalone-s3.sh
+
+    - name: Upload artifact
+      uses: actions/upload-artifact@v2
+      if: failure()
+      with:
+        path: |
+          _local_fs/
+          _logs*/
+          _meta*/
+          metasrv/_logs*/
+          query/_logs*/
+          store/_logs*/

common/ast/Cargo.toml

@@ -19,7 +19,7 @@ common-functions = { path = "../functions" }
 # TODO (andylokandy): Use the version from crates.io once
 # https://github.com/brendanzab/codespan/pull/331 is released.
 codespan-reporting = { git = "https://github.com/brendanzab/codespan", rev = "c84116f5" }
-sqlparser = { git = "https://github.com/datafuse-extras/sqlparser-rs", rev = "472f5b6" }
+sqlparser = { git = "https://github.com/datafuse-extras/sqlparser-rs", rev = "64795ad" }
 
 # Crates.io dependencies
 async-trait = "0.1.52"

common/contexts/Cargo.toml

@@ -10,4 +10,4 @@ test = false
 
 [dependencies]
 async-trait = "0.1.52"
-opendal = "0.2.2"
+opendal = "0.2.3"

common/contexts/src/dal/dal_context.rs

@@ -17,13 +17,15 @@ use std::time::Instant;
 use async_trait::async_trait;
 use opendal::error::Result as DalResult;
 use opendal::ops::OpDelete;
+use opendal::ops::OpList;
 use opendal::ops::OpRead;
 use opendal::ops::OpStat;
 use opendal::ops::OpWrite;
 use opendal::readers::ObserveReader;
 use opendal::readers::ReadEvent;
 use opendal::Accessor;
 use opendal::BoxedAsyncReader;
+use opendal::BoxedObjectStream;
 use opendal::Layer;
 use opendal::Metadata;
 
@@ -102,4 +104,8 @@ impl Accessor for DalContext {
     async fn delete(&self, args: &OpDelete) -> DalResult<()> {
         self.inner.as_ref().unwrap().delete(args).await
     }
+
+    async fn list(&self, args: &OpList) -> DalResult<BoxedObjectStream> {
+        self.inner.as_ref().unwrap().list(args).await
+    }
 }

common/exception/Cargo.toml

@@ -16,7 +16,7 @@ common-arrow = { path = "../arrow" }
 anyhow = "1.0.55"
 backtrace = "0.3.64"
 octocrab = "0.15.4"
-opendal = "0.2.2"
+opendal = "0.2.3"
 paste = "1.0.6"
 prost = "0.9.0"
 serde = { version = "1.0.136", features = ["derive"] }
@@ -26,4 +26,4 @@ tonic = "0.6.2"
 
 # Github dependencies
 bincode = { git = "https://github.com/datafuse-extras/bincode", rev = "fd3f9ff" }
-sqlparser = { git = "https://github.com/datafuse-extras/sqlparser-rs", rev = "472f5b6" }
+sqlparser = { git = "https://github.com/datafuse-extras/sqlparser-rs", rev = "64795ad" }

common/functions/Cargo.toml

@@ -19,7 +19,7 @@ common-exception = { path = "../exception" }
 common-io = { path = "../io" }
 
 # Github dependencies
-sqlparser = { git = "https://github.com/datafuse-extras/sqlparser-rs", rev = "472f5b6" }
+sqlparser = { git = "https://github.com/datafuse-extras/sqlparser-rs", rev = "64795ad" }
 
 # Crates.io dependencies
 base64 = "0.13.0"

common/io/Cargo.toml

@@ -22,7 +22,7 @@ bincode = { git = "https://github.com/datafuse-extras/bincode", rev = "fd3f9ff"
 # Crates.io dependencies
 bytes = "1.1.0"
 futures = "0.3.21"
-opendal = "0.2.2"
+opendal = "0.2.3"
 serde = { version = "1.0.136", features = ["derive"] }
 
 [dev-dependencies]

common/io/src/files/file_s3.rs

@@ -18,7 +18,6 @@ use futures::StreamExt;
 use opendal::credential::Credential;
 use opendal::ObjectMode;
 use opendal::Operator;
-use opendal::Reader;
 
 pub struct S3File {}
 
@@ -51,31 +50,9 @@ impl S3File {
         Ok(opendal::Operator::new(accessor))
     }
 
-    // Read a file, returns the reader.
-    // file_name is the Some(/path/to/path/xx.csv)
-    pub async fn read(
-        file_name: Option<String>,
-        s3_endpoint: &str,
-        s3_bucket: &str,
-        aws_key_id: &str,
-        aws_secret_key: &str,
-    ) -> Result<Reader> {
-        let operator = Self::open(s3_endpoint, s3_bucket, aws_key_id, aws_secret_key).await?;
-        let path = file_name.unwrap_or_else(|| "".to_string());
-        Ok(operator.object(&path).reader())
-    }
-
     // Get the files in the path.
-    pub async fn list(
-        s3_endpoint: &str,
-        s3_bucket: &str,
-        path: &str,
-        aws_key_id: &str,
-        aws_secret_key: &str,
-    ) -> Result<Vec<String>> {
+    pub async fn list(operator: &Operator, path: &str) -> Result<Vec<String>> {
         let mut list: Vec<String> = vec![];
-        let operator = Self::open(s3_endpoint, s3_bucket, aws_key_id, aws_secret_key).await?;
-
         // Check the path object mode is DIR or FILE.
         let mode = operator.object(path).metadata().await?.mode();
         match mode {

common/io/src/utils.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::cmp;
+use std::path::PathBuf;
 
 use bincode::Options;
 use bytes::BufMut;
@@ -91,3 +92,16 @@ pub fn deserialize_from_slice<T: serde::de::DeserializeOwned>(slice: &mut &[u8])
 
     Ok(value)
 }
+
+#[inline]
+pub fn get_abs_path(root: &str, path: &str) -> String {
+    // Joining an absolute path replaces the existing path, we need to
+    // normalize it before.
+    let path = path
+        .split('/')
+        .filter(|v| !v.is_empty())
+        .collect::<Vec<&str>>()
+        .join("/");
+
+    PathBuf::from(root).join(path).to_string_lossy().to_string()
+}

common/planners/src/plan_user_stage_create.rs

@@ -21,6 +21,7 @@ use common_meta_types::UserStageInfo;
 #[derive(serde::Serialize, serde::Deserialize, Clone, Debug, PartialEq)]
 pub struct CreateUserStagePlan {
     pub if_not_exists: bool,
+    pub tenant: String,
     pub user_stage_info: UserStageInfo,
 }
 

common/planners/src/plan_user_stage_describe.rs

@@ -21,12 +21,12 @@ pub struct DescribeUserStagePlan {
 impl DescribeUserStagePlan {
     pub fn schema(&self) -> DataSchemaRef {
         DataSchemaRefExt::create(vec![
-            DataField::new("parent_properties", Vu8::to_data_type()),
-            DataField::new("properties", Vu8::to_data_type()),
-            DataField::new("property_types", Vu8::to_data_type()),
-            DataField::new("property_values", Vu8::to_data_type()),
-            DataField::new("property_defaults", Vu8::to_data_type()),
-            DataField::new("property_changed", bool::to_data_type()),
+            DataField::new("name", Vu8::to_data_type()),
+            DataField::new("stage_type", Vu8::to_data_type()),
+            DataField::new("stage_params", Vu8::to_data_type()),
+            DataField::new("copy_options", Vu8::to_data_type()),
+            DataField::new("file_format_options", Vu8::to_data_type()),
+            DataField::new("comment", Vu8::to_data_type()),
         ])
     }
 }

common/streams/Cargo.toml

@@ -33,4 +33,4 @@ tempfile = "3.3.0"
 tokio-stream = { version = "0.1.8", features = ["net"] }
 
 [dev-dependencies]
-opendal = "0.2.2"
+opendal = "0.2.3"

docs/doc/03-reference/03-sql/01-ddl/04-stage/_category_.yml

@@ -1,4 +1,4 @@
-label: 'Data Loading DDL'
+label: 'DDL for User Stage'
 link:
   type: generated-index
-  title: 'Data Loading DDL'
+  title: 'DDL for User Stage'

docs/doc/03-reference/03-sql/01-ddl/04-stage/ddl-create-stage.md

@@ -0,0 +1,26 @@
+---
+title: CREATE Stage
+---
+
+Create a new stage.
+
+## Syntax
+
+- Create Internal Stage
+
+```sql
+CREATE STAGE [ IF NOT EXISTS ] <stage_name>;
+```
+
+- Create External Stage
+```sql
+CREATE STAGE [ IF NOT EXISTS ] <stage_name> uri = 's3://<bucket>[/<path>]' credentials=(access_key_id='1a2b3c' secret_access_key='4x5y6z');
+```
+
+## Examples
+
+```sql
+mysql> CREATE STAGE test_stage url='s3://load/files/' credentials=(access_key_id='1a2b3c' secret_access_key='4x5y6z');
+
+mysql> desc STAGE;
+```

docs/doc/03-reference/03-sql/01-ddl/04-stage/ddl-drop-stage.md

@@ -0,0 +1,17 @@
+---
+title: DROP Stage
+---
+
+Drop a stage.
+
+## Syntax
+
+```sql
+DROP STAGE [ IF EXISTS ] <stage_name>;
+```
+
+## Examples
+
+```sql
+mysql> drop stage if exists test_stage;
+```

query/Cargo.toml

@@ -55,7 +55,7 @@ common-tracing = { path = "../common/tracing" }
 bincode = { git = "https://github.com/datafuse-extras/bincode", rev = "fd3f9ff" }
 cargo-license = { git = "https://github.com/datafuse-extras/cargo-license", rev = "f1ce4a2" }
 msql-srv = { git = "https://github.com/datafuse-extras/msql-srv", rev = "af29f7b" }
-sqlparser = { git = "https://github.com/datafuse-extras/sqlparser-rs", rev = "472f5b6" }
+sqlparser = { git = "https://github.com/datafuse-extras/sqlparser-rs", rev = "64795ad" }
 
 # Crates.io dependencies
 ahash = "0.7.6"
@@ -85,7 +85,7 @@ num = "0.4.0"
 num_cpus = "1.13.1"
 octocrab = "0.15.4"
 once_cell = "1.9.0"
-opendal = "0.2.2"
+opendal = "0.2.3"
 parquet-format-async-temp = "0.2.0"
 paste = "1.0.6"
 petgraph = "0.6.0"

query/src/interpreters/interpreter_copy.rs

@@ -37,6 +37,7 @@ use crate::interpreters::InterpreterPtr;
 use crate::pipelines::new::executor::PipelinePullingExecutor;
 use crate::pipelines::new::QueryPipelineBuilder;
 use crate::sessions::QueryContext;
+use crate::storages::ExternalSource;
 
 pub struct CopyInterpreter {
     ctx: Arc<QueryContext>,
@@ -61,7 +62,6 @@ impl CopyInterpreter {
                 match &storage {
                     StageStorage::S3(s3) => {
                         let path = &s3.path;
-
                         // Here we add the path to the file: /path/to/path/file1.
                         if !self.plan.files.is_empty() {
                             let mut files_with_path = vec![];
@@ -71,13 +71,8 @@ impl CopyInterpreter {
                             }
                             Ok(files_with_path)
                         } else {
-                            let endpoint = &self.ctx.get_config().storage.s3.endpoint_url;
-                            let bucket = &s3.bucket;
-
-                            let key_id = &s3.credentials_aws_key_id;
-                            let secret_key = &s3.credentials_aws_secret_key;
-
-                            S3File::list(endpoint, bucket, path, key_id, secret_key).await
+                            let op = ExternalSource::get_op(&self.ctx, table_info).await?;
+                            S3File::list(&op, path).await
                         }
                     }
                 }