-
Notifications
You must be signed in to change notification settings - Fork 8.8k
/
query_executer_combiner.go
84 lines (72 loc) · 2.72 KB
/
query_executer_combiner.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
/*
Copyright IBM Corp. All Rights Reserved.
SPDX-License-Identifier: Apache-2.0
*/
package queryutil
import (
"github.com/hyperledger/fabric/common/flogging"
commonledger "github.com/hyperledger/fabric/common/ledger"
"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/statedb"
)
var logger = flogging.MustGetLogger("util")
//go:generate counterfeiter -o mock/query_executer.go -fake-name QueryExecuter . QueryExecuter
// QueryExecuter encapsulates query functions
type QueryExecuter interface {
GetState(namespace, key string) (*statedb.VersionedValue, error)
GetStateRangeScanIterator(namespace, startKey, endKey string) (statedb.ResultsIterator, error)
}
// QECombiner combines the query results from one or more underlying 'queryExecuters'
// In case, the same key is returned by multiple 'queryExecuters', the first 'queryExecuter'
// in the input is considered having the latest state of the key
type QECombiner struct {
QueryExecuters []QueryExecuter // actual executers in decending order of priority
}
// GetState implements function in the interface ledger.SimpleQueryExecutor
func (c *QECombiner) GetState(namespace string, key string) ([]byte, error) {
var vv *statedb.VersionedValue
var val []byte
var err error
for _, qe := range c.QueryExecuters {
if vv, err = qe.GetState(namespace, key); err != nil {
return nil, err
}
if vv != nil {
if !vv.IsDelete() {
val = vv.Value
}
break
}
}
return val, nil
}
// GetStateRangeScanIterator implements function in the interface ledger.SimpleQueryExecutor
func (c *QECombiner) GetStateRangeScanIterator(namespace string, startKey string, endKey string) (commonledger.ResultsIterator, error) {
var itrs []statedb.ResultsIterator
for _, qe := range c.QueryExecuters {
itr, err := qe.GetStateRangeScanIterator(namespace, startKey, endKey)
if err != nil {
for _, itr := range itrs {
itr.Close()
}
return nil, err
}
itrs = append(itrs, itr)
}
itrCombiner, err := newItrCombiner(namespace, itrs)
if err != nil {
return nil, err
}
return itrCombiner, nil
}
// UpdateBatchBackedQueryExecuter wraps an update batch for providing functions in the interface 'queryExecuter'
type UpdateBatchBackedQueryExecuter struct {
UpdateBatch *statedb.UpdateBatch
}
// GetState implements function in interface 'queryExecuter'
func (qe *UpdateBatchBackedQueryExecuter) GetState(ns, key string) (*statedb.VersionedValue, error) {
return qe.UpdateBatch.Get(ns, key), nil
}
// GetStateRangeScanIterator implements function in interface 'queryExecuter'
func (qe *UpdateBatchBackedQueryExecuter) GetStateRangeScanIterator(namespace, startKey, endKey string) (statedb.ResultsIterator, error) {
return qe.UpdateBatch.GetRangeScanIterator(namespace, startKey, endKey), nil
}