This is an automated email from the ASF dual-hosted git repository.
danielcweeks 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 8f825e6e19 Spark: support default values with DDL (#18250)
8f825e6e19 is described below
commit 8f825e6e19088f5b0db2e5095f49bd18ffafeb92
Author: Daniel Weeks <[email protected]>
AuthorDate: Thu Oct 1 08:17:56 2026 -0700
Spark: support default values with DDL (#18250)
* Spark: support default values with DDL
---
.../action/IcebergDataCompactionBenchmark.java | 9 +-
.../action/IcebergSortCompactionBenchmark.java | 8 +-
.../java/org/apache/iceberg/spark/BaseCatalog.java | 7 +
.../org/apache/iceberg/spark/SparkCatalog.java | 49 +++---
.../org/apache/iceberg/spark/SparkSchemaUtil.java | 35 ++++
.../apache/iceberg/spark/SparkSessionCatalog.java | 43 +++--
.../org/apache/iceberg/spark/SparkTypeToType.java | 80 ++++++++-
.../org/apache/iceberg/spark/SparkV2Filters.java | 8 +-
.../actions/BaseTableCreationSparkAction.java | 10 +-
.../iceberg/spark/TestSparkSessionCatalog.java | 24 ++-
.../actions/TestRemoveOrphanFilesAction3.java | 41 ++++-
.../iceberg/spark/source/TestPathIdentifier.java | 14 +-
.../iceberg/spark/sql/TestSparkDefaultValues.java | 178 +++++++++++++++++++--
13 files changed, 406 insertions(+), 100 deletions(-)
diff --git
a/spark/v4.2/spark/src/jmh/java/org/apache/iceberg/spark/action/IcebergDataCompactionBenchmark.java
b/spark/v4.2/spark/src/jmh/java/org/apache/iceberg/spark/action/IcebergDataCompactionBenchmark.java
index 6199726e18..de00afd376 100644
---
a/spark/v4.2/spark/src/jmh/java/org/apache/iceberg/spark/action/IcebergDataCompactionBenchmark.java
+++
b/spark/v4.2/spark/src/jmh/java/org/apache/iceberg/spark/action/IcebergDataCompactionBenchmark.java
@@ -24,7 +24,6 @@ import static org.apache.spark.sql.functions.col;
import static org.apache.spark.sql.functions.concat;
import static org.apache.spark.sql.functions.lit;
-import java.util.Collections;
import org.apache.iceberg.Schema;
import org.apache.iceberg.actions.SizeBasedFileRewritePlanner;
import org.apache.iceberg.spark.Spark3Util;
@@ -35,7 +34,7 @@ import org.apache.iceberg.types.Types;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.connector.catalog.Identifier;
-import org.apache.spark.sql.connector.expressions.Transform;
+import org.apache.spark.sql.connector.catalog.TableInfo;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.Measurement;
import org.openjdk.jmh.annotations.Param;
@@ -92,8 +91,10 @@ public class IcebergDataCompactionBenchmark extends
IcebergCompactionBenchmark {
(SparkSessionCatalog<?>)
Spark3Util.catalogAndIdentifier(spark(),
"spark_catalog").catalog();
catalog.dropTable(IDENT);
- catalog.createTable(
- IDENT, SparkSchemaUtil.convert(schema), new Transform[0],
Collections.emptyMap());
+
+ TableInfo tableInfo =
+ new
TableInfo.Builder().withSchema(SparkSchemaUtil.convert(schema)).build();
+ catalog.createTable(IDENT, tableInfo);
} catch (Exception e) {
throw new RuntimeException(e);
}
diff --git
a/spark/v4.2/spark/src/jmh/java/org/apache/iceberg/spark/action/IcebergSortCompactionBenchmark.java
b/spark/v4.2/spark/src/jmh/java/org/apache/iceberg/spark/action/IcebergSortCompactionBenchmark.java
index 8900ae8136..d4d21dde93 100644
---
a/spark/v4.2/spark/src/jmh/java/org/apache/iceberg/spark/action/IcebergSortCompactionBenchmark.java
+++
b/spark/v4.2/spark/src/jmh/java/org/apache/iceberg/spark/action/IcebergSortCompactionBenchmark.java
@@ -25,7 +25,6 @@ import static org.apache.spark.sql.functions.current_date;
import static org.apache.spark.sql.functions.date_add;
import static org.apache.spark.sql.functions.expr;
-import java.util.Collections;
import java.util.concurrent.TimeUnit;
import org.apache.iceberg.NullOrder;
import org.apache.iceberg.Schema;
@@ -40,7 +39,7 @@ import org.apache.iceberg.types.Types;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.connector.catalog.Identifier;
-import org.apache.spark.sql.connector.expressions.Transform;
+import org.apache.spark.sql.connector.catalog.TableInfo;
import org.apache.spark.sql.types.DataTypes;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
@@ -268,8 +267,9 @@ public class IcebergSortCompactionBenchmark extends
IcebergCompactionBenchmark {
(SparkSessionCatalog<?>)
Spark3Util.catalogAndIdentifier(spark(),
"spark_catalog").catalog();
catalog.dropTable(IDENT);
- catalog.createTable(
- IDENT, SparkSchemaUtil.convert(schema), new Transform[0],
Collections.emptyMap());
+ TableInfo tableInfo =
+ new
TableInfo.Builder().withSchema(SparkSchemaUtil.convert(schema)).build();
+ catalog.createTable(IDENT, tableInfo);
} catch (Exception e) {
throw new RuntimeException(e);
}
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/BaseCatalog.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/BaseCatalog.java
index 7435fa9874..c2a43306ce 100644
--- a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/BaseCatalog.java
+++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/BaseCatalog.java
@@ -18,6 +18,7 @@
*/
package org.apache.iceberg.spark;
+import java.util.Set;
import org.apache.iceberg.spark.procedures.SparkProcedures;
import org.apache.iceberg.spark.procedures.SparkProcedures.ProcedureBuilder;
import org.apache.iceberg.spark.source.HasIcebergCatalog;
@@ -30,6 +31,7 @@ import org.apache.spark.sql.connector.catalog.Relation;
import org.apache.spark.sql.connector.catalog.RelationCatalog;
import org.apache.spark.sql.connector.catalog.StagingTableCatalog;
import org.apache.spark.sql.connector.catalog.SupportsNamespaces;
+import org.apache.spark.sql.connector.catalog.TableCatalogCapability;
import org.apache.spark.sql.connector.catalog.procedures.UnboundProcedure;
import org.apache.spark.sql.util.CaseInsensitiveStringMap;
@@ -118,4 +120,9 @@ abstract class BaseCatalog
private static boolean isSystemNamespace(String[] namespace) {
return namespace.length == 1 && namespace[0].equalsIgnoreCase("system");
}
+
+ @Override
+ public Set<TableCatalogCapability> capabilities() {
+ return Set.of(TableCatalogCapability.SUPPORT_COLUMN_DEFAULT_VALUE);
+ }
}
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 67c75c3a63..9e2e7138e0 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
@@ -82,10 +82,9 @@ import org.apache.spark.sql.connector.catalog.TableChange;
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;
-import org.apache.spark.sql.types.StructType;
import org.apache.spark.sql.util.CaseInsensitiveStringMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -202,17 +201,16 @@ public class SparkCatalog extends BaseCatalog {
}
@Override
- public Table createTable(
- Identifier ident, StructType schema, Transform[] transforms, Map<String,
String> properties)
- throws TableAlreadyExistsException {
- Schema icebergSchema = SparkSchemaUtil.convert(schema);
+ public Table createTable(Identifier ident, TableInfo tableInfo)
+ throws TableAlreadyExistsException, NoSuchNamespaceException {
+ Schema icebergSchema = SparkSchemaUtil.convertWithDefaults(tableInfo);
try {
Catalog.TableBuilder builder = newBuilder(ident, icebergSchema);
org.apache.iceberg.Table icebergTable =
builder
- .withPartitionSpec(Spark3Util.toPartitionSpec(icebergSchema,
transforms))
- .withLocation(properties.get("location"))
- .withProperties(Spark3Util.rebuildCreateProperties(properties))
+ .withPartitionSpec(Spark3Util.toPartitionSpec(icebergSchema,
tableInfo.partitions()))
+ .withLocation(tableInfo.properties().get("location"))
+
.withProperties(Spark3Util.rebuildCreateProperties(tableInfo.properties()))
.create();
return new SparkTable(icebergTable);
} catch (AlreadyExistsException e) {
@@ -221,17 +219,16 @@ public class SparkCatalog extends BaseCatalog {
}
@Override
- public StagedTable stageCreate(
- Identifier ident, StructType schema, Transform[] transforms, Map<String,
String> properties)
+ public StagedTable stageCreate(Identifier ident, TableInfo tableInfo)
throws TableAlreadyExistsException {
- Schema icebergSchema = SparkSchemaUtil.convert(schema);
+ Schema icebergSchema = SparkSchemaUtil.convert(tableInfo);
try {
Catalog.TableBuilder builder = newBuilder(ident, icebergSchema);
Transaction transaction =
builder
- .withPartitionSpec(Spark3Util.toPartitionSpec(icebergSchema,
transforms))
- .withLocation(properties.get("location"))
- .withProperties(Spark3Util.rebuildCreateProperties(properties))
+ .withPartitionSpec(Spark3Util.toPartitionSpec(icebergSchema,
tableInfo.partitions()))
+ .withLocation(tableInfo.properties().get("location"))
+
.withProperties(Spark3Util.rebuildCreateProperties(tableInfo.properties()))
.createTransaction();
return new StagedSparkTable(transaction);
} catch (AlreadyExistsException e) {
@@ -240,17 +237,16 @@ public class SparkCatalog extends BaseCatalog {
}
@Override
- public StagedTable stageReplace(
- Identifier ident, StructType schema, Transform[] transforms, Map<String,
String> properties)
+ public StagedTable stageReplace(Identifier ident, TableInfo tableInfo)
throws NoSuchTableException {
- Schema icebergSchema = SparkSchemaUtil.convert(schema);
+ Schema icebergSchema = SparkSchemaUtil.convert(tableInfo);
try {
Catalog.TableBuilder builder = newBuilder(ident, icebergSchema);
Transaction transaction =
builder
- .withPartitionSpec(Spark3Util.toPartitionSpec(icebergSchema,
transforms))
- .withLocation(properties.get("location"))
- .withProperties(Spark3Util.rebuildCreateProperties(properties))
+ .withPartitionSpec(Spark3Util.toPartitionSpec(icebergSchema,
tableInfo.partitions()))
+ .withLocation(tableInfo.properties().get("location"))
+
.withProperties(Spark3Util.rebuildCreateProperties(tableInfo.properties()))
.replaceTransaction();
return new StagedSparkTable(transaction);
} catch (org.apache.iceberg.exceptions.NoSuchTableException e) {
@@ -259,15 +255,14 @@ public class SparkCatalog extends BaseCatalog {
}
@Override
- public StagedTable stageCreateOrReplace(
- Identifier ident, StructType schema, Transform[] transforms, Map<String,
String> properties) {
- Schema icebergSchema = SparkSchemaUtil.convert(schema);
+ public StagedTable stageCreateOrReplace(Identifier ident, TableInfo
tableInfo) {
+ Schema icebergSchema = SparkSchemaUtil.convert(tableInfo);
Catalog.TableBuilder builder = newBuilder(ident, icebergSchema);
Transaction transaction =
builder
- .withPartitionSpec(Spark3Util.toPartitionSpec(icebergSchema,
transforms))
- .withLocation(properties.get("location"))
- .withProperties(Spark3Util.rebuildCreateProperties(properties))
+ .withPartitionSpec(Spark3Util.toPartitionSpec(icebergSchema,
tableInfo.partitions()))
+ .withLocation(tableInfo.properties().get("location"))
+
.withProperties(Spark3Util.rebuildCreateProperties(tableInfo.properties()))
.createOrReplaceTransaction();
return new StagedSparkTable(transaction);
}
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkSchemaUtil.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkSchemaUtil.java
index 9f299cb276..85583c8711 100644
---
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkSchemaUtil.java
+++
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkSchemaUtil.java
@@ -41,6 +41,7 @@ import org.apache.iceberg.types.Types;
import org.apache.spark.sql.AnalysisException;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.catalog.Column;
+import org.apache.spark.sql.connector.catalog.TableInfo;
import org.apache.spark.sql.types.DataType;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
@@ -108,6 +109,40 @@ public class SparkSchemaUtil {
return TypeUtil.visit(type, new TypeToSparkType());
}
+ /**
+ * Convert a Spark {@link TableInfo tableInfo} to a {@link Schema} with new
field ids.
+ *
+ * <p>This conversion assigns fresh ids.
+ *
+ * <p>Default values are converted from spark default values.
+ *
+ * @param tableInfo a Spark TableInfo
+ * @return the equivalent Schema
+ * @throws IllegalArgumentException if the type cannot be converted
+ */
+ public static Schema convertWithDefaults(TableInfo tableInfo) {
+ Type converted =
+ SparkTypeVisitor.visit(tableInfo.schema(), new
SparkTypeToType(tableInfo, true));
+ return new Schema(converted.asNestedType().asStructType().fields());
+ }
+
+ /**
+ * Convert a Spark {@link TableInfo tableInfo} to a {@link Schema} with new
field ids.
+ *
+ * <p>This conversion assigns fresh ids.
+ *
+ * <p>Default values are ignored.
+ *
+ * @param tableInfo a Spark TableInfo
+ * @return the equivalent Schema
+ * @throws IllegalArgumentException if the type cannot be converted
+ */
+ public static Schema convert(TableInfo tableInfo) {
+ Type converted =
+ SparkTypeVisitor.visit(tableInfo.schema(), new
SparkTypeToType(tableInfo, false));
+ return new Schema(converted.asNestedType().asStructType().fields());
+ }
+
/**
* Convert a Spark {@link StructType struct} to a {@link Schema} with new
field ids.
*
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 d754f84b27..d8a42eec97 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
@@ -52,12 +52,11 @@ import
org.apache.spark.sql.connector.catalog.SupportsNamespaces;
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;
import org.apache.spark.sql.connector.catalog.functions.UnboundFunction;
-import org.apache.spark.sql.connector.expressions.Transform;
-import org.apache.spark.sql.types.StructType;
import org.apache.spark.sql.util.CaseInsensitiveStringMap;
/**
@@ -238,31 +237,29 @@ public class SparkSessionCatalog<
}
@Override
- public Table createTable(
- Identifier ident, StructType schema, Transform[] partitions, Map<String,
String> properties)
+ public Table createTable(Identifier ident, TableInfo info)
throws TableAlreadyExistsException, NoSuchNamespaceException {
checkViewNotExists(ident);
- String provider = properties.get("provider");
+ String provider = info.properties().get("provider");
if (useIceberg(provider)) {
- return icebergCatalog.createTable(ident, schema, partitions, properties);
+ return icebergCatalog.createTable(ident, info);
} else {
// delegate to the session catalog
- return getSessionCatalog().createTable(ident, schema, partitions,
properties);
+ return getSessionCatalog().createTable(ident, info);
}
}
@Override
- public StagedTable stageCreate(
- Identifier ident, StructType schema, Transform[] partitions, Map<String,
String> properties)
+ public StagedTable stageCreate(Identifier ident, TableInfo info)
throws TableAlreadyExistsException, NoSuchNamespaceException {
checkViewNotExists(ident);
- String provider = properties.get("provider");
+ String provider = info.properties().get("provider");
TableCatalog catalog;
if (useIceberg(provider)) {
if (asStagingCatalog != null) {
- return asStagingCatalog.stageCreate(ident, schema, partitions,
properties);
+ return asStagingCatalog.stageCreate(ident, info);
}
catalog = icebergCatalog;
} else {
@@ -271,19 +268,18 @@ public class SparkSessionCatalog<
// create the table with the session catalog, then wrap it in a staged
table that will delete to
// roll back
- Table table = catalog.createTable(ident, schema, partitions, properties);
+ Table table = catalog.createTable(ident, info);
return new RollbackStagedTable(catalog, ident, table);
}
@Override
- public StagedTable stageReplace(
- Identifier ident, StructType schema, Transform[] partitions, Map<String,
String> properties)
+ public StagedTable stageReplace(Identifier ident, TableInfo info)
throws NoSuchNamespaceException, NoSuchTableException {
- String provider = properties.get("provider");
+ String provider = info.properties().get("provider");
TableCatalog catalog;
if (useIceberg(provider)) {
if (asStagingCatalog != null) {
- return asStagingCatalog.stageReplace(ident, schema, partitions,
properties);
+ return asStagingCatalog.stageReplace(ident, info);
}
catalog = icebergCatalog;
} else {
@@ -298,18 +294,17 @@ public class SparkSessionCatalog<
try {
// create the table with the session catalog, then wrap it in a staged
table that will delete
// to roll back
- Table table = catalog.createTable(ident, schema, partitions, properties);
+ Table table = catalog.createTable(ident, info);
return new RollbackStagedTable(catalog, ident, table);
} catch (TableAlreadyExistsException e) {
// the table was deleted, but now already exists again. retry the
replace.
- return stageReplace(ident, schema, partitions, properties);
+ return stageReplace(ident, info);
}
}
@Override
- public StagedTable stageCreateOrReplace(
- Identifier ident, StructType schema, Transform[] partitions, Map<String,
String> properties)
+ public StagedTable stageCreateOrReplace(Identifier ident, TableInfo info)
throws NoSuchNamespaceException {
if (viewExists(ident)) {
// StagingTableCatalog does not declare a checked collision exception
for create-or-replace.
@@ -317,11 +312,11 @@ public class SparkSessionCatalog<
"Cannot create or replace table %s: a view with the same name
already exists", ident);
}
- String provider = properties.get("provider");
+ String provider = info.properties().get("provider");
TableCatalog catalog;
if (useIceberg(provider)) {
if (asStagingCatalog != null) {
- return asStagingCatalog.stageCreateOrReplace(ident, schema,
partitions, properties);
+ return asStagingCatalog.stageCreateOrReplace(ident, info);
}
catalog = icebergCatalog;
} else {
@@ -334,12 +329,12 @@ public class SparkSessionCatalog<
try {
// create the table with the session catalog, then wrap it in a staged
table that will delete
// to roll back
- Table sessionCatalogTable = catalog.createTable(ident, schema,
partitions, properties);
+ Table sessionCatalogTable = catalog.createTable(ident, info);
return new RollbackStagedTable(catalog, ident, sessionCatalogTable);
} catch (TableAlreadyExistsException e) {
// the table was deleted, but now already exists again. retry the
replace.
- return stageCreateOrReplace(ident, schema, partitions, properties);
+ return stageCreateOrReplace(ident, info);
}
}
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java
index 6ce085a1b7..e658ebe1c0 100644
---
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java
+++
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java
@@ -20,10 +20,19 @@ package org.apache.iceberg.spark;
import java.util.List;
import java.util.Locale;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.types.EdgeAlgorithm;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
+import org.apache.spark.sql.connector.catalog.Column;
+import org.apache.spark.sql.connector.catalog.ColumnDefaultValue;
+import org.apache.spark.sql.connector.catalog.TableInfo;
+import org.apache.spark.sql.connector.expressions.Literal;
import org.apache.spark.sql.types.ArrayType;
import org.apache.spark.sql.types.BinaryType;
import org.apache.spark.sql.types.BooleanType;
@@ -52,6 +61,8 @@ import org.apache.spark.sql.types.VariantType;
class SparkTypeToType extends SparkTypeVisitor<Type> {
private final StructType root;
+ private boolean includeDefaults;
+ private Map<String, Column> nameToColumnMap = Map.of();
private int nextId = 0;
SparkTypeToType() {
@@ -64,6 +75,14 @@ class SparkTypeToType extends SparkTypeVisitor<Type> {
this.nextId = root.fields().length;
}
+ SparkTypeToType(TableInfo tableInfo, boolean includeDefaults) {
+ this(tableInfo.schema());
+
+ this.includeDefaults = includeDefaults;
+ this.nameToColumnMap =
+ Stream.of(tableInfo.columns()).collect(Collectors.toMap(Column::name,
Function.identity()));
+ }
+
private int getNextId() {
int next = nextId;
nextId += 1;
@@ -71,11 +90,10 @@ class SparkTypeToType extends SparkTypeVisitor<Type> {
}
@Override
- @SuppressWarnings("ReferenceEquality")
public Type struct(StructType struct, List<Type> types) {
StructField[] fields = struct.fields();
List<Types.NestedField> newFields =
Lists.newArrayListWithExpectedSize(fields.length);
- boolean isRoot = root == struct;
+ boolean isRoot = root != null && root.equals(struct);
for (int i = 0; i < fields.length; i += 1) {
StructField field = fields[i];
Type type = types.get(i);
@@ -88,13 +106,23 @@ class SparkTypeToType extends SparkTypeVisitor<Type> {
id = getNextId();
}
- String doc = field.getComment().isDefined() ? field.getComment().get() :
null;
+ Types.NestedField.Builder fieldBuilder =
+ Types.NestedField.builder()
+ .isOptional(field.nullable())
+ .withId(id)
+ .withName(field.name())
+ .ofType(type);
- if (field.nullable()) {
- newFields.add(Types.NestedField.optional(id, field.name(), type, doc));
- } else {
- newFields.add(Types.NestedField.required(id, field.name(), type, doc));
+ if (field.getComment().isDefined()) {
+ fieldBuilder.withDoc(field.getComment().get());
+ }
+
+ // spark only supports defaults at the top-level
+ if (isRoot && includeDefaults) {
+ convertDefaultValue(fieldBuilder, field);
}
+
+ newFields.add(fieldBuilder.build());
}
return Types.StructType.of(newFields);
@@ -201,4 +229,42 @@ class SparkTypeToType extends SparkTypeVisitor<Type> {
"Iceberg does not support Spark geography edge algorithm: " +
algorithm);
}
}
+
+ private void convertDefaultValue(Types.NestedField.Builder icebergField,
StructField sparkField) {
+ Column column = nameToColumnMap.get(sparkField.name());
+
+ if (column == null) {
+ return;
+ }
+
+ ColumnDefaultValue columnDefaultValue = column.defaultValue();
+
+ if (columnDefaultValue == null) {
+ return;
+ }
+
+ if (columnDefaultValue.getExpression() == null &&
columnDefaultValue.getSql() != null) {
+ throw new UnsupportedOperationException(
+ "Unsupported default value expression: " +
columnDefaultValue.getSql());
+ }
+
+ if (columnDefaultValue.getExpression() != null
+ && !SparkV2Filters.isLiteral(columnDefaultValue.getExpression())) {
+ throw new UnsupportedOperationException(
+ "Default value expressions are not supported in Iceberg");
+ }
+
+ // ColumnDefaultValue.getValue is equivalent to initial-default in Iceberg
+ Literal<?> initialDefault = columnDefaultValue.getValue();
+ if (initialDefault != null && initialDefault.value() != null) {
+ icebergField.withInitialDefault(
+ Expressions.lit(SparkV2Filters.convertLiteral(initialDefault)));
+ }
+
+ // ColumnDefaultValue.getExpression() is equivalent to write-default in
Iceberg
+ Literal<?> writeDefault = (Literal<?>) columnDefaultValue.getExpression();
+ if (writeDefault != null && writeDefault.value() != null) {
+
icebergField.withWriteDefault(Expressions.lit(SparkV2Filters.convertLiteral(writeDefault)));
+ }
+ }
}
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkV2Filters.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkV2Filters.java
index 57b9d61e38..739d4e1caf 100644
---
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkV2Filters.java
+++
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkV2Filters.java
@@ -372,15 +372,19 @@ public class SparkV2Filters {
return false;
}
- private static boolean
isLiteral(org.apache.spark.sql.connector.expressions.Expression expr) {
+ static boolean
isLiteral(org.apache.spark.sql.connector.expressions.Expression expr) {
return expr instanceof Literal;
}
- private static Object convertLiteral(Literal<?> literal) {
+ static Object convertLiteral(Literal<?> literal) {
if (literal.value() instanceof UTF8String) {
return ((UTF8String) literal.value()).toString();
} else if (literal.value() instanceof Decimal) {
return ((Decimal) literal.value()).toJavaBigDecimal();
+ } else if (literal.value() instanceof Short) {
+ return ((Short) literal.value()).intValue();
+ } else if (literal.value() instanceof Byte) {
+ return ((Byte) literal.value()).intValue();
}
return literal.value();
}
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseTableCreationSparkAction.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseTableCreationSparkAction.java
index 37e1ec4ce7..574bed18de 100644
---
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseTableCreationSparkAction.java
+++
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseTableCreationSparkAction.java
@@ -45,6 +45,7 @@ import org.apache.spark.sql.connector.catalog.CatalogPlugin;
import org.apache.spark.sql.connector.catalog.Identifier;
import org.apache.spark.sql.connector.catalog.StagingTableCatalog;
import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.connector.catalog.TableInfo;
import org.apache.spark.sql.connector.catalog.V1Table;
import org.apache.spark.sql.connector.expressions.Transform;
import org.apache.spark.sql.types.StructType;
@@ -158,8 +159,13 @@ abstract class BaseTableCreationSparkAction<ThisT> extends
BaseSparkAction<ThisT
Map<String, String> props = destTableProps();
StructType schema = sourceTable.schema();
Transform[] partitioning = sourceTable.partitioning();
- return (StagedSparkTable)
- destCatalog().stageCreate(destTableIdent(), schema, partitioning,
props);
+ TableInfo tableInfo =
+ new TableInfo.Builder()
+ .withSchema(schema)
+ .withPartitions(partitioning)
+ .withProperties(props)
+ .build();
+ return (StagedSparkTable) destCatalog().stageCreate(destTableIdent(),
tableInfo);
} catch (org.apache.spark.sql.catalyst.analysis.NoSuchNamespaceException
e) {
throw new NoSuchNamespaceException(
"Cannot create table %s as the namespace does not exist",
destTableIdent());
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 46897c2f9a..4438e22cf5 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
@@ -43,6 +43,7 @@ import
org.apache.spark.sql.connector.catalog.SupportsDeleteV2;
import org.apache.spark.sql.connector.catalog.SupportsNamespaces;
import org.apache.spark.sql.connector.catalog.Table;
import org.apache.spark.sql.connector.catalog.TableCatalog;
+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;
@@ -390,17 +391,26 @@ public class TestSparkSessionCatalog extends TestBase {
SparkSessionCatalog<?> catalog = catalogWithViews(icebergCatalog,
sessionCatalog);
StructType schema = new StructType();
Transform[] partitions = new Transform[0];
-
- assertThatThrownBy(() -> catalog.createTable(ident, schema, partitions,
Collections.emptyMap()))
+ TableInfo tableInfo =
+ new TableInfo.Builder()
+ .withSchema(schema)
+ .withPartitions(partitions)
+ .withProperties(Collections.emptyMap())
+ .build();
+ TableInfo parquetTableInfo =
+ new TableInfo.Builder()
+ .withSchema(schema)
+ .withPartitions(partitions)
+ .withProperties(Collections.singletonMap("provider", "parquet"))
+ .build();
+
+ assertThatThrownBy(() -> catalog.createTable(ident, tableInfo))
.isInstanceOf(TableAlreadyExistsException.class)
.hasMessageContaining(ident.name());
- assertThatThrownBy(() -> catalog.stageCreate(ident, schema, partitions,
Collections.emptyMap()))
+ assertThatThrownBy(() -> catalog.stageCreate(ident, tableInfo))
.isInstanceOf(TableAlreadyExistsException.class)
.hasMessageContaining(ident.name());
- assertThatThrownBy(
- () ->
- catalog.stageCreateOrReplace(
- ident, schema, partitions,
Collections.singletonMap("provider", "parquet")))
+ assertThatThrownBy(() -> catalog.stageCreateOrReplace(ident,
parquetTableInfo))
.isInstanceOf(AlreadyExistsException.class)
.hasMessage(
"Cannot create or replace table %s: a view with the same name
already exists", ident);
diff --git
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveOrphanFilesAction3.java
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveOrphanFilesAction3.java
index 88ac800b15..4a46ca1a00 100644
---
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveOrphanFilesAction3.java
+++
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveOrphanFilesAction3.java
@@ -28,6 +28,7 @@ import org.apache.iceberg.spark.SparkSchemaUtil;
import org.apache.iceberg.spark.SparkSessionCatalog;
import org.apache.iceberg.spark.source.SparkTable;
import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.TableInfo;
import org.apache.spark.sql.connector.expressions.Transform;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.TestTemplate;
@@ -43,7 +44,13 @@ public class TestRemoveOrphanFilesAction3 extends
TestRemoveOrphanFilesAction {
String[] database = {"default"};
Identifier id = Identifier.of(database, randomName("table"));
Transform[] transforms = {};
- cat.createTable(id, SparkSchemaUtil.convert(SCHEMA), transforms,
properties);
+ TableInfo tableInfo =
+ new TableInfo.Builder()
+ .withSchema(SparkSchemaUtil.convert(SCHEMA))
+ .withPartitions(transforms)
+ .withProperties(properties)
+ .build();
+ cat.createTable(id, tableInfo);
SparkTable table = (SparkTable) cat.loadTable(id);
sql("INSERT INTO mycat.default.%s VALUES (1,1,1)", id.name());
@@ -73,7 +80,13 @@ public class TestRemoveOrphanFilesAction3 extends
TestRemoveOrphanFilesAction {
String[] database = {"default"};
Identifier id = Identifier.of(database, randomName("table"));
Transform[] transforms = {};
- cat.createTable(id, SparkSchemaUtil.convert(SCHEMA), transforms,
properties);
+ TableInfo tableInfo =
+ new TableInfo.Builder()
+ .withSchema(SparkSchemaUtil.convert(SCHEMA))
+ .withPartitions(transforms)
+ .withProperties(properties)
+ .build();
+ cat.createTable(id, tableInfo);
SparkTable table = (SparkTable) cat.loadTable(id);
sql("INSERT INTO hadoop.default.%s VALUES (1,1,1)", id.name());
@@ -103,7 +116,13 @@ public class TestRemoveOrphanFilesAction3 extends
TestRemoveOrphanFilesAction {
String[] database = {"default"};
Identifier id = Identifier.of(database, randomName("table"));
Transform[] transforms = {};
- cat.createTable(id, SparkSchemaUtil.convert(SCHEMA), transforms,
properties);
+ TableInfo tableInfo =
+ new TableInfo.Builder()
+ .withSchema(SparkSchemaUtil.convert(SCHEMA))
+ .withPartitions(transforms)
+ .withProperties(properties)
+ .build();
+ cat.createTable(id, tableInfo);
SparkTable table = (SparkTable) cat.loadTable(id);
sql("INSERT INTO hive.default.%s VALUES (1,1,1)", id.name());
@@ -137,7 +156,13 @@ public class TestRemoveOrphanFilesAction3 extends
TestRemoveOrphanFilesAction {
String[] database = {"default"};
Identifier id = Identifier.of(database, randomName("table"));
Transform[] transforms = {};
- cat.createTable(id, SparkSchemaUtil.convert(SCHEMA), transforms,
properties);
+ TableInfo tableInfo =
+ new TableInfo.Builder()
+ .withSchema(SparkSchemaUtil.convert(SCHEMA))
+ .withPartitions(transforms)
+ .withProperties(properties)
+ .build();
+ cat.createTable(id, tableInfo);
SparkTable table = (SparkTable) cat.loadTable(id);
sql("INSERT INTO default.%s VALUES (1,1,1)", id.name());
@@ -170,7 +195,13 @@ public class TestRemoveOrphanFilesAction3 extends
TestRemoveOrphanFilesAction {
Identifier id = Identifier.of(database, "sessioncattest");
Transform[] transforms = {};
cat.dropTable(id);
- cat.createTable(id, SparkSchemaUtil.convert(SCHEMA), transforms,
properties);
+ TableInfo tableInfo =
+ new TableInfo.Builder()
+ .withSchema(SparkSchemaUtil.convert(SCHEMA))
+ .withPartitions(transforms)
+ .withProperties(properties)
+ .build();
+ cat.createTable(id, tableInfo);
SparkTable table = (SparkTable) cat.loadTable(id);
sql("INSERT INTO default.sessioncattest VALUES (1,1,1)");
diff --git
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestPathIdentifier.java
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestPathIdentifier.java
index aa8fe047b3..b244f71b8a 100644
---
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestPathIdentifier.java
+++
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestPathIdentifier.java
@@ -33,9 +33,10 @@ import org.apache.iceberg.spark.SparkCatalog;
import org.apache.iceberg.spark.SparkSchemaUtil;
import org.apache.iceberg.spark.TestBase;
import org.apache.iceberg.types.Types;
+import org.apache.spark.sql.catalyst.analysis.NoSuchNamespaceException;
import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException;
-import org.apache.spark.sql.connector.expressions.Transform;
+import org.apache.spark.sql.connector.catalog.TableInfo;
import org.apache.spark.sql.util.CaseInsensitiveStringMap;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
@@ -67,11 +68,12 @@ public class TestPathIdentifier extends TestBase {
}
@Test
- public void testPathIdentifier() throws TableAlreadyExistsException,
NoSuchTableException {
- SparkTable table =
- (SparkTable)
- sparkCatalog.createTable(
- identifier, SparkSchemaUtil.convert(SCHEMA), new Transform[0],
ImmutableMap.of());
+ public void testPathIdentifier()
+ throws TableAlreadyExistsException, NoSuchTableException,
NoSuchNamespaceException {
+ TableInfo tableInfo =
+ new
TableInfo.Builder().withSchema(SparkSchemaUtil.convert(SCHEMA)).build();
+
+ SparkTable table = (SparkTable) sparkCatalog.createTable(identifier,
tableInfo);
assertThat(tableLocation.getAbsolutePath()).isEqualTo(table.table().location());
assertThat(table.table()).isInstanceOf(BaseTable.class);
diff --git
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkDefaultValues.java
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkDefaultValues.java
index 838e735c75..fa6c1ec03c 100644
---
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkDefaultValues.java
+++
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkDefaultValues.java
@@ -21,29 +21,30 @@ package org.apache.iceberg.spark.sql;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import java.nio.ByteBuffer;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.util.HexFormat;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.expressions.Literal;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.spark.CatalogTestBase;
import org.apache.iceberg.types.Types;
-import org.apache.spark.sql.AnalysisException;
+import org.apache.iceberg.util.DateTimeUtil;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.TestTemplate;
-/**
- * Tests for Spark SQL Default values integration with Iceberg default values.
- *
- * <p>Note: These tests use {@code validationCatalog.createTable()} to create
tables with default
- * values because the Iceberg Spark integration does not yet support default
value clauses in Spark
- * DDL.
- */
+/** Tests for Spark SQL Default values integration with Iceberg default
values. */
public class TestSparkDefaultValues extends CatalogTestBase {
@AfterEach
public void dropTestTable() {
sql("DROP TABLE IF EXISTS %s", tableName);
+ sql("DROP TABLE IF EXISTS %s", tableName("dest"));
}
@TestTemplate
@@ -141,7 +142,104 @@ public class TestSparkDefaultValues extends
CatalogTestBase {
}
@TestTemplate
- public void testCreateTableWithDefaultsUnsupported() {
+ public void testCreateTableWithDefaults() {
+ assertThat(validationCatalog.tableExists(tableIdent))
+ .as("Table should not already exist")
+ .isFalse();
+
+ sql(
+ "CREATE TABLE %s (id INT DEFAULT -1, data STRING DEFAULT '-') USING
iceberg "
+ + "TBLPROPERTIES ('format-version'='3')",
+ tableName);
+
+ Types.NestedField idField =
validationCatalog.loadTable(tableIdent).schema().findField("id");
+ assertThat(idField.initialDefault()).isEqualTo(-1);
+ assertThat(idField.writeDefault()).isEqualTo(-1);
+
+ Types.NestedField dataField =
+ validationCatalog.loadTable(tableIdent).schema().findField("data");
+ assertThat(dataField.initialDefault()).isEqualTo("-");
+ assertThat(dataField.writeDefault()).isEqualTo("-");
+ }
+
+ @TestTemplate
+ public void testCreateTableWithBoxedDefaults() {
+ assertThat(validationCatalog.tableExists(tableIdent))
+ .as("Table should not already exist")
+ .isFalse();
+
+ sql(
+ "CREATE TABLE %s (ti TINYINT DEFAULT -1, s SHORT DEFAULT '-2') USING
iceberg "
+ + "TBLPROPERTIES ('format-version'='3')",
+ tableName);
+
+ Types.NestedField tinyField =
validationCatalog.loadTable(tableIdent).schema().findField("ti");
+ assertThat(tinyField.initialDefault()).isEqualTo(-1);
+ assertThat(tinyField.writeDefault()).isEqualTo(-1);
+
+ Types.NestedField shortField =
validationCatalog.loadTable(tableIdent).schema().findField("s");
+ assertThat(shortField.initialDefault()).isEqualTo(-2);
+ assertThat(shortField.writeDefault()).isEqualTo(-2);
+ }
+
+ @TestTemplate
+ public void testCreateTableWithBinaryDefaults() {
+ assertThat(validationCatalog.tableExists(tableIdent))
+ .as("Table should not already exist")
+ .isFalse();
+
+ sql(
+ "CREATE TABLE %s (b BINARY DEFAULT X'1ABF') USING iceberg "
+ + "TBLPROPERTIES ('format-version'='3')",
+ tableName);
+
+ ByteBuffer expected = ByteBuffer.wrap(HexFormat.of().parseHex("1ABF"));
+
+ Types.NestedField binaryField =
validationCatalog.loadTable(tableIdent).schema().findField("b");
+ assertThat(binaryField.initialDefault()).isEqualTo(expected);
+ assertThat(binaryField.writeDefault()).isEqualTo(expected);
+ }
+
+ @TestTemplate
+ public void testCreateTableWithTimestampDefaults() {
+ assertThat(validationCatalog.tableExists(tableIdent))
+ .as("Table should not already exist")
+ .isFalse();
+
+ sql(
+ "CREATE TABLE %s (ts TIMESTAMP DEFAULT timestamp '2026-09-24
21:26:02.98269') USING iceberg "
+ + "TBLPROPERTIES ('format-version'='3')",
+ tableName);
+
+ long expectedMicros =
+ DateTimeUtil.microsFromInstant(
+ LocalDateTime.parse("2026-09-24T21:26:02.98269")
+
.atZone(ZoneId.of(spark.sessionState().conf().sessionLocalTimeZone()))
+ .toInstant());
+
+ Types.NestedField tsField =
validationCatalog.loadTable(tableIdent).schema().findField("ts");
+ assertThat(tsField.initialDefault()).isEqualTo(expectedMicros);
+ assertThat(tsField.writeDefault()).isEqualTo(expectedMicros);
+ }
+
+ @TestTemplate
+ public void testCreateTableWithNullDefaults() {
+ assertThat(validationCatalog.tableExists(tableIdent))
+ .as("Table should not already exist")
+ .isFalse();
+
+ sql(
+ "CREATE TABLE %s (s String DEFAULT null) USING iceberg "
+ + "TBLPROPERTIES ('format-version'='3')",
+ tableName);
+
+ Types.NestedField stringField =
validationCatalog.loadTable(tableIdent).schema().findField("s");
+ assertThat(stringField.initialDefault()).isNull();
+ assertThat(stringField.writeDefault()).isNull();
+ }
+
+ @TestTemplate
+ public void testCreateTableWithDefaultsExpressionsUnsupported() {
assertThat(validationCatalog.tableExists(tableIdent))
.as("Table should not already exist")
.isFalse();
@@ -149,10 +247,10 @@ public class TestSparkDefaultValues extends
CatalogTestBase {
assertThatThrownBy(
() ->
sql(
- "CREATE TABLE %s (id INT, data STRING DEFAULT
'default-value') USING iceberg",
+ "CREATE TABLE %s (id INT, data STRING DEFAULT
CURRENT_USER) USING iceberg",
tableName))
- .isInstanceOf(AnalysisException.class)
- .hasMessageContaining("does not support column default value");
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("Unsupported default value expression:
CURRENT_USER");
}
@TestTemplate
@@ -172,6 +270,62 @@ public class TestSparkDefaultValues extends
CatalogTestBase {
.hasMessageContaining("default values in Spark is currently
unsupported");
}
+ @TestTemplate
+ public void testCtasExcludesDefault() {
+ assertThat(validationCatalog.tableExists(tableIdent))
+ .as("Table should not already exist")
+ .isFalse();
+
+ sql(
+ "CREATE TABLE %s (data STRING DEFAULT '-') USING iceberg "
+ + "TBLPROPERTIES ('format-version'='3')",
+ tableName);
+
+ sql("INSERT INTO %s VALUES (DEFAULT)", tableName);
+
+ sql("CREATE TABLE %s USING iceberg AS SELECT * FROM %s",
tableName("dest"), tableName);
+
+ TableIdentifier destIdent = TableIdentifier.of(Namespace.of("default"),
"dest");
+ Types.NestedField dataField =
validationCatalog.loadTable(destIdent).schema().findField("data");
+ assertThat(dataField.initialDefault()).isNull();
+ assertThat(dataField.writeDefault()).isNull();
+
+ assertEquals(
+ "Should have correct default values from source table",
+ ImmutableList.of(row("-")),
+ sql("SELECT * FROM %s", tableName("dest")));
+ }
+
+ @TestTemplate
+ public void testRtasExcludesDefault() {
+ assertThat(validationCatalog.tableExists(tableIdent))
+ .as("Table should not already exist")
+ .isFalse();
+
+ sql(
+ "CREATE TABLE %s (data STRING DEFAULT '-') USING iceberg "
+ + "TBLPROPERTIES ('format-version'='3')",
+ tableName);
+
+ sql("INSERT INTO %s VALUES (DEFAULT)", tableName);
+
+ sql(
+ "CREATE TABLE %s (data STRING DEFAULT '-') USING iceberg TBLPROPERTIES
('format-version'='3')",
+ tableName("dest"));
+
+ sql("REPLACE TABLE %s USING iceberg AS SELECT * FROM %s",
tableName("dest"), tableName);
+
+ TableIdentifier destIdent = TableIdentifier.of(Namespace.of("default"),
"dest");
+ Types.NestedField dataField =
validationCatalog.loadTable(destIdent).schema().findField("data");
+ assertThat(dataField.initialDefault()).isNull();
+ assertThat(dataField.writeDefault()).isNull();
+
+ assertEquals(
+ "Should have correct default values from source table",
+ ImmutableList.of(row("-")),
+ sql("SELECT * FROM %s", tableName("dest")));
+ }
+
@TestTemplate
public void testSchemaEvolutionWithDefaultValueChanges() {
assertThat(validationCatalog.tableExists(tableIdent))
