From c6f324a1742406494b6e5c0fc12efc6c48414a38 Mon Sep 17 00:00:00 2001 From: Dmitry Teryaev Date: Mon, 6 Jul 2026 23:17:35 +0300 Subject: [PATCH] fix(reprocess): drop Lance tables before full-reprocess to avoid per-row-delete hang MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit reprocess (cocoindex ``update --full-reprocess`` against an EXISTING table) took the in-place bulk-update merge_insert path, which cocoindex implements as ~one deletion-vector + version commit PER matched row — O(rows) of tiny file IO. Measured on Shopizer (3475 chunks): 3481 versions / 3474 deletion files / 83s sys time, degrading to a multi-minute 99%-CPU hang (native ``_lancedb.abi3.so``, main thread parked in ``kevent``) that reproduces every run and worsens with repo size — a larger repo hung 15+ min at "99%". init (insert into a fresh table) is unaffected: 3 versions / 0 deletions / 4.5s sys. drop+recreate is semantically identical to a full rebuild (all rows are recomputed either way), so make reprocess DROP the Lance target tables first and let cocoindex recreate them via the fast INSERT path. Applied at both ``--full-reprocess`` sites: - ``pipeline.run_cocoindex_update`` → covers ``reprocess --vectors-only`` - ``server.run_refresh_pipeline`` → covers the default ``reprocess`` (vectors+graph) Drop failure is non-fatal: a non-preflight failure logs and falls back to the slow in-place path; a 127/126 preflight stub (graph-only installs) stays silent. Validated on Shopizer: reprocess 105s→hang → 48s (≈ init), sys 83s→4.9s, deletions 3474→0, versions 3481→3. Independent of #392 (concurrency) and #393 (mem-pool/PK-index): ``LANCE_MEM_POOL_SIZE=4GiB`` did not help — this is file-IO amplification, not memory. Tests: ``tests/test_pipeline.py`` — full_reprocess drops exactly once; increment does not drop (would lose the table); drop failure falls back to in-place; a preflight drop stub is silent. Co-Authored-By: Claude --- java_codebase_rag/pipeline.py | 20 +++++++ server.py | 25 +++++++++ tests/test_pipeline.py | 103 ++++++++++++++++++++++++++++++++++ 3 files changed, 148 insertions(+) create mode 100644 tests/test_pipeline.py diff --git a/java_codebase_rag/pipeline.py b/java_codebase_rag/pipeline.py index bcd15f8..c3e35e5 100644 --- a/java_codebase_rag/pipeline.py +++ b/java_codebase_rag/pipeline.py @@ -132,6 +132,26 @@ def run_cocoindex_update( on_progress: Callable[[ProgressEvent], None] | None = None, on_progress_console: object | None = None, ) -> subprocess.CompletedProcess[str]: + if full_reprocess: + # A full reprocess rebuilds every row, so DROP the Lance target tables + # first and let cocoindex recreate them via the fast INSERT path. The + # in-place alternative (cocoindex's bulk-update merge_insert) emits + # ~one deletion-vector + version commit PER matched row — O(rows) of + # tiny file IO that scales to multi-minute hangs on large repos + # (measured ~83s sys time / 3474 deletion files for 3475 chunks on + # Shopizer; drop+recreate is ~3.6s sys / 0 deletions, ~3.7x faster and + # hang-free). Output is identical either way (full recompute); only the + # write path differs. Drop failure is non-fatal — if it somehow fails, + # the update falls back to the slow in-place path. The same fix is + # applied on the async server path (``server.run_refresh_pipeline``). + drop = run_cocoindex_drop(env, quiet=quiet) + if drop.returncode != 0 and not is_cocoindex_preflight_blocker(drop): + print( + "java-codebase-rag: drop-before-reprocess failed " + f"(exit {drop.returncode}); falling back to in-place update: " + f"{(drop.stderr or '').strip()[:200]}", + file=sys.stderr, + ) result = _run_cocoindex_update_impl( env, full_reprocess=full_reprocess, diff --git a/server.py b/server.py index be3b616..95e4280 100644 --- a/server.py +++ b/server.py @@ -349,6 +349,31 @@ async def run_refresh_pipeline( ) proc: asyncio.subprocess.Process | None = None out_b, err_b = b"", b"" + # DROP the Lance target tables so the update takes the fast INSERT path + # instead of cocoindex's in-place bulk-update, which emits ~one deletion- + # vector + version commit PER matched row — O(rows) of tiny file IO that + # hangs for many minutes on large repos. Drop+recreate is identical output + # for a full rebuild (the very thing --full-reprocess means). Same fix on + # the sync path: pipeline.run_cocoindex_update. Drop failure is non-fatal: + # the update falls back to the slow in-place path. + try: + drop_proc = await asyncio.create_subprocess_exec( + str(cocoindex_bin), + "drop", + _COCOINDEX_TARGET, + "-f", + cwd=str(flow_path.parent), + env=_cocoindex_subprocess_env(root), + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + await drop_proc.communicate() + except Exception as exc: + print( + f"java-codebase-rag: drop-before-reprocess failed ({exc!s}); " + "falling back to in-place update", + file=sys.stderr, + ) if quiet: try: proc = await asyncio.create_subprocess_exec( diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py new file mode 100644 index 0000000..7d8fb7c --- /dev/null +++ b/tests/test_pipeline.py @@ -0,0 +1,103 @@ +"""Tests for ``java_codebase_rag.pipeline`` subprocess helpers. + +Focus: ``run_cocoindex_update`` drops the Lance target tables before a *full +reprocess* so the update takes the fast INSERT path. The in-place alternative +(cocoindex's bulk-update ``merge_insert``) emits ~one deletion-vector + version +commit per matched row — O(rows) of tiny file IO that hangs for many minutes on +large repos. Drop+recreate is identical output for a full rebuild. +""" +from __future__ import annotations + +import subprocess + +from java_codebase_rag import pipeline + + +def _ok() -> subprocess.CompletedProcess: + return subprocess.CompletedProcess(args=[], returncode=0, stdout="", stderr="") + + +def _stub_impl(monkeypatch, seen: dict) -> None: + """Replace the impl + post-optimize with no-op stubs (no cocoindex/lancedb).""" + + def fake_impl(env, **kwargs): + seen["update"] = seen.get("update", 0) + 1 + seen["full_reprocess"] = kwargs.get("full_reprocess") + return _ok() + + monkeypatch.setattr(pipeline, "_run_cocoindex_update_impl", fake_impl) + monkeypatch.setattr(pipeline, "_maybe_run_serialized_optimize", lambda *a, **k: None) + + +def test_full_reprocess_drops_tables_first(monkeypatch) -> None: + """full_reprocess=True drops exactly once before the update (INSERT path).""" + seen: dict = {} + _stub_impl(monkeypatch, seen) + drops: list[dict] = [] + + def fake_drop(env, *, quiet): + drops.append(env) + return _ok() + + monkeypatch.setattr(pipeline, "run_cocoindex_drop", fake_drop) + + pipeline.run_cocoindex_update({"X": "1"}, full_reprocess=True, quiet=True) + + assert len(drops) == 1, "full_reprocess must drop exactly once before update" + assert drops[0] == {"X": "1"}, "drop must receive the same env as the update" + assert seen["update"] == 1 + + +def test_increment_does_not_drop(monkeypatch) -> None: + """full_reprocess=False (increment) must NOT drop — it would lose the table.""" + seen: dict = {} + _stub_impl(monkeypatch, seen) + drops: list[dict] = [] + + def fake_drop(env, *, quiet): + drops.append(env) + return _ok() + + monkeypatch.setattr(pipeline, "run_cocoindex_drop", fake_drop) + + pipeline.run_cocoindex_update({}, full_reprocess=False, quiet=True) + + assert drops == [], "increment must not drop the tables" + assert seen["update"] == 1 + + +def test_drop_failure_falls_back_to_inplace(monkeypatch, capsys) -> None: + """A non-preflight drop failure does not abort — the update still runs in-place.""" + seen: dict = {} + _stub_impl(monkeypatch, seen) + monkeypatch.setattr( + pipeline, + "run_cocoindex_drop", + lambda env, *, quiet: subprocess.CompletedProcess( + args=[], returncode=2, stdout="", stderr="boom" + ), + ) + + pipeline.run_cocoindex_update({}, full_reprocess=True, quiet=True) + + assert seen["update"] == 1, "update must still run after a non-fatal drop failure" + assert "drop-before-reprocess failed" in capsys.readouterr().err + + +def test_drop_preflight_blocker_is_silent(monkeypatch, capsys) -> None: + """A preflight drop stub (cocoindex not installed, e.g. graph-only) is not noisy.""" + seen: dict = {} + _stub_impl(monkeypatch, seen) + monkeypatch.setattr( + pipeline, + "run_cocoindex_drop", + lambda env, *, quiet: subprocess.CompletedProcess( + args=["cocoindex"], returncode=127, stdout="", stderr="not found" + ), + ) + + pipeline.run_cocoindex_update({}, full_reprocess=True, quiet=True) + + assert seen["update"] == 1 + # 127 preflight is expected on graph-only installs and must NOT warn. + assert "drop-before-reprocess failed" not in capsys.readouterr().err