Skip to content

Commit 4fb9327

Browse files
cxzl25ulysses-you
authored andcommitted
[KYUUBI #2201] Show ExecutionId when running status on query engine page
### _Why are the changes needed?_ Now we can display the ExecutionId on the query engine page, but generally the sql has been finished. We can update it when SQL is running and there is an ExecutionId. close #2201, #921 ### _How was this patch tested?_ - [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible - [x] Add screenshots for manual tests if appropriate - [x] [Run test](https://kyuubi.apache.org/docs/latest/develop_tools/testing.html#running-tests) locally before make a pull request Closes #2202 from cxzl25/KYUUBI-2201. Closes #2201 ed4f8b0 [sychen] compile time 607a18c [sychen] import 9096b6a [sychen] compiledTime abb0cc0 [sychen] add sync e1273e5 [sychen] use running state to judge 3d413fe [sychen] fix style f619fe4 [sychen] use PARSING a6d91e2 [sychen] use SparkListenerSQLExecutionStart to get executionId 4a03626 [sychen] set COMPILED state early 60de793 [sychen] fix ut e2b879f [sychen] update ExecutionId Lead-authored-by: sychen <sychen@trip.com> Co-authored-by: sychen <sychen@ctrip.com> Signed-off-by: ulysses-you <ulyssesyou@apache.org>
1 parent a09ad0b commit 4fb9327

File tree

2 files changed

+30
-3
lines changed

2 files changed

+30
-3
lines changed

externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecuteStatement.scala

Lines changed: 24 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -79,9 +79,6 @@ class ExecuteStatement(
7979
// TODO: Make it configurable
8080
spark.sparkContext.addSparkListener(operationListener)
8181
result = spark.sql(statement)
82-
// TODO #921: COMPILED need consider eagerly executed commands
83-
setState(OperationState.COMPILED)
84-
debug(result.queryExecution)
8582
iter =
8683
if (incrementalCollect) {
8784
info("Execute in incremental collect mode")
@@ -97,6 +94,7 @@ class ExecuteStatement(
9794
new ArrayFetchIterator(result.take(resultMaxRows))
9895
}
9996
}
97+
setCompiledStateIfNeeded()
10098
setState(OperationState.FINISHED)
10199
} catch {
102100
onError(cancel = true)
@@ -158,4 +156,27 @@ class ExecuteStatement(
158156
EventBus.post(
159157
SparkOperationEvent(this, operationListener.getExecutionId))
160158
}
159+
160+
def setCompiledStateIfNeeded(): Unit = synchronized {
161+
if (getStatus.state == OperationState.RUNNING) {
162+
val lastAccessCompiledTime =
163+
if (result != null) {
164+
val phase = result.queryExecution.tracker.phases
165+
if (phase.contains("parsing") && phase.contains("planning")) {
166+
val compiledTime = phase("planning").endTimeMs - phase("parsing").startTimeMs
167+
lastAccessTime + compiledTime
168+
} else {
169+
0L
170+
}
171+
} else {
172+
0L
173+
}
174+
super.setState(OperationState.COMPILED)
175+
if (lastAccessCompiledTime > 0L) {
176+
lastAccessTime = lastAccessCompiledTime
177+
}
178+
EventBus.post(
179+
SparkOperationEvent(this, operationListener.getExecutionId))
180+
}
181+
}
161182
}

externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/spark/kyuubi/SQLOperationListener.scala

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ import org.apache.kyuubi.config.KyuubiConf
2929
import org.apache.kyuubi.config.KyuubiConf.{ENGINE_SPARK_SHOW_PROGRESS, ENGINE_SPARK_SHOW_PROGRESS_TIME_FORMAT, ENGINE_SPARK_SHOW_PROGRESS_UPDATE_INTERVAL}
3030
import org.apache.kyuubi.config.KyuubiReservedKeys.KYUUBI_STATEMENT_ID_KEY
3131
import org.apache.kyuubi.engine.spark.KyuubiSparkUtil.SPARK_SQL_EXECUTION_ID_KEY
32+
import org.apache.kyuubi.engine.spark.operation.ExecuteStatement
3233
import org.apache.kyuubi.operation.Operation
3334
import org.apache.kyuubi.operation.log.OperationLog
3435

@@ -86,6 +87,11 @@ class SQLOperationListener(
8687
if (executionId.isEmpty) {
8788
executionId = Option(jobStart.properties.getProperty(SPARK_SQL_EXECUTION_ID_KEY))
8889
.map(_.toLong)
90+
operation match {
91+
case executeStatement: ExecuteStatement =>
92+
executeStatement.setCompiledStateIfNeeded()
93+
case _ =>
94+
}
8995
}
9096
withOperationLog {
9197
activeJobs.add(jobId)

0 commit comments

Comments
 (0)