chore: bump Arrow stack to 59

Cargo.toml

@@ -34,7 +34,7 @@ members = ["crates/fluss", "crates/fluss-test-cluster", "crates/examples", "bind
 fluss = { package = "fluss-rs", version = "0.2.0", path = "crates/fluss", features = ["storage-all"] }
 tokio = { version = "1.44.2", features = ["full"] }
 clap = { version = "4.5.37", features = ["derive"] }
-arrow = { version = "57.0.0", features = ["ipc_compression", "ffi"] }
+arrow = { version = "59.0.0", features = ["ipc_compression", "ffi"] }
 bigdecimal = "0.4"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"

bindings/python/Cargo.toml

@@ -27,14 +27,14 @@ name = "fluss"
 crate-type = ["cdylib"]
 
 [dependencies]
-pyo3 = { version = "0.26.0", features = ["extension-module", "generate-import-lib"] }
+pyo3 = { version = "0.28.0", features = ["extension-module", "generate-import-lib"] }
 fluss = { workspace = true, features = ["storage-all"] }
 tokio = { workspace = true }
 arrow = { workspace = true }
-arrow-pyarrow = "57.0.0"
-arrow-schema = "57.0.0"
-arrow-array = "57.0.0"
-pyo3-async-runtimes = { version = "0.26.0", features = ["tokio-runtime"] }
+arrow-pyarrow = "59.0.0"
+arrow-schema = "59.0.0"
+arrow-array = "59.0.0"
+pyo3-async-runtimes = { version = "0.28.0", features = ["tokio-runtime"] }
 jiff = { workspace = true }
 bigdecimal = "0.4"
 indexmap = "2"

bindings/python/src/config.rs

@@ -19,7 +19,7 @@ use crate::*;
 use pyo3::types::PyDict;
 
 /// Configuration for Fluss client
-#[pyclass]
+#[pyclass(from_py_object)]
 #[derive(Clone)]
 pub struct Config {
     inner: fcore::config::Config,

bindings/python/src/error.rs

@@ -26,7 +26,7 @@ use pyo3::prelude::*;
 const CLIENT_ERROR_CODE: i32 = -2;
 
 /// Fluss errors
-#[pyclass(extends=PyException)]
+#[pyclass(extends=PyException, from_py_object)]
 #[derive(Debug, Clone)]
 pub struct FlussError {
     #[pyo3(get)]

bindings/python/src/lib.rs

@@ -56,7 +56,7 @@ static TOKIO_RUNTIME: LazyLock<Runtime> = LazyLock::new(|| {
 ///   OffsetSpec.earliest()
 ///   OffsetSpec.latest()
 ///   OffsetSpec.timestamp(ts)
-#[pyclass]
+#[pyclass(from_py_object)]
 #[derive(Clone)]
 pub struct OffsetSpec {
     pub(crate) inner: fcore::rpc::message::OffsetSpec,

bindings/python/src/metadata.rs

@@ -20,7 +20,7 @@ use pyo3::types::PyDict;
 use std::collections::HashMap;
 
 /// Represents the type of change for a record in a log
-#[pyclass(eq, eq_int)]
+#[pyclass(eq, eq_int, from_py_object)]
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub enum ChangeType {
     /// Append-only operation
@@ -71,7 +71,7 @@ impl ChangeType {
 }
 
 /// Represents a table path with database and table name
-#[pyclass]
+#[pyclass(from_py_object)]
 #[derive(Clone)]
 pub struct TablePath {
     database_name: String,
@@ -260,7 +260,7 @@ impl TableDistribution {
 }
 
 /// Table descriptor containing schema and metadata
-#[pyclass]
+#[pyclass(from_py_object)]
 #[derive(Clone)]
 pub struct TableDescriptor {
     __tbl_desc: fcore::metadata::TableDescriptor,
@@ -363,7 +363,7 @@ impl TableDescriptor {
 }
 
 /// Information about a Fluss table
-#[pyclass]
+#[pyclass(from_py_object)]
 #[derive(Clone)]
 pub struct TableInfo {
     __table_info: fcore::metadata::TableInfo,
@@ -479,15 +479,15 @@ impl TableInfo {
 }
 
 /// Represents a lake snapshot with snapshot ID and table bucket offsets
-#[pyclass]
+#[pyclass(from_py_object)]
 #[derive(Clone)]
 pub struct LakeSnapshot {
     snapshot_id: i64,
     table_buckets_offset: HashMap<fcore::metadata::TableBucket, i64>,
 }
 
 /// Represents a table bucket with table ID, partition ID, and bucket ID
-#[pyclass]
+#[pyclass(from_py_object)]
 #[derive(Eq, Hash, PartialEq, Clone)]
 pub struct TableBucket {
     table_id: i64,
@@ -665,7 +665,7 @@ impl LakeSnapshot {
 }
 
 /// Descriptor for a Fluss database (comment and custom properties)
-#[pyclass]
+#[pyclass(from_py_object)]
 #[derive(Clone)]
 pub struct DatabaseDescriptor {
     __descriptor: fcore::metadata::DatabaseDescriptor,

bindings/python/src/table.rs

@@ -238,7 +238,7 @@ impl ScanRecords {
             ));
         }
         // Try slice
-        if let Ok(slice) = key.downcast::<PySlice>() {
+        if let Ok(slice) = key.cast::<PySlice>() {
             let indices = slice.indices(self.total_count as isize)?;
             let mut result: Vec<Py<ScanRecord>> = Vec::new();
             let mut i = indices.start;
@@ -1364,11 +1364,11 @@ fn python_value_to_datum(value: &Bound<PyAny>, data_type: &DataType) -> PyResult
             Ok(v.into())
         }
         DataType::Bytes(_) | DataType::Binary(_) => {
-            // Efficient extraction: downcast to specific type and use bulk copy.
+            // Efficient extraction: cast to specific type and use bulk copy.
             // PyBytes::as_bytes() and PyByteArray::to_vec() are O(n) bulk copies of the underlying data.
-            if let Ok(bytes) = value.downcast::<PyBytes>() {
+            if let Ok(bytes) = value.cast::<PyBytes>() {
                 Ok(bytes.as_bytes().to_vec().into())
-            } else if let Ok(bytearray) = value.downcast::<PyByteArray>() {
+            } else if let Ok(bytearray) = value.cast::<PyByteArray>() {
                 Ok(bytearray.to_vec().into())
             } else {
                 Err(FlussError::new_err(format!(
@@ -1392,7 +1392,7 @@ fn python_value_to_datum(value: &Bound<PyAny>, data_type: &DataType) -> PyResult
                     get_type_name(value)
                 )));
             }
-            let seq = value.downcast::<PySequence>().map_err(|_| {
+            let seq = value.cast::<PySequence>().map_err(|_| {
                 FlussError::new_err(format!(
                     "Expected sequence for Array column, got {}",
                     get_type_name(value)
@@ -1480,7 +1480,7 @@ fn python_value_to_datum(value: &Bound<PyAny>, data_type: &DataType) -> PyResult
 fn python_map_pairs<'py>(
     value: &Bound<'py, PyAny>,
 ) -> PyResult<Vec<(Bound<'py, PyAny>, Bound<'py, PyAny>)>> {
-    if let Ok(dict) = value.downcast::<PyDict>() {
+    if let Ok(dict) = value.cast::<PyDict>() {
         return Ok(dict.iter().collect());
     }
     if value.is_instance_of::<PyString>() {
@@ -1489,7 +1489,7 @@ fn python_map_pairs<'py>(
             get_type_name(value)
         )));
     }
-    let seq = value.downcast::<PySequence>().map_err(|_| {
+    let seq = value.cast::<PySequence>().map_err(|_| {
         FlussError::new_err(format!(
             "Expected dict or sequence of (key, value) pairs for Map column, got {}",
             get_type_name(value)
@@ -1499,7 +1499,7 @@ fn python_map_pairs<'py>(
     let mut pairs = Vec::with_capacity(len);
     for i in 0..len {
         let entry = seq.get_item(i)?;
-        let pair = entry.downcast::<PySequence>().map_err(|_| {
+        let pair = entry.cast::<PySequence>().map_err(|_| {
             FlussError::new_err("Map entries must be (key, value) pairs".to_string())
         })?;
         if pair.len()? != 2 {
@@ -1941,7 +1941,7 @@ static UTC_EPOCH: PyOnceLock<Py<PyAny>> = PyOnceLock::new();
 fn get_decimal_type(py: Python) -> PyResult<Bound<PyType>> {
     let ty = DECIMAL_TYPE.get_or_try_init(py, || -> PyResult<_> {
         let decimal_mod = py.import("decimal")?;
-        let decimal_ty = decimal_mod.getattr("Decimal")?.downcast_into::<PyType>()?;
+        let decimal_ty = decimal_mod.getattr("Decimal")?.cast_into::<PyType>()?;
         Ok(decimal_ty.unbind())
     })?;
     Ok(ty.bind(py).clone())
@@ -1955,8 +1955,8 @@ fn get_utc_timezone(py: Python) -> PyResult<Bound<PyTzInfo>> {
         let utc = timezone.getattr("utc")?;
         Ok(utc.unbind())
     })?;
-    // Downcast to PyTzInfo for use with PyDateTime::new()
-    Ok(tz.bind(py).clone().downcast_into::<PyTzInfo>()?)
+    // Cast to PyTzInfo for use with PyDateTime::new()
+    Ok(tz.bind(py).clone().cast_into::<PyTzInfo>()?)
 }
 
 /// Get the cached UTC epoch datetime, creating it once per interpreter.
@@ -2013,13 +2013,13 @@ fn python_decimal_to_datum(
 /// Convert Python datetime.date to Datum::Date.
 fn python_date_to_datum(value: &Bound<PyAny>) -> PyResult<fcore::row::Datum<'static>> {
     // Reject datetime.datetime (subclass of date) - use timestamp columns for those
-    if value.downcast::<PyDateTime>().is_ok() {
+    if value.cast::<PyDateTime>().is_ok() {
         return Err(FlussError::new_err(
             "Expected datetime.date, got datetime.datetime. Use a TIMESTAMP column for datetime values.",
         ));
     }
 
-    let date = value.downcast::<PyDate>().map_err(|_| {
+    let date = value.cast::<PyDate>().map_err(|_| {
         FlussError::new_err(format!(
             "Expected datetime.date, got {}",
             get_type_name(value)
@@ -2048,7 +2048,7 @@ fn python_date_to_datum(value: &Bound<PyAny>) -> PyResult<fcore::row::Datum<'sta
 /// Sub-millisecond precision (microseconds not divisible by 1000) will raise an error
 /// to prevent silent data loss and ensure fail-fast behavior.
 fn python_time_to_datum(value: &Bound<PyAny>) -> PyResult<fcore::row::Datum<'static>> {
-    let time = value.downcast::<PyTime>().map_err(|_| {
+    let time = value.cast::<PyTime>().map_err(|_| {
         FlussError::new_err(format!(
             "Expected datetime.time, got {}",
             get_type_name(value)
@@ -2107,7 +2107,7 @@ fn python_datetime_to_timestamp_ltz(value: &Bound<PyAny>) -> PyResult<fcore::row
 /// For clarity, tz-aware datetimes are rejected - use TimestampLtz for those.
 fn extract_datetime_components_ntz(value: &Bound<PyAny>) -> PyResult<(i64, i32)> {
     // Try PyDateTime first
-    if let Ok(dt) = value.downcast::<PyDateTime>() {
+    if let Ok(dt) = value.cast::<PyDateTime>() {
         // Reject tz-aware datetime for NTZ - it's ambiguous what the user wants
         let tzinfo = dt.getattr("tzinfo")?;
         if !tzinfo.is_none() {
@@ -2137,7 +2137,7 @@ fn extract_datetime_components_ntz(value: &Bound<PyAny>) -> PyResult<(i64, i32)>
 
     // Try to_pydatetime() for objects that support it
     if let Ok(py_dt) = value.call_method0("to_pydatetime") {
-        if let Ok(dt) = py_dt.downcast::<PyDateTime>() {
+        if let Ok(dt) = py_dt.cast::<PyDateTime>() {
             let tzinfo = dt.getattr("tzinfo")?;
             if !tzinfo.is_none() {
                 return Err(FlussError::new_err(
@@ -2159,7 +2159,7 @@ fn extract_datetime_components_ntz(value: &Bound<PyAny>) -> PyResult<(i64, i32)>
 /// For naive datetimes, assumes UTC. For aware datetimes, converts to UTC.
 fn extract_datetime_components_ltz(value: &Bound<PyAny>) -> PyResult<(i64, i32)> {
     // Try PyDateTime first
-    if let Ok(dt) = value.downcast::<PyDateTime>() {
+    if let Ok(dt) = value.cast::<PyDateTime>() {
         // Check if timezone-aware
         let tzinfo = dt.getattr("tzinfo")?;
         if tzinfo.is_none() {
@@ -2180,7 +2180,7 @@ fn extract_datetime_components_ltz(value: &Bound<PyAny>) -> PyResult<(i64, i32)>
 
     // Try to_pydatetime()
     if let Ok(py_dt) = value.call_method0("to_pydatetime") {
-        if let Ok(dt) = py_dt.downcast::<PyDateTime>() {
+        if let Ok(dt) = py_dt.cast::<PyDateTime>() {
             let tzinfo = dt.getattr("tzinfo")?;
             if tzinfo.is_none() {
                 return datetime_to_epoch_millis_as_utc(dt);
@@ -2234,7 +2234,7 @@ fn datetime_to_epoch_millis_utc_aware(dt: &Bound<'_, PyDateTime>) -> PyResult<(i
 
     // Compute delta = dt - epoch (this handles timezone conversion correctly)
     let delta = dt.call_method1("__sub__", (epoch,))?;
-    let delta = delta.downcast::<PyDelta>()?;
+    let delta = delta.cast::<PyDelta>()?;
 
     // Extract components using integer arithmetic
     let days = delta.get_days() as i64;

crates/fluss/Cargo.toml

@@ -44,7 +44,7 @@ integration_tests = []
 
 [dependencies]
 arrow = { workspace = true }
-arrow-schema = "57.0.0"
+arrow-schema = "59.0.0"
 bitvec = "1"
 byteorder = "1.5"
 futures = "0.3"