From f580ce2e79bbbc0dc5f4e45c296c8625642ab4a3 Mon Sep 17 00:00:00 2001 From: Hengfei Yang Date: Fri, 16 Oct 2026 08:30:52 +0800 Subject: [PATCH] fix(tantivy): stop index builds from fsyncing their scratch files (v1.1.0) (#15420) (#15425) --- Cargo.lock | 1 + src/tantivy_utils/Cargo.toml | 1 + .../src/index_builder/parallel.rs | 9 +- src/tantivy_utils/src/lib.rs | 1 + .../src/puffin_directory/writer.rs | 8 +- src/tantivy_utils/src/scratch_directory.rs | 187 ++++++++++++++++++ 6 files changed, 200 insertions(+), 7 deletions(-) create mode 100644 src/tantivy_utils/src/scratch_directory.rs diff --git a/Cargo.lock b/Cargo.lock index 667f9ab005..0ba6cc6782 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -12740,6 +12740,7 @@ dependencies = [ "serde", "serde_json", "tantivy", + "tempfile", "tokio", "vortex", "zstd", diff --git a/src/tantivy_utils/Cargo.toml b/src/tantivy_utils/Cargo.toml index afb7ef636a..12ec8c97b3 100644 --- a/src/tantivy_utils/Cargo.toml +++ b/src/tantivy_utils/Cargo.toml @@ -27,6 +27,7 @@ parquet.workspace = true serde.workspace = true serde_json.workspace = true tantivy.workspace = true +tempfile.workspace = true tokio.workspace = true vortex = { workspace = true } zstd.workspace = true diff --git a/src/tantivy_utils/src/index_builder/parallel.rs b/src/tantivy_utils/src/index_builder/parallel.rs index ed191906d4..2e7c6ad004 100644 --- a/src/tantivy_utils/src/index_builder/parallel.rs +++ b/src/tantivy_utils/src/index_builder/parallel.rs @@ -23,7 +23,7 @@ use config::{ }; use futures::future::join_all; use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; -use tantivy::{directory::MmapDirectory, indexer::merge_indices}; +use tantivy::indexer::merge_indices; use tokio::task::JoinHandle; use vortex::{ VortexSessionDefault, @@ -36,7 +36,10 @@ use vortex::{ }; use super::{TantivyIndexSchema, convert_batch_to_docs_sync}; -use crate::index_builder::reader::{ChunkSelector, chunk_iter}; +use crate::{ + index_builder::reader::{ChunkSelector, chunk_iter}, + scratch_directory::ScratchDirectory, +}; pub(super) async fn build_index( tantivy_dir: D, @@ -173,7 +176,7 @@ where let selector = make_selector(chunk_idx); tasks.push(tokio::task::spawn_blocking(move || { let _permit = permit; - let dir = MmapDirectory::create_from_tempdir()?; + let dir = ScratchDirectory::new()?; build_segment(chunk_idx, buf_c, selector, index_schema_c, dir) })); } diff --git a/src/tantivy_utils/src/lib.rs b/src/tantivy_utils/src/lib.rs index 98409a6934..c44c556722 100644 --- a/src/tantivy_utils/src/lib.rs +++ b/src/tantivy_utils/src/lib.rs @@ -16,3 +16,4 @@ pub mod index_builder; pub mod puffin; pub mod puffin_directory; +pub mod scratch_directory; diff --git a/src/tantivy_utils/src/puffin_directory/writer.rs b/src/tantivy_utils/src/puffin_directory/writer.rs index 21241c82ec..67a33369a5 100644 --- a/src/tantivy_utils/src/puffin_directory/writer.rs +++ b/src/tantivy_utils/src/puffin_directory/writer.rs @@ -23,19 +23,20 @@ use anyhow::{Context, Result}; use hashbrown::{HashMap, HashSet}; use tantivy::{ HasLen, - directory::{Directory, MmapDirectory, WatchCallback, WatchHandle, error::OpenReadError}, + directory::{Directory, WatchCallback, WatchHandle, error::OpenReadError}, }; use super::{FOOTER_CACHE, footer_cache::build_footer_cache}; use crate::{ puffin::{BlobTypes, writer::PuffinBytesWriter}, puffin_directory::{ALLOWED_FILE_EXT, META_JSON}, + scratch_directory::ScratchDirectory, }; /// Puffin directory is a puffin file which contains all the tantivy files. /// Each tantivy file is stored as a blob in the puffin file, along with their file name. #[derive(Debug)] pub struct PuffinDirWriter { - mmap_directory: Arc, + mmap_directory: Arc, /// record all the files paths in the puffin file file_paths: Arc>>, /// file-level puffin properties (written to PuffinMeta.properties) @@ -62,8 +63,7 @@ impl PuffinDirWriter { pub fn new() -> Self { PuffinDirWriter { mmap_directory: Arc::new( - MmapDirectory::create_from_tempdir() - .expect("failed to create temporary mmap directory"), + ScratchDirectory::new().expect("failed to create temporary mmap directory"), ), file_paths: Arc::new(RwLock::new(HashSet::default())), properties: Arc::new(RwLock::new(HashMap::default())), diff --git a/src/tantivy_utils/src/scratch_directory.rs b/src/tantivy_utils/src/scratch_directory.rs new file mode 100644 index 0000000000..4aae32f446 --- /dev/null +++ b/src/tantivy_utils/src/scratch_directory.rs @@ -0,0 +1,187 @@ +// Copyright 2026 OpenObserve Inc. +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +use std::{ + fs::{File, OpenOptions}, + io::{self, BufWriter, Write}, + path::{Path, PathBuf}, + sync::Arc, +}; + +use tantivy::directory::{ + AntiCallToken, Directory, DirectoryLock, FileHandle, Lock, MmapDirectory, TerminatingWrite, + WatchCallback, WatchHandle, WritePtr, + error::{DeleteError, LockError, OpenDirectoryError, OpenReadError, OpenWriteError}, +}; +use tempfile::TempDir; + +/// Temp-dir `MmapDirectory` without fsync: its files are scratch, and fsync stalls all disk writes. +#[derive(Clone, Debug)] +pub struct ScratchDirectory { + inner: MmapDirectory, + root: Arc, +} + +impl ScratchDirectory { + pub fn new() -> Result { + let root = + TempDir::new().map_err(|e| OpenDirectoryError::FailedToCreateTempDir(Arc::new(e)))?; + let inner = MmapDirectory::open(root.path())?; + Ok(Self { + inner, + root: Arc::new(root), + }) + } + + fn resolve_path(&self, path: &Path) -> PathBuf { + self.root.path().join(path) + } +} + +impl Directory for ScratchDirectory { + fn get_file_handle(&self, path: &Path) -> Result, OpenReadError> { + self.inner.get_file_handle(path) + } + + fn delete(&self, path: &Path) -> Result<(), DeleteError> { + self.inner.delete(path) + } + + fn exists(&self, path: &Path) -> Result { + self.inner.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(|e| { + if e.kind() == io::ErrorKind::AlreadyExists { + OpenWriteError::FileAlreadyExists(path.to_path_buf()) + } else { + OpenWriteError::wrap_io_error(e, path.to_path_buf()) + } + })?; + Ok(BufWriter::new(Box::new(NoSyncFileWriter(file)))) + } + + fn atomic_read(&self, path: &Path) -> Result, OpenReadError> { + self.inner.atomic_read(path) + } + + fn atomic_write(&self, path: &Path, data: &[u8]) -> io::Result<()> { + let full_path = self.resolve_path(path); + let mut tmp_path = full_path.clone().into_os_string(); + tmp_path.push(".tmp"); + std::fs::write(&tmp_path, data)?; + std::fs::rename(&tmp_path, &full_path) + } + + fn acquire_lock(&self, lock: &Lock) -> Result { + self.inner.acquire_lock(lock) + } + + fn sync_directory(&self) -> io::Result<()> { + Ok(()) + } + + fn watch(&self, watch_callback: WatchCallback) -> tantivy::Result { + self.inner.watch(watch_callback) + } +} + +struct NoSyncFileWriter(File); + +impl Write for NoSyncFileWriter { + fn write(&mut self, buf: &[u8]) -> io::Result { + self.0.write(buf) + } + + fn flush(&mut self) -> io::Result<()> { + self.0.flush() + } +} + +impl TerminatingWrite for NoSyncFileWriter { + fn terminate_ref(&mut self, _: AntiCallToken) -> io::Result<()> { + self.0.flush() + } +} + +#[cfg(test)] +mod tests { + use tantivy::{ + Index, IndexWriter, TantivyDocument, doc, + schema::{STORED, Schema, TEXT}, + }; + + use super::*; + + #[test] + fn test_scratch_directory_round_trip() { + let dir = ScratchDirectory::new().unwrap(); + let path = Path::new("a.bin"); + let mut w = dir.open_write(path).unwrap(); + w.write_all(b"hello").unwrap(); + w.terminate().unwrap(); + assert!(dir.exists(path).unwrap()); + assert_eq!( + dir.open_read(path) + .unwrap() + .read_bytes() + .unwrap() + .as_slice(), + b"hello" + ); + assert!(matches!( + dir.open_write(path), + Err(OpenWriteError::FileAlreadyExists(_)) + )); + + let meta = Path::new("meta.json"); + dir.atomic_write(meta, b"v1").unwrap(); + dir.atomic_write(meta, b"v2").unwrap(); + assert_eq!(dir.atomic_read(meta).unwrap(), b"v2"); + } + + #[test] + fn test_scratch_directory_builds_index() { + let mut builder = Schema::builder(); + let body = builder.add_text_field("body", TEXT | STORED); + let index = Index::create( + ScratchDirectory::new().unwrap(), + builder.build(), + Default::default(), + ) + .unwrap(); + let mut writer: IndexWriter = index.writer(15_000_000).unwrap(); + writer.add_document(doc!(body => "hello world")).unwrap(); + writer.commit().unwrap(); + let searcher = index.reader().unwrap().searcher(); + assert_eq!(searcher.num_docs(), 1); + } + + #[test] + fn test_scratch_directory_removed_on_drop() { + let dir = ScratchDirectory::new().unwrap(); + let root = dir.root.path().to_path_buf(); + let clone = dir.clone(); + drop(dir); + assert!(root.exists()); + drop(clone); + assert!(!root.exists()); + } +}