drivers

Documentation

Index

• Variables
• func Register(name string, driver Driver)
• type CatalogEntry
  • func (e *CatalogEntry) GetMetricsView() *runtimev1.MetricsView
  • func (e *CatalogEntry) GetModel() *runtimev1.Model
  • func (e *CatalogEntry) GetSource() *runtimev1.Source
  • func (e *CatalogEntry) GetTable() *runtimev1.Table
• type CatalogStore
• type Connection
  • func Open(driver, dsn string, logger *zap.Logger) (Connection, error)
• type Dialect
  • func (d Dialect) String() string
• type Driver
• type InformationSchema
• type IngestionSummary
• type Instance
  • func (i *Instance) ResolveVariables() map[string]string
• type OLAPStore
• type ObjectType
• type RegistryStore
• type RepoObjectStat
• type RepoStore
• type Result
  • func (r *Result) Close() error
  • func (r *Result) SetCleanupFunc(fn func() error)
• type Statement
• type Table
• type WithConnectionFunc

Variables

var Drivers = make(map[string]Driver)

Drivers is a registry of drivers.

var ErrFileAlreadyExists = errors.New("file already exists")

var ErrNotFound = errors.New("driver: not found")

ErrNotFound indicates the resource wasn't found.

var ErrUnsupportedConnector = errors.New("drivers: connector not supported")

ErrUnsupportedConnector is returned from Ingest for unsupported connectors.

Functions

func Register

func Register(name string, driver Driver)

Register registers a new driver.

Types

type CatalogEntry

type CatalogEntry struct {
	Name          string
	Type          ObjectType
	Object        proto.Message
	Path          string
	Embedded      bool
	BytesIngested int64
	Parents       []string
	Children      []string
	CreatedOn     time.Time
	UpdatedOn     time.Time
	RefreshedOn   time.Time
}

CatalogEntry represents one object in the catalog, such as a source.

func (*CatalogEntry) GetMetricsView

func (e *CatalogEntry) GetMetricsView() *runtimev1.MetricsView

func (*CatalogEntry) GetModel

func (e *CatalogEntry) GetModel() *runtimev1.Model

func (*CatalogEntry) GetSource

func (e *CatalogEntry) GetSource() *runtimev1.Source

func (*CatalogEntry) GetTable

func (e *CatalogEntry) GetTable() *runtimev1.Table

type CatalogStore

type CatalogStore interface {
	FindEntries(ctx context.Context, instanceID string, t ObjectType) ([]*CatalogEntry, error)
	FindEntry(ctx context.Context, instanceID string, name string) (*CatalogEntry, error)
	CreateEntry(ctx context.Context, instanceID string, entry *CatalogEntry) error
	UpdateEntry(ctx context.Context, instanceID string, entry *CatalogEntry) error
	DeleteEntry(ctx context.Context, instanceID string, name string) error
	DeleteEntries(ctx context.Context, instanceID string) error
}

CatalogStore is implemented by drivers capable of storing catalog info for a specific instance.

type Connection

type Connection interface {
	// Migrate prepares the connection for use. It will be called before the connection is first used.
	// (Not to be confused with migrating artifacts, which is handled by the runtime and tracked in the catalog.)
	Migrate(ctx context.Context) error

	// MigrationStatus returns the connection's current and desired migration version (if applicable)
	MigrationStatus(ctx context.Context) (current int, desired int, err error)

	// Close closes the connection
	Close() error

	// RegistryStore returns a RegistryStore if the driver can serve as such, otherwise returns false.
	// The registry is responsible for tracking instances and repos.
	RegistryStore() (RegistryStore, bool)

	// CatalogStore returns a CatalogStore if the driver can serve as such, otherwise returns false.
	// A catalog is used to store state about migrated/deployed objects (such as sources and metrics views).
	CatalogStore() (CatalogStore, bool)

	// RepoStore returns a RepoStore if the driver can serve as such, otherwise returns false.
	// A repo stores file artifacts (either in a folder or virtualized in a database).
	RepoStore() (RepoStore, bool)

	// OLAPStore returns an OLAPStore if the driver can serve as such, otherwise returns false.
	// OLAP stores are where we actually store, transform, and query users' data.
	OLAPStore() (OLAPStore, bool)
}

Connection represents a connection to an underlying DB. It should implement one or more of RegistryStore, CatalogStore, RepoStore, and OLAPStore.

func Open

func Open(driver, dsn string, logger *zap.Logger) (Connection, error)

Open opens a new connection

type Dialect

type Dialect int

Dialect enumerates OLAP query languages.

const (
	DialectUnspecified Dialect = iota
	DialectDuckDB
	DialectDruid
)

func (Dialect) String

func (d Dialect) String() string

type Driver

type Driver interface {
	Open(dsn string, logger *zap.Logger) (Connection, error)
}

Driver represents an underlying DB.

type InformationSchema

type InformationSchema interface {
	All(ctx context.Context) ([]*Table, error)
	Lookup(ctx context.Context, name string) (*Table, error)
}

