Release RAM-backed splits after serialization

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(())
+    }
+}

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();
+}

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;