feat(dashboard): add query texts to pg_stat_statements table

Add query_text column to the "Detailed table view" in the Query
Performance Analysis dashboard. This allows users to see the actual
SQL query text alongside the query metrics.

Changes:
- Add PostgreSQL sink database connection to Flask backend for
  fetching query texts from pgss_queryid_queries table
- Update process_pgss_data() to include query_text in CSV output
- Add query_text column selector to Grafana dashboard with
  400px width and text wrapping enabled
- Update docker-compose.yml with POSTGRES_SINK_URL env var
- Update Helm chart flask-deployment with sink database connection
- Add psycopg2-binary dependency to requirements.txt

Author: claude

config/grafana/dashboards/Dashboard_2_Aggregated_query_analysis.json

@@ -163,6 +163,25 @@
                     ]
                   }
                 ]
+              },
+              {
+                "matcher": {
+                  "id": "byName",
+                  "options": "Query Text"
+                },
+                "properties": [
+                  {
+                    "id": "custom.width",
+                    "value": 400
+                  },
+                  {
+                    "id": "custom.cellOptions",
+                    "value": {
+                      "type": "auto",
+                      "wrapText": true
+                    }
+                  }
+                ]
               }
             ]
           },
@@ -200,6 +219,11 @@
                   "text": "Query ID",
                   "type": "string"
                 },
+                {
+                  "selector": "query_text",
+                  "text": "Query Text",
+                  "type": "string"
+                },
                 {
                   "selector": "duration_seconds",
                   "text": "Duration (seconds)",
@@ -395,37 +419,38 @@
                 },
                 "includeByName": {},
                 "indexByName": {
-                  "Block read time (ms)": 27,
-                  "Block read time/call (ms)": 29,
-                  "Block read time/sec (ms)": 28,
-                  "Block write time (ms)": 24,
-                  "Block write time/call (ms)": 26,
-                  "Block write time/sec (ms)": 25,
-                  "Calls": 1,
-                  "Calls/sec": 2,
-                  "Duration (seconds)": 30,
-                  "Exec time (ms)": 3,
-                  "Exec time/call (ms)": 5,
-                  "Exec time/sec (ms/s)": 4,
-                  "Planning time (ms)": 6,
-                  "Planning time/call (ms)": 8,
-                  "Planning time/sec (ms/s)": 7,
+                  "Block read time (ms)": 28,
+                  "Block read time/call (ms)": 30,
+                  "Block read time/sec (ms)": 29,
+                  "Block write time (ms)": 25,
+                  "Block write time/call (ms)": 27,
+                  "Block write time/sec (ms)": 26,
+                  "Calls": 2,
+                  "Calls/sec": 3,
+                  "Duration (seconds)": 31,
+                  "Exec time (ms)": 4,
+                  "Exec time/call (ms)": 6,
+                  "Exec time/sec (ms/s)": 5,
+                  "Planning time (ms)": 7,
+                  "Planning time/call (ms)": 9,
+                  "Planning time/sec (ms/s)": 8,
                   "Query ID": 0,
-                  "Rows": 9,
-                  "Rows/call": 11,
-                  "Rows/sec": 10,
-                  "Shared blocks dirtied": 21,
-                  "Shared blocks dirtied/call": 23,
-                  "Shared blocks dirtied/sec": 22,
-                  "Shared blocks hit": 12,
-                  "Shared blocks hit/call": 14,
-                  "Shared blocks hit/sec": 13,
-                  "Shared blocks read": 15,
-                  "Shared blocks read/call": 17,
-                  "Shared blocks read/sec": 16,
-                  "Shared blocks written": 18,
-                  "Shared blocks written/call": 20,
-                  "Shared blocks written/sec": 19
+                  "Query Text": 1,
+                  "Rows": 10,
+                  "Rows/call": 12,
+                  "Rows/sec": 11,
+                  "Shared blocks dirtied": 22,
+                  "Shared blocks dirtied/call": 24,
+                  "Shared blocks dirtied/sec": 23,
+                  "Shared blocks hit": 13,
+                  "Shared blocks hit/call": 15,
+                  "Shared blocks hit/sec": 14,
+                  "Shared blocks read": 16,
+                  "Shared blocks read/call": 18,
+                  "Shared blocks read/sec": 17,
+                  "Shared blocks written": 19,
+                  "Shared blocks written/call": 21,
+                  "Shared blocks written/sec": 20
                 },
                 "renameByName": {}
               }

