Repository: cloudspannerecosystem/harbourbridge
Pull request #240 (2021-12-16 12:24:08)

Feature :  Sql server support
#### This PR adds SQL server data and schema migration functionality
Related to https://github.com/cloudspannerecosystem/harbourbridge/issues/251

#### Brief of the changes 
Added new value for the source flag [ -source = sqlserver OR mssql] and implemented ToDdl and common.InfoSchema interface for the SQL server.

- [x] schema migration
- [x] data migration
- [x] Unit test added for each file
- [x] Appropriate changes to README are included in PR

Deep1998: LGTM!
Just one thing to go over, with the current changes we kind of still support sql server through the legacy cmd mode (not sure if it works though). Do we want to add some checks on the allowed drivers for legacy mode or do we go with the stance that new engines are not tested so it is not recommended to use legacy mode for them but we dont error out if they put sql server in the driver?
bharadwaj-aditya: > LGTM! Just one thing to go over, with the current changes we kind of still support sql server through the legacy cmd mode (not sure if it works though). Do we want to add some checks on the allowed drivers for legacy mode or do we go with the stance that new engines are not tested so it is not recommended to use legacy mode for them but we dont error out if they put sql server in the driver?

Let us throw up an error if the driver is manually set which is not one of the previously supported databases - i.e. mysql, postgres or Dynamo

Patch:
diff --git a/.vscode/launch.json b/.vscode/launch.json
index d41e359..c21bb35 100644
--- a/.vscode/launch.json
+++ b/.vscode/launch.json
@@ -11,9 +11,9 @@
             "mode": "auto",
             "program": "${fileDirname}",
             "args": [
-                "-driver=mysqldump", 
-                "-instance=e2e-spanner-testing", 
-                "-v=true", 
+                "-driver=mysqldump",
+                "-instance=e2e-spanner-testing",
+                "-v=true",
                 "-dump-file="
             ]
         }
