Prepare overwrites existing prepared statement

Prepare now sends a Close message as part of preparing a statement. This
allows changes Prepare behavior in that it is no longer an error to
prepare a statement with a name that already exists. In addition, it
resolves issues where the connection gets confused and doesn't realize
a statement is already prepared. This could happen if a statement_timout
triggered between the Parse and Describe messages.

#2223

Author: jackc

conn_test.go

@@ -503,9 +503,10 @@ func TestPrepareIdempotency(t *testing.T) {
 			}
 		}
 
+		// Previously, preparing a statement with the same name but different SQL would fail. Now it should succeed.
 		_, err := conn.Prepare(context.Background(), "test", "select 'fail'::varchar")
-		if err == nil {
-			t.Fatalf("Prepare statement with same name but different SQL should have failed but it didn't")
+		if err != nil {
+			t.Fatalf("Prepare statement with same name but different SQL should have succeeded but it didn't: %v", err)
 			return
 		}
 	})

pgconn/pgconn.go

@@ -854,7 +854,8 @@ type StatementDescription struct {
 }
 
 // Prepare creates a prepared statement. If the name is empty, the anonymous prepared statement will be used. This
-// allows Prepare to also to describe statements without creating a server-side prepared statement.
+// allows Prepare to also to describe statements without creating a server-side prepared statement. Prepare is safe to
+// call with a name that is already in use. The previous statement will be closed and replaced with the new one.
 //
 // Prepare does not send a PREPARE statement to the server. It uses the PostgreSQL Parse and Describe protocol messages
 // directly.
@@ -874,6 +875,9 @@ func (pgConn *PgConn) Prepare(ctx context.Context, name, sql string, paramOIDs [
 		defer pgConn.contextWatcher.Unwatch()
 	}
 
+	if name != "" {
+		pgConn.frontend.SendClose(&pgproto3.Close{ObjectType: 'S', Name: name})
+	}
 	pgConn.frontend.SendParse(&pgproto3.Parse{Name: name, Query: sql, ParameterOIDs: paramOIDs})
 	pgConn.frontend.SendDescribe(&pgproto3.Describe{ObjectType: 'S', Name: name})
 	pgConn.frontend.SendSync(&pgproto3.Sync{})

pgconn/pgconn_test.go

@@ -639,6 +639,33 @@ func TestConnectConfigRequiresConfigFromParseConfig(t *testing.T) {
 	require.PanicsWithValue(t, "config must be created by ParseConfig", func() { pgconn.ConnectConfig(ctx, config) })
 }
 
+func TestConnPrepareOverwritesExistingPreparedStatement(t *testing.T) {
+	t.Parallel()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
+	defer cancel()
+
+	pgConn, err := pgconn.Connect(ctx, os.Getenv("PGX_TEST_DATABASE"))
+	require.NoError(t, err)
+	defer closeConn(t, pgConn)
+
+	psd, err := pgConn.Prepare(ctx, "ps1", "select 1", nil)
+	require.NoError(t, err)
+	require.NotNil(t, psd)
+
+	psd, err = pgConn.Prepare(ctx, "ps1", "select 2", nil)
+	require.NoError(t, err)
+	require.NotNil(t, psd)
+
+	result := pgConn.ExecPrepared(ctx, "ps1", nil, nil, nil).Read()
+	require.NoError(t, result.Err)
+	require.Len(t, result.Rows, 1)
+	require.Len(t, result.Rows[0], 1)
+	require.Equal(t, "2", string(result.Rows[0][0]))
+
+	ensureConnValid(t, pgConn)
+}
+
 func TestConnPrepareSyntaxError(t *testing.T) {
 	t.Parallel()