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.

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