Copilot commented on code in PR #12589:
URL: https://github.com/apache/gluten/pull/12589#discussion_r3629322730
##########
tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/Suite.scala:
##########
@@ -70,7 +70,6 @@ abstract class Suite(
new SparkSessionSwitcher(appName, masterUrl, logLevel.toString)
// define initial configs
- sessionSwitcher.addDefaultConf("spark.sql.sources.useV1SourceList", "")
sessionSwitcher.addDefaultConf("spark.sql.shuffle.partitions",
s"$shufflePartitions")
sessionSwitcher.addDefaultConf("spark.storage.blockManagerSlaveTimeoutMs",
"3600000")
sessionSwitcher.addDefaultConf("spark.executor.heartbeatInterval", "10s")
Review Comment:
The removed default `spark.sql.sources.useV1SourceList = ""` changes the
integration harness to potentially prefer V1 file sources again, which can
prevent DataSource V2 batch scans from being used and therefore make the new
`batchscan.maxInputPartitions` behavior untestable in gluten-it runs. Consider
keeping the previous default (forcing V2) or making it an
explicit/parameterized suite option so V2 behavior remains the baseline for
these integration tests.
##########
docs/Configuration.md:
##########
@@ -46,6 +46,7 @@ nav_order: 15
| spark.gluten.sql.columnar.appendData | π
Dynamic | true | Enable or disable columnar v2 command append
data.
|
| spark.gluten.sql.columnar.arrowUdf | π
Dynamic | true | Enable or disable columnar arrow udf.
|
| spark.gluten.sql.columnar.batchscan | π
Dynamic | true | Enable or disable columnar batchscan.
|
+| spark.gluten.sql.columnar.batchscan.maxInputPartitions | π
Dynamic | 2147483647 | Maximum number of Spark task partitions for
supported DataSource V2 batch scans.
|
Review Comment:
The documentation currently reads as if this always limits task partitions,
but the implementation only coalesces when `outputPartitioning` is
`UnknownPartitioning`. Please clarify this constraint (and the rationale around
key-grouped/storage-partitioned joins) so users understand when the setting
will be ignored.
##########
gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala:
##########
@@ -854,6 +856,14 @@ object GlutenConfig extends ConfigRegistry {
.booleanConf
.createWithDefault(true)
+ val COLUMNAR_BATCHSCAN_MAX_INPUT_PARTITIONS =
+ buildConf("spark.gluten.sql.columnar.batchscan.maxInputPartitions")
+ .doc(
+ "Maximum number of Spark task partitions for supported DataSource V2
batch scans. ")
+ .intConf
+ .checkValue(_ > 0, s"must be positive.")
+ .createWithDefault(Int.MaxValue)
Review Comment:
Minor doc/registry polish: the `.doc(...)` string has a trailing space, and
`.checkValue(..., s"must be positive.")` uses interpolated-string syntax
without interpolation. Consider removing the trailing space and using a plain
string literal for the check message to keep generated docs/messages clean and
consistent.
##########
gluten-substrait/src/main/scala/org/apache/gluten/execution/BatchScanExecTransformer.scala:
##########
@@ -184,11 +186,31 @@ abstract class BatchScanExecTransformerBase(
commonPartitionValues,
applyPartialClustering,
replicatePartitions)
- .zipWithIndex
- .map {
- case (inputPartitions, index) => new
SparkDataSourceRDDPartition(index, inputPartitions)
+
+ val target = GlutenConfig.get.batchScanMaxInputPartitions
+ val taskPartitions =
+ if (
+ orderedPartitions.size > target &&
+ // Coalescing changes task boundaries. Only do it when Spark does not
advertise a
+ // distribution whose partition groups must remain aligned, such as
key-grouped
+ // partitioning used by storage-partitioned joins.
+ outputPartitioning.isInstanceOf[UnknownPartitioning]
+ ) {
+ Seq.tabulate(target) {
+ index =>
+ val from = index * orderedPartitions.size / target
+ val until = (index + 1) * orderedPartitions.size / target
+ orderedPartitions.slice(from, until).flatten
+ }
+ } else {
+ orderedPartitions
}
Review Comment:
The new behavior has a semantic guard
(`outputPartitioning.isInstanceOf[UnknownPartitioning]`) that bypasses
coalescing even when `maxInputPartitions` is set. Thereβs test coverage for the
coalescing path (VeloxScanSuite), but it would be valuable to add a test that
asserts coalescing does *not* happen when the scan advertises a non-unknown
partitioning (e.g., a key-grouped/storage-partitioned join scenario), to
prevent regressions that could break partition-group alignment guarantees.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
