Collect PostgreSQL view schema metadata

postgres/assets/configuration/spec.yaml

@@ -988,6 +988,14 @@ files:
           value:
             type: boolean
             example: true
+        - name: collect_views
+          fleet_configurable: true
+          description: |
+            Enable collection of PostgreSQL views and materialized views. Set to false to continue collecting table
+            schemas without collecting views. Defaults to true.
+          value:
+            type: boolean
+            example: true
         - name: max_tables
           fleet_configurable: true
           description: |
@@ -996,6 +1004,14 @@ files:
             type: number
             example: 300
             display_default: 300
+        - name: max_views
+          fleet_configurable: true
+          description: |
+            Maximum amount of views and materialized views the Agent collects from each database.
+          value:
+            type: number
+            example: 1000
+            display_default: 1000
         - name: max_query_duration
           fleet_configurable: true
           description: |

postgres/changelog.d/24972.added

@@ -0,0 +1 @@
+Collect PostgreSQL view definitions and column metadata during schema collection.

postgres/datadog_checks/postgres/config.py

@@ -158,6 +158,8 @@ def build_config(check: PostgreSql) -> Tuple[InstanceConfig, ValidationResult]:
             },
             "collect_schemas": {
                 **dict_defaults.instance_collect_schemas().model_dump(),
+                "collect_views": True,
+                "max_views": 1000,
                 **(instance.get('collect_schemas', {})),
             },
             "collect_column_statistics": {

postgres/datadog_checks/postgres/config_models/instance.py

@@ -116,6 +116,7 @@ class CollectSchemas(BaseModel):
         arbitrary_types_allowed=True,
         frozen=True,
     )
+    collect_views: Optional[bool] = None
     collection_interval: Optional[float] = None
     enabled: Optional[bool] = None
     exclude_databases: Optional[tuple[str, ...]] = None
@@ -127,6 +128,7 @@ class CollectSchemas(BaseModel):
     max_columns: Optional[float] = None
     max_query_duration: Optional[float] = None
     max_tables: Optional[float] = None
+    max_views: Optional[float] = None
 
 
 class CollectSettings(BaseModel):

postgres/datadog_checks/postgres/data/conf.yaml.example

@@ -515,11 +515,22 @@ instances:
         #
         # enabled: true
 
+        ## @param collect_views - boolean - optional - default: true
+        ## Enable collection of PostgreSQL views and materialized views. Set to false to continue collecting table
+        ## schemas without collecting views. Defaults to true.
+        #
+        # collect_views: true
+
         ## @param max_tables - number - optional - default: 300
         ## Maximum amount of tables the Agent collects from the instance.
         #
         # max_tables: 300
 
+        ## @param max_views - number - optional - default: 1000
+        ## Maximum amount of views and materialized views the Agent collects from each database.
+        #
+        # max_views: 1000
+
         ## @param max_query_duration - number - optional - default: 60
         ## Maximum duration of the query to collect schema information in seconds.
         #

postgres/datadog_checks/postgres/metadata.py

@@ -10,9 +10,12 @@
 import psycopg
 from psycopg.rows import dict_row
 
+from datadog_checks.base.utils.db.schemas import SchemaCollector
+
 from .column_statistics import PostgresColumnStatisticsCollector
 from .schemas import PostgresSchemaCollector
 from .util import collection_interval_gcd
+from .views import PostgresViewCollector
 
 try:
     import datadog_agent  # type: ignore
@@ -121,6 +124,13 @@ def __init__(self, check: PostgreSql, config: InstanceConfig):
         self._collect_extensions_enabled = self._collect_pg_settings_enabled
         self._collect_schemas_enabled = config.collect_schemas.enabled
         self._schema_collector = PostgresSchemaCollector(check) if config.collect_schemas.enabled else None
