deltalake

Documentation

Index

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

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

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 New

func New() *Deltalake

func (*Deltalake) AddColumns

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

func (*Deltalake) AlterColumn

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

AlterColumn alter table with column name and type

func (*Deltalake) Cleanup

func (dl *Deltalake) Cleanup(ctx context.Context)

Cleanup cleanup when upload is done.

func (*Deltalake) Connect

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

Connect returns Client

func (*Deltalake) CrashRecover

func (dl *Deltalake) CrashRecover(ctx context.Context)

CrashRecover crash recover scenarios

func (*Deltalake) CreateSchema

func (dl *Deltalake) CreateSchema(ctx context.Context) (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(ctx context.Context, tableName string, columns model.TableSchema) (err error)

CreateTable creates tables with table name and columns

func (*Deltalake) DeleteBy

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

func (*Deltalake) DownloadIdentityRules

func (*Deltalake) DownloadIdentityRules(context.Context, *misc.GZipWriter) (err error)

DownloadIdentityRules download identity rules

func (*Deltalake) DropTable

func (dl *Deltalake) DropTable(ctx context.Context, tableName string) (err error)

func (*Deltalake) ErrorMappings

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

func (*Deltalake) ExecuteSQLClient

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

ExecuteSQLClient executes sql client using grpc Client

func (*Deltalake) FetchSchema

func (dl *Deltalake) FetchSchema(ctx context.Context) (model.Schema, model.Schema, 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(context.Context, model.Warehouse) (empty bool, err error)

IsEmpty checks if the warehouse is empty or not

func (*Deltalake) LoadIdentityMappingsTable

func (*Deltalake) LoadIdentityMappingsTable(context.Context) (err error)

LoadIdentityMappingsTable loads identifies mappings table

func (*Deltalake) LoadIdentityMergeRulesTable

func (*Deltalake) LoadIdentityMergeRulesTable(context.Context) (err error)

LoadIdentityMergeRulesTable loads identifies merge rules tables

func (*Deltalake) LoadTable

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

LoadTable loads table for table name

func (*Deltalake) LoadTestTable

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

func (*Deltalake) LoadUserTables

func (dl *Deltalake) LoadUserTables(ctx context.Context) map[string]error

LoadUserTables loads user tables

func (*Deltalake) NewClient

func (dl *Deltalake) NewClient(ctx context.Context, 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(ctx context.Context, warehouse model.Warehouse, uploader warehouseutils.Uploader) (err error)

Setup populate the Deltalake

func (*Deltalake) TestConnection

func (*Deltalake) TestConnection(context.Context, model.Warehouse) error

TestConnection test the connection for the warehouse