Skip to content

feat(api): ReturnValues and ConditionExpression for PutItem/DeleteItem #7

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
poyrazK wants to merge 3 commits into
base: main
Choose a base branch
from
Open

feat(api): ReturnValues and ConditionExpression for PutItem/DeleteItem #7

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f1b6ab8
feat(api): implement ReturnValues and ConditionExpression for PutItem…
poyrazK Jun 13, 2026
82d7477
feat(storage-engine): track per-item failures in BatchWriteItem
poyrazK Jun 13, 2026
3e2fe4c
fix(api): address PR #7 review findings
poyrazK Jun 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
211 changes: 208 additions & 3 deletions control-plane/src/main/java/com/clouddb/controlplane/controller/DynamoController.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -192,6 +192,25 @@ private ResponseEntity<JsonNode> 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<String> lsiKeys = new java.util.ArrayList<>();
java.util.List<com.clouddb.controlplane.client.ItemArgs.GsiItem> gsiItems =
new java.util.ArrayList<>();
Expand Down Expand Up @@ -234,10 +253,19 @@ private ResponseEntity<JsonNode> 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)
Expand Down Expand Up @@ -325,6 +353,33 @@ private ResponseEntity<JsonNode> 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,
Expand All @@ -335,6 +390,11 @@ private ResponseEntity<JsonNode> 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)
Expand Down Expand Up @@ -715,12 +775,38 @@ private ResponseEntity<JsonNode> 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()));
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()));
}
}
}
}
responseNode.put("ConsumedCapacity", objectMapper.createObjectNode());
}
return ResponseEntity.ok(responseNode);
}
Expand Down Expand Up @@ -1276,6 +1362,125 @@ 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");
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 hasPath(item, path);
}
if ("attribute_not_exists".equals(name)) {
return !hasPath(item, 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 && 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);
}
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 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();
Expand Down
25 changes: 22 additions & 3 deletions storage-engine/src/main.zig
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -740,13 +740,32 @@ 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
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.print("{}", .{std.json.fmt(failed_item, .{})}) 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;
Expand Down
Loading