2014-04-17 22:55:53 +02:00
|
|
|
// Copyright (c) 2014 Couchbase, Inc.
|
|
|
|
// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file
|
|
|
|
// except in compliance with the License. You may obtain a copy of the License at
|
|
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
// Unless required by applicable law or agreed to in writing, software distributed under the
|
|
|
|
// License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
|
|
|
|
// either express or implied. See the License for the specific language governing permissions
|
|
|
|
// and limitations under the License.
|
2014-09-02 16:54:50 +02:00
|
|
|
|
2014-04-17 22:55:53 +02:00
|
|
|
package upside_down
|
|
|
|
|
|
|
|
import (
|
|
|
|
"bytes"
|
2014-10-02 20:12:22 +02:00
|
|
|
"encoding/json"
|
2015-01-16 20:07:47 +01:00
|
|
|
"fmt"
|
2014-04-17 22:55:53 +02:00
|
|
|
"math"
|
2015-05-08 14:07:20 +02:00
|
|
|
"sync"
|
2014-10-02 20:12:22 +02:00
|
|
|
"sync/atomic"
|
|
|
|
"time"
|
2014-04-17 22:55:53 +02:00
|
|
|
|
2014-08-28 21:38:57 +02:00
|
|
|
"github.com/blevesearch/bleve/analysis"
|
|
|
|
"github.com/blevesearch/bleve/document"
|
|
|
|
"github.com/blevesearch/bleve/index"
|
|
|
|
"github.com/blevesearch/bleve/index/store"
|
2015-09-02 19:12:08 +02:00
|
|
|
"github.com/blevesearch/bleve/registry"
|
2014-08-19 14:58:26 +02:00
|
|
|
|
2014-12-09 23:24:59 +01:00
|
|
|
"github.com/golang/protobuf/proto"
|
2014-04-17 22:55:53 +02:00
|
|
|
)
|
|
|
|
|
2015-09-02 19:12:08 +02:00
|
|
|
const Name = "upside_down"
|
|
|
|
|
2014-09-04 00:17:26 +02:00
|
|
|
var VersionKey = []byte{'v'}
|
2014-04-17 22:55:53 +02:00
|
|
|
|
2015-05-15 21:04:52 +02:00
|
|
|
var UnsafeBatchUseDetected = fmt.Errorf("bleve.Batch is NOT thread-safe, modification after execution detected")
|
|
|
|
|
2015-05-18 09:16:35 +02:00
|
|
|
const Version uint8 = 5
|
2015-02-17 12:45:53 +01:00
|
|
|
|
|
|
|
var IncompatibleVersion = fmt.Errorf("incompatible version, %d is supported", Version)
|
2014-04-17 22:55:53 +02:00
|
|
|
|
|
|
|
type UpsideDownCouch struct {
|
2015-09-02 19:12:08 +02:00
|
|
|
version uint8
|
|
|
|
path string
|
|
|
|
store store.KVStore
|
|
|
|
fieldCache *index.FieldCache
|
|
|
|
analysisQueue *index.AnalysisQueue
|
|
|
|
stats *indexStat
|
2015-05-08 14:07:20 +02:00
|
|
|
|
|
|
|
m sync.RWMutex
|
|
|
|
// fields protected by m
|
|
|
|
docCount uint64
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
|
|
|
|
2015-09-02 19:12:08 +02:00
|
|
|
func NewUpsideDownCouch(s store.KVStore, analysisQueue *index.AnalysisQueue) *UpsideDownCouch {
|
2014-04-17 22:55:53 +02:00
|
|
|
return &UpsideDownCouch{
|
2015-09-02 19:12:08 +02:00
|
|
|
version: Version,
|
|
|
|
fieldCache: index.NewFieldCache(),
|
|
|
|
store: s,
|
|
|
|
analysisQueue: analysisQueue,
|
|
|
|
stats: &indexStat{},
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2014-09-12 23:21:35 +02:00
|
|
|
func (udc *UpsideDownCouch) init(kvwriter store.KVWriter) (err error) {
|
2014-04-17 22:55:53 +02:00
|
|
|
// prepare a list of rows
|
|
|
|
rows := make([]UpsideDownCouchRow, 0)
|
|
|
|
|
|
|
|
// version marker
|
|
|
|
rows = append(rows, NewVersionRow(udc.version))
|
|
|
|
|
2014-09-12 23:21:35 +02:00
|
|
|
return udc.batchRows(kvwriter, nil, rows, nil)
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
|
|
|
|
2014-09-12 23:21:35 +02:00
|
|
|
func (udc *UpsideDownCouch) loadSchema(kvreader store.KVReader) (err error) {
|
2014-04-17 22:55:53 +02:00
|
|
|
|
2014-05-09 22:37:04 +02:00
|
|
|
keyPrefix := []byte{'f'}
|
2014-09-12 23:21:35 +02:00
|
|
|
it := kvreader.Iterator(keyPrefix)
|
2015-04-07 19:26:54 +02:00
|
|
|
defer func() {
|
|
|
|
if cerr := it.Close(); err == nil && cerr != nil {
|
|
|
|
err = cerr
|
|
|
|
}
|
|
|
|
}()
|
2014-04-17 22:55:53 +02:00
|
|
|
|
|
|
|
it.Seek(keyPrefix)
|
2014-05-09 22:37:04 +02:00
|
|
|
key, val, valid := it.Current()
|
|
|
|
for valid {
|
|
|
|
|
2014-04-17 22:55:53 +02:00
|
|
|
// stop when
|
2014-05-09 22:37:04 +02:00
|
|
|
if !bytes.HasPrefix(key, keyPrefix) {
|
2014-04-17 22:55:53 +02:00
|
|
|
break
|
|
|
|
}
|
2015-04-07 19:26:54 +02:00
|
|
|
var fieldRow *FieldRow
|
|
|
|
fieldRow, err = NewFieldRowKV(key, val)
|
2014-04-19 03:07:41 +02:00
|
|
|
if err != nil {
|
2015-04-07 19:26:54 +02:00
|
|
|
return
|
2014-04-19 03:07:41 +02:00
|
|
|
}
|
2015-09-02 19:12:08 +02:00
|
|
|
udc.fieldCache.AddExisting(fieldRow.name, fieldRow.index)
|
2014-05-09 22:37:04 +02:00
|
|
|
|
|
|
|
it.Next()
|
|
|
|
key, val, valid = it.Current()
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
|
|
|
|
2015-02-17 12:45:53 +01:00
|
|
|
keyPrefix = []byte{'v'}
|
|
|
|
val, err = kvreader.Get(keyPrefix)
|
|
|
|
if err != nil {
|
2015-04-07 19:26:54 +02:00
|
|
|
return
|
2015-02-17 12:45:53 +01:00
|
|
|
}
|
2015-04-07 19:26:54 +02:00
|
|
|
var vr *VersionRow
|
|
|
|
vr, err = NewVersionRowKV(keyPrefix, val)
|
2015-02-17 12:45:53 +01:00
|
|
|
if err != nil {
|
2015-04-07 19:26:54 +02:00
|
|
|
return
|
2015-02-17 12:45:53 +01:00
|
|
|
}
|
|
|
|
if vr.version != Version {
|
2015-04-07 19:26:54 +02:00
|
|
|
err = IncompatibleVersion
|
|
|
|
return
|
2015-02-17 12:45:53 +01:00
|
|
|
}
|
|
|
|
|
2014-04-17 22:55:53 +02:00
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2014-09-12 23:21:35 +02:00
|
|
|
func (udc *UpsideDownCouch) batchRows(writer store.KVWriter, addRows []UpsideDownCouchRow, updateRows []UpsideDownCouchRow, deleteRows []UpsideDownCouchRow) (err error) {
|
2014-08-11 22:27:18 +02:00
|
|
|
|
2014-04-17 22:55:53 +02:00
|
|
|
// prepare batch
|
2014-09-12 23:21:35 +02:00
|
|
|
wb := writer.NewBatch()
|
2014-04-17 22:55:53 +02:00
|
|
|
|
|
|
|
// add
|
|
|
|
for _, row := range addRows {
|
|
|
|
tfr, ok := row.(*TermFrequencyRow)
|
|
|
|
if ok {
|
|
|
|
// need to increment counter
|
2015-03-10 21:22:19 +01:00
|
|
|
dictionaryKey := tfr.DictionaryRowKey()
|
2015-04-24 23:11:47 +02:00
|
|
|
wb.Merge(dictionaryKey, dictionaryTermIncr)
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
2014-05-09 22:37:04 +02:00
|
|
|
wb.Set(row.Key(), row.Value())
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
// update
|
|
|
|
for _, row := range updateRows {
|
2014-05-09 22:37:04 +02:00
|
|
|
wb.Set(row.Key(), row.Value())
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
// delete
|
|
|
|
for _, row := range deleteRows {
|
|
|
|
tfr, ok := row.(*TermFrequencyRow)
|
|
|
|
if ok {
|
|
|
|
// need to decrement counter
|
2015-03-10 21:22:19 +01:00
|
|
|
dictionaryKey := tfr.DictionaryRowKey()
|
2015-04-24 23:11:47 +02:00
|
|
|
wb.Merge(dictionaryKey, dictionaryTermDecr)
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
2014-04-18 22:09:34 +02:00
|
|
|
wb.Delete(row.Key())
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
// write out the batch
|
2015-10-09 09:57:12 +02:00
|
|
|
return wb.Execute()
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
|
|
|
|
2014-10-31 14:40:23 +01:00
|
|
|
func (udc *UpsideDownCouch) DocCount() (uint64, error) {
|
2015-05-08 14:07:20 +02:00
|
|
|
udc.m.RLock()
|
|
|
|
defer udc.m.RUnlock()
|
2014-10-31 14:40:23 +01:00
|
|
|
return udc.docCount, nil
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
|
|
|
|
2015-04-07 19:26:54 +02:00
|
|
|
func (udc *UpsideDownCouch) Open() (err error) {
|
2015-04-24 23:11:47 +02:00
|
|
|
// install the merge operator
|
|
|
|
udc.store.SetMergeOperator(&mergeOperator)
|
|
|
|
|
|
|
|
// now open the kv store
|
|
|
|
err = udc.store.Open()
|
|
|
|
if err != nil {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2014-09-12 23:21:35 +02:00
|
|
|
// start a writer for the open process
|
2015-04-07 19:26:54 +02:00
|
|
|
var kvwriter store.KVWriter
|
|
|
|
kvwriter, err = udc.store.Writer()
|
2014-10-31 14:40:23 +01:00
|
|
|
if err != nil {
|
2015-04-07 19:26:54 +02:00
|
|
|
return
|
2014-10-31 14:40:23 +01:00
|
|
|
}
|
2015-04-07 19:26:54 +02:00
|
|
|
defer func() {
|
|
|
|
if cerr := kvwriter.Close(); err == nil && cerr != nil {
|
|
|
|
err = cerr
|
|
|
|
}
|
|
|
|
}()
|
2014-09-12 23:21:35 +02:00
|
|
|
|
2015-04-07 19:26:54 +02:00
|
|
|
var value []byte
|
|
|
|
value, err = kvwriter.Get(VersionKey)
|
2014-04-17 22:55:53 +02:00
|
|
|
if err != nil {
|
2015-04-07 19:26:54 +02:00
|
|
|
return
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
// init new index OR load schema
|
|
|
|
if value == nil {
|
2014-09-12 23:21:35 +02:00
|
|
|
err = udc.init(kvwriter)
|
2014-04-17 22:55:53 +02:00
|
|
|
if err != nil {
|
2015-04-07 19:26:54 +02:00
|
|
|
return
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
|
|
|
} else {
|
2014-09-12 23:21:35 +02:00
|
|
|
err = udc.loadSchema(kvwriter)
|
2014-04-17 22:55:53 +02:00
|
|
|
if err != nil {
|
2015-04-07 19:26:54 +02:00
|
|
|
return
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
// set doc count
|
2015-05-08 14:07:20 +02:00
|
|
|
udc.m.Lock()
|
2015-04-07 19:26:54 +02:00
|
|
|
udc.docCount, err = udc.countDocs(kvwriter)
|
2015-05-08 14:07:20 +02:00
|
|
|
udc.m.Unlock()
|
2015-04-07 19:26:54 +02:00
|
|
|
return
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
|
|
|
|
2015-04-07 19:26:54 +02:00
|
|
|
func (udc *UpsideDownCouch) countDocs(kvreader store.KVReader) (count uint64, err error) {
|
2014-09-12 23:21:35 +02:00
|
|
|
it := kvreader.Iterator([]byte{'b'})
|
2015-04-07 19:26:54 +02:00
|
|
|
defer func() {
|
|
|
|
if cerr := it.Close(); err == nil && cerr != nil {
|
|
|
|
err = cerr
|
|
|
|
}
|
|
|
|
}()
|
2014-04-17 22:55:53 +02:00
|
|
|
|
2014-05-09 22:37:04 +02:00
|
|
|
key, _, valid := it.Current()
|
|
|
|
for valid {
|
|
|
|
if !bytes.HasPrefix(key, []byte{'b'}) {
|
2014-04-17 22:55:53 +02:00
|
|
|
break
|
|
|
|
}
|
2015-04-07 19:26:54 +02:00
|
|
|
count++
|
2014-05-09 22:37:04 +02:00
|
|
|
it.Next()
|
|
|
|
key, _, valid = it.Current()
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
2014-05-09 22:37:04 +02:00
|
|
|
|
2015-04-07 19:26:54 +02:00
|
|
|
return
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
|
|
|
|
2015-04-07 19:26:54 +02:00
|
|
|
func (udc *UpsideDownCouch) rowCount() (count uint64, err error) {
|
2014-09-12 23:21:35 +02:00
|
|
|
// start an isolated reader for use during the rowcount
|
2014-10-31 14:40:23 +01:00
|
|
|
kvreader, err := udc.store.Reader()
|
|
|
|
if err != nil {
|
2015-04-07 19:26:54 +02:00
|
|
|
return
|
2014-10-31 14:40:23 +01:00
|
|
|
}
|
2015-04-07 19:26:54 +02:00
|
|
|
defer func() {
|
|
|
|
if cerr := kvreader.Close(); err == nil && cerr != nil {
|
|
|
|
err = cerr
|
|
|
|
}
|
|
|
|
}()
|
2014-09-12 23:21:35 +02:00
|
|
|
it := kvreader.Iterator([]byte{0})
|
2015-04-07 19:26:54 +02:00
|
|
|
defer func() {
|
|
|
|
if cerr := it.Close(); err == nil && cerr != nil {
|
|
|
|
err = cerr
|
|
|
|
}
|
|
|
|
}()
|
2014-04-17 22:55:53 +02:00
|
|
|
|
2014-05-09 22:37:04 +02:00
|
|
|
_, _, valid := it.Current()
|
|
|
|
for valid {
|
2015-04-07 19:26:54 +02:00
|
|
|
count++
|
2014-05-09 22:37:04 +02:00
|
|
|
it.Next()
|
|
|
|
_, _, valid = it.Current()
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
2014-05-09 22:37:04 +02:00
|
|
|
|
2015-04-07 19:26:54 +02:00
|
|
|
return
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
|
|
|
|
2014-10-31 14:40:23 +01:00
|
|
|
func (udc *UpsideDownCouch) Close() error {
|
|
|
|
return udc.store.Close()
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
|
|
|
|
2015-04-07 19:26:54 +02:00
|
|
|
func (udc *UpsideDownCouch) Update(doc *document.Document) (err error) {
|
2014-09-24 14:13:14 +02:00
|
|
|
// do analysis before acquiring write lock
|
2014-10-02 20:12:22 +02:00
|
|
|
analysisStart := time.Now()
|
2015-09-02 19:12:08 +02:00
|
|
|
resultChan := make(chan *index.AnalysisResult)
|
|
|
|
aw := index.NewAnalysisWork(udc, doc, resultChan)
|
2014-09-24 14:13:14 +02:00
|
|
|
// put the work on the queue
|
2015-10-12 10:42:14 +02:00
|
|
|
udc.analysisQueue.Queue(aw)
|
2014-09-24 14:13:14 +02:00
|
|
|
|
|
|
|
// wait for the result
|
|
|
|
result := <-resultChan
|
|
|
|
close(resultChan)
|
2014-10-02 20:12:22 +02:00
|
|
|
atomic.AddUint64(&udc.stats.analysisTime, uint64(time.Since(analysisStart)))
|
2014-09-24 14:13:14 +02:00
|
|
|
|
2014-09-12 23:21:35 +02:00
|
|
|
// start a writer for this update
|
2014-10-02 20:12:22 +02:00
|
|
|
indexStart := time.Now()
|
2015-04-07 19:26:54 +02:00
|
|
|
var kvwriter store.KVWriter
|
|
|
|
kvwriter, err = udc.store.Writer()
|
2014-10-31 14:40:23 +01:00
|
|
|
if err != nil {
|
2015-04-07 19:26:54 +02:00
|
|
|
return
|
2014-10-31 14:40:23 +01:00
|
|
|
}
|
2015-04-07 19:26:54 +02:00
|
|
|
defer func() {
|
|
|
|
if cerr := kvwriter.Close(); err == nil && cerr != nil {
|
|
|
|
err = cerr
|
|
|
|
}
|
|
|
|
}()
|
2014-09-12 23:21:35 +02:00
|
|
|
|
2014-04-17 22:55:53 +02:00
|
|
|
// first we lookup the backindex row for the doc id if it exists
|
|
|
|
// lookup the back index row
|
2015-04-07 19:26:54 +02:00
|
|
|
var backIndexRow *BackIndexRow
|
|
|
|
backIndexRow, err = udc.backIndexRowForDoc(kvwriter, doc.ID)
|
2014-04-17 22:55:53 +02:00
|
|
|
if err != nil {
|
2014-10-02 20:12:22 +02:00
|
|
|
atomic.AddUint64(&udc.stats.errors, 1)
|
2015-04-07 19:26:54 +02:00
|
|
|
return
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
|
|
|
|
2014-08-11 22:27:18 +02:00
|
|
|
// prepare a list of rows
|
|
|
|
addRows := make([]UpsideDownCouchRow, 0)
|
|
|
|
updateRows := make([]UpsideDownCouchRow, 0)
|
|
|
|
deleteRows := make([]UpsideDownCouchRow, 0)
|
|
|
|
|
2015-09-02 19:12:08 +02:00
|
|
|
addRows, updateRows, deleteRows = udc.mergeOldAndNew(backIndexRow, result.Rows, addRows, updateRows, deleteRows)
|
2014-08-11 22:27:18 +02:00
|
|
|
|
2014-09-12 23:21:35 +02:00
|
|
|
err = udc.batchRows(kvwriter, addRows, updateRows, deleteRows)
|
2014-08-11 22:27:18 +02:00
|
|
|
if err == nil && backIndexRow == nil {
|
2015-05-08 14:07:20 +02:00
|
|
|
udc.m.Lock()
|
2014-09-04 00:17:26 +02:00
|
|
|
udc.docCount++
|
2015-05-08 14:07:20 +02:00
|
|
|
udc.m.Unlock()
|
2014-08-11 22:27:18 +02:00
|
|
|
}
|
2014-10-02 20:12:22 +02:00
|
|
|
atomic.AddUint64(&udc.stats.indexTime, uint64(time.Since(indexStart)))
|
|
|
|
if err == nil {
|
|
|
|
atomic.AddUint64(&udc.stats.updates, 1)
|
|
|
|
} else {
|
|
|
|
atomic.AddUint64(&udc.stats.errors, 1)
|
|
|
|
}
|
2015-04-07 19:26:54 +02:00
|
|
|
return
|
2014-08-11 22:27:18 +02:00
|
|
|
}
|
|
|
|
|
2015-09-02 19:12:08 +02:00
|
|
|
func (udc *UpsideDownCouch) mergeOldAndNew(backIndexRow *BackIndexRow, rows []index.IndexRow, addRows, updateRows, deleteRows []UpsideDownCouchRow) ([]UpsideDownCouchRow, []UpsideDownCouchRow, []UpsideDownCouchRow) {
|
2014-08-19 14:58:26 +02:00
|
|
|
existingTermKeys := make(map[string]bool)
|
|
|
|
for _, key := range backIndexRow.AllTermKeys() {
|
|
|
|
existingTermKeys[string(key)] = true
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
2014-08-19 14:58:26 +02:00
|
|
|
|
|
|
|
existingStoredKeys := make(map[string]bool)
|
|
|
|
for _, key := range backIndexRow.AllStoredKeys() {
|
|
|
|
existingStoredKeys[string(key)] = true
|
2014-06-26 17:43:13 +02:00
|
|
|
}
|
2014-04-17 22:55:53 +02:00
|
|
|
|
2014-09-24 14:13:14 +02:00
|
|
|
for _, row := range rows {
|
|
|
|
switch row := row.(type) {
|
|
|
|
case *TermFrequencyRow:
|
|
|
|
rowKey := string(row.Key())
|
|
|
|
if _, ok := existingTermKeys[rowKey]; ok {
|
|
|
|
updateRows = append(updateRows, row)
|
|
|
|
delete(existingTermKeys, rowKey)
|
|
|
|
} else {
|
|
|
|
addRows = append(addRows, row)
|
2014-06-26 17:43:13 +02:00
|
|
|
}
|
2014-09-24 14:13:14 +02:00
|
|
|
case *StoredRow:
|
|
|
|
rowKey := string(row.Key())
|
|
|
|
if _, ok := existingStoredKeys[rowKey]; ok {
|
|
|
|
updateRows = append(updateRows, row)
|
|
|
|
delete(existingStoredKeys, rowKey)
|
|
|
|
} else {
|
|
|
|
addRows = append(addRows, row)
|
|
|
|
}
|
|
|
|
default:
|
|
|
|
updateRows = append(updateRows, row)
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
// any of the existing rows that weren't updated need to be deleted
|
2014-09-04 00:17:26 +02:00
|
|
|
for existingTermKey := range existingTermKeys {
|
2014-08-19 14:58:26 +02:00
|
|
|
termFreqRow, err := NewTermFrequencyRowK([]byte(existingTermKey))
|
|
|
|
if err == nil {
|
|
|
|
deleteRows = append(deleteRows, termFreqRow)
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
|
|
|
}
|
2014-08-19 14:58:26 +02:00
|
|
|
|
2014-06-26 17:43:13 +02:00
|
|
|
// any of the existing stored fields that weren't updated need to be deleted
|
2014-09-04 00:17:26 +02:00
|
|
|
for existingStoredKey := range existingStoredKeys {
|
2014-08-19 14:58:26 +02:00
|
|
|
storedRow, err := NewStoredRowK([]byte(existingStoredKey))
|
|
|
|
if err == nil {
|
|
|
|
deleteRows = append(deleteRows, storedRow)
|
|
|
|
}
|
2014-06-26 17:43:13 +02:00
|
|
|
}
|
2014-04-17 22:55:53 +02:00
|
|
|
|
2014-08-11 22:27:18 +02:00
|
|
|
return addRows, updateRows, deleteRows
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
|
|
|
|
2015-09-02 19:12:08 +02:00
|
|
|
func (udc *UpsideDownCouch) storeField(docID string, field document.Field, fieldIndex uint16) ([]index.IndexRow, []*BackIndexStoreEntry) {
|
|
|
|
rows := make([]index.IndexRow, 0, 100)
|
2014-08-19 14:58:26 +02:00
|
|
|
backIndexStoredEntries := make([]*BackIndexStoreEntry, 0)
|
2014-08-06 19:52:20 +02:00
|
|
|
fieldType := encodeFieldType(field)
|
2014-09-04 01:53:59 +02:00
|
|
|
storedRow := NewStoredRow(docID, fieldIndex, field.ArrayPositions(), fieldType, field.Value())
|
2014-08-19 14:58:26 +02:00
|
|
|
|
|
|
|
// record the back index entry
|
|
|
|
backIndexStoredEntry := BackIndexStoreEntry{Field: proto.Uint32(uint32(fieldIndex)), ArrayPositions: field.ArrayPositions()}
|
|
|
|
backIndexStoredEntries = append(backIndexStoredEntries, &backIndexStoredEntry)
|
|
|
|
|
2014-09-24 14:13:14 +02:00
|
|
|
rows = append(rows, storedRow)
|
|
|
|
return rows, backIndexStoredEntries
|
2014-07-21 23:05:55 +02:00
|
|
|
}
|
|
|
|
|
2014-08-06 19:52:20 +02:00
|
|
|
func encodeFieldType(f document.Field) byte {
|
|
|
|
fieldType := byte('x')
|
|
|
|
switch f.(type) {
|
|
|
|
case *document.TextField:
|
|
|
|
fieldType = 't'
|
|
|
|
case *document.NumericField:
|
|
|
|
fieldType = 'n'
|
|
|
|
case *document.DateTimeField:
|
|
|
|
fieldType = 'd'
|
|
|
|
case *document.CompositeField:
|
|
|
|
fieldType = 'c'
|
|
|
|
}
|
|
|
|
return fieldType
|
|
|
|
}
|
|
|
|
|
2015-09-02 19:12:08 +02:00
|
|
|
func (udc *UpsideDownCouch) indexField(docID string, field document.Field, fieldIndex uint16, fieldLength int, tokenFreqs analysis.TokenFrequencies) ([]index.IndexRow, []*BackIndexTermEntry) {
|
2014-07-21 23:05:55 +02:00
|
|
|
|
2015-09-02 19:12:08 +02:00
|
|
|
rows := make([]index.IndexRow, 0, 100)
|
2014-08-19 14:58:26 +02:00
|
|
|
backIndexTermEntries := make([]*BackIndexTermEntry, 0)
|
2014-07-21 23:05:55 +02:00
|
|
|
fieldNorm := float32(1.0 / math.Sqrt(float64(fieldLength)))
|
|
|
|
|
|
|
|
for _, tf := range tokenFreqs {
|
|
|
|
var termFreqRow *TermFrequencyRow
|
|
|
|
if field.Options().IncludeTermVectors() {
|
2015-07-31 17:16:11 +02:00
|
|
|
tv, newFieldRows := udc.termVectorsFromTokenFreq(fieldIndex, tf)
|
2014-09-24 14:13:14 +02:00
|
|
|
rows = append(rows, newFieldRows...)
|
2014-09-04 01:53:59 +02:00
|
|
|
termFreqRow = NewTermFrequencyRowWithTermVectors(tf.Term, fieldIndex, docID, uint64(frequencyFromTokenFreq(tf)), fieldNorm, tv)
|
2014-07-21 23:05:55 +02:00
|
|
|
} else {
|
2014-09-04 01:53:59 +02:00
|
|
|
termFreqRow = NewTermFrequencyRow(tf.Term, fieldIndex, docID, uint64(frequencyFromTokenFreq(tf)), fieldNorm)
|
2014-07-21 23:05:55 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
// record the back index entry
|
2014-08-19 14:58:26 +02:00
|
|
|
backIndexTermEntry := BackIndexTermEntry{Term: proto.String(string(tf.Term)), Field: proto.Uint32(uint32(fieldIndex))}
|
|
|
|
backIndexTermEntries = append(backIndexTermEntries, &backIndexTermEntry)
|
|
|
|
|
2014-09-24 14:13:14 +02:00
|
|
|
rows = append(rows, termFreqRow)
|
2014-07-21 23:05:55 +02:00
|
|
|
}
|
|
|
|
|
2014-09-24 14:13:14 +02:00
|
|
|
return rows, backIndexTermEntries
|
2014-07-21 23:05:55 +02:00
|
|
|
}
|
|
|
|
|
2015-04-07 19:26:54 +02:00
|
|
|
func (udc *UpsideDownCouch) Delete(id string) (err error) {
|
2014-10-02 20:12:22 +02:00
|
|
|
indexStart := time.Now()
|
2014-09-12 23:21:35 +02:00
|
|
|
// start a writer for this delete
|
2015-04-07 19:26:54 +02:00
|
|
|
var kvwriter store.KVWriter
|
|
|
|
kvwriter, err = udc.store.Writer()
|
2014-10-31 14:40:23 +01:00
|
|
|
if err != nil {
|
2015-04-07 19:26:54 +02:00
|
|
|
return
|
2014-10-31 14:40:23 +01:00
|
|
|
}
|
2015-04-07 19:26:54 +02:00
|
|
|
defer func() {
|
|
|
|
if cerr := kvwriter.Close(); err == nil && cerr != nil {
|
|
|
|
err = cerr
|
|
|
|
}
|
|
|
|
}()
|
2014-09-12 23:21:35 +02:00
|
|
|
|
2014-04-17 22:55:53 +02:00
|
|
|
// lookup the back index row
|
2015-04-07 19:26:54 +02:00
|
|
|
var backIndexRow *BackIndexRow
|
|
|
|
backIndexRow, err = udc.backIndexRowForDoc(kvwriter, id)
|
2014-04-17 22:55:53 +02:00
|
|
|
if err != nil {
|
2014-10-02 20:12:22 +02:00
|
|
|
atomic.AddUint64(&udc.stats.errors, 1)
|
2015-04-07 19:26:54 +02:00
|
|
|
return
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
|
|
|
if backIndexRow == nil {
|
2014-10-02 20:12:22 +02:00
|
|
|
atomic.AddUint64(&udc.stats.deletes, 1)
|
2015-04-07 19:26:54 +02:00
|
|
|
return
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
|
|
|
|
2014-08-11 22:27:18 +02:00
|
|
|
deleteRows := make([]UpsideDownCouchRow, 0)
|
|
|
|
deleteRows = udc.deleteSingle(id, backIndexRow, deleteRows)
|
|
|
|
|
2014-09-12 23:21:35 +02:00
|
|
|
err = udc.batchRows(kvwriter, nil, nil, deleteRows)
|
2014-08-11 22:27:18 +02:00
|
|
|
if err == nil {
|
2015-05-08 14:07:20 +02:00
|
|
|
udc.m.Lock()
|
2014-09-04 00:17:26 +02:00
|
|
|
udc.docCount--
|
2015-05-08 14:07:20 +02:00
|
|
|
udc.m.Unlock()
|
2014-08-11 22:27:18 +02:00
|
|
|
}
|
2014-10-02 20:12:22 +02:00
|
|
|
atomic.AddUint64(&udc.stats.indexTime, uint64(time.Since(indexStart)))
|
|
|
|
if err == nil {
|
|
|
|
atomic.AddUint64(&udc.stats.deletes, 1)
|
|
|
|
} else {
|
|
|
|
atomic.AddUint64(&udc.stats.errors, 1)
|
|
|
|
}
|
2015-04-07 19:26:54 +02:00
|
|
|
return
|
2014-08-11 22:27:18 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
func (udc *UpsideDownCouch) deleteSingle(id string, backIndexRow *BackIndexRow, deleteRows []UpsideDownCouchRow) []UpsideDownCouchRow {
|
|
|
|
|
2014-08-19 14:58:26 +02:00
|
|
|
for _, backIndexEntry := range backIndexRow.termEntries {
|
|
|
|
tfr := NewTermFrequencyRow([]byte(*backIndexEntry.Term), uint16(*backIndexEntry.Field), id, 0, 0)
|
2014-08-11 22:27:18 +02:00
|
|
|
deleteRows = append(deleteRows, tfr)
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
2014-08-19 14:58:26 +02:00
|
|
|
for _, se := range backIndexRow.storedEntries {
|
|
|
|
sf := NewStoredRow(id, uint16(*se.Field), se.ArrayPositions, 'x', nil)
|
2014-08-11 22:27:18 +02:00
|
|
|
deleteRows = append(deleteRows, sf)
|
2014-06-26 17:43:13 +02:00
|
|
|
}
|
2014-04-17 22:55:53 +02:00
|
|
|
|
|
|
|
// also delete the back entry itself
|
2014-08-11 22:27:18 +02:00
|
|
|
deleteRows = append(deleteRows, backIndexRow)
|
|
|
|
return deleteRows
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
|
|
|
|
2014-09-12 23:21:35 +02:00
|
|
|
func (udc *UpsideDownCouch) backIndexRowForDoc(kvreader store.KVReader, docID string) (*BackIndexRow, error) {
|
2014-04-17 22:55:53 +02:00
|
|
|
// use a temporary row structure to build key
|
|
|
|
tempRow := &BackIndexRow{
|
2014-09-04 01:53:59 +02:00
|
|
|
doc: []byte(docID),
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
2014-04-18 22:09:34 +02:00
|
|
|
key := tempRow.Key()
|
2014-09-12 23:21:35 +02:00
|
|
|
value, err := kvreader.Get(key)
|
2014-04-17 22:55:53 +02:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
if value == nil {
|
|
|
|
return nil, nil
|
|
|
|
}
|
2014-04-19 03:07:41 +02:00
|
|
|
backIndexRow, err := NewBackIndexRowKV(key, value)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
2014-04-17 22:55:53 +02:00
|
|
|
return backIndexRow, nil
|
|
|
|
}
|
|
|
|
|
2014-10-31 14:40:23 +01:00
|
|
|
func (udc *UpsideDownCouch) backIndexRowsForBatch(kvreader store.KVReader, batch *index.Batch) (map[string]*BackIndexRow, error) {
|
2014-08-11 22:27:18 +02:00
|
|
|
// FIXME faster to order the ids and scan sequentially
|
|
|
|
// for now just get it working
|
|
|
|
rv := make(map[string]*BackIndexRow, 0)
|
2014-10-31 14:40:23 +01:00
|
|
|
for docID := range batch.IndexOps {
|
2014-09-12 23:21:35 +02:00
|
|
|
backIndexRow, err := udc.backIndexRowForDoc(kvreader, docID)
|
2014-08-11 22:27:18 +02:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
2014-09-04 01:53:59 +02:00
|
|
|
rv[docID] = backIndexRow
|
2014-08-11 22:27:18 +02:00
|
|
|
}
|
|
|
|
return rv, nil
|
|
|
|
}
|
|
|
|
|
2015-05-15 21:47:54 +02:00
|
|
|
func decodeFieldType(typ byte, name string, pos []uint64, value []byte) document.Field {
|
2014-08-06 19:52:20 +02:00
|
|
|
switch typ {
|
|
|
|
case 't':
|
2015-05-15 21:47:54 +02:00
|
|
|
return document.NewTextField(name, pos, value)
|
2014-08-06 19:52:20 +02:00
|
|
|
case 'n':
|
2015-05-15 21:47:54 +02:00
|
|
|
return document.NewNumericFieldFromBytes(name, pos, value)
|
2014-08-06 19:52:20 +02:00
|
|
|
case 'd':
|
2015-05-15 21:47:54 +02:00
|
|
|
return document.NewDateTimeFieldFromBytes(name, pos, value)
|
2014-08-06 19:52:20 +02:00
|
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2014-04-17 22:55:53 +02:00
|
|
|
func frequencyFromTokenFreq(tf *analysis.TokenFreq) int {
|
|
|
|
return len(tf.Locations)
|
|
|
|
}
|
|
|
|
|
2015-09-02 19:12:08 +02:00
|
|
|
func (udc *UpsideDownCouch) termVectorsFromTokenFreq(field uint16, tf *analysis.TokenFreq) ([]*TermVector, []index.IndexRow) {
|
2014-04-17 22:55:53 +02:00
|
|
|
rv := make([]*TermVector, len(tf.Locations))
|
2015-09-02 19:12:08 +02:00
|
|
|
newFieldRows := make([]index.IndexRow, 0)
|
2014-04-17 22:55:53 +02:00
|
|
|
|
|
|
|
for i, l := range tf.Locations {
|
2014-07-21 23:05:55 +02:00
|
|
|
var newFieldRow *FieldRow
|
|
|
|
fieldIndex := field
|
|
|
|
if l.Field != "" {
|
|
|
|
// lookup correct field
|
2015-09-02 19:12:08 +02:00
|
|
|
fieldIndex, newFieldRow = udc.fieldIndexOrNewRow(l.Field)
|
2014-07-21 23:05:55 +02:00
|
|
|
if newFieldRow != nil {
|
|
|
|
newFieldRows = append(newFieldRows, newFieldRow)
|
|
|
|
}
|
|
|
|
}
|
2014-04-17 22:55:53 +02:00
|
|
|
tv := TermVector{
|
2015-05-17 07:07:14 +02:00
|
|
|
field: fieldIndex,
|
2015-07-31 17:16:11 +02:00
|
|
|
arrayPositions: l.ArrayPositions,
|
2015-05-17 07:07:14 +02:00
|
|
|
pos: uint64(l.Position),
|
|
|
|
start: uint64(l.Start),
|
|
|
|
end: uint64(l.End),
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
|
|
|
rv[i] = &tv
|
|
|
|
}
|
|
|
|
|
2014-07-21 23:05:55 +02:00
|
|
|
return rv, newFieldRows
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
func (udc *UpsideDownCouch) termFieldVectorsFromTermVectors(in []*TermVector) []*index.TermFieldVector {
|
|
|
|
rv := make([]*index.TermFieldVector, len(in))
|
|
|
|
|
|
|
|
for i, tv := range in {
|
2015-09-02 19:12:08 +02:00
|
|
|
fieldName := udc.fieldCache.FieldIndexed(tv.field)
|
2014-04-17 22:55:53 +02:00
|
|
|
tfv := index.TermFieldVector{
|
2015-05-17 07:07:14 +02:00
|
|
|
Field: fieldName,
|
|
|
|
ArrayPositions: tv.arrayPositions,
|
|
|
|
Pos: tv.pos,
|
|
|
|
Start: tv.start,
|
|
|
|
End: tv.end,
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
|
|
|
rv[i] = &tfv
|
|
|
|
}
|
|
|
|
return rv
|
|
|
|
}
|
|
|
|
|
2015-04-07 19:26:54 +02:00
|
|
|
func (udc *UpsideDownCouch) Batch(batch *index.Batch) (err error) {
|
2014-10-02 20:12:22 +02:00
|
|
|
analysisStart := time.Now()
|
2015-09-02 19:12:08 +02:00
|
|
|
resultChan := make(chan *index.AnalysisResult)
|
2014-09-24 14:13:14 +02:00
|
|
|
|
2014-10-02 20:12:22 +02:00
|
|
|
var numUpdates uint64
|
2014-10-31 14:40:23 +01:00
|
|
|
for _, doc := range batch.IndexOps {
|
2014-09-24 14:13:14 +02:00
|
|
|
if doc != nil {
|
|
|
|
numUpdates++
|
2014-04-17 22:55:53 +02:00
|
|
|
}
|
|
|
|
}
|
2014-08-11 22:27:18 +02:00
|
|
|
|
2015-05-15 21:04:52 +02:00
|
|
|
var detectedUnsafeMutex sync.RWMutex
|
|
|
|
detectedUnsafe := false
|
|
|
|
|
2014-09-24 14:13:14 +02:00
|
|
|
go func() {
|
2015-05-15 21:04:52 +02:00
|
|
|
sofar := uint64(0)
|
2014-10-31 14:40:23 +01:00
|
|
|
for _, doc := range batch.IndexOps {
|
2014-09-24 14:13:14 +02:00
|
|
|
if doc != nil {
|
2015-05-15 21:04:52 +02:00
|
|
|
sofar++
|
|
|
|
if sofar > numUpdates {
|
|
|
|
detectedUnsafeMutex.Lock()
|
|
|
|
detectedUnsafe = true
|
|
|
|
detectedUnsafeMutex.Unlock()
|
|
|
|
return
|
|
|
|
}
|
2015-09-02 19:12:08 +02:00
|
|
|
aw := index.NewAnalysisWork(udc, doc, resultChan)
|
2014-09-24 14:13:14 +02:00
|
|
|
// put the work on the queue
|
2015-09-02 19:12:08 +02:00
|
|
|
udc.analysisQueue.Queue(aw)
|
2014-09-24 14:13:14 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}()
|
|
|
|
|
2015-09-02 19:12:08 +02:00
|
|
|
newRowsMap := make(map[string][]index.IndexRow)
|
2014-09-24 14:13:14 +02:00
|
|
|
// wait for the result
|
2014-10-02 20:12:22 +02:00
|
|
|
var itemsDeQueued uint64
|
2014-09-24 14:13:14 +02:00
|
|
|
for itemsDeQueued < numUpdates {
|
|
|
|
result := <-resultChan
|
2015-09-02 19:12:08 +02:00
|
|
|
newRowsMap[result.DocID] = result.Rows
|
2014-09-24 14:13:14 +02:00
|
|
|
itemsDeQueued++
|
|
|
|
}
|
|
|
|
close(resultChan)
|
|
|
|
|
2015-05-15 21:04:52 +02:00
|
|
|
detectedUnsafeMutex.RLock()
|
|
|
|
defer detectedUnsafeMutex.RUnlock()
|
|
|
|
if detectedUnsafe {
|
|
|
|
return UnsafeBatchUseDetected
|
|
|
|
}
|
|
|
|
|
2014-10-02 20:12:22 +02:00
|
|
|
atomic.AddUint64(&udc.stats.analysisTime, uint64(time.Since(analysisStart)))
|
|
|
|
|
|
|
|
indexStart := time.Now()
|
2014-09-12 23:21:35 +02:00
|
|
|
// start a writer for this batch
|
2015-04-07 19:26:54 +02:00
|
|
|
var kvwriter store.KVWriter
|
|
|
|
kvwriter, err = udc.store.Writer()
|
2014-10-31 14:40:23 +01:00
|
|
|
if err != nil {
|
2015-04-07 19:26:54 +02:00
|
|
|
return
|
2014-10-31 14:40:23 +01:00
|
|
|
}
|
2014-09-12 23:21:35 +02:00
|
|
|
|
2014-08-11 22:27:18 +02:00
|
|
|
// first lookup all the back index rows
|
2015-04-07 19:26:54 +02:00
|
|
|
var backIndexRows map[string]*BackIndexRow
|
|
|
|
backIndexRows, err = udc.backIndexRowsForBatch(kvwriter, batch)
|
2014-08-11 22:27:18 +02:00
|
|
|
if err != nil {
|
2015-06-08 20:14:05 +02:00
|
|
|
_ = kvwriter.Close()
|
2015-04-07 19:26:54 +02:00
|
|
|
return
|
2014-08-11 22:27:18 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
// prepare a list of rows
|
|
|
|
addRows := make([]UpsideDownCouchRow, 0)
|
|
|
|
updateRows := make([]UpsideDownCouchRow, 0)
|
|
|
|
deleteRows := make([]UpsideDownCouchRow, 0)
|
|
|
|
|
|
|
|
docsAdded := uint64(0)
|
|
|
|
docsDeleted := uint64(0)
|
2014-10-31 14:40:23 +01:00
|
|
|
for docID, doc := range batch.IndexOps {
|
2014-09-04 01:53:59 +02:00
|
|
|
backIndexRow := backIndexRows[docID]
|
2014-08-11 22:27:18 +02:00
|
|
|
if doc == nil && backIndexRow != nil {
|
2014-12-18 18:43:12 +01:00
|
|
|
// delete
|
2014-09-04 01:53:59 +02:00
|
|
|
deleteRows = udc.deleteSingle(docID, backIndexRow, deleteRows)
|
2014-08-11 22:27:18 +02:00
|
|
|
docsDeleted++
|
|
|
|
} else if doc != nil {
|
2014-09-24 14:13:14 +02:00
|
|
|
addRows, updateRows, deleteRows = udc.mergeOldAndNew(backIndexRow, newRowsMap[docID], addRows, updateRows, deleteRows)
|
2014-08-11 22:27:18 +02:00
|
|
|
if backIndexRow == nil {
|
|
|
|
docsAdded++
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2014-10-31 14:40:23 +01:00
|
|
|
// add the internal ops
|
|
|
|
for internalKey, internalValue := range batch.InternalOps {
|
|
|
|
if internalValue == nil {
|
|
|
|
// delete
|
|
|
|
deleteInternalRow := NewInternalRow([]byte(internalKey), nil)
|
|
|
|
deleteRows = append(deleteRows, deleteInternalRow)
|
|
|
|
} else {
|
|
|
|
updateInternalRow := NewInternalRow([]byte(internalKey), internalValue)
|
|
|
|
updateRows = append(updateRows, updateInternalRow)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2014-09-12 23:21:35 +02:00
|
|
|
err = udc.batchRows(kvwriter, addRows, updateRows, deleteRows)
|
2015-06-08 20:14:05 +02:00
|
|
|
if err != nil {
|
|
|
|
_ = kvwriter.Close()
|
|
|
|
atomic.AddUint64(&udc.stats.errors, 1)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
err = kvwriter.Close()
|
2014-10-02 20:12:22 +02:00
|
|
|
atomic.AddUint64(&udc.stats.indexTime, uint64(time.Since(indexStart)))
|
2015-06-08 20:14:05 +02:00
|
|
|
|
2014-08-11 22:27:18 +02:00
|
|
|
if err == nil {
|
2015-05-08 14:07:20 +02:00
|
|
|
udc.m.Lock()
|
2014-08-11 22:27:18 +02:00
|
|
|
udc.docCount += docsAdded
|
|
|
|
udc.docCount -= docsDeleted
|
2015-05-08 14:07:20 +02:00
|
|
|
udc.m.Unlock()
|
2014-10-02 20:12:22 +02:00
|
|
|
atomic.AddUint64(&udc.stats.updates, numUpdates)
|
|
|
|
atomic.AddUint64(&udc.stats.deletes, docsDeleted)
|
|
|
|
atomic.AddUint64(&udc.stats.batches, 1)
|
|
|
|
} else {
|
|
|
|
atomic.AddUint64(&udc.stats.errors, 1)
|
2014-08-11 22:27:18 +02:00
|
|
|
}
|
2015-04-07 19:26:54 +02:00
|
|
|
return
|
2014-08-11 22:27:18 +02:00
|
|
|
}
|
2014-08-14 03:14:47 +02:00
|
|
|
|
2015-04-07 19:26:54 +02:00
|
|
|
func (udc *UpsideDownCouch) SetInternal(key, val []byte) (err error) {
|
2014-08-14 03:14:47 +02:00
|
|
|
internalRow := NewInternalRow(key, val)
|
2015-04-07 19:26:54 +02:00
|
|
|
var writer store.KVWriter
|
|
|
|
writer, err = udc.store.Writer()
|
2014-10-31 14:40:23 +01:00
|
|
|
if err != nil {
|
2015-04-07 19:26:54 +02:00
|
|
|
return
|
2014-10-31 14:40:23 +01:00
|
|
|
}
|
2015-04-07 19:26:54 +02:00
|
|
|
defer func() {
|
|
|
|
if cerr := writer.Close(); err == nil && cerr != nil {
|
|
|
|
err = cerr
|
|
|
|
}
|
|
|
|
}()
|
2014-09-12 23:21:35 +02:00
|
|
|
return writer.Set(internalRow.Key(), internalRow.Value())
|
2014-08-14 03:14:47 +02:00
|
|
|
}
|
|
|
|
|
2015-04-07 19:26:54 +02:00
|
|
|
func (udc *UpsideDownCouch) DeleteInternal(key []byte) (err error) {
|
2014-08-15 15:39:41 +02:00
|
|
|
internalRow := NewInternalRow(key, nil)
|
2015-04-07 19:26:54 +02:00
|
|
|
var writer store.KVWriter
|
|
|
|
writer, err = udc.store.Writer()
|
2014-10-31 14:40:23 +01:00
|
|
|
if err != nil {
|
2015-04-07 19:26:54 +02:00
|
|
|
return
|
2014-10-31 14:40:23 +01:00
|
|
|
}
|
2015-04-07 19:26:54 +02:00
|
|
|
defer func() {
|
|
|
|
if cerr := writer.Close(); err == nil && cerr != nil {
|
|
|
|
err = cerr
|
|
|
|
}
|
|
|
|
}()
|
2014-09-12 23:21:35 +02:00
|
|
|
return writer.Delete(internalRow.Key())
|
2014-08-14 03:14:47 +02:00
|
|
|
}
|
|
|
|
|
2014-10-31 14:40:23 +01:00
|
|
|
func (udc *UpsideDownCouch) Reader() (index.IndexReader, error) {
|
|
|
|
kvr, err := udc.store.Reader()
|
|
|
|
if err != nil {
|
2015-01-16 20:07:47 +01:00
|
|
|
return nil, fmt.Errorf("error opening store reader: %v", err)
|
2014-10-31 14:40:23 +01:00
|
|
|
}
|
2015-05-08 14:07:20 +02:00
|
|
|
udc.m.RLock()
|
|
|
|
defer udc.m.RUnlock()
|
2014-09-12 23:21:35 +02:00
|
|
|
return &IndexReader{
|
|
|
|
index: udc,
|
2014-10-31 14:40:23 +01:00
|
|
|
kvreader: kvr,
|
2014-09-12 23:21:35 +02:00
|
|
|
docCount: udc.docCount,
|
2014-10-31 14:40:23 +01:00
|
|
|
}, nil
|
2014-08-15 15:39:41 +02:00
|
|
|
}
|
2014-10-02 20:12:22 +02:00
|
|
|
|
|
|
|
func (udc *UpsideDownCouch) Stats() json.Marshaler {
|
|
|
|
return udc.stats
|
|
|
|
}
|
2015-09-02 19:12:08 +02:00
|
|
|
|
|
|
|
func (udc *UpsideDownCouch) fieldIndexOrNewRow(name string) (uint16, *FieldRow) {
|
|
|
|
index, existed := udc.fieldCache.FieldNamed(name, true)
|
|
|
|
if !existed {
|
|
|
|
return index, NewFieldRow(uint16(index), name)
|
|
|
|
}
|
|
|
|
return index, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func IndexTypeConstructor(store store.KVStore, analysisQueue *index.AnalysisQueue) (index.Index, error) {
|
|
|
|
return NewUpsideDownCouch(store, analysisQueue), nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func init() {
|
|
|
|
registry.RegisterIndexType(Name, IndexTypeConstructor)
|
|
|
|
}
|