(improvement) additional improvements to HostFilter and default policy.

DefaultLoadBalancingPolicy: add make_query_plan_with_exclusion
- forward exclusions to child policy
- preserve target_host preference while skipping excluded hosts

HostFilterPolicy: add make_query_plan_with_exclusion
- forward exclusions to child policy
- filter excluded hosts via predicate in exclusion-aware plans
- add isinstance guard to avoid redundant set() conversion when
  excluded is already a set

Also add isinstance guard in DefaultLoadBalancingPolicy.make_query_plan_with_exclusion
for consistency.

Benchmark (100K queries, 45-node/5-DC topology, Python 3.14, median of 5 runs):
Policy                    |  Kops/s | vs master | delta | Mem KB
-----------------------------------------------------------------
DCAware                   |     206 |      +94% |       |    1.5
RackAware                 |     172 |     +153% |       |    2.0
TokenAware(DCAware)       |      97 |     +439% |       |    1.7
TokenAware(RackAware)     |      89 |     +424% |       |    2.2
Default(DCAware)          |     132 |      +45% |   -4% |    1.6
HostFilter(DCAware)       |      54 |       +2% |  -17% |    1.7

This commit primarily adds exclusion-aware interfaces to Default and
HostFilter policies. The throughput deltas are within noise; the main
benefit is architectural — enabling these policies to participate in
the exclusion protocol for composition with TokenAware.

Signed-off-by: Yaniv Kaul <yaniv.kaul@scylladb.com>

Author: mykaul

cassandra/policies.py

@@ -924,6 +924,18 @@ def make_query_plan(self, working_keyspace=None, query=None):
             if self.predicate(host):
                 yield host
 
+    def make_query_plan_with_exclusion(
+        self, working_keyspace=None, query=None, excluded=()
+    ):
+        if excluded and not isinstance(excluded, set):
+            excluded = set(excluded)
+        child_qp = self._child_policy.make_query_plan_with_exclusion(
+            working_keyspace=working_keyspace, query=query, excluded=excluded
+        )
+        for host in child_qp:
+            if self.predicate(host):
+                yield host
+
     def check_supported(self):
         return self._child_policy.check_supported()
 
@@ -1641,6 +1653,29 @@ def make_query_plan(self, working_keyspace=None, query=None):
             for h in child.make_query_plan(keyspace, query):
                 yield h
 
+    def make_query_plan_with_exclusion(
+        self, working_keyspace=None, query=None, excluded=()
+    ):
+        if query and query.keyspace:
+            keyspace = query.keyspace
+        else:
+            keyspace = working_keyspace
+
+        addr = getattr(query, "target_host", None) if query else None
+        target_host = self._cluster_metadata.get_host(addr)
+
+        if excluded and not isinstance(excluded, set):
+            excluded = set(excluded)
+
+        child = self._child_policy
+        if target_host and target_host.is_up and target_host not in excluded:
+            yield target_host
+            for h in child.make_query_plan_with_exclusion(keyspace, query, excluded):
+                if h != target_host:
+                    yield h
+        else:
+            yield from child.make_query_plan_with_exclusion(keyspace, query, excluded)
+
 
 # TODO for backward compatibility, remove in next major
 class DSELoadBalancingPolicy(DefaultLoadBalancingPolicy):