🧭 Concept Batch processing Pipelines & automation Updated

Incremental Processing: How a Batch Job Knows What Changed

Problem statement

Nine hundred files arrive every night. Four of them are different from yesterday.

02:00  job starts
02:00  discovered 900 files
08:40  job finished — 900 processed, 900 written

Six hours and forty minutes to redo work that was already correct. The outputs for the other 896 files are byte-identical to the ones they replaced, their modification times all jumped to 08:40, and anything downstream watching for changes now sees 900 changes instead of four.

The obvious fix — skip files whose output already exists — is wrong in a way that only shows up later: it never reprocesses a file whose content changed but whose output still exists. The job silently stops updating.

Getting this right means answering one question precisely: what does "changed" mean, and what evidence proves it?

Quick answer

Four ways to decide, in increasing order of reliability and cost:

Signal Detects Misses Cost
output exists never-processed items any content change free
mtime newer than output edits and re-deliveries touched-but-unchanged files (false positive) one stat()
content hash genuine content changes only nothing reads every byte
manifest changes to input or to the code/config nothing a state file 
import hashlib, json
from pathlib import Path

def content_hash(path, chunk=1 << 20):
    h = hashlib.blake2b(digest_size=16)
    with open(path, "rb") as f:
        while block := f.read(chunk):
            h.update(block)
    return h.hexdigest()

def needs_processing(src: Path, state: dict, recipe: str) -> bool:
    prev = state.get(str(src))
    if prev is None:
        return True                              # never seen
    if prev["recipe"] != recipe:
        return True                              # the code or config changed
    if prev["size"] != src.stat().st_size:
        return True                              # cheap pre-filter
    return prev["hash"] != content_hash(src)     # the authority
02:00  job starts
02:00  discovered 900 files, 4 changed  (896 unchanged, 0 new)
02:03  job finished — 4 processed, 896 skipped

The recipe field is the part people leave out and regret. If the transformation changes, every output is stale regardless of whether any input moved — and a job that only watches inputs will never notice.

The four signals

Grid of four change-detection signals against what each detects and misses.
Only the bottom row notices that the code changed.

Step-by-step solution

Decision diagram from how inputs change to the appropriate change-detection signal.
How the data arrives decides which signal is honest.

Signal 1: output exists — the one that quietly stops working 

def needs_processing(src, dst):
    return not output_for(src, dst).exists()

This is correct for exactly one situation: inputs are immutable. Archive tiles, dated deliveries in dated folders, anything with a content-addressed name. If 2024-03-01_parcels.gpkg never changes after it lands, its output never needs rebuilding.

It is wrong the moment a supplier overwrites a file in place. The output exists, so the file is skipped, and the job reports success while serving data from March forever. This failure is completely silent — the counts look normal, no error is raised, and the only symptom is that a number downstream stops moving.

If you use this signal, be explicit about the assumption:

# Inputs are immutable: the supplier writes dated files and never edits them.
# If that stops being true, this check must change to a content hash.
def needs_processing(src, dst):
    return not output_for(src, dst).exists()

Signal 2: mtime — cheap, and a little bit of a liar 

def needs_processing(src, dst):
    out = output_for(src, dst)
    return not out.exists() or src.stat().st_mtime > out.stat().st_mtime

Almost free, and catches genuine edits. Its failure modes are all in the direction of doing too much work rather than too little, which is the safe direction:

  • rsync and cp update mtime even when the content is identical, so a re-delivery of unchanged data reprocesses everything.
  • touch marks a file changed without changing a byte.
  • Clock skew between machines makes a file look older than an output produced from it.
  • Filesystem granularity is one second on some systems, so a file rewritten within the same second as its output looks unchanged — the one direction where mtime does too little.

Guard against the last one by comparing with a margin:

return src.stat().st_mtime > out.stat().st_mtime + 1

Signal 3: content hash — the authority 

def content_hash(path, chunk=1 << 20):
    h = hashlib.blake2b(digest_size=16)      # faster than sha256, plenty for this
    with open(path, "rb") as f:
        while block := f.read(chunk):
            h.update(block)
    return h.hexdigest()

A hash answers the real question: is this file's content the same as the content we processed? It has no false positives and no false negatives.

The cost is reading every byte — about 1–2 GB/s from local SSD, far slower over a network share. On 41 GB that is 20–40 seconds locally and several minutes over NFS, which is still nothing next to six hours of reprocessing.

Use size as a free pre-filter, because a different size guarantees different content:

if prev["size"] != src.stat().st_size:
    return True                     # certainly changed, no need to hash
return prev["hash"] != content_hash(src)

