libqpu

type APIClient interface {
	Query(QueryRequest) (ResponseStream, error)
	QueryUnary(string) (*qpuextapi.QueryResp, error)
	QuerySQL(string, map[string]string, bool) (ResponseStream, error)
	QuerySubscribe(string) (qpuapi.QPUAPI_QuerySubscribeClient, context.CancelFunc, error)
	CloseConnection() error
	GetConfig() (*qpuapi.ConfigResponse, error)
}

type APIProcessor interface {
	Query(QueryRequest, RequestStream) error
	QueryUnary(QueryRequest, opentracing.Span) (*qpuextapi.QueryResp, error)
	QuerySubscribe(*qpuextapi.QueryReq, qpuapi.QPUAPI_QuerySubscribeServer) error
	GetConfig(context.Context, *qpuapi.ConfigRequest) (*qpuapi.ConfigResponse, error)
}

type ASTQuery struct {
	Q *qpuapi.ASTQuery
}

type AdjacentQPU struct {
	APIClient           APIClient
	OutputSchema        []string
	MeasureDataTransfer bool
}

type AggregationType int

type DatastoreAttributeType int

type DatastoreType int

type DownstreamQueryConf struct {
	IsNull    []string
	IsNotNull []string
}

type LogOperation struct {
	Op   *qpu.LogOperation
	InTs time.Time
}

type OperatorType int

type QPU struct {
	Class        QPUClass
	Config       *QPUConfig
	InputSchema  Schema
	AdjacentQPUs []*AdjacentQPU
	State        QPUState
}

type QPUClass interface {
	ClientQuery(ASTQuery, string, opentracing.Span) (*qpuextapi.QueryResp, error)
	ProcessQuerySnapshot(ASTQuery, map[string]string, bool, opentracing.Span) (<-chan LogOperation, <-chan error)
	ProcessQuerySubscribe(ASTQuery, map[string]string, bool) (int, <-chan LogOperation, <-chan error)
	QuerySubscribe(ASTQuery, *qpuextapi.QueryReq) (chan LogOperation, chan bool, chan error)
	RemovePersistentQuery(string, int)
	GetConfig() *qpuapi.ConfigResponse
}

type QPUConfig struct {
	Operator     OperatorType
	State        StateType
	Port         string
	Connections  []QPUConnection
	StateBackend struct {
		Endpoint    string
		Credentials struct {
			AccessKeyID     string
			SecretAccessKey string
		}
	}
	DatastoreConfig struct {
		Type              DatastoreType
		Endpoint          string
		LogStreamEndpoint string
		DBName            string
		Credentials       struct {
			AccessKeyID     string
			SecretAccessKey string
		}
	}
	IndexConfig struct {
		Table         string
		AttributeName string
		AttributeType DatastoreAttributeType
		LBound        *qpu.Value
		UBound        *qpu.Value
	}
	AggregationConfig struct {
		AggregationFunc      AggregationType
		GroupBy              string
		AggregationAttribute string
	}
	JoinConfig struct {
		OutputTableAlias     string
		JoinedAttributeAlias string
		JoinAttribute        map[string]string
	}
	CacheConfig struct {
		Size int
		TTL  int
	}
	Evaluation struct {
		Tracing                    bool
		LogTimestamps              bool
		MeasureNotificationLatency bool
		MeasureDataTransfer        bool
	}
	ProcessingConfig struct {
		API struct {
			MaxWorkers  int
			MaxJobQueue int
		}
		Input struct {
			MaxWorkers  int
			MaxJobQueue int
		}
		GrpcServers int
	}
}

type QPUConnection struct {
	Address             string
	MeasureDataTransfer bool
}

type QPUService interface {
	Start() error
	Shutdown()
}

type QPUState interface {
	Init(string, string, string) error
	Insert(string, map[string]interface{}, map[string]*timestamp.Timestamp) error
	Update(string, map[string]interface{}, map[string]interface{}, map[string]*timestamp.Timestamp) error
	Get(string, []string, []string, string, int64, opentracing.Span) (<-chan map[string]interface{}, error)
	GetRow(string, []string, []string, opentracing.Span) *sql.Row
	Cleanup()
}

type QueryLogEntry struct {
	RowIDs []string
	Ts     time.Time
}

type QueryRequest struct {
	Req *qpuapi.QueryRequest
}

type QueryType int

type RequestStream struct {
	Stream qpuapi.QPUAPI_QueryServer
}

type ResponseRecord struct {
	Rec  *qpuapi.ResponseStreamRecord
	InTs time.Time
}

type ResponseRecordType int

type ResponseStream struct {
	Stream     qpuapi.QPUAPI_QueryClient
	CancelFunc context.CancelFunc
}

type Schema map[string]SchemaTable

type SchemaTable struct {
	Attributes      map[string]DatastoreAttributeType
	DownstreamQuery DownstreamQueryConf
}

type StateType int

type WriteLogEntry struct {
	RowID string
	T0    time.Time
	T1    time.Time
}