diff --git a/crates/core/src/validation/mod.rs b/crates/core/src/validation/mod.rs
index c186071a..9bf4ea69 100755
--- a/crates/core/src/validation/mod.rs
+++ b/crates/core/src/validation/mod.rs
@@ -1007,6 +1007,12 @@ fn ordered_indexes<'b, 'a>(indexes: &'b [IndexKeyRef<'a>]) -> Vec<&'b IndexKeyRe
     ordered
 }
 
+/// Readable flags for the `is_query` parameter of [`validate_select_projection`].
+/// Real DynamoDB prepends `1 validation error detected: ` to some rejections for
+/// Query but not for Scan, so callers pass one of these instead of a bare bool.
+pub const IS_QUERY: bool = true;
+pub const IS_SCAN: bool = false;
+
 /// Validate the `Select` value against `ProjectionExpression` / `AttributesToGet`
 /// presence and `IndexName`. Shared by Query and Scan so both reject the same
 /// invalid combinations with the same messages.
@@ -1026,6 +1032,7 @@ pub fn validate_select_projection(
     has_projection: bool,
     has_attributes_to_get: bool,
     has_index_name: bool,
+    is_query: bool,
 ) -> Result<(), DynamoDbError> {
     if has_projection {
         let incompatible = match select {
@@ -1035,9 +1042,16 @@ pub fn validate_select_projection(
             _ => None,
         };
         if let Some(what) = incompatible {
-            return Err(DynamoDbError::ValidationException(format!(
-                "Cannot specify the ProjectionExpression when choosing to get {what}"
-            )));
+            // Real DynamoDB prepends "1 validation error detected: " to this
+            // rejection for Query, but NOT for Scan.
+            let body =
+                format!("Cannot specify the ProjectionExpression when choosing to get {what}");
+            let msg = if is_query {
+                format!("1 validation error detected: {body}")
+            } else {
+                body
+            };
+            return Err(DynamoDbError::ValidationException(msg));
         }
     }
     if matches!(select, Some(Select::SpecificAttributes))
@@ -1119,6 +1133,29 @@ pub fn validate_conditional_operator_usage(
     }
 }
 
+/// Reject `Select=ALL_ATTRIBUTES` against a global secondary index whose
+/// projection type is not `ALL` (such an index cannot serve every attribute).
+/// Shared by Query and Scan so both reject with the identical message; a no-op
+/// unless a GSI is targeted with `Select=ALL_ATTRIBUTES`.
+///
+/// # Errors
+///
+/// Returns `DynamoDbError::ValidationException` when the combination is invalid.
+pub fn validate_all_attributes_index_support(
+    select: Option<Select>,
+    is_gsi: bool,
+    projection_is_all: bool,
+    index_name: &str,
+) -> Result<(), DynamoDbError> {
+    if matches!(select, Some(Select::AllAttributes)) && is_gsi && !projection_is_all {
+        return Err(DynamoDbError::ValidationException(format!(
+            "One or more parameter values were invalid: Select type ALL_ATTRIBUTES is not \
+             supported for global secondary index {index_name} because its projection type is not ALL"
+        )));
+    }
+    Ok(())
+}
+
 /// Validate partition key and sort key sizes against limits.
 ///
 /// # Errors
@@ -1132,31 +1169,86 @@ pub fn validate_key_sizes(
     for ks in key_schema {
         if let Some(value) = item.get(&ks.attribute_name) {
             validate_no_empty_key_value(&ks.attribute_name, value)?;
-            let size = key_value_byte_size(value);
-            let max_size = match ks.key_type {
-                KeyType::Hash => limits.max_partition_key_size_bytes,
-                KeyType::Range => limits.max_sort_key_size_bytes,
-            };
-            if size > max_size {
-                // Hash and range use different wording, matching Amazon
-                // DynamoDB (the hash variant has no space before the size).
-                let msg = match ks.key_type {
-                    KeyType::Hash => format!(
-                        "One or more parameter values were invalid: \
-                         Size of hashkey has exceeded the maximum size limit of{max_size} bytes"
-                    ),
-                    KeyType::Range => format!(
-                        "One or more parameter values were invalid: \
-                         Aggregated size of all range keys has exceeded the size limit of {max_size} bytes"
-                    ),
-                };
-                return Err(DynamoDbError::ValidationException(msg));
-            }
+            check_key_size(ks, value, limits)?;
+        }
+    }
+    Ok(())
+}
+
+/// Validate only the byte-size limit of primary-key values (no empty-value
+/// check).
+///
+/// Used by the transaction path, which surfaces an oversized key as a per-item
+/// `TransactionCanceledException` / `ValidationError` cancellation reason, while
+/// an empty key value remains a top-level `ValidationException` — matching real
+/// `DynamoDB`.
+///
+/// # Errors
+///
+/// Returns `DynamoDbError::ValidationException` if a key value exceeds its size limit.
+pub fn validate_key_size_limits(
+    item: &Item,
+    key_schema: &[KeySchemaElement],
+    limits: &LimitsConfig,
+) -> Result<(), DynamoDbError> {
+    for ks in key_schema {
+        if let Some(value) = item.get(&ks.attribute_name) {
+            check_key_size(ks, value, limits)?;
         }
     }
     Ok(())
 }
 
+/// Validate only that primary-key values are non-empty (no size check).
+///
+/// The transaction path uses this to keep the empty-key rejection as a
+/// top-level `ValidationException` (real `DynamoDB` behavior) while the size
+/// limit is enforced separately as a per-item cancellation reason.
+///
+/// # Errors
+///
+/// Returns `DynamoDbError::ValidationException` if a key value is empty.
+pub fn validate_key_not_empty(
+    item: &Item,
+    key_schema: &[KeySchemaElement],
+) -> Result<(), DynamoDbError> {
+    for ks in key_schema {
+        if let Some(value) = item.get(&ks.attribute_name) {
+            validate_no_empty_key_value(&ks.attribute_name, value)?;
+        }
+    }
+    Ok(())
+}
+
+/// Check a single primary-key value against its size limit. Hash and range use
+/// different wording, matching Amazon `DynamoDB` (the hash variant has no space
+/// before the size).
+fn check_key_size(
+    ks: &KeySchemaElement,
+    value: &AttributeValue,
+    limits: &LimitsConfig,
+) -> Result<(), DynamoDbError> {
+    let size = key_value_byte_size(value);
+    let max_size = match ks.key_type {
+        KeyType::Hash => limits.max_partition_key_size_bytes,
+        KeyType::Range => limits.max_sort_key_size_bytes,
+    };
+    if size > max_size {
+        let msg = match ks.key_type {
+            KeyType::Hash => format!(
+                "One or more parameter values were invalid: \
+                 Size of hashkey has exceeded the maximum size limit of{max_size} bytes"
+            ),
+            KeyType::Range => format!(
+                "One or more parameter values were invalid: \
+                 Aggregated size of all range keys has exceeded the size limit of {max_size} bytes"
+            ),
+        };
+        return Err(DynamoDbError::ValidationException(msg));
+    }
+    Ok(())
+}
+
 /// Get the byte size of a key attribute value.
 ///
 /// For `N` keys this uses the digit-string length. Amazon DynamoDB caps a
@@ -1629,6 +1721,70 @@ mod tests {
         assert!(validate_key_sizes(&item, &[make_ks("pk", KeyType::Hash)], &limits).is_ok());
     }
 
+    #[test]
+    fn validate_key_size_limits_rejects_oversized_but_ignores_empty() {
+        // Size-only helper: oversized hash key rejected with the exact message,
+        // but an empty key value is NOT rejected here (that stays a separate,
+        // top-level check for the transaction path).
+        let limits = LimitsConfig::default();
+        let mut big = Item::new();
+        big.insert(
+            "pk".to_owned(),
+            AttributeValue::S("a".repeat(limits.max_partition_key_size_bytes + 1)),
+        );
+        let err =
+            validate_key_size_limits(&big, &[make_ks("pk", KeyType::Hash)], &limits).unwrap_err();
+        assert_eq!(
+            err.to_string(),
+            "One or more parameter values were invalid: \
+             Size of hashkey has exceeded the maximum size limit of2048 bytes"
+        );
+
+        let mut empty = Item::new();
+        empty.insert("pk".to_owned(), AttributeValue::S(String::new()));
+        assert!(
+            validate_key_size_limits(&empty, &[make_ks("pk", KeyType::Hash)], &limits).is_ok(),
+            "size-only check must ignore empty key values"
+        );
+    }
+
+    #[test]
+    fn validate_key_size_limits_range_message_matches_amazon_dynamodb() {
+        let limits = LimitsConfig::default();
+        let mut item = Item::new();
+        item.insert(
+            "sk".to_owned(),
+            AttributeValue::S("b".repeat(limits.max_sort_key_size_bytes + 1)),
+        );
+        let err =
+            validate_key_size_limits(&item, &[make_ks("sk", KeyType::Range)], &limits).unwrap_err();
+        assert_eq!(
+            err.to_string(),
+            "One or more parameter values were invalid: \
+             Aggregated size of all range keys has exceeded the size limit of 1024 bytes"
+        );
+    }
+
+    #[test]
+    fn validate_key_not_empty_rejects_empty_but_ignores_oversized() {
+        // Empty-only helper: rejects an empty key value, but a merely-oversized
+        // (non-empty) key passes (size is enforced separately).
+        let limits = LimitsConfig::default();
+        let mut empty = Item::new();
+        empty.insert("pk".to_owned(), AttributeValue::S(String::new()));
+        assert!(validate_key_not_empty(&empty, &[make_ks("pk", KeyType::Hash)]).is_err());
+
+        let mut big = Item::new();
+        big.insert(
+            "pk".to_owned(),
+            AttributeValue::S("a".repeat(limits.max_partition_key_size_bytes + 1)),
+        );
+        assert!(
+            validate_key_not_empty(&big, &[make_ks("pk", KeyType::Hash)]).is_ok(),
+            "empty-only check must ignore oversized (non-empty) key values"
+        );
+    }
+
     #[test]
     fn validate_key_sizes_hash_message_matches_amazon_dynamodb() {
         let limits = LimitsConfig::default();
@@ -2031,11 +2187,19 @@ mod tests {
             (Select::Count, "only the Count"),
         ];
         for (select, what) in cases {
-            let err = validate_select_projection(Some(select), true, false, true).unwrap_err();
+            let body =
+                format!("Cannot specify the ProjectionExpression when choosing to get {what}");
+            // Query prepends the "1 validation error detected: " prefix.
+            let q =
+                validate_select_projection(Some(select), true, false, true, IS_QUERY).unwrap_err();
             assert_eq!(
-                err.to_string(),
-                format!("Cannot specify the ProjectionExpression when choosing to get {what}")
+                q.to_string(),
+                format!("1 validation error detected: {body}")
             );
+            // Scan does NOT prepend the prefix (matches real DynamoDB).
+            let s =
+                validate_select_projection(Some(select), true, false, true, IS_SCAN).unwrap_err();
+            assert_eq!(s.to_string(), body);
         }
     }
 
@@ -2043,33 +2207,62 @@ mod tests {
     fn select_specific_attributes_requires_projection() {
         // No projection and no AttributesToGet -> rejected.
         assert!(
-            validate_select_projection(Some(Select::SpecificAttributes), false, false, false)
-                .is_err()
+            validate_select_projection(
+                Some(Select::SpecificAttributes),
+                false,
+                false,
+                false,
+                IS_QUERY
+            )
+            .is_err()
         );
         // A projection satisfies it.
         assert!(
-            validate_select_projection(Some(Select::SpecificAttributes), true, false, false)
-                .is_ok()
+            validate_select_projection(
+                Some(Select::SpecificAttributes),
+                true,
+                false,
+                false,
+                IS_QUERY
+            )
+            .is_ok()
         );
         // Legacy AttributesToGet satisfies it.
         assert!(
-            validate_select_projection(Some(Select::SpecificAttributes), false, true, false)
-                .is_ok()
+            validate_select_projection(
+                Some(Select::SpecificAttributes),
+                false,
+                true,
+                false,
+                IS_QUERY
+            )
+            .is_ok()
         );
     }
 
     #[test]
     fn select_all_projected_requires_index() {
-        let err =
-            validate_select_projection(Some(Select::AllProjectedAttributes), false, false, false)
-                .unwrap_err();
+        let err = validate_select_projection(
+            Some(Select::AllProjectedAttributes),
+            false,
+            false,
+            false,
+            IS_QUERY,
+        )
+        .unwrap_err();
         assert_eq!(
             err.to_string(),
             "ALL_PROJECTED_ATTRIBUTES can be used only when Querying using an IndexName"
         );
         assert!(
-            validate_select_projection(Some(Select::AllProjectedAttributes), false, false, true)
-                .is_ok()
+            validate_select_projection(
+                Some(Select::AllProjectedAttributes),
+                false,
+                false,
+                true,
+                IS_QUERY
+            )
+            .is_ok()
         );
     }
 
@@ -2144,9 +2337,14 @@ mod tests {
     fn select_projection_rule_precedes_index_rule() {
         // ALL_PROJECTED_ATTRIBUTES + ProjectionExpression + no IndexName: the
         // ProjectionExpression rule is reported, not the IndexName one.
-        let err =
-            validate_select_projection(Some(Select::AllProjectedAttributes), true, false, false)
-                .unwrap_err();
+        let err = validate_select_projection(
+            Some(Select::AllProjectedAttributes),
+            true,
+            false,
+            false,
+            IS_QUERY,
+        )
+        .unwrap_err();
         assert!(
             err.to_string().contains("ALL_PROJECTED_ATTRIBUTES")
                 && err
diff --git a/crates/engine/src/delete_item.rs b/crates/engine/src/delete_item.rs
index 36a5ad0e..bfadb562 100755
--- a/crates/engine/src/delete_item.rs
+++ b/crates/engine/src/delete_item.rs
@@ -29,15 +29,13 @@ pub async fn handle_delete_item(
 ) -> Result<DispatchResult, DynamoDbError> {
     crate::validate_enum_fields(
         &body,
-        &[
-            ("ReturnValues", "returnValues", &["NONE", "ALL_OLD"]),
-            (
-                "ReturnConsumedCapacity",
-                "returnConsumedCapacity",
-                &["INDEXES", "TOTAL", "NONE"],
-            ),
-        ],
+        &[(
+            "ReturnConsumedCapacity",
+            "returnConsumedCapacity",
+            &["INDEXES", "TOTAL", "NONE"],
+        )],
     )?;
+    crate::validate_put_delete_return_values(&body)?;
     let input: DeleteItemInput = serde_json::from_value(body).map_err(crate::deserialize_error)?;
 
     extenddb_core::validation::validate_table_name(&input.table_name, &ctx.limits)?;
diff --git a/crates/engine/src/lib.rs b/crates/engine/src/lib.rs
index f85e4785..02049abe 100755
--- a/crates/engine/src/lib.rs
+++ b/crates/engine/src/lib.rs
@@ -202,6 +202,37 @@ pub(crate) fn validate_enum_fields(
     );
     Err(DynamoDbError::ValidationException(msg))
 }
+
+/// Validate the `ReturnValues` field for `PutItem` / `DeleteItem`, which accept
+/// only `NONE` or `ALL_OLD`. Matches real DynamoDB's two distinct messages:
+///
+/// - A value that IS a valid `ReturnValues` enum member but is not allowed for
+///   these operations (e.g. `UPDATED_OLD`) → `ReturnValues can only be ALL_OLD
+///   or NONE`.
+/// - A value that is not a `ReturnValues` enum member at all (e.g. `GARBAGE`) →
+///   the generic constraint error listing the full enum set.
+pub(crate) fn validate_put_delete_return_values(
+    body: &serde_json::Value,
+) -> Result<(), DynamoDbError> {
+    // Full ReturnValues enum set, in the order real DynamoDB reports it.
+    const ALL: &[&str] = &["ALL_NEW", "UPDATED_OLD", "ALL_OLD", "NONE", "UPDATED_NEW"];
+    if let Some(rv) = body.get("ReturnValues").and_then(serde_json::Value::as_str) {
+        if rv == "NONE" || rv == "ALL_OLD" {
+            return Ok(());
+        }
+        if ALL.contains(&rv) {
+            return Err(DynamoDbError::ValidationException(
+                "ReturnValues can only be ALL_OLD or NONE".to_owned(),
+            ));
+        }
+        return Err(DynamoDbError::ValidationException(format!(
+            "1 validation error detected: Value '{rv}' at 'returnValues' failed to satisfy \
+             constraint: Member must satisfy enum value set: [{}]",
+            ALL.join(", ")
+        )));
+    }
+    Ok(())
+}
 ///
 /// Populated by engine handlers so the server layer can record capacity,
 /// returned item counts, and returned byte counts without parsing the JSON
diff --git a/crates/engine/src/put_item.rs b/crates/engine/src/put_item.rs
index cf964fcf..d94a9b07 100755
--- a/crates/engine/src/put_item.rs
+++ b/crates/engine/src/put_item.rs
@@ -51,7 +51,6 @@ pub async fn handle_put_item(
     crate::validate_enum_fields(
         &body,
         &[
-            ("ReturnValues", "returnValues", &["NONE", "ALL_OLD"]),
             (
                 "ReturnConsumedCapacity",
                 "returnConsumedCapacity",
@@ -64,6 +63,7 @@ pub async fn handle_put_item(
             ),
         ],
     )?;
+    crate::validate_put_delete_return_values(&body)?;
 
     let input: PutItemInput = serde_json::from_value(body).map_err(|e| {
         let msg = e.to_string();
diff --git a/crates/engine/src/query.rs b/crates/engine/src/query.rs
index 42a7e36f..d5b33fff 100755
--- a/crates/engine/src/query.rs
+++ b/crates/engine/src/query.rs
@@ -11,7 +11,8 @@ use extenddb_core::error::DynamoDbError;
 use extenddb_core::expression::PathElement;
 use extenddb_core::expression::{ExpressionKind, ExpressionMaps, Projection};
 use extenddb_core::types::{
-    IndexType, KeyType, QueryInput, QueryOutput, Select, TableKeyInfo, extract_key, item_size_bytes,
+    IndexType, KeyType, ProjectionType, QueryInput, QueryOutput, Select, TableKeyInfo, extract_key,
+    item_size_bytes,
 };
 
 use crate::OperationContext;
@@ -69,6 +70,16 @@ pub async fn handle_query(
         ));
     }
 
+    // Select=ALL_ATTRIBUTES requires an ALL-projection GSI (shared with Scan).
+    if let Some(ref idx) = index_info {
+        extenddb_core::validation::validate_all_attributes_index_support(
+            input.select,
+            idx.index_type == IndexType::Gsi,
+            idx.projection.projection_type == ProjectionType::All,
+            &idx.index_name,
+        )?;
+    }
+
     // Validate Limit >= 1 (REQ-QUERY-001)
     if let Some(limit) = input.limit
         && limit < 1
@@ -350,6 +361,7 @@ pub async fn handle_query(
             .as_ref()
             .is_some_and(|a| !a.is_empty()),
         input.index_name.is_some(),
+        extenddb_core::validation::IS_QUERY,
     )?;
 
     // When Select=ALL_PROJECTED_ATTRIBUTES, capture the index info for post-read filtering.
diff --git a/crates/engine/src/scan.rs b/crates/engine/src/scan.rs
index 0bad171a..4f5a513d 100755
--- a/crates/engine/src/scan.rs
+++ b/crates/engine/src/scan.rs
@@ -10,7 +10,8 @@ use serde_json::Value;
 use extenddb_core::error::DynamoDbError;
 use extenddb_core::expression::{ExpressionKind, ExpressionMaps, Projection};
 use extenddb_core::types::{
-    IndexType, ScanInput, ScanOutput, Select, TableKeyInfo, extract_key, item_size_bytes,
+    IndexType, ProjectionType, ScanInput, ScanOutput, Select, TableKeyInfo, extract_key,
+    item_size_bytes,
 };
 
 use crate::OperationContext;
@@ -169,6 +170,7 @@ pub async fn handle_scan(
             .as_ref()
             .is_some_and(|a| !a.is_empty()),
         input.index_name.is_some(),
+        extenddb_core::validation::IS_SCAN,
     )?;
 
     // Validate unused expression attributes
@@ -225,6 +227,16 @@ pub async fn handle_scan(
         ));
     }
 
+    // Select=ALL_ATTRIBUTES requires an ALL-projection GSI (shared with Query).
+    if let Some(ref idx) = index_info {
+        extenddb_core::validation::validate_all_attributes_index_support(
+            input.select,
+            idx.index_type == IndexType::Gsi,
+            idx.projection.projection_type == ProjectionType::All,
+            &idx.index_name,
+        )?;
+    }
+
     // Validate Segment/TotalSegments — DynamoDB returns different messages per direction
     match (input.segment, input.total_segments) {
         (Some(_), None) => {
diff --git a/crates/engine/src/transact_get_items.rs b/crates/engine/src/transact_get_items.rs
index ebc37da0..76121317 100755
--- a/crates/engine/src/transact_get_items.rs
+++ b/crates/engine/src/transact_get_items.rs
@@ -10,9 +10,12 @@ use serde_json::Value;
 use extenddb_core::error::DynamoDbError;
 use extenddb_core::expression::Projection;
 use extenddb_core::types::{
-    ItemResponse, TransactGetItemsInput, TransactGetItemsOutput, item_size_bytes,
+    CancellationReason, ItemResponse, TransactGetItemsInput, TransactGetItemsOutput,
+    item_size_bytes,
 };
+use extenddb_core::validation;
 use extenddb_storage::TransactGetOp;
+use extenddb_storage::error::StorageError;
 
 use crate::OperationContext;
 use crate::capacity_helpers;
@@ -117,6 +120,29 @@ pub async fn handle_transact_get_items(
         }
     }
 
+    // Reject oversized primary keys as a per-item cancellation, matching real
+    // DynamoDB (TransactionCanceledException with a ValidationError reason for
+    // the offending item). Key type/emptiness is validated in the storage
+    // layer; this enforces the size limit the storage layer does not check.
+    {
+        let mut reasons: Vec<CancellationReason> = Vec::with_capacity(input.transact_items.len());
+        let mut any_oversized = false;
+        for (tgi, ki) in input.transact_items.iter().zip(key_infos.iter()) {
+            match validation::validate_key_size_limits(&tgi.get.key, &ki.key_schema, &ctx.limits) {
+                Ok(()) => reasons.push(CancellationReason::none()),
+                Err(e) => {
+                    any_oversized = true;
+                    reasons.push(CancellationReason::validation_error(e.to_string()));
+                }
+            }
+        }
+        if any_oversized {
+            return Err(storage_err_to_dynamo(StorageError::TransactionCanceled(
+                reasons,
+            )));
+        }
+    }
+
     // Build storage operations
     let ops: Vec<TransactGetOp<'_>> = input
         .transact_items
diff --git a/crates/engine/src/transact_write_helpers.rs b/crates/engine/src/transact_write_helpers.rs
index e28ceb69..8295ef6d 100755
--- a/crates/engine/src/transact_write_helpers.rs
+++ b/crates/engine/src/transact_write_helpers.rs
@@ -205,6 +205,35 @@ impl PreparedOp {
         };
         capacity_helpers::item_metrics(ricm, &key_info.key_schema, item_or_key, key_info.has_lsi)
     }
+
+    /// Return a `ValidationError` cancellation reason if this op's primary key
+    /// exceeds the size limit (hash > 2048 / range > 1024 bytes by default).
+    ///
+    /// Real `DynamoDB` cancels the whole transaction with a per-item
+    /// `ValidationError` reason for an oversized key in ANY sub-op (Put,
+    /// Delete, Update, ConditionCheck). Empty-key values are handled separately
+    /// as a top-level `ValidationException`, so this checks size only.
+    pub(crate) fn oversized_key_reason(
+        &self,
+        limits: &extenddb_core::limits::LimitsConfig,
+    ) -> Option<extenddb_core::types::CancellationReason> {
+        let (key_info, item_or_key) = match self {
+            Self::Put { key_info, item, .. } => (key_info, item),
+            Self::Delete { key_info, key, .. }
+            | Self::Update { key_info, key, .. }
+            | Self::ConditionCheck { key_info, key, .. } => (key_info, key),
+        };
+        match extenddb_core::validation::validate_key_size_limits(
+            item_or_key,
+            &key_info.key_schema,
+            limits,
+        ) {
+            Ok(()) => None,
+            Err(e) => Some(extenddb_core::types::CancellationReason::validation_error(
+                e.to_string(),
+            )),
+        }
+    }
 }
 
 /// Validate `ClientRequestToken` format.
diff --git a/crates/engine/src/transact_write_items.rs b/crates/engine/src/transact_write_items.rs
index a6f13eca..068b155d 100755
--- a/crates/engine/src/transact_write_items.rs
+++ b/crates/engine/src/transact_write_items.rs
@@ -21,7 +21,7 @@ use extenddb_core::error::DynamoDbError;
 use extenddb_core::types::{TransactWriteItem, TransactWriteItemsInput, TransactWriteItemsOutput};
 use extenddb_core::validation::{
     validate_attribute_name_sizes, validate_attribute_values_nesting_depth,
-    validate_item_nesting_depth, validate_item_size, validate_key_sizes,
+    validate_item_nesting_depth, validate_item_size, validate_key_not_empty,
 };
 
 /// Maximum number of items in a single `TransactWriteItems` request.
@@ -115,6 +115,31 @@ pub async fn handle_transact_write_items(
         ));
     }
 
+    // Reject oversized primary keys as a per-item cancellation, matching real
+    // DynamoDB: an oversized hash/range key in any sub-op returns
+    // TransactionCanceledException with a ValidationError cancellation reason
+    // for the offending item (an EMPTY key value, by contrast, is a top-level
+    // ValidationException and is handled in prepare_write_op).
+    {
+        let mut reasons: Vec<extenddb_core::types::CancellationReason> =
+            Vec::with_capacity(prepared.len());
+        let mut any_oversized = false;
+        for op in &prepared {
+            match op.oversized_key_reason(&ctx.limits) {
+                Some(reason) => {
+                    any_oversized = true;
+                    reasons.push(reason);
+                }
+                None => reasons.push(extenddb_core::types::CancellationReason::none()),
+            }
+        }
+        if any_oversized {
+            return Err(storage_err_to_dynamo(
+                extenddb_storage::error::StorageError::TransactionCanceled(reasons),
+            ));
+        }
+    }
+
     // Build storage operations
     let ops: Vec<extenddb_storage::TransactWriteOp<'_>> =
         prepared.iter().map(|p| p.to_storage_op()).collect();
@@ -218,7 +243,7 @@ async fn prepare_write_op(
         validate_item_nesting_depth(&put.item)?;
         validate_item_size(&put.item, ctx.limits.max_item_size_bytes)?;
         validate_attribute_name_sizes(&put.item, &ctx.limits)?;
-        validate_key_sizes(&put.item, &key_info.key_schema, &ctx.limits)?;
+        validate_key_not_empty(&put.item, &key_info.key_schema)?;
         let maps = build_expression_maps(
             put.expression_attribute_names.as_ref(),
             put.expression_attribute_values.as_ref(),
@@ -262,7 +287,7 @@ async fn prepare_write_op(
         // Empty or oversize key values are up-front input validation in
         // DynamoDB (a top-level ValidationException), unlike a key type
         // mismatch which surfaces as a per-item cancellation reason.
-        validate_key_sizes(&del.key, &key_info.key_schema, &ctx.limits)?;
+        validate_key_not_empty(&del.key, &key_info.key_schema)?;
         let maps = build_expression_maps(
             del.expression_attribute_names.as_ref(),
             del.expression_attribute_values.as_ref(),
@@ -304,7 +329,7 @@ async fn prepare_write_op(
         // Empty or oversize key values are up-front input validation in
         // DynamoDB (a top-level ValidationException), unlike a key type
         // mismatch which surfaces as a per-item cancellation reason.
-        validate_key_sizes(&upd.key, &key_info.key_schema, &ctx.limits)?;
+        validate_key_not_empty(&upd.key, &key_info.key_schema)?;
         let maps = build_expression_maps(
             upd.expression_attribute_names.as_ref(),
             upd.expression_attribute_values.as_ref(),
@@ -365,7 +390,7 @@ async fn prepare_write_op(
         // Empty or oversize key values are up-front input validation in
         // DynamoDB (a top-level ValidationException), unlike a key type
         // mismatch which surfaces as a per-item cancellation reason.
-        validate_key_sizes(&cc.key, &key_info.key_schema, &ctx.limits)?;
+        validate_key_not_empty(&cc.key, &key_info.key_schema)?;
         let maps = build_expression_maps(
             cc.expression_attribute_names.as_ref(),
             cc.expression_attribute_values.as_ref(),
diff --git a/tests/rust/src/main.rs b/tests/rust/src/main.rs
index 5ddb2697..c5cad40e 100755
--- a/tests/rust/src/main.rs
+++ b/tests/rust/src/main.rs
@@ -82,6 +82,8 @@ mod transact_write_items;
 #[cfg(test)]
 mod transact_write_items_more;
 #[cfg(test)]
+mod transaction_key_size_validation;
+#[cfg(test)]
 mod transaction_validation;
 #[cfg(test)]
 mod ttl;
@@ -101,6 +103,8 @@ mod update_item_more;
 mod update_item_number_validation;
 #[cfg(test)]
 mod update_table_billing_validation;
+#[cfg(test)]
+mod wording_parity_validation;
 
 fn main() {
     eprintln!("Run with `cargo test` to execute integration tests.");
diff --git a/tests/rust/src/transaction_key_size_validation.rs b/tests/rust/src/transaction_key_size_validation.rs
new file mode 100644
index 00000000..6c8e81fe
--- /dev/null
+++ b/tests/rust/src/transaction_key_size_validation.rs
@@ -0,0 +1,189 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+
+//! Transaction-API primary-key size validation.
+//!
+//! An oversized hash key (> 2048 bytes) or range key (> 1024 bytes) in any
+//! transaction sub-op must cancel the transaction with a per-item
+//! `ValidationError` cancellation reason (`TransactionCanceledException`),
+//! matching real DynamoDB. This covers TransactGetItems (Get) and each
+//! TransactWriteItems sub-op (Put / Delete / Update / ConditionCheck).
+//!
+//! An EMPTY key value, by contrast, remains a top-level `ValidationException`
+//! (covered in `transact_key_validation`); this file exercises the size path.
+
+use crate::test_base::*;
+use aws_sdk_dynamodb::types::{
+    ConditionCheck, Delete, Get, Put, TransactGetItem, TransactWriteItem, Update,
+};
+use std::collections::HashMap;
+
+fn oversized_hash() -> String {
+    "a".repeat(2049)
+}
+
+fn assert_cancelled_validation(err_code_opt: Option<&str>, msg: &str) {
+    assert_eq!(
+        err_code_opt,
+        Some("TransactionCanceledException"),
+        "expected TransactionCanceledException, got: {msg}"
+    );
+    assert!(
+        msg.contains("ValidationError"),
+        "expected a ValidationError cancellation reason, got: {msg}"
+    );
+}
+
+#[tokio::test]
+async fn transact_write_put_oversized_hash_key_cancels() {
+    let c = client();
+    let t = tables().await;
+    let mut item: HashMap<String, _> = HashMap::new();
+    item.insert(HASH_KEY_S.into(), s(&oversized_hash()));
+    let err = c
+        .transact_write_items()
+        .transact_items(
+            TransactWriteItem::builder()
+                .put(
+                    Put::builder()
+                        .table_name(&t.simple_key_string)
+                        .set_item(Some(item))
+                        .build()
+                        .unwrap(),
+                )
+                .build(),
+        )
+        .send()
+        .await
+        .expect_err("oversized hash key must cancel the transaction");
+    assert_cancelled_validation(err_code(&err), &err_msg(&err));
+}
+
+#[tokio::test]
+async fn transact_write_delete_oversized_hash_key_cancels() {
+    let c = client();
+    let t = tables().await;
+    let mut key: HashMap<String, _> = HashMap::new();
+    key.insert(HASH_KEY_S.into(), s(&oversized_hash()));
+    let err = c
+        .transact_write_items()
+        .transact_items(
+            TransactWriteItem::builder()
+                .delete(
+                    Delete::builder()
+                        .table_name(&t.simple_key_string)
+                        .set_key(Some(key))
+                        .build()
+                        .unwrap(),
+                )
+                .build(),
+        )
+        .send()
+        .await
+        .expect_err("oversized hash key must cancel the transaction");
+    assert_cancelled_validation(err_code(&err), &err_msg(&err));
+}
+
+#[tokio::test]
+async fn transact_write_update_oversized_hash_key_cancels() {
+    let c = client();
+    let t = tables().await;
+    let mut key: HashMap<String, _> = HashMap::new();
+    key.insert(HASH_KEY_S.into(), s(&oversized_hash()));
+    let err = c
+        .transact_write_items()
+        .transact_items(
+            TransactWriteItem::builder()
+                .update(
+                    Update::builder()
+                        .table_name(&t.simple_key_string)
+                        .set_key(Some(key))
+                        .update_expression("SET #d = :v")
+                        .expression_attribute_names("#d", "data")
+                        .expression_attribute_values(":v", s("x"))
+                        .build()
+                        .unwrap(),
+                )
+                .build(),
+        )
+        .send()
+        .await
+        .expect_err("oversized hash key must cancel the transaction");
+    assert_cancelled_validation(err_code(&err), &err_msg(&err));
+}
+
+#[tokio::test]
+async fn transact_write_condition_check_oversized_hash_key_cancels() {
+    let c = client();
+    let t = tables().await;
+    let mut key: HashMap<String, _> = HashMap::new();
+    key.insert(HASH_KEY_S.into(), s(&oversized_hash()));
+    let err = c
+        .transact_write_items()
+        .transact_items(
+            TransactWriteItem::builder()
+                .condition_check(
+                    ConditionCheck::builder()
+                        .table_name(&t.simple_key_string)
+                        .set_key(Some(key))
+                        .condition_expression("attribute_exists(#h)")
+                        .expression_attribute_names("#h", HASH_KEY_S)
+                        .build()
+                        .unwrap(),
+                )
+                .build(),
+        )
+        .send()
+        .await
+        .expect_err("oversized hash key must cancel the transaction");
+    assert_cancelled_validation(err_code(&err), &err_msg(&err));
+}
+
+#[tokio::test]
+async fn transact_get_oversized_hash_key_cancels() {
+    let c = client();
+    let t = tables().await;
+    let mut key: HashMap<String, _> = HashMap::new();
+    key.insert(HASH_KEY_S.into(), s(&oversized_hash()));
+    let err = c
+        .transact_get_items()
+        .transact_items(
+            TransactGetItem::builder()
+                .get(
+                    Get::builder()
+                        .table_name(&t.simple_key_string)
+                        .set_key(Some(key))
+                        .build()
+                        .unwrap(),
+                )
+                .build(),
+        )
+        .send()
+        .await
+        .expect_err("oversized hash key must cancel the transaction");
+    assert_cancelled_validation(err_code(&err), &err_msg(&err));
+}
+
+#[tokio::test]
+async fn transact_get_valid_key_succeeds() {
+    // Sanity: a normal-sized key is not rejected by the size check.
+    let c = client();
+    let t = tables().await;
+    let mut key: HashMap<String, _> = HashMap::new();
+    key.insert(HASH_KEY_S.into(), s(&format!("tks_{}", ts())));
+    c.transact_get_items()
+        .transact_items(
+            TransactGetItem::builder()
+                .get(
+                    Get::builder()
+                        .table_name(&t.simple_key_string)
+                        .set_key(Some(key))
+                        .build()
+                        .unwrap(),
+                )
+                .build(),
+        )
+        .send()
+        .await
+        .expect("valid key must be accepted");
+}
diff --git a/tests/rust/src/wording_parity_validation.rs b/tests/rust/src/wording_parity_validation.rs
new file mode 100644
index 00000000..6615484f
--- /dev/null
+++ b/tests/rust/src/wording_parity_validation.rs
@@ -0,0 +1,196 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+
+//! Validation message/behavior parity for a few request-validation cases that
+//! previously diverged from real DynamoDB (verified against us-east-1):
+//!
+//! - PutItem/DeleteItem `ReturnValues`: a valid-but-disallowed enum value
+//!   (e.g. UPDATED_OLD) → "ReturnValues can only be ALL_OLD or NONE"; a
+//!   non-enum value (e.g. GARBAGE) → the generic constraint error listing the
+//!   full enum set.
+//! - Query/Scan `Select` + `ProjectionExpression`: the rejection carries the
+//!   "1 validation error detected: " prefix.
+//! - Query/Scan `Select=ALL_ATTRIBUTES` on a non-ALL GSI is rejected.
+
+use crate::test_base::*;
+use aws_sdk_dynamodb::types::{
+    AttributeDefinition, BillingMode, GlobalSecondaryIndex, KeySchemaElement, KeyType, Projection,
+    ProjectionType, ReturnValue, ScalarAttributeType, Select,
+};
+use std::collections::HashMap;
+
+#[tokio::test]
+async fn put_item_disallowed_return_values_message() {
+    let c = client();
+    let t = tables().await;
+    let mut item: HashMap<String, _> = HashMap::new();
+    item.insert(HASH_KEY_S.into(), s(&format!("rv_{}", ts())));
+    let err = c
+        .put_item()
+        .table_name(&t.simple_key_string)
+        .set_item(Some(item))
+        .return_values(ReturnValue::UpdatedOld) // valid enum, not allowed for Put
+        .send()
+        .await
+        .expect_err("UPDATED_OLD not allowed for PutItem");
+    assert_eq!(
+        err_code(&err),
+        Some("ValidationException"),
+        "{}",
+        err_msg(&err)
+    );
+    assert_eq!(err_msg(&err), "ReturnValues can only be ALL_OLD or NONE");
+}
+
+#[tokio::test]
+async fn delete_item_disallowed_return_values_message() {
+    let c = client();
+    let t = tables().await;
+    let mut key: HashMap<String, _> = HashMap::new();
+    key.insert(HASH_KEY_S.into(), s(&format!("rv_{}", ts())));
+    let err = c
+        .delete_item()
+        .table_name(&t.simple_key_string)
+        .set_key(Some(key))
+        .return_values(ReturnValue::UpdatedOld)
+        .send()
+        .await
+        .expect_err("UPDATED_OLD not allowed for DeleteItem");
+    assert_eq!(err_msg(&err), "ReturnValues can only be ALL_OLD or NONE");
+}
+
+#[tokio::test]
+async fn query_count_with_projection_has_validation_prefix() {
+    let c = client();
+    let t = tables().await;
+    let err = c
+        .query()
+        .table_name(&t.simple_key_string)
+        .key_condition_expression("#h = :p")
+        .expression_attribute_names("#h", HASH_KEY_S)
+        .expression_attribute_values(":p", s("x"))
+        .select(Select::Count)
+        .projection_expression("#h")
+        .send()
+        .await
+        .expect_err("COUNT with ProjectionExpression is rejected");
+    assert_eq!(
+        err_msg(&err),
+        "1 validation error detected: Cannot specify the ProjectionExpression \
+         when choosing to get only the Count"
+    );
+}
+
+#[tokio::test]
+async fn scan_count_with_projection_no_prefix() {
+    // Scan (unlike Query) does NOT carry the "1 validation error detected: "
+    // prefix on this rejection — matches real DynamoDB.
+    let c = client();
+    let t = tables().await;
+    let err = c
+        .scan()
+        .table_name(&t.simple_key_string)
+        .select(Select::Count)
+        .projection_expression("#h")
+        .expression_attribute_names("#h", HASH_KEY_S)
+        .send()
+        .await
+        .expect_err("COUNT with ProjectionExpression is rejected");
+    assert_eq!(
+        err_msg(&err),
+        "Cannot specify the ProjectionExpression when choosing to get only the Count"
+    );
+}
+
+#[tokio::test]
+async fn scan_all_attributes_on_non_all_gsi_rejected() {
+    let c = client();
+    let name = format!("WordingGsi_{}", ts());
+    c.create_table()
+        .table_name(&name)
+        .attribute_definitions(
+            AttributeDefinition::builder()
+                .attribute_name("pk")
+                .attribute_type(ScalarAttributeType::S)
+                .build()
+                .unwrap(),
+        )
+        .attribute_definitions(
+            AttributeDefinition::builder()
+                .attribute_name("g")
+                .attribute_type(ScalarAttributeType::S)
+                .build()
+                .unwrap(),
+        )
+        .key_schema(
+            KeySchemaElement::builder()
+                .attribute_name("pk")
+                .key_type(KeyType::Hash)
+                .build()
+                .unwrap(),
+        )
+        .billing_mode(BillingMode::PayPerRequest)
+        .global_secondary_indexes(
+            GlobalSecondaryIndex::builder()
+                .index_name("g_index")
+                .key_schema(
+                    KeySchemaElement::builder()
+                        .attribute_name("g")
+                        .key_type(KeyType::Hash)
+                        .build()
+                        .unwrap(),
+                )
+                .projection(
+                    Projection::builder()
+                        .projection_type(ProjectionType::KeysOnly)
+                        .build(),
+                )
+                .build()
+                .unwrap(),
+        )
+        .send()
+        .await
+        .expect("create table with KEYS_ONLY gsi");
+    wait_for_active(c, &name).await;
+
+    let scan_err = c
+        .scan()
+        .table_name(&name)
+        .index_name("g_index")
+        .select(Select::AllAttributes)
+        .send()
+        .await
+        .expect_err("ALL_ATTRIBUTES on a KEYS_ONLY GSI must be rejected");
+    assert_eq!(
+        err_code(&scan_err),
+        Some("ValidationException"),
+        "{}",
+        err_msg(&scan_err)
+    );
+    assert!(
+        err_msg(&scan_err).contains(
+            "Select type ALL_ATTRIBUTES is not supported for global secondary index g_index \
+             because its projection type is not ALL"
+        ),
+        "got: {}",
+        err_msg(&scan_err)
+    );
+
+    let query_err = c
+        .query()
+        .table_name(&name)
+        .index_name("g_index")
+        .key_condition_expression("g = :v")
+        .expression_attribute_values(":v", s("x"))
+        .select(Select::AllAttributes)
+        .send()
+        .await
+        .expect_err("ALL_ATTRIBUTES on a KEYS_ONLY GSI must be rejected");
+    assert!(
+        err_msg(&query_err).contains("Select type ALL_ATTRIBUTES is not supported"),
+        "got: {}",
+        err_msg(&query_err)
+    );
+
+    let _ = c.delete_table().table_name(&name).send().await;
+}
diff --git a/tests/test_transaction_key_size_validation.py b/tests/test_transaction_key_size_validation.py
new file mode 100644
index 00000000..aa3608f0
--- /dev/null
+++ b/tests/test_transaction_key_size_validation.py
@@ -0,0 +1,119 @@
+# Copyright 2026 ExtendDB contributors
+# SPDX-License-Identifier: Apache-2.0
+
+"""Transaction-API primary-key size validation.
+
+An oversized hash key (> 2048 bytes) or range key (> 1024 bytes) in any
+transaction sub-op must cancel the transaction with a per-item ValidationError
+cancellation reason (TransactionCanceledException), matching real DynamoDB.
+Covers TransactGetItems (Get) and each TransactWriteItems sub-op
+(Put / Delete / Update / ConditionCheck).
+
+An EMPTY key value, by contrast, is a top-level ValidationException — verified
+here to lock the size-vs-empty distinction.
+"""
+
+from __future__ import annotations
+
+import pytest
+from botocore.exceptions import ClientError
+
+OVERSIZED_HASH = "a" * 2049
+
+
+@pytest.fixture()
+def table(create_and_cleanup_table):
+    return create_and_cleanup_table()["TableDescription"]["TableName"]
+
+
+def _assert_cancelled_validation(ei):
+    err = ei.value.response
+    assert err["Error"]["Code"] == "TransactionCanceledException", err["Error"]
+    reasons = err.get("CancellationReasons")
+    assert reasons, "expected CancellationReasons in the error response"
+    assert any(r.get("Code") == "ValidationError" for r in reasons), reasons
+
+
+def test_transact_write_put_oversized_hash_key_cancels(dynamodb_client, table):
+    with pytest.raises(ClientError) as ei:
+        dynamodb_client.transact_write_items(
+            TransactItems=[
+                {"Put": {"TableName": table, "Item": {"pk": {"S": OVERSIZED_HASH}}}}
+            ]
+        )
+    _assert_cancelled_validation(ei)
+
+
+def test_transact_write_delete_oversized_hash_key_cancels(dynamodb_client, table):
+    with pytest.raises(ClientError) as ei:
+        dynamodb_client.transact_write_items(
+            TransactItems=[
+                {"Delete": {"TableName": table, "Key": {"pk": {"S": OVERSIZED_HASH}}}}
+            ]
+        )
+    _assert_cancelled_validation(ei)
+
+
+def test_transact_write_update_oversized_hash_key_cancels(dynamodb_client, table):
+    with pytest.raises(ClientError) as ei:
+        dynamodb_client.transact_write_items(
+            TransactItems=[
+                {
+                    "Update": {
+                        "TableName": table,
+                        "Key": {"pk": {"S": OVERSIZED_HASH}},
+                        "UpdateExpression": "SET #d = :v",
+                        "ExpressionAttributeNames": {"#d": "data"},
+                        "ExpressionAttributeValues": {":v": {"S": "x"}},
+                    }
+                }
+            ]
+        )
+    _assert_cancelled_validation(ei)
+
+
+def test_transact_write_condition_check_oversized_hash_key_cancels(
+    dynamodb_client, table
+):
+    with pytest.raises(ClientError) as ei:
+        dynamodb_client.transact_write_items(
+            TransactItems=[
+                {
+                    "ConditionCheck": {
+                        "TableName": table,
+                        "Key": {"pk": {"S": OVERSIZED_HASH}},
+                        "ConditionExpression": "attribute_exists(pk)",
+                    }
+                }
+            ]
+        )
+    _assert_cancelled_validation(ei)
+
+
+def test_transact_get_oversized_hash_key_cancels(dynamodb_client, table):
+    with pytest.raises(ClientError) as ei:
+        dynamodb_client.transact_get_items(
+            TransactItems=[
+                {"Get": {"TableName": table, "Key": {"pk": {"S": OVERSIZED_HASH}}}}
+            ]
+        )
+    _assert_cancelled_validation(ei)
+
+
+def test_transact_write_empty_hash_key_is_top_level_validation(dynamodb_client, table):
+    # Empty key value is a top-level ValidationException, NOT a per-item
+    # cancellation — the size-vs-empty distinction.
+    with pytest.raises(ClientError) as ei:
+        dynamodb_client.transact_write_items(
+            TransactItems=[
+                {"Put": {"TableName": table, "Item": {"pk": {"S": ""}}}}
+            ]
+        )
+    assert ei.value.response["Error"]["Code"] == "ValidationException"
+
+
+def test_transact_get_valid_key_succeeds(dynamodb_client, table):
+    resp = dynamodb_client.transact_get_items(
+        TransactItems=[{"Get": {"TableName": table, "Key": {"pk": {"S": "ok"}}}}]
+    )
+    assert "Responses" in resp
diff --git a/tests/test_wording_parity_validation.py b/tests/test_wording_parity_validation.py
new file mode 100644
index 00000000..a5d16fb5
--- /dev/null
+++ b/tests/test_wording_parity_validation.py
@@ -0,0 +1,125 @@
+# Copyright 2026 ExtendDB contributors
+# SPDX-License-Identifier: Apache-2.0
+
+"""Validation message/behavior parity (verified against real DynamoDB, us-east-1):
+
+- PutItem/DeleteItem ReturnValues: a valid-but-disallowed enum value
+  (UPDATED_OLD) -> "ReturnValues can only be ALL_OLD or NONE"; a non-enum value
+  (GARBAGE) -> generic constraint error with the full enum set.
+- Query/Scan Select + ProjectionExpression rejection carries the
+  "1 validation error detected: " prefix.
+- Query/Scan Select=ALL_ATTRIBUTES on a non-ALL GSI is rejected.
+"""
+
+from __future__ import annotations
+
+import pytest
+from botocore.exceptions import ClientError
+
+
+@pytest.fixture()
+def table(create_and_cleanup_table):
+    return create_and_cleanup_table()["TableDescription"]["TableName"]
+
+
+@pytest.fixture()
+def gsi_table(create_and_cleanup_table):
+    # KEYS_ONLY GSI so Select=ALL_ATTRIBUTES against it is invalid.
+    result = create_and_cleanup_table(
+        AttributeDefinitions=[
+            {"AttributeName": "pk", "AttributeType": "S"},
+            {"AttributeName": "g", "AttributeType": "S"},
+        ],
+        KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}],
+        GlobalSecondaryIndexes=[
+            {
+                "IndexName": "g_index",
+                "KeySchema": [{"AttributeName": "g", "KeyType": "HASH"}],
+                "Projection": {"ProjectionType": "KEYS_ONLY"},
+            }
+        ],
+    )
+    return result["TableDescription"]["TableName"]
+
+
+def test_put_disallowed_return_values(dynamodb_client, table):
+    with pytest.raises(ClientError) as ei:
+        dynamodb_client.put_item(
+            TableName=table, Item={"pk": {"S": "k"}}, ReturnValues="UPDATED_OLD"
+        )
+    assert ei.value.response["Error"]["Message"] == "ReturnValues can only be ALL_OLD or NONE"
+
+
+def test_delete_disallowed_return_values(dynamodb_client, table):
+    with pytest.raises(ClientError) as ei:
+        dynamodb_client.delete_item(
+            TableName=table, Key={"pk": {"S": "k"}}, ReturnValues="ALL_NEW"
+        )
+    assert ei.value.response["Error"]["Message"] == "ReturnValues can only be ALL_OLD or NONE"
+
+
+def test_put_invalid_return_values_enum(dynamodb_client, table):
+    with pytest.raises(ClientError) as ei:
+        dynamodb_client.put_item(
+            TableName=table, Item={"pk": {"S": "k"}}, ReturnValues="GARBAGE"
+        )
+    msg = ei.value.response["Error"]["Message"]
+    assert msg == (
+        "1 validation error detected: Value 'GARBAGE' at 'returnValues' failed to "
+        "satisfy constraint: Member must satisfy enum value set: "
+        "[ALL_NEW, UPDATED_OLD, ALL_OLD, NONE, UPDATED_NEW]"
+    )
+
+
+def test_query_count_with_projection_prefix(dynamodb_client, table):
+    with pytest.raises(ClientError) as ei:
+        dynamodb_client.query(
+            TableName=table,
+            KeyConditionExpression="pk = :p",
+            ExpressionAttributeValues={":p": {"S": "x"}},
+            Select="COUNT",
+            ProjectionExpression="pk",
+        )
+    assert ei.value.response["Error"]["Message"] == (
+        "1 validation error detected: Cannot specify the ProjectionExpression "
+        "when choosing to get only the Count"
+    )
+
+
+def test_scan_count_with_projection_no_prefix(dynamodb_client, table):
+    # Scan (unlike Query) has NO "1 validation error detected: " prefix here.
+    with pytest.raises(ClientError) as ei:
+        dynamodb_client.scan(
+            TableName=table, Select="COUNT", ProjectionExpression="pk"
+        )
+    assert ei.value.response["Error"]["Message"] == (
+        "Cannot specify the ProjectionExpression when choosing to get only the Count"
+    )
+
+
+def test_scan_all_attributes_on_non_all_gsi(dynamodb_client, gsi_table):
+    with pytest.raises(ClientError) as ei:
+        dynamodb_client.scan(
+            TableName=gsi_table, IndexName="g_index", Select="ALL_ATTRIBUTES"
+        )
+    assert ei.value.response["Error"]["Code"] == "ValidationException"
+    assert (
+        "Select type ALL_ATTRIBUTES is not supported for global secondary index "
+        "g_index because its projection type is not ALL"
+        in ei.value.response["Error"]["Message"]
+    )
+
+
+def test_query_all_attributes_on_non_all_gsi(dynamodb_client, gsi_table):
+    with pytest.raises(ClientError) as ei:
+        dynamodb_client.query(
+            TableName=gsi_table,
+            IndexName="g_index",
+            KeyConditionExpression="g = :v",
+            ExpressionAttributeValues={":v": {"S": "x"}},
+            Select="ALL_ATTRIBUTES",
+        )
+    assert (
+        "Select type ALL_ATTRIBUTES is not supported"
+        in ei.value.response["Error"]["Message"]
+    )