Use tablets in token and shard awareness

Add mechanism to parse system.tablets periodically. In TokenAwarePolicy
check if keyspace uses tablets if so try to use them to find replicas.

Make shard awareness work when using tablets.

Everything is wrapped in experimental setting, because tablets are still
experimental in ScyllaDB and changes in the tablets format are possible.

Author: sylwiaszunejko

cassandra/cluster.py

@@ -41,7 +41,7 @@
 import weakref
 from weakref import WeakValueDictionary
 
-from cassandra import (ConsistencyLevel, AuthenticationFailed,
+from cassandra import (ConsistencyLevel, AuthenticationFailed, InvalidRequest,
                        OperationTimedOut, UnsupportedOperation,
                        SchemaTargetType, DriverException, ProtocolVersion,
                        UnresolvableContactPoints)
@@ -79,6 +79,7 @@
                              named_tuple_factory, dict_factory, tuple_factory, FETCH_SIZE_UNSET,
                              HostTargetingStatement)
 from cassandra.marshal import int64_pack
+from cassandra.tablets import Tablet, Tablets
 from cassandra.timestamps import MonotonicTimestampGenerator
 from cassandra.compat import Mapping
 from cassandra.util import _resolve_contact_points_to_string_map, Version
@@ -938,6 +939,17 @@ def default_retry_policy(self, policy):
     establish connection pools. This can cause a rush of connections and queries if not mitigated with this factor.
     """
 
+    experimental_tablet_refresh_time = 60
+    """
+    The time at which the system.tablets table is polled periodically.
+    This field is experimental and may be changed / removed in the future.
+    """
+
+    experimental_tablet_feature_enabled = False
+    """
+    If true tablet experimental feature is enabled, its interface and use may change.
+    """
+
     prepare_on_all_hosts = True
     """
     Specifies whether statements should be prepared on all hosts, or just one.
