diff --git a/aerospike-client-c b/aerospike-client-c
index bce75a68ad..53f245112d 160000
--- a/aerospike-client-c
+++ b/aerospike-client-c
@@ -1 +1 @@
-Subproject commit bce75a68adcce897c7e58168113bf928097fb0bd
+Subproject commit 53f245112dc64282978fe0b510e84cebacf65131
diff --git a/doc/client.rst b/doc/client.rst
index 3f80225b34..07bad616c7 100755
--- a/doc/client.rst
+++ b/doc/client.rst
@@ -2141,10 +2141,7 @@ Write Policies
     .. hlist::
         :columns: 1
 
-        * **key**
-            | One of the :ref:`POLICY_KEY` values such as :data:`aerospike.POLICY_KEY_DIGEST`
-            |
-            | Default: :data:`aerospike.POLICY_KEY_DIGEST`
+        * .. include:: ./key_option_with_union_precedence.rst
         * **exists**
             | One of the :ref:`POLICY_EXISTS` values such as :data:`aerospike.POLICY_EXISTS_CREATE`
             |
@@ -2256,10 +2253,7 @@ Operate Policies
     .. hlist::
         :columns: 1
 
-        * **key**
-            | One of the :ref:`POLICY_KEY` values such as :data:`aerospike.POLICY_KEY_DIGEST`
-            |
-            | Default: :data:`aerospike.POLICY_KEY_DIGEST`
+        * .. include:: ./key_option_with_union_precedence.rst
         * **gen**
             | One of the :ref:`POLICY_GEN` values such as :data:`aerospike.POLICY_GEN_IGNORE`
             |
@@ -2343,10 +2337,7 @@ Apply Policies
     .. hlist::
         :columns: 1
 
-        * **key**
-            | One of the :ref:`POLICY_KEY` values such as :data:`aerospike.POLICY_KEY_DIGEST`
-            |
-            | Default: :data:`aerospike.POLICY_KEY_DIGEST`
+        * .. include:: ./key_option_with_union_precedence.rst
         * **replica**
             | One of the :ref:`POLICY_REPLICA` values such as :data:`aerospike.POLICY_REPLICA_MASTER`
             |
@@ -2509,10 +2500,7 @@ Batch Write Policies
     .. hlist::
         :columns: 1
 
-        * **key**
-            | One of the :ref:`POLICY_KEY` values such as :data:`aerospike.POLICY_KEY_DIGEST`
-            |
-            | Default: :data:`aerospike.POLICY_KEY_DIGEST`
+        * .. include:: ./key_option_with_union_precedence.rst
         * **commit_level**
             | One of the :ref:`POLICY_COMMIT_LEVEL` values such as :data:`aerospike.POLICY_COMMIT_LEVEL_ALL`
             |
@@ -2564,10 +2552,7 @@ Batch Apply Policies
     .. hlist::
         :columns: 1
 
-        * **key**
-            | One of the :ref:`POLICY_KEY` values such as :data:`aerospike.POLICY_KEY_DIGEST`
-            |
-            | Default: :data:`aerospike.POLICY_KEY_DIGEST`
+        * .. include:: ./key_option_with_union_precedence.rst
         * **commit_level**
             | One of the :ref:`POLICY_COMMIT_LEVEL` values such as :data:`aerospike.POLICY_COMMIT_LEVEL_ALL`
             |
diff --git a/doc/key_option_with_union_precedence.rst b/doc/key_option_with_union_precedence.rst
new file mode 100644
index 0000000000..57395936b1
--- /dev/null
+++ b/doc/key_option_with_union_precedence.rst
@@ -0,0 +1,9 @@
+**key**
+
+One of the :ref:`POLICY_KEY` values such as :data:`aerospike.POLICY_KEY_DIGEST`
+
+This option is unique from other options such that it uses union precedence;
+if this is set to :data:`aerospike.POLICY_KEY_SEND` either in dynamic config, config-level, or command-level,
+:data:`aerospike.POLICY_KEY_SEND` will always be applied.
+
+Default: :data:`aerospike.POLICY_KEY_DIGEST`
diff --git a/test/dyn_config.yml b/test/dyn_config.yml
index df3b9cb7a6..b6824e5eb3 100644
--- a/test/dyn_config.yml
+++ b/test/dyn_config.yml
@@ -4,6 +4,10 @@ dynamic:
     enable: true
   write:
     send_key: true
