From 1ca42ca6046adf462755849736a79e8786dd5ce1 Mon Sep 17 00:00:00 2001 From: Pascal Seitz Date: Wed, 14 Oct 2026 08:24:12 +0200 Subject: [PATCH] Release RAM-backed splits after serialization --- .../src/models/indexed_split.rs | 73 ++++++++++++++- .../src/tests/in_memory_indexing_tests.rs | 91 +++++++++++++++++++ .../src/tests/mod.rs | 1 + 3 files changed, 162 insertions(+), 3 deletions(-) create mode 100644 quickwit/quickwit-integration-tests/src/tests/in_memory_indexing_tests.rs diff --git a/quickwit/quickwit-indexing/src/models/indexed_split.rs b/quickwit/quickwit-indexing/src/models/indexed_split.rs index 3175ec90612..b79cdf26997 100644 --- a/quickwit/quickwit-indexing/src/models/indexed_split.rs +++ b/quickwit/quickwit-indexing/src/models/indexed_split.rs @@ -169,10 +169,16 @@ impl IndexedSplitBuilder { } else { self.index_writer.finalize()? }; - if let Some(ram_directory) = &self.ram_directory_opt { - // The packager and uploader consume split files from the scratch directory. + let index = if let Some(ram_directory) = self.ram_directory_opt { ram_directory.persist(&self.controlled_directory)?; - } + // Persist copies files but leaves RAM intact. Return a disk-backed index so packaging + // does not retain the RAM files. The builder's RAM handle drops at this block's end; + // the original RAM-backed `index` (shadowed by this binding) drops when + // finalize returns. + tantivy::Index::open(self.controlled_directory.clone())? + } else { + index + }; Ok(IndexedSplit { split_attrs, index, @@ -241,3 +247,64 @@ pub struct EmptySplit { pub publish_lock: PublishLock, pub batch_parent_span: Span, } + +#[cfg(test)] +mod tests { + use quickwit_proto::types::NodeId; + use tantivy::schema::{STORED, Schema}; + use tantivy::{Directory, DocAddress, TantivyDocument, doc}; + + use super::*; + + #[test] + fn test_finalize_ram_split_uses_disk() -> anyhow::Result<()> { + let mut schema = Schema::builder(); + let id = schema.add_u64_field("id", STORED); + let ram_directory = RamDirectory::default(); + let mut index_writer = IndexBuilder::new() + .schema(schema.build()) + .single_segment_index_writer(ram_directory.clone(), 15_000_000)?; + index_writer.add_document(doc!(id => 42u64))?; + let split_scratch_directory = TempDirectory::for_test(); + let controlled_directory = ControlledDirectory::new( + Box::new(MmapDirectory::open(split_scratch_directory.path())?), + IoControls::default(), + ); + // Select RAM directly: the production environment flag is cached process-wide. + let builder = IndexedSplitBuilder { + split_attrs: SplitAttrs { + node_id: NodeId::from_str("test-node"), + index_uid: IndexUid::new_with_random_ulid("test-index"), + source_id: "test-source".to_string(), + doc_mapping_uid: DocMappingUid::default(), + split_id: SplitId::new(), + partition_id: 0, + num_docs: 1, + uncompressed_docs_size_in_bytes: 0, + time_range: None, + replaced_split_ids: Vec::new(), + delete_opstamp: 0, + num_merge_ops: 0, + }, + index_writer, + split_scratch_directory, + controlled_directory, + doc_id_clusterer_opt: None, + ram_directory_opt: Some(ram_directory.clone()), + }; + let split = builder.finalize()?; + // A file added only after persistence must not be visible through the returned index. + let marker = Path::new("ram-only"); + ram_directory.atomic_write(marker, b"not persisted")?; + assert!(!split.index.directory().exists(marker)?); + drop(ram_directory); + + let reader = split.index.reader()?; + let searcher = reader.searcher(); + assert_eq!(searcher.num_docs(), 1); + let id = split.index.schema().get_field("id")?; + let document: TantivyDocument = searcher.doc(DocAddress::new(0, 0))?; + assert_eq!(document, doc!(id => 42u64)); + Ok(()) + } +} diff --git a/quickwit/quickwit-integration-tests/src/tests/in_memory_indexing_tests.rs b/quickwit/quickwit-integration-tests/src/tests/in_memory_indexing_tests.rs new file mode 100644 index 00000000000..1be3aa79638 --- /dev/null +++ b/quickwit/quickwit-integration-tests/src/tests/in_memory_indexing_tests.rs @@ -0,0 +1,91 @@ +// 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::process::Command; + +use quickwit_config::ConfigFormat; +use quickwit_config::service::QuickwitService; +use quickwit_rest_client::models::IngestSource; +use quickwit_rest_client::rest_client::CommitType; +use quickwit_serve::SearchRequestQueryString; +use serde_json::json; + +use crate::test_utils::{ClusterSandboxBuilder, ingest}; + +#[tokio::test] +async fn test_in_memory_indexing() { + // The flag is cached on first use. Set it before startup in a fresh process, rather than + // mutating the environment shared by concurrent tests (including under cargo test). + if std::env::var("QW_ENABLE_IN_MEMORY_INDEXING").as_deref() != Ok("true") { + let status = Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + "tests::in_memory_indexing_tests::test_in_memory_indexing", + "--nocapture", + ]) + .env("QW_ENABLE_IN_MEMORY_INDEXING", "true") + .status() + .unwrap(); + assert!(status.success(), "in-memory indexing test failed: {status}"); + return; + } + quickwit_common::setup_logging_for_tests(); + let sandbox = ClusterSandboxBuilder::build_and_start_standalone().await; + let client = sandbox.rest_client(QuickwitService::Indexer); + let index_id = "in-memory-index"; + client + .indexes() + .create( + r#" + version: 0.8 + index_id: in-memory-index + doc_mapping: + field_mappings: + - name: body + type: text + "#, + ConfigFormat::Yaml, + false, + ) + .await + .unwrap(); + sandbox.wait_for_indexing_pipelines(1).await.unwrap(); + ingest( + &client, + index_id, + IngestSource::Str( + "{\"body\":\"first message\"}\n{\"body\":\"second message\"}".to_string(), + ), + CommitType::Force, + ) + .await + .unwrap(); + // Search the published split, after serialization, packaging, and upload have completed. + sandbox.assert_hit_count(index_id, "*", 2).await; + let response = client + .search( + index_id, + SearchRequestQueryString { + query: "body:first".to_string(), + max_hits: 10, + ..Default::default() + }, + ) + .await + .unwrap(); + assert!(response.errors.is_empty(), "{:?}", response.errors); + assert_eq!(response.num_hits, 1); + assert_eq!(response.hits, vec![json!({"body": "first message"})]); + sandbox.shutdown().await.unwrap(); +} diff --git a/quickwit/quickwit-integration-tests/src/tests/mod.rs b/quickwit/quickwit-integration-tests/src/tests/mod.rs index 4f9d9eebdfe..7e613bd1f95 100644 --- a/quickwit/quickwit-integration-tests/src/tests/mod.rs +++ b/quickwit/quickwit-integration-tests/src/tests/mod.rs @@ -14,6 +14,7 @@ mod basic_tests; mod compactor_tests; +mod in_memory_indexing_tests; mod ingest_v1_tests; mod ingest_v2_tests; mod no_cp_tests;