[SPARK-51064][SQL] Enable spark.sql.sources.v2.bucketing.enabled by default

[dongjoon-hyun]

What changes were proposed in this pull request?

This PR aims to enable spark.sql.sources.v2.bucketing.enabled by default for Apache Spark 4.1.0.

Why are the changes needed?

We have been using spark.sql.sources.v2.bucketing.enabled since Apache Spark 3.3.0 stably to improve Spark performance on V2 data source. Although Apache Spark enables this configuration, Spark checks if all InputPartitions have HasPartitionKey or not still.

spark/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala

Lines 140 to 142 in 8fc6a20

	if (results.length != inputPartitions.length || inputPartitions.isEmpty) {
	// Not all of the `InputPartitions` implements `HasPartitionKey`, therefore skip here.
	None

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Pass the CIs.

Was this patch authored or co-authored using generative AI tooling?

No.

[dongjoon-hyun]

What do you think about this change, @sunchao and @szehon-ho ?

[dongjoon-hyun]

Also, cc @aokolnychyi , @viirya , @huaxingao . I'd like to get your opinions on this.

[sunchao]

thanks @dongjoon-hyun ! +1 on making this as the default config - the feature has been stabilized for several release cycles already.

[viirya]

As it is used since Apache Spark 3.3.0, the feature should be stable enough for now. Except for any concerns from others, I'm okay for this change.

[dongjoon-hyun]

Thank you, @sunchao and @viirya .
Merged to master for Apache Spark 4.1.0.

[guangyu-yang-rokt]

Hi @dongjoon-hyun, sorry this might be a question a bit unrelated to this PR.

Context:
We are currently introducing SPJ to our production environment. Our iceberg table is partitioned by timestamp with day transformation and in our ML processing job we will read past 30 days worth of data with filter on timestamp column which will be pushed down to iceberg. So 30 partitions will be reported by iceberg. I have observed that with spark.sql.sources.v2.bucketing.enabled, spark will then generate one task per partition, which will be only 30 tasks in our case when we are doing batchScan. This has led to cluster resource under utilisation since we have 40 execs and 15 cores each (so at max 600 tasks in parallel). That impact the batchScan performance a lot - same stage from 2.4 mins to 10+mins.

Have you encountered the same issue in your use case? I must be missing something here. Any insights will be much appreciated!

[szehon-ho]

@guangyu-yang-rokt yea its right, by design in SPJ the number of Iceberg partitions is equal to number of Spark partitions, so sizing that is critical. There is an option to enable more parallelism in spark.sql.sources.v2.bucketing.partiallyClusteredDistribution.enabled, but it leads to more times the data read, and doesnt work in all cases (not FULL OUTER joins)

[guangyu-yang-rokt]

Thanks @szehon-ho! one follow up question - in our query, we only do filtering on timestamp column but join key is something different (joining on non-partition keys). I have checked that BatchScanExec is reporting groupedBy=[timestamp_day] in query plan. I'm not too familiar with spark codebase but I guess filter pushdown to iceberg also tell BatchScanExec to group by partition key if there is a filter on partition key. With spark.sql.sources.v2.bucketing.enabledset to true, it will slow down batchScan for joins that are not joining on partition keys. (we have a self-implemented featurestore which will spin up multiple joins to gather features in so I need to enable all SPJ related configs globally)

This is kind not making sense to me since I'm not joining on timestamp so I would expect SPJ shouldn't kick in. Or I would imagine a configuration like spark.sql.sources.v2.ignoreFiltering to tell BatchScanExec don't grouped by partition key if it is just a filter and not a join key