From 97975816ead419ca9489e21a7da7d16dffc56a42 Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Mon, 26 Oct 2026 19:47:08 +0000 Subject: [PATCH 1/3] fix(cassandra): a Scan reads on past rows without an item Scan skipped rows without item_data after Cassandra had applied the LIMIT, and decided "more pages" from the items it kept. Such rows are the static-only row of a partition whose items have all been deleted, on a table with a sort key; the row a TTL claim on an absent key writes; and the row a transaction prepare leaves on an item deleted after the transaction read it. A read that met enough of them came back short with no LastEvaluatedKey, so the Scan ended early, and so did CreateBackup, export, and the TTL backfill and audit, which page through the same function. Scan now decides "more pages" from the rows Cassandra returned, so a short page carries a LastEvaluatedKey. When every row of a full read holds no item, Scan reads on after the last row it read: a base-table read also selects pk and the sort key column, and the next read goes on within the partition after a row with a sort key and from the next token after a row without one. Each further read takes up to twice as many rows and at most max(1000, LIMIT + 1), and no read goes back over rows already read. Cassandra counts a tombstone for each deleted row in the partitions such a read passes, so a continuation read that Cassandra refuses with a read failure is read again with half the rows, down to one, and later reads in the same call stay at that size. The row handling, the continuation and the two-query read that resumes after an ExclusiveStartKey are in the new data/scan_rows.rs; scan.rs and scan_paginated use it. --- crates/storage-cassandra/src/data/mod.rs | 1 + crates/storage-cassandra/src/data/scan.rs | 135 ++++++------ .../storage-cassandra/src/data/scan_rows.rs | 207 ++++++++++++++++++ 3 files changed, 276 insertions(+), 67 deletions(-) create mode 100644 crates/storage-cassandra/src/data/scan_rows.rs diff --git a/crates/storage-cassandra/src/data/mod.rs b/crates/storage-cassandra/src/data/mod.rs index 946dbd04..8acc16d0 100644 --- a/crates/storage-cassandra/src/data/mod.rs +++ b/crates/storage-cassandra/src/data/mod.rs @@ -22,6 +22,7 @@ mod put_get_item; mod query; mod query_helpers; mod scan; +mod scan_rows; pub mod transaction_ledger; mod transactions; pub(crate) mod ttl; diff --git a/crates/storage-cassandra/src/data/scan.rs b/crates/storage-cassandra/src/data/scan.rs index 6e93d4a3..be0a096c 100644 --- a/crates/storage-cassandra/src/data/scan.rs +++ b/crates/storage-cassandra/src/data/scan.rs @@ -41,6 +41,12 @@ //! additionally bounding the "next partitions" query with the segment's upper //! token bound. //! +//! ## Rows that hold no item +//! +//! The scan skips rows that hold no item (see `scan_rows`). A page they leave +//! short still carries a LastEvaluatedKey when its read filled the LIMIT, and +//! a full read of nothing but such rows reads on after its last row. +//! //! ## Known limitation //! //! Token-based pagination can, in theory, skip partition keys that hash to the @@ -49,7 +55,6 @@ //! the "token range pagination" approach approved in the workplan. use cdrs_tokio::query::QueryValues; -use cdrs_tokio::types::IntoRustByName; use cdrs_tokio::types::blob::Blob; use cdrs_tokio::types::value::Value; use extenddb_core::types::{Item, ScalarAttributeType, TableKeyInfo}; @@ -59,7 +64,9 @@ use std::sync::Arc; use super::ddl::{data_table_name, index_table_name}; use super::index::fetch_index_by_name; -use super::json_to_item; +use super::scan_rows::{ + RUN_READ_ROWS, read_after_position, read_on_after_position, row_position, rows_to_items, +}; use crate::cassandra_util::{self, CassandraSession}; /// Compute the inclusive `[lower, upper]` token bounds for a parallel-scan @@ -154,7 +161,6 @@ impl crate::CassandraEngine { }; let account_keyspace = self.account_keyspace(&key_info.account_id); - let select_prefix = format!("SELECT item_data FROM {account_keyspace}.{table_name}"); // Sort-key info for the scanned table/index, and (for index scans) the // base table's sort key — both are clustering columns on the index table. @@ -165,6 +171,20 @@ impl crate::CassandraEngine { None }; + // A base-table read also selects the primary key columns, so that a + // read whose rows all hold no item can go on after its last row. + let resume_sk_type = sk_info_opt + .filter(|_| index_name.is_none()) + .map(|(_, sk_type)| sk_type); + let select_prefix = if index_name.is_none() { + let sk_col = resume_sk_type + .map(|sk_type| format!(", {}", sk_column(sk_type))) + .unwrap_or_default(); + format!("SELECT item_data, pk{sk_col} FROM {account_keyspace}.{table_name}") + } else { + format!("SELECT item_data FROM {account_keyspace}.{table_name}") + }; + let token_bounds = segment_token_bounds(segment, total_segments); let token_upper = token_bounds.map(|(_, hi)| hi); @@ -220,22 +240,41 @@ impl crate::CassandraEngine { } }; - // Parse item_data JSON into items. - let items: Vec = rows - .into_iter() - .map(|row| { - let item_data: Option = row.get_by_name("item_data").map_err(|error| { - StorageError::Internal(format!("scan parse item_data failed: {error}")) - })?; - item_data.map(json_to_item).transpose() - }) - .collect::, StorageError>>()? - .into_iter() - .flatten() - .collect(); + // Rows without item_data are skipped; a full read of nothing but such + // rows reads on after its last row (see "Rows that hold no item"). + let mut rows = rows; + let mut read_limit = fetch_limit; + let mut ceiling = RUN_READ_ROWS.max(fetch_limit); + let (items, full) = loop { + let full = rows.len() >= read_limit; + let resume = match rows.last() { + Some(row) if full && index_name.is_none() => { + Some(row_position(row, resume_sk_type)?) + } + _ => None, + }; + let items = rows_to_items(rows)?; + match resume { + Some((pk_text, clustering)) if items.is_empty() => { + read_limit = read_limit.saturating_mul(2).min(ceiling); + rows = read_on_after_position( + &session, + &select_prefix, + &pk_text, + clustering.as_ref(), + token_upper, + &mut read_limit, + &mut ceiling, + ) + .await?; + } + _ => break (items, full), + } + }; // Enforce the limit and derive the LastEvaluatedKey from the last item. - let has_more = items.len() > actual_limit; + // A full read may have items after it even when this page is short. + let has_more = full || items.len() > actual_limit; let final_items: Vec = items.into_iter().take(actual_limit).collect(); let last_key = if has_more { @@ -346,18 +385,9 @@ impl crate::CassandraEngine { // Partition key text of the resume row (single or composite HASH key). let pk_text = composite_pk_to_text(start_key, scan_key_schema)?; - let mut all_rows = Vec::new(); - - // "Next partitions" upper token bound (bound parameter, if segmented). - let token_clause: &str = if token_upper.is_some() { - " AND token(pk) <= ?" - } else { - "" - }; - - // ── Query 1: finish the resume partition (only if there are clustering - // columns to advance past — i.e. the table/index has a sort key, or an - // index carries base-table key clustering columns). ────────────────── + // Query 1 finishes the resume partition only if there are clustering + // columns to advance past: the table or index has a sort key, or an + // index carries base-table key clustering columns. let q1_clustering = self.build_resume_clustering( key_info, sk_info_opt, @@ -366,44 +396,15 @@ impl crate::CassandraEngine { start_key, )?; - if let Some((clustering_predicate, clustering_binds)) = q1_clustering { - let query1 = format!( - "{select_prefix} WHERE pk = ? AND {clustering_predicate} LIMIT {fetch_limit}" - ); - let mut values: Vec = Vec::with_capacity(1 + clustering_binds.len()); - values.push(Value::from(pk_text.as_str())); - values.extend(clustering_binds); - let rows1 = cassandra_util::query_rows::( - session, - &query1, - QueryValues::SimpleValues(values), - "scan_finish_partition", - ) - .await?; - all_rows.extend(rows1); - } - - // ── Query 2: subsequent partitions in token order. ────────────────────── - if all_rows.len() < fetch_limit { - let remaining = fetch_limit - all_rows.len(); - let query2 = format!( - "{select_prefix} WHERE token(pk) > token(?){token_clause} LIMIT {remaining}" - ); - let mut values: Vec = vec![Value::from(pk_text.as_str())]; - if let Some(hi) = token_upper { - values.push(Value::from(hi)); - } - let rows2 = cassandra_util::query_rows::( - session, - &query2, - QueryValues::SimpleValues(values), - "scan_next_partitions", - ) - .await?; - all_rows.extend(rows2); - } - - Ok(all_rows) + read_after_position( + session, + select_prefix, + &pk_text, + q1_clustering.as_ref(), + token_upper, + fetch_limit, + ) + .await } /// Build the clustering-comparison predicate for query 1 of pagination diff --git a/crates/storage-cassandra/src/data/scan_rows.rs b/crates/storage-cassandra/src/data/scan_rows.rs new file mode 100644 index 00000000..911f70e5 --- /dev/null +++ b/crates/storage-cassandra/src/data/scan_rows.rs @@ -0,0 +1,207 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Row handling for the Cassandra Scan: skipping rows that hold no item, +//! and reading on after the last row of a read. +//! +//! A partition whose items have all been deleted answers a read with one row +//! of its static columns. A TTL claim on an absent key, and a transaction +//! prepare that lands on an item deleted after the transaction read it, leave +//! a row with no `item_data`. The scan skips such rows. When every row of a +//! full read holds no item, the scan reads on after the last row read (its +//! partition key and, for a row with one, its sort key), with up to twice as +//! many rows each time and at most [`RUN_READ_ROWS`] (or the read's own +//! LIMIT, if larger), so no statement reads through the whole run. Cassandra +//! counts a tombstone for each deleted row in the partitions such a read +//! passes, so a read Cassandra refuses with a read failure (for example +//! `READ_TOO_MANY_TOMBSTONES`) is tried again with half as many rows, down to +//! one, and later reads in the same call stay at that size. + +use cdrs_tokio::error::Error as CdrsError; +use cdrs_tokio::frame::message_error::ErrorType; +use cdrs_tokio::query::QueryValues; +use cdrs_tokio::types::IntoRustByName; +use cdrs_tokio::types::blob::Blob; +use cdrs_tokio::types::decimal::Decimal; +use cdrs_tokio::types::value::Value; +use extenddb_core::types::{Item, ScalarAttributeType}; +use extenddb_storage::error::StorageError; +use extenddb_storage::util::sk_column; +use std::sync::Arc; + +use super::json_to_item; +use crate::cassandra_util::CassandraSession; + +/// The most rows one read takes while the scan reads on through rows that +/// hold no item, unless the read's own LIMIT is larger. +pub(super) const RUN_READ_ROWS: usize = 1_000; + +pub(super) type Row = cdrs_tokio::types::rows::Row; + +/// A clustering predicate after `pk = ? AND`, and the values it binds. +pub(super) type Clustering = (String, Vec); + +/// Parses the item of every row that holds one, in order. +pub(super) fn rows_to_items(rows: Vec) -> Result, StorageError> { + rows.into_iter() + .map(|row| { + let item_data: Option = row.get_by_name("item_data").map_err(|error| { + StorageError::Internal(format!("scan parse item_data failed: {error}")) + })?; + item_data.map(json_to_item).transpose() + }) + .filter_map(Result::transpose) + .collect() +} + +/// The position of a base-table row, to read on after it: its partition key +/// text, and its sort key as a clustering predicate when the row has one. A +/// static-only row has no sort key, and neither does a hash-only table's row. +pub(super) fn row_position( + row: &Row, + sk_type: Option, +) -> Result<(String, Option), StorageError> { + let parse = + |error: CdrsError| StorageError::Internal(format!("scan parse row key failed: {error}")); + let pk: Option = row.get_by_name("pk").map_err(parse)?; + let pk = pk.ok_or_else(|| StorageError::Internal("scan row without pk".to_owned()))?; + let Some(sk_type) = sk_type else { + return Ok((pk, None)); + }; + let col = sk_column(sk_type); + let value = match sk_type { + ScalarAttributeType::S => IntoRustByName::::get_by_name(row, col) + .map_err(parse)? + .map(Value::from), + ScalarAttributeType::N => IntoRustByName::::get_by_name(row, col) + .map_err(parse)? + .map(Value::from), + ScalarAttributeType::B => IntoRustByName::::get_by_name(row, col) + .map_err(parse)? + .map(Value::from), + }; + Ok((pk, value.map(|v| (format!("{col} > ?"), vec![v])))) +} + +/// Runs one statement and returns its rows, keeping the driver's error. +async fn rows_of( + session: &Arc, + query: &str, + values: Vec, + context: &'static str, +) -> Result, (&'static str, CdrsError)> { + let result = session + .query_with_values(query, QueryValues::SimpleValues(values)) + .await + .map_err(|e| (context, e))?; + let body = result.response_body().map_err(|e| (context, e))?; + Ok(body.into_rows().unwrap_or_default()) +} + +/// Rows after a position, `limit` in all: the rest of partition `pk_text` +/// past `clustering` (query 1, when given), then the partitions after it in +/// token order up to `token_upper` (query 2). +async fn rows_after_position( + session: &Arc, + select_prefix: &str, + pk_text: &str, + clustering: Option<&Clustering>, + token_upper: Option, + limit: usize, +) -> Result, (&'static str, CdrsError)> { + let mut all_rows = Vec::new(); + + if let Some((clustering_predicate, clustering_binds)) = clustering { + let query1 = + format!("{select_prefix} WHERE pk = ? AND {clustering_predicate} LIMIT {limit}"); + let mut values: Vec = Vec::with_capacity(1 + clustering_binds.len()); + values.push(Value::from(pk_text)); + values.extend(clustering_binds.iter().cloned()); + all_rows.extend(rows_of(session, &query1, values, "scan_finish_partition").await?); + } + + if all_rows.len() < limit { + let remaining = limit - all_rows.len(); + let token_clause = if token_upper.is_some() { + " AND token(pk) <= ?" + } else { + "" + }; + let query2 = + format!("{select_prefix} WHERE token(pk) > token(?){token_clause} LIMIT {remaining}"); + let mut values: Vec = vec![Value::from(pk_text)]; + if let Some(hi) = token_upper { + values.push(Value::from(hi)); + } + all_rows.extend(rows_of(session, &query2, values, "scan_next_partitions").await?); + } + + Ok(all_rows) +} + +fn storage_error((context, error): (&'static str, CdrsError)) -> StorageError { + tracing::error!("{context} query failed: {error}"); + StorageError::Internal(format!("{context}: {error}")) +} + +/// [`rows_after_position`], with driver errors as storage errors. +pub(super) async fn read_after_position( + session: &Arc, + select_prefix: &str, + pk_text: &str, + clustering: Option<&Clustering>, + token_upper: Option, + limit: usize, +) -> Result, StorageError> { + rows_after_position( + session, + select_prefix, + pk_text, + clustering, + token_upper, + limit, + ) + .await + .map_err(storage_error) +} + +/// A continuation read of `*limit` rows after a position. A read Cassandra +/// refuses with a read failure is tried again with half the rows, down to +/// one; `*limit` ends as the size that was read and `*ceiling` is lowered to +/// it, so the reads after it do not grow back past it. +pub(super) async fn read_on_after_position( + session: &Arc, + select_prefix: &str, + pk_text: &str, + clustering: Option<&Clustering>, + token_upper: Option, + limit: &mut usize, + ceiling: &mut usize, +) -> Result, StorageError> { + loop { + match rows_after_position( + session, + select_prefix, + pk_text, + clustering, + token_upper, + *limit, + ) + .await + { + Ok(rows) => return Ok(rows), + Err((context, CdrsError::Server { body, .. })) + if *limit > 1 && matches!(body.ty, ErrorType::ReadFailure(_)) => + { + tracing::warn!( + rows = *limit, + "{context}: read failure ({}), reading again with half the rows", + body.message + ); + *limit /= 2; + *ceiling = *limit; + } + Err(error) => return Err(storage_error(error)), + } + } +} From 0b1da901754a4186332d6cc41f19b4b4146d5ba4 Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Mon, 26 Oct 2026 19:47:08 +0000 Subject: [PATCH 2/3] test(cassandra): Scan past rows without an item Thirty live partitions among 300 emptied ones and 20 claim rows, on a table with a sort key and on a hash-only table, paged with no Limit and with Limit 1, 2, 7 and 50, and as two parallel segments; a run of claim rows inside one partition ahead of its item; a run of 5,000 emptied partitions between items, after which Cassandra's tombstones_per_read histogram for the table must show no read of 1,500 tombstones or more; a table with no item left; and a table of 101,000 emptied partitions, more than Cassandra's tombstone failure threshold. Claim rows among items with number, binary and string sort keys, read back and bound again by the continuation; and 2,100 emptied partitions that each held 110 rows, with one item after them in token order, which a read of 1,000 rows cannot pass without a read failure. --- .../tests/direct/scan_rows_without_item.rs | 385 ++++++++++++++++++ .../tests/direct/scan_sort_key_types.rs | 307 ++++++++++++++ .../tests/direct_integration.rs | 4 + 3 files changed, 696 insertions(+) create mode 100644 crates/storage-cassandra/tests/direct/scan_rows_without_item.rs create mode 100644 crates/storage-cassandra/tests/direct/scan_sort_key_types.rs diff --git a/crates/storage-cassandra/tests/direct/scan_rows_without_item.rs b/crates/storage-cassandra/tests/direct/scan_rows_without_item.rs new file mode 100644 index 00000000..c0825b37 --- /dev/null +++ b/crates/storage-cassandra/tests/direct/scan_rows_without_item.rs @@ -0,0 +1,385 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Scan over data rows that hold no item. +//! +//! A partition whose items have all been deleted answers a read with one row +//! of its static columns, and a TTL claim on an absent key writes a row with +//! no `item_data`. 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. A long run of such rows is +//! read in reads of at most about 1,000 rows, never from its start again. + +use extenddb_core::expression::ExpressionMaps; +use extenddb_core::types::{AttributeValue, Item}; +use extenddb_storage::DataEngine; +use extenddb_storage_cassandra::CassandraEngine; +use futures::StreamExt; +use std::collections::{BTreeMap, BTreeSet}; + +use crate::helpers::{TestTable, setup_engine}; + +fn key(table: &TestTable, id: &str) -> Item { + let mut key = BTreeMap::from([("id".to_string(), AttributeValue::S(id.to_string()))]); + if table.key_info.key_schema.len() > 1 { + key.insert("sort".to_string(), AttributeValue::S("a".to_string())); + } + key +} + +async fn put(engine: &CassandraEngine, table: &TestTable, id: &str) { + let mut item = key(table, id); + 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"); +} + +async fn put_and_delete(engine: &CassandraEngine, table: &TestTable, id: &str) { + put(engine, table, id).await; + engine + .delete_item( + &table.key_info, + &key(table, id), + false, + None, + &Default::default(), + None, + ) + .await + .expect("delete_item"); +} + +/// Writes the row a TTL claim on the absent key `id` leaves: no `item_data`. +async fn claim_row(engine: &CassandraEngine, table: &TestTable, id: &str) { + let keyspace = engine.account_keyspace(&table.key_info.account_id); + let data_table = format!("items_{}", table.key_info.table_id.replace('-', "_")); + let session = engine.session_arc(); + if table.key_info.key_schema.len() > 1 { + session + .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!(id, "a"), + ) + .await + .expect("claim row"); + } else { + session + .query_with_values( + format!( + "UPDATE {keyspace}.{data_table} SET prepared_txn_id = uuid(), \ + prepared_txn_timestamp = 1 WHERE pk = ?" + ), + cdrs_tokio::query_values!(id), + ) + .await + .expect("claim row"); + } +} + +fn ids(items: &[Item]) -> Vec { + items + .iter() + .map(|item| match item.get("id") { + Some(AttributeValue::S(s)) => s.clone(), + other => panic!("partition key {other:?}"), + }) + .collect() +} + +/// Pages through the table (or one segment of it) with `limit` and returns +/// every partition key read, in the order read. +async fn read_all( + engine: &CassandraEngine, + table: &TestTable, + limit: Option, + segment: Option<(i64, i64)>, +) -> Vec { + let mut all = Vec::new(); + let mut start: Option = None; + for _ in 0..1000 { + let (items, last_key) = engine + .scan( + &table.key_info, + limit, + start.as_ref(), + segment.map(|(s, _)| s), + segment.map(|(_, t)| t), + None, + ) + .await + .expect("scan"); + all.extend(ids(&items)); + match last_key { + Some(k) => start = Some(k), + None => return all, + } + } + panic!("more than 1000 pages"); +} + +/// 30 live partitions among 300 emptied ones and 20 claim rows, so runs of +/// rows without an item sit between the items in token order. +async fn fill(engine: &CassandraEngine, table: &TestTable) -> BTreeSet { + let live: Vec = (0..30).map(|i| format!("live{i:03}")).collect(); + let emptied: Vec = (0..300).map(|i| format!("gone{i:03}")).collect(); + let claims: Vec = (0..20).map(|i| format!("claim{i:03}")).collect(); + futures::stream::iter(&live) + .for_each_concurrent(32, |id| put(engine, table, id)) + .await; + futures::stream::iter(&emptied) + .for_each_concurrent(32, |id| put_and_delete(engine, table, id)) + .await; + futures::stream::iter(&claims) + .for_each_concurrent(32, |id| claim_row(engine, table, id)) + .await; + live.into_iter().collect() +} + +async fn assert_scans_read_every_item(engine: &CassandraEngine, table: &TestTable) { + let live = fill(engine, table).await; + for limit in [None, Some(1), Some(2), Some(7), Some(50)] { + let read = read_all(engine, table, limit, None).await; + assert_eq!(read.len(), live.len(), "limit {limit:?}: {read:?}"); + assert_eq!( + read.into_iter().collect::>(), + live, + "limit {limit:?}" + ); + } + for limit in [None, Some(1), Some(3)] { + let mut read = read_all(engine, table, limit, Some((0, 2))).await; + read.extend(read_all(engine, table, limit, Some((1, 2))).await); + assert_eq!( + read.len(), + live.len(), + "segments, limit {limit:?}: {read:?}" + ); + assert_eq!( + read.into_iter().collect::>(), + live, + "segments, limit {limit:?}" + ); + } +} + +#[tokio::test] +async fn test_scan_past_partitions_without_an_item() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "ScanRowsWithoutItem", true).await; + assert_scans_read_every_item(&engine, &table).await; +} + +#[tokio::test] +async fn test_scan_of_a_hash_only_table_past_claim_rows() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "ScanHashOnlyClaims", false).await; + assert_scans_read_every_item(&engine, &table).await; +} + +#[tokio::test] +async fn test_scan_of_a_table_with_no_item_left() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "ScanNoItemLeft", true).await; + let emptied: Vec = (0..50).map(|i| format!("gone{i:03}")).collect(); + futures::stream::iter(&emptied) + .for_each_concurrent(32, |id| put_and_delete(&engine, &table, id)) + .await; + for limit in [None, Some(1), Some(7)] { + let (items, last_key) = engine + .scan(&table.key_info, limit, None, None, None, None) + .await + .expect("scan"); + assert!(items.is_empty(), "limit {limit:?}: {:?}", ids(&items)); + assert_eq!(last_key, None, "limit {limit:?}"); + } +} + +/// The CQL table behind `table`. +fn data_table(engine: &CassandraEngine, table: &TestTable) -> String { + format!( + "{}.items_{}", + engine.account_keyspace(&table.key_info.account_id), + table.key_info.table_id.replace('-', "_") + ) +} + +/// Writes `n` emptied partitions with CQL, as DeleteItem leaves them on a +/// table with a sort key: the static delete timestamp and a deleted row. +async fn emptied_partitions(engine: &CassandraEngine, table: &TestTable, n: usize) { + let cql = format!( + "BEGIN UNLOGGED BATCH \ + UPDATE {t} SET partition_max_delete_timestamp = 1 WHERE pk = ?; \ + DELETE FROM {t} WHERE pk = ? AND sk_s = 'a'; \ + APPLY BATCH", + t = data_table(engine, table) + ); + let session = engine.session_arc(); + let ids: Vec = (0..n).map(|i| format!("gone{i:06}")).collect(); + futures::stream::iter(&ids) + .for_each_concurrent(8, |id| { + let (session, cql) = (session.clone(), cql.clone()); + async move { + session + .query_with_values(cql, cdrs_tokio::query_values!(id.as_str(), id.as_str())) + .await + .expect("emptied partition"); + } + }) + .await; +} + +/// A run of claim rows inside one partition, ahead of an item in the same +/// partition: the scan reads on within the partition (query 1), not past it. +#[tokio::test] +async fn test_scan_past_claim_rows_inside_a_partition() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "ScanClaimRun", true).await; + let t = data_table(&engine, &table); + let session = engine.session_arc(); + for i in 0..40 { + session + .query_with_values( + format!( + "UPDATE {t} SET prepared_txn_id = uuid(), prepared_txn_timestamp = 1 \ + WHERE pk = 'p' AND sk_s = ?" + ), + cdrs_tokio::query_values!(format!("c{i:03}")), + ) + .await + .expect("claim row"); + } + let mut item = BTreeMap::from([ + ("id".to_string(), AttributeValue::S("p".to_string())), + ("sort".to_string(), AttributeValue::S("z".to_string())), + ]); + 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"); + for limit in [None, Some(1), Some(3)] { + assert_eq!( + read_all(&engine, &table, limit, None).await, + ["p"], + "limit {limit:?}" + ); + } +} + +/// The most tombstones Cassandra has counted in one read of `table`'s data +/// table, from its `system_views.tombstones_per_read` histogram. +async fn max_tombstones_per_read(engine: &CassandraEngine, table: &TestTable) -> f64 { + use cdrs_tokio::types::IntoRustByName; + + let keyspace = engine.account_keyspace(&table.key_info.account_id); + let data_table = format!("items_{}", table.key_info.table_id.replace('-', "_")); + let result = engine + .session_arc() + .query_with_values( + "SELECT max FROM system_views.tombstones_per_read \ + WHERE keyspace_name = ? AND table_name = ?", + cdrs_tokio::query_values!(keyspace.as_str(), data_table.as_str()), + ) + .await + .expect("tombstones_per_read"); + let rows = result + .response_body() + .expect("tombstones_per_read body") + .into_rows() + .unwrap_or_default(); + let max: Option = rows + .first() + .map(|row| row.get_by_name("max").expect("max column")) + .unwrap_or_default(); + max.unwrap_or(0.0) +} + +/// A run of emptied partitions longer than one bounded read, between items. +/// A paged Scan reads it in statements of at most about 1,000 rows each, so +/// no single read of the data table counts more tombstones than that; a scan +/// that read the run again from its start with a growing LIMIT, or let the +/// read size grow without a cap, would read the whole run in one statement. +#[tokio::test] +async fn test_scan_past_a_long_run_of_emptied_partitions() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "ScanEmptiedRun", true).await; + emptied_partitions(&engine, &table, 5_000).await; + let live: BTreeSet = (0..5).map(|i| format!("live{i}")).collect(); + for id in &live { + put(&engine, &table, id).await; + } + for limit in [Some(1), Some(2), Some(1000)] { + let read = read_all(&engine, &table, limit, None).await; + assert_eq!(read.len(), live.len(), "limit {limit:?}: {read:?}"); + assert_eq!( + read.into_iter().collect::>(), + live, + "limit {limit:?}" + ); + } + let max = max_tombstones_per_read(&engine, &table).await; + assert!( + max > 0.0, + "no tombstones counted, so the check proves nothing" + ); + // The histogram rounds up: a read of 1,001 rows shows as about 1,109. + assert!(max < 1_500.0, "one read counted {max} tombstones"); + // No Limit reads up to 1,000,001 rows in one statement, by design. + let read = read_all(&engine, &table, None, None).await; + assert_eq!(read.into_iter().collect::>(), live, "no limit"); +} + +/// A table with more emptied partitions than Cassandra's +/// `tombstone_failure_threshold` (100,000 by default). Each read of the run +/// stays below it, so the scan answers no item and no key. +#[tokio::test] +async fn test_scan_of_a_table_emptied_past_the_tombstone_threshold() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "ScanEmptiedPastThreshold", true).await; + emptied_partitions(&engine, &table, 101_000).await; + for limit in [Some(1000), Some(128)] { + let (items, last_key) = engine + .scan(&table.key_info, limit, None, None, None, None) + .await + .expect("scan of an emptied table"); + assert!(items.is_empty(), "limit {limit:?}"); + assert_eq!(last_key, None, "limit {limit:?}"); + } +} diff --git a/crates/storage-cassandra/tests/direct/scan_sort_key_types.rs b/crates/storage-cassandra/tests/direct/scan_sort_key_types.rs new file mode 100644 index 00000000..6a2b68f4 --- /dev/null +++ b/crates/storage-cassandra/tests/direct/scan_sort_key_types.rs @@ -0,0 +1,307 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Scan past rows that hold no item, by sort key type, and past emptied +//! partitions that each hold many deleted rows. +//! +//! A full read of rows that hold no item reads on after its last row, with +//! the row's sort key read back from Cassandra and bound again; the typed +//! tests check that this works for number, binary and string sort keys. A +//! read through partitions with many deleted rows can reach Cassandra's +//! `tombstone_failure_threshold` (100,000 by default) at 1,000 rows; the scan +//! then reads again with fewer rows. + +use bigdecimal::BigDecimal; +use cdrs_tokio::types::IntoRustByName; +use cdrs_tokio::types::blob::Blob; +use cdrs_tokio::types::decimal::Decimal; +use cdrs_tokio::types::value::Value; +use extenddb_core::expression::ExpressionMaps; +use extenddb_core::types::{AttributeValue, Item, ScalarAttributeType}; +use extenddb_storage::DataEngine; +use extenddb_storage_cassandra::CassandraEngine; +use futures::StreamExt; +use std::collections::BTreeMap; +use std::str::FromStr; + +use crate::helpers::{TestTable, setup_engine}; + +#[derive(Clone, Debug)] +enum Sk { + S(&'static str), + N(&'static str), + B(Vec), +} + +impl Sk { + fn attr(&self) -> AttributeValue { + match self { + Sk::S(s) => AttributeValue::S((*s).to_string()), + Sk::N(n) => AttributeValue::N((*n).to_string()), + Sk::B(b) => AttributeValue::B(b.clone()), + } + } + + fn bind(&self) -> Value { + match self { + Sk::S(s) => Value::from(*s), + Sk::N(n) => { + let (unscaled, scale) = BigDecimal::from_str(n) + .expect("decimal") + .as_bigint_and_exponent(); + Value::from(Decimal::new(unscaled, i32::try_from(scale).expect("scale"))) + } + Sk::B(b) => Value::from(Blob::new(b.clone())), + } + } + + fn column(&self) -> &'static str { + match self { + Sk::S(_) => "sk_s", + Sk::N(_) => "sk_n", + Sk::B(_) => "sk_b", + } + } +} + +fn data_table(engine: &CassandraEngine, table: &TestTable) -> String { + format!( + "{}.items_{}", + engine.account_keyspace(&table.key_info.account_id), + table.key_info.table_id.replace('-', "_") + ) +} + +/// The row a TTL claim on the absent key (`pk`, `sk`) leaves: no `item_data`. +async fn claim(engine: &CassandraEngine, table: &TestTable, pk: &str, sk: &Sk) { + let t = data_table(engine, table); + engine + .session_arc() + .query_with_values( + format!( + "UPDATE {t} SET prepared_txn_id = uuid(), prepared_txn_timestamp = 1 \ + WHERE pk = ? AND {} = ?", + sk.column() + ), + cdrs_tokio::query_values!(pk, sk.bind()), + ) + .await + .expect("claim row"); +} + +/// A partition as DeleteItem leaves it once its only item is gone. +async fn emptied(engine: &CassandraEngine, table: &TestTable, pk: &str, sk: &Sk) { + let t = data_table(engine, table); + engine + .session_arc() + .query_with_values( + format!( + "BEGIN UNLOGGED BATCH \ + UPDATE {t} SET partition_max_delete_timestamp = 1 WHERE pk = ?; \ + DELETE FROM {t} WHERE pk = ? AND {} = ?; \ + APPLY BATCH", + sk.column() + ), + cdrs_tokio::query_values!(pk, pk, sk.bind()), + ) + .await + .expect("emptied partition"); +} + +async fn put_at(engine: &CassandraEngine, table: &TestTable, pk: &str, sk: &Sk) { + let item: Item = BTreeMap::from([ + ("id".to_string(), AttributeValue::S(pk.to_string())), + ("sort".to_string(), sk.attr()), + ("v".to_string(), AttributeValue::S("x".to_string())), + ]); + engine + .put_item( + &table.key_info, + item, + false, + None, + &ExpressionMaps::default(), + None, + ) + .await + .expect("put_item"); +} + +/// Pages a Scan to the end and returns the items in the order read. +async fn read_all(engine: &CassandraEngine, table: &TestTable, limit: Option) -> Vec { + let (mut all, mut start): (Vec, Option) = (Vec::new(), None); + for _ in 0..3000 { + let (items, last_key) = engine + .scan(&table.key_info, limit, start.as_ref(), None, None, None) + .await + .expect("scan"); + all.extend(items); + match last_key { + Some(k) => start = Some(k), + None => return all, + } + } + panic!("more than 3000 pages"); +} + +/// Numbers are stored in a normal form, so they are compared by value. +fn canon(v: &AttributeValue) -> String { + match v { + AttributeValue::N(n) => BigDecimal::from_str(n) + .expect("number") + .normalized() + .to_string(), + other => format!("{other:?}"), + } +} + +/// Partition `p` holding `rows` in sort key order (a claim row or an item +/// each), and 40 emptied partitions. Every page size reads exactly the items +/// of `p`, once each and in order. +async fn check_typed(ty: ScalarAttributeType, name: &str, rows: Vec<(Sk, bool)>, gone: Sk) { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::with_sort_key_type(&engine, name, ty).await; + for (sk, is_item) in &rows { + if *is_item { + put_at(&engine, &table, "p", sk).await; + } else { + claim(&engine, &table, "p", sk).await; + } + } + for i in 0..40 { + emptied(&engine, &table, &format!("gone{i:02}"), &gone).await; + } + let want: Vec = rows + .iter() + .filter(|(_, is_item)| *is_item) + .map(|(sk, _)| canon(&sk.attr())) + .collect(); + for limit in [Some(1), Some(2), Some(3), Some(7), None] { + let items = read_all(&engine, &table, limit).await; + let got: Vec = items + .iter() + .map(|item| canon(item.get("sort").expect("sort"))) + .collect(); + assert_eq!(got, want, "limit {limit:?}"); + } +} + +#[tokio::test] +async fn test_scan_past_claim_rows_with_a_number_sort_key() { + let rows = vec![ + (Sk::N("-123456789012345678901234567890.123456789"), false), + (Sk::N("-99999.5"), false), + (Sk::N("-99999"), true), + (Sk::N("-0.000000000000000000000000000001"), false), + (Sk::N("0"), true), + (Sk::N("0.1"), false), + (Sk::N("0.10000000000000000000000000000000000001"), false), + (Sk::N("0.2"), true), + (Sk::N("1.000000000000000000000000000000000001"), false), + (Sk::N("1E+5"), false), + (Sk::N("1E+10"), true), + (Sk::N("9.9999999999999999999999999999999999998E+125"), false), + (Sk::N("9.9999999999999999999999999999999999999E+125"), false), + ]; + check_typed(ScalarAttributeType::N, "ScanClaimNumber", rows, Sk::N("5")).await; +} + +#[tokio::test] +async fn test_scan_past_claim_rows_with_a_binary_sort_key() { + let b = |bytes: &[u8]| Sk::B(bytes.to_vec()); + let rows = vec![ + (b(&[0x00]), false), + (b(&[0x00, 0x00]), false), + (b(&[0x00, 0x01]), true), + (b(&[0x7f, 0xff]), false), + (b(&[0x80]), true), + (b(&[0x80, 0x00]), false), + (b(&[0xab; 1024]), false), + (b(&[0xfe]), true), + (b(&[0xff, 0x00]), false), + (b(&[0xff, 0xff, 0x01]), true), + (b(&[0xff, 0xff, 0xff]), false), + ]; + check_typed(ScalarAttributeType::B, "ScanClaimBinary", rows, b(&[0x01])).await; +} + +#[tokio::test] +async fn test_scan_past_claim_rows_with_a_string_sort_key() { + let rows = vec![ + (Sk::S("A"), false), + (Sk::S("a"), true), + (Sk::S("\u{e9}"), false), + (Sk::S("\u{65e5}\u{672c}"), false), + (Sk::S("\u{1f600}"), true), + (Sk::S("\u{1f601}"), false), + ]; + check_typed(ScalarAttributeType::S, "ScanClaimString", rows, Sk::S("a")).await; +} + +async fn token_of(engine: &CassandraEngine, pk: &str) -> i64 { + let rows = engine + .session_arc() + .query_with_values( + "SELECT token(?) AS t FROM system.local", + cdrs_tokio::query_values!(pk), + ) + .await + .expect("token") + .response_body() + .expect("body") + .into_rows() + .expect("rows"); + IntoRustByName::::get_by_name(&rows[0], "t") + .expect("decode") + .expect("not null") +} + +/// 2,100 emptied partitions that each held 110 rows, and one item in the +/// partition with the highest token. A read of 1,000 such rows counts about +/// 110,000 tombstones, past the threshold; a read of 500 counts about 55,000. +/// A Scan with a small Limit reads through all of them to the item. +#[tokio::test] +async fn test_scan_past_emptied_partitions_with_many_deleted_rows() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "ScanHeavyEmptied", true).await; + let t = data_table(&engine, &table); + let session = engine.session_arc(); + let mut tokens: Vec<(i64, String)> = Vec::new(); + for i in 0..2101 { + let pk = format!("heavy{i:04}"); + tokens.push((token_of(&engine, &pk).await, pk)); + } + tokens.sort(); + let (_, live) = tokens.pop().expect("partition keys"); + futures::stream::iter(tokens.iter().map(|(_, pk)| pk)) + .for_each_concurrent(8, |pk| { + let (session, t) = (session.clone(), t.clone()); + async move { + // Test-generated ASCII keys, so literals are safe here. + let mut cql = format!( + "BEGIN UNLOGGED BATCH UPDATE {t} SET partition_max_delete_timestamp = 1 \ + WHERE pk = '{pk}'; " + ); + for j in 0..110 { + cql.push_str(&format!( + "DELETE FROM {t} WHERE pk = '{pk}' AND sk_s = 's{j:03}'; " + )); + } + cql.push_str("APPLY BATCH"); + session.query(cql).await.expect("emptied partition"); + } + }) + .await; + put_at(&engine, &table, &live, &Sk::S("a")).await; + for limit in [Some(1), Some(128)] { + let items = read_all(&engine, &table, limit).await; + let ids: Vec<&AttributeValue> = items.iter().filter_map(|i| i.get("id")).collect(); + assert_eq!(ids, [&AttributeValue::S(live.clone())], "limit {limit:?}"); + } +} diff --git a/crates/storage-cassandra/tests/direct_integration.rs b/crates/storage-cassandra/tests/direct_integration.rs index 43eb28b8..e351b6ba 100644 --- a/crates/storage-cassandra/tests/direct_integration.rs +++ b/crates/storage-cassandra/tests/direct_integration.rs @@ -43,6 +43,10 @@ mod query; mod roles; #[path = "direct/scan.rs"] mod scan; +#[path = "direct/scan_rows_without_item.rs"] +mod scan_rows_without_item; +#[path = "direct/scan_sort_key_types.rs"] +mod scan_sort_key_types; #[path = "direct/settings_store.rs"] mod settings_store; #[path = "direct/streams.rs"] From 1b1e8140afac309d32629df200cad67a0313fb1e Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Mon, 26 Oct 2026 19:47:08 +0000 Subject: [PATCH 3/3] test: Scan of a table with partitions whose items were deleted Dual-target wire tests from the service measurements on 2026-10-26. At 15:29 UTC: every live item is returned and only live items are counted in ScannedCount, with no Limit, with Limit 1, 7 and 50, across two parallel segments, with Select COUNT and with a FilterExpression, on a table with a sort key and on a hash-only table. At 16:39 UTC: a table whose items have all been deleted answers Count 0, ScannedCount 0 and no LastEvaluatedKey. --- tests/test_scan_emptied_partitions.py | 126 ++++++++++++++++++++++++++ 1 file changed, 126 insertions(+) create mode 100644 tests/test_scan_emptied_partitions.py diff --git a/tests/test_scan_emptied_partitions.py b/tests/test_scan_emptied_partitions.py new file mode 100644 index 00000000..42314f8d --- /dev/null +++ b/tests/test_scan_emptied_partitions.py @@ -0,0 +1,126 @@ +# Copyright 2026 ExtendDB contributors +# SPDX-License-Identifier: Apache-2.0 + +"""Scan of a table whose items sit among partitions whose items were deleted. + +Amazon DynamoDB answers such a Scan with every live item and counts only +live items in ScannedCount, with no Limit, with a Limit, across parallel +segments, with Select COUNT and with a FilterExpression. With Limit 1, a +Scan of 59 live items among 100 partitions whose items were deleted took 60 +pages: 59 with one item and a last one with none. Measured on 2026-10-26 at +15:29 UTC (us-east-1), each case twice. A table whose 50 items have all been +deleted, with a sort key or hash only, answers Count 0, ScannedCount 0 and +no LastEvaluatedKey with no Limit and with Limit 1 and 7. Measured at 16:39 +UTC, each case twice. + +Dual-target against Amazon DynamoDB and extenddb. +""" + +from __future__ import annotations + +import pytest + +from conftest import scoped_table + +LIVE = [f"live-{i:02d}" for i in range(30)] +GONE = [f"gone-{i:03d}" for i in range(100)] + + +def _key(pk, sort_key): + key = {"pk": {"S": pk}} + if sort_key: + key["sk"] = {"S": "a"} + return key + + +def _fill(client, name, sort_key): + for pk in LIVE: + client.put_item(TableName=name, Item={**_key(pk, sort_key), "x": {"S": "x"}}) + for pk in GONE: + client.put_item(TableName=name, Item={**_key(pk, sort_key), "x": {"S": "x"}}) + client.delete_item(TableName=name, Key=_key(pk, sort_key)) + + +SORT_KEY_TABLE = { + "attribute_definitions": [ + {"AttributeName": "pk", "AttributeType": "S"}, + {"AttributeName": "sk", "AttributeType": "S"}, + ], + "key_schema": [ + {"AttributeName": "pk", "KeyType": "HASH"}, + {"AttributeName": "sk", "KeyType": "RANGE"}, + ], +} + + +@pytest.fixture(scope="module", params=["sort_key", "hash_only"]) +def table(request, dynamodb_client): + sort_key = request.param == "sort_key" + with scoped_table(dynamodb_client, **(SORT_KEY_TABLE if sort_key else {})) as name: + _fill(dynamodb_client, name, sort_key) + yield name + + +def _page_all(client, name, **kw): + pks, pages, start = [], [], None + for _ in range(500): + args = {"TableName": name, "ConsistentRead": True, **kw} + if start: + args["ExclusiveStartKey"] = start + resp = client.scan(**args) + pages.append(resp) + pks += [item["pk"]["S"] for item in resp.get("Items", [])] + start = resp.get("LastEvaluatedKey") + if not start: + return pks, pages + raise AssertionError("more than 500 pages") + + +def test_scan_with_no_limit(dynamodb_client, table): + pks, pages = _page_all(dynamodb_client, table) + assert len(pages) == 1 + assert (pages[0]["Count"], pages[0]["ScannedCount"]) == (30, 30) + assert sorted(pks) == LIVE + + +@pytest.mark.parametrize("limit", [1, 7, 50]) +def test_scan_pages_through_every_item(dynamodb_client, table, limit): + pks, pages = _page_all(dynamodb_client, table, Limit=limit) + assert sorted(pks) == LIVE + first = pages[0] + expected = min(limit, 30) + assert (first["Count"], first["ScannedCount"]) == (expected, expected) + + +def test_parallel_scan_reads_every_item(dynamodb_client, table): + pks = [] + for segment in (0, 1): + seg_pks, _ = _page_all(dynamodb_client, table, Limit=3, Segment=segment, TotalSegments=2) + pks += seg_pks + assert sorted(pks) == LIVE + + +def test_scan_count(dynamodb_client, table): + _, pages = _page_all(dynamodb_client, table, Select="COUNT") + assert sum(p["Count"] for p in pages) == 30 + + +def test_scan_with_a_filter(dynamodb_client, table): + flt = {"FilterExpression": "begins_with(pk, :p)", "ExpressionAttributeValues": {":p": {"S": "live-0"}}} + pks, pages = _page_all(dynamodb_client, table, **flt) + assert (pages[0]["Count"], pages[0]["ScannedCount"]) == (10, 30) + assert sorted(pks) == LIVE[:10] + pks, _ = _page_all(dynamodb_client, table, Limit=7, **flt) + assert sorted(pks) == LIVE[:10] + + +@pytest.mark.parametrize("sort_key", [True, False], ids=["sort_key", "hash_only"]) +def test_scan_of_a_table_with_no_item_left(dynamodb_client, sort_key): + with scoped_table(dynamodb_client, **(SORT_KEY_TABLE if sort_key else {})) as name: + for pk in GONE[:50]: + dynamodb_client.put_item(TableName=name, Item={**_key(pk, sort_key), "x": {"S": "x"}}) + dynamodb_client.delete_item(TableName=name, Key=_key(pk, sort_key)) + for kw in ({}, {"Limit": 1}, {"Limit": 7}): + resp = dynamodb_client.scan(TableName=name, ConsistentRead=True, **kw) + assert (resp["Count"], resp["ScannedCount"]) == (0, 0), kw + assert "LastEvaluatedKey" not in resp, kw