[Draft] Remote query cache plugin

.env.example

@@ -0,0 +1,5 @@
+DB_CONNECTION_STRING=jdbc:aws-wrapper:postgresql://localhost:5432/dbname
+CACHE_RW_SERVER_ADDR=localhost:6379
+CACHE_RO_SERVER_ADDR=localhost:6380
+DB_USERNAME=postgres
+DB_PASSWORD=admin

benchmarks/build.gradle.kts

@@ -25,6 +25,10 @@ dependencies {
     implementation("org.mariadb.jdbc:mariadb-java-client:3.5.6")
     implementation("com.zaxxer:HikariCP:4.0.3")
     implementation("org.checkerframework:checker-qual:3.49.5")
+    implementation("io.lettuce:lettuce-core:6.6.0.RELEASE")
+    implementation("org.apache.commons:commons-pool2:2.11.1")
+    annotationProcessor("org.openjdk.jmh:jmh-core:1.36")
+    jmhAnnotationProcessor ("org.openjdk.jmh:jmh-generator-annprocess:1.36")
 
     testImplementation("org.junit.jupiter:junit-jupiter-api:5.12.2")
     testImplementation("org.mockito:mockito-inline:4.11.0") // 4.11.0 is the last version compatible with Java 8

benchmarks/src/jmh/java/software/amazon/jdbc/benchmarks/PgCacheBenchmarks.java

@@ -0,0 +1,144 @@
+package software.amazon.jdbc.benchmarks;
+
+import org.openjdk.jmh.annotations.*;
+import org.openjdk.jmh.infra.Blackhole;
+import java.sql.*;
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+
+import org.openjdk.jmh.profile.GCProfiler;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.RunnerException;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+
+/**
+ * Performance benchmark program against PG.
+ *
+ * This test program runs JMH benchmark tests the performance of the remote cache plugin against a
+ * a remote PG database and a remote cache server for both indexed queries and non-indexed queries.
+ *
+ * The database table schema is as follows:
+ *
+ * postgres=# CREATE TABLE test (id SERIAL PRIMARY KEY, int_col INTEGER, varchar_col varchar(50) NOT NULL, text_col TEXT,
+ *    num_col DOUBLE PRECISION, date_col date, time_col TIME WITHOUT TIME ZONE, time_tz TIME WITH TIME ZONE,
+ *    ts_col TIMESTAMP WITHOUT TIME ZONE, ts_tz TIMESTAMP WITH TIME ZONE, description TEXT);
+ * CREATE TABLE
+ * postgres=# select * from test;
+ *  id | int_col | varchar_col | text_col | num_col | date_col | time_col | time_tz | ts_col | ts_tz | description
+ * ----+---------+-------------+----------+---------+----------+----------+---------+--------+-------+--------------
+ * (0 rows)
+ *
+ */
+@State(Scope.Thread)
+@Fork(1)
+@Warmup(iterations = 1)
+@Measurement(iterations = 60, time = 1)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+public class PgCacheBenchmarks {
+  private static final String DB_CONNECTION_STRING = "jdbc:aws-wrapper:postgresql://db-0.XYZ.us-east-2.rds.amazonaws.com:5432/postgres";
+  private static final String CACHE_RW_SERVER_ADDR = "cache-0.XYZ.us-east-2.rds.amazonaws.com:6379";
+  private static final String CACHE_RO_SERVER_ADDR = "cache-0.XYZ.us-east-2.rds.amazonaws.com:6380";
+
+  private Connection connection;
+  private int counter;
+  long startTime;
+
+  public static void main(String[] args) throws RunnerException {
+    Options opt = new OptionsBuilder()
+        .include(PgCacheBenchmarks.class.getSimpleName())
+        .addProfiler(GCProfiler.class)
+        .detectJvmArgs()
+        .build();
+
+    new Runner(opt).run();
+  }
+
+  @Setup(Level.Trial)
+  public void setup() throws SQLException {
+    try {
+      software.amazon.jdbc.Driver.register();
+    } catch (IllegalStateException e) {
+      System.out.println("exception during register() is " + e.getMessage());
+    }
+    Properties properties = new Properties();
+    properties.setProperty("wrapperPlugins", "dataRemoteCache");
+    properties.setProperty("cacheEndpointAddrRw", CACHE_RW_SERVER_ADDR);
+    properties.setProperty("cacheEndpointAddrRo", CACHE_RO_SERVER_ADDR);
+    properties.setProperty("wrapperLogUnclosedConnections", "true");
+    counter = 0;
+    connection = DriverManager.getConnection(DB_CONNECTION_STRING, properties);
+    startTime = System.currentTimeMillis();
+  }
+
+  @TearDown(Level.Trial)
+  public void tearDown() throws SQLException {
+    connection.close();
+  }
+
+  // Code to warm up the data in the table
+  public void warmUpDataSet() throws SQLException {
+    String desc_1KB = "mP48pHrR5vreBo3N6ecmlDgvfEAz0kQEOUQ89U3Rh05BTG9LhB8R0HBFBp5RIqc8vVcrphu89kW1OE2c2xApwpczFMdDAuk2SxOl9OrLvfk9zGYrdfzedcepT8LVeE6NTtYDeP3yo6UFC6AiOeqRBY5NEaNcZ8fuoXVpqOrqAhz910v5XrFxeXUyPDFxuaKFLaHfEFq7BRasUc9nfhP8gblKAGfEEmgYBpUKio27Rfo0xnavfVJQkAA2kME2PT4qZRSqeDkLmn7VBAzT9ghHqe9D4kQLQKjIyIPKqYoS8kW3ShW44VqYENwPSRAXw7UqOJqlKJ4pnmx4sPZO2kI4NYOl1JZXNlbGaSzJR0cOloKiY0z2OmUNvmD0Wju1DC9TT4OY6a6DOfFvk265BfDVxT6ufN68YG9sZuVsl7jq8SZSJg3x2cqlJuAtdSTIoKmJT1a6cEXxVusmdO27kRRp1BfWR4gz4w9HawYf9nBQOq76ObctlNvj0fYUUG3I49s3iP33CL8qZjj9RnyNUus6ieiZgta6L3mZuMRYOgCLyJrAKUYEL9KND7qirCPzVgmJHWIOnVewu8mldYFhroL89yvV3bZx4MGeyPU4KvbCsRgdORCTN0XhuLYUdiehHXnDBfuZ5yyR0saWLh8gjkLV5GkxTeKpOhpoK1o1cMiCDPYqTa64g5JundlW707c9zxc3Xnf2pW7E74YJl5oBu5vWEyPqXtYOtZOjOIRxxDY8QpoW8mpbQXxgB8DjkZZMiUCe0qHZYxvktVZJmHoaYBwpYpXVTZCfq9WajmkIOdIad1VnH5HpaECLRs6loa259yH8qesak2feDiKjfb8p3uj3s7WZUvPJwAWX9PIW1p7x6OiszXQCntOFRC3bQFNz1c98wlCBJnBSxbbYhU057TDNnoaib1h9bH7LAcqD1caE5KwLMAc5HqugkkRzT5NszkdJcpF0SxakdrAQLOKS6sNwDUzBJA76F775vmaqe3XIYecPmGtfoAKMychfEI4vfNr";
+    for (int i = 0; i < 400000; i++) {
+      Statement stmt = connection.createStatement();
+      String description = "description " + i;
+      String text = "here is my text data " + i;
+      String query = "insert into test values (" + i + ", " + i * 10 + ", '" + description + "', '" + text + "', " + i * 100 + 0.1234 + ", '2024-01-10', '10:00:00', '10:00:00-07', '2025-07-15 10:00:00', '2025-07-15 10:00:00-07'" + ", '" + desc_1KB + "');";
+      int rs = stmt.executeUpdate(query);
+      assert rs == 1;
+    }
+  }
+
+  private void validateResultSet(ResultSet rs, Blackhole b) throws SQLException {
+    while (rs.next()) {
+      b.consume(rs.getInt(1));
+      b.consume(rs.getInt(2));
+      b.consume(rs.getString(3));
+      b.consume(rs.getString(4));
+      b.consume(rs.getDouble(5));
+      b.consume(rs.getDate(6));
+      b.consume(rs.getTime(7));
+      b.consume(rs.getTime(8));
+      b.consume(rs.getTimestamp(9));
+      b.consume(rs.getTimestamp(10));
+      b.consume(rs.wasNull());
+    }
+  }
+
+  @Benchmark
+  public void runBenchmarkPrimaryKeyLookupNoCaching(Blackhole b) throws SQLException {
+    try (Statement stmt = connection.createStatement();
+         ResultSet rs = stmt.executeQuery("SELECT * FROM test where id = " + counter)) {
+      validateResultSet(rs, b);
+    }
+    counter++;
+  }
+
+  @Benchmark
+  public void runBenchmarkNonIndexedLookupNoCaching(Blackhole b) throws SQLException {
+    try (Statement stmt = connection.createStatement();
+         ResultSet rs = stmt.executeQuery("SELECT * FROM test where int_col = " + counter*10)) {
+      validateResultSet(rs, b);
+    }
+    counter++;
+  }
+
+  @Benchmark
+  public void runBenchmarkPrimaryKeyLookupWithCaching(Blackhole b) throws SQLException {
+    try (Statement stmt = connection.createStatement();
+         ResultSet rs = stmt.executeQuery("/*+ CACHE_PARAM(ttl=172800s) */ SELECT * FROM test where id = " + counter)) {
+         validateResultSet(rs, b);
+    }
+    counter++;
+  }
+
+  @Benchmark
+  public void runBenchmarkNonIndexedLookupWithCaching(Blackhole b) throws SQLException {
+    try (Statement stmt = connection.createStatement();
+         ResultSet rs = stmt.executeQuery("/*+ CACHE_PARAM(ttl=172800s) */ SELECT * FROM test where int_col = " + counter*10)) {
+      validateResultSet(rs, b);
+    }
+    counter++;
+  }
+}

docs/using-the-jdbc-driver/UsingTheJdbcDriver.md

@@ -220,7 +220,7 @@ The AWS JDBC Driver has several built-in plugins that are available to use. Plea
 [^2]: Federated Identity and Okta rely on IAM. Due to [^1], RDS Multi-AZ Clusters are not supported.
 
 > [!NOTE]\
-> To see information logged by plugins such as `DataCacheConnectionPlugin` and `LogQueryConnectionPlugin`, see the [Logging](#logging) section.
+> To see information logged by plugins such as `DataLocalCacheConnectionPlugin` and `LogQueryConnectionPlugin`, see the [Logging](#logging) section.
 
 In addition to the built-in plugins, you can also create custom plugins more suitable for your needs.
 For more information, see [Custom Plugins](../development-guide/LoadablePlugins.md#using-custom-plugins).

examples/AWSDriverExample/build.gradle.kts

@@ -29,6 +29,8 @@ dependencies {
     implementation("com.amazonaws:aws-xray-recorder-sdk-core:2.18.2")
     implementation("org.jsoup:jsoup:1.21.1")
     implementation("com.mchange:c3p0:0.11.0")
+    implementation("io.lettuce:lettuce-core:6.6.0.RELEASE")
+    implementation("org.apache.commons:commons-pool2:2.11.1")
 }
 
 tasks.withType<JavaExec> {

examples/AWSDriverExample/src/main/java/software/amazon/DatabaseConnectionWithCacheExample.java

@@ -0,0 +1,84 @@
+package software.amazon;
+
+import software.amazon.util.EnvLoader;
+import java.sql.*;
+import java.util.*;
+import java.util.logging.Logger;
+
+public class DatabaseConnectionWithCacheExample {
+
+  private static final EnvLoader env = new EnvLoader();
+
+  private static final String DB_CONNECTION_STRING = env.get("DB_CONNECTION_STRING");
+  private static final String CACHE_RW_SERVER_ADDR = env.get("CACHE_RW_SERVER_ADDR");
+  private static final String CACHE_RO_SERVER_ADDR = env.get("CACHE_RO_SERVER_ADDR");
+  // If the cache server is authenticated with IAM
+  private static final String CACHE_NAME = env.get("CACHE_NAME");
+  // Both IAM and traditional auth uses the same CACHE_USERNAME
+  private static final String CACHE_USERNAME = env.get("CACHE_USERNAME"); // e.g., "iam-user-01" / "username"
+  private static final String CACHE_IAM_REGION = env.get("CACHE_IAM_REGION"); // e.g., "us-west-2"
+  private static final String CACHE_USE_SSL = env.get("CACHE_USE_SSL");
+  // If the cache server is authenticated with traditional username password
+  // private static final String CACHE_PASSWORD = env.get("CACHE_PASSWORD");
+  private static final String USERNAME = env.get("DB_USERNAME");
+  private static final String PASSWORD = env.get("DB_PASSWORD");
+  private static final int THREAD_COUNT = 8; //Use 8 Threads
+  private static final long TEST_DURATION_MS = 16000; //Test duration for 16 seconds
+  private static final String CACHE_CONNECTION_TIMEOUT = env.get("CACHE_CONNECTION_TIMEOUT"); //Set connection timeout in milliseconds
+  private static final String CACHE_CONNECTION_POOL_SIZE = env.get("CACHE_CONNECTION_POOL_SIZE"); //Set connection pool size
+
+  public static void main(String[] args) throws SQLException {
+    final Properties properties = new Properties();
+    final Logger LOGGER = Logger.getLogger(DatabaseConnectionWithCacheExample.class.getName());
+
+    // Configuring connection properties for the underlying JDBC driver.
+    properties.setProperty("user", USERNAME);
+    properties.setProperty("password", PASSWORD);
+
+    // Configuring connection properties for the JDBC Wrapper.
+    properties.setProperty("wrapperPlugins", "dataRemoteCache");
+    properties.setProperty("cacheEndpointAddrRw", CACHE_RW_SERVER_ADDR);
+    properties.setProperty("cacheEndpointAddrRo", CACHE_RO_SERVER_ADDR);
+    // If the cache server is authenticated with IAM
+    properties.setProperty("cacheName", CACHE_NAME);
+    properties.setProperty("cacheUsername", CACHE_USERNAME);
+    properties.setProperty("cacheIamRegion", CACHE_IAM_REGION);
+    // If the cache server is authenticated with traditional username password
+    // properties.setProperty("cachePassword", PASSWORD);
+    properties.setProperty("cacheUseSSL", CACHE_USE_SSL); // "true" or "false"
+    properties.setProperty("wrapperLogUnclosedConnections", "true");
+    properties.setProperty("cacheConnectionTimeout", CACHE_CONNECTION_TIMEOUT);
+    properties.setProperty("cacheConnectionPoolSize", CACHE_CONNECTION_POOL_SIZE);
+    String queryStr = "/*+ CACHE_PARAM(ttl=300s) */ select * from cinemas";
+
+    // Create threads for concurrent connection testing
+    Thread[] threads = new Thread[THREAD_COUNT];
+    for (int t = 0; t < THREAD_COUNT; t++) {
+      // Each thread uses a single connection for multiple queries
+      threads[t] = new Thread(() -> {
+        try {
+          try (Connection conn = DriverManager.getConnection(DB_CONNECTION_STRING, properties)) {
+            long endTime = System.currentTimeMillis() + TEST_DURATION_MS;
+            try (Statement stmt = conn.createStatement()) {
+              while (System.currentTimeMillis() < endTime) {
+                ResultSet rs = stmt.executeQuery(queryStr);
+                System.out.println("Executed the SQL query with result sets: " + rs.toString());
+              }
+            }
+          }
+        } catch (Exception e) {
+          LOGGER.warning("Error: " + e.getMessage());
+        }
+      });
+      threads[t].start();
+    }
+    // Wait for all threads to complete
+    for (Thread thread : threads) {
+      try {
+        thread.join();
+      } catch (InterruptedException e) {
+        LOGGER.warning("Thread interrupted: " + e.getMessage());
+      }
+    }
+  }
+}

examples/AWSDriverExample/src/main/java/software/amazon/util/EnvLoader.java

@@ -0,0 +1,83 @@
+package software.amazon.util;
+
+import java.io.BufferedReader;
+import java.io.FileReader;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * A simple utility class to load environment variables from a .env file.
+ */
+public class EnvLoader {
+  private final Map<String, String> envVars = new HashMap<>();
+
+  /**
+   * Loads environment variables from a .env file in the current directory.
+   */
+  public EnvLoader() {
+    this(Paths.get(".env"));
+  }
+
+  /**
+   * Loads environment variables from the specified file path.
+   *
+   * @param envPath Path to the .env file
+   */
+  public EnvLoader(Path envPath) {
+    if (Files.exists(envPath)) {
+      try (BufferedReader reader = new BufferedReader(new FileReader(envPath.toFile()))) {
+        String line;
+        while ((line = reader.readLine()) != null) {
+          parseLine(line);
+        }
+      } catch (IOException e) {
+        System.err.println("Error reading .env file: " + e.getMessage());
+      }
+    }
+  }
+
+  private void parseLine(String line) {
+    line = line.trim();
+    // Skip empty lines and comments
+    if (line.isEmpty() || line.startsWith("#")) {
+      return;
+    }
+
+    // Split on the first equals sign
+    int delimiterPos = line.indexOf('=');
+    if (delimiterPos > 0) {
+      String key = line.substring(0, delimiterPos).trim();
+      String value = line.substring(delimiterPos + 1).trim();
+
+      // Remove quotes if present
+      if ((value.startsWith("\"") && value.endsWith("\"")) || 
+          (value.startsWith("'") && value.endsWith("'"))) {
+        value = value.substring(1, value.length() - 1);
+      }
+
+      envVars.put(key, value);
+    }
+  }
+
+  /**
+   * Gets the value of an environment variable.
+   *
+   * @param key The name of the environment variable
+   * @return The value of the environment variable, or null if not found
+   */
+  public String get(String key) {
+    // First check the loaded .env file
+    String value = envVars.get(key);
+
+    // If not found, check system environment variables
+    if (value == null) {
+      value = System.getenv(key);
+    }
+
+    return value;
+  }
+}

examples/AWSDriverExample/src/main/resources/logback.xml

@@ -0,0 +1,6 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+    <!-- Root logger defines the default logging level and appenders -->
+    <root level="info">
+    </root>
+</configuration>

wrapper/build.gradle.kts

@@ -44,8 +44,10 @@ dependencies {
     optionalImplementation("com.mchange:c3p0:0.11.0")
     optionalImplementation("org.apache.httpcomponents:httpclient:4.5.14")
     optionalImplementation("com.fasterxml.jackson.core:jackson-databind:2.19.0")
+    optionalImplementation("org.apache.commons:commons-pool2:2.11.1")
     optionalImplementation("org.jsoup:jsoup:1.21.1")
     optionalImplementation("com.amazonaws:aws-xray-recorder-sdk-core:2.18.2")
+    optionalImplementation("io.lettuce:lettuce-core:6.6.0.RELEASE")
     optionalImplementation("io.opentelemetry:opentelemetry-api:1.52.0")
     optionalImplementation("io.opentelemetry:opentelemetry-sdk:1.52.0")
     optionalImplementation("io.opentelemetry:opentelemetry-sdk-metrics:1.52.0")
@@ -98,10 +100,12 @@ dependencies {
     testImplementation("org.slf4j:slf4j-simple:2.0.17")
     testImplementation("com.fasterxml.jackson.core:jackson-databind:2.19.0")
     testImplementation("com.amazonaws:aws-xray-recorder-sdk-core:2.18.2")
+    testImplementation("io.lettuce:lettuce-core:6.6.0.RELEASE")
     testImplementation("io.opentelemetry:opentelemetry-api:1.52.0")
     testImplementation("io.opentelemetry:opentelemetry-sdk:1.52.0")
     testImplementation("io.opentelemetry:opentelemetry-sdk-metrics:1.52.0")
     testImplementation("io.opentelemetry:opentelemetry-exporter-otlp:1.52.0")
+    testImplementation("org.apache.commons:commons-pool2:2.11.1")
     testImplementation("org.jsoup:jsoup:1.21.1")
     testImplementation("de.vandermeer:asciitable:0.3.2")
     testImplementation("org.hibernate:hibernate-core:5.6.15.Final") // the latest version compatible with Java 8
@@ -208,7 +212,7 @@ if (useJacoco) {
                         "software/amazon/jdbc/wrapper/*",
                         "software/amazon/jdbc/util/*",
                         "software/amazon/jdbc/profile/*",
-                        "software/amazon/jdbc/plugin/DataCacheConnectionPlugin*"
+                        "software/amazon/jdbc/plugin/cache/DataLocalCacheConnectionPlugin*"
                     )
                 }
             }))
@@ -223,7 +227,7 @@ if (useJacoco) {
                         "software/amazon/jdbc/wrapper/*",
                         "software/amazon/jdbc/util/*",
                         "software/amazon/jdbc/profile/*",
-                        "software/amazon/jdbc/plugin/DataCacheConnectionPlugin*"
+                        "software/amazon/jdbc/plugin/cache/DataLocalCacheConnectionPlugin*"
                     )
                 }
             }))