hhr293 commented on code in PR #12756: URL: https://github.com/apache/gluten/pull/12756#discussion_r3823030189
########## backends-velox/src/test/scala/org/apache/gluten/extension/RewriteSelfJoinInequalityToAggregateSuite.scala: ########## @@ -0,0 +1,446 @@ +/* + * 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.gluten.extension + +import org.apache.gluten.execution.WholeStageTransformerSuite + +import org.apache.spark.SparkConf +import org.apache.spark.sql.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 + +/** + * Correctness tests for [[RewriteSelfJoinInequalityToAggregate]]. + * + * Assertions center on **result equivalence** between `rewrite=true` and `rewrite=false`. That is + * the direct check of the rule's semantic contract: the rewrite must not change what the query + * returns. Plan-shape assertions (e.g. "an Aggregate node exists") are avoided because downstream + * optimizer rules differ across Spark versions -- e.g. Spark 4.0's `RewritePredicateSubquery` and + * constant folding collapse LocalRelation-based EXISTS bodies so aggressively that our rule may + * never see the original self-join shape yet the final result is still correct. + * + * Where a plan-level signal is useful, we look for the alias name our rule injects + * (`_gluten_rw_selfjoin_cnt_distinct`) as a soft indicator that the rule fired. Its absence is not + * treated as a failure -- an equivalent result via a different path is still a pass. + */ +class RewriteSelfJoinInequalityToAggregateSuite extends WholeStageTransformerSuite { + + override protected val resourcePath: String = "/tpch-data-parquet" + override protected val fileFormat: String = "parquet" + + override protected def sparkConf: SparkConf = super.sparkConf + .set("spark.gluten.sql.rewrite.selfJoinInequality", "true") + .set(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key, "-1") + + /** Signature alias produced by the rewrite; presence => rule definitely fired. */ + private val CountDistinctAlias = "_gluten_rw_selfjoin_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 + }) + } + + /** Run `sql` twice, first with rewrite ON then OFF, and return the two result row sets. */ + private def runBoth(sql: String): (Set[Row], Set[Row]) = { + var on: Set[Row] = null + var off: Set[Row] = null + withSQLConf("spark.gluten.sql.rewrite.selfJoinInequality" -> "true") { + on = spark.sql(sql).collect().toSet + } + withSQLConf("spark.gluten.sql.rewrite.selfJoinInequality" -> "false") { + off = spark.sql(sql).collect().toSet + } + (on, off) + } + + private def setupTable(): Unit = { + // k=1: distinct v={10,20} -> matches (has 2 non-null distinct) + // k=2: distinct v={30} -> no match (only 1) + // k=3: distinct v={40,50,60} -> matches + // k=4: v={70, NULL} -> no match (only 1 non-null) + // k=5: v={NULL, NULL} -> no match (0 non-null) + // k=6: v={80, 90, NULL} -> matches + spark.sql( + """CREATE OR REPLACE TEMP VIEW T AS SELECT * FROM VALUES + | (1, 10), (1, 10), (1, 20), + | (2, 30), + | (3, 40), (3, 50), (3, 60), + | (4, 70), (4, CAST(NULL AS INT)), + | (5, CAST(NULL AS INT)), (5, CAST(NULL AS INT)), + | (6, 80), (6, 90), (6, CAST(NULL AS INT)) + |AS T(k, v)""".stripMargin) + } + + // ==================== Positive: rewrite is semantically equivalent ==================== + + test("Pattern A': EXISTS subquery with bare self-join produces equivalent results") { + setupTable() + val sql = + """SELECT k FROM T ws1 WHERE EXISTS ( + | SELECT 1 FROM T s WHERE s.k = ws1.k AND s.v <> ws1.v)""".stripMargin + val (on, off) = runBoth(sql) + assert(on == off, s"rewrite ON $on != OFF $off") + // Ground truth: only k in {1,3,6} have >=2 non-null distinct v. + assert(on == Set(Row(1), Row(3), Row(6)), s"expected {1,3,6}, got $on") + } + + test("Pattern A': InSubquery with bare self-join produces equivalent results") { + setupTable() + val sql = + """SELECT k FROM T ws1 WHERE k IN ( + | SELECT s1.k FROM T s1, T s2 + | WHERE s1.k = s2.k AND s1.v <> s2.v)""".stripMargin + val (on, off) = runBoth(sql) + assert(on == off, s"rewrite ON $on != OFF $off") + assert(on == Set(Row(1), Row(3), Row(6))) + } + + test("Pattern A2: self-join nested inside outer InnerJoin produces equivalent results") { + setupTable() + // Only k in {1,3,6} qualify from the self-join side; the outer InnerJoin with D + // (values {1,3,6}) intersects, so the final answer is again {1,3,6}. + spark.sql( + """CREATE OR REPLACE TEMP VIEW D AS SELECT * FROM VALUES + | (1), (3), (6) AS D(k)""".stripMargin) + val sql = + """SELECT k FROM T outer_t WHERE k IN ( + | SELECT d.k + | FROM D d, (SELECT s1.k FROM T s1, T s2 + | WHERE s1.k = s2.k AND s1.v <> s2.v) sj + | WHERE d.k = sj.k)""".stripMargin + val (on, off) = runBoth(sql) + assert(on == off, s"Pattern A2 rewrite ON $on != OFF $off") + assert(on == Set(Row(1), Row(3), Row(6))) + } + + // ==================== Semantic parity on NULL / 3VL ==================== + + test("NULL / 3VL: rows with only-NULL or single-non-null inequality column are excluded") { + setupTable() + val sql = + """SELECT k FROM T ws1 WHERE EXISTS ( + | SELECT 1 FROM T s WHERE s.k = ws1.k AND s.v <> ws1.v)""".stripMargin + val (on, off) = runBoth(sql) + // k=4 (v={70,NULL}) fails: <> with NULL is UNKNOWN -> filtered by WHERE. + // k=5 (v={NULL,NULL}) fails: every <> is UNKNOWN. + assert(on == Set(Row(1), Row(3), Row(6)), s"expected {1,3,6}, got $on") + assert(off == on, s"NULL/3VL semantics diverge between rewrite ON and OFF: $on vs $off") + } + + // ==================== Negative: rewrite must produce equivalent results (or bail) ========== + + test("Plain InnerJoin at top level: results unchanged (rewrite must not touch it)") { + setupTable() + val sql = + """SELECT ws1.k FROM T ws1 JOIN T ws2 + |ON ws1.k = ws2.k AND ws1.v <> ws2.v""".stripMargin + // Row-multiplicity matters here; using count() to catch any drop or dup. + var onCount: Long = -1L + var offCount: Long = -1L + withSQLConf("spark.gluten.sql.rewrite.selfJoinInequality" -> "true") { + onCount = spark.sql(sql).count() + } + withSQLConf("spark.gluten.sql.rewrite.selfJoinInequality" -> "false") { + offCount = spark.sql(sql).count() + } + assert( + onCount == offCount, + s"plain InnerJoin row-count differs: rewrite=$onCount vs baseline=$offCount") + } + + test("IS DISTINCT FROM: NULL-safe inequality preserves original semantics") { + setupTable() + // IS DISTINCT FROM treats NULL as distinguishable (NULL IS DISTINCT FROM x = TRUE, + // NULL IS DISTINCT FROM NULL = FALSE). Our rewrite must NOT fold this into + // COUNT(DISTINCT), because COUNT(DISTINCT) ignores NULL. + val sql = + """SELECT k FROM T ws1 WHERE EXISTS ( + | SELECT 1 FROM T s WHERE s.k = ws1.k + | AND (s.v IS DISTINCT FROM ws1.v))""".stripMargin + val (on, off) = runBoth(sql) + assert(on == off, s"IS DISTINCT FROM semantics diverge: ON=$on OFF=$off") + // Sanity check: k=4 has (70, NULL) -- pair (v=70, v=NULL) IS DISTINCT FROM => TRUE + // so k=4 must be included (unlike the plain-neq case above where it's excluded). + assert(on.contains(Row(4)), s"k=4 should be in IS DISTINCT FROM result: $on") + // And rule fire signal must be absent: this is a rejection path. + val plan = spark.sql(sql).queryExecution.optimizedPlan + assert(!ruleFired(plan), s"rule must not fire on IS DISTINCT FROM:\n$plan") Review Comment: I'll rework the negative cases so they are uncorrelated and contain an actual self-join, ensuring each test reaches the guard it is intended to validate. I'll also add assert(ruleFired(plan)) to the positive A'/A2 cases, and add a NULL equi-key + NOT IN regression for the IsNotNull(equiKeys) protection. -- 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]
