From 78d3891a885f79c8b4d8070b76c938c9f37c6f78 Mon Sep 17 00:00:00 2001
From: Koen Vossen <info@koenvossen.nl>
Date: Wed, 12 Aug 2026 10:47:17 +0200
Subject: [PATCH 1/3] Add failing tests for FetchPolicy.can_skip hook (TDD red)

can_skip is a cheap, one-sided pre-check (bloom-filter style): a policy may
return True only when it is certain, from a lightweight per-dataset summary,
that an existing dataset is up-to-date. The engine then skips it without loading
the full Dataset graph and without reaching should_refetch. Returning False
means "unknown" -> fall through to the authoritative path. Base FetchPolicy
returns False, so the hook is additive and non-breaking.

Red test asserts that, on a second run, can_skip=True skips the fetch (loader
never runs) and short-circuits before should_refetch. A regression guard asserts
the base policy still reaches should_refetch and refetches (can_skip=False).

Implementation (summary type + get_dataset_summary_map + the hook + engine call)
follows.

Claude-Session: https://claude.ai/code/session_01B5EfLJqoafjW1FhvkxGSmg
---
 ingestify/tests/test_can_skip.py | 123 +++++++++++++++++++++++++++++++
 1 file changed, 123 insertions(+)
 create mode 100644 ingestify/tests/test_can_skip.py

diff --git a/ingestify/tests/test_can_skip.py b/ingestify/tests/test_can_skip.py
new file mode 100644
index 0000000..39305d0
--- /dev/null
+++ b/ingestify/tests/test_can_skip.py
@@ -0,0 +1,123 @@
+"""Tests for the ``can_skip`` fetch-policy hook.
+
+``can_skip`` is a cheap, one-sided pre-check (à la a bloom filter): it may return
+True *only* when the policy is certain, from a lightweight per-dataset summary,
+that the existing dataset is up-to-date. In that case the engine skips the
+dataset without loading the full ``Dataset`` graph and without reaching the
+authoritative ``should_refetch``. Returning False means "unknown" — fall through
+to ``get_dataset_collection`` + ``should_refetch``. The base ``FetchPolicy``
+returns False, so the hook is purely additive and non-breaking.
+
+Two observables:
+- the file loader (does the fetch task run at all), and
+- ``should_refetch`` (is the authoritative, full-dataset path reached).
+A ``can_skip`` that works avoids both on an up-to-date dataset. (Revision count
+can't tell a skip apart from a refetch-that-got-squashed-to-ignored, so it isn't
+used here.)
+"""
+from ingestify import Source, DatasetResource
+from ingestify.domain import DataSpecVersionCollection, DraftFile, Selector
+from ingestify.domain.models.fetch_policy import FetchPolicy
+from ingestify.domain.models.ingestion.ingestion_plan import IngestionPlan
+from ingestify.utils import utcnow
+
+
+class CountingSource(Source):
+    """Yields 5 datasets; counts how often their file loader is invoked."""
+
+    provider = "test_provider"
+
+    def __init__(self, name):
+        super().__init__(name)
+        self.load_calls = 0
+
+    def find_datasets(
+        self, dataset_type, data_spec_versions, dataset_collection_metadata, **kwargs
+    ):
+        def loader(file_resource, current_file, **kwargs):
+            self.load_calls += 1
+            return DraftFile.from_input("data", data_feed_key="f1")
+
+        for i in range(5):
+            r = DatasetResource(
+                dataset_resource_id={"item_id": i},
+                provider=self.provider,
+                dataset_type="test",
+                name=f"item-{i}",
+            )
+            r.add_file(
+                last_modified=utcnow(),
+                data_feed_key="f1",
+                data_spec_version="v1",
+                file_loader=loader,
+            )
+            yield r
+
+
+class CountingPolicy(FetchPolicy):
+    """Base-behaviour policy that records how often should_refetch is reached."""
+
+    def __init__(self):
+        super().__init__()
+        self.can_skip_calls = 0
+        self.should_refetch_calls = 0
+
+    def should_refetch(self, dataset, dataset_resource) -> bool:
+        self.should_refetch_calls += 1
+        return super().should_refetch(dataset, dataset_resource)
+
+
+class SkipExistingPolicy(CountingPolicy):
+    """Always certain an existing dataset is up-to-date."""
+
+    def can_skip(self, summary, dataset_resource) -> bool:
+        self.can_skip_calls += 1
+        return True
+
+
+def _add_plan(engine, source, fetch_policy):
+    dsv = DataSpecVersionCollection.from_dict({"default": {"v1"}})
+    engine.add_ingestion_plan(
+        IngestionPlan(
+            source=source,
+            fetch_policy=fetch_policy,
+            dataset_type="test",
+            selectors=[Selector.build({}, data_spec_versions=dsv)],
+            data_spec_versions=dsv,
+        )
+    )
+
+
+def test_can_skip_true_avoids_fetch_and_should_refetch(engine):
+    """can_skip=True short-circuits before the full-dataset path: on the second
+    run the file loader never runs and should_refetch is never reached."""
+    source = CountingSource("s")
+    policy = SkipExistingPolicy()
+    _add_plan(engine, source, policy)
+
+    engine.run()  # first run: nothing exists yet -> 5 creates
+    engine.run()  # second run: can_skip -> True -> skipped
+
+    assert source.load_calls == 5, "can_skip=True must skip the fetch on the second run"
+    assert (
+        policy.should_refetch_calls == 0
+    ), "can_skip=True must short-circuit before should_refetch"
+    assert (
+        policy.can_skip_calls == 5
+    ), "can_skip must be consulted once per existing dataset"
+
+
+def test_base_fetch_policy_still_fetches_can_skip_false(engine):
+    """Regression guard: base FetchPolicy.can_skip is False (default), so the
+    second run still reaches should_refetch and runs the fetch — unchanged."""
+    source = CountingSource("s")
+    policy = CountingPolicy()
+    _add_plan(engine, source, policy)
+
+    engine.run()
+    engine.run()
+
+    assert source.load_calls == 10, "base policy must still fetch (can_skip=False)"
+    assert (
+        policy.should_refetch_calls == 5
+    ), "base policy must still reach should_refetch on the second run"

