// NewTxRecord creates a new transaction record that may be inserted into the
// store.  It uses memoization to save the transaction hash and the serialized
// transaction.
191490
func NewTxRecord(serializedTx []byte, received time.Time) (*TxRecord, error) {
	rec := &TxRecord{
		Received:     received,
		SerializedTx: serializedTx,
	}
	err := rec.MsgTx.Deserialize(bytes.NewReader(serializedTx))
	if err != nil {
		str := "failed to deserialize transaction"
		return nil, storeError(ErrInput, str, err)
	}
	copy(rec.Hash[:], chainhash.DoubleHashB(serializedTx))
	return rec, nil
}
717487
func NewTxStatusEvent(txID string, txValidationCode pb.TxValidationCode, blockNum uint64, sourceURL string) *fab.TxStatusEvent {
	return &fab.TxStatusEvent{
		TxID:             txID,
		TxValidationCode: txValidationCode,
		BlockNumber:      blockNum,
		SourceURL:        sourceURL,
	}
}
180948
func MaybeCommitTx(ctx context.Context, storage interface{}) error {
	txnStorage, transactional := storage.(Transactional)
	if transactional {
		return txnStorage.Commit(ctx)
	} else {
		return nil
	}
}
175199
func (t *tableCommon) AddRecord(ctx sessionctx.Context, r []types.Datum, opts ...*table.AddRecordOpt) (recordID int64, err error) {
	opt := &table.AddRecordOpt{}
	if len(opts) != 0 {
		opt = opts[0]
	}
	var hasRecordID bool
	cols := t.Cols()
	// opt.IsUpdate is a flag for update.
	// If handle ID is changed when update, update will remove the old record first, and then call `AddRecord` to add a new record.
	// Currently, only insert can set _tidb_rowid, update can not update _tidb_rowid.
	if len(r) > len(cols) && !opt.IsUpdate {
		// The last value is _tidb_rowid.
		recordID = r[len(r)-1].GetInt64()
		hasRecordID = true
	} else {
		for _, col := range cols {
			if col.IsPKHandleColumn(t.meta) {
				recordID = r[col.Offset].GetInt64()
				hasRecordID = true
				break
			}
		}
	}
	if !hasRecordID {
		recordID, err = t.AllocAutoID(ctx)
		if err != nil {
			return 0, err
		}
	}

	txn, err := ctx.Txn(true)
	if err != nil {
		return 0, err
	}

	sessVars := ctx.GetSessionVars()

	rm, err := t.getRollbackableMemStore(ctx)
	// Insert new entries into indices.
	h, err := t.addIndices(ctx, recordID, r, rm, &opt.CreateIdxOpt)
	if err != nil {
		return h, err
	}

	var colIDs, binlogColIDs []int64
	var row, binlogRow []types.Datum
	colIDs = make([]int64, 0, len(r))
	row = make([]types.Datum, 0, len(r))

	for _, col := range t.WritableCols() {
		var value types.Datum
		// Update call `AddRecord` will already handle the write only column default value.
		// Only insert should add default value for write only column.
		if col.State != model.StatePublic && !opt.IsUpdate {
			// If col is in write only or write reorganization state, we must add it with its default value.
			value, err = table.GetColOriginDefaultValue(ctx, col.ToInfo())
			if err != nil {
				return 0, err
			}
			// add value to `r` for dirty db in transaction.
			// Otherwise when update will panic cause by get value of column in write only state from dirty db.
			if col.Offset < len(r) {
				r[col.Offset] = value
			} else {
				r = append(r, value)
			}
		} else {
			value = r[col.Offset]
		}
		if !t.canSkip(col, value) {
			colIDs = append(colIDs, col.ID)
			row = append(row, value)
		}
	}
	writeBufs := sessVars.GetWriteStmtBufs()
	adjustRowValuesBuf(writeBufs, len(row))
	key := t.RecordKey(recordID)
	writeBufs.RowValBuf, err = tablecodec.EncodeRow(ctx.GetSessionVars().StmtCtx, row, colIDs, writeBufs.RowValBuf, writeBufs.AddRowValues)
	if err != nil {
		return 0, err
	}
	value := writeBufs.RowValBuf
	if err = txn.Set(key, value); err != nil {
		return 0, err
	}
	txn.SetAssertion(key, kv.None)

	if !sessVars.LightningMode {
		if err = rm.(*kv.BufferStore).SaveTo(txn); err != nil {
			return 0, err
		}
	}

	if !ctx.GetSessionVars().LightningMode {
		ctx.StmtAddDirtyTableOP(table.DirtyTableAddRow, t.physicalTableID, recordID, r)
	}
	if shouldWriteBinlog(ctx) {
		// For insert, TiDB and Binlog can use same row and schema.
		binlogRow = row
		binlogColIDs = colIDs
		err = t.addInsertBinlog(ctx, recordID, binlogRow, binlogColIDs)
		if err != nil {
			return 0, err
		}
	}
	sessVars.StmtCtx.AddAffectedRows(1)
	colSize := make(map[int64]int64)
	for id, col := range t.Cols() {
		val := int64(len(r[id].GetBytes()))
		if val != 0 {
			colSize[col.ID] = val
		}
	}
	sessVars.TxnCtx.UpdateDeltaForTable(t.physicalTableID, 1, 1, colSize)
	return recordID, nil
}
172428
func NewTxnWatcher(config TxnWatcherConfig) (*TxnWatcher, error) {
	if err := config.Validate(); err != nil {
		return nil, errors.Annotate(err, "new TxnWatcher invalid config")
	}

	w := &TxnWatcher{
		hub:           config.Hub,
		clock:         config.Clock,
		logger:        config.Logger,
		log:           config.ChangeLog,
		iteratorFunc:  config.IteratorFunc,
		notifySync:    TxnPollNotifyFunc,
		reportRequest: make(chan chan map[string]interface{}),
	}
	if w.iteratorFunc == nil {
		w.iteratorFunc = w.iter
	}
	if w.logger == nil {
		w.logger = noOpLogger{}
	}
	w.tomb.Go(func() error {
		err := w.loop()
		cause := errors.Cause(err)
		// tomb expects ErrDying or ErrStillAlive as
		// exact values, so we need to log and unwrap
		// the error first.
		if err != nil && cause != tomb.ErrDying {
			w.logger.Infof("watcher loop failed: %v", err)
		}
		return cause
	})
	return w, nil
}
1326819
func (mem *Mempool) recheckTxs(txs []types.Tx) {
	if len(txs) == 0 {
		return
	}
	atomic.StoreInt32(&mem.rechecking, 1)
	mem.recheckCursor = mem.txs.Front()
	mem.recheckEnd = mem.txs.Back()

	// Push txs to proxyAppConn
	// NOTE: globalCb may be called concurrently.
	for _, tx := range txs {
		mem.proxyAppConn.CheckTxAsync(tx)
	}
	mem.proxyAppConn.FlushAsync()
}
587649
func (l *kvLedger) NewTxSimulator(txid string) (ledger.TxSimulator, error) {
	return l.txtmgmt.NewTxSimulator(txid)
}
355295
func dbPutTxIndexEntry(dbTx database.Tx, txHash *chainhash.Hash, serializedData []byte) error {
	txIndex := dbTx.Metadata().Bucket(txIndexKey)
	return txIndex.Put(txHash[:], serializedData)
}
1196601
func NewTxWriter(tx *Transaction, p Procedure) quad.Writer {
	if p == 0 {
		p = Add
	}
	return &txWriter{tx: tx, p: p}
}
1362880
func NewTransactionOrDie(ctx context.Context, store Store, params ...TransactionParams) Transaction {
	txn, err := store.NewTransaction(ctx, params...)
	if err != nil {
		panic(err)
	}
	return txn
}
981635
func (c *Client) AddTxtRecord(zoneName string, fqdn, value string, ttl int) error {
	host := dns01.UnFqdn(strings.TrimSuffix(dns01.UnFqdn(fqdn), zoneName))

	reqURL := *c.BaseURL
	reqURL.Path += "add-record.json"

	q := reqURL.Query()
	q.Add("domain-name", zoneName)
	q.Add("host", host)
	q.Add("record", value)
	q.Add("ttl", strconv.Itoa(ttlRounder(ttl)))
	q.Add("record-type", "TXT")
	reqURL.RawQuery = q.Encode()

	raw, err := c.doRequest(http.MethodPost, &reqURL)
	if err != nil {
		return err
	}

	resp := apiResponse{}
	if err = json.Unmarshal(raw, &resp); err != nil {
		return fmt.Errorf("apiResponse unmarshaling error: %v: %s", err, string(raw))
	}

	if resp.Status != "Success" {
		return fmt.Errorf("fail to add TXT record: %s %s", resp.Status, resp.StatusDescription)
	}

	return nil
}
985688
func NewTxPreparedPool(capacity int) *TxPreparedPool {
	if capacity < 0 {
		// If capacity is 0 all prepares will fail.
		capacity = 0
	}
	return &TxPreparedPool{
		conns:    make(map[string]*TxConnection, capacity),
		reserved: make(map[string]error),
		capacity: capacity,
	}
}
945980
func (eng *engine) AddRecord(key string, member interface{}, unixstamp int64) error {
	s := eng.newSession()
	defer s.Close()
	action, err := s.addRecord(key, member, unixstamp)
	if err != nil {
		return s.catch("AddRecord: "+action, err)
	}
	return nil
}
97938
func NewTxIndex(store dbm.DB, options ...func(*TxIndex)) *TxIndex {
	txi := &TxIndex{store: store, tagsToIndex: make([]string, 0), indexAllTags: false}
	for _, o := range options {
		o(txi)
	}
	return txi
}
655046
func (idx *index) AppendRecord(compSize, rawSize int64, typ int) bool {
	if rawSize < 0 || compSize < 0 {
		return false // Invalid size
	}

	lastRec := idx.LastRecord()
	rec := record{
		CompOffset: lastRec.CompOffset + compSize,
		RawOffset:  lastRec.RawOffset + rawSize,
		Type:       typ,
	}
	if rec.CompOffset < lastRec.CompOffset || rec.RawOffset < lastRec.RawOffset {
		return false // Overflow detected
	}
	idx.Records = append(idx.Records, rec)
	return true
}
704999
func (t *tableCommon) UpdateRecord(ctx sessionctx.Context, h int64, oldData, newData []types.Datum, touched []bool) error {
	txn, err := ctx.Txn(true)
	if err != nil {
		return err
	}

	// TODO: reuse bs, like AddRecord does.
	bs := kv.NewBufferStore(txn, kv.DefaultTxnMembufCap)

	// rebuild index
	err = t.rebuildIndices(ctx, bs, h, touched, oldData, newData)
	if err != nil {
		return err
	}
	numColsCap := len(newData) + 1 // +1 for the extra handle column that we may need to append.

	var colIDs, binlogColIDs []int64
	var row, binlogOldRow, binlogNewRow []types.Datum
	colIDs = make([]int64, 0, numColsCap)
	row = make([]types.Datum, 0, numColsCap)
	if shouldWriteBinlog(ctx) {
		binlogColIDs = make([]int64, 0, numColsCap)
		binlogOldRow = make([]types.Datum, 0, numColsCap)
		binlogNewRow = make([]types.Datum, 0, numColsCap)
	}

	for _, col := range t.WritableCols() {
		var value types.Datum
		if col.State != model.StatePublic {
			// If col is in write only or write reorganization state we should keep the oldData.
			// Because the oldData must be the orignal data(it's changed by other TiDBs.) or the orignal default value.
			// TODO: Use newData directly.
			value = oldData[col.Offset]
		} else {
			value = newData[col.Offset]
		}
		if !t.canSkip(col, value) {
			colIDs = append(colIDs, col.ID)
			row = append(row, value)
		}
		if shouldWriteBinlog(ctx) && !t.canSkipUpdateBinlog(col, value) {
			binlogColIDs = append(binlogColIDs, col.ID)
			binlogOldRow = append(binlogOldRow, oldData[col.Offset])
			binlogNewRow = append(binlogNewRow, value)
		}
	}

	key := t.RecordKey(h)
	value, err := tablecodec.EncodeRow(ctx.GetSessionVars().StmtCtx, row, colIDs, nil, nil)
	if err != nil {
		return err
	}
	if err = bs.Set(key, value); err != nil {
		return err
	}
	if err = bs.SaveTo(txn); err != nil {
		return err
	}
	ctx.StmtAddDirtyTableOP(table.DirtyTableDeleteRow, t.physicalTableID, h, nil)
	ctx.StmtAddDirtyTableOP(table.DirtyTableAddRow, t.physicalTableID, h, newData)
	if shouldWriteBinlog(ctx) {
		if !t.meta.PKIsHandle {
			binlogColIDs = append(binlogColIDs, model.ExtraHandleID)
			binlogOldRow = append(binlogOldRow, types.NewIntDatum(h))
			binlogNewRow = append(binlogNewRow, types.NewIntDatum(h))
		}
		err = t.addUpdateBinlog(ctx, binlogOldRow, binlogNewRow, binlogColIDs)
		if err != nil {
			return err
		}
	}
	colSize := make(map[int64]int64)
	for id, col := range t.Cols() {
		val := int64(len(newData[id].GetBytes()) - len(oldData[id].GetBytes()))
		if val != 0 {
			colSize[col.ID] = val
		}
	}
	ctx.GetSessionVars().TxnCtx.UpdateDeltaForTable(t.physicalTableID, 0, 1, colSize)
	return nil
}
376204
func NewTxnPool(env *lmdb.Env) *TxnPool {
	p := &TxnPool{
		env: env,
	}
	return p
}