diff --git a/mysql/changelog.d/21097.fixed b/mysql/changelog.d/21097.fixed
new file mode 100644
index 0000000000000..2046fd6863c92
--- /dev/null
+++ b/mysql/changelog.d/21097.fixed
@@ -0,0 +1 @@
+Change statement_rows to TTL cache to fix mysql memory leaks
diff --git a/mysql/datadog_checks/mysql/config.py b/mysql/datadog_checks/mysql/config.py
index 2dca99c2688e2..2997022e0aa69 100644
--- a/mysql/datadog_checks/mysql/config.py
+++ b/mysql/datadog_checks/mysql/config.py
@@ -45,6 +45,8 @@ def __init__(self, instance, init_config):
         self.full_statement_text_samples_per_hour_per_query = instance.get(
             'full_statement_text_samples_per_hour_per_query', 1
         )
+        self.statement_rows_cache_max_size = instance.get('statement_rows_cache_max_size', 10000)
+        self.statement_rows_cache_ttl = instance.get('statement_rows_cache_ttl', 3600)
         self.statement_samples_config = instance.get('query_samples', instance.get('statement_samples', {})) or {}
         self.statement_metrics_config = instance.get('query_metrics', {}) or {}
         self.settings_config = instance.get('collect_settings', {}) or {}
diff --git a/mysql/datadog_checks/mysql/statements.py b/mysql/datadog_checks/mysql/statements.py
index 365bf681e53e9..f158ae337dc22 100644
--- a/mysql/datadog_checks/mysql/statements.py
+++ b/mysql/datadog_checks/mysql/statements.py
@@ -96,7 +96,10 @@ def __init__(self, check, config, connection_args):
 
         # statement_rows: cache of all rows for each digest, keyed by (schema_name, query_signature)
         # This is used to cache the metrics for queries that have the same query_signature but different digests
-        self._statement_rows = {}  # type: Dict[(str, str), Dict[str, PyMysqlRow]]
+        self._statement_rows = TTLCache(
+            maxsize=self._config.statement_rows_cache_max_size,
+            ttl=self._config.statement_rows_cache_ttl,
+        )
 
     def _get_db_connection(self):
         """
@@ -119,6 +122,7 @@ def _close_db_conn(self):
 
     def run_job(self):
         start = time.time()
+        self._statement_rows.expire()
         self.collect_per_statement_metrics()
         self._check.gauge(
             "dd.mysql.statement_metrics.collect_metrics.elapsed_ms",
@@ -300,9 +304,9 @@ def _add_associated_rows(self, rows):
         """
         for row in rows:
             key = (row['schema_name'], row['query_signature'])
-            if key not in self._statement_rows:
-                self._statement_rows[key] = {}
-            self._statement_rows[key][row['digest']] = row
+            digest_rows = self._statement_rows.get(key, {})
+            digest_rows[row['digest']] = row
+            self._statement_rows[key] = digest_rows
 
         return [row for statement_row in self._statement_rows.values() for row in statement_row.values()]