use crate::actions::helpers;
use crate::errors::{DynoxideError, Result};
use crate::expressions;
use crate::storage_backend::StorageBackend;
use crate::types::{AttributeValue, Item};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
const MAX_RESPONSE_SIZE: usize = 1_048_576;
#[derive(Debug, Default, Deserialize)]
struct QueryRequestRaw {
#[serde(rename = "TableName", default)]
table_name: Option<String>,
#[serde(rename = "KeyConditionExpression", default)]
key_condition_expression: Option<String>,
#[serde(rename = "FilterExpression", default)]
filter_expression: Option<String>,
#[serde(rename = "ProjectionExpression", default)]
projection_expression: Option<String>,
#[serde(rename = "ExpressionAttributeNames", default)]
expression_attribute_names: Option<HashMap<String, String>>,
#[serde(
rename = "ExpressionAttributeValues",
default,
deserialize_with = "crate::types::deserialize_named_empty_and_null_refusals"
)]
expression_attribute_values: Option<HashMap<String, AttributeValue>>,
#[serde(
rename = "ScanIndexForward",
default,
deserialize_with = "crate::types::deserialize_lenient_flag"
)]
scan_index_forward: Option<bool>,
#[serde(rename = "Limit", default)]
limit: Option<usize>,
#[serde(rename = "ExclusiveStartKey", default)]
exclusive_start_key: Option<serde_json::Value>,
#[serde(rename = "Select", default)]
select: Option<String>,
#[serde(
rename = "ConsistentRead",
default,
deserialize_with = "crate::types::deserialize_lenient_flag"
)]
consistent_read: Option<bool>,
#[serde(rename = "IndexName", default)]
index_name: Option<String>,
#[serde(rename = "ReturnConsumedCapacity", default)]
return_consumed_capacity: Option<String>,
#[serde(rename = "KeyConditions", default)]
key_conditions: Option<serde_json::Value>,
#[serde(rename = "AttributesToGet", default)]
attributes_to_get: Option<Vec<String>>,
#[serde(rename = "QueryFilter", default)]
query_filter: Option<serde_json::Value>,
#[serde(rename = "ConditionalOperator", default)]
conditional_operator: Option<String>,
}
#[derive(Debug, Default)]
pub struct QueryRequest {
pub table_name: String,
pub key_condition_expression: Option<String>,
pub filter_expression: Option<String>,
pub projection_expression: Option<String>,
pub expression_attribute_names: Option<HashMap<String, String>>,
pub expression_attribute_values: Option<HashMap<String, AttributeValue>>,
pub scan_index_forward: bool,
pub limit: Option<usize>,
pub exclusive_start_key: Option<HashMap<String, AttributeValue>>,
pub select: Option<String>,
pub consistent_read: Option<bool>,
pub index_name: Option<String>,
pub return_consumed_capacity: Option<String>,
pub key_conditions: Option<serde_json::Value>,
pub attributes_to_get: Option<Vec<String>>,
pub query_filter: Option<serde_json::Value>,
pub conditional_operator: Option<String>,
pub exclusive_start_key_raw: Option<serde_json::Value>,
}
impl<'de> serde::Deserialize<'de> for QueryRequest {
fn deserialize<D: serde::Deserializer<'de>>(
deserializer: D,
) -> std::result::Result<Self, D::Error> {
let raw = QueryRequestRaw::deserialize(deserializer)?;
use crate::validation::format_validation_errors;
if let Some(message) =
crate::types::malformed_condition_boolean(raw.key_conditions.as_ref())
.or_else(|| crate::types::malformed_condition_boolean(raw.query_filter.as_ref()))
{
return Err(serde::de::Error::custom(message));
}
let mut first_pass = Vec::new();
if raw.limit == Some(0) {
first_pass.push(
"Value at 'Limit' failed to satisfy constraint: \
Member must have value greater than or equal to 1"
.to_string(),
);
}
if raw.table_name.as_deref() == Some("") {
first_pass.push(crate::validation::EMPTY_TABLE_NAME_REJECTION.to_string());
}
if let Some(msg) = format_validation_errors(&first_pass) {
return Err(serde::de::Error::custom(format!("VALIDATION:{}", msg)));
}
let mut errors = crate::validation::item_table_name_errors(raw.table_name.as_deref());
let table_name = raw.table_name.unwrap_or_default();
if let Some(msg) = crate::validation::return_consumed_capacity_rejection(
raw.return_consumed_capacity.as_deref(),
) {
errors.push(msg);
}
errors.extend(helpers::missing_comparison_operators(
raw.key_conditions.as_ref(),
"keyConditions",
));
errors.extend(helpers::missing_comparison_operators(
raw.query_filter.as_ref(),
"queryFilter",
));
if let Some(msg) = format_validation_errors(&errors) {
return Err(serde::de::Error::custom(format!("VALIDATION:{}", msg)));
}
Ok(QueryRequest {
table_name,
key_condition_expression: raw.key_condition_expression,
filter_expression: raw.filter_expression,
projection_expression: raw.projection_expression,
expression_attribute_names: raw.expression_attribute_names,
expression_attribute_values: raw.expression_attribute_values,
scan_index_forward: raw.scan_index_forward.unwrap_or(true),
limit: raw.limit,
exclusive_start_key: None,
select: raw.select,
consistent_read: raw.consistent_read,
index_name: raw.index_name,
return_consumed_capacity: raw.return_consumed_capacity,
key_conditions: raw.key_conditions,
attributes_to_get: raw.attributes_to_get,
query_filter: raw.query_filter,
conditional_operator: raw.conditional_operator,
exclusive_start_key_raw: raw.exclusive_start_key,
})
}
}
#[derive(Debug, Default, Serialize)]
pub struct QueryResponse {
#[serde(rename = "Items", skip_serializing_if = "Option::is_none")]
pub items: Option<Vec<Item>>,
#[serde(rename = "Count")]
pub count: usize,
#[serde(rename = "ScannedCount")]
pub scanned_count: usize,
#[serde(rename = "LastEvaluatedKey", skip_serializing_if = "Option::is_none")]
pub last_evaluated_key: Option<HashMap<String, AttributeValue>>,
#[serde(rename = "ConsumedCapacity", skip_serializing_if = "Option::is_none")]
pub consumed_capacity: Option<crate::types::ConsumedCapacity>,
}
pub async fn execute<S: StorageBackend>(
storage: &S,
mut request: QueryRequest,
) -> Result<QueryResponse> {
crate::validation::validate_item_table_name(&request.table_name)?;
if crate::empty_names::names_include_empty(request.attributes_to_get.as_ref())
|| crate::empty_names::filter_values_have_empty_name(request.query_filter.as_ref())
|| request
.exclusive_start_key
.as_ref()
.is_some_and(crate::empty_names::map_has_empty_name)
|| request
.exclusive_start_key_raw
.as_ref()
.is_some_and(crate::empty_names::json_map_has_empty_name)
{
return Err(crate::empty_names::bare());
}
crate::empty_names::check_names(request.expression_attribute_names.as_ref())?;
if crate::empty_names::value_placeholder(request.expression_attribute_values.as_ref(), None)
.is_some()
{
return Err(crate::empty_names::bare());
}
{
let mut non_expr = Vec::new();
let mut expr = Vec::new();
if request.attributes_to_get.is_some() {
non_expr.push("AttributesToGet");
}
if request.query_filter.is_some()
&& request.query_filter.as_ref().is_some_and(|v| !v.is_null())
{
non_expr.push("QueryFilter");
}
if request.conditional_operator.is_some() {
non_expr.push("ConditionalOperator");
}
if request.key_conditions.is_some()
&& request
.key_conditions
.as_ref()
.is_some_and(|v| !v.is_null())
{
non_expr.push("KeyConditions");
}
if request.projection_expression.is_some() {
expr.push("ProjectionExpression");
}
if request.filter_expression.is_some() {
expr.push("FilterExpression");
}
if request.key_condition_expression.is_some() {
expr.push("KeyConditionExpression");
}
let no_raw_eav: Option<serde_json::Value> = None;
let ctx = helpers::ExpressionParamContext {
non_expression_params: non_expr,
expression_params: expr,
all_expression_param_names: vec!["FilterExpression", "KeyConditionExpression"],
expression_attribute_names: &request.expression_attribute_names,
expression_attribute_values: &request.expression_attribute_values,
expression_attribute_values_raw: &no_raw_eav,
layer: crate::validation::ValueLayer::Current,
};
helpers::validate_expression_params(&ctx)?;
}
if let Some(ref sel) = request.select
&& ![
"ALL_ATTRIBUTES",
"ALL_PROJECTED_ATTRIBUTES",
"COUNT",
"SPECIFIC_ATTRIBUTES",
]
.contains(&sel.as_str())
{
return Err(DynoxideError::ValidationException(
crate::validation::envelope_message(&format!(
"Value '{sel}' at 'select' failed to satisfy constraint: Member must satisfy \
enum value set: [SPECIFIC_ATTRIBUTES, COUNT, ALL_ATTRIBUTES, \
ALL_PROJECTED_ATTRIBUTES]"
)),
));
}
if let Some(msg) = crate::validation::conditional_operator_misuse(
request.conditional_operator.as_deref(),
crate::validation::legacy_condition_count(request.query_filter.as_ref()),
) {
return Err(DynoxideError::ValidationException(msg.to_string()));
}
crate::value_nesting::check_query_conditions(request.query_filter.as_ref())?;
crate::value_nesting::check_query_conditions(request.key_conditions.as_ref())?;
helpers::validate_filter_conditions_raw(
request.query_filter.as_ref(),
"QueryFilter",
crate::validation::ValueLayer::Current,
)?;
helpers::validate_filter_conditions_raw(
request.key_conditions.as_ref(),
"KeyConditions",
crate::validation::ValueLayer::Current,
)?;
helpers::validate_filter_condition_args(request.query_filter.as_ref(), "queryFilter")?;
helpers::validate_filter_condition_args(request.key_conditions.as_ref(), "keyConditions")?;
if let Some(ref attrs) = request.attributes_to_get {
helpers::validate_attributes_to_get_no_duplicates(attrs)?;
}
let exclusive_start_key = if let Some(ref esk_val) = request.exclusive_start_key_raw {
Some(
helpers::parse_exclusive_start_key(esk_val, crate::validation::ValueLayer::Current)
.map_err(start_key_value_error)?,
)
} else {
request.exclusive_start_key.clone()
};
crate::value_nesting::check_query_start_key(exclusive_start_key.as_ref())?;
if let Some(ref kce) = request.key_condition_expression
&& kce.is_empty()
{
return Err(DynoxideError::ValidationException(
"Invalid KeyConditionExpression: The expression can not be empty;".to_string(),
));
}
if let Some(ref fe) = request.filter_expression {
if fe.is_empty() {
if request.query_filter.is_none() || request.filter_expression.as_deref() == Some("") {
return Err(DynoxideError::ValidationException(
"Invalid FilterExpression: The expression can not be empty;".to_string(),
));
}
} else {
let parsed_fe = expressions::condition::parse(fe).map_err(|e| {
DynoxideError::ValidationException(format!("Invalid FilterExpression: {e}"))
})?;
if let Err(e) = expressions::condition::validate_name_refs(
&parsed_fe,
&request.expression_attribute_names,
) {
return Err(DynoxideError::ValidationException(format!(
"Invalid FilterExpression: {e}"
)));
}
if let Err(e) = expressions::condition::validate_operand_semantics(
&parsed_fe,
&request.expression_attribute_names,
&request.expression_attribute_values,
) {
return Err(DynoxideError::ValidationException(format!(
"Invalid FilterExpression: {e}"
)));
}
}
}
if let Some(ref pe) = request.projection_expression
&& pe.is_empty()
{
return Err(DynoxideError::ValidationException(
"Invalid ProjectionExpression: The expression can not be empty;".to_string(),
));
}
if request.select.as_deref() == Some("SPECIFIC_ATTRIBUTES")
&& request.projection_expression.is_none()
&& request.attributes_to_get.is_none()
{
return Err(DynoxideError::ValidationException(
"1 validation error detected: Must specify the AttributesToGet or ProjectionExpression when choosing to get SPECIFIC_ATTRIBUTES"
.to_string(),
));
}
if request.projection_expression.is_some()
&& let Some(select) = request.select.as_deref()
&& select != "SPECIFIC_ATTRIBUTES"
{
let target = if select == "COUNT" {
"only the Count"
} else {
select
};
return Err(DynoxideError::ValidationException(format!(
"Cannot specify the ProjectionExpression when choosing to get {target}"
)));
}
if request.select.as_deref() == Some("ALL_PROJECTED_ATTRIBUTES") && request.index_name.is_none()
{
return Err(DynoxideError::ValidationException(
"ALL_PROJECTED_ATTRIBUTES can be used only when Querying using an IndexName"
.to_string(),
));
}
if let Some(ref kce) = request.key_condition_expression
&& !kce.is_empty()
{
let temp_tracker = crate::expressions::TrackedExpressionAttributes::new(
&request.expression_attribute_names,
&request.expression_attribute_values,
);
if let Err(e) = expressions::key_condition::parse(kce, &temp_tracker) {
return Err(DynoxideError::ValidationException(key_condition_error(
kce, e,
)));
}
}
let meta = helpers::require_table_for_item_op(storage, &request.table_name).await?;
let table_key_schema = helpers::parse_key_schema(&meta)?;
let effective_pk_for_kc = if let Some(ref index_name) = request.index_name {
if let Some((pk, _)) = request
.index_name
.as_ref()
.and_then(|idx| super::lsi::parse_lsi_key_schema(&meta, idx).ok())
{
pk
} else if let Ok((pk, _)) = super::gsi::parse_gsi_key_schema(&meta, index_name) {
pk
} else {
table_key_schema.partition_key.clone()
}
} else {
table_key_schema.partition_key.clone()
};
let key_conditions: Option<HashMap<String, helpers::KeyCondition>> = match request
.key_conditions
.as_ref()
{
Some(kc_val) if request.key_condition_expression.is_none() && !kc_val.is_null() => Some(
helpers::read_legacy_conditions(kc_val, crate::validation::ValueLayer::Current)?,
),
_ => None,
};
if let Some(kc) = key_conditions
&& !kc.is_empty()
{
check_legacy_key_condition_types(&kc, &meta)?;
let converted = helpers::convert_key_conditions(&kc, Some(&effective_pk_for_kc))?;
request.key_condition_expression = Some(converted.expression);
let expr_values = request
.expression_attribute_values
.get_or_insert_with(HashMap::new);
expr_values.extend(converted.attribute_values);
let expr_names = request
.expression_attribute_names
.get_or_insert_with(HashMap::new);
expr_names.extend(converted.attribute_names);
}
let query_filter: Option<HashMap<String, helpers::FilterCondition>> =
match request.query_filter.as_ref() {
Some(qf_val) if request.filter_expression.is_none() && !qf_val.is_null() => Some(
helpers::read_legacy_conditions(qf_val, crate::validation::ValueLayer::Current)?,
),
_ => None,
};
if let Some(qf) = query_filter
&& !qf.is_empty()
{
let converted =
helpers::convert_filter_conditions(&qf, request.conditional_operator.as_deref())?;
if !converted.expression.is_empty() {
request.filter_expression = Some(converted.expression);
let expr_values = request
.expression_attribute_values
.get_or_insert_with(HashMap::new);
expr_values.extend(converted.attribute_values);
let expr_names = request
.expression_attribute_names
.get_or_insert_with(HashMap::new);
expr_names.extend(converted.attribute_names);
}
}
let legacy_projection = if request.projection_expression.is_none() {
request
.attributes_to_get
.as_ref()
.map(|attrs| helpers::attributes_to_get_to_projection(attrs))
} else {
None
};
let key_condition_expression = request.key_condition_expression.as_deref().ok_or_else(|| {
DynoxideError::ValidationException(
"Either the KeyConditions or KeyConditionExpression parameter must be specified in the request."
.to_string(),
)
})?;
let key_condition_expression = key_condition_expression.to_string();
let lsi_keys = request
.index_name
.as_ref()
.and_then(|idx| super::lsi::parse_lsi_key_schema(&meta, idx).ok());
let is_lsi = lsi_keys.is_some();
if request.consistent_read.unwrap_or(false) && request.index_name.is_some() && !is_lsi {
return Err(DynoxideError::ValidationException(
"Consistent reads are not supported on global secondary indexes".to_string(),
));
}
let index_projection_type = if let Some(ref index_name) = request.index_name {
if is_lsi {
super::lsi::parse_lsi_defs(&meta)?
.into_iter()
.find(|l| l.index_name == *index_name)
.map(|l| l.projection_type)
} else {
super::gsi::parse_gsi_defs(&meta)?
.into_iter()
.find(|g| g.index_name == *index_name)
.map(|g| g.projection_type)
}
} else {
None
};
let (effective_pk, effective_sk) = if let Some(ref index_name) = request.index_name {
if let Some(keys) = lsi_keys.clone() {
keys
} else {
reject_vector_index(&meta, index_name)?;
super::gsi::parse_gsi_key_schema(&meta, index_name)?
}
} else {
(
table_key_schema.partition_key.clone(),
table_key_schema.sort_key.clone(),
)
};
let (key_hash, key_range): (Vec<String>, Vec<String>) = match request.index_name {
Some(ref index_name) if !is_lsi => super::gsi::parse_gsi_key_attrs(&meta, index_name)?,
_ => (
vec![effective_pk.clone()],
effective_sk.iter().cloned().collect(),
),
};
if let Some(ref esk) = exclusive_start_key {
helpers::validate_esk_count_and_index_keys(
esk,
&meta,
request.index_name.as_deref(),
"The provided starting key is invalid",
)?;
helpers::validate_esk_table_keys(esk, &meta).map_err(bare_start_key_mismatch)?;
check_start_key_not_empty(
esk,
&table_key_schema.partition_key,
table_key_schema.sort_key.as_deref(),
)?;
}
let tracker = crate::expressions::TrackedExpressionAttributes::new(
&request.expression_attribute_names,
&request.expression_attribute_values,
);
let key_cond =
expressions::key_condition::parse(&key_condition_expression, &tracker).map_err(|e| {
DynoxideError::ValidationException(key_condition_error(&key_condition_expression, e))
})?;
let bound = expressions::key_condition::bind(
&key_cond,
expressions::key_condition::KeyAttributes {
hash: &key_hash,
range: &key_range,
},
&tracker,
)
.map_err(DynoxideError::ValidationException)?;
check_key_condition_types(&bound, &meta)?;
let pk_str = bound.partition_key_string().ok_or_else(|| {
DynoxideError::ValidationException(
"Cannot convert partition key value to string".to_string(),
)
})?;
let (sk_condition_sql, sk_param_values) = bound.sort_key_sql(key_range.len());
if let Some(ref esk) = exclusive_start_key {
let table_without_sort_key = request.index_name.is_none() && effective_sk.is_none();
check_start_key_within_condition(esk, &bound, table_without_sort_key)?;
}
let effective_key_attrs: Vec<String> = key_hash.iter().chain(&key_range).cloned().collect();
if let Some(ref qf_val) = request.query_filter
&& let Some(obj) = qf_val.as_object()
{
for attr_name in obj.keys() {
if effective_key_attrs.contains(attr_name) {
return Err(DynoxideError::ValidationException(format!(
"QueryFilter can only contain non-primary key attributes: \
Primary key attribute: {attr_name}"
)));
}
}
}
if request.query_filter.is_none()
&& let Some(ref fe) = request.filter_expression
&& let Ok(parsed_fe) = expressions::condition::parse(fe)
{
let top_attrs = expressions::condition::extract_top_level_attributes(
&parsed_fe,
&request.expression_attribute_names,
);
for attr in &top_attrs {
if effective_key_attrs.contains(attr) {
return Err(DynoxideError::ValidationException(format!(
"Filter Expression can only contain non-primary key attributes: \
Primary key attribute: {attr}"
)));
}
}
let mut index_key_attrs = Vec::new();
if request.index_name.is_some() {
for k in &effective_key_attrs {
if ![table_key_schema.partition_key.clone()]
.iter()
.chain(table_key_schema.sort_key.iter())
.any(|tk| tk == k)
{
index_key_attrs.push(k.clone());
}
}
}
let base_key_attrs: Vec<String> = {
let mut v = vec![table_key_schema.partition_key.clone()];
if let Some(ref sk) = table_key_schema.sort_key {
v.push(sk.clone());
}
v
};
if let Some((attr, is_index)) = expressions::condition::check_non_scalar_key_access(
&parsed_fe,
&request.expression_attribute_names,
&base_key_attrs,
&index_key_attrs,
) {
let prefix = if is_index { "IndexKey" } else { "Key" };
return Err(DynoxideError::ValidationException(format!(
"Key attributes must be scalars; \
list random access '[]' and map lookup '.' are not allowed: {prefix}: {attr}"
)));
}
}
let is_index_query = request.index_name.is_some();
let start_sk = if let Some(ref esk) = exclusive_start_key {
if !key_range.is_empty() {
let values: Option<Vec<&AttributeValue>> =
key_range.iter().map(|a| esk.get(a)).collect();
values.and_then(|v| super::gsi::composite_key_string(&v))
} else if is_index_query {
Some(String::new())
} else {
None
}
} else {
None
};
let (start_base_pk, start_base_sk) = if is_index_query {
if let Some(ref esk) = exclusive_start_key {
let base_pk = esk
.get(&table_key_schema.partition_key)
.and_then(|v| v.to_key_string());
let base_sk = if let Some(sk_name) = table_key_schema.sort_key.as_ref() {
esk.get(sk_name).and_then(|v| v.to_key_string())
} else {
Some(String::new())
};
(base_pk, base_sk)
} else {
(None, None)
}
} else {
(None, None)
};
let is_select_all_attributes = request
.select
.as_deref()
.map(|s| s.eq_ignore_ascii_case("ALL_ATTRIBUTES"))
.unwrap_or(false);
let fetch_from_base_table = if is_select_all_attributes {
if let Some(ref proj_type) = index_projection_type {
if *proj_type != crate::types::ProjectionType::ALL {
if !is_lsi {
return Err(DynoxideError::ValidationException(format!(
"One or more parameter values were invalid: \
Select type ALL_ATTRIBUTES is not supported for global secondary index {} \
because its projection type is not ALL",
request.index_name.as_deref().unwrap_or("")
)));
}
true
} else {
false
}
} else {
false
}
} else {
false
};
let fetch_limit = request.limit;
let sk_params_refs: Vec<&str> = sk_param_values.iter().map(|s| s.as_str()).collect();
let query_params = crate::storage::QueryParams {
sk_condition: sk_condition_sql.as_deref(),
sk_params: &sk_params_refs,
forward: request.scan_index_forward,
limit: fetch_limit,
exclusive_start_sk: start_sk.as_deref(),
exclusive_start_base_pk: start_base_pk.as_deref(),
exclusive_start_base_sk: start_base_sk.as_deref(),
};
let rows = if let Some(ref index_name) = request.index_name {
if is_lsi {
storage
.query_lsi_items(&request.table_name, index_name, &pk_str, &query_params)
.await?
} else {
storage
.query_gsi_items(&request.table_name, index_name, &pk_str, &query_params)
.await?
}
} else {
storage
.query_items(&request.table_name, &pk_str, &query_params)
.await?
};
let filter_expr = request
.filter_expression
.as_ref()
.map(|expr| expressions::condition::parse(expr))
.transpose()
.map_err(DynoxideError::ValidationException)?;
let projection = if let Some(ref proj_expr) = request.projection_expression {
Some(
expressions::projection::parse(proj_expr)
.map_err(DynoxideError::ValidationException)?,
)
} else {
legacy_projection.clone()
};
if let Some(ref filter) = filter_expr {
tracker.track_condition_expr(filter);
}
if let Some(ref proj) = projection {
tracker.track_projection_expr(proj);
expressions::projection::validate(proj, &tracker)
.map_err(DynoxideError::ValidationException)?;
}
let loop_tracker = crate::expressions::TrackedExpressionAttributes::without_tracking(
&request.expression_attribute_names,
&request.expression_attribute_values,
);
let is_count = request
.select
.as_deref()
.map(|s| s.eq_ignore_ascii_case("COUNT"))
.unwrap_or(false);
let mut key_attrs = vec![effective_pk.clone()];
if let Some(ref sk) = effective_sk {
key_attrs.push(sk.clone());
}
if request.index_name.is_some() {
if !key_attrs.contains(&table_key_schema.partition_key) {
key_attrs.push(table_key_schema.partition_key.clone());
}
if let Some(ref sk) = table_key_schema.sort_key
&& !key_attrs.contains(sk)
{
key_attrs.push(sk.clone());
}
}
let mut items = Vec::new();
let mut scanned_count = 0;
let mut filtered_count = 0;
let mut cumulative_size = 0;
let mut last_evaluated_item: Option<Item> = None;
let mut truncated_by_size = false;
let mut base_table_cumulative_size = 0usize;
let mut index_cumulative_size = 0usize;
for (_pk, _sk, item_json) in &rows {
let index_item: Item = serde_json::from_str(item_json).map_err(|e| {
DynoxideError::InternalServerError(format!("Bad item JSON in storage: {e}"))
})?;
index_cumulative_size += crate::types::item_size(&index_item);
let item = if fetch_from_base_table {
let base_pk = index_item
.get(&table_key_schema.partition_key)
.and_then(|v| v.to_key_string())
.unwrap_or_default();
let base_sk = table_key_schema
.sort_key
.as_ref()
.and_then(|sk_name| index_item.get(sk_name))
.and_then(|v| v.to_key_string())
.unwrap_or_default();
if let Some(full_json) = storage
.get_item(&request.table_name, &base_pk, &base_sk)
.await?
{
let full_item: Item = serde_json::from_str(&full_json).map_err(|e| {
DynoxideError::InternalServerError(format!("Bad item JSON: {e}"))
})?;
base_table_cumulative_size += crate::types::item_size(&full_item);
full_item
} else {
index_item.clone()
}
} else {
index_item.clone()
};
scanned_count += 1;
let item_size = crate::types::item_size(&item);
if cumulative_size + item_size > MAX_RESPONSE_SIZE && scanned_count > 1 {
truncated_by_size = true;
break;
}
cumulative_size += item_size;
if let Some(ref filter) = filter_expr {
let passes = expressions::condition::evaluate(filter, &item, &loop_tracker)
.map_err(DynoxideError::ValidationException)?;
if !passes {
last_evaluated_item = Some(index_item);
continue;
}
}
filtered_count += 1;
let result_item = if let Some(ref proj) = projection {
let no_keys: &[String] = &[];
expressions::projection::apply(&item, proj, &loop_tracker, no_keys)
.map_err(DynoxideError::ValidationException)?
} else {
item
};
last_evaluated_item = Some(index_item);
if !is_count {
items.push(result_item);
}
}
tracker.check_unused()?;
let count = if is_count {
filtered_count
} else {
items.len()
};
let has_more = truncated_by_size
|| (fetch_limit.is_some() && scanned_count >= fetch_limit.unwrap_or(usize::MAX));
let is_gsi_query = request.index_name.is_some() && !is_lsi;
let last_evaluated_key = if has_more {
last_evaluated_item.map(|item| {
let mut key = build_last_evaluated_key(&item, &key_hash, &key_range);
if is_lsi
&& let Some(tsk) = table_key_schema.sort_key.as_deref()
&& !key.contains_key(tsk)
&& let Some(v) = item.get(tsk)
{
key.insert(tsk.to_string(), v.clone());
}
if is_gsi_query {
if !key.contains_key(&table_key_schema.partition_key)
&& let Some(v) = item.get(&table_key_schema.partition_key)
{
key.insert(table_key_schema.partition_key.clone(), v.clone());
}
if let Some(ref tsk) = table_key_schema.sort_key
&& !key.contains_key(tsk)
&& let Some(v) = item.get(tsk)
{
key.insert(tsk.clone(), v.clone());
}
}
key
})
} else {
None
};
let is_gsi = is_gsi_query;
let consistent = request.consistent_read.unwrap_or(false);
let consumed_capacity = if is_gsi {
let mut gsi_units = std::collections::HashMap::new();
gsi_units.insert(
request.index_name.as_ref().unwrap().clone(),
crate::types::read_capacity_units_with_consistency(cumulative_size, consistent),
);
crate::types::consumed_capacity_with_indexes(
&request.table_name,
0.0,
&gsi_units,
&request.return_consumed_capacity,
)
} else if is_lsi {
let (table_cap, lsi_cap) = if fetch_from_base_table {
let table_rcu = crate::types::read_capacity_units_with_consistency(
base_table_cumulative_size,
consistent,
);
let lsi_rcu = crate::types::read_capacity_units_with_consistency(
index_cumulative_size,
consistent,
);
(table_rcu, lsi_rcu)
} else {
(
0.0,
crate::types::read_capacity_units_with_consistency(cumulative_size, consistent),
)
};
let mut lsi_units = std::collections::HashMap::new();
lsi_units.insert(request.index_name.as_ref().unwrap().clone(), lsi_cap);
crate::types::consumed_capacity_with_secondary_indexes(
&request.table_name,
table_cap,
&std::collections::HashMap::new(),
&lsi_units,
&request.return_consumed_capacity,
)
} else {
crate::types::consumed_capacity(
&request.table_name,
crate::types::read_capacity_units_with_consistency(cumulative_size, consistent),
&request.return_consumed_capacity,
)
};
Ok(QueryResponse {
items: if is_count { None } else { Some(items) },
count,
scanned_count,
last_evaluated_key,
consumed_capacity,
})
}
fn build_last_evaluated_key(
item: &Item,
hash: &[String],
range: &[String],
) -> HashMap<String, AttributeValue> {
hash.iter()
.chain(range)
.filter_map(|name| item.get(name).map(|v| (name.clone(), v.clone())))
.collect()
}
fn check_key_condition_types(
bound: &expressions::key_condition::BoundKeyCondition,
meta: &crate::storage::TableMetadata,
) -> Result<()> {
let defs: Vec<crate::types::AttributeDefinition> =
serde_json::from_str(&meta.attribute_definitions).map_err(|e| {
DynoxideError::InternalServerError(format!("Bad attribute definitions JSON: {e}"))
})?;
for (attr, value) in bound.compared_values() {
if let Some(def) = defs.iter().find(|d| d.attribute_name == attr)
&& !helpers::attr_value_matches_scalar_type(value, &def.attribute_type)
{
return Err(DynoxideError::ValidationException(
"One or more parameter values were invalid: Condition parameter type does not \
match schema type"
.to_string(),
));
}
}
Ok(())
}
fn reject_vector_index(
meta: &crate::storage::TableMetadata,
index_name: &str,
) -> crate::errors::Result<()> {
if super::vector_index::parse_vector_defs(meta)?
.iter()
.any(|v| v.index_name == index_name)
{
return Err(DynoxideError::ValidationException(
"Query operation not supported on this index type.".to_string(),
));
}
Ok(())
}
fn key_condition_error(expr: &str, err: String) -> String {
if err.contains("Expression size has exceeded") || err.contains("redundant parentheses") {
return err;
}
match expressions::condition::syntax_error(expr) {
Some(syntax) => format!("Invalid KeyConditionExpression: {syntax}"),
None => err,
}
}
const START_KEY_INVALID: &str = "The provided starting key is invalid";
fn bare_start_key_mismatch(err: DynoxideError) -> DynoxideError {
match err {
DynoxideError::ValidationException(msg)
if msg
== "The provided starting key is invalid: \
The provided key element does not match the schema" =>
{
DynoxideError::ValidationException(START_KEY_INVALID.to_string())
}
other => other,
}
}
fn start_key_value_error(err: DynoxideError) -> DynoxideError {
match err {
DynoxideError::ValidationException(msg) => {
match msg.strip_prefix("The provided starting key is invalid: ") {
Some(rest) if rest.starts_with("One or more parameter values were invalid:") => {
DynoxideError::ValidationException(rest.to_string())
}
_ => DynoxideError::ValidationException(msg),
}
}
other => other,
}
}
fn check_start_key_within_condition(
esk: &Item,
bound: &expressions::key_condition::BoundKeyCondition,
table_without_sort_key: bool,
) -> Result<()> {
use crate::expressions::condition::{CompOp, compare_values};
const AT_MOST_ONE_ROW: &str = "The query can return at most one row and cannot be restarted";
let refuse = |message: &str| Err(DynoxideError::ValidationException(message.to_string()));
let holds = |name: &str, value: &AttributeValue| {
esk.get(name)
.is_some_and(|v| compare_values(v, &CompOp::Eq, value))
};
let Some(((last_name, last_value), leading)) = bound.hash.split_last() else {
return Ok(());
};
if leading.iter().any(|(name, value)| !holds(name, value)) {
return refuse(AT_MOST_ONE_ROW);
}
let outside_equalities = bound
.range_eq
.iter()
.any(|(name, value)| esk.contains_key(name) && !holds(name, value));
let outside_last = bound.range_last.as_ref().is_some_and(|condition| {
esk.get(condition.sk_name())
.is_some_and(|value| !sort_key_satisfies(value, condition))
});
if outside_equalities || outside_last {
return refuse("The provided starting key does not match the range key predicate");
}
if !holds(last_name, last_value) {
if !bound.range_eq.is_empty() || bound.range_last.is_some() {
return refuse(AT_MOST_ONE_ROW);
}
if table_without_sort_key {
return refuse("The provided exclusive start key is invalid");
}
return refuse(
"The provided starting key is outside query boundaries based on provided conditions",
);
}
Ok(())
}
fn sort_key_satisfies(
value: &AttributeValue,
condition: &expressions::key_condition::ResolvedSortKeyCondition,
) -> bool {
use crate::expressions::condition::{CompOp, compare_values};
use expressions::key_condition::ResolvedSortKeyCondition as Sk;
match condition {
Sk::Eq(_, v) => compare_values(value, &CompOp::Eq, v),
Sk::Lt(_, v) => compare_values(value, &CompOp::Lt, v),
Sk::Le(_, v) => compare_values(value, &CompOp::Le, v),
Sk::Gt(_, v) => compare_values(value, &CompOp::Gt, v),
Sk::Ge(_, v) => compare_values(value, &CompOp::Ge, v),
Sk::Between(_, low, high) => {
compare_values(value, &CompOp::Ge, low) && compare_values(value, &CompOp::Le, high)
}
Sk::BeginsWith(_, prefix) => match (value, prefix) {
(AttributeValue::S(s), AttributeValue::S(p)) => s.starts_with(p.as_str()),
(AttributeValue::B(b), AttributeValue::B(p)) => b.starts_with(p),
_ => false,
},
}
}
fn check_legacy_key_condition_types(
conditions: &HashMap<String, helpers::KeyCondition>,
meta: &crate::storage::TableMetadata,
) -> Result<()> {
use crate::types::{AttributeDefinition, ScalarAttributeType};
let definitions: Vec<AttributeDefinition> = serde_json::from_str(&meta.attribute_definitions)
.map_err(|e| {
DynoxideError::InternalServerError(format!("Bad attribute definitions JSON: {e}"))
})?;
let lists = || {
conditions
.iter()
.filter_map(|(name, condition)| Some((name, condition.attribute_value_list.as_ref()?)))
};
if lists().any(|(_, values)| {
values
.windows(2)
.any(|pair| std::mem::discriminant(&pair[0]) != std::mem::discriminant(&pair[1]))
}) {
return Err(DynoxideError::ValidationException(
"One or more parameter values were invalid: \
AttributeValues inside AttributeValueList must be of same type"
.to_string(),
));
}
for (name, values) in lists() {
let Some(definition) = definitions.iter().find(|d| &d.attribute_name == name) else {
continue;
};
let matches = |value: &AttributeValue| {
matches!(
(value, &definition.attribute_type),
(AttributeValue::S(_), ScalarAttributeType::S)
| (AttributeValue::N(_), ScalarAttributeType::N)
| (AttributeValue::B(_), ScalarAttributeType::B)
)
};
if !values.iter().all(matches) {
return Err(DynoxideError::ValidationException(
"One or more parameter values were invalid: \
Condition parameter type does not match schema type"
.to_string(),
));
}
}
Ok(())
}
fn check_start_key_not_empty(
esk: &Item,
partition_key: &str,
sort_key: Option<&str>,
) -> Result<()> {
for name in std::iter::once(partition_key).chain(sort_key) {
let kind = match esk.get(name) {
Some(AttributeValue::S(s)) if s.is_empty() => "string",
Some(AttributeValue::B(b)) if b.is_empty() => "binary",
_ => continue,
};
return Err(DynoxideError::KeyEmptyValueValidation(format!(
"One or more parameter values are not valid. The AttributeValue for a key \
attribute cannot contain an empty {kind} value. Key: {name}"
)));
}
Ok(())
}