mbutrovich commented on code in PR #6447:
URL: https://github.com/apache/datafusion-comet/pull/6447#discussion_r4190054555
##########
native/spark-expr/src/array_funcs/nested_comparison.rs:
##########
@@ -219,31 +217,83 @@ impl PhysicalExpr for NestedPredicate {
}
}
-/// Build equality after the planner has reconciled nested operand nullability.
+/// How [`spark_comparison`] treats floating-point operands.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum FloatOperands {
+ /// Normalize them, so that the comparison follows Spark's SQL ordering.
+ Normalize,
+ /// Leave a Float32 or Float64 column compared with a literal as it is,
and normalize every
+ /// other operand. Only a scan's pushed-down data filters use this, and
only when the Parquet
+ /// reader prunes with them but does not filter rows: Parquet pruning
recognizes a column
+ /// compared with a literal but not a normalized column, and Spark's
Filter above the scan
+ /// applies Spark's semantics to every row. With row-level pushdown the
reader would drop the
+ /// rows such a comparison rejects, including a stored NaN whose bits
differ from the
+ /// normalized literal, so the data filters use
[`FloatOperands::Normalize`] there instead.
+ Raw,
+}
+
+/// Builds a comparison with Spark's SQL ordering for floats, in which `-0.0`
equals `0.0`, all
+/// NaNs are equal and NaN sorts above every other value, at any depth of a
list or struct.
+///
+/// Arrow compares floats by IEEE 754 total order instead, so each float
operand is normalized
+/// first with [`normalize_comparison_operand`], after which the two orders
agree. Nested `=` and
+/// `<>` compare with `spark_equality` instead, without building normalized
copies of the nested
+/// values. Any other operator, such as `AND`, builds a plain [`BinaryExpr`].
+///
+/// The planner reconciles the nullability of nested operands before calling
this.
pub fn spark_comparison(
left: Arc<dyn PhysicalExpr>,
op: Operator,
right: Arc<dyn PhysicalExpr>,
schema: &Schema,
+ float_operands: FloatOperands,
) -> Result<Arc<dyn PhysicalExpr>> {
+ use Operator::*;
+ if !matches!(
+ op,
+ Eq | NotEq | Lt | LtEq | Gt | GtEq | IsDistinctFrom | IsNotDistinctFrom
+ ) {
+ return Ok(Arc::new(BinaryExpr::new(left, op, right)));
+ }
// An operand whose type does not resolve against this schema falls back
to the plain
// comparison, the way `reconcile_nested_comparison_types` already leaves
such operands alone.
- let nested = matches!(op, Operator::Eq | Operator::NotEq)
- && match (left.data_type(schema), right.data_type(schema)) {
- (Ok(lt), Ok(_)) => needs_spark_equality(<),
- _ => false,
- };
- if nested {
+ let (Ok(left_type), Ok(_)) = (left.data_type(schema),
right.data_type(schema)) else {
+ return Ok(Arc::new(BinaryExpr::new(left, op, right)));
+ };
+ if matches!(op, Eq | NotEq) && is_nested_with_float_leaf(&left_type) {
validate_types(&left, std::slice::from_ref(&right), schema)?;
- Ok(Arc::new(NestedPredicate {
+ return Ok(Arc::new(NestedPredicate {
value: left,
candidates: vec![right],
- negated: op == Operator::NotEq,
+ negated: op == NotEq,
membership: false,
- }))
- } else {
- Ok(Arc::new(BinaryExpr::new(left, op, right)))
+ }));
}
+ let raw = float_operands == FloatOperands::Raw;
+ let (left, right) = if raw && is_float_column(&left, schema) &&
is_literal(&right) {
+ (left, normalize_comparison_operand(right, schema)?)
+ } else if raw && is_literal(&left) && is_float_column(&right, schema) {
+ (normalize_comparison_operand(left, schema)?, right)
Review Comment:
With `FloatOperands::Raw`, the literal is still normalized. Since 2f003c2,
`Raw` is used only when the reader prunes and doesn't filter rows. Parquet
bloom filter pruning probes the literal's exact bits
([`check_scalar`](https://github.com/apache/datafusion/blob/7d3835c71f30cbd3c3ae4041732267f1f453097a/datafusion/datasource-parquet/src/bloom_filter.rs#L102-L103)
hashes the `f64` as is). So `d = -0.0D` now probes for `0.0`.
What happens to a row group whose `d` values are all `-0.0`? I wrote a file
with arrow-rs, with a bloom filter on `d` and two `-0.0` rows, and scanned it
with the `scan_parquet_file` test helper in `schema_adapter.rs`. With main's
data filter (`d@0 = -0`) the scan returns both rows. With the filter
`spark_comparison(..., FloatOperands::Raw)` builds at 2f003c2 (`d@0 = 0`) it
returns none, with or without column statistics. The Filter above the scan
can't bring those rows back, and Spark matches them, since `-0.0 = -0.0`.
Should `Raw` leave a zero literal as it is, as main does, or expand it into
both signed zeros the way the `IN` path does for constant lists? Leaving it as
it is goes back to main's behavior, where a row group holding only `0.0` is
still pruned for `d = -0.0D`. Expanding it covers both cases. Could you also
add a test with a bloom filter on a `DOUBLE` column, for example in
`CometNativeReaderSuite` with `parquet.bloom.filter.enabled#d`, that checks
`WHERE d = -0.0D` returns the `-0.0` rows?
##########
docs/source/user-guide/latest/compatibility/floating-point.md:
##########
@@ -23,20 +23,34 @@ Spark normalizes NaN and zero for floating point numbers
for several cases. See
However, one exception is comparison. Spark does not normalize NaN and zero
when comparing values
because they are handled well in Spark (e.g.,
`SQLOrderingUtil.compareFloats`). But the comparison
functions of arrow-rs used by DataFusion do not normalize NaN and zero (e.g.,
[arrow::compute::kernels::cmp::eq](https://docs.rs/arrow/latest/arrow/compute/kernels/cmp/fn.eq.html#)).
-For top-level `FLOAT` and `DOUBLE` comparisons, Comet normalizes both operands
before native
-execution, including noncanonical NaN literals. Top-level `IN`, `InSet`, and
`NOT IN` membership
-also normalize dynamic candidates and lists containing NaN. When every
candidate is a non-NaN
-literal, Comet keeps DataFusion's static filter and pruning path, enumerating
both signed-zero
-forms when a list contains zero.
+For `FLOAT` and `DOUBLE` comparisons (`=`, `<>`, `<=>`, `<`, `<=`, `>` and
`>=`), Comet
+normalizes both operands before native execution, including noncanonical NaN
literals. This
+applies wherever a comparison appears: projections, filters, aggregate
arguments and `FILTER`
+clauses, join conditions, sort keys, and generator arguments.
+
+A native Parquet scan skips row groups with its data filters. So that
statistics pruning still
+applies, a `FLOAT` or `DOUBLE` column compared with a constant in a data
filter is compared
+without normalizing the column, and the filter above the scan evaluates the
comparison again with
+Spark's semantics. Every other comparison in a data filter is normalized. With
+`spark.comet.parquet.rowFilterPushdown.enabled=true` the scan also filters
rows with these
+comparisons, so a noncanonical NaN stored in the file, such as one with the
sign bit set, can be
+filtered differently from Spark when compared with a constant. Spark's Parquet
writer only writes
+canonical NaNs.
Review Comment:
This paragraph describes the data filters as they were before 2f003c2. As I
read `data_filter_float_operands`, with
`spark.comet.parquet.rowFilterPushdown.enabled=true` both operands are
normalized. A noncanonical NaN is no longer filtered differently in that mode,
and float comparisons stop pruning instead. What do you think about this
wording?
```suggestion
A native Parquet scan prunes row groups and pages with its data filters. So
that this pruning
still applies, a `FLOAT` or `DOUBLE` column compared with a constant in a
data filter is compared
without normalizing the column, and the filter above the scan evaluates the
comparison again with
Spark's semantics. Every other comparison in a data filter is normalized.
With
`spark.comet.parquet.rowFilterPushdown.enabled=true` the scan also drops the
rows its data filters
reject, so every comparison in a data filter is normalized, and a `FLOAT` or
`DOUBLE` comparison
in a data filter does not prune row groups or pages.
```
--
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]
