You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/09/19 18:09:02 UTC
(datafusion) 01/01: Match Parquet map key/value by position in nested schema coercion (#25342)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-25342-42f3888ca2fc82da8548dc4f13533af7af9b5a9f
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 710c748db8931bbe28ade5f25436d0672a19b355
Author: Adrian Garcia Badaracco <17...@users.noreply.github.com>
AuthorDate: Sat Sep 19 18:08:40 2026 +0000
Match Parquet map key/value by position in nested schema coercion (#25342)
# Match Parquet map key/value by position in nested schema coercion
## Which issue does this PR close?
- Follow-up to #25193 (which closed #25192). No separate issue.
## Rationale for this change
#25193 made `apply_file_schema_type_coercions` recurse into nested types
so that, for example, a nested `Utf8` column in a Parquet file is
decoded directly as `Utf8View` when the table schema asks for it,
instead of being cast afterwards.
For `Map` columns the recursion delegated the `key`/`value` children to
the by-name `Struct` path, even though the code comment said container
children match by position. Parquet readers always name map children
`key_value`/`key`/`value`, while Arrow producers commonly use
`entries`/`keys`/`values` (e.g. `MapBuilder` defaults). With that naming
mismatch the view coercion was silently skipped for maps and queries
kept the slower cast path.
## What changes are included in this PR?
- `Map` key and value children are now matched by position, regardless
of their names. Field names, nullability, metadata, the container kind,
`FixedSizeList` width and `Map` ordering continue to come from the file
schema.
- `apply_file_schema_type_coercions` is restructured around a recursive
`coerce_data_type` helper that operates on `DataType`s directly. This
removes the throwaway single-field `Schema`s built for each nested
level, the `needs_*` pre-scan flags (the result is compared to the input
instead), and an unreachable match arm in the container branch (the one
uncovered line in the #25193 codecov report).
- The doc comment now describes the nested matching rules (structs by
name, lists and maps by position) and which parts of the result always
come from the file.
Behaviour is otherwise unchanged; the existing nested parquet `.slt`
files (`schema_evolution_nested`, `parquet_nested_schema_pruning`,
`parquet_filter_pushdown`) pass.
## What is the testing strategy for this PR?
Unit tests in `schema_coercion.rs`:
- `nested_coercion_preserves_list_containers` now uses a table-side
child that differs from the file-side child in name, nullability,
metadata, `FixedSizeList` width and `Map` ordering, and asserts the
file's values win. Mutation testing against #25193 showed that taking
the width, ordering or child field from the table survived the previous
version of this test; a wrong `FixedSizeList` width is accepted silently
by parquet-rs and would decode data with the wrong list size.
- `nested_coercion_matches_map_entries_by_position`: Arrow-named table
map vs Parquet-named file map, plus a map with a different number of
entry children that must be left alone.
- `nested_coercion_map_reader_produces_string_views`: writes a file with
Parquet's map naming and no embedded Arrow schema, coerces against an
Arrow-named table schema, and checks the Parquet reader accepts the
schema and produces `StringViewArray` keys and values.
- `nested_coercion_list_inside_struct`.
The two map tests fail on `main` without the fix.
## Are there any user-facing changes?
No API changes. Map columns whose table schema uses `keys`/`values`
naming now get the same nested view/string coercion as structs and
lists, avoiding a cast during scan.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
https://claude.ai/code/session_01LA4Rks1XaN8wxpszj1hXQQ
---------
Co-authored-by: Claude <no...@anthropic.com>
Co-authored-by: Copilot Autofix powered by AI <17...@users.noreply.github.com>
---
.../datasource-parquet/src/schema_coercion.rs | 741 ++++++++++++++++-----
1 file changed, 577 insertions(+), 164 deletions(-)
diff --git a/datafusion/datasource-parquet/src/schema_coercion.rs b/datafusion/datasource-parquet/src/schema_coercion.rs
index ce51ab33fd..828070659b 100644
--- a/datafusion/datasource-parquet/src/schema_coercion.rs
+++ b/datafusion/datasource-parquet/src/schema_coercion.rs
@@ -22,13 +22,23 @@
//! These helpers are independent of the [`ParquetFormat`](crate::file_format::ParquetFormat)
//! type and several have been re-exported at the crate root for use by
//! callers outside the format implementation.
+//!
+//! # Binary to string coercion
+//!
+//! Coercing a binary file column to a string type makes the Parquet reader
+//! build the string array itself. It only validates UTF-8 for columns carrying
+//! the `UTF8` logical annotation, which such a column by definition does not,
+//! so invalid bytes surface as an invalid string array rather than an error.
+//! That is long-standing behaviour for top-level and struct fields; it is not
+//! extended to map children, whose positional matching is newer, so binary map
+//! values keep going through the validating cast instead.
use std::cell::RefCell;
use std::collections::{HashMap, HashSet};
use std::rc::Rc;
use std::sync::Arc;
-use arrow::datatypes::{DataType, Field, FieldRef, Schema, TimeUnit};
+use arrow::datatypes::{DataType, Field, FieldRef, Fields, Schema, TimeUnit};
use parquet::basic::Type;
use parquet::schema::types::SchemaDescriptor;
@@ -40,6 +50,21 @@ use parquet::schema::types::SchemaDescriptor;
/// 2. Regular to view types conversion - Converts standard string/binary types to
/// view types when the table schema uses view types
///
+/// The same coercions are applied recursively to nested types: children of
+/// [`DataType::Struct`] fields are matched by name, while the children of
+/// list-like types ([`DataType::List`], [`DataType::LargeList`],
+/// [`DataType::ListView`], [`DataType::LargeListView`],
+/// [`DataType::FixedSizeList`]) and the key/value children of
+/// [`DataType::Map`] are matched by position, regardless of their names.
+///
+/// Only leaf data types are ever changed: field names, nullability, metadata,
+/// the list/map container kind, [`DataType::FixedSizeList`] width and
+/// [`DataType::Map`] ordering are always taken from `file_schema`, since the
+/// resulting schema must still describe the data physically present in the
+/// file (see [`ArrowReaderOptions::with_schema`]).
+///
+/// [`ArrowReaderOptions::with_schema`]: parquet::arrow::arrow_reader::ArrowReaderOptions::with_schema
+///
/// # Arguments
/// * `table_schema` - The table schema containing the desired types
/// * `file_schema` - The file schema to be transformed
@@ -51,158 +76,170 @@ pub fn apply_file_schema_type_coercions(
table_schema: &Schema,
file_schema: &Schema,
) -> Option<Schema> {
- let mut needs_view_transform = false;
- let mut needs_string_transform = false;
- let mut needs_nested_transform = false;
+ let fields =
+ coerce_fields_by_name(table_schema.fields(), file_schema.fields(), true)?;
+ Some(Schema::new_with_metadata(
+ fields,
+ file_schema.metadata.clone(),
+ ))
+}
+/// Coerce `file_fields` towards `table_fields`, matching fields by name.
+///
+/// File fields with no counterpart in `table_fields` are kept unchanged and
+/// table fields missing from the file are ignored. Returns `None` if no field
+/// changed.
+fn coerce_fields_by_name(
+ table_fields: &Fields,
+ file_fields: &Fields,
+ binary_to_string: bool,
+) -> Option<Fields> {
// Create a mapping of table field names to their data types for fast lookup
- // and simultaneously check if we need any transformations
- let table_fields: HashMap<_, _> = table_schema
- .fields()
+ let table_types: HashMap<_, _> = table_fields
.iter()
- .map(|f| {
- let dt = f.data_type();
- // Check if we need view type transformation
- if matches!(dt, &DataType::Utf8View | &DataType::BinaryView) {
- needs_view_transform = true;
- }
- // Check if we need string type transformation
- if matches!(
- dt,
- &DataType::Utf8 | &DataType::LargeUtf8 | &DataType::Utf8View
- ) {
- needs_string_transform = true;
- }
- // Nested fields can need transformations even when their parent does not.
- if matches!(
- dt,
- DataType::Struct(_)
- | DataType::List(_)
- | DataType::LargeList(_)
- | DataType::ListView(_)
- | DataType::LargeListView(_)
- | DataType::FixedSizeList(_, _)
- | DataType::Map(_, _)
- ) {
- needs_nested_transform = true;
- }
-
- (f.name(), dt)
- })
+ .map(|f| (f.name(), f.data_type()))
.collect();
- // Early return if no transformation needed
- if !needs_view_transform && !needs_string_transform && !needs_nested_transform {
- return None;
- }
+ coerce_fields(file_fields, |_, field| {
+ let table_type = table_types.get(field.name())?;
+ coerce_data_type(table_type, field.data_type(), binary_to_string)
+ .map(|new_type| field_with_new_type(field, new_type))
+ })
+}
- let transformed_fields: Vec<Arc<Field>> = file_schema
- .fields()
- .iter()
- .map(|field| {
- let field_name = field.name();
- let field_type = field.data_type();
-
- // Look up the corresponding field type in the table schema
- if let Some(table_type) = table_fields.get(field_name) {
- match (table_type, field_type) {
- // table schema uses string type, coerce the file schema to use string type
- (
- &DataType::Utf8,
- DataType::Binary | DataType::LargeBinary | DataType::BinaryView,
- ) => {
- return field_with_new_type(field, DataType::Utf8);
- }
- // table schema uses large string type, coerce the file schema to use large string type
- (
- &DataType::LargeUtf8,
- DataType::Binary | DataType::LargeBinary | DataType::BinaryView,
- ) => {
- return field_with_new_type(field, DataType::LargeUtf8);
- }
- // table schema uses string view type, coerce the file schema to use view type
- (
- &DataType::Utf8View,
- DataType::Binary | DataType::LargeBinary | DataType::BinaryView,
- ) => {
- return field_with_new_type(field, DataType::Utf8View);
- }
- // Handle view type conversions
- (&DataType::Utf8View, DataType::Utf8 | DataType::LargeUtf8) => {
- return field_with_new_type(field, DataType::Utf8View);
- }
- (&DataType::BinaryView, DataType::Binary | DataType::LargeBinary) => {
- return field_with_new_type(field, DataType::BinaryView);
- }
- // Apply the same coercions to matching fields inside structs.
- (DataType::Struct(table_fields), DataType::Struct(file_fields)) => {
- if let Some(schema) = apply_file_schema_type_coercions(
- &Schema::new(table_fields.clone()),
- &Schema::new(file_fields.clone()),
- ) {
- return field_with_new_type(
- field,
- DataType::Struct(schema.fields),
- );
- }
- }
- // Container children match by position, regardless of their names.
- (DataType::List(table_child), DataType::List(file_child))
- | (
- DataType::LargeList(table_child),
- DataType::LargeList(file_child),
- )
- | (DataType::ListView(table_child), DataType::ListView(file_child))
- | (
- DataType::LargeListView(table_child),
- DataType::LargeListView(file_child),
- )
- | (
- DataType::FixedSizeList(table_child, _),
- DataType::FixedSizeList(file_child, _),
- )
- | (DataType::Map(table_child, _), DataType::Map(file_child, _)) => {
- if let Some(schema) = apply_file_schema_type_coercions(
- &Schema::new(vec![field_with_new_type(
- file_child,
- table_child.data_type().clone(),
- )]),
- &Schema::new(vec![Arc::clone(file_child)]),
- ) {
- let child = Arc::clone(&schema.fields()[0]);
- let new_type = match field_type {
- DataType::List(_) => DataType::List(child),
- DataType::LargeList(_) => DataType::LargeList(child),
- DataType::ListView(_) => DataType::ListView(child),
- DataType::LargeListView(_) => {
- DataType::LargeListView(child)
- }
- DataType::FixedSizeList(_, size) => {
- DataType::FixedSizeList(child, *size)
- }
- DataType::Map(_, sorted) => DataType::Map(child, *sorted),
- _ => return Arc::clone(field),
- };
- return field_with_new_type(field, new_type);
- }
- }
- _ => {}
+/// Rebuild `file_fields`, replacing every field for which `coerce` returns a
+/// new one. Returns `None` if no field changed.
+///
+/// The output is only allocated once a field actually changes, so schemas
+/// needing no coercion at all (the common case) are walked without allocating
+/// or touching the reference counts of the file fields.
+fn coerce_fields(
+ file_fields: &Fields,
+ mut coerce: impl FnMut(usize, &FieldRef) -> Option<FieldRef>,
+) -> Option<Fields> {
+ let mut coerced: Option<Vec<FieldRef>> = None;
+ for (idx, field) in file_fields.iter().enumerate() {
+ match coerce(idx, field) {
+ Some(new_field) => coerced
+ .get_or_insert_with(|| {
+ // The fields before the first change are carried over as is
+ let mut fields = Vec::with_capacity(file_fields.len());
+ fields.extend_from_slice(&file_fields[..idx]);
+ fields
+ })
+ .push(new_field),
+ // Unchanged fields are only copied once something else changed
+ None => {
+ if let Some(coerced) = &mut coerced {
+ coerced.push(Arc::clone(field));
}
}
+ }
+ }
- // If no transformation is needed, keep the original field
- Arc::clone(field)
- })
- .collect();
+ coerced.map(Fields::from)
+}
- if transformed_fields.iter().eq(file_schema.fields().iter()) {
+/// Coerce `file_type` towards `table_type`, recursing into nested types.
+///
+/// `binary_to_string` allows binary file types to be read as string types.
+/// Parquet only validates UTF-8 when the column carries the `UTF8` logical
+/// annotation, which a binary column by definition does not, so such a
+/// conversion hands back an unvalidated string array; see the module docs.
+///
+/// Returns the new type for the file field, or `None` if no transformation
+/// is needed (including when the two types are unrelated).
+fn coerce_data_type(
+ table_type: &DataType,
+ file_type: &DataType,
+ binary_to_string: bool,
+) -> Option<DataType> {
+ use DataType::*;
+ match (table_type, file_type) {
+ // table schema uses string type, coerce the file schema to use string type
+ (Utf8, Binary | LargeBinary | BinaryView) if binary_to_string => Some(Utf8),
+ // table schema uses large string type, coerce the file schema to use large string type
+ (LargeUtf8, Binary | LargeBinary | BinaryView) if binary_to_string => {
+ Some(LargeUtf8)
+ }
+ // table schema uses string view type, coerce the file schema to use view type
+ (Utf8View, Binary | LargeBinary | BinaryView) if binary_to_string => {
+ Some(Utf8View)
+ }
+ (Utf8View, Utf8 | LargeUtf8) => Some(Utf8View),
+ (BinaryView, Binary | LargeBinary) => Some(BinaryView),
+ // Struct children match by name
+ (Struct(table_fields), Struct(file_fields)) => {
+ coerce_fields_by_name(table_fields, file_fields, binary_to_string).map(Struct)
+ }
+ // List-like children match by position, regardless of their names.
+ // The container kind and FixedSizeList width always come from the file.
+ (List(table_child), List(file_child)) => {
+ coerce_child(table_child, file_child, binary_to_string).map(List)
+ }
+ (LargeList(table_child), LargeList(file_child)) => {
+ coerce_child(table_child, file_child, binary_to_string).map(LargeList)
+ }
+ (ListView(table_child), ListView(file_child)) => {
+ coerce_child(table_child, file_child, binary_to_string).map(ListView)
+ }
+ (LargeListView(table_child), LargeListView(file_child)) => {
+ coerce_child(table_child, file_child, binary_to_string).map(LargeListView)
+ }
+ (FixedSizeList(table_child, _), FixedSizeList(file_child, size)) => {
+ coerce_child(table_child, file_child, binary_to_string)
+ .map(|child| FixedSizeList(child, *size))
+ }
+ // Map keys and values match by position: Parquet always names them
+ // `key`/`value` while Arrow producers commonly use `keys`/`values`.
+ (Map(table_entries, _), Map(file_entries, sorted)) => {
+ coerce_map_entries(table_entries, file_entries)
+ .map(|entries| Map(entries, *sorted))
+ }
+ _ => None,
+ }
+}
+
+/// Coerce a single nested child field, keeping everything but its data type
+/// from `file_child`.
+fn coerce_child(
+ table_child: &FieldRef,
+ file_child: &FieldRef,
+ binary_to_string: bool,
+) -> Option<FieldRef> {
+ coerce_data_type(
+ table_child.data_type(),
+ file_child.data_type(),
+ binary_to_string,
+ )
+ .map(|new_type| field_with_new_type(file_child, new_type))
+}
+
+/// Coerce the `entries` struct of a [`DataType::Map`], matching the key and
+/// value children by position.
+///
+/// Binary children are never read as strings here: matching by position newly
+/// reaches map children whose names differ, and turning those into unvalidated
+/// string arrays would be a regression over leaving them to the (validating)
+/// cast. See the module docs.
+fn coerce_map_entries(
+ table_entries: &FieldRef,
+ file_entries: &FieldRef,
+) -> Option<FieldRef> {
+ let (DataType::Struct(table_fields), DataType::Struct(file_fields)) =
+ (table_entries.data_type(), file_entries.data_type())
+ else {
+ return None;
+ };
+ if table_fields.len() != file_fields.len() {
return None;
}
- Some(Schema::new_with_metadata(
- transformed_fields,
- file_schema.metadata.clone(),
- ))
+ let fields = coerce_fields(file_fields, |idx, file_child| {
+ coerce_child(&table_fields[idx], file_child, false)
+ })?;
+
+ Some(field_with_new_type(file_entries, DataType::Struct(fields)))
}
/// Coerces the file schema's Timestamps to the provided TimeUnit if the
@@ -607,7 +644,11 @@ mod tests {
#[test]
fn nested_coercion_preserves_list_containers() {
- let wrap: [fn(FieldRef) -> DataType; 6] = [
+ // The table side deliberately differs from the file side in the child
+ // field name, nullability and metadata, as well as in the
+ // FixedSizeList width and Map ordering: all of those must be taken
+ // from the file, only the leaf types come from the table.
+ let file_wrap: [fn(FieldRef) -> DataType; 6] = [
DataType::List,
DataType::LargeList,
DataType::ListView,
@@ -615,37 +656,51 @@ mod tests {
|field| DataType::FixedSizeList(field, 3),
|field| DataType::Map(field, false),
];
- for wrap in wrap {
- let file_element = Arc::new(Field::new_struct(
- "entries",
- vec![
- Field::new("key", DataType::Utf8, false),
- Field::new("value", DataType::Binary, true),
- ],
- false,
- ));
+ let table_wrap: [fn(FieldRef) -> DataType; 6] = [
+ DataType::List,
+ DataType::LargeList,
+ DataType::ListView,
+ DataType::LargeListView,
+ |field| DataType::FixedSizeList(field, 7),
+ |field| DataType::Map(field, true),
+ ];
+ for (file_wrap, table_wrap) in file_wrap.into_iter().zip(table_wrap) {
+ let file_element = Arc::new(
+ Field::new_struct(
+ "key_value",
+ vec![
+ Field::new("key", DataType::Utf8, false),
+ Field::new("value", DataType::Utf8, true),
+ ],
+ false,
+ )
+ .with_metadata(HashMap::from([("source".into(), "file".into())])),
+ );
let table_element = Arc::new(Field::new_struct(
"entries",
vec![
- Field::new("value", DataType::Utf8View, true),
- Field::new("key", DataType::Utf8View, false),
+ Field::new("value", DataType::Utf8View, false),
+ Field::new("key", DataType::Utf8View, true),
],
- false,
- ));
- let expected_element = Arc::new(Field::new_struct(
- "entries",
- vec![
- Field::new("key", DataType::Utf8View, false),
- Field::new("value", DataType::Utf8View, true),
- ],
- false,
+ true,
));
+ let expected_element = Arc::new(
+ Field::new_struct(
+ "key_value",
+ vec![
+ Field::new("key", DataType::Utf8View, false),
+ Field::new("value", DataType::Utf8View, true),
+ ],
+ false,
+ )
+ .with_metadata(HashMap::from([("source".into(), "file".into())])),
+ );
let file_schema =
- Schema::new(vec![Field::new("data", wrap(file_element), true)]);
+ Schema::new(vec![Field::new("data", file_wrap(file_element), true)]);
let table_schema =
- Schema::new(vec![Field::new("data", wrap(table_element), true)]);
+ Schema::new(vec![Field::new("data", table_wrap(table_element), false)]);
let expected =
- Schema::new(vec![Field::new("data", wrap(expected_element), true)]);
+ Schema::new(vec![Field::new("data", file_wrap(expected_element), true)]);
assert_eq!(
apply_file_schema_type_coercions(&table_schema, &file_schema),
Some(expected)
@@ -653,6 +708,364 @@ mod tests {
}
}
+ #[test]
+ fn nested_coercion_matches_map_entries_by_position() {
+ // Parquet readers always name map children `key_value`/`key`/`value`,
+ // while Arrow producers (e.g. `MapBuilder`) default to
+ // `entries`/`keys`/`values`. Names must not matter for maps.
+ let file_schema = Schema::new(vec![Field::new_map(
+ "m",
+ "key_value",
+ Field::new("key", DataType::Utf8, false),
+ Field::new("value", DataType::Binary, true),
+ false,
+ true,
+ )]);
+ let table_schema = Schema::new(vec![Field::new_map(
+ "m",
+ "entries",
+ Field::new("keys", DataType::Utf8View, false),
+ Field::new("values", DataType::Utf8View, true),
+ false,
+ true,
+ )]);
+ // The key is matched by position and coerced to a view, while the
+ // binary value is left to the validating cast (see
+ // `coerce_map_entries`)
+ let expected = Schema::new(vec![Field::new_map(
+ "m",
+ "key_value",
+ Field::new("key", DataType::Utf8View, false),
+ Field::new("value", DataType::Binary, true),
+ false,
+ true,
+ )]);
+ assert_eq!(
+ apply_file_schema_type_coercions(&table_schema, &file_schema),
+ Some(expected)
+ );
+
+ // A map whose entries do not have the same shape is left alone
+ let odd_table_schema = Schema::new(vec![Field::new(
+ "m",
+ DataType::Map(
+ Arc::new(Field::new_struct(
+ "entries",
+ vec![Field::new("keys", DataType::Utf8View, false)],
+ false,
+ )),
+ false,
+ ),
+ true,
+ )]);
+ assert_eq!(
+ apply_file_schema_type_coercions(&odd_table_schema, &file_schema),
+ None
+ );
+ }
+
+ #[test]
+ fn nested_coercion_map_entry_struct_children_match_by_name() {
+ // A map whose key and value are both structs: the two entry children
+ // are matched by position (the file's `key`/`value` take the types of
+ // the table's first/second entry child, whatever they are named),
+ // while the fields inside those structs go back to matching by name.
+ //
+ // Both structs put `shared` in a different position on each side, so a
+ // positional match inside them would coerce `file_only` instead, and a
+ // by-name match on the entries themselves would coerce nothing at all.
+ let struct_field =
+ |name: &str, first: (&str, DataType), second: (&str, DataType)| {
+ Field::new_struct(
+ name,
+ vec![
+ Field::new(first.0, first.1, true),
+ Field::new(second.0, second.1, true),
+ ],
+ true,
+ )
+ };
+ let file_schema = Schema::new(vec![Field::new_map(
+ "m",
+ "key_value",
+ struct_field(
+ "key",
+ ("file_only", DataType::Utf8),
+ ("shared", DataType::Utf8),
+ ),
+ struct_field(
+ "value",
+ ("file_only", DataType::Utf8),
+ ("shared", DataType::Utf8),
+ ),
+ false,
+ true,
+ )]);
+ let table_schema = Schema::new(vec![Field::new_map(
+ "m",
+ "entries",
+ struct_field(
+ "keys",
+ ("shared", DataType::Utf8View),
+ ("table_only", DataType::Utf8View),
+ ),
+ struct_field(
+ "values",
+ ("shared", DataType::Utf8View),
+ ("table_only", DataType::Utf8View),
+ ),
+ false,
+ true,
+ )]);
+ let expected = Schema::new(vec![Field::new_map(
+ "m",
+ "key_value",
+ struct_field(
+ "key",
+ ("file_only", DataType::Utf8),
+ ("shared", DataType::Utf8View),
+ ),
+ struct_field(
+ "value",
+ ("file_only", DataType::Utf8),
+ ("shared", DataType::Utf8View),
+ ),
+ false,
+ true,
+ )]);
+ assert_eq!(
+ apply_file_schema_type_coercions(&table_schema, &file_schema),
+ Some(expected)
+ );
+ }
+
+ #[test]
+ fn nested_coercion_leaves_binary_map_children_to_the_validating_cast() {
+ use arrow::array::{
+ Array, ArrayRef, AsArray, BinaryBuilder, MapBuilder, MapFieldNames,
+ StringBuilder,
+ };
+ use arrow::record_batch::RecordBatch;
+ use bytes::Bytes;
+ use parquet::arrow::ArrowWriter;
+ use parquet::arrow::arrow_reader::{
+ ArrowReaderOptions, ParquetRecordBatchReaderBuilder,
+ };
+ use parquet::arrow::arrow_writer::ArrowWriterOptions;
+
+ // A binary map value holding bytes that are not valid UTF-8. Reading it
+ // as a string would produce an invalid array, since the Parquet reader
+ // only validates columns annotated `UTF8`, so the value must be left
+ // binary for the (validating) cast to reject.
+ let parquet_names = MapFieldNames {
+ entry: "key_value".to_string(),
+ key: "key".to_string(),
+ value: "value".to_string(),
+ };
+ let mut builder = MapBuilder::new(
+ Some(parquet_names),
+ StringBuilder::new(),
+ BinaryBuilder::new(),
+ );
+ builder.keys().append_value("k1");
+ builder.values().append_value([0xff]);
+ builder.append(true).unwrap();
+ let batch =
+ RecordBatch::try_from_iter([("m", Arc::new(builder.finish()) as ArrayRef)])
+ .unwrap();
+
+ let mut bytes = Vec::new();
+ let mut writer = ArrowWriter::try_new_with_options(
+ &mut bytes,
+ batch.schema(),
+ ArrowWriterOptions::new().with_skip_arrow_metadata(true),
+ )
+ .unwrap();
+ writer.write(&batch).unwrap();
+ writer.close().unwrap();
+ let bytes = Bytes::from(bytes);
+
+ let file_schema = ParquetRecordBatchReaderBuilder::try_new(bytes.clone())
+ .unwrap()
+ .schema()
+ .clone();
+ let table_schema = Schema::new(vec![Field::new_map(
+ "m",
+ "entries",
+ Field::new("keys", DataType::Utf8View, false),
+ Field::new("values", DataType::Utf8View, true),
+ false,
+ true,
+ )]);
+
+ // The key is still coerced by position, the binary value is not
+ let coerced =
+ apply_file_schema_type_coercions(&table_schema, &file_schema).unwrap();
+ assert_eq!(
+ coerced.field(0).data_type(),
+ &DataType::Map(
+ Arc::new(Field::new_struct(
+ "key_value",
+ vec![
+ Field::new("key", DataType::Utf8View, false),
+ Field::new("value", DataType::Binary, true),
+ ],
+ false,
+ )),
+ false,
+ )
+ );
+
+ let mut reader = ParquetRecordBatchReaderBuilder::try_new_with_options(
+ bytes,
+ ArrowReaderOptions::new().with_schema(Arc::new(coerced)),
+ )
+ .unwrap()
+ .build()
+ .unwrap();
+ let decoded = reader.next().unwrap().unwrap();
+ // The decoded batch is valid: the invalid bytes are still binary
+ decoded.column(0).to_data().validate_full().unwrap();
+ assert_eq!(
+ decoded
+ .column(0)
+ .as_map()
+ .values()
+ .as_binary::<i32>()
+ .value(0),
+ [0xff]
+ );
+ }
+
+ #[test]
+ fn nested_coercion_list_inside_struct() {
+ let file_schema = Schema::new(vec![Field::new_struct(
+ "s",
+ vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new_list("tags", Field::new("item", DataType::Utf8, true), true),
+ ],
+ true,
+ )]);
+ let table_schema = Schema::new(vec![Field::new_struct(
+ "s",
+ vec![Field::new_list(
+ "tags",
+ Field::new("element", DataType::Utf8View, false),
+ false,
+ )],
+ false,
+ )]);
+ let expected = Schema::new(vec![Field::new_struct(
+ "s",
+ vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new_list(
+ "tags",
+ Field::new("item", DataType::Utf8View, true),
+ true,
+ ),
+ ],
+ true,
+ )]);
+ assert_eq!(
+ apply_file_schema_type_coercions(&table_schema, &file_schema),
+ Some(expected)
+ );
+ }
+
+ #[test]
+ fn nested_coercion_map_reader_produces_string_views() {
+ use arrow::array::{
+ Array, ArrayRef, AsArray, MapBuilder, MapFieldNames, StringBuilder,
+ };
+ use arrow::record_batch::RecordBatch;
+ use bytes::Bytes;
+ use parquet::arrow::ArrowWriter;
+ use parquet::arrow::arrow_reader::{
+ ArrowReaderOptions, ParquetRecordBatchReaderBuilder,
+ };
+ use parquet::arrow::arrow_writer::ArrowWriterOptions;
+
+ // Build the file with Parquet's `key_value`/`key`/`value` naming, as
+ // written by non-Arrow writers, while the table schema below uses
+ // Arrow's default `entries`/`keys`/`values` naming.
+ let parquet_names = MapFieldNames {
+ entry: "key_value".to_string(),
+ key: "key".to_string(),
+ value: "value".to_string(),
+ };
+ let mut builder = MapBuilder::new(
+ Some(parquet_names),
+ StringBuilder::new(),
+ StringBuilder::new(),
+ );
+ builder.keys().append_value("k1");
+ builder.values().append_value("v1");
+ builder.append(true).unwrap();
+ builder.append(false).unwrap();
+ let map = builder.finish();
+ let batch =
+ RecordBatch::try_from_iter([("m", Arc::new(map) as ArrayRef)]).unwrap();
+ let table_schema = Schema::new(vec![Field::new_map(
+ "m",
+ "entries",
+ Field::new("keys", DataType::Utf8View, false),
+ Field::new("values", DataType::Utf8View, true),
+ false,
+ true,
+ )]);
+
+ // Write without the embedded Arrow schema, as a non-Arrow writer would
+ let mut bytes = Vec::new();
+ let mut writer = ArrowWriter::try_new_with_options(
+ &mut bytes,
+ batch.schema(),
+ ArrowWriterOptions::new().with_skip_arrow_metadata(true),
+ )
+ .unwrap();
+ writer.write(&batch).unwrap();
+ writer.close().unwrap();
+ let bytes = Bytes::from(bytes);
+
+ let file_schema = ParquetRecordBatchReaderBuilder::try_new(bytes.clone())
+ .unwrap()
+ .schema()
+ .clone();
+ assert_eq!(
+ file_schema.field(0).data_type(),
+ &DataType::Map(
+ Arc::new(Field::new_struct(
+ "key_value",
+ vec![
+ Field::new("key", DataType::Utf8, false),
+ Field::new("value", DataType::Utf8, true),
+ ],
+ false,
+ )),
+ false,
+ )
+ );
+
+ let reader_schema = Arc::new(
+ apply_file_schema_type_coercions(&table_schema, &file_schema).unwrap(),
+ );
+ let mut reader = ParquetRecordBatchReaderBuilder::try_new_with_options(
+ bytes,
+ ArrowReaderOptions::new().with_schema(Arc::clone(&reader_schema)),
+ )
+ .unwrap()
+ .build()
+ .unwrap();
+ let decoded = reader.next().unwrap().unwrap();
+ assert_eq!(decoded.schema(), reader_schema);
+ let map = decoded.column(0).as_map();
+ assert_eq!(map.keys().as_string_view().value(0), "k1");
+ assert_eq!(map.values().as_string_view().value(0), "v1");
+ assert!(map.is_null(1));
+ assert!(reader.next().is_none());
+ }
+
#[test]
fn nested_coercion_preserves_leaf_conversion_rules() {
let nested_schema = |data_type| {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org