scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1072 from molecula/uni-wal

Browse source 
Refactor RBF to use a single WAL file
This commit is contained in:
jaten-molecula 2020-11-11 15:44:45 -06:00 • committed by GitHub
commit cb73c71e02
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
19 changed files with 469 additions and 1256 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
api.go
View file

@ -463,7 +463,6 @@ func (api *API) ImportRoaring(ctx context.Context, indexName, fieldName string,
}
qcx := api.Txf().NewQcx()
qcx.Direct = req.Direct
defer qcx.Abort()
nodes := api.cluster.shardNodes(indexName, shard)

11
cmd/slurp/slurp.go
View file

@ -45,7 +45,6 @@ type stateMachine struct {
state string
client *http.InternalClient
start time.Time
direct bool
profile string
host string
@ -132,8 +131,7 @@ func (r *stateMachine) NewHeader(h *tar.Header, tr *tar.Reader) error {
func (r *stateMachine) Upload() error {
if len(r.viewData) > 0 {
request := &pilosa.ImportRoaringRequest{
Views: r.viewData,
Direct: r.direct,
Views: r.viewData,
}
uri := GetImportRoaringURI(r.lastIndex, r.lastShard)
err := r.client.ImportRoaring(context.Background(), uri, r.lastIndex, r.lastField, r.lastShard, false, request)
@ -145,7 +143,7 @@ func (r *stateMachine) Upload() error {
return nil
}
func UploadTar(srcFile string, direct bool, client *http.InternalClient, profile, host string) error {
func UploadTar(srcFile string, client *http.InternalClient, profile, host string) error {
f, err := os.Open(srcFile)
if err != nil {
@ -165,7 +163,6 @@ func UploadTar(srcFile string, direct bool, client *http.InternalClient, profile
runner := &stateMachine{
viewData: make(map[string][]byte),
start: time.Now(),
direct: direct,
profile: profile,
host: host,
}
@ -187,11 +184,9 @@ func UploadTar(srcFile string, direct bool, client *http.InternalClient, profile
func main() {
var host string
var direct bool
var profile string
var tarSrcPath string
flag.StringVar(&host, "host", "127.0.0.1:10101", "host to import into")
flag.BoolVar(&direct, "direct", false, "direct write to database (unsafe)")
flag.StringVar(&profile, "profile", "", "profile and save a cpu profile of the import to this file")
flag.StringVar(&tarSrcPath, "src", "q2.tar.gz", "data to import")
flag.Parse()
@ -209,7 +204,7 @@ func main() {
t0 := time.Now()
println("uploading", tarSrcPath)
panicOn(UploadTar(tarSrcPath, direct, c, profile, host))
panicOn(UploadTar(tarSrcPath, c, profile, host))
vv("total elapsed '%v'", time.Since(t0))
}

2
encoding/proto/proto.go
View file

@ -460,7 +460,6 @@ func (s Serializer) encodeImportRoaringRequest(m *pilosa.ImportRoaringRequest) *
Action: m.Action,
Block: uint64(m.Block),
Views: views,
Direct: m.Direct,
}
}
@ -1235,7 +1234,6 @@ func (s Serializer) decodeImportRoaringRequest(pb *internal.ImportRoaringRequest
m.Views = views
m.IndexCreatedAt = pb.IndexCreatedAt
m.FieldCreatedAt = pb.FieldCreatedAt
m.Direct = pb.Direct
}
func (s Serializer) decodeImportColumnAttrsRequest(pb *internal.ImportColumnAttrsRequest, m *pilosa.ImportColumnAttrsRequest) {

1
handler.go
View file

@ -242,7 +242,6 @@ type ImportRoaringRequest struct {
Action string // [set, clear, overwrite]
Block int
Views map[string][]byte
Direct bool
}
// ValidateWithTimestamp ensures that the payload of the request is valid.

247
internal/public.pb.go
View file

@ -2402,7 +2402,6 @@ type ImportRoaringRequest struct {
Block uint64 `protobuf:"varint,4,opt,name=Block,proto3" json:"Block,omitempty"`
IndexCreatedAt int64 `protobuf:"varint,5,opt,name=IndexCreatedAt,proto3" json:"IndexCreatedAt,omitempty"`
FieldCreatedAt int64 `protobuf:"varint,6,opt,name=FieldCreatedAt,proto3" json:"FieldCreatedAt,omitempty"`
Direct bool `protobuf:"varint,7,opt,name=Direct,proto3" json:"Direct,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
@ -2483,13 +2482,6 @@ func (m *ImportRoaringRequest) GetFieldCreatedAt() int64 {
return 0
}
func (m *ImportRoaringRequest) GetDirect() bool {
if m != nil {
return m.Direct
}
return false
}
type ImportColumnAttrsRequest struct {
Index string `protobuf:"bytes,1,opt,name=Index,proto3" json:"Index,omitempty"`
Shard int64 `protobuf:"varint,2,opt,name=Shard,proto3" json:"Shard,omitempty"`
@ -2619,111 +2611,111 @@ func init() {
func init() { proto.RegisterFile("public.proto", fileDescriptor_413a91106d7bcce8) }
var fileDescriptor_413a91106d7bcce8 = []byte{
// 1663 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x58, 0xcd, 0x6e, 0xdb, 0xce,
0x11, 0x37, 0x45, 0xea, 0x6b, 0x24, 0xfb, 0xef, 0x6c, 0x94, 0x94, 0x48, 0x1d, 0x47, 0x20, 0xdc,
// 1653 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x58, 0x4f, 0x6f, 0xdb, 0xca,
0x11, 0x37, 0x45, 0xca, 0x92, 0x46, 0xb2, 0xe3, 0x6c, 0x94, 0x94, 0x48, 0x1d, 0x47, 0x20, 0xdc,
0x46, 0x2d, 0x0a, 0x07, 0x4e, 0x93, 0x20, 0x97, 0xb6, 0xb1, 0x23, 0xa7, 0x26, 0x52, 0xbb, 0xe9,
0xca, 0x70, 0x6e, 0x05, 0x68, 0x69, 0xeb, 0x10, 0xa5, 0x44, 0x95, 0xa2, 0x22, 0xfb, 0x52, 0xa0,
0xcf, 0x90, 0x4b, 0x1f, 0xa1, 0xcf, 0xd1, 0x4b, 0x7b, 0xec, 0xb1, 0x40, 0x2f, 0x45, 0xfa, 0x18,
0xe9, 0xa1, 0x98, 0x59, 0xae, 0x76, 0x49, 0xd1, 0x8e, 0x11, 0xf4, 0xb6, 0xf3, 0xb1, 0xb3, 0x33,
0xbf, 0x99, 0x9d, 0x1d, 0x12, 0xda, 0xd3, 0xf9, 0x79, 0x14, 0x0e, 0x77, 0xa7, 0x49, 0x9c, 0xc6,
0xac, 0x11, 0x4e, 0x52, 0x91, 0x4c, 0x82, 0xc8, 0x9b, 0x81, 0xcd, 0xe3, 0x05, 0x73, 0xa1, 0xfe,
0x3a, 0x8e, 0xe6, 0xe3, 0xc9, 0xcc, 0xb5, 0xba, 0x76, 0xcf, 0xe1, 0x8a, 0x64, 0x0c, 0x9c, 0xb7,
0xe2, 0x6a, 0xe6, 0xda, 0x5d, 0xbb, 0xd7, 0xe4, 0xb4, 0x66, 0x3b, 0x50, 0xdd, 0x4f, 0xd3, 0x64,
0xe6, 0x56, 0xba, 0x76, 0xaf, 0xf5, 0x74, 0x63, 0x57, 0x99, 0xdb, 0x45, 0x36, 0x97, 0x42, 0xb4,
0xc9, 0xe3, 0x20, 0x09, 0x27, 0x17, 0xae, 0xd3, 0xb5, 0x7a, 0x6d, 0xae, 0x48, 0xef, 0x18, 0x9a,
0x83, 0xf0, 0x62, 0x22, 0x46, 0x78, 0xf4, 0x23, 0xb0, 0xdf, 0xc5, 0x78, 0xac, 0xd5, 0x6b, 0x3d,
0x5d, 0xd7, 0xa6, 0x78, 0xbc, 0xe0, 0x28, 0x41, 0x85, 0x13, 0x71, 0xe1, 0x56, 0x4a, 0x15, 0x4e,
0xc4, 0x85, 0xf7, 0x12, 0x36, 0x78, 0xbc, 0xf0, 0x47, 0x62, 0x92, 0x86, 0xbf, 0x0b, 0x45, 0x42,
0x4e, 0xf3, 0x78, 0xa1, 0x62, 0xa1, 0xf5, 0x32, 0x90, 0x8a, 0x0e, 0xc4, 0x7b, 0x00, 0x35, 0xbf,
0xff, 0xab, 0x70, 0x96, 0xb2, 0x4d, 0xb0, 0xfd, 0xbe, 0xda, 0x80, 0x4b, 0xcf, 0x87, 0x3b, 0x87,
0x97, 0x69, 0x12, 0x0c, 0x53, 0x31, 0xf2, 0xfb, 0x12, 0x0e, 0xb6, 0x01, 0x15, 0xbf, 0x4f, 0xbe,
0x3a, 0xbc, 0xe2, 0xf7, 0xd9, 0x0e, 0x38, 0x67, 0x41, 0xa4, 0x80, 0xd8, 0xd4, 0xce, 0x49, 0xb3,
0x9c, 0xa4, 0xde, 0x79, 0xce, 0xd4, 0x71, 0x90, 0x26, 0xe1, 0x25, 0xbb, 0x0f, 0xb5, 0x37, 0xa1,
0x88, 0x46, 0xf2, 0xd0, 0x26, 0xcf, 0x28, 0xf6, 0x5c, 0xa7, 0x42, 0x5a, 0xfd, 0xbe, 0xb6, 0xba,
0xe2, 0xd0, 0x32, 0x4f, 0xde, 0x43, 0xa8, 0xbf, 0x15, 0x57, 0x14, 0x8b, 0x8a, 0xd4, 0x32, 0x22,
0xfd, 0x97, 0x05, 0x77, 0x97, 0xbb, 0x4f, 0x83, 0xf3, 0x48, 0x9c, 0x05, 0xd1, 0x5c, 0xb0, 0x1d,
0x15, 0xb7, 0x55, 0xe6, 0xff, 0xd1, 0x1a, 0x61, 0xc1, 0x1e, 0x2f, 0xb1, 0x43, 0xb5, 0x3b, 0x5a,
0x2d, 0x3b, 0xf2, 0x68, 0x2d, 0xab, 0x8c, 0x2d, 0x68, 0x1c, 0x0c, 0x7c, 0x32, 0xed, 0xda, 0x5d,
0xab, 0x67, 0x1f, 0xad, 0xf1, 0x25, 0x87, 0x3d, 0x80, 0xfa, 0xf1, 0x3c, 0x15, 0x97, 0x7e, 0x9f,
0x2a, 0xc2, 0x39, 0x5a, 0xe3, 0x8a, 0x81, 0x3b, 0x69, 0xf9, 0x56, 0x5c, 0xb9, 0xd5, 0xae, 0xd5,
0x6b, 0xe2, 0x4e, 0xc5, 0x61, 0x1d, 0x70, 0x0e, 0xe2, 0x38, 0x72, 0x6b, 0x5d, 0xab, 0xd7, 0xc0,
0xd3, 0x90, 0x3a, 0xa8, 0x43, 0x95, 0x0c, 0x7b, 0x7f, 0x84, 0x4e, 0x3e, 0xb8, 0x2c, 0x5d, 0x0c,
0x6c, 0xb4, 0x67, 0x65, 0xf6, 0x90, 0x60, 0x9b, 0x94, 0xc2, 0x4a, 0x76, 0x3e, 0x26, 0xf1, 0x39,
0xd4, 0xc8, 0x8c, 0x2c, 0xf2, 0xd6, 0xd3, 0x87, 0x25, 0x80, 0x6b, 0xc8, 0x78, 0xa6, 0x7c, 0xd0,
0x24, 0xc4, 0x7f, 0x9d, 0xf8, 0x7d, 0xef, 0x67, 0x45, 0x70, 0x29, 0x97, 0x98, 0x88, 0x93, 0x60,
0x2c, 0xe4, 0xf9, 0x9c, 0xd6, 0xc8, 0x3b, 0xbd, 0x9a, 0x0a, 0x72, 0xa0, 0xc9, 0x69, 0xed, 0xfd,
0xc9, 0x82, 0x8d, 0xfc, 0x7e, 0xf4, 0xc9, 0xa8, 0x8e, 0x1b, 0x7c, 0x22, 0xad, 0x65, 0xf1, 0xbc,
0x2c, 0x16, 0xcf, 0xf6, 0x75, 0xfb, 0x8a, 0xf5, 0xf3, 0x73, 0x70, 0xde, 0x05, 0x61, 0xb2, 0x52,
0xe1, 0x9b, 0x12, 0x42, 0x9b, 0xdc, 0xb5, 0x65, 0x2e, 0xaa, 0xaf, 0xe3, 0xf9, 0x24, 0x95, 0x18,
0x72, 0x49, 0x78, 0x87, 0xd0, 0xc4, 0xfd, 0x32, 0x70, 0x4f, 0x1a, 0xcb, 0xca, 0xca, 0xe8, 0x0f,
0xc8, 0xe5, 0xf2, 0xa0, 0x0e, 0x54, 0x49, 0x39, 0x43, 0x42, 0x12, 0xde, 0x11, 0x00, 0x4a, 0x67,
0xd2, 0xce, 0x0e, 0x54, 0x89, 0xca, 0x40, 0x28, 0x1a, 0x92, 0xc2, 0x6b, 0x2c, 0x3d, 0x84, 0xaa,
0x3f, 0x49, 0x5f, 0x3c, 0x43, 0xb1, 0x2c, 0x48, 0xf4, 0xc6, 0xe6, 0x59, 0xc9, 0xcc, 0xa1, 0x21,
0xa1, 0x8b, 0x17, 0xda, 0x80, 0x65, 0x18, 0x40, 0x2e, 0xb6, 0x95, 0xbe, 0x8a, 0x93, 0x08, 0xbc,
0xb6, 0x3c, 0x5e, 0x68, 0x48, 0x32, 0x8a, 0xfd, 0x40, 0x9d, 0xe2, 0x50, 0xcc, 0xdf, 0x19, 0x57,
0x09, 0xbd, 0x50, 0xc7, 0xfe, 0x16, 0xe0, 0x97, 0x49, 0x3c, 0x9f, 0x12, 0x68, 0xac, 0x07, 0x55,
0xa2, 0xb2, 0xf8, 0x98, 0xde, 0xa4, 0x7c, 0xe3, 0x52, 0xa1, 0x1c, 0x74, 0x4c, 0xce, 0x60, 0x3e,
0x96, 0x37, 0x8d, 0xe3, 0x12, 0x4b, 0xa9, 0x71, 0x16, 0x44, 0x4b, 0xf1, 0x59, 0x10, 0x65, 0x71,
0xe3, 0x32, 0x6f, 0xc6, 0x56, 0x66, 0x1e, 0x40, 0xe3, 0x4d, 0x14, 0x07, 0x29, 0x2a, 0xa3, 0x2d,
0x8b, 0x2f, 0x69, 0xb6, 0x07, 0xd0, 0x17, 0xc3, 0x70, 0x1c, 0x44, 0x28, 0x75, 0x8a, 0x0d, 0x20,
0x93, 0x71, 0x43, 0xc9, 0x7b, 0x0e, 0xf5, 0x8c, 0x2a, 0xc7, 0x1e, 0xb9, 0x83, 0x61, 0x10, 0x09,
0xe5, 0x05, 0x11, 0xde, 0x7b, 0x58, 0x97, 0xc5, 0x88, 0xcf, 0xc7, 0x40, 0xa4, 0xb7, 0x28, 0xc5,
0x5b, 0x3d, 0x44, 0xde, 0x5f, 0x2c, 0x70, 0x70, 0xa5, 0x0c, 0x58, 0xda, 0x80, 0x79, 0x1b, 0x1d,
0x79, 0x1b, 0x59, 0x17, 0x5a, 0x83, 0x14, 0xdf, 0x29, 0xdd, 0xc6, 0x9a, 0xdc, 0x64, 0x21, 0x5e,
0xfe, 0x24, 0xd5, 0xe9, 0xb6, 0xf9, 0x92, 0x66, 0x5b, 0xd0, 0xc4, 0xde, 0x24, 0x85, 0xd8, 0xc8,
0x1a, 0x5c, 0x33, 0xd8, 0x36, 0x80, 0x42, 0x76, 0x2e, 0xa8, 0x9b, 0x59, 0xdc, 0xe0, 0x78, 0x4f,
0xa0, 0x8e, 0x9e, 0x1e, 0x07, 0x53, 0x1d, 0x9b, 0x75, 0x53, 0x6c, 0x5f, 0x2c, 0x68, 0xff, 0x66,
0x2e, 0x92, 0x2b, 0x2e, 0xfe, 0x30, 0x17, 0xb3, 0x14, 0xb1, 0x25, 0x5a, 0xd5, 0x32, 0x11, 0x58,
0xb5, 0x83, 0x0f, 0x41, 0x32, 0x92, 0x48, 0x39, 0x3c, 0xa3, 0x30, 0x56, 0x8d, 0xf9, 0x8c, 0x62,
0x6d, 0x70, 0x93, 0x45, 0xf5, 0x2e, 0xc6, 0x71, 0xaa, 0x82, 0xc9, 0x28, 0xd6, 0x83, 0xef, 0x0e,
0x2f, 0x87, 0xd1, 0x7c, 0x24, 0x78, 0xbc, 0x90, 0xbb, 0xa9, 0x39, 0xf3, 0x22, 0x9b, 0xfd, 0x10,
0x9b, 0x1b, 0xb1, 0x54, 0x6b, 0xaa, 0x93, 0x62, 0x81, 0xcb, 0xf6, 0xa0, 0x7d, 0x38, 0x3e, 0x17,
0xa3, 0x91, 0x18, 0xf5, 0x83, 0x34, 0x70, 0x1b, 0x14, 0x77, 0xe1, 0xc1, 0xcf, 0xa9, 0x78, 0x9f,
0x2c, 0x58, 0xcf, 0xa2, 0x9f, 0x4d, 0xe3, 0xc9, 0x4c, 0x60, 0x8a, 0x0f, 0x93, 0x44, 0xa5, 0xf8,
0x30, 0x49, 0xd8, 0x13, 0xa8, 0x73, 0x31, 0x9b, 0x47, 0xa9, 0xaa, 0x92, 0x7b, 0xda, 0xa2, 0xda,
0x3b, 0x8f, 0x52, 0xae, 0xb4, 0xd8, 0x2f, 0x60, 0x23, 0x57, 0x87, 0xea, 0x59, 0xf8, 0x9e, 0xde,
0x97, 0x93, 0xf3, 0x82, 0xba, 0xf7, 0xc5, 0x81, 0x96, 0x61, 0x79, 0x59, 0x64, 0x88, 0xcf, 0x7a,
0x56, 0x64, 0x8f, 0x68, 0xee, 0xba, 0x66, 0xea, 0xc1, 0x9e, 0xd4, 0x06, 0xeb, 0x24, 0x2b, 0x4b,
0xeb, 0x44, 0x37, 0x42, 0xfb, 0xa6, 0x46, 0x88, 0x53, 0xdc, 0x87, 0x60, 0x72, 0x21, 0x46, 0x54,
0x96, 0x0d, 0xae, 0x48, 0xb6, 0xab, 0xbb, 0x02, 0xe5, 0x31, 0xd7, 0x6b, 0x94, 0x84, 0xeb, 0xce,
0x21, 0xbb, 0x1c, 0x4e, 0x06, 0x75, 0x59, 0x2f, 0x92, 0x62, 0x2f, 0xa0, 0xa5, 0xdb, 0xd7, 0x2c,
0x4b, 0x51, 0x47, 0x9b, 0xd2, 0x42, 0x6e, 0x2a, 0xb2, 0x57, 0xc5, 0x11, 0xcd, 0x6d, 0x92, 0x17,
0x6e, 0x2e, 0x72, 0x43, 0xce, 0x8b, 0x23, 0xdd, 0x9e, 0x31, 0x33, 0xba, 0x40, 0x9b, 0xef, 0xea,
0xcd, 0x4b, 0x11, 0x37, 0x26, 0xcb, 0x67, 0xe6, 0x5b, 0xe2, 0xb6, 0x68, 0x4f, 0x27, 0x8f, 0x9c,
0x94, 0x71, 0xf3, 0xcd, 0xd9, 0x33, 0x1e, 0x32, 0xb7, 0x5d, 0x3c, 0x68, 0x29, 0xe2, 0xc6, 0x73,
0xe7, 0x97, 0xcc, 0x77, 0xee, 0x3a, 0x6d, 0x2d, 0x1f, 0xde, 0xa4, 0x0a, 0x2f, 0x99, 0x0a, 0x5f,
0x15, 0x27, 0x01, 0x77, 0xa3, 0x08, 0x54, 0x5e, 0xce, 0x0b, 0xfa, 0xde, 0xdf, 0x2a, 0xb0, 0xee,
0x8f, 0xa7, 0x71, 0x92, 0x1a, 0x2d, 0xc1, 0x9f, 0x8c, 0xc4, 0xa5, 0x6a, 0x09, 0x44, 0x94, 0xbf,
0x9a, 0xd4, 0x9a, 0xb1, 0x35, 0x50, 0x2b, 0x70, 0xb8, 0x24, 0x8c, 0x72, 0x70, 0x72, 0xe5, 0xb0,
0x05, 0x4d, 0x59, 0xfb, 0x28, 0xaa, 0x92, 0x48, 0x33, 0xe4, 0x07, 0xc0, 0x82, 0x06, 0xc7, 0x3a,
0x8d, 0xa2, 0x8a, 0xc4, 0x36, 0x28, 0xd5, 0x48, 0xd8, 0x20, 0xa1, 0xc1, 0x41, 0xf9, 0x69, 0x38,
0x16, 0xb3, 0x34, 0x18, 0x4f, 0xb1, 0xaf, 0xd8, 0x3d, 0x9b, 0x1b, 0x1c, 0x6c, 0x29, 0x14, 0xc4,
0xeb, 0x44, 0x04, 0xa9, 0x18, 0xed, 0xa7, 0x54, 0x4e, 0x36, 0x2f, 0x70, 0x51, 0x8f, 0xc2, 0xd2,
0x7a, 0x20, 0xf5, 0xf2, 0x5c, 0x7a, 0x16, 0x23, 0x11, 0x24, 0x54, 0x24, 0x0d, 0x2e, 0x09, 0xef,
0x9f, 0x15, 0x60, 0x12, 0x49, 0x39, 0xf8, 0xfd, 0xdf, 0xe0, 0xbc, 0x19, 0xb6, 0x3c, 0x38, 0xf5,
0x15, 0x70, 0xee, 0x2f, 0xc7, 0x55, 0x09, 0x4c, 0x46, 0x61, 0x2f, 0xd7, 0x2f, 0x89, 0x44, 0xd5,
0xe2, 0x26, 0x8b, 0x79, 0xd0, 0x36, 0x9e, 0x31, 0xbc, 0x83, 0x68, 0x3b, 0xc7, 0x2b, 0x81, 0x16,
0x6e, 0x09, 0x6d, 0xeb, 0x66, 0x68, 0xdb, 0x26, 0xb4, 0x9f, 0x2c, 0x68, 0xef, 0xa7, 0xf1, 0x38,
0x1c, 0x72, 0x31, 0x8c, 0x93, 0xd1, 0xf5, 0xa0, 0x4a, 0xf8, 0x2a, 0x26, 0x7c, 0xbb, 0x60, 0xfb,
0x1f, 0x93, 0xac, 0x15, 0x6e, 0x19, 0x83, 0xd6, 0x4a, 0xae, 0x38, 0x2a, 0xb2, 0xc7, 0x50, 0xf1,
0x13, 0xaa, 0xdc, 0x5c, 0x13, 0xcf, 0x5d, 0x12, 0x5e, 0xf1, 0x13, 0xef, 0x27, 0xd0, 0x91, 0x4e,
0x29, 0x51, 0xf6, 0xa8, 0x74, 0xa0, 0x7a, 0x98, 0x24, 0xb1, 0x7a, 0x56, 0x24, 0xe1, 0x5d, 0x42,
0xe7, 0x34, 0x09, 0x26, 0xb3, 0x28, 0x48, 0x05, 0x26, 0xe6, 0x5b, 0xea, 0xa3, 0xec, 0xeb, 0xba,
0x0b, 0xad, 0x93, 0x38, 0x7d, 0x9f, 0x84, 0x29, 0xdd, 0x7f, 0xd9, 0xc9, 0x4d, 0x96, 0xf7, 0x23,
0xb8, 0x57, 0x38, 0x59, 0xbf, 0x7e, 0x58, 0x52, 0xb6, 0xfe, 0x8a, 0x1d, 0xc0, 0xdd, 0xa5, 0xaa,
0xdf, 0xff, 0x26, 0x1f, 0x57, 0x8d, 0xfe, 0xd8, 0x88, 0x9c, 0x8c, 0x66, 0xc7, 0x97, 0x44, 0xe3,
0x1d, 0x80, 0x9b, 0xa1, 0x29, 0x3f, 0xfe, 0x33, 0x0f, 0xce, 0x42, 0xb1, 0xb8, 0xee, 0xfb, 0x88,
0x5e, 0xff, 0x0a, 0xfd, 0x32, 0xa0, 0xb5, 0xf7, 0x5f, 0x0b, 0x3a, 0x65, 0x46, 0x74, 0x71, 0x59,
0x46, 0x71, 0xb1, 0x97, 0x50, 0xfd, 0x18, 0x8a, 0x85, 0x7a, 0xef, 0xbd, 0x95, 0x94, 0xaf, 0x78,
0xc2, 0xe5, 0x06, 0xbc, 0x5a, 0xfb, 0xc3, 0x34, 0x8c, 0x27, 0x6a, 0xb8, 0x97, 0x14, 0x9e, 0x73,
0x10, 0xc5, 0xc3, 0xdf, 0xcb, 0xcf, 0x56, 0x2e, 0x89, 0x92, 0xab, 0x52, 0xbd, 0xe5, 0x55, 0xa9,
0x95, 0x5e, 0x95, 0xfb, 0x50, 0xeb, 0x87, 0x89, 0x18, 0xa6, 0xd9, 0x80, 0x94, 0x51, 0xde, 0x5f,
0x2d, 0x85, 0xa1, 0x31, 0x98, 0x7d, 0x35, 0x93, 0xfa, 0xe2, 0xd8, 0xea, 0xe2, 0xb8, 0x72, 0xba,
0xd4, 0x43, 0xb4, 0x22, 0x71, 0xa2, 0xc5, 0x25, 0xfd, 0xcb, 0x70, 0x28, 0x7b, 0x4b, 0xfa, 0x2b,
0xdd, 0x6a, 0x15, 0x84, 0x5a, 0x19, 0x08, 0x07, 0x9b, 0x7f, 0xff, 0xbc, 0x6d, 0xfd, 0xe3, 0xf3,
0xb6, 0xf5, 0xef, 0xcf, 0xdb, 0xd6, 0x9f, 0xff, 0xb3, 0xbd, 0x76, 0x5e, 0xa3, 0x7f, 0x51, 0x3f,
0xfd, 0x5f, 0x00, 0x00, 0x00, 0xff, 0xff, 0xf9, 0x65, 0xe6, 0x0c, 0x9b, 0x12, 0x00, 0x00,
0x9f, 0x21, 0x97, 0x7e, 0x84, 0x7e, 0x8e, 0x5e, 0xda, 0x63, 0x8f, 0x05, 0xde, 0xe5, 0x21, 0xef,
0x7d, 0x8b, 0x5c, 0x1e, 0x66, 0x96, 0xab, 0x5d, 0x52, 0xb4, 0x63, 0x04, 0xef, 0xb6, 0xf3, 0x67,
0x67, 0x67, 0x7e, 0x33, 0x3b, 0x3b, 0x24, 0xb4, 0x26, 0xb3, 0xb3, 0x28, 0x1c, 0xec, 0x4c, 0x92,
0x38, 0x8d, 0x59, 0x3d, 0x1c, 0xa7, 0x22, 0x19, 0x07, 0x91, 0x37, 0x05, 0x9b, 0xc7, 0x73, 0xe6,
0x42, 0xed, 0x55, 0x1c, 0xcd, 0x46, 0xe3, 0xa9, 0x6b, 0x75, 0xec, 0xae, 0xc3, 0x15, 0xc9, 0x18,
0x38, 0x6f, 0xc4, 0xe5, 0xd4, 0xb5, 0x3b, 0x76, 0xb7, 0xc1, 0x69, 0xcd, 0xb6, 0xa1, 0xba, 0x97,
0xa6, 0xc9, 0xd4, 0xad, 0x74, 0xec, 0x6e, 0xf3, 0xc9, 0xfa, 0x8e, 0x32, 0xb7, 0x83, 0x6c, 0x2e,
0x85, 0x68, 0x93, 0xc7, 0x41, 0x12, 0x8e, 0xcf, 0x5d, 0xa7, 0x63, 0x75, 0x5b, 0x5c, 0x91, 0xde,
0x11, 0x34, 0xfa, 0xe1, 0xf9, 0x58, 0x0c, 0xf1, 0xe8, 0x87, 0x60, 0xbf, 0x8d, 0xf1, 0x58, 0xab,
0xdb, 0x7c, 0xb2, 0xa6, 0x4d, 0xf1, 0x78, 0xce, 0x51, 0x82, 0x0a, 0xc7, 0xe2, 0xdc, 0xad, 0x94,
0x2a, 0x1c, 0x8b, 0x73, 0xef, 0x05, 0xac, 0xf3, 0x78, 0xee, 0x0f, 0xc5, 0x38, 0x0d, 0xff, 0x12,
0x8a, 0x84, 0x9c, 0xe6, 0xf1, 0x5c, 0xc5, 0x42, 0xeb, 0x45, 0x20, 0x15, 0x1d, 0x88, 0x77, 0x1f,
0x56, 0xfd, 0xde, 0x1f, 0xc2, 0x69, 0xca, 0x36, 0xc0, 0xf6, 0x7b, 0x6a, 0x03, 0x2e, 0x3d, 0x1f,
0x6e, 0x1f, 0x5c, 0xa4, 0x49, 0x30, 0x48, 0xc5, 0xd0, 0xef, 0x49, 0x38, 0xd8, 0x3a, 0x54, 0xfc,
0x1e, 0xf9, 0xea, 0xf0, 0x8a, 0xdf, 0x63, 0xdb, 0xe0, 0x9c, 0x06, 0x91, 0x02, 0x62, 0x43, 0x3b,
0x27, 0xcd, 0x72, 0x92, 0x7a, 0x67, 0x39, 0x53, 0x47, 0x41, 0x9a, 0x84, 0x17, 0xec, 0x1e, 0xac,
0xbe, 0x0e, 0x45, 0x34, 0x94, 0x87, 0x36, 0x78, 0x46, 0xb1, 0x67, 0x3a, 0x15, 0xd2, 0xea, 0x4f,
0xb5, 0xd5, 0x25, 0x87, 0x16, 0x79, 0xf2, 0x1e, 0x40, 0xed, 0x8d, 0xb8, 0xa4, 0x58, 0x54, 0xa4,
0x96, 0x11, 0xe9, 0x37, 0x16, 0xdc, 0x59, 0xec, 0x3e, 0x09, 0xce, 0x22, 0x71, 0x1a, 0x44, 0x33,
0xc1, 0xb6, 0x55, 0xdc, 0x56, 0x99, 0xff, 0x87, 0x2b, 0x84, 0x05, 0x7b, 0xb4, 0xc0, 0x0e, 0xd5,
0x6e, 0x6b, 0xb5, 0xec, 0xc8, 0xc3, 0x95, 0xac, 0x32, 0x36, 0xa1, 0xbe, 0xdf, 0xf7, 0xc9, 0xb4,
0x6b, 0x77, 0xac, 0xae, 0x7d, 0xb8, 0xc2, 0x17, 0x1c, 0x76, 0x1f, 0x6a, 0x47, 0xb3, 0x54, 0x5c,
0xf8, 0x3d, 0xaa, 0x08, 0xe7, 0x70, 0x85, 0x2b, 0x06, 0xee, 0xa4, 0xe5, 0x1b, 0x71, 0xe9, 0x56,
0x3b, 0x56, 0xb7, 0x81, 0x3b, 0x15, 0x87, 0xb5, 0xc1, 0xd9, 0x8f, 0xe3, 0xc8, 0x5d, 0xed, 0x58,
0xdd, 0x3a, 0x9e, 0x86, 0xd4, 0x7e, 0x0d, 0xaa, 0x64, 0xd8, 0xfb, 0x3b, 0xb4, 0xf3, 0xc1, 0x65,
0xe9, 0x62, 0x60, 0xa3, 0x3d, 0x2b, 0xb3, 0x87, 0x04, 0xdb, 0xa0, 0x14, 0x56, 0xb2, 0xf3, 0x31,
0x89, 0xcf, 0x60, 0x95, 0xcc, 0xc8, 0x22, 0x6f, 0x3e, 0x79, 0x50, 0x02, 0xb8, 0x86, 0x8c, 0x67,
0xca, 0xfb, 0x0d, 0x42, 0xfc, 0x8f, 0x89, 0xdf, 0xf3, 0x7e, 0x53, 0x04, 0x97, 0x72, 0x89, 0x89,
0x38, 0x0e, 0x46, 0x42, 0x9e, 0xcf, 0x69, 0x8d, 0xbc, 0x93, 0xcb, 0x89, 0x20, 0x07, 0x1a, 0x9c,
0xd6, 0xde, 0x3f, 0x2c, 0x58, 0xcf, 0xef, 0x47, 0x9f, 0x8c, 0xea, 0xb8, 0xc6, 0x27, 0xd2, 0x5a,
0x14, 0xcf, 0x8b, 0x62, 0xf1, 0x6c, 0x5d, 0xb5, 0xaf, 0x58, 0x3f, 0xbf, 0x05, 0xe7, 0x6d, 0x10,
0x26, 0x4b, 0x15, 0xbe, 0x21, 0x21, 0xb4, 0xc9, 0x5d, 0x5b, 0xe6, 0xa2, 0xfa, 0x2a, 0x9e, 0x8d,
0x53, 0x89, 0x21, 0x97, 0x84, 0x77, 0x00, 0x0d, 0xdc, 0x2f, 0x03, 0xf7, 0xa4, 0xb1, 0xac, 0xac,
0x8c, 0xfe, 0x80, 0x5c, 0x2e, 0x0f, 0x6a, 0x43, 0x95, 0x94, 0x33, 0x24, 0x24, 0xe1, 0x1d, 0x02,
0xa0, 0x74, 0x2a, 0xed, 0x6c, 0x43, 0x95, 0xa8, 0x0c, 0x84, 0xa2, 0x21, 0x29, 0xbc, 0xc2, 0xd2,
0x03, 0xa8, 0xfa, 0xe3, 0xf4, 0xf9, 0x53, 0x14, 0xcb, 0x82, 0x44, 0x6f, 0x6c, 0x9e, 0x95, 0xcc,
0x0c, 0xea, 0x12, 0xba, 0x78, 0xae, 0x0d, 0x58, 0x86, 0x01, 0xe4, 0x62, 0x5b, 0xe9, 0xa9, 0x38,
0x89, 0xc0, 0x6b, 0xcb, 0xe3, 0xb9, 0x86, 0x24, 0xa3, 0xd8, 0xcf, 0xd4, 0x29, 0x0e, 0xc5, 0x7c,
0xcb, 0xb8, 0x4a, 0xe8, 0x85, 0x3a, 0xf6, 0xcf, 0x00, 0xbf, 0x4f, 0xe2, 0xd9, 0x84, 0x40, 0x63,
0x5d, 0xa8, 0x12, 0x95, 0xc5, 0xc7, 0xf4, 0x26, 0xe5, 0x1b, 0x97, 0x0a, 0xe5, 0xa0, 0x63, 0x72,
0xfa, 0xb3, 0x91, 0xbc, 0x69, 0x1c, 0x97, 0x58, 0x4a, 0xf5, 0xd3, 0x20, 0x5a, 0x88, 0x4f, 0x83,
0x28, 0x8b, 0x1b, 0x97, 0x79, 0x33, 0xb6, 0x32, 0x73, 0x1f, 0xea, 0xaf, 0xa3, 0x38, 0x48, 0x51,
0x19, 0x6d, 0x59, 0x7c, 0x41, 0xb3, 0x5d, 0x80, 0x9e, 0x18, 0x84, 0xa3, 0x20, 0x42, 0xa9, 0x53,
0x6c, 0x00, 0x99, 0x8c, 0x1b, 0x4a, 0xde, 0x33, 0xa8, 0x65, 0x54, 0x39, 0xf6, 0xc8, 0xed, 0x0f,
0x82, 0x48, 0x28, 0x2f, 0x88, 0xf0, 0xde, 0xc1, 0x9a, 0x2c, 0x46, 0x7c, 0x3e, 0xfa, 0x22, 0xbd,
0x41, 0x29, 0xde, 0xe8, 0x21, 0xf2, 0xfe, 0x65, 0x81, 0x83, 0x2b, 0x65, 0xc0, 0xd2, 0x06, 0xcc,
0xdb, 0xe8, 0xc8, 0xdb, 0xc8, 0x3a, 0xd0, 0xec, 0xa7, 0xf8, 0x4e, 0xe9, 0x36, 0xd6, 0xe0, 0x26,
0x0b, 0xf1, 0xf2, 0xc7, 0xa9, 0x4e, 0xb7, 0xcd, 0x17, 0x34, 0xdb, 0x84, 0x06, 0xf6, 0x26, 0x29,
0xc4, 0x46, 0x56, 0xe7, 0x9a, 0xc1, 0xb6, 0x00, 0x14, 0xb2, 0x33, 0x41, 0xdd, 0xcc, 0xe2, 0x06,
0xc7, 0x7b, 0x0c, 0x35, 0xf4, 0xf4, 0x28, 0x98, 0xe8, 0xd8, 0xac, 0xeb, 0x62, 0xfb, 0x6c, 0x41,
0xeb, 0x4f, 0x33, 0x91, 0x5c, 0x72, 0xf1, 0xb7, 0x99, 0x98, 0xa6, 0x88, 0x2d, 0xd1, 0xaa, 0x96,
0x89, 0xc0, 0xaa, 0xed, 0xbf, 0x0f, 0x92, 0xa1, 0x44, 0xca, 0xe1, 0x19, 0x85, 0xb1, 0x6a, 0xcc,
0xa7, 0x14, 0x6b, 0x9d, 0x9b, 0x2c, 0xaa, 0x77, 0x31, 0x8a, 0x53, 0x15, 0x4c, 0x46, 0xb1, 0x2e,
0xdc, 0x3a, 0xb8, 0x18, 0x44, 0xb3, 0xa1, 0xe0, 0xf1, 0x5c, 0xee, 0xa6, 0xe6, 0xcc, 0x8b, 0x6c,
0xf6, 0x73, 0x6c, 0x6e, 0xc4, 0x52, 0xad, 0xa9, 0x46, 0x8a, 0x05, 0x2e, 0xdb, 0x85, 0xd6, 0xc1,
0xe8, 0x4c, 0x0c, 0x87, 0x62, 0xd8, 0x0b, 0xd2, 0xc0, 0xad, 0x53, 0xdc, 0x85, 0x07, 0x3f, 0xa7,
0xe2, 0x7d, 0xb4, 0x60, 0x2d, 0x8b, 0x7e, 0x3a, 0x89, 0xc7, 0x53, 0x81, 0x29, 0x3e, 0x48, 0x12,
0x95, 0xe2, 0x83, 0x24, 0x61, 0x8f, 0xa1, 0xc6, 0xc5, 0x74, 0x16, 0xa5, 0xaa, 0x4a, 0xee, 0x6a,
0x8b, 0x6a, 0xef, 0x2c, 0x4a, 0xb9, 0xd2, 0x62, 0xbf, 0x83, 0xf5, 0x5c, 0x1d, 0xaa, 0x67, 0xe1,
0x27, 0x7a, 0x5f, 0x4e, 0xce, 0x0b, 0xea, 0xde, 0x67, 0x07, 0x9a, 0x86, 0xe5, 0x45, 0x91, 0x21,
0x3e, 0x6b, 0x59, 0x91, 0x3d, 0xa4, 0xb9, 0xeb, 0x8a, 0xa9, 0x07, 0x7b, 0x52, 0x0b, 0xac, 0xe3,
0xac, 0x2c, 0xad, 0x63, 0xdd, 0x08, 0xed, 0xeb, 0x1a, 0x21, 0x4e, 0x71, 0xef, 0x83, 0xf1, 0xb9,
0x18, 0x52, 0x59, 0xd6, 0xb9, 0x22, 0xd9, 0x8e, 0xee, 0x0a, 0x94, 0xc7, 0x5c, 0xaf, 0x51, 0x12,
0xae, 0x3b, 0x87, 0xec, 0x72, 0x38, 0x19, 0xd4, 0x64, 0xbd, 0x48, 0x8a, 0x3d, 0x87, 0xa6, 0x6e,
0x5f, 0xd3, 0x2c, 0x45, 0x6d, 0x6d, 0x4a, 0x0b, 0xb9, 0xa9, 0xc8, 0x5e, 0x16, 0x47, 0x34, 0xb7,
0x41, 0x5e, 0xb8, 0xb9, 0xc8, 0x0d, 0x39, 0x2f, 0x8e, 0x74, 0xbb, 0xc6, 0xcc, 0xe8, 0x02, 0x6d,
0xbe, 0xa3, 0x37, 0x2f, 0x44, 0xdc, 0x98, 0x2c, 0x9f, 0x9a, 0x6f, 0x89, 0xdb, 0xa4, 0x3d, 0xed,
0x3c, 0x72, 0x52, 0xc6, 0xcd, 0x37, 0x67, 0xd7, 0x78, 0xc8, 0xdc, 0x56, 0xf1, 0xa0, 0x85, 0x88,
0x1b, 0xcf, 0x9d, 0x5f, 0x32, 0xdf, 0xb9, 0x6b, 0xb4, 0xb5, 0x7c, 0x78, 0x93, 0x2a, 0xbc, 0x64,
0x2a, 0x7c, 0x59, 0x9c, 0x04, 0xdc, 0xf5, 0x22, 0x50, 0x79, 0x39, 0x2f, 0xe8, 0x7b, 0xff, 0xa9,
0xc0, 0x9a, 0x3f, 0x9a, 0xc4, 0x49, 0x6a, 0xb4, 0x04, 0x7f, 0x3c, 0x14, 0x17, 0xaa, 0x25, 0x10,
0x51, 0xfe, 0x6a, 0x52, 0x6b, 0xc6, 0xd6, 0x40, 0xad, 0xc0, 0xe1, 0x92, 0x30, 0xca, 0xc1, 0xc9,
0x95, 0xc3, 0x26, 0x34, 0x64, 0xed, 0xa3, 0xa8, 0x4a, 0x22, 0xcd, 0x90, 0x1f, 0x00, 0x73, 0x1a,
0x1c, 0x6b, 0x34, 0x8a, 0x2a, 0x12, 0xdb, 0xa0, 0x54, 0x23, 0x61, 0x9d, 0x84, 0x06, 0x07, 0xe5,
0x27, 0xe1, 0x48, 0x4c, 0xd3, 0x60, 0x34, 0xc1, 0xbe, 0x62, 0x77, 0x6d, 0x6e, 0x70, 0xb0, 0xa5,
0x50, 0x10, 0xaf, 0x12, 0x11, 0xa4, 0x62, 0xb8, 0x97, 0x52, 0x39, 0xd9, 0xbc, 0xc0, 0x45, 0x3d,
0x0a, 0x4b, 0xeb, 0x81, 0xd4, 0xcb, 0x73, 0xe9, 0x59, 0x8c, 0x44, 0x90, 0x50, 0x91, 0xd4, 0xb9,
0x24, 0xbc, 0xff, 0x57, 0x80, 0x49, 0x24, 0xe5, 0xe0, 0xf7, 0xa3, 0xc1, 0x79, 0x3d, 0x6c, 0x79,
0x70, 0x6a, 0x4b, 0xe0, 0xdc, 0x5b, 0x8c, 0xab, 0x12, 0x98, 0x8c, 0xc2, 0x5e, 0xae, 0x5f, 0x12,
0x89, 0xaa, 0xc5, 0x4d, 0x16, 0xf3, 0xa0, 0x65, 0x3c, 0x63, 0x78, 0x07, 0xd1, 0x76, 0x8e, 0x57,
0x02, 0x2d, 0xdc, 0x10, 0xda, 0xe6, 0xf5, 0xd0, 0xb6, 0x4c, 0x68, 0x3f, 0x5a, 0xd0, 0xda, 0x4b,
0xe3, 0x51, 0x38, 0xe0, 0x62, 0x10, 0x27, 0xc3, 0xab, 0x41, 0x95, 0xf0, 0x55, 0x4c, 0xf8, 0x76,
0xc0, 0xf6, 0x3f, 0x24, 0x59, 0x2b, 0xdc, 0x34, 0x06, 0xad, 0xa5, 0x5c, 0x71, 0x54, 0x64, 0x8f,
0xa0, 0xe2, 0x27, 0x54, 0xb9, 0xb9, 0x26, 0x9e, 0xbb, 0x24, 0xbc, 0xe2, 0x27, 0xde, 0xaf, 0xa0,
0x2d, 0x9d, 0x52, 0xa2, 0xec, 0x51, 0x69, 0x43, 0xf5, 0x20, 0x49, 0x62, 0xf5, 0xac, 0x48, 0xc2,
0xbb, 0x80, 0xf6, 0x49, 0x12, 0x8c, 0xa7, 0x51, 0x90, 0x0a, 0x4c, 0xcc, 0xd7, 0xd4, 0x47, 0xd9,
0xd7, 0x75, 0x07, 0x9a, 0xc7, 0x71, 0xfa, 0x2e, 0x09, 0x53, 0xba, 0xff, 0xb2, 0x93, 0x9b, 0x2c,
0xef, 0x17, 0x70, 0xb7, 0x70, 0xb2, 0x7e, 0xfd, 0xb0, 0xa4, 0x6c, 0xfd, 0x15, 0xdb, 0x87, 0x3b,
0x0b, 0x55, 0xbf, 0xf7, 0x55, 0x3e, 0x2e, 0x1b, 0xfd, 0xa5, 0x11, 0x39, 0x19, 0xcd, 0x8e, 0x2f,
0x89, 0xc6, 0xdb, 0x07, 0x37, 0x43, 0x53, 0x7e, 0xfc, 0x67, 0x1e, 0x9c, 0x86, 0x62, 0x7e, 0xd5,
0xf7, 0x11, 0xbd, 0xfe, 0x15, 0xfa, 0x65, 0x40, 0x6b, 0xef, 0x7b, 0x0b, 0xda, 0x65, 0x46, 0x74,
0x71, 0x59, 0x46, 0x71, 0xb1, 0x17, 0x50, 0xfd, 0x10, 0x8a, 0xb9, 0x7a, 0xef, 0xbd, 0xa5, 0x94,
0x2f, 0x79, 0xc2, 0xe5, 0x06, 0xbc, 0x5a, 0x7b, 0x83, 0x34, 0x8c, 0xc7, 0x6a, 0xb8, 0x97, 0x14,
0x9e, 0xb3, 0x1f, 0xc5, 0x83, 0xbf, 0xca, 0xcf, 0x56, 0x2e, 0x89, 0x92, 0xab, 0x52, 0xbd, 0xe1,
0x55, 0x59, 0x2d, 0xbb, 0x2a, 0xde, 0xbf, 0x2d, 0x85, 0x95, 0x31, 0x80, 0x7d, 0x31, 0x63, 0xfa,
0x82, 0xd8, 0xea, 0x82, 0xb8, 0x72, 0x8a, 0xd4, 0xc3, 0xb2, 0x22, 0x71, 0x72, 0xc5, 0x25, 0xfd,
0xb3, 0x70, 0x28, 0x4b, 0x0b, 0xfa, 0x0b, 0x5d, 0x69, 0x39, 0xd8, 0xd5, 0xb2, 0x60, 0xf7, 0x37,
0xfe, 0xfb, 0x69, 0xcb, 0xfa, 0xdf, 0xa7, 0x2d, 0xeb, 0xdb, 0x4f, 0x5b, 0xd6, 0x3f, 0xbf, 0xdb,
0x5a, 0x39, 0x5b, 0xa5, 0x7f, 0x4e, 0xbf, 0xfe, 0x21, 0x00, 0x00, 0xff, 0xff, 0x51, 0x90, 0x81,
0x4c, 0x83, 0x12, 0x00, 0x00,
}
func (m *Row) Marshal() (dAtA []byte, err error) {
@ -4913,16 +4905,6 @@ func (m *ImportRoaringRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) {
i -= len(m.XXX_unrecognized)
copy(dAtA[i:], m.XXX_unrecognized)
}
if m.Direct {
i--
if m.Direct {
dAtA[i] = 1
} else {
dAtA[i] = 0
}
i--
dAtA[i] = 0x38
}
if m.FieldCreatedAt != 0 {
i = encodeVarintPublic(dAtA, i, uint64(m.FieldCreatedAt))
i--
@ -6086,9 +6068,6 @@ func (m *ImportRoaringRequest) Size() (n int) {
if m.FieldCreatedAt != 0 {
n += 1 + sovPublic(uint64(m.FieldCreatedAt))
}
if m.Direct {
n += 2
}
if m.XXX_unrecognized != nil {
n += len(m.XXX_unrecognized)
}
@ -12138,26 +12117,6 @@ func (m *ImportRoaringRequest) Unmarshal(dAtA []byte) error {
break
}
}
case 7:
if wireType != 0 {
return fmt.Errorf("proto: wrong wireType = %d for field Direct", wireType)
}
var v int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPublic
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
v |= int(b&0x7F) << shift
if b < 0x80 {
break
}
}
m.Direct = bool(v != 0)
default:
iNdEx = preIndex
skippy, err := skipPublic(dAtA[iNdEx:])

1
internal/public.proto
View file

@ -236,7 +236,6 @@ message ImportRoaringRequest {
uint64 Block = 4;
int64 IndexCreatedAt = 5;
int64 FieldCreatedAt = 6;
bool Direct = 7;
}
message ImportColumnAttrsRequest {

7
rbf.go
View file

@ -531,12 +531,7 @@ func (w *RbfDBWrapper) Close() error {
var globalNextTxSnRBFTx int64
func (w *RbfDBWrapper) NewTx(write bool, initialIndex string, o Txo) (_ Tx, err error) {
var tx *rbf.Tx
if write && o.Direct { // obtain exclusive lock if writing directly to db.
tx, err = w.db.BeginWithExclusiveLock()
} else {
tx, err = w.db.Begin(write)
}
tx, err := w.db.Begin(write)
if err != nil {
return nil, err
}

40
rbf/cfg/cfg.go
View file

@ -15,11 +15,14 @@
package cfg
import (
"time"
"github.com/spf13/pflag"
)
const (
DefaultMinWALCheckpointSize = 1 * (1 << 20) // 1MB
DefaultMaxWALCheckpointSize = DefaultMaxWALSize / 2
)
// Config defines externally configurable rbf options.
// The separate package avoids circular import.
type Config struct {
@ -27,37 +30,38 @@ type Config struct {
// The maximum allowed database size. Required by mmap.
MaxSize int64
// The maximum allowed WAL size. Required by mmap.
MaxWALSize int64
// The minimum WAL size before the WAL is copied to the DB.
MinWALCheckpointSize int64
// The maximum WAL size before transactions are halted to allow a checkpoint.
MaxWALCheckpointSize int64
// Set before calling db.Open()
FsyncEnabled bool
// for mmap correctness testing.
DoAllocZero bool
// CheckpointEveryDur if zero means checkpoint after every write.
// Otherwise, wait and checkpoint at the next write that happens
// after CheckpointEveryDur since the previous.
CheckpointEveryDur time.Duration
// Maximum size of a single WAL segment.
// May exceed by one page if last page is a bitmap header + bitmap.
MaxWALSegmentFileSize int
}
func NewDefaultConfig() *Config {
return &Config{
MaxSize: DefaultMaxSize,
FsyncEnabled: true,
CheckpointEveryDur: time.Millisecond,
MaxWALSegmentFileSize: 1 << 20,
MaxSize: DefaultMaxSize,
MaxWALSize: DefaultMaxWALSize,
MinWALCheckpointSize: DefaultMinWALCheckpointSize,
MaxWALCheckpointSize: DefaultMaxWALCheckpointSize,
FsyncEnabled: true,
}
}
func (cfg *Config) DefineFlags(flags *pflag.FlagSet) {
default0 := NewDefaultConfig()
flags.IntVar(&cfg.MaxWALSegmentFileSize, "rbf-max-wal", default0.MaxWALSegmentFileSize, "RBF write-Ahead-Log file size in bytes")
flags.DurationVar(&cfg.CheckpointEveryDur, "rbf-checkpoint-dur", default0.CheckpointEveryDur,
"RBF checkpoint on the next write that occurs this long or more after the previous write. 0 means checkpoint after every write.")
flags.Int64Var(&cfg.MaxSize, "rbf-max-db-size", default0.MaxSize, "RBF maximum size in bytes of a database file (distinct from a WAL file)")
flags.Int64Var(&cfg.MaxWALSize, "rbf-max-wal-size", default0.MaxWALSize, "RBF maximum size in bytes of a WAL file (distinct from a DB file)")
flags.Int64Var(&cfg.MinWALCheckpointSize, "rbf-min-wal-checkpoint-size", default0.MinWALCheckpointSize, "RBF minimum size in bytes of a WAL file before attempting checkpoint")
flags.Int64Var(&cfg.MaxWALCheckpointSize, "rbf-max-wal-checkpoint-size", default0.MaxWALCheckpointSize, "RBF maximum size in bytes of a WAL file before forcing checkpoint")
// renamed from --rbf-fsync to just --fsync because now it applies to all Tx backends.
flags.BoolVar(&cfg.FsyncEnabled, "fsync", default0.FsyncEnabled, "enable fsync fully safe flush-to-disk")

5
rbf/cfg/os.go
View file

@ -20,3 +20,8 @@ package cfg
// size of the database. The size can be increased by updating the DB.MaxSize
// and reopening the database. This setting mainly affects virtual space usage.
const DefaultMaxSize = 4 * (1 << 30)
// DefaultMaxWALSize is the default mmap size and therefore the maximum allowed
// size of the WAL. The size can be increased by updating the DB.MaxWALSize
// and reopening the database. This setting mainly affects virtual space usage.
const DefaultMaxWALSize = 4 * (1 << 30)

5
rbf/cfg/os_386.go
View file

@ -18,3 +18,8 @@ package cfg
// size of the database. The size can be increased by updating the DB.MaxSize
// and reopening the database. This setting mainly affects virtual space usage.
const DefaultMaxSize = 256 * (1 << 20) // 256MB
// DefaultMaxWALSize is the default mmap size and therefore the maximum allowed
// size of the WAL. The size can be increased by updating the DB.MaxWALSize
// and reopening the database. This setting mainly affects virtual space usage.
const DefaultMaxWALSize = 64 * (1 << 20) // 64MB

14
rbf/cursor.go
View file

@ -322,13 +322,7 @@ func toPgno(val []byte) uint32 {
return binary.LittleEndian.Uint32(val)
}
func (c *Cursor) putLeafCell(in leafCell) (err error) {
// Copy target page if we are using direct writes because a split will
// cause the source data to be overwritten after the first page is written.
leafPage := c.leafPage
if c.tx.exclusive {
leafPage = make([]byte, PageSize)
copy(leafPage, c.leafPage)
}
cells := readLeafCells(leafPage, c.leafCells[:])
elem := &c.stack.elems[c.stack.index]
@ -506,14 +500,6 @@ func (c *Cursor) putBranchCells(stackIndex int, newCells []branchCell) (err erro
return err
}
// Copy target page if we are using direct writes because a split will
// cause the source data to be overwritten after the first page is written.
if c.tx.exclusive {
tmp := make([]byte, PageSize)
copy(tmp, page)
page = tmp
}
cells := readBranchCells(page)
// Update current cell & insert additional cells after it.

1
rbf/cursor_test.go
View file

@ -787,7 +787,6 @@ func TestCursor_RLEConversion(t *testing.T) {
}()...)
if err != nil {
t.Fatalf("ERR adding bits: %v\n", err)
}
if err := c.First(); err != nil {

365
rbf/db.go
View file

@ -18,12 +18,10 @@ import (
"errors"
"fmt"
"io"
"io/ioutil"
"os"
"path/filepath"
"sync"
"syscall"
"time"
"github.com/benbjohnson/immutable"
"github.com/pilosa/pilosa/v2/syswrap"
@ -40,24 +38,23 @@ var (
type DB struct {
cfg rbfcfg.Config
data []byte // mmap data
file *os.File // file descriptor
data []byte // database mmap
file *os.File // database file descriptor
rootRecords []*RootRecord // cached root records
pageMap *immutable.Map // pgno-to-WALID mapping
txs map[*Tx]struct{} // active transactions
opened bool // true if open
wcache []byte // wal write cache
segments []WALSegment // write-ahead log
wal []byte // wal mmap
walFile *os.File // wal file descriptor
walPageN int // wal page count
mu sync.RWMutex // general mutex
rwmu sync.Mutex // mutex for restricting single writer
exclmu sync.RWMutex // mutex for locking out everyone but a single writer
mu sync.RWMutex // general mutex
rwmu sync.Mutex // mutex for restricting single writer
haltCond *sync.Cond // condition for resuming txs after checkpoint
// Path represents the path to the database file.
Path string
lastCheckpoint time.Time
}
// NewDB returns a new instance of DB.
@ -70,9 +67,9 @@ func NewDB(path string, cfg *rbfcfg.Config) *DB {
cfg: *cfg,
txs: make(map[*Tx]struct{}),
pageMap: immutable.NewMap(&uint32Hasher{}),
wcache: make([]byte, cfg.MaxWALSegmentFileSize+PageSize),
Path: path,
}
db.haltCond = sync.NewCond(&db.mu)
return db
}
@ -81,7 +78,7 @@ func (db *DB) DataPath() string {
return filepath.Join(db.Path, "data")
}
// WALPath returns the path to the WAL directory.
// WALPath returns the path to the WAL file.
func (db *DB) WALPath() string {
return filepath.Join(db.Path, "wal")
}
@ -115,7 +112,7 @@ func (db *DB) Open() (err error) {
return fmt.Errorf("open file: %w", err)
}
// Open read-only mmap.
// Open read-only database mmap.
if f, err := os.OpenFile(db.DataPath(), os.O_RDONLY, 0666); err != nil {
return fmt.Errorf("open mmap file: %w", err)
} else if db.data, err = syswrap.Mmap(int(f.Fd()), 0, int(db.cfg.MaxSize), syscall.PROT_READ, syscall.MAP_SHARED); err != nil {
@ -136,181 +133,112 @@ func (db *DB) Open() (err error) {
// TODO(BBJ): Obtain advisory lock on file.
// Ensure WAL directory exists.
if err := os.MkdirAll(db.WALPath(), 0777); err != nil {
return fmt.Errorf("create wal dir: %w", err)
}
db.opened = true
// Open write-ahead log & checkpoint to the end since no transactions are open.
if err := db.openWALSegments(); err != nil {
if err := db.openWAL(); err != nil {
return fmt.Errorf("wal open: %w", err)
} else if err := db.checkpoint(true); err != nil {
} else if err := db.checkpoint(); err != nil {
return fmt.Errorf("checkpoint: %w", err)
}
return nil
}
func (db *DB) openWALSegments() error {
fis, err := ioutil.ReadDir(db.WALPath())
if err != nil {
return fmt.Errorf("read dir: %w", err)
func (db *DB) openWAL() (err error) {
// Open WAL file writer.
if db.walFile, err = os.OpenFile(db.WALPath(), os.O_WRONLY|os.O_CREATE, 0666); err != nil {
return fmt.Errorf("open wal file: %w", err)
}
// Open all WAL segments.
for _, fi := range fis {
if filepath.Ext(fi.Name()) != ".wal" {
continue
}
// Open read-only mmap.
if f, err := os.OpenFile(db.WALPath(), os.O_RDONLY, 0666); err != nil {
return fmt.Errorf("open wal mmap file: %w", err)
} else if db.wal, err = syswrap.Mmap(int(f.Fd()), 0, int(db.cfg.MaxWALSize), syscall.PROT_READ, syscall.MAP_SHARED); err != nil {
f.Close()
return fmt.Errorf("open wal mmap file: %w", err)
} else if err := f.Close(); err != nil {
return fmt.Errorf("cannot close wal mmap file: %w", err)
}
segment := db.NewWALSegment(filepath.Join(db.WALPath(), fi.Name()))
if err := segment.Open(); err != nil {
_ = db.closeWALSegments()
// Determine the number of whole pages in the WAL.
var pageN int
if fi, err := db.walFile.Stat(); err != nil {
return fmt.Errorf("wal stat: %w", err)
} else {
pageN = int(fi.Size() / PageSize)
}
// Read backwards through the WAL to find the last valid meta page.
for ; pageN > 0; pageN-- {
if page, err := db.readWALPageAt(pageN - 1); err != nil {
return err
} else if IsMetaPage(page) {
break
}
db.segments = append(db.segments, segment)
}
// Truncate everything after the last successful meta page.
if walID, err := findLastWALMetaPage(db.segments); err != nil {
return err
} else if db.segments, err = db.truncateWALAfter(db.segments, walID); err != nil {
return err
// Truncate WAL to the last valid meta page.
if err := db.walFile.Truncate(int64(pageN * PageSize)); err != nil {
return fmt.Errorf("wal truncate: %w", err)
} else if _, err := db.walFile.Seek(int64(pageN*PageSize), io.SeekStart); err != nil {
return fmt.Errorf("wal seek: %w", err)
}
db.walPageN = pageN
return nil
}
// updateWALSegment updates or adds a segment.
func (db *DB) updateWALSegment(s WALSegment) {
segments := make([]WALSegment, len(db.segments), len(db.segments)+1)
copy(segments, db.segments)
// Find the matching segment using the path.
segment := walSegmentByPath(segments, s.Path)
// Update existing segment if it already exists.
// Otherwise append segment to the end.
if segment != nil {
*segment = s
} else {
assert(len(segments) == 0 || segments[len(segments)-1].MinWALID < s.MinWALID)
segments = append(segments, s)
}
// Replace DB segment list.
db.segments = segments
}
// checkpoint moves WAL segments to the main DB file.
// checkpoint moves all WAL pages to the main DB file.
// Must be called by a write transaction while under db.mu lock.
func (db *DB) checkpoint(exclusive bool) error {
func (db *DB) checkpoint() error {
if !db.opened {
return nil
} else if len(db.txs) > 0 {
return nil // skip if transactions open
}
// Determine last checkpointed WAL ID.
page, err := db.readDBPage(0)
if err != nil {
return err
}
walID := readMetaWALID(page)
// INVAR: walID represents everything already in the DB, and
// any wal page k > walID is in the WAL not the DB.
// Loop over each transaction
// We could be looking at a recovery. When there is no new
// meta page further down in the WAL, then there was power
// failure or the process was killed. So we have to search
// and find the next meta page, if present.
//
// Read ahead to the next meta page, if present, in the WAL. If we
// we find it, then we must ensure the pages between
// [walID, next_meta_page.walID] are committed.
// If there is NOT another meta page after, then those writes get
// rolled back.
walID++
for {
// Determine last page of transaction.
metaWALID, err := findNextWALMetaPage(db.segments, walID)
if err == ErrNoMetaFound {
break
} else if err != nil {
for i := 0; i < db.walPageN; i++ {
page, err := db.readWALPageAt(i)
if err != nil {
return err
}
// Loop over pages in the transaction.
for ; walID <= metaWALID; walID++ {
page, err := readWALPage(db.segments, walID)
if err != nil {
// Determine page number. Meta pages are always on zero & bitmap
// headers specify the page number of the next page in the WAL.
// All other pages have their page number in the page data.
var pgno uint32
if IsBitmapHeader(page) {
pgno = readPageNo(page)
if page, err = db.readWALPageAt(i + 1); err != nil {
return err
}
isBitmapHeader := IsBitmapHeader(page)
i++ // bitmaps in WAL are two pages
} else if !IsMetaPage(page) {
pgno = readPageNo(page)
}
// Determine page number. Meta pages are always on zero & bitmap
// headers specify the page number of the next page in the WAL.
// All other pages have their page number in the page data.
var pgno uint32
if isBitmapHeader {
pgno, walID = readPageNo(page), walID+1 // skip next page
} else if !IsMetaPage(page) {
pgno = readPageNo(page)
}
// Ensure we actually read the bitmap data in when we checkpoint.
// NOTE: The walID variable is incremented above in the pgno check.
if isBitmapHeader {
if page, err = readWALPage(db.segments, walID); err != nil {
return err
}
}
// TODO: address this problem: if we write a WAL meta page to database page 0 before fsyncing
// the transactions updates from the WAL into the DB, then (upon
// power failure in the middle of a fsync), the meta page might
// get updated before all of the databases pages that included the changes
// that the meta page represents. The only way to have a strict
// ordering that the meta page is updated only after the other
// pages is to fsync it in a 2nd fsync that follows the
// the first. SSDs and HDs both exhibit these "unsynchronized writes".
// reference https://www.usenix.org/system/files/conference/fast13/fast13-final80.pdf
//
// needed pattern:
// 1) write tx-content pages;
// 2) fsync the tx-content pages;
// 3) write meta page;
// 4) fsync the meta page.
// The OS can also be inserting fsyncs at any point (e.g. due to memory pressure)
// and so we have to be certain that the meta page is written after a separate fsync.
// Write page data into main db file.
if err := db.writeDBPage(pgno, page); err != nil {
return err
}
// Write data to the data file.
if err := db.writeDBPage(pgno, page); err != nil {
return err
}
}
// Ensure WAL pages are fully copied & synced to DB file.
// Ensure database file is synced and then truncate the WAL file.
if err := db.fsync(db.file); err != nil {
return fmt.Errorf("db file sync: %w", err)
} else if err := db.walFile.Truncate(0); err != nil {
return fmt.Errorf("truncate wal file: %w", err)
} else if err := db.fsync(db.walFile); err != nil {
return fmt.Errorf("wal file sync: %w", err)
} else if _, err := db.walFile.Seek(0, io.SeekStart); err != nil {
return fmt.Errorf("seek wal file: %w", err)
}
db.walPageN = 0
db.pageMap = immutable.NewMap(&uint32Hasher{})
// Remove WAL segments that have been checkpointed.
for _, segment := range db.segments {
if err := segment.Close(); err != nil {
return err
} else if err := os.Remove(segment.Path); err != nil {
return err
}
}
db.segments = nil
// Notify halted tranactions that the WAL has been checkpointed.
db.haltCond.Broadcast()
return nil
}
@ -344,21 +272,22 @@ func (db *DB) Close() (err error) {
db.file = nil
}
if e := db.closeWALSegments(); e != nil && err == nil {
err = e
}
return err
}
// closeWALSegments closes the WAL and all its segments.
func (db *DB) closeWALSegments() (err error) {
for _, s := range db.segments {
if e := s.Close(); e != nil && err == nil {
// Close WAL mmap handle.
if db.wal != nil {
if e := syswrap.Munmap(db.wal); e != nil && err == nil {
err = e
}
db.wal = nil
}
db.segments = nil
// Close wal writer handler.
if db.walFile != nil {
if e := db.walFile.Close(); e != nil && err == nil {
err = e
}
db.walFile = nil
}
return err
}
@ -431,23 +360,18 @@ func (db *DB) Size() (int64, error) {
if err != nil {
return 0, err
}
return walSize(db.segments) + fi.Size(), nil
return db.walSize() + fi.Size(), nil
}
// WALSize returns the size of all WAL segments, in bytes.
// WALSize returns the size of the WAL, in bytes.
func (db *DB) WALSize() int64 {
db.mu.RLock()
defer db.mu.RUnlock()
return walSize(db.segments)
return db.walSize()
}
// WALSegments returns the WAL segments currently on the DB.
func (db *DB) WALSegments() []WALSegment {
db.mu.RLock()
defer db.mu.RUnlock()
other := make([]WALSegment, len(db.segments))
copy(other, db.segments)
return other
func (db *DB) walSize() int64 {
return int64(db.walPageN * PageSize)
}
// init initializes a new database file.
@ -496,32 +420,6 @@ func (db *DB) initFreelistPage() error {
// Begin starts a new transaction.
func (db *DB) Begin(writable bool) (_ *Tx, err error) {
return db.begin(writable, false)
}
// BeginWithExclusiveLock starts a new transaction with an exclusive lock.
//
// This waits for all read transactions to finish and disallows any other
// transactions on the database. All WAL writes are flushed to disk and page
// writes during this transaction are written directly to the database file.
//
// Note that because page writes are direct, write failures can corrupt the
// database. This should only be used during bulk loading of data.
func (db *DB) BeginWithExclusiveLock() (_ *Tx, err error) {
return db.begin(true, true)
}
func (db *DB) begin(writable, exclusive bool) (_ *Tx, err error) {
if exclusive {
assert(writable) // exclusive transactions must be writable
}
if exclusive {
db.exclmu.Lock()
} else {
db.exclmu.RLock()
}
// Ensure only one writable transaction at a time.
if writable {
db.rwmu.Lock()
@ -530,12 +428,6 @@ func (db *DB) begin(writable, exclusive bool) (_ *Tx, err error) {
// This local function is called at exit points that occur before we can
// call Rollback() which would normally release these locks.
cleanup := func() {
if exclusive {
db.exclmu.Unlock()
} else {
db.exclmu.RUnlock()
}
if writable {
db.rwmu.Unlock()
}
@ -554,35 +446,24 @@ func (db *DB) begin(writable, exclusive bool) (_ *Tx, err error) {
return nil, ErrClosed
}
// Flush all WAL writes to disk before an exclusive writer so that we can
// work directly with the on-disk database.
if exclusive {
if err := db.checkpoint(true); err != nil {
cleanup()
db.mu.Unlock()
return nil, err
}
// Wait for WAL size to be below threshold.
for int64(db.walPageN*PageSize) > db.cfg.MaxWALCheckpointSize {
db.haltCond.Wait()
}
tx := &Tx{
db: db,
rootRecords: db.rootRecords,
pageMap: db.pageMap,
walPageN: db.walPageN,
writable: writable,
exclusive: exclusive,
DeleteEmptyContainer: true,
}
if writable {
tx.wcache = db.wcache[:0]
}
// Copy list of WAL segments so they can be altered by the tx.
// Add last segment to the list of segments that will be updated/added.
if len(db.segments) != 0 {
tx.segments = make([]WALSegment, len(db.segments))
copy(tx.segments, db.segments)
tx.updatedSegmentPaths = []string{tx.segments[len(tx.segments)-1].Path}
if writable {
tx.dirtyPages = make(map[uint32][]byte)
tx.dirtyBitmapPages = make(map[uint32][]byte)
}
// Copy meta page into transaction's buffer.
@ -609,12 +490,6 @@ func (db *DB) begin(writable, exclusive bool) (_ *Tx, err error) {
// removeTx removes an active transaction from the database.
func (db *DB) removeTx(tx *Tx) error {
if tx.exclusive {
db.exclmu.Unlock()
} else {
db.exclmu.RUnlock()
}
// Release writer lock if tx is writable.
if tx.writable {
tx.db.rwmu.Unlock()
@ -625,25 +500,17 @@ func (db *DB) removeTx(tx *Tx) error {
delete(tx.db.txs, tx)
// Write pages from WAL to DB.
// TODO(bbj): Move this to an async goroutine.
// TODO(jea): Make the time-based checkpointing work at all, and update the
// comment in cfg/cfg.go for CheckpointEveryDur. Seems that
// wal.go readWALPage() can receive a request for a walID that
// comes before the segments it is passed if we do not
// checkpoint eagerly.
if tx.writable {
if db.cfg.CheckpointEveryDur == 0 || time.Since(db.lastCheckpoint) > db.cfg.CheckpointEveryDur {
if err := db.checkpoint(false); err != nil {
return fmt.Errorf("checkpoint: %w", err)
}
db.lastCheckpoint = time.Now()
}
}
// Disassociate from db.
tx.db = nil
// Write pages from WAL to DB.
// TODO(bbj): Move this to an async goroutine.
if len(db.txs) == 0 && db.walSize() > db.cfg.MinWALCheckpointSize {
if err := db.checkpoint(); err != nil {
return fmt.Errorf("checkpoint: %w", err)
}
}
return nil
}
@ -669,9 +536,25 @@ func (db *DB) readDBPage(pgno uint32) ([]byte, error) {
return db.data[offset : offset+PageSize], nil
}
// baseWALID returns the WAL ID stored in the database file meta page.
func (db *DB) baseWALID() int64 {
return readMetaWALID(db.data)
}
// readWALPageByID reads a WAL page by WAL ID.
func (db *DB) readWALPageByID(id int64) ([]byte, error) {
return db.readWALPageAt(int(id - db.baseWALID() - 1))
}
// readWALPageAt reads the i-th page in the WAL file.
func (db *DB) readWALPageAt(i int) ([]byte, error) {
offset := int64(i) * PageSize
return db.wal[offset : offset+PageSize], nil
}
func (db *DB) readMetaPage() ([]byte, error) {
if walID, ok := db.pageMap.Get(uint32(0)); ok {
return readWALPage(db.segments, walID.(int64))
return db.readWALPageByID(walID.(int64))
}
return db.readDBPage(0)
}

173
rbf/db_test.go
View file

@ -39,44 +39,88 @@ func TestDB_Open(t *testing.T) {
}
}
/* optimization of wal size means there may certainly be more than 2 WAL segments.
func TestDB_Checkpoint(t *testing.T) {
if testing.Short() {
t.Skip("-short enabled, skipping")
}
func TestDB_WAL(t *testing.T) {
t.Run("ErrTxTooLarge", func(t *testing.T) {
config := rbfcfg.NewDefaultConfig()
config.MaxWALSize = 4 * rbf.PageSize
db := MustOpenDB(t)
defer MustCloseDB(t, db)
db := MustOpenDB(t, config)
defer MustCloseDB(t, db)
// Create bitmap.
if tx, err := db.Begin(true); err != nil {
t.Fatal(err)
} else if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
} else if err := tx.Commit(); err != nil {
t.Fatal(err)
}
tx := MustBegin(t, db, true)
defer tx.Rollback()
// Create a bunch of transactions to generate WAL segments.
rand := rand.New(rand.NewSource(0))
for i := 0; i < 1000; i++ {
if tx, err := db.Begin(true); err != nil {
if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
} else if _, err := tx.Add("x", rand.Uint64()); err != nil {
t.Fatal(err)
} else if _, err := tx.Add("x", rand.Uint64()); err != nil {
t.Fatal(err)
} else if err := tx.Commit(); err != nil {
} else if err := tx.CreateBitmap("y"); err != rbf.ErrTxTooLarge {
t.Fatalf("unexpected error: %v", err)
}
})
t.Run("Halt", func(t *testing.T) {
if testing.Short() {
t.Skip("-short enabled, skipping")
}
config := rbfcfg.NewDefaultConfig()
config.MaxWALSize = 16 * rbf.PageSize
config.MaxWALCheckpointSize = 8 * rbf.PageSize
config.MinWALCheckpointSize = 4 * rbf.PageSize
db := MustOpenDB(t, config)
defer MustCloseDB(t, db)
// Continuously run read overlapping transactions.
ctx, cancel := context.WithCancel(context.Background())
g, ctx := errgroup.WithContext(ctx)
for i := 0; i < 10; i++ {
i := i
g.Go(func() error {
time.Sleep(time.Duration(i) * 10 * time.Millisecond) // stagger
for {
if err := ctx.Err(); err != nil {
return nil
}
if err := func() error {
tx, err := db.Begin(false)
if err != nil {
return err
}
defer tx.Rollback()
time.Sleep(20 * time.Millisecond)
return nil
}(); err != nil {
return err
}
}
})
}
// Generate updates to the DB/WAL.
for i := 0; i < 100; i++ {
func() {
tx := MustBegin(t, db, true)
defer tx.Rollback()
if err := tx.CreateBitmapIfNotExists("x"); err != nil {
t.Fatal(err)
} else if _, err := tx.Add("x", uint64(i)); err != nil {
t.Fatal(err)
} else if err := tx.Commit(); err != nil {
t.Fatal(err)
}
}()
}
// Stop read transactions & wait.
cancel()
if err := g.Wait(); err != nil {
t.Fatal(err)
}
}
// Ensure there is no more than two WAL segments.
if n := len(db.WALSegments()); n > 2 {
t.Fatalf("expected two or fewer WAL segments, got %d", n)
}
})
}
*/
func TestDB_Recovery(t *testing.T) {
// Ensure a bitmap header written without a bitmap is truncated.
@ -121,11 +165,10 @@ func TestDB_Recovery(t *testing.T) {
tx1.Rollback()
// Close database & truncate WAL to remove commit page & bitmap data page.
segments := db.WALSegments()
segment := segments[len(segments)-1]
walPath, walSize := db.WALPath(), db.WALSize()
if err := db.Close(); err != nil {
t.Fatal(err)
} else if err := os.Truncate(segment.Path, segment.Size()-(2*rbf.PageSize)); err != nil {
} else if err := os.Truncate(walPath, walSize-(2*rbf.PageSize)); err != nil {
t.Fatal(err)
}
@ -152,67 +195,6 @@ func TestDB_Recovery(t *testing.T) {
})
}
func TestDB_BeginWithExclusiveLock(t *testing.T) {
t.Run("EnsureBlock", func(t *testing.T) {
db := MustOpenDB(t)
defer MustCloseDB(t, db)
tx, err := db.BeginWithExclusiveLock()
if err != nil {
t.Fatal(err)
} else if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
}
// Attempt to start another transaction in a second goroutine.
ch := make(chan struct{})
go func() {
tx1, err := db.Begin(false)
if err != nil {
panic(err)
}
defer tx1.Rollback()
close(ch) // signal
}()
// Ensure other transctions are blocked during an exclusive lock.
select {
case <-ch:
t.Fatal("secondary transaction too soon")
case <-time.After(100 * time.Millisecond):
}
// Release exclusive lock.
if err := tx.Commit(); err != nil {
t.Fatal(err)
}
// Ensure other transaction to begin after exclusive lock released.
select {
case <-time.After(1 * time.Second):
t.Fatal("expected secondary transaction")
case <-ch:
}
})
t.Run("EnsureNoWAL", func(t *testing.T) {
db := MustOpenDB(t)
defer MustCloseDB(t, db)
tx, err := db.BeginWithExclusiveLock()
if err != nil {
t.Fatal(err)
}
defer tx.Rollback()
if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
} else if got, want := db.WALSize(), int64(0); got != want {
t.Fatalf("WALSize()=%d, want %d", got, want)
}
})
}
func TestDB_HasData(t *testing.T) {
db := MustOpenDB(t)
@ -300,7 +282,6 @@ func TestDB_MultiTx(t *testing.T) {
}
cfg := rbfcfg.NewDefaultConfig()
cfg.CheckpointEveryDur = 1 * time.Millisecond
db := MustOpenDB(t, cfg)
defer MustCloseDB(t, db)

41
rbf/rbf.go
View file

@ -105,6 +105,7 @@ var (
ErrBitmapNameRequired = errors.New("bitmap name required")
ErrBitmapNotFound = errors.New("bitmap not found")
ErrBitmapExists = errors.New("bitmap already exists")
ErrTxTooLarge = errors.New("rbf tx too large")
)
// Debug is just a temporary flag used for debugging.
@ -672,25 +673,33 @@ func RowValues(b []uint64) []uint64 {
// return fmt.Sprintf("%s:%d", file, line)
// }
// truncate truncates the file at path to sz bytes. File must exist.
func (db *DB) truncate(path string, sz int64) error {
f, err := os.OpenFile(path, os.O_WRONLY, 0666)
if err != nil {
return fmt.Errorf("open file: %w", err)
}
defer f.Close()
if err := f.Truncate(sz); err != nil {
return fmt.Errorf("truncate: %w", err)
} else if err := db.fsync(f); err != nil {
return fmt.Errorf("sync: %w", err)
}
return f.Close()
}
func (db *DB) fsync(f *os.File) error {
if !db.cfg.FsyncEnabled {
return nil
}
return f.Sync()
}
// uint32Hasher implements Hasher for uint32 keys.
type uint32Hasher struct{}
// Hash returns a hash for key.
func (h *uint32Hasher) Hash(key interface{}) uint32 {
return hashUint64(uint64(key.(uint32)))
}
// Equal returns true if a is equal to b. Otherwise returns false.
// Panics if a and b are not ints.
func (h *uint32Hasher) Equal(a, b interface{}) bool {
return a.(uint32) == b.(uint32)
}
// hashUint64 returns a 32-bit hash for a 64-bit value.
func hashUint64(value uint64) uint32 {
hash := value
for value > 0xffffffff {
value /= 0xffffffff
hash ^= value
}
return uint32(hash)
}

286
rbf/tx.go
View file

@ -14,11 +14,10 @@
package rbf
import (
"bufio"
"fmt"
"io"
"math"
"os"
"path/filepath"
"sort"
"strings"
"sync"
@ -33,23 +32,21 @@ var _ = txkey.ToString
// Tx represents a transaction.
type Tx struct {
mu sync.RWMutex
db *DB // parent db
segments []WALSegment // copy of WAL segments
updatedSegmentPaths []string // updated or added segment paths
meta [PageSize]byte // copy of current meta page
walID int64 // max WAL ID at start of tx
rootRecords []*RootRecord // read-only cache of root records
mu sync.RWMutex
db *DB // parent db
meta [PageSize]byte // copy of current meta page
walID int64 // max WAL ID at start of tx
walPageN int // wal page count
rootRecords []*RootRecord // read-only cache of root records
// pageMap holds WAL pages that have not yet been transferred
// into the database pages. So it can be empty, if the whole previous
// WAL has been checkpointed back into the database.
pageMap *immutable.Map // mapping of database pages to WAL IDs
writable bool // if true, tx can write
exclusive bool // if true, tx writes directly to db file (no wal)
dirty bool // if true, changes have been made
pageMap *immutable.Map // mapping of database pages to WAL IDs
writable bool // if true, tx can write
wcache []byte // write cache
dirtyPages map[uint32][]byte // updated pages in this tx
dirtyBitmapPages map[uint32][]byte // updated bitmap pages in this tx
// If Rollback() has already completed, don't do it again.
// Note db == nil means that commit has already been done.
@ -70,6 +67,16 @@ func (tx *Tx) Writable() bool {
return tx.writable
}
// dirty returns true if any pages have been updated in this tx.
func (tx *Tx) dirty() bool {
return tx.dirtyN() != 0
}
// dirtyN returns the number of dirty pages.
func (tx *Tx) dirtyN() int {
return len(tx.dirtyPages) + len(tx.dirtyBitmapPages)
}
// PageN returns the number of pages in the database as seen by this transaction.
func (tx *Tx) PageN() int {
return int(readMetaPageN(tx.meta[:]))
@ -86,13 +93,10 @@ func (tx *Tx) Commit() error {
// If any pages have been written, ensure we write a new meta page with
// the commit flag to mark the end of the transaction.
if tx.dirty {
if err := tx.writeMetaPage(MetaPageFlagCommit); err != nil {
return err
} else if err := tx.flushWALWriter(); err != nil {
if tx.dirty() {
if err := tx.flush(); err != nil {
return err
}
// future plan: after checkpoint is moved to background
// or not every removeTx, then we can move the
// tx.db.rootRecords = tx.rootRecords into removeTx().
@ -102,12 +106,7 @@ func (tx *Tx) Commit() error {
tx.db.mu.Lock()
tx.db.rootRecords = tx.rootRecords
tx.db.pageMap = tx.pageMap
for _, path := range tx.updatedSegmentPaths {
segment := walSegmentByPath(tx.segments, path)
assert(segment != nil)
//lint:ignore SA5011 the assert above prevents this from being nil
tx.db.updateWALSegment(*segment) //nolint:staticcheck
}
tx.db.walPageN = tx.walPageN
tx.db.mu.Unlock()
}
@ -129,16 +128,6 @@ func (tx *Tx) Rollback() {
return
}
// TODO(bbj): Invalidate DB if rollback fails. Possibly attempt reopen?
if tx.dirty {
if _, err := tx.db.truncateWALAfter(tx.segments, tx.walID); err != nil {
panicOn(err)
}
tx.segments = nil
tx.updatedSegmentPaths = nil
}
// Disconnect transaction from DB.
panicOn(tx.db.removeTx(tx))
}
@ -950,80 +939,38 @@ func (tx *Tx) readPage(pgno uint32) ([]byte, error) {
return nil, fmt.Errorf("rbf: page read out of bounds: pgno=%d max=%d", pgno, pageN)
}
// Check if page is remapped.
if walID, ok := tx.pageMap.Get(pgno); ok {
walID64 := walID.(int64)
// Read from write cache if not yet flushed to disk.
maxWALID := activeWALSegment(tx.segments).MaxWALID()
if walID64 > maxWALID {
offset := (walID64 - maxWALID - 1) * PageSize
return tx.wcache[offset : offset+PageSize], nil
// Check if page has been updated in this tx.
if tx.writable {
if page := tx.dirtyPages[pgno]; page != nil {
return page, nil
} else if page := tx.dirtyBitmapPages[pgno]; page != nil {
return page, nil
}
// Otherwise return remapped page from WAL segment.
return readWALPage(tx.segments, walID64)
}
// Check if page is remapped in WAL.
if walID, ok := tx.pageMap.Get(pgno); ok {
return tx.db.readWALPageByID(walID.(int64))
}
// Otherwise read directly from DB.
return tx.db.readDBPage(pgno)
}
func (tx *Tx) writePage(page []byte) error {
// Mark transaction as dirty so we write a meta page on commit/rollback.
tx.dirty = true
// If we are running in exclusive mode, directly write page to database.
if tx.exclusive {
return tx.db.writeDBPage(readPageNo(page), page)
}
// Write page to WAL and obtain position in WAL.
walID, err := tx.writeWALPage(page, false)
if err != nil {
return err
}
// Update page map with WAL position.
tx.pageMap = tx.pageMap.Set(readPageNo(page), walID)
return nil
tx.dirtyPages[readPageNo(page)] = page
return tx.checkTxSize()
}
func (tx *Tx) writeBitmapPage(pgno uint32, page []byte) error {
// Mark transaction as dirty so we write a meta page on commit/rollback.
tx.dirty = true
// If we are running in exclusive mode, directly write page to database.
if tx.exclusive {
return tx.db.writeDBPage(pgno, page)
}
// Write bitmap to WAL and obtain WAL position of the actual page data (not the prefix page).
walID, err := tx.writeBitmapWALPage(pgno, page)
if err != nil {
return err
}
// Update page map with WAL position.
tx.pageMap = tx.pageMap.Set(pgno, walID)
return nil
tx.dirtyBitmapPages[pgno] = page
return tx.checkTxSize()
}
func (tx *Tx) writeMetaPage(flag uint32) error {
// Set meta flags.
writeFlags(tx.meta[:], flag)
// If we are running in exclusive mode, directly write page to database.
if tx.exclusive {
return tx.db.writeDBPage(0, tx.meta[:])
func (tx *Tx) checkTxSize() error {
if (tx.walPageN+tx.dirtyN())*PageSize >= len(tx.db.wal) {
return ErrTxTooLarge
}
// Write page to WAL and obtain position in WAL.
walID, err := tx.writeWALPage(tx.meta[:], true)
if err != nil {
return err
}
tx.pageMap = tx.pageMap.Set(uint32(0), walID)
return nil
}
@ -1609,113 +1556,72 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear
return
}
func (tx *Tx) flushWALWriter() error {
// Ignore if we have no data in the write cache.
if len(tx.wcache) == 0 {
return nil
// flush writes the dirty pages & meta page to the WAL.
func (tx *Tx) flush() error {
w := bufio.NewWriterSize(tx.db.walFile, 65536)
builder := immutable.NewMapBuilder(tx.pageMap)
// Write non-bitmap pages to WAL.
for _, pgno := range dirtyPageMapKeys(tx.dirtyPages) {
walID, err := tx.writeToWAL(w, tx.dirtyPages[pgno])
if err != nil {
return fmt.Errorf("write page to wal: %w", err)
}
builder.Set(pgno, walID)
}
// Determine active WAL segment.
assert(len(tx.segments) != 0)
segment := &tx.segments[len(tx.segments)-1]
// Write bitmap headers & pages to WAL.
for _, pgno := range dirtyPageMapKeys(tx.dirtyBitmapPages) {
// Write header page.
hdr := make([]byte, PageSize)
writePageNo(hdr[:], pgno)
writeFlags(hdr[:], PageTypeBitmapHeader)
if _, err := tx.writeToWAL(w, hdr); err != nil {
return fmt.Errorf("write bitmap header page to wal: %w", err)
}
// Open write handle to active segment.
w, err := os.OpenFile(segment.Path, os.O_WRONLY, 0666)
// Write bitmap page.
walID, err := tx.writeToWAL(w, tx.dirtyBitmapPages[pgno])
if err != nil {
return fmt.Errorf("write bitmap page to wal: %w", err)
}
builder.Set(pgno, walID)
}
// Write meta page to WAL.
walID, err := tx.writeToWAL(w, tx.meta[:])
if err != nil {
return fmt.Errorf("open wal segment write handle: %w", err)
return fmt.Errorf("write meta page to wal: %w", err)
}
defer w.Close()
builder.Set(uint32(0), walID)
// Flush cache to writer.
if _, err := w.WriteAt(tx.wcache, int64(segment.PageN)*PageSize); err != nil {
return fmt.Errorf("write wal segment: %w", err)
} else if err := tx.db.fsync(w); err != nil {
return fmt.Errorf("sync wal segment: %w", err)
} else if err := w.Close(); err != nil {
return fmt.Errorf("close wal segment: %w", err)
// Flush & sync WAL.
if err := w.Flush(); err != nil {
return fmt.Errorf("flush wal: %w", err)
} else if err := tx.db.fsync(tx.db.walFile); err != nil {
return fmt.Errorf("sync wal: %w", err)
}
// Increase the size of the last WAL segment & clear cache.
assert(len(tx.wcache)%PageSize == 0)
segment.PageN += len(tx.wcache) / PageSize
tx.wcache = tx.wcache[:0]
// Save page map for new WAL pages.
tx.pageMap = builder.Map()
return nil
}
func (tx *Tx) writeWALPage(page []byte, isMeta bool) (walID int64, err error) {
if err := tx.ensureWritableWALSegment(); err != nil {
return 0, err
}
func (tx *Tx) writeToWAL(w io.Writer, page []byte) (walID int64, err error) {
// Determine next WAL ID from cached meta page.
walID = readMetaWALID(tx.meta[:]) + 1
// Update WAL ID on cached meta page.
writeMetaWALID(tx.meta[:], walID)
// Append write to write buffer.
tx.wcache = append(tx.wcache, page...)
return walID, nil
}
func (tx *Tx) writeBitmapWALPage(pgno uint32, page []byte) (walID int64, err error) {
if err := tx.ensureWritableWALSegment(); err != nil {
// Append to WAL and increment WAL size.
if _, err := w.Write(page); err != nil {
return 0, err
}
tx.walPageN++
// Write header page for next bitmap page.
buf := make([]byte, PageSize)
writePageNo(buf[:], pgno)
writeFlags(buf[:], PageTypeBitmapHeader)
// TODO(BBJ): Write checksum.
if _, err := tx.writeWALPage(buf, false); err != nil {
return 0, fmt.Errorf("write bitmap header: %w", err)
}
// Write the bitmap page and return its WALID.
return tx.writeWALPage(page, false)
}
func (tx *Tx) ensureWritableWALSegment() error {
// Ignore if we still have space in the write cache.
writeCacheSize := int64(len(tx.wcache))
if len(tx.segments) != 0 && activeWALSegment(tx.segments).Size()+writeCacheSize < int64(tx.db.cfg.MaxWALSegmentFileSize) {
return nil
}
// Flush write cache out to file before adding new segment.
if err := tx.flushWALWriter(); err != nil {
return err
}
// If we have a current active WAL segment then close it and start the
// next segment from the next WAL ID. If there is no existing WAL segments,
// read the last checkpointed WAL ID from the DB and start after that.
var base int64
if len(tx.segments) != 0 {
base = activeWALSegment(tx.segments).MaxWALID() + 1
} else {
page, err := tx.readPage(0)
if err != nil {
return err
}
base = readMetaWALID(page) + 1
}
// Create new segment file.
s := tx.db.NewWALSegment(filepath.Join(tx.db.WALPath(), FormatWALSegmentPath(base)))
if err := s.Open(); err != nil {
return fmt.Errorf("add wal segment: %w", err)
}
// Track all segments that need to be added back to DB.
// The DB can remove segments in the background so we don't want to replace.
tx.segments = append(tx.segments, s)
tx.updatedSegmentPaths = append(tx.updatedSegmentPaths, s.Path)
return nil
return walID, nil
}
// Pages returns meta & record data for a list of pages.
@ -2019,3 +1925,19 @@ type BitmapPage struct {
type FreePage struct {
*FreePageInfo
}
// dirtyPageMapKeys returns a sorted slice slice of keys for a dirty page map.
func dirtyPageMapKeys(m map[uint32][]byte) []uint32 {
a := make([]uint32, 0, len(m))
for k := range m {
a = append(a, k)
}
sort.Sort(uint32Slice(a))
return a
}
type uint32Slice []uint32
func (p uint32Slice) Len() int { return len(p) }
func (p uint32Slice) Less(i, j int) bool { return p[i] < p[j] }
func (p uint32Slice) Swap(i, j int) { p[i], p[j] = p[j], p[i] }

329
rbf/wal.go
View file

@ -1,329 +0,0 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package rbf
import (
"fmt"
"os"
"path/filepath"
"sort"
"syscall"
"github.com/pilosa/pilosa/v2/syswrap"
)
var _ = sort.Search
// WALSegment represents a single file in the WAL.
type WALSegment struct {
db *DB
Path string // path to file
MinWALID int64 // base WALID; calculated from path
PageN int // number of written pages
data []byte // read-only mmap data
}
// NewWALSegment returns a new instance of WALSegment for a given path.
func (db *DB) NewWALSegment(path string) WALSegment {
return WALSegment{
db: db,
Path: path,
}
}
// MaxWALID returns the maximum WAL ID of the segment. Only available after Open().
func (s WALSegment) MaxWALID() int64 {
return s.MinWALID + int64(s.PageN) - 1
}
// Size returns the current size of the segment, in bytes.
func (s WALSegment) Size() int64 {
return int64(s.PageN) * PageSize
}
func (s *WALSegment) Open() (err error) {
// Extract base WAL ID and validate path.
if s.MinWALID, err = ParseWALSegmentPath(s.Path); err != nil {
return err
}
// Determine file size & create if necessary.
var sz int64
if fi, err := os.Stat(s.Path); os.IsNotExist(err) {
if f, err := os.OpenFile(s.Path, os.O_RDWR|os.O_CREATE, 0666); err != nil {
return fmt.Errorf("touch wal segment file: %w", err)
} else if err := f.Close(); err != nil {
return fmt.Errorf("close touched wal segment file: %w", err)
}
} else if err != nil {
return fmt.Errorf("stat wal segment file: %w", err)
} else {
sz = fi.Size()
}
// Determine page count & truncate if a partial page is written.
s.PageN = int(sz / PageSize)
if sz%PageSize != 0 {
sz = int64(s.PageN * PageSize)
if err := s.db.truncate(s.Path, sz); err != nil {
return fmt.Errorf("truncate wal file: %w", err)
}
}
// Default the mmap size to the max size plus a page of padding for bitmap pages.
// If the actual size is larger, then increase to that size.
mmapSize := int64(s.db.cfg.MaxWALSegmentFileSize + PageSize)
if sz > mmapSize {
mmapSize = sz
}
// Open file as a read-only memory map.
if f, err := os.OpenFile(s.Path, os.O_RDONLY, 0666); err != nil {
return fmt.Errorf("open wal segment file: %w", err)
} else if s.data, err = syswrap.Mmap(int(f.Fd()), 0, int(mmapSize), syscall.PROT_READ, syscall.MAP_SHARED); err != nil {
f.Close()
return fmt.Errorf("mmap wal segment: %w", err)
} else if err := f.Close(); err != nil {
return fmt.Errorf("close wal segment mmap file: %w", err)
}
return nil
}
// Close closes the write handle and the read-only mmap.
func (s *WALSegment) Close() error {
if s.data != nil {
if err := syswrap.Munmap(s.data); err != nil {
return err
}
s.data = nil
}
return nil
}
// ReadWALPage reads a single page at the given WAL ID.
func (s *WALSegment) ReadWALPage(walID int64) ([]byte, error) {
// Ensure requested ID is contained in this file.
if walID < s.MinWALID || walID > s.MinWALID+int64(s.PageN) {
return nil, fmt.Errorf("wal segment page read out of range: id=%d base=%d pageN=%d", walID, s.MinWALID, s.PageN)
}
offset := (walID - s.MinWALID) * PageSize
return s.data[offset : offset+PageSize], nil
}
func walSegmentByPath(segments []WALSegment, path string) *WALSegment {
for i := range segments {
if segments[i].Path == path {
return &segments[i]
}
}
return nil
}
func activeWALSegment(segments []WALSegment) WALSegment {
if len(segments) == 0 {
return WALSegment{}
}
return segments[len(segments)-1]
}
func minWALID(segments []WALSegment) int64 {
if len(segments) == 0 {
return 0
}
return segments[0].MinWALID
}
func maxWALID(segments []WALSegment) int64 {
if len(segments) == 0 {
return 0
}
s := segments[len(segments)-1]
return s.MaxWALID()
}
func walSize(segments []WALSegment) int64 {
var sz int64
for _, s := range segments {
sz += s.Size()
}
return sz
}
// readWALPage reads a single page at the given WAL ID.
func readWALPage(segments []WALSegment, walID int64) ([]byte, error) {
n := len(segments)
i := sort.Search(n, func(i int) bool {
return walID < segments[i].MinWALID
})
minWALID, maxWALID := int64(-1), int64(-1)
if i > 0 {
s := segments[i-1]
minWALID = s.MinWALID
maxWALID = s.MaxWALID()
if walID >= minWALID && walID <= maxWALID {
return s.ReadWALPage(walID)
}
}
// ok, we're about to error, which should never happen.
// So we can afford to provide detailed diagnostics.
// Report min and max WALID over all supplied segments.
for _, s := range segments {
if minWALID < 0 || s.MinWALID < minWALID {
minWALID = s.MinWALID
}
max := s.MaxWALID()
if maxWALID < 0 || max > maxWALID {
maxWALID = max
}
}
// show all the current segments too.
detail := WALSegmentsAsString(segments)
return nil, fmt.Errorf("cannot find segment containing WAL page: %d; over all supplied segments, minWALID=%v, maxWALID=%v; detail='%v'", walID, minWALID, maxWALID, detail)
}
var ErrNoMetaFound = fmt.Errorf("no meta page found")
func findNextWALMetaPage(segments []WALSegment, walID int64) (metaWALID int64, err error) {
maxWALID := maxWALID(segments)
for ; walID <= maxWALID; walID++ {
// Read page data from WAL and return if it is a meta page.
page, err := readWALPage(segments, walID)
if err != nil {
return walID, err
} else if IsMetaPage(page) {
return walID, nil
}
// Skip over next page if this is a bitmap header.
if IsBitmapHeader(page) {
walID++
}
}
return -1, ErrNoMetaFound
}
func findLastWALMetaPage(segments []WALSegment) (walID int64, err error) {
if len(segments) == 0 {
return 0, nil
}
var maxMetaWALID int64
maxWALID := maxWALID(segments)
for walID := minWALID(segments); walID <= maxWALID; walID++ {
if page, err := readWALPage(segments, walID); err != nil {
return walID, err
} else if IsBitmapHeader(page) {
walID++ // skip next page for bitmap headers
} else if IsMetaPage(page) {
maxMetaWALID = walID // save max meta WAL ID
}
}
return maxMetaWALID, nil
}
// truncateWALAfter removes all pages in the WAL after walID.
func (db *DB) truncateWALAfter(segments []WALSegment, walID int64) ([]WALSegment, error) {
var newSegments []WALSegment
for i := range segments {
segment := &segments[i]
// Append entire segment if WAL range entirely before target WAL ID.
if walID > segment.MaxWALID() {
newSegments = append(newSegments, *segment)
continue
}
// If we only remove some of the WAL pages then truncate and append.
if segment.MinWALID < walID {
newSegment := *segment
newSegment.PageN = int((walID - newSegment.MinWALID) + 1)
if err := db.truncate(newSegment.Path, int64(newSegment.PageN)*PageSize); err != nil {
return segments, err
}
newSegments = append(newSegments, newSegment)
continue
}
// Drop entire segment if all pages are after WAL ID.
if err := segment.Close(); err != nil {
return segments, err
} else if err := os.Remove(segment.Path); err != nil {
return segments, err
}
}
return newSegments, nil
}
func DumpWALSegments(segments []WALSegment) {
fmt.Printf("WAL (%d segments)\n", len(segments))
for i, s := range segments {
fmt.Printf("[%d] WALIDs=(%d-%d) PageN=%d\n", i, s.MinWALID, s.MaxWALID(), s.PageN)
}
}
func WALSegmentsAsString(segments []WALSegment) (r string) {
r = fmt.Sprintf("WAL (%d segments)\n", len(segments))
for i, s := range segments {
r += fmt.Sprintf("[%d] WALIDs=(%d-%d) PageN=%d\n", i, s.MinWALID, s.MaxWALID(), s.PageN)
}
return
}
// FormatWALSegmentPath returns a path for a WAL segment using a WAL ID.
func FormatWALSegmentPath(walID int64) string {
return fmt.Sprintf("%016x.wal", walID)
}
// ParseWALSegmentPath returns the WAL ID for a given WAL segment path.
func ParseWALSegmentPath(s string) (walID int64, err error) {
if _, err = fmt.Sscanf(filepath.Base(s), "%016x.wal", &walID); err != nil {
return 0, fmt.Errorf("invalid WAL path: %s", s)
}
return walID, nil
}
// uint32Hasher implements Hasher for uint32 keys.
type uint32Hasher struct{}
// Hash returns a hash for key.
func (h *uint32Hasher) Hash(key interface{}) uint32 {
return hashUint64(uint64(key.(uint32)))
}
// Equal returns true if a is equal to b. Otherwise returns false.
// Panics if a and b are not ints.
func (h *uint32Hasher) Equal(a, b interface{}) bool {
return a.(uint32) == b.(uint32)
}
// hashUint64 returns a 32-bit hash for a 64-bit value.
func hashUint64(value uint64) uint32 {
hash := value
for value > 0xffffffff {
value /= 0xffffffff
hash ^= value
}
return uint32(hash)
}

186
rbf/wal_test.go
View file

@ -1,186 +0,0 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package rbf_test
import (
// "bytes"
// "encoding/hex"
"io/ioutil"
// "math/rand"
"os"
"path/filepath"
"testing"
"github.com/pilosa/pilosa/v2/rbf"
)
func TestWALSegment_Open(t *testing.T) {
t.Run("OK", func(t *testing.T) {
db := MustOpenDB(t)
defer MustCloseDB(t, db)
s := MustOpenWALSegment(t, db, 10)
defer MustCloseWALSegment(t, s)
if got, want := s.MinWALID, int64(10); got != want {
t.Fatalf("Base()=%d, want %d", got, want)
} else if got, want := s.PageN, 0; got != want {
t.Fatalf("PageN()=%d, want %d", got, want)
}
})
// TODO(BBJ): Test open w/ partially written pages.
}
/*
func TestWALSegment_WritePage(t *testing.T) {
rand := rand.New(rand.NewSource(0))
s := MustOpenWALSegment(t, 10)
defer MustCloseWALSegment(t, s)
pages := [][]byte{
make([]byte, rbf.PageSize),
make([]byte, rbf.PageSize),
}
rand.Read(pages[0])
rand.Read(pages[1])
// Write first page.
if walID, err := s.WriteWALPage(pages[0], false); err != nil {
t.Fatal(err)
} else if got, want := walID, int64(10); got != want {
t.Fatalf("WALID=%d, want %d", got, want)
} else if got, want := s.PageN(), 1; got != want {
t.Fatalf("PageN()=%d, want %d", got, want)
}
// Write second page.
if walID, err := s.WriteWALPage(pages[1], false); err != nil {
t.Fatal(err)
} else if got, want := walID, int64(11); got != want {
t.Fatalf("WALID=%d, want %d", got, want)
} else if got, want := s.PageN(), 2; got != want {
t.Fatalf("PageN()=%d, want %d", got, want)
}
// Read & verify first page.
if buf, err := s.ReadWALPage(10); err != nil {
t.Fatal(err)
} else if !bytes.Equal(pages[0], buf) {
t.Fatalf("unexpected first page:\n%s", hex.Dump(buf))
}
// Read & verify second page.
if buf, err := s.ReadWALPage(11); err != nil {
t.Fatal(err)
} else if !bytes.Equal(pages[1], buf) {
t.Fatal("unexpected second page")
}
}
*/
func TestFormatWALSegmentPath(t *testing.T) {
if got, want := rbf.FormatWALSegmentPath(1234), "00000000000004d2.wal"; got != want {
t.Fatalf("FormatWALSegmentPath()=%q, want %q", got, want)
}
}
func TestParseWALSegmentPath(t *testing.T) {
t.Run("OK", func(t *testing.T) {
if walID, err := rbf.ParseWALSegmentPath("/tmp/00000000000004d2.wal"); err != nil {
t.Fatal(err)
} else if got, want := walID, int64(1234); got != want {
t.Fatalf("ParseWALSegmentPath()=%q, want %q", got, want)
}
})
t.Run("ErrInvalidWALPath", func(t *testing.T) {
if _, err := rbf.ParseWALSegmentPath("/tmp/xyz"); err == nil || err.Error() != "invalid WAL path: /tmp/xyz" {
t.Fatalf("unexpected error: %#v", err)
}
})
}
/*
func BenchmarkWALSegment_WriteWALPage(b *testing.B) {
b.Run("8KB", func(b *testing.B) { benchmarkWALSegment_WriteWALPage(b, 8*(1<<10)) })
b.Run("16KB", func(b *testing.B) { benchmarkWALSegment_WriteWALPage(b, 16*(1<<10)) })
b.Run("64KB", func(b *testing.B) { benchmarkWALSegment_WriteWALPage(b, 64*(1<<10)) })
b.Run("256KB", func(b *testing.B) { benchmarkWALSegment_WriteWALPage(b, 256*(1<<10)) })
b.Run("1MB", func(b *testing.B) { benchmarkWALSegment_WriteWALPage(b, (1 << 20)) })
b.Run("10MB", func(b *testing.B) { benchmarkWALSegment_WriteWALPage(b, 10*(1<<20)) })
}
func benchmarkWALSegment_WriteWALPage(b *testing.B, flushSize int) {
page := make([]byte, rbf.PageSize)
for i := 0; i < b.N; i++ {
func() {
s := MustOpenWALSegment(b, 0)
defer MustCloseWALSegment(b, s)
// Fill the segment but stop after each flush interval to flush the write buffer.
for j := 0; j < rbf.MaxWALSegmentFileSize; j += rbf.PageSize {
if _, err := s.WriteWALPage(page, false); err != nil {
b.Fatal(err)
}
// Flush write buffer.
if j != 0 && j%flushSize == 0 {
if err := s.Flush(); err != nil {
b.Fatal(err)
}
}
}
// Fsync to disk at the end.
if err := s.Sync(); err != nil {
b.Fatal(err)
}
}()
}
b.SetBytes(rbf.MaxWALSegmentFileSize)
}
*/
// MustOpenWALSegment opens a WAL segment in a temporary path. Fails on error.
func MustOpenWALSegment(tb testing.TB, db *rbf.DB, walID int64) rbf.WALSegment {
tb.Helper()
dir, err := ioutil.TempDir("", "")
if err != nil {
tb.Fatal(err)
}
path := filepath.Join(dir, rbf.FormatWALSegmentPath(walID))
if err := ioutil.WriteFile(path, nil, 0666); err != nil {
tb.Fatal(err)
}
s := db.NewWALSegment(path)
if err := s.Open(); err != nil {
tb.Fatal(err)
}
return s
}
// MustCloseWALSegment closes s. Fails on error.
func MustCloseWALSegment(tb testing.TB, s rbf.WALSegment) {
tb.Helper()
if err := s.Close(); err != nil {
tb.Fatal(err)
} else if err := os.Remove(s.Path); err != nil {
tb.Fatal(err)
}
}

10
txfactory.go
View file

@ -128,10 +128,6 @@ type Qcx struct {
// efficient access to the options for RequiredForAtomicWriteTx
RequiredTxo *Txo
// Option for direct writes to the database. RBF only.
// This option is unsafe and should only be used for imports.
Direct bool
isRoaring bool
// top-level context is for a write, so re-use a
@ -255,11 +251,6 @@ func (qcx *Qcx) GetTx(o Txo) (tx Tx, finisher func(perr *error), err error) {
return nil, nil, ErrQcxDone
}
// Use direct option if set on QCX.
if qcx.Direct {
o.Direct = true
}
// roaring uses finer grain, a file per fragment rather than
// db per shard. So we can't re-use the readTx. Moreover,
// roaring Tx are No-ops anyway, so just give it a new Tx
@ -573,7 +564,6 @@ func (f *TxFactory) UseRowCache() bool {
// Txo holds the transaction options
type Txo struct {
Write bool
Direct bool // directly write to the database. rbf only. (unsafe)
Field *Field
Index *Index
Fragment *fragment