[CLIENT-4000] Add support for bin projection.

doc/query.rst

@@ -12,31 +12,45 @@ Overview
 Constructing A Query
 --------------------
 
-.. warning:: :meth:`aerospike.Query` should not be called directly to create a :class:`~aerospike.Query` object.
-
 The query object is used for executing queries over a secondary index of a specified set.
-It can be created by calling :meth:`aerospike.Client.query`.
+It can be created by calling :meth:`aerospike.Client.query`. If the set is initialized to :py:obj:`None`,
+then the query will only apply to records without a set.
+
+.. warning:: :meth:`aerospike.Query` should not be called directly to create a :class:`~aerospike.Query` object.
 
 A query without a secondary index filter will apply to all records in the namespace,
 similar to a :class:`~aerospike.Scan`.
 
-Otherwise, the query can optionally be assigned one of the secondary index filters in :mod:`aerospike.predicates`
+Secondary Index Filters
+-----------------------
+
+The query can optionally be assigned one of the secondary index filters in :mod:`aerospike.predicates`
 to filter out records using their bin values.
 These secondary index filters are applied to the query using :meth:`~aerospike.Query.where`.
-In this case, if the set is initialized to :py:obj:`None`, then the query will only apply to records without a set.
 
 .. note::
     The secondary index filters in :mod:`aerospike.predicates` are **not** the same as
     the deprecated `predicate expressions`.
     For more details, read this `guide <https://aerospike.com/docs/develop/learn/queries/secondary-index>`_.
 
-Writing Using Query
--------------------
+Filtering Bins
+--------------
+
+The returned bins can be filtered by using :meth:`~aerospike.Query.select`.
+
+Background Queries
+------------------
 
 If a list of write operations is added to the query with :meth:`~aerospike.Query.add_ops`, \
 they will be applied to each record processed by the query. \
 See available write operations at :mod:`aerospike_helpers.operations`.
 
+
+Foreground Queries
+------------------
+
+Read operations can also be added in a foreground query using :meth:`~aerospike.Query.add_ops`.
+
 Query Aggregations
 ------------------
 
@@ -47,13 +61,12 @@ records streaming back from the query.
 Getting Results From Query
 --------------------------
 
-The returned bins can be filtered by using :meth:`select`.
+For performing write operations or applying record UDF's, :meth:`~aerospike.Query.execute_background` must be used.
 
-Finally, the query is invoked using one of these methods:
+Otherwise, the query is invoked using one of these synchronous methods:
 
 - :meth:`~aerospike.Query.foreach`
 - :meth:`~aerospike.Query.results`
-- :meth:`~aerospike.Query.execute_background`
 
 .. seealso::
     `Queries <https://aerospike.com/docs/develop/learn/queries/>`_ and \
@@ -107,12 +120,16 @@ Assume this boilerplate code is run before all examples below:
 
     .. method:: select(bin1[, bin2[, bin3..]])
 
-        Set a filter on the record bins resulting from :meth:`results` or \
-        :meth:`foreach`.
+        .. warning:: In the next major client release, calling this method after :meth:`Query.add_ops` was called on the same Query object will raise a :py:exc:`~aerospike.exception.ParamError` exception.
+
+        Set a filter on the record bins resulting from :meth:`results` or :meth:`foreach`.
 
         If this method is called more than once on the same query instance, a :py:exc:`~aerospike.exception.ClientError` exception will be raised.
 
-        If a selected bin does not exist in a record it will not appear in the *bins* portion of that record tuple.
+        If a selected bin does not exist in a record, it will not appear in the *bins* portion of that record tuple.
+
+        If this method is called after :meth:`Query.add_ops` was called on the same Query object, the selected bins in
+        this call will be ignored during the query.
 
     .. method:: where(predicate[, ctx])
 
@@ -250,6 +267,8 @@ Assume this boilerplate code is run before all examples below:
         predicate is attached to the  :class:`~aerospike.Query` the stream UDF \
         will aggregate over all the records in the specified set.
 