From 2a7f1a5a9177752b12b7c0dbf49d6c2dd3edb5ba Mon Sep 17 00:00:00 2001
From: Koen Vossen <info@koenvossen.nl>
Date: Wed, 12 Aug 2026 11:18:22 +0200
Subject: [PATCH 2/3] Implement FetchPolicy.can_skip hook + dataset summary map
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Turns the engine's inline timestamp pre-check into a single, overridable policy
hook so a policy can skip an up-to-date dataset without loading the full
dataset+revision+file graph or reaching should_refetch.

- DatasetSummary (+ DatasetSummaryMap): lightweight per-dataset projection
  (last_modified, current revision created_at/state, has_revisions).
- DatasetRepository.get_dataset_summary_map + SqlAlchemy implementation: one
  grouped "latest revision per dataset" query (portable — no Postgres-only
  DISTINCT ON, no correlated subquery-in-join), restricted to the provider.
- FetchPolicy.can_skip(summary, dataset_resource) -> bool: base keeps the old
  timestamp behaviour (skip when stored is at least as new as every reported
  file), so this is additive and non-breaking. One-sided: True only when
  certain; False = fall through to the authoritative path.
- Engine: the inline last_modified pre-check is replaced by a single can_skip
  call (no duplicate pre-check logic); loader builds the summary map once per
  (provider, dataset_type).

Makes test_can_skip green; existing fast-skip/engine tests unchanged.

Claude-Session: https://claude.ai/code/session_01B5EfLJqoafjW1FhvkxGSmg
---
 ingestify/application/dataset_store.py        |  9 +++
 ingestify/application/loader.py               | 15 ++--
 ingestify/domain/models/dataset/dataset.py    | 22 +++++-
 .../models/dataset/dataset_repository.py      | 14 +++-
 ingestify/domain/models/fetch_policy.py       | 21 ++++++
 .../domain/models/ingestion/ingestion_job.py  | 32 +++++----
 .../store/dataset/sqlalchemy/repository.py    | 71 +++++++++++++++++++
 7 files changed, 160 insertions(+), 24 deletions(-)

