Limit MySQL discovery to the configured database

CHANGELOG.md

@@ -11,6 +11,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - **Common values of declared columns.** The workload bundle now has `distributions.json`. It holds the most common values from `pg_stats` for each column under `database.workload.distributions`. Values ship verbatim unless `hash_values: true`. The capture role needs `SELECT` on each declared column. See [Recording the common values of chosen columns](docs/workload_capture.md#recording-the-common-values-of-chosen-columns).
 
+### Fixed
+
+- **`mysql.database` did not limit discovery to one database.** Schema discovery and schema-based feature detection now cover only the configured database, as [MySQL Setup](docs/mysql.md#target-a-specific-database) describes. An empty value still discovers every user database. Configuration, performance, replication, and security checks remain server-wide.
+
 ## [2.1.0] - 2026-10-05
 
 ### Added

docs/mysql.md

@@ -253,7 +253,9 @@ mysql:
 
 ### Target a Specific Database
 
-Set the `database` field to focus on one database:
+Set the `database` field to limit schema discovery and schema-based feature
+detection to one database. Configuration, performance, replication, and security
+checks still describe the server as a whole:
 
 ```yaml
 engine: mysql

planetscale_discovery/database/mysql_analyzers/feature_analyzer.py

@@ -180,5 +180,9 @@ def _detect_galera(self, variables: Dict[str, str]) -> bool:
         return provider not in ("", "none")
 
     def _user_schema_filter(self, col: str) -> str:
+        database = self.config.get("database")
+        if database:
+            # Hex literal: see MySQLSchemaAnalyzer._user_databases_filter.
+            return f"{col} = 0x{database.encode('utf-8').hex()}"
         quoted = ", ".join(f"'{db}'" for db in SYSTEM_DATABASES)
         return f"{col} NOT IN ({quoted})"

planetscale_discovery/database/mysql_analyzers/schema_analyzer.py

@@ -242,11 +242,21 @@ def _analyze_per_database(self, database_list: List[str]) -> Dict[str, Any]:
         return merged
 
     def _user_databases_filter(self, col: str = "table_schema") -> str:
-        """SQL fragment to exclude system databases."""
+        """Limit metadata queries to the configured database, if any."""
+        database = self.config.get("database")
+        if database:
+            # A hex literal keeps the configured name out of SQL syntax without
+            # relying on the connection's string-escaping mode. Left as a plain
+            # literal, it takes the column's character set in the comparison,
+            # like the quoted system database names below.
+            return f"{col} = 0x{database.encode('utf-8').hex()}"
         quoted = ", ".join(f"'{db}'" for db in SYSTEM_DATABASES)
         return f"{col} NOT IN ({quoted})"
 
     def _get_database_list(self) -> List[str]:
+        database = self.config.get("database")
+        if database:
+            return [database]
         try:
             cursor = self.connection.cursor()
             cursor.execute("SHOW DATABASES")

planetscale_discovery/database/mysql_discovery.py

@@ -179,7 +179,10 @@ def run_analysis(self, modules: list = None) -> Dict[str, Any]:
 
             try:
                 self.logger.info(f"Running {module_name} analysis...")
-                analyzer = available_modules[module_name](self.connection)
+                analyzer = available_modules[module_name](
+                    self.connection,
+                    config={"database": self.connection_params.get("database", "")},
+                )
                 module_results = analyzer.analyze()
                 self.results["analysis_results"][module_name] = module_results
 

tests/unit/test_mysql_discovery.py

@@ -117,6 +117,26 @@ def test_unknown_ssl_mode_falls_back_to_disabled(self, base_params):
         assert captured.get("ssl_disabled") is True
 
 
+class TestDatabaseScope:
+    @pytest.mark.parametrize("database", ["app", ""])
+    def test_run_analysis_passes_database_scope_to_analyzers(
+        self, base_params, database
+    ):
+        base_params["database"] = database
+        discovery = MySQLDiscovery(base_params)
+        discovery.connection = MagicMock()
+
+        with patch(
+            "planetscale_discovery.database.mysql_discovery.MySQLSchemaAnalyzer"
+        ) as analyzer_class:
+            analyzer_class.return_value.analyze.return_value = {}
+            discovery.run_analysis(["schema"])
+
+        analyzer_class.assert_called_once_with(
+            discovery.connection, config={"database": database}
+        )
+
+
 class TestScanForErrors:
     """Verify _scan_for_errors catches partial failures."""
 

tests/unit/test_mysql_feature_analyzer.py

@@ -188,6 +188,12 @@ def test_filter_excludes_all_system_databases(self):
         for system_db in SYSTEM_DATABASES:
             assert f"'{system_db}'" in clause
 
+    def test_filter_uses_configured_database(self):
+        analyzer = MySQLFeatureAnalyzer(MagicMock(), config={"database": "billing"})
+        clause = analyzer._user_schema_filter("table_schema")
+
+        assert clause == "table_schema = 0x62696c6c696e67"
+
 
 class TestAnalyzeTopLevel:
     def test_analyze_returns_technologies_key(self):

tests/unit/test_mysql_schema_analyzer.py

@@ -28,6 +28,23 @@ def analyzer(self, mock_connection):
         connection, _ = mock_connection
         return MySQLSchemaAnalyzer(connection)
 
+    def test_configured_database_limits_catalog_and_table_query(self, mock_connection):
+        connection, cursor = mock_connection
+        database = "app' OR 1=1"
+        analyzer = MySQLSchemaAnalyzer(connection, config={"database": database})
+
+        assert analyzer._get_database_list() == [database]
+        cursor.execute.assert_not_called()
+        with patch.object(analyzer, "execute_query", return_value=[]) as query:
+            analyzer._get_table_analysis()
+
+        sql = query.call_args.args[0]
+        assert f"t.table_schema = 0x{database.encode().hex()}" in sql
+        assert database not in sql
+
+    def test_empty_database_keeps_all_user_databases(self, analyzer):
+        assert analyzer._user_databases_filter().startswith("table_schema NOT IN (")
+
     # ---------------------------------------------------------------
     # analyze() - structure includes new keys
     # ---------------------------------------------------------------