database

package
v1.5.3 Latest Latest
Warning

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

 Go to latest Published: Dec 23, 2025 License: GPL-3.0 Imports: 123 Imported by: 3

Documentation

Index

Constants

View Source 
const RelationManyToMany = "many_to_many"
View Source 
const RelationManyToOne = "many_to_one"
View Source 
const RelationOneToMany = "one_to_many"
View Source 
const RelationOneToOne = "one_to_one"

Variables

View Source 
var ChunkByColumn = func(conn Connection, table Table, c string, p int) ([]Table, error) {
	return []Table{table}, g.Error("please use the official sling-cli release for chunking columns")
}
View Source 
var ChunkByColumnRange = func(conn Connection, t Table, c string, cs, min, max string) ([]Chunk, error) {
	return []Chunk{}, g.Error("please use the official sling-cli release for chunking")
}
View Source 
var ChunkByCount = func(conn Connection, t Table, c string, cc int, min, max string) ([]Chunk, string, error) {
	return []Chunk{}, "", g.Error("please use the official sling-cli release for chunking")
}
View Source 
var ChunkByExpression = func(conn Connection, t Table, e string, cc int) ([]Chunk, error) {
	return []Chunk{}, g.Error("please use the official sling-cli release for chunking")
}
View Source 
var Debug = false

Debug prints queries when true

View Source 
var DuckDbFileCmd = map[string]*exec.Cmd{}
View Source 
var DuckDbFileContext = map[string]*g.Context{} // so that collision doesn't happen
View Source 
var DuckDbMux = sync.Mutex{}
View Source 
var DuckDbUseTempFile = false
View Source 
var InferDBStream = false

InferDBStream may need to be `true`, since precision and scale is not guaranteed. If `false`, will use the database stream source schema

View Source 
var SQLiteVersion = "3.41"
View Source 
var (
	// UseBulkExportFlowCSV to use BulkExportFlowCSV
	UseBulkExportFlowCSV = false
)

Functions

func ChangeColumnTypeViaAdd 

func ChangeColumnTypeViaAdd(conn Connection, table Table, col iop.Column) (err error)

ChangeColumnTypeViaAdd swaps a new column with the old in order to change the type need to use this with snowflake when changing from date to string, or number to string

func CommonColumns 

func CommonColumns(colNames1 []string, colNames2 []string) (commCols []string)

CommonColumns return common columns

func CopyFromAzure 

func CopyFromAzure(conn Connection, tableFName, azPath string) (err error)

CopyFromAzure uses the Snowflake COPY INTO Table command from Azure https://docs.snowflake.com/en/sql-reference/sql/copy-into-table.html

func CopyFromS3 

func CopyFromS3(conn Connection, tableFName, s3Path string) (err error)

func EnsureBinSQLite 

func EnsureBinSQLite() (binPath string, err error)

EnsureBinSQLite ensures sqlite binary exists if missing, downloads and uses

func GenerateAlterDDL 

func GenerateAlterDDL(conn Connection, table Table, newColumns iop.Columns) (bool, error)

GenerateAlterDDL generate a DDL based on a dataset

func GetArrowDBCDriverType added in v1.5.2 

func GetArrowDBCDriverType(driverName string) dbio.Type

GetArrowDBCDriverType maps ADBC driver names to corresponding database types This allows using driver-specific SQL templates

func GetOptimizeTableStatements added in v1.1.8 

func GetOptimizeTableStatements(conn Connection, table *Table, newColumns iop.Columns, isTemp bool) (ok bool, ddlParts []string, err error)

GetOptimizeTableStatements analyzes the table and alters the table with the columns data type based on its analysis result if table is missing, it is created with a new DDl Hole in this: will truncate data points, since it is based only on new data being inserted... would need a complete stats of the target table to properly optimize.

func GetQualifierQuote 

func GetQualifierQuote(dialect dbio.Type) string

func InsertBatchStream 

func InsertBatchStream(conn Connection, tx Transaction, tableFName string, ds *iop.Datastream) (count uint64, err error)

InsertBatchStream inserts a stream into a table in batch

func InsertStream 

func InsertStream(conn Connection, tx *BaseTransaction, tableFName string, ds *iop.Datastream) (count uint64, err error)

InsertStream inserts a stream

func MakeDuckDbSecretProps added in v1.4.11 

func MakeDuckDbSecretProps(conn Connection, secretType iop.DuckDbSecretType) (secretProps map[string]string)

MakeDuckDbSecretProps convert conn props to duckdb secret props

func Merge added in v1.4.20 

func Merge(conn Connection, tx Transaction, sourceTable, targetTable string, pkFields []string) (count int64, err error)

Merge upserts from source table into target table

func NativeTypeToGeneral 

func NativeTypeToGeneral(name, dbType string, conn Connection) (colType iop.ColumnType)

func PK 

func PK(obj interface{}) (pk []string)

PK returns the primary keys of a model

func ParseColumnName 

func ParseColumnName(text string, dialect dbio.Type) (colName string, err error)

func ParseSQLMultiStatements 

func ParseSQLMultiStatements(sql string, Dialect ...dbio.Type) (sqls []string)

ParseSQLMultiStatements splits a sql text into statements typically by a ';'

func PrintSessionID added in v1.4.24 

func PrintSessionID(conn Connection)

func QueryOperation added in v1.4.24 

func QueryOperation(conn Connection, operation Operation, params map[string]any) (query string, err error)

QueryOperation uses the operation input to render/generate a query based on a template. This should be connection agnostic.

func SQLColumns 

func SQLColumns(colTypes []ColumnType, conn Connection) (columns iop.Columns)

SQLColumns returns the columns from database ColumnType

func SplitTableFullName 

func SplitTableFullName(tableName string) (string, string)

SplitTableFullName retrusn the schema / table name

func TestPermissions 

func TestPermissions(conn Connection, tableName string) (err error)

TestPermissions tests the needed permissions in a given connection

func TrimSQLComments added in v1.4.7 

func TrimSQLComments(sql string) (string, error)

TrimSQLComments removes all SQL comments (line and block comments) from the input SQL Line comments start with -- and end with a newline Block comments start with /* and end with */

func TruncateTable added in v1.4.14 

func TruncateTable(conn Connection, tableName string) error

func UID 

func UID(obj interface{}) string

UID returns a unique ID for that object

Types

type ArrowDBConn added in v1.5.2 

type ArrowDBConn struct {
	BaseConn
	URL string

	Conn adbc.Connection
	// contains filtered or unexported fields
}

ArrowDBConn is an Arrow FlightSQL connection

func (*ArrowDBConn) BulkExportFlow added in v1.5.2 

func (conn *ArrowDBConn) BulkExportFlow(table Table) (df *iop.Dataflow, err error)

BulkExportFlow exports data as a dataflow

func (*ArrowDBConn) BulkExportStream added in v1.5.2 

func (conn *ArrowDBConn) BulkExportStream(table Table) (ds *iop.Datastream, err error)

BulkExportStream streams the rows in bulk

func (*ArrowDBConn) BulkImportFlow added in v1.5.2 

func (conn *ArrowDBConn) BulkImportFlow(tableFName string, df *iop.Dataflow) (count uint64, err error)

BulkImportFlow imports data from a dataflow using ADBC bulk ingestion

func (*ArrowDBConn) BulkImportStream added in v1.5.2 

func (conn *ArrowDBConn) BulkImportStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

BulkImportStream imports data from a datastream using ADBC bulk ingestion

func (*ArrowDBConn) Close added in v1.5.2 

func (conn *ArrowDBConn) Close() error

Close closes the ADBC connection and database

func (*ArrowDBConn) Connect added in v1.5.2 

func (conn *ArrowDBConn) Connect(timeOut ...int) (err error)

Connect opens the ADBC connection

func (*ArrowDBConn) ExecContext added in v1.5.2 

func (conn *ArrowDBConn) ExecContext(ctx context.Context, sql string, args ...interface{}) (result sql.Result, err error)

ExecContext executes a SQL statement (read-only operations)

func (*ArrowDBConn) GetNativeType added in v1.5.2 

func (conn *ArrowDBConn) GetNativeType(col iop.Column) (nativeType string, err error)

GetNativeType returns the native column type from generic

func (*ArrowDBConn) GetSQLColumns added in v1.5.2 

func (conn *ArrowDBConn) GetSQLColumns(table Table) (columns iop.Columns, err error)

GetSQLColumns returns columns for a SQL query using Arrow schema This avoids wrapping with LIMIT which may not work for all database types

func (*ArrowDBConn) GetTemplateValue added in v1.5.2 

func (conn *ArrowDBConn) GetTemplateValue(path string) string

GetTemplateValue returns the template value for the given path It first checks the driver-specific template, then falls back to ADBC template

func (*ArrowDBConn) Init added in v1.5.2 

func (conn *ArrowDBConn) Init() error

Init initiates the connection

func (*ArrowDBConn) LoadTemplates added in v1.5.2 

func (conn *ArrowDBConn) LoadTemplates() error

LoadTemplates loads the appropriate yaml template For ADBC, it merges the driver-specific template with the ADBC template Driver template is base, ADBC template overrides for ADBC-specific behavior

func (*ArrowDBConn) Quote added in v1.5.2 

func (conn *ArrowDBConn) Quote(field string) string

func (*ArrowDBConn) StreamRowsContext added in v1.5.2 

func (conn *ArrowDBConn) StreamRowsContext(ctx context.Context, sql string, options ...map[string]interface{}) (ds *iop.Datastream, err error)

StreamRowsContext streams query results as a datastream using Arrow record batches

func (*ArrowDBConn) Template added in v1.5.2 

func (conn *ArrowDBConn) Template() dbio.Template

func (*ArrowDBConn) Unquote added in v1.5.2 

func (conn *ArrowDBConn) Unquote(field string) string

type AthenaConn added in v1.4.7 

type AthenaConn struct {
	BaseConn
	Client          *athena.Client
	URL             string
	DataLocation    string
	StagingLocation string
}

AthenaConn is an Athena connection

func (*AthenaConn) BulkExportFlow added in v1.4.7 

func (conn *AthenaConn) BulkExportFlow(table Table) (df *iop.Dataflow, err error)

BulkExportFlow reads in bulk

func (*AthenaConn) BulkExportStream added in v1.4.7 

func (conn *AthenaConn) BulkExportStream(table Table) (ds *iop.Datastream, err error)

BulkExportStream reads in bulk

func (*AthenaConn) BulkImportFlow added in v1.4.7 

func (conn *AthenaConn) BulkImportFlow(tableFName string, df *iop.Dataflow) (count uint64, err error)

BulkImportFlow inserts a flow of streams into a table. For athena we need to create CSVs in S3 and then use the COPY command.

func (*AthenaConn) BulkImportStream added in v1.4.7 

func (conn *AthenaConn) BulkImportStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

BulkImportStream inserts a stream into a table. For athena we need to create CSVs in S3 and then use the COPY command.

func (*AthenaConn) CastColumnForSelect added in v1.4.7 

func (conn *AthenaConn) CastColumnForSelect(srcCol iop.Column, tgtCol iop.Column) (selectStr string)

CastColumnForSelect casts to the correct target column type

func (*AthenaConn) Close added in v1.4.7 

func (conn *AthenaConn) Close() error

Close closes the connection

func (*AthenaConn) Connect added in v1.4.7 

func (conn *AthenaConn) Connect(timeOut ...int) (err error)

Connect connects to the database

func (*AthenaConn) ExecContext added in v1.4.7 

func (conn *AthenaConn) ExecContext(ctx context.Context, sql string, args ...interface{}) (result sql.Result, err error)

ExecContext executes the sql query

func (*AthenaConn) GenerateDDL added in v1.4.7 

func (conn *AthenaConn) GenerateDDL(table Table, data iop.Dataset, temporary bool) (sql string, err error)

GenerateDDL generates a DDL based on a dataset

func (*AthenaConn) GenerateMergeSQL added in v1.4.20 

func (conn *AthenaConn) GenerateMergeSQL(srcTable string, tgtTable string, pkFields []string) (sql string, err error)

GenerateMergeSQL generates the upsert SQL

func (*AthenaConn) GetDatabases added in v1.4.7 

func (conn *AthenaConn) GetDatabases() (iop.Dataset, error)

GetDatabases returns databases for given connection

func (*AthenaConn) GetSchemata added in v1.4.7 

func (conn *AthenaConn) GetSchemata(level SchemataLevel, schemaName string, tableNames ...string) (Schemata, error)

GetSchemata obtain full schemata info for a schema and/or table in current database

func (*AthenaConn) Init added in v1.4.7 

func (conn *AthenaConn) Init() error

Init initiates the object

func (*AthenaConn) InsertBatchStream added in v1.4.7 

func (conn *AthenaConn) InsertBatchStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

InsertBatchStream inserts a stream into a table in batch

func (*AthenaConn) InsertStream added in v1.4.7 

func (conn *AthenaConn) InsertStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

InsertStream demonstrates loading data into a Athena table using a file on the local filesystem.

func (*AthenaConn) LoadFromS3 added in v1.4.10 

func (conn *AthenaConn) LoadFromS3(tableFName, s3Path string, columns iop.Columns) (count uint64, err error)

LoadFromS3 creates a temporary external table pointing to S3, then inserts into the target table

func (*AthenaConn) NewTransaction added in v1.4.7 

func (conn *AthenaConn) NewTransaction(ctx context.Context, options ...*sql.TxOptions) (tx Transaction, err error)

NewTransaction creates a new transaction

func (*AthenaConn) StagingBucket added in v1.4.10 

func (conn *AthenaConn) StagingBucket() string

func (*AthenaConn) StreamRowsContext added in v1.4.7 

func (conn *AthenaConn) StreamRowsContext(ctx context.Context, sql string, options ...map[string]interface{}) (ds *iop.Datastream, err error)

StreamRowsContext executes the query and streams the result into a datastream

func (*AthenaConn) Unload added in v1.4.7 

func (conn *AthenaConn) Unload(ctx *g.Context, tables ...Table) (s3Path string, err error)

Unload exports the sql query into an S3 bucket

type AzureTableConn added in v1.4.16 

type AzureTableConn struct {
	BaseConn
	URL           string
	Client        *aztables.ServiceClient
	AccountName   string
	AccountKey    string
	SASToken      string
	ConnectionStr string
	TableName     string // Specific table name for table-specific SAS tokens
}

AzureTableConn is an Azure Table connection

func (*AzureTableConn) BulkExportFlow added in v1.4.16 

func (conn *AzureTableConn) BulkExportFlow(table Table) (df *iop.Dataflow, err error)

func (*AzureTableConn) BulkImportFlow added in v1.4.16 

func (conn *AzureTableConn) BulkImportFlow(tableFName string, df *iop.Dataflow) (count uint64, err error)

BulkImportFlow imports data into Azure Tables

func (*AzureTableConn) Close added in v1.4.16 

func (conn *AzureTableConn) Close() error

func (*AzureTableConn) Connect added in v1.4.16 

func (conn *AzureTableConn) Connect(timeOut ...int) error

Connect connects to the database

func (*AzureTableConn) ExecContext added in v1.4.16 

func (conn *AzureTableConn) ExecContext(ctx context.Context, sql string, args ...any) (result sql.Result, err error)

func (*AzureTableConn) GetCount added in v1.4.16 

func (conn *AzureTableConn) GetCount(tableFName string) (int64, error)

