feat(migration): refuse batch plans with overlapping index prefixes

Two RediSearch indexes whose key prefixes overlap (one is a literal
string-prefix of the other, matching FT.CREATE PREFIX semantics) cover
the same Redis keyspace. Running a batch quantization migration over
them re-reads vectors that an earlier index in the batch has already
quantized, producing garbage bytes and corrupting data.

Reject these batches at plan time so the operator must split the
migration into disjoint groups before any vectors are touched.

- utils: add normalize_prefixes and find_overlapping_index_groups, and
  align build_scan_match_patterns with FT.CREATE PREFIX semantics
  (literal prefix + glob, no injected key separator) so the scan
  pattern matches the same keyspace the planner reasons about.
- planner: same scan-pattern fix in MigrationPlanner; bump the SCAN
  COUNT hint while we're touching the loop.
- batch_planner: collect prefixes during applicability checks and
  raise ValueError listing the overlapping (index, index, [pairs])
  before create_batch_plan returns.
- tests/unit: TestBatchMigrationPlannerOverlapDetection covers
  identical, nested, multi-prefix, and disjoint cases plus the
  applicable-only filtering rule.
- tests/integration: TestBatchMigrationOverlapDetectionIntegration
  exercises the same paths against a real Redis with FT.CREATE'd
  indexes.

Author: nkanu17

redisvl/migration/batch_planner.py

@@ -13,7 +13,12 @@
 from redisvl.index import SearchIndex
 from redisvl.migration.models import BatchIndexEntry, BatchPlan, SchemaPatch
 from redisvl.migration.planner import MigrationPlanner
-from redisvl.migration.utils import list_indexes, timestamp_utc
+from redisvl.migration.utils import (
+    find_overlapping_index_groups,
+    list_indexes,
+    normalize_prefixes,
+    timestamp_utc,
+)
 from redisvl.redis.connection import RedisConnectionFactory
 
 
@@ -83,17 +88,27 @@ def create_batch_plan(
 
         # Check applicability for each index
         batch_entries: List[BatchIndexEntry] = []
+        applicable_prefixes: List[Tuple[str, List[str]]] = []
         requires_quantization = False
 
         for index_name in index_names:
-            entry, has_quantization = self._check_index_applicability(
+            entry, has_quantization, prefixes = self._check_index_applicability(
                 index_name=index_name,
                 shared_patch=shared_patch,
                 redis_client=client,
             )
             batch_entries.append(entry)
             if has_quantization:
                 requires_quantization = True
+            if entry.applicable:
+                applicable_prefixes.append((index_name, prefixes))
+
+        # Refuse plan creation when applicable indexes share keyspace.
+        # Overlapping indexes cause double-mutation of the same keys during
+        # sequential batch execution (e.g., double-quantization of vectors).
+        overlaps = find_overlapping_index_groups(applicable_prefixes)
+        if overlaps:
+            raise ValueError(self._format_overlap_error(overlaps))
 
         batch_id = f"batch_{uuid.uuid4().hex[:12]}"
 
@@ -156,16 +171,19 @@ def _check_index_applicability(
         index_name: str,
         shared_patch: SchemaPatch,
         redis_client: Any,
-    ) -> Tuple[BatchIndexEntry, bool]:
+    ) -> Tuple[BatchIndexEntry, bool, List[str]]:
         """Check if the shared patch can be applied to a specific index.
 
         Returns:
-            Tuple of (BatchIndexEntry, requires_quantization).
+            Tuple of (BatchIndexEntry, requires_quantization, prefixes).
+            ``prefixes`` is the list of key prefixes the index is bound to,
+            or an empty list when the index could not be loaded.
         """
         try:
             index = SearchIndex.from_existing(index_name, redis_client=redis_client)
             schema_dict = index.schema.to_dict()
             field_names = {f["name"] for f in schema_dict.get("fields", [])}
+            prefixes = normalize_prefixes(schema_dict.get("index", {}).get("prefix"))
 
             # Build a set of field names that includes rename targets so
             # that update_fields referencing the NEW name of a renamed field
@@ -189,6 +207,7 @@ def _check_index_applicability(
                         skip_reason=f"Missing fields: {', '.join(missing_fields)}",
                     ),
                     False,
+                    prefixes,
                 )
 
             # Validate rename targets don't collide with each other or
@@ -212,6 +231,7 @@ def _check_index_applicability(
                             skip_reason=f"Rename targets collide: {', '.join(duplicates)}",
                         ),
                         False,
+                        prefixes,
                     )
                 # Check if any rename target already exists and isn't itself being renamed away
                 collisions = [
@@ -227,6 +247,7 @@ def _check_index_applicability(
                             skip_reason=f"Rename targets already exist: {', '.join(collisions)}",
                         ),
                         False,
+                        prefixes,
                     )
 
             # Check that add_fields don't already exist.
@@ -247,6 +268,7 @@ def _check_index_applicability(
                         skip_reason=f"Fields already exist: {', '.join(existing_adds)}",
                     ),
                     False,
+                    prefixes,
                 )
 
             # Try creating a plan to check for blocked changes
@@ -268,6 +290,7 @@ def _check_index_applicability(
                         ),
                     ),
                     False,
+                    prefixes,
                 )
 
             # Detect quantization from the plan we already created
@@ -279,7 +302,11 @@ def _check_index_applicability(
                 )
             )
 
