From f1b6ab8e2869b4c1e116410fbf875f4f6b9848f9 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: Sat, 13 Jun 2026 16:37:22 +0300 Subject: [PATCH 1/3] feat(api): implement ReturnValues and ConditionExpression for PutItem/DeleteItem - PutItem: add ReturnValues ALL_NEW support (fetch item after write) - DeleteItem: add ReturnValues ALL_OLD support (get item before tombstone) - PutItem: add ConditionExpression evaluation (attribute_not_exists, etc.) - DeleteItem: add ConditionExpression evaluation before tombstone write - Add evaluateCondition() helper mirroring expression_evaluator.zig logic - BatchWriteItem: build UnprocessedKeys from failed items in conflicts list --- .../controller/DynamoController.java | 183 +++++++++++++++++- 1 file changed, 180 insertions(+), 3 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 d84b6e5..36e09a2 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 @@ -192,6 +192,25 @@ private ResponseEntity handlePutItem(JsonNode body) throws IOException .body(createError("Could not extract partition key from Item")); } + // ConditionExpression check: evaluate before write + JsonNode condExpr = body.get("ConditionExpression"); + if (condExpr != null && !condExpr.isNull()) { + RpcReply preGet = tcpClient.getItem(partitionKeyValue, tableName, sortKeyValue); + JsonNode existingItem = null; + if (preGet.success && preGet.value != null && !preGet.value.isEmpty()) { + existingItem = objectMapper.readTree(preGet.value); + } else { + existingItem = objectMapper.createObjectNode(); + } + JsonNode condAttribNames = body.get("ExpressionAttributeNames"); + JsonNode condAttribValues = body.get("ExpressionAttributeValues"); + JsonNode condAst = compileExpression(condExpr.asText(), condAttribNames, condAttribValues); + if (condAst != null && !evaluateCondition(condAst, existingItem)) { + return ResponseEntity.status(HttpStatus.BAD_REQUEST) + .body(createError("ConditionalCheckFailedException")); + } + } + java.util.List lsiKeys = new java.util.ArrayList<>(); java.util.List gsiItems = new java.util.ArrayList<>(); @@ -234,10 +253,19 @@ private ResponseEntity handlePutItem(JsonNode body) throws IOException } String value = objectMapper.writeValueAsString(itemNode); + String returnValues = body.has("ReturnValues") ? body.get("ReturnValues").asText() : "NONE"; RpcReply reply = tcpClient.putItem(partitionKeyValue, tableName, sortKeyValue, value, lsiKeys, gsiItems); if (reply.success) { + if ("ALL_NEW".equals(returnValues)) { + RpcReply getReply = tcpClient.getItem(partitionKeyValue, tableName, sortKeyValue); + var resp = objectMapper.createObjectNode(); + if (getReply.success && getReply.value != null && !getReply.value.isEmpty()) { + resp.set("Attributes", objectMapper.readTree(getReply.value)); + } + return ResponseEntity.ok(resp); + } return ResponseEntity.ok(objectMapper.createObjectNode()); } else { return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR) @@ -325,6 +353,33 @@ private ResponseEntity handleDeleteItem(JsonNode body) throws IOExcept .body(createError("Could not extract partition key from Key")); } + // ConditionExpression check: evaluate before tombstone write + JsonNode condExpr = body.get("ConditionExpression"); + if (condExpr != null && !condExpr.isNull()) { + RpcReply preGet = tcpClient.getItem(partitionKeyValue, tableName, sortKeyValue); + JsonNode existingItem = null; + if (preGet.success && preGet.value != null && !preGet.value.isEmpty()) { + existingItem = objectMapper.readTree(preGet.value); + } else { + existingItem = objectMapper.createObjectNode(); + } + JsonNode condAttribNames = body.get("ExpressionAttributeNames"); + JsonNode condAttribValues = body.get("ExpressionAttributeValues"); + JsonNode condAst = compileExpression(condExpr.asText(), condAttribNames, condAttribValues); + if (condAst != null && !evaluateCondition(condAst, existingItem)) { + return ResponseEntity.status(HttpStatus.BAD_REQUEST) + .body(createError("ConditionalCheckFailedException")); + } + } + + // GET existing item BEFORE tombstone (for ReturnValues) + RpcReply preReply = tcpClient.getItem(partitionKeyValue, tableName, sortKeyValue); + JsonNode preItem = null; + if (preReply.success && preReply.value != null && !preReply.value.isEmpty()) { + preItem = objectMapper.readTree(preReply.value); + } + + String returnValues = body.has("ReturnValues") ? body.get("ReturnValues").asText() : "NONE"; RpcReply reply = tcpClient.putItem( partitionKeyValue, @@ -335,6 +390,11 @@ private ResponseEntity handleDeleteItem(JsonNode body) throws IOExcept new java.util.ArrayList<>()); if (reply.success) { + if ("ALL_OLD".equals(returnValues) && preItem != null) { + var resp = objectMapper.createObjectNode(); + resp.set("Attributes", preItem); + return ResponseEntity.ok(resp); + } return ResponseEntity.ok(objectMapper.createObjectNode()); } else { return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR) @@ -715,12 +775,32 @@ 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) - .body(createError("BatchWriteItem failed: " + reply.errorMsg)); + // Build UnprocessedKeys from failed items in conflicts + var unprocessed = responseNode.putObject("UnprocessedKeys"); + if (reply.conflicts != null) { + for (String conflictJson : reply.conflicts) { + JsonNode failedItem = objectMapper.readTree(conflictJson); + String tname = failedItem.get("table").asText(); + var tableUnproc = unprocessed.has(tname) + ? (com.fasterxml.jackson.databind.node.ObjectNode) unprocessed.get(tname) + : unprocessed.putObject(tname); + var writeReqs = tableUnproc.putArray("WriteRequest"); + var writeReq = writeReqs.addObject(); + if (failedItem.has("value") && !failedItem.get("value").asText().isEmpty()) { + var putReq = writeReq.putObject("PutRequest"); + var itemNode = putReq.putObject("Item"); + itemNode.put(failedItem.get("partitionKey").asText(), objectMapper.createObjectNode().put("S", failedItem.get("partitionKey").asText())); + } else { + var delReq = writeReq.putObject("DeleteRequest"); + var keyNode = delReq.putObject("Key"); + keyNode.put(failedItem.get("partitionKey").asText(), objectMapper.createObjectNode().put("S", failedItem.get("partitionKey").asText())); + } + } + } + responseNode.put("ConsumedCapacity", objectMapper.createObjectNode()); } return ResponseEntity.ok(responseNode); } @@ -1276,6 +1356,103 @@ else if (c == ',' && parenDepth == 0) { return fallback; } + private boolean evaluateCondition(JsonNode ast, JsonNode item) { + if (ast == null || item == null) return true; + String type = ast.has("type") ? ast.get("type").asText() : ""; + + switch (type) { + case "and": { + JsonNode left = ast.get("left"); + JsonNode right = ast.get("right"); + return evaluateCondition(left, item) && evaluateCondition(right, item); + } + case "or": { + JsonNode left = ast.get("left"); + JsonNode right = ast.get("right"); + return evaluateCondition(left, item) || evaluateCondition(right, item); + } + case "not": { + JsonNode exprNode = ast.get("expr"); + return !evaluateCondition(exprNode, item); + } + case "function": { + String name = ast.has("name") ? ast.get("name").asText() : ""; + String path = ast.has("path") ? ast.get("path").asText() : ""; + if ("attribute_exists".equals(name)) { + return item.has(path); + } + if ("attribute_not_exists".equals(name)) { + return !item.has(path); + } + if ("begins_with".equals(name) || "contains".equals(name)) { + JsonNode valNode = ast.get("value"); + String itemVal = extractScalarValue(item, path); + String compareVal = extractScalarValue(valNode, null); + if ("begins_with".equals(name)) { + return itemVal.startsWith(compareVal); + } + if ("contains".equals(name)) { + return itemVal.contains(compareVal); + } + } + return true; + } + case "comparison": { + String op = ast.has("op") ? ast.get("op").asText() : "="; + String path = ast.has("path") ? ast.get("path").asText() : ""; + JsonNode valNode = ast.get("value"); + String itemVal = extractScalarValue(item, path); + String compareVal = extractScalarValue(valNode, null); + switch (op) { + case "=": return itemVal.equals(compareVal); + case "<>": return !itemVal.equals(compareVal); + case ">": return itemVal.compareTo(compareVal) > 0; + case "<": return itemVal.compareTo(compareVal) < 0; + case ">=": return itemVal.compareTo(compareVal) >= 0; + case "<=": return itemVal.compareTo(compareVal) <= 0; + default: return false; + } + } + case "between": { + String path = ast.has("path") ? ast.get("path").asText() : ""; + JsonNode v1Node = ast.get("value1"); + JsonNode v2Node = ast.get("value2"); + String itemVal = extractScalarValue(item, path); + String val1 = extractScalarValue(v1Node, null); + String val2 = extractScalarValue(v2Node, null); + return itemVal.compareTo(val1) >= 0 && itemVal.compareTo(val2) <= 0; + } + case "in": { + String path = ast.has("path") ? ast.get("path").asText() : ""; + JsonNode valuesNode = ast.get("values"); + String itemVal = extractScalarValue(item, path); + if (valuesNode != null && valuesNode.isArray()) { + for (JsonNode v : valuesNode) { + if (itemVal.equals(extractScalarValue(v, null))) { + return true; + } + } + } + return false; + } + default: + return true; + } + } + + private String extractScalarValue(JsonNode node, String path) { + if (node == null) return ""; + if (path != null && node.has(path)) { + node = node.get(path); + } + if (node.has("S")) return node.get("S").asText(); + if (node.has("N")) return node.get("N").asText(); + if (node.has("BOOL")) return node.get("BOOL").asText(); + if (node.isTextual()) return node.asText(); + if (node.isNumber()) return node.asText(); + return ""; + } + private int findTopLevelOperator(String expr, String op) { int balance = 0; int opLen = op.length(); From 82d74778d7086e6c93805fcf6429db35b54c1179 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: Sat, 13 Jun 2026 16:37:29 +0300 Subject: [PATCH 2/3] feat(storage-engine): track per-item failures in BatchWriteItem - BatchWriteItem now collects failed item partition keys in reply_conflicts - Each failed item is serialized as JSON for UnprocessedKeys reconstruction - reply_success is false when any item fails --- storage-engine/src/main.zig | 21 ++++++++++++++++++--- 1 file changed, 18 insertions(+), 3 deletions(-) diff --git a/storage-engine/src/main.zig b/storage-engine/src/main.zig index 52304e7..112a00e 100644 --- a/storage-engine/src/main.zig +++ b/storage-engine/src/main.zig @@ -740,13 +740,28 @@ fn rpcHandler( for (list.items.items) |item| { // Empty value means delete (tombstone), non-empty means put + var ok = false; if (item.value.len > 0) { - _ = s_ctx.engine.put(item.partition_key, item.value, &.{}, item.table, item.sort_key); + ok = 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); + ok = s_ctx.engine.delete(item.partition_key, &.{}, item.table, item.sort_key); + } + if (!ok) { + // Serialize failed item as JSON for UnprocessedKeys + var failed_json = ArrayList(u8).init(s_ctx.allocator); + failed_json.appendSlice("{\"table\":\"") catch continue; + failed_json.appendSlice(item.table) catch continue; + failed_json.appendSlice("\",\"partitionKey\":\"") catch continue; + failed_json.appendSlice(item.partition_key) catch continue; + failed_json.appendSlice("\",\"sortKey\":\"") catch continue; + failed_json.appendSlice(item.sort_key) catch continue; + failed_json.appendSlice("\",\"value\":\"") catch continue; + failed_json.appendSlice(item.value) catch continue; + failed_json.appendSlice("\"}") catch continue; + reply_conflicts.append(failed_json.toOwnedSlice() catch &.{}) catch {}; } } - reply_success = true; + reply_success = (reply_conflicts.items.len == 0); } } else if (rpc_type == 15) { // TransactGetItems var count: usize = 0; From 3e2fe4c49009a57889e8349c2bbd91b3c4ddc407 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: Sun, 14 Jun 2026 14:31:29 +0300 Subject: [PATCH 3/3] fix(api): address PR #7 review findings - Fix Zig JSON serialization: use std.json.fmt instead of manual concat to properly escape special characters in failed item JSON - BatchWriteItem UnprocessedKeys: include sortKey in reconstructed DeleteRequest Key for composite key tables - evaluateCondition: guard null children in AND/OR/NOT nodes - Add hasPath() helper for nested attribute paths (address.city) and update extractScalarValue() to traverse dot-notation paths --- .../controller/DynamoController.java | 32 +++++++++++++++++-- storage-engine/src/main.zig | 22 +++++++------ 2 files changed, 43 insertions(+), 11 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 36e09a2..b9d49c0 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 @@ -793,10 +793,16 @@ private ResponseEntity handleBatchWriteItem(JsonNode body) throws IOEx var putReq = writeReq.putObject("PutRequest"); var itemNode = putReq.putObject("Item"); itemNode.put(failedItem.get("partitionKey").asText(), objectMapper.createObjectNode().put("S", failedItem.get("partitionKey").asText())); + if (failedItem.has("sortKey") && !failedItem.get("sortKey").asText().isEmpty()) { + itemNode.put(failedItem.get("sortKey").asText(), objectMapper.createObjectNode().put("S", failedItem.get("sortKey").asText())); + } } else { var delReq = writeReq.putObject("DeleteRequest"); var keyNode = delReq.putObject("Key"); keyNode.put(failedItem.get("partitionKey").asText(), objectMapper.createObjectNode().put("S", failedItem.get("partitionKey").asText())); + if (failedItem.has("sortKey") && !failedItem.get("sortKey").asText().isEmpty()) { + keyNode.put(failedItem.get("sortKey").asText(), objectMapper.createObjectNode().put("S", failedItem.get("sortKey").asText())); + } } } } @@ -1364,25 +1370,30 @@ private boolean evaluateCondition(JsonNode ast, JsonNode item) { case "and": { JsonNode left = ast.get("left"); JsonNode right = ast.get("right"); + if (left == null || right == null) return false; return evaluateCondition(left, item) && evaluateCondition(right, item); } case "or": { JsonNode left = ast.get("left"); JsonNode right = ast.get("right"); + if (left == null && right == null) return false; + if (left == null) return evaluateCondition(right, item); + if (right == null) return evaluateCondition(left, item); return evaluateCondition(left, item) || evaluateCondition(right, item); } case "not": { JsonNode exprNode = ast.get("expr"); + if (exprNode == null) return false; return !evaluateCondition(exprNode, item); } case "function": { String name = ast.has("name") ? ast.get("name").asText() : ""; String path = ast.has("path") ? ast.get("path").asText() : ""; if ("attribute_exists".equals(name)) { - return item.has(path); + return hasPath(item, path); } if ("attribute_not_exists".equals(name)) { - return !item.has(path); + return !hasPath(item, path); } if ("begins_with".equals(name) || "contains".equals(name)) { JsonNode valNode = ast.get("value"); @@ -1442,6 +1453,13 @@ private boolean evaluateCondition(JsonNode ast, JsonNode item) { private String extractScalarValue(JsonNode node, String path) { if (node == null) return ""; + if (path != null && path.contains(".")) { + int dot = path.indexOf('.'); + String first = path.substring(0, dot); + String rest = path.substring(dot + 1); + if (!node.has(first)) return ""; + return extractScalarValue(node.get(first), rest); + } if (path != null && node.has(path)) { node = node.get(path); } @@ -1453,6 +1471,16 @@ private String extractScalarValue(JsonNode node, String path) { return ""; } + private boolean hasPath(JsonNode node, String path) { + if (path == null || path.isEmpty()) return false; + if (!path.contains(".")) return node.has(path); + int dot = path.indexOf('.'); + String first = path.substring(0, dot); + String rest = path.substring(dot + 1); + if (!node.has(first)) return false; + return hasPath(node.get(first), rest); + } + private int findTopLevelOperator(String expr, String op) { int balance = 0; int opLen = op.length(); diff --git a/storage-engine/src/main.zig b/storage-engine/src/main.zig index 112a00e..b2a7aab 100644 --- a/storage-engine/src/main.zig +++ b/storage-engine/src/main.zig @@ -748,16 +748,20 @@ fn rpcHandler( } if (!ok) { // Serialize failed item as JSON for UnprocessedKeys + const FailedItemJson = struct { + table: []const u8, + partitionKey: []const u8, + sortKey: []const u8, + value: []const u8, + }; + const failed_item = FailedItemJson{ + .table = item.table, + .partitionKey = item.partition_key, + .sortKey = item.sort_key, + .value = item.value, + }; var failed_json = ArrayList(u8).init(s_ctx.allocator); - failed_json.appendSlice("{\"table\":\"") catch continue; - failed_json.appendSlice(item.table) catch continue; - failed_json.appendSlice("\",\"partitionKey\":\"") catch continue; - failed_json.appendSlice(item.partition_key) catch continue; - failed_json.appendSlice("\",\"sortKey\":\"") catch continue; - failed_json.appendSlice(item.sort_key) catch continue; - failed_json.appendSlice("\",\"value\":\"") catch continue; - failed_json.appendSlice(item.value) catch continue; - failed_json.appendSlice("\"}") catch continue; + failed_json.print("{}", .{std.json.fmt(failed_item, .{})}) catch continue; reply_conflicts.append(failed_json.toOwnedSlice() catch &.{}) catch {}; } }