@@ -1124,6 +1136,8 @@ def __init__(self,
                  schema_metadata_page_size=1000,
                  address_translator=None,
                  status_event_refresh_window=2,
+                 experimental_tablet_refresh_time=60, #  This field is experimental and may be changed / removed in the future
+                 experimental_tablet_feature_enabled=False,  #  This field is experimental and may be changed / removed in the future
                  prepare_on_all_hosts=True,
                  reprepare_on_up=True,
                  execution_profiles=None,
@@ -1364,6 +1378,8 @@ def __init__(self,
         self.schema_event_refresh_window = schema_event_refresh_window
         self.topology_event_refresh_window = topology_event_refresh_window
         self.status_event_refresh_window = status_event_refresh_window
+        self.experimental_tablet_refresh_time = experimental_tablet_refresh_time
+        self.experimental_tablet_feature_enabled = experimental_tablet_feature_enabled
         self.connect_timeout = connect_timeout
         self.prepare_on_all_hosts = prepare_on_all_hosts
         self.reprepare_on_up = reprepare_on_up
@@ -1416,7 +1432,7 @@ def __init__(self,
         self.control_connection = ControlConnection(
             self, self.control_connection_timeout,
             self.schema_event_refresh_window, self.topology_event_refresh_window,
-            self.status_event_refresh_window,
+            self.status_event_refresh_window, self.experimental_tablet_refresh_time,
             schema_metadata_enabled, token_metadata_enabled,
             schema_meta_page_size=schema_metadata_page_size)
 
@@ -2381,6 +2397,9 @@ def add_prepared(self, query_id, prepared_statement):
         with self._prepared_statement_lock:
             self._prepared_statements[query_id] = prepared_statement
 
+    # Experimental, this interface and use may change
+    def check_tablets_enabled(self):
+        return self.experimental_tablet_feature_enabled
 
 class Session(object):
     """
@@ -3513,6 +3532,8 @@ class ControlConnection(object):
     _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"
 
+    _SELECT_TABLETS = "SELECT * FROM system.tablets"
+
     _MINIMUM_NATIVE_ADDRESS_DSE_VERSION = Version("6.0.0")
 
     class PeersQueryType(object):
@@ -3527,6 +3548,7 @@ class PeersQueryType(object):
     _schema_event_refresh_window = None
     _topology_event_refresh_window = None
     _status_event_refresh_window = None
+    _tablet_refresh_time = 60
 
     _schema_meta_enabled = True
     _token_meta_enabled = True
@@ -3541,6 +3563,7 @@ def __init__(self, cluster, timeout,
                  schema_event_refresh_window,
                  topology_event_refresh_window,
                  status_event_refresh_window,
+                 tablet_refresh_time=60,
                  schema_meta_enabled=True,
                  token_meta_enabled=True,
                  schema_meta_page_size=1000):
@@ -3553,6 +3576,7 @@ def __init__(self, cluster, timeout,
         self._schema_event_refresh_window = schema_event_refresh_window
         self._topology_event_refresh_window = topology_event_refresh_window
         self._status_event_refresh_window = status_event_refresh_window
+        self._tablet_refresh_time = tablet_refresh_time
         self._schema_meta_enabled = schema_meta_enabled
         self._token_meta_enabled = token_meta_enabled
         self._schema_meta_page_size = schema_meta_page_size
@@ -3658,6 +3682,18 @@ def _try_connect(self, host):
                 "SCHEMA_CHANGE": partial(_watch_callback, self_weakref, '_handle_schema_change')
             }, register_timeout=self._timeout)
 
+            if self._cluster.check_tablets_enabled():
+                try:
+                    self._refresh_tablets(connection)
+                except InvalidRequest:
+                    self._cluster.experimental_tablet_feature_enabled = False
+                    # The information about tablets is passed to load_balancing_policy during `populate` that happens
+                    # before this check, so if there is an error during `_refresh_tablets` we need to change the field
+                    # in the cluster, but also propagate this information to the load_balancing_policy using `populate`
+                    self._cluster.load_balancing_policy.populate(self._cluster, self._cluster.metadata.all_hosts())
+                else:
+                    self._cluster.scheduler.schedule_unique(self._tablet_refresh_time, self._refresh_tablets_periodically, connection)
+
             sel_peers = self._get_peers_query(self.PeersQueryType.PEERS, connection)
             sel_local = self._SELECT_LOCAL if self._token_meta_enabled else self._SELECT_LOCAL_NO_TOKENS
             peers_query = QueryMessage(query=sel_peers, consistency_level=ConsistencyLevel.ONE)
@@ -3964,6 +4000,40 @@ def _refresh_node_list_and_token_map(self, connection, preloaded_results=None,
             log.debug("[control connection] Rebuilding token map due to topology changes")
             self._cluster.metadata.rebuild_token_map(partitioner, token_map)
 
+    def _refresh_tablets_periodically(self, connection):
+        try:
+            self._refresh_tablets(connection)
+        except ReferenceError:
+            return
+        except Exception:
+            log.debug("[control connection] Error refreshing tablets", exc_info=True)
+            self._signal_error()
+        self._cluster.scheduler.schedule_unique(self._tablet_refresh_time, self._refresh_tablets, connection)
+    
+    # Experimental, this interface and use may change
+    def _refresh_tablets(self, connection):
+        sel_tablets = self._SELECT_TABLETS
+        tablets_query = QueryMessage(query=sel_tablets, consistency_level=ConsistencyLevel.ONE)
+        tablets_result = connection.wait_for_response(
+            tablets_query, timeout=self._timeout)
+        tablets_result = dict_factory(tablets_result.column_names, tablets_result.parsed_rows)
+
+        tablets = {}
+        for row in tablets_result:
+            tablet = Tablet.from_row(row)
+            if tablet is None:
+                log.warning(
+                    "Found an invalid row for tablet (%s). Ignoring tablet." %
+                    _NodeInfo.get_broadcast_rpc_address(row))
+                continue
+
+            keyspace_name = row.get("keyspace_name")
+            table_name = row.get("table_name")
+
+            tablets.setdefault((keyspace_name, table_name), []).append(tablet)
+        
+        self._cluster.metadata._tablets = Tablets(tablets)
+
     @staticmethod
     def _is_valid_peer(row):
         return bool(_NodeInfo.get_broadcast_rpc_address(row) and row.get("host_id") and
@@ -4571,7 +4641,10 @@ def _query(self, host, message=None, cb=None):
         connection = None
         try:
             # TODO get connectTimeout from cluster settings
-            connection, request_id = pool.borrow_connection(timeout=2.0, routing_key=self.query.routing_key if self.query else None)
+            if self.query:
+                connection, request_id = pool.borrow_connection(timeout=2.0, routing_key=self.query.routing_key, keyspace=self.query.keyspace, table=self.query.table)
+            else:
+                connection, request_id = pool.borrow_connection(timeout=2.0)
             self._connection = connection
             result_meta = self.prepared_statement.result_metadata if self.prepared_statement else []
 

cassandra/metadata.py

@@ -126,6 +126,7 @@ def __init__(self):
         self._hosts = {}
         self._host_id_by_endpoint = {}
         self._hosts_lock = RLock()
+        self._tablets = None
 
     def export_schema_as_string(self):
         """

cassandra/policies.py

@@ -335,6 +335,7 @@ class TokenAwarePolicy(LoadBalancingPolicy):
 
     _child_policy = None
     _cluster_metadata = None
+    _experimental_tablets_enabled = False
     shuffle_replicas = False
     """
     Yield local replicas in a random order.
@@ -346,6 +347,7 @@ def __init__(self, child_policy, shuffle_replicas=False):
 
     def populate(self, cluster, hosts):
         self._cluster_metadata = cluster.metadata
+        self._experimental_tablets_enabled = cluster.check_tablets_enabled()
         self._child_policy.populate(cluster, hosts)
 
     def check_supported(self):
@@ -376,7 +378,19 @@ def make_query_plan(self, working_keyspace=None, query=None):
                 for host in child.make_query_plan(keyspace, query):
                     yield host
             else:
-                replicas = self._cluster_metadata.get_replicas(keyspace, routing_key)
+                replicas = []
+                if self._experimental_tablets_enabled:
+                    tablet = self._cluster_metadata._tablets.get_tablet_for_key(keyspace, query.table, self._cluster_metadata.token_map.token_class.from_key(routing_key))
+
+                    if tablet is not None:
+                        replicas_mapped = set(map(lambda r: r[0], tablet.replicas))
+                        child_plan = child.make_query_plan(keyspace, query)
+
+                        replicas = [host for host in child_plan if host.host_id in replicas_mapped]
+
+                if replicas == []:
+                    replicas = self._cluster_metadata.get_replicas(keyspace, routing_key)
+
                 if self.shuffle_replicas:
                     shuffle(replicas)
                 for replica in replicas:

cassandra/pool.py

@@ -439,7 +439,7 @@ def __init__(self, host, host_distance, session):
 
         log.debug("Finished initializing connection for host %s", self.host)
 
-    def _get_connection_for_routing_key(self, routing_key=None):
+    def _get_connection_for_routing_key(self, routing_key=None, keyspace=None, table=None):
         if self.is_shutdown:
             raise ConnectionException(
                 "Pool for %s is shutdown" % (self.host,), self.host)
@@ -450,7 +450,19 @@ def _get_connection_for_routing_key(self, routing_key=None):
         shard_id = None
         if not self._session.cluster.shard_aware_options.disable and self.host.sharding_info and routing_key:
             t = self._session.cluster.metadata.token_map.token_class.from_key(routing_key)
-            shard_id = self.host.sharding_info.shard_id_from_token(t.value)
+            
+            shard_id = None
+            if self._session.cluster.check_tablets_enabled() and table is not None:
+                if keyspace is None:
+                    keyspace = self._keyspace
+
+                tablet = self._session.cluster._load_balancing_policy._cluster_metadata._tablets.get_tablet_for_key(keyspace, table, t)
+
+                if tablet is not None:
+                    shard_id = tablet.replicas[0][1]
+
+            if shard_id is None:
+                shard_id = self.host.sharding_info.shard_id_from_token(t.value)
 
         conn = self._connections.get(shard_id)
 
@@ -496,15 +508,15 @@ def _get_connection_for_routing_key(self, routing_key=None):
             return random.choice(active_connections)
         return random.choice(list(self._connections.values()))
 
-    def borrow_connection(self, timeout, routing_key=None):
-        conn = self._get_connection_for_routing_key(routing_key)
+    def borrow_connection(self, timeout, routing_key=None, keyspace=None, table=None):
+        conn = self._get_connection_for_routing_key(routing_key, keyspace, table)
         start = time.time()
         remaining = timeout
         last_retry = False
         while True:
             if conn.is_closed:
                 # The connection might have been closed in the meantime - if so, try again
-                conn = self._get_connection_for_routing_key(routing_key)
+                conn = self._get_connection_for_routing_key(routing_key, keyspace, table)
             with conn.lock:
                 if (not conn.is_closed or last_retry) and conn.in_flight < conn.max_request_id:
                     # On last retry we ignore connection status, since it is better to return closed connection than

cassandra/query.py

@@ -253,6 +253,13 @@ class Statement(object):
     .. versionadded:: 2.1.3
     """
 
+    table = None
+    """
+    The string name of the table this query acts on. This is used when the tablet
+    experimental feature is enabled and in the same time :class`~.TokenAwarePolicy`
+    is configured in the profile load balancing policy.
+    """
+
     custom_payload = None
     """
     :ref:`custom_payload` to be passed to the server.
@@ -272,7 +279,7 @@ class Statement(object):
 
     def __init__(self, retry_policy=None, consistency_level=None, routing_key=None,
                  serial_consistency_level=None, fetch_size=FETCH_SIZE_UNSET, keyspace=None, custom_payload=None,
-                 is_idempotent=False):
+                 is_idempotent=False, table=None):
         if retry_policy and not hasattr(retry_policy, 'on_read_timeout'):  # just checking one method to detect positional parameter errors
             raise ValueError('retry_policy should implement cassandra.policies.RetryPolicy')
         if retry_policy is not None:
@@ -286,6 +293,8 @@ def __init__(self, retry_policy=None, consistency_level=None, routing_key=None,
             self.fetch_size = fetch_size
         if keyspace is not None:
             self.keyspace = keyspace
+        if table is not None:
+            self.table = table
         if custom_payload is not None:
             self.custom_payload = custom_payload
         self.is_idempotent = is_idempotent
@@ -548,6 +557,7 @@ def __init__(self, prepared_statement, retry_policy=None, consistency_level=None
         meta = prepared_statement.column_metadata
         if meta:
             self.keyspace = meta[0].keyspace_name
+            self.table = meta[0].table_name
 
         Statement.__init__(self, retry_policy, consistency_level, routing_key,
                            serial_consistency_level, fetch_size, keyspace, custom_payload,

cassandra/tablets.py

@@ -0,0 +1,80 @@
+# Experimental, this interface and use may change
+class Tablet(object):
+    """
+    Represents a single ScyllaDB tablet.
+    """
+    last_token = 0
+    new_replicas = None
+    replicas = None
+    stage = ""
+
+    def __init__(self, last_token = 0, new_replicas = None, replicas = None, stage = ""):
+        self.last_token = last_token
+        self.new_replicas = new_replicas
+        self.replicas = replicas
+        self.stage = stage
+
+    # Experimental, this interface and use may change
+    @staticmethod
+    def _is_valid_tablet(row):
+        return row.get("replicas") is not None and len(row.get("replicas")) != 0 and row.get("table_name") is not None and row.get("keyspace_name") is not None
+
+    # Experimental, this interface and use may change
+    @staticmethod
+    def from_row(row):
+        if Tablet._is_valid_tablet(row):
+            tablet = Tablet(row.get("last_token"), row.get("new_replicas"), row.get("replicas"), row.get("stage"))
+            return tablet
+        return None
+
+# Experimental, this interface and use may change
+class Tablets(object):
+    _tablets = {}
+
+    def __init__(self, tablets):
+        self._tablets = tablets
+    
+    # Experimental, this interface and use may change
+    def get_tablet_for_key(self, keyspace, table, t):
+        tablet = self._tablets.get((keyspace, table), [])
+        if tablet == []:
+            return None
+        
+        id = bisect_left(tablet, t.value, key = lambda tablet: tablet.last_token)
+        return tablet[id]
+
+# bisect.bisect_left implementation from Python 3.11, needed untill support for
+# Python < 3.10 is dropped, it is needed to use `key` to extract last_token from
+# Tablet list - better solution performance-wise than materialize list of last_tokens
+def bisect_left(a, x, lo=0, hi=None, *, key=None):
+    """Return the index where to insert item x in list a, assuming a is sorted.
+
+    The return value i is such that all e in a[:i] have e < x, and all e in
+    a[i:] have e >= x.  So if x already appears in the list, a.insert(i, x) will
+    insert just before the leftmost x already there.
+
+    Optional args lo (default 0) and hi (default len(a)) bound the
+    slice of a to be searched.
+    """
+
+    if lo < 0:
+        raise ValueError('lo must be non-negative')
+    if hi is None:
+        hi = len(a)
+    # Note, the comparison uses "<" to match the
+    # __lt__() logic in list.sort() and in heapq.
+    if key is None:
+        while lo < hi:
+            mid = (lo + hi) // 2
+            if a[mid] < x:
+                lo = mid + 1
+            else:
+                hi = mid
+    else:
+        while lo < hi:
+            mid = (lo + hi) // 2
+            if key(a[mid]) < x:
+                lo = mid + 1
+            else:
+                hi = mid
+    return lo