Method Batch

pkg/ring/kv/dynamodb/dynamodb.go:218–281

(ctx context.Context, put map[dynamodbKey]dynamodbItem, delete []dynamodbKey)

}

func (kv dynamodbKV) Batch(ctx context.Context, put map[dynamodbKey]dynamodbItem, delete []dynamodbKey) (float64, bool, error) {
	totalCapacity := float64(0)
	writeRequestSize := len(put) + len(delete)
	if writeRequestSize == 0 {
		return totalCapacity, false, nil
	}

	writeRequestsSlices := make([][]types.TransactWriteItem, int(math.Ceil(float64(writeRequestSize)/float64(DdbBatchSizeLimit))))
	for i := range writeRequestsSlices {
		writeRequestsSlices[i] = make([]types.TransactWriteItem, 0, DdbBatchSizeLimit)
	}

	currIdx := 0
	for key, ddbItem := range put {
		item := kv.generatePutItemRequest(key, ddbItem)
		ddbPut := &types.Put{
			TableName:           kv.tableName,
			Item:                item,
			ConditionExpression: aws.String("attribute_not_exists(version) OR version = :v"),
			ExpressionAttributeValues: map[string]types.AttributeValue{
				":v": &types.AttributeValueMemberN{Value: strconv.FormatInt(ddbItem.version, 10)},
			},
		}

		writeRequestsSlices[currIdx] = append(writeRequestsSlices[currIdx], types.TransactWriteItem{Put: ddbPut})
		if len(writeRequestsSlices[currIdx]) == DdbBatchSizeLimit {
			currIdx++
		}
	}

	for _, key := range delete {
		item := generateItemKey(key)
		ddbDelete := &types.Delete{
			TableName: kv.tableName,
			Key:       item,
		}
		writeRequestsSlices[currIdx] = append(writeRequestsSlices[currIdx], types.TransactWriteItem{Delete: ddbDelete})
		if len(writeRequestsSlices[currIdx]) == DdbBatchSizeLimit {
			currIdx++
		}
	}

	for _, slice := range writeRequestsSlices {
		if len(slice) == 0 {
			continue
		}
		resp, err := kv.ddbClient.TransactWriteItems(ctx, &dynamodb.TransactWriteItemsInput{
			TransactItems: slice,
		})
		if err != nil {
			var checkFailed *types.ConditionalCheckFailedException
			isCheckFailed := errors.As(err, &checkFailed)
			if isCheckFailed {
				kv.logger.Log("msg", "conditional check failed on DynamoDB Batch", "err", err)
			}
			return totalCapacity, isCheckFailed, err
		}
		for _, consumedCapacity := range resp.ConsumedCapacity {