biondizzle/jormun-db
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

make batch operations work

Browse Source
This commit is contained in:
biondizzle
2026-02-16 00:18:20 -05:00
parent c6a78ca054
commit ffd3eda63c
6 changed files with 877 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

551
main.odin
View File

@@ -102,6 +102,10 @@ handle_dynamodb_request :: proc(ctx: rawptr, request: ^HTTP_Request, request_all
handle_query(engine, request, &response)
case .Scan:
handle_scan(engine, request, &response)
case .BatchWriteItem:
handle_batch_write_item(engine, request, &response)
case .BatchGetItem:
handle_batch_get_item(engine, request, &response)
case .Unknown:
return make_error_response(&response, .ValidationException, "Unknown operation")
case:
@@ -300,6 +304,95 @@ handle_put_item :: proc(engine: ^dynamodb.Storage_Engine, request: ^HTTP_Request
}
defer dynamodb.item_destroy(&item)
// ---- ConditionExpression evaluation ----
_, has_condition := dynamodb.parse_condition_expression_string(request.body)
if has_condition {
// Parse shared expression attributes
attr_names := dynamodb.parse_expression_attribute_names(request.body)
defer {
if names, has_names := attr_names.?; has_names {
for k, v in names {
delete(k)
delete(v)
}
names_copy := names
delete(names_copy)
}
}
attr_values, vals_ok := dynamodb.parse_expression_attribute_values(request.body)
if !vals_ok {
make_error_response(response, .ValidationException, "Invalid ExpressionAttributeValues")
return
}
defer {
for k, v in attr_values {
delete(k)
v_copy := v
dynamodb.attr_value_destroy(&v_copy)
}
delete(attr_values)
}
// Fetch existing item to evaluate condition against
key_item, key_ok := dynamodb.parse_key_from_request(request.body)
existing_item: Maybe(dynamodb.Item)
if !key_ok {
// If no explicit Key field, extract key from Item
// (PutItem doesn't have a Key field — the key is in the Item itself)
existing_maybe, get_err := dynamodb.get_item(engine, table_name, item)
if get_err != .None && get_err != .Table_Not_Found {
// Table not found is handled by put_item below
if get_err == .Missing_Key_Attribute || get_err == .Invalid_Key {
handle_storage_error(response, get_err)
return
}
}
existing_item = existing_maybe
} else {
defer dynamodb.item_destroy(&key_item)
existing_maybe, get_err := dynamodb.get_item(engine, table_name, key_item)
if get_err != .None && get_err != .Table_Not_Found {
if get_err == .Missing_Key_Attribute || get_err == .Invalid_Key {
handle_storage_error(response, get_err)
return
}
}
existing_item = existing_maybe
}
defer {
if ex, has_ex := existing_item.?; has_ex {
ex_copy := ex
dynamodb.item_destroy(&ex_copy)
}
}
// Evaluate condition
cond_result := dynamodb.evaluate_condition_expression(
request.body, existing_item, attr_names, attr_values,
)
switch cond_result {
case .Failed:
make_error_response(
response, .ConditionalCheckFailedException,
"The conditional request failed",
)
return
case .Parse_Error:
make_error_response(
response, .ValidationException,
"Invalid ConditionExpression",
)
return
case .Passed:
// Continue with put
}
}
// ---- Execute PutItem ----
err := dynamodb.put_item(engine, table_name, item)
if err != .None {
handle_storage_error(response, err)
@@ -353,6 +446,70 @@ handle_delete_item :: proc(engine: ^dynamodb.Storage_Engine, request: ^HTTP_Requ
}
defer dynamodb.item_destroy(&key)
// ---- ConditionExpression evaluation ----
_, has_condition := dynamodb.parse_condition_expression_string(request.body)
if has_condition {
attr_names := dynamodb.parse_expression_attribute_names(request.body)
defer {
if names, has_names := attr_names.?; has_names {
for k, v in names {
delete(k)
delete(v)
}
names_copy := names
delete(names_copy)
}
}
attr_values, vals_ok := dynamodb.parse_expression_attribute_values(request.body)
if !vals_ok {
make_error_response(response, .ValidationException, "Invalid ExpressionAttributeValues")
return
}
defer {
for k, v in attr_values {
delete(k)
v_copy := v
dynamodb.attr_value_destroy(&v_copy)
}
delete(attr_values)
}
// Fetch existing item
existing_item, get_err := dynamodb.get_item(engine, table_name, key)
if get_err != .None && get_err != .Table_Not_Found {
if get_err == .Missing_Key_Attribute || get_err == .Invalid_Key {
handle_storage_error(response, get_err)
return
}
}
defer {
if ex, has_ex := existing_item.?; has_ex {
ex_copy := ex
dynamodb.item_destroy(&ex_copy)
}
}
cond_result := dynamodb.evaluate_condition_expression(
request.body, existing_item, attr_names, attr_values,
)
switch cond_result {
case .Failed:
make_error_response(
response, .ConditionalCheckFailedException,
"The conditional request failed",
)
return
case .Parse_Error:
make_error_response(response, .ValidationException, "Invalid ConditionExpression")
return
case .Passed:
// Continue with delete
}
}
// ---- Execute DeleteItem ----
err := dynamodb.delete_item(engine, table_name, key)
if err != .None {
handle_storage_error(response, err)
@@ -413,6 +570,43 @@ handle_update_item :: proc(engine: ^dynamodb.Storage_Engine, request: ^HTTP_Requ
delete(attr_values)
}
// ---- ConditionExpression evaluation ----
_, has_condition := dynamodb.parse_condition_expression_string(request.body)
if has_condition {
// Fetch existing item to evaluate condition against
existing_item, get_err := dynamodb.get_item(engine, table_name, key_item)
if get_err != .None && get_err != .Table_Not_Found {
if get_err == .Missing_Key_Attribute || get_err == .Invalid_Key {
handle_storage_error(response, get_err)
return
}
}
defer {
if ex, has_ex := existing_item.?; has_ex {
ex_copy := ex
dynamodb.item_destroy(&ex_copy)
}
}
cond_result := dynamodb.evaluate_condition_expression(
request.body, existing_item, attr_names, attr_values,
)
switch cond_result {
case .Failed:
make_error_response(
response, .ConditionalCheckFailedException,
"The conditional request failed",
)
return
case .Parse_Error:
make_error_response(response, .ValidationException, "Invalid ConditionExpression")
return
case .Passed:
// Continue with update
}
}
// Parse update plan
plan, plan_ok := dynamodb.parse_update_expression(update_expr, attr_names, attr_values)
if !plan_ok {
@@ -462,8 +656,6 @@ handle_update_item :: proc(engine: ^dynamodb.Storage_Engine, request: ^HTTP_Requ
}
case "UPDATED_NEW":
// Return only the attributes that were updated (in the new item)
// For simplicity, return the full new item (DynamoDB returns affected attributes)
if new_val, has := new_item.?; has {
item_json := dynamodb.serialize_item(new_val)
resp := fmt.aprintf(`{"Attributes":%s}`, item_json)
@@ -487,6 +679,361 @@ handle_update_item :: proc(engine: ^dynamodb.Storage_Engine, request: ^HTTP_Requ
}
}
handle_batch_write_item :: proc(engine: ^dynamodb.Storage_Engine, request: ^HTTP_Request, response: ^HTTP_Response) {
data, parse_err := json.parse(request.body, allocator = context.allocator)
if parse_err != nil {
make_error_response(response, .SerializationException, "Invalid JSON")
return
}
defer json.destroy_value(data)
root, root_ok := data.(json.Object)
if !root_ok {
make_error_response(response, .SerializationException, "Request must be an object")
return
}
request_items_val, found := root["RequestItems"]
if !found {
make_error_response(response, .ValidationException, "Missing RequestItems")
return
}
request_items, ri_ok := request_items_val.(json.Object)
if !ri_ok {
make_error_response(response, .ValidationException, "RequestItems must be an object")
return
}
// Count total operations for limit enforcement
total_ops := 0
table_requests := make([dynamic]dynamodb.Batch_Write_Table_Request)
defer {
for &tr in table_requests {
for &req in tr.requests {
dynamodb.item_destroy(&req.item)
}
delete(tr.requests)
}
delete(table_requests)
}
for table_name, table_val in request_items {
table_array, arr_ok := table_val.(json.Array)
if !arr_ok {
make_error_response(response, .ValidationException,
fmt.tprintf("RequestItems for table '%s' must be an array", table_name))
return
}
requests := make([dynamic]dynamodb.Write_Request)
for elem in table_array {
elem_obj, elem_ok := elem.(json.Object)
if !elem_ok {
for &r in requests {
dynamodb.item_destroy(&r.item)
}
delete(requests)
make_error_response(response, .ValidationException, "Each write request must be an object")
return
}
// Check for PutRequest
if put_val, has_put := elem_obj["PutRequest"]; has_put {
put_obj, put_ok := put_val.(json.Object)
if !put_ok {
for &r in requests {
dynamodb.item_destroy(&r.item)
}
delete(requests)
make_error_response(response, .ValidationException, "PutRequest must be an object")
return
}
item_val, item_found := put_obj["Item"]
if !item_found {
for &r in requests {
dynamodb.item_destroy(&r.item)
}
delete(requests)
make_error_response(response, .ValidationException, "PutRequest missing Item")
return
}
item, item_ok := dynamodb.parse_item_from_value(item_val)
if !item_ok {
for &r in requests {
dynamodb.item_destroy(&r.item)
}
delete(requests)
make_error_response(response, .ValidationException, "Invalid Item in PutRequest")
return
}
append(&requests, dynamodb.Write_Request{type = .Put, item = item})
total_ops += 1
continue
}
// Check for DeleteRequest
if del_val, has_del := elem_obj["DeleteRequest"]; has_del {
del_obj, del_ok := del_val.(json.Object)
if !del_ok {
for &r in requests {
dynamodb.item_destroy(&r.item)
}
delete(requests)
make_error_response(response, .ValidationException, "DeleteRequest must be an object")
return
}
key_val, key_found := del_obj["Key"]
if !key_found {
for &r in requests {
dynamodb.item_destroy(&r.item)
}
delete(requests)
make_error_response(response, .ValidationException, "DeleteRequest missing Key")
return
}
key_item, key_ok := dynamodb.parse_item_from_value(key_val)
if !key_ok {
for &r in requests {
dynamodb.item_destroy(&r.item)
}
delete(requests)
make_error_response(response, .ValidationException, "Invalid Key in DeleteRequest")
return
}
append(&requests, dynamodb.Write_Request{type = .Delete, item = key_item})
total_ops += 1
continue
}
// Neither PutRequest nor DeleteRequest
for &r in requests {
dynamodb.item_destroy(&r.item)
}
delete(requests)
make_error_response(response, .ValidationException,
"Each write request must contain PutRequest or DeleteRequest")
return
}
append(&table_requests, dynamodb.Batch_Write_Table_Request{
table_name = string(table_name),
requests = requests[:],
})
}
// Enforce 25-operation limit
if total_ops > 25 {
make_error_response(response, .ValidationException,
"Too many items requested for the BatchWriteItem call (max 25)")
return
}
if total_ops == 0 {
make_error_response(response, .ValidationException,
"RequestItems must contain at least one table with at least one request")
return
}
// Execute batch
result, err := dynamodb.batch_write_item(engine, table_requests[:])
if err != .None {
handle_storage_error(response, err)
return
}
defer dynamodb.batch_write_result_destroy(&result)
// Build response
builder := strings.builder_make()
strings.write_string(&builder, `{"UnprocessedItems":{`)
unprocessed_count := 0
for table_req, ti in result.unprocessed {
if ti > 0 {
strings.write_string(&builder, ",")
}
fmt.sbprintf(&builder, `"%s":[`, table_req.table_name)
for req, ri in table_req.requests {
if ri > 0 {
strings.write_string(&builder, ",")
}
item_json := dynamodb.serialize_item(req.item)
switch req.type {
case .Put:
fmt.sbprintf(&builder, `{"PutRequest":{"Item":%s}}`, item_json)
case .Delete:
fmt.sbprintf(&builder, `{"DeleteRequest":{"Key":%s}}`, item_json)
}
}
strings.write_string(&builder, "]")
unprocessed_count += len(table_req.requests)
}
strings.write_string(&builder, "}}")
resp_body := strings.to_string(builder)
response_set_body(response, transmute([]byte)resp_body)
}
handle_batch_get_item :: proc(engine: ^dynamodb.Storage_Engine, request: ^HTTP_Request, response: ^HTTP_Response) {
data, parse_err := json.parse(request.body, allocator = context.allocator)
if parse_err != nil {
make_error_response(response, .SerializationException, "Invalid JSON")
return
}
defer json.destroy_value(data)
root, root_ok := data.(json.Object)
if !root_ok {
make_error_response(response, .SerializationException, "Request must be an object")
return
}
request_items_val, found := root["RequestItems"]
if !found {
make_error_response(response, .ValidationException, "Missing RequestItems")
return
}
request_items, ri_ok := request_items_val.(json.Object)
if !ri_ok {
make_error_response(response, .ValidationException, "RequestItems must be an object")
return
}
total_keys := 0
table_requests := make([dynamic]dynamodb.Batch_Get_Table_Request)
defer {
for &tr in table_requests {
for &key in tr.keys {
dynamodb.item_destroy(&key)
}
delete(tr.keys)
}
delete(table_requests)
}
for table_name, table_val in request_items {
table_obj, obj_ok := table_val.(json.Object)
if !obj_ok {
make_error_response(response, .ValidationException,
fmt.tprintf("RequestItems for table '%s' must be an object", table_name))
return
}
keys_val, keys_found := table_obj["Keys"]
if !keys_found {
make_error_response(response, .ValidationException,
fmt.tprintf("Missing Keys for table '%s'", table_name))
return
}
keys_array, keys_ok := keys_val.(json.Array)
if !keys_ok {
make_error_response(response, .ValidationException,
fmt.tprintf("Keys for table '%s' must be an array", table_name))
return
}
keys := make([dynamic]dynamodb.Item)
for key_val in keys_array {
key_item, key_ok := dynamodb.parse_item_from_value(key_val)
if !key_ok {
for &k in keys {
dynamodb.item_destroy(&k)
}
delete(keys)
make_error_response(response, .ValidationException, "Invalid key in BatchGetItem")
return
}
append(&keys, key_item)
total_keys += 1
}
append(&table_requests, dynamodb.Batch_Get_Table_Request{
table_name = string(table_name),
keys = keys[:],
})
}
// Enforce 100-key limit
if total_keys > 100 {
make_error_response(response, .ValidationException,
"Too many items requested for the BatchGetItem call (max 100)")
return
}
if total_keys == 0 {
make_error_response(response, .ValidationException,
"RequestItems must contain at least one table with at least one key")
return
}
// Execute batch get
result, err := dynamodb.batch_get_item(engine, table_requests[:])
if err != .None {
handle_storage_error(response, err)
return
}
defer dynamodb.batch_get_result_destroy(&result)
// Build response
builder := strings.builder_make()
strings.write_string(&builder, `{"Responses":{`)
for table_result, ti in result.responses {
if ti > 0 {
strings.write_string(&builder, ",")
}
fmt.sbprintf(&builder, `"%s":[`, table_result.table_name)
for item, ii in table_result.items {
if ii > 0 {
strings.write_string(&builder, ",")
}
item_json := dynamodb.serialize_item(item)
strings.write_string(&builder, item_json)
}
strings.write_string(&builder, "]")
}
strings.write_string(&builder, `},"UnprocessedKeys":{`)
for table_req, ti in result.unprocessed_keys {
if ti > 0 {
strings.write_string(&builder, ",")
}
fmt.sbprintf(&builder, `"%s":{"Keys":[`, table_req.table_name)
for key, ki in table_req.keys {
if ki > 0 {
strings.write_string(&builder, ",")
}
key_json := dynamodb.serialize_item(key)
strings.write_string(&builder, key_json)
}
strings.write_string(&builder, "]}")
}
strings.write_string(&builder, "}}")
resp_body := strings.to_string(builder)
response_set_body(response, transmute([]byte)resp_body)
}
// ============================================================================
// Query and Scan Operations
// ============================================================================