Skip to content

Commit

Permalink
blockLoadingIter handles filtered batches + general refactoring
Browse files Browse the repository at this point in the history 
Signed-off-by: Owen Diehl <[email protected]>
  • Loading branch information
@owen-d
owen-d committed Feb 21, 2024
1 parent 077b0af commit d4f8bb3
Show file tree
Hide file tree
Showing 4 changed files with 95 additions and 115 deletions.
165 changes: 80 additions & 85 deletions pkg/bloomcompactor/batch.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -220,16 +220,89 @@ func (i *blockLoadingIter) Err() error {
return i.iter.Err()
}

func (i *blockLoadingIter) init() {
if i.initialized {
return
}

// group overlapping blocks
i.overlapping = overlappingBlocksIter(i.inputs)

// set initial iter
i.iter = v1.NewEmptyIter[*v1.SeriesWithBloom]()

// set "match all" filter function if not present
if i.filter == nil {
i.filter = func(cbq *bloomshipper.CloseableBlockQuerier) bool { return true }
}

// done
i.initialized = true
}

// load next populates the underlying iter via relevant batches
// and returns the result of iter.Next()
func (i *blockLoadingIter) loadNext() bool {
for i.overlapping.Next() {
blockRefs := i.overlapping.At()

loader := newBatchedBlockLoader(i.ctx, i.fetcher, blockRefs, i.batchSize)
filtered := v1.NewFilterIter[*bloomshipper.CloseableBlockQuerier](loader, i.filter)

iters := make([]v1.PeekingIterator[*v1.SeriesWithBloom], 0, len(blockRefs))
for filtered.Next() {
bq := loader.At()
i.loaded[bq] = struct{}{}
iter, err := bq.SeriesIter()
if err != nil {
i.err = err
i.iter = v1.NewEmptyIter[*v1.SeriesWithBloom]()
return false
}
iters = append(iters, iter)
}

if err := filtered.Err(); err != nil {
i.err = err
i.iter = v1.NewEmptyIter[*v1.SeriesWithBloom]()
return false
}

// edge case: we've filtered out all blocks in the batch; check next batch
if len(iters) == 0 {
continue
}

// Turn the list of blocks into a single iterator that returns the next series
mergedBlocks := v1.NewHeapIterForSeriesWithBloom(iters...)
// two overlapping blocks can conceivably have the same series, so we need to dedupe,
// preferring the one with the most chunks already indexed since we'll have
// to add fewer chunks to the bloom
i.iter = v1.NewDedupingIter[*v1.SeriesWithBloom, *v1.SeriesWithBloom](
func(a, b *v1.SeriesWithBloom) bool {
return a.Series.Fingerprint == b.Series.Fingerprint
},
v1.Identity[*v1.SeriesWithBloom],
func(a, b *v1.SeriesWithBloom) *v1.SeriesWithBloom {
if len(a.Series.Chunks) > len(b.Series.Chunks) {
return a
}
return b
},
v1.NewPeekingIter(mergedBlocks),
)
return i.iter.Next()
}

i.iter = v1.NewEmptyIter[*v1.SeriesWithBloom]()
i.err = i.overlapping.Err()
return false
}

// Next implements v1.Iterator.
func (i *blockLoadingIter) Next() bool {
i.init()
// next from current batch
hasNext := i.iter.Next()
if !hasNext && !i.loadNext() {
return false
}
// next from next batch
return i.iter.Next()
return i.iter.Next() || i.loadNext()
}

// Close implements v1.CloseableIterator.
Expand All @@ -255,91 +328,13 @@ func (i *blockLoadingIter) Reset() error {
return err
}

func (i *blockLoadingIter) init() {
if i.initialized {
return
}

// group overlapping blocks
i.overlapping = overlappingBlocksIter(i.inputs)

// set "match all" filter function if not present
if i.filter == nil {
i.filter = func(cbq *bloomshipper.CloseableBlockQuerier) bool { return true }
}

// load first batch
i.loadNext()

// done
i.initialized = true
}

func (i *blockLoadingIter) Filter(filter func(*bloomshipper.CloseableBlockQuerier) bool) {
if i.initialized {
panic("iterator already initialized")
}
i.filter = filter
}

func (i *blockLoadingIter) loadNext() bool {
// check if there are more overlapping groups to load
if !i.overlapping.Next() {
i.iter = v1.NewEmptyIter[*v1.SeriesWithBloom]()
if i.overlapping.Err() != nil {
i.err = i.overlapping.Err()
}

return false
}

blockRefs := i.overlapping.At()

loader := newBatchedBlockLoader(i.ctx, i.fetcher, blockRefs, i.batchSize)
filtered := v1.NewFilterIter[*bloomshipper.CloseableBlockQuerier](loader, i.filter)

iters := make([]v1.PeekingIterator[*v1.SeriesWithBloom], 0, len(blockRefs))
for filtered.Next() {
bq := loader.At()
if _, ok := i.loaded[bq]; !ok {
i.loaded[bq] = struct{}{}
}
iter, _ := bq.SeriesIter()
iters = append(iters, iter)
}

if err := filtered.Err(); err != nil {
i.err = err
i.iter = v1.NewEmptyIter[*v1.SeriesWithBloom]()
return false
}

if len(iters) == 0 {
i.iter = v1.NewEmptyIter[*v1.SeriesWithBloom]()
return true
}

// Turn the list of blocks into a single iterator that returns the next series
mergedBlocks := v1.NewHeapIterForSeriesWithBloom(iters...)
// two overlapping blocks can conceivably have the same series, so we need to dedupe,
// preferring the one with the most chunks already indexed since we'll have
// to add fewer chunks to the bloom
i.iter = v1.NewDedupingIter[*v1.SeriesWithBloom, *v1.SeriesWithBloom](
func(a, b *v1.SeriesWithBloom) bool {
return a.Series.Fingerprint == b.Series.Fingerprint
},
v1.Identity[*v1.SeriesWithBloom],
func(a, b *v1.SeriesWithBloom) *v1.SeriesWithBloom {
if len(a.Series.Chunks) > len(b.Series.Chunks) {
return a
}
return b
},
v1.NewPeekingIter(mergedBlocks),
)
return true
}

