class IndexOps:
"""Orchestrates k-mer index build and merge operations against a kmindex registry.
This class drives the full lifecycle of index management: loading index
definitions or span registries from YAML/JSON files, building sub-indexes
via ``IndexBuilder``, merging partial indexes into a combined index, and
cleaning up superseded segments. It supports a plan/dry-run mode that
logs commands without executing them and can emit a shell script of the
equivalent commands for manual replay.
Args:
config: Runtime configuration governing paths, build parameters,
filtering, and execution behaviour. See ``IndexOpsConfig``.
Attributes:
config (IndexOpsConfig): The resolved configuration (paths are
converted to absolute paths on construction).
work_dir (str): Absolute path to the working directory.
asset_dir (str): ``<work_dir>/assets`` - output location for generated
shell scripts.
log_dir (str): ``<work_dir>/logs`` - log file destination.
kmindex_registry_dir (str): Path to the kmindex registry directory.
kmindex_data_dir (str): Path to the folder that holds index data.
timestamp (str): ``YYYYmmdd_HHMMSS`` string captured at construction.
Note:
The execution mode (dry-run, plan, apply, apply with progress) is
controlled by the ``mode`` argument passed to ``run()``. In non-apply
modes, build and merge commands are collected internally and can be
written to a shell script via ``write_script()``.
"""
# MAGIC METHODS
def __init__(self, config: IndexOpsConfig) -> None:
self._config = config
self._config.workdir = os.path.realpath(self.config.workdir)
self._config.index_data_folder = os.path.realpath(self.config.index_data_folder)
self._config.registry_dir = os.path.realpath(self.config.registry_dir)
self._timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
self._mode = ApplyMode.APPLY
self._fail_on_error = True
self._dbs = dict[str, list[IndexDB]]()
self._building = set[str]()
self._loaded_sample_files = set[int]()
self._sample_file_cache = dict[str, dict[str, list[str]]]()
self._script_lines = [
"#!/usr/bin/bash",
f"WORKDIR='{self.work_dir}'",
"cd ${WORKDIR}",
]
logger.debug(f"Init {type(self).__name__}")
logger.debug("workdir: " + self.work_dir)
logger.debug("registry_dir: " + self.kmindex_registry_dir)
logger.debug("asset_dir: " + self.asset_dir)
logger.debug("data_dir: " + self.kmindex_data_dir)
os.makedirs(self.log_dir, exist_ok=True)
os.makedirs(self.asset_dir, exist_ok=True)
os.makedirs(self.kmindex_registry_dir, exist_ok=True)
os.makedirs(self.kmindex_data_dir, exist_ok=True)
# ---
# PROPERTIES AND GETTERS
@property
def config(self) -> IndexOpsConfig:
return self._config
@property
def work_dir(self) -> str:
return self.config.workdir
@property
def asset_dir(self) -> str:
return os.path.join(self.work_dir, "assets")
@property
def log_dir(self) -> str:
return os.path.join(self.work_dir, "logs")
@property
def kmindex_registry_dir(self) -> str:
return self.config.registry_dir
@property
def kmindex_data_dir(self) -> str:
return self.config.index_data_folder
@property
def timestamp(self) -> str:
return self._timestamp
# ---
# PUBLIC METHODS
def write_script(self):
"""Write the collected build/merge commands to a shell script.
The script is placed under ``asset_dir`` and named
``kmhelpers_apply.sh``. Any pre-existing script at that path is
backed up with a ``.bak`` suffix before being overwritten. Only
meaningful after ``run()`` has been called in a non-apply mode, since
that is when commands are accumulated into ``_script_lines``.
"""
script_path = os.path.join(self.asset_dir, "kmhelpers_apply.sh")
if os.path.exists(script_path):
backup_path = script_path + ".bak"
os.replace(script_path, backup_path)
logger.debug(f"Backed up existing script to {backup_path}")
with open(script_path, "w") as f:
f.write("\n".join(self._script_lines) + "\n")
logger.info(f"Script written to {script_path}")
def run(
self, path: str, mode: ApplyMode, fail_on_error: bool = False
) -> ApplyResult:
"""Apply an index definition or span registry file to the kmindex registry.
Reads ``path``, detects whether it is an index definition or a span
registry, then builds any missing sub-indexes and merges partial indexes
as required. Skips any sub-index already present in the registry.
Args:
path: Path to a YAML or JSON file containing either an
``IndexDefinition`` or a span registry.
mode: Execution mode - controls whether to dry-run, plan, or apply.
fail_on_error: Abort this run on the first build or merge error
instead of continuing and returning ``PARTIAL``.
Returns:
An ``ApplyResult`` with the overall status and a per-index details
dict. Status is ``SUCCESS`` when all operations complete without
error, ``PARTIAL`` when at least one operation fails but others
succeed (only when ``fail_on_error=False``), or ``FAILED`` when a
fatal error occurs before any index is built.
"""
path = os.path.realpath(path)
self._mode = mode
self._fail_on_error = fail_on_error
self._building = set[str]()
self._loaded_sample_files = set[int]()
self._sample_file_cache = dict[str, dict[str, list[str]]]()
self._script_lines = [
"#!/usr/bin/bash",
f"WORKDIR='{self.work_dir}'",
"cd ${WORKDIR}",
]
result = ApplyResult()
idt = IndexDefinitionTools()
self._init_result(path, mode, result)
data = self._deserialize_data(path, result, idt)
if data is None or result.input_type is ApplyInputType.UNKNOWN:
logger.error(f"Could not retrieve data type.")
result.status = ApplyStatus.FAILED
return result
dbs = list[IndexDB]()
merges = dict[str, list[str]]()
builder = IndexBuilder(
workdir=self.work_dir,
registry_name=self.kmindex_registry_dir,
data_folder=self.kmindex_data_dir,
log_folder=self.log_dir,
)
try:
if result.input_type is ApplyInputType.INDEX_DEFINITION:
dbs = self._get_dbs(path, idt)
elif result.input_type is ApplyInputType.SPAN_REGISTRY:
dbs, merges = self._load_span_registry(path, idt, data)
except Exception as e:
Log.handle_exception(logger, e, f"Failed to load definition file '{path}'")
result.status = ApplyStatus.FAILED
return result
for db in dbs:
for i in db.index_table.values():
if self._is_filtered(result, i) or not i.name:
continue
logger.info(f"► Processing index definition '{i.name}'...")
if builder.index.has_index(i.name):
logger.info(f" └── {i.name} found in registry: skip")
self._record_run_result(result, i.name, ApplyStatus.NONE)
continue
try:
self._update_span_stats(result, i)
self._build(result, builder, i)
except Exception as e:
if self._handle_error(
result, e, f" Failed to build index '{i.name}'", key=i.name
):
return result
for to_index, parts in merges.items():
try:
self._merge(result, builder, to_index, parts)
except Exception as e:
if self._handle_error(
result, e, f"Failed to merge index '{to_index}'", key=to_index
):
return result
result.status = self._aggregate_status(result)
return result
# ---
# PRIVATE METHODS
def _update_span_stats(self, result: ApplyResult, i: IndexDefinition) -> None:
index_size = i.get_stored_size()
sample_count = i.sample_count
span_data = result.details["span"].setdefault(
i.span, {"sample_count": 0, "bytes": 0, "size_str": "0B"}
)
span_data["sample_count"] += sample_count
span_data["bytes"] += index_size.byte_count
span_data["size_str"] = str(ByteCounter.auto(span_data["bytes"]))
logger.info(f" └── Sample count: {sample_count}")
logger.info(f" └── Estimated build size: {index_size}")
def _is_filtered(self, result: ApplyResult, i: IndexDefinition) -> bool:
# if ApplyInputType.SPAN_REGISTRY, filter has been already applied by _load_span_registry
if result.input_type is not ApplyInputType.INDEX_DEFINITION:
return False
name_filtered = (
self.config.filter_names is not None
and i.name not in self.config.filter_names
)
span_filtered = (
self.config.filter_spans is not None
and i.span not in self.config.filter_spans
)
return name_filtered or span_filtered
def _deserialize_data(
self, path: str, result: ApplyResult, idt: IndexDefinitionTools
) -> dict | None:
data = None
if os.path.isfile(path) and path.endswith((".yaml", ".yml", ".json")):
try:
data = dict(idt.deserialize(path))
except Exception as e:
Log.handle_exception(
logger=logger, msg=f"Could not parse schema from {path}", e=e
)
return None
if data:
try:
type_value = data.get("type")
if type_value == SerializedDataType.INDEX_DEFINITION.value:
result.input_type = ApplyInputType.INDEX_DEFINITION
elif type_value == SerializedDataType.SPAN_DEFINITION.value:
result.input_type = ApplyInputType.SPAN_REGISTRY
else:
result.input_type = ApplyInputType.UNKNOWN
except (ValueError, KeyError) as e:
Log.handle_exception(
logger, e, f"Invalid type value: {data.get('type')}"
)
return None
return data
def _init_result(self, path: str, mode: ApplyMode, result: ApplyResult) -> None:
result.mode = mode
result.status = ApplyStatus.NONE
result.input_type = ApplyInputType.UNKNOWN
wrapper = KmindexWrapper(dry_run=False)
result.details = {
"input_file": path,
"kmindex": {
"version": wrapper.kmindex_version(),
"path": wrapper.which,
},
"kmhelpers": {
"version": pykmhelpers.__version__,
"path": sys.argv[0],
},
"system": {
"os": platform.system(),
"os_version": platform.version(),
},
"span": {},
"run": {},
}
def _append_script(self, cmd: str) -> None:
self._script_lines.append(cmd.replace(self.config.workdir, "${WORKDIR}"))
def _record_run_result(
self,
result: ApplyResult,
name: str,
status: ApplyStatus,
extra: str | None = None,
issues: list[str] | None = None,
) -> None:
entry: dict = {"result": status.value}
if issues:
entry["issues"] = issues
if extra:
entry["error"] = extra
result.details["run"][name] = entry
def _handle_error(
self,
result: ApplyResult,
e: Exception,
msg: str,
key: str | None = None,
) -> bool:
"""Log an error and update result status. Returns True if the caller should abort."""
Log.handle_exception(logger, e, msg)
if key:
self._record_run_result(
result, key, ApplyStatus.FAILED, Log.format_exception(e)
)
result.status = ApplyStatus.PARTIAL
if self._fail_on_error:
result.status = ApplyStatus.FAILED
return True
return False
def _aggregate_status(self, result: ApplyResult) -> ApplyStatus:
"""Derive the overall status from the per-item run results."""
entries = [
v.get("result")
for v in result.details["run"].values()
if isinstance(v, dict)
]
if not entries:
return ApplyStatus.NONE
has_failed = any(r == ApplyStatus.FAILED.value for r in entries)
has_ok = any(r == ApplyStatus.SUCCESS.value for r in entries)
has_partial = any(r == ApplyStatus.PARTIAL.value for r in entries)
if has_partial or (has_failed and has_ok):
return ApplyStatus.PARTIAL
if has_failed:
return ApplyStatus.FAILED
if has_ok:
return ApplyStatus.SUCCESS
return ApplyStatus.NONE # all NONE
def _indent_prefix(self) -> str:
return " └── " if logger.isEnabledFor(logging.INFO) else ""
def _find_source_dir(self, i: IndexDefinition) -> str | None:
for path, dbs in self._dbs.items():
if i.parent_db in dbs:
return os.path.dirname(path)
return None
def _load_sample_file(self, i: IndexDefinition) -> None:
if not i.sample_file or id(i) in self._loaded_sample_files:
return
self._loaded_sample_files.add(id(i))
source_dir = self._find_source_dir(i)
path = (
os.path.join(source_dir, i.sample_file)
if source_dir and not os.path.isabs(i.sample_file)
else i.sample_file
)
if not os.path.isfile(path):
logger.warning(f"Sample file not found: {path}")
return
# Parse JSONL once per file path, reuse across definitions
if path not in self._sample_file_cache:
sample_data: dict[str, list[str]] = {}
with open(path) as f:
header = json.loads(f.readline())
file_root = header.get("root_path", "")
root_path = self.config.sample_rootpath or file_root
for line in f:
data = json.loads(line)
name = data.get("name")
files = data.get("files", [])
if name and files:
sample_data[name] = (
[os.path.join(root_path, fp) for fp in files]
if root_path
else files
)
self._sample_file_cache[path] = sample_data
logger.debug(f"Loaded sample paths from {path}")
# Enrich existing Sample objects with their file paths
sample_data = self._sample_file_cache[path]
for sample in i.samples.values():
if not sample.files:
lookup = sample.get_link(DbFields.ORIGINAL_ID) or sample.name
if lookup and lookup in sample_data:
sample.files = sample_data[lookup]
def _resolve_build_params(
self, i: IndexDefinition, sample_count: int
) -> tuple[int, int, Optional[int]]:
"""Resolve ``(threads, partition_count, chunk_size)`` for building index ``i``.
If ``kmindex_threads`` is explicitly configured, it is used as-is
together with the storage-driven partition count already computed
by ``IndexComposer`` (``i.partition_count``), preserving prior
behaviour exactly; ``chunk_size`` is always ``None`` in that case.
Otherwise, threads and a RAM-driven partition floor are computed
from ``i``'s max k-mer count and ``sample_count`` via
``auto_params``. The returned partition count is never lower than
the storage-driven value, since both are separate constraints on
the same ``--nb-partitions`` flag. If the open-files ceiling can't
fit ``sample_count`` samples in one kmtricks build, ``chunk_size``
is set to the max samples one physical sub-build can hold; the
caller is expected to split the build into
``ceil(sample_count / chunk_size)`` chunks (see ``_build_chunked``).
"""
partition_count = self.config.partition_count or i.partition_count
if self.config.kmindex_threads:
return self.config.kmindex_threads, partition_count, None
kmers = max((s.kmer_count for s in i.samples.values()), default=0)
params = auto_params(
kmers=kmers,
samples=sample_count,
limits=self.config.limits or "{}",
safety_margin=self.config.safety_margin,
)
if (
params.threads is None
or params.samples is None
or params.partitions is None
):
raise TypeError(
f"expected auto_params() to set threads/samples/partitions, got "
f"threads={params.threads!r}, samples={params.samples!r}, "
f"partitions={params.partitions!r}"
)
chunk_size = params.samples if params.samples < sample_count else None
if chunk_size is not None:
logger.info(
f" └── '{i.name}' has {sample_count} samples, exceeding the "
f"{chunk_size} a single kmtricks build can fit under the "
f"current open-files limit; splitting into chunks"
)
logger.debug(
f" └── Auto-sized: threads={params.threads}, "
f"partitions={max(partition_count, params.partitions)}"
)
return params.threads, max(partition_count, params.partitions), chunk_size
def _build_one(
self,
builder: IndexBuilder,
name: str,
fof: FofManager,
i: IndexDefinition,
threads: int,
partition_count: int,
) -> dict | None:
"""Build a single physical sub-index, showing a progress spinner or bar if configured.
Delegates to ``IndexBuilder.create_subindex`` using ``i``'s shared
build parameters (bloom size, abundance min, k-mer size) with a
caller-supplied ``name``/``fof``, so this can build either the
whole index or one chunk of it (see ``_build_chunked``). The
generated command is appended to ``_script_lines`` for later script
export.
"""
# --- Progress handlers setup
stop_event = None
wait_handler = None
progress_handler = None
if self._mode == ApplyMode.APPLY_SHOW_PROGRESS:
start = datetime.now()
stop_event = threading.Event()
def _progress_worker():
sleep(2)
s = 0
wait_steps = ["⢿", "⣻", "⣽", "⣾", "⣷", "⣯", "⣟", "⡿"]
while not stop_event.wait(timeout=0.5):
print(
f"\r\033[1;32m{wait_steps[s]} Building index '{name}'...\033[0m ",
end="",
flush=True,
)
s = (s + 1) % len(wait_steps)
wait_handler = threading.Thread(target=_progress_worker, daemon=True)
wait_handler.start()
def _on_progress(value: float):
elapsed = (datetime.now() - start).total_seconds()
bar_len = 30
filled = int(round(bar_len * value))
bar = "■" * filled + " " * (bar_len - filled)
print(
f"\r[{bar}] {value * 100:.1f}% elapsed: {int(elapsed // 60)}m{int(elapsed % 60):02d}s ",
end="",
flush=True,
)
if stop_event:
stop_event.set()
if wait_handler:
wait_handler.join()
progress_handler = IndexBuilder.Progress(_on_progress, delay=60)
elif self._mode >= ApplyMode.APPLY:
logger.info(f" └── Building '{name}'...")
# --- Call builder
try:
result = builder.create_subindex(
name=name,
samples=fof,
abundance_min=i.abundance_min,
bloom_size=i.bf_size,
n_partitions=partition_count,
n_threads=threads,
auto_check=True,
compress_intermediate=not self.config.kmindex_skip_compression,
minim_size=self.config.minimizer_length,
dry_run=self._mode < ApplyMode.APPLY,
kmer_size=i.kmer_size,
on_existing=self.config.on_existing,
progress=progress_handler,
)
if result and "command" in result:
self._append_script(result["command"])
finally:
if stop_event:
stop_event.set()
if wait_handler:
wait_handler.join()
# --- Post-build verification
if self._mode >= ApplyMode.APPLY:
builder.index.load_json()
if not builder.has_subindex(name):
raise RuntimeError(f"Could not find index '{name}'")
return result
def _build_chunked(
self,
builder: IndexBuilder,
i: IndexDefinition,
fof: FofManager,
chunk_size: int,
threads: int,
partition_count: int,
) -> dict | None:
"""Build index ``i`` as several ``chunk_size``-sized sub-builds, then merge them.
Used when the open-files ceiling can't fit every sample of ``i`` in
a single kmtricks build (see ``_resolve_build_params``). Each chunk
is built under a transient ``{i.name}__chunk{n}`` name via
``_build_one``, then merged into ``i.name`` via the same
``builder.merge()`` call ``_merge`` uses for span-registry merges,
after which the transient chunk sub-indexes are deleted.
"""
if not i.name:
raise ValueError("IndexDefinition is missing required 'name' field")
items = list(fof.samples.items())
chunks = [items[n : n + chunk_size] for n in range(0, len(items), chunk_size)]
chunk_names = [f"{i.name}__chunk{n}" for n in range(len(chunks))]
logger.info(
f" └── Splitting '{i.name}' into {len(chunks)} chunks of up to "
f"{chunk_size} samples (open-files limit)"
)
for chunk_name, chunk_items in zip(chunk_names, chunks):
chunk_fof = FofManager(samples=dict(chunk_items))
self._build_one(builder, chunk_name, chunk_fof, i, threads, partition_count)
result = builder.merge(
i.name,
chunk_names,
delete_old=False,
dry_run=self._mode < ApplyMode.APPLY,
threads=threads,
)
if result and "command" in result:
self._append_script(result["command"])
if self._mode >= ApplyMode.APPLY:
builder.index.load_json()
if not builder.has_subindex(i.name):
raise RuntimeError(f"Could not find index '{i.name}'")
if builder.index.get_index(i.name).check_structure():
for chunk_name in chunk_names:
self._delete_segment(builder, chunk_name)
return result
def _run_build(
self, builder: IndexBuilder, i: IndexDefinition
) -> tuple[dict | None, list[str]]:
"""Build a single sub-index, showing a progress spinner or bar if configured.
Resolves sample file paths, populates a ``FofManager``, and delegates
the actual build to ``_build_one`` or, when the open-files ceiling
requires it, to ``_build_chunked``.
Args:
builder: The ``IndexBuilder`` instance managing the target registry.
i: The index definition describing the sub-index to build.
Returns:
A ``(build_result, issues)`` tuple where ``build_result`` is the
dict returned by the build/merge call (or ``None`` if the index
was already building or no samples were added), and ``issues``
is a list of sample-level warning strings.
"""
if not i.name:
raise ValueError("IndexDefinition is missing required 'name' field")
if i.name in self._building or builder.has_subindex(i.name):
return None, []
if not i.bf_size:
raise ValueError(
f"IndexDefinition {i.name} is missing required 'bf_size' field"
)
# --- Build FofManager from samples
self._load_sample_file(i)
fof = FofManager()
issues: list[str] = []
for s in i.samples.values():
self._add_sample_to_fof(i, fof, s, issues)
result = None
self._building.add(i.name)
if fof.get_sample_count() > 0:
threads, partition_count, chunk_size = self._resolve_build_params(
i, fof.get_sample_count()
)
if chunk_size is not None:
result = self._build_chunked(
builder, i, fof, chunk_size, threads, partition_count
)
else:
result = self._build_one(
builder, i.name, fof, i, threads, partition_count
)
else:
logger.warning(
f"{self._indent_prefix()}Skipping index '{i.name}' as no sample was added to it"
)
return result, issues
def _add_sample_to_fof(
self, i: IndexDefinition, fof: FofManager, s: Sample, issues: list[str]
) -> None:
try:
if not s.name:
raise ValueError("Empty name")
if not s.files:
raise ValueError("Empty file list")
if s.name != "_":
sample_files = (
[
(
os.path.join(self.config.sample_rootpath, f)
if not os.path.isabs(f)
else f
)
for f in s.files
]
if self.config.sample_rootpath
else s.files
)
if self._mode > ApplyMode.DRY_RUN:
for f in sample_files:
if not os.path.isfile(f):
raise FileNotFoundError(f"Sample file not found: {f}")
fof.add_sample(sample_files, s.name)
except Exception as e:
msg = f"Error adding sample '{s.name or 'UNNAMED'}' to '{i.name}' | {e}"
Log.handle_exception(
logger=logger,
e=e,
msg=f"{self._indent_prefix()}{msg}",
level=logging.WARNING,
)
issues.append(msg)
def _get_dbs(self, path: str, idt: IndexDefinitionTools) -> list[IndexDB]:
"""Load and cache ``IndexDB`` objects from a definition file.
Args:
path: Absolute path to the definition file.
idt: ``IndexDefinitionTools`` instance used for deserialization.
Returns:
A list of ``IndexDB`` objects loaded from ``path``. Subsequent
calls with the same path return the cached result without re-reading
the file.
"""
if path in self._dbs:
dbs = self._dbs[path]
else:
dbs = idt.load_db(path)
self._dbs[path] = dbs
return dbs
def _load_span_registry(
self, path: str, idt: IndexDefinitionTools, data: dict
) -> tuple[list[IndexDB], dict[str, list[str]]]:
"""Parse a span registry and return ``(dbs, merges)``.
Iterates over each span in the registry, optionally filtering by
``config.filter_spans`` and ``config.filter_names``. For each index
that passes the filter, the merge target and its constituent sub-index
names are recorded in ``merges``, and the corresponding definition files
are loaded into ``dbs``.
Args:
path: Absolute path to the span registry file (used to resolve
sibling definition files).
idt: ``IndexDefinitionTools`` instance for loading definition files.
data: Deserialized registry dict (``{"data": {span_id: {...}}}``)
Returns:
A ``(dbs, merges)`` tuple where ``dbs`` is a list of ``IndexDB``
objects and ``merges`` maps each merge target to its sub-index names.
Raises:
ValueError: If a span entry is missing the ``"indices"`` field.
FileNotFoundError: If a required definition file does not exist on disk.
"""
dbs = list[IndexDB]()
merges = dict[str, list[str]]()
spans = dict[int, dict](data["data"])
for to_index, parts in spans.items():
if self.config.filter_spans and to_index not in self.config.filter_spans:
continue
parts = parts.get("indices")
if not parts:
raise ValueError("Span registry is missing field 'indices'")
indices = dict[str, list[str]](parts)
for name, subindices in indices.items():
# if len(subindices) == 1:
# # Single sub-index: build directly under the merge target name
# if not self.config.filter_names or name in self.config.filter_names:
# subindex = subindices[0]
# db_path = os.path.join(
# os.path.dirname(path),
# subindex + os.path.splitext(path)[1],
# )
# assert os.path.isfile(
# db_path
# ), f"Could not find required data file at {db_path}"
# loaded_dbs = self._get_dbs(db_path, idt)
# for db in loaded_dbs:
# if subindex in db.index_table:
# db.index_table[subindex].name = name
# dbs.extend(loaded_dbs)
# else:
if not self.config.filter_names or name in self.config.filter_names:
merges[name] = subindices
for subindex in subindices:
if name in merges or (
self.config.filter_names
and subindex in self.config.filter_names
):
db_path = os.path.join(
os.path.dirname(path),
subindex + os.path.splitext(path)[1],
)
if not os.path.isfile(db_path):
raise FileNotFoundError(
f"Could not find required data file at {db_path}"
)
dbs.extend(self._get_dbs(db_path, idt))
return dbs, merges
def _build(
self, result: ApplyResult, builder: IndexBuilder, i: IndexDefinition
) -> None:
"""Build a sub-index and record the outcome in ``result``.
Args:
result: The ``ApplyResult`` being accumulated; ``details`` is
updated in place.
builder: The ``IndexBuilder`` managing the target registry.
i: The ``IndexDefinition`` of the index to build.
Raises:
ValueError: If ``name`` or ``bf_size`` is not set on the definition.
"""
if not i.name:
raise ValueError("IndexDefinition is missing required 'name' field")
if not i.bf_size > 0:
raise ValueError(
f"IndexDefinition {i.name} is missing required 'bf_size' field"
)
builder.index.load_json()
build_result, issues = self._run_build(builder, i)
if build_result:
if self._mode < ApplyMode.APPLY or build_result.get("return_code", -1) == 0:
self._record_run_result(
result, i.name, ApplyStatus.SUCCESS, issues=issues
)
else:
self._record_run_result(
result,
i.name,
ApplyStatus.FAILED,
extra=f"error_code={build_result['return_code']}",
issues=issues,
)
else:
status = ApplyStatus.FAILED if issues else ApplyStatus.NONE
self._record_run_result(result, i.name, status, issues=issues or None)
def _merge(
self,
result: ApplyResult,
builder: IndexBuilder,
to_index: str,
parts: list[str],
) -> None:
"""Merge a list of sub-indexes into a combined index and clean up the parts.
Verifies that all constituent sub-indexes are present in the registry
(skipped in plan mode), calls ``IndexBuilder.merge``, and if the
resulting index passes a structure check, removes each constituent
segment via ``_delete_segment``.
Args:
result: The ``ApplyResult`` being accumulated; ``details`` is
updated in place.
builder: The ``IndexBuilder`` managing the target registry.
to_index: Name of the target merged index.
parts: List of sub-index names to merge into ``to_index``.
Raises:
Exception: If the builder returns a malformed result dict.
RuntimeError: If the merged sub-index is not found in the registry
after the merge.
"""
builder.index.load_json()
missing = None
if self._mode >= ApplyMode.APPLY:
missing = [name for name in parts if not builder.index.has_index(name)]
if missing:
logger.warning(
f"Cannot merge '{to_index}' due to some sub-indexes missing: {missing}"
)
result.details["run"][to_index] = {
"result": ApplyStatus.FAILED.value,
"error": f"Missing sub-indexes: {missing}",
}
else:
merge_result = builder.merge(
to_index,
parts,
delete_old=False,
dry_run=self._mode < ApplyMode.APPLY,
threads=self._config.kmindex_threads or os.cpu_count() or 1,
)
if merge_result and "command" in merge_result:
self._append_script(merge_result["command"])
merge_entry: dict = {"result": ApplyStatus.SUCCESS.value}
for sub in parts:
merge_entry[sub] = result.details["run"].pop(
sub, {"result": ApplyStatus.NONE.value}
)
sub_results = [
v.get("result") for v in merge_entry.values() if isinstance(v, dict)
]
if sub_results and all(
r == ApplyStatus.FAILED.value for r in sub_results
):
merge_entry["result"] = ApplyStatus.FAILED.value
elif any(r == ApplyStatus.FAILED.value for r in sub_results):
merge_entry["result"] = ApplyStatus.PARTIAL.value
elif sub_results and all(
r == ApplyStatus.NONE.value for r in sub_results
):
merge_entry["result"] = ApplyStatus.NONE.value
else:
merge_entry["result"] = ApplyStatus.SUCCESS.value
result.details["run"][to_index] = merge_entry
else:
raise Exception("Malformed result")
if self._mode >= ApplyMode.APPLY:
builder.index.load_json()
if not builder.has_subindex(to_index):
raise RuntimeError(f"Sub-index {to_index} not found")
if builder.index.get_index(to_index).check_structure():
for segment in parts:
self._delete_segment(builder, segment)
def _delete_segment(self, builder: IndexBuilder, segment: str) -> None:
"""Remove a segment sub-index from the registry and delete its files.
Unregisters ``segment`` from the kmindex registry (ignoring it if
already unregistered), then removes the corresponding directory under
``index_data_folder`` and any dangling symlink at that path. Failures
during file deletion are logged as warnings rather than raised.
Args:
builder: The ``IndexBuilder`` whose registry entry should be removed.
segment: Name of the sub-index segment to delete.
"""
logger.info(f"Delete {segment}...")
# --- Unregister from registry
try:
builder.index.remove_index(
segment, delete_files=False, skip_unregistered=True
)
except Exception as e:
Log.handle_exception(
logger, e, f"Failed to remove {segment} from registry", logging.WARNING
)
index_path = os.path.join(self.config.index_data_folder, segment)
# --- Delete files from disk
if self._mode >= ApplyMode.APPLY:
try:
# Get index before removal (needed to delete files)
shutil.rmtree(
os.path.realpath(index_path),
ignore_errors=True,
)
except Exception as e:
Log.handle_exception(
logger,
e,
f"Failed to delete some files",
logging.WARNING,
)
try:
if os.path.islink(index_path):
os.unlink(index_path)
except Exception as e:
Log.handle_exception(
logger,
e,
f"Error deleting link {index_path}",
logging.WARNING,
)
if os.path.exists(index_path):
logger.warning(
f"Could not remove dir {index_path}, please remove it manually."
)
else:
self._append_script(f"rm -rf $(realpath {index_path})")
self._append_script(f"[ -L {index_path} ] && unlink {index_path}")