Платформа ЦРНП "Мирокод" для разработки проектов
https://git.mirocod.ru
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.
85 lines
1.8 KiB
85 lines
1.8 KiB
package themis |
|
|
|
import ( |
|
"bytes" |
|
"encoding/binary" |
|
|
|
"github.com/ngaut/log" |
|
"github.com/pingcap/go-hbase" |
|
) |
|
|
|
type ThemisScanner struct { |
|
scan *hbase.Scan |
|
txn *themisTxn |
|
tbl []byte |
|
} |
|
|
|
func newThemisScanner(tbl []byte, txn *themisTxn, batchSize int, c hbase.HBaseClient) *ThemisScanner { |
|
s := hbase.NewScan(tbl, batchSize, c) |
|
// add start ts |
|
b := bytes.NewBuffer(nil) |
|
binary.Write(b, binary.BigEndian, txn.startTs) |
|
s.AddAttr("_themisTransationStartTs_", b.Bytes()) |
|
return &ThemisScanner{ |
|
scan: s, |
|
txn: txn, |
|
tbl: tbl, |
|
} |
|
} |
|
|
|
func (s *ThemisScanner) setStartRow(start []byte) { |
|
s.scan.StartRow = start |
|
} |
|
|
|
func (s *ThemisScanner) setStopRow(stop []byte) { |
|
s.scan.StopRow = stop |
|
} |
|
|
|
func (s *ThemisScanner) SetTimeRange(tsRangeFrom uint64, tsRangeTo uint64) { |
|
s.scan.TsRangeFrom = tsRangeFrom |
|
s.scan.TsRangeTo = tsRangeTo |
|
} |
|
|
|
func (s *ThemisScanner) SetMaxVersions(maxVersions uint32) { |
|
s.scan.MaxVersions = maxVersions |
|
} |
|
|
|
func (s *ThemisScanner) createGetFromScan(row []byte) *hbase.Get { |
|
return s.scan.CreateGetFromScan(row) |
|
} |
|
|
|
func (s *ThemisScanner) Next() *hbase.ResultRow { |
|
r := s.scan.Next() |
|
if r == nil { |
|
return nil |
|
} |
|
// if we encounter conflict locks, we need to clean lock for this row and read again |
|
if isLockResult(r) { |
|
g := s.createGetFromScan(r.Row) |
|
r, err := s.txn.tryToCleanLockAndGetAgain(s.tbl, g, r.SortedColumns) |
|
if err != nil { |
|
log.Error(err) |
|
return nil |
|
} |
|
// empty result indicates the current row has been erased, we should get next row |
|
if r == nil { |
|
return s.Next() |
|
} else { |
|
return r |
|
} |
|
} |
|
return r |
|
} |
|
|
|
func (s *ThemisScanner) Closed() bool { |
|
return s.scan.Closed() |
|
} |
|
|
|
func (s *ThemisScanner) Close() { |
|
if !s.scan.Closed() { |
|
// TODO: handle error, now just log |
|
if err := s.scan.Close(); err != nil { |
|
log.Warnf("scanner close error, scan: %s, error: %v", s.scan, err) |
|
} |
|
} |
|
}
|
|
|