pipeline

Documentation

Index

• func DeleteEmpty(n *filterNode, c *Collection, msg *DeleteMsg) error
• func DeleteNotAligned(n *filterNode, c *Collection, msg *DeleteMsg) error
• func DeleteOutOfTarget(n *filterNode, c *Collection, msg *DeleteMsg) error
• func GetEmbeddingFieldData(datas []*schemapb.FieldData, fieldID int64) ([]string, error)
• func InsertEmpty(n *filterNode, c *Collection, msg *InsertMsg) error
• func InsertNotAligned(n *filterNode, c *Collection, msg *InsertMsg) error
• func InsertOutOfTarget(n *filterNode, c *Collection, msg *InsertMsg) error
• type BaseNode
• type Collection
• type DataManager
• type DeleteMsg
• type DeleteMsgFilter
• type InsertMsg
• type InsertMsgFilter
• type Manager
  • func NewManager(dataManager *DataManager, tSafeManager TSafeManager, ...) Manager
• type Msg
• type Pipeline
  • func NewPipeLine(collectionID UniqueID, channel string, manager *DataManager, ...) (Pipeline, error)
• type PrimaryKey
• type Segment
• type TSafeManager
• type TimeRange
• type Timestamp
• type UniqueID

Functions

func DeleteEmpty

func DeleteEmpty(n *filterNode, c *Collection, msg *DeleteMsg) error

func DeleteNotAligned

func DeleteNotAligned(n *filterNode, c *Collection, msg *DeleteMsg) error

func DeleteOutOfTarget

func DeleteOutOfTarget(n *filterNode, c *Collection, msg *DeleteMsg) error

func GetEmbeddingFieldData

func GetEmbeddingFieldData(datas []*schemapb.FieldData, fieldID int64) ([]string, error)

func InsertEmpty

func InsertEmpty(n *filterNode, c *Collection, msg *InsertMsg) error

func InsertNotAligned

func InsertNotAligned(n *filterNode, c *Collection, msg *InsertMsg) error

Chack msg is aligned -- len of each kind of infos in InsertMsg should match each other

func InsertOutOfTarget

func InsertOutOfTarget(n *filterNode, c *Collection, msg *InsertMsg) error

Types

type BaseNode

type BaseNode = base.BaseNode

type Collection

type Collection = segments.Collection

type DataManager

type DataManager = segments.Manager

type DeleteMsg

type DeleteMsg = msgstream.DeleteMsg

type DeleteMsgFilter

type DeleteMsgFilter = func(n *filterNode, c *Collection, msg *DeleteMsg) error

type InsertMsg

type InsertMsg = msgstream.InsertMsg

type InsertMsgFilter

type InsertMsgFilter = func(n *filterNode, c *Collection, msg *InsertMsg) error

MsgFilter will return error if Msg was invalid

type Manager

type Manager interface {
	Num() int
	Add(collectionID UniqueID, channel string) (Pipeline, error)
	Get(channel string) Pipeline
	Remove(channels ...string)
	Start(channels ...string) error
	Close()
	GetChannelStats() []*metricsinfo.Channel
}

Manager manage pipeline in querynode

func NewManager

func NewManager(dataManager *DataManager,
	tSafeManager TSafeManager,
	dispatcher msgdispatcher.Client,
	delegators *typeutil.ConcurrentMap[string, delegator.ShardDelegator],
) Manager

type Msg

type Msg = base.Msg

type Pipeline

type Pipeline interface {
	base.StreamPipeline
	GetCollectionID() UniqueID
}

pipeline used for querynode

func NewPipeLine

func NewPipeLine(
	collectionID UniqueID,
	channel string,
	manager *DataManager,
	tSafeManager TSafeManager,
	dispatcher msgdispatcher.Client,
	delegator delegator.ShardDelegator,
) (Pipeline, error)

type PrimaryKey

type PrimaryKey = storage.PrimaryKey

type Segment

type Segment = segments.Segment

type TSafeManager

type TSafeManager = tsafe.Manager

type TimeRange

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

type Timestamp

type Timestamp = typeutil.Timestamp

type UniqueID

type UniqueID = typeutil.UniqueID