duck

package
v0.11.813 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Nov 9, 2026 License: Apache-2.0 Imports: 51 Imported by: 0

Documentation

Index

Constants

View Source 
const (

	// QuackDriverVersion pins the DuckDB ADBC driver used for Quack connections.
	// Quack needs DuckDB 2.0+, while the default driver stays on a release
	// MotherDuck supports, so the two are installed side by side.
	QuackDriverVersion = "2.0.0"
)

Variables

This section is empty.

Functions

func ADBCDriverName added in v0.11.367 

func ADBCDriverName() string

ADBCDriverName returns the registered SQL driver name for ADBC DuckDB.

func BuildIngestrLakehouseURI added in v0.11.604 

func BuildIngestrLakehouseURI(lh *config.LakehouseConfig) string

Serialises a LakehouseConfig into a URI that ingestr's matching destination parses on the other side.

func EnsureADBCDriverInstalled added in v0.11.367 

func EnsureADBCDriverInstalled(ctx context.Context) error

func EnsureQuackADBCDriverInstalled added in v0.11.810 

func EnsureQuackADBCDriverInstalled(ctx context.Context) error

EnsureQuackADBCDriverInstalled installs the DuckDB driver used by Quack connections into its own directory, leaving the default driver untouched.

func IngestrTypeHints added in v0.11.789 

func IngestrTypeHints() map[string]string

IngestrTypeHints maps DuckDB type names whose ingestr defaults lose width or add a time zone.

func InstallDrivers added in v0.11.810 

func InstallDrivers(ctx context.Context) error

InstallDrivers installs both DuckDB drivers ahead of use, so images can include them instead of downloading them on every run. The quack extension is left to the first Quack connection, so only runs that use Quack download it.

func IsQuackURI added in v0.11.810 

func IsQuackURI(path string) bool

IsQuackURI reports whether the URI points at a Quack server, which serves many clients at once and so needs no local lock.

func LockDatabase added in v0.11.88 

func LockDatabase(path string)

func LockDatabases added in v0.11.782 

func LockDatabases(paths ...string) func()

LockDatabases acquires source/destination files once, in a consistent order, including when different URIs refer to the same file.

func NewColumnCheckOperator 

func NewColumnCheckOperator(manager config.ConnectionGetter) *ansisql.ColumnCheckOperator

func NewMaterializer 

func NewMaterializer(fullRefresh bool) *pipeline.Materializer

func QuackADBCDriverDir added in v0.11.810 

func QuackADBCDriverDir() (string, error)

QuackADBCDriverDir returns the directory holding the Quack DuckDB driver.

func SuspendConnectionReuse added in v0.11.782 

func SuspendConnectionReuse(ctx context.Context) func()

SuspendConnectionReuse releases cached file locks for arbitrary Python/R code. SQL continues with ephemeral connections until every overlapping script exits. Scripts must still declare dependencies when sharing files with other work.

func UnlockDatabase added in v0.11.88 

func UnlockDatabase(path string)

func ValidateLakehouseConfig added in v0.11.442 

func ValidateLakehouseConfig(lh *config.LakehouseConfig) error

Validates the lakehouse config for DuckDB-specific requirements.

func WithConnectionReuse added in v0.11.782 

func WithConnectionReuse(ctx context.Context) (context.Context, func())

WithConnectionReuse keeps DuckDB instances (including lakehouse attachments) alive for a pipeline run. Cleanup cancels the scope and closes idle databases; active sessions are closed asynchronously once their workers release them. This preserves bounded CLI shutdown when a worker ignores cancellation. Outside this scope, connections remain ephemeral.

Types

type AcceptedValuesCheck 

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

func (*AcceptedValuesCheck) Check 

func (c *AcceptedValuesCheck) Check(ctx context.Context, ti *scheduler.ColumnCheckInstance) error

type BasicOperator 

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

func NewBasicOperator 

func NewBasicOperator(conn config.ConnectionGetter, extractor query.QueryExtractor, materializer materializer, parser *sqlparser.SQLParser) *BasicOperator

func (BasicOperator) Run 

func (o BasicOperator) Run(ctx context.Context, ti scheduler.TaskInstance) error

func (BasicOperator) RunTask 

func (o BasicOperator) RunTask(ctx context.Context, p *pipeline.Pipeline, t *pipeline.Asset) error

type Client 

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

func NewClient 

func NewClient(c DuckDBConfig) (*Client, error)

func (*Client) Close added in v0.11.367 

func (c *Client) Close()

Close is a no-op: sessions close per operation; WithConnectionReuse owns database cleanup for a run, not individual assets using this shared client.

