model

Documentation

Index

• Constants
• Variables
• func GetLastFailedStatus(timingsMap Timings) (status string)
• func GetLoadFileGenTime(timingsMap Timings) (t time.Time)
• type AlterTableResponse
• type DestinationValidationResponse
• type JobError
• type JobErrorType
• type LoadFile
• type Matcher
• type PendingTableUpload
• type Schema
• type SchemaType
• type StagingFile
  • func (s StagingFile) WithSchema(schema json.RawMessage) StagingFileWithSchema
• type StagingFileWithSchema
• type Step
• type StepsResponse
• type TableSchema
• type TableUpload
• type Timings
• type Upload
• type UploadJob
• type UploadJobsStats
• type UploadStatus
• type ValidationRequest
• type ValidationResponse
• type Warehouse
  • func (w *Warehouse) GetBoolDestinationConfig(key string) bool

Constants

const (
	TableUploadWaiting              = "waiting"
	TableUploadExecuting            = "executing"
	TableUploadUpdatingSchema       = "updating_schema"
	TableUploadUpdatingSchemaFailed = "updating_schema_failed"
	TableUploadUpdatedSchema        = "updated_schema"
	TableUploadExporting            = "exporting_data"
	TableUploadExportingFailed      = "exporting_data_failed"
	TableUploadExported             = "exported_data"
)

const (
	Waiting                   = "waiting"
	GeneratedUploadSchema     = "generated_upload_schema"
	CreatedTableUploads       = "created_table_uploads"
	GeneratingLoadFiles       = "generating_load_files"
	GeneratedLoadFiles        = "generated_load_files"
	UpdatedTableUploadsCounts = "updated_table_uploads_counts"
	CreatedRemoteSchema       = "created_remote_schema"
	ExportedData              = "exported_data"
	ExportingData             = "exporting_data"
	ExportingDataFailed       = "exporting_data_failed"
	Aborted                   = "aborted"
	Failed                    = "failed"
)

const (
	VerifyingObjectStorage       = "Verifying Object Storage"
	VerifyingConnections         = "Verifying Connections"
	VerifyingCreateSchema        = "Verifying Create Schema"
	VerifyingCreateAndAlterTable = "Verifying Create and Alter Table"
	VerifyingFetchSchema         = "Verifying Fetch Schema"
	VerifyingLoadTable           = "Verifying Load Table"
)

Variables

var ErrUploadNotFound = errors.New("upload not found")

Functions

func GetLastFailedStatus

func GetLastFailedStatus(timingsMap Timings) (status string)

func GetLoadFileGenTime

func GetLoadFileGenTime(timingsMap Timings) (t time.Time)

Types

type AlterTableResponse

type AlterTableResponse struct {
	IsDependent bool // true if the column is dependent on another view or rules, false otherwise
	Query       string
}

type DestinationValidationResponse

type DestinationValidationResponse struct {
	Success bool    `json:"success"`
	Error   string  `json:"error"`
	Steps   []*Step `json:"steps"`
}

type JobError

type JobError struct {
	Type   JobErrorType
	Format Matcher
}

type JobErrorType

type JobErrorType string

const (
	PermissionError           JobErrorType = "permission_error"
	AlterColumnError          JobErrorType = "alter_column_error"
	ResourceNotFoundError     JobErrorType = "resource_not_found_error"
	ColumnCountError          JobErrorType = "column_count_error"
	ColumnSizeError           JobErrorType = "column_size_error"
	InsufficientResourceError JobErrorType = "insufficient_resource_error"
	ConcurrentQueriesError    JobErrorType = "concurrent_queries_error"
	UnknownError              JobErrorType = "unknown_error"
	Noop                      JobErrorType = "noop"
)

type LoadFile

type LoadFile struct {
	ID                    int64
	TableName             string
	Location              string
	TotalRows             int
	ContentLength         int64
	StagingFileID         int64
	DestinationRevisionID string
	UseRudderStorage      bool
	SourceID              string
	DestinationID         string
	DestinationType       string
}

type Matcher

type Matcher interface {
	MatchString(string) bool
}

