Skip to content

[SPARK-26504][SQL] Rope-wise dumping of Spark plans #23406

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

Closed
MaxGekk wants to merge 13 commits into from
Closed

[SPARK-26504][SQL] Rope-wise dumping of Spark plans #23406

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
de38cf8
Use append function instead of Writer
MaxGekk Dec 29, 2018
af5c9ed
Fix typo
MaxGekk Dec 29, 2018
0a36a26
StringRope on List
MaxGekk Dec 29, 2018
07ae72d
Tests for StringRope
MaxGekk Dec 29, 2018
f621de7
Porting on StringRope
MaxGekk Dec 29, 2018
ca4aed8
Rope other methods
MaxGekk Dec 29, 2018
d727959
Added docs
MaxGekk Dec 30, 2018
544b80e
Replacing List by ArrayBuffer
MaxGekk Dec 30, 2018
059dcf4
Inline stringOrError
MaxGekk Dec 30, 2018
fe3bbcf
Move appendOrError to QueryPlan
MaxGekk Dec 30, 2018
707c80b
Renaming StringRope to StringConcat
MaxGekk Dec 30, 2018
074e9b8
rope -> concat in tests
MaxGekk Dec 30, 2018
29b62bf
StringBuffer -> StringBuilder
MaxGekk Dec 31, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 17 additions & 21 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreeNode.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,13 +17,11 @@

package org.apache.spark.sql.catalyst.trees

import java.io.Writer
import java.util.UUID

import scala.collection.Map
import scala.reflect.ClassTag

import org.apache.commons.io.output.StringBuilderWriter
import org.apache.commons.lang3.ClassUtils
import org.json4s.JsonAST._
import org.json4s.JsonDSL._
Expand All @@ -37,6 +35,7 @@ import org.apache.spark.sql.catalyst.errors._
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.plans.JoinType
import org.apache.spark.sql.catalyst.plans.physical.{BroadcastMode, Partitioning}
import org.apache.spark.sql.catalyst.util.StringUtils.StringRope
import org.apache.spark.sql.catalyst.util.truncatedString
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
Expand Down Expand Up @@ -481,21 +480,18 @@ abstract class TreeNode[BaseType <: TreeNode[BaseType]] extends Product {
verbose: Boolean,
addSuffix: Boolean = false,
maxFields: Int = SQLConf.get.maxToStringFields): String = {
val writer = new StringBuilderWriter()
try {
treeString(writer, verbose, addSuffix, maxFields)
writer.toString
} finally {
writer.close()
}
val rope = new StringRope()

treeString(rope.append, verbose, addSuffix, maxFields)
rope.toString
}

def treeString(
writer: Writer,
append: String => Unit,
verbose: Boolean,
addSuffix: Boolean,
maxFields: Int): Unit = {
generateTreeString(0, Nil, writer, verbose, "", addSuffix, maxFields)
generateTreeString(0, Nil, append, verbose, "", addSuffix, maxFields)
}

