Fix aggregate crashing on operations that report null metrics

An operation that produced no valid samples reports its metric fields as null.
The geonames workload does this for `optimize`, which comes back with
error_rate 1.0 and a fully null throughput. Aggregating any test run that
contains such an operation fails:

  [ERROR] Cannot aggregate. '<' not supported between instances of
          'NoneType' and 'NoneType'.

calculate_weighted_average reached min()/max() with None values, and
`value.get(metric_field, 0)` did not help because the key is present with a
null value rather than absent.

Null values are now left out of the min, the max, and the weighted mean
instead of being coerced to 0, and the result is None when no test run
contributed a value, so an unmeasured metric stays distinct from a measured
zero. calculate_rsd has the same problem on a second, independent code path
and returns "NA" when nothing was measured.

The weighted-mean arithmetic moves into a helper because the divisor now
depends on which runs contributed, so it can no longer be a single total
computed up front.

Signed-off-by: Serhiy Bzhezytskyy <me@serhiy-bzhezytskyy.com>

Author: serhiy-bzhezytskyy

osbenchmark/aggregator.py

@@ -213,7 +213,6 @@ def calculate_weighted_average(self, task_metrics: Dict[str, List[Any]], task_na
         # Get iterations for each test run
         iterations_per_run = [self.accumulated_iterations[test_id][task_name]
                                     for test_id in self.test_runs.keys()]
-        total_iterations = sum(iterations_per_run)
 
         for metric, values in task_metrics.items():
             if isinstance(values[0], dict):
@@ -222,23 +221,44 @@ def calculate_weighted_average(self, task_metrics: Dict[str, List[Any]], task_na
                     if metric_field == 'unit':
                         weighted_metrics[metric][metric_field] = values[0][metric_field]
                     elif metric_field == 'min':
-                        weighted_metrics[metric]['overall_min'] = min(value.get(metric_field, 0) for value in values)
+                        item_values = [value.get(metric_field) for value in values]
+                        weighted_metrics[metric]['overall_min'] = min(
+                            (value for value in item_values if value is not None), default=None)
                     elif metric_field == 'max':
-                        weighted_metrics[metric]['overall_max'] = max(value.get(metric_field, 0) for value in values)
+                        item_values = [value.get(metric_field) for value in values]
+                        weighted_metrics[metric]['overall_max'] = max(
+                            (value for value in item_values if value is not None), default=None)
                     else:
                         # for items like median or containing percentile values
-                        item_values = [value.get(metric_field, 0) for value in values]
-                        weighted_sum = sum(value * iterations for value, iterations in zip(item_values, iterations_per_run))
-                        weighted_metrics[metric][metric_field] = weighted_sum / total_iterations
+                        item_values = [value.get(metric_field) for value in values]
+                        weighted_metrics[metric][metric_field] = self.weighted_mean(item_values, iterations_per_run)
             else:
-                weighted_sum = sum(value * iterations for value, iterations in zip(values, iterations_per_run))
-                weighted_metrics[metric] = weighted_sum / total_iterations
+                weighted_metrics[metric] = self.weighted_mean(values, iterations_per_run)
 
         return weighted_metrics
 
+    @staticmethod
+    def weighted_mean(values: List[Any], iterations_per_run: List[int]) -> Any:
+        """
+        Weights each test run's value by that run's iteration count.
+        Operations that produced no valid samples report their metrics as None; those are left out
+        of both the sum and the divisor, and the result is None when no run contributed a value,
+        so that "not measured" stays distinct from a measured zero
+        """
+        contributions = [(value, iterations) for value, iterations in zip(values, iterations_per_run)
+                         if value is not None]
+        total_iterations = sum(iterations for _, iterations in contributions)
+        if not total_iterations:
+            return None
+        return sum(value * iterations for value, iterations in contributions) / total_iterations
+
     def calculate_rsd(self, values: List[Union[int, float]], metric_name: str) -> Union[float, str]:
         if not values:
             raise ValueError(f"Cannot calculate RSD for metric '{metric_name}': empty list of values")
+        # operations that produced no valid samples report None, which cannot contribute to a deviation
+        values = [value for value in values if value is not None]
+        if not values:
+            return "NA"  # no test run measured this metric
         if len(values) == 1:
             return "NA"  # RSD is not applicable for a single value
         mean = statistics.mean(values)

tests/aggregator_test.py

@@ -136,6 +136,47 @@ def test_update_config_object_reads_attributes_that_test_runs_actually_have(aggr
     aggregator.config.add.assert_any_call(config.Scope.applicationOverride, "builder",
                                           "cluster_config.params", {"heap_size": "6g"})
 
+def test_calculate_weighted_average_with_null_metric_fields(aggregator):
+    # An operation that produced no valid samples reports null metric fields, e.g. `optimize`
+    # in the geonames workload, which reports error_rate 1.0 and a fully null throughput.
+    task_metrics = {
+        "throughput": [
+            {"min": None, "mean": None, "median": None, "max": None, "unit": "ops/s"},
+            {"min": None, "mean": None, "median": None, "max": None, "unit": "ops/s"}
+        ]
+    }
+    aggregator.accumulated_iterations = {"test1": {"op1": 1}, "test2": {"op1": 1}}
+    aggregator.test_runs = {"test1": Mock(), "test2": Mock()}
+
+    result = aggregator.calculate_weighted_average(task_metrics, "op1")
+
+    # null is carried through rather than replaced by 0, which would read as a real
+    # measurement of zero throughput
+    assert result["throughput"]["overall_min"] is None
+    assert result["throughput"]["overall_max"] is None
+    assert result["throughput"]["mean"] is None
+    assert result["throughput"]["median"] is None
+    assert result["throughput"]["unit"] == "ops/s"
+
+def test_calculate_weighted_average_with_partially_null_metric_fields(aggregator):
+    # A metric that has samples in one test run but not another: the valid values are
+    # still aggregated, weighted only by the runs that contributed them.
+    task_metrics = {
+        "throughput": [
+            {"min": 10, "mean": 20, "median": 20, "max": 30, "unit": "ops/s"},
+            {"min": None, "mean": None, "median": None, "max": None, "unit": "ops/s"}
+        ]
+    }
+    aggregator.accumulated_iterations = {"test1": {"op1": 2}, "test2": {"op1": 3}}
+    aggregator.test_runs = {"test1": Mock(), "test2": Mock()}
+
+    result = aggregator.calculate_weighted_average(task_metrics, "op1")
+
+    assert result["throughput"]["overall_min"] == 10
+    assert result["throughput"]["overall_max"] == 30
+    assert result["throughput"]["mean"] == 20
+    assert result["throughput"]["unit"] == "ops/s"
+
 def test_calculate_rsd(aggregator):
     values = [1, 2, 3, 4, 5]
     rsd = aggregator.calculate_rsd(values, "test_metric")