stream.scotty.slicing.aggregationstore

Class LazyAggregateStore<InputType>

java.lang.Object

aggregationstore.stream.scotty.slicing.LazyAggregateStore<InputType>

Type Parameters:

InputType -

All Implemented Interfaces:

AggregationStore<InputType>

public class LazyAggregateStore<InputType>
extends java.lang.Object
implements AggregationStore<InputType>

A lazy in memory aggregation store implementation. All slices are stored in a plain java array.

Field Summary

Fields

Modifier and Type	Field and Description
private java.util.List<Slice<InputType,?>>	slices

Constructor Summary

Constructors

Constructor and Description
LazyAggregateStore()

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 start)
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()

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

slices

private final java.util.List<Slice<InputType,?>> slices

Constructor Detail

LazyAggregateStore

public LazyAggregateStore()

Method Detail

getCurrentSlice

public Slice<InputType,?> getCurrentSlice()

Specified by:

getCurrentSlice in interface AggregationStore<InputType>

Returns:

get the newest slice.

findSliceIndexByTimestamp

public int findSliceIndexByTimestamp(long ts)

Description copied from interface: AggregationStore

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

Specified by:

findSliceIndexByTimestamp in interface AggregationStore<InputType>

Parameters:

ts - element timestamp

Returns:

index of containing slice.

getSlice

public Slice<InputType,?> getSlice(int index)

Description copied from interface: AggregationStore

Returns slice for a given index or @IndexOutOfBoundsException

Specified by:

getSlice in interface AggregationStore<InputType>

Parameters:

index - >= 0 < size

Returns:

@Slice

insertValueToCurrentSlice

public void insertValueToCurrentSlice(InputType element,
                                      long ts)

Description copied from interface: AggregationStore

Insert the element to the current slice.

Specified by:

insertValueToCurrentSlice in interface AggregationStore<InputType>

Parameters:

element - the element

ts - timestamp of record

insertValueToSlice

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

Specified by:

insertValueToSlice in interface AggregationStore<InputType>

appendSlice

public void appendSlice(Slice<InputType,?> newSlice)

Description copied from interface: AggregationStore

Appends a new slice as new current slice

Specified by:

appendSlice in interface AggregationStore<InputType>

size

public int size()

Specified by:

size in interface AggregationStore<InputType>

isEmpty

public boolean isEmpty()

Specified by:

isEmpty in interface AggregationStore<InputType>

Returns:

true if AggregationStore contains no slices

aggregate

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

Description copied from interface: AggregationStore

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

Specified by:

aggregate in interface AggregationStore<InputType>

Parameters:

aggregateWindows - definition of the requested window

minTs - startTimestamp of the earliest window.

maxTs - endTimestamp of the latest window

addSlice

public void addSlice(int index,
                     Slice newSlice)

Description copied from interface: AggregationStore

Add a new Slice at a specific index

Specified by:

addSlice in interface AggregationStore<InputType>

mergeSlice

public void mergeSlice(int sliceIndex)

Description copied from interface: AggregationStore

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.

Specified by:

mergeSlice in interface AggregationStore<InputType>

Parameters:

sliceIndex - index of slice A

findSliceByEnd

public int findSliceByEnd(long start)

Specified by:

findSliceByEnd in interface AggregationStore<InputType>