sunchao commented on code in PR #6447:
URL: https://github.com/apache/datafusion-comet/pull/6447#discussion_r4185987363
##########
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:
[P2] [P2] Preserve divisor normalization until percentile ordering handles
noncanonical NaNs. For a Spark-written Parquet table `t` containing one NaN in
`d`, run `SELECT percentile_approx(q, 1.0D) FROM (SELECT /*+ COALESCE(1) */ q
FROM (SELECT 1.0D / (-d) AS q FROM t UNION ALL SELECT 0.0D AS q FROM t))`.
Spark returns NaN. On x86-64, removing `normalizePlan` exposes a negative NaN
quotient to `QuantileSummaries`, whose `total_cmp` places it below zero, so the
native accumulator returns `0.0` in both ANSI modes. The former divisor wrapper
made this projection correct. Retain that wrapper or correct percentile NaN
ordering before removing it.
Evidence: Fresh exact-head native reproduction exercised
`create_negate_expr`, the non-ANSI `IfExpr` division guard or ANSI
`checked_div`, and the shipped `ApproxPercentile` accumulator. With divisor
normalization, both modes returned NaN. Without it, both returned `Float64(0)`,
failing the Spark-result assertion. Fresh Spark 3.5.9 execution of the SQL
returned NaN in both modes. Source:
`/tmp/review6447-f2c2-session2-consumers.rs`. Logs:
`/tmp/review6447-f2c2-session2-consumers.log` and
`/tmp/review6447-f2c2-session2-spark.log`.
##########
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:
[P2] [P2] Do not use one-sided normalization to reject Parquet rows. With
`spark.comet.parquet.rowFilterPushdown.enabled=true`, a Parquet column
containing negative NaN bits `0xfff8000000000000`, and normal constant folding,
`SELECT d FROM t WHERE d = -double('NaN')` previously retained the row. This
branch canonicalizes the negative NaN literal while leaving the stored column
raw, so Arrow equality now rejects it. Spark considers the NaNs equal. The
filter above the scan cannot recover the dropped row. This materially worsens
the existing noncanonical-NaN limitation. Keep raw pruning predicates separate
from normalized row predicates, or prevent these raw float comparisons from
rejecting rows.
Evidence: A fresh planner-level test wrote the negative NaN with ArrowWriter
and disabled statistics to isolate row filtering. Through the actual Parquet
reader, the base-equivalent predicate retained 1 row, `create_data_filter` at
this head retained 0, and the fully normalized predicate retained 1. The
regression assertion failed. Fresh Spark 3.5.9 retained the row with Parquet
pushdown both off and on, and inspection confirmed the optimized literal bits
were `fff8000000000000`. Source: `/tmp/review6447-f2c2-session2-scan-probe.rs`.
Logs: `/tmp/review6447-f2c2-session2-scan.log` and
`/tmp/review6447-f2c2-session2-spark.log`.
--
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]
