[RFC] feat(metadata): add table spec metadata

docs/content/integrations/dbt/reference.mdx

@@ -571,18 +571,18 @@ def my_dbt_assets(context: AssetExecutionContext, dbt: DbtCliResource):
 
 ---
 
-## Emit column schema as materialization metadata <Experimental />
+## Emit column-level metadata as materialization metadata <Experimental />
 
 <Note>
   <strong>
-    Emitting column schema as materialization metadata is currently an
+    Emitting column-level metadata as materialization metadata is currently an
     experimental feature.{" "}
   </strong>{" "}
-  To use this feature, you'll need to be on at least `dagster==1.6.6` and
-  `dagster-dbt==0.22.6`.
+  To use this feature, you'll need to be on at least `dagster>=1.6.12` and
+  `dagster-dbt>=0.22.12`.
 </Note>
 
-Dagster allows you to emit column schema [materialization metadata](/concepts/assets/software-defined-assets#recording-materialization-metadata), which includes the column names and data types of your materialized dbt models, seeds, and snapshots.
+Dagster allows you to emit column-level metadata, like column schema and column dependencies, as [materialization metadata](/concepts/assets/software-defined-assets#recording-materialization-metadata).
 
 With this metadata, you can view documentation in Dagster for all columns, not just columns described in your dbt project.
 
@@ -595,20 +595,36 @@ packages:
     revision: DAGSTER_VERSION # replace with the version of `dagster` you are using.
 ```
 
-Then, enable the `dagster.log_columns_in_relation()` macro as a [post-hook](https://docs.getdbt.com/reference/resource-configs/pre-hook-post-hook) for the dbt resources that should emit column schema metadata. For example, adding the following snippet in `dbt_project.yml` enables this macro for all dbt models, seeds, and snapshots:
+Then, enable the `dagster.log_column_level_metadata()` macro as a [post-hook](https://docs.getdbt.com/reference/resource-configs/pre-hook-post-hook) for the dbt resources that should emit column schema metadata. For example, adding the following snippet in `dbt_project.yml` enables this macro for all dbt models, seeds, and snapshots:
 
 ```yaml
 models:
   +post-hook:
-    - "{{ dagster.log_columns_in_relation() }}"
+    - "{{ dagster.log_column_level_metadata() }}"
 
 seeds:
   +post-hook:
-    - "{{ dagster.log_columns_in_relation() }}"
+    - "{{ dagster.log_column_level_metadata() }}"
 
 snapshots:
   +post-hook:
-    - "{{ dagster.log_columns_in_relation() }}"
+    - "{{ dagster.log_column_level_metadata() }}"
+```
+
+Column dependencies can be removed from materialization metadata by disabling the collection of parent relation metadata. This can be done by setting the `enable_parent_relation_metadata_collection` argument to `False` in the `dagster.log_column_level_metadata()` macro:
+
+```yaml
+models:
+  +post-hook:
+    - "{{ dagster.log_column_level_metadata(enable_parent_relation_metadata_collection=false) }}"
+
+seeds:
+  +post-hook:
+    - "{{ dagster.log_column_level_metadata(enable_parent_relation_metadata_collection=false) }}"
+
+snapshots:
+  +post-hook:
+    - "{{ dagster.log_column_level_metadata(enable_parent_relation_metadata_collection=false) }}"
 ```
 
 ---

docs/sphinx/sections/api/apidocs/ops.rst

@@ -101,6 +101,8 @@ All metadata types inherit from `MetadataValue`. The following types are defined
 
 .. autoclass:: TableSchemaMetadataValue
 
+.. autoclass:: TableSpecMetadataValue
+
 .. autoclass:: TextMetadataValue
 
 .. autoclass:: TimestampMetadataValue

pyright/alt-1/requirements-pinned.txt

@@ -17,6 +17,7 @@ asn1crypto==1.5.1
 astroid==3.1.0
 asttokens==2.4.1
 async-lru==2.0.4
+async-timeout==4.0.3
 attrs==23.2.0
 babel==2.14.0
 backoff==2.2.1
@@ -256,6 +257,8 @@ snowflake-sqlalchemy==1.5.1
 sortedcontainers==2.4.0
 soupsieve==2.5
 sqlalchemy==1.4.52
+sqlglot==22.3.1
+sqlglotrs==0.1.2
 sqlparse==0.4.4
 stack-data==0.6.3
 starlette==0.37.2

pyright/master/requirements-pinned.txt

@@ -35,6 +35,7 @@ asn1crypto==1.5.1
 -e examples/assets_pandas_pyspark
 asttokens==2.4.1
 async-lru==2.0.4
+async-timeout==4.0.3
 attrs==23.2.0
 autodocsumm==0.2.12
 autoflake==2.3.1
@@ -196,6 +197,7 @@ duckdb==0.10.1
 ecdsa==0.18.0
 email-validator==1.3.1
 entrypoints==0.4
+exceptiongroup==1.2.0
 execnet==2.0.2
 executing==2.0.1
 expandvars==0.12.0
@@ -509,6 +511,8 @@ sphinxcontrib-serializinghtml==1.1.10
 sqlalchemy==1.4.52
 sqlalchemy-jsonfield==1.0.2
 sqlalchemy-utils==0.41.1
+sqlglot==22.3.1
+sqlglotrs==0.1.2
 sqlparse==0.4.4
 sshpubkeys==3.3.1
 sshtunnel==0.4.0

python_modules/dagster-graphql/dagster_graphql/implementation/events.py

@@ -25,6 +25,7 @@
 from dagster._core.definitions.metadata import (
     DagsterRunMetadataValue,
     MetadataValue,
+    TableSpecMetadataValue,
 )
 from dagster._core.events import (
     DagsterEventType,
@@ -36,6 +37,9 @@
 from dagster._core.execution.plan.inputs import StepInputData
 from dagster._core.execution.plan.outputs import StepOutputData
 
+from dagster_graphql.schema.metadata import GrapheneTableSpecMetadataEntry
+from dagster_graphql.schema.table import GrapheneTableSpec
+
 MAX_INT = 2147483647
 MIN_INT = -2147483648
 
@@ -165,6 +169,11 @@ def iterate_metadata_entries(metadata: Mapping[str, MetadataValue]) -> Iterator[
                     columns=value.schema.columns,
                 ),
             )
+        elif isinstance(value, TableSpecMetadataValue):
+            yield GrapheneTableSpecMetadataEntry(
+                label=key,
+                spec=GrapheneTableSpec(spec=value.table_spec),
+            )
         elif isinstance(value, TimestampMetadataValue):
             yield GrapheneTimestampMetadataEntry(label=key, timestamp=value.value)
         else:

python_modules/dagster-graphql/dagster_graphql/schema/metadata.py

@@ -1,7 +1,7 @@
 import graphene
 
 from .asset_key import GrapheneAssetKey
-from .table import GrapheneTable, GrapheneTableSchema
+from .table import GrapheneTable, GrapheneTableSchema, GrapheneTableSpec
 
 
 class GrapheneMetadataItemDefinition(graphene.ObjectType):
@@ -52,6 +52,14 @@ class Meta:
         name = "TableSchemaMetadataEntry"
 
 
+class GrapheneTableSpecMetadataEntry(graphene.ObjectType):
+    spec = graphene.NonNull(GrapheneTableSpec)
+
+    class Meta:
+        interfaces = (GrapheneMetadataEntry,)
+        name = "TableSpecMetadataEntry"
+
+
 class GrapheneJsonMetadataEntry(graphene.ObjectType):
     jsonString = graphene.NonNull(graphene.String)
 

python_modules/dagster-graphql/dagster_graphql/schema/table.py

@@ -1,5 +1,6 @@
 import graphene
 
+from .asset_key import GrapheneAssetKey
 from .util import non_null_list
 
 
@@ -45,10 +46,36 @@ class Meta:
         name = "Table"
 
 
+class GrapheneTableColumnDep(graphene.ObjectType):
+    assetKey = graphene.NonNull(GrapheneAssetKey)
+    columnName = graphene.NonNull(graphene.String)
+
+    class Meta:
+        name = "TableColumnDep"
+
+
+class GrapheneTableColumnSpec(graphene.ObjectType):
+    columnName = graphene.NonNull(graphene.String)
+    tableColumnDeps = non_null_list(GrapheneTableColumnDep)
+
+    class Meta:
+        name = "TableColumnSpec"
+
+
+class GrapheneTableSpec(graphene.ObjectType):
+    columnSpecs = graphene.NonNull(GrapheneTableColumnSpec)
+
+    class Meta:
+        name = "TableSpec"
+
+
 types = [
     GrapheneTable,
     GrapheneTableSchema,
     GrapheneTableColumn,
     GrapheneTableColumnConstraints,
     GrapheneTableConstraints,
+    GrapheneTableColumnDep,
+    GrapheneTableColumnSpec,
+    GrapheneTableSpec,
 ]

python_modules/dagster/dagster/__init__.py

@@ -263,6 +263,7 @@
     PythonArtifactMetadataValue as PythonArtifactMetadataValue,
     TableMetadataValue as TableMetadataValue,
     TableSchemaMetadataValue as TableSchemaMetadataValue,
+    TableSpecMetadataValue as TableSpecMetadataValue,
     TextMetadataValue as TextMetadataValue,
     TimestampMetadataValue as TimestampMetadataValue,
     UrlMetadataValue as UrlMetadataValue,

python_modules/dagster/dagster/_core/definitions/__init__.py

@@ -67,6 +67,7 @@
     TableRecord as TableRecord,
     TableSchema as TableSchema,
     TableSchemaMetadataValue as TableSchemaMetadataValue,
+    TableSpecMetadataValue as TableSpecMetadataValue,
     TextMetadataValue as TextMetadataValue,
     UrlMetadataValue as UrlMetadataValue,
 )