Caching data frames to disk between executions in an automated/smart way

We have a system where multiple different tables are being computed to eventually be loaded into a database. Some of them do use each other for some computation. A common structure would be something looking like:

class Store:

    def products():
        ...
        return products # A data frame

    def sales():
        products = self.products()
        ...
        return sales # A data frame

Now let's say I've completed the whole pipeline for products and I'm just now working on sales. Every time I run the code to do some test, the whole products pipeline needs to be computed (possibly takes quite long), but I do know it is not changing between executions.

Because of that, I'm working on an automated caching system that can save intermediate results to disk and be able to identify if something has changed or not, so it caches again or retrieves already computed data from disk.

Not that the specific method is particularly relevant for the question but I'm using a decorator for each of these methods that can try to automate the behavior, so the code actually looks more like:

class Store:

    @smartdecorator
    def products():
        ...
        return products # A data frame

    @smartdecorator
    def sales():
        products = self.products()
        ...
        return sales # A data frame

My problem is then, how do I identify that what I'm trying to compute is already cached in disk?

My approach has been to produce a hash/digest of the query plan to be able to identify that I already know what the result will be (cached in the disk). So for the case of the products for example I would save it in a file named products_<hash_of_query_plan>. The next time I just need to recompute that hash and check if I already have it on disk.

The way I'm getting that hash, or some kind of unique identifier of that the results will be is basically the whole question. I've experimented with:

def df_digest(df):
    m = hashlib.sha256()
    query_plan = df._jdf.queryExecution().simpleString()
    m.update(query_plan)
    return m.hexdigest()

I've also tried toString instead of simpleString.

In any case, it seems like a dataframe that would contain exactly the same data as before, is outputting a different digest, which leads me to believe that there's some element to that queryExecution string that is unique to each time I run it.

Does anyone have any idea of what else I could use to create that digest?

NOTE: The fact that this would work and actually be a performance improvement is heavily based in spark being lazy when evaluating. Assuming I don't have any actions in the code and am really only defining transformations, no computation is performed when I retrieve the data frame to produce a digest of its query plan.