CLIENT-5011 server side idx selection

aerospike_sdk/ael/server_filter.py

@@ -0,0 +1,45 @@
+# Copyright 2025-2026 Aerospike, Inc.
+#
+# Portions may be licensed to Aerospike, Inc. under one or more contributor
+# license agreements WHICH ARE COMPATIBLE WITH THE APACHE LICENSE, VERSION 2.0.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may not
+# use this file except in compliance with the License. You may obtain a copy of
+# the License at http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations under
+# the License.
+
+"""Pick client-parsed vs server-compiled filter wire form for AEL strings."""
+
+from __future__ import annotations
+
+from aerospike_async import FilterExpression
+
+from aerospike_sdk.ael.parser import parse_ael
+
+# Resolved once at import — the PAC factory does not change at runtime.
+_SERVER_COMPILED_FACTORY = getattr(
+    FilterExpression, "from_server_compiled_ael", None,
+)
+_PAC_EXPOSES_SERVER_COMPILED: bool = callable(_SERVER_COMPILED_FACTORY)
+
+
+def filter_expression_from_ael_string(
+    ael: str,
+    *,
+    supports_server_compiled_ael: bool,
+) -> FilterExpression:
+    """Return a ``FilterExpression`` for *ael*, using server-compiled wire when allowed.
+
+    When ``supports_server_compiled_ael`` is true and PAC exposes the factory,
+    returns field **43** MessagePack ``[128, "<utf-8 ael>"]`` via
+    :meth:`~aerospike_async.FilterExpression.from_server_compiled_ael`.
+    Otherwise parses on the client via :func:`~aerospike_sdk.ael.parser.parse_ael`.
+    """
+    if supports_server_compiled_ael and _PAC_EXPOSES_SERVER_COMPILED:
+        return _SERVER_COMPILED_FACTORY(ael)  # type: ignore[misc]
+    return parse_ael(ael)

aerospike_sdk/aio/background.py

@@ -40,7 +40,7 @@
     reject_unsupported_background_write_ops,
 )
 from aerospike_sdk.dataset import DataSet
-from aerospike_sdk.ael.parser import parse_ael
+from aerospike_sdk.ael.server_filter import filter_expression_from_ael_string
 from aerospike_sdk.exceptions import _convert_pac_exception
 from aerospike_sdk.operations_shared import _seconds_from_timedelta, _seconds_until
 
