(spark) branch master updated: [SPARK-47633][SQL] Include right-side plan output in `LateralJoin#allAttributes` for more consistent canonicalization

[we...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

wenchen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/master by this push:
     new eba6364dbb36 [SPARK-47633][SQL] Include right-side plan output in `LateralJoin#allAttributes` for more consistent canonicalization
eba6364dbb36 is described below

commit eba6364dbb36b7009ff79a3cd8638aae6eec280f
Author: Bruce Robbins <be...@gmail.com>
AuthorDate: Tue Apr 23 14:09:01 2024 +0800

    [SPARK-47633][SQL] Include right-side plan output in `LateralJoin#allAttributes` for more consistent canonicalization
    
    ### What changes were proposed in this pull request?
    
    Modify `LateralJoin` to include right-side plan output in `allAttributes`.
    
    ### Why are the changes needed?
    
    In the following example, the view v1 is cached, but a query of v1 does not use the cache:
    ```
    CREATE or REPLACE TEMP VIEW t1(c1, c2) AS VALUES (0, 1), (1, 2);
    CREATE or REPLACE TEMP VIEW t2(c1, c2) AS VALUES (0, 1), (1, 2);
    
    create or replace temp view v1 as
    select *
    from t1
    join lateral (
      select c1 as a, c2 as b
      from t2)
    on c1 = a;
    
    cache table v1;
    
    explain select * from v1;
    == Physical Plan ==
    AdaptiveSparkPlan isFinalPlan=false
    +- BroadcastHashJoin [c1#180], [a#173], Inner, BuildRight, false
       :- LocalTableScan [c1#180, c2#181]
       +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)),false), [plan_id=113]
          +- LocalTableScan [a#173, b#174]
    ```
    
    The canonicalized version of the `LateralJoin` node is not consistent when there is a join condition. For example, for the above query, the join condition is canonicalized as follows:
    ```
    Before canonicalization: Some((c1#174 = a#167))
    After canonicalization:  Some((none#0 = none#167))
    ```
    You can see that the `exprId` for the second operand of `EqualTo` is not normalized (it remains 167). That's because the attribute `a` from the right-side plan is not included `allAttributes`.
    
    This PR adds right-side attributes to `allAttributes` so that references to right-side attributes in the join condition are normalized during canonicalization.
    
    ### Does this PR introduce _any_ user-facing change?
    
    No.
    
    ### How was this patch tested?
    
    New test.
    
    ### Was this patch authored or co-authored using generative AI tooling?
    
    No.
    
    Closes #45763 from bersprockets/lj_canonical_issue.
    
    Authored-by: Bruce Robbins <be...@gmail.com>
    Signed-off-by: Wenchen Fan <we...@databricks.com>
---
 .../plans/logical/basicLogicalOperators.scala         |  2 ++
 .../scala/org/apache/spark/sql/CachedTableSuite.scala | 19 +++++++++++++++++++
 2 files changed, 21 insertions(+)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
index 7c36e3bc79af..4fd640afe3b2 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
@@ -2057,6 +2057,8 @@ case class LateralJoin(
     joinType: JoinType,
     condition: Option[Expression]) extends UnaryNode {
 
+  override lazy val allAttributes: AttributeSeq = left.output ++ right.plan.output
+
   require(Seq(Inner, LeftOuter, Cross).contains(joinType),
     s"Unsupported lateral join type $joinType")
 
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala
index 7865e7f1f864..0ad9ceefc419 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala
@@ -1770,4 +1770,23 @@ class CachedTableSuite extends QueryTest with SQLTestUtils
       withSQLConf(SQLConf.DEFAULT_CACHE_STORAGE_LEVEL.key -> "DISK") {}
     }
   }
+
+  test("SPARK-47633: Cache hit for lateral join with join condition") {
+    withTempView("t", "q1") {
+      sql("create or replace temp view t(c1, c2) as values (0, 1), (1, 2)")
+      val query = """select *
+                    |from t
+                    |join lateral (
+                    |  select c1 as a, c2 as b
+                    |  from t)
+                    |on c1 = a;
+                    |""".stripMargin
+      sql(s"cache table q1 as $query")
+      val df = sql(query)
+      checkAnswer(df,
+        Row(0, 1, 0, 1) :: Row(1, 2, 1, 2) :: Nil)
+      assert(getNumInMemoryRelations(df) == 1)
+    }
+
+  }
 }


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org