[SPARK-28156][SQL] Self-join should not miss cached view

[viirya]

What changes were proposed in this pull request?

The issue is when self-join a cached view, only one side of join uses cached relation. The cause is in ResolveReferences we do deduplicate for a view to have new output attributes. Then in AliasViewChild, the rule adds extra project under a view. So it breaks cache matching.

The fix is when dedup, we only dedup a view which has output different to its child plan. Otherwise, we dedup on the view's child plan.

val df = Seq.tabulate(5) { x => (x, x + 1, x + 2, x + 3) }.toDF("a", "b", "c", "d")
df.write.mode("overwrite").format("orc").saveAsTable("table1")

sql("drop view if exists table1_vw")
sql("create view table1_vw as select * from table1")

val cachedView = sql("select a, b, c, d from table1_vw")

cachedView.createOrReplaceTempView("cachedview")
cachedView.persist()

val queryDf = sql(
  s"""select leftside.a, leftside.b
      |from cachedview leftside
      |join cachedview rightside
      |on leftside.a = rightside.a
    """.stripMargin)

Query plan before this PR:

== Physical Plan ==                                                                                                                                    
*(2) Project [a#12664, b#12665]                                                                                                                        
+- *(2) BroadcastHashJoin [a#12664], [a#12660], Inner, BuildRight                                                                                      
   :- *(2) Filter isnotnull(a#12664)                                                                                                                   
   :  +- *(2) InMemoryTableScan [a#12664, b#12665], [isnotnull(a#12664)]                                                                               
   :        +- InMemoryRelation [a#12664, b#12665, c#12666, d#12667], StorageLevel(disk, memory, deserialized, 1 replicas)                             
   :              +- *(1) FileScan orc default.table1[a#12660,b#12661,c#12662,d#12663] Batched: true, DataFilters: [], Format: ORC, Location: InMemoryF
ileIndex[file:/Users/viirya/repos/spark-1/sql/core/spark-warehouse/org.apache.spark.sql...., PartitionFilters: [], PushedFilters: [], ReadSchema: struc
t<a:int,b:int,c:int,d:int>                                                                                                                             
   +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, true] as bigint)))                                                         
      +- *(1) Project [a#12660]                                                                                                                        
         +- *(1) Filter isnotnull(a#12660)                                                                                                             
            +- *(1) FileScan orc default.table1[a#12660] Batched: true, DataFilters: [isnotnull(a#12660)], Format: ORC, Location: InMemoryFileIndex[fil
e:/Users/viirya/repos/spark-1/sql/core/spark-warehouse/org.apache.spark.sql...., PartitionFilters: [], PushedFilters: [IsNotNull(a)], ReadSchema: struc
t<a:int>       

Query plan after this PR:

== Physical Plan ==
*(2) Project [a#12664, b#12665]
+- *(2) BroadcastHashJoin [a#12664], [a#12692], Inner, BuildRight
   :- *(2) Filter isnotnull(a#12664)
   :  +- *(2) InMemoryTableScan [a#12664, b#12665], [isnotnull(a#12664)]
   :        +- InMemoryRelation [a#12664, b#12665, c#12666, d#12667], StorageLevel(disk, memory, deserialized, 1 replicas)                            
   :              +- *(1) FileScan orc default.table1[a#12660,b#12661,c#12662,d#12663] Batched: true, DataFilters: [], Format: ORC, Location: InMemoryFileIndex[file:/Users/viirya/repos/spark-1/sql/core/spark-warehouse/org.apache.spark.sql...., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<a:int,b:int,c:int,d:int>
   +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)))                                                       
      +- *(1) Filter isnotnull(a#12692)
         +- *(1) InMemoryTableScan [a#12692], [isnotnull(a#12692)]
               +- InMemoryRelation [a#12692, b#12693, c#12694, d#12695], StorageLevel(disk, memory, deserialized, 1 replicas)                         
                     +- *(1) FileScan orc default.table1[a#12660,b#12661,c#12662,d#12663] Batched: true, DataFilters: [], Format: ORC, Location: InMemoryFileIndex[file:/Users/viirya/repos/spark-1/sql/core/spark-warehouse/org.apache.spark.sql...., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<a:int,b:int,c:int,d:int>

How was this patch tested?

Added test.

[SparkQA]

Test build #106868 has finished for PR 24960 at commit 81cbabc.

• This patch fails due to an unknown error code, -9.
• This patch merges cleanly.
• This patch adds no public classes.

[viirya]

retest this please.

[SparkQA]

Test build #106883 has finished for PR 24960 at commit 81cbabc.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[viirya]

cc @cloud-fan @dongjoon-hyun

[cloud-fan]

How about we add a Project above View to do dedup? And make View not extend MultiInstanceRelation.

[viirya]

In dedupRight, we don't add a Project that can break caching like this.

[cloud-fan]

Then we should revisit #24236 which adds Project to do dedup for Union. I don't know if there are more places like this.

The cache manager uses QueryPlan.sameResult to do cache look up, and it looks weird if an extra no-op project breaks QueryPlan.sameResult. Shall we update Project.canonicalized to remove no-op project node?

[viirya]

A project node just aliasing child's output in same output order, looks like can be a candidate to be removed in canonicalized query plan. But we should be careful about QueryPlan.sameResult. Not sure if any corner cases I'm not aware of.

If removing a project node like that is safe when canonicalization, #24236 and this issue should be fine.

[viirya]

I can do Project.canonicalized here and see jenkins result first? WDYT?

[viirya]

I read QueryPlan.doCanonicalize and rethink project canonicalization. I'm wondering if we can drop a project node like that, safely. For an intermediate project node, looks like it causes some problems in canonicalization of above nodes on it. Looks like it's not safe to remove a project node?

[cloud-fan]

Another idea: shall we apply AliasViewChild right before EliminateView?

[viirya]

Does it mean moving AliasViewChild to "Finish Analysis" in optimizer? I guess that makes sense. If moving AliasViewChild wouldn't bring negative impact, I think it is fine.

[cloud-fan]

I think we can even merge AliasViewChild and EliminateView, so that we only have one view related rule in "Finish Analysis".

[viirya]

Deferring AliasViewChild to optimizer, could result fail in CheckAnalysis, because non empty missingInput of View operator.

Shall we make View free of the check? Sounds not so good.

[cloud-fan]

Can we make View a leaf node instead of unary node?

[viirya]

It sounds not safe but I did try it locally. A rule, e.g., SubqueryAlias, needs to transform children of View. Making it as leaf node can bring unexpected error, if we forget to take care View in a rule like that.

[cloud-fan]

Deferring AliasViewChild to optimizer, could result fail in CheckAnalysis, because non empty missingInput of View operator.

This is easy to fix, just overwrite references in View.

[viirya]

Ok. Let's try this approach and see Jenkins results. Thanks.

[maropu]

(This might be off-topic and this is just a question though...) btw, any reason not to use optimized plans instead of analyzed plans for plan caches? Just to avoid overheads? In case that we do so, it seems self-joins in the description uses the cached view;

== Optimized Logical Plan ==
Project [a#37, b#38]
+- Join Inner, (a#37 = a#73)
   :- Project [a#37, b#38]
   :  +- Filter isnotnull(a#37)
   :     +- InMemoryRelation [a#37, b#38, c#39, d#40], StorageLevel(disk, memory, deserialized, 1 replicas)
   :           +- *(1) FileScan orc default.table1[a#37,b#38,c#39,d#40] Batched: true, DataFilters: [], Format: ORC, Location: InMemoryFileIndex[file:/Users/maropu/Repositories/spark/spark-master/spark-warehouse/table1], PartitionFilters: [], PushedFilters: [], ReadSchema: struct<a:int,b:int,c:int,d:int>
   +- Project [a#37 AS a#73]
      +- Filter isnotnull(a#37)
         +- InMemoryRelation [a#37, b#38, c#39, d#40], StorageLevel(disk, memory, deserialized, 1 replicas)
               +- *(1) FileScan orc default.table1[a#37,b#38,c#39,d#40] Batched: true, DataFilters: [], Format: ORC, Location: InMemoryFileIndex[file:/Users/maropu/Repositories/spark/spark-master/spark-warehouse/table1], PartitionFilters: [], PushedFilters: [], ReadSchema: struct<a:int,b:int,c:int,d:int>

[SparkQA]

Test build #107097 has finished for PR 24960 at commit ab3c6ce.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[maropu]

retest this please

[SparkQA]

Test build #107100 has finished for PR 24960 at commit ab3c6ce.

• This patch fails due to an unknown error code, -9.
• This patch merges cleanly.
• This patch adds no public classes.

[cloud-fan]

retest this please

[SparkQA]

Test build #107108 has finished for PR 24960 at commit ab3c6ce.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[cloud-fan]

We still need a rule in analyzer to make sure the view is valid... We can put it in CheckAnalysis

[viirya]

Yeah, I noticed that. Will add it later.

[viirya]

Basically, I moved analysis exception check from AliasViewChild. In EliminateView rule, it doesn't need to check this.

[SparkQA]

Test build #107122 has finished for PR 24960 at commit ca94e9c.

• This patch fails SparkR unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #107123 has finished for PR 24960 at commit 6a0e7d6.

• This patch fails SparkR unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[viirya]

oh, the CRAN problem happens again...

* checking CRAN incoming feasibility ...Error in .check_package_CRAN_incoming(pkgdir) : 
  dims [product 26] do not match the length of object [0]

I will contact CRAN sysadmin to solve it.

[viirya]

retest this please.

[SparkQA]

Test build #107155 has finished for PR 24960 at commit 6a0e7d6.

• This patch fails due to an unknown error code, -9.
• This patch merges cleanly.
• This patch adds no public classes.

[viirya]

retest this please

[viirya]

@maropu from my understanding, optimization can change query plan structure largely. A cached query fragment can be transformed to very different form and miss the cache.

[SparkQA]

Test build #107167 has finished for PR 24960 at commit 6a0e7d6.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[cloud-fan]

thanks, merging to master!

[viirya]

Thanks @cloud-fan

I also visited #24236 as you mentioned. I think it is fine from missing caches. Because when you cache an union with duplicate children, you actually cache the deduplicated plan processed by #24236. In later you have this query fragment in other queries, the deduplicated union will match cached version.

[maropu]

@viirya ur, I see. thanks for your kind explanation.