+        collect_views_requested = config.collect_schemas.enabled and config.collect_schemas.collect_views is not False
+        self._collect_views_enabled = collect_views_requested and hasattr(SchemaCollector, 'object_count_metric_name')
+        self._view_collector = PostgresViewCollector(check) if self._collect_views_enabled else None
+        if collect_views_requested and not self._collect_views_enabled:
+            self._log.warning(
+                "PostgreSQL view collection requires object-specific schema count telemetry; skipping view collection"
+            )
         self._collect_column_statistics_enabled = config.collect_column_statistics.enabled and config.dbm
         self._column_statistics_collector = (
             PostgresColumnStatisticsCollector(check, self._cancel_event)
@@ -140,6 +150,7 @@ def __init__(self, check: PostgreSql, config: InstanceConfig):
     def shutdown(self) -> None:
         self._check = None
         self._schema_collector = None
+        self._view_collector = None
         self._column_statistics_collector = None
         self._compiled_patterns_cache = None
 
@@ -245,6 +256,8 @@ def _collect_postgres_schemas(self):
         if not started:
             # TODO: Emit health event for over-long collection
             self._log.warning("Previous schema collection still in progress, skipping this collection")
+        if self._collect_views_enabled:
+            self._view_collector.collect_schemas()
 
     @tracked_method(agent_check_getter=agent_check_getter)
     def _collect_postgres_settings(self):

postgres/datadog_checks/postgres/views.py

@@ -0,0 +1,140 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, TypedDict
+
+from datadog_checks.base.utils.db.schemas import SchemaCollector
+from datadog_checks.postgres.schemas import DatabaseInfo, DatabaseObject, PostgresSchemaCollector
+
+if TYPE_CHECKING:
+    from datadog_checks.postgres import PostgreSql
+
+
+VIEW_COLUMNS_QUERY = """
+SELECT a.attname                            AS name,
+       format_type(a.atttypid, a.atttypmod) AS data_type,
+       NOT a.attnotnull                     AS nullable,
+       pg_get_expr(ad.adbin, ad.adrelid)    AS default,
+       selected_views.view_id,
+       a.attnum                            AS ordinal_position
+FROM   selected_views
+       INNER JOIN pg_attribute a
+               ON a.attrelid = selected_views.view_id
+       LEFT JOIN pg_attrdef ad
+              ON ad.adrelid = a.attrelid
+                 AND ad.adnum = a.attnum
+WHERE  a.attnum > 0
+       AND NOT a.attisdropped
+"""
+
+
+class ViewColumnObject(TypedDict):
+    data_type: str
+    default: str | None
+    name: str
+    nullable: bool
+
+
+class ViewObject(TypedDict):
+    columns: list[ViewColumnObject]
+    definition: str | None
+    id: str
+    name: str
+    owner: str
+    relkind: str
+
+
+class ViewSchemaObject(TypedDict):
+    id: str
+    name: str
+    owner: str
+    tables: list
+    views: list[ViewObject]
+
+
+class PostgresViewCollector(PostgresSchemaCollector):
+    _check: PostgreSql
+
+    def __init__(self, check: PostgreSql):
+        super().__init__(check)
+        self._max_views = int(check._config.collect_schemas.max_views or 1000)
+
+    @property
+    def kind(self) -> str:
+        return "pg_views"
+
+    @property
+    def object_count_metric_name(self) -> str:
+        return f"dd.{self._check.dbms}.schema.views_count"
+
+    def get_rows_query(self) -> tuple[str, list[str]]:
+        schemas_query, schemas_params = self._get_schemas_query()
+        limit = self._max_views or 1_000_000
+        query = f"""
+            WITH
+            schemas AS (
+                {schemas_query}
+            ),
+            selected_views AS (
+                SELECT schemas.schema_id, schemas.schema_name, schemas.schema_owner,
+                    c.oid AS view_id, c.relname AS view_name,
+                    c.relowner::regrole::text AS view_owner, c.relkind::text AS relkind
+                FROM schemas
+                INNER JOIN pg_class c ON schemas.schema_id = c.relnamespace
+                WHERE c.relkind IN ('v', 'm')
+                ORDER BY schemas.schema_name, c.relname
+                LIMIT {limit}
+            ),
+            views AS (
+                SELECT selected_views.*,
+                    pg_get_viewdef(selected_views.view_id, true) AS definition
+                FROM selected_views
+            ),
+            columns AS (
+                {VIEW_COLUMNS_QUERY}
+            )
+
+            SELECT views.schema_id, views.schema_name, views.schema_owner,
+                views.view_id, views.view_name, views.view_owner,
+                views.relkind, views.definition,
+                array_agg(
+                    json_build_object(
+                        'data_type', columns.data_type,
+                        'default', columns.default,
+                        'name', columns.name,
+                        'nullable', columns.nullable
+                    ) ORDER BY columns.ordinal_position
+                ) FILTER (WHERE columns.name IS NOT NULL) AS columns
+            FROM views
+                LEFT JOIN columns ON views.view_id = columns.view_id
+            GROUP BY views.schema_id, views.schema_name, views.schema_owner,
+                views.view_id, views.view_name, views.view_owner,
+                views.relkind, views.definition
+            ;
+        """
+        return query, schemas_params
+
+    def _map_row(self, database: DatabaseInfo, cursor_row) -> DatabaseObject:
+        object = SchemaCollector._map_row(self, database, cursor_row)
+        object["schemas"] = [
+            {
+                "id": str(cursor_row.get("schema_id")),
+                "name": cursor_row.get("schema_name"),
+                "owner": cursor_row.get("schema_owner"),
+                "tables": [],
+                "views": [
+                    {
+                        "columns": (cursor_row.get("columns") or [])[: self._config.max_columns],
+                        "definition": cursor_row.get("definition"),
+                        "id": str(cursor_row.get("view_id")),
+                        "name": cursor_row.get("view_name"),
+                        "owner": cursor_row.get("view_owner"),
+                        "relkind": cursor_row.get("relkind"),
+                    }
+                ],
+            }
+        ]
+        return object

postgres/tests/compose/resources/02_load_data.sh

@@ -40,6 +40,8 @@ psql -v ON_ERROR_STOP=1 --username "$POSTGRES_USER" datadog_test <<-EOSQL
     SELECT * FROM persons;
     CREATE SCHEMA public2;
     CREATE TABLE public2.cities (city VARCHAR(255), country VARCHAR(255), PRIMARY KEY(city));
+    CREATE VIEW public2.active_persons AS SELECT personid, lastname, city FROM persons;
+    CREATE MATERIALIZED VIEW public2.materialized_persons AS SELECT personid, firstname, city FROM persons;
     GRANT ALL PRIVILEGES ON ALL TABLES IN SCHEMA public TO bob;
     GRANT ALL PRIVILEGES ON ALL TABLES IN SCHEMA public TO blocking_bob;
 EOSQL

postgres/tests/test_config_defaults.py

@@ -127,7 +127,9 @@
     # === DBM: Schema collection ===
     'collect_schemas': {
         'enabled': True,
+        'collect_views': True,
         'max_tables': 300,
+        'max_views': 1000,
         'max_columns': 50,
         'collection_interval': 600,
         'max_query_duration': 60,

postgres/tests/test_metadata.py

@@ -7,6 +7,7 @@
 
 import pytest
 
+from datadog_checks.base.utils.db.schemas import SchemaCollector
 from datadog_checks.base.utils.db.utils import DBMAsyncJob
 
 from .common import POSTGRES_LOCALE, POSTGRES_VERSION
@@ -100,6 +101,45 @@ def test_collect_schema_snapshot(integration_check, dbm_instance, aggregator):
     assert normalize_object(snapshot) == normalize_object(schema_events[0]['metadata'])
 
 
+def test_collect_views(integration_check, dbm_instance, aggregator):
+    dbm_instance["collect_schemas"] = {
+        'enabled': True,
+        'run_sync': True,
+        'include_databases': ['datadog_test'],
+        'include_schemas': ['^public2$'],
+    }
+    dbm_instance['dbname'] = 'datadog_test'
+    check = integration_check(dbm_instance)
+
+    run_one_check(check, dbm_instance)
+
+    view_events = [
+        event for event in aggregator.get_event_platform_events("dbm-metadata") if event['kind'] == 'pg_views'
+    ]
+    if not hasattr(SchemaCollector, 'object_count_metric_name'):
+        assert view_events == []
+        return
+    assert len(view_events) == 1
+    schemas = [database['schemas'][0] for event in view_events for database in event['metadata']]
+    assert all(schema['tables'] == [] for schema in schemas)
+    views = [view for schema in schemas for view in schema['views']]
+    views_by_name = {view['name']: view for view in views}
+
+    assert set(views_by_name) == {'active_persons', 'materialized_persons'}
+    assert views_by_name['active_persons']['relkind'] == 'v'
+    assert views_by_name['materialized_persons']['relkind'] == 'm'
+    for view in (views_by_name['active_persons'], views_by_name['materialized_persons']):
+        assert view['id'].isdigit()
+        assert view['owner'] == 'postgres'
+        assert view['definition']
+        assert [column['name'] for column in view['columns']] in [
+            ['personid', 'lastname', 'city'],
+            ['personid', 'firstname', 'city'],
+        ]
+        for column in view['columns']:
+            assert set(column) == {'data_type', 'default', 'name', 'nullable'}
+
+
 @pytest.mark.parametrize(
     "use_default_ignore_schemas_owned_by",
     [