// partitionInnerRows re-order e.innerResults into sub-relations.
323758
func (e *RadixHashJoinExec) partitionInnerRows() error {
	e.evalRadixBit()
	if err := e.preAlloc4InnerParts(); err != nil {
		return err
	}

	wg := sync.WaitGroup{}
	defer wg.Wait()
	wg.Add(int(e.concurrency))
	for i := 0; i < int(e.concurrency); i++ {
		workerID := i
		go util.WithRecovery(func() {
			defer wg.Done()
			e.doInnerPartition(workerID)
		}, e.handlePartitionPanic)
	}
	return nil
}
337246
func (builder *SearchCommandBuilder) WithNumRows(numRows uint32) *SearchCommandBuilder {
	builder.protobuf.Rows = &numRows
	return builder
}
647462
func (e *MergeJoinExec) fetchNextOuterRows(ctx context.Context, requiredRows int) (err error) {
	// It's hard to calculate selectivity if there is any filter or it's inner join,
	// so we just push the requiredRows down when it's outer join and has no filter.
	if e.isOuterJoin && len(e.outerTable.filter) == 0 {
		e.outerTable.chk.SetRequiredRows(requiredRows, e.maxChunkSize)
	}

	err = e.outerTable.reader.Next(ctx, chunk.NewRecordBatch(e.outerTable.chk))
	if err != nil {
		return err
	}

	e.outerTable.iter.Begin()
	e.outerTable.selected, err = expression.VectorizedFilter(e.ctx, e.outerTable.filter, e.outerTable.iter, e.outerTable.selected)
	if err != nil {
		return err
	}
	e.outerTable.row = e.outerTable.iter.Begin()
	return nil
}
932576
func NewRow(data ...interface{}) *Row {
	r := &Row{Cells: make([]*Cell, len(data))}
	for i, d := range data {
		r.Cells[i] = &Cell{Data: d}
	}
	return r
}
1246974
func selectRows(row virtualRow, predicates []PredicateLinker, functors []selectFunctor) error {
	var res bool
	var err error

	// If the row validate all predicates, write it
	for _, predicate := range predicates {
		if res, err = predicate.Eval(row); err != nil {
			return err
		}
		if res == false {
			return nil
		}
	}

	for i := range functors {
		err := functors[i].FeedVirtualRow(row)
		if err != nil {
			return err
		}
	}
	return nil
}
1039569
func (rd *RowDiffer2) reconcileRow(row []sqltypes.Value, typ DiffType) error {
	if typ == DiffNotEqual {
		panic(fmt.Sprintf("reconcileRow() called with wrong type: %v", typ))
	}
	destShardIndex, err := rd.router.Route(row)
	if err != nil {
		return vterrors.Wrapf(err, "failed to route row (%v) to correct shard", row)
	}

	if err := rd.aggregators[destShardIndex][typ].Add(row); err != nil {
		return vterrors.Wrap(err, "failed to add row update to RowAggregator")
	}
	// TODO(mberlin): Add more fine granular stats here.
	rd.tableStatusList.addCopiedRows(rd.tableIndex, 1)
	return nil
}
703936
func (m matrix) SwapRows(r1, r2 int) error {
	if r1 < 0 || len(m) <= r1 || r2 < 0 || len(m) <= r2 {
		return errInvalidRowSize
	}
	m[r2], m[r1] = m[r1], m[r2]
	return nil
}
285765
func NewIterator4RowPtr(li *List, ptrs []RowPtr) Iterator {
	return &iterator4RowPtr{li: li, ptrs: ptrs}
}
789417
func CountMergedRows(rows []*AddressRow, txnView AddrTxnViewType) (numMerged int, err error) {
	var wrongDirection func(funding bool) bool
	switch txnView {
	case AddrMergedTxn:
		wrongDirection = func(_ bool) bool {
			return false
		}
	case AddrMergedTxnCredit:
		wrongDirection = func(funding bool) bool {
			return !funding
		}
	case AddrMergedTxnDebit:
		wrongDirection = func(funding bool) bool {
			return funding
		}
	default:
		return 0, fmt.Errorf("MergedTxnCount: requested count for non-merged view")
	}

	merged := make(map[string]struct{})
	for _, r := range rows {
		if r.MergedCount != 0 {
			return 0, fmt.Errorf("CountMergedRows: merged row found in input; " +
				"only non-merged rows may be merged")
		}

		if wrongDirection(r.IsFunding) {
			continue
		}

		hash := r.TxHash
		_, found := merged[hash]
		if !found {
			numMerged++
			merged[hash] = struct{}{}
			continue
		}
	}
	return
}
1242613
func buildRowNumber(aggFuncDesc *aggregation.AggFuncDesc, ordinal int) AggFunc {
	base := baseAggFunc{
		args:    aggFuncDesc.Args,
		ordinal: ordinal,
	}
	return &rowNumber{base}
}
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
}
1179934
func (e *HashJoinExec) fetchInnerRows(ctx context.Context) error {
	e.innerResult = chunk.NewList(e.innerExec.retTypes(), e.initCap, e.maxChunkSize)
	e.innerResult.GetMemTracker().AttachTo(e.memTracker)
	e.innerResult.GetMemTracker().SetLabel(innerResultLabel)
	var err error
	for {
		if e.finished.Load().(bool) {
			return nil
		}
		chk := e.children[e.innerIdx].newFirstChunk()
		err = e.innerExec.Next(ctx, chunk.NewRecordBatch(chk))
		if err != nil || chk.NumRows() == 0 {
			return err
		}
		e.innerResult.Add(chk)
	}
}
856917
func buildFirstRow(aggFuncDesc *aggregation.AggFuncDesc, ordinal int) AggFunc {
	base := baseAggFunc{
		args:    aggFuncDesc.Args,
		ordinal: ordinal,
	}

	evalType, fieldType := aggFuncDesc.RetTp.EvalType(), aggFuncDesc.RetTp
	if fieldType.Tp == mysql.TypeBit {
		evalType = types.ETString
	}
	switch aggFuncDesc.Mode {
	case aggregation.DedupMode:
	default:
		switch evalType {
		case types.ETInt:
			return &firstRow4Int{base}
		case types.ETReal:
			switch fieldType.Tp {
			case mysql.TypeFloat:
				return &firstRow4Float32{base}
			case mysql.TypeDouble:
				return &firstRow4Float64{base}
			}
		case types.ETDecimal:
			return &firstRow4Decimal{base}
		case types.ETDatetime, types.ETTimestamp:
			return &firstRow4Time{base}
		case types.ETDuration:
			return &firstRow4Duration{base}
		case types.ETString:
			return &firstRow4String{base}
		case types.ETJson:
			return &firstRow4JSON{base}
		}
	}
	return nil
}
1337913
func (ckb *ckb_IR) WeekdayNarrow(weekday time.Weekday) string {
	return ckb.daysNarrow[weekday]
}
925336
func newRows(qr *sqltypes.Result, c *converter) driver.Rows {
	return &rows{qr: qr, convert: c}
}
217545
func (m *Table) AsRows() [][]driver.Value {
	if len(m.rows) > 0 {
		return m.rows
	}
	m.rows = make([][]driver.Value, len(m.Fields))
	for i, f := range m.Fields {
		m.rows[i] = f.AsRow()
	}
	return m.rows
}
1254830
func (nds *nds_DE) WeekdayNarrow(weekday time.Weekday) string {
	return nds.daysNarrow[weekday]
}