fix(aerospike): wire batch max_retries and fix empty projection handling

Copilot review feedback on PR #6532: - Add max_retries to the batch client policy (batch_operate/batch_write path) - Treat empty projected feature maps as present FV slots (is not None) - Return {} from _normalize_projected_features([]) instead of None - Fix projection unit test mock/assertions - Correct prewriting_hook config docstring Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> Signed-off-by: Valentyn Kahamlyk <valentin.kagamlyk@gmail.com>
feast-dev · ntkathole · Jul 2, 2026 · Apr 20, 2026 · Apr 20, 2026 · Apr 20, 2026
commit 1d016f6f26405d2d5c84947bf9abd45dcd33b0a8
@@ -206,10 +206,9 @@ def hook(
 
     The hook MUST return a row list with the same shape as its input.
     Hooks that raise will fail the whole batch — there is no per-row
-    fallback. The resolved callable is cached on the store instance, so
-    swapping the config to a different hook also requires a new
-    ``AerospikeOnlineStore`` instance (which Feast already does on
-    ``RepoConfig`` change).
+    fallback. The resolved callable is cached on the store instance; if the
+    configured import string changes between calls, it is re-resolved
+    automatically on the next write.
     """
 
     user: Optional[str] = None
@@ -318,6 +317,7 @@ def _get_client(self, config: RepoConfig) -> aerospike.Client:
         }
         batch_policy: Dict[str, Any] = {
             "total_timeout": store_cfg.batch_total_timeout_ms,
+            "max_retries": store_cfg.max_retries,
         }
         if store_cfg.socket_timeout_ms is not None:
             # socket_timeout is the per-attempt deadline; without it,
@@ -666,7 +666,7 @@ def online_read(
             fv_event_ts_ms = bins.get("event_ts") if bins else None
             fv_features = self._normalize_projected_features(raw_features)
             docs[user_key] = {
-                "features": {table.name: fv_features} if fv_features else {},
+                "features": {table.name: fv_features} if fv_features is not None else {},
                 "event_timestamps": {table.name: _epoch_ms_to_datetime(fv_event_ts_ms)},
             }
 
@@ -719,7 +719,7 @@ def _normalize_projected_features(
             return raw
         if isinstance(raw, list):
             if not raw:
-                return None
+                return {}
             return dict(zip(raw[0::2], raw[1::2]))
         return None
 

@@ -578,7 +578,7 @@ def fake_batch_operate(keys, ops):
                 _fake_batch_record(
                     key,
                     {
-                        "features": ["rating", 4.91, "trips_last_7d", 132],
+                        "features": ["rating", 4.91],
                         "event_ts": _datetime_to_epoch_ms(ts),
                     },
                 )
@@ -595,8 +595,7 @@ def fake_batch_operate(keys, ops):
     ts_out, feats = results[0]
     assert ts_out == ts
     assert abs(feats["rating"].double_val - 4.91) < 1e-9
-    assert feats["trips_last_7d"].int64_val == 132
-
+    assert feats["trips_last_7d"] == ValueProto()
     features_op = captured_ops[0]
     assert features_op["op"] == aerospike.OP_MAP_GET_BY_KEY_LIST
     assert features_op["bin"] == "features"
@@ -611,7 +610,7 @@ def test_normalize_projected_features_handles_all_payload_shapes():
     """The shape of the ``features`` payload depends on which op produced it;
     the helper must accept all of them."""
     assert AerospikeOnlineStore._normalize_projected_features(None) is None
-    assert AerospikeOnlineStore._normalize_projected_features([]) is None
+    assert AerospikeOnlineStore._normalize_projected_features([]) == {}
     assert AerospikeOnlineStore._normalize_projected_features(["a", 1, "b", 2]) == {
         "a": 1,
         "b": 2,