Skip to content

Benchmarking

vector_search_study.benchmarking

Deterministic workload and implementation definitions for benchmark matrices.

WorkloadSpec dataclass

One deterministic semantic exact-search workload.

Source code in src/vector_search_study/benchmarking.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
@dataclass(frozen=True, slots=True)
class WorkloadSpec:
    """One deterministic semantic exact-search workload."""

    objective: SearchObjective
    corpus_size: int
    dimension: int
    query_count: int
    k: int
    dtype: str = "float32"
    seed: int = DEFAULT_SEED
    profile: str = "discovery-core"

    def __post_init__(self) -> None:
        """Validate and canonicalize workload fields."""
        object.__setattr__(self, "objective", resolve_search_objective(self.objective))
        object.__setattr__(self, "corpus_size", validate_positive_int(self.corpus_size, name="corpus_size"))
        object.__setattr__(self, "dimension", validate_positive_int(self.dimension, name="dimension"))
        object.__setattr__(self, "query_count", validate_positive_int(self.query_count, name="query_count"))
        resolved_k = validate_positive_int(self.k, name="k")
        if resolved_k > self.corpus_size:
            raise ValueError("k must not exceed corpus_size")
        object.__setattr__(self, "k", resolved_k)
        object.__setattr__(self, "dtype", resolve_float_dtype(self.dtype).name)
        if isinstance(self.seed, bool) or not isinstance(self.seed, int) or self.seed < 0:
            raise ValueError("seed must be a non-negative integer")
        if not self.profile:
            raise ValueError("profile must be non-empty")

    @property
    def name(self) -> str:
        """Return a stable human-readable case name."""
        return (
            f"{self.profile}__{self.objective.value}__n{self.corpus_size}__d{self.dimension}"
            f"__q{self.query_count}__k{self.k}__{self.dtype}"
        )

    @property
    def coordinate_evaluations(self) -> int:
        """Return scalar query-corpus coordinate evaluations per search."""
        return self.corpus_size * self.dimension * self.query_count

    def metadata(self) -> dict[str, object]:
        """Return strict-JSON-safe deterministic benchmark metadata."""
        distribution = "uniform_sphere" if self.objective.requires_normalization else "gaussian"
        return {
            "schema_version": _SCHEMA_VERSION,
            "study": "exact_top_k_vector_search",
            "profile": self.profile,
            "objective": self.objective.value,
            "score_convention": _score_convention(self.objective),
            "corpus_size": self.corpus_size,
            "dimension": self.dimension,
            "query_count": self.query_count,
            "k": self.k,
            "dtype": self.dtype,
            "normalization": "l2_unit_rows" if self.objective.requires_normalization else "none",
            "dataset_family": "synthetic",
            "distribution": distribution,
            "generator": "numpy.random.PCG64",
            "generator_revision": 1,
            "seed": self.seed,
            "dataset_id": (
                f"synthetic-pcg64-v1:{distribution}:{self.corpus_size}:{self.dimension}:"
                f"{self.query_count}:{self.dtype}:{self.seed}"
            ),
            "boundary_policy": "strict_top_k_margin",
            "threads": 1,
        }

    def make_dataset(self) -> SyntheticDataset:
        """Materialize the deterministic corpus and queries."""
        if self.objective.requires_normalization:
            return make_uniform_sphere_dataset(
                self.corpus_size,
                self.dimension,
                self.query_count,
                dtype=self.dtype,
                seed=self.seed,
            )
        return make_gaussian_dataset(
            self.corpus_size,
            self.dimension,
            self.query_count,
            objective=self.objective,
            dtype=self.dtype,
            seed=self.seed,
        )

name property 

name: str

Return a stable human-readable case name.

coordinate_evaluations property 

coordinate_evaluations: int

Return scalar query-corpus coordinate evaluations per search.

__post_init__ 

__post_init__() -> None

Validate and canonicalize workload fields.

