Allow for a same-thread doc compressor.

fulmicoton · fulmicoton · commit dbd1ddd419b6 · 2022-09-08T10:28:04.000+09:00
In addition, it isolates the doc compressor logic,
better reports io::Result.

In the case of the same-thread doc compressor,
the blocks are also not copied.
diff --git a/src/core/index_meta.rs b/src/core/index_meta.rs
@@ -235,6 +235,14 @@ impl InnerSegmentMeta {
     }
 }
 
+fn return_true() -> bool {
+    true
+}
+
+fn is_true(val: &bool) -> bool {
+    *val
+}
+
 /// Search Index Settings.
 ///
 /// Contains settings which are applied on the whole
@@ -248,6 +256,12 @@ pub struct IndexSettings {
     /// The `Compressor` used to compress the doc store.
     #[serde(default)]
     pub docstore_compression: Compressor,
+    /// If set to true, docstore compression will happen on a dedicated thread.
+    /// (defaults: true)
+    #[doc(hidden)]
+    #[serde(default = "return_true")]
+    #[serde(skip_serializing_if = "is_true")]
+    pub docstore_compress_dedicated_thread: bool,
     #[serde(default = "default_docstore_blocksize")]
     /// The size of each block that will be compressed and written to disk
     pub docstore_blocksize: usize,
@@ -264,6 +278,7 @@ impl Default for IndexSettings {
             sort_by_field: None,
             docstore_compression: Compressor::default(),
             docstore_blocksize: default_docstore_blocksize(),
+            docstore_compress_dedicated_thread: true,
         }
     }
 }
@@ -395,7 +410,7 @@ mod tests {
     use super::IndexMeta;
     use crate::core::index_meta::UntrackedIndexMeta;
     use crate::schema::{Schema, TEXT};
-    use crate::store::ZstdCompressor;
+    use crate::store::{ZstdCompressor, Compressor};
     use crate::{IndexSettings, IndexSortByField, Order};
 
     #[test]
@@ -447,6 +462,7 @@ mod tests {
                     compression_level: Some(4),
                 }),
                 docstore_blocksize: 1_000_000,
+                docstore_compress_dedicated_thread: true,
             },
             segments: Vec::new(),
             schema,
@@ -485,4 +501,39 @@ mod tests {
             "unknown zstd option \"bla\" at line 1 column 103".to_string()
         );
     }
+
+    #[test]
+    #[cfg(feature="lz4-compression")]
+    fn test_index_settings_default() {
+        let mut index_settings = IndexSettings::default();
+        assert_eq!(
+            index_settings,
+            IndexSettings {
+                sort_by_field: None,
+                docstore_compression: Compressor::default(),
+                docstore_compress_dedicated_thread: true,
+                docstore_blocksize: 16_384
+            }
+        );
+        {
+            let index_settings_json = serde_json::to_value(&index_settings).unwrap();
+            assert_eq!(index_settings_json, serde_json::json!({
+                "docstore_compression": "lz4",
+                "docstore_blocksize": 16384
+            }));
+            let index_settings_deser: IndexSettings = serde_json::from_value(index_settings_json).unwrap();
+            assert_eq!(index_settings_deser, index_settings);
+        }
+        {
+            index_settings.docstore_compress_dedicated_thread = false;
+            let index_settings_json = serde_json::to_value(&index_settings).unwrap();
+            assert_eq!(index_settings_json, serde_json::json!({
+                "docstore_compression": "lz4",
+                "docstore_blocksize": 16384,
+                "docstore_compress_dedicated_thread": false,
+            }));
+            let index_settings_deser: IndexSettings = serde_json::from_value(index_settings_json).unwrap();
+            assert_eq!(index_settings_deser, index_settings);
+        }
+   }
 }
