[FLINK-39252][sqlserver] Address review feedback for pipeline and source connectors

- handle null namespace in metadata accessor
- register chunk-key option and improve tables validation messages
- relax SQL Server database-name validation and escape bracket quoting
- fix FLOAT/REAL type mapping and clean up option/docs wording
- tighten tests for chunk-key support, malformed tables format, and quote escaping
- fail fast on SQL Server error 313 when CDC LSN window has been purged

Author: Daishuyuan

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/main/java/org/apache/flink/cdc/connectors/sqlserver/factory/SqlServerDataSourceFactory.java

@@ -50,7 +50,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.Optional;
 import java.util.Properties;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -77,8 +76,6 @@
 import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_STARTUP_TIMESTAMP_MILLIS;
 import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCHEMA_CHANGE_ENABLED;
 import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SERVER_TIME_ZONE;
-import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND;
-import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND;
 import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.TABLES;
 import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.TABLES_EXCLUDE;
 import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.USERNAME;
@@ -141,7 +138,7 @@ public DataSource createDataSource(Context context) {
 
         Map<String, String> configMap = config.toMap();
         mergeJdbcPropertiesIntoDebeziumProperties(configMap);
-        String databaseName = getValidateDatabaseName(tables).orElseThrow();
+        String databaseName = getValidateDatabaseName(tables);
 
         SqlServerSourceConfigFactory configFactory = new SqlServerSourceConfigFactory();
         configFactory
@@ -249,6 +246,7 @@ public Set<ConfigOption<?>> optionalOptions() {
         Set<ConfigOption<?>> options = new HashSet<>();
         options.add(PORT);
         options.add(TABLES_EXCLUDE);
+        options.add(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN);
         options.add(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE);
         options.add(SCAN_SNAPSHOT_FETCH_SIZE);
         options.add(SCHEMA_CHANGE_ENABLED);
@@ -341,7 +339,7 @@ private void validateDistributionFactorUpper(double distributionFactorUpper) {
                 doubleCompare(distributionFactorUpper, 1.0d) >= 0,
                 String.format(
                         "The value of option '%s' must larger than or equals %s, but is %s",
-                        SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND.key(),
+                        CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND.key(),
                         1.0d,
                         distributionFactorUpper));
     }
@@ -353,7 +351,7 @@ private void validateDistributionFactorLower(double distributionFactorLower) {
                         && doubleCompare(distributionFactorLower, 1.0d) <= 0,
                 String.format(
                         "The value of option '%s' must between %s and %s inclusively, but is %s",
-                        SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND.key(),
+                        CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND.key(),
                         0.0d,
                         1.0d,
                         distributionFactorLower));
@@ -365,41 +363,35 @@ && doubleCompare(distributionFactorLower, 1.0d) <= 0,
      * @param tables Table name list, format is "db.schema.table,db.schema.table,..." Each table
      *     name consists of three parts separated by ".", which are database name, schema name, and
      *     table name.
-     * @return Database name if found, otherwise returns Optional.empty()
      * @throws IllegalArgumentException If the input parameter is null or does not match the
      *     expected format, or if database names are inconsistent.
      */
-    private Optional<String> getValidateDatabaseName(String tables) {
-        // Input validation
+    private String getValidateDatabaseName(String tables) {
         if (tables == null || tables.trim().isEmpty()) {
             throw new IllegalArgumentException("Parameter tables cannot be null or empty");
         }
 
-        // Split table name list
         String[] tableNames = tables.split(",");
         String dbName = null;
 
         for (String tableName : tableNames) {
-            // Trim whitespace and split table name
             String trimmedTableName = tableName.trim();
-            if (!trimmedTableName.contains(".")) {
-                continue; // Skip table names that do not match the expected format
-            }
-
-            String[] tableNameParts =
-                    trimmedTableName.split(
-                            "(?<!\\\\)\\.", -1); // Use -1 to avoid ignoring trailing empty elements
+            String[] tableNameParts = trimmedTableName.split("(?<!\\\\)\\.", -1);
 
             checkState(
                     tableNameParts.length == 3,
                     String.format(
-                            "Tables format must db.schema.table, can not 'tables' = %s",
-                            TABLES.key()));
+                            "Table '%s' does not match the expected 'database.schema.table' "
+                                    + "format. Please check the value of option '%s'.",
+                            trimmedTableName, TABLES.key()));
             String currentDbName = tableNameParts[0];
 
             checkState(
                     isValidSqlServerDbName(currentDbName),
-                    String.format("%s is not a valid SQL Server database name", currentDbName));
+                    String.format(
+                            "Database name '%s' exceeds SQL Server's maximum identifier length "
+                                    + "of 128 characters.",
+                            currentDbName));
             if (dbName == null) {
                 dbName = currentDbName;
             } else {
@@ -411,20 +403,18 @@ private Optional<String> getValidateDatabaseName(String tables) {
             }
         }
 
-        // If no valid table name is found, return Optional.empty()
-        return Optional.ofNullable(dbName);
+        if (dbName == null) {
+            throw new IllegalArgumentException(
+                    "Cannot determine database name from 'tables' option: "
+                            + tables
+                            + ". Expected format: database.schema.table");
+        }
+        return dbName;
     }
 
     /** Validate if the database name conforms to SQL Server naming conventions. */
     private boolean isValidSqlServerDbName(String dbName) {
-        // SQL Server database name conventions:
-        // 1. Length does not exceed 128 characters
-        // 2. Can contain letters, numbers, underscores, at sign, and dollar signs
-        // 3. Cannot start with a digit or special character (except underscore)
-        if (dbName == null || dbName.length() > 128) {
-            return false;
-        }
-        return dbName.matches("[a-zA-Z_@#][a-zA-Z0-9_@#$]*");
+        return !StringUtils.isNullOrWhitespaceOnly(dbName) && dbName.length() <= 128;
     }
 
     /** Replaces the default timezone placeholder with session timezone, if applicable. */

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/main/java/org/apache/flink/cdc/connectors/sqlserver/source/SqlServerDataSourceOptions.java