GetCount returns the row count for a table

func (*AzureTableConn) GetSchemas added in v1.4.16 

func (conn *AzureTableConn) GetSchemas() (data iop.Dataset, err error)

GetSchemas returns schemas

func (*AzureTableConn) GetSchemata added in v1.4.16 

func (conn *AzureTableConn) GetSchemata(level SchemataLevel, schemaName string, tableNames ...string) (Schemata, error)

GetSchemata obtain full schemata info for a schema and/or table in current database

func (*AzureTableConn) GetTableColumns added in v1.4.16 

func (conn *AzureTableConn) GetTableColumns(table *Table, fields ...string) (columns iop.Columns, err error)

GetTableColumns returns columns for a table

func (*AzureTableConn) GetTables added in v1.4.16 

func (conn *AzureTableConn) GetTables(schema string) (data iop.Dataset, err error)

GetTables returns tables

func (*AzureTableConn) Init added in v1.4.16 

func (conn *AzureTableConn) Init() error

Init initiates the object

func (*AzureTableConn) NewTransaction added in v1.4.16 

func (conn *AzureTableConn) NewTransaction(ctx context.Context, options ...*sql.TxOptions) (tx Transaction, err error)

NewTransaction creates a new transaction

func (*AzureTableConn) StreamRowsContext added in v1.4.16 

func (conn *AzureTableConn) StreamRowsContext(ctx context.Context, tableName string, Opts ...map[string]any) (ds *iop.Datastream, err error)

func (*AzureTableConn) TableExists added in v1.4.16 

func (conn *AzureTableConn) TableExists(table Table) (exists bool, err error)

TableExists checks if a table exists

type AzureTableResult added in v1.4.16 

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

AzureTableResult implements sql.Result interface

func (*AzureTableResult) LastInsertId added in v1.4.16 

func (r *AzureTableResult) LastInsertId() (int64, error)

func (*AzureTableResult) RowsAffected added in v1.4.16 

func (r *AzureTableResult) RowsAffected() (int64, error)

type BaseConn 

type BaseConn struct {
	Connection
	URL  string
	Type dbio.Type // the type of database for sqlx: postgres, mysql, sqlite

	Data iop.Dataset

	Log []string
	// contains filtered or unexported fields
}

BaseConn is a database connection

func (*BaseConn) AddMissingColumns added in v1.1.8 

func (conn *BaseConn) AddMissingColumns(table Table, newCols iop.Columns) (ok bool, err error)

func (*BaseConn) Base 

func (conn *BaseConn) Base() *BaseConn

BaseURL returns the base Conn

func (*BaseConn) BaseURL 

func (conn *BaseConn) BaseURL() string

BaseURL returns the base URL with default port

func (*BaseConn) Begin 

func (conn *BaseConn) Begin(options ...*sql.TxOptions) (err error)

Begin starts a connection wide transaction

func (*BaseConn) BeginContext 

func (conn *BaseConn) BeginContext(ctx context.Context, options ...*sql.TxOptions) (err error)

BeginContext starts a connection wide transaction

func (*BaseConn) BulkExportFlow 

func (conn *BaseConn) BulkExportFlow(table Table) (df *iop.Dataflow, err error)

BulkExportFlow creates a dataflow from a sql query

func (*BaseConn) BulkExportFlowCSV 

func (conn *BaseConn) BulkExportFlowCSV(table Table) (df *iop.Dataflow, err error)

BulkExportFlowCSV creates a dataflow from a sql query, using CSVs

func (*BaseConn) BulkExportStream 

func (conn *BaseConn) BulkExportStream(table Table) (ds *iop.Datastream, err error)

BulkExportStream streams the rows in bulk

func (*BaseConn) BulkImportFlow 

func (conn *BaseConn) BulkImportFlow(tableFName string, df *iop.Dataflow) (count uint64, err error)

BulkImportFlow imports the streams rows in bulk concurrently using channels

func (*BaseConn) BulkImportStream 

func (conn *BaseConn) BulkImportStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

BulkImportStream import the stream rows in bulk

func (*BaseConn) CastColumnForSelect 

func (conn *BaseConn) CastColumnForSelect(srcCol iop.Column, tgtCol iop.Column) string

CastColumnForSelect casts to the correct target column type

func (*BaseConn) CastColumnsForSelect 

func (conn *BaseConn) CastColumnsForSelect(srcColumns iop.Columns, tgtColumns iop.Columns) []string

CastColumnsForSelect cast the source columns into the target Column types

func (*BaseConn) Close 

func (conn *BaseConn) Close() error

Close closes the connection

func (*BaseConn) Commit 

func (conn *BaseConn) Commit() (err error)

Commit commits a connection wide transaction

func (*BaseConn) CompareChecksums 

func (conn *BaseConn) CompareChecksums(tableName string, columns iop.Columns) (err error)

CompareChecksums compares the checksum values from the database side to the checkum values from the StreamProcessor

func (*BaseConn) ConnString 

func (conn *BaseConn) ConnString() string

ConnString returns the connection string needed for connection

func (*BaseConn) Connect 

func (conn *BaseConn) Connect(timeOut ...int) (err error)

Connect connects to the database

func (*BaseConn) Context 

func (conn *BaseConn) Context() *g.Context

Context returns the db context

func (*BaseConn) CreateTable 

func (conn *BaseConn) CreateTable(tableName string, cols iop.Columns, tableDDL string) (err error)

CreateTable creates a new table based on provided columns `tableName` should have 'schema.table' format

func (*BaseConn) CreateTemporaryTable 

func (conn *BaseConn) CreateTemporaryTable(tableName string, cols iop.Columns) (err error)

CreateTemporaryTable creates a temp table based on provided columns

func (*BaseConn) CurrentDatabase 

func (conn *BaseConn) CurrentDatabase() (dbName string, err error)

CurrentDatabase returns the name of the current database

func (*BaseConn) Db 

func (conn *BaseConn) Db() *sqlx.DB

Db returns the sqlx db object

func (*BaseConn) DbX 

func (conn *BaseConn) DbX() *DbX

DbX returns the DbX object

func (*BaseConn) DropTable 

func (conn *BaseConn) DropTable(tableNames ...string) (err error)

DropTable drops given table.

func (*BaseConn) DropView 

func (conn *BaseConn) DropView(viewNames ...string) (err error)

DropView drops given view.

func (*BaseConn) Exec 

func (conn *BaseConn) Exec(sql string, args ...interface{}) (result sql.Result, err error)

Exec runs a sql query, returns `error`

func (*BaseConn) ExecContext 

func (conn *BaseConn) ExecContext(ctx context.Context, q string, args ...interface{}) (result sql.Result, err error)

ExecContext runs a sql query with context, returns `error`

func (*BaseConn) ExecMulti 

func (conn *BaseConn) ExecMulti(sqls ...string) (result sql.Result, err error)

ExecMulti runs mutiple sql queries, returns `error`

func (*BaseConn) ExecMultiContext 

func (conn *BaseConn) ExecMultiContext(ctx context.Context, qs ...string) (result sql.Result, err error)

ExecMultiContext runs multiple sql queries with context, returns `error`

func (*BaseConn) GenerateDDL 

func (conn *BaseConn) GenerateDDL(table Table, data iop.Dataset, temporary bool) (string, error)

GenerateDDL genrate a DDL based on a dataset

func (*BaseConn) GenerateInsertStatement 

func (conn *BaseConn) GenerateInsertStatement(tableName string, cols iop.Columns, numRows int) string

GenerateInsertStatement returns the proper INSERT statement

func (*BaseConn) GenerateMergeConfig added in v1.4.20 

func (conn *BaseConn) GenerateMergeConfig(srcTable string, tgtTable string, pkFields []string) (mc MergeConfig, err error)

GenerateMergeConfig returns the merge config

func (*BaseConn) GenerateMergeExpressions added in v1.4.20 

func (conn *BaseConn) GenerateMergeExpressions(srcTable string, tgtTable string, pkFields []string) (upsertMap map[string]string, err error)

GenerateMergeExpressions returns a map with needed expressions

func (*BaseConn) GenerateMergeSQL added in v1.4.20 

func (conn *BaseConn) GenerateMergeSQL(srcTable string, tgtTable string, pkFields []string) (sql string, err error)

GenerateMergeSQL returns a sql for upsert

func (*BaseConn) GetAnalysis 

func (conn *BaseConn) GetAnalysis(analysisName string, values map[string]interface{}) (sql string, err error)

GetAnalysis runs an analysis

func (*BaseConn) GetColumnStats 

func (conn *BaseConn) GetColumnStats(tableName string, fields ...string) (columns iop.Columns, err error)

GetColumnStats analyzes the table and returns the column statistics

func (*BaseConn) GetColumns 

func (conn *BaseConn) GetColumns(tableFName string, fields ...string) (columns iop.Columns, err error)

func (*BaseConn) GetColumnsFull 

func (conn *BaseConn) GetColumnsFull(tableFName string) (iop.Dataset, error)

GetColumnsFull returns columns for given table. `tableName` should include schema and table, example: `schema1.table2` fields should be `schema_name|table_name|table_type|column_name|data_type|column_id`

func (*BaseConn) GetCount 

func (conn *BaseConn) GetCount(tableFName string) (int64, error)

GetCount returns count of records

func (*BaseConn) GetDDL 

func (conn *BaseConn) GetDDL(tableFName string) (string, error)

GetDDL returns DDL for given table.

func (*BaseConn) GetDatabases 

func (conn *BaseConn) GetDatabases() (iop.Dataset, error)

GetDatabases returns databases for given connection

func (*BaseConn) GetGormConn 

func (conn *BaseConn) GetGormConn(config *gorm.Config) (*gorm.DB, error)

GetGormConn returns the gorm db connection

func (*BaseConn) GetIndexes 

func (conn *BaseConn) GetIndexes(tableFName string) (iop.Dataset, error)

GetIndexes returns indexes for given table.

func (*BaseConn) GetMaxValue added in v1.4.10 

func (conn *BaseConn) GetMaxValue(table Table, colName string) (value any, maxCol iop.Column, err error)

GetMaxValue get the max value of a column

func (*BaseConn) GetNativeType 

func (conn *BaseConn) GetNativeType(col iop.Column) (nativeType string, err error)

GetNativeType returns the native column type from generic

func (*BaseConn) GetObjects 

func (conn *BaseConn) GetObjects(schema string, objectType string) (iop.Dataset, error)

GetObjects returns objects (tables or views) for given schema `objectType` can be either 'table', 'view' or 'all'

func (*BaseConn) GetPrimaryKeys 

func (conn *BaseConn) GetPrimaryKeys(tableFName string) (iop.Dataset, error)

GetPrimaryKeys returns primark keys for given table.

func (*BaseConn) GetProp 

func (conn *BaseConn) GetProp(key ...string) string

GetProp returns the value of a property

func (*BaseConn) GetSQLColumns 

func (conn *BaseConn) GetSQLColumns(table Table) (columns iop.Columns, err error)

GetSQLColumns return columns from a sql query result

func (*BaseConn) GetSchemas 

func (conn *BaseConn) GetSchemas() (iop.Dataset, error)

GetSchemas returns schemas

func (*BaseConn) GetSchemata 

func (conn *BaseConn) GetSchemata(level SchemataLevel, schemaName string, tableNames ...string) (Schemata, error)

GetSchemata obtain full schemata info for a schema and/or table in current database

func (*BaseConn) GetTableColumns 

func (conn *BaseConn) GetTableColumns(table *Table, fields ...string) (columns iop.Columns, err error)

GetColumns returns columns for given table. `tableFName` should include schema and table, example: `schema1.table2` fields should be `column_name|data_type`

func (*BaseConn) GetTables 

func (conn *BaseConn) GetTables(schema string) (iop.Dataset, error)

GetTables returns tables for given schema

func (*BaseConn) GetTablesAndViews added in v1.2.19 

func (conn *BaseConn) GetTablesAndViews(schema string) (iop.Dataset, error)

GetTablesAndViews returns tables/views for given schema

func (*BaseConn) GetTemplateValue 

func (conn *BaseConn) GetTemplateValue(path string) (value string)

GetTemplateValue returns the value of the path

func (*BaseConn) GetType 

func (conn *BaseConn) GetType() dbio.Type

GetType returns the type db object

func (*BaseConn) GetURL 

func (conn *BaseConn) GetURL(newURL ...string) string

GetURL returns the processed URL

func (*BaseConn) GetViews 

func (conn *BaseConn) GetViews(schema string) (iop.Dataset, error)

GetViews returns views for given schema

func (*BaseConn) Import 

func (conn *BaseConn) Import(data iop.Dataset, tableName string) error

Import imports `data` into `tableName`

func (*BaseConn) Info 

func (conn *BaseConn) Info() (ci ConnInfo)

Info returns connection information

func (*BaseConn) Init 

func (conn *BaseConn) Init() (err error)

Init initiates the connection object & add default port if missing

func (*BaseConn) InsertBatchStream 

func (conn *BaseConn) InsertBatchStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

InsertBatchStream inserts a stream into a table in batch

func (*BaseConn) InsertStream 

func (conn *BaseConn) InsertStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

InsertStream inserts a stream into a table

func (*BaseConn) Kill 

func (conn *BaseConn) Kill() error

Kill kill the database connection

func (*BaseConn) LoadTemplates 

func (conn *BaseConn) LoadTemplates() (err error)

LoadTemplates loads the appropriate yaml template

func (*BaseConn) LogSQL added in v1.1.8 

func (conn *BaseConn) LogSQL(query string, args ...any)

LogSQL logs a query for debugging

func (*BaseConn) Merge added in v1.4.20 

func (conn *BaseConn) Merge(srcTable string, tgtTable string, primKeys []string) (rowAffCnt int64, err error)

Merge inserts / updates from a srcTable into a target table. Assuming the srcTable has some or all of the tgtTable fields with matching types

func (*BaseConn) MustExec 

func (conn *BaseConn) MustExec(sql string, args ...interface{}) (result sql.Result)

MustExec execs the query using e and panics if there was an error. Any placeholder parameters are replaced with supplied args.

func (*BaseConn) NewTransaction 

func (conn *BaseConn) NewTransaction(ctx context.Context, options ...*sql.TxOptions) (Transaction, error)

NewTransaction creates a new transaction

func (*BaseConn) OptimizeTable 

func (conn *BaseConn) OptimizeTable(table *Table, newColumns iop.Columns, isTemp ...bool) (ok bool, err error)

func (*BaseConn) Prepare 

func (conn *BaseConn) Prepare(query string) (stmt *sql.Stmt, err error)

Prepare prepares the statement

func (*BaseConn) ProcessTemplate 

func (conn *BaseConn) ProcessTemplate(level, text string, values map[string]interface{}) (sql string, err error)

ProcessTemplate processes a template SQL text at a given level

func (*BaseConn) PropArr 

func (conn *BaseConn) PropArr() []string

PropArr returns an array of properties

func (*BaseConn) PropArrExclude added in v1.2.14 

func (conn *BaseConn) PropArrExclude(exclude ...string) []string

func (*BaseConn) Props 

func (conn *BaseConn) Props() map[string]string

Props returns a map properties

func (*BaseConn) Query 

func (conn *BaseConn) Query(sql string, options ...map[string]interface{}) (data iop.Dataset, err error)

Query runs a sql query, returns `result`, `error`

func (*BaseConn) QueryContext 

