feat: redesign Run History page with stats dashboard, row counts, and job switcher

backend/backend/core/scheduler/celery_tasks.py

@@ -30,6 +30,15 @@
 
 logger = logging.getLogger(__name__)
 
+
+def _sum_or_none(results, attr):
+    """Sum an attribute across results. Return None only if ALL values are None,
+    return 0 if the sum is legitimately zero (not all missing)."""
+    values = [getattr(r, attr, None) for r in results]
+    if all(v is None for v in values):
+        return None
+    return sum(v or 0 for v in values)
+
 # Default max duration before a job is considered stuck (1 hour)
 DEFAULT_STUCK_JOB_THRESHOLD_SECONDS = 3600
 
@@ -361,12 +370,22 @@ def _clean_name(raw):
                         "status": r.status,
                         "end_status": r.end_status,
                         "sequence": r.sequence_num,
+                        "rows_affected": getattr(r, "rows_affected", None),
+                        "rows_inserted": getattr(r, "rows_inserted", None),
+                        "rows_updated": getattr(r, "rows_updated", None),
+                        "rows_deleted": getattr(r, "rows_deleted", None),
+                        "type": getattr(r, "materialization", "") or "",
+                        "duration_ms": getattr(r, "duration_ms", None),
                     }
                     for r in user_results
                 ],
                 "total": len(user_results),
                 "passed": sum(1 for r in user_results if r.end_status == "OK"),
                 "failed": sum(1 for r in user_results if r.end_status == "FAIL"),
+                "rows_processed": _sum_or_none(user_results, "rows_affected"),
+                "rows_added": _sum_or_none(user_results, "rows_inserted"),
+                "rows_modified": _sum_or_none(user_results, "rows_updated"),
+                "rows_deleted": _sum_or_none(user_results, "rows_deleted"),
             }
         except Exception:
             _clear_base_result()

backend/backend/core/scheduler/serializer.py

@@ -1,17 +1,105 @@
+from django.contrib.auth import get_user_model
 from rest_framework import serializers
 
 from backend.core.scheduler.models import TaskRunHistory
 
+User = get_user_model()
+
 
 class TaskRunHistorySerializer(serializers.ModelSerializer):
     duration = serializers.SerializerMethodField()
+    duration_ms = serializers.SerializerMethodField()
+    run_number = serializers.SerializerMethodField()
+    triggered_by = serializers.SerializerMethodField()
+    model_count = serializers.SerializerMethodField()
+    failed_models = serializers.SerializerMethodField()
+    skipped_count = serializers.SerializerMethodField()
 
     class Meta:
         model = TaskRunHistory
-        fields = "__all__"  # Include all fields or specify fields like ['id', 'start_time', 'end_time', 'status']
+        fields = [
+            "id", "task_id", "status", "start_time", "end_time",
+            "trigger", "scope", "error_message", "result", "retry_num",
+            "user_task_detail",
+            "duration", "duration_ms", "run_number", "triggered_by",
+            "model_count", "failed_models", "skipped_count",
+        ]
+
+    def _get_user_cache(self):
+        """Batch-load users for all runs in one query, cached per serializer instance."""
+        if not hasattr(self, "_user_cache"):
+            user_ids = set()
+            for obj in self.instance if hasattr(self.instance, '__iter__') else [self.instance]:
+                if obj and isinstance(obj.kwargs, dict) and obj.kwargs.get("user_id"):
+                    user_ids.add(obj.kwargs["user_id"])
+            if user_ids:
+                self._user_cache = {
+                    str(u.id): u for u in User.objects.filter(id__in=user_ids)
+                }
+            else:
+                self._user_cache = {}
+        return self._user_cache
 
     def get_duration(self, obj):
-        """Calculate duration (end_time - start_time)"""
+        """Human-readable duration string."""
         if obj.start_time and obj.end_time:
