perf(managed-agent): cache parsed statements on the MySQL pools by default (#16414)

Connector/J parses a statement's SQL on every prepareStatement unless its
per-connection parse cache is on, and that parse was the largest share of
the control plane's Java CPU under load. The server and embedded MySQL
pools now add cachePrepStmts=true, prepStmtCacheSize=512 and
prepStmtCacheSqlLimit=8192 as connection properties.

A connection property overrides the URL in Connector/J, so each setting is
added only when neither the URL nor the Druid connection properties name
it, ignoring case, percent-encoding and surrounding spaces. Nothing is
added when either names useServerPrepStmts or useCursorFetch, since the
cache would then hold statements open on the server.

The server pool binds spring.datasource.druid inside its factory, before
the defaults are added: @ConfigurationProperties bound it only after the
factory returned, and the string form of connection-properties replaces
every connection property.

docs/users/managed-agent-configuration-reference.md

@@ -25,7 +25,7 @@ Spring settings the packaged `application.yml` maps to variables, and the listen
 | `server.port` | `SERVER_PORT` | integer | `8080` | The port of the public listener the API and `/actuator/health` are served on. Spring Boot reads it from `SERVER_PORT`. |
 | `spring.datasource.druid.max-active` | `QWEN_MANAGED_AGENT_DB_MAX_POOL_SIZE` | integer | `20` | How many connections the control plane's Druid database pool may hold at most. When all are busy, a request for a connection waits up to 30 seconds and then fails. Ignored under the `managed-local` profile, whose pool is fixed at 20. |
 | `spring.datasource.password` | `SPRING_DATASOURCE_PASSWORD` | string | *(empty)* | The password of `spring.datasource.username`. A secret: keep it in a private environment file such as `server.env`, not on a command line. |
-| `spring.datasource.url` | `SPRING_DATASOURCE_URL` | string | `jdbc:mysql://127.0.0.1:3306/qwen_managed_agent` | The JDBC URL of the MySQL 8 database that holds the control plane's state; Flyway applies the schema migrations to it at startup. The `managed-local` profile replaces it with a private H2 file database and refuses any other URL. |
+| `spring.datasource.url` | `SPRING_DATASOURCE_URL` | string | `jdbc:mysql://127.0.0.1:3306/qwen_managed_agent` | The JDBC URL of the MySQL 8 database that holds the control plane's state; Flyway applies the schema migrations to it at startup. The `managed-local` profile replaces it with a private H2 file database and refuses any other URL. For a `jdbc:mysql:` URL the server turns on Connector/J's cache of parsed statements: it adds `cachePrepStmts=true`, `prepStmtCacheSize=512` and `prepStmtCacheSqlLimit=8192` as connection properties, each only when neither this URL nor the Druid connection properties (`spring.datasource.druid.connect-properties` or `connection-properties`) name it, and none of them when either names `useServerPrepStmts` or `useCursorFetch`, with any value, since both can make Connector/J prepare statements on the server. `cachePrepStmts=false` in the URL turns the cache off. |
 | `spring.datasource.username` | `SPRING_DATASOURCE_USERNAME` | string | `qwen` | The database user the control plane connects as. Besides rights on the control plane's database, it needs `SELECT` on `performance_schema.user_variables_by_thread`, which the preflight `--check` verifies. |
 
 ## `qwen.managed-agent`

docs/users/managed-agent-configuration-reference.zh-CN.md

@@ -25,7 +25,7 @@ mvn -f packages/managed-agent-server/pom.xml -DskipTests clean process-classes &
 | `server.port` | `SERVER_PORT` | 整数 | `8080` | 公共监听端口，API 和 `/actuator/health` 都在该端口上提供。Spring Boot 从 `SERVER_PORT` 读取。 |
 | `spring.datasource.druid.max-active` | `QWEN_MANAGED_AGENT_DB_MAX_POOL_SIZE` | 整数 | `20` | 控制面 Druid 数据库连接池最多可持有的连接数。连接全部占用时，获取连接的请求最多等待 30 秒，随后失败。`managed-local` profile 下不生效，其连接池固定为 20。 |
 | `spring.datasource.password` | `SPRING_DATASOURCE_PASSWORD` | 字符串 | *（空）* | `spring.datasource.username` 的密码。属于机密：应放在 `server.env` 这类私有环境文件中，不要写在命令行上。 |
-| `spring.datasource.url` | `SPRING_DATASOURCE_URL` | 字符串 | `jdbc:mysql://127.0.0.1:3306/qwen_managed_agent` | 存放控制面状态的 MySQL 8 数据库的 JDBC URL；启动时 Flyway 会对其执行 schema 迁移。`managed-local` profile 改用私有 H2 文件数据库，其他 URL 启动时拒绝。 |
+| `spring.datasource.url` | `SPRING_DATASOURCE_URL` | 字符串 | `jdbc:mysql://127.0.0.1:3306/qwen_managed_agent` | 存放控制面状态的 MySQL 8 数据库的 JDBC URL；启动时 Flyway 会对其执行 schema 迁移。`managed-local` profile 改用私有 H2 文件数据库，其他 URL 启动时拒绝。对 `jdbc:mysql:` URL，服务端会开启 Connector/J 的语句解析缓存：以连接属性加上 `cachePrepStmts=true`、`prepStmtCacheSize=512` 和 `prepStmtCacheSqlLimit=8192`，每一项只在该 URL 和 Druid 连接属性（`spring.datasource.druid.connect-properties` 或 `connection-properties`）都没有设置它时才加；任一处提到 `useServerPrepStmts` 或 `useCursorFetch` 时，无论取值如何，一项都不加，因为二者都可能让 Connector/J 在服务端准备语句。在 URL 里写 `cachePrepStmts=false` 即可关闭缓存。 |
 | `spring.datasource.username` | `SPRING_DATASOURCE_USERNAME` | 字符串 | `qwen` | 控制面连接数据库所用的用户。除控制面数据库的权限外，还需要 `performance_schema.user_variables_by_thread` 的 `SELECT` 权限，预检 `--check` 会校验该权限。 |
 
 ## `qwen.managed-agent`

packages/managed-agent-server/configuration-descriptions.json

@@ -821,8 +821,8 @@
       "type": "string"
     },
     "spring.datasource.url": {
-      "en": "The JDBC URL of the MySQL 8 database that holds the control plane's state; Flyway applies the schema migrations to it at startup. The `managed-local` profile replaces it with a private H2 file database and refuses any other URL.",
-      "zh": "存放控制面状态的 MySQL 8 数据库的 JDBC URL；启动时 Flyway 会对其执行 schema 迁移。`managed-local` profile 改用私有 H2 文件数据库，其他 URL 启动时拒绝。",
+      "en": "The JDBC URL of the MySQL 8 database that holds the control plane's state; Flyway applies the schema migrations to it at startup. The `managed-local` profile replaces it with a private H2 file database and refuses any other URL. For a `jdbc:mysql:` URL the server turns on Connector/J's cache of parsed statements: it adds `cachePrepStmts=true`, `prepStmtCacheSize=512` and `prepStmtCacheSqlLimit=8192` as connection properties, each only when neither this URL nor the Druid connection properties (`spring.datasource.druid.connect-properties` or `connection-properties`) name it, and none of them when either names `useServerPrepStmts` or `useCursorFetch`, with any value, since both can make Connector/J prepare statements on the server. `cachePrepStmts=false` in the URL turns the cache off.",
+      "zh": "存放控制面状态的 MySQL 8 数据库的 JDBC URL；启动时 Flyway 会对其执行 schema 迁移。`managed-local` profile 改用私有 H2 文件数据库，其他 URL 启动时拒绝。对 `jdbc:mysql:` URL，服务端会开启 Connector/J 的语句解析缓存：以连接属性加上 `cachePrepStmts=true`、`prepStmtCacheSize=512` 和 `prepStmtCacheSqlLimit=8192`，每一项只在该 URL 和 Druid 连接属性（`spring.datasource.druid.connect-properties` 或 `connection-properties`）都没有设置它时才加；任一处提到 `useServerPrepStmts` 或 `useCursorFetch` 时，无论取值如何，一项都不加，因为二者都可能让 Connector/J 在服务端准备语句。在 URL 里写 `cachePrepStmts=false` 即可关闭缓存。",
       "type": "string"
     },
     "spring.datasource.username": {

packages/managed-agent-server/server/pom.xml

@@ -163,9 +163,10 @@
                 <executions>
                     <!-- Writes META-INF/spring-configuration-metadata.json, from
                          which docs/users/managed-agent-configuration-reference.md
-                         is generated. Only ManagedAgentProperties: the two Druid
-                         beans share a prefix, which the processor refuses. It
-                         reads Lombok's annotations itself. -->
+                         is generated. Only ManagedAgentProperties: the generator
+                         documents every property in this metadata, and the Druid
+                         pools bound to spring.datasource.druid would add hundreds.
+                         It reads Lombok's annotations itself. -->
                     <execution>
                         <id>configuration-metadata</id>
                         <phase>process-classes</phase>

packages/managed-agent-server/server/src/main/java/com/alibaba/qwen/code/managedagent/ManagedAgentServerApplication.java

@@ -4,13 +4,13 @@ import com.alibaba.druid.pool.DruidDataSource;
 import com.alibaba.qwen.code.managedagent.config.ManagedAgentConfiguration;
 import com.alibaba.qwen.code.managedagent.config.ManagedAgentWebConfiguration;
 import com.alibaba.qwen.code.managedagent.config.ManagedLocalConfiguration;
+import com.alibaba.qwen.code.managedagent.config.MySqlStatementCache;
 import com.alibaba.qwen.code.managedagent.preflight.ManagedAgentPreflight;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.SpringBootConfiguration;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.boot.autoconfigure.flyway.FlywayMigrationStrategy;
 import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
-import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Import;
 import org.springframework.context.annotation.Profile;
@@ -30,14 +30,18 @@ public class ManagedAgentServerApplication {
 
     @Bean(name = "managedAgentDataSource", destroyMethod = "close")
     @Profile("!managed-local")
-    @ConfigurationProperties("spring.datasource.druid")
     DruidDataSource standaloneDataSource(org.springframework.core.env.Environment environment) {
         var source = new DruidDataSource();
         source.setMaxActive(20);
         source.setMaxWait(30_000);
         source.setValidationQuery("SELECT 1");
-        org.springframework.boot.context.properties.bind.Binder.get(environment).bind(
-                "spring.datasource", org.springframework.boot.context.properties.bind.Bindable.ofInstance(source));
+        var binder = org.springframework.boot.context.properties.bind.Binder.get(environment);
+        binder.bind("spring.datasource", org.springframework.boot.context.properties.bind.Bindable.ofInstance(source));
+        // Bound here, not through @ConfigurationProperties, which binds only after
+        // this method returns: the statement cache defaults must see the final URL
+        // and connection properties.
+        binder.bind("spring.datasource.druid", org.springframework.boot.context.properties.bind.Bindable.ofInstance(source));
+        MySqlStatementCache.fillDefaults(source);
         return source;
     }
 

packages/managed-agent-server/server/src/main/java/com/alibaba/qwen/code/managedagent/config/ManagedAgentPersistenceConfiguration.java

@@ -39,6 +39,7 @@ public class ManagedAgentPersistenceConfiguration {
             source.setConnectionInitSqls(List.of("SET SESSION sql_mode ="
                     + " CONCAT_WS(',', NULLIF(@@SESSION.sql_mode, ''), 'STRICT_TRANS_TABLES')"));
         }
+        MySqlStatementCache.fillDefaults(source);
         return source;
     }
 

packages/managed-agent-server/server/src/main/java/com/alibaba/qwen/code/managedagent/config/MySqlStatementCache.java

@@ -0,0 +1,95 @@
+package com.alibaba.qwen.code.managedagent.config;
+
+import com.alibaba.druid.pool.DruidDataSource;
+
+import java.net.URLDecoder;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Properties;
+
+/**
+ * Turns on Connector/J's per-connection cache of parsed SQL for a MySQL pool.
+ * Without it the driver parses a statement's SQL again on every
+ * {@code prepareStatement}, which was the largest single share of the control
+ * plane's Java CPU under load.
+ *
+ * <p>Each setting is added only when neither the URL nor the pool's connection
+ * properties name it, because Connector/J lets a connection property override
+ * the URL. When either names {@code useServerPrepStmts}, or {@code useCursorFetch},
+ * which turns server-side statements on, nothing is added: with server-side
+ * statements the cache would keep statements open on the server, counted
+ * against its global {@code max_prepared_stmt_count}.
+ */
+public final class MySqlStatementCache {
+    /**
+     * In a Hosted Shell run the control plane prepared 270 distinct statements,
+     * 184 to 265 on each connection; apart from one metadata query of the
+     * driver's own, the longest was 2,023 characters. Connector/J's own sizes
+     * would hold 25 per connection and skip the 104 longer than 256 characters.
+     */
+    static final Map<String, String> DEFAULTS = defaults();
+
+    private static final List<String> SERVER_STATEMENTS = List.of("useServerPrepStmts", "useCursorFetch");
+
+    private MySqlStatementCache() {
+    }
+
+    /** Adds the settings this pool leaves unset; call it before the pool opens. */
+    public static void fillDefaults(DruidDataSource source) {
+        String url = source.getUrl();
+        if (url == null || !(url.startsWith("jdbc:mysql:") || url.startsWith("jdbc:mysql+srv:"))) {
+            return;
+        }
+        Properties properties = source.getConnectProperties();
+        if (SERVER_STATEMENTS.stream().anyMatch(key -> named(url, properties, key))) {
+            return;
+        }
+        DEFAULTS.forEach((key, value) -> {
+            if (!named(url, properties, key)) {
+                source.addConnectionProperty(key, value);
+            }
+        });
+    }
+
+    /**
+     * Whether the URL or the properties name the key, ignoring case as
+     * Connector/J does. Any mention in the URL counts, percent-encoded or not,
+     * in its query or in a host's own properties, so a setting is never added
+     * over one the URL holds. Druid keeps the spaces around a name in
+     * {@code connection-properties}, so they are ignored here too.
+     */
+    private static boolean named(String url, Properties properties, String key) {
+        String lower = key.toLowerCase(Locale.ROOT);
+        return url.toLowerCase(Locale.ROOT).contains(lower)
+                || decoded(url).toLowerCase(Locale.ROOT).contains(lower)
+                || properties.stringPropertyNames().stream()
+                        .anyMatch(name -> name.strip().toLowerCase(Locale.ROOT).equals(lower));
+    }
+
+    /**
+     * The URL with its percent escapes decoded, up to a fragment, which
+     * Connector/J neither decodes nor reads; as it is when an escape is
+     * malformed, which Connector/J refuses anyway.
+     */
+    private static String decoded(String url) {
+        int fragment = url.indexOf('#');
+        String read = fragment < 0 ? url : url.substring(0, fragment);
+        try {
+            return URLDecoder.decode(read.replace("+", "%2B"), StandardCharsets.UTF_8);
+        } catch (IllegalArgumentException malformed) {
+            return url;
+        }
+    }
+
+    private static Map<String, String> defaults() {
+        var defaults = new LinkedHashMap<String, String>();
+        defaults.put("cachePrepStmts", "true");
+        defaults.put("prepStmtCacheSize", "512");
+        defaults.put("prepStmtCacheSqlLimit", "8192");
+        return Collections.unmodifiableMap(defaults);
+    }
+}

packages/managed-agent-server/verification/src/test/java/com/alibaba/qwen/code/managedagent/StandaloneDataSourceConfigurationTest.java

@@ -4,7 +4,12 @@ import com.alibaba.druid.pool.DruidDataSource;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.boot.context.properties.ConfigurationPropertiesBean;
 import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.ApplicationContext;
+import org.springframework.mock.env.MockEnvironment;
+
+import java.util.Map;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -24,6 +29,9 @@ class StandaloneDataSourceConfigurationTest {
     @Qualifier("managedAgentDataSource")
     DruidDataSource source;
 
+    @Autowired
+    ApplicationContext context;
+
     @Test
     void standaloneBeanBindsCommonAndDruidSettingsBeforeOpeningThePool() {
         assertThat(source.getUrl()).startsWith("jdbc:h2:mem:standalone-pool-binding;");
@@ -35,4 +43,24 @@ class StandaloneDataSourceConfigurationTest {
         assertThat(source.getConnectProperties().getProperty("socketTimeout")).isEqualTo("10000");
         assertThat(source.isInited()).isTrue();
     }
+
+    @Test
+    void standaloneBeanFillsStatementCacheDefaultsAfterItsDruidSettings() {
+        // The string form replaces every connection property, so a default added
+        // before the Druid settings were bound would be gone.
+        try (DruidDataSource mysql = new ManagedAgentServerApplication().standaloneDataSource(new MockEnvironment()
+                .withProperty("spring.datasource.url", "jdbc:mysql://db/qwen?prepStmtCacheSqlLimit=4096")
+                .withProperty("spring.datasource.druid.max-active", "3")
+                .withProperty("spring.datasource.druid.connection-properties", "prepStmtCacheSize=64;connectTimeout=5000"))) {
+            assertThat(mysql.getMaxActive()).isEqualTo(3);
+            assertThat(mysql.getUrl()).isEqualTo("jdbc:mysql://db/qwen?prepStmtCacheSqlLimit=4096");
+            assertThat(Map.copyOf(mysql.getConnectProperties())).isEqualTo(Map.of(
+                    "prepStmtCacheSize", "64", "connectTimeout", "5000", "cachePrepStmts", "true"));
+            assertThat(mysql.isInited()).isFalse();
+        }
+        assertThat(source.getConnectProperties()).doesNotContainKey("cachePrepStmts");
+        // Spring would bind a @ConfigurationProperties bean again after its factory
+        // returned, and the string form would drop the defaults.
+        assertThat(ConfigurationPropertiesBean.getAll(context)).doesNotContainKey("managedAgentDataSource");
+    }
 }

packages/managed-agent-server/verification/src/test/java/com/alibaba/qwen/code/managedagent/config/MySqlStatementCacheMySqlIT.java

@@ -0,0 +1,107 @@
+package com.alibaba.qwen.code.managedagent.config;
+
+import com.alibaba.druid.pool.DruidDataSource;
+import com.alibaba.qwen.code.managedagent.ManagedAgentServerApplication;
+import com.alibaba.qwen.code.managedagent.persistence.IsolatedManagedSchemaMySqlTestDatabase;
+import com.mysql.cj.conf.PropertyKey;
+import com.mysql.cj.conf.PropertySet;
+import com.mysql.cj.jdbc.ClientPreparedStatement;
+import com.mysql.cj.jdbc.JdbcConnection;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.mock.env.MockEnvironment;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * The pools reach Connector/J with its parse cache on, so a statement prepared
+ * again on one connection reuses the first parse, and a URL that turns the
+ * cache off keeps it off.
+ */
+class MySqlStatementCacheMySqlIT {
+    @Test
+    void theServerPoolCachesParsesAfterBindingItsDruidSettings() throws Exception {
+        try (var database = new IsolatedManagedSchemaMySqlTestDatabase();
+                var context = server(database.url(),
+                        "--spring.datasource.druid.connection-properties=prepStmtCacheSize=64;connectTimeout=5000");
+                Connection connection = context.getBean("managedAgentDataSource", DruidDataSource.class).getConnection()) {
+            PropertySet properties = properties(connection);
+            assertThat(properties.getBooleanProperty(PropertyKey.cachePrepStmts).getValue()).isTrue();
+            assertThat(properties.getIntegerProperty(PropertyKey.prepStmtCacheSize).getValue()).isEqualTo(64);
+            assertThat(properties.getIntegerProperty(PropertyKey.prepStmtCacheSqlLimit).getValue()).isEqualTo(8192);
+            assertThat(properties.getIntegerProperty(PropertyKey.connectTimeout).getValue()).isEqualTo(5000);
+            assertThat(reusesParse(connection)).isTrue();
+        }
+    }
+
+    @Test
+    void theServerPoolKeepsACacheTheUrlTurnsOff() throws Exception {
+        try (var database = new IsolatedManagedSchemaMySqlTestDatabase();
+                var context = server(withParameter(database.url(), "cachePrepStmts=false"));
+                Connection connection = context.getBean("managedAgentDataSource", DruidDataSource.class).getConnection()) {
+            assertThat(properties(connection).getBooleanProperty(PropertyKey.cachePrepStmts).getValue()).isFalse();
+            assertThat(reusesParse(connection)).isFalse();
+        }
+    }
+
+    @Test
+    void theEmbeddedPoolCachesParsesUnlessStatementsArePreparedOnTheServer() throws Exception {
+        try (var database = new IsolatedManagedSchemaMySqlTestDatabase()) {
+            try (DruidDataSource source = embedded(database.url()); Connection connection = source.getConnection()) {
+                PropertySet properties = properties(connection);
+                assertThat(properties.getBooleanProperty(PropertyKey.cachePrepStmts).getValue()).isTrue();
+                assertThat(properties.getIntegerProperty(PropertyKey.prepStmtCacheSize).getValue()).isEqualTo(512);
+                assertThat(reusesParse(connection)).isTrue();
+            }
+            try (DruidDataSource source = embedded(withParameter(database.url(), "useServerPrepStmts=true"));
+                    Connection connection = source.getConnection()) {
+                assertThat(properties(connection).getBooleanProperty(PropertyKey.cachePrepStmts).getValue()).isFalse();
+            }
+        }
+    }
+
+    private static ConfigurableApplicationContext server(String url, String... more) {
+        var args = new ArrayList<>(List.of(
+                "--server.address=127.0.0.1", "--server.port=0",
+                "--spring.datasource.url=" + url,
+                "--spring.datasource.username=" + System.getProperty("mysql.user"),
+                "--spring.datasource.password=" + System.getProperty("mysql.password", ""),
+                "--spring.datasource.druid.max-active=2",
+                "--spring.datasource.druid.max-wait=5000",
+                "--qwen.managed-agent.harness.enabled=false",
+                "--qwen.managed-agent.runtime-broker.enabled=false"));
+        args.addAll(List.of(more));
+        return new SpringApplicationBuilder(ManagedAgentServerApplication.class).run(args.toArray(String[]::new));
+    }
+
+    private static DruidDataSource embedded(String url) {
+        return new ManagedAgentPersistenceConfiguration().managedAgentDataSource(new MockEnvironment()
+                .withProperty("qwen.managed-agent.datasource.url", url)
+                .withProperty("qwen.managed-agent.datasource.username", System.getProperty("mysql.user"))
+                .withProperty("qwen.managed-agent.datasource.password", System.getProperty("mysql.password", ""))
+                .withProperty("qwen.managed-agent.datasource.max-active", "1"));
+    }
+
+    private static String withParameter(String url, String parameter) {
+        return url + (url.contains("?") ? "&" : "?") + parameter;
+    }
+
+    private static PropertySet properties(Connection connection) throws SQLException {
+        return connection.unwrap(JdbcConnection.class).getPropertySet();
+    }
+
+    /** Whether preparing the same SQL twice on the connection parses it once. */
+    private static boolean reusesParse(Connection connection) throws SQLException {
+        String sql = "SELECT ? FROM DUAL";
+        try (var first = connection.prepareStatement(sql); var second = connection.prepareStatement(sql)) {
+            return first.unwrap(ClientPreparedStatement.class).getQueryInfo()
+                    == second.unwrap(ClientPreparedStatement.class).getQueryInfo();
+        }
+    }
+}

packages/managed-agent-server/verification/src/test/java/com/alibaba/qwen/code/managedagent/config/MySqlStatementCacheTest.java

@@ -0,0 +1,94 @@
+package com.alibaba.qwen.code.managedagent.config;
+
+import com.alibaba.druid.pool.DruidDataSource;
+import org.junit.jupiter.api.Test;
+import org.springframework.mock.env.MockEnvironment;
+
+import java.util.Map;
+import java.util.Properties;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class MySqlStatementCacheTest {
+    private static final Map<String, String> ALL = Map.of(
+            "cachePrepStmts", "true", "prepStmtCacheSize", "512", "prepStmtCacheSqlLimit", "8192");
+
+    @Test
+    void aMySqlPoolGetsEveryDefault() {
+        assertThat(filled("jdbc:mysql://db:3306/qwen_managed_agent", new Properties())).isEqualTo(ALL);
+        assertThat(filled("jdbc:mysql://db:3306/qwen_managed_agent?useSSL=false", new Properties())).isEqualTo(ALL);
+        assertThat(filled("jdbc:mysql:loadbalance://a,b/qwen_managed_agent", new Properties())).isEqualTo(ALL);
+        assertThat(filled("jdbc:mysql+srv://cluster/qwen_managed_agent", new Properties())).isEqualTo(ALL);
+    }
+
+    @Test
+    void aSettingTheUrlNamesKeepsItsValueInAnyCaseOrPlace() {
+        assertThat(filled("jdbc:mysql://db/qwen?CachePrepStmts=false", new Properties())).isEqualTo(Map.of(
+                "prepStmtCacheSize", "512", "prepStmtCacheSqlLimit", "8192"));
+        assertThat(filled("jdbc:mysql://address=(host=db)(port=3306)(prepstmtcachesize=64)/qwen", new Properties()))
+                .isEqualTo(Map.of("cachePrepStmts", "true", "prepStmtCacheSqlLimit", "8192"));
+        // Connector/J decodes percent escapes in names before it reads them.
+        assertThat(filled("jdbc:mysql://db/qwen?%63achePrepStmts=false", new Properties())).isEqualTo(Map.of(
+                "prepStmtCacheSize", "512", "prepStmtCacheSqlLimit", "8192"));
+        // A fragment is not decoded, so a stray escape in it does not hide the name.
+        assertThat(filled("jdbc:mysql://db/qwen?prepStmt%43acheSize=16#50%", new Properties())).isEqualTo(Map.of(
+                "cachePrepStmts", "true", "prepStmtCacheSqlLimit", "8192"));
+    }
+
+    @Test
+    void aSettingTheConnectionPropertiesNameKeepsItsValue() {
+        var properties = new Properties();
+        properties.setProperty("connectTimeout", "5000");
+        properties.setProperty("PREPSTMTCACHESQLLIMIT", "2048");
+        assertThat(filled("jdbc:mysql://db/qwen", properties)).isEqualTo(Map.of(
+                "connectTimeout", "5000", "PREPSTMTCACHESQLLIMIT", "2048",
+                "cachePrepStmts", "true", "prepStmtCacheSize", "512"));
+        // Druid keeps the space after a separator in the name.
+        var source = new DruidDataSource();
+        source.setUrl("jdbc:mysql://db/qwen");
+        source.setConnectionProperties("connectTimeout=5000; cachePrepStmts=false");
+        MySqlStatementCache.fillDefaults(source);
+        assertThat(Map.copyOf(source.getConnectProperties())).isEqualTo(Map.of(
+                "connectTimeout", "5000", " cachePrepStmts", "false",
+                "prepStmtCacheSize", "512", "prepStmtCacheSqlLimit", "8192"));
+    }
+
+    @Test
+    void serverSideStatementsInEitherPlaceGetNoDefaults() {
+        assertThat(filled("jdbc:mysql://db/qwen?useServerPrepStmts=true", new Properties())).isEmpty();
+        var properties = new Properties();
+        properties.setProperty("useServerPrepStmts", "false");
+        assertThat(filled("jdbc:mysql://db/qwen", properties)).isEqualTo(Map.of("useServerPrepStmts", "false"));
+        // A cursor fetch turns server-side statements on.
+        assertThat(filled("jdbc:mysql://db/qwen?useCursorFetch=true", new Properties())).isEmpty();
+        var cursor = new Properties();
+        cursor.setProperty("useCursorFetch", "true");
+        assertThat(filled("jdbc:mysql://db/qwen", cursor)).isEqualTo(Map.of("useCursorFetch", "true"));
+    }
+
+    @Test
+    void anotherDriverGetsNoDefaults() {
+        assertThat(filled("jdbc:h2:mem:qwen;MODE=MySQL", new Properties())).isEmpty();
+        assertThat(filled("jdbc:mariadb://db/qwen", new Properties())).isEmpty();
+        var source = new DruidDataSource();
+        MySqlStatementCache.fillDefaults(source);
+        assertThat(source.getConnectProperties()).isEmpty();
+    }
+
+    @Test
+    void theEmbeddedPoolGetsTheDefaults() {
+        try (DruidDataSource source = new ManagedAgentPersistenceConfiguration().managedAgentDataSource(
+                new MockEnvironment().withProperty("qwen.managed-agent.datasource.url", "jdbc:mysql://db/qwen"))) {
+            assertThat(Map.copyOf(source.getConnectProperties())).isEqualTo(ALL);
+            assertThat(source.isInited()).isFalse();
+        }
+    }
+
+    private static Map<Object, Object> filled(String url, Properties properties) {
+        var source = new DruidDataSource();
+        source.setUrl(url);
+        source.setConnectProperties(properties);
+        MySqlStatementCache.fillDefaults(source);
+        return Map.copyOf(source.getConnectProperties());
+    }
+}