chore!: bump dependencies

Cargo.toml

@@ -61,36 +61,40 @@ license = "Apache-2.0"
 
 [workspace.dependencies]
 aquamarine = "0.3"
-arrow = { version = "43.0" }
-arrow-array = "43.0"
-arrow-flight = "43.0"
-arrow-schema = { version = "43.0", features = ["serde"] }
+arrow = { version = "47.0" }
+arrow-array = "47.0"
+arrow-flight = "47.0"
+arrow-schema = { version = "47.0", features = ["serde"] }
 async-stream = "0.3"
 async-trait = "0.1"
 chrono = { version = "0.4", features = ["serde"] }
-datafusion = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "b6f3b28b6fe91924cc8dd3d83726b766f2a706ec" }
-datafusion-common = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "b6f3b28b6fe91924cc8dd3d83726b766f2a706ec" }
-datafusion-expr = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "b6f3b28b6fe91924cc8dd3d83726b766f2a706ec" }
-datafusion-optimizer = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "b6f3b28b6fe91924cc8dd3d83726b766f2a706ec" }
-datafusion-physical-expr = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "b6f3b28b6fe91924cc8dd3d83726b766f2a706ec" }
-datafusion-sql = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "b6f3b28b6fe91924cc8dd3d83726b766f2a706ec" }
-datafusion-substrait = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "b6f3b28b6fe91924cc8dd3d83726b766f2a706ec" }
+datafusion = { git = "https://github.com/apache/arrow-datafusion.git", rev = "26e43acac3a96cec8dd4c8365f22dfb1a84306e9" }
+datafusion-common = { git = "https://github.com/apache/arrow-datafusion.git", rev = "26e43acac3a96cec8dd4c8365f22dfb1a84306e9" }
+datafusion-expr = { git = "https://github.com/apache/arrow-datafusion.git", rev = "26e43acac3a96cec8dd4c8365f22dfb1a84306e9" }
+datafusion-optimizer = { git = "https://github.com/apache/arrow-datafusion.git", rev = "26e43acac3a96cec8dd4c8365f22dfb1a84306e9" }
+datafusion-physical-expr = { git = "https://github.com/apache/arrow-datafusion.git", rev = "26e43acac3a96cec8dd4c8365f22dfb1a84306e9" }
+datafusion-sql = { git = "https://github.com/apache/arrow-datafusion.git", rev = "26e43acac3a96cec8dd4c8365f22dfb1a84306e9" }
+datafusion-substrait = { git = "https://github.com/apache/arrow-datafusion.git", rev = "26e43acac3a96cec8dd4c8365f22dfb1a84306e9" }
 derive_builder = "0.12"
 etcd-client = "0.11"
 futures = "0.3"
 futures-util = "0.3"
-greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "1f1dd532a111e3834cc3019c5605e2993ffb9dc3" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "365922c3581ad954b3b3851af522b0ed0096970c" }
 humantime-serde = "1.1"
 itertools = "0.10"
 lazy_static = "1.4"
 meter-core = { git = "https://github.com/GreptimeTeam/greptime-meter.git", rev = "abbd357c1e193cd270ea65ee7652334a150b628f" }
 metrics = "0.20"
 moka = "0.12"
 once_cell = "1.18"
-opentelemetry-proto = { version = "0.2", features = ["gen-tonic", "metrics", "traces"] }
-parquet = "43.0"
+opentelemetry-proto = { git = "https://github.com/waynexia/opentelemetry-rust.git", rev = "33841b38dda79b15f2024952be5f32533325ca02", features = [
+    "gen-tonic",
+    "metrics",
+    "trace",
+] }
+parquet = "47.0"
 paste = "1.0"
-prost = "0.11"
+prost = "0.12"
 raft-engine = { git = "https://github.com/tikv/raft-engine.git", rev = "22dfb426cd994602b57725ef080287d3e53db479" }
 rand = "0.8"
 regex = "1.8"
