Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[SPARK-45592][SPARK-45282][SQL][3.4] Correctness issue in AQE with InMemoryTableScanExec #43729

Closed
wants to merge 2 commits into from
Closed

[SPARK-45592][SPARK-45282][SQL][3.4] Correctness issue in AQE with InMemoryTableScanExec #43729

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4bbec91
[SPARK-45592][SQL] Correctness issue in AQE with InMemoryTableScanExec
eejbyfeldt Oct 31, 2023
cdc4498
Use test case from SPARK-45282
eejbyfeldt Nov 9, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
49 changes: 49 additions & 0 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -306,6 +306,35 @@ case class HashPartitioning(expressions: Seq[Expression], numPartitions: Int)

override protected def withNewChildrenInternal(
newChildren: IndexedSeq[Expression]): HashPartitioning = copy(expressions = newChildren)

}

case class CoalescedBoundary(startReducerIndex: Int, endReducerIndex: Int)

/**
* Represents a partitioning where partitions have been coalesced from a HashPartitioning into a
* fewer number of partitions.
*/
case class CoalescedHashPartitioning(from: HashPartitioning, partitions: Seq[CoalescedBoundary])
extends Expression with Partitioning with Unevaluable {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

hmm why this needs to extend Expression and Unevaluable, I thought just Partitioning is enough.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It was just based on how it was done for HashPartitioning, could be that it not needed.


override def children: Seq[Expression] = from.expressions
override def nullable: Boolean = from.nullable
override def dataType: DataType = from.dataType

override def satisfies0(required: Distribution): Boolean = from.satisfies0(required)

override def createShuffleSpec(distribution: ClusteredDistribution): ShuffleSpec =
CoalescedHashShuffleSpec(from.createShuffleSpec(distribution), partitions)

override protected def withNewChildrenInternal(
newChildren: IndexedSeq[Expression]): CoalescedHashPartitioning =
copy(from = from.copy(expressions = newChildren))

override val numPartitions: Int = partitions.length

override def toString: String = from.toString
override def sql: String = from.sql
}

/**
Expand Down Expand Up @@ -661,6 +690,26 @@ case class HashShuffleSpec(
override def numPartitions: Int = partitioning.numPartitions
}

case class CoalescedHashShuffleSpec(
from: ShuffleSpec,
partitions: Seq[CoalescedBoundary]) extends ShuffleSpec {

override def isCompatibleWith(other: ShuffleSpec): Boolean = other match {
case SinglePartitionShuffleSpec =>
numPartitions == 1
case CoalescedHashShuffleSpec(otherParent, otherPartitions) =>
partitions == otherPartitions && from.isCompatibleWith(otherParent)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suppose both from and otherParent are HashShuffleSpec, is it possible that they may have different number of partitions, but after coalescing the number of partitions become the same?

In this case we'll consider the two incompatible but they actually should be?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Even they are coalesced to same number of partitions, the coalesced boundary could be different. I think this is root of the issue and why it needs to make sure boundaries are the same when checking compatibility.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm, this is not related to my above comment. The check partitions == otherPartitions is OK, my question is on from.isCompatibleWith(otherParent), which checks (assuming both are HashShuffleSpec) whether the original number of partitions are the same.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think it must to be HashShuffleSpec. You mean that their partition numbers can be different but compatible?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I mean their partition numbers are different and thus from.isCompatibleWith(otherParent) will return false, and thus cause this CoalescedHashShuffleSpec.isCompatibleWith to also return false. But should we return true instead if the partitions after coalescing are the same?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In other words, from.isCompatibleWith(otherParent) could be too conservative

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm, if first hash partitioning is 5 partitions, second is 4 partitions. How can we get same coalesced partitions with that?

For example:

[[0, 3], [3, 5]] != [[0, 3], [3, 4]]
[[0, 2], [2, 3], [3, 5]] != [[0, 2], [2, 3], [3, 4]]

The end reducer index of last coalesced partition should be different always, no?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not sure whether it is possible for this case to happen. But irrespective of that, I feel this check is unnecessary here.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Of course this is relatively minor stuff and not related to this backport.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

But what about two (nonsensical) CoalescedHashShuffleSpec where is partition is just coalesced from single partition from the some parent HashShuffleSpec. Is it not then correct to do the from.isCompatibleWith(otherParent) check? or is it expected that we can make them compatible by just coalescing them in such a way?

case ShuffleSpecCollection(specs) =>
specs.exists(isCompatibleWith)
case _ =>
false
}

override def canCreatePartitioning: Boolean = false

override def numPartitions: Int = partitions.length
}

case class KeyGroupedShuffleSpec(
partitioning: KeyGroupedPartitioning,
distribution: ClusteredDistribution) extends ShuffleSpec {
Expand Down
124 changes: 68 additions & 56 deletions sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/DistributionSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ package org.apache.spark.sql.catalyst
import org.apache.spark.SparkFunSuite
/* Implicit conversions */
import org.apache.spark.sql.catalyst.dsl.expressions._
import org.apache.spark.sql.catalyst.expressions.{Literal, Murmur3Hash, Pmod}
import org.apache.spark.sql.catalyst.expressions.{Expression, Literal, Murmur3Hash, Pmod}
import org.apache.spark.sql.catalyst.plans.physical._
import org.apache.spark.sql.types.IntegerType