+  batch_write:
+    send_key: true
+  batch_udf:
+    send_key: true
 static:
   client:
     config_interval: 1000
diff --git a/test/new_tests/conftest.py b/test/new_tests/conftest.py
index a4de68835b..10f25c1d65 100644
--- a/test/new_tests/conftest.py
+++ b/test/new_tests/conftest.py
@@ -270,16 +270,32 @@ def wait_for_job_completion(as_connection, job_id, job_module: int = aerospike.J
 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):
+def insert_records(request, as_connection):
+
+    # - Some tests don't make use of unique sets,
+    # so we leave them alone for backwards compatibility
+    # - make_set_unique ensures that if a test case's cleanup stage fails to run
+    # e.g when the test case's setup fixture fails out,
+    # that test case's records does not interfere with future test cases that need to perform a query
+    num_keys, make_set_unique = request.param
+
+    if make_set_unique:
+        set_name = f"{TEST_SET}-{time.time_ns()}"
+    else:
+        set_name = TEST_SET
+
+    request.cls.set_name = set_name
+    keys = [(TEST_NS, set_name, i) for i in range(num_keys)]
+    request.cls.keys = keys
+
     batch_records = []
     brs = BatchRecords(batch_records=batch_records)
-    for i, key in enumerate(KEYS):
+
+    for i, key in enumerate(keys):
         ops = [
             operations.write(BIN_NAME, i),
             operations.write(MAP_BIN_NAME, {"a": i})
@@ -287,9 +303,22 @@ def clean_test_background(as_connection):
         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)
