voonhous commented on code in PR #19836:
URL: https://github.com/apache/hudi/pull/19836#discussion_r3927318777
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureFilterUtils.scala:
##########
@@ -496,11 +496,15 @@ object HoodieProcedureFilterUtils {
right: org.apache.spark.sql.catalyst.expressions.Expression,
constructor:
(org.apache.spark.sql.catalyst.expressions.Expression,
org.apache.spark.sql.catalyst.expressions.Expression) => T,
original: T): T = {
- (left, right) match {
- case (boundRef:
org.apache.spark.sql.catalyst.expressions.BoundReference, literal:
org.apache.spark.sql.catalyst.expressions.Literal)
- if boundRef.dataType == org.apache.spark.sql.types.LongType &&
literal.dataType == org.apache.spark.sql.types.IntegerType =>
- val castExpr =
org.apache.spark.sql.catalyst.expressions.Cast(boundRef,
org.apache.spark.sql.types.IntegerType)
- constructor(castExpr, literal)
+ (left.dataType, right.dataType) match {
Review Comment:
**major:** `left.dataType` now runs on every comparison node during the
transform, before eval. `UnresolvedAttribute` and `UnresolvedFunction` throw
`UnresolvedException` there, so the per-row `Try` drops the row even where `Or`
previously short-circuited past the broken branch.
On 3.5.5 and 4.1.1, ANSI on and off: `keep(scalarRows, "id = 1 OR
concat(name, 'x') = 'a1x'")` returns `Seq(scalarRows.head)` on master and
`Seq.empty` here. `validateFilter` only checks column references, so such
filters do reach `evaluateFilter`.
Could we skip coercion when either side is unresolved?
```suggestion
if (!left.resolved || !right.resolved) original
else (left.dataType, right.dataType) match {
```
I replayed all 86 filter strings this test file asserts on through that
guard on both ANSI settings: no change except the OR cases above.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureFilterUtils.scala:
##########
@@ -496,11 +496,15 @@ object HoodieProcedureFilterUtils {
right: org.apache.spark.sql.catalyst.expressions.Expression,
constructor:
(org.apache.spark.sql.catalyst.expressions.Expression,
org.apache.spark.sql.catalyst.expressions.Expression) => T,
original: T): T = {
- (left, right) match {
- case (boundRef:
org.apache.spark.sql.catalyst.expressions.BoundReference, literal:
org.apache.spark.sql.catalyst.expressions.Literal)
- if boundRef.dataType == org.apache.spark.sql.types.LongType &&
literal.dataType == org.apache.spark.sql.types.IntegerType =>
- val castExpr =
org.apache.spark.sql.catalyst.expressions.Cast(boundRef,
org.apache.spark.sql.types.IntegerType)
- constructor(castExpr, literal)
+ (left.dataType, right.dataType) match {
+ case (_: NumericType, _: NumericType) =>
+ TypeCoercion.findWiderTypeForTwo(left.dataType, right.dataType)
Review Comment:
**minor:** Not blocking. Spark 4 defaults to ANSI and its analyzer uses
`AnsiTypeCoercion`, which differs from `TypeCoercion` here:
`findWiderTypeForTwo(BIGINT, FLOAT)` gives `FLOAT` vs `DOUBLE` (same for
INT/SMALLINT/TINYINT). With `ts = 16777217`, `ts > 16777216.0f` is false via
float and true via double. Still an improvement on master, which dropped every
row on that pair.
Could we pick the object from the session, `if (SQLConf.get.ansiEnabled)
AnsiTypeCoercion else TypeCoercion`? `AnsiTypeCoercion.findWiderTypeForTwo` is
present in catalyst 3.3.4 through 4.1.1.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHoodieProcedureFilterUtils.scala:
##########
@@ -74,32 +74,23 @@ class TestHoodieProcedureFilterUtils extends
HoodieSparkProcedureTestBase {
assertResult(Seq(scalarRows.head))(keep(scalarRows, "flag = true",
scalarSchema))
}
- test("evaluateFilter coerces Long columns against integer literals") {
+ test("evaluateFilter widens Long columns and integer literals") {
// Exercises applyTypeCoercion for every comparison operator (Long
boundRef vs Int literal).
assertResult(Seq(scalarRows.head))(keep(scalarRows, "ts = 1000",
scalarSchema))
assertResult(Seq(scalarRows(1)))(keep(scalarRows, "ts > 1500",
scalarSchema))
assertResult(Seq(scalarRows(1)))(keep(scalarRows, "ts >= 2000",
scalarSchema))
assertResult(Seq(scalarRows.head))(keep(scalarRows, "ts < 2000",
scalarSchema))
assertResult(Seq(scalarRows.head))(keep(scalarRows, "ts <= 1000",
scalarSchema))
- // Known limitation: the coercion narrows the Long column to Int instead
of widening the Int
- // literal, so a Long value beyond Int range never matches (wrong results
under non-ANSI Spark,
- // swallowed overflow error under ANSI). Pinned here so a fix flips this
assertion; see #19632.
+ // The integer literal is widened, preserving Long values beyond the Int
range.
val bigRow = Seq(Row(3, "c3", 30.0d, 3000000000L, true, -9,
Date.valueOf("2024-03-16"), Timestamp.valueOf("2024-03-16 12:30:00")))
- assertResult(Seq.empty)(keep(bigRow, "ts > 2000", scalarSchema))
- // Known limitation: the coercion only matches column-on-left, so a
literal-on-left comparison
- // never coerces and drops every row instead of mirroring the equivalent
column-on-left filter.
- // Pinned here so a fix flips these assertions; see #19632.
- assertResult(Seq.empty)(keep(scalarRows, "1500 < ts", scalarSchema))
- assertResult(Seq.empty)(keep(scalarRows, "1000 = ts", scalarSchema))
+ assertResult(bigRow)(keep(bigRow, "ts > 2000", scalarSchema))
Review Comment:
**minor:** Not blocking. This pins the false-negative half of #19632 but not
the false-positive half: master narrowed 3000000000 to -1294967296, so `ts <
2000` matched the big row. Nothing here would catch a fix that still wraps. The
one-row fixture also means `ts > 2000` selects 1 of 1.
Could we add `assertResult(Seq.empty)(keep(bigRow, "ts < 2000",
scalarSchema))` and run both over `scalarRows ++ bigRow`?
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureFilterUtils.scala:
##########
@@ -496,11 +496,15 @@ object HoodieProcedureFilterUtils {
right: org.apache.spark.sql.catalyst.expressions.Expression,
constructor:
(org.apache.spark.sql.catalyst.expressions.Expression,
org.apache.spark.sql.catalyst.expressions.Expression) => T,
original: T): T = {
- (left, right) match {
- case (boundRef:
org.apache.spark.sql.catalyst.expressions.BoundReference, literal:
org.apache.spark.sql.catalyst.expressions.Literal)
- if boundRef.dataType == org.apache.spark.sql.types.LongType &&
literal.dataType == org.apache.spark.sql.types.IntegerType =>
- val castExpr =
org.apache.spark.sql.catalyst.expressions.Cast(boundRef,
org.apache.spark.sql.types.IntegerType)
- constructor(castExpr, literal)
+ (left.dataType, right.dataType) match {
+ case (_: NumericType, _: NumericType) =>
+ TypeCoercion.findWiderTypeForTwo(left.dataType, right.dataType)
+ .map { widerType =>
+ val coercedLeft = if (left.dataType == widerType) left else
Cast(left, widerType)
Review Comment:
**minor:** Not blocking. `DecimalType.bounded` clamps precision at 38, so
the wider type can be too narrow for one side:
`findWiderTypeForTwo(DECIMAL(38,18), DECIMAL(38,0))` is `DECIMAL(38,18)`. The
cast then yields null (non-ANSI) or raises (ANSI) and the row is dropped, where
master compared the two `Decimal` values exactly. Latent today, since no
procedure declares a DecimalType column.
Could we add `case (_: DecimalType, _: DecimalType) => original` above the
numeric case? Decimal ordering is already precision and scale independent.
`Cast.canUpCast` is too strict as a guard: `canUpCast(DECIMAL(3,1), DOUBLE)` is
false, which would undo the `price > 15.0` fix.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHoodieProcedureFilterUtils.scala:
##########
@@ -111,13 +102,13 @@ class TestHoodieProcedureFilterUtils extends
HoodieSparkProcedureTestBase {
assertResult(rows)(keep(rows, "f > 1.0f", schema))
assertResult(rows)(keep(rows, "dec > 1.00", schema))
- // Mismatched literal types no-match even though the values would satisfy
the predicate.
- assertResult(Seq.empty)(keep(rows, "sh = 3", schema))
- assertResult(Seq.empty)(keep(rows, "by = 4", schema))
- assertResult(Seq.empty)(keep(rows, "f > 1.0d", schema))
- assertResult(Seq.empty)(keep(rows, "dec > 1", schema))
- // Same gap for a double column: a plain 15.0 parses as decimal, not
double.
- assertResult(Seq.empty)(keep(scalarRows, "price > 15.0", scalarSchema))
+ // Mismatched numeric types are widened to Spark's common type.
+ assertResult(rows)(keep(rows, "sh = 3", schema))
+ assertResult(rows)(keep(rows, "by = 4", schema))
+ assertResult(rows)(keep(rows, "f > 1.0d", schema))
+ assertResult(rows)(keep(rows, "dec > 1", schema))
+ // A plain 15.0 decimal literal is coerced with the double column.
+ assertResult(Seq(scalarRows(1)))(keep(scalarRows, "price > 15.0",
scalarSchema))
Review Comment:
**minor:** Not blocking. These unit assertions are the only coverage for the
fix. All 8 procedure-level `filter =>` tests use String or Int columns, so none
of them changes behaviour with this commit, and the two reproducers named in
#19632 (`show_fsview_all` data_file_size, `show_column_stats_overlap` Average
overlap) have no filter test at all.
Could we add one `show_fsview_all(filter => "data_file_size > <n>")`
assertion so the reported path is covered end to end?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHoodieProcedureFilterUtils.scala:
##########
@@ -74,32 +74,23 @@ class TestHoodieProcedureFilterUtils extends
HoodieSparkProcedureTestBase {
assertResult(Seq(scalarRows.head))(keep(scalarRows, "flag = true",
scalarSchema))
}
- test("evaluateFilter coerces Long columns against integer literals") {
+ test("evaluateFilter widens Long columns and integer literals") {
// Exercises applyTypeCoercion for every comparison operator (Long
boundRef vs Int literal).
assertResult(Seq(scalarRows.head))(keep(scalarRows, "ts = 1000",
scalarSchema))
assertResult(Seq(scalarRows(1)))(keep(scalarRows, "ts > 1500",
scalarSchema))
assertResult(Seq(scalarRows(1)))(keep(scalarRows, "ts >= 2000",
scalarSchema))
assertResult(Seq(scalarRows.head))(keep(scalarRows, "ts < 2000",
scalarSchema))
assertResult(Seq(scalarRows.head))(keep(scalarRows, "ts <= 1000",
scalarSchema))
- // Known limitation: the coercion narrows the Long column to Int instead
of widening the Int
- // literal, so a Long value beyond Int range never matches (wrong results
under non-ANSI Spark,
- // swallowed overflow error under ANSI). Pinned here so a fix flips this
assertion; see #19632.
+ // The integer literal is widened, preserving Long values beyond the Int
range.
val bigRow = Seq(Row(3, "c3", 30.0d, 3000000000L, true, -9,
Date.valueOf("2024-03-16"), Timestamp.valueOf("2024-03-16 12:30:00")))
- assertResult(Seq.empty)(keep(bigRow, "ts > 2000", scalarSchema))
- // Known limitation: the coercion only matches column-on-left, so a
literal-on-left comparison
- // never coerces and drops every row instead of mirroring the equivalent
column-on-left filter.
- // Pinned here so a fix flips these assertions; see #19632.
- assertResult(Seq.empty)(keep(scalarRows, "1500 < ts", scalarSchema))
- assertResult(Seq.empty)(keep(scalarRows, "1000 = ts", scalarSchema))
+ assertResult(bigRow)(keep(bigRow, "ts > 2000", scalarSchema))
+ // Coercion applies symmetrically when the literal is on the left.
+ assertResult(Seq(scalarRows(1)))(keep(scalarRows, "1500 < ts",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "1000 = ts",
scalarSchema))
}
- test("evaluateFilter does not coerce other numeric column/literal type
pairs") {
- // Known limitation: applyTypeCoercion only special-cases a Long column
against an Int literal.
- // Every other numeric column/literal pair is left alone, so the
mismatched comparison fails to
- // evaluate; the per-row Try swallows the failure and drops the row. The
filter therefore
- // returns no rows instead of erroring on the type mismatch.
- // Pinned here so a fix flips the Seq.empty assertions; see #19632.
+ test("evaluateFilter coerces numeric column and literal type pairs") {
Review Comment:
**major:** The rename says "numeric column and literal type pairs", but the
coercion is wired only to
EqualTo/GreaterThan/GreaterThanOrEqual/LessThan/LessThanOrEqual, so `In` and
`EqualNullSafe` still hit the #19632 silent drop: `ts IN (1000, 2000)` and `ts
<=> 1000` throw `ClassCastException` per row and return nothing, before and
after. The only IN assertion (`id IN (1, 3)`, line 136) is Int vs Int, so it
cannot catch that, and 79 of the numeric procedure output columns are LongType.
Could we extend the transform to `In` and `EqualNullSafe`, or pin
`assertResult(Seq.empty)(keep(scalarRows, "ts IN (1000, 2000)", scalarSchema))`
with an issue reference so the name does not overclaim?
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureFilterUtils.scala:
##########
@@ -496,11 +496,15 @@ object HoodieProcedureFilterUtils {
right: org.apache.spark.sql.catalyst.expressions.Expression,
constructor:
(org.apache.spark.sql.catalyst.expressions.Expression,
org.apache.spark.sql.catalyst.expressions.Expression) => T,
original: T): T = {
- (left, right) match {
- case (boundRef:
org.apache.spark.sql.catalyst.expressions.BoundReference, literal:
org.apache.spark.sql.catalyst.expressions.Literal)
- if boundRef.dataType == org.apache.spark.sql.types.LongType &&
literal.dataType == org.apache.spark.sql.types.IntegerType =>
- val castExpr =
org.apache.spark.sql.catalyst.expressions.Cast(boundRef,
org.apache.spark.sql.types.IntegerType)
- constructor(castExpr, literal)
+ (left.dataType, right.dataType) match {
+ case (_: NumericType, _: NumericType) =>
+ TypeCoercion.findWiderTypeForTwo(left.dataType, right.dataType)
+ .map { widerType =>
+ val coercedLeft = if (left.dataType == widerType) left else
Cast(left, widerType)
+ val coercedRight = if (right.dataType == widerType) right else
Cast(right, widerType)
+ constructor(coercedLeft, coercedRight)
Review Comment:
**nit:** Feel free to ignore. When both sides already equal `widerType` the
node is still rebuilt, and all three transform passes run per row
(`evaluateFilter` calls `evaluateExpressionOnRow` inside `rows.filter`), so
this adds a `findWiderTypeForTwo` lookup plus allocations per comparison per
row. `show_fsview_all` returns one row per file slice.
Could we return `original` when neither side needs a cast, and hoist the
three transforms out of the row loop as a follow-up?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHoodieProcedureFilterUtils.scala:
##########
@@ -111,13 +102,13 @@ class TestHoodieProcedureFilterUtils extends
HoodieSparkProcedureTestBase {
assertResult(rows)(keep(rows, "f > 1.0f", schema))
assertResult(rows)(keep(rows, "dec > 1.00", schema))
- // Mismatched literal types no-match even though the values would satisfy
the predicate.
- assertResult(Seq.empty)(keep(rows, "sh = 3", schema))
- assertResult(Seq.empty)(keep(rows, "by = 4", schema))
- assertResult(Seq.empty)(keep(rows, "f > 1.0d", schema))
- assertResult(Seq.empty)(keep(rows, "dec > 1", schema))
- // Same gap for a double column: a plain 15.0 parses as decimal, not
double.
- assertResult(Seq.empty)(keep(scalarRows, "price > 15.0", scalarSchema))
+ // Mismatched numeric types are widened to Spark's common type.
+ assertResult(rows)(keep(rows, "sh = 3", schema))
+ assertResult(rows)(keep(rows, "by = 4", schema))
+ assertResult(rows)(keep(rows, "f > 1.0d", schema))
+ assertResult(rows)(keep(rows, "dec > 1", schema))
+ // A plain 15.0 decimal literal is coerced with the double column.
Review Comment:
**nit:** Feel free to ignore. Two comments now describe behaviour this
commit removed: lines 68-69 ("The literal must match the column type ... the
plain 15.0 (decimal) form is pinned in the numeric-coercion test below") and
lines 157-159 ("The util only special-cases a long column vs an integer
literal"). Line 71 (`price > 15.0d`) also duplicates line 111 now that both
literal forms work.
Could we reword both comments and drop one of the two `price > 15`
assertions?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHoodieProcedureFilterUtils.scala:
##########
@@ -111,13 +102,13 @@ class TestHoodieProcedureFilterUtils extends
HoodieSparkProcedureTestBase {
assertResult(rows)(keep(rows, "f > 1.0f", schema))
assertResult(rows)(keep(rows, "dec > 1.00", schema))
- // Mismatched literal types no-match even though the values would satisfy
the predicate.
- assertResult(Seq.empty)(keep(rows, "sh = 3", schema))
- assertResult(Seq.empty)(keep(rows, "by = 4", schema))
- assertResult(Seq.empty)(keep(rows, "f > 1.0d", schema))
- assertResult(Seq.empty)(keep(rows, "dec > 1", schema))
- // Same gap for a double column: a plain 15.0 parses as decimal, not
double.
- assertResult(Seq.empty)(keep(scalarRows, "price > 15.0", scalarSchema))
+ // Mismatched numeric types are widened to Spark's common type.
+ assertResult(rows)(keep(rows, "sh = 3", schema))
Review Comment:
**minor:** Not blocking. `rows` is a single row and all four assertions are
positive-only, so they cannot distinguish "coerced correctly" from "matched
everything". Reversed operands are covered only for the Long/Int pair (lines
89-90), not for the pairs this commit newly handles.
Could we add a second, non-matching row to `rows`, plus `3 = sh` and `by <
300`? Both discriminate against master, which returns empty for them.
--
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]
