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 b71b10bf477f fix(spark): reject unsupported procedure filter functions
(#19850)
b71b10bf477f is described below
commit b71b10bf477f45ac161fc4ef4fd8da71e6c3ec8a
Author: Ryu <[email protected]>
AuthorDate: Sun Sep 6 18:32:58 2026 +0200
fix(spark): reject unsupported procedure filter functions (#19850)
* fix(spark): reject unsupported procedure filter functions
* fix(spark): validate resolved filter expressions
* fix(spark): align filter validation coercion
---
.../procedures/HoodieProcedureFilterUtils.scala | 78 ++++++++++++++--------
.../procedure/TestHoodieProcedureFilterUtils.scala | 42 +++++++-----
.../hudi/procedure/TestShowCleansProcedures.scala | 4 ++
3 files changed, 81 insertions(+), 43 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 b0938733cbac..d0783fb872a5 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
-import org.apache.spark.sql.catalyst.expressions.{Expression,
GenericInternalRow}
+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.util.DateTimeUtils
import org.apache.spark.sql.types.{DataType, StructType}
import org.apache.spark.unsafe.types.UTF8String
+import java.util.Locale
+
import scala.collection.JavaConverters._
import scala.util.{Failure, Success, Try}
@@ -70,13 +72,9 @@ object HoodieProcedureFilterUtils {
}
}
- private def evaluateExpressionOnRow(expression: Expression, row: Row,
schema: StructType): Boolean = {
-
- val internalRow = convertRowToInternalRow(row, schema)
-
- Try {
- // First pass: bind attributes
- val attributeBound = expression.transform {
+ private def bindAndResolveExpression(expression: Expression, schema:
StructType): Expression = {
+ // First pass: bind attributes
+ val attributeBound = expression.transform {
case attr: org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
=>
try {
val fieldIndex = schema.fieldIndex(attr.name)
@@ -87,10 +85,10 @@ object HoodieProcedureFilterUtils {
}
}
- // Second pass: resolve functions
- val functionResolved = attributeBound.transform {
+ // Second pass: resolve functions
+ val functionResolved = attributeBound.transform {
case unresolvedFunc:
org.apache.spark.sql.catalyst.analysis.UnresolvedFunction =>
- unresolvedFunc.nameParts.head.toLowerCase match {
+ unresolvedFunc.nameParts.head.toLowerCase(Locale.ROOT) match {
case "upper" =>
if (unresolvedFunc.arguments.length == 1) {
org.apache.spark.sql.catalyst.expressions.Upper(unresolvedFunc.arguments.head)
@@ -352,21 +350,29 @@ object HoodieProcedureFilterUtils {
}
case _ => unresolvedFunc
}
- }
+ }
- // Third pass: handle type coercion for numeric comparisons
- val boundExpr = 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)
- case gt: org.apache.spark.sql.catalyst.expressions.GreaterThan =>
- applyTypeCoercion(gt.left, gt.right,
org.apache.spark.sql.catalyst.expressions.GreaterThan.apply, gt)
- case gte: org.apache.spark.sql.catalyst.expressions.GreaterThanOrEqual
=>
- applyTypeCoercion(gte.left, gte.right,
org.apache.spark.sql.catalyst.expressions.GreaterThanOrEqual.apply, gte)
- case lt: org.apache.spark.sql.catalyst.expressions.LessThan =>
- applyTypeCoercion(lt.left, lt.right,
org.apache.spark.sql.catalyst.expressions.LessThan.apply, lt)
- case lte: org.apache.spark.sql.catalyst.expressions.LessThanOrEqual =>
- applyTypeCoercion(lte.left, lte.right,
org.apache.spark.sql.catalyst.expressions.LessThanOrEqual.apply, lte)
- }
+ // 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)
+ case gt: org.apache.spark.sql.catalyst.expressions.GreaterThan =>
+ applyTypeCoercion(gt.left, gt.right,
org.apache.spark.sql.catalyst.expressions.GreaterThan.apply, gt)
+ case gte: org.apache.spark.sql.catalyst.expressions.GreaterThanOrEqual =>
+ applyTypeCoercion(gte.left, gte.right,
org.apache.spark.sql.catalyst.expressions.GreaterThanOrEqual.apply, gte)
+ case lt: org.apache.spark.sql.catalyst.expressions.LessThan =>
+ applyTypeCoercion(lt.left, lt.right,
org.apache.spark.sql.catalyst.expressions.LessThan.apply, lt)
+ case lte: org.apache.spark.sql.catalyst.expressions.LessThanOrEqual =>
+ applyTypeCoercion(lte.left, lte.right,
org.apache.spark.sql.catalyst.expressions.LessThanOrEqual.apply, lte)
+ }
+ }
+
+ private def evaluateExpressionOnRow(expression: Expression, row: Row,
schema: StructType): Boolean = {
+
+ val internalRow = convertRowToInternalRow(row, schema)
+
+ Try {
+ val boundExpr = bindAndResolveExpression(expression, schema)
val result = boundExpr.eval(internalRow)
result match {
@@ -468,9 +474,22 @@ object HoodieProcedureFilterUtils {
val columnNames = schema.fieldNames.toSet
val referencedColumns = extractColumnReferences(parsedExpr)
val invalidColumns = referencedColumns -- columnNames
+ val resolvedExpr = bindAndResolveExpression(parsedExpr, schema)
+ val unsupportedFunctions = extractFunctionReferences(resolvedExpr)
+ val unsupportedExpressions = resolvedExpr.collect {
+ case expression: Unevaluable
+ if !expression.isInstanceOf[UnresolvedAttribute]
+ && !expression.isInstanceOf[UnresolvedFunction] =>
expression.prettyName
+ }.toSet
if (invalidColumns.nonEmpty) {
Left(s"Invalid column references: ${invalidColumns.mkString(", ")}.
Available columns: ${columnNames.mkString(", ")}")
+ } else if (unsupportedFunctions.nonEmpty) {
+ Left(s"Unsupported functions:
${unsupportedFunctions.toSeq.sorted.mkString(", ")}")
+ } else if (!resolvedExpr.resolved || unsupportedExpressions.nonEmpty) {
+ val names = unsupportedExpressions.toSeq.sorted
+ val detail = if (names.nonEmpty) s": ${names.mkString(", ")}" else ""
+ Left(s"Unsupported filter expression$detail")
} else {
Right(())
}
@@ -481,6 +500,12 @@ object HoodieProcedureFilterUtils {
}
}
+ private def extractFunctionReferences(expression: Expression): Set[String] =
expression match {
+ case unresolved: UnresolvedFunction =>
+ Set(unresolved.nameParts.mkString(".")) ++
unresolved.children.flatMap(extractFunctionReferences)
+ case _ => expression.children.flatMap(extractFunctionReferences).toSet
+ }
+
private def extractColumnReferences(expression: Expression): Set[String] = {
import org.apache.spark.sql.catalyst.expressions._
@@ -505,4 +530,3 @@ object HoodieProcedureFilterUtils {
}
}
}
-
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 595b49a67929..8478129a6f8c 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
@@ -37,6 +37,9 @@ class TestHoodieProcedureFilterUtils extends
HoodieSparkProcedureTestBase {
private def keep(rows: Seq[Row], expr: String, schema: StructType): Seq[Row]
=
HoodieProcedureFilterUtils.evaluateFilter(rows, expr, schema, spark)
+ private def validate(expr: String, schema: StructType = scalarSchema):
Either[String, Unit] =
+ HoodieProcedureFilterUtils.validateFilterExpression(expr, schema, spark)
+
// A rich scalar schema reused across the function tests.
private val scalarSchema = schemaOf(
"id" -> IntegerType,
@@ -120,22 +123,12 @@ class TestHoodieProcedureFilterUtils extends
HoodieSparkProcedureTestBase {
assertResult(Seq.empty)(keep(scalarRows, "price > 15.0", scalarSchema))
}
- test("evaluateFilter silently drops rows for functions outside the
resolution table") {
- // Known limitation: a function missing from the resolution table falls
through as an
- // UnresolvedFunction. validateFilterExpression only checks column
references, so nothing
- // rejects it; instead evaluation fails per row and the row is dropped,
which looks like an
- // empty result rather than an error. Pinned here so a fix flips these;
see #19638.
+ test("evaluateFilter silently drops rows for expressions it cannot resolve")
{
assertResult(Seq.empty)(keep(scalarRows, "concat(name, 'x') = 'a1x'",
scalarSchema))
assertResult(Seq.empty)(keep(scalarRows, "instr(name, 'a') = 1",
scalarSchema))
- assertResult(Right(()))(
- HoodieProcedureFilterUtils.validateFilterExpression("concat(name, 'x') =
'a1x'", scalarSchema, spark))
- // if() is parsed as a function call and hits the same gap, while the
equivalent CASE WHEN is
- // lowered by the parser without an UnresolvedFunction and evaluates fine.
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))
- // Control: a function that is in the resolution table resolves and
matches.
- assertResult(Seq(scalarRows.head))(keep(scalarRows, "upper(name) = 'A1'",
scalarSchema))
}
test("evaluateFilter handles AND / OR / NOT / IN / BETWEEN") {
@@ -291,20 +284,37 @@ class TestHoodieProcedureFilterUtils extends
HoodieSparkProcedureTestBase {
test("validateFilterExpression accepts valid references and rejects unknown
ones") {
assertResult(Right(()))(
- HoodieProcedureFilterUtils.validateFilterExpression("id > 1 AND name =
'a1'", scalarSchema, spark))
+ validate("id > 1 AND name = 'a1'"))
+ assertResult(Right(()))(
+ validate("ts >= 0 AND ts BETWEEN 0 AND 999999"))
assertResult(Right(()))(
- HoodieProcedureFilterUtils.validateFilterExpression(null, scalarSchema,
spark))
+ validate(null))
assertResult(Right(()))(
- HoodieProcedureFilterUtils.validateFilterExpression(" ", scalarSchema,
spark))
+ validate(" "))
- val invalidCol =
HoodieProcedureFilterUtils.validateFilterExpression("missing_col > 1",
scalarSchema, spark)
+ val invalidCol = validate("missing_col > 1")
assert(invalidCol.isLeft)
val invalidColMsg = invalidCol.fold(identity, _ => "")
assert(invalidColMsg.contains("Invalid column references"))
assert(invalidColMsg.contains("missing_col"))
- val parseError = HoodieProcedureFilterUtils.validateFilterExpression("id
>< 1", scalarSchema, spark)
+ val parseError = validate("id >< 1")
assert(parseError.isLeft)
assert(parseError.fold(identity, _ => "").contains("Invalid filter
expression"))
}
+
+ test("validateFilterExpression rejects expressions the evaluator cannot
resolve") {
+ val unknown = validate("concat(name, 'x') = 'a1x' OR instr(name, 'a') = 1")
+ assert(unknown.left.exists(_.contains("Unsupported functions: concat,
instr")))
+
+ 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("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)
+
+ assertResult(Right(()))(validate("upper(name) = 'A1'"))
+ }
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestShowCleansProcedures.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestShowCleansProcedures.scala
index 9fac4df2f072..c5d8792712c5 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestShowCleansProcedures.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestShowCleansProcedures.scala
@@ -636,6 +636,10 @@ class TestShowCleansProcedures extends
HoodieSparkProcedureTestBase {
checkExceptionContain(
s"""call show_clean_plans(table => '$tableName', filter =>
"nonexistent_col > 1")""")(
"Invalid column references: nonexistent_col")
+
+ checkExceptionContain(
+ s"""call show_clean_plans(table => '$tableName', filter =>
"concat(action, 'x') = 'cleanx'")""")(
+ "Unsupported functions: concat")
}
}
}
