[pull] dev from apache:dev

.github/workflows/backend.yml

@@ -449,7 +449,7 @@ jobs:
       matrix:
         java: [ '8', '11' ]
         os: [ 'ubuntu-latest' ]
-    timeout-minutes: 120
+    timeout-minutes: 180
     steps:
       - uses: actions/checkout@v2
       - name: Set up JDK ${{ matrix.java }}
@@ -510,7 +510,7 @@ jobs:
       matrix:
         java: [ '8', '11' ]
         os: [ 'ubuntu-latest' ]
-    timeout-minutes: 120
+    timeout-minutes: 180
     steps:
       - uses: actions/checkout@v2
       - name: Set up JDK ${{ matrix.java }}
@@ -540,7 +540,7 @@ jobs:
       matrix:
         java: [ '8', '11' ]
         os: [ 'ubuntu-latest' ]
-    timeout-minutes: 120
+    timeout-minutes: 180
     steps:
       - uses: actions/checkout@v2
       - name: Set up JDK ${{ matrix.java }}

seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/util/ClickhouseCatalogUtil.java

@@ -21,15 +21,43 @@
 
 import org.apache.seatunnel.api.table.catalog.Column;
 import org.apache.seatunnel.api.table.catalog.TablePath;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
 import org.apache.seatunnel.connectors.seatunnel.clickhouse.catalog.ClickhouseTypeConverter;
 import org.apache.seatunnel.connectors.seatunnel.common.util.CatalogUtil;
 
