wenshao commented on code in PR #4320: URL: https://github.com/apache/flink-cdc/pull/4320#discussion_r3042811470
########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/main/java/org/apache/flink/cdc/connectors/sqlserver/utils/SqlServerSchemaUtils.java: ########## @@ -0,0 +1,326 @@ +/* + * 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.flink.cdc.connectors.sqlserver.utils; + +import org.apache.flink.cdc.common.event.TableId; +import org.apache.flink.cdc.common.schema.Column; +import org.apache.flink.cdc.common.schema.Schema; +import org.apache.flink.cdc.connectors.sqlserver.source.config.SqlServerSourceConfig; +import org.apache.flink.cdc.connectors.sqlserver.source.dialect.SqlServerDialect; +import org.apache.flink.cdc.connectors.sqlserver.source.utils.SqlServerConnectionUtils; +import org.apache.flink.table.api.ValidationException; + +import io.debezium.connector.sqlserver.SqlServerConnection; +import io.debezium.jdbc.JdbcConnection; +import io.debezium.relational.Table; +import io.debezium.relational.history.TableChanges.TableChange; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.stream.Collectors; + +/** Utilities for converting from debezium {@link Table} types to {@link Schema}. */ +public class SqlServerSchemaUtils { + + private static final Logger LOG = LoggerFactory.getLogger(SqlServerSchemaUtils.class); + + private static final String SQL_SERVER_AGENT_STATUS_QUERY = + "SELECT TOP(1) status_desc FROM sys.dm_server_services " + + "WHERE servicename LIKE 'SQL Server Agent (%'"; + + public static List<String> listSchemas(SqlServerSourceConfig sourceConfig, String namespace) { + try (JdbcConnection jdbc = + getSqlServerDialect(sourceConfig).openJdbcConnection(sourceConfig)) { + return listSchemas(jdbc, namespace); + } catch (SQLException e) { + throw new RuntimeException("Error to list schemas: " + e.getMessage(), e); + } + } + + public static List<String> listNamespaces(SqlServerSourceConfig sourceConfig) { + try (JdbcConnection jdbc = + getSqlServerDialect(sourceConfig).openJdbcConnection(sourceConfig)) { + return listNamespaces(jdbc); + } catch (SQLException e) { + throw new RuntimeException("Error to list namespaces: " + e.getMessage(), e); + } + } + + public static void validateSqlServerAgentRunning(SqlServerSourceConfig sourceConfig) { + try (JdbcConnection jdbc = + getSqlServerDialect(sourceConfig).openJdbcConnection(sourceConfig)) { + String agentStatus = + jdbc.queryAndMap( + SQL_SERVER_AGENT_STATUS_QUERY, + rs -> rs.next() ? rs.getString(1) : null); + if (!"Running".equalsIgnoreCase(agentStatus)) { + throw new ValidationException( + "SQL Server Agent is not running. Please start SQL Server Agent before" + + " creating the SQL Server pipeline."); + } + } catch (SQLException e) { + throw new ValidationException( + "Failed to validate SQL Server Agent status from sys.dm_server_services.", e); + } + } + + public static List<TableId> listTables( + SqlServerSourceConfig sourceConfig, + @Nullable String dbName, + @Nullable String schemaName) { + try (JdbcConnection jdbc = + getSqlServerDialect(sourceConfig).openJdbcConnection(sourceConfig)) { + List<String> databases = + dbName != null + ? Collections.singletonList(dbName) + : sourceConfig.getDatabaseList(); + + List<io.debezium.relational.TableId> dbzTableIds = + SqlServerConnectionUtils.listTables( + jdbc, sourceConfig.getTableFilters(), databases); + + return dbzTableIds.stream() + .filter( + tableId -> + schemaName == null + || schemaName.equalsIgnoreCase(tableId.schema())) + .map(SqlServerSchemaUtils::toCdcTableId) + .collect(Collectors.toList()); + } catch (SQLException e) { + throw new RuntimeException("Error to list tables: " + e.getMessage(), e); + } + } + + public static Schema getTableSchema(SqlServerSourceConfig sourceConfig, TableId tableId) { + SqlServerDialect dialect = getSqlServerDialect(sourceConfig); + try (JdbcConnection jdbc = dialect.openJdbcConnection(sourceConfig)) { + return getTableSchema(tableId, (SqlServerConnection) jdbc, dialect); + } catch (SQLException e) { + throw new RuntimeException("Error to get table schema: " + e.getMessage(), e); + } + } + + public static SqlServerDialect getSqlServerDialect(SqlServerSourceConfig sourceConfig) { + return new SqlServerDialect(sourceConfig); + } + + public static List<String> listSchemas(JdbcConnection jdbc, String namespace) + throws SQLException { + LOG.info("Read list of available schemas"); + final List<String> schemaNames = new ArrayList<>(); + + String querySql = + String.format( + "SELECT SCHEMA_NAME FROM %s.INFORMATION_SCHEMA.SCHEMATA", quote(namespace)); + + jdbc.query( + querySql, + rs -> { + while (rs.next()) { + schemaNames.add(rs.getString(1)); + } + }); + LOG.info("\t list of available schemas are: {}", schemaNames); + return schemaNames; + } + + public static List<String> listNamespaces(JdbcConnection jdbc) throws SQLException { + LOG.info("Read list of available namespaces (databases)"); + final List<String> namespaceNames = new ArrayList<>(); + jdbc.query( + "SELECT name FROM sys.databases", + rs -> { + while (rs.next()) { + namespaceNames.add(rs.getString(1)); + } + }); + LOG.info("\t list of available namespaces are: {}", namespaceNames); + return namespaceNames; + } + + public static String quote(String dbOrTableName) { Review Comment: **[Suggestion]** The `quote()` method wraps identifiers in `[...]` but does not escape embedded `]` characters as `]]`. The source connector module's `SqlServerUtils.quote()` correctly does `"[" + name.replace("]", "]]") + "]"`. This inconsistency means database names containing `]` would produce malformed SQL in the pipeline module's `listSchemas()` method. **Suggested fix:** ```java return "[" + dbOrTableName.replace("]", "]]") + "]"; ``` _— glm-5.1_ -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
