Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
93 changes: 93 additions & 0 deletions crates/fakecloud-athena/src/service/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1227,6 +1227,99 @@ mod tests {
serde_json::from_slice(resp.body.expect_bytes()).unwrap()
}

#[test]
fn get_query_results_paginates_large_result_set() {
// Regression: GetQueryResults honors MaxResults + NextToken. Before
// the fix it returned only the first MaxResults rows and emitted no
// NextToken, leaving the rest of a large result set unreachable.
let svc = AthenaService::new(SharedAthenaState::default());
let id = "qe-paginate";
{
let mut state = svc.state.write();
let account = super::account_mut(&mut state, "123456789012");
let now = Utc::now();
account.query_executions.insert(
id.to_string(),
crate::state::QueryExecution {
query_execution_id: id.to_string(),
query: "SELECT n FROM t".to_string(),
statement_type: "DML".to_string(),
work_group: "primary".to_string(),
state: "SUCCEEDED".to_string(),
state_change_reason: None,
submission_time: now,
completion_time: Some(now),
query_execution_context: None,
result_configuration: None,
engine_version: None,
data_scanned_bytes: 0,
engine_execution_time_ms: 1,
query_planning_time_ms: 1,
total_execution_time_ms: 2,
result_rows: (0..5).map(|n| vec![n.to_string()]).collect(),
result_columns: vec![("n".to_string(), "integer".to_string())],
},
);
}

let mut data_rows: Vec<String> = Vec::new();
let mut next: Option<String> = None;
let mut pages = 0;
loop {
pages += 1;
let mut body = json!({ "QueryExecutionId": id, "MaxResults": 3 });
if let Some(t) = &next {
body["NextToken"] = json!(t);
}
let resp = svc
.get_query_results(&req("GetQueryResults", body))
.unwrap();
let parsed = parse_json(&resp);
let rows = parsed["ResultSet"]["Rows"].as_array().unwrap();
// The column-header row is emitted only on the first page.
let start = if pages == 1 { 1 } else { 0 };
for row in &rows[start..] {
data_rows.push(row["Data"][0]["VarCharValue"].as_str().unwrap().to_string());
}
match parsed.get("NextToken").and_then(|v| v.as_str()) {
Some(t) => next = Some(t.to_string()),
None => break,
}
assert!(pages < 10, "pagination did not terminate");
}

assert!(pages >= 2, "large result set must span multiple pages");
assert_eq!(data_rows, vec!["0", "1", "2", "3", "4"]);

// MaxResults=1 (header consumes the only slot on page 1) must still
// terminate and surface every data row rather than loop forever.
let mut data_rows: Vec<String> = Vec::new();
let mut next: Option<String> = None;
let mut pages = 0;
loop {
pages += 1;
let mut body = json!({ "QueryExecutionId": id, "MaxResults": 1 });
if let Some(t) = &next {
body["NextToken"] = json!(t);
}
let resp = svc
.get_query_results(&req("GetQueryResults", body))
.unwrap();
let parsed = parse_json(&resp);
let rows = parsed["ResultSet"]["Rows"].as_array().unwrap();
let start = if pages == 1 { 1 } else { 0 };
for row in &rows[start..] {
data_rows.push(row["Data"][0]["VarCharValue"].as_str().unwrap().to_string());
}
match parsed.get("NextToken").and_then(|v| v.as_str()) {
Some(t) => next = Some(t.to_string()),
None => break,
}
assert!(pages < 100, "MaxResults=1 pagination did not terminate");
}
assert_eq!(data_rows, vec!["0", "1", "2", "3", "4"]);
}

