CLIENT-4878 serverside ael parsing

aerospike_sdk/ael/server_filter.py

@@ -0,0 +1,39 @@
+# 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
+
+def filter_expression_from_ael_string(
+    ael: str,
+    *,
+    supports_server_compiled_ael: bool,
+) -> FilterExpression:
+    """Return a ``FilterExpression`` for *ael*, using server-compiled wire form when allowed.
+
+    When ``supports_server_compiled_ael`` is true
+    returns that (MessagePack ``[128, "<utf-8 ael>"]``). Otherwise, parses on the
+    client via :func:`~aerospike_sdk.ael.parser.parse_ael`.
+    """
+    if supports_server_compiled_ael:
+        factory = getattr(FilterExpression, "from_server_compiled_ael", None)
+        if callable(factory):
+            return factory(ael)
+    return parse_ael(ael)

aerospike_sdk/aio/background.py

@@ -38,7 +38,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
 
@@ -243,7 +243,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._session.client.supports_server_compiled_ael,
+            )
         else:
             self._filter_expression = expression
         return self
@@ -584,7 +587,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._session.client.supports_server_compiled_ael,
+            )
         else:
             self._filter_expression = expression
         return self

aerospike_sdk/aio/client.py

@@ -26,6 +26,7 @@
     AdminPolicy,
     Client as AsyncClient,
     ClientPolicy,
+    FilterExpression,
     Key,
     RegisterTask,
     UDFLang,
@@ -48,6 +49,57 @@
 log = logging.getLogger(__name__)
 
 
+def _pac_version_supports_server_compiled_filter(version_obj: object) -> bool:
+    """Delegate to PAC :class:`~aerospike_async.Version` server-compiled AEL support.
+
+    Calls :meth:`~aerospike_async.Version.supports_server_compiled_ael` when present.
+    Older PAC builds omit that method; then return ``False`` so the SDK keeps
+    client-side :func:`~aerospike_sdk.ael.parser.parse_ael` for string predicates.
+    """
+    fn = getattr(version_obj, "supports_server_compiled_ael", None)
+    if not callable(fn):
+        return False
+    return bool(fn())
+
+
+async def _first_active_node_supports_server_compiled_ael(pac: AsyncClient) -> bool:
+    """Whether the first **active** node's version reports server-compiled AEL support.
+
+    Assumes **homogeneous** cluster builds (all nodes same server version); only
+    the first active node is consulted. Returns ``False`` if there are no active
+    nodes.
+    """
+    nodes = await pac.nodes()
+    for n in nodes:
+        if n.is_active:
+            return _pac_version_supports_server_compiled_filter(n.version)
+    return False
+
+
+def _first_active_node_supports_server_compiled_ael_blocking(pac: AsyncClient) -> bool:
+    """Blocking counterpart of :func:`_first_active_node_supports_server_compiled_ael`.
+
+    Uses :meth:`~aerospike_async.Client.nodes_blocking` so ``new_client_blocking()``
+    clients never call async-only PAC APIs (which raise at runtime).
+    """
+    nodes = pac.nodes_blocking()
+    for n in nodes:
+        if n.is_active:
+            return _pac_version_supports_server_compiled_filter(n.version)
+    return False
+
+
+def _compute_server_compiled_ael_support_blocking(pac: AsyncClient) -> bool:
+    """Gate for server-compiled string ``where()`` on a blocking PAC client.
+
+    Same rules as :meth:`Client._compute_server_compiled_ael_support`, but uses
+    :func:`_first_active_node_supports_server_compiled_ael_blocking` — no asyncio.
+    """
+    if not callable(getattr(FilterExpression, "from_server_compiled_ael", None)):
+        return False
+    return _first_active_node_supports_server_compiled_ael_blocking(pac)
+
+
 class Client:
     """Async entry point for the SDK API over the Aerospike Python Async Client.
 
@@ -134,6 +186,7 @@ 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_server_compiled_ael: Optional[bool] = None
 
     async def connect(self) -> None:
         """Open a connection to the cluster using the configured seeds and policy.
@@ -176,6 +229,9 @@ async def connect(self) -> None:
         # IndexesMonitor starts lazily on the first AEL ``where()`` query.
         # Sync benches and callers that never touch the AEL filter-generation
         # path pay zero daemon-thread cost.
