Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 10 additions & 1 deletion docs/docs/spark-ddl.md
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,16 @@ 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.

### `PARTITIONED BY`

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import static org.apache.iceberg.TableProperties.GC_ENABLED_DEFAULT;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
Expand All @@ -36,6 +37,8 @@
import org.apache.iceberg.HasTableOperations;
import org.apache.iceberg.MetadataTableType;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SortField;
import org.apache.iceberg.SortOrder;
import org.apache.iceberg.Transaction;
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.Namespace;
Expand All @@ -44,6 +47,7 @@
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;
Expand Down Expand Up @@ -82,6 +86,7 @@
import org.apache.spark.sql.connector.catalog.TableChange.ColumnChange;
import org.apache.spark.sql.connector.catalog.TableChange.RemoveProperty;
import org.apache.spark.sql.connector.catalog.TableChange.SetProperty;
import org.apache.spark.sql.connector.catalog.TableInfo;
import org.apache.spark.sql.connector.catalog.TableSummary;
import org.apache.spark.sql.connector.catalog.View;
import org.apache.spark.sql.connector.expressions.Transform;
Expand Down Expand Up @@ -206,11 +211,45 @@ public Table createTable(
Identifier ident, StructType schema, Transform[] transforms, Map<String, String> properties)
throws TableAlreadyExistsException {
Schema icebergSchema = SparkSchemaUtil.convert(schema);
return createTable(ident, icebergSchema, transforms, 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
// properties and sort order, then let user-specified LIKE options (in tableInfo) take
// precedence.
Schema icebergSchema = SparkSchemaUtil.convert(tableInfo.schema());
Map<String, String> properties = new HashMap<>();
SortOrder sortOrder = SortOrder.unsorted();

if (sourceTable instanceof SparkTable) {
org.apache.iceberg.Table sourceIcebergTable = ((SparkTable) sourceTable).table();
properties.putAll(sourceIcebergTable.properties());
sortOrder =
copySortOrder(sourceIcebergTable.schema(), icebergSchema, sourceIcebergTable.sortOrder());
}

properties.putAll(tableInfo.properties());

return createTable(ident, icebergSchema, tableInfo.partitions(), properties, sortOrder);
}

private Table createTable(
Identifier ident,
Schema icebergSchema,
Transform[] transforms,
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, transforms))
.withSortOrder(sortOrder)
.withLocation(properties.get("location"))
.withProperties(Spark3Util.rebuildCreateProperties(properties))
.create();
Expand All @@ -220,6 +259,24 @@ public Table createTable(
}
}

private static SortOrder copySortOrder(
Schema sourceSchema, Schema targetSchema, SortOrder sourceSortOrder) {
if (sourceSortOrder.isUnsorted()) {
return SortOrder.unsorted();
}

SortOrder.Builder builder = SortOrder.builderFor(targetSchema);
for (SortField field : sourceSortOrder.fields()) {
String sourceName = sourceSchema.findColumnName(field.sourceId());
builder.sortBy(
Expressions.transform(sourceName, field.transform()),
field.direction(),
field.nullOrder());
}

return builder.build();
}

@Override
public StagedTable stageCreate(
Identifier ident, StructType schema, Transform[] transforms, Map<String, String> properties)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,7 @@
import org.apache.spark.sql.connector.catalog.Table;
import org.apache.spark.sql.connector.catalog.TableCatalog;
import org.apache.spark.sql.connector.catalog.TableChange;
import org.apache.spark.sql.connector.catalog.TableInfo;
import org.apache.spark.sql.connector.catalog.TableSummary;
import org.apache.spark.sql.connector.catalog.View;
import org.apache.spark.sql.connector.catalog.ViewCatalog;
Expand Down Expand Up @@ -252,6 +253,19 @@ public Table createTable(
}
}

@Override
public Table createTableLike(Identifier ident, TableInfo tableInfo, Table sourceTable)
throws TableAlreadyExistsException, NoSuchNamespaceException {
checkViewNotExists(ident);

String provider = tableInfo.properties().get("provider");
if (useIceberg(provider)) {
return icebergCatalog.createTableLike(ident, tableInfo, sourceTable);
} else {
return getSessionCatalog().createTableLike(ident, tableInfo, sourceTable);
}
}

@Override
public StagedTable stageCreate(
Identifier ident, StructType schema, Transform[] partitions, Map<String, String> properties)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -31,9 +31,12 @@
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.hadoop.HadoopCatalog;
import org.apache.iceberg.spark.CatalogTestBase;
Expand Down Expand Up @@ -69,6 +72,102 @@ public void testTransformIgnoreCase() {
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").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
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
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))
Expand Down
Loading