-            return BatchIndexEntry(name=index_name, applicable=True), has_quantization
+            return (
+                BatchIndexEntry(name=index_name, applicable=True),
+                has_quantization,
+                prefixes,
+            )
 
         except (
             ConnectionError,
@@ -298,8 +325,36 @@ def _check_index_applicability(
                     skip_reason=str(e),
                 ),
                 False,
+                [],
             )
 
+    @staticmethod
+    def _format_overlap_error(
+        overlaps: List[Tuple[str, str, List[Tuple[str, str]]]],
+    ) -> str:
+        """Build a human-readable error for overlapping index prefixes."""
+        lines = [
+            "Refusing to create batch plan: overlapping indexes detected.",
+            "",
+            "Multiple indexes in the batch share Redis key prefixes. Running a",
+            "batch migration over overlapping indexes can mutate the same keys",
+            "more than once (e.g., double-quantization of vectors), corrupting",
+            "the underlying data.",
+            "",
+            "Conflicts:",
+        ]
+        for name_a, name_b, pairs in overlaps:
+            pretty_pairs = ", ".join(f"'{pa}' <-> '{pb}'" for pa, pb in pairs)
+            lines.append(f"  - {name_a} <-> {name_b}: {pretty_pairs}")
+        lines.extend(
+            [
+                "",
+                "Resolve by migrating overlapping indexes one at a time, or by",
+                "narrowing the batch to a set of indexes with disjoint prefixes.",
+            ]
+        )
+        return "\n".join(lines)
+
     def write_batch_plan(self, batch_plan: BatchPlan, path: str) -> None:
         """Write batch plan to YAML file."""
         plan_path = Path(path).resolve()

redisvl/migration/planner.py

@@ -639,16 +639,18 @@ def _sample_keys(
                 break
             if prefix == "":
                 match_pattern = "*"
-            elif prefix.endswith(key_separator):
-                match_pattern = f"{prefix}*"
             else:
-                match_pattern = f"{prefix}{key_separator}*"
+                # Use literal prefix + glob, matching Redis Search PREFIX
+                # semantics (pure string-prefix match).  Do NOT insert the
+                # key_separator — a PREFIX of "doc" must match "doc:1",
+                # "doca:1", etc., exactly like FT.CREATE does.
+                match_pattern = f"{prefix}*"
             cursor = 0
             while True:
                 cursor, keys = client.scan(
                     cursor=cursor,
                     match=match_pattern,
-                    count=max(self.key_sample_limit, 10),
+                    count=max(self.key_sample_limit, 1000),
                 )
                 for key in keys:
                     decoded_key = key.decode() if isinstance(key, bytes) else str(key)

redisvl/migration/utils.py

@@ -100,10 +100,11 @@ def build_scan_match_patterns(prefixes: List[str], key_separator: str) -> List[s
                 "Using '*' which will scan the entire keyspace."
             )
             return ["*"]
-        if key_separator and not prefix.endswith(key_separator):
-            patterns.add(f"{prefix}{key_separator}*")
-        else:
-            patterns.add(f"{prefix}*")
+        # Use literal prefix + glob, matching Redis Search PREFIX semantics
+        # (pure string-prefix match).  Do NOT insert the key_separator — a
+        # PREFIX of "doc" must match "doc:1", "doca:1", etc., exactly like
+        # FT.CREATE does.
+        patterns.add(f"{prefix}*")
     return sorted(patterns)
 
 
@@ -340,6 +341,59 @@ def timestamp_utc() -> str:
     return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
 
 
+def normalize_prefixes(prefix: Any) -> List[str]:
+    """Normalize an IndexInfo.prefix value to a list of strings."""
+    if prefix is None:
+        return []
+    if isinstance(prefix, str):
+        return [prefix]
+    if isinstance(prefix, (list, tuple)):
+        return [str(p) for p in prefix]
+    return [str(prefix)]
+
+
+def _prefixes_overlap(a: List[str], b: List[str]) -> List[Tuple[str, str]]:
+    """Return concrete (prefix_a, prefix_b) pairs whose keyspaces overlap.
+
+    Two prefixes overlap when one is a literal string-prefix of the other,
+    matching RediSearch FT.CREATE PREFIX semantics. An empty prefix matches
+    every key.
+    """
+    pairs: List[Tuple[str, str]] = []
+    for pa in a:
+        for pb in b:
+            if pa == "" or pb == "" or pa.startswith(pb) or pb.startswith(pa):
+                pairs.append((pa, pb))
+    return pairs
+
+
+def find_overlapping_index_groups(
+    indexes_with_prefixes: List[Tuple[str, List[str]]],
+) -> List[Tuple[str, str, List[Tuple[str, str]]]]:
+    """Find pairs of indexes whose key prefixes overlap.
+
+    Args:
+        indexes_with_prefixes: list of (index_name, prefixes) tuples.
+
+    Returns:
+        A list of (index_a, index_b, overlapping_prefix_pairs) tuples.
+        Empty list when no overlaps exist.
+    """
+    overlaps: List[Tuple[str, str, List[Tuple[str, str]]]] = []
+    for i in range(len(indexes_with_prefixes)):
+        name_a, prefixes_a = indexes_with_prefixes[i]
+        if not prefixes_a:
+            continue
+        for j in range(i + 1, len(indexes_with_prefixes)):
+            name_b, prefixes_b = indexes_with_prefixes[j]
+            if not prefixes_b:
+                continue
+            pairs = _prefixes_overlap(prefixes_a, prefixes_b)
+            if pairs:
+                overlaps.append((name_a, name_b, pairs))
+    return overlaps
+
+
 def estimate_disk_space(
     plan: MigrationPlan,
     *,