loserwang1024 commented on code in PR #4316: URL: https://github.com/apache/flink-cdc/pull/4316#discussion_r2946177770
########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/utils/SchemaChangeUtil.java: ########## @@ -0,0 +1,307 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.cdc.connectors.postgres.utils; + +import org.apache.flink.cdc.common.event.AddColumnEvent; +import org.apache.flink.cdc.common.event.AlterColumnTypeEvent; +import org.apache.flink.cdc.common.event.CreateTableEvent; +import org.apache.flink.cdc.common.event.DropColumnEvent; +import org.apache.flink.cdc.common.event.RenameColumnEvent; +import org.apache.flink.cdc.common.event.SchemaChangeEvent; +import org.apache.flink.cdc.common.event.TableId; +import org.apache.flink.cdc.common.types.DataType; +import org.apache.flink.cdc.connectors.postgres.source.PostgresDialect; +import org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfig; + +import io.debezium.connector.postgresql.PostgresConnectorConfig; +import io.debezium.connector.postgresql.TypeRegistry; +import io.debezium.connector.postgresql.connection.PostgresConnection; +import io.debezium.relational.Column; +import io.debezium.relational.Table; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import static org.apache.flink.cdc.connectors.postgres.utils.PostgresSchemaUtils.toCdcTableId; +import static org.apache.flink.cdc.connectors.postgres.utils.PostgresSchemaUtils.toSchema; + +/** + * Utilities for inferring CDC schema change events by comparing before/after Debezium table + * schemas. + * + * <p>Ideally, PostgreSQL's {@code pg_attribute.attnum} would be the best way to distinguish whether + * a column was dropped-and-recreated or simply renamed. However, the pgoutput relation message does + * not include {@code attnum}. Even querying the database for the current attnum after the fact is + * unreliable due to temporal mismatch — for example, if column {@code C} is renamed to {@code D} + * and a new column {@code C} is added between two relation messages, a retroactive attnum query + * would see the <em>new</em> {@code C}'s attnum, not the original one, leading to incorrect + * inference. + * + * <p>Instead, this utility infers schema changes by comparing the before/after column lists and + * computing the <b>minimum edit</b> using four operations: rename, drop, add-at-last, and + * alter-type. When multiple interpretations are possible (e.g. {@code [c1,c2,c3,c4,c5] → + * [c1,c2,c4,c5]} could be "drop c3" or "drop c5 then rename c3→c6, c4→..." etc.), the algorithm + * favors the one with the fewest total changes. + */ +public class SchemaChangeUtil { + /** + * Infers the minimum list of schema change events that transform {@code tableBefore} into + * {@code tableAfter}. Returns a {@link CreateTableEvent} if {@code tableBefore} is null. + */ + public static List<SchemaChangeEvent> inferSchemaChangeEvent( + io.debezium.relational.TableId dbzTableId, + @Nullable Table tableBefore, + Table tableAfter, + PostgresSourceConfig sourceConfig, + PostgresDialect dialect) { + + if (tableBefore == null) { + return Collections.singletonList(toCreateTableEvent(tableAfter, sourceConfig, dialect)); + } + + TableId cdcTableId = + toCdcTableId( + dbzTableId, + sourceConfig.getDatabaseList().get(0), + sourceConfig.isIncludeDatabaseInTableId()); + PostgresConnectorConfig dbzConfig = sourceConfig.getDbzConnectorConfig(); + + try (PostgresConnection connection = dialect.openJdbcConnection()) { + TypeRegistry typeRegistry = connection.getTypeRegistry(); + return inferMinimalSchemaChanges( + cdcTableId, + tableBefore.columns(), + tableAfter.columns(), + dbzConfig, + typeRegistry); + } + } + + public static CreateTableEvent toCreateTableEvent( + Table table, PostgresSourceConfig sourceConfig, PostgresDialect dialect) { + try (PostgresConnection connection = dialect.openJdbcConnection()) { + return toCreateTableEvent(table, sourceConfig, connection.getTypeRegistry()); + } + } + + private static CreateTableEvent toCreateTableEvent( + Table table, PostgresSourceConfig sourceConfig, TypeRegistry typeRegistry) { + return new CreateTableEvent( + toCdcTableId( + table.id(), + sourceConfig.getDatabaseList().get(0), + sourceConfig.isIncludeDatabaseInTableId()), + toSchema(table, sourceConfig.getDbzConnectorConfig(), typeRegistry)); + } + + /** + * Finds the minimum schema change events to transform beforeCols into afterCols using recursion + * with memoization. Available operations: rename column, add column at last, drop column, alter + * column type. Recursion depth bounded by total column count. + */ + private static List<SchemaChangeEvent> inferMinimalSchemaChanges( Review Comment: @Daishuyuan how to head-insert in postgres sql? I have seen the document, postgres only support add column at last. https://www.postgresql.org/docs/current/sql-altertable.html Why I use this algorithm, i try best to keep the original ddl. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
