kazuyukitanimura commented on code in PR #6246:
URL: https://github.com/apache/datafusion-comet/pull/6246#discussion_r4299018024
##########
native/core/src/parquet/parquet_exec.rs:
##########
@@ -52,6 +54,81 @@ mod variant_tests;
/// scan would.
pub(crate) const METADATA_SIZE_HINT: usize = 512 * 1024;
+/// Keep top-level Spark string fields as binary while arrow-rs decodes
Parquet. arrow-rs only
+/// validates UTF-8 for columns annotated as strings, so decoding an
unannotated BINARY column that
+/// `spark.sql.parquet.binaryAsString=true` types as a string straight into
Arrow Utf8 would admit
+/// invalid UTF-8. A projection added at the scan boundary converts the binary
arrays to valid Utf8
+/// with Spark-compatible replacement semantics.
+fn binary_as_string_read_schema(schema: &SchemaRef) -> SchemaRef {
+ let fields = schema.fields().iter().map(|field| {
+ binary_as_string_read_type(field.data_type()).map_or_else(
+ || Arc::clone(field),
+ |data_type|
Arc::new(field.as_ref().clone().with_data_type(data_type)),
+ )
+ });
+ Arc::new(Schema::new_with_metadata(
+ fields.collect::<Vec<_>>(),
+ schema.metadata().clone(),
+ ))
+}
+
+/// The type `binary_as_string_read_schema` reads a top-level field of
`data_type` as, or `None`
+/// when the field is read as declared.
+fn binary_as_string_read_type(data_type: &DataType) -> Option<DataType> {
+ match data_type {
+ DataType::Utf8 => Some(DataType::Binary),
+ DataType::LargeUtf8 => Some(DataType::LargeBinary),
+ _ => None,
+ }
+}
+
+/// Whether `expr` references a column whose folded name is in `folded_names`.
+fn references_any_column(
+ expr: &Arc<dyn PhysicalExpr>,
+ folded_names: &HashSet<String>,
+ case_sensitive: bool,
+) -> Result<bool, ExecutionError> {
+ for column in collect_columns(expr) {
+ if folded_names.contains(&fold_name(column.name(), case_sensitive)?) {
+ return Ok(true);
+ }
+ }
+ Ok(false)
+}
+
+fn binary_as_string_projection(
Review Comment:
For a file column that is already annotated UTF8, I think this reads Utf8,
casts to Binary in the adapter, and then casts back to Utf8 in the projection.
Could you benchmark a scan of annotated string columns with
binaryAsString=false and true against branch-1.1? If there is a regression,
deciding per file whether the column is annotated would avoid the double cast.
##########
spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala:
##########
@@ -218,6 +218,55 @@ abstract class ParquetReadSuite extends CometTestBase {
}
}
+ test("spark.sql.parquet.binaryAsString uses native scan") {
Review Comment:
Could you add these cases to the test?
- A WHERE value = 'one' filter, which exercises the dropped-filter path
and compares against Spark.
- An array<string> and a struct<string> built from unannotated binary
with a malformed byte.
- A partitioned or multi-file read
##########
native/core/src/parquet/parquet_exec.rs:
##########
@@ -272,7 +381,22 @@ pub(crate) fn init_datasource_exec(
.filter(|filter| !calls_jvm(filter))
.collect::<Vec<_>>()
});
- let file_source: Arc<dyn FileSource> = match data_filters {
+ // In binaryAsString mode the filters are typed against the Utf8 output
schema, but the scan
+ // reads the string columns as Binary, so drop the filters that reference
one for the same
+ // reason. Filters on other columns still prune.
+ let data_filters = match data_filters {
Review Comment:
Thanks for fixing this. Reading the string columns as Binary and decoding at
the boundary is the right shape. I'm worried about the pruning we lose. Before
this PR, a filter like s = 'x' with binaryAsString=true pruned row groups. Now
it never reaches the reader. Could we keep that pushdown, for example by
rewriting the filter against the Binary read schema with a binary literal for
=, IN and IS NULL? If that isn't feasible here, could you show benchmark
numbers and explain why losing pruning is acceptable for users who turn this
config on?
##########
native/core/src/parquet/parquet_support.rs:
##########
@@ -330,6 +330,24 @@ fn parquet_convert_array_impl(
Ok(Arc::new(string_array))
}
+ // Parquet's unannotated BINARY is read as Spark StringType when
+ // spark.sql.parquet.binaryAsString is enabled or a read schema
declares a string.
+ // Arrow's cast below is safe, so it would turn malformed UTF-8 into
NULL. Use Spark's
+ // cast so the scan's top-level projection and nested values replace
malformed input
+ // with U+FFFD, as the JVM's `new String(bytes, UTF_8)` does.
+ (Binary, Utf8) | (LargeBinary, LargeUtf8) => {
Review Comment:
This arm also changes behavior with binaryAsString=false when a read schema
declares a string over a binary column. Malformed bytes used to become NULL and
now become U+FFFD. That looks like the right Spark behavior. Could you say so
in the PR description, add a test for it, and consider whether this part should
be backported?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
