How can I use the

runtime_metadata_fn

function to retrieve the number of created rows if DBT was materializing a table? The run_results.json in

results.adapter_response.rows_affected

seems to contain the desired value. How is this different to:

*node_info_to_definition_metadata_fn*

?

hi @geoHeil! the

runtime_metadata_fn

takes two arguments, one being the context and the other being the node info. so, one option here would be to have your runtime_metadata_fn use the dbt resource available on the context to fetch the run_results.json object. This would be something like:

def my_runtime_metada_fn(context, node_info):
    run_results = context.resources.dbt.get_run_results_json()
    unique_id = node_info["unique_id"]
    # ... parse out rows_affected from adapter response
    return {"rows_affected": rows_affected}

How would this work (the assigment to an asset key and the metadata? DBT run is returning metadata for many assets I guess?

runtime_metadata_fn is invoked once for each output asset

So for an example like this:

{
  "metadata": {
    "dbt_schema_version": "<https://schemas.getdbt.com/dbt/run-results/v4.json>",
    "dbt_version": "1.4.6",
    "generated_at": "2023-05-23T15:54:40.621481Z",
    "invocation_id": "e9e2e65a-1307-476c-b68c-ddf8f7c55128",
    "env": {}
  },
  "results": [
    {
      "status": "success",
      "timing": [
        {
          "name": "compile",
          "started_at": "2023-05-23T15:54:20.016828Z",
          "completed_at": "2023-05-23T15:54:20.021289Z"
        },
        {
          "name": "execute",
          "started_at": "2023-05-23T15:54:20.021870Z",
          "completed_at": "2023-05-23T15:54:40.500335Z"
        }
      ],
      "thread_id": "Thread-1 (worker)",
      "execution_time": 20.486584186553955,
      "adapter_response": {
        "_message": "SELECT 3196779",
        "code": "SELECT",
        "rows_affected": 3196779
      },
      "message": "SELECT 3196779",
      "failures": null,
      "unique_id": "<<<my_unique_id>>"
    }
  ],
  "elapsed_time": 21.433478832244873,
  "args": {
    "write_json": true,
    "use_colors": true,
    "printer_width": 80,
    "version_check": true,
    "partial_parse": true,
    "static_parser": true,
    "profiles_dir": "xxxx",
    "send_anonymous_usage_stats": true,
    "quiet": false,
    "no_print": false,
    "cache_selected_only": false,
    "target": "<<<<",
    "select": [
      "kkkk"
    ],
    "which": "run",
    "rpc_method": "run",
    "indirect_selection": "eager"
  }
}

Would dagster automatically parse the right asset key to each individual element of the

results

? If yes, how is dagsters asset key filtering to the right element in the results object? I would like to get the

adapter_response.rows_affected

for each result (or at leats for each result where the materialization is of type table). I have to admit, it is still unclear to me how the right filter/assigment between asset key and unique id would look like. Assuming asset key and uniqe ID are identical - would I need to (manually) search the array? However, most likely they are not identical (as the asset key is only constructed ) by feeding additional parameters (key_prefix, source_key_prefix) to the load assets form DBT function. So I would somehow need to unpack/unparse these as well?

the

node_info

argument to

runtime_metadata_fn

is the specific node info associated with a given asset key (so technically the asset key is not even an input to this function). To expand on the pseudocode, it'd look like:

def my_runtime_metada_fn(context, node_info):    
    run_results = context.resources.dbt.get_run_results_json()
    unique_id = node_info["unique_id"]
    
    for result_dict in run_results["results"]:
        if result_dict["unique_id"] == unique_id:
           return {"rows_affected": result_dict["adapter_response"]["rows_affected"]}
    return {}

so you/dagster are already remapping the asset key to the unique ID?

yep exactly

cool - let me try it and give it a shot

ah I think I might know what's happening here -- are you on dbt-core version >= 1.4?

should be 1.4.6

for those versions, we emit the outputs as the dbt command is running, rather than waiting for the command to complete, and so the run_results.json will not be available at that time (forgot about that quirk). if that's the case, then there's not a great existing solution with

load_assets_from_dbt_manifest

. However, in this week's release we're adding a new

dbt_assets

decorator that provides much more flexibility in how to handle events. This is still a work in progress / experimental, so things might shift around a bit, but here's a tentative example of how adding custom metadata might work: https://github.com/dagster-io/dagster/pull/14301/files#diff-acd3f930fb0f90b373feafce2bdaea26a238014bcb6d1dc757b1a578ced14cdbR10

would it be more practical if I send the manifest file?

cc @rex

@owen @rex https://dagster.slack.com/archives/C04CW71AGBW/p1685051912387029 the new version seems to be released now. Can you explain me in an example how this issue would be solved now with the new version?

We gave some examples in the repository. You can compose them to accomplish your use case. • This example, which fetches the artifact after the run is done • This example, which allows you to add metadata to an output

The @dbt_asset requires the manifest. Does this mean in the future you no longer support the dynamic generation of the maifest like: load_assets_from_dbt_project? Where is the

io_manager_key

specified in the new approach?