Skip to content

[SPARK-26129][SQL] Instrumentation for per-query planning time #23096

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
rxin wants to merge 6 commits into from
Closed

[SPARK-26129][SQL] Instrumentation for per-query planning time #23096

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
b6a3d02
[SPARK-26129][SQL] Instrumentation for query planning time
rxin Nov 20, 2018
b2a2a01
Add parsing
rxin Nov 20, 2018
dd61273
remove merge
rxin Nov 20, 2018
f36a231
thread local
rxin Nov 20, 2018
2cd069c
fix compile
rxin Nov 20, 2018
34f8bfe
fix Hive test cases
rxin Nov 21, 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
Prev Previous commit
fix Hive test cases
  • Loading branch information
@rxin
rxin committed Nov 21, 2018
commit 34f8bfe69b70ff702324ec7f38d78ae920410ef7
2 changes: 1 addition & 1 deletion sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/QueryPlanningTracker.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,7 +60,7 @@ object QueryPlanningTracker {
}

/** Returns the current tracker in scope, based on the thread local variable. */
def get: QueryPlanningTracker = localTracker.get()
def get: Option[QueryPlanningTracker] = Option(localTracker.get())

/** Sets the current tracker for the execution of function f. We assume f is single-threaded. */
def withTracker[T](tracker: QueryPlanningTracker)(f: => T): T = {
Expand Down
7 changes: 3 additions & 4 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleExecutor.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,7 +86,7 @@ abstract class RuleExecutor[TreeType <: TreeNode[_]] extends Logging {
var curPlan = plan
val queryExecutionMetrics = RuleExecutor.queryExecutionMeter
val planChangeLogger = new PlanChangeLogger()
val tracker = QueryPlanningTracker.get
val tracker: Option[QueryPlanningTracker] = QueryPlanningTracker.get

batches.foreach { batch =>
val batchStartPlan = curPlan
Expand All @@ -111,9 +111,8 @@ abstract class RuleExecutor[TreeType <: TreeNode[_]] extends Logging {
queryExecutionMetrics.incExecutionTimeBy(rule.ruleName, runTime)
queryExecutionMetrics.incNumExecution(rule.ruleName)

if (tracker ne null) {
tracker.recordRuleInvocation(rule.ruleName, runTime, effective)
}
// Record timing information using QueryPlanningTracker
tracker.foreach(_.recordRuleInvocation(rule.ruleName, runTime, effective))
Copy link

@abehm abehm Nov 20, 2018

Choose a reason for hiding this comment

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

Doesn't this make the query-local and the global metrics inconsistent when tracker is None?

Copy link
Contributor Author

@rxin rxin Nov 20, 2018

Choose a reason for hiding this comment

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

yes! (not great -- but I'd probably remove the global tracker at some point)

Copy link

@abehm abehm Nov 20, 2018

Choose a reason for hiding this comment

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

removing the global tracker would be great!


// Run the structural integrity checker against the plan after each rule.
if (!isPlanIntegral(result)) {
Expand Down
17 changes: 13 additions & 4 deletions sql/hive/src/main/scala/org/apache/spark/sql/hive/test/TestHive.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,10 +33,9 @@ import org.apache.hadoop.hive.serde2.`lazy`.LazySimpleSerDe

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.internal.Logging
import org.apache.spark.sql.{SparkSession, SQLContext}
import org.apache.spark.sql.catalyst.QueryPlanningTracker
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession, SQLContext}
import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
import org.apache.spark.sql.catalyst.catalog.{ExternalCatalog, ExternalCatalogWithListener}
import org.apache.spark.sql.catalyst.catalog.ExternalCatalogWithListener
import org.apache.spark.sql.catalyst.optimizer.ConvertToLocalRelation
import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, OneRowRelation}
import org.apache.spark.sql.execution.{QueryExecution, SQLExecution}
Expand Down Expand Up @@ -220,6 +219,16 @@ private[hive] class TestHiveSparkSession(
sharedState.externalCatalog.unwrapped.asInstanceOf[HiveExternalCatalog].client.newSession()
}

/**
* This is a temporary hack to override SparkSession.sql so we can still use the version of
* Dataset.ofRows that creates a TestHiveQueryExecution (rather than a normal QueryExecution
* which wouldn't load all the test tables).
*/
override def sql(sqlText: String): DataFrame = {
val plan = sessionState.sqlParser.parsePlan(sqlText)
Dataset.ofRows(self, plan)
}

override def newSession(): TestHiveSparkSession = {
new TestHiveSparkSession(sc, Some(sharedState), None, loadTestTables)
}
Expand Down Expand Up @@ -587,7 +596,7 @@ private[hive] class TestHiveQueryExecution(
logDebug(s"Query references test tables: ${referencedTestTables.mkString(", ")}")
referencedTestTables.foreach(sparkSession.loadTestTable)
// Proceed with analysis.
sparkSession.sessionState.analyzer.executeAndCheck(logical, new QueryPlanningTracker)
sparkSession.sessionState.analyzer.executeAndCheck(logical, tracker)
}
}

Expand Down