+import java.util.HashSet;
+import java.util.Set;
+
 import static org.apache.seatunnel.shade.com.google.common.base.Preconditions.checkNotNull;
 
 public class ClickhouseCatalogUtil extends CatalogUtil {
 
+    private static final ThreadLocal<Set<String>> PRIMARY_KEY_COLUMNS =
+            ThreadLocal.withInitial(HashSet::new);
+
     public static final ClickhouseCatalogUtil INSTANCE = new ClickhouseCatalogUtil();
 
+    @Override
+    public String getCreateTableSql(
+            String template,
+            String database,
+            String table,
+            TableSchema tableSchema,
+            String comment,
+            String optionsKey) {
+        Set<String> pkColumns = PRIMARY_KEY_COLUMNS.get();
+        pkColumns.clear();
+        if (tableSchema.getPrimaryKey() != null) {
+            pkColumns.addAll(tableSchema.getPrimaryKey().getColumnNames());
+        }
+        try {
+            return super.getCreateTableSql(
+                    template, database, table, tableSchema, comment, optionsKey);
+        } finally {
+            pkColumns.clear();
+        }
+    }
+
     public String columnToConnectorType(Column column) {
         checkNotNull(column, "The column is required.");
         String columnType;
@@ -38,6 +66,14 @@ public String columnToConnectorType(Column column) {
         } else {
             columnType = ClickhouseTypeConverter.INSTANCE.reconvert(column).getColumnType();
         }
+
+        Set<String> pkColumns = PRIMARY_KEY_COLUMNS.get();
+        boolean isPrimaryKeyColumn = pkColumns != null && pkColumns.contains(column.getName());
+
+        if (column.isNullable() && !isUnsupportedNullableType(columnType) && !isPrimaryKeyColumn) {
+            columnType = "Nullable(" + columnType + ")";
+        }
+
         return String.format(
                 "`%s` %s %s",
                 column.getName(),
@@ -49,6 +85,10 @@ public String columnToConnectorType(Column column) {
                                 + "'");
     }
 
+    private static boolean isUnsupportedNullableType(String columnType) {
+        return columnType.startsWith("Map(") || columnType.startsWith("Array(");
+    }
+
     public String getDropTableSql(TablePath tablePath, boolean ignoreIfNotExists) {
         if (ignoreIfNotExists) {
             return "DROP TABLE IF EXISTS "

seatunnel-connectors-v2/connector-clickhouse/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/ClickhouseCreateTableTest.java

@@ -99,14 +99,16 @@ public void test() {
                                 .build(),
                         "clickhouse test table",
                         ClickhouseSinkOptions.SAVE_MODE_CREATE_TEMPLATE.key());
+        // Primary key columns (id, age) should NOT be wrapped in Nullable
+        // because ClickHouse does not allow nullable columns in ORDER BY / PRIMARY KEY
         Assertions.assertEquals(
                 createTableSql,
                 "CREATE TABLE IF NOT EXISTS  `test1`.`test2` (\n"
                         + "    `id` Int64 ,`age` Int32 COMMENT 'test comment',\n"
-                        + "    `name` String ,\n"
-                        + "`score` Int32 COMMENT '''N''-N',\n"
-                        + "`gender` Int8 ,\n"
-                        + "`create_time` Int64 \n"
+                        + "    `name` Nullable(String) ,\n"
+                        + "`score` Nullable(Int32) COMMENT '\''N''-N',\n"
+                        + "`gender` Nullable(Int8) ,\n"
+                        + "`create_time` Nullable(Int64) \n"
                         + ") ENGINE = MergeTree()\n"
                         + "ORDER BY (`id`,`age`)\n"
                         + "PRIMARY KEY (`id`,`age`)\n"

seatunnel-connectors-v2/connector-clickhouse/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/util/ClickhouseCatalogUtilTest.java

@@ -24,6 +24,7 @@
 import org.junit.jupiter.api.Test;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
@@ -33,6 +34,8 @@ void returnsReconvertedTypeWhenSinkTypeNotNull() {
         Column column = mock(Column.class);
         when(column.getName()).thenReturn("col1");
         when(column.getSinkType()).thenReturn("String");
+        when(column.isNullable()).thenReturn(false);
+        when(column.getComment()).thenReturn("");
 
         String result = ClickhouseCatalogUtil.INSTANCE.columnToConnectorType(column);
 
@@ -44,6 +47,8 @@ void returnsReconvertedTypeWhenSinkTypeIsNull() {
         Column column = mock(Column.class);
         when(column.getName()).thenReturn("col1");
         when(column.getDataType()).thenReturn((SeaTunnelDataType) BasicType.INT_TYPE);
+        when(column.isNullable()).thenReturn(false);
+        when(column.getComment()).thenReturn("");
 
         String result = ClickhouseCatalogUtil.INSTANCE.columnToConnectorType(column);
 
@@ -56,9 +61,44 @@ void returnsReconvertedTypeWhenTypesNotNull() {
         when(column.getName()).thenReturn("col1");
         when(column.getDataType()).thenReturn((SeaTunnelDataType) BasicType.INT_TYPE);
         when(column.getSinkType()).thenReturn("String");
+        when(column.isNullable()).thenReturn(false);
+        when(column.getComment()).thenReturn("");
 
         String result = ClickhouseCatalogUtil.INSTANCE.columnToConnectorType(column);
 
         assertEquals("`col1` String ", result);
     }
+
+    @Test
+    void wrapsTypeWithNullableWhenColumnIsNullable() {
+        Column column = mock(Column.class);
+        when(column.getName()).thenReturn("col1");
+        when(column.getSinkType()).thenReturn("String");
+        when(column.isNullable()).thenReturn(true);
+        when(column.getComment()).thenReturn("");
+
+        String result = ClickhouseCatalogUtil.INSTANCE.columnToConnectorType(column);
+
+        assertEquals("`col1` Nullable(String) ", result);
+    }
+
+    @Test
+    void escapesSingleQuoteAndBackslashInComment() {
+        Column column = mock(Column.class);
+        when(column.getName()).thenReturn("col1");
+        when(column.getSinkType()).thenReturn("String");
+        when(column.isNullable()).thenReturn(false);
+        when(column.getComment()).thenReturn("O'Reilly \\ path");
+
+        String result = ClickhouseCatalogUtil.INSTANCE.columnToConnectorType(column);
+
+        assertEquals("`col1` String COMMENT 'O''Reilly \\\\ path'", result);
+    }
+
+    @Test
+    void throwsExceptionWhenColumnIsNull() {
+        assertThrows(
+                NullPointerException.class,
+                () -> ClickhouseCatalogUtil.INSTANCE.columnToConnectorType(null));
+    }
 }

seatunnel-connectors-v2/connector-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/util/CatalogUtil.java

@@ -32,6 +32,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.function.Function;
+import java.util.regex.Matcher;
 import java.util.stream.Collectors;
 
 @Slf4j
@@ -70,7 +71,7 @@ public String getCreateTableSql(
         template =
                 template.replaceAll(
                         SaveModePlaceHolder.ROWTYPE_PRIMARY_KEY.getReplacePlaceHolder(),
-                        primaryKey);
+                        Matcher.quoteReplacement(primaryKey));
         SqlTemplate.canHandledByTemplateWithPlaceholder(
                 template,
                 SaveModePlaceHolder.ROWTYPE_UNIQUE_KEY.getPlaceHolder(),
@@ -80,7 +81,8 @@ public String getCreateTableSql(
 
         template =
                 template.replaceAll(
-                        SaveModePlaceHolder.ROWTYPE_UNIQUE_KEY.getReplacePlaceHolder(), uniqueKey);
+                        SaveModePlaceHolder.ROWTYPE_UNIQUE_KEY.getReplacePlaceHolder(),
+                        Matcher.quoteReplacement(uniqueKey));
         Map<String, CreateTableParser.ColumnInfo> columnInTemplate =
                 CreateTableParser.getColumnList(template);
         template = mergeColumnInTemplate(columnInTemplate, tableSchema, template);
@@ -95,20 +97,27 @@ public String getCreateTableSql(
             // TODO: Remove this compatibility config
             template =
                     template.replaceAll(
-                            SaveModePlaceHolder.TABLE_NAME.getReplacePlaceHolder(), table);
+                            SaveModePlaceHolder.TABLE_NAME.getReplacePlaceHolder(),
+                            Matcher.quoteReplacement(table));
             log.warn(
                     "The variable placeholder `${table_name}` has been marked as deprecated and will be removed soon, please use `${table}`");
         }
 
-        return template.replaceAll(SaveModePlaceHolder.DATABASE.getReplacePlaceHolder(), database)
-                .replaceAll(SaveModePlaceHolder.TABLE.getReplacePlaceHolder(), table)
+        return template.replaceAll(
+                        SaveModePlaceHolder.DATABASE.getReplacePlaceHolder(),
+                        Matcher.quoteReplacement(database))
                 .replaceAll(
-                        SaveModePlaceHolder.ROWTYPE_FIELDS.getReplacePlaceHolder(), rowTypeFields)
+                        SaveModePlaceHolder.TABLE.getReplacePlaceHolder(),
+                        Matcher.quoteReplacement(table))
+                .replaceAll(
+                        SaveModePlaceHolder.ROWTYPE_FIELDS.getReplacePlaceHolder(),
+                        Matcher.quoteReplacement(rowTypeFields))
                 .replaceAll(
                         SaveModePlaceHolder.COMMENT.getReplacePlaceHolder(),
-                        Objects.isNull(comment)
-                                ? ""
-                                : comment.replace("'", "''").replace("\\", "\\\\"));
+                        Matcher.quoteReplacement(
+                                Objects.isNull(comment)
+                                        ? ""
+                                        : comment.replace("'", "''").replace("\\", "\\\\")));
     }
 
     private String mergeColumnInTemplate(

seatunnel-connectors-v2/connector-hbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/hbase/catalog/HbaseCatalog.java

@@ -104,7 +104,13 @@ public List<String> listTables(String databaseName)
     @Override
     public boolean tableExists(TablePath tablePath) throws CatalogException {
         checkNotNull(tablePath);
-        return hbaseClient.tableExists(tablePath.getTableName());
+        String databaseName = tablePath.getDatabaseName();
+        String tableName = tablePath.getTableName();
+        String fullTableName =
+                (databaseName == null || databaseName.isEmpty())
+                        ? tableName
+                        : databaseName + ":" + tableName;
+        return hbaseClient.tableExists(fullTableName);
     }
 
     @Override

seatunnel-connectors-v2/connector-hbase/src/test/java/org/apache/seatunnel/connectors/seatunnel/hbase/HbaseCatalogTest.java

@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.hbase;
+
+import org.apache.seatunnel.api.table.catalog.TablePath;
+import org.apache.seatunnel.connectors.seatunnel.hbase.catalog.HbaseCatalog;
+import org.apache.seatunnel.connectors.seatunnel.hbase.client.HbaseClient;
+import org.apache.seatunnel.connectors.seatunnel.hbase.config.HbaseParameters;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.lang.reflect.Field;
+
+public class HbaseCatalogTest {
+
+    @Test
+    public void testTableExistsWithNamespace() throws Exception {
+        HbaseParameters parameters =
+                HbaseParameters.builder()
+                        .zookeeperQuorum("localhost")
+                        .namespace("ns1")
+                        .table("tbl")
+                        .build();
+        HbaseCatalog catalog = new HbaseCatalog("hbase", "ns1", parameters);
+
+        HbaseClient hbaseClient = Mockito.mock(HbaseClient.class);
+        Mockito.when(hbaseClient.tableExists("ns1:tbl")).thenReturn(true);
+
+        injectHbaseClient(catalog, hbaseClient);
+
+        TablePath tablePath = TablePath.of("ns1", "tbl");
+        Assertions.assertTrue(catalog.tableExists(tablePath));
+        Mockito.verify(hbaseClient, Mockito.times(1)).tableExists("ns1:tbl");
+    }
+
+    @Test
+    public void testTableExistsWithoutNamespace() throws Exception {
+        HbaseParameters parameters =
+                HbaseParameters.builder()
+                        .zookeeperQuorum("localhost")
+                        .namespace("default")
+                        .table("tbl")
+                        .build();
+        HbaseCatalog catalog = new HbaseCatalog("hbase", "default", parameters);
+
+        HbaseClient hbaseClient = Mockito.mock(HbaseClient.class);
+        Mockito.when(hbaseClient.tableExists("tbl")).thenReturn(true);
+
+        injectHbaseClient(catalog, hbaseClient);
+
+        TablePath tablePath = TablePath.of("tbl");
+        Assertions.assertTrue(catalog.tableExists(tablePath));
+        Mockito.verify(hbaseClient, Mockito.times(1)).tableExists("tbl");
+    }
+
+    private void injectHbaseClient(HbaseCatalog catalog, HbaseClient hbaseClient) throws Exception {
+        Field clientField = HbaseCatalog.class.getDeclaredField("hbaseClient");
+        clientField.setAccessible(true);
+        clientField.set(catalog, hbaseClient);
+    }
+}