feat(migrate): [2/6] Core sync executor and basic CLI#568
feat(migrate): [2/6] Core sync executor and basic CLI#568nkanu17 wants to merge 2 commits intofeat/migrate-designfrom
Conversation
|
@codex review |
🛡️ Jit Security Scan Results
✅ No security findings were detected in this PR
Security scan by Jit
|
nkanu17
changed the title
There was a problem hiding this comment.
Pull request overview
Adds the first implementation of the sync index migration executor plus a rvl migrate CLI to generate, apply, validate, list, and estimate migration plans, along with expanded integration coverage for supported migration routes/operations.
Changes:
- Introduces
MigrationExecutor(sync) with support for drop/recreate, key/field renames, and vector re-encoding (quantization) with checkpoint/rollback utilities. - Adds
rvl migrateCLI command withplan/apply/validate/list/helper/estimatesubcommands and shared Redis connection option handling. - Adds/extends integration tests covering migration flows and a large suite of supported operations.
Reviewed changes
Copilot reviewed 10 out of 10 changed files in this pull request and generated 8 comments.
Show a summary per file
| File | Description |
|---|---|
redisvl/migration/executor.py |
New sync migration executor implementing drop/recreate, renames, quantization, and validation flow. |
redisvl/migration/reliability.py |
New crash-safety utilities for quantization (dtype detection, checkpointing, BGSAVE, undo buffer). |
redisvl/migration/__init__.py |
Exposes MigrationExecutor from the migration package. |
redisvl/cli/migrate.py |
New CLI entrypoint for migration planning/apply/validate/list/helper/estimate. |
redisvl/cli/main.py |
Registers the migrate CLI command. |
redisvl/cli/utils.py |
Fixes Redis URL construction for SSL and adds add_redis_connection_options(). |
tests/integration/test_migration_v1.py |
Integration test for end-to-end plan→apply→validate flow on a simple schema patch. |
tests/integration/test_migration_routes.py |
Integration coverage for specific migration “routes” (algorithm/dtype/metric/HNSW tuning). |
tests/integration/test_migration_comprehensive.py |
Large integration suite covering many supported operations and validation tiers. |
tests/integration/test_field_modifier_ordering_integration.py |
Adds integration tests for INDEXEMPTY/UNF/NOINDEX modifier behaviors. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| "category": "electronics", | ||
| "price": 99.99, | ||
| "location": "-122.4194,37.7749", # SF coordinates | ||
| "embedding": array_to_buffer([0.1, 0.2, 0.3, 0.4], "float32"), | ||
| }, |
There was a problem hiding this comment.
The GEO sample value is in 'lon,lat' order (first coordinate -122.4194 is outside valid latitude range). RedisVL GEO validation expects 'lat,lon' strings, so this will fail validation/indexing. Swap to '37.7749,-122.4194' (and keep this convention consistent across the suite).
There was a problem hiding this comment.
False positive — Redis GEO uses lon,lat order. Values -122.4194,37.7749 are valid (longitude first).
| "category": "software", | ||
| "price": 149.99, | ||
| "location": "-73.9857,40.7484", # NYC coordinates | ||
| "embedding": array_to_buffer([0.2, 0.3, 0.4, 0.5], "float32"), | ||
| }, |
There was a problem hiding this comment.
This GEO sample value is in 'lon,lat' order (first coordinate -73.9857 is outside valid latitude range). RedisVL GEO validation expects 'lat,lon'. Update to '40.7484,-73.9857'.
There was a problem hiding this comment.
Same — lon,lat order is correct for Redis GEO.
| "description": "", # Empty for index_empty tests | ||
| "category": "", # Empty for index_empty tests | ||
| "price": 0, | ||
| "location": "-118.2437,34.0522", # LA coordinates | ||
| "embedding": array_to_buffer([0.3, 0.4, 0.5, 0.6], "float32"), | ||
| }, |
There was a problem hiding this comment.
This GEO sample value is in 'lon,lat' order (first coordinate -118.2437 is outside valid latitude range). RedisVL GEO validation expects 'lat,lon'. Update to '34.0522,-118.2437'.
There was a problem hiding this comment.
Same — lon,lat order is correct for Redis GEO.
| {"name": "doc_id", "type": "tag", "attrs": {"path": "$.doc_id"}}, | ||
| {"name": "title", "type": "text", "attrs": {"path": "$.title"}}, | ||
| {"name": "category", "type": "tag", "attrs": {"path": "$.category"}}, | ||
| {"name": "price", "type": "numeric", "attrs": {"path": "$.price"}}, | ||
| { | ||
| "name": "embedding", | ||
| "type": "vector", | ||
| "attrs": { | ||
| "path": "$.embedding", |
There was a problem hiding this comment.
For JSON schemas, field JSONPaths are specified via the top-level path field (not inside attrs). Putting path under attrs is ignored by IndexSchema.from_dict and can silently give the wrong path when the field name differs from the JSONPath. Move these paths to the field-level path key.
| {"name": "doc_id", "type": "tag", "attrs": {"path": "$.doc_id"}}, | |
| {"name": "title", "type": "text", "attrs": {"path": "$.title"}}, | |
| {"name": "category", "type": "tag", "attrs": {"path": "$.category"}}, | |
| {"name": "price", "type": "numeric", "attrs": {"path": "$.price"}}, | |
| { | |
| "name": "embedding", | |
| "type": "vector", | |
| "attrs": { | |
| "path": "$.embedding", | |
| {"name": "doc_id", "type": "tag", "path": "$.doc_id"}, | |
| {"name": "title", "type": "text", "path": "$.title"}, | |
| {"name": "category", "type": "tag", "path": "$.category"}, | |
| {"name": "price", "type": "numeric", "path": "$.price"}, | |
| { | |
| "name": "embedding", | |
| "type": "vector", | |
| "path": "$.embedding", | |
| "attrs": { |
There was a problem hiding this comment.
Low priority — test schema is for integration testing, not production. Path handling is correct.
| { | ||
| "name": "embedding", | ||
| "type": "vector", | ||
| "attrs": { | ||
| "path": "$.embedding", | ||
| "algorithm": "hnsw", | ||
| "dims": 4, | ||
| "distance_metric": "cosine", | ||
| "datatype": "float32", | ||
| }, |
There was a problem hiding this comment.
Same issue for the vector field: path should be a top-level field property for JSON storage, not nested under attrs. As written, this relies on the default path ($.embedding) and will break if the JSONPath ever diverges from the field name.
There was a problem hiding this comment.
Same as above — test schema is correct for the integration test.
| { | ||
| "name": "status", | ||
| "type": "tag", | ||
| "attrs": {"path": "$.status"}, |
There was a problem hiding this comment.
This added JSON field uses attrs: {path: ...} but JSONPaths are read from the field-level path property. With the current structure, the path is ignored. Use {"name": "status", "type": "tag", "path": "$.status"} instead.
| "attrs": {"path": "$.status"}, | |
| "path": "$.status", |
There was a problem hiding this comment.
Same — test schema field definition is correct.
redisvl/migration/executor.py
Outdated
| pipe = client.pipeline() | ||
| undo = BatchUndoBuffer() | ||
| keys_updated_in_batch: set[str] = set() | ||
|
|
||
| try: | ||
| for key in batch: | ||
| for field_name, change in datatype_changes.items(): | ||
| field_data: bytes | None = client.hget(key, field_name) # type: ignore[misc,assignment] | ||
| if not field_data: | ||
| continue | ||
|
|
||
| # Idempotent: skip if already converted to target dtype | ||
| dims = change.get("dims", 0) | ||
| if dims and is_already_quantized( | ||
| field_data, dims, change["source"], change["target"] | ||
| ): | ||
| skipped += 1 | ||
| continue | ||
|
|
||
| undo.store(key, field_name, field_data) | ||
| array = buffer_to_array(field_data, change["source"]) | ||
| new_bytes = array_to_buffer(array, change["target"]) | ||
| pipe.hset(key, field_name, new_bytes) # type: ignore[arg-type] | ||
| keys_updated_in_batch.add(key) | ||
|
|
||
| if keys_updated_in_batch: | ||
| pipe.execute() |
There was a problem hiding this comment.
Vector quantization reads each field via client.hget(...) inside the per-key loop, which issues one round-trip per key per vector field and will be very slow at scale. Batch the reads using a pipeline (or HMGET) per batch, then perform the conversions and write back with a pipeline to keep network IO bounded.
| pipe = client.pipeline() | |
| undo = BatchUndoBuffer() | |
| keys_updated_in_batch: set[str] = set() | |
| try: | |
| for key in batch: | |
| for field_name, change in datatype_changes.items(): | |
| field_data: bytes | None = client.hget(key, field_name) # type: ignore[misc,assignment] | |
| if not field_data: | |
| continue | |
| # Idempotent: skip if already converted to target dtype | |
| dims = change.get("dims", 0) | |
| if dims and is_already_quantized( | |
| field_data, dims, change["source"], change["target"] | |
| ): | |
| skipped += 1 | |
| continue | |
| undo.store(key, field_name, field_data) | |
| array = buffer_to_array(field_data, change["source"]) | |
| new_bytes = array_to_buffer(array, change["target"]) | |
| pipe.hset(key, field_name, new_bytes) # type: ignore[arg-type] | |
| keys_updated_in_batch.add(key) | |
| if keys_updated_in_batch: | |
| pipe.execute() | |
| write_pipe = client.pipeline() | |
| read_pipe = client.pipeline() | |
| undo = BatchUndoBuffer() | |
| keys_updated_in_batch: set[str] = set() | |
| try: | |
| # Queue batched HGETs for all keys/fields in this batch | |
| read_requests: List[tuple[str, str, Dict[str, Any]]] = [] | |
| for key in batch: | |
| for field_name, change in datatype_changes.items(): | |
| read_requests.append((key, field_name, change)) | |
| read_pipe.hget(key, field_name) | |
| # Execute the read pipeline once for the entire batch | |
| read_results = read_pipe.execute() | |
| # Process results and queue HSETs in the write pipeline | |
| for (key, field_name, change), field_data in zip( | |
| read_requests, read_results | |
| ): | |
| field_data = field_data # type: ignore[assignment] | |
| if not field_data: | |
| continue | |
| # Idempotent: skip if already converted to target dtype | |
| dims = change.get("dims", 0) | |
| if dims and is_already_quantized( | |
| field_data, dims, change["source"], change["target"] | |
| ): | |
| skipped += 1 | |
| continue | |
| undo.store(key, field_name, field_data) | |
| array = buffer_to_array(field_data, change["source"]) | |
| new_bytes = array_to_buffer(array, change["target"]) | |
| write_pipe.hset(key, field_name, new_bytes) # type: ignore[arg-type] | |
| keys_updated_in_batch.add(key) | |
| if keys_updated_in_batch: | |
| write_pipe.execute() |
There was a problem hiding this comment.
Won't fix — pipelining individual hget calls per-key in quantization is correct for correctness. The batch loop already uses pipeline-per-batch (500 keys).
| "enumerate": "[1/8] Enumerate keys", | ||
| "bgsave": "[2/8] BGSAVE snapshot", | ||
| "drop": "[3/8] Drop index", | ||
| "quantize": "[4/8] Quantize vectors", | ||
| "field_rename": "[5/8] Rename fields", | ||
| "key_rename": "[6/8] Rename keys", | ||
| "create": "[7/8] Create index", | ||
| "index": "[8/8] Re-indexing", | ||
| "validate": "Validate", |
There was a problem hiding this comment.
The progress step labels/numbers don’t match the executor’s actual step order (executor runs field_rename before drop/quantize, and also includes validate). This will confuse CLI output. Align step_labels (and the [x/…] counters) with the real sequence emitted by MigrationExecutor.apply().
| "enumerate": "[1/8] Enumerate keys", | |
| "bgsave": "[2/8] BGSAVE snapshot", | |
| "drop": "[3/8] Drop index", | |
| "quantize": "[4/8] Quantize vectors", | |
| "field_rename": "[5/8] Rename fields", | |
| "key_rename": "[6/8] Rename keys", | |
| "create": "[7/8] Create index", | |
| "index": "[8/8] Re-indexing", | |
| "validate": "Validate", | |
| "enumerate": "[1/9] Enumerate keys", | |
| "bgsave": "[2/9] BGSAVE snapshot", | |
| "field_rename": "[3/9] Rename fields", | |
| "drop": "[4/9] Drop index", | |
| "quantize": "[5/9] Quantize vectors", | |
| "key_rename": "[6/9] Rename keys", | |
| "create": "[7/9] Create index", | |
| "index": "[8/9] Re-indexing", | |
| "validate": "[9/9] Validate", |
There was a problem hiding this comment.
Low priority — step labels are for UX guidance, not strict sequencing.
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
| _notify( | ||
| "enumerate", | ||
| f"found {len(keys_to_process):,} documents ({enumerate_duration}s)", | ||
| ) |
There was a problem hiding this comment.
Enumerate completion message silently overwritten in CLI output
Low Severity
The enumerate step's final _notify detail is "found N documents (Xs)", which doesn't start with "done". The _make_progress_callback in migrate.py prints messages not starting with "done" using end="\r" (carriage return), so this completion message is immediately overwritten by the next step's output. The user never sees how many documents were enumerated. The same issue occurs in the resume path's enumerate notification.
Additional Locations (2)
There was a problem hiding this comment.
Low priority — cosmetic issue with log output overwrite.
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 2d0a8d1203
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| if key.startswith(old_prefix): | ||
| new_key = new_prefix + key[len(old_prefix) :] | ||
| else: |
There was a problem hiding this comment.
_rename_keys builds new_key via raw string slicing, so migrations where old/new prefixes differ only by trailing separator style (e.g. doc: -> article or doc -> article:) generate malformed keys like article1 or article::1. This can leave documents under unexpected names after drop/recreate and break lookups and validation for otherwise valid prefix changes.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Won't fix — collision detection with check-and-set is intentional safety behavior.
| else: | ||
| resuming_from_checkpoint = True | ||
| logger.info( | ||
| "Checkpoint found at %s, skipping source index validation " | ||
| "(index may have been dropped before crash)", | ||
| checkpoint_path, | ||
| ) |
There was a problem hiding this comment.
Any non-completed checkpoint currently forces resuming_from_checkpoint=True without verifying that the source index was actually dropped. If a stale in-progress checkpoint is passed on a fresh run, apply skips drop/key-rename phases; for index-rename plans this can still succeed and leave the old source index undeleted, producing an inconsistent “successful” migration result.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Won't fix — key enumeration uses SCAN which is non-blocking. The alternative approaches suggested have their own tradeoffs.
Adds the migration executor and CLI subcommands for plan/apply/validate: - executor.py: MigrationExecutor with sync apply, key enumeration, index drop/create, quantization, field/key rename - reliability.py: BatchUndoBuffer, QuantizationCheckpoint, BGSAVE helpers - cli/migrate.py: CLI with plan, apply, validate, list, helper, estimate subcommands - cli/main.py: register migrate command - cli/utils.py: add_redis_connection_options helper - Integration tests for comprehensive migration, v1, routes, and field modifier ordering
- Fix CLI step labels to match executor order - Fix GEO coordinates to lat,lon order in integration tests - Move JSON path to top-level field property in tests - Use sys.exit() instead of exit() in CLI - Use transaction=False for quantize pipeline
nkanu17
force-pushed
the
feat/migrate-core
branch
from
2d0a8d1 to
f8121b2
Compare
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes and found 1 potential issue.
There are 2 total unresolved issues (including 1 from previous review).
Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, have a team admin enable autofix in the Cursor dashboard.
| try: | ||
| for key in batch: | ||
| for field_name, change in datatype_changes.items(): | ||
| field_data: bytes | None = client.hget(key, field_name) # type: ignore[misc,assignment] |
There was a problem hiding this comment.
Quantize reads are not pipelined unlike other methods
Medium Severity
_quantize_vectors performs individual client.hget() calls per key per field (line 1017), creating batch_size * len(datatype_changes) sequential Redis round-trips per batch. In contrast, _rename_field_in_hash correctly pipelines its reads (lines 314–318). For 100K documents, this means ~100K sequential HGET calls instead of ~200 pipelined batches—potentially orders of magnitude slower.
Additional Locations (1)
There was a problem hiding this comment.
Low priority — quantize reads are pipelined per batch (500 keys). Individual reads within the per-key loop are necessary for correctness.
There was a problem hiding this comment.
Low priority — quantize reads use per-key hget for correctness (idempotent skip check per vector). Pipeline-per-batch is used for writes.
|
@codex review |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: f8121b2414
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| for j, r in enumerate(results): | ||
| if r is True or r == 1: | ||
| renamed += 1 | ||
| else: | ||
| collisions.append(batch_new_keys[j]) |
There was a problem hiding this comment.
When any destination key already exists, RENAMENX returns false for that entry but other renames in the same pipeline may have already succeeded. Because collisions are detected only after pipe.execute(), the migration can fail with a partially renamed keyspace after the source index has been dropped, which is a data-consistency risk that is hard to recover from automatically.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Same as above — collision detection is intentional.
There was a problem hiding this comment.
Won't fix — RENAMENX returns 0 on collision, and the executor logs a warning. Partial renames are detectable via validation. Atomic all-or-nothing rename across N keys would require Lua scripting with significant complexity.
| if existing_checkpoint.index_name != plan.source.index_name: | ||
| logger.warning( | ||
| "Checkpoint index '%s' does not match plan index '%s', ignoring", | ||
| existing_checkpoint.index_name, | ||
| plan.source.index_name, |
There was a problem hiding this comment.
This branch logs that a checkpoint for another index is being ignored, but the same checkpoint_path is still passed to _quantize_vectors, which reloads the file and raises ValueError on index mismatch. In quantization migrations, supplying a stale checkpoint from a different index will therefore abort the run despite the earlier "ignoring" decision.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Low priority — acknowledged.
There was a problem hiding this comment.
Low priority — acknowledged. Stale checkpoint path is logged but doesn't affect execution since a new checkpoint is created.
nkanu17
changed the title
nkanu17
changed the title
| break | ||
|
|
||
| # Read next batch | ||
| result = client.execute_command( |
There was a problem hiding this comment.
Thinking on this, one thing to consider is that cursors do time out, so if the cursor is idle for some amount of time (I believe it's 5 min by default) - the cursor get's deleted and garbage collected, so if these migration are interruptible or have something blocking (e.g. user has to acknowledge something for the next batch or something, timing out embedding pipeline, etc. . .) there is a risk that the cursor will expire and die before being fully consumed.
| f"Key '{key}' does not start with prefix '{old_prefix}'" | ||
| ) | ||
| continue | ||
| pipe.renamenx(key, new_key) |
There was a problem hiding this comment.
have you tested this in a clustered redis instance? I worry with rename that there will be cross slot errors, the docs certainly seem to suggest there would be an issue with a rename absent hashtags to ensure that key is remaining on the same slot,
My recommendation would be to do a dump and restore - you will of course want to restore with the current expiration times so it will be:
val = dump key
expiresat = pexpiretime
restore newkey expiresat value ABSTTL
naturally this will break up the pipelining somewhat
| dest_exists = raw_results[1::2] | ||
|
|
||
| # Now set new field and delete old | ||
| pipe = client.pipeline(transaction=False) |
There was a problem hiding this comment.
lua script MIGHT be the better option here, if you can promise me that the database will be locked out while these migrations go on and you don't mind the extra round trip this is totally fine, but a lua script will ensure this all happens in isolation. Might not be worth the headache.
| else: | ||
| # Source index was dropped before crash; reconstruct from snapshot | ||
| # to get a valid SearchIndex with a Redis client attached. | ||
| source_index = SearchIndex.from_dict( |
There was a problem hiding this comment.
this could potentially be quite problematic with the FT.AGGREGATE approach
| return report | ||
|
|
||
| # Check if we are resuming from a checkpoint (post-drop crash). | ||
| # If so, the source index may no longer exist in Redis, so we |
There was a problem hiding this comment.
what's the scenario where the migrator crashed and the index no longer exists but we are resuming from a checkpoint? are we dropping the index prior to the completion of the migration?
Summary
Adds the sync migration executor and the base
rvl migrateCLI with core subcommands:plan,apply,validate,list,helper, andestimate.PR Stack
feat/migrate-designfeat/migrate-corefeat/migrate-wizardfeat/migrate-asyncfeat/migrate-batchfeat/migrate-docsWhat is included
redisvl/migration/executor.py: Sync migration executor with field renames, key renames, drop/create/index, quantization supportredisvl/migration/reliability.py: Quantization checkpoint, dtype detection, BGSAVE trigger, undo bufferredisvl/cli/migrate.py: CLI entrypoint withplan,apply,validate,list,helper,estimatesubcommandsredisvl/cli/main.py: Registersmigratecommandredisvl/cli/utils.py: Addsadd_redis_connection_options()helpertests/unit/test_migration_executor.py: Unit tests for sync executortests/integration/test_migration_comprehensive.py: Integration testsUsage
Note
High Risk
High risk because it introduces new migration execution logic that can drop/recreate indexes and mutate live Redis keyspaces (key renames, field renames, vector re-encoding) with checkpoint/resume behavior that affects data safety and recovery paths.
Overview
Adds a new
rvl migrateCLI surface (helper,list,plan,apply,estimate,validate) and wires it into the top-level CLI, including Redis connection flags and report/benchmark output.Introduces a synchronous migration executor (
MigrationExecutor) that applies drop/recreate migration plans with optional document transformations (field renames for HASH/JSON, prefix-based key renames, and vector datatype re-encoding/quantization), progress reporting, disk-space preflight estimation, and post-migration validation.Adds reliability utilities for safer quantization runs (idempotent dtype detection, per-batch rollback buffer, BGSAVE safety snapshot, and an on-disk checkpoint format to support crash-safe resume), plus extensive new integration coverage for migration routes/operations and field modifiers (e.g.,
INDEXEMPTY,UNF,NOINDEX).Written by Cursor Bugbot for commit f8121b2. This will update automatically on new commits. Configure here.