Repository: cloudspannerecosystem/harbourbridge (go)

Issue:
UI: Global Data type mapping box not populated in UI when using direct connect
<img width="552" alt="Screenshot 2021-10-07 at 12 36 50 PM" src="https://user-images.githubusercontent.com/19630018/136335817-1917ea92-ec3b-436c-8ae3-23e2aeb0048f.png">
This box is not populated when mysql in direct connect has unsupported data types such as geometry.
However, using the corresponding dump file with unsupported types works as expected.

Sample schema to replicate this:
```

CREATE TABLE `productshapes` (
  `productId` int NOT NULL AUTO_INCREMENT,
  `productLine` varchar(50) NOT NULL ,
  `productShape` geometry,
  `attributes` JSON,
  PRIMARY KEY (`productId`)
) ENGINE=InnoDB DEFAULT CHARSET=latin1;
```

Pull request description:
added spatial type support in mysql direct connect(web)
Fixes #192 

added spatial type in web data type map so will not get null in the frontend, hence global data type map will populate properly. 

Patch:
diff --git a/conversion/conversion.go b/conversion/conversion.go
index 3c59b8b..fde16b1 100644
--- a/conversion/conversion.go
+++ b/conversion/conversion.go
@@ -101,7 +101,7 @@ func DataConv(ctx context.Context, sourceProfile profiles.SourceProfile, targetP
 		Verbose:    internal.Verbose(),
 	}
 	switch sourceProfile.Driver {
-	case constants.POSTGRES, constants.MYSQL, constants.DYNAMODB, constants.SQLSERVER:
+	case constants.POSTGRES, constants.MYSQL, constants.DYNAMODB, constants.SQLSERVER, constants.ORACLE:
 		return dataFromDatabase(sourceProfile, config, client, conv)
 	case constants.PGDUMP, constants.MYSQLDUMP:
 		if conv.SpSchema.CheckInterleaved() {
diff --git a/frontend/package-lock.json b/frontend/package-lock.json
index bb55a9d..cd9a85f 100644
--- a/frontend/package-lock.json
+++ b/frontend/package-lock.json
@@ -4380,9 +4380,9 @@
             "optional": true
         },
         "simple-get": {
-            "version": "3.1.0",
-            "resolved": "https://registry.npmjs.org/simple-get/-/simple-get-3.1.0.tgz",
-            "integrity": "sha512-bCR6cP+aTdScaQCnQKbPKtJOKDp/hj9EDLJo3Nw4y1QksqaovlW/bnptB6/c1e+qmNIDHRK+oXFDdEqBT8WzUA==",
+            "version": "3.1.1",
+            "resolved": "https://registry.npmjs.org/simple-get/-/simple-get-3.1.1.tgz",
+            "integrity": "sha512-CQ5LTKGfCpvE1K0n2us+kuMPbk/q0EKl82s4aheV9oXjFEz6W/Y7oQFVJuU6QG77hRT4Ghb5RURteF5vnWjupA==",
             "optional": true,
             "requires": {
                 "decompress-response": "^4.2.0",
diff --git a/go.mod b/go.mod
index d13c19e..28102ea 100644
--- a/go.mod
+++ b/go.mod
@@ -8,7 +8,8 @@ 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
+	github.com/basgys/goxml2json v1.1.0 // indirect
+	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/google/go-cmp v0.5.6
diff --git a/go.sum b/go.sum
index e7104f2..9678b4f 100644
--- a/go.sum
+++ b/go.sum
@@ -36,6 +36,8 @@ github.com/appleboy/gin-jwt/v2 v2.6.3/go.mod h1:MfPYA4ogzvOcVkRwAxT7quHOtQmVKDpT
 github.com/appleboy/gofight/v2 v2.1.2/go.mod h1:frW+U1QZEdDgixycTj4CygQ48yLTUhplt43+Wczp3rw=
 github.com/aws/aws-sdk-go v1.34.5 h1:FwubVVX9u+kW9qDCjVzyWOdsL+W5wPq683wMk2R2GXk=
 github.com/aws/aws-sdk-go v1.34.5/go.mod h1:5zCpMtNQVjRREroY7sYe8lOMRSxkhG6MZveU8YkpAk0=
+github.com/basgys/goxml2json v1.1.0 h1:4ln5i4rseYfXNd86lGEB+Vi652IsIXIvggKM/BhUKVw=
+github.com/basgys/goxml2json v1.1.0/go.mod h1:wH7a5Np/Q4QoECFIU8zTQlZwZkrilY0itPfecMw41Dw=
 github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
 github.com/beorn7/perks v1.0.0 h1:HWo1m869IqiPhD389kmkxeTalrjNbbJTC8LXupb+sl0=
 github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=
diff --git a/sources/oracle/README.md b/sources/oracle/README.md
new file mode 100644
index 0000000..dc7bc9e
--- /dev/null
+++ b/sources/oracle/README.md
@@ -0,0 +1,67 @@
+# HarbourBridge: Oracle-to-Spanner Evaluation and Migration
+
+HarbourBridge is a stand-alone open source tool for Cloud Spanner evaluation,
+using data from an existing Oracle database. This README provides
+details of the tool's Oracle capabilities. For general HarbourBridge information
+see this [README](https://github.com/cloudspannerecosystem/harbourbridge#harbourbridge-spanner-evaluation-and-migration).
+
+Note that _'oracle'_ can be used as an identifier with the flag `-source` in the command line.
+
+## Example Oracle Usage
+
+HarbourBridge can be run directly on a Oracle 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 an Oracle database
+
+In this case, HarbourBridge connects directly to the Oracle database to
+retrieve table schema and data. Set the `-source=oracle` and corresponding
+source profile connection parameters `host`, `port`, `user`, `db_name` and
+`password`.
+
+For example to perform schema conversion, run
+
+```sh
+harbourbridge schema -source=oracle -source-profile="host=<>,port=<>,user=<>,db_name=<>,password=<>"
+```
+
+In Oracle, USER is the account name, SCHEMA is the set of objects owned by that user. Oracle creates the SCHEMA object as part of the CREATE USER statement and the SCHEMA has the same name as the USER. 
+
+db_name will be name of the Oracle service being used.
+
+
+## Schema Conversion
+
+| SQL_Server_Type        | Spanner_Type |
+| ---------------------- | ------------ |
+| NUMBER (* , 0)         | INT64        |
+| FLOAT                  | FLOAT64      |
+| BINARY_FLOAT           | FLOAT64      |
+| BINARY_DOUBLE          | FLOAT64      |
+| NUMBER (* , >0)        | NUMERIC      |
+| CHAR                   | STRING       |
+| NCHAR                  | STRING       |
+| VARCHAR                | STRING       |
+| VARCHAR2               | STRING       |
+| NVARCHAR2              | STRING       |
+| CLOB                   | STRING       |
+| NCLOB                  | STRING       |
+| LONG                   | STRING       |
+| ROWID                  | STRING       |
+| UROWID                 | STRING       |
+| DATE                   | DATE         |
+| TIMESTAMP              | TIMESTAMP    |
+| BLOB                   | BYTES        |
+| BFILE                  | BYTES        |
+| RAW                    | BYTES        |
+| LONG RAW               | BYTES        |
+| XMLTYPE                | STRING       |
+| INTERVAL YEAR          | STRING       |
+| INTERVAL DAY           | STRING       |
+| GEOMETRY               | STRING       |
+| JSON                   | JSON         |
+
+
+
diff --git a/sources/oracle/data.go b/sources/oracle/data.go
new file mode 100644
index 0000000..ee251ba
--- /dev/null
+++ b/sources/oracle/data.go
@@ -0,0 +1,315 @@
+// Copyright 2022 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 oracle
+
+import (
+	"encoding/json"
+	"fmt"
+	"math/big"
+	"math/bits"
+	"strconv"
+	"strings"
+	"time"
+
+	"cloud.google.com/go/civil"
+	"cloud.google.com/go/spanner"
+	xj "github.com/basgys/goxml2json"
+	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/schema"
+	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
+)
+
+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., these values
+		// 'NULL' values are represented as "NULL" (because we retrieve the values as strings).
+		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
+		if spColDef.T.IsArray {
+			x, err = convArray(spColDef.T, srcColDef.Type.Name, vals[i])
+		} else {
+			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(conv, 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)
+	case ddl.JSON:
+		if srcTypeName == "OBJECT" {
+			return convertXmlToJson(val)
+		}
+		return val, nil
+	default:
+		return val, fmt.Errorf("data conversion not implemented for type %v", spannerType.Name)
+	}
+}
+
+func convBool(conv *internal.Conv, val string) (bool, error) {
+	b, err := strconv.ParseBool(val)
+	if err != nil {
+		i, err2 := convInt64(val)
+		if err2 == nil && i >= -128 && i <= 127 {
+			b = i != 0
+			conv.Unexpected(fmt.Sprintf("Expected boolean value, but found integer value %v; mapping it to %v\n", val, b))
+			return b, err2
+		}
+		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.Split(val, "T")[0]
+	d, err := civil.ParseDate(date)
+	if err != nil {
+		return d, fmt.Errorf("can't convert to date: %w", err)
+	}
+	return d, err
+}
+
+func convFloat64(val string) (float64, error) {
+	f, err := strconv.ParseFloat(val, 64)
+	if err != nil {
+		return f, fmt.Errorf("can't convert to float64: %w", err)
+	}
+	return f, 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 timestamp into a go Time Spanner timestamp
+func convTimestamp(srcTypeName string, val string) (t time.Time, err error) {
+	// we are getting all timestamp value in UTC from the oracle.
+	// e.g. 2022-02-01T08:14:36.254Z 			(timestamp)
+	// e.g. 2022-02-01T12:14:36.254Z 		    (timestamp with timezone)
+	// e.g. 2022-02-01T06:14:36.254Z 			(timestamp with local timezone)
+
+	t, err = time.Parse(time.RFC3339, val)
+	if err != nil {
+		return t, fmt.Errorf("can't convert to timestamp (type: %s)", srcTypeName)
+	}
+	return t, err
+}
+
+func convArray(spannerType ddl.Type, srcTypeName string, v string) (interface{}, error) {
+	v = strings.TrimSpace(v)
+	// Handle empty array. Note that we use an empty NullString array
+	// for all Spanner array types since this will be converted to the
+	// appropriate type by the Spanner client.
+	if v == "" {
+		return []spanner.NullString{}, nil
+	}
+
+	// The Spanner client for go does not accept []interface{} for arrays.
+	// Instead it only accepts slices of a specific type eg: []string
+	// Hence we have to do the following case analysis.
+	switch spannerType.Name {
+	case ddl.String:
+		var a []string
+		var r []spanner.NullString
+		err := json.Unmarshal([]byte(v), &a)
+		if err != nil {
+			return []spanner.NullString{}, err
+		}
+
+		for _, s := range a {
+			if s == "NULL" {
+				r = append(r, spanner.NullString{Valid: false})
+				continue
+			}
+			r = append(r, spanner.NullString{StringVal: s, Valid: true})
+		}
+		return r, nil
+	case ddl.Numeric:
+		var a []interface{}
+		var r []spanner.NullNumeric
+		err := json.Unmarshal([]byte(v), &a)
+		if err != nil {
+			return []spanner.NullNumeric{}, err
+		}
+		for _, s := range a {
+			if s == "NULL" {
+				r = append(r, spanner.NullNumeric{Valid: false})
+				continue
+			}
+			val := new(big.Rat)
+			if _, ok := val.SetString(fmt.Sprint(s)); !ok {
+				return []spanner.NullNumeric{}, fmt.Errorf("can't convert %q to big.Rat", s)
+			}
+			r = append(r, spanner.NullNumeric{Numeric: *val, Valid: true})
+		}
+		return r, nil
+	case ddl.Int64:
+		var a []interface{}
+		var r []spanner.NullInt64
+		err := json.Unmarshal([]byte(v), &a)
+		if err != nil {
+			return []spanner.NullInt64{}, err
+		}
+		for _, s := range a {
+			if s == "NULL" {
+				r = append(r, spanner.NullInt64{Valid: false})
+				continue
+			}
+			val, err := convInt64(fmt.Sprint(s))
+			if err != nil {
+				return []spanner.NullInt64{}, err
+			}
+			r = append(r, spanner.NullInt64{Int64: val, Valid: true})
+		}
+		return r, nil
+	case ddl.Float64:
+		var a []interface{}
+		var r []spanner.NullFloat64
+		err := json.Unmarshal([]byte(v), &a)
+		if err != nil {
+			return []spanner.NullFloat64{}, err
+		}
+		for _, s := range a {
+			if s == "NULL" {
+				r = append(r, spanner.NullFloat64{Valid: false})
+				continue
+			}
+			val, err := convFloat64(fmt.Sprint(s))
+			if err != nil {
+				return []spanner.NullFloat64{}, err
+			}
+			r = append(r, spanner.NullFloat64{Float64: val, Valid: true})
+		}
+		return r, nil
+	case ddl.Date:
+		var a []interface{}
+		err := json.Unmarshal([]byte(v), &a)
+		if err != nil {
+			return []spanner.NullDate{}, err
+		}
+		var r []spanner.NullDate
+		for _, s := range a {
+			if s == "NULL" {
+				r = append(r, spanner.NullDate{Valid: false})
+				continue
+			}
+			val, err := convDate(fmt.Sprint(s))
+			if err != nil {
+				return []spanner.NullDate{}, err
+			}
+			r = append(r, spanner.NullDate{Date: val, Valid: true})
+		}
+		return r, nil
+	}
+	return []interface{}{}, fmt.Errorf("array type conversion not implemented for type %v", spannerType.Name)
+}
+
+func convertXmlToJson(v string) (ans string, err error) {
+	xml := strings.NewReader(v)
+	j, err := xj.Convert(xml)
+	if err != nil {
+		return "", fmt.Errorf("not able to convert object to JSON: %v ", v)
+	}
+	return j.String(), nil
+}
diff --git a/sources/oracle/infoschema.go b/sources/oracle/infoschema.go
index 07d5722..d07cf71 100644
--- a/sources/oracle/infoschema.go
+++ b/sources/oracle/infoschema.go
@@ -1,3 +1,17 @@
+// Copyright 2022 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 oracle
 
 import (
@@ -29,21 +43,92 @@ func (isi InfoSchemaImpl) GetTableName(dbName string, tableName string) string {
 
 // GetRowsFromTable returns a sql Rows object for a table.
 func (isi InfoSchemaImpl) GetRowsFromTable(conv *internal.Conv, srcTable string) (interface{}, error) {
-	panic("Not Implemented")
+	tbl := conv.SrcSchema[srcTable]
+	srcCols := tbl.ColNames
+	if len(srcCols) == 0 {
+		conv.Unexpected(fmt.Sprintf("Couldn't get source columns for table %s ", srcTable))
+		return nil, nil
+	}
+	q := getSelectQuery(isi.DbName, tbl.Schema, tbl.Name, tbl.ColNames, tbl.ColDefs)
+	rows, err := isi.Db.Query(q)
+	return rows, err
 }
 
-// ProcessData performs data conversion for source database.
-func (isi InfoSchemaImpl) ProcessData(conv *internal.Conv, srcTable string, srcSchema schema.Table, spTable string, spCols []string, spSchema ddl.CreateTable) error {
-	panic("Not Implemented")
+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
+		if timestampReg.MatchString(colDefs[cn].Type.Name) {
+			s = fmt.Sprintf(`SYS_EXTRACT_UTC("%s") AS "%s"`, cn, cn)
+		} else if len(colDefs[cn].Type.ArrayBounds) == 1 {
+			s = fmt.Sprintf(`(SELECT JSON_ARRAYAGG(COLUMN_VALUE RETURNING VARCHAR2(4000)) 
+				FROM TABLE ("%s"."%s")) AS "%s"`, tableName, cn, cn)
+		} else {
+			switch colDefs[cn].Type.Name {
+			case "XMLTYPE":
+				s = fmt.Sprintf(`CAST(XMLTYPE.getStringVal("%s") AS VARCHAR2(4000)) AS "%s"`, cn, cn)
+			case "SDO_GEOMETRY":
+				s = fmt.Sprintf(`SDO_UTIL.TO_WKTGEOMETRY("%s") AS "%s"`, cn, cn)
+			case "OBJECT":
+				s = fmt.Sprintf(`
+				(
+					CASE WHEN "%s" IS NULL THEN ''
+					ELSE
+						XMLTYPE("%s").getStringVal() 
+					END
+				) AS "%s"
+				`, cn, cn, cn)
+			default:
+				s = fmt.Sprintf(`"%s"`, cn)
+			}
+		}
+		selects[i] = s
+	}
+
+	return fmt.Sprintf(`SELECT %s FROM "%s"."%s"`, strings.Join(selects, ", "), schemaName, tableName)
 }
 
-func ProcessDataRow(conv *internal.Conv, srcTable string, srcCols []string, srcSchema schema.Table, spTable string, spCols []string, spSchema ddl.CreateTable, values []string) {
-	panic("Not Implemented")
+// ProcessData performs data conversion for source database.
+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
 }
 
 // GetRowCount with number of rows in each table.
 func (isi InfoSchemaImpl) GetRowCount(table common.SchemaAndName) (int64, error) {
-	panic("Not Implemented")
+	q := fmt.Sprintf(`SELECT count(*) FROM "%s"`, 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
 }
 
 func (isi InfoSchemaImpl) GetTables() ([]common.SchemaAndName, error) {
@@ -65,16 +150,23 @@ func (isi InfoSchemaImpl) GetTables() ([]common.SchemaAndName, error) {
 // 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 := fmt.Sprintf(`
-					SELECT 
+						SELECT 
 						column_name, 
 						data_type, 
 						nullable, 
 						data_default, 
 						data_length, 
 						data_precision, 
-						data_scale 
-					FROM all_tab_columns 
-					WHERE owner = '%s' AND table_name = '%s'
+						data_scale,
+						at.typecode,
+						act.elem_type_name,
+						act.length,
+						act.precision,
+						act.scale
+					FROM all_tab_columns atc
+					LEFT JOIN all_types at ON atc.data_type=at.type_name AND atc.owner = at.owner
+					LEFT JOIN all_coll_types act ON atc.data_type=act.type_name AND atc.owner = at.owner
+					WHERE atc.owner = '%s' AND atc.table_name = '%s'
 					`, table.Schema, table.Name)
 	cols, err := isi.Db.Query(q)
 	if err != nil {
@@ -84,10 +176,10 @@ func (isi InfoSchemaImpl) GetColumns(conv *internal.Conv, table common.SchemaAnd
 	var colNames []string
 	var colName, dataType string
 	var isNullable string
-	var colDefault sql.NullString
-	var charMaxLen, numericPrecision, numericScale sql.NullInt64
+	var colDefault, typecode, elementDataType sql.NullString
+	var charMaxLen, numericPrecision, numericScale, elementCharMaxLen, elementNumericPrecision, elementNumericScale sql.NullInt64
 	for cols.Next() {
-		err := cols.Scan(&colName, &dataType, &isNullable, &colDefault, &charMaxLen, &numericPrecision, &numericScale)
+		err := cols.Scan(&colName, &dataType, &isNullable, &colDefault, &charMaxLen, &numericPrecision, &numericScale, &typecode, &elementDataType, &elementCharMaxLen, &elementNumericPrecision, &elementNumericScale)
 		if err != nil {
 			conv.Unexpected(fmt.Sprintf("Can't scan: %v", err))
 			continue
@@ -101,14 +193,23 @@ func (isi InfoSchemaImpl) GetColumns(conv *internal.Conv, table common.SchemaAnd
 			switch c {
 			case "C":
 				ignored.Check = true
-			case "R", "P", "U":
-				// Nothing to do here -- these are handled elsewhere.
+			// Oracle 21c introduces a JSON datatype, before that we used to store json as VARCHAR2, CLOB, and BLOB.
+			// If column has check constraints IS JSON(check for J in constraints array as per GetConstraints function) then update src datatype to JSON
+			// so toSpannerTypeInternal function map this datatype to spanner JSON.
+			case "J":
+				dataType = "JSON"
+				charMaxLen.Valid = false
 			}
 		}
+		if typecode.Valid && typecode.String == "OBJECT" {
+			dataType = "OBJECT"
+			charMaxLen.Valid = false
+		}
+
 		ignored.Default = colDefault.Valid
 		c := schema.Column{
 			Name:    colName,
-			Type:    toType(dataType, charMaxLen, numericPrecision, numericScale),
+			Type:    toType(dataType, typecode, elementDataType, charMaxLen, numericPrecision, numericScale, elementCharMaxLen, elementNumericPrecision, elementNumericScale),
 			NotNull: strings.ToUpper(isNullable) == "N",
 			Ignored: ignored,
 		}
@@ -126,7 +227,8 @@ func (isi InfoSchemaImpl) GetConstraints(conv *internal.Conv, table common.Schem
 	q := fmt.Sprintf(`
 					SELECT 
 						k.column_name,
-						t.constraint_type
+						t.constraint_type,
+						t.search_condition
 	   				FROM all_constraints t
        				INNER JOIN all_cons_columns k
        				ON (k.constraint_name = t.constraint_name) 
@@ -139,9 +241,10 @@ func (isi InfoSchemaImpl) GetConstraints(conv *internal.Conv, table common.Schem
 	defer rows.Close()
 	var primaryKeys []string
 	var col, constraint string
+	var condition sql.NullString
 	m := make(map[string][]string)
 	for rows.Next() {
-		err := rows.Scan(&col, &constraint)
+		err := rows.Scan(&col, &constraint, &condition)
 		if err != nil {
 			conv.Unexpected(fmt.Sprintf("Can't scan: %v", err))
 			continue
@@ -154,6 +257,13 @@ func (isi InfoSchemaImpl) GetConstraints(conv *internal.Conv, table common.Schem
 		switch constraint {
 		case "P":
 			primaryKeys = append(primaryKeys, col)
+		case "C":
+			// If column has IS JSON check constraints then add extra string "J" in constraints array.
+			// condition value example `column_name IS JSON`,null etc.
+			if condition.Valid && strings.Contains(condition.String, "IS JSON") {
+				m[col] = append(m[col], "J")
+			}
+			m[col] = append(m[col], constraint)
 		default:
 			m[col] = append(m[col], constraint)
 		}
@@ -281,16 +391,60 @@ func (isi InfoSchemaImpl) GetIndexes(conv *internal.Conv, table common.SchemaAnd
 	return indexes, nil
 }
 
-func toType(dataType string, charLen sql.NullInt64, numericPrecision, numericScale sql.NullInt64) schema.Type {
+func toType(dataType string, typecode, elementDataType sql.NullString, charLen sql.NullInt64, numericPrecision, numericScale, elementCharMaxLen, elementNumericPrecision, elementNumericScale sql.NullInt64) schema.Type {
+	switch {
+	case typecode.Valid && typecode.String == "COLLECTION":
+		return modifyType(elementDataType.String, elementCharMaxLen, elementNumericPrecision, elementNumericScale, true)
+	default:
+		return modifyType(dataType, charLen, numericPrecision, numericScale, false)
+	}
+}
+
+func modifyType(dataType string, charLen sql.NullInt64, numericPrecision, numericScale sql.NullInt64, isArray bool) schema.Type {
+	var t schema.Type
 	switch {
 	case dataType == "NUMBER" && numericPrecision.Valid && numericScale.Valid && numericScale.Int64 != 0:
-		return schema.Type{Name: dataType, Mods: []int64{numericPrecision.Int64, numericScale.Int64}}
+		t = schema.Type{Name: dataType, Mods: []int64{numericPrecision.Int64, numericScale.Int64}}
 	case dataType == "NUMBER" && numericPrecision.Valid:
-		return schema.Type{Name: dataType, Mods: []int64{numericPrecision.Int64}}
+		t = schema.Type{Name: dataType, Mods: []int64{numericPrecision.Int64}}
 	// Oracle get column query return data length for the Number type.
 	case dataType != "NUMBER" && charLen.Valid:
-		return schema.Type{Name: dataType, Mods: []int64{charLen.Int64}}
+		t = schema.Type{Name: dataType, Mods: []int64{charLen.Int64}}
 	default:
-		return schema.Type{Name: dataType}
+		t = schema.Type{Name: dataType}
+	}
+	if isArray {
+		t.ArrayBounds = []int64{-1}
+		return t
+	}
+	return t
+}
+
+// buildVals constructs []sql.RawBytes 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 []sql.RawBytes, iv []interface{}) {
+	v = make([]sql.RawBytes, n)
+	// rows.Scan wants '[]interface{}' as an argument, so we must copy the
+	// references into such a slice.
+	iv = make([]interface{}, len(v))
+	for i := range v {
+		iv[i] = &v[i]
+	}
+	return v, iv
+}
+
+func valsToStrings(vals []sql.RawBytes) []string {
+	toString := func(val sql.RawBytes) string {
+		if val == nil {
+			return "NULL"
+		}
+		return string(val)
+	}
+	var s []string
+	for _, v := range vals {
+		s = append(s, toString(v))
 	}
+	return s
 }
diff --git a/sources/oracle/toddl.go b/sources/oracle/toddl.go
index e97ae80..9cb69f2 100644
--- a/sources/oracle/toddl.go
+++ b/sources/oracle/toddl.go
@@ -1,4 +1,4 @@
-// Copyright 2021 Google LLC
+// Copyright 2022 Google LLC
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -41,6 +41,12 @@ func (tdi ToDdlImpl) ToSpannerType(conv *internal.Conv, columnType schema.Type)
 	ty, issues := toSpannerTypeInternal(conv, columnType.Name, columnType.Mods)
 	if conv.TargetDb == constants.TargetExperimentalPostgres {
 		ty = overrideExperimentalType(columnType, ty)
+	} else {
+		if len(columnType.ArrayBounds) > 1 {
+			ty = ddl.Type{Name: ddl.String, Len: ddl.MaxLength}
+			issues = append(issues, internal.MultiDimensionalArray)
+		}
+		ty.IsArray = len(columnType.ArrayBounds) == 1
 	}
 	return ty, issues
 }
@@ -99,6 +105,8 @@ func toSpannerTypeInternal(conv *internal.Conv, id string, mods []int64) (ddl.Ty
 		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, nil
 	case "XMLTYPE":
 		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, nil
+	case "JSON", "OBJECT":
+		return ddl.Type{Name: ddl.JSON}, nil
 	default:
 		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, []internal.SchemaIssue{internal.NoGoodType}
 	}
@@ -108,6 +116,8 @@ func toSpannerTypeInternal(conv *internal.Conv, id string, mods []int64) (ddl.Ty
 func overrideExperimentalType(columnType schema.Type, originalType ddl.Type) ddl.Type {
 	if columnType.Name == "DATE" {
 		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}
+	} else if columnType.Name == "JSON" {
+		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}
 	}
 	return originalType
 }
diff --git a/web/web.go b/web/web.go
index 485111d..5a3a19d 100644
--- a/web/web.go
+++ b/web/web.go
@@ -1157,7 +1157,7 @@ func addTypeToList(convertedType string, spType string, issues []internal.Schema
 }
 func init() {
 	// Initialize mysqlTypeMap.
-	for _, srcType := range []string{"bool", "boolean", "varchar", "char", "text", "tinytext", "mediumtext", "longtext", "set", "enum", "json", "bit", "binary", "varbinary", "blob", "tinyblob", "mediumblob", "longblob", "tinyint", "smallint", "mediumint", "int", "integer", "bigint", "double", "float", "numeric", "decimal", "date", "datetime", "timestamp", "time", "year"} {
+	for _, srcType := range []string{"bool", "boolean", "varchar", "char", "text", "tinytext", "mediumtext", "longtext", "set", "enum", "json", "bit", "binary", "varbinary", "blob", "tinyblob", "mediumblob", "longblob", "tinyint", "smallint", "mediumint", "int", "integer", "bigint", "double", "float", "numeric", "decimal", "date", "datetime", "timestamp", "time", "year", "geometrycollection", "multipoint", "multilinestring", "multipolygon", "point", "linestring", "polygon", "geometry"} {
 		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 := toSpannerTypeMySQL(srcType, spType, []int64{})


Test patch:
diff --git a/.github/workflows/integration-tests-against-emulator.yaml b/.github/workflows/integration-tests-against-emulator.yaml
index b139b25..2617250 100644
--- a/.github/workflows/integration-tests-against-emulator.yaml
+++ b/.github/workflows/integration-tests-against-emulator.yaml
@@ -74,6 +74,10 @@ jobs:
         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
+      oracle:
+        image: oracleinanutshell/oracle-xe-11g
+        ports: 
+          - 1521:1521
 
     steps:
       - uses: actions/checkout@v2
@@ -104,6 +108,18 @@ jobs:
       - run: sqlcmd -?
       - run: sqlcmd -U sa -P ${MSSQL_SA_PASSWORD} -i test_data/sqlserver.test.out
 
+      # sqlplus set up init oracle db.
+      - name: Install sqlplus required for loading .sql files
+        run: |
+          sudo apt install alien libaio1
+          curl -O https://download.oracle.com/otn_software/linux/instantclient/215000/oracle-instantclient-basic-21.5.0.0.0-1.x86_64.rpm          
+          curl -O https://download.oracle.com/otn_software/linux/instantclient/215000/oracle-instantclient-sqlplus-21.5.0.0.0-1.x86_64.rpm
+          sudo alien -i oracle-instantclient-basic-21.5.0.0.0-1.x86_64.rpm
+          sudo alien -i oracle-instantclient-sqlplus-21.5.0.0.0-1.x86_64.rpm
+          sudo sh -c 'echo /usr/lib/oracle/21.5/client64/lib/ > /etc/ld.so.conf.d/oracle.conf'
+          sudo ldconfig
+      - run: sqlplus SYS/oracle@127.0.0.1:1521/xe as sysdba @test_data/oracle.test.out
+     
       # create a spanner instance
       - uses: google-github-actions/setup-gcloud@master
         with:
diff --git a/sources/oracle/data_test.go b/sources/oracle/data_test.go
new file mode 100644
index 0000000..6739013
--- /dev/null
+++ b/sources/oracle/data_test.go
@@ -0,0 +1,314 @@
+// Copyright 2022 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 oracle
+
+import (
+	"fmt"
+	"math/big"
+	"math/bits"
+	"testing"
+	"time"
+
+	"cloud.google.com/go/civil"
+	"cloud.google.com/go/spanner"
+	"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{}
+}
+
+func TestProcessDataRow(t *testing.T) {
+	tableName := "testtable"
+	cols := []string{"a", "b", "c", "d"}
+	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: "b", T: ddl.Type{Name: ddl.Timestamp}},
+				"d": {Name: "d", T: ddl.Type{Name: ddl.String, Len: int64(1)}},
+			}},
+		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: "NUMER"}},
+				"c": {Name: "c", Type: schema.Type{Name: "TIMESTAMP(6)"}},
+				"d": {Name: "d", Type: schema.Type{Name: "CHAR", Mods: []int64{1}}},
+			}})
+	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", "2022-01-19T09:34:06.47Z", "p"})
+	assert.Equal(t, []spannerData{{table: tableName, cols: cols, vals: []interface{}{float64(4.2), int64(6), getTime("2022-01-19T09:34:06.47Z"), "p"}}}, rows)
+}
+
+func TestConvertData(t *testing.T) {
+	numStr := "33753785954695469456.33333333982343435"
+	numVal := new(big.Rat)
+	numVal.SetString(numStr)
+	outputJson := "{\"PERSON_TYP\": {\"IDNO\": \"1\", \"NAME\": \"test\", \"PHONE\": \"123456\"}}\n"
+
+	singleColTests := []struct {
+		name  string
+		ty    ddl.Type
+		srcTy string      // Source DB type
+		in    string      // Input value for conversion.
+		e     interface{} // Expected result.
+	}{
+		{"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")},
+		{"float", ddl.Type{Name: ddl.Float64}, "", "42.6", float64(42.6)},
+		{"int", ddl.Type{Name: ddl.Int64}, "", "42", int64(42)},
+		{"string", ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, "VARCHAR2", "eh", "eh"},
+		{"number", ddl.Type{Name: ddl.Numeric}, "NUMBER", numStr, numVal},
+		{"timestamp", ddl.Type{Name: ddl.Timestamp}, "TIMESTAMP(6)", "2022-01-19T09:34:06.47Z", getTime("2022-01-19T09:34:06.47Z")},
+		{"json", ddl.Type{Name: ddl.JSON}, "VARCHAR2", "{\"abc\": 123}", "{\"abc\": 123}"},
+		{"bool", ddl.Type{Name: ddl.Bool}, "CHAR(1)", "T", true},
+		{"arrayStr", ddl.Type{Name: ddl.String, IsArray: true}, "", "[\"CA\",\"CDSC\",\"DSCCS\"]", []spanner.NullString{{StringVal: "CA", Valid: true}, {StringVal: "CDSC", Valid: true}, {StringVal: "DSCCS", Valid: true}}},
+		{"arrayInt", ddl.Type{Name: ddl.Int64, IsArray: true}, "", "[1,2,3]", []spanner.NullInt64{{Int64: 1, Valid: true}, {Int64: 2, Valid: true}, {Int64: 3, Valid: true}}},
+		{"arrayFloat", ddl.Type{Name: ddl.Float64, IsArray: true}, "", "[1.5,0.00002,357657]", []spanner.NullFloat64{{Float64: 1.5, Valid: true}, {Float64: 0.00002, Valid: true}, {Float64: 357657, Valid: true}}},
+		{"arrayFloat", ddl.Type{Name: ddl.Date, IsArray: true}, "", "[\"2022-04-12\", \"2022-11-12\", \"2022-09-12\"]", getDateArray()},
+		{"object", ddl.Type{Name: ddl.JSON}, "OBJECT", "<PERSON_TYP><IDNO>1</IDNO><NAME>test</NAME><PHONE>123456</PHONE></PERSON_TYP>", outputJson},
+	}
+	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})
+			if tc.srcTy == "OBJECT" {
+				assert.Nil(t, err, tc.name)
+				assert.Equal(t, at, tableName, tc.name+": table mismatch")
+				assert.Equal(t, ac, []string{col}, tc.name+": column mismatch")
+				assert.Equal(t, len(fmt.Sprint(av[0])), len(outputJson), tc.name+": value mismatch")
+			} else {
+				checkResults(t, at, ac, av, err, tableName, []string{col}, []interface{}{tc.e}, 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", "t"},
+			ecols: []string{"a", "b", "c", "synth_id"},
+			evals: []interface{}{int64(6), float64(6.6), "t", 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.String, Len: int64(1)}},
+		}}
+	srcTable := schema.Table{
+		Name:     tableName,
+		ColNames: []string{"a", "b", "c"},
+		ColDefs: map[string]schema.Column{
+			"a": {Type: schema.Type{Name: "NUMBER"}},
+			"b": {Type: schema.Type{Name: "FLOAT"}},
+			"c": {Type: schema.Type{Name: "CHAR", Mods: []int64{1}}},
+		}}
+	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 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), "1"},
+		},
+		{
+			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), "1", int64(6)},
+		},
+		{
+			name:  "Null column",
+			cols:  []string{"a", "b", "c"},
+			vals:  []string{"6", "NULL", "1"},
+			ecols: []string{"a", "c"},
+			evals: []interface{}{int64(6), "1"},
+		},
+		{
+			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.String, Len: int64(1)}},
+		}}
+	srcTable := schema.Table{
+		Name:     tableName,
+		ColNames: []string{"a", "b", "c"},
+		ColDefs: map[string]schema.Column{
+			"a": {Type: schema.Type{Name: "NUMBER", Mods: []int64{5}}},
+			"b": {Type: schema.Type{Name: "FLOAT"}},
+			"c": {Type: schema.Type{Name: "CHAR", Mods: []int64{1}}},
+		}}
+	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 timestamp",
+			cols: []string{"a", "b", "c"},
+			vals: []string{"6", "6.6", "2022-01-199:34:06.47Z"},
+		},
+	}
+	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.Timestamp}},
+		}}
+	srcTable := schema.Table{
+		Name:     tableName,
+		ColNames: []string{"a", "b", "c"},
+		ColDefs: map[string]schema.Column{
+			"a": {Type: schema.Type{Name: "NUMBER"}},
+			"b": {Type: schema.Type{Name: "FLOAT"}},
+			"c": {Type: schema.Type{Name: "TIMESTAMP(6)"}},
+		}}
+	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 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 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 getTime(val string) time.Time {
+	t, _ := time.Parse(time.RFC3339, val)
+	return t
+}
+
+func getDate(s string) civil.Date {
+	d, _ := civil.ParseDate(s)
+	return d
+}
+
+func getDateArray() []spanner.NullDate {
+	ds := []string{"2022-04-12", "2022-11-12", "2022-09-12"}
+	var dates []spanner.NullDate
+	for _, d := range ds {
+		date, _ := civil.ParseDate(d)
+		dates = append(dates, spanner.NullDate{Date: date, Valid: true})
+	}
+	return dates
+}
diff --git a/sources/oracle/infoschema_test.go b/sources/oracle/infoschema_test.go
index de93d18..0097e09 100644
--- a/sources/oracle/infoschema_test.go
+++ b/sources/oracle/infoschema_test.go
@@ -1,4 +1,4 @@
-// Copyright 2021 Google LLC
+// Copyright 2022 Google LLC
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -33,7 +33,7 @@ type mockSpec struct {
 	rows  [][]driver.Value // Set of rows returned.
 }
 
