feat(Spanner): implement initial SQL Server source connector and DML generation support for Spanner-to-SourceDB migration

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<String, HikariDataSource> 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());

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<String, Object> astraConfigMap = parseConfigToConfigMap(astraFileContent);
         return mapper.convertValue(astraConfigMap, AstraConnectionConfig.class);
+      case SQLSERVER:
       case ORACLE:
       case MYSQL:
       case PG:

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;
 

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<String, SourceTable> 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<SourceColumn> 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<String> 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();
+  }
+}

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";

v2/spanner-to-sourcedb/pom.xml

@@ -154,6 +154,11 @@
             <artifactId>ojdbc8</artifactId>
             <version>23.26.1.0.0</version>
         </dependency>
+        <dependency>
+            <groupId>com.microsoft.sqlserver</groupId>
+            <artifactId>mssql-jdbc</artifactId>
+            <version>${mssql-jdbc.version}</version>
+        </dependency>
     </dependencies>
 
     <profiles>

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 =

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) {