push

Documentation

Index

• func SetInitBatchSize(batchSize int)
• type Engine
  • func NewPushEngine(cluster cluster.Cluster, sharder *sharder.Sharder, meta *meta.Controller, ...) *Engine
  • func (p *Engine) CreateIndex(indexInfo *common.IndexInfo, fill bool, shardIDs []uint64, ...) error
  • func (p *Engine) CreateShardListener(shardID uint64) cluster.ShardListener
  • func (p *Engine) CreateSource(sourceInfo *common.SourceInfo, initTable *common.TableInfo) (*source.Source, error)
  • func (p *Engine) ExistRowsInLocalTable(tableID uint64, localShards []uint64) (bool, error)
  • func (p *Engine) GetForwardWriteHandler() remoting.ClusterMessageHandler
  • func (p *Engine) GetLoadClientSetRateHandler() remoting.ClusterMessageHandler
  • func (p *Engine) GetMaterializedView(mvID uint64) (*MaterializedView, error)
  • func (p *Engine) GetSource(sourceID uint64) (*source.Source, error)
  • func (p *Engine) HandleBatch(shardID uint64, rowsBatch []cluster.ForwardRow, first bool) (int64, error)
  • func (p *Engine) HandleForwardWrite(shardID uint64, writeBatch []byte) error
  • func (p *Engine) IsEmpty() bool
  • func (p *Engine) IsStarted() bool
  • func (p *Engine) LoadInitialStateForTable(shardIDs []uint64, initTableID uint64, targetTableID uint64, ...) error
  • func (p *Engine) RegisterMV(mv *MaterializedView) error
  • func (p *Engine) RegisterRemoteConsumer(id uint64, rc *RemoteConsumer) error
  • func (p *Engine) RemoveIndex(indexInfo *common.IndexInfo) error
  • func (p *Engine) RemoveMV(mvID uint64) error
  • func (p *Engine) RemoveSource(sourceInfo *common.SourceInfo) (*source.Source, error)
  • func (p *Engine) ShardFailed(shardID uint64)
  • func (p *Engine) Start() error
  • func (p *Engine) StartSources()
  • func (p *Engine) Stop() error
  • func (p *Engine) UnattachIndex(indexInfo *common.IndexInfo) error
  • func (p *Engine) UnregisterRemoteConsumer(id uint64) error
  • func (p *Engine) VerifyNoSourcesOrMVs() error
  • func (p *Engine) WaitForProcessingToComplete() error
  • func (p *Engine) WaitForSchedulers() error
• type MaterializedView
  • func CreateMaterializedView(pe *Engine, pl *parplan.Planner, schema *common.Schema, mvName string, ...) (*MaterializedView, error)
  • func (m *MaterializedView) Connect(addConsuming bool, registerRemote bool) error
  • func (m *MaterializedView) Disconnect() error
  • func (m *MaterializedView) Drop() error
  • func (m *MaterializedView) Fill(shardIDs []uint64, interruptor *interruptor.Interruptor) error
  • func (m *MaterializedView) GetConsumingMVs() []string
  • func (m *MaterializedView) TableExecutor() *exec.TableExecutor
• type RawRow
• type RemoteConsumer

Functions

func SetInitBatchSize

func SetInitBatchSize(batchSize int)

Types

type Engine

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

func NewPushEngine

func NewPushEngine(cluster cluster.Cluster, sharder *sharder.Sharder, meta *meta.Controller, cfg *conf.Config,
	queryExec common.SimpleQueryExec, registry protolib.Resolver, failInject failinject.Injector) *Engine

func (*Engine) CreateIndex

func (p *Engine) CreateIndex(indexInfo *common.IndexInfo, fill bool, shardIDs []uint64, interruptor *interruptor.Interruptor) error

func (*Engine) CreateShardListener

func (p *Engine) CreateShardListener(shardID uint64) cluster.ShardListener

func (*Engine) CreateSource

func (p *Engine) CreateSource(sourceInfo *common.SourceInfo, initTable *common.TableInfo) (*source.Source, error)

func (*Engine) ExistRowsInLocalTable

func (p *Engine) ExistRowsInLocalTable(tableID uint64, localShards []uint64) (bool, error)

func (*Engine) GetForwardWriteHandler

func (p *Engine) GetForwardWriteHandler() remoting.ClusterMessageHandler

func (*Engine) GetLoadClientSetRateHandler

