Optimize PreparedStatementIT, BatchStatementIT, and BoundStatementCcmIT to reuse sessions across tests

Convert @rule to @ClassRule for CcmRule and SessionRule so that the
CCM cluster and CQL session are created once per test class instead of
once per test method. This eliminates repeated cluster startup/teardown
overhead across all tests in these classes.

For PreparedStatementIT specifically:
- Replace DROP+CREATE TABLE with TRUNCATE between tests that don't
  modify schema, since most tests only need a clean table
- Track schema-altering tests (ALTER TABLE, DROP TABLE) with a flag
  so the table is only recreated when actually needed
- Use regular session.execute() for INSERT statements instead of
  routing them through the slow DDL execution profile
- Use try-with-resources in cross-session test to avoid prepare cache
  interference from the shared session

Retain CcmRule as a per-method @rule so that @BackendRequirement,
@ScyllaOnly, and @ScyllaSkip annotations are still evaluated per test.

Author: dkropachev

integration-tests/src/test/java/com/datastax/oss/driver/core/cql/BatchStatementIT.java

@@ -56,7 +56,8 @@
 import com.datastax.oss.driver.internal.core.util.LoggerTest;
 import java.util.Iterator;
 import java.util.List;
-import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
 import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
@@ -68,18 +69,21 @@
 @Category(ParallelizableTests.class)
 public class BatchStatementIT {
 
-  private final CcmRule CCM_RULE = CcmRule.getInstance();
+  private static final CcmRule CCM_RULE = CcmRule.getInstance();
 
-  private final SessionRule<CqlSession> SESSION_RULE = SessionRule.builder(CCM_RULE).build();
+  private static final SessionRule<CqlSession> SESSION_RULE = SessionRule.builder(CCM_RULE).build();
 
-  @Rule public TestRule chain = RuleChain.outerRule(CCM_RULE).around(SESSION_RULE);
+  @ClassRule public static TestRule classChain = RuleChain.outerRule(CCM_RULE).around(SESSION_RULE);
+
+  // CcmRule as @Rule for per-method @BackendRequirement annotation checking
+  @Rule public TestRule methodChain = CCM_RULE;
 
   @Rule public TestName name = new TestName();
 
   private static final int batchCount = 100;
 
-  @Before
-  public void createTable() {
+  @BeforeClass
+  public static void createTable() {
     String[] schemaStatements =
         new String[] {
           "CREATE TABLE test (k0 text, k1 int, v int, PRIMARY KEY (k0, k1))",

integration-tests/src/test/java/com/datastax/oss/driver/core/cql/BoundStatementCcmIT.java

@@ -77,7 +77,8 @@
 import java.util.Map;
 import java.util.concurrent.CompletionStage;
 import java.util.function.Function;
-import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
 import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
@@ -91,28 +92,31 @@
 @RunWith(DataProviderRunner.class)
 public class BoundStatementCcmIT {
 
-  private CcmRule ccmRule = CcmRule.getInstance();
+  private static final CcmRule ccmRule = CcmRule.getInstance();
 
-  private final boolean atLeastV4 = ccmRule.getHighestProtocolVersion().getCode() >= 4;
+  private static final boolean atLeastV4 = ccmRule.getHighestProtocolVersion().getCode() >= 4;
 
-  private SessionRule<CqlSession> sessionRule =
+  private static final SessionRule<CqlSession> sessionRule =
       SessionRule.builder(ccmRule)
           .withConfigLoader(
               SessionUtils.configLoaderBuilder()
                   .withInt(DefaultDriverOption.REQUEST_PAGE_SIZE, 20)
                   .build())
           .build();
 
-  @Rule public TestRule chain = RuleChain.outerRule(ccmRule).around(sessionRule);
+  @ClassRule public static TestRule classChain = RuleChain.outerRule(ccmRule).around(sessionRule);
+
+  // CcmRule as @Rule for per-method @ScyllaSkip / @BackendRequirement annotation checking
+  @Rule public TestRule methodChain = ccmRule;
 
   @Rule public TestName name = new TestName();
 
   private static final String KEY = "test";
 
   private static final int VALUE = 7;
 
-  @Before
-  public void setupSchema() {
+  @BeforeClass
+  public static void setupSchema() {
     // table where every column forms the primary key.
     SchemaChangeSynchronizer.withLock(
         () -> {

integration-tests/src/test/java/com/datastax/oss/driver/core/cql/PreparedStatementIT.java

@@ -64,6 +64,7 @@
 import junit.framework.TestCase;
 import org.assertj.core.api.AbstractThrowableAssert;
 import org.junit.Before;
+import org.junit.ClassRule;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
@@ -84,9 +85,9 @@ public class PreparedStatementIT {
   private static final Version SCYLLA_METADATA_ID_SUPPORT_VERSION =
       Objects.requireNonNull(Version.parse("2025.3"));
 
-  private final CcmRule ccmRule = CcmRule.getInstance();
+  private static final CcmRule ccmRule = CcmRule.getInstance();
 
-  private final SessionRule<CqlSession> sessionRule =
+  private static final SessionRule<CqlSession> sessionRule =
       SessionRule.builder(ccmRule)
           .withConfigLoader(
               SessionUtils.configLoaderBuilder()
@@ -95,19 +96,31 @@ public class PreparedStatementIT {
                   .build())
           .build();
 
-  @Rule public TestRule chain = RuleChain.outerRule(ccmRule).around(sessionRule);
+  @ClassRule public static TestRule classChain = RuleChain.outerRule(ccmRule).around(sessionRule);
+
+  // CcmRule as @Rule for per-method @ScyllaOnly / @BackendRequirement annotation checking
+  @Rule public TestRule methodChain = ccmRule;
+
+  // When true, @Before will DROP+CREATE the table (needed after tests that ALTER TABLE).
+  // Starts as true so the table is created on first run.
+  private static boolean needsTableRecreate = true;
 
   @Before
   public void setupSchema() {
+    if (needsTableRecreate) {
+      executeDdl("DROP TABLE IF EXISTS prepared_statement_test");
+      executeDdl("CREATE TABLE prepared_statement_test (a int PRIMARY KEY, b int, c int)");
+      needsTableRecreate = false;
+    } else {
+      executeDdl("TRUNCATE prepared_statement_test");
+    }
     for (String query :
         ImmutableList.of(
-            "DROP TABLE IF EXISTS prepared_statement_test",
-            "CREATE TABLE prepared_statement_test (a int PRIMARY KEY, b int, c int)",
             "INSERT INTO prepared_statement_test (a, b, c) VALUES (1, 1, 1)",
             "INSERT INTO prepared_statement_test (a, b, c) VALUES (2, 2, 2)",
             "INSERT INTO prepared_statement_test (a, b, c) VALUES (3, 3, 3)",
             "INSERT INTO prepared_statement_test (a, b, c) VALUES (4, 4, 4)")) {
-      executeDdl(query);
+      sessionRule.session().execute(query);
     }
   }
 
@@ -176,6 +189,7 @@ public void should_update_metadata_when_schema_changed_across_executions() {
     }
 
     // When
+    needsTableRecreate = true;
     session.execute(
         SimpleStatement.builder("ALTER TABLE prepared_statement_test ADD d int")
             .setExecutionProfile(sessionRule.slowProfile())
@@ -239,6 +253,7 @@ public void should_update_metadata_when_schema_changed_across_pages() {
     }
 
     // When
+    needsTableRecreate = true;
     session.execute(
         SimpleStatement.builder("ALTER TABLE prepared_statement_test ADD d int")
             .setExecutionProfile(sessionRule.slowProfile())
@@ -271,71 +286,71 @@ public void should_update_metadata_when_schema_changed_across_pages() {
   @BackendRequirement(type = BackendType.CASSANDRA, minInclusive = "4.0")
   @BackendRequirement(type = BackendType.SCYLLA)
   public void should_update_metadata_when_schema_changed_across_sessions() {
-    // Given
-    CqlSession session1 = sessionRule.session();
-    CqlSession session2 = SessionUtils.newSession(ccmRule, sessionRule.keyspace());
-
-    PreparedStatement ps1 = session1.prepare("SELECT * FROM prepared_statement_test WHERE a = ?");
-    PreparedStatement ps2 = session2.prepare("SELECT * FROM prepared_statement_test WHERE a = ?");
-
-    ByteBuffer id1a = ps1.getResultMetadataId();
-    ByteBuffer id2a = ps2.getResultMetadataId();
-    if (hasNoScyllaMetadataIdSupport()) {
-      // Scylla does not support CQL5 extensions and metadata id
-      assertThat(id1a).isNull();
-      assertThat(id2a).isNull();
-    }
+    // Use fresh sessions to avoid prepare cache interference from other tests that use the
+    // same query string on the shared session.
+    try (CqlSession session1 = SessionUtils.newSession(ccmRule, sessionRule.keyspace());
+        CqlSession session2 = SessionUtils.newSession(ccmRule, sessionRule.keyspace())) {
+      // Given
+      PreparedStatement ps1 = session1.prepare("SELECT * FROM prepared_statement_test WHERE a = ?");
+      PreparedStatement ps2 = session2.prepare("SELECT * FROM prepared_statement_test WHERE a = ?");
+
+      ByteBuffer id1a = ps1.getResultMetadataId();
+      ByteBuffer id2a = ps2.getResultMetadataId();
+      if (hasNoScyllaMetadataIdSupport()) {
+        // Scylla does not support CQL5 extensions and metadata id
+        assertThat(id1a).isNull();
+        assertThat(id2a).isNull();
+      }
 
-    ResultSet rows1 = session1.execute(ps1.bind(1));
-    ResultSet rows2 = session2.execute(ps2.bind(1));
+      ResultSet rows1 = session1.execute(ps1.bind(1));
+      ResultSet rows2 = session2.execute(ps2.bind(1));
 
-    assertThat(rows1.getColumnDefinitions()).hasSize(3);
-    assertThat(rows1.getColumnDefinitions().contains("d")).isFalse();
-    assertThat(rows2.getColumnDefinitions()).hasSize(3);
-    assertThat(rows2.getColumnDefinitions().contains("d")).isFalse();
+      assertThat(rows1.getColumnDefinitions()).hasSize(3);
+      assertThat(rows1.getColumnDefinitions().contains("d")).isFalse();
+      assertThat(rows2.getColumnDefinitions()).hasSize(3);
+      assertThat(rows2.getColumnDefinitions().contains("d")).isFalse();
 
-    // When
-    session1.execute("ALTER TABLE prepared_statement_test ADD d int");
+      // When
+      needsTableRecreate = true;
+      session1.execute("ALTER TABLE prepared_statement_test ADD d int");
 
-    rows1 = session1.execute(ps1.bind(1));
-    rows2 = session2.execute(ps2.bind(1));
+      rows1 = session1.execute(ps1.bind(1));
+      rows2 = session2.execute(ps2.bind(1));
 
-    ByteBuffer id1b = ps1.getResultMetadataId();
-    ByteBuffer id2b = ps2.getResultMetadataId();
+      ByteBuffer id1b = ps1.getResultMetadataId();
+      ByteBuffer id2b = ps2.getResultMetadataId();
 
-    // Then
-    if (hasNoScyllaMetadataIdSupport()) {
-      // Scylla does not support CQL5 extensions and metadata id
-      assertThat(id1b).isNull();
-      assertThat(id2b).isNull();
+      // Then
+      if (hasNoScyllaMetadataIdSupport()) {
+        // Scylla does not support CQL5 extensions and metadata id
+        assertThat(id1b).isNull();
+        assertThat(id2b).isNull();
+
+        assertThat(ps1.getResultSetDefinitions()).hasSize(3);
+        assertThat(ps1.getResultSetDefinitions().contains("d")).isFalse();
+        assertThat(ps2.getResultSetDefinitions()).hasSize(3);
+        assertThat(ps2.getResultSetDefinitions().contains("d")).isFalse();
+
+        assertThat(rows1.getColumnDefinitions()).hasSize(4);
+        assertThat(rows1.getColumnDefinitions().contains("d")).isTrue();
+        assertThat(rows2.getColumnDefinitions()).hasSize(4);
+        assertThat(rows2.getColumnDefinitions().contains("d")).isTrue();
+
+        return;
+      }
+      assertThat(Bytes.toHexString(id1b)).isNotEqualTo(Bytes.toHexString(id1a));
+      assertThat(Bytes.toHexString(id2b)).isNotEqualTo(Bytes.toHexString(id2a));
 
-      assertThat(ps1.getResultSetDefinitions()).hasSize(3);
-      assertThat(ps1.getResultSetDefinitions().contains("d")).isFalse();
-      assertThat(ps2.getResultSetDefinitions()).hasSize(3);
-      assertThat(ps2.getResultSetDefinitions().contains("d")).isFalse();
+      assertThat(ps1.getResultSetDefinitions()).hasSize(4);
+      assertThat(ps1.getResultSetDefinitions().contains("d")).isTrue();
+      assertThat(ps2.getResultSetDefinitions()).hasSize(4);
+      assertThat(ps2.getResultSetDefinitions().contains("d")).isTrue();
 
       assertThat(rows1.getColumnDefinitions()).hasSize(4);
       assertThat(rows1.getColumnDefinitions().contains("d")).isTrue();
       assertThat(rows2.getColumnDefinitions()).hasSize(4);
       assertThat(rows2.getColumnDefinitions().contains("d")).isTrue();
-
-      session2.close();
-      return;
     }
-    assertThat(Bytes.toHexString(id1b)).isNotEqualTo(Bytes.toHexString(id1a));
-    assertThat(Bytes.toHexString(id2b)).isNotEqualTo(Bytes.toHexString(id2a));
-
-    assertThat(ps1.getResultSetDefinitions()).hasSize(4);
-    assertThat(ps1.getResultSetDefinitions().contains("d")).isTrue();
-    assertThat(ps2.getResultSetDefinitions()).hasSize(4);
-    assertThat(ps2.getResultSetDefinitions().contains("d")).isTrue();
-
-    assertThat(rows1.getColumnDefinitions()).hasSize(4);
-    assertThat(rows1.getColumnDefinitions().contains("d")).isTrue();
-    assertThat(rows2.getColumnDefinitions()).hasSize(4);
-    assertThat(rows2.getColumnDefinitions().contains("d")).isTrue();
-
-    session2.close();
   }
 
   @Test
@@ -344,6 +359,7 @@ public void should_update_metadata_when_schema_changed_across_sessions() {
   public void should_fail_to_reprepare_if_query_becomes_invalid() {
     // Given
     CqlSession session = sessionRule.session();
+    needsTableRecreate = true;
     session.execute("ALTER TABLE prepared_statement_test ADD d int");
     PreparedStatement ps =
         session.prepare("SELECT a, b, c, d FROM prepared_statement_test WHERE a = ?");
@@ -439,6 +455,7 @@ private void should_not_store_metadata_for_conditional_updates(CqlSession sessio
     assertThat(Bytes.toHexString(ps.getResultMetadataId())).isEqualTo(Bytes.toHexString(idBefore));
 
     // When
+    needsTableRecreate = true;
     session.execute("ALTER TABLE prepared_statement_test ADD d int");
     rs = session.execute(ps.bind(5, 5, 5));
 
@@ -565,6 +582,7 @@ public void should_create_separate_instances_for_different_statement_parameters(
       session.execute("USE " + sessionRule.keyspace().asCql(false));
 
       // Drop and recreate the table to invalidate the prepared statement server-side
+      needsTableRecreate = true;
       executeDdl("DROP TABLE prepared_statement_test");
       executeDdl("CREATE TABLE prepared_statement_test (a int PRIMARY KEY, b int, c int)");