execution

package

Go to main page

Versions in this module

v0

v0.13.0

May 11, 2024

v0.12.2

May 30, 2023

v0.12.1

Feb 25, 2023

v0.12.0

Oct 19, 2022

v0.11.1

Oct 15, 2022

v0.11.0

Oct 2, 2022

v0.10.0

Sep 25, 2022

v0.9.4

Sep 4, 2022

v0.9.3

Aug 17, 2022

v0.9.2

Jul 24, 2022

Changes in this version

+ type TypeCast struct

+ func NewTypeCast(targetTypeID octosql.TypeID, expr Expression) *TypeCast

+ func (c *TypeCast) Evaluate(ctx ExecutionContext) (octosql.Value, error)

v0.9.1

Jul 17, 2022

v0.9.0

Jul 17, 2022

v0.8.1

Jul 17, 2022

v0.8.0

Jul 1, 2022

v0.7.4

Jun 21, 2022

v0.7.3

Jun 1, 2022

v0.7.2

May 19, 2022

Changes in this version

type RecordEventTimeBuffer

+ func (b *RecordEventTimeBuffer) Empty() bool

v0.7.1

May 14, 2022

v0.7.0

May 12, 2022

Changes in this version

+ func CompareValueSlices(key, than []octosql.Value) bool

v0.6.2

Apr 18, 2022

v0.6.1

Apr 18, 2022

v0.6.0

Apr 15, 2022

v0.5.0

Mar 27, 2022

Changes in this version

+ type LayoutMapping struct

+ List ...

+ Struct ...

+ Tuple ...

+ type ObjectFieldAccess struct

+ func NewObjectFieldAccess(object Expression, fieldIndex int) *ObjectFieldAccess

+ func (c *ObjectFieldAccess) Evaluate(ctx ExecutionContext) (octosql.Value, error)

+ type ObjectLayoutFixer struct

+ func NewObjectLayoutFixer(targetType octosql.Type, sourceTypes []octosql.Type) *ObjectLayoutFixer

+ func (f *ObjectLayoutFixer) FixLayout(index int, value octosql.Value) octosql.Value

v0.4.2

Jan 3, 2022

v0.4.1

Jan 3, 2022

v0.4.0

Dec 30, 2021

Changes in this version

+ const BTreeDefaultDegree

+ var WatermarkMaxValue = time.Unix(0, math.MaxInt64)

+ func NewCountingTriggerPrototype(triggerAfter uint) func() Trigger

+ func NewEndOfStreamTriggerPrototype() func() Trigger

+ func NewMultiTriggerPrototype(triggerPrototypes []func() Trigger) func() Trigger

+ func NewWatermarkTriggerPrototype(timeFieldKeyIndex int) func() Trigger

+ func ProduceFnApplyContext(fn ProduceFn, ctx ProduceContext) func(record Record) error

+ type Cast struct

+ func NewCast(targetType octosql.Type, expr Expression) *Cast

+ func (c *Cast) Evaluate(ctx ExecutionContext) (octosql.Value, error)

+ type Coalesce struct

+ func NewCoalesce(args []Expression) *Coalesce

+ func (c *Coalesce) Evaluate(ctx ExecutionContext) (octosql.Value, error)

type CountingTrigger

+ func (c *CountingTrigger) EndOfStreamReached()

+ func (c *CountingTrigger) KeyReceived(key GroupKey)

+ func (c *CountingTrigger) Poll() []GroupKey

+ func (c *CountingTrigger) WatermarkReceived(watermark time.Time)

+ type EndOfStreamTrigger struct

+ func (c *EndOfStreamTrigger) EndOfStreamReached()

+ func (c *EndOfStreamTrigger) KeyReceived(key GroupKey)

+ func (c *EndOfStreamTrigger) Poll() []GroupKey

+ func (c *EndOfStreamTrigger) WatermarkReceived(watermark time.Time)

+ type ExecutionContext struct

+ VariableContext *VariableContext

+ func (ctx ExecutionContext) WithRecord(record Record) ExecutionContext

type Expression

+ Evaluate func(ctx ExecutionContext) (octosql.Value, error)

+ type FunctionCall struct

+ func NewFunctionCall(function func([]octosql.Value) (octosql.Value, error), args []Expression, ...) *FunctionCall

+ func (c *FunctionCall) Evaluate(ctx ExecutionContext) (octosql.Value, error)

+ type GroupKey []octosql.Value

+ func (key GroupKey) GetGroupKey() GroupKey

+ func (key GroupKey) Less(than btree.Item) bool

+ type GroupKeyIface interface

+ GetGroupKey func() GroupKey

+ type MetaSendFn func(ctx ProduceContext, msg MetadataMessage) error

+ type MetadataMessage struct

+ Type MetadataMessageType

+ Watermark time.Time

+ type MetadataMessageType int

+ const MetadataMessageTypeWatermark

+ type MultiColumnQueryExpression struct

+ func NewMultiColumnQueryExpression(source Node) *MultiColumnQueryExpression

+ func (e *MultiColumnQueryExpression) Evaluate(ctx ExecutionContext) (octosql.Value, error)

type MultiTrigger

+ func (c *MultiTrigger) EndOfStreamReached()

+ func (c *MultiTrigger) KeyReceived(key GroupKey)

+ func (c *MultiTrigger) Poll() []GroupKey

+ func (c *MultiTrigger) WatermarkReceived(watermark time.Time)

type Node

+ Run func(ctx ExecutionContext, produce ProduceFn, metaSend MetaSendFn) error

+ type ProduceContext struct

+ func ProduceFromExecutionContext(ctx ExecutionContext) ProduceContext

+ type ProduceFn func(ctx ProduceContext, record Record) error

