From 86597fd6208250bbf077753f9709a1c7ea742d48 Mon Sep 17 00:00:00 2001 From: Aleksandr Romanenko Date: Wed, 19 Aug 2026 16:44:30 +0200 Subject: [PATCH 1/7] fix(cubestore): classify a missing remote file as its own error cause A download that fails because the object is gone is not a transient failure, but the only signal callers had was CorruptData, which they cannot act on: Cube Cloud retries such a download ten times with exponential backoff, and the startup warmup logs it at ERROR. A single file that compaction removed while the warmup walks its snapshot therefore costs ~51s of sleeps, 20 remote calls and 10 log lines, which slows the pass down enough to make the rest of the snapshot staler. Give it CubeErrorCauseType::FileNotFound so the process holding the remote fs can tell the two apart. A listing stays the only classifier, since object stores answer 404 both for a missing key and for a missing bucket. Nothing acts on the new cause yet, so behaviour is unchanged: it still counts as corrupt data for is_corrupt_data(), and it keeps the name and index of CorruptData everywhere it leaves the process - the wire, which a node of any version has to be able to read, and Display, whose output the scheduler matches on to deactivate a table whose import job failed. Co-Authored-By: Claude Opus 5 (1M context) --- rust/cubestore/cubestore/src/lib.rs | 104 +++++++++++++++++- .../cubestore/cubestore/src/remotefs/queue.rs | 6 +- 2 files changed, 105 insertions(+), 5 deletions(-) diff --git a/rust/cubestore/cubestore/src/lib.rs b/rust/cubestore/cubestore/src/lib.rs index 82fb3dcc3d3f4..1092bf1f65cd4 100644 --- a/rust/cubestore/cubestore/src/lib.rs +++ b/rust/cubestore/cubestore/src/lib.rs @@ -15,6 +15,7 @@ use datafusion::cube_ext::catch_unwind::PanicError; use datafusion::parquet::errors::ParquetError; use flexbuffers::{DeserializationError, ReaderError}; use log::SetLoggerError; +use serde::Serializer; use serde_derive::{Deserialize, Serialize}; use sqlparser::parser::ParserError; use std::any::Any; @@ -62,13 +63,41 @@ pub struct CubeError { impl std::error::Error for CubeError {} -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[derive(Debug, Clone, Deserialize, PartialEq)] pub enum CubeErrorCauseType { User, Internal, CorruptData, WrongConnection, Panic, + /// A remote object is absent rather than temporarily unreachable. + FileNotFound, +} + +impl CubeErrorCauseType { + /// How the cause looks to everything outside the process that produced it: peers reading a + /// serialized [CubeError] and [crate::scheduler] matching the job error strings that + /// [CubeError::to_string] produced. [CubeErrorCauseType::FileNotFound] is a refinement of + /// [CubeErrorCauseType::CorruptData] that only the local process acts on, so it takes the same + /// name and index and stays readable by a node or a stored job row of any version. + fn external_repr(&self) -> (u32, &'static str) { + match self { + CubeErrorCauseType::User => (0, "User"), + CubeErrorCauseType::Internal => (1, "Internal"), + CubeErrorCauseType::CorruptData | CubeErrorCauseType::FileNotFound => { + (2, "CorruptData") + } + CubeErrorCauseType::WrongConnection => (3, "WrongConnection"), + CubeErrorCauseType::Panic => (4, "Panic"), + } + } +} + +impl serde::Serialize for CubeErrorCauseType { + fn serialize(&self, serializer: S) -> Result { + let (index, name) = self.external_repr(); + serializer.serialize_unit_variant("CubeErrorCauseType", index, name) + } } impl CubeError { @@ -135,9 +164,24 @@ impl CubeError { } } + pub fn file_not_found(message: String) -> CubeError { + CubeError { + message, + backtrace: String::new(), + cause: CubeErrorCauseType::FileNotFound, + } + } + pub fn is_corrupt_data(&self) -> bool { match self.cause { - CubeErrorCauseType::CorruptData => true, + CubeErrorCauseType::CorruptData | CubeErrorCauseType::FileNotFound => true, + _ => false, + } + } + + pub fn is_file_not_found(&self) -> bool { + match self.cause { + CubeErrorCauseType::FileNotFound => true, _ => false, } } @@ -180,7 +224,11 @@ impl fmt::Display for CubeError { fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { match self.cause { CubeErrorCauseType::User => f.write_fmt(format_args!("{}", self.message)), - _ => f.write_fmt(format_args!("{:?}: {}", self.cause, self.message)), + _ => f.write_fmt(format_args!( + "{}: {}", + self.cause.external_repr().1, + self.message + )), } } } @@ -519,3 +567,53 @@ impl Into for CubeError { ArrowError::ExternalError(Box::new(self)) } } + +#[cfg(test)] +mod tests { + use super::*; + use serde::{Deserialize as _, Serialize as _}; + + fn to_flexbuffer(e: &CubeError) -> Vec { + let mut ser = flexbuffers::FlexbufferSerializer::new(); + e.serialize(&mut ser).unwrap(); + ser.take_buffer() + } + + #[test] + fn file_not_found_travels_as_corrupt_data() { + let not_found = CubeError::file_not_found("f.parquet is gone".to_string()); + let corrupt_data = CubeError::corrupt_data("f.parquet is gone".to_string()); + + // A node running any other version has to be able to read the result. + assert_eq!(to_flexbuffer(¬_found), to_flexbuffer(&corrupt_data)); + + let buffer = to_flexbuffer(¬_found); + let reader = flexbuffers::Reader::get_root(buffer.as_slice()).unwrap(); + assert_eq!(CubeError::deserialize(reader).unwrap(), corrupt_data); + } + + #[test] + fn file_not_found_reads_as_corrupt_data() { + // `scheduler` classifies a failed job by looking for "CorruptData" in the error string a + // worker stored via `to_string`, so the rendering has to stay the same. + let not_found = CubeError::file_not_found("f.parquet is gone".to_string()); + assert_eq!( + not_found.to_string(), + CubeError::corrupt_data("f.parquet is gone".to_string()).to_string() + ); + assert!(not_found.to_string().contains("CorruptData")); + } + + #[test] + fn file_not_found_is_corrupt_data_locally() { + let not_found = CubeError::file_not_found("f.parquet is gone".to_string()); + assert!(not_found.is_file_not_found()); + assert!(not_found.is_corrupt_data()); + + let corrupt_data = CubeError::corrupt_data("f.parquet is gone".to_string()); + assert!(!corrupt_data.is_file_not_found()); + assert!(corrupt_data.is_corrupt_data()); + + assert!(!CubeError::internal("nope".to_string()).is_file_not_found()); + } +} diff --git a/rust/cubestore/cubestore/src/remotefs/queue.rs b/rust/cubestore/cubestore/src/remotefs/queue.rs index bb5beb585f87a..9e716019d1ea6 100644 --- a/rust/cubestore/cubestore/src/remotefs/queue.rs +++ b/rust/cubestore/cubestore/src/remotefs/queue.rs @@ -288,9 +288,11 @@ impl RemoteFs for QueueRemoteFs { return Ok(f); } Err(err) => { - //Check if file doesn't exists in remoteFs + // A listing is the only way to tell an absent object from a remote + // that can't be reached at all: object stores answer 404 for both a + // missing key and a missing bucket. if self.remote_fs.list(file.clone()).await?.is_empty() { - return Err(CubeError::corrupt_data(format!( + return Err(CubeError::file_not_found(format!( "File {} doesn't exist in remote file system", file ))); From b3b31cbd1d2fd3d64868521f10533566a2fcda55 Mon Sep 17 00:00:00 2001 From: Aleksandr Romanenko Date: Wed, 19 Aug 2026 16:51:40 +0200 Subject: [PATCH 2/7] fix(cubestore): treat a GCS delete of an absent object as done Deleting a file that is already gone is the state the caller asked for, and S3 reports it that way: it answers 204 for a missing key. GCS answers 404 instead, so every GC task and every cleanup pass that raced with an earlier deletion logged an error nobody could act on. On one cluster those lines were 76% of all ERROR output, enough on their own to fire the error-rate alert. Report an absent object as a successful delete, and keep dropping the local copy so callers see the same end state on either driver. The check is a function of its own so a test can pin it to the payload a bucket really answers with: the reason mapping lives in the client crate, and if it ever changes the flood comes back silently. Co-Authored-By: Claude Opus 5 (1M context) --- rust/cubestore/cubestore/src/remotefs/gcs.rs | 52 +++++++++++++++++++- 1 file changed, 50 insertions(+), 2 deletions(-) diff --git a/rust/cubestore/cubestore/src/remotefs/gcs.rs b/rust/cubestore/cubestore/src/remotefs/gcs.rs index a8069021e431c..21c1de7b7ed99 100644 --- a/rust/cubestore/cubestore/src/remotefs/gcs.rs +++ b/rust/cubestore/cubestore/src/remotefs/gcs.rs @@ -118,6 +118,18 @@ impl GCSRemoteFs { di_service!(GCSRemoteFs, [RemoteFs, ExtendedRemoteFs]); +/// GCS answers 404 for an object that is already gone where S3 answers 204. Absent is the state a +/// caller of `delete_file` asked for and none of them can act on the difference, so it is reported +/// as done rather than as a failure they have to log. +fn is_absent_object(e: &cloud_storage::Error) -> bool { + match e { + cloud_storage::Error::Google(response) => { + response.errors_has_reason(&cloud_storage::Reason::NotFound) + } + _ => false, + } +} + #[async_trait] impl RemoteFs for GCSRemoteFs { async fn temp_upload_path(&self, remote_path: String) -> Result { @@ -247,8 +259,13 @@ impl RemoteFs for GCSRemoteFs { ); let time = SystemTime::now(); debug!("Deleting {}", remote_path); - Object::delete(self.bucket.as_str(), self.gcs_path(&remote_path).as_str()).await?; - info!("Deleting {} ({:?})", remote_path, time.elapsed()?); + match Object::delete(self.bucket.as_str(), self.gcs_path(&remote_path).as_str()).await { + Ok(()) => info!("Deleting {} ({:?})", remote_path, time.elapsed()?), + Err(e) if is_absent_object(&e) => { + debug!("File {} is already absent in remote fs", remote_path); + } + Err(e) => return Err(e.into()), + } let _guard = acquire_lock("delete file", self.delete_mut.lock()).await?; let local = self.dir.as_path().join(remote_path); @@ -353,3 +370,34 @@ impl GCSRemoteFs { ) } } + +#[cfg(test)] +mod tests { + use super::*; + + /// The payload a bucket really answers with, so a change in how the crate maps reasons shows + /// up here instead of as a returning flood of errors from the scheduler. + const NOT_FOUND_RESPONSE: &str = r#"{"error":{"errors":[{"domain":"global", + "reason":"notFound","message":"No such object: bucket/1829139-ihr4wfxi.parquet"}], + "code":404,"message":"No such object: bucket/1829139-ihr4wfxi.parquet"}}"#; + + const FORBIDDEN_RESPONSE: &str = r#"{"error":{"errors":[{"domain":"global", + "reason":"forbidden","message":"Access denied"}],"code":403,"message":"Access denied"}}"#; + + fn google_error(payload: &str) -> cloud_storage::Error { + cloud_storage::Error::Google(serde_json::from_str(payload).unwrap()) + } + + #[test] + fn absent_object_is_recognized() { + assert!(is_absent_object(&google_error(NOT_FOUND_RESPONSE))); + } + + #[test] + fn other_failures_are_not_absent_objects() { + assert!(!is_absent_object(&google_error(FORBIDDEN_RESPONSE))); + assert!(!is_absent_object(&cloud_storage::Error::Other( + "connection reset".to_string() + ))); + } +} From 9175e9955412743eca18faf26f1b158ea0fea9fd Mon Sep 17 00:00:00 2001 From: Aleksandr Romanenko Date: Wed, 19 Aug 2026 17:00:45 +0200 Subject: [PATCH 3/7] fix(cubestore): stop reporting files compaction removed under a running warmup The startup warmup takes one metastore snapshot and then walks it one download at a time, which on a large node runs for hours. Everything compaction replaces in the meantime is gone from remote storage by the time the pass asks for it, and every one of those was logged at ERROR, so a worker kept reporting errors about files nothing referenced any more for as long as the pass lasted. Ask the metastore what the file's row looks like now. A row that is inactive or gone means compaction replaced it, which is routine and belongs in debug output. A row that is still active means the data is really missing, which is the case worth an error and worth a metric of its own - it is the one this alert was supposed to be about. Co-Authored-By: Claude Opus 5 (1M context) --- rust/cubestore/cubestore/src/app_metrics.rs | 6 + rust/cubestore/cubestore/src/cluster/mod.rs | 131 ++++++++++++++++++-- 2 files changed, 126 insertions(+), 11 deletions(-) diff --git a/rust/cubestore/cubestore/src/app_metrics.rs b/rust/cubestore/cubestore/src/app_metrics.rs index 4059f0a8b0db3..e7fd6af48a848 100644 --- a/rust/cubestore/cubestore/src/app_metrics.rs +++ b/rust/cubestore/cubestore/src/app_metrics.rs @@ -148,6 +148,12 @@ pub static REMOTE_FS_FILES_TO_REMOVE: Gauge = metrics::gauge("cs.remote_fs.files pub static REMOTE_FS_FILES_SIZE_TO_REMOVE: Gauge = metrics::gauge("cs.remote_fs.files_to_remove.size"); +/// Startup warmup files that were gone by the time the pass reached them. `stale` counts the ones +/// whose metastore row compaction had already replaced, which is routine on a busy node; `active` +/// counts the ones an active row still points at, which means the data is really missing. +pub static WARMUP_MISSING_STALE: Counter = metrics::counter("cs.warmup.missing.stale"); +pub static WARMUP_MISSING_ACTIVE: Counter = metrics::counter("cs.warmup.missing.active"); + /// Cache Store Cache pub static CACHESTORE_TTL_PERSIST: Counter = metrics::counter("cs.cachestore.ttl.persist"); pub static CACHESTORE_TTL_BUFFER: Gauge = metrics::gauge("cs.cachestore.ttl.buffer"); diff --git a/rust/cubestore/cubestore/src/cluster/mod.rs b/rust/cubestore/cubestore/src/cluster/mod.rs index 446f0a34c2445..d0255aa177c1f 100644 --- a/rust/cubestore/cubestore/src/cluster/mod.rs +++ b/rust/cubestore/cubestore/src/cluster/mod.rs @@ -19,7 +19,7 @@ use crate::cluster::worker_services::{ WorkerProcessing, }; -use crate::ack_error; +use crate::app_metrics; use crate::cluster::message::NetworkMessage; use crate::cluster::rate_limiter::{ProcessRateLimiter, TaskType, TraceIndex}; use crate::cluster::transport::{ClusterTransport, MetaStoreTransport, WorkerConnection}; @@ -2118,13 +2118,11 @@ impl ClusterImpl { log::debug!("Startup warmup cancelled"); return; } - // TODO: propagate 'not found' and log in debug mode. Compaction might remove files, - // so they are not errors most of the time. - ack_error!( - self.remote_fs - .download_file(file, p.get_row().file_size()) - .await - ); + let result = self + .remote_fs + .download_file(file, p.get_row().file_size()) + .await; + self.report_warmup_download(result, p.get_id(), None).await; } for c in chunks { if self.stop_token.is_cancelled() { @@ -2141,14 +2139,54 @@ impl ClusterImpl { c.get_row().file_size(), ) .await; - // TODO: propagate 'not found' and log in debug mode. Compaction might remove files, - // so they are not errors most of the time. - ack_error!(result); + self.report_warmup_download(result, p.get_id(), Some(c.get_id())) + .await; } } log::debug!("Startup warmup finished"); return; } + + /// A file that is gone by the time the pass reaches it is the normal case: the walk takes as + /// long as it takes and compaction removes what it replaces meanwhile. One that an active row + /// still points at means the data is really missing, which is worth an error. + async fn report_warmup_download( + &self, + result: Result, + partition_id: u64, + chunk_id: Option, + ) { + let e = match result { + Ok(_) => return, + Err(e) => e, + }; + if e.is_file_not_found() { + if self.is_warmup_file_referenced(partition_id, chunk_id).await { + app_metrics::WARMUP_MISSING_ACTIVE.increment(); + log::error!("Warmup file of an active row is missing: {}", e.message); + } else { + app_metrics::WARMUP_MISSING_STALE.increment(); + log::debug!("Skipping warmup of a replaced file: {}", e.message); + } + return; + } + log::error!("Error: {:?}", e); + } + + /// Whether the metastore still points at the file. A row that cannot be read counts as gone, + /// the same way the corrupt data deactivation path treats it. + async fn is_warmup_file_referenced(&self, partition_id: u64, chunk_id: Option) -> bool { + match chunk_id { + Some(chunk_id) => match self.meta_store.get_chunk(chunk_id).await { + Ok(c) => c.get_row().active(), + Err(_) => false, + }, + None => match self.meta_store.get_partition(partition_id).await { + Ok(p) => p.get_row().has_main_table_file(), + Err(_) => false, + }, + } + } } struct LoopbackConnection { @@ -2314,6 +2352,8 @@ pub fn pick_worker_by_partitions<'a>( mod tests { use super::*; use crate::config::Config; + use crate::queryplanner::trace_data_loaded::DataLoadedSize; + use crate::store::compaction::CompactionService; use std::fs; fn config_with_workers(name: &str, workers: Vec) -> Arc { @@ -2325,6 +2365,75 @@ mod tests { .config_obj() } + /// Compaction replacing a chunk with a partition file is exactly the race the warmup loses when + /// it reaches a file hours after the snapshot that named it. + #[tokio::test] + async fn warmup_file_reference_follows_the_metastore() { + Config::test("warmup_file_reference") + .start_test(async move |services| { + let service = services.sql_service; + let meta_store = services.meta_store; + let cluster = services.cluster; + + service.exec_query("CREATE SCHEMA test").await?; + service.exec_query("CREATE TABLE test.warm (a int)").await?; + service + .exec_query("INSERT INTO test.warm (a) VALUES (1), (2)") + .await?; + + let partition = meta_store + .get_partitions_with_chunks_created_seconds_ago(-1000) + .await? + .remove(0); + let partition_id = partition.get_id(); + let index_id = partition.get_row().get_index_id(); + let chunk_id = meta_store + .get_chunks_by_partition(partition_id, false) + .await?[0] + .get_id(); + + assert!( + cluster + .is_warmup_file_referenced(partition_id, Some(chunk_id)) + .await + ); + // The partition holds no data of its own until it is compacted. + assert!(!cluster.is_warmup_file_referenced(partition_id, None).await); + + services + .injector + .get_service_typed::() + .await + .compact(partition_id, DataLoadedSize::new()) + .await?; + + // Compaction moved the rows into a partition file of its own and left the chunk + // and its parent behind, which is what makes their files disappear under a pass + // that is still walking an older snapshot. + let compacted_id = meta_store + .get_active_partitions_by_index_id(index_id) + .await? + .remove(0) + .get_id(); + assert_ne!(compacted_id, partition_id); + assert!(cluster.is_warmup_file_referenced(compacted_id, None).await); + assert!(!cluster.is_warmup_file_referenced(partition_id, None).await); + assert!( + !cluster + .is_warmup_file_referenced(partition_id, Some(chunk_id)) + .await + ); + assert!( + !cluster + .is_warmup_file_referenced(partition_id, Some(chunk_id + 1000)) + .await + ); + + Ok::<(), CubeError>(()) + }) + .await; + } + #[test] fn pick_import_worker_load_aware() { let config = config_with_workers( From 99a20184c9c1951ced9445c5652b6161c935ab95 Mon Sep 17 00:00:00 2001 From: Aleksandr Romanenko Date: Wed, 19 Aug 2026 17:30:03 +0200 Subject: [PATCH 4/7] fix(cubestore): tell an unreachable metastore from a row that is gone The warmup check read one row per absent file and treated any failure to read it as proof the row was gone. Both halves were wrong in the same direction: warmup runs at startup, when a select worker reaches the metastore over an RPC link that may still be coming up, so a flaky link made every missing file look routine and counted it as such - fail-silent, in a check whose whole purpose is to report data that is really missing. And a node whose snapshot compaction had long moved on issued a router point-get per stale file, with every warming worker doing it at once, exactly while the router was busy recovering. Re-read the rows in one batch per partition with the out of queue readers, which leave out ids they no longer hold. That gives the classification its third answer: an id missing from the result is a deleted row, while an error is a check that did not happen and is now reported as such instead of being folded into either verdict. Also note in the GCS delete path that a missing bucket carries the same reason code as a missing object and that the client crate keeps the detail private, so the two cannot be told apart there. Co-Authored-By: Claude Opus 5 (1M context) --- rust/cubestore/cubestore/src/cluster/mod.rs | 176 +++++++++++++------ rust/cubestore/cubestore/src/remotefs/gcs.rs | 4 + 2 files changed, 130 insertions(+), 50 deletions(-) diff --git a/rust/cubestore/cubestore/src/cluster/mod.rs b/rust/cubestore/cubestore/src/cluster/mod.rs index d0255aa177c1f..cc1af7941ccad 100644 --- a/rust/cubestore/cubestore/src/cluster/mod.rs +++ b/rust/cubestore/cubestore/src/cluster/mod.rs @@ -67,7 +67,7 @@ use opentelemetry::Context as OtelContext; use regex::Regex; use serde::{Deserialize, Serialize}; use std::collections::hash_map::DefaultHasher; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::hash::{Hash, Hasher}; use std::pin::Pin; use std::sync::Weak; @@ -2113,6 +2113,8 @@ impl ClusterImpl { if self.node_name_by_partition(&p) != self.server_name { continue; } + let mut missing_partition_file = false; + let mut missing_chunk_ids = Vec::new(); if let Some(file) = p.get_row().get_full_name(p.get_id()) { if self.stop_token.is_cancelled() { log::debug!("Startup warmup cancelled"); @@ -2122,7 +2124,7 @@ impl ClusterImpl { .remote_fs .download_file(file, p.get_row().file_size()) .await; - self.report_warmup_download(result, p.get_id(), None).await; + missing_partition_file = self.warmup_file_is_absent(result); } for c in chunks { if self.stop_token.is_cancelled() { @@ -2139,53 +2141,131 @@ impl ClusterImpl { c.get_row().file_size(), ) .await; - self.report_warmup_download(result, p.get_id(), Some(c.get_id())) - .await; + if self.warmup_file_is_absent(result) { + missing_chunk_ids.push(c.get_id()); + } + } + if missing_partition_file || !missing_chunk_ids.is_empty() { + self.report_absent_warmup_files( + p.get_id(), + missing_partition_file, + &missing_chunk_ids, + ) + .await; } } log::debug!("Startup warmup finished"); return; } + /// Reports everything but an absent file, which on its own says nothing: only the metastore can + /// tell whether the pass is simply behind or the data is gone. + fn warmup_file_is_absent(&self, result: Result) -> bool { + match result { + Ok(_) => false, + Err(e) if e.is_file_not_found() => true, + Err(e) => { + log::error!("Error: {:?}", e); + false + } + } + } + /// A file that is gone by the time the pass reaches it is the normal case: the walk takes as - /// long as it takes and compaction removes what it replaces meanwhile. One that an active row - /// still points at means the data is really missing, which is worth an error. - async fn report_warmup_download( + /// long as it takes and compaction removes what it replaces meanwhile. A file the metastore + /// still names is really missing, which is worth an error of its own. + async fn report_absent_warmup_files( &self, - result: Result, partition_id: u64, - chunk_id: Option, + partition_file: bool, + chunk_ids: &[u64], ) { - let e = match result { - Ok(_) => return, - Err(e) => e, + let (partition_named, named_chunks) = match self + .named_warmup_files(partition_id, partition_file, chunk_ids) + .await + { + Ok(named) => named, + Err(e) => { + // Assuming either verdict would either hide missing data or report every file of a + // partition as lost, so report that the check itself did not happen. + log::warn!( + "Could not check {} absent warmup file(s) of partition {} against the metastore: {}", + chunk_ids.len() + partition_file as usize, + partition_id, + e + ); + return; + } }; - if e.is_file_not_found() { - if self.is_warmup_file_referenced(partition_id, chunk_id).await { - app_metrics::WARMUP_MISSING_ACTIVE.increment(); - log::error!("Warmup file of an active row is missing: {}", e.message); + + let mut active = 0; + let mut stale = 0; + if partition_file { + if partition_named { + active += 1; + log::error!( + "Warmup file of active partition {} is missing in remote fs", + partition_id + ); } else { - app_metrics::WARMUP_MISSING_STALE.increment(); - log::debug!("Skipping warmup of a replaced file: {}", e.message); + stale += 1; } - return; } - log::error!("Error: {:?}", e); + for id in chunk_ids { + if named_chunks.contains(id) { + active += 1; + log::error!("Warmup file of active chunk {} is missing in remote fs", id); + } else { + stale += 1; + } + } + + if 0 < active { + app_metrics::WARMUP_MISSING_ACTIVE.add(active); + } + if 0 < stale { + app_metrics::WARMUP_MISSING_STALE.add(stale); + log::debug!( + "Skipping {} warmup file(s) of partition {} replaced since the snapshot", + stale, + partition_id + ); + } } - /// Whether the metastore still points at the file. A row that cannot be read counts as gone, - /// the same way the corrupt data deactivation path treats it. - async fn is_warmup_file_referenced(&self, partition_id: u64, chunk_id: Option) -> bool { - match chunk_id { - Some(chunk_id) => match self.meta_store.get_chunk(chunk_id).await { - Ok(c) => c.get_row().active(), - Err(_) => false, - }, - None => match self.meta_store.get_partition(partition_id).await { - Ok(p) => p.get_row().has_main_table_file(), - Err(_) => false, - }, + /// Which of the files the metastore still names, read in one batch per partition so that a node + /// whose snapshot compaction has long moved on does not turn the check into a request per file. + /// The batch reads leave out the ids they no longer hold, which is what tells a deleted row + /// apart from a metastore we could not reach. + async fn named_warmup_files( + &self, + partition_id: u64, + partition_file: bool, + chunk_ids: &[u64], + ) -> Result<(bool, HashSet), CubeError> { + let partition_named = if partition_file { + self.meta_store + .get_partitions_out_of_queue(vec![partition_id]) + .await? + .first() + .map_or(false, |p| p.get_row().has_main_table_file()) + } else { + false + }; + + let mut named_chunks = HashSet::new(); + if !chunk_ids.is_empty() { + for c in self + .meta_store + .get_chunks_out_of_queue(chunk_ids.to_vec()) + .await? + { + if c.get_row().active() { + named_chunks.insert(c.get_id()); + } + } } + Ok((partition_named, named_chunks)) } } @@ -2392,13 +2472,12 @@ mod tests { .await?[0] .get_id(); - assert!( - cluster - .is_warmup_file_referenced(partition_id, Some(chunk_id)) - .await - ); + let (partition_named, named_chunks) = cluster + .named_warmup_files(partition_id, true, &[chunk_id]) + .await?; + assert!(named_chunks.contains(&chunk_id)); // The partition holds no data of its own until it is compacted. - assert!(!cluster.is_warmup_file_referenced(partition_id, None).await); + assert!(!partition_named); services .injector @@ -2416,18 +2495,15 @@ mod tests { .remove(0) .get_id(); assert_ne!(compacted_id, partition_id); - assert!(cluster.is_warmup_file_referenced(compacted_id, None).await); - assert!(!cluster.is_warmup_file_referenced(partition_id, None).await); - assert!( - !cluster - .is_warmup_file_referenced(partition_id, Some(chunk_id)) - .await - ); - assert!( - !cluster - .is_warmup_file_referenced(partition_id, Some(chunk_id + 1000)) - .await - ); + assert!(cluster.named_warmup_files(compacted_id, true, &[]).await?.0); + + // An id the metastore no longer holds at all is left out of the batch read, the + // same as one it holds as inactive. + let (partition_named, named_chunks) = cluster + .named_warmup_files(partition_id, true, &[chunk_id, chunk_id + 1000]) + .await?; + assert!(!partition_named); + assert!(named_chunks.is_empty()); Ok::<(), CubeError>(()) }) diff --git a/rust/cubestore/cubestore/src/remotefs/gcs.rs b/rust/cubestore/cubestore/src/remotefs/gcs.rs index 21c1de7b7ed99..2737b3da87cc3 100644 --- a/rust/cubestore/cubestore/src/remotefs/gcs.rs +++ b/rust/cubestore/cubestore/src/remotefs/gcs.rs @@ -121,6 +121,10 @@ di_service!(GCSRemoteFs, [RemoteFs, ExtendedRemoteFs]); /// GCS answers 404 for an object that is already gone where S3 answers 204. Absent is the state a /// caller of `delete_file` asked for and none of them can act on the difference, so it is reported /// as done rather than as a failure they have to log. +/// +/// A missing bucket carries the same reason, and the client crate keeps the message and the code +/// private, so the two cannot be told apart here. A bucket that is not there fails every upload and +/// download loudly, which is the signal to go by. fn is_absent_object(e: &cloud_storage::Error) -> bool { match e { cloud_storage::Error::Google(response) => { From e1d495772d975553c60732dc7c372df6c4a28eef Mon Sep 17 00:00:00 2001 From: Aleksandr Romanenko Date: Wed, 19 Aug 2026 18:00:50 +0200 Subject: [PATCH 5/7] fix(cubestore): check absent warmup files in batches across partitions The check ran per partition, which on the path the pass hits most is still one or two requests to the main node for every group of files compaction has replaced. A worker returning after a long absence finds its whole snapshot replaced, so the count follows the number of partitions it holds, and it pays them while the main node is busy with the rest of the cluster coming up. Slowing the pass down is also what makes files go stale in the first place, so the check was working against the fix it belongs to. Collect absent files as the pass goes and look them up once per 256, which is the same verdict for a fraction of the round trips. A cancelled pass drops what it has not looked up: the next start walks the whole list again. Report per batch as well. A table whose objects an operator or a lifecycle rule really removed would otherwise produce an error line per chunk, which is the flood this check exists to remove wearing a different message. Co-Authored-By: Claude Opus 5 (1M context) --- rust/cubestore/cubestore/src/cluster/mod.rs | 169 +++++++++++--------- 1 file changed, 94 insertions(+), 75 deletions(-) diff --git a/rust/cubestore/cubestore/src/cluster/mod.rs b/rust/cubestore/cubestore/src/cluster/mod.rs index cc1af7941ccad..9892871ace4ce 100644 --- a/rust/cubestore/cubestore/src/cluster/mod.rs +++ b/rust/cubestore/cubestore/src/cluster/mod.rs @@ -2109,12 +2109,16 @@ impl ClusterImpl { }; log::debug!("Got {} partitions, running the warmup", partitions.len()); + // Absent files are looked up in batches: the pass is expected to find plenty of them, and + // the metastore of a select worker is a request to the main node, which is busy with the + // rest of the cluster starting up. + const ABSENT_BATCH: usize = 256; + let mut absent = AbsentWarmupFiles::default(); + for (p, chunks) in partitions { if self.node_name_by_partition(&p) != self.server_name { continue; } - let mut missing_partition_file = false; - let mut missing_chunk_ids = Vec::new(); if let Some(file) = p.get_row().get_full_name(p.get_id()) { if self.stop_token.is_cancelled() { log::debug!("Startup warmup cancelled"); @@ -2124,7 +2128,9 @@ impl ClusterImpl { .remote_fs .download_file(file, p.get_row().file_size()) .await; - missing_partition_file = self.warmup_file_is_absent(result); + if self.warmup_file_is_absent(result) { + absent.partition_ids.push(p.get_id()); + } } for c in chunks { if self.stop_token.is_cancelled() { @@ -2142,18 +2148,17 @@ impl ClusterImpl { ) .await; if self.warmup_file_is_absent(result) { - missing_chunk_ids.push(c.get_id()); + absent.chunk_ids.push(c.get_id()); } } - if missing_partition_file || !missing_chunk_ids.is_empty() { - self.report_absent_warmup_files( - p.get_id(), - missing_partition_file, - &missing_chunk_ids, - ) - .await; + if ABSENT_BATCH <= absent.len() { + self.report_absent_warmup_files(std::mem::take(&mut absent)) + .await; } } + // A cancelled pass drops what it has not looked up yet, since the next start walks the + // whole list again anyway and the metastore is on its way down with us. + self.report_absent_warmup_files(absent).await; log::debug!("Startup warmup finished"); return; } @@ -2171,87 +2176,90 @@ impl ClusterImpl { } } - /// A file that is gone by the time the pass reaches it is the normal case: the walk takes as - /// long as it takes and compaction removes what it replaces meanwhile. A file the metastore - /// still names is really missing, which is worth an error of its own. - async fn report_absent_warmup_files( - &self, - partition_id: u64, - partition_file: bool, - chunk_ids: &[u64], - ) { - let (partition_named, named_chunks) = match self - .named_warmup_files(partition_id, partition_file, chunk_ids) + /// Files that are gone by the time the pass reaches them are the normal case: the walk takes as + /// long as it takes and compaction removes what it replaces meanwhile. Files the metastore still + /// names are really missing, which is worth an error of its own, reported per batch so that a + /// table whose objects an operator or a lifecycle rule really removed does not bring back the + /// flood of lines this check exists to remove. + async fn report_absent_warmup_files(&self, absent: AbsentWarmupFiles) { + if absent.len() == 0 { + return; + } + let (named_partitions, named_chunks) = match self + .named_warmup_files(&absent.partition_ids, &absent.chunk_ids) .await { Ok(named) => named, Err(e) => { - // Assuming either verdict would either hide missing data or report every file of a - // partition as lost, so report that the check itself did not happen. + // Assuming either verdict would either hide missing data or report a batch of live + // files as lost, so report that the check itself did not happen. log::warn!( - "Could not check {} absent warmup file(s) of partition {} against the metastore: {}", - chunk_ids.len() + partition_file as usize, - partition_id, + "Could not check {} absent warmup file(s) against the metastore: {}", + absent.len(), e ); return; } }; - let mut active = 0; - let mut stale = 0; - if partition_file { - if partition_named { - active += 1; - log::error!( - "Warmup file of active partition {} is missing in remote fs", - partition_id - ); - } else { - stale += 1; - } + let active_partitions = absent + .partition_ids + .iter() + .filter(|id| named_partitions.contains(id)) + .collect::>(); + if !active_partitions.is_empty() { + log::error!( + "Files of {} active partition(s) are missing in remote fs: {:?}", + active_partitions.len(), + active_partitions + ); } - for id in chunk_ids { - if named_chunks.contains(id) { - active += 1; - log::error!("Warmup file of active chunk {} is missing in remote fs", id); - } else { - stale += 1; - } + let active_chunks = absent + .chunk_ids + .iter() + .filter(|id| named_chunks.contains(id)) + .collect::>(); + if !active_chunks.is_empty() { + log::error!( + "Files of {} active chunk(s) are missing in remote fs: {:?}", + active_chunks.len(), + active_chunks + ); } + let active = active_partitions.len() + active_chunks.len(); + let stale = absent.len() - active; if 0 < active { - app_metrics::WARMUP_MISSING_ACTIVE.add(active); + app_metrics::WARMUP_MISSING_ACTIVE.add(active as i64); } if 0 < stale { - app_metrics::WARMUP_MISSING_STALE.add(stale); + app_metrics::WARMUP_MISSING_STALE.add(stale as i64); log::debug!( - "Skipping {} warmup file(s) of partition {} replaced since the snapshot", - stale, - partition_id + "Skipping {} warmup file(s) replaced since the snapshot", + stale ); } } - /// Which of the files the metastore still names, read in one batch per partition so that a node - /// whose snapshot compaction has long moved on does not turn the check into a request per file. - /// The batch reads leave out the ids they no longer hold, which is what tells a deleted row - /// apart from a metastore we could not reach. + /// Which of the files the metastore still names. The batch reads leave out the ids they no + /// longer hold, which is what tells a deleted row apart from a metastore we could not reach. async fn named_warmup_files( &self, - partition_id: u64, - partition_file: bool, + partition_ids: &[u64], chunk_ids: &[u64], - ) -> Result<(bool, HashSet), CubeError> { - let partition_named = if partition_file { - self.meta_store - .get_partitions_out_of_queue(vec![partition_id]) + ) -> Result<(HashSet, HashSet), CubeError> { + let mut named_partitions = HashSet::new(); + if !partition_ids.is_empty() { + for p in self + .meta_store + .get_partitions_out_of_queue(partition_ids.to_vec()) .await? - .first() - .map_or(false, |p| p.get_row().has_main_table_file()) - } else { - false - }; + { + if p.get_row().has_main_table_file() { + named_partitions.insert(p.get_id()); + } + } + } let mut named_chunks = HashSet::new(); if !chunk_ids.is_empty() { @@ -2265,7 +2273,20 @@ impl ClusterImpl { } } } - Ok((partition_named, named_chunks)) + Ok((named_partitions, named_chunks)) + } +} + +/// Warmup files that turned out to be absent, waiting to be checked against the metastore. +#[derive(Default)] +struct AbsentWarmupFiles { + partition_ids: Vec, + chunk_ids: Vec, +} + +impl AbsentWarmupFiles { + fn len(&self) -> usize { + self.partition_ids.len() + self.chunk_ids.len() } } @@ -2472,12 +2493,12 @@ mod tests { .await?[0] .get_id(); - let (partition_named, named_chunks) = cluster - .named_warmup_files(partition_id, true, &[chunk_id]) + let (named_partitions, named_chunks) = cluster + .named_warmup_files(&[partition_id], &[chunk_id]) .await?; assert!(named_chunks.contains(&chunk_id)); // The partition holds no data of its own until it is compacted. - assert!(!partition_named); + assert!(named_partitions.is_empty()); services .injector @@ -2495,14 +2516,12 @@ mod tests { .remove(0) .get_id(); assert_ne!(compacted_id, partition_id); - assert!(cluster.named_warmup_files(compacted_id, true, &[]).await?.0); - // An id the metastore no longer holds at all is left out of the batch read, the // same as one it holds as inactive. - let (partition_named, named_chunks) = cluster - .named_warmup_files(partition_id, true, &[chunk_id, chunk_id + 1000]) + let (named_partitions, named_chunks) = cluster + .named_warmup_files(&[partition_id, compacted_id], &[chunk_id, chunk_id + 1000]) .await?; - assert!(!partition_named); + assert_eq!(named_partitions, HashSet::from([compacted_id])); assert!(named_chunks.is_empty()); Ok::<(), CubeError>(()) From 644af58ce479a393169421492c6b64c2a9606dea Mon Sep 17 00:00:00 2001 From: Aleksandr Romanenko Date: Wed, 19 Aug 2026 18:28:55 +0200 Subject: [PATCH 6/7] fix(cubestore): keep the identities the warmup check reports Three ways the reporting lost the one thing an operator needs, the name of the file: A recheck the metastore could not answer dropped the batch with a count and no ids, and this runs at startup, when the link to the main node is likeliest to blip. There is no second pass, so the ids were the only trace of files that may be the genuinely missing ones. They are bounded by the batch, so log them. A download that failed for any other reason was reported as the bare error, which carries the path only when the error happens to be about the path - a size check does, a transport error may not. The caller knows it either way. A listing that failed while probing whether the object exists replaced the download error with its own, so the reason the download failed was lost and the classification the warmup now depends on turned into a guess. Let the download error stand and note the failed probe separately. Also name the predicate for what it does, since it reports as well as answers. Co-Authored-By: Claude Opus 5 (1M context) --- rust/cubestore/cubestore/src/cluster/mod.rs | 35 ++++++++++--------- .../cubestore/cubestore/src/remotefs/queue.rs | 21 +++++++---- 2 files changed, 34 insertions(+), 22 deletions(-) diff --git a/rust/cubestore/cubestore/src/cluster/mod.rs b/rust/cubestore/cubestore/src/cluster/mod.rs index 9892871ace4ce..ef2e4c220916e 100644 --- a/rust/cubestore/cubestore/src/cluster/mod.rs +++ b/rust/cubestore/cubestore/src/cluster/mod.rs @@ -2126,9 +2126,9 @@ impl ClusterImpl { } let result = self .remote_fs - .download_file(file, p.get_row().file_size()) + .download_file(file.clone(), p.get_row().file_size()) .await; - if self.warmup_file_is_absent(result) { + if self.report_unless_absent(result, &file) { absent.partition_ids.push(p.get_id()); } } @@ -2140,14 +2140,12 @@ impl ClusterImpl { if c.get_row().in_memory() { continue; } + let file = chunk_file_name(c.get_id(), c.get_row().suffix()); let result = self .remote_fs - .download_file( - chunk_file_name(c.get_id(), c.get_row().suffix()), - c.get_row().file_size(), - ) + .download_file(file.clone(), c.get_row().file_size()) .await; - if self.warmup_file_is_absent(result) { + if self.report_unless_absent(result, &file) { absent.chunk_ids.push(c.get_id()); } } @@ -2156,21 +2154,23 @@ impl ClusterImpl { .await; } } - // A cancelled pass drops what it has not looked up yet, since the next start walks the - // whole list again anyway and the metastore is on its way down with us. + // Only a pass that ran to the end gets here: the cancellation paths above leave their + // batch unchecked, since the next start walks the whole list again anyway and the metastore + // is on its way down with us. self.report_absent_warmup_files(absent).await; log::debug!("Startup warmup finished"); return; } - /// Reports everything but an absent file, which on its own says nothing: only the metastore can - /// tell whether the pass is simply behind or the data is gone. - fn warmup_file_is_absent(&self, result: Result) -> bool { + /// Reports a failed download and returns whether the reason was the file being absent, which on + /// its own says nothing: only the metastore can tell whether the pass is simply behind or the + /// data is gone, and it is asked in batches. + fn report_unless_absent(&self, result: Result, remote_path: &str) -> bool { match result { Ok(_) => false, Err(e) if e.is_file_not_found() => true, Err(e) => { - log::error!("Error: {:?}", e); + log::error!("Warmup of {} failed: {:?}", remote_path, e); false } } @@ -2194,9 +2194,12 @@ impl ClusterImpl { // Assuming either verdict would either hide missing data or report a batch of live // files as lost, so report that the check itself did not happen. log::warn!( - "Could not check {} absent warmup file(s) against the metastore: {}", + "Could not check {} absent warmup file(s) against the metastore: {}. \ + Unchecked partitions: {:?}, chunks: {:?}", absent.len(), - e + e, + absent.partition_ids, + absent.chunk_ids ); return; } @@ -2519,7 +2522,7 @@ mod tests { // An id the metastore no longer holds at all is left out of the batch read, the // same as one it holds as inactive. let (named_partitions, named_chunks) = cluster - .named_warmup_files(&[partition_id, compacted_id], &[chunk_id, chunk_id + 1000]) + .named_warmup_files(&[partition_id, compacted_id], &[chunk_id, u64::MAX]) .await?; assert_eq!(named_partitions, HashSet::from([compacted_id])); assert!(named_chunks.is_empty()); diff --git a/rust/cubestore/cubestore/src/remotefs/queue.rs b/rust/cubestore/cubestore/src/remotefs/queue.rs index 9e716019d1ea6..4f5208a06b4e1 100644 --- a/rust/cubestore/cubestore/src/remotefs/queue.rs +++ b/rust/cubestore/cubestore/src/remotefs/queue.rs @@ -290,12 +290,21 @@ impl RemoteFs for QueueRemoteFs { Err(err) => { // A listing is the only way to tell an absent object from a remote // that can't be reached at all: object stores answer 404 for both a - // missing key and a missing bucket. - if self.remote_fs.list(file.clone()).await?.is_empty() { - return Err(CubeError::file_not_found(format!( - "File {} doesn't exist in remote file system", - file - ))); + // missing key and a missing bucket. A listing that fails itself leaves + // the download error to speak for both, rather than replacing it. + match self.remote_fs.list(file.clone()).await { + Ok(listing) if listing.is_empty() => { + return Err(CubeError::file_not_found(format!( + "File {} doesn't exist in remote file system", + file + ))); + } + Ok(_) => {} + Err(list_err) => log::warn!( + "Could not check whether {} exists in remote fs: {}", + file, + list_err + ), } return Err(err); } From 6e3a78b56e272573f29a0283bc1d5be40061b616 Mon Sep 17 00:00:00 2001 From: Aleksandr Romanenko Date: Mon, 24 Aug 2026 18:21:46 +0200 Subject: [PATCH 7/7] fix(cubestore): leave the warmup to say only what it knows The pass was asking the metastore whether each absent file was still named, so that it could tell a file compaction had replaced from one that is really gone. That distinction is worth having, but not here: the query that needs a missing file reports it anyway, and it is better placed to judge, since by then there is no snapshot to be behind. A whole-cluster answer is also nearly free where the cleanup loop already holds both the remote listing and the metastore filenames, and it covers every file rather than one node's share of one snapshot. So the pass reports what it can actually tell: a file that is not there is a file it did not warm, at debug, with a counter that says how far behind the snapshot it ran. Everything else stays an error. This drops the batching, the id bookkeeping and the two metastore reads that came with the classification. Co-Authored-By: Claude Opus 5 (1M context) --- rust/cubestore/cubestore/src/app_metrics.rs | 8 +- rust/cubestore/cubestore/src/cluster/mod.rs | 225 ++------------------ 2 files changed, 17 insertions(+), 216 deletions(-) diff --git a/rust/cubestore/cubestore/src/app_metrics.rs b/rust/cubestore/cubestore/src/app_metrics.rs index e7fd6af48a848..02745f67926cc 100644 --- a/rust/cubestore/cubestore/src/app_metrics.rs +++ b/rust/cubestore/cubestore/src/app_metrics.rs @@ -148,11 +148,9 @@ pub static REMOTE_FS_FILES_TO_REMOVE: Gauge = metrics::gauge("cs.remote_fs.files pub static REMOTE_FS_FILES_SIZE_TO_REMOVE: Gauge = metrics::gauge("cs.remote_fs.files_to_remove.size"); -/// Startup warmup files that were gone by the time the pass reached them. `stale` counts the ones -/// whose metastore row compaction had already replaced, which is routine on a busy node; `active` -/// counts the ones an active row still points at, which means the data is really missing. -pub static WARMUP_MISSING_STALE: Counter = metrics::counter("cs.warmup.missing.stale"); -pub static WARMUP_MISSING_ACTIVE: Counter = metrics::counter("cs.warmup.missing.active"); +/// Startup warmup files that were gone by the time the pass reached them, which says how far +/// behind its snapshot a pass ran rather than that anything is wrong. +pub static WARMUP_MISSING: Counter = metrics::counter("cs.warmup.missing"); /// Cache Store Cache pub static CACHESTORE_TTL_PERSIST: Counter = metrics::counter("cs.cachestore.ttl.persist"); diff --git a/rust/cubestore/cubestore/src/cluster/mod.rs b/rust/cubestore/cubestore/src/cluster/mod.rs index ef2e4c220916e..86110d43462a5 100644 --- a/rust/cubestore/cubestore/src/cluster/mod.rs +++ b/rust/cubestore/cubestore/src/cluster/mod.rs @@ -67,7 +67,7 @@ use opentelemetry::Context as OtelContext; use regex::Regex; use serde::{Deserialize, Serialize}; use std::collections::hash_map::DefaultHasher; -use std::collections::{HashMap, HashSet}; +use std::collections::HashMap; use std::hash::{Hash, Hasher}; use std::pin::Pin; use std::sync::Weak; @@ -2109,12 +2109,6 @@ impl ClusterImpl { }; log::debug!("Got {} partitions, running the warmup", partitions.len()); - // Absent files are looked up in batches: the pass is expected to find plenty of them, and - // the metastore of a select worker is a request to the main node, which is busy with the - // rest of the cluster starting up. - const ABSENT_BATCH: usize = 256; - let mut absent = AbsentWarmupFiles::default(); - for (p, chunks) in partitions { if self.node_name_by_partition(&p) != self.server_name { continue; @@ -2128,9 +2122,7 @@ impl ClusterImpl { .remote_fs .download_file(file.clone(), p.get_row().file_size()) .await; - if self.report_unless_absent(result, &file) { - absent.partition_ids.push(p.get_id()); - } + self.report_warmup_download(result, &file); } for c in chunks { if self.stop_token.is_cancelled() { @@ -2145,151 +2137,27 @@ impl ClusterImpl { .remote_fs .download_file(file.clone(), c.get_row().file_size()) .await; - if self.report_unless_absent(result, &file) { - absent.chunk_ids.push(c.get_id()); - } - } - if ABSENT_BATCH <= absent.len() { - self.report_absent_warmup_files(std::mem::take(&mut absent)) - .await; + self.report_warmup_download(result, &file); } } - // Only a pass that ran to the end gets here: the cancellation paths above leave their - // batch unchecked, since the next start walks the whole list again anyway and the metastore - // is on its way down with us. - self.report_absent_warmup_files(absent).await; log::debug!("Startup warmup finished"); return; } - /// Reports a failed download and returns whether the reason was the file being absent, which on - /// its own says nothing: only the metastore can tell whether the pass is simply behind or the - /// data is gone, and it is asked in batches. - fn report_unless_absent(&self, result: Result, remote_path: &str) -> bool { + /// A file that is gone by the time the pass reaches it is the normal case: the walk takes as + /// long as it takes and compaction removes what it replaces meanwhile, so its absence says how + /// far behind the snapshot the pass ran rather than that anything is wrong. A file that is + /// really missing is reported by the query that needs it, which is also the one that can tell, + /// since by then the metastore either still names the file or does not. + fn report_warmup_download(&self, result: Result, remote_path: &str) { match result { - Ok(_) => false, - Err(e) if e.is_file_not_found() => true, - Err(e) => { - log::error!("Warmup of {} failed: {:?}", remote_path, e); - false - } - } - } - - /// Files that are gone by the time the pass reaches them are the normal case: the walk takes as - /// long as it takes and compaction removes what it replaces meanwhile. Files the metastore still - /// names are really missing, which is worth an error of its own, reported per batch so that a - /// table whose objects an operator or a lifecycle rule really removed does not bring back the - /// flood of lines this check exists to remove. - async fn report_absent_warmup_files(&self, absent: AbsentWarmupFiles) { - if absent.len() == 0 { - return; - } - let (named_partitions, named_chunks) = match self - .named_warmup_files(&absent.partition_ids, &absent.chunk_ids) - .await - { - Ok(named) => named, - Err(e) => { - // Assuming either verdict would either hide missing data or report a batch of live - // files as lost, so report that the check itself did not happen. - log::warn!( - "Could not check {} absent warmup file(s) against the metastore: {}. \ - Unchecked partitions: {:?}, chunks: {:?}", - absent.len(), - e, - absent.partition_ids, - absent.chunk_ids - ); - return; - } - }; - - let active_partitions = absent - .partition_ids - .iter() - .filter(|id| named_partitions.contains(id)) - .collect::>(); - if !active_partitions.is_empty() { - log::error!( - "Files of {} active partition(s) are missing in remote fs: {:?}", - active_partitions.len(), - active_partitions - ); - } - let active_chunks = absent - .chunk_ids - .iter() - .filter(|id| named_chunks.contains(id)) - .collect::>(); - if !active_chunks.is_empty() { - log::error!( - "Files of {} active chunk(s) are missing in remote fs: {:?}", - active_chunks.len(), - active_chunks - ); - } - - let active = active_partitions.len() + active_chunks.len(); - let stale = absent.len() - active; - if 0 < active { - app_metrics::WARMUP_MISSING_ACTIVE.add(active as i64); - } - if 0 < stale { - app_metrics::WARMUP_MISSING_STALE.add(stale as i64); - log::debug!( - "Skipping {} warmup file(s) replaced since the snapshot", - stale - ); - } - } - - /// Which of the files the metastore still names. The batch reads leave out the ids they no - /// longer hold, which is what tells a deleted row apart from a metastore we could not reach. - async fn named_warmup_files( - &self, - partition_ids: &[u64], - chunk_ids: &[u64], - ) -> Result<(HashSet, HashSet), CubeError> { - let mut named_partitions = HashSet::new(); - if !partition_ids.is_empty() { - for p in self - .meta_store - .get_partitions_out_of_queue(partition_ids.to_vec()) - .await? - { - if p.get_row().has_main_table_file() { - named_partitions.insert(p.get_id()); - } + Ok(_) => {} + Err(e) if e.is_file_not_found() => { + app_metrics::WARMUP_MISSING.increment(); + log::debug!("Skipping warmup of {}: {}", remote_path, e.message); } + Err(e) => log::error!("Warmup of {} failed: {:?}", remote_path, e), } - - let mut named_chunks = HashSet::new(); - if !chunk_ids.is_empty() { - for c in self - .meta_store - .get_chunks_out_of_queue(chunk_ids.to_vec()) - .await? - { - if c.get_row().active() { - named_chunks.insert(c.get_id()); - } - } - } - Ok((named_partitions, named_chunks)) - } -} - -/// Warmup files that turned out to be absent, waiting to be checked against the metastore. -#[derive(Default)] -struct AbsentWarmupFiles { - partition_ids: Vec, - chunk_ids: Vec, -} - -impl AbsentWarmupFiles { - fn len(&self) -> usize { - self.partition_ids.len() + self.chunk_ids.len() } } @@ -2456,8 +2324,6 @@ pub fn pick_worker_by_partitions<'a>( mod tests { use super::*; use crate::config::Config; - use crate::queryplanner::trace_data_loaded::DataLoadedSize; - use crate::store::compaction::CompactionService; use std::fs; fn config_with_workers(name: &str, workers: Vec) -> Arc { @@ -2469,69 +2335,6 @@ mod tests { .config_obj() } - /// Compaction replacing a chunk with a partition file is exactly the race the warmup loses when - /// it reaches a file hours after the snapshot that named it. - #[tokio::test] - async fn warmup_file_reference_follows_the_metastore() { - Config::test("warmup_file_reference") - .start_test(async move |services| { - let service = services.sql_service; - let meta_store = services.meta_store; - let cluster = services.cluster; - - service.exec_query("CREATE SCHEMA test").await?; - service.exec_query("CREATE TABLE test.warm (a int)").await?; - service - .exec_query("INSERT INTO test.warm (a) VALUES (1), (2)") - .await?; - - let partition = meta_store - .get_partitions_with_chunks_created_seconds_ago(-1000) - .await? - .remove(0); - let partition_id = partition.get_id(); - let index_id = partition.get_row().get_index_id(); - let chunk_id = meta_store - .get_chunks_by_partition(partition_id, false) - .await?[0] - .get_id(); - - let (named_partitions, named_chunks) = cluster - .named_warmup_files(&[partition_id], &[chunk_id]) - .await?; - assert!(named_chunks.contains(&chunk_id)); - // The partition holds no data of its own until it is compacted. - assert!(named_partitions.is_empty()); - - services - .injector - .get_service_typed::() - .await - .compact(partition_id, DataLoadedSize::new()) - .await?; - - // Compaction moved the rows into a partition file of its own and left the chunk - // and its parent behind, which is what makes their files disappear under a pass - // that is still walking an older snapshot. - let compacted_id = meta_store - .get_active_partitions_by_index_id(index_id) - .await? - .remove(0) - .get_id(); - assert_ne!(compacted_id, partition_id); - // An id the metastore no longer holds at all is left out of the batch read, the - // same as one it holds as inactive. - let (named_partitions, named_chunks) = cluster - .named_warmup_files(&[partition_id, compacted_id], &[chunk_id, u64::MAX]) - .await?; - assert_eq!(named_partitions, HashSet::from([compacted_id])); - assert!(named_chunks.is_empty()); - - Ok::<(), CubeError>(()) - }) - .await; - } - #[test] fn pick_import_worker_load_aware() { let config = config_with_workers(