postgres

Documentation

Index

• func ColumnsWithDataTypes(columns model.TableSchema, prefix string) string
• func Connect(cred Credentials) (*sql.DB, error)
• func Init()
• func WithConfig(h *Postgres, config *config.Config)
• type Credentials
• type Diagnostic
  • func (d *Diagnostic) TxnExecute(ctx context.Context, txn *sql.Tx, tableName, query string) error
  • func (d *Diagnostic) TxnRollback(txn Tx, tableName, stage string)
• type LoadTable
  • func (lt *LoadTable) Load(ctx context.Context, tableName string, tableSchemaInUpload model.TableSchema) (string, error)
• type LoadUsersTable
  • func (lut *LoadUsersTable) Load(ctx context.Context, ...) map[string]error
• type Postgres
  • func NewPostgres() *Postgres
  • func (pg *Postgres) AddColumns(tableName string, columnsInfo []warehouseutils.ColumnInfo) (err error)
  • func (*Postgres) AlterColumn(_, _, _ string) (model.AlterTableResponse, error)
  • func (pg *Postgres) Cleanup()
  • func (pg *Postgres) Connect(warehouse model.Warehouse) (client.Client, error)
  • func (pg *Postgres) CrashRecover(model.Warehouse) error
  • func (pg *Postgres) CreateSchema() (err error)
  • func (pg *Postgres) CreateTable(tableName string, columnMap model.TableSchema) (err error)
  • func (pg *Postgres) DeleteBy(tableNames []string, params warehouseutils.DeleteByParams) (err error)
  • func (*Postgres) DownloadIdentityRules(*misc.GZipWriter) (err error)
  • func (pg *Postgres) DropTable(tableName string) (err error)
  • func (pg *Postgres) ErrorMappings() []model.JobError
  • func (pg *Postgres) FetchSchema(warehouse model.Warehouse) (schema, unrecognizedSchema model.Schema, err error)
  • func (pg *Postgres) GetTotalCountInTable(ctx context.Context, tableName string) (int64, error)
  • func (*Postgres) IsEmpty(_ model.Warehouse) (empty bool, err error)
  • func (*Postgres) LoadIdentityMappingsTable() (err error)
  • func (*Postgres) LoadIdentityMergeRulesTable() (err error)
  • func (pg *Postgres) LoadTable(tableName string) error
  • func (pg *Postgres) LoadTestTable(_, tableName string, payloadMap map[string]interface{}, _ string) (err error)
  • func (pg *Postgres) LoadUserTables() map[string]error
  • func (pg *Postgres) SetConnectionTimeout(timeout time.Duration)
  • func (pg *Postgres) Setup(warehouse model.Warehouse, uploader warehouseutils.Uploader) (err error)
  • func (pg *Postgres) TestConnection(warehouse model.Warehouse) (err error)
• type Tx

Functions

func ColumnsWithDataTypes

func ColumnsWithDataTypes(columns model.TableSchema, prefix string) string

func Connect

func Connect(cred Credentials) (*sql.DB, error)

func Init

func Init()

func WithConfig

func WithConfig(h *Postgres, config *config.Config)

Types

type Credentials

type Credentials struct {
	Host       string
	DBName     string
	User       string
	Password   string
	Port       string
	SSLMode    string
	SSLDir     string
	TunnelInfo *tunnelling.TunnelInfo
	// contains filtered or unexported fields
}

type Diagnostic

type Diagnostic struct {
	Logger    logger.Logger
	Namespace string
	Warehouse *model.Warehouse
	Stats     stats.Stats
	Config    *config.Config
}

func (*Diagnostic) TxnExecute

func (d *Diagnostic) TxnExecute(ctx context.Context, txn *sql.Tx, tableName, query string) error

TxnExecute Print execution plan if enableWithQueryPlan is set to true else return result set. Currently, these statements are supported by EXPLAIN Any INSERT, UPDATE, DELETE whose execution plan you wish to see.

func (*Diagnostic) TxnRollback

func (d *Diagnostic) TxnRollback(txn Tx, tableName, stage string)

TxnRollback rolls back a transaction and logs the error.

type LoadTable