-            return str(obj.end_time - obj.start_time)  # Convert timedelta to string
-        return None  # If end_time is missing, return None
+            delta = obj.end_time - obj.start_time
+            total_ms = int(delta.total_seconds() * 1000)
+            if total_ms < 1000:
+                return f"{total_ms}ms"
+            elif total_ms < 60000:
+                return f"{total_ms / 1000:.1f}s"
+            else:
+                minutes = total_ms // 60000
+                seconds = (total_ms % 60000) / 1000
+                return f"{minutes}m {seconds:.0f}s"
+        return None
+
+    def get_duration_ms(self, obj):
+        """Duration in milliseconds for sorting/comparison."""
+        if obj.start_time and obj.end_time:
+            return int((obj.end_time - obj.start_time).total_seconds() * 1000)
+        return None
+
+    def get_run_number(self, obj):
+        """Sequential run number from view context (total - offset - idx)."""
+        run_numbers = self.context.get("run_numbers", {})
+        return run_numbers.get(obj.id, 0)
+
+    def get_triggered_by(self, obj):
+        """Resolve user_id from kwargs to username using batch-loaded cache."""
+        if not isinstance(obj.kwargs, dict):
+            return None
+        user_id = obj.kwargs.get("user_id")
+        if not user_id:
+            return None
+        cache = self._get_user_cache()
+        user = cache.get(str(user_id))
+        if user:
+            return {
+                "id": str(user.id),
+                "username": user.get_full_name() or user.username or user.email,
+            }
+        return {"id": str(user_id), "username": "Unknown user"}
+
+    def get_model_count(self, obj):
+        """Total model count from result."""
+        if obj.result and isinstance(obj.result, dict):
+            return obj.result.get("total", 0)
+        return 0
+
+    def get_failed_models(self, obj):
+        """List of failed model names."""
+        if obj.result and isinstance(obj.result, dict):
+            models = obj.result.get("models", [])
+            return [m["name"] for m in models if m.get("end_status") == "FAIL" or m.get("status") == "failure"]
+        return []
+
+    def get_skipped_count(self, obj):
+        """Count of skipped models (total - passed - failed)."""
+        if obj.result and isinstance(obj.result, dict):
+            total = obj.result.get("total", 0)
+            passed = obj.result.get("passed", 0)
+            failed = obj.result.get("failed", 0)
+            return max(0, total - passed - failed)
+        return 0

backend/backend/core/scheduler/urls.py

@@ -7,6 +7,7 @@
     delete_periodic_task,
     update_periodic_task,
     task_run_history,
+    run_stats,
     trigger_task_once,
     trigger_task_once_for_model,
     list_deploy_candidates,
@@ -30,6 +31,7 @@
         name="get_periodic_task",
     ),
     path("/run-history/<int:user_task_id>", task_run_history, name="task_run_history"),
+    path("/run-stats/<int:user_task_id>", run_stats, name="run_stats"),
     path(
         "/trigger-periodic-task/<int:user_task_id>",
         trigger_task_once,

backend/backend/core/scheduler/views.py

@@ -4,6 +4,7 @@
 from datetime import timedelta
 
 from django.utils import timezone
+from django.utils.dateparse import parse_datetime
 from django_celery_beat.models import CrontabSchedule, IntervalSchedule, PeriodicTask
 from rest_framework import status
 from rest_framework.decorators import api_view, permission_classes
@@ -583,6 +584,112 @@ def delete_periodic_task(request, project_id, task_id):
         )
 
 
