Add Iceberg MinIO connector smoke test

plugin/trino-hive/src/test/java/io/trino/plugin/hive/containers/HiveMinioDataLake.java

@@ -64,7 +64,7 @@ public HiveMinioDataLake(String bucketName, Map<String, String> hiveHadoopFilesT
         this.hiveHadoop = closer.register(
                 HiveHadoop.builder()
                         .withFilesToMount(ImmutableMap.<String, String>builder()
-                                .put("hive_s3_insert_overwrite/hive-core-site.xml", "/etc/hadoop/conf/core-site.xml")
+                                .put("hive_minio_datalake/hive-core-site.xml", "/etc/hadoop/conf/core-site.xml")
                                 .putAll(hiveHadoopFilesToMount)
                                 .buildOrThrow())
                         .withImage(hiveHadoopImage)

plugin/trino-hive/src/test/resources/hive_s3_insert_overwrite/hive-core-site.xml → plugin/trino-hive/src/test/resources/hive_minio_datalake/hive-core-site.xml

@@ -20,4 +20,8 @@
         <name>fs.s3a.path.style.access</name>
         <value>true</value>
     </property>
+    <property>
+        <name>fs.s3.impl</name>
+        <value>org.apache.hadoop.fs.s3a.S3AFileSystem</value>
+    </property>
 </configuration>

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/BaseIcebergConnectorSmokeTest.java

@@ -0,0 +1,82 @@
+/*
+ * 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.iceberg;
+
+import io.trino.testing.BaseConnectorSmokeTest;
+import io.trino.testing.TestingConnectorBehavior;
+import org.apache.iceberg.FileFormat;
+import org.testng.annotations.Test;
+
+import static java.lang.String.format;
+import static java.util.Objects.requireNonNull;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+public abstract class BaseIcebergConnectorSmokeTest
+        extends BaseConnectorSmokeTest
+{
+    protected final FileFormat format;
+
+    public BaseIcebergConnectorSmokeTest(FileFormat format)
+    {
+        this.format = requireNonNull(format, "format is null");
+    }
+
+    @Override
+    protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
+    {
+        switch (connectorBehavior) {
+            case SUPPORTS_TOPN_PUSHDOWN:
+                return false;
+
+            case SUPPORTS_CREATE_VIEW:
+                return true;
+
+            case SUPPORTS_CREATE_MATERIALIZED_VIEW:
+                return true;
+
+            case SUPPORTS_DELETE:
+                return true;
+            default:
+                return super.hasBehavior(connectorBehavior);
+        }
+    }
+
+    @Test
+    @Override
+    public void testRowLevelDelete()
+    {
+        // Deletes are covered AbstractTestIcebergConnectorTest
+        assertThatThrownBy(super::testRowLevelDelete)
+                .hasStackTraceContaining("This connector only supports delete where one or more identity-transformed partitions are deleted entirely");
+    }
+
+    @Test
+    @Override
+    public void testShowCreateTable()
+    {
+        String schemaName = getSession().getSchema().orElseThrow();
+        assertThat((String) computeScalar("SHOW CREATE TABLE region"))
+                .matches("" +
+                        "CREATE TABLE iceberg." + schemaName + ".region \\(\n" +
+                        "   regionkey bigint,\n" +
+                        "   name varchar,\n" +
+                        "   comment varchar\n" +
+                        "\\)\n" +
+                        "WITH \\(\n" +
+                        "   format = '" + format.name() + "',\n" +
+                        format("   location = '.*/" + schemaName + "/region'\n") +
+                        "\\)");
+    }
+}

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/BaseIcebergMinioConnectorSmokeTest.java

