feat(migrate): add rename operations to planner

- Add FieldRename and RenameOperations models
- Add _extract_rename_operations to detect index/prefix/field renames
- Update classify_diff to support rename detection
- Update tests for prefix change (now supported, not blocked)

Author: nkanu17

redisvl/migration/__init__.py

@@ -13,8 +13,10 @@
     BatchPlan,
     BatchReport,
     BatchState,
+    FieldRename,
     MigrationPlan,
     MigrationReport,
+    RenameOperations,
     SchemaPatch,
 )
 from redisvl.migration.planner import MigrationPlanner
@@ -29,6 +31,8 @@
     "MigrationReport",
     "MigrationValidator",
     "MigrationWizard",
+    "FieldRename",
+    "RenameOperations",
     "SchemaPatch",
     # Batch
     "BatchMigrationExecutor",

redisvl/migration/async_planner.py

@@ -92,16 +92,28 @@ async def create_plan_from_patch(
         merged_target_schema = self._sync_planner.merge_patch(
             source_schema, schema_patch
         )
+
+        # Extract rename operations first
+        rename_operations, rename_warnings = (
+            self._sync_planner._extract_rename_operations(source_schema, schema_patch)
+        )
+
+        # Classify diff with awareness of rename operations
         diff_classification = self._sync_planner.classify_diff(
-            source_schema, schema_patch, merged_target_schema
+            source_schema, schema_patch, merged_target_schema, rename_operations
         )
 
+        # Build warnings list
+        warnings = ["Index downtime is required"]
+        warnings.extend(rename_warnings)
+
         return MigrationPlan(
             source=snapshot,
             requested_changes=schema_patch.model_dump(exclude_none=True),
             merged_target_schema=merged_target_schema.to_dict(),
             diff_classification=diff_classification,
-            warnings=["Index downtime is required"],
+            rename_operations=rename_operations,
+            warnings=warnings,
         )
 
     async def snapshot_source(

redisvl/migration/models.py

@@ -24,10 +24,18 @@ def merge_options_into_attrs(self) -> "FieldUpdate":
         return self
 
 
+class FieldRename(BaseModel):
+    """Field rename specification for schema patch inputs."""
+
+    old_name: str
+    new_name: str
+
+
 class SchemaPatchChanges(BaseModel):
     add_fields: List[Dict[str, Any]] = Field(default_factory=list)
     remove_fields: List[str] = Field(default_factory=list)
     update_fields: List[FieldUpdate] = Field(default_factory=list)
+    rename_fields: List[FieldRename] = Field(default_factory=list)
     index: Dict[str, Any] = Field(default_factory=dict)
 
 
@@ -60,13 +68,26 @@ class ValidationPolicy(BaseModel):
     require_schema_match: bool = True
 
 
+class RenameOperations(BaseModel):
+    """Tracks which rename operations are required for a migration."""
+
+    rename_index: Optional[str] = None  # New index name if renaming
+    change_prefix: Optional[str] = None  # New prefix if changing
+    rename_fields: List[FieldRename] = Field(default_factory=list)
+
+    @property
+    def has_operations(self) -> bool:
+        return bool(self.rename_index or self.change_prefix or self.rename_fields)
+
+
 class MigrationPlan(BaseModel):
     version: int = 1
     mode: str = "drop_recreate"
     source: SourceSnapshot
     requested_changes: Dict[str, Any]
     merged_target_schema: Dict[str, Any]
     diff_classification: DiffClassification
+    rename_operations: RenameOperations = Field(default_factory=RenameOperations)
     warnings: List[str] = Field(default_factory=list)
     validation: ValidationPolicy = Field(default_factory=ValidationPolicy)
 
@@ -90,6 +111,8 @@ class MigrationTimings(BaseModel):
     total_migration_duration_seconds: Optional[float] = None
     drop_duration_seconds: Optional[float] = None
     quantize_duration_seconds: Optional[float] = None
+    field_rename_duration_seconds: Optional[float] = None
+    key_rename_duration_seconds: Optional[float] = None
     recreate_duration_seconds: Optional[float] = None
     initial_indexing_duration_seconds: Optional[float] = None
     validation_duration_seconds: Optional[float] = None

redisvl/migration/planner.py

@@ -2,15 +2,17 @@
 
 from copy import deepcopy
 from pathlib import Path
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Tuple
 
 import yaml
 
 from redisvl.index import SearchIndex
 from redisvl.migration.models import (
     DiffClassification,
+    FieldRename,
     KeyspaceSnapshot,
     MigrationPlan,
+    RenameOperations,
     SchemaPatch,
     SourceSnapshot,
 )
@@ -94,16 +96,28 @@ def create_plan_from_patch(
         )
         source_schema = IndexSchema.from_dict(snapshot.schema_snapshot)
         merged_target_schema = self.merge_patch(source_schema, schema_patch)
+
+        # Extract rename operations first
+        rename_operations, rename_warnings = self._extract_rename_operations(
+            source_schema, schema_patch
+        )
+
+        # Classify diff with awareness of rename operations
         diff_classification = self.classify_diff(
-            source_schema, schema_patch, merged_target_schema
+            source_schema, schema_patch, merged_target_schema, rename_operations
         )
 
+        # Build warnings list
+        warnings = ["Index downtime is required"]
+        warnings.extend(rename_warnings)
+
         return MigrationPlan(
             source=snapshot,
             requested_changes=schema_patch.model_dump(exclude_none=True),
             merged_target_schema=merged_target_schema.to_dict(),
             diff_classification=diff_classification,
-            warnings=["Index downtime is required"],
+            rename_operations=rename_operations,
+            warnings=warnings,
         )
 
     def snapshot_source(
@@ -223,29 +237,100 @@ def merge_patch(
         schema_dict["index"].update(changes.index)
         return IndexSchema.from_dict(schema_dict)
 
+    def _extract_rename_operations(
+        self,
+        source_schema: IndexSchema,
+        schema_patch: SchemaPatch,
+    ) -> Tuple[RenameOperations, List[str]]:
+        """Extract rename operations from the patch and generate warnings.
+
+        Returns:
+            Tuple of (RenameOperations, warnings list)
+        """
+        source_dict = source_schema.to_dict()
+        changes = schema_patch.changes
+        warnings: List[str] = []
+
+        # Index rename
+        rename_index: Optional[str] = None
+        if "name" in changes.index:
+            new_name = changes.index["name"]
+            old_name = source_dict["index"].get("name")
+            if new_name != old_name:
+                rename_index = new_name
+                warnings.append(
+                    f"Index rename: '{old_name}' -> '{new_name}' (index-only change, no document migration needed)"
+                )
+
+        # Prefix change
+        change_prefix: Optional[str] = None
+        if "prefix" in changes.index:
+            new_prefix = changes.index["prefix"]
+            old_prefix = source_dict["index"].get("prefix")
+            if new_prefix != old_prefix:
+                change_prefix = new_prefix
+                warnings.append(
+                    f"Prefix change: '{old_prefix}' -> '{new_prefix}' "
+                    "(requires RENAME for all keys, may be slow for large datasets)"
+                )
+
+        # Field renames from explicit rename_fields
+        rename_fields: List[FieldRename] = list(changes.rename_fields)
+        for field_rename in rename_fields:
+            warnings.append(
+                f"Field rename: '{field_rename.old_name}' -> '{field_rename.new_name}' "
+                "(requires read/write for all documents, may be slow for large datasets)"
+            )
+
+        return (
+            RenameOperations(
+                rename_index=rename_index,
+                change_prefix=change_prefix,
+                rename_fields=rename_fields,
+            ),
+            warnings,
+        )
+
     def classify_diff(
         self,
         source_schema: IndexSchema,
         schema_patch: SchemaPatch,
         merged_target_schema: IndexSchema,
+        rename_operations: Optional[RenameOperations] = None,
     ) -> DiffClassification:
         blocked_reasons: List[str] = []
         changes = schema_patch.changes
         source_dict = source_schema.to_dict()
         target_dict = merged_target_schema.to_dict()
 
+        # Check which rename operations are being handled
+        has_index_rename = rename_operations and rename_operations.rename_index
+        has_prefix_change = rename_operations and rename_operations.change_prefix
+        has_field_renames = (
+            rename_operations and len(rename_operations.rename_fields) > 0
+        )
+        renamed_field_names = set()
+        if has_field_renames and rename_operations:
+            renamed_field_names = {
+                fr.old_name for fr in rename_operations.rename_fields
+            }
+
         for index_key, target_value in changes.index.items():
             source_value = source_dict["index"].get(index_key)
             if source_value == target_value:
                 continue
             if index_key == "name":
-                blocked_reasons.append(
-                    "Changing the index name requires document migration (not yet supported)."
-                )
+                # Index rename is now supported - skip blocking if we have rename_operations
+                if not has_index_rename:
+                    blocked_reasons.append(
+                        "Changing the index name requires document migration (not yet supported)."
+                    )
             elif index_key == "prefix":
-                blocked_reasons.append(
-                    "Changing index prefixes requires document migration (not yet supported)."
-                )
+                # Prefix change is now supported
+                if not has_prefix_change:
+                    blocked_reasons.append(
+                        "Changing index prefixes requires document migration (not yet supported)."
+                    )
             elif index_key == "key_separator":
                 blocked_reasons.append(
                     "Changing the key separator requires document migration (not yet supported)."
@@ -291,9 +376,11 @@ def classify_diff(
                 )
                 blocked_reasons.extend(vector_blocked)
 
-        blocked_reasons.extend(
-            self._detect_possible_field_renames(source_fields, target_fields)
-        )
+        # Detect possible field renames only if not explicitly provided
+        if not has_field_renames:
+            blocked_reasons.extend(
+                self._detect_possible_field_renames(source_fields, target_fields)
+            )
 
         return DiffClassification(
             supported=len(blocked_reasons) == 0,

tests/unit/test_async_migration_planner.py

@@ -276,8 +276,8 @@ async def mock_from_existing(*args, **kwargs):
 
 
 @pytest.mark.asyncio
-async def test_async_planner_prefix_change_blocked(monkeypatch, tmp_path):
-    """Prefix change is blocked: documents are at wrong keys."""
+async def test_async_planner_prefix_change_is_supported(monkeypatch, tmp_path):
+    """Prefix change is supported: executor will rename keys."""
     source_schema = _make_source_schema()
     dummy_index = AsyncDummyIndex(source_schema, {"num_docs": 2}, [b"docs:1"])
 
@@ -312,8 +312,8 @@ async def mock_from_existing(*args, **kwargs):
         target_schema_path=str(target_schema_path),
     )
 
-    assert plan.diff_classification.supported is False
-    assert any(
-        "prefix" in reason.lower()
-        for reason in plan.diff_classification.blocked_reasons
-    )
+    # Prefix change is now supported
+    assert plan.diff_classification.supported is True
+    assert plan.rename_operations.change_prefix == "docs_v2"
+    # Should have a warning about key renaming
+    assert any("prefix" in w.lower() for w in plan.warnings)

tests/unit/test_migration_planner.py

@@ -278,8 +278,8 @@ def test_target_schema_vector_algorithm_change_is_allowed(monkeypatch, tmp_path)
 # =============================================================================
 
 
-def test_target_schema_prefix_change_is_blocked(monkeypatch, tmp_path):
-    """Prefix change is blocked: documents are at wrong keys."""
+def test_target_schema_prefix_change_is_supported(monkeypatch, tmp_path):
+    """Prefix change is now supported via key rename operations."""
     source_schema = _make_source_schema()
     dummy_index = DummyIndex(source_schema, {"num_docs": 2}, [b"docs:1"])
     monkeypatch.setattr(
@@ -310,11 +310,12 @@ def test_target_schema_prefix_change_is_blocked(monkeypatch, tmp_path):
         target_schema_path=str(target_schema_path),
     )
 
-    assert plan.diff_classification.supported is False
-    assert any(
-        "prefix" in reason.lower() and "iterative_shadow" in reason
-        for reason in plan.diff_classification.blocked_reasons
-    )
+    # Prefix change is now supported
+    assert plan.diff_classification.supported is True
+    # Verify rename operation is populated
+    assert plan.rename_operations.change_prefix == "docs_v2"
+    # Verify warning is present
+    assert any("Prefix change" in w for w in plan.warnings)
 
 
 def test_key_separator_change_is_blocked(monkeypatch, tmp_path):
@@ -466,7 +467,7 @@ def test_vector_dimension_change_is_blocked(monkeypatch, tmp_path):
 
     assert plan.diff_classification.supported is False
     assert any(
-        "dims" in reason and "iterative_shadow" in reason
+        "dims" in reason and "document migration" in reason
         for reason in plan.diff_classification.blocked_reasons
     )