Hi Flink CDC community, I opened a PR for FLINK-39797 to add a Fluss pipeline sink option

auto.create.table.enabled

. PR: https://github.com/apache/flink-cdc/pull/4420 JIRA: https://issues.apache.org/jira/browse/FLINK-39797 The option defaults to

true

to preserve the current behavior. When set to

false

, the Fluss sink will fail if the target table does not already exist, instead of creating it automatically. This is useful for users who want to manage Fluss table creation explicitly. Would appreciate your review when you have time. Thanks!

@litiliu Currently, org.apache.flink.cdc.connectors.fluss.sink.FlussMetaDataApplier#applyCreateTable already creates tables in an idempotent manner. If you want to support not creating tables within the meta applier, I think we could provide a unified interface at the framework level. Not only fluss can use it. But we must also consider how to handle cases where the schemas of the upstream and downstream are inconsistent.

@Hongshun Wang Thanks for the clarification! My intention is to avoid automatically creating target tables, not to solve duplicate table creation. I agree that making this a framework-level configuration is a better approach, so other connectors can reuse the same semantics as well. I’ll rethink the design around a common table creation mode, including how to validate schema compatibility when the target table already exists, and submit an updated PR.

Thanks for the proposal. I think the core goal — validating schema compatibility for pre-existing tables — is valuable, but I'm not sure we need a new

schema.create.mode

parameter to achieve it. Schema validation on existing tables should be the default behavior of

MetadataApplier

, not something gated behind a configuration option. When a

CreateTableEvent

arrives: - If the table doesn't exist → create it (as we do today). - If the table already exists → validate schema compatibility (columns, primary keys, partition keys, etc.) and skip creation if compatible, or throw an exception if not. This is a correctness concern, not a mode choice. The

ERROR_IF_NOT_EXISTS

mode addresses a fairly niche scenario — requiring tables to be pre-created. In practice, users can already achieve this through database-level permission controls (e.g., not granting

CREATE TABLE

privileges). Adding a framework-level parameter for this seems like over-engineering. From an implementation perspective, the

SchemaCreateModeMetadataApplier

wrapper adds an extra layer that every call passes through, but the validation logic is inherently sink-specific (e.g., Fluss needs to verify PKs and partition keys, Iceberg may need to verify field IDs). It would be more natural to put this validation directly inside each sink's

MetadataApplier

implementation. My suggestion: enhance

MetadataApplier

to handle existing-table validation as part of its default

CreateTableEvent

processing, without introducing a new config parameter or wrapper class. This achieves the same goal with less complexity.

@kunni Thanks for the feedback. I have narrowed the scope of this PR. The updated design no longer introduces a framework-level schema validation abstraction. The new option is now

target.table.create.mode

, and it only controls whether Flink CDC is allowed to create a missing target table. In

ERROR_IF_NOT_EXISTS

mode, the framework only checks whether the target table already exists before delegating the

CreateTableEvent

to the original sink

MetadataApplier

. If the table does not exist, it fails early. If the table exists, the original sink-specific logic still runs as before, including any existing schema/PK/partition-key compatibility checks. I agree that database-level permissions can be used as a workaround in some cases. However, I think it is not fully equivalent to an explicit pipeline policy. Permission control depends on the target system and deployment setup, and failures usually surface as system-specific permission errors after the sink attempts table creation. With

target.table.create.mode=ERROR_IF_NOT_EXISTS

, users can declare the expected behavior directly in the pipeline config, and the framework can fail fast with a consistent error before automatic table creation is attempted. I kept this as a framework-level option because this capability is not Fluss-specific: other connectors may also need a clear way to require target tables to be pre-created and explicitly disable automatic table creation, while preserving the default

CREATE_IF_NOT_EXISTS

behavior for compatibility. What do you think about this updated direction?