diff --git a/linux/crates/drivers/clickhouse/src/commit_resolution.rs b/linux/crates/drivers/clickhouse/src/commit_resolution.rs index 18f31c4c9c..3b2edc1e6b 100644 --- a/linux/crates/drivers/clickhouse/src/commit_resolution.rs +++ b/linux/crates/drivers/clickhouse/src/commit_resolution.rs @@ -4,11 +4,11 @@ //! An ALTER runs its mutation wait inside the interpreter, so while it //! waits the query log has no entry for it at all, and a wait that fails //! is logged as ExceptionBeforeStart with no table named. The statement -//! is therefore looked for in `system.processes` when the log is silent, -//! and its mutation in `system.mutations` by table and creation time when -//! its ALTER failed while the mutation may live on. The table comes from -//! the write itself when the log names none. `system.mutations` has no -//! column linking a mutation to its ALTER, so any doubt is Unresolvable. +//! is therefore looked for in `system.processes` too, and its mutation in +//! `system.mutations` by table and creation time when its ALTER failed +//! while the mutation may live on. The table comes from the write itself +//! when the log names none. `system.mutations` has no column linking a +//! mutation to its ALTER, so any doubt is Unresolvable. use tablepro_core::meta::TableRef; use tablepro_core::{CommitResolution, CommitToken, DriverError, Value}; @@ -44,13 +44,10 @@ pub(crate) struct MutatedTable { pub(crate) table: String, } -/// What the log alone says, or what else to ask. +/// What the log says, or what else to ask. #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) enum LogVerdict { Resolved(CommitResolution), - /// Nothing says the statement ended, so only the server's list of - /// running queries can tell whether it is still going. - NotEnded, /// A mutation's ALTER failed while the mutation may live on. Mutation { table: Option, @@ -72,7 +69,11 @@ fn mutation_may_live_on(exception_code: i32) -> bool { ) } -pub(crate) fn read_log(entries: &[LogEntry]) -> LogVerdict { +/// `listed` is whether `system.processes` held the statement before the +/// log was flushed. The server queues a statement's last log entry +/// before it takes the statement off that list, so a statement in +/// neither is one the server cannot account for. +pub(crate) fn read_log(entries: &[LogEntry], listed: bool) -> LogVerdict { let find = |event: &str| entries.iter().find(|entry| entry.event == event); if find("QueryFinish").is_some() { return LogVerdict::Resolved(CommitResolution::Applied); @@ -80,7 +81,11 @@ pub(crate) fn read_log(entries: &[LogEntry]) -> LogVerdict { let before_start = find("ExceptionBeforeStart"); let failed = before_start.or_else(|| find("ExceptionWhileProcessing")); let Some(failed) = failed else { - return LogVerdict::NotEnded; + return LogVerdict::Resolved(if listed { + CommitResolution::StillRunning + } else { + CommitResolution::Unresolvable + }); }; let mutation = 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 } } +/// The statement a resolution asks about: the first one the write could +/// not confirm. +struct Unconfirmed<'a> { + query_id: Uuid, + index: usize, + statements: usize, + written: &'a TableRef, +} + +impl<'a> Unconfirmed<'a> { + fn of(token: &'a CommitToken) -> Option { + let CommitToken::ClickHouseQuery { + query_ids, + first_unconfirmed, + table, + } = token + else { + return None; + }; + Some(Self { + query_id: *query_ids.get(*first_unconfirmed)?, + index: *first_unconfirmed, + statements: query_ids.len(), + written: table, + }) + } +} + /// Ask the server about the first statement a write could not confirm. +/// +/// The process list is read before the log is flushed, so a statement +/// that ends between the two reads is either still listed or has its end +/// in the flushed log. Read the other way round, it would be in neither. pub(crate) async fn resolve(calls: &CallClient, token: &CommitToken) -> Result { - let CommitToken::ClickHouseQuery { - query_ids, - first_unconfirmed, - table: written, - } = token - else { - return Ok(CommitResolution::Unresolvable); - }; - let Some(query_id) = query_ids.get(*first_unconfirmed) else { + let Some(statement) = Unconfirmed::of(token) else { return Ok(CommitResolution::Unresolvable); }; + let listed = listed(calls, statement.query_id).await?; + from_log(calls, &statement, listed).await +} + +async fn from_log( + calls: &CallClient, + statement: &Unconfirmed<'_>, + listed: bool, +) -> Result { match drain(calls.raw("SYSTEM FLUSH LOGS")?).await { Ok(_) => {} Err(error) if server_code(&error) == Some(code::ACCESS_DENIED) => {} Err(error) => return Err(error), } - let entries = match log_entries(calls, *query_id).await { + let entries = match log_entries(calls, statement.query_id).await { Ok(entries) => entries, Err(error) if matches!(server_code(&error), Some(code::ACCESS_DENIED | UNKNOWN_TABLE)) => { return Ok(CommitResolution::Unresolvable); } Err(error) => return Err(error), }; - let statement = match read_log(&entries) { + let resolution = match read_log(&entries, listed) { LogVerdict::Resolved(resolution) => resolution, - LogVerdict::NotEnded => still_running(calls, *query_id).await?, - LogVerdict::Mutation { table, started, ended } => match mutation_table(table, written) { + LogVerdict::Mutation { table, started, ended } => match mutation_table(table, statement.written) { Some(table) => mutation(calls, &table, &started, &ended).await?, None => CommitResolution::Unresolvable, }, }; - Ok(compose(*first_unconfirmed, query_ids.len(), statement)) + Ok(compose(statement.index, statement.statements, resolution)) } async fn log_entries(calls: &CallClient, query_id: Uuid) -> Result, DriverError> { @@ -199,24 +236,25 @@ async fn log_entries(calls: &CallClient, query_id: Uuid) -> Result .collect() } -/// Logged as started and never ended: still running if the server still -/// lists it, and lost otherwise. -async fn still_running(calls: &CallClient, query_id: Uuid) -> Result { +/// The stock config's `select_from_system_db_requires_grant` refuses +/// `system.processes` to a login without a grant. Such a login learns +/// nothing here, which leaves the log to answer. +async fn listed(calls: &CallClient, query_id: Uuid) -> Result { let query = calls .raw("SELECT count() FROM system.processes WHERE query_id = {id:String}")? .param("id", query_id.to_string()); - let result = read_table(query, None).await?; + let result = match read_table(query, None).await { + Ok(result) => result, + Err(error) if server_code(&error) == Some(code::ACCESS_DENIED) => return Ok(false), + Err(error) => return Err(error), + }; let running = result .rows .first() .and_then(|row| row.first()) .and_then(Value::as_exact_u64) .unwrap_or(0); - Ok(if running > 0 { - CommitResolution::StillRunning - } else { - CommitResolution::Unresolvable - }) + Ok(running > 0) } async fn mutation( @@ -245,7 +283,17 @@ async fn mutation( #[cfg(test)] mod tests { + use std::sync::Arc; + use std::time::Duration; + + use tablepro_core::StopRequest; + use tablepro_test_fixtures::ClickHouseFixture; + use tablepro_test_support::conformance::test_liveness; + use tokio::time::Instant; + use super::*; + use crate::cancel::ClickHouseStop; + use crate::endpoint::ClickHouseEndpoint; fn entry(event: &str, query_kind: &str, exception_code: i32) -> LogEntry { LogEntry { @@ -259,10 +307,17 @@ mod tests { } } + /// A logged end answers the same whatever the list said. + fn ended(entries: &[LogEntry]) -> LogVerdict { + let verdict = read_log(entries, true); + assert_eq!(read_log(entries, false), verdict, "{entries:?}"); + verdict + } + fn resolved(entries: &[LogEntry]) -> Option { - match read_log(entries) { + match ended(entries) { LogVerdict::Resolved(resolution) => Some(resolution), - _ => None, + LogVerdict::Mutation { .. } => None, } } @@ -285,19 +340,12 @@ mod tests { fn mapping_table() { use CommitResolution::{Applied, NotApplied, StillRunning, Unresolvable}; - assert_eq!( - read_log(&[]), - LogVerdict::NotEnded, - "an ALTER still waiting logs nothing" - ); for kind in ["Insert", "Alter", "Create"] { - let start = entry("QueryStart", kind, 0); assert_eq!( - resolved(&[start.clone(), entry("QueryFinish", kind, 0)]), + resolved(&[entry("QueryStart", kind, 0), entry("QueryFinish", kind, 0)]), Some(Applied), "{kind} finished" ); - assert_eq!(read_log(&[start]), LogVerdict::NotEnded, "{kind} only started"); assert_eq!( resolved(&[entry("ExceptionBeforeStart", kind, 62)]), Some(NotApplied), @@ -323,7 +371,7 @@ mod tests { "a failed statement that is not a mutation" ); assert_eq!( - read_log(&[ + ended(&[ entry("QueryStart", "Alter", 0), entry("ExceptionWhileProcessing", "Alter", 241) ]), @@ -334,7 +382,7 @@ mod tests { failed.databases.clear(); failed.tables.clear(); assert_eq!( - read_log(&[failed]), + ended(&[failed]), mutation(None), "an ALTER whose wait ended with {code}" ); @@ -352,6 +400,33 @@ mod tests { } } + #[test] + fn with_no_logged_end_the_list_answers() { + use CommitResolution::{StillRunning, Unresolvable}; + let start = entry("QueryStart", "Alter", 0); + + assert_eq!( + read_log(&[], true), + LogVerdict::Resolved(StillRunning), + "an ALTER still waiting logs nothing" + ); + assert_eq!( + read_log(std::slice::from_ref(&start), true), + LogVerdict::Resolved(StillRunning), + "only started" + ); + assert_eq!( + read_log(&[], false), + LogVerdict::Resolved(Unresolvable), + "neither listed nor logged" + ); + assert_eq!( + read_log(&[start], false), + LogVerdict::Resolved(Unresolvable), + "off the list with no end queued" + ); + } + #[test] fn a_mutation_is_looked_for_in_the_logged_table_then_the_written_one() { let written = TableRef::new(Some("tablepro".to_owned()), "orders"); @@ -397,4 +472,65 @@ mod tests { assert_eq!(compose(1, 3, StillRunning), StillRunning); assert_eq!(compose(1, 3, Unresolvable), Unresolvable); } + + async fn wait_until_listed(calls: &CallClient, query_id: Uuid, listing: bool) -> bool { + let give_up = Instant::now() + Duration::from_secs(10); + while listed(calls, query_id).await.expect("the process list") != listing { + assert!(Instant::now() < give_up, "listed stayed {}", !listing); + tokio::time::sleep(Duration::from_millis(50)).await; + } + listing + } + + /// Stopped merges hold the ALTER in its mutation wait while the list + /// is read, and it ends before the log is. The old order, log then + /// list, found it in neither and answered Unresolvable. + #[tokio::test] + #[ignore = "requires docker"] + async fn an_alter_that_ends_between_the_list_and_the_log_is_applied() { + let fixture = ClickHouseFixture::start().await.expect("a ClickHouse container"); + let endpoint = ClickHouseEndpoint::for_fixture(&fixture, test_liveness()); + let client = endpoint.workspace(); + for sql in [ + "CREATE TABLE held (id Int32, n Int32) ENGINE = MergeTree ORDER BY id", + "INSERT INTO held VALUES (1, 0)", + "SYSTEM STOP MERGES held", + ] { + client.query(sql).execute().await.expect(sql); + } + let query_id = Uuid::new_v4(); + let altering = tokio::spawn( + client + .query("ALTER TABLE held UPDATE n = 1 WHERE 1") + .with_setting("query_id", query_id.to_string()) + .execute(), + ); + let control = endpoint.control().clone(); + let calls = CallClient::new( + control.clone(), + Arc::new(ClickHouseStop::new(control)), + StopRequest::new(), + ); + let token = CommitToken::ClickHouseQuery { + query_ids: vec![query_id], + first_unconfirmed: 0, + table: TableRef::new(Some("tablepro".to_owned()), "held"), + }; + let statement = Unconfirmed::of(&token).expect("a ClickHouse token"); + + let listed = wait_until_listed(&calls, query_id, true).await; + assert_eq!(resolve(&calls, &token).await, Ok(CommitResolution::StillRunning)); + client + .query("SYSTEM START MERGES held") + .execute() + .await + .expect("the merges"); + altering.await.expect("the ALTER task").expect("the ALTER"); + wait_until_listed(&calls, query_id, false).await; + + assert_eq!( + from_log(&calls, &statement, listed).await, + Ok(CommitResolution::Applied) + ); + } } diff --git a/linux/crates/drivers/clickhouse/src/editor_session.rs b/linux/crates/drivers/clickhouse/src/editor_session.rs index 4bc240a0c0..0a3b2ccf21 100644 --- a/linux/crates/drivers/clickhouse/src/editor_session.rs +++ b/linux/crates/drivers/clickhouse/src/editor_session.rs @@ -140,12 +140,7 @@ impl EditorSession for ClickHouseEditorSession { #[cfg(test)] mod tests { use async_trait::async_trait; - use secrecy::SecretString; - use tablepro_core::{ - ConnectEnvironment, Credentials, LivenessPolicy, NetworkEndpoint, NetworkTarget, ResultEvent, RunEnd, SinkFlow, - TlsConfig, - }; - use tablepro_net::DirectTcpTransport; + use tablepro_core::{LivenessPolicy, ResultEvent, RunEnd, SinkFlow}; use tablepro_test_fixtures::ClickHouseFixture; use tablepro_test_support::conformance::test_liveness; @@ -170,26 +165,12 @@ mod tests { .await } - fn endpoint(fixture: &ClickHouseFixture, liveness: LivenessPolicy) -> ClickHouseEndpoint { - let address = NetworkEndpoint::new(fixture.host(), fixture.http_port()).expect("an endpoint"); - let credentials = fixture.password_credentials(); - let target = NetworkTarget { - transport: Arc::new(DirectTcpTransport::new(address, &liveness)), - credentials: Credentials::Password { - username: credentials.username.clone(), - password: SecretString::from(credentials.password.clone()), - }, - database: Some("tablepro".to_owned()), - tls: TlsConfig::disabled(), - }; - ClickHouseEndpoint::new(&target, liveness, &ConnectEnvironment::new("test"), false).expect("an endpoint") - } - #[tokio::test] #[ignore = "requires docker"] async fn a_session_the_server_let_expire_is_lost_not_replaced() { let fixture = ClickHouseFixture::start().await.expect("a ClickHouse container"); - let endpoint = endpoint(&fixture, test_liveness()).with_session_timeout(Duration::from_secs(1)); + let endpoint = + ClickHouseEndpoint::for_fixture(&fixture, test_liveness()).with_session_timeout(Duration::from_secs(1)); let session = ClickHouseEditorSession::open(&endpoint, EditorSessionOptions::default(), CancellationToken::new()) .await @@ -223,7 +204,7 @@ mod tests { cancel_grace: Duration::from_millis(300), ..test_liveness() }; - let endpoint = endpoint(&fixture, liveness); + let endpoint = ClickHouseEndpoint::for_fixture(&fixture, liveness); let session = ClickHouseEditorSession::open(&endpoint, EditorSessionOptions::default(), CancellationToken::new()) .await @@ -263,7 +244,7 @@ mod tests { cancel_grace: Duration::from_secs(10), ..test_liveness() }; - let endpoint = endpoint(&fixture, liveness); + let endpoint = ClickHouseEndpoint::for_fixture(&fixture, liveness); let session = ClickHouseEditorSession::open(&endpoint, EditorSessionOptions::default(), CancellationToken::new()) .await diff --git a/linux/crates/drivers/clickhouse/src/endpoint.rs b/linux/crates/drivers/clickhouse/src/endpoint.rs index 6d75c008fa..74ca4f2969 100644 --- a/linux/crates/drivers/clickhouse/src/endpoint.rs +++ b/linux/crates/drivers/clickhouse/src/endpoint.rs @@ -49,6 +49,26 @@ impl ClickHouseEndpoint { }) } + #[cfg(test)] + pub(crate) fn for_fixture(fixture: &tablepro_test_fixtures::ClickHouseFixture, liveness: LivenessPolicy) -> Self { + use secrecy::SecretString; + use tablepro_core::{Credentials, NetworkEndpoint, TlsConfig}; + use tablepro_net::DirectTcpTransport; + + let address = NetworkEndpoint::new(fixture.host(), fixture.http_port()).expect("an endpoint"); + let credentials = fixture.password_credentials(); + let target = NetworkTarget { + transport: Arc::new(DirectTcpTransport::new(address, &liveness)), + credentials: Credentials::Password { + username: credentials.username.clone(), + password: SecretString::from(credentials.password.clone()), + }, + database: Some("tablepro".to_owned()), + tls: TlsConfig::disabled(), + }; + Self::new(&target, liveness, &ConnectEnvironment::new("test"), false).expect("an endpoint") + } + #[cfg(test)] pub(crate) fn with_session_timeout(mut self, timeout: Duration) -> Self { self.session_timeout = timeout; diff --git a/linux/crates/drivers/clickhouse/tests/integration/writes.rs b/linux/crates/drivers/clickhouse/tests/integration/writes.rs index 2ab2d45114..f4207dfaca 100644 --- a/linux/crates/drivers/clickhouse/tests/integration/writes.rs +++ b/linux/crates/drivers/clickhouse/tests/integration/writes.rs @@ -5,14 +5,15 @@ use std::time::Duration; use tablepro_core::save::{ChangeSet, RowChange, RowUpdate}; use tablepro_core::{ - CommitResolution, CommitToken, Connection, DriverError, ReadOnlyRefusal, RunEnd, ServerCode, StatementOutcome, - Value, WriteKind, + CommitResolution, CommitToken, Connection, DriverError, ReadOnlyRefusal, RowLimit, RunEnd, ServerCode, + StatementOutcome, TlsConfig, Value, WorkspaceSettings, WriteKind, }; +use tablepro_test_support::conformance::test_liveness; use tokio::time::Instant; use crate::support::{ - CALL, DATABASE, admin, apply, batch_on, connect, count, flush_logs, planned, quoted, run, scalar, session, setup, - start, table, text, unique, within, + CALL, DATABASE, admin, apply, batch_on, connect, connect_with, count, direct, flush_logs, options_over, password, + planned, quoted, run, scalar, session, setup, start, statement, table, text, unique, within, }; /// A mutation that takes seconds on a small table, so it cannot finish @@ -190,17 +191,28 @@ async fn write_statements_never_carry_max_execution_time() { connection.close().await.expect("a close"); } +/// Stopped merges hold the mutation, so the ALTER is still waiting when +/// first asked about. #[tokio::test] #[ignore = "requires docker"] async fn alter_update_deadline_commit_outcome_unknown_resolves_still_running_then_applied() { let fixture = start().await; let connection = connect(&fixture, false).await; let name = unique("deadline"); - let qualified = slow_table(connection.as_ref(), &name).await; + let qualified = quoted(&name); + setup( + connection.as_ref(), + &[ + &format!("CREATE TABLE {qualified} (id Int32, n Int32) ENGINE = MergeTree ORDER BY id"), + &format!("INSERT INTO {qualified} VALUES (1, 0)"), + ], + ) + .await; + admin(connection.as_ref(), &format!("SYSTEM STOP MERGES {qualified}")).await; let outcome = connection .write_batch( - &batch_on(&name, &[&slow_update(&qualified)]), + &batch_on(&name, &[&format!("ALTER TABLE {qualified} UPDATE n = 1 WHERE 1")]), within(Duration::from_secs(1)), ) .await; @@ -212,16 +224,25 @@ async fn alter_update_deadline_commit_outcome_unknown_resolves_still_running_the resolution(connection.as_ref(), &token).await, CommitResolution::StillRunning ); + admin(connection.as_ref(), &format!("SYSTEM START MERGES {qualified}")).await; let give_up = Instant::now() + Duration::from_secs(120); loop { match resolution(connection.as_ref(), &token).await { CommitResolution::Applied => break, CommitResolution::StillRunning if Instant::now() < give_up => { - tokio::time::sleep(Duration::from_millis(500)).await; + tokio::time::sleep(Duration::from_millis(100)).await; } other => panic!("the mutation resolved as {other:?}"), } } + assert_eq!( + count( + connection.as_ref(), + &format!("SELECT count() FROM {qualified} WHERE n = 1") + ) + .await, + 1 + ); connection.close().await.expect("a close"); } @@ -417,6 +438,67 @@ async fn resolve_finished_insert_applied() { connection.close().await.expect("a close"); } +#[tokio::test] +#[ignore = "requires docker"] +async fn resolve_refused_the_process_list_answers_from_the_log() { + let fixture = start().await; + let owner = connect(&fixture, false).await; + let name = unique("limited"); + let qualified = quoted(&name); + setup( + owner.as_ref(), + &[ + &format!("CREATE TABLE {qualified} (id Int32) ENGINE = MergeTree ORDER BY id"), + &format!("INSERT INTO {qualified} VALUES (1)"), + ], + ) + .await; + let query_id = logged_query_id( + owner.as_ref(), + &format!("type = 'QueryFinish' AND query_kind = 'Insert' AND has(tables, '{DATABASE}.{name}')"), + ) + .await; + let liveness = test_liveness(); + let limited = connect_with(options_over( + direct(fixture.host(), fixture.http_port(), &liveness), + password(&fixture.limited_credentials()), + TlsConfig::disabled(), + WorkspaceSettings::workspace(false), + liveness, + )) + .await; + let refused = limited + .query_bound( + &statement("SELECT count() FROM system.processes"), + RowLimit::EDITOR_DEFAULT, + within(CALL), + ) + .await; + assert!( + matches!(&refused, Err(DriverError::Server(diagnostics)) + if diagnostics.code == Some(ServerCode::ClickHouse { code: 497 })), + "{refused:?}" + ); + + let finished = CommitToken::ClickHouseQuery { + query_ids: vec![query_id], + first_unconfirmed: 0, + table: table(&name), + }; + let unknown = CommitToken::ClickHouseQuery { + query_ids: vec![uuid::Uuid::new_v4()], + first_unconfirmed: 0, + table: table(&name), + }; + assert_eq!(resolution(limited.as_ref(), &finished).await, CommitResolution::Applied); + assert_eq!( + resolution(limited.as_ref(), &unknown).await, + CommitResolution::Unresolvable + ); + limited.close().await.expect("a close"); + owner.close().await.expect("a close"); +} + #[tokio::test] #[ignore = "requires docker"] async fn resolve_random_id_unresolvable() { diff --git a/linux/crates/test-fixtures/src/clickhouse_fixture.rs b/linux/crates/test-fixtures/src/clickhouse_fixture.rs index 9e42f5445b..60e86eb9d1 100644 --- a/linux/crates/test-fixtures/src/clickhouse_fixture.rs +++ b/linux/crates/test-fixtures/src/clickhouse_fixture.rs @@ -15,6 +15,7 @@ use crate::{ContainerFile, ExecOutput, FixtureCredentials, FixtureError, Fixture const HTTP_PORT: u16 = ClickHouseFixture::HTTP_PORT; const HTTPS_PORT: u16 = ClickHouseFixture::HTTPS_PORT; const PASSWORD_USER: &str = "tablepro"; +const LIMITED_USER: &str = "tablepro_limited"; const DATABASE: &str = "tablepro"; const SERVICE: &str = "clickhouse"; const READY_TIMEOUT: Duration = Duration::from_secs(60); @@ -110,6 +111,16 @@ impl ClickHouseFixture { ::/0\n\ defaultdefault\n\ \n\ + <{LIMITED_USER}>\n\ + {password}\n\ + ::/0\n\ + defaultdefault\n\ + \n\ + GRANT SELECT ON {DATABASE}.*\n\ + GRANT SELECT ON system.query_log\n\ + GRANT SYSTEM FLUSH LOGS ON *.*\n\ + \n\ + \n\ <{client}>\n\ {client}\n\ ::/0\n\ @@ -156,6 +167,12 @@ impl ClickHouseFixture { FixtureCredentials::new(PASSWORD_USER, &self.password) } + /// A login granted the query log but not `system.processes`, which the + /// stock server config holds back without a grant. + pub fn limited_credentials(&self) -> FixtureCredentials { + FixtureCredentials::new(LIMITED_USER, &self.password) + } + pub fn certificate_credentials(&self) -> FixtureCredentials { FixtureCredentials::new(TestPki::CLIENT_COMMON_NAME, "") } diff --git a/linux/docs/drivers/clickhouse.md b/linux/docs/drivers/clickhouse.md index c588b90a9a..cd5bc683a9 100644 --- a/linux/docs/drivers/clickhouse.md +++ b/linux/docs/drivers/clickhouse.md @@ -79,8 +79,9 @@ A failed statement is sorted by whether it can have landed. A server error that `CommitResolutionSupport::QueryLog`. `resolve_commit` asks about the first statement the save could not confirm: -1. `SYSTEM FLUSH LOGS`, so the log has the latest entries. A login without the privilege skips it. -2. That query id in `system.query_log` from today and yesterday: +1. Whether `system.processes` lists that query id. It is read before the flush: the server queues a query's last log entry before it takes the query off the list, so a query that ends between the two reads is still listed or has its end in the flushed log. A login refused `system.processes` (the stock config's `select_from_system_db_requires_grant` refuses it without a grant) learns nothing here, and a query still running reads as `Unresolvable` for it. +2. `SYSTEM FLUSH LOGS`, so the log has the latest entries. A login without the privilege skips it, and then a query that ended within the server's `query_log` flush interval (7.5 s by default) reads as `Unresolvable`. +3. That query id in `system.query_log` from today and yesterday: | What the log says | `CommitResolution` | |---|---| @@ -88,11 +89,11 @@ A failed statement is sorted by whether it can have landed. A server error that | `ExceptionBeforeStart`, unless an `ALTER` as in the mutation row | `NotApplied` | | `ExceptionWhileProcessing`, not an `ALTER` | `Unresolvable`: an `INSERT` may have written some blocks | | An `ALTER` that failed while processing, or whose wait ended with 394, 159, 341, 209 or 210 (cancelled, timed out, the wait gave up, a network error) | The mutation, below | -| Started, never ended | `StillRunning` if `system.processes` still lists it, else `Unresolvable` | +| No end, or no entry at all | `StillRunning` if step 1 found it listed, else `Unresolvable` | | No `system.query_log`, or no access to it | `Unresolvable` | -3. For a mutation: `system.mutations` has no column that links a mutation to its `ALTER`, so the driver looks for mutations on the table the log names, or the token's table when the log names none, created between the `ALTER`'s start and end. One that is done is `Applied`; one still running is `StillRunning`. None, because someone ran `KILL MUTATION` (which removes the row and keeps the parts already changed), or more than one, is `Unresolvable`. -4. The batch's answer: statements confirmed before the unconfirmed one, or not yet sent after it, turn a whole answer into `PartiallyApplied`. +4. For a mutation: `system.mutations` has no column that links a mutation to its `ALTER`, so the driver looks for mutations on the table the log names, or the token's table when the log names none, created between the `ALTER`'s start and end. One that is done is `Applied`; one still running is `StillRunning`. None, because someone ran `KILL MUTATION` (which removes the row and keeps the parts already changed), or more than one, is `Unresolvable`. +5. The batch's answer: statements confirmed before the unconfirmed one, or not yet sent after it, turn a whole answer into `PartiallyApplied`. The query log and the process list are each node's own, and the driver asks the node its connection reaches; it does not query the cluster. Behind a load balancer that sends the next request to another node, that node's log has no entry for the statement and the answer is `Unresolvable`.