Recovering files from S3 using Delete Markers

0. Intro

APIs and data lakes are unforgiving: one wrong path and you can “delete” an entire table. This post shows exactly how we recovered Iceberg tables deleted by mistake by leveraging S3 Versioning and delete markers, with a notebook‑friendly pandas workflow.

1. What happened

One of our cleanup scripts was designed to delete Glue tables along with their associated S3 data paths. Normally, each table is safely scoped to its own directory, e.g.:

raw-bucket/
└── schema/
    ├── customers/
    ├── orders/
    └── payments/

However, one table was mistakenly defined with its path set to the schema root instead of its dedicated folder:

raw-bucket/
└── schema/   ❌  (should have been raw-bucket/schema/table_name/)

When the cleanup ran, it deleted the entire schema/ prefix — effectively removing all tables under that schema, not just one. This included both data and metadata files for every Iceberg table in that schema.

Because we use Iceberg tables, recovering the wrong files could break table consistency since mismatched metadata and data files would render the tables unreadable. Our goal was therefore to recover exactly the files deleted during the incident, nothing more.

Fortunately, all S3 deletions are logged, and our buckets had versioning and delete markers enabled, letting us precisely identify the deletion window and safely restore the affected files.

2. S3 Versioning

When working with critical datasets like Iceberg tables, S3 Versioning is your safety net. It allows you to recover any object or undo accidental deletions by maintaining historical versions of every file.

aws s3api put-bucket-versioning \
  --bucket <your-bucket-name> \
  --versioning-configuration Status=Enabled

3. How Delete Markers Work

When S3 versioning is enabled, deleting an object doesn’t actually remove its data. Instead, AWS adds a delete marker which is a special placeholder that marks the object as deleted while keeping all its previous versions safely stored.

Key: data/table.parquet
├── v3 ─ Delete Marker   ← hides everything below (latest)
├── v2 ─ Data version
└── v1 ─ Data version

Key: logs/2025-10-31.json
├── v4 ─ Delete Marker   ← current hidden state
├── v3 ─ Delete Marker
├── v2 ─ Data version
└── v1 ─ Data version

4. Our Recovery Strategy

After confirming versioning was enabled, the goal was to recover only the files deleted during the incident window — not everything that had ever changed. Recovering too much could break Iceberg table consistency, so precision was essential.

5. Investigate delete markers

Before restoring anything, we first enumerate all delete markers for a given prefix and filter them to the exact incident window. This gives us a clear, auditable list of what would be restored.

import boto3
import pandas as pd
from loguru import logger
from tqdm.notebook import tqdm

COLS_SORTING = {"eligible": False, "last_modified": True, "key": True}


def ensure_versioning_enabled(s3, bucket):
    logger.info(f"Checking if versioning is enabled for {bucket=}")
    status = s3.get_bucket_versioning(Bucket=bucket).get("Status", "NotEnabled")
    if status not in ("Enabled", "Suspended"):
        raise RuntimeError(
            f"Versioning not enabled for {bucket=}. Unable to auto-restore via delete markers."
        )
    logger.info(f"Versioning is {status=} for {bucket=}")
    return status


def _iter_delete_markers(s3, bucket, prefix):
    paginator = s3.get_paginator("list_object_versions")
    pages = paginator.paginate(Bucket=bucket, Prefix=prefix)

    for page in tqdm(pages, desc="Listing pages", unit="page"):
        delete_markers = page.get("DeleteMarkers", []) or []
        for dm in delete_markers:
            yield dm


def scan_delete_markers(s3, bucket, prefix, start_ts, end_ts):
    logger.info(f"Scanning delete markers under {bucket=} {prefix=}")

    rows = []
    for dm in _iter_delete_markers(s3, bucket, prefix):
        rows.append(
            {
                "key": dm["Key"],
                "version_id": dm["VersionId"],
                "is_latest": bool(dm.get("IsLatest")),
                "last_modified": dm["LastModified"],
            }
        )

    logger.info(f"Scan found {len(rows)} delete markers. Building dataframe")
    df = pd.DataFrame(rows)
    if df.empty:
        logger.warning("There are no delete markers")
        return df

    df["last_modified"] = pd.to_datetime(df["last_modified"], utc=True)

    start_ts = pd.to_datetime(start_ts, utc=True)
    end_ts = pd.to_datetime(end_ts, utc=True)

    logger.info(f"Filtering with {start_ts=} and {end_ts=}")
    df["eligible"] = (df["last_modified"] >= start_ts) & (df["last_modified"] <= end_ts)

    eligible_count = int(df["eligible"].sum())
    unique_keys = df.loc[df["eligible"], "key"].nunique()
    logger.info(f"{eligible_count=} ({unique_keys=}) for {bucket=} {prefix=}")

    return df.sort_values(
        list(COLS_SORTING.keys()), ascending=list(COLS_SORTING.values())
    )

6. Actually recovering files

Now that we have an audited list of delete markers, we restore by deleting those markers. Start with a dry‑run to validate scope, then apply in batches.

def remove_delete_markers(s3, bucket, df_in, dry_run=True, batch_size=1000):
    logger.info(f"Preparing delete-marker removal with {dry_run=} and {batch_size=}")
    if df_in.empty:
        logger.info("No delete markers were provided to the remover (dataframe is empty)")
        return False

    df = df_in.loc[df_in["eligible"], ["key", "version_id"]].copy()
    planned = len(df)
    logger.info(f"There are {planned} files to be restored")

    if planned == 0 or dry_run:
        logger.warning("Dry-run active: will not remove any delete markers")
        return False

    logger.info("Removing eligible delete markers in batches")
    restored = 0
    batch = []
    for rec in tqdm(df.to_dict(orient="records"), desc="Removing markers", unit="dm"):
        batch.append({"Key": rec["key"], "VersionId": rec["version_id"]})

        if len(batch) == batch_size:
            logger.info(f"Restoring {len(batch)} files")
            s3.delete_objects(Bucket=bucket, Delete={"Objects": batch})
            restored += len(batch)
            batch = []

    if batch:
        logger.info(f"Restoring {len(batch)} files (done {restored}/{planned})")
        s3.delete_objects(Bucket=bucket, Delete={"Objects": batch})
        restored += len(batch)

    logger.info(f"Job done {restored=} out of {planned=}")
    return True

Usage skeleton

def recover_s3_delete_markers(bucket, prefix, start_ts, end_ts, dry_run=True, export_excel=False):
    logger.info("Starting delete-marker recovery flow"
                f" ({bucket=} {prefix=} {start_ts=} {end_ts=} {dry_run=})")

    assert prefix is not None, "prefix must be non-empty"
    assert start_ts is not None, "start_ts must be non-empty"
    assert end_ts is not None, "end_ts must be non-empty"

    s3 = boto3.client("s3")
    ensure_versioning_enabled(s3, bucket)
    df = scan_delete_markers(s3, bucket, prefix, start_ts, end_ts)
    remove_delete_markers(s3, bucket, df, dry_run=dry_run)

    if export_excel:
        logger.info("Exporting delete markers list to FILE_EXCEL")
        df_x = df.copy()
        df_x["last_modified"] = df_x["last_modified"].dt.tz_localize(None)
        df_x.to_excel("files_to_recover.xlsx", index=False)

    return df

7. Conclusion

Recovering the Iceberg tables wasn’t about luck, it was about preparation. Versioning gave us a rewind button, and precise recovery logic kept metadata and data aligned.