cloud-fan commented on code in PR #58867:
URL: https://github.com/apache/spark/pull/58867#discussion_r4499550366
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/NondeterministicExpressionCollection.scala:
##########
@@ -22,19 +22,42 @@ import java.util.LinkedHashMap
import org.apache.spark.sql.catalyst.expressions._
object NondeterministicExpressionCollection {
+ /**
+ * Returns a copy used only to classify the parent, plus the minimal
nondeterministic
+ * evaluation frontier below `expression`.
+ *
+ * A selected expression is represented by a deterministic attribute while
walking upward. If
+ * its parent is still nondeterministic after that substitution, the parent
is the evaluation
+ * unit and replaces the selected descendants in the frontier.
+ */
+ private def collectNondeterministicFrontier(
+ expression: Expression): (Expression, Seq[Expression]) = {
+ if (expression.deterministic) {
+ (expression, Seq.empty)
+ } else {
+ val childResults =
expression.children.map(collectNondeterministicFrontier)
Review Comment:
**Blocking (P1):** This recursion enters every child, including CASE WHEN,
IF and COALESCE branches and the lazily evaluated side of AND and OR, which run
only for some rows. A random-IV or non-constant-IV `aes_encrypt` in such a
branch becomes the extraction unit, and PullOutNondeterministic evaluates it in
a Project below the operator for every input row.
By this review's reading of the code, `SELECT * FROM t ORDER BY CASE WHEN
octet_length(k) IN (16, 24, 32) THEN aes_encrypt(v, k) END` runs on the base
branch but now throws invalidAesKeyLengthError for rows with a 10-byte key,
which the guard was written to skip. A per-row IV guarded by `octet_length(iv)
= 12` fails the same way with invalidAesIvLengthError, and so do the same
guards in GROUP BY and DISTRIBUTE BY. Rows the guard excludes also pay for an
AES evaluation. Nondeterministic UDFs inside conditionals already have this gap
on the base branch. No test covers an extraction unit under a conditional.
See **Shared repair plan 1** in the review body.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/misc.scala:
##########
@@ -412,6 +418,38 @@ case class CurrentUser()
final override val nodePatterns: Seq[TreePattern] = Seq(CURRENT_LIKE)
}
+private object AesEncryptDeterminism {
+ def apply(arguments: Seq[Expression]): Boolean = {
+ arguments.forall(_.deterministic) &&
+ (hasNonEmptyFixedValue(arguments(4)) || isEcbFixedValue(arguments(2)))
+ }
+
+ private def fixedValue(expression: Expression): Option[Any] = {
+ if (expression.resolved && expression.foldable && expression.deterministic
&&
Review Comment:
**Blocking (P1):** This check recognizes a constant IV or mode only if it is
already foldable and context-independent when `deterministic` is computed
during analysis. `to_binary(...)` and `encode(...)` are RuntimeReplaceable, so
neither is foldable until the optimizer's ReplaceExpressions runs, and a
session-variable reference is foldable but not context-independent. A GCM or
CBC call with such a constant non-empty IV is therefore classified as
nondeterministic, and its StaticInvoke replacement keeps `isDeterministic =
false` even after the IV folds, because AesEncrypt is replaced before its
children.
By this review's reading of the code, `SELECT * FROM a JOIN b ON a.enc =
aes_encrypt(b.x, '0000111122223333', 'GCM', 'DEFAULT',
to_binary('000000000000000000000000', 'hex'))` runs on the base branch but now
fails with INVALID_NON_DETERMINISTIC_EXPRESSIONS, although every row uses the
same 12-byte IV. The same call is rejected as a generated column or a new
DEFAULT value, INSERT into a table whose stored DEFAULT uses it fails because
the default no longer folds, and these calls lose constant folding, pushdown
and subexpression elimination. A constant 'ECB' mode built the same way is
affected too. Literal, cast and `unhex` IVs, which the tests use, are
unaffected, and no query returns a wrong result.
This extends an [existing
discussion](https://github.com/apache/spark/pull/58867#discussion_r4047296328).
**Recommended change:** Make fixedValue classify the expression the call
will actually execute: replace RuntimeReplaceable subexpressions the way
ReplaceExpressions does, then treat a resolved, deterministic and foldable
result as fixed, which is the condition under which constant folding would turn
it into a literal. Session-variable references and constant wrapper calls such
as to_binary(...) and encode(...) then count as fixed IVs, and a mode built by
a string function can count as ECB. Keep the current handling of null, empty,
invalid and unevaluable values, which still count as not fixed. Add expression
tests and analyzer-level SQL tests for these IV and mode forms, including a
join condition and a DEFAULT value.
**Why this works:** The classifier and constant folding then agree on what a
constant IV is. Any IV that the optimizer would fold to a non-empty value makes
the call deterministic before analysis checks run, so the wrapper's flag and
the frozen StaticInvoke flag both describe the IV the call executes with.
**Scope:** Align AES fixed-value classification with what constant folding
produces, and cover wrapper and variable IVs in tests.
**Compatibility:** Random-IV and non-constant-IV classification, runtime
encryption output, and the conservative handling of null, empty and unevaluable
IVs stay as this PR defines them.
**Risks:** Classification now evaluates foldable IVs and modes whose value
depends on session state, such as a session-timezone cast, during analysis.
Constant folding evaluates the same expressions later in the same query, and
the classifier uses only emptiness and the ECB name, so the classification
matches what the call executes. Calls with these IV forms become deterministic
again and their replacements foldable, so join conditions accept them, DEFAULT
values and generated columns accept the to_binary and encode forms, and the
optimizer folds and pushes them as on the base branch.
**Constraints:** An omitted, empty, null or per-row IV in CBC or GCM mode
must still make the call nondeterministic. A nondeterministic input, key, mode,
padding, IV or AAD must still make the call nondeterministic. An IV or mode
whose evaluation fails must still count as not fixed.
**Exclusions:**
- Per-row IV and mode columns stay nondeterministic, because a row may carry
an empty IV; whether that change needs a migration note is the open decision
about the new determinism rejections.
**Success:** aes_encrypt in GCM or CBC mode with an IV of to_binary('<hex>',
'hex'), encode('<text>', 'utf-8'), or a session variable holding a non-empty
binary value reports deterministic, and its replacement reports deterministic
and foldable. With a to_binary or encode IV, such a call is accepted in a join
condition, as a generated-column expression and as a DEFAULT value, as on the
base branch, and INSERT using such a DEFAULT succeeds. With a session-variable
IV it is accepted in a join condition; DEFAULT values and generated columns
reject session variables on both branches. Omitted, empty, null and column IVs
in CBC or GCM mode still report nondeterministic.
**Verification:**
- **Behavior:** Analyzer-resolved aes_encrypt calls with to_binary, encode
and session-variable IVs in GCM and CBC mode report deterministic, their
replacements report deterministic and foldable, and an ECB mode produced by a
string function is recognized as ECB.
- **Regression:** A join condition that uses aes_encrypt with a to_binary IV
analyzes without INVALID_NON_DETERMINISTIC_EXPRESSIONS, and a column DEFAULT
with such a call is accepted at table creation and resolves to a constant on
INSERT.
- **Compatibility:** Omitted-, empty-, null- and column-IV calls in CBC and
GCM mode remain nondeterministic and unfoldable, and a nondeterministic
argument in any position still makes the call nondeterministic.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/NondeterministicExpressionCollection.scala:
##########
@@ -22,19 +22,42 @@ import java.util.LinkedHashMap
import org.apache.spark.sql.catalyst.expressions._
object NondeterministicExpressionCollection {
+ /**
+ * Returns a copy used only to classify the parent, plus the minimal
nondeterministic
+ * evaluation frontier below `expression`.
+ *
+ * A selected expression is represented by a deterministic attribute while
walking upward. If
+ * its parent is still nondeterministic after that substitution, the parent
is the evaluation
+ * unit and replaces the selected descendants in the frontier.
+ */
+ private def collectNondeterministicFrontier(
+ expression: Expression): (Expression, Seq[Expression]) = {
+ if (expression.deterministic) {
+ (expression, Seq.empty)
+ } else {
+ val childResults =
expression.children.map(collectNondeterministicFrontier)
+ val expressionWithDeterministicChildren =
+ expression.withNewChildren(childResults.map(_._1))
Review Comment:
**Non-blocking (P2):** This rebuild breaks when a selected child is the
ListQuery of an IN predicate: `InSubquery.withNewChildrenInternal` casts its
last child to ListQuery, so the placeholder substitution throws
ClassCastException. The base branch rejects `SELECT * FROM t ORDER BY a IN
(SELECT b FROM s WHERE rand() > 0.5)` with a named analysis error and the GROUP
BY form with INTERNAL_ERROR. By this review's reading of the code, both now
fail with a raw ClassCastException that has no error class, because
QueryExecution converts only NullPointerException, AssertionError and
MatchError into internal errors.
An EXISTS whose subquery is nondeterministic is selected as a unit too and
moves into the child Project, where the IN/EXISTS placement check accepts it.
`ORDER BY EXISTS (SELECT 1 FROM s WHERE rand() > 0.5)` is therefore accepted,
while `ORDER BY EXISTS (SELECT 1 FROM s)` is still rejected with
UNSUPPORTED_IN_EXISTS_SUBQUERY. These queries are invalid, and none returns a
wrong result. No test covers subquery units.
**Verification:**
- **Regression:** IN and EXISTS subqueries with nondeterministic plans in
ORDER BY, GROUP BY and DISTRIBUTE BY produce the base branch's errors and never
a ClassCastException.
- **Compatibility:** A nondeterministic scalar subquery in GROUP BY and
ORDER BY is still extracted once into the child Project.
- **Compatibility:** rand() in the left-hand values of an IN subquery whose
plan is deterministic is still extracted on its own in GROUP BY.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/NondeterministicExpressionCollection.scala:
##########
@@ -22,19 +22,42 @@ import java.util.LinkedHashMap
import org.apache.spark.sql.catalyst.expressions._
object NondeterministicExpressionCollection {
+ /**
+ * Returns a copy used only to classify the parent, plus the minimal
nondeterministic
+ * evaluation frontier below `expression`.
+ *
+ * A selected expression is represented by a deterministic attribute while
walking upward. If
+ * its parent is still nondeterministic after that substitution, the parent
is the evaluation
+ * unit and replaces the selected descendants in the frontier.
+ */
+ private def collectNondeterministicFrontier(
+ expression: Expression): (Expression, Seq[Expression]) = {
+ if (expression.deterministic) {
+ (expression, Seq.empty)
+ } else {
+ val childResults =
expression.children.map(collectNondeterministicFrontier)
+ val expressionWithDeterministicChildren =
+ expression.withNewChildren(childResults.map(_._1))
+
+ if (!expressionWithDeterministicChildren.deterministic) {
+ val placeholder = AttributeReference(
+ "_nondeterministic", expression.dataType, expression.nullable)()
+ (placeholder, Seq(expression))
Review Comment:
**Blocking (P1):** When the call reads a lambda variable, this returns it as
the unit by itself. For `SELECT count(*) FROM t GROUP BY transform(arr, x ->
aes_encrypt(x, '0000111122223333'))`, the random-IV `aes_encrypt(x, ...)` is
selected because the enclosing LambdaFunction and ArrayTransform are
deterministic once it is a placeholder. PullOutNondeterministic then aliases it
in a Project over `t`, where `x` is not bound.
By this review's reading of the code, that query runs on the base branch but
now fails analysis with a missing-attributes error naming an internal lambda
variable. The failure does not depend on the data or on whether the ciphertext
is used: `GROUP BY size(transform(arr, x -> aes_encrypt(x, key)))` fails too,
and ORDER BY and DISTRIBUTE BY fail the same way. Lambda-bound `shuffle(x)` and
nondeterministic UDFs already have this gap on the base branch. No test covers
an extraction unit under a lambda.
This extends an [existing
discussion](https://github.com/apache/spark/pull/58867#discussion_r4047296335).
See **Shared repair plan 1** in the review body.
--
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]
