dynoxide/actions/
batch_write_item.rs

1use crate::actions::helpers;
2use crate::errors::{DynoxideError, Result};
3use crate::storage_backend::StorageBackend;
4use crate::types::{self, AttributeValue, Item};
5use serde::{Deserialize, Serialize};
6use std::collections::HashMap;
7
8#[derive(Debug, Default, Deserialize)]
9pub struct BatchWriteItemRequest {
10    #[serde(rename = "RequestItems")]
11    pub request_items: HashMap<String, Vec<WriteRequest>>,
12    #[serde(rename = "ReturnConsumedCapacity", default)]
13    pub return_consumed_capacity: Option<String>,
14    #[serde(rename = "ReturnItemCollectionMetrics", default)]
15    pub return_item_collection_metrics: Option<String>,
16}
17
18#[derive(Debug, Default, Deserialize)]
19pub struct WriteRequest {
20    #[serde(rename = "PutRequest", default)]
21    pub put_request: Option<PutRequest>,
22    #[serde(rename = "DeleteRequest", default)]
23    pub delete_request: Option<DeleteRequest>,
24}
25
26#[derive(Debug, Default, Deserialize)]
27pub struct PutRequest {
28    #[serde(rename = "Item")]
29    pub item: Item,
30}
31
32#[derive(Debug, Default, Deserialize)]
33pub struct DeleteRequest {
34    #[serde(rename = "Key")]
35    pub key: HashMap<String, AttributeValue>,
36}
37
38#[derive(Debug, Default, Serialize)]
39pub struct BatchWriteItemResponse {
40    #[serde(rename = "UnprocessedItems")]
41    pub unprocessed_items: HashMap<String, serde_json::Value>,
42    #[serde(rename = "ConsumedCapacity", skip_serializing_if = "Option::is_none")]
43    pub consumed_capacity: Option<Vec<crate::types::ConsumedCapacity>>,
44    #[serde(
45        rename = "ItemCollectionMetrics",
46        skip_serializing_if = "Option::is_none"
47    )]
48    pub item_collection_metrics: Option<HashMap<String, Vec<crate::types::ItemCollectionMetrics>>>,
49}
50
51/// Fold one item's per-index units into the running per-table totals. A batch
52/// reports the sum across its items, and an index no item in the batch touched
53/// stays absent rather than appearing as a zero.
54fn accumulate_index_units(
55    totals: &mut HashMap<String, HashMap<String, f64>>,
56    table_name: &str,
57    units: &HashMap<String, f64>,
58) {
59    if units.is_empty() {
60        return;
61    }
62    let table_entry = totals.entry(table_name.to_string()).or_default();
63    for (index_name, units) in units {
64        *table_entry.entry(index_name.clone()).or_insert(0.0) += units;
65    }
66}
67
68pub async fn execute<S: StorageBackend>(
69    storage: &S,
70    mut request: BatchWriteItemRequest,
71) -> Result<BatchWriteItemResponse> {
72    const MAX_REQUEST_SIZE: usize = 16 * 1024 * 1024; // 16MB
73
74    // Validate RequestItems is not empty.
75    // AWS routes the empty-map case through a separate parameter-required path
76    // rather than the standard "N validation errors detected" envelope.
77    //
78    // Do not align this with BatchGetItem, which now answers the standard
79    // envelope. The two have diverged: as of the 2026-09-02 sweep 31 of the 33
80    // regions observed still return this sentence for BatchWriteItem, against
81    // 1 of 33 for BatchGetItem. Matching them would break the common case here.
82    if request.request_items.is_empty() {
83        return Err(DynoxideError::ValidationException(
84            "The requestItems parameter is required for BatchWriteItem".to_string(),
85        ));
86    }
87
88    // Validate each table entry has at least one write request
89    for (table_name, wrs) in &request.request_items {
90        if wrs.is_empty() {
91            return Err(DynoxideError::ValidationException(format!(
92                "1 validation error detected: Value at 'requestItems.{table_name}.member' failed to satisfy constraint: Member must have length greater than or equal to 1"
93            )));
94        }
95    }
96
97    // Validate table name format for all tables before checking existence
98    for table_name in request.request_items.keys() {
99        crate::validation::validate_table_name(table_name)?;
100    }
101
102    // Validate total request count.
103    // AWS surfaces this as the standard "1 validation error detected" envelope
104    // and echoes the WriteRequest list inside `Value '{<table>=[<dump>]}'`. The
105    // conformance suite anchors a regex around the envelope and the constraint
106    // phrase but leaves the dump body unconstrained (because the AWS SDK's
107    // Java-toString shape adds new AttributeValue fields over time). We emit
108    // the table name verbatim and a Rust Debug dump of the WriteRequests so
109    // the envelope matches without coupling to a specific SDK version. If a
110    // future suite tightens the regex to pin the dump exactly, this site
111    // will need a follow-up change.
112    let total_requests: usize = request.request_items.values().map(|v| v.len()).sum();
113    if total_requests > 25 {
114        let empty: Vec<WriteRequest> = Vec::new();
115        let (table_name, requests) = request
116            .request_items
117            .iter()
118            .max_by_key(|(_, v)| v.len())
119            .map(|(name, v)| (name.as_str(), v))
120            .unwrap_or(("", &empty));
121        let dump = format!("{requests:?}");
122        return Err(DynoxideError::ValidationException(format!(
123            "1 validation error detected: Value '{{{table_name}=[{dump}]}}' at 'requestItems' failed to satisfy constraint: Map value must satisfy constraint: [Member must have length less than or equal to 25, Member must have length greater than or equal to 1]"
124        )));
125    }
126
127    // --- Pre-table validations ---
128    // DynamoDB validates attribute values, item size, and empty write requests
129    // BEFORE checking table existence.
130    for write_requests in request.request_items.values() {
131        for wr in write_requests {
132            if wr.put_request.is_none() && wr.delete_request.is_none() {
133                return Err(DynoxideError::ValidationException(
134                    "Supplied AttributeValue has more than one datatypes set, must contain exactly one of the supported datatypes".to_string(),
135                ));
136            }
137            if let Some(ref put_req) = wr.put_request {
138                // Validate attribute values (empty strings, empty sets, invalid numbers)
139                crate::validation::validate_item_attribute_values(&put_req.item)?;
140
141                // Validate item size before table lookup
142                let size = types::item_size(&put_req.item);
143                if size > types::MAX_ITEM_SIZE {
144                    return Err(DynoxideError::ValidationException(
145                        "Item size has exceeded the maximum allowed size".to_string(),
146                    ));
147                }
148            }
149            if let Some(ref del_req) = wr.delete_request {
150                crate::validation::validate_item_attribute_values(&del_req.key)?;
151            }
152        }
153    }
154
155    // Validate aggregate request size
156    let total_size: usize = request
157        .request_items
158        .values()
159        .flat_map(|wrs| wrs.iter())
160        .map(|wr| {
161            if let Some(ref put_req) = wr.put_request {
162                types::item_size(&put_req.item)
163            } else if let Some(ref del_req) = wr.delete_request {
164                types::item_size(&del_req.key)
165            } else {
166                0
167            }
168        })
169        .sum();
170    if total_size > MAX_REQUEST_SIZE {
171        return Err(DynoxideError::ValidationException(
172            "Item collection too large: aggregate size of items in BatchWriteItem exceeds 16MB limit".to_string(),
173        ));
174    }
175
176    // Validate: no duplicate keys across all operations
177    {
178        let mut seen_keys: std::collections::HashSet<(String, String, String)> =
179            std::collections::HashSet::new();
180        for (table_name, write_requests) in &request.request_items {
181            let meta = helpers::require_table_for_item_op(storage, table_name).await?;
182            let key_schema = helpers::parse_key_schema(&meta)?;
183            for wr in write_requests {
184                // Validate keys BEFORE extract_key_strings: that helper returns
185                // InternalServerError (HTTP 500) for a missing partition or sort
186                // key, but a key-less request is client input and must surface as
187                // a 400 ValidationException. Mirrors put_item.rs, which validates
188                // before extracting.
189                let key_item = if let Some(ref put) = wr.put_request {
190                    helpers::validate_item_keys_for_batch(&put.item, &key_schema, &meta)?;
191                    &put.item
192                } else if let Some(ref del) = wr.delete_request {
193                    helpers::validate_key_only(&del.key, &key_schema)?;
194                    &del.key
195                } else {
196                    continue;
197                };
198                let (pk, sk) = helpers::extract_key_strings(key_item, &key_schema)?;
199                let key = (table_name.clone(), pk, sk);
200                if !seen_keys.insert(key) {
201                    return Err(DynoxideError::ValidationException(
202                        "Provided list of item keys contains duplicates".to_string(),
203                    ));
204                }
205            }
206        }
207    }
208
209    // Track per-table secondary index capacity and affected partition keys for
210    // deferred metrics
211    let mut table_gsi_units: HashMap<String, HashMap<String, f64>> = HashMap::new();
212    let mut table_lsi_units: HashMap<String, HashMap<String, f64>> = HashMap::new();
213    // Vector replication is billed in bytes, so it accumulates on its own map
214    // and never joins the unit totals.
215    let mut table_vector_bytes: HashMap<String, HashMap<String, f64>> = HashMap::new();
216    // Track per-table WCU (table-level, excludes GSI)
217    let mut table_wcu: HashMap<String, f64> = HashMap::new();
218    // Collect unique (table, pk_str, pk_attr, pk_value) for deferred metrics computation
219    let mut affected_partitions: Vec<(String, String, String, AttributeValue)> = Vec::new();
220
221    for (table_name, write_requests) in &mut request.request_items {
222        let meta = helpers::require_table_for_item_op(storage, table_name).await?;
223        let key_schema = helpers::parse_key_schema(&meta)?;
224        // Parse the index definitions once per table: the maintain helpers'
225        // meta-accepting forms deserialise the JSON on every call, which for
226        // a 25-item batch against one table would repeat the work per item.
227        let gsi_defs = super::gsi::parse_gsi_defs(&meta)?;
228        let lsi_defs = super::lsi::parse_lsi_defs(&meta)?;
229        let vector_defs = super::vector_index::parse_vector_defs(&meta)?;
230        let attr_defs = super::vector_index::parse_attr_defs(&meta)?;
231
232        for wr in write_requests {
233            if let Some(ref mut put_req) = wr.put_request {
234                // Validate keys
235                helpers::validate_item_keys_for_batch(&put_req.item, &key_schema, &meta)?;
236
237                // Validate attribute values (empty strings, empty sets)
238                crate::validation::validate_item_attribute_values(&put_req.item)?;
239
240                // Normalize sets (deduplication)
241                crate::validation::normalize_item_sets(&mut put_req.item);
242
243                // Validate item size
244                let size = types::item_size(&put_req.item);
245                if size > types::MAX_ITEM_SIZE {
246                    return Err(DynoxideError::ValidationException(
247                        "Item size has exceeded the maximum allowed size".to_string(),
248                    ));
249                }
250
251                // TODO: validation must precede this call -- if reaching this line, caller has already validated keys.
252                let (pk, sk) = helpers::extract_key_strings(&put_req.item, &key_schema)?;
253                let item_json = serde_json::to_string(&put_req.item)
254                    .map_err(|e| DynoxideError::InternalServerError(e.to_string()))?;
255                let hash_prefix = put_req
256                    .item
257                    .get(&key_schema.partition_key)
258                    .map(crate::storage::compute_hash_prefix)
259                    .unwrap_or_default();
260                // Base write + index fan-out + stream are one atomic unit per
261                // item: a mid-fan-out failure rolls this item's write back.
262                // BatchWriteItem items are independent, so this is one
263                // transaction per write request.
264                let (old_size, gsi_units, lsi_units, vector_bytes) =
265                    helpers::with_write_transaction(storage, async {
266                        let old_json = storage
267                            .put_item_with_hash(
268                                table_name,
269                                &pk,
270                                &sk,
271                                &item_json,
272                                size,
273                                &hash_prefix,
274                            )
275                            .await?;
276                        let old_item: Option<Item> =
277                            old_json.and_then(|j| serde_json::from_str(&j).ok());
278                        let target = super::gsi::IndexWrite {
279                            table_name,
280                            pk: &pk,
281                            sk: &sk,
282                            pk_attr: &key_schema.partition_key,
283                            sk_attr: key_schema.sort_key.as_deref(),
284                            old_item: old_item.as_ref(),
285                            capacity_mode: request.return_consumed_capacity.as_deref(),
286                        };
287                        let gsi_units = super::gsi::maintain_gsis_after_write_with_defs(
288                            storage,
289                            &gsi_defs,
290                            &target,
291                            &put_req.item,
292                        )
293                        .await?;
294                        let lsi_units = super::lsi::maintain_lsis_after_write_with_defs(
295                            storage,
296                            &lsi_defs,
297                            &target,
298                            &put_req.item,
299                        )
300                        .await?;
301                        let vector_bytes =
302                            super::vector_index::maintain_vector_indexes_after_write_with_defs(
303                                storage,
304                                &vector_defs,
305                                &attr_defs,
306                                &target,
307                                &put_req.item,
308                                false,
309                            )
310                            .await?;
311                        crate::streams::record_stream_event(
312                            storage,
313                            &meta,
314                            old_item.as_ref(),
315                            Some(&put_req.item),
316                        )
317                        .await?;
318                        Ok((
319                            old_item.as_ref().map(types::item_size),
320                            gsi_units,
321                            lsi_units,
322                            vector_bytes,
323                        ))
324                    })
325                    .await?;
326
327                // Accumulate WCU on the larger of the displaced and new images
328                *table_wcu.entry(table_name.clone()).or_insert(0.0) +=
329                    types::table_write_capacity_units(old_size, Some(size));
330
331                // Accumulate secondary index units per table
332                accumulate_index_units(&mut table_gsi_units, table_name, &gsi_units);
333                accumulate_index_units(&mut table_lsi_units, table_name, &lsi_units);
334                accumulate_index_units(&mut table_vector_bytes, table_name, &vector_bytes);
335
336                // Track affected partition for deferred metrics
337                if let Some(pk_val) = put_req.item.get(&key_schema.partition_key) {
338                    affected_partitions.push((
339                        table_name.clone(),
340                        pk.clone(),
341                        key_schema.partition_key.clone(),
342                        pk_val.clone(),
343                    ));
344                }
345            } else if let Some(ref del_req) = wr.delete_request {
346                helpers::validate_key_only(&del_req.key, &key_schema)?;
347                // TODO: validation must precede this call -- if reaching this line, caller has already validated keys.
348                let (pk, sk) = helpers::extract_key_strings(&del_req.key, &key_schema)?;
349
350                // Base delete + index fan-out + stream are one atomic unit per item.
351                let (old_item, gsi_units, lsi_units, vector_bytes) =
352                    helpers::with_write_transaction(storage, async {
353                        let old_json = storage.delete_item(table_name, &pk, &sk).await?;
354                        let old_item: Option<Item> =
355                            old_json.as_ref().and_then(|j| serde_json::from_str(j).ok());
356                        let target = super::gsi::IndexWrite {
357                            table_name,
358                            pk: &pk,
359                            sk: &sk,
360                            pk_attr: &key_schema.partition_key,
361                            sk_attr: key_schema.sort_key.as_deref(),
362                            old_item: old_item.as_ref(),
363                            capacity_mode: request.return_consumed_capacity.as_deref(),
364                        };
365                        let gsi_units = super::gsi::maintain_gsis_after_delete_with_defs(
366                            storage, &gsi_defs, &target,
367                        )
368                        .await?;
369                        let lsi_units = super::lsi::maintain_lsis_after_delete_with_defs(
370                            storage, &lsi_defs, &target,
371                        )
372                        .await?;
373                        let vector_bytes =
374                            super::vector_index::maintain_vector_indexes_after_delete_with_defs(
375                                storage,
376                                &vector_defs,
377                                &attr_defs,
378                                &target,
379                            )
380                            .await?;
381                        if old_item.is_some() {
382                            crate::streams::record_stream_event(
383                                storage,
384                                &meta,
385                                old_item.as_ref(),
386                                None,
387                            )
388                            .await?;
389                        }
390                        Ok((old_item, gsi_units, lsi_units, vector_bytes))
391                    })
392                    .await?;
393
394                // Accumulate WCU: based on old item size if it existed, else 1 WCU
395                *table_wcu.entry(table_name.clone()).or_insert(0.0) +=
396                    types::table_write_capacity_units(
397                        old_item.as_ref().map(types::item_size),
398                        None,
399                    );
400
401                // Accumulate secondary index units per table
402                accumulate_index_units(&mut table_gsi_units, table_name, &gsi_units);
403                accumulate_index_units(&mut table_lsi_units, table_name, &lsi_units);
404                accumulate_index_units(&mut table_vector_bytes, table_name, &vector_bytes);
405
406                // Track affected partition for deferred metrics
407                if let Some(pk_val) = del_req.key.get(&key_schema.partition_key) {
408                    affected_partitions.push((
409                        table_name.clone(),
410                        pk.clone(),
411                        key_schema.partition_key.clone(),
412                        pk_val.clone(),
413                    ));
414                }
415            } else {
416                return Err(DynoxideError::ValidationException(
417                    "WriteRequest must contain either PutRequest or DeleteRequest".to_string(),
418                ));
419            }
420        }
421    }
422
423    // Build consumed capacity per table using pre-tracked WCU
424    let consumed_capacity = if matches!(
425        request.return_consumed_capacity.as_deref(),
426        Some("TOTAL") | Some("INDEXES")
427    ) {
428        let mut caps = Vec::new();
429        for table_name in request.request_items.keys() {
430            let total_wcu = table_wcu.get(table_name).copied().unwrap_or(0.0);
431            let gsi_units = table_gsi_units.get(table_name).cloned().unwrap_or_default();
432            let lsi_units = table_lsi_units.get(table_name).cloned().unwrap_or_default();
433            let vector_bytes = table_vector_bytes
434                .get(table_name)
435                .cloned()
436                .unwrap_or_default();
437            if let Some(cc) = crate::types::consumed_capacity_with_vector_indexes(
438                table_name,
439                total_wcu,
440                &gsi_units,
441                &lsi_units,
442                &vector_bytes,
443                &request.return_consumed_capacity,
444            ) {
445                caps.push(cc);
446            }
447        }
448        Some(caps)
449    } else {
450        None
451    };
452
453    // Item collection metrics are computed once per unique (table, pk), deferred from the write loop
454    let mut all_item_collection_metrics: HashMap<String, Vec<crate::types::ItemCollectionMetrics>> =
455        HashMap::new();
456    if matches!(
457        request.return_item_collection_metrics.as_deref(),
458        Some("SIZE")
459    ) {
460        // Deduplicate by (table, pk) to avoid redundant queries
461        let mut seen = std::collections::HashSet::new();
462        for (tbl, pk_str, pk_attr, pk_val) in &affected_partitions {
463            let key = (tbl.as_str(), pk_str.as_str());
464            if !seen.insert(key) {
465                continue;
466            }
467            let meta = helpers::require_table(storage, tbl).await?;
468            if let Some(icm) = helpers::build_item_collection_metrics(
469                storage,
470                &meta,
471                tbl,
472                pk_str,
473                pk_attr,
474                pk_val,
475                &request.return_item_collection_metrics,
476            )
477            .await?
478            {
479                all_item_collection_metrics
480                    .entry(tbl.clone())
481                    .or_default()
482                    .push(icm);
483            }
484        }
485    }
486    let item_collection_metrics = if all_item_collection_metrics.is_empty() {
487        None
488    } else {
489        Some(all_item_collection_metrics)
490    };
491
492    Ok(BatchWriteItemResponse {
493        unprocessed_items: HashMap::new(),
494        consumed_capacity,
495        item_collection_metrics,
496    })
497}
498
499#[cfg(test)]
500mod tests {
501    use crate::actions::{batch_write_item, create_table};
502    use crate::storage::Storage;
503    use crate::storage_backend::StorageBackend;
504
505    /// Each batch put is atomic with its own GSI fan-out: a mid-fan-out failure
506    /// rolls that item's base write back rather than leaving a torn index.
507    #[test]
508    fn batch_put_rolls_back_base_write_when_gsi_fan_out_fails() {
509        let storage = Storage::memory().unwrap();
510
511        let create = serde_json::from_value(serde_json::json!({
512            "TableName": "Orders",
513            "KeySchema": [{"AttributeName": "UserId", "KeyType": "HASH"}],
514            "AttributeDefinitions": [
515                {"AttributeName": "UserId", "AttributeType": "S"},
516                {"AttributeName": "Status", "AttributeType": "S"},
517                {"AttributeName": "Priority", "AttributeType": "S"}
518            ],
519            "GlobalSecondaryIndexes": [
520                {"IndexName": "StatusIndex", "KeySchema": [{"AttributeName": "Status", "KeyType": "HASH"}], "Projection": {"ProjectionType": "ALL"}},
521                {"IndexName": "PriorityIndex", "KeySchema": [{"AttributeName": "Priority", "KeyType": "HASH"}], "Projection": {"ProjectionType": "ALL"}}
522            ]
523        }))
524        .unwrap();
525        pollster::block_on(create_table::execute(&storage, create)).unwrap();
526
527        // Break the second GSI's fan-out by dropping its physical table.
528        storage.drop_gsi_table("Orders", "PriorityIndex").unwrap();
529
530        let batch = serde_json::from_value(serde_json::json!({
531            "RequestItems": {
532                "Orders": [
533                    {"PutRequest": {"Item": {"UserId": {"S": "u1"}, "Status": {"S": "SHIPPED"}, "Priority": {"S": "HIGH"}}}}
534                ]
535            }
536        }))
537        .unwrap();
538        let res = pollster::block_on(batch_write_item::execute(&storage, batch));
539        assert!(
540            res.is_err(),
541            "a mid-fan-out failure must surface as an error"
542        );
543
544        // The base write must roll back: no item landed.
545        let count =
546            pollster::block_on(<Storage as StorageBackend>::count_items(&storage, "Orders"))
547                .unwrap();
548        assert_eq!(
549            count, 0,
550            "batch put base write must roll back when fan-out fails"
551        );
552    }
553}
dynoxide/actions/batch_write_item.rs

dynoxide/actions/
batch_write_item.rs
