Fix rollback in auto-commit mode and handle numeric autocommit responses (#1206)

## Summary
- **rollback() now throws** `DatabricksTransactionException` when called
while the connection is in auto-commit mode (no active transaction),
aligning behavior with the JDBC spec
- **fetchAutoCommitStateFromServer()** now accepts both `"1"`/`"0"` and
`"true"`/`"false"` responses from the `SET AUTOCOMMIT` query, since
different server implementations return different formats
- Updated E2E and unit tests to reflect the corrected behavior

## Context
1. `connection.rollback()` in auto-commit mode should throw an
exception, not silently succeed
2. `SET AUTOCOMMIT` query returns `"1"`/`"0"` rather than
`"true"`/`"false"` depending on the server

## Test plan
- [x] Unit tests pass: `DatabricksConnectionTest` (42/42 pass)
- [ ] E2E tests: `TransactionTests` and
`ExplicitTransactionStatementTests` against staging warehouse

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Signed-off-by: Vikrant Puppala <vikrant.puppala@databricks.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: vikrantpuppala

NEXT_CHANGELOG.md

@@ -7,6 +7,8 @@
 ### Updated
 
 ### Fixed
+- Fixed `rollback()` to throw `SQLException` when called in auto-commit mode (no active transaction), aligning with JDBC spec. Previously it silently sent a ROLLBACK command to the server.
+- Fixed `fetchAutoCommitStateFromServer()` to accept both `"1"`/`"0"` and `"true"`/`"false"` responses from `SET AUTOCOMMIT` query, since different server implementations return different formats.
 
 ---
 *Note: When making changes, please add your change under the appropriate section

src/main/java/com/databricks/jdbc/api/impl/DatabricksConnection.java

@@ -252,13 +252,22 @@ private boolean fetchAutoCommitStateFromServer() throws SQLException {
       ResultSet rs = statement.executeQuery("SET AUTOCOMMIT");
 
       if (rs.next()) {
-        // The result should contain the value = "true" or "false"
+        // The result may contain "true"/"false" or "1"/"0" depending on the server
         String value = rs.getString(1); // Column 1: value
 
         LOGGER.debug(
             "Fetched autoCommit state from server: value={}. Updating session cache.", value);
 
-        boolean autoCommitState = "true".equalsIgnoreCase(value);
+        boolean autoCommitState;
+        if ("true".equalsIgnoreCase(value) || "1".equals(value)) {
+          autoCommitState = true;
+        } else if ("false".equalsIgnoreCase(value) || "0".equals(value)) {
+          autoCommitState = false;
+        } else {
+          throw new DatabricksSQLException(
+              "Unexpected autoCommit value from server: " + value,
+              DatabricksDriverErrorCode.TRANSACTION_SET_AUTOCOMMIT_ERROR);
+        }
 
         // Update the session cache with the server value
         session.setAutoCommit(autoCommitState);
@@ -348,24 +357,19 @@ public void commit() throws SQLException {
    *
    * <ul>
    *   <li>Rolls back the current transaction
-   *   <li>A new transaction begins automatically (per autocommit design)
+   *   <li>A new transaction begins automatically
    * </ul>
    *
    * <p>When auto-commit is TRUE:
    *
    * <ul>
-   *   <li>ROLLBACK is a safe no-op (does not throw exception)
-   *   <li>This is more forgiving than COMMIT, which throws an exception when there's no active
-   *       transaction
+   *   <li>This operation throws {@link DatabricksTransactionException} (if ignoreTransactions flag
+   *       is not set)
    * </ul>
    *
-   * <p><b>Note:</b> ROLLBACK is designed to be safe to call even when there is no active
-   * transaction. It can be used to recover from error states without needing to check transaction
-   * status first.
-   *
    * @throws DatabricksSQLException if the connection is closed
-   * @throws DatabricksTransactionException for transaction-specific errors (rare - ROLLBACK is
-   *     typically very forgiving)
+   * @throws DatabricksTransactionException for transaction-specific errors such as
+   *     MULTI_STATEMENT_TRANSACTION_NO_ACTIVE_TRANSACTION or calling rollback in auto-commit mode
    * @see #setAutoCommit(boolean)
    * @see #commit()
    */
@@ -382,13 +386,20 @@ public void rollback() throws SQLException {
       return;
     }
 
+    // Rollback is not valid when auto-commit is enabled (no active transaction)
+    if (getAutoCommit()) {
+      throw new DatabricksTransactionException(
+          "Cannot use rollback while Connection is in auto-commit mode.",
+          new SQLException("Cannot use rollback while Connection is in auto-commit mode."),
+          DatabricksDriverErrorCode.TRANSACTION_ROLLBACK_ERROR);
+    }
+
     // Execute ROLLBACK command
     Statement statement = null;
     try {
       statement = createStatement();
       statement.execute("ROLLBACK");
       // Note: Server auto-starts new transaction if autocommit=false
-      // Note: ROLLBACK is more forgiving - typically succeeds even on unexpected states
 
     } catch (SQLException e) {
       LOGGER.error(e, "Error {} while rolling back transaction", e.getMessage());

src/test/java/com/databricks/jdbc/api/impl/DatabricksConnectionTest.java

@@ -18,6 +18,7 @@
 import com.databricks.jdbc.exception.DatabricksSQLException;
 import com.databricks.jdbc.exception.DatabricksSQLFeatureNotImplementedException;
 import com.databricks.jdbc.exception.DatabricksSQLFeatureNotSupportedException;
+import com.databricks.jdbc.exception.DatabricksTransactionException;
 import com.databricks.jdbc.model.telemetry.enums.DatabricksDriverErrorCode;
 import java.sql.*;
 import java.util.*;
@@ -824,12 +825,42 @@ public void testRollbackSuccess() throws SQLException {
     DatabricksConnection spyConnection = spy(connection);
     DatabricksStatement mockStatement = mock(DatabricksStatement.class);
     doReturn(mockStatement).when(spyConnection).createStatement();
+    when(mockStatement.execute("SET AUTOCOMMIT = FALSE")).thenReturn(true);
     when(mockStatement.execute("ROLLBACK")).thenReturn(true);
 
+    // Must set autocommit to false first — rollback is not valid in auto-commit mode
+    spyConnection.setAutoCommit(false);
     spyConnection.rollback();
 
     verify(mockStatement).execute("ROLLBACK");
-    verify(mockStatement).close();
+    verify(mockStatement, atLeast(2)).close();
+
+    spyConnection.close();
+  }
+
+  @Test
+  public void testRollbackThrowsWhenAutoCommitEnabled() throws SQLException {
+    when(databricksClient.createSession(
+            new Warehouse(WAREHOUSE_ID), CATALOG, SCHEMA, new HashMap<>()))
+        .thenReturn(IMMUTABLE_SESSION_INFO);
+    connection = new DatabricksConnection(transactionsEnabledContext, databricksClient);
+    connection.open();
+
+    DatabricksConnection spyConnection = spy(connection);
+
+    // Auto-commit is true by default — rollback should throw
+    assertTrue(spyConnection.getAutoCommit());
+
+    DatabricksTransactionException thrown =
+        assertThrows(DatabricksTransactionException.class, spyConnection::rollback);
+
+    assertTrue(
+        thrown.getMessage().contains("auto-commit"),
+        "Exception should indicate rollback is not valid in auto-commit mode. Got: "
+            + thrown.getMessage());
+
+    // Connection should still be usable
+    assertFalse(spyConnection.isClosed());
 
     spyConnection.close();
   }
@@ -865,6 +896,10 @@ public void testRollbackWithServerError() throws SQLException {
     DatabricksConnection spyConnection = spy(connection);
     DatabricksStatement mockStatement = mock(DatabricksStatement.class);
     doReturn(mockStatement).when(spyConnection).createStatement();
+    when(mockStatement.execute("SET AUTOCOMMIT = FALSE")).thenReturn(true);
+
+    // Must set autocommit to false first — rollback is not valid in auto-commit mode
+    spyConnection.setAutoCommit(false);
 
     SQLException serverError = new SQLException("Unexpected rollback error", "HY000", 99999);
     when(mockStatement.execute("ROLLBACK")).thenThrow(serverError);
@@ -875,7 +910,7 @@ public void testRollbackWithServerError() throws SQLException {
     assertEquals("HY000", thrown.getSQLState());
     assertEquals(99999, thrown.getErrorCode()); // Vendor code preserved
 
-    verify(mockStatement).close();
+    verify(mockStatement, atLeast(2)).close();
 
     spyConnection.close();
   }
@@ -986,6 +1021,68 @@ public void testGetAutoCommitWithFetchFromServerEnabled_ReturnsFalse() throws SQ
     spyConnection.close();
   }
 
+  @Test
+  public void testGetAutoCommitWithFetchFromServerEnabled_ReturnsNumeric1() throws SQLException {
+    // Server may return "1" instead of "true" for autocommit enabled
+    String urlWithFetch = CATALOG_SCHEMA_JDBC_URL + ";FetchAutoCommitFromServer=1";
+    IDatabricksConnectionContext contextWithFetch =
+        DatabricksConnectionContext.parse(urlWithFetch, new Properties());
+
+    when(databricksClient.createSession(
+            new Warehouse(WAREHOUSE_ID), CATALOG, SCHEMA, new HashMap<>()))
+        .thenReturn(IMMUTABLE_SESSION_INFO);
+    connection = new DatabricksConnection(contextWithFetch, databricksClient);
+    connection.open();
+
+    DatabricksConnection spyConnection = spy(connection);
+    DatabricksStatement mockStatement = mock(DatabricksStatement.class);
+    ResultSet mockResultSet = mock(ResultSet.class);
+
+    doReturn(mockStatement).when(spyConnection).createStatement();
+    when(mockStatement.executeQuery("SET AUTOCOMMIT")).thenReturn(mockResultSet);
+    when(mockResultSet.next()).thenReturn(true);
+    when(mockResultSet.getString(1)).thenReturn("1"); // Server returns "1"
+
+    boolean result = spyConnection.getAutoCommit();
+
+    verify(mockStatement).executeQuery("SET AUTOCOMMIT");
+    assertTrue(result);
+    assertTrue(spyConnection.getSession().getAutoCommit());
+
+    spyConnection.close();
+  }
+
+  @Test
+  public void testGetAutoCommitWithFetchFromServerEnabled_ReturnsNumeric0() throws SQLException {
+    // Server may return "0" instead of "false" for autocommit disabled
+    String urlWithFetch = CATALOG_SCHEMA_JDBC_URL + ";FetchAutoCommitFromServer=1";
+    IDatabricksConnectionContext contextWithFetch =
+        DatabricksConnectionContext.parse(urlWithFetch, new Properties());
+
+    when(databricksClient.createSession(
+            new Warehouse(WAREHOUSE_ID), CATALOG, SCHEMA, new HashMap<>()))
+        .thenReturn(IMMUTABLE_SESSION_INFO);
+    connection = new DatabricksConnection(contextWithFetch, databricksClient);
+    connection.open();
+
+    DatabricksConnection spyConnection = spy(connection);
+    DatabricksStatement mockStatement = mock(DatabricksStatement.class);
+    ResultSet mockResultSet = mock(ResultSet.class);
+
+    doReturn(mockStatement).when(spyConnection).createStatement();
+    when(mockStatement.executeQuery("SET AUTOCOMMIT")).thenReturn(mockResultSet);
+    when(mockResultSet.next()).thenReturn(true);
+    when(mockResultSet.getString(1)).thenReturn("0"); // Server returns "0"
+
+    boolean result = spyConnection.getAutoCommit();
+
+    verify(mockStatement).executeQuery("SET AUTOCOMMIT");
+    assertFalse(result);
+    assertFalse(spyConnection.getSession().getAutoCommit());
+
+    spyConnection.close();
+  }
+
   @Test
   public void testGetAutoCommitWithFetchFromServerEnabled_ServerQueryFails() throws SQLException {
     // Create connection context with FetchAutoCommitFromServer=1

src/test/java/com/databricks/jdbc/integration/e2e/ExplicitTransactionStatementTests.java

@@ -475,8 +475,11 @@ void testSetAutocommitWithoutValue() throws SQLException {
     ResultSet rs1 = stmt.executeQuery("SET AUTOCOMMIT");
     assertTrue(rs1.next(), "SET AUTOCOMMIT should return a result");
     // The result should indicate autocommit is TRUE (default)
+    // Note: The server may return "1"/"0" instead of "true"/"false"
     String value1 = rs1.getString(1);
-    assertTrue(Boolean.parseBoolean(value1), "Should return a value");
+    assertTrue(
+        "true".equalsIgnoreCase(value1) || "1".equals(value1),
+        "Default autocommit should be true/1. Got: " + value1);
     rs1.close();
 
     // Set autocommit to FALSE

src/test/java/com/databricks/jdbc/integration/e2e/TransactionTests.java

@@ -344,16 +344,25 @@ void testCommitWithoutActiveTransaction() throws SQLException {
   }
 
   @Test
-  @DisplayName("Should safely handle ROLLBACK without active transaction (no-op)")
-  void testRollbackWithoutActiveTransactionIsNoOp() throws SQLException {
+  @DisplayName("Should throw exception when rolling back without active transaction")
+  void testRollbackWithoutActiveTransactionThrows() throws SQLException {
     // With autoCommit=true (no active transaction)
     assertTrue(connection.getAutoCommit());
 
-    // ROLLBACK is more forgiving than COMMIT - it should succeed as a no-op
-    // when there's no active transaction
-    assertDoesNotThrow(
-        () -> connection.rollback(),
-        "ROLLBACK should be a safe no-op when autocommit=true (no active transaction)");
+    // ROLLBACK should throw an exception when there is no active transaction
+    // (connection is in auto-commit mode)
+    DatabricksTransactionException exception =
+        assertThrows(
+            DatabricksTransactionException.class,
+            () -> connection.rollback(),
+            "ROLLBACK should throw exception when autocommit=true (no active transaction)");
+
+    assertTrue(
+        exception.getMessage().contains("auto-commit")
+            || exception.getMessage().contains("rollback")
+            || exception.getMessage().contains("No active transaction"),
+        "Exception message should indicate rollback is not valid in auto-commit mode. Got: "
+            + exception.getMessage());
 
     // Verify connection is still usable
     assertTrue(connection.getAutoCommit());