From 4b6da6bd9c84d1fc51a8ce218c879049095ec979 Mon Sep 17 00:00:00 2001 From: tprevot Date: Sat, 17 Oct 2026 21:01:29 +0200 Subject: [PATCH 1/4] feat: metastore recovery cli --- docs/operating/metastore-recovery.md | 98 ++++ quickwit/quickwit-cli/src/tool.rs | 248 +++++++++- .../quickwit-index-management/src/index.rs | 427 +++++++++++++++++- quickwit/quickwit-index-management/src/lib.rs | 5 +- 4 files changed, 765 insertions(+), 13 deletions(-) create mode 100644 docs/operating/metastore-recovery.md diff --git a/docs/operating/metastore-recovery.md b/docs/operating/metastore-recovery.md new file mode 100644 index 00000000000..51c7a3aaed7 --- /dev/null +++ b/docs/operating/metastore-recovery.md @@ -0,0 +1,98 @@ +--- +title: Metastore recovery +sidebar_position: 5 +--- + +In case of disaster recovery you can rebuild the metastore state from +split recovery metadata. +The `quickwit tool recover` command reads that embedded metadata from each split and republishes +the split metadata into an existing index. + +## Requirements + +- **The index storage must support object listing.** Recovery discovers splits by listing the + index URI, which only S3-compatible storages implement. A `file://` index cannot be recovered + this way. +- **The splits must embed the recovery metadata.** It is written on every upload, but splits + produced by an older version of Quickwit do not have it and are skipped. +- **The index must exist, with the doc mapping the splits were built with.** Recovery restores + splits, not the index configuration. A mismatched doc mapping yields an index that is searchable + but returns wrong or missing fields for the recovered splits. + +## Recovering an index + +### 1. Recreate the index + +Create the index with the same doc mapping as before the loss, and make sure its `index_uri` +points at the directory holding the splits: + +```bash +curl -X POST http://localhost:7280/api/v1/indexes \ + -H "Content-Type: application/yaml" \ + --data-binary @index-config.yaml +``` + +### 2. Preview the recovery + +```bash +quickwit tool recover --index otel-logs-v0_9 --config ./config/quickwit.yaml --dry-run +``` + +``` +Recovering index otel-logs-v0_9 from s3://my-bucket/indexes/otel-logs-v0_9 + ++-------------------------------------+-------+ +| Recovery summary (dry run) | ++-------------------------------------+-------+ +| Splits found in storage | 12 | +| Publishable | 5 | +| Documents | 1,099 | +| Skipped (merged into another split) | 0 | +| Skipped (already in metastore) | 6 | +| Unreadable | 1 | ++-------------------------------------+-------+ +``` + +The command then lists the splits it would publish, the IDs it skipped, and the split files it +could not read along with the reason. + +### 3. Publish the splits + +Run the same command without `--dry-run`. The splits are staged and published in batches. + +### 4. Verify + +Compare the hit count over the time range covered by the recovered splits with the document count +reported by the command: + +```bash +curl "http://localhost:7280/api/v1/otel-logs-v0_9/search?query=*&max_hits=0\ +&start_timestamp=1792257120&end_timestamp=1792258260" +``` + +## What the command does + +1. Lists the `.split` objects at the root of the index URI. +2. Fetches the recovery metadata of each split, ten splits at a time. The object size comes from + the listing, and the footer and the metadata file usually land in a single `GET` per split. +3. Rebuilds the split metadata, replacing the index UID recorded in the split with the UID of the + target index. +4. Skips the splits that were superseded by a merge. A merged split and its parents coexist in the + storage until garbage collection removes the parents. +5. Leaves out the splits the metastore already knows about, which makes a second run a no-op. + Splits left staged by an interrupted run are not considered known: they are staged and + published again. +6. Stages and publishes the remaining splits, 1000 per request. + +Splits whose metadata cannot be read are reported individually and do not interrupt the run: a +single truncated upload would otherwise make every other split unrecoverable. + +## What is not recovered + +- **Source checkpoints.** The index checkpoint is left empty. A source restarted against a + recovered index resumes from the position its own configuration dictates, which may re-ingest or + skip data depending on the source. +- **Delete tasks.** The delete opstamp recorded in each split is restored, but the delete task + queue itself is not. Pending deletes have to be submitted again. +- **Splits without embedded recovery metadata.** They are listed as unreadable and left in the + storage, untouched. diff --git a/quickwit/quickwit-cli/src/tool.rs b/quickwit/quickwit-cli/src/tool.rs index dfff0b937c2..5e6625ccff3 100644 --- a/quickwit/quickwit-cli/src/tool.rs +++ b/quickwit/quickwit-cli/src/tool.rs @@ -22,11 +22,13 @@ use std::time::{Duration, Instant}; use std::{env, fmt, io}; use anyhow::{Context, bail}; +use bytesize::ByteSize; use clap::{ArgMatches, Command, arg}; use colored::{ColoredString, Colorize}; use humantime::format_duration; use quickwit_actors::{ActorExitStatus, ActorHandle, Mailbox, Universe}; use quickwit_cluster::{ChitchatTransport, Cluster, ClusterMember, FailureDetectorConfig}; +use quickwit_common::pretty::PrettySample; use quickwit_common::pubsub::EventBroker; use quickwit_common::runtimes::RuntimesConfig; use quickwit_common::uri::Uri; @@ -35,7 +37,7 @@ use quickwit_config::{ CLI_SOURCE_ID, IndexerConfig, NodeConfig, SourceConfig, SourceInputFormat, SourceParams, TransformConfig, VecSourceParams, }; -use quickwit_index_management::{IndexService, clear_cache_directory}; +use quickwit_index_management::{IndexService, SplitRecoveryInfo, clear_cache_directory}; use quickwit_indexing::actors::{IndexingService, MergePipeline, MergeSchedulerService}; use quickwit_indexing::docs_clustering::Fingerprinter; use quickwit_indexing::models::{ @@ -43,7 +45,7 @@ use quickwit_indexing::models::{ }; use quickwit_indexing::{IndexingPipeline, IndexingSplitCache}; use quickwit_ingest::IngesterPool; -use quickwit_metastore::IndexMetadataResponseExt; +use quickwit_metastore::{IndexMetadataResponseExt, SplitMetadata}; use quickwit_proto::indexing::CpuCapacity; use quickwit_proto::ingest::ingester::IngesterStatus; use quickwit_proto::metastore::{IndexMetadataRequest, MetastoreService, MetastoreServiceClient}; @@ -53,14 +55,17 @@ use quickwit_search::{SearchResponseRest, single_node_search}; use quickwit_serve::{ BodyFormat, SearchRequestQueryString, SortBy, search_request_from_api_request, }; -use quickwit_storage::{BundleStorage, Storage}; +use quickwit_storage::{BundleStorage, Storage, StorageResolver}; use quickwit_transport::ChannelFactory; +use tabled::Tabled; use thousands::Separable; +use time::OffsetDateTime; +use time::format_description::well_known::Rfc3339; use tracing::debug; -use crate::checklist::{GREEN_COLOR, RED_COLOR}; +use crate::checklist::{GREEN_COLOR, RED_COLOR, WHITE_COLOR}; use crate::{ - THROUGHPUT_WINDOW_SIZE, config_cli_arg, get_resolvers, info, load_node_config, + THROUGHPUT_WINDOW_SIZE, config_cli_arg, get_resolvers, info, load_node_config, make_table, run_index_checklist, start_actor_runtimes, }; @@ -165,9 +170,25 @@ pub fn build_tool_command() -> Command { .required(true), ]) ) + .subcommand(Command::new("recover").display_order(11) + .about("Recovers an index from the metadata embedded in its splits. Only splits written with a footer trailer (QW_ENABLE_SPLIT_FOOTER_TRAILER=true) are recovered. Requires an S3-compatible index storage.").args(&[ + arg!(--index "ID of the target index") + .display_order(1) + .required(true), + arg!(--"dry-run" "Executes the command in dry run mode and only displays the splits that are going to be added to the index") + .required(false), + ]) + ) .arg_required_else_help(true) } +#[derive(Debug, Eq, PartialEq)] +pub struct RecoverIndexArgs { + pub config_uri: Uri, + pub index_id: IndexId, + pub dry_run: bool, +} + #[derive(Debug, Eq, PartialEq)] pub struct LocalIngestDocsArgs { pub config_uri: Uri, @@ -223,6 +244,7 @@ pub enum ToolCliCommand { LocalIngest(LocalIngestDocsArgs), LocalSearch(LocalSearchArgs), ExtractSplit(ExtractSplitArgs), + RecoverIndex(RecoverIndexArgs), Merge(MergeArgs), } @@ -237,6 +259,7 @@ impl ToolCliCommand { "local-search" => Self::parse_local_search_args(submatches), "merge" => Self::parse_merge_args(submatches), "extract-split" => Self::parse_extract_split_args(submatches), + "recover" => Self::parse_recover_index_args(submatches), _ => bail!("unknown tool subcommand `{subcommand}`"), } } @@ -364,6 +387,23 @@ impl ToolCliCommand { })) } + fn parse_recover_index_args(mut matches: ArgMatches) -> anyhow::Result { + let config_uri = matches + .get_one("config") + .map(|uri_str: &String| Uri::from_str(uri_str)) + .expect("`config` should be a required arg.")?; + let index_id = matches + .remove_one::("index") + .expect("`index` should be a required arg."); + let dry_run = matches.get_flag("dry-run"); + + Ok(Self::RecoverIndex(RecoverIndexArgs { + config_uri, + index_id, + dry_run, + })) + } + fn parse_extract_split_args(mut matches: ArgMatches) -> anyhow::Result { let index_id = matches .remove_one::("index") @@ -395,6 +435,7 @@ impl ToolCliCommand { Self::LocalSearch(args) => local_search_cli(args).await, Self::Merge(args) => merge_cli(args).await, Self::ExtractSplit(args) => extract_split_cli(args).await, + Self::RecoverIndex(args) => recover_index_cli(args).await, } } } @@ -685,6 +726,203 @@ pub async fn garbage_collect_index_cli(args: GarbageCollectIndexArgs) -> anyhow: print_tantivy_gc_result(args.dry_run, removal_info) } +pub async fn recover_index_cli(args: RecoverIndexArgs) -> anyhow::Result<()> { + let config = load_node_config(&args.config_uri, None).await?; + let (storage_resolver, metastore_resolver) = + get_resolvers(&config.storage_configs, &config.metastore_configs); + let metastore = metastore_resolver.resolve(&config.metastore_uri).await?; + let mut index_service = IndexService::new(metastore.clone(), storage_resolver); + + let index_metadata_request = IndexMetadataRequest::for_index_id(args.index_id.to_string()); + let index_metadata = metastore + .index_metadata(index_metadata_request) + .await? + .deserialize_index_metadata()?; + + let storage = StorageResolver::configured(&config.storage_configs) + .resolve(&index_metadata.index_config().index_uri) + .await?; + let index_uri = index_metadata.index_config().index_uri.clone(); + println!( + "Recovering index {} from {}\n", + args.index_id.bold(), + index_uri.to_string().bold() + ); + let recovery_info = index_service + .recover_index_from_storage(&args.index_id, storage, args.dry_run) + .await?; + + print_split_recovery_info(&args.index_id, &recovery_info); + Ok(()) +} + +#[derive(Tabled)] +struct RecoveredSplitRow { + #[tabled(rename = "Split ID")] + split_id: String, + #[tabled(rename = "Docs")] + num_docs: String, + #[tabled(rename = "Split size")] + split_size: String, + #[tabled(rename = "Time range")] + time_range: String, + #[tabled(rename = "Created at")] + created_at: String, + #[tabled(rename = "Merge ops")] + num_merge_ops: usize, + #[tabled(rename = "Doc mapping UID")] + doc_mapping_uid: String, +} + +impl From<&SplitMetadata> for RecoveredSplitRow { + fn from(split_metadata: &SplitMetadata) -> Self { + // The footer ends at the very end of the split file, so its end offset is the file size. + let split_size = ByteSize::b(split_metadata.footer_offsets.end); + RecoveredSplitRow { + split_id: split_metadata.split_id().to_string(), + num_docs: split_metadata.num_docs.separate_with_commas(), + split_size: split_size.to_string(), + time_range: format_time_range(split_metadata), + created_at: format_timestamp(split_metadata.create_timestamp), + num_merge_ops: split_metadata.num_merge_ops, + doc_mapping_uid: split_metadata.doc_mapping_uid.to_string(), + } + } +} + +fn format_timestamp(unix_timestamp: i64) -> String { + let Ok(date_time) = OffsetDateTime::from_unix_timestamp(unix_timestamp) else { + return unix_timestamp.to_string(); + }; + date_time + .format(&Rfc3339) + .unwrap_or_else(|_| unix_timestamp.to_string()) +} + +fn format_time_range(split_metadata: &SplitMetadata) -> String { + let Some(time_range) = &split_metadata.time_range else { + return "*".to_string(); + }; + format!( + "{} → {}", + format_timestamp(*time_range.start()), + format_timestamp(*time_range.end()) + ) +} + +#[derive(Tabled)] +struct SplitRecoverySummaryRow { + #[tabled(rename = "Splits found in storage")] + num_split_files: String, + #[tabled(rename = "Publishable")] + num_published: String, + #[tabled(rename = "Documents")] + num_docs_published: String, + #[tabled(rename = "Skipped (merged into another split)")] + num_merged: String, + #[tabled(rename = "Skipped (already in metastore)")] + num_already_known: String, + #[tabled(rename = "Unreadable")] + num_unreadable: String, +} + +fn print_split_recovery_info(index_id: &str, recovery_info: &SplitRecoveryInfo) { + let num_published = recovery_info.published_splits.len(); + let num_unreadable = recovery_info.unreadable_splits.len(); + + let summary_row = SplitRecoverySummaryRow { + num_split_files: recovery_info.num_split_files().separate_with_commas(), + num_published: num_published.separate_with_commas(), + num_docs_published: recovery_info.num_docs_published().separate_with_commas(), + num_merged: recovery_info.merged_split_ids.len().separate_with_commas(), + num_already_known: recovery_info + .already_known_split_ids + .len() + .separate_with_commas(), + num_unreadable: num_unreadable.separate_with_commas(), + }; + let summary_header = if recovery_info.dry_run { + "Recovery summary (dry run)" + } else { + "Recovery summary" + }; + println!("{}", make_table(summary_header, [summary_row], true)); + + if !recovery_info.published_splits.is_empty() { + let splits_header = if recovery_info.dry_run { + "Splits that would be published" + } else { + "Published splits" + }; + let rows = recovery_info + .published_splits + .iter() + .map(RecoveredSplitRow::from); + println!("\n{}", make_table(splits_header, rows, false)); + } + print_skipped_split_ids( + "merged into another recovered split", + &recovery_info.merged_split_ids, + ); + print_skipped_split_ids( + "already known to the metastore", + &recovery_info.already_known_split_ids, + ); + + if num_unreadable > 0 { + println!( + "\n{} {} split file(s) could not be read:", + "✖".color(RED_COLOR), + num_unreadable + ); + for unreadable_split in &recovery_info.unreadable_splits { + println!( + " {} {}: {:#}", + "✖".color(RED_COLOR), + unreadable_split.split_file.display(), + unreadable_split.error + ); + } + } + + println!(); + if recovery_info.dry_run { + println!( + "{} Dry run: nothing was published. Re-run without `--dry-run` to publish {} split(s) \ + into `{index_id}`.", + "!".color(WHITE_COLOR), + num_published + ); + return; + } + if num_published == 0 { + println!( + "{} No split to publish into `{index_id}`.", + "✔".color(GREEN_COLOR) + ); + return; + } + println!( + "{} Published {} split(s) holding {} document(s) into `{index_id}`.", + "✔".color(GREEN_COLOR), + num_published.separate_with_commas(), + recovery_info.num_docs_published().separate_with_commas() + ); +} + +/// Prints the IDs of the splits left out of the publication, sampled to keep the output readable. +fn print_skipped_split_ids(reason: &str, split_ids: &[SplitId]) { + if split_ids.is_empty() { + return; + } + println!( + "\n{} {} split(s) skipped, {reason}: {:?}", + "-".color(WHITE_COLOR), + split_ids.len(), + PrettySample::new(split_ids, 10) + ); +} + fn print_tantivy_gc_result( dry_run: bool, removal_info: quickwit_index_management::SplitRemovalInfo, diff --git a/quickwit/quickwit-index-management/src/index.rs b/quickwit/quickwit-index-management/src/index.rs index 4c49bd3b5e6..3d281f05d35 100644 --- a/quickwit/quickwit-index-management/src/index.rs +++ b/quickwit/quickwit-index-management/src/index.rs @@ -13,33 +13,41 @@ // limitations under the License. use std::collections::{HashMap, HashSet}; -use std::path::Path; +use std::ffi::OsStr; +use std::path::{Path, PathBuf}; +use std::sync::Arc; use std::time::Duration; +use anyhow::Context; +use futures::TryStreamExt; use futures_util::StreamExt; use itertools::Itertools; use quickwit_common::fs::{empty_dir, get_cache_directory_path}; use quickwit_common::pretty::PrettySample; use quickwit_common::rate_limited_error; +use quickwit_common::shared_consts::SPLIT_RECOVERY_METADATA_FILE_NAME; use quickwit_config::{IndexConfig, SourceConfig, validate_identifier}; use quickwit_indexing::check_source_connectivity; use quickwit_metastore::{ AddSourceRequestExt, CreateIndexResponseExt, IndexMetadata, IndexMetadataResponseExt, ListIndexesMetadataResponseExt, ListSplitsQuery, ListSplitsRequestExt, - MetastoreServiceStreamSplitsExt, SplitInfo, SplitMetadata, SplitState, UpdateIndexRequestExt, - UpdateSourceRequestExt, + MetastoreServiceStreamSplitsExt, SplitInfo, SplitMetadata, SplitState, StageSplitsRequestExt, + UpdateIndexRequestExt, UpdateSourceRequestExt, }; use quickwit_proto::metastore::{ AddSourceRequest, CreateIndexRequest, DeleteIndexRequest, EntityKind, IndexMetadataRequest, ListIndexesMetadataRequest, ListSplitsRequest, MarkSplitsForDeletionRequest, MetastoreError, - MetastoreService, MetastoreServiceClient, ResetSourceCheckpointRequest, UpdateIndexRequest, - UpdateSourceRequest, serde_utils, + MetastoreService, MetastoreServiceClient, PublishSplitsRequest, ResetSourceCheckpointRequest, + SplitRecoveryMetadata, StageSplitsRequest, UpdateIndexRequest, UpdateSourceRequest, + serde_utils, }; use quickwit_proto::types::{IndexUid, SplitId}; use quickwit_proto::{ServiceError, ServiceErrorCode}; -use quickwit_storage::{StorageResolver, StorageResolverError}; +use quickwit_storage::{ + BundleStorage, ObjectMetadata, Storage, StorageResolver, StorageResolverError, +}; use thiserror::Error; -use tracing::{error, info}; +use tracing::{error, info, warn}; use crate::garbage_collection::{ DeleteSplitsError, SplitRemovalInfo, delete_splits_from_storage_and_metastore, @@ -563,6 +571,304 @@ impl IndexService { Ok(source_config) } + + pub async fn recover_index_from_storage( + &mut self, + index_id: &str, + storage: Arc, + dry_run: bool, + ) -> Result { + let index_metadata_request = IndexMetadataRequest::for_index_id(index_id.to_string()); + let index_uid = self + .metastore + .index_metadata(index_metadata_request) + .await? + .deserialize_index_metadata()? + .index_uid; + + let read_outcome = read_splits_recovery_metadata(&storage).await?; + + if !read_outcome.unreadable_splits.is_empty() { + warn!( + index_id=%index_id, + num_unreadable_splits=read_outcome.unreadable_splits.len(), + "some split files could not be read" + ); + } + info!( + index_id=%index_id, + dry_run=dry_run, + num_recovered_splits=read_outcome.recovered_splits.len(), + num_unreadable_splits=read_outcome.unreadable_splits.len(), + "read split recovery metadata" + ); + + let known_split_ids: HashSet = + self.list_metastore_split_ids(index_uid.clone()).await?; + let plan = plan_split_recovery(&index_uid, read_outcome.recovered_splits, &known_split_ids); + + let recovery_info = SplitRecoveryInfo { + dry_run, + published_splits: plan.splits_to_publish, + merged_split_ids: plan.merged_split_ids, + already_known_split_ids: plan.already_known_split_ids, + unreadable_splits: read_outcome.unreadable_splits, + }; + if dry_run { + return Ok(recovery_info); + } + + for splits_batch in recovery_info + .published_splits + .chunks(PUBLISH_SPLITS_BATCH_SIZE) + { + self.publish_recovered_splits(&index_uid, splits_batch) + .await?; + } + info!( + index_id=%index_id, + num_published_splits=recovery_info.published_splits.len(), + num_published_docs=recovery_info.num_docs_published(), + "published recovered splits" + ); + Ok(recovery_info) + } + + /// Lists the split ids the metastore already holds in a final state. + /// + /// Staged splits are left out on purpose: they are leftovers of a run that failed between + /// staging and publishing, and the next run must stage them again (which the metastore allows) + /// and publish them. Counting them as known would skip them forever and let the janitor delete + /// their files once the staged grace period elapses. + async fn list_metastore_split_ids( + &mut self, + index_uid: IndexUid, + ) -> Result, IndexServiceError> { + let list_splits_query = ListSplitsQuery::for_index(index_uid) + .with_split_states([SplitState::Published, SplitState::MarkedForDeletion]); + let list_splits_request = + ListSplitsRequest::try_from_list_splits_query(&list_splits_query)?; + let split_ids: Vec = self + .metastore + .list_splits(list_splits_request) + .await? + .collect_split_ids() + .await?; + Ok(split_ids.into_iter().collect()) + } + + /// Stages then publishes a batch of recovered splits. + /// + /// `replaced_split_ids` is left empty on purpose: the splits these ones superseded were never + /// published in the rebuilt metastore, so there is nothing to mark for deletion. They are kept + /// out of the publication plan instead. + async fn publish_recovered_splits( + &mut self, + index_uid: &IndexUid, + splits_metadata: &[SplitMetadata], + ) -> Result<(), IndexServiceError> { + let stage_splits_request = StageSplitsRequest::try_from_splits_metadata( + index_uid.clone(), + splits_metadata.iter().cloned(), + )?; + self.metastore.stage_splits(stage_splits_request).await?; + + let publish_splits_request = PublishSplitsRequest { + index_uid: Some(index_uid.clone()), + staged_split_ids: splits_metadata + .iter() + .map(|split_metadata| split_metadata.split_id().to_string()) + .collect(), + replaced_split_ids: Vec::new(), + index_checkpoint_delta_json_opt: None, + publish_token_opt: None, + }; + self.metastore + .publish_splits(publish_splits_request) + .await?; + Ok(()) + } +} + +/// Splits to publish, and the counts of those left out. +struct SplitRecoveryPlan { + splits_to_publish: Vec, + merged_split_ids: Vec, + already_known_split_ids: Vec, +} + +/// Decides which recovered splits should be published into `index_uid`. +/// +/// Ignores split ids already stored in the metastore (see `known_split_ids`) and splits that got +/// merged together and that still remain (probably with a pending deletion before the disaster). +fn plan_split_recovery( + index_uid: &IndexUid, + recovered_splits: Vec, + known_split_ids: &HashSet, +) -> SplitRecoveryPlan { + let merged_split_ids: HashSet<&SplitId> = recovered_splits + .iter() + .flat_map(|recovered_split| recovered_split.parent_split_ids.iter()) + .collect(); + + let mut plan = SplitRecoveryPlan { + splits_to_publish: Vec::new(), + merged_split_ids: Vec::new(), + already_known_split_ids: Vec::new(), + }; + for recovered_split in &recovered_splits { + let split_id = recovered_split.split_metadata.split_id(); + if merged_split_ids.contains(split_id) { + plan.merged_split_ids.push(split_id.clone()); + continue; + } + if known_split_ids.contains(split_id) { + plan.already_known_split_ids.push(split_id.clone()); + continue; + } + // The index_uid might be different so we are overwriting it with the new one. + let mut split_metadata = recovered_split.split_metadata.clone(); + split_metadata.index_uid = index_uid.clone(); + plan.splits_to_publish.push(split_metadata); + } + plan +} + +const READ_RECOVERY_METADATA_CONCURRENCY: usize = 10; + +const PUBLISH_SPLITS_BATCH_SIZE: usize = 1_000; + +#[derive(Debug)] +struct RecoveredSplit { + split_metadata: SplitMetadata, + parent_split_ids: Vec, +} + +/// A split file whose recovery metadata could not be read. +#[derive(Debug)] +pub struct UnreadableSplit { + /// Path of the split file, relative to the index URI. + pub split_file: PathBuf, + /// Why the recovery metadata could not be read. + pub error: anyhow::Error, +} + +#[derive(Debug, Default)] +struct SplitsRecoveryMetadata { + pub recovered_splits: Vec, + pub unreadable_splits: Vec, +} + +pub struct SplitRecoveryInfo { + /// Whether the splits were only planned, and not actually published. + pub dry_run: bool, + /// Splits published into the index, or that a dry run would have published. + pub published_splits: Vec, + /// Recovered splits that was merged into another split. + pub merged_split_ids: Vec, + /// Recovered splits the metastore already knew about. + pub already_known_split_ids: Vec, + /// Split files whose recovery metadata could not be read. + pub unreadable_splits: Vec, +} + +impl SplitRecoveryInfo { + /// Number of split files found in the storage, whether their recovery metadata could be read + /// or not. + pub fn num_split_files(&self) -> usize { + self.published_splits.len() + + self.merged_split_ids.len() + + self.already_known_split_ids.len() + + self.unreadable_splits.len() + } + + /// Number of documents held by the published splits. + pub fn num_docs_published(&self) -> u64 { + self.published_splits + .iter() + .map(|split_metadata| split_metadata.num_docs as u64) + .sum() + } +} + +async fn read_splits_recovery_metadata( + storage: &Arc, +) -> Result { + let split_objects: Vec = list_split_objects(storage).await?; + + let read_results: Vec<(PathBuf, anyhow::Result)> = + futures::stream::iter(split_objects) + .map(|split_object| { + let storage = storage.clone(); + async move { + let split_file = split_object.path.clone(); + let read_result = read_split_recovery_metadata(storage, split_object).await; + (split_file, read_result) + } + }) + .buffer_unordered(READ_RECOVERY_METADATA_CONCURRENCY) + .collect() + .await; + + let mut outcome = SplitsRecoveryMetadata::default(); + for (split_file, read_result) in read_results { + match read_result { + Ok(recovered_split) => outcome.recovered_splits.push(recovered_split), + Err(error) => outcome + .unreadable_splits + .push(UnreadableSplit { split_file, error }), + } + } + Ok(outcome) +} + +async fn list_split_objects( + storage: &Arc, +) -> Result, IndexServiceError> { + let object_pages: Vec> = storage + .list(Path::new("")) + .try_collect() + .await + .map_err(|error| { + IndexServiceError::Internal(format!( + "failed to list objects in `{}`: {error}", + storage.uri() + )) + })?; + + let split_objects: Vec = object_pages + .into_iter() + .flatten() + .filter(|object| object.path.extension() == Some(OsStr::new("split"))) + .collect(); + + Ok(split_objects) +} + +async fn read_split_recovery_metadata( + storage: Arc, + split_object: ObjectMetadata, +) -> anyhow::Result { + let (recovery_metadata_bytes, footer_offsets) = BundleStorage::fetch_file_from_split( + storage, + PathBuf::from(SPLIT_RECOVERY_METADATA_FILE_NAME), + &split_object.path, + split_object.size.as_u64(), + ) + .await + .context("failed to fetch split recovery metadata")?; + + let recovery_metadata = SplitRecoveryMetadata::deserialize(&recovery_metadata_bytes) + .context("failed to deserialize split recovery metadata")?; + + let (split_metadata, parent_split_ids) = + SplitMetadata::try_from_recovery_metadata(recovery_metadata, footer_offsets) + .context("failed to rebuild split metadata")?; + + Ok(RecoveredSplit { + split_metadata, + parent_split_ids, + }) } /// Clears the cache directory of a given source. @@ -770,4 +1076,111 @@ mod tests { assert!(splits.is_empty()); assert!(!storage.exists(split_path).await.unwrap()); } + + #[tokio::test] + async fn test_list_metastore_split_ids_ignores_staged_splits() { + let metastore = metastore_for_test(); + let storage_resolver = StorageResolver::for_test(); + let mut index_service = IndexService::new(metastore.clone(), storage_resolver); + let index_config = IndexConfig::for_test("test-index", "ram://indexes/test-index"); + let index_uid = index_service + .create_index(index_config, false) + .await + .unwrap() + .index_uid; + + let staged_split = SplitMetadata::for_test("staged-split".into()); + let published_split = SplitMetadata::for_test("published-split".into()); + let stage_splits_request = StageSplitsRequest::try_from_splits_metadata( + index_uid.clone(), + [staged_split, published_split], + ) + .unwrap(); + metastore.stage_splits(stage_splits_request).await.unwrap(); + let publish_splits_request = PublishSplitsRequest { + index_uid: Some(index_uid.clone()), + staged_split_ids: vec!["published-split".to_string()], + replaced_split_ids: Vec::new(), + index_checkpoint_delta_json_opt: None, + publish_token_opt: None, + }; + metastore + .publish_splits(publish_splits_request) + .await + .unwrap(); + + let known_split_ids = index_service + .list_metastore_split_ids(index_uid) + .await + .unwrap(); + let expected_split_ids: HashSet = ["published-split".into()].into_iter().collect(); + assert_eq!(known_split_ids, expected_split_ids); + } + + fn recovered_split(split_id: &str, parent_split_ids: &[&str]) -> RecoveredSplit { + let mut split_metadata = SplitMetadata::for_test(split_id.into()); + split_metadata.index_uid = IndexUid::for_test("stale-index", 1); + split_metadata.num_docs = 10; + RecoveredSplit { + split_metadata, + parent_split_ids: parent_split_ids.iter().map(|id| (*id).into()).collect(), + } + } + + #[test] + fn test_plan_split_recovery_restamps_index_uid() { + let index_uid = IndexUid::for_test("recovered-index", 2); + let plan = plan_split_recovery( + &index_uid, + vec![recovered_split("split-1", &[])], + &HashSet::new(), + ); + assert_eq!(plan.splits_to_publish.len(), 1); + assert_eq!(plan.splits_to_publish[0].index_uid, index_uid); + } + + #[test] + fn test_plan_split_recovery_skips_splits_superseded_by_a_merge_chain() { + // `split-3` was merged out of `split-2`, itself merged out of `split-1`: all three files + // can still be in the storage, but only the last one may be published. + let index_uid = IndexUid::for_test("recovered-index", 2); + let plan = plan_split_recovery( + &index_uid, + vec![ + recovered_split("split-1", &[]), + recovered_split("split-2", &["split-1"]), + recovered_split("split-3", &["split-2"]), + ], + &HashSet::new(), + ); + let published_split_ids: Vec<&str> = plan + .splits_to_publish + .iter() + .map(|split_metadata| split_metadata.split_id().as_str()) + .collect(); + assert_eq!(published_split_ids, ["split-3"]); + assert_eq!(plan.merged_split_ids.len(), 2); + assert!(plan.already_known_split_ids.is_empty()); + } + + #[test] + fn test_plan_split_recovery_skips_splits_already_in_the_metastore() { + let index_uid = IndexUid::for_test("recovered-index", 2); + let known_split_ids: HashSet = ["split-1".into()].into_iter().collect(); + let plan = plan_split_recovery( + &index_uid, + vec![ + recovered_split("split-1", &[]), + recovered_split("split-2", &[]), + ], + &known_split_ids, + ); + let published_split_ids: Vec<&str> = plan + .splits_to_publish + .iter() + .map(|split_metadata| split_metadata.split_id().as_str()) + .collect(); + assert_eq!(published_split_ids, ["split-2"]); + assert_eq!(plan.already_known_split_ids.len(), 1); + } } diff --git a/quickwit/quickwit-index-management/src/lib.rs b/quickwit/quickwit-index-management/src/lib.rs index b8ab6ea2850..55db30e00db 100644 --- a/quickwit/quickwit-index-management/src/lib.rs +++ b/quickwit/quickwit-index-management/src/lib.rs @@ -16,4 +16,7 @@ mod garbage_collection; mod index; pub use garbage_collection::{GcMetrics, SplitRemovalInfo, run_garbage_collect}; -pub use index::{IndexService, IndexServiceError, clear_cache_directory, validate_storage_uri}; +pub use index::{ + IndexService, IndexServiceError, SplitRecoveryInfo, UnreadableSplit, clear_cache_directory, + validate_storage_uri, +}; From c370affa845f281a97ff596977dd83354033a2c6 Mon Sep 17 00:00:00 2001 From: tprevot Date: Sat, 17 Oct 2026 21:35:06 +0200 Subject: [PATCH 2/4] fix: subcommand description --- quickwit/quickwit-cli/src/tool.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/quickwit/quickwit-cli/src/tool.rs b/quickwit/quickwit-cli/src/tool.rs index 5e6625ccff3..cd4617de9dd 100644 --- a/quickwit/quickwit-cli/src/tool.rs +++ b/quickwit/quickwit-cli/src/tool.rs @@ -171,7 +171,7 @@ pub fn build_tool_command() -> Command { ]) ) .subcommand(Command::new("recover").display_order(11) - .about("Recovers an index from the metadata embedded in its splits. Only splits written with a footer trailer (QW_ENABLE_SPLIT_FOOTER_TRAILER=true) are recovered. Requires an S3-compatible index storage.").args(&[ + .about("Recovers an index from the metadata embedded in its splits. Splits written by a version of Quickwit that does not embed this metadata are skipped. Requires an existing index and an S3-compatible index storage.").args(&[ arg!(--index "ID of the target index") .display_order(1) .required(true), From bc46f586aa534e1b45d1713fd469d7d607ca5d6c Mon Sep 17 00:00:00 2001 From: tprevot Date: Sat, 17 Oct 2026 21:44:18 +0200 Subject: [PATCH 3/4] fix: de-duplicate storage loading --- quickwit/quickwit-cli/src/tool.rs | 18 ++--------- .../quickwit-index-management/src/index.rs | 30 ++++++------------- 2 files changed, 11 insertions(+), 37 deletions(-) diff --git a/quickwit/quickwit-cli/src/tool.rs b/quickwit/quickwit-cli/src/tool.rs index cd4617de9dd..6549decff58 100644 --- a/quickwit/quickwit-cli/src/tool.rs +++ b/quickwit/quickwit-cli/src/tool.rs @@ -733,23 +733,9 @@ pub async fn recover_index_cli(args: RecoverIndexArgs) -> anyhow::Result<()> { let metastore = metastore_resolver.resolve(&config.metastore_uri).await?; let mut index_service = IndexService::new(metastore.clone(), storage_resolver); - let index_metadata_request = IndexMetadataRequest::for_index_id(args.index_id.to_string()); - let index_metadata = metastore - .index_metadata(index_metadata_request) - .await? - .deserialize_index_metadata()?; - - let storage = StorageResolver::configured(&config.storage_configs) - .resolve(&index_metadata.index_config().index_uri) - .await?; - let index_uri = index_metadata.index_config().index_uri.clone(); - println!( - "Recovering index {} from {}\n", - args.index_id.bold(), - index_uri.to_string().bold() - ); + println!("Recovering index {}\n", args.index_id.bold()); let recovery_info = index_service - .recover_index_from_storage(&args.index_id, storage, args.dry_run) + .recover_index_from_storage(&args.index_id, args.dry_run) .await?; print_split_recovery_info(&args.index_id, &recovery_info); diff --git a/quickwit/quickwit-index-management/src/index.rs b/quickwit/quickwit-index-management/src/index.rs index 3d281f05d35..5d0ff2d1b29 100644 --- a/quickwit/quickwit-index-management/src/index.rs +++ b/quickwit/quickwit-index-management/src/index.rs @@ -575,18 +575,22 @@ impl IndexService { pub async fn recover_index_from_storage( &mut self, index_id: &str, - storage: Arc, dry_run: bool, ) -> Result { let index_metadata_request = IndexMetadataRequest::for_index_id(index_id.to_string()); - let index_uid = self + let index_metadata = self .metastore .index_metadata(index_metadata_request) .await? - .deserialize_index_metadata()? - .index_uid; + .deserialize_index_metadata()?; + + let storage = self + .storage_resolver + .resolve(&index_metadata.index_config().index_uri) + .await?; let read_outcome = read_splits_recovery_metadata(&storage).await?; + let index_uid = index_metadata.index_uid; if !read_outcome.unreadable_splits.is_empty() { warn!( @@ -635,11 +639,6 @@ impl IndexService { } /// Lists the split ids the metastore already holds in a final state. - /// - /// Staged splits are left out on purpose: they are leftovers of a run that failed between - /// staging and publishing, and the next run must stage them again (which the metastore allows) - /// and publish them. Counting them as known would skip them forever and let the janitor delete - /// their files once the staged grace period elapses. async fn list_metastore_split_ids( &mut self, index_uid: IndexUid, @@ -658,10 +657,6 @@ impl IndexService { } /// Stages then publishes a batch of recovered splits. - /// - /// `replaced_split_ids` is left empty on purpose: the splits these ones superseded were never - /// published in the rebuilt metastore, so there is nothing to mark for deletion. They are kept - /// out of the publication plan instead. async fn publish_recovered_splits( &mut self, index_uid: &IndexUid, @@ -690,7 +685,6 @@ impl IndexService { } } -/// Splits to publish, and the counts of those left out. struct SplitRecoveryPlan { splits_to_publish: Vec, merged_split_ids: Vec, @@ -699,7 +693,7 @@ struct SplitRecoveryPlan { /// Decides which recovered splits should be published into `index_uid`. /// -/// Ignores split ids already stored in the metastore (see `known_split_ids`) and splits that got +/// Ignores split ids already stored in the metastore and splits that got /// merged together and that still remain (probably with a pending deletion before the disaster). fn plan_split_recovery( index_uid: &IndexUid, @@ -744,12 +738,9 @@ struct RecoveredSplit { parent_split_ids: Vec, } -/// A split file whose recovery metadata could not be read. #[derive(Debug)] pub struct UnreadableSplit { - /// Path of the split file, relative to the index URI. pub split_file: PathBuf, - /// Why the recovery metadata could not be read. pub error: anyhow::Error, } @@ -773,8 +764,6 @@ pub struct SplitRecoveryInfo { } impl SplitRecoveryInfo { - /// Number of split files found in the storage, whether their recovery metadata could be read - /// or not. pub fn num_split_files(&self) -> usize { self.published_splits.len() + self.merged_split_ids.len() @@ -782,7 +771,6 @@ impl SplitRecoveryInfo { + self.unreadable_splits.len() } - /// Number of documents held by the published splits. pub fn num_docs_published(&self) -> u64 { self.published_splits .iter() From f14996cc5797a7b7e6023f8dcaa04f3b6830f139 Mon Sep 17 00:00:00 2001 From: tprevot Date: Mon, 19 Oct 2026 11:07:03 +0200 Subject: [PATCH 4/4] fix: rebase delete opstamp + retire known parent + reject index_uri != --- docs/operating/metastore-recovery.md | 14 +- quickwit/quickwit-cli/src/tool.rs | 38 +- .../quickwit-index-management/src/index.rs | 326 +++++++++++++++--- quickwit/quickwit-index-management/src/lib.rs | 4 +- 4 files changed, 329 insertions(+), 53 deletions(-) diff --git a/docs/operating/metastore-recovery.md b/docs/operating/metastore-recovery.md index 51c7a3aaed7..607faa9d05f 100644 --- a/docs/operating/metastore-recovery.md +++ b/docs/operating/metastore-recovery.md @@ -49,6 +49,7 @@ Recovering index otel-logs-v0_9 from s3://my-bucket/indexes/otel-logs-v0_9 | Documents | 1,099 | | Skipped (merged into another split) | 0 | | Skipped (already in metastore) | 6 | +| Replaced (marked for deletion) | 2 | | Unreadable | 1 | +-------------------------------------+-------+ ``` @@ -76,13 +77,16 @@ curl "http://localhost:7280/api/v1/otel-logs-v0_9/search?query=*&max_hits=0\ 2. Fetches the recovery metadata of each split, ten splits at a time. The object size comes from the listing, and the footer and the metadata file usually land in a single `GET` per split. 3. Rebuilds the split metadata, replacing the index UID recorded in the split with the UID of the - target index. + target index. Splits belonging to another index are skipped. 4. Skips the splits that were superseded by a merge. A merged split and its parents coexist in the storage until garbage collection removes the parents. 5. Leaves out the splits the metastore already knows about, which makes a second run a no-op. Splits left staged by an interrupted run are not considered known: they are staged and published again. -6. Stages and publishes the remaining splits, 1000 per request. +6. Stages and publishes the remaining splits, 1000 per request. A recovered merged split replaces + the splits it was merged out of, transitively: if the metastore still holds any of them as + published, for instance because it was restored from a backup predating the merge, they are + marked for deletion in the same operation, so that their documents are not returned twice. Splits whose metadata cannot be read are reported individually and do not interrupt the run: a single truncated upload would otherwise make every other split unrecoverable. @@ -92,7 +96,9 @@ single truncated upload would otherwise make every other split unrecoverable. - **Source checkpoints.** The index checkpoint is left empty. A source restarted against a recovered index resumes from the position its own configuration dictates, which may re-ingest or skip data depending on the source. -- **Delete tasks.** The delete opstamp recorded in each split is restored, but the delete task - queue itself is not. Pending deletes have to be submitted again. +- **Delete tasks.** Neither the delete task queue nor the delete opstamps are restored: the + recovered splits are published with an opstamp of `0`, since there is no way to tell which + delete tasks were applied to them. Pending deletes have to be submitted again, and already + applied ones are not replayed. - **Splits without embedded recovery metadata.** They are listed as unreadable and left in the storage, untouched. diff --git a/quickwit/quickwit-cli/src/tool.rs b/quickwit/quickwit-cli/src/tool.rs index 6549decff58..5275948c1f6 100644 --- a/quickwit/quickwit-cli/src/tool.rs +++ b/quickwit/quickwit-cli/src/tool.rs @@ -37,7 +37,9 @@ use quickwit_config::{ CLI_SOURCE_ID, IndexerConfig, NodeConfig, SourceConfig, SourceInputFormat, SourceParams, TransformConfig, VecSourceParams, }; -use quickwit_index_management::{IndexService, SplitRecoveryInfo, clear_cache_directory}; +use quickwit_index_management::{ + IndexService, SplitRecoveryInfo, SplitToPublish, clear_cache_directory, +}; use quickwit_indexing::actors::{IndexingService, MergePipeline, MergeSchedulerService}; use quickwit_indexing::docs_clustering::Fingerprinter; use quickwit_indexing::models::{ @@ -55,7 +57,7 @@ use quickwit_search::{SearchResponseRest, single_node_search}; use quickwit_serve::{ BodyFormat, SearchRequestQueryString, SortBy, search_request_from_api_request, }; -use quickwit_storage::{BundleStorage, Storage, StorageResolver}; +use quickwit_storage::{BundleStorage, Storage}; use quickwit_transport::ChannelFactory; use tabled::Tabled; use thousands::Separable; @@ -758,10 +760,13 @@ struct RecoveredSplitRow { num_merge_ops: usize, #[tabled(rename = "Doc mapping UID")] doc_mapping_uid: String, + #[tabled(rename = "Replaces")] + num_replaced_splits: usize, } -impl From<&SplitMetadata> for RecoveredSplitRow { - fn from(split_metadata: &SplitMetadata) -> Self { +impl From<&SplitToPublish> for RecoveredSplitRow { + fn from(split_to_publish: &SplitToPublish) -> Self { + let split_metadata = &split_to_publish.split_metadata; // The footer ends at the very end of the split file, so its end offset is the file size. let split_size = ByteSize::b(split_metadata.footer_offsets.end); RecoveredSplitRow { @@ -772,6 +777,7 @@ impl From<&SplitMetadata> for RecoveredSplitRow { created_at: format_timestamp(split_metadata.create_timestamp), num_merge_ops: split_metadata.num_merge_ops, doc_mapping_uid: split_metadata.doc_mapping_uid.to_string(), + num_replaced_splits: split_to_publish.replaced_split_ids.len(), } } } @@ -808,6 +814,8 @@ struct SplitRecoverySummaryRow { num_merged: String, #[tabled(rename = "Skipped (already in metastore)")] num_already_known: String, + #[tabled(rename = "Replaced (marked for deletion)")] + num_replaced: String, #[tabled(rename = "Unreadable")] num_unreadable: String, } @@ -825,6 +833,7 @@ fn print_split_recovery_info(index_id: &str, recovery_info: &SplitRecoveryInfo) .already_known_split_ids .len() .separate_with_commas(), + num_replaced: recovery_info.num_replaced_splits().separate_with_commas(), num_unreadable: num_unreadable.separate_with_commas(), }; let summary_header = if recovery_info.dry_run { @@ -854,6 +863,7 @@ fn print_split_recovery_info(index_id: &str, recovery_info: &SplitRecoveryInfo) "already known to the metastore", &recovery_info.already_known_split_ids, ); + print_replaced_split_ids(recovery_info); if num_unreadable > 0 { println!( @@ -896,6 +906,26 @@ fn print_split_recovery_info(index_id: &str, recovery_info: &SplitRecoveryInfo) ); } +/// Prints the IDs of the published splits superseded by a recovered merge, sampled to keep the +/// output readable. +fn print_replaced_split_ids(recovery_info: &SplitRecoveryInfo) { + let replaced_split_ids: Vec<&SplitId> = recovery_info.replaced_split_ids().collect(); + if replaced_split_ids.is_empty() { + return; + } + let outcome = if recovery_info.dry_run { + "would be marked for deletion" + } else { + "marked for deletion" + }; + println!( + "\n{} {} split(s) {outcome}, replaced by a recovered merged split: {:?}", + "-".color(WHITE_COLOR), + replaced_split_ids.len(), + PrettySample::new(&replaced_split_ids, 10) + ); +} + /// Prints the IDs of the splits left out of the publication, sampled to keep the output readable. fn print_skipped_split_ids(reason: &str, split_ids: &[SplitId]) { if split_ids.is_empty() { diff --git a/quickwit/quickwit-index-management/src/index.rs b/quickwit/quickwit-index-management/src/index.rs index 5d0ff2d1b29..cfea461a74b 100644 --- a/quickwit/quickwit-index-management/src/index.rs +++ b/quickwit/quickwit-index-management/src/index.rs @@ -607,9 +607,13 @@ impl IndexService { "read split recovery metadata" ); - let known_split_ids: HashSet = - self.list_metastore_split_ids(index_uid.clone()).await?; - let plan = plan_split_recovery(&index_uid, read_outcome.recovered_splits, &known_split_ids); + let known_split_states: HashMap = + self.list_metastore_split_states(index_uid.clone()).await?; + let plan = plan_split_recovery( + &index_uid, + read_outcome.recovered_splits, + &known_split_states, + ); let recovery_info = SplitRecoveryInfo { dry_run, @@ -633,48 +637,59 @@ impl IndexService { index_id=%index_id, num_published_splits=recovery_info.published_splits.len(), num_published_docs=recovery_info.num_docs_published(), + num_replaced_splits=recovery_info.num_replaced_splits(), "published recovered splits" ); Ok(recovery_info) } - /// Lists the split ids the metastore already holds in a final state. - async fn list_metastore_split_ids( + /// Lists the state of each split the metastore already holds as published or marked for + /// deletion. + async fn list_metastore_split_states( &mut self, index_uid: IndexUid, - ) -> Result, IndexServiceError> { + ) -> Result, IndexServiceError> { let list_splits_query = ListSplitsQuery::for_index(index_uid) .with_split_states([SplitState::Published, SplitState::MarkedForDeletion]); let list_splits_request = ListSplitsRequest::try_from_list_splits_query(&list_splits_query)?; - let split_ids: Vec = self + let split_states: HashMap = self .metastore .list_splits(list_splits_request) .await? - .collect_split_ids() - .await?; - Ok(split_ids.into_iter().collect()) + .collect_splits() + .await? + .into_iter() + .map(|split| (split.split_metadata.split_id, split.split_state)) + .collect(); + Ok(split_states) } /// Stages then publishes a batch of recovered splits. async fn publish_recovered_splits( &mut self, index_uid: &IndexUid, - splits_metadata: &[SplitMetadata], + splits_to_publish: &[SplitToPublish], ) -> Result<(), IndexServiceError> { let stage_splits_request = StageSplitsRequest::try_from_splits_metadata( index_uid.clone(), - splits_metadata.iter().cloned(), + splits_to_publish + .iter() + .map(|split_to_publish| split_to_publish.split_metadata.clone()), )?; self.metastore.stage_splits(stage_splits_request).await?; let publish_splits_request = PublishSplitsRequest { index_uid: Some(index_uid.clone()), - staged_split_ids: splits_metadata + staged_split_ids: splits_to_publish .iter() - .map(|split_metadata| split_metadata.split_id().to_string()) + .map(|split_to_publish| split_to_publish.split_metadata.split_id().to_string()) + .collect(), + replaced_split_ids: splits_to_publish + .iter() + .flat_map(|split_to_publish| split_to_publish.replaced_split_ids.iter()) + .map(|split_id| split_id.to_string()) .collect(), - replaced_split_ids: Vec::new(), index_checkpoint_delta_json_opt: None, publish_token_opt: None, }; @@ -686,7 +701,7 @@ impl IndexService { } struct SplitRecoveryPlan { - splits_to_publish: Vec, + splits_to_publish: Vec, merged_split_ids: Vec, already_known_split_ids: Vec, } @@ -698,8 +713,17 @@ struct SplitRecoveryPlan { fn plan_split_recovery( index_uid: &IndexUid, recovered_splits: Vec, - known_split_ids: &HashSet, + known_split_states: &HashMap, ) -> SplitRecoveryPlan { + let parent_split_ids: HashMap<&SplitId, &[SplitId]> = recovered_splits + .iter() + .map(|recovered_split| { + ( + recovered_split.split_metadata.split_id(), + recovered_split.parent_split_ids.as_slice(), + ) + }) + .collect(); let merged_split_ids: HashSet<&SplitId> = recovered_splits .iter() .flat_map(|recovered_split| recovered_split.parent_split_ids.iter()) @@ -710,24 +734,79 @@ fn plan_split_recovery( merged_split_ids: Vec::new(), already_known_split_ids: Vec::new(), }; + let mut visited_ancestor_ids: HashSet = HashSet::new(); for recovered_split in &recovered_splits { let split_id = recovered_split.split_metadata.split_id(); if merged_split_ids.contains(split_id) { plan.merged_split_ids.push(split_id.clone()); continue; } - if known_split_ids.contains(split_id) { + if known_split_states.contains_key(split_id) { plan.already_known_split_ids.push(split_id.clone()); continue; } - // The index_uid might be different so we are overwriting it with the new one. + let recovered_index_uid = &recovered_split.split_metadata.index_uid; + if recovered_index_uid.index_id != index_uid.index_id { + warn!( + split_id = %recovered_split.split_metadata.split_id, + found_index_id = %recovered_index_uid.index_id, + expected_index_id = %index_uid.index_id, + "skipping split belonging to another index" + ); + continue; + } + let replaced_split_ids = collect_replaced_split_ids( + split_id, + &parent_split_ids, + known_split_states, + &mut visited_ancestor_ids, + ); + // we've just checked index_id is the same, we overwriting the uid to keep + // the incarnation id uniform let mut split_metadata = recovered_split.split_metadata.clone(); split_metadata.index_uid = index_uid.clone(); - plan.splits_to_publish.push(split_metadata); + // set opstamp to 0 as we dont know the state of delete tasks + split_metadata.delete_opstamp = 0; + plan.splits_to_publish.push(SplitToPublish { + split_metadata, + replaced_split_ids, + }); } plan } +/// Collects the ancestors of `split_id` in the merge graph that the metastore still holds as +/// published: publishing `split_id` must replace them, otherwise their documents would show up +/// twice. +fn collect_replaced_split_ids( + split_id: &SplitId, + parent_split_ids: &HashMap<&SplitId, &[SplitId]>, + known_split_states: &HashMap, + visited_ancestor_ids: &mut HashSet, +) -> Vec { + let mut replaced_split_ids: Vec = Vec::new(); + let mut ancestor_ids_to_visit: Vec<&SplitId> = parent_split_ids + .get(split_id) + .copied() + .unwrap_or_default() + .iter() + .collect(); + + while let Some(ancestor_id) = ancestor_ids_to_visit.pop() { + if !visited_ancestor_ids.insert(ancestor_id.clone()) { + continue; + } + if known_split_states.get(ancestor_id) == Some(&SplitState::Published) { + replaced_split_ids.push(ancestor_id.clone()); + } + // The ancestor may itself be the result of a merge: its own parents are superseded too. + if let Some(grand_parent_split_ids) = parent_split_ids.get(ancestor_id) { + ancestor_ids_to_visit.extend(grand_parent_split_ids.iter()); + } + } + replaced_split_ids +} + const READ_RECOVERY_METADATA_CONCURRENCY: usize = 10; const PUBLISH_SPLITS_BATCH_SIZE: usize = 1_000; @@ -750,11 +829,17 @@ struct SplitsRecoveryMetadata { pub unreadable_splits: Vec, } +#[derive(Debug)] +pub struct SplitToPublish { + pub split_metadata: SplitMetadata, + pub replaced_split_ids: Vec, +} + pub struct SplitRecoveryInfo { /// Whether the splits were only planned, and not actually published. pub dry_run: bool, /// Splits published into the index, or that a dry run would have published. - pub published_splits: Vec, + pub published_splits: Vec, /// Recovered splits that was merged into another split. pub merged_split_ids: Vec, /// Recovered splits the metastore already knew about. @@ -774,9 +859,21 @@ impl SplitRecoveryInfo { pub fn num_docs_published(&self) -> u64 { self.published_splits .iter() - .map(|split_metadata| split_metadata.num_docs as u64) + .map(|split_to_publish| split_to_publish.split_metadata.num_docs as u64) .sum() } + + /// Returns the ids of the published splits replaced by the recovered splits, and hence marked + /// for deletion. + pub fn replaced_split_ids(&self) -> impl Iterator { + self.published_splits + .iter() + .flat_map(|split_to_publish| split_to_publish.replaced_split_ids.iter()) + } + + pub fn num_replaced_splits(&self) -> usize { + self.replaced_split_ids().count() + } } async fn read_splits_recovery_metadata( @@ -1066,7 +1163,7 @@ mod tests { } #[tokio::test] - async fn test_list_metastore_split_ids_ignores_staged_splits() { + async fn test_list_metastore_split_states_ignores_staged_splits() { let metastore = metastore_for_test(); let storage_resolver = StorageResolver::for_test(); let mut index_service = IndexService::new(metastore.clone(), storage_resolver); @@ -1097,17 +1194,74 @@ mod tests { .await .unwrap(); - let known_split_ids = index_service - .list_metastore_split_ids(index_uid) + let known_split_states = index_service + .list_metastore_split_states(index_uid) .await .unwrap(); - let expected_split_ids: HashSet = ["published-split".into()].into_iter().collect(); - assert_eq!(known_split_ids, expected_split_ids); + let expected_split_states: HashMap = + [("published-split".into(), SplitState::Published)] + .into_iter() + .collect(); + assert_eq!(known_split_states, expected_split_states); } + #[tokio::test] + async fn test_publish_recovered_splits_marks_replaced_splits_for_deletion() { + let metastore = metastore_for_test(); + let storage_resolver = StorageResolver::for_test(); + let mut index_service = IndexService::new(metastore.clone(), storage_resolver); + let index_config = IndexConfig::for_test("test-index", "ram://indexes/test-index"); + let index_uid = index_service + .create_index(index_config, false) + .await + .unwrap() + .index_uid; + + let stage_splits_request = StageSplitsRequest::try_from_splits_metadata( + index_uid.clone(), + [SplitMetadata::for_test("merged-split".into())], + ) + .unwrap(); + metastore.stage_splits(stage_splits_request).await.unwrap(); + let publish_splits_request = PublishSplitsRequest { + index_uid: Some(index_uid.clone()), + staged_split_ids: vec!["merged-split".to_string()], + replaced_split_ids: Vec::new(), + index_checkpoint_delta_json_opt: None, + publish_token_opt: None, + }; + metastore + .publish_splits(publish_splits_request) + .await + .unwrap(); + + let split_to_publish = SplitToPublish { + split_metadata: SplitMetadata::for_test("merge-result-split".into()), + replaced_split_ids: vec!["merged-split".into()], + }; + index_service + .publish_recovered_splits(&index_uid, &[split_to_publish]) + .await + .unwrap(); + + let split_states = index_service + .list_metastore_split_states(index_uid) + .await + .unwrap(); + let expected_split_states: HashMap = [ + ("merge-result-split".into(), SplitState::Published), + ("merged-split".into(), SplitState::MarkedForDeletion), + ] + .into_iter() + .collect(); + assert_eq!(split_states, expected_split_states); + } + + /// Builds a split recovered from the storage of `recovered-index`, stamped with the index + /// incarnation that preceded the one the metastore holds today. fn recovered_split(split_id: &str, parent_split_ids: &[&str]) -> RecoveredSplit { let mut split_metadata = SplitMetadata::for_test(split_id.into()); - split_metadata.index_uid = IndexUid::for_test("stale-index", 1); + split_metadata.index_uid = IndexUid::for_test("recovered-index", 1); split_metadata.num_docs = 10; RecoveredSplit { split_metadata, @@ -1115,16 +1269,44 @@ mod tests { } } + fn known_split_states( + split_states: impl IntoIterator, + ) -> HashMap { + split_states + .into_iter() + .map(|(split_id, split_state)| (split_id.into(), split_state)) + .collect() + } + + fn publishable_split_ids(plan: &SplitRecoveryPlan) -> Vec<&str> { + plan.splits_to_publish + .iter() + .map(|split_to_publish| split_to_publish.split_metadata.split_id().as_str()) + .collect() + } + #[test] fn test_plan_split_recovery_restamps_index_uid() { let index_uid = IndexUid::for_test("recovered-index", 2); let plan = plan_split_recovery( &index_uid, vec![recovered_split("split-1", &[])], - &HashSet::new(), + &HashMap::new(), ); assert_eq!(plan.splits_to_publish.len(), 1); - assert_eq!(plan.splits_to_publish[0].index_uid, index_uid); + assert_eq!( + plan.splits_to_publish[0].split_metadata.index_uid, + index_uid + ); + } + + #[test] + fn test_plan_split_recovery_skips_splits_of_another_index() { + let index_uid = IndexUid::for_test("recovered-index", 2); + let mut foreign_split = recovered_split("split-1", &[]); + foreign_split.split_metadata.index_uid = IndexUid::for_test("another-index", 1); + let plan = plan_split_recovery(&index_uid, vec![foreign_split], &HashMap::new()); + assert!(plan.splits_to_publish.is_empty()); } #[test] @@ -1139,36 +1321,94 @@ mod tests { recovered_split("split-2", &["split-1"]), recovered_split("split-3", &["split-2"]), ], - &HashSet::new(), + &HashMap::new(), ); - let published_split_ids: Vec<&str> = plan - .splits_to_publish - .iter() - .map(|split_metadata| split_metadata.split_id().as_str()) - .collect(); - assert_eq!(published_split_ids, ["split-3"]); + assert_eq!(publishable_split_ids(&plan), ["split-3"]); assert_eq!(plan.merged_split_ids.len(), 2); assert!(plan.already_known_split_ids.is_empty()); + assert!(plan.splits_to_publish[0].replaced_split_ids.is_empty()); } #[test] fn test_plan_split_recovery_skips_splits_already_in_the_metastore() { let index_uid = IndexUid::for_test("recovered-index", 2); - let known_split_ids: HashSet = ["split-1".into()].into_iter().collect(); + let known_split_states = known_split_states([("split-1", SplitState::Published)]); let plan = plan_split_recovery( &index_uid, vec![ recovered_split("split-1", &[]), recovered_split("split-2", &[]), ], - &known_split_ids, + &known_split_states, ); - let published_split_ids: Vec<&str> = plan + assert_eq!(publishable_split_ids(&plan), ["split-2"]); + assert_eq!(plan.already_known_split_ids.len(), 1); + } + + #[test] + fn test_plan_split_recovery_replaces_published_ancestors() { + // The metastore was restored from a backup taken before the merges: publishing + // `split-4` must retire the splits it was merged out of, transitively. + let index_uid = IndexUid::for_test("recovered-index", 2); + let known_split_states = known_split_states([ + ("split-1", SplitState::Published), + ("split-2", SplitState::Published), + ("split-3", SplitState::Published), + ]); + let plan = plan_split_recovery( + &index_uid, + vec![ + recovered_split("split-3", &["split-1", "split-2"]), + recovered_split("split-4", &["split-3"]), + ], + &known_split_states, + ); + assert_eq!(publishable_split_ids(&plan), ["split-4"]); + let mut replaced_split_ids: Vec<&str> = plan.splits_to_publish[0] + .replaced_split_ids + .iter() + .map(SplitId::as_str) + .collect(); + replaced_split_ids.sort(); + assert_eq!(replaced_split_ids, ["split-1", "split-2", "split-3"]); + } + + #[test] + fn test_plan_split_recovery_does_not_replace_unpublished_ancestors() { + // Replacing a split that the metastore does not hold as published is rejected, and an + // ancestor already marked for deletion is on its way out anyway. + let index_uid = IndexUid::for_test("recovered-index", 2); + let known_split_states = known_split_states([("split-1", SplitState::MarkedForDeletion)]); + let plan = plan_split_recovery( + &index_uid, + vec![recovered_split("split-3", &["split-1", "split-2"])], + &known_split_states, + ); + assert_eq!(publishable_split_ids(&plan), ["split-3"]); + assert!(plan.splits_to_publish[0].replaced_split_ids.is_empty()); + } + + #[test] + fn test_plan_split_recovery_replaces_an_ancestor_at_most_once() { + // `split-1` was merged twice: the first merge result was never published, so both + // results sit in the storage. The metastore only accepts it as replaced once. + let index_uid = IndexUid::for_test("recovered-index", 2); + let known_split_states = known_split_states([("split-1", SplitState::Published)]); + let plan = plan_split_recovery( + &index_uid, + vec![ + recovered_split("split-2", &["split-1"]), + recovered_split("split-3", &["split-1"]), + ], + &known_split_states, + ); + assert_eq!(publishable_split_ids(&plan), ["split-2", "split-3"]); + let replaced_split_ids: Vec<&str> = plan .splits_to_publish .iter() - .map(|split_metadata| split_metadata.split_id().as_str()) + .flat_map(|split_to_publish| split_to_publish.replaced_split_ids.iter()) + .map(SplitId::as_str) .collect(); - assert_eq!(published_split_ids, ["split-2"]); - assert_eq!(plan.already_known_split_ids.len(), 1); + assert_eq!(replaced_split_ids, ["split-1"]); } } diff --git a/quickwit/quickwit-index-management/src/lib.rs b/quickwit/quickwit-index-management/src/lib.rs index 55db30e00db..a95c7b1071b 100644 --- a/quickwit/quickwit-index-management/src/lib.rs +++ b/quickwit/quickwit-index-management/src/lib.rs @@ -17,6 +17,6 @@ mod index; pub use garbage_collection::{GcMetrics, SplitRemovalInfo, run_garbage_collect}; pub use index::{ - IndexService, IndexServiceError, SplitRecoveryInfo, UnreadableSplit, clear_cache_directory, - validate_storage_uri, + IndexService, IndexServiceError, SplitRecoveryInfo, SplitToPublish, UnreadableSplit, + clear_cache_directory, validate_storage_uri, };