diff --git a/bindings/python/src/arrow_input.rs b/bindings/python/src/arrow_input.rs new file mode 100644 index 000000000..18cb54278 --- /dev/null +++ b/bindings/python/src/arrow_input.rs @@ -0,0 +1,350 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Preserve C Data offsets before Arrow Rust materializes sparse Union arrays. + +use std::sync::Arc; + +use arrow::array::{make_array, ArrayData, ArrayRef, RecordBatch, RecordBatchOptions}; +use arrow::datatypes::{DataType, Schema, UnionMode}; +use arrow::error::ArrowError; +use arrow::pyarrow::FromPyArrow; +use pyo3::exceptions::PyValueError; +use pyo3::prelude::*; + +fn contains_sparse_union(data_type: &DataType) -> bool { + match data_type { + DataType::Union(_, UnionMode::Sparse) => true, + DataType::Union(fields, _) => fields + .iter() + .any(|(_, f)| contains_sparse_union(f.data_type())), + DataType::Struct(fields) => fields.iter().any(|f| contains_sparse_union(f.data_type())), + DataType::List(f) + | DataType::LargeList(f) + | DataType::FixedSizeList(f, _) + | DataType::ListView(f) + | DataType::LargeListView(f) + | DataType::Map(f, _) => contains_sparse_union(f.data_type()), + DataType::Dictionary(_, values) => contains_sparse_union(values), + DataType::RunEndEncoded(_, values) => contains_sparse_union(values.data_type()), + _ => false, + } +} + +pub(crate) fn array_from_pyarrow(value: &Bound<'_, PyAny>) -> PyResult { + let mut data = ArrayData::from_pyarrow_bound(value)?; + // Foreign buffers may be unaligned; normalize only after making them + // valid for Arrow Rust's typed access and ArrayData builder validation. + data.align_buffers(); + let data = normalize_sparse_unions(data).map_err(|e| PyValueError::new_err(e.to_string()))?; + Ok(make_array(data)) +} + +pub(crate) fn batch_from_pyarrow(value: &Bound<'_, PyAny>) -> PyResult { + let schema = Arc::new(Schema::from_pyarrow_bound(&value.getattr("schema")?)?); + if !schema + .fields() + .iter() + .any(|f| contains_sparse_union(f.data_type())) + { + return RecordBatch::from_pyarrow_bound(value); + } + // The standard batch importer materializes its children before we can fix + // Union offsets. Import each column's C Data representation first instead. + let columns = value + .getattr("columns")? + .try_iter()? + .map(|column| array_from_pyarrow(&column?)) + .collect::>>()?; + let rows = value.getattr("num_rows")?.extract::()?; + RecordBatch::try_new_with_options( + schema, + columns, + &RecordBatchOptions::default().with_row_count(Some(rows)), + ) + .map_err(|e| PyValueError::new_err(e.to_string())) +} + +fn normalize_sparse_unions(data: ArrayData) -> Result { + if !contains_sparse_union(data.data_type()) { + return Ok(data); + } + normalize_imported_data(data) +} + +fn normalize_imported_data(data: ArrayData) -> Result { + let len = data.len(); + normalize_imported_slice(data, 0, len) +} + +fn normalize_imported_slice( + data: ArrayData, + relative_offset: usize, + len: usize, +) -> Result { + // Once a parent is sliced, normalize every descendant, including ordinary + // STRUCT siblings and Union children. Arrow's STRUCT materializer uses + // ArrayData::slice, which can otherwise apply nested offsets twice. + let (data_type, _, nulls, offset, mut buffers, children) = data.into_parts(); + let nulls = nulls.map(|n| n.slice(relative_offset, len)); + let mut offset = offset + relative_offset; + // Defer validation until both parent and children share their coordinates. + // An intermediate ArrayData slice would validate a sliced parent's NULLs + // against unsliced non-nullable children, or propagate STRUCT offsets twice. + let children = match &data_type { + DataType::Struct(_) | DataType::Union(_, UnionMode::Sparse) => { + if matches!(data_type, DataType::Union(_, UnionMode::Sparse)) { + // UnionArray::from(ArrayData) applies the parent offset to type + // IDs but not sparse children. Canonicalize both before that. + buffers[0] = buffers[0].slice_with_length(offset, len); + } + let children = children + .into_iter() + .map(|c| normalize_imported_slice(c, offset, len)) + .collect::, _>>()?; + offset = 0; + children + } + DataType::FixedSizeList(_, width) => { + let width = *width as usize; + let children = children + .into_iter() + .map(|c| normalize_imported_slice(c, offset * width, len * width)) + .collect::, _>>()?; + offset = 0; + children + } + _ => children + .into_iter() + .map(normalize_imported_data) + .collect::, _>>()?, + }; + ArrayData::builder(data_type) + .len(len) + .offset(offset) + .nulls(nulls) + .buffers(buffers) + .child_data(children) + .build() +} + +#[cfg(test)] +mod tests { + use super::*; + use arrow::array::{ + Array, FixedSizeListArray, Int32Array, StringArray, StructArray, UnionArray, + }; + use arrow::buffer::{NullBuffer, ScalarBuffer}; + use arrow::datatypes::{Field, UnionFields}; + + fn input() -> ArrayData { + UnionArray::try_new( + UnionFields::try_new( + [7, 42], + [ + Field::new("n", DataType::Int32, true), + Field::new("s", DataType::Utf8, true), + ], + ) + .unwrap(), + ScalarBuffer::from(vec![42, 7, 7, 42]), + None, + vec![ + Arc::new(Int32Array::from(vec![0, 1, 2, 0])), + Arc::new(StringArray::from(vec!["hidden"; 4])), + ], + ) + .unwrap() + .to_data() + } + + fn assert_values(data: ArrayData) { + let array = UnionArray::from(data); + assert_eq!(array.type_ids().as_ref(), &[7, 7]); + assert_eq!( + array.child(7).to_data(), + Int32Array::from(vec![1, 2]).to_data() + ); + } + + #[test] + fn c_data_sparse_union_slice_rebases_children_and_type_ids() { + let raw = input().into_builder().offset(1).len(2).build().unwrap(); + assert_values(normalize_sparse_unions(raw).unwrap()); + } + + #[test] + fn c_data_parent_offsets_are_applied_before_union_materialization() { + let child = input(); + let rows = ArrayData::builder(DataType::Struct( + vec![Field::new("u", child.data_type().clone(), true)].into(), + )) + .offset(1) + .len(2) + .child_data(vec![child.clone()]) + .build() + .unwrap(); + let rows = StructArray::from(normalize_sparse_unions(rows).unwrap()); + assert_values(rows.column(0).to_data()); + // Nested STRUCT children are still in the original C Data coordinate + // system; the outer slice must not be propagated into them twice. + let nested = ArrayData::builder(DataType::Struct( + vec![Field::new("u", child.data_type().clone(), true)].into(), + )) + .len(4) + .child_data(vec![child.clone()]) + .build() + .unwrap(); + let outer = ArrayData::builder(DataType::Struct( + vec![Field::new("row", nested.data_type().clone(), true)].into(), + )) + .offset(1) + .len(2) + .child_data(vec![nested]) + .build() + .unwrap(); + let outer = StructArray::from(normalize_sparse_unions(outer).unwrap()); + let nested = outer + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + assert_values(nested.column(0).to_data()); + let lists = ArrayData::builder(DataType::FixedSizeList( + Arc::new(Field::new("item", child.data_type().clone(), true)), + 1, + )) + .offset(1) + .len(2) + .child_data(vec![child]) + .build() + .unwrap(); + let lists = FixedSizeListArray::from(normalize_sparse_unions(lists).unwrap()); + assert_values(lists.values().to_data()); + } + + #[test] + fn c_data_slices_normalize_ordinary_nested_union_children_and_siblings() { + let leaf = Int32Array::from(vec![0, 1, 2, 3]).to_data(); + let inner = ArrayData::builder(DataType::Struct( + vec![Field::new("b", DataType::Int32, true)].into(), + )) + .len(4) + .child_data(vec![leaf]) + .build() + .unwrap(); + let ordinary = ArrayData::builder(DataType::Struct( + vec![Field::new("a", inner.data_type().clone(), true)].into(), + )) + .len(4) + .child_data(vec![inner]) + .build() + .unwrap(); + let sibling = ArrayData::builder(DataType::Struct( + vec![ + Field::new("u", input().data_type().clone(), true), + Field::new("ordinary", ordinary.data_type().clone(), true), + ] + .into(), + )) + .len(2) + .offset(1) + .child_data(vec![input(), ordinary.clone()]) + .build() + .unwrap(); + let sibling = StructArray::from(normalize_sparse_unions(sibling).unwrap()); + let expected = Int32Array::from(vec![1, 2]).to_data(); + let assert_ordinary = |data: ArrayData| { + let outer = StructArray::from(data); + let inner = outer + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(inner.column(0).to_data(), expected); + }; + assert_ordinary(sibling.column(1).to_data()); + let union = ArrayData::builder(DataType::Union( + UnionFields::try_new( + [7], + [Field::new("ordinary", ordinary.data_type().clone(), true)], + ) + .unwrap(), + UnionMode::Sparse, + )) + .len(2) + .offset(1) + .buffers(vec![ScalarBuffer::from(vec![7_i8; 4]).into_inner()]) + .child_data(vec![ordinary]) + .build() + .unwrap(); + let union = UnionArray::from(normalize_sparse_unions(union).unwrap()); + assert_ordinary(union.child(7).to_data()); + } + + #[test] + fn c_data_nullability_is_validated_after_parent_and_children_are_sliced() { + let numbers = Int32Array::from(vec![None, Some(1), Some(2), None]); + let parents = NullBuffer::from(vec![false, true, true, false]); + let rows = ArrayData::builder(DataType::Struct( + vec![Field::new("n", DataType::Int32, false)].into(), + )) + .len(4) + .nulls(Some(parents.clone())) + .child_data(vec![numbers.to_data()]) + .build() + .unwrap(); + let lists = ArrayData::builder(DataType::FixedSizeList( + Arc::new(Field::new("item", DataType::Int32, false)), + 1, + )) + .len(4) + .nulls(Some(parents)) + .child_data(vec![numbers.to_data()]) + .build() + .unwrap(); + for ordinary in [rows, lists] { + // Both children are initially valid in their complete C Data + // coordinates. The nested parent's NULL masks must move together. + let input = ArrayData::builder(DataType::Struct( + vec![ + Field::new("u", input().data_type().clone(), true), + Field::new("ordinary", ordinary.data_type().clone(), true), + ] + .into(), + )) + .len(2) + .offset(1) + .child_data(vec![input(), ordinary]) + .build() + .unwrap(); + let result = StructArray::from(normalize_sparse_unions(input).unwrap()); + let ordinary = result.column(1); + assert_eq!(ordinary.null_count(), 0); + let values = if let Some(row) = ordinary.as_any().downcast_ref::() { + row.column(0) + } else { + ordinary + .as_any() + .downcast_ref::() + .unwrap() + .values() + }; + assert_eq!(values.to_data(), Int32Array::from(vec![1, 2]).to_data()); + } + } +} diff --git a/bindings/python/src/lib.rs b/bindings/python/src/lib.rs index 72d17d428..d579ee28a 100644 --- a/bindings/python/src/lib.rs +++ b/bindings/python/src/lib.rs @@ -17,6 +17,7 @@ use pyo3::prelude::*; +mod arrow_input; mod blob; mod blob_uri_reader; mod catalog_version; diff --git a/bindings/python/src/update_assignment.rs b/bindings/python/src/update_assignment.rs index d0cc3134e..290d95607 100644 --- a/bindings/python/src/update_assignment.rs +++ b/bindings/python/src/update_assignment.rs @@ -17,8 +17,8 @@ //! Python object and callback conversion for core update assignments. -use arrow::array::{make_array, ArrayData, ArrayRef}; -use arrow::pyarrow::{FromPyArrow, ToPyArrow}; +use arrow::array::ArrayRef; +use arrow::pyarrow::ToPyArrow; use paimon::spec::DataField; use paimon::table::UpdateAssignment; use pyo3::exceptions::PyValueError; @@ -41,12 +41,12 @@ fn bridge_error(error: PyErr, target: &CallbackError) -> paimon::Error { fn arrow_arrays(py: Python<'_>, value: &Bound<'_, PyAny>) -> PyResult> { let pa = py.import("pyarrow")?; if value.is_instance(&pa.getattr("Array")?)? { - Ok(vec![make_array(ArrayData::from_pyarrow_bound(value)?)]) + Ok(vec![crate::arrow_input::array_from_pyarrow(value)?]) } else if value.is_instance(&pa.getattr("ChunkedArray")?)? { value .getattr("chunks")? .try_iter()? - .map(|chunk| Ok(make_array(ArrayData::from_pyarrow_bound(&chunk?)?))) + .map(|chunk| crate::arrow_input::array_from_pyarrow(&chunk?)) .collect() } else { Err(PyValueError::new_err( @@ -121,7 +121,7 @@ pub(crate) fn from_python( (PyList::new(py, [scalar])?,), Some(&kwargs), )?; - Ok(make_array(ArrayData::from_pyarrow_bound(&one)?)) + crate::arrow_input::array_from_pyarrow(&one) }) }) .map_err(|cause| bridge_error(cause, &error)) diff --git a/bindings/python/src/write.rs b/bindings/python/src/write.rs index 312990d0a..f9a1716ab 100644 --- a/bindings/python/src/write.rs +++ b/bindings/python/src/write.rs @@ -540,7 +540,7 @@ fn arrow_table_batches(table: &Bound<'_, PyAny>) -> PyResult> { let mut batches = table .call_method0("to_batches")? .try_iter()? - .map(|batch| RecordBatch::from_pyarrow_bound(&batch?)) + .map(|batch| crate::arrow_input::batch_from_pyarrow(&batch?)) .collect::>>()?; // Preserve empty input schema so core validates columns even with no rows. if batches.is_empty() { @@ -559,7 +559,7 @@ fn arrow_upsert_batches(input: &Bound<'_, PyAny>) -> PyResult> } input .try_iter()? - .map(|batch| RecordBatch::from_pyarrow_bound(&batch?)) + .map(|batch| crate::arrow_input::batch_from_pyarrow(&batch?)) .collect() } diff --git a/crates/paimon/src/arrow/evolution_cast.rs b/crates/paimon/src/arrow/evolution_cast.rs index 5e85e1102..991932f0c 100644 --- a/crates/paimon/src/arrow/evolution_cast.rs +++ b/crates/paimon/src/arrow/evolution_cast.rs @@ -131,45 +131,56 @@ pub(crate) fn cast_numeric( let strings = super::evolution_decimal::decimal_to_string(source)?; return Ok(Some(super::apply_declared_width(&strings, to))); } - // DecimalUtils.doubleValue divides the compact long by its scale; - // only wide decimals use BigDecimal.doubleValue. FLOAT rounds that - // double rather than the decimal coefficient directly. - let values: Vec> = if decimal.precision() <= 18 { - let divisor = 10_f64.powi(decimal.scale() as i32); - source - .as_primitive::() - .iter() - .map(|value| value.map(|value| value as f64 / divisor)) - .collect() - } else { - let strings = super::evolution_decimal::decimal_to_string(source)?; - strings - .as_string::() - .iter() - .map(|value| { - value - .map(|value| { - value - .parse::() - .map_err(|_| invalid_numeric("Invalid DECIMAL value")) - }) - .transpose() - }) - .collect::>>()? - }; - return Ok(Some(match target { - ArrowType::Float32 => Arc::new(Float32Array::from_iter( - values.into_iter().map(|v| v.map(|v| v as f32)), - )) as ArrayRef, - ArrowType::Float64 => Arc::new(Float64Array::from(values)) as ArrayRef, - _ => unreachable!("validated DECIMAL numeric target"), - })); + return cast_decimal_to_floating(source, decimal.precision(), decimal.scale(), &target) + .map(Some); } arrow_cast::cast(source, &target) .map(Some) .map_err(|error| invalid_numeric(error.to_string())) } +/// Java DECIMAL conversion shared by schema reads and external update inputs. +pub(crate) fn cast_decimal_to_floating( + source: &ArrayRef, + precision: u32, + scale: u32, + target: &ArrowType, +) -> crate::Result { + // DecimalUtils.doubleValue divides the compact long by its scale; + // only wide decimals use BigDecimal.doubleValue. FLOAT rounds that + // double rather than the decimal coefficient directly. + let values: Vec> = if precision <= 18 { + let divisor = 10_f64.powi(scale as i32); + source + .as_primitive::() + .iter() + .map(|value| value.map(|value| value as f64 / divisor)) + .collect() + } else { + let strings = super::evolution_decimal::decimal_to_string(source)?; + strings + .as_string::() + .iter() + .map(|value| { + value + .map(|value| { + value + .parse::() + .map_err(|_| invalid_numeric("Invalid DECIMAL value")) + }) + .transpose() + }) + .collect::>>()? + }; + Ok(match target { + ArrowType::Float32 => Arc::new(Float32Array::from_iter( + values.into_iter().map(|v| v.map(|v| v as f32)), + )) as ArrayRef, + ArrowType::Float64 => Arc::new(Float64Array::from(values)) as ArrayRef, + _ => unreachable!("validated DECIMAL numeric target"), + }) +} + fn invalid_numeric(message: impl Into) -> crate::Error { crate::Error::DataInvalid { message: message.into(), diff --git a/crates/paimon/src/arrow/mod.rs b/crates/paimon/src/arrow/mod.rs index 7a057c0af..fddf9f193 100644 --- a/crates/paimon/src/arrow/mod.rs +++ b/crates/paimon/src/arrow/mod.rs @@ -26,7 +26,7 @@ mod cast_completion_java_test; mod cast_text; pub(crate) use cast_text::parse_timestamp as parse_java_timestamp; mod evolution_cast; -pub(crate) use evolution_cast::{cast_numeric, supports_numeric_cast}; +pub(crate) use evolution_cast::{cast_decimal_to_floating, cast_numeric, supports_numeric_cast}; mod evolution_decimal; mod evolution_string; mod evolution_temporal; diff --git a/crates/paimon/src/table/key_normalization.rs b/crates/paimon/src/table/key_normalization.rs index 3ea04dc59..50b78dc84 100644 --- a/crates/paimon/src/table/key_normalization.rs +++ b/crates/paimon/src/table/key_normalization.rs @@ -36,34 +36,6 @@ pub(super) fn normalize_match_key( return Ok(new_null_array(expected, decoded.len())); } let source = decoded.data_type(); - if source.is_decimal() && matches!(expected, DataType::Decimal128(..)) { - let source_scale = match source { - DataType::Decimal32(_, scale) - | DataType::Decimal64(_, scale) - | DataType::Decimal128(_, scale) - | DataType::Decimal256(_, scale) => *scale, - _ => unreachable!(), - }; - let DataType::Decimal128(_, target_scale) = expected else { - unreachable!() - }; - let difference = i16::from(*target_scale) - i16::from(source_scale); - let text_fallback = || { - // Arrow's decimal rescaler can reject a large scale difference - // before inspecting values, even for zero and NULL. Reuse the - // exact text parser as a fallback; it still rejects nonzero values - // that exceed precision or lose fractional digits. - let text = cast_update_value(&decoded, &DataType::Utf8, CastMode::Assignment)?; - cast_update_value(&text, expected, CastMode::Assignment) - }; - // Arrow computes scale differences in i8 and uses a Decimal128 power - // table. Skip that kernel before either can overflow. - return if !(-38..=38).contains(&difference) { - text_fallback() - } else { - cast_update_value(&decoded, expected, CastMode::Assignment).or_else(|_| text_fallback()) - }; - } let exact_family = source.is_integer() && expected.is_integer() || source.is_decimal() && expected.is_decimal() || matches!( diff --git a/crates/paimon/src/table/update_input.rs b/crates/paimon/src/table/update_input.rs index f6a809db9..38f1fccc9 100644 --- a/crates/paimon/src/table/update_input.rs +++ b/crates/paimon/src/table/update_input.rs @@ -28,6 +28,9 @@ use arrow_array::{ use arrow_buffer::OffsetBuffer; use arrow_schema::{DataType, Schema, TimeUnit}; +mod decimal; +mod encoded; + /// PyPaimon distinguishes evaluated assignments (safe casts) from row-ID /// input (numeric coercion). Both use the same encodings, strings and nested /// traversal; callers select the contract explicitly. @@ -133,27 +136,37 @@ pub(super) fn cast_update_value( if mode == CastMode::RowUpdate { cast_values(array, target, CastMode::Assignment).or_else(|_| { validate_map_constructor_input(array, target)?; - let decoded = decode_dictionary(array)?; - // Python converts lists of pairs to tuples only for a top-level MAP. - if let DataType::Map(field, sorted) = target { - if matches!( - decoded.data_type(), - DataType::List(_) - | DataType::LargeList(_) - | DataType::ListView(_) - | DataType::LargeListView(_) - | DataType::FixedSizeList(..) - ) { - return constructor_map(&decoded, field, *sorted); - } - } - cast_values(array, target, CastMode::Constructor) + construct_update_value(array, target) }) } else { cast_values(array, target, mode) } } +/// The root MAP constructor converts list pairs to tuples. Preserve that +/// context through encoding wrappers, but not through ROW/ARRAY children. +fn construct_update_value(array: &ArrayRef, target: &DataType) -> crate::Result { + if matches!(array.data_type(), DataType::Dictionary(..)) { + return encoded::construct_dictionary(array, target, construct_update_value); + } + if matches!(array.data_type(), DataType::Union(..)) { + return encoded::construct_union(array, target, construct_update_value); + } + if let DataType::Map(field, sorted) = target { + if matches!( + array.data_type(), + DataType::List(_) + | DataType::LargeList(_) + | DataType::ListView(_) + | DataType::LargeListView(_) + | DataType::FixedSizeList(..) + ) { + return constructor_map(array, field, *sorted); + } + } + cast_values(array, target, CastMode::Constructor) +} + /// Cast logical values recursively, so dictionary encoding and nesting cannot /// bypass primitive validation. The final outer cast only converts the layout. fn cast_values(array: &ArrayRef, target: &DataType, mode: CastMode) -> crate::Result { @@ -164,8 +177,18 @@ fn cast_values(array: &ArrayRef, target: &DataType, mode: CastMode) -> crate::Re return Ok(array.clone()); } if matches!(array.data_type(), DataType::Dictionary(..)) { + if mode == CastMode::Constructor { + return encoded::construct_dictionary(array, target, |array, target| { + cast_values(array, target, CastMode::Constructor) + }); + } return cast_values(&decode_dictionary(array)?, target, mode); } + if mode == CastMode::Constructor && matches!(array.data_type(), DataType::Union(..)) { + return encoded::construct_union(array, target, |array, target| { + cast_values(array, target, CastMode::Constructor) + }); + } if let DataType::ListView(field) | DataType::LargeListView(field) = array.data_type() { // Convert only the offsets/layout first. Casting children here would // bypass the recursive precision and formatting rules below. @@ -209,10 +232,7 @@ fn cast_values(array: &ArrayRef, target: &DataType, mode: CastMode) -> crate::Re let columns = fields .iter() .map(|field| match input.column_by_name(field.name()) { - Some(column) => { - let column = constructor_child(column, input.nulls(), 1, mode)?; - cast_values(&column, field.data_type(), mode) - } + Some(column) => cast_child(column, field.data_type(), input.nulls(), 1, mode), None => Ok(new_null_array(field.data_type(), input.len())), }) .collect::>>()?; @@ -242,9 +262,13 @@ fn cast_values(array: &ArrayRef, target: &DataType, mode: CastMode) -> crate::Re )?), DataType::FixedSizeList(_, size) => { let input = array.as_any().downcast_ref::().unwrap(); - let values = - constructor_child(input.values(), input.nulls(), *size as usize, mode)?; - let values = cast_values(&values, field.data_type(), mode)?; + let values = cast_child( + input.values(), + field.data_type(), + input.nulls(), + *size as usize, + mode, + )?; Some(Arc::new( FixedSizeListArray::try_new_with_length( field.clone(), @@ -256,35 +280,32 @@ fn cast_values(array: &ArrayRef, target: &DataType, mode: CastMode) -> crate::Re .map_err(|error| invalid(error.to_string()))?, )) } + DataType::Map(..) => { + let input = array.as_any().downcast_ref::().unwrap(); + let DataType::Struct(fields) = field.data_type() else { + return Err(invalid( + "Map entries can only be cast to two-field ROW elements", + )); + }; + let (entries, offsets) = cast_map_entries(input, fields, mode)?; + Some(Arc::new( + ListArray::try_new( + field.clone(), + offsets, + Arc::new(entries), + input.nulls().cloned(), + ) + .map_err(|error| invalid(error.to_string()))?, + )) + } _ => None, }, (DataType::Map(_, _), DataType::Map(field, sorted)) => { let input = array.as_any().downcast_ref::().unwrap(); - let offsets = input.value_offsets(); - let start = offsets[0]; - let length = offsets[offsets.len() - 1] - start; - let (offsets, indices) = constructor_offsets(offsets, input.nulls(), mode); - let length = indices.as_ref().map_or(length as usize, |i| i.len()); let DataType::Struct(fields) = field.data_type() else { return Err(invalid("Map entries must have a struct type")); }; - // Map fields may use custom names. Key/value roles are positional, - // unlike ROW fields, so never match these two children by name. - let columns = [input.keys(), input.values()] - .into_iter() - .zip(fields) - .map(|(values, field)| { - let values = match &indices { - Some(indices) => arrow_select::take::take(values.as_ref(), indices, None) - .map_err(|error| invalid(error.to_string()))?, - None => values.slice(start as usize, length), - }; - cast_values(&values, field.data_type(), mode) - }) - .collect::>>()?; - let entries = StructArray::try_new_with_length(fields.clone(), columns, None, length) - .map_err(|error| invalid(error.to_string()))?; - let offsets = OffsetBuffer::new(offsets.into()); + let (entries, offsets) = cast_map_entries(input, fields, mode)?; Some(Arc::new( MapArray::try_new( field.clone(), @@ -323,6 +344,41 @@ fn cast_values(array: &ArrayRef, target: &DataType, mode: CastMode) -> crate::Re cast_primitive(array, target, mode) } +/// Arrow represents a MAP as a list of key/value ROWs. Adapt these children +/// positionally for both MAP and ARRAY inputs, retaining recursive value +/// conversion and the whole-column constructor's visibility rules. +fn cast_map_entries( + input: &MapArray, + fields: &arrow_schema::Fields, + mode: CastMode, +) -> crate::Result<(StructArray, OffsetBuffer)> { + if fields.len() != 2 { + return Err(invalid("Map entries require exactly two ROW fields")); + } + let offsets = input.value_offsets(); + let start = offsets[0]; + let length = offsets[offsets.len() - 1] - start; + let (offsets, indices) = constructor_offsets(offsets, input.nulls(), mode); + let length = indices.as_ref().map_or(length as usize, |i| i.len()); + // MAP field names are aliases. Their key/value roles do not depend on the + // source names or the target ROW names, unlike a normal ROW-to-ROW cast. + let columns = [input.keys(), input.values()] + .into_iter() + .zip(fields) + .map(|(values, field)| { + let values = match &indices { + Some(indices) => arrow_select::take::take(values.as_ref(), indices, None) + .map_err(|error| invalid(error.to_string()))?, + None => values.slice(start as usize, length), + }; + cast_values(&values, field.data_type(), mode) + }) + .collect::>>()?; + let entries = StructArray::try_new_with_length(fields.clone(), columns, None, length) + .map_err(|error| invalid(error.to_string()))?; + Ok((entries, OffsetBuffer::new(offsets.into()))) +} + // PyPaimon's row-ID fallback reconstructs Python dicts only after a whole-column // safe cast fails. At the top level it rejects ambiguous schema-less null keys. fn validate_map_constructor_input(array: &ArrayRef, target: &DataType) -> crate::Result<()> { @@ -406,13 +462,12 @@ fn constructor_struct_from_map( .collect::>>()?; // Missing fields, NULL parents and trailing entries do not participate // in conversion, matching PyArrow's ordered pair constructor for ROW. - let values = arrow_select::take::take( - input.values().as_ref(), + encoded::construct_selected( + input.values(), &arrow_array::UInt64Array::from(indices), - None, + field.data_type(), + |array, target| cast_values(array, target, CastMode::Constructor), ) - .map_err(|error| invalid(error.to_string()))?; - cast_values(&values, field.data_type(), CastMode::Constructor) }) .collect::>>()?; Ok(Arc::new( @@ -451,8 +506,13 @@ fn constructor_map( .columns() .iter() .map(|column| { - let visible = constructor_child(column, input.nulls(), 1, CastMode::Constructor)?; - cast_values(&visible, fields[1].data_type(), CastMode::Constructor) + cast_child( + column, + fields[1].data_type(), + input.nulls(), + 1, + CastMode::Constructor, + ) }) .collect::>>()?; let values = if columns.is_empty() { @@ -580,21 +640,23 @@ fn cast_list( // Rebuilding Python values discards children hidden by a null parent. Safe // casts deliberately retain them, since Arrow C++ validates their values too. -fn constructor_child( +fn cast_child( array: &ArrayRef, + target: &DataType, nulls: Option<&arrow_buffer::NullBuffer>, width: usize, mode: CastMode, ) -> crate::Result { if mode != CastMode::Constructor || nulls.is_none_or(|n| n.null_count() == 0) { - return Ok(array.clone()); + return cast_values(array, target, mode); } let nulls = nulls.unwrap(); let indices = arrow_array::UInt64Array::from_iter( (0..array.len()).map(|i| nulls.is_valid(i / width).then_some(i as u64)), ); - arrow_select::take::take(array.as_ref(), &indices, None) - .map_err(|error| invalid(error.to_string())) + encoded::construct_selected(array, &indices, target, |array, target| { + cast_values(array, target, CastMode::Constructor) + }) } fn constructor_offsets( @@ -706,7 +768,22 @@ fn cast_to_string(array: &ArrayRef, target: &DataType) -> crate::Result = match array.data_type() { + DataType::Float16 => Some( + array + .as_any() + .downcast_ref::() + .unwrap() + .iter() + .map(|value| value.map(|value| format_float(value.to_f64()))) + .collect(), + ), DataType::Float32 => Some( array .as_any() @@ -1365,6 +1442,12 @@ fn cast_primitive(array: &ArrayRef, target: &DataType, mode: CastMode) -> crate: return cast_float_decimal(array, *precision, *scale); } } + if source.is_decimal() && matches!(target, DataType::Decimal128(..)) { + return decimal::cast_exact_decimal(array, target); + } + if source.is_decimal() && matches!(target, DataType::Float32 | DataType::Float64) { + return decimal::cast_to_floating(array, target); + } if mode == CastMode::Constructor { if let Some(converted) = coerce_temporal(array, target)? { return Ok(converted); @@ -1467,6 +1550,12 @@ fn cast_primitive(array: &ArrayRef, target: &DataType, mode: CastMode) -> crate: Ok(converted) } +#[cfg(test)] +mod container_tests; + +#[cfg(test)] +mod scalar_tests; + #[cfg(test)] mod tests { use super::*; diff --git a/crates/paimon/src/table/update_input/container_tests.rs b/crates/paimon/src/table/update_input/container_tests.rs new file mode 100644 index 000000000..85dd4186f --- /dev/null +++ b/crates/paimon/src/table/update_input/container_tests.rs @@ -0,0 +1,240 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use super::*; +use arrow_array::{types::Int8Type, DictionaryArray, Int32Array, Int8Array}; +use arrow_buffer::NullBuffer; +use arrow_schema::Field; + +fn map(keys: ArrayRef, values: ArrayRef, lengths: &[usize], valid: &[bool]) -> ArrayRef { + let entries = StructArray::from(vec![ + ( + Arc::new(Field::new("encoded_key", keys.data_type().clone(), false)), + keys, + ), + ( + Arc::new(Field::new( + "encoded_value", + values.data_type().clone(), + true, + )), + values, + ), + ]); + Arc::new(MapArray::new( + Arc::new(Field::new("pairs", entries.data_type().clone(), false)), + OffsetBuffer::from_lengths(lengths.iter().copied()), + entries, + Some(NullBuffer::from(valid.to_vec())), + false, + )) +} + +fn pair_type(key: DataType, value: DataType) -> DataType { + DataType::Struct( + vec![ + Field::new("label", key, true), + Field::new("amount", value, true), + ] + .into(), + ) +} + +#[test] +fn map_to_array_rows_preserves_aliases_slices_empty_and_null_values() { + let input = map( + Arc::new(StringArray::from(vec!["unused", "a", "b", "z"])), + Arc::new(Int32Array::from(vec![Some(-1), Some(1), None, Some(4)])), + &[1, 2, 0, 0, 1], + &[true, true, false, true, true], + ) + .slice(1, 4); + let fields = pair_type(DataType::Utf8, DataType::Int64); + let item = Arc::new(Field::new("element", fields.clone(), false)); + for target in [ + DataType::List(item.clone()), + DataType::LargeList(item.clone()), + ] { + let output = cast_update_value(&input, &target, CastMode::Assignment).unwrap(); + assert_eq!(output.data_type(), &target); + assert!(output.is_null(1)); + // Normalize only the container to inspect both offset widths. + let output = arrow_cast::cast(output.as_ref(), &DataType::List(item.clone())).unwrap(); + let output = output.as_any().downcast_ref::().unwrap(); + assert_eq!(output.value_offsets(), &[0, 2, 2, 2, 3]); + let entries = output + .values() + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(entries.data_type(), &fields); + assert_eq!( + entries.column(0).to_data(), + StringArray::from(vec!["a", "b", "z"]).to_data() + ); + assert_eq!( + entries.column(1).to_data(), + Int64Array::from(vec![Some(1), None, Some(4)]).to_data() + ); + } +} + +#[test] +fn map_to_array_recurses_through_rows_lists_and_dictionaries() { + let dictionary: ArrayRef = Arc::new( + DictionaryArray::::try_new( + Int8Array::from(vec![0, 1]), + Arc::new(StringArray::from(vec!["42", "7"])), + ) + .unwrap(), + ); + let input = map( + Arc::new(StringArray::from(vec!["a", "b"])), + dictionary, + &[1, 1], + &[true, true], + ); + let target = DataType::List(Arc::new(Field::new( + "item", + pair_type(DataType::Utf8, DataType::Int32), + true, + ))); + let expected = cast_update_value(&input, &target, CastMode::Assignment).unwrap(); + let outer_row: ArrayRef = Arc::new(StructArray::from(vec![( + Arc::new(Field::new("nested", input.data_type().clone(), true)), + input.clone(), + )])); + let result = cast_update_value( + &outer_row, + &DataType::Struct(vec![Field::new("nested", target.clone(), true)].into()), + CastMode::Assignment, + ) + .unwrap(); + assert_eq!( + result + .as_any() + .downcast_ref::() + .unwrap() + .column(0) + .to_data(), + expected.to_data() + ); + let outer_list: ArrayRef = Arc::new(ListArray::new( + Arc::new(Field::new("item", input.data_type().clone(), true)), + OffsetBuffer::from_lengths([2]), + input.clone(), + None, + )); + let result = cast_update_value( + &outer_list, + &DataType::List(Arc::new(Field::new("item", target.clone(), true))), + CastMode::Assignment, + ) + .unwrap(); + assert_eq!( + result + .as_any() + .downcast_ref::() + .unwrap() + .values() + .to_data(), + expected.to_data() + ); + let encoded: ArrayRef = + Arc::new(DictionaryArray::::try_new(Int8Array::from(vec![0, 1]), input).unwrap()); + assert_eq!( + cast_update_value(&encoded, &target, CastMode::Assignment) + .unwrap() + .to_data(), + expected.to_data() + ); +} + +#[test] +fn map_to_array_keeps_whole_column_constructor_visibility() { + let input = map( + Arc::new(StringArray::from(vec!["a", "hidden"])), + Arc::new(Int64Array::from(vec![7, i64::MAX])), + &[1, 1, 0], + &[true, false, true], + ); + let target = DataType::List(Arc::new(Field::new( + "item", + pair_type(DataType::Utf8, DataType::Int32), + true, + ))); + assert!(cast_update_value(&input, &target, CastMode::Assignment).is_err()); + let output = cast_update_value(&input, &target, CastMode::RowUpdate).unwrap(); + let output = output.as_any().downcast_ref::().unwrap(); + assert_eq!(output.value_offsets(), &[0, 1, 1, 1]); + assert!(output.is_null(1)); + let entries = output + .values() + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!( + entries.column(1).to_data(), + Int32Array::from(vec![7]).to_data() + ); + let visible = map( + Arc::new(StringArray::from(vec!["bad"])), + Arc::new(Int64Array::from(vec![i64::MAX])), + &[1], + &[true], + ); + assert!(cast_update_value(&visible, &target, CastMode::RowUpdate).is_err()); +} + +#[test] +fn map_entries_require_two_fields_and_actual_child_nullability() { + let input = map( + Arc::new(StringArray::from(vec!["a"])), + Arc::new(Int32Array::from(vec![7])), + &[1], + &[true], + ); + for size in [0, 1, 3] { + let fields = (0..size) + .map(|i| Field::new(format!("f{i}"), DataType::Utf8, true)) + .collect::>(); + let target = DataType::List(Arc::new(Field::new( + "item", + DataType::Struct(fields.into()), + true, + ))); + assert!(cast_update_value(&input, &target, CastMode::Assignment).is_err()); + } + let fields = vec![ + Field::new("k", DataType::Utf8, false), + Field::new("v", DataType::Int32, false), + ]; + let target = DataType::List(Arc::new(Field::new( + "item", + DataType::Struct(fields.into()), + false, + ))); + // A nullable source value FIELD with non-null actual values can be cast. + assert!(cast_update_value(&input, &target, CastMode::Assignment).is_ok()); + let null_value = map( + Arc::new(StringArray::from(vec!["a"])), + Arc::new(Int32Array::from(vec![None])), + &[1], + &[true], + ); + assert!(cast_update_value(&null_value, &target, CastMode::Assignment).is_err()); +} diff --git a/crates/paimon/src/table/update_input/decimal.rs b/crates/paimon/src/table/update_input/decimal.rs new file mode 100644 index 000000000..609d47e7c --- /dev/null +++ b/crates/paimon/src/table/update_input/decimal.rs @@ -0,0 +1,239 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Exact DECIMAL input rescaling shared by updates and logical match keys. + +use arrow_array::{cast::AsArray, ArrayRef, Decimal128Array, Float32Array, Float64Array}; +use arrow_schema::DataType; +use std::sync::Arc; + +use super::{invalid, parse_decimal_text}; + +fn precision_and_scale(array: &ArrayRef) -> (u8, i8) { + match array.data_type() { + DataType::Decimal32(p, s) + | DataType::Decimal64(p, s) + | DataType::Decimal128(p, s) + | DataType::Decimal256(p, s) => (*p, *s), + _ => unreachable!("validated DECIMAL input"), + } +} + +fn map_decimal( + array: &ArrayRef, + parse: impl Fn(&str) -> crate::Result, +) -> crate::Result>> { + let (_, scale) = precision_and_scale(array); + // Preserve the full coefficient. A generic display can clip digits to the + // declared source precision for foreign buffers containing more digits. + macro_rules! values { + ($ty:ty) => { + array + .as_primitive::<$ty>() + .iter() + .map(|value| { + value + .map(|value| parse(&format!("{value}e{}", -i16::from(scale)))) + .transpose() + }) + .collect() + }; + } + match array.data_type() { + DataType::Decimal32(..) => values!(arrow_array::types::Decimal32Type), + DataType::Decimal64(..) => values!(arrow_array::types::Decimal64Type), + DataType::Decimal128(..) => values!(arrow_array::types::Decimal128Type), + DataType::Decimal256(..) => values!(arrow_array::types::Decimal256Type), + _ => unreachable!("validated DECIMAL input"), + } +} + +pub(super) fn cast_to_floating(array: &ArrayRef, target: &DataType) -> crate::Result { + let (precision, scale) = precision_and_scale(array); + if precision <= 38 && scale >= 0 && scale as u8 <= precision { + let canonical = cast_exact_decimal(array, &DataType::Decimal128(precision, scale))?; + return crate::arrow::cast_decimal_to_floating( + &canonical, + u32::from(precision), + scale as u32, + target, + ); + } + // External Arrow decimals can exceed Paimon's logical precision or use a + // negative scale. They still update FLOAT/DOUBLE columns via the exact + // decimal value, without first rounding an oversized coefficient to f64. + let values = map_decimal(array, |text| { + text.parse::() + .map_err(|_| invalid("Invalid decimal floating assignment")) + })?; + Ok(match target { + DataType::Float32 => Arc::new(Float32Array::from_iter( + values.into_iter().map(|v| v.map(|v| v as f32)), + )) as ArrayRef, + DataType::Float64 => Arc::new(Float64Array::from(values)), + _ => unreachable!("validated DECIMAL floating target"), + }) +} + +pub(super) fn cast_exact_decimal(array: &ArrayRef, target: &DataType) -> crate::Result { + let (_, source_scale) = precision_and_scale(array); + let DataType::Decimal128(target_precision, target_scale) = target else { + unreachable!("validated DECIMAL target") + }; + let text_fallback = || { + // A zero or NULL can cross scales beyond the rescaler's power table. + // The exact text parser still rejects fractional loss and overflow. + let values = map_decimal(array, |text| { + parse_decimal_text(text, *target_precision, *target_scale) + })?; + Decimal128Array::from(values) + .with_precision_and_scale(*target_precision, *target_scale) + .map(|array| Arc::new(array) as ArrayRef) + .map_err(|error| invalid(error.to_string())) + }; + let difference = i16::from(*target_scale) - i16::from(source_scale); + if !(-38..=38).contains(&difference) { + // Arrow subtracts scales in i8 and uses a Decimal128 power table. + // Check before entering the kernel, including its reverse cast. + return text_fallback(); + } + let options = arrow_cast::CastOptions { + safe: false, + ..Default::default() + }; + let exact = || { + let converted = arrow_cast::cast_with_options(array.as_ref(), target, &options) + .map_err(|error| invalid(error.to_string()))?; + let restored = + arrow_cast::cast_with_options(converted.as_ref(), array.data_type(), &options) + .map_err(|error| invalid(error.to_string()))?; + if restored.to_data() != array.to_data() { + return Err(invalid("Decimal assignment would lose scale precision")); + } + Ok(converted) + }; + exact().or_else(|_| text_fallback()) +} + +#[cfg(test)] +mod tests { + use super::super::{cast_update_value, CastMode}; + use super::*; + use arrow_array::{Array, Decimal128Array, Decimal256Array, Decimal32Array, Decimal64Array}; + use arrow_buffer::i256; + use std::sync::Arc; + + fn decimal(data_type: &DataType, value: i128) -> ArrayRef { + match data_type { + DataType::Decimal32(p, s) => Arc::new( + Decimal32Array::from(vec![Some(value as i32), None]) + .with_precision_and_scale(*p, *s) + .unwrap(), + ), + DataType::Decimal64(p, s) => Arc::new( + Decimal64Array::from(vec![Some(value as i64), None]) + .with_precision_and_scale(*p, *s) + .unwrap(), + ), + DataType::Decimal128(p, s) => Arc::new( + Decimal128Array::from(vec![Some(value), None]) + .with_precision_and_scale(*p, *s) + .unwrap(), + ), + DataType::Decimal256(p, s) => Arc::new( + Decimal256Array::from(vec![Some(i256::from_i128(value)), None]) + .with_precision_and_scale(*p, *s) + .unwrap(), + ), + _ => unreachable!(), + } + } + + #[test] + fn zero_and_null_cross_scale_and_width_boundaries_without_overflow() { + for (source, target) in [ + (DataType::Decimal32(9, -128), DataType::Decimal128(38, 38)), + (DataType::Decimal64(18, -90), DataType::Decimal128(38, 38)), + (DataType::Decimal128(38, -2), DataType::Decimal128(38, 38)), + (DataType::Decimal128(38, 38), DataType::Decimal128(20, 0)), + (DataType::Decimal256(76, -128), DataType::Decimal128(20, 2)), + (DataType::Decimal256(60, -40), DataType::Decimal128(20, 2)), + (DataType::Decimal256(76, 76), DataType::Decimal128(38, 38)), + ] { + for mode in [CastMode::Assignment, CastMode::RowUpdate] { + let output = cast_update_value(&decimal(&source, 0), &target, mode).unwrap(); + assert_eq!( + output.to_data(), + decimal(&target, 0).to_data(), + "{source:?} -> {target:?}" + ); + assert!( + cast_update_value(&decimal(&source, 1), &target, mode).is_err(), + "{source:?} -> {target:?}" + ); + } + } + } + + #[test] + fn exact_rescaling_preserves_precision_and_nulls_for_every_input_width() { + for source in [ + DataType::Decimal32(9, 3), + DataType::Decimal64(18, 3), + DataType::Decimal128(38, 3), + DataType::Decimal256(76, 3), + ] { + let target = DataType::Decimal128(10, 2); + let output = + cast_update_value(&decimal(&source, -1250), &target, CastMode::Assignment).unwrap(); + assert_eq!(output.to_data(), decimal(&target, -125).to_data()); + assert!( + cast_update_value(&decimal(&source, -1251), &target, CastMode::Assignment).is_err() + ); + assert!(cast_update_value( + &decimal(&source, 1250), + &DataType::Decimal128(2, 2), + CastMode::Assignment + ) + .is_err()); + } + } + + #[test] + fn floating_input_keeps_coefficients_wider_than_decimal128() { + let coefficient = format!("1508079088755887087797858531882816082{}", "0".repeat(38)) + .parse::() + .unwrap(); + let source: ArrayRef = Arc::new( + Decimal256Array::from(vec![Some(coefficient), None, Some(-coefficient)]) + .with_precision_and_scale(76, 39) + .unwrap(), + ); + for mode in [CastMode::Assignment, CastMode::RowUpdate] { + let result = cast_update_value(&source, &DataType::Float64, mode).unwrap(); + assert_eq!( + result.to_data(), + Float64Array::from(vec![ + Some(1.508079088755887e35), + None, + Some(-1.508079088755887e35), + ]) + .to_data() + ); + } + } +} diff --git a/crates/paimon/src/table/update_input/encoded.rs b/crates/paimon/src/table/update_input/encoded.rs new file mode 100644 index 000000000..4bb99363a --- /dev/null +++ b/crates/paimon/src/table/update_input/encoded.rs @@ -0,0 +1,599 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Rebuild selected encoded values after the whole-column safe cast has failed. + +use arrow_array::cast::AsArray; +use arrow_array::{Array, ArrayRef, UInt64Array, UnionArray}; +use arrow_schema::DataType; +use indexmap::IndexMap; + +use super::invalid; + +/// Decode references before constructor conversion, then restore NULL keys. +/// Unlike taking a nullable index directly from a Union, taking from the typed +/// output can represent NULL without assigning it an unrelated active type. +pub(super) fn construct_dictionary( + array: &ArrayRef, + target: &DataType, + convert: fn(&ArrayRef, &DataType) -> crate::Result, +) -> crate::Result { + let dictionary = array.as_any_dictionary(); + let keys = arrow_cast::cast(dictionary.keys(), &DataType::UInt64) + .map_err(|error| invalid(error.to_string()))?; + let keys = keys.as_any().downcast_ref::().unwrap(); + construct_selected(dictionary.values(), keys, target, convert) +} + +/// Restore missing values only after conversion to a type with a validity +/// bitmap. Union inputs cannot represent a missing selection independently of +/// their active child, including below NULL parents or absent ROW fields. +pub(super) fn construct_selected( + array: &ArrayRef, + indices: &UInt64Array, + target: &DataType, + convert: fn(&ArrayRef, &DataType) -> crate::Result, +) -> crate::Result { + let mut selected = Vec::new(); + let restore = indices + .iter() + .map(|key| { + key.map(|key| { + let index = selected.len() as u64; + selected.push(key); + index + }) + }) + .collect::>(); + if selected.is_empty() { + return Ok(arrow_array::new_null_array(target, indices.len())); + } + let values = if let Some(union) = array.as_any().downcast_ref::() { + // Taking a sparse Union would also take every inactive child. Select + // its active values directly, including through dictionary wrappers. + construct_union_rows( + union, + selected.into_iter().map(|row| row as usize), + target, + convert, + )? + } else { + let values = arrow_select::take::take(array.as_ref(), &UInt64Array::from(selected), None) + .map_err(|error| invalid(error.to_string()))?; + convert(&values, target)? + }; + arrow_select::take::take(values.as_ref(), &UInt64Array::from(restore), None) + .map_err(|error| invalid(error.to_string())) +} + +/// Only active children participate in the Python-value constructor. A sparse +/// Union can contain invalid values at inactive positions; a dense Union can +/// select any child offset repeatedly or in a different order. +pub(super) fn construct_union( + array: &ArrayRef, + target: &DataType, + convert: fn(&ArrayRef, &DataType) -> crate::Result, +) -> crate::Result { + let input = array.as_any().downcast_ref::().unwrap(); + construct_union_rows(input, 0..input.len(), target, convert) +} + +fn construct_union_rows( + input: &UnionArray, + rows: impl ExactSizeIterator, + target: &DataType, + convert: fn(&ArrayRef, &DataType) -> crate::Result, +) -> crate::Result { + if rows.len() == 0 { + return Ok(arrow_array::new_empty_array(target)); + } + let mut selected = IndexMap::>::new(); + let mut positions = Vec::with_capacity(rows.len()); + for row in rows { + let entry = selected.entry(input.type_id(row)); + let child = entry.index(); + let offsets = entry.or_default(); + positions.push((child, offsets.len())); + offsets.push(input.value_offset(row) as u64); + } + let values = selected + .into_iter() + .map(|(id, offsets)| { + let selected = arrow_select::take::take( + input.child(id).as_ref(), + &UInt64Array::from(offsets), + None, + ) + .map_err(|error| invalid(error.to_string()))?; + convert(&selected, target) + }) + .collect::>>()?; + let values = values + .iter() + .map(|array| array.as_ref()) + .collect::>(); + arrow_select::interleave::interleave(&values, &positions) + .map_err(|error| invalid(error.to_string())) +} + +#[cfg(test)] +mod tests { + use super::super::{cast_update_value, CastMode}; + use super::*; + use arrow_array::{ + types::Int8Type, DictionaryArray, FixedSizeListArray, Float64Array, Int32Array, Int64Array, + Int8Array, ListArray, MapArray, NullArray, StringArray, StructArray, + }; + use arrow_buffer::{NullBuffer, OffsetBuffer, ScalarBuffer}; + use arrow_schema::{Field, UnionFields}; + use std::sync::Arc; + + fn numeric_union(dense: bool) -> ArrayRef { + let fields = UnionFields::try_new( + [7, 42, 99], + [ + Field::new("small", DataType::Int32, true), + Field::new("large", DataType::Int64, true), + Field::new("unused", DataType::Utf8, true), + ], + ) + .unwrap(); + let (left, right, offsets) = if dense { + ( + vec![Some(-1), None, Some(2)], + vec![Some(3), Some(1), Some(i64::MAX)], + Some(ScalarBuffer::from(vec![0, 1, 2, 0, 1, 2])), + ) + } else { + ( + vec![Some(-1), Some(0), Some(2), Some(0), None, Some(0)], + vec![ + Some(i64::MAX), + Some(1), + Some(i64::MAX), + Some(3), + Some(i64::MAX), + Some(i64::MAX), + ], + None, + ) + }; + Arc::new( + UnionArray::try_new( + fields, + ScalarBuffer::from(vec![7, 42, 7, 42, 7, 42]), + offsets, + vec![ + Arc::new(Int32Array::from(left)), + Arc::new(Int64Array::from(right)), + Arc::new(StringArray::from(vec![ + "invalid"; + if dense { 1 } else { 6 } + ])), + ], + ) + .unwrap(), + ) + } + + #[test] + fn active_values_keep_slices_nulls_and_nonstandard_type_ids() { + for dense in [true, false] { + let input = numeric_union(dense); + let slice = input.slice(1, 4); + assert_eq!(slice.null_count(), 0); + assert!(cast_update_value(&slice, &DataType::Int32, CastMode::Assignment).is_err()); + let result = cast_update_value(&slice, &DataType::Int32, CastMode::RowUpdate).unwrap(); + assert_eq!( + result.to_data(), + Int32Array::from(vec![Some(1), Some(2), Some(3), None]).to_data() + ); + // The same wide value must fail as soon as its type/offset is active. + assert!(cast_update_value(&input, &DataType::Int32, CastMode::RowUpdate).is_err()); + assert_eq!( + cast_update_value(&slice.slice(0, 0), &DataType::Int32, CastMode::RowUpdate) + .unwrap() + .len(), + 0 + ); + } + } + + #[test] + fn dense_offsets_can_repeat_and_reorder_child_values() { + let input = numeric_union(true); + let fields = match input.data_type() { + DataType::Union(fields, _) => fields.clone(), + _ => unreachable!(), + }; + let input = input.as_any().downcast_ref::().unwrap(); + let reordered: ArrayRef = Arc::new( + UnionArray::try_new( + fields, + ScalarBuffer::from(vec![7, 7, 42, 7]), + Some(ScalarBuffer::from(vec![2, 0, 1, 2])), + vec![ + input.child(7).clone(), + input.child(42).clone(), + input.child(99).clone(), + ], + ) + .unwrap(), + ); + assert_eq!( + cast_update_value(&reordered, &DataType::Int32, CastMode::RowUpdate) + .unwrap() + .to_data(), + Int32Array::from(vec![2, -1, 1, 2]).to_data(), + ); + } + + #[test] + fn every_selected_child_uses_constructor_semantics() { + let fields = UnionFields::try_new( + [3, 8], + [ + Field::new("number", DataType::Float64, true), + Field::new("text", DataType::Utf8, true), + ], + ) + .unwrap(); + let input: ArrayRef = Arc::new( + UnionArray::try_new( + fields, + ScalarBuffer::from(vec![3, 8]), + Some(ScalarBuffer::from(vec![0, 0])), + vec![ + Arc::new(Float64Array::from(vec![1.5])), + Arc::new(StringArray::from(vec!["42"])), + ], + ) + .unwrap(), + ); + assert_eq!( + cast_update_value(&input.slice(0, 1), &DataType::Int32, CastMode::RowUpdate) + .unwrap() + .to_data(), + Int32Array::from(vec![1]).to_data(), + ); + // A numeric text child must not get a fresh safe-cast attempt after + // the Union's whole-column safe cast has already failed. + assert!(cast_update_value(&input, &DataType::Int32, CastMode::RowUpdate).is_err()); + assert!( + cast_update_value(&input.slice(0, 1), &DataType::Utf8, CastMode::RowUpdate).is_err() + ); + } + + #[test] + fn constructors_recurse_through_rows_and_arrays() { + for dense in [true, false] { + let input = numeric_union(dense).slice(1, 4); + let rows: ArrayRef = Arc::new(StructArray::from(vec![( + Arc::new(Field::new("n", input.data_type().clone(), true)), + input.clone(), + )])); + let target = DataType::Struct(vec![Field::new("n", DataType::Int32, true)].into()); + let output = cast_update_value(&rows, &target, CastMode::RowUpdate).unwrap(); + let expected = Int32Array::from(vec![Some(1), Some(2), Some(3), None]); + assert_eq!( + output + .as_any() + .downcast_ref::() + .unwrap() + .column(0) + .to_data(), + expected.to_data() + ); + let lists: ArrayRef = Arc::new(ListArray::new( + Arc::new(Field::new("item", input.data_type().clone(), true)), + OffsetBuffer::from_lengths([2, 2]), + input, + None, + )); + let target = DataType::List(Arc::new(Field::new("item", DataType::Int32, true))); + let output = cast_update_value(&lists, &target, CastMode::RowUpdate).unwrap(); + assert_eq!( + output + .as_any() + .downcast_ref::() + .unwrap() + .values() + .to_data(), + expected.to_data() + ); + } + } + + fn pair_union() -> ArrayRef { + let pairs: ArrayRef = Arc::new(ListArray::new( + Arc::new(Field::new("item", DataType::Int32, true)), + OffsetBuffer::from_lengths([2]), + Arc::new(Int32Array::from(vec![1, 2])), + None, + )); + let pairs: ArrayRef = Arc::new(ListArray::new( + Arc::new(Field::new("item", pairs.data_type().clone(), true)), + OffsetBuffer::from_lengths([1]), + pairs, + None, + )); + Arc::new( + UnionArray::try_new( + UnionFields::try_new( + [7, 42], + [ + Field::new("pairs", pairs.data_type().clone(), true), + Field::new("empty", DataType::Null, true), + ], + ) + .unwrap(), + ScalarBuffer::from(vec![7, 42]), + Some(ScalarBuffer::from(vec![0, 0])), + vec![pairs, Arc::new(NullArray::new(1))], + ) + .unwrap(), + ) + } + + fn map_type(key: DataType) -> DataType { + DataType::Map( + Arc::new(Field::new( + "entries", + DataType::Struct( + vec![ + Field::new("key", key, false), + Field::new("value", DataType::Int64, true), + ] + .into(), + ), + false, + )), + false, + ) + } + + #[test] + fn map_pair_conversion_is_root_only_through_encoding_wrappers() { + let input = pair_union(); + let target = map_type(DataType::Int32); + let output = cast_update_value(&input, &target, CastMode::RowUpdate).unwrap(); + let output = output.as_any().downcast_ref::().unwrap(); + assert_eq!(output.value_offsets(), &[0, 1, 1]); + assert!(output.is_null(1)); + assert_eq!( + output.values().to_data(), + Int64Array::from(vec![2]).to_data() + ); + let encoded: ArrayRef = Arc::new( + DictionaryArray::::try_new(Int8Array::from(vec![0, 1]), input.clone()) + .unwrap(), + ); + assert_eq!( + cast_update_value(&encoded, &target, CastMode::RowUpdate) + .unwrap() + .to_data(), + output.to_data() + ); + let nested: ArrayRef = Arc::new(StructArray::from(vec![( + Arc::new(Field::new("m", input.data_type().clone(), true)), + input, + )])); + let target = DataType::Struct(vec![Field::new("m", target, true)].into()); + assert!(cast_update_value(&nested, &target, CastMode::RowUpdate).is_err()); + } + + #[test] + fn dictionary_union_preserves_null_indices_and_active_null_values() { + for dense in [true, false] { + let values = numeric_union(dense).slice(1, 4); + let input: ArrayRef = Arc::new( + DictionaryArray::::try_new( + Int8Array::from(vec![Some(3), None, Some(0), Some(1)]), + values.clone(), + ) + .unwrap(), + ); + let output = cast_update_value(&input, &DataType::Int32, CastMode::RowUpdate).unwrap(); + assert_eq!( + output.to_data(), + Int32Array::from(vec![None, None, Some(1), Some(2)]).to_data() + ); + let input: ArrayRef = Arc::new( + DictionaryArray::::try_new(Int8Array::from(vec![None, None]), values) + .unwrap(), + ); + assert_eq!( + cast_update_value(&input, &DataType::Int32, CastMode::RowUpdate) + .unwrap() + .null_count(), + 2 + ); + } + let input: ArrayRef = Arc::new( + DictionaryArray::::try_new( + Int8Array::from(vec![None, Some(0), Some(1), None]), + pair_union(), + ) + .unwrap(), + ); + let output = + cast_update_value(&input, &map_type(DataType::Int32), CastMode::RowUpdate).unwrap(); + let output = output.as_any().downcast_ref::().unwrap(); + assert_eq!(output.value_offsets(), &[0, 0, 1, 1, 1]); + assert_eq!(output.null_count(), 3); + } + + #[test] + fn union_dict_values_allow_explicit_null_map_values() { + let row: ArrayRef = Arc::new(StructArray::from(vec![ + ( + Arc::new(Field::new("a", DataType::Int32, true)), + Arc::new(Int32Array::from(vec![Some(1)])) as ArrayRef, + ), + ( + Arc::new(Field::new("b", DataType::Int32, true)), + Arc::new(Int32Array::from(vec![None])) as ArrayRef, + ), + ])); + let target = map_type(DataType::Utf8); + assert!(cast_update_value(&row, &target, CastMode::RowUpdate).is_err()); + let input: ArrayRef = Arc::new( + UnionArray::try_new( + UnionFields::try_new([42], [Field::new("row", row.data_type().clone(), true)]) + .unwrap(), + ScalarBuffer::from(vec![42]), + Some(ScalarBuffer::from(vec![0])), + vec![row], + ) + .unwrap(), + ); + let output = cast_update_value(&input, &target, CastMode::RowUpdate).unwrap(); + let output = output.as_any().downcast_ref::().unwrap(); + assert_eq!( + output.keys().to_data(), + StringArray::from(vec!["a", "b"]).to_data() + ); + assert_eq!( + output.values().to_data(), + Int64Array::from(vec![Some(1), None]).to_data() + ); + } + + #[test] + fn null_parents_hide_union_values_before_constructor_conversion() { + for dense in [false, true] { + let input: ArrayRef = Arc::new( + UnionArray::try_new( + UnionFields::try_new( + [7, 42], + [ + Field::new("number", DataType::Int32, true), + Field::new("text", DataType::Utf8, true), + ], + ) + .unwrap(), + ScalarBuffer::from(vec![42, 7, 7, 42]), + dense.then(|| ScalarBuffer::from(vec![0, 0, 1, 1])), + if dense { + vec![ + Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef, + Arc::new(StringArray::from(vec!["hidden", "also hidden"])), + ] + } else { + vec![ + Arc::new(Int32Array::from(vec![0, 1, 2, 0])) as ArrayRef, + Arc::new(StringArray::from(vec!["hidden"; 4])), + ] + }, + ) + .unwrap(), + ); + let dictionary: ArrayRef = Arc::new( + DictionaryArray::::try_new( + Int8Array::from(vec![0, 1, 2, 3]), + input.clone(), + ) + .unwrap(), + ); + for input in [input, dictionary] { + let parents = NullBuffer::from(vec![false, true, true, false]); + let rows: ArrayRef = Arc::new(StructArray::new( + vec![Field::new("n", input.data_type().clone(), true)].into(), + vec![input.clone()], + Some(parents.clone()), + )); + let target = DataType::Struct(vec![Field::new("n", DataType::Int32, true)].into()); + assert!(cast_update_value(&rows, &target, CastMode::Assignment).is_err()); + let output = cast_update_value(&rows, &target, CastMode::RowUpdate).unwrap(); + let output = output.as_struct(); + assert_eq!(output.nulls(), Some(&parents)); + assert_eq!( + output.column(0).to_data(), + Int32Array::from(vec![None, Some(1), Some(2), None]).to_data() + ); + + let lists: ArrayRef = Arc::new(FixedSizeListArray::new( + Arc::new(Field::new("item", input.data_type().clone(), true)), + 1, + input, + Some(parents.clone()), + )); + let target = DataType::List(Arc::new(Field::new("item", DataType::Int32, true))); + let output = cast_update_value(&lists, &target, CastMode::RowUpdate).unwrap(); + let output = output.as_list::(); + assert_eq!(output.nulls(), Some(&parents)); + assert_eq!( + output.value(1).to_data(), + Int32Array::from(vec![1]).to_data() + ); + assert_eq!( + output.value(2).to_data(), + Int32Array::from(vec![2]).to_data() + ); + } + } + } + + #[test] + fn map_union_values_leave_absent_row_fields_null() { + for dense in [false, true] { + let values: ArrayRef = Arc::new( + UnionArray::try_new( + UnionFields::try_new([7], [Field::new("number", DataType::Int32, true)]) + .unwrap(), + ScalarBuffer::from(vec![7; 4]), + dense.then(|| ScalarBuffer::from(vec![0, 1, 2, 3])), + vec![Arc::new(Int32Array::from(vec![1, 2, 3, 4]))], + ) + .unwrap(), + ); + let entries = StructArray::new( + vec![ + Field::new("key", DataType::Utf8, false), + Field::new("value", values.data_type().clone(), true), + ] + .into(), + vec![Arc::new(StringArray::from(vec!["n"; 4])), values], + None, + ); + let input: ArrayRef = Arc::new(MapArray::new( + Arc::new(Field::new("entries", entries.data_type().clone(), false)), + OffsetBuffer::from_lengths([1; 4]), + entries, + None, + false, + )); + let target = DataType::Struct( + vec![ + Field::new("n", DataType::Int32, true), + Field::new("v", DataType::Utf8, true), + ] + .into(), + ); + let output = cast_update_value(&input, &target, CastMode::RowUpdate).unwrap(); + let output = output.as_struct(); + assert_eq!( + output.column(0).to_data(), + Int32Array::from(vec![1, 2, 3, 4]).to_data() + ); + assert_eq!( + output.column(1).to_data(), + StringArray::from(vec![None::<&str>; 4]).to_data() + ); + } + } +} diff --git a/crates/paimon/src/table/update_input/scalar_tests.rs b/crates/paimon/src/table/update_input/scalar_tests.rs new file mode 100644 index 000000000..56f06debc --- /dev/null +++ b/crates/paimon/src/table/update_input/scalar_tests.rs @@ -0,0 +1,207 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use super::*; +use arrow_array::{Decimal128Array, Decimal256Array, FixedSizeBinaryArray}; +use arrow_buffer::{Buffer, NullBuffer}; +use arrow_schema::Field; + +#[test] +fn fixed_binary_strings_keep_utf8_nulls_slices_and_empty_values() { + let input: ArrayRef = Arc::new( + FixedSizeBinaryArray::try_from_sparse_iter_with_size( + [ + Some(b"old".as_slice()), + Some(b"a\0b".as_slice()), + None, + Some("中".as_bytes()), + ] + .into_iter(), + 3, + ) + .unwrap(), + ); + let input = input.slice(1, 3); + for target in [DataType::Utf8, DataType::LargeUtf8] { + for mode in [CastMode::Assignment, CastMode::RowUpdate] { + let actual = cast_update_value(&input, &target, mode).unwrap(); + let expected = arrow_cast::cast( + &StringArray::from(vec![Some("a\0b"), None, Some("中")]), + &target, + ) + .unwrap(); + assert_eq!(actual.to_data(), expected.to_data()); + } + } + let empty: ArrayRef = Arc::new( + FixedSizeBinaryArray::try_new( + 0, + Buffer::from_vec(Vec::::new()), + Some(NullBuffer::from(vec![true, false, true])), + ) + .unwrap(), + ); + assert_eq!( + cast_update_value(&empty, &DataType::Utf8, CastMode::Assignment) + .unwrap() + .to_data(), + StringArray::from(vec![Some(""), None, Some("")]).to_data() + ); + let invalid: ArrayRef = + Arc::new(FixedSizeBinaryArray::try_from_iter([b"\xffab".as_slice()].into_iter()).unwrap()); + assert!(cast_update_value(&invalid, &DataType::Utf8, CastMode::RowUpdate).is_err()); + // The adaptation is for text, not a new binary-to-numeric cast. + let digits: ArrayRef = + Arc::new(FixedSizeBinaryArray::try_from_iter([b"123".as_slice()].into_iter()).unwrap()); + assert!(cast_update_value(&digits, &DataType::Int32, CastMode::RowUpdate).is_err()); +} + +#[test] +fn half_float_text_uses_all_significant_digits() { + let input = Float64Array::from(vec![ + Some(0.1), + None, + Some(1.5), + Some(-3.25), + Some(65504.0), + Some(2_f64.powi(-24)), + ]); + let input = arrow_cast::cast(&input, &DataType::Float16).unwrap(); + let expected = StringArray::from(vec![ + Some("0.0999755859375"), + None, + Some("1.5"), + Some("-3.25"), + Some("65504"), + Some("5.960464477539063e-8"), + ]); + for mode in [CastMode::Assignment, CastMode::RowUpdate] { + assert_eq!( + cast_update_value(&input, &DataType::Utf8, mode) + .unwrap() + .to_data(), + expected.to_data() + ); + } +} + +#[test] +fn scalar_encodings_recurse_inside_containers() { + let input: ArrayRef = Arc::new( + FixedSizeBinaryArray::try_from_iter([b"abc".as_slice(), "中".as_bytes()].into_iter()) + .unwrap(), + ); + let rows: ArrayRef = Arc::new(StructArray::from(vec![( + Arc::new(Field::new("s", input.data_type().clone(), true)), + input.clone(), + )])); + let target = DataType::Struct(vec![Field::new("s", DataType::Utf8, true)].into()); + let output = cast_update_value(&rows, &target, CastMode::Assignment).unwrap(); + let expected = StringArray::from(vec!["abc", "中"]); + assert_eq!( + output + .as_any() + .downcast_ref::() + .unwrap() + .column(0) + .to_data(), + expected.to_data() + ); + let lists: ArrayRef = Arc::new(ListArray::new( + Arc::new(Field::new("item", input.data_type().clone(), true)), + OffsetBuffer::from_lengths([2]), + input, + None, + )); + let target = DataType::List(Arc::new(Field::new("item", DataType::Utf8, true))); + let output = cast_update_value(&lists, &target, CastMode::Assignment).unwrap(); + assert_eq!( + output + .as_any() + .downcast_ref::() + .unwrap() + .values() + .to_data(), + expected.to_data() + ); +} + +#[test] +fn wide_decimal_floating_values_use_java_big_decimal_rounding() { + let coefficient = "1508079088755887087797858531882816082" + .parse::() + .unwrap(); + let source: ArrayRef = Arc::new( + Decimal128Array::from(vec![Some(coefficient), None, Some(-coefficient)]) + .with_precision_and_scale(38, 1) + .unwrap(), + ); + let expected = [ + Some(1.508079088755887e35), + None, + Some(-1.508079088755887e35), + ]; + let mut sources = vec![source]; + for precision in [38, 39, 76] { + let wide: ArrayRef = Arc::new( + Decimal256Array::from_iter( + [Some(coefficient), None, Some(-coefficient)] + .into_iter() + .map(|value| value.map(arrow_buffer::i256::from_i128)), + ) + .with_precision_and_scale(precision, 1) + .unwrap(), + ); + sources.push(wide); + } + for source in sources { + for mode in [CastMode::Assignment, CastMode::RowUpdate] { + let output = cast_update_value(&source, &DataType::Float64, mode).unwrap(); + assert_eq!( + output.to_data(), + Float64Array::from(expected.to_vec()).to_data() + ); + let output = cast_update_value(&source, &DataType::Float32, mode).unwrap(); + assert_eq!( + output.to_data(), + Float32Array::from_iter(expected.into_iter().map(|v| v.map(|v| v as f32))) + .to_data() + ); + } + } +} + +#[test] +fn compact_decimal_floating_values_keep_java_long_division() { + let coefficient = 9_007_199_254_740_993_i128; + let source: ArrayRef = Arc::new( + Decimal128Array::from(vec![Some(coefficient), None]) + .with_precision_and_scale(18, 3) + .unwrap(), + ); + let expected = coefficient as f64 / 1000.0; + let output = cast_update_value(&source, &DataType::Float64, CastMode::Assignment).unwrap(); + assert_eq!( + output.to_data(), + Float64Array::from(vec![Some(expected), None]).to_data() + ); + let output = cast_update_value(&source, &DataType::Float32, CastMode::Assignment).unwrap(); + assert_eq!( + output.to_data(), + Float32Array::from(vec![Some(expected as f32), None]).to_data() + ); +}