diff --git a/ingestify/application/dataset_store.py b/ingestify/application/dataset_store.py index 451564c..b840b7d 100644 --- a/ingestify/application/dataset_store.py +++ b/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, 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 e832fe9..044eed3 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 @@ -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): diff --git a/ingestify/domain/models/dataset/dataset_repository.py b/ingestify/domain/models/dataset/dataset_repository.py index 27e8f2a..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 +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): 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 f67bce1..84a9eae 100644 --- a/ingestify/domain/models/ingestion/ingestion_job.py +++ b/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,7 +303,7 @@ 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() @@ -313,7 +311,7 @@ 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 diff --git a/ingestify/infra/store/dataset/sqlalchemy/repository.py b/ingestify/infra/store/dataset/sqlalchemy/repository.py index efa8bf6..1c10c61 100644 --- a/ingestify/infra/store/dataset/sqlalchemy/repository.py +++ b/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, 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" 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