Moving the StoreWriter out of the SegmentSerializer.

CHANGELOG.md

@@ -1,7 +1,3 @@
-Tantivy 0.13.0
-======================
-- Bugfix in `FuzzyTermQuery` not matching terms by prefix when it should (@Peachball)
-
 Tantivy 0.12.0
 ======================
 - Removing static dispatch in tokenizers for simplicity. (#762)

Cargo.toml

@@ -13,12 +13,12 @@ keywords = ["search", "information", "retrieval"]
 edition = "2018"
 
 [dependencies]
-base64 = "0.12.0"
+base64 = "0.11.0"
 byteorder = "1.0"
 crc32fast = "1.2.0"
 once_cell = "1.0"
 regex ={version = "1.3.0", default-features = false, features = ["std"]}
-tantivy-fst = {path="../tantivy-fst", version="0.3"}
+tantivy-fst = "0.2.1"
 memmap = {version = "0.7", optional=true}
 lz4 = {version="1.20", optional=true}
 snap = "1"
@@ -29,7 +29,8 @@ serde = {version="1.0", features=["derive"]}
 serde_json = "1.0"
 num_cpus = "1.2"
 fs2={version="0.4", optional=true}
-levenshtein_automata = "0.2"
+itertools = "0.8"
+levenshtein_automata = "0.1"
 notify = {version="4", optional=true}
 uuid = { version = "0.8", features = ["v4", "serde"] }
 crossbeam = "0.7"
@@ -38,7 +39,7 @@ owning_ref = "0.4"
 stable_deref_trait = "1.0.0"
 rust-stemmers = "1.2"
 downcast-rs = { version="1.0" }
-tantivy-query-grammar = { version="0.13", path="./query-grammar" }
+tantivy-query-grammar = { version="0.12", path="./query-grammar" }
 bitpacking = {version="0.8", default-features = false, features=["bitpacker4x"]}
 census = "0.4"
 fnv = "1.0.6"

appveyor.yml

@@ -18,5 +18,5 @@ install:
 build: false
 
 test_script:
-  - REM SET RUST_LOG=tantivy,test & cargo test --all --verbose --no-default-features --features mmap
+  - REM SET RUST_LOG=tantivy,test & cargo test --verbose --no-default-features --features mmap
   - REM SET RUST_BACKTRACE=1 & cargo build --examples

query-grammar/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "tantivy-query-grammar"
-version = "0.13.0"
+version = "0.12.0"
 authors = ["Paul Masurel <paul.masurel@gmail.com>"]
 license = "MIT"
 categories = ["database-implementations", "data-structures"]

query-grammar/src/query_grammar.rs

@@ -154,11 +154,17 @@ fn negate(expr: UserInputAST) -> UserInputAST {
     expr.unary(Occur::MustNot)
 }
 
+fn must(expr: UserInputAST) -> UserInputAST {
+    expr.unary(Occur::Must)
+}
+
 fn leaf<'a>() -> impl Parser<&'a str, Output = UserInputAST> {
     parser(|input| {
-        char('(')
-            .with(ast())
-            .skip(char(')'))
+        char('-')
+            .with(leaf())
+            .map(negate)
+            .or(char('+').with(leaf()).map(must))
+            .or(char('(').with(ast()).skip(char(')')))
             .or(char('*').map(|_| UserInputAST::from(UserInputLeaf::All)))
             .or(attempt(
                 string("NOT").skip(spaces1()).with(leaf()).map(negate),
@@ -170,16 +176,6 @@ fn leaf<'a>() -> impl Parser<&'a str, Output = UserInputAST> {
     })
 }
 
-fn occur_symbol<'a>() -> impl Parser<&'a str, Output = Occur> {
-    char('-')
-        .map(|_| Occur::MustNot)
-        .or(char('+').map(|_| Occur::Must))
-}
-
-fn occur_leaf<'a>() -> impl Parser<&'a str, Output = (Option<Occur>, UserInputAST)> {
-    (optional(occur_symbol()), boosted_leaf())
-}
-
 fn positive_float_number<'a>() -> impl Parser<&'a str, Output = f32> {
     (many1(digit()), optional((char('.'), many1(digit())))).map(
         |(int_part, decimal_part_opt): (String, Option<(char, String)>)| {
@@ -243,29 +239,21 @@ fn aggregate_binary_expressions(
     }
 }
 
-fn operand_leaf<'a>() -> impl Parser<&'a str, Output = (BinaryOperand, UserInputAST)> {
-    (
+pub fn ast<'a>() -> impl Parser<&'a str, Output = UserInputAST> {
+    let operand_leaf = (
         binary_operand().skip(spaces()),
         boosted_leaf().skip(spaces()),
-    )
-}
-
-pub fn ast<'a>() -> impl Parser<&'a str, Output = UserInputAST> {
-    let boolean_expr = (boosted_leaf().skip(spaces()), many1(operand_leaf()))
+    );
+    let boolean_expr = (boosted_leaf().skip(spaces().silent()), many1(operand_leaf))
         .map(|(left, right)| aggregate_binary_expressions(left, right));
-    let whitespace_separated_leaves = many1(occur_leaf().skip(spaces().silent())).map(
-        |subqueries: Vec<(Option<Occur>, UserInputAST)>| {
+    let whitespace_separated_leaves =
+        many1(boosted_leaf().skip(spaces().silent())).map(|subqueries: Vec<UserInputAST>| {
             if subqueries.len() == 1 {
-                let (occur_opt, ast) = subqueries.into_iter().next().unwrap();
-                match occur_opt.unwrap_or(Occur::Should) {
-                    Occur::Must | Occur::Should => ast,
-                    Occur::MustNot => UserInputAST::Clause(vec![(Some(Occur::MustNot), ast)]),
-                }
+                subqueries.into_iter().next().unwrap()
             } else {
                 UserInputAST::Clause(subqueries.into_iter().collect())
             }
-        },
-    );
+        });
     let expr = attempt(boolean_expr).or(whitespace_separated_leaves);
     spaces().with(expr).skip(spaces())
 }
@@ -295,12 +283,6 @@ mod test {
         );
     }
 
-    #[test]
-    fn test_occur_symbol() {
-        assert_eq!(super::occur_symbol().parse("-"), Ok((Occur::MustNot, "")));
-        assert_eq!(super::occur_symbol().parse("+"), Ok((Occur::Must, "")));
-    }
-
     #[test]
     fn test_positive_float_number() {
         fn valid_parse(float_str: &str, expected_val: f32, expected_remaining: &str) {
@@ -348,7 +330,7 @@ mod test {
             "Err(UnexpectedParse)"
         );
         test_parse_query_to_ast_helper("NOTa", "\"NOTa\"");
-        test_parse_query_to_ast_helper("NOT a", "(-\"a\")");
+        test_parse_query_to_ast_helper("NOT a", "-(\"a\")");
     }
 
     #[test]
@@ -356,16 +338,16 @@ mod test {
         assert!(parse_to_ast().parse("a^2^3").is_err());
         assert!(parse_to_ast().parse("a^2^").is_err());
         test_parse_query_to_ast_helper("a^3", "(\"a\")^3");
-        test_parse_query_to_ast_helper("a^3 b^2", "(*(\"a\")^3 *(\"b\")^2)");
+        test_parse_query_to_ast_helper("a^3 b^2", "((\"a\")^3 (\"b\")^2)");
         test_parse_query_to_ast_helper("a^1", "\"a\"");
     }
 
     #[test]
     fn test_parse_query_to_ast_binary_op() {
-        test_parse_query_to_ast_helper("a AND b", "(+\"a\" +\"b\")");
-        test_parse_query_to_ast_helper("a OR b", "(?\"a\" ?\"b\")");
-        test_parse_query_to_ast_helper("a OR b AND c", "(?\"a\" ?(+\"b\" +\"c\"))");
-        test_parse_query_to_ast_helper("a AND b         AND c", "(+\"a\" +\"b\" +\"c\")");
+        test_parse_query_to_ast_helper("a AND b", "(+(\"a\") +(\"b\"))");
+        test_parse_query_to_ast_helper("a OR b", "(?(\"a\") ?(\"b\"))");
+        test_parse_query_to_ast_helper("a OR b AND c", "(?(\"a\") ?((+(\"b\") +(\"c\"))))");
+        test_parse_query_to_ast_helper("a AND b         AND c", "(+(\"a\") +(\"b\") +(\"c\"))");
         assert_eq!(
             format!("{:?}", parse_to_ast().parse("a OR b aaa")),
             "Err(UnexpectedParse)"
@@ -403,13 +385,6 @@ mod test {
         test_parse_query_to_ast_helper("weight: <= 70.5", "weight:{\"*\" TO \"70.5\"]");
     }
 
-    #[test]
-    fn test_occur_leaf() {
-        let ((occur, ast), _) = super::occur_leaf().parse("+abc").unwrap();
-        assert_eq!(occur, Some(Occur::Must));
-        assert_eq!(format!("{:?}", ast), "\"abc\"");
-    }
-
     #[test]
     fn test_range_parser() {
         // testing the range() parser separately
@@ -438,67 +413,32 @@ mod test {
     fn test_parse_query_to_triming_spaces() {
         test_parse_query_to_ast_helper("   abc", "\"abc\"");
         test_parse_query_to_ast_helper("abc ", "\"abc\"");
-        test_parse_query_to_ast_helper("(  a OR abc)", "(?\"a\" ?\"abc\")");
-        test_parse_query_to_ast_helper("(a  OR abc)", "(?\"a\" ?\"abc\")");
-        test_parse_query_to_ast_helper("(a OR  abc)", "(?\"a\" ?\"abc\")");
-        test_parse_query_to_ast_helper("a OR abc ", "(?\"a\" ?\"abc\")");
-        test_parse_query_to_ast_helper("(a OR abc )", "(?\"a\" ?\"abc\")");
-        test_parse_query_to_ast_helper("(a OR  abc) ", "(?\"a\" ?\"abc\")");
+        test_parse_query_to_ast_helper("(  a OR abc)", "(?(\"a\") ?(\"abc\"))");
+        test_parse_query_to_ast_helper("(a  OR abc)", "(?(\"a\") ?(\"abc\"))");
+        test_parse_query_to_ast_helper("(a OR  abc)", "(?(\"a\") ?(\"abc\"))");
+        test_parse_query_to_ast_helper("a OR abc ", "(?(\"a\") ?(\"abc\"))");
+        test_parse_query_to_ast_helper("(a OR abc )", "(?(\"a\") ?(\"abc\"))");
+        test_parse_query_to_ast_helper("(a OR  abc) ", "(?(\"a\") ?(\"abc\"))");
     }
 
     #[test]
-    fn test_parse_query_single_term() {
+    fn test_parse_query_to_ast() {
         test_parse_query_to_ast_helper("abc", "\"abc\"");
-    }
-
-    #[test]
-    fn test_parse_query_default_clause() {
-        test_parse_query_to_ast_helper("a b", "(*\"a\" *\"b\")");
-    }
-
-    #[test]
-    fn test_parse_query_must_default_clause() {
-        test_parse_query_to_ast_helper("+(a b)", "(*\"a\" *\"b\")");
-    }
-
-    #[test]
-    fn test_parse_query_must_single_term() {
-        test_parse_query_to_ast_helper("+d", "\"d\"");
-    }
-
-    #[test]
-    fn test_single_term_with_field() {
+        test_parse_query_to_ast_helper("a b", "(\"a\" \"b\")");
+        test_parse_query_to_ast_helper("+(a b)", "+((\"a\" \"b\"))");
+        test_parse_query_to_ast_helper("+d", "+(\"d\")");
+        test_parse_query_to_ast_helper("+(a b) +d", "(+((\"a\" \"b\")) +(\"d\"))");
+        test_parse_query_to_ast_helper("(+a +b) d", "((+(\"a\") +(\"b\")) \"d\")");
+        test_parse_query_to_ast_helper("(+a)", "+(\"a\")");
+        test_parse_query_to_ast_helper("(+a +b)", "(+(\"a\") +(\"b\"))");
         test_parse_query_to_ast_helper("abc:toto", "abc:\"toto\"");
-    }
-
-    #[test]
-    fn test_single_term_with_float() {
         test_parse_query_to_ast_helper("abc:1.1", "abc:\"1.1\"");
-    }
-
-    #[test]
-    fn test_must_clause() {
-        test_parse_query_to_ast_helper("(+a +b)", "(+\"a\" +\"b\")");
-    }
-
-    #[test]
-    fn test_parse_test_query_plus_a_b_plus_d() {
-        test_parse_query_to_ast_helper("+(a b) +d", "(+(*\"a\" *\"b\") +\"d\")");
-    }
-
-    #[test]
-    fn test_parse_test_query_other() {
-        test_parse_query_to_ast_helper("(+a +b) d", "(*(+\"a\" +\"b\") *\"d\")");
-        test_parse_query_to_ast_helper("+abc:toto", "abc:\"toto\"");
-        test_parse_query_to_ast_helper("(+abc:toto -titi)", "(+abc:\"toto\" -\"titi\")");
-        test_parse_query_to_ast_helper("-abc:toto", "(-abc:\"toto\")");
-        test_parse_query_to_ast_helper("abc:a b", "(*abc:\"a\" *\"b\")");
+        test_parse_query_to_ast_helper("+abc:toto", "+(abc:\"toto\")");
+        test_parse_query_to_ast_helper("(+abc:toto -titi)", "(+(abc:\"toto\") -(\"titi\"))");
+        test_parse_query_to_ast_helper("-abc:toto", "-(abc:\"toto\")");
+        test_parse_query_to_ast_helper("abc:a b", "(abc:\"a\" \"b\")");
         test_parse_query_to_ast_helper("abc:\"a b\"", "abc:\"a b\"");
         test_parse_query_to_ast_helper("foo:[1 TO 5]", "foo:[\"1\" TO \"5\"]");
-    }
-
-    #[test]
-    fn test_parse_query_with_range() {
         test_parse_query_to_ast_helper("[1 TO 5]", "[\"1\" TO \"5\"]");
         test_parse_query_to_ast_helper("foo:{a TO z}", "foo:{\"a\" TO \"z\"}");
         test_parse_query_to_ast_helper("foo:[1 TO toto}", "foo:[\"1\" TO \"toto\"}");

query-grammar/src/user_input_ast.rs

@@ -85,14 +85,15 @@ impl UserInputBound {
 }
 
 pub enum UserInputAST {
-    Clause(Vec<(Option<Occur>, UserInputAST)>),
+    Clause(Vec<UserInputAST>),
+    Unary(Occur, Box<UserInputAST>),
     Leaf(Box<UserInputLeaf>),
     Boost(Box<UserInputAST>, f32),
 }
 
 impl UserInputAST {
     pub fn unary(self, occur: Occur) -> UserInputAST {
-        UserInputAST::Clause(vec![(Some(occur), self)])
+        UserInputAST::Unary(occur, Box::new(self))
     }
 
     fn compose(occur: Occur, asts: Vec<UserInputAST>) -> UserInputAST {
@@ -103,7 +104,7 @@ impl UserInputAST {
         } else {
             UserInputAST::Clause(
                 asts.into_iter()
-                    .map(|ast: UserInputAST| (Some(occur), ast))
+                    .map(|ast: UserInputAST| ast.unary(occur))
                     .collect::<Vec<_>>(),
             )
         }
@@ -134,36 +135,25 @@ impl From<UserInputLeaf> for UserInputAST {
     }
 }
 
-fn print_occur_ast(
-    occur_opt: Option<Occur>,
-    ast: &UserInputAST,
-    formatter: &mut fmt::Formatter,
-) -> fmt::Result {
-    if let Some(occur) = occur_opt {
-        write!(formatter, "{}{:?}", occur, ast)?;
-    } else {
-        write!(formatter, "*{:?}", ast)?;
-    }
-    Ok(())
-}
-
 impl fmt::Debug for UserInputAST {
-    fn fmt(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
         match *self {
             UserInputAST::Clause(ref subqueries) => {
                 if subqueries.is_empty() {
                     write!(formatter, "<emptyclause>")?;
                 } else {
                     write!(formatter, "(")?;
-                    print_occur_ast(subqueries[0].0, &subqueries[0].1, formatter)?;
+                    write!(formatter, "{:?}", &subqueries[0])?;
                     for subquery in &subqueries[1..] {
-                        write!(formatter, " ")?;
-                        print_occur_ast(subquery.0, &subquery.1, formatter)?;
+                        write!(formatter, " {:?}", subquery)?;
                     }
                     write!(formatter, ")")?;
                 }
                 Ok(())
             }
+            UserInputAST::Unary(ref occur, ref subquery) => {
+                write!(formatter, "{}({:?})", occur, subquery)
+            }
             UserInputAST::Leaf(ref subquery) => write!(formatter, "{:?}", subquery),
             UserInputAST::Boost(ref leaf, boost) => write!(formatter, "({:?})^{}", leaf, boost),
         }

src/collector/top_score_collector.rs

@@ -450,6 +450,7 @@ mod tests {
     use crate::Index;
     use crate::IndexWriter;
     use crate::Score;
+    use itertools::Itertools;
 
     fn make_index() -> Index {
         let mut schema_builder = Schema::builder();
@@ -523,8 +524,8 @@ mod tests {
 
         // precondition for the test to be meaningful: we did get documents
         // with the same score
-        assert!(page_1.iter().all(|result| result.0 == page_1[0].0));
-        assert!(page_2.iter().all(|result| result.0 == page_2[0].0));
+        assert!(page_1.iter().map(|result| result.0).all_equal());
+        assert!(page_2.iter().map(|result| result.0).all_equal());
 
         // sanity check since we're relying on make_index()
         assert_eq!(page_1.len(), 2);

src/common/mod.rs

@@ -18,19 +18,6 @@ pub use byteorder::LittleEndian as Endianness;
 /// We do not allow segments with more than
 pub const MAX_DOC_LIMIT: u32 = 1 << 31;
 
-pub fn minmax<I, T>(mut vals: I) -> Option<(T, T)>
-where
-    I: Iterator<Item = T>,
-    T: Copy + Ord,
-{
-    if let Some(first_el) = vals.next() {
-        return Some(vals.fold((first_el, first_el), |(min_val, max_val), el| {
-            (min_val.min(el), max_val.max(el))
-        }));
-    }
-    None
-}
-
 /// Computes the number of bits that will be used for bitpacking.
 ///
 /// In general the target is the minimum number of bits
@@ -147,7 +134,6 @@ pub fn u64_to_f64(val: u64) -> f64 {
 #[cfg(test)]
 pub(crate) mod test {
 
-    pub use super::minmax;
     pub use super::serialize::test::fixed_size_test;
     use super::{compute_num_bits, f64_to_u64, i64_to_u64, u64_to_f64, u64_to_i64};
     use std::f64;
@@ -213,21 +199,4 @@ pub(crate) mod test {
         assert!(((super::MAX_DOC_LIMIT - 1) as i32) >= 0);
         assert!((super::MAX_DOC_LIMIT as i32) < 0);
     }
-
-    #[test]
-    fn test_minmax_empty() {
-        let vals: Vec<u32> = vec![];
-        assert_eq!(minmax(vals.into_iter()), None);
-    }
-
-    #[test]
-    fn test_minmax_one() {
-        assert_eq!(minmax(vec![1].into_iter()), Some((1, 1)));
-    }
-
-    #[test]
-    fn test_minmax_two() {
-        assert_eq!(minmax(vec![1, 2].into_iter()), Some((1, 2)));
-        assert_eq!(minmax(vec![2, 1].into_iter()), Some((1, 2)));
-    }
 }

src/directory/mod.rs

@@ -13,6 +13,7 @@ mod footer;
 mod managed_directory;
 mod ram_directory;
 mod read_only_source;
+mod spilling_writer;
 mod watch_event_router;
 
 /// Errors specific to the directory module.
@@ -22,6 +23,7 @@ pub use self::directory::DirectoryLock;
 pub use self::directory::{Directory, DirectoryClone};
 pub use self::directory_lock::{Lock, INDEX_WRITER_LOCK, META_LOCK};
 pub use self::ram_directory::RAMDirectory;
+pub(crate) use self::spilling_writer::SpillingWriter;
 pub use self::read_only_source::ReadOnlySource;
 pub use self::watch_event_router::{WatchCallback, WatchCallbackList, WatchHandle};
 use std::io::{self, BufWriter, Write};
@@ -79,10 +81,16 @@ impl<W: TerminatingWrite> TerminatingWrite for BufWriter<W> {
     }
 }
 
+impl TerminatingWrite for Vec<u8> {
+    fn terminate_ref(&mut self, _a: AntiCallToken) -> io::Result<()> {
+        Ok(())
+    }
+}
+
 #[cfg(test)]
 impl<'a> TerminatingWrite for &'a mut Vec<u8> {
     fn terminate_ref(&mut self, _a: AntiCallToken) -> io::Result<()> {
-        self.flush()
+        Ok(())
     }
 }
 

src/directory/spilling_writer.rs

@@ -0,0 +1,180 @@
+use crate::directory::{WritePtr, TerminatingWrite};
+use std::io::{self, Write};
+
+enum SpillingState {
+    Buffer {
+        buffer: Vec<u8>,
+        capacity: usize,
+        write_factory: Box<dyn FnOnce() -> io::Result<WritePtr>>,
+    },
+    Spilled(WritePtr),
+}
+
+impl SpillingState {
+
+    fn new(
+        limit: usize,
+        write_factory: Box<dyn FnOnce() -> io::Result<WritePtr>>,
+    ) -> SpillingState {
+        SpillingState::Buffer {
+            buffer: Vec::with_capacity(limit),
+            capacity: limit,
+            write_factory,
+        }
+    }
+
+    fn reserve(self, extra_capacity: usize) -> io::Result<SpillingState> {
+        match self {
+            SpillingState::Buffer {
+                buffer,
+                capacity,
+                write_factory,
+            } => {
+                if capacity >= extra_capacity {
+                    Ok(SpillingState::Buffer {
+                        buffer,
+                        capacity: capacity - extra_capacity,
+                        write_factory,
+                    })
+                } else {
+                    let mut wrt = write_factory()?;
+                    wrt.write_all(&buffer[..])?;
+                    Ok(SpillingState::Spilled(wrt))
+                }
+            }
+            SpillingState::Spilled(wrt) => Ok(SpillingState::Spilled(wrt)),
+        }
+    }
+}
+
+pub struct SpillingWriter {
+    state: Option<SpillingState>,
+}
+
+impl SpillingWriter {
+    pub fn new(
+        limit: usize,
+        write_factory: Box<dyn FnOnce() -> io::Result<WritePtr>>,
+    ) -> SpillingWriter {
+        let state = SpillingState::new(limit, write_factory);
+        SpillingWriter {
+            state: Some(state)
+        }
+    }
+
+    pub fn flush_and_finalize(self) -> io::Result<()> {
+        if let SpillingState::Buffer {
+            buffer,
+            write_factory,
+            ..
+        } = self.state.expect("State cannot be none") {
+            let mut wrt = write_factory()?;
+            wrt.write_all(&buffer[..])?;
+            wrt.flush()?;
+            wrt.terminate()?;
+        }
+        Ok(())
+    }
+
+    pub fn finalize(self) -> io::Result<SpillingResult> {
+        match self.state.expect("state cannot be None") {
+            SpillingState::Spilled(mut wrt) => {
+                wrt.flush()?;
+                Ok(SpillingResult::Spilled)
+            }
+            SpillingState::Buffer { buffer, .. } => Ok(SpillingResult::Buffer(buffer)),
+
+        }
+    }
+}
+
+pub enum SpillingResult {
+    Spilled,
+    Buffer(Vec<u8>),
+}
+
+impl io::Write for SpillingWriter {
+    fn write(&mut self, payload: &[u8]) -> io::Result<usize> {
+        self.write_all(payload)?;
+        Ok(payload.len())
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        if let Some(SpillingState::Spilled(wrt)) = &mut self.state {
+            wrt.flush()?;
+        }
+        Ok(())
+    }
+
+    fn write_all(&mut self, payload: &[u8]) -> io::Result<()> {
+        let state_opt: Option<io::Result<SpillingState>> =  self.state
+            .take()
+            .map(|mut state| {
+                state = state.reserve(payload.len())?;
+                match &mut state {
+                    SpillingState::Buffer { buffer, .. } => {
+                        buffer.extend_from_slice(payload);
+                    }
+                    SpillingState::Spilled(wrt) => {
+                        wrt.write_all(payload)?;
+                    }
+                }
+                Ok(state)
+            });
+        self.state = state_opt.transpose()?;
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::SpillingWriter;
+    use crate::directory::spilling_writer::SpillingResult;
+    use crate::directory::RAMDirectory;
+    use crate::Directory;
+    use std::io::{self, Write};
+    use std::path::Path;
+
+    #[test]
+    fn test_no_spilling() {
+        let ram_directory = RAMDirectory::create();
+        let mut ram_directory_clone = ram_directory.clone();
+        let path = Path::new("test");
+        let write_factory = Box::new(move || {
+            ram_directory_clone
+                .open_write(path)
+                .map_err(|err| io::Error::new(io::ErrorKind::Other, err))
+        });
+        let mut spilling_wrt = SpillingWriter::new(10, write_factory);
+        assert!(spilling_wrt.write_all(b"abcd").is_ok());
+        if let SpillingResult::Buffer(buf) = spilling_wrt.finalize().unwrap() {
+            assert_eq!(buf, b"abcd")
+        } else {
+            panic!("spill writer should not have spilled");
+        }
+        assert!(!ram_directory.exists(path));
+    }
+
+    #[test]
+    fn test_spilling() {
+        let ram_directory = RAMDirectory::create();
+        let mut ram_directory_clone = ram_directory.clone();
+        let path = Path::new("test");
+        let write_factory = Box::new(move || {
+            ram_directory_clone
+                .open_write(path)
+                .map_err(|err| io::Error::new(io::ErrorKind::Other, err))
+        });
+        let mut spilling_wrt = SpillingWriter::new(10, write_factory);
+        assert!(spilling_wrt.write_all(b"abcd").is_ok());
+        assert!(spilling_wrt.write_all(b"efghijklmnop").is_ok());
+        if let SpillingResult::Spilled = spilling_wrt.finalize().unwrap() {
+        } else {
+            panic!("spill writer should have spilled");
+        }
+        assert_eq!(
+            ram_directory.atomic_read(path).unwrap(),
+            b"abcdefghijklmnop"
+        );
+    }
+}

src/fastfield/multivalued/writer.rs

@@ -6,6 +6,7 @@ use crate::schema::{Document, Field};
 use crate::termdict::TermOrdinal;
 use crate::DocId;
 use fnv::FnvHashMap;
+use itertools::Itertools;
 use std::io;
 
 /// Writer for multi-valued (as in, more than one value per document)
@@ -150,8 +151,8 @@ impl MultiValueIntFastFieldWriter {
                     }
                 }
                 None => {
-                    let val_min_max = crate::common::minmax(self.vals.iter().cloned());
-                    let (val_min, val_max) = val_min_max.unwrap_or((0u64, 0u64));
+                    let val_min_max = self.vals.iter().cloned().minmax();
+                    let (val_min, val_max) = val_min_max.into_option().unwrap_or((0u64, 0u64));
                     value_serializer =
                         serializer.new_u64_fast_field_with_idx(self.field, val_min, val_max, 1)?;
                     for &val in &self.vals {

src/indexer/merger.rs

@@ -2,6 +2,7 @@ use crate::common::MAX_DOC_LIMIT;
 use crate::core::Segment;
 use crate::core::SegmentReader;
 use crate::core::SerializableSegment;
+use crate::directory::WritePtr;
 use crate::docset::DocSet;
 use crate::fastfield::BytesFastFieldReader;
 use crate::fastfield::DeleteBitSet;
@@ -21,6 +22,7 @@ use crate::store::StoreWriter;
 use crate::termdict::TermMerger;
 use crate::termdict::TermOrdinal;
 use crate::DocId;
+use itertools::Itertools;
 use std::cmp;
 use std::collections::HashMap;
 
@@ -69,11 +71,11 @@ fn compute_min_max_val(
             Some(delete_bitset) => {
                 // some deleted documents,
                 // we need to recompute the max / min
-                crate::common::minmax(
-                    (0..max_doc)
-                        .filter(|doc_id| delete_bitset.is_alive(*doc_id))
-                        .map(|doc_id| u64_reader.get(doc_id)),
-                )
+                (0..max_doc)
+                    .filter(|doc_id| delete_bitset.is_alive(*doc_id))
+                    .map(|doc_id| u64_reader.get(doc_id))
+                    .minmax()
+                    .into_option()
             }
             None => {
                 // no deleted documents,
@@ -660,7 +662,8 @@ impl IndexMerger {
         Ok(term_ordinal_mappings)
     }
 
-    fn write_storable_fields(&self, store_writer: &mut StoreWriter) -> crate::Result<()> {
+    pub fn write_storable_fields(&self, store_wrt: WritePtr) -> crate::Result<()> {
+        let mut store_writer = StoreWriter::new(store_wrt);
         for reader in &self.readers {
             let store_reader = reader.get_store_reader();
             if reader.num_deleted_docs() > 0 {
@@ -672,6 +675,7 @@ impl IndexMerger {
                 store_writer.stack(&store_reader)?;
             }
         }
+        store_writer.close()?;
         Ok(())
     }
 }
@@ -681,7 +685,6 @@ impl SerializableSegment for IndexMerger {
         let term_ord_mappings = self.write_postings(serializer.get_postings_serializer())?;
         self.write_fieldnorms(serializer.get_fieldnorms_serializer())?;
         self.write_fast_fields(serializer.get_fast_field_serializer(), term_ord_mappings)?;
-        self.write_storable_fields(serializer.get_store_writer())?;
         serializer.close()?;
         Ok(self.max_doc)
     }

src/indexer/segment_serializer.rs

@@ -3,12 +3,10 @@ use crate::core::SegmentComponent;
 use crate::fastfield::FastFieldSerializer;
 use crate::fieldnorm::FieldNormsSerializer;
 use crate::postings::InvertedIndexSerializer;
-use crate::store::StoreWriter;
 
 /// Segment serializer is in charge of laying out on disk
 /// the data accumulated and sorted by the `SegmentWriter`.
 pub struct SegmentSerializer {
-    store_writer: StoreWriter,
     fast_field_serializer: FastFieldSerializer,
     fieldnorms_serializer: FieldNormsSerializer,
     postings_serializer: InvertedIndexSerializer,
@@ -17,8 +15,6 @@ pub struct SegmentSerializer {
 impl SegmentSerializer {
     /// Creates a new `SegmentSerializer`.
     pub fn for_segment(segment: &mut Segment) -> crate::Result<SegmentSerializer> {
-        let store_write = segment.open_write(SegmentComponent::STORE)?;
-
         let fast_field_write = segment.open_write(SegmentComponent::FASTFIELDS)?;
         let fast_field_serializer = FastFieldSerializer::from_write(fast_field_write)?;
 
@@ -27,7 +23,6 @@ impl SegmentSerializer {
 
         let postings_serializer = InvertedIndexSerializer::open(segment)?;
         Ok(SegmentSerializer {
-            store_writer: StoreWriter::new(store_write),
             fast_field_serializer,
             fieldnorms_serializer,
             postings_serializer,
@@ -49,16 +44,10 @@ impl SegmentSerializer {
         &mut self.fieldnorms_serializer
     }
 
-    /// Accessor to the `StoreWriter`.
-    pub fn get_store_writer(&mut self) -> &mut StoreWriter {
-        &mut self.store_writer
-    }
-
     /// Finalize the segment serialization.
     pub fn close(self) -> crate::Result<()> {
         self.fast_field_serializer.close()?;
         self.postings_serializer.close()?;
-        self.store_writer.close()?;
         self.fieldnorms_serializer.close()?;
         Ok(())
     }

src/indexer/segment_updater.rs

@@ -18,7 +18,7 @@ use crate::indexer::SegmentSerializer;
 use crate::indexer::{DefaultMergePolicy, MergePolicy};
 use crate::indexer::{MergeCandidate, MergeOperation};
 use crate::schema::Schema;
-use crate::Opstamp;
+use crate::{Opstamp, SegmentComponent};
 use futures::channel::oneshot;
 use futures::executor::{ThreadPool, ThreadPoolBuilder};
 use futures::future::Future;
@@ -134,8 +134,10 @@ fn merge(
     // ... we just serialize this index merger in our new segment to merge the two segments.
     let segment_serializer = SegmentSerializer::for_segment(&mut merged_segment)?;
 
-    let num_docs = merger.write(segment_serializer)?;
+    let store_wrt = merged_segment.open_write(SegmentComponent::STORE)?;
+    merger.write_storable_fields(store_wrt)?;
 
+    let num_docs = merger.write(segment_serializer)?;
     let segment_meta = index.new_segment_meta(merged_segment.id(), num_docs);
 
     Ok(SegmentEntry::new(segment_meta, delete_cursor, None))

src/indexer/segment_writer.rs

@@ -11,13 +11,15 @@ use crate::schema::Schema;
 use crate::schema::Term;
 use crate::schema::Value;
 use crate::schema::{Field, FieldEntry};
+use crate::store::StoreWriter;
 use crate::tokenizer::{BoxTokenStream, PreTokenizedStream};
 use crate::tokenizer::{FacetTokenizer, TextAnalyzer};
 use crate::tokenizer::{TokenStreamChain, Tokenizer};
-use crate::DocId;
 use crate::Opstamp;
+use crate::{DocId, SegmentComponent};
 use std::io;
 use std::str;
+use crate::directory::SpillingWriter;
 
 /// Computes the initial size of the hash table.
 ///
@@ -43,11 +45,12 @@ fn initial_table_size(per_thread_memory_budget: usize) -> crate::Result<usize> {
 pub struct SegmentWriter {
     max_doc: DocId,
     multifield_postings: MultiFieldPostingsWriter,
-    segment_serializer: SegmentSerializer,
+    segment: Segment,
     fast_field_writers: FastFieldsWriter,
     fieldnorms_writer: FieldNormsWriter,
     doc_opstamps: Vec<Opstamp>,
     tokenizers: Vec<Option<TextAnalyzer>>,
+    store_writer: StoreWriter<SpillingWriter>,
 }
 
 impl SegmentWriter {
@@ -62,11 +65,10 @@ impl SegmentWriter {
     /// - schema
     pub fn for_segment(
         memory_budget: usize,
-        mut segment: Segment,
+        segment: Segment,
         schema: &Schema,
     ) -> crate::Result<SegmentWriter> {
         let table_num_bits = initial_table_size(memory_budget)?;
-        let segment_serializer = SegmentSerializer::for_segment(&mut segment)?;
         let multifield_postings = MultiFieldPostingsWriter::new(schema, table_num_bits);
         let tokenizers = schema
             .fields()
@@ -82,14 +84,22 @@ impl SegmentWriter {
                 },
             )
             .collect();
+        let mut segment_clone = segment.clone();
+        let spilling_wrt = SpillingWriter::new(1_000, Box::new(move || {
+            segment_clone
+                .open_write(SegmentComponent::STORE)
+                .map_err(|err| io::Error::new(io::ErrorKind::Other, err))
+        }));
+        let store_writer = StoreWriter::new(spilling_wrt);
         Ok(SegmentWriter {
             max_doc: 0,
             multifield_postings,
             fieldnorms_writer: FieldNormsWriter::for_schema(schema),
-            segment_serializer,
+            segment,
             fast_field_writers: FastFieldsWriter::from_schema(schema),
             doc_opstamps: Vec::with_capacity(1_000),
             tokenizers,
+            store_writer,
         })
     }
 
@@ -99,11 +109,14 @@ impl SegmentWriter {
     /// be used afterwards.
     pub fn finalize(mut self) -> crate::Result<Vec<u64>> {
         self.fieldnorms_writer.fill_up_to_max_doc(self.max_doc);
+        let spilling_wrt = self.store_writer.close()?;
+        spilling_wrt.flush_and_finalize()?;
+        let segment_serializer = SegmentSerializer::for_segment(&mut self.segment)?;
         write(
             &self.multifield_postings,
             &self.fast_field_writers,
             &self.fieldnorms_writer,
-            self.segment_serializer,
+            segment_serializer,
         )?;
         Ok(self.doc_opstamps)
     }
@@ -246,8 +259,7 @@ impl SegmentWriter {
         }
         doc.filter_fields(|field| schema.get_field_entry(field).is_stored());
         doc.prepare_for_store();
-        let doc_writer = self.segment_serializer.get_store_writer();
-        doc_writer.store(&doc)?;
+        self.store_writer.store(&doc)?;
         self.max_doc += 1;
         Ok(())
     }

src/lib.rs

@@ -269,144 +269,6 @@ impl DocAddress {
     }
 }
 
-
-
-mod sse2 {
-    use crate::postings::compression::{AlignedBuffer, COMPRESSION_BLOCK_SIZE};
-    use std::arch::x86_64::__m128i as DataType;
-    use std::arch::x86_64::_mm_add_epi32 as op_add;
-    use std::arch::x86_64::_mm_cmplt_epi32 as op_lt;
-    use std::arch::x86_64::_mm_load_si128 as op_load;
-    // requires 128-bits alignment
-    use std::arch::x86_64::_mm_set1_epi32 as set1;
-    use std::arch::x86_64::_mm_setzero_si128 as set0;
-    use std::arch::x86_64::_mm_sub_epi32 as op_sub;
-    use std::arch::x86_64::{_mm_cvtsi128_si32, _mm_shuffle_epi32};
-
-    const MASK1: i32 = 78;
-    const MASK2: i32 = 177;
-
-    /// Performs an exhaustive linear search over the
-    ///
-    /// There is no early exit here. We simply count the
-    /// number of elements that are `< target`.
-    pub unsafe fn linear_search_sse2_128(arr: &AlignedBuffer, target: u32) -> usize {
-        let ptr = arr as *const AlignedBuffer as *const DataType;
-        let vkey = set1(target as i32);
-        let mut cnt = set0();
-        // We work over 4 `__m128i` at a time.
-        // A single `__m128i` actual contains 4 `u32`.
-        for i in 0..(COMPRESSION_BLOCK_SIZE as isize) / (4 * 4) {
-            let cmp1 = op_lt(op_load(ptr.offset(i * 4)), vkey);
-            let cmp2 = op_lt(op_load(ptr.offset(i * 4 + 1)), vkey);
-            let cmp3 = op_lt(op_load(ptr.offset(i * 4 + 2)), vkey);
-            let cmp4 = op_lt(op_load(ptr.offset(i * 4 + 3)), vkey);
-            let sum = op_add(op_add(cmp1, cmp2), op_add(cmp3, cmp4));
-            cnt = op_sub(cnt, sum);
-        }
-        cnt = op_add(cnt, _mm_shuffle_epi32(cnt, MASK1));
-        cnt = op_add(cnt, _mm_shuffle_epi32(cnt, MASK2));
-        _mm_cvtsi128_si32(cnt) as usize
-    }
-}
-
-
-const MAX_DOC: u32 = 2_000_000_000u32;
-
-use sse2::linear_search_sse2_128;
-use crate::postings::BlockSegmentPostings;
-use crate::schema::IndexRecordOption;
-
-struct DocCursor {
-    block_segment_postings: BlockSegmentPostings,
-    cursor: usize,
-}
-
-impl DocCursor {
-    fn new(mut block_segment_postings: BlockSegmentPostings) -> DocCursor {
-        block_segment_postings.advance();
-        DocCursor {
-            block_segment_postings,
-            cursor: 0,
-        }
-    }
-    fn doc(&self,) -> DocId {
-        self.block_segment_postings.doc(self.cursor)
-    }
-
-    fn len(&self,) -> usize {
-        self.block_segment_postings.doc_freq()
-    }
-
-    fn advance(&mut self) {
-        if self.cursor == 127 {
-            self.block_segment_postings.advance();
-            self.cursor = 0;
-        } else {
-            self.cursor += 1;
-        }
-    }
-
-    fn skip_to(&mut self, target: DocId) {
-        if self.block_segment_postings.skip_reader.doc() >= target {
-            unsafe {
-                let mut ptr = self.block_segment_postings.docs_aligned_b().0.get_unchecked(self.cursor) as *const u32;
-                while *ptr < target {
-                    self.cursor += 1;
-                    ptr = ptr.offset(1);
-                }
-            }
-            return;
-        }
-        self.block_segment_postings.skip_to_b(target);
-        let block= self.block_segment_postings.docs_aligned_b();
-        self.cursor = unsafe { linear_search_sse2_128(&block, target) };
-
-    }
-}
-
-pub fn intersection(idx: &InvertedIndexReader, terms: &[Term]) -> crate::Result<u32> {
-    let mut posting_lists: Vec<DocCursor> = Vec::with_capacity(terms.len());
-    for term in terms {
-        if let Some(mut block_postings) = idx.read_block_postings(term, IndexRecordOption::Basic) {
-            posting_lists.push(DocCursor::new(block_postings ));
-        } else {
-            return Ok(0);
-        }
-    }
-    posting_lists.sort_by_key(|posting_list| posting_list.len());
-    Ok({ intersection_idx(&mut posting_lists[..]) })
-}
-
-fn intersection_idx(posting_lists: &mut [DocCursor]) -> DocId {
-    let mut candidate = posting_lists[0].doc();
-    let mut i = 1;
-
-    let mut count = 0u32;
-
-    let num_posting_lists = posting_lists.len();
-    loop {
-        while i < num_posting_lists {
-            posting_lists[i].skip_to(candidate);
-            if posting_lists[i].doc() != candidate {
-                candidate = posting_lists[i].doc();
-                i = 0;
-                continue;
-            }
-            i += 1;
-        }
-        count += 1;
-        if candidate == MAX_DOC {
-            break;
-        }
-        posting_lists[0].advance();
-        candidate = posting_lists[0].doc();
-        i = 1;
-    }
-    count - 1u32
-}
-
-
 /// `DocAddress` contains all the necessary information
 /// to identify a document given a `Searcher` object.
 ///

src/postings/block_search.rs

@@ -25,7 +25,7 @@ mod sse2 {
     ///
     /// There is no early exit here. We simply count the
     /// number of elements that are `< target`.
-    pub fn linear_search_sse2_128(arr: &AlignedBuffer, target: u32) -> usize {
+    pub(crate) fn linear_search_sse2_128(arr: &AlignedBuffer, target: u32) -> usize {
         unsafe {
             let ptr = arr as *const AlignedBuffer as *const DataType;
             let vkey = set1(target as i32);
@@ -63,8 +63,6 @@ mod sse2 {
     }
 }
 
-pub use self::sse2::linear_search_sse2_128;
-
 /// This `linear search` browser exhaustively through the array.
 /// but the early exit is very difficult to predict.
 ///

src/postings/compression/mod.rs

@@ -46,7 +46,7 @@ impl BlockEncoder {
 /// We ensure that the OutputBuffer is align on 128 bits
 /// in order to run SSE2 linear search on it.
 #[repr(align(128))]
-pub struct AlignedBuffer(pub [u32; COMPRESSION_BLOCK_SIZE]);
+pub(crate) struct AlignedBuffer(pub [u32; COMPRESSION_BLOCK_SIZE]);
 
 pub struct BlockDecoder {
     bitpacker: BitPacker4x,
@@ -67,18 +67,6 @@ impl BlockDecoder {
         }
     }
 
-    pub fn uncompress_vint_sorted_b<'a>(
-        &mut self,
-        compressed_data: &'a [u8],
-        offset: u32,
-        num_els: usize,
-    ) {
-        if num_els > 0 {
-            vint::uncompress_sorted(compressed_data, &mut self.output.0[..num_els], offset);
-        }
-        self.output.0[num_els..].iter_mut().for_each(|val| *val = 2_000_000_000u32);
-    }
-
     pub fn uncompress_block_sorted(
         &mut self,
         compressed_data: &[u8],
@@ -106,11 +94,6 @@ impl BlockDecoder {
         (&self.output, self.output_len)
     }
 
-    #[inline]
-    pub(crate) fn output_aligned_2(&self) -> &AlignedBuffer {
-        &self.output
-    }
-
     #[inline]
     pub fn output(&self, idx: usize) -> u32 {
         self.output.0[idx]

src/postings/mod.rs

@@ -3,7 +3,7 @@ Postings module (also called inverted index)
 */
 
 mod block_search;
-pub mod compression;
+pub(crate) mod compression;
 /// Postings module
 ///
 /// Postings, also called inverted lists, is the key datastructure

src/postings/segment_postings.rs

@@ -317,7 +317,7 @@ pub struct BlockSegmentPostings {
     num_vint_docs: usize,
 
     remaining_data: OwnedRead,
-    pub skip_reader: SkipReader,
+    skip_reader: SkipReader,
 }
 
 fn split_into_skips_and_postings(
@@ -414,12 +414,7 @@ impl BlockSegmentPostings {
         self.doc_decoder.output_array()
     }
 
-    #[inline(always)]
-    pub fn docs_aligned_b(&self) -> &AlignedBuffer {
-        self.doc_decoder.output_aligned_2()
-    }
-
-    pub fn docs_aligned(&self) -> (&AlignedBuffer, usize) {
+    pub(crate) fn docs_aligned(&self) -> (&AlignedBuffer, usize) {
         self.doc_decoder.output_aligned()
     }
 
@@ -491,7 +486,6 @@ impl BlockSegmentPostings {
                     }
                 }
                 self.doc_offset = self.skip_reader.doc();
-                unsafe { core::arch::x86_64::_mm_prefetch(self.remaining_data.as_ref() as *mut i8, _MM_HINT_T1) };
                 return BlockSegmentPostingsSkipResult::Success(skip_freqs);
             } else {
                 skip_freqs += self.skip_reader.tf_sum();
@@ -532,41 +526,6 @@ impl BlockSegmentPostings {
         BlockSegmentPostingsSkipResult::Terminated
     }
 
-    /// Ensure we are located on the right block.
-    #[inline(never)]
-    pub fn skip_to_b(&mut self, target_doc: DocId) {
-       while self.skip_reader.advance() {
-            if self.skip_reader.doc() >= target_doc {
-                // the last document of the current block is larger
-                // than the target.
-                //
-                // We found our block!
-                let num_bits = self.skip_reader.doc_num_bits();
-                let num_consumed_bytes = self.doc_decoder.uncompress_block_sorted(
-                    self.remaining_data.as_ref(),
-                    self.doc_offset,
-                    num_bits,
-                );
-                let tf_num_bits = self.skip_reader.tf_num_bits();
-                let num_bytes_to_skip = compressed_block_size(tf_num_bits);
-                self.remaining_data.advance(num_consumed_bytes + num_bytes_to_skip);
-                self.doc_offset = self.skip_reader.doc();
-                return;
-            } else {
-                let advance_len = self.skip_reader.total_block_len();
-                self.doc_offset = self.skip_reader.doc();
-                self.remaining_data.advance(advance_len);
-            }
-        }
-
-        // we are now on the last, incomplete, variable encoded block.
-        self.doc_decoder.uncompress_vint_sorted_b(
-           self.remaining_data.as_ref(),
-           self.doc_offset,
-          self.num_vint_docs
-        );
-    }
-
     /// Advance to the next block.
     ///
     /// Returns false iff there was no remaining blocks.

src/postings/skip.rs

@@ -49,7 +49,7 @@ impl SkipSerializer {
     }
 }
 
-pub struct SkipReader {
+pub(crate) struct SkipReader {
     doc: DocId,
     owned_read: OwnedRead,
     doc_num_bits: u8,

src/query/boolean_query/mod.rs

@@ -31,11 +31,24 @@ mod tests {
             // writing the segment
             let mut index_writer = index.writer_with_num_threads(1, 3_000_000).unwrap();
             {
-                index_writer.add_document(doc!(text_field => "a b c"));
-                index_writer.add_document(doc!(text_field => "a c"));
-                index_writer.add_document(doc!(text_field => "b c"));
-                index_writer.add_document(doc!(text_field => "a b c d"));
-                index_writer.add_document(doc!(text_field => "d"));
+                let doc = doc!(text_field => "a b c");
+                index_writer.add_document(doc);
+            }
+            {
+                let doc = doc!(text_field => "a c");
+                index_writer.add_document(doc);
+            }
+            {
+                let doc = doc!(text_field => "b c");
+                index_writer.add_document(doc);
+            }
+            {
+                let doc = doc!(text_field => "a b c d");
+                index_writer.add_document(doc);
+            }
+            {
+                let doc = doc!(text_field => "d");
+                index_writer.add_document(doc);
             }
             assert!(index_writer.commit().is_ok());
         }

src/query/fuzzy_query.rs

@@ -2,36 +2,10 @@ use crate::query::{AutomatonWeight, Query, Weight};
 use crate::schema::Term;
 use crate::Searcher;
 use crate::TantivyError::InvalidArgument;
-use levenshtein_automata::{Distance, LevenshteinAutomatonBuilder, DFA};
+use levenshtein_automata::{LevenshteinAutomatonBuilder, DFA};
 use once_cell::sync::Lazy;
 use std::collections::HashMap;
 use std::ops::Range;
-use tantivy_fst::Automaton;
-
-pub(crate) struct DFAWrapper(pub DFA);
-
-impl Automaton for DFAWrapper {
-    type State = u32;
-
-    fn start(&self) -> Self::State {
-        self.0.initial_state()
-    }
-
-    fn is_match(&self, state: &Self::State) -> bool {
-        match self.0.distance(*state) {
-            Distance::Exact(_) => true,
-            Distance::AtLeast(_) => false,
-        }
-    }
-
-    fn can_match(&self, state: &u32) -> bool {
-        *state != levenshtein_automata::SINK_STATE
-    }
-
-    fn accept(&self, state: &Self::State, byte: u8) -> Self::State {
-        self.0.transition(*state, byte)
-    }
-}
 
 /// A range of Levenshtein distances that we will build DFAs for our terms
 /// The computation is exponential, so best keep it to low single digits
@@ -127,20 +101,13 @@ impl FuzzyTermQuery {
         }
     }
 
-    fn specialized_weight(&self) -> crate::Result<AutomatonWeight<DFAWrapper>> {
+    fn specialized_weight(&self) -> crate::Result<AutomatonWeight<DFA>> {
         // LEV_BUILDER is a HashMap, whose `get` method returns an Option
         match LEV_BUILDER.get(&(self.distance, false)) {
             // Unwrap the option and build the Ok(AutomatonWeight)
             Some(automaton_builder) => {
-                let automaton = if self.prefix {
-                    automaton_builder.build_prefix_dfa(self.term.text())
-                } else {
-                    automaton_builder.build_dfa(self.term.text())
-                };
-                Ok(AutomatonWeight::new(
-                    self.term.field(),
-                    DFAWrapper(automaton),
-                ))
+                let automaton = automaton_builder.build_dfa(self.term.text());
+                Ok(AutomatonWeight::new(self.term.field(), automaton))
             }
             None => Err(InvalidArgument(format!(
                 "Levenshtein distance of {} is not allowed. Choose a value in the {:?} range",
@@ -199,17 +166,5 @@ mod test {
             let (score, _) = top_docs[0];
             assert_nearly_equals(1f32, score);
         }
-
-        {
-            let term = Term::from_field_text(country_field, "jap");
-
-            let fuzzy_query = FuzzyTermQuery::new_prefix(term, 1, true);
-            let top_docs = searcher
-                .search(&fuzzy_query, &TopDocs::with_limit(2))
-                .unwrap();
-            assert_eq!(top_docs.len(), 1, "Expected only 1 document");
-            let (score, _) = top_docs[0];
-            assert_nearly_equals(1f32, score);
-        }
     }
 }

src/query/mod.rs

@@ -24,7 +24,6 @@ mod term_query;
 mod union;
 mod weight;
 
-
 #[cfg(test)]
 mod vec_docset;
 
@@ -44,7 +43,6 @@ pub use self::empty_query::{EmptyQuery, EmptyScorer, EmptyWeight};
 pub use self::exclude::Exclude;
 pub use self::explanation::Explanation;
 pub use self::fuzzy_query::FuzzyTermQuery;
-pub(crate) use self::fuzzy_query::DFAWrapper;
 pub use self::intersection::intersect_scorers;
 pub use self::phrase_query::PhraseQuery;
 pub use self::query::Query;

src/query/phrase_query/mod.rs

@@ -12,6 +12,7 @@ pub mod tests {
     use super::*;
     use crate::collector::tests::{TEST_COLLECTOR_WITHOUT_SCORE, TEST_COLLECTOR_WITH_SCORE};
     use crate::core::Index;
+    use crate::error::TantivyError;
     use crate::schema::{Schema, Term, TEXT};
     use crate::tests::assert_nearly_equals;
     use crate::DocAddress;
@@ -126,16 +127,21 @@ pub mod tests {
             Term::from_field_text(text_field, "a"),
             Term::from_field_text(text_field, "b"),
         ]);
-
-        let search_result = searcher
+        match searcher
             .search(&phrase_query, &TEST_COLLECTOR_WITH_SCORE)
-            .map(|_| ());
-        assert!(matches!(
-            search_result,
-            Err(crate::TantivyError::SchemaError(msg))
-            if msg == "Applied phrase query on field \"text\", which does not have positions \
-            indexed"
-        ));
+            .map(|_| ())
+            .unwrap_err()
+        {
+            TantivyError::SchemaError(ref msg) => {
+                assert_eq!(
+                    "Applied phrase query on field \"text\", which does not have positions indexed",
+                    msg.as_str()
+                );
+            }
+            _ => {
+                panic!("Should have returned an error");
+            }
+        }
     }
 
     #[test]

src/query/query.rs

@@ -1,17 +1,11 @@
 use super::Weight;
 use crate::core::searcher::Searcher;
 use crate::query::Explanation;
-use crate::{DocAddress, Score};
+use crate::DocAddress;
 use crate::Term;
 use downcast_rs::impl_downcast;
 use std::collections::BTreeSet;
 use std::fmt;
-use crate::collector::{TopDocs, Collector, SegmentCollector, Count};
-
-pub struct TopKResult {
-    pub count: u64,
-    docs: Vec<(Score, DocAddress)>
-}
 
 /// The `Query` trait defines a set of documents and a scoring method
 /// for those documents.
@@ -62,27 +56,6 @@ pub trait Query: QueryClone + downcast_rs::Downcast + fmt::Debug {
         weight.explain(reader, doc_address.doc())
     }
 
-    fn top_k(&self, searcher: &Searcher, num_hits: usize) -> crate::Result<TopKResult> {
-        let top_docs = TopDocs::with_limit(num_hits);
-        let collector = (Count, top_docs);
-        let weight = self.weight(searcher, false)?;
-        let mut count = 0u64;
-        let mut result = 0;
-        let mut child_fruits = Vec::with_capacity(searcher.segment_readers().len());
-        for (segment_ord, reader) in searcher.segment_readers().iter().enumerate() {
-            let mut child_top_k = collector.for_segment(segment_ord as u32, reader)?;
-            let mut scorer = weight.scorer(reader, 1.0f32)?;
-            // handle deletes
-            scorer.top_k(&mut child_top_k);
-            child_fruits.push(child_top_k.harvest());
-        }
-        let (count, docs) = collector.merge_fruits(child_fruits)?;
-        Ok(TopKResult {
-            count: count as u64,
-            docs
-        })
-    }
-
     /// Returns the number of documents matching the query.
     fn count(&self, searcher: &Searcher) -> crate::Result<usize> {
         let weight = self.weight(searcher, false)?;

src/query/query_parser/query_parser.rs

@@ -174,16 +174,6 @@ pub struct QueryParser {
     boost: HashMap<Field, f32>,
 }
 
-fn all_negative(ast: &LogicalAST) -> bool {
-    match ast {
-        LogicalAST::Leaf(_) => false,
-        LogicalAST::Boost(ref child_ast, _) => all_negative(&*child_ast),
-        LogicalAST::Clause(children) => children
-            .iter()
-            .all(|(ref occur, child)| (*occur == Occur::MustNot) || all_negative(child)),
-    }
-}
-
 impl QueryParser {
     /// Creates a `QueryParser`, given
     /// * schema - index Schema
@@ -263,13 +253,8 @@ impl QueryParser {
         &self,
         user_input_ast: UserInputAST,
     ) -> Result<LogicalAST, QueryParserError> {
-        let ast = self.compute_logical_ast_with_occur(user_input_ast)?;
-        if let LogicalAST::Clause(children) = &ast {
-            if children.is_empty() {
-                return Ok(ast);
-            }
-        }
-        if all_negative(&ast) {
+        let (occur, ast) = self.compute_logical_ast_with_occur(user_input_ast)?;
+        if occur == Occur::MustNot {
             return Err(QueryParserError::AllButQueryForbidden);
         }
         Ok(ast)
@@ -425,23 +410,31 @@ impl QueryParser {
     fn compute_logical_ast_with_occur(
         &self,
         user_input_ast: UserInputAST,
-    ) -> Result<LogicalAST, QueryParserError> {
+    ) -> Result<(Occur, LogicalAST), QueryParserError> {
         match user_input_ast {
             UserInputAST::Clause(sub_queries) => {
                 let default_occur = self.default_occur();
                 let mut logical_sub_queries: Vec<(Occur, LogicalAST)> = Vec::new();
-                for (occur_opt, sub_ast) in sub_queries {
-                    let sub_ast = self.compute_logical_ast_with_occur(sub_ast)?;
-                    let occur = occur_opt.unwrap_or(default_occur);
-                    logical_sub_queries.push((occur, sub_ast));
+                for sub_query in sub_queries {
+                    let (occur, sub_ast) = self.compute_logical_ast_with_occur(sub_query)?;
+                    let new_occur = Occur::compose(default_occur, occur);
+                    logical_sub_queries.push((new_occur, sub_ast));
                 }
-                Ok(LogicalAST::Clause(logical_sub_queries))
+                Ok((Occur::Should, LogicalAST::Clause(logical_sub_queries)))
+            }
+            UserInputAST::Unary(left_occur, subquery) => {
+                let (right_occur, logical_sub_queries) =
+                    self.compute_logical_ast_with_occur(*subquery)?;
+                Ok((Occur::compose(left_occur, right_occur), logical_sub_queries))
             }
             UserInputAST::Boost(ast, boost) => {
-                let ast = self.compute_logical_ast_with_occur(*ast)?;
-                Ok(ast.boost(boost))
+                let (occur, ast_without_occur) = self.compute_logical_ast_with_occur(*ast)?;
+                Ok((occur, ast_without_occur.boost(boost)))
+            }
+            UserInputAST::Leaf(leaf) => {
+                let result_ast = self.compute_logical_ast_from_leaf(*leaf)?;
+                Ok((Occur::Should, result_ast))
             }
-            UserInputAST::Leaf(leaf) => self.compute_logical_ast_from_leaf(*leaf),
         }
     }
 
@@ -789,20 +782,6 @@ mod test {
         );
     }
 
-    #[test]
-    fn test_parse_query_to_ast_ab_c() {
-        test_parse_query_to_logical_ast_helper(
-            "(+title:a +title:b) title:c",
-            "((+Term(field=0,bytes=[97]) +Term(field=0,bytes=[98])) Term(field=0,bytes=[99]))",
-            false,
-        );
-        test_parse_query_to_logical_ast_helper(
-            "(+title:a +title:b) title:c",
-            "(+(+Term(field=0,bytes=[97]) +Term(field=0,bytes=[98])) +Term(field=0,bytes=[99]))",
-            true,
-        );
-    }
-
     #[test]
     pub fn test_parse_query_to_ast_single_term() {
         test_parse_query_to_logical_ast_helper(
@@ -822,13 +801,11 @@ mod test {
              Term(field=1,bytes=[116, 105, 116, 105])))",
             false,
         );
-    }
-
-    #[test]
-    fn test_single_negative_term() {
-        assert_matches!(
-            parse_query_to_logical_ast("-title:toto", false),
-            Err(QueryParserError::AllButQueryForbidden)
+        assert_eq!(
+            parse_query_to_logical_ast("-title:toto", false)
+                .err()
+                .unwrap(),
+            QueryParserError::AllButQueryForbidden
         );
     }
 
@@ -988,18 +965,6 @@ mod test {
         assert!(query_parser.parse_query("with_stop_words:the").is_ok());
     }
 
-    #[test]
-    pub fn test_parse_query_single_negative_term_through_error() {
-        assert_matches!(
-            parse_query_to_logical_ast("-title:toto", true),
-            Err(QueryParserError::AllButQueryForbidden)
-        );
-        assert_matches!(
-            parse_query_to_logical_ast("-title:toto", false),
-            Err(QueryParserError::AllButQueryForbidden)
-        );
-    }
-
     #[test]
     pub fn test_parse_query_to_ast_conjunction() {
         test_parse_query_to_logical_ast_helper(
@@ -1019,6 +984,12 @@ mod test {
              Term(field=1,bytes=[116, 105, 116, 105])))",
             true,
         );
+        assert_eq!(
+            parse_query_to_logical_ast("-title:toto", true)
+                .err()
+                .unwrap(),
+            QueryParserError::AllButQueryForbidden
+        );
         test_parse_query_to_logical_ast_helper(
             "title:a b",
             "(+Term(field=0,bytes=[97]) \
@@ -1042,26 +1013,4 @@ mod test {
             false
         );
     }
-
-    #[test]
-    fn test_and_default_regardless_of_default_conjunctive() {
-        for &default_conjunction in &[false, true] {
-            test_parse_query_to_logical_ast_helper(
-                "title:a AND title:b",
-                "(+Term(field=0,bytes=[97]) +Term(field=0,bytes=[98]))",
-                default_conjunction,
-            );
-        }
-    }
-
-    #[test]
-    fn test_or_default_conjunctive() {
-        for &default_conjunction in &[false, true] {
-            test_parse_query_to_logical_ast_helper(
-                "title:a OR title:b",
-                "(Term(field=0,bytes=[97]) Term(field=0,bytes=[98]))",
-                default_conjunction,
-            );
-        }
-    }
 }

src/query/scorer.rs

@@ -4,8 +4,6 @@ use crate::DocId;
 use crate::Score;
 use downcast_rs::impl_downcast;
 use std::ops::DerefMut;
-use crate::collector::{SegmentCollector, Collector, TopDocs, Count};
-
 
 /// Scored set of documents matching a query within a specific segment.
 ///
@@ -23,12 +21,6 @@ pub trait Scorer: downcast_rs::Downcast + DocSet + 'static {
             callback(self.doc(), self.score());
         }
     }
-
-    fn top_k(&mut self, collector: &mut <(Count, TopDocs) as Collector>::Child) {
-        while self.advance() {
-            collector.collect(self.doc(), self.score());
-        }
-    }
 }
 
 impl_downcast!(Scorer);

src/schema/document.rs

@@ -3,9 +3,9 @@ use crate::common::BinarySerializable;
 use crate::common::VInt;
 use crate::tokenizer::PreTokenizedString;
 use crate::DateTime;
+use itertools::Itertools;
 use serde;
 use std::io::{self, Read, Write};
-use std::mem;
 
 /// Tantivy's Document is the object that can
 /// be indexed and then searched for.
@@ -132,34 +132,12 @@ impl Document {
     pub fn get_sorted_field_values(&self) -> Vec<(Field, Vec<&FieldValue>)> {
         let mut field_values: Vec<&FieldValue> = self.field_values().iter().collect();
         field_values.sort_by_key(|field_value| field_value.field());
-
-        let mut grouped_field_values = vec![];
-
-        let mut current_field;
-        let mut current_group;
-
-        let mut field_values_it = field_values.into_iter();
-        if let Some(field_value) = field_values_it.next() {
-            current_field = field_value.field();
-            current_group = vec![field_value]
-        } else {
-            return grouped_field_values;
-        }
-
-        for field_value in field_values_it {
-            if field_value.field() == current_field {
-                current_group.push(field_value);
-            } else {
-                grouped_field_values.push((
-                    current_field,
-                    mem::replace(&mut current_group, vec![field_value]),
-                ));
-                current_field = field_value.field();
-            }
-        }
-
-        grouped_field_values.push((current_field, current_group));
-        grouped_field_values
+        field_values
+            .into_iter()
+            .group_by(|field_value| field_value.field())
+            .into_iter()
+            .map(|(key, group)| (key, group.collect()))
+            .collect::<Vec<(Field, Vec<&FieldValue>)>>()
     }
 
     /// Returns all of the `FieldValue`s associated the given field

src/schema/text_options.rs

@@ -156,17 +156,30 @@ mod tests {
 
     #[test]
     fn test_field_options() {
-        let field_options = STORED | TEXT;
-        assert!(field_options.is_stored());
-        assert!(field_options.get_indexing_options().is_some());
-        let mut schema_builder = Schema::builder();
-        schema_builder.add_text_field("body", TEXT);
-        let schema = schema_builder.build();
-        let field = schema.get_field("body").unwrap();
-        let field_entry = schema.get_field_entry(field);
-        assert!(matches!(field_entry.field_type(),
-                &FieldType::Str(ref text_options)
-                if text_options.get_indexing_options().unwrap().tokenizer() == "default"));
+        {
+            let field_options = STORED | TEXT;
+            assert!(field_options.is_stored());
+            assert!(field_options.get_indexing_options().is_some());
+        }
+        {
+            let mut schema_builder = Schema::builder();
+            schema_builder.add_text_field("body", TEXT);
+            let schema = schema_builder.build();
+            let field = schema.get_field("body").unwrap();
+            let field_entry = schema.get_field_entry(field);
+            match field_entry.field_type() {
+                &FieldType::Str(ref text_options) => {
+                    assert!(text_options.get_indexing_options().is_some());
+                    assert_eq!(
+                        text_options.get_indexing_options().unwrap().tokenizer(),
+                        "default"
+                    );
+                }
+                _ => {
+                    panic!("");
+                }
+            }
+        }
     }
 
     #[test]

src/store/writer.rs

@@ -3,8 +3,6 @@ use super::skiplist::SkipListBuilder;
 use super::StoreReader;
 use crate::common::CountingWriter;
 use crate::common::{BinarySerializable, VInt};
-use crate::directory::TerminatingWrite;
-use crate::directory::WritePtr;
 use crate::schema::Document;
 use crate::DocId;
 use std::io::{self, Write};
@@ -19,20 +17,20 @@ const BLOCK_SIZE: usize = 16_384;
 ///
 /// The skip list index on the other hand, is built in memory.
 ///
-pub struct StoreWriter {
+pub struct StoreWriter<W: io::Write> {
     doc: DocId,
     offset_index_writer: SkipListBuilder<u64>,
-    writer: CountingWriter<WritePtr>,
+    writer: CountingWriter<W>,
     intermediary_buffer: Vec<u8>,
     current_block: Vec<u8>,
 }
 
-impl StoreWriter {
+impl<W: io::Write> StoreWriter<W> {
     /// Create a store writer.
     ///
     /// The store writer will writes blocks on disc as
     /// document are added.
-    pub fn new(writer: WritePtr) -> StoreWriter {
+    pub fn new(writer: W) -> StoreWriter<W> {
         StoreWriter {
             doc: 0,
             offset_index_writer: SkipListBuilder::new(4),
@@ -102,7 +100,7 @@ impl StoreWriter {
     ///
     /// Compress the last unfinished block if any,
     /// and serializes the skip list index on disc.
-    pub fn close(mut self) -> io::Result<()> {
+    pub fn close(mut self) -> io::Result<W> {
         if !self.current_block.is_empty() {
             self.write_and_compress_block()?;
         }
@@ -110,6 +108,9 @@ impl StoreWriter {
         self.offset_index_writer.write(&mut self.writer)?;
         header_offset.serialize(&mut self.writer)?;
         self.doc.serialize(&mut self.writer)?;
-        self.writer.terminate()
+        self.writer.flush()?;
+        let (wrt, _) = self.writer.finish()?;
+        Ok(wrt)
+
     }
 }

src/termdict/mod.rs

@@ -435,7 +435,6 @@ mod tests {
     #[test]
     fn test_automaton_search() {
         use levenshtein_automata::LevenshteinAutomatonBuilder;
-        use crate::query::DFAWrapper;
 
         const COUNTRIES: [&'static str; 7] = [
             "San Marino",
@@ -464,7 +463,7 @@ mod tests {
 
         // We can now build an entire dfa.
         let lev_automaton_builder = LevenshteinAutomatonBuilder::new(2, true);
-        let automaton = DFAWrapper(lev_automaton_builder.build_dfa("Spaen"));
+        let automaton = lev_automaton_builder.build_dfa("Spaen");
 
         let mut range = term_dict.search(automaton).into_stream();