Documentation
¶
Overview ¶
Package sql provides SQL executor capabilities for PostgreSQL and SQLite databases.
Index ¶
- func Contains(slice []string, item string) bool
- func ConvertNamedToPositional(query string, params map[string]any, placeholder string) (string, []any, error)
- func ConvertPositionalParams(query string, params []any, placeholder string) ([]any, error)
- func DetectFormat(path string) string
- func ExtractParamNames(query string) []string
- func ParseConflictTarget(target string) []string
- func PrepareParams(query string, cfg *Config, driver Driver) (string, []any, error)
- func QuoteIdentifier(name string) string
- func RegisterDriver(driver Driver)
- func SanitizeIdentifier(identifier string) (string, error)
- func ScanRow(rows *sql.Rows, columns []string) ([]any, error)
- func ValidateParams(query string, params map[string]any) error
- func WithPoolManager(ctx context.Context, pm *GlobalPoolManager) context.Context
- type CSVReader
- type CSVWriter
- type Config
- type ConnectionManager
- type Driver
- type DriverRegistry
- type ExecutionMetrics
- type GlobalPoolConfig
- type GlobalPoolManager
- type ImportConfig
- type ImportMetrics
- type Importer
- type InputOptions
- type InputReader
- type JSONLReader
- type JSONLWriter
- type JSONWriter
- type QueryExecutor
- type ResultWriter
- type Transaction
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func ConvertNamedToPositional ¶
func ConvertNamedToPositional(query string, params map[string]any, placeholder string) (string, []any, error)
ConvertNamedToPositional converts named parameters (:param) to positional parameters ($1, $2, ...). Returns the converted query and ordered parameter values.
func ConvertPositionalParams ¶
ConvertPositionalParams validates and returns positional parameters. Ensures the query has the correct number of placeholders.
func DetectFormat ¶
DetectFormat attempts to detect the format from a file path.
func ExtractParamNames ¶
ExtractParamNames extracts parameter names from a query with named parameters. Returns the names in the order they appear.
func ParseConflictTarget ¶
ParseConflictTarget extracts column names from a conflict target string. Handles both single column "id" and composite "(user_id, org_id)" formats.
func PrepareParams ¶
PrepareParams prepares parameters for query execution. Handles both named and positional parameters.
func QuoteIdentifier ¶
QuoteIdentifier quotes a table or column name to handle reserved words and special characters. This is a shared implementation for SQL databases that use double quotes (PostgreSQL, SQLite). Any existing double quotes in the identifier are escaped by doubling them.
func RegisterDriver ¶
func RegisterDriver(driver Driver)
RegisterDriver registers a driver in the global registry.
func SanitizeIdentifier ¶
SanitizeIdentifier sanitizes a SQL identifier (table/column name) to prevent injection. Only allows alphanumeric characters, underscores, and dots (for schema.table notation).
func ValidateParams ¶
ValidateParams checks if all required parameters are provided.
func WithPoolManager ¶
func WithPoolManager(ctx context.Context, pm *GlobalPoolManager) context.Context
WithPoolManager returns a context with the global pool manager.
Types ¶
type CSVReader ¶
type CSVReader struct {
// contains filtered or unexported fields
}
CSVReader implements InputReader for CSV/TSV formatted input.
func NewCSVReader ¶
func NewCSVReader(r io.Reader, opts InputOptions) *CSVReader
NewCSVReader creates a new CSV/TSV reader.
func (*CSVReader) Close ¶
Close is a no-op for CSVReader (underlying reader should be closed by caller).
func (*CSVReader) ReadHeader ¶
ReadHeader reads the header row from CSV.
type CSVWriter ¶
type CSVWriter struct {
// contains filtered or unexported fields
}
CSVWriter writes results in CSV format.
func NewCSVWriter ¶
NewCSVWriter creates a new CSV writer.
func (*CSVWriter) WriteHeader ¶
WriteHeader writes the CSV header row if enabled.
type Config ¶
type Config struct {
// DSN is the data source name for database connection.
// Format depends on the driver:
// - PostgreSQL: "postgres://user:pass@host:port/dbname?sslmode=disable"
// - SQLite: "file:./data.db?mode=rw" or ":memory:"
DSN string `mapstructure:"dsn"`
// Parameterized queries (SQL injection prevention)
// Can be map[string]any for named params or []any for positional params
Params any `mapstructure:"params"`
// Execution settings
Timeout int `mapstructure:"timeout"` // Query timeout in seconds (default: 60)
Transaction bool `mapstructure:"transaction"` // Wrap execution in transaction
IsolationLevel string `mapstructure:"isolation_level"` // Transaction isolation level
// Locking
AdvisoryLock string `mapstructure:"advisory_lock"` // Named advisory lock (PostgreSQL)
FileLock bool `mapstructure:"file_lock"` // Use file locking (SQLite)
SharedMemory bool `mapstructure:"shared_memory"` // Enable shared cache for :memory: databases (SQLite)
// Output settings
OutputFormat string `mapstructure:"output_format"` // jsonl (default), json, csv
Headers bool `mapstructure:"headers"` // Include headers in CSV output
NullString string `mapstructure:"null_string"` // String representation for NULL values
// Large result handling
MaxRows int `mapstructure:"max_rows"` // Maximum rows to return (0 = unlimited)
Streaming bool `mapstructure:"streaming"` // Stream results to file
OutputFile string `mapstructure:"output_file"` // File path for streaming output
// Import settings
Import *ImportConfig `mapstructure:"import"` // Import data from file
}
Config represents the SQL executor configuration.
func DefaultConfig ¶
func DefaultConfig() *Config
DefaultConfig returns a Config with default values.
func ParseConfig ¶
ParseConfig parses the executor configuration from a map.
func (*Config) GetNamedParams ¶
GetNamedParams returns params as a map if they are named parameters.
func (*Config) GetPositionalParams ¶
GetPositionalParams returns params as a slice if they are positional parameters.
type ConnectionManager ¶
type ConnectionManager struct {
// contains filtered or unexported fields
}
ConnectionManager manages database connections with pooling.
func NewConnectionManager ¶
func NewConnectionManager(ctx context.Context, driver Driver, cfg *Config) (*ConnectionManager, error)
NewConnectionManager creates a new connection manager.
func (*ConnectionManager) Acquire ¶
func (m *ConnectionManager) Acquire()
Acquire increments the reference count.
func (*ConnectionManager) Close ¶
func (m *ConnectionManager) Close() error
Close closes the connection manager.
func (*ConnectionManager) Config ¶
func (m *ConnectionManager) Config() *Config
Config returns the configuration.
func (*ConnectionManager) DB ¶
func (m *ConnectionManager) DB() *sql.DB
DB returns the underlying database connection.
func (*ConnectionManager) Driver ¶
func (m *ConnectionManager) Driver() Driver
Driver returns the database driver.
func (*ConnectionManager) Release ¶
func (m *ConnectionManager) Release() error
Release decrements the reference count and closes if zero.
type Driver ¶
type Driver interface {
// Name returns the driver identifier (e.g., "postgres", "sqlite")
Name() string
// Connect establishes a connection to the database using the provided configuration.
// Returns a *sql.DB instance and any error encountered.
Connect(ctx context.Context, cfg *Config) (*sql.DB, func() error, error)
// SupportsAdvisoryLock indicates if the driver supports advisory locking.
SupportsAdvisoryLock() bool
// AcquireAdvisoryLock acquires a named advisory lock (PostgreSQL-specific).
// Returns a release function and any error.
AcquireAdvisoryLock(ctx context.Context, db *sql.DB, lockName string) (func() error, error)
// ConvertNamedParams converts named parameters (:param) to driver-specific format ($1, ?).
// Returns the converted query and ordered parameter values.
ConvertNamedParams(query string, params map[string]any) (string, []any, error)
// PlaceholderFormat returns the placeholder format for the driver ("$" for PostgreSQL, "?" for SQLite).
PlaceholderFormat() string
// BuildInsertQuery generates a multi-row INSERT statement for batch imports.
// table: target table name
// columns: column names to insert
// rowCount: number of rows in this batch
// onConflict: conflict handling strategy ("error", "ignore", "replace")
// conflictTarget: column(s) for conflict detection (required for PostgreSQL UPSERT with "replace")
// updateColumns: columns to update on conflict (if empty, updates all non-key columns)
// Returns the SQL query string with placeholders.
BuildInsertQuery(table string, columns []string, rowCount int, onConflict, conflictTarget string, updateColumns []string) string
// QuoteIdentifier quotes a table or column name to handle reserved words and special characters.
// For PostgreSQL and SQLite, this wraps the identifier in double quotes.
QuoteIdentifier(name string) string
}
Driver defines the interface for SQL database drivers. Each database (PostgreSQL, SQLite) implements this interface.
type DriverRegistry ¶
type DriverRegistry struct {
// contains filtered or unexported fields
}
DriverRegistry holds registered database drivers with thread-safe access.
func NewDriverRegistry ¶
func NewDriverRegistry() *DriverRegistry
NewDriverRegistry creates a new driver registry.
func (*DriverRegistry) Get ¶
func (r *DriverRegistry) Get(name string) (Driver, bool)
Get retrieves a driver by name (thread-safe).
func (*DriverRegistry) Register ¶
func (r *DriverRegistry) Register(driver Driver)
Register adds a driver to the registry (thread-safe).
type ExecutionMetrics ¶
type ExecutionMetrics struct {
QueryHash string `json:"query_hash"`
StartedAt time.Time `json:"started_at"`
FinishedAt time.Time `json:"finished_at"`
DurationMs int64 `json:"duration_ms"`
RowsAffected int64 `json:"rows_affected,omitempty"`
RowsReturned int64 `json:"rows_returned,omitempty"`
Status string `json:"status"`
Error string `json:"error,omitempty"`
}
ExecutionMetrics holds metrics from SQL execution.
type GlobalPoolConfig ¶
type GlobalPoolConfig struct {
// MaxOpenConns is the maximum total open connections across all DSNs.
MaxOpenConns int
// MaxIdleConns is the maximum idle connections per DSN.
MaxIdleConns int
// ConnMaxLifetime is the maximum lifetime of a connection.
ConnMaxLifetime time.Duration
// ConnMaxIdleTime is the maximum idle time for a connection.
ConnMaxIdleTime time.Duration
}
GlobalPoolConfig holds configuration for the global pool manager.
type GlobalPoolManager ¶
type GlobalPoolManager struct {
// contains filtered or unexported fields
}
GlobalPoolManager manages PostgreSQL connection pools across all DAG executions. It is designed for workers where multiple DAGs run concurrently in a single process and share database connections.
func GetPoolManager ¶
func GetPoolManager(ctx context.Context) *GlobalPoolManager
GetPoolManager retrieves the global pool manager from context. Returns nil if no pool manager is configured.
func NewGlobalPoolManager ¶
func NewGlobalPoolManager(cfg GlobalPoolConfig) *GlobalPoolManager
NewGlobalPoolManager creates a new global pool manager.
func (*GlobalPoolManager) Close ¶
func (m *GlobalPoolManager) Close() error
Close closes all pools and the manager. This should be called when the worker shuts down.
func (*GlobalPoolManager) GetOrCreatePool ¶
func (m *GlobalPoolManager) GetOrCreatePool(ctx context.Context, driver Driver, cfg *Config) (*sql.DB, error)
GetOrCreatePool returns an existing pool or creates a new one for the DSN. The pool is configured with the global limits.
func (*GlobalPoolManager) ReleasePool ¶
func (m *GlobalPoolManager) ReleasePool(dsn string)
ReleasePool decrements the reference count for a DSN's pool. The pool is kept open for reuse; it will be closed when the manager is closed.
func (*GlobalPoolManager) Stats ¶
func (m *GlobalPoolManager) Stats() map[string]any
Stats returns statistics about the pool manager.
type ImportConfig ¶
type ImportConfig struct {
// Required fields
InputFile string `mapstructure:"input_file"` // Path to input file
Table string `mapstructure:"table"` // Target table name
// Format options
Format string `mapstructure:"format"` // csv, tsv, jsonl (auto-detect if empty)
HasHeader *bool `mapstructure:"has_header"` // Whether first row is header (default: true for csv/tsv)
Delimiter string `mapstructure:"delimiter"` // Field delimiter (default: "," for csv, "\t" for tsv)
// Column mapping
Columns []string `mapstructure:"columns"` // Explicit column names (overrides header)
// NULL handling
NullValues []string `mapstructure:"null_values"` // Values to treat as NULL
// Batch settings
BatchSize int `mapstructure:"batch_size"` // Rows per INSERT statement (default: 1000)
// Conflict handling
OnConflict string `mapstructure:"on_conflict"` // error (default), ignore, replace
ConflictTarget string `mapstructure:"conflict_target"` // Column(s) for conflict detection (required for PostgreSQL UPSERT with "replace")
UpdateColumns []string `mapstructure:"update_columns"` // Columns to update on conflict (driver support varies)
// Row limits
SkipRows int `mapstructure:"skip_rows"` // Skip first N data rows
MaxRows int `mapstructure:"max_rows"` // Limit import (0 = unlimited)
// Validation
DryRun bool `mapstructure:"dry_run"` // Validate without importing
}
ImportConfig configures data import from CSV/TSV/JSONL files.
type ImportMetrics ¶
type ImportMetrics struct {
StartedAt time.Time `json:"started_at"`
FinishedAt time.Time `json:"finished_at"`
DurationMs int64 `json:"duration_ms"`
RowsRead int64 `json:"rows_read"`
RowsImported int64 `json:"rows_imported"`
RowsSkipped int64 `json:"rows_skipped"`
BatchCount int `json:"batch_count"`
Status string `json:"status"`
Error string `json:"error,omitempty"`
}
ImportMetrics tracks import operation statistics.
type Importer ¶
type Importer struct {
// contains filtered or unexported fields
}
Importer handles data import from files to database tables.
func NewImporter ¶
NewImporter creates a new Importer instance.
type InputOptions ¶
type InputOptions struct {
HasHeader bool // Whether first row is header (CSV/TSV)
Delimiter rune // Field delimiter (default: ',' for CSV, '\t' for TSV)
NullValues []string // Values to treat as NULL
Columns []string // Expected column names (for JSONL)
}
InputOptions configures input reader behavior.
func DefaultInputOptions ¶
func DefaultInputOptions(format string) InputOptions
DefaultInputOptions returns default options for the given format.
type InputReader ¶
type InputReader interface {
// ReadHeader reads the column headers from the input.
// Returns nil for formats without headers or when hasHeader is false.
ReadHeader() ([]string, error)
// ReadRow reads the next row of data.
// Returns io.EOF when no more rows are available.
ReadRow() ([]any, error)
// Close releases any resources held by the reader.
Close() error
}
InputReader defines the interface for reading input data for import. This is the symmetric inverse of ResultWriter - it reads rows for input rather than writing rows for output.
func NewInputReader ¶
func NewInputReader(r io.Reader, format string, opts InputOptions) (InputReader, error)
NewInputReader creates an InputReader based on the input format.
type JSONLReader ¶
type JSONLReader struct {
// contains filtered or unexported fields
}
JSONLReader implements InputReader for JSON Lines formatted input.
func NewJSONLReader ¶
func NewJSONLReader(r io.Reader, opts InputOptions) *JSONLReader
NewJSONLReader creates a new JSON Lines reader.
func (*JSONLReader) ReadHeader ¶
func (r *JSONLReader) ReadHeader() ([]string, error)
ReadHeader returns the expected columns for JSONL. If columns were not specified, reads the first line to determine keys.
func (*JSONLReader) ReadRow ¶
func (r *JSONLReader) ReadRow() ([]any, error)
ReadRow reads the next JSON object from the input.
type JSONLWriter ¶
type JSONLWriter struct {
// contains filtered or unexported fields
}
JSONLWriter writes results in JSON Lines format.
func NewJSONLWriter ¶
func NewJSONLWriter(w io.Writer, nullString string) *JSONLWriter
NewJSONLWriter creates a new JSONL writer.
func (*JSONLWriter) RowCount ¶
func (w *JSONLWriter) RowCount() int
RowCount returns the number of rows written.
func (*JSONLWriter) WriteHeader ¶
func (w *JSONLWriter) WriteHeader(columns []string) error
WriteHeader stores column names for row writing.
func (*JSONLWriter) WriteRow ¶
func (w *JSONLWriter) WriteRow(values []any) error
WriteRow writes a row as a JSON object.
type JSONWriter ¶
type JSONWriter struct {
// contains filtered or unexported fields
}
JSONWriter writes results as a JSON array.
func NewJSONWriter ¶
func NewJSONWriter(w io.Writer, nullString string) *JSONWriter
NewJSONWriter creates a new JSON writer.
func (*JSONWriter) Close ¶
func (w *JSONWriter) Close() error
Close writes the accumulated JSON array.
func (*JSONWriter) RowCount ¶
func (w *JSONWriter) RowCount() int
RowCount returns the number of rows accumulated.
func (*JSONWriter) WriteHeader ¶
func (w *JSONWriter) WriteHeader(columns []string) error
WriteHeader stores column names for row writing.
func (*JSONWriter) WriteRow ¶
func (w *JSONWriter) WriteRow(values []any) error
WriteRow accumulates a row for later writing.
type QueryExecutor ¶
type QueryExecutor interface {
QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)
ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
}
QueryExecutor provides methods to execute SQL queries.
func GetQueryExecutor ¶
func GetQueryExecutor(db *sql.DB, tx *Transaction) QueryExecutor
GetQueryExecutor returns the appropriate query executor (transaction or db).
type ResultWriter ¶
type ResultWriter interface {
// WriteHeader writes the column headers (if applicable).
WriteHeader(columns []string) error
// WriteRow writes a single row of data.
WriteRow(values []any) error
// Flush ensures all buffered data is written.
Flush() error
// Close finalizes the writer.
Close() error
}
ResultWriter defines the interface for writing query results.
func NewResultWriter ¶
func NewResultWriter(w io.Writer, format, nullString string, headers bool) ResultWriter
NewResultWriter creates a ResultWriter based on the output format.
type Transaction ¶
type Transaction struct {
// contains filtered or unexported fields
}
Transaction represents a database transaction with isolation level support.
func BeginTransaction ¶
BeginTransaction starts a new transaction with the specified isolation level.
func (*Transaction) Rollback ¶
func (t *Transaction) Rollback() error
Rollback rolls back the transaction.
Source Files
¶
Directories
¶
| Path | Synopsis |
|---|---|
|
drivers
|
|
|
postgres
Package postgres provides the PostgreSQL driver for the SQL executor.
|
Package postgres provides the PostgreSQL driver for the SQL executor. |
|
sqlite
Package sqlite provides the SQLite driver for the SQL executor.
|
Package sqlite provides the SQLite driver for the SQL executor. |