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 1 commit
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
Next Next commit
Use append function instead of Writer
  • Loading branch information
@MaxGekk
MaxGekk committed Dec 29, 2018
commit de38cf895f12f4433b10702fab12efaa179dce27
26 changes: 13 additions & 13 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 @@ -483,19 +483,19 @@ abstract class TreeNode[BaseType <: TreeNode[BaseType]] extends Product {
maxFields: Int = SQLConf.get.maxToStringFields): String = {
val writer = new StringBuilderWriter()
try {
treeString(writer, verbose, addSuffix, maxFields)
treeString(writer.write, verbose, addSuffix, maxFields)
writer.toString
} finally {
writer.close()
}
}

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 +558,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
32 changes: 16 additions & 16 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 @@ -202,34 +202,34 @@ class QueryExecution(
""".stripMargin.trim
}

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

@hvanhovell hvanhovell Dec 29, 2018

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.

Copy link
Member Author

@MaxGekk MaxGekk Dec 29, 2018

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.

Copy link
Member Author

@MaxGekk MaxGekk Dec 30, 2018

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.

Copy link
Contributor

@hvanhovell hvanhovell Dec 30, 2018

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.

Copy link
Member Author

@MaxGekk MaxGekk Dec 30, 2018

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 = { () }

Copy link
Contributor

@hvanhovell hvanhovell Dec 30, 2018

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 = {
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)
writePlans(writer.write, SQLConf.get.maxToStringFields)
writer.toString
} finally {
writer.close()
Expand Down Expand Up @@ -290,7 +290,7 @@ class QueryExecution(
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)
} finally {
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