diff --git a/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelper.java b/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelper.java index 4dca05c070..14b9280b80 100644 --- a/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelper.java +++ b/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelper.java @@ -51,15 +51,29 @@ public synchronized void init(ConnectionHelperRequest connectionHelperRequest) { "Initializing connection pool with size: {}", connectionHelperRequest.getMaxConnections()); Map localMap = new HashMap<>(); for (Shard shard : connectionHelperRequest.getShards()) { - String sourceConnectionUrl = - new StringBuilder() - .append(connectionHelperRequest.getJdbcUrlPrefix()) - .append(shard.getHost()) - .append(":") - .append(shard.getPort()) - .append("/") - .append(shard.getDbName()) - .toString(); + String sourceConnectionUrl; + if (connectionHelperRequest.getJdbcUrlPrefix() != null + && connectionHelperRequest.getJdbcUrlPrefix().startsWith("jdbc:sqlserver://")) { + sourceConnectionUrl = + new StringBuilder() + .append(connectionHelperRequest.getJdbcUrlPrefix()) + .append(shard.getHost()) + .append(":") + .append(shard.getPort()) + .append(";databaseName=") + .append(shard.getDbName()) + .toString(); + } else { + sourceConnectionUrl = + new StringBuilder() + .append(connectionHelperRequest.getJdbcUrlPrefix()) + .append(shard.getHost()) + .append(":") + .append(shard.getPort()) + .append("/") + .append(shard.getDbName()) + .toString(); + } HikariConfig config = new HikariConfig(); config.setJdbcUrl(sourceConnectionUrl); config.setUsername(shard.getUserName()); diff --git a/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/source/config/SourceConfigParser.java b/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/source/config/SourceConfigParser.java index b99111dafb..8311662fc1 100644 --- a/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/source/config/SourceConfigParser.java +++ b/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/source/config/SourceConfigParser.java @@ -92,6 +92,7 @@ public SourceConnectionConfig parseConfiguration( String astraFileContent = FileLoader.readConfigFilePath(sourceConfigFilePath); Map astraConfigMap = parseConfigToConfigMap(astraFileContent); return mapper.convertValue(astraConfigMap, AstraConnectionConfig.class); + case SQLSERVER: case ORACLE: case MYSQL: case PG: diff --git a/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/source/config/SourceType.java b/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/source/config/SourceType.java index fbff9eb158..e08d3395a5 100644 --- a/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/source/config/SourceType.java +++ b/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/source/config/SourceType.java @@ -30,7 +30,10 @@ public enum SourceType { PG(new String[] {"postgresql"}), /** Oracle source database type. */ - ORACLE(new String[] {"oracle"}); + ORACLE(new String[] {"oracle"}), + + /** SQL Server source database type. */ + SQLSERVER(new String[] {"sqlserver"}); private final String[] sourceTypeStringValues; diff --git a/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/sourceddl/SQLServerInformationSchemaScanner.java b/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/sourceddl/SQLServerInformationSchemaScanner.java new file mode 100644 index 0000000000..43388e447c --- /dev/null +++ b/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/sourceddl/SQLServerInformationSchemaScanner.java @@ -0,0 +1,103 @@ +/* + * Copyright (C) 2026 Google LLC + * + * Licensed 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 com.google.cloud.teleport.v2.spanner.sourceddl; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import java.sql.Connection; +import java.sql.DatabaseMetaData; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +public class SQLServerInformationSchemaScanner implements SourceSchemaScanner { + private final Connection connection; + private final String databaseName; + + public SQLServerInformationSchemaScanner(Connection connection, String databaseName) { + this.connection = connection; + this.databaseName = databaseName; + } + + @Override + public SourceSchema scan() { + Map tablesMap = new HashMap<>(); + SourceSchema.Builder builder = + SourceSchema.builder(SourceDatabaseType.SQLSERVER).databaseName(databaseName); + try { + DatabaseMetaData metaData = connection.getMetaData(); + String schemaPattern = "dbo"; + + try (ResultSet rs = metaData.getTables(null, schemaPattern, "%", new String[] {"TABLE"})) { + while (rs.next()) { + String tableName = rs.getString("TABLE_NAME"); + if (tableName == null + || tableName.startsWith("trace_xe_") + || tableName.startsWith("spt_")) { + continue; + } + SourceTable table = scanTable(metaData, schemaPattern, tableName); + tablesMap.put(tableName, table); + } + } + } catch (SQLException e) { + throw new RuntimeException("Failed to scan SQL Server information schema", e); + } + return builder.tables(ImmutableMap.copyOf(tablesMap)).build(); + } + + private SourceTable scanTable(DatabaseMetaData metaData, String schemaPattern, String tableName) + throws SQLException { + SourceTable.Builder tableBuilder = + SourceTable.builder(SourceDatabaseType.SQLSERVER).name(tableName).schema(schemaPattern); + List columns = new ArrayList<>(); + + try (ResultSet colsRs = metaData.getColumns(null, schemaPattern, tableName, "%")) { + while (colsRs.next()) { + String columnName = colsRs.getString("COLUMN_NAME"); + String dataType = colsRs.getString("TYPE_NAME"); + + SourceColumn.Builder colBuilder = + SourceColumn.builder(SourceDatabaseType.SQLSERVER) + .name(columnName) + .type(dataType) + .isNullable("YES".equalsIgnoreCase(colsRs.getString("IS_NULLABLE"))); + + String isAutoIncrement = ""; + try { + isAutoIncrement = colsRs.getString("IS_AUTOINCREMENT"); + } catch (Exception e) { + } + colBuilder.isGenerated("YES".equalsIgnoreCase(isAutoIncrement)); + columns.add(colBuilder.build()); + } + } + + List pks = new ArrayList<>(); + try (ResultSet pkRs = metaData.getPrimaryKeys(null, schemaPattern, tableName)) { + while (pkRs.next()) { + pks.add(pkRs.getString("COLUMN_NAME")); + } + } + + tableBuilder.columns(ImmutableList.copyOf(columns)); + tableBuilder.primaryKeyColumns(ImmutableList.copyOf(pks)); + return tableBuilder.build(); + } +} diff --git a/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/avro/GenericRecordTypeConvertorTest.java b/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/avro/GenericRecordTypeConvertorTest.java index 3c326c220a..1df8d50484 100644 --- a/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/avro/GenericRecordTypeConvertorTest.java +++ b/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/avro/GenericRecordTypeConvertorTest.java @@ -303,10 +303,7 @@ public void testHandleLogicalFieldType() { col = "json_col"; result = GenericRecordTypeConvertor.handleLogicalFieldType( - col, - genericRecord.get(col), - genericRecord.getSchema().getField(col).schema(), - null); + col, genericRecord.get(col), genericRecord.getSchema().getField(col).schema(), null); assertEquals("Test json_col conversion: ", "{\"k1\":\"476F6F676C65\"}", result); col = "json_col"; diff --git a/v2/spanner-to-sourcedb/pom.xml b/v2/spanner-to-sourcedb/pom.xml index 059f7d2ac5..4a58d82262 100644 --- a/v2/spanner-to-sourcedb/pom.xml +++ b/v2/spanner-to-sourcedb/pom.xml @@ -154,6 +154,11 @@ ojdbc8 23.26.1.0.0 + + com.microsoft.sqlserver + mssql-jdbc + ${mssql-jdbc.version} + diff --git a/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/constants/Constants.java b/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/constants/Constants.java index 75285e9ee1..ace1debbb9 100644 --- a/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/constants/Constants.java +++ b/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/constants/Constants.java @@ -81,6 +81,7 @@ public class Constants { public static final String SOURCE_SPANNER = "spanner"; public static final String SOURCE_ORACLE = "oracle"; + public static final String SOURCE_SQLSERVER = "sqlserver"; // Message written to the file for filtered records public static final String FILTERED_TAG_MESSAGE = diff --git a/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/dbutils/processor/SourceProcessorFactory.java b/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/dbutils/processor/SourceProcessorFactory.java index 3ff4f55ab2..4690a0b50b 100644 --- a/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/dbutils/processor/SourceProcessorFactory.java +++ b/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/dbutils/processor/SourceProcessorFactory.java @@ -41,6 +41,10 @@ public class SourceProcessorFactory { sourceMap.put(Constants.SOURCE_CASSANDRA, new CassandraSpToSrcSourceConnector()); sourceMap.put(Constants.SOURCE_SPANNER, new SpannerSpToSrcSourceConnector()); sourceMap.put(Constants.SOURCE_ORACLE, new OracleSpToSrcSourceConnector()); + sourceMap.put( + Constants.SOURCE_SQLSERVER, + new com.google.cloud.teleport.v2.templates.source.sqlserver + .SQLServerSpToSrcSourceConnector()); } public static void registerSource(String sourceName, ISpToSrcSourceConnector source) { diff --git a/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/sqlserver/SQLServerDMLGenerator.java b/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/sqlserver/SQLServerDMLGenerator.java new file mode 100644 index 0000000000..bc94a82a9c --- /dev/null +++ b/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/sqlserver/SQLServerDMLGenerator.java @@ -0,0 +1,315 @@ +/* + * Copyright (C) 2026 Google LLC + * + * Licensed 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 com.google.cloud.teleport.v2.templates.source.sqlserver; + +import com.google.cloud.teleport.v2.spanner.ddl.Column; +import com.google.cloud.teleport.v2.spanner.ddl.Ddl; +import com.google.cloud.teleport.v2.spanner.ddl.Table; +import com.google.cloud.teleport.v2.spanner.migrations.schema.ISchemaMapper; +import com.google.cloud.teleport.v2.spanner.sourceddl.SourceColumn; +import com.google.cloud.teleport.v2.spanner.sourceddl.SourceSchema; +import com.google.cloud.teleport.v2.spanner.sourceddl.SourceTable; +import com.google.cloud.teleport.v2.spanner.type.Type; +import com.google.cloud.teleport.v2.templates.dbutils.dml.DMLGeneratorUtils; +import com.google.cloud.teleport.v2.templates.dbutils.dml.IDMLGenerator; +import com.google.cloud.teleport.v2.templates.exceptions.InvalidDMLGenerationException; +import com.google.cloud.teleport.v2.templates.models.DMLGeneratorRequest; +import com.google.cloud.teleport.v2.templates.models.DMLGeneratorResponse; +import com.google.common.annotations.VisibleForTesting; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.NoSuchElementException; +import org.apache.commons.lang3.StringUtils; +import org.json.JSONObject; + +public class SQLServerDMLGenerator implements IDMLGenerator { + + public DMLGeneratorResponse getDMLStatement(DMLGeneratorRequest dmlGeneratorRequest) { + if (dmlGeneratorRequest == null) { + throw new InvalidDMLGenerationException( + "DMLGeneratorRequest is null. Cannot process the request."); + } + String spannerTableName = dmlGeneratorRequest.getSpannerTableName(); + ISchemaMapper schemaMapper = dmlGeneratorRequest.getSchemaMapper(); + Ddl spannerDdl = dmlGeneratorRequest.getSpannerDdl(); + SourceSchema sourceSchema = dmlGeneratorRequest.getSourceSchema(); + + if (schemaMapper == null) { + throw new InvalidDMLGenerationException("Schema Mapper must be not null"); + } + if (spannerDdl == null) { + throw new InvalidDMLGenerationException("Spanner Ddl must be not null."); + } + if (sourceSchema == null) { + throw new InvalidDMLGenerationException("SourceSchema must be not null."); + } + + Table spannerTable = spannerDdl.table(spannerTableName); + if (spannerTable == null) { + throw new InvalidDMLGenerationException( + String.format( + "The spanner table %s was not found in ddl found on spanner", spannerTableName)); + } + + String sourceTableName = ""; + try { + sourceTableName = schemaMapper.getSourceTableName("", spannerTableName); + } catch (NoSuchElementException e) { + throw new InvalidDMLGenerationException( + "Could not find source table name for spanner table: " + spannerTableName, e); + } + SourceTable sourceTable = sourceSchema.table(sourceTableName); + if (sourceTable == null) { + throw new InvalidDMLGenerationException( + String.format( + "Equivalent table %s was not found in source for spanner table %s", + sourceTableName, spannerTableName)); + } + + if (sourceTable.primaryKeyColumns() == null || sourceTable.primaryKeyColumns().size() == 0) { + throw new InvalidDMLGenerationException( + String.format( + "Cannot reverse replicate for source table %s without primary key, skipping the record.", + sourceTableName)); + } + + Map pkcolumnNameValues = + DMLGeneratorUtils.getPkColumnValues( + schemaMapper, + spannerTable, + sourceTable, + dmlGeneratorRequest.getNewValuesJson(), + dmlGeneratorRequest.getKeyValuesJson(), + dmlGeneratorRequest.getSourceDbTimezoneOffset(), + dmlGeneratorRequest.getCustomTransformationResponse(), + SQLServerDMLGenerator::getMappedColumnValue, + new ArrayList<>()); + if (pkcolumnNameValues == null || pkcolumnNameValues.isEmpty()) { + throw new InvalidDMLGenerationException( + String.format( + "Cannot reverse replicate for table %s without primary key, skipping the record", + sourceTableName)); + } + + if ("INSERT".equals(dmlGeneratorRequest.getModType()) + || "UPDATE".equals(dmlGeneratorRequest.getModType())) { + return generateUpsertStatement( + spannerTable, sourceTable, dmlGeneratorRequest, pkcolumnNameValues); + + } else if ("DELETE".equals(dmlGeneratorRequest.getModType())) { + return getDeleteStatement(sourceTable.name(), pkcolumnNameValues); + } else { + throw new InvalidDMLGenerationException( + String.format( + "Unsupported modType: %s for table %s", + dmlGeneratorRequest.getModType(), spannerTableName)); + } + } + + private static DMLGeneratorResponse getUpsertStatement( + String tableName, + Map allColumnNameValues, + Map pkColumnNameValues) { + + String updateValues = ""; + String insertColumns = ""; + String insertValues = ""; + String onCondition = ""; + + int pkIndex = 0; + for (Map.Entry entry : pkColumnNameValues.entrySet()) { + if (pkIndex > 0) { + onCondition += " AND "; + } + onCondition += "target.[" + entry.getKey() + "] = " + entry.getValue(); + pkIndex++; + } + + int index = 0; + for (Map.Entry entry : allColumnNameValues.entrySet()) { + String colName = entry.getKey(); + String colValue = entry.getValue(); + String sqlValue = (colValue == null) ? "NULL" : colValue; + + if (index > 0) { + insertColumns += ", "; + insertValues += ", "; + } + insertColumns += "[" + colName + "]"; + insertValues += sqlValue; + + if (!pkColumnNameValues.containsKey(colName)) { + if (updateValues.length() > 0) { + updateValues += ", "; + } + updateValues += "target.[" + colName + "] = " + sqlValue; + } + index++; + } + + String returnVal = + "MERGE INTO [" + + tableName + + "] AS target " + + "USING (SELECT 1 AS dummy) AS source " + + "ON (" + + onCondition + + ") "; + + if (updateValues.length() > 0) { + returnVal += "WHEN MATCHED THEN UPDATE SET " + updateValues + " "; + } + returnVal += + "WHEN NOT MATCHED THEN INSERT (" + insertColumns + ") VALUES (" + insertValues + ");"; + + return new DMLGeneratorResponse(returnVal); + } + + private static DMLGeneratorResponse getDeleteStatement( + String tableName, Map pkcolumnNameValues) { + String deleteValues = ""; + + int index = 0; + for (Map.Entry entry : pkcolumnNameValues.entrySet()) { + String colName = entry.getKey(); + String colValue = entry.getValue(); + + deleteValues += " [" + colName + "] = " + colValue; + if (index + 1 < pkcolumnNameValues.size()) { + deleteValues += " AND "; + } + index++; + } + String returnVal = "DELETE FROM [" + tableName + "] WHERE " + deleteValues; + + return new DMLGeneratorResponse(returnVal); + } + + private static DMLGeneratorResponse generateUpsertStatement( + Table spannerTable, + SourceTable sourceTable, + DMLGeneratorRequest dmlGeneratorRequest, + Map pkcolumnNameValues) { + Map columnNameValues = + DMLGeneratorUtils.getColumnValues( + dmlGeneratorRequest.getSchemaMapper(), + spannerTable, + sourceTable, + dmlGeneratorRequest.getNewValuesJson(), + dmlGeneratorRequest.getKeyValuesJson(), + dmlGeneratorRequest.getSourceDbTimezoneOffset(), + dmlGeneratorRequest.getCustomTransformationResponse(), + SQLServerDMLGenerator::getMappedColumnValue, + new ArrayList<>()); + columnNameValues.putAll(pkcolumnNameValues); + return getUpsertStatement(sourceTable.name(), columnNameValues, pkcolumnNameValues); + } + + @VisibleForTesting + static String getMappedColumnValue( + Column spannerColDef, + SourceColumn sourceColDef, + JSONObject valuesJson, + String sourceDbTimezoneOffset, + List preparedStatementParameters) { + + String colInputValue = ""; + Type colType = spannerColDef.type(); + String colName = spannerColDef.name(); + if (colType.getCode().equals(Type.Code.FLOAT64) + || colType.getCode().equals(Type.Code.FLOAT32) + || colType.getCode().equals(Type.Code.PG_FLOAT4) + || colType.getCode().equals(Type.Code.PG_FLOAT8) + || colType.getCode().equals(Type.Code.PG_NUMERIC)) { + colInputValue = valuesJson.getBigDecimal(colName).toString(); + } else if (colType.getCode().equals(Type.Code.BOOL) + || colType.getCode().equals(Type.Code.PG_BOOL)) { + // SQL Server bit: 1 for true, 0 for false + boolean b = valuesJson.getBoolean(colName); + colInputValue = b ? "1" : "0"; + } else if (colType.getCode().equals(Type.Code.BYTES) + || colType.getCode().equals(Type.Code.PG_BYTEA)) { + colInputValue = convertBase64ToHex(valuesJson.getString(colName)); + } else { + colInputValue = valuesJson.getString(colName); + } + String response = + getColumnValueByType( + sourceColDef.type(), colInputValue, sourceDbTimezoneOffset, colType.toString()); + return response; + } + + @VisibleForTesting + protected static String convertBase64ToHex(String base64EncodedString) { + String rawHex = DMLGeneratorUtils.convertBase64ToRawHex(base64EncodedString); + if (rawHex == null) { + return null; + } + return rawHex.isEmpty() ? "0x" : "0x" + rawHex; + } + + @VisibleForTesting + static String getColumnValueByType( + String columnType, String colValue, String sourceDbTimezoneOffset, String spannerColType) { + String response = ""; + switch (columnType.toLowerCase()) { + case "varchar": + case "char": + case "text": + case "nvarchar": + case "nchar": + case "ntext": + case "sysname": + case "uniqueidentifier": + case "xml": + case "date": + case "time": + case "datetime2": + case "datetimeoffset": + case "datetime": + case "smalldatetime": + response = getQuotedEscapedString(colValue, spannerColType); + break; + case "binary": + case "varbinary": + case "image": + response = colValue; // Already formatted as 0x... + break; + case "bit": + response = colValue.equals("true") || colValue.equals("1") ? "1" : "0"; + break; + default: + response = colValue; + } + return response; + } + + private static String escapeString(String input) { + String cleanedNullBytes = StringUtils.replace(input, "\u0000", ""); + cleanedNullBytes = StringUtils.replace(cleanedNullBytes, "'", "''"); + return cleanedNullBytes; + } + + private static String getQuotedEscapedString(String input, String spannerColType) { + if ("BYTES".equals(spannerColType) || "PG_BYTEA".equals(spannerColType)) { + return input; + } + String cleanedString = escapeString(input); + String response = "\'" + cleanedString + "\'"; + return response; + } +} diff --git a/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/sqlserver/SQLServerSpToSrcSourceConnector.java b/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/sqlserver/SQLServerSpToSrcSourceConnector.java new file mode 100644 index 0000000000..1ca5625bff --- /dev/null +++ b/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/sqlserver/SQLServerSpToSrcSourceConnector.java @@ -0,0 +1,143 @@ +/* + * Copyright (C) 2026 Google LLC + * + * Licensed 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 com.google.cloud.teleport.v2.templates.source.sqlserver; + +import com.google.cloud.teleport.v2.spanner.migrations.connection.ConnectionHelperRequest; +import com.google.cloud.teleport.v2.spanner.migrations.connection.IConnectionHelper; +import com.google.cloud.teleport.v2.spanner.migrations.connection.JdbcConnectionHelper; +import com.google.cloud.teleport.v2.spanner.migrations.shard.Shard; +import com.google.cloud.teleport.v2.spanner.migrations.source.config.JdbcShardConfig; +import com.google.cloud.teleport.v2.spanner.migrations.source.config.SourceConfigParser; +import com.google.cloud.teleport.v2.spanner.migrations.source.config.SourceConnectionConfig; +import com.google.cloud.teleport.v2.spanner.migrations.utils.ISecretManagerAccessor; +import com.google.cloud.teleport.v2.spanner.migrations.utils.SecretManagerAccessorImpl; +import com.google.cloud.teleport.v2.spanner.sourceddl.SourceSchema; +import com.google.cloud.teleport.v2.templates.dbutils.dao.source.IDao; +import com.google.cloud.teleport.v2.templates.dbutils.dao.source.JdbcDao; +import com.google.cloud.teleport.v2.templates.dbutils.dml.IDMLGenerator; +import com.google.cloud.teleport.v2.templates.dbutils.processor.ISpToSrcSourceConnector; +import com.google.common.annotations.VisibleForTesting; +import java.sql.Connection; +import java.util.List; +import org.apache.beam.sdk.options.PipelineOptions; + +public class SQLServerSpToSrcSourceConnector implements ISpToSrcSourceConnector { + + private final IConnectionHelper connectionHelper; + + public SQLServerSpToSrcSourceConnector() { + this.connectionHelper = new JdbcConnectionHelper(); + } + + @VisibleForTesting + SQLServerSpToSrcSourceConnector(IConnectionHelper connectionHelper) { + this.connectionHelper = connectionHelper; + } + + @Override + public IDMLGenerator getDmlGenerator() { + return new SQLServerDMLGenerator(); + } + + @Override + public IConnectionHelper getConnectionHelper() { + return connectionHelper; + } + + String getConnectionUrl(Shard shard) { + return "jdbc:sqlserver://" + + shard.getHost() + + ":" + + shard.getPort() + + ";databaseName=" + + shard.getDbName(); + } + + @Override + public IDao getDao(Shard shard) { + return new JdbcDao(getConnectionUrl(shard), shard.getUserName(), getConnectionHelper()); + } + + @Override + public void initConnectionHelper(List shards, int maxConnections) { + if (!connectionHelper.isConnectionPoolInitialized()) { + ConnectionHelperRequest request = + new ConnectionHelperRequest( + shards, + null, + maxConnections, + "com.microsoft.sqlserver.jdbc.SQLServerDriver", + null, + "jdbc:sqlserver://"); + connectionHelper.init(request); + } + } + + @Override + public List parseShardConfig(String shardFilePath) throws Exception { + ISecretManagerAccessor secretManagerAccessor = new SecretManagerAccessorImpl(); + SourceConfigParser sourceConfigParser = new SourceConfigParser(secretManagerAccessor); + SourceConnectionConfig sourceConnectionConfig = + sourceConfigParser.parseConfiguration("sqlserver", shardFilePath); + if (sourceConnectionConfig instanceof JdbcShardConfig) { + return ((JdbcShardConfig) sourceConnectionConfig).getShardConfigs(); + } + throw new IllegalArgumentException( + "Expected JdbcShardConfig but got: " + sourceConnectionConfig.getClass()); + } + + @Override + public void validate(List shards, PipelineOptions options) throws Exception { + // Basic validation could be implemented here + } + + @Override + public SourceSchema getInformationSchema(List shards) throws Exception { + try (Connection connection = createConnection(shards.get(0))) { + return new com.google.cloud.teleport.v2.spanner.sourceddl.SQLServerInformationSchemaScanner( + connection, shards.get(0).getDbName()) + .scan(); + } + } + + @VisibleForTesting + Connection createConnection(Shard shard) throws Exception { + Class.forName("com.microsoft.sqlserver.jdbc.SQLServerDriver"); + String url = + getConnectionUrl(shard) + + ";user=" + + shard.getUserName() + + ";password=" + + shard.getPassword() + + ";trustServerCertificate=true;encrypt=false"; + return java.sql.DriverManager.getConnection(url); + } + + @Override + public boolean supportsSharding() { + return true; + } + + @Override + public boolean shouldUpdateReadValuesToSpannerRecord() { + return true; + } + + @Override + public org.apache.beam.sdk.values.TupleTag classifyException(Throwable cause) { + return null; + } +} diff --git a/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/sqlserver/package-info.java b/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/sqlserver/package-info.java new file mode 100644 index 0000000000..4ed788a2d1 --- /dev/null +++ b/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/sqlserver/package-info.java @@ -0,0 +1,5 @@ +/** + * This package contains the SQL Server source connector implementation for the spanner-to-sourcedb + * template. + */ +package com.google.cloud.teleport.v2.templates.source.sqlserver; diff --git a/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/dbutils/processor/SourceProcessorFactoryTest.java b/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/dbutils/processor/SourceProcessorFactoryTest.java index 2930c1221a..b8180bbb23 100644 --- a/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/dbutils/processor/SourceProcessorFactoryTest.java +++ b/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/dbutils/processor/SourceProcessorFactoryTest.java @@ -162,6 +162,13 @@ public void testGetSource_Success() throws Exception { oracleSource instanceof com.google.cloud.teleport.v2.templates.source.oracle.OracleSpToSrcSourceConnector); + + ISpToSrcSourceConnector sqlServerSource = SourceProcessorFactory.getSource("sqlserver"); + Assert.assertTrue( + sqlServerSource + instanceof + com.google.cloud.teleport.v2.templates.source.sqlserver + .SQLServerSpToSrcSourceConnector); } @Test(expected = UnsupportedSourceException.class) diff --git a/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/source/sqlserver/SQLServerSpToSrcSourceConnectorTest.java b/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/source/sqlserver/SQLServerSpToSrcSourceConnectorTest.java new file mode 100644 index 0000000000..7c6a84734a --- /dev/null +++ b/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/source/sqlserver/SQLServerSpToSrcSourceConnectorTest.java @@ -0,0 +1,93 @@ +/* + * Copyright (C) 2026 Google LLC + * + * Licensed 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 com.google.cloud.teleport.v2.templates.source.sqlserver; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.google.cloud.teleport.v2.spanner.migrations.connection.ConnectionHelperRequest; +import com.google.cloud.teleport.v2.spanner.migrations.connection.IConnectionHelper; +import com.google.cloud.teleport.v2.spanner.migrations.shard.Shard; +import com.google.cloud.teleport.v2.templates.dbutils.dao.source.IDao; +import com.google.cloud.teleport.v2.templates.dbutils.dml.IDMLGenerator; +import java.util.List; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; + +@RunWith(MockitoJUnitRunner.class) +public class SQLServerSpToSrcSourceConnectorTest { + + @Mock private IConnectionHelper mockConnectionHelper; + @Mock private Shard mockShard; + + private SQLServerSpToSrcSourceConnector connector; + + @Before + public void setUp() { + connector = new SQLServerSpToSrcSourceConnector(mockConnectionHelper); + } + + @Test + public void testGetDmlGenerator() { + IDMLGenerator dmlGenerator = connector.getDmlGenerator(); + assertNotNull(dmlGenerator); + assertTrue(dmlGenerator instanceof SQLServerDMLGenerator); + } + + @Test + public void testGetConnectionHelper() { + assertEquals(mockConnectionHelper, connector.getConnectionHelper()); + } + + @Test + public void testGetConnectionUrl() { + when(mockShard.getHost()).thenReturn("localhost"); + when(mockShard.getPort()).thenReturn("1433"); + when(mockShard.getDbName()).thenReturn("testdb"); + + String url = connector.getConnectionUrl(mockShard); + assertEquals("jdbc:sqlserver://localhost:1433;databaseName=testdb", url); + } + + @Test + public void testGetDao() { + when(mockShard.getHost()).thenReturn("localhost"); + when(mockShard.getPort()).thenReturn("1433"); + when(mockShard.getDbName()).thenReturn("testdb"); + when(mockShard.getUserName()).thenReturn("user"); + + IDao dao = connector.getDao(mockShard); + assertNotNull(dao); + } + + @Test + public void testInitConnectionHelper() { + when(mockConnectionHelper.isConnectionPoolInitialized()).thenReturn(false); + doNothing().when(mockConnectionHelper).init(any(ConnectionHelperRequest.class)); + + connector.initConnectionHelper(List.of(mockShard), 10); + + verify(mockConnectionHelper).init(any(ConnectionHelperRequest.class)); + } +}