// in a transaction, truncate, create or update, and then copy from the s3 data file or manifest
// yell loudly if there is anything different in the target table compared to config (different distkey, etc)
8872
func runCopy(
	db *redshift.Redshift, inputConf s3filepath.S3File, inputTable redshift.Table, targetTable *redshift.Table,
	truncate, gzip bool, delimiter, timeGranularity, targetTimeZone, streamStart, streamEnd string,
) error {
	tx, err := db.Begin()
	if err != nil {
		return err
	}

	// TRUNCATE for dimension tables, but not fact tables
	if truncate && targetTable != nil {
		log.Println("truncating table!")
		if err := db.Truncate(tx, inputConf.Schema, inputTable.Name); err != nil {
			return fmt.Errorf("err running truncate table: %s", err)
		}
	}
	if targetTable == nil {
		if err := db.CreateTable(tx, inputTable); err != nil {
			return fmt.Errorf("err running create table: %s", err)
		}
	} else {
		var start, end time.Time
		var err error
		if timeGranularity == "stream" {
			start, err = time.Parse("2006-01-02T15:04:05", streamStart)
			if err != nil {
				return err
			}
			end, err = time.Parse("2006-01-02T15:04:05", streamEnd)
			if err != nil {
				return err
			}
		} else {
			start, end = startEndFromGranularity(inputConf.DataDate, timeGranularity, targetTimeZone)
		}
		// To prevent duplicates, clear away any existing data within a certain time range as the data date
		// (that is, sharing the same data date up to a certain time granularity)
		if err := db.TruncateInTimeRange(tx, inputConf.Schema, inputTable.Name, inputTable.Meta.DataDateColumn, start, end); err != nil {
			return fmt.Errorf("err truncating data for data refresh: %s", err)
		}

		if err := db.UpdateTable(tx, inputTable, *targetTable); err != nil {
			return fmt.Errorf("err running update table: %s", err)
		}
	}

	// COPY direct into it, ok to do since we're in a transaction
	// can't switch on file ending as manifest files b/c
	// manifest files obscure the underlying file types
	// instead just pass the delimiter along even if it's null
	if err := db.Copy(tx, inputConf, delimiter, true, gzip); err != nil {
		return fmt.Errorf("err running copy: %s", err)
	}

	if err := tx.Commit(); err != nil {
		return fmt.Errorf("err committing transaction: %s", err)
	}

	if truncate {
		// If we've truncated the table we should run vacuum to clear out the old data
		// Only one vacuum can be run at a time, so we're going to throw this over the wall to
		// redshift-vacuum and use gearman-admin as a queueing service.
		if len(gearmanAdminURL) == 0 {
			log.Fatalf("Unable to post vacuum job to %s", vacuumWorker)
		} else {
			log.Println("Submitting job to Gearman admin")

			// N.B. We need to pass backslashes to escape the quotation marks as required
			// by Golang's os.Args for command line arguments
			payload, err := json.Marshal(map[string]string{
				"analyze": inputConf.Schema + `."` + inputTable.Name + `"`,
			})
			if err != nil {
				log.Fatalf("Error creating new payload: %s", err)
			}

			client := &http.Client{}
			endpoint := gearmanAdminURL + fmt.Sprintf("/%s", vacuumWorker)
			req, err := http.NewRequest("POST", endpoint, bytes.NewReader(payload))
			if err != nil {
				log.Fatalf("Error creating new request: %s", err)
			}
			req.Header.Add("Content-Type", "text/plain")
			_, err = client.Do(req)
			if err != nil {
				log.Fatalf("Error submitting job: %s", err)
			}
		}
	}
	return nil
}
311531
func (b *Builder) Truncate(n int) {
	b.lastRead = opInvalid
	switch {
	case n < 0 || n > b.Len():
		// Originall this was a panic, but I hate panicing (hehe),
		// so instead let's just ignore this issue.
		// panic("bytes.Buffer: truncation out of range")
		return
	case n == 0:
		// Reuse buffer space.
		b.off = 0
	}
	b.buf = b.buf[0 : b.off+n]
}
8112
func (d *Datastore) NewTransaction(readOnly bool) (ds.Txn, error) {
	d.closeLk.RLock()
	defer d.closeLk.RUnlock()
	if d.closed {
		return nil, ErrClosed
	}

	return &txn{d, d.DB.NewTransaction(!readOnly), false}, nil
}
115726
func (tm *transactionManager) announcePeer(
	no *node, infoHash string, impliedPort, port int, token string) {

	tm.sendQuery(no, announcePeerType, map[string]interface{}{
		"id":           tm.dht.id(no.id.RawString()),
		"info_hash":    infoHash,
		"implied_port": impliedPort,
		"port":         port,
		"token":        token,
	})
}
924957
func (m *Migrator) Exec(dir MigrationDirection, migrations ...Migration) error {
	m.Lock()
	defer m.Unlock()

	_, err := m.db.Exec(fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (version integer primary key not null)", m.table()))
	if err != nil {
		return err
	}

	var tx *sql.Tx
	if m.TransactionMode == SingleTransaction {
		tx, err = m.db.Begin()
		if err != nil {
			return err
		}
	}

	for _, migration := range sortMigrations(dir, migrations) {
		if m.TransactionMode == IndividualTransactions {
			tx, err = m.db.Begin()
			if err != nil {
				return err
			}
		}

		if err := m.runMigration(tx, dir, migration); err != nil {
			tx.Rollback()
			return err
		}

		if m.TransactionMode == IndividualTransactions {
			if err := tx.Commit(); err != nil {
				return err
			}
		}
	}

	if m.TransactionMode == SingleTransaction {
		if err := tx.Commit(); err != nil {
			return err
		}
	}

	return nil
}
380593
func (rp *Reprovider) Trigger(ctx context.Context) error {
	progressCtx, done := context.WithCancel(ctx)

	var err error
	df := func(e error) {
		err = e
		done()
	}

	select {
	case <-rp.closing:
		return errors.New("reprovider is closed")
	case <-rp.ctx.Done():
		return rp.ctx.Err()
	case <-ctx.Done():
		return ctx.Err()
	case rp.trigger <- df:
		<-progressCtx.Done()
		return err
	}
}
729072
func (sm *SourceMgr) doRelease() {
	// Send the signal to the supervisor to cancel all running calls
	sm.cancelAll()
	sm.suprvsr.wait()

	// Close the file handle for the lock file and remove it from disk
	sm.lf.Close()
	os.Remove(filepath.Join(sm.cachedir, "sm.lock"))

	// Close the qch, if non-nil, so the signal handlers run out. This will
	// also deregister the sig channel, if any has been set up.
	if sm.qch != nil {
		close(sm.qch)
	}
}
796398
func (dt *DirtyTable) TruncateTable() {
	dt.addedRows = make(map[int64]struct{})
	dt.truncated = true
}
392691
func Transactional(tm TransactionManager, todo func(f TransactionalResources) error) error {
	var tx Transaction
	var err error
	if tx, err = tm.BeginTransaction(); err != nil {
		log.Error(nil, map[string]interface{}{
			"err": err,
		}, "database BeginTransaction failed!")

		return errors.WithStack(err)
	}

	return func() error {
		errorChan := make(chan error, 1)
		txTimeout := time.After(databaseTransactionTimeout)

		go func(f TransactionalResources) {
			defer func() {
				if err := recover(); err != nil {
					errorChan <- errors.New(fmt.Sprintf("Unknown error: %v", err))
				}
			}()
			errorChan <- todo(tx)
		}(tx)

		select {
		case err := <-errorChan:
			if err != nil {
				log.Debug(nil, nil, "Rolling back the transaction...")
				tx.Rollback()
				log.Error(nil, map[string]interface{}{
					"err": err,
				}, "database transaction failed!")
				return errors.WithStack(err)
			}

			tx.Commit()
			log.Debug(nil, nil, "Commit the transaction!")
			return nil
		case <-txTimeout:
			log.Debug(nil, nil, "Rolling back the transaction...")
			tx.Rollback()
			log.Error(nil, nil, "database transaction timeout!")
			return errors.New("database transaction timeout!")
		}
	}()
}
375254
func (sc *StatementContext) HandleTruncate(err error) error {
	// TODO: At present we have not checked whether the error can be ignored or treated as warning.
	// We will do that later, and then append WarnDataTruncated instead of the error itself.
	if err == nil {
		return nil
	}
	if sc.IgnoreTruncate {
		return nil
	}
	if sc.TruncateAsWarning {
		sc.AppendWarning(err)
		return nil
	}
	return err
}
564840
func (db *Database) ValidateTransaction(selector *string, tx *core.SendTxArgs) (*core.ValidationMessages, error) {
	messages := new(core.ValidationMessages)

	// Prevent accidental erroneous usage of both 'input' and 'data' (show stopper)
	if tx.Data != nil && tx.Input != nil && !bytes.Equal(*tx.Data, *tx.Input) {
		return nil, errors.New(`ambiguous request: both "data" and "input" are set and are not identical`)
	}
	// Place data on 'data', and nil 'input'
	var data []byte
	if tx.Input != nil {
		tx.Data = tx.Input
		tx.Input = nil
	}
	if tx.Data != nil {
		data = *tx.Data
	}
	// Contract creation doesn't validate call data, handle first
	if tx.To == nil {
		// Contract creation should contain sufficient data to deploy a contract. A
		// typical error is omitting sender due to some quirk in the javascript call
		// e.g. https://github.com/ethereum/go-ethereum/issues/16106.
		if len(data) == 0 {
			// Prevent sending ether into black hole (show stopper)
			if tx.Value.ToInt().Cmp(big.NewInt(0)) > 0 {
				return nil, errors.New("tx will create contract with value but empty code")
			}
			// No value submitted at least, critically Warn, but don't blow up
			messages.Crit("Transaction will create contract with empty code")
		} else if len(data) < 40 { // arbitrary heuristic limit
			messages.Warn(fmt.Sprintf("Transaction will will create contract, but payload is suspiciously small (%d bytes)", len(data)))
		}
		// Method selector should be nil for contract creation
		if selector != nil {
			messages.Warn("Transaction will create contract, but method selector supplied, indicating intent to call a method")
		}
		return messages, nil
	}
	// Not a contract creation, validate as a plain transaction
	if !tx.To.ValidChecksum() {
		messages.Warn("Invalid checksum on recipient address")
	}
	if bytes.Equal(tx.To.Address().Bytes(), common.Address{}.Bytes()) {
		messages.Crit("Transaction recipient is the zero address")
	}
	// Semantic fields validated, try to make heads or tails of the call data
	db.validateCallData(selector, data, messages)
	return messages, nil
}
549046
func (t *Transaction) Add(cmd ...*Command) {
	t.commands = append(t.commands, cmd...)
}
792947
func (p *peer) AsyncSendTransactions(txs []*types.Transaction) {
	select {
	case p.queuedTxs <- txs:
		for _, tx := range txs {
			p.knownTxs.Add(tx.Hash())
		}
	default:
		p.Log().Debug("Dropping transaction propagation", "count", len(txs))
	}
}
946719
func (db *schemaUpdater) updateSchema6to7() {
	t := db.newReadWriteTransaction()
	defer t.close()

	var gk []byte
	var nk []byte

	for _, folderStr := range db.ListFolders() {
		folder := []byte(folderStr)
		db.withNeedLocal(folder, false, func(f FileIntf) bool {
			name := []byte(f.FileName())
			global := f.(protocol.FileInfo)
			gk = db.keyer.GenerateGlobalVersionKey(gk, folder, name)
			svl, err := t.Get(gk, nil)
			if err != nil {
				// If there is no global list, we hardly need it.
				t.Delete(t.keyer.GenerateNeedFileKey(nk, folder, name))
				return true
			}
			var fl VersionList
			err = fl.Unmarshal(svl)
			if err != nil {
				// This can't happen, but it's ignored everywhere else too,
				// so lets not act on it.
				return true
			}
			if localFV, haveLocalFV := fl.Get(protocol.LocalDeviceID[:]); !need(global, haveLocalFV, localFV.Version) {
				t.Delete(t.keyer.GenerateNeedFileKey(nk, folder, name))
			}
			return true
		})
	}
}
604330
func (t *explainTablet) CreateTransaction(ctx context.Context, target *querypb.Target, dtid string, participants []*querypb.Target) (err error) {
	t.mu.Lock()
	t.currentTime = batchTime.Wait()
	t.mu.Unlock()
	return t.tsv.CreateTransaction(ctx, target, dtid, participants)
}
83694
func (t *Table) Truncate() {
	t.mu.Lock()
	defer t.mu.Unlock()
	t.Data = nil
}
587690
func (txn *Txn) Abort() {
	if txn.managed {
		panic("managed transaction cannot be aborted directly")
	}

	runtime.SetFinalizer(txn, nil)
	txn.abort()
}