Skip to content

Commit 0a354b2

Browse files
HyukjinKwonHyukjinKwon
committed
Initial UI
1 parent 525caed commit 0a354b2

9 files changed

Lines changed: 899 additions & 5 deletions

File tree

‎python/pyspark/sql/connect/proto/base_pb2.py‎

Lines changed: 13 additions & 5 deletions
Large diffs are not rendered by default.

‎python/pyspark/sql/connect/proto/base_pb2.pyi‎

Lines changed: 247 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4630,3 +4630,250 @@ class GetStatusResponse(google.protobuf.message.Message):
46304630
) -> None: ...
46314631

46324632
global___GetStatusResponse = GetStatusResponse
4633+
4634+
class ListSqlExecutionsRequest(google.protobuf.message.Message):
4635+
DESCRIPTOR: google.protobuf.descriptor.Descriptor
4636+
4637+
SESSION_ID_FIELD_NUMBER: builtins.int
4638+
USER_CONTEXT_FIELD_NUMBER: builtins.int
4639+
CLIENT_TYPE_FIELD_NUMBER: builtins.int
4640+
CLIENT_OBSERVED_SERVER_SIDE_SESSION_ID_FIELD_NUMBER: builtins.int
4641+
OFFSET_FIELD_NUMBER: builtins.int
4642+
LENGTH_FIELD_NUMBER: builtins.int
4643+
session_id: builtins.str
4644+
"""(Required) Spark session for the user identified by user_context.user_id."""
4645+
@property
4646+
def user_context(self) -> global___UserContext:
4647+
"""(Required) user_context.user_id and session_id identify a unique remote Spark session."""
4648+
client_type: builtins.str
4649+
"""(Optional) Client information for logging only; not interpreted by the server."""
4650+
client_observed_server_side_session_id: builtins.str
4651+
"""(Optional) Server-side generated idempotency key from a previous response. The server uses
4652+
this to validate that the server-side session has not changed since the client last saw it.
4653+
"""
4654+
offset: builtins.int
4655+
"""(Optional) Pagination. Negative offsets are clamped to 0. A length <= 0 or larger than the
4656+
server-side maximum is clamped by the server.
4657+
"""
4658+
length: builtins.int
4659+
def __init__(
4660+
self,
4661+
*,
4662+
session_id: builtins.str = ...,
4663+
user_context: global___UserContext | None = ...,
4664+
client_type: builtins.str | None = ...,
4665+
client_observed_server_side_session_id: builtins.str | None = ...,
4666+
offset: builtins.int = ...,
4667+
length: builtins.int = ...,
4668+
) -> None: ...
4669+
def HasField(
4670+
self,
4671+
field_name: typing_extensions.Literal[
4672+
"_client_observed_server_side_session_id",
4673+
b"_client_observed_server_side_session_id",
4674+
"_client_type",
4675+
b"_client_type",
4676+
"client_observed_server_side_session_id",
4677+
b"client_observed_server_side_session_id",
4678+
"client_type",
4679+
b"client_type",
4680+
"user_context",
4681+
b"user_context",
4682+
],
4683+
) -> builtins.bool: ...
4684+
def ClearField(
4685+
self,
4686+
field_name: typing_extensions.Literal[
4687+
"_client_observed_server_side_session_id",
4688+
b"_client_observed_server_side_session_id",
4689+
"_client_type",
4690+
b"_client_type",
4691+
"client_observed_server_side_session_id",
4692+
b"client_observed_server_side_session_id",
4693+
"client_type",
4694+
b"client_type",
4695+
"length",
4696+
b"length",
4697+
"offset",
4698+
b"offset",
4699+
"session_id",
4700+
b"session_id",
4701+
"user_context",
4702+
b"user_context",
4703+
],
4704+
) -> None: ...
4705+
@typing.overload
4706+
def WhichOneof(
4707+
self,
4708+
oneof_group: typing_extensions.Literal[
4709+
"_client_observed_server_side_session_id", b"_client_observed_server_side_session_id"
4710+
],
4711+
) -> typing_extensions.Literal["client_observed_server_side_session_id"] | None: ...
4712+
@typing.overload
4713+
def WhichOneof(
4714+
self, oneof_group: typing_extensions.Literal["_client_type", b"_client_type"]
4715+
) -> typing_extensions.Literal["client_type"] | None: ...
4716+
4717+
global___ListSqlExecutionsRequest = ListSqlExecutionsRequest
4718+
4719+
class ListSqlExecutionsResponse(google.protobuf.message.Message):
4720+
DESCRIPTOR: google.protobuf.descriptor.Descriptor
4721+
4722+
class _SqlExecutionStatus:
4723+
ValueType = typing.NewType("ValueType", builtins.int)
4724+
V: typing_extensions.TypeAlias = ValueType
4725+
4726+
class _SqlExecutionStatusEnumTypeWrapper(
4727+
google.protobuf.internal.enum_type_wrapper._EnumTypeWrapper[
4728+
ListSqlExecutionsResponse._SqlExecutionStatus.ValueType
4729+
],
4730+
builtins.type,
4731+
): # noqa: F821
4732+
DESCRIPTOR: google.protobuf.descriptor.EnumDescriptor
4733+
SQL_EXECUTION_STATUS_UNSPECIFIED: (
4734+
ListSqlExecutionsResponse._SqlExecutionStatus.ValueType
4735+
) # 0
4736+
SQL_EXECUTION_STATUS_RUNNING: ListSqlExecutionsResponse._SqlExecutionStatus.ValueType # 1
4737+
SQL_EXECUTION_STATUS_COMPLETED: ListSqlExecutionsResponse._SqlExecutionStatus.ValueType # 2
4738+
SQL_EXECUTION_STATUS_FAILED: ListSqlExecutionsResponse._SqlExecutionStatus.ValueType # 3
4739+
4740+
class SqlExecutionStatus(_SqlExecutionStatus, metaclass=_SqlExecutionStatusEnumTypeWrapper): ...
4741+
SQL_EXECUTION_STATUS_UNSPECIFIED: ListSqlExecutionsResponse.SqlExecutionStatus.ValueType # 0
4742+
SQL_EXECUTION_STATUS_RUNNING: ListSqlExecutionsResponse.SqlExecutionStatus.ValueType # 1
4743+
SQL_EXECUTION_STATUS_COMPLETED: ListSqlExecutionsResponse.SqlExecutionStatus.ValueType # 2
4744+
SQL_EXECUTION_STATUS_FAILED: ListSqlExecutionsResponse.SqlExecutionStatus.ValueType # 3
4745+
4746+
class SqlExecutionSummary(google.protobuf.message.Message):
4747+
"""Lightweight summary of a single SQL execution. Plan strings and per-node metrics are
4748+
intentionally omitted here -- list responses must stay cheap. A future GetSqlExecution RPC
4749+
can return the heavy fields for a single execution_id.
4750+
"""
4751+
4752+
DESCRIPTOR: google.protobuf.descriptor.Descriptor
4753+
4754+
EXECUTION_ID_FIELD_NUMBER: builtins.int
4755+
ROOT_EXECUTION_ID_FIELD_NUMBER: builtins.int
4756+
DESCRIPTION_FIELD_NUMBER: builtins.int
4757+
STATUS_FIELD_NUMBER: builtins.int
4758+
SUBMISSION_TIME_MS_FIELD_NUMBER: builtins.int
4759+
COMPLETION_TIME_MS_FIELD_NUMBER: builtins.int
4760+
ERROR_MESSAGE_FIELD_NUMBER: builtins.int
4761+
JOB_IDS_FIELD_NUMBER: builtins.int
4762+
execution_id: builtins.int
4763+
root_execution_id: builtins.int
4764+
description: builtins.str
4765+
status: global___ListSqlExecutionsResponse.SqlExecutionStatus.ValueType
4766+
submission_time_ms: builtins.int
4767+
completion_time_ms: builtins.int
4768+
"""Unset while the execution is still running."""
4769+
error_message: builtins.str
4770+
@property
4771+
def job_ids(
4772+
self,
4773+
) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.int]:
4774+
"""Job IDs associated with this execution. Job statuses are not included; clients can look
4775+
them up via the existing /api/v1/applications/{appId}/jobs REST endpoint if needed.
4776+
"""
4777+
def __init__(
4778+
self,
4779+
*,
4780+
execution_id: builtins.int = ...,
4781+
root_execution_id: builtins.int = ...,
4782+
description: builtins.str = ...,
4783+
status: global___ListSqlExecutionsResponse.SqlExecutionStatus.ValueType = ...,
4784+
submission_time_ms: builtins.int = ...,
4785+
completion_time_ms: builtins.int | None = ...,
4786+
error_message: builtins.str | None = ...,
4787+
job_ids: collections.abc.Iterable[builtins.int] | None = ...,
4788+
) -> None: ...
4789+
def HasField(
4790+
self,
4791+
field_name: typing_extensions.Literal[
4792+
"_completion_time_ms",
4793+
b"_completion_time_ms",
4794+
"_error_message",
4795+
b"_error_message",
4796+
"completion_time_ms",
4797+
b"completion_time_ms",
4798+
"error_message",
4799+
b"error_message",
4800+
],
4801+
) -> builtins.bool: ...
4802+
def ClearField(
4803+
self,
4804+
field_name: typing_extensions.Literal[
4805+
"_completion_time_ms",
4806+
b"_completion_time_ms",
4807+
"_error_message",
4808+
b"_error_message",
4809+
"completion_time_ms",
4810+
b"completion_time_ms",
4811+
"description",
4812+
b"description",
4813+
"error_message",
4814+
b"error_message",
4815+
"execution_id",
4816+
b"execution_id",
4817+
"job_ids",
4818+
b"job_ids",
4819+
"root_execution_id",
4820+
b"root_execution_id",
4821+
"status",
4822+
b"status",
4823+
"submission_time_ms",
4824+
b"submission_time_ms",
4825+
],
4826+
) -> None: ...
4827+
@typing.overload
4828+
def WhichOneof(
4829+
self,
4830+
oneof_group: typing_extensions.Literal["_completion_time_ms", b"_completion_time_ms"],
4831+
) -> typing_extensions.Literal["completion_time_ms"] | None: ...
4832+
@typing.overload
4833+
def WhichOneof(
4834+
self, oneof_group: typing_extensions.Literal["_error_message", b"_error_message"]
4835+
) -> typing_extensions.Literal["error_message"] | None: ...
4836+
4837+
SESSION_ID_FIELD_NUMBER: builtins.int
4838+
SERVER_SIDE_SESSION_ID_FIELD_NUMBER: builtins.int
4839+
EXECUTIONS_FIELD_NUMBER: builtins.int
4840+
TOTAL_COUNT_FIELD_NUMBER: builtins.int
4841+
session_id: builtins.str
4842+
"""Session id of the session for which executions were requested."""
4843+
server_side_session_id: builtins.str
4844+
"""Server-side generated idempotency key that the client can use to assert that the
4845+
server-side session has not changed.
4846+
"""
4847+
@property
4848+
def executions(
4849+
self,
4850+
) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[
4851+
global___ListSqlExecutionsResponse.SqlExecutionSummary
4852+
]:
4853+
"""Page of executions, in the order returned by the underlying status store."""
4854+
total_count: builtins.int
4855+
"""Total number of executions known to the server, regardless of pagination."""
4856+
def __init__(
4857+
self,
4858+
*,
4859+
session_id: builtins.str = ...,
4860+
server_side_session_id: builtins.str = ...,
4861+
executions: collections.abc.Iterable[global___ListSqlExecutionsResponse.SqlExecutionSummary]
4862+
| None = ...,
4863+
total_count: builtins.int = ...,
4864+
) -> None: ...
4865+
def ClearField(
4866+
self,
4867+
field_name: typing_extensions.Literal[
4868+
"executions",
4869+
b"executions",
4870+
"server_side_session_id",
4871+
b"server_side_session_id",
4872+
"session_id",
4873+
b"session_id",
4874+
"total_count",
4875+
b"total_count",
4876+
],
4877+
) -> None: ...
4878+
4879+
global___ListSqlExecutionsResponse = ListSqlExecutionsResponse

