Exchange reuse is not working in Spark 3.2.1

[zhouyuan]

Describe the bug
TPC-DS Q2, columnar exchange is not reused.

== Physical Plan ==
AdaptiveSparkPlan (95)
+- == Final Plan ==
   ArrowColumnarToRow (56)
   +- ColumnarSort (55)
      +- ColumnarCustomShuffleReader (54)
         +- ShuffleQueryStage (53)
            +- ColumnarExchange (52)
               +- * ColumnarConditionProject (50)
                  +- * ColumnarShuffledHashJoin Inner (49)
                     :- * ColumnarBroadcastHashJoin Inner (22)
                     :  :- ColumnarHashAggregate (17)
                     :  :  +- CoalesceBatches (16)
                     :  :     +- ColumnarCustomShuffleReader (15)
                     :  :        +- ShuffleQueryStage (14)
                     :  :           +- ColumnarExchange (13)
                     :  :              +- * ColumnarHashAggregate (11)
                     :  :                 +- * ColumnarBroadcastHashJoin Inner (10)
                     :  :                    :- ColumnarUnion (5)
                     :  :                    :  :- ColumnarConditionProject (2)
                     :  :                    :  :  +- Scan arrow arrow_part_tpcds_decimal_1000.web_sales (1)
                     :  :                    :  +- ColumnarConditionProject (4)
                     :  :                    :     +- Scan arrow arrow_part_tpcds_decimal_1000.catalog_sales (3)
                     :  :                    +- BroadcastQueryStage (9)
                     :  :                       +- ColumnarBroadcastExchange (8)
                     :  :                          +- ColumnarConditionProject (7)
                     :  :                             +- Scan arrow arrow_part_tpcds_decimal_1000.date_dim (6)
                     :  +- BroadcastQueryStage (21)
                     :     +- ColumnarBroadcastExchange (20)
                     :        +- ColumnarConditionProject (19)
                     :           +- Scan arrow arrow_part_tpcds_decimal_1000.date_dim (18)
                     +- CoalesceBatches (48)
                        +- ColumnarCustomShuffleReader (47)
                           +- ShuffleQueryStage (46)
                              +- ColumnarExchange (45)
                                 +- * ColumnarBroadcastHashJoin Inner (43)
                                    :- ColumnarHashAggregate (38)
                                    :  +- CoalesceBatches (37)
                                    :     +- ColumnarCustomShuffleReader (36)
                                    :        +- ShuffleQueryStage (35)
                                    :           +- ColumnarExchange (34)
                                    :              +- * ColumnarHashAggregate (32)
                                    :                 +- * ColumnarBroadcastHashJoin Inner (31)
                                    :                    :- ColumnarUnion (27)
                                    :                    :  :- ColumnarConditionProject (24)
                                    :                    :  :  +- Scan arrow arrow_part_tpcds_decimal_1000.web_sales (23)
                                    :                    :  +- ColumnarConditionProject (26)
                                    :                    :     +- Scan arrow arrow_part_tpcds_decimal_1000.catalog_sales (25)
                                    :                    +- ColumnarBroadcastExchange (30)
                                    :                       +- DataToArrowColumnar (29)
                                    :                          +- ReusedExchange (28)
                                    +- BroadcastQueryStage (42)
                                       +- ColumnarBroadcastExchange (41)
                                          +- ColumnarConditionProject (40)
                                             +- Scan arrow arrow_part_tpcds_decimal_1000.date_dim (39)

To Reproduce
Run TPC-DS Q2

Expected behavior
reuse the exchange between ColumnarExchange (13) and ColumnarExchange (34)

Additional context
Add any other context about the problem here.

[PHILO-HE]

As discussed, this issues may be related to broadcast exchange fallback to handle DPP issue. I will track it.