deltalake

Documentation

Index

• Constants
• func ColumnsWithDataTypes(columns model.TableSchema, prefix string) string
• func GetDatabricksVersion() (databricksBuildVersion string)
• func Init()
• func WithConfig(h *Deltalake, config *config.Config)
• type Deltalake
  • func NewDeltalake() *Deltalake
  • func (dl *Deltalake) AddColumns(tableName string, columnsInfo []warehouseutils.ColumnInfo) error
  • func (*Deltalake) AlterColumn(_, _, _ string) (model.AlterTableResponse, error)
  • func (dl *Deltalake) Cleanup()
  • func (dl *Deltalake) Connect(warehouse model.Warehouse) (warehouseclient.Client, error)
  • func (dl *Deltalake) CrashRecover(warehouse model.Warehouse) (err error)
  • func (dl *Deltalake) CreateSchema() (err error)
  • func (dl *Deltalake) CreateTable(tableName string, columns model.TableSchema) (err error)
  • func (*Deltalake) DeleteBy([]string, warehouseutils.DeleteByParams) error
  • func (*Deltalake) DownloadIdentityRules(*misc.GZipWriter) (err error)
  • func (dl *Deltalake) DropTable(tableName string) (err error)
  • func (dl *Deltalake) ErrorMappings() []model.JobError
  • func (*Deltalake) ExecuteSQLClient(client *client.Client, sqlStatement string) (err error)
  • func (dl *Deltalake) FetchSchema(warehouse model.Warehouse) (schema, unrecognizedSchema model.Schema, err error)
  • func (dl *Deltalake) GetLogIdentifier(args ...string) string
  • func (dl *Deltalake) GetTotalCountInTable(ctx context.Context, tableName string) (int64, error)
  • func (*Deltalake) IsEmpty(model.Warehouse) (empty bool, err error)
  • func (*Deltalake) LoadIdentityMappingsTable() (err error)
  • func (*Deltalake) LoadIdentityMergeRulesTable() (err error)
  • func (dl *Deltalake) LoadTable(tableName string) error
  • func (dl *Deltalake) LoadTestTable(location, tableName string, _ map[string]interface{}, format string) (err error)
  • func (dl *Deltalake) LoadUserTables() map[string]error
  • func (dl *Deltalake) NewClient(cred *client.Credentials, connectTimeout time.Duration) (Client *client.Client, err error)
  • func (dl *Deltalake) SetConnectionTimeout(timeout time.Duration)
  • func (dl *Deltalake) Setup(warehouse model.Warehouse, uploader warehouseutils.Uploader) (err error)
  • func (dl *Deltalake) TestConnection(warehouse model.Warehouse) (err error)

Constants

const (
	Host                   = "host"
	Port                   = "port"
	Path                   = "path"
	Token                  = "token"
	Catalog                = "catalog"
	UseSTSTokens           = "useSTSTokens"
	EnableExternalLocation = "enableExternalLocation"
	ExternalLocation       = "externalLocation"
)

Database configuration

Functions

func ColumnsWithDataTypes

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

ColumnsWithDataTypes returns columns with specified prefix and data type

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

func WithConfig

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

Types

type Deltalake

type Deltalake struct {
	Client                 *client.Client
	Namespace              string
	ObjectStorage          string
	Warehouse              model.Warehouse
	Uploader               warehouseutils.Uploader
	ConnectTimeout         time.Duration
	Logger                 logger.Logger
	Stats                  stats.Stats
	Schema                 string
	SparkServerType        string
	AuthMech               string
	UID                    string
	ThriftTransport        string
	SSL                    string
	UserAgent              string
	GrpcTimeout            time.Duration
	HealthTimeout          time.Duration
	LoadTableStrategy      string
	EnablePartitionPruning bool
	ConnectorURL           string
}

func NewDeltalake

func NewDeltalake() *Deltalake

func (*Deltalake) AddColumns

func (dl *Deltalake) AddColumns(tableName string, columnsInfo []warehouseutils.ColumnInfo) error

func (*Deltalake) AlterColumn

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

AlterColumn alter table with column name and type

func (*Deltalake) Cleanup

func (dl *Deltalake) Cleanup()

Cleanup cleanup when upload is done.

func (*Deltalake) Connect

func (dl *Deltalake) Connect(warehouse model.Warehouse) (warehouseclient.Client, error)

Connect returns Client

func (*Deltalake) CrashRecover

func (dl *Deltalake) CrashRecover(warehouse model.Warehouse) (err error)

CrashRecover crash recover scenarios

func (*Deltalake) CreateSchema

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

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

func (*Deltalake) CreateTable

func (dl *Deltalake) CreateTable(tableName string, columns model.TableSchema) (err error)

CreateTable creates tables with table name and columns

func (*Deltalake) DeleteBy

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

func (*Deltalake) DownloadIdentityRules

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

DownloadIdentityRules download identity rules

func (*Deltalake) DropTable

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

func (*Deltalake) ErrorMappings

func (dl *Deltalake) ErrorMappings() []model.JobError

func (*Deltalake) ExecuteSQLClient

func (*Deltalake) ExecuteSQLClient(client *client.Client, sqlStatement string) (err error)

ExecuteSQLClient executes sql client using grpc Client

func (*Deltalake) FetchSchema

func (dl *Deltalake) FetchSchema(warehouse model.Warehouse) (schema, unrecognizedSchema model.Schema, err error)

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

func (*Deltalake) GetLogIdentifier

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

GetLogIdentifier returns log identifier

func (*Deltalake) GetTotalCountInTable

func (dl *Deltalake) GetTotalCountInTable(ctx context.Context, tableName string) (int64, error)

GetTotalCountInTable returns the total count in the table

func (*Deltalake) IsEmpty

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

IsEmpty checks if the warehouse is empty or not

func (*Deltalake) LoadIdentityMappingsTable

func (*Deltalake) LoadIdentityMappingsTable() (err error)

LoadIdentityMappingsTable loads identifies mappings table

func (*Deltalake) LoadIdentityMergeRulesTable

func (*Deltalake) LoadIdentityMergeRulesTable() (err error)

LoadIdentityMergeRulesTable loads identifies merge rules tables

func (*Deltalake) LoadTable

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

LoadTable loads table for table name

func (*Deltalake) LoadTestTable

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

func (*Deltalake) LoadUserTables

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

LoadUserTables loads user tables

func (*Deltalake) NewClient

func (dl *Deltalake) NewClient(cred *client.Credentials, connectTimeout time.Duration) (Client *client.Client, err error)

NewClient creates deltalake client

func (*Deltalake) SetConnectionTimeout

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

func (*Deltalake) Setup

func (dl *Deltalake) Setup(warehouse model.Warehouse, uploader warehouseutils.Uploader) (err error)

Setup populate the Deltalake

func (*Deltalake) TestConnection

func (dl *Deltalake) TestConnection(warehouse model.Warehouse) (err error)

TestConnection test the connection for the warehouse