reader.go 27 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139
  1. // Copyright (c) 2012, Suryandaru Triandana <syndtr@gmail.com>
  2. // All rights reserved.
  3. //
  4. // Use of this source code is governed by a BSD-style license that can be
  5. // found in the LICENSE file.
  6. package table
  7. import (
  8. "encoding/binary"
  9. "fmt"
  10. "io"
  11. "sort"
  12. "strings"
  13. "sync"
  14. "github.com/golang/snappy"
  15. "github.com/syndtr/goleveldb/leveldb/cache"
  16. "github.com/syndtr/goleveldb/leveldb/comparer"
  17. "github.com/syndtr/goleveldb/leveldb/errors"
  18. "github.com/syndtr/goleveldb/leveldb/filter"
  19. "github.com/syndtr/goleveldb/leveldb/iterator"
  20. "github.com/syndtr/goleveldb/leveldb/opt"
  21. "github.com/syndtr/goleveldb/leveldb/storage"
  22. "github.com/syndtr/goleveldb/leveldb/util"
  23. )
  24. // Reader errors.
  25. var (
  26. ErrNotFound = errors.ErrNotFound
  27. ErrReaderReleased = errors.New("leveldb/table: reader released")
  28. ErrIterReleased = errors.New("leveldb/table: iterator released")
  29. )
  30. // ErrCorrupted describes error due to corruption. This error will be wrapped
  31. // with errors.ErrCorrupted.
  32. type ErrCorrupted struct {
  33. Pos int64
  34. Size int64
  35. Kind string
  36. Reason string
  37. }
  38. func (e *ErrCorrupted) Error() string {
  39. return fmt.Sprintf("leveldb/table: corruption on %s (pos=%d): %s", e.Kind, e.Pos, e.Reason)
  40. }
  41. func max(x, y int) int {
  42. if x > y {
  43. return x
  44. }
  45. return y
  46. }
  47. type block struct {
  48. bpool *util.BufferPool
  49. bh blockHandle
  50. data []byte
  51. restartsLen int
  52. restartsOffset int
  53. }
  54. func (b *block) seek(cmp comparer.Comparer, rstart, rlimit int, key []byte) (index, offset int, err error) {
  55. index = sort.Search(b.restartsLen-rstart-(b.restartsLen-rlimit), func(i int) bool {
  56. offset := int(binary.LittleEndian.Uint32(b.data[b.restartsOffset+4*(rstart+i):]))
  57. offset++ // shared always zero, since this is a restart point
  58. v1, n1 := binary.Uvarint(b.data[offset:]) // key length
  59. _, n2 := binary.Uvarint(b.data[offset+n1:]) // value length
  60. m := offset + n1 + n2
  61. return cmp.Compare(b.data[m:m+int(v1)], key) > 0
  62. }) + rstart - 1
  63. if index < rstart {
  64. // The smallest key is greater-than key sought.
  65. index = rstart
  66. }
  67. offset = int(binary.LittleEndian.Uint32(b.data[b.restartsOffset+4*index:]))
  68. return
  69. }
  70. func (b *block) restartIndex(rstart, rlimit, offset int) int {
  71. return sort.Search(b.restartsLen-rstart-(b.restartsLen-rlimit), func(i int) bool {
  72. return int(binary.LittleEndian.Uint32(b.data[b.restartsOffset+4*(rstart+i):])) > offset
  73. }) + rstart - 1
  74. }
  75. func (b *block) restartOffset(index int) int {
  76. return int(binary.LittleEndian.Uint32(b.data[b.restartsOffset+4*index:]))
  77. }
  78. func (b *block) entry(offset int) (key, value []byte, nShared, n int, err error) {
  79. if offset >= b.restartsOffset {
  80. if offset != b.restartsOffset {
  81. err = &ErrCorrupted{Reason: "entries offset not aligned"}
  82. }
  83. return
  84. }
  85. v0, n0 := binary.Uvarint(b.data[offset:]) // Shared prefix length
  86. v1, n1 := binary.Uvarint(b.data[offset+n0:]) // Key length
  87. v2, n2 := binary.Uvarint(b.data[offset+n0+n1:]) // Value length
  88. m := n0 + n1 + n2
  89. n = m + int(v1) + int(v2)
  90. if n0 <= 0 || n1 <= 0 || n2 <= 0 || offset+n > b.restartsOffset {
  91. err = &ErrCorrupted{Reason: "entries corrupted"}
  92. return
  93. }
  94. key = b.data[offset+m : offset+m+int(v1)]
  95. value = b.data[offset+m+int(v1) : offset+n]
  96. nShared = int(v0)
  97. return
  98. }
  99. func (b *block) Release() {
  100. b.bpool.Put(b.data)
  101. b.bpool = nil
  102. b.data = nil
  103. }
  104. type dir int
  105. const (
  106. dirReleased dir = iota - 1
  107. dirSOI
  108. dirEOI
  109. dirBackward
  110. dirForward
  111. )
  112. type blockIter struct {
  113. tr *Reader
  114. block *block
  115. blockReleaser util.Releaser
  116. releaser util.Releaser
  117. key, value []byte
  118. offset int
  119. // Previous offset, only filled by Next.
  120. prevOffset int
  121. prevNode []int
  122. prevKeys []byte
  123. restartIndex int
  124. // Iterator direction.
  125. dir dir
  126. // Restart index slice range.
  127. riStart int
  128. riLimit int
  129. // Offset slice range.
  130. offsetStart int
  131. offsetRealStart int
  132. offsetLimit int
  133. // Error.
  134. err error
  135. }
  136. func (i *blockIter) sErr(err error) {
  137. i.err = err
  138. i.key = nil
  139. i.value = nil
  140. i.prevNode = nil
  141. i.prevKeys = nil
  142. }
  143. func (i *blockIter) reset() {
  144. if i.dir == dirBackward {
  145. i.prevNode = i.prevNode[:0]
  146. i.prevKeys = i.prevKeys[:0]
  147. }
  148. i.restartIndex = i.riStart
  149. i.offset = i.offsetStart
  150. i.dir = dirSOI
  151. i.key = i.key[:0]
  152. i.value = nil
  153. }
  154. func (i *blockIter) isFirst() bool {
  155. switch i.dir {
  156. case dirForward:
  157. return i.prevOffset == i.offsetRealStart
  158. case dirBackward:
  159. return len(i.prevNode) == 1 && i.restartIndex == i.riStart
  160. }
  161. return false
  162. }
  163. func (i *blockIter) isLast() bool {
  164. switch i.dir {
  165. case dirForward, dirBackward:
  166. return i.offset == i.offsetLimit
  167. }
  168. return false
  169. }
  170. func (i *blockIter) First() bool {
  171. if i.err != nil {
  172. return false
  173. } else if i.dir == dirReleased {
  174. i.err = ErrIterReleased
  175. return false
  176. }
  177. if i.dir == dirBackward {
  178. i.prevNode = i.prevNode[:0]
  179. i.prevKeys = i.prevKeys[:0]
  180. }
  181. i.dir = dirSOI
  182. return i.Next()
  183. }
  184. func (i *blockIter) Last() bool {
  185. if i.err != nil {
  186. return false
  187. } else if i.dir == dirReleased {
  188. i.err = ErrIterReleased
  189. return false
  190. }
  191. if i.dir == dirBackward {
  192. i.prevNode = i.prevNode[:0]
  193. i.prevKeys = i.prevKeys[:0]
  194. }
  195. i.dir = dirEOI
  196. return i.Prev()
  197. }
  198. func (i *blockIter) Seek(key []byte) bool {
  199. if i.err != nil {
  200. return false
  201. } else if i.dir == dirReleased {
  202. i.err = ErrIterReleased
  203. return false
  204. }
  205. ri, offset, err := i.block.seek(i.tr.cmp, i.riStart, i.riLimit, key)
  206. if err != nil {
  207. i.sErr(err)
  208. return false
  209. }
  210. i.restartIndex = ri
  211. i.offset = max(i.offsetStart, offset)
  212. if i.dir == dirSOI || i.dir == dirEOI {
  213. i.dir = dirForward
  214. }
  215. for i.Next() {
  216. if i.tr.cmp.Compare(i.key, key) >= 0 {
  217. return true
  218. }
  219. }
  220. return false
  221. }
  222. func (i *blockIter) Next() bool {
  223. if i.dir == dirEOI || i.err != nil {
  224. return false
  225. } else if i.dir == dirReleased {
  226. i.err = ErrIterReleased
  227. return false
  228. }
  229. if i.dir == dirSOI {
  230. i.restartIndex = i.riStart
  231. i.offset = i.offsetStart
  232. } else if i.dir == dirBackward {
  233. i.prevNode = i.prevNode[:0]
  234. i.prevKeys = i.prevKeys[:0]
  235. }
  236. for i.offset < i.offsetRealStart {
  237. key, value, nShared, n, err := i.block.entry(i.offset)
  238. if err != nil {
  239. i.sErr(i.tr.fixErrCorruptedBH(i.block.bh, err))
  240. return false
  241. }
  242. if n == 0 {
  243. i.dir = dirEOI
  244. return false
  245. }
  246. i.key = append(i.key[:nShared], key...)
  247. i.value = value
  248. i.offset += n
  249. }
  250. if i.offset >= i.offsetLimit {
  251. i.dir = dirEOI
  252. if i.offset != i.offsetLimit {
  253. i.sErr(i.tr.newErrCorruptedBH(i.block.bh, "entries offset not aligned"))
  254. }
  255. return false
  256. }
  257. key, value, nShared, n, err := i.block.entry(i.offset)
  258. if err != nil {
  259. i.sErr(i.tr.fixErrCorruptedBH(i.block.bh, err))
  260. return false
  261. }
  262. if n == 0 {
  263. i.dir = dirEOI
  264. return false
  265. }
  266. i.key = append(i.key[:nShared], key...)
  267. i.value = value
  268. i.prevOffset = i.offset
  269. i.offset += n
  270. i.dir = dirForward
  271. return true
  272. }
  273. func (i *blockIter) Prev() bool {
  274. if i.dir == dirSOI || i.err != nil {
  275. return false
  276. } else if i.dir == dirReleased {
  277. i.err = ErrIterReleased
  278. return false
  279. }
  280. var ri int
  281. if i.dir == dirForward {
  282. // Change direction.
  283. i.offset = i.prevOffset
  284. if i.offset == i.offsetRealStart {
  285. i.dir = dirSOI
  286. return false
  287. }
  288. ri = i.block.restartIndex(i.restartIndex, i.riLimit, i.offset)
  289. i.dir = dirBackward
  290. } else if i.dir == dirEOI {
  291. // At the end of iterator.
  292. i.restartIndex = i.riLimit
  293. i.offset = i.offsetLimit
  294. if i.offset == i.offsetRealStart {
  295. i.dir = dirSOI
  296. return false
  297. }
  298. ri = i.riLimit - 1
  299. i.dir = dirBackward
  300. } else if len(i.prevNode) == 1 {
  301. // This is the end of a restart range.
  302. i.offset = i.prevNode[0]
  303. i.prevNode = i.prevNode[:0]
  304. if i.restartIndex == i.riStart {
  305. i.dir = dirSOI
  306. return false
  307. }
  308. i.restartIndex--
  309. ri = i.restartIndex
  310. } else {
  311. // In the middle of restart range, get from cache.
  312. n := len(i.prevNode) - 3
  313. node := i.prevNode[n:]
  314. i.prevNode = i.prevNode[:n]
  315. // Get the key.
  316. ko := node[0]
  317. i.key = append(i.key[:0], i.prevKeys[ko:]...)
  318. i.prevKeys = i.prevKeys[:ko]
  319. // Get the value.
  320. vo := node[1]
  321. vl := vo + node[2]
  322. i.value = i.block.data[vo:vl]
  323. i.offset = vl
  324. return true
  325. }
  326. // Build entries cache.
  327. i.key = i.key[:0]
  328. i.value = nil
  329. offset := i.block.restartOffset(ri)
  330. if offset == i.offset {
  331. ri--
  332. if ri < 0 {
  333. i.dir = dirSOI
  334. return false
  335. }
  336. offset = i.block.restartOffset(ri)
  337. }
  338. i.prevNode = append(i.prevNode, offset)
  339. for {
  340. key, value, nShared, n, err := i.block.entry(offset)
  341. if err != nil {
  342. i.sErr(i.tr.fixErrCorruptedBH(i.block.bh, err))
  343. return false
  344. }
  345. if offset >= i.offsetRealStart {
  346. if i.value != nil {
  347. // Appends 3 variables:
  348. // 1. Previous keys offset
  349. // 2. Value offset in the data block
  350. // 3. Value length
  351. i.prevNode = append(i.prevNode, len(i.prevKeys), offset-len(i.value), len(i.value))
  352. i.prevKeys = append(i.prevKeys, i.key...)
  353. }
  354. i.value = value
  355. }
  356. i.key = append(i.key[:nShared], key...)
  357. offset += n
  358. // Stop if target offset reached.
  359. if offset >= i.offset {
  360. if offset != i.offset {
  361. i.sErr(i.tr.newErrCorruptedBH(i.block.bh, "entries offset not aligned"))
  362. return false
  363. }
  364. break
  365. }
  366. }
  367. i.restartIndex = ri
  368. i.offset = offset
  369. return true
  370. }
  371. func (i *blockIter) Key() []byte {
  372. if i.err != nil || i.dir <= dirEOI {
  373. return nil
  374. }
  375. return i.key
  376. }
  377. func (i *blockIter) Value() []byte {
  378. if i.err != nil || i.dir <= dirEOI {
  379. return nil
  380. }
  381. return i.value
  382. }
  383. func (i *blockIter) Release() {
  384. if i.dir != dirReleased {
  385. i.tr = nil
  386. i.block = nil
  387. i.prevNode = nil
  388. i.prevKeys = nil
  389. i.key = nil
  390. i.value = nil
  391. i.dir = dirReleased
  392. if i.blockReleaser != nil {
  393. i.blockReleaser.Release()
  394. i.blockReleaser = nil
  395. }
  396. if i.releaser != nil {
  397. i.releaser.Release()
  398. i.releaser = nil
  399. }
  400. }
  401. }
  402. func (i *blockIter) SetReleaser(releaser util.Releaser) {
  403. if i.dir == dirReleased {
  404. panic(util.ErrReleased)
  405. }
  406. if i.releaser != nil && releaser != nil {
  407. panic(util.ErrHasReleaser)
  408. }
  409. i.releaser = releaser
  410. }
  411. func (i *blockIter) Valid() bool {
  412. return i.err == nil && (i.dir == dirBackward || i.dir == dirForward)
  413. }
  414. func (i *blockIter) Error() error {
  415. return i.err
  416. }
  417. type filterBlock struct {
  418. bpool *util.BufferPool
  419. data []byte
  420. oOffset int
  421. baseLg uint
  422. filtersNum int
  423. }
  424. func (b *filterBlock) contains(filter filter.Filter, offset uint64, key []byte) bool {
  425. i := int(offset >> b.baseLg)
  426. if i < b.filtersNum {
  427. o := b.data[b.oOffset+i*4:]
  428. n := int(binary.LittleEndian.Uint32(o))
  429. m := int(binary.LittleEndian.Uint32(o[4:]))
  430. if n < m && m <= b.oOffset {
  431. return filter.Contains(b.data[n:m], key)
  432. } else if n == m {
  433. return false
  434. }
  435. }
  436. return true
  437. }
  438. func (b *filterBlock) Release() {
  439. b.bpool.Put(b.data)
  440. b.bpool = nil
  441. b.data = nil
  442. }
  443. type indexIter struct {
  444. *blockIter
  445. tr *Reader
  446. slice *util.Range
  447. // Options
  448. fillCache bool
  449. }
  450. func (i *indexIter) Get() iterator.Iterator {
  451. value := i.Value()
  452. if value == nil {
  453. return nil
  454. }
  455. dataBH, n := decodeBlockHandle(value)
  456. if n == 0 {
  457. return iterator.NewEmptyIterator(i.tr.newErrCorruptedBH(i.tr.indexBH, "bad data block handle"))
  458. }
  459. var slice *util.Range
  460. if i.slice != nil && (i.blockIter.isFirst() || i.blockIter.isLast()) {
  461. slice = i.slice
  462. }
  463. return i.tr.getDataIterErr(dataBH, slice, i.tr.verifyChecksum, i.fillCache)
  464. }
  465. // Reader is a table reader.
  466. type Reader struct {
  467. mu sync.RWMutex
  468. fd storage.FileDesc
  469. reader io.ReaderAt
  470. cache *cache.NamespaceGetter
  471. err error
  472. bpool *util.BufferPool
  473. // Options
  474. o *opt.Options
  475. cmp comparer.Comparer
  476. filter filter.Filter
  477. verifyChecksum bool
  478. dataEnd int64
  479. metaBH, indexBH, filterBH blockHandle
  480. indexBlock *block
  481. filterBlock *filterBlock
  482. }
  483. func (r *Reader) blockKind(bh blockHandle) string {
  484. switch bh.offset {
  485. case r.metaBH.offset:
  486. return "meta-block"
  487. case r.indexBH.offset:
  488. return "index-block"
  489. case r.filterBH.offset:
  490. if r.filterBH.length > 0 {
  491. return "filter-block"
  492. }
  493. }
  494. return "data-block"
  495. }
  496. func (r *Reader) newErrCorrupted(pos, size int64, kind, reason string) error {
  497. return &errors.ErrCorrupted{Fd: r.fd, Err: &ErrCorrupted{Pos: pos, Size: size, Kind: kind, Reason: reason}}
  498. }
  499. func (r *Reader) newErrCorruptedBH(bh blockHandle, reason string) error {
  500. return r.newErrCorrupted(int64(bh.offset), int64(bh.length), r.blockKind(bh), reason)
  501. }
  502. func (r *Reader) fixErrCorruptedBH(bh blockHandle, err error) error {
  503. if cerr, ok := err.(*ErrCorrupted); ok {
  504. cerr.Pos = int64(bh.offset)
  505. cerr.Size = int64(bh.length)
  506. cerr.Kind = r.blockKind(bh)
  507. return &errors.ErrCorrupted{Fd: r.fd, Err: cerr}
  508. }
  509. return err
  510. }
  511. func (r *Reader) readRawBlock(bh blockHandle, verifyChecksum bool) ([]byte, error) {
  512. data := r.bpool.Get(int(bh.length + blockTrailerLen))
  513. if _, err := r.reader.ReadAt(data, int64(bh.offset)); err != nil && err != io.EOF {
  514. return nil, err
  515. }
  516. if verifyChecksum {
  517. n := bh.length + 1
  518. checksum0 := binary.LittleEndian.Uint32(data[n:])
  519. checksum1 := util.NewCRC(data[:n]).Value()
  520. if checksum0 != checksum1 {
  521. r.bpool.Put(data)
  522. return nil, r.newErrCorruptedBH(bh, fmt.Sprintf("checksum mismatch, want=%#x got=%#x", checksum0, checksum1))
  523. }
  524. }
  525. switch data[bh.length] {
  526. case blockTypeNoCompression:
  527. data = data[:bh.length]
  528. case blockTypeSnappyCompression:
  529. decLen, err := snappy.DecodedLen(data[:bh.length])
  530. if err != nil {
  531. r.bpool.Put(data)
  532. return nil, r.newErrCorruptedBH(bh, err.Error())
  533. }
  534. decData := r.bpool.Get(decLen)
  535. decData, err = snappy.Decode(decData, data[:bh.length])
  536. r.bpool.Put(data)
  537. if err != nil {
  538. r.bpool.Put(decData)
  539. return nil, r.newErrCorruptedBH(bh, err.Error())
  540. }
  541. data = decData
  542. default:
  543. r.bpool.Put(data)
  544. return nil, r.newErrCorruptedBH(bh, fmt.Sprintf("unknown compression type %#x", data[bh.length]))
  545. }
  546. return data, nil
  547. }
  548. func (r *Reader) readBlock(bh blockHandle, verifyChecksum bool) (*block, error) {
  549. data, err := r.readRawBlock(bh, verifyChecksum)
  550. if err != nil {
  551. return nil, err
  552. }
  553. restartsLen := int(binary.LittleEndian.Uint32(data[len(data)-4:]))
  554. b := &block{
  555. bpool: r.bpool,
  556. bh: bh,
  557. data: data,
  558. restartsLen: restartsLen,
  559. restartsOffset: len(data) - (restartsLen+1)*4,
  560. }
  561. return b, nil
  562. }
  563. func (r *Reader) readBlockCached(bh blockHandle, verifyChecksum, fillCache bool) (*block, util.Releaser, error) {
  564. if r.cache != nil {
  565. var (
  566. err error
  567. ch *cache.Handle
  568. )
  569. if fillCache {
  570. ch = r.cache.Get(bh.offset, func() (size int, value cache.Value) {
  571. var b *block
  572. b, err = r.readBlock(bh, verifyChecksum)
  573. if err != nil {
  574. return 0, nil
  575. }
  576. return cap(b.data), b
  577. })
  578. } else {
  579. ch = r.cache.Get(bh.offset, nil)
  580. }
  581. if ch != nil {
  582. b, ok := ch.Value().(*block)
  583. if !ok {
  584. ch.Release()
  585. return nil, nil, errors.New("leveldb/table: inconsistent block type")
  586. }
  587. return b, ch, err
  588. } else if err != nil {
  589. return nil, nil, err
  590. }
  591. }
  592. b, err := r.readBlock(bh, verifyChecksum)
  593. return b, b, err
  594. }
  595. func (r *Reader) readFilterBlock(bh blockHandle) (*filterBlock, error) {
  596. data, err := r.readRawBlock(bh, true)
  597. if err != nil {
  598. return nil, err
  599. }
  600. n := len(data)
  601. if n < 5 {
  602. return nil, r.newErrCorruptedBH(bh, "too short")
  603. }
  604. m := n - 5
  605. oOffset := int(binary.LittleEndian.Uint32(data[m:]))
  606. if oOffset > m {
  607. return nil, r.newErrCorruptedBH(bh, "invalid data-offsets offset")
  608. }
  609. b := &filterBlock{
  610. bpool: r.bpool,
  611. data: data,
  612. oOffset: oOffset,
  613. baseLg: uint(data[n-1]),
  614. filtersNum: (m - oOffset) / 4,
  615. }
  616. return b, nil
  617. }
  618. func (r *Reader) readFilterBlockCached(bh blockHandle, fillCache bool) (*filterBlock, util.Releaser, error) {
  619. if r.cache != nil {
  620. var (
  621. err error
  622. ch *cache.Handle
  623. )
  624. if fillCache {
  625. ch = r.cache.Get(bh.offset, func() (size int, value cache.Value) {
  626. var b *filterBlock
  627. b, err = r.readFilterBlock(bh)
  628. if err != nil {
  629. return 0, nil
  630. }
  631. return cap(b.data), b
  632. })
  633. } else {
  634. ch = r.cache.Get(bh.offset, nil)
  635. }
  636. if ch != nil {
  637. b, ok := ch.Value().(*filterBlock)
  638. if !ok {
  639. ch.Release()
  640. return nil, nil, errors.New("leveldb/table: inconsistent block type")
  641. }
  642. return b, ch, err
  643. } else if err != nil {
  644. return nil, nil, err
  645. }
  646. }
  647. b, err := r.readFilterBlock(bh)
  648. return b, b, err
  649. }
  650. func (r *Reader) getIndexBlock(fillCache bool) (b *block, rel util.Releaser, err error) {
  651. if r.indexBlock == nil {
  652. return r.readBlockCached(r.indexBH, true, fillCache)
  653. }
  654. return r.indexBlock, util.NoopReleaser{}, nil
  655. }
  656. func (r *Reader) getFilterBlock(fillCache bool) (*filterBlock, util.Releaser, error) {
  657. if r.filterBlock == nil {
  658. return r.readFilterBlockCached(r.filterBH, fillCache)
  659. }
  660. return r.filterBlock, util.NoopReleaser{}, nil
  661. }
  662. func (r *Reader) newBlockIter(b *block, bReleaser util.Releaser, slice *util.Range, inclLimit bool) *blockIter {
  663. bi := &blockIter{
  664. tr: r,
  665. block: b,
  666. blockReleaser: bReleaser,
  667. // Valid key should never be nil.
  668. key: make([]byte, 0),
  669. dir: dirSOI,
  670. riStart: 0,
  671. riLimit: b.restartsLen,
  672. offsetStart: 0,
  673. offsetRealStart: 0,
  674. offsetLimit: b.restartsOffset,
  675. }
  676. if slice != nil {
  677. if slice.Start != nil {
  678. if bi.Seek(slice.Start) {
  679. bi.riStart = b.restartIndex(bi.restartIndex, b.restartsLen, bi.prevOffset)
  680. bi.offsetStart = b.restartOffset(bi.riStart)
  681. bi.offsetRealStart = bi.prevOffset
  682. } else {
  683. bi.riStart = b.restartsLen
  684. bi.offsetStart = b.restartsOffset
  685. bi.offsetRealStart = b.restartsOffset
  686. }
  687. }
  688. if slice.Limit != nil {
  689. if bi.Seek(slice.Limit) && (!inclLimit || bi.Next()) {
  690. bi.offsetLimit = bi.prevOffset
  691. bi.riLimit = bi.restartIndex + 1
  692. }
  693. }
  694. bi.reset()
  695. if bi.offsetStart > bi.offsetLimit {
  696. bi.sErr(errors.New("leveldb/table: invalid slice range"))
  697. }
  698. }
  699. return bi
  700. }
  701. func (r *Reader) getDataIter(dataBH blockHandle, slice *util.Range, verifyChecksum, fillCache bool) iterator.Iterator {
  702. b, rel, err := r.readBlockCached(dataBH, verifyChecksum, fillCache)
  703. if err != nil {
  704. return iterator.NewEmptyIterator(err)
  705. }
  706. return r.newBlockIter(b, rel, slice, false)
  707. }
  708. func (r *Reader) getDataIterErr(dataBH blockHandle, slice *util.Range, verifyChecksum, fillCache bool) iterator.Iterator {
  709. r.mu.RLock()
  710. defer r.mu.RUnlock()
  711. if r.err != nil {
  712. return iterator.NewEmptyIterator(r.err)
  713. }
  714. return r.getDataIter(dataBH, slice, verifyChecksum, fillCache)
  715. }
  716. // NewIterator creates an iterator from the table.
  717. //
  718. // Slice allows slicing the iterator to only contains keys in the given
  719. // range. A nil Range.Start is treated as a key before all keys in the
  720. // table. And a nil Range.Limit is treated as a key after all keys in
  721. // the table.
  722. //
  723. // WARNING: Any slice returned by interator (e.g. slice returned by calling
  724. // Iterator.Key() or Iterator.Key() methods), its content should not be modified
  725. // unless noted otherwise.
  726. //
  727. // The returned iterator is not safe for concurrent use and should be released
  728. // after use.
  729. //
  730. // Also read Iterator documentation of the leveldb/iterator package.
  731. func (r *Reader) NewIterator(slice *util.Range, ro *opt.ReadOptions) iterator.Iterator {
  732. r.mu.RLock()
  733. defer r.mu.RUnlock()
  734. if r.err != nil {
  735. return iterator.NewEmptyIterator(r.err)
  736. }
  737. fillCache := !ro.GetDontFillCache()
  738. indexBlock, rel, err := r.getIndexBlock(fillCache)
  739. if err != nil {
  740. return iterator.NewEmptyIterator(err)
  741. }
  742. index := &indexIter{
  743. blockIter: r.newBlockIter(indexBlock, rel, slice, true),
  744. tr: r,
  745. slice: slice,
  746. fillCache: !ro.GetDontFillCache(),
  747. }
  748. return iterator.NewIndexedIterator(index, opt.GetStrict(r.o, ro, opt.StrictReader))
  749. }
  750. func (r *Reader) find(key []byte, filtered bool, ro *opt.ReadOptions, noValue bool) (rkey, value []byte, err error) {
  751. r.mu.RLock()
  752. defer r.mu.RUnlock()
  753. if r.err != nil {
  754. err = r.err
  755. return
  756. }
  757. indexBlock, rel, err := r.getIndexBlock(true)
  758. if err != nil {
  759. return
  760. }
  761. defer rel.Release()
  762. index := r.newBlockIter(indexBlock, nil, nil, true)
  763. defer index.Release()
  764. if !index.Seek(key) {
  765. if err = index.Error(); err == nil {
  766. err = ErrNotFound
  767. }
  768. return
  769. }
  770. dataBH, n := decodeBlockHandle(index.Value())
  771. if n == 0 {
  772. r.err = r.newErrCorruptedBH(r.indexBH, "bad data block handle")
  773. return nil, nil, r.err
  774. }
  775. // The filter should only used for exact match.
  776. if filtered && r.filter != nil {
  777. filterBlock, frel, ferr := r.getFilterBlock(true)
  778. if ferr == nil {
  779. if !filterBlock.contains(r.filter, dataBH.offset, key) {
  780. frel.Release()
  781. return nil, nil, ErrNotFound
  782. }
  783. frel.Release()
  784. } else if !errors.IsCorrupted(ferr) {
  785. return nil, nil, ferr
  786. }
  787. }
  788. data := r.getDataIter(dataBH, nil, r.verifyChecksum, !ro.GetDontFillCache())
  789. if !data.Seek(key) {
  790. data.Release()
  791. if err = data.Error(); err != nil {
  792. return
  793. }
  794. // The nearest greater-than key is the first key of the next block.
  795. if !index.Next() {
  796. if err = index.Error(); err == nil {
  797. err = ErrNotFound
  798. }
  799. return
  800. }
  801. dataBH, n = decodeBlockHandle(index.Value())
  802. if n == 0 {
  803. r.err = r.newErrCorruptedBH(r.indexBH, "bad data block handle")
  804. return nil, nil, r.err
  805. }
  806. data = r.getDataIter(dataBH, nil, r.verifyChecksum, !ro.GetDontFillCache())
  807. if !data.Next() {
  808. data.Release()
  809. if err = data.Error(); err == nil {
  810. err = ErrNotFound
  811. }
  812. return
  813. }
  814. }
  815. // Key doesn't use block buffer, no need to copy the buffer.
  816. rkey = data.Key()
  817. if !noValue {
  818. if r.bpool == nil {
  819. value = data.Value()
  820. } else {
  821. // Value does use block buffer, and since the buffer will be
  822. // recycled, it need to be copied.
  823. value = append([]byte{}, data.Value()...)
  824. }
  825. }
  826. data.Release()
  827. return
  828. }
  829. // Find finds key/value pair whose key is greater than or equal to the
  830. // given key. It returns ErrNotFound if the table doesn't contain
  831. // such pair.
  832. // If filtered is true then the nearest 'block' will be checked against
  833. // 'filter data' (if present) and will immediately return ErrNotFound if
  834. // 'filter data' indicates that such pair doesn't exist.
  835. //
  836. // The caller may modify the contents of the returned slice as it is its
  837. // own copy.
  838. // It is safe to modify the contents of the argument after Find returns.
  839. func (r *Reader) Find(key []byte, filtered bool, ro *opt.ReadOptions) (rkey, value []byte, err error) {
  840. return r.find(key, filtered, ro, false)
  841. }
  842. // FindKey finds key that is greater than or equal to the given key.
  843. // It returns ErrNotFound if the table doesn't contain such key.
  844. // If filtered is true then the nearest 'block' will be checked against
  845. // 'filter data' (if present) and will immediately return ErrNotFound if
  846. // 'filter data' indicates that such key doesn't exist.
  847. //
  848. // The caller may modify the contents of the returned slice as it is its
  849. // own copy.
  850. // It is safe to modify the contents of the argument after Find returns.
  851. func (r *Reader) FindKey(key []byte, filtered bool, ro *opt.ReadOptions) (rkey []byte, err error) {
  852. rkey, _, err = r.find(key, filtered, ro, true)
  853. return
  854. }
  855. // Get gets the value for the given key. It returns errors.ErrNotFound
  856. // if the table does not contain the key.
  857. //
  858. // The caller may modify the contents of the returned slice as it is its
  859. // own copy.
  860. // It is safe to modify the contents of the argument after Find returns.
  861. func (r *Reader) Get(key []byte, ro *opt.ReadOptions) (value []byte, err error) {
  862. r.mu.RLock()
  863. defer r.mu.RUnlock()
  864. if r.err != nil {
  865. err = r.err
  866. return
  867. }
  868. rkey, value, err := r.find(key, false, ro, false)
  869. if err == nil && r.cmp.Compare(rkey, key) != 0 {
  870. value = nil
  871. err = ErrNotFound
  872. }
  873. return
  874. }
  875. // OffsetOf returns approximate offset for the given key.
  876. //
  877. // It is safe to modify the contents of the argument after Get returns.
  878. func (r *Reader) OffsetOf(key []byte) (offset int64, err error) {
  879. r.mu.RLock()
  880. defer r.mu.RUnlock()
  881. if r.err != nil {
  882. err = r.err
  883. return
  884. }
  885. indexBlock, rel, err := r.readBlockCached(r.indexBH, true, true)
  886. if err != nil {
  887. return
  888. }
  889. defer rel.Release()
  890. index := r.newBlockIter(indexBlock, nil, nil, true)
  891. defer index.Release()
  892. if index.Seek(key) {
  893. dataBH, n := decodeBlockHandle(index.Value())
  894. if n == 0 {
  895. r.err = r.newErrCorruptedBH(r.indexBH, "bad data block handle")
  896. return
  897. }
  898. offset = int64(dataBH.offset)
  899. return
  900. }
  901. err = index.Error()
  902. if err == nil {
  903. offset = r.dataEnd
  904. }
  905. return
  906. }
  907. // Release implements util.Releaser.
  908. // It also close the file if it is an io.Closer.
  909. func (r *Reader) Release() {
  910. r.mu.Lock()
  911. defer r.mu.Unlock()
  912. if closer, ok := r.reader.(io.Closer); ok {
  913. closer.Close()
  914. }
  915. if r.indexBlock != nil {
  916. r.indexBlock.Release()
  917. r.indexBlock = nil
  918. }
  919. if r.filterBlock != nil {
  920. r.filterBlock.Release()
  921. r.filterBlock = nil
  922. }
  923. r.reader = nil
  924. r.cache = nil
  925. r.bpool = nil
  926. r.err = ErrReaderReleased
  927. }
  928. // NewReader creates a new initialized table reader for the file.
  929. // The fi, cache and bpool is optional and can be nil.
  930. //
  931. // The returned table reader instance is safe for concurrent use.
  932. func NewReader(f io.ReaderAt, size int64, fd storage.FileDesc, cache *cache.NamespaceGetter, bpool *util.BufferPool, o *opt.Options) (*Reader, error) {
  933. if f == nil {
  934. return nil, errors.New("leveldb/table: nil file")
  935. }
  936. r := &Reader{
  937. fd: fd,
  938. reader: f,
  939. cache: cache,
  940. bpool: bpool,
  941. o: o,
  942. cmp: o.GetComparer(),
  943. verifyChecksum: o.GetStrict(opt.StrictBlockChecksum),
  944. }
  945. if size < footerLen {
  946. r.err = r.newErrCorrupted(0, size, "table", "too small")
  947. return r, nil
  948. }
  949. footerPos := size - footerLen
  950. var footer [footerLen]byte
  951. if _, err := r.reader.ReadAt(footer[:], footerPos); err != nil && err != io.EOF {
  952. return nil, err
  953. }
  954. if string(footer[footerLen-len(magic):footerLen]) != magic {
  955. r.err = r.newErrCorrupted(footerPos, footerLen, "table-footer", "bad magic number")
  956. return r, nil
  957. }
  958. var n int
  959. // Decode the metaindex block handle.
  960. r.metaBH, n = decodeBlockHandle(footer[:])
  961. if n == 0 {
  962. r.err = r.newErrCorrupted(footerPos, footerLen, "table-footer", "bad metaindex block handle")
  963. return r, nil
  964. }
  965. // Decode the index block handle.
  966. r.indexBH, n = decodeBlockHandle(footer[n:])
  967. if n == 0 {
  968. r.err = r.newErrCorrupted(footerPos, footerLen, "table-footer", "bad index block handle")
  969. return r, nil
  970. }
  971. // Read metaindex block.
  972. metaBlock, err := r.readBlock(r.metaBH, true)
  973. if err != nil {
  974. if errors.IsCorrupted(err) {
  975. r.err = err
  976. return r, nil
  977. }
  978. return nil, err
  979. }
  980. // Set data end.
  981. r.dataEnd = int64(r.metaBH.offset)
  982. // Read metaindex.
  983. metaIter := r.newBlockIter(metaBlock, nil, nil, true)
  984. for metaIter.Next() {
  985. key := string(metaIter.Key())
  986. if !strings.HasPrefix(key, "filter.") {
  987. continue
  988. }
  989. fn := key[7:]
  990. if f0 := o.GetFilter(); f0 != nil && f0.Name() == fn {
  991. r.filter = f0
  992. } else {
  993. for _, f0 := range o.GetAltFilters() {
  994. if f0.Name() == fn {
  995. r.filter = f0
  996. break
  997. }
  998. }
  999. }
  1000. if r.filter != nil {
  1001. filterBH, n := decodeBlockHandle(metaIter.Value())
  1002. if n == 0 {
  1003. continue
  1004. }
  1005. r.filterBH = filterBH
  1006. // Update data end.
  1007. r.dataEnd = int64(filterBH.offset)
  1008. break
  1009. }
  1010. }
  1011. metaIter.Release()
  1012. metaBlock.Release()
  1013. // Cache index and filter block locally, since we don't have global cache.
  1014. if cache == nil {
  1015. r.indexBlock, err = r.readBlock(r.indexBH, true)
  1016. if err != nil {
  1017. if errors.IsCorrupted(err) {
  1018. r.err = err
  1019. return r, nil
  1020. }
  1021. return nil, err
  1022. }
  1023. if r.filter != nil {
  1024. r.filterBlock, err = r.readFilterBlock(r.filterBH)
  1025. if err != nil {
  1026. if !errors.IsCorrupted(err) {
  1027. return nil, err
  1028. }
  1029. // Don't use filter then.
  1030. r.filter = nil
  1031. }
  1032. }
  1033. }
  1034. return r, nil
  1035. }