-
Notifications
You must be signed in to change notification settings - Fork 44
/
writer.go
76 lines (59 loc) · 2.15 KB
/
writer.go
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
package index
import (
"context"
"fmt"
"github.com/streamingfast/substreams/block"
"go.uber.org/zap"
"github.com/RoaringBitmap/roaring/roaring64"
"github.com/streamingfast/dstore"
"github.com/streamingfast/substreams/manifest"
pbsubstreams "github.com/streamingfast/substreams/pb/sf/substreams/v1"
)
type Writer struct {
indexFile *File
}
func NewWriter(indexFile *File) *Writer {
return &Writer{
indexFile: indexFile,
}
}
func (w *Writer) Write(indexes map[string]*roaring64.Bitmap) {
w.indexFile.Set(indexes)
}
func (w *Writer) Close(ctx context.Context) error {
err := w.indexFile.Save(ctx)
if err != nil {
return fmt.Errorf("saving index file %s: %w", w.indexFile.moduleName, err)
}
return nil
}
// startblock=500
// look for 0->1000
// GenrateBlockIndexWriters will only generate writers for modules that have no preexisting index file and that are aligned with the bundle size
func GenerateBlockIndexWriters(ctx context.Context, baseStore dstore.Store, indexModules []*pbsubstreams.Module, ModuleHashes *manifest.ModuleHashes, logger *zap.Logger, blockRange *block.Range, bundleSize uint64) (writers map[string]*Writer, existingIndices map[string]map[string]*roaring64.Bitmap, err error) {
writers = make(map[string]*Writer)
existingIndices = make(map[string]map[string]*roaring64.Bitmap)
isAligned := blockRange.StartBlock%bundleSize == 0 && blockRange.ExclusiveEndBlock%bundleSize == 0
if !isAligned { // we align it, but won't write it because it would be missing blocks...
alignedStartBlock := blockRange.StartBlock - (blockRange.StartBlock % bundleSize)
blockRange = &block.Range{
StartBlock: alignedStartBlock,
ExclusiveEndBlock: alignedStartBlock + bundleSize,
}
}
for _, module := range indexModules {
currentFile, err := NewFile(baseStore, ModuleHashes.Get(module.Name), module.Name, logger, blockRange)
if err != nil {
return nil, nil, fmt.Errorf("creating new index file for %q: %w", module.Name, err)
}
if err := currentFile.Load(ctx); err == nil {
existingIndices[module.Name] = currentFile.Indices
continue
}
if !isAligned {
continue
}
writers[module.Name] = NewWriter(currentFile)
}
return
}