diff --git a/cmd/cmd.go b/cmd/cmd.go
index 1f4470d..a3509c1 100644
--- a/cmd/cmd.go
+++ b/cmd/cmd.go
@@ -18,6 +18,7 @@ package cmd
 import (
 	"context"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
@@ -40,6 +41,12 @@ var (
 // 3. Run data conversion (if schemaOnly is set to false)
 // 4. Generate report
 func CommandLine(ctx context.Context, driver, targetDb, dbURI string, dataOnly, schemaOnly, skipForeignKeys bool, schemaSampleSize int64, sessionJSON string, ioHelper *utils.IOStreams, outputFilePrefix string, now time.Time) error {
+
+	// Legacy mode is only supported for MySQL, PostgreSQL and DynamoDB
+	if driver != "" && utils.IsValidDriver(driver) && !utils.IsLegacyModeSupportedDriver(driver) {
+		return fmt.Errorf("legacy mode is not supported for drivers other than %s", strings.Join(utils.GetLegacyModeSupportedDrivers(), ", "))
+	}
+
 	var conv *internal.Conv
 	var err error
 	// Creating profiles from legacy flags. We only pass schema-sample-size here because thats the
diff --git a/common/constants/constants.go b/common/constants/constants.go
index 2768c8f..53e70c3 100644
--- a/common/constants/constants.go
+++ b/common/constants/constants.go
@@ -16,6 +16,9 @@ const (
 	// MYSQL is the driver name for MySQL.
 	MYSQL string = "mysql"
 
+	// SQLSERVER is the driver name for sqlserver.
+	SQLSERVER string = "sqlserver"
+
 	// DYNAMODB is the driver name for AWS DynamoDB.
 	// This is an experimental driver; implementation in progress.
 	DYNAMODB string = "dynamodb"
diff --git a/common/utils/utils.go b/common/utils/utils.go
index 7568bec..4ba6456 100644
--- a/common/utils/utils.go
+++ b/common/utils/utils.go
@@ -377,3 +377,41 @@ func SumMapValues(m map[string]int64) int64 {
 func GetBanner(now time.Time, db string) string {
 	return fmt.Sprintf("Generated at %s for db %s\n\n", now.Format("2006-01-02 15:04:05"), db)
 }
+
+func IsValidDriver(driver string) bool {
+	d := strings.ToLower(driver)
+	for _, vd := range GetValidDrivers() {
+		if d == vd {
+			return true
+		}
+	}
+	return false
+}
+
+func GetValidDrivers() []string {
+	//First 5 drivers support legacy mode. Rest dont.
+	return []string{
+		constants.POSTGRES,
+		constants.PGDUMP,
+		constants.MYSQL,
+		constants.MYSQLDUMP,
+		constants.DYNAMODB,
+
+		constants.SQLSERVER,
+	}
+}
+
+func IsLegacyModeSupportedDriver(driver string) bool {
+	d := strings.ToLower(driver)
+	lds := GetLegacyModeSupportedDrivers()
+	for _, ld := range lds {
+		if d == ld {
+			return true
+		}
+	}
+	return false
+}
+
+func GetLegacyModeSupportedDrivers() []string {
+	return GetValidDrivers()[:5]
+}
diff --git a/conversion/conversion.go b/conversion/conversion.go
index ab2f1da..20d0976 100644
--- a/conversion/conversion.go
+++ b/conversion/conversion.go
@@ -52,6 +52,7 @@ import (
 	"github.com/cloudspannerecosystem/harbourbridge/sources/dynamodb"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/mysql"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/postgres"
+	"github.com/cloudspannerecosystem/harbourbridge/sources/sqlserver"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 )
@@ -75,7 +76,7 @@ var (
 // When using source-profile, the sqlConnectionStr is constructed from the input params.
 func SchemaConv(sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile, ioHelper *utils.IOStreams) (*internal.Conv, error) {
 	switch sourceProfile.Driver {
-	case constants.POSTGRES, constants.MYSQL, constants.DYNAMODB:
+	case constants.POSTGRES, constants.MYSQL, constants.DYNAMODB, constants.SQLSERVER:
 		return schemaFromDatabase(sourceProfile, targetProfile)
 	case constants.PGDUMP, constants.MYSQLDUMP:
 		return schemaFromDump(sourceProfile.Driver, targetProfile.TargetDb, ioHelper)
@@ -98,7 +99,7 @@ func DataConv(sourceProfile profiles.SourceProfile, targetProfile profiles.Targe
 		Verbose:    internal.Verbose(),
 	}
 	switch sourceProfile.Driver {
-	case constants.POSTGRES, constants.MYSQL, constants.DYNAMODB:
+	case constants.POSTGRES, constants.MYSQL, constants.DYNAMODB, constants.SQLSERVER:
 		return dataFromDatabase(sourceProfile, config, client, conv)
 	case constants.PGDUMP, constants.MYSQLDUMP:
 		if conv.SpSchema.CheckInterleaved() {
@@ -135,6 +136,8 @@ func connectionConfig(sourceProfile profiles.SourceProfile) (interface{}, error)
 	// For Dynamodb, both legacy and new flows use env vars.
 	case constants.DYNAMODB:
 		return getDynamoDBClientConfig()
+	case constants.SQLSERVER:
+		return profiles.GetSQLConnectionStr(sourceProfile), nil
 	default:
 		return "", fmt.Errorf("driver %s not supported", sourceProfile.Driver)
 	}
@@ -147,6 +150,9 @@ func getDbNameFromSQLConnectionStr(driver, sqlConnectionStr string) string {
 		return strings.Split(dbParam, "=")[1]
 	case constants.MYSQL:
 		return strings.Split(sqlConnectionStr, ")/")[1]
+	case constants.SQLSERVER:
+		splts := strings.Split(sqlConnectionStr, "?database=")
+		return splts[len(splts)-1]
 	}
 	return ""
 }
@@ -728,6 +734,13 @@ func GetInfoSchema(sourceProfile profiles.SourceProfile) (common.InfoSchema, err
 		mySession := session.Must(session.NewSession())
 		dydbClient := dydb.New(mySession, connectionConfig.(*aws.Config))
 		return dynamodb.InfoSchemaImpl{DynamoClient: dydbClient, SampleSize: profiles.GetSchemaSampleSize(sourceProfile)}, nil
+	case constants.SQLSERVER:
+		db, err := sql.Open(driver, connectionConfig.(string))
+		dbName := getDbNameFromSQLConnectionStr(driver, connectionConfig.(string))
+		if err != nil {
+			return nil, err
+		}
+		return sqlserver.InfoSchemaImpl{DbName: dbName, Db: db}, nil
 	default:
 		return nil, fmt.Errorf("driver %s not supported", driver)
 	}
diff --git a/frontend/js/components/ConnectToDbForm/ConnectToDbForm.component.js b/frontend/js/components/ConnectToDbForm/ConnectToDbForm.component.js
index 9f1d725..502ab89 100644
--- a/frontend/js/components/ConnectToDbForm/ConnectToDbForm.component.js
+++ b/frontend/js/components/ConnectToDbForm/ConnectToDbForm.component.js
@@ -8,10 +8,10 @@ class ConnectToDbForm extends HTMLElement {
         let response;
         if (document.getElementById("connectToDbModal")) {
             document.getElementById("connectToDbModal")
-            .querySelector("i")
-            .addEventListener("click", () => {
-                Forms.resetConnectToDbModal();
-            });
+                .querySelector("i")
+                .addEventListener("click", () => {
+                    Forms.resetConnectToDbModal();
+                });
         }
         document.getElementById("db-type").addEventListener("change", () => {
             Forms.toggleDbType();
@@ -54,6 +54,7 @@ class ConnectToDbForm extends HTMLElement {
                     <option value="" class="template"></option>
                     <option value="mysql">MySQL</option>
                     <option value="postgres">Postgres</option>
+                    <option value="sqlserver">SQL Server</option>
                 </select>
             </div>
             <div id="sql-fields" class="template">
diff --git a/frontend/js/components/DataTable/DataTable.component.js b/frontend/js/components/DataTable/DataTable.component.js
index d611658..b9eb568 100644
--- a/frontend/js/components/DataTable/DataTable.component.js
+++ b/frontend/js/components/DataTable/DataTable.component.js
@@ -184,9 +184,9 @@ class DataTable extends HTMLElement {
         let maxPossiblePageNumber =  Math.ceil(tableColumnsArrayLength / columnPerPage);
         let tableColumnsArrayCurrent = [];
         let z = 0;
-        for(let i= pageNumber*columnPerPage ;i < Math.min(tableColumnsArrayLength, pageNumber*columnPerPage+15) ; i++ )
-        {
-            tableColumnsArrayCurrent[z]=tableColumnsArray[i];
+        for (let i = pageNumber * columnPerPage; i < Math.min(tableColumnsArrayLength, pageNumber * columnPerPage + 15); i++) {
+            if (tableColumnsArray[i] != 'synth_id')
+                tableColumnsArrayCurrent[z] = tableColumnsArray[i];
             z++;
         }
 
@@ -253,9 +253,9 @@ class DataTable extends HTMLElement {
                                     </div>
                                 </span>`: `<div></div>`}
                                 <span class="column left">
-                                    ${(currentColumnSrc != srcTable.PrimaryKeys[0].Column || srcTable.PrimaryKeys === null) ?
-                        `<img class="hidden ml-3" src="./Icons/Icons/ic_vpn_key_24px.svg" />` :
-                        `<img class="ml-3" src="./Icons/Icons/ic_vpn_key_24px.svg" />`}
+                                    ${(srcTable.PrimaryKeys !== null && currentColumnSrc === srcTable.PrimaryKeys[0].Column) ?
+                        `<img class="ml-3" src="./Icons/Icons/ic_vpn_key_24px.svg" />` :
+                        `<img class="hidden ml-3" src="./Icons/Icons/ic_vpn_key_24px.svg" />`}
                                 </span>
                                 <span class="column right src-column"
                                     id="src-column-name-${tableIndex}${index}${index}">${currentColumnSrc}</span>
diff --git a/frontend/js/components/EditGlobalDataTypeForm/EditGlobalDataTypeForm.componenet.js b/frontend/js/components/EditGlobalDataTypeForm/EditGlobalDataTypeForm.componenet.js
index 7fed821..9a74e01 100644
--- a/frontend/js/components/EditGlobalDataTypeForm/EditGlobalDataTypeForm.componenet.js
+++ b/frontend/js/components/EditGlobalDataTypeForm/EditGlobalDataTypeForm.componenet.js
@@ -5,9 +5,9 @@ class EditGlobalDataTypeForm extends HTMLElement {
     connectedCallback() {
         this.render();
         document.getElementById("data-type-button")?.addEventListener("click", async () => {
-                await Actions.setGlobalDataType();
-                await Actions.ddlSummaryAndConversionApiCall();
-            });
+            await Actions.setGlobalDataType();
+            await Actions.ddlSummaryAndConversionApiCall();
+        });
     }
 
     render() {
@@ -21,21 +21,23 @@ class EditGlobalDataTypeForm extends HTMLElement {
                         <th>Source</th>
                         <th>Spanner</th>
                     </tr>
-                    ${dataTypeListKeyArray.map((dataType, index) => {return `
+                    ${dataTypeListKeyArray.map((dataType, index) => {
+            return `
                     <tr class='global-data-type-row' id="data-type-row${index + 1}">
                         <td class='src-td' id="data-type-key${index + 1}">${dataType}</td>
                         <td id="data-type-val${index + 1}">
                             <div class="label-container">
                                 <i id="warning${index + 1}" 
-                                class="large material-icons warning ${globalDataTypeList[dataType][0].Brief? "show":"hidden"}" 
+                                class="large material-icons warning ${globalDataTypeList[dataType] ? globalDataTypeList[dataType][0].Brief ? "show" : "hidden" : "hidden"}"
                                  data-toggle="tooltip" data-placement="bottom" 
-                                title="${globalDataTypeList[dataType][0].Brief}">warning</i>
+                                title="${globalDataTypeList[dataType] ? globalDataTypeList[dataType][0].Brief : ""}">warning</i>
                         
                                 <select class='form-control table-select' 
                                 id="data-type-option${index + 1}">
-                                    ${globalDataTypeList[dataType].map((option, idx) => {return `
+                                    ${globalDataTypeList[dataType]?.map((option, idx) => {
+                return `
                                         <option class='data-type-option' value="${option.T}">${option.T}</option>`;
-                                    }).join("")}
+            }).join("")}
                                 </select>
                             </div>
                         </td>
@@ -44,10 +46,10 @@ class EditGlobalDataTypeForm extends HTMLElement {
             </table>
         </div>`;
         tooltipHandler();
-        
+
         for (let i = 0; i < dataTypeListKeyArray.length; i++) {
             document.getElementById(`data-type-option${i + 1}`).addEventListener("change", () => {
-                    Actions.dataTypeUpdate(`data-type-option${i + 1}`, globalDataTypeList);
+                Actions.dataTypeUpdate(`data-type-option${i + 1}`, globalDataTypeList);
             });
         }
     }
@@ -57,4 +59,4 @@ class EditGlobalDataTypeForm extends HTMLElement {
     }
 }
 
-window.customElements.define("hb-edit-global-datatype-form",EditGlobalDataTypeForm);
+window.customElements.define("hb-edit-global-datatype-form", EditGlobalDataTypeForm);
diff --git a/frontend/js/components/TableCarousel/TableCarousel.component.js b/frontend/js/components/TableCarousel/TableCarousel.component.js
index 6eebd08..c34e603 100644
--- a/frontend/js/components/TableCarousel/TableCarousel.component.js
+++ b/frontend/js/components/TableCarousel/TableCarousel.component.js
@@ -25,17 +25,17 @@ class TableCarousel extends HTMLElement {
   }
 
   get stringData() {
-     return this.getAttribute("stringData");
+    return this.getAttribute("stringData");
   }
 
-  set data(value){
+  set data(value) {
     this._data = value;
     this.render();
     this.addEventListenertoCarausal();
-    if(Actions.carouselStatus(this.tabId)[this.tableIndex]){
-      document.querySelector(`hb-data-table[tableName=${this.tableTitle}`).data =this._data; 
+    if (Actions.carouselStatus(this.tabId)[this.tableIndex]) {
+      document.querySelector(`hb-data-table[tableName="${this.tableTitle}"`).data = this._data;
     }
-   
+
   }
 
   get borderData() {
@@ -45,7 +45,7 @@ class TableCarousel extends HTMLElement {
   addEventListenertoCarausal() {
     document.getElementById(`id-${this.tabId}-${this.tableIndex}`).addEventListener('click',()=>{
       Actions.showSpinner()
-      if(Actions.carouselStatus(this.tabId)[this.tableIndex])
+      if(Actions.carouselStatus(this.tabId)[this.tableIndex]) 
       {
         Actions.closeCarousel(this.tabId , this.tableIndex)
       }
@@ -60,11 +60,11 @@ class TableCarousel extends HTMLElement {
     {
       this.render();
       this.addEventListenertoCarausal();
-    } 
+    }
   }
 
   render() {
-    let {tableTitle, tabId, tableIndex, data, borderData, stringData} = this;
+    let { tableTitle, tabId, tableIndex, data, borderData, stringData } = this;
     let color = borderData;
     let panelColor = panelBorderClass(color);
     let cardColor = mdcCardBorder(color);
@@ -72,17 +72,17 @@ class TableCarousel extends HTMLElement {
     let editButtonVisibleClass = carouselStatus ? 'show-content' : 'hide-content';
     let tableMode = Actions.getTableMode(tableIndex);
 
- 
+
     this.innerHTML = `
     <section class="${tabId}-section" id="${tableIndex}">
       <div class="card">
-        <div role="tab" class="card-header ${tabId}-card-header ${carouselStatus ? 'no-border-bottom  rem-border-bottom':''}  ${panelColor}">
+        <div role="tab" class="card-header ${tabId}-card-header ${carouselStatus ? 'no-border-bottom  rem-border-bottom' : ''}  ${panelColor}">
           <h5 class="mb-0">
             <a data-toggle="collapse" id="id-${tabId}-${tableIndex}">
               Table: <span>${tableTitle}</span>
-              <i class="rotate-icon fas fa-angle-${carouselStatus?'up':'down'} "></i>
+              <i class="rotate-icon fas fa-angle-${carouselStatus ? 'up' : 'down'} "></i>
             </a>
-            ${ tabId ==="report" ? `
+            ${tabId === "report" ? `
                 <span class="spanner-text right-align ${editButtonVisibleClass}">Spanner</span>
                 <span class="spanner-icon right-align ${editButtonVisibleClass}">
                   <i class="large material-icons round-icon-size">circle</i>
@@ -92,33 +92,32 @@ class TableCarousel extends HTMLElement {
                   <i class="large material-icons round-icon-size">circle</i>
                 </span>
                 <button class="edit-button ${editButtonVisibleClass}" id="editSpanner${tableIndex}">
-                  ${tableMode?"Save Changes" : " Edit Spanner Schema"}
+                  ${tableMode ? "Save Changes" : " Edit Spanner Schema"}
                 </button>
-                ${tableMode==false && carouselStatus ? `
+                ${tableMode == false && carouselStatus ? `
                   <span id="edit-instruction${tableIndex}" class="right-align edit-instruction blink ">
                   Schema locked for editing. Unlock to change =>
-                </span> `:`<div></div>`}`
-                :
-                ` <div></div> `
-             }
+                </span> `: `<div></div>`}`
+        :
+        ` <div></div> `
+      }
           </h5>
         </div>
-         ${
-           carouselStatus ? `<div class="collapse ${tabId}-collapse show bs collapse show-carausel" id="${tabId}-${tableTitle}">
+         ${carouselStatus ? `<div class="collapse ${tabId}-collapse show bs collapse show-carausel" id="${tabId}-${tableTitle}">
            <div class="mdc-card mdc-card-content table-card-border ${cardColor}">
-             ${ tabId == "report" ? `
-             <hb-data-table tableName="${tableTitle}" tableIndex="${tableIndex}" ></hb-data-table>` 
-             :
-             `<hb-list-table tabName="${tabId}" tableName="${tableTitle}" dta="${stringData}"></hb-list-table>`
-            }
+             ${tabId == "report" ? `
+             <hb-data-table tableName="${tableTitle}" tableIndex="${tableIndex}" ></hb-data-table>`
+          :
+          `<hb-list-table tabName="${tabId}" tableName="${tableTitle}" dta="${stringData}"></hb-list-table>`
+        }
            </div>
-         </div>`:`<div></div>`
-         }
+         </div>`: `<div></div>`
+      }
         
 
       </div>
     </section> `;
-   
+
   }
 
   constructor() {
diff --git a/frontend/js/pages/SchemaConversionScreen/SchemaConversionScreen.page.js b/frontend/js/pages/SchemaConversionScreen/SchemaConversionScreen.page.js
index 3620b8b..cb26966 100644
--- a/frontend/js/pages/SchemaConversionScreen/SchemaConversionScreen.page.js
+++ b/frontend/js/pages/SchemaConversionScreen/SchemaConversionScreen.page.js
@@ -16,81 +16,81 @@ import "../../services/Fetch.service.js";
 
 class SchemaConversionScreen extends HTMLElement {
 
-    get testing() {
-        return this.getAttribute("testing");
-    }
+  get testing() {
+    return this.getAttribute("testing");
+  }
 
-    connectedCallback() {
-        this.stateObserver = setInterval(this.observeState, 150);
-        Actions.showSpinner()
-        this.render();
-    }
+  connectedCallback() {
+    this.stateObserver = setInterval(this.observeState, 150);
+    Actions.showSpinner()
+    this.render();
+  }
+
+  disconnectedCallback() {
+    clearInterval(this.stateObserver);
+  }
 
-    disconnectedCallback() {
-        clearInterval(this.stateObserver);
+  sendDatatoReportTab(tableNameArray, currentTabContent) {
+    for (let i = 0; i < tableNameArray.length; i++) {
+      let filterdata = {
+        SpSchema: currentTabContent.SpSchema[tableNameArray[i]],
+        SrcSchema: currentTabContent.SrcSchema[currentTabContent.ToSource[tableNameArray[i]].Name],
+        ToSource: currentTabContent.ToSource[tableNameArray[i]],
+        ToSpanner: currentTabContent.ToSpanner[currentTabContent.ToSource[tableNameArray[i]].Name],
+        summary: Store.getinstance().tableData["summaryTabContent"][tableNameArray[i]],
+        currentPageNumber: Actions.getCurrentPageNumber(i)
+      };
+      let component = document.querySelector(`#reportTab${i}`);
+      component.data = filterdata;
     }
+  }
+
+  set Data(data) {
+    this.data = data;
+    this.render();
+  }
 
-    sendDatatoReportTab(tableNameArray, currentTabContent) {
-        for (let i = 0; i < tableNameArray.length; i++) {
-            let filterdata = {
-                SpSchema: currentTabContent.SpSchema[tableNameArray[i]],
-                SrcSchema: currentTabContent.SrcSchema[tableNameArray[i]],
-                ToSource: currentTabContent.ToSource[tableNameArray[i]],
-                ToSpanner: currentTabContent.ToSpanner[tableNameArray[i]],
-                summary: Store.getinstance().tableData["summaryTabContent"][tableNameArray[i]],
-                currentPageNumber: Actions.getCurrentPageNumber(i)
-            };
-            let component = document.querySelector(`#reportTab${i}`);
-            component.data = filterdata;
-        }
+  getChangingValue(currentTab) {
+    currentTab = currentTab.substring(0, currentTab.length - 3);
+    let currentArray = Actions.carouselStatus(currentTab);
+    let flag = "Expand All";
+    for (let i = 0; i < currentArray.length; i++) {
+      if (currentArray[i] == true) {
+        flag = "Collapse All";
+      }
     }
+    return flag;
+  }
 
-    set Data(data) {
-        this.data = data;
-        this.render();
+  observeState = () => {
+    let updatedData = Store.getinstance();
+    if (JSON.stringify(updatedData) !== JSON.stringify(this.data)) {
+      this.data = JSON.parse(JSON.stringify(updatedData));
+      this.render();
     }
+  };
 
-    getChangingValue(currentTab) {
-        currentTab = currentTab.substring(0, currentTab.length - 3);
-        let currentArray = Actions.carouselStatus(currentTab);
-        let flag = "Expand All";
-        for (let i = 0; i < currentArray.length; i++) {
-            if (currentArray[i] == true) {
-                flag = "Collapse All";
-            }
-        }
-        return flag;
+  render() {
+    if (!this.data) {
+      return;
+    }
+    const { currentTab, tableData, tableBorderData, searchInputValue } = this.data;
+    let currentTabContent = tableData[`${currentTab}Content`];
+    if (Object.keys(currentTabContent).length == 0) {
+      Actions.hideSpinner();
+      return;
+    }
+    const changingText = this.getChangingValue(currentTab);
+    let tableNameArray;
+    if (currentTab === "reportTab") {
+      tableNameArray = Object.keys(currentTabContent.SpSchema)
+        .filter((title) => title.indexOf(searchInputValue[currentTab]) > -1);
+    } else {
+      tableNameArray = Object.keys(currentTabContent)
+        .filter((title) => title.indexOf(searchInputValue[currentTab]) > -1);
     }
 
-    observeState = () => {
-        let updatedData = Store.getinstance();
-        if (JSON.stringify(updatedData) !== JSON.stringify(this.data)) {
-            this.data = JSON.parse(JSON.stringify(updatedData));
-            this.render();
-        }
-    };
-
-    render() {
-        if (!this.data) {
-            return;
-        }
-        const { currentTab, tableData, tableBorderData, searchInputValue } = this.data;
-        let currentTabContent = tableData[`${currentTab}Content`];
-        if (Object.keys(currentTabContent).length == 0) {
-            Actions.hideSpinner();
-            return;
-        }
-        const changingText = this.getChangingValue(currentTab);
-        let tableNameArray;
-        if (currentTab === "reportTab") {
-            tableNameArray = Object.keys(currentTabContent.SpSchema)
-                .filter((title) => title.indexOf(searchInputValue[currentTab]) > -1);
-        } else {
-            tableNameArray = Object.keys(currentTabContent)
-                .filter((title) => title.indexOf(searchInputValue[currentTab]) > -1);
-        }
-
-        this.innerHTML = `
+    this.innerHTML = `
     <div class="summary-main-content" id='schema-screen-content'>
       <div id="snackbar"></div>
       <div>
@@ -110,61 +110,59 @@ class SchemaConversionScreen extends HTMLElement {
         <div class="tab-content">
           ${currentTab === 'reportTab' ? `<div id="report" class="tab-pane fade show active">
             <div class="accordion md-accordion" id="accordion" role="tablist" aria-multiselectable="true">
-            ${tableNameArray.length > 0 ? `<hb-site-button buttonid="reportExpandButton" classname="expand" buttonaction="expandAll" text="${changingText}"></hb-site-button>`:''}
-            ${tableNameArray.length > 0 ? `<hb-site-button buttonid="editButton" classname="expand right-align" buttonaction="editGlobalDataType" text="Edit Global Data Type"></hb-site-button>` :''}
+            ${tableNameArray.length > 0 ? `<hb-site-button buttonid="reportExpandButton" classname="expand" buttonaction="expandAll" text="${changingText}"></hb-site-button>` : ''}
+            ${tableNameArray.length > 0 ? `<hb-site-button buttonid="editButton" classname="expand right-align" buttonaction="editGlobalDataType" text="Edit Global Data Type"></hb-site-button>` : ''}
               <div id='reportDiv'>
                 ${tableNameArray.map((tableName, index) => {
-                    return `
+      return `
                     <hb-table-carousel tableTitle="${tableName}" id="${currentTab}${index}" tabId="report" 
                     tableIndex="${index}" borderData = "${tableBorderData[tableName]}"></hb-table-carousel>`;
-                  })
-                  .join("")}                    
+    })
+          .join("")}                    
                 </div>
             </div>
-            ${tableNameArray.length <=0 ? '<h5 class="no-text" >No Match Found</h5>':''}
+            ${tableNameArray.length <= 0 ? '<h5 class="no-text" >No Match Found</h5>' : ''}
           </div>`
-              : `<div></div>`
-          }
+        : `<div></div>`
+      }
 
-          ${
-            currentTab === "ddlTab"
-              ? `<div id="ddl" class="tab-pane fade show active">
+          ${currentTab === "ddlTab"
+        ? `<div id="ddl" class="tab-pane fade show active">
             <div class="panel-group" id="ddl-accordion">
-            ${tableNameArray.length > 0 ? `<hb-site-button buttonid="ddlExpandButton" classname="expand" buttonaction="expandAll" text="${changingText}"></hb-site-button>`:''}
-            ${tableNameArray.length > 0 ?`<hb-site-button buttonid="download-ddl" classname="expand right-align" buttonaction="downloadDdl" text="Download DDL Statements"></hb-site-button>`:''}
+            ${tableNameArray.length > 0 ? `<hb-site-button buttonid="ddlExpandButton" classname="expand" buttonaction="expandAll" text="${changingText}"></hb-site-button>` : ''}
+            ${tableNameArray.length > 0 ? `<hb-site-button buttonid="download-ddl" classname="expand right-align" buttonaction="downloadDdl" text="Download DDL Statements"></hb-site-button>` : ''}
               <div id='ddlDiv'>
                 ${tableNameArray.map((tableName, index) => {
-                    return `
+          return `
                     <hb-table-carousel tableTitle="${tableName}" stringData="${currentTabContent[tableName]}" tabId="ddl" id="${currentTab}${index}" tableIndex=${index} borderData = "${tableBorderData[tableName]}">
                     </hb-table-carousel>`;
-                  })
-                  .join("")} 
+        })
+          .join("")} 
                 </div>
             </div>
-           ${tableNameArray.length <=0 ? '<h5 class="no-text" >No Match Found</h5>':''}
+           ${tableNameArray.length <= 0 ? '<h5 class="no-text" >No Match Found</h5>' : ''}
           </div>`
-              : `<div></div>`
-          }
+        : `<div></div>`
+      }
 
-          ${
-            currentTab === "summaryTab"
-              ? `<div id="summary" class="tab-pane fade show active">
+          ${currentTab === "summaryTab"
+        ? `<div id="summary" class="tab-pane fade show active">
             <div class="panel-group" id="summary-accordion">
-            ${tableNameArray.length > 0 ?`<hb-site-button buttonid="summaryExpandButton" classname="expand" buttonaction="expandAll" text="${changingText}"></hb-site-button>`:''}
-            ${tableNameArray.length > 0 ?`<hb-site-button buttonid="download-report" classname="expand right-align" buttonaction="downloadReport" text="Download Summary Report"></hb-site-button>`:''}
+            ${tableNameArray.length > 0 ? `<hb-site-button buttonid="summaryExpandButton" classname="expand" buttonaction="expandAll" text="${changingText}"></hb-site-button>` : ''}
+            ${tableNameArray.length > 0 ? `<hb-site-button buttonid="download-report" classname="expand right-align" buttonaction="downloadReport" text="Download Summary Report"></hb-site-button>` : ''}
               <div id='summaryDiv'>
                 ${tableNameArray.map((tableName, index) => {
-                    return `
+          return `
                     <hb-table-carousel  tableTitle="${tableName}" stringData="${currentTabContent[tableName]}" id="${currentTab}${index}" tabId="summary" 
                     tableIndex=${index} borderData = "${tableBorderData[tableName]}"></hb-table-carousel>`;
-                  })
-                  .join("")} 
+        })
+          .join("")} 
               </div>
             </div>
-            ${tableNameArray.length <=0 ? '<h5 class="no-text">No Match Found</h5>':''}
+            ${tableNameArray.length <= 0 ? '<h5 class="no-text">No Match Found</h5>' : ''}
           </div>`
-              : `<div></div>`
-          }
+        : `<div></div>`
+      }
 
           </div>
       </div>
@@ -181,10 +179,10 @@ class SchemaConversionScreen extends HTMLElement {
     initSchemaScreenTasks();
     if (currentTab === "reportTab" && !this.testing) {
       this.sendDatatoReportTab(tableNameArray
-        .filter((title)=>title.indexOf(searchInputValue[currentTab]) > -1), currentTabContent);
-       if(!this.testing){ 
-          window.scrollTo(0,Actions.getPageYOffset());
-       }
+        .filter((title) => title.indexOf(searchInputValue[currentTab]) > -1), currentTabContent);
+      if (!this.testing) {
+        window.scrollTo(0, Actions.getPageYOffset());
+      }
     }
     Actions.hideSpinner();
   }
diff --git a/frontend/js/services/Action.service.js b/frontend/js/services/Action.service.js
index ad0b395..8c6ef9b 100644
--- a/frontend/js/services/Action.service.js
+++ b/frontend/js/services/Action.service.js
@@ -51,7 +51,7 @@ const Actions = (() => {
             tableData.data = {};
         },
 
-        onLoadDatabase: async(dbType, dumpFilePath) => {
+        onLoadDatabase: async (dbType, dumpFilePath) => {
             let reportData, sourceTableFlag, reportDataResp, reportDataCopy, jsonReportDataResp, requestCode;
             reportData = await Fetch.getAppData("POST", "/convert/dump", { Driver: dbType, Path: dumpFilePath });
             reportDataCopy = reportData.clone();
@@ -78,7 +78,7 @@ const Actions = (() => {
             return true;
         },
 
-        onconnect: async(dbType, dbHost, dbPort, dbUser, dbName, dbPassword) => {
+        onconnect: async (dbType, dbHost, dbPort, dbUser, dbName, dbPassword) => {
             let sourceTableFlag = "",
                 response;
             let payload = { Driver: dbType, Database: dbName, Password: dbPassword, User: dbUser, Port: dbPort, Host: dbHost };
@@ -86,6 +86,7 @@ const Actions = (() => {
             if (response.ok) {
                 if (dbType === "mysql") sourceTableFlag = "MySQL";
                 else if (dbType === "postgres") sourceTableFlag = "Postgres";
+                else if (dbType === 'sqlserver') sourceTableFlag = "SQL Server";
                 Store.setSourceDbName(sourceTableFlag)
                 jQuery("#connectToDbModal").modal("hide");
                 jQuery("#connectModalSuccess").modal();
@@ -96,7 +97,7 @@ const Actions = (() => {
             return response;
         },
 
-        showSchemaAssessment: async() => {
+        showSchemaAssessment: async () => {
             let reportDataResp, reportData, sourceTableFlag;
             reportData = await Fetch.getAppData("GET", "/convert/infoschema");
             reportDataResp = await reportData.json();
@@ -106,7 +107,7 @@ const Actions = (() => {
             jQuery("#connectModalSuccess").modal("hide");
         },
 
-        onLoadSessionFile: async(filePath) => {
+        onLoadSessionFile: async (filePath) => {
             let driver = '',
                 response, payload;
             let srcDb = Store.getSourceDbName()
@@ -139,7 +140,7 @@ const Actions = (() => {
             }
         },
 
-        ddlSummaryAndConversionApiCall: async() => {
+        ddlSummaryAndConversionApiCall: async () => {
             let conversionRate, conversionRateJson, ddlData, ddlDataJson, summaryData, summaryDataJson;
             ddlData = await Fetch.getAppData("GET", "/ddl");
             summaryData = await Fetch.getAppData("GET", "/summary");
@@ -157,7 +158,7 @@ const Actions = (() => {
             return true;
         },
 
-        sessionRetrieval: async(dbType) => {
+        sessionRetrieval: async (dbType) => {
             let sessionStorageArr, sessionInfo, sessionResp;
             sessionResp = await Fetch.getAppData("GET", "/session");
             sessionInfo = await sessionResp.json();
@@ -169,7 +170,7 @@ const Actions = (() => {
             sessionStorage.setItem('currentSessionIdx', 0)
         },
 
-        resumeSessionHandler: async(index, sessionArray) => {
+        resumeSessionHandler: async (index, sessionArray) => {
             Actions.showSpinner()
             let driver, path, dbName, sourceDb, pathArray, fileName, filePath;
             Store.setSourceDbName(sessionArray[index].sourceDbType)
@@ -180,7 +181,7 @@ const Actions = (() => {
             pathArray = path.split("/");
             fileName = pathArray[pathArray.length - 1];
             filePath = "./" + fileName;
-            readTextFile(filePath, async(error, text) => {
+            readTextFile(filePath, async (error, text) => {
                 if (error) {
                     let storage = JSON.parse(sessionStorage.getItem('sessionStorage'))
                     storage.splice(index, 1);
@@ -217,31 +218,31 @@ const Actions = (() => {
             }
         },
 
-        downloadSession: async() => {
+        downloadSession: async () => {
             let reportJsonObj = JSON.stringify(Store.getinstance().tableData.reportTabContent);
             reportJsonObj = reportJsonObj.replaceAll("9223372036854776000", "9223372036854775807");
             jQuery("<a />", {
-                    download: "session.json",
-                    href: "data:application/json;charset=utf-8," + encodeURIComponent(reportJsonObj, null, 4),
-                }).appendTo("body").click(function() { jQuery(this).remove(); })[0]
+                download: "session.json",
+                href: "data:application/json;charset=utf-8," + encodeURIComponent(reportJsonObj, null, 4),
+            }).appendTo("body").click(function () { jQuery(this).remove(); })[0]
                 .click();
         },
 
-        downloadDdl: async() => {
+        downloadDdl: async () => {
             let ddlreport = await Fetch.getAppData("GET", "/schema");
             if (ddlreport.ok) {
-                await ddlreport.text().then(function(result) {
+                await ddlreport.text().then(function (result) {
                     localStorage.setItem("schemaFilePath", result);
                 });
                 let schemaFilePath = localStorage.getItem("schemaFilePath");
                 if (schemaFilePath) {
                     let schemaFileName = schemaFilePath.split("/")[schemaFilePath.split("/").length - 1];
                     let filePath = "./" + schemaFileName;
-                    readTextFile(filePath, function(error, text) {
+                    readTextFile(filePath, function (error, text) {
                         jQuery("<a />", {
-                                download: schemaFileName,
-                                href: "data:application/json;charset=utf-8," + encodeURIComponent(text),
-                            }).appendTo("body").click(function() { jQuery(this).remove(); })[0]
+                            download: schemaFileName,
+                            href: "data:application/json;charset=utf-8," + encodeURIComponent(text),
+                        }).appendTo("body").click(function () { jQuery(this).remove(); })[0]
                             .click();
                     });
                 }
@@ -249,20 +250,20 @@ const Actions = (() => {
             }
         },
 
-        downloadReport: async() => {
+        downloadReport: async () => {
             let summaryreport = await Fetch.getAppData("GET", "/report");
             if (summaryreport.ok) {
-                await summaryreport.text().then(function(result) {
+                await summaryreport.text().then(function (result) {
                     localStorage.setItem("reportFilePath", result);
                 });
                 let reportFilePath = localStorage.getItem("reportFilePath");
                 let reportFileName = reportFilePath.split("/")[reportFilePath.split("/").length - 1];
                 let filePath = "./" + reportFileName;
-                readTextFile(filePath, function(error, text) {
+                readTextFile(filePath, function (error, text) {
                     jQuery("<a />", {
-                            download: reportFileName,
-                            href: "data:application/json;charset=utf-8," + encodeURIComponent(text),
-                        }).appendTo("body").click(function() { jQuery(this).remove(); })[0]
+                        download: reportFileName,
+                        href: "data:application/json;charset=utf-8," + encodeURIComponent(text),
+                    }).appendTo("body").click(function () { jQuery(this).remove(); })[0]
                         .click();
                 });
             }
@@ -272,7 +273,7 @@ const Actions = (() => {
             jQuery("#globalDataTypeModal").modal();
         },
 
-        checkInterleaveConversion: async(tableName) => {
+        checkInterleaveConversion: async (tableName) => {
             let interleaveApiCall;
             interleaveApiCall = await Fetch.getAppData("GET", "/setparent?table=" + tableName);
             let interleaveApiCallResp = await interleaveApiCall.json();
@@ -280,7 +281,7 @@ const Actions = (() => {
             Store.setInterleave(tableName, value);
         },
 
-        setGlobalDataType: async() => {
+        setGlobalDataType: async () => {
             Actions.showSpinner()
             let globalDataTypeList = Store.getGlobalDataTypeList();
             let dataTypeListLength = Object.keys(globalDataTypeList).length;
@@ -313,7 +314,7 @@ const Actions = (() => {
             }
         },
 
-        setGlobalDataTypeList: async() => {
+        setGlobalDataTypeList: async () => {
             let res = await Fetch.getAppData("GET", "/typemap");
             if (res) {
                 let result = await res.json();
@@ -340,7 +341,7 @@ const Actions = (() => {
             }
         },
 
-        fetchIndexFormValues: async(tableIndex, tableName, name, uniqueness) => {
+        fetchIndexFormValues: async (tableIndex, tableName, name, uniqueness) => {
             Actions.showSpinner()
             if (keysList.length === 0) {
                 Actions.hideSpinner()
@@ -397,7 +398,7 @@ const Actions = (() => {
             if (document.getElementById("editSpanner" + tableIndex).innerHTML.trim() == "Save Changes") {
                 let pendingChanges = false;
                 let dataTable = jQuery(`#src-sp-table${tableIndex} tr`)
-                dataTable.each(function(index) {
+                dataTable.each(function (index) {
                     if (index > 1) {
                         let newColumnName;
                         let srcColumnName = jQuery(this).find('.src-column').html().trim();
@@ -409,7 +410,7 @@ const Actions = (() => {
                         if (newColumnNameEle) {
                             newColumnName = newColumnNameEle.value;
                         }
-                        let oldColumnName = jsonObj.ToSpanner[tableName].Cols[srcColumnName];
+                        let oldColumnName = jsonObj.ToSpanner[jsonObj.ToSource[tableName].Name].Cols[srcColumnName];
                         if (newColumnName !== oldColumnName || !(jQuery(this).find("input[type=checkbox]").is(":checked"))) {
                             let errorModal = document.querySelector("hb-modal[modalId = editTableWarningModal]");
                             let content = "There are pending changes to this table, please save the same before creating the index";
@@ -465,9 +466,9 @@ const Actions = (() => {
             }
         },
 
-        SaveButtonHandler: async(tableNumber, tableName, notNullConstraint) => {
+        SaveButtonHandler: async (tableNumber, tableName, notNullConstraint) => {
             var errorMessage = [];
-            let schemaConversionObj = {...Store.getinstance().tableData.reportTabContent };
+            let schemaConversionObj = { ...Store.getinstance().tableData.reportTabContent };
             let fkStatus = false,
                 secIndexStatus = false,
                 columnStatus = false;
@@ -505,7 +506,7 @@ const Actions = (() => {
                 }
             }
             let flagofcheckbox = false;
-            jQuery(tableId).each(function(index) {
+            jQuery(tableId).each(function (index) {
                 if (!(jQuery(this).find("input[type=checkbox]").is(":checked"))) {
                     flagofcheckbox = true;
                     return false;
@@ -517,15 +518,15 @@ const Actions = (() => {
             return false;
         },
 
-        saveColumn: async(schemaConversionObj, tableNumber, tableName, notNullConstraint, tableData, errorMessage, updateInStore = false) => {
+        saveColumn: async (schemaConversionObj, tableNumber, tableName, notNullConstraint, tableData, errorMessage, updateInStore = false) => {
             let data;
             if (tableData.data.SpSchema != undefined) {
-                data = {...tableData.data };
+                data = { ...tableData.data };
             } else {
                 if (updateInStore) {
-                    data = {...Store.getinstance().tableData.reportTabContent };
+                    data = { ...Store.getinstance().tableData.reportTabContent };
                 } else {
-                    data = {...schemaConversionObj };
+                    data = { ...schemaConversionObj };
                 }
             }
 
@@ -547,16 +548,16 @@ const Actions = (() => {
             let columnsNamesArray = [];
             // We follow this approach to create columnsNamesArray to ensure the order of column names in the array is the consistent with what is seen in the UI.
             // This is needed later on where we compare the array index with tableColumnNumber.
-            jQuery(tableId).each(function(index) {
+            jQuery(tableId).each(function (index) {
                 if (index > 1) {
                     let srcColumnName = document.getElementById('src-column-name-' + tableNumber + tableColumnNumber + tableColumnNumber).innerHTML;
-                    let originalColumnName = data.ToSpanner[tableName].Cols[srcColumnName];
+                    let originalColumnName = data.ToSpanner[data.ToSource[tableName].Name].Cols[srcColumnName];
                     columnsNamesArray.push(originalColumnName)
                     tableColumnNumber++;
                 }
             });
             tableColumnNumber = 0;
-            jQuery(tableId).each(function(index) {
+            jQuery(tableId).each(function (index) {
                 if (index > 1) {
                     let newColumnName;
                     let srcColumnName = document.getElementById('src-column-name-' + tableNumber + tableColumnNumber + tableColumnNumber).innerHTML;
@@ -565,7 +566,7 @@ const Actions = (() => {
                         newColumnName = newColumnNameEle.value;
                         newColArrayForDuplicateCheck.push(newColumnName)
                     }
-                    let originalColumnName = data.ToSpanner[tableName].Cols[srcColumnName];
+                    let originalColumnName = data.ToSpanner[data.ToSource[tableName].Name].Cols[srcColumnName];
                     updatedColsData.UpdateCols[originalColumnName] = {};
                     updatedColsData.UpdateCols[originalColumnName]['Removed'] = false;
                     if (newColumnName === originalColumnName) {
@@ -659,15 +660,15 @@ const Actions = (() => {
             return true;
         },
 
-        saveForeignKeys: async(schemaConversionObj, tableNumber, tableName, tableData, errorMessage) => {
+        saveForeignKeys: async (schemaConversionObj, tableNumber, tableName, tableData, errorMessage) => {
             let fkTableData, renameFkMap = {},
                 fkLength;
             let uniquevals;
             let newFkValueArray = [];
             let data;
             if (tableData.data.SpSchema != undefined) {
-                data = {...tableData.data };
-            } else data = {...schemaConversionObj };
+                data = { ...tableData.data };
+            } else data = { ...schemaConversionObj };
             if (data.SpSchema[tableName].Fks != null && data.SpSchema[tableName].Fks.length != 0) {
                 fkLength = data.SpSchema[tableName].Fks.length;
                 for (let x = 0; x < fkLength; x++) {
@@ -683,7 +684,7 @@ const Actions = (() => {
                     let keys = Object.keys(renameFkMap);
                     let flag = false;
                     let dummyobj = {};
-                    keys.forEach(function(key) {
+                    keys.forEach(function (key) {
 
                         for (let x = 0; x < fkLength; x++) {
                             if (data.SpSchema[tableName].Fks[x].Name === renameFkMap[key]) {
@@ -708,12 +709,12 @@ const Actions = (() => {
                     });
                     if (flag) {
                         let dummyrenameFkMap = {};
-                        keys.forEach(function(key) {
+                        keys.forEach(function (key) {
                             dummyobj[key] = new Date().toString() + key;
                         });
                         fkTableData = await Fetch.getAppData('POST', '/rename/fks?table=' + tableName, dummyobj);
                         if (fkTableData.ok) {
-                            keys.forEach(function(key) {
+                            keys.forEach(function (key) {
                                 dummyrenameFkMap[dummyobj[key]] = renameFkMap[key];
                             })
                         }
@@ -739,7 +740,7 @@ const Actions = (() => {
             return true;
         },
 
-        saveSecondaryIndexes: async(schemaConversionObj, tableNumber, tableName, tableData, errorMessage) => {
+        saveSecondaryIndexes: async (schemaConversionObj, tableNumber, tableName, tableData, errorMessage) => {
             let data;
             let newSecIndexArray = [];
             let uniquevals;
@@ -747,8 +748,8 @@ const Actions = (() => {
                 secIndexLength;
 
             if (tableData.data.SpSchema != undefined) {
-                data = {...tableData.data };
-            } else data = {...schemaConversionObj };
+                data = { ...tableData.data };
+            } else data = { ...schemaConversionObj };
 
             if (data.SpSchema[tableName].Indexes != null && data.SpSchema[tableName].Indexes.length != 0) {
                 secIndexLength = data.SpSchema[tableName].Indexes.length;
@@ -764,7 +765,7 @@ const Actions = (() => {
                     let duplicateFound = false;
                     let keys = Object.keys(renameIndexMap);
                     let flag = false;
-                    keys.forEach(function(key) {
+                    keys.forEach(function (key) {
                         for (let x = 0; x < secIndexLength; x++) {
                             if (data.SpSchema[tableName].Indexes[x].Name === renameIndexMap[key]) {
                                 if (uniquevals.length == newSecIndexArray.length) {
@@ -789,12 +790,12 @@ const Actions = (() => {
                     if (flag) {
                         let dummyobj = {};
                         let dummyrenameSecIndexMap = {};
-                        keys.forEach(function(key) {
+                        keys.forEach(function (key) {
                             dummyobj[key] = new Date().toString() + key;
                         });
                         secIndexTableData = await Fetch.getAppData('POST', '/rename/indexes?table=' + tableName, dummyobj);
                         if (secIndexTableData.ok) {
-                            keys.forEach(function(key) {
+                            keys.forEach(function (key) {
                                 dummyrenameSecIndexMap[dummyobj[key]] = renameIndexMap[key];
                             })
                         }
@@ -820,7 +821,7 @@ const Actions = (() => {
             return true;
         },
 
-        dropForeignKeyHandler: async(tableName, tableNumber, pos) => {
+        dropForeignKeyHandler: async (tableName, tableNumber, pos) => {
             let response;
             Actions.showSpinner();
             response = await Fetch.getAppData('GET', '/drop/fk?table=' + tableName + '&pos=' + pos);
@@ -837,7 +838,7 @@ const Actions = (() => {
             }
         },
 
-        dropSecondaryIndexHandler: async(tableName, tableNumber, pos) => {
+        dropSecondaryIndexHandler: async (tableName, tableNumber, pos) => {
             Actions.showSpinner()
             let response;
             response = await Fetch.getAppData('GET', '/drop/secondaryindex?table=' + tableName + '&pos=' + pos);
diff --git a/frontend/js/services/Forms.service.js b/frontend/js/services/Forms.service.js
index fec5e4a..6f40fa2 100644
--- a/frontend/js/services/Forms.service.js
+++ b/frontend/js/services/Forms.service.js
@@ -38,6 +38,12 @@ const Forms = (() => {
         document.getElementById("sql-fields").style.display = "block";
         sourceTableFlag = "Postgres";
         Actions.setSourceDbName(sourceTableFlag);
+      } else if (val.value === "sqlserver") {
+        jQuery(".form-error").html("");
+        jQuery(".db-input").val("");
+        document.getElementById("sql-fields").style.display = "block";
+        sourceTableFlag = "SQL Server";
+        Actions.setSourceDbName(sourceTableFlag);
       }
       else if (val.value === "dynamodb") {
         document.getElementById("sql-fields").style.display = "none";
diff --git a/go.mod b/go.mod
index 073cf3f..1cff300 100644
--- a/go.mod
+++ b/go.mod
@@ -8,8 +8,10 @@ require (
 	cloud.google.com/go/storage v1.17.0
 	github.com/DATA-DOG/go-sqlmock v1.4.1
 	github.com/aws/aws-sdk-go v1.34.5
+	github.com/denisenkom/go-mssqldb v0.11.0 // indirect
 	github.com/form3tech-oss/jwt-go v3.2.5+incompatible // indirect
 	github.com/go-sql-driver/mysql v1.5.0
+	github.com/golang-sql/civil v0.0.0-20190719163853-cb61b32ac6fe // indirect
 	github.com/google/go-cmp v0.5.6
 	github.com/google/subcommands v1.2.0
 	github.com/gorilla/handlers v1.5.0
diff --git a/go.sum b/go.sum
index f5c60c7..83aa8e0 100644
--- a/go.sum
+++ b/go.sum
@@ -76,6 +76,8 @@ github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/denisenkom/go-mssqldb v0.0.0-20191124224453-732737034ffd/go.mod h1:xbL0rPBG9cCiLr28tMa8zpbdarY27NDyej4t/EjAShU=
+github.com/denisenkom/go-mssqldb v0.11.0 h1:9rHa233rhdOyrz2GcP9NM+gi2psgJZ4GWDpL/7ND8HI=
+github.com/denisenkom/go-mssqldb v0.11.0/go.mod h1:xbL0rPBG9cCiLr28tMa8zpbdarY27NDyej4t/EjAShU=
 github.com/dgraph-io/ristretto v0.0.1 h1:cJwdnj42uV8Jg4+KLrYovLiCgIfz9wtWm6E6KA+1tLs=
 github.com/dgraph-io/ristretto v0.0.1/go.mod h1:T40EBc7CJke8TkpiYfGGKAeFjSaxuFXhuXRyumBd6RE=
 github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2 h1:tdlZCpZ/P9DhczCTSixgIKmwPv6+wP5DGjqLYw5SUiA=
@@ -149,6 +151,7 @@ github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7a
 github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zVXpSg4=
 github.com/gogo/protobuf v1.3.1 h1:DqDEcV5aeaTmdFBePNpYsp3FlcVH/2ISVVM9Qf8PSls=
 github.com/gogo/protobuf v1.3.1/go.mod h1:SlYgWuQ5SjCEi6WLHjHCa1yvBfUnHcTbrrZtXPKa29o=
+github.com/golang-sql/civil v0.0.0-20190719163853-cb61b32ac6fe h1:lXe2qZdvpiX5WZkZR4hgp4KJVfY3nMkvmwbVkpv1rVY=
 github.com/golang-sql/civil v0.0.0-20190719163853-cb61b32ac6fe/go.mod h1:8vg3r2VgvsThLBIFL93Qb5yWzgyZWhEmBwUJWevAkK0=
 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
 github.com/golang/groupcache v0.0.0-20160516000752-02826c3e7903/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
diff --git a/internal/convert.go b/internal/convert.go
index 814356e..d8071f2 100644
--- a/internal/convert.go
+++ b/internal/convert.go
@@ -79,6 +79,7 @@ const (
 	Datetime
 	Widened
 	Time
+	StringOverflow
 )
 
 // NameAndCols contains the name of a table and its columns.
diff --git a/internal/report.go b/internal/report.go
index c44c1fb..b3fc4f3 100644
--- a/internal/report.go
+++ b/internal/report.go
@@ -275,6 +275,7 @@ var IssueDB = map[SchemaIssue]struct {
 	Datetime:              {Brief: "Spanner timestamp is closer to MySQL timestamp", severity: note, batch: true},
 	Time:                  {Brief: "Spanner does not support time/year types", severity: note, batch: true},
 	Widened:               {Brief: "Some columns will consume more storage in Spanner", severity: note, batch: true},
+	StringOverflow:        {Brief: "String overflow issue might occur as maximum supported length in Spanner is 2621440", severity: warning},
 }
 
 type severity int
diff --git a/main.go b/main.go
index 1f90c49..9a2aa45 100644
--- a/main.go
+++ b/main.go
@@ -25,6 +25,7 @@ import (
 	"strings"
 	"time"
 
+	_ "github.com/denisenkom/go-mssqldb"
 	_ "github.com/go-sql-driver/mysql"
 	_ "github.com/lib/pq"
 
diff --git a/profiles/common.go b/profiles/common.go
index c6cf1bd..c700534 100644
--- a/profiles/common.go
+++ b/profiles/common.go
@@ -93,6 +93,9 @@ func GetSQLConnectionStr(sourceProfile SourceProfile) string {
 			// Thus, there is no need to create sqlConnectionStr for the same. We instead set the env variables
 			// programmatically if not set.
 			return ""
+		case SourceProfileConnectionTypeSqlServer:
+			connParams := sourceProfile.Conn.SqlServer
+			return getSQLSERVERConnectionStr(connParams.Host, connParams.Port, connParams.User, connParams.Pwd, connParams.Db)
 		}
 	}
 	return sqlConnectionStr
@@ -138,6 +141,10 @@ func getMYSQLConnectionStr(server, port, user, password, dbname string) string {
 	return fmt.Sprintf("%s:%s@tcp(%s:%s)/%s", user, password, server, port, dbname)
 }
 
+func getSQLSERVERConnectionStr(server, port, user, password, dbname string) string {
+	return fmt.Sprintf(`sqlserver://%s:%s@%s:%s?database=%s`, user, password, server, port, dbname)
+}
+
 func GetSchemaSampleSize(sourceProfile SourceProfile) int64 {
 	schemaSampleSize := int64(100000)
 	if sourceProfile.Ty == SourceProfileTypeConnection {
diff --git a/profiles/source_profile.go b/profiles/source_profile.go
index 9e7ac30..acef38d 100644
--- a/profiles/source_profile.go
+++ b/profiles/source_profile.go
@@ -46,6 +46,7 @@ const (
 	SourceProfileConnectionTypeMySQL
 	SourceProfileConnectionTypePostgreSQL
 	SourceProfileConnectionTypeDynamoDB
+	SourceProfileConnectionTypeSqlServer
 )
 
 type SourceProfileConnectionMySQL struct {
@@ -160,6 +161,72 @@ func NewSourceProfileConnectionPostgreSQL(params map[string]string) (SourceProfi
 	return pg, nil
 }
 
+type SourceProfileConnectionSqlServer struct {
+	Host string
+	Port string
+	User string
+	Db   string
+	Pwd  string
+}
+
+func NewSourceProfileConnectionSqlServer(params map[string]string) (SourceProfileConnectionSqlServer, error) {
+	ss := SourceProfileConnectionSqlServer{}
+	host, hostOk := params["host"]
+	user, userOk := params["user"]
+	db, dbOk := params["db_name"]
+	port, portOk := params["port"]
+	pwd, pwdOk := params["password"]
+
+	// We don't allow users to mix and match params from source-profile and environment variables.
+	// We either try to get all params from the source-profile and if none are set, we read from the env variables.
+	if !(hostOk || userOk || dbOk || portOk || pwdOk) {
+		// No connection params provided through source-profile. Fetching from env variables.
+		fmt.Printf("Connection parameters not specified in source-profile. Reading from " +
+			"environment variables MSSQL_IP_ADDRESS, MSSQL_USER, MSSQL_DATABASE, MSSQL_TCP_PORT, MSSQL_SA_PASSWORD...\n")
+		ss.Host = os.Getenv("MSSQL_IP_ADDRESS") //For default SQL Server instances.
+		ss.Port = os.Getenv("MSSQL_TCP_PORT")
+		ss.Pwd = os.Getenv("MSSQL_SA_PASSWORD")
+
+		ss.Db = os.Getenv("MSSQL_DATABASE")  //Non standard env variable. Defined for HarbourBridge.
+		ss.User = os.Getenv("MSSQL_SA_USER") //Non standard env variable. Defined for HarbourBridge.
+		if ss.User == "" {
+			fmt.Printf("MSSQL_SA_USER environment variable is not set. Default admin user 'SA' will be used for further processing.\n")
+			ss.User = "SA"
+		}
+		// Throw error if the input entered is empty.
+		if ss.Host == "" || ss.Db == "" {
+			return ss, fmt.Errorf("found empty string for MSSQL_IP_ADDRESS/MSSQL_DATABASE. Please specify these environment variables with correct values")
+		}
+	} else if hostOk && userOk && dbOk {
+		// All connection params provided through source-profile. Port and password handled later.
+		ss.Host, ss.User, ss.Db, ss.Port, ss.Pwd = host, user, db, port, pwd
+		// Throw error if the input entered is empty.
+		if ss.Host == "" || ss.User == "" || ss.Db == "" {
+			return ss, fmt.Errorf("found empty string for host/user/db_name. Please specify host, port, user and db_name in the source-profile")
+		}
+	} else {
+		// Partial params provided through source-profile. Ask user to provide all through the source-profile.
+		return ss, fmt.Errorf("please specify host, port, user and db_name in the source-profile")
+	}
+
+	if ss.Port == "" {
+		// Set default port for sql server, which rarely changes.
+		ss.Port = "1433"
+	}
+
+	// Try to get admin password from env
+	if saPas := os.Getenv("MSSQL_SA_PASSWORD"); saPas != "" {
+		ss.Pwd = saPas
+	}
+
+	// If source profile and env do not have password then get password via prompt.
+	if ss.Pwd == "" {
+		ss.Pwd = utils.GetPassword()
+	}
+
+	return ss, nil
+}
+
 type SourceProfileConnectionDynamoDB struct {
 	// These connection params are not used currently because the SDK reads directly from the env variables.
 	// These are still kept around as reference when we refactor passing
@@ -200,10 +267,11 @@ func NewSourceProfileConnectionDynamoDB(params map[string]string) (SourceProfile
 }
 
 type SourceProfileConnection struct {
-	Ty    SourceProfileConnectionType
-	Mysql SourceProfileConnectionMySQL
-	Pg    SourceProfileConnectionPostgreSQL
-	Dydb  SourceProfileConnectionDynamoDB
+	Ty        SourceProfileConnectionType
+	Mysql     SourceProfileConnectionMySQL
+	Pg        SourceProfileConnectionPostgreSQL
+	Dydb      SourceProfileConnectionDynamoDB
+	SqlServer SourceProfileConnectionSqlServer
 }
 
 func NewSourceProfileConnection(source string, params map[string]string) (SourceProfileConnection, error) {
@@ -234,6 +302,16 @@ func NewSourceProfileConnection(source string, params map[string]string) (Source
 				return conn, err
 			}
 		}
+
+	case "sqlserver", "mssql":
+		{
+			conn.Ty = SourceProfileConnectionTypeSqlServer
+			conn.SqlServer, err = NewSourceProfileConnectionSqlServer(params)
+			if err != nil {
+				return conn, err
+			}
+		}
+
 	default:
 		return conn, fmt.Errorf("please specify a valid source database using -source flag, received source = %v", source)
 	}
@@ -284,6 +362,8 @@ func (src SourceProfile) ToLegacyDriver(source string) (string, error) {
 				return constants.POSTGRES, nil
 			case "dynamodb":
 				return constants.DYNAMODB, nil
+			case "sqlserver", "mssql":
+				return constants.SQLSERVER, nil
 			default:
 				return "", fmt.Errorf("please specify a valid source database using -source flag, received source = %v", source)
 			}
diff --git a/sources/sqlserver/README.md b/sources/sqlserver/README.md
new file mode 100644
index 0000000..ccbb8a0
--- /dev/null
+++ b/sources/sqlserver/README.md
@@ -0,0 +1,70 @@
+# HarbourBridge: SQLServer-to-Spanner Evaluation and Migration
+
+HarbourBridge is a stand-alone open source tool for Cloud Spanner evaluation,
+using data from an existing SQL Server database. This README provides
+details of the tool's SQLserver capabilities. For general HarbourBridge information
+see this [README](https://github.com/cloudspannerecosystem/harbourbridge#harbourbridge-spanner-evaluation-and-migration).
+
+Note that either _'sqlserver'_ or _'mssql'_ can be used as an identifier with the flag `-source` in the command line.
+
+## Example SQLServer Usage
+
+HarbourBridge can be run directly on a sqlserver database (via go's database/sql package).
+
+The following examples assume a `harbourbridge` alias has been setup as described
+in the [Installing HarbourBridge](https://github.com/cloudspannerecosystem/harbourbridge#installing-harbourbridge) section of the main README.
+
+### Directly connecting to a Sql server database
+
+In this case, HarbourBridge connects directly to the Sql server database to
+retrieve table schema and data. Set the `-source=sqlserver` and corresponding
+source profile connection parameters `host`, `port`, `user`, `db_name` and
+`password`.
+
+For example to perform schema conversion, run
+
+```sh
+harbourbridge schema -source=sqlserver -source-profile="host=<>,port=<>,user=<>,db_name=<>"
+```
+
+Parameters `port` and `password` are optional. Port (`port`) defaults to `1433`
+for SQLserver source. Password can be provided at the password prompt.
+
+## Schema Conversion
+
+| SQL_Server_Type        | Spanner_Type |
+| ---------------------- | ------------ |
+| INT                    | INT64        |
+| TINYINT                | INT64        |
+| SMALLINT               | INT64        |
+| BIGINT                 | INT64        |
+| TIMESTAMP              | INT64        |
+| ROWVERSION             | INT64        |
+| BIT                    | BOOL         |
+| FLOAT                  | FLOAT64      |
+| REAL                   | FLOAT64      |
+| NUMERIC                | NUMERIC      |
+| DECIMAL                | NUMERIC      |
+| MONEY                  | NUMERIC      |
+| SMALLMONEY             | NUMERIC      |
+| CHAR                   | STRING       |
+| NCHAR                  | STRING       |
+| VARCHAR                | STRING       |
+| NVARCHAR               | STRING       |
+| TEXT                   | STRING       |
+| NTEXT                  | STRING       |
+| DATE                   | DATE         |
+| DATETIME               | TIMESTAMP    |
+| DATETIME2              | TIMESTAMP    |
+| SMALLDATETIME          | TIMESTAMP    |
+| DATETIMEOFFSET         | TIMESTAMP    |
+| TIME                   | STRING       |
+| BINARY                 | BYTES        |
+| VARBINARY              | BYTES        |
+| IMAGE                  | BYTES        |
+| XML                    | STRING       |
+| UNIQUEIDENTIFIER       | STRING       |
+| SQL_VARIANT            | STRING       |
+| HIERARCHYID            | STRING       |
+| Spatial Geography Type | STRING       |
+| Spatial Geometry Types | STRING       |
diff --git a/sources/sqlserver/data.go b/sources/sqlserver/data.go
new file mode 100644
index 0000000..2a4fe3b
--- /dev/null
+++ b/sources/sqlserver/data.go
@@ -0,0 +1,188 @@
+// Copyright 2021 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 sqlserver
+
+import (
+	"fmt"
+	"math/big"
+	"math/bits"
+	"strconv"
+	"strings"
+	"time"
+
+	"cloud.google.com/go/civil"
+	"cloud.google.com/go/spanner"
+	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/schema"
+	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
+)
+
+// ProcessDataRow converts a row of data and writes it out to Spanner.
+// srcTable and srcCols are the source table and columns respectively,
+// and vals contains string data to be converted to appropriate types
+// to send to Spanner.  ProcessDataRow is only called in DataMode.
+func ProcessDataRow(conv *internal.Conv, srcTable string, srcCols []string, srcSchema schema.Table, spTable string, spCols []string, spSchema ddl.CreateTable, vals []string) {
+	spTable, cvtCols, cvtVals, err := ConvertData(conv, srcTable, srcCols, srcSchema, spTable, spCols, spSchema, vals)
+	if err != nil {
+		conv.Unexpected(fmt.Sprintf("Error while converting data: %s\n", err))
+		conv.StatsAddBadRow(srcTable, conv.DataMode())
+		conv.CollectBadRow(srcTable, srcCols, vals)
+	} else {
+		conv.WriteRow(srcTable, spTable, cvtCols, cvtVals)
+	}
+}
+
+// ConvertData maps the source DB data in vals into Spanner data,
+// based on the Spanner and source DB schemas. Note that since entries
+// in vals may be empty, we also return the list of columns (empty
+// cols are dropped).
+func ConvertData(conv *internal.Conv, srcTable string, srcCols []string, srcSchema schema.Table, spTable string, spCols []string, spSchema ddl.CreateTable, vals []string) (string, []string, []interface{}, error) {
+	var c []string
+	var v []interface{}
+	if len(spCols) != len(srcCols) || len(spCols) != len(vals) {
+		return "", []string{}, []interface{}{}, fmt.Errorf("ConvertData: spCols, srcCols and vals don't all have the same lengths: len(spCols)=%d, len(srcCols)=%d, len(vals)=%d", len(spCols), len(srcCols), len(vals))
+	}
+	for i, spCol := range spCols {
+		srcCol := srcCols[i]
+		// Skip columns with 'NULL' values.
+		if vals[i] == "NULL" {
+			continue
+		}
+		spColDef, ok1 := spSchema.ColDefs[spCol]
+		srcColDef, ok2 := srcSchema.ColDefs[srcCol]
+		if !ok1 || !ok2 {
+			return "", []string{}, []interface{}{}, fmt.Errorf("can't find Spanner and source-db schema for col %s", spCol)
+		}
+		var x interface{}
+		var err error
+		x, err = convScalar(conv, spColDef.T, srcColDef.Type.Name, conv.TimezoneOffset, vals[i])
+		if err != nil {
+			return "", []string{}, []interface{}{}, err
+		}
+		v = append(v, x)
+		c = append(c, spCol)
+	}
+	if aux, ok := conv.SyntheticPKeys[spTable]; ok {
+		c = append(c, aux.Col)
+		v = append(v, int64(bits.Reverse64(uint64(aux.Sequence))))
+		aux.Sequence++
+		conv.SyntheticPKeys[spTable] = aux
+	}
+	return spTable, c, v, nil
+}
+
+// convScalar converts a source database string value to an
+// appropriate Spanner value. It is the caller's responsibility to
+// detect and handle NULL values: convScalar will return error if a
+// NULL value is passed.
+func convScalar(conv *internal.Conv, spannerType ddl.Type, srcTypeName string, timezoneOffset string, val string) (interface{}, error) {
+	// Whitespace within the val string is considered part of the data value.
+	// Note that many of the underlying conversions functions we use (like
+	// strconv.ParseFloat and strconv.ParseInt) return "invalid syntax"
+	// errors if whitespace were to appear at the start or end of a string.
+	switch spannerType.Name {
+	case ddl.Bool:
+		return convBool(val)
+	case ddl.Bytes:
+		return convBytes(val)
+	case ddl.Date:
+		return convDate(val)
+	case ddl.Float64:
+		return convFloat64(val)
+	case ddl.Int64:
+		return convInt64(val)
+	case ddl.Numeric:
+		return convNumeric(conv, val)
+	case ddl.String:
+		return val, nil
+	case ddl.Timestamp:
+		return convTimestamp(srcTypeName, val)
+	default:
+		return val, fmt.Errorf("data conversion not implemented for type %v", spannerType.Name)
+	}
+}
+
+func convBool(val string) (bool, error) {
+	b, err := strconv.ParseBool(val)
+	if err != nil {
+		return b, fmt.Errorf("can't convert to bool: %w", err)
+	}
+	return b, err
+}
+
+func convBytes(val string) ([]byte, error) {
+	// convert a string to a byte slice.
+	b := []byte(val)
+	return b, nil
+}
+
+func convDate(val string) (civil.Date, error) {
+	date := strings.Fields(val)
+	d, err := civil.ParseDate(date[0])
+	if err != nil {
+		return d, fmt.Errorf("can't convert to date: %w", err)
+	}
+	return d, err
+}
+
+func convFloat64(val string) (float64, error) {
+	float, err := strconv.ParseFloat(val, 64)
+	if err != nil {
+		return float, fmt.Errorf("can't convert to float64: %w", err)
+	}
+	return float, err
+}
+
+func convInt64(val string) (int64, error) {
+	i, err := strconv.ParseInt(val, 10, 64)
+	if err != nil {
+		return i, fmt.Errorf("can't convert to int64: %w", err)
+	}
+	return i, err
+}
+
+// convNumeric maps a source database string value (representing a numeric)
+// into a string representing a valid Spanner numeric.
+func convNumeric(conv *internal.Conv, val string) (interface{}, error) {
+	if conv.TargetDb == constants.TargetExperimentalPostgres {
+		return spanner.PGNumeric{Numeric: val, Valid: true}, nil
+	} else {
+		r := new(big.Rat)
+		if _, ok := r.SetString(val); !ok {
+			return "", fmt.Errorf("can't convert %q to big.Rat", val)
+		}
+		return r, nil
+	}
+}
+
+// convTimestamp maps a source DB datetime types to Spanner timestamp
+func convTimestamp(srcTypeName string, val string) (t time.Time, err error) {
+	// the query returns the datetime in ISO8601
+	// e.g. 2021-12-15T07:39:52.943 			(datetime)
+	// e.g. 2021-12-15T07:39:52.9433333 		(datetime2)
+	// e.g. 2021-12-15T07:40:00 				(smalldatetime)
+	// e.g. 2021-12-08T03:00:52.9500000+01:00 	(datetimeoffset)
+
+	if srcTypeName == dateTimeOffsetType {
+		t, err = time.Parse(time.RFC3339, val)
+	} else {
+		t, err = time.Parse("2006-01-02T15:04:05", val)
+	}
+	if err != nil {
+		return t, fmt.Errorf("can't convert to timestamp (mssql type: %s)", srcTypeName)
+	}
+	return t, err
+}
diff --git a/sources/sqlserver/infoschema.go b/sources/sqlserver/infoschema.go
new file mode 100644
index 0000000..a238cca
--- /dev/null
+++ b/sources/sqlserver/infoschema.go
@@ -0,0 +1,435 @@
+// Copyright 2021 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 sqlserver
+
+import (
+	"database/sql"
+	"fmt"
+	"sort"
+	"strings"
+
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/schema"
+	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
+	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
+)
+
+const (
+	uuidType           string = "uniqueidentifier"
+	geographyType      string = "geography"
+	geometryType       string = "geometry"
+	timeType           string = "time"
+	hierarchyIdType    string = "hierarchyid"
+	timestampType      string = "timestamp"
+	dateTimeType       string = "datetime"
+	dateTime2Type      string = "datetime2"
+	dateTimeOffsetType string = "datetimeoffset"
+	smallDateTimeType  string = "smalldatetime"
+	dateType           string = "date"
+)
+
+type InfoSchemaImpl struct {
+	DbName string
+	Db     *sql.DB
+}
+
+// GetToDdl function below implement the common.InfoSchema interface.
+func (isi InfoSchemaImpl) GetToDdl() common.ToDdl {
+	return ToDdlImpl{}
+}
+
+// GetTableName returns table name.
+func (isi InfoSchemaImpl) GetTableName(schema string, tableName string) string {
+	if schema == "dbo" { // Drop 'dbo' prefix.
+		return tableName
+	}
+	return fmt.Sprintf("%s.%s", schema, tableName)
+}
+
+// ProcessDataRows performs data conversion for source database
+// 'db'. For each table, we extract data using a "SELECT *" query,
+// convert the data to Spanner data (based on the source and Spanner
+// schemas), and write it to Spanner.  If we can't get/process data
+// for a table, we skip that table and process the remaining tables.
+//
+// Note that the database/sql library has a somewhat complex model for
+// returning data from rows.Scan. Scalar values can be returned using
+// the native value used by the underlying driver (by passing
+// *interface{} to rows.Scan), or they can be converted to specific go
+// types.
+// We choose to do all type conversions explicitly ourselves so that
+// we can generate more targeted error messages: hence we pass
+// *interface{} parameters to row.Scan.
+func (isi InfoSchemaImpl) ProcessData(conv *internal.Conv, srcTable string, srcSchema schema.Table, spTable string, spCols []string, spSchema ddl.CreateTable) error {
+	rowsInterface, err := isi.GetRowsFromTable(conv, srcTable)
+	if err != nil {
+		conv.Unexpected(fmt.Sprintf("Couldn't get data for table %s : err = %s", srcTable, err))
+		return err
+	}
+	rows := rowsInterface.(*sql.Rows)
+	defer rows.Close()
+	srcCols, _ := rows.Columns()
+	v, scanArgs := buildVals(len(srcCols))
+	for rows.Next() {
+		// get RawBytes from data.
+		err := rows.Scan(scanArgs...)
+		if err != nil {
+			conv.Unexpected(fmt.Sprintf("Couldn't process sql data row: %s", err))
+			// Scan failed, so we don't have any data to add to bad rows.
+			conv.StatsAddBadRow(srcTable, conv.DataMode())
+			continue
+		}
+		values := valsToStrings(v)
+		ProcessDataRow(conv, srcTable, srcCols, srcSchema, spTable, spCols, spSchema, values)
+	}
+	return nil
+}
+
+// GetRowsFromTable returns a sql Rows object for a table.
+func (isi InfoSchemaImpl) GetRowsFromTable(conv *internal.Conv, srcTable string) (interface{}, error) {
+	tbl := conv.SrcSchema[srcTable]
+	//To get only the table name by removing the schema name prefix
+	tblName := strings.Replace(srcTable, tbl.Schema+".", "", 1)
+
+	q := getSelectQuery(isi.DbName, tbl.Schema, tblName, tbl.ColNames, tbl.ColDefs)
+	rows, err := isi.Db.Query(q)
+	if err != nil {
+		return nil, err
+	}
+	return rows, err
+}
+
+func getSelectQuery(srcDb string, schemaName string, tableName string, colNames []string, colDefs map[string]schema.Column) string {
+	var selects = make([]string, len(colNames))
+
+	for i, cn := range colNames {
+		var s string
+		switch colDefs[cn].Type.Name {
+		case geometryType, geographyType:
+			s = fmt.Sprintf("[%s].STAsText() AS %s", cn, cn)
+		case uuidType:
+			s = fmt.Sprintf("CAST([%s] AS VARCHAR(36)) AS %s", cn, cn)
+		case hierarchyIdType:
+			s = fmt.Sprintf("CAST([%s] AS VARCHAR(4000)) AS %s", cn, cn)
+		case timeType:
+			s = fmt.Sprintf("CAST([%s] AS VARCHAR(12)) AS %s", cn, cn)
+		case timestampType:
+			s = fmt.Sprintf("CAST([%s] AS BIGINT) AS %s", cn, cn)
+		case smallDateTimeType, dateTimeType, dateTime2Type, dateTimeOffsetType:
+			s = fmt.Sprintf("CONVERT(VARCHAR(33), [%s], 126) AS %s", cn, cn)
+		case dateType:
+			s = fmt.Sprintf("CONVERT(VARCHAR(10), [%s], 23) AS %s", cn, cn)
+		default:
+			s = fmt.Sprintf("[%s]", cn)
+		}
+		selects[i] = s
+	}
+
+	return fmt.Sprintf("SELECT %s FROM [%s].[%s].[%s]", strings.Join(selects, ", "), srcDb, schemaName, tableName)
+}
+
+// buildVals contructs interface{} value containers to scan row
+// results into.  Returns both the underlying containers (as a slice)
+// as well as an interface{} of pointers to containers to pass to
+// rows.Scan.
+func buildVals(n int) (v []interface{}, iv []interface{}) {
+	v = make([]interface{}, n)
+	for i := range v {
+		iv = append(iv, &v[i])
+	}
+	return v, iv
+}
+
+// GetRowCount with number of rows in each table.
+func (isi InfoSchemaImpl) GetRowCount(table common.SchemaAndName) (int64, error) {
+	q := fmt.Sprintf(`SELECT COUNT(1) FROM [%s].[%s].[%s];`, isi.DbName, table.Schema, table.Name)
+	rows, err := isi.Db.Query(q)
+	if err != nil {
+		return 0, err
+	}
+	defer rows.Close()
+	var count int64
+	if rows.Next() {
+		err := rows.Scan(&count)
+		return count, err
+	}
+	return 0, nil
+}
+
+// GetTables return list of tables in the selected database.
+func (isi InfoSchemaImpl) GetTables() ([]common.SchemaAndName, error) {
+	q := `
+	SELECT 
+		SCH.name AS table_schema, 
+		TBL.name AS table_name
+	FROM sys.tables AS TBL
+	INNER JOIN sys.schemas AS SCH 
+	ON SCH.schema_id = TBL.schema_id
+	WHERE TBL.type = 'U' AND TBL.is_tracked_by_cdc = 0 AND TBL.is_ms_shipped = 0 AND TBL.name <> 'sysdiagrams'
+	`
+	rows, err := isi.Db.Query(q)
+	if err != nil {
+		return nil, fmt.Errorf("couldn't get tables: %w", err)
+	}
+
+	defer rows.Close()
+	var tableSchema, tableName string
+	var tables []common.SchemaAndName
+	for rows.Next() {
+		rows.Scan(&tableSchema, &tableName)
+		tables = append(tables, common.SchemaAndName{Schema: tableSchema, Name: tableName})
+	}
+	return tables, nil
+}
+
+// GetColumns returns a list of Column objects and names
+func (isi InfoSchemaImpl) GetColumns(conv *internal.Conv, table common.SchemaAndName, constraints map[string][]string, primaryKeys []string) (map[string]schema.Column, []string, error) {
+	q := `
+		SELECT 
+			column_name, 
+			data_type, 
+			is_nullable, 
+			column_default, 
+			character_maximum_length, 
+			numeric_precision, 
+			numeric_scale
+		FROM information_schema.COLUMNS 
+		WHERE table_schema = @p1 and table_name = @p2 
+		ORDER BY ordinal_position;
+	`
+	cols, err := isi.Db.Query(q, table.Schema, table.Name)
+	if err != nil {
+		return nil, nil, fmt.Errorf("couldn't get schema for table %s.%s: %s", table.Schema, table.Name, err)
+	}
+	colDefs := make(map[string]schema.Column)
+	var colNames []string
+	var colName, dataType string
+	var isNullable string
+	var colDefault sql.NullString
+	// elementDataType
+	var charMaxLen, numericPrecision, numericScale sql.NullInt64
+	for cols.Next() {
+		err := cols.Scan(&colName, &dataType, &isNullable, &colDefault, &charMaxLen, &numericPrecision, &numericScale)
+		if err != nil {
+			conv.Unexpected(fmt.Sprintf("Can't scan: %v", err))
+			continue
+		}
+		ignored := schema.Ignored{}
+		for _, c := range constraints[colName] {
+			// c can be UNIQUE, PRIMARY KEY, FOREIGN KEY,
+			// or CHECK (based on msql, sql server, postgres docs).
+			// We've already filtered out PRIMARY KEY.
+			switch c {
+			case "CHECK":
+				ignored.Check = true
+			case "FOREIGN KEY", "PRIMARY KEY", "UNIQUE":
+				// Nothing to do here -- these are handled elsewhere.
+			}
+		}
+		ignored.Default = colDefault.Valid
+		c := schema.Column{
+			Name:    colName,
+			Type:    toType(dataType, charMaxLen, numericPrecision, numericScale),
+			NotNull: strings.ToUpper(isNullable) == "NO",
+			Ignored: ignored,
+		}
+		colDefs[colName] = c
+		colNames = append(colNames, colName)
+	}
+	return colDefs, colNames, nil
+}
+
+// GetConstraints returns a list of primary keys and by-column map of
+// other constraints.  Note: we need to preserve ordinal order of
+// columns in primary key constraints.
+// Note that foreign key constraints are handled in getForeignKeys.
+func (isi InfoSchemaImpl) GetConstraints(conv *internal.Conv, table common.SchemaAndName) ([]string, map[string][]string, error) {
+	q := `
+		SELECT 
+			k.COLUMN_NAME, 
+			t.CONSTRAINT_TYPE
+		FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS AS t
+		INNER JOIN INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS k 
+			ON t.CONSTRAINT_NAME = k.CONSTRAINT_NAME AND t.CONSTRAINT_SCHEMA = k.CONSTRAINT_SCHEMA
+		WHERE k.TABLE_SCHEMA = @p1 AND k.TABLE_NAME = @p2 ORDER BY k.ordinal_position;
+	`
+	rows, err := isi.Db.Query(q, table.Schema, table.Name)
+	if err != nil {
+		return nil, nil, err
+	}
+	defer rows.Close()
+	var primaryKeys []string
+	var col, constraint string
+	m := make(map[string][]string)
+	for rows.Next() {
+		err := rows.Scan(&col, &constraint)
+		if err != nil {
+			conv.Unexpected(fmt.Sprintf("Can't scan: %v", err))
+			continue
+		}
+		if col == "" || constraint == "" {
+			conv.Unexpected("Got empty col or constraint")
+			continue
+		}
+		switch constraint {
+		case "PRIMARY KEY":
+			primaryKeys = append(primaryKeys, col)
+		default:
+			m[col] = append(m[col], constraint)
+		}
+	}
+	return primaryKeys, m, nil
+}
+
+// GetForeignKeys returns a list of all the foreign key constraints.
+func (isi InfoSchemaImpl) GetForeignKeys(conv *internal.Conv, table common.SchemaAndName) (foreignKeys []schema.ForeignKey, err error) {
+	q := `
+	SELECT 
+		OBJECT_SCHEMA_NAME (FK.referenced_object_id) AS [schema_name],
+		OBJECT_NAME (FK.referenced_object_id) AS [referenced_table],
+		COL_NAME(FKC.parent_object_id, FKC.parent_column_id) AS [column],  
+		COL_NAME(FKC.referenced_object_id, FKC.referenced_column_id) AS [referenced_column],  
+		FK.name AS [foreign_key_name]
+	FROM sys.foreign_keys AS FK  
+	INNER JOIN sys.foreign_key_columns AS FKC   
+    ON FK.object_id = FKC.constraint_object_id  
+	WHERE FK.parent_object_id = OBJECT_ID(@p1);
+	`
+
+	rows, err := isi.Db.Query(q, fmt.Sprintf("%s.%s", table.Schema, table.Name))
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var refTable common.SchemaAndName
+	var col, refCol, fKeyName string
+	fKeys := make(map[string]common.FkConstraint)
+	var keyNames []string
+	for rows.Next() {
+		err := rows.Scan(&refTable.Schema, &refTable.Name, &col, &refCol, &fKeyName)
+		if err != nil {
+			conv.Unexpected(fmt.Sprintf("Can't scan: %v", err))
+			continue
+		}
+		tableName := isi.GetTableName(refTable.Schema, refTable.Name)
+		if _, found := fKeys[fKeyName]; found {
+			fk := fKeys[fKeyName]
+			fk.Cols = append(fk.Cols, col)
+			fk.Refcols = append(fk.Refcols, refCol)
+			fKeys[fKeyName] = fk
+			continue
+		}
+		fKeys[fKeyName] = common.FkConstraint{Name: fKeyName, Table: tableName, Refcols: []string{refCol}, Cols: []string{col}}
+		keyNames = append(keyNames, fKeyName)
+	}
+
+	sort.Strings(keyNames)
+	for _, k := range keyNames {
+		foreignKeys = append(foreignKeys,
+			schema.ForeignKey{
+				Name:         fKeys[k].Name,
+				Columns:      fKeys[k].Cols,
+				ReferTable:   fKeys[k].Table,
+				ReferColumns: fKeys[k].Refcols})
+	}
+	return foreignKeys, nil
+}
+
+// GetIndexes return a list of all indexes for the specified table.
+func (isi InfoSchemaImpl) GetIndexes(conv *internal.Conv, table common.SchemaAndName) ([]schema.Index, error) {
+	q2 := `
+		SELECT
+			IX.name, 
+			COL_NAME(IX.object_id, IXC.column_id) as [Column Name],
+			IX.is_unique,
+			IXC.is_descending_key 
+		FROM sys.indexes IX 
+		INNER JOIN sys.index_columns IXC 
+			ON  IX.object_id = IXC.object_id AND IX.index_id = IXC.index_id
+		INNER JOIN sys.tables TAB 
+			ON IX.object_id = TAB.object_id 
+		WHERE
+			IX.is_primary_key = 0          
+			AND IX.is_unique_constraint = 0 
+			AND TAB.is_ms_shipped = 0   
+			AND TAB.name=@p1
+			AND TAB.schema_id = SCHEMA_ID(@p2)
+			ORDER BY IX.name ;
+	`
+	rows, err := isi.Db.Query(q2, table.Name, table.Schema)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var name, column, isUnique, collation string
+	indexMap := make(map[string]schema.Index)
+	var indexNames []string
+	var indexes []schema.Index
+	for rows.Next() {
+		if err := rows.Scan(&name, &column, &isUnique, &collation); err != nil {
+			conv.Unexpected(fmt.Sprintf("Can't scan: %v", err))
+			continue
+		}
+
+		if _, found := indexMap[name]; !found {
+			indexNames = append(indexNames, name)
+			indexMap[name] = schema.Index{Name: name, Unique: (isUnique == "true")}
+		}
+		index := indexMap[name]
+		index.Keys = append(index.Keys, schema.Key{Column: column, Desc: (collation == "DESC")})
+		indexMap[name] = index
+	}
+	for _, k := range indexNames {
+		indexes = append(indexes, indexMap[k])
+	}
+	return indexes, nil
+}
+
+func toType(dataType string, charLen sql.NullInt64, numericPrecision, numericScale sql.NullInt64) schema.Type {
+	switch {
+	case charLen.Valid:
+		return schema.Type{Name: dataType, Mods: []int64{charLen.Int64}}
+	case dataType == "numeric" && numericPrecision.Valid && numericScale.Valid && numericScale.Int64 != 0:
+		return schema.Type{Name: dataType, Mods: []int64{numericPrecision.Int64, numericScale.Int64}}
+	case dataType == "numeric" && numericPrecision.Valid:
+		return schema.Type{Name: dataType, Mods: []int64{numericPrecision.Int64}}
+	case dataType == "decimal" && numericPrecision.Valid && numericScale.Valid && numericScale.Int64 != 0:
+		return schema.Type{Name: dataType, Mods: []int64{numericPrecision.Int64, numericScale.Int64}}
+	case dataType == "decimal" && numericPrecision.Valid:
+		return schema.Type{Name: dataType, Mods: []int64{numericPrecision.Int64}}
+	default:
+		return schema.Type{Name: dataType}
+	}
+}
+
+func valsToStrings(vals []interface{}) []string {
+	toString := func(val interface{}) string {
+		if val == nil {
+			return "NULL"
+		}
+		switch v := val.(type) {
+		case []uint8:
+			val = string([]byte(v))
+		case *interface{}:
+			val = *v
+		}
+		return fmt.Sprintf("%v", val)
+	}
+	var s []string
+	for _, v := range vals {
+		s = append(s, toString(v))
+	}
+	return s
+}
diff --git a/sources/sqlserver/toddl.go b/sources/sqlserver/toddl.go
new file mode 100644
index 0000000..c120e33
--- /dev/null
+++ b/sources/sqlserver/toddl.go
@@ -0,0 +1,95 @@
+// Copyright 2021 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 sqlserver handles schema and data migrations from sqlserver.
+package sqlserver
+
+import (
+	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/schema"
+	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
+)
+
+// ToDdlImpl sql server specific implementation for ToDdl.
+type ToDdlImpl struct {
+}
+
+// ToSpannerType maps a scalar source schema type (defined by id and
+// mods) into a Spanner type. This is the core source-to-Spanner type
+// mapping.  toSpannerType returns the Spanner type and a list of type
+// conversion issues encountered.
+func (tdi ToDdlImpl) ToSpannerType(conv *internal.Conv, columnType schema.Type) (ddl.Type, []internal.SchemaIssue) {
+	ty, issues := toSpannerTypeInternal(columnType.Name, columnType.Mods)
+	if conv.TargetDb == constants.TargetExperimentalPostgres {
+		ty = overrideExperimentalType(columnType, ty)
+	}
+	return ty, issues
+}
+
+// toSpannerType maps a scalar source schema type (defined by id and
+// mods) into a Spanner type. This is the core source-to-Spanner type
+// mapping.  toSpannerType returns the Spanner type and a list of type
+// conversion issues encountered.
+func toSpannerTypeInternal(srcType string, mods []int64) (ddl.Type, []internal.SchemaIssue) {
+	switch srcType {
+	case "bit":
+		return ddl.Type{Name: ddl.Bool}, nil
+	case "uniqueidentifier":
+		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, nil
+	case "binary", "varbinary", "image":
+		return ddl.Type{Name: ddl.Bytes, Len: ddl.MaxLength}, nil
+	case "date":
+		return ddl.Type{Name: ddl.Date}, nil
+	case "float", "real":
+		return ddl.Type{Name: ddl.Float64}, []internal.SchemaIssue{internal.Widened}
+	case "bigint":
+		return ddl.Type{Name: ddl.Int64}, nil
+	case "tinyint", "smallint", "int":
+		return ddl.Type{Name: ddl.Int64}, []internal.SchemaIssue{internal.Widened}
+	case "numeric", "money", "smallmoney", "decimal":
+		return ddl.Type{Name: ddl.Numeric}, nil
+	case "ntext", "text", "xml":
+		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, nil
+	case "smalldatetime", "datetimeoffset", "datetime2", "datetime":
+		return ddl.Type{Name: ddl.Timestamp}, []internal.SchemaIssue{internal.Timestamp}
+	case "time":
+		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, []internal.SchemaIssue{internal.Time}
+	case "varchar", "char", "nvarchar", "nchar":
+		// Sets the source length only if it falls within the allowed length range in Spanner.
+		if len(mods) > 0 && mods[0] > 0 && mods[0] <= ddl.StringMaxLength {
+			return ddl.Type{Name: ddl.String, Len: mods[0]}, nil
+		}
+		// Raises warning and sets length to MAX when -
+		// Source length is greater than maximum allowed length
+		// -OR-
+		// Source length is "-1" which represents MAX in SQL Server
+		if len(mods) > 0 && (mods[0] > ddl.StringMaxLength || mods[0] < 0) {
+			return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, []internal.SchemaIssue{internal.StringOverflow}
+		}
+		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, nil
+	case "timestamp":
+		return ddl.Type{Name: ddl.Int64}, nil
+	default:
+		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, []internal.SchemaIssue{internal.NoGoodType}
+	}
+}
+
+// Override the types to map to experimental postgres types.
+func overrideExperimentalType(columnType schema.Type, originalType ddl.Type) ddl.Type {
+	if columnType.Name == "date" {
+		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}
+	}
+	return originalType
+}
diff --git a/spanner/ddl/ast.go b/spanner/ddl/ast.go
index 3f5cc05..21a011e 100644
--- a/spanner/ddl/ast.go
+++ b/spanner/ddl/ast.go
@@ -52,6 +52,8 @@ const (
 	JSON string = "JSON"
 	// MaxLength is a sentinel for Type's Len field, representing the MAX value.
 	MaxLength = math.MaxInt64
+	// StringMaxLength represents maximum allowed STRING length.
+	StringMaxLength = 2621440
 
 	// Types specific to Spanner with postgresql dialect, when they differ from
 	// Spanner with google_standard_sql.
diff --git a/web/sqlserver_typemap.go b/web/sqlserver_typemap.go
new file mode 100644
index 0000000..b68a942
--- /dev/null
+++ b/web/sqlserver_typemap.go
@@ -0,0 +1,124 @@
+// Copyright 2020 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 web
+
+import (
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
+)
+
+// toSpannerTypeSQLserver defines the mapping of source types into Spanner
+// types. Each source type has a default Spanner type, as well as other potential
+// Spanner types it could map to. When calling toSpannerTypeSQLserver, you specify
+// the source type name (along with any modifiers), and optionally you specify
+// a target Spanner type name (empty string if you don't have one). If the target
+// Spanner type name is specified and is a potential mapping for this source type,
+// then it will be used to build the returned ddl.Type. If not, the default
+// Spanner type for this source type will be used.
+// Note that toSpannerTypeSQLserver is extensively tested via tests in web_test.go.
+//
+// TODO: Move the type remapping function to toddl.go (once we've merged
+// dynamodb/toddl.go, mysql/toddl.go, sqlserver/toddl.go and postgres/toddl.go).
+// Consider some refactoring to reduce code duplication (although note
+// that this type remapping has to preserve all previous changes done via the UI!)
+func toSpannerTypeSQLserver(srcType string, spType string, mods []int64) (ddl.Type, []internal.SchemaIssue) {
+	switch srcType {
+	case "tinyint", "smallint", "int", "bigint":
+		switch spType {
+		case ddl.String:
+			return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, []internal.SchemaIssue{internal.Widened}
+		case ddl.Int64:
+			return ddl.Type{Name: ddl.Int64}, []internal.SchemaIssue{internal.Widened}
+		default:
+			return ddl.Type{Name: ddl.Int64}, []internal.SchemaIssue{internal.Widened}
+		}
+	case "float", "real":
+		switch spType {
+		case ddl.String:
+			return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, []internal.SchemaIssue{internal.Widened}
+		default:
+			return ddl.Type{Name: ddl.Float64}, []internal.SchemaIssue{internal.Widened}
+		}
+	case "numeric", "decimal", "money", "smallmoney":
+		switch spType {
+		case ddl.String:
+			return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, []internal.SchemaIssue{internal.Widened}
+		default:
+			// TODO: check mod[0] and mod[1] and generate a warning
+			// if this numeric won't fit in Spanner's NUMERIC.
+			return ddl.Type{Name: ddl.Numeric}, nil
+		}
+
+	case "bit":
+		switch spType {
+		case ddl.String:
+			return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, nil
+		default:
+			return ddl.Type{Name: ddl.Bool}, nil
+		}
+	case "varchar", "char", "nvarchar", "nchar", "uniqueidentifier":
+		switch spType {
+		case ddl.Bytes:
+			if len(mods) > 0 && mods[0] > 0 {
+				return ddl.Type{Name: ddl.Bytes, Len: mods[0]}, nil
+			}
+			return ddl.Type{Name: ddl.Bytes, Len: ddl.MaxLength}, nil
+		default:
+			if len(mods) > 0 && mods[0] > 0 {
+				return ddl.Type{Name: ddl.String, Len: mods[0]}, nil
+			}
+			return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, nil
+		}
+	case "ntext", "text", "xml":
+		switch spType {
+		case ddl.Bytes:
+			return ddl.Type{Name: ddl.Bytes, Len: ddl.MaxLength}, nil
+		default:
+			return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, nil
+		}
+
+	case "binary", "varbinary", "image":
+		switch spType {
+		case ddl.String:
+			return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, nil
+		default:
+			return ddl.Type{Name: ddl.Bytes, Len: ddl.MaxLength}, nil
+		}
+	case "date":
+		switch spType {
+		case ddl.String:
+			return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, []internal.SchemaIssue{internal.Widened}
+		default:
+			return ddl.Type{Name: ddl.Date}, nil
+		}
+	case "datetime2", "datetime", "datetimeoffset", "smalldatetime", "rowversion":
+		switch spType {
+		case ddl.String:
+			return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, []internal.SchemaIssue{internal.Widened}
+		default:
+			return ddl.Type{Name: ddl.Timestamp}, []internal.SchemaIssue{internal.Datetime}
+		}
+	case "timestamp":
+		switch spType {
+		case ddl.String:
+			return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, []internal.SchemaIssue{internal.Widened}
+		default:
+			return ddl.Type{Name: ddl.Int64}, nil
+		}
+	case "time":
+		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, []internal.SchemaIssue{internal.Time}
+	}
+	return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, []internal.SchemaIssue{internal.NoGoodType}
+}
diff --git a/web/web.go b/web/web.go
index 5fc6cef..e0d8635 100644
--- a/web/web.go
+++ b/web/web.go
@@ -42,6 +42,7 @@ import (
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/mysql"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/postgres"
+	"github.com/cloudspannerecosystem/harbourbridge/sources/sqlserver"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/gorilla/handlers"
@@ -58,6 +59,7 @@ import (
 // 8) Add an overview in summary report API
 var mysqlTypeMap = make(map[string][]typeIssue)
 var postgresTypeMap = make(map[string][]typeIssue)
+var sqlserverTypeMap = make(map[string][]typeIssue)
 
 // TODO:(searce) organize this file according to go style guidelines: generally
 // have public constants and public type definitions first, then public
@@ -94,6 +96,8 @@ func databaseConnection(w http.ResponseWriter, r *http.Request) {
 		dataSourceName = fmt.Sprintf("host=%s port=%s user=%s password=%s dbname=%s sslmode=disable", config.Host, config.Port, config.User, config.Password, config.Database)
 	case constants.MYSQL:
 		dataSourceName = fmt.Sprintf("%s:%s@tcp(%s:%s)/%s", config.User, config.Password, config.Host, config.Port, config.Database)
+	case constants.SQLSERVER:
+		dataSourceName = fmt.Sprintf(`sqlserver://%s:%s@%s:%s?database=%s`, config.User, config.Password, config.Host, config.Port, config.Database)
 	default:
 		http.Error(w, fmt.Sprintf("Driver : '%s' is not supported", config.Driver), http.StatusBadRequest)
 		return
@@ -130,6 +134,8 @@ func convertSchemaSQL(w http.ResponseWriter, r *http.Request) {
 		err = common.ProcessSchema(conv, mysql.InfoSchemaImpl{DbName: sessionState.dbName, Db: sessionState.sourceDB})
 	case constants.POSTGRES:
 		err = common.ProcessSchema(conv, postgres.InfoSchemaImpl{Db: sessionState.sourceDB})
+	case constants.SQLSERVER:
+		err = common.ProcessSchema(conv, sqlserver.InfoSchemaImpl{DbName: sessionState.dbName, Db: sessionState.sourceDB})
 	default:
 		http.Error(w, fmt.Sprintf("Driver : '%s' is not supported", sessionState.driver), http.StatusBadRequest)
 		return
@@ -250,6 +256,8 @@ func getTypeMap(w http.ResponseWriter, r *http.Request) {
 		typeMap = mysqlTypeMap
 	case constants.POSTGRES, constants.PGDUMP:
 		typeMap = postgresTypeMap
+	case constants.SQLSERVER:
+		typeMap = sqlserverTypeMap
 	default:
 		http.Error(w, fmt.Sprintf("Driver : '%s' is not supported", sessionState.driver), http.StatusBadRequest)
 		return
@@ -1037,6 +1045,8 @@ func getType(newType, table, colName string, srcTableName string) (ddl.CreateTab
 		ty, issues = toSpannerTypeMySQL(srcCol.Type.Name, newType, srcCol.Type.Mods)
 	case constants.PGDUMP, constants.POSTGRES:
 		ty, issues = toSpannerTypePostgres(srcCol.Type.Name, newType, srcCol.Type.Mods)
+	case constants.SQLSERVER:
+		ty, issues = toSpannerTypeSQLserver(srcCol.Type.Name, newType, srcCol.Type.Mods)
 	default:
 		return sp, ty, fmt.Errorf("driver : '%s' is not supported", sessionState.driver)
 	}
@@ -1164,6 +1174,17 @@ func init() {
 		}
 		postgresTypeMap[srcType] = l
 	}
+
+	// Initialize sqlserverTypeMap.
+	for _, srcType := range []string{"int", "tinyint", "smallint", "bigint", "bit", "float", "real", "numeric", "decimal", "money", "smallmoney", "char", "nchar", "varchar", "nvarchar", "text", "ntext", "date", "datetime", "datetime2", "smalldatetime", "datetimeoffset", "time", "timestamp", "rowversion", "binary", "varbinary", "image", "xml", "geography", "geometry", "uniqueidentifier", "sql_variant", "hierarchyid"} {
+		var l []typeIssue
+		for _, spType := range []string{ddl.Bool, ddl.Bytes, ddl.Date, ddl.Float64, ddl.Int64, ddl.String, ddl.Timestamp, ddl.Numeric} {
+			ty, issues := toSpannerTypeSQLserver(srcType, spType, []int64{})
+			l = addTypeToList(ty.Name, spType, issues, l)
+		}
+		sqlserverTypeMap[srcType] = l
+	}
+
 	sessionState.conv = internal.MakeConv()
 }
 


Test patch:
diff --git a/.github/workflows/integration-tests-against-emulator.yaml b/.github/workflows/integration-tests-against-emulator.yaml
index 96688fd..3df45d3 100644
--- a/.github/workflows/integration-tests-against-emulator.yaml
+++ b/.github/workflows/integration-tests-against-emulator.yaml
@@ -29,6 +29,10 @@ jobs:
       AWS_REGION: dummyRegion
       DYNAMODB_ENDPOINT_OVERRIDE: http://localhost:8000
 
+      # sql server envs
+      # Password for connection as SA
+      MSSQL_SA_PASSWORD : tCUE9c1&Ucp0 
+
     services:
       spanner_emulator:
         image: gcr.io/cloud-spanner-emulator/emulator:1.4.0
@@ -59,6 +63,17 @@ jobs:
         ports:
         - 8000:8000
         options: --workdir /home/dynamodblocal --health-cmd "curl --fail http://127.0.0.1:8000/shell/ || exit 1" --health-interval 10s --health-timeout 5s --health-retries 5
+      sqlserver:
+        image: mcr.microsoft.com/mssql/server
+        env:
+          SA_PASSWORD: ${{env.MSSQL_SA_PASSWORD}}
+          MSSQL_PID: Express
+          ACCEPT_EULA: Y
+        ports:
+          - 1433:1433
+        options:
+          --health-cmd "/opt/mssql-tools/bin/sqlcmd -U sa -P $SA_PASSWORD -Q 'select 1' -b -o /dev/null"
+          --health-interval 10s --health-timeout 5s --health-retries 3
 
     steps:
       - uses: actions/checkout@v2
@@ -78,6 +93,17 @@ jobs:
       - run: mysql -v -P 3306 --protocol=tcp -u root -proot test < test_data/mysqldump.test.out
       - run: mysql -v -P 3306 --protocol=tcp -u root -proot < test_data/mysql_interleave_dump.test.out
 
+      # init sql server with test_data
+      - name: Install sqlcmd required for loading .sql files
+        run: |
+          curl https://packages.microsoft.com/keys/microsoft.asc | sudo apt-key add -
+          curl https://packages.microsoft.com/config/ubuntu/16.04/prod.list | sudo tee /etc/apt/sources.list.d/msprod.list
+          sudo apt-get update 
+          sudo apt-get install mssql-tools unixodbc-dev
+          echo 'export PATH="$PATH:/opt/mssql-tools/bin"' >> ~/.bash_profile
+      - run: sqlcmd -?
+      - run: sqlcmd -U sa -P ${MSSQL_SA_PASSWORD} -i test_data/sqlserver.test.out
+
       # create a spanner instance
       - uses: google-github-actions/setup-gcloud@master
         with:
diff --git a/sources/sqlserver/data_test.go b/sources/sqlserver/data_test.go
new file mode 100644
index 0000000..4e43c9b
--- /dev/null
+++ b/sources/sqlserver/data_test.go
@@ -0,0 +1,306 @@
+// Copyright 2021 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 sqlserver
+
+import (
+	"fmt"
+	"math/big"
+	"math/bits"
+	"testing"
+	"time"
+
+	"cloud.google.com/go/civil"
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/schema"
+	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
+	"github.com/stretchr/testify/assert"
+)
+
+type spannerData struct {
+	table string
+	cols  []string
+	vals  []interface{}
+}
+
+// Basic smoke test of ProcessDataRow. The core part of this code path
+// (ConvertData) is tested in TestConvertData.
+func TestProcessDataRow(t *testing.T) {
+	tableName := "testtable"
+	cols := []string{"a", "b", "c"}
+	conv := buildConv(
+		ddl.CreateTable{
+			Name:     tableName,
+			ColNames: cols,
+			ColDefs: map[string]ddl.ColumnDef{
+				"a": {Name: "a", T: ddl.Type{Name: ddl.Float64}},
+				"b": {Name: "b", T: ddl.Type{Name: ddl.Int64}},
+				"c": {Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+			}},
+		schema.Table{
+			Name:     tableName,
+			ColNames: cols,
+			ColDefs: map[string]schema.Column{
+				"a": {Name: "a", Type: schema.Type{Name: "float"}},
+				"b": {Name: "b", Type: schema.Type{Name: "int"}},
+				"c": {Name: "c", Type: schema.Type{Name: "text"}},
+			}})
+	conv.SetDataMode()
+	var rows []spannerData
+	conv.SetDataSink(func(table string, cols []string, vals []interface{}) {
+		rows = append(rows, spannerData{table: table, cols: cols, vals: vals})
+	})
+	ProcessDataRow(conv, tableName, cols, conv.SrcSchema[tableName], tableName, cols, conv.SpSchema[tableName], []string{"4.2", "6", "prisoner zero"})
+	assert.Equal(t, []spannerData{{table: tableName, cols: cols, vals: []interface{}{float64(4.2), int64(6), "prisoner zero"}}}, rows)
+}
+
+func TestConvertData(t *testing.T) {
+
+	numStr := "33753785954695469456094506940.33333333982343435"
+	numVal := new(big.Rat)
+	numVal.SetString(numStr)
+
+	singleColTests := []struct {
+		name  string
+		ty    ddl.Type
+		srcTy string      // Source DB type
+		in    string      // Input value for conversion.
+		e     interface{} // Expected result.
+	}{
+		{"bit 0", ddl.Type{Name: ddl.Bool}, "", "0", false},
+		{"bit 1", ddl.Type{Name: ddl.Bool}, "", "1", true},
+		{"bytes", ddl.Type{Name: ddl.Bytes, Len: ddl.MaxLength}, "", string([]byte{137, 80}), []byte{0x89, 0x50}}, // need some other approach to testblob type
+		{"date", ddl.Type{Name: ddl.Date}, "", "2019-10-29", getDate("2019-10-29")},
+		{"float64", ddl.Type{Name: ddl.Float64}, "", "42.6", float64(42.6)},
+		{"int64", ddl.Type{Name: ddl.Int64}, "", "42", int64(42)},
+		{"string", ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, "", "eh", "eh"},
+		{"datetime", ddl.Type{Name: ddl.Timestamp}, "datetime", "2019-10-29T05:30:00", getTimeWithoutTimezone(t, "2019-10-29T05:30:00")},
+		{"datetimeoffset", ddl.Type{Name: ddl.Timestamp}, "datetimeoffset", "2021-12-15T07:39:52.9433333+01:20", getTimeWithTimezone(t, "2021-12-15T07:39:52.9433333+01:20")},
+		{"decimal", ddl.Type{Name: ddl.Numeric}, "decimal", "234.90909090909", big.NewRat(23490909090909, 100000000000)},
+		{"numeric", ddl.Type{Name: ddl.Numeric}, "numeric", numStr, numVal},
+	}
+	tableName := "testtable"
+	for _, tc := range singleColTests {
+		col := "a"
+		conv := buildConv(
+			ddl.CreateTable{
+				Name:     tableName,
+				ColNames: []string{col},
+				ColDefs:  map[string]ddl.ColumnDef{col: {Name: col, T: tc.ty, NotNull: false}},
+				Pks:      []ddl.IndexKey{}},
+			schema.Table{Name: tableName, ColNames: []string{col}, ColDefs: map[string]schema.Column{col: {Type: schema.Type{Name: tc.srcTy}}}})
+		conv.TimezoneOffset = "+05:30"
+		t.Run(tc.in, func(t *testing.T) {
+			at, ac, av, err := ConvertData(conv, tableName, []string{col}, conv.SrcSchema[tableName], tableName, []string{col}, conv.SpSchema[tableName], []string{tc.in})
+			checkResults(t, at, ac, av, err, tableName, []string{col}, []interface{}{tc.e}, tc.name)
+		})
+	}
+}
+
+func TestConvertMultiColData(t *testing.T) {
+	multiColTests := []struct {
+		name  string
+		cols  []string      // Input columns.
+		vals  []string      // Input values.
+		ecols []string      // Expected columns.
+		evals []interface{} // Expected values.
+	}{
+		{
+			name:  "Cols in order",
+			cols:  []string{"a", "b", "c"},
+			vals:  []string{"6", "6.6", "1"},
+			ecols: []string{"a", "b", "c"},
+			evals: []interface{}{int64(6), float64(6.6), true},
+		},
+		{
+			name:  "Cols out of order",
+			cols:  []string{"b", "c", "a"},
+			vals:  []string{"6.6", "1", "6"},
+			ecols: []string{"b", "c", "a"},
+			evals: []interface{}{float64(6.6), true, int64(6)},
+		},
+		{
+			name:  "Null column",
+			cols:  []string{"a", "b", "c"},
+			vals:  []string{"6", "NULL", "1"},
+			ecols: []string{"a", "c"},
+			evals: []interface{}{int64(6), true},
+		},
+		{
+			name:  "Missing columns",
+			cols:  []string{"a"},
+			vals:  []string{"6"},
+			ecols: []string{"a"},
+			evals: []interface{}{int64(6)},
+		},
+	}
+	tableName := "testtable"
+	spTable := ddl.CreateTable{
+		Name:     tableName,
+		ColNames: []string{"a", "b", "c"},
+		ColDefs: map[string]ddl.ColumnDef{
+			"a": {Name: "a", T: ddl.Type{Name: ddl.Int64}},
+			"b": {Name: "b", T: ddl.Type{Name: ddl.Float64}},
+			"c": {Name: "c", T: ddl.Type{Name: ddl.Bool}},
+		}}
+	srcTable := schema.Table{
+		Name:     tableName,
+		ColNames: []string{"a", "b", "c"},
+		ColDefs: map[string]schema.Column{
+			"a": {Type: schema.Type{Name: "int"}},
+			"b": {Type: schema.Type{Name: "float"}},
+			"c": {Type: schema.Type{Name: "bool"}},
+		}}
+	for _, tc := range multiColTests {
+		t.Run(tc.name, func(t *testing.T) {
+			conv := buildConv(spTable, srcTable)
+			atable, acols, avals, err := ConvertData(conv, srcTable.Name, tc.cols, conv.SrcSchema[tableName], spTable.Name, tc.cols, conv.SpSchema[tableName], tc.vals)
+			checkResults(t, atable, acols, avals, err, tableName, tc.ecols, tc.evals, tc.name)
+		})
+	}
+}
+
+func TestConvertError(t *testing.T) {
+	errorTests := []struct {
+		name string
+		cols []string // Input columns.
+		vals []string // Input values.
+	}{
+		{
+			name: "Error in int64",
+			cols: []string{"a", "b", "c"},
+			vals: []string{" 6", "6.6", "true"},
+		},
+		{
+			name: "Error in float64",
+			cols: []string{"a", "b", "c"},
+			vals: []string{"6", "6.6e", "true"},
+		},
+		{
+			name: "Error in bool",
+			cols: []string{"a", "b", "c"},
+			vals: []string{"6", "6.6", "truee"},
+		},
+		{
+			name: "Error in bool 128",
+			cols: []string{"a", "b", "c"},
+			vals: []string{"6", "6.6", "128"},
+		},
+	}
+	tableName := "testtable"
+	spTable := ddl.CreateTable{
+		Name:     tableName,
+		ColNames: []string{"a", "b", "c"},
+		ColDefs: map[string]ddl.ColumnDef{
+			"a": {Name: "a", T: ddl.Type{Name: ddl.Int64}},
+			"b": {Name: "b", T: ddl.Type{Name: ddl.Float64}},
+			"c": {Name: "c", T: ddl.Type{Name: ddl.Bool}},
+		}}
+	srcTable := schema.Table{
+		Name:     tableName,
+		ColNames: []string{"a", "b", "c"},
+		ColDefs: map[string]schema.Column{
+			"a": {Type: schema.Type{Name: "int"}},
+			"b": {Type: schema.Type{Name: "float"}},
+			"c": {Type: schema.Type{Name: "bool"}},
+		}}
+	for _, tc := range errorTests {
+		t.Run(tc.name, func(t *testing.T) {
+			conv := buildConv(spTable, srcTable)
+			_, _, _, err := ConvertData(conv, srcTable.Name, tc.cols, conv.SrcSchema[tableName], spTable.Name, tc.cols, conv.SpSchema[tableName], tc.vals)
+			assert.NotNil(t, err, tc.name)
+		})
+	}
+}
+
+func TestConvertsyntheticPKey(t *testing.T) {
+	syntheticPKeyTests := []struct {
+		name  string
+		cols  []string      // Input columns.
+		vals  []string      // Input values.
+		ecols []string      // Expected columns.
+		evals []interface{} // Expected values.
+	}{
+		{
+			name:  "Sequence 0",
+			cols:  []string{"a", "b", "c"},
+			vals:  []string{"6", "6.6", "true"},
+			ecols: []string{"a", "b", "c", "synth_id"},
+			evals: []interface{}{int64(6), float64(6.6), true, int64(0)},
+		},
+		{
+			name:  "Sequence 1",
+			cols:  []string{"a"},
+			vals:  []string{"7"},
+			ecols: []string{"a", "synth_id"},
+			evals: []interface{}{int64(7), int64(bits.Reverse64(1))},
+		},
+	}
+	tableName := "testtable"
+	spTable := ddl.CreateTable{
+		Name:     tableName,
+		ColNames: []string{"a", "b", "c"},
+		ColDefs: map[string]ddl.ColumnDef{
+			"a": {Name: "a", T: ddl.Type{Name: ddl.Int64}},
+			"b": {Name: "b", T: ddl.Type{Name: ddl.Float64}},
+			"c": {Name: "c", T: ddl.Type{Name: ddl.Bool}},
+		}}
+	srcTable := schema.Table{
+		Name:     tableName,
+		ColNames: []string{"a", "b", "c"},
+		ColDefs: map[string]schema.Column{
+			"a": {Type: schema.Type{Name: "int"}},
+			"b": {Type: schema.Type{Name: "float"}},
+			"c": {Type: schema.Type{Name: "bool"}},
+		}}
+	conv := buildConv(spTable, srcTable)
+	conv.SyntheticPKeys[spTable.Name] = internal.SyntheticPKey{Col: "synth_id", Sequence: 0}
+	for _, tc := range syntheticPKeyTests {
+		t.Run(tc.name, func(t *testing.T) {
+			atable, acols, avals, err := ConvertData(conv, srcTable.Name, tc.cols, conv.SrcSchema[tableName], spTable.Name, tc.cols, conv.SpSchema[tableName], tc.vals)
+			checkResults(t, atable, acols, avals, err, tableName, tc.ecols, tc.evals, tc.name)
+		})
+	}
+}
+
+func buildConv(spTable ddl.CreateTable, srcTable schema.Table) *internal.Conv {
+	conv := internal.MakeConv()
+	conv.SpSchema[spTable.Name] = spTable
+	conv.SrcSchema[srcTable.Name] = srcTable
+	return conv
+}
+
+func checkResults(t *testing.T, atable string, acols []string, avals []interface{}, err error, etable string, ecols []string, evals []interface{}, name string) {
+	assert.Nil(t, err, name)
+	assert.Equal(t, atable, etable, name+": table mismatch")
+	assert.Equal(t, ecols, acols, name+": column mismatch")
+	assert.Equal(t, evals, avals, name+": value mismatch")
+}
+
+func getDate(s string) civil.Date {
+	d, _ := civil.ParseDate(s)
+	return d
+}
+
+func getTimeWithoutTimezone(t *testing.T, s string) time.Time {
+	x, err := time.Parse("2006-01-02T15:04:05", s)
+	assert.Nil(t, err, fmt.Sprintf("getTime can't parse %s:", s))
+	return x
+}
+
+func getTimeWithTimezone(t *testing.T, s string) time.Time {
+	x, err := time.Parse(time.RFC3339, s)
+	assert.Nil(t, err, fmt.Sprintf("getTime can't parse %s:", s))
+	return x
+}
diff --git a/sources/sqlserver/infoschema_test.go b/sources/sqlserver/infoschema_test.go
new file mode 100644
index 0000000..e392497
--- /dev/null
+++ b/sources/sqlserver/infoschema_test.go
@@ -0,0 +1,356 @@
+// Copyright 2021 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 sqlserver
+
+import (
+	"database/sql"
+	"database/sql/driver"
+	"testing"
+
+	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
+	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
+	"github.com/stretchr/testify/assert"
+)
+
+type mockSpec struct {
+	query string
+	args  []driver.Value   // Query args.
+	cols  []string         // Columns names for returned rows.
+	rows  [][]driver.Value // Set of rows returned.
+}
+
+func TestProcessSchema(t *testing.T) {
+	ms := []mockSpec{
+		{
+			query: `SELECT (.+) WHERE TBL.type = 'U' AND TBL.is_tracked_by_cdc = 0 AND TBL.is_ms_shipped = 0 AND TBL.name <> 'sysdiagrams'`,
+			cols:  []string{"table_schema", "table_name"},
+			rows: [][]driver.Value{
+				{"dbo", "user"},
+				{"dbo", "test"},
+				{"dbo", "cart"},
+				{"production", "product"},
+				{"dbo", "test_ref"},
+			},
+		}, {
+			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
+			args:  []driver.Value{"dbo", "user"},
+			cols:  []string{"column_name", "constraint_type"},
+			rows: [][]driver.Value{
+				{"user_id", "PRIMARY KEY"},
+				{"ref", "FOREIGN KEY"}},
+		}, {
+			query: "SELECT (.+) FROM sys.foreign_keys AS FK (.+)",
+			args:  []driver.Value{"dbo.user"},
+			cols:  []string{"TABLE_SCHEMA", "TABLE_NAME", "COLUMN_NAME", "REF_COLUMN_NAME", "CONSTRAINT_NAME"},
+			rows: [][]driver.Value{
+				{"dbo", "test", "ref", "Id", "fk_test"},
+			},
+		}, {
+			query: "SELECT (.+) FROM sys.indexes (.+)",
+			args:  []driver.Value{"user", "dbo"},
+			cols:  []string{"index_name", "column_name", "column_position", "is_unique", "order"},
+		}, {
+			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
+			args:  []driver.Value{"dbo", "user"},
+			cols:  []string{"column_name", "data_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale"},
+			rows: [][]driver.Value{
+				{"user_id", "text", "NO", nil, nil, nil, nil},
+				{"name", "text", "NO", nil, nil, nil, nil},
+				{"ref", "bigint", "YES", nil, nil, nil, nil}},
+		}, {
+			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
+			args:  []driver.Value{"dbo", "test"},
+			cols:  []string{"column_name", "constraint_type"},
+			rows: [][]driver.Value{
+				{"Id", "PRIMARY KEY"},
+			},
+		}, {
+			query: "SELECT (.+) FROM sys.foreign_keys AS FK (.+)",
+			args:  []driver.Value{"dbo.test"},
+			cols:  []string{"TABLE_SCHEMA", "TABLE_NAME", "COLUMN_NAME", "REF_COLUMN_NAME", "CONSTRAINT_NAME"},
+			rows:  [][]driver.Value{{"dbo", "test_ref", "id", "ref_id", "fk_test4"}},
+		}, {
+			query: "SELECT (.+) FROM sys.indexes (.+)",
+			args:  []driver.Value{"test", "dbo"},
+			cols:  []string{"index_name", "column_name", "column_position", "is_unique", "order"},
+		}, {
+			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
+			args:  []driver.Value{"dbo", "test"},
+			cols:  []string{"column_name", "data_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale"},
+			rows: [][]driver.Value{
+				{"Id", "int", "NO", nil, nil, 10, 0},
+				{"BigInt", "bigint", "YES", nil, nil, 19, 0},
+				{"Binary", "binary", "YES", nil, 50, nil, nil},
+				{"Bit", "bit", "YES", nil, nil, nil, nil},
+				{"Char", "char", "YES", nil, 10, nil, nil},
+				{"Date", "date", "YES", nil, nil, nil, nil},
+				{"DateTime", "datetime", "YES", nil, nil, nil, nil},
+				{"DateTime2", "datetime2", "YES", nil, nil, nil, nil},
+				{"DateTimeOffset", "datetimeoffset", "YES", nil, nil, nil, nil},
+				{"Decimal", "decimal", "YES", nil, nil, 18, 9},
+				{"Float", "float", "YES", nil, nil, 53, nil},
+				{"Geography", "geography", "YES", nil, -1, nil, nil},
+				{"Geometry", "geometry", "YES", nil, -1, nil, nil},
+				{"HierarchyId", "hierarchyid", "YES", nil, 892, nil, nil},
+				{"Image", "image", "YES", nil, 2147483647, nil, nil},
+				{"Int", "int", "YES", nil, nil, 10, 0},
+				{"Money", "money", "YES", nil, nil, 19, 4},
+				{"NChar", "nchar", "YES", nil, 10, nil, nil},
+				{"NText", "ntext", "YES", nil, 1073741823, nil, nil},
+				{"Numeric", "numeric", "YES", nil, nil, 18, 17},
+				{"NVarChar", "nvarchar", "YES", nil, 50, nil, nil},
+				{"NVarCharMax", "nvarchar", "YES", nil, -1, nil, nil},
+				{"Real", "real", "YES", nil, nil, 24, nil},
+				{"SmallDateTime", "smalldatetime", "YES", nil, nil, nil, nil},
+				{"SmallInt", "smallint", "YES", nil, nil, 5, 0},
+				{"SmallMoney", "smallmoney", "YES", nil, nil, 10, 4},
+				{"SQLVariant", "sql_variant", "YES", nil, 0, nil, nil},
+				{"Text", "text", "YES", nil, 2147483647, nil, nil},
+				{"Time", "time", "YES", nil, nil, nil, nil},
+				{"TimeStamp", "timestamp", "YES", nil, nil, nil, nil},
+				{"TinyInt", "tinyint", "YES", nil, nil, 3, 0},
+				{"UniqueIdentifier", "uniqueidentifier", "YES", nil, nil, nil, nil},
+				{"VarBinary", "varbinary", "YES", nil, 50, nil, nil},
+				{"VarBinaryMax", "varbinary", "YES", nil, -1, nil, nil},
+				{"VarChar", "varchar", "YES", nil, 50, nil, nil},
+				{"VarCharMax", "varchar", "YES", nil, -1, nil, nil},
+				{"Xml", "xml", "YES", nil, -1, nil, nil},
+			},
+		},
+
+		{
+			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
+			args:  []driver.Value{"dbo", "cart"},
+			cols:  []string{"column_name", "constraint_type"},
+			rows: [][]driver.Value{
+				{"productid", "PRIMARY KEY"},
+				{"userid", "PRIMARY KEY"},
+			},
+		}, {
+			query: "SELECT (.+) FROM sys.foreign_keys AS FK (.+)",
+			args:  []driver.Value{"dbo.cart"},
+			cols:  []string{"TABLE_SCHEMA", "TABLE_NAME", "COLUMN_NAME", "REF_COLUMN_NAME", "CONSTRAINT_NAME"},
+			rows: [][]driver.Value{
+				{"production", "product", "productid", "product_id", "fk_test2"},
+				{"dbo", "user", "userid", "user_id", "fk_test3"}},
+		}, {
+			query: "SELECT (.+) FROM sys.indexes (.+)",
+			args:  []driver.Value{"cart", "dbo"},
+			cols:  []string{"index_name", "column_name", "is_unique", "order"},
+			rows: [][]driver.Value{{"index1", "userid", "false", "ASC"},
+				{"index2", "userid", "true", "ASC"},
+				{"index2", "productid", "true", "DESC"},
+				{"index3", "productid", "true", "DESC"},
+				{"index3", "userid", "true", "ASC"},
+			},
+		}, {
+			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
+			args:  []driver.Value{"dbo", "cart"},
+			cols:  []string{"column_name", "data_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale"},
+			rows: [][]driver.Value{
+				{"productid", "text", "NO", nil, nil, nil, nil},
+				{"userid", "text", "NO", nil, nil, nil, nil},
+				{"quantity", "bigint", "YES", nil, nil, 64, 0}},
+		},
+
+		{
+			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
+			args:  []driver.Value{"production", "product"},
+			cols:  []string{"column_name", "constraint_type"},
+			rows: [][]driver.Value{
+				{"product_id", "PRIMARY KEY"},
+			},
+		}, {
+			query: "SELECT (.+) FROM sys.foreign_keys AS FK (.+)",
+			args:  []driver.Value{"production.product"},
+			cols:  []string{"TABLE_SCHEMA", "TABLE_NAME", "COLUMN_NAME", "REF_COLUMN_NAME", "CONSTRAINT_NAME"},
+		}, {
+			query: "SELECT (.+) FROM sys.indexes (.+)",
+			args:  []driver.Value{"product", "production"},
+			cols:  []string{"index_name", "column_name", "column_position", "is_unique", "order"},
+		}, {
+			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
+			args:  []driver.Value{"production", "product"},
+			cols:  []string{"column_name", "data_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale"},
+			rows: [][]driver.Value{
+				{"product_id", "text", "NO", nil, nil, nil, nil},
+				{"product_name", "text", "NO", nil, nil, nil, nil},
+			},
+		},
+
+		{
+			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
+			args:  []driver.Value{"dbo", "test_ref"},
+			cols:  []string{"column_name", "constraint_type"},
+			rows: [][]driver.Value{
+				{"ref_id", "PRIMARY KEY"},
+				{"ref_txt", "PRIMARY KEY"},
+			},
+		}, {
+			query: "SELECT (.+) FROM sys.foreign_keys AS FK (.+)",
+			args:  []driver.Value{"dbo.test_ref"},
+			cols:  []string{"TABLE_SCHEMA", "TABLE_NAME", "COLUMN_NAME", "REF_COLUMN_NAME", "CONSTRAINT_NAME"},
+		}, {
+			query: "SELECT (.+) FROM sys.indexes (.+)",
+			args:  []driver.Value{"test_ref", "dbo"},
+			cols:  []string{"index_name", "column_name", "column_position", "is_unique", "order"},
+		}, {
+			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
+			args:  []driver.Value{"dbo", "test_ref"},
+			cols:  []string{"column_name", "data_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale"},
+			rows: [][]driver.Value{
+				{"ref_id", "bigint", "NO", nil, nil, 64, 0},
+				{"ref_txt", "text", "NO", nil, nil, nil, nil},
+				{"abc", "text", "NO", nil, nil, nil, nil},
+			},
+		},
+	}
+	db := mkMockDB(t, ms)
+	conv := internal.MakeConv()
+	err := common.ProcessSchema(conv, InfoSchemaImpl{"test", db})
+	assert.Nil(t, err)
+	expectedSchema := map[string]ddl.CreateTable{
+		"user": {
+			Name:     "user",
+			ColNames: []string{"user_id", "name", "ref"},
+			ColDefs: map[string]ddl.ColumnDef{
+				"user_id": {Name: "user_id", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, NotNull: true},
+				"name":    {Name: "name", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, NotNull: true},
+				"ref":     {Name: "ref", T: ddl.Type{Name: ddl.Int64}},
+			},
+			Pks: []ddl.IndexKey{{Col: "user_id"}},
+			Fks: []ddl.Foreignkey{{Name: "fk_test", Columns: []string{"ref"}, ReferTable: "test", ReferColumns: []string{"Id"}}}},
+		"test": {
+			Name: "test",
+			ColNames: []string{"Id", "BigInt", "Binary", "Bit", "Char", "Date", "DateTime",
+				"DateTime2", "DateTimeOffset", "Decimal", "Float", "Geography", "Geometry", "HierarchyId",
+				"Image", "Int", "Money", "NChar", "NText", "Numeric", "NVarChar", "NVarCharMax", "Real", "SmallDateTime",
+				"SmallInt", "SmallMoney", "SQLVariant", "Text", "Time", "TimeStamp",
+				"TinyInt", "UniqueIdentifier", "VarBinary", "VarBinaryMax", "VarChar", "VarCharMax", "Xml"},
+			ColDefs: map[string]ddl.ColumnDef{
+				"Id":               {Name: "Id", T: ddl.Type{Name: ddl.Int64}, NotNull: true},
+				"BigInt":           {Name: "BigInt", T: ddl.Type{Name: ddl.Int64}, NotNull: false},
+				"Binary":           {Name: "Binary", T: ddl.Type{Name: ddl.Bytes, Len: ddl.MaxLength}, NotNull: false},
+				"Bit":              {Name: "Bit", T: ddl.Type{Name: ddl.Bool}, NotNull: false},
+				"Char":             {Name: "Char", T: ddl.Type{Name: ddl.String, Len: 10, IsArray: false}, NotNull: false},
+				"Date":             {Name: "Date", T: ddl.Type{Name: ddl.Date}, NotNull: false},
+				"DateTime":         {Name: "DateTime", T: ddl.Type{Name: ddl.Timestamp}, NotNull: false},
+				"DateTime2":        {Name: "DateTime2", T: ddl.Type{Name: ddl.Timestamp}, NotNull: false},
+				"DateTimeOffset":   {Name: "DateTimeOffset", T: ddl.Type{Name: ddl.Timestamp}, NotNull: false},
+				"Decimal":          {Name: "Decimal", T: ddl.Type{Name: ddl.Numeric}, NotNull: false},
+				"Float":            {Name: "Float", T: ddl.Type{Name: ddl.Float64}, NotNull: false},
+				"Geography":        {Name: "Geography", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, NotNull: false},
+				"Geometry":         {Name: "Geometry", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, NotNull: false},
+				"HierarchyId":      {Name: "HierarchyId", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, NotNull: false},
+				"Image":            {Name: "Image", T: ddl.Type{Name: ddl.Bytes, Len: ddl.MaxLength}, NotNull: false},
+				"Int":              {Name: "Int", T: ddl.Type{Name: ddl.Int64}, NotNull: false},
+				"Money":            {Name: "Money", T: ddl.Type{Name: ddl.Numeric}, NotNull: false},
+				"NChar":            {Name: "NChar", T: ddl.Type{Name: ddl.String, Len: 10}, NotNull: false},
+				"NText":            {Name: "NText", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, NotNull: false},
+				"Numeric":          {Name: "Numeric", T: ddl.Type{Name: ddl.Numeric}, NotNull: false},
+				"NVarChar":         {Name: "NVarChar", T: ddl.Type{Name: ddl.String, Len: 50}, NotNull: false},
+				"NVarCharMax":      {Name: "NVarCharMax", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, NotNull: false},
+				"Real":             {Name: "Real", T: ddl.Type{Name: ddl.Float64}, NotNull: false},
+				"SmallDateTime":    {Name: "SmallDateTime", T: ddl.Type{Name: ddl.Timestamp}, NotNull: false},
+				"SmallInt":         {Name: "SmallInt", T: ddl.Type{Name: ddl.Int64}, NotNull: false},
+				"SmallMoney":       {Name: "SmallMoney", T: ddl.Type{Name: ddl.Numeric}, NotNull: false},
+				"SQLVariant":       {Name: "SQLVariant", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, NotNull: false},
+				"Text":             {Name: "Text", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, NotNull: false},
+				"Time":             {Name: "Time", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, NotNull: false},
+				"TimeStamp":        {Name: "TimeStamp", T: ddl.Type{Name: ddl.Int64}, NotNull: false},
+				"TinyInt":          {Name: "TinyInt", T: ddl.Type{Name: ddl.Int64}, NotNull: false},
+				"UniqueIdentifier": {Name: "UniqueIdentifier", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, NotNull: false},
+				"VarBinary":        {Name: "VarBinary", T: ddl.Type{Name: ddl.Bytes, Len: ddl.MaxLength}, NotNull: false},
+				"VarBinaryMax":     {Name: "VarBinaryMax", T: ddl.Type{Name: ddl.Bytes, Len: ddl.MaxLength}, NotNull: false},
+				"VarChar":          {Name: "VarChar", T: ddl.Type{Name: ddl.String, Len: 50}, NotNull: false},
+				"VarCharMax":       {Name: "VarCharMax", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, NotNull: false},
+				"Xml":              {Name: "Xml", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, NotNull: false},
+			},
+			Pks: []ddl.IndexKey{{Col: "Id"}},
+			Fks: []ddl.Foreignkey{{Name: "fk_test4", Columns: []string{"Id"}, ReferTable: "test_ref", ReferColumns: []string{"ref_id"}}},
+		},
+		"cart": {
+			Name:     "cart",
+			ColNames: []string{"productid", "userid", "quantity"},
+			ColDefs: map[string]ddl.ColumnDef{
+				"productid": {Name: "productid", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, NotNull: true},
+				"userid":    {Name: "userid", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, NotNull: true},
+				"quantity":  {Name: "quantity", T: ddl.Type{Name: ddl.Int64}},
+			},
+			Pks: []ddl.IndexKey{{Col: "productid"}, {Col: "userid"}},
+			Fks: []ddl.Foreignkey{{Name: "fk_test2", Columns: []string{"productid"}, ReferTable: "production_product", ReferColumns: []string{"product_id"}},
+				{Name: "fk_test3", Columns: []string{"userid"}, ReferTable: "user", ReferColumns: []string{"user_id"}}},
+			Indexes: []ddl.CreateIndex{{Name: "index1", Table: "cart", Unique: false, Keys: []ddl.IndexKey{{Col: "userid", Desc: false}}},
+				{Name: "index2", Table: "cart", Unique: true, Keys: []ddl.IndexKey{{Col: "userid", Desc: false}, {Col: "productid", Desc: true}}},
+				{Name: "index3", Table: "cart", Unique: true, Keys: []ddl.IndexKey{{Col: "productid", Desc: true}, {Col: "userid", Desc: false}}}}},
+		"production_product": {
+			Name:     "production_product",
+			ColNames: []string{"product_id", "product_name"},
+			ColDefs: map[string]ddl.ColumnDef{
+				"product_id":   {Name: "product_id", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, NotNull: true},
+				"product_name": {Name: "product_name", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, NotNull: true},
+			},
+			Pks: []ddl.IndexKey{{Col: "product_id"}}},
+		"test_ref": {
+			Name:     "test_ref",
+			ColNames: []string{"ref_id", "ref_txt", "abc"},
+			ColDefs: map[string]ddl.ColumnDef{
+				"ref_id":  {Name: "ref_id", T: ddl.Type{Name: ddl.Int64}, NotNull: true},
+				"ref_txt": {Name: "ref_txt", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, NotNull: true},
+				"abc":     {Name: "abc", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, NotNull: true},
+			},
+			Pks: []ddl.IndexKey{{Col: "ref_id"}, {Col: "ref_txt"}}},
+	}
+	assert.Equal(t, expectedSchema, stripSchemaComments(conv.SpSchema))
+	assert.Equal(t, len(conv.Issues["cart"]), 0)
+	assert.Equal(t, len(conv.Issues["test"]), 17)
+	assert.Equal(t, int64(0), conv.Unexpecteds())
+
+}
+
+func mkMockDB(t *testing.T, ms []mockSpec) *sql.DB {
+	db, mock, err := sqlmock.New()
+	assert.Nil(t, err)
+	for _, m := range ms {
+		rows := sqlmock.NewRows(m.cols)
+		for _, r := range m.rows {
+			rows.AddRow(r...)
+		}
+		if len(m.args) > 0 {
+			mock.ExpectQuery(m.query).WithArgs(m.args...).WillReturnRows(rows)
+		} else {
+			mock.ExpectQuery(m.query).WillReturnRows(rows)
+		}
+
+	}
+	return db
+}
+
+// stripSchemaComments returns a schema with all comments removed.
+// We mostly ignore schema comments in testing since schema comments
+// are often changed and are not a core part of conversion functionality.
+func stripSchemaComments(spSchema map[string]ddl.CreateTable) map[string]ddl.CreateTable {
+	for t, ct := range spSchema {
+		for c, cd := range ct.ColDefs {
+			cd.Comment = ""
+			ct.ColDefs[c] = cd
+		}
+		ct.Comment = ""
+		spSchema[t] = ct
+	}
+	return spSchema
+}
diff --git a/sources/sqlserver/toddl_test.go b/sources/sqlserver/toddl_test.go
new file mode 100644
index 0000000..f8eb790
--- /dev/null
+++ b/sources/sqlserver/toddl_test.go
@@ -0,0 +1,121 @@
+// Copyright 2021 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 sqlserver
+
+import (
+	"testing"
+
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/schema"
+	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
+	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
+	"github.com/stretchr/testify/assert"
+)
+
+// This is just a very basic smoke-test for toSpannerType.
+func TestToSpannerType(t *testing.T) {
+	conv := internal.MakeConv()
+	conv.SetSchemaMode()
+	name := "test"
+	srcSchema := schema.Table{
+		Name:     name,
+		ColNames: []string{"a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k", "l", "m"},
+		ColDefs: map[string]schema.Column{
+			"a": {Name: "a", Type: schema.Type{Name: "int"}},
+			"b": {Name: "b", Type: schema.Type{Name: "float"}},
+			"c": {Name: "c", Type: schema.Type{Name: "tinyint", Mods: []int64{1}}},
+			"d": {Name: "d", Type: schema.Type{Name: "varchar", Mods: []int64{6}}},
+			"e": {Name: "e", Type: schema.Type{Name: "numeric"}},
+			"f": {Name: "f", Type: schema.Type{Name: "timestamp"}},
+			"g": {Name: "g", Type: schema.Type{Name: "binary", Mods: []int64{4000}}},
+			"h": {Name: "h", Type: schema.Type{Name: "date"}},
+			"i": {Name: "i", Type: schema.Type{Name: "money"}},
+			"j": {Name: "j", Type: schema.Type{Name: "smalldatetime"}},
+			"k": {Name: "k", Type: schema.Type{Name: "nvarchar", Mods: []int64{50}}},
+			"l": {Name: "l", Type: schema.Type{Name: "image"}},
+			"m": {Name: "m", Type: schema.Type{Name: "geometry"}},
+		},
+		PrimaryKeys: []schema.Key{{Column: "a"}},
+		ForeignKeys: []schema.ForeignKey{{Name: "fk_test", Columns: []string{"d"}, ReferTable: "ref_table", ReferColumns: []string{"dref"}},
+			{Name: "fk_test2", Columns: []string{"a"}, ReferTable: "ref_table2", ReferColumns: []string{"aRef"}}},
+		Indexes: []schema.Index{{Name: "index1", Unique: true, Keys: []schema.Key{{Column: "a", Desc: false}, {Column: "d", Desc: true}}}},
+	}
+	conv.SrcSchema[name] = srcSchema
+	conv.SpSchema["ref_table"] = ddl.CreateTable{
+		Name:     "ref_table",
+		ColNames: []string{"dref", "b", "c"},
+		ColDefs: map[string]ddl.ColumnDef{
+			"dref": {Name: "dref", T: ddl.Type{Name: ddl.String, Len: int64(6)}},
+			"b":    {Name: "b", T: ddl.Type{Name: ddl.Float64}},
+			"c":    {Name: "c", T: ddl.Type{Name: ddl.Bool}},
+		},
+		Pks: []ddl.IndexKey{{Col: "dref"}},
+	}
+	conv.SpSchema["ref_table2"] = ddl.CreateTable{
+		Name:     "ref_table2",
+		ColNames: []string{"aref", "b", "c"},
+		ColDefs: map[string]ddl.ColumnDef{
+			"aref": {Name: "aref", T: ddl.Type{Name: ddl.Int64}},
+			"b":    {Name: "b", T: ddl.Type{Name: ddl.Float64}},
+			"c":    {Name: "c", T: ddl.Type{Name: ddl.Bool}},
+		},
+		Pks: []ddl.IndexKey{{Col: "aref"}},
+	}
+	assert.Nil(t, common.SchemaToSpannerDDL(conv, ToDdlImpl{}))
+	actual := conv.SpSchema[name]
+	dropComments(&actual) // Don't test comment.
+	expected := ddl.CreateTable{
+		Name:     name,
+		ColNames: []string{"a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k", "l", "m"},
+		ColDefs: map[string]ddl.ColumnDef{
+			"a": {Name: "a", T: ddl.Type{Name: ddl.Int64}},
+			"b": {Name: "b", T: ddl.Type{Name: ddl.Float64}},
+			"c": {Name: "c", T: ddl.Type{Name: ddl.Int64}},
+			"d": {Name: "d", T: ddl.Type{Name: ddl.String, Len: int64(6)}},
+			"e": {Name: "e", T: ddl.Type{Name: ddl.Numeric}},
+			"f": {Name: "f", T: ddl.Type{Name: ddl.Int64}},
+			"g": {Name: "g", T: ddl.Type{Name: ddl.Bytes, Len: ddl.MaxLength}},
+			"h": {Name: "h", T: ddl.Type{Name: ddl.Date}},
+			"i": {Name: "i", T: ddl.Type{Name: ddl.Numeric}},
+			"j": {Name: "j", T: ddl.Type{Name: ddl.Timestamp}},
+			"k": {Name: "k", T: ddl.Type{Name: ddl.String, Len: int64(50)}},
+			"l": {Name: "l", T: ddl.Type{Name: ddl.Bytes, Len: ddl.MaxLength}},
+			"m": {Name: "m", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+		},
+
+		Pks: []ddl.IndexKey{{Col: "a"}},
+		Fks: []ddl.Foreignkey{{Name: "fk_test", Columns: []string{"d"}, ReferTable: "ref_table", ReferColumns: []string{"dref"}},
+			{Name: "fk_test2", Columns: []string{"a"}, ReferTable: "ref_table2", ReferColumns: []string{"aref"}}},
+		Indexes: []ddl.CreateIndex{{Name: "index1", Table: name, Unique: true, Keys: []ddl.IndexKey{{Col: "a", Desc: false}, {Col: "d", Desc: true}}}},
+	}
+	assert.Equal(t, expected, actual)
+	expectedIssues := map[string][]internal.SchemaIssue{
+		"a": {internal.Widened},
+		"b": {internal.Widened},
+		"c": {internal.Widened},
+		"j": {internal.Timestamp},
+		"m": {internal.NoGoodType},
+	}
+	assert.Equal(t, expectedIssues, conv.Issues[name])
+}
+
+func dropComments(t *ddl.CreateTable) {
+	t.Comment = ""
+	for _, c := range t.ColNames {
+		cd := t.ColDefs[c]
+		cd.Comment = ""
+		t.ColDefs[c] = cd
+	}
+}
diff --git a/test_data/sqlserver.test.out b/test_data/sqlserver.test.out
new file mode 100644
index 0000000..342485f
--- /dev/null
+++ b/test_data/sqlserver.test.out
@@ -0,0 +1,108 @@
+CREATE DATABASE SqlServer_IntTest;
+Go
+USE [SqlServer_IntTest]
+GO
+/****** Object:  Table [dbo].[AllTypes]    Script Date: 12/29/2021 8:56:38 AM ******/
+SET ANSI_NULLS ON
+GO
+SET QUOTED_IDENTIFIER ON
+GO
+CREATE TABLE [dbo].[AllTypes](
+	[Id] [int] IDENTITY(1,1) NOT NULL,
+	[BigInt] [bigint] NULL,
+	[Binary] [binary](50) NULL,
+	[Bit] [bit] NULL,
+	[Char] [char](10) NULL,
+	[Date] [date] NULL,
+	[DateTime] [datetime] NULL,
+	[DateTime2] [datetime2](7) NULL,
+	[DateTimeOffset] [datetimeoffset](7) NULL,
+	[Decimal] [decimal](18, 9) NULL,
+	[Float] [float] NULL,
+	[Geography] [geography] NULL,
+	[Geometry] [geometry] NULL,
+	[HierarchyId] [hierarchyid] NULL,
+	[Image] [image] NULL,
+	[Int] [int] NULL,
+	[Money] [money] NULL,
+	[NChar] [nchar](10) NULL,
+	[NText] [ntext] NULL,
+	[Numeric] [numeric](18, 17) NULL,
+	[NVarChar] [nvarchar](50) NULL,
+	[NVarCharMax] [nvarchar](max) NULL,
+	[Real] [real] NULL,
+	[SmallDateTime] [smalldatetime] NULL,
+	[SmallInt] [smallint] NULL,
+	[SmallMoney] [smallmoney] NULL,
+	[SQLVariant] [sql_variant] NULL,
+	[Text] [text] NULL,
+	[Time] [time](7) NULL,
+	[TimeStamp] [timestamp] NULL,
+	[TinyInt] [tinyint] NULL,
+	[UniqueIdentifier] [uniqueidentifier] NULL,
+	[VarBinary] [varbinary](50) NULL,
+	[VarBinaryMax] [varbinary](max) NULL,
+	[VarChar] [varchar](50) NULL,
+	[VarCharMax] [varchar](max) NULL,
+	[Xml] [xml] NULL,
+ CONSTRAINT [PK_AllTypes] PRIMARY KEY CLUSTERED 
+(
+	[Id] ASC
+)WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON, OPTIMIZE_FOR_SEQUENTIAL_KEY = OFF) ON [PRIMARY]
+) ON [PRIMARY] TEXTIMAGE_ON [PRIMARY]
+GO
+/****** Object:  Table [dbo].[go.get]    Script Date: 12/29/2021 8:56:38 AM ******/
+SET ANSI_NULLS ON
+GO
+SET QUOTED_IDENTIFIER ON
+GO
+CREATE TABLE [dbo].[go.get](
+	[Name] [nchar](10) NULL
+) ON [PRIMARY]
+GO
+/****** Object:  Table [dbo].[RowVersionType]    Script Date: 12/29/2021 8:56:38 AM ******/
+SET ANSI_NULLS ON
+GO
+SET QUOTED_IDENTIFIER ON
+GO
+CREATE TABLE [dbo].[RowVersionType](
+	[Id] [int] NOT NULL,
+	[RowVersionCol] [timestamp] NOT NULL,
+PRIMARY KEY CLUSTERED 
+(
+	[Id] ASC
+)WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON, OPTIMIZE_FOR_SEQUENTIAL_KEY = OFF) ON [PRIMARY]
+) ON [PRIMARY]
+GO
+SET IDENTITY_INSERT [dbo].[AllTypes] ON 
+GO
+INSERT [dbo].[AllTypes] ([Id], [BigInt], [Binary], [Bit], [Char], [Date], [DateTime], [DateTime2], [DateTimeOffset], [Decimal], [Float], [Geography], [Geometry], [HierarchyId], [Image], [Int], [Money], [NChar], [NText], [Numeric], [NVarChar], [NVarCharMax], [Real], [SmallDateTime], [SmallInt], [SmallMoney], [SQLVariant], [Text], [Time], [TinyInt], [UniqueIdentifier], [VarBinary], [VarBinaryMax], [VarChar], [VarCharMax], [Xml]) VALUES (1, -9223372036854775808, 0x42696E6172792064617461000000000000000000000000000000000000000000000000000000000000000000000000000000, 0, N'ABC       ', CAST(N'2021-12-15' AS Date), CAST(N'2021-12-15T07:39:52.943' AS DateTime), CAST(N'2021-12-15T07:39:52.9433333' AS DateTime2), CAST(N'2021-12-15T07:39:52.9433333+01:20' AS DateTimeOffset), CAST(123456789.123456789 AS Decimal(18, 9)), 1.2, 0xE6100000010C1E1B81785DD34740E5D022DBF9965EC0, 0xE61000000114D7A3703D0A975EC08716D9CEF7D34740CBA145B6F3955EC08716D9CEF7D34740, N'/2/', 0x5465787420617320696D616765, 42, 922337203685477.5807, N'A         ', N'Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.', CAST(1.12345678912345679 AS Numeric(18, 17)), N'ABC', N'ABCD', 5.5, CAST(N'2021-12-15T07:40:00' AS SmallDateTime), 32767, 214748.3647, CAST(1.200000000000000001234 AS numeric(22,21)), N'Lorem ipsum dolor sit amet', CAST(N'07:39:52.9500000' AS Time), 255, N'5d434705-4f3e-461e-ae36-0f96834a70ba', 0x56617242696E617279283530292064617461, 0x56617242696E617279286D6178292064617461, N'ABCDE ABCDE ABCDE ABCDE ABCDE ABCDE ABCDE ABCDE AB', N'ABCDE ABCDE ABCDE ABCDE ABCDE ABCDE ABCDE ABCDE ABCDE ABCDE', N'<note><to>Tove</to><from>Jani</from><heading>Reminder</heading><body>Don''t forget me this weekend!</body></note>')
+GO
+INSERT [dbo].[AllTypes] ([Id], [BigInt], [Binary], [Bit], [Char], [Date], [DateTime], [DateTime2], [DateTimeOffset], [Decimal], [Float], [Geography], [Geometry], [HierarchyId], [Image], [Int], [Money], [NChar], [NText], [Numeric], [NVarChar], [NVarCharMax], [Real], [SmallDateTime], [SmallInt], [SmallMoney], [SQLVariant], [Text], [Time], [TinyInt], [UniqueIdentifier], [VarBinary], [VarBinaryMax], [VarChar], [VarCharMax], [Xml]) VALUES (2, 9223372036854775807, 0x42696E6172792064617461000000000000000000000000000000000000000000000000000000000000000000000000000000, 1, N'ABC       ', CAST(N'2021-12-15' AS Date), CAST(N'2021-12-15T07:39:52.960' AS DateTime), NULL, CAST(N'2021-12-08T03:00:52.9500000+00:00' AS DateTimeOffset), CAST(123456789.123456789 AS Decimal(18, 9)), 1.2, 0xE6100000010C1E1B81785DD34740E5D022DBF9965EC0, NULL, N'/2/', NULL, 42, 922337203685477.5807, N'A         ', N'Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.', CAST(1.12345678912345679 AS Numeric(18, 17)), N'ABC', N'ABCD', 5.5, CAST(N'2021-12-15T07:40:00' AS SmallDateTime), 32767, 214748.3647, CAST(1.200000000000000001234 AS numeric(22,21)), N'Lorem ipsum dolor sit amet', NULL, 255, N'0f7df523-faba-4efa-bfa9-8c39cc7fa6a6', NULL, 0x56617242696E617279286D6178292064617461, N'ABCDE ABCDE ABCDE ABCDE ABCDE ABCDE ABCDE ABCDE AB', N'ABCDE ABCDE ABCDE ABCDE ABCDE ABCDE ABCDE ABCDE ABCDE ABCDE', NULL)
+GO
+SET IDENTITY_INSERT [dbo].[AllTypes] OFF
+GO
+INSERT [dbo].[go.get] ([Name]) VALUES (N'A         ')
+GO
+INSERT [dbo].[go.get] ([Name]) VALUES (N'B         ')
+GO
+INSERT [dbo].[go.get] ([Name]) VALUES (N'C         ')
+GO
+INSERT [dbo].[go.get] ([Name]) VALUES (N'D         ')
+GO
+INSERT [dbo].[go.get] ([Name]) VALUES (N'E         ')
+GO
+INSERT [dbo].[go.get] ([Name]) VALUES (N'F         ')
+GO
+INSERT [dbo].[go.get] ([Name]) VALUES (N'G         ')
+GO
+INSERT [dbo].[RowVersionType] ([Id]) VALUES (5)
+GO
+INSERT [dbo].[RowVersionType] ([Id]) VALUES (6)
+GO
+INSERT [dbo].[RowVersionType] ([Id]) VALUES (7)
+GO
+USE [master]
+GO
+ALTER DATABASE [SqlServer_IntTest] SET  READ_WRITE 
+GO
diff --git a/testing/sqlserver/integration_test.go b/testing/sqlserver/integration_test.go
new file mode 100644
index 0000000..3f26f10
--- /dev/null
+++ b/testing/sqlserver/integration_test.go
@@ -0,0 +1,202 @@
+// Copyright 2021 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 sqlserver
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"math/big"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"cloud.google.com/go/spanner"
+	database "cloud.google.com/go/spanner/admin/database/apiv1"
+	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
+	"github.com/cloudspannerecosystem/harbourbridge/testing/common"
+
+	"google.golang.org/api/iterator"
+	databasepb "google.golang.org/genproto/googleapis/spanner/admin/database/v1"
+)
+
+var (
+	projectID  string
+	instanceID string
+
+	ctx           context.Context
+	databaseAdmin *database.DatabaseAdminClient
+)
+
+func TestMain(m *testing.M) {
+	cleanup := initIntegrationTests()
+	res := m.Run()
+	cleanup()
+	os.Exit(res)
+}
+
+func initIntegrationTests() (cleanup func()) {
+	projectID = os.Getenv("HARBOURBRIDGE_TESTS_GCLOUD_PROJECT_ID")
+	instanceID = os.Getenv("HARBOURBRIDGE_TESTS_GCLOUD_INSTANCE_ID")
+
+	ctx = context.Background()
+	flag.Parse() // Needed for testing.Short().
+	noop := func() {}
+
+	if testing.Short() {
+		log.Println("Integration tests skipped in -short mode.")
+		return noop
+	}
+
+	if projectID == "" {
+		log.Println("Integration tests skipped: HARBOURBRIDGE_TESTS_GCLOUD_PROJECT_ID is missing")
+		return noop
+	}
+
+	if instanceID == "" {
+		log.Println("Integration tests skipped: HARBOURBRIDGE_TESTS_GCLOUD_INSTANCE_ID is missing")
+		return noop
+	}
+
+	var err error
+	databaseAdmin, err = database.NewDatabaseAdminClient(ctx)
+	if err != nil {
+		log.Fatalf("cannot create databaseAdmin client: %v", err)
+	}
+
+	return func() {
+		databaseAdmin.Close()
+	}
+}
+
+func dropDatabase(t *testing.T, dbURI string) {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+	defer cancel()
+	// Drop the testing database.
+	if err := databaseAdmin.DropDatabase(ctx, &databasepb.DropDatabaseRequest{Database: dbURI}); err != nil {
+		t.Fatalf("failed to drop testing database %v: %v", dbURI, err)
+	}
+}
+
+func prepareIntegrationTest(t *testing.T) string {
+	if databaseAdmin == nil {
+		t.Skip("Integration tests skipped")
+	}
+	tmpdir, err := ioutil.TempDir(".", "int-test-")
+	if err != nil {
+		log.Fatal(err)
+	}
+	return tmpdir
+}
+
+func TestIntegration_SQLserver_SchemaSubcommand(t *testing.T) {
+	onlyRunForEmulatorTest(t)
+	t.Parallel()
+	tmpdir := prepareIntegrationTest(t)
+	defer os.RemoveAll(tmpdir)
+	filePrefix := filepath.Join(tmpdir, "SqlServer_IntTest.")
+
+	args := fmt.Sprintf("schema -prefix %s -source=sqlserver -source-profile='host=localhost,user=sa,db_name=SqlServer_IntTest'", filePrefix)
+	err := common.RunCommand(args, projectID)
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+func TestIntegration_SQLserver_SchemaAndDataSubcommand(t *testing.T) {
+	onlyRunForEmulatorTest(t)
+	tmpdir := prepareIntegrationTest(t)
+	defer os.RemoveAll(tmpdir)
+
+	dbName := "sqlserver-schema-and-data"
+	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
+	filePrefix := filepath.Join(tmpdir, "SqlServer_IntTest.")
+
+	args := fmt.Sprintf("schema-and-data -prefix %s -source=%s  -source-profile='host=localhost,user=sa,db_name=SqlServer_IntTest' -target-profile='instance=%s,dbname=%s'", filePrefix, constants.SQLSERVER, instanceID, dbName)
+	err := common.RunCommand(args, projectID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer dropDatabase(t, dbURI)
+	checkResults(t, dbURI)
+}
+func checkResults(t *testing.T, dbURI string) {
+	// Make a query to check results.
+	client, err := spanner.NewClient(ctx, dbURI)
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer client.Close()
+
+	checkCommonDataType(ctx, t, client)
+}
+
+func checkCommonDataType(ctx context.Context, t *testing.T, client *spanner.Client) {
+	var boolVal bool
+	var date spanner.NullDate
+	var floatVal float64
+	var intVal int64
+	var numericVal big.Rat
+	var stringVal string
+	var timeVal string
+	var moneyVal big.Rat
+	iter := client.Single().Read(ctx, "AllTypes", spanner.Key{1}, []string{"Bit", "Date", "Float", "BigInt", "Numeric", "Text", "Time", "Money"})
+	defer iter.Stop()
+	for {
+		row, err := iter.Next()
+		if err == iterator.Done {
+			break
+		}
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := row.Columns(&boolVal, &date, &floatVal, &intVal, &numericVal, &stringVal, &timeVal, &moneyVal); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	if got, want := boolVal, false; got != want {
+		t.Fatalf("Bit are not correct: got %v, want %v", got, want)
+	}
+	if got, want := date.String(), "2021-12-15"; got != want {
+		t.Fatalf("Date are not correct: got %v, want %v", got, want)
+	}
+	if got, want := floatVal, 1.2; got != want {
+		t.Fatalf("float are not correct: got %v, want %v", got, want)
+	}
+	if got, want := intVal, int64(-9223372036854775808); got != want {
+		t.Fatalf("Bigint are not correct: got %v, want %v", got, want)
+	}
+	if got, want := numericVal.FloatString(9), "1.123456789"; got != want {
+		t.Fatalf("Numeric are not correct: got %v, want %v", got, want)
+	}
+	if got, want := stringVal, "Lorem ipsum dolor sit amet"; got != want {
+		t.Fatalf("Text are not correct: got %v, want %v", got, want)
+	}
+	if got, want := timeVal, "07:39:52.950"; got != want {
+		t.Fatalf("Time are not correct: got %v, want %v", got, want)
+	}
+	if got, want := moneyVal.FloatString(4), "922337203685477.5807"; got != want {
+		t.Fatalf("quantities are not correct: got %v, want %v", got, want)
+	}
+}
+
+func onlyRunForEmulatorTest(t *testing.T) {
+	if os.Getenv("SPANNER_EMULATOR_HOST") == "" {
+		t.Skip("Skipping tests only running against the emulator.")
+	}
+}