Method CopyFrom

pgconn/pgconn.go:1438–1566

CopyFrom executes the copy command sql and copies all of r to the PostgreSQL server. Note: context cancellation will only interrupt operations on the underlying PostgreSQL network connection. Reads on r could still block.

(ctx context.Context, r io.Reader, sql string)

// Note: context cancellation will only interrupt operations on the underlying PostgreSQL network connection. Reads on r
// could still block.
func (pgConn *PgConn) CopyFrom(ctx context.Context, r io.Reader, sql string) (CommandTag, error) {
	if err := pgConn.lock(); err != nil {
		return CommandTag{}, err
	}
	defer pgConn.unlock()

	if ctx != context.Background() {
		select {
		case <-ctx.Done():
			return CommandTag{}, newContextAlreadyDoneError(ctx)
		default:
		}
		pgConn.contextWatcher.Watch(ctx)
		defer pgConn.contextWatcher.Unwatch()
	}

	// Send copy from query
	pgConn.frontend.SendQuery(&pgproto3.Query{String: sql})
	err := pgConn.flushWithPotentialWriteReadDeadlock()
	if err != nil {
		pgConn.asyncClose()
		return CommandTag{}, err
	}

	// Send copy data
	abortCopyChan := make(chan struct{})
	copyErrChan := make(chan error, 1)
	signalMessageChan := pgConn.signalMessage()
	var wg sync.WaitGroup
	wg.Go(func() {
		buf := iobufpool.Get(65536)
		defer iobufpool.Put(buf)
		(*buf)[0] = 'd'

		for {
			n, readErr := r.Read((*buf)[5:cap(*buf)])
			if n > 0 {
				*buf = (*buf)[0 : n+5]
				pgio.SetInt32((*buf)[1:], int32(n+4))

				writeErr := pgConn.frontend.SendUnbufferedEncodedCopyData(*buf)
				if writeErr != nil {
					// Write errors are always fatal, but we can't use asyncClose because we are in a different goroutine. Not
					// setting pgConn.status or closing pgConn.cleanupDone for the same reason.
					pgConn.conn.Close()

					copyErrChan <- writeErr
					return
				}
			}
			if readErr != nil {
				copyErrChan <- readErr
				return
			}

			select {
			case <-abortCopyChan:
				return