func BenchmarkSplitBAM(b *testing.B) {
	if !*splitFlag {
		b.Skipf("--split not set")
		return
	}
	require.NoError(b, increaseRlimit())
	p := bamprovider.NewProvider(*bamFlag, bamprovider.ProviderOpts{Index: *baiFlag})
	header, err := p.GetHeader()
	require.NoError(b, err)
	shards := generateShards(b, *bamFlag, *baiFlag, *bytesPerShardFlag)
	sortCh := make(chan *sam.Record, 1<<20)
	sortWg := sync.WaitGroup{}
	shardSeq := int32(0)
	for i := 0; i < *sortParallelismFlag; i++ {
		sortWg.Add(1)
		go func() {
			var sorter *Sorter
			var nRecsInShard int64
			for rec := range sortCh {
				if sorter == nil {
					seq := atomic.AddInt32(&shardSeq, 1)
					shardPath := fmt.Sprintf("%s/test-%04d.sortshard", *sortShardDirFlag, seq)
					log.Printf("Creating %v", shardPath)
					sorter = NewSorter(shardPath, header, SortOptions{
						TmpDir:             *tmpDirFlag,
						NoCompressTmpFiles: !*compressFlag,
						ShardIndex:         uint32(seq),
					})
				}
				sorter.AddRecord(rec)
				nRecsInShard++
				if nRecsInShard >= *recsPerShardFlag {
					require.NoError(b, sorter.Close())
					sorter = nil
					nRecsInShard = 0
				}
			}
			require.NoError(b, sorter.Close())
			sortWg.Done()
		}()
	}
	ch := gbam.NewShardChannel(shards)
	readWg := sync.WaitGroup{}
	totalNumRecs := int64(0)
	for i := 0; i < runtime.NumCPU(); i++ {
		readWg.Add(1)
		go func() {
			n := int64(0)
			for shard := range ch {
				iter := p.NewIterator(shard)
				for iter.Scan() {
					sortCh <- iter.Record()
					n++
					if n == 1<<20 {
						if total := atomic.AddInt64(&totalNumRecs, n); total%(16<<20) == 0 {
							log.Printf("Read %d records", total)
						}
						n = 0
					}
				}
				require.NoError(b, iter.Close())
			}
			readWg.Done()
		}()
	}
	readWg.Wait()
	close(sortCh)
	sortWg.Wait()
	require.NoError(b, p.Close())
}
Example of sorting & merging a large BAM file for benchmarking:

Example of splitting a BAM file ~60 ways and creating a sortshard file for each.

bazel-bin/cmd/bio-bam-sort/sorter/go_default_test --test.run xxxx -test.v -test.bench Split -logtostderr --split -bam /scratch-nvme/bam/CNVS-NORM-110033752-cfDNA-WGBS-Rep1.bam

Example of merging the sortshard files created above.

