compute

Documentation

Index

• type CompileManager
  • func NewCompileManager(computeClients *client.ComputeClients, ...) CompileManager
• type ComputeManager
  • func NewComputeManager(errGroup *errgroup.Group, compileManager *CompileManager, ...) ComputeManager
• type FileManager
  • func NewFileManager(computeClients *client.ComputeClients) FileManager
• type MaterializationManager
  • func NewMaterializationManager(compileManager *CompileManager, computeClients *client.ComputeClients, ...) MaterializationManager
• type PrepareManager
  • func NewPrepareManager(computeClients *client.ComputeClients, ...) PrepareManager
• type QueryContext
  • func GetNewQueryContext(ctx context.Context, owner *ent.Owner, changedSinceTime *timestamppb.Timestamp, ...) (*QueryContext, context.CancelFunc)
  • func (qc *QueryContext) Cancelled() bool
  • func (qc *QueryContext) GetComputeTables() []*v1alpha.ComputeTable
  • func (qc *QueryContext) GetSlices() []*internal.SliceInfo
  • func (qc *QueryContext) GetTableIDs() []uuid.UUID
• type QueryResult

Types

type CompileManager

type CompileManager interface {
	CompileEntMaterialization(ctx context.Context, owner *ent.Owner, materialization *ent.Materialization) (*v1alpha.CompileResponse, []*v1alpha.View, error)
	CompileV1Materialization(ctx context.Context, owner *ent.Owner, materialization *v1alpha.Materialization, isStreamBased bool) (*v1alpha.CompileResponse, []*v1alpha.View, error)
	CompileV1Query(ctx context.Context, owner *ent.Owner, query *v1alpha.Query, queryOptions *v1alpha.QueryOptions) (*v1alpha.CompileResponse, []*v1alpha.View, error)
	CompileV2Query(ctx context.Context, owner *ent.Owner, expression string, views []*v2alpha.QueryView, queryConfig *v2alpha.QueryConfig) (*v1alpha.CompileResponse, []*v1alpha.View, error)
	CompileV1View(ctx context.Context, owner *ent.Owner, view *v1alpha.View) (*v1alpha.CompileResponse, error)
}

func NewCompileManager

func NewCompileManager(computeClients *client.ComputeClients, kaskadaTableClient *internal.KaskadaTableClient, kaskadaViewClient *internal.KaskadaViewClient) CompileManager

type ComputeManager

type ComputeManager interface {
	CompileManager

	// execute related
	GetOutputURI(owner *ent.Owner, planHash []byte) string
	InitiateQuery(queryContext *QueryContext) (client.ComputeServiceClient, v1alpha.ComputeService_ExecuteClient, error)
	SaveComputeSnapshots(queryContext *QueryContext, computeSnapshots []*v1alpha.ComputeSnapshot)

	// Runs all existing file-based materializations for the given owner
	// Note: this exists in the ComputeManager interface instead of the MaterializationManager interface because
	// it runs materializations in a similar way to InitiateQuery
	RunMaterializations(ctx context.Context, owner *ent.Owner)
}

func NewComputeManager

func NewComputeManager(errGroup *errgroup.Group, compileManager *CompileManager, computeClients *client.ComputeClients, dataTokenClient *internal.DataTokenClient, kaskadaTableClient *internal.KaskadaTableClient, materializationClient *internal.MaterializationClient, objectStoreClient *client.ObjectStoreClient, prepareManager *PrepareManager) ComputeManager

NewComputeManager creates a new compute manager

type FileManager

type FileManager interface {

	// GetFileSchema returns the schema of the file at the given URI
	GetFileSchema(ctx context.Context, fileInput internal.FileInput) (*v1alpha.Schema, error)

	// GetPulsarSchema returns the schema of the pulsar topic
	GetPulsarSchema(ctx context.Context, pulsarConfig *v1alpha.PulsarConfig) (*v1alpha.Schema, error)
}

func NewFileManager

func NewFileManager(computeClients *client.ComputeClients) FileManager

type MaterializationManager

type MaterializationManager interface {
	CompileManager

	// StartMaterialization starts a materialization on the compute backend
	StartMaterialization(ctx context.Context, owner *ent.Owner, materializationID string, compileResp *v1alpha.CompileResponse, destination *v1alpha.Destination) error

	// StopMaterialization stops a materialization on the compute backend
	StopMaterialization(ctx context.Context, materializationID string) error

	// GetMaterializationStatus gets the status of a materialization on the compute backend
	GetMaterializationStatus(ctx context.Context, materializationID string) (*v1alpha.GetMaterializationStatusResponse, error)

	// ReconcileMaterializations reconciles the materializations in the database with the materializations on the compute backend
	ReconcileMaterializations(ctx context.Context) error
}

func NewMaterializationManager

func NewMaterializationManager(compileManager *CompileManager, computeClients *client.ComputeClients, kaskadaTableClient *internal.KaskadaTableClient, materializationClient *internal.MaterializationClient) MaterializationManager

type PrepareManager

type PrepareManager interface {
	PrepareTablesForCompute(ctx context.Context, owner *ent.Owner, dataToken *ent.DataToken, slicePlans []*v1alpha.SlicePlan) (map[uuid.UUID]*internal.SliceTable, error)
	GetPrepareCacheBuster(ctx context.Context) (*int32, error)
}

func NewPrepareManager

func NewPrepareManager(computeClients *client.ComputeClients, kaskadaTableClient *internal.KaskadaTableClient, objectStore *client.ObjectStoreClient, prepareJobClient *internal.PrepareJobClient, parallelizeConfig *utils.ParallelizeConfig, tableStore *store.TableStore) PrepareManager

type QueryContext

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

func GetNewQueryContext

func GetNewQueryContext(ctx context.Context, owner *ent.Owner, changedSinceTime *timestamppb.Timestamp, compileResp *v1alpha.CompileResponse, dataToken *ent.DataToken, finalResultTime *timestamppb.Timestamp, isCurrentDataToken bool, limits *v1alpha.ExecuteRequest_Limits, destination *v1alpha.Destination, sliceRequest *v1alpha.SliceRequest, sliceTableMap map[uuid.UUID]*internal.SliceTable) (*QueryContext, context.CancelFunc)

func (*QueryContext) Cancelled

func (qc *QueryContext) Cancelled() bool

func (*QueryContext) GetComputeTables

func (qc *QueryContext) GetComputeTables() []*v1alpha.ComputeTable

func (*QueryContext) GetSlices

func (qc *QueryContext) GetSlices() []*internal.SliceInfo

func (*QueryContext) GetTableIDs

func (qc *QueryContext) GetTableIDs() []uuid.UUID

type QueryResult

type QueryResult struct {
	DataTokenId string
	Paths       []string
}