+@api_view(["GET"])
+@permission_classes([IsAuthenticated])
+def run_stats(request, project_id, user_task_id):
+    """Get aggregated run statistics for a job — stats cards data."""
+    try:
+        query = {"id": user_task_id}
+        if _is_valid_project_id(project_id):
+            query["project__project_uuid"] = project_id
+        task = UserTaskDetails.objects.get(**query)
+        runs = TaskRunHistory.objects.filter(user_task_detail=task)
+
+        now = timezone.now()
+        last_7d = now - timedelta(days=7)
+        last_24h = now - timedelta(hours=24)
+        prev_24h_start = now - timedelta(hours=48)
+
+        # Success rate (7 days) — only count completed runs in denominator
+        runs_7d = runs.filter(start_time__gte=last_7d)
+        completed_7d = runs_7d.filter(status__in=["SUCCESS", "FAILURE"])
+        total_7d = completed_7d.count()
+        success_7d = completed_7d.filter(status="SUCCESS").count()
+        success_rate = round((success_7d / total_7d * 100), 1) if total_7d > 0 else None
+
+        # Average duration (successful runs, 7 days)
+        successful_runs_7d = runs_7d.filter(status="SUCCESS", start_time__isnull=False, end_time__isnull=False)
+        avg_duration_ms = None
+        if successful_runs_7d.exists():
+            durations = [(r.end_time - r.start_time).total_seconds() * 1000 for r in successful_runs_7d]
+            avg_duration_ms = int(sum(durations) / len(durations))
+
+        # Failures (24h) + comparison with previous 24h
+        failures_24h = runs.filter(start_time__gte=last_24h, status="FAILURE").count()
+        failures_prev_24h = runs.filter(
+            start_time__gte=prev_24h_start, start_time__lt=last_24h, status="FAILURE"
+        ).count()
+
+        # Last successful run
+        last_success = runs.filter(status="SUCCESS").order_by("-end_time").first()
+        last_success_time = last_success.end_time if last_success else None
+
+        # Expected duration (avg of last 5 successful runs)
+        recent_successes = runs.filter(
+            status="SUCCESS", start_time__isnull=False, end_time__isnull=False
+        ).order_by("-end_time")[:5]
+        expected_duration_ms = None
+        if recent_successes.exists():
+            durations = [(r.end_time - r.start_time).total_seconds() * 1000 for r in recent_successes]
+            expected_duration_ms = int(sum(durations) / len(durations))
+
+        # Duration trend (last 10 completed runs for sparkline)
+        recent_runs = list(runs.filter(
+            start_time__isnull=False, end_time__isnull=False
+        ).order_by("-end_time")[:10])
+        recent_runs.reverse()  # chronological order for sparkline
+        duration_trend = [
+            int((r.end_time - r.start_time).total_seconds() * 1000) for r in recent_runs
+        ]
+
+        # Schedule info
+        schedule_type = None
+        schedule_label = None
+        periodic = None
+        try:
+            periodic = task.periodic_task
+            if periodic:
+                if periodic.crontab:
+                    schedule_type = "cron"
+                    c = periodic.crontab
+                    schedule_label = f"{c.minute} {c.hour} {c.day_of_month} {c.month_of_year} {c.day_of_week}"
+                elif periodic.interval:
+                    schedule_type = "interval"
+                    schedule_label = f"Every {periodic.interval.every} {periodic.interval.period}"
+        except Exception:
+            periodic = None
+
+        return Response({
+            "success": True,
+            "data": {
+                "success_rate_7d": success_rate,
+                "success_count_7d": success_7d,
+                "total_count_7d": total_7d,
+                "avg_duration_ms": avg_duration_ms,
+                "failures_24h": failures_24h,
+                "failures_prev_24h": failures_prev_24h,
+                "failures_change": failures_24h - failures_prev_24h,
+                "last_successful_run": last_success_time,
+                "expected_duration_ms": expected_duration_ms,
+                "duration_trend": duration_trend,
+                "total_runs": runs.count(),
+                "job_name": task.task_name,
+                "environment": {
+                    "name": task.environment.environment_name if task.environment else None,
+                    "type": task.environment.deployment_type if task.environment else None,
+                },
+                "schedule_type": schedule_type,
+                "schedule_label": schedule_label,
+                "schedule_enabled": periodic.enabled if periodic else False,
+            },
+        }, status=status.HTTP_200_OK)
+    except UserTaskDetails.DoesNotExist:
+        return Response({"error": "Task not found"}, status=status.HTTP_404_NOT_FOUND)
+    except Exception as e:
+        logger.error(f"Error getting run stats: {e}", exc_info=True)
+        return Response({"error": "Internal server error"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+
 @api_view(["GET"])
 @permission_classes([IsAuthenticated])
 def task_run_history(request, project_id, user_task_id):
@@ -600,18 +707,40 @@ def task_run_history(request, project_id, user_task_id):
         trigger_filter = request.GET.get("trigger")
         scope_filter = request.GET.get("scope")
         status_filter = request.GET.get("status")
+        date_from = request.GET.get("date_from")
+        date_to = request.GET.get("date_to")
+        search = request.GET.get("search")
+
         if trigger_filter:
             runs = runs.filter(trigger=trigger_filter)
         if scope_filter:
             runs = runs.filter(scope=scope_filter)
         if status_filter:
             runs = runs.filter(status=status_filter)
+        if date_from:
+            dt = parse_datetime(date_from)
+            if dt:
+                runs = runs.filter(start_time__gte=dt)
+        if date_to:
+            dt = parse_datetime(date_to)
+            if dt:
+                runs = runs.filter(start_time__lte=dt)
+        if search:
+            runs = runs.filter(error_message__icontains=search)
 
         runs = runs.order_by("-start_time")
         total = runs.count()
 
         offset = (page - 1) * limit
-        serializer = TaskRunHistorySerializer(runs[offset : offset + limit], many=True)
+        page_qs = runs[offset : offset + limit]
+        # Compute run numbers from total and offset — no extra query needed
+        run_numbers = {
+            run.id: total - offset - idx
+            for idx, run in enumerate(page_qs)
+        }
+        serializer = TaskRunHistorySerializer(
+            page_qs, many=True, context={"run_numbers": run_numbers}
+        )
 
         return Response(
             {
@@ -620,6 +749,7 @@ def task_run_history(request, project_id, user_task_id):
                     "page_items": {
                         "id": task.id,
                         "job_name": task.task_name,
+                        "project_id": str(task.project.project_uuid) if task.project else None,
                         "env_type": task.environment.deployment_type
                         if task.environment
                         else None,
@@ -705,9 +835,10 @@ def trigger_task_once(request, project_id, user_task_id):
     synchronous (in-process) execution so local dev works without Redis.
     """
     try:
-        task = UserTaskDetails.objects.get(
-            id=user_task_id, project__project_uuid=project_id
-        )
+        query = {"id": user_task_id}
+        if _is_valid_project_id(project_id):
+            query["project__project_uuid"] = project_id
+        task = UserTaskDetails.objects.get(**query)
     except UserTaskDetails.DoesNotExist:
         return Response(
             {"error": "Task not found"}, status=status.HTTP_404_NOT_FOUND

backend/visitran/adapters/adapter.py

@@ -67,10 +67,10 @@ def db_scd(self) -> BaseSCD:
     def db_reader(self) -> BaseDBReader:
         return self._db_reader
 
-    def run_model(self, visitran_model: VisitranModel) -> None:
+    def run_model(self, visitran_model: VisitranModel):
         self.load_model(model=visitran_model)
         fire_event(MaterializationType(materialization=str(visitran_model.materialization)))
-        self.db_model.execute()
+        return self.db_model.execute()
 
     def run_seeds(self, schema: str, abs_path: str) -> None:
         seed_obj = self.load_seed(schema, abs_path)

backend/visitran/adapters/bigquery/connection.py

@@ -269,16 +269,9 @@ def merge_into_table(
         target_table_name: str,
         select_statement: Table,
         primary_key: Union[str, list[str]] = None,
-    ) -> None:
+    ) -> dict:
         """Efficient upsert using DELETE + INSERT for BigQuery.
-
-        This approach is more efficient than MERGE for BigQuery because:
-        1. BigQuery is optimized for bulk operations
-        2. DELETE + INSERT performs better than UPDATE operations
-        3. Works better with BigQuery's partitioning strategy
-
-        Args:
-            primary_key: Can be a single column name (str) or list of column names for composite keys
+        Returns dict with rows_affected.
         """
         try:
             fire_event(
@@ -378,6 +371,7 @@ def merge_into_table(
             raise Exception(
                 f"BigQuery incremental upsert failed for {schema_name}.{target_table_name}: {str(e)}"
             ) from e
+        return {"rows_affected": None}  # BigQuery: fallback to get_table_row_count in BaseModel