diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 2a02f1ac54..d1ab4f1a8a 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -9,6 +9,10 @@ Features Others ------ +* Token-aware request routing is now maintained when ``schema_metadata_enabled`` is + disabled: the driver fetches the keyspace replication strategies from the lightweight + ``system_schema.keyspaces`` table (on connect and on keyspace schema change events), + and drops tablet metadata for dropped tables and keyspaces. * ``PreparedStatement.result_metadata`` and ``PreparedStatement.result_metadata_id`` are now read-only. They are replaced together by ``PreparedStatement.update_result_metadata()``, so a request can never observe a metadata diff --git a/cassandra/cluster.py b/cassandra/cluster.py index bcc7852c33..0b17f2bd67 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -69,7 +69,7 @@ RESULT_KIND_SET_KEYSPACE, RESULT_KIND_ROWS, RESULT_KIND_SCHEMA_CHANGE, ProtocolHandler, RESULT_KIND_VOID, ProtocolException) -from cassandra.metadata import Metadata, Token, protect_name, murmur3, _NodeInfo +from cassandra.metadata import Metadata, Token, protect_name, murmur3, _NodeInfo, ReplicationStrategy from cassandra.policies import (TokenAwarePolicy, DCAwareRoundRobinPolicy, SimpleConvictionPolicy, ExponentialReconnectionPolicy, HostDistance, RetryPolicy, IdentityTranslator, NoSpeculativeExecutionPlan, @@ -1153,8 +1153,9 @@ def schema_metadata_enabled(self): Flag indicating whether internal schema metadata is updated. When disabled, the driver does not populate Cluster.metadata.keyspaces on connect, or on schema change events. This - can be used to speed initial connection, and reduce load on client and server during operation. Turning this off - gives away token aware request routing, and programmatic inspection of the metadata model. + can be used to speed initial connection, and reduce load on client and server during operation. Token aware request + routing is still maintained using a lightweight query of keyspace replication strategies, but programmatic inspection + of the metadata model is not available. """ return self.control_connection._schema_meta_enabled @@ -3843,6 +3844,8 @@ class ControlConnection(object): _SELECT_SCHEMA_PEERS_TEMPLATE = "SELECT peer, host_id, {nt_col_name}, schema_version FROM system.peers" _SELECT_SCHEMA_LOCAL = "SELECT schema_version FROM system.local WHERE key='local'" + _SELECT_KEYSPACES_REPLICATION = "SELECT keyspace_name, replication FROM system_schema.keyspaces" + _SELECT_PEERS_V2 = "SELECT * FROM system.peers_v2" _SELECT_PEERS_NO_TOKENS_V2 = "SELECT host_id, peer, peer_port, data_center, rack, native_address, native_port, release_version, schema_version FROM system.peers_v2" _SELECT_SCHEMA_PEERS_V2 = "SELECT host_id, peer, peer_port, native_address, native_port, schema_version FROM system.peers_v2" @@ -4132,6 +4135,53 @@ def refresh_schema(self, force=False, **kwargs): self._signal_error() return False + def _refresh_replication_strategies(self, connection, keyspace=None): + """ + Fetch the keyspace replication strategies from the lightweight + ``system_schema.keyspaces`` table. Used to support token-aware routing + when schema metadata is disabled. When ``keyspace`` is given, only that + keyspace is re-read and merged into the current strategies. + """ + if not self._token_meta_enabled: + return + select = self._SELECT_KEYSPACES_REPLICATION + if keyspace is not None: + select += bind_params(" WHERE keyspace_name = %s", (keyspace,), Encoder()) + cl = ConsistencyLevel.ONE + query = QueryMessage( + query=maybe_add_timeout_to_query(select, self._metadata_request_timeout), + consistency_level=cl) + try: + result = connection.wait_for_response(query, timeout=self._timeout) + rows = dict_factory(result.column_names, result.parsed_rows) + except Exception as exc: + # not supported by very old servers; degrade to no token-aware routing + log.warning("[control connection] Failed to fetch keyspace replication strategies, " + "token-aware routing may be degraded: %s", exc) + return + + strategies = {} + for row in rows: + try: + replication = dict(row["replication"]) + strategy_class = replication.pop("class") + except (TypeError, KeyError): + log.warning("[control connection] Skipping keyspace %s with unparseable " + "replication settings", row.get("keyspace_name")) + continue + strategy = ReplicationStrategy.create(strategy_class, replication) + if strategy: + strategies[row["keyspace_name"]] = strategy + + if keyspace is not None: + # merge the single re-read keyspace into the current map; a missing + # row means the keyspace was dropped + merged = dict(self._cluster.metadata._keyspace_replication_strategies) + merged.pop(keyspace, None) + merged.update(strategies) + strategies = merged + self._cluster.metadata._update_replication_strategies(strategies) + def _refresh_schema(self, connection, preloaded_results=None, schema_agreement_wait=None, force=False, **kwargs): if self._cluster.is_shutdown: return False @@ -4140,14 +4190,28 @@ def _refresh_schema(self, connection, preloaded_results=None, schema_agreement_w preloaded_results=preloaded_results, wait_time=schema_agreement_wait) - if not self._schema_meta_enabled and not force: - log.debug("[control connection] Skipping schema refresh because schema metadata is disabled") - return False - if not agreed: log.debug("Skipping schema refresh due to lack of schema agreement") return False + if not self._schema_meta_enabled and not force: + target_type = kwargs.get("target_type") + change_type = kwargs.get("change_type") + if not target_type: + # keep token-aware routing functional without fetching the full schema + self._refresh_replication_strategies(connection) + elif target_type.lower() == "keyspace": + # a single keyspace changed; no need to re-read the whole table + self._refresh_replication_strategies(connection, keyspace=kwargs.get("keyspace")) + elif target_type.lower() == "table" and change_type == "DROPPED": + # keep tablet metadata up to date for dropped tables + keyspace = kwargs.get("keyspace") + table = kwargs.get("table") + if keyspace and table: + self._cluster.metadata._table_removed(keyspace, table) + log.debug("[control connection] Skipping schema refresh because schema metadata is disabled") + return False + self._cluster.metadata.refresh( connection, self._timeout, diff --git a/cassandra/metadata.py b/cassandra/metadata.py index 5669e6a80e..23205f375c 100644 --- a/cassandra/metadata.py +++ b/cassandra/metadata.py @@ -24,6 +24,7 @@ import re import sys from threading import RLock +from warnings import warn import struct import random import itertools @@ -129,6 +130,7 @@ def __init__(self): self._host_id_by_endpoint = {} self._hosts_lock = RLock() self._tablets = Tablets({}) + self._keyspace_replication_strategies = {} def export_schema_as_string(self): """ @@ -273,16 +275,52 @@ def _table_removed(self, keyspace, table): def _keyspace_added(self, ksname): if self.token_map: self.token_map.rebuild_keyspace(ksname, build_if_absent=False) + ks_meta = self.keyspaces.get(ksname) + if ks_meta and ks_meta.replication_strategy: + self._keyspace_replication_strategies[ksname] = ks_meta.replication_strategy def _keyspace_updated(self, ksname): if self.token_map: self.token_map.rebuild_keyspace(ksname, build_if_absent=False) self._tablets.drop_tablets(ksname) + ks_meta = self.keyspaces.get(ksname) + if ks_meta and ks_meta.replication_strategy: + self._keyspace_replication_strategies[ksname] = ks_meta.replication_strategy def _keyspace_removed(self, ksname): if self.token_map: self.token_map.remove_keyspace(ksname) self._tablets.drop_tablets(ksname) + self._keyspace_replication_strategies.pop(ksname, None) + + def _update_replication_strategies(self, replication_strategies): + """ + Update the keyspace replication strategies used for token-aware routing. + + Used to maintain the token map when schema metadata is disabled and + full keyspace metadata is not available. + """ + removed = set(self._keyspace_replication_strategies) - set(replication_strategies) + for keyspace in removed: + self._keyspace_replication_strategies.pop(keyspace, None) + self.keyspaces.pop(keyspace, None) # stale metadata if it was previously enabled + self._tablets.drop_tablets(keyspace) + if self.token_map: + self.token_map.remove_keyspace(keyspace) + changed = [(keyspace, strategy) for keyspace, strategy in replication_strategies.items() + if self._keyspace_replication_strategies.get(keyspace) != strategy] + self._keyspace_replication_strategies.update(replication_strategies) + token_map = self.token_map + for keyspace, strategy in changed: + ks_meta = self.keyspaces.get(keyspace) + if ks_meta is not None and ks_meta.replication_strategy != strategy: + # keep the metadata in sync so the token map rebuild uses the fresh strategy + ks_meta.replication_strategy = strategy + # a replication change can alter the tablet layout, so stale tablets are dropped + self._tablets.drop_tablets(keyspace) + if token_map: + token_map.remove_keyspace(keyspace) + token_map.rebuild_keyspace(keyspace, build_if_absent=True) def rebuild_token_map(self, partitioner, token_map): """ @@ -1847,10 +1885,17 @@ def rebuild_keyspace(self, keyspace, build_if_absent=False): try: current = self.tokens_to_hosts_by_ks.get(keyspace, None) if (build_if_absent and current is None) or (not build_if_absent and current is not None): + strategy = None ks_meta = self._metadata.keyspaces.get(keyspace) if ks_meta: - replica_map = self.replica_map_for_keyspace(self._metadata.keyspaces[keyspace]) - self.tokens_to_hosts_by_ks[keyspace] = replica_map + strategy = ks_meta.replication_strategy + if strategy is None: + strategy = self._metadata._keyspace_replication_strategies.get(keyspace) + if strategy: + replica_map = strategy.make_token_replica_map(self.token_to_host_owner, self.ring) + else: + replica_map = None + self.tokens_to_hosts_by_ks[keyspace] = replica_map except Exception: # should not happen normally, but we don't want to blow up queries because of unexpected meta state # bypass until new map is generated @@ -1858,6 +1903,8 @@ def rebuild_keyspace(self, keyspace, build_if_absent=False): log.exception("Failed creating a token map for keyspace '%s' with %s. PLEASE REPORT THIS: https://datastax-oss.atlassian.net/projects/PYTHON", keyspace, self.token_to_host_owner) def replica_map_for_keyspace(self, ks_metadata): + warn("TokenMap.replica_map_for_keyspace is deprecated and will be " + "removed in a future release.", DeprecationWarning, stacklevel=2) strategy = ks_metadata.replication_strategy if strategy: return strategy.make_token_replica_map(self.token_to_host_owner, self.ring) diff --git a/tests/integration/standard/test_metadata.py b/tests/integration/standard/test_metadata.py index 562f457a32..26169dbfbd 100644 --- a/tests/integration/standard/test_metadata.py +++ b/tests/integration/standard/test_metadata.py @@ -30,6 +30,7 @@ RegisteredTableExtension, _RegisteredExtensionType, get_schema_parser, group_keys_by_replica, NO_VALID_REPLICA) from cassandra.protocol import QueryMessage, ProtocolHandler +from cassandra.query import SimpleStatement from tests.integration import (get_cluster, use_singledc, PROTOCOL_VERSION, execute_until_pass, BasicSegregatedKeyspaceUnitTestCase, BasicSharedKeyspaceUnitTestCase, @@ -170,6 +171,40 @@ def test_schema_metadata_disable(self): no_schema.shutdown() no_token.shutdown() + def test_schema_metadata_disable_token_aware_routing(self): + """ + Token-aware routing is maintained when schema metadata is disabled. + + @since 3.30 + @expected_result The token map is populated from keyspace replication strategies without full schema metadata + + @test_category metadata + """ + no_schema = TestCluster(schema_metadata_enabled=False) + try: + no_schema_session = no_schema.connect() + assert len(no_schema.metadata.keyspaces) == 0 + + # replication strategies are fetched without the full schema + assert no_schema.metadata._keyspace_replication_strategies.get(self.ks_name) is not None + + # token-aware routing works via the replication strategies + token_map = no_schema.metadata.token_map + replicas = token_map.get_replicas(self.ks_name, token_map.token_class.from_string("0")) + assert replicas + + # a token-aware query is routed to a replica of the keyspace + statement = SimpleStatement( + "SELECT * FROM system.local WHERE key='local'", + keyspace=self.ks_name, routing_key=b"routing-key") + rs = no_schema_session.execute(statement) + assert rs.one() is not None + replicas = set(no_schema.metadata.get_replicas(self.ks_name, statement.routing_key)) + # the first host attempted; retries may move the query off a replica + assert rs.response_future.attempted_hosts[0] in replicas + finally: + no_schema.shutdown() + def make_create_statement(self, partition_cols, clustering_cols=None, other_cols=None): clustering_cols = clustering_cols or [] other_cols = other_cols or [] diff --git a/tests/unit/test_control_connection.py b/tests/unit/test_control_connection.py index fd62323f33..fa8938c5a6 100644 --- a/tests/unit/test_control_connection.py +++ b/tests/unit/test_control_connection.py @@ -24,6 +24,8 @@ from cassandra.connection import EndPoint, DefaultEndPoint, DefaultEndPointFactory from cassandra.policies import (SimpleConvictionPolicy, RoundRobinPolicy, ConstantReconnectionPolicy, IdentityTranslator) +from cassandra.metadata import SimpleStrategy, LocalStrategy, Metadata +from cassandra.tablets import Tablet PEER_IP = "foobar" @@ -49,6 +51,7 @@ def __init__(self): self.partitioner = None self.token_map = {} self.removed_hosts = [] + self._keyspace_replication_strategies = {} def get_host(self, endpoint_or_address, port=None): if not isinstance(endpoint_or_address, EndPoint): @@ -491,6 +494,217 @@ def test_handle_topology_change(self): self.control_connection._handle_topology_change(event) self.cluster.scheduler.schedule_unique.assert_called_once_with(ANY, self.control_connection._refresh_nodes_if_not_up, None) + def test_refresh_schema_disabled_fetches_replication_strategies(self): + """ + When schema metadata is disabled, the full schema refresh is skipped + but keyspace replication strategies are fetched for token-aware routing. + """ + self.control_connection._schema_meta_enabled = False + result = ResultMessage(kind=RESULT_KIND_ROWS) + result.column_names = ["keyspace_name", "replication"] + result.parsed_rows = [ + ["ks", {"class": "org.apache.cassandra.locator.SimpleStrategy", "replication_factor": "1"}], + ["system", {"class": "org.apache.cassandra.locator.LocalStrategy", "replication_factor": "1"}], + ] + self.connection.wait_for_response = Mock(return_value=result) + self.cluster.metadata._update_replication_strategies = Mock() + + self.control_connection._refresh_schema(self.connection) + + assert self.connection.wait_for_response.call_count == 1 + strategies = self.cluster.metadata._update_replication_strategies.call_args[0][0] + assert isinstance(strategies["ks"], SimpleStrategy) + assert strategies["ks"].replication_factor == 1 + assert isinstance(strategies["system"], LocalStrategy) + + def test_refresh_schema_disabled_keyspace_event_fetches_replication_strategies(self): + """ + Keyspace schema change events are processed when schema metadata is + disabled to keep token-aware routing up to date. The read is filtered + to the changed keyspace and merged into the known strategies. + """ + self.control_connection._schema_meta_enabled = False + self.cluster.metadata._keyspace_replication_strategies = {"other": SimpleStrategy({"replication_factor": "3"})} + result = ResultMessage(kind=RESULT_KIND_ROWS) + result.column_names = ["keyspace_name", "replication"] + result.parsed_rows = [ + ["ks", {"class": "org.apache.cassandra.locator.SimpleStrategy", "replication_factor": "1"}], + ] + self.connection.wait_for_response = Mock(return_value=result) + self.cluster.metadata._update_replication_strategies = Mock() + + self.control_connection._refresh_schema( + self.connection, target_type="KEYSPACE", change_type="CREATED", keyspace="ks") + + query = self.connection.wait_for_response.call_args[0][0].query + assert "WHERE keyspace_name = 'ks'" in query + self.cluster.metadata._update_replication_strategies.assert_called_once() + strategies = self.cluster.metadata._update_replication_strategies.call_args[0][0] + assert set(strategies) == {"ks", "other"} + + def test_refresh_schema_disabled_keyspace_drop_removes_strategy(self): + """ + A dropped keyspace's strategy is removed: the filtered read returns no + row and the merge drops the keyspace from the known strategies. + """ + self.control_connection._schema_meta_enabled = False + self.cluster.metadata._keyspace_replication_strategies = { + "ks": SimpleStrategy({"replication_factor": "1"}), + "other": SimpleStrategy({"replication_factor": "3"}), + } + result = ResultMessage(kind=RESULT_KIND_ROWS) + result.column_names = ["keyspace_name", "replication"] + result.parsed_rows = [] + self.connection.wait_for_response = Mock(return_value=result) + self.cluster.metadata._update_replication_strategies = Mock() + + self.control_connection._refresh_schema( + self.connection, target_type="KEYSPACE", change_type="DROPPED", keyspace="ks") + + strategies = self.cluster.metadata._update_replication_strategies.call_args[0][0] + assert set(strategies) == {"other"} + + def test_refresh_schema_disabled_waits_for_schema_agreement(self): + """ + The replication-strategy read runs at ConsistencyLevel.ONE, so it must + only happen after schema agreement; a pre-agreement read can return the + stale pre-change replication settings. + """ + self.control_connection._schema_meta_enabled = False + + # one peer starts on a different schema version and converges later + self.connection.peer_results[1][1][2] = 'b' + + def converge_then_respond(*args, **kwargs): + if self.time.clock >= 1: + self.connection.peer_results[1][1][2] = 'a' + return _node_meta_results(self.connection.local_results, self.connection.peer_results) + + self.connection.wait_for_responses = Mock(side_effect=converge_then_respond) + + result = ResultMessage(kind=RESULT_KIND_ROWS) + result.column_names = ["keyspace_name", "replication"] + result.parsed_rows = [ + ["ks", {"class": "org.apache.cassandra.locator.SimpleStrategy", "replication_factor": "1"}], + ] + self.connection.wait_for_response = Mock(return_value=result) + self.cluster.metadata._update_replication_strategies = Mock() + + self.control_connection._refresh_schema( + self.connection, target_type="KEYSPACE", change_type="UPDATED", keyspace="ks") + + # slept until the disagreeing peer converged, then fetched strategies + assert self.time.clock > 0 + self.cluster.metadata._update_replication_strategies.assert_called_once() + + def test_refresh_schema_disabled_no_agreement_skips_strategies(self): + """ + Without schema agreement the strategies are not fetched at all, rather + than being fetched from a node with a stale schema. + """ + self.control_connection._schema_meta_enabled = False + # a peer that never agrees + self.connection.peer_results[1][1][2] = 'b' + self.connection.wait_for_response = Mock() + self.cluster.metadata._update_replication_strategies = Mock() + + assert self.control_connection._refresh_schema(self.connection) is False + + self.connection.wait_for_response.assert_not_called() + self.cluster.metadata._update_replication_strategies.assert_not_called() + + def test_refresh_schema_disabled_table_dropped_invalidates_tablets(self): + """ + Dropped table events are processed when schema metadata is disabled so + tablet metadata stays up to date. + """ + self.control_connection._schema_meta_enabled = False + metadata = Metadata() + metadata._tablets.add_tablet("ks", "tb", Tablet(0, 100, [("host1", 0)])) + self.cluster.metadata = metadata + + self.control_connection._refresh_schema( + self.connection, target_type="TABLE", change_type="DROPPED", keyspace="ks", table="tb") + + assert metadata._tablets.table_has_tablets("ks", "tb") is False + + def test_refresh_schema_disabled_table_dropped_without_names(self): + """ + A dropped table event without keyspace/table names is ignored without + raising, keeping the control connection healthy. + """ + self.control_connection._schema_meta_enabled = False + self.cluster.metadata._table_removed = Mock() + + self.control_connection._refresh_schema( + self.connection, target_type="TABLE", change_type="DROPPED") + + self.cluster.metadata._table_removed.assert_not_called() + + def test_refresh_schema_disabled_ignores_other_events(self): + """ + Non-keyspace, non-dropped-table schema events are ignored when schema + metadata is disabled. + """ + self.control_connection._schema_meta_enabled = False + self.cluster.metadata._update_replication_strategies = Mock() + self.cluster.metadata._table_removed = Mock() + + self.control_connection._refresh_schema( + self.connection, target_type="TABLE", change_type="CREATED", keyspace="ks", table="tb") + + self.cluster.metadata._update_replication_strategies.assert_not_called() + self.cluster.metadata._table_removed.assert_not_called() + + def test_refresh_schema_disabled_skips_strategies_without_token_metadata(self): + """ + Replication strategies are not fetched when token metadata is disabled. + """ + self.control_connection._schema_meta_enabled = False + self.control_connection._token_meta_enabled = False + self.connection.wait_for_response = Mock() + self.cluster.metadata._update_replication_strategies = Mock() + + self.control_connection._refresh_schema(self.connection) + + self.connection.wait_for_response.assert_not_called() + self.cluster.metadata._update_replication_strategies.assert_not_called() + + def test_refresh_schema_disabled_degrades_gracefully_on_fetch_error(self): + """ + A failure fetching replication strategies is not propagated, so the + control connection stays healthy on servers without system_schema.keyspaces. + """ + self.control_connection._schema_meta_enabled = False + self.connection.wait_for_response = Mock(side_effect=OperationTimedOut()) + self.cluster.metadata._update_replication_strategies = Mock() + + assert self.control_connection._refresh_schema(self.connection) is False + + self.cluster.metadata._update_replication_strategies.assert_not_called() + + def test_refresh_schema_disabled_skips_malformed_strategy_row(self): + """ + A malformed row is skipped without discarding the strategies parsed + from the remaining valid rows. + """ + self.control_connection._schema_meta_enabled = False + result = ResultMessage(kind=RESULT_KIND_ROWS) + result.column_names = ["keyspace_name", "replication"] + result.parsed_rows = [ + ["bad", None], + ["good", {"class": "org.apache.cassandra.locator.SimpleStrategy", "replication_factor": "1"}], + ] + self.connection.wait_for_response = Mock(return_value=result) + self.cluster.metadata._update_replication_strategies = Mock() + + self.control_connection._refresh_schema(self.connection) + + self.cluster.metadata._update_replication_strategies.assert_called_once() + strategies = self.cluster.metadata._update_replication_strategies.call_args[0][0] + assert set(strategies) == {"good"} + assert isinstance(strategies["good"], SimpleStrategy) + def test_handle_status_change(self): event = { 'change_type': 'UP', diff --git a/tests/unit/test_metadata.py b/tests/unit/test_metadata.py index a058f73c61..170ae14b2b 100644 --- a/tests/unit/test_metadata.py +++ b/tests/unit/test_metadata.py @@ -15,7 +15,7 @@ from binascii import unhexlify import logging -from unittest.mock import Mock +from unittest.mock import Mock, patch import os import timeit import uuid @@ -443,6 +443,145 @@ def test_bytes_tokens(self): self._get_replicas(BytesToken) +class TokenMapReplicationStrategiesTest(unittest.TestCase): + """ + TokenMap should use Metadata._keyspace_replication_strategies to compute + replicas when keyspace metadata is unavailable (schema_metadata_enabled=False). + """ + + def _build_metadata(self): + tokens = [Murmur3Token(i) for i in range(0, 500, 100)] + hosts = [Host("ip%d" % i, SimpleConvictionPolicy, datacenter="dc1", rack="rack1", host_id=uuid.uuid4()) + for i in range(len(tokens))] + metadata = Metadata() + metadata.rebuild_token_map( + "org.apache.cassandra.dht.Murmur3Partitioner", + {host: [str(token.value)] for host, token in zip(hosts, tokens)}) + return metadata, tokens, hosts + + def test_get_replicas_without_keyspace_metadata(self): + """Replica lookup works without keyspace metadata when the replication strategy is known.""" + metadata, tokens, hosts = self._build_metadata() + metadata._update_replication_strategies({ + "ks": ReplicationStrategy.create( + "org.apache.cassandra.locator.SimpleStrategy", {"replication_factor": "1"})}) + + replicas = metadata.token_map.get_replicas("ks", tokens[0]) + assert set(replicas) == {hosts[0]} + assert metadata.keyspaces == {} + + def test_update_replication_strategies_removes_dropped_keyspaces(self): + """Removing a keyspace drops its token map entry and tablets.""" + metadata, _, _ = self._build_metadata() + metadata._update_replication_strategies({ + "ks": ReplicationStrategy.create( + "org.apache.cassandra.locator.SimpleStrategy", {"replication_factor": "1"})}) + metadata._tablets.add_tablet("ks", "tb", Tablet(0, 100, [("host1", 0)])) + assert "ks" in metadata._keyspace_replication_strategies + assert "ks" in metadata.token_map.tokens_to_hosts_by_ks + + metadata._update_replication_strategies({}) + + assert metadata._keyspace_replication_strategies == {} + assert metadata.token_map.tokens_to_hosts_by_ks == {} + assert metadata._tablets.table_has_tablets("ks", "tb") is False + + def test_keyspace_added_syncs_replication_strategies(self): + """Schema refresh keeps _keyspace_replication_strategies in sync with keyspace metadata.""" + metadata, _, _ = self._build_metadata() + keyspace = KeyspaceMetadata("ks", True, "NetworkTopologyStrategy", {"dc1": "1"}) + metadata.keyspaces["ks"] = keyspace + metadata._keyspace_added("ks") + + assert metadata._keyspace_replication_strategies["ks"] is keyspace.replication_strategy + + def test_keyspace_removed_syncs_replication_strategies(self): + """Removing a keyspace clears its replication strategy and tablets.""" + metadata, _, _ = self._build_metadata() + metadata._update_replication_strategies({ + "ks": ReplicationStrategy.create( + "org.apache.cassandra.locator.SimpleStrategy", {"replication_factor": "1"})}) + metadata._tablets.add_tablet("ks", "tb", Tablet(0, 100, [("host1", 0)])) + + metadata._keyspace_removed("ks") + + assert metadata._keyspace_replication_strategies == {} + assert metadata._tablets.table_has_tablets("ks", "tb") is False + + def test_rebuild_token_map_keeps_routing_after_topology_change(self): + """A rebuilt token map still routes when schema metadata is disabled (rebuilt lazily).""" + metadata, tokens, hosts = self._build_metadata() + metadata._update_replication_strategies({ + "ks": ReplicationStrategy.create( + "org.apache.cassandra.locator.SimpleStrategy", {"replication_factor": "1"})}) + + metadata.rebuild_token_map( + "org.apache.cassandra.dht.Murmur3Partitioner", + {host: [str(token.value)] for host, token in zip(hosts, tokens)}) + + assert set(metadata.token_map.get_replicas("ks", tokens[0])) == {hosts[0]} + + def test_update_replication_strategies_change_drops_tablets_and_rebuilds(self): + """A changed replication strategy drops tablets and rebuilds the token map.""" + metadata, tokens, hosts = self._build_metadata() + metadata._update_replication_strategies({ + "ks": ReplicationStrategy.create( + "org.apache.cassandra.locator.SimpleStrategy", {"replication_factor": "1"})}) + metadata._tablets.add_tablet("ks", "tb", Tablet(0, 100, [("host1", 0)])) + + metadata._update_replication_strategies({ + "ks": ReplicationStrategy.create( + "org.apache.cassandra.locator.SimpleStrategy", {"replication_factor": "2"})}) + + assert metadata._tablets.table_has_tablets("ks", "tb") is False + assert set(metadata.token_map.get_replicas("ks", tokens[0])) == {hosts[0], hosts[1]} + + def test_update_replication_strategies_skips_unchanged(self): + """An unchanged replication strategy does not rebuild the token map or drop tablets.""" + metadata, _, _ = self._build_metadata() + metadata._update_replication_strategies({ + "ks": ReplicationStrategy.create( + "org.apache.cassandra.locator.SimpleStrategy", {"replication_factor": "1"})}) + metadata._tablets.add_tablet("ks", "tb", Tablet(0, 100, [("host1", 0)])) + before = dict(metadata.token_map.tokens_to_hosts_by_ks["ks"]) + + with patch.object(TokenMap, "rebuild_keyspace") as rebuild_keyspace: + metadata._update_replication_strategies({ + "ks": ReplicationStrategy.create( + "org.apache.cassandra.locator.SimpleStrategy", {"replication_factor": "1"})}) + + rebuild_keyspace.assert_not_called() + assert metadata._tablets.table_has_tablets("ks", "tb") is True + assert metadata.token_map.tokens_to_hosts_by_ks["ks"] == before + + def test_update_replication_strategies_change_syncs_keyspace_metadata(self): + """A strategy change updates stale keyspace metadata so the token map uses the fresh strategy.""" + metadata, tokens, hosts = self._build_metadata() + keyspace = KeyspaceMetadata("ks", True, "NetworkTopologyStrategy", {"dc1": "1"}) + metadata.keyspaces["ks"] = keyspace + metadata._keyspace_added("ks") + assert set(metadata.token_map.get_replicas("ks", tokens[0])) == {hosts[0]} + + metadata._update_replication_strategies({ + "ks": ReplicationStrategy.create( + "org.apache.cassandra.locator.NetworkTopologyStrategy", {"dc1": "2"})}) + + assert keyspace.replication_strategy is metadata._keyspace_replication_strategies["ks"] + assert set(metadata.token_map.get_replicas("ks", tokens[0])) == {hosts[0], hosts[1]} + + def test_update_replication_strategies_removed_syncs_keyspace_metadata(self): + """A dropped keyspace is removed from keyspace metadata when it was previously enabled.""" + metadata, _, _ = self._build_metadata() + metadata.keyspaces["ks"] = KeyspaceMetadata("ks", True, "NetworkTopologyStrategy", {"dc1": "1"}) + metadata._keyspace_added("ks") + + metadata._update_replication_strategies({}) + + assert metadata.keyspaces == {} + assert metadata._keyspace_replication_strategies == {} + assert metadata.token_map.tokens_to_hosts_by_ks == {} + + class Murmur3TokensTest(unittest.TestCase): def test_murmur3_init(self): @@ -641,10 +780,12 @@ class ScyllaKeyspaceConsistencyParsingTest(unittest.TestCase): """ def _parser_with_rows(self, rows): - # Build the parser without a connection and drive only the aggregation + # Build the parser without a connection and drive the real aggregation # step; _query_all's batching is exercised by the integration tests. - parser = SchemaParserV3.__new__(SchemaParserV3) + parser = SchemaParserV3(None, 1.0, 100, None) + parser.views_result = [] # populated by _query_all, not __init__ parser.scylla_keyspaces_result = rows + parser._aggregate_results() return parser def test_consistency_modes_are_mapped_from_rows(self): @@ -657,8 +798,7 @@ def test_consistency_modes_are_mapped_from_rows(self): {'keyspace_name': 'e', 'consistency': 'eventual'}, {'keyspace_name': 'n', 'consistency': None}, ]) - modes = {row["keyspace_name"]: _consistency_mode_from_string(row.get("consistency")) - for row in parser.scylla_keyspaces_result} + modes = parser.keyspace_consistency_modes assert modes['g'] == _ConsistencyMode.GLOBAL assert modes['l'] == _ConsistencyMode.LOCAL assert modes['e'] == _ConsistencyMode.EVENTUAL @@ -667,8 +807,9 @@ def test_consistency_modes_are_mapped_from_rows(self): def test_keyspace_absent_from_the_map_is_eventual(self): # Covers the whole-cluster fallbacks too: no rows is what a skipped read # (no TABLETS_ROUTING_V2) and a missing table/column both produce. - parser = SchemaParserV3.__new__(SchemaParserV3) - parser.keyspace_consistency_modes = {'g': _ConsistencyMode.GLOBAL} + parser = self._parser_with_rows([ + {'keyspace_name': 'g', 'consistency': 'global'}, + ]) assert parser.keyspace_consistency_modes.get( 'absent', _ConsistencyMode.EVENTUAL) == _ConsistencyMode.EVENTUAL