diff --git a/crates/core/src/types/stream.rs b/crates/core/src/types/stream.rs index 870169b9a..ad0edad0c 100755 --- a/crates/core/src/types/stream.rs +++ b/crates/core/src/types/stream.rs @@ -119,6 +119,14 @@ pub struct StreamDescription { pub stream_arn: String, #[serde(rename = "StreamLabel")] pub stream_label: String, + /// When the stream was requested, epoch seconds. The service returns it + /// on every stream, enabled or disabled. Backends leave it `None`; the + /// engine derives it from `stream_label` at the response boundary. + #[serde( + rename = "CreationRequestDateTime", + skip_serializing_if = "Option::is_none" + )] + pub creation_request_date_time: Option, #[serde(rename = "StreamStatus")] pub stream_status: StreamStatus, #[serde(rename = "StreamViewType")] diff --git a/crates/core/src/validation/key_definitions.rs b/crates/core/src/validation/key_definitions.rs new file mode 100644 index 000000000..ee14e246f --- /dev/null +++ b/crates/core/src/validation/key_definitions.rs @@ -0,0 +1,169 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Checks that a key schema's attributes are defined in `AttributeDefinitions`, +//! shared by `CreateTable` and an `UpdateTable` GSI `Create`. +//! +//! The service's rules, measured on 2026-10-25 and 2026-10-26 (us-east-1): +//! +//! - A name defined twice answers `Attribute Name is duplicated: `, +//! naming the first name whose second definition comes first in the list. +//! The service checks it after the table billing checks, the one-name-twice +//! rule and the length rule below, and before the other key schema checks; +//! ExtendDB checks it after all key schema shape checks. +//! - Each key schema is then checked in turn: the table's, each LSI's, each +//! GSI's. A schema with more keys than there are definitions answers +//! `Invalid KeySchema: Some index key attribute have no definition`. A +//! schema with a key that has no definition answers `Some index key +//! attributes are not defined in AttributeDefinitions`, listing every key of +//! that schema and every definition. +//! +//! The service lists the keys and definitions in an order that follows neither +//! the request nor the alphabet; ExtendDB lists them in request order. + +use std::collections::HashSet; + +use crate::error::DynamoDbError; +use crate::types::{AttributeDefinition, KeySchemaElement}; + +/// Refuse a name defined twice in `defs`. +/// +/// # Errors +/// +/// `ValidationException` `Attribute Name is duplicated: `. +pub fn validate_unique_attribute_definitions( + defs: &[AttributeDefinition], +) -> Result<(), DynamoDbError> { + let mut seen = HashSet::with_capacity(defs.len()); + for def in defs { + if !seen.insert(def.attribute_name.as_str()) { + return Err(DynamoDbError::ValidationException(format!( + "Attribute Name is duplicated: {}", + def.attribute_name + ))); + } + } + Ok(()) +} + +/// Refuse a key schema whose keys are not all defined in `defs`. +/// +/// # Errors +/// +/// `ValidationException` with the `Invalid KeySchema` sentence when the schema +/// has more keys than `defs` has entries, otherwise with the `Some index key +/// attributes are not defined` sentence when any key has no definition. +pub fn validate_key_schema_defined( + key_schema: &[KeySchemaElement], + defs: &[AttributeDefinition], +) -> Result<(), DynamoDbError> { + if defs.len() < key_schema.len() { + return Err(DynamoDbError::ValidationException( + "Invalid KeySchema: Some index key attribute have no definition".to_owned(), + )); + } + if key_schema + .iter() + .all(|k| defs.iter().any(|d| d.attribute_name == k.attribute_name)) + { + return Ok(()); + } + let keys: Vec<&str> = key_schema + .iter() + .map(|k| k.attribute_name.as_str()) + .collect(); + let names: Vec<&str> = defs.iter().map(|d| d.attribute_name.as_str()).collect(); + Err(DynamoDbError::ValidationException(format!( + "One or more parameter values were invalid: Some index key attributes are not defined in \ + AttributeDefinitions. Keys: [{}], AttributeDefinitions: [{}]", + keys.join(", "), + names.join(", ") + ))) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::types::{KeyType, ScalarAttributeType}; + + fn defs(names: &[&str]) -> Vec { + names + .iter() + .map(|n| AttributeDefinition { + attribute_name: (*n).to_owned(), + attribute_type: ScalarAttributeType::S, + }) + .collect() + } + + fn ks(hash: &str, range: Option<&str>) -> Vec { + let mut k = vec![KeySchemaElement { + attribute_name: hash.to_owned(), + key_type: KeyType::Hash, + }]; + if let Some(r) = range { + k.push(KeySchemaElement { + attribute_name: r.to_owned(), + key_type: KeyType::Range, + }); + } + k + } + + fn message(r: Result<(), DynamoDbError>) -> String { + match r { + Err(DynamoDbError::ValidationException(m)) => m, + other => panic!("expected a ValidationException, got {other:?}"), + } + } + + const INVALID: &str = "Invalid KeySchema: Some index key attribute have no definition"; + + #[test] + fn the_first_repeated_name_is_named() { + assert_eq!( + message(validate_unique_attribute_definitions(&defs(&[ + "b", "a", "b", "a" + ]))), + "Attribute Name is duplicated: b" + ); + assert!(validate_unique_attribute_definitions(&defs(&["a", "b"])).is_ok()); + } + + #[test] + fn fewer_definitions_than_keys_is_the_key_schema_sentence() { + assert_eq!( + message(validate_key_schema_defined( + &ks("g", Some("r")), + &defs(&["pk"]) + )), + INVALID + ); + assert_eq!( + message(validate_key_schema_defined(&ks("g", None), &[])), + INVALID + ); + } + + #[test] + fn a_missing_key_lists_every_key_and_every_definition() { + assert_eq!( + message(validate_key_schema_defined( + &ks("g", Some("r")), + &defs(&["pk", "g"]) + )), + "One or more parameter values were invalid: Some index key attributes are not defined \ + in AttributeDefinitions. Keys: [g, r], AttributeDefinitions: [pk, g]" + ); + // As many definitions as keys, none of them a key. + assert_eq!( + message(validate_key_schema_defined( + &ks("g", Some("r")), + &defs(&["pk", "z"]) + )), + "One or more parameter values were invalid: Some index key attributes are not defined \ + in AttributeDefinitions. Keys: [g, r], AttributeDefinitions: [pk, z]" + ); + assert!(validate_key_schema_defined(&ks("g", Some("r")), &defs(&["r", "g"])).is_ok()); + } +} diff --git a/crates/core/src/validation/mod.rs b/crates/core/src/validation/mod.rs index c144b46b6..f7221ac24 100755 --- a/crates/core/src/validation/mod.rs +++ b/crates/core/src/validation/mod.rs @@ -1,8 +1,10 @@ // Copyright 2026 ExtendDB contributors // SPDX-License-Identifier: Apache-2.0 +mod key_definitions; pub mod number; pub mod vector_item; +pub use key_definitions::{validate_key_schema_defined, validate_unique_attribute_definitions}; pub use vector_item::{ MAX_HASH_KEY_SIZE, MAX_INLINE_FILTER_SIZE, validate_vector_write, validate_vector_write_changed, vector_components, vector_norm, @@ -101,8 +103,10 @@ pub fn validate_create_table( validate_key_schema(input, limits.allow_multipart_table_keys)?; validate_gsi_key_schemas(input)?; validate_lsi_key_schemas(input)?; - validate_attribute_definitions(input)?; + // The service reports a table billing fault before any definition fault + // (measured 2026-10-26). validate_provisioned_throughput(input)?; + validate_attribute_definitions(input)?; validate_gsi_provisioned_throughput(input)?; validate_gsi_throughput_minimum(input)?; validate_gsi_count(input, limits)?; @@ -943,14 +947,16 @@ fn validate_attribute_definitions(input: &CreateTableInput) -> Result<(), Dynamo } } - // Every key attribute must have a definition - for attr in &key_attrs { - if !def_names.contains(attr) { - return Err(DynamoDbError::ValidationException(format!( - "One or more parameter values were invalid: Some index key attributes are not defined in AttributeDefinitions. Keys: [{attr}], AttributeDefinitions: [{}]", - format_attr_defs(&input.attribute_definitions) - ))); - } + // Every key attribute must have a definition: the table's key schema, then + // each LSI's, then each GSI's (measured order, see key_definitions.rs). + validate_unique_attribute_definitions(&input.attribute_definitions)?; + let defs = &input.attribute_definitions; + validate_key_schema_defined(&input.key_schema, defs)?; + for lsi in input.local_secondary_indexes.iter().flatten() { + validate_key_schema_defined(&lsi.key_schema, defs)?; + } + for gsi in input.global_secondary_indexes.iter().flatten() { + validate_key_schema_defined(&gsi.key_schema, defs)?; } // Every definition must be used by a key @@ -966,13 +972,6 @@ fn validate_attribute_definitions(input: &CreateTableInput) -> Result<(), Dynamo Ok(()) } -fn format_attr_defs(defs: &[AttributeDefinition]) -> String { - defs.iter() - .map(|d| d.attribute_name.clone()) - .collect::>() - .join(", ") -} - /// Validate table-level throughput against the effective billing mode. /// /// # Errors diff --git a/crates/engine/src/streams.rs b/crates/engine/src/streams.rs index 83b297253..9ea574d95 100755 --- a/crates/engine/src/streams.rs +++ b/crates/engine/src/streams.rs @@ -7,9 +7,10 @@ use extenddb_core::error::DynamoDbError; use extenddb_core::types::{ DescribeStreamInput, DescribeStreamOutput, GetRecordsInput, GetRecordsOutput, GetShardIteratorInput, GetShardIteratorOutput, ListStreamsInput, ListStreamsOutput, - ShardIteratorType, + ShardIteratorType, StreamDescription, }; use extenddb_storage::error::StorageError; +use extenddb_storage::util::stream_label_creation_date_time; use serde_json::Value; use crate::OperationContext; @@ -35,11 +36,19 @@ pub async fn handle_describe_stream( .map_err(storage_to_dynamo)?; let output = DescribeStreamOutput { - stream_description: desc, + stream_description: with_creation_request_date_time(desc), }; serialize_output(&output) } +/// The service returns `CreationRequestDateTime` on every stream, enabled or +/// disabled. The label is the creation instant to the millisecond, so it is +/// derived here for every backend rather than stored by each. +fn with_creation_request_date_time(mut desc: StreamDescription) -> StreamDescription { + desc.creation_request_date_time = stream_label_creation_date_time(&desc.stream_label); + desc +} + /// Handle `ListStreams`. /// /// # Errors @@ -362,6 +371,49 @@ fn storage_to_dynamo(e: StorageError) -> DynamoDbError { #[cfg(test)] mod tests { + /// `DescribeStream` renders `CreationRequestDateTime` as the label's + /// instant in epoch seconds, next to the members it already had, and + /// omits the member only when the label is not a timestamp. + #[test] + fn describe_stream_output_carries_creation_request_date_time() { + use extenddb_core::types::{ + DescribeStreamOutput, StreamDescription, StreamStatus, StreamViewType, + }; + let desc = StreamDescription { + creation_request_date_time: None, + stream_arn: "arn:aws:dynamodb:us-east-1:1:table/T/stream/2026-10-24T11:31:45.748" + .to_owned(), + stream_label: "2026-10-24T11:31:45.748".to_owned(), + stream_status: StreamStatus::Disabled, + stream_view_type: StreamViewType::KeysOnly, + table_name: "T".to_owned(), + key_schema: Vec::new(), + shards: Vec::new(), + last_evaluated_shard_id: None, + }; + let v = crate::serialize_output(&DescribeStreamOutput { + stream_description: super::with_creation_request_date_time(desc.clone()), + }) + .unwrap(); + let sd = &v["StreamDescription"]; + assert_eq!(sd["CreationRequestDateTime"], 1_792_841_505.748); + assert_eq!(sd["StreamLabel"], "2026-10-24T11:31:45.748"); + assert_eq!(sd["StreamStatus"], "DISABLED"); + + let v = crate::serialize_output(&DescribeStreamOutput { + stream_description: super::with_creation_request_date_time(StreamDescription { + stream_label: "not-a-label".to_owned(), + ..desc + }), + }) + .unwrap(); + assert!( + v["StreamDescription"] + .get("CreationRequestDateTime") + .is_none() + ); + } + /// AT_SEQUENCE_NUMBER converts to AFTER by subtracting 1 and padding to /// the backend's stored width. Verify that an unpadded client input is /// normalised correctly for both the 21-digit (postgres/sqlite/mongodb) diff --git a/crates/engine/src/update_table.rs b/crates/engine/src/update_table.rs index 6db763f82..9b965b3ca 100755 --- a/crates/engine/src/update_table.rs +++ b/crates/engine/src/update_table.rs @@ -103,22 +103,12 @@ pub async fn handle_update_table( )); } - // Validate: enabling streams requires a view type. - if let Some(spec) = &input.stream_specification - && spec.stream_enabled - && spec.stream_view_type.is_none() - { - return Err(DynamoDbError::ValidationException( - "StreamViewType must be specified when StreamEnabled is true".to_owned(), - )); - } - // Switching to PROVISIONED requires explicit throughput values. if matches!(input.billing_mode, Some(BillingMode::Provisioned)) && input.provisioned_throughput.is_none() { return Err(DynamoDbError::ValidationException( - "One or more parameter values were invalid: ProvisionedThroughput must be specified when changing BillingMode to PROVISIONED".to_owned(), + "One or more parameter values were invalid: ProvisionedThroughput must be specified when BillingMode is PROVISIONED".to_owned(), )); } @@ -141,6 +131,7 @@ pub async fn handle_update_table( } validate_gsi_updates(&input)?; + validate_stream_specification(&input)?; // A created GSI's projected attributes count toward the table's limit // together with the indexes it already has: the total CreateTable checks, @@ -203,6 +194,25 @@ pub async fn handle_update_table( serialize_output(&output) } +/// Enabling a stream requires a view type. The sentence is the service's +/// (measured 2026-10-24 on a never-streamed table and on a table whose stream +/// is enabled). The service reports a defective index Create and a missing +/// ProvisionedThroughput before this (measured 2026-10-25), and this before +/// the table lookup, so the call sits after the billing and index rules and +/// before the backend call, where it also precedes the already-enabled +/// refusal. The `CreateTable` form has its own sentence. +fn validate_stream_specification(input: &UpdateTableInput) -> Result<(), DynamoDbError> { + if let Some(spec) = &input.stream_specification + && spec.stream_enabled + && spec.stream_view_type.is_none() + { + return Err(DynamoDbError::ValidationException( + "One or more parameter values were invalid: If stream is being enabled then UpdateViewType is required".to_owned(), + )); + } + Ok(()) +} + /// The per-entry rules for `GlobalSecondaryIndexUpdates`, checked before any /// storage read: each entry has exactly one of Create, Update, or Delete, /// and a Create has a valid name, a key schema of defined attributes, and @@ -239,25 +249,19 @@ fn validate_gsi_updates(input: &UpdateTableInput) -> Result<(), DynamoDbError> { "One or more parameter values were invalid: KeySchema must not be empty for GSI creation".to_owned(), )); } - // Validate that all key attributes are defined in AttributeDefinitions - let attr_defs = input.attribute_definitions.as_deref().unwrap_or(&[]); - for ks in &create.key_schema { - if !attr_defs - .iter() - .any(|ad| ad.attribute_name == ks.attribute_name) - { - return Err(DynamoDbError::ValidationException(format!( - "One or more parameter values were invalid: Some index key attributes are not defined in AttributeDefinitions. \ - Keys: [{}], AttributeDefinitions: [{}]", - ks.attribute_name, - attr_defs - .iter() - .map(|ad| ad.attribute_name.as_str()) - .collect::>() - .join(", ") - ))); - } - } + // Every key attribute must be in AttributeDefinitions. The + // service's sentence depends on the member's shape (measured + // 2026-10-25 and 2026-10-26): absent names the index; a name + // defined twice is refused first; then the CreateTable rules + // for the index's key schema (core's key_definitions.rs). + let Some(defs) = &input.attribute_definitions else { + return Err(DynamoDbError::ValidationException(format!( + "One or more parameter values were invalid: AttributeDefinitions is not specified for index: {}", + create.index_name + ))); + }; + extenddb_core::validation::validate_unique_attribute_definitions(defs)?; + extenddb_core::validation::validate_key_schema_defined(&create.key_schema, defs)?; } if let Some(delete) = &update.delete { extenddb_core::validation::validate_index_name(&delete.index_name)?; @@ -347,157 +351,5 @@ fn update_table_err_to_dynamo( } #[cfg(test)] -mod tests { - use super::*; - use extenddb_storage::error::StorageError; - - #[test] - fn a_created_gsi_throughput_below_one_is_refused() { - let input = |pt: Value| -> UpdateTableInput { - serde_json::from_value(serde_json::json!({ - "TableName": "t", - "AttributeDefinitions": [{ "AttributeName": "g", "AttributeType": "S" }], - "GlobalSecondaryIndexUpdates": [{ "Create": { - "IndexName": "gix", - "KeySchema": [{ "AttributeName": "g", "KeyType": "HASH" }], - "Projection": { "ProjectionType": "ALL" }, - "ProvisionedThroughput": pt, - }}], - })) - .unwrap() - }; - for (read, write) in [(0, 0), (0, 1), (1, 0), (-1, 5)] { - let pt = serde_json::json!({ "ReadCapacityUnits": read, "WriteCapacityUnits": write }); - match validate_gsi_updates(&input(pt)).unwrap_err() { - DynamoDbError::ValidationException(m) => { - assert!(m.ends_with("for index: gix"), "{m}"); - } - other => panic!("({read}, {write}): {other:?}"), - } - } - let ok = serde_json::json!({ "ReadCapacityUnits": 1, "WriteCapacityUnits": 1 }); - validate_gsi_updates(&input(ok)).unwrap(); - } - - #[test] - fn projected_total_counts_kept_indexes_and_created_ones() { - use extenddb_core::types::{ - GlobalSecondaryIndexUpdate, GsiDescription, LsiDescription, Projection, ProjectionType, - TableDescription, - }; - let include = |n: usize| Projection { - projection_type: ProjectionType::Include, - non_key_attributes: Some((0..n).map(|i| format!("a{i}")).collect()), - }; - let existing = TableDescription { - global_secondary_indexes: Some(vec![ - GsiDescription { - index_name: "kept".to_owned(), - key_schema: Vec::new(), - projection: include(30), - index_status: "ACTIVE".to_owned(), - provisioned_throughput: None, - index_size_bytes: 0, - item_count: 0, - index_arn: String::new(), - }, - GsiDescription { - index_name: "dropped".to_owned(), - key_schema: Vec::new(), - projection: include(50), - index_status: "ACTIVE".to_owned(), - provisioned_throughput: None, - index_size_bytes: 0, - item_count: 0, - index_arn: String::new(), - }, - ]), - local_secondary_indexes: Some(vec![LsiDescription { - index_name: "lsi".to_owned(), - key_schema: Vec::new(), - projection: include(20), - index_size_bytes: 0, - item_count: 0, - index_arn: String::new(), - }]), - ..Default::default() - }; - let create = |n: usize| -> GlobalSecondaryIndexUpdate { - serde_json::from_value(serde_json::json!({ "Create": { - "IndexName": "new", "KeySchema": [], - "Projection": serde_json::to_value(include(n)).unwrap(), - }})) - .unwrap() - }; - let delete: GlobalSecondaryIndexUpdate = - serde_json::from_value(serde_json::json!({ "Delete": { "IndexName": "dropped" } })) - .unwrap(); - - assert_eq!(projected_after_gsi_updates(&existing, &[create(1)]), 101); - assert_eq!( - projected_after_gsi_updates(&existing, &[create(1), delete]), - 51 - ); - assert!( - extenddb_core::validation::validate_projected_attribute_total( - projected_after_gsi_updates(&existing, &[create(1)]) - ) - .is_err() - ); - } - - /// A backend that cannot serve the request reports a 400, not a 500. - /// - /// This arm was missing while CreateTable had it, so the same refusal - /// answered differently depending on which operation carried it. It matters - /// for a backend whose vector capability is decided at runtime rather than at - /// compile time: its refusal arrives through this path. - #[test] - fn an_unsupported_feature_is_a_validation_exception() { - let err = update_table_err_to_dynamo( - StorageError::Unsupported("Vector indexes are not supported here".to_owned()), - "t", - ); - match err { - DynamoDbError::ValidationException(msg) => { - assert_eq!(msg, "Vector indexes are not supported here"); - } - other => panic!("expected ValidationException, got {other:?}"), - } - } - - /// The backend's whole message survives, unwrapped and unprefixed, because - /// the service's own wording for this case names both the table and the index - /// and no layer above the backend knows either. - #[test] - fn a_resource_in_use_refusal_keeps_the_backend_message() { - let measured = extenddb_core::types::vector_index_delete_in_allocation_phase("t", "vidx"); - let err = update_table_err_to_dynamo(StorageError::IndexesInUse(measured.clone()), "t"); - match err { - DynamoDbError::ResourceInUseException(msg) => assert_eq!(msg, measured), - other => panic!("expected ResourceInUseException, got {other:?}"), - } - } - - /// The measured whole string, byte for byte, from probe P2 on 2026-08-19. - #[test] - fn the_allocation_phase_refusal_is_the_measured_whole_string() { - assert_eq!( - extenddb_core::types::vector_index_delete_in_allocation_phase( - "eddbprobe-backfill", - "vidx2" - ), - "Attempt to change a resource which is still in use: Index creation is in resource \ - allocation phase. Retry deletion during backfilling phase or when the index is \ - active. Table: eddbprobe-backfill Index: vidx2" - ); - } - - /// A genuine fault still reports a 500 and is still logged as one. - #[test] - fn an_internal_failure_is_still_an_internal_server_error() { - let err = - update_table_err_to_dynamo(StorageError::Internal("disk on fire".to_owned()), "t"); - assert!(matches!(err, DynamoDbError::InternalServerError(_))); - } -} +#[path = "update_table_tests.rs"] +mod tests; diff --git a/crates/engine/src/update_table_tests.rs b/crates/engine/src/update_table_tests.rs new file mode 100644 index 000000000..cd8828388 --- /dev/null +++ b/crates/engine/src/update_table_tests.rs @@ -0,0 +1,247 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Tests for `update_table`. + +use super::*; +use extenddb_storage::error::StorageError; + +/// Enabling a stream without a view type is refused with the service's +/// sentence; a view type, or a disable, passes this check. +#[test] +fn enabling_a_stream_without_a_view_type_is_refused() { + let input = |spec: Value| -> UpdateTableInput { + serde_json::from_value(serde_json::json!({ + "TableName": "t", + "StreamSpecification": spec, + })) + .unwrap() + }; + match validate_stream_specification(&input(serde_json::json!({ "StreamEnabled": true }))) + .unwrap_err() + { + DynamoDbError::ValidationException(m) => assert_eq!( + m, + "One or more parameter values were invalid: If stream is being enabled then UpdateViewType is required" + ), + other => panic!("{other:?}"), + } + validate_stream_specification(&input(serde_json::json!({ + "StreamEnabled": true, "StreamViewType": "NEW_IMAGE" + }))) + .unwrap(); + validate_stream_specification(&input(serde_json::json!({ "StreamEnabled": false }))).unwrap(); +} + +/// A created index whose key attribute is not in AttributeDefinitions is +/// refused with the sentence the service picks by the member's shape +/// (measured 2026-10-25 and 2026-10-26): absent names the index; fewer +/// definitions than keys, including none, has the key schema sentence; +/// otherwise the sentence lists every key and every definition. +#[test] +fn a_created_gsi_on_an_undefined_key_attribute_is_refused_by_member_shape() { + let input = |defs: Option| -> UpdateTableInput { + let mut v = serde_json::json!({ + "TableName": "t", + "GlobalSecondaryIndexUpdates": [{ "Create": { + "IndexName": "gix", + "KeySchema": [ + { "AttributeName": "g", "KeyType": "HASH" }, + { "AttributeName": "r", "KeyType": "RANGE" }, + ], + "Projection": { "ProjectionType": "ALL" }, + }}], + }); + if let Some(d) = defs { + v["AttributeDefinitions"] = d; + } + serde_json::from_value(v).unwrap() + }; + let message = |defs: Option| match validate_gsi_updates(&input(defs)).unwrap_err() { + DynamoDbError::ValidationException(m) => m, + other => panic!("{other:?}"), + }; + assert_eq!( + message(None), + "One or more parameter values were invalid: AttributeDefinitions is not specified for index: gix" + ); + assert_eq!( + message(Some(serde_json::json!([]))), + "Invalid KeySchema: Some index key attribute have no definition" + ); + assert_eq!( + message(Some( + serde_json::json!([{ "AttributeName": "g", "AttributeType": "S" }]) + )), + "Invalid KeySchema: Some index key attribute have no definition" + ); + assert_eq!( + message(Some(serde_json::json!([ + { "AttributeName": "pk", "AttributeType": "S" }, + { "AttributeName": "g", "AttributeType": "S" }, + ]))), + "One or more parameter values were invalid: Some index key attributes are not defined in AttributeDefinitions. Keys: [g, r], AttributeDefinitions: [pk, g]" + ); + assert_eq!( + message(Some(serde_json::json!([ + { "AttributeName": "g", "AttributeType": "S" }, + { "AttributeName": "g", "AttributeType": "S" }, + ]))), + "Attribute Name is duplicated: g" + ); + validate_gsi_updates(&input(Some(serde_json::json!([ + { "AttributeName": "g", "AttributeType": "S" }, + { "AttributeName": "r", "AttributeType": "S" }, + ])))) + .unwrap(); +} + +#[test] +fn a_created_gsi_throughput_below_one_is_refused() { + let input = |pt: Value| -> UpdateTableInput { + serde_json::from_value(serde_json::json!({ + "TableName": "t", + "AttributeDefinitions": [{ "AttributeName": "g", "AttributeType": "S" }], + "GlobalSecondaryIndexUpdates": [{ "Create": { + "IndexName": "gix", + "KeySchema": [{ "AttributeName": "g", "KeyType": "HASH" }], + "Projection": { "ProjectionType": "ALL" }, + "ProvisionedThroughput": pt, + }}], + })) + .unwrap() + }; + for (read, write) in [(0, 0), (0, 1), (1, 0), (-1, 5)] { + let pt = serde_json::json!({ "ReadCapacityUnits": read, "WriteCapacityUnits": write }); + match validate_gsi_updates(&input(pt)).unwrap_err() { + DynamoDbError::ValidationException(m) => { + assert!(m.ends_with("for index: gix"), "{m}"); + } + other => panic!("({read}, {write}): {other:?}"), + } + } + let ok = serde_json::json!({ "ReadCapacityUnits": 1, "WriteCapacityUnits": 1 }); + validate_gsi_updates(&input(ok)).unwrap(); +} + +#[test] +fn projected_total_counts_kept_indexes_and_created_ones() { + use extenddb_core::types::{ + GlobalSecondaryIndexUpdate, GsiDescription, LsiDescription, Projection, ProjectionType, + TableDescription, + }; + let include = |n: usize| Projection { + projection_type: ProjectionType::Include, + non_key_attributes: Some((0..n).map(|i| format!("a{i}")).collect()), + }; + let existing = TableDescription { + global_secondary_indexes: Some(vec![ + GsiDescription { + index_name: "kept".to_owned(), + key_schema: Vec::new(), + projection: include(30), + index_status: "ACTIVE".to_owned(), + provisioned_throughput: None, + index_size_bytes: 0, + item_count: 0, + index_arn: String::new(), + }, + GsiDescription { + index_name: "dropped".to_owned(), + key_schema: Vec::new(), + projection: include(50), + index_status: "ACTIVE".to_owned(), + provisioned_throughput: None, + index_size_bytes: 0, + item_count: 0, + index_arn: String::new(), + }, + ]), + local_secondary_indexes: Some(vec![LsiDescription { + index_name: "lsi".to_owned(), + key_schema: Vec::new(), + projection: include(20), + index_size_bytes: 0, + item_count: 0, + index_arn: String::new(), + }]), + ..Default::default() + }; + let create = |n: usize| -> GlobalSecondaryIndexUpdate { + serde_json::from_value(serde_json::json!({ "Create": { + "IndexName": "new", "KeySchema": [], + "Projection": serde_json::to_value(include(n)).unwrap(), + }})) + .unwrap() + }; + let delete: GlobalSecondaryIndexUpdate = + serde_json::from_value(serde_json::json!({ "Delete": { "IndexName": "dropped" } })) + .unwrap(); + + assert_eq!(projected_after_gsi_updates(&existing, &[create(1)]), 101); + assert_eq!( + projected_after_gsi_updates(&existing, &[create(1), delete]), + 51 + ); + assert!( + extenddb_core::validation::validate_projected_attribute_total(projected_after_gsi_updates( + &existing, + &[create(1)] + )) + .is_err() + ); +} + +/// A backend that cannot serve the request reports a 400, not a 500. +/// +/// This arm was missing while CreateTable had it, so the same refusal +/// answered differently depending on which operation carried it. It matters +/// for a backend whose vector capability is decided at runtime rather than at +/// compile time: its refusal arrives through this path. +#[test] +fn an_unsupported_feature_is_a_validation_exception() { + let err = update_table_err_to_dynamo( + StorageError::Unsupported("Vector indexes are not supported here".to_owned()), + "t", + ); + match err { + DynamoDbError::ValidationException(msg) => { + assert_eq!(msg, "Vector indexes are not supported here"); + } + other => panic!("expected ValidationException, got {other:?}"), + } +} + +/// The backend's whole message survives, unwrapped and unprefixed, because +/// the service's own wording for this case names both the table and the index +/// and no layer above the backend knows either. +#[test] +fn a_resource_in_use_refusal_keeps_the_backend_message() { + let measured = extenddb_core::types::vector_index_delete_in_allocation_phase("t", "vidx"); + let err = update_table_err_to_dynamo(StorageError::IndexesInUse(measured.clone()), "t"); + match err { + DynamoDbError::ResourceInUseException(msg) => assert_eq!(msg, measured), + other => panic!("expected ResourceInUseException, got {other:?}"), + } +} + +/// The measured whole string, byte for byte, from probe P2 on 2026-08-19. +#[test] +fn the_allocation_phase_refusal_is_the_measured_whole_string() { + assert_eq!( + extenddb_core::types::vector_index_delete_in_allocation_phase( + "eddbprobe-backfill", + "vidx2" + ), + "Attempt to change a resource which is still in use: Index creation is in resource \ + allocation phase. Retry deletion during backfilling phase or when the index is \ + active. Table: eddbprobe-backfill Index: vidx2" + ); +} + +/// A genuine fault still reports a 500 and is still logged as one. +#[test] +fn an_internal_failure_is_still_an_internal_server_error() { + let err = update_table_err_to_dynamo(StorageError::Internal("disk on fire".to_owned()), "t"); + assert!(matches!(err, DynamoDbError::InternalServerError(_))); +} diff --git a/crates/storage-cassandra/src/stream_engine.rs b/crates/storage-cassandra/src/stream_engine.rs index e3c6088fa..60601bab7 100755 --- a/crates/storage-cassandra/src/stream_engine.rs +++ b/crates/storage-cassandra/src/stream_engine.rs @@ -345,6 +345,7 @@ impl StreamEngine for CassandraEngine { }; Ok(StreamDescription { + creation_request_date_time: None, stream_arn: stream_arn_str, stream_label, stream_status, diff --git a/crates/storage-mongodb/src/stream_engine.rs b/crates/storage-mongodb/src/stream_engine.rs index e976a667a..3075910b1 100644 --- a/crates/storage-mongodb/src/stream_engine.rs +++ b/crates/storage-mongodb/src/stream_engine.rs @@ -494,6 +494,7 @@ impl StreamEngine for MongoEngine { }; Ok(StreamDescription { + creation_request_date_time: None, stream_arn: stream_arn_val, stream_label, stream_status, diff --git a/crates/storage-postgres/src/stream_engine.rs b/crates/storage-postgres/src/stream_engine.rs index 6eee09658..574b2b332 100755 --- a/crates/storage-postgres/src/stream_engine.rs +++ b/crates/storage-postgres/src/stream_engine.rs @@ -376,6 +376,7 @@ impl StreamEngine for PostgresEngine { }; Ok(StreamDescription { + creation_request_date_time: None, stream_arn, stream_label, stream_status, diff --git a/crates/storage-sqlite/src/stream.rs b/crates/storage-sqlite/src/stream.rs index 62943f678..49dbc0541 100644 --- a/crates/storage-sqlite/src/stream.rs +++ b/crates/storage-sqlite/src/stream.rs @@ -263,6 +263,7 @@ impl StreamEngine for SqliteEngine { }; Ok(StreamDescription { + creation_request_date_time: None, stream_arn, stream_label, stream_status, diff --git a/crates/storage/src/util/arn.rs b/crates/storage/src/util/arn.rs index 083e62c35..459bbee21 100755 --- a/crates/storage/src/util/arn.rs +++ b/crates/storage/src/util/arn.rs @@ -54,6 +54,57 @@ pub fn new_stream_label() -> String { format_stream_label(time::OffsetDateTime::now_utc()) } +/// The instant a [`format_stream_label`] label names, as the epoch-seconds +/// number `DescribeStream` renders in `CreationRequestDateTime` (measured on +/// the service 2026-10-24: present on every stream, enabled or disabled). The +/// label is the stream's creation instant to the millisecond, so no stored +/// column is needed. +/// +/// Two shapes are accepted: the 23-byte `YYYY-MM-DDThh:mm:ss.sss` this module +/// writes, and the 19-byte `YYYY-MM-DDThh:mm:ss` that SQLite, PostgreSQL and +/// MongoDB wrote on v0.1.3 to v0.1.13 and Cassandra up to v0.1.14, which a +/// stream enabled on those versions and never re-enabled still carries. The +/// legacy shape names the instant with a `.000` fraction. `None` for any +/// other shape. +#[must_use] +#[allow(clippy::cast_precision_loss)] // epoch ms fit f64's 53-bit mantissa +pub fn stream_label_creation_date_time(label: &str) -> Option { + let b = label.as_bytes(); + let has_fraction = match b.len() { + 19 => false, + 23 if b[19] == b'.' => true, + _ => return None, + }; + if b[4] != b'-' || b[7] != b'-' || b[10] != b'T' || b[13] != b':' || b[16] != b':' { + return None; + } + let num = |range: std::ops::Range| -> Option { + let s = &label[range]; + if s.bytes().all(|c| c.is_ascii_digit()) { + s.parse().ok() + } else { + None + } + }; + let year = i32::try_from(num(0..4)?).ok()?; + let month = time::Month::try_from(u8::try_from(num(5..7)?).ok()?).ok()?; + let date = time::Date::from_calendar_date(year, month, u8::try_from(num(8..10)?).ok()?).ok()?; + let millis = if has_fraction { + u16::try_from(num(20..23)?).ok()? + } else { + 0 + }; + let clock = time::Time::from_hms_milli( + u8::try_from(num(11..13)?).ok()?, + u8::try_from(num(14..16)?).ok()?, + u8::try_from(num(17..19)?).ok()?, + millis, + ) + .ok()?; + let secs = date.with_time(clock).assume_utc().unix_timestamp(); + Some((secs * 1000 + i64::from(millis)) as f64 / 1000.0) +} + /// Returns an ARN for the specified `DynamoDB` stream. #[must_use] pub fn stream_arn(region: &str, account_id: &str, table_name: &str, stream_label: &str) -> String { @@ -130,4 +181,56 @@ mod stream_label_tests { "2026-01-01T22:04:05.500" ); } + + /// `CreationRequestDateTime` is the label's instant, to the millisecond, + /// and round-trips through the formatter. + #[test] + fn creation_date_time_is_the_label_instant() { + use super::stream_label_creation_date_time; + assert_eq!( + stream_label_creation_date_time("1970-01-01T00:00:00.000"), + Some(0.0) + ); + assert_eq!( + stream_label_creation_date_time("1970-01-01T00:00:01.748"), + Some(1.748) + ); + // 2026-10-24T11:31:45.748Z, the label in the issue. + assert_eq!( + stream_label_creation_date_time("2026-10-24T11:31:45.748"), + Some(1_792_841_505.748) + ); + // The 19-byte legacy label (v0.1.3 to v0.1.13 writers) is the instant + // with a .000 fraction. + assert_eq!( + stream_label_creation_date_time("2026-07-04T10:20:30"), + Some(1_783_160_430.0) + ); + // 2026-01-02T03:04:05.007 UTC, through the formatter. + let label = format_stream_label(at((2026, 1, 2), (3, 4, 5, 7_000), 0)); + assert_eq!( + stream_label_creation_date_time(&label), + Some(1_767_323_045.007) + ); + } + + #[test] + fn creation_date_time_refuses_other_shapes() { + use super::stream_label_creation_date_time; + for bad in [ + "", + "2026-10-24T11:31:4", + "2026-10-24T11:31:4S", + "2026-10-24T11:31:45.7480", + "2026-10-24T11:31:45.748Z", + "2026-10-24 11:31:45.748", + "2026-10-24 11:31:45", + "2026-13-24T11:31:45.748", + "2026-10-24T25:31:45.748", + "2026-1O-24T11:31:45.748", + "2026-10-24T11:31:45:748", + ] { + assert_eq!(stream_label_creation_date_time(bad), None, "{bad:?}"); + } + } } diff --git a/crates/storage/src/util/mod.rs b/crates/storage/src/util/mod.rs index b9c68a0f3..ce3c5a495 100755 --- a/crates/storage/src/util/mod.rs +++ b/crates/storage/src/util/mod.rs @@ -13,7 +13,7 @@ mod key; pub use arn::{ ARN_PARTITION, format_stream_label, index_arn, new_stream_label, parse_stream_arn, stream_arn, - table_arn, + stream_label_creation_date_time, table_arn, }; pub use backup_id::{ BackupArnError, BackupArnParts, MAX_BACKUP_ARN_LEN, backup_arn, backup_creation_date_time, diff --git a/docs/differences-from-dynamodb.md b/docs/differences-from-dynamodb.md index 2d6469b7e..feecf83a1 100755 --- a/docs/differences-from-dynamodb.md +++ b/docs/differences-from-dynamodb.md @@ -48,6 +48,7 @@ adaptation when switching between ExtendDB and the real service. |------|----------|------| | Table creation delay | Returns `CREATING` immediately; transitions to `ACTIVE` typically within seconds. Same behavior for on-demand and provisioned | Configurable via `control_plane_delay_seconds` runtime setting (default: 5s) | | DeletionProtectionEnabled | Enforced | Enforced (accepted and stored, DeleteTable rejects when enabled) | +| Order of the lists in `Some index key attributes are not defined in AttributeDefinitions. Keys: [...], AttributeDefinitions: [...]` (CreateTable, UpdateTable GSI Create) | Every key of the schema and every definition, in an order that follows neither the request nor the alphabet and does not depend on the attribute types (measured 2026-10-26: definitions `[pk, r]` render `[r, pk]`, keys `hk, rk` render `[rk, hk]`) | The same keys and definitions, the keys in key schema order and the definitions in request order. A client that reads the first element of `Keys` can see a different name | ## Time to Live (TTL) diff --git a/tests/python/test_streams.py b/tests/python/test_streams.py index cc3b3d2e5..d6fdcea3a 100755 --- a/tests/python/test_streams.py +++ b/tests/python/test_streams.py @@ -16,6 +16,7 @@ import os import time import uuid +from datetime import datetime, timezone import boto3 import botocore.config @@ -167,6 +168,48 @@ def test_describe_stream_invalid_arn(self, streams_client): "ValidationException", ) + def test_creation_request_date_time_is_the_label_instant( + self, dynamodb_client, streams_client + ): + """CreationRequestDateTime is present on an enabled and on a disabled + stream and equals the instant the StreamLabel names (UTC, to the + millisecond).""" + name, stream_arn = _create_stream_table(dynamodb_client) + try: + desc = streams_client.describe_stream(StreamArn=stream_arn)["StreamDescription"] + assert desc["StreamStatus"] == "ENABLED" + self._assert_creation_matches_label(desc) + + dynamodb_client.update_table( + TableName=name, StreamSpecification={"StreamEnabled": False} + ) + wait_for_active(dynamodb_client, name) + table = dynamodb_client.describe_table(TableName=name)["Table"] + # The service drops StreamSpecification from DescribeTable once the + # stream is disabled; ExtendDB answers {StreamEnabled: false} until + # every backend does the same. + assert "StreamSpecification" not in table or table["StreamSpecification"] == { + "StreamEnabled": False + } + assert table["LatestStreamArn"] == stream_arn + desc = streams_client.describe_stream(StreamArn=stream_arn)["StreamDescription"] + self._assert_creation_matches_label(desc) + finally: + dynamodb_client.delete_table(TableName=name) + wait_for_deleted(dynamodb_client, name) + + @staticmethod + def _assert_creation_matches_label(desc: dict) -> None: + label = desc["StreamLabel"] + expected = ( + datetime.strptime(label, "%Y-%m-%dT%H:%M:%S.%f") + .replace(tzinfo=timezone.utc) + .timestamp() + ) + created = desc["CreationRequestDateTime"] + assert isinstance(created, datetime), created + assert abs(created.timestamp() - expected) < 0.0005, (label, created) + class TestStreamRecords: """GetShardIterator and GetRecords tests. diff --git a/tests/python/test_update_table_sentences.py b/tests/python/test_update_table_sentences.py new file mode 100644 index 000000000..09d35d909 --- /dev/null +++ b/tests/python/test_update_table_sentences.py @@ -0,0 +1,170 @@ +# Copyright 2026 ExtendDB contributors +# SPDX-License-Identifier: Apache-2.0 + +"""UpdateTable refusals whose sentence the service fixes. + +Each row asserts the code and the whole message as DynamoDB returns them +(measured 2026-10-24 and 2026-10-25). +""" + +from __future__ import annotations + +import pytest +from botocore.exceptions import ClientError + +from helpers import wait_for_active + +STREAM_VIEW_TYPE_REQUIRED = ( + "One or more parameter values were invalid: " + "If stream is being enabled then UpdateViewType is required" +) + + +def _validation_message(exc_info) -> str: + err = exc_info.value.response["Error"] + assert err["Code"] == "ValidationException" + assert exc_info.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400 + return err["Message"] + + +class TestStreamSpecification: + """StreamEnabled without StreamViewType.""" + + def test_never_streamed_table(self, table_factory, dynamodb_client): + name = table_factory() + with pytest.raises(ClientError) as exc_info: + dynamodb_client.update_table( + TableName=name, StreamSpecification={"StreamEnabled": True} + ) + assert _validation_message(exc_info) == STREAM_VIEW_TYPE_REQUIRED + + def test_table_with_an_enabled_stream(self, table_factory, dynamodb_client): + """The view type check comes before the already-enabled check.""" + name = table_factory( + StreamSpecification={ + "StreamEnabled": True, + "StreamViewType": "NEW_AND_OLD_IMAGES", + } + ) + with pytest.raises(ClientError) as exc_info: + dynamodb_client.update_table( + TableName=name, StreamSpecification={"StreamEnabled": True} + ) + assert _validation_message(exc_info) == STREAM_VIEW_TYPE_REQUIRED + + +class TestValidationOrder: + """Which refusal a request with two defects gets (measured 2026-10-25).""" + + def test_index_without_definitions_precedes_the_view_type(self, table_factory, dynamodb_client): + name = table_factory() + with pytest.raises(ClientError) as exc_info: + dynamodb_client.update_table( + TableName=name, + StreamSpecification={"StreamEnabled": True}, + GlobalSecondaryIndexUpdates=[ + { + "Create": { + "IndexName": "by-g", + "KeySchema": [{"AttributeName": "g", "KeyType": "HASH"}], + "Projection": {"ProjectionType": "ALL"}, + } + } + ], + ) + assert _validation_message(exc_info) == ( + "One or more parameter values were invalid: " + "AttributeDefinitions is not specified for index: by-g" + ) + + def test_missing_throughput_precedes_the_view_type(self, table_factory, dynamodb_client): + name = table_factory() + with pytest.raises(ClientError) as exc_info: + dynamodb_client.update_table( + TableName=name, + StreamSpecification={"StreamEnabled": True}, + BillingMode="PROVISIONED", + ) + assert _validation_message(exc_info) == ( + "One or more parameter values were invalid: " + "ProvisionedThroughput must be specified when BillingMode is PROVISIONED" + ) + + +class TestIndexKeyAttributeUndefined: + """A secondary index key attribute that AttributeDefinitions does not define.""" + + @staticmethod + def _create_by_g(): + return [ + { + "Create": { + "IndexName": "by-g", + "KeySchema": [{"AttributeName": "g", "KeyType": "HASH"}], + "Projection": {"ProjectionType": "ALL"}, + } + } + ] + + def test_update_table_without_the_member_names_the_index(self, table_factory, dynamodb_client): + name = table_factory() + with pytest.raises(ClientError) as exc_info: + dynamodb_client.update_table( + TableName=name, + GlobalSecondaryIndexUpdates=self._create_by_g(), + ) + assert _validation_message(exc_info) == ( + "One or more parameter values were invalid: " + "AttributeDefinitions is not specified for index: by-g" + ) + # The refusal left the table as it was. + wait_for_active(dynamodb_client, name) + desc = dynamodb_client.describe_table(TableName=name)["Table"] + assert "GlobalSecondaryIndexes" not in desc + + def test_update_table_with_the_member_lacking_the_key_lists_keys_and_definitions( + self, table_factory, dynamodb_client + ): + name = table_factory() + with pytest.raises(ClientError) as exc_info: + dynamodb_client.update_table( + TableName=name, + AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}], + GlobalSecondaryIndexUpdates=self._create_by_g(), + ) + assert _validation_message(exc_info) == ( + "One or more parameter values were invalid: " + "Some index key attributes are not defined in AttributeDefinitions. " + "Keys: [g], AttributeDefinitions: [pk]" + ) + + def test_update_table_with_an_empty_member_has_its_own_sentence(self, table_factory, dynamodb_client): + name = table_factory() + with pytest.raises(ClientError) as exc_info: + dynamodb_client.update_table( + TableName=name, + AttributeDefinitions=[], + GlobalSecondaryIndexUpdates=self._create_by_g(), + ) + assert _validation_message(exc_info) == "Invalid KeySchema: Some index key attribute have no definition" + + def test_create_table_lists_keys_and_definitions(self, dynamodb_client): + with pytest.raises(ClientError) as exc_info: + dynamodb_client.create_table( + TableName="never-created-by-g", + AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}], + KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}], + BillingMode="PAY_PER_REQUEST", + GlobalSecondaryIndexes=[ + { + "IndexName": "by-g", + "KeySchema": [{"AttributeName": "g", "KeyType": "HASH"}], + "Projection": {"ProjectionType": "ALL"}, + } + ], + ) + assert _validation_message(exc_info) == ( + "One or more parameter values were invalid: " + "Some index key attributes are not defined in AttributeDefinitions. " + "Keys: [g], AttributeDefinitions: [pk]" + ) diff --git a/tests/test_key_schema_definitions.py b/tests/test_key_schema_definitions.py new file mode 100644 index 000000000..29da06e51 --- /dev/null +++ b/tests/test_key_schema_definitions.py @@ -0,0 +1,231 @@ +# Copyright 2026 ExtendDB contributors +# SPDX-License-Identifier: Apache-2.0 + +"""Key schemas whose attributes are missing from AttributeDefinitions. + +Amazon DynamoDB, measured on 2026-10-25 and 2026-10-26 at 01:27, 01:29, +02:27 and 02:34 UTC (us-east-1), each request twice, on CreateTable and on +an UpdateTable GSI Create: + +- A table billing fault is reported first. +- A name defined twice answers ``Attribute Name is duplicated: ``, + naming the first name whose second definition comes first, before a + missing key. +- The table's key schema is checked first, then each LSI's, then each GSI's. + A schema with more keys than there are definitions answers ``Invalid + KeySchema: Some index key attribute have no definition``. Otherwise a key + with no definition answers ``Some index key attributes are not defined in + AttributeDefinitions`` listing every key of that schema and every + definition. + +The service orders both lists in a way that follows neither the request nor +the alphabet, and ExtendDB lists them in request order, so these tests compare +the lists as sets. + +On UpdateTable a definition that redefines the table's own key with another +type is accepted beside a valid Create, and the table keeps its type +(measured 2026-10-25). + +Dual-target against Amazon DynamoDB and extenddb. +""" + +from __future__ import annotations + +import re +import time + +import pytest +from botocore.exceptions import ClientError + +from conftest import scoped_table + +INVALID = "Invalid KeySchema: Some index key attribute have no definition" +LISTED = re.compile( + r"^One or more parameter values were invalid: Some index key attributes are not defined in " + r"AttributeDefinitions\. Keys: \[([^\]]*)\], AttributeDefinitions: \[([^\]]*)\]$" +) + + +def defs(spec: str) -> list[dict]: + return [{"AttributeName": p.split(":")[0], "AttributeType": p.split(":")[1]} for p in spec.split()] + + +def ks(hash_key: str, range_key: str | None = None) -> list[dict]: + out = [{"AttributeName": hash_key, "KeyType": "HASH"}] + if range_key: + out.append({"AttributeName": range_key, "KeyType": "RANGE"}) + return out + + +def _message(call) -> str: + with pytest.raises(ClientError) as exc_info: + call() + err = exc_info.value.response["Error"] + assert err["Code"] == "ValidationException", err + return err["Message"] + + +def _listed(message: str) -> tuple[set[str], set[str]]: + m = LISTED.match(message) + assert m, message + return set(m.group(1).split(", ")), set(m.group(2).split(", ")) + + +@pytest.fixture(scope="module") +def table(dynamodb_client): + with scoped_table(dynamodb_client) as name: + yield name + + +def _update(client, table, spec, hash_key, range_key=None): + return lambda: client.update_table( + TableName=table, + AttributeDefinitions=defs(spec), + GlobalSecondaryIndexUpdates=[{"Create": { + "IndexName": "never-created", + "KeySchema": ks(hash_key, range_key), + "Projection": {"ProjectionType": "KEYS_ONLY"}, + }}], + ) + + +def _create(client, spec, table_keys, gsis=(), lsis=()): + kwargs = { + "TableName": "never-created-key-definitions", + "AttributeDefinitions": defs(spec), + "KeySchema": table_keys, + "BillingMode": "PAY_PER_REQUEST", + } + if gsis: + kwargs["GlobalSecondaryIndexes"] = [ + {"IndexName": f"gsi{i}", "KeySchema": k, "Projection": {"ProjectionType": "KEYS_ONLY"}} + for i, k in enumerate(gsis) + ] + if lsis: + kwargs["LocalSecondaryIndexes"] = [ + {"IndexName": f"lsi{i}", "KeySchema": k, "Projection": {"ProjectionType": "KEYS_ONLY"}} + for i, k in enumerate(lsis) + ] + return lambda: client.create_table(**kwargs) + + +# UpdateTable GSI Create: (definitions, index hash, index range, keys listed, definitions listed) +UPDATE_LISTED = [ + ("range_missing", "pk:S g:S", "g", "r", {"g", "r"}, {"pk", "g"}), + ("hash_missing", "pk:S r:S", "g", "r", {"g", "r"}, {"pk", "r"}), + ("both_missing_two_defs", "pk:S z:S", "g", "r", {"g", "r"}, {"pk", "z"}), + ("both_missing_three_defs", "pk:S y:S z:S", "g", "r", {"g", "r"}, {"pk", "y", "z"}), + ("hash_only_missing", "pk:S", "g", None, {"g"}, {"pk"}), + ("other_types", "pk:S n:N b:B", "n", "x", {"n", "x"}, {"pk", "n", "b"}), +] + + +@pytest.mark.parametrize(("case", "spec", "h", "r", "keys", "names"), UPDATE_LISTED, + ids=[c[0] for c in UPDATE_LISTED]) +def test_update_table_lists_every_key_and_definition(dynamodb_client, table, case, spec, h, r, keys, names): + message = _message(_update(dynamodb_client, table, spec, h, r)) + assert _listed(message) == (keys, names) + + +@pytest.mark.parametrize(("spec", "h", "r"), [("pk:S", "g", "r"), ("pk:S", "pk", "r")], + ids=["both_missing_one_def", "range_missing_one_def"]) +def test_update_table_with_fewer_definitions_than_keys(dynamodb_client, table, spec, h, r): + assert _message(_update(dynamodb_client, table, spec, h, r)) == INVALID + + +@pytest.mark.parametrize(("spec", "h", "r", "name"), [ + ("h:S h:S", "h", None, "h"), + ("h:S h:S", "h", "r", "h"), + ("b:S a:S b:S a:S h:S", "h", "r", "b"), + ("h:S h:N", "h", "r", "h"), +], ids=["index_key", "and_a_missing_key", "first_repeated", "two_types"]) +def test_update_table_refuses_a_name_defined_twice(dynamodb_client, table, spec, h, r, name): + assert _message(_update(dynamodb_client, table, spec, h, r)) == f"Attribute Name is duplicated: {name}" + + +# CreateTable: (definitions, table keys, GSIs, LSIs, keys listed or None for the +# key schema sentence, definitions listed) +CREATE = [ + ("gsi_range_missing", "pk:S g:S", ks("pk"), [ks("g", "r")], [], {"g", "r"}, {"pk", "g"}), + ("gsi_both_missing", "pk:S", ks("pk"), [ks("g", "r")], [], None, None), + ("table_sort_missing_one_def", "pk:S", ks("pk", "sk"), [], [], None, None), + ("table_sort_missing", "pk:S x:S", ks("pk", "sk"), [], [], {"pk", "sk"}, {"pk", "x"}), + ("table_pk_missing", "x:S", ks("pk"), [], [], {"pk"}, {"x"}), + ("table_before_gsi", "pk:S g:S", ks("pk", "sk"), [ks("g", "r")], [], {"pk", "sk"}, {"pk", "g"}), + ("table_before_lsi", "pk:S x:S", ks("pk", "sk"), [], [ks("pk", "l")], {"pk", "sk"}, {"pk", "x"}), + ("lsi_missing", "pk:S sk:S", ks("pk", "sk"), [], [ks("pk", "l")], {"pk", "l"}, {"pk", "sk"}), + ("lsi_before_gsi", "pk:S sk:S g:S", ks("pk", "sk"), [ks("g", "r")], [ks("pk", "l")], + {"pk", "l"}, {"pk", "sk", "g"}), + ("first_gsi_first", "pk:S g:S h:S", ks("pk"), [ks("g", "r"), ks("h", "q")], [], + {"g", "r"}, {"pk", "g", "h"}), +] + + +@pytest.mark.parametrize(("case", "spec", "table_keys", "gsis", "lsis", "keys", "names"), CREATE, + ids=[c[0] for c in CREATE]) +def test_create_table_checks_each_key_schema_in_turn(dynamodb_client, case, spec, table_keys, gsis, lsis, + keys, names): + message = _message(_create(dynamodb_client, spec, table_keys, gsis, lsis)) + if keys is None: + assert message == INVALID + else: + assert _listed(message) == (keys, names) + + +BILLING = ("One or more parameter values were invalid: Neither ReadCapacityUnits nor WriteCapacityUnits " + "can be specified when BillingMode is PAY_PER_REQUEST") + + +@pytest.mark.parametrize(("spec", "table_keys"), [ + ("pk:S pk:S", ks("pk")), + ("pk:S x:S", ks("pk", "sk")), + ("pk:S", ks("pk", "sk")), +], ids=["duplicate", "missing_key", "fewer_definitions"]) +def test_create_table_reports_a_billing_fault_first(dynamodb_client, spec, table_keys): + def call(): + dynamodb_client.create_table( + TableName="never-created-key-definitions", AttributeDefinitions=defs(spec), KeySchema=table_keys, + BillingMode="PAY_PER_REQUEST", ProvisionedThroughput={"ReadCapacityUnits": 1, "WriteCapacityUnits": 1}) + assert _message(call) == BILLING + + +@pytest.mark.parametrize(("spec", "table_keys", "gsis", "name"), [ + ("pk:S pk:S", ks("pk"), [], "pk"), + ("pk:S pk:N", ks("pk"), [], "pk"), + ("pk:S pk:S", ks("pk", "sk"), [], "pk"), + ("pk:S g:S g:S", ks("pk"), [ks("g")], "g"), + ("b:S a:S b:S a:S pk:S", ks("pk"), [], "b"), +], ids=["table_key", "two_types", "and_a_missing_key", "index_key", "first_repeated"]) +def test_create_table_refuses_a_name_defined_twice(dynamodb_client, spec, table_keys, gsis, name): + message = _message(_create(dynamodb_client, spec, table_keys, gsis)) + assert message == f"Attribute Name is duplicated: {name}" + + +def _wait_for_index(client, table, index, timeout=120.0): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + desc = client.describe_table(TableName=table)["Table"] + status = {i["IndexName"]: i["IndexStatus"] for i in desc.get("GlobalSecondaryIndexes", [])} + if desc["TableStatus"] == "ACTIVE" and status.get(index) == "ACTIVE": + return + time.sleep(0.5) + raise TimeoutError(f"{index} on {table} not ACTIVE after {timeout} s") + + +def test_update_table_keeps_the_type_of_a_redefined_table_key(dynamodb_client): + with scoped_table(dynamodb_client) as name: + dynamodb_client.update_table( + TableName=name, + AttributeDefinitions=defs("pk:N g:S"), + GlobalSecondaryIndexUpdates=[{"Create": { + "IndexName": "by-g", + "KeySchema": ks("g"), + "Projection": {"ProjectionType": "KEYS_ONLY"}, + }}], + ) + desc = dynamodb_client.describe_table(TableName=name)["Table"] + stored = {d["AttributeName"]: d["AttributeType"] for d in desc["AttributeDefinitions"]} + assert stored == {"pk": "S", "g": "S"} + # The table cannot be deleted while the index is being created. + _wait_for_index(dynamodb_client, name, "by-g") + dynamodb_client.put_item(TableName=name, Item={"pk": {"S": "a"}, "g": {"S": "b"}})