docker-compose.yml

@@ -175,8 +175,10 @@ services:
     environment:
       - FLASK_ENV=production
       - PROMETHEUS_URL=http://sink-prometheus:9090
+      - POSTGRES_SINK_URL=postgresql://pgwatch@sink-postgres:5432/measurements
     depends_on:
       - sink-prometheus
+      - sink-postgres
     restart: unless-stopped
 
   # PostgreSQL Reports Generator - Runs reports after 30 minutes

monitoring_flask_backend/app.py

@@ -5,13 +5,18 @@
 from datetime import datetime, timezone, timedelta
 import logging
 import os
+import psycopg2
+import psycopg2.extras
 
 # Configure logging
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
 app = Flask(__name__)
 
+# PostgreSQL sink connection for query text lookups
+POSTGRES_SINK_URL = os.environ.get('POSTGRES_SINK_URL', 'postgresql://pgwatch@sink-postgres:5432/measurements')
+
 # Prometheus connection - use environment variable with fallback
 PROMETHEUS_URL = os.environ.get('PROMETHEUS_URL', 'http://localhost:8428')
 
@@ -37,6 +42,64 @@ def get_prometheus_client():
         logger.error(f"Failed to connect to Prometheus: {e}")
         raise
 
+
+def get_query_texts_from_sink(db_name: str = None) -> dict:
+    """
+    Fetch queryid-to-query text mappings from the PostgreSQL sink database.
+
+    Args:
+        db_name: Optional database name to filter results
+
+    Returns:
+        Dictionary mapping queryid to query text
+    """
+    query_texts = {}
+
+    try:
+        conn = psycopg2.connect(POSTGRES_SINK_URL)
+        with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor:
+            if db_name:
+                query = """
+                    SELECT DISTINCT ON (data->>'queryid')
+                        data->>'queryid' as queryid,
+                        data->>'query' as query
+                    FROM public.pgss_queryid_queries
+                    WHERE
+                        dbname = %s
+                        AND data->>'queryid' IS NOT NULL
+                        AND data->>'query' IS NOT NULL
+                    ORDER BY data->>'queryid', time DESC
+                """
+                cursor.execute(query, (db_name,))
+            else:
+                query = """
+                    SELECT DISTINCT ON (data->>'queryid')
+                        data->>'queryid' as queryid,
+                        data->>'query' as query
+                    FROM public.pgss_queryid_queries
+                    WHERE
+                        data->>'queryid' IS NOT NULL
+                        AND data->>'query' IS NOT NULL
+                    ORDER BY data->>'queryid', time DESC
+                """
+                cursor.execute(query)
+
+            for row in cursor:
+                queryid = row['queryid']
+                query_text = row['query']
+                if queryid:
+                    # Truncate very long queries for display
+                    if query_text and len(query_text) > 500:
+                        query_text = query_text[:500] + '...'
+                    query_texts[queryid] = query_text or ''
+
+        conn.close()
+    except Exception as e:
+        logger.warning(f"Failed to fetch query texts from sink database: {e}")
+
+    return query_texts
+
+
 def read_version_file(filepath, default='unknown'):
     """Read version information from file"""
     try:
@@ -182,29 +245,33 @@ def get_pgss_metrics_csv():
                 logger.warning(f"Failed to query metric {metric}: {e}")
                 continue
 
+        # Fetch query texts from sink database
+        query_texts = get_query_texts_from_sink(db_name)
+        logger.info(f"Fetched {len(query_texts)} query texts from sink database")
+
         # Process the data to calculate differences
-        csv_data = process_pgss_data(start_data, end_data, start_dt, end_dt)
+        csv_data = process_pgss_data(start_data, end_data, start_dt, end_dt, query_texts)
 
         # Create CSV response
         output = io.StringIO()
         if csv_data:
