The idea here is run multiple DynamoDB queries in parallel as the query is run over a GSI. As of now BatchGetItems doesn't support querying over Indexes and the recommended approach is to Query the data in parallel. I'm using go routines with wg to take care of the executions of the routines in parallel.
The input to the function is an array of strings with an ID, the outputs are the attributes of the Ids.
When the function is run locally, there is no issue, however, when the function is run on AWS-Lambda, the returned data keeps growing;
ie; Input 2 items should output 2 items. If the function is tested on AWS-Lambda,
- 1st time the function returns 2 item
- 2nd time it returns 4 items (same items are repeated 2 times)
- 3rd time it returns 6 items (same items are repeated 4 times)
and so on. Here is a snippet of the code. Is there something not handled correctly that's having the lambda output the extra set of data every-time the lambda is run?
package main
import (
"context"
"fmt"
"os"
"sync"
"github.com/aws/aws-lambda-go/lambda"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/aws/session"
"github.com/aws/aws-sdk-go/service/dynamodb"
"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbattribute"
)
//Final Output Interface
var bulkOutput []interface{}
func exitWithError(err error) {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
//LambdaInputJSON input for the lambda handler
type LambdaInputJSON struct {
Ids []string `json:"ids,omitempty"`
}
//HandleRequest : Lambda entry point
func HandleRequest(ctx context.Context, data LambdaInputJSON) ([]interface{}, error) {
return DynamoDBBatchGetRecords(data), nil
}
func main() {
lambda.Start(HandleRequest)
}
func DynamoDBBatchGetRecords(a LambdaInputJSON) []interface{} {
var wg sync.WaitGroup
var mutex = &sync.Mutex{}
iterations := len(a.Ids)
wg.Add(iterations)
for i := 0; i < iterations; i++ {
go QueryOutput(a.Ids[i], &wg, mutex)
}
wg.Wait()
return bulkOutput
}
//QueryOutput GoRoutine
func QueryOutput(data string, wg *sync.WaitGroup, mtx *sync.Mutex) {
var outputData []interface{}
defer wg.Done()
sess, err := session.NewSession(&aws.Config{
Region: aws.String("aws-region"),
})
if err != nil {
exitWithError(fmt.Errorf("failed to make Query API call, %v", err))
}
ddb := dynamodb.New(sess)
queryInput := &dynamodb.QueryInput{
Limit: aws.Int64(1),
TableName: aws.String("table-name"),
IndexName: aws.String("gsi-index"),
ScanIndexForward: aws.Bool(false),
ConsistentRead: aws.Bool(false),
KeyConditions: map[string]*dynamodb.Condition{
"column_name": {
ComparisonOperator: aws.String("EQ"),
AttributeValueList: []*dynamodb.AttributeValue{
{
S: aws.String(data),
},
},
},
},
}
output, err := ddb.Query(queryInput)
if err != nil {
exitWithError(fmt.Errorf("Failed to make Query API call, %v", err))
}
err = dynamodbattribute.UnmarshalListOfMaps(output.Items, &outputData)
if err != nil {
exitWithError(fmt.Errorf("Failed to unmarshal Query result items, %v", err))
}
mtx.Lock()
bulkOutput = append(bulkOutput, outputData)
mtx.Unlock()
}