#[test]
fn substitute_parameters_replaces_placeholders() {
let out = substitute_parameters(
Expand Down
57 changes: 50 additions & 7 deletions crates/fakecloud-athena/src/service/queries.rs
Original file line number Diff line number Diff line change
Expand Up @@ -266,7 +266,14 @@ impl AthenaService {
) -> Result<AwsResponse, AwsServiceError> {
let body = req.json_body();
let id = require_str(&body, "QueryExecutionId")?;
// Smithy: MaxResults targets MaxRowsCount @range(1,1000);
// NextToken targets Token @length(1,1024).
let max_results = validate_max_results(&body, 1, 1000)?;
validate_opt_string_len(&body, "NextToken", 1, 1024)?;
let next_token = body
.get("NextToken")
.and_then(Value::as_str)
.map(str::to_owned);
let mut state = self.state.write();
let account = account_mut(&mut state, &req.account_id);
let q = account
Expand Down Expand Up @@ -297,22 +304,58 @@ impl AthenaService {
})
})
.collect();
let header_row = json!({
"Data": q.result_columns.iter().map(|(n, _)| json!({"VarCharValue": n})).collect::<Vec<_>>(),
});
let mut rows = vec![header_row];
for row in q.result_rows.iter().take(max_results.saturating_sub(1)) {
// Pagination: the header row (column names) is emitted only on the
// first page (NextToken absent) and counts against MaxResults there.
// The NextToken is a numeric offset into the data rows; subsequent
// pages carry data rows only. Without this, result sets larger than
// MaxResults were unreachable past page 1 (bug-hunt 2026-07-16).
let offset: usize = match next_token.as_deref() {
None => 0,
Some(tok) => tok
.parse()
.map_err(|_| invalid_request("Invalid NextToken"))?,
};
let total = q.result_rows.len();
let start = offset.min(total);
// The column-header row is emitted only on the first invocation (no
// NextToken supplied). Keying this on token absence rather than
// `offset == 0` avoids an infinite loop when MaxResults is 1: the
// first page returns only the header with NextToken "0", and the
// resumed page must then advance past offset 0 rather than treating
// itself as the first page again.
let include_header = next_token.is_none();
let data_budget = if include_header {
max_results.saturating_sub(1)
} else {
max_results
};
let end = start.saturating_add(data_budget).min(total);

let mut rows = Vec::new();
if include_header {
rows.push(json!({
"Data": q.result_columns.iter().map(|(n, _)| json!({"VarCharValue": n})).collect::<Vec<_>>(),
}));
}
for row in &q.result_rows[start..end] {
rows.push(json!({
"Data": row.iter().map(|v| json!({"VarCharValue": v})).collect::<Vec<_>>(),
}));
}
Ok(AwsResponse::ok_json(json!({
let mut response = json!({
"ResultSet": {
"Rows": rows,
"ResultSetMetadata": {"ColumnInfo": column_info},
},
"UpdateCount": 0,
})))
});
if end < total {
response
.as_object_mut()
.unwrap()
.insert("NextToken".to_string(), Value::String(end.to_string()));
}
Ok(AwsResponse::ok_json(response))
}

pub(super) fn get_query_runtime_statistics(
Expand Down
54 changes: 47 additions & 7 deletions crates/fakecloud-cloudwatch/src/service.rs
Original file line number Diff line number Diff line change
Expand Up @@ -532,6 +532,45 @@ pub(crate) fn parse_dimensions_query(req: &AwsRequest, prefix: &str) -> BTreeMap
out
}

/// Parse `{prefix}.member.N.Name` / `.Value` query params into ListMetrics'
/// `DimensionFilter` list, where `Value` is OPTIONAL (per the Smithy model).
/// A name-only filter matches any metric carrying a dimension with that name
/// (any value); a name+value filter is an exact match.
///
/// Distinct from [`parse_dimensions_query`], which drops a name-only entry —
/// correct for the put/statistics APIs (exact dimension sets) but wrong for
/// ListMetrics, where a name-only filter must still narrow the results
/// instead of silently returning every metric in the namespace.
pub(crate) fn parse_dimension_filters(
req: &AwsRequest,
prefix: &str,
) -> Vec<(String, Option<String>)> {
let mut dims: BTreeMap<u32, (Option<String>, Option<String>)> = BTreeMap::new();
let needle = format!("{prefix}.member.");
for (k, v) in req.query_params.iter() {
let Some(rest) = k.strip_prefix(&needle) else {
continue;
};
let mut parts = rest.splitn(2, '.');
let Some(idx_str) = parts.next() else {
continue;
};
let Ok(idx) = idx_str.parse::<u32>() else {
continue;
};
let field = parts.next().unwrap_or("");
let entry = dims.entry(idx).or_default();
match field {
"Name" => entry.0 = Some(v.clone()),
"Value" => entry.1 = Some(v.clone()),
_ => {}
}
}
dims.into_values()
.filter_map(|(name, value)| name.map(|n| (n, value)))
.collect()
}