Source code in src/vector_search_study/benchmarking.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def __post_init__(self) -> None:
    """Validate and canonicalize workload fields."""
    object.__setattr__(self, "objective", resolve_search_objective(self.objective))
    object.__setattr__(self, "corpus_size", validate_positive_int(self.corpus_size, name="corpus_size"))
    object.__setattr__(self, "dimension", validate_positive_int(self.dimension, name="dimension"))
    object.__setattr__(self, "query_count", validate_positive_int(self.query_count, name="query_count"))
    resolved_k = validate_positive_int(self.k, name="k")
    if resolved_k > self.corpus_size:
        raise ValueError("k must not exceed corpus_size")
    object.__setattr__(self, "k", resolved_k)
    object.__setattr__(self, "dtype", resolve_float_dtype(self.dtype).name)
    if isinstance(self.seed, bool) or not isinstance(self.seed, int) or self.seed < 0:
        raise ValueError("seed must be a non-negative integer")
    if not self.profile:
        raise ValueError("profile must be non-empty")

metadata 

metadata() -> dict[str, object]

Return strict-JSON-safe deterministic benchmark metadata.

Source code in src/vector_search_study/benchmarking.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def metadata(self) -> dict[str, object]:
    """Return strict-JSON-safe deterministic benchmark metadata."""
    distribution = "uniform_sphere" if self.objective.requires_normalization else "gaussian"
    return {
        "schema_version": _SCHEMA_VERSION,
        "study": "exact_top_k_vector_search",
        "profile": self.profile,
        "objective": self.objective.value,
        "score_convention": _score_convention(self.objective),
        "corpus_size": self.corpus_size,
        "dimension": self.dimension,
        "query_count": self.query_count,
        "k": self.k,
        "dtype": self.dtype,
        "normalization": "l2_unit_rows" if self.objective.requires_normalization else "none",
        "dataset_family": "synthetic",
        "distribution": distribution,
        "generator": "numpy.random.PCG64",
        "generator_revision": 1,
        "seed": self.seed,
        "dataset_id": (
            f"synthetic-pcg64-v1:{distribution}:{self.corpus_size}:{self.dimension}:"
            f"{self.query_count}:{self.dtype}:{self.seed}"
        ),
        "boundary_policy": "strict_top_k_margin",
        "threads": 1,
    }

make_dataset 

make_dataset() -> SyntheticDataset

Materialize the deterministic corpus and queries.

Source code in src/vector_search_study/benchmarking.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def make_dataset(self) -> SyntheticDataset:
    """Materialize the deterministic corpus and queries."""
    if self.objective.requires_normalization:
        return make_uniform_sphere_dataset(
            self.corpus_size,
            self.dimension,
            self.query_count,
            dtype=self.dtype,
            seed=self.seed,
        )
    return make_gaussian_dataset(
        self.corpus_size,
        self.dimension,
        self.query_count,
        objective=self.objective,
        dtype=self.dtype,
        seed=self.seed,
    )

NaturalDatasetSpec dataclass

Pinned manifest for the approved natural-embedding slice.

Source code in src/vector_search_study/benchmarking.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
@dataclass(frozen=True, slots=True)
class NaturalDatasetSpec:
    """Pinned manifest for the approved natural-embedding slice."""

    dataset: str = "BeIR/scifact"
    dataset_revision: str = "a75ae049398addde9b70f6b268875f5cbce99089"  # pragma: allowlist secret
    model: str = "sentence-transformers/all-MiniLM-L6-v2"
    model_revision: str = "c9745ed1d9f207416be6d2e6f8de32d1f16199bf"  # pragma: allowlist secret
    dimension: int = 384
    corpus_size: int = 5_183
    query_count: int = 1_109

    def metadata(self) -> dict[str, object]:
        """Return the immutable natural-data provenance manifest."""
        return {
            "schema_version": _SCHEMA_VERSION,
            "dataset_family": "natural",
            "dataset": self.dataset,
            "dataset_revision": self.dataset_revision,
            "model": self.model,
            "model_revision": self.model_revision,
            "dimension": self.dimension,
            "corpus_size": self.corpus_size,
            "query_count": self.query_count,
            "dtype": "float32",
            "normalization": "l2_unit_rows",
        }

metadata 

metadata() -> dict[str, object]

Return the immutable natural-data provenance manifest.

