1use crate::actions::create_table::StreamSpecification;
2use crate::actions::vector_lifecycle::VectorIndexLifecycle;
3use crate::actions::{TableDescription, build_table_description};
4use crate::actions::{gsi, helpers};
5use crate::errors::{DynoxideError, Result};
6use crate::storage_backend::StorageBackend;
7use crate::streams;
8use crate::types::{
9 AttributeDefinition, GlobalSecondaryIndex, Item, KeySchemaElement, Projection, VectorIndex,
10};
11use crate::validation;
12use serde::{Deserialize, Serialize};
13
14#[derive(Debug, Default, Deserialize)]
16struct UpdateTableRequestRaw {
17 #[serde(rename = "TableName", default)]
18 table_name: Option<String>,
19
20 #[serde(rename = "AttributeDefinitions", default)]
21 attribute_definitions: Option<Vec<AttributeDefinition>>,
22
23 #[serde(rename = "GlobalSecondaryIndexUpdates", default)]
24 global_secondary_index_updates: Option<Vec<GlobalSecondaryIndexUpdate>>,
25
26 #[serde(rename = "VectorIndexUpdates", default)]
30 vector_index_updates: Option<serde_json::Value>,
31
32 #[serde(rename = "StreamSpecification", default)]
33 stream_specification: Option<StreamSpecification>,
34
35 #[serde(rename = "DeletionProtectionEnabled", default)]
36 deletion_protection_enabled: Option<bool>,
37
38 #[serde(rename = "ProvisionedThroughput", default)]
39 provisioned_throughput: Option<serde_json::Value>,
40
41 #[serde(rename = "BillingMode", default)]
42 billing_mode: Option<String>,
43
44 #[serde(rename = "TableClass", default)]
45 table_class: Option<String>,
46
47 #[serde(rename = "OnDemandThroughput", default)]
48 on_demand_throughput: Option<crate::types::OnDemandThroughput>,
49}
50
51#[derive(Debug, Default)]
52pub struct UpdateTableRequest {
53 pub table_name: String,
54 pub attribute_definitions: Option<Vec<AttributeDefinition>>,
55 pub global_secondary_index_updates: Option<Vec<GlobalSecondaryIndexUpdate>>,
56 pub vector_index_updates: Option<Vec<VectorIndexUpdate>>,
57 pub stream_specification: Option<StreamSpecification>,
58 pub deletion_protection_enabled: Option<bool>,
59 pub provisioned_throughput: Option<serde_json::Value>,
60 pub billing_mode: Option<String>,
61 pub table_class: Option<String>,
62 pub on_demand_throughput: Option<crate::types::OnDemandThroughput>,
63}
64
65impl<'de> serde::Deserialize<'de> for UpdateTableRequest {
66 fn deserialize<D: serde::Deserializer<'de>>(
67 deserializer: D,
68 ) -> std::result::Result<Self, D::Error> {
69 let raw = UpdateTableRequestRaw::deserialize(deserializer)?;
70
71 if raw.table_name.is_none() || raw.table_name.as_deref() == Some("") {
73 let msg = if raw.table_name.is_none() {
74 "The parameter 'TableName' is required but was not present in the request"
75 } else {
76 "TableName must be at least 3 characters long and at most 255 characters long"
77 };
78 return Err(serde::de::Error::custom(format!("VALIDATION:{}", msg)));
79 }
80
81 let table_name = raw.table_name.unwrap_or_default();
82
83 if table_name.len() < 3 || table_name.len() > 255 {
85 return Err(serde::de::Error::custom(
86 "VALIDATION:TableName must be at least 3 characters long and at most 255 characters long",
87 ));
88 }
89
90 let mut errors = Vec::new();
92
93 if !table_name
94 .chars()
95 .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-' || c == '.')
96 {
97 errors.push(format!(
98 "Value '{}' at 'tableName' failed to satisfy constraint: \
99 Member must satisfy regular expression pattern: [a-zA-Z0-9_.-]+",
100 table_name
101 ));
102 }
103
104 if let Some(msg) = crate::validation::format_validation_errors(&errors) {
105 return Err(serde::de::Error::custom(format!("VALIDATION:{}", msg)));
106 }
107
108 if let Some(ref vix_val) = raw.vector_index_updates {
118 let mut vix_errors = Vec::new();
119 collect_vix_update_errors(vix_val, &mut vix_errors);
120 if !vix_errors.is_empty() {
121 let mut errors = Vec::new();
122 if let Some(ref gsi_updates) = raw.global_secondary_index_updates {
123 collect_gsi_update_errors(gsi_updates, &mut errors);
124 }
125 errors.append(&mut vix_errors);
126 errors.truncate(10);
127 let msg = format!(
128 "{} validation error{} detected: {}",
129 errors.len(),
130 if errors.len() == 1 { "" } else { "s" },
131 errors.join("; ")
132 );
133 return Err(serde::de::Error::custom(format!("VALIDATION:{}", msg)));
134 }
135 }
136
137 let vector_index_updates = raw
138 .vector_index_updates
139 .as_ref()
140 .map(parse_vector_index_updates)
141 .transpose()
142 .map_err(serde::de::Error::custom)?;
143
144 Ok(UpdateTableRequest {
145 table_name,
146 attribute_definitions: raw.attribute_definitions,
147 global_secondary_index_updates: raw.global_secondary_index_updates,
148 vector_index_updates,
149 stream_specification: raw.stream_specification,
150 deletion_protection_enabled: raw.deletion_protection_enabled,
151 provisioned_throughput: raw.provisioned_throughput,
152 billing_mode: raw.billing_mode,
153 table_class: raw.table_class,
154 on_demand_throughput: raw.on_demand_throughput,
155 })
156 }
157}
158
159#[derive(Debug, Default, Deserialize)]
160pub struct GlobalSecondaryIndexUpdate {
161 #[serde(rename = "Update", default)]
162 pub update: Option<UpdateGsiAction>,
163
164 #[serde(rename = "Create", default)]
165 pub create: Option<CreateGsiAction>,
166
167 #[serde(rename = "Delete", default)]
168 pub delete: Option<DeleteGsiAction>,
169}
170
171#[derive(Debug, Default, Deserialize)]
172pub struct UpdateGsiAction {
173 #[serde(rename = "IndexName")]
174 pub index_name: String,
175
176 #[serde(rename = "ProvisionedThroughput", default)]
177 pub provisioned_throughput: Option<crate::types::ProvisionedThroughput>,
178}
179
180#[derive(Debug, Default, Deserialize)]
181pub struct CreateGsiAction {
182 #[serde(rename = "IndexName")]
183 pub index_name: String,
184
185 #[serde(rename = "KeySchema")]
186 pub key_schema: Vec<KeySchemaElement>,
187
188 #[serde(rename = "Projection")]
189 pub projection: Projection,
190}
191
192#[derive(Debug, Default, Deserialize)]
193pub struct DeleteGsiAction {
194 #[serde(rename = "IndexName")]
195 pub index_name: String,
196}
197
198#[derive(Debug, Default)]
203pub struct VectorIndexUpdate {
204 pub create: Option<VectorIndex>,
207 pub delete: Option<DeleteVectorIndexAction>,
208}
209
210#[derive(Debug, Default, Deserialize)]
211pub struct DeleteVectorIndexAction {
212 #[serde(rename = "IndexName")]
213 pub index_name: String,
214}
215
216fn collect_vix_update_errors(val: &serde_json::Value, errors: &mut Vec<String>) {
221 let Some(arr) = val.as_array() else {
222 return;
223 };
224 for (i, elem) in arr.iter().enumerate().take(10) {
225 let Some(obj) = elem.as_object() else {
226 continue;
227 };
228 if let Some(create) = obj.get("Create").and_then(|v| v.as_object()) {
229 crate::actions::create_table::collect_vix_obj_errors(
230 create,
231 &format!("vectorIndexUpdates.{}.member.create", i + 1),
232 errors,
233 );
234 }
235 }
236}
237
238#[cfg(feature = "mcp-server")]
243pub(crate) fn vector_index_updates_request_model_error(val: &serde_json::Value) -> Option<String> {
244 let mut errors = Vec::new();
245 collect_vix_update_errors(val, &mut errors);
246 errors.truncate(10);
247 if errors.is_empty() {
248 return None;
249 }
250 Some(format!(
251 "{} validation error{} detected: {}",
252 errors.len(),
253 if errors.len() == 1 { "" } else { "s" },
254 errors.join("; ")
255 ))
256}
257
258pub(crate) fn parse_vector_index_updates(
266 val: &serde_json::Value,
267) -> std::result::Result<Vec<VectorIndexUpdate>, String> {
268 let arr = val
269 .as_array()
270 .ok_or_else(|| "Unexpected field type".to_string())?;
271 let mut out = Vec::with_capacity(arr.len());
272 for elem in arr {
273 let obj = elem
274 .as_object()
275 .ok_or_else(|| "Unexpected value type in payload".to_string())?;
276 let mut update = VectorIndexUpdate::default();
277 if let Some(create) = obj.get("Create").filter(|v| !v.is_null()) {
284 let mut create = create.clone();
285 if let Some(create_obj) = create.as_object_mut() {
286 crate::actions::create_table::normalise_vix_dimensions_obj(create_obj);
287 }
288 update.create = Some(serde_json::from_value(create).map_err(|e| e.to_string())?);
289 }
290 if let Some(delete) = obj.get("Delete").filter(|v| !v.is_null()) {
291 update.delete =
292 Some(serde_json::from_value(delete.clone()).map_err(|e| e.to_string())?);
293 }
294 out.push(update);
295 }
296 Ok(out)
297}
298
299#[derive(Debug, Default, Serialize)]
300pub struct UpdateTableResponse {
301 #[serde(rename = "TableDescription")]
302 pub table_description: TableDescription,
303}
304
305const ONE_ONLINE_INDEX: &str = "Subscriber limit exceeded: Only 1 online index can be created or deleted \
308 simultaneously per table";
309
310pub async fn execute<S: StorageBackend>(
311 storage: &S,
312 mut request: UpdateTableRequest,
313 lifecycle: &VectorIndexLifecycle,
314) -> Result<UpdateTableResponse> {
315 if request.on_demand_throughput.as_ref().is_some_and(|odt| {
323 odt.max_read_request_units.is_none() && odt.max_write_request_units.is_none()
324 }) {
325 request.on_demand_throughput = None;
326 }
327
328 validate_update_request(&request)?;
332
333 let meta = helpers::require_table(storage, &request.table_name).await?;
335
336 let current_billing_mode = meta.billing_mode.as_deref().unwrap_or("PROVISIONED");
337
338 let mut current_vixs: Vec<VectorIndex> = meta
341 .vector_index_definitions
342 .as_ref()
343 .map(|json| serde_json::from_str(json))
344 .transpose()
345 .map_err(|e| DynoxideError::InternalServerError(format!("Bad vector index JSON: {e}")))?
346 .unwrap_or_default();
347
348 if current_billing_mode == "PAY_PER_REQUEST"
352 && request.billing_mode.is_none()
353 && request.provisioned_throughput.is_some()
354 {
355 return Err(DynoxideError::ValidationException(
356 "One or more parameter values were invalid: \
357 Neither ReadCapacityUnits nor WriteCapacityUnits can be \
358 specified when BillingMode is PAY_PER_REQUEST"
359 .to_string(),
360 ));
361 }
362
363 if request.billing_mode.as_deref() == Some("PROVISIONED")
365 && request.provisioned_throughput.is_none()
366 {
367 return Err(DynoxideError::ValidationException(
368 "One or more parameter values were invalid: \
369 ProvisionedThroughput must be specified when BillingMode is PROVISIONED"
370 .to_string(),
371 ));
372 }
373
374 if request.billing_mode.as_deref() == Some("PROVISIONED") && !current_vixs.is_empty() {
380 return Err(DynoxideError::ValidationException(
381 "One or more parameter values were invalid: Tables with vector indexes must \
382 be in PAY_PER_REQUEST mode"
383 .to_string(),
384 ));
385 }
386
387 if let Some(ref odt) = request.on_demand_throughput {
395 let target_is_provisioned = match request.billing_mode.as_deref() {
396 Some("PAY_PER_REQUEST") => false,
397 Some(_) => true,
398 None => current_billing_mode == "PROVISIONED",
399 };
400 let members = [
401 ("MaxReadRequestUnits", odt.max_read_request_units),
402 ("MaxWriteRequestUnits", odt.max_write_request_units),
403 ];
404 if target_is_provisioned
405 && let Some((member, _)) = members.iter().find(|(_, v)| v.is_some())
406 {
407 return Err(DynoxideError::ValidationException(format!(
408 "One or more parameter values were invalid: {member} for \
409 OnDemandThroughput cannot be specified when the table BillingMode \
410 is PROVISIONED"
411 )));
412 }
413 for (member, value) in members {
416 if value.is_some_and(|v| v == 0 || v < -1) {
417 return Err(DynoxideError::ValidationException(format!(
418 "One or more parameter values were invalid: Requested {member} for \
419 OnDemandThroughput for table is outside of valid range"
420 )));
421 }
422 }
423 }
424
425 if let Some(ref pt) = request.provisioned_throughput
427 && let Some(obj) = pt.as_object()
428 {
429 let new_rcu = obj
430 .get("ReadCapacityUnits")
431 .and_then(|v| v.as_i64())
432 .unwrap_or(0);
433 let new_wcu = obj
434 .get("WriteCapacityUnits")
435 .and_then(|v| v.as_i64())
436 .unwrap_or(0);
437
438 let (cur_rcu, cur_wcu) = parse_current_throughput(&meta);
440
441 let billing_mode_unchanged = request.billing_mode.is_none()
442 || (request.billing_mode.as_deref() == Some("PROVISIONED")
443 && current_billing_mode == "PROVISIONED");
444
445 if new_rcu == cur_rcu && new_wcu == cur_wcu && billing_mode_unchanged {
446 return Err(DynoxideError::ValidationException(format!(
447 "The provisioned throughput for the table will not change. \
448 The requested value equals the current value. \
449 Current ReadCapacityUnits provisioned for the table: {}. \
450 Requested ReadCapacityUnits: {}. \
451 Current WriteCapacityUnits provisioned for the table: {}. \
452 Requested WriteCapacityUnits: {}. \
453 Refer to the Amazon DynamoDB Developer Guide for current limits \
454 and how to request higher limits.",
455 cur_rcu, new_rcu, cur_wcu, new_wcu
456 )));
457 }
458 }
459
460 if request.stream_specification.is_some() && !storage.supports_streams() {
465 return Err(crate::storage_backend::BackendError::Unsupported {
466 capability: "streams",
467 }
468 .into());
469 }
470
471 let mut current_gsis: Vec<GlobalSecondaryIndex> = meta
473 .gsi_definitions
474 .as_ref()
475 .map(|json| serde_json::from_str(json))
476 .transpose()
477 .map_err(|e| DynoxideError::InternalServerError(format!("Bad GSI JSON: {e}")))?
478 .unwrap_or_default();
479
480 let lsi_defs = crate::actions::lsi::parse_lsi_defs(&meta)?;
483
484 if let Some(ref updates) = request.global_secondary_index_updates {
486 for update in updates {
487 if let Some(ref upd) = update.update
488 && !current_gsis.iter().any(|g| g.index_name == upd.index_name)
489 {
490 return Err(DynoxideError::ValidationException(
493 "This operation cannot be performed with given input values. \
494 Please contact DynamoDB service team for more info: \
495 Action Blocked: IndexUpdate"
496 .to_string(),
497 ));
498 }
499 }
500 }
501
502 if let Some(ref updates) = request.global_secondary_index_updates
504 && updates.len() > 5
505 {
506 return Err(DynoxideError::LimitExceededException(
507 "Subscriber limit exceeded: Only 1 online index can be created or \
508 deleted simultaneously per table"
509 .to_string(),
510 ));
511 }
512
513 let mut attr_defs: Vec<AttributeDefinition> = serde_json::from_str(&meta.attribute_definitions)
519 .map_err(|e| DynoxideError::InternalServerError(format!("Bad attr defs JSON: {e}")))?;
520
521 if let Some(ref provided) = request.attribute_definitions {
522 for def in provided {
523 if !attr_defs
528 .iter()
529 .any(|d| d.attribute_name == def.attribute_name)
530 {
531 attr_defs.push(def.clone());
532 }
533 }
534 }
535
536 let key_schema = helpers::parse_key_schema(&meta)?;
538
539 if let Some(ref updates) = request.global_secondary_index_updates {
541 for update in updates {
542 if let Some(ref create) = update.create {
543 if current_gsis
544 .iter()
545 .any(|g| g.index_name == create.index_name)
546 {
547 return Err(DynoxideError::ValidationException(format!(
548 "One or more parameter values were invalid: \
549 Index already exists: {}",
550 create.index_name
551 )));
552 }
553 if current_vixs
558 .iter()
559 .any(|v| v.index_name == create.index_name)
560 {
561 return Err(DynoxideError::ValidationException(
562 "Attempting to create an index which already exists".to_string(),
563 ));
564 }
565 for elem in &create.key_schema {
575 let Some(vix) = current_vixs
576 .iter()
577 .find(|v| v.vector_attribute.attribute_name == elem.attribute_name)
578 else {
579 continue;
580 };
581 let Some(def) = request
582 .attribute_definitions
583 .as_deref()
584 .unwrap_or(&[])
585 .iter()
586 .find(|d| d.attribute_name == elem.attribute_name)
587 else {
588 continue;
589 };
590 let type_letter = match def.attribute_type {
591 crate::types::ScalarAttributeType::S => "S",
592 crate::types::ScalarAttributeType::N => "N",
593 crate::types::ScalarAttributeType::B => "B",
594 };
595 let key_type = match elem.key_type {
596 crate::types::KeyType::HASH => "HASH",
597 crate::types::KeyType::RANGE => "RANGE",
598 };
599 return Err(DynoxideError::ValidationException(format!(
600 "One or more parameter values were invalid: Attributes cannot be \
601 redefined. Please check that your attribute has the same type as \
602 previously defined. Existing schema: \
603 VectorIndexSchema:[VectorAttribute: key{{{attr}:L:{dims}}}] \
604 New schema: Schema:[SchemaElement: key{{{attr}:{type_letter}:{key_type}}}]",
605 attr = elem.attribute_name,
606 dims = vix.dimensions,
607 )));
608 }
609 let gsi_def = GlobalSecondaryIndex {
610 index_name: create.index_name.clone(),
611 key_schema: create.key_schema.clone(),
612 projection: create.projection.clone(),
613 provisioned_throughput: None,
614 };
615 validation::validate_gsi(
620 &gsi_def,
621 request.attribute_definitions.as_deref().unwrap_or(&[]),
622 )?;
623 }
624 if let Some(ref delete) = update.delete
625 && !current_gsis
626 .iter()
627 .any(|g| g.index_name == delete.index_name)
628 {
629 return Err(DynoxideError::ResourceNotFoundException(format!(
630 "Requested resource not found: Table: {} not found",
631 delete.index_name
632 )));
633 }
634 }
635 }
636
637 if let Some(ref updates) = request.vector_index_updates {
639 if updates.len() > 1 {
645 return Err(DynoxideError::LimitExceededException(
646 ONE_ONLINE_INDEX.to_string(),
647 ));
648 }
649 let creating = lifecycle.phases_of(
660 storage,
661 &request.table_name,
662 current_vixs.iter().map(|v| v.index_name.as_str()),
663 );
664 let targeted = updates.first().and_then(|u| u.delete.as_ref());
665 let cancels_the_creating_index =
666 targeted.is_some_and(|d| creating.get(&d.index_name).is_creating());
667 if !updates.is_empty() && creating.any_creating() && !cancels_the_creating_index {
668 return Err(DynoxideError::LimitExceededException(
669 ONE_ONLINE_INDEX.to_string(),
670 ));
671 }
672
673 if let Some(delete) = targeted
678 && !creating.get(&delete.index_name).accepts_delete()
679 {
680 return Err(DynoxideError::ResourceInUseException(format!(
681 "Attempt to change a resource which is still in use: Index creation is \
682 in resource allocation phase. Retry deletion during backfilling phase \
683 or when the index is active. Table: {} Index: {}",
684 request.table_name, delete.index_name
685 )));
686 }
687
688 for update in updates {
689 if let Some(ref create) = update.create {
690 let target_billing_mode = request
695 .billing_mode
696 .as_deref()
697 .unwrap_or(current_billing_mode);
698 if target_billing_mode != "PAY_PER_REQUEST" {
699 return Err(DynoxideError::ValidationException(
700 "One or more parameter values were invalid: Vector indexes are \
701 only supported for PAY_PER_REQUEST tables"
702 .to_string(),
703 ));
704 }
705
706 if current_vixs
713 .iter()
714 .any(|v| v.index_name == create.index_name)
715 || current_gsis
716 .iter()
717 .any(|g| g.index_name == create.index_name)
718 || lsi_defs.iter().any(|l| l.index_name == create.index_name)
719 {
720 return Err(DynoxideError::ValidationException(
721 "Attempting to create an index which already exists".to_string(),
722 ));
723 }
724
725 validation::validate_vector_index(
730 create,
731 request.attribute_definitions.as_deref().unwrap_or(&[]),
732 "vectorIndexUpdates.1.member.create",
733 )?;
734
735 if current_vixs.len() >= 5 {
741 return Err(DynoxideError::ValidationException(
742 "One or more parameter values were invalid: VectorIndex count \
743 exceeds the per-table limit of 5"
744 .to_string(),
745 ));
746 }
747
748 if current_vixs.iter().any(|v| {
753 v.vector_attribute.attribute_name == create.vector_attribute.attribute_name
754 && v.dimensions != create.dimensions
755 }) {
756 return Err(DynoxideError::ValidationException(format!(
757 "One or more parameter values were invalid: Conflicting attribute \
758 definition for '{}'. All VectorIndexes on the same vector attribute \
759 must use the same dimensions.",
760 create.vector_attribute.attribute_name
761 )));
762 }
763
764 let attr = create.vector_attribute.attribute_name.as_str();
770 if attr_defs.iter().any(|d| d.attribute_name == attr) {
771 return Err(DynoxideError::ValidationException(format!(
772 "One or more parameter values were invalid: Conflicting attribute \
773 definition for '{attr}'. An attribute cannot be defined in \
774 AttributeDefinitions when used as a VectorAttribute."
775 )));
776 }
777 }
778 if let Some(ref delete) = update.delete
779 && !current_vixs
780 .iter()
781 .any(|v| v.index_name == delete.index_name)
782 {
783 return Err(DynoxideError::ResourceNotFoundException(format!(
786 "Requested resource not found: Index {} for table {}",
787 delete.index_name, request.table_name
788 )));
789 }
790 }
791 }
792
793 let now = {
797 use std::sync::atomic::{AtomicU64, Ordering};
798 static LAST_TS: AtomicU64 = AtomicU64::new(0);
799 let wall = web_time::SystemTime::now()
800 .duration_since(web_time::UNIX_EPOCH)
801 .unwrap_or_default()
802 .as_secs_f64();
803 loop {
804 let prev_bits = LAST_TS.load(Ordering::SeqCst);
805 let prev_f = f64::from_bits(prev_bits);
806 let candidate = if wall > prev_f { wall } else { prev_f + 0.001 };
807 let candidate_bits = candidate.to_bits();
808 if LAST_TS
809 .compare_exchange(
810 prev_bits,
811 candidate_bits,
812 Ordering::SeqCst,
813 Ordering::SeqCst,
814 )
815 .is_ok()
816 {
817 break candidate;
818 }
819 }
820 };
821
822 let (cur_rcu, cur_wcu) = parse_current_throughput(&meta);
823 let is_pt_update = request.provisioned_throughput.is_some();
824 let (new_rcu, new_wcu) = if let Some(ref pt) = request.provisioned_throughput {
825 let obj = pt.as_object();
826 (
827 obj.and_then(|o| o.get("ReadCapacityUnits"))
828 .and_then(|v| v.as_i64())
829 .unwrap_or(0),
830 obj.and_then(|o| o.get("WriteCapacityUnits"))
831 .and_then(|v| v.as_i64())
832 .unwrap_or(0),
833 )
834 } else {
835 (cur_rcu, cur_wcu)
836 };
837
838 let is_increase = new_rcu > cur_rcu || new_wcu > cur_wcu;
839 let is_decrease = new_rcu < cur_rcu || new_wcu < cur_wcu;
840
841 let odt_change = request.on_demand_throughput.as_ref().map(|req_odt| {
846 let stored: crate::types::OnDemandThroughput = meta
847 .on_demand_throughput
848 .as_deref()
849 .and_then(|json| serde_json::from_str(json).ok())
850 .unwrap_or_default();
851 let echo = crate::types::OnDemandThroughput {
852 max_read_request_units: req_odt
853 .max_read_request_units
854 .or(stored.max_read_request_units),
855 max_write_request_units: req_odt
856 .max_write_request_units
857 .or(stored.max_write_request_units),
858 };
859 let strip = |v: Option<i64>| v.filter(|&v| v != -1);
860 let effective = crate::types::OnDemandThroughput {
861 max_read_request_units: strip(echo.max_read_request_units),
862 max_write_request_units: strip(echo.max_write_request_units),
863 };
864 (echo, effective)
865 });
866
867 helpers::with_write_transaction(storage, async {
869 if let Some(ref updates) = request.global_secondary_index_updates {
870 for update in updates {
871 if let Some(ref create) = update.create {
872 let gsi_def = GlobalSecondaryIndex {
873 index_name: create.index_name.clone(),
874 key_schema: create.key_schema.clone(),
875 projection: create.projection.clone(),
876 provisioned_throughput: None,
877 };
878
879 storage
880 .create_gsi_table(&request.table_name, &create.index_name)
881 .await?;
882
883 let gsi_p = gsi::gsi_to_def(&gsi_def)?;
884 backfill_gsi(storage, &request.table_name, &key_schema, &gsi_p).await?;
885
886 current_gsis.push(gsi_def);
887 }
888
889 if let Some(ref delete) = update.delete {
890 storage
891 .drop_gsi_table(&request.table_name, &delete.index_name)
892 .await?;
893 current_gsis.retain(|g| g.index_name != delete.index_name);
894 }
895 }
896 }
897
898 if let Some(ref updates) = request.vector_index_updates {
899 for update in updates {
900 if let Some(ref create) = update.create {
901 storage
902 .create_vector_table(&request.table_name, &create.index_name)
903 .await?;
904
905 backfill_vector_index(
911 storage,
912 &request.table_name,
913 &key_schema,
914 create,
915 &attr_defs,
916 )
917 .await?;
918
919 current_vixs.push(create.clone());
920 }
921
922 if let Some(ref delete) = update.delete {
923 storage
924 .drop_vector_table(&request.table_name, &delete.index_name)
925 .await?;
926 current_vixs.retain(|v| v.index_name != delete.index_name);
927 }
928 }
929 }
930
931 reconcile_attribute_definitions(
937 &mut attr_defs,
938 &key_schema,
939 ¤t_gsis,
940 &lsi_defs,
941 ¤t_vixs,
942 );
943
944 let attr_defs_json = serde_json::to_string(&attr_defs)
946 .map_err(|e| DynoxideError::InternalServerError(e.to_string()))?;
947 let gsi_json = if current_gsis.is_empty() {
948 None
949 } else {
950 Some(
951 serde_json::to_string(¤t_gsis)
952 .map_err(|e| DynoxideError::InternalServerError(e.to_string()))?,
953 )
954 };
955
956 storage
957 .update_table_metadata(&request.table_name, &attr_defs_json, gsi_json.as_deref())
958 .await?;
959
960 if request
964 .vector_index_updates
965 .as_ref()
966 .is_some_and(|u| !u.is_empty())
967 {
968 let vix_json = if current_vixs.is_empty() {
969 None
970 } else {
971 Some(
972 serde_json::to_string(¤t_vixs)
973 .map_err(|e| DynoxideError::InternalServerError(e.to_string()))?,
974 )
975 };
976 storage
977 .update_vector_index_definitions(&request.table_name, vix_json.as_deref())
978 .await?;
979 }
980
981 if is_pt_update {
983 let prev = parse_stored_throughput(&meta);
984 let mut stored = StoredProvisionedThroughput {
985 read_capacity_units: new_rcu,
986 write_capacity_units: new_wcu,
987 last_increase_date_time: prev.as_ref().and_then(|p| p.last_increase_date_time),
988 last_decrease_date_time: prev.as_ref().and_then(|p| p.last_decrease_date_time),
989 number_of_decreases_today: prev
990 .as_ref()
991 .and_then(|p| p.number_of_decreases_today)
992 .or(Some(0)),
993 };
994 if is_increase {
995 stored.last_increase_date_time = Some(now);
996 }
997 if is_decrease {
998 stored.last_decrease_date_time = Some(now);
999 stored.number_of_decreases_today =
1000 Some(stored.number_of_decreases_today.unwrap_or(0) + 1);
1001 }
1002 let pt_json = serde_json::to_string(&stored)
1003 .map_err(|e| DynoxideError::InternalServerError(e.to_string()))?;
1004 storage
1005 .update_provisioned_throughput(&request.table_name, &pt_json)
1006 .await?;
1007 }
1008
1009 if let Some(enabled) = request.deletion_protection_enabled {
1011 storage
1012 .update_deletion_protection(&request.table_name, enabled)
1013 .await?;
1014 }
1015
1016 if let Some(ref table_class) = request.table_class {
1018 storage
1019 .update_table_class(&request.table_name, table_class)
1020 .await?;
1021 }
1022
1023 if let Some((_, ref effective)) = odt_change {
1025 if effective.max_read_request_units.is_none()
1026 && effective.max_write_request_units.is_none()
1027 {
1028 storage
1029 .clear_on_demand_throughput(&request.table_name)
1030 .await?;
1031 } else {
1032 let json = serde_json::to_string(effective)
1033 .map_err(|e| DynoxideError::InternalServerError(e.to_string()))?;
1034 storage
1035 .update_on_demand_throughput(&request.table_name, &json)
1036 .await?;
1037 }
1038 }
1039
1040 if let Some(ref billing_mode) = request.billing_mode {
1042 storage
1043 .update_billing_mode(&request.table_name, billing_mode)
1044 .await?;
1045 if billing_mode == "PAY_PER_REQUEST" {
1046 storage
1048 .clear_provisioned_throughput(&request.table_name)
1049 .await?;
1050 } else if billing_mode == "PROVISIONED" {
1051 storage
1054 .clear_on_demand_throughput(&request.table_name)
1055 .await?;
1056 }
1057 }
1058
1059 if let Some(ref spec) = request.stream_specification {
1061 if spec.stream_enabled {
1062 let view_type = spec
1063 .stream_view_type
1064 .as_deref()
1065 .unwrap_or("NEW_AND_OLD_IMAGES");
1066 let label = streams::generate_stream_label(storage.clock());
1067 storage
1068 .enable_stream(&request.table_name, view_type, &label)
1069 .await?;
1070 } else {
1071 storage.disable_stream(&request.table_name).await?;
1072 }
1073 }
1074
1075 Ok(())
1076 })
1077 .await?;
1078
1079 if let Some(ref updates) = request.vector_index_updates {
1085 for update in updates {
1086 if let Some(ref create) = update.create {
1087 lifecycle.arm(storage, &request.table_name, &create.index_name);
1088 }
1089 if let Some(ref delete) = update.delete {
1090 lifecycle.disarm(&request.table_name, &delete.index_name);
1091 }
1092 }
1093 }
1094
1095 let updated_meta = helpers::require_table(storage, &request.table_name).await?;
1097 let vector_phases = lifecycle.phases_armed_on(storage, &request.table_name);
1098 let mut desc = build_table_description(&updated_meta, Some(0), Some(0), &vector_phases);
1099
1100 if let Some((echo, _)) = odt_change {
1103 desc.on_demand_throughput = Some(echo);
1104 }
1105
1106 if let Some(ref updates) = request.vector_index_updates
1114 && updates.iter().any(|u| u.create.is_some())
1115 {
1116 desc.table_status = "UPDATING".to_string();
1117 }
1118
1119 if is_pt_update {
1121 desc.table_status = "UPDATING".to_string();
1122
1123 let stored = parse_stored_throughput(&updated_meta);
1126 if let Some(ref mut pt) = desc.provisioned_throughput {
1127 pt.read_capacity_units = cur_rcu as u64;
1128 pt.write_capacity_units = cur_wcu as u64;
1129 if let Some(ref s) = stored {
1130 pt.last_increase_date_time = s.last_increase_date_time;
1131 pt.last_decrease_date_time = s.last_decrease_date_time;
1132 pt.number_of_decreases_today = s.number_of_decreases_today.unwrap_or(0);
1133 }
1134 }
1135 }
1136
1137 Ok(UpdateTableResponse {
1138 table_description: desc,
1139 })
1140}
1141
1142fn reconcile_attribute_definitions(
1154 attr_defs: &mut Vec<AttributeDefinition>,
1155 key_schema: &helpers::KeySchema,
1156 gsis: &[GlobalSecondaryIndex],
1157 lsi_defs: &[crate::actions::lsi::LsiDef],
1158 vixs: &[VectorIndex],
1159) {
1160 let mut used: std::collections::HashSet<&str> = std::collections::HashSet::new();
1161 used.insert(key_schema.partition_key.as_str());
1162 if let Some(ref sk) = key_schema.sort_key {
1163 used.insert(sk.as_str());
1164 }
1165 for g in gsis {
1166 for k in &g.key_schema {
1167 used.insert(k.attribute_name.as_str());
1168 }
1169 }
1170 for lsi in lsi_defs {
1171 used.insert(lsi.pk_attr.as_str());
1172 if let Some(ref sk) = lsi.sk_attr {
1173 used.insert(sk.as_str());
1174 }
1175 }
1176 for vix in vixs {
1177 if let Some(ref schema) = vix.search_schema {
1178 for elem in schema {
1179 used.insert(elem.attribute_name.as_str());
1180 }
1181 }
1182 }
1183 attr_defs.retain(|d| used.contains(d.attribute_name.as_str()));
1184}
1185
1186fn collect_gsi_update_errors(updates: &[GlobalSecondaryIndexUpdate], errors: &mut Vec<String>) {
1190 for (i, update) in updates.iter().enumerate() {
1191 if let Some(ref upd) = update.update {
1192 if upd.index_name.len() < 3 {
1194 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));
1195 }
1196 if !upd.index_name.is_empty()
1197 && !upd
1198 .index_name
1199 .chars()
1200 .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-' || c == '.')
1201 {
1202 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));
1203 }
1204 if let Some(ref pt) = upd.provisioned_throughput {
1206 let wcu = pt.write_capacity_units;
1207 let rcu = pt.read_capacity_units;
1208 if wcu.is_none() {
1209 errors.push(format!("Value null at 'globalSecondaryIndexUpdates.{}.member.update.provisionedThroughput.writeCapacityUnits' failed to satisfy constraint: Member must not be null", i + 1));
1210 } else if let Some(w) = wcu
1211 && w < 1
1212 {
1213 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));
1214 }
1215 if rcu.is_none() {
1216 errors.push(format!("Value null at 'globalSecondaryIndexUpdates.{}.member.update.provisionedThroughput.readCapacityUnits' failed to satisfy constraint: Member must not be null", i + 1));
1217 } else if let Some(r) = rcu
1218 && r < 1
1219 {
1220 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));
1221 }
1222 } else {
1223 errors.push(format!("Value null at 'globalSecondaryIndexUpdates.{}.member.update.provisionedThroughput' failed to satisfy constraint: Member must not be null", i + 1));
1224 }
1225 }
1226 }
1227}
1228
1229fn validate_update_request(request: &UpdateTableRequest) -> Result<()> {
1234 let mut errors = Vec::new();
1236
1237 if let Some(ref pt) = request.provisioned_throughput
1239 && let Some(obj) = pt.as_object()
1240 {
1241 let wcu = obj.get("WriteCapacityUnits");
1242 let rcu = obj.get("ReadCapacityUnits");
1243 if wcu.is_none() || wcu == Some(&serde_json::Value::Null) {
1244 errors.push("Value null at 'provisionedThroughput.writeCapacityUnits' failed to satisfy constraint: Member must not be null".to_string());
1245 } else if let Some(w) = wcu.and_then(|v| v.as_i64())
1246 && w < 1
1247 {
1248 errors.push(format!("Value '{}' at 'provisionedThroughput.writeCapacityUnits' failed to satisfy constraint: Member must have value greater than or equal to 1", w));
1249 }
1250 if rcu.is_none() || rcu == Some(&serde_json::Value::Null) {
1251 errors.push("Value null at 'provisionedThroughput.readCapacityUnits' failed to satisfy constraint: Member must not be null".to_string());
1252 } else if let Some(r) = rcu.and_then(|v| v.as_i64())
1253 && r < 1
1254 {
1255 errors.push(format!("Value '{}' at 'provisionedThroughput.readCapacityUnits' failed to satisfy constraint: Member must have value greater than or equal to 1", r));
1256 }
1257 }
1258
1259 if let Some(ref updates) = request.global_secondary_index_updates {
1261 collect_gsi_update_errors(updates, &mut errors);
1262 }
1263
1264 errors.truncate(10);
1266
1267 if !errors.is_empty() {
1268 let prefix = format!(
1269 "{} validation error{} detected: ",
1270 errors.len(),
1271 if errors.len() == 1 { "" } else { "s" }
1272 );
1273 return Err(DynoxideError::ValidationException(format!(
1274 "{}{}",
1275 prefix,
1276 errors.join("; ")
1277 )));
1278 }
1279
1280 if let Some(ref bm) = request.billing_mode
1284 && bm != "PROVISIONED"
1285 && bm != "PAY_PER_REQUEST"
1286 {
1287 return Err(DynoxideError::ValidationException(format!(
1288 "1 validation error detected: Value '{}' at 'billingMode' \
1289 failed to satisfy constraint: Member must satisfy enum value set: \
1290 [PROVISIONED, PAY_PER_REQUEST]",
1291 bm
1292 )));
1293 }
1294
1295 if let Some(ref tc) = request.table_class
1297 && tc != "STANDARD"
1298 && tc != "STANDARD_INFREQUENT_ACCESS"
1299 {
1300 return Err(DynoxideError::ValidationException(format!(
1301 "1 validation error detected: Value '{tc}' at 'tableClass' failed to satisfy \
1302 constraint: Member must satisfy enum value set: \
1303 [STANDARD, STANDARD_INFREQUENT_ACCESS]"
1304 )));
1305 }
1306
1307 if request.billing_mode.as_deref() == Some("PAY_PER_REQUEST")
1309 && request.provisioned_throughput.is_some()
1310 {
1311 return Err(DynoxideError::ValidationException(
1312 "One or more parameter values were invalid: \
1313 Neither ReadCapacityUnits nor WriteCapacityUnits can be \
1314 specified when BillingMode is PAY_PER_REQUEST"
1315 .to_string(),
1316 ));
1317 }
1318
1319 if let Some(ref pt) = request.provisioned_throughput
1321 && let Some(obj) = pt.as_object()
1322 {
1323 let rcu = obj
1324 .get("ReadCapacityUnits")
1325 .and_then(|v| v.as_i64())
1326 .unwrap_or(0);
1327 let wcu = obj
1328 .get("WriteCapacityUnits")
1329 .and_then(|v| v.as_i64())
1330 .unwrap_or(0);
1331 const MAX_THROUGHPUT: i64 = 1_000_000_000_000;
1332 if rcu > MAX_THROUGHPUT {
1333 return Err(DynoxideError::ValidationException(format!(
1334 "Given value {} for ReadCapacityUnits is out of bounds",
1335 rcu
1336 )));
1337 }
1338 if wcu > MAX_THROUGHPUT {
1339 return Err(DynoxideError::ValidationException(format!(
1340 "Given value {} for WriteCapacityUnits is out of bounds",
1341 wcu
1342 )));
1343 }
1344 }
1345
1346 let no_config_change = request.provisioned_throughput.is_none()
1352 && request.billing_mode.is_none()
1353 && request.stream_specification.is_none()
1354 && request.deletion_protection_enabled.is_none()
1355 && request.table_class.is_none()
1356 && request.on_demand_throughput.is_none();
1357 let no_gsi_change = request
1358 .global_secondary_index_updates
1359 .as_ref()
1360 .is_none_or(|updates| updates.is_empty());
1361 let no_vector_change = request
1362 .vector_index_updates
1363 .as_ref()
1364 .is_none_or(|updates| updates.is_empty());
1365 if no_gsi_change && no_config_change && no_vector_change {
1366 return Err(DynoxideError::ValidationException(
1367 "At least one of ProvisionedThroughput, BillingMode, UpdateStreamEnabled, GlobalSecondaryIndexUpdates or SSESpecification or ReplicaUpdates is required".to_string(),
1368 ));
1369 }
1370
1371 if let Some(ref updates) = request.global_secondary_index_updates {
1373 for update in updates {
1375 if update.update.is_none() && update.create.is_none() && update.delete.is_none() {
1376 return Err(DynoxideError::ValidationException(
1377 "One or more parameter values were invalid: One of GlobalSecondaryIndexUpdate.Update, GlobalSecondaryIndexUpdate.Create, GlobalSecondaryIndexUpdate.Delete must not be null".to_string(),
1378 ));
1379 }
1380 }
1381
1382 let mut seen_names = std::collections::HashSet::new();
1384 for update in updates {
1385 let name = if let Some(ref u) = update.update {
1386 Some(u.index_name.as_str())
1387 } else if let Some(ref c) = update.create {
1388 Some(c.index_name.as_str())
1389 } else {
1390 update.delete.as_ref().map(|d| d.index_name.as_str())
1391 };
1392 if let Some(name) = name
1393 && !seen_names.insert(name.to_string())
1394 {
1395 return Err(DynoxideError::ValidationException(format!(
1396 "One or more parameter values were invalid: Only one global secondary index update per index is allowed simultaneously. Index: {}",
1397 name
1398 )));
1399 }
1400 }
1401 }
1402
1403 if let Some(ref updates) = request.vector_index_updates {
1407 for update in updates {
1408 if update.create.is_none() && update.delete.is_none() {
1409 return Err(DynoxideError::ValidationException(
1410 "One or more parameter values were invalid: One of \
1411 VectorIndexUpdate.Create, VectorIndexUpdate.Delete must not be null"
1412 .to_string(),
1413 ));
1414 }
1415 if update.create.is_some() && update.delete.is_some() {
1420 return Err(DynoxideError::ValidationException(
1421 "One or more parameter values were invalid: Only one vector index \
1422 action is allowed per VectorIndexUpdate object"
1423 .to_string(),
1424 ));
1425 }
1426 }
1427 }
1428
1429 Ok(())
1430}
1431
1432#[derive(Debug, Clone, Default, Serialize, Deserialize)]
1434struct StoredProvisionedThroughput {
1435 #[serde(rename = "ReadCapacityUnits")]
1436 read_capacity_units: i64,
1437 #[serde(rename = "WriteCapacityUnits")]
1438 write_capacity_units: i64,
1439 #[serde(
1440 rename = "LastIncreaseDateTime",
1441 skip_serializing_if = "Option::is_none"
1442 )]
1443 last_increase_date_time: Option<f64>,
1444 #[serde(
1445 rename = "LastDecreaseDateTime",
1446 skip_serializing_if = "Option::is_none"
1447 )]
1448 last_decrease_date_time: Option<f64>,
1449 #[serde(
1450 rename = "NumberOfDecreasesToday",
1451 skip_serializing_if = "Option::is_none"
1452 )]
1453 number_of_decreases_today: Option<u64>,
1454}
1455
1456fn parse_current_throughput(meta: &crate::storage::TableMetadata) -> (i64, i64) {
1458 parse_stored_throughput(meta)
1459 .map(|pt| (pt.read_capacity_units, pt.write_capacity_units))
1460 .unwrap_or((0, 0))
1461}
1462
1463fn parse_stored_throughput(
1465 meta: &crate::storage::TableMetadata,
1466) -> Option<StoredProvisionedThroughput> {
1467 meta.provisioned_throughput
1468 .as_ref()
1469 .and_then(|pt_json| serde_json::from_str(pt_json).ok())
1470}
1471
1472async fn backfill_gsi<S: StorageBackend>(
1474 storage: &S,
1475 table_name: &str,
1476 key_schema: &helpers::KeySchema,
1477 gsi_def: &gsi::GsiDef,
1478) -> Result<()> {
1479 const BATCH_SIZE: usize = 1000;
1480 let mut last_pk: Option<String> = None;
1481 let mut last_sk: Option<String> = None;
1482
1483 loop {
1484 let items = storage
1485 .scan_items(
1486 table_name,
1487 &crate::storage::ScanParams {
1488 limit: Some(BATCH_SIZE),
1489 exclusive_start_pk: last_pk.as_deref(),
1490 exclusive_start_sk: last_sk.as_deref(),
1491 ..Default::default()
1492 },
1493 )
1494 .await?;
1495
1496 if items.is_empty() {
1497 break;
1498 }
1499
1500 let mut rows: Vec<crate::storage_backend::GsiItemRow> = Vec::new();
1501 for (pk, sk, item_json) in &items {
1502 let item: crate::types::Item = serde_json::from_str(item_json)
1503 .map_err(|e| DynoxideError::InternalServerError(format!("Bad item JSON: {e}")))?;
1504
1505 if let Some((gsi_pk, gsi_sk)) = gsi_def.index_key_strings(&item) {
1507 let projected = gsi::build_index_item(
1508 &item,
1509 gsi_def,
1510 &key_schema.partition_key,
1511 key_schema.sort_key.as_deref(),
1512 );
1513 let projected_json = serde_json::to_string(&projected)
1514 .map_err(|e| DynoxideError::InternalServerError(e.to_string()))?;
1515
1516 rows.push(crate::storage_backend::GsiItemRow {
1517 gsi_pk,
1518 gsi_sk,
1519 table_pk: pk.clone(),
1520 table_sk: sk.clone(),
1521 item_json: projected_json,
1522 });
1523 }
1524 }
1525
1526 storage
1527 .insert_gsi_items(table_name, &gsi_def.index_name, &rows)
1528 .await?;
1529
1530 let last = &items[items.len() - 1];
1531 last_pk = Some(last.0.clone());
1532 last_sk = Some(last.1.clone());
1533
1534 if items.len() < BATCH_SIZE {
1535 break;
1536 }
1537 }
1538
1539 Ok(())
1540}
1541
1542async fn backfill_vector_index<S: StorageBackend>(
1550 storage: &S,
1551 table_name: &str,
1552 key_schema: &helpers::KeySchema,
1553 vix: &VectorIndex,
1554 attr_defs: &[AttributeDefinition],
1555) -> Result<()> {
1556 const BATCH_SIZE: usize = 1000;
1557 let mut last_pk: Option<String> = None;
1558 let mut last_sk: Option<String> = None;
1559
1560 loop {
1561 let items = storage
1562 .scan_items(
1563 table_name,
1564 &crate::storage::ScanParams {
1565 limit: Some(BATCH_SIZE),
1566 exclusive_start_pk: last_pk.as_deref(),
1567 exclusive_start_sk: last_sk.as_deref(),
1568 ..Default::default()
1569 },
1570 )
1571 .await?;
1572
1573 if items.is_empty() {
1574 break;
1575 }
1576
1577 let mut rows: Vec<crate::storage_backend::VectorItemRow> = Vec::new();
1578 for (pk, sk, item_json) in &items {
1579 let item: Item = serde_json::from_str(item_json)
1580 .map_err(|e| DynoxideError::InternalServerError(format!("Bad item JSON: {e}")))?;
1581
1582 if let Some(row) = super::vector_index::vector_index_row(
1583 &item,
1584 vix,
1585 &key_schema.partition_key,
1586 key_schema.sort_key.as_deref(),
1587 attr_defs,
1588 pk,
1589 sk,
1590 )? {
1591 rows.push(row);
1592 }
1593 }
1594
1595 storage
1596 .insert_vector_items(table_name, &vix.index_name, &rows)
1597 .await?;
1598
1599 let last = &items[items.len() - 1];
1600 last_pk = Some(last.0.clone());
1601 last_sk = Some(last.1.clone());
1602
1603 if items.len() < BATCH_SIZE {
1604 break;
1605 }
1606 }
1607
1608 Ok(())
1609}