Repository: cloudspannerecosystem/harbourbridge
Pull request #80 (2020-09-02 06:56:29)

feat: support DynamoDB data conversion
This PR is based on #79. 

* Data conversion: 
    - [x] initial data conversion
    - [x] encode List & Map to json strings
    - [x] add a flag for ConsistentRead mode scan. By default, set `ConsistentRead=false`. 
    - [x] set total rows & report progress (use `describe_table` api to get an inaccurate number of total rows). 
    - [x] unit tests


hengfengli: @nevinheintze I have done the implementation. Please take a review. Two main questions: 

1. There are too many null fields when converting List and Map (dynamodb) to a String (spanner). You can take a look at [here](https://github.com/cloudspannerecosystem/harbourbridge/pull/80/files#diff-14c703eaae9731ccdc2115db6e01fa9cfb4073d50c332240974c018e3b7b6f22R149-R150). The good part is that it is easier for users to convert it back to a Map/List (a struct in dynamodb sdk). 
2. For the consistent read mode, do we need to enable it for both schema and data conversion? Because they are two different api calls and it means that we cannot guarantee a consistent view when sampling for schema and doing the actual data conversion. 

Patch:
diff --git a/dynamodb/data.go b/dynamodb/data.go
new file mode 100644
index 0000000..cba4728
--- /dev/null
+++ b/dynamodb/data.go
@@ -0,0 +1,242 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package dynamodb
+
+import (
+	"encoding/json"
+	"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 cvtRow(attrsMap map[string]*dynamodb.AttributeValue, srcSchema schema.Table, spSchema ddl.CreateTable, spCols []string) ([]interface{}, []string, []string) {
+	var err error
+	var srcStrVals []string
+	var spVals []interface{}
+	var badCols []string
+	for i, srcCol := range srcSchema.ColNames {
+		var spVal interface{}
+		var srcStrVal string
+		if attrsMap[srcCol] == nil {
+			spVal = nil
+			srcStrVal = "null"
+		} else {
+			// Convert data to the target type.
+			spVal, err = cvtColValue(attrsMap[srcCol], srcSchema.ColDefs[srcCol].Type.Name, spSchema.ColDefs[spCols[i]].T.Name)
+			if err != nil {
+				badCols = append(badCols, srcCol)
+			}
+			srcStrVal = attrsMap[srcCol].GoString()
+		}
+		srcStrVals = append(srcStrVals, srcStrVal)
+		spVals = append(spVals, spVal)
+	}
+	return spVals, badCols, srcStrVals
+}
+
+func cvtColValue(attrVal *dynamodb.AttributeValue, srcType string, spType string) (interface{}, error) {
+	switch spType {
+	case ddl.Bool:
+		switch srcType {
+		case typeBool:
+			return *attrVal.BOOL, nil
+		}
+	case ddl.Bytes:
+		switch srcType {
+		case typeBinary:
+			return attrVal.B, nil
+		case typeBinarySet:
+			return attrVal.BS, nil
+		}
+	case ddl.String:
+		switch srcType {
+		case typeMap, typeList:
+			// For typeMap and typeList, attrVal is a very verbose data
+			// structure that contains null entries for unused type cases. We
+			// strip these out using stripNull. If it is important that the
+			// Spanner values can be easily unmarshalled back to
+			// dynamodb.AttributeValue types, then replace the following five
+			// lines with just:
+			// b, err := json.Marshal(attrVal)
+			// but note that this will consume extra Spanner storage.
+			val, err := stripNull(attrVal)
+			if err != nil {
+				return nil, fmt.Errorf("failed to convert %v to a go struct", attrVal.GoString())
+			}
+			b, err := json.Marshal(val)
+			if err != nil {
+				return nil, fmt.Errorf("failed to convert %v to a json string", attrVal.GoString())
+			}
+			return string(b), nil
+		case typeString:
+			return *attrVal.S, nil
+		case typeStringSet:
+			var strArr []string
+			for _, s := range attrVal.SS {
+				strArr = append(strArr, *s)
+			}
+			return strArr, nil
+		case typeNumberString:
+			return *attrVal.N, nil
+		case typeNumberStringSet:
+			var strArr []string
+			for _, s := range attrVal.NS {
+				strArr = append(strArr, *s)
+			}
+			return strArr, nil
+		}
+	case ddl.Numeric:
+		switch srcType {
+		case typeNumber:
+			s := *attrVal.N
+			val, ok := (&big.Rat{}).SetString(s)
+			if !ok {
+				return nil, fmt.Errorf("failed to convert '%v' to an NUMERIC type", s)
+			}
+			return *val, nil
+		case typeNumberSet:
+			var numArr []big.Rat
+			for _, s := range attrVal.NS {
+				val, ok := (&big.Rat{}).SetString(*s)
+				if !ok {
+					return nil, fmt.Errorf("failed to convert '%v' to an NUMERIC array", attrVal.NS)
+				}
+				numArr = append(numArr, *val)
+			}
+			return numArr, nil
+		}
+	}
+	return nil, fmt.Errorf("can't convert value of type %s to Spanner type %s", attrVal.GoString(), spType)
+}
+
+// stripNull converts a dynamodb.AttributeValue to a Go struct which can
+// be easily encoded to a json string. If we use the normal json encoder, it
+// will have many null values. The purpose of this function is to remove the
+// null values in the json string.
+func stripNull(a *dynamodb.AttributeValue) (interface{}, error) {
+	var err error
+	switch {
+	case a.M != nil:
+		cvtMap := make(map[string]interface{})
+		for k, v := range a.M {
+			cvtMap[k], err = stripNull(v)
+			if err != nil {
+				return nil, err
+			}
+		}
+		return cvtMap, nil
+	case a.L != nil:
+		var cvtList []interface{}
+		for _, v := range a.L {
+			c, err := stripNull(v)
+			if err != nil {
+				return nil, err
+			}
+			cvtList = append(cvtList, c)
+		}
+		return cvtList, nil
+	case a.B != nil:
+		return string(a.B), nil
+	case a.BOOL != nil:
+		return a.BOOL, nil
+	case a.BS != nil:
+		var bs []string
+		for _, b := range a.BS {
+			bs = append(bs, string(b))
+		}
+		return bs, nil
+	case a.N != nil:
+		return *a.N, nil
+	case a.NS != nil:
+		return a.NS, nil
+	case a.NULL != nil:
+		return a.NULL, nil
+	case a.S != nil:
+		return *a.S, nil
+	case a.SS != nil:
+		return a.SS, nil
+	default:
+		return nil, fmt.Errorf("unknown type of AttributeValue: %v", a)
+	}
+}
diff --git a/dynamodb/schema.go b/dynamodb/schema.go
index 427a7b4..6b3e3a8 100644
--- a/dynamodb/schema.go
+++ b/dynamodb/schema.go
@@ -325,3 +325,33 @@ 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
+	}
+}
diff --git a/main.go b/main.go
index 6bea778..b4cbaf5 100644
--- a/main.go
+++ b/main.go
@@ -203,11 +203,6 @@ func toSpanner(driver, projectID, instanceID, dbName string, ioHelper *ioStreams
 		}
 	}
 
