diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index fc7202a6c5a45..c0da43821fcfc 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -1380,6 +1380,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 3d904290b222c..41ed7a9c757aa 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: _, @@ -412,6 +413,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, @@ -546,6 +548,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/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); +} diff --git a/datafusion/datasource-parquet/src/file_format.rs b/datafusion/datasource-parquet/src/file_format.rs index 50e76d2d211e1..2a1e1fff0e32c 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 0c7c1ee28031d..f74dc7515cb03 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 24429f2fd1fad..9ea8fe0159962 100644 --- a/datafusion/proto-common/src/from_proto/mod.rs +++ b/datafusion/proto-common/src/from_proto/mod.rs @@ -1284,6 +1284,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 22b291843def5..0774f5f8e8efc 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 6d3cadc577acb..bec43122dee33 100644 --- a/datafusion/proto-models/src/from_proto.rs +++ b/datafusion/proto-models/src/from_proto.rs @@ -471,6 +471,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 25ada65f69ba2..9b9d6e39407d1 100644 --- a/datafusion/sqllogictest/test_files/information_schema.slt +++ b/datafusion/sqllogictest/test_files/information_schema.slt @@ -237,6 +237,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 @@ -402,6 +403,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 44868ae2c977a..ff42b77d7b873 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -89,6 +89,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 |