// Reduce runs the reducer registered to the reducerId,
// combining all rows into one row
6500
func (d *Dataset) Reduce(name string, reducerId gio.ReducerId) (ret *Dataset) {

	name = name + ".Reduce"

	ret = d.LocalReduceBy(name+".LocalReduce", reducerId, nil)
	if len(d.Shards) > 1 {
		ret = ret.MergeTo(name, 1).LocalReduceBy(name+".LocalReduce2", reducerId, nil)
	}
	return ret
}
932633
func (rows *OCI8Rows) Close() error {
	if rows.closed {
		return nil
	}

	rows.closed = true
	close(rows.done)

	if rows.cls {
		rows.stmt.Close()
	}

	freeDefines(rows.defines)

	return nil
}
1069971
func (iter *iterator) exhaust() keys {
	keys := make(keys, 0, 10)
	for iter := iter; iter.Next(); {
		keys = append(keys, iter.Value())
	}

	return keys
}
784346
func Finalize(finalizers ...Finalizer) error {
	var errs ErrorGroup
	for _, finalizer := range finalizers {
		errs.Add(finalizer.Finalize())
	}
	return errs.Finalize()
}
421618
func (r VisvalingamReducer) Reduce(path *geo.Path) *geo.Path {
	return Visvalingam(path, r.Threshold, r.ToKeep)
}
215048
func (r *Chain) Reduce(input chan *Entry, output chan *Entry) {
	// Make input and output channel for each reducer
	subInput := make([]chan *Entry, len(r.reducers))
	subOutput := make([]chan *Entry, len(r.reducers))
	for i, reducer := range r.reducers {
		subInput[i] = make(chan *Entry, cap(input))
		subOutput[i] = make(chan *Entry, cap(output))
		go reducer.Reduce(subInput[i], subOutput[i])
	}

	// Read reducer master input channel
	for entry := range input {
		for _, f := range r.filters {
			entry = f.Filter(entry)
			if entry == nil {
				break
			}
		}
		// Publish input entry for each sub-reducers to process
		if entry != nil {
			for _, sub := range subInput {
				sub <- entry
			}
		}
	}
	for _, ch := range subInput {
		close(ch)
	}

	// Merge all results
	entry := NewEmptyEntry()
	for _, result := range subOutput {
		entry.Merge(<-result)
	}

	output <- entry
	close(output)
}
1062302
func (s *containerSorter) Less(i, j int) bool {
	return s.by(s.containers[i], s.containers[j])
}
809953
func (s *IndicesForcemergeService) Flush(flush bool) *IndicesForcemergeService {
	s.flush = &flush
	return s
}
80321
func (rs *ReducerStruct) ReduceEnd(c []sophie.Collector) error {
	if rs.ReduceEndF != nil {
		return rs.ReduceEndF(c)
	}
	return nil
}
1287971
func (r RadialGeoReducer) GeoReduce(path *geo.Path) *geo.Path {
	return RadialGeo(path, r.Threshold)
}
1286263
func (rh *rowHeap) Less(i, j int) bool {
	l := rh.ims[i].value.key
	r := rh.ims[j].value.key
	ret, err := lessThan(rh.sc, l, r, rh.byDesc)
	if rh.err == nil {
		rh.err = err
	}
	return ret
}
997051
func (rows *Rows) Close() error {
	if rows.session.IsAutoClose {
		defer rows.session.Close()
	}

	if rows.lastError == nil {
		if rows.rows != nil {
			rows.lastError = rows.rows.Close()
			if rows.lastError != nil {
				defer rows.stmt.Close()
				return rows.lastError
			}
		}
		if rows.stmt != nil {
			rows.lastError = rows.stmt.Close()
		}
	} else {
		if rows.stmt != nil {
			defer rows.stmt.Close()
		}
		if rows.rows != nil {
			defer rows.rows.Close()
		}
	}
	return rows.lastError
}
550391
func (r RadialGeoReducer) Reduce(path *geo.Path) *geo.Path {
	return RadialGeo(path, r.Threshold)
}
1005338
func (d *Dataset) Sort(name string, sortOption *SortOption) *Dataset {
	ret := d.LocalSort(name, sortOption)
	ret = ret.TreeMergeSortedTo(name, 1, 10)
	return ret
}
581982
func (r *Redis) Delete(key string) error {
	return r.Client.Del(key).Err()
}
883877
func (m *podContainerManagerImpl) ReduceCPULimits(podCgroup CgroupName) error {
	return m.cgroupManager.ReduceCPULimits(podCgroup)
}
289464
func (r *Row) Count() uint64 {
	var n uint64
	for i := range r.segments {
		n += r.segments[i].Count()
	}
	return n
}