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-12877-cb3351ed124f68eeac6304ac4c1d328edafbed8f in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 7b686d8387557ae4d90de35176d6dcafdc0e1fff Author: hutiefang76 <[email protected]> AuthorDate: Mon Oct 26 17:10:24 2026 +0000 [Fix][Connector-V2] Truncate the table path database (#12877) --- docs/en/connectors/sink/Jdbc.md | 2 + docs/zh/connectors/sink/Jdbc.md | 2 + .../jdbc/catalog/AbstractJdbcCatalog.java | 15 ++- .../jdbc/catalog/AbstractJdbcCatalogTest.java | 94 ++++++++++++++++++ .../connectors/seatunnel/jdbc/JdbcPostgresIT.java | 105 ++++++++++++++++++++- 5 files changed, 214 insertions(+), 4 deletions(-) diff --git a/docs/en/connectors/sink/Jdbc.md b/docs/en/connectors/sink/Jdbc.md index b2ba4e00d1..f5a70fb80f 100644 --- a/docs/en/connectors/sink/Jdbc.md +++ b/docs/en/connectors/sink/Jdbc.md @@ -411,6 +411,8 @@ Option introduction： `CUSTOM_PROCESSING`：User defined processing `ERROR_WHEN_DATA_EXISTS`：When there is data, an error is reported +When the full target table path specifies a database (`database.schema.table`), `DROP_DATA` executes `TRUNCATE TABLE` in that database. + ### custom_sql [String] When data_save_mode selects CUSTOM_PROCESSING, you should fill in the CUSTOM_SQL parameter. This parameter usually fills in a SQL that can be executed. SQL will be executed before synchronization tasks. diff --git a/docs/zh/connectors/sink/Jdbc.md b/docs/zh/connectors/sink/Jdbc.md index 574f0d3138..8b85b2bf1a 100644 --- a/docs/zh/connectors/sink/Jdbc.md +++ b/docs/zh/connectors/sink/Jdbc.md @@ -407,6 +407,8 @@ Sink插件常用参数，请参考 [Sink常用选项](../common-options/sink-com `CUSTOM_PROCESSING`：允许用户自定义数据处理方式<br/> `ERROR_WHEN_DATA_EXISTS`：当有数据时抛出错误<br/> +目标表的完整路径指定数据库（`database.schema.table`）时，`DROP_DATA` 会连接到该数据库执行 `TRUNCATE TABLE`。 + ### table_options [Map] Sink 在自动建表（SaveMode DDL）时附加的表级选项。仅在 `schema_save_mode` 触发建表时生效，例如 `CREATE_SCHEMA_WHEN_NOT_EXIST`、`RECREATE_SCHEMA`；**不影响**数据写入阶段的 INSERT/UPSERT，也**不会**对已存在表执行 `ALTER TABLE`。 diff --git a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/AbstractJdbcCatalog.java b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/AbstractJdbcCatalog.java index 9a40d2123f..882473551c 100644 --- a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/AbstractJdbcCatalog.java +++ b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/AbstractJdbcCatalog.java @@ -688,9 +688,22 @@ public abstract class AbstractJdbcCatalog implements Catalog { return CatalogUtils.getCatalogTable(defaultConnection, sqlQuery); } + /** + * Truncates the table on the connection of the database that the table path points to. + * + * <p>Most catalogs render the truncate DDL with schema and table only (no database qualifier), + * so the connection URL is what decides which database is truncated. A table path that carries + * an explicit database different from the catalog database must therefore be truncated through + * that database's URL, like the other DDL siblings in this class. A table path without a + * database keeps using the catalog default URL. + */ protected void truncateTableInternal(TablePath tablePath) throws CatalogException { + String dbUrl = + StringUtils.isNotBlank(tablePath.getDatabaseName()) + ? getUrlFromDatabaseName(tablePath.getDatabaseName()) + : defaultUrl; try { - executeInternal(defaultUrl, getTruncateTableSql(tablePath)); + executeInternal(dbUrl, getTruncateTableSql(tablePath)); } catch (Exception e) { throw new CatalogException( String.format( diff --git a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/AbstractJdbcCatalogTest.java b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/AbstractJdbcCatalogTest.java index c1a1abf389..13d2a9f4d6 100644 --- a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/AbstractJdbcCatalogTest.java +++ b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/AbstractJdbcCatalogTest.java @@ -17,6 +17,7 @@ package org.apache.seatunnel.connectors.seatunnel.jdbc.catalog; +import org.apache.seatunnel.api.table.catalog.TablePath; import org.apache.seatunnel.api.table.catalog.exception.CatalogException; import org.apache.seatunnel.common.utils.JdbcUrlUtil; @@ -24,14 +25,29 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import java.sql.Connection; +import java.sql.PreparedStatement; import java.sql.SQLException; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.contains; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; class AbstractJdbcCatalogTest { + /** Database configured in the catalog URL. */ + private static final String CATALOG_DATABASE = "db_a"; + + private static final String CATALOG_DATABASE_URL = "jdbc:testdb://localhost:5432/db_a"; + + /** Database referenced by the table path, different from the catalog database. */ + private static final String TABLE_DATABASE = "db_b"; + + private static final String TABLE_DATABASE_URL = "jdbc:testdb://localhost:5432/db_b"; + @Test void shouldCloseRemainingConnectionsAndKeepSuppressedExceptionsWhenCloseFails() throws SQLException { @@ -51,6 +67,64 @@ class AbstractJdbcCatalogTest { Assertions.assertEquals(1, exception.getSuppressed().length); } + /** + * Control: a table inside the catalog database is truncated on the catalog default connection. + */ + @Test + void shouldTruncateTableOnCatalogDefaultConnection() throws SQLException { + TestJdbcCatalog catalog = new TestJdbcCatalog(CATALOG_DATABASE_URL); + Connection catalogConnection = registerMockConnection(catalog, CATALOG_DATABASE_URL); + Connection tableDatabaseConnection = registerMockConnection(catalog, TABLE_DATABASE_URL); + + catalog.truncateTable(TablePath.of(CATALOG_DATABASE, "public", "t"), false); + + verify(catalogConnection).prepareStatement(contains("TRUNCATE")); + verify(tableDatabaseConnection, never()).prepareStatement(anyString()); + } + + /** + * The truncate DDL only qualifies schema and table (same as {@code PostgresCatalog}), so the + * connection URL is the only thing that decides which database gets truncated. Truncating + * {@code db_b.public.t} through a catalog whose own database is {@code db_a} must therefore run + * on the {@code db_b} connection, not on the catalog default one. + */ + @Test + void shouldRouteTruncateToTableDatabaseWhenItDiffersFromCatalogDefault() throws SQLException { + TestJdbcCatalog catalog = new TestJdbcCatalog(CATALOG_DATABASE_URL); + Connection catalogConnection = registerMockConnection(catalog, CATALOG_DATABASE_URL); + Connection tableDatabaseConnection = registerMockConnection(catalog, TABLE_DATABASE_URL); + + catalog.truncateTable(TablePath.of(TABLE_DATABASE, "public", "t"), false); + + verify(tableDatabaseConnection).prepareStatement(contains("TRUNCATE")); + verify(catalogConnection, never()).prepareStatement(anyString()); + } + + /** + * Control: a table path without database has no database to route to and keeps using the + * catalog default connection. + */ + @Test + void shouldTruncateTableOnCatalogDefaultConnectionWhenTablePathHasNoDatabase() + throws SQLException { + TestJdbcCatalog catalog = new TestJdbcCatalog(CATALOG_DATABASE_URL); + Connection catalogConnection = registerMockConnection(catalog, CATALOG_DATABASE_URL); + + // truncateTableInternal() is invoked directly because tableExists() cannot probe a + // database-less table path; the existence check is not part of the routing contract here. + catalog.truncateTableInternal(new TablePath(null, "public", "t")); + + verify(catalogConnection).prepareStatement(contains("TRUNCATE")); + } + + private static Connection registerMockConnection(TestJdbcCatalog catalog, String url) + throws SQLException { + Connection connection = mock(Connection.class); + when(connection.prepareStatement(anyString())).thenReturn(mock(PreparedStatement.class)); + catalog.addConnection(url, connection); + return connection; + } + private static class TestJdbcCatalog extends AbstractJdbcCatalog { private TestJdbcCatalog() { @@ -64,8 +138,28 @@ class AbstractJdbcCatalogTest { null); } + private TestJdbcCatalog(String origin) { + super("test", "user", "password", JdbcUrlUtil.getUrlInfo(origin), null, null); + } + private void addConnection(String url, Connection connection) { connectionMap.put(url, connection); } + + /** The table is assumed to exist: existence checks are not part of the reproduction. */ + @Override + public boolean tableExists(TablePath tablePath) { + return true; + } + + /** Mirrors {@code PostgresCatalog}: schema and table only, no database qualifier. */ + @Override + protected String getTruncateTableSql(TablePath tablePath) { + return "TRUNCATE TABLE \"" + + tablePath.getSchemaName() + + "\".\"" + + tablePath.getTableName() + + "\""; + } } } diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-3/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcPostgresIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-3/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcPostgresIT.java index fcc3028878..a654f841e6 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-3/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcPostgresIT.java +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-3/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcPostgresIT.java @@ -269,6 +269,15 @@ public class JdbcPostgresIT extends TestSuiteBase implements TestResource { private static final String NUMERIC_SPECIAL_VALUE_SINK_QUERY = "select gid, numeric_col from " + NUMERIC_SPECIAL_VALUE_SINK_TABLE + " order by gid"; + /** + * Dedicated database created and dropped by the truncate routing test, so truncating a table + * path that points to another database never touches the container's default fixture tables. + */ + private static final String TRUNCATE_ROUTE_DATABASE = "pg_e2e_truncate_route_db"; + + /** Same table name in both the catalog database and {@link #TRUNCATE_ROUTE_DATABASE}. */ + private static final String TRUNCATE_ROUTE_TABLE = "pg_e2e_truncate_route_table"; + @TestContainerExtension private final ContainerExtendedFactory extendedFactory = container -> { @@ -643,6 +652,88 @@ public class JdbcPostgresIT extends TestSuiteBase implements TestResource { catalog.close(); } + /** + * Truncating {@code database.schema.table} through a catalog whose own database is the + * container default database must truncate the database of the table path, and must leave the + * table with the same name in the catalog database untouched. + * + * <p>Only the dedicated database and table created here are truncated and dropped. + */ + @Test + public void testTruncateTableUsesTablePathDatabase() throws SQLException { + String schema = "public"; + String catalogDatabaseName = POSTGRESQL_CONTAINER.getDatabaseName(); + TablePath catalogDatabaseTablePath = + TablePath.of(catalogDatabaseName, schema, TRUNCATE_ROUTE_TABLE); + TablePath routedDatabaseTablePath = + TablePath.of(TRUNCATE_ROUTE_DATABASE, schema, TRUNCATE_ROUTE_TABLE); + String createTableSql = + "CREATE TABLE " + schema + "." + TRUNCATE_ROUTE_TABLE + " (gid INT)"; + String insertRowSql = "INSERT INTO " + schema + "." + TRUNCATE_ROUTE_TABLE + " VALUES (1)"; + + PostgresCatalog postgresCatalog = + new PostgresCatalog( + DatabaseIdentifier.POSTGRESQL, + POSTGRESQL_CONTAINER.getUsername(), + POSTGRESQL_CONTAINER.getPassword(), + JdbcUrlUtil.getUrlInfo(POSTGRESQL_CONTAINER.getJdbcUrl()), + schema, + null); + postgresCatalog.open(); + try { + postgresCatalog.createDatabase(routedDatabaseTablePath, true); + postgresCatalog.executeSql(catalogDatabaseTablePath, createTableSql); + postgresCatalog.executeSql(routedDatabaseTablePath, createTableSql); + postgresCatalog.executeSql(catalogDatabaseTablePath, insertRowSql); + postgresCatalog.executeSql(routedDatabaseTablePath, insertRowSql); + Assertions.assertEquals(1, countRows(catalogDatabaseName, TRUNCATE_ROUTE_TABLE)); + Assertions.assertEquals(1, countRows(TRUNCATE_ROUTE_DATABASE, TRUNCATE_ROUTE_TABLE)); + + postgresCatalog.truncateTable(routedDatabaseTablePath, false); + + Assertions.assertEquals( + 1, + countRows(catalogDatabaseName, TRUNCATE_ROUTE_TABLE), + "The table in the catalog database must not be truncated"); + Assertions.assertEquals( + 0, + countRows(TRUNCATE_ROUTE_DATABASE, TRUNCATE_ROUTE_TABLE), + "The table in the database of the table path must be truncated"); + } finally { + try { + postgresCatalog.dropTable(routedDatabaseTablePath, true); + postgresCatalog.dropTable(catalogDatabaseTablePath, true); + } finally { + try { + postgresCatalog.dropDatabase(routedDatabaseTablePath, true); + } finally { + postgresCatalog.close(); + } + } + } + } + + /** + * Counts the rows of a table in the given database on the PostgreSQL container. + * + * @param databaseName database to connect to + * @param tableName table to count rows of, in the {@code public} schema + * @return the number of rows in the table + */ + private int countRows(String databaseName, String tableName) throws SQLException { + String baseJdbcUrl = POSTGRESQL_CONTAINER.getJdbcUrl(); + String databaseUrl = + baseJdbcUrl.replace( + "/" + POSTGRESQL_CONTAINER.getDatabaseName(), "/" + databaseName); + String countSql = "SELECT count(1) FROM public." + tableName; + try (Connection connection = getJdbcConnection(databaseUrl); + Statement statement = connection.createStatement(); + ResultSet resultSet = statement.executeQuery(countSql)) { + resultSet.next(); + return resultSet.getInt(1); + } + } + /** * Verifies PostgreSQL catalog discovery preserves visible column order after a column is * dropped. @@ -799,10 +890,18 @@ public class JdbcPostgresIT extends TestSuiteBase implements TestResource { } private Connection getJdbcConnection() throws SQLException { + return getJdbcConnection(POSTGRESQL_CONTAINER.getJdbcUrl()); + } + + /** + * Opens a connection to the given PostgreSQL URL on the container. + * + * @param jdbcUrl JDBC URL to connect to, including the database to use + * @return an open connection, owned by the caller + */ + private Connection getJdbcConnection(String jdbcUrl) throws SQLException { return DriverManager.getConnection( - POSTGRESQL_CONTAINER.getJdbcUrl(), - POSTGRESQL_CONTAINER.getUsername(), - POSTGRESQL_CONTAINER.getPassword()); + jdbcUrl, POSTGRESQL_CONTAINER.getUsername(), POSTGRESQL_CONTAINER.getPassword()); } private List<List<Object>> querySql(String sql) {