-	// TODO(hengfeng): delete the following code after data conversion is done.
-	if driver == DYNAMODB {
-		return nil
-	}
-
 	db, err := createDatabase(projectID, instanceID, dbName, conv, ioHelper.out)
 	if err != nil {
 		fmt.Printf("\nCan't create database: %v\n", err)
@@ -256,6 +251,8 @@ func dataConv(driver string, ioHelper *ioStreams, client *sp.Client, conv *inter
 		return dataFromSQL(driver, config, client, conv)
 	case PGDUMP, MYSQLDUMP:
 		return dataFromDump(driver, config, ioHelper, client, conv)
+	case DYNAMODB:
+		return dataFromDynamoDB(config, client, conv)
 	default:
 		return nil, fmt.Errorf("data conversion for driver %s not supported", driver)
 	}
@@ -376,6 +373,39 @@ func schemaFromDynamoDB(sampleSize int64) (*internal.Conv, error) {
 	return conv, nil
 }
 
+func dataFromDynamoDB(config spanner.BatchWriterConfig, client *sp.Client, conv *internal.Conv) (*spanner.BatchWriter, error) {
+	mySession := session.Must(session.NewSession())
+	dyclient := dydb.New(mySession)
+
+	dynamodb.SetRowStats(conv, dyclient)
+	totalRows := conv.Rows()
+	p := internal.NewProgress(totalRows, "Writing data to Spanner", internal.Verbose())
+
+	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, dyclient)
+	if err != nil {
+		return nil, err
+	}
+	writer.Flush()
+	return writer, nil
+}
+
 type ioStreams struct {
 	in, seekableIn, out *os.File
 	bytesRead           int64


Test patch:
diff --git a/dynamodb/data_test.go b/dynamodb/data_test.go
new file mode 100644
index 0000000..ea5f7f1
--- /dev/null
+++ b/dynamodb/data_test.go
@@ -0,0 +1,262 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package dynamodb
+
+import (
+	"encoding/json"
+	"fmt"
+	"math/big"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/service/dynamodb"
+	"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 TestProcessData(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{
+				{
+					"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},
+				},
+			},
+		},
+	}
+
+	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})
+		})
+	ProcessData(conv, client)
+	assert.Equal(t,
+		[]spannerData{
+			{
+				table: tableName,
+				cols:  cols,
+				vals:  []interface{}{"str-1", *numVal1, "12.34", true},
+			},
+			{
+				table: tableName,
+				cols:  cols,
+				vals:  []interface{}{"str-2", *numVal2, nil, nil},
+			},
+		},
+		rows,
+	)
+}
+
+func TestCvtRowWithError(t *testing.T) {
+	tableName := "testtable"
+	cols := []string{"a"}
+	spSchema := ddl.CreateTable{
+		Name:     tableName,
+		ColNames: cols,
+		ColDefs: map[string]ddl.ColumnDef{
+			// Give a wrong target type.
+			"a": {Name: "a", T: ddl.Type{Name: ddl.Float64}},
+		},
+	}
+	srcSchema := schema.Table{
+		Name:     tableName,
+		ColNames: cols,
+		ColDefs: map[string]schema.Column{
+			"a": {Name: "a", Type: schema.Type{Name: typeString}},
+		},
+	}
+	strA := "str-1"
+	attrs := map[string]*dynamodb.AttributeValue{
+		"a": {S: &strA},
+	}
+	_, badCols, srcStrVals := cvtRow(attrs, srcSchema, spSchema, cols)
+
+	assert.Equal(t, []string{"a"}, badCols)
+	assert.Equal(t, []string{attrs["a"].GoString()}, srcStrVals)
+}
+
+func TestCvtColValue(t *testing.T) {
+	str := "str-1"
+	numStr := "1234.56789"
+	boolVal := true
+	binaryVal := []byte("ABC")
+	binarySetVal := [][]byte{binaryVal}
+	listVal := []*dynamodb.AttributeValue{
+		{S: &str},
+		{N: &numStr},
+	}
+	mapVal := map[string]*dynamodb.AttributeValue{
+		"list": {L: listVal},
+	}
+	stringSetVal := []*string{&str}
+	numVal := big.NewRat(123456789, 100000)
+
+	testcases := []struct {
+		name    string
+		srcType string                   // Source DB type.
+		spType  string                   // Spanner DB type.
+		in      *dynamodb.AttributeValue // Input value for conversion.
+		want    interface{}              // Expected result.
+	}{
+		{"bool", typeBool, ddl.Bool, &dynamodb.AttributeValue{BOOL: &boolVal}, true},
+		{"binary", typeBinary, ddl.Bytes, &dynamodb.AttributeValue{B: binaryVal}, binaryVal},
+		{"binary set", typeBinarySet, ddl.Bytes, &dynamodb.AttributeValue{BS: binarySetVal}, binarySetVal},
+		{"map", typeMap, ddl.String, &dynamodb.AttributeValue{M: mapVal}, "{\"list\":[\"str-1\",\"1234.56789\"]}"},
+		{"list", typeList, ddl.String, &dynamodb.AttributeValue{L: listVal}, "[\"str-1\",\"1234.56789\"]"},
+		{"string", typeString, ddl.String, &dynamodb.AttributeValue{S: &str}, str},
+		{"string set", typeStringSet, ddl.String, &dynamodb.AttributeValue{SS: stringSetVal}, []string{str}},
+		{"number string", typeNumberString, ddl.String, &dynamodb.AttributeValue{N: &numStr}, numStr},
+		{"number string set", typeNumberStringSet, ddl.String, &dynamodb.AttributeValue{NS: []*string{&numStr}}, []string{numStr}},
+		{"number", typeNumber, ddl.Numeric, &dynamodb.AttributeValue{N: &numStr}, *numVal},
+		{"number set", typeNumberSet, ddl.Numeric, &dynamodb.AttributeValue{NS: []*string{&numStr}}, []big.Rat{*numVal}},
+	}
+
+	for _, tc := range testcases {
+		cvtVal, err := cvtColValue(tc.in, tc.srcType, tc.spType)
+		assert.Nil(t, err, fmt.Sprintf("Failed to convert %v from %s to %s", tc.in, typeString, ddl.String))
+		assert.Equal(t, tc.want, cvtVal, tc.name)
+	}
+}
+
+func TestStripNull(t *testing.T) {
+	str := "str-1"
+	numStr := "1234.56789"
+	boolTrue := true
+	boolFalse := false
+	binaryVal := []byte("ABC")
+	binarySetVal := [][]byte{binaryVal}
+	stringSetVal := []*string{&str}
+	numberSetVal := []*string{&numStr}
+	listVal1 := []*dynamodb.AttributeValue{
+		{S: &str},
+	}
+	mapVal1 := map[string]*dynamodb.AttributeValue{
+		"list": {L: listVal1},
+	}
+	listVal2 := []*dynamodb.AttributeValue{
+		{B: binaryVal},
+		{S: &str},
+		{N: &numStr},
+		{BOOL: &boolTrue},
+		{SS: stringSetVal},
+		{BS: binarySetVal},
+		{NS: numberSetVal},
+		{NULL: &boolFalse},
+		{L: listVal1},
+		{M: mapVal1},
+	}
+	mapVal2 := map[string]*dynamodb.AttributeValue{
+		"list": {L: listVal2},
+	}
+
+	testcases := []struct {
+		name string
+		in   *dynamodb.AttributeValue // Input value for conversion.
+		want string                   // Expected result.
+	}{
+		{"binary", &dynamodb.AttributeValue{B: binaryVal}, "\"ABC\""},
+		{"string", &dynamodb.AttributeValue{S: &str}, "\"str-1\""},
+		{"number", &dynamodb.AttributeValue{N: &numStr}, "\"1234.56789\""},
+		{"bool", &dynamodb.AttributeValue{BOOL: &boolTrue}, "true"},
+		{"string set", &dynamodb.AttributeValue{SS: stringSetVal}, "[\"str-1\"]"},
+		{"binary set", &dynamodb.AttributeValue{BS: binarySetVal}, "[\"ABC\"]"},
+		{"number set", &dynamodb.AttributeValue{NS: []*string{&numStr}}, "[\"1234.56789\"]"},
+		{"null", &dynamodb.AttributeValue{NULL: &boolFalse}, "false"},
+		{"list", &dynamodb.AttributeValue{L: listVal2}, "[\"ABC\",\"str-1\",\"1234.56789\",true,[\"str-1\"],[\"ABC\"],[\"1234.56789\"],false,[\"str-1\"],{\"list\":[\"str-1\"]}]"},
+		{"map", &dynamodb.AttributeValue{M: mapVal2}, "{\"list\":[\"ABC\",\"str-1\",\"1234.56789\",true,[\"str-1\"],[\"ABC\"],[\"1234.56789\"],false,[\"str-1\"],{\"list\":[\"str-1\"]}]}"},
+	}
+	for _, tc := range testcases {
+		s, err := stripNull(tc.in)
+		assert.Nil(t, err, fmt.Sprintf("Failed to marshal an attribute value: %v", tc.in))
+		b, err := json.Marshal(s)
+		assert.Nil(t, err, fmt.Sprintf("Failed to marshal to a json string: %v", s))
+		assert.Equal(t, tc.want, string(b), tc.name)
+	}
+}
+
+func buildConv(spTable ddl.CreateTable, srcTable schema.Table) *internal.Conv {
+	conv := internal.MakeConv()
+	conv.SpSchema[spTable.Name] = spTable
+	conv.SrcSchema[srcTable.Name] = srcTable
+	conv.ToSource[spTable.Name] = internal.NameAndCols{Name: srcTable.Name, Cols: make(map[string]string)}
+	conv.ToSpanner[srcTable.Name] = internal.NameAndCols{Name: spTable.Name, Cols: make(map[string]string)}
+	for i := range spTable.ColNames {
+		conv.ToSource[spTable.Name].Cols[spTable.ColNames[i]] = srcTable.ColNames[i]
+		conv.ToSpanner[srcTable.Name].Cols[srcTable.ColNames[i]] = spTable.ColNames[i]
+	}
+	return conv
+}
diff --git a/dynamodb/schema_test.go b/dynamodb/schema_test.go
index aeeb4ef..8e7f513 100644
--- a/dynamodb/schema_test.go
+++ b/dynamodb/schema_test.go
@@ -506,3 +506,39 @@ func TestIncTypeCount_NumericArray(t *testing.T) {
 		assert.Equal(t, s[test.wantType], int64(1))
 	}
 }
+
+func TestSetRowStats(t *testing.T) {
+	tableNameA := "test_a"
+	tableNameB := "test_b"
+	tableItemCountA := int64(10)
+	tableItemCountB := int64(20)
+
+	listTableOutputs := []dynamodb.ListTablesOutput{
+		{TableNames: []*string{&tableNameA, &tableNameB}},
+	}
+	describeTableOutputs := []dynamodb.DescribeTableOutput{
+		{
+			Table: &dynamodb.TableDescription{
+				TableName: &tableNameA,
+				ItemCount: &tableItemCountA,
+			},
+		},
+		{
+			Table: &dynamodb.TableDescription{
+				TableName: &tableNameB,
+				ItemCount: &tableItemCountB,
+			},
+		},
+	}
+
+	conv := internal.MakeConv()
+	client := &mockDynamoClient{
+		listTableOutputs:     listTableOutputs,
+		describeTableOutputs: describeTableOutputs,
+	}
+
+	SetRowStats(conv, client)
+
+	assert.Equal(t, tableItemCountA, conv.Stats.Rows[tableNameA])
+	assert.Equal(t, tableItemCountB, conv.Stats.Rows[tableNameB])
+}