+        This function can also be used to apply a record UDF.
+
         :param str module: the name of the Lua module.
         :param str function: the name of the Lua function within the *module*.
         :param list arguments: optional arguments to pass to the *function*. NOTE: these arguments must be types supported by Aerospike See: `supported data types <https://aerospike.com/docs/develop/client/python/data-types/>`_.
@@ -280,11 +299,22 @@ Assume this boilerplate code is run before all examples below:
 
     .. method:: add_ops(ops)
 
-        Add a list of write ops to the query.
-        When used with :meth:`Query.execute_background` the query will perform the write ops on any records found.
+        .. warning:: In the next major client release, if this is called after :meth:`~Query.select` was called on the same object, an :exc:`~aerospike.exception.ParamError` will be raised.
+
+        Add a list of operations to the query.
+
+        For background queries, only write operations are allowed.
+        For foreground queries, only read operations are allowed.
+
+        For server versions < 8.1.2, basic read operations are allowed in foreground queries. Otherwise with this
+        server version, using a non-basic read operation will raise a :exc:`~aerospike.exception.ParamError`.
+
         If no predicate is attached to the Query it will apply ops to all the records in the specified set.
 
-        :param ops: `list` A list of write operations generated by the aerospike_helpers e.g. list_operations, map_operations, etc.
+        If there are selected bins in this Query object via :meth:`~Query.select`, those selected bins will be ignored
+        during the query.
+
+        :param ops: `list` A list of operations generated from :ref:`aerospike_operation_helpers.operations`.
 
         .. note::
             Requires server version >= 4.7.0.

src/include/client.h

@@ -586,3 +586,8 @@ PyObject *AerospikeClient_Commit(AerospikeClient *self, PyObject *args,
                                  PyObject *kwds);
 PyObject *AerospikeClient_Abort(AerospikeClient *self, PyObject *args,
                                 PyObject *kwds);
+
+#define SELECT_AND_ADD_OPS_ARE_MUTUALLY_EXCLUSIVE_MESSAGE                      \
+    "Operations and bin names are mutually exclusive."                         \
+    "In the next major client release, when this %s object is executed, a "    \
+    "ParamError will be raised."

src/main/query/add_ops.c

