deltalake

Documentation

Index

• Constants
• func ColumnsWithDataTypes(columns map[string]string, prefix string) string
• func Connect(cred *databricks.CredentialsT, connectTimeout time.Duration) (dbHandleT *databricks.DBHandleT, err error)
• func GetDatabricksConnectorURL() string
• func GetDatabricksVersion() (databricksBuildVersion string)
• func Init()
• type HandleT
  • func (dl *HandleT) AddColumn(name, columnName, columnType string) (err error)
  • func (*HandleT) AlterColumn(_, _, _ string) (err error)
  • func (dl *HandleT) Cleanup()
  • func (dl *HandleT) Connect(warehouse warehouseutils.Warehouse) (client.Client, error)
  • func (dl *HandleT) CrashRecover(warehouse warehouseutils.Warehouse) (err error)
  • func (dl *HandleT) CreateSchema() (err error)
  • func (dl *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 (dl *HandleT) DropTable(tableName string) (err error)
  • func (dl *HandleT) ExecuteSQL(sqlStatement, queryType string) (err error)
  • func (*HandleT) ExecuteSQLClient(dbClient *databricks.DBHandleT, sqlStatement string) (err error)
  • func (dl *HandleT) FetchSchema(warehouse warehouseutils.Warehouse) (schema warehouseutils.SchemaT, err error)
  • func (dl *HandleT) GetLogIdentifier(args ...string) string
  • func (dl *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 (dl *HandleT) LoadTable(tableName string) error
  • func (dl *HandleT) LoadTestTable(location, tableName string, _ map[string]interface{}, format string) (err error)
  • func (dl *HandleT) LoadUserTables() map[string]error
  • func (dl *HandleT) SetConnectionTimeout(timeout time.Duration)
  • func (dl *HandleT) Setup(warehouse warehouseutils.Warehouse, uploader warehouseutils.UploaderI) (err error)
  • func (dl *HandleT) TestConnection(warehouse warehouseutils.Warehouse) (err error)

Constants

const (
	DLHost                 = "host"
	DLPort                 = "port"
	DLPath                 = "path"
	DLToken                = "token"
	AWSTokens              = "useSTSTokens"
	EnableExternalLocation = "enableExternalLocation"
	ExternalLocation       = "externalLocation"
)

Database configuration

Functions

func ColumnsWithDataTypes

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

ColumnsWithDataTypes returns columns with specified prefix and data type

func Connect

func Connect(cred *databricks.CredentialsT, connectTimeout time.Duration) (dbHandleT *databricks.DBHandleT, err error)

Connect creates database connection with CredentialsT

func GetDatabricksConnectorURL

func GetDatabricksConnectorURL() string

GetDatabricksConnectorURL returns databricks connector url.

func GetDatabricksVersion

func GetDatabricksVersion() (databricksBuildVersion string)

GetDatabricksVersion Gets the databricks version by making a grpc call to Version stub.

func Init

func Init()

Init initializes the delta lake warehouse

Types

type HandleT

type HandleT struct {
	Namespace      string
	ObjectStorage  string
	Warehouse      warehouseutils.Warehouse
	Uploader       warehouseutils.UploaderI
	ConnectTimeout time.Duration
	// contains filtered or unexported fields
}

func (*HandleT) AddColumn

func (dl *HandleT) AddColumn(name, columnName, columnType string) (err error)

AddColumn adds column for column name and type

func (*HandleT) AlterColumn

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

AlterColumn alter table with column name and type

func (*HandleT) Cleanup

func (dl *HandleT) Cleanup()

Cleanup handle cleanup when upload is done.

func (*HandleT) Connect

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

Connect returns Client

func (*HandleT) CrashRecover

func (dl *HandleT) CrashRecover(warehouse warehouseutils.Warehouse) (err error)

CrashRecover handle crash recover scenarios

func (*HandleT) CreateSchema

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

CreateSchema checks if schema exists or not. If it does not exist, it creates the schema.

func (*HandleT) CreateTable

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

CreateTable creates tables with table name and columns

func (*HandleT) DeleteBy

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

func (*HandleT) DownloadIdentityRules

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

DownloadIdentityRules download identity rules

func (*HandleT) DropTable

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

func (*HandleT) ExecuteSQL

func (dl *HandleT) ExecuteSQL(sqlStatement, queryType string) (err error)

ExecuteSQL executes sql using grpc Client

func (*HandleT) ExecuteSQLClient

func (*HandleT) ExecuteSQLClient(dbClient *databricks.DBHandleT, sqlStatement string) (err error)

ExecuteSQLClient executes sql client using grpc Client

func (*HandleT) FetchSchema

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

FetchSchema queries delta lake and returns the schema associated with provided namespace

func (*HandleT) GetLogIdentifier

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

GetLogIdentifier returns log identifier

func (*HandleT) GetTotalCountInTable

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

GetTotalCountInTable returns total count in tables.

func (*HandleT) IsEmpty

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

IsEmpty checks if the warehouse is empty or not

func (*HandleT) LoadIdentityMappingsTable

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

LoadIdentityMappingsTable loads identifies mappings table

func (*HandleT) LoadIdentityMergeRulesTable

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

LoadIdentityMergeRulesTable loads identifies merge rules tables

func (*HandleT) LoadTable

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

LoadTable loads table for table name

func (*HandleT) LoadTestTable

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

func (*HandleT) LoadUserTables

func (dl *HandleT) LoadUserTables() map[string]error

LoadUserTables loads user tables

func (*HandleT) SetConnectionTimeout

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

func (*HandleT) Setup

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

Setup populate the HandleT

func (*HandleT) TestConnection

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

TestConnection test the connection for the warehouse