diff --git a/src/indexer/segment_serializer.rs b/src/indexer/segment_serializer.rs
@@ -38,11 +38,16 @@ impl SegmentSerializer {
         let fieldnorms_serializer = FieldNormsSerializer::from_write(fieldnorms_write)?;
 
         let postings_serializer = InvertedIndexSerializer::open(&mut segment)?;
-        let compressor = segment.index().settings().docstore_compression;
-        let blocksize = segment.index().settings().docstore_blocksize;
+        let settings = segment.index().settings();
+        let store_writer = StoreWriter::new(
+                store_write,
+                settings.docstore_compression,
+                settings.docstore_blocksize,
+                settings.docstore_compress_dedicated_thread,
+        )?;
         Ok(SegmentSerializer {
             segment,
-            store_writer: StoreWriter::new(store_write, compressor, blocksize)?,
+            store_writer,
             fast_field_serializer,
             fieldnorms_serializer: Some(fieldnorms_serializer),
             postings_serializer,
diff --git a/src/indexer/segment_writer.rs b/src/indexer/segment_writer.rs
@@ -380,12 +380,14 @@ fn remap_and_write(
         let store_write = serializer
             .segment_mut()
             .open_write(SegmentComponent::Store)?;
-        let compressor = serializer.segment().index().settings().docstore_compression;
-        let block_size = serializer.segment().index().settings().docstore_blocksize;
-        let old_store_writer = std::mem::replace(
-            &mut serializer.store_writer,
-            StoreWriter::new(store_write, compressor, block_size)?,
-        );
+        let settings = serializer.segment().index().settings();
+        let store_writer = StoreWriter::new(
+            store_write,
+            settings.docstore_compression,
+            settings.docstore_blocksize,
+            settings.docstore_compress_dedicated_thread,
+        )?;
+        let old_store_writer = std::mem::replace(&mut serializer.store_writer, store_writer);
         old_store_writer.close()?;
         let store_read = StoreReader::open(
             serializer
diff --git a/src/store/mod.rs b/src/store/mod.rs
@@ -43,6 +43,7 @@ pub use self::decompressors::Decompressor;
 pub(crate) use self::reader::DOCSTORE_CACHE_CAPACITY;
 pub use self::reader::{CacheStats, StoreReader};
 pub use self::writer::StoreWriter;
+mod store_compressor;
 
 #[cfg(feature = "lz4-compression")]
 mod compression_lz4_block;
@@ -82,14 +83,15 @@ pub mod tests {
         num_docs: usize,
         compressor: Compressor,
         blocksize: usize,
+        separate_thread: bool
     ) -> Schema {
         let mut schema_builder = Schema::builder();
         let field_body = schema_builder.add_text_field("body", TextOptions::default().set_stored());
         let field_title =
             schema_builder.add_text_field("title", TextOptions::default().set_stored());
         let schema = schema_builder.build();
         {
-            let mut store_writer = StoreWriter::new(writer, compressor, blocksize).unwrap();
+            let mut store_writer = StoreWriter::new(writer, compressor, blocksize, separate_thread).unwrap();
             for i in 0..num_docs {
                 let mut doc = Document::default();
                 doc.add_field_value(field_body, LOREM.to_string());
@@ -112,7 +114,7 @@ pub mod tests {
         let path = Path::new("store");
         let directory = RamDirectory::create();
         let store_wrt = directory.open_write(path)?;
-        let schema = write_lorem_ipsum_store(store_wrt, NUM_DOCS, Compressor::Lz4, BLOCK_SIZE);
+        let schema = write_lorem_ipsum_store(store_wrt, NUM_DOCS, Compressor::Lz4, BLOCK_SIZE, true);
         let field_title = schema.get_field("title").unwrap();
         let store_file = directory.open_read(path)?;
         let store = StoreReader::open(store_file, 10)?;
@@ -148,11 +150,11 @@ pub mod tests {
         Ok(())
     }
 
-    fn test_store(compressor: Compressor, blocksize: usize) -> crate::Result<()> {
+    fn test_store(compressor: Compressor, blocksize: usize, separate_thread: bool) -> crate::Result<()> {
         let path = Path::new("store");
         let directory = RamDirectory::create();
         let store_wrt = directory.open_write(path)?;
-        let schema = write_lorem_ipsum_store(store_wrt, NUM_DOCS, compressor, blocksize);
+        let schema = write_lorem_ipsum_store(store_wrt, NUM_DOCS, compressor, blocksize, separate_thread);
         let field_title = schema.get_field("title").unwrap();
         let store_file = directory.open_read(path)?;
         let store = StoreReader::open(store_file, 10)?;
@@ -177,13 +179,19 @@ pub mod tests {
     }
 
     #[test]
-    fn test_store_noop() -> crate::Result<()> {
-        test_store(Compressor::None, BLOCK_SIZE)
+    fn test_store_no_compression_same_thread() -> crate::Result<()> {
+        test_store(Compressor::None, BLOCK_SIZE, false)
     }
+
+    #[test]
+    fn test_store_no_compression() -> crate::Result<()> {
+        test_store(Compressor::None, BLOCK_SIZE, true)
+    }
+
     #[cfg(feature = "lz4-compression")]
     #[test]
     fn test_store_lz4_block() -> crate::Result<()> {
-        test_store(Compressor::Lz4, BLOCK_SIZE)
+        test_store(Compressor::Lz4, BLOCK_SIZE, true)
     }
     #[cfg(feature = "snappy-compression")]
     #[test]
diff --git a/src/store/reader.rs b/src/store/reader.rs
@@ -393,7 +393,7 @@ mod tests {
         let directory = RamDirectory::create();
         let path = Path::new("store");
         let writer = directory.open_write(path)?;
-        let schema = write_lorem_ipsum_store(writer, 500, Compressor::default(), BLOCK_SIZE);
+        let schema = write_lorem_ipsum_store(writer, 500, Compressor::default(), BLOCK_SIZE, true);
         let title = schema.get_field("title").unwrap();
         let store_file = directory.open_read(path)?;
         let store = StoreReader::open(store_file, DOCSTORE_CACHE_CAPACITY)?;
diff --git a/src/store/store_compressor.rs b/src/store/store_compressor.rs
diff --git a/src/store/writer.rs b/src/store/writer.rs
