[Feature] DbtCompileOperator + allow `partial_parse.msgpack` to be read from S3/GCS/Azure Blob Storage.

[dwreeves]

Our CICD uses a different config than what Cosmos does, and the difference in the profile is close to unavoidable, which makes it so partial parsing doesn't actually work:

Unable to do partial parsing because config vars, config profile, or config target have changed
Unable to do partial parsing because profile has changed
Unable to do partial parsing because env vars used in profiles.yml have changed

I think the only way you can reasonably get this to work is to have a DbtCompileOperator, and then read the partial_parse.msgpack from S3.

The output directory should be templated, ideally, so that you can use ti.xcoms_pull() or otherwise avoid any "clashing" across multiple simultaneous DAG runs. (Then at the end of each DAG run the user can do e.g. S3DeleteObjectsOperator() to clean things up at their leisure.)

[dosubot]

It seems like you've got this under control, if you want help or have specific questions, let me know what I can do for you!

About Dosu

This response is meant to be useful and save you time. It is not meant to be a precise solution, but rather a starting point for your own research.

Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.

If you want to continue the conversation, start your reply with @dosu-bot.

[yusufshalaby]

This would be a gamechanger. Similar to Dagster's implementation where models are split into different data assets but a full dbt run only requires a single dbt parse at the beginning. This would unlock that functionality in Airflow, where we can run each model as a seperate Airflow task that reads from the same cached manifest rather than forcing each task to perform a dbt parse and incurring a high cpu cost. Previous solutions involved building the manifest during CI however this prevents users from injecting any vars at runtime.

I've given some thought to writing an Airflow task that parses the dbt project and passes the manifest to each Cosmos dbt task, but it's tough when you're using Celery/Kubernetes executor with workers spread across multiple machines.

[dwreeves]

I've somewhat caught up on how partial parsing is implemented in 1.4.0, and I'm thinking this really should be two issues:

1. Support both reading from and writing to S3/GCS/Azure Blob Storage with dbt artifacts post-execution.
2. Support DagRun-level parameterization of the target path. (To more elegantly support e.g. dbt_vars={"logical_date": "{{ logical_date }}"})

Will have to think a little more about this. I know there are some use cases for dbt artifacts post-execution in addition to just using the partial_parse.msgpack for subsequent executions, and I want to make sure we are supporting that (or, at least, we are designing in a direction that will support that), since the caching directory approach taken in 1.4.0 will work OK enough as is for many simpler use cases.

There is a bit of overlap in the writing part with what the dbt docs operator already does, which is both a blessing and a curse. A blessing because we already have an API (we don't need to reinvent the wheel), but a curse because we already have an API (we don't want to create two different pathways to interacting with S3/Azure/GCS). Unfortunately mimicking exactly the pattern used by the dbt docs operators, and having something like DbtLocalRunS3Operator, would explode the operator space 4x or 5x, and that's untenable.