Method executeInTransactionMode

backend/plugin/db/redshift/redshift.go:231–288

executeInTransactionMode executes statements within a single transaction

(ctx context.Context, commands []base.Statement, opts db.ExecuteOptions)

// executeInTransactionMode executes statements within a single transaction
func (d *Driver) executeInTransactionMode(ctx context.Context, commands []base.Statement, opts db.ExecuteOptions) (int64, error) {
	totalRowsAffected := int64(0)
	if len(commands) == 0 {
		return 0, nil
	}

	tx, err := d.db.BeginTx(ctx, nil)
	if err != nil {
		opts.LogTransactionControl(storepb.TaskRunLog_TransactionControl_BEGIN, err.Error())
		return 0, err
	}
	opts.LogTransactionControl(storepb.TaskRunLog_TransactionControl_BEGIN, "")

	committed := false
	defer func() {
		err := tx.Rollback()
		if committed {
			return
		}
		var rerr string
		if err != nil && !errors.Is(err, sql.ErrTxDone) {
			rerr = err.Error()
			slog.Debug("failed to rollback transaction", log.BBError(err))
		}
		opts.LogTransactionControl(storepb.TaskRunLog_TransactionControl_ROLLBACK, rerr)
	}()

	for i, command := range commands {
		opts.LogCommandExecute(command.Range, command.Text)
		// Log the query statement in char code to see if there are some control characters that cause issues.
		var charCode []rune
		for _, r := range command.Text {
			charCode = append(charCode, r)
		}
		slog.Debug("executing command", slog.Any("command", charCode), slog.Int("index", i))
		sqlResult, err := tx.ExecContext(ctx, command.Text)
		if err != nil {
			opts.LogCommandResponse(0, nil, err.Error())
			return 0, err
		}
		rowsAffected, err := sqlResult.RowsAffected()
		if err != nil {
			// Since we cannot differentiate DDL and DML yet, we have to ignore the error.
			slog.Debug("rowsAffected returns error", log.BBError(err))
		}
		opts.LogCommandResponse(rowsAffected, nil, "")
		totalRowsAffected += rowsAffected
	}

	if err := tx.Commit(); err != nil {
		opts.LogTransactionControl(storepb.TaskRunLog_TransactionControl_COMMIT, err.Error())
		return 0, err
	}
	opts.LogTransactionControl(storepb.TaskRunLog_TransactionControl_COMMIT, "")
	committed = true

	return totalRowsAffected, nil
}