[VL] Parquet writer rowgroup number is bigger than Spark using the same parameter

[FelixYBW]

Description

Using default parameters, Gluten parquet writer creates more rowgroup than Spark, which leads to much worse perf than using Spark files

df.repartition(tbl_filenum[tbl]['part_cnt'], tbl_filenum[tbl]['part']).write.mode("overwrite").format("parquet").option("compression", "zstd").partitionBy(tbl_filenum[tbl]['part']).option('fs.s3a.committer.name', 'magic').save(f"s3a://presto-workload/{databasename}/{tbl}")

The table shows one stage of TPCDS q9 when reading spark and Gluten generated store_sales table

Metric	spark generated	gluten generated	Difference
Time of scan and filter	18.8 min	1.31 hours	4.2x slower
I/O wait time	2.09 hours	10.09 hours	4.8x slower
Time of scan	1.85 hours	8.84 hours	4.8x slower
Page load time	14.0 min	1.25 hours	5.4x slower
Data source read time	4.8 min	21.6 min	4.5x slower
Storage read bytes	31.5 GiB	34.7 GiB	10% more data read
Size of files read	240.7 GiB	256.6 GiB	6.6% larger
Peak memory	16.0 GiB	14.1 GiB	12% less memory used
Row groups processed	2,666	8,727	3.3x more row groups
Memory allocations	17,584,444	14,466,793	18% fewer allocations
Preloaded splits	1,585	1,710	8% more preloaded

Gluten version

None

[FelixYBW]

block.size and block.rows don't take effect

[FelixYBW]

both

df.write.option("block.size", 512*1024*1024)\
            .option("block.rows", 400000000)

or

sc.conf.set("spark.sql.parquet.block.size", 256*1024*1024)
sc.conf.set("spark.sql.parquet.block.rows", 400000000)

doesn't pass the config to parquet writer

[FelixYBW]

both

df.write.option("block.size", 512*1024*1024)\
            .option("block.rows", 400000000)

or

sc.conf.set("spark.sql.parquet.block.size", 256*1024*1024)
sc.conf.set("spark.sql.parquet.block.rows", 400000000)

doesn't pass the config to parquet writer

df.write.option("parquet.block.size", 512*1024*1024)\
            .option("parquet.block.rows", 400000000)

can pass to velox.

sc.conf.set("spark.gluten.sql.parquet.block.size", 256*1024*1024)
sc.conf.set("spark.gluten.sql.parquet.block.rows", 400000000)

doesn't work

[FelixYBW]

here are some tests. We need to set block.size to 500M to get the same rowgroup number as spark.

sql is part of Q9:

select count(*)
                 from store_sales
                 where ss_quantity between 1 and 20)

Test	zstd.level	block.size	block.rows	Elapsed Time
Test 1	0	128m	100m	40s
Test 2	3	128m	100m	35s
Test 3	3	256m	100m	25s
Test 4	3	512m	100m	16s
Test 5	3	512m	400m	17s
Test 6	3	512m	800m	17s

Metric	Test 1	Test 2	Test 3	Test 4	Test 5	Test 6
Elapsed Time	40s	35s	25s	16s	17s	17s
Scan & Filter Time	22.6m	22.2m	15.8m	9.5m	9.4m	9.6m
IO Wait Time	2.91h	2.90h	1.99h	1.18h	1.18h	1.20h
Scan Time	2.57h	2.56h	1.76h	1.05h	1.06h	1.07h
Page Load Time	20.6m	20.3m	13.7m	7.4m	7.4m	7.6m
Data Source Read Time	21.0m	20.7m	14.1m	7.8m	7.8m	8.0m

Metric	Test 1	Test 2	Test 3	Test 4	Test 5	Test 6
Storage Read Bytes	5.9 GiB	5.9 GiB	5.9 GiB	5.9 GiB	5.9 GiB	5.9 GiB
Size of Files Read	256.6 GiB	256.6 GiB	245.1 GiB	237.6 GiB	237.6 GiB	237.6 GiB
Peak Memory	2.5 GiB	2.5 GiB	2.4 GiB	2.8 GiB	2.9 GiB	2.9 GiB
Output Bytes	40.1 GiB	40.1 GiB	40.1 GiB	40.1 GiB	40.1 GiB	40.1 GiB

Metric	Test 1	Test 2	Test 3	Test 4	Test 5	Test 6
Processed Row Groups	8,727	8,727	4,894	2,677	2,677	2,677
Output Rows	1,375,234,677	1,375,234,677	1,375,234,677	1,375,234,677	1,375,234,677	1,375,234,677
Raw Input Rows	7,199,920,789	7,199,920,789	7,199,920,789	7,199,920,789	7,199,920,789	7,199,920,789
Output Vectors	1,763,258	1,763,213	1,760,583	1,759,144	1,759,159	1,759,168
Memory Allocations	3,617,435	3,617,334	3,594,017	3,579,321	3,579,441	3,579,406

[FelixYBW]

Spark rule, apply to hive, iceberg and deltalake all. Gluten currently doesn't respect the rule fully
Copied from https://docs.delta.io/delta-batch/#spark-configurations

Spark configurations

When you start a Spark application on a cluster, you can set the Spark configurations in the form of spark.hadoop.* to pass your custom Hadoop configurations. For example, setting a value for spark.hadoop.a.b.c will pass the value as a Hadoop configuration a.b.c, and Delta Lake will use it to access Hadoop FileSystem APIs.

See Spark documentation for more details.

SQL session configurations

Spark SQL will pass all of the current SQL session configurations to Delta Lake, and Delta Lake will use them to access Hadoop FileSystem APIs. For example, SET a.b.c=x.y.z will tell Delta Lake to pass the value x.y.z as a Hadoop configuration a.b.c, and Delta Lake will use it to access Hadoop FileSystem APIs.

[FelixYBW]

Rowgroup size a velox bug, fixed by facebookincubator/velox#15751