From bef8ffd05db26e7a3eba823e902ee284188cf44f Mon Sep 17 00:00:00 2001 From: Manu Zhang Date: Sun, 27 Sep 2026 23:00:24 +0800 Subject: [PATCH 1/2] feat(parquet): add coerce_binary_to_string option to read binary file columns as binary When a table schema declares a string type for a column the Parquet file stores as unannotated BINARY, the opener always coerces the file schema so the reader decodes the column straight into a string array. The reader only validates UTF-8 for UTF8-annotated columns, so invalid bytes produce an invalid string array (and panic in debug builds), and callers cannot apply their own conversion because the coercion runs before the PhysicalExprAdapter. Add datafusion.execution.parquet.coerce_binary_to_string (default true). When false, binary file columns stay binary and the adapter casts them to the table type, so the default adapter rejects invalid UTF-8 and a custom adapter can convert it. Expose apply_file_schema_type_coercions_with_options for callers of the coercion helper. Co-Authored-By: Claude Opus 5.5 (1M context) --- datafusion/common/src/config.rs | 12 ++ .../common/src/file_options/parquet_writer.rs | 9 +- .../datasource-parquet/src/file_format.rs | 1 + datafusion/datasource-parquet/src/mod.rs | 3 +- .../datasource-parquet/src/opener/mod.rs | 93 ++++++++++++++ .../datasource-parquet/src/schema_coercion.rs | 116 +++++++++++++++--- datafusion/datasource-parquet/src/source.rs | 4 + .../proto/datafusion_common.proto | 1 + datafusion/proto-common/src/from_proto/mod.rs | 1 + .../proto-common/src/generated/pbjson.rs | 18 +++ .../proto-common/src/generated/prost.rs | 3 + datafusion/proto-common/src/to_proto/mod.rs | 1 + datafusion/proto-models/src/from_proto.rs | 1 + .../src/generated/datafusion_proto_common.rs | 3 + .../test_files/information_schema.slt | 2 + docs/source/user-guide/configs.md | 1 + 16 files changed, 251 insertions(+), 18 deletions(-) diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index b2cfdbfc2e34e..da50d50525304 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -1403,6 +1403,18 @@ config_namespace! { /// reading them as binary and then casting to string. pub binary_as_string: bool, default = false + /// (reading) If true, a column the parquet file stores as binary is + /// read directly as a string when the table schema declares a string + /// type for it. The parquet reader only validates UTF-8 for columns + /// with the `UTF8` annotation, so invalid bytes in such a column + /// produce an invalid string array. + /// + /// If false, such a column is read as binary and cast to the table's + /// string type by the `PhysicalExprAdapter`, which rejects invalid + /// UTF-8 unless a custom adapter converts it differently. Statistics + /// and bloom filter pruning do not apply to predicates on these columns. + pub coerce_binary_to_string: bool, default = true + /// (reading) If set, parquet reader will read columns of /// physical type int96 as originating from this resolution instead of /// nanosecond. Valid values are: ns, us, ms, and s. This is useful for diff --git a/datafusion/common/src/file_options/parquet_writer.rs b/datafusion/common/src/file_options/parquet_writer.rs index f6485c2f54b61..528638aa44be8 100644 --- a/datafusion/common/src/file_options/parquet_writer.rs +++ b/datafusion/common/src/file_options/parquet_writer.rs @@ -244,9 +244,10 @@ impl ParquetOptions { maximum_buffered_record_batches_per_stream: _, bloom_filter_on_read: _, // reads not used for writer props schema_force_view_types: _, - binary_as_string: _, // not used for writer props - coerce_int96: _, // not used for writer props - coerce_int96_tz: _, // not used for writer props + binary_as_string: _, // not used for writer props + coerce_binary_to_string: _, // not used for writer props + coerce_int96: _, // not used for writer props + coerce_int96_tz: _, // not used for writer props skip_arrow_metadata: _, max_predicate_cache_size: _, max_in_list_size: _, @@ -431,6 +432,7 @@ mod tests { bloom_filter_on_read: defaults.bloom_filter_on_read, schema_force_view_types: defaults.schema_force_view_types, binary_as_string: defaults.binary_as_string, + coerce_binary_to_string: defaults.coerce_binary_to_string, skip_arrow_metadata: defaults.skip_arrow_metadata, coerce_int96: None, coerce_int96_tz: None, @@ -562,6 +564,7 @@ mod tests { .max_predicate_cache_size, schema_force_view_types: global_options_defaults.schema_force_view_types, binary_as_string: global_options_defaults.binary_as_string, + coerce_binary_to_string: global_options_defaults.coerce_binary_to_string, skip_arrow_metadata: global_options_defaults.skip_arrow_metadata, write_row_group_number_distinct_values: props .write_row_group_number_distinct_values(), diff --git a/datafusion/datasource-parquet/src/file_format.rs b/datafusion/datasource-parquet/src/file_format.rs index ca5a4bddec5a8..26da50460e9ca 100644 --- a/datafusion/datasource-parquet/src/file_format.rs +++ b/datafusion/datasource-parquet/src/file_format.rs @@ -754,6 +754,7 @@ impl From<&ParquetFormatFactory> for protobuf::TableParquetOptions { maximum_buffered_record_batches_per_stream: global_options.global.maximum_buffered_record_batches_per_stream as u64, schema_force_view_types: global_options.global.schema_force_view_types, binary_as_string: global_options.global.binary_as_string, + coerce_binary_to_string: global_options.global.coerce_binary_to_string, skip_arrow_metadata: global_options.global.skip_arrow_metadata, write_row_group_number_distinct_values: global_options .global diff --git a/datafusion/datasource-parquet/src/mod.rs b/datafusion/datasource-parquet/src/mod.rs index 6837e0519486f..288475a4048a5 100644 --- a/datafusion/datasource-parquet/src/mod.rs +++ b/datafusion/datasource-parquet/src/mod.rs @@ -64,7 +64,8 @@ pub use row_group_filter::RowGroupAccessPlanFilter; #[expect(deprecated)] pub use schema_coercion::coerce_int96_to_resolution; pub use schema_coercion::{ - Int96Coercer, apply_file_schema_type_coercions, transform_binary_to_string, + Int96Coercer, apply_file_schema_type_coercions, + apply_file_schema_type_coercions_with_options, transform_binary_to_string, transform_schema_to_view, }; pub use sink::ParquetSink; diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs index b048a23eea896..300da7feb932a 100644 --- a/datafusion/datasource-parquet/src/opener/mod.rs +++ b/datafusion/datasource-parquet/src/opener/mod.rs @@ -284,6 +284,10 @@ pub(super) struct ParquetMorselizer { /// coerced column type becomes `Timestamp(, Some())`. /// No effect when `coerce_int96` is `None`. pub coerce_int96_tz: Option>, + /// Read binary file columns directly as strings when the table schema + /// declares a string type. When `false`, they are read as binary and the + /// `expr_adapter_factory`'s adapter casts them instead. + pub coerce_binary_to_string: bool, /// Optional parquet FileDecryptionProperties #[cfg(feature = "parquet_encryption")] pub file_decryption_properties: Option>, @@ -496,6 +500,7 @@ struct PreparedParquetOpen { limit: Option, coerce_int96: Option, coerce_int96_tz: Option>, + coerce_binary_to_string: bool, expr_adapter_factory: Arc, predicate_creation_errors: Count, max_predicate_cache_size: Option, @@ -999,6 +1004,7 @@ impl ParquetMorselizer { limit: self.limit, coerce_int96: self.coerce_int96, coerce_int96_tz: self.coerce_int96_tz.clone(), + coerce_binary_to_string: self.coerce_binary_to_string, expr_adapter_factory: Arc::clone(&self.expr_adapter_factory), predicate_creation_errors, max_predicate_cache_size: self.max_predicate_cache_size, @@ -1123,6 +1129,7 @@ impl MetadataLoadedParquetOpen { if let Some(merged) = apply_file_schema_type_coercions_with_rle( &prepared.logical_file_schema, &physical_file_schema, + prepared.coerce_binary_to_string, prepared.enable_rle_to_dictionary, ) { physical_file_schema = Arc::new(merged); @@ -2215,6 +2222,7 @@ mod test { enable_bloom_filter: bool, enable_row_group_stats_pruning: bool, coerce_int96: Option, + coerce_binary_to_string: bool, max_predicate_cache_size: Option, max_in_list_size: usize, enable_rle_to_dictionary: bool, @@ -2435,6 +2443,7 @@ mod test { enable_bloom_filter: false, enable_row_group_stats_pruning: false, coerce_int96: None, + coerce_binary_to_string: true, max_predicate_cache_size: None, max_in_list_size: MAX_IN_LIST_SIZE, enable_rle_to_dictionary: false, @@ -2549,6 +2558,12 @@ mod test { self } + /// Set whether binary file columns are read directly as strings. + fn with_coerce_binary_to_string(mut self, enable: bool) -> Self { + self.coerce_binary_to_string = enable; + self + } + /// Set whether the scan must preserve file order. fn with_preserve_order(mut self, enable: bool) -> Self { self.preserve_order = enable; @@ -2632,6 +2647,7 @@ mod test { enable_bloom_filter: self.enable_bloom_filter, enable_row_group_stats_pruning: self.enable_row_group_stats_pruning, coerce_int96: self.coerce_int96, + coerce_binary_to_string: self.coerce_binary_to_string, // End-to-end coercion behavior (including timezone) is // covered by parquet.slt. No opener-level test currently // needs a non-default value here. @@ -6051,4 +6067,81 @@ mod test { &DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Binary)) ); } + + /// Write a single unannotated `BINARY` column `b` holding `values`, with no + /// `ARROW:schema` hint, so the reader infers `Binary` for it. + async fn write_unannotated_binary_file( + store: &dyn ObjectStore, + name: &str, + values: &[&[u8]], + ) -> PartitionedFile { + use parquet::data_type::{ByteArray, ByteArrayType}; + use parquet::file::writer::SerializedFileWriter; + use parquet::schema::parser::parse_message_type; + + let schema = + Arc::new(parse_message_type("message test { required binary b; }").unwrap()); + let mut bytes = Vec::new(); + let mut writer = SerializedFileWriter::new( + &mut bytes, + schema, + Arc::new(WriterProperties::builder().build()), + ) + .unwrap(); + let mut row_group = writer.next_row_group().unwrap(); + let mut column = row_group.next_column().unwrap().unwrap(); + let values: Vec = + values.iter().map(|v| ByteArray::from(v.to_vec())).collect(); + column + .typed::() + .write_batch(&values, None, None) + .unwrap(); + column.close().unwrap(); + row_group.close().unwrap(); + writer.close().unwrap(); + let path = Path::from(name); + store.put(&path, bytes.into()).await.unwrap(); + PartitionedFile::from(store.head(&path).await.unwrap()) + } + + /// With `coerce_binary_to_string` disabled, a binary file column read into + /// a string table column is decoded as binary and converted by the + /// expression adapter's cast, so invalid UTF-8 is rejected by that cast + /// rather than surfacing as an invalid string array. + #[tokio::test] + async fn test_coerce_binary_to_string_disabled_casts_in_adapter() { + let store = Arc::new(InMemory::new()) as Arc; + let table_schema = + Arc::new(Schema::new(vec![Field::new("b", DataType::Utf8, false)])); + let morselizer = ParquetMorselizerBuilder::new() + .with_store(Arc::clone(&store)) + .with_schema(table_schema) + .with_coerce_binary_to_string(false) + .build(); + + let file = write_unannotated_binary_file( + store.as_ref(), + "valid.parquet", + &[b"a", b"bc"], + ) + .await; + let stream = open_file(&morselizer, file).await.unwrap(); + let batches: Vec = stream.try_collect().await.unwrap(); + assert_eq!(batches.len(), 1); + let values = batches[0].column(0).as_string::(); + assert_eq!( + values.iter().collect::>(), + vec![Some("a"), Some("bc")] + ); + + let file = write_unannotated_binary_file( + store.as_ref(), + "invalid.parquet", + &[b"a", &[0xff]], + ) + .await; + let stream = open_file(&morselizer, file).await.unwrap(); + let err = stream.try_collect::>().await.unwrap_err(); + assert_contains!(err.to_string(), "Encountered non UTF-8 data"); + } } diff --git a/datafusion/datasource-parquet/src/schema_coercion.rs b/datafusion/datasource-parquet/src/schema_coercion.rs index fbfc37a33f31b..6b71bf50cda30 100644 --- a/datafusion/datasource-parquet/src/schema_coercion.rs +++ b/datafusion/datasource-parquet/src/schema_coercion.rs @@ -32,6 +32,13 @@ //! 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. +//! +//! A scan can turn this coercion off with the +//! `datafusion.execution.parquet.coerce_binary_to_string` option (see +//! [`apply_file_schema_type_coercions_with_options`]). Such columns are then +//! read as binary and converted by the `PhysicalExprAdapter`'s cast, which +//! rejects invalid UTF-8 by default and which a custom adapter can replace +//! with its own conversion. use std::cell::RefCell; use std::collections::{HashMap, HashSet}; @@ -76,26 +83,49 @@ pub fn apply_file_schema_type_coercions( table_schema: &Schema, file_schema: &Schema, ) -> Option { - let fields = - coerce_fields_by_name(table_schema.fields(), file_schema.fields(), true, false)?; - Some(Schema::new_with_metadata( - fields, - file_schema.metadata.clone(), - )) + apply_file_schema_type_coercions_with_options(table_schema, file_schema, true) } -/// Like [`apply_file_schema_type_coercions`], but also coerces compatible -/// string/binary file fields to dictionary types already present in the table -/// schema. +/// Like [`apply_file_schema_type_coercions`], with control over the binary to +/// string coercion. +/// +/// When `binary_to_string` is `false`, binary file columns stay binary even +/// when the table schema declares a string type for them. The Parquet reader +/// then decodes them as binary and the [`PhysicalExprAdapter`] casts them to +/// the table type. The regular to view type coercions are unaffected. +/// +/// The reader only validates UTF-8 for columns with the `UTF8` annotation, so +/// the coercion turns invalid bytes in a binary column into an invalid string +/// array. Disabling it lets the adapter's cast reject those bytes, or lets a +/// custom adapter convert them its own way. +/// +/// [`PhysicalExprAdapter`]: datafusion_physical_expr_adapter::PhysicalExprAdapter +pub fn apply_file_schema_type_coercions_with_options( + table_schema: &Schema, + file_schema: &Schema, + binary_to_string: bool, +) -> Option { + apply_file_schema_type_coercions_with_rle( + table_schema, + file_schema, + binary_to_string, + false, + ) +} + +/// Like [`apply_file_schema_type_coercions_with_options`], but also coerces +/// compatible string/binary file fields to dictionary types already present in +/// the table schema. pub(crate) fn apply_file_schema_type_coercions_with_rle( table_schema: &Schema, file_schema: &Schema, + binary_to_string: bool, enable_rle_to_dictionary: bool, ) -> Option { let fields = coerce_fields_by_name( table_schema.fields(), file_schema.fields(), - true, + binary_to_string, enable_rle_to_dictionary, )?; Some(Schema::new_with_metadata( @@ -794,6 +824,59 @@ mod tests { use parquet::schema::parser::parse_message_type; + /// Disabling the binary to string coercion leaves binary file columns + /// binary at every nesting level, while view type coercions still apply. + #[test] + fn binary_to_string_coercion_can_be_disabled() { + let list_of = + |data_type| DataType::List(Arc::new(Field::new("item", data_type, true))); + let file_schema = Schema::new(vec![ + Field::new("b", DataType::Binary, true), + Field::new_struct( + "s", + vec![Field::new("b", DataType::LargeBinary, true)], + true, + ), + Field::new("l", list_of(DataType::Binary), true), + Field::new("u", DataType::Utf8, true), + ]); + let table_schema = Schema::new(vec![ + Field::new("b", DataType::Utf8, true), + Field::new_struct("s", vec![Field::new("b", DataType::Utf8, true)], true), + Field::new("l", list_of(DataType::Utf8View), true), + Field::new("u", DataType::Utf8View, true), + ]); + + let expected = Schema::new(vec![ + Field::new("b", DataType::Binary, true), + Field::new_struct( + "s", + vec![Field::new("b", DataType::LargeBinary, true)], + true, + ), + Field::new("l", list_of(DataType::Binary), true), + Field::new("u", DataType::Utf8View, true), + ]); + assert_eq!( + apply_file_schema_type_coercions_with_options( + &table_schema, + &file_schema, + false + ), + Some(expected) + ); + + // The default still reads every binary column as a string. + let coerced = + apply_file_schema_type_coercions(&table_schema, &file_schema).unwrap(); + assert_eq!(coerced.field(0).data_type(), &DataType::Utf8); + assert_eq!( + coerced.field(1).data_type(), + &DataType::Struct(vec![Field::new("b", DataType::Utf8, true)].into()) + ); + assert_eq!(coerced.field(2).data_type(), &list_of(DataType::Utf8View)); + } + #[test] fn nested_coercion_preserves_file_schema() { let field_metadata = HashMap::from([("source".into(), "file".into())]); @@ -1990,6 +2073,7 @@ mod tests { &table_schema, &file_schema, true, + true, ) .as_ref() .map(|s| s.field(0).data_type().clone()) @@ -2006,9 +2090,13 @@ mod tests { Field::new("dict_col", DataType::Utf8, true), Field::new("string_col", DataType::Binary, true), ]); - let result = - apply_file_schema_type_coercions_with_rle(&table_schema, &file_schema, false) - .unwrap(); + let result = apply_file_schema_type_coercions_with_rle( + &table_schema, + &file_schema, + true, + false, + ) + .unwrap(); assert_eq!(result.field(0).data_type(), &DataType::Utf8); assert_eq!(result.field(1).data_type(), &DataType::Utf8); @@ -2123,7 +2211,7 @@ mod tests { let coerce = |table: DataType, file: DataType| { let t = one_field_schema(table); let f = one_field_schema(file.clone()); - apply_file_schema_type_coercions_with_rle(&t, &f, true) + apply_file_schema_type_coercions_with_rle(&t, &f, true, true) .as_ref() .map(|s| s.field(0).data_type().clone()) .unwrap_or(file) diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs index 2a8b5d428f5ee..ecfed52453612 100644 --- a/datafusion/datasource-parquet/src/source.rs +++ b/datafusion/datasource-parquet/src/source.rs @@ -651,6 +651,10 @@ impl FileSource for ParquetSource { enable_row_group_stats_pruning: self.table_parquet_options.global.pruning, coerce_int96, coerce_int96_tz, + coerce_binary_to_string: self + .table_parquet_options + .global + .coerce_binary_to_string, #[cfg(feature = "parquet_encryption")] file_decryption_properties, expr_adapter_factory, diff --git a/datafusion/proto-common/proto/datafusion_common.proto b/datafusion/proto-common/proto/datafusion_common.proto index ba6e6937daf38..cd8dfcc6661be 100644 --- a/datafusion/proto-common/proto/datafusion_common.proto +++ b/datafusion/proto-common/proto/datafusion_common.proto @@ -579,6 +579,7 @@ message ParquetOptions { bool bloom_filter_on_write = 27; // default = false bool schema_force_view_types = 28; // default = false bool binary_as_string = 29; // default = false + bool coerce_binary_to_string = 42; // default = true bool skip_arrow_metadata = 30; // default = false bool write_row_group_number_distinct_values = 41; // default = false diff --git a/datafusion/proto-common/src/from_proto/mod.rs b/datafusion/proto-common/src/from_proto/mod.rs index 42b7c4976cbce..72b39fd721f24 100644 --- a/datafusion/proto-common/src/from_proto/mod.rs +++ b/datafusion/proto-common/src/from_proto/mod.rs @@ -1283,6 +1283,7 @@ impl TryFrom<&protobuf::ParquetOptions> for ParquetOptions { )?, schema_force_view_types: value.schema_force_view_types, binary_as_string: value.binary_as_string, + coerce_binary_to_string: value.coerce_binary_to_string, coerce_int96: value.coerce_int96_opt.as_ref().map(|opt| match opt { protobuf::parquet_options::CoerceInt96Opt::CoerceInt96(v) => v.parse(), }).transpose()?, diff --git a/datafusion/proto-common/src/generated/pbjson.rs b/datafusion/proto-common/src/generated/pbjson.rs index 09386a01eb556..5fc5117f1184e 100644 --- a/datafusion/proto-common/src/generated/pbjson.rs +++ b/datafusion/proto-common/src/generated/pbjson.rs @@ -6493,6 +6493,9 @@ impl serde::Serialize for ParquetOptions { if self.binary_as_string { len += 1; } + if self.coerce_binary_to_string { + len += 1; + } if self.skip_arrow_metadata { len += 1; } @@ -6619,6 +6622,9 @@ impl serde::Serialize for ParquetOptions { if self.binary_as_string { struct_ser.serialize_field("binaryAsString", &self.binary_as_string)?; } + if self.coerce_binary_to_string { + struct_ser.serialize_field("coerceBinaryToString", &self.coerce_binary_to_string)?; + } if self.skip_arrow_metadata { struct_ser.serialize_field("skipArrowMetadata", &self.skip_arrow_metadata)?; } @@ -6801,6 +6807,8 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { "schemaForceViewTypes", "binary_as_string", "binaryAsString", + "coerce_binary_to_string", + "coerceBinaryToString", "skip_arrow_metadata", "skipArrowMetadata", "write_row_group_number_distinct_values", @@ -6865,6 +6873,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { BloomFilterOnWrite, SchemaForceViewTypes, BinaryAsString, + CoerceBinaryToString, SkipArrowMetadata, WriteRowGroupNumberDistinctValues, DictionaryPageSizeLimit, @@ -6925,6 +6934,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { "bloomFilterOnWrite" | "bloom_filter_on_write" => Ok(GeneratedField::BloomFilterOnWrite), "schemaForceViewTypes" | "schema_force_view_types" => Ok(GeneratedField::SchemaForceViewTypes), "binaryAsString" | "binary_as_string" => Ok(GeneratedField::BinaryAsString), + "coerceBinaryToString" | "coerce_binary_to_string" => Ok(GeneratedField::CoerceBinaryToString), "skipArrowMetadata" | "skip_arrow_metadata" => Ok(GeneratedField::SkipArrowMetadata), "writeRowGroupNumberDistinctValues" | "write_row_group_number_distinct_values" => Ok(GeneratedField::WriteRowGroupNumberDistinctValues), "dictionaryPageSizeLimit" | "dictionary_page_size_limit" => Ok(GeneratedField::DictionaryPageSizeLimit), @@ -6983,6 +6993,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { let mut bloom_filter_on_write__ = None; let mut schema_force_view_types__ = None; let mut binary_as_string__ = None; + let mut coerce_binary_to_string__ = None; let mut skip_arrow_metadata__ = None; let mut write_row_group_number_distinct_values__ = None; let mut dictionary_page_size_limit__ = None; @@ -7112,6 +7123,12 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { } binary_as_string__ = Some(map_.next_value()?); } + GeneratedField::CoerceBinaryToString => { + if coerce_binary_to_string__.is_some() { + return Err(serde::de::Error::duplicate_field("coerceBinaryToString")); + } + coerce_binary_to_string__ = Some(map_.next_value()?); + } GeneratedField::SkipArrowMetadata => { if skip_arrow_metadata__.is_some() { return Err(serde::de::Error::duplicate_field("skipArrowMetadata")); @@ -7277,6 +7294,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { bloom_filter_on_write: bloom_filter_on_write__.unwrap_or_default(), schema_force_view_types: schema_force_view_types__.unwrap_or_default(), binary_as_string: binary_as_string__.unwrap_or_default(), + coerce_binary_to_string: coerce_binary_to_string__.unwrap_or_default(), skip_arrow_metadata: skip_arrow_metadata__.unwrap_or_default(), write_row_group_number_distinct_values: write_row_group_number_distinct_values__.unwrap_or_default(), dictionary_page_size_limit: dictionary_page_size_limit__.unwrap_or_default(), diff --git a/datafusion/proto-common/src/generated/prost.rs b/datafusion/proto-common/src/generated/prost.rs index ee586fa75ee37..cbed1d1c9de6b 100644 --- a/datafusion/proto-common/src/generated/prost.rs +++ b/datafusion/proto-common/src/generated/prost.rs @@ -866,6 +866,9 @@ pub struct ParquetOptions { /// default = false #[prost(bool, tag = "29")] pub binary_as_string: bool, + /// default = true + #[prost(bool, tag = "42")] + pub coerce_binary_to_string: bool, /// default = false #[prost(bool, tag = "30")] pub skip_arrow_metadata: bool, diff --git a/datafusion/proto-common/src/to_proto/mod.rs b/datafusion/proto-common/src/to_proto/mod.rs index 01f37a478f624..ac0e8e305750b 100644 --- a/datafusion/proto-common/src/to_proto/mod.rs +++ b/datafusion/proto-common/src/to_proto/mod.rs @@ -965,6 +965,7 @@ impl TryFrom<&ParquetOptions> for protobuf::ParquetOptions { maximum_buffered_record_batches_per_stream: value.maximum_buffered_record_batches_per_stream as u64, schema_force_view_types: value.schema_force_view_types, binary_as_string: value.binary_as_string, + coerce_binary_to_string: value.coerce_binary_to_string, skip_arrow_metadata: value.skip_arrow_metadata, write_row_group_number_distinct_values: value.write_row_group_number_distinct_values, coerce_int96_opt: value.coerce_int96.map(|v| protobuf::parquet_options::CoerceInt96Opt::CoerceInt96(v.to_string())), diff --git a/datafusion/proto-models/src/from_proto.rs b/datafusion/proto-models/src/from_proto.rs index cdf7c30d00670..4829ae7761f27 100644 --- a/datafusion/proto-models/src/from_proto.rs +++ b/datafusion/proto-models/src/from_proto.rs @@ -469,6 +469,7 @@ impl TryFrom<&ParquetOptionsProto> for ParquetOptions { )?, schema_force_view_types: proto.schema_force_view_types, binary_as_string: proto.binary_as_string, + coerce_binary_to_string: proto.coerce_binary_to_string, skip_arrow_metadata: proto.skip_arrow_metadata, write_row_group_number_distinct_values: proto .write_row_group_number_distinct_values, diff --git a/datafusion/proto-models/src/generated/datafusion_proto_common.rs b/datafusion/proto-models/src/generated/datafusion_proto_common.rs index ee586fa75ee37..cbed1d1c9de6b 100644 --- a/datafusion/proto-models/src/generated/datafusion_proto_common.rs +++ b/datafusion/proto-models/src/generated/datafusion_proto_common.rs @@ -866,6 +866,9 @@ pub struct ParquetOptions { /// default = false #[prost(bool, tag = "29")] pub binary_as_string: bool, + /// default = true + #[prost(bool, tag = "42")] + pub coerce_binary_to_string: bool, /// default = false #[prost(bool, tag = "30")] pub skip_arrow_metadata: bool, diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt index 2381fc14e3d88..60c78466ca32b 100644 --- a/datafusion/sqllogictest/test_files/information_schema.slt +++ b/datafusion/sqllogictest/test_files/information_schema.slt @@ -241,6 +241,7 @@ datafusion.execution.parquet.bloom_filter_fpp NULL datafusion.execution.parquet.bloom_filter_ndv NULL datafusion.execution.parquet.bloom_filter_on_read true datafusion.execution.parquet.bloom_filter_on_write false +datafusion.execution.parquet.coerce_binary_to_string true datafusion.execution.parquet.coerce_int96 NULL datafusion.execution.parquet.coerce_int96_tz NULL datafusion.execution.parquet.column_index_truncate_length 64 @@ -410,6 +411,7 @@ datafusion.execution.parquet.bloom_filter_fpp NULL (writing) Sets bloom filter f datafusion.execution.parquet.bloom_filter_ndv NULL (writing) Sets bloom filter number of distinct values. If NULL, uses default parquet writer setting datafusion.execution.parquet.bloom_filter_on_read true (reading) Use any available bloom filters when reading parquet files datafusion.execution.parquet.bloom_filter_on_write false (writing) Write bloom filters for all columns when creating parquet files +datafusion.execution.parquet.coerce_binary_to_string true (reading) If true, a column the parquet file stores as binary is read directly as a string when the table schema declares a string type for it. The parquet reader only validates UTF-8 for columns with the `UTF8` annotation, so invalid bytes in such a column produce an invalid string array. If false, such a column is read as binary and cast to the table's string type by the `PhysicalExprAdapter`, which rejects invalid UTF-8 unless a custom adapter converts it differently. Statistics and bloom filter pruning do not apply to predicates on these columns. datafusion.execution.parquet.coerce_int96 NULL (reading) If set, parquet reader will read columns of physical type int96 as originating from this resolution instead of nanosecond. Valid values are: ns, us, ms, and s. This is useful for reading data from systems like Spark which store microsecond resolution timestamps in an int96, allowing them to write values with a larger date range than 64-bit timestamps with nanosecond resolution. datafusion.execution.parquet.coerce_int96_tz NULL (reading) Optional timezone applied to INT96 columns when `coerce_int96` is set. When `Some`, INT96 columns coerce to `Timestamp(, Some())` instead of the default `Timestamp(, None)`. Spark and other systems write INT96 values as UTC-adjusted instants, so callers that need the resulting Arrow type to be timezone-aware (e.g. for Spark `TimestampType` semantics) should set this to `"UTC"`. No effect when `coerce_int96` is `None`. datafusion.execution.parquet.column_index_truncate_length 64 (writing) Sets column index truncate length diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index 273b0e39cfedd..cc2232a315f94 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -93,6 +93,7 @@ The following configuration settings are available: | datafusion.execution.parquet.force_filter_selections | false | (reading) Force the use of RowSelections for filter results, when pushdown_filters is enabled. If false, the reader will automatically choose between a RowSelection and a Bitmap based on the number and pattern of selected rows. | | datafusion.execution.parquet.schema_force_view_types | true | (reading) If true, parquet reader will read columns of `Utf8/Utf8Large` with `Utf8View`, and `Binary/BinaryLarge` with `BinaryView`. The parquet reader is optimized for reading `Utf8View` and `BinaryView`, so such queries are significantly faster than reading `Utf8`/`Binary` and then casting to the view types. | | datafusion.execution.parquet.binary_as_string | false | (reading) If true, parquet reader will read columns of `Binary/LargeBinary` with `Utf8`, and `BinaryView` with `Utf8View`. Parquet files generated by some legacy writers do not correctly set the UTF8 flag for strings, causing string columns to be loaded as BLOB instead. The parquet reader has special optimizations for `Utf8` validation, so reading such columns as strings is significantly faster than reading them as binary and then casting to string. | +| datafusion.execution.parquet.coerce_binary_to_string | true | (reading) If true, a column the parquet file stores as binary is read directly as a string when the table schema declares a string type for it. The parquet reader only validates UTF-8 for columns with the `UTF8` annotation, so invalid bytes in such a column produce an invalid string array. If false, such a column is read as binary and cast to the table's string type by the `PhysicalExprAdapter`, which rejects invalid UTF-8 unless a custom adapter converts it differently. Statistics and bloom filter pruning do not apply to predicates on these columns. | | datafusion.execution.parquet.coerce_int96 | NULL | (reading) If set, parquet reader will read columns of physical type int96 as originating from this resolution instead of nanosecond. Valid values are: ns, us, ms, and s. This is useful for reading data from systems like Spark which store microsecond resolution timestamps in an int96, allowing them to write values with a larger date range than 64-bit timestamps with nanosecond resolution. | | datafusion.execution.parquet.coerce_int96_tz | NULL | (reading) Optional timezone applied to INT96 columns when `coerce_int96` is set. When `Some`, INT96 columns coerce to `Timestamp(, Some())` instead of the default `Timestamp(, None)`. Spark and other systems write INT96 values as UTC-adjusted instants, so callers that need the resulting Arrow type to be timezone-aware (e.g. for Spark `TimestampType` semantics) should set this to `"UTC"`. No effect when `coerce_int96` is `None`. | | datafusion.execution.parquet.bloom_filter_on_read | true | (reading) Use any available bloom filters when reading parquet files | From 8dec0ea76409a134c3fd0f0d9ce1b39a57f6d40a Mon Sep 17 00:00:00 2001 From: Manu Zhang Date: Tue, 20 Oct 2026 19:17:21 +0800 Subject: [PATCH 2/2] test(parquet): cover coerce_binary_to_string through session config Add scan tests that turn off datafusion.execution.parquet.coerce_binary_to_string with SET and read a binary file column into a Utf8 ListingTable column, so a regression in the path from the session config through ParquetSource to the opener is caught. With the default adapter, valid values come back as strings in the projection and in a pushed down filter, and invalid UTF-8 fails with the cast error. A custom adapter that turns invalid UTF-8 into U+FFFD shows that callers can apply their own conversion. Co-Authored-By: Claude Opus 5.5 --- datafusion/core/tests/parquet/expr_adapter.rs | 224 +++++++++++++++++- 1 file changed, 219 insertions(+), 5 deletions(-) diff --git a/datafusion/core/tests/parquet/expr_adapter.rs b/datafusion/core/tests/parquet/expr_adapter.rs index dfed2eb5bda74..385122bd3b36d 100644 --- a/datafusion/core/tests/parquet/expr_adapter.rs +++ b/datafusion/core/tests/parquet/expr_adapter.rs @@ -18,9 +18,9 @@ use std::sync::Arc; use arrow::array::{ - Array, ArrayRef, BooleanArray, FixedSizeListArray, Int32Array, Int64Array, - LargeListArray, ListArray, MapArray, RecordBatch, StringArray, StructArray, - record_batch, + Array, ArrayRef, AsArray, BinaryArray, BooleanArray, FixedSizeListArray, Int32Array, + Int64Array, LargeListArray, ListArray, MapArray, RecordBatch, StringArray, + StructArray, record_batch, }; use arrow::buffer::{NullBuffer, OffsetBuffer}; use arrow::compute::concat_batches; @@ -34,11 +34,14 @@ use datafusion::datasource::listing::{ use datafusion::prelude::{SessionConfig, SessionContext}; use datafusion_common::DataFusionError; use datafusion_common::ScalarValue; +use datafusion_common::config::ConfigOptions; use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode}; +use datafusion_common::{assert_contains, exec_err}; use datafusion_datasource::ListingTableUrl; use datafusion_execution::object_store::ObjectStoreUrl; -use datafusion_physical_expr::PhysicalExpr; -use datafusion_physical_expr::expressions::{self, Column}; +use datafusion_expr::{ColumnarValue, Volatility, create_udf}; +use datafusion_physical_expr::expressions::{self, CastExpr, Column}; +use datafusion_physical_expr::{PhysicalExpr, ScalarFunctionExpr}; use datafusion_physical_expr_adapter::{ DefaultPhysicalExprAdapter, DefaultPhysicalExprAdapterFactory, PhysicalExprAdapter, PhysicalExprAdapterFactory, @@ -1581,3 +1584,214 @@ async fn test_physical_expr_adapter_factory_reuse_across_tables() { ]; assert_batches_eq!(expected, &batches); } + +/// Register a table `t` with a `Utf8` column `b` over a file that stores `b` +/// as unannotated `BINARY`, after disabling +/// `datafusion.execution.parquet.coerce_binary_to_string` for the session. +async fn register_binary_as_utf8_table( + values: &[&[u8]], + expr_adapter_factory: Arc, +) -> SessionContext { + let ctx = test_context(); + ctx.sql("SET datafusion.execution.parquet.coerce_binary_to_string = false") + .await + .unwrap(); + + let store = Arc::new(InMemory::new()) as Arc; + let batch = RecordBatch::try_from_iter([( + "b", + Arc::new(BinaryArray::from(values.to_vec())) as ArrayRef, + )]) + .unwrap(); + write_parquet(batch, Arc::clone(&store), "binary.parquet").await; + let store_url = ObjectStoreUrl::parse("memory://").unwrap(); + ctx.register_object_store(store_url.as_ref(), store); + + let table_schema = Arc::new(Schema::new(vec![Field::new("b", DataType::Utf8, true)])); + let listing_table_config = + ListingTableConfig::new(ListingTableUrl::parse("memory:///").unwrap()) + .infer_options(&ctx.state()) + .await + .unwrap() + .with_schema(table_schema) + .with_expr_adapter_factory(expr_adapter_factory); + let table = ListingTable::try_new(listing_table_config).unwrap(); + ctx.register_table("t", Arc::new(table)).unwrap(); + ctx +} + +/// With `coerce_binary_to_string` disabled in the session configuration, the +/// default adapter casts binary file columns to the table's string type, in +/// both the projection and the pushed down filter, and rejects invalid UTF-8. +#[tokio::test] +async fn test_binary_to_string_cast_by_default_adapter() { + let ctx = register_binary_as_utf8_table( + &[b"a", b"bc"], + Arc::new(DefaultPhysicalExprAdapterFactory), + ) + .await; + + let batches = ctx + .sql("SELECT b FROM t") + .await + .unwrap() + .collect() + .await + .unwrap(); + #[rustfmt::skip] + let expected = [ + "+----+", + "| b |", + "+----+", + "| a |", + "| bc |", + "+----+", + ]; + assert_batches_eq!(expected, &batches); + + let batches = ctx + .sql("SELECT b FROM t WHERE b = 'bc'") + .await + .unwrap() + .collect() + .await + .unwrap(); + #[rustfmt::skip] + let expected = [ + "+----+", + "| b |", + "+----+", + "| bc |", + "+----+", + ]; + assert_batches_eq!(expected, &batches); + + let ctx = register_binary_as_utf8_table( + &[b"a", b"b\xff"], + Arc::new(DefaultPhysicalExprAdapterFactory), + ) + .await; + let err = ctx + .sql("SELECT b FROM t") + .await + .unwrap() + .collect() + .await + .unwrap_err(); + assert_contains!(err.to_string(), "Encountered non UTF-8 data"); +} + +/// Replaces the default adapter's cast from a binary file column to `Utf8` +/// with a conversion that turns invalid UTF-8 into U+FFFD, as Spark does. +#[derive(Debug)] +struct LossyUtf8AdapterFactory; + +impl PhysicalExprAdapterFactory for LossyUtf8AdapterFactory { + fn create( + &self, + logical_file_schema: SchemaRef, + physical_file_schema: SchemaRef, + ) -> Result> { + Ok(Arc::new(LossyUtf8Adapter { + inner: DefaultPhysicalExprAdapter::new( + logical_file_schema, + Arc::clone(&physical_file_schema), + ), + physical_file_schema, + })) + } +} + +#[derive(Debug)] +struct LossyUtf8Adapter { + inner: DefaultPhysicalExprAdapter, + physical_file_schema: SchemaRef, +} + +impl PhysicalExprAdapter for LossyUtf8Adapter { + fn rewrite(&self, expr: Arc) -> Result> { + self.inner + .rewrite(expr)? + .transform(|expr| { + if let Some(cast) = expr.downcast_ref::() + && cast.cast_type() == &DataType::Utf8 + && cast.expr().data_type(&self.physical_file_schema)? + == DataType::Binary + { + let lossy_utf8 = create_udf( + "lossy_utf8", + vec![DataType::Binary], + DataType::Utf8, + Volatility::Immutable, + Arc::new(|args: &[ColumnarValue]| { + let [ColumnarValue::Array(array)] = args else { + return exec_err!("lossy_utf8 expects one array"); + }; + let strings: StringArray = array + .as_binary::() + .iter() + .map(|value| value.map(String::from_utf8_lossy)) + .collect(); + Ok(ColumnarValue::Array(Arc::new(strings))) + }), + ); + return Ok(Transformed::yes(Arc::new(ScalarFunctionExpr::new( + "lossy_utf8", + Arc::new(lossy_utf8), + vec![Arc::clone(cast.expr())], + Arc::clone(cast.target_field()), + Arc::new(ConfigOptions::default()), + )))); + } + Ok(Transformed::no(expr)) + }) + .data() + } +} + +/// With `coerce_binary_to_string` disabled in the session configuration, a +/// custom adapter can convert binary file columns its own way, in both the +/// projection and the pushed down filter. +#[tokio::test] +async fn test_binary_to_string_converted_by_custom_adapter() { + let ctx = register_binary_as_utf8_table( + &[b"a", b"b\xff"], + Arc::new(LossyUtf8AdapterFactory), + ) + .await; + + let batches = ctx + .sql("SELECT b FROM t") + .await + .unwrap() + .collect() + .await + .unwrap(); + #[rustfmt::skip] + let expected = [ + "+----+", + "| b |", + "+----+", + "| a |", + "| b\u{FFFD} |", + "+----+", + ]; + assert_batches_eq!(expected, &batches); + + let batches = ctx + .sql("SELECT b FROM t WHERE b = 'b\u{FFFD}'") + .await + .unwrap() + .collect() + .await + .unwrap(); + #[rustfmt::skip] + let expected = [ + "+----+", + "| b |", + "+----+", + "| b\u{FFFD} |", + "+----+", + ]; + assert_batches_eq!(expected, &batches); +}