Idempotent loads on S3 + Athena: skip unchanged partitions with content signatures

Your upstream source sends a cumulative, month-to-date file: every
delivery carries the whole month so far. Sometimes it arrives once a
month, sometimes every day. Your pipeline lands it in Amazon S3 as
Parquet, partitioned by day, and then builds stage and analytics layers
with Amazon Athena.

Two problems show up quickly:

1. Reruns duplicate data. Run the same day twice and Athena returns the rows twice.
2. Every delivery reprocesses the whole month. Thirty days, one Athena query per day per layer, even though most days did not change at all.

This post shows a small pattern that fixes both: an idempotent write per
partition, plus a content signature per partition that lets each
layer skip the days whose input did not change.

1. Why reruns duplicate data

INSERT INTO in Athena never touches the files that are already there.
The Athena documentation
says it plainly: Athena "does not rewrite or modify existing files.
Instead, it writes the rows as one or more new files."

So the second run of the same day adds a second file next to the first
one, and the table now reads both.

The fix is to make each partition a unit you can replace:

1. Delete every object under the partition prefix.
2. Run INSERT INTO ... WHERE <partition filter>.
3. Register the partition with ALTER TABLE ... ADD IF NOT EXISTS PARTITION.

Each step is safe to repeat. Run it once or ten times and the partition
ends up with the same rows.

def _delete_partition_prefix(self, partition: dict[str, str]) -> None:
    """Delete every object of the partition so INSERT INTO cannot duplicate."""
    prefix = f"{self._settings.data_prefix}/{self._partition_path(partition)}/"
    paginator = self._s3.get_paginator("list_objects_v2")
    pages = paginator.paginate(Bucket=self._settings.data_bucket, Prefix=prefix)
    for page in pages:
        objects = [{"Key": obj["Key"]} for obj in page.get("Contents", [])]
        if objects:
            self._s3.delete_objects(
                Bucket=self._settings.data_bucket,
                Delete={"Objects": objects, "Quiet": True},
            )

That solves correctness. It does not solve cost: we still rebuild all
thirty days on every delivery.

2. A content signature per partition

A signature is a hash of the business content of one partition. If
the hash is the same as last time, the partition did not change and
there is nothing to do.

Two details matter more than the hash function itself:

• Hash only the columns that mean "the data changed". Many exports include an updated_at-style column that moves even when nothing meaningful changed. If you include it, almost every partition looks "changed" on every run and the whole pattern becomes useless. Look at real consecutive deliveries before you pick the columns.
• Make it order-insensitive. Sort by a stable key before hashing. The same rows in a different order must give the same signature.

def content_signature(
    df: pd.DataFrame,
    columns: Sequence[str],
    sort_by: Sequence[str],
) -> str:
    """Return an order-insensitive SHA-256 hash of the given columns."""
    canonical = (
        df.loc[:, list(columns)]
        .sort_values(list(sort_by))
        .reset_index(drop=True)
        .to_csv(index=False)
    )
    return hashlib.sha256(canonical.encode(_ENCODING)).hexdigest()

The landing (raw) job computes this for each day it writes and stores it.

3. Chain the signatures through the layers

The stage layer does not need to read the raw data to know whether it
changed. It only needs the raw signature. But stage output also depends
on its own SQL: if you fix a bug in the transformation, every day
must be rebuilt even though the input is the same.

So each downstream layer stores:

signature(stage) = sha256( signature(raw) | fingerprint(stage SQL) )

and the analytics layer does the same one level up.

For the SQL fingerprint, the S3 ETag of the SQL file is enough: it
changes whenever the object content changes.

def chained_signature(upstream_signature: str, transform_fingerprint: str) -> str:
    """Combine the upstream signature with the transformation fingerprint."""
    payload = f"{upstream_signature}{_SEPARATOR}{transform_fingerprint}"
    return hashlib.sha256(payload.encode(_ENCODING)).hexdigest()


