Skip to main content

risingwave_common/array/arrow/
arrow_impl.rs

1// Copyright 2023 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Converts between arrays and Apache Arrow arrays.
16//!
17//! This file acts as a template file for conversion code between
18//! arrays and different version of Apache Arrow.
19//!
20//! The conversion logic will be implemented for the arrow version specified in the outer mod by
21//! `super::arrow_xxx`, such as `super::arrow_array`.
22//!
23//! When we want to implement the conversion logic for an arrow version, we first
24//! create a new mod file, and rename the corresponding arrow package name to `arrow_xxx`
25//! using the `use` clause, and then declare a sub-mod and set its file path with attribute
26//! `#[path = "./arrow_impl.rs"]` so that the code in this template file can be embedded to
27//! the new mod file, and the conversion logic can be implemented for the corresponding arrow
28//! version.
29//!
30//! Example can be seen in `arrow_default.rs`, which is also as followed:
31//! ```ignore
32//! use {arrow_array, arrow_buffer, arrow_cast, arrow_schema};
33//!
34//! #[allow(clippy::duplicate_mod)]
35//! #[path = "./arrow_impl.rs"]
36//! mod arrow_impl;
37//! ```
38
39// Is this a bug? Why do we have these lints?
40#![allow(unused_imports)]
41#![allow(dead_code)]
42
43use std::fmt::Write;
44use std::sync::Arc;
45
46use arrow_array::array;
47use arrow_array::cast::AsArray;
48use arrow_buffer::OffsetBuffer;
49use arrow_schema::TimeUnit;
50use chrono::{DateTime, Timelike as _};
51use itertools::Itertools;
52use thiserror_ext::AsReport;
53
54use super::arrow_schema::IntervalUnit;
55// This is important because we want to use the arrow version specified by the outer mod.
56use super::{ArrowIntervalType, arrow_array, arrow_buffer, arrow_cast, arrow_schema};
57// Other import should always use the absolute path.
58use crate::array::*;
59use crate::types::{DataType as RwDataType, Scalar, *};
60use crate::util::iter_util::ZipEqFast;
61
62/// Defines how to convert RisingWave arrays to Arrow arrays.
63///
64/// This trait allows for customized conversion logic for different external systems using Arrow.
65/// The default implementation is based on the `From` implemented in this mod.
66pub trait ToArrow {
67    /// Converts RisingWave `DataChunk` to Arrow `RecordBatch` with specified schema.
68    ///
69    /// This function will try to convert the array if the type is not same with the schema.
70    fn to_record_batch(
71        &self,
72        schema: arrow_schema::SchemaRef,
73        chunk: &DataChunk,
74    ) -> Result<arrow_array::RecordBatch, ArrayError> {
75        // compact the chunk if it's not compacted
76        if !chunk.is_vis_compacted() {
77            let c = chunk.clone();
78            return self.to_record_batch(schema, &c.compact_vis());
79        }
80
81        // convert each column to arrow array
82        let columns: Vec<_> = chunk
83            .columns()
84            .iter()
85            .zip_eq_fast(schema.fields().iter())
86            .map(|(column, field)| self.to_array(field.data_type(), column))
87            .try_collect()?;
88
89        // create record batch
90        let opts =
91            arrow_array::RecordBatchOptions::default().with_row_count(Some(chunk.capacity()));
92        arrow_array::RecordBatch::try_new_with_options(schema, columns, &opts)
93            .map_err(ArrayError::to_arrow)
94    }
95
96    /// Converts RisingWave array to Arrow array.
97    fn to_array(
98        &self,
99        data_type: &arrow_schema::DataType,
100        array: &ArrayImpl,
101    ) -> Result<arrow_array::ArrayRef, ArrayError> {
102        let arrow_array = match array {
103            ArrayImpl::Bool(array) => self.bool_to_arrow(array),
104            ArrayImpl::Int16(array) => self.int16_to_arrow(array),
105            ArrayImpl::Int32(array) => self.int32_to_arrow(array),
106            ArrayImpl::Int64(array) => self.int64_to_arrow(array),
107            ArrayImpl::Int256(array) => self.int256_to_arrow(array),
108            ArrayImpl::Float32(array) => self.float32_to_arrow(array),
109            ArrayImpl::Float64(array) => self.float64_to_arrow(array),
110            ArrayImpl::Date(array) => self.date_to_arrow(array),
111            ArrayImpl::Time(array) => self.time_to_arrow(array),
112            ArrayImpl::Timestamp(array) => self.timestamp_to_arrow(array),
113            ArrayImpl::Timestamptz(array) => self.timestamptz_to_arrow(array),
114            ArrayImpl::Interval(array) => self.interval_to_arrow(array),
115            ArrayImpl::Utf8(array) => self.utf8_to_arrow(array),
116            ArrayImpl::Bytea(array) => self.bytea_to_arrow(array),
117            ArrayImpl::Decimal(array) => self.decimal_to_arrow(data_type, array),
118            ArrayImpl::Jsonb(array) => self.jsonb_to_arrow(array),
119            ArrayImpl::Variant(array) => self.variant_to_arrow(array),
120            ArrayImpl::Serial(array) => self.serial_to_arrow(array),
121            ArrayImpl::List(array) => self.list_to_arrow(data_type, array),
122            ArrayImpl::Struct(array) => self.struct_to_arrow(data_type, array),
123            ArrayImpl::Map(array) => self.map_to_arrow(data_type, array),
124            ArrayImpl::Vector(inner) => self.vector_to_arrow(data_type, inner),
125        }?;
126        if arrow_array.data_type() != data_type {
127            arrow_cast::cast(&arrow_array, data_type).map_err(ArrayError::to_arrow)
128        } else {
129            Ok(arrow_array)
130        }
131    }
132
133    #[inline]
134    fn bool_to_arrow(&self, array: &BoolArray) -> Result<arrow_array::ArrayRef, ArrayError> {
135        Ok(Arc::new(arrow_array::BooleanArray::from(array)))
136    }
137
138    #[inline]
139    fn int16_to_arrow(&self, array: &I16Array) -> Result<arrow_array::ArrayRef, ArrayError> {
140        Ok(Arc::new(arrow_array::Int16Array::from(array)))
141    }
142
143    #[inline]
144    fn int32_to_arrow(&self, array: &I32Array) -> Result<arrow_array::ArrayRef, ArrayError> {
145        Ok(Arc::new(arrow_array::Int32Array::from(array)))
146    }
147
148    #[inline]
149    fn int64_to_arrow(&self, array: &I64Array) -> Result<arrow_array::ArrayRef, ArrayError> {
150        Ok(Arc::new(arrow_array::Int64Array::from(array)))
151    }
152
153    #[inline]
154    fn float32_to_arrow(&self, array: &F32Array) -> Result<arrow_array::ArrayRef, ArrayError> {
155        Ok(Arc::new(arrow_array::Float32Array::from(array)))
156    }
157
158    #[inline]
159    fn float64_to_arrow(&self, array: &F64Array) -> Result<arrow_array::ArrayRef, ArrayError> {
160        Ok(Arc::new(arrow_array::Float64Array::from(array)))
161    }
162
163    #[inline]
164    fn utf8_to_arrow(&self, array: &Utf8Array) -> Result<arrow_array::ArrayRef, ArrayError> {
165        Ok(Arc::new(arrow_array::StringArray::from(array)))
166    }
167
168    #[inline]
169    fn int256_to_arrow(&self, array: &Int256Array) -> Result<arrow_array::ArrayRef, ArrayError> {
170        Ok(Arc::new(arrow_array::Decimal256Array::from(array)))
171    }
172
173    #[inline]
174    fn date_to_arrow(&self, array: &DateArray) -> Result<arrow_array::ArrayRef, ArrayError> {
175        Ok(Arc::new(arrow_array::Date32Array::try_from(array)?))
176    }
177
178    #[inline]
179    fn timestamp_to_arrow(
180        &self,
181        array: &TimestampArray,
182    ) -> Result<arrow_array::ArrayRef, ArrayError> {
183        Ok(Arc::new(arrow_array::TimestampMicrosecondArray::try_from(
184            array,
185        )?))
186    }
187
188    #[inline]
189    fn timestamptz_to_arrow(
190        &self,
191        array: &TimestamptzArray,
192    ) -> Result<arrow_array::ArrayRef, ArrayError> {
193        Ok(Arc::new(
194            arrow_array::TimestampMicrosecondArray::try_from(array)?.with_timezone_utc(),
195        ))
196    }
197
198    #[inline]
199    fn time_to_arrow(&self, array: &TimeArray) -> Result<arrow_array::ArrayRef, ArrayError> {
200        Ok(Arc::new(arrow_array::Time64MicrosecondArray::try_from(
201            array,
202        )?))
203    }
204
205    #[inline]
206    fn interval_to_arrow(
207        &self,
208        array: &IntervalArray,
209    ) -> Result<arrow_array::ArrayRef, ArrayError> {
210        Ok(Arc::new(arrow_array::IntervalMonthDayNanoArray::try_from(
211            array,
212        )?))
213    }
214
215    #[inline]
216    fn bytea_to_arrow(&self, array: &BytesArray) -> Result<arrow_array::ArrayRef, ArrayError> {
217        Ok(Arc::new(arrow_array::BinaryArray::from(array)))
218    }
219
220    // Decimal values are stored as ASCII text representation in a string array.
221    #[inline]
222    fn decimal_to_arrow(
223        &self,
224        _data_type: &arrow_schema::DataType,
225        array: &DecimalArray,
226    ) -> Result<arrow_array::ArrayRef, ArrayError> {
227        Ok(Arc::new(arrow_array::StringArray::from(array)))
228    }
229
230    // JSON values are stored as text representation in a string array.
231    #[inline]
232    fn jsonb_to_arrow(&self, array: &JsonbArray) -> Result<arrow_array::ArrayRef, ArrayError> {
233        Ok(Arc::new(arrow_array::StringArray::from(array)))
234    }
235
236    // TODO: support the Parquet Variant Arrow extension layout.
237    #[inline]
238    fn variant_to_arrow(&self, _array: &VariantArray) -> Result<arrow_array::ArrayRef, ArrayError> {
239        Err(ArrayError::to_arrow(
240            "VARIANT is not supported in Arrow conversion yet",
241        ))
242    }
243
244    #[inline]
245    fn serial_to_arrow(&self, array: &SerialArray) -> Result<arrow_array::ArrayRef, ArrayError> {
246        Ok(Arc::new(arrow_array::Int64Array::from(array)))
247    }
248
249    #[inline]
250    fn list_to_arrow(
251        &self,
252        data_type: &arrow_schema::DataType,
253        array: &ListArray,
254    ) -> Result<arrow_array::ArrayRef, ArrayError> {
255        let arrow_schema::DataType::List(field) = data_type else {
256            return Err(ArrayError::to_arrow("Invalid list type"));
257        };
258        let values = self.to_array(field.data_type(), array.values())?;
259        let offsets = OffsetBuffer::new(array.offsets().iter().map(|&o| o as i32).collect());
260        let nulls = (!array.null_bitmap().all()).then(|| array.null_bitmap().into());
261        Ok(Arc::new(arrow_array::ListArray::new(
262            field.clone(),
263            offsets,
264            values,
265            nulls,
266        )))
267    }
268
269    #[inline]
270    fn vector_to_arrow(
271        &self,
272        data_type: &arrow_schema::DataType,
273        array: &VectorArray,
274    ) -> Result<arrow_array::ArrayRef, ArrayError> {
275        let arrow_schema::DataType::List(field) = data_type else {
276            return Err(ArrayError::to_arrow("Invalid list type"));
277        };
278        if field.data_type() != &arrow_schema::DataType::Float32 {
279            return Err(ArrayError::to_arrow("Invalid list inner type for vector"));
280        }
281        let values = Arc::new(arrow_array::Float32Array::from(
282            array.as_raw_slice().to_vec(),
283        ));
284        let offsets = OffsetBuffer::new(array.offsets().iter().map(|&o| o as i32).collect());
285        let nulls = (!array.null_bitmap().all()).then(|| array.null_bitmap().into());
286        Ok(Arc::new(arrow_array::ListArray::new(
287            field.clone(),
288            offsets,
289            values,
290            nulls,
291        )))
292    }
293
294    #[inline]
295    fn struct_to_arrow(
296        &self,
297        data_type: &arrow_schema::DataType,
298        array: &StructArray,
299    ) -> Result<arrow_array::ArrayRef, ArrayError> {
300        let arrow_schema::DataType::Struct(fields) = data_type else {
301            return Err(ArrayError::to_arrow("Invalid struct type"));
302        };
303        // Use `try_new_with_length` so that empty-field structs keep their row count;
304        // `StructArray::new` panics for empty `fields` because it derives length from
305        // the child arrays.
306        let len = array.len();
307        let child_arrays = array
308            .fields()
309            .zip_eq_fast(fields)
310            .map(|(arr, field)| self.to_array(field.data_type(), arr))
311            .try_collect::<_, _, ArrayError>()?;
312        let nulls = Some(array.null_bitmap().into());
313        Ok(Arc::new(
314            arrow_array::StructArray::try_new_with_length(fields.clone(), child_arrays, nulls, len)
315                .map_err(ArrayError::from_arrow)?,
316        ))
317    }
318
319    #[inline]
320    fn map_to_arrow(
321        &self,
322        data_type: &arrow_schema::DataType,
323        array: &MapArray,
324    ) -> Result<arrow_array::ArrayRef, ArrayError> {
325        let arrow_schema::DataType::Map(field, ordered) = data_type else {
326            return Err(ArrayError::to_arrow("Invalid map type"));
327        };
328        if *ordered {
329            return Err(ArrayError::to_arrow("Sorted map is not supported"));
330        }
331        let values = self
332            .struct_to_arrow(field.data_type(), array.as_struct())?
333            .as_struct()
334            .clone();
335        let offsets = OffsetBuffer::new(array.offsets().iter().map(|&o| o as i32).collect());
336        let nulls = (!array.null_bitmap().all()).then(|| array.null_bitmap().into());
337        Ok(Arc::new(arrow_array::MapArray::new(
338            field.clone(),
339            offsets,
340            values,
341            nulls,
342            *ordered,
343        )))
344    }
345
346    /// Convert RisingWave data type to Arrow data type.
347    ///
348    /// This function returns a `Field` instead of `DataType` because some may be converted to
349    /// extension types which require additional metadata in the field.
350    fn to_arrow_field(
351        &self,
352        name: &str,
353        value: &DataType,
354    ) -> Result<arrow_schema::Field, ArrayError> {
355        let data_type = match value {
356            // using the inline function
357            DataType::Boolean => self.bool_type_to_arrow(),
358            DataType::Int16 => self.int16_type_to_arrow(),
359            DataType::Int32 => self.int32_type_to_arrow(),
360            DataType::Int64 => self.int64_type_to_arrow(),
361            DataType::Int256 => self.int256_type_to_arrow(),
362            DataType::Float32 => self.float32_type_to_arrow(),
363            DataType::Float64 => self.float64_type_to_arrow(),
364            DataType::Date => self.date_type_to_arrow(),
365            DataType::Time => self.time_type_to_arrow(),
366            DataType::Timestamp => self.timestamp_type_to_arrow(),
367            DataType::Timestamptz => self.timestamptz_type_to_arrow(),
368            DataType::Interval => self.interval_type_to_arrow(),
369            DataType::Varchar => self.varchar_type_to_arrow(),
370            DataType::Bytea => self.bytea_type_to_arrow(),
371            DataType::Serial => self.serial_type_to_arrow(),
372            DataType::Decimal => return Ok(self.decimal_type_to_arrow(name)),
373            DataType::Jsonb => return Ok(self.jsonb_type_to_arrow(name)),
374            // TODO: support the Parquet Variant Arrow extension layout.
375            DataType::Variant => {
376                return Err(ArrayError::to_arrow(
377                    "VARIANT is not supported in Arrow conversion yet",
378                ));
379            }
380            DataType::Struct(fields) => self.struct_type_to_arrow(fields)?,
381            DataType::List(list) => self.list_type_to_arrow(list)?,
382            DataType::Map(map) => self.map_type_to_arrow(map)?,
383            DataType::Vector(_) => self.vector_type_to_arrow()?,
384        };
385        Ok(arrow_schema::Field::new(name, data_type, true))
386    }
387
388    #[inline]
389    fn bool_type_to_arrow(&self) -> arrow_schema::DataType {
390        arrow_schema::DataType::Boolean
391    }
392
393    #[inline]
394    fn int16_type_to_arrow(&self) -> arrow_schema::DataType {
395        arrow_schema::DataType::Int16
396    }
397
398    #[inline]
399    fn int32_type_to_arrow(&self) -> arrow_schema::DataType {
400        arrow_schema::DataType::Int32
401    }
402
403    #[inline]
404    fn int64_type_to_arrow(&self) -> arrow_schema::DataType {
405        arrow_schema::DataType::Int64
406    }
407
408    #[inline]
409    fn int256_type_to_arrow(&self) -> arrow_schema::DataType {
410        arrow_schema::DataType::Decimal256(arrow_schema::DECIMAL256_MAX_PRECISION, 0)
411    }
412
413    #[inline]
414    fn float32_type_to_arrow(&self) -> arrow_schema::DataType {
415        arrow_schema::DataType::Float32
416    }
417
418    #[inline]
419    fn float64_type_to_arrow(&self) -> arrow_schema::DataType {
420        arrow_schema::DataType::Float64
421    }
422
423    #[inline]
424    fn date_type_to_arrow(&self) -> arrow_schema::DataType {
425        arrow_schema::DataType::Date32
426    }
427
428    #[inline]
429    fn time_type_to_arrow(&self) -> arrow_schema::DataType {
430        arrow_schema::DataType::Time64(arrow_schema::TimeUnit::Microsecond)
431    }
432
433    #[inline]
434    fn timestamp_type_to_arrow(&self) -> arrow_schema::DataType {
435        arrow_schema::DataType::Timestamp(arrow_schema::TimeUnit::Microsecond, None)
436    }
437
438    #[inline]
439    fn timestamptz_type_to_arrow(&self) -> arrow_schema::DataType {
440        arrow_schema::DataType::Timestamp(
441            arrow_schema::TimeUnit::Microsecond,
442            Some("+00:00".into()),
443        )
444    }
445
446    #[inline]
447    fn interval_type_to_arrow(&self) -> arrow_schema::DataType {
448        arrow_schema::DataType::Interval(arrow_schema::IntervalUnit::MonthDayNano)
449    }
450
451    #[inline]
452    fn varchar_type_to_arrow(&self) -> arrow_schema::DataType {
453        arrow_schema::DataType::Utf8
454    }
455
456    #[inline]
457    fn jsonb_type_to_arrow(&self, name: &str) -> arrow_schema::Field {
458        arrow_schema::Field::new(name, arrow_schema::DataType::Utf8, true)
459            .with_metadata([("ARROW:extension:name".into(), "arrowudf.json".into())].into())
460    }
461
462    #[inline]
463    fn bytea_type_to_arrow(&self) -> arrow_schema::DataType {
464        arrow_schema::DataType::Binary
465    }
466
467    #[inline]
468    fn decimal_type_to_arrow(&self, name: &str) -> arrow_schema::Field {
469        arrow_schema::Field::new(name, arrow_schema::DataType::Utf8, true)
470            .with_metadata([("ARROW:extension:name".into(), "arrowudf.decimal".into())].into())
471    }
472
473    #[inline]
474    fn serial_type_to_arrow(&self) -> arrow_schema::DataType {
475        arrow_schema::DataType::Int64
476    }
477
478    #[inline]
479    fn list_type_to_arrow(
480        &self,
481        list_type: &ListType,
482    ) -> Result<arrow_schema::DataType, ArrayError> {
483        Ok(arrow_schema::DataType::List(Arc::new(
484            self.to_arrow_field("item", list_type.elem())?,
485        )))
486    }
487
488    #[inline]
489    fn struct_type_to_arrow(
490        &self,
491        fields: &StructType,
492    ) -> Result<arrow_schema::DataType, ArrayError> {
493        Ok(arrow_schema::DataType::Struct(
494            fields
495                .iter()
496                .map(|(name, ty)| self.to_arrow_field(name, ty))
497                .try_collect::<_, _, ArrayError>()?,
498        ))
499    }
500
501    #[inline]
502    fn map_type_to_arrow(&self, map_type: &MapType) -> Result<arrow_schema::DataType, ArrayError> {
503        let sorted = false;
504        // "key" is always non-null
505        let key = self
506            .to_arrow_field("key", map_type.key())?
507            .with_nullable(false);
508        let value = self.to_arrow_field("value", map_type.value())?;
509        Ok(arrow_schema::DataType::Map(
510            Arc::new(arrow_schema::Field::new(
511                "entries",
512                arrow_schema::DataType::Struct([Arc::new(key), Arc::new(value)].into()),
513                // "entries" is always non-null
514                false,
515            )),
516            sorted,
517        ))
518    }
519
520    #[inline]
521    fn vector_type_to_arrow(&self) -> Result<arrow_schema::DataType, ArrayError> {
522        Ok(arrow_schema::DataType::List(Arc::new(
523            self.to_arrow_field("item", &VECTOR_ITEM_TYPE)?,
524        )))
525    }
526}
527
528/// Defines how to convert Arrow arrays to RisingWave arrays.
529#[allow(clippy::wrong_self_convention)]
530pub trait FromArrow {
531    /// Converts Arrow `RecordBatch` to RisingWave `DataChunk`.
532    fn from_record_batch(&self, batch: &arrow_array::RecordBatch) -> Result<DataChunk, ArrayError> {
533        let mut columns = Vec::with_capacity(batch.num_columns());
534        for (array, field) in batch.columns().iter().zip_eq_fast(batch.schema().fields()) {
535            let column = Arc::new(self.from_array(field, array)?);
536            columns.push(column);
537        }
538        Ok(DataChunk::new(columns, batch.num_rows()))
539    }
540
541    /// Converts Arrow `Fields` to RisingWave `StructType`.
542    fn from_fields(&self, fields: &arrow_schema::Fields) -> Result<StructType, ArrayError> {
543        Ok(StructType::new(
544            fields
545                .iter()
546                .map(|f| Ok((f.name().clone(), self.from_field(f)?)))
547                .try_collect::<_, Vec<_>, ArrayError>()?,
548        ))
549    }
550
551    /// Converts Arrow `Field` to RisingWave `DataType`.
552    fn from_field(&self, field: &arrow_schema::Field) -> Result<DataType, ArrayError> {
553        use arrow_schema::DataType::*;
554        use arrow_schema::IntervalUnit::*;
555        use arrow_schema::TimeUnit::*;
556
557        // extension type
558        if let Some(type_name) = field.metadata().get("ARROW:extension:name") {
559            return self.from_extension_type(type_name, field.data_type());
560        }
561
562        Ok(match field.data_type() {
563            Boolean => DataType::Boolean,
564            Int16 => DataType::Int16,
565            Int32 => DataType::Int32,
566            Int64 => DataType::Int64,
567            Int8 => DataType::Int16,
568            UInt8 => DataType::Int16,
569            UInt16 => DataType::Int32,
570            UInt32 => DataType::Int64,
571            UInt64 => DataType::Decimal,
572            Float16 => DataType::Float32,
573            Float32 => DataType::Float32,
574            Float64 => DataType::Float64,
575            Decimal128(_, _) => DataType::Decimal,
576            Decimal256(_, _) => DataType::Int256,
577            Date32 => DataType::Date,
578            Time32(Second) | Time32(Millisecond) | Time64(Microsecond) | Time64(Nanosecond) => {
579                DataType::Time
580            }
581            Timestamp(Microsecond, None) => DataType::Timestamp,
582            Timestamp(Microsecond, Some(_)) => DataType::Timestamptz,
583            Timestamp(Second, None) => DataType::Timestamp,
584            Timestamp(Second, Some(_)) => DataType::Timestamptz,
585            Timestamp(Millisecond, None) => DataType::Timestamp,
586            Timestamp(Millisecond, Some(_)) => DataType::Timestamptz,
587            Timestamp(Nanosecond, None) => DataType::Timestamp,
588            Timestamp(Nanosecond, Some(_)) => DataType::Timestamptz,
589            Interval(MonthDayNano) => DataType::Interval,
590            Utf8 => DataType::Varchar,
591            Utf8View => DataType::Varchar,
592            Binary => DataType::Bytea,
593            // Iceberg `uuid` maps to `FixedSizeBinary(16)` and `fixed[L]` maps to
594            // `FixedSizeBinary(L)`. Both are represented as `Bytea` in RisingWave.
595            FixedSizeBinary(_) => self.from_fixed_size_binary()?,
596            LargeUtf8 => self.from_large_utf8()?,
597            LargeBinary => self.from_large_binary()?,
598            List(field) => DataType::list(self.from_field(field)?),
599            Struct(fields) => DataType::Struct(self.from_fields(fields)?),
600            Map(field, _is_sorted) => {
601                let entries = self.from_field(field)?;
602                DataType::Map(MapType::try_from_entries(entries).map_err(|e| {
603                    ArrayError::from_arrow(format!("invalid arrow map field: {field:?}, err: {e}"))
604                })?)
605            }
606            t => {
607                return Err(ArrayError::from_arrow(format!(
608                    "unsupported arrow data type: {t:?}"
609                )));
610            }
611        })
612    }
613
614    /// Converts Arrow `LargeUtf8` type to RisingWave data type.
615    fn from_large_utf8(&self) -> Result<DataType, ArrayError> {
616        Ok(DataType::Varchar)
617    }
618
619    /// Converts Arrow `LargeBinary` type to RisingWave data type.
620    fn from_large_binary(&self) -> Result<DataType, ArrayError> {
621        Ok(DataType::Bytea)
622    }
623
624    /// Converts Arrow `FixedSizeBinary` type to RisingWave data type.
625    fn from_fixed_size_binary(&self) -> Result<DataType, ArrayError> {
626        Ok(DataType::Bytea)
627    }
628
629    /// Converts Arrow extension type to RisingWave `DataType`.
630    fn from_extension_type(
631        &self,
632        type_name: &str,
633        physical_type: &arrow_schema::DataType,
634    ) -> Result<DataType, ArrayError> {
635        match (type_name, physical_type) {
636            ("arrowudf.decimal", arrow_schema::DataType::Utf8) => Ok(DataType::Decimal),
637            ("arrowudf.json", arrow_schema::DataType::Utf8) => Ok(DataType::Jsonb),
638            _ => Err(ArrayError::from_arrow(format!(
639                "unsupported extension type: {type_name:?}"
640            ))),
641        }
642    }
643
644    /// Converts Arrow `Array` to RisingWave `ArrayImpl`.
645    ///
646    /// `expected_field` is the declared-side field: it selects the extension decode and
647    /// authoritatively drives the alignment of nested struct/list/map children.
648    fn from_array(
649        &self,
650        expected_field: &arrow_schema::Field,
651        array: &arrow_array::ArrayRef,
652    ) -> Result<ArrayImpl, ArrayError> {
653        use arrow_schema::DataType::*;
654        use arrow_schema::IntervalUnit::*;
655        use arrow_schema::TimeUnit::*;
656
657        // extension type
658        if let Some(type_name) = expected_field.metadata().get("ARROW:extension:name") {
659            return self.from_extension_array(type_name, array);
660        }
661
662        match array.data_type() {
663            Boolean => self.from_bool_array(array.as_any().downcast_ref().unwrap()),
664            Int8 => self.from_int8_array(array.as_any().downcast_ref().unwrap()),
665            Int16 => self.from_int16_array(array.as_any().downcast_ref().unwrap()),
666            Int32 => self.from_int32_array(array.as_any().downcast_ref().unwrap()),
667            Int64 => self.from_int64_array(array.as_any().downcast_ref().unwrap()),
668            UInt8 => self.from_uint8_array(array.as_any().downcast_ref().unwrap()),
669            UInt16 => self.from_uint16_array(array.as_any().downcast_ref().unwrap()),
670            UInt32 => self.from_uint32_array(array.as_any().downcast_ref().unwrap()),
671
672            UInt64 => self.from_uint64_array(array.as_any().downcast_ref().unwrap()),
673            Decimal128(_, _) => self.from_decimal128_array(array.as_any().downcast_ref().unwrap()),
674            Decimal256(_, _) => self.from_int256_array(array.as_any().downcast_ref().unwrap()),
675            Float16 => self.from_float16_array(array.as_any().downcast_ref().unwrap()),
676            Float32 => self.from_float32_array(array.as_any().downcast_ref().unwrap()),
677            Float64 => self.from_float64_array(array.as_any().downcast_ref().unwrap()),
678            Date32 => self.from_date32_array(array.as_any().downcast_ref().unwrap()),
679            Time32(Second) => self.from_time32s_array(array.as_any().downcast_ref().unwrap()),
680            Time32(Millisecond) => self.from_time32ms_array(array.as_any().downcast_ref().unwrap()),
681            Time64(Microsecond) => self.from_time64us_array(array.as_any().downcast_ref().unwrap()),
682            Time64(Nanosecond) => self.from_time64ns_array(array.as_any().downcast_ref().unwrap()),
683            Timestamp(Second, None) => {
684                self.from_timestampsecond_array(array.as_any().downcast_ref().unwrap())
685            }
686            Timestamp(Second, Some(_)) => {
687                self.from_timestampsecond_some_array(array.as_any().downcast_ref().unwrap())
688            }
689            Timestamp(Millisecond, None) => {
690                self.from_timestampms_array(array.as_any().downcast_ref().unwrap())
691            }
692            Timestamp(Millisecond, Some(_)) => {
693                self.from_timestampms_some_array(array.as_any().downcast_ref().unwrap())
694            }
695            Timestamp(Microsecond, None) => {
696                self.from_timestampus_array(array.as_any().downcast_ref().unwrap())
697            }
698            Timestamp(Microsecond, Some(_)) => {
699                self.from_timestampus_some_array(array.as_any().downcast_ref().unwrap())
700            }
701            Timestamp(Nanosecond, None) => {
702                self.from_timestampns_array(array.as_any().downcast_ref().unwrap())
703            }
704            Timestamp(Nanosecond, Some(_)) => {
705                self.from_timestampns_some_array(array.as_any().downcast_ref().unwrap())
706            }
707            Interval(MonthDayNano) => {
708                self.from_interval_array(array.as_any().downcast_ref().unwrap())
709            }
710            Utf8 => self.from_utf8_array(array.as_any().downcast_ref().unwrap()),
711            Utf8View => self.from_utf8_view_array(array.as_any().downcast_ref().unwrap()),
712            Binary => self.from_binary_array(array.as_any().downcast_ref().unwrap()),
713            FixedSizeBinary(_) => {
714                self.from_fixed_size_binary_array(array.as_any().downcast_ref().unwrap())
715            }
716            LargeUtf8 => self.from_large_utf8_array(array.as_any().downcast_ref().unwrap()),
717            LargeBinary => self.from_large_binary_array(array.as_any().downcast_ref().unwrap()),
718            List(_) => self.from_list_array(expected_field, array.as_any().downcast_ref().unwrap()),
719            Struct(_) => {
720                self.from_struct_array(expected_field, array.as_any().downcast_ref().unwrap())
721            }
722            Map(_, _) => {
723                self.from_map_array(expected_field, array.as_any().downcast_ref().unwrap())
724            }
725            t => Err(ArrayError::from_arrow(format!(
726                "unsupported arrow data type: {t:?}",
727            ))),
728        }
729    }
730
731    /// Converts Arrow extension array to RisingWave `ArrayImpl`.
732    fn from_extension_array(
733        &self,
734        type_name: &str,
735        array: &arrow_array::ArrayRef,
736    ) -> Result<ArrayImpl, ArrayError> {
737        match type_name {
738            "arrowudf.decimal" => {
739                let array: &arrow_array::StringArray =
740                    array.as_any().downcast_ref().ok_or_else(|| {
741                        ArrayError::from_arrow(
742                            "expected string array for `arrowudf.decimal`".to_owned(),
743                        )
744                    })?;
745                Ok(ArrayImpl::Decimal(array.try_into()?))
746            }
747            "arrowudf.json" => {
748                let array: &arrow_array::StringArray =
749                    array.as_any().downcast_ref().ok_or_else(|| {
750                        ArrayError::from_arrow(
751                            "expected string array for `arrowudf.json`".to_owned(),
752                        )
753                    })?;
754                Ok(ArrayImpl::Jsonb(array.try_into()?))
755            }
756            _ => Err(ArrayError::from_arrow(format!(
757                "unsupported extension type: {type_name:?}"
758            ))),
759        }
760    }
761
762    fn from_bool_array(&self, array: &arrow_array::BooleanArray) -> Result<ArrayImpl, ArrayError> {
763        Ok(ArrayImpl::Bool(array.into()))
764    }
765
766    fn from_int16_array(&self, array: &arrow_array::Int16Array) -> Result<ArrayImpl, ArrayError> {
767        Ok(ArrayImpl::Int16(array.into()))
768    }
769
770    fn from_int8_array(&self, array: &arrow_array::Int8Array) -> Result<ArrayImpl, ArrayError> {
771        Ok(ArrayImpl::Int16(array.into()))
772    }
773
774    fn from_uint8_array(&self, array: &arrow_array::UInt8Array) -> Result<ArrayImpl, ArrayError> {
775        Ok(ArrayImpl::Int16(array.into()))
776    }
777
778    fn from_uint16_array(&self, array: &arrow_array::UInt16Array) -> Result<ArrayImpl, ArrayError> {
779        Ok(ArrayImpl::Int32(array.into()))
780    }
781
782    fn from_uint32_array(&self, array: &arrow_array::UInt32Array) -> Result<ArrayImpl, ArrayError> {
783        Ok(ArrayImpl::Int64(array.into()))
784    }
785
786    fn from_int32_array(&self, array: &arrow_array::Int32Array) -> Result<ArrayImpl, ArrayError> {
787        Ok(ArrayImpl::Int32(array.into()))
788    }
789
790    fn from_int64_array(&self, array: &arrow_array::Int64Array) -> Result<ArrayImpl, ArrayError> {
791        Ok(ArrayImpl::Int64(array.into()))
792    }
793
794    fn from_int256_array(
795        &self,
796        array: &arrow_array::Decimal256Array,
797    ) -> Result<ArrayImpl, ArrayError> {
798        Ok(ArrayImpl::Int256(array.into()))
799    }
800
801    fn from_decimal128_array(
802        &self,
803        array: &arrow_array::Decimal128Array,
804    ) -> Result<ArrayImpl, ArrayError> {
805        Ok(ArrayImpl::Decimal(array.try_into()?))
806    }
807
808    fn from_uint64_array(&self, array: &arrow_array::UInt64Array) -> Result<ArrayImpl, ArrayError> {
809        Ok(ArrayImpl::Decimal(array.try_into()?))
810    }
811
812    fn from_float16_array(
813        &self,
814        array: &arrow_array::Float16Array,
815    ) -> Result<ArrayImpl, ArrayError> {
816        Ok(ArrayImpl::Float32(array.try_into()?))
817    }
818
819    fn from_float32_array(
820        &self,
821        array: &arrow_array::Float32Array,
822    ) -> Result<ArrayImpl, ArrayError> {
823        Ok(ArrayImpl::Float32(array.into()))
824    }
825
826    fn from_float64_array(
827        &self,
828        array: &arrow_array::Float64Array,
829    ) -> Result<ArrayImpl, ArrayError> {
830        Ok(ArrayImpl::Float64(array.into()))
831    }
832
833    fn from_date32_array(&self, array: &arrow_array::Date32Array) -> Result<ArrayImpl, ArrayError> {
834        Ok(ArrayImpl::Date(array.try_into()?))
835    }
836
837    fn from_time32s_array(
838        &self,
839        array: &arrow_array::Time32SecondArray,
840    ) -> Result<ArrayImpl, ArrayError> {
841        Ok(ArrayImpl::Time(array.try_into()?))
842    }
843
844    fn from_time32ms_array(
845        &self,
846        array: &arrow_array::Time32MillisecondArray,
847    ) -> Result<ArrayImpl, ArrayError> {
848        Ok(ArrayImpl::Time(array.try_into()?))
849    }
850
851    fn from_time64us_array(
852        &self,
853        array: &arrow_array::Time64MicrosecondArray,
854    ) -> Result<ArrayImpl, ArrayError> {
855        Ok(ArrayImpl::Time(array.try_into()?))
856    }
857
858    fn from_time64ns_array(
859        &self,
860        array: &arrow_array::Time64NanosecondArray,
861    ) -> Result<ArrayImpl, ArrayError> {
862        Ok(ArrayImpl::Time(array.try_into()?))
863    }
864
865    fn from_timestampsecond_array(
866        &self,
867        array: &arrow_array::TimestampSecondArray,
868    ) -> Result<ArrayImpl, ArrayError> {
869        Ok(ArrayImpl::Timestamp(array.try_into()?))
870    }
871    fn from_timestampsecond_some_array(
872        &self,
873        array: &arrow_array::TimestampSecondArray,
874    ) -> Result<ArrayImpl, ArrayError> {
875        Ok(ArrayImpl::Timestamptz(array.try_into()?))
876    }
877
878    fn from_timestampms_array(
879        &self,
880        array: &arrow_array::TimestampMillisecondArray,
881    ) -> Result<ArrayImpl, ArrayError> {
882        Ok(ArrayImpl::Timestamp(array.try_into()?))
883    }
884
885    fn from_timestampms_some_array(
886        &self,
887        array: &arrow_array::TimestampMillisecondArray,
888    ) -> Result<ArrayImpl, ArrayError> {
889        Ok(ArrayImpl::Timestamptz(array.try_into()?))
890    }
891
892    fn from_timestampus_array(
893        &self,
894        array: &arrow_array::TimestampMicrosecondArray,
895    ) -> Result<ArrayImpl, ArrayError> {
896        Ok(ArrayImpl::Timestamp(array.try_into()?))
897    }
898
899    fn from_timestampus_some_array(
900        &self,
901        array: &arrow_array::TimestampMicrosecondArray,
902    ) -> Result<ArrayImpl, ArrayError> {
903        Ok(ArrayImpl::Timestamptz(array.try_into()?))
904    }
905
906    fn from_timestampns_array(
907        &self,
908        array: &arrow_array::TimestampNanosecondArray,
909    ) -> Result<ArrayImpl, ArrayError> {
910        Ok(ArrayImpl::Timestamp(array.try_into()?))
911    }
912
913    fn from_timestampns_some_array(
914        &self,
915        array: &arrow_array::TimestampNanosecondArray,
916    ) -> Result<ArrayImpl, ArrayError> {
917        Ok(ArrayImpl::Timestamptz(array.try_into()?))
918    }
919
920    fn from_interval_array(
921        &self,
922        array: &arrow_array::IntervalMonthDayNanoArray,
923    ) -> Result<ArrayImpl, ArrayError> {
924        Ok(ArrayImpl::Interval(array.try_into()?))
925    }
926
927    fn from_utf8_array(&self, array: &arrow_array::StringArray) -> Result<ArrayImpl, ArrayError> {
928        Ok(ArrayImpl::Utf8(array.into()))
929    }
930
931    fn from_utf8_view_array(
932        &self,
933        array: &arrow_array::StringViewArray,
934    ) -> Result<ArrayImpl, ArrayError> {
935        Ok(ArrayImpl::Utf8(array.into()))
936    }
937
938    fn from_binary_array(&self, array: &arrow_array::BinaryArray) -> Result<ArrayImpl, ArrayError> {
939        Ok(ArrayImpl::Bytea(array.into()))
940    }
941
942    fn from_large_utf8_array(
943        &self,
944        array: &arrow_array::LargeStringArray,
945    ) -> Result<ArrayImpl, ArrayError> {
946        Ok(ArrayImpl::Utf8(array.into()))
947    }
948
949    fn from_large_binary_array(
950        &self,
951        array: &arrow_array::LargeBinaryArray,
952    ) -> Result<ArrayImpl, ArrayError> {
953        Ok(ArrayImpl::Bytea(array.into()))
954    }
955
956    fn from_fixed_size_binary_array(
957        &self,
958        array: &arrow_array::FixedSizeBinaryArray,
959    ) -> Result<ArrayImpl, ArrayError> {
960        Ok(ArrayImpl::Bytea(array.iter().collect()))
961    }
962
963    /// Converts an Arrow `ListArray`, decoding elements by the expected element field so that
964    /// nested decode keeps following the declared schema. Falls back to the array's own element
965    /// field when the expected field does not describe a list.
966    fn from_list_array(
967        &self,
968        expected_field: &arrow_schema::Field,
969        array: &arrow_array::ListArray,
970    ) -> Result<ArrayImpl, ArrayError> {
971        use arrow_array::Array;
972        let elem_field = match (expected_field.data_type(), array.data_type()) {
973            (arrow_schema::DataType::List(elem), _) | (_, arrow_schema::DataType::List(elem)) => {
974                elem
975            }
976            _ => unreachable!("a list array must have a list data type"),
977        };
978        Ok(ArrayImpl::List(ListArray {
979            value: Box::new(self.from_array(elem_field, array.values())?),
980            bitmap: match array.nulls() {
981                Some(nulls) => nulls.iter().collect(),
982                None => Bitmap::ones(array.len()),
983            },
984            offsets: array.offsets().iter().map(|o| *o as u32).collect(),
985        }))
986    }
987
988    /// Converts an Arrow `StructArray` by the expected struct field: children align by name
989    /// (first occurrence, extras dropped), positionally when a name is missing but the arity
990    /// matches (an external UDF may label struct children differently, as its signature check
991    /// ignores nested field names), and error otherwise. Extensions are taken from the
992    /// expected side only.
993    ///
994    /// The result carries the expected field *names* with the *decoded* child types: the
995    /// expected field may be a lossy rendering of the declared type (e.g. iceberg has no
996    /// 16-bit int), so the decoded types are authoritative and any divergence from the
997    /// declared type stays visible to the callers' boundary checks.
998    ///
999    /// Falls back to the array's own fields when the expected field does not describe a struct.
1000    fn from_struct_array(
1001        &self,
1002        expected_field: &arrow_schema::Field,
1003        array: &arrow_array::StructArray,
1004    ) -> Result<ArrayImpl, ArrayError> {
1005        use std::collections::HashMap;
1006
1007        use arrow_array::Array;
1008
1009        let arrow_schema::DataType::Struct(actual_fields) = array.data_type() else {
1010            unreachable!("a struct array must have a struct data type");
1011        };
1012        let expected_fields = match expected_field.data_type() {
1013            arrow_schema::DataType::Struct(fields) => fields,
1014            _ => actual_fields,
1015        };
1016
1017        let len = array.len();
1018        let decode_positionally = |fields: &arrow_schema::Fields| {
1019            array
1020                .columns()
1021                .iter()
1022                .zip_eq_fast(fields)
1023                .map(|(column, field)| self.from_array(field, column).map(Arc::new))
1024                .try_collect()
1025        };
1026        // Children decode by the expected field either way, so aligning names in order is
1027        // enough for the zip fast path — no deep field comparison needed.
1028        let names_aligned = expected_fields.len() == actual_fields.len()
1029            && expected_fields
1030                .iter()
1031                .zip_eq_fast(actual_fields.iter())
1032                .all(|(e, a)| e.name() == a.name());
1033        let columns: Vec<Arc<ArrayImpl>> = if names_aligned {
1034            decode_positionally(expected_fields)?
1035        } else {
1036            // First occurrence wins. The schema matcher rejects duplicate sibling names,
1037            // so on the parquet path this tie-break is never exercised.
1038            let mut actual_name_to_index = HashMap::new();
1039            for (idx, f) in actual_fields.iter().enumerate() {
1040                actual_name_to_index.entry(f.name().as_str()).or_insert(idx);
1041            }
1042            if expected_fields
1043                .iter()
1044                .all(|f| actual_name_to_index.contains_key(f.name().as_str()))
1045            {
1046                expected_fields
1047                    .iter()
1048                    .map(|expected_field| {
1049                        let idx = actual_name_to_index[expected_field.name().as_str()];
1050                        self.from_array(expected_field, &array.columns()[idx])
1051                            .map(Arc::new)
1052                    })
1053                    .try_collect()?
1054            } else if expected_fields.len() == actual_fields.len() {
1055                // Positional fallback for external UDFs. Unreachable on the parquet path:
1056                // the schema matcher requires every declared name to be present.
1057                decode_positionally(expected_fields)?
1058            } else {
1059                let names =
1060                    |fields: &arrow_schema::Fields| fields.iter().map(|f| f.name()).join(", ");
1061                return Err(ArrayError::from_arrow(format!(
1062                    "unable to align struct fields: expected [{}], actual [{}]",
1063                    names(expected_fields),
1064                    names(actual_fields),
1065                )));
1066            }
1067        };
1068
1069        Ok(ArrayImpl::Struct(StructArray::new(
1070            StructType::new(
1071                expected_fields
1072                    .iter()
1073                    .zip_eq_fast(columns.iter())
1074                    .map(|(f, c)| (f.name().clone(), c.data_type())),
1075            ),
1076            columns,
1077            (0..len).map(|i| array.is_valid(i)).collect(),
1078        )))
1079    }
1080
1081    /// Converts an Arrow `MapArray`, decoding entries by the expected entries field. Falls back
1082    /// to the array's own entries field when the expected field does not describe a map.
1083    fn from_map_array(
1084        &self,
1085        expected_field: &arrow_schema::Field,
1086        array: &arrow_array::MapArray,
1087    ) -> Result<ArrayImpl, ArrayError> {
1088        use arrow_array::Array;
1089        let expected_entries = match (expected_field.data_type(), array.data_type()) {
1090            (arrow_schema::DataType::Map(entries, _), _)
1091            | (_, arrow_schema::DataType::Map(entries, _)) => entries,
1092            _ => unreachable!("a map array must have a map data type"),
1093        };
1094        let struct_array = self.from_struct_array(expected_entries, array.entries())?;
1095        // RW restricts map key types, so deriving a map type from decoded entries is
1096        // fallible and `MapArray::data_type()` panics on an invalid key. Reject it here.
1097        MapType::try_from_entries(struct_array.data_type())
1098            .map_err(|e| ArrayError::from_arrow(format!("invalid arrow map array: {e}")))?;
1099        let list_array = ListArray {
1100            value: Box::new(struct_array),
1101            bitmap: match array.nulls() {
1102                Some(nulls) => nulls.iter().collect(),
1103                None => Bitmap::ones(array.len()),
1104            },
1105            offsets: array.offsets().iter().map(|o| *o as u32).collect(),
1106        };
1107
1108        Ok(ArrayImpl::Map(MapArray { inner: list_array }))
1109    }
1110}
1111
1112impl From<&Bitmap> for arrow_buffer::NullBuffer {
1113    fn from(bitmap: &Bitmap) -> Self {
1114        bitmap.iter().collect()
1115    }
1116}
1117
1118/// Implement bi-directional `From` between concrete array types.
1119macro_rules! converts {
1120    ($ArrayType:ty, $ArrowType:ty) => {
1121        impl From<&$ArrayType> for $ArrowType {
1122            fn from(array: &$ArrayType) -> Self {
1123                array.iter().collect()
1124            }
1125        }
1126        impl From<&$ArrowType> for $ArrayType {
1127            fn from(array: &$ArrowType) -> Self {
1128                array.iter().collect()
1129            }
1130        }
1131        impl From<&[$ArrowType]> for $ArrayType {
1132            fn from(arrays: &[$ArrowType]) -> Self {
1133                arrays.iter().flat_map(|a| a.iter()).collect()
1134            }
1135        }
1136    };
1137    // convert values using FromIntoArrow
1138    ($ArrayType:ty, $ArrowType:ty, @map) => {
1139        impl From<&$ArrayType> for $ArrowType {
1140            fn from(array: &$ArrayType) -> Self {
1141                array.iter().map(|o| o.map(|v| v.into_arrow())).collect()
1142            }
1143        }
1144        impl From<&$ArrowType> for $ArrayType {
1145            fn from(array: &$ArrowType) -> Self {
1146                array
1147                    .iter()
1148                    .map(|o| {
1149                        o.map(|v| {
1150                            <<$ArrayType as Array>::RefItem<'_> as FromIntoArrow>::from_arrow(v)
1151                        })
1152                    })
1153                    .collect()
1154            }
1155        }
1156        impl From<&[$ArrowType]> for $ArrayType {
1157            fn from(arrays: &[$ArrowType]) -> Self {
1158                arrays
1159                    .iter()
1160                    .flat_map(|a| a.iter())
1161                    .map(|o| {
1162                        o.map(|v| {
1163                            <<$ArrayType as Array>::RefItem<'_> as FromIntoArrow>::from_arrow(v)
1164                        })
1165                    })
1166                    .collect()
1167            }
1168        }
1169    };
1170    // convert values using TryFromIntoArrow
1171    ($ArrayType:ty, $ArrowType:ty, @try_map) => {
1172        impl TryFrom<&$ArrayType> for $ArrowType {
1173            type Error = ArrayError;
1174
1175            fn try_from(array: &$ArrayType) -> Result<Self, Self::Error> {
1176                // Collecting `Result`s loses the iterator's size hint, so build with an
1177                // explicit capacity instead.
1178                let mut builder = <$ArrowType>::builder(array.len());
1179                for o in array.iter() {
1180                    builder.append_option(o.map(|v| v.try_into_arrow()).transpose()?);
1181                }
1182                Ok(builder.finish())
1183            }
1184        }
1185        impl TryFrom<&$ArrowType> for $ArrayType {
1186            type Error = ArrayError;
1187
1188            fn try_from(array: &$ArrowType) -> Result<Self, Self::Error> {
1189                use arrow_array::Array as _;
1190
1191                let mut builder = <$ArrayType as Array>::Builder::new(array.len());
1192                for o in array.iter() {
1193                    builder.append(
1194                        o.map(<<$ArrayType as Array>::RefItem<'_> as TryFromIntoArrow>::try_from_arrow)
1195                            .transpose()?,
1196                    );
1197                }
1198                Ok(builder.finish())
1199            }
1200        }
1201    };
1202}
1203
1204/// Used to convert different types.
1205macro_rules! converts_with_type {
1206    ($ArrayType:ty, $ArrowType:ty, $FromType:ty, $ToType:ty) => {
1207        impl From<&$ArrayType> for $ArrowType {
1208            fn from(array: &$ArrayType) -> Self {
1209                let values: Vec<Option<$ToType>> =
1210                    array.iter().map(|x| x.map(|v| v as $ToType)).collect();
1211                <$ArrowType>::from_iter(values)
1212            }
1213        }
1214
1215        impl From<&$ArrowType> for $ArrayType {
1216            fn from(array: &$ArrowType) -> Self {
1217                let values: Vec<Option<$FromType>> =
1218                    array.iter().map(|x| x.map(|v| v as $FromType)).collect();
1219                <$ArrayType>::from_iter(values)
1220            }
1221        }
1222
1223        impl From<&[$ArrowType]> for $ArrayType {
1224            fn from(arrays: &[$ArrowType]) -> Self {
1225                let values: Vec<Option<$FromType>> = arrays
1226                    .iter()
1227                    .flat_map(|a| a.iter().map(|x| x.map(|v| v as $FromType)))
1228                    .collect();
1229                <$ArrayType>::from_iter(values)
1230            }
1231        }
1232    };
1233}
1234
1235macro_rules! converts_with_time_unit {
1236    ($ArrayType:ty, $ArrowType:ident, $time_unit:ident) => {
1237        impl TryFrom<&$ArrayType> for arrow_array::$ArrowType {
1238            type Error = ArrayError;
1239
1240            fn try_from(array: &$ArrayType) -> Result<Self, Self::Error> {
1241                let mut builder = arrow_array::$ArrowType::builder(array.len());
1242                for o in array.iter() {
1243                    builder.append_option(
1244                        o.map(|v| into_arrow_temporal_value(v, TimeUnit::$time_unit))
1245                            .transpose()?,
1246                    );
1247                }
1248                Ok(builder.finish())
1249            }
1250        }
1251
1252        impl TryFrom<&arrow_array::$ArrowType> for $ArrayType {
1253            type Error = ArrayError;
1254
1255            fn try_from(array: &arrow_array::$ArrowType) -> Result<Self, Self::Error> {
1256                Self::try_from(std::slice::from_ref(array))
1257            }
1258        }
1259
1260        impl TryFrom<&[arrow_array::$ArrowType]> for $ArrayType {
1261            type Error = ArrayError;
1262
1263            fn try_from(arrays: &[arrow_array::$ArrowType]) -> Result<Self, Self::Error> {
1264                use arrow_array::Array as _;
1265
1266                let mut builder =
1267                    <$ArrayType as Array>::Builder::new(arrays.iter().map(|a| a.len()).sum());
1268                for o in arrays.iter().flat_map(|a| a.iter()) {
1269                    builder.append(
1270                        o.map(|v| try_from_arrow_temporal_value(v, TimeUnit::$time_unit))
1271                            .transpose()?,
1272                    );
1273                }
1274                Ok(builder.finish())
1275            }
1276        }
1277    };
1278}
1279
1280converts!(BoolArray, arrow_array::BooleanArray);
1281converts!(I16Array, arrow_array::Int16Array);
1282converts!(I32Array, arrow_array::Int32Array);
1283converts!(I64Array, arrow_array::Int64Array);
1284converts!(F32Array, arrow_array::Float32Array, @map);
1285converts!(F64Array, arrow_array::Float64Array, @map);
1286converts!(BytesArray, arrow_array::BinaryArray);
1287converts!(BytesArray, arrow_array::LargeBinaryArray);
1288converts!(Utf8Array, arrow_array::StringArray);
1289converts!(Utf8Array, arrow_array::LargeStringArray);
1290converts!(Utf8Array, arrow_array::StringViewArray);
1291converts!(DateArray, arrow_array::Date32Array, @try_map);
1292converts!(IntervalArray, arrow_array::IntervalMonthDayNanoArray, @try_map);
1293converts!(SerialArray, arrow_array::Int64Array, @map);
1294
1295converts_with_type!(I16Array, arrow_array::Int8Array, i16, i8);
1296converts_with_type!(I16Array, arrow_array::UInt8Array, i16, u8);
1297converts_with_type!(I32Array, arrow_array::UInt16Array, i32, u16);
1298converts_with_type!(I64Array, arrow_array::UInt32Array, i64, u32);
1299
1300converts_with_time_unit!(TimeArray, Time32SecondArray, Second);
1301converts_with_time_unit!(TimeArray, Time32MillisecondArray, Millisecond);
1302converts_with_time_unit!(TimeArray, Time64MicrosecondArray, Microsecond);
1303converts_with_time_unit!(TimeArray, Time64NanosecondArray, Nanosecond);
1304
1305converts_with_time_unit!(TimestampArray, TimestampSecondArray, Second);
1306converts_with_time_unit!(TimestampArray, TimestampMillisecondArray, Millisecond);
1307converts_with_time_unit!(TimestampArray, TimestampMicrosecondArray, Microsecond);
1308converts_with_time_unit!(TimestampArray, TimestampNanosecondArray, Nanosecond);
1309
1310converts_with_time_unit!(TimestamptzArray, TimestampSecondArray, Second);
1311converts_with_time_unit!(TimestamptzArray, TimestampMillisecondArray, Millisecond);
1312converts_with_time_unit!(TimestamptzArray, TimestampMicrosecondArray, Microsecond);
1313converts_with_time_unit!(TimestamptzArray, TimestampNanosecondArray, Nanosecond);
1314
1315/// Converts RisingWave value from and into Arrow value.
1316trait FromIntoArrow {
1317    /// The corresponding element type in the Arrow array.
1318    type ArrowType;
1319    fn from_arrow(value: Self::ArrowType) -> Self;
1320    fn into_arrow(self) -> Self::ArrowType;
1321}
1322
1323/// Like [`FromIntoArrow`], for values whose Arrow representation does not cover the whole
1324/// RisingWave domain, or vice versa.
1325trait TryFromIntoArrow: Sized {
1326    /// The corresponding element type in the Arrow array.
1327    type ArrowType;
1328    fn try_from_arrow(value: Self::ArrowType) -> Result<Self, ArrayError>;
1329    fn try_into_arrow(self) -> Result<Self::ArrowType, ArrayError>;
1330}
1331
1332/// Converts a RisingWave temporal scalar to and from Arrow's physical primitive
1333/// value using an Arrow time unit.
1334trait TemporalArrowConvert<ArrowNative>: Sized {
1335    fn try_from_arrow_with_time_unit(
1336        value: ArrowNative,
1337        time_unit: TimeUnit,
1338    ) -> Result<Self, ArrayError>;
1339    fn into_arrow_with_time_unit(self, time_unit: TimeUnit) -> Result<ArrowNative, ArrayError>;
1340}
1341
1342fn try_from_arrow_temporal_value<Value, ArrowNative>(
1343    value: ArrowNative,
1344    time_unit: TimeUnit,
1345) -> Result<Value, ArrayError>
1346where
1347    Value: TemporalArrowConvert<ArrowNative>,
1348{
1349    Value::try_from_arrow_with_time_unit(value, time_unit)
1350}
1351
1352fn into_arrow_temporal_value<Value, ArrowNative>(
1353    value: Value,
1354    time_unit: TimeUnit,
1355) -> Result<ArrowNative, ArrayError>
1356where
1357    Value: TemporalArrowConvert<ArrowNative>,
1358{
1359    value.into_arrow_with_time_unit(time_unit)
1360}
1361
1362impl FromIntoArrow for Serial {
1363    type ArrowType = i64;
1364
1365    fn from_arrow(value: Self::ArrowType) -> Self {
1366        value.into()
1367    }
1368
1369    fn into_arrow(self) -> Self::ArrowType {
1370        self.into()
1371    }
1372}
1373
1374impl FromIntoArrow for F32 {
1375    type ArrowType = f32;
1376
1377    fn from_arrow(value: Self::ArrowType) -> Self {
1378        value.into()
1379    }
1380
1381    fn into_arrow(self) -> Self::ArrowType {
1382        self.into()
1383    }
1384}
1385
1386impl FromIntoArrow for F64 {
1387    type ArrowType = f64;
1388
1389    fn from_arrow(value: Self::ArrowType) -> Self {
1390        value.into()
1391    }
1392
1393    fn into_arrow(self) -> Self::ArrowType {
1394        self.into()
1395    }
1396}
1397
1398impl TryFromIntoArrow for Date {
1399    type ArrowType = i32;
1400
1401    #[allow(deprecated)]
1402    fn try_from_arrow(value: Self::ArrowType) -> Result<Self, ArrayError> {
1403        arrow_array::types::Date32Type::to_naive_date_opt(value)
1404            .map(Date)
1405            .ok_or_else(|| ArrayError::from_arrow(format!("invalid Arrow date {value}")))
1406    }
1407
1408    fn try_into_arrow(self) -> Result<Self::ArrowType, ArrayError> {
1409        Ok(arrow_array::types::Date32Type::from_naive_date(self.0))
1410    }
1411}
1412
1413/// Arrow `Time32` arrays are defined only for second and millisecond units.
1414impl TemporalArrowConvert<i32> for Time {
1415    fn try_from_arrow_with_time_unit(value: i32, time_unit: TimeUnit) -> Result<Self, ArrayError> {
1416        u32::try_from(value)
1417            .ok()
1418            .and_then(|v| match time_unit {
1419                TimeUnit::Second => Time::with_secs_nano(v, 0).ok(),
1420                TimeUnit::Millisecond => Time::with_milli(v).ok(),
1421                unit => unreachable!("{unit:?} is not a Time32 unit"),
1422            })
1423            .ok_or_else(|| invalid_arrow_temporal_value(value, time_unit))
1424    }
1425
1426    fn into_arrow_with_time_unit(self, time_unit: TimeUnit) -> Result<i32, ArrayError> {
1427        Ok(match time_unit {
1428            TimeUnit::Second => self.0.num_seconds_from_midnight() as i32,
1429            TimeUnit::Millisecond => {
1430                (self.0.num_seconds_from_midnight() * 1_000 + self.0.nanosecond() / 1_000_000)
1431                    as i32
1432            }
1433            unit => unreachable!("{unit:?} is not a Time32 unit"),
1434        })
1435    }
1436}
1437
1438/// Arrow `Time64` arrays are defined only for microsecond and nanosecond units.
1439///
1440/// RisingWave's `time` is microsecond-precision, so nanoseconds are truncated.
1441impl TemporalArrowConvert<i64> for Time {
1442    fn try_from_arrow_with_time_unit(value: i64, time_unit: TimeUnit) -> Result<Self, ArrayError> {
1443        u64::try_from(value)
1444            .ok()
1445            .and_then(|v| match time_unit {
1446                TimeUnit::Microsecond => Time::with_micro(v).ok(),
1447                TimeUnit::Nanosecond => Time::with_micro(v / 1_000).ok(),
1448                unit => unreachable!("{unit:?} is not a Time64 unit"),
1449            })
1450            .ok_or_else(|| invalid_arrow_temporal_value(value, time_unit))
1451    }
1452
1453    fn into_arrow_with_time_unit(self, time_unit: TimeUnit) -> Result<i64, ArrayError> {
1454        Ok(match time_unit {
1455            TimeUnit::Microsecond => self.micros_of_day() as i64,
1456            TimeUnit::Nanosecond => self.nanos_of_day() as i64,
1457            unit => unreachable!("{unit:?} is not a Time64 unit"),
1458        })
1459    }
1460}
1461
1462impl TemporalArrowConvert<i64> for Timestamp {
1463    fn try_from_arrow_with_time_unit(value: i64, time_unit: TimeUnit) -> Result<Self, ArrayError> {
1464        match time_unit {
1465            TimeUnit::Second => Timestamp::with_secs_nsecs(value, 0)
1466                .map_err(|_| invalid_arrow_temporal_value(value, time_unit)),
1467            TimeUnit::Millisecond => Timestamp::with_millis(value)
1468                .map_err(|_| invalid_arrow_temporal_value(value, time_unit)),
1469            TimeUnit::Microsecond => Timestamp::with_micros(value)
1470                .map_err(|_| invalid_arrow_temporal_value(value, time_unit)),
1471            TimeUnit::Nanosecond => Ok(Timestamp::with_nanos(value)),
1472        }
1473    }
1474
1475    fn into_arrow_with_time_unit(self, time_unit: TimeUnit) -> Result<i64, ArrayError> {
1476        match time_unit {
1477            TimeUnit::Second => Ok(self.0.and_utc().timestamp()),
1478            TimeUnit::Millisecond => Ok(self.0.and_utc().timestamp_millis()),
1479            TimeUnit::Microsecond => Ok(self.0.and_utc().timestamp_micros()),
1480            TimeUnit::Nanosecond => self
1481                .0
1482                .and_utc()
1483                .timestamp_nanos_opt()
1484                .ok_or_else(|| arrow_temporal_value_overflow(self, time_unit)),
1485        }
1486    }
1487}
1488
1489impl TemporalArrowConvert<i64> for Timestamptz {
1490    fn try_from_arrow_with_time_unit(value: i64, time_unit: TimeUnit) -> Result<Self, ArrayError> {
1491        match time_unit {
1492            TimeUnit::Second => Timestamptz::from_secs(value)
1493                .ok_or_else(|| invalid_arrow_temporal_value(value, time_unit)),
1494            TimeUnit::Millisecond => Timestamptz::from_millis(value)
1495                .ok_or_else(|| invalid_arrow_temporal_value(value, time_unit)),
1496            TimeUnit::Microsecond => Timestamptz::from_micros(value)
1497                .ok_or_else(|| invalid_arrow_temporal_value(value, time_unit)),
1498            TimeUnit::Nanosecond => Ok(Timestamptz::from_nanos(value)),
1499        }
1500    }
1501
1502    fn into_arrow_with_time_unit(self, time_unit: TimeUnit) -> Result<i64, ArrayError> {
1503        match time_unit {
1504            TimeUnit::Second => Ok(self.timestamp()),
1505            TimeUnit::Millisecond => Ok(self.timestamp_millis()),
1506            TimeUnit::Microsecond => Ok(self.timestamp_micros()),
1507            TimeUnit::Nanosecond => self
1508                .timestamp_nanos()
1509                .ok_or_else(|| arrow_temporal_value_overflow(self, time_unit)),
1510        }
1511    }
1512}
1513
1514fn invalid_arrow_temporal_value<T: std::fmt::Display>(value: T, time_unit: TimeUnit) -> ArrayError {
1515    ArrayError::from_arrow(format!(
1516        "invalid Arrow temporal value {value} for unit {time_unit:?}"
1517    ))
1518}
1519
1520fn arrow_temporal_value_overflow<T: std::fmt::Display>(
1521    value: T,
1522    time_unit: TimeUnit,
1523) -> ArrayError {
1524    ArrayError::to_arrow(format!(
1525        "temporal value {value} overflows Arrow unit {time_unit:?}"
1526    ))
1527}
1528
1529impl TryFromIntoArrow for Interval {
1530    type ArrowType = ArrowIntervalType;
1531
1532    /// RisingWave's `interval` is microsecond-precision, so nanoseconds are truncated.
1533    fn try_from_arrow(value: Self::ArrowType) -> Result<Self, ArrayError> {
1534        Ok(Interval::from_month_day_usec(
1535            value.months,
1536            value.days,
1537            value.nanoseconds / 1_000,
1538        ))
1539    }
1540
1541    fn try_into_arrow(self) -> Result<Self::ArrowType, ArrayError> {
1542        Ok(ArrowIntervalType {
1543            months: self.months(),
1544            days: self.days(),
1545            nanoseconds: self.usecs().checked_mul(1_000).ok_or_else(|| {
1546                ArrayError::to_arrow(format!(
1547                    "interval with {} microseconds is out of range for Arrow",
1548                    self.usecs()
1549                ))
1550            })?,
1551        })
1552    }
1553}
1554
1555impl From<&DecimalArray> for arrow_array::LargeBinaryArray {
1556    fn from(array: &DecimalArray) -> Self {
1557        let mut builder =
1558            arrow_array::builder::LargeBinaryBuilder::with_capacity(array.len(), array.len() * 8);
1559        for value in array.iter() {
1560            builder.append_option(value.map(|d| d.to_string()));
1561        }
1562        builder.finish()
1563    }
1564}
1565
1566impl From<&DecimalArray> for arrow_array::StringArray {
1567    fn from(array: &DecimalArray) -> Self {
1568        let mut builder =
1569            arrow_array::builder::StringBuilder::with_capacity(array.len(), array.len() * 8);
1570        for value in array.iter() {
1571            builder.append_option(value.map(|d| d.to_string()));
1572        }
1573        builder.finish()
1574    }
1575}
1576
1577// This arrow decimal type is used by iceberg source to read iceberg decimal into RW decimal.
1578impl TryFrom<&arrow_array::Decimal128Array> for DecimalArray {
1579    type Error = ArrayError;
1580
1581    fn try_from(array: &arrow_array::Decimal128Array) -> Result<Self, Self::Error> {
1582        if array.scale() < 0 {
1583            bail!("support negative scale for arrow decimal")
1584        }
1585
1586        // Calculate the max value based on the Arrow decimal's precision
1587        // When writing Inf to Arrow Decimal128(precision, scale), we use 10^precision - 1
1588        let precision = array.precision();
1589        let max_value = 10_i128.pow(precision as u32) - 1;
1590
1591        let from_arrow = |value| {
1592            const NAN: i128 = i128::MIN + 1;
1593            let res = match value {
1594                // Check for special values using Arrow Decimal's max value, not i128::MAX
1595                NAN => Decimal::NaN,
1596                v if v == max_value => Decimal::PositiveInf,
1597                v if v == -max_value => Decimal::NegativeInf,
1598                i128::MAX => Decimal::PositiveInf, // Fallback for old data
1599                i128::MIN => Decimal::NegativeInf, // Fallback for old data
1600                _ => Decimal::truncated_i128_and_scale(value, array.scale() as u32)
1601                    .ok_or_else(|| ArrayError::from_arrow("decimal overflow"))?,
1602            };
1603            Ok(res)
1604        };
1605        array
1606            .iter()
1607            .map(|o| o.map(from_arrow).transpose())
1608            .collect::<Result<Self, Self::Error>>()
1609    }
1610}
1611
1612// Since RisingWave does not support UInt type, convert UInt64Array to Decimal.
1613impl TryFrom<&arrow_array::UInt64Array> for DecimalArray {
1614    type Error = ArrayError;
1615
1616    fn try_from(array: &arrow_array::UInt64Array) -> Result<Self, Self::Error> {
1617        let from_arrow = |value| {
1618            // Convert the value to a Decimal with scale 0
1619            let res = Decimal::from(value);
1620            Ok(res)
1621        };
1622
1623        // Map over the array and convert each value
1624        array
1625            .iter()
1626            .map(|o| o.map(from_arrow).transpose())
1627            .collect::<Result<Self, Self::Error>>()
1628    }
1629}
1630
1631impl TryFrom<&arrow_array::Float16Array> for F32Array {
1632    type Error = ArrayError;
1633
1634    fn try_from(array: &arrow_array::Float16Array) -> Result<Self, Self::Error> {
1635        let from_arrow = |value| Ok(f32::from(value));
1636
1637        array
1638            .iter()
1639            .map(|o| o.map(from_arrow).transpose())
1640            .collect::<Result<Self, Self::Error>>()
1641    }
1642}
1643
1644impl TryFrom<&arrow_array::LargeBinaryArray> for DecimalArray {
1645    type Error = ArrayError;
1646
1647    fn try_from(array: &arrow_array::LargeBinaryArray) -> Result<Self, Self::Error> {
1648        array
1649            .iter()
1650            .map(|o| {
1651                o.map(|s| {
1652                    let s = std::str::from_utf8(s)
1653                        .map_err(|_| ArrayError::from_arrow(format!("invalid decimal: {s:?}")))?;
1654                    s.parse()
1655                        .map_err(|_| ArrayError::from_arrow(format!("invalid decimal: {s:?}")))
1656                })
1657                .transpose()
1658            })
1659            .try_collect()
1660    }
1661}
1662
1663impl TryFrom<&arrow_array::StringArray> for DecimalArray {
1664    type Error = ArrayError;
1665
1666    fn try_from(array: &arrow_array::StringArray) -> Result<Self, Self::Error> {
1667        array
1668            .iter()
1669            .map(|o| {
1670                o.map(|s| {
1671                    s.parse()
1672                        .map_err(|_| ArrayError::from_arrow(format!("invalid decimal: {s:?}")))
1673                })
1674                .transpose()
1675            })
1676            .try_collect()
1677    }
1678}
1679
1680impl From<&JsonbArray> for arrow_array::StringArray {
1681    fn from(array: &JsonbArray) -> Self {
1682        let mut builder =
1683            arrow_array::builder::StringBuilder::with_capacity(array.len(), array.len() * 16);
1684        for value in array.iter() {
1685            match value {
1686                Some(jsonb) => {
1687                    write!(&mut builder, "{}", jsonb).unwrap();
1688                    builder.append_value("");
1689                }
1690                None => builder.append_null(),
1691            }
1692        }
1693        builder.finish()
1694    }
1695}
1696
1697impl TryFrom<&arrow_array::StringArray> for JsonbArray {
1698    type Error = ArrayError;
1699
1700    fn try_from(array: &arrow_array::StringArray) -> Result<Self, Self::Error> {
1701        array
1702            .iter()
1703            .map(|o| {
1704                o.map(|s| {
1705                    s.parse()
1706                        .map_err(|_| ArrayError::from_arrow(format!("invalid json: {s}")))
1707                })
1708                .transpose()
1709            })
1710            .try_collect()
1711    }
1712}
1713
1714impl From<&IntervalArray> for arrow_array::StringArray {
1715    fn from(array: &IntervalArray) -> Self {
1716        let mut builder =
1717            arrow_array::builder::StringBuilder::with_capacity(array.len(), array.len() * 16);
1718        for value in array.iter() {
1719            match value {
1720                Some(interval) => {
1721                    write!(&mut builder, "{}", interval).unwrap();
1722                    builder.append_value("");
1723                }
1724                None => builder.append_null(),
1725            }
1726        }
1727        builder.finish()
1728    }
1729}
1730
1731impl From<&JsonbArray> for arrow_array::LargeStringArray {
1732    fn from(array: &JsonbArray) -> Self {
1733        let mut builder =
1734            arrow_array::builder::LargeStringBuilder::with_capacity(array.len(), array.len() * 16);
1735        for value in array.iter() {
1736            match value {
1737                Some(jsonb) => {
1738                    write!(&mut builder, "{}", jsonb).unwrap();
1739                    builder.append_value("");
1740                }
1741                None => builder.append_null(),
1742            }
1743        }
1744        builder.finish()
1745    }
1746}
1747
1748impl TryFrom<&arrow_array::LargeStringArray> for JsonbArray {
1749    type Error = ArrayError;
1750
1751    fn try_from(array: &arrow_array::LargeStringArray) -> Result<Self, Self::Error> {
1752        array
1753            .iter()
1754            .map(|o| {
1755                o.map(|s| {
1756                    s.parse()
1757                        .map_err(|_| ArrayError::from_arrow(format!("invalid json: {s}")))
1758                })
1759                .transpose()
1760            })
1761            .try_collect()
1762    }
1763}
1764
1765impl From<arrow_buffer::i256> for Int256 {
1766    fn from(value: arrow_buffer::i256) -> Self {
1767        let buffer = value.to_be_bytes();
1768        Int256::from_be_bytes(buffer)
1769    }
1770}
1771
1772impl<'a> From<Int256Ref<'a>> for arrow_buffer::i256 {
1773    fn from(val: Int256Ref<'a>) -> Self {
1774        let buffer = val.to_be_bytes();
1775        arrow_buffer::i256::from_be_bytes(buffer)
1776    }
1777}
1778
1779impl From<&Int256Array> for arrow_array::Decimal256Array {
1780    fn from(array: &Int256Array) -> Self {
1781        array
1782            .iter()
1783            .map(|o| o.map(arrow_buffer::i256::from))
1784            .collect()
1785    }
1786}
1787
1788impl From<&arrow_array::Decimal256Array> for Int256Array {
1789    fn from(array: &arrow_array::Decimal256Array) -> Self {
1790        let values = array.iter().map(|o| o.map(Int256::from)).collect_vec();
1791
1792        values
1793            .iter()
1794            .map(|i| i.as_ref().map(|v| v.as_scalar_ref()))
1795            .collect()
1796    }
1797}
1798
1799/// Field-aware version of [`is_parquet_schema_match_source_schema`]: it inspects the field
1800/// metadata to match an `arrow.parquet.variant` struct against `Variant`. A variant extension
1801/// binds exclusively to `Variant` at every nesting depth — declaring such a column as anything
1802/// else (e.g. the raw physical struct) is an illegal type mismatch, so the parser NULL-fills it
1803/// instead of decoding a diverging type. Other extensions do not affect matching: decode
1804/// follows the declared side, which simply ignores them. Prefer this whenever a `Field` is
1805/// available.
1806pub fn is_parquet_field_match_source_schema(
1807    arrow_field: &arrow_schema::Field,
1808    rw_data_type: &crate::types::DataType,
1809) -> bool {
1810    use arrow_schema::extension::ExtensionType as _;
1811
1812    // A file-side variant extension binds exclusively to `Variant` at every depth; matching it
1813    // against any other declared type is an illegal type mismatch, not a lenient physical
1814    // match. Other extensions do not affect matching: decode follows the declared side, which
1815    // simply ignores them.
1816    if arrow_field.extension_type_name() == Some(parquet_variant_compute::VariantType::NAME) {
1817        return matches!(arrow_field.data_type(), arrow_schema::DataType::Struct(_))
1818            && matches!(rw_data_type, crate::types::DataType::Variant);
1819    }
1820    is_parquet_schema_match_source_schema(arrow_field.data_type(), rw_data_type)
1821}
1822
1823/// This function checks whether the schema of a Parquet file matches the user-defined schema in RisingWave.
1824/// It handles the following special cases:
1825/// - Arrow's `timestamp(_, None)` types (all four time units) match with RisingWave's `Timestamp` type.
1826/// - Arrow's `timestamp(_, Some)` matches with RisingWave's `Timestamptz` type.
1827/// - Since RisingWave does not have an `UInt` type:
1828///   - Arrow's `UInt8` matches with RisingWave's `Int16`.
1829///   - Arrow's `UInt16` matches with RisingWave's `Int32`.
1830///   - Arrow's `UInt32` matches with RisingWave's `Int64`.
1831///   - Arrow's `UInt64` matches with RisingWave's `Decimal`.
1832/// - Arrow's `Float16` matches with RisingWave's `Float32`.
1833///
1834/// Nested data type matching:
1835/// - Struct: Arrow's `Struct` type matches with RisingWave's `Struct` type recursively, requiring that all expected fields exist and match by name and type. Extra Arrow fields are allowed, but a declared name matching multiple Arrow siblings is ambiguous and rejected.
1836/// - List: Arrow's `List` type matches with RisingWave's `List` type recursively, requiring the same element type.
1837/// - Map: Arrow's `Map` type matches with RisingWave's `Map` type recursively, requiring the key and value types to match, and the inner struct must have exactly two fields named "key" and "value".
1838///
1839/// Nested positions recurse through [`is_parquet_field_match_source_schema`] so a variant
1840/// extension is recognized at any depth.
1841pub fn is_parquet_schema_match_source_schema(
1842    arrow_data_type: &arrow_schema::DataType,
1843    rw_data_type: &crate::types::DataType,
1844) -> bool {
1845    use arrow_schema::DataType as ArrowType;
1846
1847    use crate::types::{DataType as RwType, MapType, StructType};
1848
1849    match (arrow_data_type, rw_data_type) {
1850        // Primitive type matching and special cases
1851        (ArrowType::Boolean, RwType::Boolean)
1852        | (ArrowType::Int8 | ArrowType::Int16 | ArrowType::UInt8, RwType::Int16)
1853        | (ArrowType::Int32 | ArrowType::UInt16, RwType::Int32)
1854        | (ArrowType::Int64 | ArrowType::UInt32, RwType::Int64)
1855        | (ArrowType::UInt64 | ArrowType::Decimal128(_, _), RwType::Decimal)
1856        | (ArrowType::Decimal256(_, _), RwType::Int256)
1857        | (ArrowType::Float16 | ArrowType::Float32, RwType::Float32)
1858        | (ArrowType::Float64, RwType::Float64)
1859        | (ArrowType::Timestamp(_, None), RwType::Timestamp)
1860        | (ArrowType::Timestamp(_, Some(_)), RwType::Timestamptz)
1861        | (ArrowType::Date32, RwType::Date)
1862        | (ArrowType::Time32(_) | ArrowType::Time64(_), RwType::Time)
1863        | (ArrowType::Interval(arrow_schema::IntervalUnit::MonthDayNano), RwType::Interval)
1864        | (ArrowType::Utf8 | ArrowType::LargeUtf8, RwType::Varchar)
1865        | (
1866            ArrowType::Binary | ArrowType::LargeBinary | ArrowType::FixedSizeBinary(_),
1867            RwType::Bytea,
1868        ) => true,
1869
1870        // Struct type recursive matching
1871        // Arrow's Struct matches RisingWave's Struct if all expected field names exist and types
1872        // match recursively. Extra Arrow fields are allowed and field order is ignored.
1873        (ArrowType::Struct(arrow_fields), RwType::Struct(rw_struct)) => {
1874            if arrow_fields.len() < rw_struct.len() {
1875                return false;
1876            }
1877            for (rw_name, rw_ty) in rw_struct.iter() {
1878                let mut candidates = arrow_fields.iter().filter(|f| f.name() == rw_name);
1879                let Some(arrow_field) = candidates.next() else {
1880                    return false;
1881                };
1882                // Parquet permits duplicate sibling names; which one holds the data is
1883                // ambiguous, so reject the match.
1884                if candidates.next().is_some() {
1885                    return false;
1886                }
1887                if !is_parquet_field_match_source_schema(arrow_field, rw_ty) {
1888                    return false;
1889                }
1890            }
1891            true
1892        }
1893        // List type recursive matching
1894        // Arrow's List matches RisingWave's List if the element type matches recursively
1895        (ArrowType::List(arrow_field), RwType::List(rw_list_ty)) => {
1896            is_parquet_field_match_source_schema(arrow_field, rw_list_ty.elem())
1897        }
1898        // Map type recursive matching
1899        // Arrow's Map matches RisingWave's Map if the key and value types match recursively,
1900        // and the inner struct has exactly two fields named "key" and "value"
1901        (ArrowType::Map(arrow_field, _), RwType::Map(rw_map_ty)) => {
1902            if let ArrowType::Struct(fields) = arrow_field.data_type() {
1903                if fields.len() != 2 {
1904                    return false;
1905                }
1906                let key_field = &fields[0];
1907                let value_field = &fields[1];
1908                if key_field.name() != "key" || value_field.name() != "value" {
1909                    return false;
1910                }
1911                let (rw_key_ty, rw_value_ty) = (rw_map_ty.key(), rw_map_ty.value());
1912                is_parquet_field_match_source_schema(key_field, rw_key_ty)
1913                    && is_parquet_field_match_source_schema(value_field, rw_value_ty)
1914            } else {
1915                false
1916            }
1917        }
1918        // Fallback: types do not match
1919        _ => false,
1920    }
1921}
1922#[cfg(test)]
1923mod tests {
1924
1925    use arrow_schema::{DataType as ArrowType, Field as ArrowField};
1926
1927    use super::*;
1928    use crate::array::arrow::IcebergArrowConvert;
1929    use crate::types::{DataType as RwType, MapType, StructType};
1930
1931    /// A default-only `FromArrow` for exercising the shared decode logic.
1932    struct Dummy;
1933    impl FromArrow for Dummy {}
1934
1935    fn variant_field(name: &str) -> ArrowField {
1936        use std::collections::HashMap;
1937        ArrowField::new(
1938            name,
1939            ArrowType::Struct(
1940                vec![
1941                    ArrowField::new("metadata", ArrowType::Binary, false),
1942                    ArrowField::new("value", ArrowType::Binary, true),
1943                ]
1944                .into(),
1945            ),
1946            true,
1947        )
1948        .with_metadata(HashMap::from([(
1949            "ARROW:extension:name".to_owned(),
1950            "arrow.parquet.variant".to_owned(),
1951        )]))
1952    }
1953
1954    #[test]
1955    fn test_variant_field_schema_match() {
1956        let variant = variant_field("v");
1957
1958        assert!(is_parquet_field_match_source_schema(
1959            &variant,
1960            &RwType::Variant
1961        ));
1962        assert!(!is_parquet_schema_match_source_schema(
1963            variant.data_type(),
1964            &RwType::Variant
1965        ));
1966        // A variant field does NOT match its raw physical struct layout: the variant extension
1967        // binds exclusively to `Variant`, so declaring it as a struct is an illegal type mismatch.
1968        let rw_physical = RwType::Struct(StructType::new(vec![
1969            ("metadata".to_owned(), RwType::Bytea),
1970            ("value".to_owned(), RwType::Bytea),
1971        ]));
1972        assert!(!is_parquet_field_match_source_schema(
1973            &variant,
1974            &rw_physical
1975        ));
1976
1977        // Variant nested in struct / list / map.
1978        let arrow_struct = ArrowField::new(
1979            "s",
1980            ArrowType::Struct(vec![variant_field("v")].into()),
1981            true,
1982        );
1983        let rw_struct = RwType::Struct(StructType::new(vec![("v".to_owned(), RwType::Variant)]));
1984        assert!(is_parquet_field_match_source_schema(
1985            &arrow_struct,
1986            &rw_struct
1987        ));
1988
1989        let arrow_list = ArrowField::new(
1990            "l",
1991            ArrowType::List(Arc::new(variant_field("element"))),
1992            true,
1993        );
1994        assert!(is_parquet_field_match_source_schema(
1995            &arrow_list,
1996            &RwType::list(RwType::Variant)
1997        ));
1998
1999        let arrow_map = ArrowField::new(
2000            "m",
2001            ArrowType::Map(
2002                Arc::new(ArrowField::new(
2003                    "entries",
2004                    ArrowType::Struct(
2005                        vec![
2006                            ArrowField::new("key", ArrowType::Utf8, false),
2007                            variant_field("value"),
2008                        ]
2009                        .into(),
2010                    ),
2011                    false,
2012                )),
2013                false,
2014            ),
2015            true,
2016        );
2017        assert!(is_parquet_field_match_source_schema(
2018            &arrow_map,
2019            &RwType::Map(MapType::from_kv(RwType::Varchar, RwType::Variant))
2020        ));
2021    }
2022
2023    #[test]
2024    fn test_variant_ext_under_list_map_rejects_physical_struct() {
2025        let physical = RwType::Struct(StructType::new(vec![
2026            ("metadata".to_owned(), RwType::Bytea),
2027            ("value".to_owned(), RwType::Bytea),
2028        ]));
2029
2030        // A list element carrying the variant extension only matches a declared `variant[]`;
2031        // a declared physical struct must NOT match (decoding it would yield `list<variant>`,
2032        // diverging from the catalog, so the parser NULL-fills instead).
2033        let list_field = ArrowField::new(
2034            "l",
2035            ArrowType::List(Arc::new(variant_field("element"))),
2036            true,
2037        );
2038        assert!(!is_parquet_field_match_source_schema(
2039            &list_field,
2040            &RwType::list(physical.clone())
2041        ));
2042        assert!(is_parquet_field_match_source_schema(
2043            &list_field,
2044            &RwType::list(RwType::Variant)
2045        ));
2046
2047        // Same rule for a map value carrying the variant extension.
2048        let map_field = ArrowField::new(
2049            "m",
2050            ArrowType::Map(
2051                Arc::new(ArrowField::new(
2052                    "entries",
2053                    ArrowType::Struct(
2054                        vec![
2055                            ArrowField::new("key", ArrowType::Utf8, false),
2056                            variant_field("value"),
2057                        ]
2058                        .into(),
2059                    ),
2060                    false,
2061                )),
2062                false,
2063            ),
2064            true,
2065        );
2066        assert!(!is_parquet_field_match_source_schema(
2067            &map_field,
2068            &RwType::Map(MapType::from_kv(RwType::Varchar, physical))
2069        ));
2070        assert!(is_parquet_field_match_source_schema(
2071            &map_field,
2072            &RwType::Map(MapType::from_kv(RwType::Varchar, RwType::Variant))
2073        ));
2074    }
2075
2076    #[test]
2077    fn test_variant_ext_nested_under_list_rejects_physical_struct() {
2078        // `list<struct<v: variant-ext>>`: strictness applies one struct level below the list.
2079        let list_field = ArrowField::new(
2080            "l",
2081            ArrowType::List(Arc::new(ArrowField::new(
2082                "element",
2083                ArrowType::Struct(vec![variant_field("v")].into()),
2084                true,
2085            ))),
2086            true,
2087        );
2088        let physical = RwType::Struct(StructType::new(vec![
2089            ("metadata".to_owned(), RwType::Bytea),
2090            ("value".to_owned(), RwType::Bytea),
2091        ]));
2092        let elem_physical = RwType::Struct(StructType::new(vec![("v".to_owned(), physical)]));
2093        assert!(!is_parquet_field_match_source_schema(
2094            &list_field,
2095            &RwType::list(elem_physical)
2096        ));
2097        let elem_variant = RwType::Struct(StructType::new(vec![("v".to_owned(), RwType::Variant)]));
2098        assert!(is_parquet_field_match_source_schema(
2099            &list_field,
2100            &RwType::list(elem_variant)
2101        ));
2102    }
2103
2104    #[test]
2105    fn test_variant_ext_declared_as_scalar_rejects_match() {
2106        // The variant extension binds exclusively to `Variant`, even outside list/map boundaries.
2107        let v = variant_field("v");
2108        assert!(!is_parquet_field_match_source_schema(&v, &RwType::Varchar));
2109        assert!(!is_parquet_field_match_source_schema(&v, &RwType::Bytea));
2110        assert!(is_parquet_field_match_source_schema(&v, &RwType::Variant));
2111    }
2112
2113    #[test]
2114    fn test_nested_struct_reorder_and_superset_decode_by_declared() {
2115        // File inner is reordered and a superset: inner<b, a, c>; declared inner<a, b>.
2116        let inner: arrow_array::ArrayRef = Arc::new(arrow_array::StructArray::from(vec![
2117            (
2118                Arc::new(ArrowField::new("b", ArrowType::Utf8, true)),
2119                Arc::new(arrow_array::StringArray::from(vec![Some("x")])) as arrow_array::ArrayRef,
2120            ),
2121            (
2122                Arc::new(ArrowField::new("a", ArrowType::Int32, true)),
2123                Arc::new(arrow_array::Int32Array::from(vec![Some(1)])) as arrow_array::ArrayRef,
2124            ),
2125            (
2126                Arc::new(ArrowField::new("c", ArrowType::Int32, true)),
2127                Arc::new(arrow_array::Int32Array::from(vec![Some(9)])) as arrow_array::ArrayRef,
2128            ),
2129        ]));
2130        let st: arrow_array::ArrayRef = Arc::new(arrow_array::StructArray::from(vec![(
2131            Arc::new(ArrowField::new("inner", inner.data_type().clone(), true)),
2132            inner,
2133        )]));
2134
2135        let declared_field = ArrowField::new(
2136            "st",
2137            ArrowType::Struct(
2138                vec![ArrowField::new(
2139                    "inner",
2140                    ArrowType::Struct(
2141                        vec![
2142                            ArrowField::new("a", ArrowType::Int32, true),
2143                            ArrowField::new("b", ArrowType::Utf8, true),
2144                        ]
2145                        .into(),
2146                    ),
2147                    true,
2148                )]
2149                .into(),
2150            ),
2151            true,
2152        );
2153        let converted = IcebergArrowConvert
2154            .array_from_arrow_array(&declared_field, &st)
2155            .unwrap();
2156
2157        assert_eq!(
2158            converted.data_type(),
2159            RwType::Struct(StructType::new(vec![(
2160                "inner",
2161                RwType::Struct(StructType::new(vec![
2162                    ("a", RwType::Int32),
2163                    ("b", RwType::Varchar),
2164                ])),
2165            )])),
2166        );
2167        let ArrayImpl::Struct(s) = &converted else {
2168            panic!("expected RW struct");
2169        };
2170        assert_eq!(
2171            s.value_at(0).unwrap().to_owned_scalar(),
2172            StructValue::new(vec![Some(ScalarImpl::Struct(StructValue::new(vec![
2173                Some(ScalarImpl::Int32(1)),
2174                Some(ScalarImpl::Utf8("x".into())),
2175            ])))]),
2176        );
2177    }
2178
2179    #[test]
2180    fn test_variant_ext_grandchild_decodes_as_physical_struct() {
2181        // Actual: s<mid<v: variant-ext struct<metadata, value>>>, with binary children.
2182        let v_child: arrow_array::ArrayRef = Arc::new(arrow_array::StructArray::from(vec![
2183            (
2184                Arc::new(ArrowField::new("metadata", ArrowType::Binary, false)),
2185                Arc::new(arrow_array::BinaryArray::from_iter_values([
2186                    &[1_u8, 0, 0][..]
2187                ])) as arrow_array::ArrayRef,
2188            ),
2189            (
2190                Arc::new(ArrowField::new("value", ArrowType::Binary, true)),
2191                Arc::new(arrow_array::BinaryArray::from_iter_values([&[9_u8][..]]))
2192                    as arrow_array::ArrayRef,
2193            ),
2194        ]));
2195        let mid: arrow_array::ArrayRef = Arc::new(arrow_array::StructArray::from(vec![(
2196            Arc::new(variant_field("v")),
2197            v_child,
2198        )]));
2199        let s: arrow_array::ArrayRef = Arc::new(arrow_array::StructArray::from(vec![(
2200            Arc::new(ArrowField::new("mid", mid.data_type().clone(), true)),
2201            mid,
2202        )]));
2203
2204        // Declared as a physical struct all the way down (no variant).
2205        let declared = RwType::Struct(StructType::new(vec![(
2206            "mid".to_owned(),
2207            RwType::Struct(StructType::new(vec![(
2208                "v".to_owned(),
2209                RwType::Struct(StructType::new(vec![
2210                    ("metadata".to_owned(), RwType::Bytea),
2211                    ("value".to_owned(), RwType::Bytea),
2212                ])),
2213            )])),
2214        )]));
2215        let declared_field = IcebergArrowConvert.to_arrow_field("s", &declared).unwrap();
2216
2217        let converted = IcebergArrowConvert
2218            .array_from_arrow_array(&declared_field, &s)
2219            .unwrap();
2220        // The depth-2 variant extension is ignored: `v` decodes as raw bytea struct, not Variant.
2221        assert_eq!(
2222            converted.data_type(),
2223            RwType::Struct(StructType::new(vec![(
2224                "mid",
2225                RwType::Struct(StructType::new(vec![(
2226                    "v",
2227                    RwType::Struct(StructType::new(vec![
2228                        ("metadata", RwType::Bytea),
2229                        ("value", RwType::Bytea),
2230                    ])),
2231                )])),
2232            )])),
2233        );
2234    }
2235
2236    #[test]
2237    fn test_struct_schema_match() {
2238        // Arrow: struct<f1: Double, f2: Utf8>
2239
2240        let arrow_struct = ArrowType::Struct(
2241            vec![
2242                ArrowField::new("f1", ArrowType::Float64, true),
2243                ArrowField::new("f2", ArrowType::Utf8, true),
2244            ]
2245            .into(),
2246        );
2247        // RW: struct<f1 Double, f2 Varchar>
2248        let rw_struct = RwType::Struct(StructType::new(vec![
2249            ("f1".to_owned(), RwType::Float64),
2250            ("f2".to_owned(), RwType::Varchar),
2251        ]));
2252        assert!(is_parquet_schema_match_source_schema(
2253            &arrow_struct,
2254            &rw_struct
2255        ));
2256
2257        // Arrow is a superset of RW struct fields.
2258        let arrow_struct_superset = ArrowType::Struct(
2259            vec![
2260                ArrowField::new("f1", ArrowType::Float64, true),
2261                ArrowField::new("f2", ArrowType::Utf8, true),
2262                ArrowField::new("f3", ArrowType::Int32, true),
2263            ]
2264            .into(),
2265        );
2266        assert!(is_parquet_schema_match_source_schema(
2267            &arrow_struct_superset,
2268            &rw_struct
2269        ));
2270
2271        // Field order is ignored for struct matching.
2272        let arrow_struct_reordered = ArrowType::Struct(
2273            vec![
2274                ArrowField::new("f2", ArrowType::Utf8, true),
2275                ArrowField::new("f1", ArrowType::Float64, true),
2276            ]
2277            .into(),
2278        );
2279        assert!(is_parquet_schema_match_source_schema(
2280            &arrow_struct_reordered,
2281            &rw_struct
2282        ));
2283
2284        // Field names do not match
2285        let arrow_struct2 = ArrowType::Struct(
2286            vec![
2287                ArrowField::new("f1", ArrowType::Float64, true),
2288                ArrowField::new("f3", ArrowType::Utf8, true),
2289            ]
2290            .into(),
2291        );
2292        assert!(!is_parquet_schema_match_source_schema(
2293            &arrow_struct2,
2294            &rw_struct
2295        ));
2296    }
2297
2298    #[test]
2299    fn test_struct_duplicate_sibling_names_reject_match() {
2300        let rw_struct = RwType::Struct(StructType::new(vec![("f1".to_owned(), RwType::Float64)]));
2301
2302        // A declared name matching multiple Arrow siblings is ambiguous, even when the
2303        // duplicates carry the same type.
2304        for dup_type in [ArrowType::Float64, ArrowType::Utf8] {
2305            let arrow_struct = ArrowType::Struct(
2306                vec![
2307                    ArrowField::new("f1", ArrowType::Float64, true),
2308                    ArrowField::new("f1", dup_type, true),
2309                ]
2310                .into(),
2311            );
2312            assert!(!is_parquet_schema_match_source_schema(
2313                &arrow_struct,
2314                &rw_struct
2315            ));
2316        }
2317
2318        // Duplicates among extra (undeclared) fields are irrelevant: they are dropped anyway.
2319        let arrow_struct = ArrowType::Struct(
2320            vec![
2321                ArrowField::new("f1", ArrowType::Float64, true),
2322                ArrowField::new("extra", ArrowType::Int32, true),
2323                ArrowField::new("extra", ArrowType::Utf8, true),
2324            ]
2325            .into(),
2326        );
2327        assert!(is_parquet_schema_match_source_schema(
2328            &arrow_struct,
2329            &rw_struct
2330        ));
2331    }
2332
2333    #[test]
2334    fn test_struct_projection_from_arrow() {
2335        use itertools::Itertools;
2336
2337        // Actual Arrow struct: struct<foo:int32, bar:utf8, baz:int32>
2338        let actual_fields: arrow_schema::Fields = vec![
2339            ArrowField::new("foo", ArrowType::Int32, true),
2340            ArrowField::new("bar", ArrowType::Utf8, true),
2341            ArrowField::new("baz", ArrowType::Int32, true),
2342        ]
2343        .into();
2344        let foo: arrow_array::ArrayRef =
2345            Arc::new(arrow_array::Int32Array::from(vec![Some(10), Some(20)]));
2346        let bar: arrow_array::ArrayRef =
2347            Arc::new(arrow_array::StringArray::from(vec![Some("a"), Some("b")]));
2348        let baz: arrow_array::ArrayRef =
2349            Arc::new(arrow_array::Int32Array::from(vec![Some(100), Some(200)]));
2350        let actual_struct = arrow_array::StructArray::new(actual_fields, vec![foo, bar, baz], None);
2351        let actual_struct_ref: arrow_array::ArrayRef = Arc::new(actual_struct);
2352
2353        // Expected struct in RW schema (via to_arrow_field): struct<foo:int32, bar:utf8>
2354        let expected_field = ArrowField::new(
2355            "s",
2356            ArrowType::Struct(
2357                vec![
2358                    ArrowField::new("foo", ArrowType::Int32, true),
2359                    ArrowField::new("bar", ArrowType::Utf8, true),
2360                ]
2361                .into(),
2362            ),
2363            true,
2364        );
2365
2366        let array_impl = Dummy
2367            .from_array(&expected_field, &actual_struct_ref)
2368            .unwrap();
2369
2370        let ArrayImpl::Struct(s) = array_impl else {
2371            panic!("expected RW StructArray");
2372        };
2373
2374        let DataType::Struct(st) = s.data_type() else {
2375            panic!("expected RW struct type");
2376        };
2377        assert_eq!(st.len(), 2);
2378        assert_eq!(st.iter().map(|(n, _)| n).collect_vec(), vec!["foo", "bar"]);
2379
2380        let v0 = s.value_at(0).unwrap().to_owned_scalar();
2381        let v1 = s.value_at(1).unwrap().to_owned_scalar();
2382        assert_eq!(
2383            v0,
2384            StructValue::new(vec![
2385                Some(ScalarImpl::Int32(10)),
2386                Some(ScalarImpl::Utf8("a".into()))
2387            ])
2388        );
2389        assert_eq!(
2390            v1,
2391            StructValue::new(vec![
2392                Some(ScalarImpl::Int32(20)),
2393                Some(ScalarImpl::Utf8("b".into()))
2394            ])
2395        );
2396    }
2397
2398    /// Builds a two-element `list<struct>` array from the given element fields and columns.
2399    fn build_list_of_struct(
2400        elem_fields: arrow_schema::Fields,
2401        columns: Vec<arrow_array::ArrayRef>,
2402    ) -> arrow_array::ArrayRef {
2403        use std::sync::Arc;
2404        let elem_struct = arrow_array::StructArray::new(elem_fields.clone(), columns, None);
2405        Arc::new(arrow_array::ListArray::new(
2406            Arc::new(ArrowField::new(
2407                "element",
2408                ArrowType::Struct(elem_fields),
2409                true,
2410            )),
2411            arrow_buffer::OffsetBuffer::new(vec![0, 2].into()),
2412            Arc::new(elem_struct),
2413            None,
2414        ))
2415    }
2416
2417    #[test]
2418    fn test_list_element_struct_decodes_by_declared_field() {
2419        // File: list<struct<b utf8, a int32, extra int32>> — reordered and a superset of the
2420        // declared element struct.
2421        let file_array = build_list_of_struct(
2422            vec![
2423                ArrowField::new("b", ArrowType::Utf8, true),
2424                ArrowField::new("a", ArrowType::Int32, true),
2425                ArrowField::new("extra", ArrowType::Int32, true),
2426            ]
2427            .into(),
2428            vec![
2429                Arc::new(arrow_array::StringArray::from(vec![Some("x"), Some("y")])),
2430                Arc::new(arrow_array::Int32Array::from(vec![Some(1), Some(2)])),
2431                Arc::new(arrow_array::Int32Array::from(vec![Some(9), Some(8)])),
2432            ],
2433        );
2434        // Declared: list<struct<a int, b varchar>>.
2435        let declared_elem: arrow_schema::Fields = vec![
2436            ArrowField::new("a", ArrowType::Int32, true),
2437            ArrowField::new("b", ArrowType::Utf8, true),
2438        ]
2439        .into();
2440        let declared_field = ArrowField::new(
2441            "l",
2442            ArrowType::List(Arc::new(ArrowField::new(
2443                "element",
2444                ArrowType::Struct(declared_elem),
2445                true,
2446            ))),
2447            true,
2448        );
2449
2450        let converted = Dummy.from_array(&declared_field, &file_array).unwrap();
2451        assert_eq!(
2452            converted.data_type(),
2453            RwType::list(RwType::Struct(StructType::new(vec![
2454                ("a", RwType::Int32),
2455                ("b", RwType::Varchar),
2456            ])))
2457        );
2458        let ArrayImpl::List(list) = &converted else {
2459            panic!("expected list array");
2460        };
2461        let ArrayImpl::Struct(elems) = list.values() else {
2462            panic!("expected struct elements");
2463        };
2464        assert_eq!(
2465            elems.value_at(0).unwrap().to_owned_scalar(),
2466            StructValue::new(vec![
2467                Some(ScalarImpl::Int32(1)),
2468                Some(ScalarImpl::Utf8("x".into())),
2469            ])
2470        );
2471        assert_eq!(
2472            elems.value_at(1).unwrap().to_owned_scalar(),
2473            StructValue::new(vec![
2474                Some(ScalarImpl::Int32(2)),
2475                Some(ScalarImpl::Utf8("y".into())),
2476            ])
2477        );
2478    }
2479
2480    /// Decodes a struct array against a declared struct with the given fields.
2481    fn decode_struct(
2482        declared_fields: Vec<arrow_schema::Field>,
2483        actual: arrow_array::StructArray,
2484    ) -> Result<ArrayImpl, ArrayError> {
2485        let declared = ArrowField::new("s", ArrowType::Struct(declared_fields.into()), true);
2486        let array: arrow_array::ArrayRef = Arc::new(actual);
2487        Dummy.from_array(&declared, &array)
2488    }
2489
2490    #[test]
2491    fn test_struct_name_mismatch_same_arity_decodes_positionally() {
2492        // An external UDF may label struct children differently from the declared return
2493        // type; the correspondence defined by the signature check is positional.
2494        let actual_fields: arrow_schema::Fields = vec![
2495            ArrowField::new("total", ArrowType::Int32, true),
2496            ArrowField::new("count", ArrowType::Utf8, true),
2497        ]
2498        .into();
2499        let array = arrow_array::StructArray::new(
2500            actual_fields,
2501            vec![
2502                Arc::new(arrow_array::Int32Array::from(vec![Some(42)])),
2503                Arc::new(arrow_array::StringArray::from(vec![Some("x")])),
2504            ],
2505            None,
2506        );
2507
2508        let converted = decode_struct(
2509            vec![
2510                ArrowField::new("sum", ArrowType::Int32, true),
2511                ArrowField::new("cnt", ArrowType::Utf8, true),
2512            ],
2513            array,
2514        )
2515        .unwrap();
2516        assert_eq!(
2517            converted.data_type(),
2518            RwType::Struct(StructType::new(vec![
2519                ("sum", RwType::Int32),
2520                ("cnt", RwType::Varchar),
2521            ]))
2522        );
2523        let ArrayImpl::Struct(structs) = &converted else {
2524            panic!("expected struct array");
2525        };
2526        assert_eq!(
2527            structs.value_at(0).unwrap().to_owned_scalar(),
2528            StructValue::new(vec![
2529                Some(ScalarImpl::Int32(42)),
2530                Some(ScalarImpl::Utf8("x".into())),
2531            ])
2532        );
2533    }
2534
2535    #[test]
2536    fn test_struct_unalignable_fields_error() {
2537        // A declared name is missing and the arity differs: neither by-name nor positional
2538        // alignment applies.
2539        let array = arrow_array::StructArray::new(
2540            vec![ArrowField::new("a", ArrowType::Int32, true)].into(),
2541            vec![Arc::new(arrow_array::Int32Array::from(vec![Some(1)]))],
2542            None,
2543        );
2544
2545        let err = decode_struct(
2546            vec![
2547                ArrowField::new("a", ArrowType::Int32, true),
2548                ArrowField::new("b", ArrowType::Utf8, true),
2549            ],
2550            array,
2551        )
2552        .unwrap_err();
2553        assert!(
2554            err.to_string()
2555                .contains("unable to align struct fields: expected [a, b], actual [a]"),
2556            "unexpected error: {err}"
2557        );
2558    }
2559
2560    #[test]
2561    fn test_struct_child_type_divergence_stamped_honestly() {
2562        // Same child name, different type: the result must report the decoded child type,
2563        // not the expected one, so callers' boundary checks can see the divergence.
2564        let array = arrow_array::StructArray::new(
2565            vec![ArrowField::new("a", ArrowType::Utf8, true)].into(),
2566            vec![Arc::new(arrow_array::StringArray::from(vec![Some("oops")]))],
2567            None,
2568        );
2569
2570        let converted =
2571            decode_struct(vec![ArrowField::new("a", ArrowType::Int64, true)], array).unwrap();
2572        assert_eq!(
2573            converted.data_type(),
2574            RwType::Struct(StructType::new(vec![("a", RwType::Varchar)]))
2575        );
2576    }
2577
2578    #[test]
2579    fn test_struct_child_decodes_despite_lossy_expected_field() {
2580        // The parquet path renders a declared `struct<a smallint>` through the iceberg-lossy
2581        // to_arrow_field as struct<a: Int32>. A foreign file storing a genuine Int16 child
2582        // must still decode to Int16 with correct values instead of erroring.
2583        let array = arrow_array::StructArray::new(
2584            vec![ArrowField::new("a", ArrowType::Int16, true)].into(),
2585            vec![Arc::new(arrow_array::Int16Array::from(vec![
2586                Some(7),
2587                Some(-3),
2588            ]))],
2589            None,
2590        );
2591
2592        let converted =
2593            decode_struct(vec![ArrowField::new("a", ArrowType::Int32, true)], array).unwrap();
2594        assert_eq!(
2595            converted.data_type(),
2596            RwType::Struct(StructType::new(vec![("a", RwType::Int16)]))
2597        );
2598        let ArrayImpl::Struct(structs) = &converted else {
2599            panic!("expected struct array");
2600        };
2601        assert_eq!(
2602            structs.value_at(0).unwrap().to_owned_scalar(),
2603            StructValue::new(vec![Some(ScalarImpl::Int16(7))])
2604        );
2605    }
2606
2607    #[test]
2608    fn test_struct_duplicate_sibling_names_decode_first_occurrence() {
2609        // The by-name decode must consult the same child as the schema matcher (the first
2610        // occurrence), never a later duplicate of a different type.
2611        let actual_fields: arrow_schema::Fields = vec![
2612            ArrowField::new("a", ArrowType::Int32, true),
2613            ArrowField::new("a", ArrowType::Utf8, true),
2614            ArrowField::new("b", ArrowType::Utf8, true),
2615        ]
2616        .into();
2617        let array = arrow_array::StructArray::new(
2618            actual_fields,
2619            vec![
2620                Arc::new(arrow_array::Int32Array::from(vec![Some(1)])),
2621                Arc::new(arrow_array::StringArray::from(vec![Some("dup")])),
2622                Arc::new(arrow_array::StringArray::from(vec![Some("x")])),
2623            ],
2624            None,
2625        );
2626
2627        let converted = decode_struct(
2628            vec![
2629                ArrowField::new("a", ArrowType::Int32, true),
2630                ArrowField::new("b", ArrowType::Utf8, true),
2631            ],
2632            array,
2633        )
2634        .unwrap();
2635        let ArrayImpl::Struct(structs) = &converted else {
2636            panic!("expected struct array");
2637        };
2638        assert_eq!(
2639            structs.value_at(0).unwrap().to_owned_scalar(),
2640            StructValue::new(vec![
2641                Some(ScalarImpl::Int32(1)),
2642                Some(ScalarImpl::Utf8("x".into())),
2643            ])
2644        );
2645    }
2646
2647    #[test]
2648    fn test_extension_decode_follows_declared_field_under_list() {
2649        let json_meta: std::collections::HashMap<String, String> = [(
2650            "ARROW:extension:name".to_owned(),
2651            "arrowudf.json".to_owned(),
2652        )]
2653        .into();
2654        let strings: arrow_array::ArrayRef = Arc::new(arrow_array::StringArray::from(vec![
2655            Some(r#"{"k":1}"#),
2656            Some("2"),
2657        ]));
2658        let make_list = |elem_field: ArrowField, values: arrow_array::ArrayRef| {
2659            Arc::new(arrow_array::ListArray::new(
2660                Arc::new(elem_field),
2661                arrow_buffer::OffsetBuffer::new(vec![0, 2].into()),
2662                values,
2663                None,
2664            )) as arrow_array::ArrayRef
2665        };
2666        let plain_elem = ArrowField::new("element", ArrowType::Utf8, true);
2667        let json_elem = plain_elem.clone().with_metadata(json_meta);
2668
2669        // A file-side extension is ignored when the declared element is plain varchar.
2670        let file_json = make_list(json_elem.clone(), strings.clone());
2671        let declared_plain =
2672            ArrowField::new("l", ArrowType::List(Arc::new(plain_elem.clone())), true);
2673        let converted = Dummy.from_array(&declared_plain, &file_json).unwrap();
2674        assert_eq!(converted.data_type(), RwType::list(RwType::Varchar));
2675
2676        // A declared-side extension drives the decode even when the file element is plain.
2677        let file_plain = make_list(plain_elem, strings);
2678        let declared_json = ArrowField::new("l", ArrowType::List(Arc::new(json_elem)), true);
2679        let converted = Dummy.from_array(&declared_json, &file_plain).unwrap();
2680        assert_eq!(converted.data_type(), RwType::list(RwType::Jsonb));
2681    }
2682
2683    #[test]
2684    fn test_map_value_struct_decodes_by_declared_field() {
2685        // File: map<utf8, struct<y int32, x int32>>; declared value struct is the subset
2686        // struct<x int32>.
2687        let value_fields: arrow_schema::Fields = vec![
2688            ArrowField::new("y", ArrowType::Int32, true),
2689            ArrowField::new("x", ArrowType::Int32, true),
2690        ]
2691        .into();
2692        let entries_fields: arrow_schema::Fields = vec![
2693            ArrowField::new("key", ArrowType::Utf8, false),
2694            ArrowField::new("value", ArrowType::Struct(value_fields.clone()), true),
2695        ]
2696        .into();
2697        let value_struct = arrow_array::StructArray::new(
2698            value_fields,
2699            vec![
2700                Arc::new(arrow_array::Int32Array::from(vec![Some(7)])),
2701                Arc::new(arrow_array::Int32Array::from(vec![Some(42)])),
2702            ],
2703            None,
2704        );
2705        let entries = arrow_array::StructArray::new(
2706            entries_fields.clone(),
2707            vec![
2708                Arc::new(arrow_array::StringArray::from(vec![Some("k")])),
2709                Arc::new(value_struct),
2710            ],
2711            None,
2712        );
2713        let file_map: arrow_array::ArrayRef = Arc::new(arrow_array::MapArray::new(
2714            Arc::new(ArrowField::new(
2715                "entries",
2716                ArrowType::Struct(entries_fields),
2717                false,
2718            )),
2719            arrow_buffer::OffsetBuffer::new(vec![0, 1].into()),
2720            entries,
2721            None,
2722            false,
2723        ));
2724
2725        let declared_value =
2726            ArrowType::Struct(vec![ArrowField::new("x", ArrowType::Int32, true)].into());
2727        let declared_field = ArrowField::new(
2728            "m",
2729            ArrowType::Map(
2730                Arc::new(ArrowField::new(
2731                    "entries",
2732                    ArrowType::Struct(
2733                        vec![
2734                            ArrowField::new("key", ArrowType::Utf8, false),
2735                            ArrowField::new("value", declared_value, true),
2736                        ]
2737                        .into(),
2738                    ),
2739                    false,
2740                )),
2741                false,
2742            ),
2743            true,
2744        );
2745
2746        let converted = Dummy.from_array(&declared_field, &file_map).unwrap();
2747        assert_eq!(
2748            converted.data_type(),
2749            RwType::Map(MapType::from_kv(
2750                RwType::Varchar,
2751                RwType::Struct(StructType::new(vec![("x", RwType::Int32)])),
2752            ))
2753        );
2754        let ArrayImpl::Map(map) = &converted else {
2755            panic!("expected map array");
2756        };
2757        let ArrayImpl::Struct(entries) = map.inner.values() else {
2758            panic!("expected struct entries");
2759        };
2760        assert_eq!(
2761            entries.value_at(0).unwrap().to_owned_scalar(),
2762            StructValue::new(vec![
2763                Some(ScalarImpl::Utf8("k".into())),
2764                Some(ScalarImpl::Struct(StructValue::new(vec![Some(
2765                    ScalarImpl::Int32(42)
2766                )]))),
2767            ])
2768        );
2769    }
2770
2771    #[test]
2772    fn test_map_invalid_key_type_errors() {
2773        // A float64 map key is representable in arrow but not in RW's `MapType`;
2774        // decode must reject it instead of building a map whose `data_type()` panics.
2775        let entries_fields: arrow_schema::Fields = vec![
2776            ArrowField::new("key", ArrowType::Float64, false),
2777            ArrowField::new("value", ArrowType::Int32, true),
2778        ]
2779        .into();
2780        let entries = arrow_array::StructArray::new(
2781            entries_fields.clone(),
2782            vec![
2783                Arc::new(arrow_array::Float64Array::from(vec![Some(1.5)])),
2784                Arc::new(arrow_array::Int32Array::from(vec![Some(42)])),
2785            ],
2786            None,
2787        );
2788        let entries_field = Arc::new(ArrowField::new(
2789            "entries",
2790            ArrowType::Struct(entries_fields),
2791            false,
2792        ));
2793        let file_map: arrow_array::ArrayRef = Arc::new(arrow_array::MapArray::new(
2794            entries_field.clone(),
2795            arrow_buffer::OffsetBuffer::new(vec![0, 1].into()),
2796            entries,
2797            None,
2798            false,
2799        ));
2800        let field = ArrowField::new("m", ArrowType::Map(entries_field, false), true);
2801
2802        let err = Dummy.from_array(&field, &file_map).unwrap_err();
2803        assert!(
2804            err.to_string().contains("invalid map key type"),
2805            "unexpected error: {err}"
2806        );
2807    }
2808
2809    #[test]
2810    fn test_list_schema_match() {
2811        // Arrow: list<double>
2812        let arrow_list =
2813            ArrowType::List(Box::new(ArrowField::new("item", ArrowType::Float64, true)).into());
2814        // RW: list<double>
2815        let rw_list = RwType::Float64.list();
2816        assert!(is_parquet_schema_match_source_schema(&arrow_list, &rw_list));
2817
2818        let rw_list2 = RwType::Int32.list();
2819        assert!(!is_parquet_schema_match_source_schema(
2820            &arrow_list,
2821            &rw_list2
2822        ));
2823    }
2824
2825    #[test]
2826    fn test_map_schema_match() {
2827        // Arrow: map<utf8, int32>
2828        let arrow_map = ArrowType::Map(
2829            Arc::new(ArrowField::new(
2830                "entries",
2831                ArrowType::Struct(
2832                    vec![
2833                        ArrowField::new("key", ArrowType::Utf8, false),
2834                        ArrowField::new("value", ArrowType::Int32, true),
2835                    ]
2836                    .into(),
2837                ),
2838                false,
2839            )),
2840            false,
2841        );
2842        // RW: map<varchar, int32>
2843        let rw_map = RwType::Map(MapType::from_kv(RwType::Varchar, RwType::Int32));
2844        assert!(is_parquet_schema_match_source_schema(&arrow_map, &rw_map));
2845
2846        // Key type does not match
2847        let rw_map2 = RwType::Map(MapType::from_kv(RwType::Int32, RwType::Int32));
2848        assert!(!is_parquet_schema_match_source_schema(&arrow_map, &rw_map2));
2849
2850        // Value type does not match
2851        let rw_map3 = RwType::Map(MapType::from_kv(RwType::Varchar, RwType::Float64));
2852        assert!(!is_parquet_schema_match_source_schema(&arrow_map, &rw_map3));
2853
2854        // Arrow inner struct field name does not match
2855        let arrow_map2 = ArrowType::Map(
2856            Arc::new(ArrowField::new(
2857                "entries",
2858                ArrowType::Struct(
2859                    vec![
2860                        ArrowField::new("k", ArrowType::Utf8, false),
2861                        ArrowField::new("value", ArrowType::Int32, true),
2862                    ]
2863                    .into(),
2864                ),
2865                false,
2866            )),
2867            false,
2868        );
2869        assert!(!is_parquet_schema_match_source_schema(&arrow_map2, &rw_map));
2870    }
2871
2872    #[test]
2873    fn bool() {
2874        let array = BoolArray::from_iter([None, Some(false), Some(true)]);
2875        let arrow = arrow_array::BooleanArray::from(&array);
2876        assert_eq!(BoolArray::from(&arrow), array);
2877    }
2878
2879    #[test]
2880    fn i16() {
2881        let array = I16Array::from_iter([None, Some(-7), Some(25)]);
2882        let arrow = arrow_array::Int16Array::from(&array);
2883        assert_eq!(I16Array::from(&arrow), array);
2884    }
2885
2886    #[test]
2887    fn i32() {
2888        let array = I32Array::from_iter([None, Some(-7), Some(25)]);
2889        let arrow = arrow_array::Int32Array::from(&array);
2890        assert_eq!(I32Array::from(&arrow), array);
2891    }
2892
2893    #[test]
2894    fn i64() {
2895        let array = I64Array::from_iter([None, Some(-7), Some(25)]);
2896        let arrow = arrow_array::Int64Array::from(&array);
2897        assert_eq!(I64Array::from(&arrow), array);
2898    }
2899
2900    #[test]
2901    fn f32() {
2902        let array = F32Array::from_iter([None, Some(-7.0), Some(25.0)]);
2903        let arrow = arrow_array::Float32Array::from(&array);
2904        assert_eq!(F32Array::from(&arrow), array);
2905    }
2906
2907    #[test]
2908    fn f64() {
2909        let array = F64Array::from_iter([None, Some(-7.0), Some(25.0)]);
2910        let arrow = arrow_array::Float64Array::from(&array);
2911        assert_eq!(F64Array::from(&arrow), array);
2912    }
2913
2914    #[test]
2915    fn int8() {
2916        let array: PrimitiveArray<i16> = I16Array::from_iter([None, Some(-128), Some(127)]);
2917        let arr = arrow_array::Int8Array::from(vec![None, Some(-128), Some(127)]);
2918        let converted: PrimitiveArray<i16> = (&arr).into();
2919        assert_eq!(converted, array);
2920    }
2921
2922    #[test]
2923    fn uint8() {
2924        let array: PrimitiveArray<i16> = I16Array::from_iter([None, Some(7), Some(25)]);
2925        let arr = arrow_array::UInt8Array::from(vec![None, Some(7), Some(25)]);
2926        let converted: PrimitiveArray<i16> = (&arr).into();
2927        assert_eq!(converted, array);
2928    }
2929
2930    #[test]
2931    fn uint16() {
2932        let array: PrimitiveArray<i32> = I32Array::from_iter([None, Some(7), Some(65535)]);
2933        let arr = arrow_array::UInt16Array::from(vec![None, Some(7), Some(65535)]);
2934        let converted: PrimitiveArray<i32> = (&arr).into();
2935        assert_eq!(converted, array);
2936    }
2937
2938    #[test]
2939    fn uint32() {
2940        let array: PrimitiveArray<i64> = I64Array::from_iter([None, Some(7), Some(4294967295)]);
2941        let arr = arrow_array::UInt32Array::from(vec![None, Some(7), Some(4294967295)]);
2942        let converted: PrimitiveArray<i64> = (&arr).into();
2943        assert_eq!(converted, array);
2944    }
2945
2946    #[test]
2947    fn uint64() {
2948        let array: PrimitiveArray<Decimal> = DecimalArray::from_iter([
2949            None,
2950            Some(Decimal::Normalized("7".parse().unwrap())),
2951            Some(Decimal::Normalized("18446744073709551615".parse().unwrap())),
2952        ]);
2953        let arr = arrow_array::UInt64Array::from(vec![None, Some(7), Some(18446744073709551615)]);
2954        let converted: PrimitiveArray<Decimal> = (&arr).try_into().unwrap();
2955        assert_eq!(converted, array);
2956    }
2957
2958    #[test]
2959    fn date() {
2960        let array = DateArray::from_iter([
2961            None,
2962            Date::with_days_since_ce(12345).ok(),
2963            Date::with_days_since_ce(-12345).ok(),
2964        ]);
2965        let arrow = arrow_array::Date32Array::try_from(&array).unwrap();
2966        assert_eq!(DateArray::try_from(&arrow).unwrap(), array);
2967    }
2968
2969    #[test]
2970    fn time() {
2971        let array = TimeArray::from_iter([None, Time::with_micro(24 * 3600 * 1_000_000 - 1).ok()]);
2972        let arrow = arrow_array::Time64MicrosecondArray::try_from(&array).unwrap();
2973        assert_eq!(TimeArray::try_from(&arrow).unwrap(), array);
2974    }
2975
2976    #[test]
2977    fn time_arrow_units_round_trip() {
2978        let second_array = TimeArray::from_iter([
2979            None,
2980            Time::with_secs_nano(1, 0).ok(),
2981            Time::with_secs_nano(2, 0).ok(),
2982        ]);
2983        let arrow = arrow_array::Time32SecondArray::try_from(&second_array).unwrap();
2984        assert_eq!(
2985            arrow,
2986            arrow_array::Time32SecondArray::from(vec![None, Some(1), Some(2)])
2987        );
2988        assert_eq!(TimeArray::try_from(&arrow).unwrap(), second_array);
2989
2990        let millisecond_array = TimeArray::from_iter([
2991            None,
2992            Time::with_secs_nano(1, 0).ok(),
2993            Time::with_secs_nano(2, 123_000_000).ok(),
2994        ]);
2995        let arrow = arrow_array::Time32MillisecondArray::try_from(&millisecond_array).unwrap();
2996        assert_eq!(
2997            arrow,
2998            arrow_array::Time32MillisecondArray::from(vec![None, Some(1_000), Some(2_123)])
2999        );
3000        assert_eq!(TimeArray::try_from(&arrow).unwrap(), millisecond_array);
3001
3002        let microsecond_array = TimeArray::from_iter([
3003            None,
3004            Time::with_secs_nano(1, 0).ok(),
3005            Time::with_secs_nano(2, 123_456_000).ok(),
3006        ]);
3007        let arrow = arrow_array::Time64MicrosecondArray::try_from(&microsecond_array).unwrap();
3008        assert_eq!(
3009            arrow,
3010            arrow_array::Time64MicrosecondArray::from(vec![None, Some(1_000_000), Some(2_123_456)])
3011        );
3012        assert_eq!(TimeArray::try_from(&arrow).unwrap(), microsecond_array);
3013
3014        let nanosecond_array = TimeArray::from_iter([
3015            None,
3016            Time::with_secs_nano(1, 0).ok(),
3017            Time::with_secs_nano(2, 123_456_000).ok(),
3018        ]);
3019        let arrow = arrow_array::Time64NanosecondArray::try_from(&nanosecond_array).unwrap();
3020        assert_eq!(
3021            arrow,
3022            arrow_array::Time64NanosecondArray::from(vec![
3023                None,
3024                Some(1_000_000_000),
3025                Some(2_123_456_000)
3026            ])
3027        );
3028        assert_eq!(TimeArray::try_from(&arrow).unwrap(), nanosecond_array);
3029    }
3030
3031    #[test]
3032    fn time_arrow_units_truncate_sub_unit_precision() {
3033        let array = TimeArray::from_iter([Time::with_secs_nano(2, 123_456_789).ok()]);
3034
3035        assert_eq!(
3036            arrow_array::Time32SecondArray::try_from(&array).unwrap(),
3037            arrow_array::Time32SecondArray::from(vec![Some(2)])
3038        );
3039        assert_eq!(
3040            arrow_array::Time32MillisecondArray::try_from(&array).unwrap(),
3041            arrow_array::Time32MillisecondArray::from(vec![Some(2_123)])
3042        );
3043        assert_eq!(
3044            arrow_array::Time64MicrosecondArray::try_from(&array).unwrap(),
3045            arrow_array::Time64MicrosecondArray::from(vec![Some(2_123_456)])
3046        );
3047
3048        // RisingWave's `time` is microsecond-precision.
3049        let arrow = arrow_array::Time64NanosecondArray::from(vec![Some(2_123_456_789)]);
3050        assert_eq!(
3051            TimeArray::try_from(&arrow).unwrap(),
3052            TimeArray::from_iter([Time::with_secs_nano(2, 123_456_000).ok()])
3053        );
3054    }
3055
3056    #[test]
3057    fn time_arrow_units_from_arrow() {
3058        use std::sync::Arc;
3059
3060        struct Dummy;
3061        impl FromArrow for Dummy {}
3062
3063        for arrow_type in [
3064            ArrowType::Time32(arrow_schema::TimeUnit::Second),
3065            ArrowType::Time32(arrow_schema::TimeUnit::Millisecond),
3066            ArrowType::Time64(arrow_schema::TimeUnit::Microsecond),
3067            ArrowType::Time64(arrow_schema::TimeUnit::Nanosecond),
3068        ] {
3069            let field = ArrowField::new("t", arrow_type, true);
3070            assert!(is_parquet_schema_match_source_schema(
3071                field.data_type(),
3072                &RwType::Time
3073            ));
3074            assert_eq!(Dummy.from_field(&field).unwrap(), RwType::Time);
3075        }
3076
3077        let cases: Vec<(arrow_array::ArrayRef, TimeArray)> = vec![
3078            (
3079                Arc::new(arrow_array::Time32SecondArray::from(vec![
3080                    None,
3081                    Some(1),
3082                    Some(2),
3083                ])),
3084                TimeArray::from_iter([
3085                    None,
3086                    Time::with_secs_nano(1, 0).ok(),
3087                    Time::with_secs_nano(2, 0).ok(),
3088                ]),
3089            ),
3090            (
3091                Arc::new(arrow_array::Time32MillisecondArray::from(vec![
3092                    None,
3093                    Some(1_000),
3094                    Some(2_123),
3095                ])),
3096                TimeArray::from_iter([
3097                    None,
3098                    Time::with_secs_nano(1, 0).ok(),
3099                    Time::with_secs_nano(2, 123_000_000).ok(),
3100                ]),
3101            ),
3102            (
3103                Arc::new(arrow_array::Time64MicrosecondArray::from(vec![
3104                    None,
3105                    Some(1_000_000),
3106                    Some(2_123_456),
3107                ])),
3108                TimeArray::from_iter([
3109                    None,
3110                    Time::with_secs_nano(1, 0).ok(),
3111                    Time::with_secs_nano(2, 123_456_000).ok(),
3112                ]),
3113            ),
3114            (
3115                Arc::new(arrow_array::Time64NanosecondArray::from(vec![
3116                    None,
3117                    Some(1_000_000_000),
3118                    Some(2_123_456_789),
3119                ])),
3120                TimeArray::from_iter([
3121                    None,
3122                    Time::with_secs_nano(1, 0).ok(),
3123                    Time::with_secs_nano(2, 123_456_000).ok(),
3124                ]),
3125            ),
3126        ];
3127
3128        for (array, expected) in cases {
3129            let field = ArrowField::new("t", array.data_type().clone(), true);
3130            let ArrayImpl::Time(actual) = Dummy.from_array(&field, &array).unwrap() else {
3131                panic!("expected RW TimeArray");
3132            };
3133            assert_eq!(actual, expected);
3134        }
3135    }
3136
3137    #[test]
3138    fn time_arrow_units_reject_invalid_values() {
3139        use std::sync::Arc;
3140
3141        struct Dummy;
3142        impl FromArrow for Dummy {}
3143
3144        let cases: Vec<arrow_array::ArrayRef> = vec![
3145            Arc::new(arrow_array::Time32SecondArray::from(vec![Some(-1)])),
3146            Arc::new(arrow_array::Time32SecondArray::from(vec![Some(86_400)])),
3147            Arc::new(arrow_array::Time32MillisecondArray::from(vec![Some(-1)])),
3148            Arc::new(arrow_array::Time32MillisecondArray::from(vec![Some(
3149                86_400_000,
3150            )])),
3151            Arc::new(arrow_array::Time64MicrosecondArray::from(vec![Some(-1)])),
3152            Arc::new(arrow_array::Time64MicrosecondArray::from(vec![Some(
3153                86_400_000_000,
3154            )])),
3155            Arc::new(arrow_array::Time64NanosecondArray::from(vec![Some(-1)])),
3156            Arc::new(arrow_array::Time64NanosecondArray::from(vec![Some(
3157                86_400_000_000_000,
3158            )])),
3159            // Seconds counts at a multiple of 2^32 used to wrap into the valid range.
3160            Arc::new(arrow_array::Time64MicrosecondArray::from(vec![Some(
3161                4_294_967_296_000_000,
3162            )])),
3163            Arc::new(arrow_array::Time64NanosecondArray::from(vec![Some(
3164                4_294_967_296_000_000_000,
3165            )])),
3166        ];
3167
3168        for array in cases {
3169            let field = ArrowField::new("t", array.data_type().clone(), true);
3170            assert_from_arrow_error(Dummy.from_array(&field, &array));
3171        }
3172    }
3173
3174    #[test]
3175    fn timestamp() {
3176        let array =
3177            TimestampArray::from_iter([None, Timestamp::with_micros(123456789012345678).ok()]);
3178        let arrow = arrow_array::TimestampMicrosecondArray::try_from(&array).unwrap();
3179        assert_eq!(TimestampArray::try_from(&arrow).unwrap(), array);
3180    }
3181
3182    #[test]
3183    fn timestamp_nanosecond_export_rejects_out_of_range_values() {
3184        // Beyond ±year 2262, the value does not fit in `i64` nanoseconds.
3185        let timestamp_array =
3186            TimestampArray::from_iter([Timestamp::with_micros(9_300_000_000_000_000).ok()]);
3187        let err = arrow_array::TimestampNanosecondArray::try_from(&timestamp_array).unwrap_err();
3188        assert!(matches!(err, ArrayError::ToArrow(_)), "got {err:?}");
3189
3190        let timestamptz_array =
3191            TimestamptzArray::from_iter([Timestamptz::from_micros(9_300_000_000_000_000).unwrap()]);
3192        let err = arrow_array::TimestampNanosecondArray::try_from(&timestamptz_array).unwrap_err();
3193        assert!(matches!(err, ArrayError::ToArrow(_)), "got {err:?}");
3194    }
3195
3196    #[test]
3197    fn timestamp_arrow_units_reject_invalid_values() {
3198        let invalid_second = arrow_array::TimestampSecondArray::from(vec![Some(i64::MAX)]);
3199        let invalid_millisecond =
3200            arrow_array::TimestampMillisecondArray::from(vec![Some(i64::MAX)]);
3201        let invalid_microsecond =
3202            arrow_array::TimestampMicrosecondArray::from(vec![Some(i64::MAX)]);
3203
3204        assert_from_arrow_error(TimestampArray::try_from(&invalid_second));
3205        assert_from_arrow_error(TimestampArray::try_from(&invalid_millisecond));
3206        assert_from_arrow_error(TimestampArray::try_from(&invalid_microsecond));
3207    }
3208
3209    #[test]
3210    fn timestamptz_arrow_units_reject_invalid_values() {
3211        let invalid_second = arrow_array::TimestampSecondArray::from(vec![Some(i64::MAX)]);
3212        let invalid_millisecond =
3213            arrow_array::TimestampMillisecondArray::from(vec![Some(i64::MAX)]);
3214        let invalid_microsecond =
3215            arrow_array::TimestampMicrosecondArray::from(vec![Some(i64::MAX)]);
3216        // Within `checked_mul` range, but past what chrono can represent.
3217        let unrepresentable_second =
3218            arrow_array::TimestampSecondArray::from(vec![Some(9_000_000_000_000)]);
3219
3220        assert_from_arrow_error(TimestamptzArray::try_from(&invalid_second));
3221        assert_from_arrow_error(TimestamptzArray::try_from(&invalid_millisecond));
3222        assert_from_arrow_error(TimestamptzArray::try_from(&invalid_microsecond));
3223        assert_from_arrow_error(TimestamptzArray::try_from(&unrepresentable_second));
3224    }
3225
3226    #[test]
3227    fn date_rejects_out_of_range_arrow_value() {
3228        let valid = arrow_array::Date32Array::from(vec![Some(0), None]);
3229        assert_eq!(
3230            DateArray::try_from(&valid).unwrap(),
3231            DateArray::from_iter([Date::with_days_since_unix_epoch(0).ok(), None])
3232        );
3233
3234        // chrono's `NaiveDate` tops out well below `i32::MAX` days from the epoch.
3235        let out_of_range = arrow_array::Date32Array::from(vec![Some(1_000_000_000)]);
3236        assert_from_arrow_error(DateArray::try_from(&out_of_range));
3237    }
3238
3239    #[test]
3240    fn interval_rejects_out_of_range_microseconds() {
3241        let array = IntervalArray::from_iter([Interval::from_month_day_usec(0, 0, i64::MAX)]);
3242        let err = arrow_array::IntervalMonthDayNanoArray::try_from(&array).unwrap_err();
3243        assert!(matches!(err, ArrayError::ToArrow(_)), "got {err:?}");
3244    }
3245
3246    #[test]
3247    fn interval_truncates_sub_microsecond_nanos() {
3248        let arrow = arrow_array::IntervalMonthDayNanoArray::from(vec![
3249            Some(ArrowIntervalType::new(0, 0, 1_999)),
3250            Some(ArrowIntervalType::new(0, 0, -1)),
3251            Some(ArrowIntervalType::new(0, 0, -1_999)),
3252        ]);
3253        assert_eq!(
3254            IntervalArray::try_from(&arrow).unwrap(),
3255            IntervalArray::from_iter([
3256                Interval::from_month_day_usec(0, 0, 1),
3257                Interval::from_month_day_usec(0, 0, 0),
3258                Interval::from_month_day_usec(0, 0, -1),
3259            ])
3260        );
3261    }
3262
3263    fn assert_from_arrow_error<T>(result: Result<T, ArrayError>) {
3264        match result {
3265            Err(ArrayError::FromArrow(_)) => {}
3266            Err(err) => panic!("expected FromArrow error, got {err:?}"),
3267            Ok(_) => panic!("expected FromArrow error, got Ok"),
3268        }
3269    }
3270
3271    #[test]
3272    fn interval() {
3273        let array = IntervalArray::from_iter([
3274            None,
3275            Some(Interval::from_month_day_usec(
3276                1_000_000,
3277                1_000,
3278                1_000_000_000,
3279            )),
3280            Some(Interval::from_month_day_usec(
3281                -1_000_000,
3282                -1_000,
3283                -1_000_000_000,
3284            )),
3285        ]);
3286        let arrow = arrow_array::IntervalMonthDayNanoArray::try_from(&array).unwrap();
3287        assert_eq!(IntervalArray::try_from(&arrow).unwrap(), array);
3288    }
3289
3290    #[test]
3291    fn string() {
3292        let array = Utf8Array::from_iter([None, Some("array"), Some("arrow")]);
3293        let arrow = arrow_array::StringArray::from(&array);
3294        assert_eq!(Utf8Array::from(&arrow), array);
3295    }
3296
3297    #[test]
3298    fn binary() {
3299        let array = BytesArray::from_iter([None, Some("array".as_bytes())]);
3300        let arrow = arrow_array::BinaryArray::from(&array);
3301        assert_eq!(BytesArray::from(&arrow), array);
3302    }
3303
3304    #[test]
3305    fn fixed_size_binary() {
3306        let uuid = [
3307            0x12, 0x34, 0x56, 0x78, 0x9a, 0xbc, 0xde, 0xf0, 0x12, 0x34, 0x56, 0x78, 0x9a, 0xbc,
3308            0xde, 0xf0,
3309        ];
3310        let arrow_array = arrow_array::FixedSizeBinaryArray::try_from_sparse_iter_with_size(
3311            [None, Some(uuid)].into_iter(),
3312            16,
3313        )
3314        .unwrap();
3315        let field =
3316            arrow_schema::Field::new("u", arrow_schema::DataType::FixedSizeBinary(16), true);
3317
3318        assert_eq!(Dummy.from_field(&field).unwrap(), DataType::Bytea);
3319
3320        let rw_array = Dummy
3321            .from_array(&field, &(Arc::new(arrow_array) as arrow_array::ArrayRef))
3322            .unwrap();
3323        let expected = BytesArray::from_iter([None, Some(uuid.as_slice())]);
3324        assert_eq!(rw_array.as_bytea(), &expected);
3325    }
3326
3327    #[test]
3328    fn decimal() {
3329        let array = DecimalArray::from_iter([
3330            None,
3331            Some(Decimal::NaN),
3332            Some(Decimal::PositiveInf),
3333            Some(Decimal::NegativeInf),
3334            Some(Decimal::Normalized("123.4".parse().unwrap())),
3335            Some(Decimal::Normalized("123.456".parse().unwrap())),
3336        ]);
3337        let arrow = arrow_array::LargeBinaryArray::from(&array);
3338        assert_eq!(DecimalArray::try_from(&arrow).unwrap(), array);
3339
3340        let arrow = arrow_array::StringArray::from(&array);
3341        assert_eq!(DecimalArray::try_from(&arrow).unwrap(), array);
3342    }
3343
3344    #[test]
3345    fn jsonb() {
3346        let array = JsonbArray::from_iter([
3347            None,
3348            Some("null".parse().unwrap()),
3349            Some("false".parse().unwrap()),
3350            Some("1".parse().unwrap()),
3351            Some("[1, 2, 3]".parse().unwrap()),
3352            Some(r#"{ "a": 1, "b": null }"#.parse().unwrap()),
3353        ]);
3354        let arrow = arrow_array::LargeStringArray::from(&array);
3355        assert_eq!(JsonbArray::try_from(&arrow).unwrap(), array);
3356
3357        let arrow = arrow_array::StringArray::from(&array);
3358        assert_eq!(JsonbArray::try_from(&arrow).unwrap(), array);
3359    }
3360
3361    #[test]
3362    fn int256() {
3363        let values = [
3364            None,
3365            Some(Int256::from(1)),
3366            Some(Int256::from(i64::MAX)),
3367            Some(Int256::from(i64::MAX) * Int256::from(i64::MAX)),
3368            Some(Int256::from(i64::MAX) * Int256::from(i64::MAX) * Int256::from(i64::MAX)),
3369            Some(
3370                Int256::from(i64::MAX)
3371                    * Int256::from(i64::MAX)
3372                    * Int256::from(i64::MAX)
3373                    * Int256::from(i64::MAX),
3374            ),
3375            Some(Int256::min_value()),
3376            Some(Int256::max_value()),
3377        ];
3378
3379        let array =
3380            Int256Array::from_iter(values.iter().map(|r| r.as_ref().map(|x| x.as_scalar_ref())));
3381        let arrow = arrow_array::Decimal256Array::from(&array);
3382        assert_eq!(Int256Array::from(&arrow), array);
3383    }
3384}