Merge branch 'main' of https://github.com/datafuselabs/databend into query-tests

Signed-off-by: Chojan Shang <[email protected]>

Author: PsiACE

.github/workflows/stateless-tests-standalone.yml

@@ -53,7 +53,7 @@ jobs:
       - name: Minio Setup for (ubuntu-latest only)
         if: matrix.config.os == 'ubuntu-latest'
         run: |
-          docker run -d -p 9000:9000 --name minio \
+          docker run -d -p 9900:9000 --name minio \
                   -e "MINIO_ACCESS_KEY=minioadmin" \
                   -e "MINIO_SECRET_KEY=minioadmin" \
                   -v /tmp/data:/data \
@@ -63,8 +63,9 @@ jobs:
           export AWS_ACCESS_KEY_ID=minioadmin
           export AWS_SECRET_ACCESS_KEY=minioadmin
           export AWS_EC2_METADATA_DISABLED=true
+          aws --endpoint-url http://127.0.0.1:9900/ s3 mb s3://testbucket
+          aws --endpoint-url http://127.0.0.1:9900/ s3 cp tests/data s3://testbucket/tests/data  --recursive
 
-          aws --endpoint-url http://127.0.0.1:9000/ s3 mb s3://testbucket
 
       - name: Run Stateless Tests with Standalone mode (ubuntu-latest only)
         if: matrix.config.os == 'ubuntu-latest'

.gitignore

