1use crate::actions::helpers;
2use crate::errors::{DynoxideError, Result};
3use crate::expressions;
4use crate::storage_backend::StorageBackend;
5use crate::types::{AttributeValue, Item};
6use serde::{Deserialize, Serialize};
7use std::collections::HashMap;
8
9const MAX_RESPONSE_SIZE: usize = 1_048_576;
11
12#[derive(Debug, Default, Deserialize)]
14struct ScanRequestRaw {
15 #[serde(rename = "TableName", default)]
16 table_name: Option<String>,
17 #[serde(rename = "FilterExpression", default)]
18 filter_expression: Option<String>,
19 #[serde(rename = "ProjectionExpression", default)]
20 projection_expression: Option<String>,
21 #[serde(rename = "ExpressionAttributeNames", default)]
22 expression_attribute_names: Option<HashMap<String, String>>,
23 #[serde(rename = "ExpressionAttributeValues", default)]
24 expression_attribute_values: Option<HashMap<String, AttributeValue>>,
25 #[serde(rename = "Limit", default)]
26 limit: Option<usize>,
27 #[serde(rename = "ExclusiveStartKey", default)]
28 exclusive_start_key: Option<serde_json::Value>,
29 #[serde(rename = "Select", default)]
30 select: Option<String>,
31 #[serde(rename = "ConsistentRead", default)]
32 consistent_read: Option<bool>,
33 #[serde(rename = "IndexName", default)]
34 index_name: Option<String>,
35 #[serde(rename = "Segment", default)]
39 segment: Option<i64>,
40 #[serde(rename = "TotalSegments", default)]
41 total_segments: Option<u32>,
42 #[serde(rename = "ReturnConsumedCapacity", default)]
43 return_consumed_capacity: Option<String>,
44 #[serde(rename = "AttributesToGet", default)]
45 attributes_to_get: Option<Vec<String>>,
46 #[serde(rename = "ScanFilter", default)]
47 scan_filter: Option<serde_json::Value>,
48 #[serde(rename = "ConditionalOperator", default)]
49 conditional_operator: Option<String>,
50}
51
52#[derive(Debug, Default)]
53pub struct ScanRequest {
54 pub table_name: String,
55 pub filter_expression: Option<String>,
56 pub projection_expression: Option<String>,
57 pub expression_attribute_names: Option<HashMap<String, String>>,
58 pub expression_attribute_values: Option<HashMap<String, AttributeValue>>,
59 pub limit: Option<usize>,
60 pub exclusive_start_key: Option<HashMap<String, AttributeValue>>,
61 pub select: Option<String>,
62 pub consistent_read: Option<bool>,
63 pub index_name: Option<String>,
64 pub segment: Option<u32>,
65 pub total_segments: Option<u32>,
66 pub return_consumed_capacity: Option<String>,
67 pub attributes_to_get: Option<Vec<String>>,
68 pub scan_filter: Option<serde_json::Value>,
69 pub conditional_operator: Option<String>,
70 pub exclusive_start_key_raw: Option<serde_json::Value>,
73}
74
75impl<'de> serde::Deserialize<'de> for ScanRequest {
76 fn deserialize<D: serde::Deserializer<'de>>(
77 deserializer: D,
78 ) -> std::result::Result<Self, D::Error> {
79 let raw = ScanRequestRaw::deserialize(deserializer)?;
80 use crate::validation::{
81 TableNameContext, format_validation_errors, table_name_constraint_errors,
82 };
83
84 let mut errors = Vec::new();
85 errors.extend(table_name_constraint_errors(
86 raw.table_name.as_deref(),
87 TableNameContext::ReadWrite,
88 ));
89 let table_name = raw.table_name.unwrap_or_default();
90
91 if let Some(msg) = crate::validation::return_consumed_capacity_rejection(
93 raw.return_consumed_capacity.as_deref(),
94 ) {
95 errors.push(msg);
96 }
97
98 if let Some(ref sel) = raw.select
100 && ![
101 "ALL_ATTRIBUTES",
102 "ALL_PROJECTED_ATTRIBUTES",
103 "COUNT",
104 "SPECIFIC_ATTRIBUTES",
105 ]
106 .contains(&sel.as_str())
107 {
108 errors.push(format!(
109 "Value '{}' at 'select' failed to satisfy constraint: \
110 Member must satisfy enum value set: [SPECIFIC_ATTRIBUTES, COUNT, ALL_ATTRIBUTES, ALL_PROJECTED_ATTRIBUTES]",
111 sel
112 ));
113 }
114
115 if let Some(limit) = raw.limit
120 && limit == 0
121 {
122 errors.push(
123 "Value '0' at 'limit' failed to satisfy constraint: \
124 Member must have value greater than or equal to 1"
125 .to_string(),
126 );
127 }
128
129 if let Some(segment) = raw.segment
132 && segment < 0
133 {
134 errors.push(format!(
135 "Value '{}' at 'segment' failed to satisfy constraint: \
136 Member must have value greater than or equal to 0",
137 segment
138 ));
139 }
140
141 if let Some(msg) = format_validation_errors(&errors) {
142 return Err(serde::de::Error::custom(format!("VALIDATION:{}", msg)));
143 }
144
145 Ok(ScanRequest {
146 table_name,
147 filter_expression: raw.filter_expression,
148 projection_expression: raw.projection_expression,
149 expression_attribute_names: raw.expression_attribute_names,
150 expression_attribute_values: raw.expression_attribute_values,
151 limit: raw.limit,
152 exclusive_start_key: None,
153 select: raw.select,
154 consistent_read: raw.consistent_read,
155 index_name: raw.index_name,
156 segment: raw.segment.map(|s| s as u32),
157 total_segments: raw.total_segments,
158 return_consumed_capacity: raw.return_consumed_capacity,
159 attributes_to_get: raw.attributes_to_get,
160 scan_filter: raw.scan_filter,
161 conditional_operator: raw.conditional_operator,
162 exclusive_start_key_raw: raw.exclusive_start_key,
163 })
164 }
165}
166
167#[derive(Debug, Default, Serialize)]
168pub struct ScanResponse {
169 #[serde(rename = "Items", skip_serializing_if = "Option::is_none")]
170 pub items: Option<Vec<Item>>,
171 #[serde(rename = "Count")]
172 pub count: usize,
173 #[serde(rename = "ScannedCount")]
174 pub scanned_count: usize,
175 #[serde(rename = "LastEvaluatedKey", skip_serializing_if = "Option::is_none")]
176 pub last_evaluated_key: Option<HashMap<String, AttributeValue>>,
177 #[serde(rename = "ConsumedCapacity", skip_serializing_if = "Option::is_none")]
178 pub consumed_capacity: Option<crate::types::ConsumedCapacity>,
179}
180
181pub async fn execute<S: StorageBackend>(
182 storage: &S,
183 mut request: ScanRequest,
184) -> Result<ScanResponse> {
185 crate::validation::validate_table_name(&request.table_name)?;
187
188 {
190 let mut non_expr = Vec::new();
191 let mut expr = Vec::new();
192 if request.attributes_to_get.is_some() {
193 non_expr.push("AttributesToGet");
194 }
195 if request.scan_filter.is_some()
196 && request.scan_filter.as_ref().is_some_and(|v| !v.is_null())
197 {
198 non_expr.push("ScanFilter");
199 }
200 if request.conditional_operator.is_some() {
201 non_expr.push("ConditionalOperator");
202 }
203 if request.projection_expression.is_some() {
204 expr.push("ProjectionExpression");
205 }
206 if request.filter_expression.is_some() {
207 expr.push("FilterExpression");
208 }
209 let no_raw_eav: Option<serde_json::Value> = None;
210 let ctx = helpers::ExpressionParamContext {
211 non_expression_params: non_expr,
212 expression_params: expr,
213 all_expression_param_names: vec!["FilterExpression"],
214 expression_attribute_names: &request.expression_attribute_names,
215 expression_attribute_values: &request.expression_attribute_values,
216 expression_attribute_values_raw: &no_raw_eav,
217 };
218 helpers::validate_expression_params(&ctx)?;
219 }
220
221 helpers::validate_filter_conditions_raw(request.scan_filter.as_ref(), "ScanFilter")?;
223
224 helpers::validate_filter_condition_args(request.scan_filter.as_ref())?;
226
227 if let Some(ref attrs) = request.attributes_to_get {
229 helpers::validate_attributes_to_get_no_duplicates(attrs)?;
230 }
231
232 let exclusive_start_key = if let Some(ref esk_val) = request.exclusive_start_key_raw {
234 Some(helpers::parse_exclusive_start_key(esk_val)?)
235 } else {
236 request.exclusive_start_key.clone()
237 };
238
239 if request.filter_expression.is_none()
241 && let Some(ref sf_val) = request.scan_filter
242 && let Ok(sf) =
243 serde_json::from_value::<HashMap<String, helpers::FilterCondition>>(sf_val.clone())
244 && !sf.is_empty()
245 {
246 let converted =
247 helpers::convert_filter_conditions(&sf, request.conditional_operator.as_deref())?;
248 if !converted.expression.is_empty() {
249 request.filter_expression = Some(converted.expression);
250 let expr_values = request
251 .expression_attribute_values
252 .get_or_insert_with(HashMap::new);
253 expr_values.extend(converted.attribute_values);
254 let expr_names = request
255 .expression_attribute_names
256 .get_or_insert_with(HashMap::new);
257 expr_names.extend(converted.attribute_names);
258 }
259 }
260
261 match (request.segment, request.total_segments) {
263 (Some(segment), Some(total)) => {
264 if !(1..=1_000_000).contains(&total) {
265 return Err(DynoxideError::ValidationException(
266 "1 validation error detected: Value at 'totalSegments' failed to satisfy constraint: \
267 Member must have value between 1 and 1000000".to_string(),
268 ));
269 }
270 if segment >= total {
271 return Err(DynoxideError::ValidationException(format!(
272 "The Segment parameter is zero-based and must be less than parameter TotalSegments: Segment: {} is not less than TotalSegments: {}",
273 segment, total
274 )));
275 }
276 }
277 (Some(_), None) => {
278 return Err(DynoxideError::ValidationException(
279 "The TotalSegments parameter is required but was not present in the request when Segment parameter is present".to_string(),
280 ));
281 }
282 (None, Some(_)) => {
283 return Err(DynoxideError::ValidationException(
284 "The Segment parameter is required but was not present in the request when parameter TotalSegments is present".to_string(),
285 ));
286 }
287 (None, None) => {}
288 }
289
290 if let Some(ref filter_expr_str) = request.filter_expression {
293 if filter_expr_str.is_empty() {
294 if request.scan_filter.is_none() || request.filter_expression.as_deref() == Some("") {
297 return Err(DynoxideError::ValidationException(
298 "Invalid FilterExpression: The expression can not be empty;".to_string(),
299 ));
300 }
301 } else {
302 let parsed_fe = expressions::condition::parse(filter_expr_str).map_err(|e| {
304 DynoxideError::ValidationException(format!("Invalid FilterExpression: {e}"))
305 })?;
306 if let Err(e) = expressions::condition::validate_name_refs(
308 &parsed_fe,
309 &request.expression_attribute_names,
310 ) {
311 return Err(DynoxideError::ValidationException(format!(
312 "Invalid FilterExpression: {e}"
313 )));
314 }
315 if let Err(e) = expressions::condition::validate_operand_semantics(
316 &parsed_fe,
317 &request.expression_attribute_names,
318 &request.expression_attribute_values,
319 ) {
320 return Err(DynoxideError::ValidationException(format!(
321 "Invalid FilterExpression: {e}"
322 )));
323 }
324 }
325 }
326 if let Some(ref proj_expr_str) = request.projection_expression
327 && proj_expr_str.is_empty()
328 {
329 return Err(DynoxideError::ValidationException(
330 "Invalid ProjectionExpression: The expression can not be empty;".to_string(),
331 ));
332 }
333
334 if request.select.as_deref() == Some("SPECIFIC_ATTRIBUTES")
336 && request.projection_expression.is_none()
337 && request.attributes_to_get.is_none()
338 {
339 return Err(DynoxideError::ValidationException(
340 "Must specify the AttributesToGet or ProjectionExpression when choosing to get SPECIFIC_ATTRIBUTES"
341 .to_string(),
342 ));
343 }
344
345 if request.projection_expression.is_some()
347 && let Some(select) = request.select.as_deref()
348 && select != "SPECIFIC_ATTRIBUTES"
349 {
350 let target = if select == "COUNT" {
352 "only the Count"
353 } else {
354 select
355 };
356 return Err(DynoxideError::ValidationException(format!(
357 "Cannot specify the ProjectionExpression when choosing to get {target}"
358 )));
359 }
360
361 if request.select.as_deref() == Some("ALL_PROJECTED_ATTRIBUTES") && request.index_name.is_none()
364 {
365 return Err(DynoxideError::ValidationException(
366 "ALL_PROJECTED_ATTRIBUTES can be used only when Querying using an IndexName"
367 .to_string(),
368 ));
369 }
370
371 let meta = helpers::require_table_for_item_op(storage, &request.table_name).await?;
372 let table_key_schema = helpers::parse_key_schema(&meta)?;
373
374 let legacy_projection = if request.projection_expression.is_none() {
377 request
378 .attributes_to_get
379 .as_ref()
380 .map(|attrs| helpers::attributes_to_get_to_projection(attrs))
381 } else {
382 None
383 };
384
385 let lsi_keys = request
387 .index_name
388 .as_ref()
389 .and_then(|idx| super::lsi::parse_lsi_key_schema(&meta, idx).ok());
390 let is_lsi = lsi_keys.is_some();
391
392 if request.consistent_read.unwrap_or(false) && request.index_name.is_some() && !is_lsi {
394 return Err(DynoxideError::ValidationException(
395 "Consistent reads are not supported on global secondary indexes".to_string(),
396 ));
397 }
398
399 let (effective_pk, effective_sk) = if let Some(ref index_name) = request.index_name {
400 if let Some(keys) = lsi_keys {
401 keys
402 } else {
403 reject_vector_index(&meta, index_name)?;
409 super::gsi::parse_gsi_key_schema(&meta, index_name)?
410 }
411 } else {
412 (
413 table_key_schema.partition_key.clone(),
414 table_key_schema.sort_key.clone(),
415 )
416 };
417
418 if let Some(ref esk) = exclusive_start_key {
421 let count_msg = if request.index_name.is_some() {
422 "The provided starting key is invalid"
423 } else {
424 "The provided starting key is invalid: The provided key element does not match the schema"
425 };
426 helpers::validate_esk_count_and_index_keys(
427 esk,
428 &meta,
429 request.index_name.as_deref(),
430 count_msg,
431 )?;
432 }
433
434 if let Some(ref index_name) = request.index_name
436 && !is_lsi
437 && let Some(ref select) = request.select
438 && select == "ALL_ATTRIBUTES"
439 {
440 let gsi_defs = super::gsi::parse_gsi_defs(&meta)?;
442 if let Some(gsi) = gsi_defs.iter().find(|g| g.index_name == *index_name)
443 && gsi.projection_type != crate::types::ProjectionType::ALL
444 {
445 return Err(DynoxideError::ValidationException(format!(
446 "One or more parameter values were invalid: \
447 Select type ALL_ATTRIBUTES is not supported for global secondary index {} \
448 because its projection type is not ALL",
449 index_name
450 )));
451 }
452 }
453
454 if let Some(ref esk) = exclusive_start_key {
456 helpers::validate_esk_table_keys(esk, &meta)?;
457 }
458
459 let (start_pk, start_sk) = if let Some(ref esk) = exclusive_start_key {
461 let pk = esk.get(&effective_pk).and_then(|v| v.to_key_string());
462 let sk = if let Some(ref sk_name) = effective_sk {
463 esk.get(sk_name).and_then(|v| v.to_key_string())
464 } else {
465 Some(String::new())
466 };
467 (pk, sk)
468 } else {
469 (None, None)
470 };
471
472 let (start_base_pk, start_base_sk) = if is_lsi || request.index_name.is_some() {
478 if let Some(ref esk) = exclusive_start_key {
479 let base_pk = esk
480 .get(&table_key_schema.partition_key)
481 .and_then(|v| v.to_key_string());
482 let base_sk = if let Some(sk_name) = table_key_schema.sort_key.as_ref() {
490 esk.get(sk_name).and_then(|v| v.to_key_string())
491 } else {
492 Some(String::new())
493 };
494 (base_pk, base_sk)
495 } else {
496 (None, None)
497 }
498 } else {
499 (None, None)
500 };
501
502 let scan_params = crate::storage::ScanParams {
504 limit: request.limit,
505 exclusive_start_pk: start_pk.as_deref(),
506 exclusive_start_sk: start_sk.as_deref(),
507 segment: request.segment,
508 total_segments: request.total_segments,
509 exclusive_start_base_pk: start_base_pk.as_deref(),
510 exclusive_start_base_sk: start_base_sk.as_deref(),
511 };
512 let rows = if let Some(ref index_name) = request.index_name {
513 if is_lsi {
514 storage
515 .scan_lsi_items(&request.table_name, index_name, &scan_params)
516 .await?
517 } else {
518 storage
519 .scan_gsi_items(&request.table_name, index_name, &scan_params)
520 .await?
521 }
522 } else {
523 storage
524 .scan_items(&request.table_name, &scan_params)
525 .await?
526 };
527
528 let tracker = crate::expressions::TrackedExpressionAttributes::new(
530 &request.expression_attribute_names,
531 &request.expression_attribute_values,
532 );
533
534 let filter_expr = request
536 .filter_expression
537 .as_ref()
538 .map(|expr| expressions::condition::parse(expr))
539 .transpose()
540 .map_err(DynoxideError::ValidationException)?;
541
542 if let Some(ref filter) = filter_expr {
544 let mut base_key_attrs = vec![table_key_schema.partition_key.clone()];
546 if let Some(ref sk) = table_key_schema.sort_key {
547 base_key_attrs.push(sk.clone());
548 }
549 let mut index_key_attrs = Vec::new();
550 if request.index_name.is_some() {
551 if !base_key_attrs.contains(&effective_pk) {
552 index_key_attrs.push(effective_pk.clone());
553 }
554 if let Some(ref sk) = effective_sk
555 && !base_key_attrs.contains(sk)
556 {
557 index_key_attrs.push(sk.clone());
558 }
559 }
560 if let Some((attr, is_index)) = expressions::condition::check_non_scalar_key_access(
561 filter,
562 &request.expression_attribute_names,
563 &base_key_attrs,
564 &index_key_attrs,
565 ) {
566 let prefix = if is_index { "IndexKey" } else { "Key" };
567 return Err(DynoxideError::ValidationException(format!(
568 "Key attributes must be scalars; \
569 list random access '[]' and map lookup '.' are not allowed: {prefix}: {attr}"
570 )));
571 }
572 }
573
574 let projection = if let Some(ref proj_expr) = request.projection_expression {
576 Some(
577 expressions::projection::parse(proj_expr)
578 .map_err(DynoxideError::ValidationException)?,
579 )
580 } else {
581 legacy_projection.clone()
582 };
583
584 if let Some(ref filter) = filter_expr {
586 tracker.track_condition_expr(filter);
587 }
588 if let Some(ref proj) = projection {
589 tracker.track_projection_expr(proj);
590 expressions::projection::validate(proj, &tracker)
593 .map_err(DynoxideError::ValidationException)?;
594 }
595
596 let loop_tracker = crate::expressions::TrackedExpressionAttributes::without_tracking(
598 &request.expression_attribute_names,
599 &request.expression_attribute_values,
600 );
601
602 let is_count = request
604 .select
605 .as_deref()
606 .map(|s| s.eq_ignore_ascii_case("COUNT"))
607 .unwrap_or(false);
608
609 let mut key_attrs = vec![effective_pk.clone()];
611 if let Some(ref sk) = effective_sk {
612 key_attrs.push(sk.clone());
613 }
614 if request.index_name.is_some() {
615 if !key_attrs.contains(&table_key_schema.partition_key) {
616 key_attrs.push(table_key_schema.partition_key.clone());
617 }
618 if let Some(ref sk) = table_key_schema.sort_key
619 && !key_attrs.contains(sk)
620 {
621 key_attrs.push(sk.clone());
622 }
623 }
624
625 let mut items = Vec::new();
626 let mut scanned_count = 0;
627 let mut filtered_count = 0;
628 let mut cumulative_size = 0;
629 let mut last_evaluated_item: Option<Item> = None;
630 let mut truncated_by_size = false;
631
632 for (_pk, _sk, item_json) in &rows {
633 let item: Item = serde_json::from_str(item_json).map_err(|e| {
634 DynoxideError::InternalServerError(format!("Bad item JSON in storage: {e}"))
635 })?;
636
637 scanned_count += 1;
638
639 let item_size = crate::types::item_size(&item);
642 if cumulative_size + item_size > MAX_RESPONSE_SIZE && scanned_count > 1 {
643 truncated_by_size = true;
644 break;
645 }
646 cumulative_size += item_size;
647
648 if let Some(ref filter) = filter_expr {
650 let passes = expressions::condition::evaluate(filter, &item, &loop_tracker)
651 .map_err(DynoxideError::ValidationException)?;
652 if !passes {
653 last_evaluated_item = Some(item);
654 continue;
655 }
656 }
657
658 filtered_count += 1;
659
660 let result_item = if let Some(ref proj) = projection {
663 let no_keys: &[String] = &[];
664 expressions::projection::apply(&item, proj, &loop_tracker, no_keys)
665 .map_err(DynoxideError::ValidationException)?
666 } else {
667 item.clone()
668 };
669
670 last_evaluated_item = Some(item);
671 if !is_count {
672 items.push(result_item);
673 }
674 }
675
676 tracker.check_unused()?;
678
679 let count = if is_count {
680 filtered_count
681 } else {
682 items.len()
683 };
684
685 let has_more = truncated_by_size
687 || (request.limit.is_some() && scanned_count >= request.limit.unwrap_or(usize::MAX));
688
689 let is_gsi_scan = request.index_name.is_some() && !is_lsi;
694 let last_evaluated_key = if has_more {
695 last_evaluated_item.map(|item| {
696 let mut key = HashMap::new();
697 if let Some(pk_val) = item.get(&effective_pk) {
698 key.insert(effective_pk.clone(), pk_val.clone());
699 }
700 if let Some(ref sk_name) = effective_sk
701 && let Some(sk_val) = item.get(sk_name)
702 {
703 key.insert(sk_name.clone(), sk_val.clone());
704 }
705 if is_lsi
707 && let Some(tsk) = table_key_schema.sort_key.as_deref()
708 && !key.contains_key(tsk)
709 && let Some(v) = item.get(tsk)
710 {
711 key.insert(tsk.to_string(), v.clone());
712 }
713 if is_gsi_scan {
715 if !key.contains_key(&table_key_schema.partition_key)
716 && let Some(v) = item.get(&table_key_schema.partition_key)
717 {
718 key.insert(table_key_schema.partition_key.clone(), v.clone());
719 }
720 if let Some(ref tsk) = table_key_schema.sort_key
721 && !key.contains_key(tsk)
722 && let Some(v) = item.get(tsk)
723 {
724 key.insert(tsk.clone(), v.clone());
725 }
726 }
727 key
728 })
729 } else {
730 None
731 };
732
733 let is_gsi = is_gsi_scan;
735 let consistent = request.consistent_read.unwrap_or(false);
736 let consumed_capacity = if is_gsi {
737 let mut gsi_units = std::collections::HashMap::new();
738 gsi_units.insert(
739 request.index_name.as_ref().unwrap().clone(),
740 crate::types::read_capacity_units_with_consistency(cumulative_size, consistent),
741 );
742 crate::types::consumed_capacity_with_indexes(
743 &request.table_name,
744 0.0,
745 &gsi_units,
746 &request.return_consumed_capacity,
747 )
748 } else {
749 crate::types::consumed_capacity(
750 &request.table_name,
751 crate::types::read_capacity_units_with_consistency(cumulative_size, consistent),
752 &request.return_consumed_capacity,
753 )
754 };
755
756 Ok(ScanResponse {
757 items: if is_count { None } else { Some(items) },
758 count,
759 scanned_count,
760 last_evaluated_key,
761 consumed_capacity,
762 })
763}
764
765fn reject_vector_index(
771 meta: &crate::storage::TableMetadata,
772 index_name: &str,
773) -> crate::errors::Result<()> {
774 if super::vector_index::parse_vector_defs(meta)?
775 .iter()
776 .any(|v| v.index_name == index_name)
777 {
778 return Err(DynoxideError::ValidationException(
779 "Scan operation not supported on this index type".to_string(),
780 ));
781 }
782 Ok(())
783}