snowflake

Documentation

Index

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

Constants

const (
	StorageIntegration = "storageIntegration"
	Account            = "account"
	Warehouse          = "warehouse"
	Database           = "database"
	User               = "user"
	Role               = "role"
	Password           = "password"
	Application        = "Rudderstack"
)

String constants for snowflake destination config

Functions

func ColumnsWithDataTypes

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

func Connect

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

func Init

func Init()

func WithConfig

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

Types

type Credentials

type Credentials struct {
	Account   string
	Warehouse string
	Database  string
	User      string
	Role      string
	Password  string
	// contains filtered or unexported fields
}

type Snowflake

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

	EnableDeleteByJobs bool
}

func NewSnowflake

func NewSnowflake() *Snowflake

func (*Snowflake) AddColumns

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

func (*Snowflake) AlterColumn

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

func (*Snowflake) Cleanup

func (sf *Snowflake) Cleanup()

func (*Snowflake) Connect

func (sf *Snowflake) Connect(warehouse warehouseutils.Warehouse) (client.Client, error)

func (*Snowflake) CrashRecover

func (*Snowflake) CrashRecover(_ warehouseutils.Warehouse) (err error)

func (*Snowflake) CreateSchema

func (sf *Snowflake) CreateSchema() (err error)

func (*Snowflake) CreateTable

func (sf *Snowflake) CreateTable(tableName string, columnMap map[string]string) (err error)

func (*Snowflake) DeleteBy

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

func (*Snowflake) DownloadIdentityRules

func (sf *Snowflake) DownloadIdentityRules(gzWriter *misc.GZipWriter) (err error)

DownloadIdentityRules gets distinct combinations of anonymous_id, user_id from tables in warehouse

func (*Snowflake) DropTable

func (sf *Snowflake) DropTable(tableName string) (err error)

func (*Snowflake) ErrorMappings

func (sf *Snowflake) ErrorMappings() []model.JobError

func (*Snowflake) FetchSchema

func (sf *Snowflake) FetchSchema(warehouse warehouseutils.Warehouse) (schema, unrecognizedSchema warehouseutils.SchemaT, err error)

FetchSchema queries snowflake and returns the schema associated with provided namespace

func (*Snowflake) GetTotalCountInTable

func (sf *Snowflake) GetTotalCountInTable(ctx context.Context, tableName string) (int64, error)

func (*Snowflake) IsEmpty

func (sf *Snowflake) IsEmpty(warehouse warehouseutils.Warehouse) (empty bool, err error)

func (*Snowflake) LoadIdentityMappingsTable

func (sf *Snowflake) LoadIdentityMappingsTable() (err error)

func (*Snowflake) LoadIdentityMergeRulesTable

func (sf *Snowflake) LoadIdentityMergeRulesTable() (err error)

func (*Snowflake) LoadTable

func (sf *Snowflake) LoadTable(tableName string) error

func (*Snowflake) LoadTestTable

func (sf *Snowflake) LoadTestTable(location, tableName string, _ map[string]interface{}, _ string) (err error)

func (*Snowflake) LoadUserTables

func (sf *Snowflake) LoadUserTables() map[string]error

func (*Snowflake) SetConnectionTimeout

func (sf *Snowflake) SetConnectionTimeout(timeout time.Duration)

func (*Snowflake) Setup

func (sf *Snowflake) Setup(warehouse warehouseutils.Warehouse, uploader warehouseutils.UploaderI) (err error)

func (*Snowflake) TestConnection

func (sf *Snowflake) TestConnection(warehouse warehouseutils.Warehouse) (err error)