trinodb · ebyhr · Jul 13, 2022 · Jul 4, 2022
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -747,6 +747,7 @@ jobs:
           DATABRICKS_AWS_SECRET_ACCESS_KEY: ${{ secrets.DATABRICKS_AWS_SECRET_ACCESS_KEY }}
           DATABRICKS_73_JDBC_URL: ${{ secrets.DATABRICKS_73_JDBC_URL }}
           DATABRICKS_91_JDBC_URL: ${{ secrets.DATABRICKS_91_JDBC_URL }}
+          DATABRICKS_104_JDBC_URL: ${{ secrets.DATABRICKS_104_JDBC_URL }}
           DATABRICKS_LOGIN: token
           DATABRICKS_TOKEN: ${{ secrets.DATABRICKS_TOKEN }}
         run: |

diff --git a/docs/src/main/sphinx/connector/delta-lake.rst b/docs/src/main/sphinx/connector/delta-lake.rst
@@ -16,7 +16,7 @@ Requirements
 
 To connect to Databricks Delta Lake, you need:
 
-* Tables written by Databricks Runtime 7.3 LTS and 9.1 LTS are supported.
+* Tables written by Databricks Runtime 7.3 LTS, 9.1 LTS and 10.4 LTS are supported.
 * Deployments using AWS, HDFS, and Azure Storage are fully supported. Google
   Cloud Storage (GCS) is :ref:`partially supported<delta-lake-gcs-support>`.
 * Network access from the coordinator and workers to the Delta Lake storage.