/**
Expand Down Expand Up @@ -558,42 +554,42 @@ abstract class TreeNode[BaseType <: TreeNode[BaseType]] extends Product {
def generateTreeString(
depth: Int,
lastChildren: Seq[Boolean],
writer: Writer,
append: String => Unit,
verbose: Boolean,
prefix: String = "",
addSuffix: Boolean = false,
maxFields: Int): Unit = {

if (depth > 0) {
lastChildren.init.foreach { isLast =>
writer.write(if (isLast) " " else ": ")
append(if (isLast) " " else ": ")
}
writer.write(if (lastChildren.last) "+- " else ":- ")
append(if (lastChildren.last) "+- " else ":- ")
}

val str = if (verbose) {
if (addSuffix) verboseStringWithSuffix(maxFields) else verboseString(maxFields)
} else {
simpleString(maxFields)
}
writer.write(prefix)
writer.write(str)
writer.write("\n")
append(prefix)
append(str)
append("\n")

if (innerChildren.nonEmpty) {
innerChildren.init.foreach(_.generateTreeString(
depth + 2, lastChildren :+ children.isEmpty :+ false, writer, verbose,
depth + 2, lastChildren :+ children.isEmpty :+ false, append, verbose,
addSuffix = addSuffix, maxFields = maxFields))
innerChildren.last.generateTreeString(
depth + 2, lastChildren :+ children.isEmpty :+ true, writer, verbose,
depth + 2, lastChildren :+ children.isEmpty :+ true, append, verbose,
addSuffix = addSuffix, maxFields = maxFields)
}

if (children.nonEmpty) {
children.init.foreach(_.generateTreeString(
depth + 1, lastChildren :+ false, writer, verbose, prefix, addSuffix, maxFields))
depth + 1, lastChildren :+ false, append, verbose, prefix, addSuffix, maxFields))
children.last.generateTreeString(
depth + 1, lastChildren :+ true, writer, verbose, prefix, addSuffix, maxFields)
depth + 1, lastChildren :+ true, append, verbose, prefix, addSuffix, maxFields)
}
}

Expand Down
24 changes: 24 additions & 0 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/StringUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,4 +87,28 @@ object StringUtils {
}
funcNames.toSeq
}

class StringRope {
Comment thread
MaxGekk marked this conversation as resolved.
Outdated
private var list = List.empty[String]

@hvanhovell hvanhovell Dec 29, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Let's use a ListBuffer or an ArrayBuffer here. Those have (amortized) constant time appends and do not force you to reverse the collection when building the string.

@MaxGekk MaxGekk Dec 29, 2018 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

I supposed this reverse is relatively cheap O(n). Coping a string would be more expensive than adding element to the head of list inside of the reverse() method. In any case, need to traverse over the list in toString.

@MaxGekk MaxGekk Dec 30, 2018

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

I replaced List by ArrayBuffer

private var length: Int = 0

def append(s: String): Unit = {
if (s != null) {
list = s :: list
length += s.length
}
}

override def toString: String = {
val buffer = new StringBuffer(length)
var reversed = list.reverse

while (!reversed.isEmpty) {
buffer.append(reversed.head)
reversed = reversed.tail
}

@hvanhovell hvanhovell Dec 30, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Why the new line?

@MaxGekk MaxGekk Dec 30, 2018

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

for beauty

buffer.toString
}
}
}
13 changes: 13 additions & 0 deletions sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/StringUtilsSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,4 +43,17 @@ class StringUtilsSuite extends SparkFunSuite {
assert(filterPattern(names, " a. ") === Seq("a1", "a2"))
assert(filterPattern(names, " d* ") === Nil)
}

test("string rope") {
def toRope(seq: String*): String = {
seq.foldLeft(new StringRope())((rope, s) => {rope.append(s); rope}).toString
}

assert(new StringRope().toString == "")
assert(toRope("") == "")
assert(toRope(null) == "")
assert(toRope("a") == "a")
assert(toRope("1", "2") == "12")
assert(toRope("abc", "\n", "123") == "abc\n123")
}
}
82 changes: 44 additions & 38 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,11 +17,10 @@

package org.apache.spark.sql.execution

import java.io.{BufferedWriter, OutputStreamWriter, Writer}
import java.io.{BufferedWriter, OutputStreamWriter}
import java.nio.charset.StandardCharsets
import java.sql.{Date, Timestamp}

import org.apache.commons.io.output.StringBuilderWriter
import org.apache.hadoop.fs.Path

import org.apache.spark.rdd.RDD
Expand All @@ -31,6 +30,7 @@ import org.apache.spark.sql.catalyst.analysis.UnsupportedOperationChecker
import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, ReturnAnswer}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.catalyst.util.StringUtils.StringRope
import org.apache.spark.sql.catalyst.util.truncatedString
import org.apache.spark.sql.execution.command.{DescribeTableCommand, ExecutedCommandExec, ShowTablesCommand}
import org.apache.spark.sql.execution.exchange.{EnsureRequirements, ReuseExchange}
Expand Down Expand Up @@ -108,10 +108,6 @@ class QueryExecution(
ReuseExchange(sparkSession.sessionState.conf),
ReuseSubquery(sparkSession.sessionState.conf))

protected def stringOrError[A](f: => A): String =
try f.toString catch { case e: AnalysisException => e.toString }


/**
* Returns the result as a hive compatible sequence of strings. This is used in tests and
* `SparkSQLDriver` for CLI applications.
Expand Down Expand Up @@ -197,55 +193,65 @@ class QueryExecution(
}

def simpleString: String = withRedaction {
s"""== Physical Plan ==
|${stringOrError(executedPlan.treeString(verbose = false))}
""".stripMargin.trim
val rope = new StringRope()

rope.append("== Physical Plan ==\n")
appendOrError(rope.append)(
executedPlan.treeString(_, false, false, SQLConf.get.maxToStringFields))
rope.append("\n")

rope.toString
}

private def writeOrError(writer: Writer)(f: Writer => Unit): Unit = {
try f(writer)
private def appendOrError(append: String => Unit)(f: (String => Unit) => Unit): Unit = {

@hvanhovell hvanhovell Dec 29, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

You always call this method on a QueryPlan. Can we specialize this method for this scenario and pass the plan and all the needed treeString parameters?

Different question. Is it possible that treeString already writes to the appender before throwing an exception. It it does the output might look pretty weird, because it will and contain a part of the tree and the exception thrown.

@MaxGekk MaxGekk Dec 29, 2018

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

because it will and contain a part of the tree and the exception thrown

In the case of writing to a file, I think it is possible. I believe it will be a nice feature in trouble shooting. I would image a huge (maybe wrong) plan causes OOMs at some point. If we write some part of the plan to file, it would be helpful in debugging.

@MaxGekk MaxGekk Dec 30, 2018

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Can we specialize this method for this scenario and pass the plan and all the needed treeString parameters?

@hvanhovell Do you mean changes like in the PR MaxGekk#16 ? Unfortunately it doesn't work well because plan's constructor can produce AnalysisException which cannot be handled with this approach.

@hvanhovell hvanhovell Dec 30, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Yeah that would be it. You can pass in a call-by-name parameter if you want to capture errors during construction. I prefer his over having some overly generic function.

@MaxGekk MaxGekk Dec 30, 2018

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Will be ok if I move appendOrError to the companion object of QueryPlan? Like:

object QueryPlan extends PredicateHelper {
  /**
   * Converts the query plan to string and appends it via provided function.
   */
  def append[T <: QueryPlan[T]](
      plan: => QueryPlan[T],
      append: String => Unit,
      verbose: Boolean,
      addSuffix: Boolean,
      maxFields: Int = SQLConf.get.maxToStringFields): Unit = { () }

