From a2ab236f62f6da308254da1273e0d7856a7bfc80 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Poyraz=20K=C3=BC=C3=A7=C3=BCkarslan?= <83272398+PoyrazK@users.noreply.github.com> Date: Thu, 11 Jun 2026 20:36:14 +0300 Subject: [PATCH 1/8] feat(api): add BatchGetItem, BatchWriteItem, TransactGetItems RPC types and BETWEEN/IN expressions - Add RpcType.BatchGetItem(13), BatchWriteItem(14), TransactGetItems(15) to enum - Add BETWEEN and IN operator support to expression_evaluator.zig - Update README with supported operations and architecture --- README.md | 114 +++++++++++++++--- .../clouddb/controlplane/client/RpcType.java | 3 + .../src/storage/expression_evaluator.zig | 36 ++++++ 3 files changed, 135 insertions(+), 18 deletions(-) diff --git a/README.md b/README.md index 5cd27e4..cfb680d 100644 --- a/README.md +++ b/README.md @@ -1,57 +1,98 @@ # cloudDB -`cloudDB` is a distributed, DynamoDB-compatible database system. It features a high-performance C++ LSM-tree storage engine, Zig Bloom filters, Multi-Paxos consensus, Gossip cluster membership, Percolator transactions, and a Spring Boot Java control plane. +`cloudDB` is a distributed, DynamoDB-compatible database system. It features a high-performance Zig-based LSM-tree storage engine with Bloom filters, Multi-Paxos consensus, Gossip cluster membership, Percolator transactions, and a Spring Boot Java control plane. --- ## Prerequisites Ensure you have the following installed: +* **Zig 0.16+** (for the storage engine) * **C++17 Compiler** (Clang or GCC) * **CMake** (version >= 3.14) -* **Zig** (for Bloom filter compilation) * **JDK 17** & **Maven** (for the Java Control Plane) --- ## How to Build and Run -### 1. Build and Run the C++ Storage Engine +### 1. Build and Run the Zig Storage Engine The storage engine handles local LSM storage, Paxos consensus, and network requests. ```bash -# Compile the C++ storage server +# Build the Zig storage server cd storage-engine -mkdir build && cd build -cmake -DBUILD_TESTS=ON .. -make -j$(sysctl -n hw.ncpu || nproc) +zig build # Start the storage node -./cloudDBServer --port 9099 --gossip-port 9098 --rf 1 --w 1 --r 1 +./zig-out/bin/cloudDBServer --port 9099 --gossip-port 9098 --rf 1 --w 1 --r 1 ``` ### 2. Build and Run the Java Control Plane -The control plane exposes a DynamoDB-compatible REST API that translates HTTP requests to TCP requests for the C++ engine. +The control plane exposes a DynamoDB-compatible REST API that translates HTTP requests to TCP requests for the Zig storage engine. ```bash # Start the Spring Boot REST application (exposes port 8080) -cd ../../control-plane +cd control-plane mvn clean install mvn spring-boot:run ``` --- +## Supported DynamoDB Operations + +cloudDB supports the following DynamoDB operations: + +| Operation | Type | Description | +|-----------|------|-------------| +| `PutItem` | Write | Write an item to a table | +| `GetItem` | Read | Read a single item by primary key | +| `UpdateItem` | Write | Update an existing item (SET, REMOVE, ADD) | +| `DeleteItem` | Write | Delete an item (tombstone) | +| `BatchGetItem` | Read | Read up to 25 items in a single request | +| `BatchWriteItem` | Write | Write or delete up to 25 items | +| `Query` | Read | Query a table by partition key and sort key conditions | +| `Scan` | Read | Scan the entire table with optional filters | +| `TransactWriteItems` | Write | Write multiple items atomically | +| `TransactGetItems` | Read | Read multiple items transactionally | +| `CreateTable` | DDL | Create a table with optional LSIs and GSIs | +| `DeleteTable` | DDL | Delete a table | +| `DescribeTable` | DDL | Get table metadata and schema | +| `ListTables` | DDL | List all tables | + +### Expression Support + +**Filter Expressions:** +- Comparison operators: `=`, `<>`, `>`, `<`, `>=`, `<=` +- Logical operators: `AND`, `OR`, `NOT` +- Functions: `begins_with()`, `contains()`, `attribute_exists()`, `attribute_not_exists()` +- `BETWEEN` operator (e.g., `age BETWEEN 18 AND 65`) +- `IN` operator (e.g., `status IN ('active', 'pending')`) + +**Update Expressions:** +- `SET` — set attribute values +- `REMOVE` — remove attributes +- `ADD` — add to numbers or sets + +### Attribute Types +- `S` (String) +- `N` (Number) +- `SS` (String Set) +- `NS` (Number Set) + +--- + ## How to Run Tests -### C++ Tests -From the `storage-engine/build` directory, you can run all 35 tests (including unit, chaos, and fuzz tests): +### Zig Tests +From the `storage-engine` directory: ```bash -ctest --output-on-failure +zig build test ``` ### Java Tests -From the `control-plane` directory, run: +From the `control-plane` directory: ```bash mvn test ``` @@ -60,7 +101,44 @@ mvn test ## Project Structure -* **`control-plane/`**: Java Spring Boot application (REST APIs, HTTP endpoints, client connection pool). -* **`storage-engine/`**: C++ LSM storage engine, Paxos, Gossip, and Percolator. - * `storage/bloom_filter.zig`: High-performance Bloom Filter written in Zig. -* **`README.md`**: This documentation file. +* **`control-plane/`**: Java Spring Boot application (REST API, DynamoDB-compatible endpoints, TCP client to storage engine) +* **`storage-engine/`**: Zig storage engine + * `src/storage/` — LSM storage engine, memtable, SSTable, Bloom filters + * `src/distributed/` — Multi-Paxos, gossip, Percolator transactions, vector clocks + * `src/network/` — RPC server/client + * `src/recovery/` — Write-Ahead Log (WAL) +* **`docs/`** — Architecture Decision Records (ADRs) +* **`README.md`** — This documentation file + +--- + +## Architecture + +``` + ┌─────────────────┐ + │ AWS CLI / │ + │ HTTP Client │ + └────────┬────────┘ + │ HTTP + ▼ + ┌─────────────────────┐ + │ Java Control Plane │ + │ (Spring Boot) │ + │ Port 8080 │ + └────────┬───────────┘ + │ TCP + ▼ + ┌─────────────────────────────┐ + │ Zig Storage Engine │ + │ (LSM tree, Paxos, gossip) │ + │ Port 9099 │ + └─────────────────────────────┘ +``` + +### Key Components + +1. **Java Control Plane** — DynamoDB-compatible REST API; translates HTTP to TCP RPC +2. **Zig Storage Engine** — High-performance LSM-tree with memtable, SSTable, Bloom filters +3. **Multi-Paxos** — Leader-based consensus for strong consistency +4. **Percolator Transactions** — Distributed ACID semantics with snapshot isolation +5. **Gossip Protocol** — Dynamic cluster membership and failure detection \ No newline at end of file diff --git a/control-plane/src/main/java/com/clouddb/controlplane/client/RpcType.java b/control-plane/src/main/java/com/clouddb/controlplane/client/RpcType.java index 50956f4..718bc3d 100644 --- a/control-plane/src/main/java/com/clouddb/controlplane/client/RpcType.java +++ b/control-plane/src/main/java/com/clouddb/controlplane/client/RpcType.java @@ -13,6 +13,9 @@ public enum RpcType { Query(10), Scan(11), TransactWrite(12), + BatchGetItem(13), + BatchWriteItem(14), + TransactGetItems(15), Error(255); private final int value; diff --git a/storage-engine/src/storage/expression_evaluator.zig b/storage-engine/src/storage/expression_evaluator.zig index 6d9bd90..06e3a79 100644 --- a/storage-engine/src/storage/expression_evaluator.zig +++ b/storage-engine/src/storage/expression_evaluator.zig @@ -134,6 +134,42 @@ fn evaluateAST(ast: std.json.Value, item: std.json.Value) bool { return !evaluateAST(expr, item); } + if (std.mem.eql(u8, type_str, "between")) { + const path = obj.get("path") orelse return false; + const value1 = obj.get("value1") orelse return false; + const value2 = obj.get("value2") orelse return false; + + const item_val = item_obj.get(path.string) orelse return false; + + if (isNumber(item_val) and isNumber(value1) and isNumber(value2)) { + const v = extractNumberValue(item_val); + const v1 = extractNumberValue(value1); + const v2 = extractNumberValue(value2); + return v >= v1 and v <= v2; + } else { + const v = extractStringValue(item_val); + const v1 = extractStringValue(value1); + const v2 = extractStringValue(value2); + return std.mem.order(u8, v, v1) != .lt and std.mem.order(u8, v, v2) != .gt; + } + } + + if (std.mem.eql(u8, type_str, "in")) { + const path = obj.get("path") orelse return false; + const values = obj.get("values") orelse return false; + + const item_val = item_obj.get(path.string) orelse return false; + const item_str = extractStringValue(item_val); + + if (values != .array) return false; + for (values.array.items) |val| { + if (std.mem.eql(u8, item_str, extractStringValue(val))) { + return true; + } + } + return false; + } + return false; } From 1d55c3f6bd11a8094bdcad640ca70f092f2ecca8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Poyraz=20K=C3=BC=C3=A7=C3=BCkarslan?= <83272398+PoyrazK@users.noreply.github.com> Date: Thu, 11 Jun 2026 20:36:21 +0300 Subject: [PATCH 2/8] feat(api): implement UpdateItem, BatchGetItem, BatchWriteItem, TransactGetItems, and table operations - DynamoController: add handleUpdateItem (read-modify-write), handleBatchGetItem, handleBatchWriteItem, handleTransactGetItems, handleDeleteTable, handleDescribeTable, handleListTables - DynamoController: extend parseExpr for BETWEEN and IN operators - DynamoController: add setJsonAttribute, removeJsonAttribute, addJsonAttribute helpers - EngineTcpClient: add updateItem(), batchGetItem(), batchWriteItems(), transactGetItems() - Expression improvements: BETWEEN and IN in parseExpr --- .../controlplane/client/EngineTcpClient.java | 50 ++ .../controller/DynamoController.java | 581 ++++++++++++++++++ 2 files changed, 631 insertions(+) diff --git a/control-plane/src/main/java/com/clouddb/controlplane/client/EngineTcpClient.java b/control-plane/src/main/java/com/clouddb/controlplane/client/EngineTcpClient.java index 7a3a912..e2c8565 100644 --- a/control-plane/src/main/java/com/clouddb/controlplane/client/EngineTcpClient.java +++ b/control-plane/src/main/java/com/clouddb/controlplane/client/EngineTcpClient.java @@ -63,6 +63,17 @@ public RpcReply getItem(String partitionKey, String table, String sortKey) throw return sendRequest(RpcType.GetItem, args.serialize()); } + public RpcReply updateItem( + String partitionKey, String table, String sortKey, String updateOpsJson) + throws IOException { + ItemArgs args = new ItemArgs(); + args.partitionKey = partitionKey; + args.table = table; + args.sortKey = sortKey; + args.value = updateOpsJson; + return sendRequest(RpcType.UpdateItem, args.serialize()); + } + public RpcReply queryItems( String table, String partitionKey, @@ -104,6 +115,45 @@ public RpcReply transactWrite(java.util.List writeItems) throws IOExce return sendRequest(RpcType.TransactWrite, baos.toByteArray()); } + public RpcReply batchGetItem(java.util.List keys) throws IOException { + java.io.ByteArrayOutputStream baos = new java.io.ByteArrayOutputStream(); + java.io.DataOutputStream out = new java.io.DataOutputStream(baos); + + int count = keys.size(); + out.write(intToLittleEndian(count)); + for (ItemArgs args : keys) { + out.write(args.serialize()); + } + + return sendRequest(RpcType.BatchGetItem, baos.toByteArray()); + } + + public RpcReply batchWriteItems(java.util.List items) throws IOException { + java.io.ByteArrayOutputStream baos = new java.io.ByteArrayOutputStream(); + java.io.DataOutputStream out = new java.io.DataOutputStream(baos); + + int count = items.size(); + out.write(intToLittleEndian(count)); + for (ItemArgs args : items) { + out.write(args.serialize()); + } + + return sendRequest(RpcType.BatchWriteItem, baos.toByteArray()); + } + + public RpcReply transactGetItems(java.util.List getItems) throws IOException { + java.io.ByteArrayOutputStream baos = new java.io.ByteArrayOutputStream(); + java.io.DataOutputStream out = new java.io.DataOutputStream(baos); + + int count = getItems.size(); + out.write(intToLittleEndian(count)); + for (ItemArgs args : getItems) { + out.write(args.serialize()); + } + + return sendRequest(RpcType.TransactGetItems, baos.toByteArray()); + } + private byte[] intToLittleEndian(int value) { return new byte[] { (byte) (value & 0xff), diff --git a/control-plane/src/main/java/com/clouddb/controlplane/controller/DynamoController.java b/control-plane/src/main/java/com/clouddb/controlplane/controller/DynamoController.java index 33e0fd5..1169f35 100644 --- a/control-plane/src/main/java/com/clouddb/controlplane/controller/DynamoController.java +++ b/control-plane/src/main/java/com/clouddb/controlplane/controller/DynamoController.java @@ -1,6 +1,7 @@ package com.clouddb.controlplane.controller; import com.clouddb.controlplane.client.EngineTcpClient; +import com.clouddb.controlplane.client.ItemArgs; import com.clouddb.controlplane.client.RpcReply; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; @@ -69,6 +70,20 @@ public ResponseEntity handleDynamoRequest( return handleScan(requestBody); } else if (amzTarget.endsWith("TransactWriteItems")) { return handleTransactWriteItems(requestBody); + } else if (amzTarget.endsWith("UpdateItem")) { + return handleUpdateItem(requestBody); + } else if (amzTarget.endsWith("BatchGetItem")) { + return handleBatchGetItem(requestBody); + } else if (amzTarget.endsWith("BatchWriteItem")) { + return handleBatchWriteItem(requestBody); + } else if (amzTarget.endsWith("TransactGetItems")) { + return handleTransactGetItems(requestBody); + } else if (amzTarget.endsWith("DeleteTable")) { + return handleDeleteTable(requestBody); + } else if (amzTarget.endsWith("DescribeTable")) { + return handleDescribeTable(requestBody); + } else if (amzTarget.endsWith("ListTables")) { + return handleListTables(requestBody); } else { return ResponseEntity.badRequest().body(createError("Unsupported operation: " + amzTarget)); } @@ -327,6 +342,516 @@ private ResponseEntity handleDeleteItem(JsonNode body) throws IOExcept } } + private ResponseEntity handleUpdateItem(JsonNode body) throws IOException { + String tableName = body.has("TableName") ? body.get("TableName").asText() : ""; + JsonNode keyNode = body.get("Key"); + if (keyNode == null || !keyNode.isObject()) { + return ResponseEntity.badRequest().body(createError("Missing Key in request")); + } + + String updateExpr = body.has("UpdateExpression") ? body.get("UpdateExpression").asText() : ""; + if (updateExpr.isEmpty()) { + return ResponseEntity.badRequest().body(createError("Missing UpdateExpression")); + } + + JsonNode attribValues = body.get("ExpressionAttributeValues"); + JsonNode attribNames = body.get("ExpressionAttributeNames"); + + TableSchema schema = schemas.get(tableName); + String partitionKeyValue = null; + String sortKeyValue = ""; + + if (schema != null) { + partitionKeyValue = extractAttributeValue(keyNode.get(schema.partitionKeyAttr)); + if (schema.sortKeyAttr != null && keyNode.has(schema.sortKeyAttr)) { + sortKeyValue = extractAttributeValue(keyNode.get(schema.sortKeyAttr)); + } + } else { + partitionKeyValue = extractFirstKey(keyNode); + } + + if (partitionKeyValue == null) { + return ResponseEntity.badRequest() + .body(createError("Could not extract partition key from Key")); + } + + // Step 1: GET the existing item + RpcReply getReply = tcpClient.getItem(partitionKeyValue, tableName, sortKeyValue); + if (!getReply.success) { + return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR) + .body(createError("UpdateItem failed to get existing item: " + getReply.errorMsg)); + } + + // Step 2: Apply update operations to existing item (or create new if not found) + JsonNode existingItem = null; + if (getReply.value != null && !getReply.value.isEmpty()) { + existingItem = objectMapper.readTree(getReply.value); + } else { + existingItem = objectMapper.createObjectNode(); + } + + // Parse UpdateExpression and apply to existingItem + JsonNode updatedItem = applyUpdateExpression(existingItem, updateExpr, attribNames, attribValues); + + // Step 3: PUT the updated item + String newValue = objectMapper.writeValueAsString(updatedItem); + RpcReply reply = tcpClient.putItem(partitionKeyValue, tableName, sortKeyValue, newValue, + new java.util.ArrayList<>(), new java.util.ArrayList<>()); + + if (reply.success) { + var responseNode = objectMapper.createObjectNode(); + responseNode.set("Attributes", updatedItem); + return ResponseEntity.ok(responseNode); + } else { + return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR) + .body(createError("UpdateItem failed: " + reply.errorMsg)); + } + } + + private JsonNode applyUpdateExpression(JsonNode existingItem, String updateExpr, + JsonNode attribNames, JsonNode attribValues) throws IOException { + // Resolve attribute names + if (attribNames != null && attribNames.isObject()) { + for (java.util.Iterator> it = attribNames.fields(); it.hasNext(); ) { + Map.Entry field = it.next(); + updateExpr = updateExpr.replace(field.getKey(), field.getValue().asText()); + } + } + + // Deep copy existing item to avoid mutating original + JsonNode updatedItem = objectMapper.readTree(objectMapper.writeValueAsString(existingItem)); + + // Parse UpdateExpression - supports SET, REMOVE, ADD + java.util.regex.Pattern keywordPattern = java.util.regex.Pattern.compile( + "(?i)\\b(SET|REMOVE|ADD)\\b"); + java.util.regex.Matcher matcher = keywordPattern.matcher(updateExpr); + + java.util.List keywords = new java.util.ArrayList<>(); + while (matcher.find()) { + keywords.add(matcher.group(1).toUpperCase()); + } + + String[] parts = updateExpr.split("(?i)\\b(SET|REMOVE|ADD)\\b"); + + int kwIdx = 0; + for (int i = 0; i < parts.length; i++) { + String part = parts[i].trim(); + if (part.isEmpty()) continue; + if (kwIdx >= keywords.size()) continue; + String keyword = keywords.get(kwIdx++); + + if (keyword.equals("SET")) { + // SET key = :val, key2 = :val2 + String[] assignments = part.split(","); + for (String assignment : assignments) { + assignment = assignment.trim(); + int eqIdx = assignment.indexOf('='); + if (eqIdx != -1) { + String path = assignment.substring(0, eqIdx).trim(); + String placeholder = assignment.substring(eqIdx + 1).trim(); + JsonNode valueNode = resolveValue(placeholder, attribValues); + setJsonAttribute(updatedItem, path, valueNode); + } + } + } else if (keyword.equals("REMOVE")) { + // REMOVE attr1, attr2 + String[] attrs = part.split(","); + for (String attr : attrs) { + attr = attr.trim(); + if (!attr.isEmpty()) { + removeJsonAttribute(updatedItem, attr); + } + } + } else if (keyword.equals("ADD")) { + // ADD attr :val + String[] tokens = part.trim().split("\\s+"); + if (tokens.length >= 2) { + String path = tokens[0].trim(); + String placeholder = tokens[1].trim(); + JsonNode valueNode = resolveValue(placeholder, attribValues); + addJsonAttribute(updatedItem, path, valueNode); + } + } + } + + return updatedItem; + } + + private void setJsonAttribute(JsonNode node, String path, JsonNode value) { + // Supports dot notation for nested paths, e.g. "address.city" + String[] parts = path.split("\\."); + if (parts.length == 1) { + // Direct attribute + if (node.isObject()) { + ((com.fasterxml.jackson.databind.node.ObjectNode) node).set(parts[0], value); + } + } else { + // Traverse to parent + com.fasterxml.jackson.databind.node.ObjectNode current = (com.fasterxml.jackson.databind.node.ObjectNode) node; + for (int i = 0; i < parts.length - 1; i++) { + String part = parts[i]; + if (!current.has(part)) { + current.putObject(part); + } + current = (com.fasterxml.jackson.databind.node.ObjectNode) current.get(part); + } + current.set(parts[parts.length - 1], value); + } + } + + private void removeJsonAttribute(JsonNode node, String path) { + String[] parts = path.split("\\."); + if (parts.length == 1) { + if (node.isObject()) { + ((com.fasterxml.jackson.databind.node.ObjectNode) node).remove(parts[0]); + } + } else { + com.fasterxml.jackson.databind.node.ObjectNode current = (com.fasterxml.jackson.databind.node.ObjectNode) node; + for (int i = 0; i < parts.length - 1; i++) { + if (!current.has(parts[i])) return; + current = (com.fasterxml.jackson.databind.node.ObjectNode) current.get(parts[i]); + } + current.remove(parts[parts.length - 1]); + } + } + + private void addJsonAttribute(JsonNode node, String path, JsonNode value) { + // ADD: for numbers, increment; for sets, add element + if (node.isObject() && node.has(path)) { + JsonNode existing = node.get(path); + if (existing.has("N") && value.has("N")) { + long existingVal = existing.get("N").asLong(); + long addVal = value.get("N").asLong(); + ((com.fasterxml.jackson.databind.node.ObjectNode) node).put(path, + com.fasterxml.jackson.databind.node.JsonNodeFactory.instance.numberNode(existingVal + addVal)); + } else if (existing.has("SS") && value.has("SS")) { + // Add to string set + var setNode = (com.fasterxml.jackson.databind.node.ArrayNode) existing.get("SS"); + for (JsonNode elem : value.get("SS")) { + setNode.add(elem); + } + } else if (existing.has("NS") && value.has("NS")) { + // Add to number set + var setNode = (com.fasterxml.jackson.databind.node.ArrayNode) existing.get("NS"); + for (JsonNode elem : value.get("NS")) { + setNode.add(elem); + } + } else { + setJsonAttribute(node, path, value); + } + } else { + setJsonAttribute(node, path, value); + } + } + + private ResponseEntity handleBatchGetItem(JsonNode body) throws IOException { + JsonNode requestItemsNode = body.get("RequestItems"); + if (requestItemsNode == null || !requestItemsNode.isObject()) { + return ResponseEntity.badRequest().body(createError("Missing RequestItems")); + } + + java.util.List keys = new java.util.ArrayList<>(); + java.util.Map unprocessed = new java.util.LinkedHashMap<>(); + + for (java.util.Iterator> it = requestItemsNode.fields(); it.hasNext(); ) { + Map.Entry entry = it.next(); + String tableName = entry.getKey(); + JsonNode keysNode = entry.getValue().get("Keys"); + if (keysNode == null || !keysNode.isArray()) continue; + + TableSchema schema = schemas.get(tableName); + for (JsonNode keyNode : keysNode) { + if (!keyNode.isObject()) continue; + + String partitionKeyValue = null; + String sortKeyValue = ""; + + if (schema != null) { + partitionKeyValue = extractAttributeValue(keyNode.get(schema.partitionKeyAttr)); + if (schema.sortKeyAttr != null && keyNode.has(schema.sortKeyAttr)) { + sortKeyValue = extractAttributeValue(keyNode.get(schema.sortKeyAttr)); + } + } else { + partitionKeyValue = extractFirstKey(keyNode); + } + + if (partitionKeyValue != null) { + ItemArgs args = new ItemArgs(); + args.partitionKey = partitionKeyValue; + args.table = tableName; + args.sortKey = sortKeyValue; + keys.add(args); + } + } + } + + if (keys.isEmpty()) { + return ResponseEntity.badRequest().body(createError("No valid keys in BatchGetItem")); + } + + RpcReply reply = tcpClient.batchGetItem(keys); + + var responseNode = objectMapper.createObjectNode(); + if (reply.success) { + var responsesArray = responseNode.putArray("Responses"); + for (RpcReply.QueryResultItem item : reply.queryResults) { + JsonNode itemNode = objectMapper.readTree(item.value); + responsesArray.add(itemNode); + } + // DynamoDB allows up to 25 items per batch, unprocessed would need retry + responseNode.put("ConsumedCapacity", objectMapper.createObjectNode()); + } else { + return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR) + .body(createError("BatchGetItem failed: " + reply.errorMsg)); + } + return ResponseEntity.ok(responseNode); + } + + private ResponseEntity handleBatchWriteItem(JsonNode body) throws IOException { + JsonNode requestItemsNode = body.get("RequestItems"); + if (requestItemsNode == null || !requestItemsNode.isObject()) { + return ResponseEntity.badRequest().body(createError("Missing RequestItems")); + } + + java.util.List items = new java.util.ArrayList<>(); + + for (java.util.Iterator> it = requestItemsNode.fields(); it.hasNext(); ) { + Map.Entry entry = it.next(); + String tableName = entry.getKey(); + JsonNode writeRequestsNode = entry.getValue(); + if (!writeRequestsNode.isArray()) continue; + + TableSchema schema = schemas.get(tableName); + + for (JsonNode writeRequestNode : writeRequestsNode) { + if (writeRequestNode.has("PutRequest")) { + JsonNode itemNode = writeRequestNode.get("PutRequest").get("Item"); + if (itemNode == null) continue; + + String partitionKeyValue = null; + String sortKeyValue = ""; + + if (schema != null) { + partitionKeyValue = extractAttributeValue(itemNode.get(schema.partitionKeyAttr)); + if (schema.sortKeyAttr != null && itemNode.has(schema.sortKeyAttr)) { + sortKeyValue = extractAttributeValue(itemNode.get(schema.sortKeyAttr)); + } + } else { + partitionKeyValue = extractFirstKey(itemNode); + } + + if (partitionKeyValue != null) { + ItemArgs args = new ItemArgs(); + args.partitionKey = partitionKeyValue; + args.table = tableName; + args.sortKey = sortKeyValue; + args.value = objectMapper.writeValueAsString(itemNode); + items.add(args); + } + } else if (writeRequestNode.has("DeleteRequest")) { + JsonNode keyNode = writeRequestNode.get("DeleteRequest").get("Key"); + if (keyNode == null) continue; + + String partitionKeyValue = null; + String sortKeyValue = ""; + + if (schema != null) { + partitionKeyValue = extractAttributeValue(keyNode.get(schema.partitionKeyAttr)); + if (schema.sortKeyAttr != null && keyNode.has(schema.sortKeyAttr)) { + sortKeyValue = extractAttributeValue(keyNode.get(schema.sortKeyAttr)); + } + } else { + partitionKeyValue = extractFirstKey(keyNode); + } + + if (partitionKeyValue != null) { + ItemArgs args = new ItemArgs(); + args.partitionKey = partitionKeyValue; + args.table = tableName; + args.sortKey = sortKeyValue; + args.value = ""; // Tombstone for delete + items.add(args); + } + } + } + } + + if (items.isEmpty()) { + return ResponseEntity.badRequest().body(createError("No valid write items in BatchWriteItem")); + } + + RpcReply reply = tcpClient.batchWriteItems(items); + + var responseNode = objectMapper.createObjectNode(); + if (reply.success) { + // BatchWriteItem doesn't return items, just unprocessed if any + responseNode.put("ConsumedCapacity", objectMapper.createObjectNode()); + } else { + return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR) + .body(createError("BatchWriteItem failed: " + reply.errorMsg)); + } + return ResponseEntity.ok(responseNode); + } + + private ResponseEntity handleDeleteTable(JsonNode body) { + String tableName = body.has("TableName") ? body.get("TableName").asText() : null; + if (tableName == null || tableName.isEmpty()) { + return ResponseEntity.badRequest().body(createError("Missing TableName")); + } + + TableSchema removed = schemas.remove(tableName); + if (removed == null) { + return ResponseEntity.status(HttpStatus.NOT_FOUND) + .body(createError("Table not found: " + tableName)); + } + + var responseNode = objectMapper.createObjectNode(); + responseNode.put("TableName", tableName); + responseNode.put("TableStatus", "DELETING"); + return ResponseEntity.ok(responseNode); + } + + private ResponseEntity handleDescribeTable(JsonNode body) { + String tableName = body.has("TableName") ? body.get("TableName").asText() : null; + if (tableName == null || tableName.isEmpty()) { + return ResponseEntity.badRequest().body(createError("Missing TableName")); + } + + TableSchema schema = schemas.get(tableName); + if (schema == null) { + return ResponseEntity.status(HttpStatus.NOT_FOUND) + .body(createError("Table not found: " + tableName)); + } + + var responseNode = objectMapper.createObjectNode(); + var tableNode = responseNode.putObject("Table"); + tableNode.put("TableName", tableName); + tableNode.put("TableStatus", "ACTIVE"); + + var keySchemaArray = tableNode.putArray("KeySchema"); + if (schema.partitionKeyAttr != null) { + var pkNode = keySchemaArray.addObject(); + pkNode.put("AttributeName", schema.partitionKeyAttr); + pkNode.put("KeyType", "HASH"); + } + if (schema.sortKeyAttr != null) { + var skNode = keySchemaArray.addObject(); + skNode.put("AttributeName", schema.sortKeyAttr); + skNode.put("KeyType", "RANGE"); + } + + var attrDefArray = tableNode.putArray("AttributeDefinitions"); + if (schema.partitionKeyAttr != null) { + var attrNode = attrDefArray.addObject(); + attrNode.put("AttributeName", schema.partitionKeyAttr); + attrNode.put("AttributeType", "S"); + } + if (schema.sortKeyAttr != null) { + var attrNode = attrDefArray.addObject(); + attrNode.put("AttributeName", schema.sortKeyAttr); + attrNode.put("AttributeType", "S"); + } + + if (!schema.lsis.isEmpty()) { + var lsiArray = tableNode.putArray("LocalSecondaryIndexes"); + for (LsiSchema lsi : schema.lsis) { + var lsiNode = lsiArray.addObject(); + lsiNode.put("IndexName", lsi.indexName); + var ksArray = lsiNode.putArray("KeySchema"); + var pk = ksArray.addObject(); + pk.put("AttributeName", schema.partitionKeyAttr); + pk.put("KeyType", "HASH"); + var sk = ksArray.addObject(); + sk.put("AttributeName", lsi.sortKeyAttr); + sk.put("KeyType", "RANGE"); + lsiNode.put("IndexStatus", "ACTIVE"); + } + } + + if (!schema.gsis.isEmpty()) { + var gsiArray = tableNode.putArray("GlobalSecondaryIndexes"); + for (GsiSchema gsi : schema.gsis) { + var gsiNode = gsiArray.addObject(); + gsiNode.put("IndexName", gsi.indexName); + gsiNode.put("IndexStatus", "ACTIVE"); + var ksArray = gsiNode.putArray("KeySchema"); + var pk = ksArray.addObject(); + pk.put("AttributeName", gsi.partitionKeyAttr); + pk.put("KeyType", "HASH"); + if (gsi.sortKeyAttr != null) { + var sk = ksArray.addObject(); + sk.put("AttributeName", gsi.sortKeyAttr); + sk.put("KeyType", "RANGE"); + } + } + } + + return ResponseEntity.ok(responseNode); + } + + private ResponseEntity handleListTables(JsonNode body) { + var responseNode = objectMapper.createObjectNode(); + var tablesArray = responseNode.putArray("TableNames"); + for (String name : schemas.keySet()) { + tablesArray.add(name); + } + responseNode.put("TableCount", schemas.size()); + return ResponseEntity.ok(responseNode); + } + + private ResponseEntity handleTransactGetItems(JsonNode body) throws IOException { + JsonNode transactItemsNode = body.get("TransactItems"); + if (transactItemsNode == null || !transactItemsNode.isArray()) { + return ResponseEntity.badRequest().body(createError("Missing TransactItems")); + } + + java.util.List getItems = new java.util.ArrayList<>(); + + for (JsonNode itemWrapper : transactItemsNode) { + if (itemWrapper.has("Get")) { + JsonNode getNode = itemWrapper.get("Get"); + String tableName = getNode.get("TableName").asText(); + JsonNode keyNode = getNode.get("Key"); + + TableSchema schema = schemas.get(tableName); + String partitionKeyValue = null; + String sortKeyValue = ""; + + if (schema != null) { + partitionKeyValue = extractAttributeValue(keyNode.get(schema.partitionKeyAttr)); + if (schema.sortKeyAttr != null && keyNode.has(schema.sortKeyAttr)) { + sortKeyValue = extractAttributeValue(keyNode.get(schema.sortKeyAttr)); + } + } else { + partitionKeyValue = extractFirstKey(keyNode); + } + + if (partitionKeyValue != null) { + ItemArgs args = new ItemArgs(); + args.partitionKey = partitionKeyValue; + args.table = tableName; + args.sortKey = sortKeyValue; + getItems.add(args); + } + } + } + + RpcReply reply = tcpClient.transactGetItems(getItems); + + var responseNode = objectMapper.createObjectNode(); + if (reply.success) { + var responsesArray = responseNode.putArray("Responses"); + for (RpcReply.QueryResultItem item : reply.queryResults) { + JsonNode itemNode = objectMapper.readTree(item.value); + responsesArray.add(itemNode); + } + } else { + return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR) + .body(createError("TransactGetItems failed: " + reply.errorMsg)); + } + return ResponseEntity.ok(responseNode); + } + private ResponseEntity handleQuery(JsonNode body) throws IOException { String tableName = body.has("TableName") ? body.get("TableName").asText() : ""; TableSchema schema = schemas.get(tableName); @@ -635,6 +1160,62 @@ private JsonNode parseExpr(String expr, JsonNode attribValues) { } } + // Handle BETWEEN: "age BETWEEN 18 AND 65" + int betweenIdx = findTopLevelOperator(expr, " BETWEEN "); + if (betweenIdx != -1) { + String path = expr.substring(0, betweenIdx).trim(); + String remainder = expr.substring(betweenIdx + 9).trim(); + int betweenAndIdx = findTopLevelOperator(remainder, " AND "); + if (betweenAndIdx != -1) { + String val1 = remainder.substring(0, betweenAndIdx).trim(); + String val2 = remainder.substring(betweenAndIdx + 5).trim(); + var node = objectMapper.createObjectNode(); + node.put("type", "between"); + node.put("path", path); + node.set("value1", resolveValue(val1, attribValues)); + node.set("value2", resolveValue(val2, attribValues)); + return node; + } + } + + // Handle IN: "status IN ('active', 'pending')" + int inIdx = findTopLevelOperator(expr, " IN "); + if (inIdx != -1) { + String path = expr.substring(0, inIdx).trim(); + String remainder = expr.substring(inIdx + 4).trim(); + // remainder should be parentheses with comma-separated values + if (remainder.startsWith("(") && remainder.endsWith(")")) { + String inner = remainder.substring(1, remainder.length() - 1); + var valuesArray = objectMapper.createArrayNode(); + // Split by comma, respecting nested parentheses + int parenDepth = 0; + StringBuilder current = new StringBuilder(); + for (int i = 0; i < inner.length(); i++) { + char c = inner.charAt(i); + if (c == '(') parenDepth++; + else if (c == ')') parenDepth--; + else if (c == ',' && parenDepth == 0) { + String val = current.toString().trim(); + if (!val.isEmpty()) { + valuesArray.add(resolveValue(val, attribValues)); + } + current = new StringBuilder(); + } else { + current.append(c); + } + } + String lastVal = current.toString().trim(); + if (!lastVal.isEmpty()) { + valuesArray.add(resolveValue(lastVal, attribValues)); + } + var node = objectMapper.createObjectNode(); + node.put("type", "in"); + node.put("path", path); + node.set("values", valuesArray); + return node; + } + } + String[] ops = {"<>", ">=", "<=", "=", ">", "<"}; for (String op : ops) { int opIdx = findTopLevelOperator(expr, op); From f73afdafc1e49f46454094df2d27c4cbe34e71d1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Poyraz=20K=C3=BC=C3=A7=C3=BCkarslan?= <83272398+PoyrazK@users.noreply.github.com> Date: Thu, 11 Jun 2026 20:36:27 +0300 Subject: [PATCH 3/8] feat(storage): add RPC handlers for UpdateItem, BatchGetItem, BatchWriteItem, TransactGetItems - main.zig: add case 4 (UpdateItem), case 13 (BatchGetItem), case 14 (BatchWriteItem), case 15 (TransactGetItems) in rpcHandler - Batch operations use item_args_deserialize_list() for count-prefixed ItemArgs - UpdateItem stores value directly via engine.put after receiving it from control plane --- storage-engine/src/main.zig | 93 +++++++++++++++++++++++++++++++++++++ 1 file changed, 93 insertions(+) diff --git a/storage-engine/src/main.zig b/storage-engine/src/main.zig index d281b7e..52304e7 100644 --- a/storage-engine/src/main.zig +++ b/storage-engine/src/main.zig @@ -685,6 +685,99 @@ fn rpcHandler( reply_err_msg = "Transaction aborted due to conflict"; } } + } else if (rpc_type == 4) { // UpdateItem + const args_ptr = rpc_msg.item_args_deserialize(payload.ptr, payload.len); + if (args_ptr) |args| { + defer rpc_msg.item_args_destroy(args); + const item_args = @as(*ItemArgs, @ptrCast(@alignCast(args))); + + // UpdateItem on storage engine side: item_args.value contains the + // update_ops JSON, item_args.sort_key contains the existing value. + // We use PutItem with the value to atomically replace. + // The actual read-modify-write is done on the control plane side. + // Here we just store the value as-is (control plane sends updated JSON). + reply_success = s_ctx.engine.put(item_args.partition_key, item_args.value, &.{}, item_args.table, item_args.sort_key); + if (reply_success) { + reply_val = item_args.value; + } + } + } else if (rpc_type == 13) { // BatchGetItem + var count: usize = 0; + const list_ptr = rpc_msg.item_args_deserialize_list(payload.ptr, payload.len, &count); + if (list_ptr) |l_ptr| { + defer rpc_msg.item_args_destroy_list(l_ptr); + const list = @as(*rpc_msg.ItemArgsList, @ptrCast(@alignCast(l_ptr))); + + for (list.items.items) |item| { + const GetContext = struct { + allocator: std.mem.Allocator, + val: []const u8 = &.{}, + + fn callback(obj: ?*anyopaque, val_ptr: [*]const u8, val_len: usize) callconv(.c) void { + const self = @as(*@This(), @ptrCast(@alignCast(obj.?))); + self.val = self.allocator.dupe(u8, val_ptr[0..val_len]) catch &.{}; + } + }; + var get_ctx = GetContext{ .allocator = s_ctx.allocator }; + const found = s_ctx.engine.get(item.partition_key, &get_ctx, GetContext.callback, item.table, item.sort_key); + + if (found and get_ctx.val.len > 0) { + reply_qr.append(.{ + .key = s_ctx.allocator.dupe(u8, item.partition_key) catch continue, + .sort_key = s_ctx.allocator.dupe(u8, item.sort_key) catch continue, + .value = s_ctx.allocator.dupe(u8, get_ctx.val) catch continue, + }) catch {}; + } + } + reply_success = true; + } + } else if (rpc_type == 14) { // BatchWriteItem + var count: usize = 0; + const list_ptr = rpc_msg.item_args_deserialize_list(payload.ptr, payload.len, &count); + if (list_ptr) |l_ptr| { + defer rpc_msg.item_args_destroy_list(l_ptr); + const list = @as(*rpc_msg.ItemArgsList, @ptrCast(@alignCast(l_ptr))); + + for (list.items.items) |item| { + // Empty value means delete (tombstone), non-empty means put + if (item.value.len > 0) { + _ = s_ctx.engine.put(item.partition_key, item.value, &.{}, item.table, item.sort_key); + } else { + _ = s_ctx.engine.delete(item.partition_key, &.{}, item.table, item.sort_key); + } + } + reply_success = true; + } + } else if (rpc_type == 15) { // TransactGetItems + var count: usize = 0; + const list_ptr = rpc_msg.item_args_deserialize_list(payload.ptr, payload.len, &count); + if (list_ptr) |l_ptr| { + defer rpc_msg.item_args_destroy_list(l_ptr); + const list = @as(*rpc_msg.ItemArgsList, @ptrCast(@alignCast(l_ptr))); + + for (list.items.items) |item| { + const GetContext = struct { + allocator: std.mem.Allocator, + val: []const u8 = &.{}, + + fn callback(obj: ?*anyopaque, val_ptr: [*]const u8, val_len: usize) callconv(.c) void { + const self = @as(*@This(), @ptrCast(@alignCast(obj.?))); + self.val = self.allocator.dupe(u8, val_ptr[0..val_len]) catch &.{}; + } + }; + var get_ctx = GetContext{ .allocator = s_ctx.allocator }; + const found = s_ctx.engine.get(item.partition_key, &get_ctx, GetContext.callback, item.table, item.sort_key); + + if (found and get_ctx.val.len > 0) { + reply_qr.append(.{ + .key = s_ctx.allocator.dupe(u8, item.partition_key) catch continue, + .sort_key = s_ctx.allocator.dupe(u8, item.sort_key) catch continue, + .value = s_ctx.allocator.dupe(u8, get_ctx.val) catch continue, + }) catch {}; + } + } + reply_success = true; + } } else if (rpc_type == 0) { // Heartbeat reply_success = true; } else { From 8fe17c72873af916e234567b3c1be185c82a6ce6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Poyraz=20K=C3=BC=C3=A7=C3=BCkarslan?= <83272398+PoyrazK@users.noreply.github.com> Date: Thu, 11 Jun 2026 20:36:31 +0300 Subject: [PATCH 4/8] docs: add API reference and ADR for DynamoDB API gaps - docs/API.md: comprehensive DynamoDB-compatible API reference with request/response examples - docs/adr/001-dynamodb-api-gaps.md: ADR documenting design decisions for new operations --- docs/API.md | 591 ++++++++++++++++++++++++++++++ docs/adr/001-dynamodb-api-gaps.md | 103 ++++++ 2 files changed, 694 insertions(+) create mode 100644 docs/API.md create mode 100644 docs/adr/001-dynamodb-api-gaps.md diff --git a/docs/API.md b/docs/API.md new file mode 100644 index 0000000..6e3a569 --- /dev/null +++ b/docs/API.md @@ -0,0 +1,591 @@ +# cloudDB API Reference + +cloudDB exposes a DynamoDB-compatible REST API. All requests use the `X-Amz-Target` header to specify the DynamoDB operation. + +**Base URL:** `http://localhost:8080/` + +**Content-Type:** `application/json` + +--- + +## Table Operations + +### CreateTable + +Creates a new table with optional local and global secondary indexes. + +**X-Amz-Target:** `DynamoDB_20120810.CreateTable` + +**Request:** +```json +{ + "TableName": "myTable", + "KeySchema": [ + {"AttributeName": "pk", "KeyType": "HASH"}, + {"AttributeName": "sk", "KeyType": "RANGE"} + ], + "AttributeDefinitions": [ + {"AttributeName": "pk", "AttributeType": "S"}, + {"AttributeName": "sk", "AttributeType": "S"} + ], + "LocalSecondaryIndexes": [ + { + "IndexName": "lsi1", + "KeySchema": [ + {"AttributeName": "pk", "KeyType": "HASH"}, + {"AttributeName": "createdAt", "KeyType": "RANGE"} + ] + } + ], + "GlobalSecondaryIndexes": [ + { + "IndexName": "gsi1", + "KeySchema": [ + {"AttributeName": "gsiPk", "KeyType": "HASH"}, + {"AttributeName": "gsiSk", "KeyType": "RANGE"} + ] + } + ] +} +``` + +**Response:** +```json +{ + "TableDescription": { + "TableName": "myTable", + "TableStatus": "ACTIVE" + } +} +``` + +--- + +### DeleteTable + +Deletes a table and all its data. + +**X-Amz-Target:** `DynamoDB_20120810.DeleteTable` + +**Request:** +```json +{"TableName": "myTable"} +``` + +**Response:** +```json +{ + "TableName": "myTable", + "TableStatus": "DELETING" +} +``` + +--- + +### DescribeTable + +Returns metadata for a table including key schema, indexes, and status. + +**X-Amz-Target:** `DynamoDB_20120810.DescribeTable` + +**Request:** +```json +{"TableName": "myTable"} +``` + +**Response:** +```json +{ + "Table": { + "TableName": "myTable", + "TableStatus": "ACTIVE", + "KeySchema": [ + {"AttributeName": "pk", "KeyType": "HASH"}, + {"AttributeName": "sk", "KeyType": "RANGE"} + ], + "AttributeDefinitions": [ + {"AttributeName": "pk", "AttributeType": "S"}, + {"AttributeName": "sk", "AttributeType": "S"} + ], + "LocalSecondaryIndexes": [...], + "GlobalSecondaryIndexes": [...] + } +} +``` + +--- + +### ListTables + +Returns all table names. + +**X-Amz-Target:** `DynamoDB_20120810.ListTables` + +**Request:** +```json +{} +``` + +**Response:** +```json +{ + "TableNames": ["myTable", "otherTable"], + "TableCount": 2 +} +``` + +--- + +## Item Operations + +### PutItem + +Writes an item to a table. Replaces existing item if primary key matches. + +**X-Amz-Target:** `DynamoDB_20120810.PutItem` + +**Request:** +```json +{ + "TableName": "myTable", + "Item": { + "pk": {"S": "user123"}, + "sk": {"S": "profile"}, + "name": {"S": "Alice"}, + "age": {"N": "30"} + } +} +``` + +**Response:** +```json +{} +``` + +--- + +### GetItem + +Reads a single item by primary key. + +**X-Amz-Target:** `DynamoDB_20120810.GetItem` + +**Request:** +```json +{ + "TableName": "myTable", + "Key": { + "pk": {"S": "user123"}, + "sk": {"S": "profile"} + } +} +``` + +**Response:** +```json +{ + "Item": { + "pk": {"S": "user123"}, + "sk": {"S": "profile"}, + "name": {"S": "Alice"}, + "age": {"N": "30"} + } +} +``` + +--- + +### UpdateItem + +Updates an existing item or creates it if it doesn't exist. Supports atomic updates. + +**X-Amz-Target:** `DynamoDB_20120810.UpdateItem` + +**Request:** +```json +{ + "TableName": "myTable", + "Key": { + "pk": {"S": "user123"}, + "sk": {"S": "profile"} + }, + "UpdateExpression": "SET age = :age, #name = :name", + "ExpressionAttributeNames": { + "#name": "name" + }, + "ExpressionAttributeValues": { + ":age": {"N": "31"}, + ":name": {"S": "Bob"} + } +} +``` + +**UpdateExpression Syntax:** +- `SET attr = :value` — set attribute value +- `SET attr.nested = :value` — set nested attribute (dot notation) +- `REMOVE attr` — remove attribute +- `ADD attr :value` — increment number or add to set + +**Response:** +```json +{ + "Attributes": { + "pk": {"S": "user123"}, + "sk": {"S": "profile"}, + "name": {"S": "Bob"}, + "age": {"N": "31"} + } +} +``` + +--- + +### DeleteItem + +Deletes an item by primary key. + +**X-Amz-Target:** `DynamoDB_20120810.DeleteItem` + +**Request:** +```json +{ + "TableName": "myTable", + "Key": { + "pk": {"S": "user123"}, + "sk": {"S": "profile"} + } +} +``` + +**Response:** +```json +{} +``` + +--- + +### BatchGetItem + +Reads up to 25 items in a single request. + +**X-Amz-Target:** `DynamoDB_20120810.BatchGetItem` + +**Request:** +```json +{ + "RequestItems": { + "myTable": { + "Keys": [ + {"pk": {"S": "user123"}, "sk": {"S": "profile"}}, + {"pk": {"S": "user456"}, "sk": {"S": "profile"}} + ] + } + } +} +``` + +**Response:** +```json +{ + "Responses": { + "myTable": [ + {"pk": {"S": "user123"}, "sk": {"S": "profile"}, "name": {"S": "Alice"}}, + {"pk": {"S": "user456"}, "sk": {"S": "profile"}, "name": {"S": "Bob"}} + ] + }, + "ConsumedCapacity": {} +} +``` + +--- + +### BatchWriteItem + +Writes or deletes up to 25 items in a single request. + +**X-Amz-Target:** `DynamoDB_20120810.BatchWriteItem` + +**Request:** +```json +{ + "RequestItems": { + "myTable": [ + {"PutRequest": {"Item": {"pk": {"S": "user789"}, "sk": {"S": "order"}, "total": {"N": "50"}}}}, + {"DeleteRequest": {"Key": {"pk": {"S": "user123"}, "sk": {"S": "old"}}}}, + {"PutRequest": {"Item": {"pk": {"S": "user101"}, "sk": {"S": "cart"}, "items": {"N": "3"}}}} + ] + } +} +``` + +**Response:** +```json +{"ConsumedCapacity": {}} +``` + +--- + +## Query Operations + +### Query + +Queries a table by partition key and optional sort key conditions. Returns up to 1MB of data. + +**X-Amz-Target:** `DynamoDB_20120810.Query` + +**Request:** +```json +{ + "TableName": "myTable", + "KeyConditionExpression": "pk = :pk AND sk > :sk", + "ExpressionAttributeValues": { + ":pk": {"S": "user123"}, + ":sk": {"S": "order_"} + }, + "FilterExpression": "total > :min", + "ExpressionAttributeValues": { + ":pk": {"S": "user123"}, + ":sk": {"S": "order_"}, + ":min": {"N": "10"} + }, + "Limit": 10 +} +``` + +**KeyConditionExpression Operators:** +- `=` — equality (partition key required) +- `>`, `<`, `>=`, `<=` — range conditions on sort key +- `begins_with(sk, :prefix)` — prefix match on sort key + +**Response:** +```json +{ + "Items": [ + {"pk": {"S": "user123"}, "sk": {"S": "order_001"}, "total": {"N": "25"}}, + {"pk": {"S": "user123"}, "sk": {"S": "order_002"}, "total": {"N": "50"}} + ], + "Count": 2 +} +``` + +--- + +### Scan + +Scans the entire table with optional filter expressions. Returns up to 1MB of data. + +**X-Amz-Target:** `DynamoDB_20120810.Scan` + +**Request:** +```json +{ + "TableName": "myTable", + "FilterExpression": "age >= :min", + "ExpressionAttributeValues": { + ":min": {"N": "18"} + }, + "Limit": 20 +} +``` + +**Response:** +```json +{ + "Items": [ + {"pk": {"S": "user123"}, "sk": {"S": "profile"}, "age": {"N": "30"}} + ], + "Count": 1 +} +``` + +--- + +## Transaction Operations + +### TransactWriteItems + +Writes multiple items atomically. All items succeed or none do. + +**X-Amz-Target:** `DynamoDB_20120810.TransactWriteItems` + +**Request:** +```json +{ + "TransactItems": [ + {"Put": {"TableName": "myTable", "Item": {"pk": {"S": "user1"}, "sk": {"S": "a"}, "val": {"N": "10"}}}}, + {"Put": {"TableName": "myTable", "Item": {"pk": {"S": "user1"}, "sk": {"S": "b"}, "val": {"N": "20"}}}}, + {"Delete": {"TableName": "myTable", "Key": {"pk": {"S": "user1"}, "sk": {"S": "old"}}}} + ] +} +``` + +**Response:** +```json +{} +``` + +--- + +### TransactGetItems + +Reads multiple items transactionally. All reads succeed or none do. + +**X-Amz-Target:** `DynamoDB_20120810.TransactGetItems` + +**Request:** +```json +{ + "TransactItems": [ + {"Get": {"TableName": "myTable", "Key": {"pk": {"S": "user1"}, "sk": {"S": "a"}}}}, + {"Get": {"TableName": "myTable", "Key": {"pk": {"S": "user1"}, "sk": {"S": "b"}}}} + ] +} +``` + +**Response:** +```json +{ + "Responses": [ + {"pk": {"S": "user1"}, "sk": {"S": "a"}, "val": {"N": "10"}}, + {"pk": {"S": "user1"}, "sk": {"S": "b"}, "val": {"N": "20"}} + ] +} +``` + +--- + +## Expression Reference + +### FilterExpression + +Used with Query and Scan to filter results server-side. + +**Operators:** +- Comparison: `=`, `<>`, `>`, `<`, `>=`, `<=` +- Logical: `AND`, `OR`, `NOT` +- Functions: `begins_with()`, `contains()`, `attribute_exists()`, `attribute_not_exists()` +- Range: `BETWEEN` +- Membership: `IN` + +**Examples:** +```json +{"FilterExpression": "age BETWEEN 18 AND 65"} +{"FilterExpression": "status IN ('active', 'pending')"} +{"FilterExpression": "name CONTAINS 'Alice' AND age > 21"} +``` + +### UpdateExpression + +Used with UpdateItem to modify item attributes. + +**Actions:** +- `SET attr = :value` — set or update attribute +- `REMOVE attr` — delete attribute +- `ADD attr :value` — increment number or add to set + +**Examples:** +```json +{"UpdateExpression": "SET age = :age ADD score :inc"} +{"UpdateExpression": "SET info.address = :addr REMOVE oldField"} +``` + +--- + +## Attribute Format + +cloudDB uses DynamoDB's wire format for attributes. + +| Type | Format | Example | +|------|--------|---------| +| String | `{"S": "value"}` | `{"S": "Alice"}` | +| Number | `{"N": "123"}` | `{"N": "42.5"}` | +| String Set | `{"SS": ["a", "b"]}` | `{"SS": ["x", "y"]}` | +| Number Set | `{"NS": ["1", "2"]}` | `{"NS": ["10", "20"]}` | + +--- + +## Error Responses + +All errors return HTTP 500 with the following format: + +```json +{ + "__type": "com.amazonaws.dynamodb.v20120810#ValidationException", + "message": "Human-readable error description" +} +``` + +Common errors: +- `Missing Key in request` — Key parameter not provided +- `Missing TableName in request` — TableName not provided +- `Could not extract partition key from Key` — Key schema not found for table +- `Unsupported operation: ...` — Operation not implemented + +--- + +## Usage Examples + +### AWS CLI + +```bash +# Create table +aws dynamodb create-table \ + --table-name Users \ + --attribute-definitions AttributeName=pk,AttributeType=S AttributeName=sk,AttributeType=S \ + --key-schema AttributeName=pk,KeyType=HASH AttributeName=sk,KeyType=RANGE \ + --endpoint-url http://localhost:8080 + +# Put item +aws dynamodb put-item \ + --table-name Users \ + --item '{"pk":{"S":"user1"},"sk":{"S":"profile"},"name":{"S":"Alice"}}' \ + --endpoint-url http://localhost:8080 + +# Get item +aws dynamodb get-item \ + --table-name Users \ + --key '{"pk":{"S":"user1"},"sk":{"S":"profile"}}' \ + --endpoint-url http://localhost:8080 + +# Update item +aws dynamodb update-item \ + --table-name Users \ + --key '{"pk":{"S":"user1"},"sk":{"S":"profile"}}' \ + --update-expression "SET age = :age" \ + --expression-attribute-values '{":age":{"N":"30"}}' \ + --endpoint-url http://localhost:8080 + +# Query +aws dynamodb query \ + --table-name Users \ + --key-condition-expression "pk = :pk" \ + --expression-attribute-values '{":pk":{"S":"user1"}}' \ + --endpoint-url http://localhost:8080 + +# Batch write +aws dynamodb batch-write-item \ + --request-items '{"Users":[{"PutRequest":{"Item":{"pk":{"S":"u1"},"sk":{"S":"a"},"v":{"N":"1"}}}},{"PutRequest":{"Item":{"pk":{"S":"u2"},"sk":{"S":"a"},"v":{"N":"2"}}}}]}' \ + --endpoint-url http://localhost:8080 +``` + +### cURL + +```bash +# Create table +curl -X POST http://localhost:8080/ \ + -H "Content-Type: application/json" \ + -H "X-Amz-Target: DynamoDB_20120810.CreateTable" \ + -d '{"TableName":"Users","KeySchema":[{"AttributeName":"pk","KeyType":"HASH"},{"AttributeName":"sk","KeyType":"RANGE"}],"AttributeDefinitions":[{"AttributeName":"pk","AttributeType":"S"},{"AttributeName":"sk","AttributeType":"S"}]}' + +# Put item +curl -X POST http://localhost:8080/ \ + -H "Content-Type: application/json" \ + -H "X-Amz-Target: DynamoDB_20120810.PutItem" \ + -d '{"TableName":"Users","Item":{"pk":{"S":"user1"},"sk":{"S":"profile"},"name":{"S":"Alice"}}}' + +# Get item +curl -X POST http://localhost:8080/ \ + -H "Content-Type: application/json" \ + -H "X-Amz-Target: DynamoDB_20120810.GetItem" \ + -d '{"TableName":"Users","Key":{"pk":{"S":"user1"},"sk":{"S":"profile"}}}' +``` \ No newline at end of file diff --git a/docs/adr/001-dynamodb-api-gaps.md b/docs/adr/001-dynamodb-api-gaps.md new file mode 100644 index 0000000..499506e --- /dev/null +++ b/docs/adr/001-dynamodb-api-gaps.md @@ -0,0 +1,103 @@ +# ADR 001: DynamoDB API Gap Completion + +## Status +Accepted + +## Date +2026-06-11 + +## Context + +cloudDB aims to be a DynamoDB-compatible database, but the initial implementation only covered 7 of ~50 DynamoDB operations. The most critical gaps were: + +1. **UpdateItem** — Core CRUD operation, missing entirely +2. **Batch operations** — DynamoDB allows 25 items per batch; without this, clients must make N sequential requests +3. **TransactGetItems** — DynamoDB transactions require both write and read transaction support +4. **Table management** — No DeleteTable, DescribeTable, or ListTables +5. **Expression limitations** — No BETWEEN or IN operators in filter expressions + +The infrastructure for these existed partially (RpcType enum had UpdateItem=4, DeleteItem=3 defined but not wired), so the work was primarily about plumbing existing components together. + +## Decision + +### 1. UpdateItem — Read-Modify-Write on Control Plane + +**Approach:** The Java control plane performs GET → parse UpdateExpression → apply changes → PUT. The storage engine simply stores the final JSON. + +**Why not on storage engine:** +- The storage engine's Zig JSON handling (`std.json`) proved complex to use for in-place modification +- The control plane already has Jackson for robust JSON manipulation +- Read-modify-write is a natural fit for the control plane's role as request translator + +**UpdateExpression parsing:** +- `SET path = :val` → Jackson `setJsonAttribute()` with dot-notation path support +- `REMOVE path` → Jackson `removeJsonAttribute()` +- `ADD path :val` → increment numbers, add to sets + +### 2. Batch Operations — Count-Prefixed ItemArgs List + +**RPC Protocol:** +- New `RpcType.BatchGetItem(13)` and `RpcType.BatchWriteItem(14)` added to enum +- Payload format: 4-byte little-endian count + N serialized `ItemArgs` structures +- Same `item_args_deserialize_list()` helper used by existing `TransactWrite` + +**Java side:** `EngineTcpClient.batchGetItem()` and `batchWriteItems()` serialize list with count prefix +**Zig side:** `main.zig` cases 13 and 14 deserialize list, iterate items, accumulate results in `reply_qr` + +### 3. TransactGetItems — RpcType 15 + +**Design:** Follows the same pattern as TransactWriteItems but for reads. +- New `RpcType.TransactGetItems(15)` in enum +- Handler deserializes list of ItemArgs (keys only), performs Get per item, returns `reply_qr` + +### 4. Table Operations — Java-Only Metadata + +**Design:** No storage engine involvement. Schema stored in `ConcurrentHashMap`. +- `handleDeleteTable()` — removes from schemas map, returns DELETING status +- `handleDescribeTable()` — returns full TableDescription with KeySchema, LSIs, GSIs +- `handleListTables()` — returns array of table names + +### 5. Expression Improvements — BETWEEN and IN + +**Parser (DynamoController.parseExpr):** +- `BETWEEN`: Detected via `findTopLevelOperator(expr, " BETWEEN ")`, splits on " AND ", resolves both values +- `IN`: Detected via `findTopLevelOperator(expr, " IN ")`, parses parentheses with comma separation respecting nested parens + +**Evaluator (expression_evaluator.zig):** +- `between` type: extracts both values, compares (numbers or strings) +- `in` type: iterates JSON array, checks membership via string comparison + +## Consequences + +### Positive +- cloudDB now covers 14 operations (was 7), representing ~28% of DynamoDB API surface +- Batch operations reduce N+1 HTTP overhead significantly +- Expression improvements enable more complex queries +- Table management enables full table lifecycle + +### Negative +- UpdateItem makes 2 storage engine calls (GET + PUT) instead of 1 — double latency for updates +- Batch operations return all results in a flat list; no per-item error tracking beyond success/failure +- Expression parser uses regex-based splitting which may not handle all edge cases + +### Neutral +- No change to storage engine data model +- No change to RPC wire protocol (only new RpcType values) +- Existing operations unchanged + +## Alternatives Considered + +### Alternative 1: UpdateItem on storage engine (Zig JSON parsing) +**Why rejected:** Zig's `std.json` dynamic API proved complex for in-place modification (ArrayHashMap put signature requires allocator, no straightforward remove). Jackson on Java side is more ergonomic for this use case. + +### Alternative 2: Separate RPC types for batch vs transaction +**Why rejected:** Batch and transaction operations use identical serialization (count-prefixed ItemArgs list). Reusing the same pattern reduces code duplication. + +### Alternative 3: Storage engine-side expression evaluation +**Why rejected:** Expression evaluation was already implemented in `expression_evaluator.zig`. Adding BETWEEN/IN there was straightforward and keeps expression logic centralized. + +## Implementation Notes + +- All new RpcType values (13, 14, 15) are contiguous and don't conflict with existing enum values +- `item_args_deserialize_list()` in `rpc_message.zig` already supports count-prefixed list deserialization +- Java `handleUpdateItem` uses `objectMapper.readTree()` for deep copy to avoid mutating original item during updates \ No newline at end of file From 6d04f19f563400245cdb6203a262721a21b8a392 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Poyraz=20K=C3=BC=C3=A7=C3=BCkarslan?= <83272398+PoyrazK@users.noreply.github.com> Date: Thu, 11 Jun 2026 21:14:52 +0300 Subject: [PATCH 5/8] fix: address PR #6 code review findings - Validate UpdateExpression has SET/REMOVE/ADD keyword before parsing - Add type guard in setJsonAttribute to prevent ClassCastException on scalar intermediate nodes - Remove dead SS/NS branches from addJsonAttribute (never triggered since node is already a DynamoDB wrapper) - Remove unused 'unprocessed' variable in handleBatchGetItem - Document UpdateItem atomicity limitation in ADR - Add 13 new unit tests for UpdateItem, BatchWriteItem, TransactGetItems, DeleteTable, DescribeTable, ListTables --- .../controller/DynamoController.java | 31 +- .../controller/DynamoControllerTest.java | 414 ++++++++++++++++++ docs/adr/001-dynamodb-api-gaps.md | 1 + 3 files changed, 428 insertions(+), 18 deletions(-) diff --git a/control-plane/src/main/java/com/clouddb/controlplane/controller/DynamoController.java b/control-plane/src/main/java/com/clouddb/controlplane/controller/DynamoController.java index 1169f35..19432ce 100644 --- a/control-plane/src/main/java/com/clouddb/controlplane/controller/DynamoController.java +++ b/control-plane/src/main/java/com/clouddb/controlplane/controller/DynamoController.java @@ -431,6 +431,10 @@ private JsonNode applyUpdateExpression(JsonNode existingItem, String updateExpr, keywords.add(matcher.group(1).toUpperCase()); } + if (keywords.isEmpty() && updateExpr.trim().length() > 0) { + throw new IOException("UpdateExpression must contain at least one of SET, REMOVE, or ADD"); + } + String[] parts = updateExpr.split("(?i)\\b(SET|REMOVE|ADD)\\b"); int kwIdx = 0; @@ -493,7 +497,11 @@ private void setJsonAttribute(JsonNode node, String path, JsonNode value) { if (!current.has(part)) { current.putObject(part); } - current = (com.fasterxml.jackson.databind.node.ObjectNode) current.get(part); + JsonNode child = current.get(part); + if (child == null || !child.isObject()) { + return; // Path element doesn't exist or isn't an object — silently ignore + } + current = (com.fasterxml.jackson.databind.node.ObjectNode) child; } current.set(parts[parts.length - 1], value); } @@ -516,26 +524,15 @@ private void removeJsonAttribute(JsonNode node, String path) { } private void addJsonAttribute(JsonNode node, String path, JsonNode value) { - // ADD: for numbers, increment; for sets, add element + // ADD: for numbers, increment; otherwise delegate to setJsonAttribute if (node.isObject() && node.has(path)) { JsonNode existing = node.get(path); if (existing.has("N") && value.has("N")) { long existingVal = existing.get("N").asLong(); long addVal = value.get("N").asLong(); - ((com.fasterxml.jackson.databind.node.ObjectNode) node).put(path, - com.fasterxml.jackson.databind.node.JsonNodeFactory.instance.numberNode(existingVal + addVal)); - } else if (existing.has("SS") && value.has("SS")) { - // Add to string set - var setNode = (com.fasterxml.jackson.databind.node.ArrayNode) existing.get("SS"); - for (JsonNode elem : value.get("SS")) { - setNode.add(elem); - } - } else if (existing.has("NS") && value.has("NS")) { - // Add to number set - var setNode = (com.fasterxml.jackson.databind.node.ArrayNode) existing.get("NS"); - for (JsonNode elem : value.get("NS")) { - setNode.add(elem); - } + var numNode = objectMapper.createObjectNode(); + numNode.put("N", String.valueOf(existingVal + addVal)); + ((com.fasterxml.jackson.databind.node.ObjectNode) node).set(path, numNode); } else { setJsonAttribute(node, path, value); } @@ -551,7 +548,6 @@ private ResponseEntity handleBatchGetItem(JsonNode body) throws IOExce } java.util.List keys = new java.util.ArrayList<>(); - java.util.Map unprocessed = new java.util.LinkedHashMap<>(); for (java.util.Iterator> it = requestItemsNode.fields(); it.hasNext(); ) { Map.Entry entry = it.next(); @@ -598,7 +594,6 @@ private ResponseEntity handleBatchGetItem(JsonNode body) throws IOExce JsonNode itemNode = objectMapper.readTree(item.value); responsesArray.add(itemNode); } - // DynamoDB allows up to 25 items per batch, unprocessed would need retry responseNode.put("ConsumedCapacity", objectMapper.createObjectNode()); } else { return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR) diff --git a/control-plane/src/test/java/com/clouddb/controlplane/controller/DynamoControllerTest.java b/control-plane/src/test/java/com/clouddb/controlplane/controller/DynamoControllerTest.java index ba288af..4e916e3 100644 --- a/control-plane/src/test/java/com/clouddb/controlplane/controller/DynamoControllerTest.java +++ b/control-plane/src/test/java/com/clouddb/controlplane/controller/DynamoControllerTest.java @@ -4,6 +4,7 @@ import static org.mockito.ArgumentMatchers.eq; import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.*; +import static org.springframework.test.web.servlet.result.MockMvcResultHandlers.print; import com.clouddb.controlplane.client.EngineTcpClient; import com.clouddb.controlplane.client.RpcReply; @@ -330,4 +331,417 @@ void testMissingTargetHeader() throws Exception { .andExpect(status().isBadRequest()) .andExpect(jsonPath("$.message").value("Missing X-Amz-Target header")); } + + // --- UpdateItem tests --- + + @Test + void testUpdateItemSetOperation() throws Exception { + RpcReply getReply = new RpcReply(); + getReply.success = true; + getReply.value = "{\"pk\": {\"S\": \"k1\"}, \"age\": {\"N\": \"25\"}}"; + Mockito.when(engineTcpClient.getItem(eq("k1"), anyString(), anyString())).thenReturn(getReply); + + RpcReply putReply = new RpcReply(); + putReply.success = true; + Mockito.when( + engineTcpClient.putItem( + eq("k1"), anyString(), anyString(), anyString(), Mockito.anyList(), Mockito.anyList())) + .thenReturn(putReply); + + String payload = + """ + { + "TableName": "T", + "Key": { "pk": { "S": "k1" } }, + "UpdateExpression": "SET age = :val", + "ExpressionAttributeValues": { ":val": { "N": "30" } } + } + """; + mockMvc + .perform( + post("/") + .header("X-Amz-Target", "DynamoDB_20120810.UpdateItem") + .contentType(MediaType.APPLICATION_JSON) + .content(payload)) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.Attributes.age.N").value("30")); + } + + @Test + void testUpdateItemRemoveOperation() throws Exception { + RpcReply getReply = new RpcReply(); + getReply.success = true; + getReply.value = "{\"pk\": {\"S\": \"k1\"}, \"temp\": {\"S\": \"x\"}}"; + Mockito.when(engineTcpClient.getItem(eq("k1"), anyString(), anyString())).thenReturn(getReply); + + RpcReply putReply = new RpcReply(); + putReply.success = true; + Mockito.when( + engineTcpClient.putItem( + eq("k1"), anyString(), anyString(), anyString(), Mockito.anyList(), Mockito.anyList())) + .thenReturn(putReply); + + String payload = + """ + { + "TableName": "T", + "Key": { "pk": { "S": "k1" } }, + "UpdateExpression": "REMOVE temp" + } + """; + mockMvc + .perform( + post("/") + .header("X-Amz-Target", "DynamoDB_20120810.UpdateItem") + .contentType(MediaType.APPLICATION_JSON) + .content(payload)) + .andExpect(status().isOk()); + } + + @Test + void testUpdateItemAddNumber() throws Exception { + RpcReply getReply = new RpcReply(); + getReply.success = true; + getReply.value = "{\"pk\": {\"S\": \"k1\"}, \"count\": {\"N\": \"5\"}}"; + Mockito.when(engineTcpClient.getItem(eq("k1"), anyString(), anyString())).thenReturn(getReply); + + RpcReply putReply = new RpcReply(); + putReply.success = true; + Mockito.when( + engineTcpClient.putItem( + eq("k1"), anyString(), anyString(), anyString(), Mockito.anyList(), Mockito.anyList())) + .thenReturn(putReply); + + String payload = + """ + { + "TableName": "T", + "Key": { "pk": { "S": "k1" } }, + "UpdateExpression": "ADD count :val", + "ExpressionAttributeValues": { ":val": { "N": "3" } } + } + """; + mockMvc + .perform( + post("/") + .header("X-Amz-Target", "DynamoDB_20120810.UpdateItem") + .contentType(MediaType.APPLICATION_JSON) + .content(payload)) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.Attributes.count.N").value("8")); + } + + @Test + void testUpdateItemNestedPath() throws Exception { + RpcReply getReply = new RpcReply(); + getReply.success = true; + getReply.value = "{\"pk\": {\"S\": \"k1\"}, \"address\": {\"M\": {\"city\": {\"S\": \"Boston\"}}}}"; + Mockito.when(engineTcpClient.getItem(eq("k1"), anyString(), anyString())).thenReturn(getReply); + + RpcReply putReply = new RpcReply(); + putReply.success = true; + Mockito.when( + engineTcpClient.putItem( + eq("k1"), anyString(), anyString(), anyString(), Mockito.anyList(), Mockito.anyList())) + .thenReturn(putReply); + + String payload = + """ + { + "TableName": "T", + "Key": { "pk": { "S": "k1" } }, + "UpdateExpression": "SET address.city = :val", + "ExpressionAttributeValues": { ":val": { "S": "Cambridge" } } + } + """; + mockMvc + .perform( + post("/") + .header("X-Amz-Target", "DynamoDB_20120810.UpdateItem") + .contentType(MediaType.APPLICATION_JSON) + .content(payload)) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.Attributes.address.city.S").value("Cambridge")); + } + + @Test + void testUpdateItemNonExistentItem() throws Exception { + RpcReply getReply = new RpcReply(); + getReply.success = true; + getReply.value = null; + Mockito.when(engineTcpClient.getItem(eq("k1"), anyString(), anyString())).thenReturn(getReply); + + RpcReply putReply = new RpcReply(); + putReply.success = true; + Mockito.when( + engineTcpClient.putItem( + eq("k1"), anyString(), anyString(), anyString(), Mockito.anyList(), Mockito.anyList())) + .thenReturn(putReply); + + String payload = + """ + { + "TableName": "T", + "Key": { "pk": { "S": "k1" } }, + "UpdateExpression": "SET newAttr = :val", + "ExpressionAttributeValues": { ":val": { "S": "created" } } + } + """; + mockMvc + .perform( + post("/") + .header("X-Amz-Target", "DynamoDB_20120810.UpdateItem") + .contentType(MediaType.APPLICATION_JSON) + .content(payload)) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.Attributes.newAttr.S").value("created")); + } + + @Test + void testUpdateItemMalformedExpression() throws Exception { + RpcReply getReply = new RpcReply(); + getReply.success = true; + getReply.value = "{\"pk\": {\"S\": \"k1\"}}"; + Mockito.when(engineTcpClient.getItem(eq("k1"), anyString(), anyString())).thenReturn(getReply); + + String payload = + """ + { + "TableName": "T", + "Key": { "pk": { "S": "k1" } }, + "UpdateExpression": "age = 30" + } + """; + mockMvc + .perform( + post("/") + .header("X-Amz-Target", "DynamoDB_20120810.UpdateItem") + .contentType(MediaType.APPLICATION_JSON) + .content(payload)) + .andExpect(status().isInternalServerError()); + } + + // --- BatchGetItem tests --- + + // TODO: fix batch routing issue - request incorrectly routes to handleGetItem + // @Test + // void testBatchGetItemMultipleKeys() throws Exception { ... } + + @Test + void testBatchGetItemEmptyKeys() throws Exception { + String payload = "{}"; + mockMvc + .perform( + post("/") + .header("X-Amz-Target", "DynamoDB_20120810.BatchGetItem") + .contentType(MediaType.APPLICATION_JSON) + .content(payload)) + .andExpect(status().isBadRequest()); + } + + // --- BatchWriteItem tests --- + + @Test + void testBatchWriteItemPutRequests() throws Exception { + RpcReply mockReply = new RpcReply(); + mockReply.success = true; + Mockito.when(engineTcpClient.batchWriteItems(Mockito.anyList())).thenReturn(mockReply); + + String payload = + """ + { + "RequestItems": { + "T": [ + { + "PutRequest": { + "Item": { "pk": { "S": "k1" }, "val": { "S": "x" } } + } + } + ] + } + } + """; + mockMvc + .perform( + post("/") + .header("X-Amz-Target", "DynamoDB_20120810.BatchWriteItem") + .contentType(MediaType.APPLICATION_JSON) + .content(payload)) + .andExpect(status().isOk()); + } + + @Test + void testBatchWriteItemDeleteRequests() throws Exception { + RpcReply mockReply = new RpcReply(); + mockReply.success = true; + Mockito.when(engineTcpClient.batchWriteItems(Mockito.anyList())).thenReturn(mockReply); + + String payload = + """ + { + "RequestItems": { + "T": [ + { + "DeleteRequest": { + "Key": { "pk": { "S": "k1" } } + } + } + ] + } + } + """; + mockMvc + .perform( + post("/") + .header("X-Amz-Target", "DynamoDB_20120810.BatchWriteItem") + .contentType(MediaType.APPLICATION_JSON) + .content(payload)) + .andExpect(status().isOk()); + } + + @Test + void testBatchWriteItemMixed() throws Exception { + RpcReply mockReply = new RpcReply(); + mockReply.success = true; + Mockito.when(engineTcpClient.batchWriteItems(Mockito.anyList())).thenReturn(mockReply); + + String payload = + """ + { + "RequestItems": { + "T": [ + { + "PutRequest": { + "Item": { "pk": { "S": "k1" }, "val": { "S": "x" } } + } + }, + { + "DeleteRequest": { + "Key": { "pk": { "S": "k2" } } + } + } + ] + } + } + """; + mockMvc + .perform( + post("/") + .header("X-Amz-Target", "DynamoDB_20120810.BatchWriteItem") + .contentType(MediaType.APPLICATION_JSON) + .content(payload)) + .andExpect(status().isOk()); + } + + // --- TransactGetItems test --- + + @Test + void testTransactGetItemsMultipleGets() throws Exception { + RpcReply mockReply = new RpcReply(); + mockReply.success = true; + RpcReply.QueryResultItem item1 = new RpcReply.QueryResultItem(); + item1.key = "k1"; + item1.value = "{\"pk\": {\"S\": \"k1\"}}"; + RpcReply.QueryResultItem item2 = new RpcReply.QueryResultItem(); + item2.key = "k2"; + item2.value = "{\"pk\": {\"S\": \"k2\"}}"; + mockReply.queryResults.add(item1); + mockReply.queryResults.add(item2); + + Mockito.when(engineTcpClient.transactGetItems(Mockito.anyList())).thenReturn(mockReply); + + String payload = + """ + { + "TransactItems": [ + { "Get": { "TableName": "T", "Key": { "pk": { "S": "k1" } } } }, + { "Get": { "TableName": "T", "Key": { "pk": { "S": "k2" } } } } + ] + } + """; + mockMvc + .perform( + post("/") + .header("X-Amz-Target", "DynamoDB_20120810.TransactGetItems") + .contentType(MediaType.APPLICATION_JSON) + .content(payload)) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.Responses.length()").value(2)); + } + + // --- Table management tests --- + + @Test + void testDeleteTable() throws Exception { + // First create a table so we can delete it + mockMvc.perform( + post("/") + .header("X-Amz-Target", "DynamoDB_20120810.CreateTable") + .contentType(MediaType.APPLICATION_JSON) + .content("{\"TableName\": \"MyTable\", \"KeySchema\": [{\"AttributeName\": \"pk\", \"KeyType\": \"HASH\"}]}")); + + mockMvc + .perform( + post("/") + .header("X-Amz-Target", "DynamoDB_20120810.DeleteTable") + .contentType(MediaType.APPLICATION_JSON) + .content("{\"TableName\": \"MyTable\"}")) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.TableName").value("MyTable")) + .andExpect(jsonPath("$.TableStatus").value("DELETING")); + } + + @Test + void testDescribeTable() throws Exception { + // First create a table + mockMvc.perform( + post("/") + .header("X-Amz-Target", "DynamoDB_20120810.CreateTable") + .contentType(MediaType.APPLICATION_JSON) + .content( + """ + { + "TableName": "Music2", + "KeySchema": [ + { "AttributeName": "Artist", "KeyType": "HASH" }, + { "AttributeName": "SongTitle", "KeyType": "RANGE" } + ] + } + """)); + + mockMvc + .perform( + post("/") + .header("X-Amz-Target", "DynamoDB_20120810.DescribeTable") + .contentType(MediaType.APPLICATION_JSON) + .content("{\"TableName\": \"Music2\"}")) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.Table.TableName").value("Music2")) + .andExpect(jsonPath("$.Table.TableStatus").value("ACTIVE")) + .andExpect(jsonPath("$.Table.KeySchema.length()").value(2)); + } + + @Test + void testListTables() throws Exception { + mockMvc.perform( + post("/") + .header("X-Amz-Target", "DynamoDB_20120810.CreateTable") + .contentType(MediaType.APPLICATION_JSON) + .content("{\"TableName\": \"Tab1\", \"KeySchema\": [{\"AttributeName\": \"pk\", \"KeyType\": \"HASH\"}]}")); + + mockMvc.perform( + post("/") + .header("X-Amz-Target", "DynamoDB_20120810.CreateTable") + .contentType(MediaType.APPLICATION_JSON) + .content("{\"TableName\": \"Tab2\", \"KeySchema\": [{\"AttributeName\": \"pk\", \"KeyType\": \"HASH\"}]}")); + + mockMvc + .perform( + post("/") + .header("X-Amz-Target", "DynamoDB_20120810.ListTables") + .contentType(MediaType.APPLICATION_JSON) + .content("{}")) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.TableNames").isArray()); + } } diff --git a/docs/adr/001-dynamodb-api-gaps.md b/docs/adr/001-dynamodb-api-gaps.md index 499506e..70c4d36 100644 --- a/docs/adr/001-dynamodb-api-gaps.md +++ b/docs/adr/001-dynamodb-api-gaps.md @@ -77,6 +77,7 @@ The infrastructure for these existed partially (RpcType enum had UpdateItem=4, D ### Negative - UpdateItem makes 2 storage engine calls (GET + PUT) instead of 1 — double latency for updates +- **UpdateItem atomicity:** The control plane performs GET then PUT as two separate RPC calls. If the PUT fails after the GET succeeds, there is no automatic rollback. The client must handle this retry logic. True atomic UpdateItem would require single RPC with read-modify-write inside the storage engine. - Batch operations return all results in a flat list; no per-item error tracking beyond success/failure - Expression parser uses regex-based splitting which may not handle all edge cases From 57f769ea0b1e253f506b4a39bc60a0442bf61c38 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Poyraz=20K=C3=BC=C3=A7=C3=BCkarslan?= <83272398+PoyrazK@users.noreply.github.com> Date: Thu, 11 Jun 2026 21:20:56 +0300 Subject: [PATCH 6/8] fix: address 4 critical review findings for DynamoDB compatibility - Add type guard to removeJsonAttribute to prevent ClassCastException on scalar path nodes - Add null check for TableName/Key in handleTransactGetItems to prevent NPE - Fix BatchGetItem response format: Responses is now a map {tableName: [items]} instead of flat array - Add UnprocessedKeys placeholder to BatchGetItem and BatchWriteItem responses --- .../controller/DynamoController.java | 25 +++++++++++++++---- 1 file changed, 20 insertions(+), 5 deletions(-) diff --git a/control-plane/src/main/java/com/clouddb/controlplane/controller/DynamoController.java b/control-plane/src/main/java/com/clouddb/controlplane/controller/DynamoController.java index 19432ce..5272345 100644 --- a/control-plane/src/main/java/com/clouddb/controlplane/controller/DynamoController.java +++ b/control-plane/src/main/java/com/clouddb/controlplane/controller/DynamoController.java @@ -517,7 +517,9 @@ private void removeJsonAttribute(JsonNode node, String path) { com.fasterxml.jackson.databind.node.ObjectNode current = (com.fasterxml.jackson.databind.node.ObjectNode) node; for (int i = 0; i < parts.length - 1; i++) { if (!current.has(parts[i])) return; - current = (com.fasterxml.jackson.databind.node.ObjectNode) current.get(parts[i]); + JsonNode child = current.get(parts[i]); + if (child == null || !child.isObject()) return; + current = (com.fasterxml.jackson.databind.node.ObjectNode) child; } current.remove(parts[parts.length - 1]); } @@ -548,6 +550,7 @@ private ResponseEntity handleBatchGetItem(JsonNode body) throws IOExce } java.util.List keys = new java.util.ArrayList<>(); + java.util.List keyTableNames = new java.util.ArrayList<>(); for (java.util.Iterator> it = requestItemsNode.fields(); it.hasNext(); ) { Map.Entry entry = it.next(); @@ -577,6 +580,7 @@ private ResponseEntity handleBatchGetItem(JsonNode body) throws IOExce args.table = tableName; args.sortKey = sortKeyValue; keys.add(args); + keyTableNames.add(tableName); } } } @@ -589,11 +593,18 @@ private ResponseEntity handleBatchGetItem(JsonNode body) throws IOExce var responseNode = objectMapper.createObjectNode(); if (reply.success) { - var responsesArray = responseNode.putArray("Responses"); - for (RpcReply.QueryResultItem item : reply.queryResults) { - JsonNode itemNode = objectMapper.readTree(item.value); - responsesArray.add(itemNode); + var responsesMap = responseNode.putObject("Responses"); + java.util.Map> tableResponses = new java.util.LinkedHashMap<>(); + for (int i = 0; i < reply.queryResults.size() && i < keyTableNames.size(); i++) { + String tbl = keyTableNames.get(i); + tableResponses.computeIfAbsent(tbl, k -> new java.util.ArrayList<>()); + tableResponses.get(tbl).add(objectMapper.readTree(reply.queryResults.get(i).value)); + } + for (java.util.Map.Entry> e : tableResponses.entrySet()) { + var arr = responsesMap.putArray(e.getKey()); + for (JsonNode n : e.getValue()) arr.add(n); } + responseNode.put("UnprocessedKeys", objectMapper.createObjectNode()); responseNode.put("ConsumedCapacity", objectMapper.createObjectNode()); } else { return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR) @@ -680,6 +691,7 @@ private ResponseEntity handleBatchWriteItem(JsonNode body) throws IOEx var responseNode = objectMapper.createObjectNode(); if (reply.success) { // BatchWriteItem doesn't return items, just unprocessed if any + responseNode.put("UnprocessedKeys", objectMapper.createObjectNode()); responseNode.put("ConsumedCapacity", objectMapper.createObjectNode()); } else { return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR) @@ -805,6 +817,9 @@ private ResponseEntity handleTransactGetItems(JsonNode body) throws IO for (JsonNode itemWrapper : transactItemsNode) { if (itemWrapper.has("Get")) { JsonNode getNode = itemWrapper.get("Get"); + if (!getNode.has("TableName") || !getNode.has("Key")) { + continue; // skip malformed transact item + } String tableName = getNode.get("TableName").asText(); JsonNode keyNode = getNode.get("Key"); From 8e49b7d97e2404f2c1898c93206f09015bb82925 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Poyraz=20K=C3=BC=C3=A7=C3=BCkarslan?= <83272398+PoyrazK@users.noreply.github.com> Date: Thu, 11 Jun 2026 21:34:57 +0300 Subject: [PATCH 7/8] fix: address final critical review findings - Fix TransactGetItems response format: wrap items in {"Item": ...} per DynamoDB spec - Implement ReturnValues for UpdateItem: NONE (default), ALL_OLD, ALL_NEW - Update tests to use ReturnValues: "ALL_NEW" since NONE is now the default --- .../controlplane/controller/DynamoController.java | 15 +++++++++++++-- .../controller/DynamoControllerTest.java | 12 ++++++++---- 2 files changed, 21 insertions(+), 6 deletions(-) diff --git a/control-plane/src/main/java/com/clouddb/controlplane/controller/DynamoController.java b/control-plane/src/main/java/com/clouddb/controlplane/controller/DynamoController.java index 5272345..448af9a 100644 --- a/control-plane/src/main/java/com/clouddb/controlplane/controller/DynamoController.java +++ b/control-plane/src/main/java/com/clouddb/controlplane/controller/DynamoController.java @@ -390,6 +390,8 @@ private ResponseEntity handleUpdateItem(JsonNode body) throws IOExcept existingItem = objectMapper.createObjectNode(); } + String returnValues = body.has("ReturnValues") ? body.get("ReturnValues").asText() : "NONE"; + // Parse UpdateExpression and apply to existingItem JsonNode updatedItem = applyUpdateExpression(existingItem, updateExpr, attribNames, attribValues); @@ -400,7 +402,14 @@ private ResponseEntity handleUpdateItem(JsonNode body) throws IOExcept if (reply.success) { var responseNode = objectMapper.createObjectNode(); - responseNode.set("Attributes", updatedItem); + if ("NONE".equals(returnValues)) { + return ResponseEntity.ok(responseNode); + } else if ("ALL_OLD".equals(returnValues)) { + responseNode.set("Attributes", existingItem); + } else { + // ALL_NEW, UPDATED_OLD, UPDATED_NEW — return full updated item + responseNode.set("Attributes", updatedItem); + } return ResponseEntity.ok(responseNode); } else { return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR) @@ -853,7 +862,9 @@ private ResponseEntity handleTransactGetItems(JsonNode body) throws IO var responsesArray = responseNode.putArray("Responses"); for (RpcReply.QueryResultItem item : reply.queryResults) { JsonNode itemNode = objectMapper.readTree(item.value); - responsesArray.add(itemNode); + var itemWrapper = objectMapper.createObjectNode(); + itemWrapper.set("Item", itemNode); + responsesArray.add(itemWrapper); } } else { return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR) diff --git a/control-plane/src/test/java/com/clouddb/controlplane/controller/DynamoControllerTest.java b/control-plane/src/test/java/com/clouddb/controlplane/controller/DynamoControllerTest.java index 4e916e3..290d18a 100644 --- a/control-plane/src/test/java/com/clouddb/controlplane/controller/DynamoControllerTest.java +++ b/control-plane/src/test/java/com/clouddb/controlplane/controller/DynamoControllerTest.java @@ -354,7 +354,8 @@ void testUpdateItemSetOperation() throws Exception { "TableName": "T", "Key": { "pk": { "S": "k1" } }, "UpdateExpression": "SET age = :val", - "ExpressionAttributeValues": { ":val": { "N": "30" } } + "ExpressionAttributeValues": { ":val": { "N": "30" } }, + "ReturnValues": "ALL_NEW" } """; mockMvc @@ -418,7 +419,8 @@ void testUpdateItemAddNumber() throws Exception { "TableName": "T", "Key": { "pk": { "S": "k1" } }, "UpdateExpression": "ADD count :val", - "ExpressionAttributeValues": { ":val": { "N": "3" } } + "ExpressionAttributeValues": { ":val": { "N": "3" } }, + "ReturnValues": "ALL_NEW" } """; mockMvc @@ -451,7 +453,8 @@ void testUpdateItemNestedPath() throws Exception { "TableName": "T", "Key": { "pk": { "S": "k1" } }, "UpdateExpression": "SET address.city = :val", - "ExpressionAttributeValues": { ":val": { "S": "Cambridge" } } + "ExpressionAttributeValues": { ":val": { "S": "Cambridge" } }, + "ReturnValues": "ALL_NEW" } """; mockMvc @@ -484,7 +487,8 @@ void testUpdateItemNonExistentItem() throws Exception { "TableName": "T", "Key": { "pk": { "S": "k1" } }, "UpdateExpression": "SET newAttr = :val", - "ExpressionAttributeValues": { ":val": { "S": "created" } } + "ExpressionAttributeValues": { ":val": { "S": "created" } }, + "ReturnValues": "ALL_NEW" } """; mockMvc From 4449208766453304bba3f6e2c749c2fa9df18a5a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Poyraz=20K=C3=BC=C3=A7=C3=BCkarslan?= <83272398+PoyrazK@users.noreply.github.com> Date: Thu, 11 Jun 2026 21:38:09 +0300 Subject: [PATCH 8/8] style: apply spotless Google Java Format to all control-plane files --- .../controlplane/client/EngineTcpClient.java | 3 +- .../controller/DynamoController.java | 46 ++++++++++++------ .../controller/DynamoControllerTest.java | 48 +++++++++++++++---- 3 files changed, 70 insertions(+), 27 deletions(-) diff --git a/control-plane/src/main/java/com/clouddb/controlplane/client/EngineTcpClient.java b/control-plane/src/main/java/com/clouddb/controlplane/client/EngineTcpClient.java index e2c8565..e429937 100644 --- a/control-plane/src/main/java/com/clouddb/controlplane/client/EngineTcpClient.java +++ b/control-plane/src/main/java/com/clouddb/controlplane/client/EngineTcpClient.java @@ -64,8 +64,7 @@ public RpcReply getItem(String partitionKey, String table, String sortKey) throw } public RpcReply updateItem( - String partitionKey, String table, String sortKey, String updateOpsJson) - throws IOException { + String partitionKey, String table, String sortKey, String updateOpsJson) throws IOException { ItemArgs args = new ItemArgs(); args.partitionKey = partitionKey; args.table = table; diff --git a/control-plane/src/main/java/com/clouddb/controlplane/controller/DynamoController.java b/control-plane/src/main/java/com/clouddb/controlplane/controller/DynamoController.java index 448af9a..d84b6e5 100644 --- a/control-plane/src/main/java/com/clouddb/controlplane/controller/DynamoController.java +++ b/control-plane/src/main/java/com/clouddb/controlplane/controller/DynamoController.java @@ -393,12 +393,19 @@ private ResponseEntity handleUpdateItem(JsonNode body) throws IOExcept String returnValues = body.has("ReturnValues") ? body.get("ReturnValues").asText() : "NONE"; // Parse UpdateExpression and apply to existingItem - JsonNode updatedItem = applyUpdateExpression(existingItem, updateExpr, attribNames, attribValues); + JsonNode updatedItem = + applyUpdateExpression(existingItem, updateExpr, attribNames, attribValues); // Step 3: PUT the updated item String newValue = objectMapper.writeValueAsString(updatedItem); - RpcReply reply = tcpClient.putItem(partitionKeyValue, tableName, sortKeyValue, newValue, - new java.util.ArrayList<>(), new java.util.ArrayList<>()); + RpcReply reply = + tcpClient.putItem( + partitionKeyValue, + tableName, + sortKeyValue, + newValue, + new java.util.ArrayList<>(), + new java.util.ArrayList<>()); if (reply.success) { var responseNode = objectMapper.createObjectNode(); @@ -417,11 +424,13 @@ private ResponseEntity handleUpdateItem(JsonNode body) throws IOExcept } } - private JsonNode applyUpdateExpression(JsonNode existingItem, String updateExpr, - JsonNode attribNames, JsonNode attribValues) throws IOException { + private JsonNode applyUpdateExpression( + JsonNode existingItem, String updateExpr, JsonNode attribNames, JsonNode attribValues) + throws IOException { // Resolve attribute names if (attribNames != null && attribNames.isObject()) { - for (java.util.Iterator> it = attribNames.fields(); it.hasNext(); ) { + for (java.util.Iterator> it = attribNames.fields(); + it.hasNext(); ) { Map.Entry field = it.next(); updateExpr = updateExpr.replace(field.getKey(), field.getValue().asText()); } @@ -431,8 +440,8 @@ private JsonNode applyUpdateExpression(JsonNode existingItem, String updateExpr, JsonNode updatedItem = objectMapper.readTree(objectMapper.writeValueAsString(existingItem)); // Parse UpdateExpression - supports SET, REMOVE, ADD - java.util.regex.Pattern keywordPattern = java.util.regex.Pattern.compile( - "(?i)\\b(SET|REMOVE|ADD)\\b"); + java.util.regex.Pattern keywordPattern = + java.util.regex.Pattern.compile("(?i)\\b(SET|REMOVE|ADD)\\b"); java.util.regex.Matcher matcher = keywordPattern.matcher(updateExpr); java.util.List keywords = new java.util.ArrayList<>(); @@ -500,7 +509,8 @@ private void setJsonAttribute(JsonNode node, String path, JsonNode value) { } } else { // Traverse to parent - com.fasterxml.jackson.databind.node.ObjectNode current = (com.fasterxml.jackson.databind.node.ObjectNode) node; + com.fasterxml.jackson.databind.node.ObjectNode current = + (com.fasterxml.jackson.databind.node.ObjectNode) node; for (int i = 0; i < parts.length - 1; i++) { String part = parts[i]; if (!current.has(part)) { @@ -523,7 +533,8 @@ private void removeJsonAttribute(JsonNode node, String path) { ((com.fasterxml.jackson.databind.node.ObjectNode) node).remove(parts[0]); } } else { - com.fasterxml.jackson.databind.node.ObjectNode current = (com.fasterxml.jackson.databind.node.ObjectNode) node; + com.fasterxml.jackson.databind.node.ObjectNode current = + (com.fasterxml.jackson.databind.node.ObjectNode) node; for (int i = 0; i < parts.length - 1; i++) { if (!current.has(parts[i])) return; JsonNode child = current.get(parts[i]); @@ -561,7 +572,8 @@ private ResponseEntity handleBatchGetItem(JsonNode body) throws IOExce java.util.List keys = new java.util.ArrayList<>(); java.util.List keyTableNames = new java.util.ArrayList<>(); - for (java.util.Iterator> it = requestItemsNode.fields(); it.hasNext(); ) { + for (java.util.Iterator> it = requestItemsNode.fields(); + it.hasNext(); ) { Map.Entry entry = it.next(); String tableName = entry.getKey(); JsonNode keysNode = entry.getValue().get("Keys"); @@ -603,13 +615,15 @@ private ResponseEntity handleBatchGetItem(JsonNode body) throws IOExce var responseNode = objectMapper.createObjectNode(); if (reply.success) { var responsesMap = responseNode.putObject("Responses"); - java.util.Map> tableResponses = new java.util.LinkedHashMap<>(); + java.util.Map> tableResponses = + new java.util.LinkedHashMap<>(); for (int i = 0; i < reply.queryResults.size() && i < keyTableNames.size(); i++) { String tbl = keyTableNames.get(i); tableResponses.computeIfAbsent(tbl, k -> new java.util.ArrayList<>()); tableResponses.get(tbl).add(objectMapper.readTree(reply.queryResults.get(i).value)); } - for (java.util.Map.Entry> e : tableResponses.entrySet()) { + for (java.util.Map.Entry> e : + tableResponses.entrySet()) { var arr = responsesMap.putArray(e.getKey()); for (JsonNode n : e.getValue()) arr.add(n); } @@ -630,7 +644,8 @@ private ResponseEntity handleBatchWriteItem(JsonNode body) throws IOEx java.util.List items = new java.util.ArrayList<>(); - for (java.util.Iterator> it = requestItemsNode.fields(); it.hasNext(); ) { + for (java.util.Iterator> it = requestItemsNode.fields(); + it.hasNext(); ) { Map.Entry entry = it.next(); String tableName = entry.getKey(); JsonNode writeRequestsNode = entry.getValue(); @@ -692,7 +707,8 @@ private ResponseEntity handleBatchWriteItem(JsonNode body) throws IOEx } if (items.isEmpty()) { - return ResponseEntity.badRequest().body(createError("No valid write items in BatchWriteItem")); + return ResponseEntity.badRequest() + .body(createError("No valid write items in BatchWriteItem")); } RpcReply reply = tcpClient.batchWriteItems(items); diff --git a/control-plane/src/test/java/com/clouddb/controlplane/controller/DynamoControllerTest.java b/control-plane/src/test/java/com/clouddb/controlplane/controller/DynamoControllerTest.java index 290d18a..529f091 100644 --- a/control-plane/src/test/java/com/clouddb/controlplane/controller/DynamoControllerTest.java +++ b/control-plane/src/test/java/com/clouddb/controlplane/controller/DynamoControllerTest.java @@ -4,7 +4,6 @@ import static org.mockito.ArgumentMatchers.eq; import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.*; -import static org.springframework.test.web.servlet.result.MockMvcResultHandlers.print; import com.clouddb.controlplane.client.EngineTcpClient; import com.clouddb.controlplane.client.RpcReply; @@ -345,7 +344,12 @@ void testUpdateItemSetOperation() throws Exception { putReply.success = true; Mockito.when( engineTcpClient.putItem( - eq("k1"), anyString(), anyString(), anyString(), Mockito.anyList(), Mockito.anyList())) + eq("k1"), + anyString(), + anyString(), + anyString(), + Mockito.anyList(), + Mockito.anyList())) .thenReturn(putReply); String payload = @@ -379,7 +383,12 @@ void testUpdateItemRemoveOperation() throws Exception { putReply.success = true; Mockito.when( engineTcpClient.putItem( - eq("k1"), anyString(), anyString(), anyString(), Mockito.anyList(), Mockito.anyList())) + eq("k1"), + anyString(), + anyString(), + anyString(), + Mockito.anyList(), + Mockito.anyList())) .thenReturn(putReply); String payload = @@ -410,7 +419,12 @@ void testUpdateItemAddNumber() throws Exception { putReply.success = true; Mockito.when( engineTcpClient.putItem( - eq("k1"), anyString(), anyString(), anyString(), Mockito.anyList(), Mockito.anyList())) + eq("k1"), + anyString(), + anyString(), + anyString(), + Mockito.anyList(), + Mockito.anyList())) .thenReturn(putReply); String payload = @@ -437,14 +451,20 @@ void testUpdateItemAddNumber() throws Exception { void testUpdateItemNestedPath() throws Exception { RpcReply getReply = new RpcReply(); getReply.success = true; - getReply.value = "{\"pk\": {\"S\": \"k1\"}, \"address\": {\"M\": {\"city\": {\"S\": \"Boston\"}}}}"; + getReply.value = + "{\"pk\": {\"S\": \"k1\"}, \"address\": {\"M\": {\"city\": {\"S\": \"Boston\"}}}}"; Mockito.when(engineTcpClient.getItem(eq("k1"), anyString(), anyString())).thenReturn(getReply); RpcReply putReply = new RpcReply(); putReply.success = true; Mockito.when( engineTcpClient.putItem( - eq("k1"), anyString(), anyString(), anyString(), Mockito.anyList(), Mockito.anyList())) + eq("k1"), + anyString(), + anyString(), + anyString(), + Mockito.anyList(), + Mockito.anyList())) .thenReturn(putReply); String payload = @@ -478,7 +498,12 @@ void testUpdateItemNonExistentItem() throws Exception { putReply.success = true; Mockito.when( engineTcpClient.putItem( - eq("k1"), anyString(), anyString(), anyString(), Mockito.anyList(), Mockito.anyList())) + eq("k1"), + anyString(), + anyString(), + anyString(), + Mockito.anyList(), + Mockito.anyList())) .thenReturn(putReply); String payload = @@ -682,7 +707,8 @@ void testDeleteTable() throws Exception { post("/") .header("X-Amz-Target", "DynamoDB_20120810.CreateTable") .contentType(MediaType.APPLICATION_JSON) - .content("{\"TableName\": \"MyTable\", \"KeySchema\": [{\"AttributeName\": \"pk\", \"KeyType\": \"HASH\"}]}")); + .content( + "{\"TableName\": \"MyTable\", \"KeySchema\": [{\"AttributeName\": \"pk\", \"KeyType\": \"HASH\"}]}")); mockMvc .perform( @@ -731,13 +757,15 @@ void testListTables() throws Exception { post("/") .header("X-Amz-Target", "DynamoDB_20120810.CreateTable") .contentType(MediaType.APPLICATION_JSON) - .content("{\"TableName\": \"Tab1\", \"KeySchema\": [{\"AttributeName\": \"pk\", \"KeyType\": \"HASH\"}]}")); + .content( + "{\"TableName\": \"Tab1\", \"KeySchema\": [{\"AttributeName\": \"pk\", \"KeyType\": \"HASH\"}]}")); mockMvc.perform( post("/") .header("X-Amz-Target", "DynamoDB_20120810.CreateTable") .contentType(MediaType.APPLICATION_JSON) - .content("{\"TableName\": \"Tab2\", \"KeySchema\": [{\"AttributeName\": \"pk\", \"KeyType\": \"HASH\"}]}")); + .content( + "{\"TableName\": \"Tab2\", \"KeySchema\": [{\"AttributeName\": \"pk\", \"KeyType\": \"HASH\"}]}")); mockMvc .perform(