Add --tail=NUM to limit the number of items to keep in memory

src/matcher.go

@@ -16,7 +16,7 @@ type MatchRequest struct {
 	pattern  *Pattern
 	final    bool
 	sort     bool
-	revision int
+	revision revision
 }
 
 // Matcher is responsible for performing search
@@ -30,7 +30,7 @@ type Matcher struct {
 	partitions     int
 	slab           []*util.Slab
 	mergerCache    map[string]*Merger
-	revision       int
+	revision       revision
 }
 
 const (
@@ -40,7 +40,7 @@ const (
 
 // NewMatcher returns a new Matcher
 func NewMatcher(cache *ChunkCache, patternBuilder func([]rune) *Pattern,
-	sort bool, tac bool, eventBox *util.EventBox, revision int) *Matcher {
+	sort bool, tac bool, eventBox *util.EventBox, revision revision) *Matcher {
 	partitions := util.Min(numPartitionsMultiplier*runtime.NumCPU(), maxPartitions)
 	return &Matcher{
 		cache:          cache,
@@ -82,11 +82,13 @@ func (m *Matcher) Loop() {
 			break
 		}
 
+		cacheCleared := false
 		if request.sort != m.sort || request.revision != m.revision {
 			m.sort = request.sort
 			m.revision = request.revision
 			m.mergerCache = make(map[string]*Merger)
 			m.cache.Clear()
+			cacheCleared = true
 		}
 
 		// Restart search
@@ -95,20 +97,20 @@ func (m *Matcher) Loop() {
 		cancelled := false
 		count := CountItems(request.chunks)
 
-		foundCache := false
-		if count == prevCount {
-			// Look up mergerCache
-			if cached, found := m.mergerCache[patternString]; found {
-				foundCache = true
-				merger = cached
+		if !cacheCleared {
+			if count == prevCount {
+				// Look up mergerCache
+				if cached, found := m.mergerCache[patternString]; found {
+					merger = cached
+				}
+			} else {
+				// Invalidate mergerCache
+				prevCount = count
+				m.mergerCache = make(map[string]*Merger)
 			}
-		} else {
-			// Invalidate mergerCache
-			prevCount = count
-			m.mergerCache = make(map[string]*Merger)
 		}
 
-		if !foundCache {
+		if merger == nil {
 			merger, cancelled = m.scan(request)
 		}
 
@@ -160,6 +162,7 @@ func (m *Matcher) scan(request MatchRequest) (*Merger, bool) {
 		return PassMerger(&request.chunks, m.tac, request.revision), false
 	}
 
+	minIndex := request.chunks[0].items[0].Index()
 	cancelled := util.NewAtomicBool(false)
 
 	slices := m.sliceChunks(request.chunks)
@@ -231,11 +234,11 @@ func (m *Matcher) scan(request MatchRequest) (*Merger, bool) {
 		partialResult := <-resultChan
 		partialResults[partialResult.index] = partialResult.matches
 	}
-	return NewMerger(pattern, partialResults, m.sort, m.tac, request.revision), false
+	return NewMerger(pattern, partialResults, m.sort, m.tac, request.revision, minIndex), false
 }
 
 // Reset is called to interrupt/signal the ongoing search
-func (m *Matcher) Reset(chunks []*Chunk, patternRunes []rune, cancel bool, final bool, sort bool, revision int) {
+func (m *Matcher) Reset(chunks []*Chunk, patternRunes []rune, cancel bool, final bool, sort bool, revision revision) {
 	pattern := m.patternBuilder(patternRunes)
 
 	var event util.EventType