For very large files where even reading is too slow, hash a sample — first megabyte, last megabyte, size — and accept a small chance of missing a change:

def cheap_fingerprint(path, edge=1 << 20):
    size = path.stat().st_size
    h = hashlib.blake2b(digest_size=16)
    h.update(str(size).encode())
    with open(path, "rb") as f:
        h.update(f.read(edge))
        if size > 2 * edge:
            f.seek(-edge, 2)
            h.update(f.read(edge))
    return h.hexdigest()

Be honest in the name and the comment: this is a fingerprint, not a hash, and an edit in the middle of a large file will be missed.

Signal 4: the manifest — inputs and the recipe

The complete version records everything the output depends on:

def recipe_id(config: dict, version: str) -> str:
    """Anything that changes the output, other than the input file."""
    payload = json.dumps({"version": version, "config": config}, sort_keys=True)
    return hashlib.blake2b(payload.encode(), digest_size=8).hexdigest()
state = {
  "data/orkney.gpkg": {
      "hash": "9f2c…", "size": 8412193, "mtime": 1723190400.0,
      "recipe": "a1b2c3d4",                 # ← code + config at the time
      "output": "out/orkney.gpkg",
      "processed_at": "2026-08-15T02:03:11Z",
      "rows": 4102
  },
  ...
}

Now bumping VERSION or changing the target CRS in the config invalidates everything, exactly as it should. Without it, a change to the transformation produces a job that skips 900 files and reports success — the same silent failure as signal 1, arriving from the other direction.

What "changed" means for a database source

Files have hashes; tables do not, at least not cheaply. The equivalents:

-- an updated_at column, if the source maintains one honestly
SELECT * FROM parcels WHERE updated_at > :last_run;

-- a monotonically increasing id, for append-only tables
SELECT * FROM events WHERE id > :last_id;

-- xmin, the Postgres row version — works without cooperation from the schema
SELECT * FROM parcels WHERE xmin::text::bigint > :last_xmin;

The first is only as good as whoever maintains the column — a bulk update that forgets to touch it makes rows invisible to the job forever. The second is reliable and only works for inserts. The third needs care around transaction id wraparound but requires nothing of the schema.

Store the state where it survives 

STATE = Path("state/manifest.json")

def load_state():
    return json.loads(STATE.read_text()) if STATE.exists() else {}

def save_state(state):
    STATE.parent.mkdir(parents=True, exist_ok=True)
    tmp = STATE.with_suffix(".tmp")
    tmp.write_text(json.dumps(state, indent=2, sort_keys=True))
    tmp.replace(STATE)                  # atomic — a crash never leaves half a manifest

The write-to-temp-then-rename is not optional. A manifest truncated by a crash is worse than no manifest: the job forgets what it did and reprocesses everything, or worse, forgets only some entries.

Update the state after the output is safely written, never before:

for item in changed:
    result = apply_one(item)            # writes the output
    if result.status == "ok":
        state[str(item)] = fingerprint(item, recipe)     # only now
save_state(state)

Recording first and processing second means a crash leaves the manifest claiming work that never happened — and that file is then skipped forever.

Code examples

Example 1: the complete incremental discovery 

import hashlib, json
from dataclasses import dataclass
from pathlib import Path

@dataclass
class Change:
    path: Path
    reason: str          # "new" | "content" | "recipe" | "size"

def plan(src: Path, state: dict, recipe: str, pattern="*.gpkg") -> tuple[list, dict]:
    changed, unchanged = [], 0
    for path in sorted(src.rglob(pattern)):
        prev = state.get(str(path))
        if prev is None:
            changed.append(Change(path, "new"))
        elif prev.get("recipe") != recipe:
            changed.append(Change(path, "recipe"))
        elif prev.get("size") != path.stat().st_size:
            changed.append(Change(path, "size"))
        elif prev.get("hash") != content_hash(path):
            changed.append(Change(path, "content"))
        else:
            unchanged += 1
    return changed, {"unchanged": unchanged, "changed": len(changed)}
changed, summary = plan(Path("data"), load_state(), recipe_id(CONFIG, VERSION))
from collections import Counter
print(summary, Counter(c.reason for c in changed))
# {'unchanged': 896, 'changed': 4} Counter({'content': 3, 'new': 1})

Reporting the reason is what makes this debuggable. "900 changed, reason=recipe" says somebody bumped the version; "900 changed, reason=content" says the supplier re-exported everything; "900 changed, reason=new" says the manifest was lost.

Example 2: the safety valve every incremental job needs 

