diff --git a/pom.xml b/pom.xml
index df26e6e278a0..26dc8449ae49 100644
--- a/pom.xml
+++ b/pom.xml
@@ -161,7 +161,7 @@
         <dep.jsonwebtoken.version>0.11.5</dep.jsonwebtoken.version>
         <dep.oracle.version>21.9.0.0</dep.oracle.version>
         <dep.drift.version>1.20</dep.drift.version>
-        <dep.tempto.version>197</dep.tempto.version>
+        <dep.tempto.version>200</dep.tempto.version>
         <dep.gcs.version>2.2.8</dep.gcs.version>
 
         <dep.errorprone.version>2.20.0</dep.errorprone.version>
@@ -180,7 +180,7 @@
         <dep.jna.version>5.13.0</dep.jna.version>
         <dep.okio.version>3.3.0</dep.okio.version>
 
-        <dep.docker.images.version>80</dep.docker.images.version>
+        <dep.docker.images.version>81</dep.docker.images.version>
 
         <!--
           America/Bahia_Banderas has:
@@ -412,12 +412,6 @@
                 <classifier>all</classifier>
             </dependency>
 
-            <dependency>
-                <groupId>com.databricks</groupId>
-                <artifactId>databricks-jdbc</artifactId>
-                <version>2.6.32</version>
-            </dependency>
-
             <dependency>
                 <groupId>com.datastax.oss</groupId>
                 <artifactId>java-driver-core</artifactId>
diff --git a/testing/trino-product-tests/pom.xml b/testing/trino-product-tests/pom.xml
index d5274eb07bc4..33543312842d 100644
--- a/testing/trino-product-tests/pom.xml
+++ b/testing/trino-product-tests/pom.xml
@@ -38,6 +38,12 @@
             <artifactId>java-driver-core</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>com.google.code.findbugs</groupId>
+            <artifactId>jsr305</artifactId>
+            <optional>true</optional>
+        </dependency>
+
         <dependency>
             <groupId>com.google.guava</groupId>
             <artifactId>guava</artifactId>
@@ -104,6 +110,11 @@
             <artifactId>units</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>io.airlift.resolver</groupId>
+            <artifactId>resolver</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>io.confluent</groupId>
             <artifactId>kafka-protobuf-provider</artifactId>
@@ -223,6 +234,11 @@
             <artifactId>annotations</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.sonatype.aether</groupId>
+            <artifactId>aether-api</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>org.testng</groupId>
             <artifactId>testng</artifactId>
@@ -236,8 +252,9 @@
         </dependency>
 
         <dependency>
-            <groupId>com.databricks</groupId>
-            <artifactId>databricks-jdbc</artifactId>
+            <groupId>com.linkedin.calcite</groupId>
+            <artifactId>calcite-core</artifactId>
+            <classifier>shaded</classifier>
             <scope>runtime</scope>
         </dependency>
 
diff --git a/testing/trino-product-tests/src/main/java/io/trino/tests/product/deltalake/TestDeltaLakeAlterTableCompatibility.java b/testing/trino-product-tests/src/main/java/io/trino/tests/product/deltalake/TestDeltaLakeAlterTableCompatibility.java
index 5db8f63ccecd..8d006e24fb44 100644
--- a/testing/trino-product-tests/src/main/java/io/trino/tests/product/deltalake/TestDeltaLakeAlterTableCompatibility.java
+++ b/testing/trino-product-tests/src/main/java/io/trino/tests/product/deltalake/TestDeltaLakeAlterTableCompatibility.java
@@ -25,6 +25,7 @@
 import static io.trino.testing.TestingNames.randomNameSuffix;
 import static io.trino.tests.product.TestGroups.DELTA_LAKE_DATABRICKS;
 import static io.trino.tests.product.TestGroups.DELTA_LAKE_EXCLUDE_104;
+import static io.trino.tests.product.TestGroups.DELTA_LAKE_EXCLUDE_113;
 import static io.trino.tests.product.TestGroups.DELTA_LAKE_EXCLUDE_73;
 import static io.trino.tests.product.TestGroups.DELTA_LAKE_EXCLUDE_91;
 import static io.trino.tests.product.TestGroups.DELTA_LAKE_OSS;
@@ -70,7 +71,7 @@ public void testAddColumnWithCommentOnTrino()
         }
     }
 