+
+    if make_set_unique is False:
+        as_connection.batch_remove(keys)
+
+def expect_records_to_have_user_key_stored(client: aerospike.Client, set_name: str):
+    query = client.query(TEST_NS, set_name)
+    recs = query.results()
+
+    # Check that record key tuple has the user key
+    for record in recs:
+        pk = record[0]
+        assert pk[2] is not None
 
 BASIC_READ_BIN_OPS = [
     operations.read(BIN_NAME)
@@ -307,7 +336,7 @@ def clean_test_background(as_connection):
 NON_EXISTENT_BIN_NAME = "asdf"
 
 @pytest.fixture()
-def query(request, clean_test_background, as_connection):
+def query(request, insert_records, as_connection):
     if not hasattr(request, "param"):
         query = as_connection.query(TEST_NS, TEST_SET)
     else:
@@ -413,3 +442,6 @@ def hydrate_partitions_1000_to_1003(request, as_connection):
     yield
 
     as_connection.batch_remove(keys)
+
+AEROSPIKE_CLIENT_CONFIG_URL = "AEROSPIKE_CLIENT_CONFIG_URL"
+DYN_CONFIG_PATH = "./dyn_config.yml"
diff --git a/test/new_tests/string_helpers.py b/test/new_tests/string_helpers.py
index 1cef3d127c..c31ab9e886 100644
--- a/test/new_tests/string_helpers.py
+++ b/test/new_tests/string_helpers.py
@@ -1,11 +1,8 @@
-from .conftest import KEYS
 import unicodedata
 import pytest
 from aerospike_helpers.string_helpers import StringPolicy
 
 
-KEY = KEYS[0]
-
 NON_STR_BIN_NAME = INT_BIN_NAME = "nonstr"
 STR_BIN_NAME = "str"
 DOUBLE_BIN_NAME = "double"
diff --git a/test/new_tests/test_command_level_policies.py b/test/new_tests/test_command_level_policies.py
index d997d3aa17..51552f9bf8 100644
--- a/test/new_tests/test_command_level_policies.py
+++ b/test/new_tests/test_command_level_policies.py
@@ -6,10 +6,10 @@
 from aerospike_helpers.batch import records as br
 from .test_base_class import TestBaseClass
 from aerospike_helpers.operations import operations
-from .conftest import verify_record_ttl
+from .conftest import verify_record_ttl, TEST_NS, TEST_SET
 
 SKIP_MSG = "read_touch_ttl_percent only supported on server 7.1 or higher"
-KEY = ("test", "demo", 1)
+KEY = (TEST_NS, TEST_SET, 1)
 
 
 @pytest.mark.usefixtures("as_connection")
diff --git a/test/new_tests/test_dynamic_config.py b/test/new_tests/test_dynamic_config.py
index b68c8816ab..5703186421 100644
--- a/test/new_tests/test_dynamic_config.py
+++ b/test/new_tests/test_dynamic_config.py
@@ -1,11 +1,11 @@
 import aerospike
 from aerospike import exception as e
+from .conftest import expect_records_to_have_user_key_stored, AEROSPIKE_CLIENT_CONFIG_URL, DYN_CONFIG_PATH
 from .test_base_class import TestBaseClass
 import pytest
 import os
 import glob
 
-DYN_CONFIG_PATH = "./dyn_config.yml"
 METRICS_LOG_FILES = "./metrics-*.log"
 
 
@@ -52,8 +52,6 @@ def cleanup_metrics_logs(self):
         for item in metrics_log_filenames:
             os.remove(item)
 
-    AEROSPIKE_CLIENT_CONFIG_URL = "AEROSPIKE_CLIENT_CONFIG_URL"
-
     @pytest.fixture
     def functional_test_setup(self, request, show_more_logs, cleanup_metrics_logs):
         config = TestBaseClass.get_connection_config()
@@ -73,7 +71,7 @@ def functional_test_setup(self, request, show_more_logs, cleanup_metrics_logs):
         setup_client.close()
 
         if request.param is True:
-            del os.environ[self.AEROSPIKE_CLIENT_CONFIG_URL]
+            del os.environ[AEROSPIKE_CLIENT_CONFIG_URL]
 
     # Decide whether env var should be used or not to read dynamic config file.
     # If not using env var, use the config provider instead
@@ -82,7 +80,7 @@ def functional_test_setup(self, request, show_more_logs, cleanup_metrics_logs):
     def test_dyn_config_file_works(self, functional_test_setup):
         config = TestBaseClass.get_connection_config()
         if functional_test_setup is True:
-            os.environ[self.AEROSPIKE_CLIENT_CONFIG_URL] = DYN_CONFIG_PATH
+            os.environ[AEROSPIKE_CLIENT_CONFIG_URL] = DYN_CONFIG_PATH
         else:
             provider = aerospike.ConfigProvider(DYN_CONFIG_PATH)
             config["config_provider"] = provider
@@ -93,13 +91,7 @@ def test_dyn_config_file_works(self, functional_test_setup):
 
         # "Send key" is enabled in dynamic config
         # The key should be returned here
-        query = self.client.query("test", "demo")
-        recs = query.results()
-        assert len(recs) == 1
-        # Check that record key tuple has the primary key
-        first_record = recs[0]
-        first_record_key = first_record[0]
-        assert first_record_key[2] is self.key[2]
+        expect_records_to_have_user_key_stored(self.client, set_name="demo")
 
     def test_enable_metrics_cannot_override_dyn_config(self, show_more_logs):
         config = TestBaseClass.get_connection_config()
diff --git a/test/new_tests/test_exception_subcode.py b/test/new_tests/test_exception_subcode.py
index ec3fe62c3e..9cae295823 100644
--- a/test/new_tests/test_exception_subcode.py
+++ b/test/new_tests/test_exception_subcode.py
@@ -1,5 +1,5 @@
 import pytest
-from .conftest import KEYS, BIN_NAME
+from .conftest import TEST_NS, TEST_SET, BIN_NAME
 import aerospike
 from aerospike import exception as e
 from aerospike_helpers.operations import list_operations as list_ops
@@ -7,7 +7,7 @@
 from . import as_errors
 
 
-KEY = KEYS[0]
+KEY = (TEST_NS, TEST_SET, 1)
 OPS = [
     list_ops.list_get_by_index(BIN_NAME, index=99, return_type=aerospike.LIST_RETURN_VALUE)
 ]
@@ -49,7 +49,7 @@ def test_error_verbosity_levels(self, policy_w_verbosity_setting: dict, set_in_c
             if not set_in_client_config:
                 cmd_policy |= policy_w_verbosity_setting
 
-            self.as_connection.operate(KEYS[0], OPS, policy=cmd_policy)
+            self.as_connection.operate(KEY, OPS, policy=cmd_policy)
 
         # Make sure there's no regression with the parent error code
         assert excinfo.value.code == as_errors.AEROSPIKE_ERR_OP_NOT_APPLICABLE
@@ -83,4 +83,4 @@ def test_invalid_verbosity(self):
             ERROR_DETAIL_VERBOSITY_SETTING: 3
         }
         with pytest.raises(e.ServerError):
-            self.as_connection.operate(KEYS[0], OPS, policy=policy)
+            self.as_connection.operate(KEY, OPS, policy=policy)
diff --git a/test/new_tests/test_expressions_string.py b/test/new_tests/test_expressions_string.py
index f2c7d8d7c8..e967a417e0 100644
--- a/test/new_tests/test_expressions_string.py
+++ b/test/new_tests/test_expressions_string.py
@@ -9,7 +9,8 @@
 
 from .test_base_class import TestBaseClass
 from .string_helpers import *
-from .conftest import expect_server_version_earlier_than_8_1_3_to_fail
+from .conftest import expect_server_version_earlier_than_8_1_3_to_fail, TEST_NS, TEST_SET
+KEY = (TEST_NS, TEST_SET, 1)
 
 
 class TestExpressions:
diff --git a/test/new_tests/test_query_bin_projection.py b/test/new_tests/test_query_bin_projection.py
index 5d7a94bd78..a99a380e63 100644
--- a/test/new_tests/test_query_bin_projection.py
+++ b/test/new_tests/test_query_bin_projection.py
@@ -17,6 +17,13 @@ class TestQueryBinProjection:
                 aerospike.Client.query
             ],
             indirect=True