@@ -0,0 +1,87 @@
+/*
+ * 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.iceberg;
+
+import com.google.common.collect.ImmutableMap;
+import io.trino.plugin.hive.containers.HiveMinioDataLake;
+import io.trino.testing.QueryRunner;
+import org.apache.iceberg.FileFormat;
+import org.testng.annotations.Test;
+
+import java.util.Locale;
+import java.util.Optional;
+
+import static io.trino.plugin.hive.containers.HiveMinioDataLake.ACCESS_KEY;
+import static io.trino.plugin.hive.containers.HiveMinioDataLake.SECRET_KEY;
+import static io.trino.plugin.iceberg.IcebergQueryRunner.createIcebergQueryRunner;
+import static io.trino.testing.sql.TestTable.randomTableSuffix;
+import static java.lang.String.format;
+
+public abstract class BaseIcebergMinioConnectorSmokeTest
+        extends BaseIcebergConnectorSmokeTest
+{
+    private final String schemaName;
+    private final String bucketName;
+
+    private HiveMinioDataLake hiveMinioDataLake;
+
+    public BaseIcebergMinioConnectorSmokeTest(FileFormat format)
+    {
+        super(format);
+        this.schemaName = "tpch_" + format.name().toLowerCase(Locale.ENGLISH);
+        this.bucketName = "test-iceberg-minio-smoke-test-" + randomTableSuffix();
+    }
+
+    @Override
+    protected QueryRunner createQueryRunner()
+            throws Exception
+    {
+        this.hiveMinioDataLake = closeAfterClass(new HiveMinioDataLake(bucketName, ImmutableMap.of()));
+        this.hiveMinioDataLake.start();
+        return createIcebergQueryRunner(
+                ImmutableMap.of(),
+                ImmutableMap.<String, String>builder()
+                        .put("iceberg.file-format", format.name())
+                        .put("iceberg.catalog.type", "HIVE_METASTORE")
+                        .put("hive.metastore.uri", "thrift://" + hiveMinioDataLake.getHiveHadoop().getHiveMetastoreEndpoint())
+                        .put("hive.s3.aws-access-key", ACCESS_KEY)
+                        .put("hive.s3.aws-secret-key", SECRET_KEY)
+                        .put("hive.s3.endpoint", "http://" + hiveMinioDataLake.getMinio().getMinioApiEndpoint())
+                        .put("hive.s3.path-style-access", "true")
+                        .put("hive.s3.streaming.part-size", "5MB")
+                        .buildOrThrow(),
+                SchemaInitializer.builder()
+                        .withSchemaName(schemaName)
+                        .withClonedTpchTables(REQUIRED_TPCH_TABLES)
+                        .withSchemaProperties(ImmutableMap.of(
+                                "location", "'s3://" + bucketName + "/" + schemaName + "'"))
+                        .build(),
+                Optional.empty());
+    }
+
+    @Override
+    protected String createSchemaSql(String schemaName)
+    {
+        return "CREATE SCHEMA IF NOT EXISTS " + schemaName + " WITH (location = 's3://" + bucketName + "/" + schemaName + "')";
+    }
+
+    @Test
+    @Override
+    public void testRenameSchema()
+    {
+        assertQueryFails(
+                format("ALTER SCHEMA %s RENAME TO %s", schemaName, schemaName + randomTableSuffix()),
+                "Hive metastore does not support renaming schemas");
+    }
+}

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/IcebergQueryRunner.java

@@ -24,13 +24,11 @@
 import java.io.File;
 import java.nio.file.Path;
 import java.util.HashMap;
-import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 
+import static com.google.common.base.Preconditions.checkState;
 import static io.airlift.testing.Closeables.closeAllSuppress;
-import static io.trino.plugin.tpch.TpchMetadata.TINY_SCHEMA_NAME;
-import static io.trino.testing.QueryAssertions.copyTpchTables;
 import static io.trino.testing.TestingSession.testSessionBuilder;
 import static java.util.Objects.requireNonNull;
 
@@ -42,13 +40,20 @@ public final class IcebergQueryRunner
 
     private IcebergQueryRunner() {}
 
-    public static DistributedQueryRunner createIcebergQueryRunner()
+    public static DistributedQueryRunner createIcebergQueryRunner(TpchTable<?>... tables)
             throws Exception
     {
         return createIcebergQueryRunner(
-                Map.of(),
-                Map.of(),
-                List.of());
+                ImmutableList.copyOf(tables));
+    }
+
+    public static DistributedQueryRunner createIcebergQueryRunner(Iterable<TpchTable<?>> tables)
+            throws Exception
+    {
+        return createIcebergQueryRunner(
+                ImmutableMap.of(),
+                ImmutableMap.of(),
+                tables);
     }
 
     public static DistributedQueryRunner createIcebergQueryRunner(
@@ -70,11 +75,27 @@ public static DistributedQueryRunner createIcebergQueryRunner(
             Iterable<TpchTable<?>> tables,
             Optional<File> metastoreDirectory)
             throws Exception
+    {
+        return createIcebergQueryRunner(
+                extraProperties,
+                connectorProperties,
+                SchemaInitializer.builder()
+                        .withClonedTpchTables(tables)
+                        .build(),
+                metastoreDirectory);
+    }
+
+    public static DistributedQueryRunner createIcebergQueryRunner(
+            Map<String, String> extraProperties,
+            Map<String, String> connectorProperties,
+            SchemaInitializer schemaInitializer,
+            Optional<File> metastoreDirectory)
+            throws Exception
     {
         Builder builder = builder()
                 .setExtraProperties(extraProperties)
                 .setIcebergProperties(connectorProperties)
-                .setInitialTables(tables);
+                .setSchemaInitializer(schemaInitializer);
 
         metastoreDirectory.ifPresent(builder::setMetastoreDirectory);
         return builder.build();
@@ -90,13 +111,12 @@ public static class Builder
     {
         private Optional<File> metastoreDirectory = Optional.empty();
         private ImmutableMap.Builder<String, String> icebergProperties = ImmutableMap.builder();
-        private List<TpchTable<?>> initialTables = ImmutableList.of();
+        private Optional<SchemaInitializer> schemaInitializer = Optional.empty();
 
         protected Builder()
         {
             super(testSessionBuilder()
                     .setCatalog(ICEBERG_CATALOG)
-                    .setSchema("tpch")
                     .build());
         }
 
@@ -121,7 +141,15 @@ public Builder addIcebergProperty(String key, String value)
 
         public Builder setInitialTables(Iterable<TpchTable<?>> initialTables)
         {
-            this.initialTables = ImmutableList.copyOf(requireNonNull(initialTables, "initialTables is null"));
+            setSchemaInitializer(SchemaInitializer.builder().withClonedTpchTables(initialTables).build());
+            return self();
+        }
+
+        public Builder setSchemaInitializer(SchemaInitializer schemaInitializer)
+        {
+            checkState(this.schemaInitializer.isEmpty(), "schemaInitializer is already set");
+            this.schemaInitializer = Optional.of(requireNonNull(schemaInitializer, "schemaInitializer is null"));
+            amendSession(sessionBuilder -> sessionBuilder.setSchema(schemaInitializer.getSchemaName()));
             return self();
         }
 
@@ -134,19 +162,16 @@ public DistributedQueryRunner build()
                 queryRunner.installPlugin(new TpchPlugin());
                 queryRunner.createCatalog("tpch", "tpch");
 
-                Path dataDir = metastoreDirectory.map(File::toPath).orElseGet(() -> queryRunner.getCoordinator().getBaseDataDir().resolve("iceberg_data"));
-
                 queryRunner.installPlugin(new IcebergPlugin());
-                Map<String, String> icebergProperties = new HashMap<>();
-                icebergProperties.put("iceberg.catalog.type", "TESTING_FILE_METASTORE");
-                icebergProperties.put("hive.metastore.catalog.dir", dataDir.toString());
-                icebergProperties.putAll(this.icebergProperties.buildOrThrow());
+                Map<String, String> icebergProperties = new HashMap<>(this.icebergProperties.buildOrThrow());
+                if (!icebergProperties.containsKey("iceberg.catalog.type")) {
+                    Path dataDir = metastoreDirectory.map(File::toPath).orElseGet(() -> queryRunner.getCoordinator().getBaseDataDir().resolve("iceberg_data"));
+                    icebergProperties.put("iceberg.catalog.type", "TESTING_FILE_METASTORE");
+                    icebergProperties.put("hive.metastore.catalog.dir", dataDir.toString());
+                }
 
                 queryRunner.createCatalog(ICEBERG_CATALOG, "iceberg", icebergProperties);
-
-                queryRunner.execute("CREATE SCHEMA tpch");
-
-                copyTpchTables(queryRunner, "tpch", TINY_SCHEMA_NAME, queryRunner.getDefaultSession(), initialTables);
+                schemaInitializer.orElse(SchemaInitializer.builder().build()).accept(queryRunner);
 
                 return queryRunner;
             }