[cdc] Improve cdc schema change function to remove duplicate execution logic code.

[zhuangchong]

Purpose

UpdatedDataFieldsProcessFunction#latestDataFieldContain to exclude the latest fields in the schema，But in the parent class UpdatedDataFieldsProcessFunctionBase#extractSchemaChanges method, it has been determined whether the name, type, description attributes except id are equal.

if (oldField.type().copy(true).equalsIgnoreFieldId(newField.type().copy(true))) 

paimon/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBase.java

Lines 218 to 228 in b914243

	for (DataField newField : updatedDataFields) {
	String newFieldName = StringUtils.toLowerCaseIfNeed(newField.name(), caseSensitive);
	if (oldFields.containsKey(newFieldName)) {
	DataField oldField = oldFields.get(newFieldName);
	// 1. we compare by ignoring nullable, because partition keys and primary keys might
	// be nullable in source database, but they can't be null in Paimon
	// 2. we compare by ignoring field id, the field ID is newly created and may be
	// different, we should ignore it
	if (oldField.type().copy(true).equalsIgnoreFieldId(newField.type().copy(true))) {
	// update column comment
	if (newField.description() != null

Tests

API and Format

Documentation