refactor(migration): extract _quantize_array helper for dtype conversion

Move the per-vector dtype conversion logic out of the convert_vectors
loop into a dedicated _quantize_array helper. Float-to-float casts stay
a simple astype, while float-to-integer conversions apply per-vector
min-max scaling so int8/uint8 targets fill the full integer range
instead of truncating embedding values in [-1, 1] to zero.

Use TYPE_CHECKING + lazy_import so numpy is loaded lazily at runtime
while mypy still resolves np.ndarray annotations.

Adds unit coverage for float64-to-float32 and float64-to-float16 paths
and for the int8 scaling behavior across constant/varying vectors.

Author: nkanu17

redisvl/migration/quantize.py

@@ -12,9 +12,21 @@
 import math
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from dataclasses import dataclass, field
-from typing import Any, Callable, Dict, List, Optional
+from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional
 
 from redisvl.redis.utils import array_to_buffer, buffer_to_array
+from redisvl.utils.utils import lazy_import
+
+if TYPE_CHECKING:
+    import numpy as np
+else:
+    np = lazy_import("numpy")
+
+# Integer dtype ranges used for float-to-integer quantization scaling.
+_INTEGER_RANGES: Dict[str, tuple] = {
+    "int8": (-128, 127),
+    "uint8": (0, 255),
+}
 
 
 def pipeline_read_vectors(
@@ -82,12 +94,63 @@ def pipeline_write_vectors(
     pipe.execute()
 
 
+def _quantize_array(arr: "np.ndarray", target_dtype: str) -> "np.ndarray":
+    """Convert a numpy array to a target dtype, applying min-max scaling
+    when converting from float to integer types.
+
+    Float-to-float conversions (e.g. float32 → float16) are a simple cast.
+
+    Float-to-integer conversions (e.g. float32 → int8) require scaling
+    because most embedding models produce values in [-1, 1] or similar
+    narrow ranges. A naive ``astype("int8")`` would truncate everything
+    to zero. Instead, we apply per-vector min-max scaling to fill the
+    full integer range, matching the approach recommended in the Redis
+    vector-search documentation.
+
+    Args:
+        arr: Source vector as a numpy array (any float dtype).
+        target_dtype: Target dtype string (e.g. "float16", "int8", "uint8").
+
+    Returns:
+        Numpy array in the target dtype.
+
+    Raises:
+        ValueError: If the target dtype is an unsupported integer type.
+    """
+    target_lower = target_dtype.lower()
+    int_range = _INTEGER_RANGES.get(target_lower)
+
+    if int_range is None:
+        # Float-to-float: simple precision cast (e.g. float32 → float16).
+        return arr.astype(target_lower)
+
+    # Float-to-integer: per-vector min-max scaling.
+    lo, hi = int_range
+    vec_min = arr.min()
+    vec_max = arr.max()
+    spread = vec_max - vec_min
+
+    if spread == 0:
+        # Constant vector (rare but possible) — map to midpoint.
+        mid = (lo + hi) // 2
+        return np.full_like(arr, mid, dtype=target_lower)
+
+    # Scale [vec_min, vec_max] → [lo, hi] and round to nearest integer.
+    scaled = (arr - vec_min) / spread * (hi - lo) + lo
+    return np.clip(np.round(scaled), lo, hi).astype(target_lower)
+
+
 def convert_vectors(
     originals: Dict[str, Dict[str, bytes]],
     datatype_changes: Dict[str, Dict[str, Any]],
 ) -> Dict[str, Dict[str, bytes]]:
     """Convert vector bytes from source dtype to target dtype.
 
+    For float-to-float conversions, this performs a simple precision cast.
+    For float-to-integer conversions (int8, uint8), this applies per-vector
+    min-max scaling to map the float range into the full integer range
+    before casting. See :func:`_quantize_array` for details.
+
     Args:
         originals: {key: {field_name: original_bytes}}
         datatype_changes: {field_name: {"source", "target", "dims"}}
@@ -102,9 +165,13 @@ def convert_vectors(
             change = datatype_changes.get(field_name)
             if not change:
                 continue
-            array = buffer_to_array(data, change["source"])
-            new_bytes = array_to_buffer(array, change["target"])
-            converted[key][field_name] = new_bytes
+            source_dtype = change["source"].lower()
+            target_dtype = change["target"]
+
+            # Deserialize directly into numpy (avoids Python list round-trip).
+            arr = np.frombuffer(data, dtype=source_dtype).copy()
+            quantized = _quantize_array(arr, target_dtype)
+            converted[key][field_name] = quantized.tobytes()
     return converted
 
 

tests/unit/test_pipeline_quantize.py

@@ -162,3 +162,200 @@ def test_convert_float32_to_float16(self):
         # Verify values round-trip through float16
         arr = np.frombuffer(converted["doc:0"]["embedding"], dtype=np.float16)
         np.testing.assert_allclose(arr, [1.0, 2.0, 3.0, 4.0], rtol=1e-3)
+
+    def test_convert_float64_to_float32(self):
+        """Float64 to float32 should preserve values with minor precision loss."""
+        import numpy as np
+
+        from redisvl.migration.quantize import convert_vectors
+
+        dims = 4
+        source = np.array([1.0, -2.5, 3.14159265358979, 0.0], dtype=np.float64)
+        originals = {"doc:0": {"embedding": source.tobytes()}}
+        changes = {
+            "embedding": {"source": "float64", "target": "float32", "dims": dims}
+        }
+
+        converted = convert_vectors(originals, changes)
+
+        # float64 = 8 bytes/dim, float32 = 4 bytes/dim
+        assert len(converted["doc:0"]["embedding"]) == dims * 4
+        arr = np.frombuffer(converted["doc:0"]["embedding"], dtype=np.float32)
+        np.testing.assert_allclose(arr, source, rtol=1e-6)
+
+    def test_convert_float64_to_float16(self):
+        """Float64 to float16 should preserve values with larger precision loss."""
+        import numpy as np
+
+        from redisvl.migration.quantize import convert_vectors
+
+        dims = 4
+        source = np.array([1.0, -2.5, 0.333, 0.0], dtype=np.float64)
+        originals = {"doc:0": {"embedding": source.tobytes()}}
+        changes = {
+            "embedding": {"source": "float64", "target": "float16", "dims": dims}
+        }
+
+        converted = convert_vectors(originals, changes)
+
+        # float64 = 8 bytes/dim, float16 = 2 bytes/dim
+        assert len(converted["doc:0"]["embedding"]) == dims * 2
+        arr = np.frombuffer(converted["doc:0"]["embedding"], dtype=np.float16)
+        np.testing.assert_allclose(arr, source, rtol=1e-2)
+
+    def test_convert_float64_to_int8_scales_correctly(self):
+        """Float64 to int8 should apply scaling, not truncate."""
+        import numpy as np
+
+        from redisvl.migration.quantize import convert_vectors
+
+        source = np.array([-0.8, -0.2, 0.3, 0.9], dtype=np.float64)
+        originals = {"doc:0": {"embedding": source.tobytes()}}
+        changes = {"embedding": {"source": "float64", "target": "int8", "dims": 4}}
+
+        converted = convert_vectors(originals, changes)
+        result = np.frombuffer(converted["doc:0"]["embedding"], dtype=np.int8)
+
+        assert len(converted["doc:0"]["embedding"]) == 4
+        assert result.min() == -128
+        assert result.max() == 127
+        assert not np.all(result == 0)
+
+    def test_convert_float32_to_int8_scales_correctly(self):
+        """Float-to-int8 should scale values to [-128, 127], not truncate."""
+        import numpy as np
+
+        from redisvl.migration.quantize import convert_vectors
+
+        # Typical embedding values in [-1, 1] — would all become 0 without scaling.
+        dims = 4
+        source = np.array([-1.0, -0.5, 0.0, 1.0], dtype=np.float32)
+        originals = {"doc:0": {"embedding": source.tobytes()}}
+        changes = {"embedding": {"source": "float32", "target": "int8", "dims": dims}}
+
+        converted = convert_vectors(originals, changes)
+        result = np.frombuffer(converted["doc:0"]["embedding"], dtype=np.int8)
+
+        # 1 byte per dim
+        assert len(converted["doc:0"]["embedding"]) == dims * 1
+        # Min should map to -128, max to 127
+        assert result[0] == -128  # min value
+        assert result[3] == 127  # max value
+        # Values should span the full int8 range, NOT be all zeros
+        assert result.min() == -128
+        assert result.max() == 127
+        # Middle values should be proportionally scaled
+        # -0.5 → (-0.5 - (-1)) / 2 * 255 + (-128) = 63.75 - 128 = -64.25 → -64
+        assert result[1] == -64
+        # 0.0 → (0 - (-1)) / 2 * 255 + (-128) = 127.5 - 128 = -0.5 → 0
+        assert result[2] == 0
+
+    def test_convert_float16_to_int8_scales_correctly(self):
+        """Float16-to-int8 should also scale properly (the benchmark bug path)."""
+        import numpy as np
+
+        from redisvl.migration.quantize import convert_vectors
+
+        # Simulate what the benchmark did: random [0, 1] float16 vectors
+        source = np.array([0.1, 0.3, 0.7, 0.9], dtype=np.float16)
+        originals = {"doc:0": {"embedding": source.tobytes()}}
+        changes = {"embedding": {"source": "float16", "target": "int8", "dims": 4}}
+
+        converted = convert_vectors(originals, changes)
+        result = np.frombuffer(converted["doc:0"]["embedding"], dtype=np.int8)
+
+        # Should NOT be all zeros (the original bug)
+        assert not np.all(
+            result == 0
+        ), "INT8 conversion produced all zeros — scaling is not being applied"
+        # Should use the full range
+        assert result.min() == -128
+        assert result.max() == 127
+
+    def test_convert_float32_to_uint8_scales_correctly(self):
+        """Float-to-uint8 should scale values to [0, 255]."""
+        import numpy as np
+
+        from redisvl.migration.quantize import convert_vectors
+
+        source = np.array([-1.0, 0.0, 0.5, 1.0], dtype=np.float32)
+        originals = {"doc:0": {"embedding": source.tobytes()}}
+        changes = {"embedding": {"source": "float32", "target": "uint8", "dims": 4}}
+
+        converted = convert_vectors(originals, changes)
+        result = np.frombuffer(converted["doc:0"]["embedding"], dtype=np.uint8)
+
+        assert len(converted["doc:0"]["embedding"]) == 4 * 1
+        assert result[0] == 0  # min maps to 0
+        assert result[3] == 255  # max maps to 255
+        assert result.min() == 0
+        assert result.max() == 255
+
+    def test_convert_constant_vector_to_int8(self):
+        """A constant vector (all same value) should not divide by zero."""
+        import numpy as np
+
+        from redisvl.migration.quantize import convert_vectors
+
+        source = np.array([0.5, 0.5, 0.5, 0.5], dtype=np.float32)
+        originals = {"doc:0": {"embedding": source.tobytes()}}
+        changes = {"embedding": {"source": "float32", "target": "int8", "dims": 4}}
+
+        converted = convert_vectors(originals, changes)
+        result = np.frombuffer(converted["doc:0"]["embedding"], dtype=np.int8)
+
+        # Should not raise and should produce a valid int8 vector
+        assert len(result) == 4
+        # All values should be identical (mapped to midpoint)
+        assert np.all(result == result[0])
+
+    def test_convert_preserves_relative_ordering(self):
+        """Scaled int8 values should maintain the same ordering as the source."""
+        import numpy as np
+
+        from redisvl.migration.quantize import convert_vectors
+
+        source = np.array([0.1, 0.9, 0.5, 0.3, 0.7], dtype=np.float32)
+        originals = {"doc:0": {"embedding": source.tobytes()}}
+        changes = {"embedding": {"source": "float32", "target": "int8", "dims": 5}}
+
+        converted = convert_vectors(originals, changes)
+        result = np.frombuffer(converted["doc:0"]["embedding"], dtype=np.int8)
+
+        # The sorted order of indices should be preserved
+        assert list(np.argsort(source)) == list(np.argsort(result.astype(float)))
+
+    def test_convert_skips_unknown_fields(self):
+        """Fields not in datatype_changes should be skipped."""
+        from redisvl.migration.quantize import convert_vectors
+
+        originals = {"doc:0": {"other_field": b"\x00\x01"}}
+        changes = {"embedding": {"source": "float32", "target": "float16", "dims": 4}}
+
+        converted = convert_vectors(originals, changes)
+        assert converted["doc:0"] == {}
+
+    def test_convert_multiple_keys(self):
+        """Conversion should work across multiple keys in a batch."""
+        import numpy as np
+
+        from redisvl.migration.quantize import convert_vectors
+
+        v1 = np.array([0.0, 0.5, 1.0], dtype=np.float32)
+        v2 = np.array([-1.0, 0.0, 1.0], dtype=np.float32)
+        originals = {
+            "doc:0": {"embedding": v1.tobytes()},
+            "doc:1": {"embedding": v2.tobytes()},
+        }
+        changes = {"embedding": {"source": "float32", "target": "int8", "dims": 3}}
+
+        converted = convert_vectors(originals, changes)
+
+        r1 = np.frombuffer(converted["doc:0"]["embedding"], dtype=np.int8)
+        r2 = np.frombuffer(converted["doc:1"]["embedding"], dtype=np.int8)
+
+        # Each vector is scaled independently (per-vector min-max)
+        assert r1.min() == -128
+        assert r1.max() == 127
+        assert r2.min() == -128
+        assert r2.max() == 127