‎python/pyspark/sql/connect/proto/base_pb2_grpc.py‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -103,6 +103,12 @@ def __init__(self, channel):
103103
response_deserializer=spark_dot_connect_dot_base__pb2.GetStatusResponse.FromString,
104104
_registered_method=True,
105105
)
106+
self.ListSqlExecutions = channel.unary_unary(
107+
"/spark.connect.SparkConnectService/ListSqlExecutions",
108+
request_serializer=spark_dot_connect_dot_base__pb2.ListSqlExecutionsRequest.SerializeToString,
109+
response_deserializer=spark_dot_connect_dot_base__pb2.ListSqlExecutionsResponse.FromString,
110+
_registered_method=True,
111+
)
106112

107113

108114
class SparkConnectServiceServicer(object):
@@ -205,6 +211,15 @@ def GetStatus(self, request, context):
205211
context.set_details("Method not implemented!")
206212
raise NotImplementedError("Method not implemented!")
207213

214+
def ListSqlExecutions(self, request, context):
215+
"""List SQL executions visible from this session's SparkSession, drawing from the same
216+
SQLAppStatusStore that powers the driver-side SQL tab. Intended to let pure-Python
217+
Connect clients render a UI without reaching the server's HTTP UI port.
218+
"""
219+
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
220+
context.set_details("Method not implemented!")
221+
raise NotImplementedError("Method not implemented!")
222+
208223

