Isn't this API overlapping with RetrievalJob.to_bigquery() etc? I think it's a bit confusing to me to have two places to tell Feast to save a dataset.

Yes, you can call job.to_bigquery() or job.to_redshift() and this is what saved dataset implementation does under the hood. The difference is that you will need to add additional DataSource in order to read those saved tables back to Feast. And this is what this PR does automatically via store.get_saved_dataset

it also is slightly unclear to me what the exact contract of RetrievalJob is. RetrievalJob otherwise is a deferred fetch that can be triggered on e.g. to_df() or to_arrow().

The save_as parameter seems to be ambiguous as to when it persists the data. (the implementation is it happens immediately). If we want the timing to be consistent, then it becomes very similar to the to_bigquery way of doing things.

1. How it works: you're right that the original query will be triggered eagerly and the dataset will be persisted right away after calling get_historical_features. The caller will receive a substitution of RetrievalJob that will wrap a much simpler query: select * from persisted_table and process (contract) of materializing this query (job) remains the same (will be triggered on .to_df() or .to_arrow()).
2. Why not postpone persisting until the user will call to_df(): current implementation doesn't cache query results, it means that if I'll call .to_df() 3 times - 3 jobs to BigQuery or Redshift would be created. In case we'll put persisting on that stage - we'll have to overwrite the destination table over and over, which I don't think is good contract. Ideally, we should write once and fail if the table already exists and not risk deleting data created by the user and not Feast.

it also is slightly unclear to me what the exact contract of RetrievalJob is. RetrievalJob otherwise is a deferred fetch that can be triggered on e.g. to_df() or to_arrow().

The save_as parameter seems to be ambiguous as to when it persists the data. (the implementation is it happens immediately). If we want the timing to be consistent, then it becomes very similar to the to_bigquery way of doing things.

is there a reason you made this a set? should probably call this out in the description of the PR too since it can cause a behavior change

I was trying to reuse pull_latest_from_table_or_query because it's already implemented in all offline stores including extensions. But those implementations also include deduplication, which shouldn't happen for retrieving a saved dataset.
Now I see that it probably wasn't a good idea. I will use a separate method.

Do we want this to be a publically facing method? Seems confusing to a user why there'd be

1. this persist method, which happens lazily.
2. to_bigquery, which exists right next to this method and also saves.

I think the return type can be confusing too since .persist has indeed executed a query & persisted, but we return a RetrievalJob. I might as a user expect to need to call RetrievalJob#to_bigquery again, which actually will persist to a different location.

Two options I think could make sense:

• Have this be an internal only method by name so people don't try to call it
• Expose a separate way to instantiate a RetrievalJob (like RetrievalJob.from_saved_dataset which executes this wrapped query) so you can return it in get_historical_features

persist doesn't happen lazily. It's generic equivalent of methods to_bigquery or to_redshift which exist only in specific implementation classes. predict is part of abstract API (RetrievalJob) and aimed to unify these methods. Under the hood, it's calling to_bigquery or to_redshift.

The feature that is absent in previous methods is that persist returns RetrievalJob, which supposedly retrieves from a persisted table.

Why it's public? As we discussed on call with Shopify - it could be useful if the user doesn't want get_historical_features to run query eagerly. In this case, user would not pass save_as parameter but call persist themselves:

job = fs.get_historical_feature(...)
persisted = job.persist(storage=SavedDatasetBigQueryStorage(...))
persisted.to_df()

Regarding cyclical calls, I think that we could deprecate specific methods like .to_bigquery() and then it should be less confusing.

Expose a separate way to instantiate a RetrievalJob (like RetrievalJob.from_saved_dataset which executes this wrapped query) so you can return it in get_historical_features

This option makes sense to me as well. Please let me know what you think is better @adchia

UPD. persist signature updated. Now it doesn't return RetrievalJob to avoid duplication of responsibilities. Read of saved dataset moved to Provider.read_saved_dataset, which uses OfflineSource.pull_all_from_table_or_query.

So now .persist() is just a universal replacement for .to_bigquery() and .to_redshift().

probably makes sense to check it's not some other file format like .csv which we don't support

In this else block we treat the destination as a directory, so it could be named even as dir.csv -> dir.csv/ folder would be created.

Added comment line

not a blocking comment. More of an FYI

fyi @felixwang9817 since you are adding a new concept to the registry. Don't think feast plan could cover this since this isn't introduced at feast apply time.

Also doesn't really make sense to treat this as a real FCO since the schema is a snapshot of how feature services / feature views resolve at a point in time. Doubt we'd be able to do this properly unless we have some form of stricter versioning.

Don t we have an issue if the data is store in multiple storages ? I though that all the data should be in the same offline store (eg: BigQuery) so that we can join any tables between them.

By allowing the "saved dataset" to be stored in another offline storage, it would mean that we will never be able to JOIN a saved dataset with a feature view. Any thoughts on that ?

You won't be able to pass here different storage other than the one you have configured as a global offline store. There will be an exception if your features are being pulled from BigQuery and you will try to store them into Redshift.
This is just a generic API to support typed configuration for all kinds of stores. I couldn't find a better way (at least in Python type system is not so strong). If you know how to do it better please let me know.

Also I would believe that we should provide a "smart default" here with storage=... is inferred and will use the same one as the offline store specified in the config