@@ -43,3 +43,6 @@ tests/perfs/*-result.json
 
 # python
 venv/
+*.pyc
+__pycache__/*
+

Cargo.lock

@@ -1,28 +1,28 @@
 [package]
-name = "common-arrow"
-version = "0.1.0"
 authors = ["Databend Authors <[email protected]>"]
+edition = "2021"
 license = "Apache-2.0"
+name = "common-arrow"
 publish = false
-edition = "2021"
+version = "0.1.0"
 
 [lib]
 doctest = false
 test = false
 
 [features]
-default = ["arrow-default", "parquet-default"]
 arrow-default = ["arrow/compute", "arrow/regex", "arrow/merge_sort", "arrow/io_csv", "arrow/io_parquet", "arrow/io_json", "arrow/io_flight"]
+default = ["arrow-default", "parquet-default"]
 parquet-default = ["parquet2/stream", "parquet2/lz4"]
 simd = ["arrow/simd"]
 
 [dependencies] # In alphabetical order
 # Workspace dependencies
 
 # Github dependencies
-arrow = { package = "arrow2", git="https://github.com/datafuse-extras/arrow2", default-features = false, rev = "83d828c" }
-arrow-format = { version = "0.3.0", features = ["flight-data", "flight-service"]}
-parquet2 = { version = "0.6", default_features = false }
+arrow = {package = "arrow2", git = "https://github.com/datafuse-extras/arrow2", default-features = false, rev = "83d828c"}
+arrow-format = {version = "0.3.0", features = ["flight-data", "flight-service"]}
+parquet2 = {version = "0.6", default_features = false}
 # Crates.io dependencies
 
 [dev-dependencies]

common/arrow/Cargo.toml

@@ -30,7 +30,6 @@ pub use profiling::Profiling;
 pub use progress::Progress;
 pub use progress::ProgressCallback;
 pub use progress::ProgressValues;
-pub use runtime::BlockingWait;
 pub use runtime::Dropper;
 pub use runtime::Runtime;
 pub use runtime::TrySpawn;

common/base/src/lib.rs

@@ -13,14 +13,11 @@
 // limitations under the License.
 
 use std::future::Future;
-use std::sync::mpsc::channel;
 use std::sync::Arc;
 use std::thread;
-use std::time::Duration;
 
 use common_exception::ErrorCode;
 use common_exception::Result;
-use futures::future::Either;
 use tokio::runtime::Handle;
 use tokio::sync::oneshot;
 use tokio::task::JoinHandle;
@@ -67,79 +64,6 @@ impl<S: TrySpawn> TrySpawn for Arc<S> {
     }
 }
 
-/// Blocking wait for a future to complete.
-///
-/// This trait turns an `async` function into `sync`.
-/// It is meant to provide convenience for building a proof-of-concept demo or else.
-/// Always avoid using it in a real world production,
-/// unless **you KNOW what you are doing**:
-///
-/// - `wait()` runs the future in current thread and **blocks** current thread until the future is finished.
-/// - `wait_in(rt)` runs the future in the specified runtime, and **blocks** current thread until the future is finished.
-pub trait BlockingWait
-where
-    Self: Future + Send + 'static,
-    Self::Output: Send + 'static,
-{
-    /// Runs the future and blocks current thread.
-    ///
-    /// ```ignore
-    /// use runtime::BlockingWait;
-    /// async fn five() -> u8 { 5 }
-    /// assert_eq!(5, five().wait());
-    /// ```
-    fn wait(self, timeout: Option<Duration>) -> Result<Self::Output>;
-
-    /// Runs the future in provided runtime and blocks current thread.
-    fn wait_in<RT: TrySpawn>(self, rt: &RT, timeout: Option<Duration>) -> Result<Self::Output>;
-}
-
-impl<T> BlockingWait for T
-where
-    T: Future + Send + 'static,
-    T::Output: Send + 'static,
-{
-    fn wait(self, timeout: Option<Duration>) -> Result<T::Output> {
-        match timeout {
-            None => Ok(futures::executor::block_on(self)),
-            Some(d) => {
-                let rt = tokio::runtime::Builder::new_current_thread()
-                    .enable_time()
-                    .build()
-                    .map_err(|e| ErrorCode::TokioError(format!("{}", e)))?;
-
-                rt.block_on(async move {
-                    let sl = tokio::time::sleep(d);
-                    let sl = Box::pin(sl);
-                    let task = Box::pin(self);
-
-                    match futures::future::select(sl, task).await {
-                        Either::Left((_, _)) => Err::<T::Output, ErrorCode>(ErrorCode::Timeout(
-                            format!("timeout: {:?}", d),
-                        )),
-                        Either::Right((res, _)) => Ok(res),
-                    }
-                })
-            }
-        }
-    }
-
-    fn wait_in<RT: TrySpawn>(self, rt: &RT, timeout: Option<Duration>) -> Result<T::Output> {
-        let (tx, rx) = channel();
-        let _jh = rt.spawn(async move {
-            let r = self.await;
-            let _ = tx.send(r);
-        });
-        let reply = match timeout {
-            Some(to) => rx
-                .recv_timeout(to)
-                .map_err(|timeout_err| ErrorCode::Timeout(timeout_err.to_string()))?,
-            None => rx.recv().map_err(ErrorCode::from_std_error)?,
-        };
-        Ok(reply)
-    }
-}
-
 /// Tokio Runtime wrapper.
 /// If a runtime is in an asynchronous context, shutdown it first.
 pub struct Runtime {

common/base/src/runtime.rs

@@ -16,9 +16,7 @@ use std::sync::Arc;
 use std::sync::Mutex;
 
 use common_base::*;
-use common_exception::ErrorCode;
 use common_exception::Result;
-use tokio::time::Duration;
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
 async fn test_runtime() -> Result<()> {
@@ -61,60 +59,3 @@ async fn test_runtime() -> Result<()> {
 
     Ok(())
 }
-
-#[test]
-fn test_blocking_wait() -> Result<()> {
-    async fn five() -> Result<u8> {
-        Ok(5)
-    }
-
-    let res = five().wait(None)?;
-    assert!(res.is_ok());
-    assert_eq!(5, res.unwrap());
-
-    let rt = Runtime::with_default_worker_threads().unwrap();
-
-    let res = five().wait_in(&rt, None)?;
-    assert!(res.is_ok());
-    assert_eq!(5, res.unwrap());
-
-    Ok(())
-}
-
-#[test]
-fn test_blocking_wait_timeout() -> Result<()> {
-    async fn sleep_5_sec() -> Result<()> {
-        tokio::time::sleep(Duration::from_millis(5000)).await;
-        Ok(())
-    }
-
-    let res = sleep_5_sec().wait(Some(Duration::from_millis(1000)));
-    assert!(res.is_err());
-    assert_eq!(ErrorCode::Timeout("").code(), res.unwrap_err().code());
-
-    let rt = Runtime::with_default_worker_threads().unwrap();
-
-    let res = sleep_5_sec().wait_in(&rt, Some(Duration::from_millis(1000)));
-    assert!(res.is_err());
-    assert_eq!(ErrorCode::Timeout("").code(), res.unwrap_err().code());
-
-    Ok(())
-}
-
-#[test]
-fn test_blocking_wait_no_timeout() -> Result<()> {
-    async fn sleep_1_sec() -> Result<()> {
-        tokio::time::sleep(Duration::from_millis(1000)).await;
-        Ok(())
-    }
-
-    let res = sleep_1_sec().wait(Some(Duration::from_millis(5000)))?;
-    assert!(res.is_ok());
-
-    let rt = Runtime::with_default_worker_threads().unwrap();
-
-    let res = sleep_1_sec().wait_in(&rt, Some(Duration::from_millis(5000)))?;
-    assert!(res.is_ok());
-
-    Ok(())
-}

common/base/tests/it/runtime.rs

@@ -186,6 +186,7 @@ build_exceptions! {
     BadPredicateRows(56),
     SHA1CheckFailed(57),
     UnknownColumn(58),
+    InvalidSourceFormat(59),
 
     // uncategorized
     UnexpectedResponseType(600),

common/exception/src/exception.rs

@@ -27,6 +27,7 @@ use crate::scalars::LogFunction;
 use crate::scalars::PiFunction;
 use crate::scalars::RadiansFunction;
 use crate::scalars::SignFunction;
+use crate::scalars::SqrtFunction;
 use crate::scalars::TrigonometricAcosFunction;
 use crate::scalars::TrigonometricAsinFunction;
 use crate::scalars::TrigonometricAtan2Function;
@@ -63,5 +64,6 @@ impl MathsFunction {
         factory.register("atan", TrigonometricAtanFunction::desc());
         factory.register("atan2", TrigonometricAtan2Function::desc());
         factory.register("sign", SignFunction::desc());
+        factory.register("sqrt", SqrtFunction::desc());
     }
 }

common/functions/src/scalars/maths/math.rs

@@ -22,6 +22,7 @@ mod log;
 mod math;
 mod pi;
 mod sign;
+mod sqrt;
 mod trigonometric;
 
 pub use abs::AbsFunction;
@@ -38,6 +39,7 @@ pub use log::LogFunction;
 pub use math::MathsFunction;
 pub use pi::PiFunction;
 pub use sign::SignFunction;
+pub use sqrt::SqrtFunction;
 pub use trigonometric::Trigonometric;
 pub use trigonometric::TrigonometricAcosFunction;
 pub use trigonometric::TrigonometricAsinFunction;

common/functions/src/scalars/maths/mod.rs

@@ -0,0 +1,94 @@
+// Copyright 2020 Datafuse Labs.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::fmt;
+
+use common_datavalues::prelude::*;
+use common_datavalues::DataSchema;
+use common_datavalues::DataType;
+use common_exception::ErrorCode;
+use common_exception::Result;
+
+use crate::scalars::function_factory::FunctionDescription;
+use crate::scalars::function_factory::FunctionFeatures;
+use crate::scalars::Function;
+
+#[derive(Clone)]
+pub struct SqrtFunction {
+    display_name: String,
+}
+
+impl SqrtFunction {
+    pub fn try_create(display_name: &str) -> Result<Box<dyn Function>> {
+        Ok(Box::new(SqrtFunction {
+            display_name: display_name.to_string(),
+        }))
+    }
+
+    pub fn desc() -> FunctionDescription {
+        FunctionDescription::creator(Box::new(Self::try_create))
+            .features(FunctionFeatures::default().deterministic())
+    }
+}
+
+impl Function for SqrtFunction {
+    fn name(&self) -> &str {
+        &*self.display_name
+    }
+
+    fn num_arguments(&self) -> usize {
+        1
+    }
+
+    fn return_type(&self, args: &[DataType]) -> Result<DataType> {
+        if is_numeric(&args[0]) || args[0] == DataType::String || args[0] == DataType::Null {
+            Ok(DataType::Float64)
+        } else {
+            Err(ErrorCode::IllegalDataType(format!(
+                "Expected numeric types, but got {}",
+                args[0]
+            )))
+        }
+    }
+
+    fn nullable(&self, _input_schema: &DataSchema) -> Result<bool> {
+        Ok(true)
+    }
+
+    fn eval(&self, columns: &DataColumnsWithField, _input_rows: usize) -> Result<DataColumn> {
+        let opt_iter = columns[0]
+            .column()
+            .to_minimal_array()?
+            .cast_with_type(&DataType::Float64)?;
+        let opt_iter = opt_iter.f64()?.into_iter().map(|v| match v {
+            Some(&v) => {
+                if v >= 0_f64 {
+                    Some(v.sqrt())
+                } else {
+                    None
+                }
+            }
+            None => None,
+        });
+        let result = DFFloat64Array::new_from_opt_iter(opt_iter);
+        let column: DataColumn = result.into();
+        Ok(column.resize_constant(columns[0].column().len()))
+    }
+}
+
+impl fmt::Display for SqrtFunction {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "SQRT")
+    }
+}