-func TestProcessSchemaMYSQL(t *testing.T) {
+func TestProcessSchemaOracle(t *testing.T) {
 	ms := []mockSpec{
 		{
 			query: "SELECT table_name FROM all_tables (.+)",
@@ -42,16 +42,16 @@ func TestProcessSchemaMYSQL(t *testing.T) {
 			rows: [][]driver.Value{
 				{"USER"},
 				{"TEST"},
-			},
+				{"TEST2"}},
 		},
 		// USER table
 		{
 			query: `SELECT (.+) FROM all_constraints (.+)`,
 			args:  []driver.Value{},
-			cols:  []string{"column_name", "contraint_type"},
+			cols:  []string{"column_name", "contraint_type", "condition"},
 			rows: [][]driver.Value{
-				{"USER_ID", "P"},
-				{"REF", "F"},
+				{"USER_ID", "P", "USER ID IS NOT NULL"},
+				{"REF", "F", ""},
 			},
 		},
 		{
@@ -78,20 +78,20 @@ func TestProcessSchemaMYSQL(t *testing.T) {
 		{
 			query: "SELECT (.+) FROM all_tab_columns (.+)",
 			args:  []driver.Value{},
-			cols:  []string{"column_name", "data_type", "nullable", "data_default", "data_length", "data_precision", "data_scale"},
+			cols:  []string{"column_name", "data_type", "nullable", "data_default", "data_length", "data_precision", "data_scale", "typecode", "element_type", "element_length", "element_precision", "element_scale"},
 			rows: [][]driver.Value{
-				{"USER_ID", "VARCHAR2", "N", nil, nil, nil, nil},
-				{"NAME", "VARCHAR2", "N", nil, nil, nil, nil},
-				{"REF", "NUMBER", "Y", nil, nil, nil, nil}},
+				{"USER_ID", "VARCHAR2", "N", nil, nil, nil, nil, nil, nil, nil, nil, nil},
+				{"NAME", "VARCHAR2", "N", nil, nil, nil, nil, nil, nil, nil, nil, nil},
+				{"REF", "NUMBER", "Y", nil, nil, nil, nil, nil, nil, nil, nil, nil}},
 		},
 
 		// test table
 		{
 			query: `SELECT (.+) FROM all_constraints (.+)`,
 			args:  []driver.Value{},
-			cols:  []string{"column_name", "contraint_type"},
+			cols:  []string{"column_name", "contraint_type", "condition"},
 			rows: [][]driver.Value{
-				{"ID", "P"}},
+				{"ID", "P", "ID IS NOT NULL"}},
 		},
 		{
 			query: `SELECT (.+) all_cons_columns A JOIN all_constraints C ON (.+) JOIN all_cons_columns B (.+)`,
@@ -108,10 +108,46 @@ func TestProcessSchemaMYSQL(t *testing.T) {
 		{
 			query: "SELECT (.+) FROM all_tab_columns (.+)",
 			args:  []driver.Value{},
-			cols:  []string{"column_name", "data_type", "nullable", "data_default", "data_length", "data_precision", "data_scale"},
+			cols:  []string{"column_name", "data_type", "nullable", "data_default", "data_length", "data_precision", "data_scale", "typecode", "element_type", "element_length", "element_precision", "element_scale"},
+			rows: [][]driver.Value{
+				{"ID", "NUMBER", "N", nil, nil, nil, nil, nil, nil, nil, nil, nil}},
+		},
+
+		// test2 table [json column test]
+		{
+			query: `SELECT (.+) FROM all_constraints (.+)`,
+			args:  []driver.Value{},
+			cols:  []string{"column_name", "contraint_type", "condition"},
 			rows: [][]driver.Value{
-				{"ID", "NUMBER", "N", nil, nil, nil, nil}},
+				{"ID", "P", "ID IS NOT NULL"},
+				{"JSON", "C", "JSON IS JSON"}},
 		},
+		{
+			query: `SELECT (.+) all_cons_columns A JOIN all_constraints C ON (.+) JOIN all_cons_columns B (.+)`,
+			args:  []driver.Value{},
+			cols:  []string{"ref_table", "column_name", "ref_column_name", "name"},
+			rows:  [][]driver.Value{},
+		},
+		{
+			query: `SELECT (.+) LEFT JOIN all_ind_expressions IE (.+) LEFT JOIN all_indexes I (.+)`,
+			args:  []driver.Value{},
+			cols:  []string{"name", "column_name", "column_position", "descend", "uniqueness", "column_expression", "index_type"},
+			rows:  [][]driver.Value{},
+		},
+		{
+			query: "SELECT (.+) FROM all_tab_columns (.+)",
+			args:  []driver.Value{},
+			cols:  []string{"column_name", "data_type", "nullable", "data_default", "data_length", "data_precision", "data_scale", "typecode", "element_type", "element_length", "element_precision", "element_scale"},
+			rows: [][]driver.Value{
+				{"ID", "NUMBER", "N", nil, nil, nil, nil, nil, nil, nil, nil, nil},
+				{"JSON", "VARCHAR2", "N", nil, nil, nil, nil, nil, nil, nil, nil, nil},
+				{"REALJSON", "JSON", "N", nil, nil, nil, nil, nil, nil, nil, nil, nil},
+				{"ARRAY_NUM", "STUDENT", "N", nil, nil, nil, nil, "COLLECTION", "NUMBER", nil, 10, 5},
+				{"ARRAY_FLOAT", "STUDENT", "N", nil, nil, nil, nil, "COLLECTION", "FLOAT", nil, nil, nil},
+				{"ARRAY_STRING", "STUDENT", "N", nil, nil, nil, nil, "COLLECTION", "VARCHAR2", 15, nil, nil},
+				{"ARRAY_DATE", "STUDENT", "N", nil, nil, nil, nil, "COLLECTION", "DATE", nil, nil, nil},
+				{"ARRAY_INT", "STUDENT", "N", nil, nil, nil, nil, "COLLECTION", "NUMBER", nil, 10, 0},
+				{"OBJECT", "CONTACTS", "N", nil, nil, nil, nil, "OBJECT", nil, nil, nil, nil}}},
 	}
 	db := mkMockDB(t, ms)
 	conv := internal.MakeConv()
@@ -143,10 +179,27 @@ func TestProcessSchemaMYSQL(t *testing.T) {
 				"ID": {Name: "ID", T: ddl.Type{Name: ddl.Numeric}, NotNull: true}},
 			Pks: []ddl.IndexKey{{Col: "ID"}},
 		},
+		"TEST2": {
+			Name:     "TEST2",
+			ColNames: []string{"ID", "JSON", "REALJSON", "ARRAY_NUM", "ARRAY_FLOAT", "ARRAY_STRING", "ARRAY_DATE", "ARRAY_INT", "OBJECT"},
+			ColDefs: map[string]ddl.ColumnDef{
+				"ID":           {Name: "ID", T: ddl.Type{Name: ddl.Numeric}, NotNull: true},
+				"JSON":         {Name: "JSON", T: ddl.Type{Name: ddl.JSON}, NotNull: true},
+				"REALJSON":     {Name: "REALJSON", T: ddl.Type{Name: ddl.JSON}, NotNull: true},
+				"ARRAY_NUM":    {Name: "ARRAY_NUM", T: ddl.Type{Name: ddl.Numeric, IsArray: true}, NotNull: true},
+				"ARRAY_FLOAT":  {Name: "ARRAY_FLOAT", T: ddl.Type{Name: ddl.Float64, IsArray: true}, NotNull: true},
+				"ARRAY_STRING": {Name: "ARRAY_STRING", T: ddl.Type{Name: ddl.String, Len: int64(15), IsArray: true}, NotNull: true},
+				"ARRAY_DATE":   {Name: "ARRAY_DATE", T: ddl.Type{Name: ddl.Date, IsArray: true}, NotNull: true},
+				"ARRAY_INT":    {Name: "ARRAY_INT", T: ddl.Type{Name: ddl.Int64, IsArray: true}, NotNull: true},
+				"OBJECT":       {Name: "OBJECT", T: ddl.Type{Name: ddl.JSON}, NotNull: true},
+			},
+			Pks: []ddl.IndexKey{{Col: "ID"}},
+		},
 	}
 	assert.Equal(t, expectedSchema, stripSchemaComments(conv.SpSchema))
 	assert.Equal(t, len(conv.Issues["USER"]), 0)
 	assert.Equal(t, len(conv.Issues["TEST"]), 0)
+	assert.Equal(t, len(conv.Issues["TEST2"]), 0)
 	assert.Equal(t, int64(0), conv.Unexpecteds())
 }
 
