func generatePAMShard(readers []*sortShardReader,
	path string,
	header *sam.Header,
	start, limit recCoord,
	pool *sortShardBlockPool,
	errReporter *errors.Once) {
	opts := pam.WriteOpts{
		Range: biopb.CoordRange{recCoordToRecAddr(start), recCoordToRecAddr(limit)},
	}
	vlog.VI(1).Infof("%v: Generating PAM shard %+v", path, opts)
	pamWriter := pam.NewWriter(opts, header, path)
	readCallback := func(key sortEntry) bool {
		if key.coord >= limit {
			return false
		}
		if key.coord >= start {
			rec, err := grailbam.Unmarshal(key.body[4:], header)
			if err != nil {
				errReporter.Set(err)
				return false
			}
			pamWriter.Write(rec)
			if pamWriter.Err() != nil {
				vlog.Fatalf("ERR: %+v, opts %+v key %+v, limit %+v", pamWriter.Err(), opts, key, limit)
			}
			sam.PutInFreePool(rec)
		}
		return true
	}
	internalMergeShards(readers, readCallback, pool, errReporter)
	errReporter.Set(pamWriter.Close())
}
Generate one PAM rowshard that stores reads in range [start, limit).
package sorter
import (
	"fmt"
	"math"
	"sort"
	"sync"
	"github.com/grailbio/base/errors"
	"github.com/grailbio/bio/biopb"
	grailbam "github.com/grailbio/bio/encoding/bam"
	"github.com/grailbio/bio/encoding/pam"
	"github.com/grailbio/bio/encoding/pam/pamutil"
	"github.com/grailbio/hts/sam"
	"v.io/x/lib/vlog"
)
func computePAMShardBounds(
	shards [][]biopb.SortShardBlockIndex,
	recordsPerShard int64) (bounds []recCoord) {
	clearReverseFlag := func(coord recCoord) recCoord {
		ref, pos, _ := parseCoord(coord)
		return makeCoord(ref, pos, false)
	}
	allKeys := []sortEntry{}
	totalRecords := int64(0)
	totalBlocks := int64(0)
	for _, shard := range shards {
		for _, block := range shard {
			allKeys = append(allKeys, sortEntry{recCoord(block.StartKey), nil})
			totalRecords += int64(block.NumRecords)
		}
		totalBlocks += int64(len(shard))
	}
	if totalRecords == 0 {
		return nil
	}
	sort.SliceStable(allKeys, func(i, j int) bool {
		return allKeys[i].compare(allKeys[j]) < 0
	})
	nextGoalRecords := float64(recordsPerShard)
	recordsPerBlock := float64(totalRecords) / float64(totalBlocks)
	curNumRecords := float64(0)
	for _, key := range allKeys[1:] {
		curNumRecords += recordsPerBlock
		if curNumRecords < nextGoalRecords {
			continue
		}
		coord := clearReverseFlag(key.coord)
		if len(bounds) > 0 && bounds[len(bounds)-1] == coord {
			continue
		}
		bounds = append(bounds, coord)
		nextGoalRecords = curNumRecords + float64(recordsPerShard)
	}
	return
}
func startFileOffset(blocks []biopb.SortShardBlockIndex, coord recCoord) int64 {
	key := sortEntry{coord: coord, body: nil}
	n := sort.Search(len(blocks), func(i int) bool {
		b := sortEntry{recCoord(blocks[i].StartKey), nil}
		return b.compare(key) >= 0
	})
	if n >= len(blocks) {
		return int64(blocks[len(blocks)-1].FileOffset)
	}
	if n >= 1 {
		return int64(blocks[n-1].FileOffset)
	}
	return 0
}
func limitFileOffset(blocks []biopb.SortShardBlockIndex, coord recCoord) int64 {
	key := sortEntry{coord: coord, body: nil}
	n := sort.Search(len(blocks), func(i int) bool {
		b := sortEntry{recCoord(blocks[i].StartKey), nil}
		return b.compare(key) >= 0
	})
	if n >= len(blocks) {
		return math.MaxInt64
	}
	return int64(blocks[n].FileOffset)
}
func recCoordToRecAddr(coord recCoord) (r biopb.Coord) {
	switch coord {
	case 0:
		return
	case unmappedCoord:
		r.RefId = biopb.InfinityRefID
		r.Pos = 0
		return
	case invalidCoord:
		r.RefId = biopb.InfinityRefID
		r.Pos = math.MaxInt32
		return
	default:
		refid, pos, _ := parseCoord(coord)
		if refid < 0 {
			r = biopb.Coord{biopb.UnmappedRefID, 0, 0}
		} else {
			r = biopb.Coord{int32(refid), int32(pos), 0}
		}
		return
	}
}