type Record

+ Retraction bool

+ Values []octosql.Value

+ type RecordEventTimeBuffer struct

+ func NewRecordEventTimeBuffer() *RecordEventTimeBuffer

+ func (b *RecordEventTimeBuffer) AddRecord(record Record)

+ func (b *RecordEventTimeBuffer) Emit(watermark time.Time, produce func(record Record) error) error

+ type SingleColumnQueryExpression struct

+ func NewSingleColumnQueryExpression(source Node) *SingleColumnQueryExpression

+ func (e *SingleColumnQueryExpression) Evaluate(ctx ExecutionContext) (octosql.Value, error)

type Trigger

+ EndOfStreamReached func()

+ KeyReceived func(key GroupKey)

+ Poll func() []GroupKey

+ WatermarkReceived func(watermark time.Time)

+ type Tuple struct

+ func (c *Tuple) Evaluate(ctx ExecutionContext) (octosql.Value, error)

+ type TypeAssertion struct

+ func NewTypeAssertion(expected octosql.Type, expr Expression) *TypeAssertion

+ func (c *TypeAssertion) Evaluate(ctx ExecutionContext) (octosql.Value, error)

type Variable

+ func (r *Variable) Evaluate(ctx ExecutionContext) (octosql.Value, error)

+ type VariableContext struct

+ Parent *VariableContext

+ Values []octosql.Value

+ func (varCtx *VariableContext) WithRecord(record Record) *VariableContext

type WatermarkTrigger

+ func (c *WatermarkTrigger) EndOfStreamReached()

+ func (c *WatermarkTrigger) KeyReceived(key GroupKey)

+ func (c *WatermarkTrigger) Poll() []GroupKey

+ func (c *WatermarkTrigger) WatermarkReceived(watermark time.Time)

v0.4.0-rc1

Dec 30, 2021

v0.3.0

Aug 5, 2020

Changes in this version

+ var ErrNewTransactionRequired = fmt.Errorf("new transaction required")

+ var MaxWatermark = time.Date(9999, 1, 1, 0, 0, 0, 0, time.UTC)

+ var SystemSource string = "sys"

+ func AreStreamsEqualNoOrderingWithCount(ctx context.Context, stateStorage storage.Storage, first, second RecordStream, ...) error

+ func AreStreamsEqualNoOrderingWithIDCheck(ctx context.Context, stateStorage storage.Storage, ...) error

+ func AreStreamsEqualNoOrderingWithRetractionReductionAndIDChecking(ctx context.Context, stateStorage storage.Storage, got, want RecordStream, ...) error

+ func AreStreamsEqualWithOrdering(ctx context.Context, stateStorage storage.Storage, first, second RecordStream) error

+ func DefaultEquality(record1 *Record, record2 *Record) error

+ func EqualityOfAll(fs ...RecordEqualityFunc) func(record1 *Record, record2 *Record) error

+ func EqualityOfEventTimeField(record1 *Record, record2 *Record) error

+ func EqualityOfEverythingButIDs(record1 *Record, record2 *Record) error

+ func EqualityOfFieldsAndValues(record1 *Record, record2 *Record) error

+ func EqualityOfID(record1 *Record, record2 *Record) error

+ func EqualityOfUndo(record1 *Record, record2 *Record) error

+ func GetAndStartAllShuffles(ctx context.Context, stateStorage storage.Storage, rootStreamID *StreamID, ...) ([]RecordStream, []*ExecutionOutput, error)

+ func GetRawStringID() string

+ func GetSourceStringID(tx storage.StateTransaction, inputName octosql.Value) (string, error)

+ func NewErrWaitForChanges(subscription *storage.Subscription) error

+ func SystemField(field string) octosql.VariableName

type Aggregate

+ AddValue func(ctx context.Context, tx storage.StateTransaction, value octosql.Value) error

+ GetValue func(ctx context.Context, tx storage.StateTransaction) (octosql.Value, error)

+ RetractValue func(ctx context.Context, tx storage.StateTransaction, value octosql.Value) error

+ type AreEqualConfig struct

+ Equality RecordEqualityFunc

+ type AreEqualOpt func(*AreEqualConfig)

+ func WithEqualityBasedOn(fs ...RecordEqualityFunc) AreEqualOpt

+ type BatchSizeManager struct

+ func NewBatchSizeManager(latencyTarget time.Duration) *BatchSizeManager

+ func (bsm *BatchSizeManager) CommitAborted()

+ func (bsm *BatchSizeManager) CommitSuccessful()

+ func (bsm *BatchSizeManager) CommitTooBig()

+ func (bsm *BatchSizeManager) MarkRecordsProcessed(count int)

+ func (bsm *BatchSizeManager) RecordsLeftToTake() int

+ func (bsm *BatchSizeManager) Reset()

+ func (bsm *BatchSizeManager) ShouldTakeNextRecord() bool

+ type ConstantStrategy struct

+ func (s *ConstantStrategy) CalculatePartition(ctx context.Context, record *Record, outputs int) (int, error)

+ type ConstantStrategyPrototype struct

+ func (s *ConstantStrategyPrototype) Get(ctx context.Context, variables octosql.Variables) (ShuffleStrategy, error)

+ type ConstantValue struct

+ func NewConstantValue(value octosql.Value) *ConstantValue

+ func (dv *ConstantValue) ExpressionValue(ctx context.Context, variables octosql.Variables) (octosql.Value, error)

+ type CountingTrigger struct

+ func NewCountingTrigger(count Expression) *CountingTrigger

+ func (c *CountingTrigger) Get(ctx context.Context, variables octosql.Variables) (Trigger, error)

+ type DelayTrigger struct

+ func NewDelayTrigger(delay Expression) *DelayTrigger