@@ -31,6 +31,17 @@
 AerospikeQuery *AerospikeQuery_Add_Ops(AerospikeQuery *self, PyObject *args,
                                        PyObject *kwds)
 {
+    if (self->query.select.size) {
+        // If select() was called on this Query object before.
+
+        int retval = PyErr_WarnFormat(
+            PyExc_DeprecationWarning, STACK_LEVEL,
+            SELECT_AND_ADD_OPS_ARE_MUTUALLY_EXCLUSIVE_MESSAGE, "Query");
+        if (retval == -1) {
+            return NULL;
+        }
+    }
+
     // Python function arguments.
     PyObject *py_ops = NULL;
     // Python function keyword arguments.
@@ -40,8 +51,6 @@ AerospikeQuery *AerospikeQuery_Add_Ops(AerospikeQuery *self, PyObject *args,
         return NULL;
     }
 
-    Py_INCREF(py_ops);
-
     // Aerospike API arguments.
     long return_type = -1;
     long operation;

src/main/query/select.c

@@ -33,6 +33,16 @@ AerospikeQuery *AerospikeQuery_Select(AerospikeQuery *self, PyObject *args,
 {
     TRACE();
 
+    // If add_ops() was called on this Query object before.
+    if (as_operations_defined(self->query.ops)) {
+        int retval = PyErr_WarnFormat(
+            PyExc_DeprecationWarning, STACK_LEVEL,
+            SELECT_AND_ADD_OPS_ARE_MUTUALLY_EXCLUSIVE_MESSAGE, "Query");
+        if (retval == -1) {
+            return NULL;
+        }
+    }
+
     int nbins = (int)PyTuple_Size(args);
     char *bin = NULL;
     PyObject *py_ubin = NULL;

src/main/scan/add_ops.c

@@ -31,6 +31,17 @@
 AerospikeScan *AerospikeScan_Add_Ops(AerospikeScan *self, PyObject *args,
                                      PyObject *kwds)
 {
+    if (self->scan.select.size) {
+        // If select() was called on this Scan object before.
+
+        int retval = PyErr_WarnFormat(
+            PyExc_DeprecationWarning, STACK_LEVEL,
+            SELECT_AND_ADD_OPS_ARE_MUTUALLY_EXCLUSIVE_MESSAGE, "Scan");
+        if (retval == -1) {
+            return NULL;
+        }
+    }
+
     // Python function arguments.
     PyObject *py_ops = NULL;
     // Python function keyword arguments.

src/main/scan/select.c

@@ -32,6 +32,16 @@ AerospikeScan *AerospikeScan_Select(AerospikeScan *self, PyObject *args,
 {
     TRACE();
 
+    // If add_ops() was called on this Scan object before.
+    if (as_operations_defined(self->scan.ops)) {
+        int retval = PyErr_WarnFormat(
+            PyExc_DeprecationWarning, STACK_LEVEL,
+            SELECT_AND_ADD_OPS_ARE_MUTUALLY_EXCLUSIVE_MESSAGE, "Scan");
+        if (retval == -1) {
+            return NULL;
+        }
+    }
+
     char *bin = NULL;
     PyObject *py_ustr = NULL;
     as_error err;

test/new_tests/conftest.py

@@ -10,6 +10,8 @@
 
 import aerospike
 from aerospike import exception as e
+from aerospike_helpers.operations import operations
+from aerospike_helpers.batch.records import BatchRecords, Write
 from contextlib import nullcontext
 
 # Comment this out because nowhere in the repository is using it
@@ -260,6 +262,57 @@ def wait_for_job_completion(as_connection, job_id, job_module: int = aerospike.J
             break
         time.sleep(0.1)
 
+# Shared between bin projection and execute background tests
+
+TEST_NS = "test"
+TEST_SET = "demo"
+BIN_NAME = "number"
+MAP_BIN_NAME = "map"
+
+# TODO: scale down tests maybe
+KEYS = [(TEST_NS, TEST_SET, i) for i in range(500)]
+expected_number_bin_values = set()
+
+# Add records around the test
+@pytest.fixture(scope="function")
+def clean_test_background(as_connection):
+    batch_records = []
+    brs = BatchRecords(batch_records=batch_records)
+    for i, key in enumerate(KEYS):
+        ops = [
+            operations.write(BIN_NAME, i),
+            operations.write(MAP_BIN_NAME, {"a": i})
+        ]
+        br = Write(key, ops=ops)
+        batch_records.append(br)
+        expected_number_bin_values.add(i)
+    as_connection.batch_write(brs)
+    yield
+    as_connection.batch_remove(KEYS)
+
+BASIC_READ_BIN_OPS = [
+    operations.read(BIN_NAME)
+]
+
+READ_AND_WRITE_OPS = [
+    operations.read(BIN_NAME),
+    operations.write(BIN_NAME, 1)
+]
+
+WRITE_OPS = [
+    operations.write(BIN_NAME, 3)
+]
+
+NON_EXISTENT_BIN_NAME = "asdf"
+
+@pytest.fixture()
+def query(request, clean_test_background, as_connection):
+    if not hasattr(request, "param"):
+        query = as_connection.query(TEST_NS, TEST_SET)
+    else:
+        query = request.param(as_connection, TEST_NS, TEST_SET)
+    yield query
+
 @pytest.fixture(scope="function")
 def expect_earlier_than_server_version_to_fail(as_connection, request):
     # Some requesting test cases may not set the param. Like if it is a negative client-side test case and there is no

test/new_tests/test_query_bin_projection.py

@@ -0,0 +1,113 @@
+import pytest
+from .conftest import BIN_NAME, BASIC_READ_BIN_OPS, READ_AND_WRITE_OPS, NON_EXISTENT_BIN_NAME, WRITE_OPS, MAP_BIN_NAME, expected_number_bin_values
+import aerospike
+from aerospike_helpers.operations import map_operations
+from aerospike import Query
+from aerospike import exception as e
+from .test_base_class import TestBaseClass
+from contextlib import nullcontext
+
+
+class TestQueryBinProjection:
+    pytestmark = [
+        pytest.mark.parametrize(
+            "query",
+            [
+                aerospike.Client.scan,
+                aerospike.Client.query
+            ],
+            indirect=True
+        )
+    ]
+
+    def test_query_foreach(self, query):
+        bin_values = set()
+        def callback(record):
+            bin_values.add(record[2][BIN_NAME])
+
+        query.add_ops(BASIC_READ_BIN_OPS)
+        query.foreach(callback)
+
+        assert bin_values == expected_number_bin_values
+
+    def test_query_results(self, query):
+        query.add_ops(BASIC_READ_BIN_OPS)
+        records = query.results()
+
+        bin_values = [record[2][BIN_NAME] for record in records]
+        assert len(bin_values) == len(set(bin_values)) and set(bin_values) == expected_number_bin_values
+
+    MAP_GET_BY_KEY_OP = [
+        map_operations.map_get_by_key(MAP_BIN_NAME, "a", aerospike.MAP_RETURN_VALUE)
+    ]
+
+    @pytest.fixture()
+    def client_should_fail_if_server_version_less_than_8_1_2(self, as_connection, request):
+        # For bin projection, the server can convert complex (e.g map) read operations into a regular bin read
+        # Since the server doesn't fail, the client has to check the server version and raise an error on its end.
+        if (TestBaseClass.major_ver, TestBaseClass.minor_ver, TestBaseClass.patch_ver) >= (8, 1, 2):
+            yield nullcontext()
+        else:
+            yield pytest.raises(e.ParamError)
+
+    def test_query_nested_results(self, query, client_should_fail_if_server_version_less_than_8_1_2):
+        query.add_ops(self.MAP_GET_BY_KEY_OP)
+        with client_should_fail_if_server_version_less_than_8_1_2:
+            records = query.results()
+
+            bin_values = [record[2][MAP_BIN_NAME] for record in records]
+            assert len(bin_values) == len(set(bin_values)) and set(bin_values) == expected_number_bin_values
+
+    # Negative tests
+
+    def noop_callback(record):
+        pass
+
+    @pytest.mark.parametrize(
+        "api_method, args",
+        [
+            ("results", []),
+            ("foreach", [noop_callback])
+        ]
+    )
+    @pytest.mark.parametrize(
+        "ops",
+        [
+            READ_AND_WRITE_OPS,
+            WRITE_OPS
+        ]
+    )
+    def test_add_write_ops_in_foreground_query(self, query, api_method, args, ops):
+        query.add_ops(ops)
+        with pytest.raises(e.ParamError) as excinfo:
+            getattr(query, api_method)(*args)
+        cls_name: str = query.__class__.__name__
+        assert excinfo.value.msg == f"{cls_name} operations must be read-only. Use background {cls_name.lower()} for write-only operations."
+
+    # The following 2 test cases should work for any server version. Not just 8.1.2
+
+    def test_select_bins_then_add_ops_then_foreground_query(self, query):
+        query.select(NON_EXISTENT_BIN_NAME)
+        with pytest.warns(DeprecationWarning):
+            query.add_ops(BASIC_READ_BIN_OPS)
+
+        records = query.results()
+
+        # The "filtered out" bin should still be returned
+        for _, _, bins in records:
+            assert BIN_NAME in bins
+
+    def test_add_ops_then_select_bins_then_foreground_query(self, query):
+        query.add_ops(BASIC_READ_BIN_OPS)
+        with pytest.warns(DeprecationWarning):
+            query.select(NON_EXISTENT_BIN_NAME)
+
+        records = query.results()
+
+        # The "filtered out" bin should still be returned
+        for _, _, bins in records:
+            assert BIN_NAME in bins
+
+    def test_add_ops_invalid_args(self, query):
+        with pytest.raises(TypeError):
+            query.add_ops()