rdblue commented on a change in pull request #1156: URL: https://github.com/apache/iceberg/pull/1156#discussion_r451729723
########## File path: spark3/src/test/java/org/apache/iceberg/spark/sql/TestCreateTableAsSelect.java ########## @@ -0,0 +1,322 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.iceberg.spark.sql; + +import java.util.Map; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.relocated.com.google.common.collect.Iterables; +import org.apache.iceberg.spark.SparkCatalogTestBase; +import org.apache.iceberg.types.Types; +import org.junit.After; +import org.junit.Assert; +import org.junit.Test; + +import static org.apache.spark.sql.functions.col; +import static org.apache.spark.sql.functions.lit; +import static org.apache.spark.sql.functions.when; + +public class TestCreateTableAsSelect extends SparkCatalogTestBase { + + private final String sourceName; + + public TestCreateTableAsSelect(String catalogName, String implementation, Map<String, String> config) { + super(catalogName, implementation, config); + this.sourceName = tableName("source"); + + sql("CREATE TABLE IF NOT EXISTS %s (id bigint NOT NULL, data string) " + + "USING iceberg PARTITIONED BY (truncate(id, 3))", sourceName); + sql("INSERT INTO %s VALUES (1, 'a'), (2, 'b'), (3, 'c')", sourceName); + } + + @After + public void removeTables() { + sql("DROP TABLE IF EXISTS %s", tableName); + } + + @Test + public void testUnpartitionedCTAS() { + sql("CREATE TABLE %s USING iceberg AS SELECT * FROM %s", tableName, sourceName); + + Schema expectedSchema = new Schema( + Types.NestedField.optional(1, "id", Types.LongType.get()), + Types.NestedField.optional(2, "data", Types.StringType.get()) + ); + + Table ctasTable = validationCatalog.loadTable(tableIdent); + + Assert.assertEquals("Should have expected nullable schema", + expectedSchema.asStruct(), ctasTable.schema().asStruct()); + Assert.assertEquals("Should be an unpartitioned table", + 0, ctasTable.spec().fields().size()); + assertEquals("Should have rows matching the source table", + sql("SELECT * FROM %s ORDER BY id", sourceName), + sql("SELECT * FROM %s ORDER BY id", tableName)); + } + + @Test + public void testPartitionedCTAS() { + sql("CREATE TABLE %s USING iceberg PARTITIONED BY (id) AS SELECT * FROM %s ORDER BY id", tableName, sourceName); + + Schema expectedSchema = new Schema( + Types.NestedField.optional(1, "id", Types.LongType.get()), + Types.NestedField.optional(2, "data", Types.StringType.get()) + ); + + PartitionSpec expectedSpec = PartitionSpec.builderFor(expectedSchema) + .identity("id") + .build(); + + Table ctasTable = validationCatalog.loadTable(tableIdent); + + Assert.assertEquals("Should have expected nullable schema", + expectedSchema.asStruct(), ctasTable.schema().asStruct()); + Assert.assertEquals("Should be partitioned by id", + expectedSpec, ctasTable.spec()); + assertEquals("Should have rows matching the source table", + sql("SELECT * FROM %s ORDER BY id", sourceName), + sql("SELECT * FROM %s ORDER BY id", tableName)); + } + + @Test + public void testRTAS() { + sql("CREATE TABLE %s USING iceberg AS SELECT * FROM %s", tableName, sourceName); + + assertEquals("Should have rows matching the source table", + sql("SELECT * FROM %s ORDER BY id", sourceName), + sql("SELECT * FROM %s ORDER BY id", tableName)); + + sql("REPLACE TABLE %s USING iceberg PARTITIONED BY (part) AS " + + "SELECT id, data, CASE WHEN (id %% 2) = 0 THEN 'even' ELSE 'odd' END AS part " + + "FROM %s ORDER BY 3, 1", tableName, sourceName); + + // spark_catalog does not use an atomic replace, so the table history and old spec is dropped + // the other catalogs do use atomic replace, so the spec id is incremented + boolean isAtomic = !"spark_catalog".equals(catalogName); Review comment: This behavior is fixed in #1183. If this goes in first, I'll add the fix to that PR before merging. ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