diff --git a/sources/oracle/toddl_test.go b/sources/oracle/toddl_test.go
index 67df390..769d5ae 100644
--- a/sources/oracle/toddl_test.go
+++ b/sources/oracle/toddl_test.go
@@ -1,4 +1,4 @@
-// Copyright 2021 Google LLC
+// Copyright 2022 Google LLC
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
diff --git a/test_data/oracle.test.out b/test_data/oracle.test.out
new file mode 100644
index 0000000..b00795c
--- /dev/null
+++ b/test_data/oracle.test.out
@@ -0,0 +1,48 @@
+create user sti identified by test1;
+grant all privileges to sti;
+connect sti/test1@127.0.0.1:1521/xe;
+CREATE TABLE "STI"."ALLTYPES"(
+	"ID" NUMBER, 
+	"BINARY_DOUBLE_T" BINARY_DOUBLE, 
+	"BINARY_FLOAT_T" BINARY_FLOAT, 
+	"BLOB_T" BLOB, 
+	"CHAR_T" CHAR(1 BYTE), 
+	"CLOB_T" CLOB, 
+	"DATE_T" DATE, 
+	"DECIMAL_T" NUMBER(10,5), 
+	"DOUBLE_PRECISION_T" FLOAT(126), 
+	"FLOAT_T" FLOAT(126), 
+	"INTEGER_T" NUMBER(9,0), 
+	"INTERVAL_DAY_T" INTERVAL DAY (2) TO SECOND (6), 
+	"INTERVAL_YEAR_T" INTERVAL YEAR (2) TO MONTH, 
+	"LONG_T" LONG, 
+	"NCHAR_T" NCHAR(1), 
+	"NCLOB_T" NCLOB, 
+	"NUMBER_T" NUMBER, 
+	"NUMERIC_T" NUMBER(*,0), 
+	"NVARCHAR2_T" NVARCHAR2(100), 
+	"RAW_T" RAW(20), 
+	"REAL_T" FLOAT(63), 
+	"ROWID_T" ROWID, 
+	"SMALLINT_T" NUMBER(4,0), 
+	"TIMESTAMP_T" TIMESTAMP (6), 
+	"UROWID_T" UROWID (4000), 
+	"VARCHAR_T" VARCHAR2(20 BYTE), 
+	"VARCHAR2_T" VARCHAR2(4000 BYTE),  
+	"TIMESTAMP_W_Z" TIMESTAMP (6) WITH TIME ZONE, 
+	"TIMESTAMP_W_L_Z" TIMESTAMP (6) WITH LOCAL TIME ZONE, 
+	 PRIMARY KEY ("ID"));
+
+
+Insert into STI.ALLTYPES (ID,BINARY_DOUBLE_T,BINARY_FLOAT_T,CHAR_T,DATE_T,DECIMAL_T,DOUBLE_PRECISION_T,FLOAT_T,INTEGER_T,INTERVAL_DAY_T,
+INTERVAL_YEAR_T,LONG_T,NCHAR_T,NUMBER_T,NUMERIC_T,NVARCHAR2_T,RAW_T,REAL_T,
+ROWID_T,SMALLINT_T,TIMESTAMP_T,UROWID_T,VARCHAR_T,VARCHAR2_T,TIMESTAMP_W_Z,
+TIMESTAMP_W_L_Z) values (3,'1234.56','1234.56','C',to_date('18-JAN-22','DD-MON-RR'),
+12345.12345,1.2345,1234.56789,42,'+01 02:34:56.000000','+04-02',
+'In publishing and graphic design, Lorem ipsum is a placeholder text commonly used to demonstrate the visual form of a document or a typeface without relying on meaningful content. Lorem ipsum may be used as a placeholder before the final copy is available',
+'C',42,42,'some nvarchar2 data','736F6D65207261772064617461',42.001,null,9999,to_timestamp('19-JAN-22 11.27.18.262000000 AM',
+'DD-MON-RR HH.MI.SSXFF AM'),null,'some varchar data','some varcahr2 data',
+to_timestamp_tz('15-FEB-21 01.22.11.871000000 PM +02:00','DD-MON-RR HH.MI.SSXFF AM TZR'),
+to_timestamp('15-FEB-22 11.22.11.871000000 AM','DD-MON-RR HH.MI.SSXFF AM'));
+
+select DATE_T, FLOAT_T, INTEGER_T, NUMERIC_T, VARCHAR_T, TIMESTAMP_T from STI.ALLTYPES;
\ No newline at end of file
diff --git a/testing/oracle/integration_test.go b/testing/oracle/integration_test.go
new file mode 100644
index 0000000..ee5c344
--- /dev/null
+++ b/testing/oracle/integration_test.go
@@ -0,0 +1,206 @@
+// 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 oracle_test
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"math/big"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"cloud.google.com/go/civil"
+	"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"
+	"github.com/google/go-cmp/cmp"
+	"github.com/stretchr/testify/assert"
+
+	"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
+)
+
+type SpannerRecord struct {
+	Date_t      spanner.NullDate
+	Float_t     float64
+	Int_t       int64
+	Numeric_t   string
+	String_t    string
+	Timestamp_t string
+}
+
+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()
+	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_ORACLE_SchemaSubcommand(t *testing.T) {
+	onlyRunForEmulatorTest(t)
+	t.Parallel()
+	tmpdir := prepareIntegrationTest(t)
+	defer os.RemoveAll(tmpdir)
+	filePrefix := filepath.Join(tmpdir, "Oracle_IntTest.")
+
+	args := fmt.Sprintf("schema -prefix %s -source=%s -source-profile='host=localhost,user=STI,db_name=xe,password=test1,port=1521'", filePrefix, constants.ORACLE)
+	err := common.RunCommand(args, projectID)
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+func TestIntegration_ORACLE_SchemaAndDataSubcommand(t *testing.T) {
+	onlyRunForEmulatorTest(t)
+	tmpdir := prepareIntegrationTest(t)
+	defer os.RemoveAll(tmpdir)
+
+	dbName := "schema-and-data"
+	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
+	filePrefix := filepath.Join(tmpdir, "Oracle_IntTest.")
+
+	args := fmt.Sprintf("schema-and-data -prefix %s -source=%s  -source-profile='host=localhost,user=STI,db_name=xe,password=test1,port=1521' -target-profile='instance=%s,dbname=%s'", filePrefix, constants.ORACLE, 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) {
+	wantRecord := SpannerRecord{
+		Date_t:      spanner.NullDate{Valid: true, Date: civil.Date{Day: 18, Year: 2022, Month: 01}},
+		Float_t:     float64(1234.56789),
+		Int_t:       int64(42),
+		Numeric_t:   "42.000000000",
+		String_t:    "some varchar data",
+		Timestamp_t: "2022-01-19T11:27:18.262Z",
+	}
+	var date spanner.NullDate
+	var floatVal float64
+	var intVal int64
+	var numericVal big.Rat
+	var stringVal string
+	var timeVal spanner.NullTime
+	iter := client.Single().Read(ctx, "ALLTYPES", spanner.Key{3}, []string{"DATE_T", "FLOAT_T", "INTEGER_T", "NUMERIC_T", "VARCHAR_T", "TIMESTAMP_T"})
+	defer iter.Stop()
+	for {
+		row, err := iter.Next()
+		if err == iterator.Done {
+			break
+		}
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := row.Columns(&date, &floatVal, &intVal, &numericVal, &stringVal, &timeVal); err != nil {
+			t.Fatal(err)
+		}
+
+		fmt.Fprintf(os.Stdout, "%v,%v,%v,%v,%v,%v", date, floatVal, intVal, numericVal, stringVal, timeVal)
+		gotRecord := SpannerRecord{
+			Date_t:      date,
+			Float_t:     floatVal,
+			Int_t:       intVal,
+			Numeric_t:   numericVal.FloatString(9),
+			String_t:    stringVal,
+			Timestamp_t: timeVal.String(),
+		}
+		assert.True(t, cmp.Equal(wantRecord, gotRecord))
+	}
+}
+
+func onlyRunForEmulatorTest(t *testing.T) {
+	if os.Getenv("SPANNER_EMULATOR_HOST") == "" {
+		t.Skip("Skipping tests only running against the emulator.")
+	}
+}