Key schema definition checks for CreateTable and UpdateTable

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<f64>,
     #[serde(rename = "StreamStatus")]
     pub stream_status: StreamStatus,
     #[serde(rename = "StreamViewType")]

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: <name>`,
+//!   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: <name>`.
+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<AttributeDefinition> {
+        names
+            .iter()
+            .map(|n| AttributeDefinition {
+                attribute_name: (*n).to_owned(),
+                attribute_type: ScalarAttributeType::S,
+            })
+            .collect()
+    }
+
+    fn ks(hash: &str, range: Option<&str>) -> Vec<KeySchemaElement> {
+        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());
+    }
+}

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::<Vec<_>>()
-        .join(", ")
-}
-
 /// Validate table-level throughput against the effective billing mode.
 ///
 /// # Errors

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)