@@ -45,7 +45,7 @@ public class SqlServerDataSourceOptions {
                     .stringType()
                     .noDefaultValue()
                     .withDescription(
-                            "Name of the SQL Server database to use when connecting to the SQL Server database server.");
+                            "Name of the SQL Server user to use when connecting to the SQL Server database server.");
 
     public static final ConfigOption<String> PASSWORD =
             ConfigOptions.key("password")
@@ -131,30 +131,6 @@ public class SqlServerDataSourceOptions {
                     .withDescription(
                             "Optional timestamp used in case of \"timestamp\" startup mode");
 
-    public static final ConfigOption<Double> SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND =
-            ConfigOptions.key("chunk-key.even-distribution.factor.upper-bound")
-                    .doubleType()
-                    .defaultValue(1000.0d)
-                    .withFallbackKeys("split-key.even-distribution.factor.upper-bound")
-                    .withDescription(
-                            "The upper bound of chunk key distribution factor. The distribution factor is used to determine whether the"
-                                    + " table is evenly distribution or not."
-                                    + " The table chunks would use evenly calculation optimization when the data distribution is even,"
-                                    + " and the query for splitting would happen when it is uneven."
-                                    + " The distribution factor could be calculated by (MAX(id) - MIN(id) + 1) / rowCount.");
-
-    public static final ConfigOption<Double> SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND =
-            ConfigOptions.key("chunk-key.even-distribution.factor.lower-bound")
-                    .doubleType()
-                    .defaultValue(0.05d)
-                    .withFallbackKeys("split-key.even-distribution.factor.lower-bound")
-                    .withDescription(
-                            "The lower bound of chunk key distribution factor. The distribution factor is used to determine whether the"
-                                    + " table is evenly distribution or not."
-                                    + " The table chunks would use evenly calculation optimization when the data distribution is even,"
-                                    + " and the query for splitting would happen when it is uneven."
-                                    + " The distribution factor could be calculated by (MAX(id) - MIN(id) + 1) / rowCount.");
-
     public static final ConfigOption<Boolean> SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP =
             ConfigOptions.key("scan.incremental.snapshot.backfill.skip")
                     .booleanType()

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/main/java/org/apache/flink/cdc/connectors/sqlserver/source/SqlServerMetadataAccessor.java

@@ -23,6 +23,7 @@
 import org.apache.flink.cdc.common.source.MetadataAccessor;
 import org.apache.flink.cdc.connectors.sqlserver.source.config.SqlServerSourceConfig;
 import org.apache.flink.cdc.connectors.sqlserver.utils.SqlServerSchemaUtils;
+import org.apache.flink.table.api.ValidationException;
 
 import javax.annotation.Nullable;
 
@@ -56,7 +57,22 @@ public List<String> listNamespaces() {
      */
     @Override
     public List<String> listSchemas(@Nullable String namespace) {
-        return SqlServerSchemaUtils.listSchemas(sourceConfig, namespace);
+        return SqlServerSchemaUtils.listSchemas(sourceConfig, resolveNamespace(namespace));
+    }
+
+    private String resolveNamespace(@Nullable String namespace) {
+        if (namespace != null) {
+            return namespace;
+        }
+
+        List<String> configuredDatabases = sourceConfig.getDatabaseList();
+        if (configuredDatabases != null && !configuredDatabases.isEmpty()) {
+            return configuredDatabases.get(0);
+        }
+
+        throw new ValidationException(
+                "Namespace must not be null when listing SQL Server schemas and no database "
+                        + "is configured in the source configuration.");
     }
 
     /**

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/main/java/org/apache/flink/cdc/connectors/sqlserver/source/SqlServerSchemaDataTypeInference.java

@@ -29,8 +29,8 @@ public class SqlServerSchemaDataTypeInference extends DebeziumSchemaDataTypeInfe
 
     private static final long serialVersionUID = 1L;
 
-    // SQL Server doesn't have special geometry types like PostgreSQL,
-    // so we can use the default implementation from the parent class.
-    // If there are SQL Server specific types that need special handling,
-    // they can be added here by overriding the inferStruct method.
+    // SQL Server has database-specific types, but no special handling is currently
+    // needed here, so this class uses the default implementation from the parent class.
+    // If SQL Server-specific types require special handling in the future,
+    // it can be added here by overriding the inferStruct method.
 }

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/main/java/org/apache/flink/cdc/connectors/sqlserver/utils/SqlServerSchemaUtils.java

@@ -160,7 +160,7 @@ public static List<String> listNamespaces(JdbcConnection jdbc) throws SQLExcepti
     }
 
     public static String quote(String dbOrTableName) {
-        return "[" + dbOrTableName + "]";
+        return "[" + dbOrTableName.replace("]", "]]") + "]";
     }
 
     public static Schema getTableSchema(

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/main/java/org/apache/flink/cdc/connectors/sqlserver/utils/SqlServerTypeUtils.java

@@ -79,8 +79,9 @@ private static DataType convertFromColumn(Column column) {
             case Types.BIGINT:
                 return DataTypes.BIGINT();
             case Types.REAL:
-            case Types.FLOAT:
                 return DataTypes.FLOAT();
+            case Types.FLOAT:
+                return DataTypes.DOUBLE();
             case Types.DOUBLE:
                 return DataTypes.DOUBLE();
             case Types.NUMERIC:

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/test/java/org/apache/flink/cdc/connectors/sqlserver/factory/SqlServerDataSourceFactoryTest.java

@@ -40,6 +40,7 @@
 import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.HOSTNAME;
 import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.PASSWORD;
 import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.PORT;
+import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN;
 import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_STARTUP_MODE;
 import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_STARTUP_TIMESTAMP_MILLIS;
 import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.TABLES;
@@ -215,6 +216,26 @@ public void testOptionalOption() {
                 .isEqualTo(MSSQL_SERVER_CONTAINER.getMappedPort(MS_SQL_SERVER_PORT));
     }
 
+    @Test
+    public void testChunkKeyColumnOptionIsSupported() {
+        Map<String, String> options = new HashMap<>();
+        options.put(HOSTNAME.key(), MSSQL_SERVER_CONTAINER.getHost());
+        options.put(
+                PORT.key(),
+                String.valueOf(MSSQL_SERVER_CONTAINER.getMappedPort(MS_SQL_SERVER_PORT)));
+        options.put(USERNAME.key(), MSSQL_SERVER_CONTAINER.getUsername());
+        options.put(PASSWORD.key(), MSSQL_SERVER_CONTAINER.getPassword());
+        options.put(TABLES.key(), DATABASE_NAME + ".dbo.products");
+        options.put(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN.key(), "id");
+
+        Factory.Context context = new MockContext(Configuration.fromMap(options));
+        SqlServerDataSourceFactory factory = new SqlServerDataSourceFactory();
+
+        assertThat(factory.optionalOptions()).contains(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN);
+        SqlServerDataSource dataSource = (SqlServerDataSource) factory.createDataSource(context);
+        assertThat(dataSource.getSqlServerSourceConfig().getChunkKeyColumn()).isEqualTo("id");
+    }
+
     @Test
     public void testStartupFromTimestamp() {
         Map<String, String> options = new HashMap<>();
@@ -317,6 +338,29 @@ public void testTableValidationWithDifferentDatabases() {
                         "The value of option `tables` is `db1.dbo.table1,db2.dbo.table2`, but not all table names have the same database name");
     }
 
+    @Test
+    public void testTableValidationRequiresDatabaseSchemaTableFormat() {
+        Map<String, String> options = new HashMap<>();
+        options.put(HOSTNAME.key(), MSSQL_SERVER_CONTAINER.getHost());
+        options.put(
+                PORT.key(),
+                String.valueOf(MSSQL_SERVER_CONTAINER.getMappedPort(MS_SQL_SERVER_PORT)));
+        options.put(USERNAME.key(), MSSQL_SERVER_CONTAINER.getUsername());
+        options.put(PASSWORD.key(), MSSQL_SERVER_CONTAINER.getPassword());
+        options.put(TABLES.key(), DATABASE_NAME + ".dbo");
+
+        SqlServerDataSourceFactory factory = new SqlServerDataSourceFactory();
+        Factory.Context context = new MockContext(Configuration.fromMap(options));
+
+        assertThatThrownBy(() -> factory.createDataSource(context))
+                .isInstanceOf(IllegalStateException.class)
+                .hasMessageContaining(
+                        "Table '"
+                                + DATABASE_NAME
+                                + ".dbo' does not match the expected 'database.schema.table' format.")
+                .hasMessageContaining(TABLES.key());
+    }
+
     static class MockContext implements Factory.Context {
 
         Configuration factoryConfiguration;

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/test/java/org/apache/flink/cdc/connectors/sqlserver/source/SqlServerMetadataAccessorITCase.java

@@ -74,6 +74,14 @@ public void testListSchemas() {
         assertThat(schemas).contains("dbo");
     }
 
+    @Test
+    public void testListSchemasUsesConfiguredDatabaseWhenNamespaceIsNull() {
+        String[] tables = new String[] {"dbo.full_types"};
+        SqlServerMetadataAccessor metadataAccessor = getMetadataAccessor(tables);
+
+        assertThat(metadataAccessor.listSchemas(null)).contains("dbo");
+    }
+
     @Test
     public void testListTables() {
         String[] tables =

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/test/java/org/apache/flink/cdc/connectors/sqlserver/source/SqlServerOnlineSchemaMigrationITCase.java

@@ -106,8 +106,6 @@ void testSchemaMigrationFromScratch() throws Exception {
                                 new EventTypeInfo())
                         .executeAndCollect();
 
-        Thread.sleep(5_000);
-
         List<Event> expected = new ArrayList<>();
         Schema schemaV1 =
                 Schema.newBuilder()
@@ -124,9 +122,6 @@ void testSchemaMigrationFromScratch() throws Exception {
                 .containsExactlyInAnyOrderElementsOf(
                         expected.stream().map(Object::toString).collect(Collectors.toList()));
 
-        // Wait for a little while until we're in streaming mode.
-        Thread.sleep(5_000);
-
         // ADD COLUMN
         try (Connection connection = getJdbcConnection();
                 Statement statement = connection.createStatement()) {

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/test/java/org/apache/flink/cdc/connectors/sqlserver/source/SqlServerPipelineITCaseTest.java

@@ -336,7 +336,7 @@ private static <T> List<T> fetchResultsExcept(Iterator<T> iter, int size, T side
         while (size > 0 && iter.hasNext()) {
             T event = iter.next();
             if (sideEvent.getClass().isInstance(event)) {
-                sideResults.add(sideEvent);
+                sideResults.add(event);
             } else {
                 result.add(event);
                 size--;