+        self._cached_supports_server_compiled_ael = (
+            await self._compute_server_compiled_ael_support()
+        )
 
     async def close(self) -> None:
         """Close the underlying async client and clear connection state.
@@ -191,8 +247,42 @@ async def close(self) -> None:
             await self._client.close()
             self._client = None
             self._connected = False
+        self._cached_supports_server_compiled_ael = None
         self._namespace_mode_cache.clear()
 
+    async def _compute_server_compiled_ael_support(self) -> bool:
+        """End-to-end gate for server-compiled string ``where()`` (computed once per connect).
+
+        Requires (1) PAC :meth:`FilterExpression.from_server_compiled_ael` and
+        (2) the **first active** node's ``version.supports_server_compiled_ael()``
+        (homogeneous cluster assumption — all nodes same build). Cached on
+        :meth:`connect`; see :attr:`supports_server_compiled_ael`.
+        """
+        if self._client is None:
+            return False
+        if not callable(getattr(FilterExpression, "from_server_compiled_ael", None)):
+            return False
+        return await _first_active_node_supports_server_compiled_ael(self._client)
+
+    @property
+    def supports_server_compiled_ael(self) -> bool:
+        """Whether server-compiled AEL filters are usable on this connection.
+
+        **Source of truth:** PAC ``Version.supports_server_compiled_ael()`` on the
+        **first active** node only (homogeneous cluster: all nodes same build). The
+        SDK does **not** re-walk the node list on every read; it returns the boolean
+        computed at the last successful :meth:`connect`.
+
+        Also requires the installed PAC to expose
+        :meth:`FilterExpression.from_server_compiled_ael`; otherwise this is
+        ``False`` even when the sampled node reports support.
+
+        Returns ``False`` before :meth:`connect` completes or after :meth:`close`.
+        """
+        if self._cached_supports_server_compiled_ael is None:
+            return False
+        return self._cached_supports_server_compiled_ael
+
     def connect_blocking(self) -> None:
         """Synchronously open a connection without requiring an asyncio loop.
 
@@ -224,6 +314,9 @@ def connect_blocking(self) -> None:
         self._client = new_client_blocking(self._policy, self._seeds)
         self._connected = True
         # IndexesMonitor starts lazily on the first AEL ``where()`` query.
+        self._cached_supports_server_compiled_ael = (
+            _compute_server_compiled_ael_support_blocking(self._client)
+        )
 
     def close_blocking(self) -> None:
         """Synchronously close the underlying client. Pair with :meth:`connect_blocking`.
@@ -236,6 +329,7 @@ def close_blocking(self) -> None:
             self._client.close_blocking()
             self._client = None
             self._connected = False
+        self._cached_supports_server_compiled_ael = None
         self._namespace_mode_cache.clear()
 
     async def __aenter__(self) -> Client:
@@ -469,6 +563,8 @@ def query(
                 behavior=behavior,
                 indexes_monitor=self._indexes_monitor,
                 namespace_mode_resolver=namespace_mode_resolver,
+                namespace_mode_resolver_blocking=namespace_mode_resolver_blocking,
+                supports_server_compiled_ael=self.supports_server_compiled_ael,
             )
             builder._single_key = key
             return builder
@@ -486,6 +582,8 @@ def query(
                 behavior=behavior,
                 indexes_monitor=self._indexes_monitor,
                 namespace_mode_resolver=namespace_mode_resolver,
+                namespace_mode_resolver_blocking=namespace_mode_resolver_blocking,
+                supports_server_compiled_ael=self.supports_server_compiled_ael,
             )
             builder._keys = keys
             return builder
@@ -514,6 +612,7 @@ def query(
             indexes_monitor=self._indexes_monitor,
             namespace_mode_resolver=namespace_mode_resolver,
             namespace_mode_resolver_blocking=namespace_mode_resolver_blocking,
+            supports_server_compiled_ael=self.supports_server_compiled_ael,
         )
 
     @overload

aerospike_sdk/aio/operations/query.py

@@ -178,7 +178,8 @@ def _resolve_hll_flags(
     make_background_write_policy,
     reject_unsupported_background_write_ops,
 )
-from aerospike_sdk.ael.parser import parse_ael, parse_ael_with_index
+from aerospike_sdk.ael.parser import parse_ael_with_index
+from aerospike_sdk.ael.server_filter import filter_expression_from_ael_string
 from aerospike_sdk.error_strategy import (
     ErrorHandler,
     OnError,
@@ -397,8 +398,10 @@ def __init__(
         cached_read_policy_sc: Optional[ReadPolicy] = None,
         cached_write_policy_sc: Optional[WritePolicy] = None,
         txn: Optional[Txn] = None,
+        *,
         namespace_mode_resolver: NamespaceModeResolver = None,
         namespace_mode_resolver_blocking: Optional[Callable[[str], "Mode"]] = None,
+        supports_server_compiled_ael: bool = False,
     ) -> None:
         """
         Initialize a QueryBuilder.