Source code in src/vector_search_study/benchmarking.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def metadata(self) -> dict[str, object]:
    """Return the immutable natural-data provenance manifest."""
    return {
        "schema_version": _SCHEMA_VERSION,
        "dataset_family": "natural",
        "dataset": self.dataset,
        "dataset_revision": self.dataset_revision,
        "model": self.model,
        "model_revision": self.model_revision,
        "dimension": self.dimension,
        "corpus_size": self.corpus_size,
        "query_count": self.query_count,
        "dtype": "float32",
        "normalization": "l2_unit_rows",
    }

ImplementationSpec dataclass

A named implementation and its exact native capabilities.

Source code in src/vector_search_study/benchmarking.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
@dataclass(frozen=True, slots=True)
class ImplementationSpec:
    """A named implementation and its exact native capabilities."""

    name: str
    objectives: frozenset[SearchObjective]
    builder: SearcherBuilder
    optional_dependency: str | None = None
    float32_only: bool = False

    def build(self, corpus: FloatMatrix, objective: SearchObjective) -> ExactSearcher:
        """Build the search index after validating semantic capability."""
        resolved = resolve_search_objective(objective)
        if resolved not in self.objectives:
            raise ValueError(f"{self.name} does not support {resolved.value}")
        if self.float32_only and corpus.dtype != np.dtype(np.float32):
            raise ValueError(f"{self.name} supports only float32")
        return self.builder(corpus, resolved)

    def metadata(self) -> dict[str, object]:
        """Return deterministic implementation metadata."""
        return {
            "implementation": self.name,
            "objectives": sorted(objective.value for objective in self.objectives),
            "optional_dependency": self.optional_dependency,
            "float32_only": self.float32_only,
        }

build 

build(
    corpus: FloatMatrix, objective: SearchObjective
) -> ExactSearcher

Build the search index after validating semantic capability.

Source code in src/vector_search_study/benchmarking.py 
166
167
168
169
170
171
172
173
def build(self, corpus: FloatMatrix, objective: SearchObjective) -> ExactSearcher:
    """Build the search index after validating semantic capability."""
    resolved = resolve_search_objective(objective)
    if resolved not in self.objectives:
        raise ValueError(f"{self.name} does not support {resolved.value}")
    if self.float32_only and corpus.dtype != np.dtype(np.float32):
        raise ValueError(f"{self.name} supports only float32")
    return self.builder(corpus, resolved)

metadata 

metadata() -> dict[str, object]

Return deterministic implementation metadata.

Source code in src/vector_search_study/benchmarking.py 
175
176
177
178
179
180
181
182
def metadata(self) -> dict[str, object]:
    """Return deterministic implementation metadata."""
    return {
        "implementation": self.name,
        "objectives": sorted(objective.value for objective in self.objectives),
        "optional_dependency": self.optional_dependency,
        "float32_only": self.float32_only,
    }

FeasibilityDecision dataclass

Deterministic reason for including or excluding a matrix cell.

Source code in src/vector_search_study/benchmarking.py 
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
@dataclass(frozen=True, slots=True)
class FeasibilityDecision:
    """Deterministic reason for including or excluding a matrix cell."""

    feasible: bool
    reason: str
    estimated_peak_bytes: int

    def metadata(self) -> dict[str, object]:
        """Return strict-JSON-safe inclusion metadata."""
        return {
            "feasible": self.feasible,
            "reason": self.reason,
            "estimated_peak_bytes": self.estimated_peak_bytes,
        }

metadata 

metadata() -> dict[str, object]

Return strict-JSON-safe inclusion metadata.

Source code in src/vector_search_study/benchmarking.py 
264
265
266
267
268
269
270
def metadata(self) -> dict[str, object]:
    """Return strict-JSON-safe inclusion metadata."""
    return {
        "feasible": self.feasible,
        "reason": self.reason,
        "estimated_peak_bytes": self.estimated_peak_bytes,
    }

DiscoveryCellPlan dataclass

One implementation/workload decision in a discovery profile.

