GeezerVison is showing github.com — add init and cleanup of long lived resources · feast-dev/feast@6ca3b47
File treeExpand file treeCollapse file tree Open diff view settings Expand file treeCollapse file tree Open diff view settings Collapse file Copy file name to clipboardExpand all lines: docs/getting-started/concepts/permission.md +2-2Lines changed: 2 additions & 2 deletions - Display the source diff
- Display the rich diff
| Original file line number | Diff line number | Diff line change |
|---|
@@ -36,7 +36,7 @@ The permission model is based on the following components: | 36 | 36 | The `Permission` class identifies a single permission configured on the feature store and is identified by these attributes: | 37 | 37 | - `name`: The permission name. | 38 | 38 | - `types`: The list of protected resource types. Defaults to all managed types, e.g. the `ALL_RESOURCE_TYPES` alias. All sub-classes are included in the resource match. | 39 | | -- `name_pattern`: A regex to match the resource name. Defaults to `None`, meaning that no name filtering is applied | | 39 | +- `name_patterns`: A list of regex patterns to match resource names. If any regex matches, the `Permission` policy is applied. Defaults to `[]`, meaning no name filtering is applied. | 40 | 40 | - `required_tags`: Dictionary of key-value pairs that must match the resource tags. Defaults to `None`, meaning that no tags filtering is applied. | 41 | 41 | - `actions`: The actions authorized by this permission. Defaults to `ALL_VALUES`, an alias defined in the `action` module. | 42 | 42 | - `policy`: The policy to be applied to validate a client request. | @@ -95,7 +95,7 @@ The following permission grants authorization to read the offline store of all t | 95 | 95 | Permission( | 96 | 96 | name="reader", | 97 | 97 | types=[FeatureView], | 98 | | - name_pattern=".*risky.*", | | 98 | + name_patterns=".*risky.*", # Accepts both `str` or `list[str]` types | 99 | 99 | policy=RoleBasedPolicy(roles=["trusted"]), | 100 | 100 | actions=[AuthzedAction.READ_OFFLINE], | 101 | 101 | ) | |
Collapse file Copy file name to clipboardExpand all lines: docs/reference/feast-cli-commands.md +2-1Lines changed: 2 additions & 1 deletion - Display the source diff
- Display the rich diff
| Original file line number | Diff line number | Diff line change |
|---|
@@ -172,9 +172,10 @@ Options: | 172 | 172 |
| 173 | 173 | ```text | 174 | 174 | +-----------------------+-------------+-----------------------+-----------+----------------+-------------------------+ | 175 | | -| NAME | TYPES | NAME_PATTERN | ACTIONS | ROLES | REQUIRED_TAGS | | | 175 | +| NAME | TYPES | NAME_PATTERNS | ACTIONS | ROLES | REQUIRED_TAGS | | 176 | 176 | +=======================+=============+=======================+===========+================+================+========+ | 177 | 177 | | reader_permission1234 | FeatureView | transformed_conv_rate | DESCRIBE | reader | - | | | 178 | +| | | driver_hourly_stats | DESCRIBE | reader | - | | 178 | 179 | +-----------------------+-------------+-----------------------+-----------+----------------+-------------------------+ | 179 | 180 | | writer_permission1234 | FeatureView | transformed_conv_rate | CREATE | writer | - | | 180 | 181 | +-----------------------+-------------+-----------------------+-----------+----------------+-------------------------+ | |
Collapse file Copy file name to clipboardExpand all lines: java/serving/pom.xml +1-1Lines changed: 1 addition & 1 deletion | Original file line number | Diff line number | Diff line change |
|---|
| 287 | 287 | <dependency> | 288 | 288 | <groupId>org.apache.avro</groupId> | 289 | 289 | <artifactId>avro</artifactId> | 290 | | - <version>1.11.3</version> | | 290 | + <version>1.11.4</version> | 291 | 291 | </dependency> | 292 | 292 |
| 293 | 293 | <!-- https://mvnrepository.com/artifact/org.apache.arrow/arrow-java-root --> | |
Collapse file Copy file name to clipboardExpand all lines: protos/feast/core/Permission.proto +1-1Lines changed: 1 addition & 1 deletion | Original file line number | Diff line number | Diff line change |
|---|
@@ -50,7 +50,7 @@ message PermissionSpec { | 50 | 50 |
| 51 | 51 | repeated Type types = 3; | 52 | 52 |
| 53 | | - string name_pattern = 4; | | 53 | + repeated string name_patterns = 4; | 54 | 54 |
| 55 | 55 | map<string, string> required_tags = 5; | 56 | 56 |
| |
Collapse file Copy file name to clipboardExpand all lines: sdk/python/feast/cli.py +1-1Lines changed: 1 addition & 1 deletion | Original file line number | Diff line number | Diff line change |
|---|
@@ -1211,7 +1211,7 @@ def feast_permissions_list_command(ctx: click.Context, verbose: bool, tags: list | 1211 | 1211 | headers=[ | 1212 | 1212 | "NAME", | 1213 | 1213 | "TYPES", | 1214 | | - "NAME_PATTERN", | | 1214 | + "NAME_PATTERNS", | 1215 | 1215 | "ACTIONS", | 1216 | 1216 | "ROLES", | 1217 | 1217 | "REQUIRED_TAGS", | |
Collapse file Copy file name to clipboardExpand all lines: sdk/python/feast/cli_utils.py +1-1Lines changed: 1 addition & 1 deletion | Original file line number | Diff line number | Diff line change |
|---|
@@ -196,7 +196,7 @@ def handle_not_verbose_permissions_command( | 196 | 196 | [ | 197 | 197 | p.name, | 198 | 198 | _to_multi_line([t.__name__ for t in p.types]), # type: ignore[union-attr, attr-defined] | 199 | | - p.name_pattern, | | 199 | + _to_multi_line(p.name_patterns), | 200 | 200 | _to_multi_line([a.value.upper() for a in p.actions]), | 201 | 201 | _to_multi_line(sorted(roles)), | 202 | 202 | _dict_to_multi_line(p.required_tags), | |
Collapse file Copy file name to clipboardExpand all lines: sdk/python/feast/feature_server.py +2Lines changed: 2 additions & 0 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -100,8 +100,10 @@ def async_refresh(): | 100 | 100 | @asynccontextmanager | 101 | 101 | async def lifespan(app: FastAPI): | 102 | 102 | async_refresh() | | 103 | + await store.initialize() | 103 | 104 | yield | 104 | 105 | stop_refresh() | | 106 | + await store.close() | 105 | 107 |
| 106 | 108 | app = FastAPI(lifespan=lifespan) | 107 | 109 |
| |
Collapse file Copy file name to clipboardExpand all lines: sdk/python/feast/feature_store.py +10-2Lines changed: 10 additions & 2 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -713,7 +713,7 @@ def plan( | 713 | 713 | >>> fs = FeatureStore(repo_path="project/feature_repo") | 714 | 714 | >>> driver = Entity(name="driver_id", description="driver id") | 715 | 715 | >>> driver_hourly_stats = FileSource( | 716 | | - ... path="project/feature_repo/data/driver_stats.parquet", | | 716 | + ... path="data/driver_stats.parquet", | 717 | 717 | ... timestamp_field="event_timestamp", | 718 | 718 | ... created_timestamp_column="created", | 719 | 719 | ... ) | @@ -827,7 +827,7 @@ def apply( | 827 | 827 | >>> fs = FeatureStore(repo_path="project/feature_repo") | 828 | 828 | >>> driver = Entity(name="driver_id", description="driver id") | 829 | 829 | >>> driver_hourly_stats = FileSource( | 830 | | - ... path="project/feature_repo/data/driver_stats.parquet", | | 830 | + ... path="data/driver_stats.parquet", | 831 | 831 | ... timestamp_field="event_timestamp", | 832 | 832 | ... created_timestamp_column="created", | 833 | 833 | ... ) | @@ -2078,6 +2078,14 @@ def list_saved_datasets( | 2078 | 2078 | self.project, allow_cache=allow_cache, tags=tags | 2079 | 2079 | ) | 2080 | 2080 |
| | 2081 | + async def initialize(self) -> None: | | 2082 | + """Initialize long-lived clients and/or resources needed for accessing datastores""" | | 2083 | + await self._get_provider().initialize(self.config) | | 2084 | + | | 2085 | + async def close(self) -> None: | | 2086 | + """Cleanup any long-lived clients and/or resources""" | | 2087 | + await self._get_provider().close() | | 2088 | + | 2081 | 2089 |
| 2082 | 2090 | def _print_materialization_log( | 2083 | 2091 | start_date, end_date, num_feature_views: int, online_store: str | |
Collapse file Copy file name to clipboardExpand all lines: sdk/python/feast/infra/offline_stores/dask.py +33-7Lines changed: 33 additions & 7 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -57,6 +57,7 @@ def __init__( | 57 | 57 | self, | 58 | 58 | evaluation_function: Callable, | 59 | 59 | full_feature_names: bool, | | 60 | + repo_path: str, | 60 | 61 | on_demand_feature_views: Optional[List[OnDemandFeatureView]] = None, | 61 | 62 | metadata: Optional[RetrievalMetadata] = None, | 62 | 63 | ): | @@ -67,6 +68,7 @@ def __init__( | 67 | 68 | self._full_feature_names = full_feature_names | 68 | 69 | self._on_demand_feature_views = on_demand_feature_views or [] | 69 | 70 | self._metadata = metadata | | 71 | + self.repo_path = repo_path | 70 | 72 |
| 71 | 73 | @property | 72 | 74 | def full_feature_names(self) -> bool: | @@ -99,8 +101,13 @@ def persist( | 99 | 101 | if not allow_overwrite and os.path.exists(storage.file_options.uri): | 100 | 102 | raise SavedDatasetLocationAlreadyExists(location=storage.file_options.uri) | 101 | 103 |
| | 104 | + if not Path(storage.file_options.uri).is_absolute(): | | 105 | + absolute_path = Path(self.repo_path) / storage.file_options.uri | | 106 | + else: | | 107 | + absolute_path = Path(storage.file_options.uri) | | 108 | + | 102 | 109 | filesystem, path = FileSource.create_filesystem_and_path( | 103 | | - storage.file_options.uri, | | 110 | + str(absolute_path), | 104 | 111 | storage.file_options.s3_endpoint_override, | 105 | 112 | ) | 106 | 113 |
| @@ -243,7 +250,9 @@ def evaluate_historical_retrieval(): | 243 | 250 |
| 244 | 251 | all_join_keys = list(set(all_join_keys + join_keys)) | 245 | 252 |
| 246 | | - df_to_join = _read_datasource(feature_view.batch_source) | | 253 | + df_to_join = _read_datasource( | | 254 | + feature_view.batch_source, config.repo_path | | 255 | + ) | 247 | 256 |
| 248 | 257 | df_to_join, timestamp_field = _field_mapping( | 249 | 258 | df_to_join, | @@ -297,6 +306,7 @@ def evaluate_historical_retrieval(): | 297 | 306 | min_event_timestamp=entity_df_event_timestamp_range[0], | 298 | 307 | max_event_timestamp=entity_df_event_timestamp_range[1], | 299 | 308 | ), | | 309 | + repo_path=str(config.repo_path), | 300 | 310 | ) | 301 | 311 | return job | 302 | 312 |
| @@ -316,7 +326,7 @@ def pull_latest_from_table_or_query( | 316 | 326 |
| 317 | 327 | # Create lazy function that is only called from the RetrievalJob object | 318 | 328 | def evaluate_offline_job(): | 319 | | - source_df = _read_datasource(data_source) | | 329 | + source_df = _read_datasource(data_source, config.repo_path) | 320 | 330 |
| 321 | 331 | source_df = _normalize_timestamp( | 322 | 332 | source_df, timestamp_field, created_timestamp_column | @@ -377,6 +387,7 @@ def evaluate_offline_job(): | 377 | 387 | return DaskRetrievalJob( | 378 | 388 | evaluation_function=evaluate_offline_job, | 379 | 389 | full_feature_names=False, | | 390 | + repo_path=str(config.repo_path), | 380 | 391 | ) | 381 | 392 |
| 382 | 393 | @staticmethod | @@ -420,8 +431,13 @@ def write_logged_features( | 420 | 431 | # Since this code will be mostly used from Go-created thread, it's better to avoid producing new threads | 421 | 432 | data = pyarrow.parquet.read_table(data, use_threads=False, pre_buffer=False) | 422 | 433 |
| | 434 | + if config.repo_path is not None and not Path(destination.path).is_absolute(): | | 435 | + absolute_path = config.repo_path / destination.path | | 436 | + else: | | 437 | + absolute_path = Path(destination.path) | | 438 | + | 423 | 439 | filesystem, path = FileSource.create_filesystem_and_path( | 424 | | - destination.path, | | 440 | + str(absolute_path), | 425 | 441 | destination.s3_endpoint_override, | 426 | 442 | ) | 427 | 443 |
| @@ -456,8 +472,14 @@ def offline_write_batch( | 456 | 472 | ) | 457 | 473 |
| 458 | 474 | file_options = feature_view.batch_source.file_options | | 475 | + | | 476 | + if config.repo_path is not None and not Path(file_options.uri).is_absolute(): | | 477 | + absolute_path = config.repo_path / file_options.uri | | 478 | + else: | | 479 | + absolute_path = Path(file_options.uri) | | 480 | + | 459 | 481 | filesystem, path = FileSource.create_filesystem_and_path( | 460 | | - file_options.uri, file_options.s3_endpoint_override | | 482 | + str(absolute_path), file_options.s3_endpoint_override | 461 | 483 | ) | 462 | 484 | prev_table = pyarrow.parquet.read_table( | 463 | 485 | path, filesystem=filesystem, memory_map=True | @@ -493,7 +515,7 @@ def _get_entity_df_event_timestamp_range( | 493 | 515 | ) | 494 | 516 |
| 495 | 517 |
| 496 | | -def _read_datasource(data_source) -> dd.DataFrame: | | 518 | +def _read_datasource(data_source, repo_path) -> dd.DataFrame: | 497 | 519 | storage_options = ( | 498 | 520 | { | 499 | 521 | "client_kwargs": { | @@ -504,8 +526,12 @@ def _read_datasource(data_source) -> dd.DataFrame: | 504 | 526 | else None | 505 | 527 | ) | 506 | 528 |
| | 529 | + if not Path(data_source.path).is_absolute(): | | 530 | + path = repo_path / data_source.path | | 531 | + else: | | 532 | + path = data_source.path | 507 | 533 | return dd.read_parquet( | 508 | | - data_source.path, | | 534 | + path, | 509 | 535 | storage_options=storage_options, | 510 | 536 | ) | 511 | 537 |
| |
Collapse file Copy file name to clipboardExpand all lines: sdk/python/feast/infra/offline_stores/duckdb.py +14-3Lines changed: 14 additions & 3 deletions | Original file line number | Diff line number | Diff line change |
|---|
| 27 | 27 | from feast.repo_config import FeastConfigBaseModel, RepoConfig | 28 | 28 |
| 29 | 29 |
| 30 | | -def _read_data_source(data_source: DataSource) -> Table: | | 30 | +def _read_data_source(data_source: DataSource, repo_path: str) -> Table: | 31 | 31 | assert isinstance(data_source, FileSource) | 32 | 32 |
| 33 | 33 | if isinstance(data_source.file_format, ParquetFormat): | @@ -43,21 +43,32 @@ def _read_data_source(data_source: DataSource) -> Table: | 43 | 43 | def _write_data_source( | 44 | 44 | table: Table, | 45 | 45 | data_source: DataSource, | | 46 | + repo_path: str, | 46 | 47 | mode: str = "append", | 47 | 48 | allow_overwrite: bool = False, | 48 | 49 | ): | 49 | 50 | assert isinstance(data_source, FileSource) | 50 | 51 |
| 51 | 52 | file_options = data_source.file_options | 52 | 53 |
| 53 | | - if mode == "overwrite" and not allow_overwrite and os.path.exists(file_options.uri): | | 54 | + if not Path(file_options.uri).is_absolute(): | | 55 | + absolute_path = Path(repo_path) / file_options.uri | | 56 | + else: | | 57 | + absolute_path = Path(file_options.uri) | | 58 | + | | 59 | + if ( | | 60 | + mode == "overwrite" | | 61 | + and not allow_overwrite | | 62 | + and os.path.exists(str(absolute_path)) | | 63 | + ): | 54 | 64 | raise SavedDatasetLocationAlreadyExists(location=file_options.uri) | 55 | 65 |
| 56 | 66 | if isinstance(data_source.file_format, ParquetFormat): | 57 | 67 | if mode == "overwrite": | 58 | 68 | table = table.to_pyarrow() | | 69 | + | 59 | 70 | filesystem, path = FileSource.create_filesystem_and_path( | 60 | | - file_options.uri, | | 71 | + str(absolute_path), | 61 | 72 | file_options.s3_endpoint_override, | 62 | 73 | ) | 63 | 74 |
| |
|