Repository: kopia/kopia
Pull request #4139 (2024-09-27 08:21:48)

refactor(repository): Reduce memory usage for epoch index compaction
When there are large number of small files in the backup, epoch index compaction, as part of repository maintenance, consumes large size memory which is much larger than other part of repository maintenance.

Below is the result of my test case:
Backup 20M(20971520) files and each file is 4KB in size, so there will be 20M indexes in the backup
When the index compaction is due, it takes more than 15GB memory in the `Compacting an eligible uncompacted epoch` phase of repository maintenance:
```
    PID USER      PR  NI    VIRT    RES    SHR S  %CPU  %MEM     TIME+ COMMAND                              
 881389 root      20   0   19.0g  15.4g 859624 S 100.0  48.9  25:18.93 kopia 
```

This PR refactors the index builder, and could reduce the memory usage significantly. In my testing env, the memory reduces to 7.1GB for the same dataset:
```
   PID USER      PR  NI    VIRT    RES    SHR S  %CPU  %MEM     TIME+ COMMAND                              
 882647 root      20   0   11.1g   7.1g 860304 S 100.0  22.7  26:28.02 kopia 
```

Below are the primary reasons of large memory usage for the existing code:
1. There are 3 copies of the full indexes. The compaction has 3 stages --- index collection, shard, sort and build. The entire indexes are loaded and kept in memory in the 1st stage, the 2nd stage makes a full copy, the 3rd stage makes a third copy for each shard (2^24 indexes)
2. 1st/2nd stage uses `map[index.ID]Info` and 3rd stage uses `[][]Info` as the data struct. The Info items cannot be shared in the 3 structures even though they are identical. So the memory footprint for each index will be `(sizeof(Info) + sizeof(index.ID) + map/slice element overhead) * numOfCopies` while sizeof(Info) + sizeof(index.ID) + map/slice element overhead > 200 bytes
3. The existing index builder (in index_builder.go) is used by many scenarios. While for epoch index compaction specifically, the usage is simple --- it doesn't need to keep the full indexes all the time for further insertion or search, e.g., after shard, we don't need to keep the `map[index.ID]Info` generated in collection stage
4. There are redundancies for packBlobID(of type `blob.ID`). Multiple contents may be in one pack blob, so multiple indexes may share the same PackBlobID. However, there is one PackBlobID in each Info, while, each PackBlobID is more than 50 bytes
5. map[index.ID]Info has a redundancy of `index.ID`, between the key and value, since Info itself also containers the same `index.ID`. While `index.ID` takes 34 bytes each.

