//TODO if len(rows) > 1000, chunk it up and exec multiple inserts in the txn
344310
func (dbMan *dbManager) insert(tableName string, rows []common.Row, txn apid.Tx) error {
	if len(rows) == 0 {
		return fmt.Errorf("no rows")
	}

	var orderedColumns []string
	for column := range rows[0] {
		orderedColumns = append(orderedColumns, column)
	}
	sort.Strings(orderedColumns)

	sql := dbMan.buildInsertSql(tableName, orderedColumns, rows)

	prep, err := txn.Prepare(sql)
	if err != nil {
		log.Errorf("INSERT Fail to prepare statement %s error=%v", sql, err)
		return err
	}
	defer prep.Close()

	var values []interface{}

	for _, row := range rows {
		for _, columnName := range orderedColumns {
			//use Value so that stmt exec does not complain about common.ColumnVal being a struct
			values = append(values, row[columnName].Value)
		}
	}

	//create prepared statement from existing template statement
	_, err = prep.Exec(values...)

	if err != nil {
		log.Errorf("INSERT Fail %s values=%v error=%v", sql, values, err)
		return err
	}
	log.Debugf("INSERT Success %s values=%v", sql, values)

	return nil
}
682574
func (bt *Translator) BulkAdd(field string, values [][]byte) error {
	var batchSize uint64 = 10000
	var batch uint64
	for batch*batchSize < uint64(len(values)) {
		err := bt.Db.Batch(func(tx *bolt.Tx) error {
			fib := tx.Bucket(idBucket).Bucket([]byte(field))
			fvb := tx.Bucket(valBucket).Bucket([]byte(field))

			for i := batch * batchSize; i < (batch+1)*batchSize && i < uint64(len(values)); i++ {
				idBytes := make([]byte, 8)
				binary.BigEndian.PutUint64(idBytes, i)
				valBytes := values[i]
				err := fib.Put(idBytes, valBytes)
				if err != nil {
					return errors.Wrap(err, "putting into idKey bucket")
				}
				err = fvb.Put(valBytes, idBytes)
				if err != nil {
					return errors.Wrap(err, "putting into valKey bucket")
				}
			}
			return nil
		})
		if err != nil {
			return errors.Wrap(err, "inserting batch")
		}
		batch++
	}
	return nil
}
708703
func (itc *internalTabletConn) ExecuteBatch(ctx context.Context, target *querypb.Target, queries []*querypb.BoundQuery, asTransaction bool, transactionID int64, options *querypb.ExecuteOptions) ([]sqltypes.Result, error) {
	q := make([]*querypb.BoundQuery, len(queries))
	for i, query := range queries {
		q[i] = &querypb.BoundQuery{
			Sql:           query.Sql,
			BindVariables: sqltypes.CopyBindVariables(query.BindVariables),
		}
	}
	results, err := itc.tablet.qsc.QueryService().ExecuteBatch(ctx, target, q, asTransaction, transactionID, options)
	if err != nil {
		return nil, tabletconn.ErrorFromGRPC(vterrors.ToGRPC(err))
	}
	return results, nil
}
105476
func (e *InsertExec) batchUpdateDupRows(newRows [][]types.Datum) error {
	err := e.batchGetInsertKeys(e.ctx, e.Table, newRows)
	if err != nil {
		return err
	}

	// Batch get the to-be-updated rows in storage.
	err = e.initDupOldRowValue(e.ctx, e.Table, newRows)
	if err != nil {
		return err
	}

	for i, r := range e.toBeCheckedRows {
		if r.handleKey != nil {
			if _, found := e.dupKVs[string(r.handleKey.newKV.key)]; found {
				handle, err := tablecodec.DecodeRowKey(r.handleKey.newKV.key)
				if err != nil {
					return err
				}
				err = e.updateDupRow(r, handle, e.OnDuplicate)
				if err != nil {
					return err
				}
				continue
			}
		}
		for _, uk := range r.uniqueKeys {
			if val, found := e.dupKVs[string(uk.newKV.key)]; found {
				handle, err := tables.DecodeHandle(val)
				if err != nil {
					return err
				}
				err = e.updateDupRow(r, handle, e.OnDuplicate)
				if err != nil {
					return err
				}
				newRows[i] = nil
				break
			}
		}
		// If row was checked with no duplicate keys,
		// we should do insert the row,
		// and key-values should be filled back to dupOldRowValues for the further row check,
		// due to there may be duplicate keys inside the insert statement.
		if newRows[i] != nil {
			newHandle, err := e.addRecord(newRows[i])
			if err != nil {
				return err
			}
			e.fillBackKeys(e.Table, r, newHandle)
		}
	}
	return nil
}
1279151
func (pool *TxPool) loop() {
	defer pool.wg.Done()

	// Start the stats reporting and transaction eviction tickers
	var prevPending, prevQueued, prevStales int

	report := time.NewTicker(statsReportInterval)
	defer report.Stop()

	evict := time.NewTicker(evictionInterval)
	defer evict.Stop()

	journal := time.NewTicker(pool.config.Rejournal)
	defer journal.Stop()

	// Track the previous head headers for transaction reorgs
	head := pool.chain.CurrentBlock()

	// Keep waiting for and reacting to the various events
	for {
		select {
		// Handle ChainHeadEvent
		case ev := <-pool.chainHeadCh:
			if ev.Block != nil {
				pool.mu.Lock()
				if pool.chainconfig.IsHomestead(ev.Block.Number()) {
					pool.homestead = true
				}
				pool.reset(head.Header(), ev.Block.Header())
				head = ev.Block

				pool.mu.Unlock()
			}
		// Be unsubscribed due to system stopped
		case <-pool.chainHeadSub.Err():
			return

		// Handle stats reporting ticks
		case <-report.C:
			pool.mu.RLock()
			pending, queued := pool.stats()
			stales := pool.priced.stales
			pool.mu.RUnlock()

			if pending != prevPending || queued != prevQueued || stales != prevStales {
				log.Debug("Transaction pool status report", "executable", pending, "queued", queued, "stales", stales)
				prevPending, prevQueued, prevStales = pending, queued, stales
			}

		// Handle inactive account transaction eviction
		case <-evict.C:
			pool.mu.Lock()
			for addr := range pool.queue {
				// Skip local transactions from the eviction mechanism
				if pool.locals.contains(addr) {
					continue
				}
				// Any non-locals old enough should be removed
				if time.Since(pool.beats[addr]) > pool.config.Lifetime {
					for _, tx := range pool.queue[addr].Flatten() {
						pool.removeTx(tx.Hash(), true)
					}
				}
			}
			pool.mu.Unlock()

		// Handle local transaction journal rotation
		case <-journal.C:
			if pool.journal != nil {
				pool.mu.Lock()
				if err := pool.journal.rotate(pool.local()); err != nil {
					log.Warn("Failed to rotate local tx journal", "err", err)
				}
				pool.mu.Unlock()
			}
		}
	}
}
1054294
func (b *bucket) Tx() walletdb.ReadWriteTx {
	return &transaction{
		(*bbolt.Bucket)(b).Tx(),
	}
}
128524
func (m *dbConn) Put(ctx context.Context, key schema.Key, row interface{}) (schema.Key, error) {

	switch rowVals := row.(type) {
	case []driver.Value:
		txn := m.db.Txn(true)
		key, err := m.putValues(txn, rowVals)
		if err != nil {
			txn.Abort()
			return nil, err
		}
		txn.Commit()
		return key, nil
	default:
		return nil, fmt.Errorf("Expected []driver.Value but got %T", row)
	}
}
430106
func (d *dbBase) InsertMulti(q dbQuerier, mi *modelInfo, sind reflect.Value, bulk int, tz *time.Location) (int64, error) {
	var (
		cnt    int64
		nums   int
		values []interface{}
		names  []string
	)

	// typ := reflect.Indirect(mi.addrField).Type()

	length := sind.Len()

	for i := 1; i <= length; i++ {

		ind := reflect.Indirect(sind.Index(i - 1))

		// Is this needed ?
		// if !ind.Type().AssignableTo(typ) {
		// 	return cnt, ErrArgs
		// }

		if i == 1 {
			vus, err := d.collectValues(mi, ind, mi.fields.dbcols, true, true, &names, tz)
			if err != nil {
				return cnt, err
			}
			values = make([]interface{}, bulk*len(vus))
			nums += copy(values, vus)

		} else {

			vus, err := d.collectValues(mi, ind, mi.fields.dbcols, true, true, nil, tz)
			if err != nil {
				return cnt, err
			}

			if len(vus) != len(names) {
				return cnt, ErrArgs
			}

			nums += copy(values[nums:], vus)
		}

		if i > 1 && i%bulk == 0 || length == i {
			num, err := d.InsertValue(q, mi, true, names, values[:nums])
			if err != nil {
				return cnt, err
			}
			cnt += num
			nums = 0
		}
	}

	return cnt, nil
}
768974
func (self *MyTable) FlushInsert() error {
	if self.rowsCount == 0 {
		return nil
	}

	colCount := len(self.columnNames)
	if colCount == 0 {
		return nil
	}

	self.sqlCode = `INSERT INTO ` + self.tableName + `(`

	for _, v := range self.columnNames {
		self.sqlCode += v[0] + ","
	}

	self.sqlCode = self.sqlCode[:len(self.sqlCode)-1] + `) VALUES `

	blank := ",(" + strings.Repeat(",?", colCount)[1:] + ")"
	self.sqlCode += strings.Repeat(blank, self.rowsCount)[1:] + `;`

	defer func() {
		// 清空临时数据
		self.args = []interface{}{}
		self.rowsCount = 0
		self.size = 0
		self.sqlCode = ""
	}()

	maxConnChan <- true
	defer func() {
		<-maxConnChan
	}()

	// debug
	// println("FlushInsert():", self.sqlCode)

	_, err := db.Exec(self.sqlCode, self.args...)
	return err
}
823152
func (d *Dataset) Insert(index int, row []interface{}) error {
	if index < 0 || index >= d.rows {
		return ErrInvalidRowIndex
	}

	if len(row) != d.cols {
		return ErrInvalidDimensions
	}

	ndata := make([][]interface{}, 0, d.rows+1)
	ndata = append(ndata, d.data[:index]...)
	ndata = append(ndata, row)
	ndata = append(ndata, d.data[index:]...)
	d.data = ndata
	d.rows++

	ntags := make([][]string, 0, d.rows+1)
	ntags = append(ntags, d.tags[:index]...)
	ntags = append(ntags, make([]string, 0))
	ntags = append(ntags, d.tags[index:]...)
	d.tags = ntags

	return nil
}
715395
func RowProcessing(rows *core.Rows, fields []string, fn func(data *reflect.Value, index int, fieldName string) error) (err error) {
	length := len(fields)
	scanResultContainers := make([]interface{}, length)
	for i := 0; i < length; i++ {
		var resultContainer interface{}
		scanResultContainers[i] = &resultContainer
	}
	if err := rows.Scan(scanResultContainers...); err != nil {
		return err
	}
	for ii, key := range fields {
		rawValue := reflect.Indirect(reflect.ValueOf(scanResultContainers[ii]))
		fn(&rawValue, ii, key)
	}
	return nil
}
930519
func dbAddTxIndexEntries(dbTx database.Tx, block *btcutil.Block, blockID uint32) error {
	// The offset and length of the transactions within the serialized
	// block.
	txLocs, err := block.TxLoc()
	if err != nil {
		return err
	}

	// As an optimization, allocate a single slice big enough to hold all
	// of the serialized transaction index entries for the block and
	// serialize them directly into the slice.  Then, pass the appropriate
	// subslice to the database to be written.  This approach significantly
	// cuts down on the number of required allocations.
	offset := 0
	serializedValues := make([]byte, len(block.Transactions())*txEntrySize)
	for i, tx := range block.Transactions() {
		putTxIndexEntry(serializedValues[offset:], blockID, txLocs[i])
		endOffset := offset + txEntrySize
		err := dbPutTxIndexEntry(dbTx, tx.Hash(),
			serializedValues[offset:endOffset:endOffset])
		if err != nil {
			return err
		}
		offset += txEntrySize
	}

	return nil
}
786856
func BatchInsert(db Database, col string) DocWriter {
	if bi, ok := db.(BatchInserter); ok {
		return bi.BatchInsert(col)
	}
	return &seqInsert{db: db, col: col}
}
1281991
func (b *BulkIndexer) Flush() {
	b.mu.Lock()
	if b.docCt > 0 {
		b.send(b.buf)
	}
	b.mu.Unlock()
}
802866
func (e *executor) executeCount(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (uint64, error) {
	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeCount")
	defer span.Finish()

	if len(c.Children) == 0 {
		return 0, errors.New("Count() requires an input bitmap")
	} else if len(c.Children) > 1 {
		return 0, errors.New("Count() only accepts a single bitmap input")
	}

	// Execute calls in bulk on each remote node and merge.
	mapFn := func(shard uint64) (interface{}, error) {
		row, err := e.executeBitmapCallShard(ctx, index, c.Children[0], shard)
		if err != nil {
			return 0, err
		}
		return row.Count(), nil
	}

	// Merge returned results at coordinating node.
	reduceFn := func(prev, v interface{}) interface{} {
		other, _ := prev.(uint64)
		return other + v.(uint64)
	}

	result, err := e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn)
	if err != nil {
		return 0, err
	}
	n, _ := result.(uint64)

	return n, nil
}
1358018
func (c *conn) ExecTx(fn func(tx *trans) error) error {
	if c.flavor.executeTx != nil {
		return c.flavor.executeTx(c.db, func(sqlTx *sql.Tx) error {
			return fn(&trans{sqlTx, c})
		})
	}

	sqlTx, err := c.db.Begin()
	if err != nil {
		return err
	}
	if err := fn(&trans{sqlTx, c}); err != nil {
		sqlTx.Rollback()
		return err
	}
	return sqlTx.Commit()
}
180948
func MaybeCommitTx(ctx context.Context, storage interface{}) error {
	txnStorage, transactional := storage.(Transactional)
	if transactional {
		return txnStorage.Commit(ctx)
	} else {
		return nil
	}
}