dataspace

1
package dataspace
2

3
import (
4
	"bufio"
5
	"bytes"
6
	"errors"
7
	"io"
8
	"os"
9
	"sync"
10

11
	"github.com/RoaringBitmap/roaring/v2/roaring64"
12
	"github.com/zeebo/xxh3"
13
)
14

15
////////////////////////////////////////////////////////////////////////////////
16

17
const (
18
	Ascend uint8 = iota
19
	Descend
20
	Intersection
21
	Union
22
)
23

24
var (
25
	ErrKeyNotFound         = errors.New("key not found")
26
	ErrKeyNil              = errors.New("key is nil")
27
	ErrKeyMissing          = errors.New("empty key")
28
	ErrValueNil            = errors.New("value is nil")
29
	ErrValueMissing        = errors.New("empty value")
30
	ErrIncorrectOrder      = errors.New("incorrect order: must be Ascend or Descend")
31
	ErrIncorectOperation   = errors.New("incorrect search operation: must be Intersection or Union")
32
	ErrFactorsNotSpecified = errors.New("search factors not specified")
33
	ErrFactorsNotCreated   = errors.New("search factors not created")
34
	ErrFactorNotFound      = errors.New("search factor not found")
35
)
36

37
// //////////////////////////////////////////////////////////////////////////////
38

39
type Space struct {
40
	sync.RWMutex                              // мьютекс
41
	wal          *os.File                     // файл
42
	writer       *bufio.Writer                // обработчик записи
43
	tree         *node                        // сбалансированное дерево
44
	list         *list                        // двусвязный список
45
	bitmaps      map[string]*roaring64.Bitmap // битовые карты поиска
46
}
47

48
type bitmapStat struct {
49
	Name        string `json:"name"`
50
	Length      uint64 `json:"length"`
51
	SizeInBytes uint64 `json:"size-in-bytes"`
52
}
53

54
type Info struct {
55
	FileName     string        `json:"file-name"`
56
	FileSize     int64         `json:"file-size"`
57
	Length       uint64        `json:"length"`
58
	LastModified string        `json:"last-modified"`
59
	TreeBranches int64         `json:"tree-branches"`
60
	Bitmaps      []*bitmapStat `json:"bitmaps"`
61
}
62

63
////////////////////////////////////////////////////////////////////////////////
64

65
// Конструктор нового пространства данных.
66
func New(path string) (*Space, error) {
67
	w, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600)
68
	if err != nil {
69
		return nil, err
70
	}
71
	s := &Space{
72
		wal:     w,
73
		writer:  bufio.NewWriter(w),
74
		tree:    &node{},
75
		list:    newList(),
76
		bitmaps: make(map[string]*roaring64.Bitmap),
77
	}
78
	stat, err := s.wal.Stat()
79
	if err != nil {
80
		return nil, err
81
	}
82
	b := make([]byte, stat.Size())
83
	if _, err := w.Read(b); err != nil {
84
		return nil, err
85
	}
86
	s.load(b)
87
	return s, nil
88
}
89

90
// Метод создаёт новый фактор поиска.
91
func (s *Space) CreateSearchFactor(name string) error {
92
	s.Lock()
93
	defer s.Unlock()
94
	if _, ok := s.bitmaps[name]; ok {
95
		return nil
96
	}
97
	s.bitmaps[name] = roaring64.NewBitmap()
98
	data := encode(&fraction{
99
		m: 2,
100
		s: true,
101
		k: []byte(name),
102
		v: nil,
103
	})
104
	if _, err := s.writer.Write(data); err != nil {
105
		return err
106
	}
107
	return s.writer.Flush()
108
}
109

110
// Метод удаляет фактор поиска.
111
func (s *Space) DeleteSearchFactor(name string) error {
112
	s.Lock()
113
	defer s.Unlock()
114
	if b, ok := s.bitmaps[name]; ok {
115
		b.Clear()
116
		delete(s.bitmaps, name)
117
	}
118
	data := encode(&fraction{
119
		m: 2,
120
		s: false,
121
		k: []byte(name),
122
		v: nil,
123
	})
124
	if _, err := s.writer.Write(data); err != nil {
125
		return err
126
	}
127
	return s.writer.Flush()
128
}
129

130
// Метод записывает данные в пространство данных.
131
func (s *Space) Set(k, v []byte) error {
132
	switch {
133
	case k == nil:
134
		return ErrKeyNil
135
	case len(k) == 0:
136
		return ErrKeyMissing
137
	case v == nil:
138
		return ErrValueNil
139
	case len(v) == 0:
140
		return ErrValueMissing
141
	}
142
	h := xxh3.Hash(k)
143
	s.Lock()
144
	defer s.Unlock()
145
	if node := s.tree.search(h); node != nil {
146
		if bytes.Equal(node.value, v) {
147
			return nil
148
		}
149
		s.tree = s.tree.add(h, k, v, node.element)
150
	} else {
151
		s.tree = s.tree.add(h, k, v, s.list.push(h))
152
	}
153
	data := encode(&fraction{
154
		m: 1,
155
		s: true,
156
		k: k,
157
		v: v,
158
	})
159
	if _, err := s.writer.Write(data); err != nil {
160
		return err
161
	}
162
	return s.writer.Flush()
163
}
164

