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 7067a09840 [Bug][Connector-V2] Ensure generated MySQL primary keys are
not null (#12425)
7067a09840 is described below
commit 7067a098408ec67c4d8761d8b2d05e1555daccce
Author: hutiefang76 <[email protected]>
AuthorDate: Thu Sep 24 02:02:26 2026 +0000
[Bug][Connector-V2] Ensure generated MySQL primary keys are not null
(#12425)
---
.../catalog/mysql/MysqlCreateTableSqlBuilder.java | 10 ++++--
.../OceanBaseMysqlCreateTableSqlBuilder.java | 11 ++++--
.../mysql/MysqlCreateTableSqlBuilderTest.java | 37 ++++++++++++++++++++
.../OceanBaseMysqlCreateTableSqlBuilderTest.java | 40 ++++++++++++++++++++++
4 files changed, 94 insertions(+), 4 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/mysql/MysqlCreateTableSqlBuilder.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/mysql/MysqlCreateTableSqlBuilder.java
index f1ee8211f2..f04a8861f6 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/mysql/MysqlCreateTableSqlBuilder.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/mysql/MysqlCreateTableSqlBuilder.java
@@ -201,8 +201,8 @@ public class MysqlCreateTableSqlBuilder {
}
columnSqls.add(type);
columnTypeMap.put(column.getName(), type);
- // nullable
- if (column.isNullable()) {
+ // Primary key columns must be NOT NULL for MySQL to accept the
generated DDL.
+ if (column.isNullable() && !isPrimaryKeyColumn(column)) {
columnSqls.add("NULL");
} else {
columnSqls.add("NOT NULL");
@@ -215,6 +215,12 @@ public class MysqlCreateTableSqlBuilder {
return String.join(" ", columnSqls);
}
+ private boolean isPrimaryKeyColumn(Column column) {
+ return createIndex
+ && primaryKey != null
+ && primaryKey.getColumnNames().contains(column.getName());
+ }
+
private String escapeComment(String comment) {
return comment.replace("'", "''").replace("\\", "\\\\");
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/oceanbase/OceanBaseMysqlCreateTableSqlBuilder.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/oceanbase/OceanBaseMysqlCreateTableSqlBuilder.java
index db80ccb09c..a87ed9d6a9 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/oceanbase/OceanBaseMysqlCreateTableSqlBuilder.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/oceanbase/OceanBaseMysqlCreateTableSqlBuilder.java
@@ -202,8 +202,9 @@ public class OceanBaseMysqlCreateTableSqlBuilder {
}
columnSqls.add(type);
columnTypeMap.put(column.getName(), type);
- // nullable
- if (column.isNullable()) {
+ // Primary key columns must be NOT NULL for OceanBase MySQL mode to
accept the generated
+ // DDL.
+ if (column.isNullable() && !isPrimaryKeyColumn(column)) {
columnSqls.add("NULL");
} else {
columnSqls.add("NOT NULL");
@@ -219,6 +220,12 @@ public class OceanBaseMysqlCreateTableSqlBuilder {
return String.join(" ", columnSqls);
}
+ private boolean isPrimaryKeyColumn(Column column) {
+ return createIndex
+ && primaryKey != null
+ && primaryKey.getColumnNames().contains(column.getName());
+ }
+
private String buildPrimaryKeySql() {
String key =
primaryKey.getColumnNames().stream()
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/mysql/MysqlCreateTableSqlBuilderTest.java
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/mysql/MysqlCreateTableSqlBuilderTest.java
index 21740a282f..d45a174cf7 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/mysql/MysqlCreateTableSqlBuilderTest.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/mysql/MysqlCreateTableSqlBuilderTest.java
@@ -182,6 +182,43 @@ public class MysqlCreateTableSqlBuilderTest {
Assertions.assertTrue(createTableSql.contains("COLLATE =
utf8mb4_unicode_ci"));
}
+ @Test
+ public void testNullablePrimaryKeyColumnIsNotNullWhenCreateIndex() {
+ TablePath tablePath = TablePath.of("test_db", "test_table");
+ TableSchema tableSchema =
+ TableSchema.builder()
+ .column(PhysicalColumn.of("id", BasicType.INT_TYPE, 0,
true, null, null))
+ .column(
+ PhysicalColumn.of(
+ "description", BasicType.STRING_TYPE,
32, true, null, null))
+ .primaryKey(PrimaryKey.of("id",
Lists.newArrayList("id")))
+ .build();
+ CatalogTable catalogTable =
+ CatalogTable.of(
+ TableIdentifier.of("test_catalog", "test_db",
"test_table"),
+ tableSchema,
+ Collections.emptyMap(),
+ Collections.emptyList(),
+ "test table");
+
+ String createTableSql =
+ MysqlCreateTableSqlBuilder.builder(
+ tablePath, catalogTable,
MySqlTypeConverter.DEFAULT_INSTANCE, true)
+ .build(DatabaseIdentifier.MYSQL);
+
+ Assertions.assertTrue(createTableSql.contains("`id` INT NOT NULL"));
+ Assertions.assertTrue(createTableSql.contains("`description`
VARCHAR(32) NULL"));
+ Assertions.assertTrue(createTableSql.contains("PRIMARY KEY (`id`)"));
+
+ String createTableSqlWithoutIndex =
+ MysqlCreateTableSqlBuilder.builder(
+ tablePath, catalogTable,
MySqlTypeConverter.DEFAULT_INSTANCE, false)
+ .build(DatabaseIdentifier.MYSQL);
+
+ Assertions.assertTrue(createTableSqlWithoutIndex.contains("`id` INT
NULL"));
+ Assertions.assertFalse(createTableSqlWithoutIndex.contains("PRIMARY
KEY"));
+ }
+
@Test
public void testColumnSinkType() {
MysqlCreateTableSqlBuilder sqlBuilder =
mock(MysqlCreateTableSqlBuilder.class);
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/oceanbase/OceanBaseMysqlCreateTableSqlBuilderTest.java
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/oceanbase/OceanBaseMysqlCreateTableSqlBuilderTest.java
index 95e85035e2..ea7eb72249 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/oceanbase/OceanBaseMysqlCreateTableSqlBuilderTest.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/oceanbase/OceanBaseMysqlCreateTableSqlBuilderTest.java
@@ -68,4 +68,44 @@ public class OceanBaseMysqlCreateTableSqlBuilderTest {
Assertions.assertTrue(createTableSql.contains("DEFAULT CHARSET =
utf8mb4"));
Assertions.assertTrue(createTableSql.contains("COLLATE =
utf8mb4_unicode_ci"));
}
+
+ @Test
+ public void testNullablePrimaryKeyColumnIsNotNullWhenCreateIndex() {
+ TablePath tablePath = TablePath.of("test_db", "test_table");
+ TableSchema tableSchema =
+ TableSchema.builder()
+ .column(PhysicalColumn.of("id", BasicType.INT_TYPE, 0,
true, null, null))
+ .column(
+ PhysicalColumn.of(
+ "description", BasicType.STRING_TYPE,
32, true, null, null))
+ .primaryKey(PrimaryKey.of("id",
Lists.newArrayList("id")))
+ .build();
+ CatalogTable catalogTable =
+ CatalogTable.of(
+ TableIdentifier.of("test_catalog", "test_db",
"test_table"),
+ tableSchema,
+ Collections.emptyMap(),
+ Collections.emptyList(),
+ "test table");
+
+ String createTableSql =
+ OceanBaseMysqlCreateTableSqlBuilder.builder(
+ tablePath, catalogTable,
OceanBaseMySqlTypeConverter.INSTANCE, true)
+ .build(DatabaseIdentifier.OCEANBASE);
+
+ Assertions.assertTrue(createTableSql.contains("`id` INT NOT NULL"));
+ Assertions.assertTrue(createTableSql.contains("`description`
VARCHAR(32) NULL"));
+ Assertions.assertTrue(createTableSql.contains("PRIMARY KEY (`id`)"));
+
+ String createTableSqlWithoutIndex =
+ OceanBaseMysqlCreateTableSqlBuilder.builder(
+ tablePath,
+ catalogTable,
+ OceanBaseMySqlTypeConverter.INSTANCE,
+ false)
+ .build(DatabaseIdentifier.OCEANBASE);
+
+ Assertions.assertTrue(createTableSqlWithoutIndex.contains("`id` INT
NULL"));
+ Assertions.assertFalse(createTableSqlWithoutIndex.contains("PRIMARY
KEY"));
+ }
}
