store

Documentation

Index

• type Config
• type IndexName
• type IndexNameAdapter
• type Option
  • func WithIndexNameAdapter(a IndexNameAdapter) Option
  • func WithLogger(l loglib.Logger) Option
  • func WithMapper(m search.Mapper) Option
• type PgMapper
  • func NewPostgresMapper(mapper searchstore.Mapper) *PgMapper
  • func (m *PgMapper) ColumnToSearchMapping(column schemalog.Column) (map[string]any, error)
  • func (m *PgMapper) MapColumnValue(column schemalog.Column, value any) (any, error)
• type SearchAdapter
• type Store
  • func NewStore(cfg Config, opts ...Option) (*Store, error)
  • func NewStoreWithClient(client searchstore.Client) *Store
  • func (s *Store) ApplySchemaChange(ctx context.Context, newEntry *schemalog.LogEntry) error
  • func (s *Store) DeleteSchema(ctx context.Context, schemaName string) error
  • func (s *Store) DeleteTableDocuments(ctx context.Context, schemaName string, tableIDs []string) error
  • func (s *Store) GetMapper() search.Mapper
  • func (s *Store) SendDocuments(ctx context.Context, docs []search.Document) ([]search.DocumentError, error)

Types

type Config

type Config struct {
	OpenSearchURL    string
	ElasticsearchURL string
}

type IndexName

type IndexName interface {
	Name() string
	Version() int
	NameWithVersion() string
	SchemaName() string
}

IndexName represents an opensearch index name constructed from a schema name.

type IndexNameAdapter

type IndexNameAdapter interface {
	SchemaNameToIndex(schemaName string) IndexName
	IndexToSchemaName(index string) string
}

type Option

type Option func(*Store)

func WithIndexNameAdapter

func WithIndexNameAdapter(a IndexNameAdapter) Option

func WithLogger

func WithLogger(l loglib.Logger) Option

func WithMapper

func WithMapper(m search.Mapper) Option

type PgMapper

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

func NewPostgresMapper

func NewPostgresMapper(mapper searchstore.Mapper) *PgMapper

NewPostgresMapper returns a mapper that maps between postgres and search store types

func (*PgMapper) ColumnToSearchMapping

func (m *PgMapper) ColumnToSearchMapping(column schemalog.Column) (map[string]any, error)

ColumnToSearchMapping maps the column on input into the equivalent search mapping

func (*PgMapper) MapColumnValue

func (m *PgMapper) MapColumnValue(column schemalog.Column, value any) (any, error)

MapColumnValue maps a value emitted from PG into a value that the search store can handle. If the column is a timestamp: we need to parse it. If the column is an array of any type except json, we need to map it to a Go slice. If column type is unknown we return nil. This avoids dropping the whole record if one field type is unknown.

type SearchAdapter

type SearchAdapter interface {
	SearchDocToBulkItem(docs search.Document) searchstore.BulkItem
	BulkItemsToSearchDocErrs(items []searchstore.BulkItem) []search.DocumentError
	RecordToLogEntry(rec map[string]any) (*schemalog.LogEntry, error)
}

Adapter converts from/to search types and opensearch types

type Store

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

func NewStore

func NewStore(cfg Config, opts ...Option) (*Store, error)

func NewStoreWithClient

func NewStoreWithClient(client searchstore.Client) *Store

func (*Store) ApplySchemaChange

func (s *Store) ApplySchemaChange(ctx context.Context, newEntry *schemalog.LogEntry) error

func (*Store) DeleteSchema

func (s *Store) DeleteSchema(ctx context.Context, schemaName string) error

func (*Store) DeleteTableDocuments

func (s *Store) DeleteTableDocuments(ctx context.Context, schemaName string, tableIDs []string) error

func (*Store) GetMapper

func (s *Store) GetMapper() search.Mapper

func (*Store) SendDocuments

func (s *Store) SendDocuments(ctx context.Context, docs []search.Document) ([]search.DocumentError, error)