+ func (c *DelayTrigger) Get(ctx context.Context, variables octosql.Variables) (Trigger, error)

type DistinctStream

+ func (ds *DistinctStream) AddRecord(ctx context.Context, tx storage.StateTransaction, inputIndex int, ...) error

+ func (ds *DistinctStream) Trigger(ctx context.Context, tx storage.StateTransaction, key octosql.Value) ([]*Record, error)

+ type ErrWaitForChanges struct

+ func GetErrWaitForChanges(err error) *ErrWaitForChanges

+ func (e *ErrWaitForChanges) Error() string

+ type ExecutionOutput struct

+ NextShuffles map[string]ShuffleData

+ TasksToRun []Task

+ WatermarkSource WatermarkSource

+ func NewExecutionOutput(ws WatermarkSource, nextShuffles map[string]ShuffleData, tasksToRun []Task) *ExecutionOutput

type Expression

+ func NewRecordExpression() Expression

+ type GetTestStreamOption func(*StreamID)

+ func GetTestStreamWithStreamID(id *StreamID) GetTestStreamOption

type GroupByStream

+ func (gb *GroupByStream) AddRecord(ctx context.Context, tx storage.StateTransaction, inputIndex int, ...) error

+ func (gb *GroupByStream) Trigger(ctx context.Context, tx storage.StateTransaction, key octosql.Value) ([]*Record, error)

+ type IntermediateRecordStore interface

+ AddRecord func(ctx context.Context, tx storage.StateTransaction, inputIndex int, ...) error

+ Close func(ctx context.Context, storage storage.Storage) error

+ GetWatermark func(ctx context.Context, tx storage.StateTransaction) (time.Time, error)

+ MarkEndOfStream func(ctx context.Context, tx storage.StateTransaction) error

+ MarkError func(ctx context.Context, tx storage.StateTransaction, err error) error

+ Next func(ctx context.Context, tx storage.StateTransaction) (*Record, error)

+ ReadyForMore func(ctx context.Context, tx storage.StateTransaction) error

+ TriggerKeys func(ctx context.Context, tx storage.StateTransaction, batchSize int) (int, error)

+ UpdateWatermark func(ctx context.Context, tx storage.StateTransaction, watermark time.Time) error

+ type JobOutputQueueIntermediateRecordStore struct

+ func (j *JobOutputQueueIntermediateRecordStore) AddRecord(ctx context.Context, tx storage.StateTransaction, inputIndex int, ...) error

+ func (j *JobOutputQueueIntermediateRecordStore) Close(ctx context.Context, storage storage.Storage) error

+ func (j *JobOutputQueueIntermediateRecordStore) GetWatermark(ctx context.Context, tx storage.StateTransaction) (time.Time, error)

+ func (j *JobOutputQueueIntermediateRecordStore) MarkEndOfStream(ctx context.Context, tx storage.StateTransaction) error

+ func (j *JobOutputQueueIntermediateRecordStore) MarkError(ctx context.Context, tx storage.StateTransaction, err error) error

+ func (j *JobOutputQueueIntermediateRecordStore) Next(ctx context.Context, tx storage.StateTransaction) (*Record, error)

+ func (j *JobOutputQueueIntermediateRecordStore) ReadyForMore(ctx context.Context, tx storage.StateTransaction) error

+ func (j *JobOutputQueueIntermediateRecordStore) TriggerKeys(ctx context.Context, tx storage.StateTransaction, batchSize int) (int, error)

+ func (j *JobOutputQueueIntermediateRecordStore) UpdateWatermark(ctx context.Context, tx storage.StateTransaction, watermark time.Time) error

+ type JoinType int

+ const INNER_JOIN

+ const LEFT_JOIN

+ const OUTER_JOIN

+ func (j JoinType) String() string

+ type JoinedStream struct

+ func (js *JoinedStream) AddRecord(ctx context.Context, tx storage.StateTransaction, inputIndex int, ...) error

+ func (js *JoinedStream) Trigger(ctx context.Context, tx storage.StateTransaction, key octosql.Value) ([]*Record, error)

+ type KeyHashingStrategy struct

+ func (s *KeyHashingStrategy) CalculatePartition(ctx context.Context, record *Record, outputs int) (int, error)

+ type KeyHashingStrategyPrototype struct

+ func (s *KeyHashingStrategyPrototype) Get(ctx context.Context, variables octosql.Variables) (ShuffleStrategy, error)

+ type LookupJoin struct

+ func NewLookupJoin(maxJobsCount int, stateStorage storage.Storage, source Node, joined Node, ...) *LookupJoin

+ func (node *LookupJoin) Get(ctx context.Context, variables octosql.Variables, streamID *StreamID) (RecordStream, *ExecutionOutput, error)

+ type LookupJoinStream struct

+ func (j *LookupJoinStream) TriggerKeys(ctx context.Context, tx storage.StateTransaction, batchSize int) (int, error)

+ func (rs *LookupJoinStream) AddRecord(ctx context.Context, tx storage.StateTransaction, inputIndex int, ...) error

+ func (rs *LookupJoinStream) Close(ctx context.Context, storage storage.Storage) error

+ func (rs *LookupJoinStream) GetNextRecord(ctx context.Context, tx storage.StateTransaction) (*Record, error)

+ func (rs *LookupJoinStream) GetWatermark(ctx context.Context, tx storage.StateTransaction) (time.Time, error)

+ func (rs *LookupJoinStream) HandleControlMessages(ctx context.Context, tx storage.StateTransaction) error

+ func (rs *LookupJoinStream) MarkEndOfStream(ctx context.Context, tx storage.StateTransaction) error

