[dagster-dbt] Add general attach_metadata method to chain async metadata fetches to dbt execution #21947

benpankow · 2024-05-17T20:14:57Z

Summary

Continues the restructuring of chaining async dbt work from #12943 to specify a attach_metadata method which can be used to specify a function which will run in a thread after each dbt event completes and will attach any additional user-specified metadata to that event before it is emitted.

Right now, it's classified as private so we can adjust the signature if need be before making it public - I'd like to unblock future work stacking on this shared impl, but I think we'll want to pass a single, proper context class rather than the loose invocation, event params.

Used internally for the fetch_row_count implementation.

Test Plan

Example unit test with SUMMARIZE SQL metadata.

benpankow · 2024-05-17T20:15:11Z

This stack of pull requests is managed by Graphite. Learn more about stacking.

Join @benpankow and the rest of your teammates on Graphite

rexledesma

Let me know if it's ready for review -- It's still in draft, so I'm not sure if you were planning on more changes.

benpankow · 2024-05-21T22:28:33Z

Marked as ready-for-review, changed method to be non-public for now since I am not confident in what signature we'll want to expose and want to unblock stacking on it (it's useful as an internal abstraction at least)

rexledesma · 2024-05-23T12:39:30Z

python_modules/libraries/dagster-dbt/dagster_dbt/core/resources_v2.py

