Separate default value parsing from format version check

Author: suhwan-cheon

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/IcebergMetadataApplier.java

@@ -38,6 +38,7 @@
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.HasTableOperations;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.Table;
@@ -179,11 +180,14 @@ private void applyCreateTable(CreateTableEvent event) {
             }
             PartitionSpec partitionSpec = generatePartitionSpec(icebergSchema, partitionColumns);
             if (!catalog.tableExists(tableIdentifier)) {
-                Table table =
-                        catalog.createTable(
-                                tableIdentifier, icebergSchema, partitionSpec, tableOptions);
-
-                applyDefaultValues(table, cdcSchema);
+                boolean hasDefaults =
+                        columns.stream().anyMatch(field -> field.initialDefault() != null);
+                Map<String, String> properties = tableOptions;
+                if (hasDefaults) {
+                    properties = new HashMap<>(tableOptions);
+                    properties.put("format-version", "3");
+                }
+                catalog.createTable(tableIdentifier, icebergSchema, partitionSpec, properties);
 
                 LOG.info(
                         "Spend {} ms to create iceberg table {}",
@@ -195,25 +199,6 @@ private void applyCreateTable(CreateTableEvent event) {
         }
     }
 
-    private void applyDefaultValues(
-            Table table, org.apache.flink.cdc.common.schema.Schema cdcSchema) {
-        UpdateSchema updateSchema = null;
-        for (Column column : cdcSchema.getColumns()) {
-            Literal<?> defaultValue =
-                    IcebergTypeUtils.parseDefaultValue(
-                            column.getDefaultValueExpression(), column.getType());
-            if (defaultValue != null) {
-                if (updateSchema == null) {
-                    updateSchema = table.updateSchema();
-                }
-                updateSchema.updateColumnDefault(column.getName(), defaultValue);
-            }
-        }
-        if (updateSchema != null) {
-            updateSchema.commit();
-        }
-    }
-
     private void applyAddColumn(AddColumnEvent event) {
         TableIdentifier tableIdentifier = TableIdentifier.parse(event.tableId().identifier());
         try {
@@ -229,6 +214,10 @@ private void applyAddColumnEventWithPosition(Table table, AddColumnEvent event)
 
         try {
             UpdateSchema updateSchema = table.updateSchema();
+            int formatVersion =
+                    table instanceof HasTableOperations
+                            ? ((HasTableOperations) table).operations().current().formatVersion()
+                            : 2;
             for (AddColumnEvent.ColumnWithPosition columnWithPosition : event.getAddedColumns()) {
                 Column addColumn = columnWithPosition.getAddColumn();
                 String columnName = addColumn.getName();
@@ -237,12 +226,14 @@ private void applyAddColumnEventWithPosition(Table table, AddColumnEvent event)
                         FlinkSchemaUtil.convert(
                                 DataTypeUtils.toFlinkDataType(addColumn.getType())
                                         .getLogicalType());
-                updateSchema.addColumn(columnName, icebergType, columnComment);
                 Literal<?> defaultValue =
                         IcebergTypeUtils.parseDefaultValue(
                                 addColumn.getDefaultValueExpression(), addColumn.getType());
-                if (defaultValue != null) {
+                if (defaultValue != null && formatVersion >= 3) {
+                    updateSchema.addColumn(columnName, icebergType, columnComment, defaultValue);
                     updateSchema.updateColumnDefault(columnName, defaultValue);
+                } else {
+                    updateSchema.addColumn(columnName, icebergType, columnComment);
                 }
                 switch (columnWithPosition.getPosition()) {
                     case FIRST:

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/utils/IcebergTypeUtils.java

@@ -52,12 +52,22 @@ public class IcebergTypeUtils {
     public static Types.NestedField convertCdcColumnToIcebergField(
             int index, PhysicalColumn column) {
         DataType dataType = column.getType();
-        return Types.NestedField.of(
-                index,
-                dataType.isNullable(),
-                column.getName(),
-                convertCDCTypeToIcebergType(dataType),
-                column.getComment());
+        Literal<?> defaultValue = parseDefaultValue(column.getDefaultValueExpression(), dataType);
+        Types.NestedField.Builder builder =
+                Types.NestedField.builder()
+                        .withId(index)
+                        .withName(column.getName())
+                        .ofType(convertCDCTypeToIcebergType(dataType))
+                        .withDoc(column.getComment());
+        if (dataType.isNullable()) {
+            builder.asOptional();
+        } else {
+            builder.asRequired();
+        }
+        if (defaultValue != null) {
+            builder.withInitialDefault(defaultValue).withWriteDefault(defaultValue);
+        }
+        return builder.build();
     }
 
     /**