Add ConnectionFactory + connection-state/transaction negative suites (PR 5a/8)

src/test/java/com/databricks/jdbc/comparator/COMPARATOR_README.md

@@ -258,8 +258,17 @@ If both are present for a method, **runOnly takes precedence**: argument combina
 | `NEGATIVE_STATEMENT_DDL` | Error-provoking CREATE / ALTER / DROP (missing/duplicate objects, bad namespace, malformed) |
 | `NEGATIVE_STATEMENT_DML` | Error-provoking INSERT / UPDATE / DELETE (type mismatch, NOT NULL, missing table, overflow) |
 | `NEGATIVE_STATEMENT_BATCH` | executeBatch partial/full failure + per-element BatchUpdateException counts |
+| `NEGATIVE_CONNECTION_STATE` | setCatalog/setSchema/setClientInfo/USE to nonexistent targets (own fresh connections) |
+| `NEGATIVE_TRANSACTION` | commit/rollback with autocommit on; DDL inside a manual transaction (own fresh connections) |
 
 Negative suites compare each endpoint's **error behavior** (exception class, SQLState, vendor code,
 message) via the `ERROR_COMPARISON_MODE` gate (default `shadow`). See
 [`error/`](error/) for the comparison engine.
 
+Suites whose cases mutate or destroy connection/session state (`NEGATIVE_CONNECTION_STATE`,
+`NEGATIVE_TRANSACTION`, and — later — connection/cancel/volume cases) open their **own dedicated,
+uncached connections** via `ConnectionFactory.openFresh(side)` and close them in a `finally`, so
+they never poison the shared connections the other suites reuse. Providers request this by
+overriding the `execute(conn1, conn2, ConnectionFactory, testCase, label)` overload of
+`SuiteProvider`.
+

src/test/java/com/databricks/jdbc/comparator/JDBCDriverComparisonTest.java

@@ -3,6 +3,7 @@
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 
 import com.databricks.jdbc.comparator.config.ConnectionConfig;
+import com.databricks.jdbc.comparator.config.ConnectionFactory;
 import com.databricks.jdbc.comparator.config.ConnectionManager;
 import com.databricks.jdbc.comparator.config.Endpoint;
 import com.databricks.jdbc.comparator.config.TestSuite;