@@ -985,6 +987,32 @@ def fetch_row_counts(
                A set of corresponding Dagster events for dbt models, with row counts attached,
                yielded in the order they are emitted by dbt.
        """
+        return self._attach_metadata(_fetch_row_count_metadata)
+
+    def _attach_metadata(


We're planning on making this a public method in the future, right? We should think of that name here. We don't have to add the @public decorator though. I think map could be a compelling name here. We'd need to make the fn type simpler: Callable[DbtDagsterEventType, DbtDagsterEventType].

I can imagine this method to operate on modifying the general event stream, rather than just attaching metadata. This could be a pattern that exists outside of dagster-dbt.

Right now the function is is useful as an internal simplification/util. If we want to expose it publicly, landing on the right name/signature is more tricky and warrants a more deliberate discussion. There are trade-offs either way - if we want to make it easy for users to easily attach metadata, we can more easily cache results or fold map calls together vs giving them power to map over all events. Either way I think it's beyond the scope of this PR and I don't feel that it's blocking what's here - we're not committing to anything name or signature wise.

rexledesma · 2024-05-23T12:41:06Z

python_modules/libraries/dagster-dbt/dagster_dbt/core/resources_v2.py

+        """Runs a threaded task to attach metadata to each event in the iterator.
+
+        Args:
+            fn (Callable[[DbtCliInvocation, DbtDagsterEventType], Optional[Dict[str, Any]]]):


I think including DbtCliInvocation is unnecessary.

We can access it from self, if we make _dbt_cli_invocation public.

From the user's perspective, DbtCliInvocation will be accessible from the scope outside of the closure, so it's repetitive.

As I mentioned in the PR description, I think the signature here will change pretty meaningfully if we expose it publicly - probably some sort of context class rather than the loose arguments that it takes now. I'd prefer to leave as-is for now and re-examine it down the line if we want to expose this method publicly.

Got it, I missed that. Makes sense.

rexledesma · 2024-05-23T12:43:31Z

python_modules/libraries/dagster-dbt/dagster_dbt/core/resources_v2.py

            Iterator[Union[Output, AssetMaterialization, AssetObservation, AssetCheckResult]]
        ):
            with ThreadPoolExecutor(
                max_workers=self._dbt_cli_invocation.postprocessing_threadpool_num_threads,
-                thread_name_prefix="fetch_row_counts",
+                thread_name_prefix=fn.__name__,


Hmm we would have to check this: I'm wondering what would happen if the user passes in a lambda, rather than a declared function. We might have to manage the name on their behalf or something.

Added a prefix to the fn name which at least points to this being the mapping fn, but not sure what else we can do here. If a user ends up defining many custom lambda postprocessing steps and has to figure out which isn't working and can't figure it out from stacktrace/other context they can always give them names.

rexledesma · 2024-05-23T12:46:47Z

python_modules/libraries/dagster-dbt/dagster_dbt/core/resources_v2.py

+            if result is None:
+                return event
+
+            return event.with_metadata({**event.metadata, **result})


One thing here: if we simplify the fn type, we'd expect users to call .with_metadata.

I think a better name for .with_metadata is .replace_metadata.

And for most users, they would call something like .add_metadata, which would call .replace_metadata, but tack on the **event.metadata on the user's behalf.

Agreed - we tend to use the with_ pattern in many places for replacement behavior (e.g. AssetsDefinition) but the behavior is definitely not clear from name alone (vs a clear distinction if we used with_added_ and replace_ instead)

…ata fetches to dbt execution

rexledesma

Approving since this is basically just an internal refactor

benpankow mentioned this pull request May 17, 2024

[dagster-dbt] Refactor row count collection code #21943

Merged

benpankow force-pushed the benpankow/generify-dbt-2 branch from 4fd65d3 to ab4cd89 Compare May 20, 2024 16:31

benpankow force-pushed the benpankow/generic-dbt-3 branch from 274a65e to f42ccaa Compare May 20, 2024 16:31

benpankow mentioned this pull request May 20, 2024

[dagster-dbt] add chained method to fetch column metadata, lineage async #21979

Merged

benpankow force-pushed the benpankow/generify-dbt-2 branch from ab4cd89 to 02be352 Compare May 20, 2024 20:17

benpankow force-pushed the benpankow/generic-dbt-3 branch from f42ccaa to 216a3d2 Compare May 20, 2024 20:18

benpankow force-pushed the benpankow/generify-dbt-2 branch from 02be352 to abb8b7e Compare May 20, 2024 23:22

benpankow force-pushed the benpankow/generic-dbt-3 branch from 216a3d2 to aa8a372 Compare May 20, 2024 23:22

benpankow requested review from rexledesma and sryza and removed request for rexledesma May 20, 2024 23:38

rexledesma reviewed May 21, 2024

View reviewed changes

benpankow force-pushed the benpankow/generify-dbt-2 branch from abb8b7e to b35eeb9 Compare May 21, 2024 17:15

benpankow force-pushed the benpankow/generic-dbt-3 branch 2 times, most recently from daf0867 to 7af52c3 Compare May 21, 2024 18:25

benpankow marked this pull request as ready for review May 21, 2024 22:28

benpankow requested a review from rexledesma May 21, 2024 22:28

benpankow force-pushed the benpankow/generify-dbt-2 branch from 965d062 to 5285767 Compare May 21, 2024 23:04

benpankow force-pushed the benpankow/generic-dbt-3 branch from 6d1efcc to 5c33b4d Compare May 21, 2024 23:04

benpankow force-pushed the benpankow/generify-dbt-2 branch from 5285767 to d7fe9db Compare May 22, 2024 15:49

benpankow force-pushed the benpankow/generic-dbt-3 branch 2 times, most recently from 9adf0cb to fcfc015 Compare May 22, 2024 16:05

benpankow force-pushed the benpankow/generify-dbt-2 branch from 0a33daf to a74481a Compare May 22, 2024 16:26

benpankow force-pushed the benpankow/generic-dbt-3 branch from fcfc015 to e70d560 Compare May 22, 2024 16:26

Base automatically changed from benpankow/generify-dbt-2 to master May 22, 2024 18:18

benpankow force-pushed the benpankow/generic-dbt-3 branch 2 times, most recently from d3d3066 to 6609790 Compare May 23, 2024 03:44

rexledesma requested changes May 23, 2024

View reviewed changes

benpankow added 6 commits May 23, 2024 08:26

[dagster-dbt] Add general attach_metadata method to chain async metad…

252e822

…ata fetches to dbt execution

fix test

cb4bcad

make private method for now

ad3e83a

rename

0ef125d

prefix

008b0f1

add threadname prefix

2dad48e

benpankow force-pushed the benpankow/generic-dbt-3 branch from 6609790 to 2dad48e Compare May 23, 2024 15:29

rexledesma approved these changes May 23, 2024

View reviewed changes

benpankow merged commit 361184c into master May 23, 2024
1 check passed

benpankow deleted the benpankow/generic-dbt-3 branch May 23, 2024 16:08

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[dagster-dbt] Add general attach_metadata method to chain async metadata fetches to dbt execution #21947

[dagster-dbt] Add general attach_metadata method to chain async metadata fetches to dbt execution #21947

benpankow commented May 17, 2024 •

edited

benpankow commented May 17, 2024 •

edited

rexledesma left a comment

benpankow commented May 21, 2024

rexledesma May 23, 2024

benpankow May 23, 2024

rexledesma May 23, 2024

benpankow May 23, 2024 •

edited

rexledesma May 23, 2024

rexledesma May 23, 2024

benpankow May 23, 2024

rexledesma May 23, 2024

benpankow May 23, 2024

rexledesma left a comment

[dagster-dbt] Add general attach_metadata method to chain async metadata fetches to dbt execution #21947

[dagster-dbt] Add general attach_metadata method to chain async metadata fetches to dbt execution #21947

Conversation

benpankow commented May 17, 2024 • edited

Summary

Test Plan

benpankow commented May 17, 2024 • edited

rexledesma left a comment

Choose a reason for hiding this comment

benpankow commented May 21, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

benpankow May 23, 2024 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rexledesma left a comment

Choose a reason for hiding this comment

benpankow commented May 17, 2024 •

edited

benpankow commented May 17, 2024 •

edited

benpankow May 23, 2024 •

edited