commit on expire_snapshot tries to remove snapshot from wrong table.

[QlikFrederic]

Apache Iceberg version

main (development)

Please describe the bug 🐞

When using two iceberg tables and running the maintenance task in threads, the commit step will try to commit the snapshot expiration to the wrong table resulting in an error (snapshot does not exist).

Script to reproduce the issue:

from pyiceberg.catalog.memory import InMemoryCatalog
from datetime import datetime, timezone
import polars as pl
import threading
import time
import random

def generate_df(batch_id=0):
    df = pl.DataFrame(
        {
            "event_type": ["playback"] * 1000,
            "event_origin": [f"origin{random.randint(1, 5)}"] * 1000,
            "event_send_at": [datetime.now(timezone.utc)] * 1000,
            "event_saved_at": [datetime.now(timezone.utc)] * 1000,
            "data": [
                {
                    "calendarKey": f"calendarKey-{batch_id}",
                    "id": str(i + batch_id * 1000),
                    "referenceId": f"ref-{batch_id}-{i}",
                }
                for i in range(1000)
            ],
        }
    )
    return df

df = generate_df()
catalog = InMemoryCatalog("default", warehouse="/tmp/iceberg")
catalog.create_namespace_if_not_exists("default")
table1 = catalog.create_table_if_not_exists(
    "default.table1", schema=df.to_arrow().schema, location="/tmp/iceberg/table1"
)

table2 = catalog.create_table_if_not_exists(
    "default.table2", schema=df.to_arrow().schema, location="/tmp/iceberg/table2"
)

# Function to add multiple commits to a table
def add_commits_to_table(table, table_name, num_commits=5):
    print(f"Adding {num_commits} commits to {table_name}")
    for i in range(num_commits):
        df_batch = generate_df(batch_id=i)
        table.append(df_batch.to_arrow())
        print(f"  Added commit {i+1} to {table_name}")
        time.sleep(0.2)  # Small delay between commits

# Add multiple commits to both tables
print("Creating multiple snapshots...")
add_commits_to_table(table1, "table1")
add_commits_to_table(table2, "table2")

# Function to expire oldest 3 snapshots in a thread
def expire_oldest_snapshots(table, table_name):
    try:
        # Get all snapshots
        snapshots = list(table.snapshots())
        if len(snapshots) <= 3:
            print(f"{table_name}: Not enough snapshots to expire 3 (only {len(snapshots)})")
            return
        
        # Find the oldest 3 snapshots
        oldest_snapshots = snapshots[:3]
        oldest_ids = [snapshot.snapshot_id for snapshot in oldest_snapshots]
        
        print(f"{table_name}: Found {len(snapshots)} snapshots, expiring oldest 3: {oldest_ids}")
        
        # Expire the oldest 3 snapshots by IDs
        for id in oldest_ids:
            table.maintenance.expire_snapshots().by_id(id).commit()

       # can also be replaced with:
       # table.maintenance.expire_snapshots().by_ids(oldest_ids).commit()
        
        print(f"{table_name}: Successfully expired snapshots {oldest_ids}")
        
    except Exception as e:
        print(f"{table_name}: Error expiring snapshots: {e}")

# Run expire_snapshots in parallel threads
print("\nRunning expire_snapshots in parallel threads...")
thread1 = threading.Thread(target=expire_oldest_snapshots, args=(table1, "table1"))
thread2 = threading.Thread(target=expire_oldest_snapshots, args=(table2, "table2"))

thread1.start()
thread2.start()

thread1.join()
thread2.join()

print("\nDone!")

Willingness to contribute

• I can contribute a fix for this bug independently
• I would be willing to contribute a fix for this bug with guidance from the Iceberg community
• I cannot contribute a fix for this bug at this time

[kevinjqliu]

i suspect this is due to the stale table object passed to expire_oldest_snapshots.
i added table = table.refresh() on the top of expire_oldest_snapshots and the error vanished.

could you give that a try?

[QlikFrederic]

Hi, adding table = table.refresh() doesn't work with the in-memory catalog (then it seems to start searching for a sql catalog instead).
So I've adapted the script to use a sql catalog (and load the table from the catalog in the start of the expire_oldest_snapshots function),

• running 1 thread for snapshot expiration (commenting out the second thread): runs fine.
• running 2 threads: from table 1, snapshots expire successfully, table2 fails with snapshot ids not found (since it is trying to delete snapshots of table 1)
• adding table.refresh to the expire_oldest_snapshots function, gives me the same result (although this time it is table 2 that is successful and table 1 that fails deleting a snapshot of table 2)

from pyiceberg.catalog.sql import SqlCatalog
from datetime import datetime, timezone
import polars as pl
import os
import threading
import traceback
import time
import random
import shutil

