Repository: jackc/pgx
Pull request #1412 (2022-12-05 17:52:17)

Add QueryExecMode to CopyFrom
### Use DefaultQueryExecMode in CopyFrom 
 
 CopyFrom had to create a prepared statement to get the OIDs of the data
types that were going to be copied into the table. Every COPY operation
required an extra round trips to retrieve the type information. There
was no way to customize this behavior.

By leveraging the QueryExecMode feature, like in `Conn.Query`, users can
specify if they want to cache the prepared statements, execute
them on every request (like the old behavior), or bypass the prepared
statement relying on the pgtype.Map to get the type information.

The `QueryExecMode` behave exactly like in `Conn.Query` in the way the
data type OIDs are fetched, meaning that:

- `QueryExecModeCacheStatement`: caches the statement.
- `QueryExecModeCacheDescribe`: caches the statement and assumes they do
  not change.
- `QueryExecModeDescribeExec`: gets the statement description on every
  execution. This is like to the old behavior of `CopyFrom`.
- `QueryExecModeExec` and `QueryExecModeSimpleProtocol`: maintain the
  same behavior as before, which is the same as `QueryExecModeDescribeExec`.
  It will keep getting the statement description on every execution


The `QueryExecMode` can only be set via
`ConnConfig.DefaultQueryExecMode`, unlike `Conn.Query` there's no
support for specifying the `QueryExecMode` via optional arguments
in the function signature.

cristaloleg: Hi, thanks for the PR, are there any performance improvements in benchmarks?
jackc: I like the direction this is going, but I am concerned about the change in the `CopyFrom()` signature.  It won't directly break any calling code, but it does break any interfaces defined with `Conn`, `Tx`, etc. in mind. e.g.

```go
type DB interface {
	Query(ctx context.Context, sql string, optionsAndArgs ...interface{}) (pgx.Rows, error)
	QueryRow(ctx context.Context, sql string, optionsAndArgs ...interface{}) pgx.Row
	CopyFrom(ctx context.Context, tableName pgx.Identifier, columnNames []string, rowSrc pgx.CopyFromSource) (int64, error)
}
```

An application or library using interface `DB` would break.

Would it be sufficient to use the default query exec mode without the ability to override? AFAIK, overriding the mode per query is extremely rare. I only included it on `Query`, `QueryRow`, and `Exec` for completeness and testing.

If it really needs to be controlled per call, perhaps a new optional interface could be created that `rowSrc` could implement. This interface could have a method to get the query exec mode. A wrapper struct could be created that wrapped the original `CopyFromSource` and added the new method.
alejandrodnm: @jackc I really need to override it to avoid the the prepare statement and the cache 😅 

