| 487 | } |
| 488 | |
| 489 | func (c *LeveledCompactor) CompactWithBlockPopulator(dest string, dirs []string, open []*Block, blockPopulator BlockPopulator) ([]ulid.ULID, error) { |
| 490 | var ( |
| 491 | blocks []BlockReader |
| 492 | bs []*Block |
| 493 | metas []*BlockMeta |
| 494 | uids []string |
| 495 | ) |
| 496 | start := time.Now() |
| 497 | |
| 498 | for _, d := range dirs { |
| 499 | select { |
| 500 | case <-c.ctx.Done(): |
| 501 | return nil, c.ctx.Err() |
| 502 | default: |
| 503 | } |
| 504 | |
| 505 | meta, _, err := readMetaFile(d) |
| 506 | if err != nil { |
| 507 | return nil, err |
| 508 | } |
| 509 | |
| 510 | var b *Block |
| 511 | |
| 512 | // Use already open blocks if we can, to avoid |
| 513 | // having the index data in memory twice. |
| 514 | for _, o := range open { |
| 515 | if meta.ULID == o.Meta().ULID { |
| 516 | b = o |
| 517 | break |
| 518 | } |
| 519 | } |
| 520 | |
| 521 | if b == nil { |
| 522 | var err error |
| 523 | b, err = OpenBlock(c.logger, d, c.chunkPool, c.postingsDecoderFactory) |
| 524 | if err != nil { |
| 525 | return nil, err |
| 526 | } |
| 527 | defer b.Close() |
| 528 | } |
| 529 | |
| 530 | metas = append(metas, meta) |
| 531 | blocks = append(blocks, b) |
| 532 | bs = append(bs, b) |
| 533 | uids = append(uids, meta.ULID.String()) |
| 534 | } |
| 535 | |
| 536 | uid := ulid.MustNew(ulid.Now(), rand.Reader) |
| 537 | |
| 538 | meta := CompactBlockMetas(uid, metas...) |
| 539 | err := c.write(dest, meta, blockPopulator, blocks...) |
| 540 | if err == nil { |
| 541 | if meta.Stats.NumSamples == 0 { |
| 542 | for _, b := range bs { |
| 543 | b.meta.Compaction.Deletable = true |
| 544 | n, err := writeMetaFile(c.logger, b.dir, &b.meta) |
| 545 | if err != nil { |
| 546 | c.logger.Error( |