@@ -423,6 +426,11 @@ def __init__(
                 Session-scoped builders supply this; client-only builders omit it.
             namespace_mode_resolver_blocking: Sync counterpart used by
                 :meth:`execute_blocking` (the sync path bypassing asyncio).
+            supports_server_compiled_ael: When true (typically from
+                :attr:`~aerospike_sdk.aio.client.Client.supports_server_compiled_ael`),
+                string :meth:`where` uses server-compiled AEL. ``False`` in tests
+                or PAC-only use defaults to client-side AEL parsing for string
+                predicates.
         """
         self._client = client
         self._namespace = namespace
@@ -445,6 +453,7 @@ def __init__(
         # reused under MRT because they were pre-computed without a txn, so
         # we null them out to force re-derivation from behavior.
         self._txn: Optional[Txn] = txn
+        self._supports_server_compiled_ael = supports_server_compiled_ael
         self._namespace_mode_resolver = namespace_mode_resolver
         self._namespace_mode: Optional[Mode] = None
         if txn is None:
@@ -459,6 +468,13 @@ def __init__(
             self._base_write_policy = None
             self._base_read_policy_sc = None
             self._base_write_policy_sc = None
+
+    def _filter_expression_from_ael(self, ael: str) -> FilterExpression:
+        return filter_expression_from_ael_string(
+            ael,
+            supports_server_compiled_ael=self._supports_server_compiled_ael,
+        )
+
     def _apply_txn(self, policy: Any) -> Any:
         """Stamp this builder's captured txn on an outer policy in place.
 
@@ -722,7 +738,7 @@ def where(
         """
         if isinstance(expression, str):
             self._where_ael = expression
-            self._filter_expression = parse_ael(expression)
+            self._filter_expression = self._filter_expression_from_ael(expression)
         else:
             self._where_ael = None
             self._filter_expression = expression
@@ -1143,7 +1159,7 @@ def default_where(
             :meth:`where`: Per-operation filter on the current operation.
         """
         if isinstance(expression, str):
-            self._default_filter_expression = parse_ael(expression)
+            self._default_filter_expression = self._filter_expression_from_ael(expression)
         else:
             self._default_filter_expression = expression
         return self
@@ -3567,7 +3583,9 @@ async def _execute_dataset_query(self) -> RecordStream:
 
         partition_filter = self._partition_filter or PartitionFilter.all()
 
-        if self._where_ael is not None and self._index_context is not None:
+        if self._where_ael is not None and self._index_context is not None and (
+            not self._supports_server_compiled_ael
+        ):
             self._auto_generate_filters(hint, policy)
 
         statement = self._build_statement()
@@ -3666,6 +3684,7 @@ class _SingleKeyWriteSegment(_SingleKeyWriteSegmentBase, WriteSegmentBuilder):
         # class-level defaults on _SingleKeyWriteSegmentBase (skip them
         # here so reads fall through to class default and writes go to
         # __dict__ inherited from the non-slotted base).
+        "_supports_server_compiled_ael",
     )
 
 
@@ -3702,6 +3721,7 @@ def _promote(self) -> None:
             txn=self._txn,
             namespace_mode_resolver=self._namespace_mode_resolver,
             namespace_mode_resolver_blocking=self._namespace_mode_resolver_blocking,
+            supports_server_compiled_ael=self._supports_server_compiled_ael,
         )
         qb._op_type = self._op_type_fast
         qb._single_key = self._key
@@ -5881,6 +5901,18 @@ def __init__(self, parent: _T, bin_name: str) -> None:
         self._parent = parent
         self._bin = bin_name
 
+    def _expression_from_ael_string_for_ops(
+        self, expression: Union[str, FilterExpression]
+    ) -> FilterExpression:
+        """Resolve AEL for bin expression reads using the same path as :meth:`where`."""
+        if not isinstance(expression, str):
+            return expression
+        qb = self._parent  # type: ignore[union-attr]
+        return filter_expression_from_ael_string(
+            expression,
+            supports_server_compiled_ael=qb._supports_server_compiled_ael,
+        )
+
     # -- Simple read ----------------------------------------------------------
 
     def get(self) -> _T:
@@ -6195,7 +6227,7 @@ def select_from(
             The parent builder for method chaining.
         """
         flags = ExpReadFlags.EVAL_NO_FAIL if ignore_eval_failure else ExpReadFlags.DEFAULT
-        expr = parse_ael(expression) if isinstance(expression, str) else expression
+        expr = self._expression_from_ael_string_for_ops(expression)
         self._parent.add_operation(ExpOperation.read(self._bin, expr, flags))  # type: ignore[union-attr]
         return self._parent
 

aerospike_sdk/aio/operations/udf.py

@@ -22,7 +22,6 @@
 from aerospike_async import FilterExpression, Key
 
 from aerospike_sdk.aio.operations.query import QueryBuilder, WriteSegmentBuilder
-from aerospike_sdk.ael.parser import parse_ael
 from aerospike_sdk.error_strategy import OnError
 from aerospike_sdk.record_stream import RecordStream
 
@@ -121,7 +120,7 @@ def where(
             :meth:`QueryBuilder.where`: Same AEL for reads.
         """
         if isinstance(expression, str):
-            self._qb._filter_expression = parse_ael(expression)
+            self._qb._filter_expression = self._qb._filter_expression_from_ael(expression)
         else:
             self._qb._filter_expression = expression
         return self