feat: no longer do cache rollups in new pipeline

we don’t need to cache rollups in the new pipeline anymore since the API
is consuming from timescale directly

So we’re removing all code related to caching rollups in the timescale 
implementation of the TA pipeline

Author: joseph-sentry

apps/codecov-api/utils/test_results.py

@@ -1,5 +1,4 @@
-import tempfile
-from datetime import UTC, date, datetime, timedelta
+from datetime import UTC, datetime
 from functools import lru_cache
 
 import polars as pl
@@ -154,121 +153,3 @@ def old_get_results(
     table = dedup_table(table)
 
     return table
-
-
-def rollup_blob_path(repoid: int, branch: str | None = None) -> str:
-    return (
-        f"test_analytics/branch_rollups/{repoid}/{branch}.arrow"
-        if branch
-        else f"test_analytics/repo_rollups/{repoid}.arrow"
-    )
-
-
-def no_version_agg_table(table: pl.LazyFrame) -> pl.LazyFrame:
-    total_tests = pl.col("fail_count") + pl.col("pass_count")
-    total_time = pl.col("avg_duration") * total_tests
-
-    failure_rate_expr = (pl.col("fail_count")).sum() / total_tests.sum()
-
-    flake_rate_expr = (pl.col("flaky_fail_count")).sum() / total_tests.sum()
-
-    avg_duration_expr = total_time.sum() / total_tests.sum()
-
-    total_duration_expr = total_time.sum()
-
-    table = table.group_by(pl.col("computed_name").alias("name")).agg(
-        pl.col("flags")
-        .explode()
-        .unique()
-        .alias("flags"),  # TODO: filter by this before we aggregate
-        pl.col("failing_commits").sum().alias("commits_where_fail"),
-        pl.col("last_duration").max().alias("last_duration"),
-        failure_rate_expr.alias("failure_rate"),
-        flake_rate_expr.alias("flake_rate"),
-        avg_duration_expr.alias("avg_duration"),
-        total_duration_expr.alias("total_duration"),
-        pl.col("pass_count").sum().alias("total_pass_count"),
-        pl.col("fail_count").sum().alias("total_fail_count"),
-        pl.col("flaky_fail_count").sum().alias("total_flaky_fail_count"),
-        pl.col("skip_count").sum().alias("total_skip_count"),
-        pl.col("updated_at").max().alias("updated_at"),
-    )
-
-    return table
-
-
-def v1_agg_table(table: pl.LazyFrame) -> pl.LazyFrame:
-    total_tests = pl.col("fail_count") + pl.col("pass_count")
-    total_time = pl.col("avg_duration") * total_tests
-
-    failure_rate_expr = (pl.col("fail_count")).sum() / total_tests.sum()
-
-    flake_rate_expr = (pl.col("flaky_fail_count")).sum() / total_tests.sum()
-
-    avg_duration_expr = total_time.sum() / total_tests.sum()
-
-    total_duration_expr = total_time.sum()
-
-    table = table.group_by(pl.col("computed_name").alias("name")).agg(
-        pl.col("testsuite").alias(
-            "testsuite"
-        ),  # TODO: filter by this before we aggregate
-        pl.col("flags")
-        .explode()
-        .unique()
-        .alias("flags"),  # TODO: filter by this before we aggregate
-        pl.col("failing_commits").sum().alias("commits_where_fail"),
-        pl.col("last_duration").max().alias("last_duration"),
-        failure_rate_expr.fill_nan(0).alias("failure_rate"),
-        flake_rate_expr.fill_nan(0).alias("flake_rate"),
-        avg_duration_expr.fill_nan(0).alias("avg_duration"),
-        total_duration_expr.alias("total_duration"),
-        pl.col("pass_count").sum().alias("total_pass_count"),
-        pl.col("fail_count").sum().alias("total_fail_count"),
-        pl.col("flaky_fail_count").sum().alias("total_flaky_fail_count"),
-        pl.col("skip_count").sum().alias("total_skip_count"),
-        pl.col("updated_at").max().alias("updated_at"),
-    )
-
-    return table
-
-
-def new_get_results(
-    repoid: int,
-    branch: str | None,
-    interval_start: int,
-    interval_end: int | None = None,
-) -> pl.DataFrame | None:
-    storage_service = get_appropriate_storage_service(repoid)
-    key = rollup_blob_path(repoid, branch)
-    try:
-        with tempfile.TemporaryFile() as tmp:
-            metadata = {}
-            storage_service.read_file(
-                bucket_name=settings.GCS_BUCKET_NAME,
-                path=key,
-                file_obj=tmp,
-                metadata_container=metadata,
-            )
-
-            table = pl.scan_ipc(tmp)
-
-            # filter start
-            start_date = date.today() - timedelta(days=interval_start)
-            table = table.filter(pl.col("timestamp_bin") >= start_date)
-
-            # filter end
-            if interval_end is not None:
-                end_date = date.today() - timedelta(days=interval_end)
-                table = table.filter(pl.col("timestamp_bin") <= end_date)
-
-            # aggregate
-            match metadata.get("version"):
-                case "1":
-                    table = v1_agg_table(table)
-                case _:  # no version is missding
-                    table = no_version_agg_table(table)
-
-            return table.collect()
-    except FileNotInStorageError:
-        return None

apps/codecov-api/utils/timescale/test_results.py

@@ -153,10 +153,7 @@ def get_test_data_queryset_via_testrun(
             distinct=True,
         ),
         total_count=(
-            F("total_pass_count")
-            + F("total_fail_count")
-            + F("total_skip_count")
-            + F("total_flaky_fail_count")
+            F("total_pass_count") + F("total_fail_count") + F("total_flaky_fail_count")
         ),
         failure_rate=Case(
             When(

apps/worker/services/test_analytics/ta_cache_rollups.py

@@ -34,7 +34,7 @@
     TestResultsNotifier,
     should_do_flaky_detection,
 )
-from shared.celery_config import cache_test_rollups_task_name, process_flakes_task_name
+from shared.celery_config import process_flakes_task_name
 from shared.django_apps.reports.models import ReportSession, UploadError
 from shared.helpers.redis import get_redis_connection
 from shared.reports.types import UploadType
@@ -130,16 +130,6 @@ def queue_followup_tasks(
             },
         )
 
-    if commit.branch is not None:
-        celery_app.send_task(
-            cache_test_rollups_task_name,
-            kwargs={
-                "repo_id": repo.repoid,
-                "branch": commit.branch,
-                "impl_type": impl_type,
-            },
-        )
-
 
 @sentry_sdk.trace
 def new_impl(

apps/worker/services/test_analytics/ta_finish_upload.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
-from dataclasses import dataclass
-from datetime import datetime, timedelta
+from datetime import datetime
 from typing import TypedDict
 
 import test_results_parser
@@ -11,14 +10,10 @@
 from services.test_results import FlakeInfo
 from shared.django_apps.ta_timeseries.models import (
     Testrun,
-    TestrunBranchSummary,
-    TestrunSummary,
     calc_test_id,
 )
 from shared.django_apps.test_analytics.models import Flake
 
-LOWER_BOUND_NUM_DAYS = 60
-
 
 def get_flaky_tests_set(repo_id: int) -> set[bytes]:
     return {
@@ -177,92 +172,3 @@ def update_testrun_to_flaky(timestamp: datetime, test_id: bytes):
             "UPDATE ta_timeseries_testrun SET outcome = %s WHERE timestamp = %s AND test_id = %s",
             ["flaky_fail", timestamp, test_id],
         )
-
-
-def timestamp_lower_bound():
-    return datetime.now() - timedelta(days=LOWER_BOUND_NUM_DAYS)
-
-
-def get_summary(repo_id: int) -> list[TestrunSummary]:
-    return list(
-        TestrunSummary.objects.filter(
-            repo_id=repo_id, timestamp_bin__gte=timestamp_lower_bound()
-        )
-    )
-
-
-def get_branch_summary(repo_id: int, branch: str) -> list[TestrunBranchSummary]:
-    return list(
-        TestrunBranchSummary.objects.filter(
-            repo_id=repo_id, branch=branch, timestamp_bin__gte=timestamp_lower_bound()
-        )
-    )
-
-
-@dataclass
-class BranchSummary:
-    testsuite: str
-    classname: str
-    name: str
-    timestamp_bin: datetime
-    computed_name: str
-    failing_commits: int
-    last_duration_seconds: float
-    avg_duration_seconds: float
-    pass_count: int
-    fail_count: int
-    skip_count: int
-    flaky_fail_count: int
-    updated_at: datetime
-    flags: list[str]
-
-
-def get_testrun_branch_summary_via_testrun(
-    repo_id: int, branch: str
-) -> list[BranchSummary]:
-    with connections["ta_timeseries"].cursor() as cursor:
-        cursor.execute(
-            """
-            select
-                testsuite,
-                classname,
-                name,
-                time_bucket(interval '1 days', timestamp) as timestamp_bin,
-
-                min(computed_name) as computed_name,
-                COUNT(DISTINCT CASE WHEN outcome = 'failure' OR outcome = 'flaky_fail' THEN commit_sha ELSE NULL END) AS failing_commits,
-                last(duration_seconds, timestamp) as last_duration_seconds,
-                avg(duration_seconds) as avg_duration_seconds,
-                COUNT(*) FILTER (WHERE outcome = 'pass') AS pass_count,
-                COUNT(*) FILTER (WHERE outcome = 'failure') AS fail_count,
-                COUNT(*) FILTER (WHERE outcome = 'skip') AS skip_count,
-                COUNT(*) FILTER (WHERE outcome = 'flaky_fail') AS flaky_fail_count,
-                MAX(timestamp) AS updated_at,
-                array_merge_dedup_agg(flags) as flags
-            from ta_timeseries_testrun
-            where repo_id = %s and branch = %s and timestamp > %s
-            group by
-                testsuite, classname, name, timestamp_bin;
-            """,
-            [repo_id, branch, timestamp_lower_bound()],
-        )
-
-        return [
-            BranchSummary(
-                testsuite=row[0],
-                classname=row[1],
-                name=row[2],
-                timestamp_bin=row[3],
-                computed_name=row[4],
-                failing_commits=row[5],
-                last_duration_seconds=row[6],
-                avg_duration_seconds=row[7],
-                pass_count=row[8],
-                fail_count=row[9],
-                skip_count=row[10],
-                flaky_fail_count=row[11],
-                updated_at=row[12],
-                flags=row[13] or [],
-            )
-            for row in cursor.fetchall()
-        ]