func (*Client) CreateSchemaIfNotExist added in v0.11.209 

func (c *Client) CreateSchemaIfNotExist(ctx context.Context, asset *pipeline.Asset) error

func (*Client) DryRunQuery added in v0.11.475 

func (c *Client) DryRunQuery(ctx context.Context, q *query.Query) (*query.DryRunResult, error)

func (*Client) GetColumns added in v0.11.247 

func (c *Client) GetColumns(ctx context.Context, databaseName, tableName string) ([]*ansisql.DBColumn, error)

func (*Client) GetDBConnectionURI added in v0.11.90 

func (c *Client) GetDBConnectionURI() (string, error)

func (*Client) GetDatabaseSummary added in v0.11.242 

func (c *Client) GetDatabaseSummary(ctx context.Context) (*ansisql.DBDatabase, error)

func (*Client) GetDatabases added in v0.11.247 

func (c *Client) GetDatabases(ctx context.Context) ([]string, error)

func (*Client) GetIngestrURI 

func (c *Client) GetIngestrURI() (string, error)

func (*Client) GetTableSummary added in v0.11.224 

func (c *Client) GetTableSummary(ctx context.Context, tableName string, schemaOnly bool) (*diff.TableSummaryResult, error)

func (*Client) GetTables added in v0.11.247 

func (c *Client) GetTables(ctx context.Context, databaseName string) ([]string, error)

func (*Client) HasQuack added in v0.11.806 

func (c *Client) HasQuack() bool

func (*Client) IngestrTypeHints added in v0.11.789 

func (c *Client) IngestrTypeHints() map[string]string

func (*Client) RunQueryWithoutResult 

func (c *Client) RunQueryWithoutResult(ctx context.Context, query *query.Query) error

func (*Client) Select 

func (c *Client) Select(ctx context.Context, query *query.Query) ([][]interface{}, error)

Select runs a query and returns the results.

func (*Client) SelectArrow added in v0.11.777 

func (c *Client) SelectArrow(ctx context.Context, q *query.Query, maxRows int) (arrow.RecordBatch, error)

SelectArrow returns an owned record batch without the JSON-oriented conversions in SelectWithSchema. The caller must Release it. maxRows bounds buffered rows.

func (*Client) SelectWithSchema added in v0.11.103 

func (c *Client) SelectWithSchema(ctx context.Context, queryObject *query.Query) (*query.QueryResult, error)

func (*Client) StreamWithSchema added in v0.11.806 

func (c *Client) StreamWithSchema(ctx context.Context, queryObject *query.Query, h query.RowHandler) (*query.QueryResult, error)

type ColumnType added in v0.11.550 

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

ColumnType holds column metadata for result processing.

func (*ColumnType) DatabaseTypeName added in v0.11.550 

func (ct *ColumnType) DatabaseTypeName() string

func (*ColumnType) DecimalSize added in v0.11.550 

func (ct *ColumnType) DecimalSize() (precision, scale int64, ok bool)

func (*ColumnType) Name added in v0.11.550 

func (ct *ColumnType) Name() string

type Config 

type Config struct {
	Path       string
	ReadOnly   bool
	Lakehouse  *config.LakehouseConfig
	Quack      bool
	Host       string
	Port       int
	Token      string
	DisableSSL bool

	MemoryLimit            string
	Threads                int
	TempDirectory          string
	MaxTempDirectorySize   string
	PreserveInsertionOrder *bool
}

func (Config) GetIngestrURI 

func (c Config) GetIngestrURI() string

func (Config) GetLakehouseAlias added in v0.11.442 

func (c Config) GetLakehouseAlias() string

func (Config) HasLakehouse added in v0.11.442 

func (c Config) HasLakehouse() bool

func (Config) HasQuack added in v0.11.804 

func (c Config) HasQuack() bool

func (Config) ToDBConnectionURI 

func (c Config) ToDBConnectionURI() string

ToDBConnectionURI returns a connection URI to be used with the pgx package.

type DuckDBClient 

type DuckDBClient interface {
	RunQueryWithoutResult(ctx context.Context, query *query.Query) error
	Select(ctx context.Context, query *query.Query) ([][]interface{}, error)
	SelectWithSchema(ctx context.Context, queryObj *query.Query) (*query.QueryResult, error)
	CreateSchemaIfNotExist(ctx context.Context, asset *pipeline.Asset) error
	Close()
}

type DuckDBConfig 

type DuckDBConfig interface {
	ToDBConnectionURI() string
	GetIngestrURI() string
}

type DuckDBSchemaCreator added in v0.11.332 

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

func NewDuckDBSchemaCreator added in v0.11.332 

