Don't include descriptions and metadatas on job snapshots for outputs of asset job ops (#33092)

### Summary

For assets, these descriptions and metadata can already be found on the asset graph, and can be quite large. Including them on the output metadata in the snapshot as well (especially for __ASSET_JOB and subsets therein) significantly bloats the size of the snapshot.

On a representative large snapshot of a single dbt node, byte size is reduced by about 10x: (and the metadata is of course still available on the asset graph)

```
Original size: 52,271,581 bytes
Stripped size: 4,774,884 bytes
Size reduction: 47,496,697 bytes (90.87%)
```

### Test Plan

new test cases, sanity checked local UI in the job snapshot page

### Changelog

Performance improvements when launching runs of asset jobs, by removing unnecessary op output metadata on the job snapshot that is computed for each run.

Author: gibsondan

python_modules/dagster/dagster/_core/definitions/job_definition.py

@@ -828,6 +828,10 @@ def as_ephemeral_job(
             asset_selection=frozenset(asset_selection) if asset_selection else None,
         )
 
+    @property
+    def is_asset_job(self) -> bool:
+        return bool(self.asset_layer and self.asset_layer.selected_asset_keys)
+
     def _get_partitions_def(
         self, selected_asset_keys: Optional[Iterable[AssetKey]]
     ) -> PartitionsDefinition:

python_modules/dagster/dagster/_core/snap/node.py

@@ -1,5 +1,5 @@
 from collections.abc import Mapping, Sequence, Set
-from functools import cached_property
+from functools import cached_property, partial
 from typing import Optional, Union
 
 import dagster._check as check
@@ -135,14 +135,21 @@ def build_input_def_snap(input_def: InputDefinition) -> InputDefSnap:
     )
 
 
-def build_output_def_snap(output_def: OutputDefinition) -> OutputDefSnap:
+def build_output_def_snap(output_def: OutputDefinition, job_def: JobDefinition) -> OutputDefSnap:
     check.inst_param(output_def, "output_def", OutputDefinition)
+    check.inst_param(job_def, "job_def", JobDefinition)
+
+    # Don't include verbose user-defined metadata(description/metadata fields) on the output
+    # definition snapshots for asset jobs, since they can be quite verbose/large and they
+    # map directly to information that is already available on the asset graph
+    include_verbose_metadata = not job_def.is_asset_job
+
     return OutputDefSnap(
         name=output_def.name,
         dagster_type_key=output_def.dagster_type.key,
-        description=output_def.description,
+        description=output_def.description if include_verbose_metadata else None,
         is_required=output_def.is_required,
-        metadata=output_def.metadata,
+        metadata=output_def.metadata if include_verbose_metadata else {},
         is_dynamic=output_def.is_dynamic,
     )
 
@@ -223,9 +230,9 @@ def build_node_defs_snapshot(job_def: JobDefinition) -> NodeDefsSnapshot:
     graph_def_snaps = []
     for node_def in job_def.all_node_defs:
         if isinstance(node_def, OpDefinition):
-            op_def_snaps.append(build_op_def_snap(node_def))
+            op_def_snaps.append(build_op_def_snap(node_def, job_def))
         elif isinstance(node_def, GraphDefinition):
-            graph_def_snaps.append(build_graph_def_snap(node_def))
+            graph_def_snaps.append(build_graph_def_snap(node_def, job_def))
         else:
             check.failed(f"Unexpected NodeDefinition type {node_def}")
 
