[postgres] Add calls to query metrics derivative compute execution_indicators (#20096)

* add calls to metrics derivative compute execution_indicators

* fix tests

* add changelog

* update base package min version

Author: lu-zhengda

postgres/changelog.d/20096.added

@@ -0,0 +1 @@
+Added query call count `calls` to StatementMetrics execution indicators to filter out false positives from normalized queries being evicted and re-inserted with same call count and slight duration change.

postgres/datadog_checks/postgres/statements.py

@@ -613,7 +613,7 @@ def _collect_metrics_rows(self):
         available_columns = set(rows[0].keys())
         metric_columns = available_columns & PG_STAT_STATEMENTS_METRICS_COLUMNS
 
-        rows = self._state.compute_derivative_rows(rows, metric_columns, key=_row_key)
+        rows = self._state.compute_derivative_rows(rows, metric_columns, key=_row_key, execution_indicators=['calls'])
         self._check.gauge(
             'dd.postgres.queries.query_rows_raw',
             len(rows),

postgres/pyproject.toml

@@ -28,7 +28,7 @@ classifiers = [
     "Private :: Do Not Upload",
 ]
 dependencies = [
-    "datadog-checks-base>=37.4.0",
+    "datadog-checks-base>=37.10.0",
 ]
 dynamic = [
     "version",

postgres/tests/test_statements.py

@@ -28,6 +28,8 @@
     PG_STAT_STATEMENTS_TIMING_COLUMNS,
     PG_STAT_STATEMENTS_TIMING_COLUMNS_LT_17,
     PostgresStatementMetrics,
+    StatementMetrics,
+    _row_key,
 )
 from datadog_checks.postgres.util import payload_pg_version
 from datadog_checks.postgres.version_utils import V12
@@ -2060,6 +2062,41 @@ def test_pg_stat_statements_dealloc(aggregator, integration_check, dbm_instance_
     aggregator.assert_metric("postgresql.pg_stat_statements.count", tags=expected_tags)
 
 
+@pytest.mark.parametrize(
+    "prev_row,curr_row,expected_included",
+    [
+        # Base case - normal increase in calls and duration
+        ({"calls": 5, "total_time": 100}, {"calls": 6, "total_time": 120}, True),
+        # Test case - no change in calls but duration increased
+        ({"calls": 5, "total_time": 100}, {"calls": 5, "total_time": 120}, False),
+        # Test case - calls=0 with duration>0
+        ({"calls": 0, "total_time": 0}, {"calls": 0, "total_time": 50}, False),
+    ],
+)
+def test_statement_metrics_execution_indicators(prev_row, curr_row, expected_included):
+    """Test that queries are only included in results if execution indicators (calls) have increased."""
+    # Setup
+    state = StatementMetrics()
+    metrics = {"calls", "total_time"}
+    execution_indicators = ["calls"]
+
+    # Add query signature and other required fields
+    prev_row.update({"query_signature": "test_query", "datname": "test_db", "rolname": "test_role"})
+    curr_row.update({"query_signature": "test_query", "datname": "test_db", "rolname": "test_role"})
+
+    # First run to establish baseline
+    state.compute_derivative_rows([prev_row], metrics, key=_row_key, execution_indicators=execution_indicators)
+
+    # Second run to test the behavior
+    result = state.compute_derivative_rows([curr_row], metrics, key=_row_key, execution_indicators=execution_indicators)
+
+    # Verify
+    assert bool(result) == expected_included
+    if result:
+        assert result[0]["calls"] == curr_row["calls"] - prev_row["calls"]
+        assert result[0]["total_time"] == curr_row["total_time"] - prev_row["total_time"]
+
+
 @requires_over_13
 def test_plan_time_metrics(aggregator, integration_check, dbm_instance):
     dbm_instance['pg_stat_statements_view'] = "pg_stat_statements"