diff --git a/ingestify/application/dataset_store.py b/ingestify/application/dataset_store.py
index 8aa0156..7c13bfe 100644
--- a/ingestify/application/dataset_store.py
+++ b/ingestify/application/dataset_store.py
@@ -203,6 +203,15 @@ def get_dataset_last_modified_at_map(
             dataset_type=dataset_type,
         )
 
+    def get_dataset_summary_map(
+        self, provider: str, dataset_type: str
+    ) -> "DatasetSummaryMap":
+        return self.dataset_repository.get_dataset_summary_map(
+            bucket=self.bucket,
+            provider=provider,
+            dataset_type=dataset_type,
+        )
+
     def get_dataset_collection(
         self,
         dataset_type: Optional[str] = None,
diff --git a/ingestify/application/loader.py b/ingestify/application/loader.py
index f302a5c..1648f71 100644
--- a/ingestify/application/loader.py
+++ b/ingestify/application/loader.py
@@ -241,9 +241,10 @@ def run(self, selectors, dry_run: bool = False):
         """Execute the collected selectors."""
         ingestion_job_prefix = str(uuid.uuid1())
 
-        # Build a cache of existing dataset timestamps per (provider, dataset_type).
-        # Used as a fast pre-check to skip datasets that are already up-to-date.
-        last_modified_at_cache: dict[tuple, "DatasetLastModifiedAtMap"] = {}
+        # Build a cache of lightweight dataset summaries per (provider,
+        # dataset_type). Fed to FetchPolicy.can_skip as a fast pre-check to skip
+        # datasets that are already up-to-date without loading the full graph.
+        summary_cache: dict[tuple, "DatasetSummaryMap"] = {}
 
         for ingestion_job_idx, (ingestion_plan, selector) in enumerate(selectors):
             logger.info(
@@ -263,10 +264,8 @@ def run(self, selectors, dry_run: bool = False):
                 ingestion_plan.source.provider,
                 ingestion_plan.dataset_type,
             )
-            if cache_key not in last_modified_at_cache:
-                last_modified_at_cache[
-                    cache_key
-                ] = self.store.get_dataset_last_modified_at_map(
+            if cache_key not in summary_cache:
+                summary_cache[cache_key] = self.store.get_dataset_summary_map(
                     provider=cache_key[0],
                     dataset_type=cache_key[1],
                 )
@@ -278,7 +277,7 @@ def run(self, selectors, dry_run: bool = False):
                 for ingestion_job_summary in ingestion_job.execute(
                     self.store,
                     task_executor=task_executor,
-                    last_modified_at_map=last_modified_at_cache[cache_key],
+                    summary_map=summary_cache[cache_key],
                 ):
                     # TODO: handle task_summaries
                     #       Summarize to a IngestionJobSummary, and save to a database. This Summary can later be used in a
diff --git a/ingestify/domain/models/dataset/dataset.py b/ingestify/domain/models/dataset/dataset.py
index bf30304..358031c 100644
--- a/ingestify/domain/models/dataset/dataset.py
+++ b/ingestify/domain/models/dataset/dataset.py
@@ -1,3 +1,4 @@
+from dataclasses import dataclass
 from datetime import datetime
 from enum import Enum
 from typing import List, Optional, Dict
@@ -7,13 +8,32 @@
 from .dataset_state import DatasetState
 from .file import DraftFile
 from .identifier import Identifier
-from .revision import Revision, RevisionSource, SourceType
+from .revision import Revision, RevisionSource, SourceType, RevisionState
 from ..base import BaseModel
 
 
 DatasetLastModifiedAtMap = dict[str, datetime]
 
 
+@dataclass(frozen=True)
+class DatasetSummary:
+    """Lightweight projection of a Dataset for FetchPolicy.can_skip.
+
+    Lets a policy decide an existing dataset is up-to-date from a few cheap
+    columns, so the engine can skip it without loading the full
+    dataset+revision+file graph. Fields reflect the latest revision (highest
+    revision_id)."""
+
+    last_modified: Optional[datetime]
+    current_created_at: Optional[datetime]
+    current_state: Optional[RevisionState]
+    has_revisions: bool
+
+
+# Keyed by Identifier.key (the JSON identifier), like DatasetLastModifiedAtMap.
+DatasetSummaryMap = dict[str, DatasetSummary]
+
+
 class Dataset(BaseModel):
     bucket: str  # This must be set by the DatasetRepository
     dataset_id: str
diff --git a/ingestify/domain/models/dataset/dataset_repository.py b/ingestify/domain/models/dataset/dataset_repository.py
index eb39cdd..2f29892 100644
--- a/ingestify/domain/models/dataset/dataset_repository.py
+++ b/ingestify/domain/models/dataset/dataset_repository.py
@@ -3,7 +3,7 @@
 from typing import Optional, List, Union
 
 from .collection import DatasetCollection
-from .dataset import Dataset, DatasetLastModifiedAtMap
+from .dataset import Dataset, DatasetLastModifiedAtMap, DatasetSummaryMap
 from .dataset_state import DatasetState
 from .selector import Selector
 
@@ -36,6 +36,18 @@ def get_dataset_last_modified_at_map(
         dataset+revision+file graph."""
         return {}
 
+    def get_dataset_summary_map(
+        self,
+        bucket: str,
+        provider: str,
+        dataset_type: str,
+    ) -> DatasetSummaryMap:
+        """Return {identifier_json: DatasetSummary} for all datasets matching the
+        given provider and dataset_type. Feeds FetchPolicy.can_skip as a cheap
+        pre-check, so an up-to-date dataset is skipped without loading the full
+        dataset+revision+file graph. Each summary reflects the latest revision."""
+        return {}
+
     def invalidate_revision(self, dataset: Dataset):
         """Mark the current revision as VALIDATION_FAILED and reset
         last_modified_at on the dataset."""
diff --git a/ingestify/domain/models/fetch_policy.py b/ingestify/domain/models/fetch_policy.py
index ead637c..0995228 100644
--- a/ingestify/domain/models/fetch_policy.py
+++ b/ingestify/domain/models/fetch_policy.py
@@ -1,6 +1,7 @@
 from datetime import timedelta
 
 from ingestify.domain import Dataset, Identifier, DatasetResource
+from ingestify.domain.models.dataset.dataset import DatasetSummary
 from ingestify.domain.models.dataset.revision import RevisionState
 from ingestify.utils import utcnow
 
@@ -15,6 +16,26 @@ def should_fetch(self, dataset_resource: DatasetResource) -> bool:
         # this is called when dataset does not exist yet
         return True
 
+    def can_skip(
+        self, summary: DatasetSummary, dataset_resource: DatasetResource
+    ) -> bool:
+        """Cheap, one-sided pre-check (bloom-filter style) for an *existing*
+        dataset. Return True only when certain, from the lightweight ``summary``,
+        that the dataset is up-to-date: the engine then skips it without loading
+        the full Dataset or reaching ``should_refetch``. Returning False means
+        "unknown" — fall through to the authoritative path.
+
+        Base policy: skip when the stored dataset is at least as new as every file
+        the source reports. (This is the timestamp pre-check the engine used to do
+        inline; it now lives here as the single source of truth.)
+        """
+        if summary.last_modified is None or not dataset_resource.files:
+            return False
+        max_file_modified = max(
+            f.last_modified for f in dataset_resource.files.values()
+        )
+        return summary.last_modified >= max_file_modified
+
     def should_refetch(
         self, dataset: Dataset, dataset_resource: DatasetResource
     ) -> bool:
diff --git a/ingestify/domain/models/ingestion/ingestion_job.py b/ingestify/domain/models/ingestion/ingestion_job.py
index 7704eaa..df2c4bc 100644
--- a/ingestify/domain/models/ingestion/ingestion_job.py
+++ b/ingestify/domain/models/ingestion/ingestion_job.py
@@ -24,7 +24,7 @@
     DatasetResource,
 )
 from ingestify.domain.models.resources.batch_loader import BatchLoader
-from ingestify.domain.models.dataset.dataset import DatasetLastModifiedAtMap
+from ingestify.domain.models.dataset.dataset import DatasetSummaryMap
 from ingestify.domain.models.task.task_summary import TaskSummary
 from ingestify.exceptions import SaveError, IngestifyError, StopProcessing
 from ingestify.utils import TaskExecutor, chunker
@@ -350,7 +350,7 @@ def execute(
         self,
         store: DatasetStore,
         task_executor: TaskExecutor,
-        last_modified_at_map: Optional[DatasetLastModifiedAtMap] = None,
+        summary_map: Optional[DatasetSummaryMap] = None,
     ) -> Iterator[IngestionJobSummary]:
         is_first_chunk = True
         ingestion_job_summary = IngestionJobSummary.new(ingestion_job=self)
@@ -433,25 +433,29 @@ def execute(
                 yield ingestion_job_summary
                 return
 
-            # Fast pre-check: skip datasets that are definitely up-to-date
-            # based on the cached timestamps. Only resources that might need
-            # work proceed to the full get_dataset_collection check.
+            # Fast pre-check (bloom-filter style): let the fetch policy skip
+            # datasets it is certain are up-to-date from a cheap summary, before
+            # loading the full dataset+revision+file graph. can_skip is one-sided
+            # (True only when certain); "unknown" resources fall through to the
+            # authoritative get_dataset_collection + should_refetch path. Only
+            # existing datasets (summary present) are eligible — new ones go to
+            # the create path below.
             skipped_tasks = 0
-            if last_modified_at_map:
+            if summary_map:
                 pending_batch = []
                 for dataset_resource in batch:
                     identifier = Identifier.create_from_selector(
                         self.selector, **dataset_resource.dataset_resource_id
                     )
-                    ts = last_modified_at_map.get(identifier.key)
-                    if ts is not None:
-                        # Dataset exists — check if all files are up-to-date
-                        max_file_modified = max(
-                            f.last_modified for f in dataset_resource.files.values()
+                    summary = summary_map.get(identifier.key)
+                    if (
+                        summary is not None
+                        and self.ingestion_plan.fetch_policy.can_skip(
+                            summary, dataset_resource
                         )
-                        if ts >= max_file_modified:
-                            skipped_tasks += 1
-                            continue
+                    ):
+                        skipped_tasks += 1
+                        continue
                     pending_batch.append(dataset_resource)
                 batch = pending_batch
 
diff --git a/ingestify/infra/store/dataset/sqlalchemy/repository.py b/ingestify/infra/store/dataset/sqlalchemy/repository.py
index e58ddb0..acc82c9 100644
--- a/ingestify/infra/store/dataset/sqlalchemy/repository.py
+++ b/ingestify/infra/store/dataset/sqlalchemy/repository.py
@@ -37,6 +37,10 @@
 from ingestify.domain.models.dataset.collection_metadata import (
     DatasetCollectionMetadata,
 )
+from ingestify.domain.models.dataset.dataset import (
+    DatasetSummary,
+    DatasetSummaryMap,
+)
 from ingestify.domain.models.ingestion.ingestion_job_summary import IngestionJobSummary
 from ingestify.domain.models.task.task_summary import TaskSummary
 from ingestify.exceptions import IngestifyError
@@ -543,6 +547,73 @@ def get_dataset_last_modified_at_map(
                 for row in query
             }
 
+    def get_dataset_summary_map(
+        self,
+        bucket: str,
+        provider: str,
+        dataset_type: str,
+    ) -> DatasetSummaryMap:
+        with self.session:
+            ds = self.dataset_table
+            rev = self.revision_table
+
+            # The current revision is the highest revision_id per dataset.
+            # Compute it as a grouped subquery (portable — no Postgres-only
+            # DISTINCT ON, no correlated subquery-in-join) restricted to this
+            # provider/dataset_type, then LEFT JOIN back to fetch its
+            # created_at/state. LEFT JOIN so datasets without any revision still
+            # show up (has_revisions=False).
+            latest = (
+                self.session.query(
+                    rev.c.dataset_id.label("dataset_id"),
+                    func.max(rev.c.revision_id).label("revision_id"),
+                )
+                .select_from(rev.join(ds, ds.c.dataset_id == rev.c.dataset_id))
+                .filter(ds.c.bucket == bucket)
+                .filter(ds.c.provider == provider)
+                .filter(ds.c.dataset_type == dataset_type)
+                .group_by(rev.c.dataset_id)
+                .subquery()
+            )
+            query = (
+                self.session.query(
+                    ds.c.identifier,
+                    ds.c.last_modified_at,
+                    latest.c.revision_id,
+                    rev.c.created_at,
+                    rev.c.state,
+                )
+                .select_from(
+                    ds.outerjoin(
+                        latest, latest.c.dataset_id == ds.c.dataset_id
+                    ).outerjoin(
+                        rev,
+                        and_(
+                            rev.c.dataset_id == latest.c.dataset_id,
+                            rev.c.revision_id == latest.c.revision_id,
+                        ),
+                    )
+                )
+                .filter(ds.c.bucket == bucket)
+                .filter(ds.c.provider == provider)
+                .filter(ds.c.dataset_type == dataset_type)
+            )
+
+            result: DatasetSummaryMap = {}
+            for row in query:
+                identifier = (
+                    row.identifier
+                    if isinstance(row.identifier, dict)
+                    else json.loads(row.identifier)
+                )
+                result[key_from_dict(identifier)] = DatasetSummary(
+                    last_modified=row.last_modified_at,
+                    current_created_at=row.created_at,
+                    current_state=row.state,
+                    has_revisions=row.revision_id is not None,
+                )
+            return result
+
     def get_dataset_collection(
         self,
         bucket: str,

From 134e98a6a267c2eb230e5b2e22dd00d40dbc9f99 Mon Sep 17 00:00:00 2001
From: Koen Vossen <info@koenvossen.nl>
Date: Wed, 12 Aug 2026 11:42:06 +0200
Subject: [PATCH 3/3] Remove get_dataset_last_modified_at_map, superseded by
 summary map

get_dataset_summary_map carries last_modified too, so the old map and its
DatasetLastModifiedAtMap type alias are redundant. Drop the repository interface
method, the SqlAlchemy implementation, the DatasetStore delegation, and the type
alias; repoint the fast-skip test to get_dataset_summary_map.

Claude-Session: https://claude.ai/code/session_01B5EfLJqoafjW1FhvkxGSmg
---
 ingestify/application/dataset_store.py        |  9 -------
 ingestify/domain/models/dataset/dataset.py    |  5 +---
 .../models/dataset/dataset_repository.py      | 14 +----------
 .../store/dataset/sqlalchemy/repository.py    | 25 -------------------
 ingestify/tests/test_fast_skip.py             | 10 ++++----
 5 files changed, 7 insertions(+), 56 deletions(-)

diff --git a/ingestify/application/dataset_store.py b/ingestify/application/dataset_store.py
index b8b2e3f..b840b7d 100644
--- a/ingestify/application/dataset_store.py
+++ b/ingestify/application/dataset_store.py
@@ -199,15 +199,6 @@ def acquire_run_lock(self, job_key: str):
         Returns a held RunLock, or None if another process already holds it."""
         return self.dataset_repository.acquire_run_lock(job_key)
 
-    def get_dataset_last_modified_at_map(
-        self, provider: str, dataset_type: str
-    ) -> "DatasetLastModifiedAtMap":
-        return self.dataset_repository.get_dataset_last_modified_at_map(
-            bucket=self.bucket,
-            provider=provider,
-            dataset_type=dataset_type,
-        )
-
     def get_dataset_summary_map(
         self, provider: str, dataset_type: str
     ) -> "DatasetSummaryMap":
diff --git a/ingestify/domain/models/dataset/dataset.py b/ingestify/domain/models/dataset/dataset.py
index 6da18de..044eed3 100644
--- a/ingestify/domain/models/dataset/dataset.py
+++ b/ingestify/domain/models/dataset/dataset.py
@@ -12,9 +12,6 @@
 from ..base import BaseModel
 
 
-DatasetLastModifiedAtMap = dict[str, datetime]
-
-
 @dataclass(frozen=True)
 class DatasetSummary:
     """Lightweight projection of a Dataset for FetchPolicy.can_skip.
@@ -30,7 +27,7 @@ class DatasetSummary:
     has_revisions: bool
 
 
-# Keyed by Identifier.key (the JSON identifier), like DatasetLastModifiedAtMap.
+# Keyed by Identifier.key (the JSON identifier).
 DatasetSummaryMap = dict[str, DatasetSummary]
 
 
diff --git a/ingestify/domain/models/dataset/dataset_repository.py b/ingestify/domain/models/dataset/dataset_repository.py
index d1b06ed..9294896 100644
--- a/ingestify/domain/models/dataset/dataset_repository.py
+++ b/ingestify/domain/models/dataset/dataset_repository.py
@@ -3,7 +3,7 @@
 from typing import Optional, List, Union
 
 from .collection import DatasetCollection
-from .dataset import Dataset, DatasetLastModifiedAtMap, DatasetSummaryMap
+from .dataset import Dataset, DatasetSummaryMap
 from .dataset_state import DatasetState
 from .selector import Selector
 
@@ -43,18 +43,6 @@ def get_dataset_collection(
     ) -> DatasetCollection:
         pass
 
-    def get_dataset_last_modified_at_map(
-        self,
-        bucket: str,
-        provider: str,
-        dataset_type: str,
-    ) -> DatasetLastModifiedAtMap:
-        """Return {identifier_json: last_modified_at} for all datasets matching
-        the given provider and dataset_type. Used as a fast pre-check to skip
-        datasets that are already up-to-date without loading the full
-        dataset+revision+file graph."""
-        return {}
-
     def get_dataset_summary_map(
         self,
         bucket: str,
diff --git a/ingestify/infra/store/dataset/sqlalchemy/repository.py b/ingestify/infra/store/dataset/sqlalchemy/repository.py
index c561a65..1c10c61 100644
--- a/ingestify/infra/store/dataset/sqlalchemy/repository.py
+++ b/ingestify/infra/store/dataset/sqlalchemy/repository.py
@@ -624,31 +624,6 @@ def _debug_query(self, q: Query):
         )
         logger.debug(f"Running query: {text_}")
 
-    def get_dataset_last_modified_at_map(
-        self,
-        bucket: str,
-        provider: str,
-        dataset_type: str,
-    ) -> dict:
-        with self.session:
-            query = (
-                self.session.query(
-                    self.dataset_table.c.identifier,
-                    self.dataset_table.c.last_modified_at,
-                )
-                .filter(self.dataset_table.c.bucket == bucket)
-                .filter(self.dataset_table.c.provider == provider)
-                .filter(self.dataset_table.c.dataset_type == dataset_type)
-            )
-            return {
-                key_from_dict(
-                    row.identifier
-                    if isinstance(row.identifier, dict)
-                    else json.loads(row.identifier)
-                ): row.last_modified_at
-                for row in query
-            }
-
     def get_dataset_summary_map(
         self,
         bucket: str,
diff --git a/ingestify/tests/test_fast_skip.py b/ingestify/tests/test_fast_skip.py
index fa4ea8a..2504b77 100644
--- a/ingestify/tests/test_fast_skip.py
+++ b/ingestify/tests/test_fast_skip.py
@@ -49,18 +49,18 @@ def _setup(engine):
     )
 
 
-def test_timestamps_cache_matches_identifiers(engine):
-    """Keys from get_dataset_last_modified_at_map match Identifier.key."""
+def test_summary_map_matches_identifiers(engine):
+    """Keys from get_dataset_summary_map match Identifier.key."""
     _setup(engine)
     engine.run()
 
-    timestamps = engine.store.get_dataset_last_modified_at_map(
+    summaries = engine.store.get_dataset_summary_map(
         provider="test_provider", dataset_type="test"
     )
     datasets = engine.store.get_dataset_collection(
         provider="test_provider", dataset_type="test"
     )
 
-    assert len(timestamps) == len(datasets) == 5
+    assert len(summaries) == len(datasets) == 5
     for dataset in datasets:
-        assert dataset.identifier.key in timestamps
+        assert dataset.identifier.key in summaries