bazel-bin/cmd/bio-bam-sort/bio-bam-sort -profile-interval-s 60 -heap-profile heap -cpu-profile cpu --bam /scratch-nvme/tmp/foo.bam /scratch-nvme/sortshard/*.sortshard
package sorter
import (
	"bytes"
	"flag"
	"fmt"
	"io"
	"log"
	"math"
	"math/rand"
	"os"
	"path/filepath"
	"runtime"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"syscall"
	"testing"
	"github.com/grailbio/base/errors"
	"github.com/grailbio/base/grail"
	gbam "github.com/grailbio/bio/encoding/bam"
	"github.com/grailbio/bio/encoding/bamprovider"
	"github.com/grailbio/hts/bam"
	"github.com/grailbio/hts/bgzf"
	"github.com/grailbio/hts/sam"
	"github.com/grailbio/testutil"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"v.io/x/lib/gosh"
	"v.io/x/lib/lookpath"
)
func headerString(t *testing.T, header *sam.Header) string {
	s, err := header.MarshalText()
	require.NoError(t, err)
	return "[" + string(s) + "]"
}
func TestCompareKeys(t *testing.T) {
	ref, err := sam.NewReference("chr1", "0", "human", 10000, nil, nil)
	require.NoError(t, err)
	_, err = sam.NewHeader(nil, []*sam.Reference{ref})
	require.NoError(t, err)
	testRecord := func(name string, alignPos int, flags sam.Flags) *sam.Record {
		seq := []byte{'A'}
		qual := []byte{30}
		rec, err := sam.NewRecord(name, ref, ref, alignPos, alignPos+223  , 0  , 1  , nil  , seq, qual, nil)
		require.NoError(t, err)
		rec.Flags = flags
		return rec
	}
	type testCase struct {
		r0 *sam.Record
		r1 *sam.Record
		e  int  
	}
	for _, tc := range []testCase{
		{
			testRecord("r01", 1, sam.Flags(0)),
			testRecord("r01", 1, sam.Flags(0)),
			0},
		{
			testRecord("r01", 1, sam.Flags(0)),
			testRecord("r01", 2, sam.Flags(0)),
			-1},
		{
			testRecord("r01", 1, sam.Reverse),
			testRecord("r01", 2, sam.Flags(0)),
			-1},
		{
			testRecord("r01", 1, sam.Flags(0)),
			testRecord("r02", 1, sam.Reverse),
			-1},
	} {
		var buf0, buf1 bytes.Buffer
		err := bam.Marshal(tc.r0, &buf0)
		assert.NoError(t, err)
		err = bam.Marshal(tc.r1, &buf1)
		assert.NoError(t, err)
		r0Key := sortEntry{coordFromRecord(tc.r0), buf0.Bytes()}
		r1Key := sortEntry{coordFromRecord(tc.r1), buf1.Bytes()}
		assert.Equalf(t, 0, r0Key.compare(r0Key), "R0: %+v", r0Key)
		assert.Equalf(t, 0, r1Key.compare(r1Key), "R0: %+v", r1Key)
		assert.Equalf(t, r0Key.compare(r1Key), tc.e,
			"R0: %+v, R1: %+v, compare %d, expected %d",
			r0Key, r1Key, r0Key.compare(r1Key), tc.e)
	}
}
func readRecords(t *testing.T, path string) (*sam.Header, []*sam.Record) {
	records := []*sam.Record{}
	if strings.HasSuffix(path, ".bam") {
		in, err := os.Open(path)
		require.NoErrorf(t, err, "Open %v", path)
		defer in.Close()
		r, err := bam.NewReader(in, 16)
		require.NoErrorf(t, err, "NewReader %v", path)
		for {
			rec, err := r.Read()
			if err != nil {
				require.True(t, err == io.EOF)
				break
			}
			records = append(records, rec)
		}
		require.NoError(t, r.Close())
		return r.Header(), records
	}
	provider := bamprovider.NewProvider(path)
	header, err := provider.GetHeader()
	require.NoError(t, err)
	iter := provider.NewIterator(gbam.UniversalShard(header))
	for iter.Scan() {
		records = append(records, iter.Record())
	}
	require.NoError(t, iter.Close())
	require.NoError(t, provider.Close())
	return header, records
}
func createSortShards(t *testing.T, header *sam.Header, recs []*sam.Record, tempDir string, n int) []string {
	shards := make([]string, n)
	for i := 0; i < len(shards); i++ {
		shards[i] = fmt.Sprintf("%s/shard%d", tempDir, i)
		log.Printf("Creating sort shard %v", shards[i])
		sorter := NewSorter(shards[i], header, SortOptions{ShardIndex: uint32(i)})
		for j, rec := range recs {
			if j%len(shards) == i {
				sorter.AddRecord(rec)
			}
		}
		require.NoError(t, sorter.Close())
		log.Printf("Finished creating sort shard %v", shards[i])
	}
	return shards
}
func compareFiles(t *testing.T, orgPath, newPath string) {
	orgHeader, orgRecs := readRecords(t, orgPath)
	newHeader, newRecs := readRecords(t, newPath)
	require.Equal(t, headerString(t, orgHeader), headerString(t, newHeader))
	require.Equal(t, len(orgRecs), len(newRecs))
	orgNames := make(map[string]*sam.Record)
	newNames := make(map[string]*sam.Record)
	for i := 0; i < len(orgRecs); i++ {
		r0 := orgRecs[i]
		r1 := newRecs[i]
		orgNames[r0.Name] = r0
		newNames[r1.Name] = r1
		require.Equalf(t, r0.Ref.Name(), r1.Ref.Name(), "%d: Records: org:\n%v, new\n%v", i, r0, r1)
		require.Equalf(t, r0.Pos, r1.Pos, "%d: Records: org:\n%v, new\n%v", i, r0, r1)
	}
	require.Equal(t, len(orgNames), len(newNames))
	for name := range orgNames {
		_, ok := newNames[name]
		assert.Truef(t, ok, "Name %d not found in new", name)
	}
}
var testBAMFiles = []string{
	testutil.GetFilePath("//go/src/grail.com/bio/encoding/bam/testdata/test.bam"),
	testutil.GetFilePath("//go/src/grail.com/bio/encoding/bam/testdata/test-unmapped.bam"),
	testutil.GetFilePath("//go/src/grail.com/bio/encoding/bam/testdata/170614_WGS_LOD_Pre_Library_B3_27961B_05.merged.10000.bam"),
}
func TestSortBAMEndToEnd(t *testing.T) {
	tempDir, cleanup := testutil.TempDir(t, "", "")
	defer testutil.NoCleanupOnError(t, cleanup)
	for _, bamPath := range testBAMFiles {
		newBAMPath := filepath.Join(tempDir, "test.bam")
		log.Printf("Sorting %v to %v", bamPath, newBAMPath)
		header, recs := shuffleRecords(t, bamPath, "")
		shards := createSortShards(t, header, recs, tempDir, 3)
		require.NoError(t, BAMFromSortShards(shards[:], newBAMPath))
		compareFiles(t, bamPath, newBAMPath)
	}
}
func TestSortPAMEndToEnd(t *testing.T) {
	tempDir, cleanup := testutil.TempDir(t, "", "")
	defer testutil.NoCleanupOnError(t, cleanup)
	for _, bamPath := range testBAMFiles {
		newPAMPath := filepath.Join(tempDir, "test.pam")
		log.Printf("Sorting %v to %v", bamPath, newPAMPath)
		header, recs := shuffleRecords(t, bamPath, "")
		shards := createSortShards(t, header, recs, tempDir, 3)
		require.NoError(t, PAMFromSortShards(shards[:], newPAMPath, math.MaxInt64, 8))
		compareFiles(t, bamPath, newPAMPath)
		if !strings.Contains(bamPath, "170614") {
			continue
		}
		newPAMPath = filepath.Join(tempDir, "sharded.pam")
		require.NoError(t, PAMFromSortShards(shards[:], newPAMPath, 1024, 8))
		compareFiles(t, bamPath, newPAMPath)
		require.NoError(t, PAMFromSortShards(shards[:], newPAMPath, 3000, 8))
		compareFiles(t, bamPath, newPAMPath)
		require.NoError(t, PAMFromSortShards(shards[:], newPAMPath, 500, 8))
		compareFiles(t, bamPath, newPAMPath)
	}
}
func TestEmpty(t *testing.T) {
	tempDir, cleanup := testutil.TempDir(t, "", "")
	defer testutil.NoCleanupOnError(t, cleanup)
	shardPath := filepath.Join(tempDir, "shard0")
	bamPath := filepath.Join(tempDir, "test.bam")
	header, err := sam.NewHeader(nil, nil)
	require.NoError(t, err)
	sorter := NewSorter(shardPath, header)
	require.NoError(t, sorter.Close())
	require.NoError(t, BAMFromSortShards([]string{shardPath}, bamPath))
	_, recs := readRecords(t, bamPath)
	require.Truef(t, len(recs) == 0, "Recs: %+v", recs)
}
func sortSAM(t *testing.T, opts SortOptions, shardPath, text string) {
	r, err := sam.NewReader(bytes.NewBuffer([]byte(text)))
	require.NoError(t, err)
	sorter := NewSorter(shardPath, r.Header(), opts)
	for {
		rec, err := r.Read()
		if rec == nil {
			require.Truef(t, err == io.EOF, "err: %v", err)
			break
		}
		require.NoError(t, err)
		sorter.AddRecord(rec)
	}
	require.NoError(t, sorter.Close())
}
func TestUnmapped(t *testing.T) {
	tempDir, cleanup := testutil.TempDir(t, "", "")
	defer testutil.NoCleanupOnError(t, cleanup)
	shard0 := fmt.Sprintf("%s/shard0", tempDir)
	sortSAM(t, SortOptions{}, shard0, `@HD	VN:1.3	SO:coordinate
read10	12	*	0	0	10M	*	0	0	ACGTACGTAC	ABCDEFGHIJ	RG:Z:NA12878
read11	12	*	0	0	10M	*	0	0	ACGTACGTAC	ABCDEFGHIJ	RG:Z:NA12879
`)
	errReporter := errors.Once{}
	pool := newSortShardBlockPool()
	shardReader := newSortShardReader(shard0, pool, &errReporter)
	header, err := mergeHeader([]*sortShardReader{shardReader})
	require.NoError(t, err)
	pamPath := filepath.Join(tempDir, "test.pam")
	generatePAMShard([]*sortShardReader{shardReader},
		pamPath, header, unmappedCoord, infinityCoord, pool, &errReporter)
	require.NoError(t, errReporter.Err())
	_, recs := readRecords(t, pamPath)
	log.Printf("Read recs: %v", recs)
	expected := []string{"read10", "read11"}
	n := 0
	for _, rec := range recs {
		log.Printf("Read rec: %v", rec)
		assert.Equalf(t, expected[n], rec.Name, "rec=%+v", rec)
		n++
	}
	assert.Equal(t, n, len(expected))
}
func TestMergeDifferentRefs(t *testing.T) {
	tempDir, cleanup := testutil.TempDir(t, "", "")
	defer testutil.NoCleanupOnError(t, cleanup)
	shard0 := fmt.Sprintf("%s/shard0", tempDir)
	shard1 := fmt.Sprintf("%s/shard1", tempDir)
	sortSAM(t, SortOptions{ShardIndex: 1}, shard0, `@HD	VN:1.3	SO:coordinate
@SQ	SN:chr1	LN:10000
read1	0	chr1	123	60	10M	=	456	20	AAAAAAAAAA	ABCDEFGHIJ	NM:i:1
`)
	sortSAM(t, SortOptions{ShardIndex: 2}, shard1, `@HD	VN:1.3	SO:coordinate
@SQ	SN:chr1	LN:10000
@SQ	SN:chr2	LN:9999
read2	0	chr1	100	60	10M	=	456	20	CCCCCCCCCC	ABCDEFGHIJ	NM:i:1
read3	0	chr2	123	60	10M	=	456	20	GGGGGGGGGG	ABCDEFGHIJ	NM:i:1
`)
	bamPath := filepath.Join(tempDir, "test.bam")
	require.NoError(t, BAMFromSortShards([]string{shard0, shard1}, bamPath))
	header, recs := readRecords(t, bamPath)
	log.Printf("Read recs: %v", recs)
	require.Contains(t, headerString(t, header), "@SQ	SN:chr1	LN:10000")
	require.Contains(t, headerString(t, header), "@SQ	SN:chr2	LN:9999")
	expected := []string{"read2", "read1", "read3"}
	n := 0
	for _, rec := range recs {
		log.Printf("Read rec: %v", rec)
		assert.Equalf(t, expected[n], rec.Name, "rec=%+v", rec)
		n++
	}
	assert.Equal(t, n, len(expected))
}
func TestMergeSameShardIndex(t *testing.T) {
	tempDir, cleanup := testutil.TempDir(t, "", "")
	defer testutil.NoCleanupOnError(t, cleanup)
	samText := `@HD	VN:1.3	SO:coordinate
@SQ	SN:chr1	LN:10000
read1	0	chr1	123	60	10M	=	456	20	AAAAAAAAAA	ABCDEFGHIJ	NM:i:1
`
	shard0 := fmt.Sprintf("%s/shard0", tempDir)
	shard1 := fmt.Sprintf("%s/shard1", tempDir)
	sortSAM(t, SortOptions{ShardIndex: 1}, shard0, samText)
	sortSAM(t, SortOptions{ShardIndex: 1}, shard1, samText)
	bamPath := filepath.Join(tempDir, "test.bam")
	require.NoError(t, BAMFromSortShards([]string{shard0, shard1}, bamPath))
	header, recs := readRecords(t, bamPath)
	log.Printf("Read recs: %v", recs)
	require.Contains(t, headerString(t, header), "@SQ	SN:chr1	LN:10000")
	expected := []string{"read1", "read1"}
	n := 0
	for _, rec := range recs {
		log.Printf("Read rec: %v", rec)
		assert.Equalf(t, expected[n], rec.Name, "rec=%+v", rec)
		n++
	}
	assert.Equal(t, n, len(expected))
}
func runCmd(t *testing.T, sh *gosh.Shell, arg0 string, args ...string) {
	cmd := sh.Cmd(arg0, args...)
	cmd.Run()
	if cmd.Err != nil {
		t.Fatal(cmd.Err)
	}
}
func testAgainstSambamba(t *testing.T, bamPath string) {
	sh := gosh.NewShell(t)
	defer sh.Cleanup()
	if _, err := lookpath.Look(sh.Vars, "sambamba"); err != nil {
		t.Skipf("sambamba not found on the machine. Skipping the test for %v", bamPath)
		return
	}
	log.Printf("Comparing outputs of sambamba and Sort() for %s", bamPath)
	unsortedFile := sh.MakeTempFile()
	runCmd(t, sh, "sambamba", "sort", "-N", "-o", unsortedFile.Name(), bamPath)
	smbSortedFile := sh.MakeTempFile()
	runCmd(t, sh, "sambamba", "sort", "-o", smbSortedFile.Name(), unsortedFile.Name())
	smbSortedTextFile := sh.MakeTempFile()
	runCmd(t, sh, "sambamba", "view", "-h", "-f", "sam", "-o", smbSortedTextFile.Name(), smbSortedFile.Name())
	t.Logf("Sorted %v into %v using sambamba", unsortedFile.Name(), smbSortedFile.Name())
	bamReader, err := bam.NewReader(unsortedFile, 1)
	require.NoError(t, err)
	shardFile := sh.MakeTempFile()
	sorter := NewSorter(shardFile.Name(), bamReader.Header())
	for {
		rec, err := bamReader.Read()
		if rec == nil {
			break
		}
		require.NoError(t, err)
		sorter.AddRecord(rec)
	}
	require.NoError(t, bamReader.Close())
	require.NoError(t, sorter.Close())
	sortedFile := sh.MakeTempFile()
	require.NoError(t, BAMFromSortShards([]string{shardFile.Name()}, sortedFile.Name()))
	t.Logf("Sorted %v into %v using Sort", unsortedFile.Name(), sortedFile.Name())
	sortedTextFile := sh.MakeTempFile()
	runCmd(t, sh, "sambamba", "view", "-h", "-f", "sam", "-o", sortedTextFile.Name(), sortedFile.Name())
	diff := sh.Cmd("diff", "-I", "^@PG", smbSortedTextFile.Name(), sortedTextFile.Name()).Stdout()
	if diff != "" {
		t.Errorf("Diff: %v", diff)
	}
}
func TestAgainstSamBamba(t *testing.T) {
	for _, path := range testBAMFiles {
		testAgainstSambamba(t, path)
	}
}
func TestSortKey(t *testing.T) {
	ref0, err := sam.NewReference("chr1", "", "", 1000, nil, nil)
	require.NoError(t, err)
	ref1, err := sam.NewReference("chr2", "", "", 1000, nil, nil)
	require.NoError(t, err)
	_, err = sam.NewHeader(nil, []*sam.Reference{ref0, ref1})
	require.NoError(t, err)
	refid, pos, reverse := parseCoord(coordFromRecord(&sam.Record{Ref: ref0, Pos: 0}))
	assert.True(t, refid == 0 && pos == 0 && !reverse, "rec", refid, pos, reverse)
	refid, pos, reverse = parseCoord(coordFromRecord(&sam.Record{Ref: ref1, Pos: 1}))
	assert.True(t, refid == 1 && pos == 1 && !reverse, "rec", refid, pos, reverse)
	refid, pos, reverse = parseCoord(coordFromRecord(&sam.Record{Ref: nil, Pos: -1}))
	assert.True(t, refid == -1 && pos == -1 && !reverse, "rec", refid, pos, reverse)
}
func increaseRlimit() error {
	var l syscall.Rlimit
	err := syscall.Getrlimit(syscall.RLIMIT_NOFILE, &l)
	if err != nil {
		return err
	}
	l.Cur = l.Max
	return syscall.Setrlimit(syscall.RLIMIT_NOFILE, &l)
}
var (
	bamFlag             = flag.String("bam", "/scratch-nvme/bam/small.bam", "File to sort in benchmark")
	baiFlag             = flag.String("bai", "", "BAM index file to use during benchmarking. May be empty")
	sortShardDirFlag    = flag.String("sortsharddir", "/scratch-nvme/sortshard", "Directory to store outputs during benchmarking")
	tmpDirFlag          = flag.String("tmpdir", "/scratch-nvme/tmp", "Tmp dir to use")
	compressFlag        = flag.Bool("compress", true, "Compress tmp files using snappy")
	splitFlag           = flag.Bool("split", false, "If set, split the BAM file into multiple shards")
	bytesPerShardFlag   = flag.Int64("bytes-per-shard", 8<<30, "BAM/PAM read shard size")
	recsPerShardFlag    = flag.Int64("recs-per-shard", 32<<20, "Target sortshard size")
	sortBatchSizeFlag   = flag.Int("sort-batch-size", 1<<20, "Sort batch size during benchmark")
	sortParallelismFlag = flag.Int("parallelism", 16, "Number of sortshards to create in parallel")
)
func generateShards(b *testing.B, bamPath, baiPath string, bytesPerShard int64) []gbam.Shard {
	shards := []gbam.Shard{}
	in, err := os.Open(bamPath)
	require.NoError(b, err)
	defer func() {
		if err := in.Close(); err != nil {
			b.Fatal(err)
		}
	}()
	bamr, err := bam.NewReader(in, 1)
	defer func() {
		if err := bamr.Close(); err != nil {
			b.Fatal(err)
		}
	}()
	header := bamr.Header()
	require.NoError(b, err)
	if baiPath == "" {
		baiPath = bamPath + ".bai"
	}
	indexIn, err := os.Open(baiPath)
	require.NoError(b, err)
	defer indexIn.Close()
	index, err := gbam.ReadIndex(indexIn)
	require.NoError(b, err)
	chunks := []bgzf.Offset{}
	for _, chunksForRef := range index.AllOffsets() {
		chunks = append(chunks, chunksForRef...)
	}
	sort.SliceStable(chunks,
		func(a, b int) bool {
			return chunks[a].File < chunks[b].File
		})
	require.True(b, len(chunks) > 0)
	startChunk := chunks[0]
	for _, chunk := range chunks[1:] {
		if chunk.File-startChunk.File >= bytesPerShard {
			startCoord, err := gbam.GetCoordAtOffset(bamr, startChunk)
			require.NoError(b, err)
			limitCoord, err := gbam.GetCoordAtOffset(bamr, chunk)
			require.NoError(b, err)
			shard := gbam.Shard{
				StartRef: header.Refs()[startCoord.RefId],
				Start:    int(startCoord.Pos),
				EndRef:   header.Refs()[limitCoord.RefId],
				End:      int(limitCoord.Pos),
				ShardIdx: len(shards),
			}
			startChunk = chunk
			shards = append(shards, shard)
		}
	}
	startCoord, err := gbam.GetCoordAtOffset(bamr, startChunk)
	require.NoError(b, err)
	shard := gbam.Shard{
		StartRef: header.Refs()[startCoord.RefId],
		Start:    int(startCoord.Pos),
		EndRef:   nil,
		End:      math.MaxInt32,
		ShardIdx: len(shards),
	}
	return append(shards, shard)
}
func shuffleRecords(t testing.TB, bamPath, baiPath string) (*sam.Header, []*sam.Record) {
	provider := bamprovider.NewProvider(bamPath, bamprovider.ProviderOpts{Index: baiPath})
	header, err := provider.GetHeader()
	require.NoError(t, err)
	iter := provider.NewIterator(gbam.UniversalShard(header))
	recs := []*sam.Record{}
	for iter.Scan() {
		recs = append(recs, iter.Record())
	}
	require.NoError(t, iter.Close())
	require.NoError(t, provider.Close())
	r := rand.New(rand.NewSource(0))
	for i := range recs {
		j := r.Intn(i + 1)
		recs[i], recs[j] = recs[j], recs[i]
	}
	return header, recs
}
func benchmarkSortShard(b *testing.B, recs []*sam.Record, header *sam.Header, shardPath string, options SortOptions) {
	if _, err := os.Stat(shardPath); err == nil {
		return
	}
	log.Printf("Creating sortshard %v. This may take minutes", shardPath)
	s := NewSorter(shardPath, header, options)
	for i := range recs {
		s.AddRecord(recs[i])
		recs[i] = nil
	}
	require.NoError(b, s.Close())
}