Source code in src/vector_search_study/benchmarking.py 
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
@dataclass(frozen=True, slots=True)
class DiscoveryCellPlan:
    """One implementation/workload decision in a discovery profile."""

    workload: WorkloadSpec
    implementation: ImplementationSpec
    decision: FeasibilityDecision
    dependency_installed: bool
    metrics: tuple[str, ...]

    @property
    def included(self) -> bool:
        """Return whether this cell can be collected on the current host."""
        return self.decision.feasible and self.dependency_installed

    def metadata(self) -> dict[str, object]:
        """Return strict-JSON-safe plan metadata."""
        reason = self.decision.reason
        if self.decision.feasible and not self.dependency_installed:
            reason = "dependency_not_installed"
        return {
            "workload": self.workload.name,
            "implementation": self.implementation.name,
            "included": self.included,
            "reason": reason,
            "estimated_peak_bytes": self.decision.estimated_peak_bytes,
            "dependency": self.implementation.optional_dependency,
            "dependency_installed": self.dependency_installed,
            "metrics": list(self.metrics),
        }

included property 

included: bool

Return whether this cell can be collected on the current host.

metadata 

metadata() -> dict[str, object]

Return strict-JSON-safe plan metadata.

Source code in src/vector_search_study/benchmarking.py 
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
def metadata(self) -> dict[str, object]:
    """Return strict-JSON-safe plan metadata."""
    reason = self.decision.reason
    if self.decision.feasible and not self.dependency_installed:
        reason = "dependency_not_installed"
    return {
        "workload": self.workload.name,
        "implementation": self.implementation.name,
        "included": self.included,
        "reason": reason,
        "estimated_peak_bytes": self.decision.estimated_peak_bytes,
        "dependency": self.implementation.optional_dependency,
        "dependency_installed": self.dependency_installed,
        "metrics": list(self.metrics),
    }

assess_feasibility 

assess_feasibility(
    workload: WorkloadSpec,
    implementation: ImplementationSpec,
    *,
    memory_budget_bytes: int = DEFAULT_MEMORY_BUDGET_BYTES,
) -> FeasibilityDecision

Apply semantic, runtime, and conservative memory feasibility rules.

Source code in src/vector_search_study/benchmarking.py 
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
def assess_feasibility(
    workload: WorkloadSpec,
    implementation: ImplementationSpec,
    *,
    memory_budget_bytes: int = DEFAULT_MEMORY_BUDGET_BYTES,
) -> FeasibilityDecision:
    """Apply semantic, runtime, and conservative memory feasibility rules."""
    budget = validate_positive_int(memory_budget_bytes, name="memory_budget_bytes")
    peak = estimate_peak_bytes(workload, implementation.name)
    if workload.objective not in implementation.objectives:
        return FeasibilityDecision(False, "objective_not_supported", peak)
    if implementation.float32_only and workload.dtype != "float32":
        return FeasibilityDecision(False, "dtype_not_supported", peak)
    if implementation.name.startswith("python_") and workload.coordinate_evaluations > 1_000_000:
        return FeasibilityDecision(False, "scalar_runtime_limit", peak)
    if peak > budget:
        return FeasibilityDecision(False, "memory_budget_exceeded", peak)
    return FeasibilityDecision(True, "included", peak)

estimate_peak_bytes 

estimate_peak_bytes(
    workload: WorkloadSpec, implementation_name: str
) -> int

Estimate dominant corpus, score, selection, and index allocations.

Source code in src/vector_search_study/benchmarking.py 
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
def estimate_peak_bytes(workload: WorkloadSpec, implementation_name: str) -> int:
    """Estimate dominant corpus, score, selection, and index allocations."""
    item_size = np.dtype(workload.dtype).itemsize
    corpus = workload.corpus_size * workload.dimension * item_size
    queries = workload.query_count * workload.dimension * item_size
    scores = workload.corpus_size * workload.query_count * item_size
    output = workload.query_count * workload.k * (item_size + np.dtype(np.int64).itemsize)
    cached_dataset_and_index = corpus * 2 + queries + output
    if implementation_name == "numpy_full":
        return cached_dataset_and_index + scores + workload.corpus_size * workload.query_count * 8
    if implementation_name in {"numpy_argpartition", "torch_matmul_topk"}:
        return cached_dataset_and_index + scores + workload.corpus_size * 8
    if implementation_name == "numpy_blocked":
        return cached_dataset_and_index + min(workload.corpus_size, 16_384) * workload.query_count * item_size + output
    if implementation_name.startswith(("sklearn_", "scipy_")):
        return cached_dataset_and_index + corpus * 2
    if implementation_name.startswith("faiss_"):
        return cached_dataset_and_index + corpus
    return cached_dataset_and_index

