feat(migrate): optimize doc enumeration and simplify CLI

Document Enumeration Optimization:
- Use FT.AGGREGATE WITHCURSOR for efficient key enumeration
- Falls back to SCAN only when index has hash_indexing_failures
- Pre-enumerate keys before drop for reliable re-indexing

CLI Simplification:
- Remove redundant --allow-downtime flag from apply/batch-apply
- Plan review is now the safety mechanism

Batch Migration:
- Add BatchMigrationExecutor and BatchMigrationPlanner
- Support for multi-index migration with failure policies
- Resumable batch operations with state persistence

Bug Fixes:
- Fix mypy type errors in planner, wizard, validation, and CLI

Documentation:
- Update concepts and how-to guides for new workflow
- Remove --allow-downtime references from all docs

Author: nkanu17

docs/concepts/index-migrations.md

@@ -56,7 +56,7 @@ The process:
 4. Wait for Redis to re-index the existing documents
 5. Validate the result
 
-**Tradeoff**: The index is unavailable during the rebuild. The migrator requires explicit acknowledgment of this downtime before proceeding.
+**Tradeoff**: The index is unavailable during the rebuild. Review the migration plan carefully before applying.
 
 ## Index only vs document dependent changes
 
@@ -130,7 +130,16 @@ Adding a vector field means all existing documents need vectors for that field.
 
 ## Downtime considerations
 
-With `drop_recreate`, your index is unavailable between the drop and when re-indexing completes. Plan for:
+c lWith `drop_recreate`, your index is unavailable between the drop and when re-indexing completes.
+
+**CRITICAL**: Downtime requires both reads AND writes to be paused:
+
+| Requirement | Reason |
+|-------------|--------|
+| **Pause reads** | Index is unavailable during migration |
+| **Pause writes** | Redis updates indexes synchronously. Writes during migration may conflict with vector re-encoding or be missed |
+
+Plan for:
 
 - Search unavailability during the migration window
 - Partial results while indexing is in progress
@@ -151,8 +160,9 @@ The migration workflow has distinct phases. Here is what each mode affects:
 |-------|-----------|------------|-------|
 | **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 |
+| **Enumeration** | FT.AGGREGATE (or SCAN fallback) | FT.AGGREGATE (or SCAN fallback) | Before drop, only if quantization needed |
 | **Drop index** | `index.delete()` | `await index.delete()` | Single `FT.DROPINDEX` command |
-| **Quantization** | Sequential SCAN + HSET | Pipelined SCAN + batched HSET | See below |
+| **Quantization** | Sequential HGET + HSET | Pipelined HGET + batched HSET | Uses pre-enumerated keys |
 | **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 |
@@ -177,13 +187,13 @@ Async execution (`--async` flag) provides benefits in specific scenarios:
 
 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)
+- Enumerates documents using `FT.AGGREGATE WITHCURSOR` for index-specific enumeration (falls back to `SCAN` only if indexing failures exist)
 - 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.
+When your Redis instance has many keys and the index has indexing failures (requiring SCAN fallback), async mode yields between batches.
 
 **Async application integration**
 
@@ -205,12 +215,18 @@ asyncio.run(migrate())
 
 ### Why async helps with quantization
 
-The key difference is in the vector re-encoding loop:
+The migrator uses an optimized enumeration strategy:
+
+1. **Index-based enumeration**: Uses `FT.AGGREGATE WITHCURSOR` to enumerate only indexed documents (not the entire keyspace)
+2. **Fallback for safety**: If the index has indexing failures (`hash_indexing_failures > 0`), falls back to `SCAN` to ensure completeness
+3. **Enumerate before drop**: Captures the document list while the index still exists, then drops and quantizes
+
+This optimization provides 10-1000x speedup for sparse indexes (where only a small fraction of prefix-matching keys are indexed).
 
 **Sync quantization:**
 ```
+enumerate keys (FT.AGGREGATE or SCAN) -> store list
 for each batch of 500 keys:
-    SCAN (blocks) -> get keys
     for each key:
         HGET field (blocks)
         convert array
@@ -220,8 +236,8 @@ for each batch of 500 keys:
 
 **Async quantization:**
 ```