+        ),
+        pytest.mark.parametrize(
+            "insert_records",
+            [
+                [500, False]
+            ],
+            indirect=True
         )
     ]
 
diff --git a/test/new_tests/test_query_execute_background.py b/test/new_tests/test_query_execute_background.py
index 36a9d239ac..270134adad 100644
--- a/test/new_tests/test_query_execute_background.py
+++ b/test/new_tests/test_query_execute_background.py
@@ -41,6 +41,11 @@ def validate_records(client, keys, validator):
 
 
 
+@pytest.mark.parametrize(
+    "insert_records",
+    [[500, False]],
+    indirect=True
+)
 class TestQueryApply(object):
 
     # These functions will run once for this test class, and do all of the
@@ -52,7 +57,7 @@ class TestQueryApply(object):
     def setup(self, connection_with_config_funcs):
         pass
 
-    def test_background_execute_return_val(self, clean_test_background):
+    def test_background_execute_return_val(self, insert_records):
         """
         Ensure that Query.execute_background() returns an int like object
         """
@@ -63,7 +68,7 @@ def test_background_execute_return_val(self, clean_test_background):
         assert isinstance(res, (int, long))
 
     @pytest.mark.xfail(reason="This started failing when adding support for bin projection due to query.ttl not being applied")
-    def test_background_with_ttl(self, clean_test_background):
+    def test_background_with_ttl(self, insert_records):
         """
         Ensure that ttl is set for the record found with background query
         """
@@ -87,7 +92,7 @@ def test_background_with_ttl(self, clean_test_background):
         skew_tolerance_secs = 50
         assert meta["ttl"] in range(query.ttl - skew_tolerance_secs, query.ttl + skew_tolerance_secs)
 
-    def test_background_execute_no_predicate(self, clean_test_background):
+    def test_background_execute_no_predicate(self, insert_records):
         """
         Ensure that Query.execute_background() gets applied to all records
         """
@@ -103,7 +108,7 @@ def test_background_execute_no_predicate(self, clean_test_background):
 
         validate_records(self.as_connection, keys, lambda rec: rec[test_bin] == "aerospike")
 
-    def test_background_execute_exp_everywhere(self, clean_test_background):
+    def test_background_execute_exp_everywhere(self, insert_records):
         """
         Ensure that Query.execute_background() gets applied to records that match the exp
         """
@@ -137,7 +142,7 @@ def test_background_execute_exp_everywhere(self, clean_test_background):
                 assert bins.get(test_bin) is None
 
     @pytest.mark.xfail(reason="predicate and exp used at same time")
