fix: address review round 3 for migrate-executor (#561)

- Remap datatype_changes keys to post-rename field names before quantization
- Only resume from completed checkpoint when source index is actually gone

Author: nkanu17

redisvl/migration/executor.py

@@ -464,14 +464,29 @@ def apply(
                         plan.source.index_name,
                     )
                 elif existing_checkpoint.status == "completed":
-                    # Quantization completed before the crash -- still need
-                    # to resume from post-drop state (index recreation).
-                    resuming_from_checkpoint = True
-                    logger.info(
-                        "Checkpoint at %s is already completed; resuming "
-                        "index recreation from post-drop state",
-                        checkpoint_path,
+                    # Quantization completed previously. Only resume if
+                    # the source index is actually gone (post-drop crash).
+                    # If the source still exists, this is a fresh run and
+                    # the stale checkpoint should be ignored.
+                    source_still_exists = current_source_matches_snapshot(
+                        plan.source.index_name,
+                        plan.source.schema_snapshot,
+                        redis_url=redis_url,
+                        redis_client=redis_client,
                     )
+                    if source_still_exists:
+                        logger.info(
+                            "Checkpoint at %s is completed and source index "
+                            "still exists; treating as fresh run",
+                            checkpoint_path,
+                        )
+                    else:
+                        resuming_from_checkpoint = True
+                        logger.info(
+                            "Checkpoint at %s is already completed; resuming "
+                            "index recreation from post-drop state",
+                            checkpoint_path,
+                        )
                 else:
                     resuming_from_checkpoint = True
                     logger.info(
@@ -742,9 +757,21 @@ def _notify(step: str, detail: Optional[str] = None) -> None:
                         for k in keys_to_process
                     ]
                     keys_to_process = normalize_keys(keys_to_process)
+                # Remap datatype_changes keys from source to target field
+                # names when field renames exist, since quantization runs
+                # after field renames (step 2).
+                effective_changes = datatype_changes
+                if has_field_renames and not resuming_from_checkpoint:
+                    field_rename_map = {
+                        fr.old_name: fr.new_name for fr in rename_ops.rename_fields
+                    }
+                    effective_changes = {
+                        field_rename_map.get(k, k): v
+                        for k, v in datatype_changes.items()
+                    }
                 docs_quantized = self._quantize_vectors(
                     source_index,
-                    datatype_changes,
+                    effective_changes,
                     keys_to_process,
                     progress_callback=lambda done, total: _notify(
                         "quantize", f"{done:,}/{total:,} docs"