+ func (rs *LookupJoinStream) MarkError(ctx context.Context, tx storage.StateTransaction, err error) error

+ func (rs *LookupJoinStream) Next(ctx context.Context, tx storage.StateTransaction) (*Record, error)

+ func (rs *LookupJoinStream) ReadyForMore(ctx context.Context, tx storage.StateTransaction) error

+ func (rs *LookupJoinStream) RunScheduler(ctx context.Context)

+ func (rs *LookupJoinStream) RunWorker(ctx context.Context, id *RecordID) error

+ func (rs *LookupJoinStream) UpdateWatermark(ctx context.Context, tx storage.StateTransaction, watermark time.Time) error

+ type Metadata struct

+ EventTimeField string

+ Id *RecordID

+ Undo bool

+ XXX_NoUnkeyedLiteral struct{}

+ XXX_sizecache int32

+ XXX_unrecognized []byte

+ func (*Metadata) Descriptor() ([]byte, []int)

+ func (*Metadata) ProtoMessage()

+ func (m *Metadata) GetEventTimeField() string

+ func (m *Metadata) GetId() *RecordID

+ func (m *Metadata) GetUndo() bool

+ func (m *Metadata) Reset()

+ func (m *Metadata) String() string

+ func (m *Metadata) XXX_DiscardUnknown()

+ func (m *Metadata) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

+ func (m *Metadata) XXX_Merge(src proto.Message)

+ func (m *Metadata) XXX_Size() int

+ func (m *Metadata) XXX_Unmarshal(b []byte) error

+ type MultiTrigger struct

+ func NewMultiTrigger(triggers ...TriggerPrototype) *MultiTrigger

+ func (m *MultiTrigger) Get(ctx context.Context, variables octosql.Variables) (Trigger, error)

+ type NextShuffleMetadataChange struct

+ Partition int

+ ShuffleIDAddSuffix string

+ Source Node

+ func NewNextShuffleMetadataChange(shuffleIDAddSuffix string, partition int, source Node) *NextShuffleMetadataChange

+ func (n *NextShuffleMetadataChange) Get(ctx context.Context, variables octosql.Variables, streamID *StreamID) (RecordStream, *ExecutionOutput, error)

+ type OrderByKey struct

+ func NewOrderByKey(key []byte) *OrderByKey

+ func (k *OrderByKey) MonotonicMarshal() []byte

+ func (k *OrderByKey) MonotonicUnmarshal(data []byte) error

+ type OrderByStream struct

+ func (ob *OrderByStream) AddRecord(ctx context.Context, tx storage.StateTransaction, inputIndex int, ...) error

+ func (ob *OrderByStream) Trigger(ctx context.Context, tx storage.StateTransaction, key octosql.Value) ([]*Record, error)

+ type OutputOptions struct

+ Limit Expression

+ Offset Expression

+ OrderByDirections []OrderDirection

+ OrderByExpressions []Expression

+ func NewOutputOptions(orderByExpressions []Expression, orderByDirections []OrderDirection, ...) *OutputOptions

+ type OutputQueue struct

+ func NewOutputQueue(tx storage.StateTransaction) *OutputQueue

+ func (q *OutputQueue) Peek(ctx context.Context, msg proto.Message) error

+ func (q *OutputQueue) Pop(ctx context.Context, msg proto.Message) error

+ func (q *OutputQueue) Push(ctx context.Context, element proto.Message) error

+ type PipelineMetadata struct

+ NextShuffleID *ShuffleID

+ Partition int

+ type ProcessByKey struct

+ func (p *ProcessByKey) AddRecord(ctx context.Context, tx storage.StateTransaction, inputIndex int, ...) error

+ func (p *ProcessByKey) Close(ctx context.Context, storage storage.Storage) error

+ func (p *ProcessByKey) GetWatermark(ctx context.Context, tx storage.StateTransaction) (time.Time, error)

+ func (p *ProcessByKey) MarkEndOfStream(ctx context.Context, tx storage.StateTransaction) error

+ func (p *ProcessByKey) MarkError(ctx context.Context, tx storage.StateTransaction, err error) error

+ func (p *ProcessByKey) Next(ctx context.Context, tx storage.StateTransaction) (*Record, error)

+ func (p *ProcessByKey) ReadyForMore(ctx context.Context, tx storage.StateTransaction) error

+ func (p *ProcessByKey) TriggerKeys(ctx context.Context, tx storage.StateTransaction, batchSize int) (int, error)

+ func (p *ProcessByKey) UpdateWatermark(ctx context.Context, tx storage.StateTransaction, watermark time.Time) error

+ type ProcessFunction interface

+ AddRecord func(ctx context.Context, tx storage.StateTransaction, inputIndex int, ...) error

+ Trigger func(ctx context.Context, tx storage.StateTransaction, key octosql.Value) ([]*Record, error)

+ type PullEngine struct

+ func NewPullEngine(irs IntermediateRecordStore, storage storage.Storage, sources []RecordStream, ...) *PullEngine

+ func (engine *PullEngine) Close(ctx context.Context, storage storage.Storage) error

+ func (engine *PullEngine) GetWatermark(ctx context.Context, tx storage.StateTransaction) (time.Time, error)

+ func (engine *PullEngine) Next(ctx context.Context) (*Record, error)

+ func (engine *PullEngine) Run()

+ type QueueElement struct

+ Type isQueueElement_Type

+ XXX_NoUnkeyedLiteral struct{}

+ XXX_sizecache int32

+ XXX_unrecognized []byte

+ func (*QueueElement) Descriptor() ([]byte, []int)

+ func (*QueueElement) ProtoMessage()

+ func (*QueueElement) XXX_OneofWrappers() []interface{}

+ func (m *QueueElement) GetEndOfStream() bool