-            # Define explicit field order with queryid first, then duration, then metrics with their rates
-            base_fields = ['queryid', 'duration_seconds']
+            # Define explicit field order with queryid first, query_text second, then duration, then metrics with their rates
+            base_fields = ['queryid', 'query_text', 'duration_seconds']
             all_metric_fields = []
-            
+
             # Get metric fields from the mapping in specific order with their rates
             desired_order = [
-                'calls', 'exec_time', 'plan_time', 'rows', 'shared_blks_hit', 
+                'calls', 'exec_time', 'plan_time', 'rows', 'shared_blks_hit',
                 'shared_blks_read', 'shared_blks_dirtied', 'shared_blks_written',
                 'blk_read_time', 'blk_write_time'
             ]
-            
+
             for display_name in desired_order:
                 if display_name in METRIC_NAME_MAPPING.values():
                     all_metric_fields.append(display_name)
                     all_metric_fields.append(f'{display_name}_per_sec')
                     all_metric_fields.append(f'{display_name}_per_call')
-            
+
             # Combine all fields in desired order
             all_fields = base_fields + all_metric_fields
 
@@ -226,10 +293,20 @@ def get_pgss_metrics_csv():
         logger.error(f"Error processing request: {e}")
         return jsonify({"error": str(e)}), 500
 
-def process_pgss_data(start_data, end_data, start_time, end_time):
+def process_pgss_data(start_data, end_data, start_time, end_time, query_texts=None):
     """
     Process pg_stat_statements data and calculate differences between start and end times
+
+    Args:
+        start_data: Prometheus data at start time
+        end_data: Prometheus data at end time
+        start_time: Start datetime
+        end_time: End datetime
+        query_texts: Optional dictionary mapping queryid to query text
     """
+    if query_texts is None:
+        query_texts = {}
+
     # Convert Prometheus data to dictionaries
     start_metrics = prometheus_to_dict(start_data, start_time)
     end_metrics = prometheus_to_dict(end_data, end_time)
@@ -264,9 +341,10 @@ def process_pgss_data(start_data, end_data, start_time, end_time):
             # Fallback to query parameter duration if timestamps are missing
             actual_duration = (end_time - start_time).total_seconds()
 
-        # Create result row
+        # Create result row with query text
         row = {
             'queryid': query_id,
+            'query_text': query_texts.get(query_id, ''),
             'duration_seconds': actual_duration
         }
 

monitoring_flask_backend/requirements.txt

@@ -3,4 +3,5 @@ prometheus-api-client==0.5.4
 python-dateutil==2.8.2
 gunicorn==23.0.0
 requests==2.32.3
-pytest==8.3.4 
+psycopg2-binary==2.9.9
+pytest==8.3.4

postgres_ai_helm/templates/flask-deployment.yaml

@@ -27,6 +27,9 @@ spec:
         - name: wait-for-victoriametrics
           image: busybox:1.36
           command: ['sh', '-c', 'until nc -z {{ include "postgres-ai-monitoring.fullname" . }}-victoriametrics {{ .Values.victoriaMetrics.service.port }}; do echo waiting for victoriametrics; sleep 2; done']
+        - name: wait-for-sink-postgres
+          image: busybox:1.36
+          command: ['sh', '-c', 'until nc -z {{ include "postgres-ai-monitoring.fullname" . }}-sink-postgres 5432; do echo waiting for sink-postgres; sleep 2; done']
       containers:
         - name: flask
           image: {{ .Values.flask.image }}
@@ -36,6 +39,13 @@ spec:
               value: "production"
             - name: PROMETHEUS_URL
               value: "http://{{ include "postgres-ai-monitoring.fullname" . }}-victoriametrics:{{ .Values.victoriaMetrics.service.port }}"
+            - name: POSTGRES_PASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: {{ include "postgres-ai-monitoring.secretName" . }}
+                  key: postgres-password
+            - name: POSTGRES_SINK_URL
+              value: "postgresql://{{ .Values.sinkPostgres.user }}:$(POSTGRES_PASSWORD)@{{ include "postgres-ai-monitoring.fullname" . }}-sink-postgres:5432/{{ .Values.sinkPostgres.database }}"
             {{- range $key, $value := .Values.flask.env }}
             - name: {{ $key }}
               value: {{ $value | quote }}