fix: get_latest_materialization_event returns stale pre-wipe materialization (#25511)

## Summary & Motivation

Fixes
[#15806](#15806):
`instance.get_latest_materialization_event` returned a stale, wiped
materialization after `dagster asset wipe`, instead of `None` as
documented. Also fixes the Dagster Plus analogue of the bug for
partition wipes (reported by a cloud customer).

Asset wipes are soft deletes: `wipe_asset` records a wipe timestamp on
the `asset_keys` row, and readers treat the asset as live when
`last_materialization_timestamp > wipe_timestamp`. The problem is that
`ASSET_MATERIALIZATION_PLANNED` and `ASSET_OBSERVATION` events also bump
`last_materialization_timestamp` (intentionally, so planned/observed
assets show in the catalog) without updating the cached
`last_materialization` record — which `wipe_asset` never cleared. So the
moment any new run targeting a wiped asset started (or an observation
was reported), the row passed the wipe filter and
`_get_latest_materialization_records` returned the stale pre-wipe
record. A post-wipe observation resurfaced the stale materialization
indefinitely. This also affected `get_asset_records`
(`AssetRecord.asset_entry.last_materialization_record`) and
`get_latest_materialization_code_versions`, which share the same path.

### OSS fixes (`SqlEventLogStorage`)

- **Write side:** `wipe_asset` now also nulls out
`last_materialization`, so newly wiped rows are self-consistent. (This
matches what the Plus event log storage has always done.)
- **Read side:** `_get_latest_materialization_records` only trusts the
cached record when its timestamp postdates the wipe; otherwise it falls
back to the backcompat event-log query, which now has
`_add_assets_wipe_filter_to_query` applied. This covers rows wiped by
older Dagster versions where the stale blob is already in the database,
and also fixes the same latent staleness for legacy old-format rows that
always took the backcompat path.
- **Legacy visibility path:** `_fetch_backcompat_materialization_times`
(used by the pre-`ASSET_KEY_INDEX_COLS`-migration read path to decide
whether a wiped asset is live) now only counts
materialization/observation/planned events, mirroring what bumps
`last_materialization_timestamp` on the migrated path. Previously it
took `max(timestamp)` over *all* events, so the `ASSET_WIPED` event
stored by `instance.wipe_assets` immediately after the wipe made the
wiped asset appear live in `all_asset_keys`. This was latent until the
write-side fix above stopped the stale blob from short-circuiting that
path — it broke `TestAssetWipe::test_asset_wipe` in the postgres graphql
CI jobs, whose fixture creates tables via alembic without marking the
secondary index (postgres gates `last_materialization_timestamp` writes
on that index, so those storages stay on the legacy path forever).

### Plus fix (`HostCloudEventLogStorage`)

Full-asset wipes already clear `last_materialization`, but
`wipe_asset_partitions` deleted the per-partition materialization rows
while leaving the cached `last_materialization` on the asset row
untouched. If the latest materialization belonged to a wiped partition,
`get_latest_materialization_events` / `get_asset_records` kept returning
it. The partition wipe now recomputes the cached record from the
remaining materializations when it points at a wiped partition (falling
back to `None` when none remain). The cached-value read locks the
`asset_keys` row (`FOR UPDATE`) through the recompute, so a
materialization committing mid-wipe either lands before the recompute
reads (and is included) or after the wipe commits (and overwrites with
the newer value) — the recomputed value can never clobber a newer
concurrent materialization. The lock is acquired in the same order as
the store path (`asset_keys` last), so it adds no deadlock risk. Note
the adjacent caches (`last_observation`,
`last_failed_to_materialize_event`, `last_skipped_materialize_event`)
have the same staleness after partition wipes and are left for
follow-up.

## Test Plan

- Extended the shared-suite `test_asset_wipe`: after a wipe, a planned
event and then an observation must not resurface the old materialization
from `get_latest_materialization_events` or `get_asset_records`, and a
re-materialization must surface the new event. Runs against sqlite,
consolidated sqlite, in-memory, legacy, postgres, mysql, and the Plus
storages (host cloud, archive-aware, ursula graphql).
- Extended the shared-suite `test_asset_partitioned_wiped_event` (Plus
storages only): wiping the latest-materialized partition falls back to
the latest remaining materialization, wiping a non-latest partition
leaves it untouched, and wiping the only remaining partition clears it.
Verified to fail without the Plus fix.
- New
`test_get_latest_materialization_ignores_stale_cached_row_after_wipe`
(sqlite) simulates a row wiped by an older version (stale cached record
restored via direct SQL after the wipe) to exercise the read-side guard.
Verified to fail without the read-side fix.
- New `test_all_asset_keys_excludes_wiped_asset_on_legacy_index_path`
(dagster-postgres) simulates the unmigrated-index condition from the
graphql CI fixture and reproduces the exact `assert AssetKey(...) not in
[AssetKey(...)]` failure from build 155619 without the legacy-path fix;
passes with it.
- Locally green: `dagster_tests/storage_tests/test_event_log.py` (519
passed), `dagster_postgres_tests/test_event_log.py` (142 passed), the
previously-failing `TestAssetWipe` postgres graphql variants,
cloud-backend `test_event_log_storage.py` +
`test_archive_aware_storage.py` via tox (553 passed), and the ursula
graphql storage wipe tests.

## Changelog

Fixed a bug where `DagsterInstance.get_latest_materialization_event`
(and `get_asset_records`) could return a stale pre-wipe materialization
for a wiped asset once a new run targeting the asset started or an
observation was reported. In Dagster+, fixed the analogous bug where
wiping partitions could leave the wiped partition's materialization as
the asset's latest materialization.

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Internal-RevId: d05abfe7d759123142e63f4d191113e3aab529f2

Author: 2 people

python_modules/dagster/dagster/_core/storage/event_log/sql_event_log.py

@@ -1164,39 +1164,54 @@ def _get_latest_materialization_records(
         # Given a list of raw asset rows, returns a mapping of asset key to latest asset materialization
         # event log entry. Fetches backcompat EventLogEntry records when the last_materialization
         # in the raw asset row is an AssetMaterialization.
-        to_backcompat_fetch = set()
+        to_backcompat_fetch: list[AssetKey] = []
+        asset_details_by_key: dict[AssetKey, AssetDetails | None] = {}
         results: dict[AssetKey, EventLogRecord | None] = {}
         for row in raw_asset_rows:
             asset_key = AssetKey.from_db_string(row["asset_key"])
             if not asset_key:
                 continue
+            asset_details = AssetDetails.from_db_string(row["asset_details"])
+            last_wipe_timestamp = asset_details.last_wipe_timestamp if asset_details else None
             event_or_materialization = (
                 deserialize_value(row["last_materialization"], NamedTuple)  # ty: ignore[no-matching-overload]
                 if row["last_materialization"]
                 else None
             )
-            if isinstance(event_or_materialization, EventLogRecord):
+            if isinstance(event_or_materialization, EventLogRecord) and (
+                last_wipe_timestamp is None
+                or event_or_materialization.event_log_entry.timestamp > last_wipe_timestamp
+            ):
                 results[asset_key] = event_or_materialization
             else:
-                to_backcompat_fetch.add(asset_key)
-
-        latest_event_subquery = db_subquery(
-            db_select(
-                [
-                    SqlEventLogStorageTable.c.asset_key,
-                    db.func.max(SqlEventLogStorageTable.c.id).label("id"),
-                ]
-            )
-            .where(
-                db.and_(
-                    SqlEventLogStorageTable.c.asset_key.in_(
-                        [asset_key.to_string() for asset_key in to_backcompat_fetch]
-                    ),
-                    SqlEventLogStorageTable.c.dagster_event_type
-                    == DagsterEventType.ASSET_MATERIALIZATION.value,
-                )
+                # No stored record, an old-format record, or a record predating the latest wipe
+                # (a planned or observation event after a wipe makes the asset row visible while
+                # the stale pre-wipe materialization is still stored on it). Fall back to querying
+                # the event log table, filtering out pre-wipe events.
+                to_backcompat_fetch.append(asset_key)
+                asset_details_by_key[asset_key] = asset_details
+
+        backcompat_subquery = db_select(
+            [
+                SqlEventLogStorageTable.c.asset_key,
+                db.func.max(SqlEventLogStorageTable.c.id).label("id"),
+            ]
+        ).where(
+            db.and_(
+                SqlEventLogStorageTable.c.asset_key.in_(
+                    [asset_key.to_string() for asset_key in to_backcompat_fetch]
+                ),
+                SqlEventLogStorageTable.c.dagster_event_type
+                == DagsterEventType.ASSET_MATERIALIZATION.value,
             )
-            .group_by(SqlEventLogStorageTable.c.asset_key),
+        )
+        backcompat_subquery = self._add_assets_wipe_filter_to_query(
+            backcompat_subquery,
+            [asset_details_by_key[asset_key] for asset_key in to_backcompat_fetch],
+            to_backcompat_fetch,
+        )
+        latest_event_subquery = db_subquery(
+            backcompat_subquery.group_by(SqlEventLogStorageTable.c.asset_key),
             "latest_event_subquery",
         )
         backcompat_query = db_select(
@@ -1527,7 +1542,9 @@ def _fetch_backcompat_materialization_times(
         self, asset_keys: Sequence[AssetKey]
     ) -> Mapping[AssetKey, datetime]:
         # fetches the latest materialization timestamp for the given asset_keys.  Uses the (slower)
-        # raw event log table.
+        # raw event log table.  Restricted to the event types that update
+        # last_materialization_timestamp in `_get_asset_entry_values`, so that events like
+        # ASSET_WIPED (stored immediately after a wipe) do not make a wiped asset appear live.
         backcompat_query = (
             db_select(
                 [
@@ -1536,8 +1553,17 @@ def _fetch_backcompat_materialization_times(
                 ]
             )
             .where(
-                SqlEventLogStorageTable.c.asset_key.in_(
-                    [asset_key.to_string() for asset_key in asset_keys]
+                db.and_(
+                    SqlEventLogStorageTable.c.asset_key.in_(
+                        [asset_key.to_string() for asset_key in asset_keys]
+                    ),
+                    SqlEventLogStorageTable.c.dagster_event_type.in_(
+                        [
+                            DagsterEventType.ASSET_MATERIALIZATION.value,
+                            DagsterEventType.ASSET_OBSERVATION.value,
+                            DagsterEventType.ASSET_MATERIALIZATION_PLANNED.value,
+                        ]
+                    ),
                 )
             )
             .group_by(SqlEventLogStorageTable.c.asset_key)
@@ -1758,6 +1784,7 @@ def _get_asset_key_values_on_wipe(self) -> Mapping[str, Any]:
         wipe_timestamp = get_current_timestamp()
         values: dict[str, Any] = {
             "asset_details": serialize_value(AssetDetails(last_wipe_timestamp=wipe_timestamp)),
+            "last_materialization": None,
             "last_run_id": None,
         }
         if self.has_asset_key_index_cols():

python_modules/dagster/dagster_tests/storage_tests/test_event_log.py

@@ -12,14 +12,22 @@
 import sqlalchemy as db
 import sqlalchemy.exc
 from dagster import DagsterInstance
-from dagster._core.events import EngineEventData, SerializableErrorInfo, StepRetryData
+from dagster._core.events import (
+    AssetMaterializationPlannedData,
+    DagsterEvent,
+    DagsterEventType,
+    EngineEventData,
+    SerializableErrorInfo,
+    StepRetryData,
+)
 from dagster._core.execution.stats import (
     StepEventStatus,
     build_run_stats_from_events,
     build_run_step_stats_from_events,
     build_run_step_stats_snapshot_from_events,
 )
 from dagster._core.storage.event_log import (
+    AssetKeyTable,
     ConsolidatedSqliteEventLogStorage,
     SqlEventLogStorageMetadata,
     SqlEventLogStorageTable,
@@ -31,13 +39,15 @@
 from dagster._core.storage.sqlalchemy_compat import db_select
 from dagster._core.storage.sqlite_storage import DagsterSqliteStorage
 from dagster._core.utils import make_new_run_id
+from dagster._serdes import serialize_value
 from dagster._utils.test import ConcurrencyEnabledSqliteTestEventLogStorage
 from sqlalchemy import __version__ as sqlalchemy_version
 from sqlalchemy.engine import Connection
 
 from dagster_tests.storage_tests.utils.event_log_storage import (
     TestEventLogStorage,
     _synthesize_events,
+    one_asset_op,
 )
 
 
@@ -218,6 +228,54 @@ def test_get_latest_tags_by_partition(self, storage, instance, dagster_event_typ
         pytest.skip("skip this since legacy storage is harder to mock.patch")
 
 
+def test_get_latest_materialization_ignores_stale_cached_row_after_wipe():
+    # Regression test for https://github.com/dagster-io/dagster/issues/15806. Rows wiped by older
+    # versions of wipe_asset retain the pre-wipe materialization in the `last_materialization`
+    # column. A planned event stored after the wipe bumps the row past the wipe filter; ensure
+    # the stale cached materialization is still not returned.
+    with tempfile.TemporaryDirectory() as tmpdir_path:
+        with dg.instance_for_test(temp_dir=tmpdir_path) as instance:
+            storage = instance.event_log_storage
+            assert isinstance(storage, SqliteEventLogStorage)
+            asset_key = dg.AssetKey("asset_1")
+
+            _synthesize_events(one_asset_op, run_id=make_new_run_id(), instance=instance)
+            record = storage.get_asset_records([asset_key])[
+                0
+            ].asset_entry.last_materialization_record
+            assert record is not None
+
+            storage.wipe_asset(asset_key)
+
+            # simulate a legacy wiped row by restoring the stale cached materialization
+            with storage.index_connection() as conn:
+                conn.execute(
+                    AssetKeyTable.update()
+                    .values(last_materialization=serialize_value(record))
+                    .where(AssetKeyTable.c.asset_key == asset_key.to_string())
+                )
+
+            storage.store_event(
+                dg.EventLogEntry(
+                    error_info=None,
+                    level="debug",
+                    user_message="",
+                    run_id=make_new_run_id(),
+                    timestamp=time.time(),
+                    dagster_event=DagsterEvent(
+                        DagsterEventType.ASSET_MATERIALIZATION_PLANNED.value,
+                        "nonce",
+                        event_specific_data=AssetMaterializationPlannedData(asset_key),
+                    ),
+                )
+            )
+
+            assert storage.get_latest_materialization_events([asset_key]).get(asset_key) is None
+            asset_records = storage.get_asset_records([asset_key])
+            assert len(asset_records) == 1
+            assert asset_records[0].asset_entry.last_materialization_record is None
+
+
 def _insert_slots(conn: Connection, concurrency_key: str, num: int, delete_num: int = 0):
     rows = [
         {

python_modules/dagster/dagster_tests/storage_tests/utils/event_log_storage.py

@@ -2923,6 +2923,61 @@ def test_asset_wipe(self, storage, instance, test_run_id):
         assert len(asset_keys) == 1
         assert storage.has_asset_key(dg.AssetKey("asset_1"))
 
+        # Regression test for https://github.com/dagster-io/dagster/issues/15806 - a planned or
+        # observation event stored after a wipe makes the asset row visible again, but must not
+        # resurface the stale pre-wipe materialization.
+        asset_key = dg.AssetKey("asset_1")
+        assert storage.get_latest_materialization_events([asset_key])[asset_key] is not None
+
+        storage.wipe_asset(asset_key)
+        assert storage.get_latest_materialization_events([asset_key]).get(asset_key) is None
+
+        planned_run_id = make_new_run_id()
+        observation_run_id = make_new_run_id()
+        with create_and_delete_test_runs(instance, [planned_run_id, observation_run_id]):
+            storage.store_event(
+                dg.EventLogEntry(
+                    error_info=None,
+                    level="debug",
+                    user_message="",
+                    run_id=planned_run_id,
+                    timestamp=time.time(),
+                    dagster_event=DagsterEvent(
+                        DagsterEventType.ASSET_MATERIALIZATION_PLANNED.value,
+                        "nonce",
+                        event_specific_data=AssetMaterializationPlannedData(asset_key),
+                    ),
+                )
+            )
+            assert storage.get_latest_materialization_events([asset_key]).get(asset_key) is None
+            asset_records = storage.get_asset_records([asset_key])
+            assert len(asset_records) == 1
+            assert asset_records[0].asset_entry.last_materialization_record is None
+
+            storage.store_event(
+                dg.EventLogEntry(
+                    error_info=None,
+                    level="debug",
+                    user_message="",
+                    run_id=observation_run_id,
+                    timestamp=time.time(),
+                    dagster_event=DagsterEvent(
+                        DagsterEventType.ASSET_OBSERVATION.value,
+                        "nonce",
+                        event_specific_data=AssetObservationData(
+                            dg.AssetObservation(asset_key=asset_key)
+                        ),
+                    ),
+                )
+            )
+            assert storage.get_latest_materialization_events([asset_key]).get(asset_key) is None
+
+        rematerialization_run_id = make_new_run_id()
+        _synthesize_events(one_asset_op, run_id=rematerialization_run_id, instance=instance)
+        latest = storage.get_latest_materialization_events([asset_key]).get(asset_key)
+        assert latest is not None
+        assert latest.run_id == rematerialization_run_id
+
     def test_asset_wiped_event(self, instance):
         @dg.asset
         def asset_to_wipe():
@@ -2969,6 +3024,28 @@ def asset_to_wipe():
         assert wipe_events[0].event_log_entry.dagster_event_type == DagsterEventType.ASSET_WIPED
         assert wipe_events[0].event_log_entry.dagster_event.asset_wiped_data.partition_keys == ["a"]
 
+        # the latest materialization pointed at the wiped partition and must not be returned
+        assert instance.get_latest_materialization_event(asset_to_wipe.key) is None
+
+        materialize([asset_to_wipe], instance=instance, partition_key="b")
+        materialize([asset_to_wipe], instance=instance, partition_key="c")
+
+        # wiping the latest-materialized partition falls back to the latest remaining one
+        instance.wipe_asset_partitions(asset_to_wipe.key, ["c"])
+        latest = instance.get_latest_materialization_event(asset_to_wipe.key)
+        assert latest is not None
+        assert latest.dagster_event.partition == "b"
+
+        # wiping a partition that does not hold the latest materialization leaves it untouched
+        instance.wipe_asset_partitions(asset_to_wipe.key, ["a"])
+        latest = instance.get_latest_materialization_event(asset_to_wipe.key)
+        assert latest is not None
+        assert latest.dagster_event.partition == "b"
+
+        # wiping the only remaining materialized partition clears it
+        instance.wipe_asset_partitions(asset_to_wipe.key, ["b"])
+        assert instance.get_latest_materialization_event(asset_to_wipe.key) is None
+
     def test_asset_secondary_index(self, storage, instance):
         _synthesize_events(one_asset_op, instance=instance)
 

python_modules/libraries/dagster-postgres/dagster_postgres_tests/test_event_log.py

@@ -4,7 +4,17 @@
 
 import objgraph
 import pytest
+from dagster import AssetKey, AssetMaterialization, EventLogEntry
+from dagster._core.events import (
+    AssetWipedData,
+    DagsterEvent,
+    DagsterEventType,
+    StepMaterializationData,
+)
+from dagster._core.instance import RUNLESS_JOB_NAME, RUNLESS_RUN_ID
 from dagster._core.storage.event_log.base import EventLogCursor
+from dagster._core.storage.event_log.migration import ASSET_KEY_INDEX_COLS
+from dagster._core.storage.event_log.schema import SecondaryIndexMigrationTable
 from dagster._core.test_utils import ensure_dagster_tests_import, instance_for_test
 from dagster._core.utils import make_new_run_id
 from dagster_postgres.event_log import PostgresEventLogStorage
@@ -140,3 +150,55 @@ def test_load_from_config(self, hostname):
                 from_explicit = explicit_instance._event_storage  # noqa: SLF001
 
                 assert from_url.postgres_url == from_explicit.postgres_url  # ty: ignore[unresolved-attribute]
+
+
+def test_all_asset_keys_excludes_wiped_asset_on_legacy_index_path(conn_string):
+    # Regression test for the legacy (pre-ASSET_KEY_INDEX_COLS-migration) read path, where wiped
+    # assets are filtered by comparing the wipe timestamp against the latest event log timestamp.
+    # The ASSET_WIPED event stored immediately after a wipe (and any other event type that does
+    # not update last_materialization_timestamp on the migrated path) must not make the wiped
+    # asset appear live.
+    asset_key = AssetKey(["asset_one"])
+    with _clean_storage(conn_string) as storage:
+        # simulate a storage that has not run the ASSET_KEY_INDEX_COLS data migration
+        with storage.index_connection() as conn:
+            conn.execute(SecondaryIndexMigrationTable.delete())
+        storage._secondary_index_cache.clear()  # noqa: SLF001
+        assert not storage.has_secondary_index(ASSET_KEY_INDEX_COLS)
+
+        storage.store_event(
+            EventLogEntry(
+                error_info=None,
+                level="debug",
+                user_message="",
+                run_id=make_new_run_id(),
+                timestamp=time.time(),
+                dagster_event=DagsterEvent(
+                    DagsterEventType.ASSET_MATERIALIZATION.value,
+                    "nonce",
+                    event_specific_data=StepMaterializationData(
+                        AssetMaterialization(asset_key=asset_key)
+                    ),
+                ),
+            )
+        )
+        assert asset_key in storage.all_asset_keys()
+
+        storage.wipe_asset(asset_key)
+        storage.store_event(
+            EventLogEntry(
+                error_info=None,
+                level="debug",
+                user_message="",
+                run_id=RUNLESS_RUN_ID,
+                timestamp=time.time(),
+                dagster_event=DagsterEvent(
+                    event_type_value=DagsterEventType.ASSET_WIPED.value,
+                    job_name=RUNLESS_JOB_NAME,
+                    event_specific_data=AssetWipedData(asset_key=asset_key, partition_keys=None),
+                ),
+            )
+        )
+
+        assert asset_key not in storage.all_asset_keys()
+        assert storage.get_latest_materialization_events([asset_key]).get(asset_key) is None