redshift

Documentation

Index

• Constants
• func CheckAndIgnoreColumnAlreadyExistError(err error) bool
• func ColumnsWithDataTypes(columns map[string]string, prefix string) string
• func Connect(cred RedshiftCredentials) (*sql.DB, error)
• func Init()
• func WithConfig(h *Redshift, config *config.Config)
• type Redshift
  • func NewRedshift() *Redshift
  • func (rs *Redshift) AddColumns(tableName string, columnsInfo []warehouseutils.ColumnInfo) error
  • func (rs *Redshift) AlterColumn(tableName, columnName, columnType string) (model.AlterTableResponse, error)
  • func (rs *Redshift) Cleanup()
  • func (rs *Redshift) Connect(warehouse warehouseutils.Warehouse) (client.Client, error)
  • func (rs *Redshift) CrashRecover(warehouse warehouseutils.Warehouse) (err error)
  • func (rs *Redshift) CreateSchema() (err error)
  • func (rs *Redshift) CreateTable(tableName string, columns map[string]string) (err error)
  • func (rs *Redshift) DeleteBy(tableNames []string, params warehouseutils.DeleteByParams) (err error)
  • func (*Redshift) DownloadIdentityRules(*misc.GZipWriter) (err error)
  • func (rs *Redshift) DropTable(tableName string) (err error)
  • func (rs *Redshift) ErrorMappings() []model.JobError
  • func (rs *Redshift) FetchSchema(warehouse warehouseutils.Warehouse) (schema, unrecognizedSchema warehouseutils.SchemaT, err error)
  • func (rs *Redshift) GetTotalCountInTable(ctx context.Context, tableName string) (int64, error)
  • func (*Redshift) IsEmpty(_ warehouseutils.Warehouse) (empty bool, err error)
  • func (*Redshift) LoadIdentityMappingsTable() (err error)
  • func (*Redshift) LoadIdentityMergeRulesTable() (err error)
  • func (rs *Redshift) LoadTable(tableName string) error
  • func (rs *Redshift) LoadTestTable(location, tableName string, _ map[string]interface{}, format string) (err error)
  • func (rs *Redshift) LoadUserTables() map[string]error
  • func (rs *Redshift) SetConnectionTimeout(timeout time.Duration)
  • func (rs *Redshift) Setup(warehouse warehouseutils.Warehouse, uploader warehouseutils.UploaderI) (err error)
  • func (rs *Redshift) TestConnection(warehouse warehouseutils.Warehouse) (err error)
• type RedshiftCredentials
• type S3Manifest
• type S3ManifestEntry
• type S3ManifestEntryMetadata

Constants

const (
	RSHost     = "host"
	RSPort     = "port"
	RSDbName   = "database"
	RSUserName = "user"
	RSPassword = "password"
)

String constants for redshift destination config

Functions

func CheckAndIgnoreColumnAlreadyExistError

func CheckAndIgnoreColumnAlreadyExistError(err error) bool

func ColumnsWithDataTypes

func ColumnsWithDataTypes(columns map[string]string, prefix string) string

func Connect

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

func Init

func Init()

func WithConfig

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

Types

type Redshift

type Redshift struct {
	DB             *sql.DB
	Namespace      string
	Warehouse      warehouseutils.Warehouse
	Uploader       warehouseutils.UploaderI
	ConnectTimeout time.Duration
	Logger         logger.Logger

	DedupWindow                   bool
	DedupWindowInHours            time.Duration
	SkipDedupDestinationIDs       []string
	SkipComputingUserLatestTraits bool
	EnableDeleteByJobs            bool
	// contains filtered or unexported fields
}

func NewRedshift

func NewRedshift() *Redshift

func (*Redshift) AddColumns

func (rs *Redshift) AddColumns(tableName string, columnsInfo []warehouseutils.ColumnInfo) error

func (*Redshift) AlterColumn

func (rs *Redshift) AlterColumn(tableName, columnName, columnType string) (model.AlterTableResponse, error)

func (*Redshift) Cleanup

func (rs *Redshift) Cleanup()

func (*Redshift) Connect

func (rs *Redshift) Connect(warehouse warehouseutils.Warehouse) (client.Client, error)

func (*Redshift) CrashRecover

func (rs *Redshift) CrashRecover(warehouse warehouseutils.Warehouse) (err error)

func (*Redshift) CreateSchema

func (rs *Redshift) CreateSchema() (err error)

func (*Redshift) CreateTable

func (rs *Redshift) CreateTable(tableName string, columns map[string]string) (err error)

func (*Redshift) DeleteBy

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

func (*Redshift) DownloadIdentityRules

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

func (*Redshift) DropTable

func (rs *Redshift) DropTable(tableName string) (err error)

func (*Redshift) ErrorMappings

func (rs *Redshift) ErrorMappings() []model.JobError

func (*Redshift) FetchSchema

func (rs *Redshift) FetchSchema(warehouse warehouseutils.Warehouse) (schema, unrecognizedSchema warehouseutils.SchemaT, err error)

FetchSchema queries redshift and returns the schema associated with provided namespace

func (*Redshift) GetTotalCountInTable

func (rs *Redshift) GetTotalCountInTable(ctx context.Context, tableName string) (int64, error)

func (*Redshift) IsEmpty

func (*Redshift) IsEmpty(_ warehouseutils.Warehouse) (empty bool, err error)

func (*Redshift) LoadIdentityMappingsTable

func (*Redshift) LoadIdentityMappingsTable() (err error)

func (*Redshift) LoadIdentityMergeRulesTable

func (*Redshift) LoadIdentityMergeRulesTable() (err error)

func (*Redshift) LoadTable

func (rs *Redshift) LoadTable(tableName string) error

func (*Redshift) LoadTestTable

func (rs *Redshift) LoadTestTable(location, tableName string, _ map[string]interface{}, format string) (err error)

func (*Redshift) LoadUserTables

func (rs *Redshift) LoadUserTables() map[string]error

func (*Redshift) SetConnectionTimeout

func (rs *Redshift) SetConnectionTimeout(timeout time.Duration)

func (*Redshift) Setup

func (rs *Redshift) Setup(warehouse warehouseutils.Warehouse, uploader warehouseutils.UploaderI) (err error)

func (*Redshift) TestConnection

func (rs *Redshift) TestConnection(warehouse warehouseutils.Warehouse) (err error)

type RedshiftCredentials

type RedshiftCredentials struct {
	Host     string
	Port     string
	DbName   string
	Username string
	Password string

	TunnelInfo *tunnelling.TunnelInfo
	// contains filtered or unexported fields
}

type S3Manifest

type S3Manifest struct {
	Entries []S3ManifestEntry `json:"entries"`
}

type S3ManifestEntry

type S3ManifestEntry struct {
	Url       string                  `json:"url"`
	Mandatory bool                    `json:"mandatory"`
	Metadata  S3ManifestEntryMetadata `json:"meta"`
}

type S3ManifestEntryMetadata

type S3ManifestEntryMetadata struct {
	ContentLength int64 `json:"content_length"`
}