table

type CollectTable ¶

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

Collect is a special table that it doesn't do any computation, but simply stores everything it receives.

func NewCollectTable(ctx context.Context, spec TableSpec) (table *CollectTable, err error)

func (tbl *CollectTable) Emit(datum saw.Datum) (err error)

func (tbl *CollectTable) Result(ctx context.Context) (interface{}, error)

type InspectCallback ¶

type InspectCallback func(key saw.DatumKey, item saw.Saw) error

type Inspectable ¶

type Inspectable interface {
	Inspect(key saw.DatumKey, callback InspectCallback) (inspected int, err error)
	InspectSet(keys []saw.DatumKey, callback InspectCallback, concurrent bool) (inspected int, err error)
	InspectAll(callback InspectCallback, concurrent bool) (inspected int, err error)
}

Inspectable tables allows a callback to inspect one, a set of, or all saws in the table, table gurantees no concurrent Emit() for the saw being inspected, but other items not being currently being inspected could still receive Emit() concurrently, that means you cannot get a consistent snapshot of table using InspectAll().

InspectSet() and InspectAll() can inspect items in parallel when concurrent=true. All inspect functions return number of items been inspected without error, and one of the error callback returned if there's any, inspect functions stops as soon as it encounter error.

type KeyHashFunc ¶

type KeyHashFunc func(saw.DatumKey) int

type MemTable ¶

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

MemTable manages a set (spec.NumShards) of SimpleTables, provides concurrent safe Emit(), stores finaly result when Result() called if there is a spec.PersistentResource setting.

func NewMemTable(spec TableSpec) *MemTable

func (tbl *MemTable) Emit(kv saw.Datum) error

func (tbl *MemTable) Inspect(key saw.DatumKey, callback InspectCallback) (int, error)

func (tbl *MemTable) InspectAll(callback InspectCallback, concurrent bool) (int, error)

func (tbl *MemTable) InspectSet(
	keys []saw.DatumKey, callback InspectCallback, concurrent bool) (int, error)

func (tbl *MemTable) Result(ctx context.Context) (interface{}, error)

type SimpleTable ¶

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

SimpleTable is a in-memory, non-storable memory table, concurrent non-safe table --- it assumes you call all its methods sequentially. Good for handling small set of data in mini-batch --- aggregate stats for a single user session etc.

func NewSimpleTable(spec TableSpec) *SimpleTable

func (tbl *SimpleTable) Emit(kv saw.Datum) (err error)

func (tbl *SimpleTable) Inspect(key saw.DatumKey, callback InspectCallback) (int, error)

func (tbl *SimpleTable) InspectAll(callback InspectCallback, concurrent bool) (int, error)

func (tbl *SimpleTable) InspectSet(
	keys []saw.DatumKey, callback InspectCallback, concurrent bool) (int, error)

func (tbl *SimpleTable) Result(ctx context.Context) (interface{}, error)

type TableItemFactory ¶

type TableItemFactory func(tableName string, key saw.DatumKey) (saw.Saw, error)

func ItemFactoryOf(example saw.Saw) TableItemFactory

type TableResultMap ¶

type TableResultMap map[saw.DatumKey]interface{}

SimpleTable and MemTable result type

type TableSpec ¶

type TableSpec struct {
	Name        string
	ItemFactory TableItemFactory

	// KeyHashFunc assigns incoming Datum to one of its shard, defaults to fnv32
	KeyHashFunc KeyHashFunc
	// Defaults to 127
	NumShards int

	// When not empty, table state will be stored at external storage.
	PersistentResource storage.ResourceSpec
	// It depends on table type to determine what data get persistent and what
	// encoder to use. Defaults to verbatim (accepts and stores []byte)
	ValueEncoder saw.ValueEncoder
	// Implementation may pre-allocate and reuse buffer for encoding values, to avoid
	// frequent malloc, defaults to 4096
	ValueEncodeBufferSize int
}

TableSpec is shared configuration of Table implementations in this package.