fix: eliminate double plan creation in batch planner, use sys.exit in CLI

- Refactor _check_index_applicability to return Tuple[BatchIndexEntry, bool]
  where bool indicates quantization, avoiding redundant create_plan_from_patch
- Replace exit(1) with sys.exit(1) in batch-apply and batch-resume CLI commands
- Sanitize report filenames (colons to underscores) for Windows compat

Author: nkanu17

redisvl/cli/migrate.py

@@ -597,7 +597,7 @@ def batch_apply(self):
          If you need to preserve original vectors, backup your data first:
            redis-cli BGSAVE"""
             )
-            exit(1)
+            sys.exit(1)
 
         redis_url = create_redis_url(args)
         executor = BatchMigrationExecutor()
@@ -677,7 +677,7 @@ def batch_status(self):
         state_path = Path(args.state).resolve()
         if not state_path.exists():
             print(f"State file not found: {args.state}")
-            exit(1)
+            sys.exit(1)
 
         from redisvl.migration.models import BatchState
 

redisvl/migration/batch_executor.py

@@ -220,9 +220,12 @@ def _migrate_single_index(
                 redis_client=redis_client,
             )
 
-            # Sanitize index_name to prevent path traversal
+            # Sanitize index_name to prevent path traversal and invalid filenames
             safe_name = (
-                index_name.replace("/", "_").replace("\\", "_").replace("..", "_")
+                index_name.replace("/", "_")
+                .replace("\\", "_")
+                .replace("..", "_")
+                .replace(":", "_")
             )
             report_file = report_dir / f"{safe_name}_report.yaml"
             write_yaml(report.model_dump(exclude_none=True), str(report_file))

redisvl/migration/batch_planner.py

@@ -5,7 +5,7 @@
 import fnmatch
 import uuid
 from pathlib import Path
-from typing import Any, List, Optional
+from typing import Any, List, Optional, Tuple
 
 import redis.exceptions
 import yaml
@@ -86,30 +86,14 @@ def create_batch_plan(
         requires_quantization = False
 
         for index_name in index_names:
-            entry = self._check_index_applicability(
+            entry, has_quantization = self._check_index_applicability(
                 index_name=index_name,
                 shared_patch=shared_patch,
                 redis_client=client,
             )
             batch_entries.append(entry)
-
-            # Check if any applicable index requires quantization
-            if entry.applicable:
-                try:
-                    plan = self._single_planner.create_plan_from_patch(
-                        index_name,
-                        schema_patch=shared_patch,
-                        redis_client=client,
-                    )
-                    datatype_changes = MigrationPlanner.get_vector_datatype_changes(
-                        plan.source.schema_snapshot,
-                        plan.merged_target_schema,
-                        rename_operations=plan.rename_operations,
-                    )
-                    if datatype_changes:
-                        requires_quantization = True
-                except Exception:
-                    pass  # Already handled in applicability check
+            if has_quantization:
+                requires_quantization = True
 
         batch_id = f"batch_{uuid.uuid4().hex[:12]}"
 
@@ -171,8 +155,12 @@ def _check_index_applicability(
         index_name: str,
         shared_patch: SchemaPatch,
         redis_client: Any,
-    ) -> BatchIndexEntry:
-        """Check if the shared patch can be applied to a specific index."""
+    ) -> Tuple[BatchIndexEntry, bool]:
+        """Check if the shared patch can be applied to a specific index.
+
+        Returns:
+            Tuple of (BatchIndexEntry, requires_quantization).
+        """
         try:
             index = SearchIndex.from_existing(index_name, redis_client=redis_client)
             schema_dict = index.schema.to_dict()
@@ -193,10 +181,13 @@ def _check_index_applicability(
                     missing_fields.append(field_update.name)
 
             if missing_fields:
-                return BatchIndexEntry(
-                    name=index_name,
-                    applicable=False,
-                    skip_reason=f"Missing fields: {', '.join(missing_fields)}",
+                return (
+                    BatchIndexEntry(
+                        name=index_name,
+                        applicable=False,
+                        skip_reason=f"Missing fields: {', '.join(missing_fields)}",
+                    ),
+                    False,
                 )
 
             # Validate rename targets don't collide with each other or
@@ -213,10 +204,13 @@ def _check_index_applicability(
                     seen_targets[t] = seen_targets.get(t, 0) + 1
                 duplicates = [t for t, c in seen_targets.items() if c > 1]
                 if duplicates:
-                    return BatchIndexEntry(
-                        name=index_name,
-                        applicable=False,
-                        skip_reason=f"Rename targets collide: {', '.join(duplicates)}",
+                    return (
+                        BatchIndexEntry(
+                            name=index_name,
+                            applicable=False,
+                            skip_reason=f"Rename targets collide: {', '.join(duplicates)}",
+                        ),
+                        False,
                     )
                 # Check if any rename target already exists and isn't itself being renamed away
                 collisions = [
@@ -225,10 +219,13 @@ def _check_index_applicability(
                     if t in field_names and t not in rename_sources
                 ]
                 if collisions:
-                    return BatchIndexEntry(
-                        name=index_name,
-                        applicable=False,
-                        skip_reason=f"Rename targets already exist: {', '.join(collisions)}",
+                    return (
+                        BatchIndexEntry(
+                            name=index_name,
+                            applicable=False,
+                            skip_reason=f"Rename targets already exist: {', '.join(collisions)}",
+                        ),
+                        False,
                     )
 
             # Check that add_fields don't already exist.
@@ -242,10 +239,13 @@ def _check_index_applicability(
                     existing_adds.append(field_name)
 
             if existing_adds:
-                return BatchIndexEntry(
-                    name=index_name,
-                    applicable=False,
-                    skip_reason=f"Fields already exist: {', '.join(existing_adds)}",
+                return (
+                    BatchIndexEntry(
+                        name=index_name,
+                        applicable=False,
+                        skip_reason=f"Fields already exist: {', '.join(existing_adds)}",
+                    ),
+                    False,
                 )
 
             # Try creating a plan to check for blocked changes
@@ -256,17 +256,29 @@ def _check_index_applicability(
             )
 
             if not plan.diff_classification.supported:
-                return BatchIndexEntry(
-                    name=index_name,
-                    applicable=False,
-                    skip_reason=(
-                        plan.diff_classification.blocked_reasons[0]
-                        if plan.diff_classification.blocked_reasons
-                        else "Unsupported changes"
+                return (
+                    BatchIndexEntry(
+                        name=index_name,
+                        applicable=False,
+                        skip_reason=(
+                            plan.diff_classification.blocked_reasons[0]
+                            if plan.diff_classification.blocked_reasons
+                            else "Unsupported changes"
+                        ),
                     ),
+                    False,
                 )
 
-            return BatchIndexEntry(name=index_name, applicable=True)
+            # Detect quantization from the plan we already created
+            has_quantization = bool(
+                MigrationPlanner.get_vector_datatype_changes(
+                    plan.source.schema_snapshot,
+                    plan.merged_target_schema,
+                    rename_operations=plan.rename_operations,
+                )
+            )
+
+            return BatchIndexEntry(name=index_name, applicable=True), has_quantization
 
         except (
             ConnectionError,
@@ -278,10 +290,13 @@ def _check_index_applicability(
             # treated as "not applicable".
             raise
         except Exception as e:
-            return BatchIndexEntry(
-                name=index_name,
-                applicable=False,
-                skip_reason=str(e),
+            return (
+                BatchIndexEntry(
+                    name=index_name,
+                    applicable=False,
+                    skip_reason=str(e),
+                ),
+                False,
             )
 
     def write_batch_plan(self, batch_plan: BatchPlan, path: str) -> None: