This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 7b686d8387 [Fix][Connector-V2] Truncate the table path database
(#12877)
7b686d8387 is described below
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) {
