/
branch.go
93 lines (88 loc) · 2.15 KB
/
branch.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
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
package lakemanage
import (
"context"
"github.com/brimdata/zed/api"
lakeapi "github.com/brimdata/zed/lake/api"
"github.com/brimdata/zed/lake/pools"
"github.com/brimdata/zed/lakeparse"
"github.com/segmentio/ksuid"
"go.uber.org/zap"
"golang.org/x/sync/errgroup"
)
type branch struct {
config PoolConfig
lake lakeapi.Interface
logger *zap.Logger
pool *pools.Config
}
func newBranch(c Config, pool *pools.Config, lake lakeapi.Interface, logger *zap.Logger) *branch {
config := c.poolConfig(pool)
logger = logger.Named("pool").With(
zap.String("name", pool.Name),
zap.Stringer("id", pool.ID),
zap.String("branch", config.Branch),
zap.Bool("vectors", config.Vectors),
)
return &branch{
config: config,
lake: lake,
logger: logger,
pool: pool,
}
}
func (b *branch) run(ctx context.Context) error {
b.logger.Debug("compaction started")
head := lakeparse.Commitish{Pool: b.pool.Name, Branch: b.config.Branch}
it, err := newObjectIterator(ctx, b.lake, &head)
if err != nil {
return err
}
defer it.close()
runCh := make(chan []ksuid.KSUID)
vecCh := make(chan ksuid.KSUID)
group, ctx := errgroup.WithContext(ctx)
group.Go(func() error {
err := scan(ctx, it, b.pool, runCh, vecCh)
close(runCh)
close(vecCh)
return err
})
var found int
var compacted int
var vectors int
group.Go(func() error {
for run := range runCh {
commit, err := b.lake.Compact(ctx, b.pool.ID, b.config.Branch, run, b.config.Vectors, api.CommitMessage{})
if err != nil {
return err
}
found++
compacted += len(run)
b.logger.Debug("compacted", zap.Stringer("commit", commit), zap.Int("objects_compacted", len(run)))
}
return nil
})
group.Go(func() error {
var oids []ksuid.KSUID
for oid := range vecCh {
if b.config.Vectors {
oids = append(oids, oid)
}
}
if len(oids) == 0 {
return nil
}
_, err := b.lake.AddVectors(ctx, head.Pool, head.Branch, oids, api.CommitMessage{})
if err == nil {
vectors += len(oids)
}
return err
})
err = group.Wait()
b.logger.Info("compaction completed",
zap.Int("runs_found", found),
zap.Int("objects_compacted", compacted),
zap.Int("vectors_created", vectors),
)
return err
}