Method Query

pkg/ring/kv/dynamodb/dynamodb.go:140–194

(ctx context.Context, key dynamodbKey, isPrefix bool)

}

func (kv dynamodbKV) Query(ctx context.Context, key dynamodbKey, isPrefix bool) (map[string]dynamodbItem, float64, error) {
	keys := make(map[string]dynamodbItem)
	var totalCapacity float64

	co := types.ComparisonOperatorEq
	if isPrefix {
		co = types.ComparisonOperatorBeginsWith
	}

	input := &dynamodb.QueryInput{
		TableName:              kv.tableName,
		ReturnConsumedCapacity: types.ReturnConsumedCapacityTotal,
		KeyConditions: map[string]types.Condition{
			primaryKey: {
				ComparisonOperator: co,
				AttributeValueList: []types.AttributeValue{
					&types.AttributeValueMemberS{Value: key.primaryKey},
				},
			},
		},
	}

	paginator := dynamodb.NewQueryPaginator(kv.ddbClient, input)

	for paginator.HasMorePages() {
		page, err := paginator.NextPage(ctx)
		if err != nil {
			return nil, totalCapacity, err
		}
		totalCapacity += getCapacityUnits(page.ConsumedCapacity)

		for _, item := range page.Items {
			itemVersion := int64(0)
			if v, ok := item[version].(*types.AttributeValueMemberN); ok {
				parsedVersion, err := strconv.ParseInt(v.Value, 10, 0)
				if err != nil {
					kv.logger.Log("msg", "failed to parse item version", "version", v.Value, "err", err)
				} else {
					itemVersion = parsedVersion
				}
			}

			if d, ok := item[contentData].(*types.AttributeValueMemberB); ok {
				if s, ok := item[sortKey].(*types.AttributeValueMemberS); ok {
					keys[s.Value] = dynamodbItem{
						data:    d.Value,
						version: itemVersion,
					}
				}
			}
		}
	}

	return keys, totalCapacity, nil
}

func (kv dynamodbKV) Delete(ctx context.Context, key dynamodbKey) (float64, error) {
	input := &dynamodb.DeleteItemInput{