func (conn *BaseConn) QueryContext(ctx context.Context, sql string, options ...map[string]interface{}) (data iop.Dataset, err error)

QueryContext runs a sql query with ctx, returns `result`, `error`

func (*BaseConn) Quote 

func (conn *BaseConn) Quote(field string) string

Quote adds quotes to the field name

func (*BaseConn) ReplaceProps added in v1.2.16 

func (conn *BaseConn) ReplaceProps(newProps map[string]string)

ReplaceProps used when reusing a connection since the provided props can change, this is used to delete old original props and set new ones

func (*BaseConn) Rollback 

func (conn *BaseConn) Rollback() (err error)

Rollback rolls back a connection wide transaction

func (*BaseConn) RunAnalysis 

func (conn *BaseConn) RunAnalysis(analysisName string, values map[string]interface{}) (data iop.Dataset, err error)

RunAnalysis runs an analysis

func (*BaseConn) Schemata 

func (conn *BaseConn) Schemata() Schemata

Schemata returns the Schemata object

func (*BaseConn) Self 

func (conn *BaseConn) Self() Connection

Self returns the respective connection Instance This is useful to refer back to a subclass method from the superclass level. (Aka overloading)

func (*BaseConn) SetProp 

func (conn *BaseConn) SetProp(key string, val string)

SetProp sets the value of a property

func (*BaseConn) StreamRecords 

func (conn *BaseConn) StreamRecords(sql string) (<-chan map[string]interface{}, error)

StreamRecords the records of a sql query, returns `result`, `error`

func (*BaseConn) StreamRows 

func (conn *BaseConn) StreamRows(sql string, options ...map[string]interface{}) (ds *iop.Datastream, err error)

StreamRows the rows of a sql query, returns `result`, `error`

func (*BaseConn) StreamRowsContext 

func (conn *BaseConn) StreamRowsContext(ctx context.Context, query string, options ...map[string]interface{}) (ds *iop.Datastream, err error)

StreamRowsContext streams the rows of a sql query with context, returns `result`, `error`

func (*BaseConn) SubmitTemplate added in v1.2.4 

func (conn *BaseConn) SubmitTemplate(level string, templateMap map[string]string, name string, values map[string]interface{}) (data iop.Dataset, err error)

func (*BaseConn) SwapTable 

func (conn *BaseConn) SwapTable(srcTable string, tgtTable string) (err error)

SwapTable swaps two table

func (*BaseConn) TableExists added in v1.4.10 

func (conn *BaseConn) TableExists(table Table) (exists bool, err error)

TableExists returns true if the table exists

func (*BaseConn) Template 

func (conn *BaseConn) Template() dbio.Template

Template returns the Template object

func (*BaseConn) Tx 

func (conn *BaseConn) Tx() Transaction

Tx returns the current sqlx tx object

func (*BaseConn) Unquote 

func (conn *BaseConn) Unquote(field string) string

Unquote removes quotes to the field name

func (*BaseConn) UseADBC added in v1.5.2 

func (conn *BaseConn) UseADBC() bool

UseADBC returns if connection should use ADBC

func (*BaseConn) ValidateColumnNames 

func (conn *BaseConn) ValidateColumnNames(tgtCols iop.Columns, colNames []string) (newCols iop.Columns, err error)

ValidateColumnNames verifies that source fields are present in the target table It will return quoted field names as `newColNames`, the same length as `colNames`

type BaseTransaction 

type BaseTransaction struct {
	Tx   *sqlx.Tx
	Conn Connection
	// contains filtered or unexported fields
}

BaseTransaction is a database transaction

func (*BaseTransaction) Commit 

func (t *BaseTransaction) Commit() (err error)

func (*BaseTransaction) Connection 

func (t *BaseTransaction) Connection() Connection

Connection return the connection

func (*BaseTransaction) Context 

func (t *BaseTransaction) Context() *g.Context

Commit commits connection wide transaction

func (*BaseTransaction) DisableTrigger 

func (t *BaseTransaction) DisableTrigger(tableName, triggerName string) (err error)

DisableTrigger disables a trigger

func (*BaseTransaction) EnableTrigger 

func (t *BaseTransaction) EnableTrigger(tableName, triggerName string) (err error)

EnableTrigger enables a trigger

func (*BaseTransaction) Exec 

func (t *BaseTransaction) Exec(sql string, args ...interface{}) (result sql.Result, err error)

Exec runs a sql query, returns `error`

func (*BaseTransaction) ExecContext 

func (t *BaseTransaction) ExecContext(ctx context.Context, q string, args ...interface{}) (result sql.Result, err error)

ExecContext runs a sql query with context, returns `error`

func (*BaseTransaction) ExecMultiContext 

func (t *BaseTransaction) ExecMultiContext(ctx context.Context, q string, args ...interface{}) (result sql.Result, err error)

ExecMultiContext runs multiple sql queries with context, returns `error`

func (*BaseTransaction) InsertBatchStream 

func (t *BaseTransaction) InsertBatchStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

InsertBatchStream inserts a stream into a table in batch

func (*BaseTransaction) InsertStream 

func (t *BaseTransaction) InsertStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

InsertStream inserts a stream into a table

func (*BaseTransaction) Merge added in v1.4.20 

func (t *BaseTransaction) Merge(sourceTable, targetTable string, pkFields []string) (count uint64, err error)

Merge does an upsert from source table into target table

func (*BaseTransaction) Prepare 

func (t *BaseTransaction) Prepare(query string) (stmt *sql.Stmt, err error)

Prepare prepares the statement

func (*BaseTransaction) QueryContext 

func (t *BaseTransaction) QueryContext(ctx context.Context, q string, args ...interface{}) (result *sqlx.Rows, err error)

QueryContext queries rows

func (*BaseTransaction) Rollback 

func (t *BaseTransaction) Rollback() (err error)

Rollback rolls back connection wide transaction

func (*BaseTransaction) UpsertStream 

func (t *BaseTransaction) UpsertStream(tableFName string, ds *iop.Datastream, pk []string) (count uint64, err error)

UpsertStream inserts a stream into a table in batch

type BigQueryConn 

type BigQueryConn struct {
	BaseConn
	URL       string
	Client    *bigquery.Client
	ProjectID string
	DatasetID string
	Location  string
	Datasets  []string
	Mux       sync.Mutex
}

BigQueryConn is a Google Big Query connection

func (*BigQueryConn) BulkExportFlow 

func (conn *BigQueryConn) BulkExportFlow(table Table) (df *iop.Dataflow, err error)

BulkExportFlow reads in bulk

func (*BigQueryConn) BulkImportFlow 

func (conn *BigQueryConn) BulkImportFlow(tableFName string, df *iop.Dataflow) (count uint64, err error)

BulkImportFlow inserts a flow of streams into a table. For redshift we need to create CSVs in GCS and then use the COPY command.

func (*BigQueryConn) BulkImportStream 

func (conn *BigQueryConn) BulkImportStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

BulkImportStream demonstrates loading data into a BigQuery table using a file on the local filesystem.

func (*BigQueryConn) CastColumnForSelect 

func (conn *BigQueryConn) CastColumnForSelect(srcCol iop.Column, tgtCol iop.Column) (selectStr string)

CastColumnForSelect casts to the correct target column type

func (*BigQueryConn) Close 

func (conn *BigQueryConn) Close() error

Close closes the connection

func (*BigQueryConn) Connect 

func (conn *BigQueryConn) Connect(timeOut ...int) error

Connect connects to the database

func (*BigQueryConn) CopyFromGCS 

func (conn *BigQueryConn) CopyFromGCS(gcsURI string, table Table, dsColumns []iop.Column, format dbio.FileType) error

func (*BigQueryConn) CopyFromLocal 

func (conn *BigQueryConn) CopyFromLocal(localURI string, table Table, dsColumns iop.Columns, format dbio.FileType) error

CopyFromGCS into bigquery from google storage

func (*BigQueryConn) CopyToGCS 

func (conn *BigQueryConn) CopyToGCS(table Table, gcsURI string, format dbio.FileType) error

func (*BigQueryConn) ExecContext 

func (conn *BigQueryConn) ExecContext(ctx context.Context, sql string, args ...interface{}) (result sql.Result, err error)

func (*BigQueryConn) ExportToGCS 

func (conn *BigQueryConn) ExportToGCS(sql string, gcsURI string, format dbio.FileType) error

CopyToGCS Copy table to gc storage

func (*BigQueryConn) GenerateDDL added in v1.1.4 

func (conn *BigQueryConn) GenerateDDL(table Table, data iop.Dataset, temporary bool) (sql string, err error)

GenerateDDL generates a DDL based on a dataset

func (*BigQueryConn) GenerateMergeSQL added in v1.4.20 

func (conn *BigQueryConn) GenerateMergeSQL(srcTable string, tgtTable string, pkFields []string) (sql string, err error)

GenerateMergeSQL generates the upsert SQL

func (*BigQueryConn) GetDatabases 

func (conn *BigQueryConn) GetDatabases() (iop.Dataset, error)

GetDatabases returns databases

func (*BigQueryConn) GetSchemas 

func (conn *BigQueryConn) GetSchemas() (iop.Dataset, error)

GetSchemas returns schemas

func (*BigQueryConn) GetSchemata 

func (conn *BigQueryConn) GetSchemata(level SchemataLevel, schemaName string, tableNames ...string) (Schemata, error)

GetSchemata obtain full schemata info for a schema and/or table in current database

func (*BigQueryConn) Init 

func (conn *BigQueryConn) Init() error

Init initiates the object

func (*BigQueryConn) InsertBatchStream 

func (conn *BigQueryConn) InsertBatchStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

InsertBatchStream inserts a stream into a table in batch

func (*BigQueryConn) InsertStream 

func (conn *BigQueryConn) InsertStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

InsertStream demonstrates loading data into a BigQuery table using a file on the local filesystem.

func (*BigQueryConn) LoadFromReader added in v1.4.24 

func (conn *BigQueryConn) LoadFromReader(table Table, reader io.Reader, dsColumns []iop.Column, format dbio.FileType) error

LoadFromReader demonstrates loading data into a BigQuery table using a file on the local filesystem. https://cloud.google.com/bigquery/docs/batch-loading-data#loading_data_from_local_files

func (*BigQueryConn) NewTransaction 

func (conn *BigQueryConn) NewTransaction(ctx context.Context, options ...*sql.TxOptions) (tx Transaction, err error)

NewTransaction creates a new transaction

func (*BigQueryConn) StreamRowsContext 

func (conn *BigQueryConn) StreamRowsContext(ctx context.Context, sql string, options ...map[string]interface{}) (ds *iop.Datastream, err error)

func (*BigQueryConn) Unload 

func (conn *BigQueryConn) Unload(tables ...Table) (gsPath string, err error)

Unload to Google Cloud Storage

type BigTableAction 

type BigTableAction string
const BTCreateColumnFamily BigTableAction = "create_column_family"
const BTCreateTable BigTableAction = "create_table"
const BTDeleteTable BigTableAction = "delete_table"
const BTTableInfo BigTableAction = "table_info"

type BigTableConn 

type BigTableConn struct {
	BaseConn
	URL        string
	Client     *bigtable.Client
	ProjectID  string
	InstanceID string
	Location   string
}

BigTableConn is a Google Big Query connection

func (*BigTableConn) BulkExportFlow 

func (conn *BigTableConn) BulkExportFlow(table Table) (df *iop.Dataflow, err error)

func (*BigTableConn) Close 

func (conn *BigTableConn) Close() error

Close closes the connection

func (*BigTableConn) Connect 

func (conn *BigTableConn) Connect(timeOut ...int) error

Connect connects to the database

func (*BigTableConn) ExecContext 

func (conn *BigTableConn) ExecContext(ctx context.Context, payload string, args ...interface{}) (result sql.Result, err error)

func (*BigTableConn) GetColumns 

func (conn *BigTableConn) GetColumns(tableFName string, fields ...string) (columns iop.Columns, err error)

func (*BigTableConn) GetColumnsFull 

func (conn *BigTableConn) GetColumnsFull(tableFName string) (iop.Dataset, error)

func (*BigTableConn) GetSQLColumns 

func (conn *BigTableConn) GetSQLColumns(table Table) (columns iop.Columns, err error)

GetTables returns tables for given schema

func (*BigTableConn) GetSchemas 

func (conn *BigTableConn) GetSchemas() (iop.Dataset, error)

func (*BigTableConn) GetSchemata 

func (conn *BigTableConn) GetSchemata(level SchemataLevel, schemaName string, tableNames ...string) (schemata Schemata, err error)

func (*BigTableConn) GetTables 

func (conn *BigTableConn) GetTables(schema string) (data iop.Dataset, err error)

func (*BigTableConn) GetViews 

func (conn *BigTableConn) GetViews(schema string) (data iop.Dataset, err error)

GetTables returns tables for given schema

func (*BigTableConn) Init 

func (conn *BigTableConn) Init() error

Init initiates the object

func (*BigTableConn) InsertBatchStream 

func (conn *BigTableConn) InsertBatchStream(table string, ds *iop.Datastream) (count uint64, err error)

InsertBatchStream inserts a stream into a table in batch

func (*BigTableConn) NewTransaction 

func (conn *BigTableConn) NewTransaction(ctx context.Context, options ...*sql.TxOptions) (tx Transaction, err error)

NewTransaction creates a new transaction

func (*BigTableConn) StreamRowsContext 

func (conn *BigTableConn) StreamRowsContext(ctx context.Context, table string, options ...map[string]interface{}) (ds *iop.Datastream, err error)

type BigTableQuery 

type BigTableQuery struct {
	Action         BigTableAction `json:"action"`
	Table          string         `json:"table"`
	ColumnFamilies []string       `json:"column_family"`
}

type BlankTransaction 

type BlankTransaction struct {
	Conn Connection
	// contains filtered or unexported fields
}

func (*BlankTransaction) Commit 

func (t *BlankTransaction) Commit() (err error)

func (*BlankTransaction) Connection 

func (t *BlankTransaction) Connection() Connection

func (*BlankTransaction) Context 

func (t *BlankTransaction) Context() *g.Context

func (*BlankTransaction) ExecContext 

func (t *BlankTransaction) ExecContext(ctx context.Context, q string, args ...interface{}) (result sql.Result, err error)

func (*BlankTransaction) ExecMultiContext 

func (t *BlankTransaction) ExecMultiContext(ctx context.Context, q string, args ...interface{}) (result sql.Result, err error)

func (*BlankTransaction) Prepare 

func (t *BlankTransaction) Prepare(query string) (stmt *sql.Stmt, err error)

func (*BlankTransaction) QueryContext 

func (t *BlankTransaction) QueryContext(ctx context.Context, q string, args ...interface{}) (result *sqlx.Rows, err error)

func (*BlankTransaction) Rollback 

func (t *BlankTransaction) Rollback() (err error)

type Chunk added in v1.4.14 

type Chunk struct {
	RangeStart string `json:"range_start,omitempty"`
	RangeEnd   string `json:"range_end,omitempty"`
	Where      string `json:"where,omitempty"`
}

func (*Chunk) Range added in v1.4.14 

func (c *Chunk) Range() string

type ClickhouseConn 

type ClickhouseConn struct {
	BaseConn
	URL string
	// contains filtered or unexported fields
}

ClickhouseConn is a Clikchouse connection

