feat(migrate): add async migration support

Add async/await execution for index migrations, enabling non-blocking
operation for large quantization jobs and async application integration.

New functionality:
- CLI: --async flag for rvl migrate apply
- Python API: AsyncMigrationPlanner, AsyncMigrationExecutor, AsyncMigrationValidator
- Batched quantization with pipelined HSET operations
- Non-blocking readiness polling with asyncio.sleep()

What becomes async:
- SCAN operations (yields between batches of 500 keys)
- Pipelined HSET writes (100-1000 ops per batch)
- Index readiness polling (asyncio.sleep vs time.sleep)

What stays sync:
- CLI prompts (user interaction)
- YAML file I/O (local filesystem)

Documentation:
- Sync vs async execution guidance in concepts/index-migrations.md
- Async usage examples in how_to_guides/migrate-indexes.md

Tests:
- 4 unit tests for AsyncMigrationPlanner
- 4 unit tests for AsyncMigrationExecutor
- 1 integration test for full async flow

Author: nkanu17

docs/concepts/index-migrations.md

@@ -139,6 +139,109 @@ With `drop_recreate`, your index is unavailable between the drop and when re-ind
 
 The duration depends on document count, field count, and vector dimensions. For large indexes, consider running migrations during low traffic periods.
 
+## Sync vs async execution
+
+The migrator provides both synchronous and asynchronous execution modes.
+
+### What becomes async and what stays sync
+
+The migration workflow has distinct phases. Here is what each mode affects:
+
+| Phase | Sync mode | Async mode | Notes |
+|-------|-----------|------------|-------|
+| **Plan generation** | `MigrationPlanner.create_plan()` | `AsyncMigrationPlanner.create_plan()` | Reads index metadata from Redis |
+| **Schema snapshot** | Sync Redis calls | Async Redis calls | Single `FT.INFO` command |
+| **Drop index** | `index.delete()` | `await index.delete()` | Single `FT.DROPINDEX` command |
+| **Quantization** | Sequential SCAN + HSET | Pipelined SCAN + batched HSET | See below |
+| **Create index** | `index.create()` | `await index.create()` | Single `FT.CREATE` command |
+| **Readiness polling** | `time.sleep()` loop | `asyncio.sleep()` loop | Polls `FT.INFO` until indexed |
+| **Validation** | Sync Redis calls | Async Redis calls | Schema and doc count checks |
+| **CLI interaction** | Always sync | Always sync | User prompts, file I/O |
+| **YAML read/write** | Always sync | Always sync | Local filesystem only |
+
+### When to use sync (default)
+
+Sync execution is simpler and sufficient for most migrations:
+
+- Small to medium indexes (under 100K documents)
+- Index-only changes (algorithm, distance metric, field options)
+- Interactive CLI usage where blocking is acceptable
+
+For migrations without quantization, the Redis operations are fast single commands. Sync mode adds no meaningful overhead.
+
+### When to use async
+
+Async execution (`--async` flag) provides benefits in specific scenarios:
+
+**Large quantization jobs (1M+ vectors)**
+
+Converting float32 to float16 requires reading every vector, converting it, and writing it back. The async executor:
+
+- Uses `SCAN` with `COUNT 500` to iterate keys without blocking Redis (per [Redis SCAN docs](https://redis.io/docs/latest/commands/scan/), SCAN is O(1) per call)
+- Pipelines `HSET` operations in batches (100-1000 operations per pipeline is optimal for Redis)
+- Yields to the event loop between batches so other tasks can proceed
+
+**Large keyspaces (40M+ keys)**
+
+When your Redis instance has many keys, `SCAN` iteration can take minutes. Async mode yields between batches.
+
+**Async application integration**
+
+If your application uses asyncio, you can integrate migration directly:
+
+```python
+import asyncio
+from redisvl.migration import AsyncMigrationPlanner, AsyncMigrationExecutor
+
+async def migrate():
+    planner = AsyncMigrationPlanner()
+    plan = await planner.create_plan("myindex", redis_url="redis://localhost:6379")
+
+    executor = AsyncMigrationExecutor()
+    report = await executor.apply(plan, redis_url="redis://localhost:6379")
+
+asyncio.run(migrate())
+```
+
+### Why async helps with quantization
+
+The key difference is in the vector re-encoding loop:
+
+**Sync quantization:**
+```
+for each batch of 500 keys:
+    SCAN (blocks) -> get keys
+    for each key:
+        HGET field (blocks)
+        convert array
+        pipeline.HSET(field, new_bytes)
+    pipeline.execute() (blocks)
+```
+
+**Async quantization:**
+```
+for each batch of 500 keys:
+    await SCAN -> get keys (yields)
+    for each key:
+        await HGET field (yields)
+        convert array
+        pipeline.HSET(field, new_bytes)
+    await pipeline.execute() (yields)
+```
+
+Each `await` is a yield point where other coroutines can run. For millions of vectors, this prevents your application from freezing.
+
+### What async does NOT improve
+
+Async execution does not reduce:
+
+- **Total migration time**: Same work, different scheduling
+- **Redis server load**: Same commands execute on the server
+- **Downtime window**: Index remains unavailable during rebuild
+- **Network round trips**: Same number of Redis calls
+
+The benefit is application responsiveness, not faster migration.
+
 ## Learn more
 
 - [Migration guide](../user_guide/how_to_guides/migrate-indexes.md): Step by step instructions

docs/user_guide/how_to_guides/migrate-indexes.md

@@ -302,6 +302,41 @@ What `apply` does:
 6. validates the result
 7. writes report artifacts
 
+### Async execution for large migrations
+
+For large migrations (especially those involving vector quantization), use the `--async` flag:
+
+```bash
+rvl migrate apply \
+  --plan migration_plan.yaml \
+  --allow-downtime \
+  --async \
+  --url redis://localhost:6379
+```
+
+**What becomes async:**
+
+- Keyspace SCAN during quantization (yields between batches of 500 keys)
+- Vector read/write operations (pipelined HGET/HSET)
+- Index readiness polling (uses `asyncio.sleep()` instead of blocking)
+- Validation checks
+
+**What stays sync:**
+
+- CLI prompts and user interaction
+- YAML file reading/writing
+- Progress display
+
+**When to use async:**
+
+- Quantizing millions of vectors (float32 to float16)
+- Redis instance has 40M+ keys
+- Integrating into an async application
+
+For most migrations (index-only changes, small datasets), sync mode is sufficient and simpler.
+
+See {doc}`/concepts/index-migrations` for detailed async vs sync guidance.
+
 ## Step 5: Validate the Result
 
 Validation happens automatically during `apply`, but you can run it separately:
@@ -358,6 +393,7 @@ rvl migrate validate \
 - `--index` : Index name to migrate
 - `--plan` / `--plan-out` : Path to migration plan
 - `--allow-downtime` : Acknowledge index unavailability (required for apply)
+- `--async` : Use async executor for large migrations (apply only)
 - `--report-out` : Path for validation report
 - `--benchmark-out` : Path for performance metrics
 
@@ -389,6 +425,48 @@ If `apply` fails mid-migration:
 
 The underlying documents are never deleted by `drop_recreate`.
 
+## Python API
+
+For programmatic migrations, use the migration classes directly:
+
+### Sync API
+
+```python
+from redisvl.migration import MigrationPlanner, MigrationExecutor
+
+planner = MigrationPlanner()
+plan = planner.create_plan(
+    "myindex",
+    redis_url="redis://localhost:6379",
+    schema_patch_path="schema_patch.yaml",
+)
+
+executor = MigrationExecutor()
+report = executor.apply(plan, redis_url="redis://localhost:6379")
+print(f"Migration result: {report.result}")
+```
+
+### Async API
+
+```python
+import asyncio
+from redisvl.migration import AsyncMigrationPlanner, AsyncMigrationExecutor
+
+async def migrate():
+    planner = AsyncMigrationPlanner()
+    plan = await planner.create_plan(
+        "myindex",
+        redis_url="redis://localhost:6379",
+        schema_patch_path="schema_patch.yaml",
+    )
+
+    executor = AsyncMigrationExecutor()
+    report = await executor.apply(plan, redis_url="redis://localhost:6379")
+    print(f"Migration result: {report.result}")
+
+asyncio.run(migrate())
+```
+
 ## Learn more
 
 - {doc}`/concepts/index-migrations`: How migrations work and which changes are supported

redisvl/cli/migrate.py

@@ -1,10 +1,17 @@
 import argparse
+import asyncio
 import sys
 from argparse import Namespace
 from typing import Optional
 
 from redisvl.cli.utils import add_redis_connection_options, create_redis_url
-from redisvl.migration import MigrationExecutor, MigrationPlanner, MigrationValidator
+from redisvl.migration import (
+    AsyncMigrationExecutor,
+    AsyncMigrationValidator,
+    MigrationExecutor,
+    MigrationPlanner,
+    MigrationValidator,
+)
 from redisvl.migration.utils import (
     list_indexes,
     load_migration_plan,
@@ -26,7 +33,7 @@ class Migrate:
             "\tlist        List all available indexes",
             "\tplan        Generate a migration plan for a document-preserving drop/recreate migration",
             "\twizard      Interactively build a migration plan and schema patch",
-            "\tapply       Execute a reviewed drop/recreate migration plan",
+            "\tapply       Execute a reviewed drop/recreate migration plan (use --async for large migrations)",
             "\tvalidate    Validate a completed migration plan against the live index",
             "\n",
         ]
@@ -194,7 +201,7 @@ def apply(self):
         parser = argparse.ArgumentParser(
             usage=(
                 "rvl migrate apply --plan <migration_plan.yaml> --allow-downtime "
-                "[--report-out <migration_report.yaml>]"
+                "[--async] [--report-out <migration_report.yaml>]"
             )
         )
         parser.add_argument("--plan", help="Path to migration_plan.yaml", required=True)
@@ -203,6 +210,12 @@ def apply(self):
             help="Explicitly acknowledge downtime for drop_recreate",
             action="store_true",
         )
+        parser.add_argument(
+            "--async",
+            dest="use_async",
+            help="Use async executor (recommended for large migrations with quantization)",
+            action="store_true",
+        )
         parser.add_argument(
             "--report-out",
             help="Path to write migration_report.yaml",
@@ -228,6 +241,21 @@ def apply(self):
 
         redis_url = create_redis_url(args)
         plan = load_migration_plan(args.plan)
+
+        if args.use_async:
+            report = asyncio.run(
+                self._apply_async(plan, redis_url, args.query_check_file)
+            )
+        else:
+            report = self._apply_sync(plan, redis_url, args.query_check_file)
+
+        write_migration_report(report, args.report_out)
+        if args.benchmark_out:
+            write_benchmark_report(report, args.benchmark_out)
+        self._print_report_summary(args.report_out, report, args.benchmark_out)
+
+    def _apply_sync(self, plan, redis_url: str, query_check_file: Optional[str]):
+        """Execute migration synchronously."""
         executor = MigrationExecutor()
 
         print(f"\nApplying migration to '{plan.source.index_name}'...")
@@ -241,7 +269,6 @@ def progress_callback(step: str, detail: str) -> None:
                 "validate": "[5/5] Validate",
             }
             label = step_labels.get(step, step)
-            # Use carriage return to update in place for progress
             if detail and not detail.startswith("done"):
                 print(f"  {label}: {detail}    ", end="\r", flush=True)
             else:
@@ -250,27 +277,55 @@ def progress_callback(step: str, detail: str) -> None:
         report = executor.apply(
             plan,
             redis_url=redis_url,
-            query_check_file=args.query_check_file,
+            query_check_file=query_check_file,
             progress_callback=progress_callback,
         )
 
-        # Print completion summary
+        self._print_apply_result(report)
+        return report
+
+    async def _apply_async(self, plan, redis_url: str, query_check_file: Optional[str]):
+        """Execute migration asynchronously (non-blocking for large quantization jobs)."""
+        executor = AsyncMigrationExecutor()
+
+        print(f"\nApplying migration to '{plan.source.index_name}' (async mode)...")
+
+        def progress_callback(step: str, detail: str) -> None:
+            step_labels = {
+                "drop": "[1/5] Drop index",
+                "quantize": "[2/5] Quantize vectors",
+                "create": "[3/5] Create index",
+                "index": "[4/5] Re-indexing",
+                "validate": "[5/5] Validate",
+            }
+            label = step_labels.get(step, step)
+            if detail and not detail.startswith("done"):
+                print(f"  {label}: {detail}    ", end="\r", flush=True)
+            else:
+                print(f"  {label}: {detail}    ")
+
+        report = await executor.apply(
+            plan,
+            redis_url=redis_url,
+            query_check_file=query_check_file,
+            progress_callback=progress_callback,
+        )
+
+        self._print_apply_result(report)
+        return report
+
+    def _print_apply_result(self, report) -> None:
+        """Print the result summary after migration apply."""
         if report.result == "succeeded":
             total_time = report.timings.total_migration_duration_seconds or 0
             downtime = report.timings.downtime_duration_seconds or 0
             print(f"\nMigration completed in {total_time}s (downtime: {downtime}s)")
         else:
             print(f"\nMigration {report.result}")
-            # Show errors immediately for visibility
             if report.validation.errors:
                 for error in report.validation.errors:
                     print(f"  ERROR: {error}")
 
-        write_migration_report(report, args.report_out)
-        if args.benchmark_out:
-            write_benchmark_report(report, args.benchmark_out)
-        self._print_report_summary(args.report_out, report, args.benchmark_out)
-
     def validate(self):
         parser = argparse.ArgumentParser(
             usage=(

redisvl/migration/__init__.py

@@ -1,15 +1,32 @@
+from redisvl.migration.async_executor import AsyncMigrationExecutor
+from redisvl.migration.async_planner import AsyncMigrationPlanner
+from redisvl.migration.async_utils import (
+    async_current_source_matches_snapshot,
+    async_list_indexes,
+    async_wait_for_index_ready,
+)
+from redisvl.migration.async_validation import AsyncMigrationValidator
 from redisvl.migration.executor import MigrationExecutor
 from redisvl.migration.models import MigrationPlan, MigrationReport, SchemaPatch
 from redisvl.migration.planner import MigrationPlanner
 from redisvl.migration.validation import MigrationValidator
 from redisvl.migration.wizard import MigrationWizard
 
 __all__ = [
+    # Sync
     "MigrationExecutor",
     "MigrationPlan",
     "MigrationPlanner",
     "MigrationReport",
     "MigrationValidator",
     "MigrationWizard",
     "SchemaPatch",
+    # Async
+    "AsyncMigrationExecutor",
+    "AsyncMigrationPlanner",
+    "AsyncMigrationValidator",
+    # Async utilities
+    "async_current_source_matches_snapshot",
+    "async_list_indexes",
+    "async_wait_for_index_ready",
 ]