+ func (m *QueueElement) GetError() string

+ func (m *QueueElement) GetRecord() *Record

+ func (m *QueueElement) GetType() isQueueElement_Type

+ func (m *QueueElement) GetWatermark() *timestamp.Timestamp

+ func (m *QueueElement) Reset()

+ func (m *QueueElement) String() string

+ func (m *QueueElement) XXX_DiscardUnknown()

+ func (m *QueueElement) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

+ func (m *QueueElement) XXX_Merge(src proto.Message)

+ func (m *QueueElement) XXX_Size() int

+ func (m *QueueElement) XXX_Unmarshal(b []byte) error

+ type QueueElement_EndOfStream struct

+ EndOfStream bool

+ type QueueElement_Error struct

+ Error string

+ type QueueElement_Record struct

+ Record *Record

+ type QueueElement_Watermark struct

+ Watermark *timestamp.Timestamp

type Record

+ Data []*octosql.Value

+ FieldNames []string

+ Metadata *Metadata

+ XXX_NoUnkeyedLiteral struct{}

+ XXX_sizecache int32

+ XXX_unrecognized []byte

+ func NewRecordFromRecord(record *Record, opts ...RecordOption) *Record

+ func ReadAll(ctx context.Context, stateStorage storage.Storage, stream RecordStream) ([]*Record, error)

+ func ReadAllWithCount(ctx context.Context, stateStorage storage.Storage, stream RecordStream, ...) ([]*Record, error)

+ func (*Record) Descriptor() ([]byte, []int)

+ func (*Record) ProtoMessage()

+ func (m *Record) GetData() []*octosql.Value

+ func (m *Record) GetFieldNames() []string

+ func (m *Record) GetMetadata() *Metadata

+ func (m *Record) Reset()

+ func (m *Record) XXX_DiscardUnknown()

+ func (m *Record) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

+ func (m *Record) XXX_Merge(src proto.Message)

+ func (m *Record) XXX_Size() int

+ func (m *Record) XXX_Unmarshal(b []byte) error

+ func (r *Record) EventTimeField() octosql.VariableName

+ func (r *Record) GetVariableNames() []octosql.VariableName

+ func (r *Record) Hash() (uint64, error)

+ func (r *Record) ID() *RecordID

+ func (r *Record) Show() string

+ func (r *Record) ShowFields() []Field

+ type RecordEqualityFunc func(record1 *Record, record2 *Record) error

+ type RecordExpression struct

+ func (re *RecordExpression) ExpressionValue(ctx context.Context, variables octosql.Variables) (octosql.Value, error)

+ type RecordID struct

+ ID string

+ XXX_NoUnkeyedLiteral struct{}

+ XXX_sizecache int32

+ XXX_unrecognized []byte

+ func NewRecordID(id string) *RecordID

+ func NewRecordIDFromStreamIDWithOffset(streamID *StreamID, offset int) *RecordID

+ func (*RecordID) Descriptor() ([]byte, []int)

+ func (*RecordID) ProtoMessage()

+ func (id *RecordID) AsPrefix() []byte

+ func (id *RecordID) MonotonicMarshal() []byte

+ func (id *RecordID) MonotonicUnmarshal(data []byte) error

+ func (id RecordID) Show() string

+ func (m *RecordID) GetID() string

+ func (m *RecordID) Reset()

+ func (m *RecordID) String() string

+ func (m *RecordID) XXX_DiscardUnknown()

+ func (m *RecordID) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

+ func (m *RecordID) XXX_Merge(src proto.Message)

+ func (m *RecordID) XXX_Size() int

+ func (m *RecordID) XXX_Unmarshal(b []byte) error

type RecordOption

+ func WithEventTimeField(field octosql.VariableName) RecordOption

+ func WithID(id *RecordID) RecordOption

+ func WithNoUndo() RecordOption

type RecordStream

+ func GetTestStream(t *testing.T, stateStorage storage.Storage, variables octosql.Variables, ...) RecordStream

+ Close func(ctx context.Context, storage storage.Storage) error

+ type Regexp struct

+ func (rel *Regexp) Apply(ctx context.Context, variables octosql.Variables, left, right Expression) (bool, error)

type Relation

+ func NewRegexp() Relation

+ type Shuffle struct

+ func NewShuffle(outputPartitionCount int, strategyPrototype ShuffleStrategyPrototype, ...) *Shuffle

+ func (s *Shuffle) Get(ctx context.Context, variables octosql.Variables, streamID *StreamID) (RecordStream, *ExecutionOutput, error)

+ func (s *Shuffle) StartSources(ctx context.Context, stateStorage storage.Storage, shuffleID *ShuffleID, ...) (map[string]ShuffleData, error)

+ type ShuffleData struct

+ Shuffle *Shuffle

+ ShuffleID *ShuffleID

+ Variables octosql.Variables

+ type ShuffleID struct

+ Id string

+ XXX_NoUnkeyedLiteral struct{}

+ XXX_sizecache int32

+ XXX_unrecognized []byte

+ func GetSourceShuffleID(tx storage.StateTransaction, inputName octosql.Value) (*ShuffleID, error)

+ func NewShuffleID(id string) *ShuffleID

+ func (*ShuffleID) Descriptor() ([]byte, []int)

+ func (*ShuffleID) ProtoMessage()

+ func (id *ShuffleID) AsMapKey() string

+ func (id *ShuffleID) AsPrefix() []byte

+ func (m *ShuffleID) GetId() string

+ func (m *ShuffleID) Reset()

+ func (m *ShuffleID) String() string

+ func (m *ShuffleID) XXX_DiscardUnknown()

+ func (m *ShuffleID) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

+ func (m *ShuffleID) XXX_Merge(src proto.Message)

