[GLUTEN-9034][VL] Fix the VeloxResizeBatch not add for ReusedExchange

[jinchengchenghh]

Before that, in TPCDS Q95, there are 4 AQEShuffleRead, but only one add the VeloxResizeBatch node.

After that,

The physical plan is

AdaptiveSparkPlan (162)
+- == Final Plan ==
   VeloxColumnarToRow (100)
   +- ^ ProjectExecTransformer (98)
      +- ^ RegularHashAggregateExecTransformer (97)
         +- ^ InputIteratorTransformer (96)
            +- VeloxResizeBatches (94)
               +- ShuffleQueryStage (93), Statistics(sizeInBytes=864.0 B, rowCount=36)
                  +- ColumnarExchange (92)
                     +- VeloxResizeBatches (91)
                        +- ^ FlushableHashAggregateExecTransformer (89)
                           +- ^ RegularHashAggregateExecTransformer (88)
                              +- ^ RegularHashAggregateExecTransformer (87)
                                 +- ^ ProjectExecTransformer (86)
                                    +- ^ BroadcastHashJoinExecTransformer Inner BuildRight (85)
                                       :- ^ ProjectExecTransformer (76)
                                       :  +- ^ BroadcastHashJoinExecTransformer Inner BuildRight (75)
                                       :     :- ^ ProjectExecTransformer (66)
                                       :     :  +- ^ BroadcastHashJoinExecTransformer Inner BuildRight (65)
                                       :     :     :- ^ ShuffledHashJoinExecTransformer LeftSemi BuildRight (56)
                                       :     :     :  :- ^ ShuffledHashJoinExecTransformer LeftSemi BuildRight (30)
                                       :     :     :  :  :- ^ InputIteratorTransformer (11)
                                       :     :     :  :  :  +- VeloxResizeBatches (9)
                                       :     :     :  :  :     +- AQEShuffleRead (8)
                                       :     :     :  :  :        +- ShuffleQueryStage (7), Statistics(sizeInBytes=2.4 GiB, rowCount=7.20E+7)
                                       :     :     :  :  :           +- ColumnarExchange (6)
                                       :     :     :  :  :              +- VeloxResizeBatches (5)
                                       :     :     :  :  :                 +- ^ ProjectExecTransformer (3)
                                       :     :     :  :  :                    +- ^ FilterExecTransformer (2)
                                       :     :     :  :  :                       +- ^ ScanTransformer parquet  (1)
                                       :     :     :  :  +- ^ ProjectExecTransformer (29)
                                       :     :     :  :     +- ^ ShuffledHashJoinExecTransformer Inner BuildRight (28)
                                       :     :     :  :        :- ^ InputIteratorTransformer (22)
                                       :     :     :  :        :  +- VeloxResizeBatches (20)
                                       :     :     :  :        :     +- AQEShuffleRead (19)
                                       :     :     :  :        :        +- ShuffleQueryStage (18), Statistics(sizeInBytes=823.8 MiB, rowCount=7.20E+7)
                                       :     :     :  :        :           +- ColumnarExchange (17)
                                       :     :     :  :        :              +- VeloxResizeBatches (16)
                                       :     :     :  :        :                 +- ^ ProjectExecTransformer (14)
                                       :     :     :  :        :                    +- ^ FilterExecTransformer (13)
                                       :     :     :  :        :                       +- ^ ScanTransformer parquet  (12)
                                       :     :     :  :        +- ^ InputIteratorTransformer (27)
                                       :     :     :  :           +- AQEShuffleRead (25)
                                       :     :     :  :              +- ShuffleQueryStage (24), Statistics(sizeInBytes=823.8 MiB, rowCount=7.20E+7)
                                       :     :     :  :                 +- ReusedExchange (23)
                                       :     :     :  +- ^ ProjectExecTransformer (55)
                                       :     :     :     +- ^ ShuffledHashJoinExecTransformer Inner BuildLeft (54)
                                       :     :     :        :- ^ InputIteratorTransformer (41)
                                       :     :     :        :  +- VeloxResizeBatches (39)
                                       :     :     :        :     +- AQEShuffleRead (38)
                                       :     :     :        :        +- ShuffleQueryStage (37), Statistics(sizeInBytes=54.9 MiB, rowCount=7.20E+6)
                                       :     :     :        :           +- ColumnarExchange (36)
                                       :     :     :        :              +- VeloxResizeBatches (35)
                                       :     :     :        :                 +- ^ ProjectExecTransformer (33)
                                       :     :     :        :                    +- ^ FilterExecTransformer (32)
                                       :     :     :        :                       +- ^ ScanTransformer parquet  (31)
                                       :     :     :        +- ^ ProjectExecTransformer (53)
                                       :     :     :           +- ^ ShuffledHashJoinExecTransformer Inner BuildRight (52)
                                       :     :     :              :- ^ InputIteratorTransformer (46)
                                       :     :     :              :  +- AQEShuffleRead (44)
                                       :     :     :              :     +- ShuffleQueryStage (43), Statistics(sizeInBytes=823.8 MiB, rowCount=7.20E+7)
                                       :     :     :              :        +- ReusedExchange (42)
                                       :     :     :              +- ^ InputIteratorTransformer (51)
                                       :     :     :                 +- AQEShuffleRead (49)
                                       :     :     :                    +- ShuffleQueryStage (48), Statistics(sizeInBytes=823.8 MiB, rowCount=7.20E+7)
                                       :     :     :                       +- ReusedExchange (47)
                                       :     :     +- ^ InputIteratorTransformer (64)
                                       :     :        +- BroadcastQueryStage (62), Statistics(sizeInBytes=287.0 B, rowCount=61)
                                       :     :           +- ColumnarBroadcastExchange (61)
                                       :     :              +- ^ ProjectExecTransformer (59)
                                       :     :                 +- ^ FilterExecTransformer (58)
                                       :     :                    +- ^ ScanTransformer parquet  (57)
                                       :     +- ^ InputIteratorTransformer (74)
                                       :        +- BroadcastQueryStage (72), Statistics(sizeInBytes=320.6 KiB, rowCount=7.94E+4)
                                       :           +- ColumnarBroadcastExchange (71)
                                       :              +- ^ ProjectExecTransformer (69)
                                       :                 +- ^ FilterExecTransformer (68)
                                       :                    +- ^ ScanTransformer parquet  (67)
                                       +- ^ InputIteratorTransformer (84)
                                          +- BroadcastQueryStage (82), Statistics(sizeInBytes=59.0 B, rowCount=4)
                                             +- ColumnarBroadcastExchange (81)
                                                +- ^ ProjectExecTransformer (79)
                                                   +- ^ FilterExecTransformer (78)
                                                      +- ^ ScanTransformer parquet  (77)

