Rename relation identifier to table name (dagster-io#25400)

## Summary & Motivation

Rename relation identifier to table name to increase clarity. [Context
here](https://dagsterlabs.slack.com/archives/C06AWMPT3DF/p1729269974738339).

Does a direct rename to switch references from `relation_identifier` to
`table_name`, since integration packages should have the same version as
the dagster package.

In the UI, to handle backcompat, display metadata with either key.

## How I Tested These Changes
BK

## Changelog
Renames the `dagster/relation_identifier` metadata key to
`dagster/table_name`.

docs/content/concepts/metadata-tags/asset-metadata.mdx

@@ -473,7 +473,7 @@ The `dagster` prefix indicates that the Dagster package takes responsibility for
     </tr>
     <tr>
       <td>
-        <code>dagster/relation_identifier</code>
+        <code>dagster/table_name</code>
       </td>
       <td>
         <ul

docs/docs-beta/docs/guides/metadata.md

@@ -85,7 +85,7 @@ The following metadata keys are given special treatment in the Dagster UI.
 | `dagster/column_lineage`      | **Type:** [`TableColumnLineage`](/todo) <br/><br/> For an asset that's a table, the lineage of column inputs to column outputs for the table. Refer to the [Table and column metadata](#table-schema) section for details.                                                                                                                         |
 | `dagster/row_count`           | **Type:** `int` <br/><br/> For an asset that's a table, the number of rows in the table. Refer to the Table metadata documentation for details.                                                                                                                                                                                                                 |
 | `dagster/partition_row_count` | **Type:** `int` <br/><br/> For a partition of an asset that's a table, the number of rows in the partition.                                                                                                                                                                                                                                                     |
-| `dagster/relation_identifier` | **Type:** `str` <br/><br/> A unique identifier for the table/view, typically fully qualified. For example, my_database.my_schema.my_table                                                                                                                                                                                                                       |
+| `dagster/table_name`          | **Type:** `str` <br/><br/> A unique identifier for the table/view, typically fully qualified. For example, my_database.my_schema.my_table                                                                                                                                                                                                                       |
 | `dagster/code_references`     | **Type:** [`CodeReferencesMetadataValue`](/todo) <br/><br/> A list of code references for the asset, such as file locations or references to GitHub URLs. Refer to the [Linking assets with their source code](#source-code) section for details. Should only be provided in definition-level metadata, not materialization metadata. |
 
 ## Table and column metadata \{#table-column}

js_modules/dagster-ui/packages/ui-core/src/assets/AssetNodeOverview.tsx

@@ -67,7 +67,7 @@ import {
   TableSchema,
   TableSchemaAssetContext,
   isCanonicalCodeSourceEntry,
-  isCanonicalRelationIdentifierEntry,
+  isCanonicalTableNameEntry,
   isCanonicalUriEntry,
 } from '../metadata/TableSchema';
 import {RepositoryLink} from '../nav/RepositoryLink';
@@ -294,9 +294,7 @@ export const AssetNodeOverview = ({
   const nonSystemTags = filteredTags?.filter((tag) => !isSystemTag(tag));
   const systemTags = filteredTags?.filter(isSystemTag);
 
-  const relationIdentifierMetadata = assetNode?.metadataEntries?.find(
-    isCanonicalRelationIdentifierEntry,
-  );
+  const tableNameMetadata = assetNode?.metadataEntries?.find(isCanonicalTableNameEntry);
   const uriMetadata = assetNode?.metadataEntries?.find(isCanonicalUriEntry);
 
   const codeSource = assetNode?.metadataEntries?.find((m) => isCanonicalCodeSourceEntry(m)) as
@@ -370,12 +368,12 @@ export const AssetNodeOverview = ({
           ))}
       </AttributeAndValue>
       <AttributeAndValue label="Storage">
-        {(relationIdentifierMetadata || uriMetadata || storageKindTag) && (
+        {(tableNameMetadata || uriMetadata || storageKindTag) && (
           <Box flex={{direction: 'column', gap: 4}} style={{minWidth: 0}}>
-            {relationIdentifierMetadata && (
+            {tableNameMetadata && (
               <Box flex={{direction: 'row', gap: 4, alignItems: 'center'}}>
-                <MiddleTruncate text={relationIdentifierMetadata.text} />
-                <CopyButton value={relationIdentifierMetadata.text} />
+                <MiddleTruncate text={tableNameMetadata.text} />
+                <CopyButton value={tableNameMetadata.text} />
               </Box>
             )}
             {uriMetadata && (

js_modules/dagster-ui/packages/ui-core/src/metadata/TableSchema.tsx

@@ -52,9 +52,8 @@ export const isCanonicalCodeSourceEntry = (
 ): m is CodeReferencesMetadataEntry =>
   m && m.__typename === 'CodeReferencesMetadataEntry' && m.label === 'dagster/code_references';
 
-export const isCanonicalRelationIdentifierEntry = (
-  m: MetadataEntryLabelOnly,
-): m is TextMetadataEntry => m && m.label === 'dagster/relation_identifier';
+export const isCanonicalTableNameEntry = (m: MetadataEntryLabelOnly): m is TextMetadataEntry =>
+  m && (m.label === 'dagster/relation_identifier' || m.label === 'dagster/table_name');
 
 export const isCanonicalUriEntry = (
   m: MetadataEntryLabelOnly,

python_modules/dagster/dagster/_core/definitions/metadata/metadata_set.py

@@ -165,15 +165,15 @@ class TableMetadataSet(NamespacedMetadataSet):
             outputs for the table.
         row_count (Optional[int]): The number of rows in the table.
         partition_row_count (Optional[int]): The number of rows in the materialized or observed partition.
-        relation_identifier (Optional[str]): A unique identifier for the table/view, typically fully qualified.
+        table_name (Optional[str]): A unique identifier for the table/view, typically fully qualified.
             For example, `my_database.my_schema.my_table`.
     """
 
     column_schema: Optional[TableSchema] = None
     column_lineage: Optional[TableColumnLineage] = None
     row_count: Optional[int] = None
     partition_row_count: Optional[int] = None
-    relation_identifier: Optional[str] = None
+    table_name: Optional[str] = None
 
     @classmethod
     def namespace(cls) -> str:

python_modules/dagster/dagster/_core/storage/db_io_manager.py

@@ -78,8 +78,8 @@ def delete_table_slice(
     def get_select_statement(table_slice: TableSlice) -> str: ...
 
     @staticmethod
-    def get_relation_identifier(table_slice: TableSlice) -> Optional[str]:
-        """Returns a string which is set as the dagster/relation_identifier metadata value for an
+    def get_table_name(table_slice: TableSlice) -> Optional[str]:
+        """Returns a string which is set as the dagster/table_name metadata value for an
         emitted asset. This value should be the fully qualified name of the table, including the
         schema and database, if applicable.
         """
@@ -171,11 +171,7 @@ def handle_output(self, context: OutputContext, obj: object) -> None:
         # don't fail if it errors because the user has already attached it.
         try:
             context.add_output_metadata(
-                dict(
-                    TableMetadataSet(
-                        relation_identifier=self._db_client.get_relation_identifier(table_slice)
-                    )
-                )
+                dict(TableMetadataSet(table_name=self._db_client.get_table_name(table_slice)))
             )
         except DagsterInvalidMetadata:
             pass

python_modules/dagster/dagster_tests/definitions_tests/metadata_tests/test_table_metadata_set.py

@@ -31,11 +31,11 @@ def test_table_metadata_set() -> None:
     assert TableMetadataSet.extract(dict(TableMetadataSet())) == TableMetadataSet()
 
 
-def test_relation_identifier() -> None:
-    table_metadata = TableMetadataSet(relation_identifier="my_database.my_schema.my_table")
+def test_table_name() -> None:
+    table_metadata = TableMetadataSet(table_name="my_database.my_schema.my_table")
 
     dict_table_metadata = dict(table_metadata)
-    assert dict_table_metadata == {"dagster/relation_identifier": "my_database.my_schema.my_table"}
+    assert dict_table_metadata == {"dagster/table_name": "my_database.my_schema.my_table"}
     AssetMaterialization(asset_key="a", metadata=dict_table_metadata)
 
 

python_modules/dagster/dagster_tests/storage_tests/test_db_io_manager.py

@@ -25,8 +25,8 @@
 }
 
 
-def mock_relation_identifier(*args, **kwargs) -> str:
-    return "relation_identifier"
+def mock_table_name(*args, **kwargs) -> str:
+    return "table_name"
 
 
 class IntHandler(DbTypeHandler[int]):
@@ -81,7 +81,7 @@ def test_asset_out():
         spec=DbClient,
         get_select_statement=MagicMock(return_value=""),
         connect=connect_mock,
-        get_relation_identifier=mock_relation_identifier,
+        get_table_name=mock_table_name,
     )
     manager = build_db_io_manager(type_handlers=[handler], db_client=db_client)
     asset_key = AssetKey(["schema1", "table1"])
@@ -117,7 +117,7 @@ def test_asset_out_columns():
         spec=DbClient,
         get_select_statement=MagicMock(return_value=""),
         connect=connect_mock,
-        get_relation_identifier=mock_relation_identifier,
+        get_table_name=mock_table_name,
     )
     manager = build_db_io_manager(type_handlers=[handler], db_client=db_client)
     asset_key = AssetKey(["schema1", "table1"])
@@ -159,7 +159,7 @@ def test_asset_out_partitioned():
         spec=DbClient,
         get_select_statement=MagicMock(return_value=""),
         connect=connect_mock,
-        get_relation_identifier=mock_relation_identifier,
+        get_table_name=mock_table_name,
     )
     manager = build_db_io_manager(type_handlers=[handler], db_client=db_client)
     asset_key = AssetKey(["schema1", "table1"])
@@ -218,7 +218,7 @@ def test_asset_out_static_partitioned():
         spec=DbClient,
         get_select_statement=MagicMock(return_value=""),
         connect=connect_mock,
-        get_relation_identifier=mock_relation_identifier,
+        get_table_name=mock_table_name,
     )
     manager = build_db_io_manager(type_handlers=[handler], db_client=db_client)
     asset_key = AssetKey(["schema1", "table1"])
@@ -270,7 +270,7 @@ def test_asset_out_multiple_static_partitions():
         spec=DbClient,
         get_select_statement=MagicMock(return_value=""),
         connect=connect_mock,
-        get_relation_identifier=mock_relation_identifier,
+        get_table_name=mock_table_name,
     )
     manager = build_db_io_manager(type_handlers=[handler], db_client=db_client)
     asset_key = AssetKey(["schema1", "table1"])
@@ -323,7 +323,7 @@ def test_different_output_and_input_types():
         spec=DbClient,
         get_select_statement=MagicMock(return_value=""),
         connect=connect_mock,
-        get_relation_identifier=mock_relation_identifier,
+        get_table_name=mock_table_name,
     )
     manager = build_db_io_manager(type_handlers=[int_handler, str_handler], db_client=db_client)
     asset_key = AssetKey(["schema1", "table1"])
@@ -361,7 +361,7 @@ def test_non_asset_out():
         spec=DbClient,
         get_select_statement=MagicMock(return_value=""),
         connect=connect_mock,
-        get_relation_identifier=mock_relation_identifier,
+        get_table_name=mock_table_name,
     )
     manager = build_db_io_manager(type_handlers=[handler], db_client=db_client)
     output_context = build_output_context(
@@ -399,7 +399,7 @@ def test_asset_schema_defaults():
     db_client = MagicMock(
         spec=DbClient,
         get_select_statement=MagicMock(return_value=""),
-        get_relation_identifier=mock_relation_identifier,
+        get_table_name=mock_table_name,
     )
     manager = build_db_io_manager(type_handlers=[handler], db_client=db_client)
 
@@ -561,7 +561,7 @@ def test_default_load_type():
     db_client = MagicMock(
         spec=DbClient,
         get_select_statement=MagicMock(return_value=""),
-        get_relation_identifier=mock_relation_identifier,
+        get_table_name=mock_table_name,
     )
     manager = DbIOManager(
         type_handlers=[handler],

python_modules/libraries/dagster-airbyte/dagster_airbyte/asset_defs.py

@@ -116,11 +116,11 @@ def _build_airbyte_asset_defn_metadata(
     for table in destination_tables:
         internal_deps[table] = set(upstream_assets or [])
 
-    relation_identifiers: Dict[str, str] = {}
+    table_names: Dict[str, str] = {}
     for table in destination_tables:
         if destination_database and destination_schema and table:
-            # Use the destination raw table name to create the relation identifier
-            relation_identifiers[table] = ".".join(
+            # Use the destination raw table name to create the table name
+            table_names[table] = ".".join(
                 [
                     destination_database,
                     destination_schema,
@@ -129,7 +129,7 @@ def _build_airbyte_asset_defn_metadata(
             )
             if normalization_tables and normalization_raw_table_names_by_table:
                 for normalization_table in normalization_tables.get(table, set()):
-                    relation_identifiers[normalization_table] = ".".join(
+                    table_names[normalization_table] = ".".join(
                         [
                             destination_database,
                             destination_schema,
@@ -156,7 +156,7 @@ def _build_airbyte_asset_defn_metadata(
                 table: {
                     **TableMetadataSet(
                         column_schema=schema_by_table_name.get(table),
-                        relation_identifier=relation_identifiers.get(table),
+                        table_name=table_names.get(table),
                     ),
                 }
                 for table in tables
@@ -302,15 +302,13 @@ def build_airbyte_assets(
         chain([destination_tables], normalization_tables.values() if normalization_tables else [])
     )
 
-    relation_identifiers: Dict[str, str] = {}
+    table_names: Dict[str, str] = {}
     for table in destination_tables:
         if destination_database and destination_schema and table:
-            relation_identifiers[table] = ".".join(
-                [destination_database, destination_schema, table]
-            )
+            table_names[table] = ".".join([destination_database, destination_schema, table])
             if normalization_tables:
                 for normalization_table in normalization_tables.get(table, set()):
-                    relation_identifiers[normalization_table] = ".".join(
+                    table_names[normalization_table] = ".".join(
                         [
                             destination_database,
                             destination_schema,
@@ -328,7 +326,7 @@ def build_airbyte_assets(
                 {
                     **TableMetadataSet(
                         column_schema=schema_by_table_name.get(table),
-                        relation_identifier=relation_identifiers.get(table),
+                        table_name=table_names.get(table),
                     ),
                 }
             ),

python_modules/libraries/dagster-airbyte/dagster_airbyte_tests/test_asset_defs.py

@@ -310,7 +310,7 @@ def downstream_of_ab():
     assert downstream_of_ab.op.ins["some_prefix_bar"].dagster_type.is_nothing
 
 
-def test_built_airbyte_asset_relation_identifier():
+def test_built_airbyte_asset_table_name():
     destination_tables = ["foo", "bar"]
 
     ab_assets = build_airbyte_assets(
@@ -322,7 +322,7 @@ def test_built_airbyte_asset_relation_identifier():
     # Check relation identifier metadata is added correctly to asset def
     assets_def = ab_assets[0]
     for metadata in assets_def.metadata_by_key.values():
-        assert metadata.get("dagster/relation_identifier") is None
+        assert metadata.get("dagster/table_name") is None
 
     ab_assets = build_airbyte_assets(
         "12345",
@@ -331,15 +331,15 @@ def test_built_airbyte_asset_relation_identifier():
         destination_schema="test_schema",
     )
 
-    relation_identifiers = {"test_database.test_schema.foo", "test_database.test_schema.bar"}
+    table_names = {"test_database.test_schema.foo", "test_database.test_schema.bar"}
 
     # Check relation identifier metadata is added correctly to asset def
     assets_def = ab_assets[0]
     for key, metadata in assets_def.metadata_by_key.items():
         # Extract the table name from the asset key
         table_name = key.path[-1]
-        assert metadata["dagster/relation_identifier"] in relation_identifiers
-        assert table_name in metadata["dagster/relation_identifier"]
+        assert metadata["dagster/table_name"] in table_names
+        assert table_name in metadata["dagster/table_name"]
 
     ab_assets = build_airbyte_assets(
         "12345",
@@ -349,12 +349,12 @@ def test_built_airbyte_asset_relation_identifier():
         normalization_tables={"foo": {"baz"}},
     )
 
-    relation_identifiers.add("test_database.test_schema.foo.baz")
+    table_names.add("test_database.test_schema.foo.baz")
 
     # Check relation identifier metadata is added correctly to asset def
     assets_def = ab_assets[0]
     for key, metadata in assets_def.metadata_by_key.items():
         # Extract the table name from the asset key
         table_name = key.path[-1]
-        assert metadata["dagster/relation_identifier"] in relation_identifiers
-        assert table_name in metadata["dagster/relation_identifier"]
+        assert metadata["dagster/table_name"] in table_names
+        assert table_name in metadata["dagster/table_name"]

python_modules/libraries/dagster-airbyte/dagster_airbyte_tests/test_load_from_instance.py

@@ -234,7 +234,7 @@ def downstream_asset(dagster_tags):
             for key, metadata in assets_def.metadata_by_key.items()
         )
 
-    relation_identifiers = {
+    table_names = {
         "test_database.test_schema.releases",
         "test_database.test_schema.tags",
         "test_database.test_schema.teams",
@@ -262,10 +262,10 @@ def downstream_asset(dagster_tags):
                 .replace("_test", "")
                 .split("_")[-1]
             )
-            assert metadata["dagster/relation_identifier"] in relation_identifiers
-            assert table_name in metadata["dagster/relation_identifier"]
+            assert metadata["dagster/table_name"] in table_names
+            assert table_name in metadata["dagster/table_name"]
         else:
-            assert not metadata.get("dagster/relation_identifier")
+            assert not metadata.get("dagster/table_name")
 
     assert assets_def.keys == {AssetKey(t) for t in tables}
     assert all(

python_modules/libraries/dagster-airbyte/dagster_airbyte_tests/test_load_from_project.py

@@ -106,7 +106,7 @@ def test_load_from_project(
     # Check metadata is added correctly to asset def
     assets_def = ab_assets[0]
 
-    relation_identifiers = {
+    table_names = {
         "AIRBYTE.BEN_DEMO.releases",
         "AIRBYTE.BEN_DEMO.tags",
         "AIRBYTE.BEN_DEMO.teams",
@@ -133,8 +133,8 @@ def test_load_from_project(
             .replace("_test", "")
             .split("_")[-1]
         )
-        assert metadata["dagster/relation_identifier"] in relation_identifiers
-        assert table_name in metadata["dagster/relation_identifier"]
+        assert metadata["dagster/table_name"] in table_names
+        assert table_name in metadata["dagster/table_name"]
 
     assert assets_def.keys == {AssetKey(t) for t in tables}
     assert all(

python_modules/libraries/dagster-dbt/dagster_dbt/asset_utils.py

@@ -469,7 +469,7 @@ def default_metadata_from_dbt_resource_props(
         **DbtMetadataSet(materialization_type=materialization_type),
         **TableMetadataSet(
             column_schema=column_schema,
-            relation_identifier=relation_name,
+            table_name=relation_name,
         ),
     }