FetchPolicy.can_skip

ingestify/application/dataset_store.py

@@ -199,10 +199,10 @@ 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(
+    def get_dataset_summary_map(
         self, provider: str, dataset_type: str
-    ) -> "DatasetLastModifiedAtMap":
-        return self.dataset_repository.get_dataset_last_modified_at_map(
+    ) -> "DatasetSummaryMap":
+        return self.dataset_repository.get_dataset_summary_map(
             bucket=self.bucket,
             provider=provider,
             dataset_type=dataset_type,

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

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
@@ -11,7 +12,23 @@
 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).
+DatasetSummaryMap = dict[str, DatasetSummary]
 
 
 class Dataset(BaseModel):

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, DatasetSummaryMap
 from .dataset_state import DatasetState
 from .selector import Selector
 
@@ -43,16 +43,16 @@ def get_dataset_collection(
     ) -> DatasetCollection:
         pass
 
-    def get_dataset_last_modified_at_map(
+    def get_dataset_summary_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."""
+    ) -> 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):

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:

ingestify/domain/models/ingestion/ingestion_job.py

@@ -4,7 +4,6 @@
 import logging
 import uuid
 from enum import Enum
-from functools import lru_cache
 from typing import Optional, Iterator, Union
 
 from pydantic import ValidationError
@@ -23,7 +22,7 @@
     FileResource,
     DatasetResource,
 )
-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, Operation
 from ingestify.exceptions import SaveError, IngestifyError, StopProcessing, FatalError
 from ingestify.utils import TaskExecutor, chunker
@@ -115,7 +114,6 @@ def load_file(
         )
 
 
-@lru_cache(maxsize=None)
 def _loader_accepts_dataset_resource(loader) -> bool:
     """Return True if loader accepts a `dataset_resource` keyword argument."""
     try:
@@ -292,7 +290,7 @@ def execute(
         self,
         store: DatasetStore,
         task_executor: TaskExecutor,
-        last_modified_at_map: Optional[DatasetLastModifiedAtMap] = None,
+        summary_map: Optional[DatasetSummaryMap] = None,
     ) -> Iterator[IngestionJobSummary]:
         # Single-run guard: one job identity must never run in two processes at once
         # (design: docs/design/single-run-lock.md). The lock is a session-scoped DB lock,
@@ -305,15 +303,15 @@ def execute(
             yield summary  # Loader persists every yielded summary
             return
         try:
-            yield from self._execute_locked(store, task_executor, last_modified_at_map)
+            yield from self._execute_locked(store, task_executor, summary_map)
         finally:
             run_lock.release()
 
     def _execute_locked(
         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)
@@ -399,7 +397,7 @@ def _execute_locked(
                 batches,
                 store,
                 task_executor,
-                last_modified_at_map,
+                summary_map,
                 ingestion_job_summary,
                 is_first_chunk,
             )
@@ -427,25 +425,29 @@ def _execute_locked(
                 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
 
@@ -576,7 +578,7 @@ def _execute_async(
         batches,
         store: DatasetStore,
         task_executor: TaskExecutor,
-        last_modified_at_map,
+        summary_map,
         ingestion_job_summary: IngestionJobSummary,
         is_first_chunk: bool,
     ) -> Iterator[IngestionJobSummary]:
@@ -600,19 +602,24 @@ def filtered_stream():
                     ingestion_job_summary.set_exception(e)
                     return
 
-                # Fast pre-check
-                if last_modified_at_map:
+                # Fast pre-check (bloom-filter style): policy.can_skip decides
+                # from a cheap summary whether an existing dataset is up-to-date,
+                # before the full get_dataset_collection load. One-sided (True only
+                # when certain); only existing datasets (summary present) are
+                # eligible — new ones fall through to the create path.
+                if summary_map:
                     pending = []
                     for dr in batch:
                         identifier = Identifier.create_from_selector(
                             self.selector, **dr.dataset_resource_id
                         )
-                        ts = last_modified_at_map.get(identifier.key)
-                        if ts is not None and dr.files:
-                            max_mod = max(f.last_modified for f in dr.files.values())
-                            if ts >= max_mod:
-                                ingestion_job_summary.increase_skipped_tasks(1)
-                                continue
+                        summary = summary_map.get(identifier.key)
+                        if (
+                            summary is not None
+                            and self.ingestion_plan.fetch_policy.can_skip(summary, dr)
+                        ):
+                            ingestion_job_summary.increase_skipped_tasks(1)
+                            continue
                         pending.append(dr)
                     batch = pending
 

ingestify/infra/store/dataset/sqlalchemy/repository.py

@@ -39,6 +39,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, TaskState
 from ingestify.exceptions import IngestifyError
@@ -620,30 +624,72 @@ def _debug_query(self, q: Query):
         )
         logger.debug(f"Running query: {text_}")
 
-    def get_dataset_last_modified_at_map(
+    def get_dataset_summary_map(
         self,
         bucket: str,
         provider: str,
         dataset_type: str,
-    ) -> dict:
+    ) -> 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(
-                    self.dataset_table.c.identifier,
-                    self.dataset_table.c.last_modified_at,
+                    ds.c.identifier,
+                    ds.c.last_modified_at,
+                    latest.c.revision_id,
+                    rev.c.created_at,
+                    rev.c.state,
                 )
-                .filter(self.dataset_table.c.bucket == bucket)
-                .filter(self.dataset_table.c.provider == provider)
-                .filter(self.dataset_table.c.dataset_type == dataset_type)
+                .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)
             )
-            return {
-                key_from_dict(
+
+            result: DatasetSummaryMap = {}
+            for row in query:
+                identifier = (
                     row.identifier
                     if isinstance(row.identifier, dict)
                     else json.loads(row.identifier)
-                ): row.last_modified_at
-                for row in query
-            }
+                )
+                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,