func (*ClickhouseConn) BulkExportStream added in v1.4.18 

func (conn *ClickhouseConn) BulkExportStream(table Table) (ds *iop.Datastream, err error)

BulkExportStream streams the rows in bulk

func (*ClickhouseConn) BulkImportStream 

func (conn *ClickhouseConn) BulkImportStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

BulkImportStream inserts a stream into a table

func (*ClickhouseConn) CastColumnForSelect added in v1.5.1 

func (conn *ClickhouseConn) CastColumnForSelect(srcCol iop.Column, tgtCol iop.Column) (selectStr string)

CastColumnForSelect casts to the correct target column type

func (*ClickhouseConn) ConnString 

func (conn *ClickhouseConn) ConnString() string

func (*ClickhouseConn) Connect added in v1.1.12 

func (conn *ClickhouseConn) Connect(timeOut ...int) (err error)

func (*ClickhouseConn) GenerateDDL added in v1.1.4 

func (conn *ClickhouseConn) GenerateDDL(table Table, data iop.Dataset, temporary bool) (ddl string, err error)

GenerateDDL generates a DDL based on a dataset

func (*ClickhouseConn) GenerateInsertStatement 

func (conn *ClickhouseConn) GenerateInsertStatement(tableName string, cols iop.Columns, numRows int) string

GenerateInsertStatement returns the proper INSERT statement

func (*ClickhouseConn) GenerateMergeSQL added in v1.4.20 

func (conn *ClickhouseConn) GenerateMergeSQL(srcTable string, tgtTable string, pkFields []string) (sql string, err error)

GenerateMergeSQL generates the upsert SQL

func (*ClickhouseConn) GetNativeType added in v1.2.15 

func (conn *ClickhouseConn) GetNativeType(col iop.Column) (nativeType string, err error)

func (*ClickhouseConn) Init 

func (conn *ClickhouseConn) Init() error

Init initiates the object

func (*ClickhouseConn) Merge added in v1.4.21 

func (conn *ClickhouseConn) Merge(srcTable string, tgtTable string, primKeys []string) (rowAffCnt int64, err error)

Merge inserts / updates from a srcTable into a target table. Assuming the srcTable has some or all of the tgtTable fields with matching types

func (*ClickhouseConn) NewTransaction 

func (conn *ClickhouseConn) NewTransaction(ctx context.Context, options ...*sql.TxOptions) (Transaction, error)

NewTransaction creates a new transaction

func (*ClickhouseConn) Version added in v1.4.21 

func (conn *ClickhouseConn) Version() int

type ColumnType 

type ColumnType struct {
	Name             string
	DatabaseTypeName string
	FetchedColumn    *iop.Column
	Length           int
	Precision        int
	Scale            int
	Nullable         bool
	CT               *sql.ColumnType
	Sourced          bool
}

func (*ColumnType) IsSourced added in v1.3.5 

func (ct *ColumnType) IsSourced() bool

type ConnInfo 

type ConnInfo struct {
	Host      string
	Port      int
	Database  string
	User      string
	Password  string
	Schema    string
	Warehouse string
	Role      string
	URL       *net.URL
}

type Connection 

type Connection interface {
	Base() *BaseConn
	BaseURL() string
	Begin(options ...*sql.TxOptions) error
	BeginContext(ctx context.Context, options ...*sql.TxOptions) error
	BulkExportFlow(table Table) (*iop.Dataflow, error)
	BulkExportStream(table Table) (*iop.Datastream, error)
	BulkImportFlow(tableFName string, df *iop.Dataflow) (count uint64, err error)
	BulkImportStream(tableFName string, ds *iop.Datastream) (count uint64, err error)
	CastColumnForSelect(srcColumn iop.Column, tgtColumn iop.Column) string
	CastColumnsForSelect(srcColumns iop.Columns, tgtColumns iop.Columns) []string
	Close() error
	Commit() error
	CompareChecksums(tableName string, columns iop.Columns) (err error)
	Connect(timeOut ...int) error
	ConnString() string
	Context() *g.Context
	CreateTable(tableName string, cols iop.Columns, tableDDL string) (err error)
	CreateTemporaryTable(tableName string, cols iop.Columns) (err error)
	CurrentDatabase() (string, error)
	Db() *sqlx.DB
	DbX() *DbX
	DropTable(...string) error
	DropView(...string) error
	Exec(sql string, args ...interface{}) (result sql.Result, err error)
	ExecContext(ctx context.Context, sql string, args ...interface{}) (result sql.Result, err error)
	ExecMulti(sqls ...string) (result sql.Result, err error)
	ExecMultiContext(ctx context.Context, sqls ...string) (result sql.Result, err error)
	GenerateDDL(table Table, data iop.Dataset, temporary bool) (string, error)
	GenerateInsertStatement(tableName string, cols iop.Columns, numRows int) string
	GenerateMergeSQL(srcTable string, tgtTable string, pkFields []string) (sql string, err error)
	GetAnalysis(string, map[string]interface{}) (string, error)
	GetColumns(tableFName string, fields ...string) (iop.Columns, error)
	GetColumnsFull(string) (iop.Dataset, error)
	GetColumnStats(tableName string, fields ...string) (columns iop.Columns, err error)
	GetCount(string) (int64, error)
	GetDatabases() (iop.Dataset, error)
	GetMaxValue(t Table, colName string) (value any, col iop.Column, err error)
	GetDDL(string) (string, error)
	GetGormConn(config *gorm.Config) (*gorm.DB, error)
	GetIndexes(string) (iop.Dataset, error)
	GetNativeType(col iop.Column) (nativeType string, err error)
	GetPrimaryKeys(string) (iop.Dataset, error)
	GetProp(...string) string
	GetSchemas() (iop.Dataset, error)
	GetSchemata(level SchemataLevel, schemaName string, tableNames ...string) (Schemata, error)
	GetSQLColumns(table Table) (columns iop.Columns, err error)
	GetTableColumns(table *Table, fields ...string) (columns iop.Columns, err error)
	GetTablesAndViews(string) (iop.Dataset, error)
	GetTables(string) (iop.Dataset, error)
	GetTemplateValue(path string) (value string)
	GetType() dbio.Type
	GetURL(newURL ...string) string
	GetViews(string) (iop.Dataset, error)
	Info() ConnInfo
	Init() error
	InsertBatchStream(tableFName string, ds *iop.Datastream) (count uint64, err error)
	InsertStream(tableFName string, ds *iop.Datastream) (count uint64, err error)
	Kill() error
	LoadTemplates() error
	MustExec(sql string, args ...interface{}) (result sql.Result)
	NewTransaction(ctx context.Context, options ...*sql.TxOptions) (Transaction, error)
	OptimizeTable(table *Table, columns iop.Columns, isTemp ...bool) (ok bool, err error)
	Prepare(query string) (stmt *sql.Stmt, err error)
	ProcessTemplate(level, text string, values map[string]interface{}) (sql string, err error)
	Props() map[string]string
	PropsArr() []string
	Query(sql string, options ...map[string]interface{}) (iop.Dataset, error)
	QueryContext(ctx context.Context, sql string, options ...map[string]interface{}) (iop.Dataset, error)
	Quote(field string) string
	RenameTable(table string, newTable string) (err error)
	Rollback() error
	RunAnalysis(string, map[string]interface{}) (iop.Dataset, error)
	Schemata() Schemata
	Self() Connection
	SetProp(string, string)
	StreamRecords(sql string) (<-chan map[string]interface{}, error)
	StreamRows(sql string, options ...map[string]interface{}) (*iop.Datastream, error)
	StreamRowsContext(ctx context.Context, sql string, options ...map[string]interface{}) (ds *iop.Datastream, err error)
	SubmitTemplate(level string, templateMap map[string]string, name string, values map[string]interface{}) (data iop.Dataset, err error)
	SwapTable(srcTable string, tgtTable string) (err error)
	TableExists(table Table) (exists bool, err error)
	Template() dbio.Template
	Tx() Transaction
	Unquote(string) string
	Merge(srcTable string, tgtTable string, pkFields []string) (rowAffCnt int64, err error)
	ValidateColumnNames(tgtCols iop.Columns, colNames []string) (newCols iop.Columns, err error)
	AddMissingColumns(table Table, newCols iop.Columns) (ok bool, err error)
	UseADBC() bool
}

Connection is the Base interface for Connections

func Clone 

func Clone(conn Connection) (newConn Connection, err error)

func NewAdbcConn added in v1.5.2 

func NewAdbcConn(parentConn Connection) (adbcConn Connection, err error)

NewAdbcConn creates a new ADBC conn from a parent conn constructs the connection string with complete URIs/paths for each database type

func NewConn 

func NewConn(URL string, props ...string) (Connection, error)

NewConn return the most proper connection for a given database

func NewConnContext 

func NewConnContext(ctx context.Context, URL string, props ...string) (Connection, error)

NewConnContext return the most proper connection for a given database with context props are provided as `"Prop1=Value1", "Prop2=Value2", ...`

type D1Conn added in v1.3.5 

type D1Conn struct {
	SQLiteConn
	URL       string
	AccountID string
	Database  string
	UUID      string
	APIToken  string
	// contains filtered or unexported fields
}

D1Conn is a Cloudflare SQLite connection

func (*D1Conn) BulkImportStream added in v1.3.5 

func (conn *D1Conn) BulkImportStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

func (*D1Conn) Connect added in v1.3.5 

func (conn *D1Conn) Connect(timeOut ...int) (err error)

Connect connects to the database

func (*D1Conn) ExecContext added in v1.3.5 

func (conn *D1Conn) ExecContext(ctx context.Context, q string, args ...interface{}) (result sql.Result, err error)

ExecContext runs a sql query with context, returns `error`

func (*D1Conn) GenerateMergeSQL added in v1.4.20 

func (conn *D1Conn) GenerateMergeSQL(srcTable string, tgtTable string, pkFields []string) (sql string, err error)

GenerateMergeSQL generates the upsert SQL

func (*D1Conn) GetDatabases added in v1.4.6 

func (conn *D1Conn) GetDatabases() (data iop.Dataset, err error)

GetDatabases returns databases for given connection

func (*D1Conn) GetSchemata added in v1.3.5 

func (conn *D1Conn) GetSchemata(level SchemataLevel, schemaName string, tableNames ...string) (Schemata, error)

GetSchemata obtain full schemata info for a schema and/or table in current database

func (*D1Conn) Init added in v1.3.5 

func (conn *D1Conn) Init() error

Init initiates the object

func (*D1Conn) InsertBatchStream added in v1.3.5 

func (conn *D1Conn) InsertBatchStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

func (*D1Conn) StreamRowsContext added in v1.3.5 

func (conn *D1Conn) StreamRowsContext(ctx context.Context, query string, options ...map[string]interface{}) (ds *iop.Datastream, err error)

type DataAnalyzer 

type DataAnalyzer struct {
	Conn        Connection
	Schemata    Schemata
	ColumnMap   map[string]iop.Column
	RelationMap map[string]map[string]map[string]Relation // table > column A > column B > relation
	Options     DataAnalyzerOptions
}

func NewDataAnalyzer 

func NewDataAnalyzer(conn Connection, opts DataAnalyzerOptions) (da *DataAnalyzer, err error)

func (*DataAnalyzer) AnalyzeColumns 

func (da *DataAnalyzer) AnalyzeColumns(sampleSize int, includeViews bool) (err error)

func (*DataAnalyzer) GetManyToMany 

func (da *DataAnalyzer) GetManyToMany(nonUniqueCols iop.Columns, asString bool) (err error)

func (*DataAnalyzer) GetOneToMany 

func (da *DataAnalyzer) GetOneToMany(uniqueCols, nonUniqueCols iop.Columns, asString bool) (err error)

func (*DataAnalyzer) GetOneToOne 

func (da *DataAnalyzer) GetOneToOne(uniqueCols iop.Columns, asString bool) (err error)

func (*DataAnalyzer) GetSchemata 

func (da *DataAnalyzer) GetSchemata(force bool) (err error)

func (*DataAnalyzer) ProcessRelations 

func (da *DataAnalyzer) ProcessRelations() (err error)

func (*DataAnalyzer) ProcessRelationsInteger 

func (da *DataAnalyzer) ProcessRelationsInteger() (err error)

func (*DataAnalyzer) ProcessRelationsString 

func (da *DataAnalyzer) ProcessRelationsString() (err error)

func (*DataAnalyzer) WriteRelationsYaml 

func (da *DataAnalyzer) WriteRelationsYaml(path string) (err error)

type DataAnalyzerOptions 

type DataAnalyzerOptions struct {
	DbName      string
	SchemaNames []string
}

type Database 

type Database struct {
	Name    string            `json:"name"`
	Schemas map[string]Schema `json:"schemas"`
}

Database represents a schemata database

func (*Database) Columns 

func (db *Database) Columns() map[string]iop.Column

func (*Database) Tables 

func (db *Database) Tables() map[string]Table

type DatabricksConn added in v1.4.11 

type DatabricksConn struct {
	BaseConn
	URL        string
	Catalog    string
	Schema     string
	Warehouse  string
	CopyMethod string
}

DatabricksConn is a Databricks connection

func (*DatabricksConn) BulkExportFlow added in v1.4.11 

func (conn *DatabricksConn) BulkExportFlow(table Table) (df *iop.Dataflow, err error)

BulkExportFlow reads in bulk

func (*DatabricksConn) BulkImportFlow added in v1.4.11 

func (conn *DatabricksConn) BulkImportFlow(tableFName string, df *iop.Dataflow) (count uint64, err error)

BulkImportFlow inserts a flow of streams into a table

func (*DatabricksConn) BulkImportStream added in v1.4.11 

func (conn *DatabricksConn) BulkImportStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

BulkImportStream inserts a stream into a table

func (*DatabricksConn) CastColumnForSelect added in v1.4.11 

func (conn *DatabricksConn) CastColumnForSelect(srcCol iop.Column, tgtCol iop.Column) (selectStr string)

CastColumnForSelect casts to the correct target column type

func (*DatabricksConn) Connect added in v1.4.11 

func (conn *DatabricksConn) Connect(timeOut ...int) error

Connect connects to the database

func (*DatabricksConn) CopyFromS3 added in v1.4.11 

func (conn *DatabricksConn) CopyFromS3(tableFName, s3Path, fileFormat string) (err error)

CopyFromS3 uses the Databricks COPY INTO command from AWS S3

func (*DatabricksConn) CopyFromVolume added in v1.4.11 

func (conn *DatabricksConn) CopyFromVolume(tableFName, volumePath string, fileFormat dbio.FileType, columns iop.Columns) error

CopyFromVolume uses the Databricks COPY INTO command from volumes

func (*DatabricksConn) CopyToS3 added in v1.4.11 

func (conn *DatabricksConn) CopyToS3(table Table) (s3Path string, err error)

CopyToS3 exports a query to an S3 location using CREATE EXTERNAL TABLE

func (*DatabricksConn) CopyViaS3 added in v1.4.11 

func (conn *DatabricksConn) CopyViaS3(tableFName string, df *iop.Dataflow) (count uint64, err error)

CopyViaS3 uses the Databricks COPY INTO command from AWS S3

func (*DatabricksConn) CopyViaVolume added in v1.4.11 

func (conn *DatabricksConn) CopyViaVolume(table Table, df *iop.Dataflow) (count uint64, err error)