209224
def add_SparkConnectServiceServicer_to_server(servicer, server):
210225
rpc_method_handlers = {
@@ -268,6 +283,11 @@ def add_SparkConnectServiceServicer_to_server(servicer, server):
268283
request_deserializer=spark_dot_connect_dot_base__pb2.GetStatusRequest.FromString,
269284
response_serializer=spark_dot_connect_dot_base__pb2.GetStatusResponse.SerializeToString,
270285
),
286+
"ListSqlExecutions": grpc.unary_unary_rpc_method_handler(
287+
servicer.ListSqlExecutions,
288+
request_deserializer=spark_dot_connect_dot_base__pb2.ListSqlExecutionsRequest.FromString,
289+
response_serializer=spark_dot_connect_dot_base__pb2.ListSqlExecutionsResponse.SerializeToString,
290+
),
271291
}
272292
generic_handler = grpc.method_handlers_generic_handler(
273293
"spark.connect.SparkConnectService", rpc_method_handlers
@@ -639,3 +659,33 @@ def GetStatus(
639659
metadata,
640660
_registered_method=True,
641661
)
662+
663+
@staticmethod
664+
def ListSqlExecutions(
665+
request,
666+
target,
667+
options=(),
668+
channel_credentials=None,
669+
call_credentials=None,
670+
insecure=False,
671+
compression=None,
672+
wait_for_ready=None,
673+
timeout=None,
674+
metadata=None,
675+
):
676+
return grpc.experimental.unary_unary(
677+
request,
678+
target,
679+
"/spark.connect.SparkConnectService/ListSqlExecutions",
680+
spark_dot_connect_dot_base__pb2.ListSqlExecutionsRequest.SerializeToString,
681+
spark_dot_connect_dot_base__pb2.ListSqlExecutionsResponse.FromString,
682+
options,
683+
channel_credentials,
684+
insecure,
685+
call_credentials,
686+
compression,
687+
wait_for_ready,
688+
timeout,
689+
metadata,
690+
_registered_method=True,
691+
)

‎python/pyspark/sql/connect/session.py‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -310,6 +310,12 @@ def __init__(
310310
# Set to false to prevent client.release_session on close() (testing only)
311311
self.release_session_on_close = True
312312

313+
# Best-effort: start a local client-side UI on the first Connect session
314+
# in this process. Idempotent; disabled via PYSPARK_CONNECT_UI=0.
315+
from pyspark.sql.connect.ui import _maybe_autostart_ui
316+
317+
_maybe_autostart_ui(self)
318+
313319
@classmethod
314320
def _set_default_and_active_session(cls, session: "SparkSession") -> None:
315321
"""

0 commit comments

Comments
 (0)