+ func (m *ShuffleID) XXX_Size() int

+ func (m *ShuffleID) XXX_Unmarshal(b []byte) error

+ type ShuffleReceiver struct

+ func NewShuffleReceiver(streamID *StreamID, shuffleID *ShuffleID, sourcePartitionCount int, ...) *ShuffleReceiver

+ func (rs *ShuffleReceiver) Close(ctx context.Context, storage storage.Storage) error

+ func (rs *ShuffleReceiver) GetWatermark(ctx context.Context, tx storage.StateTransaction) (time.Time, error)

+ func (rs *ShuffleReceiver) Next(ctx context.Context) (*Record, error)

+ type ShuffleSender struct

+ func NewShuffleSender(streamID *StreamID, shuffleID *ShuffleID, shuffleStrategy ShuffleStrategy, ...) *ShuffleSender

+ func (node *ShuffleSender) AddRecord(ctx context.Context, tx storage.StateTransaction, inputIndex int, ...) error

+ func (node *ShuffleSender) Close(ctx context.Context, storage storage.Storage) error

+ func (node *ShuffleSender) GetWatermark(ctx context.Context, tx storage.StateTransaction) (time.Time, error)

+ func (node *ShuffleSender) MarkEndOfStream(ctx context.Context, tx storage.StateTransaction) error

+ func (node *ShuffleSender) MarkError(ctx context.Context, tx storage.StateTransaction, err error) error

+ func (node *ShuffleSender) Next(ctx context.Context, tx storage.StateTransaction) (*Record, error)

+ func (node *ShuffleSender) ReadyForMore(ctx context.Context, tx storage.StateTransaction) error

+ func (node *ShuffleSender) TriggerKeys(ctx context.Context, tx storage.StateTransaction, batchSize int) (int, error)

+ func (node *ShuffleSender) UpdateWatermark(ctx context.Context, tx storage.StateTransaction, watermark time.Time) error

+ type ShuffleStrategy interface

+ CalculatePartition func(ctx context.Context, record *Record, outputs int) (int, error)

+ func NewConstantStrategy(partition int) ShuffleStrategy

+ func NewKeyHashingStrategy(variables octosql.Variables, key []Expression) ShuffleStrategy

+ type ShuffleStrategyPrototype interface

+ Get func(ctx context.Context, variables octosql.Variables) (ShuffleStrategy, error)

+ func NewConstantStrategyPrototype(partition int) ShuffleStrategyPrototype

+ func NewKeyHashingStrategyPrototype(key []Expression) ShuffleStrategyPrototype

+ type StarExpression struct

+ func NewStarExpression(qualifier string) *StarExpression

+ func (se *StarExpression) ExpressionValue(ctx context.Context, variables octosql.Variables) (octosql.Value, error)

+ func (se *StarExpression) Fields(variables octosql.Variables) []octosql.VariableName

+ func (se *StarExpression) Name() octosql.VariableName

+ type StreamID struct

+ Id string

+ XXX_NoUnkeyedLiteral struct{}

+ XXX_sizecache int32

+ XXX_unrecognized []byte

+ func GetRawStreamID() *StreamID

+ func GetSourceStreamID(tx storage.StateTransaction, inputName octosql.Value) (*StreamID, error)

+ func NewStreamID(str string) *StreamID

+ func (*StreamID) Descriptor() ([]byte, []int)

+ func (*StreamID) ProtoMessage()

+ func (id *StreamID) AsPrefix() []byte

+ func (m *StreamID) GetId() string

+ func (m *StreamID) Reset()

+ func (m *StreamID) String() string

+ func (m *StreamID) XXX_DiscardUnknown()

+ func (m *StreamID) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

+ func (m *StreamID) XXX_Merge(src proto.Message)

+ func (m *StreamID) XXX_Size() int

+ func (m *StreamID) XXX_Unmarshal(b []byte) error

+ type StreamJoin struct

+ func NewStreamJoin(leftSource, rightSource Node, leftKey, rightKey []Expression, ...) *StreamJoin

+ func (node *StreamJoin) Get(ctx context.Context, variables octosql.Variables, streamID *StreamID) (RecordStream, *ExecutionOutput, error)

+ type Task func() error

+ type Trigger interface

+ KeysFired func(ctx context.Context, tx storage.StateTransaction, key []octosql.Value) error

+ PollKeysToFire func(ctx context.Context, tx storage.StateTransaction, batchSize int) ([]octosql.Value, error)

+ RecordReceived func(ctx context.Context, tx storage.StateTransaction, key octosql.Value, ...) error

+ UpdateWatermark func(ctx context.Context, tx storage.StateTransaction, watermark time.Time) error

+ type TriggerPrototype interface

+ Get func(ctx context.Context, variables octosql.Variables) (Trigger, error)

+ type UnionWatermarkGenerator struct

+ func NewUnionWatermarkGenerator(sources []WatermarkSource) *UnionWatermarkGenerator

+ func (uwg *UnionWatermarkGenerator) GetWatermark(ctx context.Context, tx storage.StateTransaction) (time.Time, error)

+ type WatermarkSource interface

+ GetWatermark func(ctx context.Context, tx storage.StateTransaction) (time.Time, error)

+ type WatermarkTrigger struct

+ func NewWatermarkTrigger() *WatermarkTrigger

+ func (c *WatermarkTrigger) Get(ctx context.Context, variables octosql.Variables) (Trigger, error)

+ type ZeroWatermarkGenerator struct

+ func NewZeroWatermarkGenerator() *ZeroWatermarkGenerator

+ func (s *ZeroWatermarkGenerator) GetWatermark(ctx context.Context, tx storage.StateTransaction) (time.Time, error)

