From e2d348b6e4faaaa731cf56c4369f50476027adff Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Mon, 26 Oct 2026 06:51:09 +0000 Subject: [PATCH 1/7] fix(cassandra): a partition with no items reads as no items DeleteItem sets the partition's static column partition_max_delete_timestamp. When the last item of a partition is deleted, Cassandra still answers a read of that partition with one row carrying only the static columns, with no item_data. Query parsed that row as an item and failed with `Column or Udt property 'item_data' is empty`, a 500, where Amazon DynamoDB answers Count 0 (measured 2026-10-26 at 06:48 UTC, every Query shape, twice). The GSI backfill's scan of the base table read the same row and failed the same way. Both now skip a row without item_data, as Scan already did. --- crates/storage-cassandra/src/data/ddl.rs | 11 +++++++++-- crates/storage-cassandra/src/data/query.rs | 18 +++++++++++++----- 2 files changed, 22 insertions(+), 7 deletions(-) diff --git a/crates/storage-cassandra/src/data/ddl.rs b/crates/storage-cassandra/src/data/ddl.rs index 9016422d..10a97b8b 100644 --- a/crates/storage-cassandra/src/data/ddl.rs +++ b/crates/storage-cassandra/src/data/ddl.rs @@ -498,8 +498,15 @@ pub(crate) async fn backfill_gsi( let mut batch_has_rows = false; for row in &rows { - let item_json: String = - crate::cassandra_util::get_column(row, "item_data", "backfill_gsi")?; + // A partition whose items have all been deleted answers with one + // row of static columns and no `item_data`; it holds no item. + let item_json: Option = + cdrs_tokio::types::IntoRustByName::get_by_name(row, "item_data").map_err(|e| { + StorageError::Internal(format!("backfill_gsi parse item_data failed: {e}")) + })?; + let Some(item_json) = item_json else { + continue; + }; let item = super::json_to_item(item_json)?; if !item_has_index_keys(&item, index_key_schema) { diff --git a/crates/storage-cassandra/src/data/query.rs b/crates/storage-cassandra/src/data/query.rs index 65b0c242..32f5da22 100644 --- a/crates/storage-cassandra/src/data/query.rs +++ b/crates/storage-cassandra/src/data/query.rs @@ -3,6 +3,7 @@ //! Query implementation for Cassandra backend. +use cdrs_tokio::types::IntoRustByName; use extenddb_core::expression::{CompareOp, ExpressionMaps, KeyCondition, SortKeyCondition}; use extenddb_core::types::{Item, ScalarAttributeType, TableKeyInfo}; use extenddb_storage::error::StorageError; @@ -726,15 +727,22 @@ impl CassandraEngine { rows_result }; - // Step 8: Parse results + // Step 8: Parse results. A partition whose items have all been deleted + // still holds its static columns (`partition_max_delete_timestamp`), + // and Cassandra answers a partition read with one row carrying only + // those, with no `item_data`. That row is no item. let items: Vec = rows .into_iter() .map(|row| { - let json_str: String = - cassandra_util::get_column(&row, "item_data", "query parse")?; - json_to_item(json_str) + let item_data: Option = row.get_by_name("item_data").map_err(|error| { + StorageError::Internal(format!("query parse item_data failed: {error}")) + })?; + item_data.map(json_to_item).transpose() }) - .collect::, _>>()?; + .collect::, _>>()? + .into_iter() + .flatten() + .collect(); // Step 9: Enforce limit and detect pagination #[allow(clippy::cast_sign_loss, clippy::cast_possible_truncation)] From 17d9fdb7fb6098a08448c83b4cf299bbab26cf04 Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Mon, 26 Oct 2026 07:59:18 +0000 Subject: [PATCH 2/7] test(cassandra): reads of a partition whose items have all been deleted Dual-target wire tests from the 2026-10-26 measurements: a Query of an emptied partition in every shape (plain, consistent, Limit 1, descending, COUNT, with a sort key condition) answers no items, as does an index Query; GetItem of a deleted item answers no Item; a partition with one item left answers it; and a GSI added by UpdateTable to a table with an emptied partition becomes ACTIVE and holds only the live item. On Cassandra before the fix the first five answered 500 and the index stayed CREATING. --- tests/test_query_emptied_partition.py | 158 ++++++++++++++++++++++++++ 1 file changed, 158 insertions(+) create mode 100644 tests/test_query_emptied_partition.py diff --git a/tests/test_query_emptied_partition.py b/tests/test_query_emptied_partition.py new file mode 100644 index 00000000..fcef3212 --- /dev/null +++ b/tests/test_query_emptied_partition.py @@ -0,0 +1,158 @@ +# Copyright 2026 ExtendDB contributors +# SPDX-License-Identifier: Apache-2.0 + +"""Reads of a partition whose items have all been deleted. + +Amazon DynamoDB answers a Query of such a partition with no items, whatever +its shape, on the table and on a global secondary index, and a GetItem of a +deleted item with no Item. A partition with one item left answers that item. +A global secondary index added by UpdateTable to a table holding such a +partition becomes ACTIVE and holds only the items that exist. Measured on +2026-10-26 at 06:48, 06:52, 06:59 and 07:09 UTC (us-east-1), each case twice. + +Dual-target against Amazon DynamoDB and extenddb. +""" + +from __future__ import annotations + +import time + +import pytest + +from conftest import scoped_table + + +@pytest.fixture(scope="module") +def table(dynamodb_client): + with scoped_table( + dynamodb_client, + attribute_definitions=[ + {"AttributeName": "pk", "AttributeType": "S"}, + {"AttributeName": "sk", "AttributeType": "S"}, + {"AttributeName": "g", "AttributeType": "S"}, + ], + key_schema=[ + {"AttributeName": "pk", "KeyType": "HASH"}, + {"AttributeName": "sk", "KeyType": "RANGE"}, + ], + GlobalSecondaryIndexes=[{ + "IndexName": "by-g", + "KeySchema": [{"AttributeName": "g", "KeyType": "HASH"}], + "Projection": {"ProjectionType": "ALL"}, + }], + ) as name: + yield name + + +def _fill_and_empty(client, table, pk, keep=()): + for sk in ("a", "b", "c"): + client.put_item(TableName=table, Item={"pk": {"S": pk}, "sk": {"S": sk}, "g": {"S": f"g-{pk}"}}) + for sk in ("a", "b", "c"): + if sk not in keep: + client.delete_item(TableName=table, Key={"pk": {"S": pk}, "sk": {"S": sk}}) + + +def _query(client, table, pk, **kw): + return client.query(TableName=table, KeyConditionExpression="pk = :p", + ExpressionAttributeValues={":p": {"S": pk}}, **kw) + + +SHAPES = [ + ("plain", {}), + ("consistent", {"ConsistentRead": True}), + ("limit_1", {"Limit": 1}), + ("descending", {"ScanIndexForward": False}), + ("count", {"Select": "COUNT"}), +] + + +@pytest.mark.parametrize(("case", "kw"), SHAPES, ids=[c[0] for c in SHAPES]) +def test_query_of_an_emptied_partition_answers_no_items(dynamodb_client, table, case, kw): + pk = f"emptied-{case}" + _fill_and_empty(dynamodb_client, table, pk) + resp = _query(dynamodb_client, table, pk, **kw) + assert resp["Count"] == 0 + assert resp.get("Items", []) == [] + + +def test_query_with_a_sort_key_condition(dynamodb_client, table): + _fill_and_empty(dynamodb_client, table, "emptied-sk") + resp = dynamodb_client.query( + TableName=table, KeyConditionExpression="pk = :p AND sk >= :a", + ExpressionAttributeValues={":p": {"S": "emptied-sk"}, ":a": {"S": "a"}}, + ) + assert resp["Count"] == 0 + + +def test_get_item_of_a_deleted_item(dynamodb_client, table): + _fill_and_empty(dynamodb_client, table, "emptied-get") + resp = dynamodb_client.get_item(TableName=table, Key={"pk": {"S": "emptied-get"}, "sk": {"S": "a"}}, + ConsistentRead=True) + assert "Item" not in resp + + +def test_a_partition_with_one_item_left(dynamodb_client, table): + _fill_and_empty(dynamodb_client, table, "one-left", keep=("c",)) + for kw in ({"ConsistentRead": True}, {"ConsistentRead": True, "Limit": 1}): + resp = _query(dynamodb_client, table, "one-left", **kw) + assert [i["sk"]["S"] for i in resp["Items"]] == ["c"], kw + + +def test_index_query_of_an_emptied_partition(dynamodb_client, table): + _fill_and_empty(dynamodb_client, table, "emptied-gsi") + query = {"TableName": table, "IndexName": "by-g", "KeyConditionExpression": "g = :g", + "ExpressionAttributeValues": {":g": {"S": "g-emptied-gsi"}}} + # The index is eventually consistent; wait for the deletes to reach it. + for _ in range(40): + resp = dynamodb_client.query(**query) + if resp["Count"] == 0: + break + time.sleep(0.5) + assert resp["Count"] == 0 + assert dynamodb_client.query(**query, ScanIndexForward=False)["Count"] == 0 + + +def test_an_index_added_later_skips_an_emptied_partition(dynamodb_client): + with scoped_table( + dynamodb_client, + attribute_definitions=[ + {"AttributeName": "pk", "AttributeType": "S"}, + {"AttributeName": "sk", "AttributeType": "S"}, + ], + key_schema=[ + {"AttributeName": "pk", "KeyType": "HASH"}, + {"AttributeName": "sk", "KeyType": "RANGE"}, + ], + ) as name: + for sk in ("a", "b"): + dynamodb_client.put_item(TableName=name, Item={"pk": {"S": "emptied"}, "sk": {"S": sk}, "g": {"S": "x"}}) + dynamodb_client.delete_item(TableName=name, Key={"pk": {"S": "emptied"}, "sk": {"S": sk}}) + dynamodb_client.put_item(TableName=name, Item={"pk": {"S": "live"}, "sk": {"S": "a"}, "g": {"S": "y"}}) + dynamodb_client.update_table( + TableName=name, + AttributeDefinitions=[{"AttributeName": n, "AttributeType": "S"} for n in ("pk", "sk", "g")], + GlobalSecondaryIndexUpdates=[{"Create": { + "IndexName": "by-g", + "KeySchema": [{"AttributeName": "g", "KeyType": "HASH"}], + "Projection": {"ProjectionType": "ALL"}, + }}], + ) + # The service took 520 s and 1,011 s to build this index. + deadline = time.monotonic() + 1500 + while time.monotonic() < deadline: + desc = dynamodb_client.describe_table(TableName=name)["Table"] + if desc["TableStatus"] == "ACTIVE" and all( + i["IndexStatus"] == "ACTIVE" for i in desc.get("GlobalSecondaryIndexes", []) + ): + break + time.sleep(1) + else: + raise AssertionError("the index did not become ACTIVE") + + def count(g): + return dynamodb_client.query( + TableName=name, IndexName="by-g", KeyConditionExpression="g = :g", + ExpressionAttributeValues={":g": {"S": g}}, + )["Count"] + + assert (count("y"), count("x")) == (1, 0) From ec8c780c793069f4c293f7d909133b7d13fb9e3e Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Mon, 26 Oct 2026 08:06:23 +0000 Subject: [PATCH 3/7] test(cassandra): a hash-only table whose item was deleted A hash-only table stores the delete timestamp as a regular column, so it has no static-only row; Query, GetItem and Scan after deleting its item answer nothing on the service (measured 2026-10-26 at 08:00 UTC, twice) and on Cassandra before and after the fix. The test keeps that covered. --- tests/test_query_emptied_partition.py | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/tests/test_query_emptied_partition.py b/tests/test_query_emptied_partition.py index fcef3212..127eb790 100644 --- a/tests/test_query_emptied_partition.py +++ b/tests/test_query_emptied_partition.py @@ -8,7 +8,8 @@ deleted item with no Item. A partition with one item left answers that item. A global secondary index added by UpdateTable to a table holding such a partition becomes ACTIVE and holds only the items that exist. Measured on -2026-10-26 at 06:48, 06:52, 06:59 and 07:09 UTC (us-east-1), each case twice. +2026-10-26 at 06:48, 06:52, 06:59, 07:09 and 08:00 UTC (us-east-1), each case +twice. Dual-target against Amazon DynamoDB and extenddb. """ @@ -156,3 +157,15 @@ def count(g): )["Count"] assert (count("y"), count("x")) == (1, 0) + + +def test_a_hash_only_table_whose_item_was_deleted(dynamodb_client): + with scoped_table(dynamodb_client) as name: + dynamodb_client.put_item(TableName=name, Item={"pk": {"S": "k"}, "x": {"S": "x"}}) + dynamodb_client.delete_item(TableName=name, Key={"pk": {"S": "k"}}) + query = {"TableName": name, "KeyConditionExpression": "pk = :p", + "ExpressionAttributeValues": {":p": {"S": "k"}}} + for kw in ({}, {"ConsistentRead": True}, {"Select": "COUNT"}): + assert dynamodb_client.query(**query, **kw)["Count"] == 0, kw + assert "Item" not in dynamodb_client.get_item(TableName=name, Key={"pk": {"S": "k"}}, ConsistentRead=True) + assert dynamodb_client.scan(TableName=name, ConsistentRead=True)["Count"] == 0 From 1df5da3845a6307730bf9d59e2a0fa7dbee2ae54 Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Mon, 26 Oct 2026 10:21:25 +0000 Subject: [PATCH 4/7] fix(cassandra): rows without an item no longer end a Query page early A TTL claim or a transaction prepare on an absent key writes a data row with only prepared_txn_id set and no item_data. Query asks Cassandra for LIMIT limit + 1 rows, so such rows take places in the read. Skipping them and then deciding "more pages" from the items kept returned a short page with no LastEvaluatedKey while items remained: a TTL table under concurrent writes to new keys returned 128 of 300 items in 9 of 1,162 full reads. On main the same rows answer 500. When rows were skipped and the read filled its LIMIT, Query now reads again with room for the skipped rows. The direct test writes such rows with CQL before, between and after the items and pages through the partition with every page size. The wire tests read with ConsistentRead where the shape allows, and the index test waits 120 s against extenddb instead of 1,500 s. --- crates/storage-cassandra/src/data/query.rs | 47 +++- .../tests/direct/query_rows_without_item.rs | 209 ++++++++++++++++++ .../tests/direct_integration.rs | 2 + tests/test_query_emptied_partition.py | 16 +- 4 files changed, 259 insertions(+), 15 deletions(-) create mode 100644 crates/storage-cassandra/tests/direct/query_rows_without_item.rs diff --git a/crates/storage-cassandra/src/data/query.rs b/crates/storage-cassandra/src/data/query.rs index 32f5da22..2c1d31e6 100644 --- a/crates/storage-cassandra/src/data/query.rs +++ b/crates/storage-cassandra/src/data/query.rs @@ -3,7 +3,6 @@ //! Query implementation for Cassandra backend. -use cdrs_tokio::types::IntoRustByName; use extenddb_core::expression::{CompareOp, ExpressionMaps, KeyCondition, SortKeyCondition}; use extenddb_core::types::{Item, ScalarAttributeType, TableKeyInfo}; use extenddb_storage::error::StorageError; @@ -727,16 +726,21 @@ impl CassandraEngine { rows_result }; - // Step 8: Parse results. A partition whose items have all been deleted - // still holds its static columns (`partition_max_delete_timestamp`), - // and Cassandra answers a partition read with one row carrying only - // those, with no `item_data`. That row is no item. + // Step 8: Parse results. A row without `item_data` holds no item: + // a partition whose items have all been deleted answers a partition + // read with one row of its static columns + // (`partition_max_delete_timestamp`), and a TTL claim or a transaction + // prepare on an absent key writes a row with only `prepared_txn_id`. + let fetched_rows = rows.len(); let items: Vec = rows .into_iter() .map(|row| { - let item_data: Option = row.get_by_name("item_data").map_err(|error| { - StorageError::Internal(format!("query parse item_data failed: {error}")) - })?; + let item_data: Option = + cdrs_tokio::types::IntoRustByName::get_by_name(&row, "item_data").map_err( + |error| { + StorageError::Internal(format!("query parse item_data failed: {error}")) + }, + )?; item_data.map(json_to_item).transpose() }) .collect::, _>>()? @@ -747,6 +751,33 @@ impl CassandraEngine { // Step 9: Enforce limit and detect pagination #[allow(clippy::cast_sign_loss, clippy::cast_possible_truncation)] let actual_limit = limit.map_or(1_000_000_usize, |l| l.max(0) as usize); + let skipped_rows = fetched_rows - items.len(); + if skipped_rows > 0 && fetched_rows > actual_limit { + // The skipped rows took places in the LIMIT, so this read cannot + // tell whether items follow. Read again with room for them. + let wider_limit = i64::try_from(actual_limit + skipped_rows).unwrap_or(i64::MAX); + // Through the trait method, whose boxed future breaks the + // recursion in this function's own future type. + let (mut wider_items, wider_key) = extenddb_storage::DataEngine::query( + self, + key_info, + key_condition, + expression_maps, + forward, + Some(wider_limit), + exclusive_start_key, + index_name, + ) + .await?; + if wider_items.len() <= actual_limit { + return Ok((wider_items, wider_key)); + } + wider_items.truncate(actual_limit); + let last_key = wider_items + .last() + .map(|item| extenddb_core::types::extract_key(item, &key_info.key_schema)); + return Ok((wider_items, last_key)); + } let has_more = items.len() > actual_limit; let final_items: Vec = items.into_iter().take(actual_limit).collect(); diff --git a/crates/storage-cassandra/tests/direct/query_rows_without_item.rs b/crates/storage-cassandra/tests/direct/query_rows_without_item.rs new file mode 100644 index 00000000..6d784c89 --- /dev/null +++ b/crates/storage-cassandra/tests/direct/query_rows_without_item.rs @@ -0,0 +1,209 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Query over data rows that hold no item. +//! +//! A partition whose items have all been deleted answers a partition read +//! with one row of its static columns, and a TTL claim or a transaction +//! prepare on an absent key writes a row with only `prepared_txn_id` set. +//! Neither row is an item, and neither may end a page early. + +use extenddb_core::expression::{Expr, ExpressionMaps, KeyCondition, PathElement}; +use extenddb_core::types::{AttributeValue, Item}; +use extenddb_storage::DataEngine; +use extenddb_storage_cassandra::CassandraEngine; +use std::collections::BTreeMap; + +use crate::helpers::{TestTable, setup_engine}; + +fn key(sk: &str) -> Item { + BTreeMap::from([ + ("id".to_string(), AttributeValue::S("p".to_string())), + ("sort".to_string(), AttributeValue::S(sk.to_string())), + ]) +} + +async fn put(engine: &CassandraEngine, table: &TestTable, sk: &str) { + let mut item = key(sk); + item.insert("v".to_string(), AttributeValue::S("x".to_string())); + engine + .put_item( + &table.key_info, + item, + false, + None, + &ExpressionMaps::default(), + None, + ) + .await + .expect("put_item"); +} + +/// Writes the row a TTL claim on the absent key `sk` leaves: no `item_data`. +async fn claim_row(engine: &CassandraEngine, table: &TestTable, sk: &str) { + let keyspace = engine.account_keyspace(&table.key_info.account_id); + let data_table = format!("items_{}", table.key_info.table_id.replace('-', "_")); + engine + .session_arc() + .query_with_values( + format!( + "UPDATE {keyspace}.{data_table} SET prepared_txn_id = uuid(), \ + prepared_txn_timestamp = 1 WHERE pk = ? AND sk_s = ?" + ), + cdrs_tokio::query_values!("p", sk), + ) + .await + .expect("claim row"); +} + +async fn query( + engine: &CassandraEngine, + table: &TestTable, + forward: bool, + limit: Option, + start: Option<&Item>, +) -> (Vec, Option) { + let condition = KeyCondition { + pk_path: vec![PathElement::Attribute("id".to_string())], + pk_value: Expr::Placeholder(":p".to_string()), + sk_condition: None, + extra_pk_conditions: vec![], + extra_sk_conditions: vec![], + }; + let mut maps = ExpressionMaps::default(); + maps.values + .insert(":p".to_string(), AttributeValue::S("p".to_string())); + let (items, last_key) = engine + .query( + &table.key_info, + &condition, + &maps, + forward, + limit, + start, + None, + ) + .await + .expect("query"); + let sort_keys = items + .iter() + .map(|item| match item.get("sort") { + Some(AttributeValue::S(s)) => s.clone(), + other => panic!("sort key {other:?}"), + }) + .collect(); + (sort_keys, last_key) +} + +/// Pages through the partition with `limit` and returns every sort key read. +async fn read_all( + engine: &CassandraEngine, + table: &TestTable, + forward: bool, + limit: i64, +) -> Vec { + let mut all = Vec::new(); + let mut start: Option = None; + for _ in 0..100 { + let (page, last_key) = query(engine, table, forward, Some(limit), start.as_ref()).await; + all.extend(page); + match last_key { + Some(k) => start = Some(k), + None => return all, + } + } + panic!("more than 100 pages"); +} + +#[tokio::test] +async fn test_query_of_an_emptied_partition() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "QueryEmptied", true).await; + for sk in ["a", "b"] { + put(&engine, &table, sk).await; + } + for sk in ["a", "b"] { + engine + .delete_item( + &table.key_info, + &key(sk), + false, + None, + &Default::default(), + None, + ) + .await + .expect("delete_item"); + } + for (forward, limit) in [ + (true, None), + (true, Some(1)), + (false, None), + (false, Some(1)), + ] { + assert_eq!( + query(&engine, &table, forward, limit, None).await, + (Vec::new(), None), + "forward {forward}, limit {limit:?}" + ); + } +} + +#[tokio::test] +async fn test_query_pages_past_rows_without_an_item() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "QueryClaimRows", true).await; + let live: Vec = (0..10).map(|i| format!("s{i:02}")).collect(); + for sk in &live { + put(&engine, &table, sk).await; + } + // Two before every item, two between s04 and s05, two after every item. + for sk in ["a0", "a1", "s04a", "s04b", "t0", "t1"] { + claim_row(&engine, &table, sk).await; + } + let reversed: Vec = live.iter().rev().cloned().collect(); + + // The first page holds the first item even when the rows before it + // fill the LIMIT. + let (page, last_key) = query(&engine, &table, true, Some(1), None).await; + assert_eq!(page, ["s00"]); + assert_eq!(last_key, Some(key("s00"))); + let (page, last_key) = query(&engine, &table, false, Some(1), None).await; + assert_eq!(page, ["s09"]); + assert_eq!(last_key, Some(key("s09"))); + + // A Limit equal to the items left reads them all on one page. + let (page, last_key) = query(&engine, &table, true, Some(10), None).await; + assert_eq!(page, live); + assert_eq!(last_key, None); + + // No Limit. + assert_eq!( + query(&engine, &table, true, None, None).await, + (live.clone(), None) + ); + assert_eq!( + query(&engine, &table, false, None, None).await, + (reversed.clone(), None) + ); + + // Every item, whatever the page size. + for limit in [1, 2, 3, 4, 5, 9, 11] { + assert_eq!( + read_all(&engine, &table, true, limit).await, + live, + "limit {limit}" + ); + assert_eq!( + read_all(&engine, &table, false, limit).await, + reversed, + "limit {limit}" + ); + } +} diff --git a/crates/storage-cassandra/tests/direct_integration.rs b/crates/storage-cassandra/tests/direct_integration.rs index 43eb28b8..4f3250b1 100644 --- a/crates/storage-cassandra/tests/direct_integration.rs +++ b/crates/storage-cassandra/tests/direct_integration.rs @@ -39,6 +39,8 @@ mod policies; mod put_get_item; #[path = "direct/query.rs"] mod query; +#[path = "direct/query_rows_without_item.rs"] +mod query_rows_without_item; #[path = "direct/roles.rs"] mod roles; #[path = "direct/scan.rs"] diff --git a/tests/test_query_emptied_partition.py b/tests/test_query_emptied_partition.py index 127eb790..2fe42c65 100644 --- a/tests/test_query_emptied_partition.py +++ b/tests/test_query_emptied_partition.py @@ -20,7 +20,7 @@ import pytest -from conftest import scoped_table +from conftest import _is_real_dynamodb, scoped_table @pytest.fixture(scope="module") @@ -61,9 +61,9 @@ def _query(client, table, pk, **kw): SHAPES = [ ("plain", {}), ("consistent", {"ConsistentRead": True}), - ("limit_1", {"Limit": 1}), - ("descending", {"ScanIndexForward": False}), - ("count", {"Select": "COUNT"}), + ("limit_1", {"Limit": 1, "ConsistentRead": True}), + ("descending", {"ScanIndexForward": False, "ConsistentRead": True}), + ("count", {"Select": "COUNT", "ConsistentRead": True}), ] @@ -81,6 +81,7 @@ def test_query_with_a_sort_key_condition(dynamodb_client, table): resp = dynamodb_client.query( TableName=table, KeyConditionExpression="pk = :p AND sk >= :a", ExpressionAttributeValues={":p": {"S": "emptied-sk"}, ":a": {"S": "a"}}, + ConsistentRead=True, ) assert resp["Count"] == 0 @@ -138,8 +139,9 @@ def test_an_index_added_later_skips_an_emptied_partition(dynamodb_client): "Projection": {"ProjectionType": "ALL"}, }}], ) - # The service took 520 s and 1,011 s to build this index. - deadline = time.monotonic() + 1500 + # The service took 520 s and 1,011 s to build this index; extenddb + # builds it in seconds. + deadline = time.monotonic() + (1500 if _is_real_dynamodb() else 120) while time.monotonic() < deadline: desc = dynamodb_client.describe_table(TableName=name)["Table"] if desc["TableStatus"] == "ACTIVE" and all( @@ -165,7 +167,7 @@ def test_a_hash_only_table_whose_item_was_deleted(dynamodb_client): dynamodb_client.delete_item(TableName=name, Key={"pk": {"S": "k"}}) query = {"TableName": name, "KeyConditionExpression": "pk = :p", "ExpressionAttributeValues": {":p": {"S": "k"}}} - for kw in ({}, {"ConsistentRead": True}, {"Select": "COUNT"}): + for kw in ({}, {"ConsistentRead": True}, {"Select": "COUNT", "ConsistentRead": True}): assert dynamodb_client.query(**query, **kw)["Count"] == 0, kw assert "Item" not in dynamodb_client.get_item(TableName=name, Key={"pk": {"S": "k"}}, ConsistentRead=True) assert dynamodb_client.scan(TableName=name, ConsistentRead=True)["Count"] == 0 From 9c1e6afbb2357dff6e7283fe7ea7a1f8eb12d01a Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Mon, 26 Oct 2026 12:07:52 +0000 Subject: [PATCH 5/7] fix(cassandra): a Query page past rows without an item reads one page The previous commit read again with a LIMIT widened by the rows it had skipped, and the wider read applied the same test to itself. A row without an item stays inside every wider window, so one such row in front of a large partition made the Query read the whole partition one item per level: 1,501 Cassandra reads for a Limit 1 page over 1,500 items, and at 3,000 items the server overflowed its stack and aborted. Query now decides "more pages" from the rows Cassandra returned: a read that filled its LIMIT reports more pages, even when rows without an item leave the page short, and the engine reads on from the last item. Only a read in which no row holds an item is read again, with twice the LIMIT, so the reads grow with the run of such rows, not with the partition. The direct tests add a 3,000-item partition behind one such row and a run of 3,000 such rows in front of two items. --- crates/storage-cassandra/src/data/query.rs | 16 +++-- .../tests/direct/query_rows_without_item.rs | 65 +++++++++++++++++-- 2 files changed, 68 insertions(+), 13 deletions(-) diff --git a/crates/storage-cassandra/src/data/query.rs b/crates/storage-cassandra/src/data/query.rs index 2c1d31e6..95c9c448 100644 --- a/crates/storage-cassandra/src/data/query.rs +++ b/crates/storage-cassandra/src/data/query.rs @@ -751,11 +751,12 @@ impl CassandraEngine { // Step 9: Enforce limit and detect pagination #[allow(clippy::cast_sign_loss, clippy::cast_possible_truncation)] let actual_limit = limit.map_or(1_000_000_usize, |l| l.max(0) as usize); - let skipped_rows = fetched_rows - items.len(); - if skipped_rows > 0 && fetched_rows > actual_limit { - // The skipped rows took places in the LIMIT, so this read cannot - // tell whether items follow. Read again with room for them. - let wider_limit = i64::try_from(actual_limit + skipped_rows).unwrap_or(i64::MAX); + if items.is_empty() && fetched_rows > actual_limit { + // Every row of a read that filled its LIMIT held no item, so there + // is no item to continue from. Read again with twice the room; a + // further read happens only if all of those rows hold no item too. + let wider_limit = + i64::try_from(actual_limit.saturating_mul(2).max(1)).unwrap_or(i64::MAX); // Through the trait method, whose boxed future breaks the // recursion in this function's own future type. let (mut wider_items, wider_key) = extenddb_storage::DataEngine::query( @@ -778,7 +779,10 @@ impl CassandraEngine { .map(|item| extenddb_core::types::extract_key(item, &key_info.key_schema)); return Ok((wider_items, last_key)); } - let has_more = items.len() > actual_limit; + // A read that filled its LIMIT may have items after it, even when + // rows without an item leave this page short; the caller reads on + // from the last item. + let has_more = fetched_rows > actual_limit; let final_items: Vec = items.into_iter().take(actual_limit).collect(); diff --git a/crates/storage-cassandra/tests/direct/query_rows_without_item.rs b/crates/storage-cassandra/tests/direct/query_rows_without_item.rs index 6d784c89..7d0d0658 100644 --- a/crates/storage-cassandra/tests/direct/query_rows_without_item.rs +++ b/crates/storage-cassandra/tests/direct/query_rows_without_item.rs @@ -6,7 +6,9 @@ //! A partition whose items have all been deleted answers a partition read //! with one row of its static columns, and a TTL claim or a transaction //! prepare on an absent key writes a row with only `prepared_txn_id` set. -//! Neither row is an item, and neither may end a page early. +//! Neither row is an item. A page may come back short because of them, but +//! while items remain it carries a LastEvaluatedKey, so a reader paging on +//! from the last item reads every item. use extenddb_core::expression::{Expr, ExpressionMaps, KeyCondition, PathElement}; use extenddb_core::types::{AttributeValue, Item}; @@ -178,11 +180,6 @@ async fn test_query_pages_past_rows_without_an_item() { assert_eq!(page, ["s09"]); assert_eq!(last_key, Some(key("s09"))); - // A Limit equal to the items left reads them all on one page. - let (page, last_key) = query(&engine, &table, true, Some(10), None).await; - assert_eq!(page, live); - assert_eq!(last_key, None); - // No Limit. assert_eq!( query(&engine, &table, true, None, None).await, @@ -194,7 +191,7 @@ async fn test_query_pages_past_rows_without_an_item() { ); // Every item, whatever the page size. - for limit in [1, 2, 3, 4, 5, 9, 11] { + for limit in [1, 2, 3, 4, 5, 9, 10, 11] { assert_eq!( read_all(&engine, &table, true, limit).await, live, @@ -207,3 +204,57 @@ async fn test_query_pages_past_rows_without_an_item() { ); } } + +#[tokio::test] +async fn test_query_past_a_row_without_an_item_in_a_large_partition() { + if crate::helpers::skip_without_cassandra() { + return; + } + use futures::StreamExt; + + let engine = setup_engine().await; + let table = TestTable::new(&engine, "QueryClaimRowLarge", true).await; + let live: Vec = (0..3000).map(|i| format!("s{i:04}")).collect(); + futures::stream::iter(&live) + .for_each_concurrent(32, |sk| put(&engine, &table, sk)) + .await; + claim_row(&engine, &table, "a0").await; + + // The page comes from the first rows read, whatever the size of the + // partition behind them. + let (page, last_key) = query(&engine, &table, true, Some(1), None).await; + assert_eq!(page, ["s0000"]); + assert_eq!(last_key, Some(key("s0000"))); + let (page, last_key) = query(&engine, &table, true, Some(100), None).await; + assert_eq!(page, live[..100]); + assert_eq!(last_key, Some(key("s0099"))); + assert_eq!(read_all(&engine, &table, true, 1000).await, live); +} + +#[tokio::test] +async fn test_query_past_a_long_run_of_rows_without_an_item() { + if crate::helpers::skip_without_cassandra() { + return; + } + use futures::StreamExt; + + let engine = setup_engine().await; + let table = TestTable::new(&engine, "QueryClaimRowRun", true).await; + let claims: Vec = (0..3000).map(|i| format!("a{i:04}")).collect(); + futures::stream::iter(&claims) + .for_each_concurrent(32, |sk| claim_row(&engine, &table, sk)) + .await; + for sk in ["s0", "s1"] { + put(&engine, &table, sk).await; + } + + assert_eq!( + query(&engine, &table, true, Some(1), None).await, + (vec!["s0".to_string()], Some(key("s0"))) + ); + assert_eq!(read_all(&engine, &table, true, 1).await, ["s0", "s1"]); + assert_eq!( + query(&engine, &table, false, Some(5), None).await, + (vec!["s1".to_string(), "s0".to_string()], Some(key("s0"))) + ); +} From 6293f595d420281e75db8c9ed2909a176aed909c Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Mon, 26 Oct 2026 12:17:07 +0000 Subject: [PATCH 6/7] test: wait for the index to hold the items before deleting them The index query test deleted the items right after putting them and then waited for the index to answer Count 0, which an index that had not yet received the puts answers too; a later descending query then saw an item on SQLite. The test now waits for the index to hold the three items, deletes them and waits for Count 0. --- tests/test_query_emptied_partition.py | 23 +++++++++++++++-------- 1 file changed, 15 insertions(+), 8 deletions(-) diff --git a/tests/test_query_emptied_partition.py b/tests/test_query_emptied_partition.py index 2fe42c65..d14ee994 100644 --- a/tests/test_query_emptied_partition.py +++ b/tests/test_query_emptied_partition.py @@ -100,17 +100,24 @@ def test_a_partition_with_one_item_left(dynamodb_client, table): assert [i["sk"]["S"] for i in resp["Items"]] == ["c"], kw +def _wait_for_index_count(client, query, count): + # The index is eventually consistent. + for _ in range(40): + resp = client.query(**query) + if resp["Count"] == count: + return + time.sleep(0.5) + raise AssertionError(f"index Count {resp['Count']}, expected {count}") + + def test_index_query_of_an_emptied_partition(dynamodb_client, table): - _fill_and_empty(dynamodb_client, table, "emptied-gsi") query = {"TableName": table, "IndexName": "by-g", "KeyConditionExpression": "g = :g", "ExpressionAttributeValues": {":g": {"S": "g-emptied-gsi"}}} - # The index is eventually consistent; wait for the deletes to reach it. - for _ in range(40): - resp = dynamodb_client.query(**query) - if resp["Count"] == 0: - break - time.sleep(0.5) - assert resp["Count"] == 0 + _fill_and_empty(dynamodb_client, table, "emptied-gsi", keep=("a", "b", "c")) + _wait_for_index_count(dynamodb_client, query, 3) + for sk in ("a", "b", "c"): + dynamodb_client.delete_item(TableName=table, Key={"pk": {"S": "emptied-gsi"}, "sk": {"S": sk}}) + _wait_for_index_count(dynamodb_client, query, 0) assert dynamodb_client.query(**query, ScanIndexForward=False)["Count"] == 0 From 9ef14ede96d52b0bb5a26c0740e231d24e5105cd Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Mon, 26 Oct 2026 15:13:47 +0000 Subject: [PATCH 7/7] test: capacity, ExclusiveStartKey and batch and transaction deletes Adds the Query shapes measured on the service at 14:45 UTC to the emptied-partition wire file: the capacity such a Query consumes (0.5 units, 1.0 with ConsistentRead), an ExclusiveStartKey before, inside and after the deleted items, and partitions emptied by TransactWriteItems and by BatchWriteItem. --- tests/test_query_emptied_partition.py | 67 +++++++++++++++++++++++++-- 1 file changed, 62 insertions(+), 5 deletions(-) diff --git a/tests/test_query_emptied_partition.py b/tests/test_query_emptied_partition.py index d14ee994..9b2ff15c 100644 --- a/tests/test_query_emptied_partition.py +++ b/tests/test_query_emptied_partition.py @@ -5,11 +5,13 @@ Amazon DynamoDB answers a Query of such a partition with no items, whatever its shape, on the table and on a global secondary index, and a GetItem of a -deleted item with no Item. A partition with one item left answers that item. -A global secondary index added by UpdateTable to a table holding such a -partition becomes ACTIVE and holds only the items that exist. Measured on -2026-10-26 at 06:48, 06:52, 06:59, 07:09 and 08:00 UTC (us-east-1), each case -twice. +deleted item with no Item. The Query consumes the capacity of an empty read, +and answers the same when the items were deleted by TransactWriteItems or +BatchWriteItem, or when it starts from an ExclusiveStartKey. A partition with +one item left answers that item. A global secondary index added by +UpdateTable to a table holding such a partition becomes ACTIVE and holds only +the items that exist. Measured on 2026-10-26 at 06:48, 06:52, 06:59, 07:09, +08:00 and 14:45 UTC (us-east-1), each case twice. Dual-target against Amazon DynamoDB and extenddb. """ @@ -76,6 +78,61 @@ def test_query_of_an_emptied_partition_answers_no_items(dynamodb_client, table, assert resp.get("Items", []) == [] +def test_capacity_of_a_query_of_an_emptied_partition(dynamodb_client, table): + _fill_and_empty(dynamodb_client, table, "emptied-cc") + cc = _query(dynamodb_client, table, "emptied-cc", ReturnConsumedCapacity="INDEXES")["ConsumedCapacity"] + assert (cc["CapacityUnits"], cc["Table"]["CapacityUnits"]) == (0.5, 0.5) + resp = _query(dynamodb_client, table, "emptied-cc", ReturnConsumedCapacity="TOTAL", ConsistentRead=True) + assert resp["Count"] == 0 + assert resp["ConsumedCapacity"]["CapacityUnits"] == 1.0 + + +ESK_SHAPES = [ + ("inside", "a", True), + ("before_all", "0", True), + ("after_all_descending", "zzz", False), +] + + +@pytest.mark.parametrize(("case", "sk", "forward"), ESK_SHAPES, ids=[c[0] for c in ESK_SHAPES]) +def test_query_from_an_exclusive_start_key_in_an_emptied_partition(dynamodb_client, table, case, sk, forward): + pk = f"emptied-esk-{case}" + _fill_and_empty(dynamodb_client, table, pk) + resp = _query(dynamodb_client, table, pk, ConsistentRead=True, ScanIndexForward=forward, + ExclusiveStartKey={"pk": {"S": pk}, "sk": {"S": sk}}) + assert resp["Count"] == 0 + assert "LastEvaluatedKey" not in resp + + +def _delete_by_transaction(client, table, pk): + client.transact_write_items(TransactItems=[ + {"Delete": {"TableName": table, "Key": {"pk": {"S": pk}, "sk": {"S": sk}}}} for sk in ("a", "b", "c") + ]) + + +def _delete_by_batch(client, table, pk): + client.batch_write_item(RequestItems={table: [ + {"DeleteRequest": {"Key": {"pk": {"S": pk}, "sk": {"S": sk}}}} for sk in ("a", "b", "c") + ]}) + + +TX_BATCH_SHAPES = [ + ("transaction", _delete_by_transaction, ({"ConsistentRead": True}, {"ConsistentRead": True, "Limit": 1})), + ("batch", _delete_by_batch, ({"ConsistentRead": True},)), +] + + +@pytest.mark.parametrize(("case", "delete", "shapes"), TX_BATCH_SHAPES, ids=[c[0] for c in TX_BATCH_SHAPES]) +def test_query_of_a_partition_emptied_by_a_transaction_or_a_batch(dynamodb_client, table, case, delete, shapes): + pk = f"emptied-by-{case}" + for sk in ("a", "b", "c"): + dynamodb_client.put_item(TableName=table, Item={"pk": {"S": pk}, "sk": {"S": sk}}) + delete(dynamodb_client, table, pk) + for kw in shapes: + resp = _query(dynamodb_client, table, pk, **kw) + assert (resp["Count"], resp.get("Items", [])) == (0, []), kw + + def test_query_with_a_sort_key_condition(dynamodb_client, table): _fill_and_empty(dynamodb_client, table, "emptied-sk") resp = dynamodb_client.query(