Skip to content

Commit ae7e20e

Browse files
authored
feat: Make batch_source optional in PushSource (#5440) (#5454)
* Make batch_source optional in PushSource (#5440) Signed-off-by: snehsuresh <[email protected]> * docs: make batch_source optional in PushSource Signed-off-by: snehsuresh <[email protected]> * refactor: remove comment Signed-off-by: snehsuresh <[email protected]> * fix: ensure batch_source is valid DataSource after null check Signed-off-by: snehsuresh <[email protected]> --------- Signed-off-by: snehsuresh <[email protected]>
1 parent 3777c00 commit ae7e20e

4 files changed

Lines changed: 38 additions & 11 deletions

File tree

‎docs/reference/data-sources/push.md‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,9 +6,12 @@ Push sources allow feature values to be pushed to the online store and offline s
66

77
Push sources can be used by multiple feature views. When data is pushed to a push source, Feast propagates the feature values to all the consuming feature views.
88

9-
Push sources must have a batch source specified. The batch source will be used for retrieving historical features. Thus users are also responsible for pushing data to a batch data source such as a data warehouse table. When using a push source as a stream source in the definition of a feature view, a batch source doesn't need to be specified in the feature view definition explicitly.
9+
Push sources can optionally have a batch_source specified. If provided, it enables retrieval of historical features and supports materialization from the offline store to the online store. However, if your features are generated post-training or are only needed online (e.g., embeddings), you can omit the batch_source.
10+
11+
When a batch_source is used, users are responsible for ensuring that data is also pushed to a batch data source, such as a data warehouse. Note that when a push source is used as a stream source in a feature view definition, a batch_source does not need to be explicitly specified in the feature view itself.
1012

1113
## Stream sources
14+
1215
Streaming data sources are important sources of feature values. A typical setup with streaming data looks like:
1316

1417
1. Raw events come in (stream 1)
@@ -20,7 +23,9 @@ Streaming data sources are important sources of feature values. A typical setup
2023
Feast allows users to push features previously registered in a feature view to the online store for fresher features. It also allows users to push batches of stream data to the offline store by specifying that the push be directed to the offline store. This will push the data to the offline store declared in the repository configuration used to initialize the feature store.
2124

2225
## Example (basic)
26+
2327
### Defining a push source
28+
2429
Note that the push schema needs to also include the entity.
2530

2631
```python
@@ -43,7 +48,9 @@ fv = FeatureView(
4348
```
4449

4550
### Pushing data
51+
4652
Note that the `to` parameter is optional and defaults to online but we can specify these options: `PushMode.ONLINE`, `PushMode.OFFLINE`, or `PushMode.ONLINE_AND_OFFLINE`.
53+
4754
```python
4855
from feast import FeatureStore
4956
import pandas as pd

‎sdk/python/feast/data_source.py‎

Lines changed: 10 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -764,13 +764,13 @@ class PushSource(DataSource):
764764

765765
# TODO(adchia): consider adding schema here in case where Feast manages pushing events to the offline store
766766
# TODO(adchia): consider a "mode" to support pushing raw vs transformed events
767-
batch_source: DataSource
767+
batch_source: Optional[DataSource] = None
768768

769769
def __init__(
770770
self,
771771
*,
772772
name: str,
773-
batch_source: DataSource,
773+
batch_source: Optional[DataSource] = None,
774774
description: Optional[str] = "",
775775
tags: Optional[Dict[str, str]] = None,
776776
owner: Optional[str] = "",
@@ -815,8 +815,11 @@ def get_table_column_names_and_types(
815815

816816
@staticmethod
817817
def from_proto(data_source: DataSourceProto):
818-
assert data_source.HasField("batch_source")
819-
batch_source = DataSource.from_proto(data_source.batch_source)
818+
batch_source = (
819+
DataSource.from_proto(data_source.batch_source)
820+
if data_source.HasField("batch_source")
821+
else None
822+
)
820823

821824
return PushSource(
822825
name=data_source.name,
@@ -827,19 +830,17 @@ def from_proto(data_source: DataSourceProto):
827830
)
828831

829832
def to_proto(self) -> DataSourceProto:
830-
batch_source_proto = None
831-
if self.batch_source:
832-
batch_source_proto = self.batch_source.to_proto()
833-
834833
data_source_proto = DataSourceProto(
835834
name=self.name,
836835
type=DataSourceProto.PUSH_SOURCE,
837836
description=self.description,
838837
tags=self.tags,
839838
owner=self.owner,
840-
batch_source=batch_source_proto,
841839
)
842840

841+
if self.batch_source:
842+
data_source_proto.batch_source.MergeFrom(self.batch_source.to_proto())
843+
843844
return data_source_proto
844845

845846
def get_table_query_string(self) -> str:

‎sdk/python/feast/inference.py‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,10 @@ def update_data_sources_with_inferred_event_timestamp_col(
2929
if isinstance(data_source, RequestSource):
3030
continue
3131
if isinstance(data_source, PushSource):
32-
data_source = data_source.batch_source
32+
if not isinstance(data_source.batch_source, DataSource):
33+
continue
34+
else:
35+
data_source = data_source.batch_source
3336
if data_source.timestamp_field is None or data_source.timestamp_field == "":
3437
# prepare right match pattern for data source
3538
ts_column_type_regex_pattern: str

‎sdk/python/tests/unit/test_data_sources.py‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,22 @@ def test_push_with_batch():
3030
assert push_source.batch_source.name == push_source_unproto.batch_source.name
3131

3232

33+
def test_push_source_without_batch_source():
34+
# Create PushSource with no batch_source
35+
push_source = PushSource(name="test_push_source")
36+
37+
# Convert to proto
38+
push_source_proto = push_source.to_proto()
39+
40+
# Assert batch_source is not present in proto
41+
assert not push_source_proto.HasField("batch_source")
42+
43+
# Deserialize and check again
44+
push_source_unproto = PushSource.from_proto(push_source_proto)
45+
assert push_source_unproto.batch_source is None
46+
assert push_source_unproto.name == "test_push_source"
47+
48+
3349
def test_request_source_primitive_type_to_proto():
3450
schema = [
3551
Field(name="f1", dtype=Float32),

0 commit comments

Comments
 (0)