viirya commented on code in PR #59992:
URL: https://github.com/apache/spark/pull/59992#discussion_r4433267282
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/constraints.scala:
##########
@@ -128,6 +132,25 @@ case class CheckConstraint(
override def inputTypes: Seq[AbstractDataType] = Seq(BooleanType)
+ /**
+ * Fails the analysis if the condition calls a user-defined function. A
write can read `condition`
+ * again and resolve the name of the function in its own session, where the
name can refer to
+ * another function or to none, so the constraint would not enforce what
this DDL declares.
+ */
+ def failIfCallsUserDefinedFunction(): Unit = {
+ child.collectFirst {
+ case f: SQLFunctionExpression => toSQLId(f.function.name.nameParts)
+ case f: UserDefinedExpression => toSQLId(Seq(f.name))
+ case i: StaticInvoke if i.scalarFunction.isDefined =>
toSQLId(Seq(i.scalarFunction.get.name))
+ case Invoke(Literal(f: ScalarFunction[_], _), _, _, _, _, _, _, _) =>
toSQLId(Seq(f.name))
+ case a: V2Aggregator[_, _] => toSQLId(Seq(a.aggrFunc.name))
+ }.foreach { functionName =>
+ child.failAnalysis(
Review Comment:
For a V2 function this reports the bound name, not the name in the
condition: `fun_test_catalog.ns.strlen(name)` is reported as ``
`strlen_default` ``, and `avg` as `` `davg` ``. Can we call `failAnalysis` on
the matched node instead of `child`, so the query context points at the call?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveTableConstraints.scala:
##########
@@ -76,7 +77,20 @@ class ResolveTableConstraints(val catalogManager:
CatalogManager) extends Rule[L
private def buildCatalystExpression(c: Check): Expression = {
Option(c.predicate())
.flatMap(V2ExpressionUtils.toCatalyst)
-
.getOrElse(catalogManager.v1SessionCatalog.parser.parseExpression(c.predicateSql()))
+ .getOrElse(qualifyBuiltinFunctions(
+
catalogManager.v1SessionCatalog.parser.parseExpression(c.predicateSql())))
+ }
+
+ /**
+ * Qualifies the built-in functions a parsed check condition calls, so that
a temporary function
+ * or the SQL path of the session of the write cannot shadow them. Any other
name is left as is:
+ * a user-defined function in a constraint that the DDL no longer accepts
still resolves here.
+ */
+ private def qualifyBuiltinFunctions(condition: Expression): Expression =
condition.transform {
+ case f @ UnresolvedFunction(Seq(name), _, _, _, _, _, _, _)
+ if FunctionRegistry.builtin.functionExists(FunctionIdentifier(name)) =>
Review Comment:
`FunctionRegistry.builtin` does not include functions injected with
`SparkSessionExtensions.injectFunction`, so those can still be shadowed. Could
the Scaladoc say so?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/constraints.scala:
##########
@@ -128,6 +132,25 @@ case class CheckConstraint(
override def inputTypes: Seq[AbstractDataType] = Seq(BooleanType)
+ /**
+ * Fails the analysis if the condition calls a user-defined function. A
write can read `condition`
+ * again and resolve the name of the function in its own session, where the
name can refer to
+ * another function or to none, so the constraint would not enforce what
this DDL declares.
+ */
+ def failIfCallsUserDefinedFunction(): Unit = {
+ child.collectFirst {
+ case f: SQLFunctionExpression => toSQLId(f.function.name.nameParts)
+ case f: UserDefinedExpression => toSQLId(Seq(f.name))
+ case i: StaticInvoke if i.scalarFunction.isDefined =>
toSQLId(Seq(i.scalarFunction.get.name))
+ case Invoke(Literal(f: ScalarFunction[_], _), _, _, _, _, _, _, _) =>
toSQLId(Seq(f.name))
+ case a: V2Aggregator[_, _] => toSQLId(Seq(a.aggrFunc.name))
Review Comment:
`ALTER TABLE` already rejects aggregates, so this case only changes the
error of `CREATE TABLE`/`REPLACE TABLE`, which accept built-in aggregates.
Rejecting all aggregates there would make this case unnecessary. Fine to keep
for now.
##########
docs/sql-migration-guide.md:
##########
@@ -67,6 +67,7 @@ license: |
- Since Spark 4.4, when a format is specified, `to_timestamp`,
`try_to_timestamp`, and `to_timestamp_ltz` cast a numeric first argument to a
string and parse it with the format, for example `to_timestamp(20201231,
'yyyyMMdd')` returns `2020-12-31 00:00:00`. This matches the existing behavior
when `spark.sql.timestampType` is `TIMESTAMP_NTZ`. Previously, with the default
`TIMESTAMP_LTZ` timestamp type, such calls failed with `INTERNAL_ERROR`. Note
that the string form of the number is parsed, so a DECIMAL `1.5` is parsed as
`"1.5"`.
- Since Spark 4.4, the Spark Thrift Server logs a warning at startup when
`hive.server2.enable.doAs` is `true` and `hive.server2.authentication` verifies
user identities (any recognized type other than `NONE`/`NOSASL`), because
executor-side data access still runs as the server's own service identity
rather than the connecting user (SPARK-5159). Set
`spark.sql.hive.thriftServer.allowIneffectiveDoAs` to `true` to acknowledge the
limitation and silence the warning. Spark 5.0 is expected to refuse to start on
this configuration unless that option is set.
- Since Spark 4.4, `UPDATE` and `MERGE` on a Data Source V2 table fail with
`DATATYPE_MISMATCH.INVALID_ROW_LEVEL_OPERATION_ASSIGNMENTS` when they assign to
a metadata column of the table. This includes `UPDATE SET *` and `INSERT *` in
`MERGE WITH SCHEMA EVOLUTION` when the source has a column with the name of a
metadata column of the target table. Previously, these assignments passed
analysis and were usually ignored. To restore the previous behavior, set
`spark.sql.legacy.ignoreMetadataColumnAssignments` to `true`.
+- Since Spark 4.4, `CREATE TABLE`, `REPLACE TABLE`, and `ALTER TABLE ... ADD
CONSTRAINT` on a Data Source V2 table fail with
`USER_DEFINED_FUNCTION_IN_CHECK_CONSTRAINT` when the condition of a CHECK
constraint calls a user-defined function, including a Data Source V2 function.
Previously, they accepted such a condition, except that `ALTER TABLE ... ADD
CONSTRAINT` failed with `UNSUPPORTED_SQL_UDF_USAGE` on a SQL function, and a
write could resolve the function name in its own session, so that it enforced
another rule or failed (SPARK-60777). To run these statements, change the
condition to call only built-in functions. Existing constraints that call a
user-defined function are not rejected. Also, a write that parses the SQL text
of a condition binds an unqualified built-in function name to `system.builtin`,
regardless of the SQL path and `spark.sql.functionResolution.sessionOrder`, so
an existing constraint no longer calls another function with the same name.
Review Comment:
Nit: could you split this into two bullets, one for the DDL rejection and
one for binding built-ins to `system.builtin` on write?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -3110,6 +3110,9 @@ class Analyzer(
// of this rule.
case f: SQLTableFunction if hasSQLFunctionExpression(f.expressions) => f
+ // CheckAnalysis rejects a SQL function in the condition of a check
constraint.
+ case a: AddCheckConstraint => a.mapChildren(rewrite)
Review Comment:
This adds a command-specific case to a generic rule only to change the error
condition. `ALTER TABLE` already rejects SQL functions. Is the consistency
worth it?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CheckConstraintSuite.scala:
##########
@@ -170,6 +186,152 @@ class CheckConstraintSuite extends QueryTest with
CommandSuiteBase with DDLComma
}
}
+ private def withV2Function(name: String, function: UnboundFunction)(f: =>
Unit): Unit = {
+ val funCatalog = spark.sessionState.catalogManager.catalog(s"fun_$catalog")
+ .asInstanceOf[InMemoryCatalog]
+ val ident = Identifier.of(Array("ns"), name)
+ withNamespace(s"fun_$catalog.ns") {
+ sql(s"CREATE NAMESPACE fun_$catalog.ns")
+ funCatalog.createFunction(ident, function)
+ try f finally funCatalog.dropFunction(ident)
+ }
+ }
+
+ // CREATE TABLE, REPLACE TABLE and, if `inAlter`, ALTER TABLE fail on the
condition.
+ private def checkUserDefinedFunctionRejected(
+ cond: String,
+ functionName: String,
+ inAlter: Boolean = true): Unit = {
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id INT, name STRING) $defaultUsing")
+ val alter = if (inAlter) Seq(s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK
($cond)") else Nil
+ (alter ++ Seq(
+ s"CREATE TABLE ${t}_2 (id INT, name STRING, CONSTRAINT c1 CHECK
($cond)) $defaultUsing",
+ s"REPLACE TABLE $t (id INT, name STRING, CONSTRAINT c1 CHECK ($cond))
$defaultUsing"
+ )).foreach { query =>
+ checkError(
+ exception = intercept[AnalysisException](sql(query)),
+ condition = "USER_DEFINED_FUNCTION_IN_CHECK_CONSTRAINT",
+ sqlState = "42621",
+ parameters = Map("checkCondition" -> cond, "functionName" ->
functionName),
+ context = ExpectedContext(fragment = cond))
+ }
+ }
+ }
+
+ test("SPARK-60777: Check constraint cannot call a user-defined function") {
+ val call = s"fun_$catalog.ns.strlen"
+ Seq(
+ (StrLenDefault, s"$call(name) < 10", "`strlen_default`"),
+ // The optimizer folds a call with constant arguments, but the DDL does
not.
+ (StrLenDefault, s"id < $call('abc')", "`strlen_default`"),
+ (StrLenMagic, s"$call(name) < 10", "`strlen_magic`"),
+ (new JavaStrLenStaticMagic, s"$call(name) < 10", "`strlen`")
+ ).foreach { case (function, cond, functionName) =>
+ withV2Function("strlen", StrLen(function)) {
+ checkUserDefinedFunctionRejected(cond, functionName)
+ }
+ }
+
+ withUserDefinedFunction("tmp_len" -> true, "tmp_sql_len" -> true,
"sql_len" -> false) {
+ spark.udf.register("tmp_len", (s: String) => s.length)
+ checkUserDefinedFunctionRejected("tmp_len(name) < 10", "`tmp_len`")
+ sql("CREATE TEMPORARY FUNCTION tmp_sql_len(x STRING) RETURNS INT RETURN
length(x)")
+ checkUserDefinedFunctionRejected("tmp_sql_len(name) < 10",
"`tmp_sql_len`")
+ sql("CREATE FUNCTION sql_len(x STRING) RETURNS INT RETURN length(x)")
+ checkUserDefinedFunctionRejected("sql_len(name) < 10",
"`spark_catalog`.`default`.`sql_len`")
+ }
+ }
+
+ test("SPARK-60777: Check constraint cannot call a V2 aggregate function") {
+ withV2Function("avg", new JavaAverage) {
+ // ALTER TABLE keeps failing with the error for an aggregate in a filter.
+ checkUserDefinedFunctionRejected(
+ s"fun_$catalog.ns.avg(CAST(id AS DOUBLE)) > 0", "`davg`", inAlter =
false)
+ }
+ }
+
+ test("SPARK-60777: Existing check constraint calling a user-defined function
still works") {
+ withUserDefinedFunction("tmp_len" -> true) {
+ spark.udf.register("tmp_len", (s: String) => s.length)
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id INT, name STRING) $defaultUsing")
+ val check = Constraint.check("c1").predicateSql("tmp_len(name) <
10").build()
+
spark.sessionState.catalogManager.catalog(nonPartitionCatalog).asInstanceOf[TableCatalog]
+ .alterTable(Identifier.of(Array("ns"), "tbl"),
TableChange.addConstraint(check, null))
+ sql(s"INSERT INTO $t VALUES (1, 'abc')")
+ checkError(
+ exception = intercept[SparkRuntimeException] {
+ sql(s"INSERT INTO $t VALUES (2, 'abcdefghijklmnop')")
+ },
+ condition = "CHECK_CONSTRAINT_VIOLATION",
+ sqlState = "23001",
+ parameters = Map(
+ "constraintName" -> "c1",
+ "expression" -> "tmp_len(name) < 10",
+ "values" -> " - name : abcdefghijklmnop"))
+ checkAnswer(spark.table(t).select("id"), Row(1))
+ }
+ }
+ }
+
+ test("SPARK-60777: Check constraint is not shadowed in the session of a
write") {
+ def checkEnforced(t: String): Unit = {
+ // The settings of the write do shadow the built-in function outside of
the constraint.
+ checkAnswer(sql("SELECT upper('x')"), Row("OK"))
+ checkError(
+ exception = intercept[SparkRuntimeException] {
+ sql(s"INSERT INTO $t VALUES (1, 'x')")
+ },
+ condition = "CHECK_CONSTRAINT_VIOLATION",
+ sqlState = "23001",
+ parameters = Map(
+ "constraintName" -> "c1",
+ "expression" -> "UPPER(s) <> 'X'",
+ "values" -> " - s : x"))
+ checkAnswer(spark.table(t), Nil)
+ }
+
+ withV2Function("upper", StrLen(ConstantString)) {
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id INT, s STRING) $defaultUsing")
+ sql(s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK (UPPER(s) <> 'X')")
+ withSQLConf(SQLConf.PATH_ENABLED.key -> "true") {
+ sql(s"SET PATH = fun_$catalog.ns, system.builtin")
+ try checkEnforced(t) finally sql("SET PATH = DEFAULT_PATH")
+ }
+ spark.udf.register("upper", (_: String) => "OK")
+ try {
+ withSQLConf(SQLConf.SESSION_FUNCTION_RESOLUTION_ORDER.key ->
"first") {
+ checkEnforced(t)
+ }
+ } finally {
+ sql("DROP TEMPORARY FUNCTION upper")
+ }
+ }
+ }
+ }
+
+ test("SPARK-60777: Check constraint calling built-in functions is enforced")
{
+ Seq(
+ "UPPER(s) <> 'X'",
Review Comment:
The shadowing test above already covers this case.
--
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]