diff --git a/.../io/trino/tests/product/launcher/env/environment/EnvSinglenodeDeltaLakeDatabricks104.java b/.../io/trino/tests/product/launcher/env/environment/EnvSinglenodeDeltaLakeDatabricks104.java
@@ -0,0 +1,38 @@
+/*
+ * 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.launcher.env.environment;
+
+import com.google.inject.Inject;
+import io.trino.tests.product.launcher.docker.DockerFiles;
+import io.trino.tests.product.launcher.env.common.Standard;
+import io.trino.tests.product.launcher.env.common.TestsEnvironment;
+
+import static java.util.Objects.requireNonNull;
+
+@TestsEnvironment
+public class EnvSinglenodeDeltaLakeDatabricks104
+        extends AbstractSinglenodeDeltaLakeDatabricks
+{
+    @Inject
+    public EnvSinglenodeDeltaLakeDatabricks104(Standard standard, DockerFiles dockerFiles)
+    {
+        super(standard, dockerFiles);
+    }
+
+    @Override
+    String databricksTestJdbcUrl()
+    {
+        return requireNonNull(System.getenv("DATABRICKS_104_JDBC_URL"), "Environment DATABRICKS_104_JDBC_URL was not set");
+    }
+}
diff --git a/.../src/main/java/io/trino/tests/product/launcher/suite/suites/SuiteDeltaLakeDatabricks.java b/.../src/main/java/io/trino/tests/product/launcher/suite/suites/SuiteDeltaLakeDatabricks.java
@@ -15,6 +15,7 @@
 
 import com.google.common.collect.ImmutableList;
 import io.trino.tests.product.launcher.env.EnvironmentConfig;
+import io.trino.tests.product.launcher.env.environment.EnvSinglenodeDeltaLakeDatabricks104;
 import io.trino.tests.product.launcher.env.environment.EnvSinglenodeDeltaLakeDatabricks73;
 import io.trino.tests.product.launcher.env.environment.EnvSinglenodeDeltaLakeDatabricks91;
 import io.trino.tests.product.launcher.suite.Suite;
@@ -56,6 +57,11 @@ public List<SuiteTestRun> getTestRuns(EnvironmentConfig config)
                         .withGroups("configured_features", "delta-lake-databricks")
                         .withExcludedGroups("delta-lake-exclude-91")
                         .withExcludedTests(excludedTests)
+                        .build(),
+
+                testOnEnvironment(EnvSinglenodeDeltaLakeDatabricks104.class)
+                        .withGroups("configured_features", "delta-lake-databricks")
+                        .withExcludedTests(excludedTests)
                         .build());
     }
 }
diff --git a/...ava/io/trino/tests/product/deltalake/TestDeltaLakeDatabricksCheckpointsCompatibility.java b/...ava/io/trino/tests/product/deltalake/TestDeltaLakeDatabricksCheckpointsCompatibility.java
@@ -34,6 +34,8 @@
 import static io.trino.tests.product.TestGroups.PROFILE_SPECIFIC_TESTS;
 import static io.trino.tests.product.deltalake.TransactionLogAssertions.assertLastEntryIsCheckpointed;
 import static io.trino.tests.product.deltalake.TransactionLogAssertions.assertTransactionLogVersion;
+import static io.trino.tests.product.deltalake.util.DeltaLakeTestUtils.DATABRICKS_104_RUNTIME_VERSION;
+import static io.trino.tests.product.deltalake.util.DeltaLakeTestUtils.getDatabricksRuntimeVersion;
 import static io.trino.tests.product.hive.util.TemporaryHiveTable.randomTableSuffix;
 import static io.trino.tests.product.utils.QueryExecutors.onDelta;
 import static io.trino.tests.product.utils.QueryExecutors.onTrino;
@@ -48,12 +50,14 @@ public class TestDeltaLakeDatabricksCheckpointsCompatibility
     private String s3ServerType;
 
     private AmazonS3 s3;
+    private String databricksRuntimeVersion;
 
     @BeforeTestWithContext
     public void setup()
     {
         super.setUp();
         s3 = new S3ClientFactory().createS3Client(s3ServerType);
+        databricksRuntimeVersion = getDatabricksRuntimeVersion();
     }
 
     @Test(groups = {DELTA_LAKE_DATABRICKS, PROFILE_SPECIFIC_TESTS})
@@ -190,18 +194,38 @@ public void testDatabricksUsesCheckpointInterval()
 
         try {
             // validate that Databricks can see the checkpoint interval
-            String showCreateTable = format(
-                    "CREATE TABLE `default`.`%s` (\n" +
-                            "  `a_number` BIGINT,\n" +
-                            "  `a_string` STRING)\n" +
-                            "USING DELTA\n" +
-                            "PARTITIONED BY (a_number)\n" +
-                            "LOCATION 's3://%s/%s'\n" +
-                            "TBLPROPERTIES (\n" +
-                            "  'delta.checkpointInterval' = '3')\n",
-                    tableName,
-                    bucketName,
-                    tableDirectory);
+            String showCreateTable;
+            if (databricksRuntimeVersion.equals(DATABRICKS_104_RUNTIME_VERSION)) {
+                showCreateTable = format(
+                        "CREATE TABLE spark_catalog.default.%s (\n" +
+                                "  a_number BIGINT,\n" +
+                                "  a_string STRING)\n" +
+                                "USING delta\n" +
+                                "PARTITIONED BY (a_number)\n" +
+                                "LOCATION 's3://%s/%s'\n" +
+                                "TBLPROPERTIES (\n" +
+                                "  'Type' = 'EXTERNAL',\n" +
+                                "  'delta.checkpointInterval' = '3',\n" +
+                                "  'delta.minReaderVersion' = '1',\n" +
+                                "  'delta.minWriterVersion' = '2')\n",
+                        tableName,
+                        bucketName,
+                        tableDirectory);
+            }
+            else {
+                showCreateTable = format(
+                        "CREATE TABLE `default`.`%s` (\n" +
+                                "  `a_number` BIGINT,\n" +
+                                "  `a_string` STRING)\n" +
+                                "USING DELTA\n" +
+                                "PARTITIONED BY (a_number)\n" +
+                                "LOCATION 's3://%s/%s'\n" +
+                                "TBLPROPERTIES (\n" +
+                                "  'delta.checkpointInterval' = '3')\n",
+                        tableName,
+                        bucketName,
+                        tableDirectory);
+            }
             assertThat(onDelta().executeQuery("SHOW CREATE TABLE default." + tableName)).containsExactlyInOrder(row(showCreateTable));
 
             // sanity check
@@ -275,7 +299,8 @@ private void testCheckpointMinMaxStatisticsForRowType(Consumer<String> sqlExecut
 
             // Assert min/max queries can be computed from just metadata
             String explainSelectMax = getOnlyElement(onDelta().executeQuery("EXPLAIN SELECT max(root.entry_one) FROM default." + tableName).column(1));
-            assertThat(explainSelectMax).matches("== Physical Plan ==\\s*LocalTableScan \\[max\\(root.entry_one AS `entry_one`\\).*]\\s*");
+            String column = databricksRuntimeVersion.equals(DATABRICKS_104_RUNTIME_VERSION) ? "root.entry_one" : "root.entry_one AS `entry_one`";
+            assertThat(explainSelectMax).matches("== Physical Plan ==\\s*LocalTableScan \\[max\\(" + column + "\\).*]\\s*");
 
             // check both engines can read both tables
             List<QueryAssert.Row> maxMin = ImmutableList.of(row(3, "ala"));
@@ -338,7 +363,8 @@ private void testCheckpointNullStatisticsForRowType(Consumer<String> sqlExecutor
 
             // Assert counting non null entries can be computed from just metadata
             String explainCountNotNull = getOnlyElement(onDelta().executeQuery("EXPLAIN SELECT count(root.entry_two) FROM default." + tableName).column(1));
-            assertThat(explainCountNotNull).matches("== Physical Plan ==\\s*LocalTableScan \\[count\\(root.entry_two AS `entry_two`\\).*]\\s*");
+            String column = databricksRuntimeVersion.equals(DATABRICKS_104_RUNTIME_VERSION) ? "root.entry_two" : "root.entry_two AS `entry_two`";
+            assertThat(explainCountNotNull).matches("== Physical Plan ==\\s*LocalTableScan \\[count\\(" + column + "\\).*]\\s*");
 
             // check both engines can read both tables
             assertThat(onDelta().executeQuery("SELECT count(root.entry_two) FROM default." + tableName))

diff --git a/...ava/io/trino/tests/product/deltalake/TestDeltaLakeDatabricksCreateTableCompatibility.java b/...ava/io/trino/tests/product/deltalake/TestDeltaLakeDatabricksCreateTableCompatibility.java
@@ -14,6 +14,7 @@
 package io.trino.tests.product.deltalake;
 
 import com.google.common.collect.ImmutableList;
+import io.trino.tempto.BeforeTestWithContext;
 import io.trino.tempto.assertions.QueryAssert;
 import org.testng.annotations.Test;
 
@@ -23,8 +24,10 @@
 import static io.trino.tempto.assertions.QueryAssert.assertThat;
 import static io.trino.tests.product.TestGroups.DELTA_LAKE_DATABRICKS;
 import static io.trino.tests.product.TestGroups.PROFILE_SPECIFIC_TESTS;
+import static io.trino.tests.product.deltalake.util.DeltaLakeTestUtils.DATABRICKS_104_RUNTIME_VERSION;
 import static io.trino.tests.product.deltalake.util.DeltaLakeTestUtils.getColumnCommentOnDelta;
 import static io.trino.tests.product.deltalake.util.DeltaLakeTestUtils.getColumnCommentOnTrino;
+import static io.trino.tests.product.deltalake.util.DeltaLakeTestUtils.getDatabricksRuntimeVersion;
 import static io.trino.tests.product.deltalake.util.DeltaLakeTestUtils.getTableCommentOnDelta;
 import static io.trino.tests.product.hive.util.TemporaryHiveTable.randomTableSuffix;
 import static io.trino.tests.product.utils.QueryExecutors.onDelta;
@@ -35,6 +38,15 @@
 public class TestDeltaLakeDatabricksCreateTableCompatibility
         extends BaseTestDeltaLakeS3Storage
 {
+    private String databricksRuntimeVersion;
+
+    @BeforeTestWithContext
+    public void setup()
+    {
+        super.setUp();
+        databricksRuntimeVersion = getDatabricksRuntimeVersion();
+    }
+
     @Test(groups = {DELTA_LAKE_DATABRICKS, PROFILE_SPECIFIC_TESTS})
     public void testDatabricksCanReadInitialCreateTable()
     {
@@ -49,11 +61,22 @@ public void testDatabricksCanReadInitialCreateTable()
         try {
             assertThat(onDelta().executeQuery("SHOW TABLES FROM default LIKE '" + tableName + "'")).contains(row("default", tableName, false));
             assertThat(onDelta().executeQuery("SELECT count(*) FROM default." + tableName)).contains(row(0));
-            String showCreateTable = format(
-                    "CREATE TABLE `default`.`%s` (\n  `integer` INT,\n  `string` STRING,\n  `timetz` TIMESTAMP)\nUSING DELTA\nLOCATION 's3://%s/%s'\n",
-                    tableName,
-                    bucketName,
-                    tableDirectory);
+            String showCreateTable;
+            if (databricksRuntimeVersion.equals(DATABRICKS_104_RUNTIME_VERSION)) {
+                showCreateTable = format(
+                        "CREATE TABLE spark_catalog.default.%s (\n  integer INT,\n  string STRING,\n  timetz TIMESTAMP)\nUSING delta\nLOCATION 's3://%s/%s'\n%s",
+                        tableName,
+                        bucketName,
+                        tableDirectory,
+                        getDatabricks104DefaultTableProperties());
+            }
+            else {
+                showCreateTable = format(
+                        "CREATE TABLE `default`.`%s` (\n  `integer` INT,\n  `string` STRING,\n  `timetz` TIMESTAMP)\nUSING DELTA\nLOCATION 's3://%s/%s'\n",
+                        tableName,
+                        bucketName,
+                        tableDirectory);
+            }
             assertThat(onDelta().executeQuery("SHOW CREATE TABLE default." + tableName))
                     .containsExactlyInOrder(row(showCreateTable));
             testInsert(tableName, ImmutableList.of());
@@ -79,12 +102,24 @@ public void testDatabricksCanReadInitialCreatePartitionedTable()
         try {
             assertThat(onDelta().executeQuery("SHOW TABLES LIKE '" + tableName + "'")).contains(row("default", tableName, false));
             assertThat(onDelta().executeQuery("SELECT count(*) FROM " + tableName)).contains(row(0));
-            String showCreateTable = format(
-                    "CREATE TABLE `default`.`%s` (\n  `integer` INT,\n  `string` STRING,\n  `timetz` TIMESTAMP)\nUSING DELTA\n" +
-                            "PARTITIONED BY (string)\nLOCATION 's3://%s/%s'\n",
-                    tableName,
-                    bucketName,
-                    tableDirectory);
+            String showCreateTable;
+            if (databricksRuntimeVersion.equals(DATABRICKS_104_RUNTIME_VERSION)) {
+                showCreateTable = format(
+                        "CREATE TABLE spark_catalog.default.%s (\n  integer INT,\n  string STRING,\n  timetz TIMESTAMP)\nUSING delta\n" +
+                                "PARTITIONED BY (string)\nLOCATION 's3://%s/%s'\n%s",
+                        tableName,
+                        bucketName,
+                        tableDirectory,
+                        getDatabricks104DefaultTableProperties());
+            }
+            else {
+                showCreateTable = format(
+                        "CREATE TABLE `default`.`%s` (\n  `integer` INT,\n  `string` STRING,\n  `timetz` TIMESTAMP)\nUSING DELTA\n" +
+                                "PARTITIONED BY (string)\nLOCATION 's3://%s/%s'\n",
+                        tableName,
+                        bucketName,
+                        tableDirectory);
+            }
             assertThat(onDelta().executeQuery("SHOW CREATE TABLE " + tableName)).containsExactlyInOrder(row(showCreateTable));
             testInsert(tableName, ImmutableList.of());
         }
@@ -108,11 +143,22 @@ public void testDatabricksCanReadInitialCreateTableAs()
         try {
             assertThat(onDelta().executeQuery("SHOW TABLES FROM default LIKE '" + tableName + "'")).contains(row("default", tableName, false));
             assertThat(onDelta().executeQuery("SELECT count(*) FROM default." + tableName)).contains(row(3));
-            String showCreateTable = format(
-                    "CREATE TABLE `default`.`%s` (\n  `integer` INT,\n  `string` STRING,\n  `timetz` TIMESTAMP)\nUSING DELTA\nLOCATION 's3://%s/%s'\n",
-                    tableName,
-                    bucketName,
-                    tableDirectory);
+            String showCreateTable;
+            if (databricksRuntimeVersion.equals(DATABRICKS_104_RUNTIME_VERSION)) {
+                showCreateTable = format(
+                        "CREATE TABLE spark_catalog.default.%s (\n  integer INT,\n  string STRING,\n  timetz TIMESTAMP)\nUSING delta\nLOCATION 's3://%s/%s'\n%s",
+                        tableName,
+                        bucketName,
+                        tableDirectory,
+                        getDatabricks104DefaultTableProperties());
+            }
+            else {
+                showCreateTable = format(
+                        "CREATE TABLE `default`.`%s` (\n  `integer` INT,\n  `string` STRING,\n  `timetz` TIMESTAMP)\nUSING DELTA\nLOCATION 's3://%s/%s'\n",
+                        tableName,
+                        bucketName,
+                        tableDirectory);
+            }
             assertThat(onDelta().executeQuery("SHOW CREATE TABLE default." + tableName)).containsExactlyInOrder(row(showCreateTable));
             testInsert(
                     tableName,
@@ -141,12 +187,24 @@ public void testDatabricksCanReadInitialCreatePartitionedTableAs()
         try {
             assertThat(onDelta().executeQuery("SHOW TABLES LIKE '" + tableName + "'")).contains(row("default", tableName, false));
             assertThat(onDelta().executeQuery("SELECT count(*) FROM " + tableName)).contains(row(3));
-            String showCreateTable = format(
-                    "CREATE TABLE `default`.`%s` (\n  `integer` INT,\n  `string` STRING,\n  `timetz` TIMESTAMP)\nUSING DELTA\n" +
-                            "PARTITIONED BY (string)\nLOCATION 's3://%s/%s'\n",
-                    tableName,
-                    bucketName,
-                    tableDirectory);
+            String showCreateTable;
+            if (databricksRuntimeVersion.equals(DATABRICKS_104_RUNTIME_VERSION)) {
+                showCreateTable = format(
+                        "CREATE TABLE spark_catalog.default.%s (\n  integer INT,\n  string STRING,\n  timetz TIMESTAMP)\nUSING delta\n" +
+                                "PARTITIONED BY (string)\nLOCATION 's3://%s/%s'\n%s",
+                        tableName,
+                        bucketName,
+                        tableDirectory,
+                        getDatabricks104DefaultTableProperties());
+            }
+            else {
+                showCreateTable = format(
+                        "CREATE TABLE `default`.`%s` (\n  `integer` INT,\n  `string` STRING,\n  `timetz` TIMESTAMP)\nUSING DELTA\n" +
+                                "PARTITIONED BY (string)\nLOCATION 's3://%s/%s'\n",
+                        tableName,
+                        bucketName,
+                        tableDirectory);
+            }
             assertThat(onDelta().executeQuery("SHOW CREATE TABLE " + tableName)).containsExactlyInOrder(row(showCreateTable));
             testInsert(
                     tableName,
@@ -241,4 +299,12 @@ public void testCreateTableWithColumnCommentOnDelta()
             onDelta().executeQuery("DROP TABLE default." + tableName);
         }
     }
+
+    private String getDatabricks104DefaultTableProperties()
+    {
+        return "TBLPROPERTIES (\n" +
+                "  'Type' = 'EXTERNAL',\n" +
+                "  'delta.minReaderVersion' = '1',\n" +
+                "  'delta.minWriterVersion' = '2')\n";
+    }
 }
diff --git a/...ain/java/io/trino/tests/product/deltalake/TestDeltaLakeDatabricksInsertCompatibility.java b/...ain/java/io/trino/tests/product/deltalake/TestDeltaLakeDatabricksInsertCompatibility.java
@@ -14,6 +14,7 @@
 package io.trino.tests.product.deltalake;
 
 import com.google.common.collect.ImmutableList;
+import io.trino.tempto.BeforeTestWithContext;
 import io.trino.tempto.assertions.QueryAssert.Row;
 import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
@@ -30,6 +31,8 @@
 import static io.trino.tests.product.TestGroups.DELTA_LAKE_EXCLUDE_73;
 import static io.trino.tests.product.TestGroups.DELTA_LAKE_OSS;
 import static io.trino.tests.product.TestGroups.PROFILE_SPECIFIC_TESTS;
+import static io.trino.tests.product.deltalake.util.DeltaLakeTestUtils.DATABRICKS_104_RUNTIME_VERSION;
+import static io.trino.tests.product.deltalake.util.DeltaLakeTestUtils.getDatabricksRuntimeVersion;
 import static io.trino.tests.product.hive.util.TemporaryHiveTable.randomTableSuffix;
 import static io.trino.tests.product.utils.QueryExecutors.onDelta;
 import static io.trino.tests.product.utils.QueryExecutors.onTrino;
@@ -38,6 +41,15 @@
 public class TestDeltaLakeDatabricksInsertCompatibility
         extends BaseTestDeltaLakeS3Storage
 {
+    private String databricksRuntimeVersion;
+
+    @BeforeTestWithContext
+    public void setup()
+    {
+        super.setUp();
+        databricksRuntimeVersion = getDatabricksRuntimeVersion();
+    }
+
     @Test(groups = {DELTA_LAKE_DATABRICKS, DELTA_LAKE_OSS, PROFILE_SPECIFIC_TESTS})
     public void testInsertCompatibility()
     {
@@ -331,7 +343,7 @@ private void testCompression(boolean optimizedWriter, String compressionCodec)
                 assertThat(onTrino().executeQuery("SELECT * FROM " + trinoTableName))
                         .containsOnly(expected);
 
-                if ("ZSTD".equals(compressionCodec)) {
+                if ("ZSTD".equals(compressionCodec) && !databricksRuntimeVersion.equals(DATABRICKS_104_RUNTIME_VERSION)) {
                     assertQueryFailure(() -> onDelta().executeQuery("SELECT * FROM default." + tableName))
                             .hasMessageContaining("java.lang.ClassNotFoundException: org.apache.hadoop.io.compress.ZStandardCodec");
                 }