// partitionByExistence partitions items into two slices: one containing items that
// are known to already exist on the server, and the other containing the remaining items.
496764
func (c *BundlingChecker) partitionByExistence(items []checkerItem) (existing, notExisting []checkerItem) {
	for _, item := range items {
		if c.exists(item.item) {
			existing = append(existing, item)
		} else {
			notExisting = append(notExisting, item)
		}
	}
	return existing, notExisting
}
605838
func (i *Index) RetainFileSet() (*FileSet, error) {
	i.mu.RLock()
	defer i.mu.RUnlock()

	fs, _ := NewFileSet(nil, i.sfile, nil)
	for _, p := range i.partitions {
		pfs, err := p.RetainFileSet()
		if err != nil {
			fs.Close()
			return nil, err
		}
		fs.files = append(fs.files, pfs.files...)
	}
	return fs, nil
}
961644
func dividePartitionsBetweenConsumers(consumers kazoo.ConsumergroupInstanceList, partitions partitionLeaders) map[string][]*kazoo.Partition {
	result := make(map[string][]*kazoo.Partition)

	plen := len(partitions)
	clen := len(consumers)
	if clen == 0 {
		return result
	}

	sort.Sort(partitions)
	sort.Sort(consumers)

	n := plen / clen
	m := plen % clen
	p := 0
	for i, consumer := range consumers {
		first := p
		last := first + n
		if m > 0 && i < m {
			last++
		}
		if last > plen {
			last = plen
		}

		for _, pl := range partitions[first:last] {
			result[consumer.ID] = append(result[consumer.ID], pl.partition)
		}
		p = last
	}

	return result
}
1047250
func OrderAndCheckPartitions(cell string, srvKeyspace *topodatapb.SrvKeyspace) error {
	// now check them all
	for _, partition := range srvKeyspace.Partitions {
		tabletType := partition.ServedType
		topoproto.ShardReferenceArray(partition.ShardReferences).Sort()

		// check the first Start is MinKey, the last End is MaxKey,
		// and the values in between match: End[i] == Start[i+1]
		first := partition.ShardReferences[0]
		if first.KeyRange != nil && len(first.KeyRange.Start) != 0 {
			return fmt.Errorf("keyspace partition for %v in cell %v does not start with min key", tabletType, cell)
		}
		last := partition.ShardReferences[len(partition.ShardReferences)-1]
		if last.KeyRange != nil && len(last.KeyRange.End) != 0 {
			return fmt.Errorf("keyspace partition for %v in cell %v does not end with max key", tabletType, cell)
		}
		for i := range partition.ShardReferences[0 : len(partition.ShardReferences)-1] {
			currShard := partition.ShardReferences[i]
			nextShard := partition.ShardReferences[i+1]
			currHasKeyRange := currShard.KeyRange != nil
			nextHasKeyRange := nextShard.KeyRange != nil
			if currHasKeyRange != nextHasKeyRange {
				return fmt.Errorf("shards with inconsistent KeyRanges for %v in cell %v. shards: %v, %v", tabletType, cell, currShard, nextShard)
			}
			if !currHasKeyRange {
				// this is the custom sharding case, all KeyRanges must be nil
				continue
			}
			if !bytes.Equal(currShard.KeyRange.End, nextShard.KeyRange.Start) {
				return fmt.Errorf("non-contiguous KeyRange values for %v in cell %v at shard %v to %v: %v != %v", tabletType, cell, i, i+1, hex.EncodeToString(currShard.KeyRange.End), hex.EncodeToString(nextShard.KeyRange.Start))
			}
		}
	}

	return nil
}
497573
func (t *Topic) Partition(id int32, replicas []int32) *Partition {
	return &Partition{ID: id, Replicas: replicas, topic: t}
}
1046347
func DiffPartitions(a []int32, b []int32) []int32 {
	var diff []int32
Iter:
	for _, eA := range a {
		for _, eB := range b {
			if eA == eB {
				continue Iter
			}
		}
		diff = append(diff, eA)
	}
	return diff
}
526460
func (m *UnpartitionedMemoryIdx) LoadPartition(partition int32, defs []schema.MetricDefinition) int {
	// UnpartitionedMemoryIdx isnt partitioned, so just ignore the partition passed and call Load()
	return m.Load(defs)
}
1336042
func (c *MemcacheCacher) IsExist(key string) bool {
	_, err := c.c.Get(key)
	return err == nil
}
986433
func (ps partitions) Partitions() []Partition {
	parts := make([]Partition, 0, len(ps))
	for i := 0; i < len(ps); i++ {
		parts = append(parts, ps[i].Partition())
	}

	return parts
}
1169453
func (c *KafkaConsumer) PartitionConsumerMetrics(topic string, partition int32) (PartitionConsumerMetrics, error) {
	if !c.config.EnableMetrics {
		return nil, ErrMetricsDisabled
	}

	c.partitionConsumersLock.Lock()
	defer c.partitionConsumersLock.Unlock()

	if !c.exists(topic, partition) {
		log.Info("Partition consumer for topic %s, partition %d does not exist", topic, partition)
		return nil, ErrPartitionConsumerDoesNotExist
	}

	return c.partitionConsumers[topic][partition].Metrics()
}
1221022
func NewPartition(pool Pool, buffers ...Buffer) Buffer {
	return &partition{
		Pool: pool,
		List: buffers,
	}
}
125744
func (t *Topic) parsePartitions(value []byte) (PartitionList, error) {
	var tm topicMetadata
	if err := json.Unmarshal(value, &tm); err != nil {
		return nil, err
	}

	result := make(PartitionList, len(tm.Partitions))
	for partitionNumber, replicas := range tm.Partitions {
		partitionID, err := strconv.ParseInt(partitionNumber, 10, 32)
		if err != nil {
			return nil, err
		}

		replicaIDs := make([]int32, 0, len(replicas))
		for _, r := range replicas {
			replicaIDs = append(replicaIDs, int32(r))
		}
		result[partitionID] = t.Partition(int32(partitionID), replicaIDs)
	}

	return result, nil
}
445369
func (i *Index) partition(key []byte) *Partition {
	return i.partitions[int(xxhash.Sum64(key)&(i.PartitionN-1))]
}
987557
func (c *SeriesPartitionCompactor) Compact(p *SeriesPartition) (time.Duration, error) {
	// Snapshot the partitions and index so we can check tombstones and replay at the end under lock.
	p.mu.RLock()
	segments := CloneSeriesSegments(p.segments)
	index := p.index.Clone()
	seriesN := p.index.Count()
	p.mu.RUnlock()

	now := time.Now()

	// Compact index to a temporary location.
	indexPath := index.path + ".compacting"
	if err := c.compactIndexTo(index, seriesN, segments, indexPath); err != nil {
		return 0, err
	}
	duration := time.Since(now)

	// Swap compacted index under lock & replay since compaction.
	if err := func() error {
		p.mu.Lock()
		defer p.mu.Unlock()

		// Reopen index with new file.
		if err := p.index.Close(); err != nil {
			return err
		} else if err := os.Rename(indexPath, index.path); err != nil {
			return err
		} else if err := p.index.Open(); err != nil {
			return err
		}

		// Replay new entries.
		if err := p.index.Recover(p.segments); err != nil {
			return err
		}
		return nil
	}(); err != nil {
		return 0, err
	}

	return duration, nil
}
1108426
func (r *ring) getPartition(key []byte) *partition {
	return r.partitions[int(xxhash.Sum64(key)%partitions)]
}
1302607
func (f *fragment) rowFromStorage(rowID uint64) *Row {
	// Only use a subset of the containers.
	// NOTE: The start & end ranges must be divisible by container width.
	data := f.storage.OffsetRange(f.shard*ShardWidth, rowID*ShardWidth, (rowID+1)*ShardWidth)

	// Reference bitmap subrange in storage. We Clone() data because otherwise
	// row will contain pointers to containers in storage. This causes
	// unexpected results when we cache the row and try to use it later.
	// Basically, since we return the Row and release the fragment lock, the
	// underlying fragment storage could be changed or snapshotted and thrown
	// out at any point.
	row := &Row{
		segments: []rowSegment{{
			data:     *data.Clone(),
			shard:    f.shard,
			writable: false, // this Row will probably be cached and shared, so it must be read only.
		}},
	}
	row.invalidateCount()

	return row
}
781462
func (rs Ranges) Partition(n uint64) (Ranges, bool) {
	i := rs.Search(n)
	if i < 0 {
		return rs, false
	}

	pn := make(Ranges, 0, len(rs)+1)
	switch pn = append(pn, rs[:i]...); {
	case rs[i].Begin == rs[i].End: // delete
	case rs[i].Begin == n: // increment lower bound
		pn = append(pn, Value_Range{rs[i].Begin + 1, rs[i].End})
	case rs[i].End == n: // decrement upper bound
		pn = append(pn, Value_Range{rs[i].Begin, rs[i].End - 1})
	default: // split
		pn = append(pn, Value_Range{rs[i].Begin, n - 1}, Value_Range{n + 1, rs[i].End})
	}
	return append(pn, rs[i+1:]...), true
}