Merge branch 'datafuselabs:main' into feat/15295_map_insert

Cargo.toml

@@ -294,7 +294,6 @@ debug-assertions = true
 overflow-checks = true
 rpath = false
 
-# If there are dependencies that need patching, they can be listed below.
 [patch.crates-io]
 arrow-format = { git = "https://github.com/Xuanwo/arrow-format", rev = "be633a0" }
 icelake = { git = "https://github.com/icelake-io/icelake", rev = "be8b2c2" }
@@ -303,3 +302,4 @@ async-backtrace = { git = "https://github.com/zhang2014/async-backtrace.git", re
 z3 = { git = "https://github.com/prove-rs/z3.rs", rev = "247d308" }
 z3-sys = { git = "https://github.com/prove-rs/z3.rs", rev = "247d308" }
 # proj = { git = "https://github.com/ariesdevil/proj", rev = "51e1c60" }
+derive-visitor = { git = 'https://github.com/andylokandy/derive-visitor.git', rev = "c07c6b6" }

src/common/cloud_control/proto/task.proto

@@ -12,6 +12,7 @@ message ScheduleOptions {
   optional string cron = 2; // CRON = '0 2 * * *' means Every night at 2 AM. UTC time zone.
   optional string time_zone = 3; // "UTC..."
   ScheduleType schedule_type = 4;
+  optional uint64 milliseconds_interval = 5; // milliseconds level interval
 }
 
 message WarehouseOptions {

src/common/cloud_control/src/task_utils.rs

@@ -93,10 +93,19 @@ pub fn format_schedule_options(s: &ScheduleOptions) -> Result<String> {
         }
     };
     return match schedule_type {
-        ScheduleType::IntervalType => Ok(format!(
-            "INTERVAL {} SECOND",
-            s.interval.unwrap_or_default(),
-        )),
+        ScheduleType::IntervalType => {
+            if s.milliseconds_interval.is_some() {
+                return Ok(format!(
+                    "INTERVAL {} SECOND {} MILLISECOND",
+                    s.interval.unwrap_or_default(),
+                    s.milliseconds_interval.unwrap_or_default(),
+                ));
+            }
+            Ok(format!(
+                "INTERVAL {} SECOND",
+                s.interval.unwrap_or_default(),
+            ))
+        }
         ScheduleType::CronType => {
             if s.cron.is_none() {
                 return Err(ErrorCode::IllegalCloudControlMessageFormat(

src/common/exception/Cargo.toml

@@ -12,11 +12,7 @@ test = true
 
 [dependencies] # In alphabetical order
 databend-common-arrow = { path = "../arrow" }
-
-# GitHub dependencies
-# TODO: 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" }
+databend-common-ast = { path = "../../query/ast" }
 
 anyhow = { workspace = true }
 arrow-schema = { workspace = true }

src/common/exception/src/exception.rs

@@ -20,11 +20,11 @@ use std::fmt::Formatter;
 use std::sync::Arc;
 
 use backtrace::Backtrace;
+use databend_common_ast::span::pretty_print_error;
+use databend_common_ast::Span;
 use thiserror::Error;
 
 use crate::exception_backtrace::capture;
-use crate::span::pretty_print_error;
-use crate::Span;
 
 #[derive(Clone)]
 pub enum ErrorCodeBacktrace {

src/common/exception/src/exception_into.rs

@@ -18,12 +18,12 @@ use std::fmt::Display;
 use std::fmt::Formatter;
 use std::sync::Arc;
 
+use databend_common_ast::Span;
 use geozero::error::GeozeroError;
 
 use crate::exception::ErrorCodeBacktrace;
 use crate::exception_backtrace::capture;
 use crate::ErrorCode;
-use crate::Span;
 
 #[derive(thiserror::Error)]
 enum OtherErrors {
@@ -231,6 +231,12 @@ impl From<std::string::FromUtf8Error> for ErrorCode {
     }
 }
 
+impl From<databend_common_ast::ParseError> for ErrorCode {
+    fn from(error: databend_common_ast::ParseError) -> Self {
+        ErrorCode::SyntaxException(error.1).set_span(error.0)
+    }
+}
+
 impl From<GeozeroError> for ErrorCode {
     fn from(value: GeozeroError) -> Self {
         ErrorCode::GeometryError(value.to_string())

src/common/exception/src/lib.rs

@@ -18,17 +18,11 @@ mod exception_backtrace;
 mod exception_code;
 mod exception_flight;
 mod exception_into;
-mod span;
 mod with_context;
 
 pub use exception::ErrorCode;
 pub use exception::Result;
 pub use exception::ToErrorCode;
 pub use exception_into::SerializedError;
-pub use span::merge_span;
-pub use span::offset_span;
-pub use span::pretty_print_error;
-pub use span::Range;
-pub use span::Span;
 pub use with_context::ErrorWithContext;
 pub use with_context::WithContext;

src/meta/api/src/schema_api_impl.rs

@@ -190,13 +190,17 @@ use databend_common_meta_kvapi::kvapi;
 use databend_common_meta_kvapi::kvapi::DirName;
 use databend_common_meta_kvapi::kvapi::Key;
 use databend_common_meta_kvapi::kvapi::UpsertKVReq;
+use databend_common_meta_types::anyerror::AnyError;
 use databend_common_meta_types::protobuf as pb;
 use databend_common_meta_types::txn_op::Request;
 use databend_common_meta_types::txn_op_response::Response;
 use databend_common_meta_types::ConditionResult;
 use databend_common_meta_types::InvalidReply;
 use databend_common_meta_types::MatchSeq;
 use databend_common_meta_types::MatchSeqExt;
+use databend_common_meta_types::MetaAPIError;
+use databend_common_meta_types::MetaDataError;
+use databend_common_meta_types::MetaDataReadError;
 use databend_common_meta_types::MetaError;
 use databend_common_meta_types::MetaId;
 use databend_common_meta_types::MetaNetworkError;
@@ -2282,7 +2286,7 @@ impl<KV: kvapi::KVApi<Error = MetaError> + ?Sized> SchemaApi for KV {
 
         // Batch get all table-name by id
         let seq_names = self.mget_kv(&id_name_kv_keys).await?;
-        let mut table_names = Vec::with_capacity(id_name_kv_keys.len());
+        let mut table_names = Vec::with_capacity(table_ids.len());
 
         // None means table_name not found, maybe immutable table id. Ignore it
         for seq_name in seq_names.into_iter().flatten() {
@@ -2297,6 +2301,15 @@ impl<KV: kvapi::KVApi<Error = MetaError> + ?Sized> SchemaApi for KV {
         }
 
         let seq_metas = self.mget_kv(&meta_kv_keys).await?;
+        if seq_metas.len() != table_names.len() {
+            return Err(KVAppError::MetaError(MetaError::APIError(
+                MetaAPIError::DataError(MetaDataError::ReadError(MetaDataReadError::new(
+                    "mget_table_names_by_ids",
+                    "",
+                    &AnyError::error("The system is experiencing high load, please retry later"),
+                ))),
+            )));
+        }
         for (i, seq_meta_opt) in seq_metas.iter().enumerate() {
             if let Some(seq_meta) = seq_meta_opt {
                 let table_meta: TableMeta = deserialize_struct(&seq_meta.data)?;
@@ -2346,7 +2359,8 @@ impl<KV: kvapi::KVApi<Error = MetaError> + ?Sized> SchemaApi for KV {
 
         // Batch get all table-name by id
         let seq_names = self.mget_kv(&kv_keys).await?;
-        let mut db_names = Vec::with_capacity(kv_keys.len());
+        // If multi drop/create db the capacity may not same
+        let mut db_names = Vec::with_capacity(db_ids.len());
 
         // None means db_name not found, maybe immutable database id. Ignore it
         for seq_name in seq_names.into_iter().flatten() {
@@ -2361,6 +2375,15 @@ impl<KV: kvapi::KVApi<Error = MetaError> + ?Sized> SchemaApi for KV {
         }
 
         let seq_metas = self.mget_kv(&meta_kv_keys).await?;
+        if seq_metas.len() != db_names.len() {
+            return Err(KVAppError::MetaError(MetaError::APIError(
+                MetaAPIError::DataError(MetaDataError::ReadError(MetaDataReadError::new(
+                    "mget_table_names_by_ids",
+                    "",
+                    &AnyError::error("The system is experiencing high load, please retry later"),
+                ))),
+            )));
+        }
         for (i, seq_meta_opt) in seq_metas.iter().enumerate() {
             if let Some(seq_meta) = seq_meta_opt {
                 let db_meta: DatabaseMeta = deserialize_struct(&seq_meta.data)?;

src/query/ast/Cargo.toml

@@ -15,10 +15,6 @@ doctest = false
 ignored = ["geos"]
 
 [dependencies] # In alphabetical order
-# Workspace dependencies
-databend-common-exception = { path = "../../common/exception" }
-databend-common-io = { path = "../../common/io" }
-
 # Crates.io dependencies
 derive-visitor = { workspace = true }
 enum-as-inner = "0.5.1"
@@ -34,13 +30,20 @@ nom-rule = "0.3.0"
 ordered-float = { workspace = true }
 pratt = "0.4.0"
 pretty = "0.11.3"
+serde = { workspace = true }
 serde_json = { workspace = true }
 strsim = "0.10"
 strum = "0.24"
 strum_macros = "0.24"
 unindent = "0.2.3"
 url = "2.3.1"
 
+# TODO: Use the version from crates.io once
+# https://github.com/brendanzab/codespan/pull/331 is released.
+[dependencies.codespan-reporting]
+git = "https://github.com/brendanzab/codespan"
+rev = "c84116f5"
+
 [dev-dependencies]
 criterion = { workspace = true }
 goldenfile = "1.4"

src/query/ast/src/ast/common.rs

@@ -14,12 +14,14 @@
 
 use std::fmt::Display;
 use std::fmt::Formatter;
+use std::fmt::Write as _;
 
-use databend_common_exception::Span;
 use derive_visitor::Drive;
 use derive_visitor::DriveMut;
+use ethnum::i256;
 
 use crate::ast::quote::QuotedIdent;
+use crate::Span;
 
 // Identifier of table name or column name.
 #[derive(Debug, Clone, PartialEq, Eq, Drive, DriveMut)]
@@ -269,3 +271,33 @@ pub(crate) fn write_space_separated_string_map(
     }
     Ok(())
 }
+
+pub fn display_decimal_256(num: i256, scale: u8) -> String {
+    let mut buf = String::new();
+    if scale == 0 {
+        write!(buf, "{}", num).unwrap();
+    } else {
+        let pow_scale = i256::from(10).pow(scale as u32);
+        // -1/10 = 0
+        if num >= 0 {
+            write!(
+                buf,
+                "{}.{:0>width$}",
+                num / pow_scale,
+                (num % pow_scale).abs(),
+                width = scale as usize
+            )
+            .unwrap();
+        } else {
+            write!(
+                buf,
+                "-{}.{:0>width$}",
+                -num / pow_scale,
+                (num % pow_scale).abs(),
+                width = scale as usize
+            )
+            .unwrap();
+        }
+    }
+    buf
+}

src/query/ast/src/ast/expr.rs

@@ -15,12 +15,6 @@
 use std::fmt::Display;
 use std::fmt::Formatter;
 
-use databend_common_exception::merge_span;
-use databend_common_exception::ErrorCode;
-use databend_common_exception::Result;
-use databend_common_exception::Span;
-use databend_common_io::display_decimal_256;
-use databend_common_io::escape_string_with_quote;
 use derive_visitor::Drive;
 use derive_visitor::DriveMut;
 use enum_as_inner::EnumAsInner;
@@ -30,10 +24,16 @@ use pratt::Precedence;
 
 use super::ColumnRef;
 use super::OrderByExpr;
+use crate::ast::display_decimal_256;
+use crate::ast::quote::QuotedString;
 use crate::ast::write_comma_separated_list;
 use crate::ast::Identifier;
 use crate::ast::Query;
 use crate::parser::expr::ExprElement;
+use crate::span::merge_span;
+use crate::ParseError;
+use crate::Result;
+use crate::Span;
 
 #[derive(Debug, Clone, PartialEq, Drive, DriveMut)]
 pub enum Expr {
@@ -819,7 +819,7 @@ impl Display for Literal {
                 write!(f, "{val}")
             }
             Literal::String(val) => {
-                write!(f, "\'{}\'", escape_string_with_quote(val, Some('\'')))
+                write!(f, "{}", QuotedString(val, '\''))
             }
             Literal::Boolean(val) => {
                 if *val {
@@ -1265,9 +1265,10 @@ impl BinaryOperator {
             BinaryOperator::Lte => Ok(BinaryOperator::Gt),
             BinaryOperator::Eq => Ok(BinaryOperator::NotEq),
             BinaryOperator::NotEq => Ok(BinaryOperator::Eq),
-            _ => Err(ErrorCode::Unimplemented(format!(
-                "Converting {self} to its contrary is not currently supported"
-            ))),
+            _ => Err(ParseError(
+                None,
+                format!("Converting {self} to its contrary is not currently supported"),
+            )),
         }
     }