This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12561-daa8197bdd415dcd65d01eb2f65521d39b20fc25 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 56b063c1d81cc3ab17ff76aabd849f69d3909a6d Author: Goutam Adwant <[email protected]> AuthorDate: Wed Oct 28 13:00:32 2026 +0000 [Fix][Connector-V2] Exclude columns not published by pgoutput from the Postgres-CDC schema (#12561) Signed-off-by: Goutam Adwant <[email protected]> --- docs/en/connectors/source/PostgreSQL-CDC.md | 10 + .../introduction/concepts/incompatible-changes.md | 6 + docs/zh/connectors/source/PostgreSQL-CDC.md | 10 + .../introduction/concepts/incompatible-changes.md | 6 + .../source/PostgresIncrementalSourceFactory.java | 20 + .../postgres/utils/PostgresPublishedColumns.java | 312 +++++++++++++++ .../config/PostgresSourceConfigFactoryTest.java | 3 + .../source/PostgresIncrementalSourceTest.java | 82 ++++ .../utils/PostgresPublishedColumnsTest.java | 351 +++++++++++++++++ .../cdc/postgres/PostgresCDCGeneratedColumnIT.java | 430 +++++++++++++++++++++ .../postgrescdc_to_postgres_with_column_list.conf | 59 +++ ...scdc_to_postgres_with_existing_publication.conf | 56 +++ ...tgrescdc_to_postgres_with_generated_column.conf | 51 +++ 13 files changed, 1396 insertions(+) diff --git a/docs/en/connectors/source/PostgreSQL-CDC.md b/docs/en/connectors/source/PostgreSQL-CDC.md index afdd0e9cbc..b19bdc948e 100644 --- a/docs/en/connectors/source/PostgreSQL-CDC.md +++ b/docs/en/connectors/source/PostgreSQL-CDC.md @@ -384,6 +384,16 @@ PostgreSQL logical replication slots must be created and consumed on the primary By default, PostgreSQL CDC requires primary keys. You can specify a custom primary key via `table-names-config` with the `primaryKeys` field if the table has a unique column that can serve as an identifier. +### How are generated columns handled? + +The publication-column filter uses the effective decoding plugin: `debezium { "plugin.name" = "..." }` takes precedence over `decoding.plugin.name`. Other plugins and snapshot-only jobs bypass this filter. + +With `pgoutput`, only the columns that the publication publishes are captured. Generated columns are not published before PostgreSQL 18, and virtual generated columns are never published. An explicit publication column list can omit columns, but must include the replica identity columns when publishing UPDATE or DELETE operations. Unpublished non-key columns are excluded from the table schema, so they are neither read in the snapshot nor created in an auto-created sink table. Publication [...] + +SeaTunnel requires every primary key column in the CDC schema (including keys configured with `table-names-config.primaryKeys`) to be published. If a required key is unpublished, the job fails at startup with the publication, table and missing key columns, rather than streaming NULL keys. This also applies to INSERT-only publications, even though PostgreSQL permits them to omit replica identity columns. Include all required primary key columns in the publication column list before starti [...] + +If SeaTunnel cannot read the publication metadata at job startup, it logs an error with the failure type and keeps all columns in the schema. In that fallback, unpublished columns may be written as NULL during streaming, or a schema-change error may stop the job. Check publication access and JDBC connectivity before relying on the captured data. + ### How are replication slots managed? SeaTunnel creates or reuses the replication slot identified by `slot.name` when the job starts. diff --git a/docs/en/introduction/concepts/incompatible-changes.md b/docs/en/introduction/concepts/incompatible-changes.md index 0433a494db..0b50a23ed2 100644 --- a/docs/en/introduction/concepts/incompatible-changes.md +++ b/docs/en/introduction/concepts/incompatible-changes.md @@ -60,6 +60,12 @@ protocol still uses `234`; its wrapper scripts continue to translate help to exi - **Impact**: Such tables are now read completely, but with one snapshot split instead of several. With `exactly_once = true`, a table whose only key is a nullable unique key, or a nullable column configured in `table-names-config.primaryKeys`, now fails at startup with `Exactly once is enabled, but not found non-nullable primary key or unique key for table ...` instead of running and losing rows. This also applies to an existing job whose configured `primaryKeys` column is nullable in [...] - **Migration Guide**: Declare a NOT NULL unique column, or configure `table-names-config.primaryKeys` / `snapshotSplitColumn` with a non-nullable column, to keep parallel snapshot reading. For `exactly_once = true`, add a primary key or a non-nullable unique key to the table, or make the configured `primaryKeys` column NOT NULL in the database. +### PostgreSQL CDC pgoutput Schema + +- **Behavior change**: For streaming jobs using `decoding.plugin.name = "pgoutput"`, columns not published by the PostgreSQL publication are excluded from the source schema. This includes generated columns before PostgreSQL 18, virtual generated columns, and columns omitted by a publication column list. Snapshot-only jobs and other decoding plugins are unchanged. +- **Impact**: On the next job start or restore, snapshots and auto-created sink tables no longer include unpublished non-key columns. Existing sinks and downstream consumers that expect them may need schema or mapping changes; previously, streamed updates could write NULL into those columns or fail on a schema mismatch. If a required primary key column is unpublished, including in an INSERT-only publication, startup now fails instead of retaining the column and streaming NULL keys. +- **Migration Guide**: Check the publication and downstream schema before upgrading. On PostgreSQL 18, a publication with `publish_generated_columns = stored` can retain stored generated columns; virtual generated columns remain unpublished. For a column-list publication, include every required primary key column (including configured `table-names-config.primaryKeys`) and all other required columns. PostgreSQL also requires replica identity columns when publishing UPDATE/DELETE. Update s [...] + ### Markdown file source `child_ids` array The Markdown file source now declares `child_ids` as `ARRAY<STRING>` and emits a `String[]` for every row. Previously, it declared `STRING`, emitted comma-separated IDs for parents, and used `null` for leaves. Update downstream schemas and consumers to read an array; leaves now contain an empty array. The element IDs, types, page numbers, and position indexes retain their existing formats. diff --git a/docs/zh/connectors/source/PostgreSQL-CDC.md b/docs/zh/connectors/source/PostgreSQL-CDC.md index 4838e5db09..83b36661b8 100644 --- a/docs/zh/connectors/source/PostgreSQL-CDC.md +++ b/docs/zh/connectors/source/PostgreSQL-CDC.md @@ -374,6 +374,16 @@ SeaTunnel PostgreSQL CDC 支持 `pgoutput`（PostgreSQL 10 起内置）、`wal2j 默认需要主键。如果表有可作为唯一标识的列，可以通过 `table-names-config` 中的 `primaryKeys` 字段自定义主键。 +### 如何处理生成列？ + +发布列过滤使用实际生效的解码插件：`debezium { "plugin.name" = "..." }` 的优先级高于 `decoding.plugin.name`。其他插件和仅快照作业会跳过此过滤。 + +使用 `pgoutput` 时，只会捕获发布（publication）中发布的列。PostgreSQL 18 之前不发布生成列，虚拟生成列始终不会发布。显式发布列列表可以省略部分列，但发布 UPDATE 或 DELETE 操作时必须包含副本标识（replica identity）列。未发布的非主键列会从表结构中排除，因此既不会在快照阶段读取，也不会在自动创建的目标表中创建。没有显式列列表的发布（包括 `FOR ALL TABLES` 和 `FOR TABLES IN SCHEMA`）仍遵循生成列规则。在 PostgreSQL 18 中，如果发布使用 `publish_generated_columns = stored` 创建，则会捕获存储生成列。 + +SeaTunnel 要求发布 CDC 表结构中的所有主键列，包括通过 `table-names-config.primaryKeys` 配置的主键。如果必需的主键未发布，作业会在启动时失败，并提示发布名、表名和缺少的主键列，而不是在流式同步中输出 NULL 主键。此要求也适用于仅发布 INSERT 的发布，即使 PostgreSQL 允许其省略副本标识列。启动或恢复作业前，请将所有必需主键列加入发布列列表。 + +如果 SeaTunnel 在作业启动时无法读取发布元数据，会记录错误及失败类型，并保留表结构中的所有列。此时未发布的列可能在流式同步时被写为 NULL，或触发表结构变更错误导致作业停止。依赖捕获数据前，请检查发布元数据的访问权限和 JDBC 连通性。 + ### 复制槽如何管理？ SeaTunnel 在任务启动时会创建或复用 `slot.name` 指定的复制槽。当 `startup.mode` 为 diff --git a/docs/zh/introduction/concepts/incompatible-changes.md b/docs/zh/introduction/concepts/incompatible-changes.md index 4ee22b301f..b648d2c65b 100644 --- a/docs/zh/introduction/concepts/incompatible-changes.md +++ b/docs/zh/introduction/concepts/incompatible-changes.md @@ -59,6 +59,12 @@ MySQL JDBC Sink 现在直接绑定 SeaTunnel `TIMESTAMP`（`LocalDateTime`）， - **影响**：这类表现在会被完整读取，但快照只有一个分片而不是多个。开启 `exactly_once = true` 时，若表唯一的键是可空唯一键，或 `table-names-config.primaryKeys` 中配置的是可空列，作业会在启动时报错 `Exactly once is enabled, but not found non-nullable primary key or unique key for table ...`，而不是继续运行并丢失数据。对于已有作业，即使配置的 `primaryKeys` 列中实际没有 NULL，只要该列在数据库中可空，也会如此。 - **迁移指南**：如需保持快照并行读取，请声明一个 NOT NULL 的唯一列，或通过 `table-names-config.primaryKeys` / `snapshotSplitColumn` 配置非空列。对于 `exactly_once = true`，请为表添加主键或非空唯一键，或将配置的 `primaryKeys` 列在数据库中改为 NOT NULL。 +### PostgreSQL CDC pgoutput 表结构 + +- **行为变更**：使用 `decoding.plugin.name = "pgoutput"` 的流式作业会从源表结构中排除 PostgreSQL 发布未发布的列，包括 PostgreSQL 18 之前的生成列、虚拟生成列以及发布列列表中省略的列。仅快照作业和其他解码插件不受影响。 +- **影响**：下次启动或恢复作业时，快照和自动创建的目标表将不再包含未发布的非主键列。依赖这些列的现有目标表和下游使用方可能需要调整表结构或字段映射；此前流式更新可能将这些列写为 NULL，或因表结构不匹配而失败。如果必需的主键未发布（包括仅发布 INSERT 的发布），作业现在会在启动时失败，而不是保留该列并在流式同步中输出 NULL 主键。 +- **迁移指南**：升级前检查发布配置及下游表结构。在 PostgreSQL 18 中，设置 `publish_generated_columns = stored` 的发布可以保留存储生成列；虚拟生成列仍不会发布。使用发布列列表时，请加入所有必需主键列（包括 `table-names-config.primaryKeys` 配置的主键）和其他必需列。发布 UPDATE/DELETE 时，PostgreSQL 还要求包含副本标识列。请调整目标表结构和下游映射，移除其他未发布的列；没有显式列列表的发布仍遵循生成列规则。 + ### Markdown File Source 的 `child_ids` 数组 Markdown File Source 现在将 `child_ids` 声明为 `ARRAY<STRING>`，每行输出 `String[]`。此前该字段声明为 `STRING`，父元素输出逗号分隔的 ID，叶子元素输出 `null`。请将下游 schema 和消费逻辑改为读取数组；叶子元素现在输出空数组。元素 ID、类型、页码和位置索引保持原有格式。 diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSourceFactory.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSourceFactory.java index 5d73d3518f..b80e54d2b3 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSourceFactory.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSourceFactory.java @@ -35,18 +35,23 @@ import org.apache.seatunnel.connectors.cdc.base.option.StartupMode; import org.apache.seatunnel.connectors.cdc.base.source.BaseChangeStreamTableSourceFactory; import org.apache.seatunnel.connectors.cdc.base.utils.CatalogTableUtils; import org.apache.seatunnel.connectors.seatunnel.cdc.postgres.config.PostgresIncrementalSourceOptions; +import org.apache.seatunnel.connectors.seatunnel.cdc.postgres.utils.PostgresPublishedColumns; import org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcCommonOptions; import com.google.auto.service.AutoService; import lombok.extern.slf4j.Slf4j; import java.io.Serializable; +import java.util.Collections; import java.util.List; import java.util.Optional; @AutoService(Factory.class) @Slf4j public class PostgresIncrementalSourceFactory extends BaseChangeStreamTableSourceFactory { + + private static final String PGOUTPUT = "pgoutput"; + @Override public String factoryIdentifier() { return org.apache.seatunnel.connectors.seatunnel.cdc.postgres.source @@ -121,6 +126,21 @@ public class PostgresIncrementalSourceFactory extends BaseChangeStreamTableSourc CatalogTableUtils.mergeCatalogTableConfig( catalogTables, tableConfigs.get(), s -> TablePath.of(s, true)); } + // Debezium passthrough properties override the connector's decoding plugin option. + String decodingPlugin = + config.getOptional(SourceOptions.DEBEZIUM_PROPERTIES) + .orElse(Collections.emptyMap()) + .getOrDefault( + "plugin.name", + config.get( + PostgresIncrementalSourceOptions.DECODING_PLUGIN_NAME)); + // pgoutput streams only published columns; snapshot-only jobs never stream + if (PGOUTPUT.equalsIgnoreCase(decodingPlugin.trim()) + && config.get(PostgresSourceOptions.STARTUP_MODE) + != StartupMode.SNAPSHOT_ONLY) { + catalogTables = + PostgresPublishedColumns.retainPublishedColumns(config, catalogTables); + } return (SeaTunnelSource<T, SplitT, StateT>) new org.apache.seatunnel.connectors.seatunnel.cdc.postgres.source .PostgresIncrementalSource<>(config, catalogTables); diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/utils/PostgresPublishedColumns.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/utils/PostgresPublishedColumns.java new file mode 100644 index 0000000000..11042af5e2 --- /dev/null +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/utils/PostgresPublishedColumns.java @@ -0,0 +1,312 @@ +/* + * 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.seatunnel.connectors.seatunnel.cdc.postgres.utils; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.table.catalog.CatalogTable; +import org.apache.seatunnel.api.table.catalog.Column; +import org.apache.seatunnel.api.table.catalog.ConstraintKey; +import org.apache.seatunnel.api.table.catalog.PrimaryKey; +import org.apache.seatunnel.api.table.catalog.TableSchema; +import org.apache.seatunnel.common.utils.SeaTunnelException; +import org.apache.seatunnel.connectors.cdc.base.option.JdbcSourceOptions; +import org.apache.seatunnel.connectors.cdc.base.option.SourceOptions; +import org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcCommonOptions; + +import lombok.extern.slf4j.Slf4j; + +import java.sql.Array; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +/** + * pgoutput only streams the columns a publication publishes: generated columns are never published + * before PostgreSQL 18, virtual generated columns are never published, and publication column lists + * can omit non-identity columns. Columns that are not streamed must not be part of the CDC schema, + * otherwise snapshot rows carry a value while streamed rows carry NULL. + */ +@Slf4j +public final class PostgresPublishedColumns { + + static final String DEFAULT_PUBLICATION_NAME = "dbz_publication"; + private static final String PUBLICATION_NAME_KEY = "publication.name"; + private static final String PUBLICATION_AUTOCREATE_MODE_KEY = "publication.autocreate.mode"; + + private PostgresPublishedColumns() {} + + public static List<CatalogTable> retainPublishedColumns( + ReadonlyConfig config, List<CatalogTable> tables) { + Map<String, String> debeziumProperties = + config.getOptional(SourceOptions.DEBEZIUM_PROPERTIES) + .orElse(Collections.emptyMap()); + String publication = publicationName(debeziumProperties); + // In filtered mode Debezium rewrites an existing publication with + // ALTER PUBLICATION ... SET TABLE, which drops any column lists. + boolean useColumnLists = + !"filtered" + .equalsIgnoreCase( + debeziumProperties + .getOrDefault(PUBLICATION_AUTOCREATE_MODE_KEY, "") + .trim()); + Map<String, Set<String>> publishedByTable; + Map<String, Map<String, String>> generatedByTable; + boolean publishesStoredColumns; + try (Connection connection = + DriverManager.getConnection( + config.get(JdbcCommonOptions.URL), + config.get(JdbcSourceOptions.USERNAME), + config.get(JdbcSourceOptions.PASSWORD))) { + int serverVersion = serverVersion(connection); + if (serverVersion < 120000) { + // no generated columns and no publication column lists before PostgreSQL 12 + return tables; + } + String publishGeneratedColumns = + publishGeneratedColumns(connection, serverVersion, publication); + publishedByTable = + useColumnLists && publishGeneratedColumns != null && serverVersion >= 150000 + ? publishedColumnsByTable(connection, publication) + : Collections.emptyMap(); + generatedByTable = generatedColumnsByTable(connection); + publishesStoredColumns = "s".equals(publishGeneratedColumns); + } catch (Exception e) { + log.error( + "Failed to read columns published by '{}'; keeping all table columns. " + + "Unpublished columns may be written as NULL during streaming or cause a " + + "schema-change failure. Check publication access and JDBC connectivity. " + + "Failure type: {}, SQL state: {}", + publication, + e.getClass().getSimpleName(), + e instanceof SQLException ? ((SQLException) e).getSQLState() : null); + return tables; + } + + // Schema validation must not fall back to keeping known unpublished primary keys. + List<CatalogTable> result = new ArrayList<>(tables.size()); + for (CatalogTable table : tables) { + String key = + tableKey(table.getTableId().getSchemaName(), table.getTableId().getTableName()); + Set<String> excluded = + excludedColumns( + columnNames(table), + publishedByTable.get(key), + generatedByTable.getOrDefault(key, Collections.emptyMap()), + publishesStoredColumns); + result.add(retainColumns(table, excluded, publication)); + } + return result; + } + + static String publicationName(Map<String, String> debeziumProperties) { + String name = debeziumProperties.get(PUBLICATION_NAME_KEY); + return name == null || name.trim().isEmpty() ? DEFAULT_PUBLICATION_NAME : name.trim(); + } + + /** + * Returns the columns pgoutput will not stream for a table. + * + * @param publishedColumns columns in an explicit publication column list, or {@code null} when + * there is no explicit list or it does not apply + * @param generatedColumns generated columns of the table and their {@code attgenerated} kind + */ + static Set<String> excludedColumns( + List<String> tableColumns, + Set<String> publishedColumns, + Map<String, String> generatedColumns, + boolean publishesStoredColumns) { + Set<String> excluded = new LinkedHashSet<>(); + for (String column : tableColumns) { + if (publishedColumns != null) { + if (!publishedColumns.contains(column)) { + excluded.add(column); + } + continue; + } + String generated = generatedColumns.get(column); + if (generated != null + && !generated.isEmpty() + && !(publishesStoredColumns && "s".equals(generated))) { + excluded.add(column); + } + } + return excluded; + } + + static CatalogTable retainColumns( + CatalogTable table, Set<String> excluded, String publication) { + TableSchema schema = table.getTableSchema(); + PrimaryKey primaryKey = schema.getPrimaryKey(); + Set<String> removed = new LinkedHashSet<>(excluded); + if (primaryKey != null) { + Set<String> unpublishedKeys = new LinkedHashSet<>(primaryKey.getColumnNames()); + unpublishedKeys.retainAll(excluded); + if (!unpublishedKeys.isEmpty()) { + throw new SeaTunnelException( + String.format( + "Publication '%s' does not publish required primary key columns %s" + + " for table %s. Include all required primary key columns in" + + " the publication column list before starting Postgres-CDC," + + " including for INSERT-only publications.", + publication, unpublishedKeys, table.getTablePath())); + } + } + if (removed.isEmpty()) { + return table; + } + log.info( + "Table {} columns {} are not published by '{}' and are excluded from the CDC schema", + table.getTablePath(), + removed, + publication); + List<Column> kept = + schema.getColumns().stream() + .filter(column -> !removed.contains(column.getName())) + .collect(Collectors.toList()); + List<ConstraintKey> constraintKeys = + schema.getConstraintKeys().stream() + .filter( + key -> + key.getColumnNames().stream() + .noneMatch( + keyColumn -> + removed.contains( + keyColumn.getColumnName()))) + .collect(Collectors.toList()); + TableSchema retained = + TableSchema.builder() + .columns(kept) + .primaryKey(primaryKey) + .constraintKey(constraintKeys) + .build(); + return CatalogTable.of( + table.getTableId(), + retained, + table.getOptions(), + table.getPartitionKeys(), + table.getComment(), + table.getCatalogName(), + table.getMetadataSchema()); + } + + private static List<String> columnNames(CatalogTable table) { + return table.getTableSchema().getColumns().stream() + .map(Column::getName) + .collect(Collectors.toList()); + } + + private static String tableKey(String schema, String table) { + return schema + "." + table; + } + + private static int serverVersion(Connection connection) throws SQLException { + try (PreparedStatement statement = + connection.prepareStatement( + "SELECT current_setting('server_version_num')"); + ResultSet rs = statement.executeQuery()) { + rs.next(); + return Integer.parseInt(rs.getString(1)); + } + } + + /** + * Returns {@code pubgencols} of an existing publication ({@code "n"} before PostgreSQL 18), or + * {@code null} when the publication does not exist yet. + */ + private static String publishGeneratedColumns( + Connection connection, int serverVersion, String publication) throws SQLException { + String column = serverVersion >= 180000 ? "pubgencols" : "'n'"; + try (PreparedStatement statement = + connection.prepareStatement( + "SELECT " + column + " FROM pg_publication WHERE pubname = ?")) { + statement.setString(1, publication); + try (ResultSet rs = statement.executeQuery()) { + return rs.next() ? rs.getString(1) : null; + } + } + } + + private static Map<String, Set<String>> publishedColumnsByTable( + Connection connection, String publication) throws SQLException { + Map<String, Set<String>> published = new HashMap<>(); + try (PreparedStatement statement = + connection.prepareStatement( + "SELECT pt.schemaname, pt.tablename, pt.attnames, pr.prattrs IS NOT NULL" + + " FROM pg_publication_tables pt" + + " JOIN pg_publication p ON p.pubname = pt.pubname" + + " JOIN pg_namespace n ON n.nspname = pt.schemaname" + + " JOIN pg_class c ON c.relnamespace = n.oid" + + " AND c.relname = pt.tablename" + + " LEFT JOIN pg_publication_rel pr ON pr.prpubid = p.oid" + + " AND pr.prrelid = c.oid" + + " WHERE pt.pubname = ?")) { + statement.setString(1, publication); + try (ResultSet rs = statement.executeQuery()) { + while (rs.next()) { + // PG15 synthesizes attnames even without a column list. All-tables and + // schema publications also have no explicit list and use generated-column + // rules. + if (!rs.getBoolean(4)) { + continue; + } + Array attnames = rs.getArray(3); + if (attnames != null) { + published.put( + tableKey(rs.getString(1), rs.getString(2)), + new HashSet<>(Arrays.asList((String[]) attnames.getArray()))); + } + } + } + } + return published; + } + + private static Map<String, Map<String, String>> generatedColumnsByTable(Connection connection) + throws SQLException { + Map<String, Map<String, String>> generated = new HashMap<>(); + try (PreparedStatement statement = + connection.prepareStatement( + "SELECT n.nspname, c.relname, a.attname, a.attgenerated" + + " FROM pg_attribute a" + + " JOIN pg_class c ON a.attrelid = c.oid" + + " JOIN pg_namespace n ON c.relnamespace = n.oid" + + " WHERE a.attnum > 0 AND NOT a.attisdropped" + + " AND a.attgenerated <> ''"); + ResultSet rs = statement.executeQuery()) { + while (rs.next()) { + generated + .computeIfAbsent( + tableKey(rs.getString(1), rs.getString(2)), k -> new HashMap<>()) + .put(rs.getString(3), rs.getString(4)); + } + } + return generated; + } +} diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/config/PostgresSourceConfigFactoryTest.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/config/PostgresSourceConfigFactoryTest.java index cbf71ae18c..2ee0b54dd8 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/config/PostgresSourceConfigFactoryTest.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/config/PostgresSourceConfigFactoryTest.java @@ -172,10 +172,13 @@ public class PostgresSourceConfigFactoryTest { PostgresSourceConfigFactory factory = baseFactory(); Properties dbzProperties = new Properties(); dbzProperties.setProperty("slot.name", "custom_slot"); + dbzProperties.setProperty("plugin.name", "decoderbufs"); factory.debeziumProperties(dbzProperties); Assertions.assertEquals( "custom_slot", factory.create(0).getDbzConfiguration().getString("slot.name")); + Assertions.assertEquals( + "decoderbufs", factory.create(0).getDbzConfiguration().getString("plugin.name")); } /** diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSourceTest.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSourceTest.java index a5d5c1e2c2..f82593de5f 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSourceTest.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSourceTest.java @@ -21,18 +21,25 @@ import org.apache.seatunnel.api.configuration.ReadonlyConfig; import org.apache.seatunnel.api.options.ConnectorCommonOptions; import org.apache.seatunnel.api.table.catalog.CatalogTable; import org.apache.seatunnel.api.table.catalog.CatalogTableUtil; +import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext; import org.apache.seatunnel.api.table.type.BasicType; import org.apache.seatunnel.api.table.type.SeaTunnelDataType; import org.apache.seatunnel.api.table.type.SeaTunnelRow; import org.apache.seatunnel.api.table.type.SeaTunnelRowType; import org.apache.seatunnel.common.utils.SeaTunnelException; import org.apache.seatunnel.connectors.cdc.base.option.JdbcSourceOptions; +import org.apache.seatunnel.connectors.cdc.base.option.SourceOptions; import org.apache.seatunnel.connectors.cdc.debezium.DebeziumDeserializationSchema; import org.apache.seatunnel.connectors.seatunnel.cdc.postgres.config.PostgresIncrementalSourceOptions; +import org.apache.seatunnel.connectors.seatunnel.cdc.postgres.utils.PostgresPublishedColumns; import org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcCommonOptions; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; +import org.mockito.MockedConstruction; +import org.mockito.MockedStatic; import io.debezium.connector.postgresql.connection.PostgresConnection; import io.debezium.connector.postgresql.connection.ServerInfo; @@ -43,9 +50,13 @@ import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.atomic.AtomicReference; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockConstruction; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.when; /** @@ -119,6 +130,77 @@ public class PostgresIncrementalSourceTest { Collections.singletonList(ORDERS))); } + @ParameterizedTest + @CsvSource( + value = { + "DEFAULT, DEFAULT, initial, true", + "decoderbufs, DEFAULT, initial, false", + "DEFAULT, decoderbufs, initial, false", + "decoderbufs, pgoutput, initial, true", + "pgoutput, pgoutput, initial, true", + "pgoutput, DEFAULT, snapshot-only, false", + "decoderbufs, pgoutput, snapshot-only, false" + }, + nullValues = "DEFAULT") + void testPublishedColumnFilterUsesEffectivePlugin( + String configuredPlugin, String overridePlugin, String startupMode, boolean filter) { + Map<String, Object> options = new HashMap<>(baseOptions()); + if (configuredPlugin != null) { + options.put( + PostgresIncrementalSourceOptions.DECODING_PLUGIN_NAME.key(), configuredPlugin); + } + if (overridePlugin != null) { + options.put( + SourceOptions.DEBEZIUM_PROPERTIES.key(), + Collections.singletonMap("plugin.name", overridePlugin)); + } + options.put(PostgresSourceOptions.STARTUP_MODE.key(), startupMode); + ReadonlyConfig config = ReadonlyConfig.fromMap(options); + ClassLoader classLoader = getClass().getClassLoader(); + List<CatalogTable> discoveredTables = + Collections.singletonList( + CatalogTableUtil.getCatalogTable( + "postgres", + "inventory", + "public", + "orders", + new SeaTunnelRowType( + new String[] {"id", "note"}, + new SeaTunnelDataType[] { + BasicType.INT_TYPE, BasicType.STRING_TYPE + }))); + List<CatalogTable> publishedTables = catalogTables(); + AtomicReference<Object> sourceTables = new AtomicReference<>(); + TableSourceFactoryContext context = new TableSourceFactoryContext(config, classLoader); + try (MockedStatic<CatalogTableUtil> catalog = mockStatic(CatalogTableUtil.class); + MockedStatic<PostgresPublishedColumns> published = + mockStatic(PostgresPublishedColumns.class); + MockedConstruction<PostgresIncrementalSource> sources = + mockConstruction( + PostgresIncrementalSource.class, + (source, construction) -> + sourceTables.set(construction.arguments().get(1)))) { + catalog.when(() -> CatalogTableUtil.getCatalogTables(config, classLoader)) + .thenReturn(discoveredTables); + published + .when( + () -> + PostgresPublishedColumns.retainPublishedColumns( + config, discoveredTables)) + .thenReturn(publishedTables); + + new PostgresIncrementalSourceFactory() + .restoreSource(context, Collections.emptyList()) + .createSource(); + + Assertions.assertEquals(1, sources.constructed().size()); + Assertions.assertSame(filter ? publishedTables : discoveredTables, sourceTables.get()); + published.verify( + () -> PostgresPublishedColumns.retainPublishedColumns(config, discoveredTables), + times(filter ? 1 : 0)); + } + } + /** * Builds the dialect through the real {@code createDataSourceDialect(config)} entry point, so * the test exercises the wiring under review rather than constructing the dialect directly. diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/utils/PostgresPublishedColumnsTest.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/utils/PostgresPublishedColumnsTest.java new file mode 100644 index 0000000000..3949a58ecb --- /dev/null +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/utils/PostgresPublishedColumnsTest.java @@ -0,0 +1,351 @@ +/* + * 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.seatunnel.connectors.seatunnel.cdc.postgres.utils; + +import org.apache.seatunnel.shade.com.typesafe.config.ConfigFactory; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.table.catalog.CatalogTable; +import org.apache.seatunnel.api.table.catalog.Column; +import org.apache.seatunnel.api.table.catalog.ConstraintKey; +import org.apache.seatunnel.api.table.catalog.PhysicalColumn; +import org.apache.seatunnel.api.table.catalog.PrimaryKey; +import org.apache.seatunnel.api.table.catalog.TableIdentifier; +import org.apache.seatunnel.api.table.catalog.TableSchema; +import org.apache.seatunnel.api.table.type.BasicType; +import org.apache.seatunnel.common.utils.SeaTunnelException; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; + +import java.sql.Array; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.when; + +public class PostgresPublishedColumnsTest { + + private static final List<String> COLUMNS = + Arrays.asList("id", "price", "note", "stored_total", "virtual_total"); + + @Test + public void testGeneratedColumnsAreExcludedWhenPublicationDoesNotCoverTable() { + Assertions.assertEquals( + new LinkedHashSet<>(Arrays.asList("stored_total", "virtual_total")), + PostgresPublishedColumns.excludedColumns(COLUMNS, null, generated(), false)); + } + + @Test + public void testStoredColumnsAreKeptWhenPublicationPublishesThem() { + Assertions.assertEquals( + Collections.singleton("virtual_total"), + PostgresPublishedColumns.excludedColumns(COLUMNS, null, generated(), true)); + } + + @Test + public void testPublicationColumnListDecidesWhenTableIsPublished() { + Assertions.assertEquals( + new LinkedHashSet<>(Arrays.asList("note", "virtual_total")), + PostgresPublishedColumns.excludedColumns( + COLUMNS, + new HashSet<>(Arrays.asList("id", "price", "stored_total")), + generated(), + true)); + } + + @Test + public void testNothingIsExcludedWithoutGeneratedColumns() { + Assertions.assertTrue( + PostgresPublishedColumns.excludedColumns( + Arrays.asList("id", "price"), null, Collections.emptyMap(), false) + .isEmpty()); + } + + @Test + public void testRetainColumnsRemovesExcludedColumnsAndTheirKeys() { + CatalogTable table = table(); + + CatalogTable retained = + PostgresPublishedColumns.retainColumns( + table, + new LinkedHashSet<>(Arrays.asList("stored_total", "virtual_total")), + "dbz_publication"); + + Assertions.assertEquals( + Arrays.asList("id", "price", "note"), + columnNames(retained.getTableSchema().getColumns())); + Assertions.assertEquals( + Collections.singletonList("id"), + retained.getTableSchema().getPrimaryKey().getColumnNames()); + Assertions.assertTrue(retained.getTableSchema().getConstraintKeys().isEmpty()); + Assertions.assertEquals(table.getTableId(), retained.getTableId()); + } + + @Test + public void testUnpublishedPrimaryKeyFailsFast() { + CatalogTable table = table(); + + SeaTunnelException error = + Assertions.assertThrows( + SeaTunnelException.class, + () -> + PostgresPublishedColumns.retainColumns( + table, Collections.singleton("id"), "orders_pub")); + Assertions.assertTrue(error.getMessage().contains("orders_pub")); + Assertions.assertTrue(error.getMessage().contains("public.orders")); + Assertions.assertTrue(error.getMessage().contains("[id]")); + Assertions.assertTrue( + error.getMessage().contains("Include all required primary key columns")); + } + + @Test + public void testExistingPg15PublicationWithoutColumnListExcludesGeneratedColumns() + throws SQLException { + assertPublishedColumns( + 150000, + "n", + false, + COLUMNS, + Collections.emptyMap(), + Arrays.asList("id", "price", "note")); + } + + @Test + public void testPg18StoredPublicationWithoutColumnListKeepsOnlyStoredGeneratedColumns() + throws SQLException { + assertPublishedColumns( + 180000, + "s", + false, + COLUMNS, + Collections.emptyMap(), + Arrays.asList("id", "price", "note", "stored_total")); + } + + @Test + public void testExistingExplicitListOverridesGeneratedColumnSetting() throws SQLException { + assertPublishedColumns( + 180000, + "s", + true, + Arrays.asList("id", "price"), + Collections.emptyMap(), + Arrays.asList("id", "price")); + } + + @Test + public void testFilteredModeIgnoresExistingColumnList() throws SQLException { + assertPublishedColumns( + 150000, + "n", + true, + Arrays.asList("id", "price"), + Collections.singletonMap("publication.autocreate.mode", "filtered"), + Arrays.asList("id", "price", "note")); + } + + @Test + public void testMissingPublicationExcludesGeneratedColumns() throws SQLException { + assertPublishedColumns( + 150000, + null, + false, + COLUMNS, + Collections.emptyMap(), + Arrays.asList("id", "price", "note")); + } + + @Test + public void testPg14DoesNotQueryColumnLists() throws SQLException { + assertPublishedColumns( + 140000, + "n", + true, + Arrays.asList("price"), + Collections.emptyMap(), + Arrays.asList("id", "price", "note")); + } + + @Test + public void testBeforePg12DoesNotQueryGeneratedColumns() throws SQLException { + assertPublishedColumns( + 110000, "n", true, Arrays.asList("price"), Collections.emptyMap(), COLUMNS); + } + + @Test + public void testUnpublishedPrimaryKeyIsNotSwallowedByMetadataFallback() throws SQLException { + Assertions.assertThrows( + SeaTunnelException.class, + () -> + assertPublishedColumns( + 150000, + "n", + true, + Arrays.asList("price"), + Collections.emptyMap(), + COLUMNS)); + } + + @Test + public void testMetadataFailureKeepsOriginalSchema() throws SQLException { + SQLException failure = new SQLException("private connection details", "08001"); + try (MockedStatic<DriverManager> driverManager = mockStatic(DriverManager.class)) { + driverManager + .when( + () -> + DriverManager.getConnection( + "jdbc:test:publication", "user", "secret")) + .thenThrow(failure); + List<CatalogTable> tables = Collections.singletonList(table()); + Assertions.assertSame( + tables, + PostgresPublishedColumns.retainPublishedColumns( + config(Collections.emptyMap()), tables)); + } + } + + @Test + public void testPublicationNameDefaultsToDebeziumPublication() { + Assertions.assertEquals( + "dbz_publication", PostgresPublishedColumns.publicationName(new HashMap<>())); + Assertions.assertEquals( + "orders_pub", + PostgresPublishedColumns.publicationName( + Collections.singletonMap("publication.name", "orders_pub"))); + } + + private static void assertPublishedColumns( + int version, + String pubgencols, + boolean explicitList, + List<String> published, + Map<String, String> debezium, + List<String> expected) + throws SQLException { + Connection connection = mock(Connection.class); + when(connection.prepareStatement(anyString())) + .thenAnswer( + invocation -> { + String sql = invocation.getArgument(0); + PreparedStatement statement = mock(PreparedStatement.class); + ResultSet rs = mock(ResultSet.class); + when(statement.executeQuery()).thenReturn(rs); + if (sql.contains("server_version_num")) { + when(rs.next()).thenReturn(true, false); + when(rs.getString(1)).thenReturn(String.valueOf(version)); + } else if (sql.contains("FROM pg_publication WHERE")) { + when(rs.next()).thenReturn(pubgencols != null, false); + when(rs.getString(1)).thenReturn(pubgencols); + } else if (sql.contains("FROM pg_publication_tables")) { + Assertions.assertTrue(version >= 150000); + when(rs.next()).thenReturn(true, false); + when(rs.getString(1)).thenReturn("public"); + when(rs.getString(2)).thenReturn("orders"); + Array columns = mock(Array.class); + when(columns.getArray()) + .thenReturn(published.toArray(new String[0])); + when(rs.getArray(3)).thenReturn(columns); + when(rs.getBoolean(4)).thenReturn(explicitList); + } else if (sql.contains("FROM pg_attribute")) { + Assertions.assertTrue(version >= 120000); + when(rs.next()).thenReturn(true, true, false); + when(rs.getString(1)).thenReturn("public"); + when(rs.getString(2)).thenReturn("orders"); + when(rs.getString(3)).thenReturn("stored_total", "virtual_total"); + when(rs.getString(4)).thenReturn("s", "v"); + } else { + Assertions.fail("Unexpected catalog query: " + sql); + } + return statement; + }); + try (MockedStatic<DriverManager> driverManager = mockStatic(DriverManager.class)) { + driverManager + .when( + () -> + DriverManager.getConnection( + "jdbc:test:publication", "user", "secret")) + .thenReturn(connection); + List<CatalogTable> retained = + PostgresPublishedColumns.retainPublishedColumns( + config(debezium), Collections.singletonList(table())); + Assertions.assertEquals( + expected, columnNames(retained.get(0).getTableSchema().getColumns())); + } + } + + private static ReadonlyConfig config(Map<String, String> debezium) { + Map<String, Object> options = new HashMap<>(); + options.put("url", "jdbc:test:publication"); + options.put("username", "user"); + options.put("password", "secret"); + options.put("debezium", debezium); + return ReadonlyConfig.fromConfig(ConfigFactory.parseMap(options)); + } + + private static Map<String, String> generated() { + Map<String, String> generated = new HashMap<>(); + generated.put("stored_total", "s"); + generated.put("virtual_total", "v"); + return generated; + } + + private static CatalogTable table() { + TableSchema.Builder builder = TableSchema.builder(); + COLUMNS.forEach(name -> builder.column(column(name))); + TableSchema schema = + builder.primaryKey(PrimaryKey.of("pk", Collections.singletonList("id"))) + .constraintKey( + ConstraintKey.of( + ConstraintKey.ConstraintType.UNIQUE_KEY, + "uk_total", + Collections.singletonList( + ConstraintKey.ConstraintKeyColumn.of( + "stored_total", null)))) + .build(); + return CatalogTable.of( + TableIdentifier.of("Postgres", "db", "public", "orders"), + schema, + new HashMap<>(), + Collections.emptyList(), + ""); + } + + private static Column column(String name) { + return PhysicalColumn.of(name, BasicType.INT_TYPE, (Long) null, true, null, null); + } + + private static List<String> columnNames(List<Column> columns) { + return columns.stream().map(Column::getName).collect(Collectors.toList()); + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/PostgresCDCGeneratedColumnIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/PostgresCDCGeneratedColumnIT.java new file mode 100644 index 0000000000..9446385685 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/PostgresCDCGeneratedColumnIT.java @@ -0,0 +1,430 @@ +/* + * 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.seatunnel.connectors.seatunnel.cdc.postgres; + +import org.apache.seatunnel.e2e.common.TestResource; +import org.apache.seatunnel.e2e.common.TestSuiteBase; +import org.apache.seatunnel.e2e.common.container.ContainerExtendedFactory; +import org.apache.seatunnel.e2e.common.container.EngineType; +import org.apache.seatunnel.e2e.common.container.TestContainer; +import org.apache.seatunnel.e2e.common.junit.DisabledOnContainer; +import org.apache.seatunnel.e2e.common.junit.TestContainerExtension; +import org.apache.seatunnel.e2e.common.util.DependencyJar; +import org.apache.seatunnel.e2e.common.util.JobIdGenerator; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.TestTemplate; +import org.testcontainers.containers.Container; +import org.testcontainers.containers.PostgreSQLContainer; +import org.testcontainers.containers.output.Slf4jLogConsumer; +import org.testcontainers.lifecycle.Startables; +import org.testcontainers.utility.DockerImageName; + +import lombok.extern.slf4j.Slf4j; + +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.stream.Stream; + +import static org.awaitility.Awaitility.given; + +/** Generated columns are not streamed by pgoutput, so they must not be part of the CDC schema. */ +@Slf4j +public class PostgresCDCGeneratedColumnIT extends TestSuiteBase implements TestResource { + + private static final String POSTGRES_HOST = "postgres_generated_column_e2e"; + private static final String COLUMN_LIST_HOST = "postgres_column_list_e2e"; + private static final String DATABASE = "generated_cdc"; + private static final String SOURCE_TABLE = "public.orders"; + private static final String SINK_TABLE = "public.sink_orders"; + private static final String COLUMN_LIST_SOURCE_TABLE = "public.column_orders"; + private static final String COLUMN_LIST_SINK_TABLE = "public.sink_column_orders"; + private static final String POSTGRES_CDC_PLUGIN_LIB = "/tmp/seatunnel/plugins/Postgres-CDC/lib"; + + // generated columns need PostgreSQL 12+ + private static final PostgreSQLContainer<?> POSTGRES_CONTAINER = + new PostgreSQLContainer<>(DockerImageName.parse("postgres:14-alpine")) + .withNetwork(NETWORK) + .withNetworkAliases(POSTGRES_HOST) + .withUsername("postgres") + .withPassword("postgres") + .withDatabaseName(DATABASE) + .withLogConsumer(new Slf4jLogConsumer(log)) + .withCommand("postgres", "-c", "wal_level=logical", "-c", "fsync=off"); + + private static final PostgreSQLContainer<?> COLUMN_LIST_CONTAINER = + new PostgreSQLContainer<>(DockerImageName.parse("postgres:15-alpine")) + .withNetwork(NETWORK) + .withNetworkAliases(COLUMN_LIST_HOST) + .withUsername("postgres") + .withPassword("postgres") + .withDatabaseName(DATABASE) + .withLogConsumer(new Slf4jLogConsumer(log)) + .withCommand("postgres", "-c", "wal_level=logical", "-c", "fsync=off"); + + @TestContainerExtension + protected final ContainerExtendedFactory extendedFactory = + container -> + DependencyJar.of(org.postgresql.Driver.class) + .copyTo(container, POSTGRES_CDC_PLUGIN_LIB); + + @BeforeAll + @Override + public void startUp() { + Startables.deepStart(Stream.of(POSTGRES_CONTAINER, COLUMN_LIST_CONTAINER)).join(); + executeSql( + "CREATE TABLE " + + SOURCE_TABLE + + " (id INT PRIMARY KEY, price NUMERIC(10, 2)," + + " total NUMERIC(10, 2) GENERATED ALWAYS AS (price * 2) STORED)"); + executeSql("ALTER TABLE " + SOURCE_TABLE + " REPLICA IDENTITY FULL"); + executeSql("INSERT INTO " + SOURCE_TABLE + " (id, price) VALUES (1, 1.00), (2, 2.00)"); + executeSql( + COLUMN_LIST_CONTAINER, + "CREATE TABLE " + + COLUMN_LIST_SOURCE_TABLE + + " (id INT PRIMARY KEY, price NUMERIC(10, 2), note TEXT)"); + executeSql( + COLUMN_LIST_CONTAINER, + "INSERT INTO " + + COLUMN_LIST_SOURCE_TABLE + + " (id, price, note) VALUES (1, 1.00, 'first'), (2, 2.00, 'second')"); + executeSql( + COLUMN_LIST_CONTAINER, + "CREATE PUBLICATION seatunnel_column_list FOR TABLE " + + COLUMN_LIST_SOURCE_TABLE + + " (id, price)"); + } + + @AfterAll + @Override + public void tearDown() { + POSTGRES_CONTAINER.close(); + COLUMN_LIST_CONTAINER.close(); + } + + @TestTemplate + @DisabledOnContainer( + value = {}, + type = {EngineType.SPARK, EngineType.FLINK}, + disabledReason = "Currently only Zeta supports Postgres-CDC schema evolution") + public void testGeneratedColumnIsExcludedFromCdcSchema(TestContainer container) { + CompletableFuture.runAsync( + () -> { + try { + container.executeJob("/postgrescdc_to_postgres_with_generated_column.conf"); + } catch (Exception e) { + log.error("Commit task exception :" + e.getMessage()); + throw new RuntimeException(e); + } + }); + + given().ignoreExceptions() + .await() + .atMost(120, TimeUnit.SECONDS) + .untilAsserted( + () -> + Assertions.assertIterableEquals( + query("SELECT id, price FROM " + SOURCE_TABLE), + query("SELECT id, price FROM " + SINK_TABLE))); + + // the first streamed change used to fail the job or write NULL into the generated column + executeSql("UPDATE " + SOURCE_TABLE + " SET price = 7.00 WHERE id = 1"); + executeSql("INSERT INTO " + SOURCE_TABLE + " (id, price) VALUES (3, 3.00)"); + + given().ignoreExceptions() + .await() + .atMost(120, TimeUnit.SECONDS) + .untilAsserted( + () -> + Assertions.assertIterableEquals( + query("SELECT id, price FROM " + SOURCE_TABLE), + query("SELECT id, price FROM " + SINK_TABLE))); + Assertions.assertTrue( + query( + "SELECT column_name FROM information_schema.columns" + + " WHERE table_name = 'sink_orders'" + + " AND column_name = 'total'") + .isEmpty()); + } + + @TestTemplate + @DisabledOnContainer( + value = {}, + type = {EngineType.SPARK, EngineType.FLINK}, + disabledReason = "Currently only Zeta supports Postgres-CDC schema evolution") + public void testPublicationColumnListExcludesUnpublishedColumn(TestContainer container) { + CompletableFuture.runAsync( + () -> { + try { + container.executeJob("/postgrescdc_to_postgres_with_column_list.conf"); + } catch (Exception e) { + log.error("Commit task exception :" + e.getMessage()); + throw new RuntimeException(e); + } + }); + + given().ignoreExceptions() + .await() + .atMost(120, TimeUnit.SECONDS) + .untilAsserted( + () -> + Assertions.assertIterableEquals( + query( + COLUMN_LIST_CONTAINER, + "SELECT id, price FROM " + + COLUMN_LIST_SOURCE_TABLE), + query( + COLUMN_LIST_CONTAINER, + "SELECT id, price FROM " + + COLUMN_LIST_SINK_TABLE))); + + executeSql( + COLUMN_LIST_CONTAINER, + "INSERT INTO " + + COLUMN_LIST_SOURCE_TABLE + + " (id, price, note) VALUES (3, 3.00, 'third')"); + + given().ignoreExceptions() + .await() + .atMost(120, TimeUnit.SECONDS) + .untilAsserted( + () -> + Assertions.assertIterableEquals( + query( + COLUMN_LIST_CONTAINER, + "SELECT id, price FROM " + + COLUMN_LIST_SOURCE_TABLE), + query( + COLUMN_LIST_CONTAINER, + "SELECT id, price FROM " + + COLUMN_LIST_SINK_TABLE))); + Assertions.assertTrue( + query( + COLUMN_LIST_CONTAINER, + "SELECT column_name FROM information_schema.columns" + + " WHERE table_name = 'sink_column_orders'" + + " AND column_name = 'note'") + .isEmpty()); + } + + @TestTemplate + @DisabledOnContainer( + value = {}, + type = {EngineType.SPARK, EngineType.FLINK}, + disabledReason = "This case requires Zeta job cancellation and schema evolution") + public void testExistingTablePublicationOnSecondRun(TestContainer container) throws Exception { + assertExistingPublicationOnSecondRun( + container, "table_pub", "FOR TABLE public.table_pub_orders"); + } + + @TestTemplate + @DisabledOnContainer( + value = {}, + type = {EngineType.SPARK, EngineType.FLINK}, + disabledReason = "This case requires Zeta job cancellation and schema evolution") + public void testAllTablesPublicationOnSecondRun(TestContainer container) throws Exception { + assertExistingPublicationOnSecondRun(container, "all_pub", "FOR ALL TABLES"); + } + + @TestTemplate + @DisabledOnContainer( + value = {}, + type = {EngineType.SPARK, EngineType.FLINK}, + disabledReason = "This case requires Zeta job cancellation and schema evolution") + public void testSchemaPublicationOnSecondRun(TestContainer container) throws Exception { + assertExistingPublicationOnSecondRun( + container, "schema_pub", "FOR TABLES IN SCHEMA public"); + } + + @TestTemplate + @DisabledOnContainer( + value = {}, + type = {EngineType.SPARK, EngineType.FLINK}, + disabledReason = "This case verifies Zeta source initialization failure") + public void testInsertOnlyPublicationWithoutPrimaryKeyFailsFast(TestContainer container) + throws Exception { + String name = "insert_only"; + executeSql( + COLUMN_LIST_CONTAINER, + "CREATE TABLE public.insert_only_orders (id INT PRIMARY KEY, price NUMERIC(10, 2))"); + executeSql(COLUMN_LIST_CONTAINER, "INSERT INTO public.insert_only_orders VALUES (1, 1.00)"); + executeSql( + COLUMN_LIST_CONTAINER, + "CREATE PUBLICATION seatunnel_insert_only FOR TABLE public.insert_only_orders" + + " (price) WITH (publish = 'insert')"); + String jobId = String.valueOf(JobIdGenerator.newJobId()); + CompletableFuture<Container.ExecResult> job = + startPublicationJob(container, name, jobId, 0); + try { + Container.ExecResult result = job.get(120, TimeUnit.SECONDS); + Assertions.assertNotEquals(0, result.getExitCode()); + String output = result.getStdout() + result.getStderr() + container.getServerLogs(); + Assertions.assertTrue( + output.contains("does not publish required primary key columns [id]")); + Assertions.assertTrue(output.contains("seatunnel_insert_only")); + Assertions.assertTrue(output.contains("public.insert_only_orders")); + Assertions.assertTrue(output.contains("Include all required primary key columns")); + Assertions.assertTrue( + query( + COLUMN_LIST_CONTAINER, + "SELECT table_name FROM information_schema.tables" + + " WHERE table_schema = 'public' AND table_name = 'sink_insert_only_orders'") + .isEmpty()); + } finally { + if (!job.isDone()) { + container.cancelJob(jobId); + job.get(120, TimeUnit.SECONDS); + } + } + } + + private void assertExistingPublicationOnSecondRun( + TestContainer container, String name, String publicationTables) throws Exception { + String source = "public." + name + "_orders"; + String sink = "public.sink_" + name + "_orders"; + executeSql( + COLUMN_LIST_CONTAINER, + "CREATE TABLE " + + source + + " (id INT PRIMARY KEY, price NUMERIC(10, 2)," + + " total NUMERIC(10, 2) GENERATED ALWAYS AS (price * 2) STORED)"); + executeSql(COLUMN_LIST_CONTAINER, "ALTER TABLE " + source + " REPLICA IDENTITY FULL"); + executeSql( + COLUMN_LIST_CONTAINER, + "INSERT INTO " + source + " (id, price) VALUES (1, 1.00), (2, 2.00)"); + executeSql( + COLUMN_LIST_CONTAINER, + "CREATE PUBLICATION seatunnel_" + name + " " + publicationTables); + + // Both starts see the same pre-created publication; use a fresh slot/sink to check + // snapshots. + for (int run = 0; run < 2; run++) { + String jobId = String.valueOf(JobIdGenerator.newJobId()); + CompletableFuture<Container.ExecResult> job = + startPublicationJob(container, name, jobId, run); + try { + awaitPublicationRows(job, source, sink); + Assertions.assertTrue( + query( + COLUMN_LIST_CONTAINER, + "SELECT column_name FROM information_schema.columns WHERE table_schema = 'public'" + + " AND table_name = 'sink_" + + name + + "_orders' AND column_name = 'total'") + .isEmpty()); + executeSql( + COLUMN_LIST_CONTAINER, + "UPDATE " + source + " SET price = " + (7 + run) + ".00 WHERE id = 1"); + executeSql( + COLUMN_LIST_CONTAINER, + "INSERT INTO " + source + " (id, price) VALUES (" + (3 + run) + ", 3.00)"); + awaitPublicationRows(job, source, sink); + } finally { + Container.ExecResult cancelled = container.cancelJob(jobId); + Assertions.assertEquals(0, cancelled.getExitCode(), cancelled.getStderr()); + job.get(120, TimeUnit.SECONDS); + } + executeSql(COLUMN_LIST_CONTAINER, "DROP TABLE " + sink); + } + } + + private CompletableFuture<Container.ExecResult> startPublicationJob( + TestContainer container, String name, String jobId, int run) { + return CompletableFuture.supplyAsync( + () -> { + try { + return container.executeJob( + "/postgrescdc_to_postgres_with_existing_publication.conf", + jobId, + "publication_name=seatunnel_" + name, + "source_table=" + name + "_orders", + "sink_table=sink_" + name + "_orders", + "slot_name=seatunnel_" + name + "_" + run); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + } + + private void awaitPublicationRows( + CompletableFuture<Container.ExecResult> job, String source, String sink) { + given().ignoreExceptions() + .await() + .atMost(120, TimeUnit.SECONDS) + .untilAsserted( + () -> { + Assertions.assertFalse( + job.isDone(), + "Publication job stopped before snapshot/stream completed"); + Assertions.assertIterableEquals( + query(COLUMN_LIST_CONTAINER, "SELECT id, price FROM " + source), + query(COLUMN_LIST_CONTAINER, "SELECT id, price FROM " + sink)); + }); + } + + private Connection getJdbcConnection(PostgreSQLContainer<?> postgres) throws SQLException { + return DriverManager.getConnection( + postgres.getJdbcUrl(), postgres.getUsername(), postgres.getPassword()); + } + + private void executeSql(String sql) { + executeSql(POSTGRES_CONTAINER, sql); + } + + private void executeSql(PostgreSQLContainer<?> postgres, String sql) { + try (Connection connection = getJdbcConnection(postgres); + Statement statement = connection.createStatement()) { + statement.execute(sql); + } catch (SQLException e) { + throw new RuntimeException(e); + } + } + + private List<List<Object>> query(String sql) { + return query(POSTGRES_CONTAINER, sql); + } + + private List<List<Object>> query(PostgreSQLContainer<?> postgres, String sql) { + try (Connection connection = getJdbcConnection(postgres); + Statement statement = connection.createStatement(); + ResultSet resultSet = statement.executeQuery(sql + " ORDER BY 1")) { + List<List<Object>> result = new ArrayList<>(); + int columnCount = resultSet.getMetaData().getColumnCount(); + while (resultSet.next()) { + List<Object> row = new ArrayList<>(); + for (int i = 1; i <= columnCount; i++) { + row.add(resultSet.getObject(i)); + } + result.add(row); + } + return result; + } catch (SQLException e) { + throw new RuntimeException(e); + } + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/resources/postgrescdc_to_postgres_with_column_list.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/resources/postgrescdc_to_postgres_with_column_list.conf new file mode 100644 index 0000000000..b5a5d502d6 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/resources/postgrescdc_to_postgres_with_column_list.conf @@ -0,0 +1,59 @@ +# +# 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. +# + +env { + parallelism = 1 + job.mode = "STREAMING" + checkpoint.interval = 5000 +} + +source { + Postgres-CDC { + plugin_output = "column_list_cdc" + username = "postgres" + password = "postgres" + database-names = ["generated_cdc"] + schema-names = ["public"] + table-names = ["generated_cdc.public.column_orders"] + url = "jdbc:postgresql://postgres_column_list_e2e:5432/generated_cdc?loggerLevel=OFF" + # The effective plugin comes from the Debezium override below. + decoding.plugin.name = "decoderbufs" + require-replica-identity-full = false + slot.name = "seatunnel_column_list" + # Column-list filtering does not require schema evolution. + schema-changes.enabled = false + debezium { + "plugin.name" = "pgoutput" + "publication.name" = "seatunnel_column_list" + "publication.autocreate.mode" = "disabled" + } + } +} + +sink { + jdbc { + plugin_input = "column_list_cdc" + url = "jdbc:postgresql://postgres_column_list_e2e:5432/generated_cdc?loggerLevel=OFF" + driver = "org.postgresql.Driver" + user = "postgres" + password = "postgres" + generate_sink_sql = true + database = generated_cdc + table = public.sink_column_orders + primary_keys = ["id"] + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/resources/postgrescdc_to_postgres_with_existing_publication.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/resources/postgrescdc_to_postgres_with_existing_publication.conf new file mode 100644 index 0000000000..fcc7d7329f --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/resources/postgrescdc_to_postgres_with_existing_publication.conf @@ -0,0 +1,56 @@ +# +# 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. +# + +env { + parallelism = 1 + job.mode = "STREAMING" + checkpoint.interval = 5000 +} + +source { + Postgres-CDC { + plugin_output = "publication_cdc" + username = "postgres" + password = "postgres" + database-names = ["generated_cdc"] + schema-names = ["public"] + table-names = ["generated_cdc.public.${source_table}"] + url = "jdbc:postgresql://postgres_column_list_e2e:5432/generated_cdc?loggerLevel=OFF" + decoding.plugin.name = "pgoutput" + require-replica-identity-full = false + slot.name = "${slot_name}" + schema-changes.enabled = true + debezium { + "publication.name" = "${publication_name}" + "publication.autocreate.mode" = "disabled" + } + } +} + +sink { + jdbc { + plugin_input = "publication_cdc" + url = "jdbc:postgresql://postgres_column_list_e2e:5432/generated_cdc?loggerLevel=OFF" + driver = "org.postgresql.Driver" + user = "postgres" + password = "postgres" + generate_sink_sql = true + database = generated_cdc + table = "public.${sink_table}" + primary_keys = ["id"] + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/resources/postgrescdc_to_postgres_with_generated_column.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/resources/postgrescdc_to_postgres_with_generated_column.conf new file mode 100644 index 0000000000..3f1b0e4f93 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/resources/postgrescdc_to_postgres_with_generated_column.conf @@ -0,0 +1,51 @@ +# +# 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. +# + +env { + parallelism = 1 + job.mode = "STREAMING" + checkpoint.interval = 5000 +} + +source { + Postgres-CDC { + plugin_output = "generated_column_cdc" + username = "postgres" + password = "postgres" + database-names = ["generated_cdc"] + schema-names = ["public"] + table-names = ["generated_cdc.public.orders"] + url = "jdbc:postgresql://postgres_generated_column_e2e:5432/generated_cdc?loggerLevel=OFF" + decoding.plugin.name = "pgoutput" + slot.name = "seatunnel_generated_column" + schema-changes.enabled = true + } +} + +sink { + jdbc { + plugin_input = "generated_column_cdc" + url = "jdbc:postgresql://postgres_generated_column_e2e:5432/generated_cdc?loggerLevel=OFF" + driver = "org.postgresql.Driver" + user = "postgres" + password = "postgres" + generate_sink_sql = true + database = generated_cdc + table = public.sink_orders + primary_keys = ["id"] + } +}