discovery_core_specs 

discovery_core_specs() -> tuple[WorkloadSpec, ...]

Return the 33-case one-factor-at-a-time discovery core.

Source code in src/vector_search_study/benchmarking.py 
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
def discovery_core_specs() -> tuple[WorkloadSpec, ...]:
    """Return the 33-case one-factor-at-a-time discovery core."""
    anchor = (10_000, 128, 32, 10)
    cases: list[WorkloadSpec] = []
    for objective in OBJECTIVES:
        cases.append(WorkloadSpec(objective, *anchor))
        for dimension in DIMENSIONS:
            if dimension != anchor[1]:
                cases.append(WorkloadSpec(objective, anchor[0], dimension, anchor[2], anchor[3]))
        for corpus_size in CORPUS_SIZES:
            if corpus_size != anchor[0]:
                cases.append(WorkloadSpec(objective, corpus_size, anchor[1], anchor[2], anchor[3]))
        for query_count in QUERY_COUNTS:
            if query_count != anchor[2]:
                cases.append(WorkloadSpec(objective, anchor[0], anchor[1], query_count, anchor[3]))
        for k in K_VALUES:
            if k != anchor[3]:
                cases.append(WorkloadSpec(objective, anchor[0], anchor[1], anchor[2], k))
    return tuple(cases)

small_specs 

small_specs() -> tuple[WorkloadSpec, ...]

Return small cases on which scalar Python implementations are credible.

Source code in src/vector_search_study/benchmarking.py 
367
368
369
370
371
372
373
def small_specs() -> tuple[WorkloadSpec, ...]:
    """Return small cases on which scalar Python implementations are credible."""
    return tuple(
        WorkloadSpec(objective, 1_000, dimension, 1, 10, profile="discovery-small")
        for objective in OBJECTIVES
        for dimension in DIMENSIONS
    )

stress_specs 

stress_specs() -> tuple[WorkloadSpec, ...]

Return core cases that isolate a high-cost factor.

Source code in src/vector_search_study/benchmarking.py 
376
377
378
379
380
381
382
def stress_specs() -> tuple[WorkloadSpec, ...]:
    """Return core cases that isolate a high-cost factor."""
    return tuple(
        replace(spec, profile="discovery-stress")
        for spec in discovery_core_specs()
        if spec.corpus_size == 1_000_000 or spec.query_count == 1_024 or spec.dimension == 768
    )

standard_specs 

standard_specs() -> tuple[WorkloadSpec, ...]

Return discovery-core cases with stress factors collected separately.

Source code in src/vector_search_study/benchmarking.py 
385
386
387
388
389
390
391
392
393
394
def standard_specs() -> tuple[WorkloadSpec, ...]:
    """Return discovery-core cases with stress factors collected separately."""
    stress_coordinates = {
        (spec.objective, spec.corpus_size, spec.dimension, spec.query_count, spec.k) for spec in stress_specs()
    }
    return tuple(
        spec
        for spec in discovery_core_specs()
        if (spec.objective, spec.corpus_size, spec.dimension, spec.query_count, spec.k) not in stress_coordinates
    )

smoke_specs 

smoke_specs() -> tuple[WorkloadSpec, ...]

Return tiny deterministic cases for benchmark-harness validation.

Source code in src/vector_search_study/benchmarking.py 
397
398
399
def smoke_specs() -> tuple[WorkloadSpec, ...]:
    """Return tiny deterministic cases for benchmark-harness validation."""
    return tuple(WorkloadSpec(objective, 1_000, 8, 1, 10, profile="smoke") for objective in OBJECTIVES)

profile_specs 

profile_specs(profile: str) -> tuple[WorkloadSpec, ...]

Return the workload slice collected by a named benchmark target.

Source code in src/vector_search_study/benchmarking.py 
402
403
404
405
406
407
408
409
410
411
412
413
414
def profile_specs(profile: str) -> tuple[WorkloadSpec, ...]:
    """Return the workload slice collected by a named benchmark target."""
    profiles = {
        "smoke": smoke_specs,
        "small": small_specs,
        "core": standard_specs,
        "stress": stress_specs,
    }
    try:
        return profiles[profile]()
    except KeyError as exc:
        choices = ", ".join(sorted(profiles))
        raise ValueError(f"profile must be one of: {choices}") from exc