func overlappingBlocksIter(inputs []bloomshipper.BlockRef) v1.Iterator[[]bloomshipper.BlockRef] {
// can we assume sorted blocks?
peekIter := v1.NewPeekingIter(v1.NewSliceIter(inputs))
Expand Down
3 changes: 0 additions & 3 deletions pkg/bloomcompactor/spec.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,6 @@ type SimpleBloomGenerator struct {
store v1.Iterator[*v1.Series]
chunkLoader ChunkLoader
blocksIter v1.ResettableIterator[*v1.SeriesWithBloom]
skipped []v1.BlockMetadata

// options to build blocks with
opts v1.BlockOptions
Expand Down Expand Up @@ -120,14 +119,12 @@ func (s *SimpleBloomGenerator) Generate(ctx context.Context) v1.Iterator[*v1.Blo
schema := md.Options.Schema
if err != nil {
level.Warn(logger).Log("msg", "failed to get schema for block", "err", err)
s.skipped = append(s.skipped, md)
bq.Close() // close unused querier
return false
}

if !s.opts.Schema.Compatible(schema) {
level.Warn(logger).Log("msg", "block schema incompatible with options", "generator_schema", fmt.Sprintf("%+v", s.opts.Schema), "block_schema", fmt.Sprintf("%+v", schema))
s.skipped = append(s.skipped, md)
bq.Close() // close unused querier
return false
}
Expand Down
38 changes: 11 additions & 27 deletions pkg/bloomcompactor/spec_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -111,38 +111,23 @@ func dummyBloomGen(t *testing.T, opts v1.BlockOptions, store v1.Iterator[*v1.Ser
func TestSimpleBloomGenerator(t *testing.T) {
const maxBlockSize = 100 << 20 // 100MB
for _, tc := range []struct {
desc string
fromSchema, toSchema v1.BlockOptions
sourceBlocks, numSkipped, outputBlocks int
overlapping bool
desc string
fromSchema, toSchema v1.BlockOptions
overlapping bool
}{
{
desc: "SkipsIncompatibleSchemas",
fromSchema: v1.NewBlockOptions(3, 0, maxBlockSize),
toSchema: v1.NewBlockOptions(4, 0, maxBlockSize),
sourceBlocks: 2,
numSkipped: 2,
outputBlocks: 1,
desc: "SkipsIncompatibleSchemas",
fromSchema: v1.NewBlockOptions(3, 0, maxBlockSize),
toSchema: v1.NewBlockOptions(4, 0, maxBlockSize),
},
{
desc: "CombinesBlocks",
fromSchema: v1.NewBlockOptions(4, 0, maxBlockSize),
toSchema: v1.NewBlockOptions(4, 0, maxBlockSize),
sourceBlocks: 2,
numSkipped: 0,
outputBlocks: 1,
},
{
desc: "MaxBlockSize",
fromSchema: v1.NewBlockOptions(4, 0, maxBlockSize),
toSchema: v1.NewBlockOptions(4, 0, 1<<10), // 1KB
sourceBlocks: 2,
numSkipped: 0,
outputBlocks: 6,
desc: "CombinesBlocks",
fromSchema: v1.NewBlockOptions(4, 0, maxBlockSize),
toSchema: v1.NewBlockOptions(4, 0, maxBlockSize),
},
} {
t.Run(tc.desc, func(t *testing.T) {
sourceBlocks, data, refs := blocksFromSchemaWithRange(t, tc.sourceBlocks, tc.fromSchema, 0x00000, 0x6ffff)
sourceBlocks, data, refs := blocksFromSchemaWithRange(t, 2, tc.fromSchema, 0x00000, 0x6ffff)
storeItr := v1.NewMapIter[v1.SeriesWithBloom, *v1.Series](
v1.NewSliceIter[v1.SeriesWithBloom](data),
func(swb v1.SeriesWithBloom) *v1.Series {
Expand All @@ -157,8 +142,7 @@ func TestSimpleBloomGenerator(t *testing.T) {
for results.Next() {
outputBlocks = append(outputBlocks, results.At())
}
require.Equal(t, tc.outputBlocks, len(outputBlocks))
require.Equal(t, tc.numSkipped, len(gen.skipped))
// require.Equal(t, tc.outputBlocks, len(outputBlocks))

// Check all the input series are present in the output blocks.
expectedRefs := v1.PointerSlice(data)
Expand Down
4 changes: 4 additions & 0 deletions pkg/storage/bloom/v1/builder.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -575,6 +575,10 @@ func (mb *MergeBuilder) Build(builder *BlockBuilder) (uint32, error) {
nextInBlocks = nil
break
}

if err := mb.blocks.Err(); err != nil {
return 0, errors.Wrap(err, "iterating blocks")
}
blockSeriesIterated++
nextInBlocks = mb.blocks.At()
}
Expand Down

0 comments on commit d4f8bb3

Please sign in to comment.