@@ -138,14 +139,26 @@ static Stream<Arguments> provideAllTests() {
     for (ConnectionConfig config : ConnectionConfig.activeConfigs()) {
       Connection leftConn;
       Connection rightConn;
+      final String leftUrl = config.buildUrl(BASE_LEFT_URL);
+      final String rightUrl = config.buildUrl(BASE_RIGHT_URL);
       try {
-        leftConn = connectionManager.getConnection(config.buildUrl(BASE_LEFT_URL));
-        rightConn = connectionManager.getConnection(config.buildUrl(BASE_RIGHT_URL));
+        leftConn = connectionManager.getConnection(leftUrl);
+        rightConn = connectionManager.getConnection(rightUrl);
       } catch (SQLException e) {
         throw new RuntimeException(
             "Failed to create connections for config: " + config.getDisplayName(), e);
       }
 
+      // Per-config factory for suites that need dedicated (uncached) connections. Bound to this
+      // config's resolved LEFT/RIGHT URLs; the returned connections are NOT cached, so the suite
+      // owns closing them and cannot poison the shared connections above.
+      ConnectionFactory factory =
+          side -> {
+            if ("LEFT".equalsIgnoreCase(side)) return connectionManager.openUncached(leftUrl);
+            if ("RIGHT".equalsIgnoreCase(side)) return connectionManager.openUncached(rightUrl);
+            throw new IllegalArgumentException("side must be LEFT or RIGHT, got: " + side);
+          };
+
       for (TestSuite suite : config.getApplicableSuites()) {
         if (allowedSuites != null && !allowedSuites.contains(suite.name())) continue;
 
@@ -158,7 +171,7 @@ static Stream<Arguments> provideAllTests() {
         String label = config.getDisplayName() + " | " + suite.name();
 
         for (TestCase tc : testCases) {
-          allTests.add(Arguments.of(label, leftConn, rightConn, suite, tc));
+          allTests.add(Arguments.of(label, leftConn, rightConn, factory, suite, tc));
         }
       }
     }
@@ -172,6 +185,7 @@ void compareDriverResults(
       String comparisonName,
       Connection conn1,
       Connection conn2,
+      ConnectionFactory factory,
       TestSuite suite,
       TestCase testCase) {
     assertDoesNotThrow(
@@ -186,7 +200,8 @@ void compareDriverResults(
                   ? comparisonName.substring(0, comparisonName.indexOf(" | "))
                   : comparisonName;
           try {
-            ComparisonResult result = suite.getProvider().execute(conn1, conn2, testCase, label);
+            ComparisonResult result =
+                suite.getProvider().execute(conn1, conn2, factory, testCase, label);
             reporter.addResult(result);
 
             // CSV: per-combo rows for metadata, single row for other suites

src/test/java/com/databricks/jdbc/comparator/config/ConnectionFactory.java

@@ -0,0 +1,26 @@
+package com.databricks.jdbc.comparator.config;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+
+/**
+ * Opens fresh, uncached connections for suites whose cases mutate or destroy connection/session
+ * state ({@code setCatalog}, {@code setAutoCommit}, {@code cancel}, {@code close}, {@code
+ * setClientInfo}, {@code USE …}). Such a case run on the shared cached connection would poison
+ * every later test; each must instead operate on its own short-lived connection and close it in a
+ * {@code finally}.
+ *
+ * <p>A factory is bound to a single {@link ConnectionConfig} — {@link #openFresh} uses that
+ * config's resolved LEFT/RIGHT URL and the shared token. The returned connection is NOT cached by
+ * {@link ConnectionManager}, so closing it is the caller's responsibility and does not affect the
+ * shared connections other suites use.
+ */
+public interface ConnectionFactory {
+
+  /**
+   * Opens a new, uncached connection for the named side.
+   *
+   * @param side "LEFT" or "RIGHT" (case-insensitive)
+   */
+  Connection openFresh(String side) throws SQLException;
+}

src/test/java/com/databricks/jdbc/comparator/config/ConnectionManager.java

@@ -33,6 +33,15 @@ public Connection getConnection(String url) throws SQLException {
     return conn;
   }
 
+  /**
+   * Opens a NEW connection for the given URL that is NOT cached — the caller owns its lifecycle and
+   * must close it. Used by suites whose cases mutate/destroy connection state and therefore need a
+   * dedicated, throwaway connection rather than the shared cached one.
+   */
+  public Connection openUncached(String url) throws SQLException {
+    return DriverManager.getConnection(url, "token", token);
+  }
+
   /** Returns all active connection URLs, useful for report headers. */
   public List<String> getActiveUrls() {
     return new ArrayList<>(cache.keySet());

src/test/java/com/databricks/jdbc/comparator/config/TestSuite.java

@@ -33,7 +33,12 @@ public enum TestSuite {
   // under comparator_ddl_tests (seeded fresh and dropped), so safe on the shared connections.
   NEGATIVE_STATEMENT_DDL(new NegativeStatementDdlProvider()),
   NEGATIVE_STATEMENT_DML(new NegativeStatementDmlProvider()),
-  NEGATIVE_STATEMENT_BATCH(new NegativeStatementBatchProvider());
+  NEGATIVE_STATEMENT_BATCH(new NegativeStatementBatchProvider()),
+
+  // Negative suites that mutate connection/session state — each opens its OWN fresh connections
+  // via ConnectionFactory (closed in a finally), so they never poison the shared connections.
+  NEGATIVE_CONNECTION_STATE(new NegativeConnectionStateProvider()),
+  NEGATIVE_TRANSACTION(new NegativeTransactionProvider());
 
   private final SuiteProvider provider;
 

src/test/java/com/databricks/jdbc/comparator/error/ErrorComparatorTest.java

@@ -204,6 +204,39 @@ void errorDiffsOffModeStillFlagsDifferingClass() {
         });
   }
 
+  @Test
+  void foldIntoPutsOneSidedInMetadataSoCsvSummaryKeepsIt() {
+    // A one-sided outcome (one threw, one returned) must land in metadataDifferences and survive
+    // csvSummary() — the bug that produced an empty CSV summary when it was mis-filed in data.
+    withMode(
+        "shadow",
+        () -> {
+          CapturedOutcome thrower = CapturedOutcome.threw(new SQLException("boom", "42P07", 0));
+          CapturedOutcome returned = CapturedOutcome.returned(Boolean.FALSE);
+          ComparisonResult result = new ComparisonResult("t", "q", new Object[0]);
+          ErrorDiffs.foldInto(result, thrower, returned, "result ", "");
+          assertTrue(
+              result.metadataDifferences.stream().anyMatch(d -> d.startsWith("Error one-sided: ")));
+          assertTrue(result.dataDifferences.isEmpty());
+          assertFalse(result.csvSummary().isEmpty(), "one-sided diff must appear in CSV summary");
+        });
+  }
+
+  @Test
+  void foldIntoPutsFieldMismatchInData() {
+    withMode(
+        "shadow",
+        () -> {
+          CapturedOutcome l = CapturedOutcome.threw(new SQLException("a", "42P01", 1));
+          CapturedOutcome r = CapturedOutcome.threw(new SQLException("b", "08000", 2));
+          ComparisonResult result = new ComparisonResult("t", "q", new Object[0]);
+          ErrorDiffs.foldInto(result, l, r, "v ", "OP: ");
+          assertTrue(result.metadataDifferences.isEmpty());
+          assertTrue(
+              result.dataDifferences.stream().anyMatch(d -> d.startsWith("OP: Error SQLState")));
+        });
+  }
+
   @Test
   void errorDiffsNeitherThrewIsEmptyInBothModes() {
     // Both sides returned (e.g. DROP TABLE IF EXISTS succeeds on both) -> no diffs, no NPE.

src/test/java/com/databricks/jdbc/comparator/error/ErrorDiffs.java

@@ -1,13 +1,15 @@
 package com.databricks.jdbc.comparator.error;
 
+import com.databricks.jdbc.comparator.ComparisonResult;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 
 /**
- * Bridges suite providers that capture their own {@link CapturedOutcome}s (DML, Volume) to the
- * error comparison, while honoring the {@code ERROR_COMPARISON_MODE} gate the same way {@code
- * ResultSetComparator.compareErrorsDeeply} does — so {@code off} is a true kill switch everywhere.
+ * Bridges suite providers that capture their own {@link CapturedOutcome}s (DML, Volume, and the
+ * connection-state/transaction suites) to the error comparison, while honoring the {@code
+ * ERROR_COMPARISON_MODE} gate the same way {@code ResultSetComparator.compareErrorsDeeply} does —
+ * so {@code off} is a true kill switch everywhere.
  *
  * <p>When the gate is on, compares errors deeply (class + SQLState + code + message). When off,
  * falls back to the minimal legacy check (exception class only) so no SQLState/code/message diffs
@@ -18,9 +20,49 @@ public final class ErrorDiffs {
   private ErrorDiffs() {}
 
   /**
-   * Returns diff strings for a pair of outcomes where at least one threw. Empty when they agree
-   * (per the active mode). {@code returnedLabel} prefixes how a returned value is rendered in a
-   * one-sided diff (e.g. {@code "update count "}).
+   * Compares two captured outcomes and folds the diffs into {@code result}, preserving the
+   * metadata/data split that {@code ComparisonResult.csvSummary()} relies on — one-sided diffs
+   * (carrying {@link ErrorComparator#ONE_SIDED_PREFIX}) go to {@code metadataDifferences}, field
+   * mismatches to {@code dataDifferences}. Callers should prefer this over {@link #compare} so a
+   * one-sided error is not mis-filed and dropped from the CSV summary. Honors the gate:
+   * neither-threw and off-mode legacy behavior match {@link #compare}. {@code prefix} (may be
+   * empty) is prepended to each diff for readable per-operation reports.
+   */
+  public static void foldInto(
+      ComparisonResult result,
+      CapturedOutcome left,
+      CapturedOutcome right,
+      String returnedLabel,
+      String prefix) {
+    if (!left.threw() && !right.threw()) {
+      return;
+    }
+    ErrorPolicy policy = ErrorPolicy.active();
+    if (!policy.isDeepComparisonEnabled()) {
+      for (String d : legacyClassOnly(left, right)) {
+        result.dataDifferences.add(prefix + d);
+      }
+      return;
+    }
+    ErrorComparison c = ErrorComparator.compare(left, right, v -> returnedLabel + v);
+    for (String d : c.metadataDiffs) {
+      result.metadataDifferences.add(prefix + d);
+    }
+    for (String d : c.dataDiffs) {
+      result.dataDifferences.add(prefix + d);
+    }
+  }
+
+  /**
+   * Returns a FLAT list of diff strings for a pair of outcomes where at least one threw (empty when
+   * they agree, per the active mode). {@code returnedLabel} prefixes how a returned value is
+   * rendered in a one-sided diff (e.g. {@code "update count "}).
+   *
+   * <p>NOTE: this flattens metadata and data diffs together, so a caller that dumps the result into
+   * {@code ComparisonResult.dataDifferences} will mis-file one-sided diffs (which belong in {@code
+   * metadataDifferences}) and drop them from {@code csvSummary()}. Prefer {@link #foldInto}, which
+   * preserves the split. Retained for callers that accumulate into their own flat list (the DML and
+   * Volume positive suites).
    */
   public static List<String> compare(
       CapturedOutcome left, CapturedOutcome right, String returnedLabel) {

src/test/java/com/databricks/jdbc/comparator/suite/NegativeConnectionStateProvider.java

@@ -0,0 +1,132 @@
+package com.databricks.jdbc.comparator.suite;
+
+import com.databricks.jdbc.comparator.ComparisonResult;
+import com.databricks.jdbc.comparator.config.ConnectionFactory;
+import com.databricks.jdbc.comparator.error.CapturedOutcome;
+import com.databricks.jdbc.comparator.error.Captures;
+import com.databricks.jdbc.comparator.error.ErrorDiffs;
+import java.sql.Connection;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * Negative connection-state cases — {@code setCatalog}/{@code setSchema} to a non-existent target,
+ * {@code getCatalog}/{@code getSchema} after a failed namespace switch, {@code setClientInfo} with
+ * an invalid value, and {@code USE} of a non-existent catalog/schema. Compares each endpoint's
+ * error behavior via {@link ErrorDiffs}.
+ *
+ * <p>Isolation: these cases mutate session state, so the suite overrides the {@link
+ * ConnectionFactory} overload and opens its OWN fresh connections per side (closed in a {@code
+ * finally}). It never touches the shared {@code conn1}/{@code conn2}, so it cannot poison other
+ * suites. A fresh pair is opened per case so state changes don't leak between cases either.
+ */
+public class NegativeConnectionStateProvider implements SuiteProvider {
+
+  /** An action on a single connection that may throw; returns a value on success. */
+  @FunctionalInterface
+  private interface ConnAction {
+    Object run(Connection conn) throws Exception;
+  }
+
+  private static final class Case {
+    final String description;
+    final ConnAction action;
+
+    Case(String description, ConnAction action) {
+      this.description = description;
+      this.action = action;
+    }
+  }
+
+  private static final List<Case> CASES =
+      Arrays.asList(
+          new Case(
+              "setCatalog to a non-existent catalog",
+              conn -> {
+                conn.setCatalog("__no_such_catalog__");
+                return conn.getCatalog();
+              }),
+          new Case(
+              "setSchema to a non-existent schema",
+              conn -> {
+                conn.setSchema("__no_such_schema__");
+                return conn.getSchema();
+              }),
+          new Case(
+              "setClientInfo with an invalid session conf value",
+              conn -> {
+                conn.setClientInfo("statement_timeout", "not_a_number");
+                return "ok";
+              }),
+          new Case(
+              "USE a non-existent catalog",
+              conn -> {
+                try (Statement s = conn.createStatement()) {
+                  return s.execute("USE CATALOG __no_such_catalog__");
+                }
+              }),
+          new Case(
+              "USE a non-existent schema",
+              conn -> {
+                try (Statement s = conn.createStatement()) {
+                  return s.execute("USE SCHEMA __no_such_schema__");
+                }
+              }));
+
+  @Override
+  public List<TestCase> getTestCases() {
+    return CASES.stream()
+        .map(c -> new TestCase(c.description, c.description))
+        .collect(Collectors.toList());
+  }
+
+  /** Not used — this suite requires the ConnectionFactory overload below. */
+  @Override
+  public ComparisonResult execute(
+      Connection conn1, Connection conn2, TestCase testCase, String label) throws Exception {
+    throw new UnsupportedOperationException(
+        "NEGATIVE_CONNECTION_STATE requires dedicated connections; use the ConnectionFactory "
+            + "overload");
+  }
+
+  @Override
+  public ComparisonResult execute(
+      Connection conn1,
+      Connection conn2,
+      ConnectionFactory factory,
+      TestCase testCase,
+      String label)
+      throws Exception {
+    Case c =
+        CASES.stream()
+            .filter(x -> x.description.equals(testCase.getIdentifier()))
+            .findFirst()
+            .orElseThrow(
+                () -> new IllegalArgumentException("Unknown case: " + testCase.getIdentifier()));
+
+    // Fresh, dedicated connections per side — opening them is bookkeeping (outside capture); a
+    // failure to connect propagates and fails the test loudly (harness problem, not a driver diff).
+    Connection left = factory.openFresh("LEFT");
+    Connection right = factory.openFresh("RIGHT");
+    try {
+      CapturedOutcome lo = Captures.capture(() -> c.action.run(left));
+      CapturedOutcome ro = Captures.capture(() -> c.action.run(right));
+      ComparisonResult result = new ComparisonResult(label, c.description, testCase.getArgs());
+      ErrorDiffs.foldInto(result, lo, ro, "value ", "");
+      return result;
+    } finally {
+      closeQuietly(left);
+      closeQuietly(right);
+    }
+  }
+
+  private static void closeQuietly(Connection conn) {
+    try {
+      if (conn != null) conn.close();
+    } catch (Exception ignored) {
+      // best-effort
+    }
+  }
+}

src/test/java/com/databricks/jdbc/comparator/suite/NegativeStatementBatchProvider.java

@@ -92,9 +92,7 @@ public ComparisonResult execute(
       CapturedOutcome right = Captures.capture(() -> s2.executeBatch());
 
       ComparisonResult result = new ComparisonResult(label, c.description, testCase.getArgs());
-      for (String d : ErrorDiffs.compare(left, right, "batch counts ")) {
-        result.dataDifferences.add(d);
-      }
+      ErrorDiffs.foldInto(result, left, right, "batch counts ", "");
       // When both threw BatchUpdateException, also compare the per-element update counts.
       String counts = compareUpdateCounts(left, right);
       if (counts != null) {