Repository: cloudspannerecosystem/harbourbridge (go)

Issue:
Standardizing DynamoDb implementation to use common structure
## Expected Behavior
The DynamoDb implementation should use the same interfaces as the common db code in sources/common/ as outlined in Issue #178 and PR #181 

## Actual Behavior
Current implementation is completely independent of the structure used in Mysql and Postgres.
Making it follow the same code flow will ensure a more standard code base and provide a template for further non-relational DB integrations.

Pull request description:
Common implementation for DynamoDB
Fixes #189 

Refactored dynamo db schema and data implementation to conform to common InfoSchema struct implementation.

- [x] Tests pass
- [x] Appropriate changes to README are included in PR

Patch:
diff --git a/cmd/cmd.go b/cmd/cmd.go
index 84bba51..ebd4a7c 100644
--- a/cmd/cmd.go
+++ b/cmd/cmd.go
@@ -81,7 +81,7 @@ func CommandLine(ctx context.Context, driver, targetDb, dbURI string, dataOnly,
 
 	// We pass an empty string to the sqlConnectionStr parameter as this is the legacy codepath,
 	// which reads the environment variables and constructs the string later on.
-	bw, err := conversion.DataConv(driver, "", ioHelper, client, conv, dataOnly)
+	bw, err := conversion.DataConv(driver, "", ioHelper, client, conv, dataOnly, schemaSampleSize)
 	if err != nil {
 		return fmt.Errorf("can't finish data conversion for db %s: %v", dbURI, err)
 	}
diff --git a/cmd/data.go b/cmd/data.go
index 922c844..0b40afd 100644
--- a/cmd/data.go
+++ b/cmd/data.go
@@ -129,7 +129,7 @@ func (cmd *DataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface
 		return subcommands.ExitFailure
 	}
 
-	bw, err := conversion.DataConv(driverName, getSQLConnectionStr(sourceProfile), &ioHelper, client, conv, true)
+	bw, err := conversion.DataConv(driverName, getSQLConnectionStr(sourceProfile), &ioHelper, client, conv, true, getSchemaSampleSize(sourceProfile))
 	if err != nil {
 		err = fmt.Errorf("can't finish data conversion for db %s: %v", dbURI, err)
 		return subcommands.ExitFailure
diff --git a/cmd/schema_and_data.go b/cmd/schema_and_data.go
index f24e247..47657e3 100644
--- a/cmd/schema_and_data.go
+++ b/cmd/schema_and_data.go
@@ -98,9 +98,9 @@ func (cmd *SchemaAndDataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...
 	}
 
 	sqlConnectionStr := getSQLConnectionStr(sourceProfile)
-
+	schemaSampleSize := getSchemaSampleSize(sourceProfile)
 	var conv *internal.Conv
-	conv, err = conversion.SchemaConv(driverName, sqlConnectionStr, targetDb, &ioHelper, getSchemaSampleSize(sourceProfile))
+	conv, err = conversion.SchemaConv(driverName, sqlConnectionStr, targetDb, &ioHelper, schemaSampleSize)
 	if err != nil {
 		panic(err)
 	}
@@ -134,7 +134,7 @@ func (cmd *SchemaAndDataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...
 		return subcommands.ExitFailure
 	}
 
-	bw, err := conversion.DataConv(driverName, sqlConnectionStr, &ioHelper, client, conv, true)
+	bw, err := conversion.DataConv(driverName, sqlConnectionStr, &ioHelper, client, conv, true, schemaSampleSize)
 	if err != nil {
 		err = fmt.Errorf("can't finish data conversion for db %s: %v", dbURI, err)
 		return subcommands.ExitFailure
diff --git a/conversion/conversion.go b/conversion/conversion.go
index dc7e12c..703ffac 100644
--- a/conversion/conversion.go
+++ b/conversion/conversion.go
@@ -74,6 +74,9 @@ var (
 )
 
 // SchemaConv performs the schema conversion
+// TODO: Pass around cmd.SourceProfile instead of sqlConnectionStr and schemaSampleSize.
+// Doing that requires refactoring since that would introduce a circular dependency between
+// conversion.go and cmd/source_profile.go.
 // The sqlConnectionStr param provides the connection details to use the go SQL library.
 // It is empty in the following cases:
 //  - Driver is DynamoDB or a dump file mode.
@@ -81,12 +84,10 @@ var (
 // When using source-profile, the sqlConnectionStr is constructed from the input params.
 func SchemaConv(driver, sqlConnectionStr, targetDb string, ioHelper *IOStreams, schemaSampleSize int64) (*internal.Conv, error) {
 	switch driver {
-	case constants.POSTGRES, constants.MYSQL:
-		return schemaFromSQL(driver, sqlConnectionStr, targetDb)
+	case constants.POSTGRES, constants.MYSQL, constants.DYNAMODB:
+		return schemaFromDatabase(driver, sqlConnectionStr, targetDb, schemaSampleSize)
 	case constants.PGDUMP, constants.MYSQLDUMP:
 		return schemaFromDump(driver, targetDb, ioHelper)
-	case constants.DYNAMODB:
-		return schemaFromDynamoDB(schemaSampleSize)
 	default:
 		return nil, fmt.Errorf("schema conversion for driver %s not supported", driver)
 	}
@@ -97,8 +98,8 @@ func SchemaConv(driver, sqlConnectionStr, targetDb string, ioHelper *IOStreams,
 // It is empty in the following cases:
 //  - Driver is DynamoDB or a dump file mode.
 //  - This function is called as part of the legacy global CLI flag mode. (This string is constructed from env variables later on)
-// When using source-profile, the sqlConnectionStr is constructed from the input params.
-func DataConv(driver, sqlConnectionStr string, ioHelper *IOStreams, client *sp.Client, conv *internal.Conv, dataOnly bool) (*spanner.BatchWriter, error) {
+// When using source-profile, the sqlConnectionStr and schemaSampleSize are constructed from the input params.
+func DataConv(driver, sqlConnectionStr string, ioHelper *IOStreams, client *sp.Client, conv *internal.Conv, dataOnly bool, schemaSampleSize int64) (*spanner.BatchWriter, error) {
 	config := spanner.BatchWriterConfig{
 		BytesLimit: 100 * 1000 * 1000,
 		WriteLimit: 40,
@@ -106,26 +107,36 @@ func DataConv(driver, sqlConnectionStr string, ioHelper *IOStreams, client *sp.C
 		Verbose:    internal.Verbose(),
 	}
 	switch driver {
-	case constants.POSTGRES, constants.MYSQL:
-		return dataFromSQL(driver, sqlConnectionStr, config, client, conv)
+	case constants.POSTGRES, constants.MYSQL, constants.DYNAMODB:
+		return dataFromDatabase(driver, sqlConnectionStr, config, client, conv, schemaSampleSize)
 	case constants.PGDUMP, constants.MYSQLDUMP:
 		if conv.SpSchema.CheckInterleaved() {
 			return nil, fmt.Errorf("harbourBridge does not currently support data conversion from dump files\nif the schema contains interleaved tables. Suggest using direct access to source database\ni.e. using drivers postgres and mysql")
 		}
 		return dataFromDump(driver, config, ioHelper, client, conv, dataOnly)
-	case constants.DYNAMODB:
-		return dataFromDynamoDB(config, client, conv)
 	default:
 		return nil, fmt.Errorf("data conversion for driver %s not supported", driver)
 	}
 }
 
-func generateSQLConnectionStr(driver string) (string, error) {
+func connectionConfig(driver string, sqlConnectionStr string) (interface{}, error) {
 	switch driver {
 	case constants.POSTGRES:
-		return generatePGSQLConnectionStr()
+		// If empty, this is called as part of the legacy mode witih global CLI flags.
+		// When using source-profile mode is used, the sqlConnectionStr is already populated.
+		if sqlConnectionStr == "" {
+			return generatePGSQLConnectionStr()
+		}
+		return sqlConnectionStr, nil
 	case constants.MYSQL:
-		return generateMYSQLConnectionStr()
+		// If empty, this is called as part of the legacy mode witih global CLI flags.
+		// When using source-profile mode is used, the sqlConnectionStr is already populated.
+		if sqlConnectionStr == "" {
+			return generateMYSQLConnectionStr()
+		}
+		return sqlConnectionStr, nil
+	case constants.DYNAMODB:
+		return getDynamoDBClientConfig()
 	default:
 		return "", fmt.Errorf("driver %s not supported", driver)
 	}
@@ -182,55 +193,22 @@ func getDbNameFromSQLConnectionStr(driver, sqlConnectionStr string) string {
 	return ""
 }
 
-// TODO: Pass around cmd.SourceProfile instead of sqlConnectionStr.
-// Doing that requires refactoring since that would introduce a circular dependency between
-// conversion.go and cmd/source_profile.go.
-func schemaFromSQL(driver, sqlConnectionStr, targetDb string) (*internal.Conv, error) {
-	// If empty, this is called as part of the legacy mode witih global CLI flags.
-	// When using source-profile mode is used, the sqlConnectionStr is already populated.
-	if sqlConnectionStr == "" {
-		var err error
-		sqlConnectionStr, err = generateSQLConnectionStr(driver)
-		if err != nil {
-			return nil, err
-		}
-	}
-	sourceDB, err := sql.Open(driver, sqlConnectionStr)
-	if err != nil {
-		return nil, err
-	}
+func schemaFromDatabase(driver, sqlConnectionStr, targetDb string, schemaSampleSize int64) (*internal.Conv, error) {
 	conv := internal.MakeConv()
 	conv.TargetDb = targetDb
-	conv.SrcDbName = getDbNameFromSQLConnectionStr(driver, sqlConnectionStr)
-	err = ProcessSchema(driver, conv, sourceDB)
+	infoSchema, err := GetInfoSchema(driver, sqlConnectionStr, schemaSampleSize)
 	if err != nil {
-		return nil, err
+		return conv, err
 	}
-	return conv, nil
+	return conv, common.ProcessSchema(conv, infoSchema)
 }
 
-func dataFromSQL(driver, sqlConnectionStr string, config spanner.BatchWriterConfig, client *sp.Client, conv *internal.Conv) (*spanner.BatchWriter, error) {
-	// TODO: Refactor to avoid redundant calls to sqlConnectionStr and
-	// Open in schemaFromSQL and dataFromSQL. Also refactor to
-	// share code with dataFromPgDump. Use single transaction for
-	// reading schema and data from source db to get consistent
-	// dump.
-	if sqlConnectionStr == "" {
-		// If empty, this is called as part of the old command line workflow.
-		var err error
-		sqlConnectionStr, err = generateSQLConnectionStr(driver)
-		if err != nil {
-			return nil, err
-		}
-	}
-	sourceDB, err := sql.Open(driver, sqlConnectionStr)
-	if err != nil {
-		return nil, err
-	}
-	err = SetRowStats(driver, conv, sourceDB)
+func dataFromDatabase(driver, sqlConnectionStr string, config spanner.BatchWriterConfig, client *sp.Client, conv *internal.Conv, schemaSampleSize int64) (*spanner.BatchWriter, error) {
+	infoSchema, err := GetInfoSchema(driver, sqlConnectionStr, schemaSampleSize)
 	if err != nil {
 		return nil, err
 	}
+	common.SetRowStats(conv, infoSchema)
 	totalRows := conv.Rows()
 	p := internal.NewProgress(totalRows, "Writing data to Spanner", internal.Verbose(), false)
 	rows := int64(0)
@@ -249,64 +227,18 @@ func dataFromSQL(driver, sqlConnectionStr string, config spanner.BatchWriterConf
 		func(table string, cols []string, vals []interface{}) {
 			writer.AddRow(table, cols, vals)
 		})
-	err = ProcessData(driver, conv, sourceDB)
-	if err != nil {
-		return nil, err
-	}
+	common.ProcessData(conv, infoSchema)
 	writer.Flush()
 	return writer, nil
 }
 
-func getDynamoDBClientConfig() *aws.Config {
+func getDynamoDBClientConfig() (*aws.Config, error) {
 	cfg := aws.Config{}
 	endpointOverride := os.Getenv("DYNAMODB_ENDPOINT_OVERRIDE")
 	if endpointOverride != "" {
 		cfg.Endpoint = aws.String(endpointOverride)
 	}
-	return &cfg
-}
-
-func schemaFromDynamoDB(sampleSize int64) (*internal.Conv, error) {
-	conv := internal.MakeConv()
-	mySession := session.Must(session.NewSession())
-	dydbClient := dydb.New(mySession, getDynamoDBClientConfig())
-	err := dynamodb.ProcessSchema(conv, dydbClient, []string{}, sampleSize)
-	if err != nil {
-		return nil, err
-	}
-	return conv, nil
-}
-
-func dataFromDynamoDB(config spanner.BatchWriterConfig, client *sp.Client, conv *internal.Conv) (*spanner.BatchWriter, error) {
-	mySession := session.Must(session.NewSession())
-	dydbClient := dydb.New(mySession, getDynamoDBClientConfig())
-	dynamodb.SetRowStats(conv, dydbClient)
-	totalRows := conv.Rows()
-	p := internal.NewProgress(totalRows, "Writing data to Spanner", internal.Verbose(), false)
-
-	rows := int64(0)
-	config.Write = func(m []*sp.Mutation) error {
-		_, err := client.Apply(context.Background(), m)
-		if err != nil {
-			return err
-		}
-		atomic.AddInt64(&rows, int64(len(m)))
-		p.MaybeReport(atomic.LoadInt64(&rows))
-		return nil
-	}
-	writer := spanner.NewBatchWriter(config)
-	conv.SetDataMode()
-	conv.SetDataSink(
-		func(table string, cols []string, vals []interface{}) {
-			writer.AddRow(table, cols, vals)
-		})
-
-	err := dynamodb.ProcessData(conv, dydbClient)
-	if err != nil {
-		return nil, err
-	}
-	writer.Flush()
-	return writer, nil
+	return &cfg, nil
 }
 
 // IOStreams is a struct that contains the file descriptor for dumpFile.
@@ -1158,41 +1090,30 @@ func ProcessDump(driver string, conv *internal.Conv, r *internal.Reader) error {
 	}
 }
 
-// ProcessSchema invokes processSchema function from a sql package based on driver selected.
-func ProcessSchema(driver string, conv *internal.Conv, db *sql.DB) error {
-	switch driver {
-	case constants.MYSQL:
-		return common.ProcessSchema(conv, mysql.InfoSchemaImpl{DbName: conv.SrcDbName, Db: db})
-	case constants.POSTGRES:
-		return common.ProcessSchema(conv, postgres.InfoSchemaImpl{Db: db})
-	default:
-		return fmt.Errorf("schema conversion for driver %s not supported", driver)
-	}
-}
-
-// SetRowStats invokes SetRowStats function from a sql package based on driver selected.
-func SetRowStats(driver string, conv *internal.Conv, db *sql.DB) error {
-	switch driver {
-	case constants.MYSQL:
-		common.SetRowStats(conv, mysql.InfoSchemaImpl{DbName: conv.SrcDbName, Db: db})
-	case constants.POSTGRES:
-		common.SetRowStats(conv, postgres.InfoSchemaImpl{Db: db})
-	default:
-		return fmt.Errorf("could not set rows stats for '%s' driver", driver)
+func GetInfoSchema(driver, sqlConnectionStr string, schemaSampleSize int64) (common.InfoSchema, error) {
+	connectionConfig, err := connectionConfig(driver, sqlConnectionStr)
+	if err != nil {
+		return nil, err
 	}
-	return nil
-}
-
-// ProcessData invokes ProcessData function from a sql package based on driver selected.
-func ProcessData(driver string, conv *internal.Conv, db *sql.DB) error {
 	switch driver {
-	//TODO - move this logic into a factory within the sources dir
 	case constants.MYSQL:
-		common.ProcessData(conv, mysql.InfoSchemaImpl{DbName: conv.SrcDbName, Db: db})
+		db, err := sql.Open(driver, connectionConfig.(string))
+		dbName := getDbNameFromSQLConnectionStr(driver, connectionConfig.(string))
+		if err != nil {
+			return nil, err
+		}
+		return mysql.InfoSchemaImpl{DbName: dbName, Db: db}, nil
 	case constants.POSTGRES:
-		common.ProcessData(conv, postgres.InfoSchemaImpl{Db: db})
+		db, err := sql.Open(driver, connectionConfig.(string))
+		if err != nil {
+			return nil, err
+		}
+		return postgres.InfoSchemaImpl{Db: db}, nil
+	case constants.DYNAMODB:
+		mySession := session.Must(session.NewSession())
+		dydbClient := dydb.New(mySession, connectionConfig.(*aws.Config))
+		return dynamodb.InfoSchemaImpl{DynamoClient: dydbClient, SampleSize: schemaSampleSize}, nil
 	default:
-		return fmt.Errorf("data conversion for driver %s is not supported", driver)
+		return nil, fmt.Errorf("driver %s not supported", driver)
 	}
-	return nil
 }
diff --git a/internal/convert.go b/internal/convert.go
index 356ca4b..acf0ed4 100644
--- a/internal/convert.go
+++ b/internal/convert.go
@@ -28,7 +28,6 @@ type Conv struct {
 	SpSchema       ddl.Schema                          // Maps Spanner table name to Spanner schema.
 	SyntheticPKeys map[string]SyntheticPKey            // Maps Spanner table name to synthetic primary key (if needed).
 	SrcSchema      map[string]schema.Table             // Maps source-DB table name to schema information.
-	SrcDbName      string                              // Name of the source database. Currently only used for mysql infoschema.
 	Issues         map[string]map[string][]SchemaIssue // Maps source-DB table/col to list of schema conversion issues.
 	ToSpanner      map[string]NameAndCols              // Maps from source-DB table name to Spanner name and column mapping.
 	ToSource       map[string]NameAndCols              // Maps from Spanner table name to source-DB table name and column mapping.
diff --git a/sources/dynamodb/data.go b/sources/dynamodb/data.go
index cba4728..d2740bc 100644
--- a/sources/dynamodb/data.go
+++ b/sources/dynamodb/data.go
@@ -19,74 +19,20 @@ import (
 	"fmt"
 	"math/big"
 
-	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/dynamodb"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 )
 
-// ProcessData performs data conversion for DynamoDB database. For each table,
-// we extract data using Scan requests, 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.
-func ProcessData(conv *internal.Conv, client dynamoClient) error {
-	for srcTable, srcSchema := range conv.SrcSchema {
-		spTable, err1 := internal.GetSpannerTable(conv, srcTable)
-		spCols, err2 := internal.GetSpannerCols(conv, srcTable, srcSchema.ColNames)
-		spSchema, ok := conv.SpSchema[spTable]
-		if err1 != nil || err2 != nil || !ok {
-			conv.Stats.BadRows[srcTable] += conv.Stats.Rows[srcTable]
-			conv.Unexpected(fmt.Sprintf("Can't get cols and schemas for table %s: err1=%s, err2=%s, ok=%t",
-				srcTable, err1, err2, ok))
-			continue
-		}
-
-		err := scan(srcTable, client, func(m map[string]*dynamodb.AttributeValue) {
-			spVals, badCols, srcStrVals := cvtRow(m, srcSchema, spSchema, spCols)
-			if len(badCols) == 0 {
-				conv.WriteRow(srcTable, spTable, spCols, spVals)
-			} else {
-				conv.Unexpected(fmt.Sprintf("Data conversion error for table %s in column(s) %s\n", srcTable, badCols))
-				conv.StatsAddBadRow(srcTable, conv.DataMode())
-				conv.CollectBadRow(srcTable, srcSchema.ColNames, srcStrVals)
-			}
-		})
-		if err != nil {
-			conv.Stats.BadRows[srcTable] += conv.Stats.Rows[srcTable]
-			conv.Unexpected(fmt.Sprintf("Can't scan the data for table %s: %s", srcTable, err))
-		}
-	}
-	return nil
-}
-
-func scan(table string, client dynamoClient, f func(map[string]*dynamodb.AttributeValue)) error {
-	var lastEvaluatedKey map[string]*dynamodb.AttributeValue
-	for {
-		// Build the query input parameters.
-		params := &dynamodb.ScanInput{
-			TableName: aws.String(table),
-		}
-		if lastEvaluatedKey != nil {
-			params.ExclusiveStartKey = lastEvaluatedKey
-		}
-
-		// Make the DynamoDB Query API call.
-		result, err := client.Scan(params)
-		if err != nil {
-			return fmt.Errorf("failed to make Query API call for table %v: %v", table, err)
-		}
-
-		// Iterate the items returned.
-		for _, attrsMap := range result.Items {
-			f(attrsMap)
-		}
-		if result.LastEvaluatedKey == nil {
-			return nil
-		}
-		// If there are more rows, then continue.
-		lastEvaluatedKey = result.LastEvaluatedKey
+func ProcessDataRow(m map[string]*dynamodb.AttributeValue, conv *internal.Conv, srcTable string, srcSchema schema.Table, spTable string, spCols []string, spSchema ddl.CreateTable) {
+	spVals, badCols, srcStrVals := cvtRow(m, srcSchema, spSchema, spCols)
+	if len(badCols) == 0 {
+		conv.WriteRow(srcTable, spTable, spCols, spVals)
+	} else {
+		conv.Unexpected(fmt.Sprintf("Data conversion error for table %s in column(s) %s\n", srcTable, badCols))
+		conv.StatsAddBadRow(srcTable, conv.DataMode())
+		conv.CollectBadRow(srcTable, srcSchema.ColNames, srcStrVals)
 	}
 }
 
diff --git a/sources/dynamodb/schema.go b/sources/dynamodb/schema.go
index 2f2ec20..868cced 100644
--- a/sources/dynamodb/schema.go
+++ b/sources/dynamodb/schema.go
@@ -23,9 +23,11 @@ import (
 	sp "cloud.google.com/go/spanner"
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/dynamodb"
+	"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbiface"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
+	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 )
 
 const (
@@ -45,47 +47,29 @@ const (
 	conflictThreshold = float64(0.05)
 )
 
-type dynamoClient interface {
-	ListTables(input *dynamodb.ListTablesInput) (*dynamodb.ListTablesOutput, error)
-	DescribeTable(input *dynamodb.DescribeTableInput) (*dynamodb.DescribeTableOutput, error)
-	Scan(input *dynamodb.ScanInput) (*dynamodb.ScanOutput, error)
+type InfoSchemaImpl struct {
+	DynamoClient dynamodbiface.DynamoDBAPI
+	SampleSize   int64
 }
 
-// ProcessSchema performs schema conversion for source tables in a DynamoDB
-// database. Since DynamoDB is a schemaless database, this process is imprecise.
-// We obtain schema information from two sources: from the table's metadata,
-// and from analyzing a sample of the table's rows.
-func ProcessSchema(conv *internal.Conv, client dynamoClient, tables []string, sampleSize int64) error {
-	if len(tables) == 0 {
-		var err error
-		tables, err = listTables(client)
-		if err != nil {
-			return err
-		}
-		if len(tables) == 0 {
-			return fmt.Errorf("no DynamoDB table exists under this account")
-		}
-	}
-	for _, t := range tables {
-		if err := processTable(conv, client, t, sampleSize); err != nil {
-			return err
-		}
-	}
-	common.SchemaToSpannerDDL(conv, ToDdlImpl{})
-	conv.AddPrimaryKeys()
-	return nil
+func (isi InfoSchemaImpl) GetToDdl() common.ToDdl {
+	return ToDdlImpl{}
 }
 
-func listTables(client dynamoClient) ([]string, error) {
-	var tables []string
+func (isi InfoSchemaImpl) GetTableName(schema string, tableName string) string {
+	return *aws.String(tableName)
+}
+
+func (isi InfoSchemaImpl) GetTables() ([]common.SchemaAndName, error) {
+	var tables []common.SchemaAndName
 	input := &dynamodb.ListTablesInput{}
 	for {
-		result, err := client.ListTables(input)
+		result, err := isi.DynamoClient.ListTables(input)
 		if err != nil {
 			return nil, err
 		}
 		for _, t := range result.TableNames {
-			tables = append(tables, *t)
+			tables = append(tables, common.SchemaAndName{Name: *t})
 		}
 
 		if result.LastEvaluatedTableName == nil {
@@ -95,40 +79,79 @@ func listTables(client dynamoClient) ([]string, error) {
 	}
 }
 
-func processTable(conv *internal.Conv, client dynamoClient, table string, sampleSize int64) error {
-	dySchema := schema.Table{Name: table}
-	err := analyzeMetadata(client, &dySchema)
+func (isi InfoSchemaImpl) GetColumns(conv *internal.Conv, table common.SchemaAndName, constraints map[string][]string, primaryKeys []string) (map[string]schema.Column, []string, error) {
+	stats, count, err := scanSampleData(isi.DynamoClient, isi.SampleSize, table.Name)
 	if err != nil {
-		return err
+		return nil, nil, err
 	}
-	stats, count, err := scanSampleData(client, sampleSize, dySchema.Name)
-	if err != nil {
-		return err
-	}
-	inferDataTypes(stats, count, &dySchema)
+	return inferDataTypes(stats, count, primaryKeys)
+}
 
-	// Sort column names in increasing order, because the server may return them
-	// in a random order.
-	sort.Strings(dySchema.ColNames)
-	conv.SrcSchema[table] = dySchema
-	return nil
+func (isi InfoSchemaImpl) GetRowsFromTable(conv *internal.Conv, srcTable string) (interface{}, error) {
+	var lastEvaluatedKey map[string]*dynamodb.AttributeValue
+	for {
+		// Build the query input parameters.
+		params := &dynamodb.ScanInput{
+			TableName: aws.String(srcTable),
+		}
+		if lastEvaluatedKey != nil {
+			params.ExclusiveStartKey = lastEvaluatedKey
+		}
+
+		// Make the DynamoDB Query API call.
+		result, err := isi.DynamoClient.Scan(params)
+		if err != nil {
+			return nil, fmt.Errorf("failed to make Query API call for table %v: %v", srcTable, err)
+		}
+
+		if result.LastEvaluatedKey == nil {
+			return result.Items, nil
+		}
+		// If there are more rows, then continue.
+		lastEvaluatedKey = result.LastEvaluatedKey
+	}
 }
 
-func analyzeMetadata(client dynamoClient, s *schema.Table) error {
+func (isi InfoSchemaImpl) GetRowCount(table common.SchemaAndName) (int64, error) {
 	input := &dynamodb.DescribeTableInput{
-		TableName: aws.String(s.Name),
+		TableName: aws.String(table.Name),
 	}
+	result, err := isi.DynamoClient.DescribeTable(input)
+	if err != nil {
+		return 0, err
+	}
+	return *result.Table.ItemCount, err
+}
 
-	result, err := client.DescribeTable(input)
+func (isi InfoSchemaImpl) GetConstraints(conv *internal.Conv, table common.SchemaAndName) (primaryKeys []string, constraints map[string][]string, err error) {
+	input := &dynamodb.DescribeTableInput{
+		TableName: aws.String(table.Name),
+	}
+	result, err := isi.DynamoClient.DescribeTable(input)
 	if err != nil {
-		return fmt.Errorf("failed to make a DescribeTable API call for table %v: %v", s.Name, err)
+		return primaryKeys, constraints, fmt.Errorf("failed to make a DescribeTable API call for table %v: %v", table.Name, err)
 	}
 
 	// Primary keys.
 	for _, i := range result.Table.KeySchema {
-		s.PrimaryKeys = append(s.PrimaryKeys, schema.Key{Column: *i.AttributeName})
+		primaryKeys = append(primaryKeys, *i.AttributeName)
 	}
+	return primaryKeys, constraints, nil
+}
+
+func (isi InfoSchemaImpl) GetForeignKeys(conv *internal.Conv, table common.SchemaAndName) (foreignKeys []schema.ForeignKey, err error) {
+	return foreignKeys, err
+}
 
+func (isi InfoSchemaImpl) GetIndexes(conv *internal.Conv, table common.SchemaAndName) (indexes []schema.Index, err error) {
+	input := &dynamodb.DescribeTableInput{
+		TableName: aws.String(table.Name),
+	}
+
+	result, err := isi.DynamoClient.DescribeTable(input)
+	if err != nil {
+		return nil, fmt.Errorf("failed to make a DescribeTable API call for table %v: %v", table.Name, err)
+	}
 	// DynamoDB supports 2 types of indexes: Global Secondary Indexes (GSI) and Local Secondary Indexes (LSI).
 	// In GSI, dydb creates another global table for indexes that scales seperately.
 	// As for LSI, every partition in dydb maintains its own local index for that partition.
@@ -137,14 +160,31 @@ func analyzeMetadata(client dynamoClient, s *schema.Table) error {
 
 	// Convert secondary indexes from GlobalSecondaryIndexes.
 	for _, i := range result.Table.GlobalSecondaryIndexes {
-		s.Indexes = append(s.Indexes, getSchemaIndexStruct(*i.IndexName, i.KeySchema))
+		indexes = append(indexes, getSchemaIndexStruct(*i.IndexName, i.KeySchema))
 	}
 
 	// Convert secondary indexes from LocalSecondaryIndexes.
 	for _, i := range result.Table.LocalSecondaryIndexes {
-		s.Indexes = append(s.Indexes, getSchemaIndexStruct(*i.IndexName, i.KeySchema))
+		indexes = append(indexes, getSchemaIndexStruct(*i.IndexName, i.KeySchema))
 	}
+	return indexes, nil
+}
 
+// ProcessData performs data conversion for DynamoDB database. For each table,
+// we extract data using Scan requests, 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.
+func (isi InfoSchemaImpl) ProcessData(conv *internal.Conv, srcTable string, srcSchema schema.Table, spTable string, spCols []string, spSchema ddl.CreateTable) error {
+	rows, 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
+	}
+	// Iterate the items returned.
+	for _, attrsMap := range rows.([]map[string]*dynamodb.AttributeValue) {
+		ProcessDataRow(attrsMap, conv, srcTable, srcSchema, spTable, spCols, spSchema)
+	}
 	return nil
 }
 
@@ -156,7 +196,7 @@ func getSchemaIndexStruct(indexName string, keySchema []*dynamodb.KeySchemaEleme
 	return schema.Index{Name: indexName, Keys: keys}
 }
 
-func scanSampleData(client dynamoClient, sampleSize int64, table string) (map[string]map[string]int64, int64, error) {
+func scanSampleData(client dynamodbiface.DynamoDBAPI, sampleSize int64, table string) (map[string]map[string]int64, int64, error) {
 	// A map from column name to a count map of possible data types.
 	stats := make(map[string]map[string]int64)
 	var count int64
@@ -248,10 +288,9 @@ type statItem struct {
 	Count int64
 }
 
-func inferDataTypes(stats map[string]map[string]int64, rows int64, s *schema.Table) {
-	if s.ColDefs == nil {
-		s.ColDefs = make(map[string]schema.Column)
-	}
+func inferDataTypes(stats map[string]map[string]int64, rows int64, primaryKeys []string) (map[string]schema.Column, []string, error) {
+	colDefs := make(map[string]schema.Column)
+	var colNames []string
 
 	for col, countMap := range stats {
 		var statItems, candidates []statItem
@@ -273,8 +312,8 @@ func inferDataTypes(stats map[string]map[string]int64, rows int64, s *schema.Tab
 
 		// Check if the column is a part of a primary key.
 		isPKey := false
-		for _, pk := range s.PrimaryKeys {
-			if pk.Column == col {
+		for _, pk := range primaryKeys {
+			if pk == col {
 				isPKey = true
 				break
 			}
@@ -294,16 +333,20 @@ func inferDataTypes(stats map[string]map[string]int64, rows int64, s *schema.Tab
 			}
 		}
 
-		s.ColNames = append(s.ColNames, col)
+		colNames = append(colNames, col)
 		if len(candidates) == 1 {
-			s.ColDefs[col] = schema.Column{Name: col, Type: schema.Type{Name: candidates[0].Type}, NotNull: !nullable}
+			colDefs[col] = schema.Column{Name: col, Type: schema.Type{Name: candidates[0].Type}, NotNull: !nullable}
 		} else {
 			// If there is no any candidate or more than a single candidate,
 			// this column has a significant conflict on data types and then
 			// defaults to a String type.
-			s.ColDefs[col] = schema.Column{Name: col, Type: schema.Type{Name: typeString}, NotNull: !nullable}
+			colDefs[col] = schema.Column{Name: col, Type: schema.Type{Name: typeString}, NotNull: !nullable}
 		}
 	}
+	// Sort column names in increasing order, because the server may return them
+	// in a random order.
+	sort.Strings(colNames)
+	return colDefs, colNames, nil
 }
 
 // numericParsable determines whether its argument is a valid Spanner numeric
@@ -340,33 +383,3 @@ func numericParsable(n string) bool {
 
 	return true
 }
-
-// SetRowStats populates conv with the number of rows in each table. In
-// DynamoDB, we use describe_table api to get the number of total rows, but this
-// number is updated approximately every six hours. This means that our row
-// count could be out of date by up to six hours. One of the primary uses of
-// row count is for calculating progress during data conversion. As a result, if
-// there have been huge changes in the number of rows in a table over the last
-// six hours, the progress calculation could be inaccurate.
-func SetRowStats(conv *internal.Conv, client dynamoClient) {
-	tables, err := listTables(client)
-	if err != nil {
-		conv.Unexpected(fmt.Sprintf("Couldn't get list of table: %s", err))
-		return
-	}
-	if len(tables) == 0 {
-		conv.Unexpected("no DynamoDB table exists under this account")
-		return
-	}
-	for _, t := range tables {
-		input := &dynamodb.DescribeTableInput{
-			TableName: aws.String(t),
-		}
-		result, err := client.DescribeTable(input)
-		if err != nil {
-			conv.Unexpected(fmt.Sprintf("failed to make a DescribeTable API call for table %v: %v", t, err))
-			return
-		}
-		conv.Stats.Rows[t] = *result.Table.ItemCount
-	}
-}


Test patch:
diff --git a/sources/dynamodb/data_test.go b/sources/dynamodb/data_test.go
index ea5f7f1..7347a82 100644
--- a/sources/dynamodb/data_test.go
+++ b/sources/dynamodb/data_test.go
@@ -33,58 +33,37 @@ type spannerData struct {
 	vals  []interface{}
 }
 
-func TestProcessData(t *testing.T) {
+func TestProcessDataRow(t *testing.T) {
 	strA := "str-1"
-	strB := "str-2"
 	numStr1 := "10.1"
 	numStr2 := "12.34"
-	numStr3 := "89.0"
 	numVal1 := big.NewRat(101, 10)
-	numVal2 := big.NewRat(89, 1)
 
 	boolVal := true
-	scanOutputs := []dynamodb.ScanOutput{
+	items := []map[string]*dynamodb.AttributeValue{
 		{
-			Items: []map[string]*dynamodb.AttributeValue{
-				{
-					"a": {S: &strA},
-					"b": {N: &numStr1},
-					"c": {N: &numStr2},
-					"d": {BOOL: &boolVal},
-				},
-			},
-			LastEvaluatedKey: map[string]*dynamodb.AttributeValue{
-				"a": {S: &strA},
-			},
-		},
-		{
-			Items: []map[string]*dynamodb.AttributeValue{
-				{
-					"a": {S: &strB},
-					"b": {N: &numStr3},
-				},
-			},
+			"a": {S: &strA},
+			"b": {N: &numStr1},
+			"c": {N: &numStr2},
+			"d": {BOOL: &boolVal},
 		},
 	}
 
-	client := &mockDynamoClient{
-		scanOutputs: scanOutputs,
-	}
-
 	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.String, Len: ddl.MaxLength}},
-				"b": {Name: "b", T: ddl.Type{Name: ddl.Numeric}},
-				"c": {Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
-				"d": {Name: "d", T: ddl.Type{Name: ddl.Bool}},
-			},
-			Pks: []ddl.IndexKey{{Col: "a"}},
+	spSchema := ddl.CreateTable{
+		Name:     tableName,
+		ColNames: cols,
+		ColDefs: map[string]ddl.ColumnDef{
+			"a": {Name: "a", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+			"b": {Name: "b", T: ddl.Type{Name: ddl.Numeric}},
+			"c": {Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+			"d": {Name: "d", T: ddl.Type{Name: ddl.Bool}},
 		},
+		Pks: []ddl.IndexKey{{Col: "a"}},
+	}
+	conv := buildConv(
+		spSchema,
 		schema.Table{
 			Name:     tableName,
 			ColNames: cols,
@@ -103,7 +82,9 @@ func TestProcessData(t *testing.T) {
 		func(table string, cols []string, vals []interface{}) {
 			rows = append(rows, spannerData{table: table, cols: cols, vals: vals})
 		})
-	ProcessData(conv, client)
+	for _, attrsMap := range items {
+		ProcessDataRow(attrsMap, conv, tableName, conv.SrcSchema[tableName], tableName, cols, spSchema)
+	}
 	assert.Equal(t,
 		[]spannerData{
 			{
@@ -111,11 +92,6 @@ func TestProcessData(t *testing.T) {
 				cols:  cols,
 				vals:  []interface{}{"str-1", *numVal1, "12.34", true},
 			},
-			{
-				table: tableName,
-				cols:  cols,
-				vals:  []interface{}{"str-2", *numVal2, nil, nil},
-			},
 		},
 		rows,
 	)
diff --git a/sources/dynamodb/schema_test.go b/sources/dynamodb/schema_test.go
index 8400392..4f7a00f 100644
--- a/sources/dynamodb/schema_test.go
+++ b/sources/dynamodb/schema_test.go
@@ -16,11 +16,14 @@ package dynamodb
 
 import (
 	"fmt"
+	"math/big"
 	"testing"
 
 	"github.com/aws/aws-sdk-go/service/dynamodb"
+	"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbiface"
 	"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"
 )
@@ -32,6 +35,7 @@ type mockDynamoClient struct {
 	describeTableOutputs   []dynamodb.DescribeTableOutput
 	scanCallCount          int
 	scanOutputs            []dynamodb.ScanOutput
+	dynamodbiface.DynamoDBAPI
 }
 
 func (m *mockDynamoClient) ListTables(input *dynamodb.ListTablesInput) (*dynamodb.ListTablesOutput, error) {
@@ -85,6 +89,23 @@ func TestProcessSchema(t *testing.T) {
 				},
 			},
 		},
+		{
+			Table: &dynamodb.TableDescription{
+				TableName: &tableNameA,
+				KeySchema: []*dynamodb.KeySchemaElement{
+					{AttributeName: &attrNameA, KeyType: &hashKeyType},
+				},
+			},
+		},
+		{
+			Table: &dynamodb.TableDescription{
+				TableName: &tableNameB,
+				KeySchema: []*dynamodb.KeySchemaElement{
+					{AttributeName: &attrNameA, KeyType: &hashKeyType},
+					{AttributeName: &attrNameB, KeyType: &sortKeyType},
+				},
+			},
+		},
 		{
 			Table: &dynamodb.TableDescription{
 				TableName: &tableNameB,
@@ -143,11 +164,10 @@ func TestProcessSchema(t *testing.T) {
 		describeTableOutputs: describeTableOutputs,
 		scanOutputs:          scanOutputs,
 	}
-	tables := []string{}
 	sampleSize := int64(10000)
 
 	conv := internal.MakeConv()
-	err := ProcessSchema(conv, client, tables, sampleSize)
+	err := common.ProcessSchema(conv, InfoSchemaImpl{client, sampleSize})
 
 	assert.Nil(t, err)
 	expectedSchema := map[string]ddl.CreateTable{
@@ -209,6 +229,15 @@ func TestProcessSchema_FullDataTypes(t *testing.T) {
 				},
 			},
 		},
+		{
+			Table: &dynamodb.TableDescription{
+				TableName: &tableNameA,
+				KeySchema: []*dynamodb.KeySchemaElement{
+					{AttributeName: &attrNameA, KeyType: &hashKeyType},
+					{AttributeName: &attrNameB, KeyType: &sortKeyType},
+				},
+			},
+		},
 	}
 	scanOutputs := []dynamodb.ScanOutput{
 		{
@@ -238,11 +267,10 @@ func TestProcessSchema_FullDataTypes(t *testing.T) {
 		describeTableOutputs: describeTableOutputs,
 		scanOutputs:          scanOutputs,
 	}
-	tables := []string{}
 	sampleSize := int64(10000)
 
 	conv := internal.MakeConv()
-	err := ProcessSchema(conv, client, tables, sampleSize)
+	err := common.ProcessSchema(conv, InfoSchemaImpl{client, sampleSize})
 
 	assert.Nil(t, err)
 	expectedSchema := map[string]ddl.CreateTable{
@@ -269,8 +297,76 @@ func TestProcessSchema_FullDataTypes(t *testing.T) {
 	assert.Equal(t, int64(0), conv.Unexpecteds())
 }
 
+func TestProcessData(t *testing.T) {
+	strA := "str-1"
+	numStr1 := "10.1"
+	numStr2 := "12.34"
+	numVal1 := big.NewRat(101, 10)
+
+	boolVal := true
+	scanOutputs := []dynamodb.ScanOutput{
+		{
+			Items: []map[string]*dynamodb.AttributeValue{
+				{
+					"a": {S: &strA},
+					"b": {N: &numStr1},
+					"c": {N: &numStr2},
+					"d": {BOOL: &boolVal},
+				},
+			},
+		},
+	}
+
+	client := &mockDynamoClient{
+		scanOutputs: scanOutputs,
+	}
+
+	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.String, Len: ddl.MaxLength}},
+				"b": {Name: "b", T: ddl.Type{Name: ddl.Numeric}},
+				"c": {Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+				"d": {Name: "d", T: ddl.Type{Name: ddl.Bool}},
+			},
+			Pks: []ddl.IndexKey{{Col: "a"}},
+		},
+		schema.Table{
+			Name:     tableName,
+			ColNames: cols,
+			ColDefs: map[string]schema.Column{
+				"a": {Name: "a", Type: schema.Type{Name: typeString}},
+				"b": {Name: "b", Type: schema.Type{Name: typeNumber}},
+				"c": {Name: "c", Type: schema.Type{Name: typeNumberString}},
+				"d": {Name: "d", Type: schema.Type{Name: typeBool}},
+			},
+			PrimaryKeys: []schema.Key{{Column: "a"}},
+		},
+	)
+	conv.SetDataMode()
+	var rows []spannerData
+	conv.SetDataSink(
+		func(table string, cols []string, vals []interface{}) {
+			rows = append(rows, spannerData{table: table, cols: cols, vals: vals})
+		})
+	common.ProcessData(conv, InfoSchemaImpl{client, 10})
+	assert.Equal(t,
+		[]spannerData{
+			{
+				table: tableName,
+				cols:  cols,
+				vals:  []interface{}{"str-1", *numVal1, "12.34", true},
+			},
+		},
+		rows,
+	)
+}
+
 func TestInferDataTypes(t *testing.T) {
-	dySchema := schema.Table{Name: "test"}
 	stats := map[string]map[string]int64{
 		"all_rows_not_null": {
 			"Number": 1000,
@@ -317,14 +413,15 @@ func TestInferDataTypes(t *testing.T) {
 		},
 		"empty_stats": {},
 	}
-	inferDataTypes(stats, 1000, &dySchema)
+	colDefs, colNames, err := inferDataTypes(stats, 1000, make([]string, 0))
+	assert.Nil(t, err)
 	expectColNames := []string{
 		"all_rows_not_null", "err_row", "err_null_row", "enough_null_row",
 		"not_conflict_row", "conflict_row", "equal_conflict_rows",
 		"not_conflict_row_with_noise", "conflict_row_with_noise",
 		"equal_conflict_row_with_noise",
 	}
-	assert.ElementsMatch(t, expectColNames, dySchema.ColNames)
+	assert.ElementsMatch(t, expectColNames, colNames)
 	assert.Equal(t, map[string]schema.Column{
 		"all_rows_not_null":             {Name: "all_rows_not_null", Type: schema.Type{Name: "Number"}, NotNull: true},
 		"err_row":                       {Name: "err_row", Type: schema.Type{Name: "Number"}, NotNull: true},
@@ -336,7 +433,7 @@ func TestInferDataTypes(t *testing.T) {
 		"not_conflict_row_with_noise":   {Name: "not_conflict_row_with_noise", Type: schema.Type{Name: "Number"}, NotNull: false},
 		"conflict_row_with_noise":       {Name: "conflict_row_with_noise", Type: schema.Type{Name: "String"}, NotNull: false},
 		"equal_conflict_row_with_noise": {Name: "equal_conflict_row_with_noise", Type: schema.Type{Name: "String"}, NotNull: false},
-	}, dySchema.ColDefs)
+	}, colDefs)
 }
 
 func TestScanSampleData(t *testing.T) {
@@ -391,7 +488,7 @@ func TestScanSampleData(t *testing.T) {
 	assert.Equal(t, expectedStats, stats)
 }
 
-func TestParseIndexes(t *testing.T) {
+func TestInfoSchemaImpl_GetIndexes(t *testing.T) {
 	tableName := "test"
 	attrNameA := "a"
 	attrNameB := "b"
@@ -433,21 +530,73 @@ func TestParseIndexes(t *testing.T) {
 		describeTableOutputs: describeTableOutputs,
 	}
 
-	dySchema := schema.Table{Name: "test"}
-
-	err := analyzeMetadata(client, &dySchema)
+	dySchema := common.SchemaAndName{Name: "test"}
+	conv := internal.MakeConv()
+	isi := InfoSchemaImpl{client, 10}
+	indexes, err := isi.GetIndexes(conv, dySchema)
 	assert.Nil(t, err)
 
-	pKeys := []schema.Key{{Column: "a"}, {Column: "b"}}
-	assert.Equal(t, pKeys, dySchema.PrimaryKeys)
 	secIndexes := []schema.Index{
 		{Name: "secondary_index_c", Keys: []schema.Key{{Column: "c"}}},
 		{Name: "secondary_index_d", Keys: []schema.Key{{Column: "d"}}},
 	}
-	assert.Equal(t, secIndexes, dySchema.Indexes)
+	assert.Equal(t, secIndexes, indexes)
 }
 
-func TestListTables(t *testing.T) {
+func TestInfoSchemaImpl_GetConstraints(t *testing.T) {
+	tableName := "test"
+	attrNameA := "a"
+	attrNameB := "b"
+	attrNameC := "c"
+	attrNameD := "d"
+	hashKeyType := "HASH"
+	sortKeyType := "RANGE"
+	globalIndexName := "secondary_index_c"
+	localIndexName := "secondary_index_d"
+	describeTableOutputs := []dynamodb.DescribeTableOutput{
+		{
+			Table: &dynamodb.TableDescription{
+				TableName: &tableName,
+				KeySchema: []*dynamodb.KeySchemaElement{
+					{AttributeName: &attrNameA, KeyType: &hashKeyType},
+					{AttributeName: &attrNameB, KeyType: &sortKeyType},
+				},
+				GlobalSecondaryIndexes: []*dynamodb.GlobalSecondaryIndexDescription{
+					{
+						IndexName: &globalIndexName,
+						KeySchema: []*dynamodb.KeySchemaElement{
+							{AttributeName: &attrNameC, KeyType: &hashKeyType},
+						},
+					},
+				},
+				LocalSecondaryIndexes: []*dynamodb.LocalSecondaryIndexDescription{
+					{
+						IndexName: &localIndexName,
+						KeySchema: []*dynamodb.KeySchemaElement{
+							{AttributeName: &attrNameD, KeyType: &hashKeyType},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	client := &mockDynamoClient{
+		describeTableOutputs: describeTableOutputs,
+	}
+
+	dySchema := common.SchemaAndName{Name: "test"}
+	conv := internal.MakeConv()
+	isi := InfoSchemaImpl{client, 10}
+	primaryKeys, constraints, err := isi.GetConstraints(conv, dySchema)
+	assert.Nil(t, err)
+
+	pKeys := []string{"a", "b"}
+	assert.Equal(t, pKeys, primaryKeys)
+	assert.Empty(t, constraints)
+}
+
+func TestInfoSchemaImpl_GetTables(t *testing.T) {
 	tableNameA := "table-a"
 	tableNameB := "table-b"
 
@@ -459,10 +608,217 @@ func TestListTables(t *testing.T) {
 	client := &mockDynamoClient{
 		listTableOutputs: listTableOutputs,
 	}
+	isi := InfoSchemaImpl{client, 10}
+	tables, err := isi.GetTables()
+	assert.Nil(t, err)
+	assert.Equal(t, []common.SchemaAndName{{"", "table-a"}, {"", "table-b"}}, tables)
+}
+
+func TestInfoSchemaImpl_GetTableName(t *testing.T) {
+	tableNameA := "table-a"
 
-	tables, err := listTables(client)
+	client := &mockDynamoClient{}
+	isi := InfoSchemaImpl{client, 10}
+	table := isi.GetTableName("", tableNameA)
+	assert.Equal(t, tableNameA, table)
+}
+
+func TestInfoSchemaImpl_GetColumns(t *testing.T) {
+	strA := "str-1"
+	strB := "str-2"
+	numStr := "10"
+	scanOutputs := []dynamodb.ScanOutput{
+		{
+			Items: []map[string]*dynamodb.AttributeValue{
+				{
+					"a": {S: &strA},
+				},
+				{
+					"a": {N: &numStr},
+				},
+			},
+			LastEvaluatedKey: map[string]*dynamodb.AttributeValue{
+				"a": {S: &strA},
+			},
+		},
+		{
+			Items: []map[string]*dynamodb.AttributeValue{
+				{
+					"a": {S: &strB},
+					"b": {N: &numStr},
+				},
+				{
+					// This will not be scaned due to the sample size.
+					"a": {N: &numStr},
+					"b": {S: &strB},
+				},
+			},
+		},
+	}
+
+	conv := internal.MakeConv()
+	client := &mockDynamoClient{
+		scanOutputs: scanOutputs,
+	}
+	dySchema := common.SchemaAndName{Name: "test"}
+
+	isi := InfoSchemaImpl{client, 10}
+
+	colDefs, colNames, err := isi.GetColumns(conv, dySchema, nil, nil)
+	assert.Nil(t, err)
+	expectColNames := []string{
+		"a", "b",
+	}
+	assert.ElementsMatch(t, expectColNames, colNames)
+	assert.Equal(t, map[string]schema.Column{
+		"a": {Name: "a", Type: schema.Type{Name: "String", Mods: []int64(nil), ArrayBounds: []int64(nil)}, NotNull: true, Ignored: schema.Ignored{Check: false, Identity: false, Default: false, Exclusion: false, ForeignKey: false, AutoIncrement: false}},
+		"b": {Name: "b", Type: schema.Type{Name: "String", Mods: []int64(nil), ArrayBounds: []int64(nil)}, NotNull: false, Ignored: schema.Ignored{Check: false, Identity: false, Default: false, Exclusion: false, ForeignKey: false, AutoIncrement: false}}},
+		colDefs)
+}
+
+func TestInfoSchemaImpl_GetForeignKeys(t *testing.T) {
+	dySchema := common.SchemaAndName{Name: "test"}
+	conv := internal.MakeConv()
+	client := &mockDynamoClient{}
+	isi := InfoSchemaImpl{client, 10}
+	fk, err := isi.GetForeignKeys(conv, dySchema)
+	assert.Nil(t, err)
+	assert.Nil(t, fk)
+}
+
+func TestInfoSchemaImpl_GetRowCount(t *testing.T) {
+	tableNameA := "test_a"
+	tableItemCountA := int64(10)
+
+	describeTableOutputs := []dynamodb.DescribeTableOutput{
+		{
+			Table: &dynamodb.TableDescription{
+				TableName: &tableNameA,
+				ItemCount: &tableItemCountA,
+			},
+		},
+	}
+
+	client := &mockDynamoClient{
+		describeTableOutputs: describeTableOutputs,
+	}
+
+	isi := InfoSchemaImpl{client, 10}
+	dySchema := common.SchemaAndName{Name: tableNameA}
+
+	rowCount, err := isi.GetRowCount(dySchema)
 	assert.Nil(t, err)
-	assert.Equal(t, []string{"table-a", "table-b"}, tables)
+	assert.Equal(t, tableItemCountA, rowCount)
+}
+
+func TestInfoSchemaImpl_GetRowsFromTable(t *testing.T) {
+	strA := "str-1"
+	numStr1 := "10.1"
+	numStr2 := "12.34"
+
+	boolVal := true
+	scanOutputs := []dynamodb.ScanOutput{
+		{
+			Items: []map[string]*dynamodb.AttributeValue{
+				{
+					"a": {S: &strA},
+					"b": {N: &numStr1},
+					"c": {N: &numStr2},
+					"d": {BOOL: &boolVal},
+				},
+			},
+		},
+	}
+
+	conv := internal.MakeConv()
+	client := &mockDynamoClient{
+		scanOutputs: scanOutputs,
+	}
+	tableName := "testtable"
+	isi := InfoSchemaImpl{client, 10}
+
+	rows, err := isi.GetRowsFromTable(conv, tableName)
+	assert.Nil(t, err)
+	assert.Equal(t, []map[string]*dynamodb.AttributeValue{{
+		"a": {S: &strA},
+		"b": {N: &numStr1},
+		"c": {N: &numStr2},
+		"d": {BOOL: &boolVal}}},
+		rows,
+	)
+}
+
+func TestInfoSchemaImpl_ProcessData(t *testing.T) {
+	strA := "str-1"
+	numStr1 := "10.1"
+	numStr2 := "12.34"
+	numVal1 := big.NewRat(101, 10)
+
+	boolVal := true
+	scanOutputs := []dynamodb.ScanOutput{
+		{
+			Items: []map[string]*dynamodb.AttributeValue{
+				{
+					"a": {S: &strA},
+					"b": {N: &numStr1},
+					"c": {N: &numStr2},
+					"d": {BOOL: &boolVal},
+				},
+			},
+		},
+	}
+
+	client := &mockDynamoClient{
+		scanOutputs: scanOutputs,
+	}
+	isi := InfoSchemaImpl{client, 10}
+
+	tableName := "testtable"
+	cols := []string{"a", "b", "c", "d"}
+	spSchema := ddl.CreateTable{
+		Name:     tableName,
+		ColNames: cols,
+		ColDefs: map[string]ddl.ColumnDef{
+			"a": {Name: "a", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+			"b": {Name: "b", T: ddl.Type{Name: ddl.Numeric}},
+			"c": {Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+			"d": {Name: "d", T: ddl.Type{Name: ddl.Bool}},
+		},
+		Pks: []ddl.IndexKey{{Col: "a"}},
+	}
+	conv := buildConv(
+		spSchema,
+		schema.Table{
+			Name:     tableName,
+			ColNames: cols,
+			ColDefs: map[string]schema.Column{
+				"a": {Name: "a", Type: schema.Type{Name: typeString}},
+				"b": {Name: "b", Type: schema.Type{Name: typeNumber}},
+				"c": {Name: "c", Type: schema.Type{Name: typeNumberString}},
+				"d": {Name: "d", Type: schema.Type{Name: typeBool}},
+			},
+			PrimaryKeys: []schema.Key{{Column: "a"}},
+		},
+	)
+
+	var rows []spannerData
+	conv.SetDataSink(
+		func(table string, cols []string, vals []interface{}) {
+			rows = append(rows, spannerData{table: table, cols: cols, vals: vals})
+		})
+	err := isi.ProcessData(conv, tableName, conv.SrcSchema[tableName], tableName,
+		cols, spSchema)
+	assert.Nil(t, err)
+	assert.Equal(t,
+		[]spannerData{
+			{
+				table: tableName,
+				cols:  cols,
+				vals:  []interface{}{"str-1", *numVal1, "12.34", true},
+			},
+		},
+		rows,
+	)
 }
 
 func stripSchemaComments(spSchema map[string]ddl.CreateTable) map[string]ddl.CreateTable {
@@ -550,7 +906,7 @@ func TestSetRowStats(t *testing.T) {
 		describeTableOutputs: describeTableOutputs,
 	}
 
-	SetRowStats(conv, client)
+	common.SetRowStats(conv, InfoSchemaImpl{client, 10})
 
 	assert.Equal(t, tableItemCountA, conv.Stats.Rows[tableNameA])
 	assert.Equal(t, tableItemCountB, conv.Stats.Rows[tableNameB])