Added connection param of ignore transactions that allows users to mu… (#1026)

## Description
<!-- Provide a brief summary of the changes made and the issue they aim
to address.-->
Allows user to mute the exceptions for transaction related methods.
Earlier the driver was throwing an exception for transaction related
methods. Now the users can use the parameter to ignore these exceptions
for transaction methods.
## Testing
<!-- Describe how the changes have been tested-->
Unit tests added.
## Additional Notes to the Reviewer
<!-- Share any additional context or insights that may help the reviewer
understand the changes better. This could include challenges faced,
limitations, or compromises made during the development process.
Also, mention any areas of the code that you would like the reviewer to
focus on specifically. -->
[PECOBLR-931](https://databricks.atlassian.net/browse/PECOBLR-931)


[PECOBLR-931]:
https://databricks.atlassian.net/browse/PECOBLR-931?atlOrigin=eyJpIjoiNWRkNTljNzYxNjVmNDY3MDlhMDU5Y2ZhYzA5YTRkZjUiLCJwIjoiZ2l0aHViLWNvbS1KU1cifQ
Fixes #1005

Author: msrathore-db

NEXT_CHANGELOG.md

@@ -7,6 +7,7 @@
 ### Updated
 - Telemetry data is now captured more efficiently and consistently due to enhancements in the log and connection close flush logic.
 - Updated Databricks SDK version to v0.65.0 (This is to fix OAuthClient to properly encode complex query parameters.)
+- Added IgnoreTransactions connection parameter to silently ignore transaction method calls.
 
 ### Fixed
 - Fixed state leaking issue in thrift client.

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

@@ -120,9 +120,11 @@ public String nativeSQL(String sql) throws SQLException {
   @Override
   public void setAutoCommit(boolean autoCommit) throws SQLException {
     if (!autoCommit) {
-      throw new DatabricksSQLFeatureNotSupportedException(
-          "In Databricks OSS JDBC, every SQL statement is committed immediately upon execution."
-              + " Setting autoCommit=false is not supported.");
+      if (!connectionContext.getIgnoreTransactions()) {
+        throw new DatabricksSQLFeatureNotSupportedException(
+            "In Databricks OSS JDBC, every SQL statement is committed immediately upon execution."
+                + " Setting autoCommit=false is not supported.");
+      }
     }
   }
 
@@ -136,15 +138,19 @@ public boolean getAutoCommit() throws SQLException {
   @Override
   public void commit() throws SQLException {
     LOGGER.debug("public void commit()");
-    throw new DatabricksSQLFeatureNotImplementedException(
-        "Not implemented in DatabricksConnection - commit()");
+    if (!connectionContext.getIgnoreTransactions()) {
+      throw new DatabricksSQLFeatureNotImplementedException(
+          "Not implemented in DatabricksConnection - commit()");
+    }
   }
 
   @Override
   public void rollback() throws SQLException {
     LOGGER.debug("public void rollback()");
-    throw new DatabricksSQLFeatureNotImplementedException(
-        "Not implemented in DatabricksConnection - rollback()");
+    if (!connectionContext.getIgnoreTransactions()) {
+      throw new DatabricksSQLFeatureNotImplementedException(
+          "Not implemented in DatabricksConnection - rollback()");
+    }
   }
 
   @Override
@@ -299,28 +305,39 @@ public int getHoldability() throws SQLException {
   @Override
   public Savepoint setSavepoint() throws SQLException {
     LOGGER.debug("public Savepoint setSavepoint()");
-    throw new DatabricksSQLFeatureNotImplementedException(
-        "Not implemented in DatabricksConnection - setSavepoint()");
+    if (!connectionContext.getIgnoreTransactions()) {
+      throw new DatabricksSQLFeatureNotImplementedException(
+          "Not implemented in DatabricksConnection - setSavepoint()");
+    }
+    return null;
   }
 
   @Override
   public Savepoint setSavepoint(String name) throws SQLException {
-    throw new DatabricksSQLFeatureNotImplementedException(
-        "Not implemented in DatabricksConnection - setSavepoint(String name)");
+    LOGGER.debug("public Savepoint setSavepoint(String name = {})", name);
+    if (!connectionContext.getIgnoreTransactions()) {
+      throw new DatabricksSQLFeatureNotImplementedException(
+          "Not implemented in DatabricksConnection - setSavepoint(String name)");
+    }
+    return null;
   }
 
   @Override
   public void rollback(Savepoint savepoint) throws SQLException {
     LOGGER.debug("public void rollback(Savepoint savepoint)");
-    throw new DatabricksSQLFeatureNotImplementedException(
-        "Not implemented in DatabricksConnection - rollback(Savepoint savepoint)");
+    if (!connectionContext.getIgnoreTransactions()) {
+      throw new DatabricksSQLFeatureNotImplementedException(
+          "Not implemented in DatabricksConnection - rollback(Savepoint savepoint)");
+    }
   }
 
   @Override
   public void releaseSavepoint(Savepoint savepoint) throws SQLException {
     LOGGER.debug("public void releaseSavepoint(Savepoint savepoint)");
-    throw new DatabricksSQLFeatureNotImplementedException(
-        "Not implemented in DatabricksConnection - releaseSavepoint(Savepoint savepoint)");
+    if (!connectionContext.getIgnoreTransactions()) {
+      throw new DatabricksSQLFeatureNotImplementedException(
+          "Not implemented in DatabricksConnection - releaseSavepoint(Savepoint savepoint)");
+    }
   }
 
   @Override

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

@@ -1045,4 +1045,9 @@ public int getTelemetryFlushIntervalInMilliseconds() {
   public boolean isBatchedInsertsEnabled() {
     return getParameter(DatabricksJdbcUrlParams.ENABLE_BATCHED_INSERTS).equals("1");
   }
+
+  @Override
+  public boolean getIgnoreTransactions() {
+    return getParameter(DatabricksJdbcUrlParams.IGNORE_TRANSACTIONS, "0").equals("1");
+  }
 }

src/main/java/com/databricks/jdbc/api/internal/IDatabricksConnectionContext.java

@@ -368,4 +368,7 @@ public interface IDatabricksConnectionContext {
 
   /** Returns whether batched INSERT optimization is enabled */
   boolean isBatchedInsertsEnabled();
+
+  /** Returns whether transaction-related method calls should be ignored */
+  boolean getIgnoreTransactions();
 }

src/main/java/com/databricks/jdbc/common/DatabricksJdbcUrlParams.java

@@ -162,7 +162,8 @@ public enum DatabricksJdbcUrlParams {
   ENABLE_SQL_VALIDATION_FOR_IS_VALID(
       "EnableSQLValidationForIsValid",
       "Enable SQL query execution for connection validation in isValid() method",
-      "0");
+      "0"),
+  IGNORE_TRANSACTIONS("IgnoreTransactions", "Ignore transaction-related method calls", "0");
 
   private final String paramName;
   private final String defaultValue;

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

@@ -59,6 +59,8 @@ public class DatabricksConnectionTest {
           SESSION_CONFIGS.entrySet().stream()
               .map(e -> e.getKey() + "=" + e.getValue())
               .collect(Collectors.joining(";")));
+  private static final String IGNORE_TRANSACTIONS_JDBC_URL =
+      "jdbc:databricks://sample-host.18.azuredatabricks.net:4423/default;transportMode=http;ssl=1;AuthMech=3;httpPath=/sql/1.0/warehouses/99999999;IgnoreTransactions=1";
   private static final ImmutableSessionInfo IMMUTABLE_SESSION_INFO =
       ImmutableSessionInfo.builder().computeResource(warehouse).sessionId(SESSION_ID).build();
   @Mock DatabricksSdkClient databricksClient;
@@ -517,4 +519,49 @@ public void testIsValidWithSQLValidationEnabled() throws SQLException {
 
     connection.close();
   }
+
+  @Test
+  public void testIgnoreTransactionsDisabled() throws Exception {
+    when(databricksClient.createSession(
+            new Warehouse(WAREHOUSE_ID), null, DEFAULT_SCHEMA, new HashMap<>()))
+        .thenReturn(IMMUTABLE_SESSION_INFO);
+
+    IDatabricksConnectionContext context =
+        DatabricksConnectionContext.parse(JDBC_URL, new Properties());
+    connection = new DatabricksConnection(context, databricksClient);
+    connection.open();
+
+    assertFalse(context.getIgnoreTransactions());
+
+    assertThrows(
+        DatabricksSQLFeatureNotSupportedException.class, () -> connection.setAutoCommit(false));
+    assertThrows(DatabricksSQLFeatureNotImplementedException.class, () -> connection.commit());
+    assertThrows(
+        DatabricksSQLFeatureNotImplementedException.class, () -> connection.setSavepoint());
+
+    connection.close();
+  }
+
+  @Test
+  public void testIgnoreTransactionsEnabled() throws Exception {
+    when(databricksClient.createSession(
+            new Warehouse(WAREHOUSE_ID), null, DEFAULT_SCHEMA, new HashMap<>()))
+        .thenReturn(IMMUTABLE_SESSION_INFO);
+
+    IDatabricksConnectionContext context =
+        DatabricksConnectionContext.parse(IGNORE_TRANSACTIONS_JDBC_URL, new Properties());
+    connection = new DatabricksConnection(context, databricksClient);
+    connection.open();
+
+    assertTrue(context.getIgnoreTransactions());
+
+    assertDoesNotThrow(() -> connection.setAutoCommit(false));
+    assertDoesNotThrow(() -> connection.commit());
+    assertDoesNotThrow(() -> connection.rollback());
+
+    assertNull(connection.setSavepoint());
+    assertNull(connection.setSavepoint("test"));
+
+    connection.close();
+  }
 }