1use crate::actions::create_table::StreamSpecification;
2use crate::actions::{TableDescription, build_table_description};
3use crate::actions::{gsi, helpers};
4use crate::errors::{DynoxideError, Result};
5use crate::storage_backend::StorageBackend;
6use crate::streams;
7use crate::types::{AttributeDefinition, GlobalSecondaryIndex, KeySchemaElement, Projection};
8use crate::validation;
9use serde::{Deserialize, Serialize};
10
11#[derive(Debug, Default, Deserialize)]
13struct UpdateTableRequestRaw {
14 #[serde(rename = "TableName", default)]
15 table_name: Option<String>,
16
17 #[serde(rename = "AttributeDefinitions", default)]
18 attribute_definitions: Option<Vec<AttributeDefinition>>,
19
20 #[serde(rename = "GlobalSecondaryIndexUpdates", default)]
21 global_secondary_index_updates: Option<Vec<GlobalSecondaryIndexUpdate>>,
22
23 #[serde(rename = "StreamSpecification", default)]
24 stream_specification: Option<StreamSpecification>,
25
26 #[serde(rename = "DeletionProtectionEnabled", default)]
27 deletion_protection_enabled: Option<bool>,
28
29 #[serde(rename = "ProvisionedThroughput", default)]
30 provisioned_throughput: Option<serde_json::Value>,
31
32 #[serde(rename = "BillingMode", default)]
33 billing_mode: Option<String>,
34
35 #[serde(rename = "TableClass", default)]
36 table_class: Option<String>,
37
38 #[serde(rename = "OnDemandThroughput", default)]
39 on_demand_throughput: Option<crate::types::OnDemandThroughput>,
40}
41
42#[derive(Debug, Default)]
43pub struct UpdateTableRequest {
44 pub table_name: String,
45 pub attribute_definitions: Option<Vec<AttributeDefinition>>,
46 pub global_secondary_index_updates: Option<Vec<GlobalSecondaryIndexUpdate>>,
47 pub stream_specification: Option<StreamSpecification>,
48 pub deletion_protection_enabled: Option<bool>,
49 pub provisioned_throughput: Option<serde_json::Value>,
50 pub billing_mode: Option<String>,
51 pub table_class: Option<String>,
52 pub on_demand_throughput: Option<crate::types::OnDemandThroughput>,
53}
54
55impl<'de> serde::Deserialize<'de> for UpdateTableRequest {
56 fn deserialize<D: serde::Deserializer<'de>>(
57 deserializer: D,
58 ) -> std::result::Result<Self, D::Error> {
59 let raw = UpdateTableRequestRaw::deserialize(deserializer)?;
60
61 if raw.table_name.is_none() || raw.table_name.as_deref() == Some("") {
63 let msg = if raw.table_name.is_none() {
64 "The parameter 'TableName' is required but was not present in the request"
65 } else {
66 "TableName must be at least 3 characters long and at most 255 characters long"
67 };
68 return Err(serde::de::Error::custom(format!("VALIDATION:{}", msg)));
69 }
70
71 let table_name = raw.table_name.unwrap_or_default();
72
73 if table_name.len() < 3 || table_name.len() > 255 {
75 return Err(serde::de::Error::custom(
76 "VALIDATION:TableName must be at least 3 characters long and at most 255 characters long",
77 ));
78 }
79
80 let mut errors = Vec::new();
82
83 if !table_name
84 .chars()
85 .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-' || c == '.')
86 {
87 errors.push(format!(
88 "Value '{}' at 'tableName' failed to satisfy constraint: \
89 Member must satisfy regular expression pattern: [a-zA-Z0-9_.-]+",
90 table_name
91 ));
92 }
93
94 if let Some(msg) = crate::validation::format_validation_errors(&errors) {
95 return Err(serde::de::Error::custom(format!("VALIDATION:{}", msg)));
96 }
97
98 Ok(UpdateTableRequest {
99 table_name,
100 attribute_definitions: raw.attribute_definitions,
101 global_secondary_index_updates: raw.global_secondary_index_updates,
102 stream_specification: raw.stream_specification,
103 deletion_protection_enabled: raw.deletion_protection_enabled,
104 provisioned_throughput: raw.provisioned_throughput,
105 billing_mode: raw.billing_mode,
106 table_class: raw.table_class,
107 on_demand_throughput: raw.on_demand_throughput,
108 })
109 }
110}
111
112#[derive(Debug, Default, Deserialize)]
113pub struct GlobalSecondaryIndexUpdate {
114 #[serde(rename = "Update", default)]
115 pub update: Option<UpdateGsiAction>,
116
117 #[serde(rename = "Create", default)]
118 pub create: Option<CreateGsiAction>,
119
120 #[serde(rename = "Delete", default)]
121 pub delete: Option<DeleteGsiAction>,
122}
123
124#[derive(Debug, Default, Deserialize)]
125pub struct UpdateGsiAction {
126 #[serde(rename = "IndexName")]
127 pub index_name: String,
128
129 #[serde(rename = "ProvisionedThroughput", default)]
130 pub provisioned_throughput: Option<crate::types::ProvisionedThroughput>,
131}
132
133#[derive(Debug, Default, Deserialize)]
134pub struct CreateGsiAction {
135 #[serde(rename = "IndexName")]
136 pub index_name: String,
137
138 #[serde(rename = "KeySchema")]
139 pub key_schema: Vec<KeySchemaElement>,
140
141 #[serde(rename = "Projection")]
142 pub projection: Projection,
143}
144
145#[derive(Debug, Default, Deserialize)]
146pub struct DeleteGsiAction {
147 #[serde(rename = "IndexName")]
148 pub index_name: String,
149}
150
151#[derive(Debug, Default, Serialize)]
152pub struct UpdateTableResponse {
153 #[serde(rename = "TableDescription")]
154 pub table_description: TableDescription,
155}
156
157pub async fn execute<S: StorageBackend>(
158 storage: &S,
159 mut request: UpdateTableRequest,
160) -> Result<UpdateTableResponse> {
161 if request.on_demand_throughput.as_ref().is_some_and(|odt| {
169 odt.max_read_request_units.is_none() && odt.max_write_request_units.is_none()
170 }) {
171 request.on_demand_throughput = None;
172 }
173
174 validate_update_request(&request)?;
178
179 let meta = helpers::require_table(storage, &request.table_name).await?;
181
182 let current_billing_mode = meta.billing_mode.as_deref().unwrap_or("PROVISIONED");
183
184 if current_billing_mode == "PAY_PER_REQUEST"
188 && request.billing_mode.is_none()
189 && request.provisioned_throughput.is_some()
190 {
191 return Err(DynoxideError::ValidationException(
192 "One or more parameter values were invalid: \
193 Neither ReadCapacityUnits nor WriteCapacityUnits can be \
194 specified when BillingMode is PAY_PER_REQUEST"
195 .to_string(),
196 ));
197 }
198
199 if request.billing_mode.as_deref() == Some("PROVISIONED")
201 && request.provisioned_throughput.is_none()
202 {
203 return Err(DynoxideError::ValidationException(
204 "One or more parameter values were invalid: \
205 ProvisionedThroughput must be specified when BillingMode is PROVISIONED"
206 .to_string(),
207 ));
208 }
209
210 if let Some(ref odt) = request.on_demand_throughput {
218 let target_is_provisioned = match request.billing_mode.as_deref() {
219 Some("PAY_PER_REQUEST") => false,
220 Some(_) => true,
221 None => current_billing_mode == "PROVISIONED",
222 };
223 let members = [
224 ("MaxReadRequestUnits", odt.max_read_request_units),
225 ("MaxWriteRequestUnits", odt.max_write_request_units),
226 ];
227 if target_is_provisioned {
228 if let Some((member, _)) = members.iter().find(|(_, v)| v.is_some()) {
229 return Err(DynoxideError::ValidationException(format!(
230 "One or more parameter values were invalid: {member} for \
231 OnDemandThroughput cannot be specified when the table BillingMode \
232 is PROVISIONED"
233 )));
234 }
235 }
236 for (member, value) in members {
239 if value.is_some_and(|v| v == 0 || v < -1) {
240 return Err(DynoxideError::ValidationException(format!(
241 "One or more parameter values were invalid: Requested {member} for \
242 OnDemandThroughput for table is outside of valid range"
243 )));
244 }
245 }
246 }
247
248 if let Some(ref pt) = request.provisioned_throughput {
250 if let Some(obj) = pt.as_object() {
251 let new_rcu = obj
252 .get("ReadCapacityUnits")
253 .and_then(|v| v.as_i64())
254 .unwrap_or(0);
255 let new_wcu = obj
256 .get("WriteCapacityUnits")
257 .and_then(|v| v.as_i64())
258 .unwrap_or(0);
259
260 let (cur_rcu, cur_wcu) = parse_current_throughput(&meta);
262
263 let billing_mode_unchanged = request.billing_mode.is_none()
264 || (request.billing_mode.as_deref() == Some("PROVISIONED")
265 && current_billing_mode == "PROVISIONED");
266
267 if new_rcu == cur_rcu && new_wcu == cur_wcu && billing_mode_unchanged {
268 return Err(DynoxideError::ValidationException(format!(
269 "The provisioned throughput for the table will not change. \
270 The requested value equals the current value. \
271 Current ReadCapacityUnits provisioned for the table: {}. \
272 Requested ReadCapacityUnits: {}. \
273 Current WriteCapacityUnits provisioned for the table: {}. \
274 Requested WriteCapacityUnits: {}. \
275 Refer to the Amazon DynamoDB Developer Guide for current limits \
276 and how to request higher limits.",
277 cur_rcu, new_rcu, cur_wcu, new_wcu
278 )));
279 }
280 }
281 }
282
283 let mut current_gsis: Vec<GlobalSecondaryIndex> = meta
285 .gsi_definitions
286 .as_ref()
287 .map(|json| serde_json::from_str(json))
288 .transpose()
289 .map_err(|e| DynoxideError::InternalServerError(format!("Bad GSI JSON: {e}")))?
290 .unwrap_or_default();
291
292 if let Some(ref updates) = request.global_secondary_index_updates {
294 for update in updates {
295 if let Some(ref upd) = update.update {
296 if !current_gsis.iter().any(|g| g.index_name == upd.index_name) {
297 return Err(DynoxideError::ValidationException(
300 "This operation cannot be performed with given input values. \
301 Please contact DynamoDB service team for more info: \
302 Action Blocked: IndexUpdate"
303 .to_string(),
304 ));
305 }
306 }
307 }
308 }
309
310 if let Some(ref updates) = request.global_secondary_index_updates {
312 if updates.len() > 5 {
313 return Err(DynoxideError::LimitExceededException(
314 "Subscriber limit exceeded: Only 1 online index can be created or \
315 deleted simultaneously per table"
316 .to_string(),
317 ));
318 }
319 }
320
321 let mut attr_defs: Vec<AttributeDefinition> = serde_json::from_str(&meta.attribute_definitions)
327 .map_err(|e| DynoxideError::InternalServerError(format!("Bad attr defs JSON: {e}")))?;
328
329 if let Some(ref provided) = request.attribute_definitions {
330 for def in provided {
331 if !attr_defs
336 .iter()
337 .any(|d| d.attribute_name == def.attribute_name)
338 {
339 attr_defs.push(def.clone());
340 }
341 }
342 }
343
344 let key_schema = helpers::parse_key_schema(&meta)?;
346
347 if let Some(ref updates) = request.global_secondary_index_updates {
349 for update in updates {
350 if let Some(ref create) = update.create {
351 if current_gsis
352 .iter()
353 .any(|g| g.index_name == create.index_name)
354 {
355 return Err(DynoxideError::ValidationException(format!(
356 "One or more parameter values were invalid: \
357 Index already exists: {}",
358 create.index_name
359 )));
360 }
361 let gsi_def = GlobalSecondaryIndex {
362 index_name: create.index_name.clone(),
363 key_schema: create.key_schema.clone(),
364 projection: create.projection.clone(),
365 provisioned_throughput: None,
366 };
367 validation::validate_gsi(
372 &gsi_def,
373 request.attribute_definitions.as_deref().unwrap_or(&[]),
374 )?;
375 }
376 if let Some(ref delete) = update.delete {
377 if !current_gsis
378 .iter()
379 .any(|g| g.index_name == delete.index_name)
380 {
381 return Err(DynoxideError::ResourceNotFoundException(format!(
382 "Requested resource not found: Table: {} not found",
383 delete.index_name
384 )));
385 }
386 }
387 }
388 }
389
390 let now = {
394 use std::sync::atomic::{AtomicU64, Ordering};
395 static LAST_TS: AtomicU64 = AtomicU64::new(0);
396 let wall = web_time::SystemTime::now()
397 .duration_since(web_time::UNIX_EPOCH)
398 .unwrap_or_default()
399 .as_secs_f64();
400 loop {
401 let prev_bits = LAST_TS.load(Ordering::SeqCst);
402 let prev_f = f64::from_bits(prev_bits);
403 let candidate = if wall > prev_f { wall } else { prev_f + 0.001 };
404 let candidate_bits = candidate.to_bits();
405 if LAST_TS
406 .compare_exchange(
407 prev_bits,
408 candidate_bits,
409 Ordering::SeqCst,
410 Ordering::SeqCst,
411 )
412 .is_ok()
413 {
414 break candidate;
415 }
416 }
417 };
418
419 let (cur_rcu, cur_wcu) = parse_current_throughput(&meta);
420 let is_pt_update = request.provisioned_throughput.is_some();
421 let (new_rcu, new_wcu) = if let Some(ref pt) = request.provisioned_throughput {
422 let obj = pt.as_object();
423 (
424 obj.and_then(|o| o.get("ReadCapacityUnits"))
425 .and_then(|v| v.as_i64())
426 .unwrap_or(0),
427 obj.and_then(|o| o.get("WriteCapacityUnits"))
428 .and_then(|v| v.as_i64())
429 .unwrap_or(0),
430 )
431 } else {
432 (cur_rcu, cur_wcu)
433 };
434
435 let is_increase = new_rcu > cur_rcu || new_wcu > cur_wcu;
436 let is_decrease = new_rcu < cur_rcu || new_wcu < cur_wcu;
437
438 let odt_change = request.on_demand_throughput.as_ref().map(|req_odt| {
443 let stored: crate::types::OnDemandThroughput = meta
444 .on_demand_throughput
445 .as_deref()
446 .and_then(|json| serde_json::from_str(json).ok())
447 .unwrap_or_default();
448 let echo = crate::types::OnDemandThroughput {
449 max_read_request_units: req_odt
450 .max_read_request_units
451 .or(stored.max_read_request_units),
452 max_write_request_units: req_odt
453 .max_write_request_units
454 .or(stored.max_write_request_units),
455 };
456 let strip = |v: Option<i64>| v.filter(|&v| v != -1);
457 let effective = crate::types::OnDemandThroughput {
458 max_read_request_units: strip(echo.max_read_request_units),
459 max_write_request_units: strip(echo.max_write_request_units),
460 };
461 (echo, effective)
462 });
463
464 helpers::with_write_transaction(storage, async {
466 if let Some(ref updates) = request.global_secondary_index_updates {
467 for update in updates {
468 if let Some(ref create) = update.create {
469 let gsi_def = GlobalSecondaryIndex {
470 index_name: create.index_name.clone(),
471 key_schema: create.key_schema.clone(),
472 projection: create.projection.clone(),
473 provisioned_throughput: None,
474 };
475
476 storage
477 .create_gsi_table(&request.table_name, &create.index_name)
478 .await?;
479
480 let gsi_p = gsi::gsi_to_def(&gsi_def)?;
481 backfill_gsi(storage, &request.table_name, &key_schema, &gsi_p).await?;
482
483 current_gsis.push(gsi_def);
484 }
485
486 if let Some(ref delete) = update.delete {
487 storage
488 .drop_gsi_table(&request.table_name, &delete.index_name)
489 .await?;
490 current_gsis.retain(|g| g.index_name != delete.index_name);
491 }
492 }
493 }
494
495 let lsi_defs = crate::actions::lsi::parse_lsi_defs(&meta)?;
499 reconcile_attribute_definitions(&mut attr_defs, &key_schema, ¤t_gsis, &lsi_defs);
500
501 let attr_defs_json = serde_json::to_string(&attr_defs)
503 .map_err(|e| DynoxideError::InternalServerError(e.to_string()))?;
504 let gsi_json = if current_gsis.is_empty() {
505 None
506 } else {
507 Some(
508 serde_json::to_string(¤t_gsis)
509 .map_err(|e| DynoxideError::InternalServerError(e.to_string()))?,
510 )
511 };
512
513 storage
514 .update_table_metadata(&request.table_name, &attr_defs_json, gsi_json.as_deref())
515 .await?;
516
517 if is_pt_update {
519 let prev = parse_stored_throughput(&meta);
520 let mut stored = StoredProvisionedThroughput {
521 read_capacity_units: new_rcu,
522 write_capacity_units: new_wcu,
523 last_increase_date_time: prev.as_ref().and_then(|p| p.last_increase_date_time),
524 last_decrease_date_time: prev.as_ref().and_then(|p| p.last_decrease_date_time),
525 number_of_decreases_today: prev
526 .as_ref()
527 .and_then(|p| p.number_of_decreases_today)
528 .or(Some(0)),
529 };
530 if is_increase {
531 stored.last_increase_date_time = Some(now);
532 }
533 if is_decrease {
534 stored.last_decrease_date_time = Some(now);
535 stored.number_of_decreases_today =
536 Some(stored.number_of_decreases_today.unwrap_or(0) + 1);
537 }
538 let pt_json = serde_json::to_string(&stored)
539 .map_err(|e| DynoxideError::InternalServerError(e.to_string()))?;
540 storage
541 .update_provisioned_throughput(&request.table_name, &pt_json)
542 .await?;
543 }
544
545 if let Some(enabled) = request.deletion_protection_enabled {
547 storage
548 .update_deletion_protection(&request.table_name, enabled)
549 .await?;
550 }
551
552 if let Some(ref table_class) = request.table_class {
554 storage
555 .update_table_class(&request.table_name, table_class)
556 .await?;
557 }
558
559 if let Some((_, ref effective)) = odt_change {
561 if effective.max_read_request_units.is_none()
562 && effective.max_write_request_units.is_none()
563 {
564 storage
565 .clear_on_demand_throughput(&request.table_name)
566 .await?;
567 } else {
568 let json = serde_json::to_string(effective)
569 .map_err(|e| DynoxideError::InternalServerError(e.to_string()))?;
570 storage
571 .update_on_demand_throughput(&request.table_name, &json)
572 .await?;
573 }
574 }
575
576 if let Some(ref billing_mode) = request.billing_mode {
578 storage
579 .update_billing_mode(&request.table_name, billing_mode)
580 .await?;
581 if billing_mode == "PAY_PER_REQUEST" {
582 storage
584 .clear_provisioned_throughput(&request.table_name)
585 .await?;
586 } else if billing_mode == "PROVISIONED" {
587 storage
590 .clear_on_demand_throughput(&request.table_name)
591 .await?;
592 }
593 }
594
595 if let Some(ref spec) = request.stream_specification {
597 if spec.stream_enabled {
598 let view_type = spec
599 .stream_view_type
600 .as_deref()
601 .unwrap_or("NEW_AND_OLD_IMAGES");
602 let label = streams::generate_stream_label(storage.clock());
603 storage
604 .enable_stream(&request.table_name, view_type, &label)
605 .await?;
606 } else {
607 storage.disable_stream(&request.table_name).await?;
608 }
609 }
610
611 Ok(())
612 })
613 .await?;
614
615 let updated_meta = helpers::require_table(storage, &request.table_name).await?;
617 let mut desc = build_table_description(&updated_meta, Some(0), Some(0));
618
619 if let Some((echo, _)) = odt_change {
622 desc.on_demand_throughput = Some(echo);
623 }
624
625 if is_pt_update {
627 desc.table_status = "UPDATING".to_string();
628
629 let stored = parse_stored_throughput(&updated_meta);
632 if let Some(ref mut pt) = desc.provisioned_throughput {
633 pt.read_capacity_units = cur_rcu as u64;
634 pt.write_capacity_units = cur_wcu as u64;
635 if let Some(ref s) = stored {
636 pt.last_increase_date_time = s.last_increase_date_time;
637 pt.last_decrease_date_time = s.last_decrease_date_time;
638 pt.number_of_decreases_today = s.number_of_decreases_today.unwrap_or(0);
639 }
640 }
641 }
642
643 Ok(UpdateTableResponse {
644 table_description: desc,
645 })
646}
647
648fn reconcile_attribute_definitions(
654 attr_defs: &mut Vec<AttributeDefinition>,
655 key_schema: &helpers::KeySchema,
656 gsis: &[GlobalSecondaryIndex],
657 lsi_defs: &[crate::actions::lsi::LsiDef],
658) {
659 let mut used: std::collections::HashSet<&str> = std::collections::HashSet::new();
660 used.insert(key_schema.partition_key.as_str());
661 if let Some(ref sk) = key_schema.sort_key {
662 used.insert(sk.as_str());
663 }
664 for g in gsis {
665 for k in &g.key_schema {
666 used.insert(k.attribute_name.as_str());
667 }
668 }
669 for lsi in lsi_defs {
670 used.insert(lsi.pk_attr.as_str());
671 if let Some(ref sk) = lsi.sk_attr {
672 used.insert(sk.as_str());
673 }
674 }
675 attr_defs.retain(|d| used.contains(d.attribute_name.as_str()));
676}
677
678fn validate_update_request(request: &UpdateTableRequest) -> Result<()> {
683 let mut errors = Vec::new();
685
686 if let Some(ref pt) = request.provisioned_throughput {
688 if let Some(obj) = pt.as_object() {
689 let wcu = obj.get("WriteCapacityUnits");
690 let rcu = obj.get("ReadCapacityUnits");
691 if wcu.is_none() || wcu == Some(&serde_json::Value::Null) {
692 errors.push("Value null at 'provisionedThroughput.writeCapacityUnits' failed to satisfy constraint: Member must not be null".to_string());
693 } else if let Some(w) = wcu.and_then(|v| v.as_i64()) {
694 if w < 1 {
695 errors.push(format!("Value '{}' at 'provisionedThroughput.writeCapacityUnits' failed to satisfy constraint: Member must have value greater than or equal to 1", w));
696 }
697 }
698 if rcu.is_none() || rcu == Some(&serde_json::Value::Null) {
699 errors.push("Value null at 'provisionedThroughput.readCapacityUnits' failed to satisfy constraint: Member must not be null".to_string());
700 } else if let Some(r) = rcu.and_then(|v| v.as_i64()) {
701 if r < 1 {
702 errors.push(format!("Value '{}' at 'provisionedThroughput.readCapacityUnits' failed to satisfy constraint: Member must have value greater than or equal to 1", r));
703 }
704 }
705 }
706 }
707
708 if let Some(ref updates) = request.global_secondary_index_updates {
710 for (i, update) in updates.iter().enumerate() {
711 if let Some(ref upd) = update.update {
712 if upd.index_name.len() < 3 {
714 errors.push(format!("Value '{}' at 'globalSecondaryIndexUpdates.{}.member.update.indexName' failed to satisfy constraint: Member must have length greater than or equal to 3", upd.index_name, i + 1));
715 }
716 if !upd.index_name.is_empty()
717 && !upd
718 .index_name
719 .chars()
720 .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-' || c == '.')
721 {
722 errors.push(format!("Value '{}' at 'globalSecondaryIndexUpdates.{}.member.update.indexName' failed to satisfy constraint: Member must satisfy regular expression pattern: [a-zA-Z0-9_.-]+", upd.index_name, i + 1));
723 }
724 if let Some(ref pt) = upd.provisioned_throughput {
726 let wcu = pt.write_capacity_units;
727 let rcu = pt.read_capacity_units;
728 if wcu.is_none() {
729 errors.push(format!("Value null at 'globalSecondaryIndexUpdates.{}.member.update.provisionedThroughput.writeCapacityUnits' failed to satisfy constraint: Member must not be null", i + 1));
730 } else if let Some(w) = wcu {
731 if w < 1 {
732 errors.push(format!("Value '{}' at 'globalSecondaryIndexUpdates.{}.member.update.provisionedThroughput.writeCapacityUnits' failed to satisfy constraint: Member must have value greater than or equal to 1", w, i + 1));
733 }
734 }
735 if rcu.is_none() {
736 errors.push(format!("Value null at 'globalSecondaryIndexUpdates.{}.member.update.provisionedThroughput.readCapacityUnits' failed to satisfy constraint: Member must not be null", i + 1));
737 } else if let Some(r) = rcu {
738 if r < 1 {
739 errors.push(format!("Value '{}' at 'globalSecondaryIndexUpdates.{}.member.update.provisionedThroughput.readCapacityUnits' failed to satisfy constraint: Member must have value greater than or equal to 1", r, i + 1));
740 }
741 }
742 } else {
743 errors.push(format!("Value null at 'globalSecondaryIndexUpdates.{}.member.update.provisionedThroughput' failed to satisfy constraint: Member must not be null", i + 1));
744 }
745 }
746 }
747 }
748
749 errors.truncate(10);
751
752 if !errors.is_empty() {
753 let prefix = format!(
754 "{} validation error{} detected: ",
755 errors.len(),
756 if errors.len() == 1 { "" } else { "s" }
757 );
758 return Err(DynoxideError::ValidationException(format!(
759 "{}{}",
760 prefix,
761 errors.join("; ")
762 )));
763 }
764
765 if let Some(ref bm) = request.billing_mode {
769 if bm != "PROVISIONED" && bm != "PAY_PER_REQUEST" {
770 return Err(DynoxideError::ValidationException(format!(
771 "1 validation error detected: Value '{}' at 'billingMode' \
772 failed to satisfy constraint: Member must satisfy enum value set: \
773 [PROVISIONED, PAY_PER_REQUEST]",
774 bm
775 )));
776 }
777 }
778
779 if let Some(ref tc) = request.table_class {
781 if tc != "STANDARD" && tc != "STANDARD_INFREQUENT_ACCESS" {
782 return Err(DynoxideError::ValidationException(format!(
783 "1 validation error detected: Value '{tc}' at 'tableClass' failed to satisfy \
784 constraint: Member must satisfy enum value set: \
785 [STANDARD, STANDARD_INFREQUENT_ACCESS]"
786 )));
787 }
788 }
789
790 if request.billing_mode.as_deref() == Some("PAY_PER_REQUEST")
792 && request.provisioned_throughput.is_some()
793 {
794 return Err(DynoxideError::ValidationException(
795 "One or more parameter values were invalid: \
796 Neither ReadCapacityUnits nor WriteCapacityUnits can be \
797 specified when BillingMode is PAY_PER_REQUEST"
798 .to_string(),
799 ));
800 }
801
802 if let Some(ref pt) = request.provisioned_throughput {
804 if let Some(obj) = pt.as_object() {
805 let rcu = obj
806 .get("ReadCapacityUnits")
807 .and_then(|v| v.as_i64())
808 .unwrap_or(0);
809 let wcu = obj
810 .get("WriteCapacityUnits")
811 .and_then(|v| v.as_i64())
812 .unwrap_or(0);
813 const MAX_THROUGHPUT: i64 = 1_000_000_000_000;
814 if rcu > MAX_THROUGHPUT {
815 return Err(DynoxideError::ValidationException(format!(
816 "Given value {} for ReadCapacityUnits is out of bounds",
817 rcu
818 )));
819 }
820 if wcu > MAX_THROUGHPUT {
821 return Err(DynoxideError::ValidationException(format!(
822 "Given value {} for WriteCapacityUnits is out of bounds",
823 wcu
824 )));
825 }
826 }
827 }
828
829 let no_config_change = request.provisioned_throughput.is_none()
835 && request.billing_mode.is_none()
836 && request.stream_specification.is_none()
837 && request.deletion_protection_enabled.is_none()
838 && request.table_class.is_none()
839 && request.on_demand_throughput.is_none();
840 let no_gsi_change = request
841 .global_secondary_index_updates
842 .as_ref()
843 .is_none_or(|updates| updates.is_empty());
844 if no_gsi_change && no_config_change {
845 return Err(DynoxideError::ValidationException(
846 "At least one of ProvisionedThroughput, BillingMode, UpdateStreamEnabled, GlobalSecondaryIndexUpdates or SSESpecification or ReplicaUpdates is required".to_string(),
847 ));
848 }
849
850 if let Some(ref updates) = request.global_secondary_index_updates {
852 for update in updates {
854 if update.update.is_none() && update.create.is_none() && update.delete.is_none() {
855 return Err(DynoxideError::ValidationException(
856 "One or more parameter values were invalid: One of GlobalSecondaryIndexUpdate.Update, GlobalSecondaryIndexUpdate.Create, GlobalSecondaryIndexUpdate.Delete must not be null".to_string(),
857 ));
858 }
859 }
860
861 let mut seen_names = std::collections::HashSet::new();
863 for update in updates {
864 let name = if let Some(ref u) = update.update {
865 Some(u.index_name.as_str())
866 } else if let Some(ref c) = update.create {
867 Some(c.index_name.as_str())
868 } else {
869 update.delete.as_ref().map(|d| d.index_name.as_str())
870 };
871 if let Some(name) = name {
872 if !seen_names.insert(name.to_string()) {
873 return Err(DynoxideError::ValidationException(format!(
874 "One or more parameter values were invalid: Only one global secondary index update per index is allowed simultaneously. Index: {}",
875 name
876 )));
877 }
878 }
879 }
880 }
881
882 Ok(())
883}
884
885#[derive(Debug, Clone, Default, Serialize, Deserialize)]
887struct StoredProvisionedThroughput {
888 #[serde(rename = "ReadCapacityUnits")]
889 read_capacity_units: i64,
890 #[serde(rename = "WriteCapacityUnits")]
891 write_capacity_units: i64,
892 #[serde(
893 rename = "LastIncreaseDateTime",
894 skip_serializing_if = "Option::is_none"
895 )]
896 last_increase_date_time: Option<f64>,
897 #[serde(
898 rename = "LastDecreaseDateTime",
899 skip_serializing_if = "Option::is_none"
900 )]
901 last_decrease_date_time: Option<f64>,
902 #[serde(
903 rename = "NumberOfDecreasesToday",
904 skip_serializing_if = "Option::is_none"
905 )]
906 number_of_decreases_today: Option<u64>,
907}
908
909fn parse_current_throughput(meta: &crate::storage::TableMetadata) -> (i64, i64) {
911 parse_stored_throughput(meta)
912 .map(|pt| (pt.read_capacity_units, pt.write_capacity_units))
913 .unwrap_or((0, 0))
914}
915
916fn parse_stored_throughput(
918 meta: &crate::storage::TableMetadata,
919) -> Option<StoredProvisionedThroughput> {
920 meta.provisioned_throughput
921 .as_ref()
922 .and_then(|pt_json| serde_json::from_str(pt_json).ok())
923}
924
925async fn backfill_gsi<S: StorageBackend>(
927 storage: &S,
928 table_name: &str,
929 key_schema: &helpers::KeySchema,
930 gsi_def: &gsi::GsiDef,
931) -> Result<()> {
932 const BATCH_SIZE: usize = 1000;
933 let mut last_pk: Option<String> = None;
934 let mut last_sk: Option<String> = None;
935
936 loop {
937 let items = storage
938 .scan_items(
939 table_name,
940 &crate::storage::ScanParams {
941 limit: Some(BATCH_SIZE),
942 exclusive_start_pk: last_pk.as_deref(),
943 exclusive_start_sk: last_sk.as_deref(),
944 ..Default::default()
945 },
946 )
947 .await?;
948
949 if items.is_empty() {
950 break;
951 }
952
953 let mut rows: Vec<crate::storage_backend::GsiItemRow> = Vec::new();
954 for (pk, sk, item_json) in &items {
955 let item: crate::types::Item = serde_json::from_str(item_json)
956 .map_err(|e| DynoxideError::InternalServerError(format!("Bad item JSON: {e}")))?;
957
958 if let Some((gsi_pk, gsi_sk)) = gsi_def.index_key_strings(&item) {
960 let projected = gsi::build_index_item(
961 &item,
962 gsi_def,
963 &key_schema.partition_key,
964 key_schema.sort_key.as_deref(),
965 );
966 let projected_json = serde_json::to_string(&projected)
967 .map_err(|e| DynoxideError::InternalServerError(e.to_string()))?;
968
969 rows.push(crate::storage_backend::GsiItemRow {
970 gsi_pk,
971 gsi_sk,
972 table_pk: pk.clone(),
973 table_sk: sk.clone(),
974 item_json: projected_json,
975 });
976 }
977 }
978
979 storage
980 .insert_gsi_items(table_name, &gsi_def.index_name, &rows)
981 .await?;
982
983 let last = &items[items.len() - 1];
984 last_pk = Some(last.0.clone());
985 last_sk = Some(last.1.clone());
986
987 if items.len() < BATCH_SIZE {
988 break;
989 }
990 }
991
992 Ok(())
993}