Retry Spark query when minReplication error happened

testing/trino-product-tests/src/main/java/io/trino/tests/product/hive/HiveProductTest.java

@@ -13,45 +13,13 @@
  */
 package io.trino.tests.product.hive;
 
-import com.google.common.base.Throwables;
-import io.airlift.log.Logger;
 import io.trino.tempto.ProductTest;
-import io.trino.tempto.query.QueryResult;
-import net.jodah.failsafe.RetryPolicy;
-import org.intellij.lang.annotations.Language;
 
 import javax.inject.Inject;
 
-import java.time.temporal.ChronoUnit;
-import java.util.regex.Pattern;
-
 public class HiveProductTest
         extends ProductTest
 {
-    private static final Logger log = Logger.get(HiveProductTest.class);
-
-    static final String ERROR_COMMITTING_WRITE_TO_HIVE_ISSUE = "https://github.com/trinodb/trino/issues/4936";
-    @Language("RegExp")
-    static final String ERROR_COMMITTING_WRITE_TO_HIVE_MATCH =
-            // "Error committing write to Hive" is present depending on when the exception is thrown.
-            // It may be absent when the underlying problem manifest earlier (e.g. during RecordFileWriter.appendRow vs RecordFileWriter.commit).
-
-            // "could only be written to 0 of the 1 minReplication" is the error wording used by e.g. HDP 3
-            "(could only be replicated to 0 nodes instead of minReplication|could only be written to 0 of the 1 minReplication)";
-
-    public static final RetryPolicy<QueryResult> ERROR_COMMITTING_WRITE_TO_HIVE_RETRY_POLICY = new RetryPolicy<QueryResult>()
-            .handleIf(HiveProductTest::isErrorCommittingToHive)
-            .withBackoff(1, 10, ChronoUnit.SECONDS)
-            .withMaxRetries(30)
-            .onRetry(event -> log.warn(event.getLastFailure(), "Query failed on attempt %d, will retry.", event.getAttemptCount()));
-
-    private static boolean isErrorCommittingToHive(Throwable throwable)
-    {
-        return Pattern.compile(ERROR_COMMITTING_WRITE_TO_HIVE_MATCH)
-                .matcher(Throwables.getStackTraceAsString(throwable))
-                .find();
-    }
-
     @Inject
     private HiveVersionProvider hiveVersionProvider;
 

testing/trino-product-tests/src/main/java/io/trino/tests/product/hive/TestAbfsSyncPartitionMetadata.java

@@ -21,9 +21,9 @@
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.collect.Iterables.getOnlyElement;
 import static io.trino.tests.product.TestGroups.AZURE;
-import static io.trino.tests.product.hive.HiveProductTest.ERROR_COMMITTING_WRITE_TO_HIVE_ISSUE;
-import static io.trino.tests.product.hive.HiveProductTest.ERROR_COMMITTING_WRITE_TO_HIVE_MATCH;
 import static io.trino.tests.product.hive.util.TemporaryHiveTable.randomTableSuffix;
+import static io.trino.tests.product.utils.HadoopTestUtils.ERROR_COMMITTING_WRITE_TO_HIVE_ISSUE;
+import static io.trino.tests.product.utils.HadoopTestUtils.ERROR_COMMITTING_WRITE_TO_HIVE_MATCH;
 import static io.trino.tests.product.utils.QueryExecutors.onHive;
 import static io.trino.tests.product.utils.QueryExecutors.onTrino;
 import static java.lang.String.format;

testing/trino-product-tests/src/main/java/io/trino/tests/product/hive/TestCsv.java

@@ -26,8 +26,8 @@
 import static io.trino.tempto.assertions.QueryAssert.assertThat;
 import static io.trino.tests.product.TestGroups.HMS_ONLY;
 import static io.trino.tests.product.TestGroups.STORAGE_FORMATS;
-import static io.trino.tests.product.hive.HiveProductTest.ERROR_COMMITTING_WRITE_TO_HIVE_ISSUE;
-import static io.trino.tests.product.hive.HiveProductTest.ERROR_COMMITTING_WRITE_TO_HIVE_MATCH;
+import static io.trino.tests.product.utils.HadoopTestUtils.ERROR_COMMITTING_WRITE_TO_HIVE_ISSUE;
+import static io.trino.tests.product.utils.HadoopTestUtils.ERROR_COMMITTING_WRITE_TO_HIVE_MATCH;
 import static io.trino.tests.product.utils.QueryExecutors.onHive;
 import static io.trino.tests.product.utils.QueryExecutors.onTrino;
 import static java.lang.String.format;

testing/trino-product-tests/src/main/java/io/trino/tests/product/hive/TestHdfsSyncPartitionMetadata.java

@@ -25,8 +25,8 @@
 import static io.trino.tests.product.TestGroups.HIVE_PARTITIONING;
 import static io.trino.tests.product.TestGroups.SMOKE;
 import static io.trino.tests.product.TestGroups.TRINO_JDBC;
-import static io.trino.tests.product.hive.HiveProductTest.ERROR_COMMITTING_WRITE_TO_HIVE_ISSUE;
-import static io.trino.tests.product.hive.HiveProductTest.ERROR_COMMITTING_WRITE_TO_HIVE_MATCH;
+import static io.trino.tests.product.utils.HadoopTestUtils.ERROR_COMMITTING_WRITE_TO_HIVE_ISSUE;
+import static io.trino.tests.product.utils.HadoopTestUtils.ERROR_COMMITTING_WRITE_TO_HIVE_MATCH;
 import static io.trino.tests.product.utils.QueryExecutors.onTrino;
 
 public class TestHdfsSyncPartitionMetadata

testing/trino-product-tests/src/main/java/io/trino/tests/product/hive/TestHiveBasicTableStatistics.java

@@ -27,9 +27,9 @@
 import java.util.OptionalLong;
 
 import static com.google.common.base.Verify.verify;
-import static io.trino.tests.product.hive.HiveProductTest.ERROR_COMMITTING_WRITE_TO_HIVE_ISSUE;
-import static io.trino.tests.product.hive.HiveProductTest.ERROR_COMMITTING_WRITE_TO_HIVE_MATCH;
 import static io.trino.tests.product.hive.util.TableLocationUtils.getTableLocation;
+import static io.trino.tests.product.utils.HadoopTestUtils.ERROR_COMMITTING_WRITE_TO_HIVE_ISSUE;
+import static io.trino.tests.product.utils.HadoopTestUtils.ERROR_COMMITTING_WRITE_TO_HIVE_MATCH;
 import static io.trino.tests.product.utils.QueryExecutors.onHive;
 import static io.trino.tests.product.utils.QueryExecutors.onTrino;
 import static java.lang.String.format;

testing/trino-product-tests/src/main/java/io/trino/tests/product/hive/TestHiveBucketedTables.java

@@ -51,6 +51,8 @@
 import static io.trino.tests.product.hive.BucketingType.BUCKETED_V2;
 import static io.trino.tests.product.hive.util.TemporaryHiveTable.randomTableSuffix;
 import static io.trino.tests.product.hive.util.TemporaryHiveTable.temporaryHiveTable;
+import static io.trino.tests.product.utils.HadoopTestUtils.ERROR_COMMITTING_WRITE_TO_HIVE_ISSUE;
+import static io.trino.tests.product.utils.HadoopTestUtils.ERROR_COMMITTING_WRITE_TO_HIVE_MATCH;
 import static io.trino.tests.product.utils.QueryExecutors.onHive;
 import static io.trino.tests.product.utils.QueryExecutors.onTrino;
 import static io.trino.tests.product.utils.TableDefinitionUtils.mutableTableInstanceOf;

testing/trino-product-tests/src/main/java/io/trino/tests/product/hive/TestHiveCreateTable.java

@@ -28,8 +28,8 @@
 import static io.trino.tests.product.TestGroups.HDP3_ONLY;
 import static io.trino.tests.product.TestGroups.PROFILE_SPECIFIC_TESTS;
 import static io.trino.tests.product.TestGroups.STORAGE_FORMATS;
-import static io.trino.tests.product.hive.HiveProductTest.ERROR_COMMITTING_WRITE_TO_HIVE_ISSUE;
-import static io.trino.tests.product.hive.HiveProductTest.ERROR_COMMITTING_WRITE_TO_HIVE_MATCH;
+import static io.trino.tests.product.utils.HadoopTestUtils.ERROR_COMMITTING_WRITE_TO_HIVE_ISSUE;
+import static io.trino.tests.product.utils.HadoopTestUtils.ERROR_COMMITTING_WRITE_TO_HIVE_MATCH;
 import static io.trino.tests.product.utils.QueryExecutors.onHive;
 import static io.trino.tests.product.utils.QueryExecutors.onTrino;
 import static java.util.Objects.requireNonNull;

testing/trino-product-tests/src/main/java/io/trino/tests/product/hive/TestHivePartitionProcedures.java

@@ -31,9 +31,9 @@
 import static io.trino.tempto.fulfillment.table.hive.InlineDataSource.createResourceDataSource;
 import static io.trino.tests.product.TestGroups.HIVE_PARTITIONING;
 import static io.trino.tests.product.TestGroups.SMOKE;
-import static io.trino.tests.product.hive.HiveProductTest.ERROR_COMMITTING_WRITE_TO_HIVE_ISSUE;
-import static io.trino.tests.product.hive.HiveProductTest.ERROR_COMMITTING_WRITE_TO_HIVE_MATCH;
 import static io.trino.tests.product.hive.util.TableLocationUtils.getTablePath;
+import static io.trino.tests.product.utils.HadoopTestUtils.ERROR_COMMITTING_WRITE_TO_HIVE_ISSUE;
+import static io.trino.tests.product.utils.HadoopTestUtils.ERROR_COMMITTING_WRITE_TO_HIVE_MATCH;
 import static io.trino.tests.product.utils.QueryExecutors.onTrino;
 import static java.lang.String.format;
 import static org.assertj.core.api.Assertions.assertThat;

testing/trino-product-tests/src/main/java/io/trino/tests/product/hive/TestHivePartitionsTable.java

@@ -43,8 +43,8 @@
 import static io.trino.tempto.fulfillment.table.hive.InlineDataSource.createStringDataSource;
 import static io.trino.tempto.fulfillment.table.hive.tpch.TpchTableDefinitions.NATION;
 import static io.trino.tests.product.TestGroups.HIVE_PARTITIONING;
-import static io.trino.tests.product.hive.HiveProductTest.ERROR_COMMITTING_WRITE_TO_HIVE_ISSUE;
-import static io.trino.tests.product.hive.HiveProductTest.ERROR_COMMITTING_WRITE_TO_HIVE_MATCH;
+import static io.trino.tests.product.utils.HadoopTestUtils.ERROR_COMMITTING_WRITE_TO_HIVE_ISSUE;
+import static io.trino.tests.product.utils.HadoopTestUtils.ERROR_COMMITTING_WRITE_TO_HIVE_MATCH;
 import static io.trino.tests.product.utils.QueryExecutors.onTrino;
 import static java.lang.Math.min;
 import static java.lang.String.format;

testing/trino-product-tests/src/main/java/io/trino/tests/product/hive/TestHiveStorageFormats.java

@@ -66,9 +66,9 @@
 import static io.trino.tests.product.TestGroups.HMS_ONLY;
 import static io.trino.tests.product.TestGroups.STORAGE_FORMATS;
 import static io.trino.tests.product.TestGroups.STORAGE_FORMATS_DETAILED;
-import static io.trino.tests.product.hive.HiveProductTest.ERROR_COMMITTING_WRITE_TO_HIVE_ISSUE;
-import static io.trino.tests.product.hive.HiveProductTest.ERROR_COMMITTING_WRITE_TO_HIVE_MATCH;
 import static io.trino.tests.product.hive.util.TemporaryHiveTable.randomTableSuffix;
+import static io.trino.tests.product.utils.HadoopTestUtils.ERROR_COMMITTING_WRITE_TO_HIVE_ISSUE;
+import static io.trino.tests.product.utils.HadoopTestUtils.ERROR_COMMITTING_WRITE_TO_HIVE_MATCH;
 import static io.trino.tests.product.utils.JdbcDriverUtils.setSessionProperty;
 import static io.trino.tests.product.utils.QueryExecutors.onHive;
 import static io.trino.tests.product.utils.QueryExecutors.onTrino;

testing/trino-product-tests/src/main/java/io/trino/tests/product/hive/TestHiveTableStatistics.java

@@ -39,6 +39,8 @@
 import static io.trino.tests.product.hive.AllSimpleTypesTableDefinitions.ALL_HIVE_SIMPLE_TYPES_TEXTFILE;
 import static io.trino.tests.product.hive.HiveTableDefinitions.NATION_PARTITIONED_BY_BIGINT_REGIONKEY;
 import static io.trino.tests.product.hive.HiveTableDefinitions.NATION_PARTITIONED_BY_VARCHAR_REGIONKEY;
+import static io.trino.tests.product.utils.HadoopTestUtils.ERROR_COMMITTING_WRITE_TO_HIVE_ISSUE;
+import static io.trino.tests.product.utils.HadoopTestUtils.ERROR_COMMITTING_WRITE_TO_HIVE_MATCH;
 import static io.trino.tests.product.utils.QueryExecutors.onHive;
 import static io.trino.tests.product.utils.QueryExecutors.onTrino;
 import static java.lang.String.format;

testing/trino-product-tests/src/main/java/io/trino/tests/product/hive/TestHiveTransactionalTable.java

@@ -65,6 +65,8 @@
 import static io.trino.tests.product.hive.TransactionalTableType.INSERT_ONLY;
 import static io.trino.tests.product.hive.util.TableLocationUtils.getTablePath;
 import static io.trino.tests.product.hive.util.TemporaryHiveTable.randomTableSuffix;
+import static io.trino.tests.product.utils.HadoopTestUtils.ERROR_COMMITTING_WRITE_TO_HIVE_ISSUE;
+import static io.trino.tests.product.utils.HadoopTestUtils.ERROR_COMMITTING_WRITE_TO_HIVE_MATCH;
 import static io.trino.tests.product.utils.QueryExecutors.onHive;
 import static io.trino.tests.product.utils.QueryExecutors.onTrino;
 import static java.lang.String.format;

testing/trino-product-tests/src/main/java/io/trino/tests/product/utils/HadoopTestUtils.java

@@ -0,0 +1,53 @@
+/*
+ * 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.tests.product.utils;
+
+import com.google.common.base.Throwables;
+import io.airlift.log.Logger;
+import io.trino.tempto.query.QueryResult;
+import io.trino.tests.product.hive.HiveProductTest;
+import net.jodah.failsafe.RetryPolicy;
+import org.intellij.lang.annotations.Language;
+
+import java.time.temporal.ChronoUnit;
+import java.util.regex.Pattern;
+
+public final class HadoopTestUtils
+{
+    private HadoopTestUtils() {}
+
+    private static final Logger log = Logger.get(HiveProductTest.class);
+
+    public static final String ERROR_COMMITTING_WRITE_TO_HIVE_ISSUE = "https://github.com/trinodb/trino/issues/4936";
+    @Language("RegExp")
+    public static final String ERROR_COMMITTING_WRITE_TO_HIVE_MATCH =
+            // "Error committing write to Hive" is present depending on when the exception is thrown.
+            // It may be absent when the underlying problem manifest earlier (e.g. during RecordFileWriter.appendRow vs RecordFileWriter.commit).
+
+            // "could only be written to 0 of the 1 minReplication" is the error wording used by e.g. HDP 3
+            "(could only be replicated to 0 nodes instead of minReplication|could only be written to 0 of the 1 minReplication)";
+
+    public static final RetryPolicy<QueryResult> ERROR_COMMITTING_WRITE_TO_HIVE_RETRY_POLICY = new RetryPolicy<QueryResult>()
+            .handleIf(HadoopTestUtils::isErrorCommittingToHive)
+            .withBackoff(1, 10, ChronoUnit.SECONDS)
+            .withMaxRetries(30)
+            .onRetry(event -> log.warn(event.getLastFailure(), "Query failed on attempt %d, will retry.", event.getAttemptCount()));
+
+    private static boolean isErrorCommittingToHive(Throwable throwable)
+    {
+        return Pattern.compile(ERROR_COMMITTING_WRITE_TO_HIVE_MATCH)
+                .matcher(Throwables.getStackTraceAsString(throwable))
+                .find();
+    }
+}

testing/trino-product-tests/src/main/java/io/trino/tests/product/utils/QueryExecutors.java

@@ -24,7 +24,7 @@
 import java.time.temporal.ChronoUnit;
 
 import static io.trino.tempto.context.ThreadLocalTestContextHolder.testContext;
-import static io.trino.tests.product.hive.HiveProductTest.ERROR_COMMITTING_WRITE_TO_HIVE_RETRY_POLICY;
+import static io.trino.tests.product.utils.HadoopTestUtils.ERROR_COMMITTING_WRITE_TO_HIVE_RETRY_POLICY;
 
 public final class QueryExecutors
 {
@@ -91,7 +91,29 @@ public static QueryExecutor onMySql()
 
     public static QueryExecutor onSpark()
     {
-        return testContext().getDependency(QueryExecutor.class, "spark");
+        return new QueryExecutor() {
+            private final QueryExecutor delegate = testContext().getDependency(QueryExecutor.class, "spark");
+
+            @Override
+            public QueryResult executeQuery(String sql, QueryParam... params)
+                    throws QueryExecutionException
+            {
+                return Failsafe.with(ERROR_COMMITTING_WRITE_TO_HIVE_RETRY_POLICY)
+                        .get(() -> delegate.executeQuery(sql, params));
+            }
+
+            @Override
+            public Connection getConnection()
+            {
+                return delegate.getConnection();
+            }
+
+            @Override
+            public void close()
+            {
+                delegate.close();
+            }
+        };
     }
 
     public static QueryExecutor onDelta()