@@ -201,6 +201,9 @@ def __init__(
         self._records_per_second: Optional[int] = None
         self._durable_delete_command_default: Optional[bool] = None
         self._durable_delete_override: Optional[bool] = None
+        self._supports_server_compiled_ael = bool(
+            getattr(session.client, "_cached_supports_server_compiled_ael", False),
+        )
 
     def default_with_durable_delete(self) -> BackgroundOperationBuilder:
         """Prefer durable deletes when resolving policy defaults (SC namespaces)."""
@@ -246,7 +249,10 @@ def where(
                 "use one narrowing mechanism.",
             )
         if isinstance(expression, str):
-            self._filter_expression = parse_ael(expression)
+            self._filter_expression = filter_expression_from_ael_string(
+                expression,
+                supports_server_compiled_ael=self._supports_server_compiled_ael,
+            )
         else:
             self._filter_expression = expression
         return self
@@ -542,6 +548,9 @@ def __init__(
         self._records_per_second: Optional[int] = None
         self._durable_delete_command_default: Optional[bool] = None
         self._durable_delete_override: Optional[bool] = None
+        self._supports_server_compiled_ael = bool(
+            getattr(session.client, "_cached_supports_server_compiled_ael", False),
+        )
 
     def default_with_durable_delete(self) -> BackgroundUdfBuilder:
         """Prefer durable deletes when resolving policy defaults (SC namespaces)."""
@@ -587,7 +596,10 @@ def where(
     ) -> BackgroundUdfBuilder:
         """Optional predicate limiting which records invoke the UDF."""
         if isinstance(expression, str):
-            self._filter_expression = parse_ael(expression)
+            self._filter_expression = filter_expression_from_ael_string(
+                expression,
+                supports_server_compiled_ael=self._supports_server_compiled_ael,
+            )
         else:
             self._filter_expression = expression
         return self

aerospike_sdk/aio/client.py

@@ -44,7 +44,20 @@
 from aerospike_sdk.policy.behavior_settings import Mode
 from aerospike_sdk.policy.sdk_config_loader import fill_hard_defaults
 from aerospike_sdk.policy.system_settings import SystemSettings
+from aerospike_sdk.feature_gates import (
+    PSDK_ENABLE_QUERY_SELECTION,
+    PSDK_ENABLE_SERVER_COMPILED_AEL,
+    cached_ael_capability_kwargs,
+)
+from aerospike_sdk.query_selection import (
+    compute_query_selection_support,
+    compute_query_selection_support_blocking,
+)
 from aerospike_sdk.sdk_config_monitor import AsyncSdkConfigMonitor, SdkConfigSource
+from aerospike_sdk.server_compiled_ael import (
+    compute_server_compiled_ael_support,
+    compute_server_compiled_ael_support_blocking,
+)
 
 if typing.TYPE_CHECKING:
     from aerospike_sdk.aio.session import Session
@@ -147,6 +160,8 @@ def __init__(
         # Shared by all Session instances from this client; avoids repeated
         # namespace/<ns> info probes when callers use multiple sessions.
         self._namespace_mode_cache: Dict[str, Mode] = {}
+        self._cached_supports_query_selection: Optional[bool] = None
+        self._cached_supports_server_compiled_ael: Optional[bool] = None
         # Resolved SDK-level settings (file over programmatic over defaults).
         # A frozen snapshot swapped wholesale by the config monitor, so the
         # operation path reads it lock-free.
@@ -220,6 +235,18 @@ async def connect(self) -> None:
             log.debug("Connecting to cluster seeds=%r", self._seeds)
         self._client = await new_client(self._policy, self._seeds)
         self._connected = True
+        if PSDK_ENABLE_QUERY_SELECTION:
+            self._cached_supports_query_selection = await compute_query_selection_support(
+                self._client,
+            )
+        else:
+            self._cached_supports_query_selection = False
+        if PSDK_ENABLE_SERVER_COMPILED_AEL:
+            self._cached_supports_server_compiled_ael = (
+                await compute_server_compiled_ael_support(self._client)
+            )
+        else:
+            self._cached_supports_server_compiled_ael = False
         log.info(
             "Connected seeds=%r", self._seeds,
             extra={"aerospike.cluster": self._policy.cluster_name},
@@ -261,9 +288,33 @@ async def close(self) -> None:
             self._client = None
             self._connected = False
             log.info("Client closed")
+        self._cached_supports_query_selection = None
+        self._cached_supports_server_compiled_ael = None
         self._namespace_mode_cache.clear()
         self._supports_mrt_cache = None
 
+    @property
+    def supports_query_selection(self) -> bool:
+        """``True`` when all cluster nodes support field ``44`` query selection (>= 8.1.3).
+
+        Computed at :meth:`connect` / :meth:`connect_blocking` from PAC
+        ``Version.supports_query_selection()`` on every node.
+        """
+        if not self._connected or self._client is None:
+            return False
+        return bool(self._cached_supports_query_selection)
+
+    @property
+    def supports_server_compiled_ael(self) -> bool:
+        """``True`` when server-compiled AEL filters are usable on this connection.
+
+        Requires all nodes >= 8.1.3 (PAC ``Version.supports_server_compiled_ael``)
+        and PAC ``FilterExpression.from_server_compiled_ael``. Cached at connect.
+        """
+        if not self._connected or self._client is None:
+            return False
+        return bool(self._cached_supports_server_compiled_ael)
+
     def connect_blocking(self) -> None:
         """Synchronously open a connection without requiring an asyncio loop.
 
@@ -295,6 +346,18 @@ def connect_blocking(self) -> None:
             log.debug("Connecting (blocking) to cluster seeds=%r", self._seeds)
         self._client = new_client_blocking(self._policy, self._seeds)
         self._connected = True
+        if PSDK_ENABLE_QUERY_SELECTION:
+            self._cached_supports_query_selection = compute_query_selection_support_blocking(
+                self._client,
+            )
+        else:
+            self._cached_supports_query_selection = False
+        if PSDK_ENABLE_SERVER_COMPILED_AEL:
+            self._cached_supports_server_compiled_ael = (
+                compute_server_compiled_ael_support_blocking(self._client)
+            )
+        else:
+            self._cached_supports_server_compiled_ael = False
         log.info(
             "Connected seeds=%r", self._seeds,
             extra={"aerospike.cluster": self._policy.cluster_name},
@@ -313,6 +376,8 @@ def close_blocking(self) -> None:
             self._client = None
             self._connected = False
             log.info("Client closed")
+        self._cached_supports_query_selection = None
+        self._cached_supports_server_compiled_ael = None
         self._namespace_mode_cache.clear()
         self._supports_mrt_cache = None
 
@@ -487,7 +552,12 @@ def _query(
                 behavior=behavior,
                 indexes_monitor=self._indexes_monitor,
                 namespace_mode_resolver=namespace_mode_resolver,
+                namespace_mode_resolver_blocking=namespace_mode_resolver_blocking,
                 sdk_client=self,
+                **cached_ael_capability_kwargs(
+                    self._cached_supports_server_compiled_ael,
+                    self._cached_supports_query_selection,
+                ),
             )
             builder._single_key = key
             return builder
@@ -505,7 +575,12 @@ def _query(
                 behavior=behavior,
                 indexes_monitor=self._indexes_monitor,
                 namespace_mode_resolver=namespace_mode_resolver,
+                namespace_mode_resolver_blocking=namespace_mode_resolver_blocking,
                 sdk_client=self,
+                **cached_ael_capability_kwargs(
+                    self._cached_supports_server_compiled_ael,
+                    self._cached_supports_query_selection,
+                ),
             )
             builder._keys = keys
             return builder
@@ -535,6 +610,10 @@ def _query(
             namespace_mode_resolver=namespace_mode_resolver,
             namespace_mode_resolver_blocking=namespace_mode_resolver_blocking,
             sdk_client=self,
+            **cached_ael_capability_kwargs(
+                self._cached_supports_server_compiled_ael,
+                self._cached_supports_query_selection,
+            ),
         )
 
     @overload

aerospike_sdk/aio/operations/query.py

@@ -78,6 +78,7 @@
     AerospikeError,
     _convert_pac_exception,
 )
+from aerospike_sdk.feature_gates import cached_ael_capability_kwargs
 from aerospike_sdk.policy.behavior_settings import Mode, OpKind, OpShape
 from aerospike_sdk.record_result import RecordResult
 from aerospike_sdk.record_stream import RecordStream
@@ -965,7 +966,9 @@ async def _execute_dataset_query(self) -> RecordStream:
         log.debug(
             "dataset query: %s.%s filter=%s chunk=%s hint=%s",
             self._namespace, self._set_name,
-            self._filter_expression is not None or bool(self._filter_records),
+            self._filter_expression is not None
+            or self._where_ael is not None
+            or bool(self._filter_records),
             self._chunk_size,
             self._query_hint is not None,
             extra={"aerospike.cluster": _cmd_cluster(self._client)},
@@ -980,40 +983,52 @@ async def _execute_dataset_query(self) -> RecordStream:
             policy = self._apply_txn(QueryPolicy())
         if self._chunk_size is not None and self._chunk_size > 0:
             policy.max_records = self._chunk_size
-        if self._filter_expression is not None:
-            policy.filter_expression = self._filter_expression
-
         hint = self._query_hint
+        use_server_query_selection = self._use_server_query_selection(hint)
+        self._apply_dataset_query_policy_filter(
+            policy, use_server_query_selection=use_server_query_selection,
+        )
+
         if hint is not None and hint.query_duration is not None:
             policy.expected_duration = hint.query_duration
 
-        if self._where_ael is not None and self._indexes_monitor is not None:
-            # Lazy start: the monitor's daemon thread only spins up on the
-            # first AEL ``where()`` query. ``start()`` is idempotent.
-            self._indexes_monitor.start(self._client)
-            # Offload the readiness wait so the event loop isn't pinned for
-            # the first-fetch case (subsequent calls return immediately).
-            await asyncio.to_thread(self._indexes_monitor.wait_until_ready)
+        self._prepare_dataset_query_index_context(
+            use_server_query_selection=use_server_query_selection,
+        )
+        await self._wait_for_dataset_query_index_context(
+            use_server_query_selection=use_server_query_selection,
+        )
 
-        self._resolve_index_context()
+        if not use_server_query_selection and self._where_ael is not None:
+            self._resolve_index_context()
 
         partition_filter = self._partition_filter or PartitionFilter.all()
 
-        if self._where_ael is not None and self._index_context is not None:
-            self._auto_generate_filters(hint, policy)
+        self._maybe_auto_generate_filters(
+            hint, policy, use_server_query_selection=use_server_query_selection,
+        )
 
         statement = self._build_statement()
 
         try:
-            recordset = await self._client.query(statement, partition_filter, policy=policy)
+            recordset, plan = await self._run_dataset_query_async(
+                policy, partition_filter, hint, statement,
+                use_server_query_selection=use_server_query_selection,
+            )
         except Exception as e:
             raise _convert_pac_exception(e) from e
 
         if self._chunk_size is not None and self._chunk_size > 0:
             client = self._client
 
-            async def _reexecute(pf: PartitionFilter) -> Any:
-                return await client.query(statement, pf, policy=policy)
+            if plan is not None:
+                async def _reexecute(pf: PartitionFilter) -> Any:
+                    return await client.query_with_plan(
+                        statement, pf, plan, policy=policy,
+                    )
+            else:
+                async def _reexecute(pf: PartitionFilter) -> Any:
+                    return await client.query(statement, pf, policy=policy)
 
             return RecordStream._from_chunked_pac_recordset(
                 recordset,
@@ -1148,6 +1163,10 @@ def _promote(self) -> None:
             txn=self._txn,
             namespace_mode_resolver=self._namespace_mode_resolver,
             namespace_mode_resolver_blocking=self._namespace_mode_resolver_blocking,
+            **cached_ael_capability_kwargs(
+                getattr(self._sdk_client_fast, "_cached_supports_server_compiled_ael", None),
+                getattr(self._sdk_client_fast, "_cached_supports_query_selection", None),
+            ),
         )
         qb._op_type = self._op_type_fast
         qb._single_key = self._key

aerospike_sdk/aio/session.py

@@ -48,6 +48,7 @@
 )
 from aerospike_sdk.aio.operations.udf import UdfFunctionBuilder
 from aerospike_sdk.dataset import DataSet
+from aerospike_sdk.feature_gates import cached_ael_capability_kwargs
 from aerospike_sdk.policy.behavior import Behavior, OpKind, OpShape
 from aerospike_sdk.policy.behavior_settings import Mode
 from aerospike_sdk.policy.policy_mapper import to_read_policy, to_write_policy
@@ -610,6 +611,10 @@ def execute_udf(self, *keys: Key) -> "UdfFunctionBuilder":
             namespace_mode_resolver=self._resolve_namespace_mode,
             namespace_mode_resolver_blocking=self._resolve_namespace_mode_blocking,
             sdk_client=self._client,
+            **cached_ael_capability_kwargs(
+                self._client._cached_supports_server_compiled_ael,
+                self._client._cached_supports_query_selection,
+            ),
         )
         qb._set_current_keys_from_varargs(keys)
         return UdfFunctionBuilder(qb)
@@ -691,6 +696,10 @@ def _build_write_segment(
             namespace_mode_resolver=self._resolve_namespace_mode,
             namespace_mode_resolver_blocking=self._resolve_namespace_mode_blocking,
             sdk_client=self._client,
+            **cached_ael_capability_kwargs(
+                self._client._cached_supports_server_compiled_ael,
+                self._client._cached_supports_query_selection,
+            ),
         )
         target: Union[Key, List[Key]] = all_keys[0] if len(all_keys) == 1 else all_keys
         return qb._start_write_verb(op_type, target)
@@ -746,6 +755,10 @@ def _fast_query_builder(self, key: Key, behavior: Behavior) -> QueryBuilder:
             self._resolve_namespace_mode,
             self._resolve_namespace_mode_blocking,
             self._client,
+            **cached_ael_capability_kwargs(
+                self._client._cached_supports_server_compiled_ael,
+                self._client._cached_supports_query_selection,
+            ),
         )
         builder._single_key = key
         return builder

aerospike_sdk/exceptions.py

@@ -661,6 +661,9 @@ def _convert_pac_exception(exc: Exception) -> AerospikeError:
     should use ``raise convert_pac_exception(e) from e``.
         :func:`_result_code_to_exception`
     """
+    if isinstance(exc, AerospikeError):
+        return exc
+
     if isinstance(exc, PacServerError):
         return _result_code_to_exception(
             exc.result_code,