source: Support since param in sources

fennel/CHANGELOG.md

@@ -1,5 +1,8 @@
 # Changelog
 
+## [0.18.10] - 2023-10-30
+- Add support for `since` in S3 source.
+
 ## [0.18.9]- 2023-10-27
 - Added support for datetime fields in struct types
 

fennel/datasets/datasets.py

@@ -1167,6 +1167,7 @@ def with_source(
         self,
         conn: DataConnector,
         every: Optional[Duration] = None,
+        starting_from: Optional[datetime.datetime] = None,
         lateness: Optional[Duration] = None,
     ):
         if len(self._pipelines) > 0:
@@ -1177,7 +1178,7 @@ def with_source(
         ds_copy = copy.deepcopy(self)
         if hasattr(ds_copy, sources.SOURCE_FIELD):
             delattr(ds_copy, sources.SOURCE_FIELD)
-        src_fn = source(conn, every, lateness)
+        src_fn = source(conn, every, starting_from, lateness)
         return src_fn(ds_copy)
 
     def dsschema(self):

fennel/gen/connector_pb2.pyi

@@ -40,13 +40,15 @@ class _CDCStrategyEnumTypeWrapper(google.protobuf.internal.enum_type_wrapper._En
     Upsert: _CDCStrategy.ValueType  # 1
     Debezium: _CDCStrategy.ValueType  # 2
     Native: _CDCStrategy.ValueType  # 3
+    Delete: _CDCStrategy.ValueType  # 4
 
 class CDCStrategy(_CDCStrategy, metaclass=_CDCStrategyEnumTypeWrapper): ...
 
 Append: CDCStrategy.ValueType  # 0
 Upsert: CDCStrategy.ValueType  # 1
 Debezium: CDCStrategy.ValueType  # 2
 Native: CDCStrategy.ValueType  # 3
+Delete: CDCStrategy.ValueType  # 4
 global___CDCStrategy = CDCStrategy
 
 @typing_extensions.final
@@ -655,6 +657,7 @@ class Source(google.protobuf.message.Message):
     LATENESS_FIELD_NUMBER: builtins.int
     TIMESTAMP_FIELD_FIELD_NUMBER: builtins.int
     CDC_FIELD_NUMBER: builtins.int
+    STARTING_FROM_FIELD_NUMBER: builtins.int
     @property
     def table(self) -> global___ExtTable: ...
     dataset: builtins.str
@@ -665,6 +668,8 @@ class Source(google.protobuf.message.Message):
     def lateness(self) -> google.protobuf.duration_pb2.Duration: ...
     timestamp_field: builtins.str
     cdc: global___CDCStrategy.ValueType
+    @property
+    def starting_from(self) -> google.protobuf.timestamp_pb2.Timestamp: ...
     def __init__(
         self,
         *,
@@ -675,9 +680,10 @@ class Source(google.protobuf.message.Message):
         lateness: google.protobuf.duration_pb2.Duration | None = ...,
         timestamp_field: builtins.str = ...,
         cdc: global___CDCStrategy.ValueType = ...,
+        starting_from: google.protobuf.timestamp_pb2.Timestamp | None = ...,
     ) -> None: ...
-    def HasField(self, field_name: typing_extensions.Literal["_cursor", b"_cursor", "cursor", b"cursor", "every", b"every", "lateness", b"lateness", "table", b"table"]) -> builtins.bool: ...
-    def ClearField(self, field_name: typing_extensions.Literal["_cursor", b"_cursor", "cdc", b"cdc", "cursor", b"cursor", "dataset", b"dataset", "every", b"every", "lateness", b"lateness", "table", b"table", "timestamp_field", b"timestamp_field"]) -> None: ...
+    def HasField(self, field_name: typing_extensions.Literal["_cursor", b"_cursor", "cursor", b"cursor", "every", b"every", "lateness", b"lateness", "starting_from", b"starting_from", "table", b"table"]) -> builtins.bool: ...
+    def ClearField(self, field_name: typing_extensions.Literal["_cursor", b"_cursor", "cdc", b"cdc", "cursor", b"cursor", "dataset", b"dataset", "every", b"every", "lateness", b"lateness", "starting_from", b"starting_from", "table", b"table", "timestamp_field", b"timestamp_field"]) -> None: ...
     def WhichOneof(self, oneof_group: typing_extensions.Literal["_cursor", b"_cursor"]) -> typing_extensions.Literal["cursor"] | None: ...
 
 global___Source = Source