-    @Test(groups = {DELTA_LAKE_DATABRICKS, DELTA_LAKE_OSS, DELTA_LAKE_EXCLUDE_73, DELTA_LAKE_EXCLUDE_91, PROFILE_SPECIFIC_TESTS})
+    @Test(groups = {DELTA_LAKE_DATABRICKS, DELTA_LAKE_OSS, DELTA_LAKE_EXCLUDE_73, DELTA_LAKE_EXCLUDE_91, DELTA_LAKE_EXCLUDE_104, DELTA_LAKE_EXCLUDE_113, PROFILE_SPECIFIC_TESTS})
     @Flaky(issue = DATABRICKS_COMMUNICATION_FAILURE_ISSUE, match = DATABRICKS_COMMUNICATION_FAILURE_MATCH)
     public void testAddColumnUnsupportedWriterVersion()
     {
@@ -80,14 +81,14 @@ public void testAddColumnUnsupportedWriterVersion()
         onDelta().executeQuery(format("" +
                         "CREATE TABLE default.%s (col int) " +
                         "USING DELTA LOCATION 's3://%s/%s'" +
-                        "TBLPROPERTIES ('delta.minWriterVersion'='6')",
+                        "TBLPROPERTIES ('delta.minWriterVersion'='7')",
                 tableName,
                 bucketName,
                 tableDirectory));
 
         try {
             assertQueryFailure(() -> onTrino().executeQuery("ALTER TABLE delta.default." + tableName + " ADD COLUMN new_col int"))
-                    .hasMessageMatching(".* Table .* requires Delta Lake writer version 6 which is not supported");
+                    .hasMessageMatching(".* Table .* requires Delta Lake writer version 7 which is not supported");
         }
         finally {
             dropDeltaTableWithRetry("default." + tableName);
@@ -193,14 +194,14 @@ public void testCommentOnTableUnsupportedWriterVersion()
         onDelta().executeQuery(format("" +
                         "CREATE TABLE default.%s (col int) " +
                         "USING DELTA LOCATION 's3://%s/%s'" +
-                        "TBLPROPERTIES ('delta.minWriterVersion'='6')",
+                        "TBLPROPERTIES ('delta.minWriterVersion'='7')",
                 tableName,
                 bucketName,
                 tableDirectory));
 
         try {
             assertQueryFailure(() -> onTrino().executeQuery("COMMENT ON TABLE delta.default." + tableName + " IS 'test comment'"))
-                    .hasMessageMatching(".* Table .* requires Delta Lake writer version 6 which is not supported");
+                    .hasMessageMatching(".* Table .* requires Delta Lake writer version 7 which is not supported");
         }
         finally {
             onTrino().executeQuery("DROP TABLE delta.default." + tableName);
@@ -239,21 +240,21 @@ public void testCommentOnColumnUnsupportedWriterVersion()
         onDelta().executeQuery(format("" +
                         "CREATE TABLE default.%s (col int) " +
                         "USING DELTA LOCATION 's3://%s/%s'" +
-                        "TBLPROPERTIES ('delta.minWriterVersion'='6')",
+                        "TBLPROPERTIES ('delta.minWriterVersion'='7')",
                 tableName,
                 bucketName,
                 tableDirectory));
 
         try {
             assertQueryFailure(() -> onTrino().executeQuery("COMMENT ON COLUMN delta.default." + tableName + ".col IS 'test column comment'"))
-                    .hasMessageMatching(".* Table .* requires Delta Lake writer version 6 which is not supported");
+                    .hasMessageMatching(".* Table .* requires Delta Lake writer version 7 which is not supported");
         }
         finally {
             onTrino().executeQuery("DROP TABLE delta.default." + tableName);
         }
     }
 
-    @Test(groups = {DELTA_LAKE_DATABRICKS, DELTA_LAKE_OSS, DELTA_LAKE_EXCLUDE_73, DELTA_LAKE_EXCLUDE_91, DELTA_LAKE_EXCLUDE_104, PROFILE_SPECIFIC_TESTS})
+    @Test(groups = {DELTA_LAKE_DATABRICKS, DELTA_LAKE_OSS, DELTA_LAKE_EXCLUDE_73, DELTA_LAKE_EXCLUDE_91, DELTA_LAKE_EXCLUDE_104, DELTA_LAKE_EXCLUDE_113, PROFILE_SPECIFIC_TESTS})
     @Flaky(issue = DATABRICKS_COMMUNICATION_FAILURE_ISSUE, match = DATABRICKS_COMMUNICATION_FAILURE_MATCH)
     public void testOptimizeUnsupportedWriterVersion()
     {
@@ -263,14 +264,14 @@ public void testOptimizeUnsupportedWriterVersion()
         onDelta().executeQuery(format("" +
                         "CREATE TABLE default.%s (col int) " +
                         "USING DELTA LOCATION 's3://%s/%s'" +
-                        "TBLPROPERTIES ('delta.minWriterVersion'='6')",
+                        "TBLPROPERTIES ('delta.minWriterVersion'='7')",
                 tableName,
                 bucketName,
                 tableDirectory));
 
         try {
             assertQueryFailure(() -> onTrino().executeQuery("ALTER TABLE delta.default." + tableName + " EXECUTE OPTIMIZE"))
-                    .hasMessageMatching(".* Table .* requires Delta Lake writer version 6 which is not supported");
+                    .hasMessageMatching(".* Table .* requires Delta Lake writer version 7 which is not supported");
         }
         finally {
             dropDeltaTableWithRetry(tableName);
diff --git a/testing/trino-product-tests/src/main/java/io/trino/tests/product/deltalake/TestDeltaLakeDatabricksDelete.java b/testing/trino-product-tests/src/main/java/io/trino/tests/product/deltalake/TestDeltaLakeDatabricksDelete.java
index db1499d3f7ad..a886a4e7106f 100644
--- a/testing/trino-product-tests/src/main/java/io/trino/tests/product/deltalake/TestDeltaLakeDatabricksDelete.java
+++ b/testing/trino-product-tests/src/main/java/io/trino/tests/product/deltalake/TestDeltaLakeDatabricksDelete.java
@@ -59,9 +59,8 @@ public void testDeleteOnAppendOnlyTableFails()
         onTrino().executeQuery("DROP TABLE " + tableName);
     }
 
-    // Databricks 12.1 added support for deletion vectors
-    // TODO: Add DELTA_LAKE_OSS group once they support creating a table with deletion vectors
-    @Test(groups = {DELTA_LAKE_DATABRICKS, DELTA_LAKE_EXCLUDE_73, DELTA_LAKE_EXCLUDE_91, DELTA_LAKE_EXCLUDE_104, DELTA_LAKE_EXCLUDE_113, PROFILE_SPECIFIC_TESTS})
+    // Databricks 12.1 and OSS Delta 2.4.0 added support for deletion vectors
+    @Test(groups = {DELTA_LAKE_DATABRICKS, DELTA_LAKE_OSS, DELTA_LAKE_EXCLUDE_73, DELTA_LAKE_EXCLUDE_91, DELTA_LAKE_EXCLUDE_104, DELTA_LAKE_EXCLUDE_113, PROFILE_SPECIFIC_TESTS})
     @Flaky(issue = DATABRICKS_COMMUNICATION_FAILURE_ISSUE, match = DATABRICKS_COMMUNICATION_FAILURE_MATCH)
     public void testDeletionVectors()
     {
diff --git a/testing/trino-product-tests/src/main/java/io/trino/tests/product/deltalake/TestDeltaLakeWriteDatabricksCompatibility.java b/testing/trino-product-tests/src/main/java/io/trino/tests/product/deltalake/TestDeltaLakeWriteDatabricksCompatibility.java
index 1f9b9cfe50f9..d414bbe051cd 100644
--- a/testing/trino-product-tests/src/main/java/io/trino/tests/product/deltalake/TestDeltaLakeWriteDatabricksCompatibility.java
+++ b/testing/trino-product-tests/src/main/java/io/trino/tests/product/deltalake/TestDeltaLakeWriteDatabricksCompatibility.java
@@ -33,6 +33,8 @@
 import static io.trino.tempto.assertions.QueryAssert.assertQueryFailure;
 import static io.trino.testing.TestingNames.randomNameSuffix;
 import static io.trino.tests.product.TestGroups.DELTA_LAKE_DATABRICKS;
+import static io.trino.tests.product.TestGroups.DELTA_LAKE_EXCLUDE_104;
+import static io.trino.tests.product.TestGroups.DELTA_LAKE_EXCLUDE_113;
 import static io.trino.tests.product.TestGroups.DELTA_LAKE_EXCLUDE_73;
 import static io.trino.tests.product.TestGroups.DELTA_LAKE_EXCLUDE_91;
 import static io.trino.tests.product.TestGroups.DELTA_LAKE_OSS;
@@ -389,7 +391,7 @@ private void testVacuumRemoveChangeDataFeedFiles(Consumer<String> vacuumExecutor
         }
     }
 
-    @Test(groups = {DELTA_LAKE_DATABRICKS, DELTA_LAKE_OSS, DELTA_LAKE_EXCLUDE_73, DELTA_LAKE_EXCLUDE_91, PROFILE_SPECIFIC_TESTS})
+    @Test(groups = {DELTA_LAKE_DATABRICKS, DELTA_LAKE_OSS, DELTA_LAKE_EXCLUDE_73, DELTA_LAKE_EXCLUDE_91, DELTA_LAKE_EXCLUDE_104, DELTA_LAKE_EXCLUDE_113, PROFILE_SPECIFIC_TESTS})
     @Flaky(issue = DATABRICKS_COMMUNICATION_FAILURE_ISSUE, match = DATABRICKS_COMMUNICATION_FAILURE_MATCH)
     public void testVacuumUnsupportedWriterVersion()
     {
@@ -400,10 +402,10 @@ public void testVacuumUnsupportedWriterVersion()
                 "(a INT)" +
                 "USING DELTA " +
                 "LOCATION '" + ("s3://" + bucketName + "/" + directoryName) + "'" +
-                "TBLPROPERTIES ('delta.minWriterVersion'='6')");
+                "TBLPROPERTIES ('delta.minWriterVersion'='7')");
         try {
             assertThatThrownBy(() -> onTrino().executeQuery("CALL delta.system.vacuum('default', '" + tableName + "', '7d')"))
-                    .hasMessageContaining("Cannot execute vacuum procedure with 6 writer version");
+                    .hasMessageContaining("Cannot execute vacuum procedure with 7 writer version");
         }
         finally {
             dropDeltaTableWithRetry("default." + tableName);
diff --git a/testing/trino-product-tests/src/main/java/io/trino/tests/product/utils/DeltaQueryExecutors.java b/testing/trino-product-tests/src/main/java/io/trino/tests/product/utils/DeltaQueryExecutors.java
new file mode 100644
index 000000000000..a91bb6dcaa91
--- /dev/null
+++ b/testing/trino-product-tests/src/main/java/io/trino/tests/product/utils/DeltaQueryExecutors.java
@@ -0,0 +1,105 @@
+/*
+ * 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.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import dev.failsafe.Failsafe;
+import dev.failsafe.RetryPolicy;
+import io.airlift.log.Logger;
+import io.airlift.resolver.ArtifactResolver;
+import io.airlift.resolver.DefaultArtifact;
+import io.trino.tempto.context.TestContext;
+import io.trino.tempto.query.JdbcConnectionsPool;
+import io.trino.tempto.query.JdbcConnectivityParamsState;
+import io.trino.tempto.query.JdbcQueryExecutor;
+
+import javax.annotation.concurrent.GuardedBy;
+
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+import static com.google.common.collect.MoreCollectors.onlyElement;
+import static io.airlift.resolver.ArtifactResolver.MAVEN_CENTRAL_URI;
+import static io.airlift.resolver.ArtifactResolver.USER_LOCAL_REPO;
+import static java.util.Map.entry;
+
+public final class DeltaQueryExecutors
+{
+    private static final Logger log = Logger.get(DeltaQueryExecutors.class);
+
+    private static final Map<String, Map.Entry<String, String>> ARTIFACTS = ImmutableMap.<String, Map.Entry<String, String>>builder()
+            .put("org.apache.hive.jdbc.HiveDriver", entry("org.apache.hive:hive-jdbc:jar:standalone:3.1.3", "hive-jdbc-3.1.3-standalone.jar"))
+            .put("com.databricks.client.jdbc.Driver", entry("com.databricks:databricks-jdbc:2.6.32", "databricks-jdbc-2.6.32.jar"))
+            .buildOrThrow();
+
+    @GuardedBy("DRIVERS")
+    private static final Map<String, String> DRIVERS = new HashMap<>();
+    @GuardedBy("JDBC_EXECUTORS")
+    private static final Map<JdbcConnectivityParamsState, JdbcQueryExecutor> JDBC_EXECUTORS = new HashMap<>();
+
+    private static final RetryPolicy<String> loadDatabaseDriverRetryPolicy = RetryPolicy.<String>builder()
+            .withMaxRetries(30)
+            .withDelay(Duration.ofSeconds(10))
+            .onRetry(event -> log.warn(event.getLastException(), "Download failed on attempt %d, will retry.", event.getAttemptCount()))
+            .build();
+
+    private DeltaQueryExecutors() {}
+
+    public static JdbcQueryExecutor createDeltaQueryExecutor(TestContext testContext)
+    {
+        JdbcConnectivityParamsState jdbcParamsState = testContext.getDependency(JdbcConnectivityParamsState.class, "delta");
+        JdbcConnectionsPool jdbcConnectionsPool = testContext.getDependency(JdbcConnectionsPool.class);
+        synchronized (JDBC_EXECUTORS) {
+            return JDBC_EXECUTORS.computeIfAbsent(jdbcParamsState, param -> new JdbcQueryExecutor(withRuntimeJar(param), jdbcConnectionsPool, testContext));
+        }
+    }
+
+    private static JdbcConnectivityParamsState withRuntimeJar(JdbcConnectivityParamsState jdbcParamsState)
+    {
+        String jarFilePath;
+        synchronized (DRIVERS) {
+            jarFilePath = DRIVERS.computeIfAbsent(jdbcParamsState.driverClass, className -> Failsafe.with(loadDatabaseDriverRetryPolicy)
+                    .get(() -> loadDatabaseDriverJar(className)));
+        }
+
+        return JdbcConnectivityParamsState.builder()
+                .setName(jdbcParamsState.getName().orElseThrow())
+                .setDriverClass(jdbcParamsState.driverClass)
+                .setUrl(jdbcParamsState.url)
+                .setUser(jdbcParamsState.user)
+                .setPassword(jdbcParamsState.password)
+                .setPooling(jdbcParamsState.pooling)
+                .setJar(Optional.of(jarFilePath))
+                .setPrepareStatements(jdbcParamsState.prepareStatements)
+                .setKerberosPrincipal(jdbcParamsState.kerberosPrincipal)
+                .setKerberosKeytab(jdbcParamsState.kerberosKeytab)
+                .build();
+    }
+
+    private static String loadDatabaseDriverJar(String driverClassName)
+    {
+        // TODO Add support for maven coordinate in tempto
+        String coords = ARTIFACTS.get(driverClassName).getKey();
+        String jar = ARTIFACTS.get(driverClassName).getValue();
+
+        ArtifactResolver resolver = new ArtifactResolver(USER_LOCAL_REPO, ImmutableList.of(MAVEN_CENTRAL_URI));
+        return resolver.resolveArtifacts(new DefaultArtifact(coords)).stream()
+                .filter(artifact -> artifact.getFile().getName().equals(jar))
+                .map(artifact -> artifact.getFile().getAbsolutePath())
+                .collect(onlyElement());
+    }
+}
diff --git a/testing/trino-product-tests/src/main/java/io/trino/tests/product/utils/QueryExecutors.java b/testing/trino-product-tests/src/main/java/io/trino/tests/product/utils/QueryExecutors.java
index b9028155877b..a89ddbf8e637 100644
--- a/testing/trino-product-tests/src/main/java/io/trino/tests/product/utils/QueryExecutors.java
+++ b/testing/trino-product-tests/src/main/java/io/trino/tests/product/utils/QueryExecutors.java
@@ -24,6 +24,7 @@
 import java.time.temporal.ChronoUnit;
 
 import static io.trino.tempto.context.ThreadLocalTestContextHolder.testContext;
+import static io.trino.tests.product.utils.DeltaQueryExecutors.createDeltaQueryExecutor;
 import static io.trino.tests.product.utils.HadoopTestUtils.ERROR_COMMITTING_WRITE_TO_HIVE_RETRY_POLICY;
 
 public final class QueryExecutors
@@ -126,7 +127,7 @@ public static QueryExecutor onDelta()
 
         return new QueryExecutor()
         {
-            private final QueryExecutor delegate = testContext().getDependency(QueryExecutor.class, "delta");
+            private final QueryExecutor delegate = createDeltaQueryExecutor(testContext());
 
             @Override
             public QueryResult executeQuery(String sql, QueryParam... params)