(shardingsphere) branch master updated: Fix PostgreSQL pipeline DDL statement splitting (#40362)

[zh...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

terrymanu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere.git


The following commit(s) were added to refs/heads/master by this push:
     new fe313582daf Fix PostgreSQL pipeline DDL statement splitting (#40362)
fe313582daf is described below

commit fe313582daf370e62e6f18be65524f4febd9aaf9
Author: Liang Zhang <zh...@apache.org>
AuthorDate: Sat Oct 24 21:51:59 2026 +0800

    Fix PostgreSQL pipeline DDL statement splitting (#40362)
---
 .../sqlbuilder/PostgreSQLPipelineSQLBuilder.java   | 27 +++++++++++++++++++---
 .../PostgreSQLPipelineSQLBuilderTest.java          | 12 ++++++----
 2 files changed, 32 insertions(+), 7 deletions(-)

diff --git a/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/PostgreSQLPipelineSQLBuilder.java b/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/PostgreSQLPipelineSQLBuilder.java
index aa7eed4d35f..278810fe385 100644
--- a/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/PostgreSQLPipelineSQLBuilder.java
+++ b/kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/PostgreSQLPipelineSQLBuilder.java
@@ -17,6 +17,9 @@
 
 package org.apache.shardingsphere.data.pipeline.postgresql.sqlbuilder;
 
+import org.antlr.v4.runtime.CharStreams;
+import org.antlr.v4.runtime.Token;
+import org.antlr.v4.runtime.misc.Interval;
 import org.apache.shardingsphere.data.pipeline.core.ingest.record.Column;
 import org.apache.shardingsphere.data.pipeline.core.ingest.record.DataRecord;
 import org.apache.shardingsphere.data.pipeline.core.sqlbuilder.dialect.DialectPipelineSQLBuilder;
@@ -27,11 +30,12 @@ import org.apache.shardingsphere.data.pipeline.postgresql.sqlbuilder.ddl.index.P
 import org.apache.shardingsphere.data.pipeline.postgresql.sqlbuilder.ddl.table.PostgreSQLTablePropertiesLoader;
 import org.apache.shardingsphere.data.pipeline.postgresql.sqlbuilder.template.PostgreSQLPipelineFreemarkerManager;
 import org.apache.shardingsphere.database.connector.core.metadata.identifier.IdentifierScope;
+import org.apache.shardingsphere.sql.parser.engine.postgresql.parser.PostgreSQLLexer;
 
 import javax.sql.DataSource;
 import java.sql.Connection;
 import java.sql.SQLException;
-import java.util.Arrays;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Map;
 import java.util.Optional;
@@ -96,8 +100,7 @@ public final class PostgreSQLPipelineSQLBuilder implements DialectPipelineSQLBui
             Map<String, Object> materials = loadMaterials(tableName, schemaName, connection, majorVersion, minorVersion);
             String tableSQL = generateCreateTableSQL(majorVersion, minorVersion, materials);
             String indexSQL = generateCreateIndexSQL(connection, majorVersion, minorVersion, materials);
-            // TODO use ";" to split is not always correct if return value's comments contains ";"
-            return Arrays.asList((tableSQL + System.lineSeparator() + indexSQL).trim().split(";"));
+            return splitStatements(tableSQL + System.lineSeparator() + indexSQL);
         }
     }
     
@@ -117,6 +120,24 @@ public final class PostgreSQLPipelineSQLBuilder implements DialectPipelineSQLBui
         return new PostgreSQLIndexSQLGenerator(connection, majorVersion, minorVersion).generate(materials);
     }
     
+    private Collection<String> splitStatements(final String sql) {
+        Collection<String> result = new ArrayList<>();
+        PostgreSQLLexer lexer = new PostgreSQLLexer(CharStreams.fromString(sql));
+        int start = 0;
+        Token token;
+        do {
+            token = lexer.nextToken();
+            if (PostgreSQLLexer.SEMI_ == token.getType() || Token.EOF == token.getType()) {
+                String statement = lexer.getInputStream().getText(Interval.of(start, token.getStartIndex() - 1)).trim();
+                if (!statement.isEmpty()) {
+                    result.add(statement);
+                }
+                start = token.getStopIndex() + 1;
+            }
+        } while (Token.EOF != token.getType());
+        return result;
+    }
+    
     @SuppressWarnings("unchecked")
     private void formatColumns(final Map<String, Object> context) {
         Collection<Map<String, Object>> columns = (Collection<Map<String, Object>>) context.get("columns");
diff --git a/kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/PostgreSQLPipelineSQLBuilderTest.java b/kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/PostgreSQLPipelineSQLBuilderTest.java
index cc73dabaaa4..48e8c65e7c0 100644
--- a/kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/PostgreSQLPipelineSQLBuilderTest.java
+++ b/kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/sqlbuilder/PostgreSQLPipelineSQLBuilderTest.java
@@ -37,6 +37,8 @@ import org.apache.shardingsphere.infra.metadata.identifier.IdentifierCasePolicyR
 import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
 import org.apache.shardingsphere.test.infra.fixture.jdbc.MockedDataSource;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 import org.mockito.MockedConstruction;
 import org.mockito.MockedStatic;
 import org.postgresql.replication.LogSequenceNumber;
@@ -145,8 +147,9 @@ class PostgreSQLPipelineSQLBuilderTest {
         assertThat(sqlBuilder.wrapWithPageQuery("SELECT * FROM foo_tbl"), is("SELECT * FROM foo_tbl LIMIT ?"));
     }
     
-    @Test
-    void assertBuildCreateTableSQLs() throws SQLException {
+    @ParameterizedTest
+    @ValueSource(strings = {"'a;b'", "E'owner''s;comment'", "$$a;b$$", "'😀;value'"})
+    void assertBuildCreateTableSQLs(final String defaultValue) throws SQLException {
         Connection connection = mock(Connection.class, RETURNS_DEEP_STUBS);
         when(connection.getMetaData().getDatabaseMajorVersion()).thenReturn(10);
         when(connection.getMetaData().getDatabaseMinorVersion()).thenReturn(1);
@@ -169,11 +172,12 @@ class PostgreSQLPipelineSQLBuilderTest {
                 MockedConstruction<PostgreSQLConstraintsPropertiesAppender> ignoredConstraintsAppender = mockConstruction(PostgreSQLConstraintsPropertiesAppender.class);
                 MockedConstruction<PostgreSQLIndexSQLGenerator> ignoredIndexGenerator =
                         mockConstruction(PostgreSQLIndexSQLGenerator.class, (mocked, mockContext) -> when(mocked.generate(materials)).thenReturn("CREATE INDEX foo_index"))) {
-            freemarkerManager.when(() -> PostgreSQLPipelineFreemarkerManager.getSQLByVersion(materials, "component/table/%s/create.ftl", 10, 1)).thenReturn("CREATE TABLE foo;");
+            freemarkerManager.when(() -> PostgreSQLPipelineFreemarkerManager.getSQLByVersion(materials, "component/table/%s/create.ftl", 10, 1))
+                    .thenReturn("CREATE TABLE foo (note text DEFAULT " + defaultValue + ");");
             Collection<String> actual = sqlBuilder.buildCreateTableSQLs(dataSource, "public", "foo_tbl");
             List<String> actualList = actual.stream().map(String::trim).toList();
             assertThat(actualList.size(), is(2));
-            assertThat(actualList.get(0), is("CREATE TABLE foo"));
+            assertThat(actualList.get(0), is("CREATE TABLE foo (note text DEFAULT " + defaultValue + ")"));
             assertThat(actualList.get(1), is("CREATE INDEX foo_index"));
             assertThat(arrayColumn.get("cltype"), is("int4"));
             assertTrue((boolean) arrayColumn.get("hasSqrBracket"));