+enumerate keys (FT.AGGREGATE or SCAN) -> store list
 for each batch of 500 keys:
-    await SCAN -> get keys (yields)
     for each key:
         await HGET field (yields)
         convert array

docs/user_guide/cli.ipynb

@@ -54,7 +54,7 @@
     "| `rvl migrate` | `helper` or `list`                     | show migration guidance and list indexes available for migration|\n",
     "| `rvl migrate` | `wizard`                               | interactively build a migration plan and schema patch|\n",
     "| `rvl migrate` | `plan`                                 | generate `migration_plan.yaml` from a patch or target schema|\n",
-    "| `rvl migrate` | `apply --allow-downtime`               | execute a reviewed `drop_recreate` migration|\n",
+    "| `rvl migrate` | `apply`                                | execute a reviewed `drop_recreate` migration|\n",
     "| `rvl migrate` | `validate`                             | validate a completed migration and emit report artifacts|"
    ]
   },

docs/user_guide/how_to_guides/migrate-indexes.md

@@ -21,7 +21,7 @@ rvl migrate list --url redis://localhost:6379
 rvl migrate wizard --index myindex --url redis://localhost:6379
 
 # 3. Apply the migration
-rvl migrate apply --plan migration_plan.yaml --allow-downtime --url redis://localhost:6379
+rvl migrate apply --plan migration_plan.yaml --url redis://localhost:6379
 
 # 4. Verify the result
 rvl migrate validate --plan migration_plan.yaml --url redis://localhost:6379
@@ -266,14 +266,45 @@ merged_target_schema:
 - `diff_classification.blocked_reasons` - Must be empty
 - `merged_target_schema` - The final schema after migration
 
+## Understanding Downtime Requirements
+
+**CRITICAL**: During a `drop_recreate` migration, your application must:
+
+| Requirement | Description |
+|-------------|-------------|
+| **Pause reads** | Index is unavailable during migration |
+| **Pause writes** | Writes during migration may be missed or cause conflicts |
+
+### Why Both Reads AND Writes Must Be Paused
+
+- **Reads**: The index definition is dropped and recreated. Any queries during this window will fail.
+- **Writes**: Redis updates indexes synchronously on every write. If your app writes documents while the index is dropped, those writes are not indexed. Additionally, if you're quantizing vectors (float32 → float16), concurrent writes may conflict with the migration's re-encoding process.
+
+### What "Downtime" Means
+
+| Downtime Type | Reads | Writes | Safe? |
+|---------------|-------|--------|-------|
+| Full quiesce (recommended) | Stopped | Stopped | **YES** |
+| Read-only pause | Stopped | Continuing | **NO** |
+| Active | Active | Active | **NO** |
+
+### Recovery from Interrupted Migration
+
+| Interruption Point | Documents | Index | Recovery |
+|--------------------|-----------|-------|----------|
+| After drop, before quantize | Unchanged | **None** | Re-run apply |
+| After quantization, before create | Quantized | **None** | Manual FT.CREATE or re-run apply |
+| After create | Correct | Rebuilding | Wait for index ready |
+
+The underlying documents are **never deleted** by `drop_recreate` mode.
+
 ## Step 4: Apply the Migration
 
-The `apply` command requires `--allow-downtime` since the index will be temporarily unavailable.
+The `apply` command executes the migration. The index will be temporarily unavailable during the drop-recreate process.
 
 ```bash
 rvl migrate apply \
   --plan migration_plan.yaml \
-  --allow-downtime \
   --url redis://localhost:6379 \
   --report-out migration_report.yaml \
   --benchmark-out benchmark_report.yaml
@@ -296,14 +327,13 @@ For large migrations (especially those involving vector quantization), use the `
 ```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)
+- Document enumeration during quantization (uses `FT.AGGREGATE WITHCURSOR` for index-specific enumeration, falling back to SCAN only if indexing failures exist)
 - Vector read/write operations (pipelined HGET/HSET)
 - Index readiness polling (uses `asyncio.sleep()` instead of blocking)
 - Validation checks
@@ -388,7 +418,6 @@ rvl migrate validate \
 - `--url` : Redis connection URL
 - `--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
