You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "HeartSaVioR (via GitHub)" <gi...@apache.org> on 2026/02/02 01:19:48 UTC
[PR] [SPARK-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
HeartSaVioR opened a new pull request, #54085:
URL: https://github.com/apache/spark/pull/54085
### What changes were proposed in this pull request?
This PR proposes to introduce the support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader.
To support Admission control, we propose to change `DataSourceStreamReader` interface as following:
(Created a table to perform side-by-side comparison)
| **Before** | **After** |
| :---: | :---: |
| `class DataSourceStreamReader(ABC):` | `class DataSourceStreamReader(ABC):` |
| `def initialOffset(self) -> dict` | `def initialOffset(self) -> dict` |
| `def latestOffset() -> dict` | `def latestOffset(self, start: dict, limit: ReadLimit) -> dict` |
| | `# NOTE: Optional to implement, default = ReadAllAvailable()` |
| | `def getDefaultReadLimit(self) -> ReadLimit` |
| | `# NOTE: Optional to implement, default = None` |
| | `def reportLatestOffset(self) -> Optional[dict]` |
| `def partitions(self, start: dict, end: dict) -> Sequence[InputPartition]` | `def partitions(self, start: dict, end: dict) -> Sequence[InputPartition]` |
| `@abstractmethod def read(self, partition: InputPartition) -> Union[Iterator[Tuple], Iterator["RecordBatch"]]` | `@abstractmethod def read(self, partition: InputPartition) -> Union[Iterator[Tuple], Iterator["RecordBatch"]]` |
| `def commit(self, end: dict) -> None` | `def commit(self, end: dict) -> None` |
| `def stop(self) -> None` | `def stop(self) -> None` |
The main change is following:
* The method signature for `latestOffset` is changed. The method is mandatory.
* The method `getDefaultReadLimit` is added, as optional.
* The method `reportLatestOffset` is added, as optional.
This way, new implementations would support Admission Control by default. We ensure the engine can handle the case of the old method signature, via Python’s built-in inspect module (similar to Java’s reflection). If the method “latestOffset” is implemented without parameters, we fall back to the source which does not enable admission control. For all new sources, implementing latestOffset with parameters is strongly recommended.
ReadLimit interface and built-in implementations will be available for source implementations to leverage. Built-in implementations are as follows: `ReadAllAvailable`, `ReadMinRows`, `ReadMaxRows`, `ReadMaxFiles`, `ReadMaxBytes`. We won’t support custom implementation of `ReadLimit` interface at this point since it requires major efforts and we don’t see a demand, but we can plan for it if there is a strong demand.
We do not make any change to `SimpleDataSourceStreamReader` for Admission Control, since it is designed for small data fetch and could be considered as already limiting the data. We could still add the `ReadLimit` later if we see strong demand of limiting the fetch size via the source option.
To support `Trigger.AvailableNow`, we propose to introduce a new interface as following:
```
class SupportsTriggerAvailableNow(ABC):
@abstractmethod
def prepareForTriggerAvailableNow(self) -> None
```
The above interface can be “mixed-up” with both `DataSourceStreamReader` and `SimpleDataSourceStreamReader`. It won’t work with `DataSourceStreamReader` implementations having the old method signature of `latestOffset()`, likewise mentioned above.
### Why are the changes needed?
This is to catch up with supported features in Scala DSv2 API, which we got reports from developers that missing features block them to implement some data sources.
### Does this PR introduce _any_ user-facing change?
Yes, users implementing streaming reader via python data source API will be able to add the support of Admission Control and Trigger.AvailableNow, which had been major lacks of features.
### How was this patch tested?
New UTs.
### Was this patch authored or co-authored using generative AI tooling?
Co-authored using claude-4.5-sonnet
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "allisonwang-db (via GitHub)" <gi...@apache.org>.
allisonwang-db commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2766305732
##########
python/pyspark/sql/streaming/datasource.py:
##########
@@ -0,0 +1,225 @@
+#
+# 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.
+#
+
+from abc import ABC, abstractmethod
+
+
+class ReadLimit(ABC):
+ """
+ Specifies limits on how much data to read from a streaming source when
+ determining the latest offset.
Review Comment:
Can we add more comments and examples on how to use this class?
##########
python/pyspark/sql/datasource.py:
##########
@@ -714,9 +715,37 @@ def initialOffset(self) -> dict:
messageParameters={"feature": "initialOffset"},
)
- def latestOffset(self) -> dict:
+ def latestOffset(self, start: dict, limit: ReadLimit) -> dict:
"""
- Returns the most recent offset available.
+ Returns the most recent offset available given a read limit. The start offset can be used
+ to figure out how much new data should be read given the limit.
+
+ The `start` will be provided from the return value of :meth:`initialOffset()` for
+ the very first micro-batch, and for subsequent micro-batches, the start offset is the
+ ending offset from the previous micro-batch. The source can return the `start` parameter
+ as it is, if there is no data to process.
+
+ :class:`ReadLimit` can be used by the source to limit the amount of data returned in this
+ call. The implementation should implement :meth:`getDefaultReadLimit()` to provide the
+ proper :class:`ReadLimit` if the source can limit the amount of data returned based on the
+ source options.
+
+ The engine can still call :meth:`latestOffset()` with :class:`ReadAllAvailable` even if the
+ source produces the different read limit from :meth:`getDefaultReadLimit()`, to respect the
+ semantic of trigger. The source must always respect the given readLimit provided by the
+ engine; e.g. if the readLimit is :class:`ReadAllAvailable`, the source must ignore the read
+ limit configured through options.
+
+ NOTE: Previous Spark versions didn't have start offset and read limit parameters for this
+ method. While Spark will ensure the backward compatibility for existing data sources, the
+ new data sources are strongly encouraged to implement this new method signature.
+
+ Parameters
+ ----------
+ start : dict
+ The start offset of the microbatch to continue reading from.
+ limit : :class:`ReadLimit`
+ The limit on the amount of data to be returned by this call.
Review Comment:
Let's add supported version here
##########
python/pyspark/sql/datasource_internal.py:
##########
@@ -79,28 +84,26 @@ class _SimpleStreamReaderWrapper(DataSourceStreamReader):
def __init__(self, simple_reader: SimpleDataSourceStreamReader):
self.simple_reader = simple_reader
- self.initial_offset: Optional[dict] = None
- self.current_offset: Optional[dict] = None
self.cache: List[PrefetchedCacheEntry] = []
def initialOffset(self) -> dict:
- if self.initial_offset is None:
- self.initial_offset = self.simple_reader.initialOffset()
- return self.initial_offset
-
- def latestOffset(self) -> dict:
- # when query start for the first time, use initial offset as the start offset.
- if self.current_offset is None:
- self.current_offset = self.initialOffset()
- (iter, end) = self.simple_reader.read(self.current_offset)
- self.cache.append(PrefetchedCacheEntry(self.current_offset, end, iter))
- self.current_offset = end
+ return self.simple_reader.initialOffset()
+
+ def getDefaultReadLimit(self) -> ReadLimit:
+ # We do not consider providing different read limit on simple stream reader.
+ return ReadAllAvailable()
+
+ def latestOffset(self, start: dict, limit: ReadLimit) -> dict:
+ assert start is not None, "start offset should not be None"
+ assert isinstance(
+ limit, ReadAllAvailable
+ ), "simple stream reader does not support read limit"
Review Comment:
Does that mean we can't use availableNow with simple streaming reader?
##########
python/pyspark/sql/streaming/datasource.py:
##########
@@ -0,0 +1,225 @@
+#
+# 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.
+#
+
+from abc import ABC, abstractmethod
+
+
+class ReadLimit(ABC):
+ """
+ Specifies limits on how much data to read from a streaming source when
+ determining the latest offset.
+ """
+
+ @classmethod
+ @abstractmethod
+ def type_name(cls) -> str:
+ """
+ The name of this :class:`ReadLimit` type. This is used to register the type into registry.
+
+ Returns
+ -------
+ str
+ The name of this :class:`ReadLimit` type.
+ """
+ pass
+
+ @classmethod
+ @abstractmethod
+ def load(cls, params: dict) -> "ReadLimit":
+ """
+ Create an instance of :class:`ReadLimit` from parameters.
+
+ Parameter
+ ---------
+ params : dict
+ The parameters to create the :class:`ReadLimit`. type name isn't included.
Review Comment:
ditto can we add more examples?
##########
python/pyspark/sql/datasource.py:
##########
@@ -714,9 +715,37 @@ def initialOffset(self) -> dict:
messageParameters={"feature": "initialOffset"},
)
- def latestOffset(self) -> dict:
+ def latestOffset(self, start: dict, limit: ReadLimit) -> dict:
"""
- Returns the most recent offset available.
+ Returns the most recent offset available given a read limit. The start offset can be used
+ to figure out how much new data should be read given the limit.
+
+ The `start` will be provided from the return value of :meth:`initialOffset()` for
+ the very first micro-batch, and for subsequent micro-batches, the start offset is the
+ ending offset from the previous micro-batch. The source can return the `start` parameter
+ as it is, if there is no data to process.
+
+ :class:`ReadLimit` can be used by the source to limit the amount of data returned in this
+ call. The implementation should implement :meth:`getDefaultReadLimit()` to provide the
+ proper :class:`ReadLimit` if the source can limit the amount of data returned based on the
+ source options.
+
+ The engine can still call :meth:`latestOffset()` with :class:`ReadAllAvailable` even if the
+ source produces the different read limit from :meth:`getDefaultReadLimit()`, to respect the
+ semantic of trigger. The source must always respect the given readLimit provided by the
+ engine; e.g. if the readLimit is :class:`ReadAllAvailable`, the source must ignore the read
+ limit configured through options.
+
+ NOTE: Previous Spark versions didn't have start offset and read limit parameters for this
+ method. While Spark will ensure the backward compatibility for existing data sources, the
+ new data sources are strongly encouraged to implement this new method signature.
Review Comment:
Let's add this NOTE below as a docstring section?
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #54085:
URL: https://github.com/apache/spark/pull/54085#issuecomment-3832453677
## JIRA Issue Information
=== New Feature SPARK-55304 ===
Summary: Introduce Admission Control and Trigger.AvailableNow into Python Data Source - reader
Assignee: None
Status: Open
Affected: ["4.2.0"]
---
*This comment was automatically generated by GitHub Actions*
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "gaogaotiantian (via GitHub)" <gi...@apache.org>.
gaogaotiantian commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2771623656
##########
python/pyspark/sql/streaming/python_streaming_source_runner.py:
##########
@@ -116,6 +138,77 @@ def send_batch_func(
write_int(EMPTY_PYARROW_RECORD_BATCHES, outfile)
+def check_support_func(reader: DataSourceStreamReader, outfile: IO) -> None:
+ support_flags = 0
+ if isinstance(reader, _SimpleStreamReaderWrapper):
Review Comment:
Yeah if the API design is already approved we don't need to change it.
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2767805562
##########
python/pyspark/sql/streaming/python_streaming_source_runner.py:
##########
@@ -116,6 +138,77 @@ def send_batch_func(
write_int(EMPTY_PYARROW_RECORD_BATCHES, outfile)
+def check_support_func(reader: DataSourceStreamReader, outfile: IO) -> None:
+ support_flags = 0
+ if isinstance(reader, _SimpleStreamReaderWrapper):
Review Comment:
While I get how your proposal works, I'd argue that this is just another workaround for doing something which we can't do from inheritance.
For example, you said SupportsTriggerAvailableNow isn't the important thing - if we come to non-python language which has strong typing, explicitly checking the type is mostly required except the hacky way like reflection in Java - it is not a best practice. We do inspect for latestOffset but this is a last resort approach, because python does not support method overloading so we actually can't do the same evolution of API we did from Scala/Java.
(It's still possible we can introduce a separate interface to achieve similar thing but it is not about overloading - it's just that the latest definition will just override the prior definition for the same name of method.)
Though I understand that in python it's duck typing and the type system is crazy loose and the availability of function/method simply matters.
Maybe you might try to give a better approach for pythonic thinking. It's just that both breaks inheritance and the proposal seems to break more than what I do. I agree if you want to say this isn't pythonic, but then I could say this is one of the pattern we heavily leverage in Scala, we call it as pattern matching.
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2771585547
##########
python/pyspark/sql/streaming/python_streaming_source_runner.py:
##########
@@ -116,6 +138,77 @@ def send_batch_func(
write_int(EMPTY_PYARROW_RECORD_BATCHES, outfile)
+def check_support_func(reader: DataSourceStreamReader, outfile: IO) -> None:
+ support_flags = 0
+ if isinstance(reader, _SimpleStreamReaderWrapper):
Review Comment:
mixin is public and yes we will have to support this in future. Though it's safer to not revisit the direction of the API design as a whole at this point, despite we may have to pay the cost of inlining it when we finally decide to do so. I think it is still cheaper to go forward than discussing about divergence and take it back if necessary.
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2767810759
##########
python/pyspark/sql/streaming/python_streaming_source_runner.py:
##########
@@ -116,6 +138,77 @@ def send_batch_func(
write_int(EMPTY_PYARROW_RECORD_BATCHES, outfile)
+def check_support_func(reader: DataSourceStreamReader, outfile: IO) -> None:
+ support_flags = 0
+ if isinstance(reader, _SimpleStreamReaderWrapper):
Review Comment:
> we should not claim that _SimpleStreamReaderWrapper is a DataSourceStreamReader
This breaks the structure of interface we design. If _SimpleStreamReaderWrapper isn't a DataSourceStreamReader, what it would be?
##########
python/pyspark/sql/streaming/python_streaming_source_runner.py:
##########
@@ -116,6 +138,77 @@ def send_batch_func(
write_int(EMPTY_PYARROW_RECORD_BATCHES, outfile)
+def check_support_func(reader: DataSourceStreamReader, outfile: IO) -> None:
+ support_flags = 0
+ if isinstance(reader, _SimpleStreamReaderWrapper):
Review Comment:
> we should not claim that _SimpleStreamReaderWrapper is a DataSourceStreamReader
This breaks the structure of interface we design. If _SimpleStreamReaderWrapper isn't a DataSourceStreamReader, what it would be?
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "huanliwang-db (via GitHub)" <gi...@apache.org>.
huanliwang-db commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2757524251
##########
python/pyspark/sql/datasource.py:
##########
@@ -714,9 +715,37 @@ def initialOffset(self) -> dict:
messageParameters={"feature": "initialOffset"},
)
- def latestOffset(self) -> dict:
+ def latestOffset(self, start: dict, limit: ReadLimit) -> dict:
"""
- Returns the most recent offset available.
+ Returns the most recent offset available given a read limit. The start offset can be used
+ to figure out how much new data should be read given the limit.
+
+ The `start` will be provided from the return value of :meth:`initialOffset()` for
+ the very first micro-batch, and the offset continues from the last micro-batch for the
+ following. The source can return the same offset as start offset if there is no data to
+ process.
+
+ :class:`ReadLimit` can be used by the source to limit the amount of data returned in this
+ call. The implementation should implement :meth:`getDefaultReadLimit()` to provide the
+ proper :class:`ReadLimit` if the source can limit the amount of data returned based on the
+ source options.
+
+ The engine can still call :meth:`latestOffset()` with :class:`ReadAllAvailable` even if the
+ source produces the different read limit from :meth:`getDefaultReadLimit()`, to respect the
+ semantic of trigger. The source must always respect the given readLimit provided by the
+ engine; e.g. if the readLimit is :class:`ReadAllAvailable`, the source must ignore the read
+ limit configured through options.
Review Comment:
nit: maybe it will be more clear if we can provide an example in which the engine provides a different read limit than the configured one
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #54085:
URL: https://github.com/apache/spark/pull/54085#issuecomment-3857597144
https://github.com/HeartSaVioR/spark/actions/runs/21731506787/job/62687169102
I don't think failure is relevant (it sounds like be broken permanently?) except linter - I'll run the linter and fix it, and merge it given I've tried several times on making CI be happy.
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2771585547
##########
python/pyspark/sql/streaming/python_streaming_source_runner.py:
##########
@@ -116,6 +138,77 @@ def send_batch_func(
write_int(EMPTY_PYARROW_RECORD_BATCHES, outfile)
+def check_support_func(reader: DataSourceStreamReader, outfile: IO) -> None:
+ support_flags = 0
+ if isinstance(reader, _SimpleStreamReaderWrapper):
Review Comment:
mixin is public and yes we will have to support this in future. Though it's safer to not revisit the direction of the API design as a whole, despite we may have to pay the cost of inlining it when we finally decide to do so.
##########
python/pyspark/sql/streaming/python_streaming_source_runner.py:
##########
@@ -116,6 +138,77 @@ def send_batch_func(
write_int(EMPTY_PYARROW_RECORD_BATCHES, outfile)
+def check_support_func(reader: DataSourceStreamReader, outfile: IO) -> None:
+ support_flags = 0
+ if isinstance(reader, _SimpleStreamReaderWrapper):
Review Comment:
mixin is public and yes we will have to support this in future. Though it's safer to not revisit the direction of the API design as a whole at this point, despite we may have to pay the cost of inlining it when we finally decide to do so.
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2761865415
##########
python/pyspark/sql/datasource_internal.py:
##########
@@ -144,3 +141,28 @@ def read(
self, input_partition: SimpleInputPartition # type: ignore[override]
) -> Iterator[Tuple]:
return self.simple_reader.readBetweenOffsets(input_partition.start, input_partition.end)
+
+
+class ReadLimitRegistry:
+ def __init__(self) -> None:
+ self._registry: Dict[str, Type[ReadLimit]] = {}
+ # Register built-in ReadLimit types
+ self.__register(ReadAllAvailable.type_name(), ReadAllAvailable)
+ self.__register(ReadMinRows.type_name(), ReadMinRows)
+ self.__register(ReadMaxRows.type_name(), ReadMaxRows)
+ self.__register(ReadMaxBytes.type_name(), ReadMaxBytes)
Review Comment:
Ouch nice finding!
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2771585547
##########
python/pyspark/sql/streaming/python_streaming_source_runner.py:
##########
@@ -116,6 +138,77 @@ def send_batch_func(
write_int(EMPTY_PYARROW_RECORD_BATCHES, outfile)
+def check_support_func(reader: DataSourceStreamReader, outfile: IO) -> None:
+ support_flags = 0
+ if isinstance(reader, _SimpleStreamReaderWrapper):
Review Comment:
mixin is public and yes we will have to support this in future. Though it's safer to not revisit the direction of the API design as a whole at this point, despite we may have to pay the cost of inlining it when we finally decide to do so. I think it is still cheaper to go forward than discussing about divergence and take it back if necessary.
Even if we change it later, mitigation wouldn't be super complicated if I understand correctly.
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2767040089
##########
python/pyspark/sql/streaming/datasource.py:
##########
@@ -0,0 +1,225 @@
+#
+# 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.
+#
+
+from abc import ABC, abstractmethod
+
+
+class ReadLimit(ABC):
+ """
+ Specifies limits on how much data to read from a streaming source when
+ determining the latest offset.
Review Comment:
The documentation should be covered in `latestOffset()` and `getDefaultReadLimit()` in DataSourceStreamReader.
Since devs have to use the built-in read limit implementations at this point, I'm going to enumerate the built-in implementations here so that they can understand which classes are available.
##########
python/pyspark/sql/datasource.py:
##########
@@ -714,9 +715,37 @@ def initialOffset(self) -> dict:
messageParameters={"feature": "initialOffset"},
)
- def latestOffset(self) -> dict:
+ def latestOffset(self, start: dict, limit: ReadLimit) -> dict:
"""
- Returns the most recent offset available.
+ Returns the most recent offset available given a read limit. The start offset can be used
+ to figure out how much new data should be read given the limit.
+
+ The `start` will be provided from the return value of :meth:`initialOffset()` for
+ the very first micro-batch, and for subsequent micro-batches, the start offset is the
+ ending offset from the previous micro-batch. The source can return the `start` parameter
+ as it is, if there is no data to process.
+
+ :class:`ReadLimit` can be used by the source to limit the amount of data returned in this
+ call. The implementation should implement :meth:`getDefaultReadLimit()` to provide the
+ proper :class:`ReadLimit` if the source can limit the amount of data returned based on the
+ source options.
+
+ The engine can still call :meth:`latestOffset()` with :class:`ReadAllAvailable` even if the
+ source produces the different read limit from :meth:`getDefaultReadLimit()`, to respect the
+ semantic of trigger. The source must always respect the given readLimit provided by the
+ engine; e.g. if the readLimit is :class:`ReadAllAvailable`, the source must ignore the read
+ limit configured through options.
+
+ NOTE: Previous Spark versions didn't have start offset and read limit parameters for this
+ method. While Spark will ensure the backward compatibility for existing data sources, the
+ new data sources are strongly encouraged to implement this new method signature.
Review Comment:
You meant moving out of doc comment? I'm OK with it.
##########
python/pyspark/sql/streaming/datasource.py:
##########
@@ -0,0 +1,225 @@
+#
+# 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.
+#
+
+from abc import ABC, abstractmethod
+
+
+class ReadLimit(ABC):
+ """
+ Specifies limits on how much data to read from a streaming source when
+ determining the latest offset.
+ """
+
+ @classmethod
+ @abstractmethod
+ def type_name(cls) -> str:
+ """
+ The name of this :class:`ReadLimit` type. This is used to register the type into registry.
+
+ Returns
+ -------
+ str
+ The name of this :class:`ReadLimit` type.
+ """
+ pass
+
+ @classmethod
+ @abstractmethod
+ def load(cls, params: dict) -> "ReadLimit":
+ """
+ Create an instance of :class:`ReadLimit` from parameters.
+
+ Parameter
+ ---------
+ params : dict
+ The parameters to create the :class:`ReadLimit`. type name isn't included.
Review Comment:
Class methods in ReadLimit aren't something user facing ones. They are Spark internal ones. Do we still need examples?
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "huanliwang-db (via GitHub)" <gi...@apache.org>.
huanliwang-db commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2757465701
##########
python/pyspark/sql/streaming/datasource.py:
##########
@@ -0,0 +1,225 @@
+#
+# 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.
+#
+
+from abc import ABC, abstractmethod
+
+
+class ReadLimit(ABC):
+ """
+ Specifies limits on how much data to read from a streaming source when
+ determining the latest offset.
+ """
+
+ @classmethod
+ @abstractmethod
+ def type_name(cls) -> str:
+ """
+ The name of this :class:`ReadLimit` type. This is used to register the type into registry.
+
+ Returns
+ -------
+ str
+ The name of this :class:`ReadLimit` type.
+ """
+ pass
+
+ @classmethod
+ @abstractmethod
+ def load(cls, params: dict) -> "ReadLimit":
+ """
+ Create an instance of :class:`ReadLimit` from parameters.
+
+ Parameter
+ ---------
+ params : dict
+ The parameters to create the :class:`ReadLimit`. type name isn't included.
+
+ Returns
+ -------
+ ReadLimit
+ The created :class:`ReadLimit` instance.
+ """
+ pass
+
+ def dump(self) -> dict:
+ """
+ Method to serialize this :class:`ReadLimit` instance. Implementations of :class:`ReadLimit`
+ are expected to not implement this method directly and rather implement the
+ :meth:`_dump()` method.
+
+ Returns
+ -------
+ dict
+ A dictionary containing the serialized parameters of this :class:`ReadLimit`,
+ including the type name.
+ """
+ params = self._dump()
+ params.update({"type": self.type_name()})
+ return params
+
+ @abstractmethod
+ def _dump(self) -> dict:
+ """
+ Method to serialize this :class:`ReadLimit` instance. Implementations of :class:`ReadLimit`
+ are expected to implement this method to handle their specific parameters. type name will
+ be handled in the :meth:`dump()` method.
+
+ Returns
+ -------
+ dict
+ A dictionary containing the serialized parameters of this :class:`ReadLimit`,
+ excluding the type name.
+ """
+ pass
+
+
+class ReadAllAvailable(ReadLimit):
+ """
+ A :class:`ReadLimit` that indicates to read all available data, regardless of the given source
+ options.
+ """
+
+ @classmethod
+ def type_name(cls) -> str:
+ return "ReadAllAvailable"
+
+ @classmethod
+ def load(cls, params: dict) -> "ReadAllAvailable":
+ return ReadAllAvailable()
+
+ def _dump(self) -> dict:
+ return {}
+
+
+class ReadMinRows(ReadLimit):
+ """
+ A :class:`ReadLimit` that indicates to read minimum N rows. If there is less than N rows
+ available for read, the source should skip producing a new offset to read and wait until more
+ data arrives.
+
+ Note that the semantic does not work properly with Trigger.AvailableNow since the source
+ may end up waiting forever for more data to arrive. It is the source's responsibility to
+ handle this case properly.
+ """
+
+ def __init__(self, min_rows: int) -> None:
+ self.min_rows = min_rows
+
+ @classmethod
+ def type_name(cls) -> str:
+ return "ReadMinRows"
+
+ @classmethod
+ def load(cls, params: dict) -> "ReadMinRows":
+ return ReadMinRows(params["min_rows"])
+
+ def _dump(self) -> dict:
+ return {"min_rows": self.min_rows}
+
+
+class ReadMaxRows(ReadLimit):
+ """
+ A :class:`ReadLimit` that indicates to read maximum N rows. The source should not read more
+ than N rows when determining the latest offset.
+ """
+
+ def __init__(self, max_rows: int) -> None:
+ self.max_rows = max_rows
+
+ @classmethod
+ def type_name(cls) -> str:
+ return "ReadMaxRows"
+
+ @classmethod
+ def load(cls, params: dict) -> "ReadMaxRows":
+ return ReadMaxRows(params["max_rows"])
+
+ def _dump(self) -> dict:
+ return {"max_rows": self.max_rows}
+
+
+class ReadMaxFiles(ReadLimit):
+ """
+ A :class:`ReadLimit` that indicates to read maximum N files. The source should not read more
+ than N files when determining the latest offset.
+ """
+
+ def __init__(self, max_files: int) -> None:
+ self.max_files = max_files
+
+ @classmethod
+ def type_name(cls) -> str:
+ return "ReadMaxFiles"
+
+ @classmethod
+ def load(cls, params: dict) -> "ReadMaxFiles":
+ return ReadMaxFiles(params["max_files"])
+
+ def _dump(self) -> dict:
+ return {"max_files": self.max_files}
+
+
+class ReadMaxBytes(ReadLimit):
+ """
+ A :class:`ReadLimit` that indicates to read maximum N bytes. The source should not read more
+ than N bytes when determining the latest offset.
+ """
+
+ def __init__(self, max_bytes: int) -> None:
+ self.max_bytes = max_bytes
+
+ @classmethod
+ def type_name(cls) -> str:
+ return "ReadMaxBytes"
+
+ @classmethod
+ def load(cls, params: dict) -> "ReadMaxBytes":
+ return ReadMaxBytes(params["max_bytes"])
+
+ def _dump(self) -> dict:
+ return {"max_bytes": self.max_bytes}
+
+
+class SupportsTriggerAvailableNow(ABC):
+ """
+ A mixin interface for streaming sources that support Trigger.AvailableNow. This interface can
+ be added to both :class:`DataSourceStreamReader` and :class:`SimpleDataSourceStreamReader`.
+ """
+
+ @abstractmethod
+ def prepareForTriggerAvailableNow(self) -> None:
+ """
+ This will be called at the beginning of streaming queries with Trigger.AvailableNow, to let
+ the source record the offset for the current latest data at the time (a.k.a the target
+ offset for the query). The source must behave as if there is no new data coming in after
+ the target offset, i.e., the source must not return an offset higher than the target offset
+ when :meth:`DataSourceStreamReader.latestOffset()` is called.
+
+ The source can extend the semantic of "current latest data" based on its own logic, but the
+ extended semantic must not violate the expectation that the source will not read any data
+ which is added later than the time this method has called.
+
+ Note that it is the source's responsibility to ensure that calling
+ :meth:`DataSourceStreamReader.latestOffset()` or :meth:`SimpleDataSourceStreamReader.read()`
+ after calling this method will eventually reach the target offset, and finally returns the
+ same offset as given start parameter, to indicate that there is no more data to read. This
+ includes the case where the query is restarted and the source is asked to read from the
Review Comment:
sorry, why do we need to handle query restart case here? shouldn't the trigger available now get the new end offset after query restart?
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2761864050
##########
python/pyspark/sql/datasource_internal.py:
##########
@@ -144,3 +141,28 @@ def read(
self, input_partition: SimpleInputPartition # type: ignore[override]
) -> Iterator[Tuple]:
return self.simple_reader.readBetweenOffsets(input_partition.start, input_partition.end)
+
+
+class ReadLimitRegistry:
+ def __init__(self) -> None:
+ self._registry: Dict[str, Type[ReadLimit]] = {}
+ # Register built-in ReadLimit types
+ self.__register(ReadAllAvailable.type_name(), ReadAllAvailable)
+ self.__register(ReadMinRows.type_name(), ReadMinRows)
+ self.__register(ReadMaxRows.type_name(), ReadMaxRows)
+ self.__register(ReadMaxBytes.type_name(), ReadMaxBytes)
+
+ def __register(self, type_name: str, read_limit_type: Type["ReadLimit"]) -> None:
+ if type_name in self._registry:
+ raise PySparkException(f"ReadLimit type '{type_name}' is already registered.")
+
+ self._registry[type_name] = read_limit_type
+
+ def get(self, type_name: str, params: dict) -> ReadLimit:
+ read_limit_type = self._registry[type_name]
Review Comment:
That's not expected to happen, but KeyError is definitely not preferable. I'll probably throw the better exception - if we have internal exception in PySpark than I'll use that.
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2767805562
##########
python/pyspark/sql/streaming/python_streaming_source_runner.py:
##########
@@ -116,6 +138,77 @@ def send_batch_func(
write_int(EMPTY_PYARROW_RECORD_BATCHES, outfile)
+def check_support_func(reader: DataSourceStreamReader, outfile: IO) -> None:
+ support_flags = 0
+ if isinstance(reader, _SimpleStreamReaderWrapper):
Review Comment:
While I get how your proposal works, I'd argue that this is just another workaround for doing something which we can't do from inheritance.
For example, you said SupportsTriggerAvailableNow isn't the important thing - if we come to non-python language which has strong typing, explicitly checking the type is mostly required except the hacky way like reflection in Java - it is not a best practice. We do inspect for latestOffset but this is a last resort approach, because python does not support method overloading so we actually can't do the same evolution of API we did from Scala/Java.
(It's still possible we can introduce a separate interface to achieve similar thing but it is not about overloading - it's just that the latest definition will just override the prior definition for the same name of method.)
Though I understand that in python it's duck typing and the type system is crazy loose and the availability of function/method simply matters.
Maybe you might have tried to give a better approach for pythonic thinking. It's just that both breaks inheritance and the proposal seems to break more than what I do. I agree if you want to say this isn't pythonic, but then I could say this is one of the pattern we heavily leverage in Scala, we call it as pattern matching. It's just that there is no language level support so I have to do this manually.
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "huanliwang-db (via GitHub)" <gi...@apache.org>.
huanliwang-db commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2757509112
##########
python/pyspark/sql/datasource.py:
##########
@@ -714,9 +715,37 @@ def initialOffset(self) -> dict:
messageParameters={"feature": "initialOffset"},
)
- def latestOffset(self) -> dict:
+ def latestOffset(self, start: dict, limit: ReadLimit) -> dict:
"""
- Returns the most recent offset available.
+ Returns the most recent offset available given a read limit. The start offset can be used
+ to figure out how much new data should be read given the limit.
+
+ The `start` will be provided from the return value of :meth:`initialOffset()` for
+ the very first micro-batch, and the offset continues from the last micro-batch for the
+ following. The source can return the same offset as start offset if there is no data to
Review Comment:
"The source can return the same offset as start offset if there is no data to process"
I feel this is also a bit confusing, which "start" offset you are referring here?
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2767465932
##########
python/pyspark/sql/streaming/datasource.py:
##########
@@ -0,0 +1,234 @@
+#
+# 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.
+#
+
+from abc import ABC, abstractmethod
+
+
+class ReadLimit(ABC):
Review Comment:
It might be the case, but when we decide to extend this, would it be a one way door and we will never be able to support user-defined ReadLimit? If not I'm happy to incorporate the feedback.
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "huanliwang-db (via GitHub)" <gi...@apache.org>.
huanliwang-db commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2760537396
##########
python/pyspark/sql/datasource_internal.py:
##########
@@ -144,3 +141,28 @@ def read(
self, input_partition: SimpleInputPartition # type: ignore[override]
) -> Iterator[Tuple]:
return self.simple_reader.readBetweenOffsets(input_partition.start, input_partition.end)
+
+
+class ReadLimitRegistry:
+ def __init__(self) -> None:
+ self._registry: Dict[str, Type[ReadLimit]] = {}
+ # Register built-in ReadLimit types
+ self.__register(ReadAllAvailable.type_name(), ReadAllAvailable)
+ self.__register(ReadMinRows.type_name(), ReadMinRows)
+ self.__register(ReadMaxRows.type_name(), ReadMaxRows)
+ self.__register(ReadMaxBytes.type_name(), ReadMaxBytes)
+
+ def __register(self, type_name: str, read_limit_type: Type["ReadLimit"]) -> None:
+ if type_name in self._registry:
+ raise PySparkException(f"ReadLimit type '{type_name}' is already registered.")
+
+ self._registry[type_name] = read_limit_type
+
+ def get(self, type_name: str, params: dict) -> ReadLimit:
+ read_limit_type = self._registry[type_name]
Review Comment:
I am not quite familiar with python, but will this throws out exception if `type_name` doesn't exist?
##########
python/pyspark/sql/datasource_internal.py:
##########
@@ -144,3 +141,28 @@ def read(
self, input_partition: SimpleInputPartition # type: ignore[override]
) -> Iterator[Tuple]:
return self.simple_reader.readBetweenOffsets(input_partition.start, input_partition.end)
+
+
+class ReadLimitRegistry:
+ def __init__(self) -> None:
+ self._registry: Dict[str, Type[ReadLimit]] = {}
+ # Register built-in ReadLimit types
+ self.__register(ReadAllAvailable.type_name(), ReadAllAvailable)
+ self.__register(ReadMinRows.type_name(), ReadMinRows)
+ self.__register(ReadMaxRows.type_name(), ReadMaxRows)
+ self.__register(ReadMaxBytes.type_name(), ReadMaxBytes)
Review Comment:
missing `ReadMaxFiles`?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/python/streaming/PythonStreamingDataSourceSuite.scala:
##########
Review Comment:
is it possible to add a test case where `latestOffset` returns the same offset?
##########
python/pyspark/sql/streaming/datasource.py:
##########
@@ -0,0 +1,225 @@
+#
+# 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.
+#
+
+from abc import ABC, abstractmethod
+
+
+class ReadLimit(ABC):
+ """
+ Specifies limits on how much data to read from a streaming source when
+ determining the latest offset.
+ """
+
+ @classmethod
+ @abstractmethod
+ def type_name(cls) -> str:
+ """
+ The name of this :class:`ReadLimit` type. This is used to register the type into registry.
+
+ Returns
+ -------
+ str
+ The name of this :class:`ReadLimit` type.
+ """
+ pass
+
+ @classmethod
+ @abstractmethod
+ def load(cls, params: dict) -> "ReadLimit":
+ """
+ Create an instance of :class:`ReadLimit` from parameters.
+
+ Parameter
+ ---------
+ params : dict
+ The parameters to create the :class:`ReadLimit`. type name isn't included.
+
+ Returns
+ -------
+ ReadLimit
+ The created :class:`ReadLimit` instance.
+ """
+ pass
+
+ def dump(self) -> dict:
+ """
+ Method to serialize this :class:`ReadLimit` instance. Implementations of :class:`ReadLimit`
+ are expected to not implement this method directly and rather implement the
+ :meth:`_dump()` method.
+
+ Returns
+ -------
+ dict
+ A dictionary containing the serialized parameters of this :class:`ReadLimit`,
+ including the type name.
+ """
+ params = self._dump()
+ params.update({"type": self.type_name()})
+ return params
+
+ @abstractmethod
+ def _dump(self) -> dict:
+ """
+ Method to serialize this :class:`ReadLimit` instance. Implementations of :class:`ReadLimit`
+ are expected to implement this method to handle their specific parameters. type name will
+ be handled in the :meth:`dump()` method.
+
+ Returns
+ -------
+ dict
+ A dictionary containing the serialized parameters of this :class:`ReadLimit`,
+ excluding the type name.
+ """
+ pass
+
+
+class ReadAllAvailable(ReadLimit):
+ """
+ A :class:`ReadLimit` that indicates to read all available data, regardless of the given source
+ options.
+ """
+
+ @classmethod
+ def type_name(cls) -> str:
+ return "ReadAllAvailable"
+
+ @classmethod
+ def load(cls, params: dict) -> "ReadAllAvailable":
+ return ReadAllAvailable()
+
+ def _dump(self) -> dict:
+ return {}
+
+
+class ReadMinRows(ReadLimit):
+ """
+ A :class:`ReadLimit` that indicates to read minimum N rows. If there is less than N rows
+ available for read, the source should skip producing a new offset to read and wait until more
+ data arrives.
+
+ Note that the semantic does not work properly with Trigger.AvailableNow since the source
+ may end up waiting forever for more data to arrive. It is the source's responsibility to
+ handle this case properly.
+ """
+
+ def __init__(self, min_rows: int) -> None:
+ self.min_rows = min_rows
+
+ @classmethod
+ def type_name(cls) -> str:
+ return "ReadMinRows"
+
+ @classmethod
+ def load(cls, params: dict) -> "ReadMinRows":
+ return ReadMinRows(params["min_rows"])
+
+ def _dump(self) -> dict:
+ return {"min_rows": self.min_rows}
+
+
+class ReadMaxRows(ReadLimit):
+ """
+ A :class:`ReadLimit` that indicates to read maximum N rows. The source should not read more
+ than N rows when determining the latest offset.
+ """
+
+ def __init__(self, max_rows: int) -> None:
+ self.max_rows = max_rows
+
+ @classmethod
+ def type_name(cls) -> str:
+ return "ReadMaxRows"
+
+ @classmethod
+ def load(cls, params: dict) -> "ReadMaxRows":
+ return ReadMaxRows(params["max_rows"])
+
+ def _dump(self) -> dict:
+ return {"max_rows": self.max_rows}
+
+
+class ReadMaxFiles(ReadLimit):
+ """
+ A :class:`ReadLimit` that indicates to read maximum N files. The source should not read more
+ than N files when determining the latest offset.
+ """
+
+ def __init__(self, max_files: int) -> None:
+ self.max_files = max_files
+
+ @classmethod
+ def type_name(cls) -> str:
+ return "ReadMaxFiles"
+
+ @classmethod
+ def load(cls, params: dict) -> "ReadMaxFiles":
+ return ReadMaxFiles(params["max_files"])
+
+ def _dump(self) -> dict:
+ return {"max_files": self.max_files}
+
+
+class ReadMaxBytes(ReadLimit):
+ """
+ A :class:`ReadLimit` that indicates to read maximum N bytes. The source should not read more
+ than N bytes when determining the latest offset.
+ """
+
+ def __init__(self, max_bytes: int) -> None:
+ self.max_bytes = max_bytes
+
+ @classmethod
+ def type_name(cls) -> str:
+ return "ReadMaxBytes"
+
+ @classmethod
+ def load(cls, params: dict) -> "ReadMaxBytes":
+ return ReadMaxBytes(params["max_bytes"])
+
+ def _dump(self) -> dict:
+ return {"max_bytes": self.max_bytes}
+
+
+class SupportsTriggerAvailableNow(ABC):
+ """
+ A mixin interface for streaming sources that support Trigger.AvailableNow. This interface can
+ be added to both :class:`DataSourceStreamReader` and :class:`SimpleDataSourceStreamReader`.
+ """
+
+ @abstractmethod
+ def prepareForTriggerAvailableNow(self) -> None:
+ """
+ This will be called at the beginning of streaming queries with Trigger.AvailableNow, to let
+ the source record the offset for the current latest data at the time (a.k.a the target
+ offset for the query). The source must behave as if there is no new data coming in after
+ the target offset, i.e., the source must not return an offset higher than the target offset
+ when :meth:`DataSourceStreamReader.latestOffset()` is called.
+
+ The source can extend the semantic of "current latest data" based on its own logic, but the
+ extended semantic must not violate the expectation that the source will not read any data
+ which is added later than the time this method has called.
+
+ Note that it is the source's responsibility to ensure that calling
+ :meth:`DataSourceStreamReader.latestOffset()` or :meth:`SimpleDataSourceStreamReader.read()`
+ after calling this method will eventually reach the target offset, and finally returns the
+ same offset as given start parameter, to indicate that there is no more data to read. This
+ includes the case where the query is restarted and the source is asked to read from the
Review Comment:
thanks for the explanation, the example sounds good to me
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR closed pull request #54085: [SPARK-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader
URL: https://github.com/apache/spark/pull/54085
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "gaogaotiantian (via GitHub)" <gi...@apache.org>.
gaogaotiantian commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2771503764
##########
python/pyspark/sql/streaming/python_streaming_source_runner.py:
##########
@@ -116,6 +138,77 @@ def send_batch_func(
write_int(EMPTY_PYARROW_RECORD_BATCHES, outfile)
+def check_support_func(reader: DataSourceStreamReader, outfile: IO) -> None:
+ support_flags = 0
+ if isinstance(reader, _SimpleStreamReaderWrapper):
Review Comment:
I think because this is internal, we don't have to make it "better" in this PR. One of the complaints we had for pyspark is that it's not pythonic. I'm not sure how important it is to make the implementation comparable to scala's.
In any case, I don't think we should make `_SimpleStreamReaderWrapper` a special case for every single place we need to deal with a `DataSourceStreamReader` - it doesn't make sense. If we can't use mixin to make `_SimpleStreamReaderWrapper` work, we should avoid it. Otherwise anytime we want to add something to data source, we need to make it special. In the future we might have another special "wrapper", and the world is going to blow up.
I think the most common way in Python is to make `DataSourceStreamReader` to have a default `prepareForTriggerAvailableNow` that raises `NotImplementedError` - you'll have a single API that you can implement in your class and the code can deal with it. Or you can have a separate API `supportTriggerAvailableNow` that returns `False` by default. Either way, we shouldn't need to access `reader.simple_reader` explicitly here.
You are right - python is duck type, and we should treat is as a duck type language.
However, I don't think you need to do it in this PR. We can refactor it later. I'm just pointing out that having special cases for `_SimpleStreamReaderWrapper` all over the code is a bit weird.
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "gaogaotiantian (via GitHub)" <gi...@apache.org>.
gaogaotiantian commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2767544702
##########
python/pyspark/sql/streaming/datasource.py:
##########
@@ -0,0 +1,234 @@
+#
+# 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.
+#
+
+from abc import ABC, abstractmethod
+
+
+class ReadLimit(ABC):
Review Comment:
Also if you think it's easier for them to do
```python
@dataclass
class MyOwnRateLimiter(RateLimit):
param1: int
param2: str
```
All you need to do is to do a bit extra work when you dump it (not even load). When you dump it, do something like
`asdict(r) | {"_type": r.__class__.__name__}` - that saved another thing you need to specify in the dataclass.
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2766485738
##########
python/pyspark/sql/datasource_internal.py:
##########
@@ -79,28 +84,26 @@ class _SimpleStreamReaderWrapper(DataSourceStreamReader):
def __init__(self, simple_reader: SimpleDataSourceStreamReader):
self.simple_reader = simple_reader
- self.initial_offset: Optional[dict] = None
- self.current_offset: Optional[dict] = None
self.cache: List[PrefetchedCacheEntry] = []
def initialOffset(self) -> dict:
- if self.initial_offset is None:
- self.initial_offset = self.simple_reader.initialOffset()
- return self.initial_offset
-
- def latestOffset(self) -> dict:
- # when query start for the first time, use initial offset as the start offset.
- if self.current_offset is None:
- self.current_offset = self.initialOffset()
- (iter, end) = self.simple_reader.read(self.current_offset)
- self.cache.append(PrefetchedCacheEntry(self.current_offset, end, iter))
- self.current_offset = end
+ return self.simple_reader.initialOffset()
+
+ def getDefaultReadLimit(self) -> ReadLimit:
+ # We do not consider providing different read limit on simple stream reader.
+ return ReadAllAvailable()
+
+ def latestOffset(self, start: dict, limit: ReadLimit) -> dict:
+ assert start is not None, "start offset should not be None"
+ assert isinstance(
+ limit, ReadAllAvailable
+ ), "simple stream reader does not support read limit"
Review Comment:
No, admission control is not available for simple streaming reader. Trigger.availableNow would be still available for simple stream reader.
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2771544709
##########
python/pyspark/sql/streaming/python_streaming_source_runner.py:
##########
@@ -116,6 +138,77 @@ def send_batch_func(
write_int(EMPTY_PYARROW_RECORD_BATCHES, outfile)
+def check_support_func(reader: DataSourceStreamReader, outfile: IO) -> None:
+ support_flags = 0
+ if isinstance(reader, _SimpleStreamReaderWrapper):
Review Comment:
I guess I agree with the general concern - though there is a tricky issue I had to deal with.
Probably you are not aware of the logic on Scala side so don't have a sufficient context.
There is a sequence 1) determine whether the data source supports trigger.availableNow (because in Scala side we have to use different implementations extending the interface) 2) call prepareForTriggerAvailableNow only when the data source reports that it supports trigger.availableNow.
Making prepareForTriggerAvailableNow that raises NotImplementedError does not work since we don't want to call this method twice just because of the implementation detail on the engine. Introducing supportTriggerAvailableNow directly into DataSourceStreamReader might be a viable alternative, but it's a clear divergence in terms of usage on interface between Scala vs Python - making two in a sync has been an implicit direction of API for python data source.
(latestOffset is an exception due to the backward compatibility issue I mentioned above, and arguably, admission control should be implemented by default. Old latestOffset signature makes no sense and I regret I hadn't pushed this in initial design of streaming reader.)
Not sure we'd love to do this just because we don't want to perform manual pattern matching. That'd be arguable and may not be something two of us just have a discuss and decide.
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "gaogaotiantian (via GitHub)" <gi...@apache.org>.
gaogaotiantian commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2771558645
##########
python/pyspark/sql/streaming/python_streaming_source_runner.py:
##########
@@ -116,6 +138,77 @@ def send_batch_func(
write_int(EMPTY_PYARROW_RECORD_BATCHES, outfile)
+def check_support_func(reader: DataSourceStreamReader, outfile: IO) -> None:
+ support_flags = 0
+ if isinstance(reader, _SimpleStreamReaderWrapper):
Review Comment:
Yeah that's one of my questions - do we need a symmetric implementation between scala and python? Or we just need API to match. I have not had the chance to look at the scala implementation.
On the other hand, is this mixin public? That users can use directly? If so, merging this means we have to go with this method in the future right? To keep compatibility.
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "gaogaotiantian (via GitHub)" <gi...@apache.org>.
gaogaotiantian commented on PR #54085:
URL: https://github.com/apache/spark/pull/54085#issuecomment-3857964216
The errors are about imcompatible protobuf versions. Try to rebase on 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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2758900088
##########
python/pyspark/sql/datasource.py:
##########
@@ -714,9 +715,37 @@ def initialOffset(self) -> dict:
messageParameters={"feature": "initialOffset"},
)
- def latestOffset(self) -> dict:
+ def latestOffset(self, start: dict, limit: ReadLimit) -> dict:
"""
- Returns the most recent offset available.
+ Returns the most recent offset available given a read limit. The start offset can be used
+ to figure out how much new data should be read given the limit.
+
+ The `start` will be provided from the return value of :meth:`initialOffset()` for
+ the very first micro-batch, and the offset continues from the last micro-batch for the
+ following. The source can return the same offset as start offset if there is no data to
Review Comment:
I'm slightly not in favor of coupling the contract with the engine's behavior, since it could limit ourselves. (I found myself doing this and I'm even OK with omitting it.) But the explanation from your comment is unlikely to change, so makes sense to me. Thanks for the suggestion.
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2771544709
##########
python/pyspark/sql/streaming/python_streaming_source_runner.py:
##########
@@ -116,6 +138,77 @@ def send_batch_func(
write_int(EMPTY_PYARROW_RECORD_BATCHES, outfile)
+def check_support_func(reader: DataSourceStreamReader, outfile: IO) -> None:
+ support_flags = 0
+ if isinstance(reader, _SimpleStreamReaderWrapper):
Review Comment:
I guess I agree with the general concern - though there is a tricky issue I had to deal with.
Probably you are not aware of the logic on Scala side so don't have a sufficient context.
There is a sequence 1) determine whether the data source supports trigger.availableNow (because in Scala side we have to use different implementations extending the interface) 2) call prepareForTriggerAvailableNow only when the data source reports that it supports trigger.availableNow.
Making prepareForTriggerAvailableNow that raises NotImplementedError does not work since we don't want to call this method twice just because of the implementation detail on the engine. This might have side effect on source implementations unless we enforce them to be idempotent.
Introducing supportTriggerAvailableNow directly into DataSourceStreamReader might be a viable alternative, but it's a clear divergence in terms of usage on interface between Scala vs Python - making two in a sync has been an implicit direction of API for python data source.
(latestOffset is an exception due to the backward compatibility issue I mentioned above, and arguably, admission control should be implemented by default. Old latestOffset signature makes no sense and I regret I hadn't pushed this in initial design of streaming reader.)
Not sure we'd love to do this just because we don't want to perform manual pattern matching. That'd be arguable and may not be something two of us just have a discuss and decide.
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2767805562
##########
python/pyspark/sql/streaming/python_streaming_source_runner.py:
##########
@@ -116,6 +138,77 @@ def send_batch_func(
write_int(EMPTY_PYARROW_RECORD_BATCHES, outfile)
+def check_support_func(reader: DataSourceStreamReader, outfile: IO) -> None:
+ support_flags = 0
+ if isinstance(reader, _SimpleStreamReaderWrapper):
Review Comment:
While I get how your proposal works, I'd argue that this is just another workaround for doing something which we can't do from inheritance.
For example, you said SupportsTriggerAvailableNow isn't the important thing - if we come to non-python language which has strong typing, explicitly checking the type is mostly required except the hacky way like reflection in Java - it is not a best practice. We do inspect for latestOffset but this is a last resort approach, because python does not support method overloading so we actually can't do the same evolution of API we did from Scala/Java.
(It's still possible we can introduce a separate interface to achieve similar thing but it is not about overloading - it's just that the latest definition will just override the prior definition for the same name of method.)
Though I understand that in python it's duck typing and the type system is crazy loose and the availability of function/method simply matters.
Maybe you might have tried to give a better approach for pythonic thinking. It's just that both breaks inheritance and the proposal seems to break more than what I do. I agree if you want to say this isn't pythonic, but then I could say this is one of the pattern we heavily leverage in Scala, we call it as pattern matching.
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2767805562
##########
python/pyspark/sql/streaming/python_streaming_source_runner.py:
##########
@@ -116,6 +138,77 @@ def send_batch_func(
write_int(EMPTY_PYARROW_RECORD_BATCHES, outfile)
+def check_support_func(reader: DataSourceStreamReader, outfile: IO) -> None:
+ support_flags = 0
+ if isinstance(reader, _SimpleStreamReaderWrapper):
Review Comment:
While I get how your proposal works, I'd argue that this is just another workaround for doing something which we can't do from inheritance.
For example, you said SupportsTriggerAvailableNow isn't the important thing - if we come to non-python language which has strong typing, explicitly checking the type is mostly required except the hacky way like reflection in Java - it is not a best practice. We do inspect for latestOffset but this is a last resort approach, because python does not support method overloading so we actually can't do the same evolution of API we did from Scala.
(It's still possible we can introduce a separate interface to achieve similar thing but it is not about overloading - it's just that the latest definition will just override the prior definition for the same name of method.)
Though I understand that in python it's duck typing and the type system is crazy loose and the availability of function/method simply matters.
Maybe you might try to give a better approach for pythonic thinking. It's just that both breaks inheritance and the proposal seems to break more than what I do. I agree if you want to say this isn't pythonic, but then I could say this is one of the pattern we heavily leverage in Scala, we call it as pattern matching.
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "jiateoh (via GitHub)" <gi...@apache.org>.
jiateoh commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2761220176
##########
python/pyspark/sql/datasource_internal.py:
##########
@@ -144,3 +141,28 @@ def read(
self, input_partition: SimpleInputPartition # type: ignore[override]
) -> Iterator[Tuple]:
return self.simple_reader.readBetweenOffsets(input_partition.start, input_partition.end)
+
+
+class ReadLimitRegistry:
+ def __init__(self) -> None:
+ self._registry: Dict[str, Type[ReadLimit]] = {}
+ # Register built-in ReadLimit types
+ self.__register(ReadAllAvailable.type_name(), ReadAllAvailable)
+ self.__register(ReadMinRows.type_name(), ReadMinRows)
+ self.__register(ReadMaxRows.type_name(), ReadMaxRows)
+ self.__register(ReadMaxBytes.type_name(), ReadMaxBytes)
+
+ def __register(self, type_name: str, read_limit_type: Type["ReadLimit"]) -> None:
+ if type_name in self._registry:
+ raise PySparkException(f"ReadLimit type '{type_name}' is already registered.")
+
+ self._registry[type_name] = read_limit_type
+
+ def get(self, type_name: str, params: dict) -> ReadLimit:
+ read_limit_type = self._registry[type_name]
Review Comment:
+1 here, it will throw a KeyError I think. Either way this should probably use `self._registry._get(type_name)` or have an appropriate check
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2761698253
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/python/streaming/PythonStreamingDataSourceSuite.scala:
##########
Review Comment:
It happens with Trigger.AvailableNow; I'll check whether we do not have one; maybe that needs some union or stream-stream join since we still need to trigger the microbatch to verify the behavior. (If latestOffset returns the same offset and it's the only stream, we don't trigger the microbatch.)
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/python/streaming/PythonStreamingDataSourceSuite.scala:
##########
Review Comment:
It happens with Trigger.AvailableNow; I'll check whether we do not have one for processing time timer; maybe that needs some union or stream-stream join since we still need to trigger the microbatch to verify the behavior. (If latestOffset returns the same offset and it's the only stream, we don't trigger the microbatch.)
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2762061031
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/python/streaming/PythonStreamingDataSourceSuite.scala:
##########
Review Comment:
https://github.com/apache/spark/pull/54085/commits/853e29ac3e6a649d03bd3a861272149afbd9a445
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "allisonwang-db (via GitHub)" <gi...@apache.org>.
allisonwang-db commented on PR #54085:
URL: https://github.com/apache/spark/pull/54085#issuecomment-3850464305
cc @shujingyang-db @gaogaotiantian
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "gaogaotiantian (via GitHub)" <gi...@apache.org>.
gaogaotiantian commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2767477462
##########
python/pyspark/sql/streaming/python_streaming_source_runner.py:
##########
@@ -47,23 +58,34 @@
utf8_deserializer,
)
+
INITIAL_OFFSET_FUNC_ID = 884
LATEST_OFFSET_FUNC_ID = 885
PARTITIONS_FUNC_ID = 886
COMMIT_FUNC_ID = 887
+CHECK_SUPPORTED_FEATURES_ID = 888
+PREPARE_FOR_TRIGGER_AVAILABLE_NOW_FUNC_ID = 889
+LATEST_OFFSET_ADMISSION_CONTROL_FUNC_ID = 890
+GET_DEFAULT_READ_LIMIT_FUNC_ID = 891
+REPORT_LATEST_OFFSET_FUNC_ID = 892
PREFETCHED_RECORDS_NOT_FOUND = 0
NON_EMPTY_PYARROW_RECORD_BATCHES = 1
EMPTY_PYARROW_RECORD_BATCHES = 2
+SUPPORTS_ADMISSION_CONTROL = 1
Review Comment:
`SUPPORTS_ADMISSION_CONTROL = 1 << 0` is better I think.
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "gaogaotiantian (via GitHub)" <gi...@apache.org>.
gaogaotiantian commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2767534350
##########
python/pyspark/sql/streaming/datasource.py:
##########
@@ -0,0 +1,234 @@
+#
+# 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.
+#
+
+from abc import ABC, abstractmethod
+
+
+class ReadLimit(ABC):
Review Comment:
Yeah there could still be custom `RateLimit` in the future. The difference is that the user only needs to define a dataclass like
```python
@dataclass
class MyOwnRateLimiter(RateLimit):
name = "MyOwnRateLimiter"
param1: int
param2: str
```
And that's it - they don't need to do anything else.
I changed `type` to `name` here because I think that's probably a bit more intuitive. Also `type` is a global variable so even though it's okay to use it here, I try to avoid it.
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "gaogaotiantian (via GitHub)" <gi...@apache.org>.
gaogaotiantian commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2767449282
##########
python/pyspark/sql/streaming/datasource.py:
##########
@@ -0,0 +1,234 @@
+#
+# 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.
+#
+
+from abc import ABC, abstractmethod
+
+
+class ReadLimit(ABC):
Review Comment:
I think this `ABC` is over-designed, especially considering that we do not even support custom `ReadLimit` class at this point.
`type_name` does not really do anything besides returning an identifier for this class. It's only used internally I think you can directly use `self.__class__.__name__`.
`cls.load(param) -> cls` is not a super common pattern in python or pyspark. This is just `__init__` I think. You just need the subclass to have an `__init__` function that takes a parameter. Save that `parameter` and use it in `dump` would be fine.
Bottom line is, you just need a serializable enum that take some argument. I think you can totally do it with just dataclass.
```python
from dataclasses import dataclass
class ReadLimit:
...
@dataclass
class ReadAllAvailable(ReadLimit):
type: str = "ReadAllAvailable"
@dataclass
class ReadMinRows(ReadLimit):
type: str = "ReadMinRows"
min_rows: int
```
You can do `dataclasses.asdict(obj)` to dump it and `registry[type](**params)` to create the class. What's the concerns to use this simple pattern?
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #54085:
URL: https://github.com/apache/spark/pull/54085#issuecomment-3863352892
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2758940103
##########
python/pyspark/sql/streaming/datasource.py:
##########
@@ -0,0 +1,225 @@
+#
+# 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.
+#
+
+from abc import ABC, abstractmethod
+
+
+class ReadLimit(ABC):
+ """
+ Specifies limits on how much data to read from a streaming source when
+ determining the latest offset.
+ """
+
+ @classmethod
+ @abstractmethod
+ def type_name(cls) -> str:
+ """
+ The name of this :class:`ReadLimit` type. This is used to register the type into registry.
+
+ Returns
+ -------
+ str
+ The name of this :class:`ReadLimit` type.
+ """
+ pass
+
+ @classmethod
+ @abstractmethod
+ def load(cls, params: dict) -> "ReadLimit":
+ """
+ Create an instance of :class:`ReadLimit` from parameters.
+
+ Parameter
+ ---------
+ params : dict
+ The parameters to create the :class:`ReadLimit`. type name isn't included.
+
+ Returns
+ -------
+ ReadLimit
+ The created :class:`ReadLimit` instance.
+ """
+ pass
+
+ def dump(self) -> dict:
+ """
+ Method to serialize this :class:`ReadLimit` instance. Implementations of :class:`ReadLimit`
+ are expected to not implement this method directly and rather implement the
+ :meth:`_dump()` method.
+
+ Returns
+ -------
+ dict
+ A dictionary containing the serialized parameters of this :class:`ReadLimit`,
+ including the type name.
+ """
+ params = self._dump()
+ params.update({"type": self.type_name()})
+ return params
+
+ @abstractmethod
+ def _dump(self) -> dict:
+ """
+ Method to serialize this :class:`ReadLimit` instance. Implementations of :class:`ReadLimit`
+ are expected to implement this method to handle their specific parameters. type name will
+ be handled in the :meth:`dump()` method.
+
+ Returns
+ -------
+ dict
+ A dictionary containing the serialized parameters of this :class:`ReadLimit`,
+ excluding the type name.
+ """
+ pass
+
+
+class ReadAllAvailable(ReadLimit):
+ """
+ A :class:`ReadLimit` that indicates to read all available data, regardless of the given source
+ options.
+ """
+
+ @classmethod
+ def type_name(cls) -> str:
+ return "ReadAllAvailable"
+
+ @classmethod
+ def load(cls, params: dict) -> "ReadAllAvailable":
+ return ReadAllAvailable()
+
+ def _dump(self) -> dict:
+ return {}
+
+
+class ReadMinRows(ReadLimit):
+ """
+ A :class:`ReadLimit` that indicates to read minimum N rows. If there is less than N rows
+ available for read, the source should skip producing a new offset to read and wait until more
+ data arrives.
+
+ Note that the semantic does not work properly with Trigger.AvailableNow since the source
+ may end up waiting forever for more data to arrive. It is the source's responsibility to
+ handle this case properly.
+ """
+
+ def __init__(self, min_rows: int) -> None:
+ self.min_rows = min_rows
+
+ @classmethod
+ def type_name(cls) -> str:
+ return "ReadMinRows"
+
+ @classmethod
+ def load(cls, params: dict) -> "ReadMinRows":
+ return ReadMinRows(params["min_rows"])
+
+ def _dump(self) -> dict:
+ return {"min_rows": self.min_rows}
+
+
+class ReadMaxRows(ReadLimit):
+ """
+ A :class:`ReadLimit` that indicates to read maximum N rows. The source should not read more
+ than N rows when determining the latest offset.
+ """
+
+ def __init__(self, max_rows: int) -> None:
+ self.max_rows = max_rows
+
+ @classmethod
+ def type_name(cls) -> str:
+ return "ReadMaxRows"
+
+ @classmethod
+ def load(cls, params: dict) -> "ReadMaxRows":
+ return ReadMaxRows(params["max_rows"])
+
+ def _dump(self) -> dict:
+ return {"max_rows": self.max_rows}
+
+
+class ReadMaxFiles(ReadLimit):
+ """
+ A :class:`ReadLimit` that indicates to read maximum N files. The source should not read more
+ than N files when determining the latest offset.
+ """
+
+ def __init__(self, max_files: int) -> None:
+ self.max_files = max_files
+
+ @classmethod
+ def type_name(cls) -> str:
+ return "ReadMaxFiles"
+
+ @classmethod
+ def load(cls, params: dict) -> "ReadMaxFiles":
+ return ReadMaxFiles(params["max_files"])
+
+ def _dump(self) -> dict:
+ return {"max_files": self.max_files}
+
+
+class ReadMaxBytes(ReadLimit):
+ """
+ A :class:`ReadLimit` that indicates to read maximum N bytes. The source should not read more
+ than N bytes when determining the latest offset.
+ """
+
+ def __init__(self, max_bytes: int) -> None:
+ self.max_bytes = max_bytes
+
+ @classmethod
+ def type_name(cls) -> str:
+ return "ReadMaxBytes"
+
+ @classmethod
+ def load(cls, params: dict) -> "ReadMaxBytes":
+ return ReadMaxBytes(params["max_bytes"])
+
+ def _dump(self) -> dict:
+ return {"max_bytes": self.max_bytes}
+
+
+class SupportsTriggerAvailableNow(ABC):
+ """
+ A mixin interface for streaming sources that support Trigger.AvailableNow. This interface can
+ be added to both :class:`DataSourceStreamReader` and :class:`SimpleDataSourceStreamReader`.
+ """
+
+ @abstractmethod
+ def prepareForTriggerAvailableNow(self) -> None:
+ """
+ This will be called at the beginning of streaming queries with Trigger.AvailableNow, to let
+ the source record the offset for the current latest data at the time (a.k.a the target
+ offset for the query). The source must behave as if there is no new data coming in after
+ the target offset, i.e., the source must not return an offset higher than the target offset
+ when :meth:`DataSourceStreamReader.latestOffset()` is called.
+
+ The source can extend the semantic of "current latest data" based on its own logic, but the
+ extended semantic must not violate the expectation that the source will not read any data
+ which is added later than the time this method has called.
+
+ Note that it is the source's responsibility to ensure that calling
+ :meth:`DataSourceStreamReader.latestOffset()` or :meth:`SimpleDataSourceStreamReader.read()`
+ after calling this method will eventually reach the target offset, and finally returns the
+ same offset as given start parameter, to indicate that there is no more data to read. This
+ includes the case where the query is restarted and the source is asked to read from the
Review Comment:
https://github.com/apache/spark/commit/4575b1da9e1d7c39c3769f2c53b213a5e5a5407d
So here is the scenario - the query reads from Kafka topic. In the first run, the topic has 3 partitions. During the downtime of the query, users perform repartition of Kafka topic and now the topic has 5 partitions. If there was uncommitted batch, the second run of query will get the `start` offset from uncommitted batch, which had only 3 partitions. In the meanwhile, prepareForTriggerAvailableNow() will identify there are 5 partitions and store the offset for 5 partitions. The source is responsible to read further from 3 partitions and figure out new partitions, and eventually touch the offset stored from prepareForTriggerAvailableNow().
The scenario is actually complicated and I might not be able to describe the case with ease of understanding. If there is proposal for better wording, I appreciate the suggestion!
##########
python/pyspark/sql/datasource.py:
##########
@@ -714,9 +715,37 @@ def initialOffset(self) -> dict:
messageParameters={"feature": "initialOffset"},
)
- def latestOffset(self) -> dict:
+ def latestOffset(self, start: dict, limit: ReadLimit) -> dict:
"""
- Returns the most recent offset available.
+ Returns the most recent offset available given a read limit. The start offset can be used
+ to figure out how much new data should be read given the limit.
+
+ The `start` will be provided from the return value of :meth:`initialOffset()` for
+ the very first micro-batch, and the offset continues from the last micro-batch for the
+ following. The source can return the same offset as start offset if there is no data to
+ process.
+
+ :class:`ReadLimit` can be used by the source to limit the amount of data returned in this
+ call. The implementation should implement :meth:`getDefaultReadLimit()` to provide the
+ proper :class:`ReadLimit` if the source can limit the amount of data returned based on the
+ source options.
+
+ The engine can still call :meth:`latestOffset()` with :class:`ReadAllAvailable` even if the
+ source produces the different read limit from :meth:`getDefaultReadLimit()`, to respect the
+ semantic of trigger. The source must always respect the given readLimit provided by the
+ engine; e.g. if the readLimit is :class:`ReadAllAvailable`, the source must ignore the read
+ limit configured through options.
Review Comment:
It's for two cases 1) Trigger.Once (deprecated), 2) fallback of Trigger.AvailableNow (when any stream does not support Trigger.AvailableNow and Trigger.AvailableNow is requested) - I'm not very sure we would like to document these cases, as once we document it's considered as "contract" rather than implementation detail.
##########
python/pyspark/sql/datasource.py:
##########
@@ -714,9 +715,37 @@ def initialOffset(self) -> dict:
messageParameters={"feature": "initialOffset"},
)
- def latestOffset(self) -> dict:
+ def latestOffset(self, start: dict, limit: ReadLimit) -> dict:
"""
- Returns the most recent offset available.
+ Returns the most recent offset available given a read limit. The start offset can be used
+ to figure out how much new data should be read given the limit.
+
+ The `start` will be provided from the return value of :meth:`initialOffset()` for
+ the very first micro-batch, and the offset continues from the last micro-batch for the
+ following. The source can return the same offset as start offset if there is no data to
Review Comment:
I meant the parameter.
"The source can return the `start` parameter as it is, if there is no data to process"
^ Would it be clearer?
##########
python/pyspark/sql/datasource.py:
##########
@@ -714,9 +715,37 @@ def initialOffset(self) -> dict:
messageParameters={"feature": "initialOffset"},
)
- def latestOffset(self) -> dict:
+ def latestOffset(self, start: dict, limit: ReadLimit) -> dict:
"""
- Returns the most recent offset available.
+ Returns the most recent offset available given a read limit. The start offset can be used
+ to figure out how much new data should be read given the limit.
+
+ The `start` will be provided from the return value of :meth:`initialOffset()` for
+ the very first micro-batch, and the offset continues from the last micro-batch for the
+ following. The source can return the same offset as start offset if there is no data to
Review Comment:
I'm slightly not in favor of coupling the contract with the engine's behavior, since it could limit ourselves. But the explanation from your comment is unlikely to change, so makes sense to me. Thanks for the suggestion.
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2767472871
##########
python/pyspark/sql/streaming/datasource.py:
##########
@@ -0,0 +1,234 @@
+#
+# 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.
+#
+
+from abc import ABC, abstractmethod
+
+
+class ReadLimit(ABC):
Review Comment:
OK looks like it's still retaining the mechanism of registry but just to remove out the classmethods. Maybe we could just ask the custom impl to provide the type and done. OK for me.
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "gaogaotiantian (via GitHub)" <gi...@apache.org>.
gaogaotiantian commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2767511348
##########
python/pyspark/sql/streaming/python_streaming_source_runner.py:
##########
@@ -116,6 +138,77 @@ def send_batch_func(
write_int(EMPTY_PYARROW_RECORD_BATCHES, outfile)
+def check_support_func(reader: DataSourceStreamReader, outfile: IO) -> None:
+ support_flags = 0
+ if isinstance(reader, _SimpleStreamReaderWrapper):
Review Comment:
I'm surprised that we wrote `_SimpleStreamReaderWrapper` which is a subclass of `DataSourceStreamReader` yet we still need to have a separate if statement for it - that to be is against the rule of inheritance...
However, in this case, do we really need to?
The same logic for `inspect.signature` applies for `_SimpleStreamReaderWrapper` because it does have the correct signature for `reader.latestOffset`. `SupportsTriggerAvailableNow` is not the important thing the important thing should be `prepareForTriggerAvailableNow`. We should have a read-through logic for `_SimpleStreamReaderWrapper` so `hasattr(reader, "prepareForTriggerAvailableNow")` returns the underlying `simple_reader`'s attributes directly.
Maybe we don't need to do this in this PR, but we should not claim that `_SimpleStreamReaderWrapper` is a `DataSourceStreamReader` and still need to have a separate case to access `simple_reader` all the time.
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2767805562
##########
python/pyspark/sql/streaming/python_streaming_source_runner.py:
##########
@@ -116,6 +138,77 @@ def send_batch_func(
write_int(EMPTY_PYARROW_RECORD_BATCHES, outfile)
+def check_support_func(reader: DataSourceStreamReader, outfile: IO) -> None:
+ support_flags = 0
+ if isinstance(reader, _SimpleStreamReaderWrapper):
Review Comment:
While I get how your proposal works, I'd argue that this is just another workaround for doing something which we can't do from inheritance.
For example, you said SupportsTriggerAvailableNow isn't the important thing - if we come to non-python language which has strong typing, explicitly checking the type is mostly required except the hacky way like reflection in Java - it is not a best practice. We do inspect for latestOffset but this is a last resort approach, because python does not support overloading so we actually can't do the same evolution of API we did from Scala.
(It's still possible we can introduce a separate interface to achieve similar thing but it is not about overloading - it's just that the latest definition will just override the prior definition for the same name of method.)
Though I understand that in python it's duck typing and the type system is crazy loose and the availability of function/method simply matters.
Maybe you might try to give a better approach for pythonic thinking. It's just that both breaks inheritance and the proposal seems to break more than what I do. I agree if you want to say this isn't pythonic, but then I could say this is one of the pattern we heavily leverage in Scala, we call it as pattern matching.
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #54085:
URL: https://github.com/apache/spark/pull/54085#issuecomment-3852026465
I think the proposal on the interface is more about arguable instead of something I totally agree. While it might be more pythonic, at least we shouldn't claim that is better for inheritance. From the proposal I don't see we respect the interface at all. I don't think this can hold the PR from merging anyway.
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2769105138
##########
python/pyspark/sql/streaming/datasource.py:
##########
@@ -0,0 +1,234 @@
+#
+# 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.
+#
+
+from abc import ABC, abstractmethod
+
+
+class ReadLimit(ABC):
Review Comment:
I'll take the class name - basically I want flexibility of the name, but that sounds to me I need to do it with class variable and that is even modifiable from outside, which could mess up the thing. I realized Final is just a type hint.
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "huanliwang-db (via GitHub)" <gi...@apache.org>.
huanliwang-db commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2757504719
##########
python/pyspark/sql/datasource.py:
##########
@@ -714,9 +715,37 @@ def initialOffset(self) -> dict:
messageParameters={"feature": "initialOffset"},
)
- def latestOffset(self) -> dict:
+ def latestOffset(self, start: dict, limit: ReadLimit) -> dict:
"""
- Returns the most recent offset available.
+ Returns the most recent offset available given a read limit. The start offset can be used
+ to figure out how much new data should be read given the limit.
+
+ The `start` will be provided from the return value of :meth:`initialOffset()` for
+ the very first micro-batch, and the offset continues from the last micro-batch for the
+ following. The source can return the same offset as start offset if there is no data to
Review Comment:
maybe "and for subsequent micro-batches, the start offset is the ending offset from the previous micro-batch." is better iirc?
--
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-55304][SS][PYTHON] Introduce support of Admission Control and Trigger.AvailableNow in Python data source - streaming reader [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #54085:
URL: https://github.com/apache/spark/pull/54085#discussion_r2761883098
##########
python/pyspark/sql/datasource_internal.py:
##########
@@ -144,3 +141,28 @@ def read(
self, input_partition: SimpleInputPartition # type: ignore[override]
) -> Iterator[Tuple]:
return self.simple_reader.readBetweenOffsets(input_partition.start, input_partition.end)
+
+
+class ReadLimitRegistry:
+ def __init__(self) -> None:
+ self._registry: Dict[str, Type[ReadLimit]] = {}
+ # Register built-in ReadLimit types
+ self.__register(ReadAllAvailable.type_name(), ReadAllAvailable)
+ self.__register(ReadMinRows.type_name(), ReadMinRows)
+ self.__register(ReadMaxRows.type_name(), ReadMaxRows)
+ self.__register(ReadMaxBytes.type_name(), ReadMaxBytes)
+
+ def __register(self, type_name: str, read_limit_type: Type["ReadLimit"]) -> None:
+ if type_name in self._registry:
+ raise PySparkException(f"ReadLimit type '{type_name}' is already registered.")
+
+ self._registry[type_name] = read_limit_type
+
+ def get(self, type_name: str, params: dict) -> ReadLimit:
+ read_limit_type = self._registry[type_name]
Review Comment:
lol I have check with None at next line. I'll just do .get().
--
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