Skip to content

[SPARK-28901][SQL] SparkThriftServer's Cancel SQL Operation show it in JDBC Tab UI #25611

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
AngersZhuuuu wants to merge 30 commits into from
Closed

[SPARK-28901][SQL] SparkThriftServer's Cancel SQL Operation show it in JDBC Tab UI #25611

Show file tree
Hide file tree
Changes from 14 commits
Commits
Show all changes
30 commits
Select commit Hold shift + click to select a range
b84d385
[SPARK-28901] Support cancel SQL operation.
AngersZhuuuu Aug 28, 2019
07d5679
commit code
AngersZhuuuu Aug 28, 2019
187a4a6
the same as other ethod
AngersZhuuuu Aug 29, 2019
44ba723
fix conflict between cancel and close status
AngersZhuuuu Aug 29, 2019
e8f902e
use old cancel method, a little mistake of code
AngersZhuuuu Aug 29, 2019
d4d1943
fix call cancel before run execut()
AngersZhuuuu Aug 30, 2019
7d77b0c
remove dunplicated code
AngersZhuuuu Aug 30, 2019
72b885d
fix throw exception
AngersZhuuuu Aug 30, 2019
d2d6cc5
close before cancel or finish, also reduce totalRunning
AngersZhuuuu Aug 30, 2019
4c9d5f1
fix other operation
AngersZhuuuu Aug 30, 2019
7b43b59
fix conflicts between cancel and finish
AngersZhuuuu Aug 30, 2019
3744fc9
move try block to satrt of execute(0
AngersZhuuuu Aug 30, 2019
5070161
fix scala style
AngersZhuuuu Aug 30, 2019
41ab7d7
fix code style
AngersZhuuuu Aug 30, 2019
87fa08f
fix error
AngersZhuuuu Aug 30, 2019
63e8b59
remove PREPARED
AngersZhuuuu Aug 30, 2019
bea260a
remove empty line
AngersZhuuuu Aug 30, 2019
7e56c14
fix scala style
AngersZhuuuu Sep 2, 2019
8b25006
remove sync operation judge terminal
AngersZhuuuu Sep 2, 2019
ccd7de9
add empty line
AngersZhuuuu Sep 2, 2019
c6651f1
revert
AngersZhuuuu Sep 2, 2019
00f3553
save code
AngersZhuuuu Sep 3, 2019
8b84e04
fix scala style and concurence problem
AngersZhuuuu Sep 3, 2019
ff5ac96
clear job group in same thread
AngersZhuuuu Sep 3, 2019
28174bd
remove all the totalRunning and onlineSessionNum vars
AngersZhuuuu Sep 3, 2019
1cbf7cc
add onStatementError for background case
AngersZhuuuu Sep 3, 2019
61c9c73
fix code style
AngersZhuuuu Sep 3, 2019
f720963
to do all the initialization bookkeeping first.
AngersZhuuuu Sep 3, 2019
c8d2ffc
fixlog
AngersZhuuuu Sep 3, 2019
536756b
code style fix
AngersZhuuuu Sep 3, 2019
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
36 changes: 25 additions & 11 deletions ...hriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/HiveThriftServer2.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -138,7 +138,7 @@ object HiveThriftServer2 extends Logging {
}

private[thriftserver] object ExecutionState extends Enumeration {
val STARTED, COMPILED, FAILED, FINISHED, CLOSED = Value
val PREPARED, STARTED, COMPILED, CANCELED, FAILED, FINISHED, CLOSED = Value
type ExecutionState = Value
}

Expand Down Expand Up @@ -219,18 +219,22 @@ object HiveThriftServer2 extends Logging {
trimSessionIfNecessary()
}

def onStatementStart(
def onStatementPrepared(
id: String,
sessionId: String,
statement: String,
groupId: String,
userName: String = "UNKNOWN"): Unit = synchronized {
val info = new ExecutionInfo(statement, sessionId, System.currentTimeMillis, userName)
info.state = ExecutionState.STARTED
info.state = ExecutionState.PREPARED
executionList.put(id, info)
trimExecutionIfNecessary()
sessionList(sessionId).totalExecution += 1
executionList(id).groupId = groupId
}

def onStatementStart(id: String): Unit = synchronized {
executionList(id).state = ExecutionState.STARTED
totalRunning += 1
}

Expand All @@ -239,14 +243,20 @@ object HiveThriftServer2 extends Logging {
executionList(id).state = ExecutionState.COMPILED
}

def onStatementError(id: String, errorMessage: String, errorTrace: String): Unit = {
synchronized {
executionList(id).finishTimestamp = System.currentTimeMillis
executionList(id).detail = errorMessage
executionList(id).state = ExecutionState.FAILED
totalRunning -= 1
trimExecutionIfNecessary()
}
def onStatementCanceled(id: String): Unit = synchronized {
executionList(id).finishTimestamp = System.currentTimeMillis
executionList(id).state = ExecutionState.CANCELED
totalRunning -= 1

@juliuszsompolski juliuszsompolski Sep 3, 2019

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.

nit leftover: you can remove all the totalRunning and onlineSessionNum vars.

@AngersZhuuuu AngersZhuuuu Sep 3, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

nit leftover: you can remove all the totalRunning and onlineSessionNum vars.

Good ideal, we have more save method to get this statistics.

trimExecutionIfNecessary()
}


@juliuszsompolski juliuszsompolski Sep 3, 2019

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.

promise final nit: remove double empty line

def onStatementError(id: String, errorMessage: String, errorTrace: String): Unit = synchronized {
Comment thread
AngersZhuuuu marked this conversation as resolved.
Outdated
executionList(id).finishTimestamp = System.currentTimeMillis
executionList(id).detail = errorMessage
executionList(id).state = ExecutionState.FAILED
totalRunning -= 1
trimExecutionIfNecessary()
}

def onStatementFinish(id: String): Unit = synchronized {
Expand All @@ -258,7 +268,11 @@ object HiveThriftServer2 extends Logging {

def onOperationClosed(id: String): Unit = synchronized {
executionList(id).closeTimestamp = System.currentTimeMillis
val lastState = executionList(id).state
executionList(id).state = ExecutionState.CLOSED
if (lastState == ExecutionState.STARTED || lastState == ExecutionState.COMPILED) {
totalRunning -= 1
}

@felixcheung felixcheung Sep 1, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

does FINISHED need to be handled somewhere?

@AngersZhuuuu AngersZhuuuu Sep 1, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

does FINISHED need to be handled somewhere?

this place is to handle situation like : we have call close before finish, then the totalRunning value should -1 too. If we call cancel() or it finished, then when we call closeOperation, we won't do totalRunning -= 1. Since for FINISHED, has do totalRunning -= 1 , so here don't need to contain FINISHED

}

private def trimExecutionIfNecessary() = {
Expand Down
83 changes: 51 additions & 32 deletions ...rc/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -160,9 +160,16 @@ private[hive] class SparkExecuteStatementOperation(
override def runInternal(): Unit = {
setState(OperationState.PENDING)
setHasResultSet(true) // avoid no resultset for async run
statementId = UUID.randomUUID().toString
HiveThriftServer2.listener.onStatementPrepared(

@juliuszsompolski juliuszsompolski Aug 30, 2019

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.

I think it's enough to move onStatementStarted here, without adding Prepared as a separate state... I don't think there's much value added in distinguishing this from when the query starts running in the thread (it will go into a new state almost immediately after that)...

To not have to modify all these other operations, I would just stick with STARTED.

@AngersZhuuuu AngersZhuuuu Aug 30, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

remove it. A little forgot why I do this, move onStatementStarted here is enough

statementId,
parentSession.getSessionHandle.getSessionId.toString,
statement,
statementId,
parentSession.getUsername)

if (!runInBackground) {
execute()
executeWhenNotTerminalStatus()
} else {
val sparkServiceUGI = Utils.getUGI()

Expand All @@ -175,7 +182,7 @@ private[hive] class SparkExecuteStatementOperation(
override def run(): Unit = {
registerCurrentOperationLog()
try {
execute()
executeWhenNotTerminalStatus()
} catch {
case e: HiveSQLException =>
setOperationException(e)
Expand Down Expand Up @@ -212,22 +219,23 @@ private[hive] class SparkExecuteStatementOperation(
}
}

private def execute(): Unit = withSchedulerPool {
statementId = UUID.randomUUID().toString
logInfo(s"Running query '$statement' with $statementId")
setState(OperationState.RUNNING)
// Always use the latest class loader provided by executionHive's state.
val executionHiveClassLoader = sqlContext.sharedState.jarClassLoader
Thread.currentThread().setContextClassLoader(executionHiveClassLoader)
private def executeWhenNotTerminalStatus(): Unit = {
if(!getStatus.getState.isTerminal) {
execute()
Comment thread
AngersZhuuuu marked this conversation as resolved.
Outdated
}
}

HiveThriftServer2.listener.onStatementStart(
statementId,
parentSession.getSessionHandle.getSessionId.toString,
statement,
statementId,
parentSession.getUsername)
sqlContext.sparkContext.setJobGroup(statementId, statement)
private def execute(): Unit = withSchedulerPool {
try {
logInfo(s"Running query '$statement' with $statementId")
setState(OperationState.RUNNING)

@juliuszsompolski juliuszsompolski Sep 2, 2019

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.

I'd change that to

synchronized {
  if (getStatus.getState.isTerminal) {
    logInfo(s"Query with $statementId in terminal state before it started running")
    return
  } else {
    logInfo(s"Running query with $statementId")
    setState(OperationState.RUNNING)
  }
}

instead of the if in runInternal. Otherwise, it can still change the state to cancelled between there and here.
(note: removed $statement from the log message; let's log it during submission)

// Always use the latest class loader provided by executionHive's state.
val executionHiveClassLoader = sqlContext.sharedState.jarClassLoader
Thread.currentThread().setContextClassLoader(executionHiveClassLoader)

@AngersZhuuuu AngersZhuuuu Sep 7, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

@juliuszsompolski call cancel here for run statement in sync mode. seems can't stop task running .

HiveThriftServer2.listener.onStatementStart(statementId)
sqlContext.sparkContext.setJobGroup(statementId, statement)

@AngersZhuuuu AngersZhuuuu Sep 7, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

How about move judgement to this place . I think setState(RUNNING) after setJobGroup is ok.
Since we can think setClassLoader and setJobGroup is prepare work for running

sqlContext.sparkContext.setJobGroup(statementId, statement)
  synchronized {
        if (getStatus.getState.isTerminal) {
          logInfo(s"Query with $statementId in terminal state before it started running")
          return
        } else {
          logInfo(s"Running query with $statementId")
          setState(OperationState.RUNNING)
        }
      }

@juliuszsompolski juliuszsompolski Sep 9, 2019

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.

@AngersZhuuuu
I don't think it will help. It's not just about setJobGroup, but about actually starting the job. You can setJobGroup, then cancelJobGroup, and then start running Jobs in that Job Group.
What cancelJobGroup does is only cancel Jobs that are currently running in the Job group, it doesn't prevent further jobs being started. So I think it can still go past here, and then be cancelled before it actually starts the Jobs.
I think it would be safer to handle it from the catch block.

@AngersZhuuuu AngersZhuuuu Sep 9, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

But when
You can setJobGroup, then cancelJobGroup, and then start running Jobs in that Job Group.
The job is under no JobGroup since when call cancelJobGroup, jobGroup in localProperties has been cleared

@juliuszsompolski juliuszsompolski Sep 9, 2019

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 have

  1. Thread1 be in execute() after this synchronized block, but before the Jobs have started
  2. Thread2 be a user connection calling cancel(). It cancels all jobs in the job group, and notifies Thread1
  3. But before Thread1 gets the notification and throws InterruptedException, it starts some Jobs.
  4. Then Thread1 gets InterruptedException, and exits through the catch and finally block. It does a clearJobGroup in the finally, but that doesn't cancel the Jobs started in 3. These Jobs keep running after Thread1 exits, and nobody cancels them.

I think that the only way to prevent this, is to call another cancelJobGroup from the catch block when an exception comes.

@AngersZhuuuu AngersZhuuuu Sep 9, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Get your point.
Cancel before setJobGroup, but there will be an interval between cancel and execute thread been interrupted. Then job won't be canceled after execute thread get into catch block.


@juliuszsompolski juliuszsompolski Sep 3, 2019

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.

ocd nit: empty lines would be more readable if it was

...
      }

      // Always use the latest class loader provided by executionHive's state.
      val executionHiveClassLoader = sqlContext.sharedState.jarClassLoader
      Thread.currentThread().setContextClassLoader(executionHiveClassLoader)

      sqlContext.sparkContext.setJobGroup(statementId, statement)
      result = sqlContext.sql(statement)
...
  • it then separates the initialization part, the hive class loader part and the Spark part with empty lines.

@AngersZhuuuu AngersZhuuuu Sep 3, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Consider a lot , learn it.

result = sqlContext.sql(statement)
logDebug(result.queryExecution.toString())
result.queryExecution.logical match {
Expand All @@ -249,32 +257,43 @@ private[hive] class SparkExecuteStatementOperation(
}
dataTypes = result.queryExecution.analyzed.output.map(_.dataType).toArray
} catch {
case e: HiveSQLException =>
if (getStatus().getState() == OperationState.CANCELED) {
// Actually do need to catch Throwable as some failures don't inherit from Exception and
// HiveServer will silently swallow them.
case e: Throwable =>

@juliuszsompolski juliuszsompolski Sep 6, 2019

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.

@AngersZhuuuu
I think I found one more problem:
If cancel() and close() is called very quickly after the query is started, then they may both call cleanup() before Spark Jobs are started. Then sqlContext.sparkContext.cancelJobGroup(statementId) does nothing.
But then the execute thread can start the jobs, and only then get interrupted and exit through here. But then it will exit here, and no-one will cancel these jobs and they will keep running even though this execution has exited.

I think it can be fixed by:

      case e: Throwable =>
        // In any case, cancel any remaining running jobs.
        // E.g. a cancel() operation could have called cleanup() which canceled the Jobs before
        // they started.
        if (statementId != null) {
          sqlContext.sparkContext.cancelJobGroup(statementId)
        }

@AngersZhuuuu AngersZhuuuu Sep 7, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Got you point.
When cleanup().
SparkContext haven't setup jobGroup.
But execute thread start execute and setup jobGroup.
cleanup() can cancel background thread task but can't promise cancelJobGroup since it may be called before sparkContext setupJobGroup

But cancelJobGroup here seem can't stop execute() method run.

val currentState = getStatus().getState()
if (currentState == OperationState.CANCELED ||
currentState == OperationState.CLOSED ||
currentState == OperationState.FINISHED) {
Comment thread
AngersZhuuuu marked this conversation as resolved.
Outdated
// This may happen if the execution was cancelled, and then closed from another thread.
logWarning(s"Ignore exception in terminal state with $statementId: $e")
return
Comment thread
AngersZhuuuu marked this conversation as resolved.
Outdated

@juliuszsompolski juliuszsompolski Sep 3, 2019

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.

nit: you either don't need the else or the return
I would keep the return and get rid of the else to save on indentation.

} else {
logError(s"Error executing query, currentState $currentState, ", e)

@juliuszsompolski juliuszsompolski Sep 2, 2019 •
edited
Loading

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.

nit: Error executing query with $statementId

setState(OperationState.ERROR)
HiveThriftServer2.listener.onStatementError(
statementId, e.getMessage, SparkUtils.exceptionString(e))
throw e
if (e.isInstanceOf[HiveSQLException]) {
throw e.asInstanceOf[HiveSQLException]
} else {
throw new HiveSQLException("Error running query: " + e.toString, e)
}
}
// Actually do need to catch Throwable as some failures don't inherit from Exception and
// HiveServer will silently swallow them.
case e: Throwable =>
val currentState = getStatus().getState()
logError(s"Error executing query, currentState $currentState, ", e)
setState(OperationState.ERROR)
HiveThriftServer2.listener.onStatementError(
statementId, e.getMessage, SparkUtils.exceptionString(e))
throw new HiveSQLException(e.toString)
}
setState(OperationState.FINISHED)
HiveThriftServer2.listener.onStatementFinish(statementId)
synchronized {
if (!getStatus.getState.isTerminal) {
setState(OperationState.FINISHED)
HiveThriftServer2.listener.onStatementFinish(statementId)

@juliuszsompolski juliuszsompolski Sep 3, 2019

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.

Missed it: You still need to wrap it in if (!currentState.isTerminal) { otherwise it will get executed in finally after e.g. an ERROR, or you can have a race condition with cancel/close.
When I wrote to put it in the finally block, I meant that then it could be together with clearJobGroup, but the if is still needed.

@AngersZhuuuu AngersZhuuuu Sep 3, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

add if here is the most direct way, otherwise add too much control

}

@juliuszsompolski juliuszsompolski Sep 2, 2019

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.

nit: I think this could become a finally { synchronized { block; the if check will make sure that it doesn't go to finished after another state.

@AngersZhuuuu AngersZhuuuu Sep 3, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

nit: I think this could become a finally { synchronized { block; the if check will make sure that it doesn't go to finished after another state.

Reasonable, I add too much control.

}
}

override def cancel(): Unit = {
logInfo(s"Cancel '$statement' with $statementId")
cleanup(OperationState.CANCELED)
synchronized {
if (!getStatus.getState.isTerminal) {
setState(OperationState.FINISHED)
HiveThriftServer2.listener.onStatementFinish(statementId)
Comment thread
AngersZhuuuu marked this conversation as resolved.
Outdated
}
}
}

private def cleanup(state: OperationState) {
Expand Down
14 changes: 8 additions & 6 deletions ...ver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetCatalogsOperation.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,18 +50,20 @@ private[hive] class SparkGetCatalogsOperation(
statementId = UUID.randomUUID().toString
val logMsg = "Listing catalogs"
logInfo(s"$logMsg with $statementId")
setState(OperationState.RUNNING)
// Always use the latest class loader provided by executionHive's state.
val executionHiveClassLoader = sqlContext.sharedState.jarClassLoader
Thread.currentThread().setContextClassLoader(executionHiveClassLoader)

HiveThriftServer2.listener.onStatementStart(
HiveThriftServer2.listener.onStatementPrepared(
statementId,
parentSession.getSessionHandle.getSessionId.toString,
logMsg,
statementId,
parentSession.getUsername)

setState(OperationState.RUNNING)
// Always use the latest class loader provided by executionHive's state.
val executionHiveClassLoader = sqlContext.sharedState.jarClassLoader
Thread.currentThread().setContextClassLoader(executionHiveClassLoader)

HiveThriftServer2.listener.onStatementStart(statementId)

try {
if (isAuthV2Enabled) {
authorizeMetaGets(HiveOperationType.GET_CATALOGS, null)
Expand Down
13 changes: 7 additions & 6 deletions ...rver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetColumnsOperation.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,18 +71,19 @@ private[hive] class SparkGetColumnsOperation(
val cmdStr = s"catalog : $catalogName, schemaPattern : $schemaName, tablePattern : $tableName"
val logMsg = s"Listing columns '$cmdStr, columnName : $columnName'"
logInfo(s"$logMsg with $statementId")
HiveThriftServer2.listener.onStatementPrepared(
statementId,
parentSession.getSessionHandle.getSessionId.toString,
logMsg,
statementId,
parentSession.getUsername)

setState(OperationState.RUNNING)
// Always use the latest class loader provided by executionHive's state.
val executionHiveClassLoader = sqlContext.sharedState.jarClassLoader
Thread.currentThread().setContextClassLoader(executionHiveClassLoader)

HiveThriftServer2.listener.onStatementStart(
statementId,
parentSession.getSessionHandle.getSessionId.toString,
logMsg,
statementId,
parentSession.getUsername)
HiveThriftServer2.listener.onStatementStart(statementId)

val schemaPattern = convertSchemaPattern(schemaName)
val tablePattern = convertIdentifierPattern(tableName, true)
Expand Down
14 changes: 8 additions & 6 deletions ...er/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetFunctionsOperation.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,13 @@ private[hive] class SparkGetFunctionsOperation(
val cmdStr = s"catalog : $catalogName, schemaPattern : $schemaName"
val logMsg = s"Listing functions '$cmdStr, functionName : $functionName'"
logInfo(s"$logMsg with $statementId")
HiveThriftServer2.listener.onStatementPrepared(
statementId,
parentSession.getSessionHandle.getSessionId.toString,
logMsg,
statementId,
parentSession.getUsername)

setState(OperationState.RUNNING)
// Always use the latest class loader provided by executionHive's state.
val executionHiveClassLoader = sqlContext.sharedState.jarClassLoader
Expand All @@ -80,12 +87,7 @@ private[hive] class SparkGetFunctionsOperation(
authorizeMetaGets(HiveOperationType.GET_FUNCTIONS, privObjs, cmdStr)
}

HiveThriftServer2.listener.onStatementStart(
statementId,
parentSession.getSessionHandle.getSessionId.toString,
logMsg,
statementId,
parentSession.getUsername)
HiveThriftServer2.listener.onStatementStart(statementId)

try {
matchingDbs.foreach { db =>
Expand Down
14 changes: 8 additions & 6 deletions ...rver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetSchemasOperation.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,13 @@ private[hive] class SparkGetSchemasOperation(
val cmdStr = s"catalog : $catalogName, schemaPattern : $schemaName"
val logMsg = s"Listing databases '$cmdStr'"
logInfo(s"$logMsg with $statementId")
HiveThriftServer2.listener.onStatementPrepared(
statementId,
parentSession.getSessionHandle.getSessionId.toString,
logMsg,
statementId,
parentSession.getUsername)

setState(OperationState.RUNNING)
// Always use the latest class loader provided by executionHive's state.
val executionHiveClassLoader = sqlContext.sharedState.jarClassLoader
Expand All @@ -67,12 +74,7 @@ private[hive] class SparkGetSchemasOperation(
authorizeMetaGets(HiveOperationType.GET_TABLES, null, cmdStr)
}

HiveThriftServer2.listener.onStatementStart(
statementId,
parentSession.getSessionHandle.getSessionId.toString,
logMsg,
statementId,
parentSession.getUsername)
HiveThriftServer2.listener.onStatementStart(statementId)

try {
val schemaPattern = convertSchemaPattern(schemaName)
Expand Down
14 changes: 8 additions & 6 deletions ...r/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetTableTypesOperation.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,13 @@ private[hive] class SparkGetTableTypesOperation(
statementId = UUID.randomUUID().toString
val logMsg = "Listing table types"
logInfo(s"$logMsg with $statementId")
HiveThriftServer2.listener.onStatementPrepared(
statementId,
parentSession.getSessionHandle.getSessionId.toString,
logMsg,
statementId,
parentSession.getUsername)

setState(OperationState.RUNNING)
// Always use the latest class loader provided by executionHive's state.
val executionHiveClassLoader = sqlContext.sharedState.jarClassLoader
Expand All @@ -60,12 +67,7 @@ private[hive] class SparkGetTableTypesOperation(
authorizeMetaGets(HiveOperationType.GET_TABLETYPES, null)
}

HiveThriftServer2.listener.onStatementStart(
statementId,
parentSession.getSessionHandle.getSessionId.toString,
logMsg,
statementId,
parentSession.getUsername)
HiveThriftServer2.listener.onStatementStart(statementId)

try {
val tableTypes = CatalogTableType.tableTypes.map(tableTypeString).toSet
Expand Down
14 changes: 8 additions & 6 deletions ...erver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetTablesOperation.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,13 @@ private[hive] class SparkGetTablesOperation(
val tableTypesStr = if (tableTypes == null) "null" else tableTypes.asScala.mkString(",")
val logMsg = s"Listing tables '$cmdStr, tableTypes : $tableTypesStr, tableName : $tableName'"
logInfo(s"$logMsg with $statementId")
HiveThriftServer2.listener.onStatementPrepared(
statementId,
parentSession.getSessionHandle.getSessionId.toString,
logMsg,
statementId,
parentSession.getUsername)

setState(OperationState.RUNNING)
// Always use the latest class loader provided by executionHive's state.
val executionHiveClassLoader = sqlContext.sharedState.jarClassLoader
Expand All @@ -85,12 +92,7 @@ private[hive] class SparkGetTablesOperation(
authorizeMetaGets(HiveOperationType.GET_TABLES, privObjs, cmdStr)
}

HiveThriftServer2.listener.onStatementStart(
statementId,
parentSession.getSessionHandle.getSessionId.toString,
logMsg,
statementId,
parentSession.getUsername)
HiveThriftServer2.listener.onStatementStart(statementId)

try {
// Tables and views
Expand Down