@@ -504,7 +533,6 @@ rvl migrate batch-plan \
 # 3. Apply the batch plan
 rvl migrate batch-apply \
   --plan batch_plan.yaml \
-  --allow-downtime \
   --accept-data-loss \
   --url redis://localhost:6379
 
@@ -587,26 +615,31 @@ created_at: "2026-03-20T10:00:00Z"
 # Apply with fail-fast (default: stop on first error)
 rvl migrate batch-apply \
   --plan batch_plan.yaml \
-  --allow-downtime \
   --accept-data-loss \
   --url redis://localhost:6379
 
-# Apply with continue-on-error (process all possible indexes)
+# Apply with continue-on-error (set at batch-plan time)
+# Note: failure_policy is set during batch-plan, not batch-apply
+rvl migrate batch-plan \
+  --pattern "*_idx" \
+  --schema-patch quantize_patch.yaml \
+  --failure-policy continue_on_error \
+  --output batch_plan.yaml \
+  --url redis://localhost:6379
+
 rvl migrate batch-apply \
   --plan batch_plan.yaml \
-  --allow-downtime \
   --accept-data-loss \
-  --failure-policy continue_on_error \
   --url redis://localhost:6379
 ```
 
-**Flags:**
-- `--allow-downtime` : Required (each index is temporarily unavailable during migration)
+**Flags for batch-apply:**
 - `--accept-data-loss` : Required when quantizing vectors (float32 → float16 is lossy)
-- `--failure-policy` : `fail_fast` (default) or `continue_on_error`
 - `--state` : Path to checkpoint file (default: `batch_state.yaml`)
 - `--report-dir` : Directory for per-index reports (default: `./reports/`)
 
+**Note:** `--failure-policy` is set during `batch-plan`, not `batch-apply`. The policy is stored in the batch plan file.
+
 ### Resume After Failure
 
 Batch migration automatically checkpoints progress. If interrupted:
@@ -615,14 +648,12 @@ Batch migration automatically checkpoints progress. If interrupted:
 # Resume from where it left off
 rvl migrate batch-resume \
   --state batch_state.yaml \
-  --allow-downtime \
   --url redis://localhost:6379
 
 # Retry previously failed indexes
 rvl migrate batch-resume \
   --state batch_state.yaml \
   --retry-failed \
-  --allow-downtime \
   --url redis://localhost:6379
 ```
 
@@ -686,7 +717,7 @@ from redisvl.migration import BatchMigrationPlanner, BatchMigrationExecutor
 
 # Create batch plan
 planner = BatchMigrationPlanner()
