Merge branch 'main' into feat/auto_discover_ip

mergify[bot] · web-flow · commit a0a597e782aa · 2022-09-15T08:22:41.000Z
diff --git a/.github/actions/fuse_compat/action.yml b/.github/actions/fuse_compat/action.yml
@@ -12,9 +12,6 @@ inputs:
 runs:
   using: "composite"
   steps:
-    - name: Setup Build Tool
-      uses: ./.github/actions/setup_build_tool
-
     - name: Download artifact
       uses: ./.github/actions/artifact_download
       with:
@@ -26,8 +23,20 @@ runs:
     - name: Test compatibility
       shell: bash
       run: |
-        build-tool bash ./tests/fuse-compat/test-fuse-compat.sh 0.7.150
-        build-tool bash ./tests/fuse-compat/test-fuse-compat.sh 0.7.151
+        docker run --rm --tty --net=host \
+          --user $(id -u):$(id -g) \
+          --env BUILD_PROFILE \
+          --volume "${PWD}:/workspace" \
+          --workdir "/workspace" \
+          datafuselabs/build-tool:sqllogic \
+          bash ./tests/fuse-compat/test-fuse-compat.sh 0.7.150
+        docker run --rm --tty --net=host \
+          --user $(id -u):$(id -g) \
+          --env BUILD_PROFILE \
+          --volume "${PWD}:/workspace" \
+          --workdir "/workspace" \
+          datafuselabs/build-tool:sqllogic \
+          bash ./tests/fuse-compat/test-fuse-compat.sh 0.7.151
 
     - name: Upload failure
       if: failure()
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/docs/doc/30-reference/30-sql/10-dml/dml-copy-into-table.md b/docs/doc/30-reference/30-sql/10-dml/dml-copy-into-table.md
@@ -224,6 +224,16 @@ COPY INTO mytable
   FROM 's3://mybucket/data.csv'
   FILE_FORMAT = (type = 'CSV' field_delimiter = ','  record_delimiter = '\n' skip_header = 1) size_limit=10;
 ```
+  
+`Parquet` file example:
+```sql
+COPY INTO mytable
+  FROM 's3://mybucket/data.parquet'
+  CONNECTION = (
+      ACCESS_KEY_ID = '<your-access-key-ID>'
+      SECRET_ACCESS_KEY = '<your-secret-access-key>')
+  FILE_FORMAT = (type = 'PARQUET');
+```
 
 **Azure Blob storage**
 
@@ -236,4 +246,4 @@ COPY INTO mytable
         ACCOUNT_NAME = '<account_name>'
         ACCOUNT_KEY = '<account_key>'
     )
-```
+```
diff --git a/scripts/ci/ci-run-sqllogic-tests-cluster.sh b/scripts/ci/ci-run-sqllogic-tests-cluster.sh
@@ -15,6 +15,9 @@ if [ $# -gt 0 ]; then
 fi
 echo "Run suites using argument: $RUN_DIR"
 
+echo "pip list"
+python3 -m pip list
+
 echo "Starting databend-sqllogic tests"
 python3 main.py $RUN_DIR
 
diff --git a/scripts/ci/ci-run-sqllogic-tests.sh b/scripts/ci/ci-run-sqllogic-tests.sh
@@ -15,6 +15,9 @@ if [ $# -gt 0 ]; then
 fi
 echo "Run suites using argument: $RUN_DIR"
 
+echo "pip list"
+python3 -m pip list
+
 echo "Starting databend-sqllogic tests"
 python3 main.py --skip-dir=mode $RUN_DIR
 
diff --git a/src/common/contexts/Cargo.toml b/src/common/contexts/Cargo.toml
@@ -12,4 +12,4 @@ test = false
 common-base = { path = "../base" }
 
 async-trait = "0.1.56"
-opendal = { version = "0.16.0", features = ["layers-retry"] }
+opendal = { version = "0.17.0", features = ["layers-retry"] }
diff --git a/src/common/contexts/src/dal/dal_context.rs b/src/common/contexts/src/dal/dal_context.rs
@@ -83,18 +83,14 @@ impl Accessor for DalContext {
             .metadata()
     }
 
-    fn presign(&self, args: &OpPresign) -> Result<PresignedRequest> {
-        self.get_inner()?.presign(args)
+    async fn create(&self, path: &str, args: OpCreate) -> Result<()> {
+        self.get_inner()?.create(path, args).await
     }
 
-    async fn create(&self, args: &OpCreate) -> Result<()> {
-        self.get_inner()?.create(args).await
-    }
-
-    async fn read(&self, args: &OpRead) -> Result<BytesReader> {
+    async fn read(&self, path: &str, args: OpRead) -> Result<BytesReader> {
         let metric = self.metrics.clone();
 
-        self.get_inner()?.read(args).await.map(|r| {
+        self.get_inner()?.read(path, args).await.map(|r| {
             let mut last_pending = None;
             let r = observe_read(r, move |e| {
                 let start = match last_pending {
@@ -117,7 +113,7 @@ impl Accessor for DalContext {
         })
     }
 
-    async fn write(&self, args: &OpWrite, r: BytesReader) -> Result<u64> {
+    async fn write(&self, path: &str, args: OpWrite, r: BytesReader) -> Result<u64> {
         let metric = self.metrics.clone();
 
         let mut last_pending = None;
@@ -139,18 +135,22 @@ impl Accessor for DalContext {
             metric.inc_write_bytes_cost(start.elapsed().as_millis() as u64);
         });
 
-        self.get_inner()?.write(args, Box::new(r)).await
+        self.get_inner()?.write(path, args, Box::new(r)).await
+    }
+
+    async fn stat(&self, path: &str, args: OpStat) -> Result<ObjectMetadata> {
+        self.get_inner()?.stat(path, args).await
     }
 
-    async fn stat(&self, args: &OpStat) -> Result<ObjectMetadata> {
-        self.get_inner()?.stat(args).await
+    async fn delete(&self, path: &str, args: OpDelete) -> Result<()> {
+        self.get_inner()?.delete(path, args).await
     }
 
-    async fn delete(&self, args: &OpDelete) -> Result<()> {
-        self.get_inner()?.delete(args).await
+    async fn list(&self, path: &str, args: OpList) -> Result<DirStreamer> {
+        self.get_inner()?.list(path, args).await
     }
 
-    async fn list(&self, args: &OpList) -> Result<DirStreamer> {
-        self.get_inner()?.list(args).await
+    fn presign(&self, path: &str, args: OpPresign) -> Result<PresignedRequest> {
+        self.get_inner()?.presign(path, args)
     }
 }
diff --git a/src/common/contexts/src/dal/dal_runtime.rs b/src/common/contexts/src/dal/dal_runtime.rs
@@ -83,61 +83,61 @@ impl Accessor for DalRuntime {
             .metadata()
     }
 
-    fn presign(&self, args: &OpPresign) -> Result<PresignedRequest> {
-        self.get_inner()?.presign(args)
-    }
-
-    async fn create(&self, args: &OpCreate) -> Result<()> {
+    async fn create(&self, path: &str, args: OpCreate) -> Result<()> {
         let op = self.get_inner()?;
-        let args = args.clone();
+        let path = path.to_string();
         self.runtime
-            .spawn(async move { op.create(&args).await })
+            .spawn(async move { op.create(&path, args).await })
             .await
             .expect("join must success")
     }
 
-    async fn read(&self, args: &OpRead) -> Result<BytesReader> {
+    async fn read(&self, path: &str, args: OpRead) -> Result<BytesReader> {
         let op = self.get_inner()?;
-        let args = args.clone();
+        let path = path.to_string();
         self.runtime
-            .spawn(async move { op.read(&args).await })
+            .spawn(async move { op.read(&path, args).await })
             .await
             .expect("join must success")
     }
 
-    async fn write(&self, args: &OpWrite, r: BytesReader) -> Result<u64> {
+    async fn write(&self, path: &str, args: OpWrite, r: BytesReader) -> Result<u64> {
         let op = self.get_inner()?;
-        let args = args.clone();
+        let path = path.to_string();
         self.runtime
-            .spawn(async move { op.write(&args, r).await })
+            .spawn(async move { op.write(&path, args, r).await })
             .await
             .expect("join must success")
     }
 
-    async fn stat(&self, args: &OpStat) -> Result<ObjectMetadata> {
+    async fn stat(&self, path: &str, args: OpStat) -> Result<ObjectMetadata> {
         let op = self.get_inner()?;
-        let args = args.clone();
+        let path = path.to_string();
         self.runtime
-            .spawn(async move { op.stat(&args).await })
+            .spawn(async move { op.stat(&path, args).await })
             .await
             .expect("join must success")
     }
 
-    async fn delete(&self, args: &OpDelete) -> Result<()> {
+    async fn delete(&self, path: &str, args: OpDelete) -> Result<()> {
         let op = self.get_inner()?;
-        let args = args.clone();
+        let path = path.to_string();
         self.runtime
-            .spawn(async move { op.delete(&args).await })
+            .spawn(async move { op.delete(&path, args).await })
             .await
             .expect("join must success")
     }
 
-    async fn list(&self, args: &OpList) -> Result<DirStreamer> {
+    async fn list(&self, path: &str, args: OpList) -> Result<DirStreamer> {
         let op = self.get_inner()?;
-        let args = args.clone();
+        let path = path.to_string();
         self.runtime
-            .spawn(async move { op.list(&args).await })
+            .spawn(async move { op.list(&path, args).await })
             .await
             .expect("join must success")
     }
+
+    fn presign(&self, path: &str, args: OpPresign) -> Result<PresignedRequest> {
+        self.get_inner()?.presign(path, args)
+    }
 }
diff --git a/src/common/storage/Cargo.toml b/src/common/storage/Cargo.toml
@@ -18,7 +18,7 @@ anyhow = "1.0.64"
 backon = "0.1.0"
 globiter = "0.1.0"
 once_cell = "1.12.0"
-opendal = { version = "0.16.0", features = [
+opendal = { version = "0.17.0", features = [
     "layers-retry",
     "layers-tracing",
     "layers-metrics",
diff --git a/src/query/catalog/Cargo.toml b/src/query/catalog/Cargo.toml
@@ -26,5 +26,5 @@ common-users = { path = "../users" }
 
 async-trait = "0.1.56"
 dyn-clone = "1.0.6"
-opendal = { version = "0.16.0", features = ["layers-retry"] }
+opendal = { version = "0.17.0", features = ["layers-retry"] }
 parking_lot = "0.12.1"
diff --git a/src/query/config/Cargo.toml b/src/query/config/Cargo.toml
@@ -21,7 +21,7 @@ common-users = { path = "../users" }
 clap = { version = "3.2.5", features = ["derive", "env"] }
 hex = "0.4.3"
 once_cell = "1.12.0"
-opendal = { version = "0.16.0", features = ["layers-retry", "compress"], optional = true }
+opendal = { version = "0.17.0", features = ["layers-retry", "compress"], optional = true }
 semver = "1.0.10"
 serde = { version = "1.0.137", features = ["derive"] }
 serfig = "0.0.2"
diff --git a/src/query/pipeline/sources/Cargo.toml b/src/query/pipeline/sources/Cargo.toml
@@ -23,5 +23,5 @@ common-streams = { path = "../../streams" }
 async-trait = { version = "0.1.0", package = "async-trait-fn" }
 futures = "0.3.21"
 futures-util = "0.3.21"
-opendal = { version = "0.16.0", features = ["layers-retry", "compress"] }
+opendal = { version = "0.17.0", features = ["layers-retry", "compress"] }
 parking_lot = "0.12.1"
diff --git a/src/query/service/Cargo.toml b/src/query/service/Cargo.toml
@@ -111,7 +111,7 @@ num = "0.4.0"
 num_cpus = "1.13.1"
 octocrab = "0.16.0"
 once_cell = "1.12.0"
-opendal = { version = "0.16.0", features = ["layers-retry", "layers-tracing", "layers-metrics", "compress"] }
+opendal = { version = "0.17.0", features = ["layers-retry", "layers-tracing", "layers-metrics", "compress"] }
 opensrv-mysql = "0.2.0"
 openssl = { version = "0.10.40", features = ["vendored"] }
 parking_lot = "0.12.1"
diff --git a/src/query/service/tests/it/storages/fuse/io.rs b/src/query/service/tests/it/storages/fuse/io.rs
@@ -227,14 +227,14 @@ where
     P: Debug + Send,
     P: Iterator<Item = IOError>,
 {
-    async fn read(&self, _args: &OpRead) -> std::io::Result<BytesReader> {
+    async fn read(&self, _: &str, _: OpRead) -> std::io::Result<BytesReader> {
         let v = &mut (*self.err.lock());
         let err = v.take_err();
         v.increase_read_op_count();
         Err(err)
     }
 
-    async fn write(&self, _args: &OpWrite, _r: BytesReader) -> std::io::Result<u64> {
+    async fn write(&self, _: &str, _: OpWrite, _: BytesReader) -> std::io::Result<u64> {
         let v = &mut (*self.err.lock());
         let err = v.take_err();
         v.increase_write_op_count();
diff --git a/src/query/storages/fuse/Cargo.toml b/src/query/storages/fuse/Cargo.toml
@@ -37,7 +37,7 @@ backoff = { version = "0.4.0", features = ["futures", "tokio"] }
 chrono = "0.4.19"
 futures = "0.3.21"
 futures-util = "0.3.21"
-opendal = { version = "0.16.0", features = ["layers-retry"] }
+opendal = { version = "0.17.0", features = ["layers-retry"] }
 serde = { version = "1.0.137", features = ["derive"] }
 serde_json = "1.0.81"
 tracing = "0.1.35"
diff --git a/src/query/storages/hive/Cargo.toml b/src/query/storages/hive/Cargo.toml
@@ -31,7 +31,7 @@ common-storages-util = { path = "../util" }
 async-recursion = "1.0.0"
 async-trait = "0.1.56"
 futures = "0.3.21"
-opendal = { version = "0.16.0", features = ["layers-retry"] }
+opendal = { version = "0.17.0", features = ["layers-retry"] }
 serde = { version = "1.0.137", features = ["derive"] }
 thrift = "0.15.0"
 tracing = "0.1.35"
diff --git a/src/query/storages/preludes/Cargo.toml b/src/query/storages/preludes/Cargo.toml
@@ -46,7 +46,7 @@ futures-util = "0.3.21"
 itertools = "0.10.3"
 num_cpus = "1.13.1"
 once_cell = "1.12.0"
-opendal = { version = "0.16.0", features = ["layers-retry"] }
+opendal = { version = "0.17.0", features = ["layers-retry"] }
 parking_lot = "0.12.1"
 reqwest = "0.11.11"
 semver = "1.0.10"
diff --git a/src/query/storages/share/Cargo.toml b/src/query/storages/share/Cargo.toml
@@ -16,7 +16,7 @@ common-exception = { path = "../../../common/exception" }
 common-meta-app = { path = "../../../meta/app" }
 common-storages-util = { path = "../util" }
 
-opendal = { version = "0.16.0", features = ["layers-retry"] }
+opendal = { version = "0.17.0", features = ["layers-retry"] }
 serde = { version = "1.0.137", features = ["derive"] }
 serde_json = "1.0.81"
 
diff --git a/src/query/streams/Cargo.toml b/src/query/streams/Cargo.toml
@@ -33,4 +33,4 @@ serde_json = { version = "1.0.81", default-features = false, features = ["preser
 tempfile = "3.3.0"
 
 [dev-dependencies]
-opendal = { version = "0.16.0", features = ["layers-retry", "compress"] }
+opendal = { version = "0.17.0", features = ["layers-retry", "compress"] }
diff --git a/tests/compat/test-compat.sh b/tests/compat/test-compat.sh
diff --git a/tests/fuse-compat/test-fuse-compat.sh b/tests/fuse-compat/test-fuse-compat.sh
diff --git a/tests/poetry.lock b/tests/poetry.lock
diff --git a/tests/pyproject.toml b/tests/pyproject.toml
