[SPARK-32765][SQL] EliminateJoinToEmptyRelation should respect exchange behavior when canChangeNumPartitions == false

[leanken-zz]

What changes were proposed in this pull request?

Currently, EliminateJoinToEmptyRelation Rule will convert Join into EmptyRelation in some cases with AQE on. But if streamedSide of Join is a ShuffleQueryStage(canChangeNumPartitions == false), which means the Exchange produced by repartition Or singlePartition, in this case, if we were to convert it into an EmptyRelation, it will lost user specified number partition information for downstream operator, it's not right.

Why are the changes needed?

NumPartition info incorrect when streamedSide is a repartition plan or SinglePartition plan.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Added case in AdaptiveQueryExecSuite.

[leanken-zz]

After some code investigation and manual test, found that Optimizer will put Repartition after LeftAnti & LeftSemi Join with Some rules. So we can see that in UT, LeftAnti & LeftSemi could still be converted from Join to EmptyRelation.

FYI, @cloud-fan

NotInSubQuery NAAJ streamedSide repartition

=== Applying Rule org.apache.spark.sql.catalyst.optimizer.RewritePredicateSubquery ===
Repartition 77, true                          
 +- Project [a#218]                           
!   +- Filter NOT a#218 IN (list#230 [])      
!      :  +- Project [c#220]                  
!      :     +- LocalRelation [c#220, d#221]  
!      +- LocalRelation [a#218, b#219]        

Repartition 77, true
 +- Project [a#218]
    +- Join LeftAnti, ((a#218 = c#220) OR isnull((a#218 = c#220)))
       :- LocalRelation [a#218, b#219]
       +- Project [c#220]
          +- LocalRelation [c#220, d#221]

hand-rewritten NAAJ streamedSide repartition

=== Applying Rule org.apache.spark.sql.catalyst.optimizer.PushDownLeftSemiAntiJoin ===
Project [a#218]                                                  
!+- Join LeftAnti, ((a#218 = c#220) OR isnull((a#218 = c#220)))  
!   :- Repartition 77, true                                      
!   :  +- Project [a#218, b#219]                                 
!   :     +- LocalRelation [a#218, b#219]                        
!   +- Project [c#220, d#221]                                    
!      +- LocalRelation [c#220, d#221]                           

Project [a#218]
 +- Repartition 77, true
    +- Project [a#218, b#219]
       +- Join LeftAnti, ((a#218 = c#220) OR isnull((a#218 = c#220)))
          :- LocalRelation [a#218, b#219]
          +- Project [c#220, d#221]
             +- LocalRelation [c#220, d#221]

LEFT SEMI streamedSide repartition

=== Applying Rule org.apache.spark.sql.catalyst.optimizer.PushDownLeftSemiAntiJoin ===
+- Project [a#218]
!   +- Join LeftSemi, (a#218 = c#220)
!      :- Repartition 77, true
!      :  +- Project [a#218, b#219]
!      :     +- LocalRelation [a#218, b#219]
!      +- Project [c#220, d#221]
!         +- Project [c#220, d#221]
!            +- Filter (cast(d#221 as decimal(4,1)) > cast(cast(100 as decimal(3,0)) as decimal(4,1)))
!               +- LocalRelation [c#220, d#221]

+- Project [a#218]
    +- Repartition 77, true
       +- Project [a#218, b#219]
          +- Join LeftSemi, (a#218 = c#220)
             :- LocalRelation [a#218, b#219]
             +- Project [c#220, d#221]
                +- Project [c#220, d#221]
                   +- Filter (cast(d#221 as decimal(4,1)) > cast(cast(100 as decimal(3,0)) as decimal(4,1)))
                      +- LocalRelation [c#220, d#221]

[cloud-fan]

ok to test

[cloud-fan]

add to whitelist

[SparkQA]

Test build #128183 has finished for PR 29614 at commit f21fd5c.

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

[SparkQA]

Test build #128189 has finished for PR 29614 at commit b0a40fa.

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

[leanken-zz]

retest this please

[SparkQA]

Test build #128192 has finished for PR 29614 at commit b68533a.

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

[SparkQA]

Test build #128202 has finished for PR 29614 at commit c67477a.

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

[SparkQA]

Test build #128209 has finished for PR 29614 at commit 82f6cc6.

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

[maropu]

nit: can we use collectFirst instead?

   val immutablePartitionStageExists = streamedPlan.collectFirst {
      case LogicalQueryStage(_, physicalPlan: SparkPlan)
        if physicalPlan.collectFirst {
          case s: ShuffleQueryStageExec if !s.shuffle.canChangeNumPartitions => s
        }.nonEmpty => true
    }.isDefined

[maropu]

nit: canChangeNumPartitions== false => canChangeNumPartitions == false

[maropu]

nit format:

    !immutablePartitionStageExists && {
      buildPlan match {
        case LogicalQueryStage(_, stage: BroadcastQueryStageExec)
          if stage.resultOption.get().isDefined
            && stage.broadcast.relationFuture.get().value == relation => true
        case _ => false
      }
    }

?

[maropu]

We already exclude this in SharedSparkSession?

spark/sql/core/src/test/scala/org/apache/spark/sql/test/SharedSparkSession.scala

Lines 70 to 74 in 95f1e95

	// Disable ConvertToLocalRelation for better test coverage. Test cases built on
	// LocalRelation will exercise the optimization rules better by disabling it as
	// this rule may potentially block testing of other optimization rules such as
	// ConstantPropagation etc.
	.set(SQLConf.OPTIMIZER_EXCLUDED_RULES.key, ConvertToLocalRelation.ruleName)

[cloud-fan]

Actually PropagateEmptyRelation doesn't respect repartition in join stream side either, maybe it's also fine in AQE.

[leanken-zz]

@cloud-fan @maropu If this is no longer considered as a bug, then I will close this PR and JIRA. Is that OK?