Fault tolerant execution for SqlServer connector

.github/workflows/ci.yml

@@ -481,6 +481,7 @@ jobs:
             - { modules: testing/trino-faulttolerant-tests, profile: test-fault-tolerant-iceberg }
             - { modules: testing/trino-faulttolerant-tests, profile: test-fault-tolerant-postgresql }
             - { modules: testing/trino-faulttolerant-tests, profile: test-fault-tolerant-mysql }
+            - { modules: testing/trino-faulttolerant-tests, profile: test-fault-tolerant-sqlserver }
             - { modules: testing/trino-tests }
           EOF
           ./.github/bin/build-matrix-from-impacted.py -v -i gib-impacted.log -m .github/test-matrix.yaml -o matrix.json

docs/src/main/sphinx/admin/fault-tolerant-execution.rst

@@ -50,6 +50,7 @@ depending on the desired :ref:`retry policy <fte-retry-policy>`.
     * :doc:`/connector/iceberg`
     * :doc:`/connector/mysql`
     * :doc:`/connector/postgresql`
+    * :doc:`/connector/sqlserver`
 
 The following configuration properties control the behavior of fault-tolerant
 execution on a Trino cluster:

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

@@ -208,7 +208,7 @@ public SqlServerClient(
             QueryBuilder queryBuilder,
             IdentifierMapping identifierMapping)
     {
-        super(config, "\"", connectionFactory, queryBuilder, identifierMapping);
+        super(config, "\"", connectionFactory, queryBuilder, identifierMapping, true);
 
         this.statisticsEnabled = statisticsConfig.isEnabled();
 

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

@@ -24,6 +24,7 @@
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.function.Consumer;
 
 import static io.airlift.testing.Closeables.closeAllSuppress;
 import static io.trino.plugin.tpch.TpchMetadata.TINY_SCHEMA_NAME;
@@ -46,9 +47,23 @@ public static QueryRunner createSqlServerQueryRunner(
             Map<String, String> connectorProperties,
             Iterable<TpchTable<?>> tables)
             throws Exception
+    {
+        return createSqlServerQueryRunner(testingSqlServer, extraProperties, Map.of(), connectorProperties, tables, runner -> {});
+    }
+
+    public static QueryRunner createSqlServerQueryRunner(
+            TestingSqlServer testingSqlServer,
+            Map<String, String> extraProperties,
+            Map<String, String> coordinatorProperties,
+            Map<String, String> connectorProperties,
+            Iterable<TpchTable<?>> tables,
+            Consumer<QueryRunner> moreSetup)
+            throws Exception
     {
         DistributedQueryRunner queryRunner = DistributedQueryRunner.builder(createSession(testingSqlServer.getUsername()))
                 .setExtraProperties(extraProperties)
+                .setCoordinatorProperties(coordinatorProperties)
+                .setAdditionalSetup(moreSetup)
                 .build();
         try {
             queryRunner.installPlugin(new TpchPlugin());

pom.xml

@@ -604,6 +604,13 @@
                 <version>${project.version}</version>
             </dependency>
 
+            <dependency>
+                <groupId>io.trino</groupId>
+                <artifactId>trino-sqlserver</artifactId>
+                <type>test-jar</type>
+                <version>${project.version}</version>
+            </dependency>
+
             <dependency>
                 <groupId>io.trino</groupId>
                 <artifactId>trino-testing</artifactId>

testing/trino-faulttolerant-tests/pom.xml

@@ -304,6 +304,19 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>io.trino</groupId>
+            <artifactId>trino-sqlserver</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>io.trino</groupId>
+            <artifactId>trino-sqlserver</artifactId>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>io.trino</groupId>
             <artifactId>trino-testing</artifactId>
@@ -390,6 +403,12 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>mssqlserver</artifactId>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>org.testcontainers</groupId>
             <artifactId>mysql</artifactId>
@@ -452,6 +471,7 @@
                                 <exclude>**/io/trino/faulttolerant/iceberg/Test*.java</exclude>
                                 <exclude>**/io/trino/faulttolerant/mysql/Test*.java</exclude>
                                 <exclude>**/io/trino/faulttolerant/postgresql/Test*.java</exclude>
+                                <exclude>**/io/trino/faulttolerant/sqlserver/Test*.java</exclude>
                             </excludes>
                         </configuration>
                     </plugin>
@@ -570,5 +590,23 @@
                 </plugins>
             </build>
         </profile>
+
+        <profile>
+            <id>test-fault-tolerant-sqlserver</id>
+            <build>
+                <plugins>
+                    <plugin>
+                        <groupId>org.apache.maven.plugins</groupId>
+                        <artifactId>maven-surefire-plugin</artifactId>
+                        <configuration>
+                            <threadCount>4</threadCount>
+                            <includes>
+                                <include>**/io/trino/faulttolerant/sqlserver/Test*.java</include>
+                            </includes>
+                        </configuration>
+                    </plugin>
+                </plugins>
+            </build>
+        </profile>
     </profiles>
 </project>

testing/trino-faulttolerant-tests/src/test/java/io/trino/faulttolerant/sqlserver/BaseSqlServerFailureRecoveryTest.java

@@ -0,0 +1,57 @@
+/*
+ * 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.faulttolerant.sqlserver;
+
+import com.google.common.collect.ImmutableMap;
+import io.trino.faulttolerant.jdbc.BaseJdbcFailureRecoveryTest;
+import io.trino.operator.RetryPolicy;
+import io.trino.plugin.exchange.filesystem.FileSystemExchangePlugin;
+import io.trino.plugin.sqlserver.TestingSqlServer;
+import io.trino.testing.QueryRunner;
+import io.trino.tpch.TpchTable;
+
+import java.util.List;
+import java.util.Map;
+
+import static io.trino.plugin.sqlserver.SqlServerQueryRunner.createSqlServerQueryRunner;
+
+public abstract class BaseSqlServerFailureRecoveryTest
+        extends BaseJdbcFailureRecoveryTest
+{
+    public BaseSqlServerFailureRecoveryTest(RetryPolicy retryPolicy)
+    {
+        super(retryPolicy);
+    }
+
+    @Override
+    protected QueryRunner createQueryRunner(
+            List<TpchTable<?>> requiredTpchTables,
+            Map<String, String> configProperties,
+            Map<String, String> coordinatorProperties)
+            throws Exception
+    {
+        return createSqlServerQueryRunner(
+                closeAfterClass(new TestingSqlServer()),
+                configProperties,
+                coordinatorProperties,
+                Map.of(),
+                requiredTpchTables,
+                runner -> {
+                    runner.installPlugin(new FileSystemExchangePlugin());
+                    runner.loadExchangeManager("filesystem", ImmutableMap.<String, String>builder()
+                            .put("exchange.base-directories", System.getProperty("java.io.tmpdir") + "/trino-local-file-system-exchange-manager")
+                            .buildOrThrow());
+                });
+    }
+}

testing/trino-faulttolerant-tests/src/test/java/io/trino/faulttolerant/sqlserver/TestSqlServerQueryFailureRecoveryTest.java

@@ -0,0 +1,25 @@
+/*
+ * 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.faulttolerant.sqlserver;
+
+import io.trino.operator.RetryPolicy;
+
+public class TestSqlServerQueryFailureRecoveryTest
+        extends BaseSqlServerFailureRecoveryTest
+{
+    public TestSqlServerQueryFailureRecoveryTest()
+    {
+        super(RetryPolicy.QUERY);
+    }
+}

testing/trino-faulttolerant-tests/src/test/java/io/trino/faulttolerant/sqlserver/TestSqlServerTaskFailureRecoveryTest.java

@@ -0,0 +1,25 @@
+/*
+ * 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.faulttolerant.sqlserver;
+
+import io.trino.operator.RetryPolicy;
+
+public class TestSqlServerTaskFailureRecoveryTest
+        extends BaseSqlServerFailureRecoveryTest
+{
+    public TestSqlServerTaskFailureRecoveryTest()
+    {
+        super(RetryPolicy.TASK);
+    }
+}