fix: add column default v2 table

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

@@ -213,11 +213,29 @@ private void applyAddColumnEventWithPosition(Table table, AddColumnEvent event)
             throws SchemaEvolveException {
 
         try {
-            UpdateSchema updateSchema = table.updateSchema();
             int formatVersion =
                     table instanceof HasTableOperations
                             ? ((HasTableOperations) table).operations().current().formatVersion()
                             : 2;
+
+            // Upgrade to format version 3 if any added column has a default value.
+            if (formatVersion < 3) {
+                boolean hasDefaults =
+                        event.getAddedColumns().stream()
+                                .anyMatch(
+                                        cwp ->
+                                                IcebergTypeUtils.parseDefaultValue(
+                                                                cwp.getAddColumn()
+                                                                        .getDefaultValueExpression(),
+                                                                cwp.getAddColumn().getType())
+                                                        != null);
+                if (hasDefaults) {
+                    table.updateProperties().set("format-version", "3").commit();
+                    formatVersion = 3;
+                }
+            }
+
+            UpdateSchema updateSchema = table.updateSchema();
             for (AddColumnEvent.ColumnWithPosition columnWithPosition : event.getAddedColumns()) {
                 Column addColumn = columnWithPosition.getAddColumn();
                 String columnName = addColumn.getName();

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/test/java/org/apache/flink/cdc/connectors/iceberg/sink/IcebergMetadataApplierTest.java

@@ -87,6 +87,97 @@ public void testApplySchemaChangeHadoopCatalog() throws InterruptedException {
         runSchemaChangeTest(catalogOptions);
     }
 
+    @Test
+    public void testAddColumnWithDefaultOnV2Table() {
+        Map<String, String> catalogOptions = new HashMap<>();
+        String warehouse =
+                new File(temporaryFolder.toFile(), UUID.randomUUID().toString()).toString();
+        catalogOptions.put("type", "hadoop");
+        catalogOptions.put("warehouse", warehouse);
+        catalogOptions.put("cache-enabled", "false");
+
+        Catalog catalog =
+                CatalogUtil.buildIcebergCatalog(
+                        "cdc-iceberg-catalog", catalogOptions, new Configuration());
+
+        IcebergMetadataApplier icebergMetadataApplier = new IcebergMetadataApplier(catalogOptions);
+        String defaultTableId = "test.v2_table";
+        TableId tableId = TableId.parse(defaultTableId);
+
+        // Create table WITHOUT default values (should be format-version 2).
+        CreateTableEvent createTableEvent =
+                new CreateTableEvent(
+                        tableId,
+                        Schema.newBuilder()
+                                .physicalColumn("id", DataTypes.BIGINT().notNull(), "column for id")
+                                .physicalColumn("name", DataTypes.VARCHAR(255), "column for name")
+                                .primaryKey("id")
+                                .build());
+        icebergMetadataApplier.applySchemaChange(createTableEvent);
+        Table table = catalog.loadTable(TableIdentifier.parse(defaultTableId));
+
+        // Verify table is format-version 2.
+        int formatVersion =
+                ((org.apache.iceberg.HasTableOperations) table)
+                        .operations()
+                        .current()
+                        .formatVersion();
+        assertThat(formatVersion).isEqualTo(2);
+
+        // Add column WITH default value on the v2 table.
+        AddColumnEvent addColumnEvent =
+                new AddColumnEvent(
+                        tableId,
+                        Collections.singletonList(
+                                AddColumnEvent.last(
+                                        new PhysicalColumn(
+                                                "status",
+                                                DataTypes.VARCHAR(50),
+                                                "column for status",
+                                                "active"))));
+        icebergMetadataApplier.applySchemaChange(addColumnEvent);
+
+        // Reload table and verify it was upgraded to format-version 3.
+        table = catalog.loadTable(TableIdentifier.parse(defaultTableId));
+        formatVersion =
+                ((org.apache.iceberg.HasTableOperations) table)
+                        .operations()
+                        .current()
+                        .formatVersion();
+        assertThat(formatVersion).isEqualTo(3);
+
+        // Verify the new column has both initial-default and write-default.
+        org.apache.iceberg.Schema expectedSchema =
+                new org.apache.iceberg.Schema(
+                        0,
+                        Arrays.asList(
+                                Types.NestedField.builder()
+                                        .withId(1)
+                                        .asRequired()
+                                        .withName("id")
+                                        .ofType(Types.LongType.get())
+                                        .withDoc("column for id")
+                                        .build(),
+                                Types.NestedField.builder()
+                                        .withId(2)
+                                        .asOptional()
+                                        .withName("name")
+                                        .ofType(Types.StringType.get())
+                                        .withDoc("column for name")
+                                        .build(),
+                                Types.NestedField.builder()
+                                        .withId(3)
+                                        .asOptional()
+                                        .withName("status")
+                                        .ofType(Types.StringType.get())
+                                        .withDoc("column for status")
+                                        .withInitialDefault(Literal.of("active"))
+                                        .withWriteDefault(Literal.of("active"))
+                                        .build()),
+                        new HashSet<>(Collections.singletonList(1)));
+        assertThat(table.schema().sameSchema(expectedSchema)).isTrue();
+    }
+
     public void runSchemaChangeTest(Map<String, String> catalogOptions) {
         Catalog catalog =
                 CatalogUtil.buildIcebergCatalog(