chore(deps): Bump arrow-*, tonic, prost

Cargo.toml

@@ -105,16 +105,16 @@ hashbrown = { version = "0.14.0", features = [
   "nightly",
 ] }
 criterion = { version = "0.5", features = ["async_futures"] }
-tonic = { package = "madsim-tonic", version = "0.3.1" }
-tonic-build = { package = "madsim-tonic-build", version = "0.3.1" }
+tonic = { package = "madsim-tonic", version = "0.4.0" }
+tonic-build = { package = "madsim-tonic-build", version = "0.4.0" }
 icelake = { git = "https://github.com/icelake-io/icelake", rev = "72a65aed6ed7b3d529b311031c2c8d99650990e2" }
-arrow-array = "46"
-arrow-cast = "46"
-arrow-schema = "46"
-arrow-buffer = "46"
-arrow-flight = "46"
-arrow-select = "46"
-arrow-ord = "46"
+arrow-array = "47"
+arrow-cast = "47"
+arrow-schema = "47"
+arrow-buffer = "47"
+arrow-flight = "47"
+arrow-select = "47"
+arrow-ord = "47"
 tikv-jemalloc-ctl = { git = "https://github.com/risingwavelabs/jemallocator.git", rev = "64a2d9" }
 tikv-jemallocator = { git = "https://github.com/risingwavelabs/jemallocator.git", features = [
   "profiling",

src/common/Cargo.toml

@@ -65,7 +65,7 @@ postgres-types = { version = "0.2.6", features = [
 ] }
 prehash = "1"
 prometheus = { version = "0.13" }
-prost = "0.11"
+prost = "0.12"
 rand = "0.8"
 regex = "1"
 reqwest = { version = "0.11", features = ["json"] }

src/common/src/array/stream_chunk.rs

@@ -58,13 +58,13 @@ impl Op {
     }
 
     pub fn from_protobuf(prost: &i32) -> ArrayResult<Op> {
-        let op = match PbOp::from_i32(*prost) {
-            Some(PbOp::Insert) => Op::Insert,
-            Some(PbOp::Delete) => Op::Delete,
-            Some(PbOp::UpdateInsert) => Op::UpdateInsert,
-            Some(PbOp::UpdateDelete) => Op::UpdateDelete,
-            Some(PbOp::Unspecified) => unreachable!(),
-            None => bail!("No such op type"),
+        let op = match PbOp::try_from(*prost) {
+            Ok(PbOp::Insert) => Op::Insert,
+            Ok(PbOp::Delete) => Op::Delete,
+            Ok(PbOp::UpdateInsert) => Op::UpdateInsert,
+            Ok(PbOp::UpdateDelete) => Op::UpdateDelete,
+            Ok(PbOp::Unspecified) => unreachable!(),
+            Err(_) => bail!("No such op type"),
         };
         Ok(op)
     }

src/connector/Cargo.toml

@@ -53,10 +53,16 @@ futures-async-stream = { workspace = true }
 glob = "0.3"
 google-cloud-pubsub = "0.20"
 http = "0.2"
-hyper = { version = "0.14", features = ["client", "tcp", "http1", "http2", "stream"] }
+hyper = { version = "0.14", features = [
+    "client",
+    "tcp",
+    "http1",
+    "http2",
+    "stream",
+] }
 hyper-tls = "0.5"
 icelake = { workspace = true }
-indexmap ={ version = "1.9.3", features = ["serde"] }
+indexmap = { version = "1.9.3", features = ["serde"] }
 itertools = "0.11"
 jni = { version = "0.21.1", features = ["invocation"] }
 jsonschema-transpiler = { git = "https://github.com/mozilla/jsonschema-transpiler", rev = "c1a89d720d118843d8bcca51084deb0ed223e4b4" }
@@ -75,8 +81,8 @@ opendal = "0.39"
 parking_lot = "0.12"
 paste = "1"
 prometheus = { version = "0.13", features = ["process"] }
-prost = { version = "0.11", features = ["no-recursion-limit"] }
-prost-reflect = "0.11"
+prost = { version = "0.12", features = ["no-recursion-limit"] }
+prost-reflect = "0.12"
 protobuf-native = "0.2.1"
 pulsar = { version = "6.0", default-features = false, features = [
     "tokio-runtime",
@@ -121,6 +127,7 @@ tokio-retry = "0.3"
 tokio-stream = "0.1"
 tokio-util = { version = "0.7", features = ["codec", "io"] }
 tonic = { workspace = true }
+tonic_0_9 = { package = "tonic", version = "0.9" }
 tracing = "0.1"
 url = "2"
 urlencoding = "2"
@@ -130,12 +137,12 @@ workspace-hack = { path = "../workspace-hack" }
 
 [dev-dependencies]
 criterion = { workspace = true, features = ["async_tokio", "async"] }
-prost-types = "0.11"
+prost-types = "0.12"
 rand = "0.8"
 tempfile = "3"
 
 [build-dependencies]
-prost-build = "0.11"
+prost-build = "0.12"
 
 [[bench]]
 name = "parser"

src/connector/src/parser/mod.rs

@@ -856,7 +856,7 @@ impl SpecificParserConfig {
                         Some(info.proto_message_name.clone())
                     },
                     key_record_name: info.key_message_name.clone(),
-                    name_strategy: PbSchemaRegistryNameStrategy::from_i32(info.name_strategy)
+                    name_strategy: PbSchemaRegistryNameStrategy::try_from(info.name_strategy)
                         .unwrap(),
                     use_schema_registry: info.use_schema_registry,
                     row_schema_location: info.row_schema_location.clone(),
@@ -887,7 +887,7 @@ impl SpecificParserConfig {
                     message_name: info.proto_message_name.clone(),
                     use_schema_registry: info.use_schema_registry,
                     row_schema_location: info.row_schema_location.clone(),
-                    name_strategy: PbSchemaRegistryNameStrategy::from_i32(info.name_strategy)
+                    name_strategy: PbSchemaRegistryNameStrategy::try_from(info.name_strategy)
                         .unwrap(),
                     key_message_name: info.key_message_name.clone(),
                     ..Default::default()
@@ -912,7 +912,7 @@ impl SpecificParserConfig {
                     } else {
                         Some(info.proto_message_name.clone())
                     },
-                    name_strategy: PbSchemaRegistryNameStrategy::from_i32(info.name_strategy)
+                    name_strategy: PbSchemaRegistryNameStrategy::try_from(info.name_strategy)
                         .unwrap(),
                     key_record_name: info.key_message_name.clone(),
                     row_schema_location: info.row_schema_location.clone(),

src/connector/src/sink/mod.rs

@@ -162,7 +162,7 @@ impl SinkParam {
                 .map(|i| *i as usize)
                 .collect(),
             sink_type: SinkType::from_proto(
-                PbSinkType::from_i32(pb_param.sink_type).expect("should be able to convert"),
+                PbSinkType::try_from(pb_param.sink_type).expect("should be able to convert"),
             ),
             format_desc,
             db_name: pb_param.db_name,

src/connector/src/source/google_pubsub/source/reader.rs

@@ -19,7 +19,7 @@ use futures_async_stream::try_stream;
 use google_cloud_pubsub::client::Client;
 use google_cloud_pubsub::subscription::{SeekTo, Subscription};
 use risingwave_common::bail;
-use tonic::Code;
+use tonic_0_9::Code;
 
 use super::TaggedReceivedMessage;
 use crate::parser::ParserConfig;

src/expr/core/src/window_function/kind.rs

@@ -42,18 +42,18 @@ impl WindowFuncKind {
         use risingwave_pb::expr::window_function::{PbGeneralType, PbType};
 
         let kind = match window_function_type {
-            PbType::General(typ) => match PbGeneralType::from_i32(*typ) {
-                Some(PbGeneralType::Unspecified) => bail!("Unspecified window function type"),
-                Some(PbGeneralType::RowNumber) => Self::RowNumber,
-                Some(PbGeneralType::Rank) => Self::Rank,
-                Some(PbGeneralType::DenseRank) => Self::DenseRank,
-                Some(PbGeneralType::Lag) => Self::Lag,
-                Some(PbGeneralType::Lead) => Self::Lead,
-                None => bail!("no such window function type"),
+            PbType::General(typ) => match PbGeneralType::try_from(*typ) {
+                Ok(PbGeneralType::Unspecified) => bail!("Unspecified window function type"),
+                Ok(PbGeneralType::RowNumber) => Self::RowNumber,
+                Ok(PbGeneralType::Rank) => Self::Rank,
+                Ok(PbGeneralType::DenseRank) => Self::DenseRank,
+                Ok(PbGeneralType::Lag) => Self::Lag,
+                Ok(PbGeneralType::Lead) => Self::Lead,
+                Err(_) => bail!("no such window function type"),
             },
-            PbType::Aggregate(agg_type) => match PbAggType::from_i32(*agg_type) {
-                Some(agg_type) => Self::Aggregate(AggKind::from_protobuf(agg_type)?),
-                None => bail!("no such aggregate function type"),
+            PbType::Aggregate(agg_type) => match PbAggType::try_from(*agg_type) {
+                Ok(agg_type) => Self::Aggregate(AggKind::from_protobuf(agg_type)?),
+                Err(_) => bail!("no such aggregate function type"),
             },
         };
         Ok(kind)

src/frontend/src/catalog/system_catalog/mod.rs

@@ -249,7 +249,7 @@ fn get_acl_items(
                 .unwrap()
                 .iter()
                 .for_each(|(action, option)| {
-                    let str = match Action::from_i32(*action).unwrap() {
+                    let str = match Action::try_from(*action).unwrap() {
                         Action::Select => "r",
                         Action::Insert => "a",
                         Action::Update => "w",

src/frontend/src/catalog/system_catalog/rw_catalog/rw_fragments.rs

@@ -51,9 +51,9 @@ impl SysCatalogReaderImpl {
         for i in 0..32 {
             let bit = 1 << i;
             if mask & bit != 0 {
-                match FragmentTypeFlag::from_i32(bit as i32) {
-                    None => continue,
-                    Some(flag) => result.push(flag),
+                match FragmentTypeFlag::try_from(bit as i32) {
+                    Err(_) => continue,
+                    Ok(flag) => result.push(flag),
                 };
             }
         }

src/frontend/src/scheduler/distributed/stage.rs

@@ -406,7 +406,7 @@ impl StageRunner {
             match status_res_inner {
                 Ok(status) => {
                     use risingwave_pb::task_service::task_info_response::TaskStatus as TaskStatusPb;
-                    match TaskStatusPb::from_i32(status.task_status).unwrap() {
+                    match TaskStatusPb::try_from(status.task_status).unwrap() {
                         TaskStatusPb::Running => {
                             running_task_cnt += 1;
                             // The task running count should always less or equal than the

src/java_binding/Cargo.toml

@@ -11,7 +11,7 @@ normal = ["workspace-hack"]
 
 [dependencies]
 jni = "0.21.1"
-prost = "0.11"
+prost = "0.12"
 risingwave_common = { workspace = true }
 risingwave_jni_core = { workspace = true }
 risingwave_pb = { workspace = true }

src/jni_core/Cargo.toml

@@ -16,7 +16,7 @@ futures = { version = "0.3", default-features = false, features = ["alloc"] }
 itertools = "0.11"
 jni = "0.21.1"
 paste = "1"
-prost = "0.11"
+prost = "0.12"
 risingwave_common = { workspace = true }
 risingwave_hummock_sdk = { workspace = true }
 risingwave_object_store = { workspace = true }

src/meta/Cargo.toml

@@ -43,7 +43,7 @@ parking_lot = { version = "0.12", features = ["arc_lock"] }
 parse-display = "0.8"
 prometheus = "0.13"
 prometheus-http-query = "0.6"
-prost = "0.11"
+prost = "0.12"
 rand = "0.8"
 regex = "1"
 reqwest = "0.11"
@@ -57,10 +57,22 @@ risingwave_pb = { workspace = true }
 risingwave_rpc_client = { workspace = true }
 risingwave_sqlparser = { workspace = true }
 scopeguard = "1.2.0"
-sea-orm = { version = "0.12.0", features = [ "sqlx-mysql", "sqlx-postgres", "sqlx-sqlite", "runtime-tokio-native-tls", "macros" ] }
+sea-orm = { version = "0.12.0", features = [
+    "sqlx-mysql",
+    "sqlx-postgres",
+    "sqlx-sqlite",
+    "runtime-tokio-native-tls",
+    "macros",
+] }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
-sqlx = { version = "0.7", features = ["runtime-tokio", "postgres", "mysql", "sqlite", "chrono"] }
+sqlx = { version = "0.7", features = [
+    "runtime-tokio",
+    "postgres",
+    "mysql",
+    "sqlite",
+    "chrono",
+] }
 sync-point = { path = "../utils/sync-point" }
 thiserror = "1"
 tokio = { version = "0.2", package = "madsim-tokio", features = [
@@ -74,6 +86,7 @@ tokio = { version = "0.2", package = "madsim-tokio", features = [
 tokio-retry = "0.3"
 tokio-stream = { version = "0.1", features = ["net"] }
 tonic = { workspace = true }
+tonic_0_9 = { package = "tonic", version = "0.9" }
 tower = { version = "0.4", features = ["util", "load-shed"] }
 tracing = "0.1"
 url = "2"

src/meta/src/dashboard/mod.rs

@@ -105,8 +105,8 @@ pub(super) mod handlers {
         let mut result = srv
             .cluster_manager
             .list_worker_node(
-                WorkerType::from_i32(ty)
-                    .ok_or_else(|| anyhow!("invalid worker type"))
+                WorkerType::try_from(ty)
+                    .map_err(|_| anyhow!("invalid worker type"))
                     .map_err(err)?,
                 None,
             )

src/meta/src/hummock/manager/mod.rs

@@ -2700,7 +2700,7 @@ impl HummockManager {
                                 sorted_output_ssts,
                                 table_stats_change
                             }) => {
-                                if let Err(e) =  hummock_manager.report_compact_task(task_id, TaskStatus::from_i32(task_status).unwrap(), sorted_output_ssts, Some(table_stats_change))
+                                if let Err(e) =  hummock_manager.report_compact_task(task_id, TaskStatus::try_from(task_status).unwrap(), sorted_output_ssts, Some(table_stats_change))
                                        .await {
                                         tracing::error!("report compact_tack fail {e:?}");
                                 }

src/meta/src/hummock/mock_hummock_meta_client.rs

@@ -292,7 +292,7 @@ impl HummockMetaClient for MockHummockMetaClient {
                         if let Err(e) = hummock_manager_compact
                             .report_compact_task(
                                 task_id,
-                                TaskStatus::from_i32(task_status).unwrap(),
+                                TaskStatus::try_from(task_status).unwrap(),
                                 sorted_output_ssts,
                                 Some(table_stats_change),
                             )

src/meta/src/rpc/service/user_service.rs

@@ -151,7 +151,7 @@ impl UserService for UserServiceImpl {
         let update_fields = req
             .update_fields
             .iter()
-            .map(|i| UpdateField::from_i32(*i).unwrap())
+            .map(|i| UpdateField::try_from(*i).unwrap())
             .collect_vec();
         let user = req.get_user()?.clone();
         let version = self

src/meta/src/storage/etcd_retry_client.rs

@@ -50,9 +50,9 @@ impl EtcdRetryClient {
     fn should_retry(err: &Error) -> bool {
         match err {
             Error::GRpcStatus(status) => {
-                status.code() == tonic::Code::Unavailable
-                    || status.code() == tonic::Code::Unknown
-                    || (status.code() == tonic::Code::Unauthenticated
+                status.code() == tonic_0_9::Code::Unavailable
+                    || status.code() == tonic_0_9::Code::Unknown
+                    || (status.code() == tonic_0_9::Code::Unauthenticated
                         && status.message().contains("invalid auth token"))
             }
             _ => false,

src/meta/src/storage/wrapped_etcd_client.rs

@@ -82,7 +82,7 @@ impl EtcdRefreshClient {
     fn should_refresh(err: &etcd_client::Error) -> bool {
         match err {
             etcd_client::Error::GRpcStatus(status) => {
-                status.code() == tonic::Code::Unauthenticated
+                status.code() == tonic_0_9::Code::Unauthenticated
                     && status.message().contains("invalid auth token")
             }
             _ => false,

src/prost/Cargo.toml

@@ -10,7 +10,7 @@ repository = { workspace = true }
 [dependencies]
 enum-as-inner = "0.6"
 pbjson = "0.6"
-prost = "0.11"
+prost = "0.12"
 prost-helpers = { path = "helpers" }
 serde = { version = "1", features = ["derive"] }
 strum = "0.25"

src/storage/Cargo.toml

@@ -38,7 +38,7 @@ more-asserts = "0.3"
 num-integer = "0.1"
 parking_lot = "0.12"
 prometheus = { version = "0.13", features = ["process"] }
-prost = "0.11"
+prost = "0.12"
 rand = "0.8"
 risingwave_backup = { workspace = true }
 risingwave_common = { workspace = true }

src/storage/backup/Cargo.toml

@@ -19,7 +19,7 @@ async-trait = "0.1"
 bytes = { version = "1", features = ["serde"] }
 itertools = "0.11"
 parking_lot = { version = "0.12", features = ["arc_lock"] }
-prost = "0.11"
+prost = "0.12"
 risingwave_common = { workspace = true }
 risingwave_hummock_sdk = { workspace = true }
 risingwave_object_store = { workspace = true }

src/storage/hummock_trace/Cargo.toml

@@ -16,7 +16,7 @@ bytes = { version = "1", features = ["serde"] }
 futures = { version = "0.3", default-features = false, features = ["alloc"] }
 futures-async-stream = "0.2"
 parking_lot = "0.12"
-prost = "0.11"
+prost = "0.12"
 risingwave_common = { workspace = true }
 risingwave_hummock_sdk = { workspace = true }
 risingwave_pb = { workspace = true }

src/stream/Cargo.toml

@@ -45,7 +45,7 @@ parking_lot = "0.12"
 parse-display = "0.8"
 pin-project = "1"
 prometheus = { version = "0.13", features = ["process"] }
-prost = "0.11"
+prost = "0.12"
 rand = "0.8"
 risingwave_common = { workspace = true }
 risingwave_connector = { workspace = true }

src/udf/src/error.rs

@@ -45,7 +45,7 @@ pub enum Error {
     ServiceError(String),
 }
 
-static_assertions::const_assert_eq!(std::mem::size_of::<Error>(), 32);
+static_assertions::const_assert_eq!(std::mem::size_of::<Error>(), 40);
 
 impl From<tonic::Status> for Error {
     fn from(status: tonic::Status) -> Self {