revert upgrade

Cargo.toml

@@ -140,32 +140,18 @@ prost-build = { version = "0.13" }
 icelake = { git = "https://github.com/risingwavelabs/icelake.git", rev = "3f4724158acee37a4785f56670a1427993a58739", features = [
     "prometheus",
 ] }
-arrow-array-iceberg = { package = "arrow-array", version = "52" }
-arrow-schema-iceberg = { package = "arrow-schema", version = "52" }
-arrow-buffer-iceberg = { package = "arrow-buffer", version = "52" }
-arrow-cast-iceberg = { package = "arrow-cast", version = "52" }
+
 # branch dev
 iceberg = { git = "https://github.com/risingwavelabs/iceberg-rust.git", rev = "84bf51c9d0d5886e4ee306ca4f383f029e1767a4" }
 iceberg-catalog-rest = { git = "https://github.com/risingwavelabs/iceberg-rust.git", rev = "84bf51c9d0d5886e4ee306ca4f383f029e1767a4" }
 iceberg-catalog-glue = { git = "https://github.com/risingwavelabs/iceberg-rust.git", rev = "84bf51c9d0d5886e4ee306ca4f383f029e1767a4" }
 opendal = "0.47"
-arrow-array = "50"
-arrow-arith = "50"
-arrow-cast = "50"
-arrow-schema = "50"
-arrow-buffer = "50"
+# used only by arrow-udf-flight
 arrow-flight = "50"
-arrow-select = "50"
-arrow-ord = "50"
-arrow-row = "50"
 arrow-udf-js = "0.3.1"
 arrow-udf-wasm = { version = "0.2.2", features = ["build"] }
 arrow-udf-python = "0.2"
 arrow-udf-flight = "0.1"
-arrow-array-deltalake = { package = "arrow-array", version = "48.0.1" }
-arrow-buffer-deltalake = { package = "arrow-buffer", version = "48.0.1" }
-arrow-cast-deltalake = { package = "arrow-cast", version = "48.0.1" }
-arrow-schema-deltalake = { package = "arrow-schema", version = "48.0.1" }
 clap = { version = "4", features = ["cargo", "derive", "env"] }
 # Use a forked version which removes the dependencies on dynamo db to reduce
 # compile time and binary size.

src/common/Cargo.toml

@@ -17,14 +17,18 @@ normal = ["workspace-hack"]
 ahash = "0.8"
 anyhow = "1"
 arc-swap = "1"
-arrow-52-array = { workspace = true }
-arrow-52-buffer = { workspace = true }
-arrow-52-cast = { workspace = true }
-arrow-52-schema = { workspace = true }
-arrow-array-deltalake = { workspace = true }
-arrow-buffer-deltalake = { workspace = true }
-arrow-cast-deltalake = { workspace = true }
-arrow-schema-deltalake = { workspace = true }
+arrow-48-array = { package = "arrow-array", version = "48" }
+arrow-48-buffer = { package = "arrow-buffer", version = "48" }
+arrow-48-cast = { package = "arrow-cast", version = "48" }
+arrow-48-schema = { package = "arrow-schema", version = "48" }
+arrow-50-array = { package = "arrow-array", version = "50" }
+arrow-50-buffer = { package = "arrow-buffer", version = "50" }
+arrow-50-cast = { package = "arrow-cast", version = "50" }
+arrow-50-schema = { package = "arrow-schema", version = "50" }
+arrow-52-array = { package = "arrow-array", version = "52" }
+arrow-52-buffer = { package = "arrow-buffer", version = "52" }
+arrow-52-cast = { package = "arrow-cast", version = "52" }
+arrow-52-schema = { package = "arrow-schema", version = "52" }
 async-trait = "0.1"
 auto_enums = { workspace = true }
 auto_impl = "1"

src/common/src/array/arrow/arrow_48.rs

@@ -0,0 +1,23 @@
+// Copyright 2024 RisingWave 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.
+
+#[allow(clippy::duplicate_mod)]
+#[path = "./arrow_impl.rs"]
+mod arrow_impl;
+type ArrowIntervalType = i128;
+pub use arrow_impl::{FromArrow, ToArrow};
+pub use {
+    arrow_48_array as arrow_array, arrow_48_buffer as arrow_buffer, arrow_48_cast as arrow_cast,
+    arrow_48_schema as arrow_schema,
+};

src/common/src/array/arrow/arrow_50.rs

@@ -0,0 +1,23 @@
+// Copyright 2024 RisingWave 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.
+
+#[allow(clippy::duplicate_mod)]
+#[path = "./arrow_impl.rs"]
+mod arrow_impl;
+type ArrowIntervalType = i128;
+pub use arrow_impl::{FromArrow, ToArrow};
+pub use {
+    arrow_50_array as arrow_array, arrow_50_buffer as arrow_buffer, arrow_50_cast as arrow_cast,
+    arrow_50_schema as arrow_schema,
+};