CopyViaVolume uses Databricks volumes for bulk import, similar to Snowflake's CopyViaStage

func (*DatabricksConn) GenerateDDL added in v1.4.11 

func (conn *DatabricksConn) GenerateDDL(table Table, data iop.Dataset, temporary bool) (sql string, err error)

GenerateDDL generates a DDL based on a dataset

func (*DatabricksConn) GenerateMergeSQL added in v1.4.20 

func (conn *DatabricksConn) GenerateMergeSQL(srcTable string, tgtTable string, pkFields []string) (sql string, err error)

GenerateMergeSQL generates the upsert SQL

func (*DatabricksConn) GetColumnsFull added in v1.4.11 

func (conn *DatabricksConn) GetColumnsFull(tableFName string) (data iop.Dataset, err error)

GetColumnsFull returns full column information for a table

func (*DatabricksConn) GetSchemata added in v1.4.24 

func (conn *DatabricksConn) GetSchemata(level SchemataLevel, schemaName string, tableNames ...string) (Schemata, error)

func (*DatabricksConn) GetURL added in v1.4.11 

func (conn *DatabricksConn) GetURL(newURL ...string) string

GetURL returns the URL

func (*DatabricksConn) Init added in v1.4.11 

func (conn *DatabricksConn) Init() error

Init initiates the object

func (*DatabricksConn) NewTransaction added in v1.4.11 

func (conn *DatabricksConn) NewTransaction(ctx context.Context, options ...*sql.TxOptions) (tx Transaction, err error)

NewTransaction creates a new transaction

func (*DatabricksConn) UnloadViaVolume added in v1.4.11 

func (conn *DatabricksConn) UnloadViaVolume(tables ...Table) (filePath string, unloadedFiles int, err error)

UnloadViaVolume exports data to a Databricks volume, similar to Snowflake's UnloadViaStage

func (*DatabricksConn) VolumeDelete added in v1.4.11 

func (conn *DatabricksConn) VolumeDelete(volumePaths ...string) (err error)

VolumeDelete delete files in a Databricks volume path

func (*DatabricksConn) VolumeGET added in v1.4.11 

func (conn *DatabricksConn) VolumeGET(volumePath, folderPath, localFilePath string) error

VolumeGET downloads a file from a Databricks volume to local filesystem

func (*DatabricksConn) VolumeList added in v1.4.11 

func (conn *DatabricksConn) VolumeList(volumePath string) (data iop.Dataset, err error)

VolumeList lists files in a Databricks volume path

func (*DatabricksConn) VolumePUT added in v1.4.11 

func (conn *DatabricksConn) VolumePUT(folderPath, localFilePath, volumePath string) error

VolumePUT uploads a local file to a Databricks volume using SQL commands

type DbX 

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

DbX is db express

func (*DbX) Delete 

func (x *DbX) Delete(o interface{}) (cnt int, err error)

Delete deletes from object or slice

func (*DbX) Get 

func (x *DbX) Get(o interface{}, fields ...string) (err error)

Get retrieves object

func (*DbX) Insert 

func (x *DbX) Insert(o interface{}, fields ...string) (err error)

Insert inserts object or slice

func (*DbX) Merge added in v1.4.20 

func (x *DbX) Merge(o interface{}, fields ...string) (cnt int, err error)

Merge upserts from object or slice

func (*DbX) Select 

func (x *DbX) Select(o interface{}, fields ...string) (err error)

Select retrieves objects

func (*DbX) TableName 

func (x *DbX) TableName(o interface{}) (name string)

TableName returns the table name of object or slice

func (*DbX) Update 

func (x *DbX) Update(o interface{}, fields ...string) (cnt int, err error)

Update updates from object or slice

func (*DbX) Where 

func (x *DbX) Where(where ...interface{}) *DbX

Where adds a where clause

type DuckDbConn 

type DuckDbConn struct {
	BaseConn
	URL string
	// contains filtered or unexported fields
}

DuckDbConn is a Duck DB connection

func (*DuckDbConn) BulkImportFlow added in v1.2.19 

func (conn *DuckDbConn) BulkImportFlow(tableFName string, df *iop.Dataflow) (count uint64, err error)

func (*DuckDbConn) CastColumnForSelect added in v1.2.10 

func (conn *DuckDbConn) CastColumnForSelect(srcCol iop.Column, tgtCol iop.Column) (selectStr string)

CastColumnForSelect casts to the correct target column type

func (*DuckDbConn) Close 

func (conn *DuckDbConn) Close() (err error)

func (*DuckDbConn) Connect 

func (conn *DuckDbConn) Connect(timeOut ...int) (err error)

func (*DuckDbConn) DuckDb added in v1.2.25 

func (conn *DuckDbConn) DuckDb() *iop.DuckDb

DuckDb returns the DuckDb instance

func (*DuckDbConn) ExecContext 

func (conn *DuckDbConn) ExecContext(ctx context.Context, sql string, args ...interface{}) (result sql.Result, err error)

func (*DuckDbConn) ExecMultiContext 

func (conn *DuckDbConn) ExecMultiContext(ctx context.Context, sqls ...string) (result sql.Result, err error)

ExecContext runs a sql query with context, returns `error`

func (*DuckDbConn) GenerateMergeSQL added in v1.4.20 

func (conn *DuckDbConn) GenerateMergeSQL(srcTable string, tgtTable string, pkFields []string) (sql string, err error)

GenerateMergeSQL generates the upsert SQL

func (*DuckDbConn) GetURL 

func (conn *DuckDbConn) GetURL(newURL ...string) string

GetURL returns the processed URL

func (*DuckDbConn) Init 

func (conn *DuckDbConn) Init() error

Init initiates the object

func (*DuckDbConn) InsertBatchStream 

func (conn *DuckDbConn) InsertBatchStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

InsertBatchStream inserts a stream into a table in batch

func (*DuckDbConn) InsertStream 

func (conn *DuckDbConn) InsertStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

InsertStream demonstrates loading data into a BigQuery table using a file on the local filesystem.

func (*DuckDbConn) StreamRowsContext 

func (conn *DuckDbConn) StreamRowsContext(ctx context.Context, sql string, options ...map[string]interface{}) (ds *iop.Datastream, err error)

type DuckLakeConn added in v1.4.10 

type DuckLakeConn struct {
	DuckDbConn

	// Catalog database configuration
	CatalogType    string // duckdb, sqlite, postgres, mysql
	CatalogSchema  string // schema to use in postgres or mysql
	CatalogConnStr string // connection string for catalog database
	DataPath       string // path to data files (local or cloud storage)
	Database       string // the database name to attached
	Encrypted      bool   // whether data is written using Parquet encryption
	InliningLimit  int    // where to enable data inlining when attaching
}

DuckLakeConn is a Ducklake connection

func (*DuckLakeConn) Close added in v1.4.10 

func (conn *DuckLakeConn) Close() error

Close closes the DuckLake connection

func (*DuckLakeConn) Connect added in v1.4.10 

func (conn *DuckLakeConn) Connect(timeOut ...int) (err error)

Connect establishes the DuckLake connection

func (*DuckLakeConn) GenerateMergeSQL added in v1.4.20 

func (conn *DuckLakeConn) GenerateMergeSQL(srcTable string, tgtTable string, pkFields []string) (sql string, err error)

GenerateMergeSQL generates the upsert SQL for DuckLake

func (*DuckLakeConn) GetURL added in v1.4.10 

func (conn *DuckLakeConn) GetURL(newURL ...string) string

GetURL returns the processed URL

func (*DuckLakeConn) Init added in v1.4.10 

func (conn *DuckLakeConn) Init() error

Init initiates the object

func (*DuckLakeConn) SubmitTemplate added in v1.4.10 

func (conn *DuckLakeConn) SubmitTemplate(level string, templateMap map[string]string, name string, values map[string]any) (data iop.Dataset, err error)

type ElasticsearchConn added in v1.3.5 

type ElasticsearchConn struct {
	BaseConn
	URL    string
	Client *elasticsearch.Client
}

ElasticsearchConn is a elasticsearch connection

func (*ElasticsearchConn) BulkExportFlow added in v1.3.5 

func (conn *ElasticsearchConn) BulkExportFlow(table Table) (df *iop.Dataflow, err error)

func (*ElasticsearchConn) Close added in v1.3.5 

func (conn *ElasticsearchConn) Close() error

func (*ElasticsearchConn) Connect added in v1.3.5 

func (conn *ElasticsearchConn) Connect(timeOut ...int) error

Connect connects to the database

func (*ElasticsearchConn) ExecContext added in v1.3.5 

func (conn *ElasticsearchConn) ExecContext(ctx context.Context, sql string, args ...interface{}) (result sql.Result, err error)

func (*ElasticsearchConn) GetSchemas added in v1.3.5 

func (conn *ElasticsearchConn) GetSchemas() (data iop.Dataset, err error)

GetSchemas returns schemas

func (*ElasticsearchConn) GetSchemata added in v1.3.5 

func (conn *ElasticsearchConn) GetSchemata(level SchemataLevel, schema string, tables ...string) (schemata Schemata, err error)

GetSchemata returns the database schemata

func (*ElasticsearchConn) GetTableColumns added in v1.3.5 

func (conn *ElasticsearchConn) GetTableColumns(table *Table, fields ...string) (columns iop.Columns, err error)

GetTableColumns returns columns for a table

func (*ElasticsearchConn) GetTables added in v1.3.5 

func (conn *ElasticsearchConn) GetTables(schema string) (data iop.Dataset, err error)

GetTables returns tables

func (*ElasticsearchConn) Init added in v1.3.5 

func (conn *ElasticsearchConn) Init() error

Init initiates the object

func (*ElasticsearchConn) NewTransaction added in v1.3.5 

func (conn *ElasticsearchConn) NewTransaction(ctx context.Context, options ...*sql.TxOptions) (tx Transaction, err error)

NewTransaction creates a new transaction

func (*ElasticsearchConn) StreamRowsContext added in v1.3.5 

func (conn *ElasticsearchConn) StreamRowsContext(ctx context.Context, tableName string, Opts ...map[string]any) (ds *iop.Datastream, err error)

type ExasolConn added in v1.4.16 

type ExasolConn struct {
	BaseConn
	URL string
}

ExasolConn is a Exasol connection

func (*ExasolConn) BulkImportFlow added in v1.4.16 

func (conn *ExasolConn) BulkImportFlow(tableFName string, df *iop.Dataflow) (count uint64, err error)

BulkImportFlow performs bulk import for Exasol using temporary CSV files

func (*ExasolConn) BulkImportStream added in v1.4.16 

func (conn *ExasolConn) BulkImportStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

BulkImportStream performs bulk import for Exasol

func (*ExasolConn) Connect added in v1.4.16 

func (conn *ExasolConn) Connect(timeOut ...int) error

Connect connects to the database

func (*ExasolConn) GenerateDDL added in v1.4.16 

func (conn *ExasolConn) GenerateDDL(table Table, data iop.Dataset, temporary bool) (string, error)

GenerateDDL generates DDL for Exasol

func (*ExasolConn) GenerateMergeSQL added in v1.4.20 

func (conn *ExasolConn) GenerateMergeSQL(srcTable string, tgtTable string, pkFields []string) (sql string, err error)

GenerateMergeSQL generates the upsert SQL for Exasol

func (*ExasolConn) GetURL added in v1.4.16 

func (conn *ExasolConn) GetURL(newURL ...string) string

GetURL returns the processed URL This method uses the official Exasol driver's configuration builder (exasol.NewConfig) to properly handle special characters in passwords and other connection parameters. This ensures passwords with backslashes, semicolons, etc. are correctly escaped.

func (*ExasolConn) Init added in v1.4.16 

func (conn *ExasolConn) Init() error

Init initiates the connection

func (*ExasolConn) NewTransaction added in v1.4.16 

func (conn *ExasolConn) NewTransaction(ctx context.Context, options ...*sql.TxOptions) (tx Transaction, err error)

NewTransaction creates a new transaction

type IcebergConn added in v1.4.10 

type IcebergConn struct {
	BaseConn
	URL            string
	CatalogType    dbio.IcebergCatalogType
	CatalogSQLConn Connection
	Catalog        catalog.Catalog
	Warehouse      string
	// contains filtered or unexported fields
}

IcebergConn is an Iceberg connection

func (*IcebergConn) BulkExportStream added in v1.4.10 

func (conn *IcebergConn) BulkExportStream(table Table) (ds *iop.Datastream, err error)

BulkExportStream reads table data in bulk

func (*IcebergConn) BulkImportStream added in v1.4.10 

func (conn *IcebergConn) BulkImportStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

BulkImportStream inserts a stream into a table using Arrow format. This method converts the incoming datastream to Apache Arrow format and uses the iceberg-go table.AppendTable API to write the data.

func (*IcebergConn) CastColumnForSelect added in v1.4.10 

func (conn *IcebergConn) CastColumnForSelect(srcCol iop.Column, tgtCol iop.Column) (selectStr string)

CastColumnForSelect casts to the correct target column type

func (*IcebergConn) Close added in v1.4.10 

func (conn *IcebergConn) Close() error

Close closes the connection

func (*IcebergConn) Connect added in v1.4.10 

func (conn *IcebergConn) Connect(timeOut ...int) (err error)

Connect connects to the Iceberg catalog

func (*IcebergConn) CreateNamespaceIfNotExists added in v1.4.10 

func (conn *IcebergConn) CreateNamespaceIfNotExists(schema string) (err error)

func (*IcebergConn) CreateTable added in v1.4.10 

func (conn *IcebergConn) CreateTable(tableName string, cols iop.Columns, tableDDL string) (err error)

func (*IcebergConn) DropTable added in v1.4.10 

func (conn *IcebergConn) DropTable(tableNames ...string) (err error)

DropTable drops given table.

func (*IcebergConn) ExecContext added in v1.4.10 

func (conn *IcebergConn) ExecContext(ctx context.Context, sql string, args ...interface{}) (result sql.Result, err error)

ExecContext executes a write operation

func (*IcebergConn) GenerateDDL added in v1.4.10 

func (conn *IcebergConn) GenerateDDL(table Table, data iop.Dataset, temporary bool) (sql string, err error)

GenerateDDL generates a DDL based on a dataset

func (*IcebergConn) GetColumns added in v1.4.10 

func (conn *IcebergConn) GetColumns(tableFName string, fields ...string) (columns iop.Columns, err error)

GetColumns returns the columns for a given table

func (*IcebergConn) GetCount added in v1.4.10 

func (conn *IcebergConn) GetCount(tableFName string) (count int64, err error)

GetCount returns -1 to skip validation

func (*IcebergConn) GetDataFiles added in v1.4.10 

func (conn *IcebergConn) GetDataFiles(t Table) (dataFiles []iceberg.DataFile, err error)

func (*IcebergConn) GetDatabases added in v1.4.10 

func (conn *IcebergConn) GetDatabases() (data iop.Dataset, err error)

GetDatabases returns databases (catalogs) for Iceberg connection

func (*IcebergConn) GetMaxValue added in v1.4.10 

func (conn *IcebergConn) GetMaxValue(t Table, colName string) (value any, maxCol iop.Column, err error)

GetMaxValue gets the maximum value of the given column

func (*IcebergConn) GetSchemas added in v1.4.10 

func (conn *IcebergConn) GetSchemas() (data iop.Dataset, err error)

GetSchemas returns the list of namespaces (schemas)

