clickhouse

Documentation

Index

• Constants
• func ColumnsWithDataTypes(tableName string, columns map[string]string, notNullableColumns []string) string
• func Connect(cred CredentialsT, includeDBInConn bool) (*sql.DB, error)
• func Init()
• type CredentialsT
• type HandleT
  • func (ch *HandleT) AddColumn(tableName, columnName, columnType string) (err error)
  • func (*HandleT) AlterColumn(_, _, _ string) (err error)
  • func (ch *HandleT) Cleanup()
  • func (ch *HandleT) Connect(warehouse warehouseutils.Warehouse) (client.Client, error)
  • func (*HandleT) CrashRecover(_ warehouseutils.Warehouse) (err error)
  • func (ch *HandleT) CreateSchema() (err error)
  • func (ch *HandleT) CreateTable(tableName string, columns map[string]string) (err error)
  • func (*HandleT) DeleteBy([]string, warehouseutils.DeleteByParams) error
  • func (*HandleT) DownloadIdentityRules(*misc.GZipWriter) (err error)
  • func (ch *HandleT) DownloadLoadFiles(tableName string) ([]string, error)
  • func (ch *HandleT) DropTable(tableName string) (err error)
  • func (ch *HandleT) FetchSchema(warehouse warehouseutils.Warehouse) (schema warehouseutils.SchemaT, err error)
  • func (ch *HandleT) GetLogIdentifier(args ...string) string
  • func (ch *HandleT) GetTotalCountInTable(ctx context.Context, tableName string) (total int64, err error)
  • func (*HandleT) IsEmpty(_ warehouseutils.Warehouse) (empty bool, err error)
  • func (*HandleT) LoadIdentityMappingsTable() (err error)
  • func (*HandleT) LoadIdentityMergeRulesTable() (err error)
  • func (ch *HandleT) LoadTable(tableName string) error
  • func (ch *HandleT) LoadTestTable(_, tableName string, payloadMap map[string]interface{}, _ string) (err error)
  • func (ch *HandleT) LoadUserTables() (errorMap map[string]error)
  • func (ch *HandleT) SetConnectionTimeout(timeout time.Duration)
  • func (ch *HandleT) Setup(warehouse warehouseutils.Warehouse, uploader warehouseutils.UploaderI) (err error)
  • func (ch *HandleT) TestConnection(warehouse warehouseutils.Warehouse) (err error)

Constants

const (
	Cluster = "cluster"
)

Functions

func ColumnsWithDataTypes

func ColumnsWithDataTypes(tableName string, columns map[string]string, notNullableColumns []string) string

ColumnsWithDataTypes creates columns and its datatype into sql format for creating table

func Connect

func Connect(cred CredentialsT, includeDBInConn bool) (*sql.DB, error)

Connect connects to warehouse with provided credentials

func Init

func Init()

Types

type CredentialsT

type CredentialsT struct {
	Host          string
	DBName        string
	User          string
	Password      string
	Port          string
	Secure        string
	SkipVerify    string
	TLSConfigName string
	// contains filtered or unexported fields
}

type HandleT

type HandleT struct {
	Db            *sql.DB
	Namespace     string
	ObjectStorage string
	Warehouse     warehouseutils.Warehouse
	Uploader      warehouseutils.UploaderI

	ConnectTimeout time.Duration
	// contains filtered or unexported fields
}

func (*HandleT) AddColumn

func (ch *HandleT) AddColumn(tableName, columnName, columnType string) (err error)

AddColumn adds column:columnName with dataType columnType to the tableName

func (*HandleT) AlterColumn

func (*HandleT) AlterColumn(_, _, _ string) (err error)

func (*HandleT) Cleanup

func (ch *HandleT) Cleanup()

func (*HandleT) Connect

func (ch *HandleT) Connect(warehouse warehouseutils.Warehouse) (client.Client, error)

func (*HandleT) CrashRecover

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

func (*HandleT) CreateSchema

func (ch *HandleT) CreateSchema() (err error)

func (*HandleT) CreateTable

func (ch *HandleT) CreateTable(tableName string, columns map[string]string) (err error)

CreateTable creates table with engine ReplacingMergeTree(), this is used for dedupe event data and replace it will the latest data if duplicate data found. This logic is handled by clickhouse The engine differs from MergeTree in that it removes duplicate entries with the same sorting key value.

func (*HandleT) DeleteBy

func (*HandleT) DeleteBy([]string, warehouseutils.DeleteByParams) error

func (*HandleT) DownloadIdentityRules

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

func (*HandleT) DownloadLoadFiles

func (ch *HandleT) DownloadLoadFiles(tableName string) ([]string, error)

DownloadLoadFiles downloads load files for the tableName and gives file names

func (*HandleT) DropTable

func (ch *HandleT) DropTable(tableName string) (err error)

func (*HandleT) FetchSchema

func (ch *HandleT) FetchSchema(warehouse warehouseutils.Warehouse) (schema warehouseutils.SchemaT, err error)

FetchSchema queries clickhouse and returns the schema associated with provided namespace

func (*HandleT) GetLogIdentifier

func (ch *HandleT) GetLogIdentifier(args ...string) string

func (*HandleT) GetTotalCountInTable

func (ch *HandleT) GetTotalCountInTable(ctx context.Context, tableName string) (total int64, err error)

func (*HandleT) IsEmpty

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

func (*HandleT) LoadIdentityMappingsTable

func (*HandleT) LoadIdentityMappingsTable() (err error)

func (*HandleT) LoadIdentityMergeRulesTable

func (*HandleT) LoadIdentityMergeRulesTable() (err error)

func (*HandleT) LoadTable

func (ch *HandleT) LoadTable(tableName string) error

func (*HandleT) LoadTestTable

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

func (*HandleT) LoadUserTables

func (ch *HandleT) LoadUserTables() (errorMap map[string]error)

func (*HandleT) SetConnectionTimeout

func (ch *HandleT) SetConnectionTimeout(timeout time.Duration)

func (*HandleT) Setup

func (ch *HandleT) Setup(warehouse warehouseutils.Warehouse, uploader warehouseutils.UploaderI) (err error)

func (*HandleT) TestConnection

func (ch *HandleT) TestConnection(warehouse warehouseutils.Warehouse) (err error)

TestConnection is used destination connection tester to test the clickhouse connection