message

package
v4.99.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 19, 2026 License: MPL-2.0 Imports: 6 Imported by: 103

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func FailedTables added in v4.99.0 

func FailedTables(err error) []string

FailedTables returns the tables a flush error names, sorted and without duplicates. It returns nil if some error joined in err names no tables, as messages of any table may then have been lost.

Types

type DeleteRecord added in v4.13.0 

type DeleteRecord struct {
	TableName      string
	TableRelations TableRelations
	WhereClause    PredicateGroups
	SyncTime       time.Time
}

type FailedTablesError added in v4.99.0 

type FailedTablesError struct {
	Tables []string
	Err    error
}

FailedTablesError is a flush error that names the tables whose messages may not have been committed.

func (*FailedTablesError) Error added in v4.99.0 

func (e *FailedTablesError) Error() string

func (*FailedTablesError) Unwrap added in v4.99.0 

func (e *FailedTablesError) Unwrap() error

type Predicate added in v4.13.0 

type Predicate struct {
	Operator string
	Column   string
	Record   arrow.RecordBatch
}

type PredicateGroup added in v4.13.0 

type PredicateGroup struct {
	// This will be AND or OR
	GroupingType string
	Predicates   Predicates
}

type PredicateGroups added in v4.13.0 

type PredicateGroups []PredicateGroup

type Predicates added in v4.13.0 

type Predicates []Predicate

type SyncCheckpoint added in v4.99.0 

type SyncCheckpoint struct {
	Record arrow.RecordBatch
	// contains filtered or unexported fields
}

SyncCheckpoint carries state backend rows (key, value, version). The CLI persists them once every message sent before the checkpoint is committed to all destinations.

func (*SyncCheckpoint) GetTable added in v4.99.0 

func (m *SyncCheckpoint) GetTable() *schema.Table

func (*SyncCheckpoint) IsSyncMessage added in v4.99.0 

func (*SyncCheckpoint) IsSyncMessage() bool

type SyncDeleteRecord added in v4.13.0 

type SyncDeleteRecord struct {

	// TODO: Instead of using this struct we should derive the DeletionKeys and parent/child relation from the schema.Table itself
	DeleteRecord
	// contains filtered or unexported fields
}

func (SyncDeleteRecord) GetTable added in v4.13.0 

func (m SyncDeleteRecord) GetTable() *schema.Table

func (*SyncDeleteRecord) IsSyncMessage added in v4.13.0 

func (*SyncDeleteRecord) IsSyncMessage() bool

type SyncError added in v4.85.0 

type SyncError struct {
	TableName string
	Error     string
	// contains filtered or unexported fields
}

func (SyncError) GetTable added in v4.85.0 

func (e SyncError) GetTable() *schema.Table

func (*SyncError) IsSyncMessage added in v4.85.0 

func (*SyncError) IsSyncMessage() bool

type SyncInsert 

type SyncInsert struct {
	Record arrow.RecordBatch
	// contains filtered or unexported fields
}

func (*SyncInsert) GetTable 

func (m *SyncInsert) GetTable() *schema.Table

func (*SyncInsert) IsSyncMessage 

func (*SyncInsert) IsSyncMessage() bool

type SyncInserts 

type SyncInserts []*SyncInsert

func (SyncInserts) Exists 

func (m SyncInserts) Exists(tableName string) bool

func (SyncInserts) GetRecords 

func (m SyncInserts) GetRecords() []arrow.RecordBatch

func (SyncInserts) GetRecordsForTable 

func (m SyncInserts) GetRecordsForTable(table *schema.Table) []arrow.RecordBatch

Get all records for a single table

type SyncMessage 

type SyncMessage interface {
	GetTable() *schema.Table
	IsSyncMessage() bool
}

type SyncMessages 

type SyncMessages []SyncMessage

func (SyncMessages) GetInserts 

func (messages SyncMessages) GetInserts() SyncInserts

func (SyncMessages) InsertItems 

func (messages SyncMessages) InsertItems() int64

type SyncMigrateTable 

type SyncMigrateTable struct {
	Table *schema.Table
	// contains filtered or unexported fields
}

func (SyncMigrateTable) GetTable 

func (m SyncMigrateTable) GetTable() *schema.Table

func (*SyncMigrateTable) IsSyncMessage 

func (*SyncMigrateTable) IsSyncMessage() bool

type SyncMigrateTables 

type SyncMigrateTables []*SyncMigrateTable