def guard_full_rebuild(changed, total, *, threshold=0.5, force=False):
    """A sudden full rebuild is usually a bug in change detection, not a real change."""
    if force or not total:
        return changed
    share = len(changed) / total
    if share > threshold:
        raise SystemExit(
            f"{len(changed):,}/{total:,} items ({share:.0%}) look changed. "
            f"That is unusual — check the manifest and the recipe id, "
            f"then re-run with --force if it is genuinely a full rebuild."
        )
    return changed

The failure this prevents is expensive and silent: a corrupted manifest, a changed hash function, or a recipe id that includes a timestamp makes every item look changed, and the job cheerfully spends six hours doing what it was built to avoid. Failing loudly costs a manual re-run; not failing costs the overnight window.

Example 3: making the recipe honest 

import inspect

def recipe_id(config: dict, *functions) -> str:
    """Hash the config AND the source of the transformation functions."""
    h = hashlib.blake2b(digest_size=8)
    h.update(json.dumps(config, sort_keys=True, default=str).encode())
    for fn in functions:
        h.update(inspect.getsource(fn).encode())
    return h.hexdigest()

RECIPE = recipe_id(CONFIG, transform_parcels, clean_attributes)

Hashing the function source means editing the transformation automatically invalidates every output, with no discipline required. It has limits — it does not see changes in functions those two call, or in library versions — so pin the library versions in the config and treat it as a strong default rather than a proof.

Explanation

Two panels showing how output-exists and hash-plus-recipe checks respond to an input edit and a code change.
Both silent failures produce the same symptom: a successful run that changed nothing.

Incremental processing is a cache, and it inherits every hard problem caches have. The output is a cached value; the input file and the transformation are its key; and the entire question is whether the key is complete.

An incomplete key produces stale results with no error. That is the failure mode running through this whole page, and it arrives in two ways. Watching only the output's existence misses input edits. Watching only the input misses code and config changes. Both produce a job that runs, reports success, and serves data from weeks ago — the worst kind of failure, because every signal says fine.

The mirror-image failure is over-invalidation: a key that includes something irrelevant, so everything looks changed and the job does full work every night while claiming to be incremental. The usual cause is a timestamp or an absolute path leaking into the recipe id. The guard in example 2 exists because this failure is otherwise invisible — the job still produces correct output, just slowly, and nobody investigates a job that works.

Between the two, stale is much worse than slow, which is why the recommended default is a content hash plus a recipe id: it never misses a change, and the cost is bounded by read bandwidth.

The last piece is ordering. State must be written after the output, and atomically. A manifest updated before processing records work that may never happen; a manifest written in place can be truncated by a crash. Temp-file-plus-rename makes the update atomic on the same filesystem, so the manifest is always either the old complete one or the new complete one — which is the same idempotency argument that applies to the outputs themselves.

Edge cases or notes

  • st_mtime granularity is 1 second on some filesystems and 1 nanosecond on others. Compare with a margin.
  • Network filesystems have unreliable mtimes and clock skew between client and server. Hash instead.
  • blake2b is faster than sha256 and entirely adequate here — this is change detection, not cryptography.
  • A hash of a GeoPackage changes when SQLite reorganises pages, even with identical data. Hashing the data rather than the file requires reading it as a layer, which is far more expensive.
  • Deleted inputs need handling. An entry in the manifest whose file is gone means the output should probably be removed too.
  • The manifest grows forever. Prune entries whose source no longer exists, or it slowly becomes the largest file in the project.
  • Do not put the manifest in the output directory if anything downstream syncs that directory — it will be treated as a deliverable.
  • Parallel workers must not write the manifest concurrently. Collect results and write once in the parent.

FAQ

Is checking whether the output exists good enough?

Only if inputs are genuinely immutable — dated files that are never overwritten. If a supplier can edit a file in place, this check silently stops updating and never reports a problem.

mtime or content hash?

mtime when inputs are local and re-deliveries are rare; a hash when correctness matters more than a few seconds of read time. mtime's errors are all in the direction of extra work, except when the filesystem's resolution is coarse.

Why include a recipe id?

Because the output depends on the code and config as well as the input. Without it, changing the target CRS produces a run that skips every file and reports success.

What if the manifest is lost?

Everything looks new and the job does a full rebuild — which is correct but expensive. The guard in example 2 turns that into a visible prompt rather than a silent six hours.

How do I do this for a database table?

Use an updated_at column if one is honestly maintained, a monotonic id for append-only data, or Postgres xmin if the schema offers nothing.

Should the manifest live in version control?

No — it is run state, not source. Keep it beside the outputs or in a state directory, and make sure it is backed up with them.

How do I test change detection?

Process, assert everything ran; re-run, assert nothing ran; touch a file's content, assert exactly one item ran; bump the recipe, assert everything ran.