A new stat for bytes read off the disk

index/scorch/merge.go

@@ -327,6 +327,7 @@ func (s *Scorch) planMergeAtSnapshot(ctx context.Context,
 			fileMergeZapStartTime := time.Now()
 
 			atomic.AddUint64(&s.stats.TotFileMergeZapBeg, 1)
+			prevBytesReadTotal := cumulateBytesRead(segmentsToMerge)
 			newDocNums, _, err := s.segPlugin.Merge(segmentsToMerge, docsToDrop, path,
 				cw.cancelCh, s)
 			atomic.AddUint64(&s.stats.TotFileMergeZapEnd, 1)
@@ -352,6 +353,14 @@ func (s *Scorch) planMergeAtSnapshot(ctx context.Context,
 				atomic.AddUint64(&s.stats.TotFileMergePlanTasksErr, 1)
 				return err
 			}
+
+			switch segI := seg.(type) {
+			case segment.DiskStatsReporter:
+				totalBytesRead := segI.BytesRead() + prevBytesReadTotal
+				segI.SetBytesRead(totalBytesRead)
+				seg = segI.(segment.Segment)
+			}
+
 			oldNewDocNums = make(map[uint64][]uint64)
 			for i, segNewDocNums := range newDocNums {
 				oldNewDocNums[task.Segments[i].Id()] = segNewDocNums
@@ -426,6 +435,16 @@ type segmentMerge struct {
 	notifyCh      chan *mergeTaskIntroStatus
 }
 
+func cumulateBytesRead(sbs []segment.Segment) uint64 {
+	rv := uint64(0)
+	for _, seg := range sbs {
+		if segI, ok := seg.(segment.DiskStatsReporter); ok {
+			rv += segI.BytesRead()
+		}
+	}
+	return rv
+}
+
 // perform a merging of the given SegmentBase instances into a new,
 // persisted segment, and synchronously introduce that new segment
 // into the root

index/scorch/scorch.go

@@ -387,13 +387,23 @@ func (s *Scorch) Batch(batch *index.Batch) (err error) {
 	analysisResults := make([]index.Document, int(numUpdates))
 	var itemsDeQueued uint64
 	var totalAnalysisSize int
+	analysisBytes := func(tokMap index.TokenFrequencies) (rv uint64) {
+		for k := range tokMap {
+			rv += uint64(len(k))
+		}
+		return rv
+	}
 	for itemsDeQueued < numUpdates {
 		result := <-resultChan
 		resultSize := result.Size()
 		atomic.AddUint64(&s.iStats.analysisBytesAdded, uint64(resultSize))
 		totalAnalysisSize += resultSize
 		analysisResults[itemsDeQueued] = result
 		itemsDeQueued++
+		result.VisitFields(func(f index.Field) {
+			atomic.AddUint64(&s.stats.TotIndexedAnalysisBytes,
+				analysisBytes(f.AnalyzedTokenFrequencies()))
+		})
 	}
 	close(resultChan)
 	defer atomic.AddUint64(&s.iStats.analysisBytesRemoved, uint64(totalAnalysisSize))
@@ -525,6 +535,10 @@ func (s *Scorch) Stats() json.Marshaler {
 	return &s.stats
 }
 
+func (s *Scorch) BytesReadQueryTime() uint64 {
+	return s.stats.TotBytesReadQueryTime
+}
+
 func (s *Scorch) diskFileStats(rootSegmentPaths map[string]struct{}) (uint64,
 	uint64, uint64) {
 	var numFilesOnDisk, numBytesUsedDisk, numBytesOnDiskByRoot uint64
@@ -582,7 +596,9 @@ func (s *Scorch) StatsMap() map[string]interface{} {
 	m["index_time"] = m["TotIndexTime"]
 	m["term_searchers_started"] = m["TotTermSearchersStarted"]
 	m["term_searchers_finished"] = m["TotTermSearchersFinished"]
+	m["num_bytes_read_query_time"] = m["TotBytesReadQueryTime"]
 	m["num_plain_text_bytes_indexed"] = m["TotIndexedPlainTextBytes"]
+	m["num_bytes_indexed_after_analysis"] = m["TotIndexedAnalysisBytes"]
 	m["num_items_introduced"] = m["TotIntroducedItems"]
 	m["num_items_persisted"] = m["TotPersistedItems"]
 	m["num_recs_to_persist"] = m["TotItemsToPersist"]

index/scorch/snapshot_index.go

@@ -60,6 +60,8 @@ var reflectStaticSizeIndexSnapshot int
 // in the kvConfig.
 var DefaultFieldTFRCacheThreshold uint64 = 10
 
+type diskStatsReporter segment.DiskStatsReporter
+
 func init() {
 	var is interface{} = IndexSnapshot{}
 	reflectStaticSizeIndexSnapshot = int(reflect.TypeOf(is).Size())
@@ -146,10 +148,18 @@ func (i *IndexSnapshot) newIndexSnapshotFieldDict(field string,
 	results := make(chan *asynchSegmentResult)
 	for index, segment := range i.segment {
 		go func(index int, segment *SegmentSnapshot) {
+			var prevBytesRead uint64
+			if seg, ok := segment.segment.(diskStatsReporter); ok {
+				prevBytesRead = seg.BytesRead()
+			}
 			dict, err := segment.segment.Dictionary(field)
 			if err != nil {
 				results <- &asynchSegmentResult{err: err}
 			} else {
+				if seg, ok := segment.segment.(diskStatsReporter); ok {
+					atomic.AddUint64(&i.parent.stats.TotBytesReadQueryTime,
+						seg.BytesRead()-prevBytesRead)
+				}
 				if randomLookup {
 					results <- &asynchSegmentResult{dict: dict}
 				} else {
@@ -424,6 +434,11 @@ func (i *IndexSnapshot) Document(id string) (rv index.Document, err error) {
 	segmentIndex, localDocNum := i.segmentIndexAndLocalDocNumFromGlobal(docNum)
 
 	rvd := document.NewDocument(id)
+	var prevBytesRead uint64
+	seg, ok := i.segment[segmentIndex].segment.(segment.DiskStatsReporter)
+	if ok {
+		prevBytesRead = seg.BytesRead()
+	}
 	err = i.segment[segmentIndex].VisitDocument(localDocNum, func(name string, typ byte, val []byte, pos []uint64) bool {
 		if name == "_id" {
 			return true
@@ -453,7 +468,10 @@ func (i *IndexSnapshot) Document(id string) (rv index.Document, err error) {
 	if err != nil {
 		return nil, err
 	}
-
+	if ok {
+		delta := seg.BytesRead() - prevBytesRead
+		atomic.AddUint64(&i.parent.stats.TotBytesReadQueryTime, delta)
+	}
 	return rvd, nil
 }
 
@@ -505,18 +523,18 @@ func (i *IndexSnapshot) InternalID(id string) (rv index.IndexInternalID, err err
 	return next.ID, nil
 }
 
-func (i *IndexSnapshot) TermFieldReader(term []byte, field string, includeFreq,
+func (is *IndexSnapshot) TermFieldReader(term []byte, field string, includeFreq,
 	includeNorm, includeTermVectors bool) (index.TermFieldReader, error) {
-	rv := i.allocTermFieldReaderDicts(field)
+	rv := is.allocTermFieldReaderDicts(field)
 
 	rv.term = term
 	rv.field = field
-	rv.snapshot = i
+	rv.snapshot = is
 	if rv.postings == nil {
-		rv.postings = make([]segment.PostingsList, len(i.segment))
+		rv.postings = make([]segment.PostingsList, len(is.segment))
 	}
 	if rv.iterators == nil {
-		rv.iterators = make([]segment.PostingsIterator, len(i.segment))
+		rv.iterators = make([]segment.PostingsIterator, len(is.segment))
 	}
 	rv.segmentOffset = 0
 	rv.includeFreq = includeFreq
@@ -526,25 +544,54 @@ func (i *IndexSnapshot) TermFieldReader(term []byte, field string, includeFreq,
 	rv.currID = rv.currID[:0]
 
 	if rv.dicts == nil {
-		rv.dicts = make([]segment.TermDictionary, len(i.segment))
-		for i, segment := range i.segment {
+		rv.dicts = make([]segment.TermDictionary, len(is.segment))
+		for i, segment := range is.segment {
+			var prevBytesRead uint64
+			segP, ok := segment.segment.(diskStatsReporter)
+			if ok {
+				prevBytesRead = segP.BytesRead()
+			}
 			dict, err := segment.segment.Dictionary(field)
 			if err != nil {
 				return nil, err
 			}
+			if ok {
+				atomic.AddUint64(&is.parent.stats.TotBytesReadQueryTime, segP.BytesRead()-prevBytesRead)
+			}
 			rv.dicts[i] = dict
 		}
 	}
 
-	for i, segment := range i.segment {
+	for i, segment := range is.segment {
+		var prevBytesReadPL uint64
+		if postings, ok := rv.postings[i].(diskStatsReporter); ok {
+			prevBytesReadPL = postings.BytesRead()
+		}
 		pl, err := rv.dicts[i].PostingsList(term, segment.deleted, rv.postings[i])
 		if err != nil {
 			return nil, err
 		}
 		rv.postings[i] = pl
+
+		var prevBytesReadItr uint64
+		if itr, ok := rv.iterators[i].(diskStatsReporter); ok {
+			prevBytesReadItr = itr.BytesRead()
+		}
 		rv.iterators[i] = pl.Iterator(includeFreq, includeNorm, includeTermVectors, rv.iterators[i])
+
+		if postings, ok := pl.(diskStatsReporter); ok &&
+			prevBytesReadPL < postings.BytesRead() {
+			atomic.AddUint64(&is.parent.stats.TotBytesReadQueryTime,
+				postings.BytesRead()-prevBytesReadPL)
+		}
+
+		if itr, ok := rv.iterators[i].(diskStatsReporter); ok &&
+			prevBytesReadItr < itr.BytesRead() {
+			atomic.AddUint64(&is.parent.stats.TotBytesReadQueryTime,
+				itr.BytesRead()-prevBytesReadItr)
+		}
 	}
-	atomic.AddUint64(&i.parent.stats.TotTermSearchersStarted, uint64(1))
+	atomic.AddUint64(&is.parent.stats.TotTermSearchersStarted, uint64(1))
 	return rv, nil
 }
 
@@ -661,10 +708,18 @@ func (i *IndexSnapshot) documentVisitFieldTermsOnSegment(
 	}
 
 	if ssvOk && ssv != nil && len(vFields) > 0 {
+		var prevBytesRead uint64
+		ssvp, ok := ssv.(segment.DiskStatsReporter)
+		if ok {
+			prevBytesRead = ssvp.BytesRead()
+		}
 		dvs, err = ssv.VisitDocValues(localDocNum, fields, visitor, dvs)
 		if err != nil {
 			return nil, nil, err
 		}
+		if ok {
+			atomic.AddUint64(&i.parent.stats.TotBytesReadQueryTime, ssvp.BytesRead()-prevBytesRead)
+		}
 	}
 
 	if errCh != nil {

index/scorch/snapshot_index_tfr.go

@@ -76,6 +76,11 @@ func (i *IndexSnapshotTermFieldReader) Next(preAlloced *index.TermFieldDoc) (*in
 	}
 	// find the next hit
 	for i.segmentOffset < len(i.iterators) {
+		prevBytesRead := uint64(0)
+		itr, ok := i.iterators[i.segmentOffset].(segment.DiskStatsReporter)
+		if ok {
+			prevBytesRead = itr.BytesRead()
+		}
 		next, err := i.iterators[i.segmentOffset].Next()
 		if err != nil {
 			return nil, err
@@ -89,6 +94,15 @@ func (i *IndexSnapshotTermFieldReader) Next(preAlloced *index.TermFieldDoc) (*in
 
 			i.currID = rv.ID
 			i.currPosting = next
+			// postingsIterators is maintain the bytesRead stat in a cumulative fashion.
+			// this is because there are chances of having a series of loadChunk calls,
+			// and they have to be added together before sending the bytesRead at this point
+			// upstream.
+			if ok {
+				delta := itr.BytesRead() - prevBytesRead
+				atomic.AddUint64(&i.snapshot.parent.stats.TotBytesReadQueryTime, uint64(delta))
+			}
+
 			return rv, nil
 		}
 		i.segmentOffset++

index/scorch/stats.go

@@ -42,7 +42,9 @@ type Stats struct {
 	TotAnalysisTime uint64
 	TotIndexTime    uint64
 
+	TotBytesReadQueryTime    uint64
 	TotIndexedPlainTextBytes uint64
+	TotIndexedAnalysisBytes  uint64
 
 	TotTermSearchersStarted  uint64
 	TotTermSearchersFinished uint64