1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
|
package state
import (
"context"
"encoding/binary"
"fmt"
"sync"
"flux/core"
"github.com/cockroachdb/pebble/v2"
)
type PebbleStateBackend struct {
maxParallelism int
currentKey string
currentGroupKey int
currentNamespace any
db *pebble.DB
// for support statename storage optimize
// PebbleDB dont have column family as RockDB does
// manually convert name <-> id
mu sync.Mutex
stateMap map[string]uint8
// nextStateID prevent the hash collision
nextStateID uint8
}
func NewPebbleStateBackend(dir string, maxParalleism int) (*PebbleStateBackend, error) {
db, err := pebble.Open(dir, &pebble.Options{})
if err != nil {
return nil, fmt.Errorf("failed to open pebbledb %w", err)
}
return &PebbleStateBackend{
maxParallelism: maxParalleism,
db: db,
stateMap: make(map[string]uint8),
nextStateID: 1,
}, nil
}
func (p *PebbleStateBackend) SetCurrentKey(key string) error {
p.currentKey = key
p.currentGroupKey = core.AssignToKeyGroup(key, p.maxParallelism)
return nil
}
func (p *PebbleStateBackend) GetCurrentKey() string {
return p.currentKey
}
func (p *PebbleStateBackend) SetCurrentNamespace(namespace any) error {
p.currentNamespace = namespace
return nil
}
func (p *PebbleStateBackend) GetCurrentNamespace() any {
return p.currentNamespace
}
// GetValueState(ctx context.Context, key string) (ValueState[any], error)
func (p *PebbleStateBackend) GetValueState(ctx context.Context, statename string) (core.ValueState[any], error) {
stateID := p.getStateID(statename)
return nil, nil
}
func (p *PebbleStateBackend) getStateID(statename string) uint8 {
p.mu.Lock()
defer p.mu.Unlock()
if _, ok := p.stateMap[statename]; !ok {
id := p.nextStateID
p.stateMap[statename] = id
p.nextStateID++
}
return p.stateMap[statename]
}
func (p *PebbleStateBackend) calculateNamespaceSize() int {
nsLen := 0
switch p.currentNamespace.(type) {
case core.TimeWindow:
nsLen = 16
case string:
nsLen = len(p.currentNamespace.(string))
default:
// void namespace
nsLen = 0
}
return nsLen
}
func (p *PebbleStateBackend) deserialize(statename string) []byte {
// this will store in SSTable, order is important, keygroup MUST be first for scaling
// 2B(KeyGroup) + 1B(StateID) + 2B(KeyLen) + len(Key) + 16B(Window)
kl := len(p.currentKey)
b := make([]byte, 0, p.calculateNamespaceSize()+5+kl)
// add keygroup
b = binary.BigEndian.AppendUint16(b, uint16(p.currentGroupKey))
// add state id
stateID := p.getStateID(statename)
b = append(b, stateID)
// add key
b = binary.BigEndian.AppendUint16(b, uint16(kl))
b = append(b, p.currentKey...)
// add window / namespace
switch ns := p.currentNamespace.(type) {
case core.TimeWindow:
b = binary.BigEndian.AppendUint64(b, uint64(ns.Start))
b = binary.BigEndian.AppendUint64(b, uint64(ns.End))
case string:
b = append(b, ns...)
default:
}
return b
}
|
