How do I configure the DuckDB IO manager to load from a custom schema?

[malow106]

Hi all,

I implement DBT+DuckDB pipeline.
I try to load data from duckDB/DBT asset into dagster asset.

My dbt_project.yml :

models:
  DBT_proj:
    DATAMART:
      +schema: datamart
      +materialized: table
      +meta:
        dagster:
          group: "DBT_PUBLISHING"
          auto_materialize_policy:
            type: 'lazy'

My profiles.yml :

duck_profile:
  target: default
  outputs:
    default:
      type: duckdb
      path: my_db.duckdb
      schema: dev

The duckdb/dbt generated schema is the concatenation of both profiles.yml and dbt_project.yml schemas
-> for my table : dev_datamart.stocks

Here my dagster asset to get this table :

@asset(
    ins={
        "stocks": AssetIn(
            key=get_asset_key_for_model([my_dbt_assets], "stocks"),
        )
    }
)
def export_stocks(stocks):
   ...

The error :

dagster._core.errors.DagsterExecutionLoadInputError: Error occurred while loading input "stocks" of step "export_stocks"::

duckdb.duckdb.CatalogException: Catalog Error: Table with name stocks does not exist!
Did you mean "dev_datamart.stocks"?
LINE 1: SELECT * FROM datamart.stocks
                      ^

Dagster seems not considering profiles.yml schema, is it intentional ?
As a workaround tried to customize get_asset_key_for_model asset key but this is out of my skills..

Thanks !

[jamiedemaria]

You can set the schema for duckdb on the IO manager configuration

defs = Definitions(
   assets=[...],
   resources={
        "io_manager": DuckDBPandasIOManager(schema="dev", ...)
   }
)

[malow106]

I have implemented this code but dagster told me :
dagster._core.errors.DagsterInvalidDefinitionError: Asset ['dagster', 'MVKE'] specifies a schema with its key prefixes ['dagster'], but schema dev was also provided via run config. Schema can only be specified one way.

I wanna apply the schema only to handle DBT related assets, other assets already has their config.

@asset(
    ins={
        "stocks": AssetIn(
            key=get_asset_key_for_model([my_dbt_assets], "stocks"),
        )
    },
    key_prefix=["dev"])
)
def export_stocks(stocks):

Output :
dagster._core.errors.DagsterInvalidSubsetError: Assets provided in asset_selection argument ["export_stocks"] do not exist in parent asset group or job.

At the end my pipeline must work with both SQL Server for prod and DuckDB for dev

[jamiedemaria]

ah yeah. this is a bug. I'm pretty sure it should be a quick fix. I'll put up a PR

[jamiedemaria]

here is the PR if you'd like to follow it #17904

[malow106]

here is the PR if you'd like to follow it #17904

Hi,

Do you think there is a quick workaround while waiting for the PR merge ?

[jamiedemaria]

You could also set the schema for the dbt assets via metadata. You want to have the metadata dict contain {"schema": "<the schema you want>"} i think you'll need to do this by customizing the dbt metadata https://docs.dagster.io/integrations/dbt/reference#customizing-metadata