-    def test_background_execute_exp_and_predicate(self, clean_test_background):
+    def test_background_execute_exp_and_predicate(self, insert_records):
         """
         Ensure that Query.execute_background() gets applied to records that match the predicate
         NOTE: the predicate overrides the exp
@@ -165,7 +170,7 @@ def test_background_execute_exp_and_predicate(self, clean_test_background):
             else:
                 assert bins.get(test_bin) is None
 
-    def test_background_execute_with_ops_and_exp(self, clean_test_background):
+    def test_background_execute_with_ops_and_exp(self, insert_records):
         """
         Ensure that Query.execute_background() applies ops to records that match the expressions.
         """
@@ -198,7 +203,7 @@ def test_background_execute_with_ops_and_exp(self, clean_test_background):
             else:
                 assert bins.get(test_bin) is None
 
-    def test_background_execute_with_ops(self, clean_test_background):
+    def test_background_execute_with_ops(self, insert_records):
         """
         Ensure that Query.execute_background() applies ops to all records
         """
@@ -217,7 +222,7 @@ def test_background_execute_with_ops(self, clean_test_background):
 
         validate_records(self.as_connection, keys, lambda rec: rec[test_bin] == "new_val")
 
-    def test_background_execute_with_ops_and_preds(self, clean_test_background):
+    def test_background_execute_with_ops_and_preds(self, insert_records):
         """
         Ensure that Query.execute_background() applies ops to records that match the predicate
         """
@@ -252,7 +257,7 @@ def test_background_execute_with_ops_and_preds(self, clean_test_background):
 
         validate_records(self.as_connection, keys, lambda rec: rec[test_bin] == "aerospike")
 
-    def test_background_execute_sindex_predicate(self, clean_test_background):
+    def test_background_execute_sindex_predicate(self, insert_records):
         """
         Ensure that Query.execute_background() only applies to records matched by
         the specified predicate
@@ -272,7 +277,7 @@ def test_background_execute_sindex_predicate(self, clean_test_background):
         _, _, num_5_record = self.as_connection.get((TEST_NS, TEST_SET, 5))
         assert num_5_record[test_bin] == "aerospike"
 
