libsq

Documentation

Overview

Package libsq implements the core sq functionality. The ExecuteSLQ function is the entrypoint for executing a SLQ query, which may interact with several data sources. The QuerySQL function executes a SQL query against a single source. Both functions ultimately send their result records to a RecordWriter. Implementations of RecordWriter write records to a destination, such as a JSON or CSV file. The NewDBWriter function returns a RecordWriter that writes records to a database.

Index

• func ExecuteSLQ(ctx context.Context, log lg.Log, dbOpener driver.DatabaseOpener, ...) error
• func QuerySQL(ctx context.Context, log lg.Log, dbase driver.Database, recw RecordWriter, ...) error
• type DBWriter
  • func NewDBWriter(log lg.Log, destDB driver.Database, destTbl string, recChSize int, ...) *DBWriter
  • func (w *DBWriter) Open(ctx context.Context, cancelFn context.CancelFunc, recMeta sqlz.RecordMeta) (chan<- sqlz.Record, <-chan error, error)
  • func (w *DBWriter) Wait() (written int64, err error)
• type DBWriterPreWriteHook
  • func DBWriterCreateTableIfNotExistsHook(destTblName string) DBWriterPreWriteHook
• type RecordWriter

Functions

func ExecuteSLQ

func ExecuteSLQ(ctx context.Context, log lg.Log, dbOpener driver.DatabaseOpener, joinDBOpener driver.JoinDatabaseOpener,
	srcs *source.Set, query string, recw RecordWriter,
) error

ExecuteSLQ executes the slq query, writing the results to recw. The caller is responsible for closing dbases.

func QuerySQL

func QuerySQL(ctx context.Context, log lg.Log, dbase driver.Database, recw RecordWriter, query string,
	args ...any,
) error

QuerySQL executes the SQL query against dbase, writing the results to recw. Note that QuerySQL may return before recw has finished writing, thus the caller may wish to wait for recw to complete. The caller is responsible for closing dbase.

Types

type DBWriter

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

DBWriter implements RecordWriter, writing records to a database table.

func NewDBWriter

func NewDBWriter(log lg.Log, destDB driver.Database, destTbl string, recChSize int,
	preWriteHooks ...DBWriterPreWriteHook,
) *DBWriter

NewDBWriter returns a new writer than implements RecordWriter. The writer writes records from recordCh to destTbl in destDB. The recChSize param controls the size of recordCh returned by the writer's Open method.

func (*DBWriter) Open

func (w *DBWriter) Open(ctx context.Context, cancelFn context.CancelFunc, recMeta sqlz.RecordMeta) (chan<- sqlz.Record,
	<-chan error, error,
)

Open implements RecordWriter.

func (*DBWriter) Wait

func (w *DBWriter) Wait() (written int64, err error)

Wait implements RecordWriter.

type DBWriterPreWriteHook

type DBWriterPreWriteHook func(ctx context.Context, recMeta sqlz.RecordMeta, destDB driver.Database, tx sqlz.DB) error

DBWriterPreWriteHook is a function that is invoked before DBWriter begins writing.

func DBWriterCreateTableIfNotExistsHook

func DBWriterCreateTableIfNotExistsHook(destTblName string) DBWriterPreWriteHook

DBWriterCreateTableIfNotExistsHook returns a hook that creates destTblName if it does not exist.

type RecordWriter

type RecordWriter interface {
	// Open opens the RecordWriter for writing records described
	// by recMeta, returning a non-nil err if the initial open
	// operation fails.
	//
	// Records received on recCh are written to the
	// destination, possibly buffered, until recCh is closed.
	// Therefore, the caller must close recCh to indicate that
	// all records have been sent, so that the writer can
	// perform any end-of-stream actions. The caller can use the Wait
	// method to wait for the writer to complete. The returned errCh
	// will also be closed when complete.
	//
	// Any underlying write error is sent on errCh,
	// at which point the writer is defunct. Thus it is the
	// responsibility of the sender to check errCh before
	// sending again on recordCh. Note that implementations
	// may send more than one error on errCh, and that errCh
	// will be closed when the writer completes. Note also that the
	// errors sent on errCh are accumulated internally by the writer and
	// returned from the Wait method (if more than one error, they
	// may be combined into a multierr).
	//
	// The caller can stop the RecordWriter by cancelling ctx.
	// When ctx is done, the writer shuts down processing
	// of recCh and returns ctx.Err on errCh  (possibly
	// with additional errors from the shutdown).
	//
	// If cancelFn is non-nil, it is invoked only by the writer's Wait method.
	// If the Open method itself returns an error, it is the caller's
	// responsibility to invoke cancelFn to prevent resource leakage.
	//
	// It is noted that the existence of the cancelFn param is an unusual
	// construction. This mechanism exists to enable a goroutine to wait
	// on the writer outside the function that invoked Open, without
	// having to pass cancelFn around.
	Open(ctx context.Context, cancelFn context.CancelFunc,
		recMeta sqlz.RecordMeta) (recCh chan<- sqlz.Record, errCh <-chan error, err error)

	// Wait waits for the writer to complete and returns the number of
	// written rows and any error (which may be a multierr).
	// The written value may be non-zero even in the presence of an error.
	// If a cancelFn was passed to Open, it will be invoked before Wait returns.
	Wait() (written int64, err error)
}

RecordWriter is the interface for writing records to a destination. The Open method returns a channel to which the records are sent. The Wait method allows the sender to wait for the writer to complete.