Documentation
¶
Index ¶
- Constants
- type PostgresSink
- func (s *PostgresSink) Begin(ctx context.Context) error
- func (s *PostgresSink) Browse(ctx context.Context, table string, limit int) ([]hermod.Message, error)
- func (s *PostgresSink) Close() error
- func (s *PostgresSink) Commit(ctx context.Context) error
- func (s *PostgresSink) CommitPrepared(ctx context.Context, txID string) error
- func (s *PostgresSink) DiscoverColumns(ctx context.Context, table string) ([]hermod.ColumnInfo, error)
- func (s *PostgresSink) DiscoverDatabases(ctx context.Context) ([]string, error)
- func (s *PostgresSink) DiscoverTables(ctx context.Context) ([]string, error)
- func (s *PostgresSink) Ping(ctx context.Context) error
- func (s *PostgresSink) PreflightTwoPhaseCommit(ctx context.Context) error
- func (s *PostgresSink) Prepare(ctx context.Context, txID string) (string, error)
- func (s *PostgresSink) Rollback(ctx context.Context) error
- func (s *PostgresSink) RollbackPrepared(ctx context.Context, txID string) error
- func (s *PostgresSink) Sample(ctx context.Context, table string) (hermod.Message, error)
- func (s *PostgresSink) SetLogger(logger hermod.Logger)
- func (s *PostgresSink) Write(ctx context.Context, msg hermod.Message) error
- func (s *PostgresSink) WriteBatch(ctx context.Context, msgs []hermod.Message) error
Constants ¶
View Source
const ( QueryUpsert = "Upsert" QueryDelete = "Delete" QueryListDatabases = "ListDatabases" QueryListTables = "ListTables" QueryBrowse = "Browse" QueryCreateTable = "CreateTable" QueryCreateSchema = "CreateSchema" QueryListColumns = "ListColumns" QueryTableExists = "TableExists" )
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type PostgresSink ¶
type PostgresSink struct {
// contains filtered or unexported fields
}
PostgresSink implements the hermod.Sink interface for PostgreSQL. All exported methods are safe for concurrent use.
func NewPostgresSink ¶
func (*PostgresSink) Close ¶
func (s *PostgresSink) Close() error
func (*PostgresSink) CommitPrepared ¶
func (s *PostgresSink) CommitPrepared(ctx context.Context, txID string) error
func (*PostgresSink) DiscoverColumns ¶
func (s *PostgresSink) DiscoverColumns(ctx context.Context, table string) ([]hermod.ColumnInfo, error)
func (*PostgresSink) DiscoverDatabases ¶
func (s *PostgresSink) DiscoverDatabases(ctx context.Context) ([]string, error)
func (*PostgresSink) DiscoverTables ¶
func (s *PostgresSink) DiscoverTables(ctx context.Context) ([]string, error)
func (*PostgresSink) PreflightTwoPhaseCommit ¶
func (s *PostgresSink) PreflightTwoPhaseCommit(ctx context.Context) error
PreflightTwoPhaseCommit reports whether this sink can genuinely take part in a distributed transaction, and refuses loudly when it cannot.
Two ways it cannot, both of which are silent without this check:
- Behind a transaction pooler. A prepared transaction has to be resolved on the backend that created it, which a pooler cannot guarantee, so Prepare degrades to committing locally and returning a sentinel. That is fine for a lone sink and catastrophic inside a group: the coordinator believes it can still roll back, the data is already committed, and a later abort leaves this sink diverged from the others with nothing reporting it.
- max_prepared_transactions = 0, which is the PostgreSQL default. PREPARE TRANSACTION then fails outright, so every batch aborts mid-flight instead of failing at start-up where an operator would see it.
func (*PostgresSink) RollbackPrepared ¶
func (s *PostgresSink) RollbackPrepared(ctx context.Context, txID string) error
func (*PostgresSink) SetLogger ¶
func (s *PostgresSink) SetLogger(logger hermod.Logger)
func (*PostgresSink) WriteBatch ¶
Click to show internal directories.
Click to hide internal directories.