165
// Метод получает данные из пространства данных.
166
func (s *Space) Get(k []byte) ([]byte, error) {
167
	switch {
168
	case k == nil:
169
		return nil, ErrKeyNil
170
	case len(k) == 0:
171
		return nil, ErrKeyMissing
172
	}
173
	s.RLock()
174
	defer s.RUnlock()
175
	if node := s.tree.search(xxh3.Hash(k)); node != nil {
176
		return node.value, nil
177
	}
178
	return nil, ErrKeyNotFound
179
}
180

181
// Метод удаляет данные из пространства данных.
182
func (s *Space) Delete(k []byte) error {
183
	switch {
184
	case k == nil:
185
		return ErrKeyNil
186
	case len(k) == 0:
187
		return ErrKeyMissing
188
	}
189
	h := xxh3.Hash(k)
190
	s.Lock()
191
	defer s.Unlock()
192
	if node := s.tree.search(h); node != nil {
193
		s.list.remove(node.element)
194
	} else {
195
		return ErrKeyNotFound
196
	}
197
	s.tree.remove(h)
198
	data := encode(&fraction{
199
		m: 1,
200
		s: false,
201
		k: k,
202
		v: []byte{},
203
	})
204
	if _, err := s.writer.Write(data); err != nil {
205
		return err
206
	}
207
	for nm, b := range s.bitmaps {
208
		if !b.Contains(h) {
209
			continue
210
		}
211
		b.Remove(h)
212
		data := encode(&fraction{
213
			m: 3,
214
			s: false,
215
			k: []byte(nm),
216
			v: k,
217
		})
218
		if _, err := s.writer.Write(data); err != nil {
219
			return err
220
		}
221
	}
222
	return s.writer.Flush()
223
}
224

225
/*
226
Метод итерирует диапазон значений.
227
Первым аргументом должен быть указан порядок итерации:
228

229
	Ascend или 0 - возрастание
230
	Descend или 1 - убывание
231

232
Вторым аргументом должно быть количество (limit) итерируемых элементов.
233
Если в качестве аргумента указан 0 - итерация будет по всем элементам.
234
Если значение больше чем количество элементов - итерация так же будет
235
по всем элементам.
236
Третим аргументом должен быть сдвиг (offset). Например сдвиг 5 значит,
237
что итерация начнется с 5 элемента.
238
*/
239
func (s *Space) Range(by uint8, limit, offset uint64) ([][]byte, error) {
240
	switch {
241
	case by != Ascend && by != Descend:
242
		return nil, ErrIncorrectOrder
243
	case s.list.len == 0:
244
		return nil, nil
245
	case limit == 0 || limit > s.list.len:
246
		limit = s.list.len
247
	}
248
	values := make([][]byte, 0, limit)
249
	s.RLock()
250
	defer s.RUnlock()
251
	switch by {
252
	case Ascend:
253
		for e := s.list.root.next; e != nil; e = e.nextElement() {
254
			if offset != 0 {
255
				offset--
256
				continue
257
			}
258
			if node := s.tree.search(e.value); node != nil {
259
				values = append(values, node.value)
260
			} else {
261
				return nil, ErrKeyNotFound
262
			}
263
			limit--
264
			if limit == 0 {
265
				break
266
			}
267
		}
268
	case Descend:
269
		for e := s.list.root.prev; e != nil; e = e.prevElement() {
270
			if offset != 0 {
271
				offset--
272
				continue
273
			}
274
			if node := s.tree.search(e.value); node != nil {
275
				values = append(values, node.value)
276
			} else {
277
				return nil, ErrKeyNotFound
278
			}
279
			limit--
280
			if limit == 0 {
281
				break
282
			}
283
		}
284
	}
285
	return values, nil
286
}
287

288
// Метод записывает ключ в факторы поиска.
289
func (s *Space) SetToSearchFactors(k []byte, factors ...string) error {
290
	switch {
291
	case k == nil:
292
		return ErrKeyNil
293
	case len(k) == 0:
294
		return ErrKeyMissing
295
	case len(factors) == 0:
296
		return ErrFactorsNotSpecified
297
	case len(s.bitmaps) == 0:
298
		return ErrFactorsNotCreated
299
	}
300
	h := xxh3.Hash(k)
301
	s.Lock()
302
	defer s.Unlock()
303
	for _, f := range factors {
304
		if b, ok := s.bitmaps[f]; ok {
305
			if b.Contains(h) {
306
				continue
307
			}
308
			b.Add(h)
309
			data := encode(&fraction{
310
				m: 3,
311
				s: true,
312
				k: []byte(f),
313
				v: k,
314
			})
315
			if _, err := s.writer.Write(data); err != nil {
316
				return err
317
			}
318
		}
319
	}
320
	return s.writer.Flush()
321
}
322

