improvement: maintain token-aware routing even when schema metadata refresh is disabled

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

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,

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,17 +1885,26 @@ 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
                 self.tokens_to_hosts_by_ks[keyspace] = {}
                 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)

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 []