Handle on-demand tables

Author: pshastry

replicate.go

@@ -90,7 +90,7 @@ func (state *syncState) copyTable(key primaryKey) error {
 	srcDynamo := state.srcDynamo
 	dstDynamo := state.dstDynamo
 
-	if state.tableConfig.ReadQps > sourceCapacity.readCapacity {
+	if sourceCapacity.readCapacity != 0 && state.tableConfig.ReadQps > sourceCapacity.readCapacity {
 		newCapacity := provisionedThroughput{
 			state.tableConfig.ReadQps,
 			sourceCapacity.writeCapacity,
@@ -104,7 +104,7 @@ func (state *syncState) copyTable(key primaryKey) error {
 		isSourceThroughputChanged = true
 	}
 
-	if state.tableConfig.WriteQps > dstCapacity.writeCapacity {
+	if dstCapacity.writeCapacity != 0 && state.tableConfig.WriteQps > dstCapacity.writeCapacity {
 		newCapacity := provisionedThroughput{
 			dstCapacity.readCapacity,
 			state.tableConfig.WriteQps,

table.go

@@ -270,22 +270,22 @@ func (sync *syncState) getCapacity(tableName string, dynamo *dynamodb.DynamoDB)
 			logger.WithFields(logging.Fields{
 				"Error": err,
 				"Table": tableName,
-			}).Debug("Error in reading provisioned throughput")
+			}).Error("Error in reading provisioned throughput")
 		} else {
 			result := output.Table.ProvisionedThroughput
 			logger.WithFields(logging.Fields{
 				"Table": tableName,
 				"Read Capacity":     *result.ReadCapacityUnits,
 				"Write Capacity":    *result.WriteCapacityUnits,
-			}).Debug("Fetched provisioned throughput of table")
+			}).Info("Fetched provisioned throughput of table")
 			return provisionedThroughput{
 				*result.ReadCapacityUnits,
 				*result.WriteCapacityUnits,
 			}
 		}
 	}
 
-	return provisionedThroughput{0, 0}
+	return provisionedThroughput{-1, -1}
 }
 
 func (sync *syncState) getTableSize(table string, dynamo *dynamodb.DynamoDB) int64 {