fix: Skip default values for Iceberg format v2 tables

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;
@@ -197,6 +198,9 @@ private void applyCreateTable(CreateTableEvent event) {
 
     private void applyDefaultValues(
             Table table, org.apache.flink.cdc.common.schema.Schema cdcSchema) {
+        if (getFormatVersion(table) < 3) {
+            return;
+        }
         UpdateSchema updateSchema = null;
         for (Column column : cdcSchema.getColumns()) {
             Literal<?> defaultValue =
@@ -240,8 +244,9 @@ private void applyAddColumnEventWithPosition(Table table, AddColumnEvent event)
                 Literal<?> defaultValue =
                         IcebergTypeUtils.parseDefaultValue(
                                 addColumn.getDefaultValueExpression(), addColumn.getType());
-                if (defaultValue != null) {
+                if (defaultValue != null && getFormatVersion(table) >= 3) {
                     updateSchema.addColumn(columnName, icebergType, columnComment, defaultValue);
+                    updateSchema.updateColumnDefault(columnName, defaultValue);
                 } else {
                     updateSchema.addColumn(columnName, icebergType, columnComment);
                 }
@@ -396,6 +401,13 @@ public Set<SchemaChangeEventType> getSupportedSchemaEvolutionTypes() {
                 SchemaChangeEventType.ALTER_COLUMN_TYPE);
     }
 
+    private int getFormatVersion(Table table) {
+        if (table instanceof HasTableOperations) {
+            return ((HasTableOperations) table).operations().current().formatVersion();
+        }
+        return 2;
+    }
+
     @Override
     public void close() {
         catalog = null;