src/common/src/array/arrow/arrow_52.rs

@@ -12,11 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#[allow(clippy::duplicate_mod)]
 #[path = "./arrow_impl.rs"]
 mod arrow_impl;
-use arrow_buffer::IntervalMonthDayNano as ArrowIntervalType;
+type ArrowIntervalType = arrow_buffer::IntervalMonthDayNano;
 pub use arrow_impl::{FromArrow, ToArrow};
-use {
+pub use {
     arrow_52_array as arrow_array, arrow_52_buffer as arrow_buffer, arrow_52_cast as arrow_cast,
     arrow_52_schema as arrow_schema,
 };

src/common/src/array/arrow/arrow_deltalake.rs

@@ -22,17 +22,11 @@ use std::sync::Arc;
 
 use arrow_array::ArrayRef;
 use num_traits::abs;
-pub use {
-    arrow_array_deltalake as arrow_array, arrow_buffer_deltalake as arrow_buffer,
-    arrow_cast_deltalake as arrow_cast, arrow_schema_deltalake as arrow_schema,
-};
-type ArrowIntervalType = i128;
 
-use self::arrow_impl::ToArrow;
+pub use super::arrow_48::{
+    arrow_array, arrow_buffer, arrow_cast, arrow_schema, FromArrow, ToArrow,
+};
 use crate::array::{Array, ArrayError, DataChunk, Decimal, DecimalArray};
-#[expect(clippy::duplicate_mod)]
-#[path = "./arrow_impl.rs"]
-mod arrow_impl;
 
 pub struct DeltaLakeConvert;
 
@@ -103,8 +97,8 @@ mod test {
     use arrow_array::cast::AsArray;
     use arrow_array::ArrayRef;
     use arrow_schema::Field;
-    use {arrow_array_deltalake as arrow_array, arrow_schema_deltalake as arrow_schema};
 
+    use super::*;
     use crate::array::arrow::arrow_deltalake::DeltaLakeConvert;
     use crate::array::{ArrayImpl, Decimal, DecimalArray, ListArray, ListValue};
     use crate::bitmap::Bitmap;

src/common/src/array/arrow/arrow_iceberg.rs

@@ -19,12 +19,10 @@ use std::sync::Arc;
 
 use arrow_array::ArrayRef;
 use num_traits::abs;
-pub use {
-    arrow_52_array as arrow_array, arrow_52_buffer as arrow_buffer, arrow_52_cast as arrow_cast,
-    arrow_52_schema as arrow_schema,
-};
 
-pub use super::arrow_52::{FromArrow, ToArrow};
+pub use super::arrow_52::{
+    arrow_array, arrow_buffer, arrow_cast, arrow_schema, FromArrow, ToArrow,
+};
 use crate::array::{Array, ArrayError, ArrayImpl, DataChunk, DataType, DecimalArray};
 use crate::types::StructType;
 
@@ -232,9 +230,8 @@ mod test {
     use std::sync::Arc;
 
     use super::arrow_array::{ArrayRef, Decimal128Array};
-    use super::arrow_impl::ToArrow;
     use super::arrow_schema::DataType;
-    use super::IcebergArrowConvert;
+    use super::*;
     use crate::array::{Decimal, DecimalArray};
 
     #[test]

src/common/src/array/arrow/arrow_udf.rs

@@ -20,12 +20,9 @@
 
 use std::sync::Arc;
 
-pub use {
-    arrow_52_array as arrow_array, arrow_52_buffer as arrow_buffer, arrow_52_cast as arrow_cast,
-    arrow_52_schema as arrow_schema,
+pub use super::arrow_50::{
+    arrow_array, arrow_buffer, arrow_cast, arrow_schema, FromArrow, ToArrow,
 };
-
-pub use super::arrow_52::{FromArrow, ToArrow};
 use crate::array::{ArrayError, ArrayImpl, DataType, DecimalArray, JsonbArray};
 
 /// Arrow conversion for UDF.

src/common/src/array/arrow/mod.rs

@@ -12,31 +12,44 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+// These mods imports arrow_impl.rs to provide FromArrow, ToArrow traits for corresponding arrow versions,
+// and the default From/To implementations.
+mod arrow_48;
+mod arrow_50;
 mod arrow_52;
+// These mods import mods above and may override some methods.
 mod arrow_deltalake;
 mod arrow_iceberg;
 mod arrow_udf;
 
 pub use arrow_deltalake::DeltaLakeConvert;
 pub use arrow_iceberg::{IcebergArrowConvert, IcebergCreateTableArrowConvert};
-pub use arrow_udf::{FromArrow, ToArrow, UdfArrowConvert};
+pub use arrow_udf::UdfArrowConvert;
 pub use reexport::*;
+/// For other RisingWave crates, they can directly use arrow re-exported here, without adding
+/// `arrow` dependencies in their `Cargo.toml`. And they don't need to care about the version.
 mod reexport {
     pub use super::arrow_deltalake::{
         arrow_array as arrow_array_deltalake, arrow_buffer as arrow_buffer_deltalake,
         arrow_cast as arrow_cast_deltalake, arrow_schema as arrow_schema_deltalake,
+        FromArrow as DeltaLakeFromArrow, ToArrow as DeltaLakeToArrow,
     };
     pub use super::arrow_iceberg::{
         arrow_array as arrow_array_iceberg, arrow_buffer as arrow_buffer_iceberg,
         arrow_cast as arrow_cast_iceberg, arrow_schema as arrow_schema_iceberg,
+        FromArrow as IcebergFromArrow, ToArrow as IcebergToArrow,
     };
     pub use super::arrow_udf::{
         arrow_array as arrow_array_udf, arrow_buffer as arrow_buffer_udf,
-        arrow_cast as arrow_cast_udf, arrow_schema as arrow_schema_udf,
+        arrow_cast as arrow_cast_udf, arrow_schema as arrow_schema_udf, FromArrow as UdfFromArrow,
+        ToArrow as UdfToArrow,
     };
 }
 use crate::types::Interval;
 
+/// Arrow 52 changed the interval type from `i128` to `arrow_buffer::IntervalMonthDayNano`, so
+/// we introduced this trait to customize the conversion in `arrow_impl.rs`.
+/// We may delete this after all arrow versions are upgraded.
 trait ArrowIntervalTypeTrait {
     fn to_interval(self) -> Interval;
     fn from_interval(value: Interval) -> Self;

src/connector/src/error.rs

@@ -64,8 +64,6 @@ def_anyhow_newtype! {
     icelake::Error => "Iceberg error",
     iceberg::Error => "IcebergV2 error",
     redis::RedisError => "Redis error",
-    // currently, the following two are the same type
-    // risingwave_common::array::arrow::arrow_schema_udf::ArrowError => "Arrow error",
     risingwave_common::array::arrow::arrow_schema_iceberg::ArrowError => "Arrow error",
     google_cloud_pubsub::client::google_cloud_auth::error::Error => "Google Cloud error",
     rumqttc::tokio_rustls::rustls::Error => "TLS error",

src/expr/core/src/aggregate/user_defined.rs

@@ -17,7 +17,7 @@ use std::sync::Arc;
 use anyhow::Context;
 use risingwave_common::array::arrow::arrow_array_udf::ArrayRef;
 use risingwave_common::array::arrow::arrow_schema_udf::{Field, Fields, Schema, SchemaRef};
-use risingwave_common::array::arrow::{FromArrow, ToArrow, UdfArrowConvert};
+use risingwave_common::array::arrow::{UdfArrowConvert, UdfFromArrow, UdfToArrow};
 use risingwave_common::array::Op;
 use risingwave_common::bitmap::Bitmap;
 use risingwave_pb::expr::PbUserDefinedFunctionMetadata;

src/expr/core/src/expr/expr_udf.rs

@@ -19,7 +19,7 @@ use anyhow::Context;
 use await_tree::InstrumentAwait;
 use prometheus::{exponential_buckets, Registry};
 use risingwave_common::array::arrow::arrow_schema_udf::{Fields, Schema, SchemaRef};
-use risingwave_common::array::arrow::{FromArrow, ToArrow, UdfArrowConvert};
+use risingwave_common::array::arrow::{UdfArrowConvert, UdfFromArrow, UdfToArrow};
 use risingwave_common::array::{Array, ArrayRef, DataChunk};
 use risingwave_common::metrics::*;
 use risingwave_common::monitor::GLOBAL_METRICS_REGISTRY;

src/expr/core/src/table_function/user_defined.rs

@@ -16,7 +16,7 @@ use std::sync::Arc;
 
 use anyhow::Context;
 use risingwave_common::array::arrow::arrow_schema_udf::{Fields, Schema, SchemaRef};
-use risingwave_common::array::arrow::{FromArrow, ToArrow, UdfArrowConvert};
+use risingwave_common::array::arrow::{UdfArrowConvert, UdfFromArrow, UdfToArrow};
 use risingwave_common::array::I32Array;
 use risingwave_common::bail;
 

src/expr/impl/src/udf/quickjs.rs

@@ -15,7 +15,7 @@
 use arrow_udf_js::{CallMode, Runtime};
 use futures_util::StreamExt;
 use risingwave_common::array::arrow::arrow_schema_udf::{DataType, Field};
-use risingwave_common::array::arrow::{ToArrow, UdfArrowConvert};
+use risingwave_common::array::arrow::{UdfArrowConvert, UdfToArrow};
 
 use super::*;