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-12425-6fac4cc44182b4a998fc351c639c4b120592baf7 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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")); + } }
