andygrove commented on code in PR #6447:
URL: https://github.com/apache/datafusion-comet/pull/6447#discussion_r4187945415
##########
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:
Parquet pruning recognizes
+ /// a column compared with a literal but not a normalized column. With
row-level pushdown the
+ /// reader also evaluates the filters on each row, and a row it drops
never reaches Spark's
+ /// Filter above the scan, so any other shape, which pruning cannot use
anyway, is normalized.
+ /// A computed operand such as `-d` can hold a NaN with the sign bit set,
which a raw
+ /// comparison sorts below every other value.
+ 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)?)
Review Comment:
Agreed, and it reaches past that literal. A stored NaN with the sign bit set
fails every ordering comparison against the raw column, so with row-level
pushdown on, `d > 0.0` dropped it on main as well. Data filters now leave the
column raw only when row-level pushdown is off, where the reader prunes with
them and the Filter above the scan applies Spark's semantics to each row. With
`spark.comet.parquet.rowFilterPushdown.enabled=true` they normalize both sides,
so in that mode float comparisons give up statistics pruning instead of
dropping rows (2f003c207).
`scan_data_filters_normalize_float_columns_with_row_level_pushdown` evaluates
`d = -double('NaN')` and `d > 0.0` on a NaN with the sign bit set, a canonical
NaN and `1.0`.
##########
spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala:
##########
@@ -957,14 +904,12 @@ case class CometExecRule(session: SparkSession,
queryStagePrep: Boolean = false)
plan
}
} else {
- val normalizedPlan = normalizePlan(plan)
-
val planWithJoinRewritten = if (CometConf.COMET_FORCE_SHJ.get()) {
- normalizedPlan.transformUp { case p =>
+ plan.transformUp { case p =>
RewriteJoin.rewrite(p)
}
} else {
- normalizedPlan
+ plan
Review Comment:
Reproduced, on aarch64 too, since negating the NaN sets the sign bit and the
division keeps it. Rather than wait for #6519, I kept the divisor
normalization, for `Divide` and `Remainder` only and in Projects and Filters as
before (2f003c207). The comment on `normalizeDivisors` says it stays for
`percentile_approx` until #6519 is fixed. `nan_divisor.sql` runs your query
with ANSI off and on. Without the wrapper Comet returns `[0.0, -0.667]` for the
maximum and minimum where Spark returns `[NaN, -0.667]`, and with it they
match. The wrapper puts `NormalizeNaNAndZero` back into the TPC-DS plans, so
their expression counts are main's again.
--
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]
