Bug fixes, test

Author: nkanu17

redisvl/migration/async_executor.py

@@ -78,14 +78,20 @@ async def _enumerate_indexed_keys(
         """Async version: Enumerate document keys using FT.AGGREGATE with SCAN fallback.
 
         Uses FT.AGGREGATE WITHCURSOR for efficient enumeration when the index
-        has no indexing failures. Falls back to SCAN if:
+        is fully built and has no indexing failures. Falls back to SCAN if:
         - Index has hash_indexing_failures > 0 (would miss failed docs)
+        - Index has percent_indexed < 1.0 (background HNSW build still in
+          progress; FT.AGGREGATE returns only fully-indexed docs and would
+          silently drop the pending tail)
         - FT.AGGREGATE command fails for any reason
         """
-        # Check for indexing failures - if any, fall back to SCAN
+        # Check for indexing failures or in-progress indexing — either
+        # condition means FT.AGGREGATE would miss documents, so fall
+        # back to SCAN for complete enumeration.
         try:
             info = await client.ft(index_name).info()
             failures = int(info.get("hash_indexing_failures", 0) or 0)
+            percent_indexed = float(info.get("percent_indexed", 1.0) or 1.0)
             if failures > 0:
                 logger.warning(
                     f"Index '{index_name}' has {failures} indexing failures. "
@@ -96,6 +102,17 @@ async def _enumerate_indexed_keys(
                 ):
                     yield key
                 return
+            if percent_indexed < 1.0:
+                logger.warning(
+                    f"Index '{index_name}' is still building "
+                    f"(percent_indexed={percent_indexed:.4f}). "
+                    "Using SCAN for complete enumeration."
+                )
+                async for key in self._enumerate_with_scan(
+                    client, index_name, batch_size, key_separator
+                ):
+                    yield key
+                return
         except Exception as e:
             logger.warning(f"Failed to check index info: {e}. Using SCAN fallback.")
             async for key in self._enumerate_with_scan(

redisvl/migration/executor.py

@@ -56,8 +56,11 @@ def _enumerate_indexed_keys(
         """Enumerate document keys using FT.AGGREGATE with SCAN fallback.
 
         Uses FT.AGGREGATE WITHCURSOR for efficient enumeration when the index
-        has no indexing failures. Falls back to SCAN if:
+        is fully built and has no indexing failures. Falls back to SCAN if:
         - Index has hash_indexing_failures > 0 (would miss failed docs)
+        - Index has percent_indexed < 1.0 (background HNSW build still in
+          progress; FT.AGGREGATE returns only fully-indexed docs and would
+          silently drop the pending tail)
         - FT.AGGREGATE command fails for any reason
 
         Args:
@@ -69,10 +72,13 @@ def _enumerate_indexed_keys(
         Yields:
             Document keys as strings
         """
-        # Check for indexing failures - if any, fall back to SCAN
+        # Check for indexing failures or in-progress indexing — either
+        # condition means FT.AGGREGATE would miss documents, so fall
+        # back to SCAN for complete enumeration.
         try:
             info = client.ft(index_name).info()
             failures = int(info.get("hash_indexing_failures", 0) or 0)
+            percent_indexed = float(info.get("percent_indexed", 1.0) or 1.0)
             if failures > 0:
                 logger.warning(
                     f"Index '{index_name}' has {failures} indexing failures. "
@@ -82,6 +88,16 @@ def _enumerate_indexed_keys(
                     client, index_name, batch_size, key_separator
                 )
                 return
+            if percent_indexed < 1.0:
+                logger.warning(
+                    f"Index '{index_name}' is still building "
+                    f"(percent_indexed={percent_indexed:.4f}). "
+                    "Using SCAN for complete enumeration."
+                )
+                yield from self._enumerate_with_scan(
+                    client, index_name, batch_size, key_separator
+                )
+                return
         except Exception as e:
             logger.warning(f"Failed to check index info: {e}. Using SCAN fallback.")
             yield from self._enumerate_with_scan(

redisvl/migration/planner.py

@@ -164,6 +164,25 @@ def create_plan_from_patch(
                 "the pre-migration count. This is expected and validation accounts for it."
             )
 
+        # Warn if source index is still building.  FT.AGGREGATE returns only
+        # fully-indexed docs, so applying a migration before the background
+        # build settles would silently drop the pending tail.  The executor
+        # falls back to SCAN automatically, but surface the condition here
+        # so users running `rvl migrate plan` can wait for indexing to
+        # complete before applying.
+        source_percent_indexed = float(
+            snapshot.stats_snapshot.get("percent_indexed", 1.0) or 1.0
+        )
+        if source_percent_indexed < 1.0:
+            warnings.append(
+                f"Source index is still building "
+                f"(percent_indexed={source_percent_indexed:.4f}). "
+                "Apply will fall back to SCAN enumeration to avoid missing "
+                "documents whose background HNSW indexing has not completed. "
+                "Wait for percent_indexed to reach 1.0 before applying for "
+                "the fastest migration path."
+            )
+
         # Check for SVS-VAMANA in target schema and add appropriate warnings
         svs_warnings = self._check_svs_vamana_requirements(
             merged_target_schema,

tests/unit/test_executor_backup_quantize.py

@@ -273,3 +273,106 @@ def test_default_backup_dir_exported_from_package(self):
 
         assert DEFAULT_BACKUP_DIR
         assert isinstance(DEFAULT_BACKUP_DIR, str)
+
+
+class TestEnumerateScanFallback:
+    """SCAN-fallback conditions in MigrationExecutor._enumerate_indexed_keys."""
+
+    def _build_executor_with_info(self, info_dict):
+        """Construct an executor and a mock client whose ft().info() returns info_dict."""
+        from redisvl.migration.executor import MigrationExecutor
+
+        executor = MigrationExecutor()
+        mock_client = MagicMock()
+        mock_ft = MagicMock()
+        mock_ft.info.return_value = info_dict
+        mock_client.ft.return_value = mock_ft
+        return executor, mock_client
+
+    def test_falls_back_to_scan_when_percent_indexed_below_one(self):
+        """percent_indexed < 1.0 must trigger SCAN fallback to avoid silent loss."""
+        executor, mock_client = self._build_executor_with_info(
+            {"hash_indexing_failures": 0, "percent_indexed": "0.5"}
+        )
+
+        with (
+            patch.object(
+                executor,
+                "_enumerate_with_scan",
+                return_value=iter(["doc:1", "doc:2"]),
+            ) as scan_mock,
+            patch.object(
+                executor,
+                "_enumerate_with_aggregate",
+                return_value=iter(["should-not-be-used"]),
+            ) as aggregate_mock,
+        ):
+            keys = list(executor._enumerate_indexed_keys(mock_client, "idx"))
+
+        scan_mock.assert_called_once()
+        aggregate_mock.assert_not_called()
+        assert keys == ["doc:1", "doc:2"]
+
+    def test_uses_aggregate_when_fully_indexed(self):
+        """percent_indexed == 1.0 with no failures should use FT.AGGREGATE."""
+        executor, mock_client = self._build_executor_with_info(
+            {"hash_indexing_failures": 0, "percent_indexed": "1"}
+        )
+
+        with (
+            patch.object(
+                executor,
+                "_enumerate_with_scan",
+                return_value=iter(["should-not-be-used"]),
+            ) as scan_mock,
+            patch.object(
+                executor,
+                "_enumerate_with_aggregate",
+                return_value=iter(["doc:1", "doc:2"]),
+            ) as aggregate_mock,
+        ):
+            keys = list(executor._enumerate_indexed_keys(mock_client, "idx"))
+
+        scan_mock.assert_not_called()
+        aggregate_mock.assert_called_once()
+        assert keys == ["doc:1", "doc:2"]
+
+    def test_failures_take_precedence_over_percent_indexed(self):
+        """hash_indexing_failures > 0 always triggers SCAN, regardless of percent_indexed."""
+        executor, mock_client = self._build_executor_with_info(
+            {"hash_indexing_failures": 7, "percent_indexed": "1"}
+        )
+
+        with patch.object(
+            executor,
+            "_enumerate_with_scan",
+            return_value=iter(["doc:1"]),
+        ) as scan_mock:
+            keys = list(executor._enumerate_indexed_keys(mock_client, "idx"))
+
+        scan_mock.assert_called_once()
+        assert keys == ["doc:1"]
+
+    def test_treats_missing_percent_indexed_as_complete(self):
+        """Missing percent_indexed key should default to 1.0 (use FT.AGGREGATE)."""
+        executor, mock_client = self._build_executor_with_info(
+            {"hash_indexing_failures": 0}
+        )
+
+        with (
+            patch.object(
+                executor,
+                "_enumerate_with_scan",
+                return_value=iter(["should-not-be-used"]),
+            ) as scan_mock,
+            patch.object(
+                executor,
+                "_enumerate_with_aggregate",
+                return_value=iter(["doc:1"]),
+            ) as aggregate_mock,
+        ):
+            keys = list(executor._enumerate_indexed_keys(mock_client, "idx"))
+
+        scan_mock.assert_not_called()
+        aggregate_mock.assert_called_once()
+        assert keys == ["doc:1"]

tests/unit/test_migration_planner.py

@@ -1008,6 +1008,124 @@ def test_plan_no_warning_when_stats_missing_failures_key(monkeypatch, tmp_path):
     assert len(failure_warnings) == 0
 
 
+def test_plan_warns_when_source_is_still_indexing(monkeypatch, tmp_path):
+    """Plan should warn when the source index has percent_indexed < 1.0."""
+    source_schema = _make_source_schema()
+    dummy_index = DummyIndex(
+        source_schema,
+        {"num_docs": 100, "hash_indexing_failures": 0, "percent_indexed": "0.42"},
+        [b"docs:1"],
+    )
+    monkeypatch.setattr(
+        "redisvl.migration.planner.SearchIndex.from_existing",
+        lambda *args, **kwargs: dummy_index,
+    )
+
+    patch_path = tmp_path / "schema_patch.yaml"
+    patch_path.write_text(
+        yaml.safe_dump(
+            {
+                "version": 1,
+                "changes": {
+                    "add_fields": [
+                        {"name": "status", "type": "tag", "path": "$.status"}
+                    ],
+                },
+            },
+            sort_keys=False,
+        )
+    )
+
+    planner = MigrationPlanner()
+    plan = planner.create_plan(
+        "docs",
+        redis_url="redis://localhost:6379",
+        schema_patch_path=str(patch_path),
+    )
+
+    indexing_warnings = [w for w in plan.warnings if "still building" in w]
+    assert len(indexing_warnings) == 1
+    assert "0.4200" in indexing_warnings[0]
+
+
+def test_plan_no_warning_when_source_fully_indexed(monkeypatch, tmp_path):
+    """Plan should NOT warn when percent_indexed == 1.0."""
+    source_schema = _make_source_schema()
+    dummy_index = DummyIndex(
+        source_schema,
+        {"num_docs": 100, "hash_indexing_failures": 0, "percent_indexed": "1"},
+        [b"docs:1"],
+    )
+    monkeypatch.setattr(
+        "redisvl.migration.planner.SearchIndex.from_existing",
+        lambda *args, **kwargs: dummy_index,
+    )
+
+    patch_path = tmp_path / "schema_patch.yaml"
+    patch_path.write_text(
+        yaml.safe_dump(
+            {
+                "version": 1,
+                "changes": {
+                    "add_fields": [
+                        {"name": "status", "type": "tag", "path": "$.status"}
+                    ],
+                },
+            },
+            sort_keys=False,
+        )
+    )
+
+    planner = MigrationPlanner()
+    plan = planner.create_plan(
+        "docs",
+        redis_url="redis://localhost:6379",
+        schema_patch_path=str(patch_path),
+    )
+
+    indexing_warnings = [w for w in plan.warnings if "still building" in w]
+    assert len(indexing_warnings) == 0
+
+
+def test_plan_no_warning_when_percent_indexed_missing(monkeypatch, tmp_path):
+    """Plan should treat missing percent_indexed as fully indexed (no warning)."""
+    source_schema = _make_source_schema()
+    dummy_index = DummyIndex(
+        source_schema,
+        {"num_docs": 100, "hash_indexing_failures": 0},
+        [b"docs:1"],
+    )
+    monkeypatch.setattr(
+        "redisvl.migration.planner.SearchIndex.from_existing",
+        lambda *args, **kwargs: dummy_index,
+    )
+
+    patch_path = tmp_path / "schema_patch.yaml"
+    patch_path.write_text(
+        yaml.safe_dump(
+            {
+                "version": 1,
+                "changes": {
+                    "add_fields": [
+                        {"name": "status", "type": "tag", "path": "$.status"}
+                    ],
+                },
+            },
+            sort_keys=False,
+        )
+    )
+
+    planner = MigrationPlanner()
+    plan = planner.create_plan(
+        "docs",
+        redis_url="redis://localhost:6379",
+        schema_patch_path=str(patch_path),
+    )
+
+    indexing_warnings = [w for w in plan.warnings if "still building" in w]
+    assert len(indexing_warnings) == 0
+
+
 # =============================================================================
 # TDD: Validation cluster-safe EXISTS + multi-prefix key translation
 # =============================================================================