Fix: Invalidate tablets when table is dropped via schema event

cassandra/metadata.py

@@ -225,6 +225,7 @@ def _drop_table(self, keyspace, table):
         except KeyError:
             # can happen if keyspace disappears while processing async event
             pass
+        self._table_removed(keyspace, table)
 
     def _update_type(self, type_meta):
         try:

tests/integration/standard/test_tablets.py

@@ -218,6 +218,37 @@ def drop_ks(_):
 
         self.run_tablets_invalidation_test(drop_ks)
 
+    def test_tablets_invalidation_drop_table(self):
+        """Dropping a table invalidates its tablet metadata via the schema change event."""
+        keyspace, table = "test_drop_table", "table1"
+
+        # Own keyspace/table so this test doesn't disturb state shared with other tests
+        self.session.execute(f"DROP KEYSPACE IF EXISTS {keyspace}")
+        self.session.execute(
+            f"""
+            CREATE KEYSPACE {keyspace}
+            WITH replication = {{
+                'class': 'NetworkTopologyStrategy',
+                'replication_factor': 2
+            }} AND tablets = {{
+                'initial': 8
+            }}
+            """)
+        self.session.execute(f"CREATE TABLE {keyspace}.{table} (pk int, ck int, v int, PRIMARY KEY (pk, ck))")
+
+        prepared = self.session.prepare(f"INSERT INTO {keyspace}.{table} (pk, ck, v) VALUES (?, ?, ?)")
+        for i in range(50):
+            self.session.execute(prepared.bind((i, i % 5, i % 2)))
+
+        def drop_table(_):
+            # Drop table to trigger tablets invalidation
+            self.session.execute(f"DROP TABLE {keyspace}.{table}")
+
+        try:
+            self.run_tablets_invalidation_test(drop_table, keyspace=keyspace, table=table)
+        finally:
+            self.session.execute(f"DROP KEYSPACE IF EXISTS {keyspace}")
+
     @pytest.mark.last
     def test_tablets_invalidation_decommission_non_cc_node(self):
         def decommission_non_cc_node(rec):
@@ -245,12 +276,12 @@ def decommission_non_cc_node(rec):
         self.run_tablets_invalidation_test(decommission_non_cc_node)
 
 
-    def run_tablets_invalidation_test(self, invalidate):
+    def run_tablets_invalidation_test(self, invalidate, keyspace="test1", table="table1"):
         # Make sure driver holds tablet info
         # By landing query to the host that is not in replica set
         bound = self.session.prepare(
-            """
-            SELECT pk, ck, v FROM test1.table1 WHERE pk = ?
+            f"""
+            SELECT pk, ck, v FROM {keyspace}.{table} WHERE pk = ?
             """).bind([(2)])
 
         rec = None

tests/unit/test_metadata.py

@@ -36,6 +36,7 @@
 from cassandra.policies import SimpleConvictionPolicy
 from cassandra.pool import Host
 from cassandra.protocol import QueryMessage
+from cassandra.tablets import Tablet
 from tests.util import assertCountEqual
 import pytest
 
@@ -441,6 +442,32 @@ def test_bytes_tokens(self):
         self._get_replicas(BytesToken)
 
 
+class DropTableMetadataTest(unittest.TestCase):
+    """Metadata._drop_table should invalidate tablets for the dropped table."""
+
+    def setUp(self):
+        """Set up metadata containing a table with a tablet record."""
+        self.metadata = Metadata()
+        keyspace = KeyspaceMetadata("ks", True, "NetworkTopologyStrategy", {"dc1": "1"})
+        keyspace.tables["tb"] = TableMetadata("ks", "tb")
+        self.metadata.keyspaces["ks"] = keyspace
+        self.metadata._tablets.add_tablet("ks", "tb", Tablet(0, 100, [("host1", 0)]))
+
+    def test_drop_table_invalidates_tablets(self):
+        """Dropping a known table removes its tablet and table metadata."""
+        self.metadata._drop_table("ks", "tb")
+
+        assert self.metadata._tablets.table_has_tablets("ks", "tb") is False
+        assert "tb" not in self.metadata.keyspaces["ks"].tables
+
+    def test_drop_table_invalidates_tablets_for_unknown_keyspace(self):
+        """Dropping a table in an unknown keyspace still removes its tablet metadata."""
+        self.metadata._tablets.add_tablet("unknown", "tb", Tablet(0, 100, [("host1", 0)]))
+        self.metadata._drop_table("unknown", "tb")
+
+        assert self.metadata._tablets.table_has_tablets("unknown", "tb") is False
+
+
 class Murmur3TokensTest(unittest.TestCase):
 
     def test_murmur3_init(self):