feat(migrate): implement rename execution in sync/async executors

- Add _rename_keys for prefix changes via RENAME command
- Add _rename_field_in_hash and _rename_field_in_json for field renames
- Execute renames before drop/recreate for safe enumeration
- Support both HASH and JSON storage types

Author: nkanu17

redisvl/migration/async_executor.py

@@ -177,6 +177,120 @@ async def _enumerate_with_scan(
             if cursor == 0:
                 break
 
+    async def _rename_keys(
+        self,
+        client: AsyncRedisClient,
+        keys: List[str],
+        old_prefix: str,
+        new_prefix: str,
+        progress_callback: Optional[Callable[[int, int], None]] = None,
+    ) -> int:
+        """Async version: Rename keys from old prefix to new prefix."""
+        renamed = 0
+        total = len(keys)
+        pipeline_size = 100
+
+        for i in range(0, total, pipeline_size):
+            batch = keys[i : i + pipeline_size]
+            pipe = client.pipeline(transaction=False)
+
+            for key in batch:
+                if key.startswith(old_prefix):
+                    new_key = new_prefix + key[len(old_prefix) :]
+                else:
+                    logger.warning(
+                        f"Key '{key}' does not start with prefix '{old_prefix}'"
+                    )
+                    continue
+                pipe.rename(key, new_key)
+
+            try:
+                results = await pipe.execute()
+                renamed += sum(1 for r in results if r is True or r == "OK")
+            except Exception as e:
+                logger.warning(f"Error in rename batch: {e}")
+
+            if progress_callback:
+                progress_callback(min(i + pipeline_size, total), total)
+
+        return renamed
+
+    async def _rename_field_in_hash(
+        self,
+        client: AsyncRedisClient,
+        keys: List[str],
+        old_name: str,
+        new_name: str,
+        progress_callback: Optional[Callable[[int, int], None]] = None,
+    ) -> int:
+        """Async version: Rename a field in hash documents."""
+        renamed = 0
+        total = len(keys)
+        pipeline_size = 100
+
+        for i in range(0, total, pipeline_size):
+            batch = keys[i : i + pipeline_size]
+
+            pipe = client.pipeline(transaction=False)
+            for key in batch:
+                pipe.hget(key, old_name)
+            values = await pipe.execute()
+
+            pipe = client.pipeline(transaction=False)
+            for key, value in zip(batch, values):
+                if value is not None:
+                    pipe.hset(key, new_name, value)
+                    pipe.hdel(key, old_name)
+
+            try:
+                results = await pipe.execute()
+                renamed += sum(1 for j, r in enumerate(results) if j % 2 == 0 and r)
+            except Exception as e:
+                logger.warning(f"Error in field rename batch: {e}")
+
+            if progress_callback:
+                progress_callback(min(i + pipeline_size, total), total)
+
+        return renamed
+
+    async def _rename_field_in_json(
+        self,
+        client: AsyncRedisClient,
+        keys: List[str],
+        old_path: str,
+        new_path: str,
+        progress_callback: Optional[Callable[[int, int], None]] = None,
+    ) -> int:
+        """Async version: Rename a field in JSON documents."""
+        renamed = 0
+        total = len(keys)
+        pipeline_size = 100
+
+        for i in range(0, total, pipeline_size):
+            batch = keys[i : i + pipeline_size]
+
+            pipe = client.pipeline(transaction=False)
+            for key in batch:
+                pipe.json().get(key, old_path)
+            values = await pipe.execute()
+
+            pipe = client.pipeline(transaction=False)
+            for key, value in zip(batch, values):
+                if value is not None:
+                    pipe.json().set(key, new_path, value)
+                    pipe.json().delete(key, old_path)
+
+            try:
+                results = await pipe.execute()
+                renamed += sum(1 for j, r in enumerate(results) if j % 2 == 0 and r)
+            except Exception as e:
+                logger.warning(f"Error in JSON field rename batch: {e}")
+
+            if progress_callback:
+                progress_callback(min(i + pipeline_size, total), total)
+
+        return renamed
+
     async def apply(
         self,
         plan: MigrationPlan,
@@ -244,6 +358,8 @@ async def apply(
         enumerate_duration = 0.0
         drop_duration = 0.0
         quantize_duration = 0.0
+        field_rename_duration = 0.0
+        key_rename_duration = 0.0
         recreate_duration = 0.0
         indexing_duration = 0.0
         target_info: Dict[str, Any] = {}
@@ -254,18 +370,26 @@ async def apply(
             plan.source.schema_snapshot, plan.merged_target_schema
         )
 
+        # Check for rename operations
+        rename_ops = plan.rename_operations
+        has_prefix_change = bool(rename_ops.change_prefix)
+        has_field_renames = bool(rename_ops.rename_fields)
+        needs_enumeration = datatype_changes or has_prefix_change or has_field_renames
+
         def _notify(step: str, detail: Optional[str] = None) -> None:
             if progress_callback:
                 progress_callback(step, detail)
 
         try:
-            # STEP 1: Enumerate keys BEFORE dropping index (if quantization needed)
-            if datatype_changes:
+            client = source_index._redis_client
+            if client is None:
+                raise ValueError("Failed to get Redis client from source index")
+            storage_type = plan.source.keyspace.storage_type
+
+            # STEP 1: Enumerate keys BEFORE any modifications
+            if needs_enumeration:
                 _notify("enumerate", "Enumerating indexed documents...")
                 enumerate_started = time.perf_counter()
-                client = source_index._redis_client
-                if client is None:
-                    raise ValueError("Failed to get Redis client from source index")
                 keys_to_process = [
                     key
                     async for key in self._enumerate_indexed_keys(
@@ -278,17 +402,85 @@ def _notify(step: str, detail: Optional[str] = None) -> None:
                     f"found {len(keys_to_process):,} documents ({enumerate_duration}s)",
                 )
 
-            # STEP 2: Drop the index
+            # STEP 2: Field renames (before dropping index)
+            if has_field_renames and keys_to_process:
+                _notify("field_rename", "Renaming fields in documents...")
+                field_rename_started = time.perf_counter()
+                for field_rename in rename_ops.rename_fields:
+                    if storage_type == "json":
+                        old_path = f"$.{field_rename.old_name}"
+                        new_path = f"$.{field_rename.new_name}"
+                        await self._rename_field_in_json(
+                            client,
+                            keys_to_process,
+                            old_path,
+                            new_path,
+                            progress_callback=lambda done, total: _notify(
+                                "field_rename",
+                                f"{field_rename.old_name} -> {field_rename.new_name}: {done:,}/{total:,}",
+                            ),
+                        )
+                    else:
+                        await self._rename_field_in_hash(
+                            client,
+                            keys_to_process,
+                            field_rename.old_name,
+                            field_rename.new_name,
+                            progress_callback=lambda done, total: _notify(
+                                "field_rename",
+                                f"{field_rename.old_name} -> {field_rename.new_name}: {done:,}/{total:,}",
+                            ),
+                        )
+                field_rename_duration = round(
+                    time.perf_counter() - field_rename_started, 3
+                )
+                _notify("field_rename", f"done ({field_rename_duration}s)")
+
+            # STEP 3: Drop the index
             _notify("drop", "Dropping index definition...")
             drop_started = time.perf_counter()
             await source_index.delete(drop=False)
             drop_duration = round(time.perf_counter() - drop_started, 3)
             _notify("drop", f"done ({drop_duration}s)")
 
-            # STEP 3: Re-encode vectors using pre-enumerated keys
+            # STEP 4: Key renames (after drop, before recreate)
+            if has_prefix_change and keys_to_process:
+                _notify("key_rename", "Renaming keys...")
+                key_rename_started = time.perf_counter()
+                old_prefix = plan.source.keyspace.prefixes[0]
+                new_prefix = rename_ops.change_prefix
+                assert new_prefix is not None
+                renamed_count = await self._rename_keys(
+                    client,
+                    keys_to_process,
+                    old_prefix,
+                    new_prefix,
+                    progress_callback=lambda done, total: _notify(
+                        "key_rename", f"{done:,}/{total:,} keys"
+                    ),
+                )
+                key_rename_duration = round(time.perf_counter() - key_rename_started, 3)
+                _notify(
+                    "key_rename",
+                    f"done ({renamed_count:,} keys in {key_rename_duration}s)",
+                )
+
+            # STEP 5: Re-encode vectors using pre-enumerated keys
             if datatype_changes and keys_to_process:
                 _notify("quantize", "Re-encoding vectors...")
                 quantize_started = time.perf_counter()
+                # If we renamed keys, update keys_to_process to new names
+                if has_prefix_change and rename_ops.change_prefix:
+                    old_prefix = plan.source.keyspace.prefixes[0]
+                    new_prefix = rename_ops.change_prefix
+                    keys_to_process = [
+                        (
+                            new_prefix + k[len(old_prefix) :]
+                            if k.startswith(old_prefix)
+                            else k
+                        )
+                        for k in keys_to_process
+                    ]
                 docs_quantized = await self._async_quantize_vectors(
                     source_index,
                     datatype_changes,
@@ -341,11 +533,19 @@ def _index_progress(indexed: int, total: int, pct: float) -> None:
                 quantize_duration_seconds=(
                     quantize_duration if quantize_duration else None
                 ),
+                field_rename_duration_seconds=(
+                    field_rename_duration if field_rename_duration else None
+                ),
+                key_rename_duration_seconds=(
+                    key_rename_duration if key_rename_duration else None
+                ),
                 recreate_duration_seconds=recreate_duration,
                 initial_indexing_duration_seconds=indexing_duration,
                 validation_duration_seconds=validation_duration,
                 downtime_duration_seconds=round(
                     drop_duration
+                    + field_rename_duration
+                    + key_rename_duration
                     + quantize_duration
                     + recreate_duration
                     + indexing_duration,
@@ -368,17 +568,23 @@ def _index_progress(indexed: int, total: int, pct: float) -> None:
                 total_migration_duration_seconds=total_duration,
                 drop_duration_seconds=drop_duration or None,
                 quantize_duration_seconds=quantize_duration or None,
+                field_rename_duration_seconds=field_rename_duration or None,
+                key_rename_duration_seconds=key_rename_duration or None,
                 recreate_duration_seconds=recreate_duration or None,
                 initial_indexing_duration_seconds=indexing_duration or None,
                 downtime_duration_seconds=(
                     round(
                         drop_duration
+                        + field_rename_duration
+                        + key_rename_duration
                         + quantize_duration
                         + recreate_duration
                         + indexing_duration,
                         3,
                     )
                     if drop_duration
+                    or field_rename_duration
+                    or key_rename_duration
                     or quantize_duration
                     or recreate_duration
                     or indexing_duration