[FLINK-39055] [Iceberg] Support default column values in Iceberg sink connector

[suhwan-cheon]

Summary

• In the Iceberg table version 3, default value support for columns
  • https://iceberg.apache.org/spec/#version-3-extended-types-and-capabilities
• Add default column value support for Iceberg sink connector
• Supported types: STRING/VARCHAR, BOOLEAN, TINYINT/SMALLINT/INTEGER, BIGINT, FLOAT, DOUBLE, DECIMAL
  • DATE/TIME/TIMESTAMP defaults not supported: parseDefaultValue() returns null for these types

Test

• Added test codes
• I tested the pipeline running in my local environment and confirmed that the initial-default/write-default value is written in the iceberg metadata file. https://github.com/apache/iceberg/blob/6d565b222a885c0e8db6c4ed2c80ef2b7c478061/core/src/main/java/org/apache/iceberg/SchemaUpdate.java#L173-L182

In MySQL Source

CREATE TABLE `default_test` (
  `id` int NOT NULL AUTO_INCREMENT,
  `username` varchar(50) COLLATE utf8mb4_general_ci NOT NULL,
  `status` varchar(20) COLLATE utf8mb4_general_ci DEFAULT 'active',
  `points` int DEFAULT '0',
  `created_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=4 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci

In Iceberg metadata file

"fields":[{"id":1,"name":"id","required":true,"type":"int"},{"id":2,"name":"username","required":true,"type":"string"},{"id":3,"name":"status","required":false,"type":"string","write-default":"active"},{"id":4,"name":"points","required":false,"type":"int","write-default":0},{"id":5,"name":"created_at","required":false,"type":"timestamptz"}]

If add default values column

{"id":6,"name":"add_col_test","required":false,"type":"string","initial-default":"active","write-default":"active"}

Reference

• [Iceberg v3] Column default values trinodb/trino#27837

[suhwan-cheon]

Moved up to avoid unnecessary repetition.

[suhwan-cheon]

NestedField.builder() to verify both schema structure and default values

[Copilot]

Pull request overview

Adds support for propagating column default values from Flink CDC schemas into Iceberg table metadata (Iceberg spec v3 defaults), primarily for table creation and ADD COLUMN evolution in the Iceberg sink connector.

Changes:

• Introduce IcebergTypeUtils.parseDefaultValue(...) to parse CDC default-value expressions into Iceberg Literals for a subset of primitive types.
• Apply parsed defaults during CREATE TABLE (via updateColumnDefault) and ADD COLUMN (via addColumn(..., default)), including column positioning moves.
• Add/extend tests to validate parsing behavior and that Iceberg schema equality reflects default values.

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 4 comments.

File	Description
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	Adds default-value parsing into Iceberg Literals.
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/IcebergMetadataApplier.java	Applies defaults on create/add-column schema updates.
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/test/java/org/apache/flink/cdc/connectors/iceberg/sink/IcebergMetadataApplierTest.java	Updates schema-evolution test assertions to include write-defaults.
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/test/java/org/apache/flink/cdc/connectors/iceberg/sink/utils/IcebergTypeUtilsTest.java	New unit tests for default-value parsing.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[suhwan-cheon]

@lvyanquan
The CI failure in flink-cdc-pipeline-connector-kafka module is unrelated to the changes in this PR.

The failure is caused by a Testcontainers issue where the Kafka Docker container (confluentinc/cp-kafka:7.2.2) failed to start within the timeout period:

ContainerLaunchException: Container startup failed for image confluentinc/cp-kafka:7.2.2
Caused by: Timed out waiting for log output matching '.*\[KafkaServer id=\d+\] started.*'

[lvyanquan]

@lvyanquan The CI failure in flink-cdc-pipeline-connector-kafka module is unrelated to the changes in this PR.

The failure is caused by a Testcontainers issue where the Kafka Docker container (confluentinc/cp-kafka:7.2.2) failed to start within the timeout period:

ContainerLaunchException: Container startup failed for image confluentinc/cp-kafka:7.2.2
Caused by: Timed out waiting for log output matching '.*\[KafkaServer id=\d+\] started.*'

You could rebase master to fix this.

[suhwan-cheon]

@lvyanquan The CI failure in flink-cdc-pipeline-connector-kafka module is unrelated to the changes in this PR.
The failure is caused by a Testcontainers issue where the Kafka Docker container (confluentinc/cp-kafka:7.2.2) failed to start within the timeout period:

ContainerLaunchException: Container startup failed for image confluentinc/cp-kafka:7.2.2
Caused by: Timed out waiting for log output matching '.*\[KafkaServer id=\d+\] started.*'

You could rebase master to fix this.

thank you. i rebased master
Could you please remove the extra labels? Sorry for the noise. it happened during a rebase/push issue.

[lvyanquan]

Looks like the test failure is related to the change here now.

[suhwan-cheon]

Looks like the test failure is related to the change here now.

Thank you. I just solved it by separating the table format v3 test function.

[lvyanquan]

LGTM.