fix(table freshness): bugs in calculating historical thresholds

Author: aarthy-dk

testgen/commands/run_test_execution.py

@@ -82,6 +82,10 @@ def run_test_execution(test_suite_id: str | UUID, username: str | None = None, r
 
         sql_generator = TestExecutionSQL(connection, table_group, test_run)
 
+        # Update the thresholds before retrieving the test definitions in the next steps
+        LOG.info("Updating historic test thresholds")
+        execute_db_queries([sql_generator.update_historic_thresholds()])
+
         LOG.info("Retrieving active test definitions in test suite")
         test_defs = fetch_dict_from_db(*sql_generator.get_active_test_definitions())
         test_defs = [TestExecutionDef(**item) for item in test_defs]
@@ -100,9 +104,6 @@ def run_test_execution(test_suite_id: str | UUID, username: str | None = None, r
             )
 
             if valid_test_defs:
-                LOG.info("Updating historic test thresholds")
-                execute_db_queries([sql_generator.update_historic_thresholds()])
-
                 column_types = {(col.schema_name, col.table_name, col.column_name): col.column_type for col in data_chars}
                 for td in valid_test_defs:
                     td.column_type = column_types.get((td.schema_name, td.table_name, td.column_name))

testgen/template/execution/update_historic_thresholds.sql

@@ -1,27 +1,71 @@
-WITH stats AS (
-  SELECT
-    d.id AS test_definition_id,
-    COALESCE(
-      MIN(r.result_signal) FILTER (WHERE d.history_calculation = 'Value'),
-      MIN(r.result_signal::NUMERIC) FILTER (WHERE d.history_calculation = 'Minimum')::VARCHAR,
-      MAX(r.result_signal::NUMERIC) FILTER (WHERE d.history_calculation = 'Maximum')::VARCHAR,
-      SUM(r.result_signal::NUMERIC) FILTER (WHERE d.history_calculation = 'Sum')::VARCHAR,
-      AVG(r.result_signal::NUMERIC) FILTER (WHERE d.history_calculation = 'Average')::VARCHAR
-    ) as calc_signal
-  FROM test_definitions d
-  INNER JOIN LATERAL (
-    SELECT result_signal
-    FROM test_results tr
-    WHERE tr.test_definition_id = d.id
-    ORDER BY tr.test_time DESC
-    LIMIT CASE WHEN d.history_calculation = 'Value' THEN 1 ELSE d.history_lookback END
-  ) AS r ON TRUE
-  WHERE d.test_suite_id    = :TEST_SUITE_ID
-    AND d.test_active      = 'Y'
-    AND d.history_lookback IS NOT NULL
-  GROUP BY d.id, d.history_calculation, d.history_lookback
+WITH filtered_defs AS (
+  -- Step 1: Filter definitions first to minimize join surface area
+  SELECT id,
+    test_suite_id,
+    schema_name,
+    table_name,
+    column_name,
+    test_type,
+    history_calculation,
+    CASE WHEN history_calculation = 'Value' THEN 1 ELSE COALESCE(history_lookback, 1) END AS lookback
+  FROM test_definitions
+  WHERE test_suite_id = :TEST_SUITE_ID
+    AND test_active = 'Y'
+    AND history_calculation IS NOT NULL
+    AND history_lookback IS NOT NULL
+),
+normalized_results AS (
+  -- Step 2: Normalize definition IDs for autogenerated tests
+  SELECT CASE
+      WHEN r.auto_gen THEN d.id
+      ELSE r.test_definition_id
+    END AS test_definition_id,
+    r.test_time,
+    r.result_signal
+  FROM test_results r
+    LEFT JOIN filtered_defs d ON r.auto_gen = TRUE
+    AND r.test_suite_id = d.test_suite_id
+    AND r.schema_name = d.schema_name
+    AND r.table_name IS NOT DISTINCT FROM d.table_name
+    AND r.column_names IS NOT DISTINCT FROM d.column_name
+    AND r.test_type = d.test_type
+  WHERE r.test_suite_id = :TEST_SUITE_ID
+),
+ranked_results AS (
+  -- Step 3: Use a Window Function to get the N most recent results
+  SELECT n.test_definition_id,
+    n.result_signal,
+    CASE
+      WHEN n.result_signal ~ '^-?[0-9]*\.?[0-9]+$' THEN n.result_signal::NUMERIC
+      ELSE NULL
+    END AS signal_numeric,
+    ROW_NUMBER() OVER (PARTITION BY n.test_definition_id ORDER BY n.test_time DESC) AS rank
+  FROM normalized_results n
+  WHERE n.test_definition_id IN (SELECT id FROM filtered_defs)
+),
+stats AS (
+  -- Step 4: Aggregate only the rows within the lookback range
+  SELECT d.id AS test_definition_id,
+    d.history_calculation,
+    MAX(CASE WHEN rr.rank = 1 THEN rr.result_signal END) AS val,
+    MIN(rr.signal_numeric) AS min,
+    MAX(rr.signal_numeric) AS max,
+    SUM(rr.signal_numeric) AS sum,
+    AVG(rr.signal_numeric) AS avg
+  FROM filtered_defs d
+    JOIN ranked_results rr ON d.id = rr.test_definition_id
+  WHERE rr.rank <= d.lookback
+  GROUP BY d.id,
+    d.history_calculation
 )
 UPDATE test_definitions t
-SET baseline_value = s.calc_signal
+SET baseline_value = CASE
+    WHEN s.history_calculation = 'Value' THEN s.val
+    WHEN s.history_calculation = 'Minimum' THEN s.min::VARCHAR
+    WHEN s.history_calculation = 'Maximum' THEN s.max::VARCHAR
+    WHEN s.history_calculation = 'Sum' THEN s.sum::VARCHAR
+    WHEN s.history_calculation = 'Average' THEN s.avg::VARCHAR
+    ELSE NULL
+  END
 FROM stats s
-WHERE t.id = s.test_definition_id;
+WHERE t.id = s.test_definition_id;