PgReplicationConnection

Namespace: Ankus

Assembly: Ankus.Runtime.dll

Connects this backend to a PostgreSQL server for logical replication and SQL, through PostgreSQL’s own libpqwalreceiver, as the built-in logical replication subscriber connects.

public sealed class PgReplicationConnection : IDisposable

The connection is a logical replication connection to the database the connection string names, which also runs ordinary SQL. It belongs to the backend that opened it, outlives transactions, and closes when disposed or when the backend exits. Waits wake when this backend’s latch is set and honor query cancellation. Connection and protocol failures raise PostgreSQL errors in this backend; a failed command on the server throws PgRemoteException and leaves this backend’s transaction untouched.

Inheritance: object

Implements: IDisposable

Methods

Connect(string, string)

Opens a logical replication connection.

public static PgReplicationConnection Connect(string connectionString, string applicationName)

Parameters:

connectionString — string

A libpq connection string naming the server and database.

applicationName — string

The name the server shows for this connection, such as in pg_stat_replication.

Returns: PgReplicationConnection

The open connection.

Exceptions:

• PgException: The server cannot be reached or rejects the connection.

CopyInto(uint, IReadOnlyList<string>?)

Reads the rows of a COPY … TO STDOUT started with StartCopy(string) into a local table, as the built-in subscriber’s table synchronization does.

public long CopyInto(uint relationOid, IReadOnlyList<string>? columns = null)

Parameters:

relationOid — uint

The local table’s OID.

columns — IReadOnlyList<string>

The table’s columns that receive the copy’s columns, in order, or null for all of them.

Returns: long

The number of rows copied.

Exceptions:

• PgException: A row is rejected, the user lacks the privilege, or the connection failed.

The copy must use COPY’s text format. The rows go through PostgreSQL’s own COPY FROM, which checks constraints, fills indexes, routes rows to partitions and fires triggers as session_replication_role selects. It needs INSERT on the table and, like the built-in subscriber, refuses a table whose row-level security applies to the user. The connection runs commands again afterward. If the copy fails, the server’s copy is left unfinished, so dispose the connection.

Dispose()

Closes the connection.

public void Dispose()

EndStreaming()

Ends a streaming copy started with START_REPLICATION, so the connection runs commands again.

public void EndStreaming()

Exceptions:

• PgException: The connection failed.

Execute(string)

Runs a command that returns no rows, such as SET, BEGIN or DROP_REPLICATION_SLOT.

public void Execute(string command)

Parameters:

command — string

The command text.

Exceptions:

• PgRemoteException: The command failed on the server.
• PgException: The command returned rows or started a copy, or the connection failed.

Query(string, int)

Runs a query and returns its rows as text, as the server formats each value.

public IReadOnlyList<string?[]> Query(string query, int columns)

Parameters:

query — string

The query text, such as SELECT, IDENTIFY_SYSTEM or CREATE_REPLICATION_SLOT.

columns — int

The number of columns the query returns; a different number is an error.

Returns: IReadOnlyList<string[]>

Each row’s values, with null for SQL NULL.

Exceptions:

• PgRemoteException: The query failed on the server.
• PgException: The query returned another number of columns, or the connection failed.

Rows need a transaction, because PostgreSQL looks up their column types in the catalog; a background worker runs such queries inside RunTransaction(Action). Commands without rows, streaming and copies work outside transactions.

Receive(TimeSpan, out ReadOnlySpan<byte>)

Receives the next message of a copy, waiting up to timeout for one.

public PgReplicationReceiveStatus Receive(TimeSpan timeout, out ReadOnlySpan<byte> message)

Parameters:

timeout — TimeSpan

How long to wait, Zero to only check, or InfiniteTimeSpan to wait until a message arrives or the latch is set.

message — ReadOnlySpan<byte>

The message, valid until the connection’s next call.

Returns: PgReplicationReceiveStatus

Whether a message arrived, the wait timed out, the latch woke it, or the copy ended.

Exceptions:

• PgException: The connection failed, or the query was canceled.

Send(ReadOnlySpan<byte>)

Sends a message during a streaming copy, such as a standby status update.

public void Send(ReadOnlySpan<byte> message)

Parameters:

message — ReadOnlySpan<byte>

The message bytes.

Exceptions:

• PgException: The connection failed.

StartCopy(string)

Starts a copy: COPY … TO STDOUT, whose rows Receive(TimeSpan, out ReadOnlySpan<byte>) then returns, or START_REPLICATION, which streams changes and accepts Send(ReadOnlySpan<byte>).

public void StartCopy(string command)

Parameters:

command — string

The command text.

Exceptions:

• PgRemoteException: The command failed on the server.
• PgException: The command did not start a copy, or the connection failed.

Properties

ServerVersion

Gets the server’s version number, such as 180006 for PostgreSQL 18.6.

public int ServerVersion { get; }

Value: int