v0.2.0

Sep 16, 2019

Changes in this version

type Record

+ func (r *Record) EventTime() octosql.Value

+ func (r *Record) IsUndo() bool

+ func (r *Record) String() string

+ type RecordOption func(stream *Record)

+ func WithEventTime(eventTime time.Time) RecordOption

+ func WithMetadataFrom(base *Record) RecordOption

+ func WithUndo() RecordOption

v0.1.1

Jul 21, 2019

Changes in this version

+ type Joiner struct

+ func NewJoiner(prefetchCount int, variables octosql.Variables, sourceStream RecordStream, ...) *Joiner

+ func (joiner *Joiner) Close() error

+ func (joiner *Joiner) GetNextRecord() (*Record, RecordStream, error)

v0.1.0

Jul 14, 2019

Changes in this version

+ var ErrEndOfStream = errors.New("end of stream")

+ var ErrNotFound = errors.New("not found")

+ func AreStreamsEqual(first, second RecordStream) (bool, error)

+ func AreStreamsEqualNoOrdering(first, second RecordStream) (bool, error)

+ func HashRecord(rec *Record) (uint64, error)

+ func ParseType(str string) octosql.Value

+ type Aggregate interface

+ AddRecord func(key octosql.Tuple, value octosql.Value) error

+ GetAggregated func(key octosql.Tuple) (octosql.Value, error)

+ String func() string

+ type AggregatePrototype func() Aggregate

+ type AliasedExpression struct

+ func NewAliasedExpression(name octosql.VariableName, expr Expression) *AliasedExpression

+ func (alExpr *AliasedExpression) ExpressionValue(variables octosql.Variables) (octosql.Value, error)

+ func (alExpr *AliasedExpression) Name() octosql.VariableName

+ type And struct

+ Left Formula

+ Right Formula

+ func NewAnd(left Formula, right Formula) *And

+ func (f *And) Evaluate(variables octosql.Variables) (bool, error)

+ type Constant struct

+ Value bool

+ func NewConstant(value bool) *Constant

+ func (f Constant) Evaluate(variables octosql.Variables) (bool, error)

+ type Datatype string

+ const DatatypeBoolean

+ const DatatypeFloat64

+ const DatatypeInt

+ const DatatypeString

+ const DatatypeTuple

+ func GetType(i octosql.Value) Datatype

+ type Distinct struct

+ func NewDistinct(child Node) *Distinct

+ func (node *Distinct) Get(variables octosql.Variables) (RecordStream, error)

+ type DistinctStream struct

+ func (ds *DistinctStream) Close() error

+ func (ds *DistinctStream) Next() (*Record, error)

+ type DummyNode struct

+ func NewDummyNode(data []*Record) *DummyNode

+ func (dn *DummyNode) Get(variables octosql.Variables) (RecordStream, error)

+ type DummyValue struct

+ func NewDummyValue(value octosql.Value) *DummyValue

+ func (dv *DummyValue) ExpressionValue(variables octosql.Variables) (octosql.Value, error)

+ type Equal struct

+ func (rel *Equal) Apply(variables octosql.Variables, left, right Expression) (bool, error)

+ type Expression interface

+ ExpressionValue func(variables octosql.Variables) (octosql.Value, error)

+ type Field struct

+ Name octosql.VariableName

+ type Filter struct

+ func NewFilter(formula Formula, child Node) *Filter

+ func (node *Filter) Get(variables octosql.Variables) (RecordStream, error)

+ type FilteredStream struct

+ func (stream *FilteredStream) Close() error

+ func (stream *FilteredStream) Next() (*Record, error)

+ type Formula interface

+ Evaluate func(variables octosql.Variables) (bool, error)

+ type Function struct

+ ArgumentNames [][]string

+ Description docs.Documentation

+ Logic func(...octosql.Value) (octosql.Value, error)

+ Name string

+ Validator Validator

+ func (f *Function) Document() docs.Documentation

+ type FunctionExpression struct

+ func NewFunctionExpression(fun *Function, args []Expression) *FunctionExpression

+ func (fe *FunctionExpression) ExpressionValue(variables octosql.Variables) (octosql.Value, error)

+ type GreaterEqual struct

+ func (rel *GreaterEqual) Apply(variables octosql.Variables, left, right Expression) (bool, error)

+ type GroupBy struct

+ func NewGroupBy(source Node, key []Expression, fields []octosql.VariableName, ...) *GroupBy

+ func (node *GroupBy) Get(variables octosql.Variables) (RecordStream, error)

+ type GroupByStream struct

+ func (stream *GroupByStream) Close() error

+ func (stream *GroupByStream) Next() (*Record, error)

+ type HashMap struct

+ func NewHashMap() *HashMap

+ func (hm *HashMap) Get(key octosql.Value) (interface{}, bool, error)

+ func (hm *HashMap) GetIterator() *Iterator

+ func (hm *HashMap) Set(key octosql.Value, value interface{}) error

+ type In struct

+ func (rel *In) Apply(variables octosql.Variables, left, right Expression) (bool, error)

+ type InMemoryStream struct

+ func NewInMemoryStream(data []*Record) *InMemoryStream

+ func (ims *InMemoryStream) Close() error

+ func (ims *InMemoryStream) Next() (*Record, error)

+ type InnerJoin struct

+ func NewInnerJoin(source Node, joined Node) *InnerJoin

+ func (node *InnerJoin) Get(variables octosql.Variables) (RecordStream, error)

+ type InnerJoinedStream struct

+ func (stream *InnerJoinedStream) Close() error

+ func (stream *InnerJoinedStream) Next() (*Record, error)

+ type Iterator struct

+ func (iter *Iterator) Next() (octosql.Value, interface{}, bool)