func NewDuckDBSchemaCreator() *DuckDBSchemaCreator

func (*DuckDBSchemaCreator) CreateSchemaIfNotExist added in v0.11.332 

func (sc *DuckDBSchemaCreator) CreateSchemaIfNotExist(ctx context.Context, qr queryRunner, asset *pipeline.Asset) error

type EphemeralConnection added in v0.11.88 

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

EphemeralConnection uses the ADBC low-level API to query DuckDB directly, bypassing the database/sql adapter which doesn't support complex Arrow types (LIST, STRUCT, MAP). During a pipeline run, sessions share a cached database.

func NewEphemeralConnection added in v0.11.88 

func NewEphemeralConnection(c DuckDBConfig) (*EphemeralConnection, error)

func (*EphemeralConnection) ExecContext added in v0.11.88 

func (e *EphemeralConnection) ExecContext(ctx context.Context, sqlStr string, arguments ...any) (sql.Result, error)

func (*EphemeralConnection) QueryContext added in v0.11.88 

func (e *EphemeralConnection) QueryContext(ctx context.Context, queryStr string, args ...any) (Rows, error)

func (*EphemeralConnection) QueryRowContext added in v0.11.224 

func (e *EphemeralConnection) QueryRowContext(ctx context.Context, queryStr string, args ...any) Row

func (*EphemeralConnection) QueryStreamContext added in v0.11.806 

func (e *EphemeralConnection) QueryStreamContext(ctx context.Context, queryStr string, args ...any) (Rows, error)

QueryStreamContext reads record batches lazily instead of buffering them; the database stays locked until the rows are closed.

type LakehouseAttacher added in v0.11.442 

type LakehouseAttacher struct{}

func NewLakehouseAttacher added in v0.11.442 

func NewLakehouseAttacher() *LakehouseAttacher

func (*LakehouseAttacher) GenerateAttachStatements added in v0.11.442 

func (l *LakehouseAttacher) GenerateAttachStatements(lh *config.LakehouseConfig, alias string) ([]string, error)

type MotherDuckConfig added in v0.11.273 

type MotherDuckConfig struct {
	ReadOnly bool
	Token    string
	Database string
}

func (MotherDuckConfig) GetIngestrURI added in v0.11.273 

func (c MotherDuckConfig) GetIngestrURI() string

func (MotherDuckConfig) ToDBConnectionURI added in v0.11.273 

func (c MotherDuckConfig) ToDBConnectionURI() string

type Mutex added in v0.11.88 

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

Mutex is the mutex with synchronized map, it allows reducing unnecessary locks among different keys. This implementation comes from the mapmutex package, I simply copied it here instead of adding it as a dependency. See the code here: https://github.com/EagleChen/mapmutex/blob/master/mutex.go

func NewCustomizedMapMutex added in v0.11.88 

func NewCustomizedMapMutex(mRetry int, mDelay, bDelay, factor, jitter float64) *Mutex

NewCustomizedMapMutex returns a customized mapmutex.

func NewMapMutex added in v0.11.88 

func NewMapMutex() *Mutex

NewMapMutex returns a mapmutex with default configs.

func (*Mutex) TryLock added in v0.11.88 

func (m *Mutex) TryLock(key interface{}) bool

TryLock tries to acquire the lock.

func (*Mutex) Unlock added in v0.11.88 

func (m *Mutex) Unlock(key interface{})

Unlock unlocks for the key please call Unlock only after having acquired the lock.

type PatternCheck 

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

func (*PatternCheck) Check 

func (c *PatternCheck) Check(ctx context.Context, ti *scheduler.ColumnCheckInstance) error

type Row added in v0.11.367 

type Row interface {
	Scan(dest ...any) error
	Err() error
}

Row interface abstracts sql.Row to allow custom implementations.

type Rows added in v0.11.367 

type Rows interface {
	Close() error
	Columns() ([]string, error)
	ColumnTypes() ([]*ColumnType, error)
	Err() error
	Next() bool
	Scan(dest ...any) error
}

Rows interface abstracts sql.Rows to allow custom implementations that manage connection lifecycle.

type TransactionConflictError added in v0.11.782 

type TransactionConflictError struct {
	Err error
}

TransactionConflictError identifies a transaction that can be retried after a competing writer completes. Bruin does not replay arbitrary SQL batches: statements before the failed transaction may already have committed.

func (*TransactionConflictError) Error added in v0.11.782 

func (e *TransactionConflictError) Error() string

func (*TransactionConflictError) Unwrap added in v0.11.782 

func (e *TransactionConflictError) Unwrap() error

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.