[bug] Fix issue causing ReconstructableJobs containing certain types …

…of metadata to error (dagster-io#21819)

## Summary & Motivation

Resolves: dagster-io#21815

Note that while the test involved here is referencing
`CacheableAssetsDefinition`, this same error would occur with regular
assets. Any asset using a `MetadataValue` which contains within it a
mutable object (which would be table schema and table column lineage
which contain lists I believe) would encounter a similar error.

After switching the core `MetadataValue` class from a `NamedTuple` to a
pydantic model, the code we use to generate hashes for our serializable
objects stopped handling them. This PR fixes this.

## How I Tested These Changes

After updating the test to include metadata of this type, observed the
mentioned error. These changes resolved it.

python_modules/dagster/dagster/_utils/__init__.py

@@ -46,6 +46,7 @@
 
 import packaging.version
 from filelock import FileLock
+from pydantic import BaseModel
 from typing_extensions import Literal, TypeAlias, TypeGuard
 
 import dagster._check as check
@@ -287,6 +288,8 @@ def make_hashable(value: Any) -> Any:
  return tuple(sorted((key, make_hashable(value)) for key, value in value.items()))
  elif isinstance(value, (list, tuple, set)):
  return tuple([make_hashable(x) for x in value])
+ elif isinstance(value, BaseModel):
+ return make_hashable(value.dict())
  else:
  return value
 

python_modules/dagster/dagster_tests/core_tests/test_external_execution_plan.py

@@ -28,6 +28,8 @@
  CacheableAssetsDefinition,
 )
 from dagster._core.definitions.job_base import InMemoryJob
+from dagster._core.definitions.metadata.metadata_value import MetadataValue
+from dagster._core.definitions.metadata.table import TableColumn, TableSchema
 from dagster._core.definitions.reconstruct import (
  ReconstructableJob,
  ReconstructableRepository,
@@ -261,7 +263,7 @@ def test_using_file_system_for_subplan_invalid_step():
  )
 
 
-def test_using_repository_data():
+def test_using_repository_data() -> None:
  with instance_for_test() as instance:
  # first, we resolve the repository to generate our cached metadata
  repository_def = pending_repo.compute_repository_definition()
@@ -316,7 +318,16 @@ def test_using_repository_data():
 
 
 class MyCacheableAssetsDefinition(CacheableAssetsDefinition):
- _cacheable_data = AssetsDefinitionCacheableData(keys_by_output_name={"result": AssetKey("foo")})
+ _cacheable_data = AssetsDefinitionCacheableData(
+ keys_by_output_name={"result": AssetKey("foo")},
+ metadata_by_output_name={
+ "result": {
+ "some_val": MetadataValue.table_schema(
+ schema=TableSchema(columns=[TableColumn("some_col")])
+ )
+ }
+ },
+ )
 
  def compute_cacheable_data(self):
  # used for tracking how many times this function gets called over an execution