func (*IcebergConn) GetSchemata added in v1.4.10 

func (conn *IcebergConn) GetSchemata(level SchemataLevel, schemaName string, tableNames ...string) (Schemata, error)

GetSchemata obtain full schemata info for a schema and/or table

func (*IcebergConn) GetTableColumns added in v1.4.10 

func (conn *IcebergConn) GetTableColumns(table *Table, fields ...string) (columns iop.Columns, err error)

func (*IcebergConn) GetTables added in v1.4.10 

func (conn *IcebergConn) GetTables(schema string) (data iop.Dataset, err error)

GetTables returns tables for given schema

func (*IcebergConn) GetTablesAndViews added in v1.4.10 

func (conn *IcebergConn) GetTablesAndViews(schema string) (data iop.Dataset, err error)

GetTablesAndViews returns tables and views for given schema

func (*IcebergConn) GetViews added in v1.4.10 

func (conn *IcebergConn) GetViews(schema string) (data iop.Dataset, err error)

GetViews returns views for given schema

func (*IcebergConn) Init added in v1.4.10 

func (conn *IcebergConn) Init() error

Init initiates the object

func (*IcebergConn) InsertBatchStream added in v1.4.10 

func (conn *IcebergConn) InsertBatchStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

InsertBatchStream inserts a stream into a table in batch

func (*IcebergConn) InsertStream added in v1.4.10 

func (conn *IcebergConn) InsertStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

InsertStream inserts data into a table

func (*IcebergConn) NewTransaction added in v1.4.10 

func (conn *IcebergConn) NewTransaction(ctx context.Context, options ...*sql.TxOptions) (tx Transaction, err error)

NewTransaction creates a new transaction

func (*IcebergConn) StreamRowsContext added in v1.4.10 

func (conn *IcebergConn) StreamRowsContext(ctx context.Context, sql string, options ...map[string]interface{}) (ds *iop.Datastream, err error)

StreamRowsContext streams the rows of a table or query

func (*IcebergConn) SwapTable added in v1.4.10 

func (conn *IcebergConn) SwapTable(srcTable string, tgtTable string) (err error)

SwapTable swaps two tables by renaming them 2025-06-09 => doesn't work, blank error

func (*IcebergConn) TableExists added in v1.4.10 

func (conn *IcebergConn) TableExists(t Table) (exists bool, err error)

type IsolationLevel added in v1.4.11 

type IsolationLevel string
const (
	IsolationLevelDefault         IsolationLevel = "default"
	IsolationLevelReadUncommitted IsolationLevel = "read_uncommitted"
	IsolationLevelReadCommitted   IsolationLevel = "read_committed"
	IsolationLevelWriteCommitted  IsolationLevel = "write_committed"
	IsolationLevelRepeatableRead  IsolationLevel = "repeatable_read"
	IsolationLevelSnapshot        IsolationLevel = "snapshot"
	IsolationLevelSerializable    IsolationLevel = "serializable"
	IsolationLevelLinearizable    IsolationLevel = "linearizable"
)

func (IsolationLevel) AsSqlIsolationLevel added in v1.4.11 

func (il IsolationLevel) AsSqlIsolationLevel() sql.IsolationLevel

type ManualTransaction 

type ManualTransaction struct {
	Conn Connection
	// contains filtered or unexported fields
}

func (*ManualTransaction) Commit 

func (t *ManualTransaction) Commit() (err error)

func (*ManualTransaction) Context 

func (t *ManualTransaction) Context() *g.Context

func (*ManualTransaction) ExecContext 

func (t *ManualTransaction) ExecContext(ctx context.Context, q string, args ...interface{}) (result sql.Result, err error)

func (*ManualTransaction) ExecMultiContext 

func (t *ManualTransaction) ExecMultiContext(ctx context.Context, q string, args ...interface{}) (result sql.Result, err error)

func (*ManualTransaction) Prepare 

func (t *ManualTransaction) Prepare(query string) (stmt *sql.Stmt, err error)

func (*ManualTransaction) QueryContext 

func (t *ManualTransaction) QueryContext(ctx context.Context, q string, args ...interface{}) (result *sqlx.Rows, err error)

func (*ManualTransaction) Rollback 

func (t *ManualTransaction) Rollback() (err error)

type MergeConfig added in v1.4.20 

type MergeConfig struct {
	Strategy MergeStrategy
	Template string
	Map      map[string]string
}

type MergeStrategy added in v1.4.20 

type MergeStrategy string

MergeStrategy is for incremental loading 

const (
	MergeStrategyNone          MergeStrategy = ""
	MergeStrategyInsert        MergeStrategy = "insert"
	MergeStrategyUpdate        MergeStrategy = "update"
	MergeStrategyUpdateInsert  MergeStrategy = "update_insert"
	MergeStrategyDeleteInsert  MergeStrategy = "delete_insert"
	MergeStrategyHistoryInsert MergeStrategy = "history_insert"
)

type ModelDbX 

type ModelDbX struct {
	Ptr          interface{} `json:"-"`
	RowsAffected int         `json:"-"`
	// contains filtered or unexported fields
}

ModelDbX is the base for any SQL model

func (*ModelDbX) Bind 

func (m *ModelDbX) Bind(bindFunc func(p interface{}) error, objPtr interface{}) (err error)

Bind extracts values from provided echo context

func (*ModelDbX) Delete 

func (m *ModelDbX) Delete(db *sqlx.DB) (err error)

Delete deletes a record

func (*ModelDbX) Fields 

func (m *ModelDbX) Fields() (fields []string)

Fields returns the model fields

func (*ModelDbX) Get 

func (m *ModelDbX) Get(db *sqlx.DB, fields ...string) (err error)

Get get the first record

func (*ModelDbX) Insert 

func (m *ModelDbX) Insert(db *sqlx.DB, fields ...string) (err error)

Insert inserts one records

func (*ModelDbX) Rec 

func (m *ModelDbX) Rec() map[string]interface{}

Rec returns the record

func (*ModelDbX) Select 

func (m *ModelDbX) Select(db *sqlx.DB, objPtr interface{}, fields ...string) (err error)

Select returns multiple records

func (*ModelDbX) TableName 

func (m *ModelDbX) TableName(objPtr interface{}) string

TableName returns the table name of the underlying pointer

func (*ModelDbX) Update 

func (m *ModelDbX) Update(db *sqlx.DB, fields ...string) (err error)

Insert inserts one records

func (*ModelDbX) Values 

func (m *ModelDbX) Values(fields []string) (values []interface{}, err error)

Where adds a where clause

func (*ModelDbX) Where 

func (m *ModelDbX) Where(where ...interface{}) *ModelDbX

Where adds a where clause

type MongoDBConn added in v1.1.14 

type MongoDBConn struct {
	BaseConn
	URL    string
	Client *mongo.Client
}

MongoDBConn is a Mongo connection

func (*MongoDBConn) BulkExportFlow added in v1.1.14 

func (conn *MongoDBConn) BulkExportFlow(table Table) (df *iop.Dataflow, err error)

func (*MongoDBConn) Close added in v1.1.14 

func (conn *MongoDBConn) Close() error

func (*MongoDBConn) Connect added in v1.1.14 

func (conn *MongoDBConn) Connect(timeOut ...int) error

Connect connects to the database

func (*MongoDBConn) ExecContext added in v1.1.14 

func (conn *MongoDBConn) ExecContext(ctx context.Context, sql string, args ...any) (result sql.Result, err error)

func (*MongoDBConn) GetSchemas added in v1.1.14 

func (conn *MongoDBConn) GetSchemas() (data iop.Dataset, err error)

GetSchemas returns schemas

func (*MongoDBConn) GetSchemata added in v1.1.15 

func (conn *MongoDBConn) GetSchemata(level SchemataLevel, schemaName string, tableNames ...string) (Schemata, error)

GetSchemata obtain full schemata info for a schema and/or table in current database

func (*MongoDBConn) GetTableColumns added in v1.1.14 

func (conn *MongoDBConn) GetTableColumns(table *Table, fields ...string) (columns iop.Columns, err error)

NewTransaction creates a new transaction

func (*MongoDBConn) GetTables added in v1.1.14 

func (conn *MongoDBConn) GetTables(schema string) (data iop.Dataset, err error)

GetSchemas returns schemas

func (*MongoDBConn) Init added in v1.1.14 

func (conn *MongoDBConn) Init() error

Init initiates the object

func (*MongoDBConn) NewTransaction added in v1.1.14 

func (conn *MongoDBConn) NewTransaction(ctx context.Context, options ...*sql.TxOptions) (tx Transaction, err error)

NewTransaction creates a new transaction

func (*MongoDBConn) StreamRowsContext added in v1.1.14 

func (conn *MongoDBConn) StreamRowsContext(ctx context.Context, collectionName string, Opts ...map[string]any) (ds *iop.Datastream, err error)

type MsFabricConn added in v1.4.24 

type MsFabricConn struct {
	MsSQLServerConn
	URL string
	// contains filtered or unexported fields
}

MsFabricConn is a Microsoft Fabric connection

func (*MsFabricConn) AddMissingColumns added in v1.4.24 

func (conn *MsFabricConn) AddMissingColumns(table Table, newCols iop.Columns) (ok bool, err error)

func (*MsFabricConn) BulkExportFlow added in v1.4.24 

func (conn *MsFabricConn) BulkExportFlow(table Table) (*iop.Dataflow, error)

BulkExportFlow exports data using cursor (OneLake export not yet supported)

func (*MsFabricConn) BulkImportFlow added in v1.4.24 

func (conn *MsFabricConn) BulkImportFlow(tableFName string, df *iop.Dataflow) (count uint64, err error)

BulkImportFlow bulk imports using OneLake staging

func (*MsFabricConn) BulkImportStream added in v1.4.24 

func (conn *MsFabricConn) BulkImportStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

BulkImportStream bulk imports a stream using OneLake staging

func (*MsFabricConn) CastColumnForSelect added in v1.4.24 

func (conn *MsFabricConn) CastColumnForSelect(srcCol iop.Column, tgtCol iop.Column) (selectStr string)

CastColumnForSelect casts to the correct target column type

func (*MsFabricConn) Connect added in v1.4.24 

func (conn *MsFabricConn) Connect(timeOut ...int) (err error)

func (*MsFabricConn) CopyFromOneLake added in v1.4.24 

func (conn *MsFabricConn) CopyFromOneLake(tableFName, oneLakePath string, columns iop.Columns, fileFormat dbio.FileType) (err error)

CopyFromOneLake uses the COPY INTO command to load data from OneLake

func (*MsFabricConn) GenerateDDL added in v1.4.24 

func (conn *MsFabricConn) GenerateDDL(table Table, data iop.Dataset, temporary bool) (string, error)

func (*MsFabricConn) Init added in v1.4.24 

func (conn *MsFabricConn) Init() error

Init initiates the object

type MsSQLServerConn 

type MsSQLServerConn struct {
	BaseConn
	URL string
	// contains filtered or unexported fields
}

MsSQLServerConn is a Microsoft SQL Server connection

func (*MsSQLServerConn) BcpExport 

func (conn *MsSQLServerConn) BcpExport() (err error)

BcpExport exports data to datastream

func (*MsSQLServerConn) BcpImportFile 

func (conn *MsSQLServerConn) BcpImportFile(tableFName, filePath string) (count uint64, err error)

BcpImportFile Import using bcp tool https://docs.microsoft.com/en-us/sql/tools/bcp-utility?view=sql-server-ver15 bcp dbo.test1 in '/tmp/LargeDataset.csv' -S tcp:sqlserver.host,51433 -d master -U sa -P 'password' -c -t ',' -b 5000 Limitation: if comma or delimite is in field, it will error. need to use delimiter not in field, or do some other transformation

func (*MsSQLServerConn) BcpImportFileParrallel 

func (conn *MsSQLServerConn) BcpImportFileParrallel(tableFName string, ds *iop.Datastream) (count uint64, err error)

BcpImportFileParrallel uses goroutine to import partitioned files

func (*MsSQLServerConn) BulkImportFlow 

func (conn *MsSQLServerConn) BulkImportFlow(tableFName string, df *iop.Dataflow) (count uint64, err error)

BulkImportFlow bulk import flow

func (*MsSQLServerConn) BulkImportStream 

func (conn *MsSQLServerConn) BulkImportStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

BulkImportStream bulk import stream

func (*MsSQLServerConn) CastColumnForSelect added in v1.3.5 

func (conn *MsSQLServerConn) CastColumnForSelect(srcCol iop.Column, tgtCol iop.Column) (selectStr string)

CastColumnForSelect casts to the correct target column type

func (*MsSQLServerConn) Close added in v1.4.25 

func (conn *MsSQLServerConn) Close() error

Close closes the SQL Server connection and cleans up Cloud SQL resources if applicable

func (*MsSQLServerConn) ConnString added in v1.2.3 

func (conn *MsSQLServerConn) ConnString() string

func (*MsSQLServerConn) Connect added in v1.2.3 

func (conn *MsSQLServerConn) Connect(timeOut ...int) (err error)

func (*MsSQLServerConn) CopyFromAzure 

func (conn *MsSQLServerConn) CopyFromAzure(tableFName, azPath string) (count uint64, err error)

CopyFromAzure uses the COPY INTO Table command from Azure https://docs.microsoft.com/en-us/sql/t-sql/statements/copy-into-transact-sql?view=azure-sqldw-latest

func (*MsSQLServerConn) CopyViaAzure 

func (conn *MsSQLServerConn) CopyViaAzure(tableFName string, df *iop.Dataflow) (count uint64, err error)

CopyViaAzure uses the Azure DWH COPY INTO Table command

func (*MsSQLServerConn) FedAuth added in v1.4.24 

func (conn *MsSQLServerConn) FedAuth() string

func (*MsSQLServerConn) GenerateDDL added in v1.2.15 

func (conn *MsSQLServerConn) GenerateDDL(table Table, data iop.Dataset, temporary bool) (string, error)

func (*MsSQLServerConn) GenerateMergeSQL added in v1.4.20 

func (conn *MsSQLServerConn) GenerateMergeSQL(srcTable string, tgtTable string, pkFields []string) (sql string, err error)

GenerateMergeSQL generates the upsert SQL

func (*MsSQLServerConn) GetTableColumns added in v1.3.5 

func (conn *MsSQLServerConn) GetTableColumns(table *Table, fields ...string) (columns iop.Columns, err error)

func (*MsSQLServerConn) GetURL 

func (conn *MsSQLServerConn) GetURL(newURL ...string) string

GetURL returns the processed URL

func (*MsSQLServerConn) Init 

func (conn *MsSQLServerConn) Init() error

Init initiates the object

func (*MsSQLServerConn) SubmitTemplate added in v1.3.5 

func (conn *MsSQLServerConn) SubmitTemplate(level string, templateMap map[string]string, name string, values map[string]any) (data iop.Dataset, err error)

type MySQLConn 

type MySQLConn struct {
	BaseConn
	URL string
	// contains filtered or unexported fields
}

MySQLConn is a MySQL or MariaDB connection

func (*MySQLConn) BulkExportStream 

func (conn *MySQLConn) BulkExportStream(table Table) (ds *iop.Datastream, err error)

BulkExportStream bulk Export

func (*MySQLConn) BulkImportStream 

func (conn *MySQLConn) BulkImportStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

BulkImportStream bulk import stream

