Speed up long-form object measurement spreadsheet folding

benchmark/results/perf_long_form_export_20260929/README.md

@@ -0,0 +1,38 @@
+# Faster long-form spreadsheet folding (2026-09-29)
+
+This report supports [issue #194](https://github.com/OpenHCSDev/openhcs/issues/194) and [PR #195](https://github.com/OpenHCSDev/openhcs/pull/195), stacked on [plate-input transport PR #193](https://github.com/OpenHCSDev/openhcs/pull/193). The only code change is inside `WideMeasurementRowAccumulator`: when a long-form columnar table declares fixed object scope, one image identity, one object ID, and no additional wide feature columns, it folds the cells without repeatedly deriving ownership and subject maps. All other row shapes use the existing loop. Duplicate-value conflict checks, feature projection, output ordering, and the accumulator's nominal authority remain in place.
+
+The saved contract-selected ImagingFlow eight-well `RuntimeArtifactBatch` renders a 92,194,901-byte spreadsheet. Its replay took about **25.1 → 20.2 seconds**, with identical SHA-256 `87b68418e19a69f20b57cf7c9bda4d1ed4f4b73d4b22507d84c61accc4a9bbca`. The original accumulator spent about 8.7 seconds folding 3.1 million long-form `BF_cells_on_grid` and `SSC` cells in that batch; the candidate spent about 3.8 seconds. The full benchmark confirms that this local gain transfers to the plate execution interval.
+
+An unprofiled replay of the candidate on the same saved batch took **20.23 seconds**: measurement-table assembly took **11.45 seconds**, relationship rows **1.02 seconds**, combined-subject rows **1.20 seconds**, and CSV rendering **6.20 seconds**. These phases account for nearly the whole render time. A dense-row prototype produced identical CSV bytes but took **3.20 seconds** to build its matrix and **3.38 seconds** to write it, slower than the existing **6.28-second** CSV writer in that run. The remaining export cost spans consolidation and serialization; a writer swap alone cannot eliminate it. The one-well runtime profile attributed only **0.064 seconds** to constructing radial-distribution rows, so changing that producer's row carrier alone has little direct single-worker payoff.
+
+All runs below use the ordinary ZMQ route, a fresh client-owned server per observation, `fork` workers, and one native thread per worker. Each checkout was warmed once before its timed comparison. Compilation and server startup contribute to total time; execution includes workers, result transfer, and plate export. Values are seconds.
+
+| Checkout and workload | Control execution | Candidate execution | Control total | Candidate total |
+| --- | ---: | ---: | ---: | ---: |
+| Integrated, ImagingFlow 1 well / 1 worker | 20.107 | **18.915** | 25.945 | **24.847** |
+| Integrated, ImagingFlow 16 wells / 4 workers | 126.078 | **115.808** | 139.305 | **128.913** |
+| Exact stacked PR, ImagingFlow 16 wells / 4 workers | 171.049 | **160.498** | 184.284 | **173.733** |
+| Integrated, official30 one-well total | 80.329 | **79.350** | 214.708 | **213.524** |
+
+The integrated control includes pending changes from PRs #157, #170, #175, #191, and #193; the candidate adds only this accumulator change. The exact stacked PR branches from #193 and lacks the separate #170/#175 changes, so it has a different absolute runtime and CSV from the integrated checkout. Its parent and candidate were compared only against each other. Both 16-well pairs saved about **10.5 seconds** of execution and total time. All **30/30** official one-well cases succeeded and **110/110** generated result files matched by relative path and bytes. The exact stacked 16-well CSV also matched its parent byte for byte. These checks establish output parity within each branch pair; they do not claim parity between branches with different pending code.
+
+The candidate 16-well total exceeds server execution by **13.106 seconds**: **10.065 seconds** of compilation and **3.041 seconds** of fresh-server/client overhead. Summed across 30 independently launched one-well cases, the **134.174-second** gap is **70.196 seconds** of compilation and **63.978 seconds** of other per-case work. The benchmark total timer begins after source workspace preparation and pipeline import.
+
+The separately measured native CellProfiler 4.2.8.1 single-sample `EXECUTE_NATIVE_CP` phase is **61.602 seconds** for ImagingFlow and includes launching a fresh native subprocess. Multiplying by 16 gives a **985.625-second cold-launch projection**, not a measured native 16-well plate or a matched warmed-run speedup. The independent warm native experiment in [PR #197](https://github.com/OpenHCSDev/openhcs/pull/197) measured **59.680 seconds** at one well and **388.431 seconds** at 16 wells with four concurrent native jobs. Against this PR's matching OpenHCS server-execution observations, the measured warm ratios are **3.16×** and **3.35×**, respectively. The figure below retains the earlier projection explicitly as historical context; [PR #197's report](https://github.com/OpenHCSDev/openhcs/blob/codex/native-synthetic-scaling/benchmark/results/perf_native_synthetic_scaling_20260929/README.md) contains the measured-native figure and timing evidence.
+
+![ImagingFlow execution scaling against native CP](imagingflow_scaling_vs_native.png)
+
+`observations.csv` and `official30_totals.csv` contain the recorded phase times and raw-run directory names; `plot.py` regenerates the figure. The native measurement comes from `benchmark/results/perf_lazy_catalog_20260929/native_cp_summary.csv`. The source batch was saved at `/home/ts/code/projects/openhcs-benchmark-runs/perf-spreadsheet-batch-imaging-8w-20260929.pkl`, and its original output is in `perf-spreadsheet-batch-capture-20260929`. Raw benchmark runs are under `/home/ts/code/projects/openhcs-benchmark-runs` with the directory names in the CSV files.
+
+Seventy-two focused measurement, spreadsheet, and analyst-export unit tests passed on the exact stacked PR branch. Black and the changed-line Ruff checks passed. Reproduce a 16-well observation with:
+
+```bash
+OPENHCS_CPU_ONLY=true \
+OPENHCS_REFERENCE_EXPORT_PIPELINES_ROOT=/absolute/path/to/benchmark/reference_exports/official30_value_completion_20260914 \
+.venv/bin/python scripts/benchmark_cppipe_well_throughput.py \
+  --manifest benchmark/manifests/official30_portable_axis1.json \
+  --output-dir /path/to/output \
+  --native-summary-csv benchmark/results/perf_lazy_catalog_20260929/native_cp_summary.csv \
+  --mode 16w_4c --case ExampleImagingFlowCytometryObjectsInGrid
+```

benchmark/results/perf_long_form_export_20260929/observations.csv

@@ -0,0 +1,7 @@
+pair,variant,case_name,well_count,worker_count,compile_seconds,execute_seconds,total_seconds,native_single_sample_execution_seconds,projected_native_execution_seconds,status,source_run
+integrated_1w,control,ExampleImagingFlowCytometryObjectsInGrid,1,1,3.2013087272644043,20.1071674823761,25.94524993514642,61.60158259188756,61.60158259188756,success,perf-long-form-control-imaging-1w-20260929
+integrated_1w,candidate,ExampleImagingFlowCytometryObjectsInGrid,1,1,3.307213544845581,18.91513967514038,24.847436835058033,61.60158259188756,61.60158259188756,success,perf-long-form-candidate-imaging-1w-20260929
+integrated_16w,control,ExampleImagingFlowCytometryObjectsInGrid,16,4,10.152604818344116,126.07837319374084,139.30536010675132,61.60158259188756,985.625321470201,success,perf-plate-transport-imaging-candidate-20260929
+integrated_16w,candidate,ExampleImagingFlowCytometryObjectsInGrid,16,4,10.064554452896118,115.80782866477966,128.91339134192094,61.60158259188756,985.625321470201,success,perf-long-form-candidate-imaging-16w-repeat-20260929
+stacked_pr_16w,control,ExampleImagingFlowCytometryObjectsInGrid,16,4,10.149492979049683,171.0489354133606,184.28444628976285,61.60158259188756,985.625321470201,success,perf-long-form-stacked-control-imaging-16w-repeat-20260929
+stacked_pr_16w,candidate,ExampleImagingFlowCytometryObjectsInGrid,16,4,10.166336059570312,160.49793696403503,173.7326781200245,61.60158259188756,985.625321470201,success,perf-long-form-stacked-imaging-16w-repeat-20260929

benchmark/results/perf_long_form_export_20260929/official30_totals.csv

@@ -0,0 +1,3 @@
+variant,cases,successful_cases,compile_seconds,execute_seconds,total_seconds,source_run
+control,30,30,70.31004929542542,80.32941341400146,214.70831610122696,perf-plate-transport-full-candidate-20260929
+candidate,30,30,70.19554257392883,79.35025358200073,213.5243737725541,perf-long-form-candidate-full30-repeat-20260929

benchmark/results/perf_long_form_export_20260929/plot.py

@@ -0,0 +1,117 @@
+"""Render native-relative ImagingFlow execution figures from recorded observations."""
+
+from __future__ import annotations
+
+import csv
+from pathlib import Path
+
+import matplotlib.pyplot as plt
+
+ROOT = Path(__file__).resolve().parent
+
+
+def main() -> None:
+    with (ROOT / "observations.csv").open(newline="") as stream:
+        rows = list(csv.DictReader(stream))
+    by_pair = {(row["pair"], row["variant"]): row for row in rows}
+
+    figure, (scaling, stacked) = plt.subplots(1, 2, figsize=(11, 4.5))
+    wells = (1, 16)
+    for variant, color in (("control", "#677687"), ("candidate", "#138579")):
+        values = [
+            float(by_pair[(f"integrated_{count}w", variant)]["execute_seconds"])
+            for count in wells
+        ]
+        scaling.plot(
+            wells,
+            values,
+            marker="o",
+            linewidth=2.2,
+            color=color,
+            label=f"OpenHCS {variant}",
+        )
+    native = [
+        float(
+            by_pair[(f"integrated_{count}w", "candidate")][
+                "projected_native_execution_seconds"
+            ]
+        )
+        for count in wells
+    ]
+    scaling.plot(
+        wells,
+        native,
+        marker="o",
+        linestyle="--",
+        linewidth=1.7,
+        color="#b46239",
+        label="Native CP (16 × cold-launch reference)",
+    )
+    scaling.set(
+        title="Integrated checkout: one vs 16 wells",
+        xlabel="Wells",
+        ylabel="Execution seconds (log scale)",
+        xticks=wells,
+        ylim=(12, 1400),
+    )
+    scaling.set_yscale("log")
+    scaling.grid(axis="y", alpha=0.25)
+    scaling.legend(loc="upper left", fontsize=8)
+    for count, native_seconds in zip(wells, native, strict=True):
+        candidate_seconds = float(
+            by_pair[(f"integrated_{count}w", "candidate")]["execute_seconds"]
+        )
+        scaling.annotate(
+            f"{native_seconds / candidate_seconds:.2f}×",
+            (count, candidate_seconds),
+            xytext=(0, 11 if count == 1 else -19),
+            textcoords="offset points",
+            ha="center",
+            fontsize=9,
+            color="#116e64",
+        )
+
+    exact = [
+        float(by_pair[("stacked_pr_16w", variant)]["execute_seconds"])
+        for variant in ("control", "candidate")
+    ]
+    bars = stacked.bar(
+        ("Control", "Candidate"),
+        exact,
+        color=("#677687", "#138579"),
+        width=0.55,
+    )
+    stacked.bar_label(bars, labels=[f"{value:.1f}s" for value in exact], padding=4)
+    stacked.set(
+        title="Exact stacked PR: 16 wells",
+        ylabel="Execution seconds",
+        ylim=(0, max(exact) * 1.18),
+    )
+    stacked.grid(axis="y", alpha=0.25)
+    stacked.set_axisbelow(True)
+    stacked.text(
+        0.5,
+        0.91,
+        f"{exact[0] - exact[1]:.1f}s saved",
+        transform=stacked.transAxes,
+        ha="center",
+        color="#116e64",
+        fontsize=10,
+    )
+
+    figure.suptitle("ImagingFlow execution: long-form spreadsheet folding")
+    figure.text(
+        0.5,
+        0.015,
+        "Native CP: one fresh subprocess measured; 16 wells = 16 × that cold-launch "
+        "measurement, not a native plate run.",
+        ha="center",
+        fontsize=8,
+        color="#555555",
+    )
+    figure.tight_layout(rect=(0, 0.04, 1, 0.95))
+    figure.savefig(ROOT / "imagingflow_scaling_vs_native.png", dpi=170)
+
+
+if __name__ == "__main__":
+    main()

openhcs/core/measurement_row_materialization.py

@@ -826,6 +826,25 @@ def add(
                 raise ValueError("Long-form measurement columns have no value column.")
             feature_values = feature_field_columns[0]
             measurement_values = value_field_columns[0]
+            if (
+                object_name_values is None
+                and default_scope is MeasurementScope.OBJECT
+                and len(identity_columns) == 1
+                and len(object_id_columns) == 1
+                and not feature_columns
+            ):
+                self._add_object_scoped_long_form(
+                    row_count=row_count,
+                    subject=default_subject,
+                    identity_column=identity_columns[0],
+                    object_id_values=object_id_columns[0],
+                    feature_values=feature_values,
+                    measurement_values=measurement_values,
+                    qualifier_columns=qualifier_columns,
+                    project_feature_name=project_feature_name,
+                    missing_cell=missing_cell,
+                )
+                return
             subject_cache: dict[object, str] = {}
             source_cache: dict[object, str | None] = {}
             projected_feature_cache: dict[
@@ -1058,6 +1077,76 @@ def add(
                     missing_cell,
                 )
 
+    def _add_object_scoped_long_form(
+        self,
+        *,
+        row_count: int,
+        subject: str,
+        identity_column: tuple[str, Sequence[object]],
+        object_id_values: Sequence[object],
+        feature_values: Sequence[object],
+        measurement_values: Sequence[object],
+        qualifier_columns: tuple[tuple[str, Sequence[object]], ...],
+        project_feature_name: MeasurementFeatureNameProjection,
+        missing_cell: object,
+    ) -> None:
+        """Fold object-owned long rows without repeating fixed ownership work."""
+        subject_rows = self._rows.setdefault(subject, {})
+        subject_order = self._order.setdefault(subject, [])
+        if subject not in self._object_subjects:
+            self._object_subjects.append(subject)
+        identity_field, image_values = identity_column
+        absent_identity = self._absent_identity
+        projection_cache: dict[tuple[str, tuple[tuple[str, object], ...]], str] = {}
+        for row_index in range(row_count):
+            image_value = image_values[row_index]
+            object_label = measurement_object_label_value(object_id_values[row_index])
+            identity = (
+                image_value,
+                object_label if object_label is not None else absent_identity,
+            )
+            if image_value is absent_identity and object_label is None:
+                identity = (self._passthrough_identity, self._passthrough_index)
+                self._passthrough_index += 1
+            target = subject_rows.get(identity)
+            if target is None:
+                target = {identity_field: image_value}
+                if object_label is not None:
+                    target[MeasurementRowAxisField.OBJECT_LABEL.value] = object_label
+                subject_rows[identity] = target
+                subject_order.append(identity)
+            feature_value = feature_values[row_index]
+            if is_structural_missing_measurement_cell(feature_value):
+                continue
+            feature_name = (
+                feature_value if isinstance(feature_value, str) else str(feature_value)
+            )
+            if not feature_name:
+                raise ValueError("Long-form measurement row has an empty feature name.")
+            qualifier_values = tuple(
+                (field_name, value)
+                for field_name, values in qualifier_columns
+                for value in (values[row_index],)
+                if not is_structural_missing_measurement_cell(value)
+            )
+            projection_key = (feature_name, qualifier_values)
+            projected_feature = projection_cache.get(projection_key)
+            if projected_feature is None:
+                projected_feature = project_feature_name(*projection_key)
+                projection_cache[projection_key] = projected_feature
+            measurement_value = measurement_values[row_index]
+            if is_structural_missing_measurement_cell(measurement_value):
+                raise ValueError(
+                    f"Long-form measurement feature {feature_name!r} has no value."
+                )
+            self._assign(
+                target,
+                identity,
+                projected_feature,
+                measurement_value,
+                missing_cell,
+            )
+
     def row_mappings_by_subject(
         self,
     ) -> dict[str, tuple[Mapping[str, object], ...]]:

tests/unit/test_cellprofiler_measurement_rows.py

@@ -753,6 +753,34 @@ def test_long_form_projection_omits_structurally_missing_qualifiers() -> None:
     )
 
 
+def test_object_scoped_long_form_projection_rejects_conflicting_values() -> None:
+    rows = _ColumnOnlyMeasurementRows(
+        {
+            "slice_index": (1, 1),
+            "object_label": (7, 7),
+            "feature_name": ("intensity", "intensity"),
+            "result_value": (4.0, 5.0),
+        },
+        fields=(
+            FieldSpec("slice_index", int),
+            FieldSpec("object_label", int),
+            FieldSpec("feature_name", str),
+            FieldSpec("result_value", float),
+        ),
+    )
+    accumulator = WideMeasurementRowAccumulator(
+        DEFAULT_RUNTIME_MEASUREMENT_ROW_IDENTITY_CONTRACT
+    )
+
+    with pytest.raises(ValueError, match="Conflicting sparse measurement values"):
+        accumulator.add(
+            rows,
+            lambda feature, _qualifiers: feature,
+            default_subject="Cells",
+            default_scope=MeasurementScope.OBJECT,
+        )
+
+
 def test_wide_measurement_projection_uses_row_owned_scope_for_artifact_table() -> None:
     accumulator = WideMeasurementRowAccumulator(
         DEFAULT_RUNTIME_MEASUREMENT_ROW_IDENTITY_CONTRACT