Expand Down Expand Up @@ -146,63 +146,75 @@ class DistributionSuite extends SparkFunSuite {
false)
}

test("HashPartitioning is the output partitioning") {
// HashPartitioning can satisfy ClusteredDistribution iff its hash expressions are a subset of
// the required clustering expressions.
checkSatisfied(
HashPartitioning(Seq($"a", $"b", $"c"), 10),
ClusteredDistribution(Seq($"a", $"b", $"c")),
true)

checkSatisfied(
HashPartitioning(Seq($"b", $"c"), 10),
ClusteredDistribution(Seq($"a", $"b", $"c")),
true)

checkSatisfied(
HashPartitioning(Seq($"a", $"b", $"c"), 10),
ClusteredDistribution(Seq($"b", $"c")),
false)

checkSatisfied(
HashPartitioning(Seq($"a", $"b", $"c"), 10),
ClusteredDistribution(Seq($"d", $"e")),
false)

// When ClusteredDistribution.requireAllClusterKeys is set to true,
// HashPartitioning can only satisfy ClusteredDistribution iff its hash expressions are
// exactly same as the required clustering expressions.
checkSatisfied(
HashPartitioning(Seq($"a", $"b", $"c"), 10),
ClusteredDistribution(Seq($"a", $"b", $"c"), requireAllClusterKeys = true),
true)

checkSatisfied(
HashPartitioning(Seq($"b", $"c"), 10),
ClusteredDistribution(Seq($"a", $"b", $"c"), requireAllClusterKeys = true),
false)

checkSatisfied(
HashPartitioning(Seq($"b", $"a", $"c"), 10),
ClusteredDistribution(Seq($"a", $"b", $"c"), requireAllClusterKeys = true),
false)

// HashPartitioning cannot satisfy OrderedDistribution
checkSatisfied(
HashPartitioning(Seq($"a", $"b", $"c"), 10),
OrderedDistribution(Seq($"a".asc, $"b".asc, $"c".asc)),
false)
private def testHashPartitioningLike(
partitioningName: String,
create: (Seq[Expression], Int) => Partitioning): Unit = {

test(s"$partitioningName is the output partitioning") {
// HashPartitioning can satisfy ClusteredDistribution iff its hash expressions are a subset of
// the required clustering expressions.
checkSatisfied(
create(Seq($"a", $"b", $"c"), 10),
ClusteredDistribution(Seq($"a", $"b", $"c")),
true)

checkSatisfied(
create(Seq($"b", $"c"), 10),
ClusteredDistribution(Seq($"a", $"b", $"c")),
true)

checkSatisfied(
create(Seq($"a", $"b", $"c"), 10),
ClusteredDistribution(Seq($"b", $"c")),
false)

checkSatisfied(
create(Seq($"a", $"b", $"c"), 10),
ClusteredDistribution(Seq($"d", $"e")),
false)

// When ClusteredDistribution.requireAllClusterKeys is set to true,
// HashPartitioning can only satisfy ClusteredDistribution iff its hash expressions are
// exactly same as the required clustering expressions.
checkSatisfied(
create(Seq($"a", $"b", $"c"), 10),
ClusteredDistribution(Seq($"a", $"b", $"c"), requireAllClusterKeys = true),
true)

checkSatisfied(
create(Seq($"b", $"c"), 10),
ClusteredDistribution(Seq($"a", $"b", $"c"), requireAllClusterKeys = true),
false)

checkSatisfied(
create(Seq($"b", $"a", $"c"), 10),
ClusteredDistribution(Seq($"a", $"b", $"c"), requireAllClusterKeys = true),
false)

// HashPartitioning cannot satisfy OrderedDistribution
checkSatisfied(
create(Seq($"a", $"b", $"c"), 10),
OrderedDistribution(Seq($"a".asc, $"b".asc, $"c".asc)),
false)

checkSatisfied(
create(Seq($"a", $"b", $"c"), 1),
OrderedDistribution(Seq($"a".asc, $"b".asc, $"c".asc)),
false) // TODO: this can be relaxed.

checkSatisfied(
create(Seq($"b", $"c"), 10),
OrderedDistribution(Seq($"a".asc, $"b".asc, $"c".asc)),
false)
}
}

checkSatisfied(
HashPartitioning(Seq($"a", $"b", $"c"), 1),
OrderedDistribution(Seq($"a".asc, $"b".asc, $"c".asc)),
false) // TODO: this can be relaxed.
testHashPartitioningLike("HashPartitioning",
(expressions, numPartitions) => HashPartitioning(expressions, numPartitions))

checkSatisfied(
HashPartitioning(Seq($"b", $"c"), 10),
OrderedDistribution(Seq($"a".asc, $"b".asc, $"c".asc)),
false)
}
testHashPartitioningLike("CoalescedHashPartitioning", (expressions, numPartitions) =>
CoalescedHashPartitioning(
HashPartitioning(expressions, numPartitions), Seq(CoalescedBoundary(0, numPartitions))))

test("RangePartitioning is the output partitioning") {
// RangePartitioning can satisfy OrderedDistribution iff its ordering is a prefix
Expand Down