This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new cebbd6a0d564 fix(spark): widen procedure filter numeric comparisons
(#19836)
cebbd6a0d564 is described below
commit cebbd6a0d56423383ffe5c8bd947db14c646e3d5
Author: Nikulin Nikita <[email protected]>
AuthorDate: Mon Sep 7 19:54:36 2026 +0300
fix(spark): widen procedure filter numeric comparisons (#19836)
Procedure filters cast a Long column down to Int when compared with
an Int literal and left every other mixed numeric pair unresolved, so
a value past the Int range matched the wrong rows, a literal on the
left failed validation, and mixed-type operands were rejected.
Widen operands to their common type instead, using TypeCoercion or
AnsiTypeCoercion as SQLConf.get.ansiEnabled selects, and apply
Spark's DecimalPrecision rules before the generic widening so decimal
literals keep their precision and decimal arithmetic keeps its result
scale. The same widening covers IN, <=>, coalesce and arithmetic;
non-decimal "/" promotes to Double and "div" promotes narrow integrals
to Long, mirroring the analyzer's Division and IntegralDivision rules;
a null operand takes the type of its peers.
Beyond the widening: an ANSI arithmetic or cast error now fails the
procedure instead of silently dropping the row, and a filter whose
result is not boolean is rejected at validation the way Spark raises
FILTER_NOT_BOOLEAN. Binding and resolution run once per batch rather
than per row, and unresolved operands are left alone so validation
keeps its own messages and Or still short-circuits.
Tests compare the evaluator with df.filter across both ANSI modes and
the decimal precision settings, pin the Spark 3 overflow and Spark 4
cast-down outcomes of a DECIMAL(38,0) vs DECIMAL(38,18) comparison,
and add procedure-level filters on show_fsview_all and
show_metadata_column_stats_overlap, the two reproducers in the issue.
Filters still run after limit in the show_* procedures (#19862), and
decimal parity under the legacy retain-fraction and precision-loss
settings is tracked in #19860.
Fixes #19632
---------
Co-authored-by: voon <[email protected]>
---
.../procedures/HoodieProcedureFilterUtils.scala | 259 +++++++++++++++--
.../sql/hudi/procedure/TestFsViewProcedure.scala | 21 ++
.../procedure/TestHoodieProcedureFilterUtils.scala | 320 ++++++++++++++++++---
.../sql/hudi/procedure/TestMetadataProcedure.scala | 15 +
4 files changed, 550 insertions(+), 65 deletions(-)
diff --git
a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureFilterUtils.scala
b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureFilterUtils.scala
index d0783fb872a5..a51bf30243cb 100644
---
a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureFilterUtils.scala
+++
b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureFilterUtils.scala
@@ -18,12 +18,14 @@
package org.apache.spark.sql.hudi.command.procedures
import org.apache.spark.sql.{Row, SparkSession}
-import org.apache.spark.sql.catalyst.analysis.{UnresolvedAttribute,
UnresolvedFunction}
-import org.apache.spark.sql.catalyst.expressions.{Expression,
GenericInternalRow, Unevaluable}
+import org.apache.spark.sql.catalyst.analysis.{AnsiTypeCoercion,
DecimalPrecision, TypeCoercion, UnresolvedAttribute, UnresolvedFunction}
+import org.apache.spark.sql.catalyst.expressions.{BinaryArithmetic,
BinaryComparison, Cast, Coalesce, Divide, EqualNullSafe, Expression,
GenericInternalRow, In, IntegralDivide, Unevaluable}
import org.apache.spark.sql.catalyst.util.DateTimeUtils
-import org.apache.spark.sql.types.{DataType, StructType}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{BooleanType, ByteType, DataType,
DecimalType, DoubleType, IntegerType, LongType, NullType, NumericType,
ShortType, StructType}
import org.apache.spark.unsafe.types.UTF8String
+import java.time.DateTimeException
import java.util.Locale
import scala.collection.JavaConverters._
@@ -58,11 +60,16 @@ object HoodieProcedureFilterUtils {
Try {
val parsedExpr =
sparkSession.sessionState.sqlParser.parseExpression(filterExpression)
- rows.filter { row =>
- evaluateExpressionOnRow(parsedExpr, row, schema)
- }
+ // Binding and resolution depend only on the schema, so run the three
passes once for the
+ // whole batch instead of per row.
+ val boundExpr = bindAndResolveExpression(parsedExpr, schema)
+ rows.filter(row => evaluateExpressionOnRow(boundExpr, row, schema))
} match {
case Success(filteredRows) => filteredRows
+ // Surface an overflowing ANSI cast or arithmetic, or an ANSI cast of
a malformed string,
+ // with Spark's own exception rather than restating it as a
filter-expression problem: the
+ // expression is fine, the data does not fit.
+ case Failure(e @ (_: ArithmeticException | _: NumberFormatException |
_: DateTimeException)) => throw e
case Failure(exception) =>
throw new IllegalArgumentException(
s"Failed to parse or evaluate filter expression
'$filterExpression': ${exception.getMessage}",
@@ -355,24 +362,38 @@ object HoodieProcedureFilterUtils {
// Third pass: handle type coercion for numeric comparisons
functionResolved.transformUp {
case eq: org.apache.spark.sql.catalyst.expressions.EqualTo =>
- applyTypeCoercion(eq.left, eq.right,
org.apache.spark.sql.catalyst.expressions.EqualTo.apply, eq)
+ applyTypeCoercion(eq)
case gt: org.apache.spark.sql.catalyst.expressions.GreaterThan =>
- applyTypeCoercion(gt.left, gt.right,
org.apache.spark.sql.catalyst.expressions.GreaterThan.apply, gt)
+ applyTypeCoercion(gt)
case gte: org.apache.spark.sql.catalyst.expressions.GreaterThanOrEqual =>
- applyTypeCoercion(gte.left, gte.right,
org.apache.spark.sql.catalyst.expressions.GreaterThanOrEqual.apply, gte)
+ applyTypeCoercion(gte)
case lt: org.apache.spark.sql.catalyst.expressions.LessThan =>
- applyTypeCoercion(lt.left, lt.right,
org.apache.spark.sql.catalyst.expressions.LessThan.apply, lt)
+ applyTypeCoercion(lt)
case lte: org.apache.spark.sql.catalyst.expressions.LessThanOrEqual =>
- applyTypeCoercion(lte.left, lte.right,
org.apache.spark.sql.catalyst.expressions.LessThanOrEqual.apply, lte)
+ applyTypeCoercion(lte)
+ case eqns: EqualNullSafe =>
+ applyTypeCoercion(eqns)
+ case in: In =>
+ applyInTypeCoercion(in)
+ // Divide and IntegralDivide are BinaryArithmetic but accept only Double
or Decimal, and only
+ // Long or Decimal, respectively, so each needs its own target type and
has to be matched
+ // before the general arithmetic case below.
+ case divide: Divide =>
+ applyDivideTypeCoercion(divide)
+ case idiv: IntegralDivide =>
+ applyIntegralDivideTypeCoercion(idiv)
+ case arith: BinaryArithmetic =>
+ applyArithmeticTypeCoercion(arith)
+ case coalesce: Coalesce =>
+ applyCoalesceTypeCoercion(coalesce)
}
}
- private def evaluateExpressionOnRow(expression: Expression, row: Row,
schema: StructType): Boolean = {
+ private def evaluateExpressionOnRow(boundExpr: Expression, row: Row, schema:
StructType): Boolean = {
val internalRow = convertRowToInternalRow(row, schema)
Try {
- val boundExpr = bindAndResolveExpression(expression, schema)
val result = boundExpr.eval(internalRow)
result match {
@@ -387,6 +408,11 @@ object HoodieProcedureFilterUtils {
}
} match {
case Success(result) => result
+ // Spark raises SparkArithmeticException for an overflowing ANSI cast or
arithmetic, and
+ // SparkNumberFormatException or SparkDateTimeException for an ANSI cast
of a malformed
+ // string; each extends the matching JDK type. Swallowing one would
silently drop a row the
+ // same query keeps, so let it out and let the caller fail the way the
equivalent query does.
+ case Failure(e @ (_: ArithmeticException | _: NumberFormatException | _:
DateTimeException)) => throw e
case Failure(_) => false
}
}
@@ -490,6 +516,11 @@ object HoodieProcedureFilterUtils {
val names = unsupportedExpressions.toSeq.sorted
val detail = if (names.nonEmpty) s": ${names.mkString(", ")}" else ""
Left(s"Unsupported filter expression$detail")
+ } else if (resolvedExpr.dataType != BooleanType) {
+ // Spark rejects any non-boolean filter condition, string included,
with
+ // DATATYPE_MISMATCH.FILTER_NOT_BOOLEAN. Without this a resolvable
expression such as
+ // "ts + 1" would report zero matching rows instead of the error the
same query raises.
+ Left(s"Filter expression must be boolean, got
${resolvedExpr.dataType.simpleString}")
} else {
Right(())
}
@@ -516,17 +547,197 @@ object HoodieProcedureFilterUtils {
}
}
- private def applyTypeCoercion[T <:
org.apache.spark.sql.catalyst.expressions.Expression](
-
left: org.apache.spark.sql.catalyst.expressions.Expression,
-
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)
- case _ => original
+ private def applyTypeCoercion(original: BinaryComparison): Expression = {
+ if (!original.childrenResolved) {
+ original
+ } else {
+ // Spark can replace an integral/decimal-literal inequality with an
integral comparison,
+ // avoiding a lossy cast of the column. It also gives integral literals
minimum decimal
+ // precision before finding the common comparison type.
+ val promoted = DecimalPrecision.transform.applyOrElse(original,
identity[Expression])
+ // Mixed decimal/integral promotion creates two decimal operands. Apply
the decimal-pair
+ // rule next, just as a subsequent analyzer iteration would.
+ val comparison = DecimalPrecision.transform.applyOrElse(promoted,
identity[Expression])
+ comparison match {
+ case binary: BinaryComparison =>
+ widenOperands(Seq(binary.left, binary.right))
+ .map(binary.withNewChildren).getOrElse(binary)
+ case other => other
+ }
+ }
+ }
+
+ private def applyInTypeCoercion(in: In): Expression = {
+ widenOperands(in.value +: in.list) match {
+ case Some(widened) => In(widened.head, widened.tail)
+ case _ => in
}
}
+
+ /**
+ * Arithmetic keeps its decimal operands exactly as they are. Unlike a
comparison,
+ * BinaryArithmetic.checkInputDataTypes accepts two decimals of different
precision and scale and
+ * derives the result type from them, so widening to a common type changes
the answer rather than
+ * enabling it: DECIMAL(38,18) * DECIMAL(2,1) yields a scale-16 product,
while casting both to
+ * DECIMAL(38,18) first drives the product to scale 6 and rounds 0.0000001
away to zero.
+ *
+ * Spark's DecimalPrecision rule promotes integral operands without changing
the existing
+ * decimal's type, including minimum precision for integral literals. It
also promotes decimals
+ * mixed with floating-point operands to Double. Null operands take the
other operand's type.
+ */
+ private def applyArithmeticTypeCoercion(arith: BinaryArithmetic): Expression
= {
+ val operands = Seq(arith.left, arith.right)
+ if (operands.exists(!_.resolved) || !operands.forall(operand =>
isNumericOrNull(operand.dataType))) {
+ arith
+ } else {
+ val decimalOperands =
operands.exists(_.dataType.isInstanceOf[DecimalType])
+ val promoted = if (decimalOperands) {
+ DecimalPrecision.transform.applyOrElse(arith, identity[Expression])
+ } else {
+ arith
+ }
+ promoted match {
+ case binary: BinaryArithmetic =>
+ val children = Seq(binary.left, binary.right)
+ val widened = if
(children.forall(_.dataType.isInstanceOf[DecimalType])) {
+ binary
+ } else {
+ widenOperands(children)
+ .map(binary.withNewChildren).getOrElse(binary)
+ }
+ // Spark 3.3 wraps decimal arithmetic in CheckOverflow after operand
promotion.
+ // Later versions calculate the result precision within
BinaryArithmetic itself.
+ if (decimalOperands) DecimalPrecision.transform.applyOrElse(widened,
identity[Expression]) else widened
+ case other => other
+ }
+ }
+ }
+
+ /**
+ * Divide only accepts Double or Decimal, so widening its operands to their
common numeric type
+ * leaves an integral pair unresolved and "ts / 2 > 500" rejected while
"price / 2 > 5" works.
+ * Mirror the analyzer's Division rule instead: leave a pair that involves a
decimal to the same
+ * widening as the other arithmetic, and promote everything else to Double.
Like Spark, that
+ * includes a null operand, so "ts / null" resolves and evaluates to null
rather than failing
+ * validation. The rule is unchanged between the two majors this builds
against, only relocated:
+ *
+ * 3.5.5 object Division, in
+ *
https://github.com/apache/spark/blob/v3.5.5/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercion.scala
+ * 4.1.1
+ *
https://github.com/apache/spark/blob/v4.1.1/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DivisionTypeCoercion.scala
+ */
+ private def applyDivideTypeCoercion(divide: Divide): Expression = {
+ val operands = Seq(divide.left, divide.right)
+ if (operands.exists(!_.resolved) || !operands.forall(operand =>
isNumericOrNull(operand.dataType))) {
+ divide
+ } else if (operands.exists(_.dataType.isInstanceOf[DecimalType])) {
+ applyArithmeticTypeCoercion(divide)
+ } else {
+ divide.withNewChildren(operands.map(operand => castTo(operand,
DoubleType)))
+ }
+ }
+
+ /**
+ * IntegralDivide accepts only Long or Decimal, and its operands are never
widened against each
+ * other, so a same-typed Int pair leaves "id div 2" unresolved while "ts
div 2" resolves. Mirror
+ * the analyzer's IntegralDivision rule, which promotes each narrower
integral operand on its own
+ * before the arithmetic widening runs.
+ */
+ private def applyIntegralDivideTypeCoercion(divide: IntegralDivide):
Expression = {
+ if (!divide.childrenResolved) {
+ divide
+ } else {
+ val promoted = divide.withNewChildren(Seq(divide.left, divide.right).map
{ operand =>
+ operand.dataType match {
+ case ByteType | ShortType | IntegerType => castTo(operand, LongType)
+ case _ => operand
+ }
+ })
+ promoted match {
+ case arith: BinaryArithmetic => applyArithmeticTypeCoercion(arith)
+ case other => other
+ }
+ }
+ }
+
+ /** Spark's own guard for the numeric coercion rules, which admit a null
literal. */
+ private def isNumericOrNull(dataType: DataType): Boolean =
+ dataType.isInstanceOf[NumericType] || dataType.isInstanceOf[NullType]
+
+ private def applyCoalesceTypeCoercion(coalesce: Coalesce): Expression = {
+ widenOperands(coalesce.children) match {
+ case Some(widened) => Coalesce(widened)
+ case _ => coalesce
+ }
+ }
+
+ /**
+ * Widens comparison operands of differing numeric types to their common
wider type, so that
+ * e.g. a LongType column compares against an IntegerType literal on the
widened Long rather
+ * than narrowing the column. A NullType operand takes the type of its peers
whatever that type
+ * is, the way Spark plans `ts IN (1000, null)` and `name IN ('a1', null)`.
Returns None when the
+ * operands need no widening or cannot be widened, in which case the caller
keeps the expression
+ * untouched.
+ *
+ * Numeric conversion can still lose precision:
+ * - Large integers may round when converted to Float or Double. For
example, Long 16777217
+ * becomes Float 16777216.
+ * - Large integers may overflow when converted to a decimal with
insufficient space before the
+ * decimal point. For example, DECIMAL(38,20) allows only 18 digits
before the decimal point,
+ * so the 19-digit Long 9000000000000000000 does not fit.
+ *
+ * Rounding can change comparison results, silently, exactly as it does in a
query. Overflow
+ * follows the session's ANSI mode the way Spark's own Cast does: without
ANSI the cast yields
+ * null and the row is filtered out, with ANSI it raises and the failure
reaches the caller.
+ */
+ private def widenOperands(operands: Seq[Expression]):
Option[Seq[Expression]] = {
+ if (operands.exists(!_.resolved)) {
+ // dataType throws on an unresolved operand. Leaving it untouched lets
validateFilterExpression
+ // report its own message ("Invalid column references", "Unsupported
functions") instead of an
+ // UnresolvedException, and keeps Or/And short-circuiting intact at eval
time.
+ None
+ } else {
+ val operandTypes = operands.map(_.dataType)
+ val nonNullTypes =
operandTypes.filterNot(_.isInstanceOf[NullType]).distinct
+ if (operandTypes.distinct.length == 1) {
+ None
+ } else if (nonNullTypes.length == 1) {
+ // Only nulls differ from a single peer type, so the nulls take that
type whether or not it
+ // is numeric. Nothing else needs widening.
+ Some(operands.map(operand => castTo(operand, nonNullTypes.head)))
+ } else if (!operandTypes.forall(isNumericOrNull)) {
+ None
+ } else {
+ findWiderNumericType(operandTypes)
+ .map(widerType => operands.map(operand => castTo(operand,
widerType)))
+ }
+ }
+ }
+
+ /**
+ * Mirrors the analyzer's choice of coercion rules, so that a filter widens
the way the same
+ * comparison would in a SQL query. The two disagree: for BIGINT with FLOAT,
AnsiTypeCoercion
+ * gives DOUBLE while TypeCoercion follows numericPrecedence and gives
FLOAT. Spark 4 defaults
+ * to ANSI mode, Spark 3 does not.
+ *
+ * Reads SQLConf.get rather than the SparkSession because Cast takes its
eval mode from that same
+ * thread-local at construction, and the two-argument Cast(child, dataType)
is the only form that
+ * is portable across Spark 3.3 to 4.x (3.3 takes ansiEnabled, 3.4+ takes
evalMode).
+ *
+ * Decimal pairs go through Spark's own precision rules, which likewise only
moved between the
+ * majors: 3.5.5 analysis/DecimalPrecision.scala, 4.1.1
analysis/DecimalPrecisionTypeCoercion
+ * .scala. Parity for a comparison whose common precision would exceed 38 is
not settled here;
+ * see HUDI #19860.
+ */
+ private def findWiderNumericType(types: Seq[DataType]): Option[DataType] = {
+ if (SQLConf.get.ansiEnabled) {
+ AnsiTypeCoercion.findWiderCommonType(types)
+ } else {
+ TypeCoercion.findWiderCommonType(types)
+ }
+ }
+
+ private def castTo(expression: Expression, dataType: DataType): Expression =
{
+ if (expression.dataType == dataType) expression else Cast(expression,
dataType)
+ }
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestFsViewProcedure.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestFsViewProcedure.scala
index d0ae75246a8d..1fbb36ebd60c 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestFsViewProcedure.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestFsViewProcedure.scala
@@ -60,6 +60,27 @@ class TestFsViewProcedure extends
HoodieSparkProcedureTestBase {
assertResult(2){
result1.length
}
+
+ // filter runs against the procedure output schema, where data_file_size
is a LongType
+ // column, so a literal of any numeric type, on either side, has to
widen to compare.
+ // See HUDI #19632.
+ val reversedFilter = spark.sql(
+ s"""call show_fsview_all(table => '$tableName', filter => '0 <
data_file_size')""".stripMargin).collect()
+ assertResult(2){
+ reversedFilter.length
+ }
+ val decimalFilter = spark.sql(
+ s"""call show_fsview_all(table => '$tableName', filter =>
'data_file_size >= 0.0')""".stripMargin).collect()
+ assertResult(2){
+ decimalFilter.length
+ }
+ // A filter that keeps nothing, widened the same way, so the pair above
is not just a filter
+ // being ignored.
+ val emptyFilter = spark.sql(
+ s"""call show_fsview_all(table => '$tableName', filter =>
'data_file_size > 9999999999.0')""".stripMargin).collect()
+ assertResult(0){
+ emptyFilter.length
+ }
}
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHoodieProcedureFilterUtils.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHoodieProcedureFilterUtils.scala
index 8478129a6f8c..64d16e581ba2 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHoodieProcedureFilterUtils.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHoodieProcedureFilterUtils.scala
@@ -17,12 +17,17 @@
package org.apache.spark.sql.hudi.procedure
+import org.apache.hudi.HoodieSparkUtils
+
import org.apache.spark.sql.Row
import org.apache.spark.sql.hudi.command.procedures.HoodieProcedureFilterUtils
import org.apache.spark.sql.types._
+import java.math.{BigDecimal => JBigDecimal}
import java.sql.{Date, Timestamp}
+import scala.collection.JavaConverters._
+
/**
* Direct unit tests for [[HoodieProcedureFilterUtils]] which evaluates SQL
filter
* expressions against procedure output rows. Covers
primitive/date/decimal/complex
@@ -40,6 +45,11 @@ class TestHoodieProcedureFilterUtils extends
HoodieSparkProcedureTestBase {
private def validate(expr: String, schema: StructType = scalarSchema):
Either[String, Unit] =
HoodieProcedureFilterUtils.validateFilterExpression(expr, schema, spark)
+ // Not the scalarRows factory: only id and ts matter to the widening tests
that use it.
+ private def tsRow(id: Int, ts: Long): Row =
+ Row(id, s"n$id", 10.0d * id, ts, true, -id,
+ Date.valueOf("2024-01-01"), Timestamp.valueOf("2024-01-01 00:00:00"))
+
// A rich scalar schema reused across the function tests.
private val scalarSchema = schemaOf(
"id" -> IntegerType,
@@ -69,58 +79,279 @@ class TestHoodieProcedureFilterUtils extends
HoodieSparkProcedureTestBase {
assertResult(2)(keep(scalarRows, "id <= 2", scalarSchema).length)
assertResult(Seq(scalarRows(1)))(keep(scalarRows, "id != 1", scalarSchema))
assertResult(Seq(scalarRows.head))(keep(scalarRows, "name = 'a1'",
scalarSchema))
- // The literal must match the column type, so use an explicit double
literal here; the plain
- // 15.0 (decimal) form is pinned in the numeric-coercion test below.
- assertResult(Seq(scalarRows(1)))(keep(scalarRows, "price > 15.0d",
scalarSchema))
+ // A plain 15.0 decimal literal is coerced with the double column.
+ assertResult(Seq(scalarRows(1)))(keep(scalarRows, "price > 15.0",
scalarSchema))
+ assertResult(Right(()))(validate("price > 15.0"))
// Bare boolean column evaluates to a Boolean result directly.
assertResult(Seq(scalarRows.head))(keep(scalarRows, "flag", scalarSchema))
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.
- 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))
+ // The integer literal is widened rather than the column narrowed, so a
Long past the Int
+ // range keeps its value instead of wrapping to -1294967296 and matching
"ts < 2000".
+ val bigRow = tsRow(3, 3000000000L)
+ val withBigRow = scalarRows :+ bigRow
+ assertResult(Seq(scalarRows(1), bigRow))(keep(withBigRow, "ts > 1500",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(withBigRow, "ts < 2000",
scalarSchema))
+ // Coercion applies symmetrically when the literal is on the left.
+ assertResult(Seq(scalarRows(1), bigRow))(keep(withBigRow, "1500 < ts",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(withBigRow, "1000 = ts",
scalarSchema))
+ // IN and <=> widen through the same path as the binary comparisons.
+ assertResult(scalarRows)(keep(withBigRow, "ts IN (1000, 2000)",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(withBigRow, "ts <=> 1000",
scalarSchema))
+ // An IN list of mixed integral widths widens to the single common type.
+ assertResult(scalarRows)(keep(withBigRow, "ts IN (1000, 2000L)",
scalarSchema))
+ // Every filterable procedure calls validateFilterExpression before
evaluateFilter, and each of
+ // these shapes was rejected there before the operands widened.
+ assertResult(Right(()))(validate("1500 < ts"))
+ assertResult(Right(()))(validate("ts IN (1000, 2000)"))
+ assertResult(Right(()))(validate("ts <=> 1000"))
+ // A non-numeric operand still bails out of the widening and stays
unresolved.
+ assert(validate("id IN (1, 'x')").isLeft)
+ // A null operand widens with the numeric ones, the plan Spark builds for
the same filter.
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "ts IN (1000, null)",
scalarSchema))
+ assertResult(Right(()))(validate("ts IN (1000, null)"))
+ assertResult(Seq.empty)(keep(scalarRows, "ts <=> null", scalarSchema))
+ assertResult(Right(()))(validate("ts <=> null"))
+ // A null operand takes its peers' type whether or not that type is
numeric.
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "name IN ('a1',
null)", scalarSchema))
+ assertResult(Right(()))(validate("name IN ('a1', null)"))
+ assertResult(Seq.empty)(keep(scalarRows, "name <=> null", scalarSchema))
+ assertResult(Right(()))(validate("name <=> null"))
}
- 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") {
val schema = schemaOf(
"f" -> FloatType,
"sh" -> ShortType,
"by" -> ByteType,
"dec" -> DecimalType(10, 2))
- val rows = Seq(Row(2.5f, 3.toShort, 4.toByte, new
java.math.BigDecimal("3.00")))
-
- // Literals whose parsed type already matches the column type evaluate
correctly.
- 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))
+ // Two rows, so an assertion that matches everything is distinguishable
from one that
+ // coerces correctly.
+ val rows = Seq(
+ Row(2.5f, 3.toShort, 4.toByte, new JBigDecimal("3.00")),
+ Row(0.5f, 9.toShort, 9.toByte, new JBigDecimal("0.50")))
+ val matched = Seq(rows.head)
+
+ // A literal whose parsed type already matches the column type evaluates
correctly.
+ assertResult(matched)(keep(rows, "f > 1.0f", schema))
+ // A decimal literal of another precision or scale widens to the common
decimal type, or the
+ // comparison stays unresolved and validateFilterExpression rejects the
filter.
+ assertResult(matched)(keep(rows, "dec > 1.00", schema))
+ assertResult(matched)(keep(rows, "dec > 1.5", schema))
+
+ // Mismatched numeric types are widened to Spark's common type.
+ assertResult(matched)(keep(rows, "sh = 3", schema))
+ assertResult(matched)(keep(rows, "by = 4", schema))
+ assertResult(matched)(keep(rows, "f > 1.0d", schema))
+ assertResult(matched)(keep(rows, "dec > 1", schema))
+ // Reversed operands widen the same way.
+ assertResult(matched)(keep(rows, "3 = sh", schema))
+ // 130 narrowed to a Byte is -126, so narrowing the literal would keep
neither row.
+ assertResult(rows)(keep(rows, "by < 130", schema))
+ // The same pairs have to pass validation, which every filterable
procedure runs first.
+ assertResult(Right(()))(validate("sh = 3", schema))
+ assertResult(Right(()))(validate("dec > 1", schema))
+ assertResult(Right(()))(validate("dec > 1.00", schema))
+ }
+
+ test("evaluateFilter surfaces an ANSI error instead of dropping the row") {
+ // An overflowing Long addition wraps to a negative without ANSI and
raises with it, exactly
+ // as it does in a query. The raise has to reach the caller: swallowing it
into "no match"
+ // would report an empty result for a filter the query answers with an
error.
+ val rows = Seq(tsRow(1, 1000L))
+ withSQLConf("spark.sql.ansi.enabled" -> "false") {
+ // 1000 + Long.MaxValue wraps to a negative, so the row genuinely fails
"> 0".
+ assertResult(Seq.empty)(keep(rows, "ts + 9223372036854775807 > 0",
scalarSchema))
+ }
+ withSQLConf("spark.sql.ansi.enabled" -> "true") {
+ intercept[ArithmeticException] {
+ keep(rows, "ts + 9223372036854775807 > 0", scalarSchema)
+ }
+ }
+ // An ANSI cast of a malformed string raises SparkNumberFormatException,
not an
+ // ArithmeticException, and has to reach the caller the same way.
+ withSQLConf("spark.sql.ansi.enabled" -> "false") {
+ assertResult(Seq.empty)(keep(scalarRows, "int(name) > 1", scalarSchema))
+ }
+ withSQLConf("spark.sql.ansi.enabled" -> "true") {
+ intercept[NumberFormatException] {
+ keep(scalarRows, "int(name) > 1", scalarSchema)
+ }
+ }
+ }
+
+ test("evaluateFilter follows ANSI mode when a decimal widening overflows") {
+ val schema = schemaOf("big" -> DecimalType(38, 0), "frac" ->
DecimalType(38, 18))
+ val rows = Seq(Row(new JBigDecimal("1" + "0" * 30), new
JBigDecimal("1.5")))
+ // On Spark 3 the widening picks DECIMAL(38,18), and the precision-38
clamp leaves it 20
+ // integral digits, too few for this 31-digit value, so the cast overflows
and the ANSI mode
+ // decides what happens. Spark 4 casts the fractional operand down
instead, planning the same
+ // filter as `big > cast(frac as decimal(38,0))`, so nothing overflows and
the row survives in
+ // either mode.
+ if (!HoodieSparkUtils.gteqSpark4_0) {
+ withSQLConf("spark.sql.ansi.enabled" -> "false") {
+ // The cast yields null and the row drops.
+ assertResult(Seq.empty)(keep(rows, "big > frac", schema))
+ }
+ withSQLConf("spark.sql.ansi.enabled" -> "true") {
+ intercept[ArithmeticException] {
+ keep(rows, "big > frac", schema)
+ }
+ }
+ } else {
+ for (ansi <- Seq("false", "true")) {
+ withSQLConf("spark.sql.ansi.enabled" -> ansi) {
+ assertResult(rows)(keep(rows, "big > frac", schema))
+ assertResult(Right(()))(validate("big > frac", schema))
+ }
+ }
+ }
+ }
+
+ test("evaluateFilter widens with the coercion rules of the active ANSI
mode") {
+ // The two coercion objects disagree on BIGINT with FLOAT:
AnsiTypeCoercion widens to DOUBLE,
+ // TypeCoercion follows numericPrecedence to FLOAT. 16777217 is the first
Long that a Float
+ // cannot hold, so it survives the widening under ANSI and rounds down to
16777216.0f without
+ // it. A filter has to widen the way the same comparison would in a query.
+ val rows = Seq(tsRow(1, 16777217L))
+ withSQLConf("spark.sql.ansi.enabled" -> "true") {
+ assertResult(rows)(keep(rows, "ts > 16777216.0f", scalarSchema))
+ }
+ withSQLConf("spark.sql.ansi.enabled" -> "false") {
+ assertResult(Seq.empty)(keep(rows, "ts > 16777216.0f", scalarSchema))
+ // The rounded-down Long still clears a smaller Float, so the widening
runs either way.
+ assertResult(rows)(keep(rows, "ts > 16777215.0f", scalarSchema))
+ }
+ // Column against column splits the same way, and only the widening
reaches it.
+ val pairSchema = schemaOf("l" -> LongType, "f" -> FloatType)
+ val pairRows = Seq(Row(16777217L, 16777216.0f))
+ withSQLConf("spark.sql.ansi.enabled" -> "true") {
+ assertResult(pairRows)(keep(pairRows, "l > f", pairSchema))
+ }
+ withSQLConf("spark.sql.ansi.enabled" -> "false") {
+ assertResult(Seq.empty)(keep(pairRows, "l > f", pairSchema))
+ }
+ }
+
+ test("evaluateFilter widens arithmetic and coalesce operands") {
+ assertResult(Seq(scalarRows(1)))(keep(scalarRows, "ts + 1 > 1500",
scalarSchema))
+ assertResult(Right(()))(validate("ts + 1 > 1500"))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "coalesce(ts, 0) =
1000", scalarSchema))
+ assertResult(Right(()))(validate("coalesce(ts, 0) = 1000"))
+ // Divide accepts only Double or Decimal, so an integral pair has to
become Double rather
+ // than a wider integral. Otherwise "ts / 2" stays unresolved while "price
/ 2" resolves.
+ assertResult(Seq(scalarRows(1)))(keep(scalarRows, "ts / 2 > 500",
scalarSchema))
+ assertResult(Right(()))(validate("ts / 2 > 500"))
+ assertResult(Seq(scalarRows(1)))(keep(scalarRows, "price / 2 > 5",
scalarSchema))
+ assertResult(Right(()))(validate("price / 2 > 5"))
+ // Spark's rule admits a null operand on either side, so these resolve and
evaluate to null
+ // rather than being rejected as an unsupported filter expression.
+ assertResult(Seq.empty)(keep(scalarRows, "ts / null > 0", scalarSchema))
+ assertResult(Right(()))(validate("ts / null > 0"))
+ assertResult(Seq.empty)(keep(scalarRows, "null / ts > 0", scalarSchema))
+ assertResult(Right(()))(validate("null / ts > 0"))
+ // The remaining arithmetic operators go through the same widening.
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "ts % 3 = 1",
scalarSchema))
+ assertResult(Right(()))(validate("ts % 3 = 1"))
+ assertResult(Seq(scalarRows(1)))(keep(scalarRows, "ts - 1 > 1500",
scalarSchema))
+ assertResult(Right(()))(validate("ts - 1 > 1500"))
+ // IntegralDivide accepts only Long or Decimal and never widens its
operands against each
+ // other, so an Int pair has to be promoted a side at a time the way
Spark's rule does.
+ assertResult(Seq(scalarRows(1)))(keep(scalarRows, "ts div 3 > 500",
scalarSchema))
+ assertResult(Right(()))(validate("ts div 3 > 500"))
+ assertResult(Seq(scalarRows(1)))(keep(scalarRows, "id div 2 > 0",
scalarSchema))
+ assertResult(Right(()))(validate("id div 2 > 0"))
+ // Coalesce widens across more than two children, and across widths as
well as kinds.
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "coalesce(ts, id, 0) =
1000", scalarSchema))
+ assertResult(Right(()))(validate("coalesce(ts, id, 0) = 1000"))
+ assertResult(Seq(scalarRows(1)))(keep(scalarRows, "coalesce(ts, price) >
1500", scalarSchema))
+ assertResult(Right(()))(validate("coalesce(ts, price) > 1500"))
+ // div is the operator this coercion newly reaches, so check it against
Spark itself.
+ val df = spark.createDataFrame(scalarRows.asJava, scalarSchema)
+ Seq("id div 2 > 0", "ts div 3 > 500").foreach { filter =>
+ withClue(s"filter=$filter: ") {
+ assertResult(df.filter(filter).collect().toSeq)(keep(scalarRows,
filter, scalarSchema))
+ }
+ }
+ }
+
+ test("evaluateFilter matches Spark for mixed decimal arithmetic") {
+ val schema = schemaOf("dec" -> DecimalType(38, 18), "i" -> IntegerType,
"f" -> FloatType)
+ val rows = Seq(
+ Row(new JBigDecimal("0.0000001"), 1, 0.5f),
+ Row(new JBigDecimal("-2.5"), 2, 0.25f))
+ val filters = Seq(
+ "dec + 1 > 0", "1 + dec > 0", "dec * 1 > 0", "1L * dec > 0",
+ "dec + i > 0", "i * dec > 0", "dec / 2 > 0", "2 / dec > 0",
+ "dec + 0.5f > 0", "0.5f + dec > 0", "dec * f > 0", "f / dec > 0",
+ "dec + 0.5d > 0", "0.5d / dec > 0",
+ "dec / null > 0", "null / dec > 0", "dec + null > 0", "null * dec > 0",
+ // DECIMAL(38,18) * DECIMAL(2,1) gives a scale-16 product that still
holds 0.0000001, which
+ // widening both operands to DECIMAL(38,18) first would round away to
zero.
+ "dec * 1.0 > 0.0")
+ for (ansi <- Seq("false", "true"); minimumPrecision <- Seq("false",
"true")) {
+ withSQLConf("spark.sql.ansi.enabled" -> ansi,
+ "spark.sql.legacy.literal.pickMinimumPrecision" -> minimumPrecision) {
+ val df = spark.createDataFrame(rows.asJava, schema)
+ filters.foreach { filter =>
+ withClue(s"filter=$filter, ansi=$ansi,
minimumPrecision=$minimumPrecision: ") {
+ assertResult(Right(()))(validate(filter, schema))
+ assertResult(df.filter(filter).collect().toSeq)(keep(rows, filter,
schema))
+ }
+ }
+ }
+ }
+ }
+
+ test("evaluateFilter matches Spark for high-precision decimal comparisons") {
+ val schema = schemaOf("ts" -> LongType, "dec" -> DecimalType(38, 30))
+ val rows = Seq(
+ Row(3000000000L, new JBigDecimal("0.00000000000000000000000000001")),
+ Row(0L, new JBigDecimal("0")),
+ Row(-3000000000L, new JBigDecimal("-0.00000000000000000000000000001")))
+ val tiny = "0.000000000000000000000000000001"
+ // A decimal literal past the Long range is folded to a constant by
DecimalPrecision, so the
+ // comparison never reaches the widening.
+ val huge = "99999999999999999999.0"
+ val filters = Seq(
+ s"ts > $tiny", s"ts >= $tiny", s"ts < $tiny", s"ts <= $tiny",
+ s"$tiny < ts", s"$tiny <= ts", s"$tiny > ts", s"$tiny >= ts",
+ s"ts > $huge", s"ts < $huge", s"$huge < ts",
+ "dec > 0", "0 < dec", "dec = 0", "dec <=> 0", "dec <= 0")
+ // spark.sql.legacy.decimal.retainFractionDigitsOnTruncate is undefined
before Spark 4.
+ val retainFractionSettings = if (HoodieSparkUtils.gteqSpark4_0)
Seq("false", "true") else Seq("false")
+ for (ansi <- Seq("false", "true"); minimumPrecision <- Seq("false",
"true");
+ retainFraction <- retainFractionSettings) {
+ withSQLConf("spark.sql.ansi.enabled" -> ansi,
+ "spark.sql.legacy.literal.pickMinimumPrecision" -> minimumPrecision,
+ "spark.sql.legacy.decimal.retainFractionDigitsOnTruncate" ->
retainFraction) {
+ val df = spark.createDataFrame(rows.asJava, schema)
+ filters.foreach { filter =>
+ withClue(s"filter=$filter, ansi=$ansi,
minimumPrecision=$minimumPrecision, retainFraction=$retainFraction: ") {
+ assertResult(Right(()))(validate(filter, schema))
+ assertResult(df.filter(filter).collect().toSeq)(keep(rows, filter,
schema))
+ }
+ }
+ }
+ }
+ }
+
+ test("evaluateFilter binds quoted column names") {
+ // show_column_stats_overlap, the second procedure named in #19632,
outputs columns like
+ // "Average overlap" and "50% overlap".
+ val schema = schemaOf("Average overlap" -> DoubleType, "50% overlap" ->
IntegerType)
+ val rows = Seq(Row(0.75d, 10), Row(0.25d, 20))
+ assertResult(Seq(rows.head))(keep(rows, "`Average overlap` > 0.5", schema))
+ assertResult(Right(()))(validate("`Average overlap` > 0.5", schema))
+ assertResult(Seq(rows(1)))(keep(rows, "`50% overlap` > 15", schema))
}
test("evaluateFilter silently drops rows for expressions it cannot resolve")
{
@@ -129,6 +360,9 @@ class TestHoodieProcedureFilterUtils extends
HoodieSparkProcedureTestBase {
assertResult(Seq.empty)(keep(scalarRows, "if(name = 'a1', true, false)",
scalarSchema))
assertResult(Seq(scalarRows.head))(
keep(scalarRows, "case when name = 'a1' then true else false end",
scalarSchema))
+ // Or short-circuits on the resolved side, which is what the
unresolved-operand guard preserves.
+ assertResult(Seq(scalarRows.head))(
+ keep(scalarRows, "id = 1 OR concat(name, 'x') = 'a1x'", scalarSchema))
}
test("evaluateFilter handles AND / OR / NOT / IN / BETWEEN") {
@@ -157,9 +391,8 @@ class TestHoodieProcedureFilterUtils extends
HoodieSparkProcedureTestBase {
}
test("evaluateFilter resolves numeric and cast functions") {
- // The util only special-cases a long column vs an integer literal; every
other numeric
- // comparison relies on the literal already matching the expression's
result type. So the
- // literals below are typed to match: round/double yield double,
ceil/floor/long yield long.
+ // Literals are typed to match each function's result type (round/double
yield double,
+ // ceil/floor/long yield long); the widening above would cover a mismatch
either way.
assertResult(Seq(scalarRows.head))(keep(scalarRows, "abs(neg) = 5",
scalarSchema))
assertResult(Seq(scalarRows.head))(keep(scalarRows, "round(price) =
10.0d", scalarSchema))
assertResult(Seq(scalarRows.head))(keep(scalarRows, "round(price, 1) =
10.0d", scalarSchema))
@@ -193,6 +426,9 @@ class TestHoodieProcedureFilterUtils extends
HoodieSparkProcedureTestBase {
test("evaluateFilter maps a string result of true / false to a boolean
decision") {
// string(flag) yields the literal strings "true"/"false", exercising the
string->boolean branch.
assertResult(Seq(scalarRows.head))(keep(scalarRows, "string(flag)",
scalarSchema))
+ // The mapping is for direct callers: Spark rejects a string filter
condition, so a filterable
+ // procedure never gets past validation with one.
+ assert(validate("string(flag)").isLeft)
}
test("evaluateFilter treats non-boolean-valued expressions as no-match") {
@@ -213,7 +449,6 @@ class TestHoodieProcedureFilterUtils extends
HoodieSparkProcedureTestBase {
}
test("evaluateFilter converts map columns and resolves map functions") {
- import scala.collection.JavaConverters._
val schema = schemaOf("id" -> IntegerType,
"mScala" -> MapType(StringType, IntegerType),
"mJava" -> MapType(StringType, IntegerType))
@@ -235,7 +470,6 @@ class TestHoodieProcedureFilterUtils extends
HoodieSparkProcedureTestBase {
}
test("evaluateFilter converts array / decimal / binary / uuid / java-time
columns without error") {
- import scala.collection.JavaConverters._
val schema = schemaOf(
"id" -> IntegerType,
"arrScala" -> ArrayType(IntegerType),
@@ -253,7 +487,7 @@ class TestHoodieProcedureFilterUtils extends
HoodieSparkProcedureTestBase {
Seq(1, 2, 3),
List(1, 2, 3).map(Int.box).asJava,
Array(1, 2, 3),
- new java.math.BigDecimal("12.50"),
+ new JBigDecimal("12.50"),
scala.math.BigDecimal("34.75"),
Array[Byte](1, 2, 3),
java.util.UUID.randomUUID(),
@@ -309,11 +543,15 @@ class TestHoodieProcedureFilterUtils extends
HoodieSparkProcedureTestBase {
assert(validate("if(name = 'a1', true, false)").isLeft)
assert(validate("substring(name, 2)").isLeft)
- assert(validate("id = 1 OR concat(name, 'x') = 'a1x'").isLeft)
+ assert(validate("id = 1 OR concat(name, 'x') =
'a1x'").left.exists(_.contains("Unsupported functions: concat")))
assert(validate("hour(t) = 12").isLeft)
assert(validate("date_format(t, 'yyyy') = '2024'").isLeft)
assert(validate("any_value(id) = 1").isLeft)
assert(validate("id = (select 1)").isLeft)
+ // Spark rejects any non-boolean filter condition. Without this these
resolve and report zero
+ // matching rows instead of the error the same query raises.
+ assert(validate("ts + 1").left.exists(_.contains("boolean")))
+ assert(validate("name").left.exists(_.contains("boolean")))
assertResult(Right(()))(validate("upper(name) = 'A1'"))
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestMetadataProcedure.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestMetadataProcedure.scala
index c4f330b38d22..00fe356ac49f 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestMetadataProcedure.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestMetadataProcedure.scala
@@ -270,6 +270,21 @@ class TestMetadataProcedure extends
HoodieSparkProcedureTestBase {
metadataStats(0)(9)
}
}
+ // The filter runs against the procedure output schema, where "Average
overlap" is a
+ // DoubleType column that a decimal literal has to widen against. See
HUDI #19632.
+ val kept = spark.sql(
+ s"""call show_metadata_column_stats_overlap(table => '$tableName',
targetColumns => 'c1',
+ | filter => '`Average overlap` >= 0.0')""".stripMargin).collect()
+ assertResult(1) {
+ kept.length
+ }
+ // Average overlap is a mean file count, never negative, so this keeps
nothing.
+ val dropped = spark.sql(
+ s"""call show_metadata_column_stats_overlap(table => '$tableName',
targetColumns => 'c1',
+ | filter => '`Average overlap` < 0.0')""".stripMargin).collect()
+ assertResult(0) {
+ dropped.length
+ }
}
}
