From 6863ced51e3f9b843b448394b3c16b5c91ccdd2f Mon Sep 17 00:00:00 2001
From: Dmitry Teryaev <doudmitry@gmail.com>
Date: Mon, 6 Jul 2026 09:19:38 +0300
Subject: [PATCH] perf(vectors): single-component bulk declare + MPS default
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

cocoindex flushes target writes once per processing component; coco.mount_each
made one component per file → one Lance merge_insert (+ fsync) per file → 1168
fragments / 1170 manifest commits, ~91s of kernel I/O on a 1167-file repo (84%
of init time). Iterate walk_dir().items() in ONE component instead → all
declare_row calls flush as one merge_insert per table.

process_*_file stay @coco.fn(memo=True) and _RowHandler.reconcile skips rows
whose fingerprint is unchanged, so increment stays correct and proportional
(no-change: 2.4s/0 new fragments; 1-file change: 10.2s/+1 fragment). MPS
default for the embedder on Apple Silicon (free in the child process; operator
overrides with SBERT_DEVICE=cpu).

Measured on Shopizer (1167 files / ~3500 chunks):
  init end-to-end ~143s cold → 44s; vectors 119.7s → 40.4s;
  sys 91.2s → 4.2s; fragments 1168 → 1.

The single-component loop is the supported workaround for cocoindex#2219
(native cross-component sink batching, open upstream).

Refs: cocoindex#2219. Follow-ups: #379, #380.

Co-Authored-By: Claude <noreply@anthropic.com>
---
 java_index_flow_lancedb.py | 53 ++++++++++++++++++++++++--------------
 1 file changed, 34 insertions(+), 19 deletions(-)

diff --git a/java_index_flow_lancedb.py b/java_index_flow_lancedb.py
index ea2ab2f4..52120b56 100644
--- a/java_index_flow_lancedb.py
+++ b/java_index_flow_lancedb.py
@@ -291,9 +291,23 @@ async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]
         root = Path(".").resolve()
     builder.provide(PROJECT_ROOT, root)
 
+    # Default to Apple Metal (MPS) when available: ~1.7x faster encode on
+    # all-MiniLM-L6-v2 (measured), and the win grows with repo size since
+    # embedding dominates on large trees. torch is already on the import path
+    # here (sentence-transformers pulls it), so the availability check is free
+    # in this child process — and it keeps the CLI parent (config.py) from ever
+    # paying a torch import. Operators force CPU with SBERT_DEVICE=cpu.
+    device = os.environ.get("SBERT_DEVICE") or None
+    if device is None:
+        try:
+            import torch  # noqa: WPS433 (local import: avoid parent-import cost)
+            if torch.backends.mps.is_available():
+                device = "mps"
+        except Exception:
+            pass
     embedder = SentenceTransformerEmbedder(
         resolved_sbert_model_for_process_env(SBERT_MODEL),
-        device=os.environ.get("SBERT_DEVICE") or None,
+        device=device,
         trust_remote_code=True,
     )
     builder.provide(EMBEDDER, embedder)
@@ -619,24 +633,25 @@ async def app_main() -> None:
         ),
     )
 
-    await coco.mount_each(
-        coco.component_subpath(coco.Symbol("java_files")),
-        process_java_file,
-        java_files.items(),
-        java_table,
-    )
-    await coco.mount_each(
-        coco.component_subpath(coco.Symbol("sql_files")),
-        process_sql_file,
-        sql_files.items(),
-        sql_table,
-    )
-    await coco.mount_each(
-        coco.component_subpath(coco.Symbol("yaml_files")),
-        process_yaml_file,
-        yaml_files.items(),
-        yaml_table,
-    )
+    # PERF: declare all rows in ONE component (app_main) instead of one
+    # component per file via coco.mount_each. cocoindex flushes target writes
+    # once per processing component, and all declare_row calls inside a
+    # component batch into a single Lance merge_insert (see _RowHandler.
+    # _apply_actions). mount_each created one component PER FILE → ~1167
+    # merge_insert transactions (one fragment + manifest commit each) → ~91s
+    # of kernel I/O on a 1167-file repo. The single-component loop collapses
+    # that to ONE merge_insert per table. cocoindex does not yet batch across
+    # mount_each components natively (open issue cocoindex#2219), so the loop
+    # is the supported workaround. process_*_file stay @coco.fn(memo=True), so
+    # unchanged files still skip re-embedding on incremental; _RowHandler.
+    # reconcile skips rows whose fingerprint is unchanged → increment carries
+    # only changed rows in its single merge_insert.
+    async for _key, _file in java_files.items():
+        await process_java_file(_file, java_table)
+    async for _key, _file in sql_files.items():
+        await process_sql_file(_file, sql_table)
+    async for _key, _file in yaml_files.items():
+        await process_yaml_file(_file, yaml_table)
 
 
 app = coco.App(