Contain SortShuffleManager with ColumnarShuffleManager

[acvictor]

We would like to set the default shuffle manager in Spark as ColumnarShuffleManager. To avoid missing any updates from Spark that would go to SortShuffleManager, is it possible to contain SortShuffleManager within ColumnarShuffleManager, handle columnar shuffle specific cases in ColumnarShuffleManager and call corresponding SortShuffleManager's functions for all other cases? For example

override def registerShuffle[K, V, C](
      shuffleId: Int,
      dependency: ShuffleDependency[K, V, C]): ShuffleHandle = {
    if (dependency.isInstanceOf[ColumnarShuffleDependency[_, _, _]]) {
      logInfo(s"Registering ColumnarShuffle shuffleId: $shuffleId")
      new ColumnarShuffleHandle[K, V](
        shuffleId,
        dependency.asInstanceOf[ColumnarShuffleDependency[K, V, V]])
    } else {
      SortShuffleManager.registerShuffle(shuffleId, dependency)
    }
  }

Is there any specific reason the code was duplicated in ColumnarShuffleManager?

cc @rui-mo @zhli1142015 @PHILO-HE

[rui-mo]

cc: @marin-ma

[acvictor]

@marin-ma do you have any suggestions/insights?

[marin-ma]

@acvictor Thanks for your proposal. I have noticed similar designs in other Spark shuffle plugin projects. However, we haven't evaluated pros/cons for different designs.

To avoid missing any updates from Spark that would go to SortShuffleManager

I agree. Would you be willing to address this improvement?

[acvictor]

I will take this up. Thank you!

[acvictor]

#6022

[zhztheplayer]

+1 for the proposal.