We are using pgx in [timescale/promscale](https://github.com/timescale/promscale) to ingest metrics and traces, we insert the data using COPY. Basically we have a table per metric, and our COPY operations are always of the same 3 columns. 

We want to get rid of the extra roundtrip per copy. In our use case the cache would be a lot of unnecessary memory since we are always sending (time, int64 and float64), which the Type map resolves correctly.

> If it really needs to be controlled per call, perhaps a new optional interface could be created that rowSrc could implement. This interface could have a method to get the query exec mode. A wrapper struct could be created that wrapped the original CopyFromSource and added the new method.

I'll try to give ☝️ a shot. 
alejandrodnm: I'm going to try our use case with allowing setting the Exec mode only via `ConnConfig.DefaultQueryExecMode`.

I reverted the signature changes to `CopyFrom`.

Patch:
diff --git a/.gitignore b/.gitignore
index 348e014f..a2ebbe9c 100644
--- a/.gitignore
+++ b/.gitignore
@@ -23,3 +23,5 @@ _testmain.go
 
 .envrc
 /.testdb
+
+.DS_Store
diff --git a/conn.go b/conn.go
index b6c1ab8f..b4bbdb5b 100644
--- a/conn.go
+++ b/conn.go
@@ -721,43 +721,10 @@ optionLoop:
 	sd, explicitPreparedStatement := c.preparedStatements[sql]
 	if sd != nil || mode == QueryExecModeCacheStatement || mode == QueryExecModeCacheDescribe || mode == QueryExecModeDescribeExec {
 		if sd == nil {
-			switch mode {
-			case QueryExecModeCacheStatement:
-				if c.statementCache == nil {
-					err = errDisabledStatementCache
-					rows.fatal(err)
-					return rows, err
-				}
-				sd = c.statementCache.Get(sql)
-				if sd == nil {
-					sd, err = c.Prepare(ctx, stmtcache.NextStatementName(), sql)
-					if err != nil {
-						rows.fatal(err)
-						return rows, err
-					}
-					c.statementCache.Put(sd)
-				}
-			case QueryExecModeCacheDescribe:
-				if c.descriptionCache == nil {
-					err = errDisabledDescriptionCache
-					rows.fatal(err)
-					return rows, err
-				}
-				sd = c.descriptionCache.Get(sql)
-				if sd == nil {
-					sd, err = c.Prepare(ctx, "", sql)
-					if err != nil {
-						rows.fatal(err)
-						return rows, err
-					}
-					c.descriptionCache.Put(sd)
-				}
-			case QueryExecModeDescribeExec:
-				sd, err = c.Prepare(ctx, "", sql)
-				if err != nil {
-					rows.fatal(err)
-					return rows, err
-				}
+			sd, err = c.getStatementDescription(ctx, mode, sql)
+			if err != nil {
+				rows.fatal(err)
+				return rows, err
 			}
 		}
 
@@ -827,6 +794,48 @@ optionLoop:
 	return rows, rows.err
 }
 
+// getStatementDescription returns the statement description of the sql query
+// according to the given mode.
+//
+// If the mode is one that doesn't require to know the param and result OIDs
+// then nil is returned without error.
+func (c *Conn) getStatementDescription(
+	ctx context.Context,
+	mode QueryExecMode,
+	sql string,
+) (sd *pgconn.StatementDescription, err error) {
+
+	switch mode {
+	case QueryExecModeCacheStatement:
+		if c.statementCache == nil {
+			return nil, errDisabledStatementCache
+		}
+		sd = c.statementCache.Get(sql)
+		if sd == nil {
+			sd, err = c.Prepare(ctx, stmtcache.NextStatementName(), sql)
+			if err != nil {
+				return nil, err
+			}
+			c.statementCache.Put(sd)
+		}
+	case QueryExecModeCacheDescribe:
+		if c.descriptionCache == nil {
+			return nil, errDisabledDescriptionCache
+		}
+		sd = c.descriptionCache.Get(sql)
+		if sd == nil {
+			sd, err = c.Prepare(ctx, "", sql)
+			if err != nil {
+				return nil, err
+			}
+			c.descriptionCache.Put(sd)
+		}
+	case QueryExecModeDescribeExec:
+		return c.Prepare(ctx, "", sql)
+	}
+	return sd, err
+}
+
 // QueryRow is a convenience wrapper over Query. Any error that occurs while
 // querying is deferred until calling Scan on the returned Row. That Row will
 // error with ErrNoRows if no rows are returned.
diff --git a/copy_from.go b/copy_from.go
index 41acafdc..40232bdf 100644
--- a/copy_from.go
+++ b/copy_from.go
@@ -85,6 +85,7 @@ type copyFrom struct {
 	columnNames   []string
 	rowSrc        CopyFromSource
 	readerErrChan chan error
+	mode          QueryExecMode
 }
 
 func (ct *copyFrom) run(ctx context.Context) (int64, error) {
@@ -105,9 +106,29 @@ func (ct *copyFrom) run(ctx context.Context) (int64, error) {
 	}
 	quotedColumnNames := cbuf.String()
 
-	sd, err := ct.conn.Prepare(ctx, "", fmt.Sprintf("select %s from %s", quotedColumnNames, quotedTableName))
-	if err != nil {
-		return 0, err
+	var sd *pgconn.StatementDescription
+	switch ct.mode {
+	case QueryExecModeExec, QueryExecModeSimpleProtocol:
+		// These modes don't support the binary format. Before the inclusion of the
+		// QueryExecModes, Conn.Prepare was called on every COPY operation to get
+		// the OIDs. These prepared statements were not cached.
+		//
+		// Since that's the same behavior provided by QueryExecModeDescribeExec,
+		// we'll default to that mode.
+		ct.mode = QueryExecModeDescribeExec
+		fallthrough
+	case QueryExecModeCacheStatement, QueryExecModeCacheDescribe, QueryExecModeDescribeExec:
+		var err error
+		sd, err = ct.conn.getStatementDescription(
+			ctx,
+			ct.mode,
+			fmt.Sprintf("select %s from %s", quotedColumnNames, quotedTableName),
+		)
+		if err != nil {
+			return 0, fmt.Errorf("statement description failed: %w", err)
+		}
+	default:
+		return 0, fmt.Errorf("unknown QueryExecMode: %v", ct.mode)
 	}
 
 	r, w := io.Pipe()
@@ -208,6 +229,7 @@ func (c *Conn) CopyFrom(ctx context.Context, tableName Identifier, columnNames [
 		columnNames:   columnNames,
 		rowSrc:        rowSrc,
 		readerErrChan: make(chan error),
+		mode:          c.config.DefaultQueryExecMode,
 	}
 
 	return ct.run(ctx)
diff --git a/doc.go b/doc.go
index fb271874..0db8cbb1 100644
--- a/doc.go
+++ b/doc.go
@@ -69,8 +69,9 @@ Use Exec to execute a query that does not return a result set.
 
 PostgreSQL Data Types
 
-The package pgtype provides extensive and customizable support for converting Go values to and from PostgreSQL values
-including array and composite types. See that package's documentation for details.
+pgx uses the pgtype package to converting Go values to and from PostgreSQL values. It supports many PostgreSQL types
+directly and is customizable and extendable. User defined data types such as enums, domains,  and composite types may
+require type registration. See that package's documentation for details.
 
 Transactions
 
diff --git a/pgtype/bytea.go b/pgtype/bytea.go
index 2e067672..a247705e 100644
--- a/pgtype/bytea.go
+++ b/pgtype/bytea.go
@@ -238,7 +238,7 @@ func decodeHexBytea(src []byte) ([]byte, error) {
 }
 
 func (c ByteaCodec) DecodeDatabaseSQLValue(m *Map, oid uint32, format int16, src []byte) (driver.Value, error) {
-	return codecDecodeToTextFormat(c, m, oid, format, src)
+	return c.DecodeValue(m, oid, format, src)
 }
 
 func (c ByteaCodec) DecodeValue(m *Map, oid uint32, format int16, src []byte) (any, error) {
diff --git a/pgtype/doc.go b/pgtype/doc.go
index 834aa0b6..6612c896 100644
--- a/pgtype/doc.go
+++ b/pgtype/doc.go
@@ -57,27 +57,7 @@ JSON Support
 
 pgtype automatically marshals and unmarshals data from json and jsonb PostgreSQL types.
 
-Array Support
-
-ArrayCodec implements support for arrays. If pgtype supports type T then it can easily support []T by registering an
-ArrayCodec for the appropriate PostgreSQL OID. In addition, Array[T] type can support multi-dimensional arrays.
-
-Composite Support
-
-CompositeCodec implements support for PostgreSQL composite types. Go structs can be scanned into if the public fields of
-the struct are in the exact order and type of the PostgreSQL type or by implementing CompositeIndexScanner and
-CompositeIndexGetter.
-
-Enum Support
-
-PostgreSQL enums can usually be treated as text. However, EnumCodec implements support for interning strings which can
-reduce memory usage.
-
-Array, Composite, and Enum Type Registration
-
-Array, composite, and enum types can be easily registered from a pgx.Conn with the LoadType method.
-
-Extending Existing Type Support
+Extending Existing PostgreSQL Type Support
 
 Generally, all Codecs will support interfaces that can be implemented to enable scanning and encoding. For example,
 PointCodec can use any Go type that implements the PointScanner and PointValuer interfaces. So rather than use
@@ -90,11 +70,58 @@ pgx support such as github.com/shopspring/decimal. These types can be registered
 logic. See https://github.com/jackc/pgx-shopspring-decimal and https://github.com/jackc/pgx-gofrs-uuid for a example
 integrations.
 
-Entirely New Type Support
+New PostgreSQL Type Support
+
+pgtype uses the PostgreSQL OID to determine how to encode or decode a value. pgtype supports array, composite, domain,
+and enum types. However, any type created in PostgreSQL with CREATE TYPE will receive a new OID. This means that the OID
+of each new PostgreSQL type must be registered for pgtype to handle values of that type with the correct Codec.
+
+The pgx.Conn LoadType method can return a *Type for array, composite, domain, and enum types by inspecting the database
+metadata. This *Type can then be registered with Map.RegisterType.
+
+For example, the following function could be called after a connection is established:
+
+    func RegisterDataTypes(ctx context.Context, conn *pgx.Conn) error {
+      dataTypeNames := []string{
+        "foo",
+        "_foo",
+        "bar",
+        "_bar",
+      }
+
+      for _, typeName := range dataTypeNames {
+        dataType, err := conn.LoadType(ctx, typeName)
+        if err != nil {
+          return err
+        }
+        conn.TypeMap().RegisterType(dataType)
+      }
+
+      return nil
+    }
+
+A type cannot be registered unless all types it depends on are already registered. e.g. An array type cannot be
+registered until its element type is registered.
+
+ArrayCodec implements support for arrays. If pgtype supports type T then it can easily support []T by registering an
+ArrayCodec for the appropriate PostgreSQL OID. In addition, Array[T] type can support multi-dimensional arrays.
+
+CompositeCodec implements support for PostgreSQL composite types. Go structs can be scanned into if the public fields of
+the struct are in the exact order and type of the PostgreSQL type or by implementing CompositeIndexScanner and
+CompositeIndexGetter.
+
+Domain types are treated as their underlying type if the underlying type and the domain type are registered.
+
+PostgreSQL enums can usually be treated as text. However, EnumCodec implements support for interning strings which can
+reduce memory usage.
+
+While pgtype will often still work with unregistered types it is highly recommended that all types be registered due to
+an improvement in performance and the elimination of certain edge cases.
 
-If the PostgreSQL type is not already supported then an OID / Codec mapping can be registered with Map.RegisterType.
-There is no difference between a Codec defined and registered by the application and a Codec built in to pgtype. See any
-of the Codecs in pgtype for Codec examples and for examples of type registration.
+If an entirely new PostgreSQL type (e.g. PostGIS types) is used then the application or a library can create a new
+Codec. Then the OID / Codec mapping can be registered with Map.RegisterType. There is no difference between a Codec
+defined and registered by the application and a Codec built in to pgtype. See any of the Codecs in pgtype for Codec
+examples and for examples of type registration.
 
 Encoding Unknown Types
 
diff --git a/pgtype/pgtype.go b/pgtype/pgtype.go
index 7cfa3d58..ad451678 100644
--- a/pgtype/pgtype.go
+++ b/pgtype/pgtype.go
@@ -1326,16 +1326,16 @@ func (m *Map) PlanEncode(oid uint32, format int16, value any) EncodePlan {
 	}
 
 	var dt *Type
-
-	if oid == 0 {
+	if dataType, ok := m.TypeForOID(oid); ok {
+		dt = dataType
+	} else {
+		// If no type for the OID was found, then either it is unknowable (e.g. the simple protocol) or it is an
+		// unregistered type. In either case try to find the type and OID that matches the value (e.g. a []byte would be
+		// registered to PostgreSQL bytea).
 		if dataType, ok := m.TypeForValue(value); ok {
 			dt = dataType
 			oid = dt.OID // Preserve assumed OID in case we are recursively called below.
 		}
-	} else {
-		if dataType, ok := m.TypeForOID(oid); ok {
-			dt = dataType
-		}
 	}
 
 	if dt != nil {


Test patch:
diff --git a/copy_from_test.go b/copy_from_test.go
index 49bfcb34..1342c14c 100644
--- a/copy_from_test.go
+++ b/copy_from_test.go
@@ -14,6 +14,129 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func TestConnCopyWithAllQueryExecModes(t *testing.T) {
+	for _, mode := range pgxtest.AllQueryExecModes {
+		t.Run(mode.String(), func(t *testing.T) {
+			t.Parallel()
+
+			cfg := mustParseConfig(t, os.Getenv("PGX_TEST_DATABASE"))
+			cfg.DefaultQueryExecMode = mode
+			conn := mustConnect(t, cfg)
+			defer closeConn(t, conn)
+
+			mustExec(t, conn, `create temporary table foo(
+			a int2,
+			b int4,
+			c int8,
+			d text,
+			e timestamptz
+		)`)
+
+			tzedTime := time.Date(2010, 2, 3, 4, 5, 6, 0, time.Local)
+
+			inputRows := [][]any{
+				{int16(0), int32(1), int64(2), "abc", tzedTime},
+				{nil, nil, nil, nil, nil},
+			}
+
+			copyCount, err := conn.CopyFrom(context.Background(), pgx.Identifier{"foo"}, []string{"a", "b", "c", "d", "e"}, pgx.CopyFromRows(inputRows))
+			if err != nil {
+				t.Errorf("Unexpected error for CopyFrom: %v", err)
+			}
+			if int(copyCount) != len(inputRows) {
+				t.Errorf("Expected CopyFrom to return %d copied rows, but got %d", len(inputRows), copyCount)
+			}
+
+			rows, err := conn.Query(context.Background(), "select * from foo")
+			if err != nil {
+				t.Errorf("Unexpected error for Query: %v", err)
+			}
+
+			var outputRows [][]any
+			for rows.Next() {
+				row, err := rows.Values()
+				if err != nil {
+					t.Errorf("Unexpected error for rows.Values(): %v", err)
+				}
+				outputRows = append(outputRows, row)
+			}
+
+			if rows.Err() != nil {
+				t.Errorf("Unexpected error for rows.Err(): %v", rows.Err())
+			}
+
+			if !reflect.DeepEqual(inputRows, outputRows) {
+				t.Errorf("Input rows and output rows do not equal: %v -> %v", inputRows, outputRows)
+			}
+
+			ensureConnValid(t, conn)
+		})
+	}
+}
+
+func TestConnCopyWithKnownOIDQueryExecModes(t *testing.T) {
+
+	for _, mode := range pgxtest.KnownOIDQueryExecModes {
+		t.Run(mode.String(), func(t *testing.T) {
+			t.Parallel()
+
+			cfg := mustParseConfig(t, os.Getenv("PGX_TEST_DATABASE"))
+			cfg.DefaultQueryExecMode = mode
+			conn := mustConnect(t, cfg)
+			defer closeConn(t, conn)
+
+			mustExec(t, conn, `create temporary table foo(
+			a int2,
+			b int4,
+			c int8,
+			d varchar,
+			e text,
+			f date,
+			g timestamptz
+		)`)
+
+			tzedTime := time.Date(2010, 2, 3, 4, 5, 6, 0, time.Local)
+
+			inputRows := [][]any{
+				{int16(0), int32(1), int64(2), "abc", "efg", time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC), tzedTime},
+				{nil, nil, nil, nil, nil, nil, nil},
+			}
+
+			copyCount, err := conn.CopyFrom(context.Background(), pgx.Identifier{"foo"}, []string{"a", "b", "c", "d", "e", "f", "g"}, pgx.CopyFromRows(inputRows))
+			if err != nil {
+				t.Errorf("Unexpected error for CopyFrom: %v", err)
+			}
+			if int(copyCount) != len(inputRows) {
+				t.Errorf("Expected CopyFrom to return %d copied rows, but got %d", len(inputRows), copyCount)
+			}
+
+			rows, err := conn.Query(context.Background(), "select * from foo")
+			if err != nil {
+				t.Errorf("Unexpected error for Query: %v", err)
+			}
+
+			var outputRows [][]any
+			for rows.Next() {
+				row, err := rows.Values()
+				if err != nil {
+					t.Errorf("Unexpected error for rows.Values(): %v", err)
+				}
+				outputRows = append(outputRows, row)
+			}
+
+			if rows.Err() != nil {
+				t.Errorf("Unexpected error for rows.Err(): %v", rows.Err())
+			}
+
+			if !reflect.DeepEqual(inputRows, outputRows) {
+				t.Errorf("Input rows and output rows do not equal: %v -> %v", inputRows, outputRows)
+			}
+
+			ensureConnValid(t, conn)
+		})
+	}
+}
+
 func TestConnCopyFromSmall(t *testing.T) {
 	t.Parallel()
 
@@ -220,7 +343,7 @@ func TestConnCopyFromEnum(t *testing.T) {
 		conn.TypeMap().RegisterType(typ)
 	}
 
-	_, err = tx.Exec(ctx, `create table foo(
+	_, err = tx.Exec(ctx, `create temporary table foo(
 		a text,
 		b color,
 		c fruit,
diff --git a/pgconn/pgconn_test.go b/pgconn/pgconn_test.go
index 9b0895d1..3b1ce436 100644
--- a/pgconn/pgconn_test.go
+++ b/pgconn/pgconn_test.go
@@ -18,6 +18,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/internal/pgio"
 	"github.com/jackc/pgx/v5/internal/pgmock"
 	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/jackc/pgx/v5/pgproto3"
@@ -1666,6 +1668,59 @@ func TestConnCopyFrom(t *testing.T) {
 	ensureConnValid(t, pgConn)
 }
 
+func TestConnCopyFromBinary(t *testing.T) {
+	t.Parallel()
+
+	pgConn, err := pgconn.Connect(context.Background(), os.Getenv("PGX_TEST_DATABASE"))
+	require.NoError(t, err)
+	defer closeConn(t, pgConn)
+
+	_, err = pgConn.Exec(context.Background(), `create temporary table foo(
+		a int4,
+		b varchar
+	)`).ReadAll()
+	require.NoError(t, err)
+
+	buf := []byte{}
+	buf = append(buf, "PGCOPY\n\377\r\n\000"...)
+	buf = pgio.AppendInt32(buf, 0)
+	buf = pgio.AppendInt32(buf, 0)
+
+	inputRows := [][][]byte{}
+	for i := 0; i < 1000; i++ {
+		// Number of elements in the tuple
+		buf = pgio.AppendInt16(buf, int16(2))
+		a := i
+
+		// Length of element for column `a int4`
+		buf = pgio.AppendInt32(buf, 4)
+		buf, err = pgtype.NewMap().Encode(pgtype.Int4OID, pgx.BinaryFormatCode, a, buf)
+		require.NoError(t, err)
+
+		b := "foo " + strconv.Itoa(a) + " bar"
+		lenB := int32(len([]byte(b)))
+		// Length of element for column `b varchar`
+		buf = pgio.AppendInt32(buf, lenB)
+		buf, err = pgtype.NewMap().Encode(pgtype.VarcharOID, pgx.BinaryFormatCode, b, buf)
+		require.NoError(t, err)
+
+		inputRows = append(inputRows, [][]byte{[]byte(strconv.Itoa(a)), []byte(b)})
+	}
+
+	srcBuf := &bytes.Buffer{}
+	srcBuf.Write(buf)
+	ct, err := pgConn.CopyFrom(context.Background(), srcBuf, "COPY foo (a, b) FROM STDIN BINARY;")
+	require.NoError(t, err)
+	assert.Equal(t, int64(len(inputRows)), ct.RowsAffected())
+
+	result := pgConn.ExecParams(context.Background(), "select * from foo", nil, nil, nil, nil).Read()
+	require.NoError(t, result.Err)
+
+	assert.Equal(t, inputRows, result.Rows)
+
+	ensureConnValid(t, pgConn)
+}
+
 func TestConnCopyFromCanceled(t *testing.T) {
 	t.Parallel()
 
diff --git a/pgtype/bytea_test.go b/pgtype/bytea_test.go
index a0d27369..336ad253 100644
--- a/pgtype/bytea_test.go
+++ b/pgtype/bytea_test.go
@@ -3,6 +3,7 @@ package pgtype_test
 import (
 	"bytes"
 	"context"
+	"fmt"
 	"testing"
 
 	pgx "github.com/jackc/pgx/v5"
@@ -114,3 +115,23 @@ func TestUndecodedBytes(t *testing.T) {
 		require.Equal(t, buf, []byte{0, 0, 0, 1})
 	})
 }
+
+func TestByteaCodecDecodeDatabaseSQLValue(t *testing.T) {
+	defaultConnTestRunner.RunTest(context.Background(), t, func(ctx context.Context, t testing.TB, conn *pgx.Conn) {
+		var buf []byte
+		err := conn.QueryRow(ctx, `select '\xa1b2c3d4'::bytea`).Scan(sqlScannerFunc(func(src any) error {
+			switch src := src.(type) {
+			case []byte:
+				buf = make([]byte, len(src))
+				copy(buf, src)
+				return nil
+			default:
+				return fmt.Errorf("expected []byte, got %T", src)
+			}
+		}))
+		require.NoError(t, err)
+
+		require.Len(t, buf, 4)
+		require.Equal(t, buf, []byte{0xa1, 0xb2, 0xc3, 0xd4})
+	})
+}
diff --git a/pgtype/pgtype_test.go b/pgtype/pgtype_test.go
index 1670db2c..c0cb3852 100644
--- a/pgtype/pgtype_test.go
+++ b/pgtype/pgtype_test.go
@@ -43,6 +43,10 @@ type _float32Slice []float32
 type _float64Slice []float64
 type _byteSlice []byte
 
+// unregisteredOID represents a actual type that is not registered. Cannot use 0 because that represents that the type
+// is not known (e.g. when using the simple protocol).
+const unregisteredOID = uint32(1)
+
 func mustParseCIDR(t testing.TB, s string) *net.IPNet {
 	_, ipnet, err := net.ParseCIDR(s)
 	if err != nil {
@@ -120,6 +124,20 @@ func skipPostgreSQLVersionLessThan(t testing.TB, minVersion int64) {
 	}
 }
 
+// sqlScannerFunc lets an arbitrary function be used as a sql.Scanner.
+type sqlScannerFunc func(src any) error
+
+func (f sqlScannerFunc) Scan(src any) error {
+	return f(src)
+}
+
+// driverValuerFunc lets an arbitrary function be used as a driver.Valuer.
+type driverValuerFunc func() (driver.Value, error)
+
+func (f driverValuerFunc) Value() (driver.Value, error) {
+	return f()
+}
+
 func TestMapScanNilIsNoOp(t *testing.T) {
 	m := pgtype.NewMap()
 
@@ -317,6 +335,36 @@ func TestMapEncodeBinaryFormatDatabaseValuerThatReturnsString(t *testing.T) {
 	require.Equal(t, []byte{0, 0, 0, 42}, buf)
 }
 
+// https://github.com/jackc/pgx/issues/1445
+func TestMapEncodeDatabaseValuerThatReturnsStringIntoUnregisteredTypeTextFormat(t *testing.T) {
+	m := pgtype.NewMap()
+	buf, err := m.Encode(unregisteredOID, pgtype.TextFormatCode, driverValuerFunc(func() (driver.Value, error) { return "foo", nil }), nil)
+	require.NoError(t, err)
+	require.Equal(t, []byte("foo"), buf)
+}
+
+// https://github.com/jackc/pgx/issues/1445
+func TestMapEncodeDatabaseValuerThatReturnsByteSliceIntoUnregisteredTypeTextFormat(t *testing.T) {
+	m := pgtype.NewMap()
+	buf, err := m.Encode(unregisteredOID, pgtype.TextFormatCode, driverValuerFunc(func() (driver.Value, error) { return []byte{0, 1, 2, 3}, nil }), nil)
+	require.NoError(t, err)
+	require.Equal(t, []byte(`\x00010203`), buf)
+}
+
+func TestMapEncodeStringIntoUnregisteredTypeTextFormat(t *testing.T) {
+	m := pgtype.NewMap()
+	buf, err := m.Encode(unregisteredOID, pgtype.TextFormatCode, "foo", nil)
+	require.NoError(t, err)
+	require.Equal(t, []byte("foo"), buf)
+}
+
+func TestMapEncodeByteSliceIntoUnregisteredTypeTextFormat(t *testing.T) {
+	m := pgtype.NewMap()
+	buf, err := m.Encode(unregisteredOID, pgtype.TextFormatCode, []byte{0, 1, 2, 3}, nil)
+	require.NoError(t, err)
+	require.Equal(t, []byte(`\x00010203`), buf)
+}
+
 // https://github.com/jackc/pgx/issues/1326
 func TestMapScanPointerToRenamedType(t *testing.T) {
 	srcBuf := []byte("foo")
diff --git a/query_test.go b/query_test.go
index c00cdec2..bf1a1fe8 100644
--- a/query_test.go
+++ b/query_test.go
@@ -190,7 +190,7 @@ func TestConnQueryValuesWhenUnableToDecode(t *testing.T) {
 	require.Equal(t, "({1},)", values[0])
 }
 
-func TestConnQueryValuesWithUnknownOID(t *testing.T) {
+func TestConnQueryValuesWithUnregisteredOID(t *testing.T) {
 	t.Parallel()
 
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
@@ -217,6 +217,24 @@ func TestConnQueryValuesWithUnknownOID(t *testing.T) {
 	require.Equal(t, "orange", values[0])
 }
 
+func TestConnQueryArgsAndScanWithUnregisteredOID(t *testing.T) {
+	t.Parallel()
+
+	pgxtest.RunWithQueryExecModes(context.Background(), t, defaultConnTestRunner, nil, func(ctx context.Context, t testing.TB, conn *pgx.Conn) {
+		tx, err := conn.Begin(ctx)
+		require.NoError(t, err)
+		defer tx.Rollback(ctx)
+
+		_, err = tx.Exec(ctx, "create type fruit as enum('orange', 'apple', 'pear')")
+		require.NoError(t, err)
+
+		var result string
+		err = conn.QueryRow(ctx, "select $1::fruit", "orange").Scan(&result)
+		require.NoError(t, err)
+		require.Equal(t, "orange", result)
+	})
+}
+
 // https://github.com/jackc/pgx/issues/478
 func TestConnQueryReadRowMultipleTimes(t *testing.T) {
 	t.Parallel()
diff --git a/tracelog/tracelog_test.go b/tracelog/tracelog_test.go
index 96b79de9..5458eaea 100644
--- a/tracelog/tracelog_test.go
+++ b/tracelog/tracelog_test.go
@@ -217,7 +217,7 @@ func TestLogCopyFrom(t *testing.T) {
 		return config
 	}
 
-	pgxtest.RunWithQueryExecModes(context.Background(), t, ctr, nil, func(ctx context.Context, t testing.TB, conn *pgx.Conn) {
+	pgxtest.RunWithQueryExecModes(context.Background(), t, ctr, pgxtest.KnownOIDQueryExecModes, func(ctx context.Context, t testing.TB, conn *pgx.Conn) {
 		_, err := conn.Exec(context.Background(), `create temporary table foo(a int4)`)
 		require.NoError(t, err)