[WIP] Do not fsync on disk on indexers when compactors are enabled

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 }

quickwit/quickwit-directories/src/lib.rs

@@ -30,13 +30,15 @@ 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;
 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 {
     () => {

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<Path>) -> Result<Self, OpenDirectoryError> {
+        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<usize> {
+        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<Arc<dyn FileHandle>, 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<bool, OpenReadError> {
+        self.mmap_directory.exists(path)
+    }
+
+    fn open_write(&self, path: &Path) -> Result<WritePtr, OpenWriteError> {
+        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<Vec<u8>, 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<DirectoryLock, LockError> {
+        self.mmap_directory.acquire_lock(lock)
+    }
+
+    fn watch(&self, watch_callback: WatchCallback) -> tantivy::Result<WatchHandle> {
+        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(())
+    }
+}

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<CooperativeIndexingCycle>,
 }
 
@@ -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)

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")]))

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<DocIdClusterer>,
+        use_unsynced_directory: bool,
     ) -> anyhow::Result<Self> {
         // 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<dyn Directory> = 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)?;