sunchao commented on code in PR #57016:
URL: https://github.com/apache/spark/pull/57016#discussion_r3537256840
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/ui/SparkConnectServerAppStatusStore.scala:
##########
@@ -55,6 +55,23 @@ class SparkConnectServerAppStatusStore(store: KVStore) {
}
}
+ /**
+ * Look up an execution by its operation id. Unlike the primary key (the job
tag, which embeds
+ * the raw user id and may therefore contain characters that are unsafe in a
URL path segment),
+ * the operation id is always a UUID and is safe to use as a REST path
parameter. Returns the
+ * first match; the operation id is unique within a session.
+ */
+ def getExecutionByOperationId(operationId: String): Option[ExecutionInfo] = {
+ KVUtils
+ .viewToSeq(
+ store
+ .view(classOf[ExecutionInfo])
+ .index("operationId")
+ .first(operationId)
+ .last(operationId))
+ .headOption
Review Comment:
[P2] Keep the full operation identity in this lookup
`operationId` is only unique within an execution key: the manager keys
operations by `(userId, sessionId, operationId)`, and clients may supply any
valid UUID. Two sessions can therefore legally create distinct executions with
the same UUID. Both rows are returned by this secondary-index range, but
`headOption` silently selects one, so following the ID from the other
`/operations` list entry returns the wrong execution. Please include enough
user/session identity in the route and lookup (or accept the unique `jobTag` as
a query parameter), and add a duplicate-operation-ID-across-sessions test.
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/ui/SparkConnectServerAppStatusStore.scala:
##########
@@ -109,6 +126,12 @@ private[connect] class ExecutionInfo(
finishTimestamp
} else -1L
+ // Secondary index so an execution can be looked up by its operation id. The
natural key (the
+ // job tag) embeds the raw user id and is not a safe URL path segment,
whereas the operation id
+ // is always a UUID. See
SparkConnectServerAppStatusStore.getExecutionByOperationId.
+ @JsonIgnore @KVIndex("operationId")
Review Comment:
[P2] Invalidate disk stores that lack this index
Existing disk-backed History Server stores were written without
`operationId` index entries. Since `AppStatusStore.CURRENT_VERSION` is
unchanged, `loadDiskStore` reopens them without replay, while LevelDB and
RocksDB populate secondary indexes only on writes and do not backfill an added
index on open. The operation list still reads the old natural records, but
every detail lookup for those cached applications returns 404 until the store
is rebuilt. Please bump the store version or add migration/fallback behavior,
with an old-store reopen test.
##########
sql/connect/server/src/main/scala/org/apache/spark/status/api/v1/connect/ConnectResource.scala:
##########
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.status.api.v1.connect
+
+import jakarta.ws.rs._
+import jakarta.ws.rs.core.MediaType
+
+import org.apache.spark.sql.connect.ui.{ExecutionInfo, SessionInfo,
SparkConnectServerAppStatusStore}
+import org.apache.spark.status.api.v1.{BaseAppResource, NotFoundException}
+
+@Produces(Array(MediaType.APPLICATION_JSON))
+private[v1] class ConnectResource extends BaseAppResource {
+
+ @GET
+ @Path("sessions")
+ def sessionList(
+ @DefaultValue("0") @QueryParam("offset") offset: Int,
+ @DefaultValue("-1") @QueryParam("length") length: Int): Seq[SessionData]
= withUI { ui =>
+ val store = new SparkConnectServerAppStatusStore(ui.store.store)
+ paginate(store.getSessionList, offset, length).map(prepareSessionData)
+ }
+
+ @GET
+ @Path("sessions/{sessionId}")
+ def session(@PathParam("sessionId") sessionId: String): SessionData = withUI
{ ui =>
+ val store = new SparkConnectServerAppStatusStore(ui.store.store)
+ store
+ .getSession(sessionId)
+ .map(prepareSessionData)
+ .getOrElse(throw new NotFoundException("unknown session id: " +
sessionId))
+ }
+
+ @GET
+ @Path("operations")
+ def operationList(
+ @DefaultValue("0") @QueryParam("offset") offset: Int,
+ @DefaultValue("-1") @QueryParam("length") length: Int):
Seq[ExecutionData] = withUI { ui =>
+ val store = new SparkConnectServerAppStatusStore(ui.store.store)
+ paginate(store.getExecutionList, offset, length).map(prepareExecutionData)
+ }
+
+ // The operation id is a UUID and is therefore safe to use as a path
segment, whereas the job
+ // tag embeds the raw user id (which may contain characters such as '/' that
are not path-safe).
+ @GET
+ @Path("operations/{operationId}")
+ def operation(@PathParam("operationId") operationId: String): ExecutionData
= withUI { ui =>
+ val store = new SparkConnectServerAppStatusStore(ui.store.store)
+ store
+ .getExecutionByOperationId(operationId)
+ .map(prepareExecutionData)
+ .getOrElse(throw new NotFoundException("unknown operation id: " +
operationId))
+ }
+
+ private def paginate[T](list: Seq[T], offset: Int, length: Int): Seq[T] = {
+ if (length < 0) list else list.slice(offset, offset + length)
Review Comment:
[P3] There is also a concrete correctness case for moving this pagination
into the KVStore view: `offset + length` is evaluated as an `Int`, so
`?offset=1&length=2147483647` wraps the end index to `Int.MinValue` and returns
an empty page even when rows remain. Using `skip(offset).max(length)` as
suggested above avoids the overflow; please cover this boundary as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