/// Validate the length of an optional string param against `[min, max]`.
/// Returns a 4xx on violation. AWS measures length in characters; the
/// conformance probe only sends ASCII so byte length is equivalent here.
Expand Down Expand Up @@ -954,7 +993,7 @@ impl CloudWatchService {
validate_enum(req, "RecentlyActive", &["PT3H"])?;
let namespace = optional_query_param(req, "Namespace");
let metric_name = optional_query_param(req, "MetricName");
let dim_filter = parse_dimensions_query(req, "Dimensions");
let dim_filter = parse_dimension_filters(req, "Dimensions");
// ListMetrics has no MaxResults param — AWS caps each page at 500 and
// round-trips a NextToken.
const LIST_METRICS_PAGE: usize = 500;
Expand All @@ -981,13 +1020,14 @@ impl CloudWatchService {
}
}
// ListMetrics filters by dimension containment (a metric
// matches if it carries all the requested name/value
// pairs), unlike the exact-set match used by the
// statistics APIs.
// matches if it carries all the requested filters),
// unlike the exact-set match used by the statistics
// APIs. A name-only DimensionFilter matches any value.
if !dim_filter.is_empty()
&& !dim_filter
.iter()
.all(|(k, v)| d.dimensions.get(k) == Some(v))
&& !dim_filter.iter().all(|(k, v)| match v {
Some(val) => d.dimensions.get(k) == Some(val),
None => d.dimensions.contains_key(k),
})
{
continue;
}
Expand Down
50 changes: 50 additions & 0 deletions crates/fakecloud-cloudwatch/src/tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -825,6 +825,56 @@ async fn put_metric_data_json_and_query_roundtrip() {
assert!(body_of(&xml_list).contains("<MetricName>Latency</MetricName>"));
}

#[tokio::test]
async fn list_metrics_name_only_dimension_filter_narrows_results() {
// Regression: a DimensionFilter with only a Name (no Value) must still
// narrow the results. Before the fix such a filter was dropped and every
// metric in the namespace was returned.
let svc = service();
call_json(
&svc,
"PutMetricData",
serde_json::json!({
"Namespace": "NS",
"MetricData": [{
"MetricName": "M1",
"Value": 1.0,
"Dimensions": [{"Name": "Host", "Value": "a"}]
}]
}),
)
.await;
call_json(
&svc,
"PutMetricData",
serde_json::json!({
"Namespace": "NS",
"MetricData": [{
"MetricName": "M2",
"Value": 2.0,
"Dimensions": [{"Name": "Region", "Value": "us"}]
}]
}),
)
.await;

let resp = call(
&svc,
"ListMetrics",
&[("Namespace", "NS"), ("Dimensions.member.1.Name", "Host")],
)
.await;
let body = body_of(&resp);
assert!(
body.contains("<MetricName>M1</MetricName>"),
"M1 (has dimension Host) expected: {body}"
);
assert!(
!body.contains("<MetricName>M2</MetricName>"),
"M2 (no Host dimension) must be filtered out: {body}"
);
}

#[tokio::test]
async fn get_metric_statistics_json_roundtrip() {
let svc = service();
Expand Down
15 changes: 15 additions & 0 deletions crates/fakecloud-dynamodb/src/service/queries.rs
Original file line number Diff line number Diff line change
Expand Up @@ -134,6 +134,21 @@ impl DynamoDbService {
let mut matched: Vec<&HashMap<String, AttributeValue>> = items_to_scan
.iter()
.filter(|item| {
// Sparse index: an item only appears in a GSI/LSI when it
// carries every one of that index's key attributes. AWS never
// returns (or counts) an item missing the index hash/range key
// on an index query, so skip those before evaluating the key
// condition (mirrors the scan() guard, bug-hunt 2026-07-16).
if index_name.is_some() {
if !item.contains_key(hash_key_name.as_str()) {
return false;
}
if let Some(ref rk) = range_key_name {
if !item.contains_key(rk.as_str()) {
return false;
}
}
}
evaluate_key_condition(&key_condition, item, &expr_attr_names, &expr_attr_values)
})
.collect();
Expand Down
Loading
Loading