You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

437 lines
10 KiB

7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
  1. package kv
  2. import (
  3. "bytes"
  4. "encoding/hex"
  5. "fmt"
  6. "sort"
  7. "strconv"
  8. "strings"
  9. "time"
  10. "github.com/pkg/errors"
  11. cmn "github.com/tendermint/tendermint/libs/common"
  12. dbm "github.com/tendermint/tendermint/libs/db"
  13. "github.com/tendermint/tendermint/libs/pubsub/query"
  14. "github.com/tendermint/tendermint/state/txindex"
  15. "github.com/tendermint/tendermint/types"
  16. )
  17. const (
  18. tagKeySeparator = "/"
  19. )
  20. var _ txindex.TxIndexer = (*TxIndex)(nil)
  21. // TxIndex is the simplest possible indexer, backed by key-value storage (levelDB).
  22. type TxIndex struct {
  23. store dbm.DB
  24. tagsToIndex []string
  25. indexAllTags bool
  26. }
  27. // NewTxIndex creates new KV indexer.
  28. func NewTxIndex(store dbm.DB, options ...func(*TxIndex)) *TxIndex {
  29. txi := &TxIndex{store: store, tagsToIndex: make([]string, 0), indexAllTags: false}
  30. for _, o := range options {
  31. o(txi)
  32. }
  33. return txi
  34. }
  35. // IndexTags is an option for setting which tags to index.
  36. func IndexTags(tags []string) func(*TxIndex) {
  37. return func(txi *TxIndex) {
  38. txi.tagsToIndex = tags
  39. }
  40. }
  41. // IndexAllTags is an option for indexing all tags.
  42. func IndexAllTags() func(*TxIndex) {
  43. return func(txi *TxIndex) {
  44. txi.indexAllTags = true
  45. }
  46. }
  47. // Get gets transaction from the TxIndex storage and returns it or nil if the
  48. // transaction is not found.
  49. func (txi *TxIndex) Get(hash []byte) (*types.TxResult, error) {
  50. if len(hash) == 0 {
  51. return nil, txindex.ErrorEmptyHash
  52. }
  53. rawBytes := txi.store.Get(hash)
  54. if rawBytes == nil {
  55. return nil, nil
  56. }
  57. txResult := new(types.TxResult)
  58. err := cdc.UnmarshalBinaryBare(rawBytes, &txResult)
  59. if err != nil {
  60. return nil, fmt.Errorf("Error reading TxResult: %v", err)
  61. }
  62. return txResult, nil
  63. }
  64. // AddBatch indexes a batch of transactions using the given list of tags.
  65. func (txi *TxIndex) AddBatch(b *txindex.Batch) error {
  66. storeBatch := txi.store.NewBatch()
  67. for _, result := range b.Ops {
  68. hash := result.Tx.Hash()
  69. // index tx by tags
  70. for _, tag := range result.Result.Tags {
  71. if txi.indexAllTags || cmn.StringInSlice(string(tag.Key), txi.tagsToIndex) {
  72. storeBatch.Set(keyForTag(tag, result), hash)
  73. }
  74. }
  75. // index tx by hash
  76. rawBytes, err := cdc.MarshalBinaryBare(result)
  77. if err != nil {
  78. return err
  79. }
  80. storeBatch.Set(hash, rawBytes)
  81. }
  82. storeBatch.Write()
  83. return nil
  84. }
  85. // Index indexes a single transaction using the given list of tags.
  86. func (txi *TxIndex) Index(result *types.TxResult) error {
  87. b := txi.store.NewBatch()
  88. hash := result.Tx.Hash()
  89. // index tx by tags
  90. for _, tag := range result.Result.Tags {
  91. if txi.indexAllTags || cmn.StringInSlice(string(tag.Key), txi.tagsToIndex) {
  92. b.Set(keyForTag(tag, result), hash)
  93. }
  94. }
  95. // index tx by hash
  96. rawBytes, err := cdc.MarshalBinaryBare(result)
  97. if err != nil {
  98. return err
  99. }
  100. b.Set(hash, rawBytes)
  101. b.Write()
  102. return nil
  103. }
  104. // Search performs a search using the given query. It breaks the query into
  105. // conditions (like "tx.height > 5"). For each condition, it queries the DB
  106. // index. One special use cases here: (1) if "tx.hash" is found, it returns tx
  107. // result for it (2) for range queries it is better for the client to provide
  108. // both lower and upper bounds, so we are not performing a full scan. Results
  109. // from querying indexes are then intersected and returned to the caller.
  110. func (txi *TxIndex) Search(q *query.Query) ([]*types.TxResult, error) {
  111. var hashes [][]byte
  112. var hashesInitialized bool
  113. // get a list of conditions (like "tx.height > 5")
  114. conditions := q.Conditions()
  115. // if there is a hash condition, return the result immediately
  116. hash, err, ok := lookForHash(conditions)
  117. if err != nil {
  118. return nil, errors.Wrap(err, "error during searching for a hash in the query")
  119. } else if ok {
  120. res, err := txi.Get(hash)
  121. if res == nil {
  122. return []*types.TxResult{}, nil
  123. }
  124. return []*types.TxResult{res}, errors.Wrap(err, "error while retrieving the result")
  125. }
  126. // conditions to skip because they're handled before "everything else"
  127. skipIndexes := make([]int, 0)
  128. // if there is a height condition ("tx.height=3"), extract it for faster lookups
  129. height, heightIndex := lookForHeight(conditions)
  130. if heightIndex >= 0 {
  131. skipIndexes = append(skipIndexes, heightIndex)
  132. }
  133. // extract ranges
  134. // if both upper and lower bounds exist, it's better to get them in order not
  135. // no iterate over kvs that are not within range.
  136. ranges, rangeIndexes := lookForRanges(conditions)
  137. if len(ranges) > 0 {
  138. skipIndexes = append(skipIndexes, rangeIndexes...)
  139. for _, r := range ranges {
  140. if !hashesInitialized {
  141. hashes = txi.matchRange(r, []byte(r.key))
  142. hashesInitialized = true
  143. } else {
  144. hashes = intersect(hashes, txi.matchRange(r, []byte(r.key)))
  145. }
  146. }
  147. }
  148. // for all other conditions
  149. for i, c := range conditions {
  150. if cmn.IntInSlice(i, skipIndexes) {
  151. continue
  152. }
  153. if !hashesInitialized {
  154. hashes = txi.match(c, startKey(c, height))
  155. hashesInitialized = true
  156. } else {
  157. hashes = intersect(hashes, txi.match(c, startKey(c, height)))
  158. }
  159. }
  160. results := make([]*types.TxResult, len(hashes))
  161. i := 0
  162. for _, h := range hashes {
  163. results[i], err = txi.Get(h)
  164. if err != nil {
  165. return nil, errors.Wrapf(err, "failed to get Tx{%X}", h)
  166. }
  167. i++
  168. }
  169. // sort by height by default
  170. sort.Slice(results, func(i, j int) bool {
  171. return results[i].Height < results[j].Height
  172. })
  173. return results, nil
  174. }
  175. func lookForHash(conditions []query.Condition) (hash []byte, err error, ok bool) {
  176. for _, c := range conditions {
  177. if c.Tag == types.TxHashKey {
  178. decoded, err := hex.DecodeString(c.Operand.(string))
  179. return decoded, err, true
  180. }
  181. }
  182. return
  183. }
  184. func lookForHeight(conditions []query.Condition) (height int64, index int) {
  185. for i, c := range conditions {
  186. if c.Tag == types.TxHeightKey {
  187. return c.Operand.(int64), i
  188. }
  189. }
  190. return 0, -1
  191. }
  192. // special map to hold range conditions
  193. // Example: account.number => queryRange{lowerBound: 1, upperBound: 5}
  194. type queryRanges map[string]queryRange
  195. type queryRange struct {
  196. key string
  197. lowerBound interface{} // int || time.Time
  198. includeLowerBound bool
  199. upperBound interface{} // int || time.Time
  200. includeUpperBound bool
  201. }
  202. func (r queryRange) lowerBoundValue() interface{} {
  203. if r.lowerBound == nil {
  204. return nil
  205. }
  206. if r.includeLowerBound {
  207. return r.lowerBound
  208. } else {
  209. switch t := r.lowerBound.(type) {
  210. case int64:
  211. return t + 1
  212. case time.Time:
  213. return t.Unix() + 1
  214. default:
  215. panic("not implemented")
  216. }
  217. }
  218. }
  219. func (r queryRange) AnyBound() interface{} {
  220. if r.lowerBound != nil {
  221. return r.lowerBound
  222. } else {
  223. return r.upperBound
  224. }
  225. }
  226. func (r queryRange) upperBoundValue() interface{} {
  227. if r.upperBound == nil {
  228. return nil
  229. }
  230. if r.includeUpperBound {
  231. return r.upperBound
  232. } else {
  233. switch t := r.upperBound.(type) {
  234. case int64:
  235. return t - 1
  236. case time.Time:
  237. return t.Unix() - 1
  238. default:
  239. panic("not implemented")
  240. }
  241. }
  242. }
  243. func lookForRanges(conditions []query.Condition) (ranges queryRanges, indexes []int) {
  244. ranges = make(queryRanges)
  245. for i, c := range conditions {
  246. if isRangeOperation(c.Op) {
  247. r, ok := ranges[c.Tag]
  248. if !ok {
  249. r = queryRange{key: c.Tag}
  250. }
  251. switch c.Op {
  252. case query.OpGreater:
  253. r.lowerBound = c.Operand
  254. case query.OpGreaterEqual:
  255. r.includeLowerBound = true
  256. r.lowerBound = c.Operand
  257. case query.OpLess:
  258. r.upperBound = c.Operand
  259. case query.OpLessEqual:
  260. r.includeUpperBound = true
  261. r.upperBound = c.Operand
  262. }
  263. ranges[c.Tag] = r
  264. indexes = append(indexes, i)
  265. }
  266. }
  267. return ranges, indexes
  268. }
  269. func isRangeOperation(op query.Operator) bool {
  270. switch op {
  271. case query.OpGreater, query.OpGreaterEqual, query.OpLess, query.OpLessEqual:
  272. return true
  273. default:
  274. return false
  275. }
  276. }
  277. func (txi *TxIndex) match(c query.Condition, startKey []byte) (hashes [][]byte) {
  278. if c.Op == query.OpEqual {
  279. it := dbm.IteratePrefix(txi.store, startKey)
  280. defer it.Close()
  281. for ; it.Valid(); it.Next() {
  282. hashes = append(hashes, it.Value())
  283. }
  284. } else if c.Op == query.OpContains {
  285. // XXX: doing full scan because startKey does not apply here
  286. // For example, if startKey = "account.owner=an" and search query = "accoutn.owner CONSISTS an"
  287. // we can't iterate with prefix "account.owner=an" because we might miss keys like "account.owner=Ulan"
  288. it := txi.store.Iterator(nil, nil)
  289. defer it.Close()
  290. for ; it.Valid(); it.Next() {
  291. if !isTagKey(it.Key()) {
  292. continue
  293. }
  294. if strings.Contains(extractValueFromKey(it.Key()), c.Operand.(string)) {
  295. hashes = append(hashes, it.Value())
  296. }
  297. }
  298. } else {
  299. panic("other operators should be handled already")
  300. }
  301. return
  302. }
  303. func (txi *TxIndex) matchRange(r queryRange, prefix []byte) (hashes [][]byte) {
  304. // create a map to prevent duplicates
  305. hashesMap := make(map[string][]byte)
  306. lowerBound := r.lowerBoundValue()
  307. upperBound := r.upperBoundValue()
  308. it := dbm.IteratePrefix(txi.store, prefix)
  309. defer it.Close()
  310. LOOP:
  311. for ; it.Valid(); it.Next() {
  312. if !isTagKey(it.Key()) {
  313. continue
  314. }
  315. switch r.AnyBound().(type) {
  316. case int64:
  317. v, err := strconv.ParseInt(extractValueFromKey(it.Key()), 10, 64)
  318. if err != nil {
  319. continue LOOP
  320. }
  321. include := true
  322. if lowerBound != nil && v < lowerBound.(int64) {
  323. include = false
  324. }
  325. if upperBound != nil && v > upperBound.(int64) {
  326. include = false
  327. }
  328. if include {
  329. hashesMap[fmt.Sprintf("%X", it.Value())] = it.Value()
  330. }
  331. // XXX: passing time in a ABCI Tags is not yet implemented
  332. // case time.Time:
  333. // v := strconv.ParseInt(extractValueFromKey(it.Key()), 10, 64)
  334. // if v == r.upperBound {
  335. // break
  336. // }
  337. }
  338. }
  339. hashes = make([][]byte, len(hashesMap))
  340. i := 0
  341. for _, h := range hashesMap {
  342. hashes[i] = h
  343. i++
  344. }
  345. return
  346. }
  347. ///////////////////////////////////////////////////////////////////////////////
  348. // Keys
  349. func startKey(c query.Condition, height int64) []byte {
  350. var key string
  351. if height > 0 {
  352. key = fmt.Sprintf("%s/%v/%d", c.Tag, c.Operand, height)
  353. } else {
  354. key = fmt.Sprintf("%s/%v", c.Tag, c.Operand)
  355. }
  356. return []byte(key)
  357. }
  358. func isTagKey(key []byte) bool {
  359. return strings.Count(string(key), tagKeySeparator) == 3
  360. }
  361. func extractValueFromKey(key []byte) string {
  362. parts := strings.SplitN(string(key), tagKeySeparator, 3)
  363. return parts[1]
  364. }
  365. func keyForTag(tag cmn.KVPair, result *types.TxResult) []byte {
  366. return []byte(fmt.Sprintf("%s/%s/%d/%d", tag.Key, tag.Value, result.Height, result.Index))
  367. }
  368. ///////////////////////////////////////////////////////////////////////////////
  369. // Utils
  370. func intersect(as, bs [][]byte) [][]byte {
  371. i := make([][]byte, 0, cmn.MinInt(len(as), len(bs)))
  372. for _, a := range as {
  373. for _, b := range bs {
  374. if bytes.Equal(a, b) {
  375. i = append(i, a)
  376. }
  377. }
  378. }
  379. return i
  380. }