func (p *Engine) GetLoadClientSetRateHandler() remoting.ClusterMessageHandler

func (*Engine) GetMaterializedView

func (p *Engine) GetMaterializedView(mvID uint64) (*MaterializedView, error)

func (*Engine) GetSource

func (p *Engine) GetSource(sourceID uint64) (*source.Source, error)

func (*Engine) HandleBatch

func (p *Engine) HandleBatch(shardID uint64, rowsBatch []cluster.ForwardRow, first bool) (int64, error)

func (*Engine) HandleForwardWrite

func (p *Engine) HandleForwardWrite(shardID uint64, writeBatch []byte) error

func (*Engine) IsEmpty

func (p *Engine) IsEmpty() bool

func (*Engine) IsStarted

func (p *Engine) IsStarted() bool

func (*Engine) LoadInitialStateForTable

func (p *Engine) LoadInitialStateForTable(shardIDs []uint64, initTableID uint64, targetTableID uint64,
	inter *interruptor.Interruptor) error

func (*Engine) RegisterMV

func (p *Engine) RegisterMV(mv *MaterializedView) error

func (*Engine) RegisterRemoteConsumer

func (p *Engine) RegisterRemoteConsumer(id uint64, rc *RemoteConsumer) error

func (*Engine) RemoveIndex

func (p *Engine) RemoveIndex(indexInfo *common.IndexInfo) error

func (*Engine) RemoveMV

func (p *Engine) RemoveMV(mvID uint64) error

func (*Engine) RemoveSource

func (p *Engine) RemoveSource(sourceInfo *common.SourceInfo) (*source.Source, error)

func (*Engine) ShardFailed

func (p *Engine) ShardFailed(shardID uint64)

func (*Engine) Start

func (p *Engine) Start() error

Start - starts - the schedulers, sources are started later

func (*Engine) StartSources

func (p *Engine) StartSources()

StartSources - starts the sources - these must be started after the engine has been started otherwise we can get into a deadlock where source start is blocked waiting for Kafka rebalance to complete but consumer thread is blocked trying to ingest messages, but this is blocked on checking whether engine is started

func (*Engine) Stop

func (p *Engine) Stop() error

func (*Engine) UnattachIndex

func (p *Engine) UnattachIndex(indexInfo *common.IndexInfo) error

func (*Engine) UnregisterRemoteConsumer

func (p *Engine) UnregisterRemoteConsumer(id uint64) error

func (*Engine) VerifyNoSourcesOrMVs

func (p *Engine) VerifyNoSourcesOrMVs() error

func (*Engine) WaitForProcessingToComplete

func (p *Engine) WaitForProcessingToComplete() error

WaitForProcessingToComplete is used in tests to wait for all rows have been processed when ingesting test data

func (*Engine) WaitForSchedulers

func (p *Engine) WaitForSchedulers() error

type MaterializedView

type MaterializedView struct {
	Info *common.MaterializedViewInfo

	InternalTables []*common.InternalTableInfo
	// contains filtered or unexported fields
}

func CreateMaterializedView

func CreateMaterializedView(pe *Engine, pl *parplan.Planner, schema *common.Schema, mvName string, query string,
	initTable string, tableID uint64, seqGenerator common.SeqGenerator) (*MaterializedView, error)

CreateMaterializedView creates the materialized view but does not register it in memory

func (*MaterializedView) Connect

func (m *MaterializedView) Connect(addConsuming bool, registerRemote bool) error

Connect connects up any executors which consumer data from sources, materialized views, or remote receivers to their feeders

func (*MaterializedView) Disconnect

func (m *MaterializedView) Disconnect() error

func (*MaterializedView) Drop

func (m *MaterializedView) Drop() error

func (*MaterializedView) Fill

func (m *MaterializedView) Fill(shardIDs []uint64, interruptor *interruptor.Interruptor) error

func (*MaterializedView) GetConsumingMVs

func (m *MaterializedView) GetConsumingMVs() []string

func (*MaterializedView) TableExecutor

func (m *MaterializedView) TableExecutor() *exec.TableExecutor

type RawRow

type RawRow struct {
	ReceiverSequence uint64
	Row              []byte
}

type RemoteConsumer

type RemoteConsumer struct {
	RowsFactory *common.RowsFactory
	ColTypes    []common.ColumnType
	RowsHandler remoteRowsHandler
}

RemoteConsumer is a wrapper for something that consumes rows that have arrived remotely from other shards e.g. a source or an aggregator