@hvanhovell hvanhovell Dec 30, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Sure that works.

try f(append)
catch {
case e: AnalysisException => writer.write(e.toString)
case e: AnalysisException => append(e.toString)
}
}

private def writePlans(writer: Writer, maxFields: Int): Unit = {
private def writePlans(append: String => Unit, maxFields: Int): Unit = {
def stringOrError[A](f: => A): String = {
try f.toString catch { case e: AnalysisException => e.toString }
}
val (verbose, addSuffix) = (true, false)

writer.write("== Parsed Logical Plan ==\n")
writeOrError(writer)(logical.treeString(_, verbose, addSuffix, maxFields))
writer.write("\n== Analyzed Logical Plan ==\n")
append("== Parsed Logical Plan ==\n")
appendOrError(append)(logical.treeString(_, verbose, addSuffix, maxFields))
append("\n== Analyzed Logical Plan ==\n")
val analyzedOutput = stringOrError(truncatedString(
analyzed.output.map(o => s"${o.name}: ${o.dataType.simpleString}"), ", ", maxFields))
writer.write(analyzedOutput)
writer.write("\n")
writeOrError(writer)(analyzed.treeString(_, verbose, addSuffix, maxFields))
writer.write("\n== Optimized Logical Plan ==\n")
writeOrError(writer)(optimizedPlan.treeString(_, verbose, addSuffix, maxFields))
writer.write("\n== Physical Plan ==\n")
writeOrError(writer)(executedPlan.treeString(_, verbose, addSuffix, maxFields))
append(analyzedOutput)
append("\n")
appendOrError(append)(analyzed.treeString(_, verbose, addSuffix, maxFields))
append("\n== Optimized Logical Plan ==\n")
appendOrError(append)(optimizedPlan.treeString(_, verbose, addSuffix, maxFields))
append("\n== Physical Plan ==\n")
appendOrError(append)(executedPlan.treeString(_, verbose, addSuffix, maxFields))
}

override def toString: String = withRedaction {
val writer = new StringBuilderWriter()
try {
writePlans(writer, SQLConf.get.maxToStringFields)
writer.toString
} finally {
writer.close()
}
val rope = new StringRope()

writePlans(rope.append, SQLConf.get.maxToStringFields)
rope.toString
}

def stringWithStats: String = withRedaction {
val rope = new StringRope()
val maxFields = SQLConf.get.maxToStringFields

// trigger to compute stats for logical plans
optimizedPlan.stats

// only show optimized logical plan and physical plan
s"""== Optimized Logical Plan ==
|${stringOrError(optimizedPlan.treeString(verbose = true, addSuffix = true))}
|== Physical Plan ==
|${stringOrError(executedPlan.treeString(verbose = true))}
""".stripMargin.trim
rope.append("== Optimized Logical Plan ==\n")
appendOrError(rope.append)(optimizedPlan.treeString(_, true, true, maxFields))
rope.append("\n== Physical Plan ==\n")
appendOrError(rope.append)(executedPlan.treeString(_, true, false, maxFields))
rope.append("\n")

rope.toString
}

/**
Expand Down Expand Up @@ -282,17 +288,17 @@ class QueryExecution(
/**
* Dumps debug information about query execution into the specified file.
*
* @param maxFields maximim number of fields converted to string representation.
Comment thread
MaxGekk marked this conversation as resolved.
* @param maxFields maximum number of fields converted to string representation.
*/
def toFile(path: String, maxFields: Int = Int.MaxValue): Unit = {
val filePath = new Path(path)
val fs = filePath.getFileSystem(sparkSession.sessionState.newHadoopConf())
val writer = new BufferedWriter(new OutputStreamWriter(fs.create(filePath)))

try {
writePlans(writer, maxFields)
writePlans(writer.write, maxFields)
writer.write("\n== Whole Stage Codegen ==\n")
org.apache.spark.sql.execution.debug.writeCodegen(writer, executedPlan)
org.apache.spark.sql.execution.debug.writeCodegen(writer.write, executedPlan)
} finally {
writer.close()
}
Expand Down
8 changes: 4 additions & 4 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/WholeStageCodegenExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -493,15 +493,15 @@ case class InputAdapter(child: SparkPlan) extends UnaryExecNode with InputRDDCod
override def generateTreeString(
depth: Int,
lastChildren: Seq[Boolean],
writer: Writer,
append: String => Unit,
verbose: Boolean,
prefix: String = "",
addSuffix: Boolean = false,
maxFields: Int): Unit = {
child.generateTreeString(
depth,
lastChildren,
writer,
append,
verbose,
prefix = "",
addSuffix = false,
Expand Down Expand Up @@ -777,15 +777,15 @@ case class WholeStageCodegenExec(child: SparkPlan)(val codegenStageId: Int)
override def generateTreeString(
depth: Int,
lastChildren: Seq[Boolean],
writer: Writer,
append: String => Unit,
verbose: Boolean,
prefix: String = "",
addSuffix: Boolean = false,
maxFields: Int): Unit = {
child.generateTreeString(
depth,
lastChildren,
writer,
append,
verbose,
s"*($codegenStageId) ",
false,
Expand Down
26 changes: 10 additions & 16 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/debug/package.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,13 +17,10 @@

package org.apache.spark.sql.execution

import java.io.Writer
import java.util.Collections

import scala.collection.JavaConverters._

import org.apache.commons.io.output.StringBuilderWriter

import org.apache.spark.internal.Logging
import org.apache.spark.rdd.RDD
import org.apache.spark.sql._
Expand All @@ -32,6 +29,7 @@ import org.apache.spark.sql.catalyst.expressions.Attribute
import org.apache.spark.sql.catalyst.expressions.codegen.{CodeFormatter, CodegenContext, ExprCode}
import org.apache.spark.sql.catalyst.plans.physical.Partitioning
import org.apache.spark.sql.catalyst.trees.TreeNodeRef
import org.apache.spark.sql.catalyst.util.StringUtils.StringRope
import org.apache.spark.sql.execution.streaming.{StreamExecution, StreamingQueryWrapper}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.streaming.StreamingQuery
Expand Down Expand Up @@ -73,24 +71,20 @@ package object debug {
* @return single String containing all WholeStageCodegen subtrees and corresponding codegen
*/
def codegenString(plan: SparkPlan): String = {
val writer = new StringBuilderWriter()
val rope = new StringRope()

try {
writeCodegen(writer, plan)
writer.toString
} finally {
writer.close()
}
writeCodegen(rope.append, plan)
rope.toString
}

def writeCodegen(writer: Writer, plan: SparkPlan): Unit = {
def writeCodegen(append: String => Unit, plan: SparkPlan): Unit = {
val codegenSeq = codegenStringSeq(plan)
writer.write(s"Found ${codegenSeq.size} WholeStageCodegen subtrees.\n")
append(s"Found ${codegenSeq.size} WholeStageCodegen subtrees.\n")
for (((subtree, code), i) <- codegenSeq.zipWithIndex) {
writer.write(s"== Subtree ${i + 1} / ${codegenSeq.size} ==\n")
writer.write(subtree)
writer.write("\nGenerated code:\n")
writer.write(s"${code}\n")
append(s"== Subtree ${i + 1} / ${codegenSeq.size} ==\n")
append(subtree)
append("\nGenerated code:\n")
append(s"${code}\n")
}
}

Expand Down