I have a job which should conduct a backfill for partitioned assets. The assets and the job have the daily partitions definition attached. However, only the normal-non backfill jobs shows the assets as green after materialization. The backfill job successfully materializes the assets. However, the partition does not show up as green - and furthermore the backfill is also not possible (no backfills for <<<job_name>>> is written on the page of this job

I'm not entirely following what you mean by "However, only the normal-non backfill jobs shows the assets as green after materialization". Mind expanding? Is there a screenshot you could share? What version are you on? cc @Ben Gotow - I know we had some bugs related to the Launch Backfill dialog recently

the latest 1.1.7

If this is in cloud, do you have a link to the partitions page in question?

sure - I even sent you this link on Friday or Saturday to share it in the dagster private slack ... 😉

ah sorry, yes you did

The Launch Backfill button is no longer there, but you should be able to replicate the same functionality by selecting partitions when you click the Materialize button on the tab that has the asset graph Daniel sent me a link to your job. I noticed that the assets inside it don't appear to be partitioned - is that expected? In general, partitioned asset jobs are only meant to be used with partitioned assets

Well it is statefully partitioned SCD2 assets. And these need a backfill

in Dagster, backfills by definition work over partitioned assets - the point of a Dagster backfill is to be able to launch separate work for each partition. do you want to have separate runs for different subcomponents of your SCD2 assets? or just a single run to backfill the entire asset? if the latter, I'd recommend adding boolean configuration options to those assets that you can toggle to recompute them from scratch, instead of using Dagster backfills

so you mean a boolean config plus a tqdm for loop ... hm this might perhaps be the better solution

Our normal recommendation for this kind of thing would be a config value that’s passed to all the assets, but I could see how a tag would be more convenient if you don’t care about being able to validate the config before launching the run

One question still remains open for me: where to put the for loop. I would have loved to use the partitions/dagster itself to run the loop - potentially parallelized, neatly isolated like normal jobs and nicely logged as well. Are you saying with this backfill=true config passed should I have a manual for loop inside the asset which a) discovers available raw partitions b) deletes the existing state c) backfills (loops the loop) or could/should parts of this be in a sensor which is launching jobs against dagster? What are best practices for such a case (I guess SCD2 stateful tables are rather common)

I think I'm struggling a bit to understand how you can have executions for different raw partitions happen independently but they're not idempotent could you potentially model the SCD2 asset as a partitioned asset where each partition depends on the prior partition?

Is this a possibility? How could/would this work?

partition self-dependencies are something that we just added support for. self-dependencies aren't yet respected in the backfill code, but I'm working on adding that right now. here's an example:

@asset(
        partitions_def=DailyPartitionsDefinition(start_date="2020-01-01"),
        ins={
            "a": AssetIn(
                partition_mapping=TimeWindowPartitionMapping(start_offset=-1, end_offset=-1)
            )
        },
    )
    def a(a):
        ...

This is super interesting! I was not yet aware of this functionality. Do you already have a timeline when you plan to add this in for backfill?

my dream is to get it done this week, but realistically it will be more like early january

Really awesome! Please keep me updated on the progress of this feature.

yes - that should still work

_TimeWindowPartitionMapping.__new__() got an unexpected keyword argument 'start_offset'

it was included in the most recent release - are you possibly on an earlier version?

it was 1.1.6 - now with 1.1.7 this works. However, it is still unclear to me how to pass the TimeWindowPartitionMapping to the define_asset_job - I can only pass the partition definition as an argument.

You don’t need to pass it to the job - just on the AssetIn like in my example above

I have a little bit different situation: you have asset a with self-reference to a. I have a ---> b where a is delivering full copies of the data daily and b is the scd2 table. b gets a daily update of a and runs the merge into operation given the fresh data and its state. What I would understand is you are saying here is that I need to explicitly get hold of the existing state as the self reference ... this seems quite complicated (but perhaps the only sane way)

my understanding of your original goal was that you wanted to be able to launch a backfill that filled in your scd2 table in a set of sequential steps, and you had to fill in day N before day N + 1. is that still right? if so, I don't think you should need to modify your I/O manager. you can specify the self-partition-dep without actually needing to have your I/O manager load that data: you can use a Nothing type on the

AssetIn

that corresponds to the self-dep. in our next release (next Thursday), asset backfills will respect the ordering of self-partition-deps

Yes this would be ideal

I don't think I'll be able to video call this week, but if you have a github gist I could take a look?

no worries - we could also do next. This one is also bad for me. Here you go: https://gist.github.com/geoHeil/12ce1e1403e474b44a84fd267323acb4 perhaps this alleviates the need for a call

I just took a look at your example. I don't think you should need to change anything except to add a Nothing self-dependency. It would look something like this:

from dagster import asset, DailyPartitionsDefinition, TimeWindowPartitionMapping, AssetIn, Nothing


@asset(partitions_def=DailyPartitionsDefinition(start_date="2020-01-01"))
def a():
    ...


@asset(
    partitions_def=DailyPartitionsDefinition(start_date="2020-01-01"),
    ins={
        "a_scd2": AssetIn(
            dagster_type=Nothing,
            partition_mapping=TimeWindowPartitionMapping(start_offset=-1, end_offset=-1),
        )
    },
)
def a_scd2(a):
    ...

(there's actually a bug that causes the above to error currently, but it will get fixed in the next release)

But is this best practice? Or would you recommend to refactor and retrieve the self reference from dagster (without the nothing dependency)?

yes - that issue with Nothing arguments to AssetIn is now fixed