+ type LeftJoin struct

+ func NewLeftJoin(source Node, joined Node) *LeftJoin

+ func (node *LeftJoin) Get(variables octosql.Variables) (RecordStream, error)

+ type LeftJoinedStream struct

+ func (stream *LeftJoinedStream) Close() error

+ func (stream *LeftJoinedStream) Next() (*Record, error)

+ type LessEqual struct

+ func (rel *LessEqual) Apply(variables octosql.Variables, left, right Expression) (bool, error)

+ type LessThan struct

+ func (rel *LessThan) Apply(variables octosql.Variables, left, right Expression) (bool, error)

+ type Like struct

+ func (rel *Like) Apply(variables octosql.Variables, left, right Expression) (bool, error)

+ type Limit struct

+ func NewLimit(data Node, limit Expression) *Limit

+ func (node *Limit) Get(variables octosql.Variables) (RecordStream, error)

+ type LimitedStream struct

+ func (node *LimitedStream) Close() error

+ func (node *LimitedStream) Next() (*Record, error)

+ type LogicExpression struct

+ func NewLogicExpression(formula Formula) *LogicExpression

+ func (le *LogicExpression) ExpressionValue(variables octosql.Variables) (octosql.Value, error)

+ type Map struct

+ func NewMap(expressions []NamedExpression, child Node, keep bool) *Map

+ func (node *Map) Get(variables octosql.Variables) (RecordStream, error)

+ type MappedStream struct

+ func (stream *MappedStream) Close() error

+ func (stream *MappedStream) Next() (*Record, error)

+ type MoreThan struct

+ func (rel *MoreThan) Apply(variables octosql.Variables, left, right Expression) (bool, error)

+ type NamedExpression interface

+ Name func() octosql.VariableName

+ type Node interface

+ Get func(variables octosql.Variables) (RecordStream, error)

+ type NodeExpression struct

+ func NewNodeExpression(node Node) *NodeExpression

+ func (ne *NodeExpression) ExpressionValue(variables octosql.Variables) (octosql.Value, error)

+ type Not struct

+ Child Formula

+ func NewNot(child Formula) *Not

+ func (f *Not) Evaluate(variables octosql.Variables) (bool, error)

+ type NotEqual struct

+ func (rel *NotEqual) Apply(variables octosql.Variables, left, right Expression) (bool, error)

+ type NotIn struct

+ func (rel *NotIn) Apply(variables octosql.Variables, left, right Expression) (bool, error)

+ type Offset struct

+ func NewOffset(data Node, offsetExpr Expression) *Offset

+ func (node *Offset) Get(variables octosql.Variables) (RecordStream, error)

+ type Or struct

+ Left Formula

+ Right Formula

+ func NewOr(left Formula, right Formula) *Or

+ func (f *Or) Evaluate(variables octosql.Variables) (bool, error)

+ type OrderBy struct

+ func NewOrderBy(exprs []Expression, directions []OrderDirection, source Node) *OrderBy

+ func (ob *OrderBy) Get(variables octosql.Variables) (RecordStream, error)

+ type OrderDirection string

+ const Ascending

+ const Descending

+ type Predicate struct

+ Left Expression

+ Relation Relation

+ Right Expression

+ func NewPredicate(left Expression, relation Relation, right Expression) *Predicate

+ func (f *Predicate) Evaluate(variables octosql.Variables) (bool, error)

+ type Record struct

+ func NewRecord(fields []octosql.VariableName, data map[octosql.VariableName]octosql.Value) *Record

+ func NewRecordFromSlice(fields []octosql.VariableName, data []octosql.Value) *Record

+ func NewRecordFromSliceWithNormalize(fields []octosql.VariableName, data []interface{}) *Record

+ func Normalize(rec *Record) *Record

+ func (r *Record) AsTuple() octosql.Tuple

+ func (r *Record) AsVariables() octosql.Variables

+ func (r *Record) Equal(other *Record) bool

+ func (r *Record) Fields() []Field

+ func (r *Record) Value(field octosql.VariableName) octosql.Value

+ type RecordStream interface

+ Next func() (*Record, error)

+ type Relation interface

+ Apply func(variables octosql.Variables, left, right Expression) (bool, error)

+ func NewEqual() Relation

+ func NewGreaterEqual() Relation

+ func NewIn() Relation

+ func NewLessEqual() Relation

+ func NewLessThan() Relation

+ func NewLike() Relation

+ func NewMoreThan() Relation

+ func NewNotEqual() Relation

+ func NewNotIn() Relation

+ type RequalifiedStream struct

+ func (stream *RequalifiedStream) Close() error

+ func (stream *RequalifiedStream) Next() (*Record, error)

+ type Requalifier struct

+ func NewRequalifier(qualifier string, child Node) *Requalifier

+ func (node *Requalifier) Get(variables octosql.Variables) (RecordStream, error)

+ type TupleExpression struct

+ func NewTuple(expressions []Expression) *TupleExpression

+ func (tup *TupleExpression) ExpressionValue(variables octosql.Variables) (octosql.Value, error)

+ type UnifiedStream struct

+ func (node *UnifiedStream) Close() error

+ func (node *UnifiedStream) Next() (*Record, error)

+ type UnionAll struct

+ func NewUnionAll(first, second Node) *UnionAll

+ func (node *UnionAll) Get(variables octosql.Variables) (RecordStream, error)

+ type Validator interface

+ Validate func(args ...octosql.Value) error

+ type Variable struct

+ func NewVariable(name octosql.VariableName) *Variable

+ func (v *Variable) ExpressionValue(variables octosql.Variables) (octosql.Value, error)

+ func (v *Variable) Name() octosql.VariableName

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL