Module `materialize.scalability.result_analyzers`

Expand source code Browse git

# Copyright Materialize, Inc. and contributors. All rights reserved.
#
# Use of this software is governed by the Business Source License
# included in the LICENSE file at the root of this repository.
#
# As of the Change Date specified in that file, in accordance with
# the Business Source License, use of this software will be governed
# by the Apache License, Version 2.0.


from materialize.scalability.comparison_outcome import ComparisonOutcome
from materialize.scalability.endpoint import Endpoint
from materialize.scalability.result_analyzer import (
    ResultAnalyzer,
)
from materialize.scalability.scalability_change import (
    Regression,
    ScalabilityImprovement,
)
from materialize.scalability.workload_result import WorkloadResult


class DefaultResultAnalyzer(ResultAnalyzer):
    def __init__(self, max_deviation_as_percent_decimal: float):
        self.max_deviation_as_percent_decimal = max_deviation_as_percent_decimal

    def perform_comparison_in_workload(
        self,
        workload_name: str,
        baseline_endpoint: Endpoint,
        other_endpoint: Endpoint,
        regression_baseline_result: WorkloadResult,
        other_result: WorkloadResult,
    ) -> ComparisonOutcome:
        # tps = transactions per seconds (higher is better)

        merged_data = regression_baseline_result.df_totals.merge(other_result.df_totals)
        tps_per_endpoint_data = merged_data.to_enriched_result_frame(
            baseline_endpoint.try_load_version(), other_endpoint.try_load_version()
        )
        entries_worse_than_threshold = tps_per_endpoint_data.to_filtered_with_threshold(
            self.max_deviation_as_percent_decimal,
            match_results_better_than_baseline=False,
        )
        entries_better_than_threshold = (
            tps_per_endpoint_data.to_filtered_with_threshold(
                self.max_deviation_as_percent_decimal,
                match_results_better_than_baseline=True,
            )
        )

        comparison_outcome = ComparisonOutcome()
        regressions = entries_worse_than_threshold.to_scalability_change(
            Regression,
            workload_name,
            other_endpoint,
        )
        improvements = entries_better_than_threshold.to_scalability_change(
            ScalabilityImprovement,
            workload_name,
            other_endpoint,
        )
        comparison_outcome.append_regressions(
            regressions,
            improvements,
            entries_worse_than_threshold,
            entries_better_than_threshold,
        )
        return comparison_outcome

Classes

class DefaultResultAnalyzer (max_deviation_as_percent_decimal: float)

Expand source code Browse git

class DefaultResultAnalyzer(ResultAnalyzer):
    def __init__(self, max_deviation_as_percent_decimal: float):
        self.max_deviation_as_percent_decimal = max_deviation_as_percent_decimal

    def perform_comparison_in_workload(
        self,
        workload_name: str,
        baseline_endpoint: Endpoint,
        other_endpoint: Endpoint,
        regression_baseline_result: WorkloadResult,
        other_result: WorkloadResult,
    ) -> ComparisonOutcome:
        # tps = transactions per seconds (higher is better)

        merged_data = regression_baseline_result.df_totals.merge(other_result.df_totals)
        tps_per_endpoint_data = merged_data.to_enriched_result_frame(
            baseline_endpoint.try_load_version(), other_endpoint.try_load_version()
        )
        entries_worse_than_threshold = tps_per_endpoint_data.to_filtered_with_threshold(
            self.max_deviation_as_percent_decimal,
            match_results_better_than_baseline=False,
        )
        entries_better_than_threshold = (
            tps_per_endpoint_data.to_filtered_with_threshold(
                self.max_deviation_as_percent_decimal,
                match_results_better_than_baseline=True,
            )
        )

        comparison_outcome = ComparisonOutcome()
        regressions = entries_worse_than_threshold.to_scalability_change(
            Regression,
            workload_name,
            other_endpoint,
        )
        improvements = entries_better_than_threshold.to_scalability_change(
            ScalabilityImprovement,
            workload_name,
            other_endpoint,
        )
        comparison_outcome.append_regressions(
            regressions,
            improvements,
            entries_worse_than_threshold,
            entries_better_than_threshold,
        )
        return comparison_outcome

Ancestors

ResultAnalyzer

Methods

def perform_comparison_in_workload(self, workload_name: str, baseline_endpoint: Endpoint, other_endpoint: Endpoint, regression_baseline_result: WorkloadResult, other_result: WorkloadResult) ‑> ComparisonOutcome

Expand source code Browse git

def perform_comparison_in_workload(
    self,
    workload_name: str,
    baseline_endpoint: Endpoint,
    other_endpoint: Endpoint,
    regression_baseline_result: WorkloadResult,
    other_result: WorkloadResult,
) -> ComparisonOutcome:
    # tps = transactions per seconds (higher is better)

    merged_data = regression_baseline_result.df_totals.merge(other_result.df_totals)
    tps_per_endpoint_data = merged_data.to_enriched_result_frame(
        baseline_endpoint.try_load_version(), other_endpoint.try_load_version()
    )
    entries_worse_than_threshold = tps_per_endpoint_data.to_filtered_with_threshold(
        self.max_deviation_as_percent_decimal,
        match_results_better_than_baseline=False,
    )
    entries_better_than_threshold = (
        tps_per_endpoint_data.to_filtered_with_threshold(
            self.max_deviation_as_percent_decimal,
            match_results_better_than_baseline=True,
        )
    )

    comparison_outcome = ComparisonOutcome()
    regressions = entries_worse_than_threshold.to_scalability_change(
        Regression,
        workload_name,
        other_endpoint,
    )
    improvements = entries_better_than_threshold.to_scalability_change(
        ScalabilityImprovement,
        workload_name,
        other_endpoint,
    )
    comparison_outcome.append_regressions(
        regressions,
        improvements,
        entries_worse_than_threshold,
        entries_better_than_threshold,
    )
    return comparison_outcome