🍁 Asymmetrical Partition I/O -- write to a dynamic number of partitions during incremental load

[mattfysh]

Dagster's definition of partitions as lifted from the docs:

An asset definition can represent a collection of partitions that can be tracked and materialized independently. In many ways, each partition functions like its own mini-asset, but they all share a common materialization function and dependencies. Typically, each partition will correspond to a separate file, or a slice of a table in a database.

However, this assumes that the asset's partitions are being written and read symmetrically, i.e. how you write the asset correlates to how you read the asset. This is not always the case, taking this example also from the docs:

Consider an online store's order data. In a database, the order data might be stored as a single orders table, which contains multiple days' worth of orders. However, if the data were ingested into Amazon Web Services (AWS) S3 as parquet files, you could create a new parquet file per day or partition.

If you were to implement this using the current dagster approach, you would be reading from the table once per partition, each time applying a date filter to retrieve only the relevant data. However, this may represent a significant inefficiency wrt. how the downstream asset is produced. This would be especially noticed if the granularity were smaller (eg. hourly partitions in dagster) and/or the dagster partitions were not aligned with database indexes, or if the query were to place undue load on the db server

A much faster, safer, and less resource-intensive way of materializing the parquet files would be to process the table's transaction log, or change data feed, using an offset cursor to incrementally load orders from the database in batches.

Dagster currently has no primitives to define an asymmetrical read/write, where you write to multiple partitions in a single run, but assets further downstream (who typically would have partition alignment with the parquet files asset) would read on a per-partition basis.

Currently I'm experimenting with having two logical assets to represent the different modes (a non-partitioned incremental load asset, and a partitioned read asset -- both representing the same physically partitioned asset).

But I couldn't help but wonder, would it make sense to implement a feature where an incremental materialization may write to zero, one, or more partitions in a single run?

[easontm]

Related to #22567