Merge branch 'main' into prune-pipeline

Author: dqhl76

Cargo.lock

@@ -212,9 +212,9 @@ arrow-ipc = { version = "53" }
 arrow-ord = { version = "53" }
 arrow-schema = { version = "53", features = ["serde"] }
 arrow-select = { version = "53" }
-arrow-udf-js = { git = "https://github.com/arrow-udf/arrow-udf", rev = "80b09d6" }
-arrow-udf-python = { git = "https://github.com/arrow-udf/arrow-udf", rev = "80b09d6" }
-arrow-udf-wasm = { git = "https://github.com/arrow-udf/arrow-udf", rev = "80b09d6" }
+arrow-udf-js = { version = "0.5.0" }
+arrow-udf-python = { version = "0.4.0" }
+arrow-udf-wasm = { version = "0.4.1" }
 async-backtrace = "0.2"
 async-channel = "1.7.1"
 async-compression = { git = "https://github.com/datafuse-extras/async-compression", rev = "dc81082", features = [
@@ -272,7 +272,7 @@ dyn-clone = "1.0.9"
 educe = "0.4" # FIXME: failed to upgrade to educe 0.6
 either = "1.9"
 enquote = "1.1.0"
-enum-as-inner = "0.5" # FIXME: failed to upgrade to enum-as-inner 0.6
+enum-as-inner = "0.6"
 enum_dispatch = "0.3.13"
 enumflags2 = { version = "0.7.7", features = ["serde"] }
 ethnum = { version = "1.5.0" }
@@ -485,6 +485,7 @@ tower = { version = "0.5.1", features = ["util"] }
 tower-service = "0.3.3"
 twox-hash = "1.6.3"
 typetag = "0.2.3"
+unicase = "2.8.0"
 unicode-segmentation = "1.10.1"
 unindent = "0.2"
 url = "2.3.1"
@@ -609,6 +610,9 @@ overflow-checks = true
 rpath = false
 
 [patch.crates-io]
+arrow-udf-js = { git = "https://github.com/arrow-udf/arrow-udf", rev = "ade868f" }
+arrow-udf-python = { git = "https://github.com/arrow-udf/arrow-udf", rev = "ade868f" }
+arrow-udf-wasm = { git = "https://github.com/arrow-udf/arrow-udf", rev = "ade868f" }
 async-backtrace = { git = "https://github.com/datafuse-extras/async-backtrace.git", rev = "dea4553" }
 async-recursion = { git = "https://github.com/datafuse-extras/async-recursion.git", rev = "a353334" }
 backtrace = { git = "https://github.com/rust-lang/backtrace-rs.git", rev = "72265be", features = [

Cargo.toml

@@ -155,3 +155,6 @@ data_cache_storage = "none"
 path = "./.databend/_cache"
 # max bytes of cached data 20G
 max_bytes = 21474836480
+
+[spill]
+spill_local_disk_path = "./.databend/temp/_query_spill"

scripts/ci/deploy/config/databend-query-node-1.toml

@@ -39,6 +39,11 @@ pub static BUCKET_MILLISECONDS: [f64; 15] = [
     300000.0, 600000.0, 1800000.0,
 ];
 
+pub static BUCKET_ROWS: [f64; 14] = [
+    1.0, 10.0, 100.0, 500.0, 1000.0, 5000.0, 10000.0, 50000.0, 100000.0, 500000.0, 1000000.0,
+    5000000.0, 10000000.0, 50000000.0,
+];
+
 /// Histogram is a port of prometheus-client's Histogram. The only difference is that
 /// we can reset the histogram.
 #[derive(Debug)]

src/common/base/src/runtime/metrics/histogram.rs

@@ -34,6 +34,7 @@ pub use registry::register_gauge_family;
 pub use registry::register_histogram;
 pub use registry::register_histogram_family;
 pub use registry::register_histogram_family_in_milliseconds;
+pub use registry::register_histogram_family_in_rows;
 pub use registry::register_histogram_family_in_seconds;
 pub use registry::register_histogram_in_milliseconds;
 pub use registry::register_histogram_in_seconds;

src/common/base/src/runtime/metrics/mod.rs

@@ -38,6 +38,7 @@ use crate::runtime::metrics::family_metrics::FamilyHistogram as InnerFamilyHisto
 use crate::runtime::metrics::gauge::Gauge;
 use crate::runtime::metrics::histogram::Histogram;
 use crate::runtime::metrics::histogram::BUCKET_MILLISECONDS;
+use crate::runtime::metrics::histogram::BUCKET_ROWS;
 use crate::runtime::metrics::histogram::BUCKET_SECONDS;
 use crate::runtime::metrics::process_collector::ProcessCollector;
 use crate::runtime::metrics::sample::MetricSample;
@@ -309,6 +310,11 @@ where T: FamilyLabels {
     register_histogram_family(name, BUCKET_MILLISECONDS.iter().copied())
 }
 
+pub fn register_histogram_family_in_rows<T>(name: &str) -> FamilyHistogram<T>
+where T: FamilyLabels {
+    register_histogram_family(name, BUCKET_ROWS.iter().copied())
+}
+
 pub type FamilyGauge<T> = Family<T, InnerFamilyGauge<T>>;
 pub type FamilyCounter<T> = Family<T, InnerFamilyCounter<T>>;
 pub type FamilyHistogram<T> = Family<T, InnerFamilyHistogram<T>>;

src/common/base/src/runtime/metrics/registry.rs

@@ -212,6 +212,10 @@ build_exceptions! {
     // Geometry errors.
     GeometryError(1801),
     InvalidGeometryFormat(1802),
+
+    // UDF errors.
+    UDFRuntimeError(1810),
+
     // Tantivy errors.
     TantivyError(1901),
     TantivyOpenReadError(1902),

src/common/exception/src/exception_code.rs

@@ -16,7 +16,8 @@ use std::sync::LazyLock;
 use std::time::Duration;
 
 use databend_common_base::runtime::metrics::register_counter_family;
-use databend_common_base::runtime::metrics::register_histogram_family_in_seconds;
+use databend_common_base::runtime::metrics::register_histogram_family_in_milliseconds;
+use databend_common_base::runtime::metrics::register_histogram_family_in_rows;
 use databend_common_base::runtime::metrics::FamilyCounter;
 use databend_common_base::runtime::metrics::FamilyHistogram;
 
@@ -28,12 +29,13 @@ const METRIC_RETRY: &str = "external_retry";
 const METRIC_ERROR: &str = "external_error";
 const METRIC_RUNNING_REQUESTS: &str = "external_running_requests";
 const METRIC_REQUESTS: &str = "external_requests";
+const METRIC_EXTERNAL_BLOCK_ROWS: &str = "external_block_rows";
 
 static REQUEST_EXTERNAL_DURATION: LazyLock<FamilyHistogram<VecLabels>> =
-    LazyLock::new(|| register_histogram_family_in_seconds(METRIC_REQUEST_EXTERNAL_DURATION));
+    LazyLock::new(|| register_histogram_family_in_milliseconds(METRIC_REQUEST_EXTERNAL_DURATION));
 
 static CONNECT_EXTERNAL_DURATION: LazyLock<FamilyHistogram<VecLabels>> =
-    LazyLock::new(|| register_histogram_family_in_seconds(METRIC_CONNECT_EXTERNAL_DURATION));
+    LazyLock::new(|| register_histogram_family_in_milliseconds(METRIC_CONNECT_EXTERNAL_DURATION));
 
 static RETRY_EXTERNAL: LazyLock<FamilyCounter<VecLabels>> =
     LazyLock::new(|| register_counter_family(METRIC_RETRY));
@@ -47,6 +49,9 @@ static RUNNING_REQUESTS_EXTERNAL: LazyLock<FamilyCounter<VecLabels>> =
 static REQUESTS_EXTERNAL_EXTERNAL: LazyLock<FamilyCounter<VecLabels>> =
     LazyLock::new(|| register_counter_family(METRIC_REQUESTS));
 
+static EXTERNAL_BLOCK_ROWS: LazyLock<FamilyHistogram<VecLabels>> =
+    LazyLock::new(|| register_histogram_family_in_rows(METRIC_EXTERNAL_BLOCK_ROWS));
+
 const LABEL_FUNCTION_NAME: &str = "function_name";
 const LABEL_ERROR_KIND: &str = "error_kind";
 
@@ -64,6 +69,13 @@ pub fn record_request_external_duration(function_name: impl Into<String>, durati
         .observe(duration.as_millis_f64());
 }
 
+pub fn record_request_external_block_rows(function_name: impl Into<String>, rows: usize) {
+    let labels = &vec![(LABEL_FUNCTION_NAME, function_name.into())];
+    EXTERNAL_BLOCK_ROWS
+        .get_or_create(labels)
+        .observe(rows as f64);
+}
+
 pub fn record_retry_external(function_name: impl Into<String>, error_kind: impl Into<String>) {
     let labels = &vec![
         (LABEL_FUNCTION_NAME, function_name.into()),

src/common/metrics/src/metrics/external_server.rs

@@ -95,6 +95,7 @@ pub use user_auth::AuthType;
 pub use user_auth::PasswordHashMethod;
 pub use user_defined_file_format::UserDefinedFileFormat;
 pub use user_defined_function::LambdaUDF;
+pub use user_defined_function::UDAFScript;
 pub use user_defined_function::UDFDefinition;
 pub use user_defined_function::UDFScript;
 pub use user_defined_function::UDFServer;

src/meta/app/src/principal/mod.rs

@@ -18,6 +18,7 @@ use std::fmt::Formatter;
 use chrono::DateTime;
 use chrono::Utc;
 use databend_common_expression::types::DataType;
+use databend_common_expression::DataField;
 
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub struct LambdaUDF {
@@ -44,11 +45,24 @@ pub struct UDFScript {
     pub runtime_version: String,
 }
 
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct UDAFScript {
+    pub code: String,
+    pub language: String,
+    // aggregate function input types
+    pub arg_types: Vec<DataType>,
+    // aggregate function state fields
+    pub state_fields: Vec<DataField>,
+    pub return_type: DataType,
+    pub runtime_version: String,
+}
+
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub enum UDFDefinition {
     LambdaUDF(LambdaUDF),
     UDFServer(UDFServer),
     UDFScript(UDFScript),
+    UDAFScript(UDAFScript),
 }
 
 #[derive(Clone, Debug, Eq, PartialEq)]
@@ -160,7 +174,6 @@ impl Display for UDFDefinition {
                     ") RETURNS {return_type} LANGUAGE {language} HANDLER = {handler} ADDRESS = {address}"
                 )?;
             }
-
             UDFDefinition::UDFScript(UDFScript {
                 code,
                 arg_types,
@@ -180,6 +193,29 @@ impl Display for UDFDefinition {
                     ") RETURNS {return_type} LANGUAGE {language} RUNTIME_VERSION = {runtime_version} HANDLER = {handler} AS $${code}$$"
                 )?;
             }
+            UDFDefinition::UDAFScript(UDAFScript {
+                code,
+                arg_types,
+                state_fields,
+                return_type,
+                language,
+                runtime_version,
+            }) => {
+                for (i, item) in arg_types.iter().enumerate() {
+                    if i > 0 {
+                        write!(f, ", ")?;
+                    }
+                    write!(f, "{item}")?;
+                }
+                write!(f, ") STATE {{ ")?;
+                for (i, item) in state_fields.iter().enumerate() {
+                    if i > 0 {
+                        write!(f, ", ")?;
+                    }
+                    write!(f, "{} {}", item.name(), item.data_type())?;
+                }
+                write!(f, " }} RETURNS {return_type} LANGUAGE {language} RUNTIME_VERSION = {runtime_version} AS $${code}$$")?;
+            }
         }
         Ok(())
     }