peter-toth commented on code in PR #59522:
URL: https://github.com/apache/spark/pull/59522#discussion_r4426649041
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -1461,5 +1461,6 @@ case class EnsureRequirements(
} else {
newPlan
}
+ PackV2ScanPartitions(finalPlan, requiredDistribution)
Review Comment:
**Finding 3.** Called from the end of `EnsureRequirements.apply`, packing
runs on every `EnsureRequirements` call. That includes the re-runs inside
`ConvertSortMergeJoinToShuffledHashJoin`
(`ConvertSortMergeJoinToShuffledHashJoin.scala:155`) and `OptimizeSkewedJoin`
(`OptimizeSkewedJoin.scala:270`), and tests or extensions that call
`EnsureRequirements()(plan)`. The plan-change log also attributes the rewrite
to `EnsureRequirements`.
Packing is a separate concern from satisfying requirements.
`DisableUnnecessaryBucketedScan` is the precedent: a standalone rule after
`EnsureRequirements` that withdraws a scan's layout claim. A packing rule can
sit right after `StampUnionDecisions` in both `QueryExecution.preparations` and
AQE's `queryStagePreparationRules`, so every rule after it sees the same plan
as today. In AQE it can take `requiredDistribution` as `EnsureRequirements`
does (`AdaptiveSparkPlanExec.scala:141`).
I tried that on this head. 357 of the 358 tests in the packing suites, the
three re-run suites and `EnsureRequirementsSuite` pass. The one failure is
"repeated requirement planning keeps packing stable", which calls
`EnsureRequirements()` directly and expects a packed scan. It would call the
new rule after it instead.
If you keep the current shape, please rename the inner `val finalPlan` at
line 1453. It shadows the new outer one, and the two have different types
(`Seq[SparkPlan]` and `SparkPlan`).
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitionPacking.scala:
##########
@@ -0,0 +1,120 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.v2
+
+import scala.collection.mutable
+
+import org.apache.spark.rdd.{PartitionCoalescer, PartitionGroup, RDD}
+import org.apache.spark.sql.catalyst.plans.physical.KeyedPartitioning
+import org.apache.spark.sql.connector.read.{Batch, HasPartitionStatistics,
InputPartition, PartitionReaderFactory, Scan}
+
+/** Immutable sizing policy captured when a scan is prepared, before runtime
filtering. */
+case class V2ScanPartitionPacking(
+ maxPartitionBytes: Long,
+ openCostInBytes: Long,
+ minPartitionNum: Int,
+ maxPartitionNum: Option[Int]) {
+
+ /**
+ * Group adjacent inputs by estimated size, keeping an oversized input on
its own. Missing,
+ * negative or mixed statistics use the explicit count cap, or leave inputs
unchanged.
+ * The cap is only a fallback: it never overrides the size limit for inputs
reporting bytes.
+ * Empty runtime-pruned slots are retained and contribute no bytes or
opening cost.
Review Comment:
**Finding 6.** Since 4372e37d08b both callers pass only `Some` inputs:
`BatchScanExec.scala:192` passes the inputs left after `:183` filters out the
pruned slots, and `PackV2ScanPartitions.scala:60` wraps the unfiltered inputs.
So the `None` case and this sentence describe a path nothing takes, and the
last assertion of "overflow, exact limits and runtime-pruned empty slots" pins
it. `pack(partitions: Seq[InputPartition])` would say what it handles.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/PackV2ScanPartitions.scala:
##########
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.exchange
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.SortOrder
+import org.apache.spark.sql.catalyst.plans.physical.{Distribution,
UnspecifiedDistribution}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.execution._
+import org.apache.spark.sql.execution.aggregate.BaseAggregateExec
+import org.apache.spark.sql.execution.command.DataWritingCommandExec
+import org.apache.spark.sql.execution.datasources.WriteFilesExec
+import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec,
V2ScanPartitionPacking, V2TableWriteExec}
+import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec,
BroadcastNestedLoopJoinExec}
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Pack V2 scan inputs after distribution and ordering requirements have been
planned.
+ * A shuffle or broadcast exchange consumes neither input property. Otherwise,
propagate
+ * consumers' requirements conservatively through known operators, including a
broadcast
+ * join's streamed side. Keep unknown operators and explicit coalesces
unchanged.
+ *
+ * Packing happens inside BatchScanExec, before partial aggregation and
UnsafeRow conversion.
+ * No keyed or ordered output is claimed, and scans used by
GroupPartitionsExec stay unchanged.
+ */
+object PackV2ScanPartitions extends Rule[SparkPlan] {
+ override def apply(plan: SparkPlan): SparkPlan = apply(plan, None)
+
+ def apply(plan: SparkPlan, requiredDistribution: Option[Distribution]):
SparkPlan = {
+ if (!conf.getConf(SQLConf.V2_SCAN_PARTITION_PACKING_ENABLED)) {
+ return plan
+ }
+ lazy val options = V2ScanPartitionPacking(
+ conf.getConf(SQLConf.V2_SCAN_PARTITION_PACKING_MAX_PARTITION_BYTES),
+ conf.getConf(SQLConf.V2_SCAN_PARTITION_PACKING_OPEN_COST_IN_BYTES),
+
conf.getConf(SQLConf.V2_SCAN_PARTITION_PACKING_MIN_PARTITION_NUM).getOrElse(
+ conf.getConf(SQLConf.LEAF_NODE_DEFAULT_PARALLELISM).getOrElse(
+ SparkSession.active.sparkContext.defaultParallelism)),
+ conf.getConf(SQLConf.V2_SCAN_PARTITION_PACKING_MAX_PARTITION_NUM))
+
+ def rewrite(node: SparkPlan, needsPartitioning: Boolean, needsOrdering:
Boolean): SparkPlan = {
+ node match {
+ case scan: BatchScanExec if !needsPartitioning && !needsOrdering &&
+ scan.partitionPacking.isEmpty &&
+ options.pack(scan.inputPartitions.map(Some(_))).size <
scan.inputPartitions.size =>
+ scan.withPartitionPacking(options)
+ case _: BatchScanExec => node
+ case _: ShuffleExchangeLike | _: BroadcastExchangeLike =>
+ node.withNewChildren(node.children.zipWithIndex.map { case (child,
i) =>
+ rewrite(child,
+ node.requiredChildDistribution(i) != UnspecifiedDistribution,
+ node.requiredChildOrdering(i).nonEmpty)
+ })
+ case sort: SortExec =>
+ // Keep a scan ordering that makes this sort redundant. Otherwise
the existing sort
+ // establishes ordering itself, so packing cannot add a new sort or
lose its output order.
+ val redundant =
SortOrder.orderingSatisfies(sort.child.outputOrdering, sort.sortOrder)
+ sort.withNewChildren(Seq(rewrite(sort.child,
+ needsPartitioning || sort.requiredChildDistribution.head !=
UnspecifiedDistribution,
+ redundant)))
+ case _: ProjectExec | _: FilterExec | _: BaseAggregateExec |
+ _: BroadcastHashJoinExec | _: BroadcastNestedLoopJoinExec |
+ _: V2TableWriteExec | _: DataWritingCommandExec | _:
WriteFilesExec =>
+ val children = node.children.zipWithIndex.map { case (child, i) =>
+ rewrite(child,
+ needsPartitioning || node.requiredChildDistribution(i) !=
UnspecifiedDistribution,
+ needsOrdering || node.requiredChildOrdering(i).nonEmpty)
+ }
+ node.withNewChildren(children)
+ case _ =>
+ // Includes GroupPartitionsExec, UnionExec, windows, explicit
coalesces and extensions.
+ // An exchange further down can still isolate a scan from these
consumers.
+ node.mapChildren(rewrite(_, needsPartitioning = true, needsOrdering
= true))
+ }
+ }
+
+ rewrite(plan, requiredDistribution.exists(_ != UnspecifiedDistribution),
needsOrdering = false)
Review Comment:
**Finding 1.** The walk starts with `needsPartitioning` only when an AQE
`requiredDistribution` is set, so the root scan of a plan being cached or
checkpointed is packed. That plan's `outputPartitioning` is what a later query
plans against: `InMemoryTableScanExec` and the checkpoint's `LogicalRDD` both
report it, and SPARK-53322 relies on it for SPJ.
Measured on this head: a cached `items` table (keyed by `id`, two splits
each for keys 1 and 3), joined to `purchases` (keyed by `item_id`), broadcast
off, with the packing settings of `PackV2ScanPartitionsPlanSuite`:
| | AQE off | AQE on |
|---|---|---|
| packing off (as on master) | 0 shuffles, 2 `GroupPartitionsExec` | 0
shuffles, 2 `GroupPartitionsExec` |
| packing on | 2 shuffles, no `GroupPartitionsExec` | 2 shuffles, no
`GroupPartitionsExec` |
All four runs return the same rows.
`CacheManager.getOrCloneSessionWithConfigsOff` already turns off the rules that
change a cached plan's output partitioning, unless
`spark.sql.optimizer.canChangeCachedPlanOutputPartitioning` is set. Adding
packing there:
```scala
disableConfigs =
disableConfigs :+
SQLConf.ADAPTIVE_EXECUTION_APPLY_FINAL_STAGE_SHUFFLE_OPTIMIZATIONS :+
SQLConf.V2_SCAN_PARTITION_PACKING_ENABLED
```
gives 0 shuffles and 2 `GroupPartitionsExec` in all four runs. None of this
PR's tests cache a packed scan.
The checkpoint case is already in this PR's own test changes. With packing
on, "SPARK-53322: checkpointed scans are used for SPJ" goes from 1 to 3
shuffles with `pushPartValues`, and "SPARK-53322: checkpointed scans can
shuffle other children on SPJ" from 2 to 3. `Dataset.checkpoint` plans through
`QueryExecution.withRegularShuffle`, which already re-plans with a config
turned off through `getOrCloneSessionWithConfigsOff`, so packing could be
turned off for checkpoints the same way. I have not measured that one. With
both in place, the paragraph in `sql-performance-tuning.md` that asks users to
turn packing off before materializing can go.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/PackV2ScanPartitionsSuite.scala:
##########
@@ -0,0 +1,219 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector
+
+import java.util.OptionalLong
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.catalyst.expressions.UnsafeRow
+import org.apache.spark.sql.catalyst.plans.physical.UnknownPartitioning
+import org.apache.spark.sql.connector.catalog.Identifier
+import org.apache.spark.sql.connector.expressions.Expressions.identity
+import org.apache.spark.sql.connector.read.{HasPartitionStatistics,
InputPartition}
+import org.apache.spark.sql.execution.{CoalesceExec, ProjectExec}
+import org.apache.spark.sql.execution.aggregate.BaseAggregateExec
+import org.apache.spark.sql.execution.datasources.v2.V2ScanPartitionPacking
+import org.apache.spark.sql.execution.exchange.{EnsureRequirements,
PackV2ScanPartitions, ShuffleExchangeExec, ValidateRequirements}
+import org.apache.spark.sql.internal.SQLConf
+
+class V2ScanPartitionPackingSuite extends org.apache.spark.SparkFunSuite {
+ private case class Sized(bytes: Option[Long]) extends HasPartitionStatistics
{
+ override def sizeInBytes(): OptionalLong = bytes.map(OptionalLong.of)
+ .getOrElse(OptionalLong.empty())
+ override def numRows(): OptionalLong = OptionalLong.empty()
+ override def filesCount(): OptionalLong = OptionalLong.empty()
+ }
+ private def inputs(sizes: Long*): Seq[Option[InputPartition]] =
+ sizes.map(size => Some(Sized(Some(size))))
+
+ test("pack actual input sizes, keeping oversized splits separate and
covering each input once") {
+ val policy = V2ScanPartitionPacking(100, 0, 1, Some(1))
+ val parts = inputs(30, 40, 200, 20, 30, 90)
+ val groups = policy.pack(parts)
+ assert(groups === Seq(Seq(0, 1), Seq(2), Seq(3, 4), Seq(5)))
+ assert(groups.flatten === parts.indices)
+ assert(policy.pack(Seq.empty).isEmpty)
+ assert(policy.pack(inputs(0, 0, 0)) === Seq(Seq(0, 1, 2)))
+ }
+
+ test("opening cost and suggested parallelism constrain small inputs") {
+ assert(V2ScanPartitionPacking(100, 10, 1, None).pack(inputs(40, 40, 0)) ===
+ Seq(Seq(0, 1), Seq(2)))
+ assert(V2ScanPartitionPacking(1000, 0, 3, None).pack(inputs(100, 100, 100,
100)) ===
+ Seq(Seq(0), Seq(1), Seq(2), Seq(3)))
+ assert(V2ScanPartitionPacking(5, 10, 1, None).pack(inputs(0, 0)) ===
+ Seq(Seq(0), Seq(1)))
+ }
+
+ test("missing, mixed and negative sizes require an explicit count fallback")
{
+ val unknown = Some(new InputPartition {})
+ Seq(unknown, Some(Sized(None)), Some(Sized(Some(-1)))).foreach { p =>
+ val parts = inputs(30, 40) :+ p
+ assert(V2ScanPartitionPacking(100, 0, 1, None).pack(parts).size === 3)
+ assert(V2ScanPartitionPacking(100, 0, 1, Some(2)).pack(parts) ===
+ Seq(Seq(0), Seq(1, 2)))
+ assert(V2ScanPartitionPacking(100, 0, 1, Some(10)).pack(parts).size ===
3)
+ }
+ }
+
+ test("overflow, exact limits and runtime-pruned empty slots") {
+ val policy = V2ScanPartitionPacking(Long.MaxValue, 10, 1, None)
+ assert(policy.pack(inputs(Long.MaxValue, 1)) === Seq(Seq(0), Seq(1)))
+ assert(V2ScanPartitionPacking(100, 0, 1, None).pack(inputs(100, 1)) ===
+ Seq(Seq(0), Seq(1)))
+ assert(V2ScanPartitionPacking(100, 0, 1, None).pack(
+ Seq(None, Some(Sized(Some(30))), None, Some(Sized(Some(40))))) ===
+ Seq(Seq(0, 1, 2, 3)))
+ }
+}
+
+class PackV2ScanPartitionsSuite extends KeyGroupedPartitioningSuiteBase {
+ private val enabled = SQLConf.V2_SCAN_PARTITION_PACKING_ENABLED.key
+ override def sparkConf: SparkConf = super.sparkConf
+ .set(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD, -1L)
+ .set(SQLConf.V2_SCAN_PARTITION_PACKING_OPEN_COST_IN_BYTES, 0L)
+ .set(SQLConf.V2_SCAN_PARTITION_PACKING_MAX_PARTITION_BYTES, 2400L)
+ .set(SQLConf.V2_SCAN_PARTITION_PACKING_MIN_PARTITION_NUM, 1)
+
+ after { catalog.clearTables() }
+
+ private def createSplits(keys: Int = 2, splits: Int = 24, keyed: Boolean =
true): Unit = {
+ createTable(items, itemsColumns, if (keyed) Array(identity("id")) else
Array.empty)
+ sql(s"INSERT INTO testcat.ns.$items VALUES " + (1 to keys).flatMap { id =>
+ (1 to splits).map(i => s"($id, 'split$i', 1.0, NULL)")
+ }.mkString(", "))
+ }
+
+ test("SPARK-60265: scan packing is disabled by default") {
+ createSplits(splits = 3)
+
assert(!spark.sessionState.conf.getConf(SQLConf.V2_SCAN_PARTITION_PACKING_ENABLED))
+ val df = sql(s"SELECT name, count(*) FROM testcat.ns.$items GROUP BY name")
+ checkAnswer(df, (1 to 3).map(i => Row(s"split$i", 2L)))
+
assert(collectScans(df.queryExecution.executedPlan).forall(_.partitionPacking.isEmpty))
+ }
+
+ test("SPARK-60265: size-pack keyed and unkeyed scans before partial
aggregation") {
+ for (keyed <- Seq(false, true)) {
+ createSplits(keyed = keyed)
+ for (adaptive <- Seq(false, true); packing <- Seq(false, true);
+ bucketing <- Seq(false, true)) {
+ withSQLConf(enabled -> packing.toString,
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> adaptive.toString,
+ SQLConf.V2_BUCKETING_ENABLED.key -> bucketing.toString) {
+ val df = sql(s"SELECT price, count(*), max(id) FROM
testcat.ns.$items GROUP BY price")
+ checkAnswer(df, Seq(Row(1.0, 48L, 2L)))
+ val plan = df.queryExecution.executedPlan
+ val scan = collectScans(plan).head
+ assert(scan.inputRDD.getNumPartitions === (if (packing) 2 else 48))
+ assert(scan.partitionPacking.isDefined === packing)
+ val exchange = collect(plan) { case s: ShuffleExchangeExec => s
}.head
+ val aggregate = collect(exchange.child) { case a: BaseAggregateExec
=> a }.head
+ // Once per packed task, rather than once per split inside a
CoalesceExec.
+ val partialRows = aggregate.execute().count()
+ assert(partialRows === (if (packing) 2L else 48L))
+ assert(collect(plan) { case c: CoalesceExec => c }.isEmpty)
+ assert(ValidateRequirements.validate(stripAQEPlan(plan)))
+ }
+ }
+ catalog.dropTable(Identifier.of(Array("ns"), items))
+ }
+ }
+
+ test("SPARK-60265: packing depends on bytes rather than key count") {
+ createSplits(keys = 1)
+ withSQLConf(enabled -> "true", SQLConf.ADAPTIVE_EXECUTION_ENABLED.key ->
"false",
+ SQLConf.V2_SCAN_PARTITION_PACKING_MAX_PARTITION_BYTES.key -> "600") {
+ val df = sql(s"SELECT name, count(*) FROM testcat.ns.$items GROUP BY
name")
+ checkAnswer(df, (1 to 24).map(i => Row(s"split$i", 1L)))
+ val scan = collectScans(df.queryExecution.executedPlan).head
+ assert(scan.inputRDD.getNumPartitions === 4)
+ assert(scan.outputPartitioning === UnknownPartitioning(0))
+ assert(scan.outputOrdering.isEmpty)
+ }
+ }
+
+ test("SPARK-60265: row conversion survives direct collection, shuffle,
filtering and codegen") {
+ createSplits(splits = 3)
+ for (adaptive <- Seq(false, true); codegen <- Seq(false, true); filter <-
Seq(false, true)) {
+ withSQLConf(enabled -> "true", SQLConf.ADAPTIVE_EXECUTION_ENABLED.key ->
adaptive.toString,
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> codegen.toString) {
+ val where = if (filter) " WHERE id > 0" else ""
+ val direct = sql(s"SELECT id, name FROM testcat.ns.$items$where")
+ checkAnswer(direct, for (id <- 1 to 2; i <- 1 to 3) yield Row(id,
s"split$i"))
+ val plan = stripAQEPlan(direct.queryExecution.executedPlan)
+ assert(collectScans(plan).head.partitionPacking.isDefined)
+ assert(collect(plan) { case p: ProjectExec => p }.nonEmpty || codegen)
+
assert(plan.execute().map(_.isInstanceOf[UnsafeRow]).collect().forall(value =>
value))
+ val shuffled = sql(s"SELECT name, count(*), max(id) FROM
testcat.ns.$items$where " +
+ "GROUP BY name")
+ checkAnswer(shuffled, (1 to 3).map(i => Row(s"split$i", 2L, 2L)))
+
assert(ValidateRequirements.validate(stripAQEPlan(shuffled.queryExecution.executedPlan)))
+ }
+ }
+ }
+
+ test("SPARK-60265: repeated requirement planning keeps packing stable") {
+ createSplits(splits = 3)
+ withSQLConf(enabled -> "true", SQLConf.ADAPTIVE_EXECUTION_ENABLED.key ->
"false") {
+ val df = sql(s"SELECT name, count(*) FROM testcat.ns.$items GROUP BY
name")
+ val once = EnsureRequirements()(df.queryExecution.sparkPlan)
+ val twice = EnsureRequirements()(once)
+ assert(PackV2ScanPartitions(twice) === twice)
+ assert(collectScans(twice).head.partitionPacking.isDefined)
+ assert(ValidateRequirements.validate(twice))
+ checkAnswer(df, (1 to 3).map(i => Row(s"split$i", 2L)))
+ }
+ }
+
+ test("SPARK-60265: empty scans and explicit coalesces remain unchanged") {
+ createTable(items, itemsColumns, Array(identity("id")))
+ withSQLConf(enabled -> "true", SQLConf.ADAPTIVE_EXECUTION_ENABLED.key ->
"false") {
+ val empty = spark.table(s"testcat.ns.$items")
+ checkAnswer(empty, Seq.empty)
+
assert(collectScans(empty.queryExecution.executedPlan).head.partitionPacking.isEmpty)
+ sql(s"INSERT INTO testcat.ns.$items VALUES (1, 'a', 1.0, NULL), (2, 'b',
1.0, NULL)")
+ val input = spark.table(s"testcat.ns.$items")
+ val df = input.coalesce(1).groupBy("name").count()
+ checkAnswer(df, Seq(Row("a", 1L), Row("b", 1L)))
+
assert(collectScans(df.queryExecution.executedPlan).head.partitionPacking.isEmpty)
+ assert(collect(df.queryExecution.executedPlan) { case c: CoalesceExec =>
c.numPartitions } ===
+ Seq(1))
+ }
+ }
+}
+
+/** Re-run existing SPJ, ordering and reuse regressions with scan packing
enabled. */
+class KeyGroupedPartitioningPackScanPartitionsSuite extends
KeyGroupedPartitioningSuite {
Review Comment:
**Finding 4.** `KeyGroupedPartitioningSuite` and both runtime-filter suites
carry `@ExtendedSQLTest`, but the annotation has no `@Inherited`
(`common/tags/src/test/java/org/apache/spark/tags/ExtendedSQLTest.java:28-30`).
That is why the parents re-annotate each runtime-filter subclass
(`KeyGroupedPartitioningSuite.scala:11108` and `:11122`). So these three
subclasses are untagged, and `build_and_test.yml` runs them in "sql - other
tests" instead of "sql - extended tests". This one is a full re-run of the
largest SPJ suite. Please add the tag to all three (plus the
`org.apache.spark.tags.ExtendedSQLTest` import):
```scala
@ExtendedSQLTest
class KeyGroupedPartitioningPackScanPartitionsSuite extends
KeyGroupedPartitioningSuite {
```
##########
docs/sql-migration-guide.md:
##########
@@ -106,6 +106,8 @@ license: |
## Upgrading from Spark SQL 4.1 to 4.2
+- Since Spark 4.2, V2 scans no longer unconditionally group input splits
sharing a reported partition key into one task (SPARK-55092). Spark still
groups splits when needed to satisfy a downstream distribution, such as for a
storage-partitioned join. Other scans can execute one task per split,
increasing overhead for connectors with many small splits per key. Before 4.2,
grouping multiple splits also discarded their reported ordering. Spark 4.4
introduces the default-off `spark.sql.sources.v2.scanPartitionPacking.enabled`
option to pack eligible batch V2 scan inputs by estimated size, with an
explicit count fallback when sizes are unavailable. This policy is independent
of partition keys and does not restore unconditional per-key grouping. See
[DataSource V2 Scan Task
Sizing](sql-performance-tuning.html#datasource-v2-scan-task-sizing) for its
scope and tradeoffs.
Review Comment:
**Finding 5.** This note documents a 4.2 behavior change, so 4.2.x users
should find it in `branch-4.2`'s migration guide. In this PR, which is a 4.4
feature, it only reaches `branch-4.x` and `master`, and its "Spark 4.4
introduces ..." sentence keeps it from being backported as written. Could the
SPARK-55092 note go into a separate docs PR that can also be merged to
`branch-4.2`? This PR then needs only the tuning section, since a new opt-in
option needs no migration note. Also, the blank line after this bullet turns
the whole section into a loose list.
--
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]