-batch_plan = planner.create_plan(
+batch_plan = planner.create_batch_plan(
     redis_url="redis://localhost:6379",
     pattern="*_idx",
     schema_patch_path="quantize_patch.yaml",

redisvl/cli/migrate.py

@@ -19,7 +19,6 @@
     load_yaml,
     write_benchmark_report,
     write_migration_report,
-    write_yaml,
 )
 from redisvl.migration.wizard import MigrationWizard
 from redisvl.utils.log import get_logger
@@ -105,7 +104,7 @@ def helper(self):
   rvl migrate list                                  List all indexes
   rvl migrate wizard --index <name>                 Guided migration builder
   rvl migrate plan --index <name> --schema-patch <patch.yaml>
-  rvl migrate apply --plan <plan.yaml> --allow-downtime
+  rvl migrate apply --plan <plan.yaml>
   rvl migrate validate --plan <plan.yaml>"""
         )
 
@@ -211,16 +210,11 @@ def wizard(self):
     def apply(self):
         parser = argparse.ArgumentParser(
             usage=(
-                "rvl migrate apply --plan <migration_plan.yaml> --allow-downtime "
+                "rvl migrate apply --plan <migration_plan.yaml> "
                 "[--async] [--report-out <migration_report.yaml>]"
             )
         )
         parser.add_argument("--plan", help="Path to migration_plan.yaml", required=True)
-        parser.add_argument(
-            "--allow-downtime",
-            help="Explicitly acknowledge downtime for drop_recreate",
-            action="store_true",
-        )
         parser.add_argument(
             "--async",
             dest="use_async",
@@ -245,11 +239,6 @@ def apply(self):
         parser = add_redis_connection_options(parser)
         args = parser.parse_args(sys.argv[3:])
 
-        if not args.allow_downtime:
-            raise ValueError(
-                "apply requires --allow-downtime for drop_recreate migrations"
-            )
-
         redis_url = create_redis_url(args)
         plan = load_migration_plan(args.plan)
 
@@ -271,7 +260,7 @@ def _apply_sync(self, plan, redis_url: str, query_check_file: Optional[str]):
 
         print(f"\nApplying migration to '{plan.source.index_name}'...")
 
-        def progress_callback(step: str, detail: str) -> None:
+        def progress_callback(step: str, detail: Optional[str]) -> None:
             step_labels = {
                 "drop": "[1/5] Drop index",
                 "quantize": "[2/5] Quantize vectors",
@@ -301,7 +290,7 @@ async def _apply_async(self, plan, redis_url: str, query_check_file: Optional[st
 
         print(f"\nApplying migration to '{plan.source.index_name}' (async mode)...")
 
-        def progress_callback(step: str, detail: str) -> None:
+        def progress_callback(step: str, detail: Optional[str]) -> None:
             step_labels = {
                 "drop": "[1/5] Drop index",
                 "quantize": "[2/5] Quantize vectors",
@@ -427,9 +416,7 @@ def _print_plan_summary(self, plan_out: str, plan) -> None:
 
         print("\nNext steps:")
         print(f"  Review the plan:     cat {plan_out}")
-        print(
-            f"  Apply the migration: rvl migrate apply --plan {plan_out} --allow-downtime"
-        )
+        print(f"  Apply the migration: rvl migrate apply --plan {plan_out}")
         print(f"  Validate the result: rvl migrate validate --plan {plan_out}")
         print(
             f"\nTo add more changes:   rvl migrate wizard --index {plan.source.index_name} --patch schema_patch.yaml"
@@ -518,16 +505,11 @@ def batch_apply(self):
         """Execute a batch migration plan with checkpointing."""
         parser = argparse.ArgumentParser(
             usage=(
-                "rvl migrate batch-apply --plan <batch_plan.yaml> --allow-downtime "
+                "rvl migrate batch-apply --plan <batch_plan.yaml> "
                 "[--state <batch_state.yaml>] [--report-dir <./reports>]"
             )
         )
         parser.add_argument("--plan", help="Path to batch_plan.yaml", required=True)
-        parser.add_argument(
-            "--allow-downtime",
-            help="Explicitly acknowledge downtime for drop_recreate",
-            action="store_true",
-        )
         parser.add_argument(
             "--accept-data-loss",
             help="Acknowledge that quantization is lossy and cannot be reverted",
@@ -546,11 +528,6 @@ def batch_apply(self):
         parser = add_redis_connection_options(parser)
         args = parser.parse_args(sys.argv[3:])
 
-        if not args.allow_downtime:
-            raise ValueError(
-                "batch-apply requires --allow-downtime for drop_recreate migrations"
-            )
-
         # Load batch plan
         from redisvl.migration.models import BatchPlan
 
@@ -701,7 +678,7 @@ def _print_batch_plan_summary(self, plan_out: str, batch_plan) -> None:
             f"""
 Next steps:
   Review the plan:      cat {plan_out}
-  Apply the migration:  rvl migrate batch-apply --plan {plan_out} --allow-downtime"""
+  Apply the migration:  rvl migrate batch-apply --plan {plan_out}"""
         )
 
         if batch_plan.requires_quantization: