Merge branch 'main' into renovate/docker-login-action-4.x

bootjp · web-flow · commit 0d17ad5ac396 · 2026-03-08T21:08:58.000+09:00
diff --git a/README.md b/README.md
@@ -9,7 +9,8 @@ Elastickv is an experimental project undertaking the challenge of creating a dis
 - **Raft-based Data Replication**: KV state replication is implemented on Raft, with leader-based commit and follower forwarding paths.
 - **Shard-aware Data Plane**: Static shard ranges across multiple Raft groups with shard routing/coordinator are implemented.
 - **Durable Route Control Plane (Milestone 1)**: Durable route catalog, versioned route snapshot apply, watcher-based route refresh, and manual `ListRoutes`/`SplitRange` (same-group split) are implemented.
-- **Protocol Adapters**: gRPC (`RawKV`/`TransactionalKV`), Redis (core commands + `MULTI/EXEC` and list operations), and DynamoDB-compatible API (`PutItem`/`GetItem`/`UpdateItem`/`TransactWriteItems`) implementations are available (runtime exposure depends on the selected server entrypoint/configuration).
+- **Protocol Adapters**: gRPC (`RawKV`/`TransactionalKV`), Redis (core commands + `MULTI/EXEC` and list operations), and DynamoDB-compatible API (`PutItem`/`GetItem`/`DeleteItem`/`UpdateItem`/`TransactWriteItems`) implementations are available (runtime exposure depends on the selected server entrypoint/configuration).
+- **DynamoDB Compatibility Scope**: `CreateTable`/`DeleteTable`/`DescribeTable`/`ListTables`/`PutItem`/`GetItem`/`DeleteItem`/`UpdateItem`/`Query`/`TransactWriteItems` are implemented. `Scan` and `BatchWriteItem` are currently unsupported.
 - **Basic Consistency Behaviors**: Write-after-read checks, leader redirection/forwarding paths, and OCC conflict detection for transactional writes are covered by tests.
 
 ## Planned Features
diff --git a/adapter/dynamodb.go b/adapter/dynamodb.go
@@ -29,6 +29,7 @@ const (
 	describeTableTarget      = targetPrefix + "DescribeTable"
 	listTablesTarget         = targetPrefix + "ListTables"
 	putItemTarget            = targetPrefix + "PutItem"
+	deleteItemTarget         = targetPrefix + "DeleteItem"
 	getItemTarget            = targetPrefix + "GetItem"
 	queryTarget              = targetPrefix + "Query"
 	updateItemTarget         = targetPrefix + "UpdateItem"
@@ -51,6 +52,7 @@ const (
 	tableLockStripeCount        = 128
 	queryDefaultLimit           = 100
 	tableCleanupDeleteBatchSize = 256
+	dynamoMaxRequestBodyBytes   = 1 << 20
 
 	dynamoTableMetaPrefix       = "!ddb|meta|table|"
 	dynamoTableGenerationPrefix = "!ddb|meta|gen|"
@@ -89,6 +91,7 @@ func NewDynamoDBServer(listen net.Listener, st store.MVCCStore, coordinate kv.Co
 		describeTableTarget:      d.describeTable,
 		listTablesTarget:         d.listTables,
 		putItemTarget:            d.putItem,
+		deleteItemTarget:         d.deleteItem,
 		getItemTarget:            d.getItem,
 		queryTarget:              d.query,
 		updateItemTarget:         d.updateItem,
@@ -671,6 +674,10 @@ func (d *DynamoDBServer) retryItemWriteWithGeneration(
 		if err != nil {
 			return err
 		}
+		if plan.req == nil {
+			return nil
+		}
+		plan.req.StartTS = readTS
 		if err = d.commitItemWrite(ctx, plan.req); err != nil {
 			if !isRetryableTransactWriteError(err) {
 				return errors.WithStack(err)
@@ -778,6 +785,129 @@ func (d *DynamoDBServer) getItem(w http.ResponseWriter, r *http.Request) {
 	writeDynamoJSON(w, map[string]any{"Item": item})
 }
 
+func (d *DynamoDBServer) deleteItem(w http.ResponseWriter, r *http.Request) {
+	in, shouldReturnOld, err := decodeDeleteItemInput(http.MaxBytesReader(w, r.Body, dynamoMaxRequestBodyBytes))
+	if err != nil {
+		writeDynamoErrorFromErr(w, err)
+		return
+	}
+	lockKey, err := dynamoItemUpdateLockKey(in.TableName, in.Key)
+	if err != nil {
+		writeDynamoError(w, http.StatusBadRequest, dynamoErrValidation, err.Error())
+		return
+	}
+	unlock := d.lockItemUpdate(lockKey)
+	defer unlock()
+	oldItem, err := d.deleteItemWithRetry(r.Context(), in)
+	if err != nil {
+		writeDynamoErrorFromErr(w, err)
+		return
+	}
+	resp := map[string]any{}
+	if shouldReturnOld && len(oldItem) > 0 {
+		resp["Attributes"] = oldItem
+	}
+	writeDynamoJSON(w, resp)
+}
+
+func decodeDeleteItemInput(bodyReader io.Reader) (deleteItemInput, bool, error) {
+	body, err := io.ReadAll(bodyReader)
+	if err != nil {
+		return deleteItemInput{}, false, newDynamoAPIError(http.StatusBadRequest, dynamoErrValidation, err.Error())
+	}
+	var in deleteItemInput
+	if err := json.Unmarshal(body, &in); err != nil {
+		return deleteItemInput{}, false, newDynamoAPIError(http.StatusBadRequest, dynamoErrValidation, err.Error())
+	}
+	if strings.TrimSpace(in.TableName) == "" {
+		return deleteItemInput{}, false, newDynamoAPIError(http.StatusBadRequest, dynamoErrValidation, "missing table name")
+	}
+	shouldReturnOld, err := parseDeleteItemReturnValues(in.ReturnValues)
+	if err != nil {
+		return deleteItemInput{}, false, err
+	}
+	return in, shouldReturnOld, nil
+}
+
+func parseDeleteItemReturnValues(returnValues string) (bool, error) {
+	switch strings.TrimSpace(returnValues) {
+	case "", "NONE":
+		return false, nil
+	case "ALL_OLD":
+		return true, nil
+	default:
+		return false, newDynamoAPIError(http.StatusBadRequest, dynamoErrValidation, "unsupported ReturnValues")
+	}
+}
+
+type deleteItemPlan struct {
+	req        *kv.OperationGroup[kv.OP]
+	generation uint64
+	oldItem    map[string]attributeValue
+}
+
+func (d *DynamoDBServer) deleteItemWithRetry(ctx context.Context, in deleteItemInput) (map[string]attributeValue, error) {
+	var oldItem map[string]attributeValue
+	err := d.retryItemWriteWithGeneration(
+		ctx,
+		in.TableName,
+		"delete retry attempts exhausted",
+		func(readTS uint64) (*itemWritePlan, error) {
+			plan, err := d.prepareDeleteItemWrite(ctx, in, readTS)
+			if err != nil {
+				return nil, err
+			}
+			oldItem = plan.oldItem
+			return &itemWritePlan{
+				req:        plan.req,
+				generation: plan.generation,
+			}, nil
+		},
+	)
+	if err != nil {
+		return nil, err
+	}
+	return oldItem, nil
+}
+
+func (d *DynamoDBServer) prepareDeleteItemWrite(ctx context.Context, in deleteItemInput, readTS uint64) (*deleteItemPlan, error) {
+	schema, exists, err := d.loadTableSchemaAt(ctx, in.TableName, readTS)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+	if !exists {
+		return nil, newDynamoAPIError(http.StatusBadRequest, dynamoErrResourceNotFound, "table not found")
+	}
+	itemKey, err := schema.itemKeyFromAttributes(in.Key)
+	if err != nil {
+		return nil, newDynamoAPIError(http.StatusBadRequest, dynamoErrValidation, err.Error())
+	}
+	current, found, err := d.readItemAtKeyAt(ctx, itemKey, readTS)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+	if err := validateConditionOnItem(
+		in.ConditionExpression,
+		in.ExpressionAttributeNames,
+		in.ExpressionAttributeValues,
+		valueOrEmptyMap(current, found),
+	); err != nil {
+		return nil, newDynamoAPIError(http.StatusBadRequest, dynamoErrConditionalFailed, err.Error())
+	}
+	if !found {
+		return &deleteItemPlan{oldItem: nil}, nil
+	}
+	req, err := buildItemDeleteRequest(schema, itemKey, current)
+	if err != nil {
+		return nil, err
+	}
+	return &deleteItemPlan{
+		req:        req,
+		generation: schema.Generation,
+		oldItem:    cloneAttributeValueMap(current),
+	}, nil
+}
+
 func (d *DynamoDBServer) updateItem(w http.ResponseWriter, r *http.Request) {
 	in, err := decodeUpdateItemInput(r.Body)
 	if err != nil {
diff --git a/adapter/dynamodb_test.go b/adapter/dynamodb_test.go
@@ -2,6 +2,8 @@ package adapter
 
 import (
 	"context"
+	"io"
+	"net/http"
 	"strconv"
 	"strings"
 	"sync"
@@ -76,6 +78,142 @@ func TestDynamoDB_PutItem_GetItem(t *testing.T) {
 	assert.Equal(t, "v", valueAttr.Value)
 }
 
+func TestDynamoDB_DeleteItem(t *testing.T) {
+	t.Parallel()
+	nodes, _, _ := createNode(t, 1)
+	defer shutdown(nodes)
+
+	cfg, err := config.LoadDefaultConfig(context.Background(),
+		config.WithRegion("us-west-2"),
+		config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider("dummy", "dummy", "")),
+	)
+	require.NoError(t, err)
+
+	client := dynamodb.NewFromConfig(cfg, func(o *dynamodb.Options) {
+		o.BaseEndpoint = aws.String("http://" + nodes[0].dynamoAddress)
+	})
+	createSimpleKeyTable(t, context.Background(), client)
+
+	_, err = client.PutItem(context.Background(), &dynamodb.PutItemInput{
+		TableName: aws.String("t"),
+		Item: map[string]types.AttributeValue{
+			"key":   &types.AttributeValueMemberS{Value: "delete-target"},
+			"value": &types.AttributeValueMemberS{Value: "v"},
+		},
+	})
+	require.NoError(t, err)
+
+	delOut, err := client.DeleteItem(context.Background(), &dynamodb.DeleteItemInput{
+		TableName: aws.String("t"),
+		Key: map[string]types.AttributeValue{
+			"key": &types.AttributeValueMemberS{Value: "delete-target"},
+		},
+		ReturnValues: types.ReturnValueAllOld,
+	})
+	require.NoError(t, err)
+	require.NotEmpty(t, delOut.Attributes)
+	oldValue, ok := delOut.Attributes["value"].(*types.AttributeValueMemberS)
+	require.True(t, ok)
+	require.Equal(t, "v", oldValue.Value)
+
+	getOut, err := client.GetItem(context.Background(), &dynamodb.GetItemInput{
+		TableName: aws.String("t"),
+		Key: map[string]types.AttributeValue{
+			"key": &types.AttributeValueMemberS{Value: "delete-target"},
+		},
+	})
+	require.NoError(t, err)
+	require.Empty(t, getOut.Item)
+}
+
+func TestDynamoDB_DeleteItem_Condition(t *testing.T) {
+	t.Parallel()
+	nodes, _, _ := createNode(t, 1)
+	defer shutdown(nodes)
+
+	cfg, err := config.LoadDefaultConfig(context.Background(),
+		config.WithRegion("us-west-2"),
+		config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider("dummy", "dummy", "")),
+	)
+	require.NoError(t, err)
+
+	client := dynamodb.NewFromConfig(cfg, func(o *dynamodb.Options) {
+		o.BaseEndpoint = aws.String("http://" + nodes[0].dynamoAddress)
+	})
+	createSimpleKeyTable(t, context.Background(), client)
+
+	_, err = client.PutItem(context.Background(), &dynamodb.PutItemInput{
+		TableName: aws.String("t"),
+		Item: map[string]types.AttributeValue{
+			"key":   &types.AttributeValueMemberS{Value: "cond-target"},
+			"value": &types.AttributeValueMemberS{Value: "v"},
+		},
+	})
+	require.NoError(t, err)
+
+	_, err = client.DeleteItem(context.Background(), &dynamodb.DeleteItemInput{
+		TableName: aws.String("t"),
+		Key: map[string]types.AttributeValue{
+			"key": &types.AttributeValueMemberS{Value: "cond-target"},
+		},
+		ConditionExpression: aws.String("attribute_not_exists(#k)"),
+		ExpressionAttributeNames: map[string]string{
+			"#k": "key",
+		},
+	})
+	require.Error(t, err)
+	var condErr *types.ConditionalCheckFailedException
+	require.ErrorAs(t, err, &condErr)
+
+	getOut, err := client.GetItem(context.Background(), &dynamodb.GetItemInput{
+		TableName: aws.String("t"),
+		Key: map[string]types.AttributeValue{
+			"key": &types.AttributeValueMemberS{Value: "cond-target"},
+		},
+	})
+	require.NoError(t, err)
+	require.NotEmpty(t, getOut.Item)
+
+	_, err = client.DeleteItem(context.Background(), &dynamodb.DeleteItemInput{
+		TableName: aws.String("t"),
+		Key: map[string]types.AttributeValue{
+			"key": &types.AttributeValueMemberS{Value: "cond-target"},
+		},
+		ConditionExpression: aws.String("attribute_exists(#k)"),
+		ExpressionAttributeNames: map[string]string{
+			"#k": "key",
+		},
+	})
+	require.NoError(t, err)
+}
+
+func TestDynamoDB_DeleteItem_RequestBodyTooLarge(t *testing.T) {
+	t.Parallel()
+	nodes, _, _ := createNode(t, 1)
+	defer shutdown(nodes)
+
+	reqBody := strings.Repeat("a", dynamoMaxRequestBodyBytes+1)
+	req, err := http.NewRequestWithContext(
+		context.Background(),
+		http.MethodPost,
+		"http://"+nodes[0].dynamoAddress+"/",
+		strings.NewReader(reqBody),
+	)
+	require.NoError(t, err)
+	req.Header.Set("X-Amz-Target", deleteItemTarget)
+	req.Header.Set("Content-Type", "application/x-amz-json-1.0")
+
+	resp, err := http.DefaultClient.Do(req)
+	require.NoError(t, err)
+	defer resp.Body.Close()
+
+	require.Equal(t, http.StatusBadRequest, resp.StatusCode)
+	body, err := io.ReadAll(resp.Body)
+	require.NoError(t, err)
+	require.Contains(t, string(body), dynamoErrValidation)
+	require.Contains(t, string(body), "too large")
+}
+
 func TestDynamoDB_TransactWriteItems(t *testing.T) {
 	t.Parallel()
 	nodes, _, _ := createNode(t, 1)
diff --git a/adapter/dynamodb_types.go b/adapter/dynamodb_types.go
@@ -21,6 +21,15 @@ type putItemInput struct {
 	Item      map[string]attributeValue `json:"Item"`
 }
 
+type deleteItemInput struct {
+	TableName                 string                    `json:"TableName"`
+	Key                       map[string]attributeValue `json:"Key"`
+	ConditionExpression       string                    `json:"ConditionExpression"`
+	ExpressionAttributeNames  map[string]string         `json:"ExpressionAttributeNames"`
+	ExpressionAttributeValues map[string]attributeValue `json:"ExpressionAttributeValues"`
+	ReturnValues              string                    `json:"ReturnValues"`
+}
+
 type transactWriteItemsInput struct {
 	TransactItems []transactWriteItem `json:"TransactItems"`
 }
diff --git a/docs/docker_multinode_manual_run.md b/docs/docker_multinode_manual_run.md
@@ -220,6 +220,28 @@ redis-cli -h 10.0.0.11 -p 6379 SET survive yes
 redis-cli -h 10.0.0.12 -p 6379 GET survive
 ```
 
+## DynamoDB Compatibility Notes
+
+Current DynamoDB-compatible API coverage includes:
+
+- `CreateTable`
+- `DeleteTable`
+- `DescribeTable`
+- `ListTables`
+- `PutItem`
+- `GetItem`
+- `DeleteItem`
+- `UpdateItem`
+- `Query`
+- `TransactWriteItems`
+
+Currently unsupported commands:
+
+- `Scan`
+- `BatchWriteItem`
+
+If you migrate existing DynamoDB data, use key-based reads (`GetItem`/`Query`) and write with `PutItem`/`TransactWriteItems` instead of `Scan`/`BatchWriteItem`.
+
 ## Stop and Cleanup
 
 Stop/remove on each VM:
