Add kill switch for SQL Server transaction isolation

plugin/trino-sqlserver/pom.xml

@@ -28,6 +28,11 @@
             <artifactId>trino-matching</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>io.airlift</groupId>
+            <artifactId>configuration</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>io.airlift</groupId>
             <artifactId>log</artifactId>

plugin/trino-sqlserver/src/main/java/io/trino/plugin/sqlserver/SqlServerClient.java

@@ -130,6 +130,7 @@
 import static java.lang.String.join;
 import static java.math.RoundingMode.UNNECESSARY;
 import static java.time.Duration.ofMinutes;
+import static java.util.Objects.requireNonNull;
 import static java.util.stream.Collectors.joining;
 
 public class SqlServerClient
@@ -140,6 +141,7 @@ public class SqlServerClient
 
     private static final Joiner DOT_JOINER = Joiner.on(".");
 
+    private final boolean snapshotIsolationDisabled;
     private final Cache<SnapshotIsolationEnabledCacheKey, Boolean> snapshotIsolationEnabled = CacheBuilder.newBuilder()
             .maximumSize(1)
             .expireAfterWrite(ofMinutes(5))
@@ -150,10 +152,13 @@ public class SqlServerClient
     private static final int MAX_SUPPORTED_TEMPORAL_PRECISION = 7;
 
     @Inject
-    public SqlServerClient(BaseJdbcConfig config, ConnectionFactory connectionFactory)
+    public SqlServerClient(BaseJdbcConfig config, SqlServerConfig sqlServerConfig, ConnectionFactory connectionFactory)
     {
         super(config, "\"", connectionFactory);
 
+        requireNonNull(sqlServerConfig, "sqlServerConfig is null");
+        snapshotIsolationDisabled = sqlServerConfig.isSnapshotIsolationDisabled();
+
         JdbcTypeHandle bigintTypeHandle = new JdbcTypeHandle(Types.BIGINT, Optional.of("bigint"), Optional.empty(), Optional.empty(), Optional.empty(), Optional.empty());
         this.aggregateFunctionRewriter = new AggregateFunctionRewriter(
                 this::quoted,
@@ -538,6 +543,9 @@ public Connection getConnection(ConnectorSession session, JdbcSplit split)
     private Connection configureConnectionTransactionIsolation(Connection connection)
             throws SQLException
     {
+        if (snapshotIsolationDisabled) {
+            return connection;
+        }
         try {
             if (hasSnapshotIsolationEnabled(connection)) {
                 // SQL Server's READ COMMITTED + SNAPSHOT ISOLATION is equivalent to ordinary READ COMMITTED in e.g. Oracle, PostgreSQL.

plugin/trino-sqlserver/src/main/java/io/trino/plugin/sqlserver/SqlServerClientModule.java

@@ -29,6 +29,7 @@
 import io.trino.plugin.jdbc.credential.CredentialProvider;
 
 import static com.google.inject.multibindings.OptionalBinder.newOptionalBinder;
+import static io.airlift.configuration.ConfigBinder.configBinder;
 import static io.trino.plugin.jdbc.JdbcModule.bindTablePropertiesProvider;
 import static io.trino.plugin.sqlserver.SqlServerClient.SQL_SERVER_MAX_LIST_EXPRESSIONS;
 
@@ -38,6 +39,7 @@ public class SqlServerClientModule
     @Override
     public void configure(Binder binder)
     {
+        configBinder(binder).bindConfig(SqlServerConfig.class);
         binder.bind(JdbcClient.class).annotatedWith(ForBaseJdbc.class).to(SqlServerClient.class).in(Scopes.SINGLETON);
         bindTablePropertiesProvider(binder, SqlServerTableProperties.class);
         newOptionalBinder(binder, Key.get(int.class, MaxDomainCompactionThreshold.class)).setBinding().toInstance(SQL_SERVER_MAX_LIST_EXPRESSIONS);

plugin/trino-sqlserver/src/main/java/io/trino/plugin/sqlserver/SqlServerConfig.java

@@ -0,0 +1,35 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.trino.plugin.sqlserver;
+
+import io.airlift.configuration.Config;
+import io.airlift.configuration.ConfigDescription;
+
+public class SqlServerConfig
+{
+    private boolean snapshotIsolationDisabled;
+
+    public boolean isSnapshotIsolationDisabled()
+    {
+        return snapshotIsolationDisabled;
+    }
+
+    @Config("sqlserver.snapshot-isolation.disabled")
+    @ConfigDescription("Disables automatic use of snapshot isolation for transactions issued by Trino in SQL Server")
+    public SqlServerConfig setSnapshotIsolationDisabled(boolean snapshotIsolationDisabled)
+    {
+        this.snapshotIsolationDisabled = snapshotIsolationDisabled;
+        return this;
+    }
+}

plugin/trino-sqlserver/src/test/java/io/trino/plugin/sqlserver/TestSqlServerClient.java

@@ -56,6 +56,7 @@ public class TestSqlServerClient
 
     private static final JdbcClient JDBC_CLIENT = new SqlServerClient(
             new BaseJdbcConfig(),
+            new SqlServerConfig(),
             session -> {
                 throw new UnsupportedOperationException();
             });

plugin/trino-sqlserver/src/test/java/io/trino/plugin/sqlserver/TestSqlServerConfig.java

@@ -0,0 +1,46 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.trino.plugin.sqlserver;
+
+import com.google.common.collect.ImmutableMap;
+import org.testng.annotations.Test;
+
+import java.util.Map;
+
+import static io.airlift.configuration.testing.ConfigAssertions.assertFullMapping;
+import static io.airlift.configuration.testing.ConfigAssertions.assertRecordedDefaults;
+import static io.airlift.configuration.testing.ConfigAssertions.recordDefaults;
+
+public class TestSqlServerConfig
+{
+    @Test
+    public void testDefaults()
+    {
+        assertRecordedDefaults(recordDefaults(SqlServerConfig.class)
+                .setSnapshotIsolationDisabled(false));
+    }
+
+    @Test
+    public void testExplicitPropertyMappings()
+    {
+        Map<String, String> properties = new ImmutableMap.Builder<String, String>()
+                .put("sqlserver.snapshot-isolation.disabled", "true")
+                .build();
+
+        SqlServerConfig expected = new SqlServerConfig()
+                .setSnapshotIsolationDisabled(true);
+
+        assertFullMapping(properties, expected);
+    }
+}