Below are the improvements in this PR:
1. Introduce a new builder for index compaction only
1. Use a new builder item struct (`InfoCompact`) which has `*blob.ID` instead of `blob.ID` for packBlobID
2. Use a map implementation that supports custom key compare function(built-in map doesn't), and use pointers to indexes as the map element (`map<*InfoCompact>`). In the PR, we use a rbtree implementation and don't need a key at all. In this way, there is always one piece of memory footprint for each index, no matter how they are copied/moved. The memory footprint for each index will be `sizeof(InfoCompact) + (8  + rbtree/slice element overhead) * numOfCopies`
3. Put the packBlobIDs pointers into a map of the same implementation (`map<*blob.ID>`), all `InfoCompact` reference packBlobIDs in the map, so as to remove the redundancies for packBlobID
4. Since the rbtree is auto sorted at insertion time, so no need to make a separate copy for sort
6. For shard stage, use `[][]*InfoCompact` as the output, index items are removed from `map<*InfoCompact>` and appended to  `[][]*InfoCompact` in the ascendant order. As a result, after shard stage, `map<*InfoCompact>` will be cleared to save memory 

This PR doesn't change any logics of the index compaction, e.g., how indexes are compared, sharded. It only changes the way to implement the logics. So the changes are safe.

Lyndon-Li: Here are the inuse memory during the running:

Before
![image](https://github.com/user-attachments/assets/3b86a479-fb60-49c2-b474-1da79bd0fbb5)

After
![image](https://github.com/user-attachments/assets/d552212f-1d77-419b-8d7c-b560324a7b78)

And the alloc memory:

Before
![image](https://github.com/user-attachments/assets/9aae2ac5-5539-4751-934b-5341a6f5af1e)

After
![image](https://github.com/user-attachments/assets/4944d35f-e3e0-485f-8504-871154218033)


From the both comparison, we can see the memory footprint is significantly reduced.
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/kopia/kopia/pull/4139?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) Report
Attention: Patch coverage is `83.07692%` with `33 lines` in your changes missing coverage. Please review.
> Project coverage is 77.24%. Comparing base [(`cb455c6`)](https://app.codecov.io/gh/kopia/kopia/commit/cb455c699a04e54cb47e25a676d7d65063ce883e?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) to head [(`0d869b8`)](https://app.codecov.io/gh/kopia/kopia/commit/0d869b897725494c71fcbebee765d4ffa230ccd4?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia).
> Report is 273 commits behind head on master.

| [Files with missing lines](https://app.codecov.io/gh/kopia/kopia/pull/4139?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) | Patch % | Lines |
|---|---|---|
| [repo/content/index/one\_use\_index\_builder.go](https://app.codecov.io/gh/kopia/kopia/pull/4139?src=pr&el=tree&filepath=repo%2Fcontent%2Findex%2Fone_use_index_builder.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia#diff-cmVwby9jb250ZW50L2luZGV4L29uZV91c2VfaW5kZXhfYnVpbGRlci5nbw==) | 69.41% | [22 Missing and 4 partials :warning: ](https://app.codecov.io/gh/kopia/kopia/pull/4139?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) |
| [repo/content/index/index\_v1.go](https://app.codecov.io/gh/kopia/kopia/pull/4139?src=pr&el=tree&filepath=repo%2Fcontent%2Findex%2Findex_v1.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia#diff-cmVwby9jb250ZW50L2luZGV4L2luZGV4X3YxLmdv) | 85.71% | [1 Missing and 2 partials :warning: ](https://app.codecov.io/gh/kopia/kopia/pull/4139?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) |
| [repo/content/index/info.go](https://app.codecov.io/gh/kopia/kopia/pull/4139?src=pr&el=tree&filepath=repo%2Fcontent%2Findex%2Finfo.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia#diff-cmVwby9jb250ZW50L2luZGV4L2luZm8uZ28=) | 95.23% | [2 Missing :warning: ](https://app.codecov.io/gh/kopia/kopia/pull/4139?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) |
| [internal/indextest/indextest.go](https://app.codecov.io/gh/kopia/kopia/pull/4139?src=pr&el=tree&filepath=internal%2Findextest%2Findextest.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia#diff-aW50ZXJuYWwvaW5kZXh0ZXN0L2luZGV4dGVzdC5nbw==) | 0.00% | [0 Missing and 1 partial :warning: ](https://app.codecov.io/gh/kopia/kopia/pull/4139?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) |
| [repo/content/index/index\_v2.go](https://app.codecov.io/gh/kopia/kopia/pull/4139?src=pr&el=tree&filepath=repo%2Fcontent%2Findex%2Findex_v2.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia#diff-cmVwby9jb250ZW50L2luZGV4L2luZGV4X3YyLmdv) | 96.77% | [1 Missing :warning: ](https://app.codecov.io/gh/kopia/kopia/pull/4139?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master    #4139      +/-   ##
==========================================
+ Coverage   75.86%   77.24%   +1.37%     
==========================================
  Files         470      485      +15     
  Lines       37301    28996    -8305     
==========================================
- Hits        28299    22398    -5901     
+ Misses       7071     4688    -2383     
+ Partials     1931     1910      -21     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/kopia/kopia/pull/4139?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia).

Lyndon-Li: I am on holiday till 8th, will continue to work on this PR when I am back.
jkowalski: > I am on holiday till 8th, will continue to work on this PR when I am back.

Awesome, I think we can approve it as soon as CI linter issues are fixed.
Lyndon-Li: @jkowalski 
For this test failure https://github.com/kopia/kopia/actions/runs/11101846488/job/30884039243?pr=4139 on ARM64, I couldn't reproduce it in my x86 env.

Do you have any clue on why it happened?

Patch:
diff --git a/go.mod b/go.mod
index 0abb4bb1..0c43a36d 100644
--- a/go.mod
+++ b/go.mod
@@ -35,6 +35,7 @@ require (
 	github.com/mocktools/go-smtp-mock/v2 v2.3.1
 	github.com/mxk/go-vss v1.2.0
 	github.com/natefinch/atomic v1.0.1
+	github.com/petar/GoLLRB v0.0.0-20210522233825-ae3b015fd3e9
 	github.com/pierrec/lz4 v2.6.1+incompatible
 	github.com/pkg/errors v0.9.1
 	github.com/pkg/profile v1.7.0
diff --git a/go.sum b/go.sum
index c5a5f951..8b06444b 100644
--- a/go.sum
+++ b/go.sum
@@ -238,6 +238,8 @@ github.com/natefinch/atomic v1.0.1 h1:ZPYKxkqQOx3KZ+RsbnP/YsgvxWQPGxjC0oBt2AhwV0
 github.com/natefinch/atomic v1.0.1/go.mod h1:N/D/ELrljoqDyT3rZrsUmtsuzvHkeB/wWjHV22AZRbM=
 github.com/orisano/pixelmatch v0.0.0-20220722002657-fb0b55479cde h1:x0TT0RDC7UhAVbbWWBzr41ElhJx5tXPWkIHA2HWPRuw=
 github.com/orisano/pixelmatch v0.0.0-20220722002657-fb0b55479cde/go.mod h1:nZgzbfBr3hhjoZnS66nKrHmduYNpc34ny7RK4z5/HM0=
+github.com/petar/GoLLRB v0.0.0-20210522233825-ae3b015fd3e9 h1:1/WtZae0yGtPq+TI6+Tv1WTxkukpXeMlviSxvL7SRgk=
+github.com/petar/GoLLRB v0.0.0-20210522233825-ae3b015fd3e9/go.mod h1:x3N5drFsm2uilKKuuYo6LdyD8vZAW55sH/9w+pbo1sw=
 github.com/pierrec/lz4 v2.6.1+incompatible h1:9UY3+iC23yxF0UfGaYrGplQ+79Rg+h/q9FV9ix19jjM=
 github.com/pierrec/lz4 v2.6.1+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
 github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c h1:+mdjkGKdHQG3305AYmdv1U2eRNDiU2ErMBj1gwrq8eQ=
diff --git a/repo/content/index/index_builder.go b/repo/content/index/index_builder.go
index 05f54482..3647b38a 100644
--- a/repo/content/index/index_builder.go
+++ b/repo/content/index/index_builder.go
@@ -18,6 +18,11 @@
 // Builder prepares and writes content index.
 type Builder map[ID]Info
 
+// BuilderCreator is an interface for caller to add indexes to builders.
+type BuilderCreator interface {
+	Add(info Info)
+}
+
 // Clone returns a deep Clone of the Builder.
 func (b Builder) Clone() Builder {
 	if b == nil {
@@ -38,7 +43,7 @@ func (b Builder) Add(i Info) {
 	cid := i.ContentID
 
 	old, found := b[cid]
-	if !found || contentInfoGreaterThanStruct(i, old) {
+	if !found || contentInfoGreaterThanStruct(&i, &old) {
 		b[cid] = i
 	}
 }
@@ -63,8 +68,8 @@ func init() {
 // sortedContents returns the list of []Info sorted lexicographically using bucket sort
 // sorting is optimized based on the format of content IDs (optional single-character
 // alphanumeric prefix (0-9a-z), followed by hexadecimal digits (0-9a-f).
-func (b Builder) sortedContents() []Info {
-	var buckets [36 * 16][]Info
+func (b Builder) sortedContents() []*Info {
+	var buckets [36 * 16][]*Info
 
 	// phase 1 - bucketize into 576 (36 *16) separate lists
 	// by first [0-9a-z] and second character [0-9a-f].
@@ -75,7 +80,7 @@ func (b Builder) sortedContents() []Info {
 		// first: 0..35, second: 0..15
 		buck := first<<4 + second //nolint:mnd
 
-		buckets[buck] = append(buckets[buck], v)
+		buckets[buck] = append(buckets[buck], &v)
 	}
 
 	// phase 2 - sort each non-empty bucket in parallel using goroutines
@@ -104,7 +109,7 @@ func (b Builder) sortedContents() []Info {
 	wg.Wait()
 
 	// Phase 3 - merge results from all buckets.
-	result := make([]Info, 0, len(b))
+	result := make([]*Info, 0, len(b))
 
 	for i := range len(buckets) {
 		result = append(result, buckets[i]...)
@@ -134,12 +139,16 @@ func (b Builder) Build(output io.Writer, version int) error {
 
 // BuildStable writes the pack index to the provided output.
 func (b Builder) BuildStable(output io.Writer, version int) error {
+	return buildSortedContents(b.sortedContents(), output, version)
+}
+
+func buildSortedContents(items []*Info, output io.Writer, version int) error {
 	switch version {
 	case Version1:
-		return b.buildV1(output)
+		return buildV1(items, output)
 
 	case Version2:
-		return b.buildV2(output)
+		return buildV2(items, output)
 
 	default:
 		return errors.Errorf("unsupported index version: %v", version)
diff --git a/repo/content/index/index_v1.go b/repo/content/index/index_v1.go
index cae8bfd1..002657c4 100644
--- a/repo/content/index/index_v1.go
+++ b/repo/content/index/index_v1.go
@@ -266,8 +266,7 @@ type indexBuilderV1 struct {
 }
 
 // buildV1 writes the pack index to the provided output.
-func (b Builder) buildV1(output io.Writer) error {
-	allContents := b.sortedContents()
+func buildV1(allContents []*Info, output io.Writer) error {
 	b1 := &indexBuilderV1{
 		packBlobIDOffsets: map[blob.ID]uint32{},
 		keyLength:         -1,
@@ -307,7 +306,7 @@ func (b Builder) buildV1(output io.Writer) error {
 	return errors.Wrap(w.Flush(), "error flushing index")
 }
 
-func (b *indexBuilderV1) prepareExtraData(allContents []Info) []byte {
+func (b *indexBuilderV1) prepareExtraData(allContents []*Info) []byte {
 	var extraData []byte
 
 	var hashBuf [maxContentIDSize]byte
@@ -330,7 +329,7 @@ func (b *indexBuilderV1) prepareExtraData(allContents []Info) []byte {
 	return extraData
 }
 
-func (b *indexBuilderV1) writeEntry(w io.Writer, it Info, entry []byte) error {
+func (b *indexBuilderV1) writeEntry(w io.Writer, it *Info, entry []byte) error {
 	var hashBuf [maxContentIDSize]byte
 
 	k := contentIDToBytes(hashBuf[:0], it.ContentID)
@@ -362,7 +361,7 @@ func (b *indexBuilderV1) writeEntry(w io.Writer, it Info, entry []byte) error {
 	return nil
 }
 
-func (b *indexBuilderV1) formatEntry(entry []byte, it Info) error {
+func (b *indexBuilderV1) formatEntry(entry []byte, it *Info) error {
 	entryTimestampAndFlags := entry[0:8]
 	entryPackFileOffset := entry[8:12]
 	entryPackedOffset := entry[12:16]
diff --git a/repo/content/index/index_v2.go b/repo/content/index/index_v2.go
index 10e34ece..0c8adfb6 100644
--- a/repo/content/index/index_v2.go
+++ b/repo/content/index/index_v2.go
@@ -373,7 +373,7 @@ type indexBuilderV2 struct {
 	baseTimestamp          int64
 }
 
-func indexV2FormatInfoFromInfo(v Info) indexV2FormatInfo {
+func indexV2FormatInfoFromInfo(v *Info) indexV2FormatInfo {
 	return indexV2FormatInfo{
 		formatVersion:       v.FormatVersion,
 		compressionHeaderID: v.CompressionHeaderID,
@@ -382,7 +382,7 @@ func indexV2FormatInfoFromInfo(v Info) indexV2FormatInfo {
 }
 
 // buildUniqueFormatToIndexMap builds a map of unique indexV2FormatInfo to their numeric identifiers.
-func buildUniqueFormatToIndexMap(sortedInfos []Info) map[indexV2FormatInfo]byte {
+func buildUniqueFormatToIndexMap(sortedInfos []*Info) map[indexV2FormatInfo]byte {
 	result := map[indexV2FormatInfo]byte{}
 
 	for _, v := range sortedInfos {
@@ -396,7 +396,7 @@ func buildUniqueFormatToIndexMap(sortedInfos []Info) map[indexV2FormatInfo]byte
 }
 
 // buildPackIDToIndexMap builds a map of unique blob IDs to their numeric identifiers.
-func buildPackIDToIndexMap(sortedInfos []Info) map[blob.ID]int {
+func buildPackIDToIndexMap(sortedInfos []*Info) map[blob.ID]int {
 	result := map[blob.ID]int{}
 
 	for _, v := range sortedInfos {
@@ -410,7 +410,7 @@ func buildPackIDToIndexMap(sortedInfos []Info) map[blob.ID]int {
 }
 
 // maxContentLengths computes max content lengths in the builder.
-func maxContentLengths(sortedInfos []Info) (maxPackedLength, maxOriginalLength, maxPackOffset uint32) {
+func maxContentLengths(sortedInfos []*Info) (maxPackedLength, maxOriginalLength, maxPackOffset uint32) {
 	for _, v := range sortedInfos {
 		if l := v.PackedLength; l > maxPackedLength {
 			maxPackedLength = l
@@ -428,7 +428,7 @@ func maxContentLengths(sortedInfos []Info) (maxPackedLength, maxOriginalLength,
 	return
 }
 
-func newIndexBuilderV2(sortedInfos []Info) (*indexBuilderV2, error) {
+func newIndexBuilderV2(sortedInfos []*Info) (*indexBuilderV2, error) {
 	entrySize := v2EntryOffsetFormatID
 
 	// compute a map of unique formats to their indexes.
@@ -487,9 +487,7 @@ func newIndexBuilderV2(sortedInfos []Info) (*indexBuilderV2, error) {
 }
 
 // buildV2 writes the pack index to the provided output.
-func (b Builder) buildV2(output io.Writer) error {
-	sortedInfos := b.sortedContents()
-
+func buildV2(sortedInfos []*Info, output io.Writer) error {
 	b2, err := newIndexBuilderV2(sortedInfos)
 	if err != nil {
 		return err
@@ -501,7 +499,7 @@ func (b Builder) buildV2(output io.Writer) error {
 	extraData := b2.prepareExtraData(sortedInfos)
 
 	if b2.keyLength <= 1 {
-		return errors.Errorf("invalid key length: %v for %v", b2.keyLength, len(b))
+		return errors.Errorf("invalid key length: %v for %v", b2.keyLength, len(sortedInfos))
 	}
 
 	// write header
@@ -558,7 +556,7 @@ func (b Builder) buildV2(output io.Writer) error {
 	return errors.Wrap(w.Flush(), "error flushing index")
 }
 
-func (b *indexBuilderV2) prepareExtraData(sortedInfos []Info) []byte {
+func (b *indexBuilderV2) prepareExtraData(sortedInfos []*Info) []byte {
 	var extraData []byte
 
 	for _, it := range sortedInfos {
@@ -581,7 +579,7 @@ func (b *indexBuilderV2) prepareExtraData(sortedInfos []Info) []byte {
 	return extraData
 }
 
-func (b *indexBuilderV2) writeIndexEntry(w io.Writer, it Info) error {
+func (b *indexBuilderV2) writeIndexEntry(w io.Writer, it *Info) error {
 	var hashBuf [maxContentIDSize]byte
 
 	k := contentIDToBytes(hashBuf[:0], it.ContentID)
@@ -624,7 +622,7 @@ func (b *indexBuilderV2) writeFormatInfoEntry(w io.Writer, f indexV2FormatInfo)
 	return errors.Wrap(err, "error writing format info entry")
 }
 
-func (b *indexBuilderV2) writeIndexValueEntry(w io.Writer, it Info) error {
+func (b *indexBuilderV2) writeIndexValueEntry(w io.Writer, it *Info) error {
 	var buf [v2EntryMaxLength]byte
 
 	//    0-3: timestamp bits 0..31 (relative to base time)
diff --git a/repo/content/index/merged.go b/repo/content/index/merged.go
index 1bd939de..603c90e4 100644
--- a/repo/content/index/merged.go
+++ b/repo/content/index/merged.go
@@ -33,7 +33,7 @@ func (m Merged) Close() error {
 	return errors.Wrap(err, "closing index shards")
 }
 
-func contentInfoGreaterThanStruct(a, b Info) bool {
+func contentInfoGreaterThanStruct(a, b *Info) bool {
 	if l, r := a.TimestampSeconds, b.TimestampSeconds; l != r {
 		// different timestamps, higher one wins
 		return l > r
@@ -66,7 +66,7 @@ func (m Merged) GetInfo(id ID, result *Info) (bool, error) {
 			continue
 		}
 
-		if !found || contentInfoGreaterThanStruct(tmp, *result) {
+		if !found || contentInfoGreaterThanStruct(&tmp, result) {
 			*result = tmp
 			found = true
 		}
@@ -88,7 +88,7 @@ func (h nextInfoHeap) Less(i, j int) bool {
 		return a.less(b)
 	}
 
-	return !contentInfoGreaterThanStruct(h[i].it, h[j].it)
+	return !contentInfoGreaterThanStruct(&h[i].it, &h[j].it)
 }
 
 func (h nextInfoHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] }
@@ -167,7 +167,7 @@ func (m Merged) Iterate(r IDRange, cb func(i Info) error) error {
 
 			pendingItem = minNextInfo.it
 			havePendingItem = true
-		} else if contentInfoGreaterThanStruct(minNextInfo.it, pendingItem) {
+		} else if contentInfoGreaterThanStruct(&minNextInfo.it, &pendingItem) {
 			pendingItem = minNextInfo.it
 		}
 
diff --git a/repo/content/index/one_use_index_builder.go b/repo/content/index/one_use_index_builder.go
new file mode 100644
index 00000000..52c3c8a0
--- /dev/null
+++ b/repo/content/index/one_use_index_builder.go
@@ -0,0 +1,157 @@
+package index
+
+import (
+	"crypto/rand"
+	"hash/fnv"
+	"io"
+
+	"github.com/pkg/errors"
+
+	"github.com/petar/GoLLRB/llrb"
+
+	"github.com/kopia/kopia/internal/gather"
+)
+
+// Less compares with another *Info by their ContentID and return true if the current one is smaller.
+func (i *Info) Less(other llrb.Item) bool {
+	return i.ContentID.less(other.(*Info).ContentID) //nolint:forcetypeassert
+}
+
+// OneUseBuilder prepares and writes content index for epoch index compaction.
+type OneUseBuilder struct {
+	indexStore *llrb.LLRB
+}
+
+// NewOneUseBuilder create a new instance of OneUseBuilder.
+func NewOneUseBuilder() *OneUseBuilder {
+	return &OneUseBuilder{
+		indexStore: llrb.New(),
+	}
+}
+
+// Add adds a new entry to the builder or conditionally replaces it if the timestamp is greater.
+func (b *OneUseBuilder) Add(i Info) {
+	cid := i.ContentID
+
+	found := b.indexStore.Get(&Info{ContentID: cid})
+	if found == nil || contentInfoGreaterThanStruct(&i, found.(*Info)) { //nolint:forcetypeassert
+		_ = b.indexStore.ReplaceOrInsert(&Info{
+			PackBlobID:          i.PackBlobID,
+			ContentID:           cid,
+			TimestampSeconds:    i.TimestampSeconds,
+			OriginalLength:      i.OriginalLength,
+			PackedLength:        i.PackedLength,
+			PackOffset:          i.PackOffset,
+			CompressionHeaderID: i.CompressionHeaderID,
+			Deleted:             i.Deleted,
+			FormatVersion:       i.FormatVersion,
+			EncryptionKeyID:     i.EncryptionKeyID,
+		})
+	}
+}
+
+// Length returns the number of indexes in the current builder.
+func (b *OneUseBuilder) Length() int {
+	return b.indexStore.Len()
+}
+
+func (b *OneUseBuilder) sortedContents() []*Info {
+	result := []*Info{}
+
+	for b.indexStore.Len() > 0 {
+		item := b.indexStore.DeleteMin()
+		result = append(result, item.(*Info)) //nolint:forcetypeassert
+	}
+
+	return result
+}
+
+func (b *OneUseBuilder) shard(maxShardSize int) [][]*Info {
+	numShards := (b.Length() + maxShardSize - 1) / maxShardSize
+	if numShards <= 1 {
+		if b.Length() == 0 {
+			return [][]*Info{}
+		}
+
+		return [][]*Info{b.sortedContents()}
+	}
+
+	result := make([][]*Info, numShards)
+
+	for b.indexStore.Len() > 0 {
+		item := b.indexStore.DeleteMin()
+
+		h := fnv.New32a()
+		io.WriteString(h, item.(*Info).ContentID.String()) //nolint:errcheck,forcetypeassert
+
+		shard := h.Sum32() % uint32(numShards) //nolint:gosec
+
+		result[shard] = append(result[shard], item.(*Info)) //nolint:forcetypeassert
+	}
+
+	var nonEmpty [][]*Info
+
+	for _, r := range result {
+		if len(r) > 0 {
+			nonEmpty = append(nonEmpty, r)
+		}
+	}
+
+	return nonEmpty
+}
+
+// BuildStable writes the pack index to the provided output.
+func (b *OneUseBuilder) BuildStable(output io.Writer, version int) error {
+	return buildSortedContents(b.sortedContents(), output, version)
+}
+
+// BuildShards builds the set of index shards ensuring no more than the provided number of contents are in each index.
+// Returns shard bytes and function to clean up after the shards have been written.
+func (b *OneUseBuilder) BuildShards(indexVersion int, stable bool, shardSize int) ([]gather.Bytes, func(), error) {
+	if shardSize == 0 {
+		return nil, nil, errors.Errorf("invalid shard size")
+	}
+
+	var (
+		shardedBuilders = b.shard(shardSize)
+		dataShardsBuf   []*gather.WriteBuffer
+		dataShards      []gather.Bytes
+		randomSuffix    [32]byte
+	)
+
+	closeShards := func() {
+		for _, ds := range dataShardsBuf {
+			ds.Close()
+		}
+	}
+
+	for _, s := range shardedBuilders {
+		buf := gather.NewWriteBuffer()
+
+		dataShardsBuf = append(dataShardsBuf, buf)
+
+		if err := buildSortedContents(s, buf, indexVersion); err != nil {
+			closeShards()
+
+			return nil, nil, errors.Wrap(err, "error building index shard")
+		}
+
+		if !stable {
+			if _, err := rand.Read(randomSuffix[:]); err != nil {
+				closeShards()
+
+				return nil, nil, errors.Wrap(err, "error getting random bytes for suffix")
+			}
+
+			if _, err := buf.Write(randomSuffix[:]); err != nil {
+				closeShards()
+
+				return nil, nil, errors.Wrap(err, "error writing extra random suffix to ensure indexes are always globally unique")
+			}
+		}
+
+		dataShards = append(dataShards, buf.Bytes())
+	}
+
+	return dataShards, closeShards, nil
+}
diff --git a/repo/content/indexblob/index_blob_manager_v0.go b/repo/content/indexblob/index_blob_manager_v0.go
index 887d1b72..cb46024f 100644
--- a/repo/content/indexblob/index_blob_manager_v0.go
+++ b/repo/content/indexblob/index_blob_manager_v0.go
@@ -550,7 +550,7 @@ func (m *ManagerV0) dropContentsFromBuilder(bld index.Builder, opt CompactOption
 	}
 }
 
-func addIndexBlobsToBuilder(ctx context.Context, enc *EncryptionManager, bld index.Builder, indexBlobID blob.ID) error {
+func addIndexBlobsToBuilder(ctx context.Context, enc *EncryptionManager, bld index.BuilderCreator, indexBlobID blob.ID) error {
 	var data gather.WriteBuffer
 	defer data.Close()
 
diff --git a/repo/content/indexblob/index_blob_manager_v1.go b/repo/content/indexblob/index_blob_manager_v1.go
index 04df58d0..edf35580 100644
--- a/repo/content/indexblob/index_blob_manager_v1.go
+++ b/repo/content/indexblob/index_blob_manager_v1.go
@@ -68,7 +68,7 @@ func (m *ManagerV1) Compact(ctx context.Context, opt CompactOptions) error {
 
 // CompactEpoch compacts the provided index blobs and writes a new set of blobs.
 func (m *ManagerV1) CompactEpoch(ctx context.Context, blobIDs []blob.ID, outputPrefix blob.ID) error {
-	tmpbld := make(index.Builder)
+	tmpbld := index.NewOneUseBuilder()
 
 	for _, indexBlob := range blobIDs {
 		if err := addIndexBlobsToBuilder(ctx, m.enc, tmpbld, indexBlob); err != nil {
diff --git a/repo/manifest/committed_manifest_manager.go b/repo/manifest/committed_manifest_manager.go
index d955ce72..109fb6ea 100644
--- a/repo/manifest/committed_manifest_manager.go
+++ b/repo/manifest/committed_manifest_manager.go
@@ -32,7 +32,7 @@ type committedManifestManager struct {
 	// +checklocks:cmmu
 	committedEntries map[ID]*manifestEntry
 	// +checklocks:cmmu
-	committedContentIDs map[content.ID]bool
+	committedContentIDs map[content.ID]struct{}
 
 	// autoCompactionThreshold controls the threshold after which the manager auto-compacts
 	// manifest contents
@@ -79,7 +79,7 @@ func (m *committedManifestManager) findCommittedEntries(ctx context.Context, lab
 	return findEntriesMatchingLabels(m.committedEntries, labels), nil
 }
 
-func (m *committedManifestManager) commitEntries(ctx context.Context, entries map[ID]*manifestEntry) (map[content.ID]bool, error) {
+func (m *committedManifestManager) commitEntries(ctx context.Context, entries map[ID]*manifestEntry) (map[content.ID]struct{}, error) {
 	if len(entries) == 0 {
 		return nil, nil
 	}
@@ -98,7 +98,7 @@ func (m *committedManifestManager) commitEntries(ctx context.Context, entries ma
 // the lock via commitEntries()) and to compact existing committed entries during compaction
 // where the lock is already being held.
 // +checklocks:m.cmmu
-func (m *committedManifestManager) writeEntriesLocked(ctx context.Context, entries map[ID]*manifestEntry) (map[content.ID]bool, error) {
+func (m *committedManifestManager) writeEntriesLocked(ctx context.Context, entries map[ID]*manifestEntry) (map[content.ID]struct{}, error) {
 	if len(entries) == 0 {
 		return nil, nil
 	}
@@ -124,12 +124,11 @@ func (m *committedManifestManager) writeEntriesLocked(ctx context.Context, entri
 
 	for _, e := range entries {
 		m.committedEntries[e.ID] = e
-		delete(entries, e.ID)
 	}
 
-	m.committedContentIDs[contentID] = true
+	m.committedContentIDs[contentID] = struct{}{}
 
-	return map[content.ID]bool{contentID: true}, nil
+	return map[content.ID]struct{}{contentID: {}}, nil
 }
 
 // +checklocks:m.cmmu
@@ -192,10 +191,10 @@ func (m *committedManifestManager) loadCommittedContentsLocked(ctx context.Conte
 // +checklocks:m.cmmu
 func (m *committedManifestManager) loadManifestContentsLocked(manifests map[content.ID]manifest) {
 	m.committedEntries = map[ID]*manifestEntry{}
-	m.committedContentIDs = map[content.ID]bool{}
+	m.committedContentIDs = map[content.ID]struct{}{}
 
 	for contentID := range manifests {
-		m.committedContentIDs[contentID] = true
+		m.committedContentIDs[contentID] = struct{}{}
 	}
 
 	for _, man := range manifests {
@@ -257,19 +256,14 @@ func (m *committedManifestManager) compactLocked(ctx context.Context) error {
 	m.b.DisableIndexFlush(ctx)
 	defer m.b.EnableIndexFlush(ctx)
 
-	tmp := map[ID]*manifestEntry{}
-	for k, v := range m.committedEntries {
-		tmp[k] = v
-	}
-
-	written, err := m.writeEntriesLocked(ctx, tmp)
+	written, err := m.writeEntriesLocked(ctx, m.committedEntries)
 	if err != nil {
 		return err
 	}
 
 	// add the newly-created content to the list, could be duplicate
 	for b := range m.committedContentIDs {
-		if written[b] {
+		if _, ok := written[b]; ok {
 			// do not delete content that was just written.
 			continue
 		}
@@ -374,7 +368,7 @@ func newCommittedManager(b contentManager, autoCompactionThreshold int) *committ
 		b:                       b,
 		debugID:                 debugID,
 		committedEntries:        map[ID]*manifestEntry{},
-		committedContentIDs:     map[content.ID]bool{},
+		committedContentIDs:     map[content.ID]struct{}{},
 		autoCompactionThreshold: autoCompactionThreshold,
 	}
 }
diff --git a/repo/manifest/manifest_manager.go b/repo/manifest/manifest_manager.go
index cb53935f..a90c1dbb 100644
--- a/repo/manifest/manifest_manager.go
+++ b/repo/manifest/manifest_manager.go
@@ -215,6 +215,9 @@ func (m *Manager) Flush(ctx context.Context) error {
 	defer m.mu.Unlock()
 
 	_, err := m.committed.commitEntries(ctx, m.pendingEntries)
+	if err == nil {
+		m.pendingEntries = map[ID]*manifestEntry{}
+	}
 
 	return err
 }


Test patch:
diff --git a/repo/content/index/packindex_test.go b/repo/content/index/packindex_test.go
index b50b59d2..bd08894f 100644
--- a/repo/content/index/packindex_test.go
+++ b/repo/content/index/packindex_test.go
@@ -144,15 +144,17 @@ func testPackIndex(t *testing.T, version int) {
 	b1 := make(Builder)
 	b2 := make(Builder)
 	b3 := make(Builder)
+	b4 := NewOneUseBuilder()
 
 	for _, info := range infos {
 		infoMap[info.ContentID] = info
 		b1.Add(info)
 		b2.Add(info)
 		b3.Add(info)
+		b4.Add(info)
 	}
 
-	var buf1, buf2, buf3 bytes.Buffer
+	var buf1, buf2, buf3, buf4 bytes.Buffer
 
 	if err := b1.Build(&buf1, version); err != nil {
 		t.Fatalf("unable to build: %v", err)
@@ -166,9 +168,14 @@ func testPackIndex(t *testing.T, version int) {
 		t.Fatalf("unable to build: %v", err)
 	}
 
+	if err := b4.BuildStable(&buf4, version); err != nil {
+		t.Fatalf("unable to build: %v", err)
+	}
+
 	data1 := buf1.Bytes()
 	data2 := buf2.Bytes()
 	data3 := buf3.Bytes()
+	data4 := buf4.Bytes()
 
 	// each build produces exactly identical prefix except for the trailing random bytes.
 	data1Prefix := data1[0 : len(data1)-randomSuffixSize]
@@ -176,13 +183,22 @@ func testPackIndex(t *testing.T, version int) {
 
 	require.Equal(t, data1Prefix, data2Prefix)
 	require.Equal(t, data2Prefix, data3)
+	require.Equal(t, data2Prefix, data4)
 	require.NotEqual(t, data1, data2)
+	require.Equal(t, data3, data4)
 
 	t.Run("FuzzTest", func(t *testing.T) {
 		fuzzTestIndexOpen(data1)
 	})
 
-	ndx, err := Open(data1, nil, func() int { return fakeEncryptionOverhead })
+	verifyPackedIndexes(t, infos, infoMap, version, data1)
+	verifyPackedIndexes(t, infos, infoMap, version, data4)
+}
+
+func verifyPackedIndexes(t *testing.T, infos []Info, infoMap map[ID]Info, version int, packed []byte) {
+	t.Helper()
+
+	ndx, err := Open(packed, nil, func() int { return fakeEncryptionOverhead })
 	if err != nil {
 		t.Fatalf("can't open index: %v", err)
 	}
@@ -276,7 +292,7 @@ func TestPackIndexPerContentLimits(t *testing.T) {
 		var result bytes.Buffer
 
 		if tc.errMsg == "" {
-			require.NoError(t, b.buildV2(&result))
+			require.NoError(t, buildV2(b.sortedContents(), &result))
 
 			pi, err := Open(result.Bytes(), nil, func() int { return fakeEncryptionOverhead })
 			require.NoError(t, err)
@@ -289,7 +305,7 @@ func TestPackIndexPerContentLimits(t *testing.T) {
 
 			require.Equal(t, got, tc.info)
 		} else {
-			err := b.buildV2(&result)
+			err := buildV2(b.sortedContents(), &result)
 			require.Error(t, err)
 			require.Contains(t, err.Error(), tc.errMsg)
 		}
@@ -366,6 +382,76 @@ func TestSortedContents2(t *testing.T) {
 	}
 }
 
+func TestSortedContents3(t *testing.T) {
+	b := NewOneUseBuilder()
+
+	for i := range 100 {
+		v := deterministicContentID(t, "", i)
+
+		b.Add(Info{
+			ContentID: v,
+		})
+	}
+
+	got := b.sortedContents()
+
+	var last ID
+	for _, info := range got {
+		if info.ContentID.less(last) {
+			t.Fatalf("not sorted %v (was %v)!", info.ContentID, last)
+		}
+
+		last = info.ContentID
+	}
+}
+
+func TestSortedContents4(t *testing.T) {
+	b := NewOneUseBuilder()
+
+	b.Add(Info{
+		ContentID: mustParseID(t, "0123"),
+	})
+	b.Add(Info{
+		ContentID: mustParseID(t, "1023"),
+	})
+	b.Add(Info{
+		ContentID: mustParseID(t, "0f23"),
+	})
+	b.Add(Info{
+		ContentID: mustParseID(t, "f023"),
+	})
+	b.Add(Info{
+		ContentID: mustParseID(t, "g0123"),
+	})
+	b.Add(Info{
+		ContentID: mustParseID(t, "g1023"),
+	})
+	b.Add(Info{
+		ContentID: mustParseID(t, "i0123"),
+	})
+	b.Add(Info{
+		ContentID: mustParseID(t, "i1023"),
+	})
+	b.Add(Info{
+		ContentID: mustParseID(t, "h0123"),
+	})
+	b.Add(Info{
+		ContentID: mustParseID(t, "h1023"),
+	})
+
+	got := b.sortedContents()
+
+	var last ID
+
+	for _, info := range got {
+		if info.ContentID.less(last) {
+			t.Fatalf("not sorted %v (was %v)!", info.ContentID, last)
+		}
+
+		last = info.ContentID
+	}
+}
+
 func TestPackIndexV2TooManyUniqueFormats(t *testing.T) {
 	b := Builder{}
 
@@ -380,7 +466,7 @@ func TestPackIndexV2TooManyUniqueFormats(t *testing.T) {
 		})
 	}
 
-	require.NoError(t, b.buildV2(io.Discard))
+	require.NoError(t, buildV2(b.sortedContents(), io.Discard))
 
 	// add one more to push it over the edge
 	b.Add(Info{
@@ -389,7 +475,7 @@ func TestPackIndexV2TooManyUniqueFormats(t *testing.T) {
 		CompressionHeaderID: compression.HeaderID(5000),
 	})
 
-	err := b.buildV2(io.Discard)
+	err := buildV2(b.sortedContents(), io.Discard)
 	require.Error(t, err)
 	require.Equal(t, "unsupported - too many unique formats 256 (max 255)", err.Error())
 }
@@ -518,6 +604,83 @@ func verifyAllShardedIDs(t *testing.T, sharded []Builder, numTotal, numShards in
 	return lens
 }
 
+func TestShard1(t *testing.T) {
+	// generate 10000 IDs in random order
+	ids := make([]int, 10000)
+	for i := range ids {
+		ids[i] = i
+	}
+
+	rand.Shuffle(len(ids), func(i, j int) {
+		ids[i], ids[j] = ids[j], ids[i]
+	})
+
+	cases := []struct {
+		shardSize int
+		numShards int
+		shardLens []int
+	}{
+		{100000, 1, nil},
+		{100, 100, nil},
+		{500, 20, []int{460, 472, 473, 477, 479, 483, 486, 492, 498, 499, 501, 503, 504, 505, 511, 519, 524, 528, 542, 544}},
+		{1000, 10, []int{945, 964, 988, 988, 993, 1002, 1014, 1017, 1021, 1068}},
+		{2000, 5, []int{1952, 1995, 2005, 2013, 2035}},
+	}
+
+	for _, tc := range cases {
+		b := NewOneUseBuilder()
+
+		// add ID to the builder
+		for _, id := range ids {
+			b.Add(Info{
+				ContentID: deterministicContentID(t, "", id),
+			})
+		}
+
+		length := b.Length()
+		shards := b.shard(tc.shardSize)
+
+		// verify number of shards
+		lens := verifyAllShardedIDsList(t, shards, length, tc.numShards)
+
+		require.Zero(t, b.Length())
+
+		// sharding will always produce stable results, verify sorted shard lengths here
+		if tc.shardLens != nil {
+			require.ElementsMatch(t, tc.shardLens, lens)
+		}
+	}
+}
+
+func verifyAllShardedIDsList(t *testing.T, sharded [][]*Info, numTotal, numShards int) []int {
+	t.Helper()
+
+	require.Len(t, sharded, numShards)
+
+	m := map[ID]bool{}
+	for i := range numTotal {
+		m[deterministicContentID(t, "", i)] = true
+	}
+
+	cnt := 0
+
+	var lens []int
+
+	for _, s := range sharded {
+		cnt += len(s)
+		lens = append(lens, len(s))
+
+		for _, v := range s {
+			delete(m, v.ContentID)
+		}
+	}
+
+	require.Equal(t, numTotal, cnt, "invalid total number of sharded elements")
+	require.Empty(t, m)
+
+	return lens
+}
+
 func withOriginalLength(is Info, originalLength uint32) Info {
 	// clone and override original length
 	is.OriginalLength = originalLength
diff --git a/repo/manifest/manifest_manager_test.go b/repo/manifest/manifest_manager_test.go
index 9c567d88..c60fa24c 100644
--- a/repo/manifest/manifest_manager_test.go
+++ b/repo/manifest/manifest_manager_test.go
@@ -3,6 +3,7 @@
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"reflect"
 	"sort"
 	"strings"
@@ -310,8 +311,8 @@ type contentManagerOpts struct {
 	readOnly bool
 }
 
-func newContentManagerForTesting(ctx context.Context, t *testing.T, data blobtesting.DataMap, opts contentManagerOpts) contentManager {
-	t.Helper()
+func newContentManagerForTesting(ctx context.Context, tb testing.TB, data blobtesting.DataMap, opts contentManagerOpts) contentManager {
+	tb.Helper()
 
 	st := blobtesting.NewMapStorage(data, nil, nil)
 
@@ -328,12 +329,12 @@ func newContentManagerForTesting(ctx context.Context, t *testing.T, data blobtes
 		},
 	}, nil)
 
-	require.NoError(t, err)
+	require.NoError(tb, err)
 
 	bm, err := content.NewManagerForTesting(ctx, st, fop, nil, nil)
-	require.NoError(t, err)
+	require.NoError(tb, err)
 
-	t.Cleanup(func() { bm.CloseShared(ctx) })
+	tb.Cleanup(func() { bm.CloseShared(ctx) })
 
 	return bm
 }
@@ -488,3 +489,50 @@ func TestManifestAutoCompactionWithReadOnly(t *testing.T) {
 	_, err = mgr.Find(ctx, map[string]string{"color": "red"})
 	require.NoError(t, err, "forcing reload of manifest manager")
 }
+
+func BenchmarkLargeCompaction(b *testing.B) {
+	item1 := map[string]int{"foo": 1, "bar": 2}
+	labels1 := map[string]string{"type": "item", "color": "red"}
+
+	table := []int{10000, 100000, 1000000}
+
+	for _, numItems := range table {
+		b.Run(fmt.Sprintf("%dItems", numItems), func(b *testing.B) {
+			for range b.N {
+				b.StopTimer()
+				// Use default context to avoid lots of log output during benchmark.
+				ctx := context.Background()
+				data := blobtesting.DataMap{}
+
+				bm := newContentManagerForTesting(ctx, b, data, contentManagerOpts{})
+
+				mgr, err := NewManager(
+					ctx,
+					bm,
+					ManagerOptions{AutoCompactionThreshold: 2},
+					nil,
+				)
+				require.NoError(b, err, "getting initial manifest manager")
+
+				for range numItems - 1 {
+					_, err = mgr.Put(ctx, labels1, item1)
+					require.NoError(b, err, "adding item to manifest manager")
+				}
+
+				require.NoError(b, mgr.Flush(ctx))
+				require.NoError(b, mgr.b.Flush(ctx))
+
+				_, err = mgr.Put(ctx, labels1, item1)
+				require.NoError(b, err, "adding item to manifest manager")
+
+				require.NoError(b, mgr.Flush(ctx))
+				require.NoError(b, mgr.b.Flush(ctx))
+
+				b.StartTimer()
+
+				err = mgr.Compact(ctx)
+				require.NoError(b, err, "forcing reload of manifest manager")
+			}
+		})
+	}
+}