fix(pytorch): compute aggregate operator metrics over the full selected population, not the output limit

[morluto]

Severity

High analysis correctness — synchronization, compilation, allocation, and repeated-small-operation results change when the caller changes only the display limit, and omitted operators are silently excluded from fields named as totals

Validated against main at 0a60b288342cd96e2608302fa9fd2917e59d7e58.

Audit only; no implementation change is included.

Summary

PyTorchRecipes.pytorch() uses limit in the SQL query that selects operator aggregates:

ORDER BY inclusive duration DESC
LIMIT ?

It then computes all of these fields only from the returned top-N operators:

synchronization_time_ns
compilation_time_ns
allocation_bytes
repeated_small_operations
25th-percentile per-event threshold
coverage flags for device/synchronization/shapes/allocations/warmup

limit is documented and modeled as a bounded output control. It should not redefine the analysis population.

The same pinned trace can therefore produce materially different aggregate conclusions solely because a caller asks for fewer rows. Omitted synchronization or compilation operators contribute zero rather than an explicit unknown/lower bound.

warmup_time_ns is computed from the unbounded metadata query, while the other aggregates use the bounded operator query, so one result object mixes full-population and top-N-population totals.

Direct code evidence

The recipe first counts the complete population:

SELECT count(DISTINCT fm.frame_id) ...
total = int(count_row[0])

It then fetches only top-N frames:

rows = snapshot.execute(
    "SELECT ... "
    "FROM frame_measurements ... "
    "GROUP BY fm.frame_id, f.function, f.module "
    "ORDER BY sum(coalesce(fm.inclusive_value, 0)) DESC, fm.frame_id "
    "LIMIT ?",
    (*parameters, bounded),
).fetchall()

flameox/src/flameox/analysis/recipe_pytorch.py