type LoadTable struct {
	Logger             logger.Logger
	DB                 *sql.DB
	Namespace          string
	Warehouse          *model.Warehouse
	Stats              stats.Stats
	Config             *config.Config
	LoadFileDownloader downloader.Downloader
}

func (*LoadTable) Load

func (lt *LoadTable) Load(ctx context.Context, tableName string, tableSchemaInUpload model.TableSchema) (string, error)

type LoadUsersTable

type LoadUsersTable struct {
	Logger             logger.Logger
	DB                 *sql.DB
	Namespace          string
	Warehouse          *model.Warehouse
	Stats              stats.Stats
	Config             *config.Config
	LoadFileDownloader downloader.Downloader
}

func (*LoadUsersTable) Load

func (lut *LoadUsersTable) Load(ctx context.Context, identifiesSchemaInUpload, usersSchemaInUpload, usersSchemaInWarehouse model.TableSchema) map[string]error

type Postgres

type Postgres struct {
	DB                          *sql.DB
	Namespace                   string
	ObjectStorage               string
	Warehouse                   model.Warehouse
	Uploader                    warehouseutils.Uploader
	ConnectTimeout              time.Duration
	Logger                      logger.Logger
	EnableDeleteByJobs          bool
	NumWorkersDownloadLoadFiles int
	LoadFileDownloader          downloader.Downloader
}

func NewPostgres

func NewPostgres() *Postgres

func (*Postgres) AddColumns

func (pg *Postgres) AddColumns(tableName string, columnsInfo []warehouseutils.ColumnInfo) (err error)

func (*Postgres) AlterColumn

func (*Postgres) AlterColumn(_, _, _ string) (model.AlterTableResponse, error)

func (*Postgres) Cleanup

func (pg *Postgres) Cleanup()

func (*Postgres) Connect

func (pg *Postgres) Connect(warehouse model.Warehouse) (client.Client, error)

func (*Postgres) CrashRecover

func (pg *Postgres) CrashRecover(model.Warehouse) error

func (*Postgres) CreateSchema

func (pg *Postgres) CreateSchema() (err error)

func (*Postgres) CreateTable

func (pg *Postgres) CreateTable(tableName string, columnMap model.TableSchema) (err error)

func (*Postgres) DeleteBy

func (pg *Postgres) DeleteBy(tableNames []string, params warehouseutils.DeleteByParams) (err error)

DeleteBy Need to create a structure with delete parameters instead of simply adding a long list of params

func (*Postgres) DownloadIdentityRules

func (*Postgres) DownloadIdentityRules(*misc.GZipWriter) (err error)

func (*Postgres) DropTable

func (pg *Postgres) DropTable(tableName string) (err error)

func (*Postgres) ErrorMappings

func (pg *Postgres) ErrorMappings() []model.JobError

func (*Postgres) FetchSchema

func (pg *Postgres) FetchSchema(warehouse model.Warehouse) (schema, unrecognizedSchema model.Schema, err error)

FetchSchema queries postgres and returns the schema associated with provided namespace

func (*Postgres) GetTotalCountInTable

func (pg *Postgres) GetTotalCountInTable(ctx context.Context, tableName string) (int64, error)

func (*Postgres) IsEmpty

func (*Postgres) IsEmpty(_ model.Warehouse) (empty bool, err error)

func (*Postgres) LoadIdentityMappingsTable

func (*Postgres) LoadIdentityMappingsTable() (err error)

func (*Postgres) LoadIdentityMergeRulesTable

func (*Postgres) LoadIdentityMergeRulesTable() (err error)

func (*Postgres) LoadTable

func (pg *Postgres) LoadTable(tableName string) error

func (*Postgres) LoadTestTable

func (pg *Postgres) LoadTestTable(_, tableName string, payloadMap map[string]interface{}, _ string) (err error)

func (*Postgres) LoadUserTables

func (pg *Postgres) LoadUserTables() map[string]error

func (*Postgres) SetConnectionTimeout

func (pg *Postgres) SetConnectionTimeout(timeout time.Duration)

func (*Postgres) Setup

func (pg *Postgres) Setup(
	warehouse model.Warehouse,
	uploader warehouseutils.Uploader,
) (err error)

func (*Postgres) TestConnection

func (pg *Postgres) TestConnection(warehouse model.Warehouse) (err error)

type Tx

type Tx interface {
	Rollback() error
}