Skip to content

Commit 1898e11

Browse files
authored
fix(linux-clickhouse): read the process list before the query log (#3449)
1 parent c0abf6c commit 1898e11

6 files changed

Lines changed: 319 additions & 82 deletions

File tree

‎linux/crates/drivers/clickhouse/src/commit_resolution.rs‎

Lines changed: 182 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -4,11 +4,11 @@
44
//! An ALTER runs its mutation wait inside the interpreter, so while it
55
//! waits the query log has no entry for it at all, and a wait that fails
66
//! is logged as ExceptionBeforeStart with no table named. The statement
7-
//! is therefore looked for in `system.processes` when the log is silent,
8-
//! and its mutation in `system.mutations` by table and creation time when
9-
//! its ALTER failed while the mutation may live on. The table comes from
10-
//! the write itself when the log names none. `system.mutations` has no
11-
//! column linking a mutation to its ALTER, so any doubt is Unresolvable.
7+
//! is therefore looked for in `system.processes` too, and its mutation in
8+
//! `system.mutations` by table and creation time when its ALTER failed
9+
//! while the mutation may live on. The table comes from the write itself
10+
//! when the log names none. `system.mutations` has no column linking a
11+
//! mutation to its ALTER, so any doubt is Unresolvable.
1212
1313
use tablepro_core::meta::TableRef;
1414
use tablepro_core::{CommitResolution, CommitToken, DriverError, Value};
@@ -44,13 +44,10 @@ pub(crate) struct MutatedTable {
4444
pub(crate) table: String,
4545
}
4646

47-
/// What the log alone says, or what else to ask.
47+
/// What the log says, or what else to ask.
4848
#[derive(Debug, Clone, PartialEq, Eq)]
4949
pub(crate) enum LogVerdict {
5050
Resolved(CommitResolution),
51-
/// Nothing says the statement ended, so only the server's list of
52-
/// running queries can tell whether it is still going.
53-
NotEnded,
5451
/// A mutation's ALTER failed while the mutation may live on.
5552
Mutation {
5653
table: Option<MutatedTable>,
@@ -72,15 +69,23 @@ fn mutation_may_live_on(exception_code: i32) -> bool {
7269
)
7370
}
7471

75-
pub(crate) fn read_log(entries: &[LogEntry]) -> LogVerdict {
72+
/// `listed` is whether `system.processes` held the statement before the
73+
/// log was flushed. The server queues a statement's last log entry
74+
/// before it takes the statement off that list, so a statement in
75+
/// neither is one the server cannot account for.
76+
pub(crate) fn read_log(entries: &[LogEntry], listed: bool) -> LogVerdict {
7677
let find = |event: &str| entries.iter().find(|entry| entry.event == event);
7778
if find("QueryFinish").is_some() {
7879
return LogVerdict::Resolved(CommitResolution::Applied);
7980
}
8081
let before_start = find("ExceptionBeforeStart");
8182
let failed = before_start.or_else(|| find("ExceptionWhileProcessing"));
8283
let Some(failed) = failed else {
83-
return LogVerdict::NotEnded;
84+
return LogVerdict::Resolved(if listed {
85+
CommitResolution::StillRunning
86+
} else {
87+
CommitResolution::Unresolvable
88+
});
8489
};
8590
let mutation =
8691
failed.query_kind == "Alter" && (before_start.is_none() || mutation_may_live_on(failed.exception_code));
@@ -142,40 +147,72 @@ pub(crate) fn compose(first_unconfirmed: usize, statements: usize, statement: Co
142147
}
143148
}
144149

150+
/// The statement a resolution asks about: the first one the write could
151+
/// not confirm.
152+
struct Unconfirmed<'a> {
153+
query_id: Uuid,
154+
index: usize,
155+
statements: usize,
156+
written: &'a TableRef,
157+
}
158+
159+
impl<'a> Unconfirmed<'a> {
160+
fn of(token: &'a CommitToken) -> Option<Self> {
161+
let CommitToken::ClickHouseQuery {
162+
query_ids,
163+
first_unconfirmed,
164+
table,
165+
} = token
166+
else {
167+
return None;
168+
};
169+
Some(Self {
170+
query_id: *query_ids.get(*first_unconfirmed)?,
171+
index: *first_unconfirmed,
172+
statements: query_ids.len(),
173+
written: table,
174+
})
175+
}
176+
}
177+
145178
/// Ask the server about the first statement a write could not confirm.
179+
///
180+
/// The process list is read before the log is flushed, so a statement
181+
/// that ends between the two reads is either still listed or has its end
182+
/// in the flushed log. Read the other way round, it would be in neither.
146183
pub(crate) async fn resolve(calls: &CallClient, token: &CommitToken) -> Result<CommitResolution, DriverError> {
147-
let CommitToken::ClickHouseQuery {
148-
query_ids,
149-
first_unconfirmed,
150-
table: written,
151-
} = token
152-
else {
153-
return Ok(CommitResolution::Unresolvable);
154-
};
155-
let Some(query_id) = query_ids.get(*first_unconfirmed) else {
184+
let Some(statement) = Unconfirmed::of(token) else {
156185
return Ok(CommitResolution::Unresolvable);
157186
};
187+
let listed = listed(calls, statement.query_id).await?;
188+
from_log(calls, &statement, listed).await
189+
}
190+
191+
async fn from_log(
192+
calls: &CallClient,
193+
statement: &Unconfirmed<'_>,
194+
listed: bool,
195+
) -> Result<CommitResolution, DriverError> {
158196
match drain(calls.raw("SYSTEM FLUSH LOGS")?).await {
159197
Ok(_) => {}
160198
Err(error) if server_code(&error) == Some(code::ACCESS_DENIED) => {}
161199
Err(error) => return Err(error),
162200
}
163-
let entries = match log_entries(calls, *query_id).await {
201+
let entries = match log_entries(calls, statement.query_id).await {
164202
Ok(entries) => entries,
165203
Err(error) if matches!(server_code(&error), Some(code::ACCESS_DENIED | UNKNOWN_TABLE)) => {
166204
return Ok(CommitResolution::Unresolvable);
167205
}
168206
Err(error) => return Err(error),
169207
};
170-
let statement = match read_log(&entries) {
208+
let resolution = match read_log(&entries, listed) {
171209
LogVerdict::Resolved(resolution) => resolution,
172-
LogVerdict::NotEnded => still_running(calls, *query_id).await?,
173-
LogVerdict::Mutation { table, started, ended } => match mutation_table(table, written) {
210+
LogVerdict::Mutation { table, started, ended } => match mutation_table(table, statement.written) {
174211
Some(table) => mutation(calls, &table, &started, &ended).await?,
175212
None => CommitResolution::Unresolvable,
176213
},
177214
};
178-
Ok(compose(*first_unconfirmed, query_ids.len(), statement))
215+
Ok(compose(statement.index, statement.statements, resolution))
179216
}
180217

181218
async fn log_entries(calls: &CallClient, query_id: Uuid) -> Result<Vec<LogEntry>, DriverError> {
@@ -199,24 +236,25 @@ async fn log_entries(calls: &CallClient, query_id: Uuid) -> Result<Vec<LogEntry>
199236
.collect()
200237
}
201238

202-
/// Logged as started and never ended: still running if the server still
203-
/// lists it, and lost otherwise.
204-
async fn still_running(calls: &CallClient, query_id: Uuid) -> Result<CommitResolution, DriverError> {
239+
/// The stock config's `select_from_system_db_requires_grant` refuses
240+
/// `system.processes` to a login without a grant. Such a login learns
241+
/// nothing here, which leaves the log to answer.
242+
async fn listed(calls: &CallClient, query_id: Uuid) -> Result<bool, DriverError> {
205243
let query = calls
206244
.raw("SELECT count() FROM system.processes WHERE query_id = {id:String}")?
207245
.param("id", query_id.to_string());
208-
let result = read_table(query, None).await?;
246+
let result = match read_table(query, None).await {
247+
Ok(result) => result,
248+
Err(error) if server_code(&error) == Some(code::ACCESS_DENIED) => return Ok(false),
249+
Err(error) => return Err(error),
250+
};
209251
let running = result
210252
.rows
211253
.first()
212254
.and_then(|row| row.first())
213255
.and_then(Value::as_exact_u64)
214256
.unwrap_or(0);
215-
Ok(if running > 0 {
216-
CommitResolution::StillRunning
217-
} else {
218-
CommitResolution::Unresolvable
219-
})
257+
Ok(running > 0)
220258
}
221259

222260
async fn mutation(
@@ -245,7 +283,17 @@ async fn mutation(
245283

246284
#[cfg(test)]
247285
mod tests {
286+
use std::sync::Arc;
287+
use std::time::Duration;
288+
289+
use tablepro_core::StopRequest;
290+
use tablepro_test_fixtures::ClickHouseFixture;
291+
use tablepro_test_support::conformance::test_liveness;
292+
use tokio::time::Instant;
293+
248294
use super::*;
295+
use crate::cancel::ClickHouseStop;
296+
use crate::endpoint::ClickHouseEndpoint;
249297

250298
fn entry(event: &str, query_kind: &str, exception_code: i32) -> LogEntry {
251299
LogEntry {
@@ -259,10 +307,17 @@ mod tests {
259307
}
260308
}
261309

310+
/// A logged end answers the same whatever the list said.
311+
fn ended(entries: &[LogEntry]) -> LogVerdict {
312+
let verdict = read_log(entries, true);
313+
assert_eq!(read_log(entries, false), verdict, "{entries:?}");
314+
verdict
315+
}
316+
262317
fn resolved(entries: &[LogEntry]) -> Option<CommitResolution> {
263-
match read_log(entries) {
318+
match ended(entries) {
264319
LogVerdict::Resolved(resolution) => Some(resolution),
265-
_ => None,
320+
LogVerdict::Mutation { .. } => None,
266321
}
267322
}
268323

@@ -285,19 +340,12 @@ mod tests {
285340
fn mapping_table() {
286341
use CommitResolution::{Applied, NotApplied, StillRunning, Unresolvable};
287342

288-
assert_eq!(
289-
read_log(&[]),
290-
LogVerdict::NotEnded,
291-
"an ALTER still waiting logs nothing"
292-
);
293343
for kind in ["Insert", "Alter", "Create"] {
294-
let start = entry("QueryStart", kind, 0);
295344
assert_eq!(
296-
resolved(&[start.clone(), entry("QueryFinish", kind, 0)]),
345+
resolved(&[entry("QueryStart", kind, 0), entry("QueryFinish", kind, 0)]),
297346
Some(Applied),
298347
"{kind} finished"
299348
);
300-
assert_eq!(read_log(&[start]), LogVerdict::NotEnded, "{kind} only started");
301349
assert_eq!(
302350
resolved(&[entry("ExceptionBeforeStart", kind, 62)]),
303351
Some(NotApplied),
@@ -323,7 +371,7 @@ mod tests {
323371
"a failed statement that is not a mutation"
324372
);
325373
assert_eq!(
326-
read_log(&[
374+
ended(&[
327375
entry("QueryStart", "Alter", 0),
328376
entry("ExceptionWhileProcessing", "Alter", 241)
329377
]),
@@ -334,7 +382,7 @@ mod tests {
334382
failed.databases.clear();
335383
failed.tables.clear();
336384
assert_eq!(
337-
read_log(&[failed]),
385+
ended(&[failed]),
338386
mutation(None),
339387
"an ALTER whose wait ended with {code}"
340388
);
@@ -352,6 +400,33 @@ mod tests {
352400
}
353401
}
354402

403+
#[test]
404+
fn with_no_logged_end_the_list_answers() {
405+
use CommitResolution::{StillRunning, Unresolvable};
406+
let start = entry("QueryStart", "Alter", 0);
407+
408+
assert_eq!(
409+
read_log(&[], true),
410+
LogVerdict::Resolved(StillRunning),
411+
"an ALTER still waiting logs nothing"
412+
);
413+
assert_eq!(
414+
read_log(std::slice::from_ref(&start), true),
415+
LogVerdict::Resolved(StillRunning),
416+
"only started"
417+
);
418+
assert_eq!(
419+
read_log(&[], false),
420+
LogVerdict::Resolved(Unresolvable),
421+
"neither listed nor logged"
422+
);
423+
assert_eq!(
424+
read_log(&[start], false),
425+
LogVerdict::Resolved(Unresolvable),
426+
"off the list with no end queued"
427+
);
428+
}
429+
355430
#[test]
356431
fn a_mutation_is_looked_for_in_the_logged_table_then_the_written_one() {
357432
let written = TableRef::new(Some("tablepro".to_owned()), "orders");
@@ -397,4 +472,65 @@ mod tests {
397472
assert_eq!(compose(1, 3, StillRunning), StillRunning);
398473
assert_eq!(compose(1, 3, Unresolvable), Unresolvable);
399474
}
475+
476+
async fn wait_until_listed(calls: &CallClient, query_id: Uuid, listing: bool) -> bool {
477+
let give_up = Instant::now() + Duration::from_secs(10);
478+
while listed(calls, query_id).await.expect("the process list") != listing {
479+
assert!(Instant::now() < give_up, "listed stayed {}", !listing);
480+
tokio::time::sleep(Duration::from_millis(50)).await;
481+
}
482+
listing
483+
}
484+
485+
/// Stopped merges hold the ALTER in its mutation wait while the list
486+
/// is read, and it ends before the log is. The old order, log then
487+
/// list, found it in neither and answered Unresolvable.
488+
#[tokio::test]
489+
#[ignore = "requires docker"]
490+
async fn an_alter_that_ends_between_the_list_and_the_log_is_applied() {
491+
let fixture = ClickHouseFixture::start().await.expect("a ClickHouse container");
492+
let endpoint = ClickHouseEndpoint::for_fixture(&fixture, test_liveness());
493+
let client = endpoint.workspace();
494+
for sql in [
495+
"CREATE TABLE held (id Int32, n Int32) ENGINE = MergeTree ORDER BY id",
496+
"INSERT INTO held VALUES (1, 0)",
497+
"SYSTEM STOP MERGES held",
498+
] {
499+
client.query(sql).execute().await.expect(sql);
500+
}
501+
let query_id = Uuid::new_v4();
502+
let altering = tokio::spawn(
503+
client
504+
.query("ALTER TABLE held UPDATE n = 1 WHERE 1")
505+
.with_setting("query_id", query_id.to_string())
506+
.execute(),
507+
);
508+
let control = endpoint.control().clone();
509+
let calls = CallClient::new(
510+
control.clone(),
511+
Arc::new(ClickHouseStop::new(control)),
512+
StopRequest::new(),
513+
);
514+
let token = CommitToken::ClickHouseQuery {
515+
query_ids: vec![query_id],
516+
first_unconfirmed: 0,
517+
table: TableRef::new(Some("tablepro".to_owned()), "held"),
518+
};
519+
let statement = Unconfirmed::of(&token).expect("a ClickHouse token");
520+
521+
let listed = wait_until_listed(&calls, query_id, true).await;
522+
assert_eq!(resolve(&calls, &token).await, Ok(CommitResolution::StillRunning));
523+
client
524+
.query("SYSTEM START MERGES held")
525+
.execute()
526+
.await
527+
.expect("the merges");
528+
altering.await.expect("the ALTER task").expect("the ALTER");
529+
wait_until_listed(&calls, query_id, false).await;
530+
531+
assert_eq!(
532+
from_log(&calls, &statement, listed).await,
533+
Ok(CommitResolution::Applied)
534+
);
535+
}
400536
}

0 commit comments

Comments
 (0)