@@ -250,12 +257,15 @@ def _by_name(
     return snap.name
 
 
-def build_graph_def_snap(graph_def: GraphDefinition) -> GraphDefSnap:
+def build_graph_def_snap(graph_def: GraphDefinition, job_def: JobDefinition) -> GraphDefSnap:
     check.inst_param(graph_def, "graph_def", GraphDefinition)
     return GraphDefSnap(
         name=graph_def.name,
         input_def_snaps=sorted(map(build_input_def_snap, graph_def.input_defs), key=_by_name),
-        output_def_snaps=sorted(map(build_output_def_snap, graph_def.output_defs), key=_by_name),
+        output_def_snaps=sorted(
+            map(partial(build_output_def_snap, job_def=job_def), graph_def.output_defs),
+            key=_by_name,
+        ),
         description=graph_def.description,
         tags=graph_def.tags,
         config_field_snap=(
@@ -278,12 +288,14 @@ def build_graph_def_snap(graph_def: GraphDefinition) -> GraphDefSnap:
     )
 
 
-def build_op_def_snap(op_def: OpDefinition) -> OpDefSnap:
+def build_op_def_snap(op_def: OpDefinition, job_def: JobDefinition) -> OpDefSnap:
     check.inst_param(op_def, "op_def", OpDefinition)
     return OpDefSnap(
         name=op_def.name,
         input_def_snaps=sorted(map(build_input_def_snap, op_def.input_defs), key=_by_name),
-        output_def_snaps=sorted(map(build_output_def_snap, op_def.output_defs), key=_by_name),
+        output_def_snaps=sorted(
+            map(partial(build_output_def_snap, job_def=job_def), op_def.output_defs), key=_by_name
+        ),
         description=op_def.description,
         tags=op_def.tags,
         required_resource_keys=sorted(op_def.required_resource_keys),

python_modules/dagster/dagster_tests/core_tests/snap_tests/test_node_def_snap.py

@@ -1,14 +1,18 @@
 import dagster as dg
 from dagster._core.snap import DependencyStructureIndex, GraphDefSnap, build_graph_def_snap
-from dagster._core.snap.node import OpDefSnap, build_op_def_snap
+from dagster._core.snap.node import OpDefSnap, build_node_defs_snapshot, build_op_def_snap
 
 
 def test_basic_op_definition():
     @dg.op
     def noop_op(_):
         pass
 
-    op_snap = build_op_def_snap(noop_op)
+    @dg.job
+    def my_job():
+        noop_op()
+
+    op_snap = build_op_def_snap(noop_op, my_job)
 
     assert op_snap
     assert dg.deserialize_value(dg.serialize_value(op_snap), OpDefSnap) == op_snap
@@ -35,7 +39,11 @@ def kitchen_sink_op(_, arg_two, arg_one):  # out of order to test positional_inp
         assert arg_two
         raise Exception("should not execute")
 
-    kitchen_sink_op_snap = build_op_def_snap(kitchen_sink_op)
+    @dg.job
+    def my_job():
+        kitchen_sink_op()
+
+    kitchen_sink_op_snap = build_op_def_snap(kitchen_sink_op, my_job)
 
     assert kitchen_sink_op_snap
     assert kitchen_sink_op_snap.name == "kitchen_sink_op"
@@ -86,7 +94,7 @@ def noop_op(_):
     def comp_graph():
         noop_op()
 
-    comp_solid_meta = build_graph_def_snap(comp_graph)
+    comp_solid_meta = build_graph_def_snap(comp_graph, comp_graph.to_job())
 
     assert isinstance(comp_solid_meta, GraphDefSnap)
     assert (
@@ -107,7 +115,7 @@ def take_one(_, one):
     def comp_graph():
         take_one(return_one())
 
-    comp_solid_meta = build_graph_def_snap(comp_graph)
+    comp_solid_meta = build_graph_def_snap(comp_graph, comp_graph.to_job())
 
     assert isinstance(comp_solid_meta, GraphDefSnap)
     assert (
@@ -134,7 +142,7 @@ def take_many(_, items):
     def comp_graph(this_number):
         take_many([return_one(), this_number, return_one.alias("return_one_also")()])
 
-    comp_solid_meta = build_graph_def_snap(comp_graph)
+    comp_solid_meta = build_graph_def_snap(comp_graph, comp_graph.to_job())
 
     assert isinstance(comp_solid_meta, GraphDefSnap)
     assert (
@@ -146,3 +154,84 @@ def comp_graph(this_number):
     assert index.get_invocation("take_many")
     assert index.get_upstream_outputs("take_many", "items")[0].node_name == "return_one"
     assert index.get_upstream_outputs("take_many", "items")[1].node_name == "return_one_also"
+
+
+def test_asset_job_omits_output_description_and_metadata():
+    @dg.asset(description="my description", metadata={"foo": "bar"})
+    def my_asset():
+        return 1
+
+    @dg.asset
+    def my_other_asset(my_asset):
+        return my_asset + 1
+
+    # Test with define_asset_job
+    asset_job = dg.define_asset_job("my_asset_job", selection=[my_asset, my_other_asset])
+    defs = dg.Definitions(assets=[my_asset, my_other_asset], jobs=[asset_job])
+    job_def = defs.get_job_def("my_asset_job")
+
+    assert job_def.is_asset_job
+
+    # Build snapshots and find the my_asset op snap
+    node_defs_snapshot = build_node_defs_snapshot(job_def)
+    my_asset_op_snap = next(s for s in node_defs_snapshot.op_def_snaps if s.name == "my_asset")
+    output_snap = my_asset_op_snap.get_output_snap("result")
+
+    assert output_snap.description is None
+    assert output_snap.metadata == {}
+
+    # Test with implicit global asset job from Definitions
+    defs_implicit = dg.Definitions(assets=[my_asset, my_other_asset])
+    implicit_job = defs_implicit.get_implicit_global_asset_job_def()
+
+    assert implicit_job.is_asset_job
+
+    node_defs_snapshot_implicit = build_node_defs_snapshot(implicit_job)
+    my_asset_op_snap_implicit = next(
+        s for s in node_defs_snapshot_implicit.op_def_snaps if s.name == "my_asset"
+    )
+    output_snap_implicit = my_asset_op_snap_implicit.get_output_snap("result")
+
+    assert output_snap_implicit.description is None
+    assert output_snap_implicit.metadata == {}
+
+    # Test with graph_asset - verifies build_graph_def_snap also strips metadata
+    @dg.op
+    def inner_op():
+        return 1
+
+    @dg.graph_asset(description="graph asset description", metadata={"graph_key": "graph_value"})
+    def my_graph_asset():
+        return inner_op()
+
+    defs_with_graph = dg.Definitions(assets=[my_graph_asset])
+    graph_asset_job = defs_with_graph.get_implicit_global_asset_job_def()
+
+    assert graph_asset_job.is_asset_job
+
+    node_defs_snapshot_graph = build_node_defs_snapshot(graph_asset_job)
+    # Graph assets create a GraphDefSnap, not an OpDefSnap
+    my_graph_asset_snap = next(
+        s for s in node_defs_snapshot_graph.graph_def_snaps if s.name == "my_graph_asset"
+    )
+    graph_output_snap = my_graph_asset_snap.get_output_snap("result")
+
+    assert graph_output_snap.description is None
+    assert graph_output_snap.metadata == {}
+
+    # Verify that non-asset jobs still include description and metadata
+    @dg.op(out=dg.Out(description="op description", metadata={"key": "value"}))
+    def my_op():
+        return 1
+
+    @dg.job
+    def my_regular_job():
+        my_op()
+
+    assert not my_regular_job.is_asset_job
+
+    my_op_snap = build_op_def_snap(my_op, my_regular_job)
+    regular_output_snap = my_op_snap.get_output_snap("result")
+
+    assert regular_output_snap.description == "op description"
+    assert regular_output_snap.metadata == {"key": dg.TextMetadataValue("value")}

python_modules/libraries/dagster-dbt/dagster_dbt_tests/core/test_asset_decorator.py

@@ -266,7 +266,7 @@ def test_snapshot_id(
     assert len(set(results)) == 1
 
     # this should only update if the dbt project or asset producing code changes
-    assert results[0] == "29a3a4ac386555a3e738867b8b25765ffb17a145"
+    assert results[0] == "4712da99dfd67a3964ed52a0b7852af831b55b20"
 
 
 @pytest.mark.parametrize("name", [None, "custom"])