func (*MySQLConn) Close added in v1.4.25 

func (conn *MySQLConn) Close() error

Close closes the MySQL connection and cleans up Cloud SQL resources if applicable

func (*MySQLConn) Connect added in v1.3.4 

func (conn *MySQLConn) Connect(timeOut ...int) (err error)

func (*MySQLConn) GenerateDDL added in v1.2.15 

func (conn *MySQLConn) GenerateDDL(table Table, data iop.Dataset, temporary bool) (string, error)

func (*MySQLConn) GenerateMergeSQL added in v1.4.20 

func (conn *MySQLConn) GenerateMergeSQL(srcTable string, tgtTable string, pkFields []string) (sql string, err error)

UPSERT https://vladmihalcea.com/how-do-upsert-and-merge-work-in-oracle-sql-server-postgresql-and-mysql/ GenerateMergeSQL generates the upsert SQL

func (*MySQLConn) GetURL 

func (conn *MySQLConn) GetURL(newURL ...string) string

GetURL returns the processed URL

func (*MySQLConn) Init 

func (conn *MySQLConn) Init() error

Init initiates the object

func (*MySQLConn) LoadDataInFile 

func (conn *MySQLConn) LoadDataInFile(tableFName string, ds *iop.Datastream) (count uint64, err error)

LoadDataInFile Bulk Import

func (*MySQLConn) LoadDataOutFile 

func (conn *MySQLConn) LoadDataOutFile(ctx *g.Context, sql string) (stdOutReader io.Reader, err error)

LoadDataOutFile Bulk Export Possible error: ERROR 1227 (42000) at line 1: Access denied; you need (at least one of) the FILE privilege(s) for this operation File privilege needs to be granted to user also the --secure-file-priv option needs to be set properly for it to work. https://stackoverflow.com/questions/9819271/why-is-mysql-innodb-insert-so-slow to improve innodb insert speed

type Operation added in v1.4.24 

type Operation string
const (
	OperationMerge        Operation = "merge"
	OperationDropTable    Operation = "drop_table"
	OperationCreateIndex  Operation = "create_index"
	OperationGenerateData Operation = "generate_data"
)

type OracleConn 

type OracleConn struct {
	BaseConn
	URL string
	// contains filtered or unexported fields
}

OracleConn is a Postgres connection

func (*OracleConn) BulkImportStream 

func (conn *OracleConn) BulkImportStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

BulkImportStream bulk import stream

func (*OracleConn) CastColumnForSelect added in v1.3.5 

func (conn *OracleConn) CastColumnForSelect(srcCol iop.Column, tgtCol iop.Column) (selectStr string)

CastColumnForSelect casts to the correct target column type

func (*OracleConn) ConnString added in v1.1.14 

func (conn *OracleConn) ConnString() string

func (*OracleConn) Connect added in v1.1.14 

func (conn *OracleConn) Connect(timeOut ...int) (err error)

func (*OracleConn) ExecMultiContext 

func (conn *OracleConn) ExecMultiContext(ctx context.Context, qs ...string) (result sql.Result, err error)

ExecMultiContext runs multiple sql queries with context, returns `error`

func (*OracleConn) GenerateDDL added in v1.2.15 

func (conn *OracleConn) GenerateDDL(table Table, data iop.Dataset, temporary bool) (string, error)

func (*OracleConn) GenerateInsertStatement 

func (conn *OracleConn) GenerateInsertStatement(tableName string, cols iop.Columns, numRows int) string

GenerateInsertStatement returns the proper INSERT statement

func (*OracleConn) GenerateMergeSQL added in v1.4.20 

func (conn *OracleConn) GenerateMergeSQL(srcTable string, tgtTable string, pkFields []string) (sql string, err error)

GenerateMergeSQL generates the upsert SQL

func (*OracleConn) GetTableColumns added in v1.1.8 

func (conn *OracleConn) GetTableColumns(table *Table, fields ...string) (columns iop.Columns, err error)

func (*OracleConn) Init 

func (conn *OracleConn) Init() error

Init initiates the object

func (*OracleConn) SQLLoad 

func (conn *OracleConn) SQLLoad(tableFName string, ds *iop.Datastream) (count uint64, err error)

SQLLoad uses sqlldr to Bulk Import cat test1.csv | sqlldr system/oracle@oracle.host:1521/xe control=sqlldr.ctl log=/dev/stdout bad=/dev/stderr cannot import when newline in value. Need to scan for new lines.

func (*OracleConn) SubmitTemplate added in v1.2.4 

func (conn *OracleConn) SubmitTemplate(level string, templateMap map[string]string, name string, values map[string]interface{}) (data iop.Dataset, err error)

func (*OracleConn) Version added in v1.1.14 

func (conn *OracleConn) Version() int

type Pool 

type Pool struct {
	Dbs     map[string]*sqlx.DB
	DuckDbs map[string]*DuckDbConn
	Mux     sync.Mutex
}

Pool is a pool of connections

type PostgresConn 

type PostgresConn struct {
	BaseConn
	URL string
	// contains filtered or unexported fields
}

PostgresConn is a Postgres connection

func (*PostgresConn) BulkExportStream 

func (conn *PostgresConn) BulkExportStream(table Table) (ds *iop.Datastream, err error)

BulkExportStream uses the bulk dumping (COPY)

func (*PostgresConn) BulkImportStream 

func (conn *PostgresConn) BulkImportStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

BulkImportStream inserts a stream into a table

func (*PostgresConn) CastColumnForSelect 

func (conn *PostgresConn) CastColumnForSelect(srcCol iop.Column, tgtCol iop.Column) (selectStr string)

CastColumnForSelect casts to the correct target column type

func (*PostgresConn) Close added in v1.4.25 

func (conn *PostgresConn) Close() error

Close closes the database connection and cleans up Cloud SQL resources

func (*PostgresConn) Connect added in v1.4.10 

func (conn *PostgresConn) Connect(timeOut ...int) error

Connect connects to the database

func (*PostgresConn) CopyToStdout 

func (conn *PostgresConn) CopyToStdout(ctx *g.Context, sql string) (stdOutReader io.Reader, err error)

CopyToStdout Copy TO STDOUT

func (*PostgresConn) GenerateDDL added in v1.1.4 

func (conn *PostgresConn) GenerateDDL(table Table, data iop.Dataset, temporary bool) (ddl string, err error)

GenerateDDL generates a DDL based on a dataset

func (*PostgresConn) GenerateMergeSQL added in v1.4.20 

func (conn *PostgresConn) GenerateMergeSQL(srcTable string, tgtTable string, pkFields []string) (sql string, err error)

GenerateMergeSQL generates the upsert SQL

func (*PostgresConn) Init 

func (conn *PostgresConn) Init() error

Init initiates the object

type PrometheusConn added in v1.2.2 

type PrometheusConn struct {
	BaseConn
	URL    string
	Client v1.API
}

PrometheusConn is a Prometheus connection

func (*PrometheusConn) BulkExportFlow added in v1.2.2 

func (conn *PrometheusConn) BulkExportFlow(table Table) (df *iop.Dataflow, err error)

func (*PrometheusConn) Close added in v1.2.2 

func (conn *PrometheusConn) Close() error

func (*PrometheusConn) Connect added in v1.2.2 

func (conn *PrometheusConn) Connect(timeOut ...int) error

Connect connects to the database

func (*PrometheusConn) ExecContext added in v1.2.2 

func (conn *PrometheusConn) ExecContext(ctx context.Context, sql string, args ...interface{}) (result sql.Result, err error)

func (*PrometheusConn) GetSQLColumns added in v1.2.2 

func (conn *PrometheusConn) GetSQLColumns(table Table) (columns iop.Columns, err error)

func (*PrometheusConn) GetSchemas added in v1.2.2 

func (conn *PrometheusConn) GetSchemas() (data iop.Dataset, err error)

GetSchemas returns schemas

func (*PrometheusConn) GetSchemata added in v1.2.2 

func (conn *PrometheusConn) GetSchemata(level SchemataLevel, schemaName string, tableNames ...string) (Schemata, error)

GetSchemata obtain full schemata info for a schema and/or table in current database

func (*PrometheusConn) GetTableColumns added in v1.2.2 

func (conn *PrometheusConn) GetTableColumns(table *Table, fields ...string) (columns iop.Columns, err error)

NewTransaction creates a new transaction

func (*PrometheusConn) GetTables added in v1.2.2 

func (conn *PrometheusConn) GetTables(schema string) (data iop.Dataset, err error)

GetSchemas returns schemas

func (*PrometheusConn) Init added in v1.2.2 

func (conn *PrometheusConn) Init() error

Init initiates the object

func (*PrometheusConn) NewTransaction added in v1.2.2 

func (conn *PrometheusConn) NewTransaction(ctx context.Context, options ...*sql.TxOptions) (tx Transaction, err error)

NewTransaction creates a new transaction

func (*PrometheusConn) StreamRowsChunked added in v1.4.11 

func (conn *PrometheusConn) StreamRowsChunked(queryContext *g.Context, query string, opts map[string]interface{}) (ds *iop.Datastream, err error)

StreamRowsChunked implements chunked streaming for Prometheus to avoid memory issues

func (*PrometheusConn) StreamRowsContext added in v1.2.2 

func (conn *PrometheusConn) StreamRowsContext(ctx context.Context, query string, Opts ...map[string]interface{}) (ds *iop.Datastream, err error)

type ProtonConn added in v1.2.7 

type ProtonConn struct {
	BaseConn
	URL string
}

ProtonConn is a Proton connection

func (*ProtonConn) BulkImportStream added in v1.2.7 

func (conn *ProtonConn) BulkImportStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

BulkImportStream inserts a stream into a table

func (*ProtonConn) ConnString added in v1.2.7 

func (conn *ProtonConn) ConnString() string

func (*ProtonConn) Connect added in v1.2.7 

func (conn *ProtonConn) Connect(timeOut ...int) (err error)

func (*ProtonConn) ExecContext added in v1.2.21 

func (conn *ProtonConn) ExecContext(ctx context.Context, q string, args ...interface{}) (result sql.Result, err error)

ExecContext runs a sql query with context, returns `error`

func (*ProtonConn) GenerateDDL added in v1.2.7 

func (conn *ProtonConn) GenerateDDL(table Table, data iop.Dataset, temporary bool) (sql string, err error)

GenerateDDL generates a DDL based on a dataset

func (*ProtonConn) GenerateInsertStatement added in v1.2.7 

func (conn *ProtonConn) GenerateInsertStatement(tableName string, cols iop.Columns, numRows int) string

GenerateInsertStatement returns the proper INSERT statement

func (*ProtonConn) GenerateMergeSQL added in v1.4.20 

func (conn *ProtonConn) GenerateMergeSQL(srcTable string, tgtTable string, pkFields []string) (sql string, err error)

GenerateMergeSQL generates the upsert SQL

func (*ProtonConn) GetCount added in v1.2.14 

func (conn *ProtonConn) GetCount(tableFName string) (int64, error)

GetCount returns count of records

func (*ProtonConn) GetNativeType added in v1.2.21 

func (conn *ProtonConn) GetNativeType(col iop.Column) (nativeType string, err error)

func (*ProtonConn) Init added in v1.2.7 

func (conn *ProtonConn) Init() error

Init initiates the object

func (*ProtonConn) NewTransaction added in v1.2.7 

func (conn *ProtonConn) NewTransaction(ctx context.Context, options ...*sql.TxOptions) (Transaction, error)

NewTransaction creates a new transaction

type RedshiftConn 

type RedshiftConn struct {
	BaseConn
	URL string
}

RedshiftConn is a Redshift connection

func (*RedshiftConn) BulkExportFlow 

func (conn *RedshiftConn) BulkExportFlow(table Table) (df *iop.Dataflow, err error)

BulkExportFlow reads in bulk

func (*RedshiftConn) BulkExportStream 

func (conn *RedshiftConn) BulkExportStream(table Table) (ds *iop.Datastream, err error)

BulkExportStream reads in bulk

func (*RedshiftConn) BulkImportFlow 

func (conn *RedshiftConn) BulkImportFlow(tableFName string, df *iop.Dataflow) (count uint64, err error)

BulkImportFlow inserts a flow of streams into a table. For redshift we need to create CSVs in S3 and then use the COPY command.

func (*RedshiftConn) BulkImportStream 

func (conn *RedshiftConn) BulkImportStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

BulkImportStream inserts a stream into a table. For redshift we need to create CSVs in S3 and then use the COPY command.

func (*RedshiftConn) CastColumnForSelect added in v1.2.10 

func (conn *RedshiftConn) CastColumnForSelect(srcCol iop.Column, tgtCol iop.Column) (selectStr string)

CastColumnForSelect casts to the correct target column type

func (*RedshiftConn) ConnString 

func (conn *RedshiftConn) ConnString() string

func (*RedshiftConn) CopyFromS3 

func (conn *RedshiftConn) CopyFromS3(tableFName, s3Path string, columns iop.Columns) (count uint64, err error)

CopyFromS3 uses the COPY INTO Table command from AWS S3

func (*RedshiftConn) GenerateDDL added in v1.1.4 

func (conn *RedshiftConn) GenerateDDL(table Table, data iop.Dataset, temporary bool) (sql string, err error)

GenerateDDL generates a DDL based on a dataset

func (*RedshiftConn) GenerateMergeSQL added in v1.4.20 

func (conn *RedshiftConn) GenerateMergeSQL(srcTable string, tgtTable string, pkFields []string) (sql string, err error)

GenerateMergeSQL generates the upsert SQL

func (*RedshiftConn) Init 

func (conn *RedshiftConn) Init() error

Init initiates the object

func (*RedshiftConn) OptimizeTable added in v1.1.9 

func (conn *RedshiftConn) OptimizeTable(table *Table, newColumns iop.Columns, isTemp ...bool) (ok bool, err error)

func (*RedshiftConn) Unload 

func (conn *RedshiftConn) Unload(ctx *g.Context, fileFormat dbio.FileType, tables ...Table) (s3Path string, err error)

Unload unloads a query to S3

func (*RedshiftConn) WarnStlLoadErrors added in v1.1.9 

func (conn *RedshiftConn) WarnStlLoadErrors(err error)

type Relation 

type Relation string

type Result 

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

func (Result) LastInsertId 

func (r Result) LastInsertId() (int64, error)

func (Result) RowsAffected 

func (r Result) RowsAffected() (int64, error)

type SQLiteConn 

type SQLiteConn struct {
	BaseConn
	URL string
}

SQLiteConn is a SQLite connection

func (*SQLiteConn) BulkImportStream 

func (conn *SQLiteConn) BulkImportStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

BulkImportStream inserts a stream into a table

func (*SQLiteConn) GenerateDDL added in v1.2.15 

func (conn *SQLiteConn) GenerateDDL(table Table, data iop.Dataset, temporary bool) (string, error)

func (*SQLiteConn) GenerateMergeSQL added in v1.4.20 

func (conn *SQLiteConn) GenerateMergeSQL(srcTable string, tgtTable string, pkFields []string) (sql string, err error)

GenerateMergeSQL generates the upsert SQL

func (*SQLiteConn) GetSchemata 

func (conn *SQLiteConn) GetSchemata(level SchemataLevel, schemaName string, tableNames ...string) (Schemata, error)

GetSchemata obtain full schemata info for a schema and/or table in current database

func (*SQLiteConn) GetURL 

