comphead commented on code in PR #6246:
URL: https://github.com/apache/datafusion-comet/pull/6246#discussion_r4354854700
##########
native/core/src/parquet/parquet_exec.rs:
##########
@@ -197,10 +384,41 @@ pub(crate) fn init_datasource_exec(
.flat_map(|s| s.fields().iter())
.map(|f| Arc::new(Field::new(f.name(), f.data_type().clone(),
f.is_nullable())))
.collect();
- let table_schema = TableSchema::builder(base_schema)
- .with_table_partition_cols(partition_fields)
+ let output_table_schema = TableSchema::builder(Arc::clone(&base_schema))
+ .with_table_partition_cols(partition_fields.clone())
+ .build();
+ let read_base_schema = if binary_as_string {
+ binary_as_string_read_schema(&base_schema)
+ } else {
+ Arc::clone(&base_schema)
+ };
+ let table_schema = TableSchema::builder(read_base_schema)
+ .with_table_partition_cols(partition_fields.clone())
.build();
+ // Folded names of the columns read as binary. Filters that reference one
are rewritten for
+ // the binary read below, or kept out of the scan.
+ let binary_read_columns: HashSet<String> = if binary_as_string {
Review Comment:
The planner binds data filters by position against `required_schema`
followed by the partition columns, and `filter_read_schema` already relies on
that. So `reads_binary` could check `column.index() <
required_schema.fields().len()` and
`required_schema.field(column.index()).data_type() == &DataType::Utf8`, and
`references_any_column` could reuse that check over `collect_columns`. The
folded-name `HashSet` and the new `fold_name` and `fold_names` imports would
then go. Gating `binary_projection` and `read_base_schema` on at least one such
column existing would also keep scans with no string column on the existing
`with_projection_indices` path. I read the binding order in `planner.rs` and
did not run it.
##########
docs/source/user-guide/latest/parquet.md:
##########
@@ -264,11 +270,14 @@ for production, so do not turn it off to avoid an entry
on this page.
and 4.0, and to Spark 4.1 and later with
`spark.sql.legacy.parquet.returnNullStructIfAllFieldsMissing=true`
([#7030](https://github.com/apache/datafusion-comet/issues/7030)).
-- **Nested BINARY values that are not valid UTF-8, read as STRING.** This
applies to a BINARY
- column without the UTF8 annotation that is read as `string`, through the
read schema or
+- **BINARY values that are not valid UTF-8, read as STRING.** This applies to
a BINARY column
Review Comment:
This bullet and #7106 rewrite the same lines, and `git merge-tree` reports a
conflict in `parquet.md` between the two heads. Each version describes the
other PR's part as still open. This one keeps "Comet returns NULL for each such
nested value", and #7106 keeps "Top-level values reach an Arrow string array
unchecked" without the `binaryAsString` case. Whichever lands second could
merge them: nested values and top-level values read with `binaryAsString=true`
become `U+FFFD`, and top-level values read through a read schema still reach an
Arrow string array unchecked (#7032). The list of decode sites at the top of
"Strings with non-UTF-8 bytes" in `compatibility/index.md` could name this scan
too, as #7106 does for nested values.
##########
docs/source/contributor-guide/spark_configs_support.md:
##########
@@ -50,6 +50,11 @@ The status column uses these values:
- Affected expressions: `date_format`, `from_unixtime`, `unix_timestamp`,
`to_unix_timestamp`, `to_timestamp`, `to_timestamp_ntz`, `to_date`,
`try_to_date` (Spark 4.1+), `try_to_timestamp` (Spark 3.4+)
- Spark versions checked: 3.4.3, 3.5.8, 4.0.1, 4.1.2
- Date: 2026-07-18
+- `spark.sql.parquet.binaryAsString`
+ - Default: `false`
+ - Status: Partial (see notes)
+ - Affected operator: native Parquet scan
Review Comment:
The other two audited entries in this file record a `Spark versions checked`
line. Could this one list the versions whose schema converter and
`ParquetFilters` were compared, so the audit entries stay uniform?
##########
native/core/src/parquet/parquet_exec.rs:
##########
@@ -63,6 +68,187 @@ pub(crate) struct CometParquetScanOptions {
pub prefetch_next_file_metadata: bool,
}
+/// 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 {
+ Arc::new(Schema::new_with_metadata(
+ schema
+ .fields()
+ .iter()
+ .map(binary_as_string_read_field)
+ .collect::<Vec<_>>(),
+ schema.metadata().clone(),
+ ))
+}
+
+/// The field `binary_as_string_read_schema` reads a top-level `field` as.
Spark's StringType is
+/// Arrow Utf8.
+fn binary_as_string_read_field(field: &FieldRef) -> FieldRef {
+ if is_binary_as_string_read(field.data_type()) {
+ Arc::new(field.as_ref().clone().with_data_type(DataType::Binary))
+ } else {
+ Arc::clone(field)
+ }
+}
+
+/// Whether `binary_as_string_read_schema` reads a top-level field of
`data_type` as Binary.
+fn is_binary_as_string_read(data_type: &DataType) -> bool {
+ data_type == &DataType::Utf8
+}
+
+/// 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)
+}
+
+/// Rewrites a data filter for a scan that reads string columns as Binary (see
+/// `binary_as_string_read_schema`), or returns `None` when it can't go into
the scan.
+///
+/// The scan decodes each such column with replacement: a valid value decodes
to itself, and an
+/// invalid one to a string containing U+FFFD. So for a literal without
U+FFFD, `s = 'lit'` on the
+/// decoded string holds exactly when `s = X'<bytes of lit>'` holds on the raw
bytes, and the same
+/// goes for `IN`. `IS [NOT] NULL` holds on both, because decoding keeps
nulls. Other filters on
+/// these columns stay above the scan. A range can disagree because decoding
can reorder values:
+/// `"a\x80"` sorts below `"aé"` as bytes, but `"a\u{FFFD}"` sorts above it.
+///
+/// Spark's Filter above the scan re-evaluates every data filter, so a pushed
filter only has to
+/// keep every row the original keeps. That's why an `AND` keeps whichever
side can be pushed.
+fn binary_as_string_pushdown_filter(
+ filter: &Arc<dyn PhysicalExpr>,
+ binary_read_columns: &HashSet<String>,
+ case_sensitive: bool,
+ filter_read_schema: &Schema,
+) -> Result<Option<Arc<dyn PhysicalExpr>>, ExecutionError> {
+ if !references_any_column(filter, binary_read_columns, case_sensitive)? {
+ return Ok(Some(Arc::clone(filter)));
+ }
+ let reads_binary = |expr: &Arc<dyn PhysicalExpr>| -> Result<bool,
ExecutionError> {
+ match expr.downcast_ref::<Column>() {
+ Some(column) => {
+ Ok(binary_read_columns.contains(&fold_name(column.name(),
case_sensitive)?))
+ }
+ None => Ok(false),
+ }
+ };
+ let pushdown = |expr: &Arc<dyn PhysicalExpr>| {
+ binary_as_string_pushdown_filter(
+ expr,
+ binary_read_columns,
+ case_sensitive,
+ filter_read_schema,
+ )
+ };
+
+ if let Some(binary) = filter.downcast_ref::<BinaryExpr>() {
+ let (left, right) = (binary.left(), binary.right());
+ match binary.op() {
+ Operator::And => {
+ return Ok(match (pushdown(left)?, pushdown(right)?) {
+ (Some(left), Some(right)) => {
+ Some(Arc::new(BinaryExpr::new(left, Operator::And,
right)))
+ }
+ (side, None) | (None, side) => side,
+ });
+ }
+ Operator::Or => {
+ return Ok(match (pushdown(left)?, pushdown(right)?) {
+ (Some(left), Some(right)) => {
+ Some(Arc::new(BinaryExpr::new(left, Operator::Or,
right)))
+ }
+ _ => None,
+ });
+ }
+ Operator::Eq => {
+ if reads_binary(left)? {
+ if let Some(literal) = binary_literal(right) {
+ let eq = BinaryExpr::new(Arc::clone(left),
Operator::Eq, literal);
+ return Ok(Some(Arc::new(eq)));
+ }
+ }
+ if reads_binary(right)? {
+ if let Some(literal) = binary_literal(left) {
+ let eq = BinaryExpr::new(literal, Operator::Eq,
Arc::clone(right));
+ return Ok(Some(Arc::new(eq)));
+ }
+ }
+ }
+ _ => {}
+ }
+ } else if let Some(list) = filter.downcast_ref::<InListExpr>() {
+ if !list.negated() && reads_binary(list.expr())? {
+ if let Some(values) =
list.list().iter().map(binary_literal).collect() {
+ let expr = Arc::clone(list.expr());
+ return Ok(Some(in_list(expr, values, &false,
filter_read_schema)?));
+ }
+ }
+ } else if let Some(is_null) = filter.downcast_ref::<IsNullExpr>() {
+ if reads_binary(is_null.arg())? {
+ return Ok(Some(Arc::clone(filter)));
+ }
+ } else if let Some(is_not_null) = filter.downcast_ref::<IsNotNullExpr>() {
+ if reads_binary(is_not_null.arg())? {
+ return Ok(Some(Arc::clone(filter)));
+ }
+ }
+ Ok(None)
+}
+
+/// `expr` as a Binary literal holding the same bytes, if it is a string
literal without U+FFFD.
+fn binary_literal(expr: &Arc<dyn PhysicalExpr>) -> Option<Arc<dyn
PhysicalExpr>> {
+ match expr.downcast_ref::<Literal>()?.value() {
+ ScalarValue::Utf8(Some(value)) if
!value.contains(char::REPLACEMENT_CHARACTER) => {
+ let bytes = value.as_bytes().to_vec();
+ Some(Arc::new(Literal::new(ScalarValue::Binary(Some(bytes)))))
+ }
+ _ => None,
+ }
+}
+
+fn binary_as_string_projection(
Review Comment:
A design question that follows from #25802 being closed, and it relates to
the double-cast thread. Could the per-file decision move into
`SparkPhysicalExprAdapter`? DataFusion 55.2's opener runs the adapter over the
pushed projection for each file whose physical schema differs from the logical
one. For a file that annotates the column, the projection should reach the
adapter as Comet's `Cast` to `Utf8` over a DataFusion `CastExpr` to `Binary`
over a `Utf8` column. A rule next to `unwrap_cast_to_binary_comparison` could
return the column itself, since arrow-rs has already validated those bytes
(`validate_utf8` is set for `ConvertedType::UTF8`), and unannotated files would
keep the decode. I would expect the annotated full scan in `CometReadBenchmark
--binary-as-string` to drop back toward its `binaryAsString=false` time, and
the same rule might let ranges and `LIKE` be pushed against the decoded value
again. I have not tried it, and I have not checked that the expression shape
reac
hes the adapter as described. arrow-rs maps JSON-annotated BINARY to `Utf8`
without validating it, so such a column would be read unchecked, as it is with
the setting off.
--
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]
