From 2ca33f98149c06a5effa6167280dfde6937cde2b Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Sun, 25 Oct 2026 20:44:49 +0000 Subject: [PATCH 1/9] fix(cassandra): record stream records for transacted writes A TransactWriteItems Put, Update or Delete on a Cassandra table with an enabled stream recorded nothing. The commit phase wrote the item mutation and such index rows as it maintained, and skipped the stream capture the request carries per operation. The commit phase now builds the stream record for each item mutation with the same stream_util::stream_record_statement the single-item paths use, with the same image pair, so the event name, the images for the table's view type, the shard assignment and the sequence number come from the same source as the equivalent PutItem, UpdateItem or DeleteItem. The record joins the secondary-effects LOGGED BATCH that already carries the index rows; Cassandra rejects a batch mixing an LWT with writes to another partition, so neither the single-item path nor this one can put the record in the base mutation's own statement. A pure builder returns that batch and the commit path executes what it returns, so a unit test holds the index rows and the record together. The old image is read at commit time, from the row together with its created_to_prepare and last_committed_txn_timestamp columns: a row PREPARE planted, which has never had a commit land on it, has no old image; otherwise the old image is the row's item_data, which is itself absent when the row holds a marker with no item. PREPARE plants the request's own item for a Put and the bare key for an Update, so the row's item_data alone would report an insert as a modify, and PREPARE's own read is one write stale when a single-item write commits inside the PREPARE window, where it is accepted and the marker's IF prepared_txn_id = null does not detect it. This costs one read per streamed Put and Delete op; the Update arm reads the row anyway. Committing, and the recovery worker's replay, now clear created_to_prepare. Releases before this one did not clear it, so a row they committed carries the flag over a committed item; the commit timestamp, which only a commit writes, tells that row apart from a planted one, and the first transacted write on it after an upgrade records MODIFY or REMOVE rather than INSERT or nothing. The index metadata read moves above the base LWT on the Put and Update arms, as occ_write does it, so the record does not wait an uncached catalog round trip behind an item that has already committed. A ConditionCheck records nothing, a cancelled transaction records nothing for any of its items, and a delete of an absent key records nothing. Records land in request order, which is ExtendDB's own behaviour on every backend rather than a promise the service makes. Storage tests cover fourteen shapes on a live Cassandra node, including a Put over an existing item, a Put over a planted row, a Put and a Delete over an item committed before the flag was cleared, and a transacted record compared field by field with the single-item record of the same verb. A wire test covers a transacted Put over an existing item, an Update, a Delete, a ConditionCheck, a cancelled transaction and a KEYS_ONLY stream; it asserts per key, since records for one key share a shard and records for different keys do not. Closes #516 --- crates/storage-cassandra/src/data/mod.rs | 1 + .../src/data/transact_stream.rs | 130 ++++ .../src/data/transactions.rs | 333 ++++++++-- .../direct/transact_write_stream_images.rs | 408 +++++++++++++ .../direct/transact_write_stream_records.rs | 576 ++++++++++++++++++ .../tests/direct_integration.rs | 4 + tests/python/test_streams_transactions.py | 295 +++++++++ 7 files changed, 1683 insertions(+), 64 deletions(-) create mode 100644 crates/storage-cassandra/src/data/transact_stream.rs create mode 100644 crates/storage-cassandra/tests/direct/transact_write_stream_images.rs create mode 100644 crates/storage-cassandra/tests/direct/transact_write_stream_records.rs create mode 100644 tests/python/test_streams_transactions.py diff --git a/crates/storage-cassandra/src/data/mod.rs b/crates/storage-cassandra/src/data/mod.rs index 946dbd042..49bb1713f 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 transact_stream; pub mod transaction_ledger; mod transactions; pub(crate) mod ttl; diff --git a/crates/storage-cassandra/src/data/transact_stream.rs b/crates/storage-cassandra/src/data/transact_stream.rs new file mode 100644 index 000000000..f4e101a3e --- /dev/null +++ b/crates/storage-cassandra/src/data/transact_stream.rs @@ -0,0 +1,130 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! The two pure pieces of a transacted mutation's stream record: which old +//! image it carries, and how its statement is batched with the index rows. +//! `transactions.rs` calls both from its commit phase. + +use cdrs_tokio::query::BatchQueryBuilder; +use extenddb_core::types::Item; + +/// The old image a transacted mutation's stream record carries, from the row as +/// the commit phase read it. +/// +/// `created_to_prepare` marks a row that exists only because a PREPARE planted +/// it. Such a row's `item_data` is the request's own item for a Put and the +/// bare key for an Update, so there is no old image: the mutation is an insert. +/// Otherwise the old image is the row's `item_data`, which is itself absent +/// when the row holds a marker with no item, as a concurrent single-item delete +/// inside the PREPARE window leaves it. +pub(super) fn commit_old_image(row: Option, created_to_prepare: bool) -> Option { + if created_to_prepare { None } else { row } +} + +/// The batches one committed transactional mutation's secondary effects run as. +/// +/// There is one, holding the index rows and the stream record together: a +/// reader must not see an index row for a mutation whose record was not +/// written, and a crash must not leave the index rows standing without it. The +/// caller executes exactly what this returns, so the one-batch property is +/// asserted in a unit test rather than resting on the shape of the calling +/// code, which no integration test can see. +pub(super) fn secondary_effect_batches( + index_batch: BatchQueryBuilder, + stream_stmt: Option, +) -> Vec { + let batch = match stream_stmt { + Some(stmt) => { + index_batch.add_query(stmt, cdrs_tokio::query::QueryValues::SimpleValues(vec![])) + } + None => index_batch, + }; + vec![batch] +} + +#[cfg(test)] +mod tests { + use super::*; + use extenddb_core::types::AttributeValue; + + fn built_statements(batch: BatchQueryBuilder) -> Vec { + use cdrs_tokio::frame::message_batch::BatchQuerySubj; + + batch + .build() + .unwrap() + .request + .queries + .into_iter() + .map(|query| match query.subject { + BatchQuerySubj::QueryString(cql) => cql.to_string(), + BatchQuerySubj::PreparedId(_) => { + panic!("secondary effects are built as CQL strings") + } + }) + .collect() + } + + fn index_batch() -> BatchQueryBuilder { + BatchQueryBuilder::new() + .add_query( + "INSERT INTO ks.index_rows (a) VALUES (1)".to_owned(), + cdrs_tokio::query::QueryValues::SimpleValues(vec![]), + ) + .add_query( + "DELETE FROM ks.index_rows WHERE a = 2".to_owned(), + cdrs_tokio::query::QueryValues::SimpleValues(vec![]), + ) + } + + const RECORD: &str = "INSERT INTO ks.stream_records (shard_id) VALUES ('s')"; + + /// The index rows and the record commit together or not at all, which is + /// what makes a transacted record as atomic with its item as the + /// single-item path's record is. + #[test] + fn index_rows_and_the_stream_record_share_one_batch() { + let batches = secondary_effect_batches(index_batch(), Some(RECORD.to_owned())); + assert_eq!(batches.len(), 1, "the record must not run as its own batch"); + let statements = built_statements(batches.into_iter().next().unwrap()); + assert_eq!(statements.len(), 3, "got {statements:?}"); + assert!(statements[0].contains("INSERT INTO ks.index_rows")); + assert!(statements[1].contains("DELETE FROM ks.index_rows")); + assert_eq!(statements[2], RECORD); + } + + #[test] + fn a_table_with_no_index_runs_the_record_in_one_batch() { + let batches = secondary_effect_batches(BatchQueryBuilder::new(), Some(RECORD.to_owned())); + assert_eq!(batches.len(), 1); + assert_eq!( + built_statements(batches.into_iter().next().unwrap()), + vec![RECORD.to_owned()] + ); + } + + #[test] + fn index_rows_without_a_record_run_unchanged() { + let batches = secondary_effect_batches(index_batch(), None); + assert_eq!(batches.len(), 1); + assert_eq!( + built_statements(batches.into_iter().next().unwrap()).len(), + 2 + ); + } + + #[test] + fn a_prepare_created_row_has_no_old_image() { + let mut item = Item::new(); + item.insert("id".to_owned(), AttributeValue::S("a".to_owned())); + + assert_eq!( + commit_old_image(Some(item.clone()), true), + None, + "a row PREPARE planted holds the request's own item, not an old image" + ); + assert_eq!(commit_old_image(Some(item.clone()), false), Some(item)); + assert_eq!(commit_old_image(None, false), None); + assert_eq!(commit_old_image(None, true), None); + } +} diff --git a/crates/storage-cassandra/src/data/transactions.rs b/crates/storage-cassandra/src/data/transactions.rs index 639aad7e4..3e82a7b66 100644 --- a/crates/storage-cassandra/src/data/transactions.rs +++ b/crates/storage-cassandra/src/data/transactions.rs @@ -3,6 +3,7 @@ //! DynamoDB transaction implementations for Cassandra backend. +use super::transact_stream::{commit_old_image, secondary_effect_batches}; use cdrs_tokio::consistency::Consistency; use cdrs_tokio::frame::Envelope; use cdrs_tokio::query::BatchQueryBuilder; @@ -13,6 +14,7 @@ use extenddb_core::types::{ AttributeValue, CancellationReason, Item, ReturnValuesOnConditionCheckFailure, TableKeyInfo, }; use extenddb_core::validation; +use extenddb_storage::StreamCapture; use extenddb_storage::TransactGetOp; use extenddb_storage::TransactWriteOp; use extenddb_storage::error::StorageError; @@ -851,6 +853,25 @@ impl CassandraEngine { } /// Execute COMMIT phase: apply changes and update ledger. + /// + /// A stream record needs the image the item held before this transaction + /// touched it, to tell an insert from a modify and to carry OldImage. Each + /// arm reads it here, from the row together with its `created_to_prepare` + /// column, and applies the rule in `commit_old_image`: a row PREPARE + /// planted has no old image, otherwise the old image is the row's + /// `item_data`, which may itself be absent. + /// + /// The row's `item_data` alone would not do, because PREPARE plants a + /// placeholder row for a key that was absent, carrying the request's own + /// item for a Put and the bare key for an Update; `created_to_prepare` is + /// what tells that row from a pre-existing one. PREPARE's own read would + /// not do either: a single-item write that commits between PREPARE's read + /// and its marker LWT is accepted (the marker's `IF prepared_txn_id = null` + /// does not detect a write that left the column null), so PREPARE's image + /// is one write stale whenever that happens, and the record would claim a + /// transition that never took place. Reading at commit time costs one + /// round trip per streamed Put and Delete op; the Update arm reads the row + /// anyway. async fn execute_commit_phase( &self, account_keyspace: &str, @@ -890,6 +911,11 @@ impl CassandraEngine { .await?; } + debug_assert_eq!( + pre_commit_images.len(), + ops.len(), + "one TTL pre-commit image per op, in request order" + ); for (op, old_image) in ops.iter().zip(&pre_commit_images) { match op { TransactWriteOp::Put { key_info, item, .. } => { @@ -1156,6 +1182,25 @@ impl CassandraEngine { key: &Item, skip_txn_id: Option, ) -> Result, StorageError> { + Ok(self + .fetch_row_for_transaction(key_info, key, skip_txn_id) + .await? + .0) + } + + /// Fetch a row for transaction: its `item_data` and whether PREPARE created + /// the row, from `created_to_prepare`. + /// + /// The flag is `false` when there is no row, when the row was not planted by + /// a PREPARE, and when the row had to be read through + /// `wait_for_commit_and_read`, which returns another transaction's committed + /// item and therefore a real old image. + async fn fetch_row_for_transaction( + &self, + key_info: &TableKeyInfo, + key: &Item, + skip_txn_id: Option, + ) -> Result<(Option, bool), StorageError> { let keyspace = self.account_keyspace(&key_info.account_id); let table = data_table_name(&key_info.table_id); let pk_text = composite_pk_to_text(key, &key_info.key_schema)?; @@ -1165,14 +1210,14 @@ impl CassandraEngine { &self.session, &keyspace, &table, - "item_data, prepared_txn_id", + "item_data, prepared_txn_id, created_to_prepare, last_committed_txn_timestamp", pk_text.as_str(), sk.as_ref(), sk_col.as_deref(), ) .await? else { - return Ok(None); + return Ok((None, false)); }; // If a *different* transaction has this row in PREPARE state, wait for @@ -1184,6 +1229,7 @@ impl CassandraEngine { return self .wait_for_commit_and_read(key_info, key) .await + .map(|item| (item, false)) .map_err(|e| { // Preserve a TransactionConflict code and message (e.g. "Transaction // is ongoing for the item" from a wait timeout) instead of flattening @@ -1199,12 +1245,29 @@ impl CassandraEngine { }); } + let created_to_prepare: Option = row.get_by_name("created_to_prepare").ok().flatten(); + let last_committed: Option = row + .get_by_name("last_committed_txn_timestamp") + .ok() + .flatten(); + // A row PREPARE planted has never had a transacted commit land on it: + // only commit_put_or_update and recover_commit_op write + // last_committed_txn_timestamp, and planting requires the row to be + // absent. A release before this branch did not clear the flag on + // commit, so a row it committed carries both; it is a committed item, + // not a placeholder. + let created_to_prepare = created_to_prepare == Some(true) && last_committed.is_none(); + let item_data: Option = row.get_by_name("item_data").ok().flatten(); let Some(item_data) = item_data else { - return Ok(None); + return Ok((None, created_to_prepare)); }; - Ok(Some( - serde_json::from_str(&item_data).map_err(|e| StorageError::Internal(e.to_string()))?, + Ok(( + Some( + serde_json::from_str(&item_data) + .map_err(|e| StorageError::Internal(e.to_string()))?, + ), + created_to_prepare, )) } @@ -1422,32 +1485,96 @@ impl CassandraEngine { let txn_id_bytes = Bytes::new(txn_id.as_bytes().to_vec()); match op { - TransactWriteOp::Put { key_info, item, .. } => { + TransactWriteOp::Put { + key_info, + item, + stream, + .. + } => { + let stream_stmt = if stream.is_some() { + // The item carries the key attributes, so it resolves the + // row by itself. + let (row, created_to_prepare) = self + .fetch_row_for_transaction(key_info, item, Some(txn_id)) + .await?; + let old_image = commit_old_image(row, created_to_prepare); + self.transact_stream_statement( + key_info, + stream.as_ref(), + old_image.as_ref(), + Some(item), + ) + } else { + None + }; + let indexes = self.indexes_for_commit(key_info).await?; self.commit_put_or_update(key_info, item, txn_id_bytes.clone(), txn_timestamp) .await?; - self.commit_sync_indexes(key_info, None, item).await + self.commit_secondary_effects(key_info, None, Some(item), &indexes, stream_stmt) + .await } TransactWriteOp::Update { key_info, key, actions, maps, + stream, .. } => { // Re-fetch and re-apply update (idempotent) - let existing = self - .fetch_item_for_transaction(key_info, key, Some(txn_id)) + let (existing, created_to_prepare) = self + .fetch_row_for_transaction(key_info, key, Some(txn_id)) .await?; let mut final_item = existing.clone().unwrap_or_else(|| (*key).clone()); expression::apply_update_validated(actions, &mut final_item, maps, &[], &[]) .map_err(|e| StorageError::Internal(e.to_string()))?; + let stream_stmt = if stream.is_some() { + let old_image = commit_old_image(existing.clone(), created_to_prepare); + self.transact_stream_statement( + key_info, + stream.as_ref(), + old_image.as_ref(), + Some(&final_item), + ) + } else { + None + }; + let indexes = self.indexes_for_commit(key_info).await?; self.commit_put_or_update(key_info, &final_item, txn_id_bytes, txn_timestamp) .await?; - self.commit_sync_indexes(key_info, existing.as_ref(), &final_item) - .await + self.commit_secondary_effects( + key_info, + existing.as_ref(), + Some(&final_item), + &indexes, + stream_stmt, + ) + .await } - TransactWriteOp::Delete { key_info, key, .. } => { + TransactWriteOp::Delete { + key_info, + key, + stream, + .. + } => { + // A key that was already absent has no pre-image, and the record + // builder then returns no statement, so a transacted delete of a + // missing item records nothing, as the single-item delete does. + let stream_stmt = if stream.is_some() { + let (row, created_to_prepare) = self + .fetch_row_for_transaction(key_info, key, Some(txn_id)) + .await?; + let old_image = commit_old_image(row, created_to_prepare); + self.transact_stream_statement( + key_info, + stream.as_ref(), + old_image.as_ref(), + None, + ) + } else { + None + }; let keyspace = self.account_keyspace(&key_info.account_id); let ddb_table = super::ddl::data_table_name(&key_info.table_id); let pk_text = composite_pk_to_text(key, &key_info.key_schema)?; @@ -1540,7 +1667,8 @@ impl CassandraEngine { check_lwt_applied(&result, "commit_single_operation").map_err(|_| { StorageError::Internal("Transaction conflict during commit".to_owned()) })?; - Ok(()) + self.commit_secondary_effects(key_info, None, None, &[], stream_stmt) + .await } TransactWriteOp::ConditionCheck { key_info, key, .. } => { self.commit_condition_check(key_info, key, txn_id).await @@ -1607,6 +1735,15 @@ impl CassandraEngine { } /// Helper to commit a Put or Update operation. + /// + /// Clears `created_to_prepare` as well as `prepared_txn_id`: the column + /// marks a row that exists only because a PREPARE planted it, and the + /// commit phase reads it to decide whether the mutation has an old image + /// (see `commit_old_image`). Left set, it would make the next transacted + /// write on this key record an insert over a committed item. Rows committed + /// by a release before this clearing still carry the flag; the commit-time + /// read tells them apart by `last_committed_txn_timestamp`, which only a + /// commit writes. async fn commit_put_or_update( &self, key_info: &TableKeyInfo, @@ -1629,13 +1766,13 @@ impl CassandraEngine { let sk_col = sk_column(sk_type); let query = format!( - "UPDATE {keyspace}.{ddb_table} SET item_data = ?, prepared_txn_id = NULL, version = version + 1, last_committed_txn_timestamp = ? \ + "UPDATE {keyspace}.{ddb_table} SET item_data = ?, prepared_txn_id = NULL, created_to_prepare = false, version = version + 1, last_committed_txn_timestamp = ? \ WHERE pk = ? AND {sk_col} = ? IF prepared_txn_id = ?" ); query_with_item_ts_pk_sk_txnid(&self.session, &query, &item_text, txn_timestamp, pk_text.as_str(), &sk, txn_id_bytes).await } else { let query = format!( - "UPDATE {keyspace}.{ddb_table} SET item_data = ?, prepared_txn_id = NULL, version = version + 1, last_committed_txn_timestamp = ? \ + "UPDATE {keyspace}.{ddb_table} SET item_data = ?, prepared_txn_id = NULL, created_to_prepare = false, version = version + 1, last_committed_txn_timestamp = ? \ WHERE pk = ? IF prepared_txn_id = ?" ); self.session @@ -1653,68 +1790,134 @@ impl CassandraEngine { Ok(()) } - /// Sync LSI/GSI index rows after a successful `commit_put_or_update`. + /// Build the stream record statement for one transacted mutation, or `None` + /// when the table has no stream enabled or the mutation records nothing (a + /// delete of a key that was already absent). /// - /// Runs as a separate non-LWT batch after the base-row LWT commit, since - /// Cassandra does not allow LWT statements to be batched with writes to - /// other partitions. The brief inconsistency window is acceptable for the - /// same reason async GSIs accept it; transaction recovery re-runs this on - /// crash. - async fn commit_sync_indexes( + /// Calls the same `stream_util::stream_record_statement` the single-item + /// write paths call, with the same image pair, so a transacted write records + /// the same event name, the same images for the table's view type, the same + /// shard assignment and the same sequence-number source as the equivalent + /// `PutItem`, `UpdateItem` or `DeleteItem`. + fn transact_stream_statement( &self, key_info: &TableKeyInfo, + stream: Option<&StreamCapture>, old_item: Option<&Item>, - new_item: &Item, - ) -> Result<(), StorageError> { - let catalog_keyspace = self.catalog_keyspace(); - let data_keyspace = self.account_keyspace(&key_info.account_id); + new_item: Option<&Item>, + ) -> Option { + let capture = stream?; + crate::stream_util::stream_record_statement( + &self.account_keyspace(&key_info.account_id), + &key_info.table_id, + key_info, + old_item, + new_item, + capture, + &self.hlc, + self.stream_retention_seconds, + ) + } - let indexes = super::index::fetch_indexes_for_table( + /// The table's index metadata, read before the base mutation commits. + /// + /// `fetch_indexes_for_table` is an uncached catalog read. Running it after + /// the base LWT would make every record and index row wait a round trip + /// behind an item that has already committed, widening the window in which + /// a crash leaves the item mutated and its secondary effects unwritten. + /// `occ_write` reads the indexes before its own base write for the same + /// reason. + async fn indexes_for_commit( + &self, + key_info: &TableKeyInfo, + ) -> Result, StorageError> { + super::index::fetch_indexes_for_table( &key_info.table_id, &self.session, - &catalog_keyspace, + &self.catalog_keyspace(), ) - .await?; + .await + } + + /// Run the secondary-effects LOGGED BATCH for one committed transactional + /// mutation: the LSI/GSI index rows and the stream record. + /// + /// Runs after the base-row LWT commit, since Cassandra does not allow LWT + /// statements to be batched with writes to other partitions. The single-item + /// paths have the same two-step shape for the same reason: `occ_write` + /// commits the base row by LWT, then carries the index rows and the stream + /// record in one LOGGED BATCH. The brief inconsistency window is acceptable + /// for the same reason async GSIs accept it; transaction recovery re-runs the + /// base mutation on crash. + /// + /// `new_item` is `None` for a delete, which writes no index rows on this + /// path, and `indexes` is then empty. `index_old_item` is the old image the + /// index sync receives, which a put leaves as `None`; the stream record's + /// own old image is already baked into `stream_stmt`. + /// + /// The statements run as `secondary_effect_batches` groups them, which is + /// one batch holding the index rows and the record together. The queue row + /// an asynchronous GSI takes (`enqueue_async_indexes`) is written on the + /// session outside that batch, as it was before the record joined it. + async fn commit_secondary_effects( + &self, + key_info: &TableKeyInfo, + index_old_item: Option<&Item>, + new_item: Option<&Item>, + indexes: &[super::index::IndexMeta], + stream_stmt: Option, + ) -> Result<(), StorageError> { + let data_keyspace = self.account_keyspace(&key_info.account_id); - if indexes.is_empty() { + if indexes.is_empty() && stream_stmt.is_none() { return Ok(()); } - let sys_delay = self - .gsi_default_delay_ms - .load(std::sync::atomic::Ordering::Relaxed); let mut batch = BatchQueryBuilder::new().with_consistency(Consistency::LocalQuorum); - super::index::sync_indexes( - &mut batch, - &data_keyspace, - &key_info.key_schema, - &key_info.attribute_definitions, - &indexes, - old_item, - Some(new_item), - sys_delay, - )?; - - let async_enqueued = super::index::enqueue_async_indexes( - &self.session, - &mut batch, - &data_keyspace, - key_info, - &indexes, - old_item, - Some(new_item), - sys_delay, - ) - .await?; + let mut async_enqueued = 0; - self.session - .batch( - batch - .build() - .map_err(|e| StorageError::Internal(e.to_string()))?, + if let Some(new_item) = new_item + && !indexes.is_empty() + { + let sys_delay = self + .gsi_default_delay_ms + .load(std::sync::atomic::Ordering::Relaxed); + super::index::sync_indexes( + &mut batch, + &data_keyspace, + &key_info.key_schema, + &key_info.attribute_definitions, + indexes, + index_old_item, + Some(new_item), + sys_delay, + )?; + + async_enqueued = super::index::enqueue_async_indexes( + &self.session, + &mut batch, + &data_keyspace, + key_info, + indexes, + index_old_item, + Some(new_item), + sys_delay, ) - .await - .map_err(|e| StorageError::Internal(format!("commit_sync_indexes batch: {e}")))?; + .await?; + } + + for batch in secondary_effect_batches(batch, stream_stmt) { + self.session + .batch( + batch + .build() + .map_err(|e| StorageError::Internal(e.to_string()))?, + ) + .await + .map_err(|e| { + StorageError::Internal(format!("commit_secondary_effects batch: {e}")) + })?; + } if async_enqueued > 0 { self.gsi_queue.notify_workers(); @@ -1962,7 +2165,8 @@ impl CassandraEngine { let result = if let Some((sk_val, sk_col)) = &sk { let query = format!( "UPDATE {keyspace}.{table} SET item_data = ?, prepared_txn_id = NULL, \ - version = version + 1, last_committed_txn_timestamp = ? \ + created_to_prepare = false, version = version + 1, \ + last_committed_txn_timestamp = ? \ WHERE pk = ? AND {sk_col} = ? IF prepared_txn_id = ?", ); query_with_item_ts_pk_sk_txnid( @@ -1978,7 +2182,8 @@ impl CassandraEngine { } else { let query = format!( "UPDATE {keyspace}.{table} SET item_data = ?, prepared_txn_id = NULL, \ - version = version + 1, last_committed_txn_timestamp = ? \ + created_to_prepare = false, version = version + 1, \ + last_committed_txn_timestamp = ? \ WHERE pk = ? IF prepared_txn_id = ?", ); self.session diff --git a/crates/storage-cassandra/tests/direct/transact_write_stream_images.rs b/crates/storage-cassandra/tests/direct/transact_write_stream_images.rs new file mode 100644 index 000000000..7c3d94305 --- /dev/null +++ b/crates/storage-cassandra/tests/direct/transact_write_stream_images.rs @@ -0,0 +1,408 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Direct integration tests for the old image a transacted mutation records, +//! and for a transacted record against the single-item record of the same verb. +//! +//! Helpers are shared with `transact_write_stream_records.rs`, which covers the +//! event names, the view types and the ops that record nothing. + +use extenddb_core::expression::ExpressionMaps; +use extenddb_core::types::{ + AttributeValue, Item, ReturnValuesOnConditionCheckFailure, StreamViewType, +}; +use extenddb_storage::{DataEngine, TransactWriteOp}; +use extenddb_storage_cassandra::CassandraEngine; + +use crate::helpers::{TestTable, setup_engine, unique_test_account}; +use crate::transact_write_stream_records::{ + capture, item_with, key_for, records_for_keys, set_val_action, +}; + +/// One transacted Put of `item`, with a `NEW_AND_OLD_IMAGES` capture. +async fn transacted_put(engine: &CassandraEngine, table: &TestTable, item: &Item) { + let maps = ExpressionMaps::default(); + let ops = vec![TransactWriteOp::Put { + key_info: &table.key_info, + item, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(capture(StreamViewType::NewAndOldImages)), + }]; + engine.transact_write_items(&ops, None).await.unwrap(); +} + +/// Write the row shape a PREPARE leaves behind: `created_to_prepare` set, with +/// `item_data` carrying a placeholder rather than a committed item, and no +/// `prepared_txn_id`. +/// +/// A commit-time read of `item_data` alone cannot tell this row from a row +/// holding a committed item, and PREPARE's own read cannot tell it either. It +/// is also the shape a single-item write that commits inside the PREPARE window +/// leaves, which is why the record's old image is decided from the flag. +async fn plant_prepare_created_row(engine: &CassandraEngine, table: &TestTable, item: &Item) { + let keyspace = format!("extenddb_ttl_test_account_{}", table.key_info.account_id); + let data_table = format!("items_{}", table.key_info.table_id.replace('-', "_")); + let AttributeValue::S(pk) = item.get("id").unwrap() else { + panic!("the test tables key on a string `id`"); + }; + let item_data = serde_json::to_string(item).unwrap(); + let query = format!( + "INSERT INTO {keyspace}.{data_table} (pk, item_data, created_to_prepare) \ + VALUES (?, ?, true)" + ); + engine + .session_arc() + .query_with_values( + &query, + cdrs_tokio::query_values!(pk.as_str(), item_data.as_str()), + ) + .await + .unwrap(); +} + +/// A transacted Put over a key that already holds an item records MODIFY with +/// the stored item as `OldImage`, as an overwriting `PutItem` does. +#[tokio::test] +async fn test_transact_write_put_over_existing_records_modify() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnStreamPutOverwrite", false).await; + let maps = ExpressionMaps::default(); + + engine + .put_item( + &table.key_info, + item_with("ovr-1", "before"), + false, + None, + &maps, + None, + ) + .await + .unwrap(); + + transacted_put(&engine, &table, &item_with("ovr-1", "after")).await; + + let records = records_for_keys( + &engine, + &table.key_info.account_id, + &table.key_info.table_id, + &["ovr-1"], + ) + .await; + assert_eq!(records.len(), 1, "got {records:?}"); + assert_eq!(records[0].1, "Modify"); + assert_eq!(records[0].2["eventName"], "MODIFY"); + assert_eq!(records[0].2["dynamodb"]["OldImage"]["val"]["S"], "before"); + assert_eq!(records[0].2["dynamodb"]["NewImage"]["val"]["S"], "after"); +} + +/// A transacted Put over a row that only exists because a PREPARE planted it +/// records INSERT: that row holds a placeholder, not a committed item. +#[tokio::test] +async fn test_transact_write_put_over_prepare_created_row_records_insert() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnStreamPrepareMarker", false).await; + + plant_prepare_created_row(&engine, &table, &item_with("mark-1", "placeholder")).await; + + transacted_put(&engine, &table, &item_with("mark-1", "committed")).await; + + let records = records_for_keys( + &engine, + &table.key_info.account_id, + &table.key_info.table_id, + &["mark-1"], + ) + .await; + assert_eq!(records.len(), 1, "got {records:?}"); + assert_eq!(records[0].1, "Insert"); + assert_eq!(records[0].2["eventName"], "INSERT"); + assert!( + records[0].2["dynamodb"].get("OldImage").is_none(), + "a planted row is not an old image: {records:?}" + ); + assert_eq!( + records[0].2["dynamodb"]["NewImage"]["val"]["S"], + "committed" + ); +} + +/// A second transacted Put of a key an earlier transaction created records +/// MODIFY: committing clears `created_to_prepare`, so the row is a committed +/// item from then on. +#[tokio::test] +async fn test_transact_write_put_over_transacted_key_records_modify() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnStreamPutTwice", false).await; + + transacted_put(&engine, &table, &item_with("twice-1", "first")).await; + transacted_put(&engine, &table, &item_with("twice-1", "second")).await; + + let records = records_for_keys( + &engine, + &table.key_info.account_id, + &table.key_info.table_id, + &["twice-1"], + ) + .await; + assert_eq!(records.len(), 2, "got {records:?}"); + let events: Vec<&str> = records.iter().map(|(_, event, _)| event.as_str()).collect(); + assert_eq!(events, vec!["Insert", "Modify"]); + assert_eq!(records[1].2["dynamodb"]["OldImage"]["val"]["S"], "first"); + assert_eq!(records[1].2["dynamodb"]["NewImage"]["val"]["S"], "second"); +} + +/// The one record for `key`, as `record_data`. +fn record_for(records: &[(String, String, serde_json::Value)], key: &str) -> serde_json::Value { + let matching: Vec<&serde_json::Value> = records + .iter() + .filter(|(_, _, data)| data["dynamodb"]["Keys"]["id"]["S"] == key) + .map(|(_, _, data)| data) + .collect(); + assert_eq!(matching.len(), 1, "one record for {key}: {matching:?}"); + matching[0].clone() +} + +/// A record without the fields that identify the write rather than describe it. +fn without_identity(record: &serde_json::Value) -> serde_json::Value { + let mut record = record.clone(); + let top = record.as_object_mut().unwrap(); + top.remove("eventID"); + let data = top.get_mut("dynamodb").unwrap().as_object_mut().unwrap(); + data.remove("SequenceNumber"); + data.remove("ApproximateCreationDateTime"); + record +} + +/// A transacted Put, Update and Delete record what the equivalent `PutItem`, +/// `UpdateItem` and `DeleteItem` record: the same fields with the same values, +/// once the per-write identity fields are removed. +/// +/// Two tables in one account carry the same keys and the same items, so the +/// records differ in nothing but which path wrote them. +#[tokio::test] +async fn test_transacted_records_match_the_single_item_records() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let account_id = unique_test_account(); + let solo = TestTable::with_account(&engine, &account_id, "TxnStreamSoloPath", false).await; + let txn = TestTable::with_account(&engine, &account_id, "TxnStreamTxnPath", false).await; + let maps = ExpressionMaps::default(); + let view = capture(StreamViewType::NewAndOldImages); + + // Seed the keys the update and the delete act on, on both tables, without a + // capture, so the only records are the ones under comparison. + for table in [&solo, &txn] { + for (id, val) in [("k-upd", "before"), ("k-del", "doomed")] { + engine + .put_item( + &table.key_info, + item_with(id, val), + false, + None, + &maps, + None, + ) + .await + .unwrap(); + } + } + + let created = item_with("k-put", "created"); + let (actions, update_maps) = set_val_action("after"); + + engine + .put_item( + &solo.key_info, + created.clone(), + false, + None, + &maps, + Some(&view), + ) + .await + .unwrap(); + engine + .update_item( + &solo.key_info, + &key_for("k-upd"), + &actions, + false, + false, + None, + &update_maps, + Some(&view), + ) + .await + .unwrap(); + engine + .delete_item( + &solo.key_info, + &key_for("k-del"), + false, + None, + &maps, + Some(&view), + ) + .await + .unwrap(); + + let update_key = key_for("k-upd"); + let delete_key = key_for("k-del"); + let ops = vec![ + TransactWriteOp::Put { + key_info: &txn.key_info, + item: &created, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(view.clone()), + }, + TransactWriteOp::Update { + key_info: &txn.key_info, + key: &update_key, + actions: &actions, + condition: None, + maps: &update_maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(view.clone()), + }, + TransactWriteOp::Delete { + key_info: &txn.key_info, + key: &delete_key, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(view.clone()), + }, + ]; + engine.transact_write_items(&ops, None).await.unwrap(); + + let keys = ["k-put", "k-upd", "k-del"]; + let solo_records = records_for_keys(&engine, &account_id, &solo.key_info.table_id, &keys).await; + let txn_records = records_for_keys(&engine, &account_id, &txn.key_info.table_id, &keys).await; + assert_eq!(solo_records.len(), 3, "got {solo_records:?}"); + assert_eq!(txn_records.len(), 3, "got {txn_records:?}"); + + for key in keys { + let single = record_for(&solo_records, key); + let transacted = record_for(&txn_records, key); + assert_eq!( + without_identity(&single), + without_identity(&transacted), + "the records for {key} differ" + ); + } +} + +/// Puts the row into the state a release before this branch leaves after a +/// transacted commit: committed `item_data`, `prepared_txn_id` null, +/// `last_committed_txn_timestamp` set, and `created_to_prepare` still true, +/// since that release's commit statement did not clear the flag. The single +/// `UPDATE` here is the only difference between that statement and this +/// branch's, so the row is what an upgraded database holds. +async fn leave_flag_set_as_an_earlier_release_did( + engine: &CassandraEngine, + table: &TestTable, + id: &str, +) { + let keyspace = format!("extenddb_ttl_test_account_{}", table.key_info.account_id); + let data_table = format!("items_{}", table.key_info.table_id.replace('-', "_")); + let query = + format!("UPDATE {keyspace}.{data_table} SET created_to_prepare = true WHERE pk = ?"); + engine + .session_arc() + .query_with_values(&query, cdrs_tokio::query_values!(id)) + .await + .unwrap(); +} + +/// A transacted Delete of an item committed by an earlier release records +/// REMOVE: the row carries the flag and a commit timestamp, and the timestamp +/// says it is a committed item, not a placeholder. +#[tokio::test] +async fn test_transact_write_delete_of_an_item_committed_before_the_flag_was_cleared_records_remove() + { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnStreamLegacyDelete", false).await; + let maps = ExpressionMaps::default(); + + transacted_put(&engine, &table, &item_with("leg-del", "committed")).await; + leave_flag_set_as_an_earlier_release_did(&engine, &table, "leg-del").await; + + let key = key_for("leg-del"); + let ops = vec![TransactWriteOp::Delete { + key_info: &table.key_info, + key: &key, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(capture(StreamViewType::NewAndOldImages)), + }]; + engine.transact_write_items(&ops, None).await.unwrap(); + + let records = records_for_keys( + &engine, + &table.key_info.account_id, + &table.key_info.table_id, + &["leg-del"], + ) + .await; + let names: Vec<&str> = records.iter().map(|r| r.1.as_str()).collect(); + assert_eq!( + names, + vec!["Insert", "Remove"], + "a transacted delete of a committed item records REMOVE: {records:?}" + ); +} + +/// A transacted Put over an item committed by an earlier release records +/// MODIFY with the committed item as `OldImage`. +#[tokio::test] +async fn test_transact_write_put_over_an_item_committed_before_the_flag_was_cleared_records_modify() +{ + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnStreamLegacyPut", false).await; + + transacted_put(&engine, &table, &item_with("leg-put", "first")).await; + leave_flag_set_as_an_earlier_release_did(&engine, &table, "leg-put").await; + + transacted_put(&engine, &table, &item_with("leg-put", "second")).await; + + let records = records_for_keys( + &engine, + &table.key_info.account_id, + &table.key_info.table_id, + &["leg-put"], + ) + .await; + let names: Vec<&str> = records.iter().map(|r| r.1.as_str()).collect(); + assert_eq!( + names, + vec!["Insert", "Modify"], + "a transacted put over a committed item records MODIFY: {records:?}" + ); + assert_eq!( + records[1].2["dynamodb"]["OldImage"]["val"]["S"], "first", + "the committed item is the old image: {records:?}" + ); +} diff --git a/crates/storage-cassandra/tests/direct/transact_write_stream_records.rs b/crates/storage-cassandra/tests/direct/transact_write_stream_records.rs new file mode 100644 index 000000000..26951bcd5 --- /dev/null +++ b/crates/storage-cassandra/tests/direct/transact_write_stream_records.rs @@ -0,0 +1,576 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Direct integration tests for stream records written by `TransactWriteItems`. + +use std::collections::HashMap; +use std::sync::Arc; + +use cdrs_tokio::types::IntoRustByName; +use extenddb_core::expression::{Expr, ExpressionMaps, PathElement, UpdateAction}; +use extenddb_core::types::{ + AttributeValue, Item, ReturnValuesOnConditionCheckFailure, StreamViewType, +}; +use extenddb_storage::{DataEngine, StreamCapture, TransactWriteOp}; +use extenddb_storage_cassandra::CassandraEngine; + +use crate::helpers::{TestTable, setup_engine}; + +pub(crate) fn capture(view_type: StreamViewType) -> StreamCapture { + StreamCapture { + view_type, + user_identity: None, + region: Arc::from("us-east-1"), + } +} + +pub(crate) fn item_with(id: &str, val: &str) -> Item { + let mut item = Item::new(); + item.insert("id".to_owned(), AttributeValue::S(id.to_owned())); + item.insert("val".to_owned(), AttributeValue::S(val.to_owned())); + item +} + +pub(crate) fn key_for(id: &str) -> Item { + let mut key = Item::new(); + key.insert("id".to_owned(), AttributeValue::S(id.to_owned())); + key +} + +/// `attribute_not_exists(id)`, the condition used to cancel a transaction on a +/// key that does exist. +fn attribute_not_exists_id() -> Expr { + Expr::Function { + name: "attribute_not_exists".to_owned(), + args: vec![Expr::Path(vec![PathElement::Attribute("id".to_owned())])], + } +} + +/// `attribute_exists(id)`, a condition that holds for a key that is present. +fn attribute_exists_id() -> Expr { + Expr::Function { + name: "attribute_exists".to_owned(), + args: vec![Expr::Path(vec![PathElement::Attribute("id".to_owned())])], + } +} + +/// `SET val = :after`, with the map that resolves the placeholder. +pub(crate) fn set_val_action(value: &str) -> (Vec, ExpressionMaps) { + let mut values = HashMap::new(); + values.insert("after".to_owned(), AttributeValue::S(value.to_owned())); + let actions = vec![UpdateAction::Set { + path: vec![PathElement::Attribute("val".to_owned())], + value: Expr::Placeholder("after".to_owned()), + }]; + (actions, ExpressionMaps::new(HashMap::new(), values)) +} + +/// All stream records for the given partition keys, as `(sequence_number, +/// event_name, record_data)` sorted by sequence number. +/// +/// The shards come from `stream_util::assign_shard_id`, the function the write +/// paths mint the record's shard with, so a change in shard naming fails the +/// assertions below rather than returning no records at all. +pub(crate) async fn records_for_keys( + engine: &CassandraEngine, + account_id: &str, + table_id: &str, + keys: &[&str], +) -> Vec<(String, String, serde_json::Value)> { + let keyspace = format!("extenddb_ttl_test_account_{account_id}"); + let mut shard_ids: Vec = keys + .iter() + .map(|key| extenddb_storage_cassandra::stream_util::assign_shard_id(key, table_id)) + .collect(); + shard_ids.sort(); + shard_ids.dedup(); + + let mut rows = Vec::new(); + for shard_id in shard_ids { + let query = format!( + "SELECT sequence_number, event_name, record_data FROM {keyspace}.stream_records \ + WHERE shard_id = ?" + ); + let result = engine + .session_arc() + .query_with_values(&query, cdrs_tokio::query_values!(shard_id.as_str())) + .await + .unwrap(); + let body = result.response_body().unwrap(); + for row in body.into_rows().unwrap_or_default() { + let sequence_number: String = row.get_r_by_name("sequence_number").unwrap(); + let event_name: String = row.get_r_by_name("event_name").unwrap(); + let data: String = row.get_r_by_name("record_data").unwrap(); + rows.push(( + sequence_number, + event_name, + serde_json::from_str(&data).unwrap(), + )); + } + } + rows.sort_by(|a, b| a.0.cmp(&b.0)); + rows +} + +/// A transaction carrying a Put, an Update and a Delete on a streamed table +/// records one record per item mutation, in request order, with the event names +/// and images the single-item writes produce. +#[tokio::test] +async fn test_transact_write_records_one_stream_record_per_mutation() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnStreamAllVerbs", false).await; + let maps = ExpressionMaps::default(); + + // Seed the keys the Update and the Delete act on, without a capture, so the + // only records in the table are the transaction's. + for (id, val) in [("upd-1", "before"), ("del-1", "doomed")] { + engine + .put_item( + &table.key_info, + item_with(id, val), + false, + None, + &maps, + None, + ) + .await + .unwrap(); + } + + let put_item = item_with("put-1", "created"); + let update_key = key_for("upd-1"); + let delete_key = key_for("del-1"); + let (actions, update_maps) = set_val_action("after"); + let view = capture(StreamViewType::NewAndOldImages); + + let ops = vec![ + TransactWriteOp::Put { + key_info: &table.key_info, + item: &put_item, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(view.clone()), + }, + TransactWriteOp::Update { + key_info: &table.key_info, + key: &update_key, + actions: &actions, + condition: None, + maps: &update_maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(view.clone()), + }, + TransactWriteOp::Delete { + key_info: &table.key_info, + key: &delete_key, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(view.clone()), + }, + ]; + + engine.transact_write_items(&ops, None).await.unwrap(); + + let records = records_for_keys( + &engine, + &table.key_info.account_id, + &table.key_info.table_id, + &["put-1", "upd-1", "del-1"], + ) + .await; + + assert_eq!( + records.len(), + 3, + "expected one record per item mutation, got {records:?}" + ); + + let events: Vec<&str> = records.iter().map(|(_, event, _)| event.as_str()).collect(); + assert_eq!(events, vec!["Insert", "Modify", "Remove"]); + + let (_, _, insert) = &records[0]; + assert_eq!(insert["eventName"], "INSERT"); + assert_eq!(insert["dynamodb"]["Keys"]["id"]["S"], "put-1"); + assert_eq!(insert["dynamodb"]["NewImage"]["val"]["S"], "created"); + assert!(insert["dynamodb"].get("OldImage").is_none()); + + let (_, _, modify) = &records[1]; + assert_eq!(modify["eventName"], "MODIFY"); + assert_eq!(modify["dynamodb"]["Keys"]["id"]["S"], "upd-1"); + assert_eq!(modify["dynamodb"]["OldImage"]["val"]["S"], "before"); + assert_eq!(modify["dynamodb"]["NewImage"]["val"]["S"], "after"); + + let (_, _, remove) = &records[2]; + assert_eq!(remove["eventName"], "REMOVE"); + assert_eq!(remove["dynamodb"]["Keys"]["id"]["S"], "del-1"); + assert_eq!(remove["dynamodb"]["OldImage"]["val"]["S"], "doomed"); + assert!(remove["dynamodb"].get("NewImage").is_none()); + + for (_, _, record) in &records { + assert_eq!(record["eventSource"], "aws:dynamodb"); + assert_eq!(record["awsRegion"], "us-east-1"); + assert_eq!( + record["dynamodb"]["StreamViewType"], "NEW_AND_OLD_IMAGES", + "record carries the capture's view type" + ); + } +} + +/// Two mutations on the same streamed table record two records, ordered as the +/// request lists them. +/// +/// This pins ExtendDB's own behaviour on every backend. The service makes no +/// such promise: one `TransactWriteItems` of a Put to `k5` and a Put to `k7` +/// that share a shard came back `k7`, `k5` (measured 2026-10-25), so a +/// consumer can rely on the order of one key's records and not on the order of +/// a transaction's. +#[tokio::test] +async fn test_transact_write_records_two_mutations_in_request_order() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnStreamTwoPuts", false).await; + let maps = ExpressionMaps::default(); + let view = capture(StreamViewType::NewAndOldImages); + + let first = item_with("ord-a", "first"); + let second = item_with("ord-b", "second"); + let ops = vec![ + TransactWriteOp::Put { + key_info: &table.key_info, + item: &first, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(view.clone()), + }, + TransactWriteOp::Put { + key_info: &table.key_info, + item: &second, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(view.clone()), + }, + ]; + + engine.transact_write_items(&ops, None).await.unwrap(); + + let records = records_for_keys( + &engine, + &table.key_info.account_id, + &table.key_info.table_id, + &["ord-a", "ord-b"], + ) + .await; + + assert_eq!(records.len(), 2, "got {records:?}"); + // The records are sorted by sequence number, so the keys in this order are + // the request's order. + assert_eq!(records[0].2["dynamodb"]["Keys"]["id"]["S"], "ord-a"); + assert_eq!(records[1].2["dynamodb"]["Keys"]["id"]["S"], "ord-b"); +} + +/// A transaction spanning a streamed and an unstreamed table records the +/// streamed table's mutation only. +#[tokio::test] +async fn test_transact_write_records_streamed_table_only() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let account_id = crate::helpers::unique_test_account(); + let streamed = TestTable::with_account(&engine, &account_id, "TxnStreamOn", false).await; + let plain = TestTable::with_account(&engine, &account_id, "TxnStreamOff", false).await; + let maps = ExpressionMaps::default(); + + let streamed_item = item_with("on-1", "recorded"); + let plain_item = item_with("off-1", "silent"); + let ops = vec![ + TransactWriteOp::Put { + key_info: &streamed.key_info, + item: &streamed_item, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(capture(StreamViewType::NewAndOldImages)), + }, + TransactWriteOp::Put { + key_info: &plain.key_info, + item: &plain_item, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: None, + }, + ]; + + engine.transact_write_items(&ops, None).await.unwrap(); + + let streamed_records = + records_for_keys(&engine, &account_id, &streamed.key_info.table_id, &["on-1"]).await; + assert_eq!(streamed_records.len(), 1, "got {streamed_records:?}"); + assert_eq!(streamed_records[0].2["dynamodb"]["Keys"]["id"]["S"], "on-1"); + + let plain_records = + records_for_keys(&engine, &account_id, &plain.key_info.table_id, &["off-1"]).await; + assert!( + plain_records.is_empty(), + "unstreamed table recorded {plain_records:?}" + ); +} + +/// A `KEYS_ONLY` stream records the keys and neither image. +#[tokio::test] +async fn test_transact_write_keys_only_view_omits_images() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnStreamKeysOnly", false).await; + let maps = ExpressionMaps::default(); + + let item = item_with("keys-1", "hidden"); + let ops = vec![TransactWriteOp::Put { + key_info: &table.key_info, + item: &item, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(capture(StreamViewType::KeysOnly)), + }]; + + engine.transact_write_items(&ops, None).await.unwrap(); + + let records = records_for_keys( + &engine, + &table.key_info.account_id, + &table.key_info.table_id, + &["keys-1"], + ) + .await; + assert_eq!(records.len(), 1, "got {records:?}"); + assert_eq!(records[0].2["dynamodb"]["Keys"]["id"]["S"], "keys-1"); + assert!(records[0].2["dynamodb"].get("NewImage").is_none()); + assert!(records[0].2["dynamodb"].get("OldImage").is_none()); + assert_eq!(records[0].2["dynamodb"]["StreamViewType"], "KEYS_ONLY"); +} + +/// A `ConditionCheck` records nothing, while the Put beside it records. +#[tokio::test] +async fn test_transact_write_condition_check_records_nothing() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnStreamCheck", false).await; + let maps = ExpressionMaps::default(); + + engine + .put_item( + &table.key_info, + item_with("chk-1", "present"), + false, + None, + &maps, + None, + ) + .await + .unwrap(); + + let check_key = key_for("chk-1"); + let put_item = item_with("chk-2", "written"); + let exists = attribute_exists_id(); + let ops = vec![ + TransactWriteOp::ConditionCheck { + key_info: &table.key_info, + key: &check_key, + condition: &exists, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + }, + TransactWriteOp::Put { + key_info: &table.key_info, + item: &put_item, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(capture(StreamViewType::NewAndOldImages)), + }, + ]; + + engine.transact_write_items(&ops, None).await.unwrap(); + + let records = records_for_keys( + &engine, + &table.key_info.account_id, + &table.key_info.table_id, + &["chk-1", "chk-2"], + ) + .await; + assert_eq!(records.len(), 1, "got {records:?}"); + assert_eq!(records[0].2["dynamodb"]["Keys"]["id"]["S"], "chk-2"); +} + +/// A transaction cancelled by a condition failure records nothing for any of +/// its items, including the ones whose conditions held. +#[tokio::test] +async fn test_cancelled_transaction_records_nothing() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnStreamCancelled", false).await; + let maps = ExpressionMaps::default(); + + engine + .put_item( + &table.key_info, + item_with("taken", "present"), + false, + None, + &maps, + None, + ) + .await + .unwrap(); + + let free_item = item_with("free", "would-be-written"); + let taken_item = item_with("taken", "would-overwrite"); + let condition = attribute_not_exists_id(); + let view = capture(StreamViewType::NewAndOldImages); + + let ops = vec![ + TransactWriteOp::Put { + key_info: &table.key_info, + item: &free_item, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(view.clone()), + }, + // attribute_not_exists(id) against a key that exists: cancels the + // transaction. + TransactWriteOp::Put { + key_info: &table.key_info, + item: &taken_item, + condition: Some(&condition), + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(view.clone()), + }, + ]; + + let result = engine.transact_write_items(&ops, None).await; + assert!( + matches!( + result, + Err(extenddb_storage::error::StorageError::TransactionCanceled( + _ + )) + ), + "expected cancellation, got {result:?}" + ); + + let records = records_for_keys( + &engine, + &table.key_info.account_id, + &table.key_info.table_id, + &["free", "taken"], + ) + .await; + assert!( + records.is_empty(), + "a cancelled transaction recorded {records:?}" + ); + + // The uncancelled op's item is not written either. + assert!( + engine + .get_item(&table.key_info, &key_for("free")) + .await + .unwrap() + .is_none(), + "cancelled transaction wrote an item" + ); +} + +/// A transacted Update of a key that does not exist creates the item and +/// records INSERT with no old image, as `UpdateItem` does. +#[tokio::test] +async fn test_transact_write_update_of_absent_key_records_insert() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnStreamUpdInsert", false).await; + + let key = key_for("upd-new"); + let (actions, update_maps) = set_val_action("created-by-update"); + let ops = vec![TransactWriteOp::Update { + key_info: &table.key_info, + key: &key, + actions: &actions, + condition: None, + maps: &update_maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(capture(StreamViewType::NewAndOldImages)), + }]; + + engine.transact_write_items(&ops, None).await.unwrap(); + + let records = records_for_keys( + &engine, + &table.key_info.account_id, + &table.key_info.table_id, + &["upd-new"], + ) + .await; + assert_eq!(records.len(), 1, "got {records:?}"); + assert_eq!(records[0].1, "Insert"); + assert_eq!(records[0].2["eventName"], "INSERT"); + assert_eq!( + records[0].2["dynamodb"]["NewImage"]["val"]["S"], + "created-by-update" + ); + assert!(records[0].2["dynamodb"].get("OldImage").is_none()); +} + +/// A transacted delete of a key that was already absent records nothing, as the +/// single-item delete does. +#[tokio::test] +async fn test_transact_write_delete_of_absent_key_records_nothing() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnStreamDeleteGhost", false).await; + let maps = ExpressionMaps::default(); + + let ghost = key_for("ghost"); + let ops = vec![TransactWriteOp::Delete { + key_info: &table.key_info, + key: &ghost, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(capture(StreamViewType::NewAndOldImages)), + }]; + + engine.transact_write_items(&ops, None).await.unwrap(); + + let records = records_for_keys( + &engine, + &table.key_info.account_id, + &table.key_info.table_id, + &["ghost"], + ) + .await; + assert!(records.is_empty(), "got {records:?}"); +} diff --git a/crates/storage-cassandra/tests/direct_integration.rs b/crates/storage-cassandra/tests/direct_integration.rs index 43eb28b81..b20ed60b1 100644 --- a/crates/storage-cassandra/tests/direct_integration.rs +++ b/crates/storage-cassandra/tests/direct_integration.rs @@ -53,6 +53,10 @@ mod table_engine; mod transact_get_items; #[path = "direct/transact_write_items.rs"] mod transact_write_items; +#[path = "direct/transact_write_stream_images.rs"] +mod transact_write_stream_images; +#[path = "direct/transact_write_stream_records.rs"] +mod transact_write_stream_records; #[path = "direct/transaction_ledger.rs"] mod transaction_ledger; #[path = "direct/users.rs"] diff --git a/tests/python/test_streams_transactions.py b/tests/python/test_streams_transactions.py new file mode 100644 index 000000000..b9699a13b --- /dev/null +++ b/tests/python/test_streams_transactions.py @@ -0,0 +1,295 @@ +# Copyright 2026 ExtendDB contributors +# SPDX-License-Identifier: Apache-2.0 + +"""Stream records for TransactWriteItems. + +A transacted Put, Update or Delete records like the equivalent single-item +write. A ConditionCheck records nothing, and a cancelled transaction records +nothing for any of its items. + +Assertions are per key. Records for one key share a shard and carry ascending +sequence numbers there, while records for different keys land in different +shards, whose sequence numbers are not comparable. + +These tests require a extenddb endpoint (DYNAMODB_ENDPOINT) because real +DynamoDB uses a separate Streams endpoint. +""" + +from __future__ import annotations + +import os +import time + +import boto3 +import botocore.config +import pytest +from botocore.exceptions import ClientError + +from helpers import unique_name, wait_for_active, wait_for_deleted + +pytestmark = pytest.mark.skipif( + not os.environ.get("DYNAMODB_ENDPOINT", "").strip(), + reason="Streams tests require extenddb (DYNAMODB_ENDPOINT)", +) + + +@pytest.fixture(scope="module") +def streams_client(endpoint_url): + """DynamoDB Streams client pointing at the extenddb endpoint.""" + kwargs: dict = { + "service_name": "dynamodbstreams", + "region_name": os.environ.get("AWS_DEFAULT_REGION", "us-east-1"), + "config": botocore.config.Config(retries={"max_attempts": 0}), + } + if endpoint_url: + kwargs["endpoint_url"] = endpoint_url + if endpoint_url.startswith("https://"): + ca_cert = os.environ.get("EXTENDDB_CA_CERT", "") + kwargs["verify"] = ca_cert if ca_cert else False + return boto3.client(**kwargs) + + +def _create_stream_table( + dynamodb_client, view_type: str = "NEW_AND_OLD_IMAGES" +) -> tuple[str, str]: + """Create a table with streams enabled, return (table_name, stream_arn).""" + name = unique_name("txnstream") + resp = dynamodb_client.create_table( + TableName=name, + AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}], + KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}], + BillingMode="PAY_PER_REQUEST", + StreamSpecification={ + "StreamEnabled": True, + "StreamViewType": view_type, + }, + ) + stream_arn = resp["TableDescription"]["LatestStreamArn"] + wait_for_active(dynamodb_client, name) + return name, stream_arn + + +def _read_all_shards(streams_client, stream_arn: str, max_polls: int = 10) -> list[dict]: + """Read records from every shard of a stream, keeping each shard's order. + + extenddb spreads records over several shards by hashing the partition key, + so every shard has to be read to collect a transaction's records. + """ + desc = streams_client.describe_stream(StreamArn=stream_arn) + records: list[dict] = [] + for shard in desc["StreamDescription"]["Shards"]: + resp = streams_client.get_shard_iterator( + StreamArn=stream_arn, + ShardId=shard["ShardId"], + ShardIteratorType="TRIM_HORIZON", + ) + iterator = resp["ShardIterator"] + for _ in range(max_polls): + resp = streams_client.get_records(ShardIterator=iterator) + records.extend(resp["Records"]) + iterator = resp.get("NextShardIterator") + if not iterator or not resp["Records"]: + break + return records + + +def _for_key(records: list[dict], key: str) -> list[dict]: + """The records for one partition key, in the order the shard returned them.""" + return [r for r in records if r["dynamodb"]["Keys"]["pk"]["S"] == key] + + +def _drop(dynamodb_client, name: str) -> None: + dynamodb_client.delete_table(TableName=name) + wait_for_deleted(dynamodb_client, name) + + +class TestTransactWriteStreamRecords: + """TransactWriteItems records to the stream like the single-item writes.""" + + def test_transacted_put_update_delete_each_record( + self, dynamodb_client, streams_client + ): + """One record per item mutation, with the event names and images.""" + name, stream_arn = _create_stream_table(dynamodb_client) + try: + # Seed the keys all three ops act on: the Put overwrites an item, so + # its record carries the seeded value as OldImage. + for key, val in (("put", "seeded"), ("upd", "before"), ("del", "doomed")): + dynamodb_client.put_item( + TableName=name, Item={"pk": {"S": key}, "v": {"S": val}} + ) + time.sleep(1) + + dynamodb_client.transact_write_items( + TransactItems=[ + { + "Put": { + "TableName": name, + "Item": {"pk": {"S": "put"}, "v": {"S": "created"}}, + } + }, + { + "Update": { + "TableName": name, + "Key": {"pk": {"S": "upd"}}, + "UpdateExpression": "SET v = :after", + "ExpressionAttributeValues": {":after": {"S": "after"}}, + } + }, + { + "Delete": { + "TableName": name, + "Key": {"pk": {"S": "del"}}, + } + }, + ] + ) + time.sleep(1) + + records = _read_all_shards(streams_client, stream_arn) + assert len(records) == 6, f"three seeded plus three transacted, got {records}" + + # The seeded INSERT and the transacted MODIFY share this key's shard, + # so the shard returns them in that order. + put_records = _for_key(records, "put") + assert [r["eventName"] for r in put_records] == ["INSERT", "MODIFY"] + overwritten = put_records[1] + assert overwritten["dynamodb"]["OldImage"]["v"]["S"] == "seeded" + assert overwritten["dynamodb"]["NewImage"]["v"]["S"] == "created" + + update_records = _for_key(records, "upd") + assert [r["eventName"] for r in update_records] == ["INSERT", "MODIFY"] + modified = update_records[1] + assert modified["dynamodb"]["OldImage"]["v"]["S"] == "before" + assert modified["dynamodb"]["NewImage"]["v"]["S"] == "after" + assert ( + modified["dynamodb"]["SequenceNumber"] + > update_records[0]["dynamodb"]["SequenceNumber"] + ) + + delete_records = _for_key(records, "del") + assert [r["eventName"] for r in delete_records] == ["INSERT", "REMOVE"] + removed = delete_records[1] + assert removed["dynamodb"]["OldImage"]["v"]["S"] == "doomed" + assert "NewImage" not in removed["dynamodb"] + + for record in records: + assert record["eventSource"] == "aws:dynamodb" + assert record["dynamodb"]["StreamViewType"] == "NEW_AND_OLD_IMAGES" + finally: + _drop(dynamodb_client, name) + + def test_two_mutations_each_record(self, dynamodb_client, streams_client): + """Two mutations on one streamed table record once each.""" + name, stream_arn = _create_stream_table(dynamodb_client) + try: + dynamodb_client.transact_write_items( + TransactItems=[ + {"Put": {"TableName": name, "Item": {"pk": {"S": "first"}}}}, + {"Put": {"TableName": name, "Item": {"pk": {"S": "second"}}}}, + ] + ) + time.sleep(1) + + records = _read_all_shards(streams_client, stream_arn) + assert len(records) == 2, f"got {records}" + assert {r["dynamodb"]["Keys"]["pk"]["S"] for r in records} == { + "first", + "second", + } + assert [r["eventName"] for r in records] == ["INSERT", "INSERT"] + finally: + _drop(dynamodb_client, name) + + def test_condition_check_records_nothing(self, dynamodb_client, streams_client): + """A ConditionCheck records nothing; the Put beside it records.""" + name, stream_arn = _create_stream_table(dynamodb_client) + try: + dynamodb_client.put_item(TableName=name, Item={"pk": {"S": "checked"}}) + time.sleep(1) + + dynamodb_client.transact_write_items( + TransactItems=[ + { + "ConditionCheck": { + "TableName": name, + "Key": {"pk": {"S": "checked"}}, + "ConditionExpression": "attribute_exists(pk)", + } + }, + {"Put": {"TableName": name, "Item": {"pk": {"S": "written"}}}}, + ] + ) + time.sleep(1) + + records = _read_all_shards(streams_client, stream_arn) + checked = _for_key(records, "checked") + assert len(checked) == 1, f"the ConditionCheck recorded: {checked}" + assert checked[0]["eventName"] == "INSERT" + written = _for_key(records, "written") + assert len(written) == 1, f"got {written}" + assert written[0]["eventName"] == "INSERT" + finally: + _drop(dynamodb_client, name) + + def test_cancelled_transaction_records_nothing( + self, dynamodb_client, streams_client + ): + """A transaction cancelled by a condition records nothing at all.""" + name, stream_arn = _create_stream_table(dynamodb_client) + try: + dynamodb_client.put_item(TableName=name, Item={"pk": {"S": "taken"}}) + time.sleep(1) + + with pytest.raises(ClientError) as excinfo: + dynamodb_client.transact_write_items( + TransactItems=[ + {"Put": {"TableName": name, "Item": {"pk": {"S": "free"}}}}, + { + "Put": { + "TableName": name, + "Item": {"pk": {"S": "taken"}}, + "ConditionExpression": "attribute_not_exists(pk)", + } + }, + ] + ) + assert ( + excinfo.value.response["Error"]["Code"] + == "TransactionCanceledException" + ) + time.sleep(1) + + records = _read_all_shards(streams_client, stream_arn) + assert len(records) == 1, f"cancelled transaction recorded: {records}" + assert records[0]["dynamodb"]["Keys"]["pk"]["S"] == "taken" + assert "Item" not in dynamodb_client.get_item( + TableName=name, Key={"pk": {"S": "free"}} + ) + finally: + _drop(dynamodb_client, name) + + def test_keys_only_stream_omits_images(self, dynamodb_client, streams_client): + """A KEYS_ONLY stream records the keys and neither image.""" + name, stream_arn = _create_stream_table(dynamodb_client, view_type="KEYS_ONLY") + try: + dynamodb_client.transact_write_items( + TransactItems=[ + { + "Put": { + "TableName": name, + "Item": {"pk": {"S": "k"}, "v": {"S": "hidden"}}, + } + } + ] + ) + time.sleep(1) + + records = _read_all_shards(streams_client, stream_arn) + assert len(records) == 1, f"got {records}" + assert records[0]["dynamodb"]["Keys"]["pk"]["S"] == "k" + assert "NewImage" not in records[0]["dynamodb"] + assert "OldImage" not in records[0]["dynamodb"] + assert records[0]["dynamodb"]["StreamViewType"] == "KEYS_ONLY" + finally: + _drop(dynamodb_client, name) From 804eea5ecb0dc3df277c92eb40f8d548a7ca6fbe Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Tue, 27 Oct 2026 09:11:51 +0000 Subject: [PATCH 2/9] fix(cassandra): run a large stream record in its own batch A transacted write on a table with a stream and an index put the stream record in the same LOGGED BATCH as the index rows. That batch runs after the base row's lightweight transaction has committed. Cassandra rejects a batch that writes more than one partition once its data passes batch_size_fail_threshold, which is 51,200 bytes unless the operator changed it. The rejection left the item written without its index rows or its record, left the ledger entry in COMMITTING, left the later operations of the transaction uncommitted, and answered 500. secondary_effect_batches now adds up the bytes of the index statements (text and bound values) and of the record statement. Up to 24,000 bytes the record joins the index batch as before. Past that, the index rows run first and the record runs after them as a batch of its own. A batch with one statement writes one partition and is not subject to the limit, so a record of any size commits. commit_secondary_effects already ran the batches in order. It now builds the index batch before calling the helper, because the size can only be read from a built batch. The budget is 24,000 bytes. On cassandra:4.1 with the default threshold, a table with one GSI and a NEW_AND_OLD_IMAGES stream took a transacted insert of 24 KB and refused one of 28 KB, and took a transacted overwrite of 16 KB and refused one of 20 KB. The index entry and the record total about 48 KB where those committed and 56 to 60 KB where they were refused, which fits the 51,200 byte limit. 24,000 is under half of the limit, so the margin covers the difference between this count and the size Cassandra measures, and a server configured with a lower threshold, which the code cannot read. The count includes the index statements because every index adds a copy of the item. An 18 KB insert on a table with two GSIs has a 19 KB record, which is under any cutoff on the record alone, and the record plus the two index entries is 57 KB. For the writes it applies to, the split means the index rows and the record are no longer one atomic unit. If the process stops after the first batch, or the second batch fails, the index rows are written and the record is not, and nothing writes the record afterwards. A storage test overwrites a 24 KB item in a transaction on a GSI table with a NEW_AND_OLD_IMAGES stream and requires success, one MODIFY record with both images, and no ledger entry. A second storage test runs an insert, an overwrite and a two-operation transaction under a client token at 16, 20, 24, 28 and 32 KB. Two more cover a transacted Update of a 20 KB item and the two-index insert above. Unit tests pin that a 30,000-byte record yields two batches with the record last, where the budget boundary lies, that bound values count toward it, and that a record with no index rows stays one batch whatever its size. --- .../src/data/transact_stream.rs | 237 +++++++++++++--- .../src/data/transactions.rs | 28 +- .../transact_write_stream_batch_limit.rs | 260 ++++++++++++++++++ .../transact_write_stream_size_sweep.rs | 105 +++++++ .../tests/direct_integration.rs | 4 + 5 files changed, 582 insertions(+), 52 deletions(-) create mode 100644 crates/storage-cassandra/tests/direct/transact_write_stream_batch_limit.rs create mode 100644 crates/storage-cassandra/tests/direct/transact_write_stream_size_sweep.rs diff --git a/crates/storage-cassandra/src/data/transact_stream.rs b/crates/storage-cassandra/src/data/transact_stream.rs index f4e101a3e..4bcff9d32 100644 --- a/crates/storage-cassandra/src/data/transact_stream.rs +++ b/crates/storage-cassandra/src/data/transact_stream.rs @@ -5,9 +5,24 @@ //! image it carries, and how its statement is batched with the index rows. //! `transactions.rs` calls both from its commit phase. -use cdrs_tokio::query::BatchQueryBuilder; +use std::collections::HashMap; + +use cdrs_tokio::frame::message_batch::{BatchQuery, BatchQuerySubj, BodyReqBatch}; +use cdrs_tokio::query::{QueryBatch, QueryValues}; +use cdrs_tokio::types::value::Value; use extenddb_core::types::Item; +/// The most bytes, counting statement text and bound values, that the index +/// rows and the stream record may total while they share one batch. +/// +/// Cassandra fails a LOGGED BATCH that writes more than one partition once its +/// data passes `batch_size_fail_threshold`, which is 50 KiB (51,200 bytes) +/// unless the operator changed it. The code does not read that setting. +/// 24,000 bytes is under half of the default, which leaves room for the +/// difference between this count and the size Cassandra measures, and for a +/// server configured with a lower threshold. +const SHARED_BATCH_BUDGET_BYTES: usize = 24_000; + /// The old image a transacted mutation's stream record carries, from the row as /// the commit phase read it. /// @@ -21,43 +36,101 @@ pub(super) fn commit_old_image(row: Option, created_to_prepare: bool) -> O if created_to_prepare { None } else { row } } -/// The batches one committed transactional mutation's secondary effects run as. +/// The batches one committed transactional mutation's secondary effects run as, +/// in the order they run. +/// +/// While the index rows and the record total at most +/// `SHARED_BATCH_BUDGET_BYTES` they share one batch, which applies all of its +/// statements or none of them. Past that the index rows run first and the +/// record runs as a batch of its own. A LOGGED BATCH that writes a single +/// partition is not subject to the size limit, so a record of any size +/// commits alone. /// -/// There is one, holding the index rows and the stream record together: a -/// reader must not see an index row for a mutation whose record was not -/// written, and a crash must not leave the index rows standing without it. The -/// caller executes exactly what this returns, so the one-batch property is -/// asserted in a unit test rather than resting on the shape of the calling -/// code, which no integration test can see. +/// The base row has already committed by the time these batches run, so a +/// batch Cassandra rejects leaves the item written without its index rows or +/// its record, and the caller answers with an error. Splitting keeps a large +/// write from reaching that state. It costs the atomicity of the index rows +/// and the record for those writes: a failure between the two batches leaves +/// the index rows written and the record unwritten, and nothing writes the +/// record afterwards. +/// +/// The caller executes exactly what this returns, in order, so the grouping is +/// asserted in unit tests rather than resting on the shape of the calling code. pub(super) fn secondary_effect_batches( - index_batch: BatchQueryBuilder, + mut index_batch: QueryBatch, stream_stmt: Option, -) -> Vec { - let batch = match stream_stmt { - Some(stmt) => { - index_batch.add_query(stmt, cdrs_tokio::query::QueryValues::SimpleValues(vec![])) - } - None => index_batch, +) -> Vec { + let Some(stmt) = stream_stmt else { + return vec![index_batch]; }; - vec![batch] + let record = BatchQuery::new( + BatchQuerySubj::QueryString(stmt), + QueryValues::SimpleValues(vec![]), + ); + let index_bytes: usize = index_batch + .request + .queries + .iter() + .map(statement_bytes) + .sum(); + if index_batch.request.queries.is_empty() + || index_bytes + statement_bytes(&record) <= SHARED_BATCH_BUDGET_BYTES + { + index_batch.request.queries.push(record); + return vec![index_batch]; + } + + let request = &index_batch.request; + let record_batch = QueryBatch::new( + BodyReqBatch::new( + request.batch_type, + vec![record], + request.consistency, + request.serial_consistency, + request.timestamp, + request.keyspace.clone(), + request.now_in_seconds, + ), + HashMap::new(), + ); + vec![index_batch, record_batch] +} + +/// What a statement adds to a batch: its text (or prepared id) and the bytes of +/// every value bound to it. +fn statement_bytes(query: &BatchQuery) -> usize { + let subject = match &query.subject { + BatchQuerySubj::QueryString(cql) => cql.len(), + BatchQuerySubj::PreparedId(id) => id.serialized_len(), + }; + let values: usize = match &query.values { + QueryValues::SimpleValues(values) => values.iter().map(value_bytes).sum(), + QueryValues::NamedValues(values) => values.values().map(value_bytes).sum(), + }; + subject + values +} + +fn value_bytes(value: &Value) -> usize { + match value { + Value::Some(bytes) => bytes.len(), + Value::Null | Value::NotSet => 0, + } } #[cfg(test)] mod tests { use super::*; + use cdrs_tokio::consistency::Consistency; + use cdrs_tokio::query::BatchQueryBuilder; use extenddb_core::types::AttributeValue; - fn built_statements(batch: BatchQueryBuilder) -> Vec { - use cdrs_tokio::frame::message_batch::BatchQuerySubj; - + fn built_statements(batch: &QueryBatch) -> Vec { batch - .build() - .unwrap() .request .queries - .into_iter() - .map(|query| match query.subject { - BatchQuerySubj::QueryString(cql) => cql.to_string(), + .iter() + .map(|query| match &query.subject { + BatchQuerySubj::QueryString(cql) => cql.clone(), BatchQuerySubj::PreparedId(_) => { panic!("secondary effects are built as CQL strings") } @@ -65,28 +138,48 @@ mod tests { .collect() } - fn index_batch() -> BatchQueryBuilder { + fn index_batch() -> QueryBatch { BatchQueryBuilder::new() + .with_consistency(Consistency::LocalQuorum) .add_query( "INSERT INTO ks.index_rows (a) VALUES (1)".to_owned(), - cdrs_tokio::query::QueryValues::SimpleValues(vec![]), + QueryValues::SimpleValues(vec![]), ) .add_query( "DELETE FROM ks.index_rows WHERE a = 2".to_owned(), - cdrs_tokio::query::QueryValues::SimpleValues(vec![]), + QueryValues::SimpleValues(vec![]), ) + .build() + .unwrap() + } + + fn batch_bytes(batch: &QueryBatch) -> usize { + batch.request.queries.iter().map(statement_bytes).sum() } const RECORD: &str = "INSERT INTO ks.stream_records (shard_id) VALUES ('s')"; + /// A record statement of exactly `len` bytes. + fn record_of_len(len: usize) -> String { + let (head, tail) = ( + "INSERT INTO ks.stream_records (record_data) VALUES ('", + "')", + ); + format!("{head}{}{tail}", "x".repeat(len - head.len() - tail.len())) + } + /// The index rows and the record commit together or not at all, which is - /// what makes a transacted record as atomic with its item as the + /// what makes a transacted record as atomic with its index rows as the /// single-item path's record is. #[test] fn index_rows_and_the_stream_record_share_one_batch() { let batches = secondary_effect_batches(index_batch(), Some(RECORD.to_owned())); - assert_eq!(batches.len(), 1, "the record must not run as its own batch"); - let statements = built_statements(batches.into_iter().next().unwrap()); + assert_eq!( + batches.len(), + 1, + "a small record must not run as its own batch" + ); + let statements = built_statements(&batches[0]); assert_eq!(statements.len(), 3, "got {statements:?}"); assert!(statements[0].contains("INSERT INTO ks.index_rows")); assert!(statements[1].contains("DELETE FROM ks.index_rows")); @@ -95,22 +188,92 @@ mod tests { #[test] fn a_table_with_no_index_runs_the_record_in_one_batch() { - let batches = secondary_effect_batches(BatchQueryBuilder::new(), Some(RECORD.to_owned())); + let empty = BatchQueryBuilder::new().build().unwrap(); + let batches = secondary_effect_batches(empty, Some(RECORD.to_owned())); assert_eq!(batches.len(), 1); - assert_eq!( - built_statements(batches.into_iter().next().unwrap()), - vec![RECORD.to_owned()] - ); + assert_eq!(built_statements(&batches[0]), vec![RECORD.to_owned()]); + } + + /// With no index rows there is nothing to split from, so a record of any + /// size is one batch holding the record alone, never an empty batch + /// followed by the record. + #[test] + fn a_large_record_with_no_index_rows_is_one_batch() { + let empty = BatchQueryBuilder::new().build().unwrap(); + let record = record_of_len(300_000); + let batches = secondary_effect_batches(empty, Some(record.clone())); + assert_eq!(batches.len(), 1); + assert_eq!(built_statements(&batches[0]), vec![record]); } #[test] fn index_rows_without_a_record_run_unchanged() { let batches = secondary_effect_batches(index_batch(), None); assert_eq!(batches.len(), 1); + assert_eq!(built_statements(&batches[0]).len(), 2); + } + + /// A record too large to share a batch with the index rows runs after + /// them, alone, with the index batch's consistency. + #[test] + fn a_30000_byte_record_runs_as_a_second_batch_after_the_index_rows() { + let record = record_of_len(30_000); + let batches = secondary_effect_batches(index_batch(), Some(record.clone())); + assert_eq!(batches.len(), 2, "the record must run as its own batch"); + + let index_statements = built_statements(&batches[0]); + assert_eq!(index_statements.len(), 2, "got {index_statements:?}"); + assert!(index_statements[0].contains("INSERT INTO ks.index_rows")); + assert!(index_statements[1].contains("DELETE FROM ks.index_rows")); + + assert_eq!( + built_statements(&batches[1]), + vec![record], + "the record is last and alone" + ); + assert_eq!(batches[1].request.consistency, Consistency::LocalQuorum); + assert_eq!(batches[1].request.batch_type, batches[0].request.batch_type); + } + + /// The record shares the batch up to the budget and runs alone one byte + /// past it, counting the index statements' own size. + #[test] + fn the_record_shares_the_batch_up_to_the_budget_and_not_past_it() { + let index_bytes = batch_bytes(&index_batch()); + + let at_budget = record_of_len(SHARED_BATCH_BUDGET_BYTES - index_bytes); + let batches = secondary_effect_batches(index_batch(), Some(at_budget)); + assert_eq!(batches.len(), 1); + assert_eq!(batch_bytes(&batches[0]), SHARED_BATCH_BUDGET_BYTES); + + let past_budget = record_of_len(SHARED_BATCH_BUDGET_BYTES - index_bytes + 1); + let batches = secondary_effect_batches(index_batch(), Some(past_budget)); + assert_eq!(batches.len(), 2); + } + + /// Values bound to the index statements count toward the budget. An async + /// GSI's queue row carries the whole new item as a bound value, so a small + /// record can still be the statement that tips a batch over. + #[test] + fn bound_values_count_toward_the_budget() { + let queue_row = BatchQueryBuilder::new() + .with_consistency(Consistency::LocalQuorum) + .add_query( + "INSERT INTO ks.gsi_pending (new_item) VALUES (?)".to_owned(), + QueryValues::SimpleValues(vec![Value::from("x".repeat(23_000).as_str())]), + ) + .build() + .unwrap(); + assert!(batch_bytes(&queue_row) > 23_000); + + let record = record_of_len(2_000); + let batches = secondary_effect_batches(queue_row, Some(record.clone())); assert_eq!( - built_statements(batches.into_iter().next().unwrap()).len(), - 2 + batches.len(), + 2, + "23,000 bound bytes and a 2,000-byte record pass the budget" ); + assert_eq!(built_statements(&batches[1]), vec![record]); } #[test] diff --git a/crates/storage-cassandra/src/data/transactions.rs b/crates/storage-cassandra/src/data/transactions.rs index 3e82a7b66..338cc9a53 100644 --- a/crates/storage-cassandra/src/data/transactions.rs +++ b/crates/storage-cassandra/src/data/transactions.rs @@ -1855,10 +1855,12 @@ impl CassandraEngine { /// index sync receives, which a put leaves as `None`; the stream record's /// own old image is already baked into `stream_stmt`. /// - /// The statements run as `secondary_effect_batches` groups them, which is - /// one batch holding the index rows and the record together. The queue row - /// an asynchronous GSI takes (`enqueue_async_indexes`) is written on the - /// session outside that batch, as it was before the record joined it. + /// The statements run as `secondary_effect_batches` groups them: one batch + /// holding the index rows and the record together, or the index batch + /// followed by the record as a batch of its own when the two are too large + /// to share one. The queue row an asynchronous GSI takes + /// (`enqueue_async_indexes`) is written on the session outside that batch, + /// as it was before the record joined it. async fn commit_secondary_effects( &self, key_info: &TableKeyInfo, @@ -1906,17 +1908,13 @@ impl CassandraEngine { .await?; } - for batch in secondary_effect_batches(batch, stream_stmt) { - self.session - .batch( - batch - .build() - .map_err(|e| StorageError::Internal(e.to_string()))?, - ) - .await - .map_err(|e| { - StorageError::Internal(format!("commit_secondary_effects batch: {e}")) - })?; + let index_batch = batch + .build() + .map_err(|e| StorageError::Internal(e.to_string()))?; + for batch in secondary_effect_batches(index_batch, stream_stmt) { + self.session.batch(batch).await.map_err(|e| { + StorageError::Internal(format!("commit_secondary_effects batch: {e}")) + })?; } if async_enqueued > 0 { diff --git a/crates/storage-cassandra/tests/direct/transact_write_stream_batch_limit.rs b/crates/storage-cassandra/tests/direct/transact_write_stream_batch_limit.rs new file mode 100644 index 000000000..025599b84 --- /dev/null +++ b/crates/storage-cassandra/tests/direct/transact_write_stream_batch_limit.rs @@ -0,0 +1,260 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Direct integration tests for the batches that carry a transacted mutation's +//! index rows and stream record. +//! +//! Cassandra fails a LOGGED BATCH over several partitions once its data passes +//! `batch_size_fail_threshold`, 50 KiB by default, and the batch runs after the +//! base row's lightweight transaction has committed. A rejection there leaves +//! the item written, writes neither the index rows nor the record, and answers +//! with an error. The record holds the new and the old image of the item, and +//! each index row or GSI queue row holds the item again, so a write of a few +//! tens of kilobytes reaches the limit. These tests run on the default +//! threshold. + +use extenddb_core::expression::ExpressionMaps; +use extenddb_core::types::{ + AttributeDefinition, AttributeValue, CreateTableInput, GsiInput, Item, KeySchemaElement, + KeyType, Projection, ProjectionType, ReturnValuesOnConditionCheckFailure, ScalarAttributeType, + StreamViewType, TableKeyInfo, +}; +use extenddb_storage::error::StorageError; +use extenddb_storage::{DataEngine, TableEngine, TransactWriteOp}; +use extenddb_storage_cassandra::CassandraEngine; + +use crate::helpers::{ + DropKeyspacesOnDrop, TestTable, ensure_test_account, setup_engine, unique_test_account, +}; +use crate::transact_write_stream_records::{ + capture, item_with, key_for, records_for_keys, set_val_action, +}; + +/// An item of about `bytes` bytes. `blob` carries the bulk, `status` and +/// `owner` are the keys of the test tables' indexes, and `val` tells two +/// versions of one item apart. +pub(crate) fn big_item(id: &str, bytes: usize, val: &str) -> Item { + let mut item = item_with(id, val); + item.insert("status".to_owned(), AttributeValue::S("s".to_owned())); + item.insert("owner".to_owned(), AttributeValue::S("o".to_owned())); + item.insert("blob".to_owned(), AttributeValue::S("x".repeat(bytes))); + item +} + +/// Write a big item without a capture, so it leaves no record. +pub(crate) async fn seed( + engine: &CassandraEngine, + key_info: &TableKeyInfo, + id: &str, + bytes: usize, + val: &str, +) { + let maps = ExpressionMaps::default(); + let item = big_item(id, bytes, val); + engine + .put_item(key_info, item, false, None, &maps, None) + .await + .unwrap(); +} + +/// One transacted Put of a big item with a `NEW_AND_OLD_IMAGES` capture. +pub(crate) async fn transacted_put( + engine: &CassandraEngine, + key_info: &TableKeyInfo, + id: &str, + bytes: usize, + val: &str, +) -> Result<(), StorageError> { + let maps = ExpressionMaps::default(); + let item = big_item(id, bytes, val); + let ops = vec![TransactWriteOp::Put { + key_info, + item: &item, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(capture(StreamViewType::NewAndOldImages)), + }]; + engine.transact_write_items(&ops, None).await +} + +/// The records of `keys` as `(sequence_number, event_name, record_data)`. +pub(crate) async fn records_of( + engine: &CassandraEngine, + key_info: &TableKeyInfo, + keys: &[&str], +) -> Vec<(String, String, serde_json::Value)> { + records_for_keys(engine, &key_info.account_id, &key_info.table_id, keys).await +} + +/// The event names recorded for `key`, in sequence order. +pub(crate) fn events_for( + records: &[(String, String, serde_json::Value)], + key: &str, +) -> Vec { + records + .iter() + .filter(|(_, _, data)| data["dynamodb"]["Keys"]["id"]["S"] == key) + .map(|(_, event, _)| event.clone()) + .collect() +} + +/// How many transactions the account's ledger still holds. A transaction that +/// committed in full deletes its entry. +pub(crate) async fn ledger_entries(engine: &CassandraEngine, key_info: &TableKeyInfo) -> usize { + let keyspace = format!("extenddb_ttl_test_account_{}", key_info.account_id); + engine + .scan_old_transactions(&keyspace, i64::MAX) + .await + .unwrap() + .len() +} + +/// A table keyed on a string `id` with two GSIs, on `status` and `owner`, that +/// each project every attribute. The guard drops the account's keyspace when +/// the test ends. +async fn two_gsi_table(engine: &CassandraEngine) -> (TableKeyInfo, DropKeyspacesOnDrop) { + let account_id = unique_test_account(); + ensure_test_account(engine, &account_id).await.unwrap(); + + let string_attribute = |name: &str| AttributeDefinition { + attribute_name: name.to_owned(), + attribute_type: ScalarAttributeType::S, + }; + let hash_key = |name: &str| { + vec![KeySchemaElement { + attribute_name: name.to_owned(), + key_type: KeyType::Hash, + }] + }; + let gsi = |index: &str, attribute: &str| GsiInput { + index_name: index.to_owned(), + key_schema: hash_key(attribute), + projection: Projection { + projection_type: ProjectionType::All, + non_key_attributes: None, + }, + provisioned_throughput: None, + }; + let attribute_definitions = vec![ + string_attribute("id"), + string_attribute("status"), + string_attribute("owner"), + ]; + let input = CreateTableInput { + table_name: "TxnStreamTwoIndexes".to_owned(), + key_schema: hash_key("id"), + attribute_definitions: attribute_definitions.clone(), + global_secondary_indexes: Some(vec![ + gsi("StatusIndex", "status"), + gsi("OwnerIndex", "owner"), + ]), + ..Default::default() + }; + let description = engine.create_table(&account_id, input).await.unwrap(); + + let key_info = TableKeyInfo { + table_name: "TxnStreamTwoIndexes".to_owned(), + account_id: account_id.clone(), + table_id: description.table_id, + key_schema: hash_key("id"), + base_key_schema: hash_key("id"), + attribute_definitions, + ..Default::default() + }; + let keyspace = format!("extenddb_ttl_test_account_{account_id}"); + (key_info, DropKeyspacesOnDrop(vec![keyspace])) +} + +/// A transacted Put that overwrites a 24 KB item on a table with a GSI and a +/// `NEW_AND_OLD_IMAGES` stream commits. The record carries both 24 KB images and +/// the index entry carries the item again, which together are over what one +/// batch may hold. +#[tokio::test] +async fn test_transact_write_put_over_a_24kb_item_on_an_indexed_table_records_modify() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::with_gsi(&engine, "TxnStreamBigPut", "StatusIndex", "status").await; + let key_info = &table.key_info; + seed(&engine, key_info, "big-1", 24_000, "before").await; + + let result = transacted_put(&engine, key_info, "big-1", 24_000, "after").await; + assert!( + result.is_ok(), + "a transacted overwrite of a 24 KB item failed: {result:?}" + ); + + let records = records_of(&engine, key_info, &["big-1"]).await; + assert_eq!(events_for(&records, "big-1"), vec!["Modify"]); + let data = &records[0].2["dynamodb"]; + assert_eq!(data["OldImage"]["val"]["S"], "before"); + assert_eq!(data["NewImage"]["val"]["S"], "after"); + assert_eq!( + ledger_entries(&engine, key_info).await, + 0, + "a committed transaction leaves no ledger entry" + ); +} + +/// A transacted Update of a 20 KB item on a table with a GSI commits and +/// records MODIFY with both images. The Update arm hands the index sync the old +/// item as well, so its index entries are larger than a Put's. +#[tokio::test] +async fn test_transact_write_update_of_a_20kb_item_on_an_indexed_table_records_modify() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::with_gsi(&engine, "TxnStreamBigUpdate", "StatusIndex", "status").await; + let key_info = &table.key_info; + seed(&engine, key_info, "upd-big", 20_000, "before").await; + + let key = key_for("upd-big"); + let (actions, update_maps) = set_val_action("after"); + let ops = vec![TransactWriteOp::Update { + key_info, + key: &key, + actions: &actions, + condition: None, + maps: &update_maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(capture(StreamViewType::NewAndOldImages)), + }]; + let result = engine.transact_write_items(&ops, None).await; + assert!( + result.is_ok(), + "a transacted update of a 20 KB item failed: {result:?}" + ); + + let records = records_of(&engine, key_info, &["upd-big"]).await; + assert_eq!(events_for(&records, "upd-big"), vec!["Modify"]); + let data = &records[0].2["dynamodb"]; + assert_eq!(data["OldImage"]["val"]["S"], "before"); + assert_eq!(data["NewImage"]["val"]["S"], "after"); + assert_eq!(ledger_entries(&engine, key_info).await, 0); +} + +/// Every index adds a copy of the item to the batch, so a table with two GSIs +/// reaches the limit at a smaller item than a table with one. An 18 KB insert +/// has a 19 KB record and two 19 KB index entries: the index entries alone fit +/// in one batch, and the record on top of them does not. +#[tokio::test] +async fn test_transact_write_insert_of_an_18kb_item_on_a_two_index_table_records_insert() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let (key_info, _keyspace_guard) = two_gsi_table(&engine).await; + + let result = transacted_put(&engine, &key_info, "two-1", 18_000, "new").await; + assert!( + result.is_ok(), + "a transacted insert of an 18 KB item failed: {result:?}" + ); + + let records = records_of(&engine, &key_info, &["two-1"]).await; + assert_eq!(events_for(&records, "two-1"), vec!["Insert"]); + assert_eq!(ledger_entries(&engine, &key_info).await, 0); +} diff --git a/crates/storage-cassandra/tests/direct/transact_write_stream_size_sweep.rs b/crates/storage-cassandra/tests/direct/transact_write_stream_size_sweep.rs new file mode 100644 index 000000000..a7bb24cbf --- /dev/null +++ b/crates/storage-cassandra/tests/direct/transact_write_stream_size_sweep.rs @@ -0,0 +1,105 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Direct integration test for the item sizes around Cassandra's batch limit. +//! +//! A table with one GSI and a `NEW_AND_OLD_IMAGES` stream takes transactions of +//! 16, 20, 24, 28 and 32 KB items. On the default `batch_size_fail_threshold` +//! (50 KiB), a transacted insert fails once the item reaches about 28 KB and a +//! transacted overwrite once it reaches about 20 KB, when the record shares the +//! batch that carries the index entry. + +use extenddb_core::expression::ExpressionMaps; +use extenddb_core::types::{ReturnValuesOnConditionCheckFailure, StreamViewType, TableKeyInfo}; +use extenddb_storage::error::StorageError; +use extenddb_storage::{DataEngine, IdempotencyKey, TransactWriteOp}; +use extenddb_storage_cassandra::CassandraEngine; + +use crate::helpers::{TestTable, setup_engine}; +use crate::transact_write_stream_batch_limit::{ + big_item, events_for, ledger_entries, records_of, seed, transacted_put, +}; +use crate::transact_write_stream_records::{capture, item_with, key_for}; + +/// One transaction with a big Put first and a small Put second, both recorded, +/// under a client token. When the first op's batch is rejected, the second op +/// is never committed and the token still answers as a replay. +async fn transacted_pair( + engine: &CassandraEngine, + key_info: &TableKeyInfo, + kb: usize, +) -> Result<(), StorageError> { + let maps = ExpressionMaps::default(); + let big = big_item(&format!("pair-big-{kb}"), kb * 1_000, "new"); + let small = item_with(&format!("pair-small-{kb}"), "tiny"); + let view = capture(StreamViewType::NewAndOldImages); + let put = |item| TransactWriteOp::Put { + key_info, + item, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(view.clone()), + }; + let ops = vec![put(&big), put(&small)]; + let token = format!("token-pair-{kb}"); + let idempotency = IdempotencyKey { + account_id: &key_info.account_id, + token: &token, + fingerprint: "fingerprint-sweep", + }; + engine.transact_write_items(&ops, Some(idempotency)).await +} + +/// An insert, an overwrite and a two-op transaction each commit with their +/// records at 16, 20, 24, 28 and 32 KB. Every size is attempted before the test +/// reports, so a failure lists each shape and size that was rejected. +#[tokio::test] +async fn test_transact_write_records_at_every_item_size_around_the_batch_limit() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::with_gsi(&engine, "TxnStreamSweep", "StatusIndex", "status").await; + let key_info = &table.key_info; + + let mut rejected = Vec::new(); + let mut expected: Vec<(String, &str)> = Vec::new(); + for kb in [16_usize, 20, 24, 28, 32] { + for (kind, event) in [("insert", "Insert"), ("overwrite", "Modify")] { + let id = format!("{kind}-{kb}"); + if kind == "overwrite" { + seed(&engine, key_info, &id, kb * 1_000, "old").await; + } + let result = transacted_put(&engine, key_info, &id, kb * 1_000, "new").await; + println!("size sweep: {kind} {kb} KB: {result:?}"); + if result.is_err() { + rejected.push(format!("{kind} {kb} KB")); + } + expected.push((id, event)); + } + + let result = transacted_pair(&engine, key_info, kb).await; + println!("size sweep: pair {kb} KB: {result:?}"); + if result.is_err() { + rejected.push(format!("pair {kb} KB")); + } + expected.push((format!("pair-big-{kb}"), "Insert")); + expected.push((format!("pair-small-{kb}"), "Insert")); + } + assert!(rejected.is_empty(), "rejected transactions: {rejected:?}"); + + let keys: Vec<&str> = expected.iter().map(|(id, _)| id.as_str()).collect(); + let records = records_of(&engine, key_info, &keys).await; + for (id, event) in &expected { + assert_eq!(events_for(&records, id), vec![*event], "records for {id}"); + } + for kb in [16_usize, 20, 24, 28, 32] { + let small = key_for(&format!("pair-small-{kb}")); + assert!( + engine.get_item(key_info, &small).await.unwrap().is_some(), + "the second op of the {kb} KB pair did not commit" + ); + } + assert_eq!(ledger_entries(&engine, key_info).await, 0); +} diff --git a/crates/storage-cassandra/tests/direct_integration.rs b/crates/storage-cassandra/tests/direct_integration.rs index b20ed60b1..9190a8332 100644 --- a/crates/storage-cassandra/tests/direct_integration.rs +++ b/crates/storage-cassandra/tests/direct_integration.rs @@ -53,10 +53,14 @@ mod table_engine; mod transact_get_items; #[path = "direct/transact_write_items.rs"] mod transact_write_items; +#[path = "direct/transact_write_stream_batch_limit.rs"] +mod transact_write_stream_batch_limit; #[path = "direct/transact_write_stream_images.rs"] mod transact_write_stream_images; #[path = "direct/transact_write_stream_records.rs"] mod transact_write_stream_records; +#[path = "direct/transact_write_stream_size_sweep.rs"] +mod transact_write_stream_size_sweep; #[path = "direct/transaction_ledger.rs"] mod transaction_ledger; #[path = "direct/users.rs"] From 8625d7c271ebda3e1e4ddc7ba3f7f483d2f6ebb0 Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Tue, 27 Oct 2026 09:18:48 +0000 Subject: [PATCH 3/9] fix(cassandra): take the stream sequence number after the item write The transacted Put, Update and Delete arms built the stream record statement before the base write. The statement takes its sequence number when it is built, so the number was taken before the item's lightweight transaction. For a Put or Update it was also taken before the catalog read for the table's indexes, and for a Delete before two or three lightweight transactions. A record numbered earlier than the write it describes can land after a record that another write on the same shard has already published, and a consumer that has read the later record can miss the earlier one. The arms now read the old image before the base write and build the statement after it: after commit_put_or_update for a Put or Update, and after check_lwt_applied for a Delete. The old image is the same row as before. The Update arm already held it from the read it uses to apply the update. On loopback against cassandra:4.1, 40 single-operation transactions (20 Puts and 20 Updates) took a sequence number whose millisecond was earlier than WRITETIME(item_data) of the stored item in 25 cases on the previous commit, 14 of the Puts and 11 of the Updates, each by 1 ms. With this change it was 0 of 40. The new storage test makes that comparison for the same 20 Puts and 20 Updates. It does not cover the Delete arm, because a deleted row has no write time to compare with. --- .../src/data/transactions.rs | 59 ++++---- .../direct/transact_write_stream_sequence.rs | 138 ++++++++++++++++++ .../tests/direct_integration.rs | 2 + 3 files changed, 174 insertions(+), 25 deletions(-) create mode 100644 crates/storage-cassandra/tests/direct/transact_write_stream_sequence.rs diff --git a/crates/storage-cassandra/src/data/transactions.rs b/crates/storage-cassandra/src/data/transactions.rs index 338cc9a53..11c6e60e2 100644 --- a/crates/storage-cassandra/src/data/transactions.rs +++ b/crates/storage-cassandra/src/data/transactions.rs @@ -1491,25 +1491,26 @@ impl CassandraEngine { stream, .. } => { - let stream_stmt = if stream.is_some() { - // The item carries the key attributes, so it resolves the - // row by itself. + // The old image is the row as it stands before the base write. + // The item carries the key attributes, so it resolves the row + // by itself. + let old_image = if stream.is_some() { let (row, created_to_prepare) = self .fetch_row_for_transaction(key_info, item, Some(txn_id)) .await?; - let old_image = commit_old_image(row, created_to_prepare); - self.transact_stream_statement( - key_info, - stream.as_ref(), - old_image.as_ref(), - Some(item), - ) + commit_old_image(row, created_to_prepare) } else { None }; let indexes = self.indexes_for_commit(key_info).await?; self.commit_put_or_update(key_info, item, txn_id_bytes.clone(), txn_timestamp) .await?; + let stream_stmt = self.transact_stream_statement( + key_info, + stream.as_ref(), + old_image.as_ref(), + Some(item), + ); self.commit_secondary_effects(key_info, None, Some(item), &indexes, stream_stmt) .await } @@ -1529,6 +1530,11 @@ impl CassandraEngine { expression::apply_update_validated(actions, &mut final_item, maps, &[], &[]) .map_err(|e| StorageError::Internal(e.to_string()))?; + let indexes = self.indexes_for_commit(key_info).await?; + self.commit_put_or_update(key_info, &final_item, txn_id_bytes, txn_timestamp) + .await?; + // `existing` was read before the base write, so it is still the + // old image. let stream_stmt = if stream.is_some() { let old_image = commit_old_image(existing.clone(), created_to_prepare); self.transact_stream_statement( @@ -1540,9 +1546,6 @@ impl CassandraEngine { } else { None }; - let indexes = self.indexes_for_commit(key_info).await?; - self.commit_put_or_update(key_info, &final_item, txn_id_bytes, txn_timestamp) - .await?; self.commit_secondary_effects( key_info, existing.as_ref(), @@ -1558,20 +1561,12 @@ impl CassandraEngine { stream, .. } => { - // A key that was already absent has no pre-image, and the record - // builder then returns no statement, so a transacted delete of a - // missing item records nothing, as the single-item delete does. - let stream_stmt = if stream.is_some() { + // The old image is the row as it stands before the delete. + let old_image = if stream.is_some() { let (row, created_to_prepare) = self .fetch_row_for_transaction(key_info, key, Some(txn_id)) .await?; - let old_image = commit_old_image(row, created_to_prepare); - self.transact_stream_statement( - key_info, - stream.as_ref(), - old_image.as_ref(), - None, - ) + commit_old_image(row, created_to_prepare) } else { None }; @@ -1667,6 +1662,15 @@ impl CassandraEngine { check_lwt_applied(&result, "commit_single_operation").map_err(|_| { StorageError::Internal("Transaction conflict during commit".to_owned()) })?; + // A key that was already absent has no pre-image, and the record + // builder then returns no statement, so a transacted delete of a + // missing item records nothing, as the single-item delete does. + let stream_stmt = self.transact_stream_statement( + key_info, + stream.as_ref(), + old_image.as_ref(), + None, + ); self.commit_secondary_effects(key_info, None, None, &[], stream_stmt) .await } @@ -1797,8 +1801,13 @@ impl CassandraEngine { /// Calls the same `stream_util::stream_record_statement` the single-item /// write paths call, with the same image pair, so a transacted write records /// the same event name, the same images for the table's view type, the same - /// shard assignment and the same sequence-number source as the equivalent + /// shard assignment and the same sequence-number clock as the equivalent /// `PutItem`, `UpdateItem` or `DeleteItem`. + /// + /// The statement takes its sequence number when it is built, so callers + /// build it once the base write has been applied. A number taken earlier + /// can be lower than the number of a record already published for a later + /// write on the same shard. fn transact_stream_statement( &self, key_info: &TableKeyInfo, diff --git a/crates/storage-cassandra/tests/direct/transact_write_stream_sequence.rs b/crates/storage-cassandra/tests/direct/transact_write_stream_sequence.rs new file mode 100644 index 000000000..9b4cdfcba --- /dev/null +++ b/crates/storage-cassandra/tests/direct/transact_write_stream_sequence.rs @@ -0,0 +1,138 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Direct integration test for when a transacted mutation takes its stream +//! sequence number. +//! +//! A sequence number starts with the millisecond at which it was taken. A +//! consumer that has read a later record of a shard can miss an +//! earlier-numbered record that lands after it, so the number has to be taken +//! once the item's write has landed. The write time of the stored item is the +//! time the lightweight transaction committed, which gives the lower bound. + +use cdrs_tokio::types::IntoRustByName; +use extenddb_core::expression::ExpressionMaps; +use extenddb_core::types::{ReturnValuesOnConditionCheckFailure, StreamViewType}; +use extenddb_storage::{DataEngine, TransactWriteOp}; +use extenddb_storage_cassandra::CassandraEngine; + +use crate::helpers::{TestTable, setup_engine}; +use crate::transact_write_stream_records::{ + capture, item_with, key_for, records_for_keys, set_val_action, +}; + +/// The millisecond at which Cassandra stored the current `item_data` of `id`. +async fn item_write_ms(engine: &CassandraEngine, table: &TestTable, id: &str) -> i64 { + let keyspace = format!("extenddb_ttl_test_account_{}", table.key_info.account_id); + let data_table = format!("items_{}", table.key_info.table_id.replace('-', "_")); + let query = + format!("SELECT WRITETIME(item_data) AS written FROM {keyspace}.{data_table} WHERE pk = ?"); + let result = engine + .session_arc() + .query_with_values(&query, cdrs_tokio::query_values!(id)) + .await + .unwrap(); + let row = result + .response_body() + .unwrap() + .into_rows() + .unwrap_or_default() + .into_iter() + .next() + .unwrap(); + let micros: i64 = row.get_r_by_name("written").unwrap(); + micros / 1_000 +} + +/// The millisecond a sequence number was taken at: its first 13 digits. +fn sequence_ms(sequence_number: &str) -> i64 { + sequence_number[..13].parse().unwrap() +} + +/// Twenty transacted Puts and twenty transacted Updates each record a sequence +/// number whose millisecond is not earlier than the millisecond at which their +/// item was written. +/// +/// Each transaction holds one op, so the gap between taking the number and +/// writing the item is the commit phase of that one op. +#[tokio::test] +async fn test_transact_write_sequence_number_is_taken_after_the_item_is_written() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnStreamSequence", false).await; + let maps = ExpressionMaps::default(); + let view = capture(StreamViewType::NewAndOldImages); + + let mut ids = Vec::new(); + for i in 0..20 { + let id = format!("seq-put-{i}"); + let item = item_with(&id, "created"); + let ops = vec![TransactWriteOp::Put { + key_info: &table.key_info, + item: &item, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(view.clone()), + }]; + engine.transact_write_items(&ops, None).await.unwrap(); + ids.push(id); + } + for i in 0..20 { + let id = format!("seq-upd-{i}"); + let item = item_with(&id, "before"); + engine + .put_item(&table.key_info, item, false, None, &maps, None) + .await + .unwrap(); + let key = key_for(&id); + let (actions, update_maps) = set_val_action("after"); + let ops = vec![TransactWriteOp::Update { + key_info: &table.key_info, + key: &key, + actions: &actions, + condition: None, + maps: &update_maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(view.clone()), + }]; + engine.transact_write_items(&ops, None).await.unwrap(); + ids.push(id); + } + + let keys: Vec<&str> = ids.iter().map(String::as_str).collect(); + let records = records_for_keys( + &engine, + &table.key_info.account_id, + &table.key_info.table_id, + &keys, + ) + .await; + assert_eq!(records.len(), ids.len(), "one record per key: {records:?}"); + + let mut early = Vec::new(); + for id in &ids { + let (sequence_number, _, _) = records + .iter() + .find(|(_, _, data)| data["dynamodb"]["Keys"]["id"]["S"] == id.as_str()) + .unwrap(); + let taken = sequence_ms(sequence_number); + let written = item_write_ms(&engine, &table, id).await; + if taken < written { + early.push(format!("{id}: {} ms early", written - taken)); + } + } + println!( + "sequence number earlier than the item's write in {} of {}", + early.len(), + ids.len() + ); + assert!( + early.is_empty(), + "{} of {} sequence numbers were taken before the item was written: {early:?}", + early.len(), + ids.len() + ); +} diff --git a/crates/storage-cassandra/tests/direct_integration.rs b/crates/storage-cassandra/tests/direct_integration.rs index 9190a8332..d7f3d360d 100644 --- a/crates/storage-cassandra/tests/direct_integration.rs +++ b/crates/storage-cassandra/tests/direct_integration.rs @@ -59,6 +59,8 @@ mod transact_write_stream_batch_limit; mod transact_write_stream_images; #[path = "direct/transact_write_stream_records.rs"] mod transact_write_stream_records; +#[path = "direct/transact_write_stream_sequence.rs"] +mod transact_write_stream_sequence; #[path = "direct/transact_write_stream_size_sweep.rs"] mod transact_write_stream_size_sweep; #[path = "direct/transaction_ledger.rs"] From c80bac2607963bf583c88156e0a7a59071813898 Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Tue, 27 Oct 2026 09:22:36 +0000 Subject: [PATCH 4/9] test(cassandra): a cancelled transaction keeps a committed item Committing a transacted Put or Update clears created_to_prepare as well as prepared_txn_id. The doc comment on commit_put_or_update gave a different reason, that a flag left set would make the next transacted write on the key record an insert over a committed item. The commit-timestamp rule in fetch_row_for_transaction already tells a committed row from a placeholder when the flag is set, so the clearing does not change which old image a record carries. The clearing affects rollback. rollback_single_operation and recover_rollback_op run DELETE .. IF prepared_txn_id = ? AND created_to_prepare = true before they clear prepared_txn_id. A committed row that kept the flag matches that DELETE when a later transaction that prepared the key is cancelled, and the committed item is deleted. The doc comment now says this. It also says that rows committed before the clearing existed still carry the flag and are still deleted this way. The new storage test commits an item through a transaction, then runs a transaction that overwrites the key and carries a condition on another key that fails. It requires the cancelled transaction to leave the first transaction's value in place and the other key unwritten. With the clearing taken out of both commit statements, get_item returns nothing after the cancelled transaction. --- .../src/data/transactions.rs | 22 +++-- .../transact_write_rollback_after_commit.rs | 98 +++++++++++++++++++ .../tests/direct_integration.rs | 2 + 3 files changed, 114 insertions(+), 8 deletions(-) create mode 100644 crates/storage-cassandra/tests/direct/transact_write_rollback_after_commit.rs diff --git a/crates/storage-cassandra/src/data/transactions.rs b/crates/storage-cassandra/src/data/transactions.rs index 11c6e60e2..9f945fa7c 100644 --- a/crates/storage-cassandra/src/data/transactions.rs +++ b/crates/storage-cassandra/src/data/transactions.rs @@ -1740,14 +1740,20 @@ impl CassandraEngine { /// Helper to commit a Put or Update operation. /// - /// Clears `created_to_prepare` as well as `prepared_txn_id`: the column - /// marks a row that exists only because a PREPARE planted it, and the - /// commit phase reads it to decide whether the mutation has an old image - /// (see `commit_old_image`). Left set, it would make the next transacted - /// write on this key record an insert over a committed item. Rows committed - /// by a release before this clearing still carry the flag; the commit-time - /// read tells them apart by `last_committed_txn_timestamp`, which only a - /// commit writes. + /// Clears `created_to_prepare` as well as `prepared_txn_id`. The column + /// marks a row that exists only because a PREPARE planted it, and rollback + /// deletes such a row: `rollback_single_operation` and `recover_rollback_op` + /// run `DELETE .. IF prepared_txn_id = ? AND created_to_prepare = true` + /// before they fall back to clearing `prepared_txn_id`. A committed row that + /// kept the flag would match that DELETE when a later transaction that + /// touched the key was cancelled, and the committed item would be lost. + /// Clearing the flag here makes the DELETE skip the row. + /// + /// Rows committed before this clearing existed still carry the flag, and a + /// cancelled transaction still deletes them. The commit phase reads the + /// flag too, to decide whether a mutation has an old image (see + /// `commit_old_image`), and tells those rows apart by + /// `last_committed_txn_timestamp`, which only a commit writes. async fn commit_put_or_update( &self, key_info: &TableKeyInfo, diff --git a/crates/storage-cassandra/tests/direct/transact_write_rollback_after_commit.rs b/crates/storage-cassandra/tests/direct/transact_write_rollback_after_commit.rs new file mode 100644 index 000000000..df71dfc44 --- /dev/null +++ b/crates/storage-cassandra/tests/direct/transact_write_rollback_after_commit.rs @@ -0,0 +1,98 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Direct integration test for a cancelled transaction that touches a key an +//! earlier transaction committed. +//! +//! A committing Put or Update clears `created_to_prepare`. Rollback deletes a +//! row that still carries the flag and has `prepared_txn_id` set to the +//! cancelled transaction, so a committed row that kept the flag loses its item +//! when a later transaction that prepared the key is cancelled. + +use extenddb_core::expression::{Expr, ExpressionMaps, PathElement}; +use extenddb_core::types::{AttributeValue, ReturnValuesOnConditionCheckFailure}; +use extenddb_storage::error::StorageError; +use extenddb_storage::{DataEngine, TransactWriteOp}; + +use crate::helpers::{TestTable, setup_engine}; +use crate::transact_write_stream_records::{item_with, key_for}; + +/// `attribute_exists(id)`, which fails against a key that is absent. +fn attribute_exists_id() -> Expr { + Expr::Function { + name: "attribute_exists".to_owned(), + args: vec![Expr::Path(vec![PathElement::Attribute("id".to_owned())])], + } +} + +/// A transaction commits an item. A second transaction overwrites the same key +/// and carries a condition on another key that fails, so it prepares the key, +/// is cancelled, and rolls back. The item the first transaction committed is +/// still stored afterwards, with its first value, and the other key was never +/// written. +#[tokio::test] +async fn test_transact_write_cancelled_transaction_keeps_an_item_an_earlier_transaction_committed() +{ + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnRollbackKeepsCommitted", false).await; + let maps = ExpressionMaps::default(); + + let committed = item_with("aaa-committed", "from-the-first-transaction"); + let first = vec![TransactWriteOp::Put { + key_info: &table.key_info, + item: &committed, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: None, + }]; + engine.transact_write_items(&first, None).await.unwrap(); + + let overwrite = item_with("aaa-committed", "from-the-cancelled-transaction"); + let absent = item_with("zzz-absent", "never-written"); + let exists = attribute_exists_id(); + let second = vec![ + TransactWriteOp::Put { + key_info: &table.key_info, + item: &overwrite, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: None, + }, + TransactWriteOp::Put { + key_info: &table.key_info, + item: &absent, + condition: Some(&exists), + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: None, + }, + ]; + let result = engine.transact_write_items(&second, None).await; + assert!( + matches!(result, Err(StorageError::TransactionCanceled(_))), + "expected cancellation, got {result:?}" + ); + + let stored = engine + .get_item(&table.key_info, &key_for("aaa-committed")) + .await + .unwrap(); + assert_eq!( + stored.as_ref().and_then(|item| item.get("val")), + Some(&AttributeValue::S("from-the-first-transaction".to_owned())), + "the cancelled transaction changed or deleted the committed item: {stored:?}" + ); + assert!( + engine + .get_item(&table.key_info, &key_for("zzz-absent")) + .await + .unwrap() + .is_none(), + "the cancelled transaction wrote a second item" + ); +} diff --git a/crates/storage-cassandra/tests/direct_integration.rs b/crates/storage-cassandra/tests/direct_integration.rs index d7f3d360d..8c8d247ec 100644 --- a/crates/storage-cassandra/tests/direct_integration.rs +++ b/crates/storage-cassandra/tests/direct_integration.rs @@ -53,6 +53,8 @@ mod table_engine; mod transact_get_items; #[path = "direct/transact_write_items.rs"] mod transact_write_items; +#[path = "direct/transact_write_rollback_after_commit.rs"] +mod transact_write_rollback_after_commit; #[path = "direct/transact_write_stream_batch_limit.rs"] mod transact_write_stream_batch_limit; #[path = "direct/transact_write_stream_images.rs"] From 4c8254aa5620803f6271a3054b1eab2a76c0476f Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Tue, 27 Oct 2026 09:31:25 +0000 Subject: [PATCH 5/9] test(cassandra): stream records on sort-key and indexed tables The scenario in test_transact_write_records_one_stream_record_per_mutation, a Put, an Update and a Delete in one transaction, now lives in a function that takes the table and the builders for its items and keys. The original test calls it on a table with a hash key only. New tests call it on tables with a string, a number and a binary sort key, checking that every record carries both key attributes, and on a table with a GSI pinned to synchronous propagation, checking that the Put's index row is written. Two more tests pin that the index rows and the record of a small item are one batch. In the first, the index's table has been removed, so the server rejects the index statement. The transaction fails with an error from the secondary-effects batch and no record exists. In the second, the account's stream_records table has been removed, so the server rejects the record statement. The transaction fails the same way and the index scan holds no row. The first test fails if the record can be written when the index statement is rejected, and the second if the index rows can be written when the record statement is rejected. --- .../direct/transact_write_stream_records.rs | 30 ++- .../direct/transact_write_stream_tables.rs | 231 ++++++++++++++++++ .../tests/direct_integration.rs | 2 + 3 files changed, 251 insertions(+), 12 deletions(-) create mode 100644 crates/storage-cassandra/tests/direct/transact_write_stream_tables.rs diff --git a/crates/storage-cassandra/tests/direct/transact_write_stream_records.rs b/crates/storage-cassandra/tests/direct/transact_write_stream_records.rs index 26951bcd5..1c0623f00 100644 --- a/crates/storage-cassandra/tests/direct/transact_write_stream_records.rs +++ b/crates/storage-cassandra/tests/direct/transact_write_stream_records.rs @@ -122,27 +122,32 @@ async fn test_transact_write_records_one_stream_record_per_mutation() { } let engine = setup_engine().await; let table = TestTable::new(&engine, "TxnStreamAllVerbs", false).await; + one_stream_record_per_mutation(&engine, &table, item_with, key_for).await; +} + +/// The scenario of the test above on a table of any key shape. `item` and `key` +/// build the item and the key for an id, with whatever sort key or index +/// attribute the table needs. Returns the records the transaction wrote. +pub(crate) async fn one_stream_record_per_mutation( + engine: &CassandraEngine, + table: &TestTable, + item: impl Fn(&str, &str) -> Item, + key: impl Fn(&str) -> Item, +) -> Vec<(String, String, serde_json::Value)> { let maps = ExpressionMaps::default(); // Seed the keys the Update and the Delete act on, without a capture, so the // only records in the table are the transaction's. for (id, val) in [("upd-1", "before"), ("del-1", "doomed")] { engine - .put_item( - &table.key_info, - item_with(id, val), - false, - None, - &maps, - None, - ) + .put_item(&table.key_info, item(id, val), false, None, &maps, None) .await .unwrap(); } - let put_item = item_with("put-1", "created"); - let update_key = key_for("upd-1"); - let delete_key = key_for("del-1"); + let put_item = item("put-1", "created"); + let update_key = key("upd-1"); + let delete_key = key("del-1"); let (actions, update_maps) = set_val_action("after"); let view = capture(StreamViewType::NewAndOldImages); @@ -177,7 +182,7 @@ async fn test_transact_write_records_one_stream_record_per_mutation() { engine.transact_write_items(&ops, None).await.unwrap(); let records = records_for_keys( - &engine, + engine, &table.key_info.account_id, &table.key_info.table_id, &["put-1", "upd-1", "del-1"], @@ -219,6 +224,7 @@ async fn test_transact_write_records_one_stream_record_per_mutation() { "record carries the capture's view type" ); } + records } /// Two mutations on the same streamed table record two records, ordered as the diff --git a/crates/storage-cassandra/tests/direct/transact_write_stream_tables.rs b/crates/storage-cassandra/tests/direct/transact_write_stream_tables.rs new file mode 100644 index 000000000..63eb21fec --- /dev/null +++ b/crates/storage-cassandra/tests/direct/transact_write_stream_tables.rs @@ -0,0 +1,231 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Direct integration tests for stream records on tables with a sort key or an +//! index, and for the batch that carries a record with its index rows. + +use extenddb_core::expression::ExpressionMaps; +use extenddb_core::types::{ + AttributeValue, Item, ReturnValuesOnConditionCheckFailure, ScalarAttributeType, StreamViewType, +}; +use extenddb_storage::error::StorageError; +use extenddb_storage::{DataEngine, TransactWriteOp}; +use extenddb_storage_cassandra::CassandraEngine; +use extenddb_storage_cassandra::data::index::fetch_indexes_for_table; + +use crate::helpers::{TestTable, setup_engine}; +use crate::transact_write_stream_records::{ + capture, item_with, key_for, one_stream_record_per_mutation, records_for_keys, +}; + +/// `item` with the sort key `sort`. +fn with_sort(mut item: Item, sort: &AttributeValue) -> Item { + item.insert("sort".to_owned(), sort.clone()); + item +} + +/// `item` with the attribute the test tables' GSI is keyed on. +fn with_status(mut item: Item) -> Item { + item.insert("status".to_owned(), AttributeValue::S("open".to_owned())); + item +} + +/// A transaction with a Put, an Update and a Delete records one record per +/// mutation on tables with a string, a number and a binary sort key, and every +/// record carries both key attributes. +#[tokio::test] +async fn test_transact_write_records_one_stream_record_per_mutation_with_a_sort_key() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let sorts = [ + ( + "TxnStreamSortS", + ScalarAttributeType::S, + AttributeValue::S("s1".to_owned()), + ), + ( + "TxnStreamSortN", + ScalarAttributeType::N, + AttributeValue::N("1".to_owned()), + ), + ( + "TxnStreamSortB", + ScalarAttributeType::B, + AttributeValue::B(vec![1, 2, 3, 4]), + ), + ]; + for (name, sort_type, sort) in sorts { + let table = TestTable::with_sort_key_type(&engine, name, sort_type).await; + let records = one_stream_record_per_mutation( + &engine, + &table, + |id, val| with_sort(item_with(id, val), &sort), + |id| with_sort(key_for(id), &sort), + ) + .await; + for (_, _, record) in &records { + let keys = record["dynamodb"]["Keys"].as_object().unwrap(); + assert!( + keys.len() == 2 && keys.contains_key("id") && keys.contains_key("sort"), + "{name}: the record's keys are {keys:?}" + ); + } + } +} + +/// Pin the table's GSI to synchronous propagation, the way a delay of 0 does in +/// production, so its index row goes into the same batch as the record. +async fn pin_gsi_sync(engine: &CassandraEngine, table: &TestTable) { + let pin = format!( + "UPDATE {}.indexes SET propagation_delay_ms = 0 \ + WHERE table_id = '{}' AND index_name = 'StatusIndex'", + engine.catalog_keyspace(), + table.key_info.table_id + ); + engine.session_arc().query(pin).await.unwrap(); +} + +/// A transaction with a Put, an Update and a Delete records one record per +/// mutation on a table with a GSI, and the Put writes its index row in the batch +/// that carries its record. +#[tokio::test] +async fn test_transact_write_records_one_stream_record_per_mutation_with_a_gsi() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::with_gsi(&engine, "TxnStreamAllVerbsGsi", "StatusIndex", "status").await; + pin_gsi_sync(&engine, &table).await; + + one_stream_record_per_mutation( + &engine, + &table, + |id, val| with_status(item_with(id, val)), + key_for, + ) + .await; + + let (rows, _) = engine + .scan(&table.key_info, None, None, None, None, Some("StatusIndex")) + .await + .unwrap(); + assert!( + rows.iter() + .any(|row| row.get("id") == Some(&AttributeValue::S("put-1".to_owned()))), + "no index row for the Put: {rows:?}" + ); +} + +/// One transacted Put of a small item with a status, so it has an index row, +/// and a `NEW_AND_OLD_IMAGES` capture. +async fn transacted_put_with_status( + engine: &CassandraEngine, + table: &TestTable, + id: &str, +) -> Result<(), StorageError> { + let maps = ExpressionMaps::default(); + let item = with_status(item_with(id, "small")); + let ops = vec![TransactWriteOp::Put { + key_info: &table.key_info, + item: &item, + condition: None, + maps: &maps, + return_values_on_ccf: ReturnValuesOnConditionCheckFailure::None, + stream: Some(capture(StreamViewType::NewAndOldImages)), + }]; + engine.transact_write_items(&ops, None).await +} + +/// The error text of a failed transaction, which has to come from the batch +/// that carries the secondary effects. +fn batch_error(result: Result<(), StorageError>) -> String { + match result { + Err(StorageError::Internal(message)) => { + assert!( + message.contains("commit_secondary_effects batch"), + "the error did not come from the secondary-effects batch: {message}" + ); + message + } + other => panic!("expected an Internal error from the batch, got {other:?}"), + } +} + +/// A transacted Put whose index statement is rejected writes no stream record. +/// +/// The record of a small item shares the batch with the index rows, and +/// Cassandra applies none of the statements of a batch it rejects. The index's +/// table is removed so the server rejects the index statement. +#[tokio::test] +async fn test_transact_write_failing_index_statement_prevents_the_record() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = + TestTable::with_gsi(&engine, "TxnStreamFailingIndex", "StatusIndex", "status").await; + pin_gsi_sync(&engine, &table).await; + let indexes = fetch_indexes_for_table( + &table.key_info.table_id, + &engine.session_arc(), + &engine.catalog_keyspace(), + ) + .await + .unwrap(); + let keyspace = engine.account_keyspace(&table.key_info.account_id); + engine + .drop_index_data_table(&keyspace, &indexes[0].index_id) + .await + .unwrap(); + + let result = transacted_put_with_status(&engine, &table, "fail-idx").await; + batch_error(result); + + let records = records_for_keys( + &engine, + &table.key_info.account_id, + &table.key_info.table_id, + &["fail-idx"], + ) + .await; + assert!( + records.is_empty(), + "the index statement failed, so no record may be written: {records:?}" + ); +} + +/// A transacted Put whose record statement is rejected writes no index row. +/// +/// The same property from the other side: the index rows and the record of a +/// small item are one batch. The account's `stream_records` table is removed so +/// the server rejects the record statement. +#[tokio::test] +async fn test_transact_write_failing_record_statement_prevents_the_index_rows() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = + TestTable::with_gsi(&engine, "TxnStreamFailingRecord", "StatusIndex", "status").await; + pin_gsi_sync(&engine, &table).await; + let keyspace = engine.account_keyspace(&table.key_info.account_id); + engine + .session_arc() + .query(format!("DROP TABLE {keyspace}.stream_records")) + .await + .unwrap(); + + let result = transacted_put_with_status(&engine, &table, "fail-rec").await; + batch_error(result); + + let (rows, _) = engine + .scan(&table.key_info, None, None, None, None, Some("StatusIndex")) + .await + .unwrap(); + assert!( + rows.is_empty(), + "the record statement failed, so no index row may be written: {rows:?}" + ); +} diff --git a/crates/storage-cassandra/tests/direct_integration.rs b/crates/storage-cassandra/tests/direct_integration.rs index 8c8d247ec..39767bc49 100644 --- a/crates/storage-cassandra/tests/direct_integration.rs +++ b/crates/storage-cassandra/tests/direct_integration.rs @@ -65,6 +65,8 @@ mod transact_write_stream_records; mod transact_write_stream_sequence; #[path = "direct/transact_write_stream_size_sweep.rs"] mod transact_write_stream_size_sweep; +#[path = "direct/transact_write_stream_tables.rs"] +mod transact_write_stream_tables; #[path = "direct/transaction_ledger.rs"] mod transaction_ledger; #[path = "direct/users.rs"] From 290f7ffde7ebce257519cad2b172d408d3a4fe84 Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Tue, 27 Oct 2026 09:38:41 +0000 Subject: [PATCH 6/9] docs(cassandra): correct comments on the stream record batch commit_secondary_effects said the queue row an asynchronous GSI takes is written outside the batch. enqueue_async_indexes adds it to the same batch as the index rows, and its own doc comment says so. The same comment called the window after the base LWT brief and compared it to the one async GSIs accept. An async GSI row converges through its queue. An index row or a record lost after the base LWT is not written again, because recovery re-runs the base mutation only. The comment now says that, and says the queue row goes into the index batch. indexes_for_commit said occ_write reads the indexes before its own base write. occ_write receives them from its callers, which read them once at entry, so the sentence is dropped. The planted-row helper in the images test described a row with no prepared_txn_id and called it the shape a single-item write leaves. PREPARE sets prepared_txn_id, and no single-item path writes created_to_prepare. The comment now describes the row PREPARE plants and what the helper writes. The comment in the transact_write_items engine handler said stream records are captured atomically within the storage transaction. That holds for PostgreSQL, SQLite and MongoDB, and not for Cassandra, and the comment now says which. The debug_assert_eq on pre_commit_images is removed. The loop above it pushes one image per op in every arm, so the assertion cannot fail. A docstring in the Python test now says "an extenddb endpoint". --- crates/engine/src/transact_write_items.rs | 5 ++++- .../storage-cassandra/src/data/transactions.rs | 18 +++++------------- .../direct/transact_write_stream_images.rs | 13 +++++++------ tests/python/test_streams_transactions.py | 2 +- 4 files changed, 17 insertions(+), 21 deletions(-) diff --git a/crates/engine/src/transact_write_items.rs b/crates/engine/src/transact_write_items.rs index 43fb78a1c..069dea418 100755 --- a/crates/engine/src/transact_write_items.rs +++ b/crates/engine/src/transact_write_items.rs @@ -209,7 +209,10 @@ pub async fn handle_transact_write_items( Err(e) => return Err(storage_err_to_dynamo(e)), } - // Stream records are now captured atomically within the storage transaction. + // The PostgreSQL, SQLite and MongoDB backends write stream records inside + // the storage transaction that writes the items. The Cassandra backend + // writes each record in a batch that runs after its item's write, so a + // failure there leaves the item without its record. // Per-item WCU: round each item individually, then sum (M-1). let mut per_table_wcu: HashMap = HashMap::new(); diff --git a/crates/storage-cassandra/src/data/transactions.rs b/crates/storage-cassandra/src/data/transactions.rs index 9f945fa7c..7d505ee37 100644 --- a/crates/storage-cassandra/src/data/transactions.rs +++ b/crates/storage-cassandra/src/data/transactions.rs @@ -911,11 +911,6 @@ impl CassandraEngine { .await?; } - debug_assert_eq!( - pre_commit_images.len(), - ops.len(), - "one TTL pre-commit image per op, in request order" - ); for (op, old_image) in ops.iter().zip(&pre_commit_images) { match op { TransactWriteOp::Put { key_info, item, .. } => { @@ -1840,8 +1835,6 @@ impl CassandraEngine { /// the base LWT would make every record and index row wait a round trip /// behind an item that has already committed, widening the window in which /// a crash leaves the item mutated and its secondary effects unwritten. - /// `occ_write` reads the indexes before its own base write for the same - /// reason. async fn indexes_for_commit( &self, key_info: &TableKeyInfo, @@ -1861,9 +1854,9 @@ impl CassandraEngine { /// statements to be batched with writes to other partitions. The single-item /// paths have the same two-step shape for the same reason: `occ_write` /// commits the base row by LWT, then carries the index rows and the stream - /// record in one LOGGED BATCH. The brief inconsistency window is acceptable - /// for the same reason async GSIs accept it; transaction recovery re-runs the - /// base mutation on crash. + /// record in one LOGGED BATCH. A failure or crash after the base LWT leaves + /// the index rows and the record unwritten; neither a retry nor recovery + /// writes them, because recovery re-runs the base mutation only. /// /// `new_item` is `None` for a delete, which writes no index rows on this /// path, and `indexes` is then empty. `index_old_item` is the old image the @@ -1873,9 +1866,8 @@ impl CassandraEngine { /// The statements run as `secondary_effect_batches` groups them: one batch /// holding the index rows and the record together, or the index batch /// followed by the record as a batch of its own when the two are too large - /// to share one. The queue row an asynchronous GSI takes - /// (`enqueue_async_indexes`) is written on the session outside that batch, - /// as it was before the record joined it. + /// to share one. The queue row that `enqueue_async_indexes` adds for an + /// asynchronous GSI goes into the index batch with the index rows. async fn commit_secondary_effects( &self, key_info: &TableKeyInfo, diff --git a/crates/storage-cassandra/tests/direct/transact_write_stream_images.rs b/crates/storage-cassandra/tests/direct/transact_write_stream_images.rs index 7c3d94305..892b64c8b 100644 --- a/crates/storage-cassandra/tests/direct/transact_write_stream_images.rs +++ b/crates/storage-cassandra/tests/direct/transact_write_stream_images.rs @@ -33,14 +33,15 @@ async fn transacted_put(engine: &CassandraEngine, table: &TestTable, item: &Item engine.transact_write_items(&ops, None).await.unwrap(); } -/// Write the row shape a PREPARE leaves behind: `created_to_prepare` set, with -/// `item_data` carrying a placeholder rather than a committed item, and no -/// `prepared_txn_id`. +/// Write a row shaped like the one a PREPARE plants for an absent key: +/// `created_to_prepare` set and `item_data` holding a placeholder rather than a +/// committed item, with no `last_committed_txn_timestamp`. The transaction under +/// test then PREPAREs this row like any other, which sets `prepared_txn_id`. /// /// A commit-time read of `item_data` alone cannot tell this row from a row -/// holding a committed item, and PREPARE's own read cannot tell it either. It -/// is also the shape a single-item write that commits inside the PREPARE window -/// leaves, which is why the record's old image is decided from the flag. +/// holding a committed item, and PREPARE's own read cannot tell it either. The +/// record's old image is decided from the flag and the commit timestamp +/// instead. async fn plant_prepare_created_row(engine: &CassandraEngine, table: &TestTable, item: &Item) { let keyspace = format!("extenddb_ttl_test_account_{}", table.key_info.account_id); let data_table = format!("items_{}", table.key_info.table_id.replace('-', "_")); diff --git a/tests/python/test_streams_transactions.py b/tests/python/test_streams_transactions.py index b9699a13b..b6e217dbd 100644 --- a/tests/python/test_streams_transactions.py +++ b/tests/python/test_streams_transactions.py @@ -11,7 +11,7 @@ sequence numbers there, while records for different keys land in different shards, whose sequence numbers are not comparable. -These tests require a extenddb endpoint (DYNAMODB_ENDPOINT) because real +These tests require an extenddb endpoint (DYNAMODB_ENDPOINT) because real DynamoDB uses a separate Streams endpoint. """ From 11aa297fd320e48076db04bd1e99645ed0c16e96 Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Tue, 27 Oct 2026 12:35:48 +0000 Subject: [PATCH 7/9] docs(cassandra): say what the stream record split does not cover The doc comment on secondary_effect_batches said that splitting the record into its own batch keeps a large write from reaching the state where the item is written without its index rows or its record. The split returns the index batch unchanged. Cassandra refuses any batch that writes more than one partition once its data passes batch_size_fail_threshold, 50 KiB by default, whatever statements the batch holds. On two synchronous GSIs, or one synchronous and one asynchronous GSI, the index rows alone pass that limit from an item of about 26 KB, with or without a stream, and the batch is refused after the base row's lightweight transaction has committed. The comment now says that the split keeps the record from being the statement that pushes a batch over the limit, that it does not bound the index batch, and when the item is written without its index rows. The sentence about the lost atomicity said "It" after the replaced sentence, so it now names the split. No code changes. --- .../storage-cassandra/src/data/transact_stream.rs | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/crates/storage-cassandra/src/data/transact_stream.rs b/crates/storage-cassandra/src/data/transact_stream.rs index 4bcff9d32..1c2c173bd 100644 --- a/crates/storage-cassandra/src/data/transact_stream.rs +++ b/crates/storage-cassandra/src/data/transact_stream.rs @@ -48,11 +48,15 @@ pub(super) fn commit_old_image(row: Option, created_to_prepare: bool) -> O /// /// The base row has already committed by the time these batches run, so a /// batch Cassandra rejects leaves the item written without its index rows or -/// its record, and the caller answers with an error. Splitting keeps a large -/// write from reaching that state. It costs the atomicity of the index rows -/// and the record for those writes: a failure between the two batches leaves -/// the index rows written and the record unwritten, and nothing writes the -/// record afterwards. +/// its record, and the caller answers with an error. Splitting keeps the +/// record from being the statement that pushes a batch over the limit. It does +/// not bound the index batch, which is sent whole: when the index rows write +/// more than one partition and pass the limit by themselves (two synchronous +/// GSIs, or one synchronous and one asynchronous GSI, from about 26 KB), the +/// item is written without them. The split costs the atomicity of the index +/// rows and the record for a record too large to share their batch: a failure +/// between the two batches leaves the index rows written and the record +/// unwritten, and nothing writes the record afterwards. /// /// The caller executes exactly what this returns, in order, so the grouping is /// asserted in unit tests rather than resting on the shape of the calling code. From 938116b3b1b7cdfa4cd567040219c76ea584e602 Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Tue, 27 Oct 2026 12:41:53 +0000 Subject: [PATCH 8/9] test(cassandra): a cancelled transaction keeps a sort-key item commit_put_or_update clears created_to_prepare in two statements, one for a table with a sort key and one for a table with a hash key only. The cancelled-transaction test created its table with no sort key, so every commit in it went through the hash-key statement and no test ran the sort-key statement's clearing. A committed sort-key row that kept the flag matches the rollback DELETE of a later cancelled transaction that prepared the key, and the committed item is deleted. The scenario now lives in a function that takes the table and the builders for its items and keys, and the existing test calls it on the hash-key table. A second test calls it on a table with a string sort key, with the sort-key item and key builders of the stream tables tests, whose with_sort helper is now visible to the other test modules. It requires the item the first transaction committed to be stored with its first value after a second transaction that prepared the same key is cancelled, and the second transaction's other key to be unwritten. --- .../transact_write_rollback_after_commit.rs | 70 ++++++++++++++----- .../direct/transact_write_stream_tables.rs | 2 +- 2 files changed, 55 insertions(+), 17 deletions(-) diff --git a/crates/storage-cassandra/tests/direct/transact_write_rollback_after_commit.rs b/crates/storage-cassandra/tests/direct/transact_write_rollback_after_commit.rs index df71dfc44..8831ed263 100644 --- a/crates/storage-cassandra/tests/direct/transact_write_rollback_after_commit.rs +++ b/crates/storage-cassandra/tests/direct/transact_write_rollback_after_commit.rs @@ -1,21 +1,27 @@ // Copyright 2026 ExtendDB contributors // SPDX-License-Identifier: Apache-2.0 -//! Direct integration test for a cancelled transaction that touches a key an +//! Direct integration tests for a cancelled transaction that touches a key an //! earlier transaction committed. //! //! A committing Put or Update clears `created_to_prepare`. Rollback deletes a //! row that still carries the flag and has `prepared_txn_id` set to the //! cancelled transaction, so a committed row that kept the flag loses its item //! when a later transaction that prepared the key is cancelled. +//! +//! `commit_put_or_update` clears the flag in one statement for a table with a +//! sort key and in another for a table without one, so the scenario runs on +//! both. use extenddb_core::expression::{Expr, ExpressionMaps, PathElement}; -use extenddb_core::types::{AttributeValue, ReturnValuesOnConditionCheckFailure}; +use extenddb_core::types::{AttributeValue, Item, ReturnValuesOnConditionCheckFailure}; use extenddb_storage::error::StorageError; use extenddb_storage::{DataEngine, TransactWriteOp}; +use extenddb_storage_cassandra::CassandraEngine; use crate::helpers::{TestTable, setup_engine}; use crate::transact_write_stream_records::{item_with, key_for}; +use crate::transact_write_stream_tables::with_sort; /// `attribute_exists(id)`, which fails against a key that is absent. fn attribute_exists_id() -> Expr { @@ -30,17 +36,18 @@ fn attribute_exists_id() -> Expr { /// is cancelled, and rolls back. The item the first transaction committed is /// still stored afterwards, with its first value, and the other key was never /// written. -#[tokio::test] -async fn test_transact_write_cancelled_transaction_keeps_an_item_an_earlier_transaction_committed() -{ - if crate::helpers::skip_without_cassandra() { - return; - } - let engine = setup_engine().await; - let table = TestTable::new(&engine, "TxnRollbackKeepsCommitted", false).await; +/// +/// `item` and `key` build the item and the key for an id, with the sort key the +/// table needs, so the scenario runs on a table of either key shape. +async fn cancelled_transaction_keeps_a_committed_item( + engine: &CassandraEngine, + table: &TestTable, + item: impl Fn(&str, &str) -> Item, + key: impl Fn(&str) -> Item, +) { let maps = ExpressionMaps::default(); - let committed = item_with("aaa-committed", "from-the-first-transaction"); + let committed = item("aaa-committed", "from-the-first-transaction"); let first = vec![TransactWriteOp::Put { key_info: &table.key_info, item: &committed, @@ -51,8 +58,8 @@ async fn test_transact_write_cancelled_transaction_keeps_an_item_an_earlier_tran }]; engine.transact_write_items(&first, None).await.unwrap(); - let overwrite = item_with("aaa-committed", "from-the-cancelled-transaction"); - let absent = item_with("zzz-absent", "never-written"); + let overwrite = item("aaa-committed", "from-the-cancelled-transaction"); + let absent = item("zzz-absent", "never-written"); let exists = attribute_exists_id(); let second = vec![ TransactWriteOp::Put { @@ -79,20 +86,51 @@ async fn test_transact_write_cancelled_transaction_keeps_an_item_an_earlier_tran ); let stored = engine - .get_item(&table.key_info, &key_for("aaa-committed")) + .get_item(&table.key_info, &key("aaa-committed")) .await .unwrap(); assert_eq!( - stored.as_ref().and_then(|item| item.get("val")), + stored.as_ref().and_then(|found| found.get("val")), Some(&AttributeValue::S("from-the-first-transaction".to_owned())), "the cancelled transaction changed or deleted the committed item: {stored:?}" ); assert!( engine - .get_item(&table.key_info, &key_for("zzz-absent")) + .get_item(&table.key_info, &key("zzz-absent")) .await .unwrap() .is_none(), "the cancelled transaction wrote a second item" ); } + +/// The scenario on a table with a hash key only. +#[tokio::test] +async fn test_transact_write_cancelled_transaction_keeps_an_item_an_earlier_transaction_committed() +{ + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnRollbackKeepsCommitted", false).await; + cancelled_transaction_keeps_a_committed_item(&engine, &table, item_with, key_for).await; +} + +/// The scenario on a table with a string sort key, which commits through its +/// own statement. +#[tokio::test] +async fn test_transact_write_cancelled_transaction_keeps_a_committed_sort_key_item() { + if crate::helpers::skip_without_cassandra() { + return; + } + let engine = setup_engine().await; + let table = TestTable::new(&engine, "TxnRollbackKeepsCommittedSk", true).await; + let sort = AttributeValue::S("s1".to_owned()); + cancelled_transaction_keeps_a_committed_item( + &engine, + &table, + |id, val| with_sort(item_with(id, val), &sort), + |id| with_sort(key_for(id), &sort), + ) + .await; +} diff --git a/crates/storage-cassandra/tests/direct/transact_write_stream_tables.rs b/crates/storage-cassandra/tests/direct/transact_write_stream_tables.rs index 63eb21fec..04588e91f 100644 --- a/crates/storage-cassandra/tests/direct/transact_write_stream_tables.rs +++ b/crates/storage-cassandra/tests/direct/transact_write_stream_tables.rs @@ -19,7 +19,7 @@ use crate::transact_write_stream_records::{ }; /// `item` with the sort key `sort`. -fn with_sort(mut item: Item, sort: &AttributeValue) -> Item { +pub(crate) fn with_sort(mut item: Item, sort: &AttributeValue) -> Item { item.insert("sort".to_owned(), sort.clone()); item } From fb59034ddabc69dc2353c8131343f1938463618a Mon Sep 17 00:00:00 2001 From: Lee Hannigan Date: Tue, 27 Oct 2026 12:42:29 +0000 Subject: [PATCH 9/9] test(cassandra): the record batch keeps the index batch's settings secondary_effect_batches builds the record's batch from the index batch's request and copies its batch type, consistency, serial consistency, timestamp, keyspace and now_in_seconds. The unit test for a record too large to share the index batch compared the consistency and the batch type only, and its index batch left the other four settings at their defaults, so a record batch that dropped them would have passed. The shared index_batch fixture now sets the serial consistency, the timestamp, the keyspace and now_in_seconds to values other than their defaults. The test checks that the index batch has all four set, then requires the record batch to carry each of them. --- .../src/data/transact_stream.rs | 38 +++++++++++++++++-- 1 file changed, 35 insertions(+), 3 deletions(-) diff --git a/crates/storage-cassandra/src/data/transact_stream.rs b/crates/storage-cassandra/src/data/transact_stream.rs index 1c2c173bd..4df80819a 100644 --- a/crates/storage-cassandra/src/data/transact_stream.rs +++ b/crates/storage-cassandra/src/data/transact_stream.rs @@ -142,9 +142,15 @@ mod tests { .collect() } + /// An index batch with every setting the record batch has to carry over set + /// to a value other than its default. fn index_batch() -> QueryBatch { BatchQueryBuilder::new() .with_consistency(Consistency::LocalQuorum) + .with_serial_consistency(Consistency::LocalSerial) + .with_timestamp(1_700_000_000_000_000) + .with_keyspace("ks".to_owned()) + .with_now_in_seconds(1_700_000_000) .add_query( "INSERT INTO ks.index_rows (a) VALUES (1)".to_owned(), QueryValues::SimpleValues(vec![]), @@ -218,7 +224,8 @@ mod tests { } /// A record too large to share a batch with the index rows runs after - /// them, alone, with the index batch's consistency. + /// them, alone, with the index batch's type and settings: consistency, + /// serial consistency, timestamp, keyspace and now_in_seconds. #[test] fn a_30000_byte_record_runs_as_a_second_batch_after_the_index_rows() { let record = record_of_len(30_000); @@ -235,8 +242,33 @@ mod tests { vec![record], "the record is last and alone" ); - assert_eq!(batches[1].request.consistency, Consistency::LocalQuorum); - assert_eq!(batches[1].request.batch_type, batches[0].request.batch_type); + let index = &batches[0].request; + let record_batch = &batches[1].request; + assert!( + index.serial_consistency.is_some() + && index.timestamp.is_some() + && index.keyspace.is_some() + && index.now_in_seconds.is_some(), + "the index batch must set every field, or the comparisons below match on defaults" + ); + assert_eq!(record_batch.consistency, Consistency::LocalQuorum); + assert_eq!(record_batch.batch_type, index.batch_type); + assert_eq!( + record_batch.serial_consistency, index.serial_consistency, + "the record batch must keep the serial consistency" + ); + assert_eq!( + record_batch.timestamp, index.timestamp, + "the record batch must keep the timestamp" + ); + assert_eq!( + record_batch.keyspace, index.keyspace, + "the record batch must keep the keyspace" + ); + assert_eq!( + record_batch.now_in_seconds, index.now_in_seconds, + "the record batch must keep now_in_seconds" + ); } /// The record shares the batch up to the budget and runs alone one byte