(Improvement) immediate DDL schema agreement (per-host system.local) + lower-memory metadata schema parsing

cassandra/cluster.py

@@ -173,6 +173,24 @@ class SchemaAgreementScope(str, Enum):
     CLUSTER = 'cluster'
 
 
+# Schema agreement is usually reached within a few milliseconds of a DDL
+# returning (both system.local and the inter-node schema apply), but the version
+# signal the peers view exposes is propagated by gossip, which can lag by up to a
+# gossip round. A fixed retry interval therefore either over-waits on a slow
+# change or adds up to a full interval of latency on a fast one. Start polling
+# quickly and back off to the historic 200ms cap instead.
+_SCHEMA_AGREEMENT_INITIAL_DELAY = 0.01
+_SCHEMA_AGREEMENT_MAX_DELAY = 0.2
+
+
+def _backoff_delays():
+    """Yield successive schema agreement poll delays, doubling up to the cap."""
+    delay = _SCHEMA_AGREEMENT_INITIAL_DELAY
+    while True:
+        yield delay
+        delay = min(delay * 2, _SCHEMA_AGREEMENT_MAX_DELAY)
+
+
 def _future_completed(future):
     """ Helper for run_in_executor() """
     exc = future.exception()
@@ -3514,6 +3532,7 @@ def wait_for_schema_agreement(self, wait_time: Optional[float] = None,
 
         deadline = time.time() + total_timeout
         schema_mismatches = None
+        delays = _backoff_delays()
         scope_label = 'local rack' if scope is SchemaAgreementScope.RACK else (
             'local datacenter' if scope is SchemaAgreementScope.DC else 'cluster')
 
@@ -3525,7 +3544,7 @@ def wait_for_schema_agreement(self, wait_time: Optional[float] = None,
             log.debug("[session] Connected hosts in the %s still disagree on schema, trying again", scope_label)
             remaining = deadline - time.time()
             if remaining > 0:
-                time.sleep(min(0.2, remaining))
+                time.sleep(min(next(delays), remaining))
 
         log.warning("[session] Connected hosts in the %s are reporting a schema disagreement: %s",
                     scope_label, schema_mismatches)
@@ -3603,7 +3622,7 @@ def _get_schema_agreement_hosts(self, scope: SchemaAgreementScope) -> Tuple[Host
     def _query_local_schema_version(self, host: Host, query: str, deadline: float) -> Future:
         remaining = max(0.0, deadline - time.time())
         try:
-            response_future = self.execute_async(
+            response_future = self._send_schema_version_query(
                 query,
                 timeout=self._schema_agreement_query_timeout(remaining),
                 host=host,
@@ -3615,7 +3634,7 @@ def _query_local_schema_version(self, host: Host, query: str, deadline: float) -
             log.debug("[session] Error querying schema version from %s: %s", host, exc)
             raise
 
-        # execute_async returns cassandra.cluster.ResponseFuture, which does not have bulk waiting logic for it.
+        # _send_schema_version_query returns cassandra.cluster.ResponseFuture, which does not have bulk waiting logic for it.
         # That is why _query_local_schema_version returns concurrent.futures.Future
         #  so that schema agreement logic could use concurrent.futures.wait_futures to wait on them.
         # schema_version_future is an adapter between cassandra.cluster.ResponseFuture and concurrent.futures.Future
@@ -3643,6 +3662,23 @@ def _set_exception(exc, result_future=schema_version_future):
 
         return schema_version_future
 
+    def _send_schema_version_query(self, query: str, timeout: float, host: Host) -> "ResponseFuture":
+        """Send an internal schema version query with a pinned row factory.
+
+        The session's configured row factory is deliberately not used: cqlengine,
+        for example, installs ``dict_factory``, which does not expose columns as
+        attributes and would make the version column unreadable.
+        """
+        response_future = self._create_response_future(
+            query, parameters=None, trace=False, custom_payload=None,
+            timeout=timeout, host=host,
+        )
+        response_future.row_factory = named_tuple_factory
+        response_future._protocol_handler = self.client_protocol_handler
+        self._on_request(response_future)
+        response_future.send_request()
+        return response_future
+
     def _schema_agreement_query_timeout(self, remaining: float) -> float:
         control_timeout = self.cluster.control_connection._timeout
         if control_timeout is None:
@@ -4046,6 +4082,8 @@ def shutdown(self):
     def refresh_schema(self, force=False, **kwargs):
         try:
             if self._connection:
+                if kwargs.get('session') is None:
+                    kwargs['session'] = next(iter(self._cluster.sessions), None)
                 return self._refresh_schema(self._connection, force=force, **kwargs)
         except ReferenceError:
             pass  # our weak reference to the Cluster is no good
@@ -4054,13 +4092,15 @@ def refresh_schema(self, force=False, **kwargs):
             self._signal_error()
         return False
 
-    def _refresh_schema(self, connection, preloaded_results=None, schema_agreement_wait=None, force=False, **kwargs):
+    def _refresh_schema(self, connection, preloaded_results=None, schema_agreement_wait=None, force=False,
+                        session=None, **kwargs):
         if self._cluster.is_shutdown:
             return False
 
         agreed = self._wait_for_schema_agreement(connection=connection,
                                                 preloaded_results=preloaded_results,
-                                                wait_time=schema_agreement_wait)
+                                                wait_time=schema_agreement_wait,
+                                                session=session)
 
         if not self._schema_meta_enabled and not force:
             log.debug("[control connection] Skipping schema refresh because schema metadata is disabled")
@@ -4374,7 +4414,8 @@ def wait_for_schema_agreement(self, connection=None, preloaded_results=None, wai
                                                preloaded_results=preloaded_results,
                                                wait_time=wait_time)
 
-    def _wait_for_schema_agreement(self, connection=None, preloaded_results=None, wait_time=None):
+    def _wait_for_schema_agreement(self, connection=None, preloaded_results=None, wait_time=None,
+                                   session=None):
         total_timeout = wait_time if wait_time is not None else self._cluster.max_schema_agreement_wait
         if total_timeout <= 0:
             return True
@@ -4399,11 +4440,20 @@ def _wait_for_schema_agreement(self, connection=None, preloaded_results=None, wa
                 if schema_mismatches is None:
                     return True
 
+            if session is not None and self._connected_host_agreement_available(session):
+                # system.local reflects an applied schema change immediately, while
+                # system.peers only learns the version on the next gossip round. When
+                # a session is available, ask the connected hosts directly so a DDL
+                # does not block on gossip propagation.
+                log.debug("[control connection] Waiting for schema agreement via connected hosts")
+                return session.wait_for_schema_agreement(wait_time=total_timeout)
+
             log.debug("[control connection] Waiting for schema agreement")
             start = self._time.time()
             elapsed = 0
             cl = ConsistencyLevel.ONE
             schema_mismatches = None
+            delays = _backoff_delays()
             select_peers_query = self._get_peers_query(self.PeersQueryType.PEERS_SCHEMA, connection)
 
             while elapsed < total_timeout:
@@ -4433,21 +4483,38 @@ def _wait_for_schema_agreement(self, connection=None, preloaded_results=None, wa
                     return True
 
                 log.debug("[control connection] Schemas mismatched, trying again")
-                self._time.sleep(0.2)
+                self._time.sleep(min(next(delays), max(0.0, total_timeout - elapsed)))
                 elapsed = self._time.time() - start
 
             log.warning("Node %s is reporting a schema disagreement: %s",
                         connection.endpoint, schema_mismatches)
             return False
 
+    @staticmethod
+    def _connected_host_agreement_available(session):
+        """Whether the session has at least one connected host to check directly.
+
+        Returns False (so the peers-based loop is used) when the session has no
+        connected pools, e.g. during control-connection startup.
+        """
+        try:
+            return bool(session._get_schema_agreement_hosts(SchemaAgreementScope.CLUSTER))
+        except Exception:
+            log.debug("[control connection] Cannot use connected hosts for schema agreement, "
+                      "falling back to peers", exc_info=True)
+            return False
+
     def _get_schema_mismatches(self, peers_result, local_result, local_address):
         peers_result = dict_factory(peers_result.column_names, peers_result.parsed_rows)
 
-        versions = defaultdict(set)
+        local_row = None
         if local_result.parsed_rows:
             local_row = dict_factory(local_result.column_names, local_result.parsed_rows)[0]
-            if local_row.get("schema_version"):
-                versions[local_row.get("schema_version")].add(local_address)
+        local_version = local_row.get("schema_version") if local_row else None
+
+        versions = defaultdict(set)
+        if local_version:
+            versions[local_version].add(local_address)
 
         for row in peers_result:
             schema_ver = row.get('schema_version')
@@ -4657,7 +4724,8 @@ def refresh_schema_and_set_result(control_conn, response_future, connection, **k
     try:
         log.debug("Refreshing schema in response to schema change. "
                   "%s", kwargs)
-        response_future.is_schema_agreed = control_conn._refresh_schema(connection, **kwargs)
+        response_future.is_schema_agreed = control_conn._refresh_schema(
+            connection, session=response_future.session, **kwargs)
     except Exception:
         log.exception("Exception refreshing schema in response to schema change:")
         response_future.session.submit(control_conn.refresh_schema, **kwargs)