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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