type PendingTableUpload

type PendingTableUpload struct {
	UploadID      int64
	DestinationID string
	Namespace     string
	TableName     string
	Status        string
	Error         string
}

type Schema

type Schema map[string]TableSchema

type SchemaType

type SchemaType string

const (
	StringDataType  SchemaType = "string"
	BooleanDataType SchemaType = "boolean"
	IntDataType     SchemaType = "int"
	BigIntDataType  SchemaType = "bigint"
	FloatDataType   SchemaType = "float"
	JSONDataType    SchemaType = "json"
	TextDataType    SchemaType = "text"
)

type StagingFile

type StagingFile struct {
	ID                    int64
	WorkspaceID           string
	Location              string
	SourceID              string
	DestinationID         string
	Status                string // enum
	Error                 error
	FirstEventAt          time.Time
	LastEventAt           time.Time
	UseRudderStorage      bool
	DestinationRevisionID string
	TotalEvents           int
	TotalBytes            int
	// cloud sources specific info
	SourceTaskRunID string
	SourceJobID     string
	SourceJobRunID  string
	TimeWindow      time.Time

	CreatedAt time.Time
	UpdatedAt time.Time
}

StagingFile a domain model for a staging file.

The staging file contains events that should be loaded into a warehouse.
It is located in a cloud storage bucket.
The model includes ownership, file location, and other metadata.

func (StagingFile) WithSchema

func (s StagingFile) WithSchema(schema json.RawMessage) StagingFileWithSchema

type StagingFileWithSchema

type StagingFileWithSchema struct {
	StagingFile
	Schema json.RawMessage
}

StagingFileWithSchema is a StagingFile with schema field for included events.

schema size can be really big, and thus it should be included only when required.

type Step

type Step struct {
	ID      int    `json:"id"`
	Name    string `json:"name"`
	Success bool   `json:"success"`
	Error   string `json:"error"`
}

type StepsResponse

type StepsResponse struct {
	Steps []*Step `json:"steps"`
}

type TableSchema

type TableSchema map[string]string

type TableUpload

type TableUpload struct {
	ID           int64
	UploadID     int64
	TableName    string
	Status       string
	Error        string
	LastExecTime time.Time
	TotalEvents  int64
	CreatedAt    time.Time
	UpdatedAt    time.Time
	Location     string
}

type Timings

type Timings []map[string]time.Time

type Upload

type Upload struct {
	ID          int64
	WorkspaceID string

	Namespace       string
	SourceID        string
	DestinationID   string
	DestinationType string
	Status          string
	Error           json.RawMessage
	FirstEventAt    time.Time
	LastEventAt     time.Time

	UseRudderStorage bool
	SourceTaskRunID  string
	SourceJobID      string
	SourceJobRunID   string
	LoadFileType     string
	NextRetryTime    time.Time
	Priority         int
	Retried          bool

	StagingFileStartID int64
	StagingFileEndID   int64

	LoadFileStartID int64
	LoadFileEndID   int64

	Timings        Timings
	FirstAttemptAt time.Time
	LastAttemptAt  time.Time
	Attempts       int64

	UploadSchema Schema
	MergedSchema Schema
}

type UploadJob

type UploadJob struct {
	Warehouse            Warehouse
	Upload               Upload
	StagingFiles         []*StagingFile
	LoadFileGenStartTime time.Time
}

type UploadJobsStats

type UploadJobsStats struct {
	PendingJobs    int64
	PickupLag      time.Duration
	PickupWaitTime time.Duration
}

type UploadStatus

type UploadStatus = string

type ValidationRequest

type ValidationRequest struct {
	Path        string
	Step        string
	Destination *backendconfig.DestinationT
}

type ValidationResponse

type ValidationResponse struct {
	Error string
	Data  string
}

type Warehouse

type Warehouse struct {
	WorkspaceID string
	Source      backendconfig.SourceT
	Destination backendconfig.DestinationT
	Namespace   string
	Type        string
	Identifier  string
}

func (*Warehouse) GetBoolDestinationConfig

func (w *Warehouse) GetBoolDestinationConfig(key string) bool