peter-toth commented on code in PR #58424: URL: https://github.com/apache/spark/pull/58424#discussion_r3966743760
########## sql/core/src/test/scala/org/apache/spark/sql/execution/RewriteSelfJoinInequalityToAggregateSuite.scala: ########## @@ -0,0 +1,1037 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.spark.sql.execution + +import org.apache.spark.SparkThrowable +import org.apache.spark.sql.{QueryTest, Row} +import org.apache.spark.sql.catalyst.expressions.Alias +import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.test.SharedSparkSession +import org.apache.spark.sql.types.{IntegerType, StructField, StructType} + +/** + * Correctness tests for [[RewriteSelfJoinInequalityToAggregate]]. + * + * Positive A' / A2 cases assert both result equivalence and that the rewrite actually fired. + * + * `assert(!ruleFired(plan))` on its own only proves the rewrite did not happen -- not that it was + * the guard under test that stopped it. A fixture whose two self-join sides are not structurally + * identical is rejected by `isSameBaseRelation` before any predicate is even parsed, and such a + * test passes while covering nothing. So six important rejection paths -- the predicate parser, the + * single-inequality requirement, output-position identity, the nondeterminism guard, the + * leaf-source allowlist (LogicalRDD vs Parquet), and the expression-type allowlist (`abs(v)` vs + * `v + 1`) -- are tested as single-variable pairs: the same fixture and the same query shape, one + * control query that must fire and one variant that changes only the feature under test and must + * not. A firing control does not pin the rejection to a particular line, but it does rule out an + * unrelated fixture mismatch as the reason its partner was rejected. The row-bag whitelist + * (Aggregate, Window) stays a plain negative: dropping the operator would change the query shape + * rather than one feature. + * + * Self-joined fixtures are real tables, not temp views over VALUES. Spark deduplicates a self-join + * over a [[org.apache.spark.sql.catalyst.analysis.MultiInstanceRelation]] via `newInstance()`, + * which refreshes one side's ExprIds without inserting a rename-only Project, so both sides stay + * structurally identical. A temp view over VALUES cannot, and Spark renames one side with a Project + * instead, which would make `isSameBaseRelation` false for every self-join below. `range()` needs + * no such treatment -- Range is a MultiInstanceRelation already. + */ +class RewriteSelfJoinInequalityToAggregateSuite extends QueryTest with SharedSparkSession { + + private val rewriteConf = SQLConf.REWRITE_SELF_JOIN_INEQUALITY_TO_AGGREGATE_ENABLED.key + + /** Signature alias produced by the rewrite; presence => rule definitely fired. */ + private val CountDistinctAlias = "_rewrite_selfjoin_inequality_cnt_distinct" + + private def ruleFired(plan: LogicalPlan): Boolean = + plan.exists { + p => + p.expressions.exists(_.exists { + case a: Alias if a.name == CountDistinctAlias => true + case _ => false + }) + } + + private def assertRuleFired(sql: String): Unit = { + withSQLConf(rewriteConf -> "true") { + val plan = spark.sql(sql).queryExecution.optimizedPlan + assert(ruleFired(plan), s"self-join inequality rewrite should fire:\n$plan") + } + } + + private def assertRuleNotFired(sql: String): Unit = { + withSQLConf(rewriteConf -> "true") { + val plan = spark.sql(sql).queryExecution.optimizedPlan + assert(!ruleFired(plan), s"self-join inequality rewrite must not fire:\n$plan") + } + } + + /** + * A real table, so that a self-join of it dedups into two structurally identical sides. See the + * class comment for why a temp view over VALUES cannot be used for a self-joined fixture. + */ + private def createTable(name: String, schema: String, values: String): Unit = { Review Comment: **Finding 2.** `createTable` drops before creating, but nothing drops at the end, so the suite leaves 19 tables (`T`, `TM`, `TN`, `AliasBase`, `TLeft`, `TRight`, `T2`, `T3`, `RddCtl`, `TFloat`, `TFloatKey`, `TCplx`, `TStrBin`, `TStrCiNeq`, `TStrCiEqui`, `TStr`, `TCharVarchar`, `TAnsi`, `TAnsiDiv`) and 4 temp views (`D`, `OuterKeys`, `OuterT`, `RddT`) behind. `SQLTestUtils` already gives you the scoped forms, so wrapping each test body in `withTable(...)` / `withTempView(...)` is the usual shape here. It also protects against a test that fails part-way leaving a half-populated table for the next one, which is a live concern for `TCharVarchar`: it is recreated inside a `Seq("CHAR(5)", "VARCHAR(5)").foreach` loop, so the second iteration depends on the first having finished. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/RewriteSelfJoinInequalityToAggregate.scala: ########## @@ -0,0 +1,746 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.execution + +import org.apache.spark.sql.catalyst.expressions._ +import org.apache.spark.sql.catalyst.expressions.aggregate._ +import org.apache.spark.sql.catalyst.plans._ +import org.apache.spark.sql.catalyst.plans.logical._ +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.catalyst.trees.TreePattern.IN_SUBQUERY +import org.apache.spark.sql.catalyst.util.CharVarcharUtils +import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, LogicalRelation} +import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types._ + +/** + * Rewrites a self-join with an inequality into GROUP BY + HAVING COUNT(DISTINCT) > 1. + * + * Targets the two uncorrelated InSubquery shapes exercised by TPC-DS Q95: + * + * - Pattern A': the subquery top-level InnerJoin is a direct self-join. + * - Pattern A2: the subquery contains an outer InnerJoin with a self-join child; only the + * self-join child is replaced with Aggregate and the outer join is preserved. + * + * Both patterns require an existence-only membership context so row-count multiplicity from the + * original self-join cross-product does not affect semantics. Correlated InSubquery expressions are + * intentionally fail-closed because the ExprId remapping performed here does not rewrite correlated + * predicates. + * + * This rule runs in `extendedOperatorOptimizationRules`, which is part of the operator optimization + * batch and therefore executes before `RewritePredicateSubquery` turns the predicate subquery into + * a semi/anti/existence join. It only observes the uncorrelated `InSubquery` shape at that phase, + * so there is no separate LeftSemi/LeftAnti ("Pattern A") or `Exists` handling. + * + * Both patterns share: + * - [[buildAggregateHavingDistinctGt1]] to construct `Filter(cnt > 1, Aggregate)` + * - [[canonicalizeWrapper]] to rebuild a wrapping Project so every equi-key reference points to + * the sjLeft-side attribute, with **fresh exprIds** (Spark's SPARK-21835 style -- no reuse of + * original exprIds), returning an old->new attribute remap for downstream rewrite. + * + * Controlled by `spark.sql.optimizer.rewriteSelfJoinInequalityToAggregate.enabled` + * (default false, opt-in). + */ +object RewriteSelfJoinInequalityToAggregate extends Rule[LogicalPlan] with PredicateHelper { + + private val CountDistinctAliasName = "_rewrite_selfjoin_inequality_cnt_distinct" + + override def apply(plan: LogicalPlan): LogicalPlan = { + if (!conf.getConf(SQLConf.REWRITE_SELF_JOIN_INEQUALITY_TO_AGGREGATE_ENABLED)) { + return plan + } + + // Pattern A' / A2: rewrite uncorrelated InSubquery plans. + // Correlated subqueries carry outer references / correlated join conditions in + // `SubqueryExpression.children`; fail closed because this rule does not remap them. + val rewritten = plan.transformAllExpressionsWithPruning(_.containsPattern(IN_SUBQUERY)) { + case in @ InSubquery(_, lq: ListQuery) if lq.children.isEmpty => + rewriteSubqueryPlan(lq.plan) match { + case Some(newSub) => in.copy(query = lq.copy(plan = newSub)) + case None => in + } + } + if (!(rewritten eq plan)) { + logDebug( + "RewriteSelfJoinInequalityToAggregate: rewrote self-join to " + + "GROUP BY + HAVING COUNT(DISTINCT) > 1") + } + rewritten + } + // ============================================================================ + // Shared helpers + // ============================================================================ + + /** + * Build `Filter(cnt > 1, Aggregate(equiKeys, [equiKeys, cnt_alias], Filter(IsNotNull(equiKeys), + * child)))`. Returns the Filter node whose output is `equiKeys ++ [count_alias_attr]`. + * + * The extra `IsNotNull(equiKeys)` filter is essential to preserve the original equi-join's NULL + * semantics. Under SQL 3VL, `left.k = right.k` never matches when either side is NULL, so the + * original self-join drops rows with NULL equi-keys. Aggregate, in contrast, groups NULL keys + * together into a single "NULL group" -- if that group has >= 2 distinct non-null neq values, + * COUNT(DISTINCT) > 1 fires and injects NULL into the subquery output. That leaked NULL then + * turns `NOT IN` into a spurious empty result (Spark's null-aware anti-join uses + * `Or(equi, IsNull(equi))` which any NULL sub-row satisfies) and can flip IN/NOT IN outcomes. The + * neq column needs no such filter: `COUNT(DISTINCT col)` already ignores NULL. + */ + private def buildAggregateHavingDistinctGt1( + equiKeys: Seq[Attribute], + neqCol: Attribute, + child: LogicalPlan): LogicalPlan = { + val countExpr = AggregateExpression( + Count(Seq(neqCol)), + mode = Complete, + isDistinct = true, + filter = None, + NamedExpression.newExprId) + val countAlias = Alias(countExpr, CountDistinctAliasName)() + // Seq[Attribute] is a Seq[NamedExpression] via covariance; no cast needed. + val aggExprs: Seq[NamedExpression] = equiKeys :+ countAlias + val nonNullChild = equiKeys + .map(a => IsNotNull(a): Expression) + .reduceOption(And) + .map(Filter(_, child)) + .getOrElse(child) + val agg = Aggregate(equiKeys, aggExprs, nonNullChild) + Filter(GreaterThan(countAlias.toAttribute, Literal(1L, LongType)), agg) + } + + /** + * Canonicalize a Project so every equi-key reference points at the sjLeft-side attribute. + * [[parseSelfJoinCondition]] has already verified that each pair refers to the same output + * position on the two structurally identical self-join sides. Uses **fresh exprIds** (no reuse of + * original wrapper output exprIds) -- the same technique Spark's own `dedupSubqueryOnSelfJoin` + * uses when it needs to change subquery output. + * + * Returns the rebuilt Project and a map `oldWrapperOutputExprId -> newWrapperOutputAttr`, so + * downstream references (outer join condition, top-level Project) can be updated consistently. + * + * `equiPairs` provides the definitive ExprId-based lookup: `equiPair (l, r)` binds + * `l.exprId -> l` (identity) and `r.exprId -> l` (sjRight -> sjLeft). Attribute identity in + * Catalyst is ExprId, not name; two columns can share a name with distinct ExprIds. Name-based + * lookup would silently drop such entries via `.toMap`. + * + * Fails (returns None) when a projectList entry is neither an equi-key Attribute (by ExprId) nor + * `Alias(equi-key Attribute, _)`. Fail-closed. + */ + private def canonicalizeWrapper( + projectList: Seq[NamedExpression], + equiPairs: Seq[(Attribute, Attribute)], + newChild: LogicalPlan): Option[(Project, Map[ExprId, Attribute])] = { + // ExprId-based canonical map: any equi-key attribute (either side) -> sjLeft attribute. + val exprIdToLeft: Map[ExprId, Attribute] = + equiPairs.flatMap { case (l, r) => Seq(l.exprId -> l, r.exprId -> l) }.toMap + val oldOutput: Seq[Attribute] = projectList.map(_.toAttribute) + val mapped: Seq[Option[NamedExpression]] = projectList.map { + case a: Attribute if exprIdToLeft.contains(a.exprId) => + // Wrap every rewritten output slot in a fresh Alias. + // + // When a wrapper reprojects BOTH sides of the same equi pair (e.g. + // `SELECT s1.k, s2.k FROM T s1 JOIN T s2 ON s1.k = s2.k AND s1.v <> s2.v`), + // both entries collapse to the same sjLeft Attribute after the self-join is + // rewritten. Duplicate output ExprIds are not illegal in Spark (`SELECT a, a` + // is a valid Project), but fresh Aliases give each output slot an independent + // identity, which keeps the `oldOutput -> newOutput` remap 1-to-1 and lets + // downstream references (outer join condition, top-level Project) be updated + // unambiguously via ExprId. + // + // The fresh ExprId is on the Alias ITSELF; the referenced child keeps its + // original ExprId. Spark's logical-plan integrity checks reject reusing a + // referenced ExprId as the Alias's own ExprId, not duplication across slots. + Some(Alias(exprIdToLeft(a.exprId), a.name)(): NamedExpression) + case al @ Alias(a: Attribute, _) if exprIdToLeft.contains(a.exprId) => + // Fresh exprId; do NOT reuse `al.exprId`. Reusing another expression's exprId + // is the pattern that Spark 3.3 flags via structural-integrity checks. + Some(Alias(exprIdToLeft(a.exprId), al.name)(): NamedExpression) Review Comment: **Finding 4.** This drops `al.explicitMetadata`, while the sibling helper `remapNamedExpressionAttributes` keeps it at line 208 (`Alias(newChild, al.name)(al.exprId, al.qualifier, al.explicitMetadata)`). I could not build a case where it changes a result: the wrapper only ever projects equi keys, and CHAR/VARCHAR is gated out before this point. So this is consistency between two helpers rather than a defect. ```scala Some(Alias(exprIdToLeft(a.exprId), al.name)( explicitMetadata = al.explicitMetadata): NamedExpression) ``` ########## sql/core/src/main/scala/org/apache/spark/sql/execution/RewriteSelfJoinInequalityToAggregate.scala: ########## @@ -0,0 +1,746 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.execution + +import org.apache.spark.sql.catalyst.expressions._ +import org.apache.spark.sql.catalyst.expressions.aggregate._ +import org.apache.spark.sql.catalyst.plans._ +import org.apache.spark.sql.catalyst.plans.logical._ +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.catalyst.trees.TreePattern.IN_SUBQUERY +import org.apache.spark.sql.catalyst.util.CharVarcharUtils +import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, LogicalRelation} +import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types._ + +/** + * Rewrites a self-join with an inequality into GROUP BY + HAVING COUNT(DISTINCT) > 1. + * + * Targets the two uncorrelated InSubquery shapes exercised by TPC-DS Q95: + * + * - Pattern A': the subquery top-level InnerJoin is a direct self-join. + * - Pattern A2: the subquery contains an outer InnerJoin with a self-join child; only the + * self-join child is replaced with Aggregate and the outer join is preserved. + * + * Both patterns require an existence-only membership context so row-count multiplicity from the + * original self-join cross-product does not affect semantics. Correlated InSubquery expressions are + * intentionally fail-closed because the ExprId remapping performed here does not rewrite correlated + * predicates. + * + * This rule runs in `extendedOperatorOptimizationRules`, which is part of the operator optimization + * batch and therefore executes before `RewritePredicateSubquery` turns the predicate subquery into + * a semi/anti/existence join. It only observes the uncorrelated `InSubquery` shape at that phase, + * so there is no separate LeftSemi/LeftAnti ("Pattern A") or `Exists` handling. + * + * Both patterns share: + * - [[buildAggregateHavingDistinctGt1]] to construct `Filter(cnt > 1, Aggregate)` + * - [[canonicalizeWrapper]] to rebuild a wrapping Project so every equi-key reference points to + * the sjLeft-side attribute, with **fresh exprIds** (Spark's SPARK-21835 style -- no reuse of + * original exprIds), returning an old->new attribute remap for downstream rewrite. + * + * Controlled by `spark.sql.optimizer.rewriteSelfJoinInequalityToAggregate.enabled` + * (default false, opt-in). + */ +object RewriteSelfJoinInequalityToAggregate extends Rule[LogicalPlan] with PredicateHelper { + + private val CountDistinctAliasName = "_rewrite_selfjoin_inequality_cnt_distinct" + + override def apply(plan: LogicalPlan): LogicalPlan = { + if (!conf.getConf(SQLConf.REWRITE_SELF_JOIN_INEQUALITY_TO_AGGREGATE_ENABLED)) { + return plan + } + + // Pattern A' / A2: rewrite uncorrelated InSubquery plans. + // Correlated subqueries carry outer references / correlated join conditions in + // `SubqueryExpression.children`; fail closed because this rule does not remap them. + val rewritten = plan.transformAllExpressionsWithPruning(_.containsPattern(IN_SUBQUERY)) { + case in @ InSubquery(_, lq: ListQuery) if lq.children.isEmpty => + rewriteSubqueryPlan(lq.plan) match { + case Some(newSub) => in.copy(query = lq.copy(plan = newSub)) + case None => in + } + } + if (!(rewritten eq plan)) { + logDebug( + "RewriteSelfJoinInequalityToAggregate: rewrote self-join to " + + "GROUP BY + HAVING COUNT(DISTINCT) > 1") + } + rewritten + } + // ============================================================================ + // Shared helpers + // ============================================================================ + + /** + * Build `Filter(cnt > 1, Aggregate(equiKeys, [equiKeys, cnt_alias], Filter(IsNotNull(equiKeys), + * child)))`. Returns the Filter node whose output is `equiKeys ++ [count_alias_attr]`. + * + * The extra `IsNotNull(equiKeys)` filter is essential to preserve the original equi-join's NULL + * semantics. Under SQL 3VL, `left.k = right.k` never matches when either side is NULL, so the + * original self-join drops rows with NULL equi-keys. Aggregate, in contrast, groups NULL keys + * together into a single "NULL group" -- if that group has >= 2 distinct non-null neq values, + * COUNT(DISTINCT) > 1 fires and injects NULL into the subquery output. That leaked NULL then + * turns `NOT IN` into a spurious empty result (Spark's null-aware anti-join uses + * `Or(equi, IsNull(equi))` which any NULL sub-row satisfies) and can flip IN/NOT IN outcomes. The + * neq column needs no such filter: `COUNT(DISTINCT col)` already ignores NULL. + */ + private def buildAggregateHavingDistinctGt1( + equiKeys: Seq[Attribute], + neqCol: Attribute, + child: LogicalPlan): LogicalPlan = { + val countExpr = AggregateExpression( + Count(Seq(neqCol)), + mode = Complete, + isDistinct = true, + filter = None, + NamedExpression.newExprId) + val countAlias = Alias(countExpr, CountDistinctAliasName)() + // Seq[Attribute] is a Seq[NamedExpression] via covariance; no cast needed. + val aggExprs: Seq[NamedExpression] = equiKeys :+ countAlias + val nonNullChild = equiKeys + .map(a => IsNotNull(a): Expression) + .reduceOption(And) + .map(Filter(_, child)) + .getOrElse(child) + val agg = Aggregate(equiKeys, aggExprs, nonNullChild) + Filter(GreaterThan(countAlias.toAttribute, Literal(1L, LongType)), agg) + } + + /** + * Canonicalize a Project so every equi-key reference points at the sjLeft-side attribute. + * [[parseSelfJoinCondition]] has already verified that each pair refers to the same output + * position on the two structurally identical self-join sides. Uses **fresh exprIds** (no reuse of + * original wrapper output exprIds) -- the same technique Spark's own `dedupSubqueryOnSelfJoin` + * uses when it needs to change subquery output. + * + * Returns the rebuilt Project and a map `oldWrapperOutputExprId -> newWrapperOutputAttr`, so + * downstream references (outer join condition, top-level Project) can be updated consistently. + * + * `equiPairs` provides the definitive ExprId-based lookup: `equiPair (l, r)` binds + * `l.exprId -> l` (identity) and `r.exprId -> l` (sjRight -> sjLeft). Attribute identity in + * Catalyst is ExprId, not name; two columns can share a name with distinct ExprIds. Name-based + * lookup would silently drop such entries via `.toMap`. + * + * Fails (returns None) when a projectList entry is neither an equi-key Attribute (by ExprId) nor + * `Alias(equi-key Attribute, _)`. Fail-closed. + */ + private def canonicalizeWrapper( + projectList: Seq[NamedExpression], + equiPairs: Seq[(Attribute, Attribute)], + newChild: LogicalPlan): Option[(Project, Map[ExprId, Attribute])] = { + // ExprId-based canonical map: any equi-key attribute (either side) -> sjLeft attribute. + val exprIdToLeft: Map[ExprId, Attribute] = + equiPairs.flatMap { case (l, r) => Seq(l.exprId -> l, r.exprId -> l) }.toMap + val oldOutput: Seq[Attribute] = projectList.map(_.toAttribute) + val mapped: Seq[Option[NamedExpression]] = projectList.map { + case a: Attribute if exprIdToLeft.contains(a.exprId) => + // Wrap every rewritten output slot in a fresh Alias. + // + // When a wrapper reprojects BOTH sides of the same equi pair (e.g. + // `SELECT s1.k, s2.k FROM T s1 JOIN T s2 ON s1.k = s2.k AND s1.v <> s2.v`), + // both entries collapse to the same sjLeft Attribute after the self-join is + // rewritten. Duplicate output ExprIds are not illegal in Spark (`SELECT a, a` + // is a valid Project), but fresh Aliases give each output slot an independent + // identity, which keeps the `oldOutput -> newOutput` remap 1-to-1 and lets + // downstream references (outer join condition, top-level Project) be updated + // unambiguously via ExprId. + // + // The fresh ExprId is on the Alias ITSELF; the referenced child keeps its + // original ExprId. Spark's logical-plan integrity checks reject reusing a + // referenced ExprId as the Alias's own ExprId, not duplication across slots. + Some(Alias(exprIdToLeft(a.exprId), a.name)(): NamedExpression) + case al @ Alias(a: Attribute, _) if exprIdToLeft.contains(a.exprId) => + // Fresh exprId; do NOT reuse `al.exprId`. Reusing another expression's exprId + // is the pattern that Spark 3.3 flags via structural-integrity checks. + Some(Alias(exprIdToLeft(a.exprId), al.name)(): NamedExpression) + case _ => None + } + if (mapped.exists(_.isEmpty)) { + None + } else { + val newProjectList = mapped.flatten + val newWrapper = Project(newProjectList, newChild) + val newOutput = newWrapper.output + val remap: Map[ExprId, Attribute] = + oldOutput.zip(newOutput).map { case (o, n) => o.exprId -> n }.toMap + Some((newWrapper, remap)) + } + } + + /** + * Replace equi-key attribute references inside a NamedExpression according to `remap`, while + * preserving the NamedExpression shape. + * + * `Expression.transformUp` returns `Expression`, not `NamedExpression`. We avoid a blanket + * `asInstanceOf[NamedExpression]` by handling the two shapes that can appear in a Project's + * `projectList` explicitly: a bare Attribute (whose top-level may itself be replaced) and an + * Alias (which stays an Alias while its child is transformed). Any other NamedExpression shape we + * do not rewrite is left as-is ONLY if it does not reference a replaced self-join output; + * otherwise it would carry a stale ExprId, so returns None to fail the whole rewrite closed. + */ + private def remapNamedExpressionAttributes( + ne: NamedExpression, + remap: Map[ExprId, Attribute]): Option[NamedExpression] = ne match { + case a: Attribute if remap.contains(a.exprId) => Some(remap(a.exprId)) + case a: Attribute => Some(a) + case al: Alias => + val newChild = al.child.transformUp { + case a: Attribute if remap.contains(a.exprId) => remap(a.exprId) + } + Some( + if (newChild eq al.child) al + else Alias(newChild, al.name)(al.exprId, al.qualifier, al.explicitMetadata)) + case other if other.references.exists(a => remap.contains(a.exprId)) => + // Fail-closed: a NamedExpression we do not rewrite (neither a bare Attribute nor an Alias) + // that still references a replaced self-join output would be left with a dangling ExprId. + // Refuse the rewrite rather than emit a plan with a stale reference. + None + case other => Some(other) + } + + // ============================================================================ + // Pattern A' / A2 dispatch (subquery plans of InSubquery) + // ============================================================================ + + private def rewriteSubqueryPlan(plan: LogicalPlan): Option[LogicalPlan] = { + // Candidate-level nondeterminism guard: reject if ANY node in the whole subquery plan + // is non-repeatable (Rand, LIMIT-without-ORDER-BY, Sample, Offset, streaming). This catches + // nondeterminism that lives ABOVE the self-join rather than on either side -- e.g. a Pattern + // A2 outer join whose condition is `d.k = sj.k AND rand() < 0.5`. Both self-join sides stay + // repeatable there, so the per-side `isSameBaseRelation` check would pass, yet the enclosing + // subquery is not repeatable. + if (!isRepeatablePlan(plan)) return None + + val (projectListOpt, innerJoin): (Option[Seq[NamedExpression]], Join) = plan match { + case Project(pl, j: Join) if j.joinType == Inner && j.condition.isDefined => + (Some(pl), j) + case j: Join if j.joinType == Inner && j.condition.isDefined => + (None, j) + case _ => return None + } + + if (isSameBaseRelation(innerJoin.left, innerJoin.right)) { + rewriteDirectSelfJoin(projectListOpt, innerJoin) + } else { + rewriteNestedSelfJoin(projectListOpt, innerJoin) + } + } + + // ============================================================================ + // Pattern A' : direct self-join at subquery top level + // ============================================================================ + + private def rewriteDirectSelfJoin( + projectListOpt: Option[Seq[NamedExpression]], + innerJoin: Join): Option[LogicalPlan] = { + val innerLeft = innerJoin.left + val innerRight = innerJoin.right + val innerCond = innerJoin.condition.get + + val parsed = parseSelfJoinCondition(innerCond, innerLeft, innerRight) + if (parsed.isEmpty) return None + // parseSelfJoinCondition has validated column correspondence and equi-key uniqueness. + val (equiPairs, neqPairs) = parsed.get + + val innerLeftEquiAttrs: Seq[Attribute] = equiPairs.map(_._1) + val innerLeftNeqAttr: Attribute = neqPairs.head._1 + val filtered = buildAggregateHavingDistinctGt1(innerLeftEquiAttrs, innerLeftNeqAttr, innerLeft) + + // Fail-closed on bare-Join subqueries: without a wrapping Project the subquery output + // is the full self-join output (both sides' columns). Replacing that with + // `Project(equiKeys, filtered)` shrinks the output; if the enclosing InSubquery + // referenced a non-equi column by position, `values.zip(sub.output).map(EqualTo.tupled)` + // inside RewritePredicateSubquery would build an incorrect semi condition. Q95's + // subqueries all have an explicit Project wrapper, so this branch does not affect it. + projectListOpt match { + case None => + None + case Some(pl) => + canonicalizeWrapper(pl, equiPairs, filtered).map { + case (newWrapper, _) => + logDebug( + s"Pattern A' - equiKeys=[${innerLeftEquiAttrs.map(_.name).mkString(",")}]" + + s", neqCol=${innerLeftNeqAttr.name}" + + s", outCols=[${newWrapper.projectList.map(_.name).mkString(",")}]") + newWrapper + } + } + } + + // ============================================================================ + // Pattern A2 : self-join nested inside another InnerJoin in the subquery + // ============================================================================ + + private def rewriteNestedSelfJoin( + projectListOpt: Option[Seq[NamedExpression]], + outerJoin: Join): Option[LogicalPlan] = { + val outerCond = outerJoin.condition.get + + val (selfJoinSide, selfJoinOnRight) = + tryExtractSelfJoin(outerJoin.right) match { + case Some(_) => (outerJoin.right, true) + case None => + tryExtractSelfJoin(outerJoin.left) match { + case Some(_) => (outerJoin.left, false) + case None => return None + } + } + + val (selfJoinProjectOpt, selfJoin) = selfJoinSide match { + case p @ Project(_, j: Join) if j.joinType == Inner && j.condition.isDefined => + (Some(p), j) + case j: Join if j.joinType == Inner && j.condition.isDefined => + (None, j) + case _ => return None + } + + val sjLeft = selfJoin.left + val sjRight = selfJoin.right + val sjCond = selfJoin.condition.get + if (!isSameBaseRelation(sjLeft, sjRight)) return None + + val parsed = parseSelfJoinCondition(sjCond, sjLeft, sjRight) + if (parsed.isEmpty) return None + // parseSelfJoinCondition has validated column correspondence and equi-key uniqueness. + val (equiPairs, neqPairs) = parsed.get + + val sjLeftEquiAttrs: Seq[Attribute] = equiPairs.map(_._1) + val sjLeftNeqAttr: Attribute = neqPairs.head._1 + + val selfJoinOutputSet = selfJoinSide.outputSet + val sjEquiExprIds: Set[ExprId] = + equiPairs.flatMap { case (l, r) => Seq(l.exprId, r.exprId) }.toSet + // wrapper Project may reproject equi-keys under fresh alias exprIds; include those. + val wrapperEquiExprIds: Set[ExprId] = selfJoinProjectOpt.toSeq.flatMap { + p => + p.projectList.flatMap { + case a: Attribute if sjEquiExprIds.contains(a.exprId) => Some(a.exprId) + case al @ Alias(a: Attribute, _) if sjEquiExprIds.contains(a.exprId) => Some(al.exprId) + case _ => None + } + }.toSet + val allEquiExprIds = sjEquiExprIds ++ wrapperEquiExprIds + + // Outer join condition may reference only equi-key attrs from the self-join side. + val outerCondRefs = outerCond.references.filter(selfJoinOutputSet.contains) + if (!outerCondRefs.forall(a => allEquiExprIds.contains(a.exprId))) return None + + // Top-level subquery Project may reference only equi-key attrs from the self-join side. + val projectOk = projectListOpt.forall { + pl => + val refs = pl.flatMap(_.references).filter(selfJoinOutputSet.contains) + refs.forall(a => allEquiExprIds.contains(a.exprId)) + } + if (!projectOk) return None + + val filtered = buildAggregateHavingDistinctGt1(sjLeftEquiAttrs, sjLeftNeqAttr, sjLeft) + + val (newSelfJoinSide, outputRemap): (LogicalPlan, Map[ExprId, Attribute]) = + selfJoinProjectOpt match { + case Some(wp) => + canonicalizeWrapper(wp.projectList, equiPairs, filtered) match { + case Some((newWrapper, remap)) => (newWrapper, remap) + case None => return None + } + case None if projectListOpt.isEmpty => + // Fail-closed: with neither a wrapper Project around the self-join nor a top-level + // subquery Project, the outer join currently exposes every self-join column, and + // replacing the self-join with `Project(equiKeys, filtered)` would shrink the outer + // join's right-hand output arity. RewritePredicateSubquery's positional zip + // (`values.zip(sub.output).map(EqualTo.tupled)`) would then bind semi predicates to + // the wrong attributes -- silently dropping components of a tuple IN. A + // top-level Project (`projectListOpt`) is what would let the arity be preserved + // by the top-level rewrite loop; without one, refuse to rewrite. + return None + case None => + // No wrapper Project but there IS a top-level subquery Project: shrinking the outer + // join's self-join-side output is safe because the top-level Project is rewritten + // consistently via `outputRemap` below and the top-level rewrite loop ensures + // subquery output arity matches what the enclosing InSubquery expects. + // Outer references may point at sjRight equi-attributes; remap them to sjLeft + // (same output position in a valid self-join). + val newP = Project(sjLeftEquiAttrs, filtered) + val remap: Map[ExprId, Attribute] = + equiPairs.map { case (l, r) => r.exprId -> l }.toMap + (newP, remap) + } + + // Rewrite outer join condition to use new wrapper output attributes. + val newOuterCond = outerCond.transformUp { + case a: Attribute if outputRemap.contains(a.exprId) => outputRemap(a.exprId) + } + + val newOuterJoin = if (selfJoinOnRight) { + outerJoin.copy(right = newSelfJoinSide, condition = Some(newOuterCond)) + } else { + outerJoin.copy(left = newSelfJoinSide, condition = Some(newOuterCond)) + } + + // Rewrite top-level Project references. + val result = projectListOpt match { + case Some(pl) => + val remapped = pl.map(ne => remapNamedExpressionAttributes(ne, outputRemap)) + if (remapped.exists(_.isEmpty)) return None + Project(remapped.flatten, newOuterJoin) + case None => newOuterJoin + } + + logDebug( + s"Pattern A2 - equiKeys=[${sjLeftEquiAttrs.map(_.name).mkString(",")}]" + + s", neqCol=${sjLeftNeqAttr.name}") + Some(result) + } + + private def tryExtractSelfJoin(plan: LogicalPlan): Option[Join] = { + val join = plan match { + case Project(_, j: Join) if j.joinType == Inner && j.condition.isDefined => j + case j: Join if j.joinType == Inner && j.condition.isDefined => j + case _ => return None + } + if (!isSameBaseRelation(join.left, join.right)) return None + val parsed = parseSelfJoinCondition(join.condition.get, join.left, join.right) + if (parsed.isEmpty) return None + Some(join) + } + + // ============================================================================ + // parseSelfJoinCondition + isSameBaseRelation + // ============================================================================ + + private def outputOrdinal(plan: LogicalPlan, attr: Attribute): Int = + plan.output.indexWhere(_.exprId == attr.exprId) + + private def sameOutputPosition( + leftPlan: LogicalPlan, + rightPlan: LogicalPlan, + leftAttr: Attribute, + rightAttr: Attribute): Boolean = { + val leftPos = outputOrdinal(leftPlan, leftAttr) + val rightPos = outputOrdinal(rightPlan, rightAttr) + leftPos >= 0 && rightPos >= 0 && leftPos == rightPos + } + + /** + * Parse a join condition into equi-pairs and inequality-pairs. Accepts only: + * - `EqualTo(attr, attr)` where the two attrs come from opposite sides, + * - `Not(EqualTo(attr, attr))` -- same side rule, + * - `IsNotNull(attr)` where the attr is one of the join columns. + * Anything else in the condition disqualifies the whole rewrite (fail-closed). + */ + private def parseSelfJoinCondition( + condition: Expression, + leftPlan: LogicalPlan, + rightPlan: LogicalPlan) + : Option[(Seq[(Attribute, Attribute)], Seq[(Attribute, Attribute)])] = { + + val leftOutput = leftPlan.outputSet + val rightOutput = rightPlan.outputSet + val predicates = splitConjunctivePredicates(condition) + + val equiPairs = predicates.collect { + case EqualTo(l: Attribute, r: Attribute) + if leftOutput.contains(l) && rightOutput.contains(r) => + (l, r) + case EqualTo(r: Attribute, l: Attribute) + if leftOutput.contains(l) && rightOutput.contains(r) => + (l, r) + } + + val neqPairs = predicates.collect { + case Not(EqualTo(l: Attribute, r: Attribute)) + if leftOutput.contains(l) && rightOutput.contains(r) => + (l, r) + case Not(EqualTo(r: Attribute, l: Attribute)) + if leftOutput.contains(l) && rightOutput.contains(r) => + (l, r) + } + + // Only IsNotNull predicates on join columns are safe to drop -- they're redundant with + // the join semantics or auto-added by InferFiltersFromConstraints. IsNotNull on other + // columns changes semantics if we drop it; bail out. + val joinAttrIds: Set[ExprId] = + (equiPairs ++ neqPairs).flatMap { case (l, r) => Seq(l.exprId, r.exprId) }.toSet + val isNotNullOnJoinCols = predicates.count { + case IsNotNull(a: Attribute) if joinAttrIds.contains(a.exprId) => true + case _ => false + } + + val totalMatched = equiPairs.size + neqPairs.size + isNotNullOnJoinCols + if (totalMatched != predicates.size) return None + + if (equiPairs.isEmpty || neqPairs.isEmpty) return None + + // Only rewrite the single-inequality case. Multiple inequality conjuncts cannot be represented + // by COUNT(DISTINCT) over a single column. + if (neqPairs.size != 1) return None + + // The rewrite swaps SQL comparison equality for grouping/DISTINCT equality: `<>` becomes + // COUNT(DISTINCT neqCol) and `=` becomes GROUP BY equiKey. It is only sound on types where + // those two notions of equality coincide, so gate every equi-key and the neq column on a + // positive `isSafeComparisonGroupingType` allowlist rather than `RowOrdering.isOrderable`: + // orderable only proves an order/hash exists, not that comparison and grouping agree. Fail + // closed on anything not proven safe (float -0.0/NaN, complex types, non-binary collations, + // and future/unknown types). + // + // Check BOTH ends of every pair, not just the sjLeft attribute. `isSameBaseRelation` proves + // the two sides are canonically equal, but `AttributeReference.canonicalized` rewrites the + // reference to drop metadata, so canonical equality does NOT prove the sjRight attribute + // carries the same CHAR/VARCHAR metadata that `isSafeComparisonGroupingAttribute` reads. Gate + // each side independently rather than assume they match. + val keyAttrs = + (equiPairs ++ neqPairs).flatMap { case (left, right) => Seq(left, right) } + if (!keyAttrs.forall(isSafeComparisonGroupingAttribute)) return None + + // Canonicalization intentionally erases cosmetic Alias names, so name equality cannot prove + // that the two predicate ends refer to the same underlying column. Resolve each end by its own + // ExprId against its child output and require matching output ordinals instead. + val equiValid = equiPairs.forall { + case (l, r) => sameOutputPosition(leftPlan, rightPlan, l, r) + } + val neqValid = neqPairs.forall { + case (l, r) => sameOutputPosition(leftPlan, rightPlan, l, r) + } + if (!equiValid || !neqValid) return None + + // Equi-key output positions must be distinct across pairs. Keep the same positional identity + // here so swapped or duplicate aliases cannot make two different underlying columns look equal. + val leftEquiOrdinals = equiPairs.map { case (l, _) => outputOrdinal(leftPlan, l) } + if (leftEquiOrdinals.exists(_ < 0)) return None + if (leftEquiOrdinals.distinct.size != leftEquiOrdinals.size) return None + + // Defensive: reject when the neq column overlaps an equi-key column + // (e.g. `t1.k = t2.k AND t1.k <> t2.k`). + val neqLeftOrdinal = outputOrdinal(leftPlan, neqPairs.head._1) + if (neqLeftOrdinal < 0 || leftEquiOrdinals.contains(neqLeftOrdinal)) return None + Some((equiPairs, neqPairs)) + } + + /** + * Attribute-aware gate applied to the equi keys and the neq column. A CHAR/VARCHAR column reaches + * the optimizer as StringType with its declared type recorded in the attribute metadata + * (CharVarcharUtils stamps it when the relation output is built), so checking `attr.dataType` + * alone would let it through the StringType branch of [[isSafeComparisonGroupingType]]. Recover + * the declared raw type from the metadata (falling back to `dataType` when there is no marker) + * and run it through the datatype allowlist, so CHAR/VARCHAR fail closed there in every config. + */ + private def isSafeComparisonGroupingAttribute(attr: Attribute): Boolean = { + val rawType = CharVarcharUtils.getRawType(attr.metadata).getOrElse(attr.dataType) + isSafeComparisonGroupingType(rawType) + } + + /** + * Types whose SQL comparison equality (`=` / `<>`) is provably identical to grouping / DISTINCT + * equality (and hashing). The rewrite turns `<>` into COUNT(DISTINCT) and `=` into GROUP BY, so + * it must never be applied to a type where those two notions of equality can diverge. This is a + * conservative positive allowlist: it admits only the types where the equivalence is well + * established and fails closed on everything else, including any new type a future version might + * add. + * + * Deliberately rejected: + * - Float / Double: this rewrite relies on comparison equality (`=` / `<>`) and + * grouping/DISTINCT equality having exactly the same contract. Signed zero and NaN are the + * representative edge cases that make that contract non-trivial for floating point, and + * whether the two paths agree rests on normalization details (e.g. NormalizeFloatingNumbers). + * Rather than depend on that, floats fail closed. Complex types are rejected wholesale below, + * which also covers floats nested inside a struct/array/map. + * - Complex types (Array / Struct / Map) and UDTs / Variant: their equality-vs-grouping + * equivalence is harder to prove and can itself embed floats. + * - Char / Varchar: CHAR / VARCHAR have dedicated declared-type semantics and are + * conservatively kept outside this rewrite. Table columns may reach the optimizer as + * annotated StringType, so the caller ([[isSafeComparisonGroupingAttribute]]) recovers their + * declared type from metadata first; this branch is the single rejection point for both that + * path and first-class CHAR/VARCHAR types. + * - Non-binary collated strings: comparison and grouping can route through different collation + * code paths, so only byte-wise `supportsBinaryEquality` strings are admitted. + */ + private def isSafeComparisonGroupingType(dt: DataType): Boolean = dt match { + case ByteType | ShortType | IntegerType | LongType => true + case _: DecimalType => true + case BooleanType => true + case DateType => true + case TimestampType | TimestampNTZType => true + case BinaryType => true + + case _: CharType | _: VarcharType => false + case st: StringType if st.supportsBinaryEquality => true + + case _ => false + } + + /** + * True iff `plan` produces the same row bag on every evaluation. + * + * This is the primary safety guard for the rewrite, which folds two occurrences of the same + * subtree into one aggregate -- sound only when both occurrences produce identical row bags. We + * check it at TWO levels: + * - candidate level: the enclosing subquery, before descending into the self-join. Catches + * nondeterminism that lives above the self-join rather than on either side -- e.g. a Pattern + * A2 outer join whose condition is `d.k = sj.k AND rand() < 0.5`. Without this, + * `isSameBaseRelation(sjLeft, sjRight)` could pass (both sides look deterministic) while the + * enclosing plan still contains `Rand`. + * - relation level: [[isSameBaseRelation]] additionally requires the two sides to be + * structurally identical. + * + * Attribute-level `plan.deterministic` alone is NOT sufficient. Catalyst's + * `Expression.deterministic` only checks explicit `Nondeterministic` annotation; several + * operators produce a runtime-nondeterministic row bag even though every expression they contain + * is `deterministic == true`: + * - `Aggregate` with `First` / `Last` / `collect_list` / `min_by` / `max_by` (tie order), + * - `Window` with `row_number()` / `rank()` over a non-total order, + * - `Limit` / `LocalLimit` / `Sample` / `Offset` (row-bag operator-level nondeterminism), + * - streaming sources. + * + * This rule collapses two evaluations of the same subtree into one aggregate; repeatability must + * be provable, not assumed. That is why both the operator check and the expression check below + * are WHITELISTS rather than blacklists -- unknown operators and unknown expression types default + * to reject. + * + * Expression support is allowlisted, not blacklisted. `plan.deterministic` relies on each + * expression's reported `deterministic` contract; that is necessary but insufficient for unknown + * expression types whose repeatability has not been established -- a builtin that carries hidden + * state yet reports `deterministic == true` would otherwise be trusted silently. For a rewrite + * that folds two evaluations of a subtree into one aggregate we prefer to miss an optimization + * than to misapply one, so new expression types are added to [[isRepeatableExpression]] only + * after their repeatability has been established. `plan.deterministic` is kept as a cheap + * fast-reject, but the expression allowlist is what actually proves repeatability. + * + * `plan.subqueriesAll.isEmpty` additionally fail-closes on any embedded expression subquery + * (scalar / IN / EXISTS). `plan.exists` in `isRowBagRepeatable` walks only the operator tree and + * does not descend into expression subqueries, and `plan.deterministic` does not prove a nested + * subquery is row-bag repeatable (e.g. an uncorrelated `LIMIT 1` without `ORDER BY`). Rejecting + * any embedded subquery keeps the repeatability proof confined to the operator whitelist below. + */ + private def isRepeatablePlan(plan: LogicalPlan): Boolean = { + // The operator/source whitelist is checked before the expression whitelist so that a plan whose + // operator is itself unknown -- e.g. Aggregate (carries AggregateExpression) or Window (carries + // WindowExpression / SortOrder) -- is attributed to isRowBagRepeatable rather than being masked + // by the fact that those operators also carry non-allowlisted expressions. + plan.deterministic && + !plan.isStreaming && + plan.subqueriesAll.isEmpty && + isRowBagRepeatable(plan) && + hasRepeatableExpressions(plan) + } + + /** + * Operator whitelist for `isRepeatablePlan`. A plan is row-bag repeatable only when every node is + * known to produce a repeatable output row bag from repeatable children. Unknown operators and + * unknown leaf sources fail closed. + * + * Kept intentionally narrow -- the target workload (Q95-shape self-join in a subquery) only needs + * a Parquet relation scan optionally wrapped in Project / Filter / SubqueryAlias plus the + * self-join itself. Range and LocalRelation are also trusted deterministic leaves. Adding an + * operator here requires proving that its output row values and multiplicities are repeatable. + * Pure row ordering is irrelevant to the row-bag contract, but operators not needed by the + * target shape remain fail-closed until explicitly reviewed. + * + * Arbitrary `LeafNode`s are intentionally not trusted. For example, `LogicalRDD` may wrap an + * arbitrary RDD lineage whose runtime behavior is invisible to Catalyst's `plan.deterministic`; + * `InMemoryRelation`, `DataSourceV2Relation` and custom leaves are likewise rejected until + * proven. Streaming sources reach here as `LeafNode`s but are already filtered upstream by + * `plan.isStreaming` in [[isRepeatablePlan]]. + * + * `LogicalRelation` is trusted only when its underlying relation is a `HadoopFsRelation` whose + * `fileFormat` is EXACTLY `ParquetFileFormat` (`getClass == classOf[ParquetFileFormat]`, not + * `isInstanceOf`). `HadoopFsRelation.fileFormat` can be any `FileFormat`, including custom + * formats whose scan is not provably a repeatable row bag; `ParquetFileFormat` is also non-final, + * so a third-party subclass could override its scan. The target workload only needs stock + * Parquet, so every other `FileFormat` -- subclasses of `ParquetFileFormat` included -- and every + * non-`HadoopFsRelation` fail closed. + * + * This helper checks operators and leaf sources only; expression-type repeatability is a separate + * concern handled by [[hasRepeatableExpressions]], and both are joined in [[isRepeatablePlan]]. + */ + private def isRowBagRepeatable(plan: LogicalPlan): Boolean = !plan.exists { + // TreeNode exposes `exists` but not `forall`, so invert: a whitelisted operator maps to `false` + // ("does not break repeatability") and everything else to `true`; negating the whole `exists` + // then means "every operator is whitelisted". + case _: Project => false + case _: Filter => false + case _: SubqueryAlias => false + // Join is included because our target pattern IS a Join; both children get recursed into. + case _: Join => false + // Explicitly trusted leaves. + case _: Range => false + case _: LocalRelation => false + case relation: LogicalRelation => + relation.relation match { + case h: HadoopFsRelation if h.fileFormat.getClass == classOf[ParquetFileFormat] => false Review Comment: **Finding 3.** A question, not a request to restructure. What does the format check buy? The property being proven is "two `sameResult` scans of this relation return the same row bag". I cannot see how a stock ORC, JSON or CSV `HadoopFsRelation` is weaker on that than Parquet. All of them go through the same `FileIndex` and the same `FileSourceScanExec`. Measured at this head, with the same schema, data and query and only the storage changed: fires on `parquet`, does not fire on `orc`, `json`, `csv`, or a `CACHE TABLE`d Parquet table. That is the boundary the source-boundary point at [r3939100186](https://github.com/apache/spark/pull/58424#discussion_r3939100186) asks you to document, but my question is about the check itself rather than its doc. The same question applies to `isRepeatableExpression` one layer down. It costs a hand-maintained list of about twenty expression types, so an ordinary `coalesce`, `case when`, `substring` or `abs` anywhere in the subquery turns the rewrite off. And it did not keep out the one repeatability hole found so far: `Cast` is on the list, and the string-to-timestamp path at [r3963602271](https://github.com/apache/spark/pull/58424#discussion_r3963602271) is exactly what the list was meant to catch. If the answer is "the risk is opaque leaves like `LogicalRDD`, not the file format", a leaf blocklist would say that directly and would let the rule work on the storage most deployments actually have. If the narrowness is deliberate for a first cut, saying so in the class doc would stop the next reader from taking the Parquet check for a correctness requirement. -- 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]
