From 89d30fbda9fefda0dad51a136b969a649bac24f2 Mon Sep 17 00:00:00 2001 From: Nick Fu Date: Fri, 31 Jul 2026 19:23:40 +0800 Subject: [PATCH] fix: honor target sorted flag and preserve map field metadata in cast_map_to_map (#5097) Co-Authored-By: Claude Opus 5 (1M context) --- .../spark-expr/src/conversion_funcs/cast.rs | 792 ++++++++++++++++-- .../expressions/cast/cast_complex.sql | 42 + 2 files changed, 758 insertions(+), 76 deletions(-) diff --git a/native/spark-expr/src/conversion_funcs/cast.rs b/native/spark-expr/src/conversion_funcs/cast.rs index 37fddb8c115..df2295218c1 100644 --- a/native/spark-expr/src/conversion_funcs/cast.rs +++ b/native/spark-expr/src/conversion_funcs/cast.rs @@ -45,8 +45,8 @@ use arrow::array::{ new_null_array, BinaryBuilder, DictionaryArray, GenericByteArray, ListArray, MapArray, StringArray, StructArray, }; +use arrow::datatypes::GenericBinaryType; use arrow::datatypes::{ArrowDictionaryKeyType, ArrowNativeType, DataType, Schema}; -use arrow::datatypes::{Field, Fields, GenericBinaryType}; use arrow::error::ArrowError; use arrow::{ array::{ @@ -78,6 +78,17 @@ static CAST_OPTIONS: CastOptions = CastOptions { .with_timestamp_format(TIMESTAMP_FORMAT), }; +/// Arrow `CastOptions` matching a Comet eval mode: ANSI surfaces conversion failures as errors, +/// the other modes substitute null. +fn arrow_cast_options(eval_mode: EvalMode) -> CastOptions<'static> { + CastOptions { + safe: !matches!(eval_mode, EvalMode::Ansi), + format_options: FormatOptions::new() + .with_timestamp_tz_format(TIMESTAMP_FORMAT) + .with_timestamp_format(TIMESTAMP_FORMAT), + } +} + #[derive(Debug, Eq)] pub struct Cast { pub child: Arc, @@ -262,12 +273,7 @@ pub(crate) fn cast_array( let array = array_with_timezone(array, cast_options.timezone.clone(), Some(to_type))?; let eval_mode = cast_options.eval_mode; - let native_cast_options: CastOptions = CastOptions { - safe: !matches!(cast_options.eval_mode, EvalMode::Ansi), // take safe mode from cast_options passed - format_options: FormatOptions::new() - .with_timestamp_tz_format(TIMESTAMP_FORMAT) - .with_timestamp_format(TIMESTAMP_FORMAT), - }; + let native_cast_options: CastOptions = arrow_cast_options(cast_options.eval_mode); let array = match &from_type { Dictionary(key_type, value_type) @@ -545,8 +551,15 @@ fn cast_struct_to_struct( } } -/// Cast between map types, handling field name differences between Parquet ("key_value") -/// and Spark ("entries") while preserving the map's structure. +/// Cast between map types (e.g. Parquet "key_value" -> Spark "entries"). +/// +/// - Rename-only (unchanged key/value types and sort order): delegate to arrow's `cast`, which +/// relabels to the target fields and preserves their metadata with no value transformation. +/// - Otherwise (a child type or the sort flag differs): recurse with Comet's `cast_array` for a +/// changed child and hand-build the result with the target sort flag. `try_new` is used so a +/// malformed target returns `Err` rather than panicking. +/// +/// Either way the result `data_type()` equals `to_type`. fn cast_map_to_map( array: &ArrayRef, from_type: &DataType, @@ -561,75 +574,71 @@ fn cast_map_to_map( match (from_type, to_type) { ( DataType::Map(from_entries_field, from_sorted), - DataType::Map(to_entries_field, _to_sorted), + DataType::Map(to_entries_field, to_sorted), ) => { - // Get the struct types for entries - let from_struct_type = from_entries_field.data_type(); - let to_struct_type = to_entries_field.data_type(); - - match (from_struct_type, to_struct_type) { - (DataType::Struct(from_fields), DataType::Struct(to_fields)) => { - // Get the key and value types - let from_key_type = from_fields[0].data_type(); - let from_value_type = from_fields[1].data_type(); - let to_key_type = to_fields[0].data_type(); - let to_value_type = to_fields[1].data_type(); - - // Cast keys if needed - let keys = map_array.keys(); - let cast_keys = if from_key_type != to_key_type { - cast_array(Arc::clone(keys), to_key_type, cast_options)? - } else { - Arc::clone(keys) - }; - - // Cast values if needed - let values = map_array.values(); - let cast_values = if from_value_type != to_value_type { - cast_array(Arc::clone(values), to_value_type, cast_options)? - } else { - Arc::clone(values) - }; - - // Build the new entries struct with the target field names - let new_key_field = Arc::new(Field::new( - to_fields[0].name(), - to_key_type.clone(), - to_fields[0].is_nullable(), - )); - let new_value_field = Arc::new(Field::new( - to_fields[1].name(), - to_value_type.clone(), - to_fields[1].is_nullable(), - )); - - let struct_fields = Fields::from(vec![new_key_field, new_value_field]); - let entries_struct = - StructArray::new(struct_fields, vec![cast_keys, cast_values], None); - - // Create the new map field with the target name - let new_entries_field = Arc::new(Field::new( - to_entries_field.name(), - DataType::Struct(entries_struct.fields().clone()), - to_entries_field.is_nullable(), - )); - - // Build the new MapArray - let new_map = MapArray::new( - new_entries_field, - map_array.offsets().clone(), - entries_struct, - map_array.nulls().cloned(), - *from_sorted, - ); - - Ok(Arc::new(new_map)) - } - _ => Err(DataFusionError::Internal(format!( - "Map entries must be structs, got {:?} and {:?}", - from_struct_type, to_struct_type - ))), + let (from_fields, to_fields) = + match (from_entries_field.data_type(), to_entries_field.data_type()) { + (DataType::Struct(f), DataType::Struct(t)) => (f, t), + (from_struct_type, to_struct_type) => { + return Err(DataFusionError::Internal(format!( + "Map entries must be structs, got {from_struct_type:?} and \ + {to_struct_type:?}" + ))) + } + }; + // A map entries struct is always exactly (key, value); guard before indexing [0]/[1] so + // a malformed target (0/1/3+ fields) returns Err instead of panicking. + if from_fields.len() != 2 || to_fields.len() != 2 { + return Err(DataFusionError::Internal(format!( + "Map entries struct must have exactly 2 fields (key, value); got from={} to={}", + from_fields.len(), + to_fields.len() + ))); + } + let key_type_unchanged = from_fields[0].data_type() == to_fields[0].data_type(); + let value_type_unchanged = from_fields[1].data_type() == to_fields[1].data_type(); + + // Rename-only fast path (the common Parquet "key_value" -> Spark "entries" case): the + // key and value types are unchanged and the sort order is unchanged, so only the field + // labels/metadata differ. Delegate to arrow's cast, whose map arm requires matching sort + // flags and relabels to the target fields, preserving their metadata and the values. + if key_type_unchanged && value_type_unchanged && from_sorted == to_sorted { + return Ok(cast_with_options( + array, + to_type, + &arrow_cast_options(cast_options.eval_mode), + )?); } + + // Otherwise a child type or the sort flag differs. Recurse with Comet's Spark-compatible + // casts for the changed children and hand-build the result carrying the target sort flag. + // `try_new` reports a malformed target as `Err` rather than panicking. + let keys = map_array.keys(); + let cast_keys = if key_type_unchanged { + Arc::clone(keys) + } else { + cast_array(Arc::clone(keys), to_fields[0].data_type(), cast_options)? + }; + let values = map_array.values(); + let cast_values = if value_type_unchanged { + Arc::clone(values) + } else { + cast_array(Arc::clone(values), to_fields[1].data_type(), cast_options)? + }; + + let entries_struct = StructArray::try_new( + to_fields.clone(), + vec![cast_keys, cast_values], + map_array.entries().nulls().cloned(), + )?; + let new_map = MapArray::try_new( + Arc::clone(to_entries_field), + map_array.offsets().clone(), + entries_struct, + map_array.nulls().cloned(), + *to_sorted, + )?; + Ok(Arc::new(new_map)) } _ => unreachable!("cast_map_to_map called with non-Map types"), } @@ -1279,4 +1288,635 @@ mod tests { assert_eq!(3, values.null_count()); assert!(values.iter().all(|value| value.is_none())); } + + fn legacy_opts() -> SparkCastOptions { + SparkCastOptions::new(EvalMode::Legacy, "UTC", false) + } + + /// Build a `Map` MapArray (Parquet-style "key_value" field names). + fn build_str_i32_map( + keys: Vec<&str>, + values: Vec>, + offsets: Vec, + map_nulls: Option, + entries_nulls: Option, + sorted: bool, + ) -> MapArray { + use arrow::array::{Int32Array, StringArray}; + let key_field = Arc::new(Field::new("key_value_key", DataType::Utf8, false)); + let value_field = Arc::new(Field::new("key_value_value", DataType::Int32, true)); + let entries_fields = Fields::from(vec![key_field, value_field]); + let ks = Arc::new(StringArray::from(keys)) as ArrayRef; + let vs = Arc::new(Int32Array::from(values)) as ArrayRef; + let entries_struct = StructArray::new(entries_fields, vec![ks, vs], entries_nulls); + let entries_field = Arc::new(Field::new( + "key_value", + DataType::Struct(entries_struct.fields().clone()), + false, + )); + MapArray::new( + entries_field, + OffsetBuffer::::new(offsets.into()), + entries_struct, + map_nulls, + sorted, + ) + } + + /// Build a target `Map` type ("entries"/"key"/"value" Spark-style names). + fn build_to_map_type(val_type: DataType, val_nullable: bool, sorted: bool) -> DataType { + let to_key = Arc::new(Field::new("key", DataType::Utf8, false)); + let to_val = Arc::new(Field::new("value", val_type, val_nullable)); + let entries = Arc::new(Field::new( + "entries", + DataType::Struct(Fields::from(vec![to_key, to_val])), + false, + )); + DataType::Map(entries, sorted) + } + + /// Assert which branch of `cast_map_to_map` a (from, to) type pair selects, by checking the + /// three inputs the branch condition is built from. + fn assert_map_child_types_and_sort( + from_type: &DataType, + to_type: &DataType, + expect_key_unchanged: bool, + expect_value_unchanged: bool, + expect_same_sort: bool, + ) { + let children = |t: &DataType| match t { + DataType::Map(entries, sorted) => match entries.data_type() { + DataType::Struct(f) => { + assert_eq!(f.len(), 2, "map entries must be (key, value)"); + (f[0].data_type().clone(), f[1].data_type().clone(), *sorted) + } + other => panic!("map entries must be a struct, got {other:?}"), + }, + other => panic!("expected a Map type, got {other:?}"), + }; + let (from_key, from_val, from_sorted) = children(from_type); + let (to_key, to_val, to_sorted) = children(to_type); + assert_eq!( + from_key == to_key, + expect_key_unchanged, + "key type unchanged: {from_key:?} vs {to_key:?}" + ); + assert_eq!( + from_val == to_val, + expect_value_unchanged, + "value type unchanged: {from_val:?} vs {to_val:?}" + ); + assert_eq!( + from_sorted == to_sorted, + expect_same_sort, + "sort flag unchanged: {from_sorted} vs {to_sorted}" + ); + } + + fn sorted_src(from_sorted: bool) -> ArrayRef { + Arc::new(build_str_i32_map( + vec!["a", "b", "c"], + vec![Some(1), Some(2), Some(3)], + vec![0, 3], + None, + None, + from_sorted, + )) as ArrayRef + } + + #[test] + fn test_cast_map_to_map_sorted_equal_flags_allowed() { + // false -> false and true -> true: the sort flag is unchanged and the result type matches. + for s in [false, true] { + let to_type = build_to_map_type(DataType::Int32, true, s); + let casted = cast_array(sorted_src(s), &to_type, &legacy_opts()).unwrap(); + assert_eq!(casted.data_type(), &to_type, "sorted={s}"); + } + } + + #[test] + fn test_cast_map_to_map_sorted_true_to_false_allowed() { + // Downgrade a sorted map to unsorted: allowed, result carries the target (false) flag. + let to_type = build_to_map_type(DataType::Int32, true, false); + let casted = cast_array(sorted_src(true), &to_type, &legacy_opts()).unwrap(); + assert_eq!(casted.data_type(), &to_type); + match casted.data_type() { + DataType::Map(_, is_sorted) => assert!(!*is_sorted), + _ => panic!("Expected Map DataType"), + } + } + + #[test] + fn test_cast_map_to_map_sorted_value_only_cast_allowed() { + use arrow::array::{Int64Array, StringArray}; + // Sorted source, key type unchanged, value Int32 -> Int64, target sorted=true. The key + // ordering is unaffected by a value cast, so this is allowed and stays sorted. + let src = Arc::new(build_str_i32_map( + vec!["a", "b", "c"], + vec![Some(10), Some(20), Some(30)], + vec![0, 3], + None, + None, + true, // source sorted + )) as ArrayRef; + let to_key = Arc::new(Field::new("key", DataType::Utf8, false)); + let to_val = Arc::new(Field::new("value", DataType::Int64, true)); + let to_entries = Arc::new(Field::new( + "entries", + DataType::Struct(Fields::from(vec![to_key, to_val])), + false, + )); + let to_type = DataType::Map(Arc::clone(&to_entries), true); + + let casted = cast_array(src, &to_type, &legacy_opts()).unwrap(); + // Exact target type, and the sort flag is preserved. + assert_eq!(casted.data_type(), &to_type); + let m = casted.as_any().downcast_ref::().unwrap(); + match m.data_type() { + DataType::Map(_, is_sorted) => assert!(*is_sorted), + _ => panic!("Expected Map DataType"), + } + // Keys unchanged; values correctly cast to Int64. + let keys = m.keys().as_any().downcast_ref::().unwrap(); + let vals = m.values().as_any().downcast_ref::().unwrap(); + assert_eq!( + (0..3).map(|i| keys.value(i)).collect::>(), + vec!["a", "b", "c"] + ); + assert_eq!(vals.values(), &[10i64, 20, 30]); + } + + #[test] + fn test_cast_map_to_map_preserves_metadata_and_child_type_casts() { + use arrow::array::{Int32Array, Int64Array, StringArray}; + use std::collections::HashMap; + + let key_field = Arc::new(Field::new("key_value_key", DataType::Utf8, false)); + let value_field = Arc::new(Field::new("key_value_value", DataType::Int32, true)); + let entries_fields = Fields::from(vec![key_field, value_field]); + + let keys = Arc::new(StringArray::from(vec!["k1", "k2"])); + let values = Arc::new(Int32Array::from(vec![10, 20])); + let entries_struct = StructArray::new(entries_fields, vec![keys, values], None); + + let from_entries_field = Arc::new(Field::new( + "key_value", + DataType::Struct(entries_struct.fields().clone()), + false, + )); + let map_array = Arc::new(MapArray::new( + from_entries_field, + OffsetBuffer::::new(vec![0, 2].into()), + entries_struct, + None, + false, + )) as ArrayRef; + + // Target key field with custom metadata + let mut key_meta = HashMap::new(); + key_meta.insert("tag".to_string(), "map_key_meta".to_string()); + let to_key_field = + Arc::new(Field::new("key", DataType::Utf8, false).with_metadata(key_meta)); + let to_value_field = Arc::new(Field::new("value", DataType::Int64, true)); + let to_entries_field = Arc::new(Field::new( + "entries", + DataType::Struct(Fields::from(vec![to_key_field, to_value_field])), + false, + )); + let to_type = DataType::Map(to_entries_field, false); + + let casted = cast_array( + map_array, + &to_type, + &SparkCastOptions::new(EvalMode::Legacy, "UTC", false), + ) + .unwrap(); + + let casted_map = casted.as_any().downcast_ref::().unwrap(); + // Result type is exactly the requested target type (incl. field metadata). + assert_eq!(casted_map.data_type(), &to_type); + // Assert key field metadata is preserved + assert_eq!( + casted_map.entries().fields()[0].metadata().get("tag"), + Some(&"map_key_meta".to_string()) + ); + + // Assert child values were cast from Int32 to Int64 + let casted_values = casted_map + .values() + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(casted_values.value(0), 10i64); + assert_eq!(casted_values.value(1), 20i64); + } + + #[test] + fn test_cast_map_to_map_preserves_map_level_nulls_and_offsets() { + use arrow::array::{Int32Array, StringArray}; + use arrow::buffer::NullBuffer; + + let key_field = Arc::new(Field::new("key", DataType::Utf8, false)); + let value_field = Arc::new(Field::new("value", DataType::Int32, true)); + let entries_fields = Fields::from(vec![key_field, value_field]); + + let keys = Arc::new(StringArray::from(vec!["a", "b"])); + let values = Arc::new(Int32Array::from(vec![1, 2])); + let entries_struct = StructArray::new(entries_fields, vec![keys, values], None); + + let from_entries_field = Arc::new(Field::new( + "key_value", + DataType::Struct(entries_struct.fields().clone()), + false, + )); + + let map_nulls = NullBuffer::from(vec![true, false]); + let src_map = MapArray::new( + from_entries_field, + OffsetBuffer::::new(vec![0, 2, 2].into()), + entries_struct, + Some(map_nulls.clone()), + false, + ); + let map_array = Arc::new(src_map) as ArrayRef; + + let to_key_field = Arc::new(Field::new("new_key", DataType::Utf8, false)); + let to_value_field = Arc::new(Field::new("new_value", DataType::Int32, true)); + let to_entries_field = Arc::new(Field::new( + "entries", + DataType::Struct(Fields::from(vec![to_key_field, to_value_field])), + false, + )); + let to_type = DataType::Map(to_entries_field, false); + + let casted = cast_array( + map_array, + &to_type, + &SparkCastOptions::new(EvalMode::Legacy, "UTC", false), + ) + .unwrap(); + + let casted_map = casted.as_any().downcast_ref::().unwrap(); + assert_eq!(casted_map.data_type(), &to_type); + assert_eq!(casted_map.nulls(), Some(&map_nulls)); + assert!(casted_map.is_null(1)); + assert_eq!(casted_map.offsets().as_ref(), &[0, 2, 2]); + } + + // NOTE on entries-struct null buffers: an entries struct carrying a non-None null buffer is not + // constructible through the inspected safe Arrow constructors — `StructArray::new`/`try_new` and + // `MapArray` all normalize an all-valid `NullBuffer` to `None`, and a struct with real nulls is + // not a valid map entries array. (ArrayData/FFI paths were not audited, so this is not a claim + // of global unobservability.) The production code still clones `entries().nulls()` to mirror + // arrow's `cast_map_values`, but entries-level null preservation is not asserted (it would be a + // vacuous `None == None`). Map-LEVEL null preservation is covered by + // `test_cast_map_to_map_preserves_map_level_nulls_and_offsets`. + #[test] + fn test_cast_map_to_map_rename_only_fast_path() { + use arrow::array::{Int32Array, StringArray}; + use arrow::buffer::NullBuffer; + use std::collections::HashMap; + + // Key/value types unchanged and sort order unchanged -> the rename-only fast path (arrow + // cast) is used. Three rows including a null row so offsets and map nulls are non-trivial. + let map_nulls = NullBuffer::from(vec![true, false, true]); + let src = build_str_i32_map( + vec!["a", "b", "c"], + vec![Some(1), Some(2), Some(3)], + vec![0, 2, 2, 3], + Some(map_nulls.clone()), + None, + false, + ); + let src_offsets: Vec = src.offsets().as_ref().to_vec(); + let map_array = Arc::new(src) as ArrayRef; + + // Complete target schema: renamed entries/key/value fields, outer + key metadata, same + // (unchanged) child types, same sort flag. + let mut outer_meta = HashMap::new(); + outer_meta.insert("outer".to_string(), "entries_meta".to_string()); + let mut key_meta = HashMap::new(); + key_meta.insert("k".to_string(), "kmeta".to_string()); + let to_key = Arc::new(Field::new("key", DataType::Utf8, false).with_metadata(key_meta)); + let to_val = Arc::new(Field::new("value", DataType::Int32, true)); + let to_entries = Arc::new( + Field::new( + "entries", + DataType::Struct(Fields::from(vec![to_key, to_val])), + false, + ) + .with_metadata(outer_meta), + ); + let to_type = DataType::Map(Arc::clone(&to_entries), false); + + let casted = cast_array(map_array, &to_type, &legacy_opts()).unwrap(); + let casted_map = casted.as_any().downcast_ref::().unwrap(); + + // The complete target schema is reproduced exactly (field names, metadata, nullability, sort). + assert_eq!(casted_map.data_type(), &to_type); + // Map-level nulls and offsets are unchanged by the relabel. + assert_eq!(casted_map.nulls(), Some(&map_nulls)); + assert!(casted_map.is_null(1)); + assert_eq!(casted_map.offsets().as_ref(), src_offsets.as_slice()); + // Keys and values are unchanged by the relabel. + let keys = casted_map + .keys() + .as_any() + .downcast_ref::() + .unwrap(); + let vals = casted_map + .values() + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!( + (0..3).map(|i| keys.value(i)).collect::>(), + vec!["a", "b", "c"] + ); + assert_eq!(vals.values(), &[1, 2, 3]); + } + + #[test] + fn test_cast_map_to_map_empty_map() { + let src = Arc::new(build_str_i32_map( + vec![], + vec![], + vec![0], + None, + None, + false, + )) as ArrayRef; + let to_type = build_to_map_type(DataType::Int64, true, false); + let casted = cast_array(src, &to_type, &legacy_opts()).unwrap(); + assert_eq!(casted.data_type(), &to_type); + let m = casted.as_any().downcast_ref::().unwrap(); + assert_eq!(m.len(), 0); + } + + #[test] + fn test_cast_map_to_map_mixed_null_and_empty_rows() { + use arrow::buffer::NullBuffer; + // row0 = NULL, row1 = empty {}, row2 = {a:1, b:2} + let map_nulls = NullBuffer::from(vec![false, true, true]); + let src = Arc::new(build_str_i32_map( + vec!["a", "b"], + vec![Some(1), Some(2)], + vec![0, 0, 0, 2], + Some(map_nulls.clone()), + None, + false, + )) as ArrayRef; + let to_type = build_to_map_type(DataType::Int64, true, false); + let casted = cast_array(src, &to_type, &legacy_opts()).unwrap(); + assert_eq!(casted.data_type(), &to_type); + let m = casted.as_any().downcast_ref::().unwrap(); + assert_eq!(m.len(), 3); + assert_eq!(m.nulls(), Some(&map_nulls)); + assert!(m.is_null(0)); + assert!(!m.is_null(1)); // empty but valid + assert_eq!(m.offsets().as_ref(), &[0, 0, 0, 2]); + } + + #[test] + fn test_cast_map_to_map_sliced() { + use arrow::array::{Int64Array, StringArray}; + // 3 rows: {a:1}, {b:2, c:3}, {d:4}; slice to rows [1, 2]. + let full = Arc::new(build_str_i32_map( + vec!["a", "b", "c", "d"], + vec![Some(1), Some(2), Some(3), Some(4)], + vec![0, 1, 3, 4], + None, + None, + false, + )) as ArrayRef; + let sliced = full.slice(1, 2); + let to_type = build_to_map_type(DataType::Int64, true, false); + let casted = cast_array(sliced, &to_type, &legacy_opts()).unwrap(); + assert_eq!(casted.data_type(), &to_type); + let m = casted.as_any().downcast_ref::().unwrap(); + assert_eq!(m.len(), 2); + let keys = m.keys().as_any().downcast_ref::().unwrap(); + let vals = m.values().as_any().downcast_ref::().unwrap(); + let o = m.offsets(); + let (start, end) = (o[0] as usize, o[2] as usize); + let got_keys: Vec<&str> = (start..end).map(|i| keys.value(i)).collect(); + let got_vals: Vec = (start..end).map(|i| vals.value(i)).collect(); + assert_eq!(got_keys, vec!["b", "c", "d"]); + assert_eq!(got_vals, vec![2i64, 3, 4]); + } + + #[test] + fn test_cast_map_to_map_sliced_rename_only_fast_path() { + use arrow::array::{Int32Array, StringArray}; + // Same slicing as `test_cast_map_to_map_sliced`, but the value type is Int32 -> Int32. + // Key type, value type and sort flag are all unchanged, so only the field labels differ + // ("key_value"/"key_value_key" -> "entries"/"key") and arrow's cast handles the relabel. + let full = Arc::new(build_str_i32_map( + vec!["a", "b", "c", "d"], + vec![Some(1), Some(2), Some(3), Some(4)], + vec![0, 1, 3, 4], + None, + None, + false, + )) as ArrayRef; + let sliced = full.slice(1, 2); + let to_type = build_to_map_type(DataType::Int32, true, false); + + // The rename-only path is selected on unchanged child types and an unchanged sort flag. + // Assert that predicate on this input rather than assuming it. + assert_map_child_types_and_sort(sliced.data_type(), &to_type, true, true, true); + + let casted = cast_array(sliced, &to_type, &legacy_opts()).unwrap(); + assert_eq!(casted.data_type(), &to_type); + let m = casted.as_any().downcast_ref::().unwrap(); + assert_eq!(m.len(), 2); + let keys = m.keys().as_any().downcast_ref::().unwrap(); + let vals = m.values().as_any().downcast_ref::().unwrap(); + let o = m.offsets(); + let (start, end) = (o[0] as usize, o[2] as usize); + let got_keys: Vec<&str> = (start..end).map(|i| keys.value(i)).collect(); + let got_vals: Vec = (start..end).map(|i| vals.value(i)).collect(); + assert_eq!(got_keys, vec!["b", "c", "d"]); + assert_eq!(got_vals, vec![2i32, 3, 4]); + + // The rename-only path transforms no values, so the eval mode cannot change the result. + let ansi = cast_array( + full.slice(1, 2), + &to_type, + &SparkCastOptions::new(EvalMode::Ansi, "UTC", false), + ) + .unwrap(); + assert_eq!(ansi.to_data(), casted.to_data()); + } + + #[test] + fn test_cast_map_to_map_both_paths_agree() { + use arrow::array::{Int32Array, StringArray}; + // Two independent implementations reach the same target type, so pin that they agree. + // The only difference between the two inputs is the source `sorted` flag, which is not + // part of the output type: equal flags select the arrow delegation, differing flags + // select the hand-built path. Everything else (values, offsets, map-level nulls) matches, + // so the two results must be identical. + let rows = |sorted: bool| { + Arc::new(build_str_i32_map( + vec!["a", "b", "c", "d"], + vec![Some(1), None, Some(3), Some(4)], + vec![0, 1, 3, 3, 4], + Some(arrow::buffer::NullBuffer::from(vec![ + true, true, false, true, + ])), + None, + sorted, + )) as ArrayRef + }; + // Slice off the first row so both paths see a non-zero offset window and a null row. + let unsorted_src = rows(false).slice(1, 3); + let sorted_src = rows(true).slice(1, 3); + let to_type = build_to_map_type(DataType::Int32, true, false); + + // Fast path: child types and sort flag all unchanged. + assert_map_child_types_and_sort(unsorted_src.data_type(), &to_type, true, true, true); + // Hand-built path: child types unchanged but the sort flag differs (true -> false). + assert_map_child_types_and_sort(sorted_src.data_type(), &to_type, true, true, false); + + let fast = cast_array(unsorted_src, &to_type, &legacy_opts()).unwrap(); + let hand_built = cast_array(sorted_src, &to_type, &legacy_opts()).unwrap(); + + assert_eq!(fast.data_type(), &to_type); + assert_eq!(hand_built.data_type(), &to_type); + assert_eq!(fast.to_data(), hand_built.to_data()); + + // Assert the shared result is actually right, so agreement on a wrong value cannot pass. + let m = fast.as_any().downcast_ref::().unwrap(); + assert_eq!(m.len(), 3); + assert!(m.is_valid(0) && !m.is_valid(1) && m.is_valid(2)); + let keys = m.keys().as_any().downcast_ref::().unwrap(); + let vals = m.values().as_any().downcast_ref::().unwrap(); + let o = m.offsets(); + let (start, end) = (o[0] as usize, o[3] as usize); + let got_keys: Vec<&str> = (start..end).map(|i| keys.value(i)).collect(); + let got_vals: Vec> = (start..end) + .map(|i| (!vals.is_null(i)).then(|| vals.value(i))) + .collect(); + assert_eq!(got_keys, vec!["b", "c", "d"]); + assert_eq!(got_vals, vec![None, Some(3), Some(4)]); + } + + #[test] + fn test_cast_map_to_map_casts_key_and_value() { + use arrow::array::{Int32Array, Int64Array}; + // Source Map -> target Map: both key and value are cast. + let key_field = Arc::new(Field::new("key_value_key", DataType::Int32, false)); + let value_field = Arc::new(Field::new("key_value_value", DataType::Int32, true)); + let entries_fields = Fields::from(vec![key_field, value_field]); + let ks = Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef; + let vs = Arc::new(Int32Array::from(vec![10, 20])) as ArrayRef; + let entries_struct = StructArray::new(entries_fields, vec![ks, vs], None); + let entries_field = Arc::new(Field::new( + "key_value", + DataType::Struct(entries_struct.fields().clone()), + false, + )); + let src = Arc::new(MapArray::new( + entries_field, + OffsetBuffer::::new(vec![0, 2].into()), + entries_struct, + None, + false, + )) as ArrayRef; + + let to_key = Arc::new(Field::new("key", DataType::Int64, false)); + let to_val = Arc::new(Field::new("value", DataType::Int64, true)); + let to_entries = Arc::new(Field::new( + "entries", + DataType::Struct(Fields::from(vec![to_key, to_val])), + false, + )); + let to_type = DataType::Map(Arc::clone(&to_entries), false); + + let casted = cast_array(src, &to_type, &legacy_opts()).unwrap(); + assert_eq!(casted.data_type(), &to_type); + let m = casted.as_any().downcast_ref::().unwrap(); + let keys = m.keys().as_any().downcast_ref::().unwrap(); + let vals = m.values().as_any().downcast_ref::().unwrap(); + assert_eq!(keys.values(), &[1i64, 2]); + assert_eq!(vals.values(), &[10i64, 20]); + } + + #[test] + fn test_cast_map_to_map_malformed_target_returns_err_without_panic() { + // Source value has a NULL; the target changes the value type (Int32 -> Int64) AND declares + // it NON-nullable. The type change routes through the hand-built child-cast path, and the + // resulting null in a non-nullable field makes `StructArray::try_new` return Err (no panic). + let src = Arc::new(build_str_i32_map( + vec!["a", "b"], + vec![Some(1), None], + vec![0, 2], + None, + None, + false, + )) as ArrayRef; + let to_type = build_to_map_type(DataType::Int64, false, false); + assert!( + cast_array(src, &to_type, &legacy_opts()).is_err(), + "non-nullable target value with null data must return Err via hand-built try_new" + ); + } + + fn one_row_src() -> ArrayRef { + Arc::new(build_str_i32_map( + vec!["a"], + vec![Some(1)], + vec![0, 1], + None, + None, + false, + )) as ArrayRef + } + + fn map_target_with_entry_fields(fields: Vec>) -> DataType { + let entries = Arc::new(Field::new( + "entries", + DataType::Struct(Fields::from(fields)), + false, + )); + DataType::Map(entries, false) + } + + fn entry_field(n: &str, t: DataType) -> Arc { + Arc::new(Field::new(n, t, false)) + } + + // A target whose entries struct does not have exactly (key, value) must return Err, never panic + // on index [0]/[1]. Split per field-count so a baseline panic in one case does not hide others. + #[test] + fn test_cast_map_to_map_zero_entry_fields_errs() { + let to_type = map_target_with_entry_fields(vec![]); + assert!( + cast_array(one_row_src(), &to_type, &legacy_opts()).is_err(), + "0 entry fields must return Err, not panic" + ); + } + + #[test] + fn test_cast_map_to_map_one_entry_field_errs() { + let to_type = map_target_with_entry_fields(vec![entry_field("key", DataType::Utf8)]); + assert!( + cast_array(one_row_src(), &to_type, &legacy_opts()).is_err(), + "1 entry field must return Err, not panic" + ); + } + + #[test] + fn test_cast_map_to_map_three_entry_fields_errs() { + let to_type = map_target_with_entry_fields(vec![ + entry_field("key", DataType::Utf8), + entry_field("value", DataType::Int32), + entry_field("extra", DataType::Int32), + ]); + assert!( + cast_array(one_row_src(), &to_type, &legacy_opts()).is_err(), + "3 entry fields must return Err, not panic" + ); + } } diff --git a/spark/src/test/resources/sql-tests/expressions/cast/cast_complex.sql b/spark/src/test/resources/sql-tests/expressions/cast/cast_complex.sql index 3898b2ac8dc..09f7b606183 100644 --- a/spark/src/test/resources/sql-tests/expressions/cast/cast_complex.sql +++ b/spark/src/test/resources/sql-tests/expressions/cast/cast_complex.sql @@ -149,3 +149,45 @@ query SELECT cast(arr_struct as string), id FROM test_cast_complex ORDER BY id + +-- map value type cast, covering null maps, empty maps and null values +statement +CREATE TABLE test_cast_map(id int, m map) USING parquet + +statement +INSERT INTO test_cast_map VALUES + (1, map('a', 1, 'b', 2)), + (2, map('a', cast(NULL as int))), + (3, map()), + (4, cast(NULL as map)) + +query +SELECT cast(m as map), id +FROM test_cast_map +ORDER BY id + +query +SELECT cast(m as map), id +FROM test_cast_map +ORDER BY id + +-- a key cast that could introduce nulls is rejected, since a null map key is not representable +query expect_error(DATATYPE_MISMATCH) +SELECT cast(m as map) +FROM test_cast_map + +-- map key cast, only legal when the key cast cannot produce null +statement +CREATE TABLE test_cast_map_int_key(id int, m map) USING parquet + +statement +INSERT INTO test_cast_map_int_key VALUES + (1, map(1, 10, 2, 20)), + (2, map(1, cast(NULL as int))), + (3, map()), + (4, cast(NULL as map)) + +query +SELECT cast(m as map), id +FROM test_cast_map_int_key +ORDER BY id