peter-toth commented on PR #58814:
URL: https://github.com/apache/spark/pull/58814#issuecomment-5679504983
@ulysses-you on reproducing the sweep, from your comment on #58659.
Two things. The invariant it measures is now a test in this PR,
`EnsureRequirementsSuite`: 18 shapes a side over six join types and eight
configuration cells, asserting that a join which skipped both shuffles left the
two sides on the same keys. It runs in under three seconds and fails without
the gate. That is the acceptance test you want for follow-up work on the
marker, and it does not rot, because it asserts the property rather than the
counts.
The full 307200-plan harness is below. It is deliberately not committed: its
value is the difference between two runs, and an absolute-count assertion would
break on every neighbouring change. To use it, drop it in as
`sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/SpjSweepScratchSuite.scala`,
run `SWEEP_TAG=<tag> build/sbt 'sql/testOnly *SpjSweepScratchSuite'` on each
side, and diff the two `/tmp/sweep-cases-<tag>.txt` files. Diff the case lists,
not the summary: two counts can match while the sets differ.
<details>
<summary>SpjSweepScratchSuite.scala</summary>
```scala
/*
* 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.rdd.RDD
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.plans._
import org.apache.spark.sql.catalyst.plans.physical._
import org.apache.spark.sql.connector.catalog.functions._
import org.apache.spark.sql.execution.{LeafExecNode, SparkPlan}
import org.apache.spark.sql.execution.joins.SortMergeJoinExec //
scalastyle:ignore
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types.IntegerType
/**
* SCRATCH. A differential sweep over generated storage-partitioned join
shapes. Not for commit.
* Set SWEEP_TAG to name the output files in /tmp.
*/
class SpjSweepScratchSuite extends SharedSparkSession {
private val tag = sys.env.getOrElse("SWEEP_TAG", "sweep")
private val aL = AttributeReference("aL", IntegerType)()
private val bL = AttributeReference("bL", IntegerType)()
private val aR = AttributeReference("aR", IntegerType)()
private val bR = AttributeReference("bR", IntegerType)()
private def bucket(n: Int, e: Expression): TransformExpression =
TransformExpression(BucketFunction, Seq(e), Some(n))
private def years(e: Expression): TransformExpression =
TransformExpression(YearsFunction, Seq(e))
private case class SweepLeaf(override val outputPartitioning: Partitioning)
extends LeafExecNode {
override def output: Seq[Attribute] = Nil
override protected def doExecute(): RDD[InternalRow] =
throw new UnsupportedOperationException("sweep leaf")
}
/** One key row per element; a repeated row makes the source ungrouped. */
private val keySets: Seq[(String, Seq[Seq[Int]])] = Seq(
"grouped-12" -> Seq(Seq(1), Seq(2)),
"grouped-123" -> Seq(Seq(1), Seq(2), Seq(3)),
"ungrouped-112" -> Seq(Seq(1), Seq(1), Seq(2)),
"disjoint-45" -> Seq(Seq(4), Seq(5)),
"empty" -> Seq.empty)
private val keySets2: Seq[(String, Seq[Seq[Int]])] = Seq(
"grouped-2d" -> Seq(Seq(1, 1), Seq(1, 2), Seq(2, 1)),
"ungrouped-2d" -> Seq(Seq(1, 1), Seq(1, 1), Seq(2, 1)),
"disjoint-2d" -> Seq(Seq(4, 4), Seq(5, 5)))
private def rows(keys: Seq[Seq[Int]]): Seq[InternalRow] =
keys.map(k => InternalRow.fromSeq(k.map(_.asInstanceOf[Any])))
/** Every partitioning one side can report, with a name. */
private def sideShapes(a: Attribute, b: Attribute): Seq[(String,
Partitioning)] = {
val single = for {
(kn, keys) <- keySets
(en, exprs) <- Seq(
"id" -> Seq[Expression](a),
"bucket4" -> Seq[Expression](bucket(4, a)),
"years" -> Seq[Expression](years(a)))
} yield (s"$en/$kn", KeyedPartitioning(exprs,
rows(keys)).asInstanceOf[Partitioning])
val pairs = for {
(kn, keys) <- keySets2
(en, exprs) <- Seq(
"id2" -> Seq[Expression](a, b),
"id2rev" -> Seq[Expression](b, a),
"bucket+id" -> Seq[Expression](bucket(4, a), b))
} yield (s"$en/$kn", KeyedPartitioning(exprs,
rows(keys)).asInstanceOf[Partitioning])
val collections = for {
(kn, keys) <- keySets2
} yield (s"coll/$kn", PartitioningCollection.fromPartitionings(Seq(
KeyedPartitioning(Seq(a, b), rows(keys)),
KeyedPartitioning(Seq(b, a), rows(keys)))).asInstanceOf[Partitioning])
// A layout that pins rows outside the declared keys to hash(key) %
numPartitions. Regrouping
// one moves those rows, which is the shape SPARK-59272 is about.
val marked = for {
(kn, keys) <- keySets
(en, exprs) <- Seq(
"id" -> Seq[Expression](a),
"bucket4" -> Seq[Expression](bucket(4, a)))
} yield (s"marked-$en/$kn",
KeyedPartitioning(exprs, rows(keys))
.withLayout(_.copy(mayContainUnknownPartitionKeys =
true)).asInstanceOf[Partitioning])
val marked2 = for {
(kn, keys) <- keySets2
} yield (s"marked-id2/$kn",
KeyedPartitioning(Seq(a, b), rows(keys))
.withLayout(_.copy(mayContainUnknownPartitionKeys =
true)).asInstanceOf[Partitioning])
single ++ pairs ++ collections ++ marked ++ marked2
}
private val joinTypes: Seq[JoinType] =
Seq(Inner, LeftOuter, RightOuter, FullOuter, LeftSemi, LeftAnti)
private val configKeys = Seq(
SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED,
SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS,
SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION,
SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED,
SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED)
private type Wrapper =
org.apache.spark.sql.catalyst.util.InternalRowComparableWrapper
/** The keys the topmost keyed partitioning of `plan` declares, or None
when it reports none. */
private def declaredKeys(plan: SparkPlan): Option[Seq[Wrapper]] =
PartitioningCollection.flatten(plan.outputPartitioning)
.collectFirst { case k: KeyedPartitioning => k.partitionKeys }
test("SWEEP") {
val rule = new EnsureRequirements()
val leftShapes = sideShapes(aL, bL)
val rightShapes = sideShapes(aR, bR)
var plans = 0
var validateFailures = 0
var coPartitionViolations = 0
var errors = 0
val errorKinds = scala.collection.mutable.Map.empty[String, Int]
val bad = scala.collection.mutable.ArrayBuffer.empty[String]
for (bits <- 0 until (1 << configKeys.length)) {
val settings = configKeys.zipWithIndex.map { case (c, i) =>
c.key -> (((bits >> i) & 1) == 1).toString
} :+ (SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true")
withSQLConf(settings: _*) {
for {
(ln, lp) <- leftShapes
(rn, rp) <- rightShapes
joinType <- joinTypes
} {
val left = SweepLeaf(lp)
val right = SweepLeaf(rp)
val smj = SortMergeJoinExec(Seq(aL), Seq(aR), joinType, None,
left, right)
plans += 1
val caseKey = s"$bits|$ln|$rn|$joinType"
try {
val planned = rule.apply(smj)
if (!ValidateRequirements.validate(planned)) {
validateFailures += 1
bad += s"VALIDATE $caseKey"
}
val shuffles = planned.children.map(c =>
c.exists(_.isInstanceOf[ShuffleExchangeLike]))
if (!shuffles.head && !shuffles(1)) {
val l = declaredKeys(planned.children.head)
val r = declaredKeys(planned.children(1))
if (l.isEmpty || l != r) {
coPartitionViolations += 1
bad += s"COPART $caseKey"
}
}
} catch {
case e: Throwable =>
errors += 1
bad += s"ERROR $caseKey"
val k = e.getClass.getSimpleName + ": " +
String.valueOf(e.getMessage).take(80)
errorKinds(k) = errorKinds.getOrElse(k, 0) + 1
}
}
}
}
val report = new StringBuilder
report ++= s"tag=$tag plans=$plans validateFailures=$validateFailures " +
s"coPartitionViolations=$coPartitionViolations errors=$errors\n"
errorKinds.toSeq.sortBy(-_._2).foreach { case (k, c) => report ++= s"
$c x $k\n" }
// scalastyle:off println
println(report.toString)
// scalastyle:on println
java.nio.file.Files.write(java.nio.file.Paths.get(s"/tmp/sweep-$tag.txt"),
report.toString.getBytes("UTF-8"))
java.nio.file.Files.write(java.nio.file.Paths.get(s"/tmp/sweep-cases-$tag.txt"),
bad.sorted.mkString("\n").getBytes("UTF-8"))
}
}
```
</details>
--
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]
