gibheer/bleve
0
0
Fork 0
Code

firestorm.Batch() collects [][]IndexRows instead of []IndexRow

Browse Source 
Rather than append() all received rows into a flat []IndexRow during
the result gathering loop, this change instead collects the analysis
result rows into a [][]IndexRow, which avoids extra copying.

As part of this, firestorm batchRows() now takes the [][]IndexRow as
its input.
This commit is contained in:
Steve Yen 2016-01-01 22:25:34 -08:00
parent 1c5b84911d
commit fb8c9a7475
1 changed files with 46 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

83
index/firestorm/firestorm.go
View File

@ -167,7 +167,7 @@ func (f *Firestorm) Update(doc *document.Document) (err error) {
}() }()
var dictionaryDeltas map[string]int64 var dictionaryDeltas map[string]int64
dictionaryDeltas, err = f.batchRows(kvwriter, result.Rows, nil) dictionaryDeltas, err = f.batchRows(kvwriter, [][]index.IndexRow{result.Rows}, nil)
if err != nil { if err != nil {
_ = kvwriter.Close() _ = kvwriter.Close()
atomic.AddUint64(&f.stats.errors, 1) atomic.AddUint64(&f.stats.errors, 1)
@ -190,7 +190,7 @@ func (f *Firestorm) Delete(id string) error {
return nil return nil
} }
func (f *Firestorm) batchRows(writer store.KVWriter, rows []index.IndexRow, deleteKeys [][]byte) (map[string]int64, error) { func (f *Firestorm) batchRows(writer store.KVWriter, rowsOfRows [][]index.IndexRow, deleteKeys [][]byte) (map[string]int64, error) {
// prepare batch // prepare batch
wb := writer.NewBatch() wb := writer.NewBatch()
@ -206,33 +206,36 @@ func (f *Firestorm) batchRows(writer store.KVWriter, rows []index.IndexRow, dele
} }
dictionaryDeltas := make(map[string]int64) dictionaryDeltas := make(map[string]int64)
for _, row := range rows {
tfr, ok := row.(*TermFreqRow) for _, rows := range rowsOfRows {
if ok { for _, row := range rows {
if tfr.Field() != 0 { tfr, ok := row.(*TermFreqRow)
kbuf = prepareBuf(kbuf, tfr.DictionaryRowKeySize()) if ok {
klen, err := tfr.DictionaryRowKeyTo(kbuf) if tfr.Field() != 0 {
if err != nil { kbuf = prepareBuf(kbuf, tfr.DictionaryRowKeySize())
return nil, err klen, err := tfr.DictionaryRowKeyTo(kbuf)
if err != nil {
return nil, err
}
dictionaryDeltas[string(kbuf[0:klen])] += 1
} }
dictionaryDeltas[string(kbuf[0:klen])] += 1
} }
}
kbuf = prepareBuf(kbuf, row.KeySize()) kbuf = prepareBuf(kbuf, row.KeySize())
klen, err := row.KeyTo(kbuf) klen, err := row.KeyTo(kbuf)
if err != nil { if err != nil {
return nil, err return nil, err
} }
vbuf = prepareBuf(vbuf, row.ValueSize()) vbuf = prepareBuf(vbuf, row.ValueSize())
vlen, err := row.ValueTo(vbuf) vlen, err := row.ValueTo(vbuf)
if err != nil { if err != nil {
return nil, err return nil, err
} }
wb.Set(kbuf[0:klen], vbuf[0:vlen]) wb.Set(kbuf[0:klen], vbuf[0:vlen])
}
} }
for _, dk := range deleteKeys { for _, dk := range deleteKeys {
@ -291,12 +294,14 @@ func (f *Firestorm) Batch(batch *index.Batch) (err error) {
} }
}() }()
allRows := make([]index.IndexRow, 0, 1000) // extra 1 capacity for internal updates.
collectRows := make([][]index.IndexRow, 0, docsUpdated+1)
// wait for the result // wait for the result
var itemsDeQueued uint64 var itemsDeQueued uint64
for itemsDeQueued < docsUpdated { for itemsDeQueued < docsUpdated {
result := <-resultChan result := <-resultChan
allRows = append(allRows, result.Rows...) collectRows = append(collectRows, result.Rows)
itemsDeQueued++ itemsDeQueued++
} }
close(resultChan) close(resultChan)
@ -309,17 +314,21 @@ func (f *Firestorm) Batch(batch *index.Batch) (err error) {
atomic.AddUint64(&f.stats.analysisTime, uint64(time.Since(analysisStart))) atomic.AddUint64(&f.stats.analysisTime, uint64(time.Since(analysisStart)))
deleteKeys := make([][]byte, 0) var deleteKeys [][]byte
// add the internal ops if len(batch.InternalOps) > 0 {
for internalKey, internalValue := range batch.InternalOps { // add the internal ops
if internalValue == nil { updateInternalRows := make([]index.IndexRow, 0, len(batch.InternalOps))
// delete for internalKey, internalValue := range batch.InternalOps {
deleteInternalRow := NewInternalRow([]byte(internalKey), nil) if internalValue == nil {
deleteKeys = append(deleteKeys, deleteInternalRow.Key()) // delete
} else { deleteInternalRow := NewInternalRow([]byte(internalKey), nil)
updateInternalRow := NewInternalRow([]byte(internalKey), internalValue) deleteKeys = append(deleteKeys, deleteInternalRow.Key())
allRows = append(allRows, updateInternalRow) } else {
updateInternalRow := NewInternalRow([]byte(internalKey), internalValue)
updateInternalRows = append(updateInternalRows, updateInternalRow)
}
} }
collectRows = append(collectRows, updateInternalRows)
} }
inflightItems := make([]*InFlightItem, 0, len(batch.IndexOps)) inflightItems := make([]*InFlightItem, 0, len(batch.IndexOps))
@ -342,7 +351,7 @@ func (f *Firestorm) Batch(batch *index.Batch) (err error) {
} }
var dictionaryDeltas map[string]int64 var dictionaryDeltas map[string]int64
dictionaryDeltas, err = f.batchRows(kvwriter, allRows, deleteKeys) dictionaryDeltas, err = f.batchRows(kvwriter, collectRows, deleteKeys)
if err != nil { if err != nil {
_ = kvwriter.Close() _ = kvwriter.Close()
atomic.AddUint64(&f.stats.errors, 1) atomic.AddUint64(&f.stats.errors, 1)