Add preserve_bytes and binary_fields parameters to fix UTF-8 encoding corruption in search results

- Add to_string_or_bytes function in _util.py for selective binary preservation
- Update Result class to accept preserve_bytes and binary_fields parameters
- Update search methods in SearchCommands and AsyncSearchCommands
- Fix field processing map object consumption issue
- Maintain backward compatibility with default behavior
- Use approximate float comparison in tests for floating-point precision

Author: swarnaprakash

valkey/commands/search/_util.py

@@ -5,3 +5,15 @@ def to_string(s):
         return s.decode("utf-8", "ignore")
     else:
         return s  # Not a string we care about
+
+
+def to_string_or_bytes(s, preserve_bytes=False, binary_fields=None, field_name=None):
+    """Convert value to string or preserve as bytes based on parameters."""
+    if isinstance(s, str):
+        return s
+    elif isinstance(s, bytes):
+        if preserve_bytes and (binary_fields is None or field_name in binary_fields):
+            return s  # Keep as bytes
+        return s.decode("utf-8", "ignore")
+    else:
+        return s  # Not a string we care about

valkey/commands/search/commands.py

@@ -80,6 +80,8 @@ def _parse_search(self, res, **kwargs):
             duration=kwargs["duration"],
             has_payload=kwargs["query"]._with_payloads,
             with_scores=kwargs["query"]._with_scores,
+            preserve_bytes=kwargs.get("preserve_bytes", False),
+            binary_fields=kwargs.get("binary_fields", None),
         )
 
     def _parse_aggregate(self, res, **kwargs):
@@ -96,6 +98,8 @@ def _parse_profile(self, res, **kwargs):
                 duration=kwargs["duration"],
                 has_payload=query._with_payloads,
                 with_scores=query._with_scores,
+                preserve_bytes=kwargs.get("preserve_bytes", False),
+                binary_fields=kwargs.get("binary_fields", None),
             )
 
         return result, parse_to_dict(res[1])
@@ -484,6 +488,8 @@ def search(
         self,
         query: Union[str, Query],
         query_params: Union[Dict[str, Union[str, int, float, bytes]], None] = None,
+        preserve_bytes: bool = False,
+        binary_fields: Optional[List[str]] = None,
     ):
         """
         Search the index for a given query, and return a result of documents
@@ -493,6 +499,11 @@ def search(
         - **query**: the search query. Either a text for simple queries with
                      default parameters, or a Query object for complex queries.
                      See RediSearch's documentation on query format
+        - **preserve_bytes**: If True, preserve binary field values as bytes
+                             instead of converting to UTF-8 strings
+        - **binary_fields**: List of field names to preserve as bytes when
+                            preserve_bytes=True. If None, all binary fields
+                            are preserved
 
         For more information see `FT.SEARCH <https://valkey.io/commands/ft.search>`_.
         """  # noqa
@@ -504,7 +515,8 @@ def search(
             return res
 
         return self._parse_results(
-            SEARCH_CMD, res, query=query, duration=(time.time() - st) * 1000.0
+            SEARCH_CMD, res, query=query, duration=(time.time() - st) * 1000.0,
+            preserve_bytes=preserve_bytes, binary_fields=binary_fields
         )
 
     def explain(
@@ -911,6 +923,8 @@ async def search(
         self,
         query: Union[str, Query],
         query_params: Dict[str, Union[str, int, float]] = None,
+        preserve_bytes: bool = False,
+        binary_fields: Optional[List[str]] = None,
     ):
         """
         Search the index for a given query, and return a result of documents
@@ -920,6 +934,11 @@ async def search(
         - **query**: the search query. Either a text for simple queries with
                      default parameters, or a Query object for complex queries.
                      See RediSearch's documentation on query format
+        - **preserve_bytes**: If True, preserve binary field values as bytes
+                             instead of converting to UTF-8 strings
+        - **binary_fields**: List of field names to preserve as bytes when
+                            preserve_bytes=True. If None, all binary fields
+                            are preserved
 
         For more information see `FT.SEARCH <https://valkey.io/commands/ft.search>`_.
         """  # noqa
@@ -931,7 +950,8 @@ async def search(
             return res
 
         return self._parse_results(
-            SEARCH_CMD, res, query=query, duration=(time.time() - st) * 1000.0
+            SEARCH_CMD, res, query=query, duration=(time.time() - st) * 1000.0,
+            preserve_bytes=preserve_bytes, binary_fields=binary_fields
         )
 
     async def aggregate(

valkey/commands/search/result.py

@@ -1,4 +1,4 @@
-from ._util import to_string
+from ._util import to_string, to_string_or_bytes
 from .document import Document
 
 
@@ -9,7 +9,8 @@ class Result:
     """
 
     def __init__(
-        self, res, hascontent, duration=0, has_payload=False, with_scores=False
+        self, res, hascontent, duration=0, has_payload=False, with_scores=False,
+        preserve_bytes=False, binary_fields=None
     ):
         """
         - **snippets**: An optional dictionary of the form
@@ -39,18 +40,19 @@ def __init__(
 
             fields = {}
             if hascontent and res[i + fields_offset] is not None:
-                fields = (
-                    dict(
-                        dict(
-                            zip(
-                                map(to_string, res[i + fields_offset][::2]),
-                                map(to_string, res[i + fields_offset][1::2]),
-                            )
-                        )
+                field_names = list(map(to_string, res[i + fields_offset][::2]))
+                field_values = res[i + fields_offset][1::2]
+                
+                # Process field values with binary preservation
+                processed_values = []
+                for field_name, field_value in zip(field_names, field_values):
+                    processed_value = to_string_or_bytes(
+                        field_value, preserve_bytes, binary_fields, field_name
                     )
-                    if hascontent
-                    else {}
-                )
+                    processed_values.append(processed_value)
+                
+                fields = dict(zip(field_names, processed_values))
+                
             try:
                 del fields["id"]
             except KeyError: