[Compiler time series data] Add fetch raw data logics (#7137)

support fetch raw data as time series for compiler data

support arch mapping between the  db and api

make the query logics as commits driven
1. if commits is provided in api, then fetch data using the list of
commits
2. if not provided, use startTime stopTime to fetch list of unique
commits, then fetch the time series data

Author: yangw-dev

aws/lambda/benchmark_regression_summary_report/common/regression_utils.py

@@ -128,11 +128,18 @@ def detect_regressions_with_policies(
         logger.info("Generating regression results ...")
         results: List[PerGroupResult] = []
 
+        missing_policy = set()  # for logging
+
         for key in sorted(dp_map.keys()):
             cur_item = dp_map.get(key)
             gi = cur_item["group_info"] if cur_item else {}
             points: List[Any] = cur_item["values"] if cur_item else []
 
+            policy = self._resolve_policy(metric_policies, gi.get("metric", ""))
+            if not policy:
+                missing_policy.add(gi.get("metric", ""))
+                continue
+
             base_item = baseline_map.get(key)
             if not base_item:
                 logger.warning("Skip. No baseline item found for %s", key)
@@ -147,20 +154,6 @@ def detect_regressions_with_policies(
                     )
                 )
                 continue
-            policy = self._resolve_policy(metric_policies, gi.get("metric", ""))
-            if not policy:
-                logger.warning("No policy for %s", gi)
-                results.append(
-                    PerGroupResult(
-                        group_info=gi,
-                        baseline_point=None,
-                        points=[],
-                        label="insufficient_data",
-                        policy=None,
-                        all_baseline_points=[],
-                    )
-                )
-                continue
             baseline_aggre_mode = policy.baseline_aggregation
             baseline_result = self._get_baseline(base_item, baseline_aggre_mode)
             if (
@@ -208,6 +201,11 @@ def detect_regressions_with_policies(
         logger.info("Done. Generated %s regression results", len(results))
         summary = self.summarize_label_counts(results)
 
+        logger.info(
+            "Found metrics existed in data, but no regression policy detected: %s",
+            missing_policy,
+        )
+
         return BenchmarkRegressionReport(
             summary=summary,
             results=results,

torchci/clickhouse_queries/compilers_benchmark_api_commit_query/params.json

@@ -0,0 +1,14 @@
+{
+  "params": {
+    "branches": "Array(String)",
+    "device": "String",
+    "arch": "Array(String)",
+    "dtype": "String",
+    "mode": "String",
+    "startTime": "DateTime64(3)",
+    "stopTime": "DateTime64(3)",
+    "suites": "Array(String)",
+    "workflowIds": "Array(Int64)"
+  },
+  "tests": []
+}

torchci/clickhouse_queries/compilers_benchmark_api_commit_query/query.sql

@@ -0,0 +1,27 @@
+SELECT DISTINCT
+    replaceOne(head_branch, 'refs/heads/', '') AS branch,
+    head_sha AS commit,
+    workflow_id AS id,
+    timestamp
+FROM benchmark.oss_ci_benchmark_torchinductor
+PREWHERE
+    timestamp >= toUnixTimestamp({startTime: DateTime64(3)})
+    AND timestamp < toUnixTimestamp({stopTime:  DateTime64(3)})
+WHERE
+    (
+        has(
+            {branches: Array(String)},
+            replaceOne(head_branch, 'refs/heads/', '')
+        )
+        OR empty({branches: Array(String)})
+    )
+    AND (
+        has({suites: Array(String) }, suite)
+        OR empty({suites: Array(String) })
+    )
+    AND benchmark_dtype = {dtype: String}
+    AND benchmark_mode = {mode: String}
+    AND device = {device: String}
+    AND multiSearchAnyCaseInsensitive(arch, {arch: Array(String)})
+ORDER BY timestamp
+SETTINGS session_timezone = 'UTC';

torchci/clickhouse_queries/compilers_benchmark_api_query/params.json

@@ -4,14 +4,11 @@
     "commits": "Array(String)",
     "compilers": "Array(String)",
     "device": "String",
-    "arch": "String",
+    "arch": "Array(String)",
     "dtype": "String",
     "granularity": "String",
     "mode": "String",
-    "startTime": "DateTime64(3)",
-    "stopTime": "DateTime64(3)",
-    "suites": "Array(String)",
-    "workflowId": "Int64"
+    "suites": "Array(String)"
   },
   "tests": []
 }

torchci/clickhouse_queries/compilers_benchmark_api_query/query.sql

@@ -1,36 +1,38 @@
 SELECT
     workflow_id,
     job_id,
-    head_sha,
-    replaceOne(head_branch, 'refs/heads/', '') AS head_branch,
+    head_sha AS commit,
+    replaceOne(head_branch, 'refs/heads/', '') AS branch,
     suite,
     model_name AS model,
     metric_name AS metric,
     value,
     metric_extra_info AS extra_info,
     benchmark_extra_info['output'] AS output,
+    benchmark_dtype AS dtype,
+    benchmark_mode AS mode,
+    device,
+    arch,
     timestamp,
     DATE_TRUNC({granularity: String}, fromUnixTimestamp(timestamp))
         AS granularity_bucket
 FROM benchmark.oss_ci_benchmark_torchinductor
 WHERE
-    (head_sha) IN (
-        SELECT DISTINCT head_sha
-        FROM benchmark.oss_ci_benchmark_torchinductor
-        PREWHERE
-            timestamp >= toUnixTimestamp({startTime: DateTime64(3,)})
-            AND timestamp < toUnixTimestamp({stopTime: DateTime64(3)})
-    )
+    head_sha IN ({commits: Array(String)})
     AND (
         has(
             {branches: Array(String)},
             replaceOne(head_branch, 'refs/heads/', '')
         )
         OR empty({branches: Array(String)})
     )
+    AND (
+        has({suites: Array(String) }, suite)
+        OR empty({suites: Array(String) })
+    )
     AND benchmark_dtype = {dtype: String}
     AND benchmark_mode = {mode: String}
     AND device = {device: String}
-    AND positionCaseInsensitive(arch, {arch: String}) > 0
-
+    AND multiSearchAnyCaseInsensitive(arch, {arch: Array(String)})
+ORDER BY timestamp
 SETTINGS session_timezone = 'UTC';

torchci/lib/benchmark/api_helper/compilers/get_compiler_benchmark_data.ts

@@ -0,0 +1,121 @@
+import { queryClickhouseSaved } from "lib/clickhouse";
+import { emptyTimeSeriesResponse } from "../utils";
+import {
+  extractBackendSqlStyle,
+  toApiArch,
+  toQueryArch,
+} from "./helpers/common";
+import { toGeneralCompilerData } from "./helpers/general";
+import { toPrecomputeCompilerData } from "./helpers/precompute";
+import { CompilerQueryType } from "./type";
+//["x86_64","NVIDIA A10G","NVIDIA H100 80GB HBM3"]
+const COMPILER_BENCHMARK_TABLE_NAME = "compilers_benchmark_api_query";
+const COMPILER_BENCHMARK_COMMITS_TABLE_NAME =
+  "compilers_benchmark_api_commit_query";
+
+export async function getCompilerBenchmarkData(
+  inputparams: any,
+  type: CompilerQueryType = CompilerQueryType.PRECOMPUTE,
+  format: string = "time_series"
+) {
+  const rows = await getCompilerDataFromClickhouse(inputparams);
+
+  if (rows.length === 0) {
+    return emptyTimeSeriesResponse();
+  }
+
+  switch (type) {
+    case CompilerQueryType.PRECOMPUTE:
+      return toPrecomputeCompilerData(rows, format);
+    case CompilerQueryType.GENERAL:
+      return toGeneralCompilerData(rows, format);
+    default:
+      throw new Error(`Invalid compiler query type, got ${type}`);
+  }
+}
+
+async function getCompilerDataFromClickhouse(inputparams: any): Promise<any[]> {
+  const start = Date.now();
+  const arch_list = toQueryArch(inputparams.device, inputparams.arch);
+  inputparams["arch"] = arch_list;
+
+  // use the startTime and endTime to fetch commits from clickhouse if commits field is not provided
+  if (!inputparams.commits || inputparams.commits.length == 0) {
+    if (!inputparams.startTime || !inputparams.stopTime) {
+      console.log("no commits or start/end time provided in request");
+      return [];
+    }
+    // get commits from clickhouse
+    const commit_results = await queryClickhouseSaved(
+      COMPILER_BENCHMARK_COMMITS_TABLE_NAME,
+      inputparams
+    );
+    // get unique commits
+    const unique_commits = [...new Set(commit_results.map((c) => c.commit))];
+    if (unique_commits.length === 0) {
+      console.log("no commits found in clickhouse using", inputparams);
+      return [];
+    }
+
+    console.log(
+      "no commits provided in request, found unqiue commits",
+      unique_commits
+    );
+
+    if (commit_results.length > 0) {
+      inputparams["commits"] = unique_commits;
+    } else {
+      console.log(`no commits found in clickhouse using ${inputparams}`);
+      return [];
+    }
+  } else {
+    console.log("commits provided in request", inputparams.commits);
+  }
+
+  let rows = await queryClickhouseSaved(
+    COMPILER_BENCHMARK_TABLE_NAME,
+    inputparams
+  );
+  const end = Date.now();
+  console.log("time to get compiler timeseris data", end - start);
+
+  if (rows.length === 0) {
+    return [];
+  }
+
+  // extract backend from output in runtime instead of doing it in the query. since it's expensive for regex matching.
+  // TODO(elainewy): we should add this as a column in the database for less runtime logics.
+  rows.map((row) => {
+    const backend =
+      row.backend && row.backend !== ""
+        ? row.backend
+        : extractBackendSqlStyle(
+            row.output,
+            row.suite,
+            row.dtype,
+            row.mode,
+            row.device
+          );
+    (row["backend"] = backend), (row["compiler"] = backend);
+    row["arch"] = toApiArch(row.device, row.arch);
+  });
+
+  if (inputparams.compilers && inputparams.compilers.length > 0) {
+    rows = rows.filter((row) => {
+      return inputparams.compilers.includes(row.backend);
+    });
+  }
+
+  if (inputparams.models && inputparams.models.length > 0) {
+    rows = rows.filter((row) => {
+      return inputparams.models.includes(row.model);
+    });
+  }
+
+  if (inputparams.metrics && inputparams.metrics.length > 0) {
+    rows = rows.filter((row) => {
+      return inputparams.metrics.includes(row.metric);
+    });
+  }
+  return rows;
+}

torchci/lib/benchmark/api_helper/compilers/helpers/common.ts

@@ -0,0 +1,83 @@
+import { groupByBenchmarkData } from "../../utils";
+
+export function to_table_compiler_data(data: any[]) {
+  const res = groupByBenchmarkData(
+    data,
+    ["dtype", "arch", "device", "mode", "workflow_id", "granularity_bucket"],
+    ["metric", "compiler"]
+  );
+  return res;
+}
+
+export function extractBackendSqlStyle(
+  output: string,
+  suite: string,
+  dtype: string,
+  mode: string,
+  device: string
+): string | null {
+  const esc = (s: string) => s.replace(/[.*+?^${}()|[\]\\]/g, "\\$&");
+  const tail = `_${esc(suite)}_${esc(dtype)}_${esc(mode)}_${esc(device)}_`;
+
+  const temp = output.replace(new RegExp(`${tail}.*$`), "");
+
+  const m = temp.match(/.*[\/\\]([^\/\\]+)$/);
+  return m ? m[1] : null;
+}
+
+export function toQueryArch(device: string, arch: string) {
+  switch (device) {
+    case "rocm":
+      if (arch === "mi300x") return ["mi300x", "mi325x"];
+      return [arch];
+    default:
+      return [arch];
+  }
+}
+
+export function toApiArch(device: string, arch: string): string {
+  const norm = arch.toLowerCase();
+  switch (device) {
+    case "cpu":
+      return norm;
+    case "cuda":
+      if (norm.includes("h100")) return "h100";
+      if (norm.includes("a100")) return "a100";
+      if (norm.includes("a10g")) return "a10g";
+      if (norm.includes("b200")) return "b200";
+      return norm;
+    case "rocm":
+      if (norm.includes("mi300x")) return "mi300x";
+      if (norm.includes("mi325x")) return "mi300x";
+      return norm;
+    case "mps":
+      return norm;
+    default:
+      return norm;
+  }
+}
+
+function deepDiff(obj1: any, obj2: any) {
+  const diffs: string[] = [];
+  const allKeys = new Set([...Object.keys(obj1), ...Object.keys(obj2)]);
+  for (const key of allKeys) {
+    const v1 = obj1[key];
+    const v2 = obj2[key];
+    if (
+      typeof v1 === "object" &&
+      v1 !== null &&
+      typeof v2 === "object" &&
+      v2 !== null
+    ) {
+      if (JSON.stringify(v1) !== JSON.stringify(v2)) {
+        diffs.push(
+          `Key "${key}" differs: ${JSON.stringify(v1)} vs ${JSON.stringify(v2)}`
+        );
+      }
+    } else if (v1 !== v2) {
+      diffs.push(`Key "${key}" differs: ${v1} vs ${v2}`);
+    }
+  }
+
+  return diffs;
+}