func (conn *SQLiteConn) GetURL(newURL ...string) string

GetURL returns the processed URL

func (*SQLiteConn) Init 

func (conn *SQLiteConn) Init() error

Init initiates the object

type Schema 

type Schema struct {
	Name     string           `json:"name"`
	Database string           `json:"database"`
	Tables   map[string]Table `json:"tables"`
}

Schema represents a schemata schema

func (*Schema) Columns 

func (schema *Schema) Columns() map[string]iop.Column

func (*Schema) ToData 

func (schema *Schema) ToData() (data iop.Dataset)

ToData converts schema objects to tabular format

type Schemata 

type Schemata struct {
	Databases map[string]Database `json:"databases"`
	// contains filtered or unexported fields
}

Schemata contains the full schema for a connection

func GetSchemataAll 

func GetSchemataAll(conn Connection) (schemata Schemata, err error)

GetSchemataAll obtains the schemata for all databases detected

func GetTablesSchemata 

func GetTablesSchemata(conn Connection, tableNames ...string) (schemata Schemata, err error)

GetTablesSchemata obtains the schemata for specified tables

func (*Schemata) Columns 

func (s *Schemata) Columns(filters ...string) map[string]iop.Column

func (*Schemata) Database 

func (s *Schemata) Database() Database

Database returns the first encountered database

func (*Schemata) Filtered added in v1.1.15 

func (s *Schemata) Filtered(columnLevel bool, filters ...string) (ns Schemata)

func (*Schemata) LoadTablesJSON 

func (s *Schemata) LoadTablesJSON(payload string) error

LoadTablesJSON loads from a json string

func (*Schemata) Tables 

func (s *Schemata) Tables(filters ...string) map[string]Table

type SchemataLevel added in v1.2.19 

type SchemataLevel string
const (
	SchemataLevelSchema SchemataLevel = "schema"
	SchemataLevelTable  SchemataLevel = "table"
	SchemataLevelColumn SchemataLevel = "column"
)

type SelectOptions added in v1.3.5 

type SelectOptions struct {
	Fields []string
	Offset int
	Limit  *int
	Where  string
}

type SnowflakeConn 

type SnowflakeConn struct {
	BaseConn
	URL        string
	Warehouse  string
	CopyMethod string
	Mux        sync.Mutex
}

SnowflakeConn is a Snowflake connection

func (*SnowflakeConn) BulkExportFlow 

func (conn *SnowflakeConn) BulkExportFlow(table Table) (df *iop.Dataflow, err error)

BulkExportFlow reads in bulk

func (*SnowflakeConn) BulkImportFlow 

func (conn *SnowflakeConn) BulkImportFlow(tableFName string, df *iop.Dataflow) (count uint64, err error)

BulkImportFlow bulk import flow

func (*SnowflakeConn) BulkImportStream 

func (conn *SnowflakeConn) BulkImportStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

BulkImportStream bulk import stream

func (*SnowflakeConn) CastColumnForSelect added in v1.1.15 

func (conn *SnowflakeConn) CastColumnForSelect(srcCol iop.Column, tgtCol iop.Column) (selectStr string)

CastColumnForSelect casts to the correct target column type

func (*SnowflakeConn) ConnString 

func (conn *SnowflakeConn) ConnString() string

func (*SnowflakeConn) Connect 

func (conn *SnowflakeConn) Connect(timeOut ...int) error

Connect connects to the database

func (*SnowflakeConn) CopyFromAzure 

func (conn *SnowflakeConn) CopyFromAzure(tableFName, azPath string) (err error)

CopyFromAzure uses the Snowflake COPY INTO Table command from Azure https://docs.snowflake.com/en/sql-reference/sql/copy-into-table.html

func (*SnowflakeConn) CopyFromS3 

func (conn *SnowflakeConn) CopyFromS3(tableFName, s3Path string) (err error)

CopyFromS3 uses the Snowflake COPY INTO Table command from AWS S3 https://docs.snowflake.com/en/sql-reference/sql/copy-into-table.html

func (*SnowflakeConn) CopyToAzure 

func (conn *SnowflakeConn) CopyToAzure(tables ...Table) (azPath string, err error)

CopyToAzure exports a query to an Azure location

func (*SnowflakeConn) CopyToS3 

func (conn *SnowflakeConn) CopyToS3(tables ...Table) (s3Path string, err error)

CopyToS3 exports a query to an S3 location

func (*SnowflakeConn) CopyViaAWS 

func (conn *SnowflakeConn) CopyViaAWS(tableFName string, df *iop.Dataflow) (count uint64, err error)

CopyViaAWS uses the Snowflake COPY INTO Table command from AWS S3 https://docs.snowflake.com/en/sql-reference/sql/copy-into-table.html

func (*SnowflakeConn) CopyViaAzure 

func (conn *SnowflakeConn) CopyViaAzure(tableFName string, df *iop.Dataflow) (count uint64, err error)

CopyViaAzure uses the Snowflake COPY INTO Table command from Azure https://docs.snowflake.com/en/sql-reference/sql/copy-into-table.html

func (*SnowflakeConn) CopyViaStage 

func (conn *SnowflakeConn) CopyViaStage(table Table, df *iop.Dataflow) (count uint64, err error)

CopyViaStage uses the Snowflake COPY INTO Table command https://docs.snowflake.com/en/sql-reference/sql/copy-into-table.html

func (*SnowflakeConn) GenerateDDL added in v1.1.4 

func (conn *SnowflakeConn) GenerateDDL(table Table, data iop.Dataset, temporary bool) (sql string, err error)

GenerateDDL generates a DDL based on a dataset

func (*SnowflakeConn) GenerateInsertStatement added in v1.2.19 

func (conn *SnowflakeConn) GenerateInsertStatement(tableName string, cols iop.Columns, numRows int) string

func (*SnowflakeConn) GenerateMergeSQL added in v1.4.20 

func (conn *SnowflakeConn) GenerateMergeSQL(srcTable string, tgtTable string, pkFields []string) (sql string, err error)

GenerateMergeSQL generates the upsert SQL

func (*SnowflakeConn) GetColumnsFull 

func (conn *SnowflakeConn) GetColumnsFull(tableFName string) (data iop.Dataset, err error)

GetColumnsFull returns columns for given table. `tableName` should include schema and table, example: `schema1.table2` fields should be `schema_name|table_name|table_type|column_name|data_type|column_id`

func (*SnowflakeConn) GetDatabases 

func (conn *SnowflakeConn) GetDatabases() (data iop.Dataset, err error)

GetDatabases returns the list of databases

func (*SnowflakeConn) GetSchemas 

func (conn *SnowflakeConn) GetSchemas() (data iop.Dataset, err error)

GetSchemas returns schemas

func (*SnowflakeConn) GetTables 

func (conn *SnowflakeConn) GetTables(schema string) (data iop.Dataset, err error)

GetTables returns tables

func (*SnowflakeConn) GetTablesAndViews added in v1.2.19 

func (conn *SnowflakeConn) GetTablesAndViews(schema string) (iop.Dataset, error)

GetTablesAndViews returns tables/views for given schema

func (*SnowflakeConn) GetViews 

func (conn *SnowflakeConn) GetViews(schema string) (data iop.Dataset, err error)

GetTables returns tables

func (*SnowflakeConn) Init 

func (conn *SnowflakeConn) Init() error

Init initiates the object

func (*SnowflakeConn) StageGET added in v1.2.14 

func (conn *SnowflakeConn) StageGET(internalStagePath, folderPath string) (filePaths []string, err error)

StageGET Copies from a staging location to a local file or folder

func (*SnowflakeConn) StagePUT added in v1.2.14 

func (conn *SnowflakeConn) StagePUT(fileURI string, internalStagePath string) (err error)

StagePUT Copies a local file or folder into a staging location

func (*SnowflakeConn) UnloadViaStage 

func (conn *SnowflakeConn) UnloadViaStage(format dbio.FileType, tables ...Table) (filePath string, unloaded int64, err error)

type SqlServerLogger added in v1.3.6 

type SqlServerLogger struct{}

func (*SqlServerLogger) Printf added in v1.3.6 

func (l *SqlServerLogger) Printf(format string, v ...any)

func (*SqlServerLogger) Println added in v1.3.6 

func (l *SqlServerLogger) Println(v ...any)

type StarRocksConn 

type StarRocksConn struct {
	BaseConn
	URL string
	// contains filtered or unexported fields
}

StarRocksConn is a StarRocks connection

func (*StarRocksConn) AddMissingColumns added in v1.1.8 

func (conn *StarRocksConn) AddMissingColumns(table Table, newCols iop.Columns) (ok bool, err error)

func (*StarRocksConn) BulkImportFlow 

func (conn *StarRocksConn) BulkImportFlow(tableFName string, df *iop.Dataflow) (count uint64, err error)

BulkImportFlow inserts a flow of streams into a table.

func (*StarRocksConn) Connect added in v1.4.16 

func (conn *StarRocksConn) Connect(timeOut ...int) (err error)

func (*StarRocksConn) ExecContext added in v1.2.19 

func (conn *StarRocksConn) ExecContext(ctx context.Context, q string, args ...interface{}) (result sql.Result, err error)

ExecContext runs a sql query with context, returns `error`

func (*StarRocksConn) GenerateDDL 

func (conn *StarRocksConn) GenerateDDL(table Table, data iop.Dataset, temporary bool) (string, error)

GenerateDDL generates a DDL based on a dataset

func (*StarRocksConn) GetDatabases added in v1.1.9 

func (conn *StarRocksConn) GetDatabases() (data iop.Dataset, err error)

GetDatabases returns the list of databases

func (*StarRocksConn) GetURL 

func (conn *StarRocksConn) GetURL(newURL ...string) string

GetURL returns the processed URL

func (*StarRocksConn) Init 

func (conn *StarRocksConn) Init() error

Init initiates the object

func (*StarRocksConn) InsertBatchStream 

func (conn *StarRocksConn) InsertBatchStream(tableFName string, ds *iop.Datastream) (count uint64, err error)

InsertBatchStream inserts a stream into a table in batch

func (*StarRocksConn) NewTransaction added in v1.4.25 

func (conn *StarRocksConn) NewTransaction(ctx context.Context, options ...*sql.TxOptions) (tx Transaction, err error)

NewTransaction creates a new transaction

func (*StarRocksConn) OptimizeTable added in v1.1.8 

func (conn *StarRocksConn) OptimizeTable(table *Table, newColumns iop.Columns, isTemp ...bool) (ok bool, err error)

func (*StarRocksConn) StreamLoad 

func (conn *StarRocksConn) StreamLoad(feURL, tableFName string, df *iop.Dataflow) (count uint64, err error)

StreamLoad bulk loads https://docs.starrocks.io/docs/loading/StreamLoad/ https://docs.starrocks.io/docs/sql-reference/sql-statements/data-manipulation/STREAM_LOAD/

func (*StarRocksConn) WaitAlterTable added in v1.1.8 

func (conn *StarRocksConn) WaitAlterTable(table Table) (err error)

type StatFieldSQL 

type StatFieldSQL struct {
	Name        string
	TemplateSQL string
}

type Table 

type Table struct {
	Name     string      `json:"name"`
	Schema   string      `json:"schema"`
	Database string      `json:"database,omitempty"`
	IsView   bool        `json:"is_view,omitempty"` // whether is a view
	SQL      string      `json:"sql,omitempty"`
	DDL      string      `json:"ddl,omitempty"`
	Dialect  dbio.Type   `json:"dialect,omitempty"`
	Columns  iop.Columns `json:"columns,omitempty"`
	Keys     TableKeys   `json:"keys,omitempty"`

	Raw string `json:"raw"`
	// contains filtered or unexported fields
}

Table represents a schemata table

func ParseTableName 

func ParseTableName(text string, dialect dbio.Type) (table Table, err error)

func (*Table) AddPrimaryKeyToDDL added in v1.2.15 

func (t *Table) AddPrimaryKeyToDDL(ddl string, columns iop.Columns) (string, error)

AddPrimaryKeyToDDL adds a primary key to the table

func (*Table) Clone added in v1.2.14 

func (t *Table) Clone() Table

func (*Table) ColumnsMap 

func (t *Table) ColumnsMap() map[string]iop.Column

func (*Table) DatabaseQ 

func (t *Table) DatabaseQ() string

func (*Table) FDQN 

func (t *Table) FDQN() string

func (*Table) FullName 

func (t *Table) FullName() string

func (*Table) Indexes added in v1.2.15 

func (t *Table) Indexes(columns iop.Columns) (indexes []TableIndex)

func (*Table) IsProcedural added in v1.3.5 

func (t *Table) IsProcedural() bool

func (*Table) IsQuery 

func (t *Table) IsQuery() bool

func (*Table) MarshalJSON added in v1.2.19 

func (t *Table) MarshalJSON() ([]byte, error)

func (*Table) NameQ 

func (t *Table) NameQ() string

func (*Table) SchemaQ 

func (t *Table) SchemaQ() string

func (*Table) Select 

func (t *Table) Select(Opts ...SelectOptions) (sql string)

func (*Table) SetKeys added in v1.1.8 

func (t *Table) SetKeys(sourcePKCols []string, updateCol string, tableKeys TableKeys) error

type TableIndex added in v1.2.15 

type TableIndex struct {
	Name    string
	Columns iop.Columns
	Unique  bool
	Table   *Table
}

func (*TableIndex) CreateDDL added in v1.2.15 

func (ti *TableIndex) CreateDDL() string

func (*TableIndex) DropDDL added in v1.2.15 

func (ti *TableIndex) DropDDL() string

type TableKeys added in v1.1.4 

type TableKeys map[iop.KeyType][]string

type Transaction 

type Transaction interface {
	Connection() Connection
	Context() *g.Context
	Commit() (err error)
	Rollback() (err error)
	Prepare(query string) (stmt *sql.Stmt, err error)
	QueryContext(ctx context.Context, q string, args ...interface{}) (result *sqlx.Rows, err error)
	ExecContext(ctx context.Context, q string, args ...interface{}) (result sql.Result, err error)
	ExecMultiContext(ctx context.Context, q string, args ...interface{}) (result sql.Result, err error)
}

type TrinoConn added in v1.1.14 

type TrinoConn struct {
	BaseConn
	URL string
}

TrinoConn is a Trino connection

func (*TrinoConn) ConnString added in v1.1.14 

func (conn *TrinoConn) ConnString() string

func (*TrinoConn) ExecContext added in v1.1.14 

func (conn *TrinoConn) ExecContext(ctx context.Context, q string, args ...interface{}) (result sql.Result, err error)

ExecContext runs a sql query with context, returns `error`

func (*TrinoConn) Init added in v1.1.14 

func (conn *TrinoConn) Init() error

Init initiates the object

func (*TrinoConn) NewTransaction added in v1.1.14 

func (conn *TrinoConn) NewTransaction(ctx context.Context, options ...*sql.TxOptions) (tx Transaction, err error)

NewTransaction creates a new transaction

type User 

type User struct {
	ModelDbX
	Name string
	Age  int
}

func NewUser 

func NewUser() *User

type WhereClause 

type WhereClause []interface{}

WhereClause is the where clause

func (WhereClause) Args 

func (wc WhereClause) Args() []interface{}

Args returns the where clause arguments

func (WhereClause) Clause 

func (wc WhereClause) Clause() string

Clause returns the string where clause

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.