|
| 1 | +// SPDX-License-Identifier: BUSL-1.1 |
| 2 | + |
| 3 | +//! A `DELETE ... RETURNING` routed through Calvin surfaces its deleted rows, |
| 4 | +//! not a bare command tag. |
| 5 | +//! |
| 6 | +//! An edge-bearing schemaless collection routes a `DELETE FROM coll WHERE id = |
| 7 | +//! 'x'` through the OLLP/Calvin dependent path (the implicit-edge routing gate |
| 8 | +//! rewrites it as a `BulkDelete` so mirrored edges are cleaned atomically). The |
| 9 | +//! Calvin completion path signals done via a Raft-replicated ack that carries no |
| 10 | +//! payload, so before this fix the applied Data-Plane response — including the |
| 11 | +//! `RETURNING` rows — was dropped and the client saw only `DELETE 1`. |
| 12 | +//! |
| 13 | +//! This test proves the fix: with the single-node Calvin stack on, deleting one |
| 14 | +//! implicit-edge document with `RETURNING *` returns that document's row through |
| 15 | +//! the completion path. |
| 16 | +//! |
| 17 | +//! File name contains "calvin" within the cluster-tests crate so nextest applies |
| 18 | +//! the cluster test-group serialization. |
| 19 | +
|
| 20 | +mod common; |
| 21 | + |
| 22 | +use std::sync::atomic::Ordering; |
| 23 | +use std::time::Duration; |
| 24 | + |
| 25 | +use common::cluster_harness::{TestClusterNode, wait_for}; |
| 26 | + |
| 27 | +/// Number of implicit-edge documents seeded before the RETURNING delete. |
| 28 | +const SOURCES: usize = 6; |
| 29 | + |
| 30 | +/// Count of transactions the single-node sequencer has admitted to an epoch, or |
| 31 | +/// `0` if the sequencer metrics handle is not installed yet. |
| 32 | +fn sequencer_admitted(node: &TestClusterNode) -> u64 { |
| 33 | + node.shared |
| 34 | + .sequencer_metrics |
| 35 | + .get() |
| 36 | + .map(|m| m.admitted_total.load(Ordering::Relaxed)) |
| 37 | + .unwrap_or(0) |
| 38 | +} |
| 39 | + |
| 40 | +/// Whether `coll` is flagged edge-bearing in this node's local catalog. The |
| 41 | +/// implicit-edge mark is committed via the replicated metadata path, so the |
| 42 | +/// DELETE must wait for it before planning — otherwise the PK delete lowers to a |
| 43 | +/// static `PointDelete` (fast path) instead of the Calvin/OLLP `BulkDelete`. |
| 44 | +fn collection_edge_bearing(node: &TestClusterNode, coll: &str) -> bool { |
| 45 | + node.shared |
| 46 | + .credentials |
| 47 | + .catalog() |
| 48 | + .load_collections_for_tenant(nodedb_types::DatabaseId::DEFAULT, 1) |
| 49 | + .map(|v| v.iter().any(|c| c.name == coll && c.has_implicit_edges)) |
| 50 | + .unwrap_or(false) |
| 51 | +} |
| 52 | + |
| 53 | +/// Flag ON: a `DELETE ... RETURNING` on an edge-bearing collection routes |
| 54 | +/// through the single-node Calvin OLLP path and returns the deleted row through |
| 55 | +/// the completion path (previously dropped as a bare tag). |
| 56 | +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] |
| 57 | +async fn calvin_returning_delete_surfaces_deleted_row() { |
| 58 | + // 4 Data-Plane cores so the edge endpoints land on distinct vShards — the |
| 59 | + // doc-delete participant carries the RETURNING rows while the edge-delete |
| 60 | + // participants carry none, exercising the deposit gate. |
| 61 | + let node = TestClusterNode::spawn_single_node_calvin(4) |
| 62 | + .await |
| 63 | + .expect("spawn standalone single-node-calvin server"); |
| 64 | + |
| 65 | + // The lone sequencer voter self-elects; wait for it before submitting. |
| 66 | + wait_for( |
| 67 | + "single-node sequencer leader elected", |
| 68 | + Duration::from_secs(10), |
| 69 | + Duration::from_millis(50), |
| 70 | + || node.sequencer_leader() == node.node_id, |
| 71 | + ) |
| 72 | + .await; |
| 73 | + |
| 74 | + // The collection name deliberately embeds "returning": it doubles as |
| 75 | + // regression coverage that a `*_returning` identifier is not mistaken for the |
| 76 | + // RETURNING keyword by the clause stripper. |
| 77 | + let coll = "sncalvin_returning"; |
| 78 | + node.client |
| 79 | + .simple_query(&format!( |
| 80 | + "CREATE COLLECTION {coll} WITH (engine='document_schemaless')" |
| 81 | + )) |
| 82 | + .await |
| 83 | + .expect("CREATE COLLECTION"); |
| 84 | + wait_for( |
| 85 | + "collection visible on the node", |
| 86 | + Duration::from_secs(10), |
| 87 | + Duration::from_millis(50), |
| 88 | + || node.cached_collection_count() >= 1, |
| 89 | + ) |
| 90 | + .await; |
| 91 | + |
| 92 | + // src_0..src_(SOURCES-1) -> hub as IMPLICIT edges (plain docs carrying |
| 93 | + // _from/_to/_type). Inserting an edge document marks the collection |
| 94 | + // edge-bearing, which is what routes the later DELETE through OLLP/Calvin. |
| 95 | + for i in 0..SOURCES { |
| 96 | + node.client |
| 97 | + .simple_query(&format!( |
| 98 | + "INSERT INTO {coll} {{ id: 'edge_{i}', _from: 'src_{i}', _to: 'hub', _type: 'l', mark: 'keep' }}" |
| 99 | + )) |
| 100 | + .await |
| 101 | + .unwrap_or_else(|e| panic!("insert implicit edge src_{i} -> hub: {e}")); |
| 102 | + } |
| 103 | + |
| 104 | + // Wait for the implicit-edge mark to land so the PK delete plans as a |
| 105 | + // `BulkDelete` and routes through Calvin (not the fast `PointDelete` path). |
| 106 | + wait_for( |
| 107 | + "collection marked edge-bearing", |
| 108 | + Duration::from_secs(10), |
| 109 | + Duration::from_millis(50), |
| 110 | + || collection_edge_bearing(&node, coll), |
| 111 | + ) |
| 112 | + .await; |
| 113 | + |
| 114 | + let admitted_before = sequencer_admitted(&node); |
| 115 | + |
| 116 | + // A PK-equality DELETE on the edge-bearing collection is rewritten to a |
| 117 | + // `BulkDelete` (id filter) and routed through the OLLP/Calvin dependent path. |
| 118 | + // With RETURNING it must come back carrying the deleted document's row. |
| 119 | + let msgs = node |
| 120 | + .client |
| 121 | + .simple_query(&format!( |
| 122 | + "DELETE FROM {coll} WHERE id = 'edge_3' RETURNING *" |
| 123 | + )) |
| 124 | + .await |
| 125 | + .expect("RETURNING delete routed through Calvin must complete"); |
| 126 | + |
| 127 | + // Proof it traversed the sequencer→scheduler path (not the fast PointDelete |
| 128 | + // path that never touches Calvin): the delete was admitted to a Calvin epoch. |
| 129 | + let admitted_after = sequencer_admitted(&node); |
| 130 | + assert!( |
| 131 | + admitted_after > admitted_before, |
| 132 | + "the RETURNING delete must be admitted to a Calvin epoch \ |
| 133 | + (before={admitted_before}, after={admitted_after})" |
| 134 | + ); |
| 135 | + |
| 136 | + let rows: Vec<&tokio_postgres::SimpleQueryRow> = msgs |
| 137 | + .iter() |
| 138 | + .filter_map(|m| match m { |
| 139 | + tokio_postgres::SimpleQueryMessage::Row(r) => Some(r), |
| 140 | + _ => None, |
| 141 | + }) |
| 142 | + .collect(); |
| 143 | + |
| 144 | + // The core assertion: the response CONTAINS the deleted row, not just a bare |
| 145 | + // `DELETE 1` command tag. Before the fix this vec is empty. |
| 146 | + assert_eq!( |
| 147 | + rows.len(), |
| 148 | + 1, |
| 149 | + "DELETE ... RETURNING routed through Calvin must surface exactly the one \ |
| 150 | + deleted row, not a bare command tag; got {} row(s)", |
| 151 | + rows.len() |
| 152 | + ); |
| 153 | + let id = rows[0].get("id").expect("returned row has an id column"); |
| 154 | + assert_eq!( |
| 155 | + id, "edge_3", |
| 156 | + "the returned row must be the deleted document (id = 'edge_3'); got id = {id:?}" |
| 157 | + ); |
| 158 | + |
| 159 | + // Sanity: the deleted document is actually gone and the others remain. |
| 160 | + let remaining = node |
| 161 | + .client |
| 162 | + .simple_query(&format!("SELECT * FROM {coll}")) |
| 163 | + .await |
| 164 | + .expect("SELECT all rows"); |
| 165 | + let remaining_count = remaining |
| 166 | + .iter() |
| 167 | + .filter(|m| matches!(m, tokio_postgres::SimpleQueryMessage::Row(_))) |
| 168 | + .count(); |
| 169 | + assert_eq!( |
| 170 | + remaining_count, |
| 171 | + SOURCES - 1, |
| 172 | + "exactly the RETURNING-deleted document must be removed" |
| 173 | + ); |
| 174 | + |
| 175 | + node.shutdown().await; |
| 176 | +} |
0 commit comments