stream.scotty.slicing.aggregationstore

Interface AggregationStore<InputType>

All Known Implementing Classes:

LazyAggregateStore

public interface AggregationStore<InputType>

Method Summary

Modifier and Type	Method and Description
void	addSlice(int index, Slice newSlice) Add a new Slice at a specific index
void	aggregate(WindowManager.AggregationWindowCollector aggregateWindows, long minTs, long maxTs) Generates the window aggregates.
void	appendSlice(Slice<InputType,?> newSlice) Appends a new slice as new current slice
int	findSliceByEnd(long pre)
int	findSliceIndexByTimestamp(long ts) Lookup for a slice which contains this timestamp.
Slice<InputType,?>	getCurrentSlice()
Slice<InputType,?>	getSlice(int index) Returns slice for a given index or @IndexOutOfBoundsException
void	insertValueToCurrentSlice(InputType element, long ts) Insert the element to the current slice.
void	insertValueToSlice(int index, InputType element, long ts)
boolean	isEmpty()
void	mergeSlice(int sliceIndex) Merging two slices A and B happens in three steps: 1.
int	size()

Method Detail

getCurrentSlice

Slice<InputType,?> getCurrentSlice()

Returns:

get the newest slice.

findSliceIndexByTimestamp

int findSliceIndexByTimestamp(long ts)

Lookup for a slice which contains this timestamp. The timestamp is >= slice start < slice end.

Parameters:

ts - element timestamp

Returns:

index of containing slice.

getSlice

Slice<InputType,?> getSlice(int index)

Returns slice for a given index or @IndexOutOfBoundsException

Parameters:

index - >= 0 < size

Returns:

@Slice

insertValueToCurrentSlice

void insertValueToCurrentSlice(InputType element,
                               long ts)

Insert the element to the current slice.

Parameters:

element - the element

ts - timestamp of record

insertValueToSlice

void insertValueToSlice(int index,
                        InputType element,
                        long ts)

appendSlice

void appendSlice(Slice<InputType,?> newSlice)

Appends a new slice as new current slice

Parameters:

newSlice -

size

int size()

isEmpty

boolean isEmpty()

Returns:

true if AggregationStore contains no slices

aggregate

void aggregate(WindowManager.AggregationWindowCollector aggregateWindows,
               long minTs,
               long maxTs)

Generates the window aggregates. On every @AggregateWindowState the aggregated value is set.

Parameters:

aggregateWindows - definition of the requested window

minTs - startTimestamp of the earliest window.

maxTs - endTimestamp of the latest window

addSlice

void addSlice(int index,
              Slice newSlice)

Add a new Slice at a specific index

Parameters:

index -

newSlice -

mergeSlice

void mergeSlice(int sliceIndex)

Merging two slices A and B happens in three steps: 1. Update the end of A such that t end (A) ← t end (B). 2. Update the aggregate of A such that a ← a ⊕ b. 3. Delete slice B, which is now merged into A.

Parameters:

sliceIndex - index of slice A

findSliceByEnd

int findSliceByEnd(long pre)