This is an automated email from the ASF dual-hosted git repository.
szehon-ho pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 53a86256e2 Spark 4.2: Support CREATE TABLE ... LIKE ... (#17954)
53a86256e2 is described below
commit 53a86256e24dda81733630b37e561787551cf81f
Author: Rahul Shivu Mahadev <[email protected]>
AuthorDate: Wed Oct 14 15:27:45 2026 -0700
Spark 4.2: Support CREATE TABLE ... LIKE ... (#17954)
---
docs/docs/spark-ddl.md | 15 +-
.../org/apache/iceberg/spark/SparkCatalog.java | 139 ++++++++-
.../apache/iceberg/spark/SparkSessionCatalog.java | 28 ++
.../apache/iceberg/spark/source/SparkTable.java | 5 +
.../iceberg/spark/TestSparkSessionCatalog.java | 129 ++++++++
.../apache/iceberg/spark/sql/TestCreateTable.java | 324 +++++++++++++++++++++
6 files changed, 636 insertions(+), 4 deletions(-)
diff --git a/docs/docs/spark-ddl.md b/docs/docs/spark-ddl.md
index 9fa6c0e7d3..09c97bffdb 100644
--- a/docs/docs/spark-ddl.md
+++ b/docs/docs/spark-ddl.md
@@ -44,7 +44,20 @@ Table create commands, including CTAS and RTAS, support the
full range of Spark
Create commands may also set the default format with the `USING` clause. This
is only supported for `SparkCatalog` because Spark handles the `USING` clause
differently for the built-in catalog.
-`CREATE TABLE ... LIKE ...` syntax is not supported.
+Spark 4.2 and later can create an Iceberg table from an existing table:
+
+```sql
+CREATE TABLE prod.db.sample_copy LIKE prod.db.sample;
+```
+
+`CREATE TABLE ... LIKE ...` copies the source schema, partitioning, sort
order, and
+table properties. The new table does not copy snapshots, data, metadata
history,
+or the source table location. Unless `LOCATION` is specified for the new table,
+the target catalog assigns its default location.
+
+Copying column default values from a source that is not an Iceberg table is not
+supported yet, so `CREATE TABLE ... LIKE ...` fails if such a source has a
column
+with a default value.
### `PARTITIONED BY`
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkCatalog.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkCatalog.java
index 9e2e7138e0..a0bdfea817 100644
--- a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkCatalog.java
+++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkCatalog.java
@@ -35,8 +35,14 @@ import org.apache.iceberg.CatalogUtil;
import org.apache.iceberg.EnvironmentContext;
import org.apache.iceberg.HasTableOperations;
import org.apache.iceberg.MetadataTableType;
+import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
+import org.apache.iceberg.SortField;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.TableUtil;
import org.apache.iceberg.Transaction;
+import org.apache.iceberg.UnboundPartitionSpec;
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.SupportsNamespaces;
@@ -44,6 +50,7 @@ import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.catalog.ViewCatalog;
import org.apache.iceberg.exceptions.AlreadyExistsException;
import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.hadoop.HadoopCatalog;
import org.apache.iceberg.hadoop.HadoopTables;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
@@ -60,6 +67,7 @@ import org.apache.iceberg.spark.source.SparkChangelogTable;
import org.apache.iceberg.spark.source.SparkTable;
import org.apache.iceberg.spark.source.SparkView;
import org.apache.iceberg.spark.source.StagedSparkTable;
+import org.apache.iceberg.types.TypeUtil;
import org.apache.iceberg.util.Pair;
import org.apache.iceberg.util.PropertyUtil;
import org.apache.iceberg.view.UpdateViewProperties;
@@ -133,6 +141,14 @@ public class SparkCatalog extends BaseCatalog {
ViewProperties.METADATA_COMPRESSION,
ViewProperties.WRITE_METADATA_LOCATION,
ViewProperties.REPLACE_DROP_DIALECT_ALLOWED);
+ private static final Set<String> COLUMN_PROPERTY_PREFIXES =
+ ImmutableSet.of(
+ TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX,
+ TableProperties.PARQUET_BLOOM_FILTER_COLUMN_ENABLED_PREFIX,
+ TableProperties.PARQUET_BLOOM_FILTER_COLUMN_FPP_PREFIX,
+ TableProperties.PARQUET_BLOOM_FILTER_COLUMN_NDV_PREFIX,
+ TableProperties.PARQUET_COLUMN_STATS_ENABLED_PREFIX,
+ TableProperties.PARQUET_DICT_ENCODING_ENABLED_COLUMN_PREFIX);
private enum ViewCommit {
CREATE,
@@ -204,13 +220,74 @@ public class SparkCatalog extends BaseCatalog {
public Table createTable(Identifier ident, TableInfo tableInfo)
throws TableAlreadyExistsException, NoSuchNamespaceException {
Schema icebergSchema = SparkSchemaUtil.convertWithDefaults(tableInfo);
+ return createTable(
+ ident,
+ icebergSchema,
+ Spark3Util.toPartitionSpec(icebergSchema, tableInfo.partitions()),
+ tableInfo.properties(),
+ SortOrder.unsorted());
+ }
+
+ @Override
+ public Table createTableLike(Identifier ident, TableInfo tableInfo, Table
sourceTable)
+ throws TableAlreadyExistsException, NoSuchNamespaceException {
+ // Spark intentionally excludes the source table's properties from
tableInfo and leaves it to
+ // the connector to decide which to clone via sourceTable. Clone the
source Iceberg table's
+ // schema, partition spec, properties and sort order, then let
user-specified LIKE options (in
+ // tableInfo) take precedence.
+ Schema icebergSchema;
+ PartitionSpec spec;
+ Map<String, String> properties = Maps.newHashMap();
+ SortOrder sortOrder = SortOrder.unsorted();
+
+ if (sourceTable instanceof SparkTable sparkTable) {
+ org.apache.iceberg.Table sourceIcebergTable = sparkTable.table();
+ icebergSchema = sparkTable.icebergSchema();
+ spec = copyPartitionSpec(icebergSchema, sourceIcebergTable.spec());
+ properties.putAll(
+ copyColumnProperties(
+ sourceIcebergTable.properties(), sourceIcebergTable.schema(),
icebergSchema));
+ properties.remove(TableProperties.WRITE_METADATA_LOCATION);
+ properties.remove(TableProperties.WRITE_DATA_LOCATION);
+ properties.remove(TableProperties.OBJECT_STORE_PATH);
+ properties.remove(TableProperties.WRITE_FOLDER_STORAGE_LOCATION);
+ properties.remove(TableProperties.DEFAULT_NAME_MAPPING);
+ properties.put(
+ TableProperties.FORMAT_VERSION,
+ String.valueOf(TableUtil.formatVersion(sourceIcebergTable)));
+ sortOrder = copySortOrder(icebergSchema, sourceIcebergTable.sortOrder());
+ } else {
+ // Spark passes the column defaults of a non-Iceberg source as SQL text
only, which
+ // convertWithDefaults rejects, so cloning a source column with a
default fails for now
+ icebergSchema = SparkSchemaUtil.convertWithDefaults(tableInfo);
+ spec = Spark3Util.toPartitionSpec(icebergSchema, tableInfo.partitions());
+ }
+
+ String provider = tableInfo.properties().get(TableCatalog.PROP_PROVIDER);
+ if (provider != null && !"iceberg".equalsIgnoreCase(provider)) {
+ properties.remove(TableProperties.DEFAULT_FILE_FORMAT);
+ }
+
+ properties.putAll(tableInfo.properties());
+
+ return createTable(ident, icebergSchema, spec, properties, sortOrder);
+ }
+
+ private Table createTable(
+ Identifier ident,
+ Schema icebergSchema,
+ PartitionSpec spec,
+ Map<String, String> properties,
+ SortOrder sortOrder)
+ throws TableAlreadyExistsException {
try {
Catalog.TableBuilder builder = newBuilder(ident, icebergSchema);
org.apache.iceberg.Table icebergTable =
builder
- .withPartitionSpec(Spark3Util.toPartitionSpec(icebergSchema,
tableInfo.partitions()))
- .withLocation(tableInfo.properties().get("location"))
-
.withProperties(Spark3Util.rebuildCreateProperties(tableInfo.properties()))
+ .withPartitionSpec(spec)
+ .withSortOrder(sortOrder)
+ .withLocation(properties.get("location"))
+ .withProperties(Spark3Util.rebuildCreateProperties(properties))
.create();
return new SparkTable(icebergTable);
} catch (AlreadyExistsException e) {
@@ -218,6 +295,62 @@ public class SparkCatalog extends BaseCatalog {
}
}
+ /**
+ * Returns the source properties with column-specific properties renamed or
dropped to match the
+ * given schema, since those properties name columns of the current schema,
which can differ from
+ * the schema of a pinned snapshot.
+ */
+ private static Map<String, String> copyColumnProperties(
+ Map<String, String> properties, Schema currentSchema, Schema schema) {
+ List<String> deletedColumns = Lists.newArrayList();
+ Map<String, String> renamedColumns = Maps.newHashMap();
+ TypeUtil.indexByName(currentSchema.asStruct())
+ .forEach(
+ (name, id) -> {
+ String nameInSchema = schema.findColumnName(id);
+ if (nameInSchema == null) {
+ deletedColumns.add(name);
+ } else if (!nameInSchema.equals(name)) {
+ renamedColumns.put(name, nameInSchema);
+ }
+ });
+
+ return PropertyUtil.applySchemaChanges(
+ properties, deletedColumns, renamedColumns, COLUMN_PROPERTY_PREFIXES);
+ }
+
+ private static PartitionSpec copyPartitionSpec(Schema schema, PartitionSpec
sourceSpec) {
+ UnboundPartitionSpec spec = sourceSpec.toUnbound();
+ for (int index = sourceSpec.fields().size() - 1; index >= 0; index -= 1) {
+ if (schema.findField(sourceSpec.fields().get(index).sourceId()) == null)
{
+ spec.fields().remove(index);
+ }
+ }
+
+ return spec.bind(schema);
+ }
+
+ private static SortOrder copySortOrder(Schema schema, SortOrder
sourceSortOrder) {
+ if (sourceSortOrder.isUnsorted()) {
+ return SortOrder.unsorted();
+ }
+
+ SortOrder.Builder builder = SortOrder.builderFor(schema);
+ for (SortField field : sourceSortOrder.fields()) {
+ String sourceName = schema.findColumnName(field.sourceId());
+ if (sourceName == null) {
+ continue;
+ }
+
+ builder.sortBy(
+ Expressions.transform(sourceName, field.transform()),
+ field.direction(),
+ field.nullOrder());
+ }
+
+ return builder.build();
+ }
+
@Override
public StagedTable stageCreate(Identifier ident, TableInfo tableInfo)
throws TableAlreadyExistsException {
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkSessionCatalog.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkSessionCatalog.java
index d8a42eec97..457c46edb6 100644
---
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkSessionCatalog.java
+++
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkSessionCatalog.java
@@ -30,6 +30,7 @@ import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.AlreadyExistsException;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
import org.apache.iceberg.spark.source.HasIcebergCatalog;
import org.apache.spark.sql.SparkSession;
@@ -250,6 +251,33 @@ public class SparkSessionCatalog<
}
}
+ @Override
+ public Table createTableLike(Identifier ident, TableInfo tableInfo, Table
sourceTable)
+ throws TableAlreadyExistsException, NoSuchNamespaceException {
+ checkViewNotExists(ident);
+
+ String provider = tableInfo.properties().get("provider");
+ if (provider == null) {
+ provider = sourceTable.properties().get("provider");
+ }
+
+ if (useIceberg(provider)) {
+ return icebergCatalog.createTableLike(ident, tableInfo, sourceTable);
+ } else {
+ // like Spark's CREATE TABLE LIKE, copy only the schema and partitioning
of the source. Its
+ // properties, such as location and external, describe the source and
are not carried over.
+ Map<String, String> properties = Maps.newHashMap(tableInfo.properties());
+ properties.put(TableCatalog.PROP_PROVIDER, provider);
+ TableInfo targetInfo =
+ new TableInfo.Builder()
+ .withColumns(sourceTable.columns())
+ .withPartitions(sourceTable.partitioning())
+ .withProperties(properties)
+ .build();
+ return getSessionCatalog().createTable(ident, targetInfo);
+ }
+ }
+
@Override
public StagedTable stageCreate(Identifier ident, TableInfo info)
throws TableAlreadyExistsException, NoSuchNamespaceException {
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java
index 9cd170f352..1e767f2e8c 100644
---
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java
+++
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java
@@ -135,6 +135,11 @@ public class SparkTable extends BaseSparkTable
return new SparkTable(table(), newBranch);
}
+ /** Returns the Iceberg schema pinned when this table was loaded. */
+ public Schema icebergSchema() {
+ return schema;
+ }
+
public Long snapshotId() {
return snapshot != null ? snapshot.snapshotId() : null;
}
diff --git
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkSessionCatalog.java
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkSessionCatalog.java
index 4438e22cf5..61b0eaa15d 100644
---
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkSessionCatalog.java
+++
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkSessionCatalog.java
@@ -22,6 +22,7 @@ import static
org.apache.hadoop.hive.conf.HiveConf.ConfVars.METASTOREURIS;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
@@ -29,7 +30,9 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.mockito.Mockito.withSettings;
+import java.nio.file.Path;
import java.util.Collections;
+import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.AlreadyExistsException;
import org.apache.spark.sql.catalyst.analysis.NoSuchNamespaceException;
import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
@@ -54,12 +57,18 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.ArgumentCaptor;
public class TestSparkSessionCatalog extends TestBase {
private final String envHmsUriKey = "spark.hadoop." + METASTOREURIS.varname;
private final String catalogHmsUriKey =
"spark.sql.catalog.spark_catalog.uri";
private final String hmsUri = hiveConf.get(METASTOREURIS.varname);
+ @TempDir private Path temp;
+
@BeforeAll
public static void setUpCatalog() {
spark
@@ -75,6 +84,126 @@ public class TestSparkSessionCatalog extends TestBase {
spark.conf().set(catalogHmsUriKey, hmsUri);
}
+ @ParameterizedTest
+ @ValueSource(strings = {"parquet", "avro", "orc"})
+ void createTableLikeDelegatesUsingSourceProvider(String provider) throws
Exception {
+ Identifier ident = Identifier.of(new String[] {"default"}, "target");
+ TableInfo tableInfo = new TableInfo.Builder().build();
+ TableInfo sourceInfo =
+ new TableInfo.Builder().withSchema(new StructType().add("id",
"long")).build();
+ Table source = mock(Table.class);
+ when(source.columns()).thenReturn(sourceInfo.columns());
+ when(source.partitioning()).thenReturn(sourceInfo.partitions());
+ when(source.properties()).thenReturn(Collections.singletonMap("provider",
provider));
+ Table target = mock(Table.class);
+ TableCatalog sessionCatalog = sessionCatalogWithViews();
+ when(sessionCatalog.createTable(eq(ident),
any(TableInfo.class))).thenReturn(target);
+ TableCatalog icebergCatalog = icebergCatalogWithViews();
+
+ SparkSessionCatalog<?> catalog = new
CatalogWithIcebergViews<>(icebergCatalog);
+ catalog.initialize(
+ "spark_catalog",
+ new CaseInsensitiveStringMap(Collections.singletonMap(provider +
"-enabled", "false")));
+ catalog.setDelegateCatalog(sessionCatalog);
+
+ assertThat(catalog.createTableLike(ident, tableInfo,
source)).isSameAs(target);
+ ArgumentCaptor<TableInfo> targetInfo =
ArgumentCaptor.forClass(TableInfo.class);
+ verify(sessionCatalog).createTable(eq(ident), targetInfo.capture());
+ assertThat(targetInfo.getValue().schema()).isEqualTo(sourceInfo.schema());
+ assertThat(targetInfo.getValue().properties()).containsEntry("provider",
provider);
+ verify(icebergCatalog, never()).createTableLike(ident, tableInfo, source);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"parquet", "orc"})
+ void createTableLikeWithoutUsingPreservesSourceProvider(String provider) {
+ String sourceName = "spark_catalog.default.like_source";
+ String targetName = "spark_catalog.default.like_target";
+ String conversionOption = "spark.sql.catalog.spark_catalog." + provider +
"-enabled";
+
+ withSQLConf(
+ Collections.singletonMap(conversionOption, "false"),
+ () -> {
+ spark.sessionState().catalogManager().reset();
+ try {
+ sql("CREATE TABLE %s (id BIGINT) USING %s", sourceName, provider);
+ sql("CREATE TABLE %s LIKE %s", targetName, sourceName);
+
+ assertThat(sql("DESCRIBE TABLE EXTENDED %s", targetName))
+ .anySatisfy(row -> assertThat(row).startsWith("Provider",
provider));
+ } finally {
+ sql("DROP TABLE IF EXISTS %s", targetName);
+ sql("DROP TABLE IF EXISTS %s", sourceName);
+ spark.sessionState().catalogManager().reset();
+ }
+ });
+ }
+
+ @Test
+ void createTableLikeUsingParquetOverridesIcebergProvider() {
+ String sourceName = "spark_catalog.default.like_iceberg_source";
+ String targetName = "spark_catalog.default.like_parquet_target";
+
+ withSQLConf(
+
Collections.singletonMap("spark.sql.catalog.spark_catalog.parquet-enabled",
"false"),
+ () -> {
+ spark.sessionState().catalogManager().reset();
+ try {
+ sql("CREATE TABLE %s (id BIGINT, data STRING) USING iceberg",
sourceName);
+ sql("INSERT INTO %s VALUES (1, 'source')", sourceName);
+
+ sql("CREATE TABLE %s LIKE %s USING parquet", targetName,
sourceName);
+
+ assertThat(sql("DESCRIBE TABLE EXTENDED %s", targetName))
+ .anySatisfy(row -> assertThat(row).startsWith("Provider",
"parquet"));
+ assertThat(spark.table(targetName).schema())
+ .isEqualTo(spark.table(sourceName).schema());
+ assertThat(sql("SELECT * FROM %s", targetName)).isEmpty();
+ } finally {
+ sql("DROP TABLE IF EXISTS %s", targetName);
+ sql("DROP TABLE IF EXISTS %s", sourceName);
+ spark.sessionState().catalogManager().reset();
+ }
+ });
+ }
+
+ @Test
+ void createTableLikeUsingParquetFromExternalIcebergSource() {
+ String sourceName = "spark_catalog.default.like_external_iceberg_source";
+ String targetName = "spark_catalog.default.like_managed_parquet_target";
+ String sourceLocation =
temp.resolve("like_external_iceberg_source").toUri().toString();
+
+ withSQLConf(
+
Collections.singletonMap("spark.sql.catalog.spark_catalog.parquet-enabled",
"false"),
+ () -> {
+ spark.sessionState().catalogManager().reset();
+ try {
+ sql(
+ "CREATE EXTERNAL TABLE %s (id BIGINT, data STRING) USING
iceberg LOCATION '%s'",
+ sourceName, sourceLocation);
+ assertThat(
+ catalog
+ .loadTable(TableIdentifier.of("default",
"like_external_iceberg_source"))
+ .properties())
+ .containsEntry(TableCatalog.PROP_EXTERNAL, "true");
+
+ // the target has no LOCATION, so it must be created as a managed
table rather than
+ // inheriting external=true from the source, which Spark rejects
without a location
+ sql("CREATE TABLE %s LIKE %s USING parquet", targetName,
sourceName);
+
+ assertThat(sql("DESCRIBE TABLE EXTENDED %s", targetName))
+ .anySatisfy(row -> assertThat(row).startsWith("Provider",
"parquet"))
+ .anySatisfy(row -> assertThat(row).startsWith("Type",
"MANAGED"));
+ assertThat(spark.table(targetName).schema())
+ .isEqualTo(spark.table(sourceName).schema());
+ } finally {
+ sql("DROP TABLE IF EXISTS %s", targetName);
+ sql("DROP TABLE IF EXISTS %s", sourceName);
+ spark.sessionState().catalogManager().reset();
+ }
+ });
+ }
+
@Test
public void testValidateHmsUri() {
// HMS uris match
diff --git
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/sql/TestCreateTable.java
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/sql/TestCreateTable.java
index afb4e99648..10ac453c84 100644
---
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/sql/TestCreateTable.java
+++
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/sql/TestCreateTable.java
@@ -31,12 +31,19 @@ import org.apache.iceberg.BaseTable;
import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
+import org.apache.iceberg.SortOrder;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableOperations;
import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.mapping.MappingUtil;
+import org.apache.iceberg.mapping.NameMappingParser;
import org.apache.iceberg.spark.CatalogTestBase;
+import org.apache.iceberg.types.TypeUtil;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.types.Types.NestedField;
import org.apache.iceberg.types.Types.StructType;
@@ -69,6 +76,323 @@ public class TestCreateTable extends CatalogTestBase {
assertThat(validationCatalog.tableExists(tableIdent)).as("Table should
already exist").isTrue();
}
+ @TestTemplate
+ public void testCreateTableLike() {
+ String sourceName = tableName("source");
+ TableIdentifier sourceIdent = TableIdentifier.of(Namespace.of("default"),
"source");
+ Schema schema =
+ new Schema(
+ NestedField.required(1, "id", Types.LongType.get()),
+ NestedField.optional(2, "category", Types.StringType.get()),
+ NestedField.optional(3, "data", Types.StringType.get()));
+ PartitionSpec spec =
+ PartitionSpec.builderFor(schema).identity("category").bucket("id", 16,
"shard").build();
+ SortOrder order =
SortOrder.builderFor(schema).desc("id").asc("data").build();
+
+ try {
+ validationCatalog
+ .buildTable(sourceIdent, schema)
+ .withPartitionSpec(spec)
+ .withSortOrder(order)
+ .withProperty("custom-property", "custom-value")
+ .create();
+
+ Table source = validationCatalog.loadTable(sourceIdent);
+ sql("CREATE TABLE %s LIKE %s", tableName, sourceName);
+
+ Table target = validationCatalog.loadTable(tableIdent);
+
assertThat(target.schema().asStruct()).isEqualTo(source.schema().asStruct());
+ assertThat(target.spec()).isEqualTo(source.spec());
+ assertThat(target.sortOrder().sameOrder(source.sortOrder())).isTrue();
+ assertThat(target.properties()).containsEntry("custom-property",
"custom-value");
+ assertThat(target.location()).isNotEqualTo(source.location());
+ } finally {
+ sql("DROP TABLE IF EXISTS %s", sourceName);
+ }
+ }
+
+ @TestTemplate
+ void createTableLikeUsesSnapshotSchema() {
+ assumeThat(catalogName)
+ .as("Spark session catalog does not support extended table names")
+ .isNotEqualTo("spark_catalog");
+
+ String sourceName = tableName("source");
+ TableIdentifier sourceIdent = TableIdentifier.of(Namespace.of("default"),
"source");
+ Schema schema =
+ new Schema(
+ NestedField.required(1, "id", Types.LongType.get()),
+ NestedField.optional(2, "uuid", Types.UUIDType.get()));
+
+ try {
+ Table source = validationCatalog.createTable(sourceIdent, schema);
+ source.newAppend().commit();
+ long snapshotId = source.currentSnapshot().snapshotId();
+ Schema snapshotSchema = source.schema();
+ sql("ALTER TABLE %s ADD COLUMN data STRING", sourceName);
+
+ sql("CREATE TABLE %s LIKE %s.snapshot_id_%s", tableName, sourceName,
snapshotId);
+
+ Table target = validationCatalog.loadTable(tableIdent);
+
assertThat(target.schema().asStruct()).isEqualTo(snapshotSchema.asStruct());
+ assertThat(target.schema().asStruct())
+
.isNotEqualTo(validationCatalog.loadTable(sourceIdent).schema().asStruct());
+ } finally {
+ sql("DROP TABLE IF EXISTS %s", sourceName);
+ }
+ }
+
+ @TestTemplate
+ void createTableLikeUsesSnapshotSortColumnNames() {
+ assumeThat(catalogName)
+ .as("Spark session catalog does not support extended table names")
+ .isNotEqualTo("spark_catalog");
+
+ String sourceName = tableName("source");
+ TableIdentifier sourceIdent = TableIdentifier.of(Namespace.of("default"),
"source");
+ Schema schema =
+ new Schema(
+ NestedField.required(1, "id", Types.LongType.get()),
+ NestedField.optional(2, "data", Types.StringType.get()));
+ SortOrder order = SortOrder.builderFor(schema).desc("data").build();
+
+ try {
+ Table source =
+ validationCatalog.buildTable(sourceIdent,
schema).withSortOrder(order).create();
+ source.newAppend().commit();
+ long snapshotId = source.currentSnapshot().snapshotId();
+ sql("ALTER TABLE %s RENAME COLUMN data TO payload", sourceName);
+
+ sql("CREATE TABLE %s LIKE %s.snapshot_id_%s", tableName, sourceName,
snapshotId);
+
+ Table target = validationCatalog.loadTable(tableIdent);
+ assertThat(target.schema().findField("data")).isNotNull();
+ assertThat(target.sortOrder().sameOrder(order)).isTrue();
+ } finally {
+ sql("DROP TABLE IF EXISTS %s", sourceName);
+ }
+ }
+
+ @TestTemplate
+ void createTableLikeUsesSnapshotColumnPropertyNames() {
+ assumeThat(catalogName)
+ .as("Spark session catalog does not support extended table names")
+ .isNotEqualTo("spark_catalog");
+
+ String sourceName = tableName("source");
+ TableIdentifier sourceIdent = TableIdentifier.of(Namespace.of("default"),
"source");
+ Schema schema =
+ new Schema(
+ NestedField.required(1, "id", Types.LongType.get()),
+ NestedField.optional(2, "data", Types.StringType.get()));
+ String dataMetrics = TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX +
"data";
+ String payloadMetrics = TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX +
"payload";
+ String extraMetrics = TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX +
"extra";
+
+ try {
+ Table source =
+ validationCatalog
+ .buildTable(sourceIdent, schema)
+ .withProperty(dataMetrics, "full")
+ .create();
+ source.newAppend().commit();
+ long snapshotId = source.currentSnapshot().snapshotId();
+ sql("ALTER TABLE %s RENAME COLUMN data TO payload", sourceName);
+ sql("ALTER TABLE %s ADD COLUMN extra STRING", sourceName);
+ sql("ALTER TABLE %s SET TBLPROPERTIES ('%s'='counts')", sourceName,
extraMetrics);
+
+ sql("CREATE TABLE %s LIKE %s.snapshot_id_%s", tableName, sourceName,
snapshotId);
+
+ Table target = validationCatalog.loadTable(tableIdent);
+ assertThat(target.properties())
+ .containsEntry(dataMetrics, "full")
+ .doesNotContainKeys(payloadMetrics, extraMetrics);
+ } finally {
+ sql("DROP TABLE IF EXISTS %s", sourceName);
+ }
+ }
+
+ @TestTemplate
+ void createTableLikeOmitsSortFieldsAddedAfterSnapshot() {
+ assumeThat(catalogName)
+ .as("Spark session catalog does not support extended table names")
+ .isNotEqualTo("spark_catalog");
+
+ String sourceName = tableName("source");
+ TableIdentifier sourceIdent = TableIdentifier.of(Namespace.of("default"),
"source");
+ Schema schema = new Schema(NestedField.required(1, "id",
Types.LongType.get()));
+ SortOrder order = SortOrder.builderFor(schema).desc("id").build();
+
+ try {
+ Table source =
+ validationCatalog.buildTable(sourceIdent,
schema).withSortOrder(order).create();
+ source.newAppend().commit();
+ long snapshotId = source.currentSnapshot().snapshotId();
+ source.updateSchema().addColumn("data", Types.StringType.get()).commit();
+ source.replaceSortOrder().desc("id").asc("data").commit();
+
+ sql("CREATE TABLE %s LIKE %s.snapshot_id_%s", tableName, sourceName,
snapshotId);
+
+ Table target = validationCatalog.loadTable(tableIdent);
+ assertThat(target.schema().asStruct()).isEqualTo(schema.asStruct());
+ assertThat(target.sortOrder().sameOrder(order)).isTrue();
+ } finally {
+ sql("DROP TABLE IF EXISTS %s", sourceName);
+ }
+ }
+
+ @TestTemplate
+ void createTableLikeOmitsPartitionFieldsWithDeletedSources() {
+ String sourceName = tableName("source");
+ TableIdentifier sourceIdent = TableIdentifier.of(Namespace.of("default"),
"source");
+ Schema schema =
+ new Schema(
+ NestedField.required(1, "id", Types.LongType.get()),
+ NestedField.optional(2, "data", Types.StringType.get()));
+
+ try {
+ Table source =
+ validationCatalog
+ .buildTable(sourceIdent, schema)
+ .withProperty(TableProperties.FORMAT_VERSION, "1")
+ .create();
+ source.updateSpec().addField("data").commit();
+ source.updateSpec().removeField("data").commit();
+ source.updateSchema().deleteColumn("data").commit();
+ source.updateSpec().addField("shard", Expressions.bucket("id",
16)).commit();
+
+ sql("CREATE TABLE %s LIKE %s", tableName, sourceName);
+
+ Table target = validationCatalog.loadTable(tableIdent);
+ PartitionSpec expectedSpec =
+ PartitionSpec.builderFor(target.schema()).bucket("id", 16,
"shard").build();
+ assertThat(target.spec()).isEqualTo(expectedSpec);
+ } finally {
+ sql("DROP TABLE IF EXISTS %s", sourceName);
+ }
+ }
+
+ @TestTemplate
+ void createTableLikeOmitsSourceNameMapping() {
+ String sourceName = tableName("source");
+ TableIdentifier sourceIdent = TableIdentifier.of(Namespace.of("default"),
"source");
+ Schema schema =
+ new Schema(
+ NestedField.required(1, "id", Types.LongType.get()),
+ NestedField.optional(2, "data", Types.StringType.get()));
+
+ try {
+ Table source = validationCatalog.createTable(sourceIdent, schema);
+ source.newAppend().commit();
+ source.updateSchema().deleteColumn("data").commit();
+ source.updateSchema().addColumn("data", Types.UUIDType.get()).commit();
+ source
+ .updateProperties()
+ .set(
+ TableProperties.DEFAULT_NAME_MAPPING,
+ NameMappingParser.toJson(MappingUtil.create(source.schema())))
+ .commit();
+
+ sql("CREATE TABLE %s LIKE %s", tableName, sourceName);
+
+ Table target = validationCatalog.loadTable(tableIdent);
+
assertThat(target.properties()).doesNotContainKey(TableProperties.DEFAULT_NAME_MAPPING);
+ assertThat(target.schema().findField("data").fieldId())
+ .isNotEqualTo(source.schema().findField("data").fieldId());
+ assertThat(target.schema().asStruct())
+
.isEqualTo(TypeUtil.assignIncreasingFreshIds(source.schema()).asStruct());
+ } finally {
+ sql("DROP TABLE IF EXISTS %s", sourceName);
+ }
+ }
+
+ @TestTemplate
+ public void testCreateTableLikeClonesAndOverridesProperties() {
+ String sourceName = tableName("source");
+ TableIdentifier sourceIdent = TableIdentifier.of(Namespace.of("default"),
"source");
+ Schema schema = new Schema(NestedField.required(1, "id",
Types.LongType.get()));
+
+ try {
+ validationCatalog
+ .buildTable(sourceIdent, schema)
+ .withProperty("clone-me", "from-source")
+ .withProperty("override-me", "from-source")
+ .create();
+
+ sql(
+ "CREATE TABLE %s LIKE %s TBLPROPERTIES
('override-me'='from-target')",
+ tableName, sourceName);
+
+ Table target = validationCatalog.loadTable(tableIdent);
+ assertThat(target.properties())
+ .containsEntry("clone-me", "from-source")
+ .containsEntry("override-me", "from-target");
+ } finally {
+ sql("DROP TABLE IF EXISTS %s", sourceName);
+ }
+ }
+
+ @TestTemplate
+ void createTableLikeUsingOverridesSourceFormat() {
+ assumeThat(catalogName)
+ .as("Spark session catalog delegates ORC tables to its session
catalog")
+ .isNotEqualTo("spark_catalog");
+
+ String sourceName = tableName("source");
+ TableIdentifier sourceIdent = TableIdentifier.of(Namespace.of("default"),
"source");
+ Schema schema = new Schema(NestedField.required(1, "id",
Types.LongType.get()));
+
+ try {
+ validationCatalog
+ .buildTable(sourceIdent, schema)
+ .withProperty(TableProperties.DEFAULT_FILE_FORMAT, "parquet")
+ .create();
+
+ sql("CREATE TABLE %s LIKE %s USING orc", tableName, sourceName);
+
+ Table target = validationCatalog.loadTable(tableIdent);
+
assertThat(target.properties()).containsEntry(TableProperties.DEFAULT_FILE_FORMAT,
"orc");
+ } finally {
+ sql("DROP TABLE IF EXISTS %s", sourceName);
+ }
+ }
+
+ @TestTemplate
+ public void testCreateTableLikeIfNotExists() {
+ String sourceName = tableName("source");
+
+ try {
+ sql(
+ "CREATE TABLE %s (id BIGINT, data STRING) "
+ + "USING iceberg TBLPROPERTIES ('source-property'='source')",
+ sourceName);
+ sql(
+ "CREATE TABLE %s (id BIGINT) "
+ + "USING iceberg TBLPROPERTIES ('target-property'='target')",
+ tableName);
+
+ sql("CREATE TABLE IF NOT EXISTS %s LIKE %s", tableName, sourceName);
+
+ Table target = validationCatalog.loadTable(tableIdent);
+ assertThat(target.schema().columns()).hasSize(1);
+ assertThat(target.properties())
+ .containsEntry("target-property", "target")
+ .doesNotContainKey("source-property");
+ } finally {
+ sql("DROP TABLE IF EXISTS %s", sourceName);
+ }
+ }
+
+ @TestTemplate
+ public void testCreateTableLikeMissingSource() {
+ String missingSource = tableName("missing_source");
+
+ assertThatThrownBy(() -> sql("CREATE TABLE %s LIKE %s", tableName,
missingSource))
+ .isInstanceOf(org.apache.spark.sql.AnalysisException.class)
+ .hasMessageContaining("missing_source");
+ assertThat(validationCatalog.tableExists(tableIdent)).isFalse();
+ }
+
@TestTemplate
public void testTransformSingularForm() {
assertThat(validationCatalog.tableExists(tableIdent))