def cleanup_workspace():
    """Clean up the iceberg workspace before starting"""
    try:
        shutil.rmtree("/tmp/iceberg", ignore_errors=True)
        # Recreate the directory for SQLite catalog
        os.makedirs("/tmp/iceberg", exist_ok=True)
        print("Cleaned up /tmp/iceberg workspace")
    except Exception as e:
        print(f"Warning: Could not clean workspace: {e}")

def generate_df(batch_id=0):
    df = pl.DataFrame(
        {
            "event_type": ["playback"] * 1000,
            "event_origin": [f"origin{random.randint(1, 5)}"] * 1000,
            "event_send_at": [datetime.now(timezone.utc)] * 1000,
            "event_saved_at": [datetime.now(timezone.utc)] * 1000,
            "data": [
                {
                    "calendarKey": f"calendarKey-{batch_id}",
                    "id": str(i + batch_id * 1000),
                    "referenceId": f"ref-{batch_id}-{i}",
                }
                for i in range(1000)
            ],
        }
    )
    return df

# Clean up workspace before starting
cleanup_workspace()

df = generate_df()
catalog = SqlCatalog(
    "default",
    warehouse="/tmp/iceberg",
    uri="sqlite:////tmp/iceberg/catalog.db"
)
catalog.create_namespace("default")
table1 = catalog.create_table(
    "default.table1", schema=df.to_arrow().schema, location="/tmp/iceberg/table1"
)

table2 = catalog.create_table(
    "default.table2", schema=df.to_arrow().schema, location="/tmp/iceberg/table2"
)

# Function to add multiple commits to a table
def add_commits_to_table(table, table_name, num_commits=5):
    print(f"Adding {num_commits} commits to {table_name}")
    for i in range(num_commits):
        df_batch = generate_df(batch_id=i)
        table.append(df_batch.to_arrow())
        print(f"  Added commit {i+1} to {table_name}")
        time.sleep(0.1)  # Small delay between commits

# Add multiple commits to both tables
print("Creating multiple snapshots...")
add_commits_to_table(table1, "table1")
add_commits_to_table(table2, "table2")

print("Tables created, starting expire operations...")

# Function to expire oldest 3 snapshots in a thread
def expire_oldest_snapshots(table_name):
    try:
        print(f"{table_name}: Starting expire operation...")
        
        # Reload table from catalog in this thread
        table = catalog.load_table(f"default.{table_name}")
        
        # Get all snapshots - get fresh snapshot list
        snapshots = list(table.snapshots())
        print(f"{table_name}: Current snapshots: {[s.snapshot_id for s in snapshots]}")
        
        if len(snapshots) <= 3:
            print(f"{table_name}: Not enough snapshots to expire 3 (only {len(snapshots)})")
            return
        
        # Find the oldest 3 snapshots
        oldest_snapshots = snapshots[:3]
        oldest_ids = [snapshot.snapshot_id for snapshot in oldest_snapshots]
        
        print(f"{table_name}: Found {len(snapshots)} snapshots, attempting to expire oldest 3: {oldest_ids}")
        
        # Verify the snapshots still exist before expiring
        current_ids = [s.snapshot_id for s in snapshots]
        for oid in oldest_ids:
            if oid not in current_ids:
                print(f"{table_name}: Snapshot {oid} no longer exists, skipping")
                return

        # Expire the oldest 3 snapshots by IDs
        table = table.refresh()
        with table.maintenance.expire_snapshots() as expire:
            for id in oldest_ids:
                expire.by_id(id)
        
        print(f"{table_name}: Successfully expired snapshots {oldest_ids}")
        
    except Exception as e:
        print(f"{table_name}: Error expiring snapshots: {e}")
        traceback.print_exc()

# Run expire_snapshots in parallel threads
print("\nRunning expire_snapshots in parallel threads...")
thread1 = threading.Thread(target=expire_oldest_snapshots, args=("table1",))
thread2 = threading.Thread(target=expire_oldest_snapshots, args=("table2",))

thread1.start()
thread2.start()

thread1.join()
thread2.join()

print("\nDone!")

[ForeverAngry]

@QlikFrederic thanks for reporting this! I'm sorry you ran into this bug :/

I believe i reproduced and found the cause.

I think the issue is here:

class ExpireSnapshots(UpdateTableMetadata["ExpireSnapshots"]):
    _snapshot_ids_to_expire: Set[int] = set()  # ❌ SHARED ACROSS ALL INSTANCES!
    _updates: Tuple[TableUpdate, ...] = ()      
    _requirements: Tuple[TableRequirement, ...] = ()  

