[NSE-461]columnar shuffle support for ArrayType

[xuechendi]

Fixed: #461

[github-actions]

#461

[xuechendi]

Now codes only support when compression is disabled.

spark = SparkSession.builder.master('local[*]')\
    .appName("udf_column")\
    .config("spark.sql.broadcastTimeout", "7200")\
    .config("spark.cleaner.periodicGC.interval", "10min")\
    .config("spark.driver.extraClassPath",
            f"{native_sql_path}:{native_arrow_datasource_path}")\
    .config("spark.sql.extensions", "com.intel.oap.ColumnarPlugin, com.intel.oap.spark.sql.ArrowWriteExtension")\
    .config("spark.shuffle.manager", "org.apache.spark.shuffle.sort.ColumnarShuffleManager")\
    .config("spark.driver.memory", "200G")\
    .config("spark.driver.memoryOverhead", "300G")\
    .config("spark.memory.offHeap.size", "300G")\
    .config("spark.oap.sql.columnar.arrowudf", "true")\
    .getOrCreate()
    #.config("spark.shuffle.compress", "false")\

df = spark.read.format("arrow").load(path_prefix + original_folder)
df = df.filter("engaged_with_user_id is not null")
df = df.groupby('engaged_with_user_id').agg(f.collect_list('present_media').alias('posted_tweet_types'), f.first('language').alias('language'))
df = df.repartition(10)
df.show(truncate=False)