We can infer the type of storage, but each type will still require its own parameters (eg, table_ref for BQ, path for File) and we need those parameters provided by user anyway. So those classes (SavedDatasetBigQueryStorage, SavedDatasetRedshiftStorage, etc) are useful at least for parameters type checking.

Is this when the saved dataset is computed ?

If by compute you mean materialized, then no. The saved dataset is computed eagerly on creation. For example, when you call get_historical_features with the save_as parameter saved dataset would be already materialized & stored to offline store & registry by the time this function returns.

For example, when you call get_historical_features with the save_as parameter saved dataset would be already materialized

That's where it could be confusing because fs.get_historical_features() does not materialize anything normally but it woudl do it when save_as is passed. Two completely different behaviour with the same method.

I still want to think out-loud about the behaviour of this new API. I just want to understand the pros/cons of each approach.

I understand that you want to have the saved dataset computed only when explicitly asked by the user.

Let's assume that store.get_historical_features() should only return one type of object (aka RetrievalJobs).
What do you think about the following ?

store = FeatureStore()
offline_job: OfflineRetrievalJob = store.get_historical_features(
    features=["driver:avg_trip"],
    entity_df=...,
)

saved_dataset: SavedDatasetJob = store.create_saved_dataset(
    name='my_training_dataset',
    storage=SavedDatasetBigQueryStorage(table_ref='<gcp-project>.<gcp-dataset>.my_training_dataset'), # Should be inferred by default and use the offline storage of the retrieval job
    retrieval_job=offline_job, # Any RetrievalJob should have methods available to write the output in the desired SavedDatasetStorage
    tags={'author': 'oleksii'},
)

retrieval_job: OfflineRetrievalJob = store.get_saved_dataset('my_training_dataset')
retrieval_job.to_df()

WDYT ?

Yep, this could work. But what are benefits of this approach?

1. Do you want to reuse offline_job? But in that case, historical query can be run several times, which is additional costs:

offline_job = store.get_historical_features(...)
store.create_saved_dataset(offline_job)
offline_job.to_df()  # historical query (with point-in-time join) executed twice

Whereas in the current proposed version saved dataset plays the role of cache. Eg,

result = store.get_historical_features(..., save_as='...')
result.to_df()
result.to_df() # both times reads from persisted table, not running the original query

2. Will create_saved_dataset run offline_job eagerly & what is the purpose of SavedDatasetJob?

But what are benefits of this approach?

Mostly consistency: store.get_historical_features() always return an OfflineRetrievalJob object

Do you want to reuse offline_job?

I don't want to have strong assumptions on that. What I know is that the BQ Retrieval job has a dry-run option. So I would not be surprised if some users use it multiple times.

query can be run several times, which is additional costs

That's on the user to decide if this additional cost is useful or not IMO

Whereas in the current proposed version saved dataset plays the role of cache

I get that and this is great 👍

what is the purpose of SavedDatasetJob?

Now I understand it better, I think that it's not needed. Here is my updated example

store = FeatureStore()
offline_job: OfflineRetrievalJob = store.get_historical_features(
    features=["driver:avg_trip"],
    entity_df=...,
)

retrieval_job: OfflineRetrievalJob = store.create_saved_dataset( # That operation executes the RetrievalJob
    name='my_training_dataset',
    storage=SavedDatasetBigQueryStorage(table_ref='<gcp-project>.<gcp-dataset>.my_training_dataset'),
    retrieval_job=offline_job,
    tags={'author': 'oleksii'},
)

retrieval_job.to_df()

As I said, I am thinking out-loud

API is updated.

Is the BigQuery offline store can only persist the SavedDataset into BigQuery, why making it possible to pass any SavedDatasetStorage and not infer it ?

As described above, that's merely for type checking: good for verification on development and could be helpful when using Feast in some kind of IDE.

If I understand correctly, you suggesting just plain dict:

fs.get_historical_features(..., save_as={'storage': {'table_ref': '...'}})

vs proposed API

fs.get_historical_features(..., save_as=SavedDatasetOpts(storage=SavedDatasetBigQueryStorage(table_ref=...))

just to free you from remembering those dict keys.

If I understand correctly, you suggesting just plain dict

Or sending parameters right away. Something like

get_historical_features(..., save_as=SavedDataset(table_ref="...")) # Example 1
get_historical_features(..., save_as=SavedDataset(table_ref="...", table_ttl=timedelta(days=30))) # Example 2
get_historical_features(..., save_as=SavedDataset(file_path="...")) # Example 3

# Under the hood
@dataclass
class SavedDataset:
    table_ref: Optional[str] = None
    table_ttl: Optional[timedelta] = None
    file_path: Optional[str] = None

def get_historical_features(..., save_as: SavedDataset):
    SavedDatasetOptions = config.offline_store.get_saved_dataset_options(**asdict(save_as))

Unsure if that makes it more confusing or easier to use for our users TBH 🤔

I think subclasses is a better approach:

1. Easier to extend (add new kind of storage)
2. You can have mandatory & optional parameters, whereas with a single class they all are optional.

Plus I believe this approach (more strict types) is used in various places in Feast, so keeping things consistent is good.

I think subclasses is a better approach

Sounds good to me 👍