def object_fingerprint(s3_client: Any, bucket: str, key: str) -> str:
    """Return the ETag of an S3 object, without quotes."""
    return s3_client.head_object(Bucket=bucket, Key=key)["ETag"].strip('"')

⚠️ Use the ETag only as "did it change?". It is not always an MD5
of the content: multipart uploads and SSE-KMS objects get a different
kind of ETag (S3 docs).
Never compare it with an MD5 you computed locally.

Nice side effects of chaining:

• Nobody keeps a separate list of "what should exist upstream". Each layer only reads the signature of the layer right above it.
• Change the SQL of one layer and that layer, plus everything below it, rebuilds on the next run. The layers above it keep skipping.

4. Where to store the signatures

Store each signature as a tiny JSON sidecar, one per layer and partition:

s3://<workspace-bucket>/<signature-prefix>/<dataset>/<layer>/year=2026/month=10/day=05.json

Do not put the sidecars under a table LOCATION. Athena reads the
files under a table location without filtering by extension, so a
.json file next to your .parquet files can break real queries.
Athena does treat files that start with _ or . as hidden
(troubleshooting guide),
but a separate prefix is simpler to reason about, works the same for
every engine, and survives the "delete the partition prefix" step.

@dataclass(frozen=True)
class SignatureStore:
    """Read and write partition signatures as JSON sidecars in S3."""

    s3_client: Any
    bucket: str
    prefix: str

    def read(self, key: str) -> str | None:
        """Return the stored signature, or ``None`` if it does not exist."""
        try:
            response = self.s3_client.get_object(Bucket=self.bucket, Key=key)
        except ClientError as error:
            if error.response["Error"]["Code"] in _NOT_FOUND_CODES:
                return None
            raise
        body = json.loads(response["Body"].read().decode(_ENCODING))
        return body[_SIGNATURE_FIELD]

5. Put it together

def process(self, partition: dict[str, str]) -> bool:
    """Process one partition. Return ``True`` if it was rebuilt."""
    settings = self._settings
    upstream = self._signatures.read(self._signatures.key(
        settings.dataset, settings.upstream_layer, partition,
    ))
    own_key = self._signatures.key(settings.dataset, settings.layer, partition)
    expected = (
        chained_signature(upstream, self._sql_fingerprint)
        if upstream is not None else None
    )
    if expected is not None and expected == self._signatures.read(own_key):
        logger.info("Skip %s: signature unchanged", partition)
        return False

    self._delete_partition_prefix(partition)
    self._athena.run(self._sql_template.format(**partition))
    self._register_partition(partition)

    # Written last: if anything above fails, the next run retries.
    if expected is not None:
        self._signatures.write(own_key, expected)
    return True

Two rules keep this safe:

• Write the signature last. If the job dies after the delete or during the insert, the old signature stays, it no longer matches, and the next run rebuilds the partition.
• No upstream signature means "process". A day the layer above has never signed is always rebuilt and is not signed, so it will be checked again next time.

6. Pitfalls I would check first

Symptom	Likely cause
Every partition rebuilds on every run	A volatile column is part of the hash, or rows are not sorted before hashing
Duplicated rows after a rerun	INSERT INTO without deleting the partition prefix first
A query fails after adding signatures	Sidecars were written under a table LOCATION
A SQL fix did not reach old partitions	The SQL fingerprint is not part of the layer signature
A partition is never rebuilt after a failure	The signature is written before the work finishes

Wrap-up

• Make each partition replaceable: delete prefix, insert, register.
• Hash only business columns, sorted by a stable key.
• Chain signatures: sha256(upstream signature | SQL fingerprint).
• Keep sidecars out of table locations and write them last.

The full, runnable example (signature module and partition processor,
configured only through environment variables) is in the repository
that goes with this post.

What do you use to avoid reprocessing unchanged data in your lake?
Let me know in the comments.