InformationSchema contains information about existing tables in an OLAP driver.

type IngestionSummary

type IngestionSummary struct {
	BytesIngested int64
}

IngestionSummary is details about ingestion

type Instance

type Instance struct {
	// Identifier
	ID string
	// Driver to connect to for OLAP (options: duckdb, druid)
	OLAPDriver string
	// DSN for connection to OLAP
	OLAPDSN string
	// Driver for reading/editing code artifacts (options: file, metastore)
	RepoDriver string
	// DSN for connecting to repo
	RepoDSN string
	// EmbedCatalog tells the runtime to store the instance's catalog in its OLAP store instead
	// of in the runtime's metadata store. Currently only supported for the duckdb driver.
	EmbedCatalog bool `db:"embed_catalog"`
	// CreatedOn is when the instance was created
	CreatedOn time.Time `db:"created_on"`
	// UpdatedOn is when the instance was last updated in the registry
	UpdatedOn time.Time `db:"updated_on"`
	// Variables contains user-provided variables
	Variables map[string]string `db:"variables"`
	// ProjectVariables contains default variables from rill.yaml
	// (NOTE: This can always be reproduced from rill.yaml, so it's really just a handy cache of the values.)
	ProjectVariables map[string]string `db:"project_variables"`
	// IngestionLimitBytes is total data allowed to ingest across all sources
	// 0 means there is no limit
	IngestionLimitBytes int64 `db:"ingestion_limit_bytes"`
}

Instance represents a single data project, meaning one OLAP connection, one repo connection, and one catalog connection.

func (*Instance) ResolveVariables

func (i *Instance) ResolveVariables() map[string]string

ResolveVariables returns the final resolved variables

type OLAPStore

type OLAPStore interface {
	Dialect() Dialect
	WithConnection(ctx context.Context, priority int, fn WithConnectionFunc) error
	Exec(ctx context.Context, stmt *Statement) error
	Execute(ctx context.Context, stmt *Statement) (*Result, error)
	Ingest(ctx context.Context, env *connectors.Env, source *connectors.Source) (*IngestionSummary, error)
	InformationSchema() InformationSchema
	DropDB() error
}

OLAPStore is implemented by drivers that are capable of storing, transforming and serving analytical queries.

type ObjectType

type ObjectType int

Constants representing the kinds of catalog objects.

const (
	ObjectTypeUnspecified ObjectType = 0
	ObjectTypeTable       ObjectType = 1
	ObjectTypeSource      ObjectType = 2
	ObjectTypeModel       ObjectType = 3
	ObjectTypeMetricsView ObjectType = 4
)

type RegistryStore

type RegistryStore interface {
	FindInstances(ctx context.Context) ([]*Instance, error)
	FindInstance(ctx context.Context, id string) (*Instance, error)
	CreateInstance(ctx context.Context, instance *Instance) error
	DeleteInstance(ctx context.Context, id string) error
	EditInstance(ctx context.Context, instance *Instance) error
}

RegistryStore is implemented by drivers capable of storing and looking up instances and repos.

type RepoObjectStat

type RepoObjectStat struct {
	LastUpdated time.Time
}

type RepoStore

type RepoStore interface {
	Driver() string
	// Root returns directory where artifacts are stored.
	Root() string
	ListRecursive(ctx context.Context, instID string, glob string) ([]string, error)
	Get(ctx context.Context, instID string, path string) (string, error)
	Stat(ctx context.Context, instID string, path string) (*RepoObjectStat, error)
	Put(ctx context.Context, instID string, path string, reader io.Reader) error
	Rename(ctx context.Context, instID string, fromPath string, toPath string) error
	Delete(ctx context.Context, instID string, path string) error
	Sync(ctx context.Context, instID string) error
}

RepoStore is implemented by drivers capable of storing code artifacts. It mirrors a file system, but may be virtualized by a database for non-local deployments.

type Result

type Result struct {
	*sqlx.Rows
	Schema *runtimev1.StructType
	// contains filtered or unexported fields
}

Result wraps the results of query.

func (*Result) Close

func (r *Result) Close() error

Close wraps rows.Close and calls the Result's cleanup function (if it is set). Close should be idempotent.

func (*Result) SetCleanupFunc

func (r *Result) SetCleanupFunc(fn func() error)

SetCleanupFunc sets a function, which will be called when the Result is closed.

type Statement

type Statement struct {
	Query    string
	Args     []any
	DryRun   bool
	Priority int
}

Statement wraps a query to execute against an OLAP driver.

type Table

type Table struct {
	Database       string
	DatabaseSchema string
	Name           string
	Schema         *runtimev1.StructType
}

Table represents a table in an information schema.

type WithConnectionFunc

type WithConnectionFunc func(wrappedCtx context.Context, ensuredCtx context.Context) error

WithConnectionFunc is a callback function that provides a context to be used in further OLAP store calls to enforce affinity to a single connection. It's called with two contexts: wrappedCtx wraps the input context (including cancellation), and ensuredCtx wraps a background context (ensuring it can never be cancelled).