query_cache

README

Caching

Overview

The plugin query cache maintains a bidirectional GRPC stream connection to the cache server (hosted by the plugin manager process)

Cache data is streamed row by row, in both directions.

Cache Set

• call CacheCommand_SET_RESULT_START command
• for each row:
  • stream to FDW
  • stream to cache with CacheCommand_SET_RESULT_ITERATE command
• mark completion with CacheCommand_SET_RESULT_END
• fetch index bucket for this table (if it exists): CacheCommand_GET_INDEX
• update index bucket and write back CacheCommand_SET_INDEX

Cache Get

• fetch index bucket for this table (if it exists): CacheCommand_GET_INDEX
• if this is a cache hit (i.e. an index item satisfying columns and quals exists), call CacheCommand_GET_RESULT
• cache will stream results to us a row at a time - for each row
  • stream to fdw

Concurrency

The plugin may make more than one request to the cache at a time (for example, if a query consists of multiple scans with the same connection, i.e. multiple concurrent plugin Execution calls).

Therefore results from the cache may be interleaved, e.g. the results from 2 different cache gets may be streaming at the same time

To handle this, each cache command is made using a CallId (unique time-based identifier). When making a cache command, the query cache registers a callback which is stored in a map keyed by call id.

When the cache stream receiver processes a response from the cache, it uses the response callId to find the appropriate callback function and invokes it.

TTL

The TTL is controlled by the client, as multiple clients might be using the same connections and each may have a different TTL.

This is achieved as follows. The cache saves data with a default TTL of 5 minutes (this is the TTL passed to the underlying cache). When fetching data, the age of the retrieved data is determined using the insertion time. If the age of the data is older than the client TTL, it is considered a cache miss

NOTE: If the client TTL is greater than the default TTL of the cache, the default TTL is increased as appropriate to ensure the data remains in the cache at lkeast as long as is needed for the client TTL

Documentation

Overview

Package query_cache is a cache used to store query results.

Each plugin process has a single QueryCache instance which stores the data for all connections

Index

• Constants
• func IsCacheMiss(err error) bool
• func NewPendingIndexItem(req *CacheRequest) *pendingIndexItem
• type CacheData
• type CacheMissError
  • func (CacheMissError) Error() string
• type CacheRequest
• type CacheStats
• type IndexBucket
  • func IndexBucketfromProto(b *proto.IndexBucket) *IndexBucket
  • func (b *IndexBucket) Append(item *IndexItem) *IndexBucket
  • func (b *IndexBucket) AsProto() *proto.IndexBucket
  • func (b *IndexBucket) Get(req *CacheRequest, keyColumns map[string]*proto.KeyColumn) *IndexItem
• type IndexItem
  • func NewIndexItem(req *CacheRequest) *IndexItem
  • func (i IndexItem) SatisfiedByRequest(req *CacheRequest, keyColumns map[string]*proto.KeyColumn) bool
  • func (i IndexItem) SatisfiesRequest(columns []string, limit int64, qualMap map[string]*proto.Quals, ...) bool
• type QueryCache
  • func NewQueryCache(pluginName string, pluginSchemaMap map[string]*grpc.PluginSchema, ...) (*QueryCache, error)
  • func (c *QueryCache) AbortSet(ctx context.Context, callId string, err error)
  • func (c *QueryCache) ClearForConnection(ctx context.Context, connectionName string)
  • func (c *QueryCache) EndSet(ctx context.Context, callId string) (err error)
  • func (c *QueryCache) Get(ctx context.Context, req *CacheRequest, streamRowFunc func(row *sdkproto.Row)) error
  • func (c *QueryCache) IterateSet(ctx context.Context, row *sdkproto.Row, callId string) error
  • func (c *QueryCache) StartSet(_ context.Context, req *CacheRequest)

Constants

const (
	// Key column CacheMatch values
	CacheMatchSubset = "subset"
	CacheMatchExact  = "exact"
)

Functions

func IsCacheMiss

func IsCacheMiss(err error) bool

func NewPendingIndexItem

func NewPendingIndexItem(req *CacheRequest) *pendingIndexItem

Types

type CacheData

type CacheData interface {
	proto.Message
	*sdkproto.QueryResult | *sdkproto.IndexBucket
}

type CacheMissError

type CacheMissError struct{}

func (CacheMissError) Error

func (CacheMissError) Error() string

type CacheRequest

type CacheRequest struct {
	CallId         string
	Table          string
	QualMap        map[string]*sdkproto.Quals
	Columns        []string
	Limit          int64
	ConnectionName string
	TtlSeconds     int64
	// contains filtered or unexported fields
}

type CacheStats

type CacheStats struct {
	// keep count of hits and misses
	Hits   int
	Misses int
}

type IndexBucket

type IndexBucket struct {
	Items []*IndexItem
}

IndexBucket contains index items for all cache results for a given table and connection

func IndexBucketfromProto

func IndexBucketfromProto(b *proto.IndexBucket) *IndexBucket

func (*IndexBucket) Append

func (b *IndexBucket) Append(item *IndexItem) *IndexBucket

func (*IndexBucket) AsProto

func (b *IndexBucket) AsProto() *proto.IndexBucket

func (*IndexBucket) Get

func (b *IndexBucket) Get(req *CacheRequest, keyColumns map[string]*proto.KeyColumn) *IndexItem

Get finds an index item which satisfies all columns

type IndexItem

type IndexItem struct {
	Columns       []string
	Key           string
	Limit         int64
	Quals         map[string]*proto.Quals
	InsertionTime time.Time
	PageCount     int64
}

IndexItem stores the columns and cached index for a single cached query result note - this index item it tied to a specific table and set of quals

func NewIndexItem

func NewIndexItem(req *CacheRequest) *IndexItem

func (IndexItem) SatisfiedByRequest

func (i IndexItem) SatisfiedByRequest(req *CacheRequest, keyColumns map[string]*proto.KeyColumn) bool

func (IndexItem) SatisfiesRequest

func (i IndexItem) SatisfiesRequest(columns []string, limit int64, qualMap map[string]*proto.Quals, keyColumns map[string]*proto.KeyColumn) bool

type QueryCache

type QueryCache struct {
	Stats *CacheStats

	// map of connection name to plugin schema
	PluginSchemaMap map[string]*grpc.PluginSchema
	// contains filtered or unexported fields
}

func NewQueryCache

func NewQueryCache(pluginName string, pluginSchemaMap map[string]*grpc.PluginSchema, maxCacheStorageMb int) (*QueryCache, error)

func (*QueryCache) AbortSet

func (c *QueryCache) AbortSet(ctx context.Context, callId string, err error)

func (*QueryCache) ClearForConnection

func (c *QueryCache) ClearForConnection(ctx context.Context, connectionName string)

ClearForConnection removes all cache entries for the given connection

func (*QueryCache) EndSet

func (c *QueryCache) EndSet(ctx context.Context, callId string) (err error)

func (*QueryCache) Get

func (c *QueryCache) Get(ctx context.Context, req *CacheRequest, streamRowFunc func(row *sdkproto.Row)) error

func (*QueryCache) IterateSet

func (c *QueryCache) IterateSet(ctx context.Context, row *sdkproto.Row, callId string) error

func (*QueryCache) StartSet

func (c *QueryCache) StartSet(_ context.Context, req *CacheRequest)

StartSet begins a streaming cache Set operation. NOTE: this mutates req