323
// Метод удаляет ключ из факторов поиска.
324
func (s *Space) DeleteFromSearchFactors(k []byte, factors ...string) error {
325
	switch {
326
	case k == nil:
327
		return ErrKeyNil
328
	case len(k) == 0:
329
		return ErrKeyMissing
330
	case len(factors) == 0:
331
		return ErrFactorsNotSpecified
332
	case len(s.bitmaps) == 0:
333
		return ErrFactorsNotCreated
334
	}
335
	h := xxh3.Hash(k)
336
	s.Lock()
337
	defer s.Unlock()
338
	for _, f := range factors {
339
		if b, ok := s.bitmaps[f]; ok {
340
			if !b.Contains(h) {
341
				continue
342
			}
343
			b.Remove(h)
344
			data := encode(&fraction{
345
				m: 3,
346
				s: false,
347
				k: []byte(f),
348
				v: k,
349
			})
350
			if _, err := s.writer.Write(data); err != nil {
351
				return err
352
			}
353
		}
354
	}
355
	return s.writer.Flush()
356
}
357

358
/*
359
Метод осуществляет поиск значений по базе данных, которые соответствуют
360
поисковым факторам, переданным в аргументах.
361
*/
362
func (s *Space) Search(by uint8, factors ...string) ([][]byte, error) {
363
	switch {
364
	case by != Intersection && by != Union:
365
		return nil, ErrIncorectOperation
366
	case len(factors) == 0:
367
		return nil, ErrFactorsNotSpecified
368
	case len(s.bitmaps) == 0:
369
		return nil, ErrFactorsNotCreated
370
	}
371
	v := make([][]byte, 0)
372
	s.RLock()
373
	defer s.RUnlock()
374
	if len(factors) == 1 {
375
		if b, ok := s.bitmaps[factors[0]]; ok {
376
			for _, k := range b.ToArray() {
377
				if node := s.tree.search(k); node != nil {
378
					v = append(v, node.value)
379
				} else {
380
					return nil, ErrKeyNotFound
381
				}
382
			}
383
		} else {
384
			return nil, ErrFactorNotFound
385
		}
386
		return v, nil
387
	}
388
	b := make([]*roaring64.Bitmap, 0, len(factors))
389
	for _, f := range factors {
390
		if bm, ok := s.bitmaps[f]; ok {
391
			b = append(b, bm)
392
		} else {
393
			return nil, ErrFactorNotFound
394
		}
395
	}
396
	switch by {
397
	case Intersection:
398
		for _, k := range roaring64.FastAnd(b...).ToArray() {
399
			if node := s.tree.search(k); node != nil {
400
				v = append(v, node.value)
401
			} else {
402
				return nil, ErrKeyNotFound
403
			}
404
		}
405
	case Union:
406
		for _, k := range roaring64.FastOr(b...).ToArray() {
407
			if node := s.tree.search(k); node != nil {
408
				v = append(v, node.value)
409
			} else {
410
				return nil, ErrKeyNotFound
411
			}
412
		}
413
	}
414
	return v, nil
415
}
416

417
// Метод возвращает информацию о пространстве данных.
418
func (s *Space) Info() (*Info, error) {
419
	i, err := s.wal.Stat()
420
	if err != nil {
421
		return nil, err
422
	}
423
	bitmaps := make([]*bitmapStat, 0, len(s.bitmaps))
424
	for k, b := range s.bitmaps {
425
		bitmaps = append(bitmaps, &bitmapStat{
426
			Name:        k,
427
			Length:      b.GetCardinality(),
428
			SizeInBytes: b.GetSizeInBytes(),
429
		})
430
	}
431
	return &Info{
432
		FileName:     i.Name(),
433
		FileSize:     i.Size(),
434
		Length:       s.list.len,
435
		LastModified: i.ModTime().Format("02.01.06 15:04"),
436
		TreeBranches: countBranches(s.tree),
437
		Bitmaps:      bitmaps,
438
	}, nil
439
}
440

441
/*
442
Метод создаёт резервный временный файл, затем копирует в него
443
все данные. Далее очищает данные из основного wal-журнала и
444
перезаписывает в порядке добавления с помощью списка.
445
*/
446
func (s *Space) Vacuum(path string) error {
447
	f, err := os.Create(path + ".temp")
448
	if err != nil {
449
		return err
450
	}
451
	s.Lock()
452
	defer s.Unlock()
453
	if _, err := s.wal.Seek(0, 0); err != nil {
454
		return err
455
	}
456
	if _, err := io.Copy(f, s.wal); err != nil {
457
		return err
458
	}
459
	if err := s.wal.Truncate(0); err != nil {
460
		return err
461
	}
462
	if _, err := s.wal.Seek(0, 0); err != nil {
463
		return err
464
	}
465
	if err := s.save(); err != nil {
466
		return err
467
	}
468
	if err := f.Close(); err != nil {
469
		return err
470
	}
471
	return os.Remove(path + ".temp")
472
}
473

474
// Метод закрывает пространство данных.
475
func (s *Space) Close() error {
476
	if err := s.wal.Sync(); err != nil {
477
		return err
478
	}
479
	return s.wal.Close()
480
}
481