-    def test_background_execute_sindex_exp(self, clean_test_background):
+    def test_background_execute_sindex_exp(self, insert_records):
         """
         Ensure that Query.execute_background() only applies to records matched by
         the specified predicate
@@ -301,7 +306,7 @@ def test_background_execute_sindex_exp(self, clean_test_background):
         #  Records with number < 10 should have had the udf applied
         validate_records(self.as_connection, keys[:10], lambda rec: rec[test_bin] == "aerospike")
 
-    def test_background_execute_with_policy(self, clean_test_background):
+    def test_background_execute_with_policy(self, insert_records):
         """
         Ensure that Query.execute_background() returns an int like object
         """
@@ -311,7 +316,7 @@ def test_background_execute_with_policy(self, clean_test_background):
         res = query.execute_background({"socket_timeout": 180000})
         assert isinstance(res, (int, long))
 
-    def test_background_execute_with_policy_kwarg(self, clean_test_background):
+    def test_background_execute_with_policy_kwarg(self, insert_records):
         """
         Ensure that Query.execute_background() returns an int like object
         """
@@ -321,7 +326,7 @@ def test_background_execute_with_policy_kwarg(self, clean_test_background):
         res = query.execute_background(policy={})
         assert isinstance(res, (int, long))
 
-    def test_background_execute_with_invalid_policy_type(self, clean_test_background):
+    def test_background_execute_with_invalid_policy_type(self, insert_records):
         """
         Ensure that Query.execute_background() returns an int like object
         """
diff --git a/test/new_tests/test_send_key_union_precedence.py b/test/new_tests/test_send_key_union_precedence.py
new file mode 100644
index 0000000000..ad624451ee
--- /dev/null
+++ b/test/new_tests/test_send_key_union_precedence.py
@@ -0,0 +1,76 @@
+import pytest
+from .test_base_class import TestBaseClass
+import aerospike
+from .conftest import TEST_NS, TEST_SET, BIN_NAME, expect_records_to_have_user_key_stored, AEROSPIKE_CLIENT_CONFIG_URL, DYN_CONFIG_PATH, WRITE_OPS
+from aerospike_helpers.operations import operations
+import os
+
+
+@pytest.mark.parametrize(
+    "insert_records",
+    [[1, True]],
+    indirect=True
+)
+@pytest.mark.usefixtures("insert_records")
+class TestSendKeyUnionPrecedence:
+    @pytest.fixture(autouse=True)
+    def set_key_option(self, request):
+        self.config = TestBaseClass.get_connection_config()
+        policy_name, override_dynamic_config = request.param
+        if override_dynamic_config:
+            os.environ[AEROSPIKE_CLIENT_CONFIG_URL] = DYN_CONFIG_PATH
+        else:
+            if policy_name not in self.config["policies"]:
+                self.config["policies"][policy_name] = {}
+            self.config["policies"][policy_name]["key"] = True
+
+        yield
+
+        if override_dynamic_config:
+            del os.environ[AEROSPIKE_CLIENT_CONFIG_URL]
+
+    @pytest.mark.parametrize("set_key_option", [("write", False), ("write", True)], indirect=True)
+    def test_client_config_overrides_command_level_write_policy(self):
+        client = aerospike.client(self.config)
+
+        client.put(self.keys[0], bins={BIN_NAME: "a"}, policy={"key": aerospike.POLICY_KEY_DIGEST})
+
+        expect_records_to_have_user_key_stored(client, self.set_name)
+
+    @pytest.mark.parametrize("set_key_option", [("operate", False)], indirect=True)
+    def test_client_config_overrides_command_level_operate_policy(self):
+        client = aerospike.client(self.config)
+
+        ops = WRITE_OPS
+        client.operate(self.keys[0], ops, policy={"key": aerospike.POLICY_KEY_DIGEST})
+
+        expect_records_to_have_user_key_stored(client, self.set_name)
+
+    udf_to_load = "query_apply.lua"
+
+    @pytest.mark.parametrize("set_key_option", [("apply", False)], indirect=True)
+    def test_client_config_overrides_command_level_apply_policy(self, connection_with_udf):
+        client = aerospike.client(self.config)
+
+        client.apply(self.keys[0], "query_apply", "mark_as_applied_one_arg", ["a"], policy={"key": aerospike.POLICY_KEY_DIGEST})
+
+        expect_records_to_have_user_key_stored(client, self.set_name)
+
+    @pytest.mark.parametrize("set_key_option", [("batch_write", False), ("batch_write", True)], indirect=True)
+    def test_client_config_overrides_command_level_batch_write_policy(self):
+        client = aerospike.client(self.config)
+
+        ops = [
+            operations.write(BIN_NAME, "a")
+        ]
+        client.batch_operate([self.keys[0]], ops, policy_batch_write={"key": aerospike.POLICY_KEY_DIGEST})
+
+        expect_records_to_have_user_key_stored(client, self.set_name)
+
+    @pytest.mark.parametrize("set_key_option", [("batch_apply", False)], indirect=True)
+    def test_client_config_overrides_command_level_batch_apply_policy(self, connection_with_udf):
+        client = aerospike.client(self.config)
+
+        client.batch_apply([self.keys[0]], "query_apply", "mark_as_applied_one_arg", ["a"], policy_batch_apply={"key": aerospike.POLICY_KEY_DIGEST})
+
+        expect_records_to_have_user_key_stored(client, self.set_name)
diff --git a/test/new_tests/test_string_operations.py b/test/new_tests/test_string_operations.py
index 3ac4042e9f..66840d7e11 100644
--- a/test/new_tests/test_string_operations.py
+++ b/test/new_tests/test_string_operations.py
@@ -7,9 +7,9 @@
 from aerospike import exception as e
 from aerospike_helpers import cdt_ctx
 
-from .conftest import expect_server_version_earlier_than_8_1_3_to_fail
-from .test_base_class import TestBaseClass
+from .conftest import expect_server_version_earlier_than_8_1_3_to_fail, TEST_NS, TEST_SET
 from .string_helpers import *
+KEY = (TEST_NS, TEST_SET, 1)
 
 
 class TestStringOperations: