source

Documentation

Index

• Variables
• func CoerceDecimal(val interface{}) (*common.Decimal, error)
• func CoerceFloat64(val interface{}) (float64, error)
• func CoerceInt64(val interface{}) (int64, error)
• func CoerceString(val interface{}) (string, error)
• func CoerceTimestamp(val interface{}) (common.Timestamp, error)
• type Decoder
• type JSONDecoder
  • func (j *JSONDecoder) Decode(bytes []byte) (interface{}, error)
• type KafkaDecoder
  • func (k *KafkaDecoder) Decode(bytes []byte) (interface{}, error)
• type MessageConsumer
  • func NewMessageConsumer(msgProvider kafka.MessageProvider, pollTimeout time.Duration, maxMessages int, ...) (*MessageConsumer, error)
  • func (m *MessageConsumer) Close() error
  • func (m *MessageConsumer) Stop() error
• type MessageParser
  • func NewMessageParser(sourceInfo *common.SourceInfo, registry protolib.Resolver) (*MessageParser, error)
  • func (m *MessageParser) ParseMessages(messages []*kafka.Message) (*common.Rows, error)
• type ProtobufDecoder
  • func (p *ProtobufDecoder) Decode(bytes []byte) (interface{}, error)
• type RowProcessor
• type Source
  • func NewSource(sourceInfo *common.SourceInfo, tableExec *exec.TableExecutor, ...) (*Source, error)
  • func (s *Source) AddConsumingExecutor(mvName string, executor exec.PushExecutor)
  • func (s *Source) Drop() error
  • func (s *Source) GetCommittedCount() int64
  • func (s *Source) GetConsumingMVs() []string
  • func (s *Source) IsRunning() bool
  • func (s *Source) RemoveConsumingExecutor(mvName string)
  • func (s *Source) SetCommitOffsets(enable bool)
  • func (s *Source) SetMaxIngestRate(rate int)
  • func (s *Source) Start() error
  • func (s *Source) Stop() error
  • func (s *Source) TableExecutor() *exec.TableExecutor

Variables

var MaxVarCharOverride = -1

Functions

func CoerceDecimal

func CoerceDecimal(val interface{}) (*common.Decimal, error)

func CoerceFloat64

func CoerceFloat64(val interface{}) (float64, error)

func CoerceInt64

func CoerceInt64(val interface{}) (int64, error)

func CoerceString

func CoerceString(val interface{}) (string, error)

func CoerceTimestamp

func CoerceTimestamp(val interface{}) (common.Timestamp, error)

Types

type Decoder

type Decoder interface {
	Decode(bytes []byte) (interface{}, error)
}

type JSONDecoder

type JSONDecoder struct {
}

func (*JSONDecoder) Decode

func (j *JSONDecoder) Decode(bytes []byte) (interface{}, error)

type KafkaDecoder

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

func (*KafkaDecoder) Decode

func (k *KafkaDecoder) Decode(bytes []byte) (interface{}, error)

type MessageConsumer

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

func NewMessageConsumer

func NewMessageConsumer(msgProvider kafka.MessageProvider, pollTimeout time.Duration, maxMessages int,
	source *Source) (*MessageConsumer, error)

func (*MessageConsumer) Close

func (m *MessageConsumer) Close() error

func (*MessageConsumer) Stop

func (m *MessageConsumer) Stop() error

type MessageParser

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

func NewMessageParser

func NewMessageParser(sourceInfo *common.SourceInfo, registry protolib.Resolver) (*MessageParser, error)

func (*MessageParser) ParseMessages

func (m *MessageParser) ParseMessages(messages []*kafka.Message) (*common.Rows, error)

type ProtobufDecoder

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

func (*ProtobufDecoder) Decode

func (p *ProtobufDecoder) Decode(bytes []byte) (interface{}, error)

type RowProcessor

type RowProcessor interface {
}

type Source

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

func NewSource

func NewSource(sourceInfo *common.SourceInfo, tableExec *exec.TableExecutor, ingestExpressions []*common.Expression, sharder *sharder.Sharder,
	cluster cluster.Cluster, cfg *conf.Config, queryExec common.SimpleQueryExec, registry protolib.Resolver) (*Source, error)

func (*Source) AddConsumingExecutor

func (s *Source) AddConsumingExecutor(mvName string, executor exec.PushExecutor)

func (*Source) Drop

func (s *Source) Drop() error

func (*Source) GetCommittedCount

func (s *Source) GetCommittedCount() int64

func (*Source) GetConsumingMVs

func (s *Source) GetConsumingMVs() []string

func (*Source) IsRunning

func (s *Source) IsRunning() bool

func (*Source) RemoveConsumingExecutor

func (s *Source) RemoveConsumingExecutor(mvName string)

func (*Source) SetCommitOffsets

func (s *Source) SetCommitOffsets(enable bool)

func (*Source) SetMaxIngestRate

func (s *Source) SetMaxIngestRate(rate int)

func (*Source) Start

func (s *Source) Start() error

func (*Source) Stop

func (s *Source) Stop() error

func (*Source) TableExecutor

func (s *Source) TableExecutor() *exec.TableExecutor