refactor: refactor arrow conversion

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/batch/Cargo.toml

@@ -15,9 +15,6 @@ normal = ["workspace-hack"]
 
 [dependencies]
 anyhow = "1"
-arrow-array = { workspace = true }
-arrow-array-iceberg = { workspace = true }
-arrow-schema = { workspace = true }
 assert_matches = "1"
 async-recursion = "1"
 async-trait = "0.1"

src/common/Cargo.toml

@@ -17,18 +17,18 @@ normal = ["workspace-hack"]
 ahash = "0.8"
 anyhow = "1"
 arc-swap = "1"
-arrow-array = { workspace = true }
-arrow-array-deltalake = { workspace = true }
-arrow-array-iceberg = { workspace = true }
-arrow-buffer = { workspace = true }
-arrow-buffer-deltalake = { workspace = true }
-arrow-buffer-iceberg = { workspace = true }
-arrow-cast = { workspace = true }
-arrow-cast-deltalake = { workspace = true }
-arrow-cast-iceberg = { workspace = true }
-arrow-schema = { workspace = true }
-arrow-schema-deltalake = { workspace = true }
-arrow-schema-iceberg = { 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

@@ -0,0 +1,48 @@
+// 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 = arrow_buffer::IntervalMonthDayNano;
+pub use arrow_impl::{FromArrow, ToArrow};
+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,
+};
+
+use crate::array::Interval;
+
+impl super::ArrowIntervalTypeTrait for ArrowIntervalType {
+    fn to_interval(self) -> Interval {
+        // XXX: the arrow-rs decoding is incorrect
+        // let (months, days, ns) = arrow_array::types::IntervalMonthDayNanoType::to_parts(value);
+        Interval::from_month_day_usec(self.months, self.days, self.nanoseconds / 1000)
+    }
+
+    fn from_interval(value: Interval) -> Self {
+        // XXX: the arrow-rs encoding is incorrect
+        // arrow_array::types::IntervalMonthDayNanoType::make_value(
+        //     self.months(),
+        //     self.days(),
+        //     // TODO: this may overflow and we need `try_into`
+        //     self.usecs() * 1000,
+        // )
+        Self {
+            months: value.months(),
+            days: value.days(),
+            nanoseconds: value.usecs() * 1000,
+        }
+    }
+}

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

@@ -22,17 +22,11 @@ use std::sync::Arc;
 
 use arrow_array::ArrayRef;
 use num_traits::abs;
-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

@@ -17,46 +17,14 @@ use std::collections::HashMap;
 use std::ops::{Div, Mul};
 use std::sync::Arc;
 
-use arrow_array_iceberg::{self as arrow_array, ArrayRef};
-use arrow_buffer_iceberg::IntervalMonthDayNano as ArrowIntervalType;
+use arrow_array::ArrayRef;
 use num_traits::abs;
-use {
-    arrow_buffer_iceberg as arrow_buffer, arrow_cast_iceberg as arrow_cast,
-    arrow_schema_iceberg as arrow_schema,
-};
 
+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::{Interval, StructType};
-
-impl ArrowIntervalTypeTrait for ArrowIntervalType {
-    fn to_interval(self) -> Interval {
-        // XXX: the arrow-rs decoding is incorrect
-        // let (months, days, ns) = arrow_array::types::IntervalMonthDayNanoType::to_parts(value);
-        Interval::from_month_day_usec(self.months, self.days, self.nanoseconds / 1000)
-    }
-
-    fn from_interval(value: Interval) -> Self {
-        // XXX: the arrow-rs encoding is incorrect
-        // arrow_array::types::IntervalMonthDayNanoType::make_value(
-        //     self.months(),
-        //     self.days(),
-        //     // TODO: this may overflow and we need `try_into`
-        //     self.usecs() * 1000,
-        // )
-        Self {
-            months: value.months(),
-            days: value.days(),
-            nanoseconds: value.usecs() * 1000,
-        }
-    }
-}
-
-#[path = "./arrow_impl.rs"]
-mod arrow_impl;
-
-use arrow_impl::{FromArrow, ToArrow};
-
-use crate::array::arrow::ArrowIntervalTypeTrait;
+use crate::types::StructType;
 
 pub struct IcebergArrowConvert;
 
@@ -261,11 +229,9 @@ impl ToArrow for IcebergCreateTableArrowConvert {
 mod test {
     use std::sync::Arc;
 
-    use arrow_array_iceberg::{ArrayRef, Decimal128Array};
-    use arrow_schema_iceberg::DataType;
-
-    use super::arrow_impl::ToArrow;
-    use super::IcebergArrowConvert;
+    use super::arrow_array::{ArrayRef, Decimal128Array};
+    use super::arrow_schema::DataType;
+    use super::*;
     use crate::array::{Decimal, DecimalArray};
 
     #[test]

src/common/src/array/arrow/arrow_impl.rs

@@ -42,8 +42,8 @@
 
 use std::fmt::Write;
 
+use arrow_array::array;
 use arrow_array::cast::AsArray;
-use arrow_array_iceberg::array;
 use arrow_buffer::OffsetBuffer;
 use chrono::{DateTime, NaiveDateTime, NaiveTime};
 use itertools::Itertools;

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

@@ -20,16 +20,11 @@
 
 use std::sync::Arc;
 
-pub use arrow_impl::{FromArrow, ToArrow};
-use {arrow_array, arrow_buffer, arrow_cast, arrow_schema};
-type ArrowIntervalType = i128;
-
+pub use super::arrow_50::{
+    arrow_array, arrow_buffer, arrow_cast, arrow_schema, FromArrow, ToArrow,
+};
 use crate::array::{ArrayError, ArrayImpl, DataType, DecimalArray, JsonbArray};
 
-#[expect(clippy::duplicate_mod)]
-#[path = "./arrow_impl.rs"]
-mod arrow_impl;
-
 /// Arrow conversion for UDF.
 #[derive(Default, Debug)]
 pub struct UdfArrowConvert {

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

@@ -12,16 +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, 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;