fix: batch planner/executor bug fixes and improvements (#563)

- Fix status mismatch: executor writes 'success' to match BatchState.success_count
- Pass rename_operations to get_vector_datatype_changes
- Validate failure_policy early (reject unknown values)
- Make update_fields applicability rename-aware
- Fix progress position during resume (correct offset)
- Fix fail-fast: leave remaining in state for checkpoint resume
- Atomic checkpoint writes (write to .tmp then rename)
- Sanitize index_name in report filenames (path traversal)
- Add assert guard for fnmatch pattern type

Author: nkanu17

redisvl/migration/batch_executor.py

@@ -80,12 +80,17 @@ def apply(
         applicable_indexes = [idx for idx in batch_plan.indexes if idx.applicable]
         total = len(applicable_indexes)
 
+        # Calculate the correct starting position for progress reporting
+        # (accounts for already-completed indexes during resume)
+        already_completed = len(state.completed)
+
         # Process each remaining index
-        for position, index_name in enumerate(state.remaining[:], start=1):
+        for offset, index_name in enumerate(state.remaining[:]):
             state.current_index = index_name
             state.updated_at = timestamp_utc()
             self._write_state(state, state_path)
 
+            position = already_completed + offset + 1
             if progress_callback:
                 progress_callback(index_name, position, total, "starting")
 
@@ -133,18 +138,8 @@ def apply(
                 index_state.status == "failed"
                 and batch_plan.failure_policy == "fail_fast"
             ):
-                # Mark remaining as skipped
-                for remaining_name in state.remaining[:]:
-                    state.remaining.remove(remaining_name)
-                    state.completed.append(
-                        BatchIndexState(
-                            name=remaining_name,
-                            status="skipped",
-                            completed_at=timestamp_utc(),
-                        )
-                    )
-                state.updated_at = timestamp_utc()
-                self._write_state(state, state_path)
+                # Leave remaining indexes in state.remaining so that
+                # checkpoint resume can pick them up later.
                 break
 
         # Build final report
@@ -225,13 +220,14 @@ def _migrate_single_index(
                 redis_client=redis_client,
             )
 
-            # Write individual report
-            report_file = report_dir / f"{index_name}_report.yaml"
+            # Sanitize index_name to prevent path traversal
+            safe_name = index_name.replace("/", "_").replace("\\", "_").replace("..", "_")
+            report_file = report_dir / f"{safe_name}_report.yaml"
             write_yaml(report.model_dump(exclude_none=True), str(report_file))
 
             return BatchIndexState(
                 name=index_name,
-                status="succeeded" if report.result == "succeeded" else "failed",
+                status="success" if report.result == "succeeded" else "failed",
                 completed_at=timestamp_utc(),
                 report_path=str(report_file),
                 error=report.validation.errors[0] if report.validation.errors else None,
@@ -277,11 +273,18 @@ def _init_or_load_state(
         )
 
     def _write_state(self, state: BatchState, state_path: str) -> None:
-        """Write checkpoint state to file."""
+        """Write checkpoint state to file atomically.
+
+        Writes to a temporary file first, then renames to avoid corruption
+        if the process crashes mid-write.
+        """
         path = Path(state_path).resolve()
         path.parent.mkdir(parents=True, exist_ok=True)
-        with open(path, "w") as f:
+        tmp_path = path.with_suffix(".tmp")
+        with open(tmp_path, "w") as f:
             yaml.safe_dump(state.model_dump(exclude_none=True), f, sort_keys=False)
+            f.flush()
+        tmp_path.replace(path)
 
     def _load_state(self, state_path: str) -> BatchState:
         """Load checkpoint state from file."""
@@ -323,13 +326,24 @@ def _build_batch_report(
                     error=idx_state.error,
                 )
             )
-            if idx_state.status == "succeeded":
+            if idx_state.status in ("succeeded", "success"):
                 succeeded += 1
             elif idx_state.status == "failed":
                 failed += 1
             else:
                 skipped += 1
 
+        # Add remaining indexes (fail-fast left them pending) as skipped
+        for remaining_name in state.remaining:
+            index_reports.append(
+                BatchIndexReport(
+                    name=remaining_name,
+                    status="skipped",
+                    error="Skipped due to fail_fast policy",
+                )
+            )
+            skipped += 1
+
         # Add non-applicable indexes as skipped
         for idx in batch_plan.indexes:
             if not idx.applicable:

redisvl/migration/batch_planner.py

@@ -37,6 +37,7 @@ def create_batch_plan(
         redis_client: Optional[Any] = None,
         failure_policy: str = "fail_fast",
     ) -> BatchPlan:
+        # --- NEW: validate failure_policy early ---
         """Create a batch migration plan for multiple indexes.
 
         Args:
@@ -51,6 +52,13 @@ def create_batch_plan(
         Returns:
             BatchPlan with shared patch and per-index applicability.
         """
+        _VALID_FAILURE_POLICIES = {"fail_fast", "continue_on_error"}
+        if failure_policy not in _VALID_FAILURE_POLICIES:
+            raise ValueError(
+                f"Invalid failure_policy '{failure_policy}'. "
+                f"Must be one of: {sorted(_VALID_FAILURE_POLICIES)}"
+            )
+
         # Get Redis client
         client = redis_client
         if client is None:
@@ -95,6 +103,7 @@ def create_batch_plan(
                     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
@@ -134,7 +143,8 @@ def _resolve_index_names(
         if indexes_file:
             return self._load_indexes_from_file(indexes_file)
 
-        # Pattern matching
+        # Pattern matching -- pattern is guaranteed non-None at this point
+        assert pattern is not None, "pattern must be set when reaching fnmatch"
         all_indexes = list_indexes(redis_client=redis_client)
         matched = [idx for idx in all_indexes if fnmatch.fnmatch(idx, pattern)]
         return sorted(matched)
@@ -167,10 +177,18 @@ def _check_index_applicability(
             schema_dict = index.schema.to_dict()
             field_names = {f["name"] for f in schema_dict.get("fields", [])}
 
-            # Check that all update_fields exist in this index
+            # Build a set of field names that includes rename targets so
+            # that update_fields referencing the NEW name of a renamed field
+            # are considered applicable.
+            rename_target_names = {
+                fr.new_name for fr in shared_patch.changes.rename_fields
+            }
+            effective_field_names = field_names | rename_target_names
+
+            # Check that all update_fields exist in this index (or are rename targets)
             missing_fields = []
             for field_update in shared_patch.changes.update_fields:
-                if field_update.name not in field_names:
+                if field_update.name not in effective_field_names:
                     missing_fields.append(field_update.name)
 
             if missing_fields: