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)), + } + } +} 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"] 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