func (SyncMigrateTables) Exists 

func (m SyncMigrateTables) Exists(tableName string) bool

type TableRelation added in v4.13.0 

type TableRelation struct {
	TableName   string
	ParentTable string
}

type TableRelations added in v4.13.0 

type TableRelations []TableRelation

type WriteDeleteRecord added in v4.13.0 

type WriteDeleteRecord struct {
	DeleteRecord
	// contains filtered or unexported fields
}

func (WriteDeleteRecord) GetTable added in v4.13.0 

func (m WriteDeleteRecord) GetTable() *schema.Table

func (*WriteDeleteRecord) IsWriteMessage added in v4.13.0 

func (*WriteDeleteRecord) IsWriteMessage() bool

type WriteDeleteRecords added in v4.13.0 

type WriteDeleteRecords []*WriteDeleteRecord

type WriteDeleteStale 

type WriteDeleteStale struct {
	TableName  string
	SourceName string
	SyncTime   time.Time
	// contains filtered or unexported fields
}

WriteDeleteStale is a pretty specific message which requires the destination to be aware of a CLI use-case thus it might be deprecated in the future in favour of MessageDelete or MessageRawQuery The message indicates that the destination needs to run something like "DELETE FROM table WHERE _cq_source_name=$1 and sync_time < $2"

func (WriteDeleteStale) GetTable 

func (m WriteDeleteStale) GetTable() *schema.Table

func (*WriteDeleteStale) IsWriteMessage 

func (*WriteDeleteStale) IsWriteMessage() bool

type WriteDeleteStales 

type WriteDeleteStales []*WriteDeleteStale

func (WriteDeleteStales) Exists 

func (m WriteDeleteStales) Exists(tableName string) bool

type WriteFlush added in v4.99.0 

type WriteFlush struct {
	// contains filtered or unexported fields
}

WriteFlush asks the writer to commit every message it received before this one. The writer calls Done once it has tried to commit them, with an error if some messages it received since the previous flush may not have been committed. A *FailedTablesError names the tables of those messages.

func (*WriteFlush) Done added in v4.99.0 

func (m *WriteFlush) Done(err error)

Done reports the outcome of the flush. Only the first call has an effect.

func (*WriteFlush) GetTable added in v4.99.0 

func (*WriteFlush) GetTable() *schema.Table

func (*WriteFlush) IsWriteMessage added in v4.99.0 

func (*WriteFlush) IsWriteMessage() bool

func (*WriteFlush) Result added in v4.99.0 

func (m *WriteFlush) Result() <-chan error

Result receives the error passed to Done.

type WriteInsert 

type WriteInsert struct {
	Record arrow.RecordBatch
	// contains filtered or unexported fields
}

func (*WriteInsert) GetTable 

func (m *WriteInsert) GetTable() *schema.Table

func (*WriteInsert) IsWriteMessage 

func (*WriteInsert) IsWriteMessage() bool

type WriteInserts 

type WriteInserts []*WriteInsert

func (WriteInserts) Exists 

func (m WriteInserts) Exists(tableName string) bool

func (WriteInserts) GetRecords 

func (m WriteInserts) GetRecords() []arrow.RecordBatch

func (WriteInserts) GetRecordsForTable 

func (m WriteInserts) GetRecordsForTable(table *schema.Table) []arrow.RecordBatch

type WriteMessage 

type WriteMessage interface {
	GetTable() *schema.Table
	IsWriteMessage() bool
}

type WriteMessages 

type WriteMessages []WriteMessage

func (WriteMessages) GetInserts 

func (messages WriteMessages) GetInserts() WriteInserts

func (WriteMessages) InsertItems 

func (messages WriteMessages) InsertItems() int64

type WriteMigrateTable 

type WriteMigrateTable struct {
	Table        *schema.Table
	MigrateForce bool
	// contains filtered or unexported fields
}

func (WriteMigrateTable) GetTable 

func (m WriteMigrateTable) GetTable() *schema.Table

func (*WriteMigrateTable) IsWriteMessage 

func (*WriteMigrateTable) IsWriteMessage() bool

type WriteMigrateTables 

type WriteMigrateTables []*WriteMigrateTable

func (WriteMigrateTables) Exists 

func (m WriteMigrateTables) Exists(tableName string) bool

func (WriteMigrateTables) GetMessageByTable 

func (m WriteMigrateTables) GetMessageByTable(tableName string) *WriteMigrateTable

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.