From 24ae0d37f2a272e6a0a2fe6bb0ce6f60882466c7 Mon Sep 17 00:00:00 2001 From: MrlixiangWE <102979255+MrlixiangWE@users.noreply.github.com> Date: Fri, 14 Aug 2026 02:26:34 +0800 Subject: [PATCH 1/2] Make AnalyzedSeries change point computation lazy --- otava/series.py | 64 +++++++++++++++++++++++++++++++++---------------- 1 file changed, 44 insertions(+), 20 deletions(-) diff --git a/otava/series.py b/otava/series.py index cb10d7b5..6237a46d 100644 --- a/otava/series.py +++ b/otava/series.py @@ -121,29 +121,56 @@ def analyze(self, options: Optional[AnalysisOptions] = None) -> "AnalyzedSeries" class AnalyzedSeries: """ Time series data with computed change points. + + Change points are computed lazily, on first access. Constructing an + instance is cheap for callers that never read them. """ __series: Series options: AnalysisOptions - change_points: ChangePointsByMetric - change_points_by_time: ChangePointsByTime - change_points_timestamp: datetime def __init__( self, series: Series, options: AnalysisOptions, change_points: Dict[str, ChangePoint] = None ): self.__series = series self.options = options - # record when these change points were calculated - self.change_points_timestamp = datetime.now(timezone.utc) - self.change_points = None - if change_points is not None: - self.change_points = change_points - else: - cp, weak_cps = self.__compute_change_points(series, options) - self.change_points = cp - self.weak_change_points = weak_cps - self.change_points_by_time = self.__group_change_points_by_time(series, self.change_points) + self.__change_points = change_points + self.__weak_change_points = ChangePointsByMetric() if change_points is not None else None + self.__change_points_by_time = None + # records when the change points were calculated + self.__change_points_timestamp = ( + datetime.now(timezone.utc) if change_points is not None else None + ) + + def __ensure_change_points_computed(self): + if self.__change_points is None: + cp, weak_cps = self.__compute_change_points(self.__series, self.options) + self.__change_points = cp + self.__weak_change_points = weak_cps + self.__change_points_timestamp = datetime.now(timezone.utc) + + @property + def change_points(self) -> ChangePointsByMetric: + self.__ensure_change_points_computed() + return self.__change_points + + @property + def weak_change_points(self) -> ChangePointsByMetric: + self.__ensure_change_points_computed() + return self.__weak_change_points + + @property + def change_points_timestamp(self) -> datetime: + self.__ensure_change_points_computed() + return self.__change_points_timestamp + + @property + def change_points_by_time(self) -> ChangePointsByTime: + if self.__change_points_by_time is None: + self.__change_points_by_time = self.__group_change_points_by_time( + self.__series, self.change_points + ) + return self.__change_points_by_time @staticmethod def __compute_change_points( @@ -320,8 +347,8 @@ def append(self, time, new_data, attributes): # r has a subset of all metrics, so can't just set change_points to r for metric, cpglist in r.items(): self.change_points[metric] = cpglist - self.weak_change_points = w - self.change_points_by_time = self.change_points.by_time() + self.__weak_change_points = w + self.__change_points_by_time = self.change_points.by_time() return r, w def test_name(self) -> str: @@ -466,14 +493,11 @@ def change_points_from_json(change_points_json): ) analyzed_series = cls(new_series, new_options, new_change_points) - analyzed_series.weak_change_points = new_weak_change_points + analyzed_series.__weak_change_points = new_weak_change_points if "change_points_timestamp" in analyzed_json.keys(): - analyzed_series.change_points_timestamp = _datetime_adapter.validate_python( + analyzed_series.__change_points_timestamp = _datetime_adapter.validate_python( analyzed_json["change_points_timestamp"] ) - analyzed_series.change_points_by_time = AnalyzedSeries.__group_change_points_by_time( - analyzed_series.__series, analyzed_series.change_points - ) return analyzed_series From 91680702098ea1fa395aa55d5cce8666432f85da Mon Sep 17 00:00:00 2001 From: MrlixiangWE <102979255+MrlixiangWE@users.noreply.github.com> Date: Fri, 14 Aug 2026 02:28:51 +0800 Subject: [PATCH 2/2] Cover lazy change point computation in series tests --- otava/series.py | 23 +++-- tests/series_test.py | 204 +++++++++++++++++++++++++++++++++++++++---- 2 files changed, 201 insertions(+), 26 deletions(-) diff --git a/otava/series.py b/otava/series.py index 6237a46d..4ddb492c 100644 --- a/otava/series.py +++ b/otava/series.py @@ -114,16 +114,12 @@ def find_by_attribute(self, name: str, value: str) -> List[int]: def analyze(self, options: Optional[AnalysisOptions] = None) -> "AnalyzedSeries": if options is None: options = AnalysisOptions() - logging.info(f"Computing change points for test {self.test_name}...") return AnalyzedSeries(self, options) class AnalyzedSeries: """ Time series data with computed change points. - - Change points are computed lazily, on first access. Constructing an - instance is cheap for callers that never read them. """ __series: Series @@ -144,6 +140,7 @@ def __init__( def __ensure_change_points_computed(self): if self.__change_points is None: + logging.info(f"Computing change points for test {self.__series.test_name}...") cp, weak_cps = self.__compute_change_points(self.__series, self.options) self.__change_points = cp self.__weak_change_points = weak_cps @@ -254,8 +251,8 @@ def can_append(self, time, new_data, attributes): return self._validate_append(time, new_data, attributes) is None def _validate_append(self, time, new_data, attributes): - if not self.change_points: - return RuntimeError("You must use __compute_change_points() once first.") + # appending updates the cached results, so they must exist first + self.__ensure_change_points_computed() if not isinstance(time, list): return ValueError("time argument must be an array.") if not isinstance(new_data, dict): @@ -300,13 +297,19 @@ def append(self, time, new_data, attributes): for metric in self.__series.data.keys(): if metric not in new_data: - weak_change_points[metric] = self.weak_change_points.select_metrics(metric) + if metric in self.weak_change_points: + weak_change_points[metric] = self.weak_change_points.select_metrics(metric) continue new_data_len = len(new_data[metric]) + previous_weak_cp = ( + self.weak_change_points.get_change_points_for_metric(metric) + if metric in self.weak_change_points + else [] + ) old_weak_cp = [ cp - for cp in self.weak_change_points.get_change_points_for_metric(metric) + for cp in previous_weak_cp if cp.index < len(self.__series.data[metric]) - new_data_len - 1 ] change_points, weak_cps = compute_change_points( @@ -348,7 +351,9 @@ def append(self, time, new_data, attributes): for metric, cpglist in r.items(): self.change_points[metric] = cpglist self.__weak_change_points = w - self.__change_points_by_time = self.change_points.by_time() + # invalidate rather than rebuild: the property recomputes it on first read + self.__change_points_by_time = None + self.__change_points_timestamp = datetime.now(timezone.utc) return r, w def test_name(self) -> str: diff --git a/tests/series_test.py b/tests/series_test.py index 0497380e..cd549662 100644 --- a/tests/series_test.py +++ b/tests/series_test.py @@ -188,7 +188,8 @@ def test_change_point_detection_performance(): data={"series": series}, attributes={}, ) - test.analyze() + # access the results so the timing covers detection, not just construction + test.analyze().change_points_by_time end_time = time.process_time() assert (end_time - start_time) < 0.5 @@ -472,22 +473,6 @@ def test_validate(): data={"series1": series_1, "series2": series_2}, attributes={}, ) - test_fail = Series( - "test", - branch=None, - time=time, - metrics={"series1": Metric(1, 1.0), "series2": Metric(1, 1.0)}, - data={"series1": series_1, "series2": series_2}, - attributes={}, - ) - - analyzed_series_fail = test_fail.analyze() - analyzed_series_fail.change_points = None - err = analyzed_series_fail._validate_append( - time=[len(time)], new_data={"series1": [0.51]}, attributes={} - ) - assert isinstance(err, RuntimeError) - analyzed_series = test.analyze() analyzed_series.append( time=[len(time)], new_data={"series1": [0.5], "series2": [1.97]}, attributes={} @@ -599,3 +584,188 @@ def test_series_raw_initialization(): assert len(series.time) == 3 assert series.data["throughput"] == [10.0, 12.0, 11.5] + + +def test_change_points_computed_lazily_and_cached(monkeypatch): + from otava import series as series_module + + calls = {"count": 0} + real_compute = series_module.compute_change_points + + def counting_compute(*args, **kwargs): + calls["count"] += 1 + return real_compute(*args, **kwargs) + + monkeypatch.setattr(series_module, "compute_change_points", counting_compute) + + data = [1.0] * 10 + [5.0] * 10 + test = Series( + "lazy_test", + branch=None, + time=list(range(len(data))), + metrics={"m1": Metric(1, 1.0), "m2": Metric(1, 1.0)}, + data={"m1": data, "m2": data.copy()}, + attributes={}, + ) + + analyzed = test.analyze() + assert calls["count"] == 0 + + change_points = analyzed.change_points + assert calls["count"] == 2 # one computation per metric + assert [c.index for c in change_points.get_change_points_for_metric("m1")] == [10] + + assert analyzed.change_points is change_points + assert len(analyzed.change_points_by_time) == 1 + assert analyzed.weak_change_points is not None + assert analyzed.change_points_timestamp is not None + assert calls["count"] == 2 + + +def test_append_on_stable_series(monkeypatch): + from otava import series as series_module + + calls = {"count": 0} + real_compute = series_module.compute_change_points + + def counting_compute(*args, **kwargs): + calls["count"] += 1 + return real_compute(*args, **kwargs) + + monkeypatch.setattr(series_module, "compute_change_points", counting_compute) + + stable = [1.0] * 20 + test = Series( + "stable_test", + branch=None, + time=list(range(len(stable))), + metrics={"m1": Metric(1, 1.0)}, + data={"m1": stable}, + attributes={}, + ) + + analyzed = test.analyze() + assert calls["count"] == 0 + + # a stable series has no change points, which must not be mistaken for "not computed yet" + assert analyzed.can_append(time=[len(stable)], new_data={"m1": [1.0]}, attributes={}) + assert calls["count"] == 1 + assert len(list(analyzed.change_points)) == 0 + + analyzed.append(time=[len(stable)], new_data={"m1": [1.0]}, attributes={}) + assert len(list(analyzed.change_points)) == 0 + + +def test_append_invalidates_by_time_view(monkeypatch): + data = [1.0] * 10 + [5.0] * 10 + test = Series( + "invalidation_test", + branch=None, + time=list(range(len(data))), + metrics={"m1": Metric(1, 1.0)}, + data={"m1": data}, + attributes={}, + ) + + analyzed = test.analyze() + assert [cpg.time for cpg in analyzed.change_points_by_time] == [10] + + # a second shift, so the by-time view must change after the append + analyzed.append( + time=list(range(20, 32)), new_data={"m1": [50.0] * 12}, attributes={} + ) + + assert [cpg.time for cpg in analyzed.change_points_by_time] == [10, 20] + + +def test_append_does_not_rebuild_unread_by_time_view(monkeypatch): + from otava.change_point_divisive import base as base_module + + data = [1.0] * 10 + [5.0] * 10 + test = Series( + "no_rebuild_test", + branch=None, + time=list(range(len(data))), + metrics={"m1": Metric(1, 1.0)}, + data={"m1": data}, + attributes={}, + ) + + analyzed = test.analyze() + calls = {"count": 0} + real_by_time = base_module.ChangePointsByMetric.by_time + + def counting_by_time(self, *args, **kwargs): + calls["count"] += 1 + return real_by_time(self, *args, **kwargs) + + monkeypatch.setattr(base_module.ChangePointsByMetric, "by_time", counting_by_time) + + analyzed.append(time=[len(data)], new_data={"m1": [5.0]}, attributes={}) + assert calls["count"] == 0 + + _ = analyzed.change_points_by_time + assert calls["count"] == 1 + + +def test_append_refreshes_change_points_timestamp(): + data = [1.0] * 10 + [5.0] * 10 + test = Series( + "timestamp_test", + branch=None, + time=list(range(len(data))), + metrics={"m1": Metric(1, 1.0)}, + data={"m1": data}, + attributes={}, + ) + + analyzed = test.analyze() + before = analyzed.change_points_timestamp + + analyzed.append(time=[len(data)], new_data={"m1": [5.0]}, attributes={}) + + assert analyzed.change_points_timestamp > before + + +def test_append_computes_change_points_first(): + data = [1.0] * 10 + [5.0] * 10 + test = Series( + "append_lazy_test", + branch=None, + time=list(range(len(data))), + metrics={"m1": Metric(1, 1.0)}, + data={"m1": data}, + attributes={}, + ) + + analyzed = test.analyze() + analyzed.append(time=[len(data)], new_data={"m1": [5.0]}, attributes={}) + + assert [c.index for c in analyzed.change_points.get_change_points_for_metric("m1")] == [10] + assert len(analyzed.change_points_by_time) == 1 + + +def test_from_json_does_not_recompute(monkeypatch): + from otava import series as series_module + + data = [1.0] * 10 + [5.0] * 10 + test = Series( + "roundtrip_lazy_test", + branch=None, + time=list(range(len(data))), + metrics={"m1": Metric(1, 1.0)}, + data={"m1": data}, + attributes={}, + ) + analyzed = test.analyze() + payload = analyzed.to_json() + + def fail_compute(*args, **kwargs): + raise AssertionError("a deserialized series must not recompute change points") + + monkeypatch.setattr(series_module, "compute_change_points", fail_compute) + + restored = AnalyzedSeries.from_json(payload) + assert restored.change_points_timestamp == analyzed.change_points_timestamp + assert [c.index for c in restored.change_points.get_change_points_for_metric("m1")] == [10] + assert len(restored.change_points_by_time) == len(analyzed.change_points_by_time)