selected_metrics 

selected_metrics(workload: WorkloadSpec) -> tuple[str, ...]

Return latency views selected before discovery collection.

Source code in src/vector_search_study/benchmarking.py 
417
418
419
420
421
422
423
424
425
426
427
428
def selected_metrics(workload: WorkloadSpec) -> tuple[str, ...]:
    """Return latency views selected before discovery collection."""
    include_tail = workload.profile in {"smoke", "discovery-small"} or (
        workload.profile == "discovery-core"
        and workload.corpus_size == 10_000
        and workload.dimension == 128
        and workload.query_count == 32
        and workload.k == 10
    )
    if include_tail:
        return (*DISCOVERY_METRICS, TAIL_LATENCY_METRIC)
    return DISCOVERY_METRICS

dependency_is_installed 

dependency_is_installed(
    implementation: ImplementationSpec,
) -> bool

Return whether an implementation's optional import is available.

Source code in src/vector_search_study/benchmarking.py 
431
432
433
434
435
436
437
def dependency_is_installed(implementation: ImplementationSpec) -> bool:
    """Return whether an implementation's optional import is available."""
    dependency = implementation.optional_dependency
    if dependency is None:
        return True
    module = {"scikit-learn": "sklearn", "faiss-cpu": "faiss"}.get(dependency, dependency)
    return find_spec(module) is not None

discovery_cell_plans 

discovery_cell_plans(
    profile: str,
    *,
    memory_budget_bytes: int = DEFAULT_MEMORY_BUDGET_BYTES,
    availability: Callable[
        [ImplementationSpec], bool
    ] = dependency_is_installed,
) -> tuple[DiscoveryCellPlan, ...]

Return every included and excluded cell for one discovery profile.

Source code in src/vector_search_study/benchmarking.py 
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
def discovery_cell_plans(
    profile: str,
    *,
    memory_budget_bytes: int = DEFAULT_MEMORY_BUDGET_BYTES,
    availability: Callable[[ImplementationSpec], bool] = dependency_is_installed,
) -> tuple[DiscoveryCellPlan, ...]:
    """Return every included and excluded cell for one discovery profile."""
    cells: list[DiscoveryCellPlan] = []
    for workload in profile_specs(profile):
        metrics = selected_metrics(workload)
        for implementation in IMPLEMENTATIONS.values():
            cells.append(
                DiscoveryCellPlan(
                    workload=workload,
                    implementation=implementation,
                    decision=assess_feasibility(
                        workload,
                        implementation,
                        memory_budget_bytes=memory_budget_bytes,
                    ),
                    dependency_installed=availability(implementation),
                    metrics=metrics,
                )
            )
    return tuple(cells)

discovery_plan_metadata 

discovery_plan_metadata(
    profile: str,
    *,
    memory_budget_bytes: int = DEFAULT_MEMORY_BUDGET_BYTES,
    availability: Callable[
        [ImplementationSpec], bool
    ] = dependency_is_installed,
) -> dict[str, object]

Return a deterministic manifest of all discovery cell decisions.

Source code in src/vector_search_study/benchmarking.py 
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
def discovery_plan_metadata(
    profile: str,
    *,
    memory_budget_bytes: int = DEFAULT_MEMORY_BUDGET_BYTES,
    availability: Callable[[ImplementationSpec], bool] = dependency_is_installed,
) -> dict[str, object]:
    """Return a deterministic manifest of all discovery cell decisions."""
    workloads = profile_specs(profile)
    cells = discovery_cell_plans(
        profile,
        memory_budget_bytes=memory_budget_bytes,
        availability=availability,
    )
    return {
        "schema_version": _SCHEMA_VERSION,
        "study": "exact_top_k_vector_search",
        "profile": profile,
        "memory_budget_bytes": memory_budget_bytes,
        "workload_count": len(workloads),
        "included_cell_count": sum(cell.included for cell in cells),
        "excluded_cell_count": sum(not cell.included for cell in cells),
        "workloads": [workload.metadata() for workload in workloads],
        "cells": [cell.metadata() for cell in cells],
    }