Related issue: #9034

[zhztheplayer]

It doesn't likely compile.

Let's use this shortcut instead: https://github.com/apache/spark/blob/9ff0fba06e758d2509dd8eb7a38b01cc8720d43d/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/QueryStageExec.scala#L203-L208

[zhztheplayer]

I found this comment:

// Since it's transformed in a bottom to up order, so we may first encounter
// ShuffeQueryStageExec, which is transformed to VeloxResizeBatchesExec(ShuffeQueryStageExec),
// then we see AQEShuffleReadExec

Maybe transformDown is more feasible in this case. cc @WangGuangxin

[jinchengchenghh]

It may caused by fix conflict, the old version is following, it can work well, I will fix it

_: ColumnarShuffleExchangeExec |
                ReusedExchangeExec(_, _: ColumnarShuffleExchangeExec)

[WangGuangxin]

Since the logic here is wrapping a new node VeloxResizeBatches when matched, if we use transformDown here, the logic here will be much more complicated .

For example, if it's transform down and when we match node a, it transform it to VeloxResizeBatches(a), then go down the tree node, it will encounter node a again, so we have to do something to avoid wrap VeloxResizeBatches again and again

[zhztheplayer]

Sounds reasonable. Thanks for the explanation.

[wForget]

Thanks @jinchengchenghh