You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "bogao007 (via GitHub)" <gi...@apache.org> on 2024/12/11 00:17:48 UTC
[PR] [SPARK-50540][PYTHON][SS] Fix string schema for StatefulProcessorHandle [spark]
bogao007 opened a new pull request, #49138:
URL: https://github.com/apache/spark/pull/49138
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
Fix string schema for StatefulProcessorHandle, it was throwing an error before when passing the schema as String type. The way we support it is to pass the String schema to JVM using the existing TransformWithStateInPandas communication channel and do the parsing on JVM side. In the end we pass back the schema in json string to Python worker to convert it back to StructType.
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
This is to fix an issue/bug with the existing code.
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
No.
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
Updated existing test cases to include string schemas.
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
No.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50540][PYTHON][SS] Fix string schema for StatefulProcessorHandle [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on PR #49138:
URL: https://github.com/apache/spark/pull/49138#issuecomment-2533522021
After executing `dev/streaming-gen-protos.sh` locally, I noticed the following changes in `StateMessage_pb2.pyi`:
```
diff --git a/python/pyspark/sql/streaming/proto/StateMessage_pb2.pyi b/python/pyspark/sql/streaming/proto/StateMessage_pb2.pyi
index 528f8dc4dcc..6d2be94f9d8 100644
--- a/python/pyspark/sql/streaming/proto/StateMessage_pb2.pyi
+++ b/python/pyspark/sql/streaming/proto/StateMessage_pb2.pyi
@@ -54,7 +54,7 @@ class _HandleState:
class _HandleStateEnumTypeWrapper(
google.protobuf.internal.enum_type_wrapper._EnumTypeWrapper[_HandleState.ValueType],
builtins.type,
-):
+): # noqa: F821
DESCRIPTOR: google.protobuf.descriptor.EnumDescriptor
CREATED: _HandleState.ValueType # 0
INITIALIZED: _HandleState.ValueType # 1
```
Is there any local file that hasn't been committed?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50540][PYTHON][SS] Fix string schema for StatefulProcessorHandle [spark]
Posted by "bogao007 (via GitHub)" <gi...@apache.org>.
bogao007 commented on code in PR #49138:
URL: https://github.com/apache/spark/pull/49138#discussion_r1881177017
##########
python/pyspark/sql/streaming/proto/StateMessage_pb2.pyi:
##########
@@ -54,7 +54,7 @@ class _HandleState:
class _HandleStateEnumTypeWrapper(
google.protobuf.internal.enum_type_wrapper._EnumTypeWrapper[_HandleState.ValueType],
builtins.type,
-): # noqa: F821
+):
Review Comment:
@LuciferYang That's because of my local env, mine is using `mypy-protobuf` with `3.4.0`, maybe that's the cause here. It's green now after I updated my local env, thanks for the help!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50540][PYTHON][SS] Fix string schema for StatefulProcessorHandle [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #49138:
URL: https://github.com/apache/spark/pull/49138#discussion_r1883337175
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/python/TransformWithStateInPandasStateServerSuite.scala:
##########
@@ -574,6 +574,16 @@ class TransformWithStateInPandasStateServerSuite extends SparkFunSuite with Befo
verify(arrowStreamWriter).finalizeCurrentArrowBatch()
}
+ test("utils request - parse string schema") {
Review Comment:
Shall we also have the test for negative case?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50540][PYTHON][SS] Fix string schema for StatefulProcessorHandle [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on code in PR #49138:
URL: https://github.com/apache/spark/pull/49138#discussion_r1879279900
##########
python/pyspark/sql/streaming/proto/StateMessage_pb2.pyi:
##########
@@ -54,7 +54,7 @@ class _HandleState:
class _HandleStateEnumTypeWrapper(
google.protobuf.internal.enum_type_wrapper._EnumTypeWrapper[_HandleState.ValueType],
builtins.type,
-): # noqa: F821
+):
Review Comment:
It seems that `# noqa: F821` on this line shouldn't be deleted.
Is the `Python CodeGen check` environment different from GA? I
https://github.com/apache/spark/blob/58c77bab93173969e37fd5dd4619f514816e1196/.github/workflows/build_and_test.yml#L727-L730
installed the following Python 3.11 modules in the newly generated execution environment on my local machine:
```
Package Version
----------------- ---------------
black 23.9.1
click 8.1.7
mypy 1.8.0
mypy-extensions 1.0.0
mypy-protobuf 3.3.0
packaging 24.1
pathspec 0.12.1
pip 24.3.1
platformdirs 4.3.6
protobuf 5.28.3
setuptools 74.1.2
types-protobuf 5.28.3.20241030
typing_extensions 4.12.2
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50540][PYTHON][SS] Fix string schema for StatefulProcessorHandle [spark]
Posted by "bogao007 (via GitHub)" <gi...@apache.org>.
bogao007 commented on code in PR #49138:
URL: https://github.com/apache/spark/pull/49138#discussion_r1884628524
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/TransformWithStateInPandasStateServer.scala:
##########
@@ -186,6 +187,19 @@ class TransformWithStateInPandasStateServer(
handleStateVariableRequest(message.getStateVariableRequest)
case StateRequest.MethodCase.TIMERREQUEST =>
handleTimerRequest(message.getTimerRequest)
+ case StateRequest.MethodCase.UTILSREQUEST =>
+ handleUtilsRequest(message.getUtilsRequest)
+ case _ =>
+ throw new IllegalArgumentException("Invalid method call")
+ }
+ }
+
+ private[sql] def handleUtilsRequest(message: UtilsRequest): Unit = {
+ message.getMethodCase match {
+ case UtilsRequest.MethodCase.PARSESTRINGSCHEMA =>
+ val stringSchema = message.getParseStringSchema.getSchema
+ val schema = CatalystSqlParser.parseTableSchema(stringSchema)
Review Comment:
This is the error we got when passing a schema like `count inttttt`
```
pyspark.errors.exceptions.base.PySparkRuntimeError: Error parsing string schema:
[UNSUPPORTED_DATATYPE] Unsupported data type "INTTTTT". SQLSTATE: 0A000 (line 1, pos 6)
== SQL ==
count inttttt
------^^^
```
This error is already handled in the bigger try-catch block [here](https://github.com/apache/spark/blob/f979bc8d37d241d65986dfcf2105db481fef6eef/sql/core/src/main/scala/org/apache/spark/sql/execution/python/TransformWithStateInPandasStateServer.scala#L161-L166) and we throw the error in `stateful_processor_api_client._parse_string_schema()`, we could just use the same `TODO(SPARK-49233)` there.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50540][PYTHON][SS] Fix string schema for StatefulProcessorHandle [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #49138:
URL: https://github.com/apache/spark/pull/49138#discussion_r1881358277
##########
python/pyspark/sql/streaming/stateful_processor.py:
##########
@@ -274,8 +274,12 @@ def getValueState(
resets the expiration time to current processing time plus ttlDuration.
If ttl is not specified the state will never expire.
"""
- self.stateful_processor_api_client.get_value_state(state_name, schema, ttl_duration_ms)
- return ValueState(ValueStateClient(self.stateful_processor_api_client), state_name, schema)
+ schema_struct = self.stateful_processor_api_client.get_value_state(
Review Comment:
But I'm open for voices, I know this reduces one round trip and there would be people who prefers performance over cleaner code.
cc. @anishshri-db It'd be good to hear about your input.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50540][PYTHON][SS] Fix string schema for StatefulProcessorHandle [spark]
Posted by "bogao007 (via GitHub)" <gi...@apache.org>.
bogao007 commented on code in PR #49138:
URL: https://github.com/apache/spark/pull/49138#discussion_r1880619868
##########
python/pyspark/sql/streaming/proto/StateMessage_pb2.pyi:
##########
@@ -54,7 +54,7 @@ class _HandleState:
class _HandleStateEnumTypeWrapper(
google.protobuf.internal.enum_type_wrapper._EnumTypeWrapper[_HandleState.ValueType],
builtins.type,
-): # noqa: F821
+):
Review Comment:
Thanks @LuciferYang, let me play around with my local env. The `StateMessage_pb2.pyi` in the PR is exactly the same as the script generated, I guess it might be related to local env.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50540][PYTHON][SS] Fix string schema for StatefulProcessorHandle [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #49138:
URL: https://github.com/apache/spark/pull/49138#discussion_r1881339180
##########
python/pyspark/sql/streaming/stateful_processor.py:
##########
@@ -274,8 +274,12 @@ def getValueState(
resets the expiration time to current processing time plus ttlDuration.
If ttl is not specified the state will never expire.
"""
- self.stateful_processor_api_client.get_value_state(state_name, schema, ttl_duration_ms)
- return ValueState(ValueStateClient(self.stateful_processor_api_client), state_name, schema)
+ schema_struct = self.stateful_processor_api_client.get_value_state(
Review Comment:
nit: It's not very straightforward to understand we request initialization for State and we get the schema as a result. (Arguably it's already confusing we get nothing from get_XXX, though it's following the current name convention.)
Why not just have a separate method? It's not very heavyweight even we have a separate request call(s) for schema, and it's only used for string schema.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50540][PYTHON][SS] Fix string schema for StatefulProcessorHandle [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR closed pull request #49138: [SPARK-50540][PYTHON][SS] Fix string schema for StatefulProcessorHandle
URL: https://github.com/apache/spark/pull/49138
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50540][PYTHON][SS] Fix string schema for StatefulProcessorHandle [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #49138:
URL: https://github.com/apache/spark/pull/49138#discussion_r1881448824
##########
python/pyspark/sql/streaming/stateful_processor.py:
##########
@@ -274,8 +274,12 @@ def getValueState(
resets the expiration time to current processing time plus ttlDuration.
If ttl is not specified the state will never expire.
"""
- self.stateful_processor_api_client.get_value_state(state_name, schema, ttl_duration_ms)
- return ValueState(ValueStateClient(self.stateful_processor_api_client), state_name, schema)
+ schema_struct = self.stateful_processor_api_client.get_value_state(
Review Comment:
yea - probably better to have a separate API for this. As @HeartSaVioR mentioned, could we only do the conversion if string is passed ?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50540][PYTHON][SS] Fix string schema for StatefulProcessorHandle [spark]
Posted by "jingz-db (via GitHub)" <gi...@apache.org>.
jingz-db commented on code in PR #49138:
URL: https://github.com/apache/spark/pull/49138#discussion_r1882962874
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/TransformWithStateInPandasStateServer.scala:
##########
@@ -186,6 +187,19 @@ class TransformWithStateInPandasStateServer(
handleStateVariableRequest(message.getStateVariableRequest)
case StateRequest.MethodCase.TIMERREQUEST =>
handleTimerRequest(message.getTimerRequest)
+ case StateRequest.MethodCase.UTILSREQUEST =>
+ handleUtilsRequest(message.getUtilsRequest)
+ case _ =>
+ throw new IllegalArgumentException("Invalid method call")
+ }
+ }
+
+ private[sql] def handleUtilsRequest(message: UtilsRequest): Unit = {
+ message.getMethodCase match {
+ case UtilsRequest.MethodCase.PARSESTRINGSCHEMA =>
+ val stringSchema = message.getParseStringSchema.getSchema
+ val schema = CatalystSqlParser.parseTableSchema(stringSchema)
Review Comment:
Shall we have an exception handling here for parse exception thrown from `parseTableSchema` when users pass in invalid schema string? It seems currently we will always send back return code 0 even if the parsing failed.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50540][PYTHON][SS] Fix string schema for StatefulProcessorHandle [spark]
Posted by "bogao007 (via GitHub)" <gi...@apache.org>.
bogao007 commented on code in PR #49138:
URL: https://github.com/apache/spark/pull/49138#discussion_r1882976836
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/TransformWithStateInPandasStateServer.scala:
##########
@@ -186,6 +187,19 @@ class TransformWithStateInPandasStateServer(
handleStateVariableRequest(message.getStateVariableRequest)
case StateRequest.MethodCase.TIMERREQUEST =>
handleTimerRequest(message.getTimerRequest)
+ case StateRequest.MethodCase.UTILSREQUEST =>
+ handleUtilsRequest(message.getUtilsRequest)
+ case _ =>
+ throw new IllegalArgumentException("Invalid method call")
+ }
+ }
+
+ private[sql] def handleUtilsRequest(message: UtilsRequest): Unit = {
+ message.getMethodCase match {
+ case UtilsRequest.MethodCase.PARSESTRINGSCHEMA =>
+ val stringSchema = message.getParseStringSchema.getSchema
+ val schema = CatalystSqlParser.parseTableSchema(stringSchema)
Review Comment:
We already have exception handling logic on the server side, if `parseTableSchema` throws an error, it will be handled [here](https://github.com/apache/spark/blob/f979bc8d37d241d65986dfcf2105db481fef6eef/sql/core/src/main/scala/org/apache/spark/sql/execution/python/TransformWithStateInPandasStateServer.scala#L161-L166)
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50540][PYTHON][SS] Fix string schema for StatefulProcessorHandle [spark]
Posted by "bogao007 (via GitHub)" <gi...@apache.org>.
bogao007 commented on PR #49138:
URL: https://github.com/apache/spark/pull/49138#issuecomment-2533508967
@LuciferYang Do you know why the python codegen test failed [here](https://github.com/bogao007/spark/actions/runs/12268004743/job/34229117566)? I was using the `./dev/streaming-gen-protos.sh` script that you created in [this](https://github.com/apache/spark/pull/48815) PR. Do we need to update the `check-protos.py` to accommodate the changes in `StateMessage.proto`?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50540][PYTHON][SS] Fix string schema for StatefulProcessorHandle [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on code in PR #49138:
URL: https://github.com/apache/spark/pull/49138#discussion_r1879279900
##########
python/pyspark/sql/streaming/proto/StateMessage_pb2.pyi:
##########
@@ -54,7 +54,7 @@ class _HandleState:
class _HandleStateEnumTypeWrapper(
google.protobuf.internal.enum_type_wrapper._EnumTypeWrapper[_HandleState.ValueType],
builtins.type,
-): # noqa: F821
+):
Review Comment:
It seems that `# noqa: F821` on this line shouldn't be deleted.
Is the environment different from GA? I
https://github.com/apache/spark/blob/58c77bab93173969e37fd5dd4619f514816e1196/.github/workflows/build_and_test.yml#L727-L730
installed the following Python 3.11 modules in the newly generated execution environment on my local machine:
```
Package Version
----------------- ---------------
black 23.9.1
click 8.1.7
mypy 1.8.0
mypy-extensions 1.0.0
mypy-protobuf 3.3.0
packaging 24.1
pathspec 0.12.1
pip 24.3.1
platformdirs 4.3.6
protobuf 5.28.3
setuptools 74.1.2
types-protobuf 5.28.3.20241030
typing_extensions 4.12.2
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50540][PYTHON][SS] Fix string schema for StatefulProcessorHandle [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #49138:
URL: https://github.com/apache/spark/pull/49138#discussion_r1884693627
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/TransformWithStateInPandasStateServer.scala:
##########
@@ -186,6 +187,19 @@ class TransformWithStateInPandasStateServer(
handleStateVariableRequest(message.getStateVariableRequest)
case StateRequest.MethodCase.TIMERREQUEST =>
handleTimerRequest(message.getTimerRequest)
+ case StateRequest.MethodCase.UTILSREQUEST =>
+ handleUtilsRequest(message.getUtilsRequest)
+ case _ =>
+ throw new IllegalArgumentException("Invalid method call")
+ }
+ }
+
+ private[sql] def handleUtilsRequest(message: UtilsRequest): Unit = {
+ message.getMethodCase match {
+ case UtilsRequest.MethodCase.PARSESTRINGSCHEMA =>
+ val stringSchema = message.getParseStringSchema.getSchema
+ val schema = CatalystSqlParser.parseTableSchema(stringSchema)
Review Comment:
OK let's deal with this in existing TODO.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50540][PYTHON][SS] Fix string schema for StatefulProcessorHandle [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #49138:
URL: https://github.com/apache/spark/pull/49138#issuecomment-2542608825
Thanks! Merging to master.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50540][PYTHON][SS] Fix string schema for StatefulProcessorHandle [spark]
Posted by "bogao007 (via GitHub)" <gi...@apache.org>.
bogao007 commented on code in PR #49138:
URL: https://github.com/apache/spark/pull/49138#discussion_r1882936510
##########
python/pyspark/sql/streaming/list_state_client.py:
##########
@@ -28,8 +28,16 @@
class ListStateClient:
- def __init__(self, stateful_processor_api_client: StatefulProcessorApiClient) -> None:
+ def __init__(
+ self,
+ stateful_processor_api_client: StatefulProcessorApiClient,
+ schema: Union[StructType, str],
+ ) -> None:
self._stateful_processor_api_client = stateful_processor_api_client
+ if isinstance(schema, str):
Review Comment:
Make schema part of the class constructor to avoid multiple API calls for parsing.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50540][PYTHON][SS] Fix string schema for StatefulProcessorHandle [spark]
Posted by "bogao007 (via GitHub)" <gi...@apache.org>.
bogao007 commented on code in PR #49138:
URL: https://github.com/apache/spark/pull/49138#discussion_r1884629577
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/python/TransformWithStateInPandasStateServerSuite.scala:
##########
@@ -574,6 +574,16 @@ class TransformWithStateInPandasStateServerSuite extends SparkFunSuite with Befo
verify(arrowStreamWriter).finalizeCurrentArrowBatch()
}
+ test("utils request - parse string schema") {
Review Comment:
Since we use a bigger try-catch block under `run()` method, it's hard to test with. We could either have a specific try-catch block just for this call, or have a negative test on python side, which one do you prefer? cc @jingz-db
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50540][PYTHON][SS] Fix string schema for StatefulProcessorHandle [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #49138:
URL: https://github.com/apache/spark/pull/49138#discussion_r1883336380
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/TransformWithStateInPandasStateServer.scala:
##########
@@ -186,6 +187,19 @@ class TransformWithStateInPandasStateServer(
handleStateVariableRequest(message.getStateVariableRequest)
case StateRequest.MethodCase.TIMERREQUEST =>
handleTimerRequest(message.getTimerRequest)
+ case StateRequest.MethodCase.UTILSREQUEST =>
+ handleUtilsRequest(message.getUtilsRequest)
+ case _ =>
+ throw new IllegalArgumentException("Invalid method call")
+ }
+ }
+
+ private[sql] def handleUtilsRequest(message: UtilsRequest): Unit = {
+ message.getMethodCase match {
+ case UtilsRequest.MethodCase.PARSESTRINGSCHEMA =>
+ val stringSchema = message.getParseStringSchema.getSchema
+ val schema = CatalystSqlParser.parseTableSchema(stringSchema)
Review Comment:
Let's think of the outcome for users. If they make a mistake on the schema string, which error message they will get? This should be a part of error class.
I know we have TODOs for error class and I'm OK to file another one - we will need to address it before Spark 4.0, but it's not very close as of now.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50540][PYTHON][SS] Fix string schema for StatefulProcessorHandle [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #49138:
URL: https://github.com/apache/spark/pull/49138#discussion_r1884693378
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/python/TransformWithStateInPandasStateServerSuite.scala:
##########
@@ -574,6 +574,16 @@ class TransformWithStateInPandasStateServerSuite extends SparkFunSuite with Befo
verify(arrowStreamWriter).finalizeCurrentArrowBatch()
}
+ test("utils request - parse string schema") {
Review Comment:
I'm OK with having negative test on python side. I'm also OK with having follow up PR for this.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50540][PYTHON][SS] Fix string schema for StatefulProcessorHandle [spark]
Posted by "bogao007 (via GitHub)" <gi...@apache.org>.
bogao007 commented on code in PR #49138:
URL: https://github.com/apache/spark/pull/49138#discussion_r1881505021
##########
python/pyspark/sql/streaming/stateful_processor.py:
##########
@@ -274,8 +274,12 @@ def getValueState(
resets the expiration time to current processing time plus ttlDuration.
If ttl is not specified the state will never expire.
"""
- self.stateful_processor_api_client.get_value_state(state_name, schema, ttl_duration_ms)
- return ValueState(ValueStateClient(self.stateful_processor_api_client), state_name, schema)
+ schema_struct = self.stateful_processor_api_client.get_value_state(
Review Comment:
Makes sense, will update, thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org