diff --git a/v2/sourcedb-to-spanner/pom.xml b/v2/sourcedb-to-spanner/pom.xml index e3d4b10821..3cd3b9de57 100644 --- a/v2/sourcedb-to-spanner/pom.xml +++ b/v2/sourcedb-to-spanner/pom.xml @@ -103,12 +103,6 @@ ${project.version} test - - com.google.apis - google-api-services-sqladmin - ${sqladmin-api.version} - test - org.apache.beam beam-it-jdbc @@ -185,7 +179,7 @@ com.microsoft.sqlserver mssql-jdbc ${mssql-jdbc.version} - test + diff --git a/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/options/SourceDbToSpannerOptions.java b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/options/SourceDbToSpannerOptions.java index e3fd9b8ba2..93bf8ef724 100644 --- a/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/options/SourceDbToSpannerOptions.java +++ b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/options/SourceDbToSpannerOptions.java @@ -26,6 +26,7 @@ public interface SourceDbToSpannerOptions extends CommonTemplateOptions { String MYSQL_SOURCE_DIALECT = "MYSQL"; String PG_SOURCE_DIALECT = "POSTGRESQL"; String ORACLE_SOURCE_DIALECT = "ORACLE"; + String SQLSERVER_SOURCE_DIALECT = "SQLSERVER"; @TemplateParameter.Enum( order = 1, @@ -35,10 +36,12 @@ public interface SourceDbToSpannerOptions extends CommonTemplateOptions { @TemplateParameter.TemplateEnumOption(CASSANDRA_SOURCE_DIALECT), @TemplateParameter.TemplateEnumOption(MYSQL_SOURCE_DIALECT), @TemplateParameter.TemplateEnumOption(PG_SOURCE_DIALECT), - @TemplateParameter.TemplateEnumOption(ORACLE_SOURCE_DIALECT) + @TemplateParameter.TemplateEnumOption(ORACLE_SOURCE_DIALECT), + @TemplateParameter.TemplateEnumOption(SQLSERVER_SOURCE_DIALECT) }, description = "Dialect of the source database", - helpText = "Possible values are `CASSANDRA`, `MYSQL`, `POSTGRESQL` and `ORACLE`.") + helpText = + "Possible values are `CASSANDRA`, `MYSQL`, `POSTGRESQL`, `ORACLE` and `SQLSERVER`.") @Default.String("MYSQL") String getSourceDbDialect(); diff --git a/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/reader/io/jdbc/iowrapper/config/SQLDialect.java b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/reader/io/jdbc/iowrapper/config/SQLDialect.java index 5e78432d27..8c82960f3f 100644 --- a/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/reader/io/jdbc/iowrapper/config/SQLDialect.java +++ b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/reader/io/jdbc/iowrapper/config/SQLDialect.java @@ -31,5 +31,6 @@ public enum SQLDialect { POSTGRESQL, CASSANDRA, ASTRA_DB, - ORACLE + ORACLE, + SQLSERVER } diff --git a/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/SourceConnectorFactory.java b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/SourceConnectorFactory.java index 766ab0a870..654df3f51b 100644 --- a/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/SourceConnectorFactory.java +++ b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/SourceConnectorFactory.java @@ -45,6 +45,8 @@ public static ISrcToSpSourceConnector getSourceConnectorByDialect( return new PostgresSrcToSpSourceConnector(); } else if (SourceDbToSpannerOptions.ORACLE_SOURCE_DIALECT.equals(dialect)) { return new com.google.cloud.teleport.v2.source.oracle.OracleSrcToSpSourceConnector(); + } else if (SourceDbToSpannerOptions.SQLSERVER_SOURCE_DIALECT.equals(dialect)) { + return new com.google.cloud.teleport.v2.source.sqlserver.SqlServerSrcToSpSourceConnector(); } /* Implementation detail, not having a default leads to failure in compile time checks enforced here */ throw new IllegalArgumentException("Unsupported source database dialect: " + dialect); @@ -70,6 +72,8 @@ public static ISrcToSpSourceConnector getSourceConnectorBySourceType(String sour return new CassandraSrcToSpSourceConnector(); case Constants.ORACLE_SOURCE_TYPE: return new com.google.cloud.teleport.v2.source.oracle.OracleSrcToSpSourceConnector(); + case Constants.SQLSERVER_SOURCE_TYPE: + return new com.google.cloud.teleport.v2.source.sqlserver.SqlServerSrcToSpSourceConnector(); default: throw new IllegalArgumentException("Unsupported source type: " + sourceType); } @@ -90,6 +94,8 @@ public static AbstractJdbcSrcToSpSourceConnector getSourceJdbcConnectorByDialect return new PostgresSrcToSpSourceConnector(); } else if (dialect == SQLDialect.ORACLE) { return new com.google.cloud.teleport.v2.source.oracle.OracleSrcToSpSourceConnector(); + } else if (dialect == SQLDialect.SQLSERVER) { + return new com.google.cloud.teleport.v2.source.sqlserver.SqlServerSrcToSpSourceConnector(); } throw new IllegalArgumentException("Unsupported SQL dialect: " + dialect); } diff --git a/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/sqlserver/SqlServerSrcToSpSourceConnector.java b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/sqlserver/SqlServerSrcToSpSourceConnector.java new file mode 100644 index 0000000000..8f6e7ef2f6 --- /dev/null +++ b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/sqlserver/SqlServerSrcToSpSourceConnector.java @@ -0,0 +1,151 @@ +/* + * 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.source.sqlserver; + +import com.google.cloud.teleport.v2.reader.io.jdbc.JdbcSchemaReference; +import com.google.cloud.teleport.v2.reader.io.jdbc.iowrapper.config.JdbcIOWrapperConfig; +import com.google.cloud.teleport.v2.reader.io.jdbc.iowrapper.config.SQLDialect; +import com.google.cloud.teleport.v2.reader.io.jdbc.rowmapper.JdbcValueMappingsProvider; +import com.google.cloud.teleport.v2.reader.io.schema.SourceSchemaReference; +import com.google.cloud.teleport.v2.reader.io.schema.typemapping.UnifiedTypeMapper; +import com.google.cloud.teleport.v2.reader.io.schema.typemapping.UnifiedTypeMapping; +import com.google.cloud.teleport.v2.reader.io.schema.typemapping.provider.unified.UnifiedMappingProvider; +import com.google.cloud.teleport.v2.source.jdbc.AbstractJdbcSrcToSpSourceConnector; +import com.google.cloud.teleport.v2.source.sqlserver.reader.io.jdbc.iowrapper.config.defaults.SqlServerConfigDefaults; +import com.google.cloud.teleport.v2.spanner.migrations.constants.Constants; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import java.util.Map; +import java.util.Map.Entry; +import org.apache.commons.lang3.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class SqlServerSrcToSpSourceConnector extends AbstractJdbcSrcToSpSourceConnector { + + private static final Logger LOG = LoggerFactory.getLogger(SqlServerSrcToSpSourceConnector.class); + + // Based on sqlserver-datatype.csv mapping matrix + private static final ImmutableMap MAPPING = + ImmutableMap.builder() + .put("TINYINT", UnifiedMappingProvider.Type.LONG) + .put("SMALLINT", UnifiedMappingProvider.Type.LONG) + .put("INT", UnifiedMappingProvider.Type.LONG) + .put("BIGINT", UnifiedMappingProvider.Type.LONG) + .put("BIT", UnifiedMappingProvider.Type.BOOLEAN) + .put("DECIMAL", UnifiedMappingProvider.Type.DECIMAL) + .put("NUMERIC", UnifiedMappingProvider.Type.DECIMAL) + .put("MONEY", UnifiedMappingProvider.Type.DECIMAL) + .put("SMALLMONEY", UnifiedMappingProvider.Type.DECIMAL) + .put("FLOAT", UnifiedMappingProvider.Type.DOUBLE) + .put("REAL", UnifiedMappingProvider.Type.FLOAT) + .put("DATE", UnifiedMappingProvider.Type.DATE) + .put("TIME", UnifiedMappingProvider.Type.STRING) + .put("DATETIME2", UnifiedMappingProvider.Type.TIMESTAMP) + .put("DATETIMEOFFSET", UnifiedMappingProvider.Type.TIMESTAMP) + .put("DATETIME", UnifiedMappingProvider.Type.TIMESTAMP) + .put("SMALLDATETIME", UnifiedMappingProvider.Type.TIMESTAMP) + .put("CHAR", UnifiedMappingProvider.Type.STRING) + .put("VARCHAR", UnifiedMappingProvider.Type.STRING) + .put("TEXT", UnifiedMappingProvider.Type.STRING) + .put("NCHAR", UnifiedMappingProvider.Type.STRING) + .put("NVARCHAR", UnifiedMappingProvider.Type.STRING) + .put("NTEXT", UnifiedMappingProvider.Type.STRING) + .put("BINARY", UnifiedMappingProvider.Type.BYTES) + .put("VARBINARY", UnifiedMappingProvider.Type.BYTES) + .put("IMAGE", UnifiedMappingProvider.Type.BYTES) + .put("ROWVERSION", UnifiedMappingProvider.Type.LONG) + .put( + "UNIQUEIDENTIFIER", + UnifiedMappingProvider.Type + .STRING) // UUID mapped to STRING in mapping provider? UUID is BYTES/STRING in + // Spanner. We can map to STRING + .put("XML", UnifiedMappingProvider.Type.STRING) + .build() + .entrySet() + .stream() + .map(e -> Map.entry(e.getKey(), UnifiedMappingProvider.getMapping(e.getValue()))) + .collect(ImmutableMap.toImmutableMap(Entry::getKey, Entry::getValue)); + + @Override + public ImmutableMap getTypeMapping() { + return MAPPING; + } + + public String getSourceType() { + return Constants.SQLSERVER_SOURCE_TYPE; + } + + @Override + public JdbcValueMappingsProvider getJdbcValueMappingsProvider() { + return SqlServerConfigDefaults.DEFAULT_SQLSERVER_VALUE_MAPPING_PROVIDER; + } + + @Override + public JdbcIOWrapperConfig.Builder getJdbcIOWrapperConfigBuilder() { + return JdbcIOWrapperConfig.builder() + .setSourceDbDialect(SQLDialect.SQLSERVER) + .setUnifiedTypeMapper(new UnifiedTypeMapper(getTypeMapping())) + .setDialectAdapter(SqlServerConfigDefaults.DEFAULT_SQLSERVER_DIALECT_ADAPTER) + .setValueMappingsProvider(SqlServerConfigDefaults.DEFAULT_SQLSERVER_VALUE_MAPPING_PROVIDER) + .setMaxConnections(SqlServerConfigDefaults.DEFAULT_SQLSERVER_MAX_CONNECTIONS) + .setSqlInitSeq(SqlServerConfigDefaults.DEFAULT_SQLSERVER_INIT_SEQ) + .setSchemaDiscoveryBackOff( + SqlServerConfigDefaults.DEFAULT_SQLSERVER_SCHEMA_DISCOVERY_BACKOFF) + .setTables(ImmutableList.of()) + .setTableVsPartitionColumns(ImmutableMap.of()) + .setMaxPartitions(null) + .setWaitOn(null) + .setDbParallelizationForReads(null) + .setDbParallelizationForSplitProcess( + JdbcIOWrapperConfig.DEFAULT_PARALLELIZATION_FOR_SLIT_PROCESS) + .setReadWithUniformPartitionsFeatureEnabled(true) + .setTestOnBorrow(JdbcIOWrapperConfig.DEFAULT_TEST_ON_BORROW) + .setTestOnCreate(JdbcIOWrapperConfig.DEFAULT_TEST_ON_CREATE) + .setTestOnReturn(JdbcIOWrapperConfig.DEFAULT_TEST_ON_RETURN) + .setTestWhileIdle(JdbcIOWrapperConfig.DEFAULT_TEST_WILE_IDLE) + .setValidationQuery(JdbcIOWrapperConfig.DEFAULT_VALIDATEION_QUERY) + .setRemoveAbandonedTimeout(JdbcIOWrapperConfig.DEFAULT_REMOVE_ABANDONED_TIMEOUT) + .setMinEvictableIdleTimeMillis(JdbcIOWrapperConfig.DEFAULT_MIN_EVICTABLE_IDLE_TIME_MILLIS) + .setSchemaDiscoveryConnectivityTimeoutMilliSeconds( + JdbcIOWrapperConfig.DEFAULT_SCHEMA_DISCOVERY_CONNECTIVITY_TIMEOUT_MILLISECONDS) + .setSplitStageCountHint(-1L) + .setWorkerMemoryBytes(null) + .setWorkerCores(null); + } + + @Override + public SourceSchemaReference getSourceSchemaReference(String dbName, String namespace) { + return SourceSchemaReference.ofJdbc( + JdbcSchemaReference.builder().setDbName(dbName).setNamespace(namespace).build()); + } + + @Override + public String getJdbcUrl( + String host, + int port, + String dbName, + String connectionProperties, + String namespace, + Integer fetchSize) { + String jdbcUrl = + "jdbc:sqlserver://" + host + ":" + port + ";databaseName=" + dbName + ";encrypt=false"; + if (StringUtils.isNotBlank(connectionProperties)) { + jdbcUrl = jdbcUrl + ";" + connectionProperties; + } + return jdbcUrl; + } +} diff --git a/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/sqlserver/package-info.java b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/sqlserver/package-info.java new file mode 100644 index 0000000000..34fd6683a2 --- /dev/null +++ b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/sqlserver/package-info.java @@ -0,0 +1,2 @@ +/** SQL Server source connector implementation. */ +package com.google.cloud.teleport.v2.source.sqlserver; diff --git a/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/sqlserver/reader/io/jdbc/dialectadapter/sqlserver/SqlServerDialectAdapter.java b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/sqlserver/reader/io/jdbc/dialectadapter/sqlserver/SqlServerDialectAdapter.java new file mode 100644 index 0000000000..3cc9866c77 --- /dev/null +++ b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/sqlserver/reader/io/jdbc/dialectadapter/sqlserver/SqlServerDialectAdapter.java @@ -0,0 +1,266 @@ +/* + * 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.source.sqlserver.reader.io.jdbc.dialectadapter.sqlserver; + +import com.google.cloud.teleport.v2.reader.io.exception.RetriableSchemaDiscoveryException; +import com.google.cloud.teleport.v2.reader.io.exception.SchemaDiscoveryException; +import com.google.cloud.teleport.v2.reader.io.jdbc.JdbcSchemaReference; +import com.google.cloud.teleport.v2.reader.io.jdbc.dialectadapter.DialectAdapter; +import com.google.cloud.teleport.v2.reader.io.schema.SourceColumnIndexInfo; +import com.google.cloud.teleport.v2.spanner.migrations.schema.SourceColumnType; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.HashMap; +import java.util.Map; +import javax.sql.DataSource; + +public class SqlServerDialectAdapter implements DialectAdapter { + + @Override + public ImmutableList discoverTables( + DataSource dataSource, JdbcSchemaReference sourceSchemaReference) + throws SchemaDiscoveryException, RetriableSchemaDiscoveryException { + String query = + "SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_TYPE = 'BASE TABLE' AND TABLE_CATALOG = ?"; + ImmutableList.Builder tablesBuilder = ImmutableList.builder(); + try (Connection conn = dataSource.getConnection(); + PreparedStatement stmt = conn.prepareStatement(query)) { + stmt.setString(1, sourceSchemaReference.dbName()); + try (ResultSet rs = stmt.executeQuery()) { + while (rs.next()) { + tablesBuilder.add(rs.getString(1)); + } + } + } catch (SQLException e) { + throw new SchemaDiscoveryException(e); + } + return tablesBuilder.build(); + } + + @Override + public ImmutableMap> discoverTableSchema( + DataSource dataSource, + JdbcSchemaReference sourceSchemaReference, + ImmutableList tables) + throws SchemaDiscoveryException, RetriableSchemaDiscoveryException { + if (tables.isEmpty()) { + return ImmutableMap.of(); + } + + String query = + "SELECT TABLE_NAME, COLUMN_NAME, DATA_TYPE, CHARACTER_MAXIMUM_LENGTH, NUMERIC_PRECISION, NUMERIC_SCALE FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_CATALOG = ? AND TABLE_NAME IN " + + DialectAdapter.generateInClause(tables.size()); + Map> builders = new HashMap<>(); + tables.forEach(table -> builders.put(table, ImmutableMap.builder())); + + try (Connection conn = dataSource.getConnection(); + PreparedStatement stmt = conn.prepareStatement(query)) { + stmt.setString(1, sourceSchemaReference.dbName()); + for (int i = 0; i < tables.size(); i++) { + stmt.setString(i + 2, tables.get(i)); + } + try (ResultSet rs = stmt.executeQuery()) { + while (rs.next()) { + String tableName = rs.getString("TABLE_NAME"); + String columnName = rs.getString("COLUMN_NAME"); + String dataType = rs.getString("DATA_TYPE"); + + long charMaxLen = rs.getLong("CHARACTER_MAXIMUM_LENGTH"); + boolean hasCharMaxLen = !rs.wasNull(); + + long numPrecision = rs.getLong("NUMERIC_PRECISION"); + boolean hasNumPrecision = !rs.wasNull(); + + long numScale = rs.getLong("NUMERIC_SCALE"); + boolean hasNumScale = !rs.wasNull(); + + SourceColumnType sourceColumnType; + if (hasCharMaxLen) { + sourceColumnType = new SourceColumnType(dataType, new Long[] {charMaxLen}, null); + } else if (hasNumPrecision && hasNumScale) { + sourceColumnType = + new SourceColumnType(dataType, new Long[] {numPrecision, numScale}, null); + } else if (hasNumPrecision) { + sourceColumnType = new SourceColumnType(dataType, new Long[] {numPrecision}, null); + } else { + sourceColumnType = new SourceColumnType(dataType, new Long[] {}, null); + } + if (builders.containsKey(tableName)) { + builders.get(tableName).put(columnName, sourceColumnType); + } + } + } + } catch (SQLException e) { + throw new SchemaDiscoveryException(e); + } + + ImmutableMap.Builder> result = + ImmutableMap.builder(); + builders.forEach((t, b) -> result.put(t, b.build())); + return result.build(); + } + + @Override + public ImmutableMap> discoverTableIndexes( + DataSource dataSource, + JdbcSchemaReference sourceSchemaReference, + ImmutableList tables) + throws SchemaDiscoveryException, RetriableSchemaDiscoveryException { + if (tables.isEmpty()) { + return ImmutableMap.of(); + } + // Simplified index discovery for SQL Server. Focus on primary keys. + String query = + "SELECT " + + " t.name AS table_name, " + + " ind.name AS index_name, " + + " col.name AS column_name, " + + " ic.key_ordinal AS ordinal_position, " + + " ind.is_unique, " + + " ind.is_primary_key AS is_primary, " + + " ty.name AS type_name " + + "FROM sys.indexes ind " + + "INNER JOIN sys.index_columns ic ON ind.object_id = ic.object_id and ind.index_id = ic.index_id " + + "INNER JOIN sys.columns col ON ic.object_id = col.object_id and ic.column_id = col.column_id " + + "INNER JOIN sys.tables t ON ind.object_id = t.object_id " + + "INNER JOIN sys.types ty ON col.system_type_id = ty.system_type_id AND col.user_type_id = ty.user_type_id " + + "WHERE t.name IN " + + DialectAdapter.generateInClause(tables.size()); + + Map> builders = new HashMap<>(); + tables.forEach(table -> builders.put(table, ImmutableList.builder())); + + try (Connection conn = dataSource.getConnection(); + PreparedStatement stmt = conn.prepareStatement(query)) { + for (int i = 0; i < tables.size(); i++) { + stmt.setString(i + 1, tables.get(i)); + } + try (ResultSet rs = stmt.executeQuery()) { + while (rs.next()) { + String tableName = rs.getString("table_name"); + String typeName = rs.getString("type_name"); + // Just map string to STRING and everything else to NUMERIC to simplify. + SourceColumnIndexInfo.IndexType indexType = SourceColumnIndexInfo.IndexType.OTHER; + if (typeName.toUpperCase().contains("CHAR")) { + indexType = SourceColumnIndexInfo.IndexType.STRING; + } else if (typeName.toUpperCase().contains("INT")) { + indexType = SourceColumnIndexInfo.IndexType.NUMERIC; + } + + SourceColumnIndexInfo info = + SourceColumnIndexInfo.builder() + .setColumnName(rs.getString("column_name")) + .setIndexName(rs.getString("index_name")) + .setIsUnique(rs.getBoolean("is_unique")) + .setIsPrimary(rs.getBoolean("is_primary")) + .setOrdinalPosition(rs.getLong("ordinal_position")) + .setCardinality(100L) // stub + .setColumnTypeName(typeName) + .setIndexType(indexType) + .build(); + + if (builders.containsKey(tableName)) { + builders.get(tableName).add(info); + } + } + } + } catch (SQLException e) { + throw new SchemaDiscoveryException(e); + } + + ImmutableMap.Builder> result = + ImmutableMap.builder(); + builders.forEach((t, b) -> result.put(t, b.build())); + return result.build(); + } + + @Override + public String getReadQuery(String tableName, ImmutableList partitionColumns) { + StringBuilder queryBuilder = new StringBuilder("SELECT * FROM " + tableName); + if (!partitionColumns.isEmpty()) { + queryBuilder.append(" WHERE "); + queryBuilder.append( + String.join( + " AND ", + partitionColumns.stream() + .map( + col -> + String.format( + "((? = 0) OR (%1$s >= ? AND (%1$s < ? OR (? = 1 AND %1$s = ?))))", + col)) + .toArray(String[]::new))); + } + return queryBuilder.toString(); + } + + @Override + public String getCountQuery( + String tableName, ImmutableList partitionColumns, long timeoutMillis) { + StringBuilder queryBuilder = new StringBuilder("SELECT COUNT(*) FROM " + tableName); + if (!partitionColumns.isEmpty()) { + queryBuilder.append(" WHERE "); + queryBuilder.append( + String.join( + " AND ", + partitionColumns.stream() + .map( + col -> + String.format( + "((? = 0) OR (%1$s >= ? AND (%1$s < ? OR (? = 1 AND %1$s = ?))))", + col)) + .toArray(String[]::new))); + } + return queryBuilder.toString(); + } + + @Override + public String getBoundaryQuery( + String tableName, ImmutableList partitionColumns, String colName) { + StringBuilder queryBuilder = + new StringBuilder( + String.format("SELECT MIN(%s), MAX(%s) FROM %s", colName, colName, tableName)); + if (!partitionColumns.isEmpty()) { + queryBuilder.append(" WHERE "); + queryBuilder.append( + String.join( + " AND ", + partitionColumns.stream() + .map( + col -> + String.format( + "((? = 0) OR (%1$s >= ? AND (%1$s < ? OR (? = 1 AND %1$s = ?))))", + col)) + .toArray(String[]::new))); + } + return queryBuilder.toString(); + } + + @Override + public boolean checkForTimeout(SQLException exception) { + return false; + } + + @Override + public String getCollationsOrderQuery(String dbCharset, String dbCollation, boolean padSpace) { + // Basic implementation to satisfy the interface. + // Real implementation would need to return the ordered characters based on collation. + return "SELECT 'a' AS char_col"; + } +} diff --git a/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/sqlserver/reader/io/jdbc/dialectadapter/sqlserver/package-info.java b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/sqlserver/reader/io/jdbc/dialectadapter/sqlserver/package-info.java new file mode 100644 index 0000000000..7debb73267 --- /dev/null +++ b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/sqlserver/reader/io/jdbc/dialectadapter/sqlserver/package-info.java @@ -0,0 +1,2 @@ +/** SQL Server dialect adapter. */ +package com.google.cloud.teleport.v2.source.sqlserver.reader.io.jdbc.dialectadapter.sqlserver; diff --git a/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/sqlserver/reader/io/jdbc/iowrapper/config/defaults/SqlServerConfigDefaults.java b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/sqlserver/reader/io/jdbc/iowrapper/config/defaults/SqlServerConfigDefaults.java new file mode 100644 index 0000000000..a7a8b9a883 --- /dev/null +++ b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/sqlserver/reader/io/jdbc/iowrapper/config/defaults/SqlServerConfigDefaults.java @@ -0,0 +1,44 @@ +/* + * 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.source.sqlserver.reader.io.jdbc.iowrapper.config.defaults; + +import com.google.cloud.teleport.v2.reader.io.jdbc.dialectadapter.DialectAdapter; +import com.google.cloud.teleport.v2.reader.io.jdbc.rowmapper.JdbcValueMappingsProvider; +import com.google.cloud.teleport.v2.source.sqlserver.reader.io.jdbc.dialectadapter.sqlserver.SqlServerDialectAdapter; +import com.google.cloud.teleport.v2.source.sqlserver.reader.io.jdbc.rowmapper.provider.SqlServerJdbcValueMappings; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import org.apache.beam.sdk.util.FluentBackoff; +import org.joda.time.Duration; + +public class SqlServerConfigDefaults { + public static final DialectAdapter DEFAULT_SQLSERVER_DIALECT_ADAPTER = + new SqlServerDialectAdapter(); + public static final JdbcValueMappingsProvider DEFAULT_SQLSERVER_VALUE_MAPPING_PROVIDER = + new SqlServerJdbcValueMappings(); + + public static final Long DEFAULT_SQLSERVER_MAX_CONNECTIONS = 160L; + + public static final ImmutableMap DEFAULT_SQLSERVER_URL_PROPERTIES = + ImmutableMap.of(); + + public static final FluentBackoff DEFAULT_SQLSERVER_SCHEMA_DISCOVERY_BACKOFF = + FluentBackoff.DEFAULT.withMaxCumulativeBackoff(Duration.standardMinutes(5L)); + + public static final ImmutableList DEFAULT_SQLSERVER_INIT_SEQ = ImmutableList.of(); + + private SqlServerConfigDefaults() {} +} diff --git a/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/sqlserver/reader/io/jdbc/iowrapper/config/defaults/package-info.java b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/sqlserver/reader/io/jdbc/iowrapper/config/defaults/package-info.java new file mode 100644 index 0000000000..c885724712 --- /dev/null +++ b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/sqlserver/reader/io/jdbc/iowrapper/config/defaults/package-info.java @@ -0,0 +1,2 @@ +/** SQL Server IO wrapper config defaults. */ +package com.google.cloud.teleport.v2.source.sqlserver.reader.io.jdbc.iowrapper.config.defaults; diff --git a/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/sqlserver/reader/io/jdbc/rowmapper/provider/SqlServerJdbcValueMappings.java b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/sqlserver/reader/io/jdbc/rowmapper/provider/SqlServerJdbcValueMappings.java new file mode 100644 index 0000000000..00c81ae670 --- /dev/null +++ b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/sqlserver/reader/io/jdbc/rowmapper/provider/SqlServerJdbcValueMappings.java @@ -0,0 +1,104 @@ +/* + * 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.source.sqlserver.reader.io.jdbc.rowmapper.provider; + +import com.google.cloud.teleport.v2.reader.io.jdbc.rowmapper.JdbcMappings; +import com.google.cloud.teleport.v2.reader.io.jdbc.rowmapper.JdbcValueMapper; +import com.google.cloud.teleport.v2.reader.io.jdbc.rowmapper.JdbcValueMappingsProvider; +import com.google.cloud.teleport.v2.reader.io.jdbc.rowmapper.ResultSetValueExtractor; +import com.google.cloud.teleport.v2.reader.io.jdbc.rowmapper.ResultSetValueMapper; +import com.google.cloud.teleport.v2.spanner.migrations.schema.SourceColumnType; +import com.google.common.collect.ImmutableMap; +import java.math.BigDecimal; +import java.nio.ByteBuffer; +import java.sql.ResultSet; +import java.util.Calendar; +import java.util.TimeZone; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class SqlServerJdbcValueMappings implements JdbcValueMappingsProvider { + + private static final Logger LOG = LoggerFactory.getLogger(SqlServerJdbcValueMappings.class); + + private static final ResultSetValueMapper valuePassThrough = (value, schema) -> value; + + private static final ResultSetValueMapper bytesToByteBuffer = + (value, schema) -> ByteBuffer.wrap(value); + + private static final ResultSetValueMapper sqlTimestampToAvroTimestampMicros = + (value, schema) -> value.getTime() * 1000 + (value.getNanos() / 1000) % 1000; + + private static final ResultSetValueExtractor utcTimeStampExtractor = + (rs, index) -> rs.getTimestamp(index, Calendar.getInstance(TimeZone.getTimeZone("UTC"))); + + private static final ResultSetValueExtractor utcDateExtractor = + (rs, index) -> rs.getDate(index, Calendar.getInstance(TimeZone.getTimeZone("UTC"))); + + private static final ResultSetValueMapper sqlDateToAvroDate = + (value, schema) -> (int) value.toLocalDate().toEpochDay(); + + private static final ResultSetValueMapper bigDecimalToByteArray = + (value, schema) -> ByteBuffer.wrap(value.unscaledValue().toByteArray()); + + private static final JdbcMappings JDBC_MAPPINGS = + JdbcMappings.builder() + .put("TINYINT", ResultSet::getLong, valuePassThrough, 1) + .put("SMALLINT", ResultSet::getLong, valuePassThrough, 2) + .put("INT", ResultSet::getLong, valuePassThrough, 4) + .put("BIGINT", ResultSet::getLong, valuePassThrough, 8) + .put("BIT", ResultSet::getBoolean, valuePassThrough, 1) + .put("DECIMAL", ResultSet::getBigDecimal, bigDecimalToByteArray, 16) + .put("NUMERIC", ResultSet::getBigDecimal, bigDecimalToByteArray, 16) + .put("MONEY", ResultSet::getBigDecimal, bigDecimalToByteArray, 8) + .put("SMALLMONEY", ResultSet::getBigDecimal, bigDecimalToByteArray, 4) + .put("FLOAT", ResultSet::getDouble, valuePassThrough, 8) + .put("REAL", ResultSet::getFloat, valuePassThrough, 4) + .put("DATE", utcDateExtractor, sqlDateToAvroDate, 4) + .put("TIME", ResultSet::getString, valuePassThrough, 12) + .put("DATETIME2", utcTimeStampExtractor, sqlTimestampToAvroTimestampMicros, 11) + .put("DATETIMEOFFSET", utcTimeStampExtractor, sqlTimestampToAvroTimestampMicros, 11) + .put("DATETIME", utcTimeStampExtractor, sqlTimestampToAvroTimestampMicros, 11) + .put("SMALLDATETIME", utcTimeStampExtractor, sqlTimestampToAvroTimestampMicros, 11) + .put("CHAR", ResultSet::getString, valuePassThrough, 255) + .put("VARCHAR", ResultSet::getString, valuePassThrough, 65535) + .put("TEXT", ResultSet::getString, valuePassThrough, 65535) + .put("NCHAR", ResultSet::getString, valuePassThrough, 255) + .put("NVARCHAR", ResultSet::getString, valuePassThrough, 65535) + .put("NTEXT", ResultSet::getString, valuePassThrough, 65535) + .put("BINARY", ResultSet::getBytes, bytesToByteBuffer, 65535) + .put("VARBINARY", ResultSet::getBytes, bytesToByteBuffer, 65535) + .put("IMAGE", ResultSet::getBytes, bytesToByteBuffer, 65535) + .put("ROWVERSION", ResultSet::getBytes, bytesToByteBuffer, 8) + .put("UNIQUEIDENTIFIER", ResultSet::getString, valuePassThrough, 36) + .put("XML", ResultSet::getString, valuePassThrough, 65535) + .build(); + + @Override + public ImmutableMap> getMappings() { + return JDBC_MAPPINGS.mappings(); + } + + @Override + public int estimateColumnSize(SourceColumnType sourceColumnType) { + String typeName = sourceColumnType.getName().toUpperCase(); + if (JDBC_MAPPINGS.sizeEstimators().containsKey(typeName)) { + return JDBC_MAPPINGS.sizeEstimators().get(typeName).apply(sourceColumnType); + } + LOG.warn("Unknown column type: {}. Defaulting to size: 65,535.", sourceColumnType); + return 65_535; + } +} diff --git a/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/sqlserver/reader/io/jdbc/rowmapper/provider/package-info.java b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/sqlserver/reader/io/jdbc/rowmapper/provider/package-info.java new file mode 100644 index 0000000000..c01f9f8385 --- /dev/null +++ b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/sqlserver/reader/io/jdbc/rowmapper/provider/package-info.java @@ -0,0 +1,2 @@ +/** SQL Server JDBC value mappings provider. */ +package com.google.cloud.teleport.v2.source.sqlserver.reader.io.jdbc.rowmapper.provider; diff --git a/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/source/SourceConnectorFactoryTest.java b/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/source/SourceConnectorFactoryTest.java index e667b5b80d..07e3bffa56 100644 --- a/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/source/SourceConnectorFactoryTest.java +++ b/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/source/SourceConnectorFactoryTest.java @@ -91,6 +91,18 @@ public void testGetSourceConnectorByDialect_optionsOracle() { com.google.cloud.teleport.v2.source.oracle.OracleSrcToSpSourceConnector.class); } + @Test + public void testGetSourceConnectorByDialect_optionsSqlServer() { + SourceDbToSpannerOptions options = PipelineOptionsFactory.as(SourceDbToSpannerOptions.class); + options.setSourceDbDialect(SourceDbToSpannerOptions.SQLSERVER_SOURCE_DIALECT); + + ISrcToSpSourceConnector connector = SourceConnectorFactory.getSourceConnectorByDialect(options); + + assertThat(connector) + .isInstanceOf( + com.google.cloud.teleport.v2.source.sqlserver.SqlServerSrcToSpSourceConnector.class); + } + @Test public void testGetSourceConnectorByDialect_optionsUnsupportedDialect() { SourceDbToSpannerOptions options = PipelineOptionsFactory.as(SourceDbToSpannerOptions.class); @@ -145,6 +157,16 @@ public void testGetSourceConnectorByDialect_sqlDialectOracle() { com.google.cloud.teleport.v2.source.oracle.OracleSrcToSpSourceConnector.class); } + @Test + public void testGetSourceConnectorByDialect_sqlDialectSqlServer() { + AbstractJdbcSrcToSpSourceConnector connector = + SourceConnectorFactory.getSourceJdbcConnectorByDialect(SQLDialect.SQLSERVER); + + assertThat(connector) + .isInstanceOf( + com.google.cloud.teleport.v2.source.sqlserver.SqlServerSrcToSpSourceConnector.class); + } + @Test public void testGetSourceConnectorByDialect_sqlDialectNull() { IllegalArgumentException thrown = @@ -185,6 +207,15 @@ public void testGetSourceConnectorBySourceType_oracle() { com.google.cloud.teleport.v2.source.oracle.OracleSrcToSpSourceConnector.class); } + @Test + public void testGetSourceConnectorBySourceType_sqlserver() { + ISrcToSpSourceConnector connector = + SourceConnectorFactory.getSourceConnectorBySourceType(Constants.SQLSERVER_SOURCE_TYPE); + assertThat(connector) + .isInstanceOf( + com.google.cloud.teleport.v2.source.sqlserver.SqlServerSrcToSpSourceConnector.class); + } + @Test public void testGetSourceConnectorBySourceType_unsupported() { IllegalArgumentException thrown = diff --git a/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/source/sqlserver/SqlServerSrcToSpSourceConnectorTest.java b/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/source/sqlserver/SqlServerSrcToSpSourceConnectorTest.java new file mode 100644 index 0000000000..2363e71f35 --- /dev/null +++ b/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/source/sqlserver/SqlServerSrcToSpSourceConnectorTest.java @@ -0,0 +1,37 @@ +/* + * 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.source.sqlserver; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import org.junit.Test; + +public class SqlServerSrcToSpSourceConnectorTest { + + @Test + public void testGetTypeMapping() { + SqlServerSrcToSpSourceConnector connector = new SqlServerSrcToSpSourceConnector(); + assertTrue(connector.getTypeMapping().containsKey("INT")); + } + + @Test + public void testGetJdbcUrl() { + SqlServerSrcToSpSourceConnector connector = new SqlServerSrcToSpSourceConnector(); + String url = connector.getJdbcUrl("localhost", 1433, "mydb", "prop1=val1", null, null); + assertEquals("jdbc:sqlserver://localhost:1433;databaseName=mydb;encrypt=false;prop1=val1", url); + } +} diff --git a/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/constants/Constants.java b/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/constants/Constants.java index 6953a3ae79..c0b1672375 100644 --- a/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/constants/Constants.java +++ b/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/constants/Constants.java @@ -30,6 +30,9 @@ public class Constants { /* The value for Postgres databases in the source type key */ public static final String POSTGRES_SOURCE_TYPE = "postgresql"; + /* The value for SQL Server databases in the source type key */ + public static final String SQLSERVER_SOURCE_TYPE = "sqlserver"; + /* The value for Spanner databases in the source type key */ public static final String SPANNER_SOURCE_TYPE = "spanner"; 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..6c28635ab0 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 @@ -95,6 +95,7 @@ public SourceConnectionConfig parseConfiguration( case ORACLE: case MYSQL: case PG: + case SQLSERVER: String jdbcFileContent = FileLoader.readConfigFilePath(sourceConfigFilePath); Map jdbcConfigMap = parseConfigToConfigMap(jdbcFileContent); JdbcShardConfig jdbcShardConfig = mapper.convertValue(jdbcConfigMap, JdbcShardConfig.class); 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;