Lines 20 to 80 in 0a60b28

	limit: int | None = None,
	corpus_commit_id: str | None = None,
	) -> PyTorchAnalysisResult:
	corpus_commit_id = self._pinned_commit_id(corpus_commit_id)
	bounded = self._limit(limit)
	with self._open_snapshot(corpus_commit_id) as snapshot:
	scope = resolve_evidence_scope(snapshot, input_id)
	self._require_pytorch_source(snapshot, scope.run_ids, scope.artifact_ids)
	where, parameters = scope.predicate(
	run_column="fm.run_id",
	artifact_column="fm.artifact_id",
	)
	count_row = snapshot.execute(
	"SELECT count(DISTINCT fm.frame_id) FROM frame_measurements fm WHERE " + where,
	parameters,
	).fetchone()
	assert count_row is not None
	total = int(count_row[0])
	if total == 0:
	if scope.run_ids:
	run_ids = scope.run_ids
	elif scope.artifact_ids:
	run_rows = snapshot.execute(
	"SELECT DISTINCT run_id FROM artifact_registrations WHERE artifact_id IN ("
	+ ", ".join("?" for _ in scope.artifact_ids)
	+ ") ORDER BY run_id",
	scope.artifact_ids,
	).fetchall()
	run_ids = tuple(str(row[0]) for row in run_rows)
	else:
	run_ids = ()
	details: dict[str, object] = {"next_tool": "extract_perfetto"}
	if run_ids:
	details["run_id"] = run_ids[0]
	raise DomainError(
	ErrorCode.CAPABILITY_UNAVAILABLE,
	"PyTorch operator analysis requires Perfetto extraction for this "
	"imported trace.",
	details=details,
	remediation=(
	"Call extract_perfetto with the reported run_id, then retry "
	"analyze_pytorch.",
	"If Trace Processor is unavailable, call start_capability_setup with "
	"adapter='perfetto'.",
	),
	)
	rows = snapshot.execute(
	"SELECT fm.frame_id, coalesce(f.function, '<unnamed>'), f.module, "
	"sum(coalesce(fm.self_value, 0)), "
	"sum(coalesce(fm.inclusive_value, 0)), "
	"sum(coalesce(fm.sample_count, 0)) "
	"FROM frame_measurements fm JOIN frames f "
	"ON f.frame_id = fm.frame_id WHERE "
	+ where
	+ " GROUP BY fm.frame_id, f.function, f.module "
	"ORDER BY sum(coalesce(fm.inclusive_value, 0)) DESC, fm.frame_id "
	"LIMIT ?",
	(*parameters, bounded),
	).fetchall()
	observation_where, observation_parameters = scope.predicate(
	run_column="run_id",

Every operator-level aggregate below is computed from operators, which is constructed only from rows:

synchronization_time_ns = sum(
    item.inclusive_ns for item in operators if item.synchronization
)

compilation_time_ns = sum(
    item.inclusive_ns for item in operators
    if any(token in item.operator.lower() for token in (...))
)

allocation_bytes = sum(item.allocation_bytes or 0 for item in operators) or None

per_event_times = [
    operator.inclusive_ns / max(operator.event_count, 1)
    for operator in operators
]
...
repeated_small = tuple(... for item in operators ...)

flameox/src/flameox/analysis/recipe_pytorch.py

Lines 150 to 235 in 0a60b28

	True
	if phases and warmup_phases == phases
	else False
	if phases and not warmup_phases
	else None
	)
	operators_list.append(
	OperatorSummary(
	frame_id=frame_id,
	operator=operator,
	category=category,
	self_cpu_ns=None if is_device else int(row[3]),
	total_cpu_ns=None if is_device else inclusive,
	device_ns=inclusive if is_device else None,
	inclusive_ns=inclusive,
	event_count=int(row[5]),
	input_shapes=shapes,
	allocation_bytes=sum(allocations) if allocations else None,
	synchronization=synchronization,
	warmup=warmup,
	)
	)
	operators = tuple(operators_list)
	synchronization_time_ns = sum(
	item.inclusive_ns for item in operators if item.synchronization
	)
	compilation_time_ns = sum(
	item.inclusive_ns
	for item in operators
	if any(
	token in item.operator.lower()
	for token in ("compile", "dynamo", "inductor", "graph_executor")
	)
	)
	warmup_time_ns = sum(
	duration
	for metadata in metadata_by_operator.values()
	for value in metadata
	if isinstance(duration := value.get("duration_ns"), int)
	and "warm" in str(value.get("phase", "")).lower()
	)
	allocation_bytes = sum(item.allocation_bytes or 0 for item in operators) or None
	per_event_times = [
	operator.inclusive_ns / max(operator.event_count, 1) for operator in operators
	]
	typical_event_ns = max(
	1.0,
	float(np.percentile(per_event_times, 25)) if per_event_times else 1.0,
	)
	repeated_small = tuple(
	sorted(
	(
	item
	for item in operators
	if item.event_count >= 3
	and item.inclusive_ns / item.event_count <= typical_event_ns
	),
	key=lambda item: (-item.event_count, item.inclusive_ns, item.frame_id),
	)[:bounded]
	)
	limitations = [
	"Operator categories and durations come from the exported torch.profiler trace.",
	"Nested operator durations can overlap; self time subtracts direct nested slices.",
	]
	if not device_time_present:
	limitations.append("The trace contains no recognized accelerator kernel categories.")
	shapes_present = any(item.input_shapes for item in operators)
	allocations_present = any(item.allocation_bytes is not None for item in operators)
	warmup_present = any(item.warmup is not None for item in operators)
	if not shapes_present:
	limitations.append("Input shapes were not present in normalized trace evidence.")
	if not allocations_present:
	limitations.append(
	"Per-operator allocation bytes were not present in normalized trace evidence."
	)
	if not warmup_present:
	limitations.append("Warm-up separation requires profiler phase annotations.")
	return PyTorchAnalysisResult(
	corpus_commit_id=snapshot.commit.commit_id,
	input_id=input_id,
	operators=operators,
	total=total,
	returned=len(operators),
	truncated=total > len(operators),
	coverage={
	"self_cpu_time": True,

By contrast, metadata_rows is not limited, and warmup_time_ns sums all matching metadata observations:

warmup_time_ns = sum(
    duration
    for metadata in metadata_by_operator.values()
    for value in metadata
    if ...
)

Thus the object has no single population definition.

The result exposes:

truncated = total > len(operators)

but the aggregate scalar fields are not typed as partial/lower bounds and no limitation states that omitted operators were excluded from them.

Deterministic aggregate counterexample

Publish one valid torch-profiler trace projection with three operator frames:

operator	category	inclusive ns	event count
aten::matmul	cpu	1,000	1
cudaDeviceSynchronize	cpu	900	1
torch.compile	cpu	800	1

Call with limit=1

SQL returns only aten::matmul.

Current result:

total = 3
returned = 1
truncated = true
synchronization_time_ns = 0
compilation_time_ns = 0

Call with limit=3

The exact same snapshot returns all frames:

total = 3
returned = 3
truncated = false
synchronization_time_ns = 900
compilation_time_ns = 800

The profiler evidence did not change. Only the presentation bound changed.

A structured consumer reading the scalar fields cannot distinguish:

there was no synchronization/compilation

from:

those operators were outside the requested top-N list

Deterministic allocation counterexample

Let the highest-duration operator have no allocation metadata and the second operator allocate 1 GiB.

At limit=1:

allocation_bytes = None
coverage.memory_allocations = false

At limit=2:

allocation_bytes = 1073741824
coverage.memory_allocations = true

Coverage of the artifact should not depend on how many operator rows the caller wants displayed.

Repeated-small-operation counterexample

The recipe is intended to identify high-frequency low-per-event operators. But SQL preselects by total inclusive duration, and only then computes the low-per-event percentile.

Suppose there are 101 operators:

• 100 large one-off operators with total durations 1,000–901 ns;
• one operator tiny_dispatch with 100 events × 8 ns = 800 ns total.

With limit=100, tiny_dispatch is omitted before the repeated-small analysis runs. The result reports no repeated small operation even though that operator is the clearest example in the trace.

Increasing limit to 101 changes the analysis population, percentile threshold, and finding.

A bounded output limit has become a biased prefilter against the phenomenon being detected.

Warmup-population inconsistency

warmup_time_ns uses every pytorch.operator observation, including operators omitted from operators, while:

• operators contains top-N frame aggregates;
• warmup flags are inferred only for those top-N operators;
• allocation_bytes, synchronization, and compilation use top-N only.

The result can therefore show:

warmup_time_ns > 0
coverage.warmup_phases = false

when all warmup-tagged operators lie outside the top-N frame list.

This is an internal semantic contradiction, not merely undercounting.

Additional population problems exposed by the same design

• device_time_present can be false when accelerator kernels exist outside top-N.
• synchronization_present can be false while omitted synchronization time is substantial.
• shapes_present/allocations_present/warmup_present describe returned rows, not trace coverage.
• typical_event_ns is the 25th percentile of the top-duration subset, not the operator population.
• repeated_small_operations is capped by the same bounded value after already being population-truncated.

Violated invariant

A query/output limit may bound materialized detail, but it must not alter the estimand or completeness claims.

For a pinned trace and analysis profile:

aggregate_scalar_result(trace, profile)

must be invariant under changes to:

operator_detail_limit

unless the field is explicitly typed as a top-N partial statistic and identifies that population.

Proposed direction

1. Compute full-population aggregates in SQL

Use separate bounded queries:

• one query for top-N operator rows;
• full-population aggregate queries for synchronization, compilation, allocation, device/shape/warmup coverage;
• a dedicated bounded candidate query for repeated-small operations based on its own predicate/order.

DuckDB can compute these without materializing every operator in Python.

2. Define repeated-small semantics independently

Select candidates by declared criteria such as:

minimum event count
per-event duration threshold/quantile over the complete eligible population
category/phase filters

Then bound returned candidates. Do not prefilter by total duration.

3. Type partial values honestly

When full aggregation is unavailable or bounded, return:

lower_bound
population_count
coverage/truncation reason

rather than an ordinary total.

4. Keep artifact/session populations explicit

Compose with #275 for multi-artifact traces. Full population still means one qualified selected session/profile, not every frame row in a run indiscriminately.

Acceptance criteria

• Synchronization, compilation, allocation, and coverage fields are invariant under operator display limit.
• The three-operator counterexample returns 900/800 at both limit=1 and limit=3.
• Omitted allocation metadata cannot flip coverage merely because of top-N selection.
• Repeated-small candidate discovery uses the complete eligible population or an explicit independent bound.
• A low-total/high-frequency operator outside top inclusive-duration N is still detectable.
• Warmup totals and warmup coverage use the same declared population.
• Every aggregate reports its population and completeness.
• Detail rows remain bounded without loading all operators into Python.
• Tests vary limit over the same snapshot and assert invariant aggregate fields.

Relationship to existing issues

• fix(comparisons): reject ambiguous measurement dimensions instead of pooling them #249/fix(scaling): bind one measurement series and unit before aggregation or model fitting #256/fix(memory): preserve series identity and chronological phase order before computing growth #277 require exact measurement-series/population identity in other analyses.
• fix(accelerator): scope trace timing and correlation to one artifact/session #275 defines trace artifact/session population boundaries.
• fix(evidence): reserve common provenance columns and remove duplicate extractor-version field #207 governs provenance needed to express aggregate populations.
• [Architecture] Add an explicit migration contract for every persisted schema #157 owns result-schema migration if aggregate fields become typed partial/full variants.