Adding drop table feature for tower tables

.gitignore

@@ -17,3 +17,5 @@ pytest.ini
 /crates/tower-api/.gitignore
 /scripts/openapi.json
 /scripts/openapi-generator-cli-*.jar
+
+.idea

src/tower/_tables.py

@@ -1,6 +1,8 @@
 from typing import Optional, Generic, TypeVar, Union, List
 from dataclasses import dataclass
 
+from pyiceberg.exceptions import NoSuchTableError
+
 TTable = TypeVar("TTable", bound="Table")
 
 import polars as pl
@@ -487,6 +489,43 @@ def create_if_not_exists(self, schema: pa.Schema) -> Table:
         return Table(self._context, table)
 
 
+    def drop(self) -> bool:
+        """
+        Drops (deletes) the Iceberg table from the catalog.
+
+        This method will:
+        1. Resolve the table's namespace (using default if not specified)
+        2. Drop the table from the catalog
+        3. Return True if successful, False if the table didn't exist
+
+        Returns:
+            bool: True if the table was successfully dropped, False if it didn't exist.
+
+        Raises:
+            CatalogError: If there are issues accessing the catalog or dropping the table.
+
+        Example:
+            >>> # Drop an existing table
+            >>> table_ref = tables("my_table", namespace="my_namespace")
+            >>> success = table_ref.drop()
+            >>> if success:
+            ...     print("Table dropped successfully")
+            ... else:
+            ...     print("Table didn't exist")
+        """
+        namespace = namespace_or_default(self._namespace)
+        table_name = make_table_name(self._name, namespace)
+
+        try:
+            self._catalog.drop_table(table_name)
+            return True
+        except NoSuchTableError:
+            # If the table doesn't exist or there's any other issue, return False
+            # The underlying PyIceberg catalog will raise different exceptions
+            # depending on the catalog implementation, so we catch all exceptions
+            return False
+
+
 def tables(
     name: str,
     catalog: Union[str, Catalog] = "default",
@@ -515,6 +554,7 @@ def tables(
             - Load an existing table using `load()`
             - Create a new table using `create()`
             - Create a table if it doesn't exist using `create_if_not_exists()`
+            - Drop an existing table using `drop()`
 
     Raises:
         CatalogError: If there are issues accessing or loading the specified catalog.
@@ -537,6 +577,11 @@ def tables(
 
         >>> # Create a table if it doesn't exist
         >>> table = tables("my_table").create_if_not_exists(schema)
+
+        >>> # Drop an existing table
+        >>> success = tables("my_table", namespace="my_namespace").drop()
+        >>> if success:
+        ...     print("Table dropped successfully")
     """
     if isinstance(catalog, str):
         catalog = load_catalog(catalog)

tests/tower/test_tables.py

@@ -560,3 +560,172 @@ def test_map_type_simple(in_memory_catalog):
     # Row 4 (null map)
     props4_series = df_read.filter(pl.col("id") == 4).select("properties").to_series()
     assert props4_series[0] is None
+
+
+def test_drop_existing_table(in_memory_catalog):
+    """Test dropping an existing table returns True."""
+    schema = pa.schema([
+        pa.field("id", pa.int64()),
+        pa.field("name", pa.string()),
+    ])
+
+    # Create a table first
+    ref = tower.tables("users_to_drop", catalog=in_memory_catalog)
+    table = ref.create(schema)
+    assert table is not None
+
+    # Insert some data to make sure the table exists and has content
+    data = pa.Table.from_pylist([{"id": 1, "name": "Alice"}], schema=schema)
+    table.insert(data)
+
+    # Verify the table exists by reading from it
+    df = table.read()
+    assert len(df) == 1
+
+    # Now drop the table - should return True
+    success = ref.drop()
+    assert success is True
+
+
+def test_drop_nonexistent_table(in_memory_catalog):
+    """Test dropping a non-existent table returns False."""
+    ref = tower.tables("nonexistent_table", catalog=in_memory_catalog)
+
+    # Try to drop a table that doesn't exist - should return False
+    success = ref.drop()
+    assert success is False
+
+
+def test_drop_table_with_namespace(in_memory_catalog):
+    """Test dropping a table with a specific namespace."""
+    schema = pa.schema([
+        pa.field("id", pa.int64()),
+        pa.field("data", pa.string()),
+    ])
+
+    # Create a table in a specific namespace
+    ref = tower.tables("test_table", catalog=in_memory_catalog, namespace="test_namespace")
+    table = ref.create(schema)
+    assert table is not None
+
+    # Insert some data to confirm the table exists
+    data = pa.Table.from_pylist([{"id": 1, "data": "test"}], schema=schema)
+    table.insert(data)
+
+    # Verify we can read from it
+    df = table.read()
+    assert len(df) == 1
+
+    # Drop the table - should succeed
+    success = ref.drop()
+    assert success is True
+
+    # Try to drop the same table again - should return False
+    success_again = ref.drop()
+    assert success_again is False
+
+
+def test_drop_and_recreate_table(in_memory_catalog):
+    """Test that we can drop a table and then recreate it."""
+    schema = pa.schema([
+        pa.field("id", pa.int64()),
+        pa.field("value", pa.string()),
+    ])
+
+    table_name = "drop_recreate_test"
+    ref = tower.tables(table_name, catalog=in_memory_catalog)
+
+    # Create and populate the table
+    table = ref.create(schema)
+    data = pa.Table.from_pylist([
+        {"id": 1, "value": "first"},
+        {"id": 2, "value": "second"}
+    ], schema=schema)
+    table.insert(data)
+
+    # Verify original data
+    df = table.read()
+    assert len(df) == 2
+
+    # Drop the table
+    success = ref.drop()
+    assert success is True
+
+    # Recreate the table with different data
+    new_table = ref.create(schema)
+    new_data = pa.Table.from_pylist([
+        {"id": 10, "value": "new_first"},
+        {"id": 20, "value": "new_second"},
+        {"id": 30, "value": "new_third"}
+    ], schema=schema)
+    new_table.insert(new_data)
+
+    # Verify new data
+    new_df = new_table.read()
+    assert len(new_df) == 3
+
+    # Make sure the old data is gone
+    ids = new_df.select("id").to_series().to_list()
+    assert 1 not in ids
+    assert 2 not in ids
+    assert 10 in ids
+    assert 20 in ids
+    assert 30 in ids
+
+
+def test_drop_multiple_tables(in_memory_catalog):
+    """Test dropping multiple tables."""
+    schema = pa.schema([
+        pa.field("id", pa.int64()),
+        pa.field("name", pa.string()),
+    ])
+
+    # Create multiple tables
+    table_names = ["table1", "table2", "table3"]
+    tables = {}
+
+    for name in table_names:
+        ref = tower.tables(name, catalog=in_memory_catalog)
+        table = ref.create(schema)
+        data = pa.Table.from_pylist([{"id": 1, "name": f"data_{name}"}], schema=schema)
+        table.insert(data)
+        tables[name] = ref
+
+    # Verify all tables exist by reading from them
+    for name, ref in tables.items():
+        table = ref.load()
+        df = table.read()
+        assert len(df) == 1
+
+    # Drop all tables
+    for name, ref in tables.items():
+        success = ref.drop()
+        assert success is True, f"Failed to drop table {name}"
+
+    # Verify all tables are gone
+    for name, ref in tables.items():
+        success = ref.drop()
+        assert success is False, f"Table {name} still exists after dropping"
+
+
+def test_drop_with_catalog_errors(in_memory_catalog):
+    """Test that catalog errors are properly propagated, not masked as False."""
+    from unittest.mock import patch
+    from pyiceberg.exceptions import NoSuchTableError
+
+    ref = tower.tables("test_table", catalog=in_memory_catalog)
+
+    # Test that NoSuchTableError returns False
+    with patch.object(in_memory_catalog, 'drop_table') as mock_drop:
+        mock_drop.side_effect = NoSuchTableError("Table not found")
+        success = ref.drop()
+        assert success is False
+        mock_drop.assert_called_once()
+
+    # Test that other exceptions are propagated
+    with patch.object(in_memory_catalog, 'drop_table') as mock_drop:
+        mock_drop.side_effect = RuntimeError("Catalog connection failed")
+
+        with pytest.raises(RuntimeError, match="Catalog connection failed"):
+            ref.drop()
+        mock_drop.assert_called_once()