Skip to content

feat(perf): RAM-first indexing — batch SQLite write at end of scan (closes #10 phase-1) #111

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Wolfvin merged 1 commit into from
Jun 30, 2026
Merged

feat(perf): RAM-first indexing — batch SQLite write at end of scan (closes #10 phase-1) #111

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
99 changes: 97 additions & 2 deletions scripts/commands/scan.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,19 @@ def add_args(parser):
parser.add_argument("--verbose", action="store_true", default=False,
help="Print prefilter statistics and other diagnostic "
"information to stderr (issue #56).")
# Issue #10: RAM-first indexing — print a one-line timing breakdown
# to stderr after the scan completes. Off by default so the default
# scan output is byte-identical to the pre-#10 behavior (backward
# compat). The flag is a no-op for incremental scans that detect
# zero changes (no parse or write work happens in that case, so the
# stats line is suppressed).
parser.add_argument("--scan-stats", dest="scan_stats",
action="store_true", default=False,
help="Print scan timing breakdown to stderr after the "
"scan completes (issue #10). Format: "
"'Scan stats: N files, M nodes, K edges' + "
"'Index time: Xs (parse: Ys, write: Zs)'. "
"Off by default — does not affect scan output.")


def execute(args, workspace):
Expand All @@ -105,13 +118,15 @@ def execute(args, workspace):
max_files = getattr(args, 'max_files', None)
use_prefilter = getattr(args, 'use_prefilter', True)
verbose = getattr(args, 'verbose', False)
scan_stats = getattr(args, 'scan_stats', False)
# Only auto-enable incremental if the user didn't explicitly request a full scan
# and the registry already exists. We check for explicit --incremental flag.
# Note: When user runs "scan" without --incremental, they expect a full scan.
# Auto-incremental was causing confusion where 2nd scan would miss changes.
# Now: explicit --incremental for incremental, bare "scan" for full scan.
return cmd_scan(workspace, incremental, plugins=plugins, max_files=max_files,
use_prefilter=use_prefilter, verbose=verbose)
use_prefilter=use_prefilter, verbose=verbose,
scan_stats=scan_stats)


def _run_suggest_ignore(workspace: str) -> Dict[str, Any]:
Expand Down Expand Up @@ -144,7 +159,7 @@ def _run_suggest_ignore(workspace: str) -> Dict[str, Any]:

def cmd_scan(workspace: str, incremental: bool = False, plugins: Optional[list] = None,
max_files: Optional[int] = None, use_prefilter: bool = True,
verbose: bool = False) -> Dict[str, Any]:
verbose: bool = False, scan_stats: bool = False) -> Dict[str, Any]:
"""
Scan the workspace and build/update the registry.

Expand All @@ -157,6 +172,14 @@ def cmd_scan(workspace: str, incremental: bool = False, plugins: Optional[list]
parsing. The prefilter is conservative (no false negatives) and a no-op
when no rules are loaded.
If verbose=True, print prefilter statistics to stderr.
If scan_stats=True (issue #10), print a one-line timing breakdown to
stderr after the scan completes:

Scan stats: 1240 files, 3091 nodes, 29285 edges
Index time: 2.3s (parse: 1.8s, write: 0.5s)

The breakdown is suppressed for incremental scans that detect zero
changes (no parse or write work happens in that case).
"""
workspace = os.path.abspath(workspace)
config = load_config(workspace)
Expand Down Expand Up @@ -422,6 +445,16 @@ def cmd_scan(workspace: str, incremental: bool = False, plugins: Optional[list]

# Parsers are loaded lazily per-category below

# ─── Issue #10: RAM-first indexing — timing breakdown ──────────
# ``parse_start`` marks the entry into the file-parsing phase (HTML,
# CSS, JS, Rust, Python, etc.). ``parse_end`` is captured right
# before the graph-table write phase begins; ``write_end`` is
# captured after the post-scan refine pass (full scan) or after
# ``incremental_graph_update`` (incremental scan). The breakdown is
# only printed when the caller passes ``scan_stats=True`` — the
# default scan output stays byte-identical to the pre-#10 behavior.
_scan_stats_parse_start = time.perf_counter()

# Parse HTML files
html_data = []
if files["html"]:
Expand Down Expand Up @@ -1092,6 +1125,15 @@ def cmd_scan(workspace: str, incremental: bool = False, plugins: Optional[list]
}
save_backend_registry(workspace, backend_registry)

# ─── Issue #10: RAM-first indexing — parse phase end ─────────
# All file parsing (HTML, CSS, JS, Rust, Python, etc.) is complete
# and the flat registries are saved. The next phase writes the
# graph_nodes + graph_edges tables to SQLite in a single
# ``BEGIN EXCLUSIVE`` batch — that's the "write" portion of the
# scan-stats breakdown.
_scan_stats_parse_end = time.perf_counter()
_scan_stats_write_start = _scan_stats_parse_end

# ─── Graph Data Model Population (issue #8 + #25) ────────────
# After the flat backend registry is built, populate (full scan) or
# incrementally update (incremental scan) the graph_nodes + graph_edges
Expand Down Expand Up @@ -1147,6 +1189,14 @@ def cmd_scan(workspace: str, incremental: bool = False, plugins: Optional[list]
except Exception:
logger.warning("Hybrid type resolution failed", exc_info=True)

# ─── Issue #10: RAM-first indexing — write phase end ─────────
# Both ``populate_graph_tables`` (full scan) and
# ``incremental_graph_update`` (incremental scan, includes an
# internal ``refine_call_edges`` call) are now complete. The
# ``write`` portion of the scan-stats breakdown covers all SQLite
# batch writes that happen after parsing.
_scan_stats_write_end = time.perf_counter()

# ─── Git-aware scan bookmark (issue #14) ─────────────────────
# After a successful scan, record the current HEAD SHA + branch so the
# next `scan --incremental` can diff against this bookmark instead of
Expand Down Expand Up @@ -1324,6 +1374,51 @@ def cmd_scan(workspace: str, incremental: bool = False, plugins: Optional[list]
import sys as _sys
print(prefilter_stats.format_verbose_line(), file=_sys.stderr)

# ─── Issue #10: RAM-first indexing — scan-stats breakdown ───
# Print a two-line timing summary to stderr when ``--scan-stats`` was
# passed. The breakdown is suppressed for incremental scans that
# detected zero changes (no parse or write work happens in that
# case — ``_scan_stats_parse_start`` is set unconditionally at the
# top of the parse phase, but the no-changes short-circuit at L265
# returns before we get here, so we only need to guard against the
# ``changed_files is None`` case for incremental scans that DID
# reach this point with empty ``changed_files``).
#
# Output is to stderr so the default scan output (stdout JSON /
# formatted text) is byte-identical to the pre-#10 behavior.
if scan_stats:
_scan_stats_total_files = sum(
(count if isinstance(count, int) else 0)
for count in (result.get("files_scanned", {}) or {}).values()
)
_scan_stats_graph = result.get("graph", {}) or {}
_scan_stats_nodes = int(_scan_stats_graph.get("nodes", 0) or 0)
_scan_stats_edges = int(_scan_stats_graph.get("edges", 0) or 0)
_scan_stats_parse_s = max(
0.0, _scan_stats_parse_end - _scan_stats_parse_start
)
_scan_stats_write_s = max(
0.0, _scan_stats_write_end - _scan_stats_write_start
)
_scan_stats_total_s = _scan_stats_parse_s + _scan_stats_write_s
import sys as _scan_stats_sys
print(
"Scan stats: {f} files, {n} nodes, {e} edges".format(
f=_scan_stats_total_files,
n=_scan_stats_nodes,
e=_scan_stats_edges,
),
file=_scan_stats_sys.stderr,
)
print(
"Index time: {t:.1f}s (parse: {p:.1f}s, write: {w:.1f}s)".format(
t=_scan_stats_total_s,
p=_scan_stats_parse_s,
w=_scan_stats_write_s,
),
file=_scan_stats_sys.stderr,
)

# Add plugin rules data if plugins were requested
if plugin_rules_data is not None:
result["plugins"] = plugin_rules_data
Expand Down
172 changes: 104 additions & 68 deletions scripts/graph_model.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -270,21 +270,6 @@ def populate_graph_tables(workspace: str, db_path: Optional[str] = None) -> Dict
# init_graph_schema already logged; continue anyway
pass

# Wipe existing CALLS rows so re-scans don't accumulate duplicates.
# Only CALLS edges are managed by populate_graph_tables (they come
# from the flat backend registry). Other edge types (IMPORTS, etc.)
# are managed by their own builders and must survive re-population.
try:
conn.execute(
"DELETE FROM {} WHERE edge_type = 'CALLS'".format(GRAPH_EDGES_TABLE)
)
conn.execute("DELETE FROM {}".format(GRAPH_NODES_TABLE))
except sqlite3.Error as e:
logger.warning(f"populate_graph_tables: clear error: {e}")
conn.rollback()
conn.close()
return {"nodes": 0, "edges": 0}

node_rows: List[Tuple[Any, ...]] = []
for node in flat_nodes:
node_id = node.get("id", "")
Expand Down Expand Up @@ -340,7 +325,31 @@ def populate_graph_tables(workspace: str, db_path: Optional[str] = None) -> Dict
src_file, src_line, confidence, extra_json,
))

# ── Issue #10: RAM-first indexing — batch write in a single
# ``BEGIN EXCLUSIVE`` transaction. The node_rows + edge_rows are
# collected fully in memory (above) BEFORE opening the transaction;
# SQLite then holds an EXCLUSIVE write lock for the shortest possible
# window: one DELETE pass + one ``executemany`` for nodes + one for
# edges, all committed atomically. This replaces the previous flow
# where the schema-clear and the inserts lived in separate implicit
# transactions, doubling the lock-cycle count and interleaving
# poorly with concurrent readers.
#
# We disable the sqlite3 module's implicit BEGIN (isolation_level=None)
# so we can issue the EXCLUSIVE lock explicitly before any DML —
# otherwise the first DELETE would trigger a deferred BEGIN and the
# EXCLUSIVE upgrade would happen later under contention.
try:
conn.isolation_level = None # autocommit; we manage BEGIN/COMMIT
conn.execute("BEGIN EXCLUSIVE")
# Wipe existing CALLS rows so re-scans don't accumulate duplicates.
# Only CALLS edges are managed by populate_graph_tables (they come
# from the flat backend registry). Other edge types (IMPORTS, etc.)
# are managed by their own builders and must survive re-population.
conn.execute(
"DELETE FROM {} WHERE edge_type = 'CALLS'".format(GRAPH_EDGES_TABLE)
)
conn.execute("DELETE FROM {}".format(GRAPH_NODES_TABLE))
if node_rows:
conn.executemany(
"INSERT INTO {t} (node_id, node_type, name, file, line, extra_json) "
Expand All @@ -353,10 +362,13 @@ def populate_graph_tables(workspace: str, db_path: Optional[str] = None) -> Dict
"VALUES (?, ?, ?, ?, ?, ?, ?)".format(t=GRAPH_EDGES_TABLE),
edge_rows,
)
conn.commit()
conn.execute("COMMIT")
except sqlite3.Error as e:
logger.warning(f"populate_graph_tables: insert error: {e}")
conn.rollback()
logger.warning(f"populate_graph_tables: batch write error: {e}")
try:
conn.execute("ROLLBACK")
except sqlite3.Error:
pass
conn.close()
return {"nodes": 0, "edges": 0}

Expand Down Expand Up @@ -494,6 +506,13 @@ def incremental_graph_update(
edges_refined = 0
edges_unresolved = 0

# ── Issue #10: RAM-first indexing — collect all rows in memory
# BEFORE opening the write transaction. The SELECT in step 1 reads
# stale node_ids (a read-only operation that doesn't need an
# EXCLUSIVE lock); the rows for steps 4 + 5 are built entirely from
# the in-memory flat registry (loaded above). Only when all rows are
# ready do we open ``BEGIN EXCLUSIVE`` and run the DELETE + INSERT
# batch in one atomic transaction, minimizing lock duration.
try:
# ── Step 1: Identify stale node ids ──────────────────────
# These are nodes whose file is in the changed set. Their ids
Expand All @@ -510,46 +529,9 @@ def incremental_graph_update(
)
affected_node_ids: Set[str] = {row[0] for row in cursor.fetchall()}

# ── Step 2: Delete affected edges ────────────────────────
# An edge is affected if ANY of:
# - its originating file (`file` column) is in the changed set
# (covers CALLS edges from changed files + IMPORTS edges
# whose importer changed), OR
# - its source_id references a stale node (covers the rare
# case where a CALLS edge has a file column value that
# doesn't match its source_id's file — defensive), OR
# - its target_id references a stale node (covers cross-file
# edges from an unchanged file into a changed file — the
# target may have been renamed/moved, so the edge must be
# dropped and re-resolved from the flat registry).
if affected_node_ids:
ph_nodes = ",".join("?" for _ in affected_node_ids)
params_nodes = tuple(affected_node_ids)
conn.execute(
"DELETE FROM {t} WHERE file IN ({phf}) "
"OR source_id IN ({phn}) "
"OR target_id IN ({phn})".format(
t=GRAPH_EDGES_TABLE, phf=ph_files, phn=ph_nodes
),
params_files + params_nodes + params_nodes,
)
else:
conn.execute(
"DELETE FROM {t} WHERE file IN ({phf})".format(
t=GRAPH_EDGES_TABLE, phf=ph_files
),
params_files,
)

# ── Step 3: Delete stale nodes ───────────────────────────
conn.execute(
"DELETE FROM {t} WHERE file IN ({phf})".format(
t=GRAPH_NODES_TABLE, phf=ph_files
),
params_files,
)

# ── Step 4: Re-insert nodes from changed files ───────────
# ── Step 4 (pre-build): Re-insert nodes from changed files ──
# Built entirely from the in-memory flat registry — no DB reads
# inside the upcoming EXCLUSIVE transaction.
node_rows: List[Tuple[Any, ...]] = []
for node in flat_nodes:
file_val = node.get("file", "")
Expand All @@ -573,14 +555,7 @@ def incremental_graph_update(
(node_id, node_type, name, file_val, line_val, extra_json)
)

if node_rows:
conn.executemany(
"INSERT INTO {t} (node_id, node_type, name, file, line, extra_json) "
"VALUES (?, ?, ?, ?, ?, ?)".format(t=GRAPH_NODES_TABLE),
node_rows,
)

# ── Step 5: Re-insert CALLS edges touching changed files ─
# ── Step 5 (pre-build): Re-insert CALLS edges touching changed files ─
# An edge qualifies for re-insertion if EITHER endpoint's file
# (looked up via the flat registry's node_id → file map) is in
# the changed set. Edges between two unchanged files are
Expand Down Expand Up @@ -644,6 +619,64 @@ def incremental_graph_update(
src_file, src_line, confidence, extra_json,
))

# ── Issue #10: single BEGIN EXCLUSIVE batch write ─────────
# All rows are now in memory. Open an EXCLUSIVE write lock and
# run steps 2 + 3 + 4-insert + 5-insert as one atomic batch:
# DELETE affected edges → DELETE stale nodes → INSERT new nodes
# → INSERT new edges, then COMMIT. Disabling isolation_level
# avoids the sqlite3 module's implicit deferred BEGIN so we get
# the EXCLUSIVE lock upfront (no upgrade deadlock risk).
conn.isolation_level = None # autocommit; we manage BEGIN/COMMIT
conn.execute("BEGIN EXCLUSIVE")

# ── Step 2: Delete affected edges ────────────────────────
# An edge is affected if ANY of:
# - its originating file (`file` column) is in the changed set
# (covers CALLS edges from changed files + IMPORTS edges
# whose importer changed), OR
# - its source_id references a stale node (covers the rare
# case where a CALLS edge has a file column value that
# doesn't match its source_id's file — defensive), OR
# - its target_id references a stale node (covers cross-file
# edges from an unchanged file into a changed file — the
# target may have been renamed/moved, so the edge must be
# dropped and re-resolved from the flat registry).
if affected_node_ids:
ph_nodes = ",".join("?" for _ in affected_node_ids)
params_nodes = tuple(affected_node_ids)
conn.execute(
"DELETE FROM {t} WHERE file IN ({phf}) "
"OR source_id IN ({phn}) "
"OR target_id IN ({phn})".format(
t=GRAPH_EDGES_TABLE, phf=ph_files, phn=ph_nodes
),
params_files + params_nodes + params_nodes,
)
else:
conn.execute(
"DELETE FROM {t} WHERE file IN ({phf})".format(
t=GRAPH_EDGES_TABLE, phf=ph_files
),
params_files,
)

# ── Step 3: Delete stale nodes ───────────────────────────
conn.execute(
"DELETE FROM {t} WHERE file IN ({phf})".format(
t=GRAPH_NODES_TABLE, phf=ph_files
),
params_files,
)

# ── Step 4 (insert): Re-insert nodes from changed files ──
if node_rows:
conn.executemany(
"INSERT INTO {t} (node_id, node_type, name, file, line, extra_json) "
"VALUES (?, ?, ?, ?, ?, ?)".format(t=GRAPH_NODES_TABLE),
node_rows,
)

# ── Step 5 (insert): Re-insert CALLS edges touching changed files ─
if edge_rows:
conn.executemany(
"INSERT INTO {t} "
Expand All @@ -652,10 +685,13 @@ def incremental_graph_update(
edge_rows,
)

conn.commit()
conn.execute("COMMIT")
except sqlite3.Error as e:
logger.warning("incremental_graph_update: db error: %s", e)
conn.rollback()
try:
conn.execute("ROLLBACK")
except sqlite3.Error:
pass
conn.close()
return zero_result

Expand Down
Loading
Loading