[GLUTEN-10933][VL] Introduce GPU ShuffleWriterType kGpuHashShuffle

[jinchengchenghh]

Related issue: #10933

[github-actions]

Run Gluten Clickhouse CI on x86

[marin-ma]

LGTM. Just one more question: how do we currently test gpu_hash shuffle?

[github-actions]

Run Gluten Clickhouse CI on x86

[jinchengchenghh]

Now it is tested on my local environment, because it relies on GPU, I test TPCDS Q95 for previous PR, this is just a refactor, I don't test it. I will add cpp test for GPU shuffle reader.

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[zhztheplayer]

BTW We have the same parameter enableCudf for Velox plan execution (See NativePlanEvaluator.java). I was hoping we could finally pass these parameters from Java to C++ using configuration options. #9785

[jinchengchenghh]

Yes, use pair parameter name enableCudf would help understand the logic. What do you think about that? @marin-ma
After resolve #9785, the logic could be simple.

[marin-ma]

@jinchengchenghh I think for shuffle, we still can stick with adding different writer type. Again the shuffle writer type is a match for different shuffle reader deserialisation, and using different writer type it's easy to distinguish them from avoiding introducing params like "isRssShuffle", "isGpuShuffle", etc.

Moreover, ShuffleWriterType.name is the display string on Spark UI. We can easily tell whether the shuffle is gpu shuffle with this change.

[github-actions]

Run Gluten Clickhouse CI on x86

[jinchengchenghh]

I find I also need to update shuffle writer, but shuffleWriterType does not send to native, do you have any suggestion? @marin-ma

          nativeShuffleWriter = if (isSort) {
            shuffleWriterJniWrapper.createSortShuffleWriter(
              numPartitions,
              dep.nativePartitioning.getShortName,
              GlutenShuffleUtils.getStartPartitionId(
                dep.nativePartitioning,
                taskContext.partitionId),
              conf.get(SHUFFLE_DISK_WRITE_BUFFER_SIZE).toInt,
              conf.get(SHUFFLE_SORT_INIT_BUFFER_SIZE).toInt,
              conf.get(SHUFFLE_SORT_USE_RADIXSORT),
              partitionWriterHandle
            )
          } else {
            shuffleWriterJniWrapper.createHashShuffleWriter(
              numPartitions,
              dep.nativePartitioning.getShortName,
              GlutenShuffleUtils.getStartPartitionId(
                dep.nativePartitioning,
                taskContext.partitionId),
              nativeBufferSize,
              reallocThreshold,
              partitionWriterHandle
            )
          }

[marin-ma]

@jinchengchenghh Do you have the code for the changes in shuffle writer? I don't see any modifications to the existing shuffle writer.

[jinchengchenghh]

In the followup PR, Velox bool is bit, but cudf is BOOL8, it is byte

[github-actions]

Run Gluten Clickhouse CI on x86

[marin-ma]

@jinchengchenghh Let's discuss and address that in your next pr. Based on the discussion offline seems like there are many discrepancies between cudf and the current hash shuffle writer implementation.

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86