Where the _snapshot_ids_to_expire is a class-level attribute, not instance-level, so when Thread 1 does table1.expire_snapshots().by_id(1001) and Thread 2 does table2.expire_snapshots().by_id(2001), they're both adding to the same shared set.

The fix seems trivia, i think... 🤞 - I moved those attributes to the __init__ method:

def __init__(self, transaction: Transaction) -> None:
    super().__init__(transaction)
    # ✅ Instance-level now - each table gets its own!
    self._snapshot_ids_to_expire: Set[int] = set()
    self._updates: Tuple[TableUpdate, ...] = ()
    self._requirements: Tuple[TableRequirement, ...] = ()

I wrote tests to reproduced the bug (literally got the exact same error message as your issue), applied the fix, and thread safety seemed to be restored.

That being said, if I'm right, the same issue exists in the ManageSnapshots class as well.

I'm traveling at the moment, once I'm home, I can push a branch for you to test (sometime in the next 24-48 hours). Let me know what you think!

[QlikFrederic]

@ForeverAngry great find! If you create a branch, I'll be happy to try it out.

[kevinjqliu]

Thanks everyone! This is a great find! @ForeverAngry looking forward to the PR!

That being said, if I'm right, the same issue exists in the ManageSnapshots class as well.

i think you're right. ManageSnapshots also declares a mutate state as class attributes

I generated a script which scans the entire codebase for class with mutable states

import ast
from pathlib import Path

# Types we consider "mutable" at class scope
MUTABLE_TYPES = (ast.List, ast.Dict, ast.Set)

def is_mutable(node: ast.AST) -> bool:
    """Return True if the node represents a mutable literal."""
    if isinstance(node, MUTABLE_TYPES):
        return True
    if isinstance(node, ast.Call) and isinstance(node.func, ast.Name):
        # e.g. list(), dict(), set(), defaultdict()
        return node.func.id in {"list", "dict", "set", "defaultdict"}
    return False

def find_mutable_class_attributes(file_path):
    with open(file_path, "r", encoding="utf-8") as f:
        source = f.read()

    try:
        tree = ast.parse(source, filename=file_path)
    except SyntaxError:
        return []

    results = []

    for node in ast.walk(tree):
        if isinstance(node, ast.ClassDef):
            class_name = node.name
            mutable_attrs = []

            for body_item in node.body:
                if isinstance(body_item, ast.Assign):
                    if is_mutable(body_item.value):
                        for target in body_item.targets:
                            if isinstance(target, ast.Name):
                                mutable_attrs.append(target.id)

                elif isinstance(body_item, ast.AnnAssign):
                    if is_mutable(body_item.value) and isinstance(body_item.target, ast.Name):
                        mutable_attrs.append(body_item.target.id)

            if mutable_attrs:
                results.append((class_name, mutable_attrs))

    return results

def scan_directory(directory="."):
    for path in Path(directory).rglob("*.py"):
        # Skip unwanted directories
        if any(part in ("tests", "vendor") for part in path.parts):
            continue

        class_attrs = find_mutable_class_attributes(path)
        for class_name, attributes in class_attrs:
            print(f"{path}: class {class_name} → {attributes}")

if __name__ == "__main__":
    scan_directory(".")

Here are the results,

➜  iceberg-python git:(main) ✗ poetry run python scan_class_attr.py
pyiceberg/catalog/sql.py: class IcebergNamespaceProperties → ['NAMESPACE_MINIMAL_PROPERTIES']
pyiceberg/utils/singleton.py: class Singleton → ['_instances']
pyiceberg/catalog/rest/auth.py: class AuthManagerFactory → ['_registry']
pyiceberg/table/update/spec.py: class UpdateSpec → ['_name_to_field', '_name_to_added_field', '_transform_to_field', '_transform_to_added_field', '_renames', '_added_time_fields']
pyiceberg/table/update/snapshot.py: class ExpireSnapshots → ['_snapshot_ids_to_expire']
pyiceberg/table/update/schema.py: class UpdateSchema → ['_adds', '_updates', '_deletes', '_moves', '_added_name_to_id', '_id_to_parent']

Some of these are false positives. But it looks like UpdateSpec, ExpireSnapshots, and UpdateSchema all have the same problem

[kevinjqliu]

Created #2427 to track this class of issues

[ForeverAngry]

@QlikFrederic take a look at: #2430

When you get a moment, try that Branch and let me know what you find! Feel free to leave your note or comments on the draft PR!

[QlikFrederic]

@ForeverAngry I tried the branch in our code and it works without issues now. 💯

[stevenayers]

Thanks for implementing this fix @ForeverAngry! Are there any timelines for when this'll be released?