From 27ab20949d1e8aecc10a0bcf1f821ccccb9d81df Mon Sep 17 00:00:00 2001 From: Luca Cominardi Date: Tue, 1 Sep 2026 18:17:35 +0200 Subject: [PATCH] Avoid syncing standalone compactor scratch files The compactor publishes splits only after remote upload succeeds, so its temporary Tantivy files do not need crash durability. --- quickwit/quickwit-directories/Cargo.toml | 4 +- quickwit/quickwit-directories/src/lib.rs | 2 + .../src/unsynced_mmap_directory.rs | 177 ++++++++++++++++++ .../quickwit-indexing/src/actors/indexer.rs | 9 + .../src/actors/indexing_pipeline.rs | 7 + .../src/models/indexed_split.rs | 13 +- 6 files changed, 205 insertions(+), 7 deletions(-) create mode 100644 quickwit/quickwit-directories/src/unsynced_mmap_directory.rs diff --git a/quickwit/quickwit-directories/Cargo.toml b/quickwit/quickwit-directories/Cargo.toml index 48036c82df8..a48365f9bb7 100644 --- a/quickwit/quickwit-directories/Cargo.toml +++ b/quickwit/quickwit-directories/Cargo.toml @@ -16,12 +16,10 @@ async-trait = { workspace = true } postcard = { workspace = true } serde = { workspace = true } tantivy = { workspace = true } +tempfile = { workspace = true } time = { workspace = true } tokio = { workspace = true } tracing = { workspace = true } quickwit-common = { workspace = true } quickwit-storage = { workspace = true } - -[dev-dependencies] -tempfile = { workspace = true } diff --git a/quickwit/quickwit-directories/src/lib.rs b/quickwit/quickwit-directories/src/lib.rs index e9b1add1692..7ab1ed9948d 100644 --- a/quickwit/quickwit-directories/src/lib.rs +++ b/quickwit/quickwit-directories/src/lib.rs @@ -30,6 +30,7 @@ mod debug_proxy_directory; mod hot_directory; mod storage_directory; mod union_directory; +mod unsynced_mmap_directory; pub use self::bundle_directory::{BundleDirectory, get_hotcache_from_split, read_split_footer}; pub use self::caching_directory::CachingDirectory; @@ -37,6 +38,7 @@ pub use self::debug_proxy_directory::{DebugProxyDirectory, ReadOperation}; pub use self::hot_directory::{HotDirectory, write_hotcache}; pub use self::storage_directory::StorageDirectory; pub use self::union_directory::UnionDirectory; +pub use self::unsynced_mmap_directory::UnsyncedMmapDirectory; macro_rules! read_only_directory { () => { diff --git a/quickwit/quickwit-directories/src/unsynced_mmap_directory.rs b/quickwit/quickwit-directories/src/unsynced_mmap_directory.rs new file mode 100644 index 00000000000..fc1e2c6acc9 --- /dev/null +++ b/quickwit/quickwit-directories/src/unsynced_mmap_directory.rs @@ -0,0 +1,177 @@ +// Copyright 2021-Present Datadog, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::fs::{File, OpenOptions}; +use std::io::{self, BufWriter, Write}; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::{fmt, result}; + +use tantivy::Directory; +use tantivy::directory::error::{ + DeleteError, LockError, OpenDirectoryError, OpenReadError, OpenWriteError, +}; +use tantivy::directory::{ + AntiCallToken, DirectoryLock, FileHandle, Lock, MmapDirectory, TerminatingWrite, WatchCallback, + WatchHandle, WritePtr, +}; + +/// A filesystem-backed Tantivy directory that does not synchronize its writes to disk. +/// +/// Files can be read through mmap, but their contents are not guaranteed to survive a crash. This +/// directory is intended for indexes whose durable copy is uploaded to remote storage before they +/// are published. +#[derive(Clone)] +pub struct UnsyncedMmapDirectory { + root_path: PathBuf, + mmap_directory: MmapDirectory, +} + +impl UnsyncedMmapDirectory { + /// Opens an existing directory. + pub fn open(directory_path: impl AsRef) -> Result { + let directory_path = directory_path.as_ref(); + let mmap_directory = MmapDirectory::open(directory_path)?; + let root_path = directory_path.canonicalize().map_err(|io_error| { + OpenDirectoryError::wrap_io_error(io_error, directory_path.to_path_buf()) + })?; + Ok(Self { + root_path, + mmap_directory, + }) + } + + fn resolve_path(&self, relative_path: &Path) -> PathBuf { + self.root_path.join(relative_path) + } +} + +impl fmt::Debug for UnsyncedMmapDirectory { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_tuple("UnsyncedMmapDirectory") + .field(&self.root_path) + .finish() + } +} + +struct UnsyncedFileWriter(File); + +impl Write for UnsyncedFileWriter { + fn write(&mut self, buffer: &[u8]) -> io::Result { + self.0.write(buffer) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +} + +impl TerminatingWrite for UnsyncedFileWriter { + fn terminate_ref(&mut self, _token: AntiCallToken) -> io::Result<()> { + Ok(()) + } +} + +impl Directory for UnsyncedMmapDirectory { + fn get_file_handle(&self, path: &Path) -> Result, OpenReadError> { + self.mmap_directory.get_file_handle(path) + } + + fn delete(&self, path: &Path) -> result::Result<(), DeleteError> { + self.mmap_directory.delete(path) + } + + fn exists(&self, path: &Path) -> Result { + self.mmap_directory.exists(path) + } + + fn open_write(&self, path: &Path) -> Result { + let file = OpenOptions::new() + .write(true) + .create_new(true) + .open(self.resolve_path(path)) + .map_err(|io_error| { + if io_error.kind() == io::ErrorKind::AlreadyExists { + OpenWriteError::FileAlreadyExists(path.to_path_buf()) + } else { + OpenWriteError::wrap_io_error(io_error, path.to_path_buf()) + } + })?; + Ok(BufWriter::new(Box::new(UnsyncedFileWriter(file)))) + } + + fn atomic_read(&self, path: &Path) -> Result, OpenReadError> { + self.mmap_directory.atomic_read(path) + } + + fn atomic_write(&self, path: &Path, data: &[u8]) -> io::Result<()> { + let full_path = self.resolve_path(path); + let parent_path = full_path.parent().ok_or_else(|| { + io::Error::new(io::ErrorKind::InvalidInput, "path has no parent directory") + })?; + let mut temp_file = tempfile::Builder::new().tempfile_in(parent_path)?; + temp_file.write_all(data)?; + temp_file + .into_temp_path() + .persist(full_path) + .map_err(|persist_error| persist_error.error)?; + Ok(()) + } + + fn sync_directory(&self) -> io::Result<()> { + Ok(()) + } + + fn acquire_lock(&self, lock: &Lock) -> Result { + self.mmap_directory.acquire_lock(lock) + } + + fn watch(&self, watch_callback: WatchCallback) -> tantivy::Result { + self.mmap_directory.watch(watch_callback) + } +} + +#[cfg(test)] +mod tests { + use std::io::Write; + use std::path::Path; + + use tantivy::Directory; + use tantivy::directory::TerminatingWrite; + + use super::UnsyncedMmapDirectory; + + #[test] + fn test_unsynced_mmap_directory() -> anyhow::Result<()> { + let temp_directory = tempfile::tempdir()?; + let directory = UnsyncedMmapDirectory::open(temp_directory.path())?; + + let file_path = Path::new("file"); + let mut writer = directory.open_write(file_path)?; + writer.write_all(b"file contents")?; + writer.terminate()?; + assert_eq!( + directory.open_read(file_path)?.read_bytes()?.as_slice(), + b"file contents" + ); + + let atomic_path = Path::new("atomic"); + directory.atomic_write(atomic_path, b"first")?; + directory.atomic_write(atomic_path, b"second")?; + assert_eq!(directory.atomic_read(atomic_path)?, b"second"); + directory.sync_directory()?; + Ok(()) + } +} diff --git a/quickwit/quickwit-indexing/src/actors/indexer.rs b/quickwit/quickwit-indexing/src/actors/indexer.rs index 5de2eda20f0..cc654d12c6c 100644 --- a/quickwit/quickwit-indexing/src/actors/indexer.rs +++ b/quickwit/quickwit-indexing/src/actors/indexer.rs @@ -100,6 +100,7 @@ struct IndexerState { tokenizer_manager: TokenizerManager, max_num_partitions: NonZeroU32, index_settings: IndexSettings, + use_unsynced_directory: bool, cooperative_indexing_opt: Option, } @@ -139,6 +140,7 @@ impl IndexerState { index_builder, io_controls, doc_id_clusterer_opt, + self.use_unsynced_directory, )?; debug!( split_id=%indexed_split.split_id(), @@ -568,6 +570,7 @@ impl Indexer { doc_mapping_uid: doc_mapper.doc_mapping_uid(), tokenizer_manager: tokenizer_manager.tantivy_manager().clone(), index_settings, + use_unsynced_directory: false, max_num_partitions: doc_mapper.max_num_partitions(), cooperative_indexing_opt, }, @@ -577,6 +580,12 @@ impl Indexer { } } + /// Uses a filesystem directory that does not synchronize its writes to disk. + pub(crate) fn use_unsynced_directory(mut self) -> Self { + self.indexer_state.use_unsynced_directory = true; + self + } + fn memory_usage(&self) -> ByteSize { if let Some(workbench) = &self.indexing_workbench_opt { ByteSize(workbench.memory_usage.delta() as u64) diff --git a/quickwit/quickwit-indexing/src/actors/indexing_pipeline.rs b/quickwit/quickwit-indexing/src/actors/indexing_pipeline.rs index e134a23ed43..5971b39f552 100644 --- a/quickwit/quickwit-indexing/src/actors/indexing_pipeline.rs +++ b/quickwit/quickwit-indexing/src/actors/indexing_pipeline.rs @@ -371,6 +371,13 @@ impl IndexingPipeline { index_serializer_mailbox, self.params.fingerprinter_opt.clone(), ); + // Without a local merge planner, the split scratch directory is only needed until the + // split is uploaded to remote storage. This is the standalone compactor configuration. + let indexer = if self.params.merge_planner_mailbox_opt.is_none() { + indexer.use_unsynced_directory() + } else { + indexer + }; let (indexer_mailbox, indexer_handle) = ctx .spawn_actor() .set_backpressure_micros_counter(counter!(parent: BACKPRESSURE_MICROS, labels: [label_values!(ACTOR_NAME => "indexer")])) diff --git a/quickwit/quickwit-indexing/src/models/indexed_split.rs b/quickwit/quickwit-indexing/src/models/indexed_split.rs index 09e6b47f831..46baf75f442 100644 --- a/quickwit/quickwit-indexing/src/models/indexed_split.rs +++ b/quickwit/quickwit-indexing/src/models/indexed_split.rs @@ -18,12 +18,13 @@ use std::path::Path; use quickwit_common::io::IoControls; use quickwit_common::metrics::index_label; use quickwit_common::temp_dir::TempDirectory; +use quickwit_directories::UnsyncedMmapDirectory; use quickwit_metastore::checkpoint::IndexCheckpointDelta; use quickwit_metrics::{GaugeGuard, label_values}; use quickwit_proto::indexing::IndexingPipelineId; use quickwit_proto::types::{DocMappingUid, IndexUid, SplitId}; -use tantivy::IndexBuilder; use tantivy::directory::MmapDirectory; +use tantivy::{Directory, IndexBuilder}; use tracing::{Span, error, instrument}; use crate::controlled_directory::ControlledDirectory; @@ -87,6 +88,7 @@ impl IndexedSplitBuilder { index_builder: IndexBuilder, io_controls: IoControls, doc_id_clusterer_opt: Option, + use_unsynced_directory: bool, ) -> anyhow::Result { // We avoid intermediary merge, and instead merge all segments in the packager. // The benefit is that we don't have to wait for potentially existing merges, @@ -95,10 +97,13 @@ impl IndexedSplitBuilder { let split_scratch_directory_prefix = format!("split-{split_id}-"); let split_scratch_directory = scratch_directory.named_temp_child(&split_scratch_directory_prefix)?; - let mmap_directory = MmapDirectory::open(split_scratch_directory.path())?; - let box_mmap_directory = Box::new(mmap_directory); + let directory: Box = if use_unsynced_directory { + Box::new(UnsyncedMmapDirectory::open(split_scratch_directory.path())?) + } else { + Box::new(MmapDirectory::open(split_scratch_directory.path())?) + }; - let controlled_directory = ControlledDirectory::new(box_mmap_directory, io_controls); + let controlled_directory = ControlledDirectory::new(directory, io_controls); let index_writer = index_builder.single_segment_index_writer(controlled_directory.clone(), 15_000_000)?;