[SDBM-2648] Add collect_execution_plans option to SQL Server DBM (DataDog#23961)

* [SDBM-2648] Add collect_execution_plans option to SQL Server DBM

Adds a new `collect_execution_plans.enabled` config flag (default: true)
that lets users disable execution plan collection independently of query
metrics. When false, the query-metrics job still runs and emits metrics
and FQT events, but skips the _collect_plans loop.

SDBM-2648

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* Add changelog entry for PR DataDog#23961

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* Move collect_plans option under query_metrics

Plans are collected by the query-metrics job and are already disabled
when query_metrics.enabled is false, so the toggle belongs there.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* Update changelog entry to reflect query_metrics.collect_plans

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* Run linting

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-authored-by: Eric Weaver <eweaver755@gmail.com>
Co-authored-by: Eric Weaver <eric.weaver@datadoghq.com>

Author: 3 people

sqlserver/assets/configuration/spec.yaml

@@ -669,6 +669,15 @@ files:
             display_default: 2 * collection_interval
             example: 600
           fleet_configurable: false
+        - name: collect_plans
+          description: |
+            Set to `false` to disable collection of query execution plans while keeping query
+            metrics collection running. Note that when `enabled` is set to `false`, plan
+            collection is also disabled regardless of this setting.
+          value:
+            type: boolean
+            example: true
+          fleet_configurable: false
     - name: procedure_metrics
       description: Configure collection of procedure metrics
       options:

sqlserver/changelog.d/23961.added

@@ -0,0 +1 @@
+Add ``query_metrics.collect_plans`` option to disable SQL Server execution plan collection independently of query metrics.

sqlserver/datadog_checks/sqlserver/config_models/instance.py

@@ -389,6 +389,7 @@ class QueryMetrics(BaseModel):
         arbitrary_types_allowed=True,
         frozen=True,
     )
+    collect_plans: Optional[bool] = None
     collection_interval: Optional[float] = None
     disable_secondary_tags: Optional[bool] = None
     dm_exec_query_stats_row_limit: Optional[int] = None

sqlserver/datadog_checks/sqlserver/data/conf.yaml.example

@@ -420,6 +420,13 @@ instances:
         #
         # lookback_window: 600
 
+        ## @param collect_plans - boolean - optional - default: true
+        ## Set to `false` to disable collection of query execution plans while keeping query
+        ## metrics collection running. Note that when `enabled` is set to `false`, plan
+        ## collection is also disabled regardless of this setting.
+        #
+        # collect_plans: true
+
     ## Configure collection of procedure metrics
     #
     # procedure_metrics:

sqlserver/datadog_checks/sqlserver/statements.py

@@ -291,6 +291,7 @@ def __init__(self, check, config: SQLServerConfig):
         self._last_stats_query_time = None
         self._max_query_metrics = self._config.statement_metrics_config.get("max_queries", 250)
 
+        self._collect_execution_plans = is_affirmative(self._config.statement_metrics_config.get('collect_plans', True))
         self._collect_raw_query_statement = self._config.collect_raw_query_statement.get("enabled", False)
 
     def _init_caches(self):
@@ -541,9 +542,12 @@ def collect_statement_metrics_and_plans(self):
                     self._check.database_monitoring_query_sample(json.dumps(event, default=default_json_event_encoding))
                 payload = self._to_metrics_payload(rows, self._max_query_metrics)
                 self._check.database_monitoring_query_metrics(json.dumps(payload, default=default_json_event_encoding))
-                for event in self._collect_plans(rows, cursor, deadline):
-                    self._check.database_monitoring_query_sample(json.dumps(event, default=default_json_event_encoding))
-                    plans_submitted += 1
+                if self._collect_execution_plans:
+                    for event in self._collect_plans(rows, cursor, deadline):
+                        self._check.database_monitoring_query_sample(
+                            json.dumps(event, default=default_json_event_encoding)
+                        )
+                        plans_submitted += 1
 
         self._check.count(
             "dd.sqlserver.statements.plans_submitted.count", plans_submitted, **self._check.debug_stats_kwargs()

sqlserver/tests/test_statements.py

@@ -1296,3 +1296,85 @@ def _execute_queries():
         database_names = {row['database_name'] for row in sqlserver_rows}
         assert 'datadog_test-1' in database_names, "should have collected metrics for datadog_test-1 databases"
         assert 'master' in database_names, "should have collected metrics for master databases"
+
+
+@pytest.mark.unit
+@pytest.mark.parametrize(
+    "collect_plans_value,expect_plans",
+    [
+        pytest.param(None, True, id="default_collects_plans"),
+        pytest.param(True, True, id="explicitly_enabled"),
+        pytest.param(False, False, id="disabled_skips_plans"),
+    ],
+)
+def test_collect_execution_plans_toggle(instance_docker, collect_plans_value, expect_plans):
+    """query_metrics.collect_plans=false skips plan loop while keeping metrics."""
+    instance_docker['dbm'] = True
+    query_metrics = {
+        'enabled': True,
+        'run_sync': True,
+        'collection_interval': 0.1,
+        'enforce_collection_interval_deadline': False,
+    }
+    if collect_plans_value is not None:
+        query_metrics['collect_plans'] = collect_plans_value
+    instance_docker['query_metrics'] = query_metrics
+
+    check = SQLServer(CHECK_NAME, {}, [instance_docker])
+
+    fake_row = {
+        'query_signature': 'abc123',
+        'query_hash': '0xDEAD',
+        'query_plan_hash': '0xBEEF',
+        'plan_handle': '0000',
+        'text': 'SELECT 1',
+        'dd_tables': [],
+        'dd_commands': [],
+        'dd_comments': None,
+        'database_name': 'master',
+        'is_proc': False,
+        'is_encrypted': False,
+        'procedure_signature': None,
+        'procedure_name': None,
+    }
+
+    fake_plan_event = {
+        'dbm_type': 'plan',
+        'db': {'query_signature': 'abc123', 'plan': {'definition': '<ShowPlanXML/>'}},
+    }
+
+    def _mock_collect_plans(*_args, **_kwargs):
+        yield fake_plan_event
+
+    with (
+        mock.patch.object(
+            check.statement_metrics,
+            '_collect_metrics_rows',
+            return_value=[fake_row],
+        ),
+        mock.patch.object(
+            check.statement_metrics,
+            '_collect_plans',
+            side_effect=_mock_collect_plans,
+        ) as mock_collect_plans,
+        mock.patch.object(
+            check.statement_metrics,
+            '_rows_to_fqt_events',
+            return_value=[],
+        ),
+        mock.patch.object(
+            check.statement_metrics,
+            '_to_metrics_payload',
+            return_value={'sqlserver_rows': [fake_row]},
+        ),
+        mock.patch.object(check, 'database_monitoring_query_metrics'),
+        mock.patch.object(check, 'database_monitoring_query_sample'),
+        mock.patch.object(check.connection, 'open_managed_default_connection'),
+        mock.patch.object(check.connection, 'get_managed_cursor'),
+    ):
+        check.statement_metrics.collect_statement_metrics_and_plans()
+
+    if expect_plans:
+        mock_collect_plans.assert_called_once()
+    else:
+        mock_collect_plans.assert_not_called()