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..95c9c448 100644 --- a/crates/storage-cassandra/src/data/query.rs +++ b/crates/storage-cassandra/src/data/query.rs @@ -726,20 +726,63 @@ impl CassandraEngine { rows_result }; - // Step 8: Parse results + // 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 json_str: String = - cassandra_util::get_column(&row, "item_data", "query parse")?; - json_to_item(json_str) + 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::, _>>()?; + .collect::, _>>()? + .into_iter() + .flatten() + .collect(); // 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 has_more = items.len() > actual_limit; + 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( + 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)); + } + // 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 new file mode 100644 index 00000000..7d0d0658 --- /dev/null +++ b/crates/storage-cassandra/tests/direct/query_rows_without_item.rs @@ -0,0 +1,260 @@ +// 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. 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}; +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"))); + + // 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, 10, 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}" + ); + } +} + +#[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"))) + ); +} 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 new file mode 100644 index 00000000..9b2ff15c --- /dev/null +++ b/tests/test_query_emptied_partition.py @@ -0,0 +1,237 @@ +# 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. 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. +""" + +from __future__ import annotations + +import time + +import pytest + +from conftest import _is_real_dynamodb, 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, "ConsistentRead": True}), + ("descending", {"ScanIndexForward": False, "ConsistentRead": True}), + ("count", {"Select": "COUNT", "ConsistentRead": True}), +] + + +@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_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( + TableName=table, KeyConditionExpression="pk = :p AND sk >= :a", + ExpressionAttributeValues={":p": {"S": "emptied-sk"}, ":a": {"S": "a"}}, + ConsistentRead=True, + ) + 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 _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): + query = {"TableName": table, "IndexName": "by-g", "KeyConditionExpression": "g = :g", + "ExpressionAttributeValues": {":g": {"S": "g-emptied-gsi"}}} + _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 + + +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; 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( + 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) + + +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", "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