@@ -103,15 +107,16 @@ serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 smallvec = "1"
 snafu = { version = "0.7", features = ["backtraces"] }
-sqlparser = { git = "https://github.com/GreptimeTeam/sqlparser-rs.git", rev = "6cf9d23d5b8fbecd65efc1d9afb7e80ad7a424da", features = [
+# on branch v0.38.x
+sqlparser = { git = "https://github.com/GreptimeTeam/sqlparser-rs.git", rev = "602d7878c9949e48512251c7f18695a50936e51c", features = [
     "visitor",
 ] }
 strum = { version = "0.25", features = ["derive"] }
 tempfile = "3"
 tokio = { version = "1.28", features = ["full"] }
 tokio-util = { version = "0.7", features = ["io-util", "compat"] }
 toml = "0.7"
-tonic = { version = "0.9", features = ["tls"] }
+tonic = { version = "0.10", features = ["tls"] }
 uuid = { version = "1", features = ["serde", "v4", "fast-rng"] }
 ## workspaces members
 api = { path = "src/api" }

src/api/src/error.rs

@@ -28,7 +28,12 @@ pub type Result<T> = std::result::Result<T, Error>;
 #[stack_trace_debug]
 pub enum Error {
     #[snafu(display("Unknown proto column datatype: {}", datatype))]
-    UnknownColumnDataType { datatype: i32, location: Location },
+    UnknownColumnDataType {
+        datatype: i32,
+        location: Location,
+        #[snafu(source)]
+        error: prost::DecodeError,
+    },
 
     #[snafu(display("Failed to create column datatype from {:?}", from))]
     IntoColumnDataType {

src/api/src/helper.rs

@@ -50,7 +50,7 @@ pub struct ColumnDataTypeWrapper(ColumnDataType);
 
 impl ColumnDataTypeWrapper {
     pub fn try_new(datatype: i32) -> Result<Self> {
-        let datatype = ColumnDataType::from_i32(datatype)
+        let datatype = ColumnDataType::try_from(datatype)
             .context(error::UnknownColumnDataTypeSnafu { datatype })?;
         Ok(Self(datatype))
     }
@@ -705,7 +705,7 @@ pub fn is_semantic_type_eq(type_value: i32, semantic_type: SemanticType) -> bool
 
 /// Returns true if the pb type value is valid.
 pub fn is_column_type_value_eq(type_value: i32, expect_type: &ConcreteDataType) -> bool {
-    let Some(column_type) = ColumnDataType::from_i32(type_value) else {
+    let Ok(column_type) = ColumnDataType::try_from(type_value) else {
         return false;
     };
 

src/client/Cargo.toml

@@ -41,11 +41,10 @@ tonic.workspace = true
 common-grpc-expr = { workspace = true }
 datanode = { workspace = true }
 derive-new = "0.5"
-prost.workspace = true
 substrait = { workspace = true }
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 
 [dev-dependencies.substrait_proto]
 package = "substrait"
-version = "0.7"
+version = "0.17"

src/common/datasource/Cargo.toml

@@ -25,7 +25,7 @@ derive_builder.workspace = true
 futures.workspace = true
 lazy_static.workspace = true
 object-store = { workspace = true }
-orc-rust = "0.2"
+orc-rust = { git = "https://github.com/WenyXu/orc-rs.git", rev = "5f6399f759ba30cb46610d06027b507949a117ca" }
 paste = "1.0"
 regex = "1.7"
 serde.workspace = true

src/common/function/src/scalars/math/rate.rs

@@ -16,7 +16,8 @@ use std::fmt;
 
 use common_query::error::{self, Result};
 use common_query::prelude::{Signature, Volatility};
-use datatypes::arrow::compute::kernels::{arithmetic, cast};
+use datafusion::arrow::compute::kernels::numeric;
+use datatypes::arrow::compute::kernels::cast;
 use datatypes::arrow::datatypes::DataType;
 use datatypes::prelude::*;
 use datatypes::vectors::{Helper, VectorRef};
@@ -51,19 +52,19 @@ impl Function for RateFunction {
         let val = &columns[0].to_arrow_array();
         let val_0 = val.slice(0, val.len() - 1);
         let val_1 = val.slice(1, val.len() - 1);
-        let dv = arithmetic::subtract_dyn(&val_1, &val_0).context(error::ArrowComputeSnafu)?;
+        let dv = numeric::sub(&val_1, &val_0).context(error::ArrowComputeSnafu)?;
         let ts = &columns[1].to_arrow_array();
         let ts_0 = ts.slice(0, ts.len() - 1);
         let ts_1 = ts.slice(1, ts.len() - 1);
-        let dt = arithmetic::subtract_dyn(&ts_1, &ts_0).context(error::ArrowComputeSnafu)?;
+        let dt = numeric::sub(&ts_1, &ts_0).context(error::ArrowComputeSnafu)?;
 
         let dv = cast::cast(&dv, &DataType::Float64).context(error::TypeCastSnafu {
             typ: DataType::Float64,
         })?;
         let dt = cast::cast(&dt, &DataType::Float64).context(error::TypeCastSnafu {
             typ: DataType::Float64,
         })?;
-        let rate = arithmetic::divide_dyn(&dv, &dt).context(error::ArrowComputeSnafu)?;
+        let rate = numeric::div(&dv, &dt).context(error::ArrowComputeSnafu)?;
         let v = Helper::try_into_vector(&rate).context(error::FromArrowArraySnafu)?;
 
         Ok(v)

src/common/function/src/scalars/timestamp/greatest.rs

@@ -18,9 +18,9 @@ use common_query::error::{
     self, ArrowComputeSnafu, InvalidFuncArgsSnafu, Result, UnsupportedInputDataTypeSnafu,
 };
 use common_query::prelude::{Signature, Volatility};
+use datafusion::arrow::compute::kernels::cmp::gt;
 use datatypes::arrow::array::AsArray;
 use datatypes::arrow::compute::cast;
-use datatypes::arrow::compute::kernels::comparison::gt_dyn;
 use datatypes::arrow::compute::kernels::zip;
 use datatypes::arrow::datatypes::{DataType as ArrowDataType, Date32Type};
 use datatypes::prelude::ConcreteDataType;
@@ -72,7 +72,7 @@ impl Function for GreatestFunction {
                 let column2 = cast(&columns[1].to_arrow_array(), &ArrowDataType::Date32)
                     .context(ArrowComputeSnafu)?;
                 let column2 = column2.as_primitive::<Date32Type>();
-                let boolean_array = gt_dyn(&column1, &column2).context(ArrowComputeSnafu)?;
+                let boolean_array = gt(&column1, &column2).context(ArrowComputeSnafu)?;
                 let result =
                     zip::zip(&boolean_array, &column1, &column2).context(ArrowComputeSnafu)?;
                 Ok(Helper::try_into_vector(&result).context(error::FromArrowArraySnafu)?)
@@ -82,7 +82,7 @@ impl Function for GreatestFunction {
                 let column1 = column1.as_primitive::<Date32Type>();
                 let column2 = columns[1].to_arrow_array();
                 let column2 = column2.as_primitive::<Date32Type>();
-                let boolean_array = gt_dyn(&column1, &column2).context(ArrowComputeSnafu)?;
+                let boolean_array = gt(&column1, &column2).context(ArrowComputeSnafu)?;
                 let result =
                     zip::zip(&boolean_array, &column1, &column2).context(ArrowComputeSnafu)?;
                 Ok(Helper::try_into_vector(&result).context(error::FromArrowArraySnafu)?)

src/common/grpc/src/flight.rs

@@ -241,7 +241,7 @@ mod test {
         .unwrap();
 
         let flight_data = batches_to_flight_data(
-            arrow_schema,
+            &arrow_schema,
             vec![
                 batch1.clone().into_df_record_batch(),
                 batch2.clone().into_df_record_batch(),

src/common/query/src/columnar_value.rs

@@ -33,7 +33,7 @@ impl ColumnarValue {
         match self {
             ColumnarValue::Vector(vector) => vector.data_type(),
             ColumnarValue::Scalar(scalar_value) => {
-                ConcreteDataType::from_arrow_type(&scalar_value.get_datatype())
+                ConcreteDataType::from_arrow_type(&scalar_value.data_type())
             }
         }
     }

src/common/substrait/Cargo.toml

@@ -27,7 +27,7 @@ table = { workspace = true }
 
 [dependencies.substrait_proto]
 package = "substrait"
-version = "0.12"
+version = "0.17"
 
 [dev-dependencies]
 datatypes = { workspace = true }

src/common/substrait/src/df_substrait.rs

@@ -47,9 +47,9 @@ impl SubstraitPlan for DFLogicalSubstraitConvertor {
         schema: &str,
     ) -> Result<Self::Plan, Self::Error> {
         let state_config = SessionConfig::new().with_default_catalog_and_schema(catalog, schema);
-        let state = SessionState::with_config_rt(state_config, Arc::new(RuntimeEnv::default()))
+        let state = SessionState::new_with_config_rt(state_config, Arc::new(RuntimeEnv::default()))
             .with_serializer_registry(Arc::new(ExtensionSerializer));
-        let mut context = SessionContext::with_state(state);
+        let mut context = SessionContext::new_with_state(state);
         context.register_catalog_list(catalog_list);
         let plan = Plan::decode(message).context(DecodeRelSnafu)?;
         let df_plan = from_substrait_plan(&mut context, &plan)
@@ -61,9 +61,9 @@ impl SubstraitPlan for DFLogicalSubstraitConvertor {
     fn encode(&self, plan: &Self::Plan) -> Result<Bytes, Self::Error> {
         let mut buf = BytesMut::new();
         let session_state =
-            SessionState::with_config_rt(SessionConfig::new(), Arc::new(RuntimeEnv::default()))
+            SessionState::new_with_config_rt(SessionConfig::new(), Arc::new(RuntimeEnv::default()))
                 .with_serializer_registry(Arc::new(ExtensionSerializer));
-        let context = SessionContext::with_state(session_state);
+        let context = SessionContext::new_with_state(session_state);
 
         let substrait_plan = to_substrait_plan(plan, &context).context(EncodeDfPlanSnafu)?;
         substrait_plan.encode(&mut buf).context(EncodeRelSnafu)?;

src/common/time/src/timestamp.rs

@@ -21,7 +21,7 @@ use std::time::Duration;
 
 use arrow::datatypes::TimeUnit as ArrowTimeUnit;
 use chrono::{
-    DateTime, LocalResult, NaiveDate, NaiveDateTime, NaiveTime, TimeZone as ChronoTimeZone, Utc,
+    DateTime, LocalResult, NaiveDate, NaiveDateTime, NaiveTime, TimeZone as ChronoTimeZone,
 };
 use serde::{Deserialize, Serialize};
 use snafu::{OptionExt, ResultExt};
@@ -63,7 +63,7 @@ impl Timestamp {
             TimeUnit::Second => now.timestamp(),
             TimeUnit::Millisecond => now.timestamp_millis(),
             TimeUnit::Microsecond => now.timestamp_micros(),
-            TimeUnit::Nanosecond => now.timestamp_nanos(),
+            TimeUnit::Nanosecond => now.timestamp_nanos_opt().unwrap_or_default(),
         };
         Timestamp { value, unit }
     }
@@ -287,6 +287,7 @@ impl FromStr for Timestamp {
     /// - `2022-09-20 14:16:43.012345Z` (Zulu timezone, without T)
     /// - `2022-09-20 14:16:43` (local timezone, without T)
     /// - `2022-09-20 14:16:43.012345` (local timezone, without T)
+    #[allow(deprecated)]
     fn from_str(s: &str) -> Result<Self, Self::Err> {
         // RFC3339 timestamp (with a T)
         let s = s.trim();
@@ -298,7 +299,7 @@ impl FromStr for Timestamp {
             return Timestamp::from_chrono_datetime(ts.naive_utc())
                 .context(ParseTimestampSnafu { raw: s });
         }
-        if let Ok(ts) = Utc.datetime_from_str(s, "%Y-%m-%d %H:%M:%S%.fZ") {
+        if let Ok(ts) = chrono::Utc.datetime_from_str(s, "%Y-%m-%d %H:%M:%S%.fZ") {
             return Timestamp::from_chrono_datetime(ts.naive_utc())
                 .context(ParseTimestampSnafu { raw: s });
         }

src/datatypes/src/value.rs

@@ -787,10 +787,11 @@ impl TryFrom<ScalarValue> for Value {
                 .map(|x| Value::Duration(Duration::new(x, TimeUnit::Nanosecond)))
                 .unwrap_or(Value::Null),
             ScalarValue::Decimal128(_, _, _)
+            | ScalarValue::Decimal256(_, _, _)
             | ScalarValue::Struct(_, _)
             | ScalarValue::Dictionary(_, _) => {
                 return error::UnsupportedArrowTypeSnafu {
-                    arrow_type: v.get_datatype(),
+                    arrow_type: v.data_type(),
                 }
                 .fail()
             }

src/datatypes/src/vectors/constant.rs

@@ -100,7 +100,8 @@ impl ConstantVector {
         let arr = indices.to_arrow_array();
         let indices_arr = arr.as_any().downcast_ref::<UInt32Array>().unwrap();
         if !arrow::compute::min_boolean(
-            &arrow::compute::lt_scalar(indices_arr, len as u32).unwrap(),
+            &arrow::compute::kernels::cmp::lt(indices_arr, &UInt32Array::new_scalar(len as u32))
+                .unwrap(),
         )
         .unwrap()
         {

src/datatypes/src/vectors/helper.rs

@@ -219,6 +219,7 @@ impl Helper {
                 ConstantVector::new(Arc::new(IntervalMonthDayNanoVector::from(vec![v])), length)
             }
             ScalarValue::Decimal128(_, _, _)
+            | ScalarValue::Decimal256(_, _, _)
             | ScalarValue::DurationSecond(_)
             | ScalarValue::DurationMillisecond(_)
             | ScalarValue::DurationMicrosecond(_)
@@ -345,15 +346,17 @@ impl Helper {
     pub fn like_utf8(names: Vec<String>, s: &str) -> Result<VectorRef> {
         let array = StringArray::from(names);
 
-        let filter = comparison::like_utf8_scalar(&array, s).context(error::ArrowComputeSnafu)?;
+        let s = StringArray::new_scalar(s);
+        let filter = comparison::like(&array, &s).context(error::ArrowComputeSnafu)?;
 
         let result = compute::filter(&array, &filter).context(error::ArrowComputeSnafu)?;
         Helper::try_into_vector(result)
     }
 
     pub fn like_utf8_filter(names: Vec<String>, s: &str) -> Result<(VectorRef, BooleanVector)> {
         let array = StringArray::from(names);
-        let filter = comparison::like_utf8_scalar(&array, s).context(error::ArrowComputeSnafu)?;
+        let s = StringArray::new_scalar(s);
+        let filter = comparison::like(&array, &s).context(error::ArrowComputeSnafu)?;
         let result = compute::filter(&array, &filter).context(error::ArrowComputeSnafu)?;
         let vector = Helper::try_into_vector(result)?;
 
@@ -481,7 +484,8 @@ mod tests {
 
         fn assert_filter(array: Vec<String>, s: &str, expected_filter: &BooleanVector) {
             let array = StringArray::from(array);
-            let actual_filter = comparison::like_utf8_scalar(&array, s).unwrap();
+            let s = StringArray::new_scalar(s);
+            let actual_filter = comparison::like(&array, &s).unwrap();
             assert_eq!(BooleanVector::from(actual_filter), *expected_filter);
         }