package entity import ( "database/sql/driver" "encoding/json" "net/url" "strings" "time" "github.com/pingcap/errors" lightningcfg "github.com/pingcap/tidb/pkg/lightning/config" "github.com/pingcap/tidb/pkg/util/filter" router "github.com/pingcap/tidb/pkg/util/table-router" "github.com/pingcap/tiflow/dm/openapi" pchecker "github.com/pingcap/tiflow/dm/pkg/checker" "github.com/tidbcloud/dataflow-service/internal/constants" "github.com/tidbcloud/dataflow-service/internal/utils" ) //go:generate ../../../tools/bin/goqueryset -in entity.go // gen:qs type ( BillingRecord struct { Base TenantID ID `gorm:"type:BIGINT;not null"` ProjectID ID `gorm:"type:BIGINT;not null"` ClusterID ID `gorm:"type:BIGINT;not null;index:idx_c_c_m,priority:1"` MeasuredTS time.Time `gorm:"type:TIMESTAMP;measured_ts;not null;index:idx_c_c_m,priority:3"` SKU string `gorm:"type:VARCHAR(1024);not null"` Total int64 `gorm:"type:BIGINT;not null"` ExtraLabels BillingRecordExtraLabel `gorm:"type:VARCHAR(1024)"` Category string `gorm:"type:VARCHAR(128);not null;index:idx_c_c_m,priority:2"` Synced bool `gorm:"not null;default:false"` CostID string `gorm:"cost_id;type:VARCHAR(64);not null"` } DBUser struct { Base ClusterID ID `gorm:"type:BIGINT;index;not null"` AuthMethod string `gorm:"type:VARCHAR(255);not null"` DBUserType string `gorm:"type:VARCHAR(255);db_user_type;not null"` DBUserName string `gorm:"type:VARCHAR(255);db_user_name;not null"` DBUserPassword string `gorm:"type:VARCHAR(1024);db_user_password;not null;default:''"` RoleName string `gorm:"type:VARCHAR(64);not null"` Status string `gorm:"type:VARCHAR(64);not null"` } Import struct { Updatable Versioned Name string `gorm:"type:VARCHAR(128)"` OldID *uint64 ClusterID ID `gorm:"index;not null" valid:"required"` ProjectID ID `gorm:"not null"` TenantID ID `gorm:"not null"` Status ImportStatus `gorm:"type:ENUM('unknown','preparing', 'importing','completed','failed', 'canceling', 'canceled');not null;default:'preparing'"` // enum Message string `gorm:"type:TEXT"` // failed message StartedAt *time.Time `gorm:"type:TIMESTAMP"` FinishedAt *time.Time `gorm:"type:TIMESTAMP;index"` Creator string `gorm:"type:VARCHAR(255)"` CleanUp bool `gorm:"index;default:false;not null"` BucketRegion string `gorm:"type:VARCHAR(255)"` AwsRoleArn string `gorm:"type:VARCHAR(255);not null;default:''"` DataFormat DataFormat DataSourceType DataSource // we just pass this to lightning's config. // for the privilege need see https://docs.pingcap.com/tidb/stable/tidb-lightning-faq#what-is-the-privilege-requirements-for-the-target-database TargetUsername string `gorm:"type:VARCHAR(255)"` TargetPassword string `gorm:"type:VARCHAR(1024)"` Filter string `gorm:"type:VARCHAR(255)"` CustomCSVFormat string `gorm:"type:VARCHAR(255)"` OptionalArgs *OptionalArgs `gorm:"type:JSON"` // Source data info TotalSize *uint64 ProcessedDataSize *uint64 TotalFiles *uint32 SourceURL string `gorm:"type:VARCHAR(1024)"` CurrentTables CurrentTables `gorm:"type:JSON"` EndpointServiceName string `gorm:"type:VARCHAR(256);index"` ConnMethod ConnMethod `gorm:"type:enum('direct', 'private-link', 'vpc-peering');default:'direct'"` // In progress info TotalTablesCount uint32 CompletedTables uint32 PendingTables uint32 CompletedPercent uint32 // 0 ~100 PostImportCompletedPercent uint32 // 0 ~100 // Table results // Tables []ImportTable `gorm:"foreignkey:ImportID"` TableResults ImportTables `gorm:"type:JSON"` TaskID int64 LocalImportCfg *LocalImportCfg `gorm:"type:JSON"` TargetTableInfos TargetTableInfos `gorm:"type:JSON"` CreatedByDM bool `gorm:"type:tinyint(1);not null;index"` NeedDropTables bool `gorm:"type:tinyint(1)"` ServerTier string `gorm:"type:VARCHAR(64)"` ClusterVersion string `gorm:"type:VARCHAR(64)"` ComponentVersion string `gorm:"type:VARCHAR(64);not null;index:idx_component_version"` TaskPhase string `gorm:"type:VARCHAR(64)"` Provider string `gorm:"type:VARCHAR(64);not null"` Region string `gorm:"type:VARCHAR(64);not null"` TaskImportingAt *time.Time `gorm:"type:TIMESTAMP"` TaskPostproceAt *time.Time `gorm:"type:TIMESTAMP"` ArchType ArchType `gorm:"type:ENUM('cloudnative','3layer');not null;default:'cloudnative'"` // enum } // Deprecated, no need this table // ImportTable is the entity model recording status for importing a single table. ImportTable struct { Updatable ImportID ID `gorm:"not null;index:idx_import_table,priority:1,unique" valid:"required"` TableName string `gorm:"not null;type:VARCHAR(255);index:idx_import_table,priority:2,unique"` Result ImportTableResult `gorm:"type:ENUM('success','warning','error');not null;default:'success'"` // enum Message string `gorm:"type:TEXT"` ImportStartedAt *time.Time `gorm:"type:TIMESTAMP"` ImportFinishedAt *time.Time `gorm:"type:TIMESTAMP"` PostImportStartedAt *time.Time `gorm:"type:TIMESTAMP"` PostImportFinishedAt *time.Time `gorm:"type:TIMESTAMP"` } // ImportUploadStub is the entity model recording uploaded file information (stub) for an import task. ImportUploadStub struct { Updatable TenantID ID `gorm:"not null"` ProjectID ID `gorm:"not null"` ClusterID ID `gorm:"not null"` LocalFileName string `gorm:"type:VARCHAR(255)"` UploadedFileName string `gorm:"type:VARCHAR(255)"` UploadedURL string `gorm:"type:VARCHAR(255)"` UploadedRegion string `gorm:"type:VARCHAR(255)"` ExpiredTime *time.Time `gorm:"type:TIMESTAMP"` } ImportIntoTask struct { Updatable Versioned Name string `gorm:"type:VARCHAR(128)"` ClusterID ID `gorm:"index;not null" valid:"required"` ProjectID ID TenantID ID `gorm:"not null"` Status ImportIntoStatus `gorm:"type:ENUM('unknown','preparing', 'importing','completed', 'failed', 'canceling', 'canceled', 'scaling');not null;default:'preparing';index"` // enum Message string `gorm:"type:LONGTEXT"` // failed message StartedAt *time.Time `gorm:"type:TIMESTAMP"` FinishedAt *time.Time `gorm:"type:TIMESTAMP;index"` CreatorID uint64 Creator string `gorm:"type:VARCHAR(255)"` TargetUsername string `gorm:"type:VARCHAR(255)"` TargetPassword string `gorm:"type:VARCHAR(1024)"` CustomCSVFormat string `gorm:"type:LONGTEXT"` Options string `gorm:"type:LONGTEXT"` // Source data info TotalSize uint64 TotalFiles uint32 SourceURI string `gorm:"type:LONGTEXT"` ImportResults ImportIntoSubTasks `gorm:"type:JSON"` ServerTier string `gorm:"type:VARCHAR(64)"` ArchType ArchType `gorm:"type:ENUM('cloudnative','3layer');not null;default:'cloudnative'"` GroupKey string `gorm:"type:VARCHAR(191)"` } ImportIntoPrecheckTask struct { Updatable ClusterID ID `gorm:"index;not null"` SourceType string `gorm:"type:VARCHAR(32);not null"` TableMetas ImportIntoPrecheckTableMetas `gorm:"type:JSON"` } // DFJob used to store common info about dataflow engine job. // for each service, such as DM or CDC, there should be another table to describe the service job // for DM, it's dm_jobs DFJob struct { Updatable DeletedAt *time.Time `gorm:"type:timestamp;"` FinishedAt *time.Time `gorm:"type:timestamp;"` TenantID ID `gorm:"type:bigint(20) unsigned;not null;index"` ProjectID ID `gorm:"type:bigint(20) unsigned;not null;index"` ClusterID ID `gorm:"type:bigint(20) unsigned;not null;index"` Name string `gorm:"type:varchar(64);not null;"` // OneTime means the job will finish at some time, not a long-running job // such as full migration job of DM OneTime bool `gorm:"type:tinyint(1);not null;index"` Phase DFJobPhase `gorm:"type:enum('creating', 'created', 'modifying', 'unrecoverable', 'deleting', 'deleted', 'finished', 'finishing');not null;index"` ServiceCode ServiceCode `gorm:"type:enum('dm','cdc');not null;index"` CreatorUID ID `gorm:"type:bigint(20) unsigned;not null;"` Region string `gorm:"type:varchar(64);not null;"` // same as the region of the cluster, for dm its downstream cluster Provider string `gorm:"type:varchar(64);not null;"` // provider of the cluster WorkerVersion string `gorm:"type:varchar(4);not null;"` // worker version, v1=Before DM refactoring, v2=After DM refactoring ErrorMessage string `gorm:"type:varchar(1024);not null;"` // error message when task failed } DFConnProfile struct { Updatable TenantID ID `gorm:"type:bigint(20) unsigned;not null;index"` ProjectID ID `gorm:"type:bigint(20) unsigned;not null;index;"` ClusterID ID `gorm:"type:bigint(20) unsigned;not null;index;"` CreatorUID ID `gorm:"type:bigint(20) unsigned;not null;"` Name string `gorm:"type:varchar(64);not null;"` Type DBType `gorm:"type:enum('mysql', 'aws-aurora-mysql', 'aws-rds-mysql', 'google-cloud-mysql', 'azure-flexible-mysql');not null;"` Version string `gorm:"type:varchar(64);not null;"` Region string `gorm:"type:varchar(64);not null;"` // region of the source database ConnMethod ConnMethod `gorm:"type:enum('direct', 'private-link', 'vpc-peering');not null;index;"` EndpointServiceName string `gorm:"type:varchar(256);"` Host string `gorm:"type:varchar(256);not null;"` Port int32 `gorm:"type:int(11);not null;"` User string `gorm:"type:varchar(64);not null;"` EncryptedPassword string `gorm:"type:varchar(1024);not null;"` SslCa string `gorm:"type:MEDIUMTEXT;not null;"` SslCert string `gorm:"type:MEDIUMTEXT;not null;"` SslKey string `gorm:"type:MEDIUMTEXT;not null;"` SameRegion bool `gorm:"type:tinyint(1);not null;"` // whether tiflow executor is connecting source database via vpc-peering and they are in the same region } DFPrivateLinkEndpoint struct { Updatable DeletedAt *time.Time `gorm:"type:timestamp;"` CreatorUID ID `gorm:"type:bigint(20) unsigned;not null;"` TenantID ID `gorm:"type:bigint(20) unsigned;not null;index"` ProjectID ID `gorm:"type:bigint(20) unsigned;not null;index;"` Provider string `gorm:"type:varchar(64);not null;"` Region string `gorm:"type:varchar(64);not null;"` NetworkNS string `gorm:"type:varchar(256);not null;index:idx_network_service_name;"` NetworkName string `gorm:"type:varchar(256);not null;index:idx_network_service_name;"` ServiceName string `gorm:"type:varchar(256);not null;index:idx_network_service_name;"` RegionalDNS string `gorm:"type:varchar(256);not null;"` ZonalDNS *ZonalDNS `gorm:"type:json;"` Phase PLEndpointPhase `gorm:"type:enum('new', 'pending', 'pending-acceptance', 'available', 'deleted', 'failed');not null;index;"` EndpointID string `gorm:"type:varchar(256);not null;"` ErrMsg string `gorm:"type:text;not null;"` IsRegional bool `gorm:"type:tinyint(1);not null;"` // whether the endpoint CR is at regional API server or infra API server } // DMJob table for dm job. // this table doesn't has a auto-increment id, it shares the same id as df_jobs // we name it 'id' not 'job_id' to make it better to reuse code(Updatable in entity.go) DMJob struct { Updatable DeletedAt *time.Time `gorm:"type:timestamp;"` TenantID ID `gorm:"type:bigint(20) unsigned;not null;"` ProjectID ID `gorm:"type:bigint(20) unsigned;not null;"` ClusterID ID `gorm:"type:bigint(20) unsigned;not null;"` ClusterUser string `gorm:"type:varchar(64);not null;"` EncryptedClusterPassword string `gorm:"type:varchar(1024);not null;"` Mode DMJobMode `gorm:"type:enum('full','incremental','all');not null;index:idx_mode_status_synced"` ShardMode DMShardMode `gorm:"type:enum('','pessimistic','optimistic');not null;"` OnlineDdl bool `gorm:"type:tinyint(1);not null;"` CollationCompatible DMCollationMode `gorm:"type:enum('loose','strict');not null;"` SpecID ID `gorm:"type:bigint(20) unsigned;not null;"` PrecheckID ID `gorm:"type:bigint(20) unsigned;not null;"` // whether status of full migration part has synced into metadb, i.e. dump/load status in dm_job_sources table. FullPartStatusSynced bool `gorm:"type:tinyint(1);not null;index:idx_mode_status_synced"` ImportMode DMImportMode `gorm:"type:enum('','physical','logical');not null;"` ComponentVersion string `gorm:"type:varchar(64);not null;index:idx_component_version"` } // DMJobSource table for dm job and source connection profile relation and some config related to this (dm job, source) combination. // DM support configure route_rules/ba_rules for sources, so this info is bound in this table, but on cloud we don't // support it now. // binlog_* is upstream meta info for MySQL like database for incremental job. need to move it out or use JSON field DMJobSource struct { JobID ID `gorm:"type:bigint(20) unsigned;not null;primaryKey;"` ConnProfileID ID `gorm:"type:bigint(20) unsigned;not null;primaryKey;"` ConnProfile *DFConnProfile `gorm:"foreignkey:ConnProfileID"` Gtid bool `gorm:"type:tinyint(1);not null;"` CaseSensitive bool `gorm:"type:tinyint(1);not null;"` BinlogName string `gorm:"type:varchar(256);not null;"` BinlogPos uint32 `gorm:"type:int(10) unsigned;not null;"` BinlogGtid string `gorm:"type:text;not null;"` RouteRules DMRouteRules `gorm:"type:json;"` BaRules *FilterRules `gorm:"type:json;not null;"` // for full-migration job, we store its status when it's finished DumpStatus *DMDumpStatus `gorm:"type:json;"` LoadStatus *DMLoadStatus `gorm:"type:json;"` SyncStatus *DMSyncStatus `gorm:"type:json;"` DeletedAt *time.Time `gorm:"type:timestamp;"` } DMPrecheck struct { Updatable Status PrecheckStatus `gorm:"type:enum('finished', 'running', 'failed', 'pending', 'canceled');"` ErrMsg string `gorm:"type:text;"` TotalItem int `gorm:"type:int;"` FullInstanceMigrate bool `gorm:"type:tinyint(1);not null;"` TenantID ID `gorm:"type:bigint(20) unsigned;not null;"` ProjectID ID `gorm:"type:bigint(20) unsigned;not null;"` ClusterID ID `gorm:"type:bigint(20) unsigned;not null;"` } DMPrecheckItem struct { Updatable PrecheckID ID `gorm:"type:bigint(20) unsigned;index;"` Desc string `gorm:"type:text;"` // this column stores item type actually Status PrecheckStatus `gorm:"type:enum('finished', 'running', 'failed', 'pending', 'canceled');"` ErrMsg string `gorm:"type:text;"` Results DMPrecheckResult `gorm:"type:JSON;"` } // DMJobSpec dm job spec. // can use below line during test // insert into `dm_job_specs` values(1, now(), now(), 'exclusive-2c2g', 'S1', 'underlying machine is 2c2g', 30, 3000); DMJobSpec struct { Updatable CapacityUnit float64 `gorm:"type:double;not null;unique;"` Name string `gorm:"type:varchar(64);not null;unique;"` // internal name, such as exclusive-1c2g DisplayName string `gorm:"type:varchar(64);not null;"` // use for display, such as S1 Description string `gorm:"type:varchar(512);not null;"` MaxFullPhaseSpeed int `gorm:"type:int;not null;"` // max full migration speed, in unit MB/s MaxIncrPhaseSpeed int `gorm:"type:int;not null;"` // max incremental migration speed, in unit rows/s } DMMetering struct { Updatable Kind MeteringKind `gorm:"type:enum('node', 'input', 'output', 'endpoint');not null;"` Cost float64 `gorm:"type:double unsigned;not null;"` JobID ID `gorm:"type:bigint(20) unsigned;not null;index"` ClusterID ID `gorm:"type:bigint(20) unsigned;not null;"` TenantID ID `gorm:"type:bigint(20) unsigned;not null;"` ProjectID ID `gorm:"type:bigint(20) unsigned;not null;"` StartTS time.Time `gorm:"type:timestamp;not null;index:idx_metering_ts"` EndTS time.Time `gorm:"type:timestamp;not null;index:idx_metering_ts"` CostID string `gorm:"type:text;not null;"` Synced bool `gorm:"type:tinyint(1);not null;index"` } DMJobModifications struct { Updatable JobID ID `gorm:"type:bigint(20) unsigned;index:idx_job_finished"` Type DMJobModifyType `gorm:"type:enum('scale-up-down') not null;"` Param DMJobModifyParam `gorm:"type:json;not null;"` FinishedAt *time.Time `gorm:"type:timestamp;index:idx_job_finished"` OperatorID ID `gorm:"type:bigint(20) unsigned;"` } ) type DMJobModifyParam struct { FromSpecID ID `json:"from_spec_id"` ToSpecID ID `json:"to_spec_id"` } func (o DMJobModifyParam) Value() (driver.Value, error) { return json.Marshal(o) } func (o *DMJobModifyParam) Scan(src interface{}) error { return json.Unmarshal(src.([]byte), o) } type CurrentTable struct { Name string `json:"name"` Size uint64 `json:"size"` Phase string `json:"phase,omitempty"` ProcessedSize uint64 `json:"processedSize,omitempty"` CompletedPercent uint32 `json:"completed_percent"` // 0 ~100 } type CurrentTables struct { Tables []*CurrentTable `json:"tables"` } // Value marshal json func (c CurrentTables) Value() (driver.Value, error) { b, err := json.Marshal(c) return string(b), err } // Scan unmarshal json func (c *CurrentTables) Scan(input interface{}) error { return json.Unmarshal(input.([]byte), c) } type OptionalArgs struct { // deprecated ObjectNamePattern string `json:"object_name_pattern"` TargetTablePattern string `json:"target_table_pattern"` DuplicationHandlingForSQL string `json:"dup_handling_for_sql,omitempty"` SkipChecksum bool `json:"skip_checksum,omitempty"` StoreWriteBWLimit lightningcfg.ByteSize `json:"store-write-bwlimit,omitempty"` } // Value marshal json func (args OptionalArgs) Value() (driver.Value, error) { b, err := json.Marshal(args) return string(b), err } // Scan unmarshal json func (args *OptionalArgs) Scan(input interface{}) error { return json.Unmarshal(input.([]byte), args) } type IDList struct { IDs []ID `json:"ids"` } // Value marshal json func (c IDList) Value() (driver.Value, error) { b, err := json.Marshal(c) return string(b), err } // Scan unmarshal json func (c *IDList) Scan(input interface{}) error { return json.Unmarshal(input.([]byte), c) } func (id ID) Uint64() uint64 { return uint64(id) } type ComponentState struct { ComponentType ComponentName `json:"component_type"` RegionName string `json:"region_name"` NodeProfileName string `json:"node_profile_name"` Replicas uint32 `json:"replicas"` StorageSizeGi uint32 `json:"storage_size_gi"` } type BillingRecordExtraLabel map[string]string func (labels BillingRecordExtraLabel) Value() (driver.Value, error) { b, err := json.Marshal(labels) return string(b), err } func (labels *BillingRecordExtraLabel) Scan(input interface{}) error { return json.Unmarshal(input.([]byte), labels) } type ZonalDNS struct { DNS map[string]string `json:"dns"` } func (z ZonalDNS) Value() (driver.Value, error) { b, err := json.Marshal(z) return string(b), err } func (z *ZonalDNS) Scan(input interface{}) error { return json.Unmarshal(input.([]byte), z) } type DMRouteRules []router.TableRule func (r DMRouteRules) Value() (driver.Value, error) { b, err := json.Marshal(r) return string(b), err } func (r *DMRouteRules) Scan(input interface{}) error { return json.Unmarshal(input.([]byte), r) } type FilterRules filter.Rules func (r FilterRules) Value() (driver.Value, error) { b, err := json.Marshal(r) return string(b), err } func (r *FilterRules) Scan(input interface{}) error { return json.Unmarshal(input.([]byte), r) } type DMPrecheckResult pchecker.Results func (r DMPrecheckResult) Value() (driver.Value, error) { b, err := json.Marshal(r) return string(b), err } func (r *DMPrecheckResult) Scan(input interface{}) error { return json.Unmarshal(input.([]byte), r) } // DMDumpStatus copy of dm worker pb.DumpStatus. type DMDumpStatus struct { BeginTime int64 `json:"begin_time"` TotalTables int64 `json:"totalTables"` CompletedTables float64 `json:"completedTables"` FinishedBytes float64 `json:"finishedBytes"` FinishedRows float64 `json:"finishedRows"` Bps int64 `json:"bsp"` Duration int64 `json:"duration"` Progress string `json:"progress"` ErrorMsg *string `json:"errorMsg"` Stage openapi.TaskStage `json:"stage"` UpdateTime int64 `json:"updateTime"` } func (r DMDumpStatus) Value() (driver.Value, error) { b, err := json.Marshal(r) return string(b), err } func (r *DMDumpStatus) Scan(input interface{}) error { return json.Unmarshal(input.([]byte), r) } func (r *DMDumpStatus) IsFinished() bool { return r != nil && (utils.RemovePercent(r.Progress) == "100.00" || utils.RemovePercent(r.Progress) == "100") && (r.Stage == openapi.TaskStageFinished || r.Stage == "") // r.Stage == "" => compatible with old data } // DMLoadStatus copy of dm worker pb.LoadStatus. type DMLoadStatus struct { BeginTime int64 `json:"begin_time"` FinishedBytes int64 `json:"finishedBytes"` TotalBytes int64 `json:"totalBytes"` Progress string `json:"progress"` MetaBinlog string `json:"metaBinlog"` MetaBinlogGTID string `json:"metaBinlogGTID"` Bps int64 `json:"bps"` Duration int64 `json:"duration"` Stage openapi.TaskStage `json:"stage"` ErrorMsg *string `json:"errorMsg"` UpdateTime int64 `json:"updateTime"` } func (r DMLoadStatus) Value() (driver.Value, error) { b, err := json.Marshal(r) return string(b), err } func (r *DMLoadStatus) Scan(input interface{}) error { return json.Unmarshal(input.([]byte), r) } func (r *DMLoadStatus) IsFinished() bool { return r != nil && (utils.RemovePercent(r.Progress) == "100.00" || utils.RemovePercent(r.Progress) == "100") && (r.Stage == openapi.TaskStageFinished || r.Stage == "") // r.Stage == "" => compatible with old data } type DMSyncStatus struct { BeginTime int64 `json:"beginTime"` Stage openapi.TaskStage `json:"stage"` ErrorMsg *string `json:"errorMsg"` UpdateTime int64 `json:"updateTime"` BinlogType string `json:"binlogType"` // sharding DDL which current is blocking BlockingDdls []string `json:"blockingDdls"` DumpIoTotalBytes uint64 `json:"dumpIoTotalBytes"` IoTotalBytes uint64 `json:"ioTotalBytes"` MasterBinlog string `json:"masterBinlog"` MasterBinlogGtid string `json:"masterBinlogGtid"` RecentTps int64 `json:"recentTps"` SecondsBehindMaster int64 `json:"secondsBehindMaster"` Synced bool `json:"synced"` SyncerBinlog string `json:"syncerBinlog"` SyncerBinlogGtid string `json:"syncerBinlogGtid"` TotalEvents int64 `json:"totalEvents"` TotalTps int64 `json:"totalTps"` } func (r DMSyncStatus) Value() (driver.Value, error) { b, err := json.Marshal(r) return string(b), err } func (r *DMSyncStatus) Scan(input interface{}) error { return json.Unmarshal(input.([]byte), r) } type TargetTableInfos struct { Infos []TargetTableInfo } type TargetTableInfo struct { Schema string `json:"schema"` Table string `json:"table"` CustomFile string `json:"custom_file"` MatchedFile string `json:"matched_file"` FilesWildcard string `json:"files_wildcard"` Database string `json:"database"` } func (t TargetTableInfos) Value() (driver.Value, error) { b, err := json.Marshal(t) return string(b), err } func (t *TargetTableInfos) Scan(input interface{}) error { return json.Unmarshal(input.([]byte), t) } type LocalImportCfg struct { FileName string `json:"file_name"` TargetDb string `json:"target_db"` TargetTable string `json:"target_table"` OriginalFileName string `json:"original_file_name"` TotalSize uint32 `json:"total_size"` } func (c LocalImportCfg) Value() (driver.Value, error) { b, err := json.Marshal(c) return string(b), err } func (c *LocalImportCfg) Scan(input interface{}) error { return json.Unmarshal(input.([]byte), c) } func (i *Import) CleanSourceUrlQuery() error { sourceUrl, err := url.Parse(i.SourceURL) if err != nil { return errors.Annotate(err, "failed to parse source url") } sourceUrl.RawQuery = "" i.SourceURL = sourceUrl.String() return nil } type ImportSourceAccessConfig struct { AWSAccessKeyID string AWSSecretAccessKey string AWSRoleARN string AWSExternalID string Region string GCPCredentialFile string AzblobSASToken string } func (i *Import) SetAccessConfigInSourceURL(accessConfig *ImportSourceAccessConfig) error { if accessConfig == nil { return nil } u, err := url.Parse(i.SourceURL) if err != nil { return errors.Annotate(err, "failed to parse source url") } queryParamVals := u.Query() switch u.Scheme { case "s3": if len(accessConfig.AWSAccessKeyID) > 0 { queryParamVals.Set("access-key", accessConfig.AWSAccessKeyID) } if len(accessConfig.AWSSecretAccessKey) > 0 { queryParamVals.Set("secret-access-key", accessConfig.AWSSecretAccessKey) } if len(accessConfig.AWSRoleARN) > 0 { queryParamVals.Set("role-arn", accessConfig.AWSRoleARN) } if len(accessConfig.AWSExternalID) > 0 { queryParamVals.Set("external-id", accessConfig.AWSExternalID) } if len(accessConfig.Region) > 0 { queryParamVals.Set("region", accessConfig.Region) } case "gs", "gcs": if len(accessConfig.GCPCredentialFile) > 0 { queryParamVals.Set("credentials-file", accessConfig.GCPCredentialFile) } case constants.AzblobScheme: if len(accessConfig.AzblobSASToken) > 0 { queryParamVals.Set("sas-token", accessConfig.AzblobSASToken) } } u.RawQuery = queryParamVals.Encode() i.SourceURL = u.String() return nil } func (i *Import) ParseSourceURL() (string, *ImportSourceAccessConfig, error) { u, err := url.Parse(i.SourceURL) if err != nil { return "", nil, errors.Annotate(err, "failed to parse source url") } queryParamVals := u.Query() accessConfig := new(ImportSourceAccessConfig) switch u.Scheme { case "s3": accessConfig.AWSAccessKeyID = queryParamVals.Get("access-key") accessConfig.AWSSecretAccessKey = queryParamVals.Get("secret-access-key") accessConfig.AWSRoleARN = queryParamVals.Get("role-arn") accessConfig.AWSExternalID = queryParamVals.Get("external-id") accessConfig.Region = queryParamVals.Get("region") case "gs", "gcs": accessConfig.GCPCredentialFile = queryParamVals.Get("credentials-file") case constants.AzblobScheme: if strings.Contains(u.Host, "blob.core.windows.net") { accessConfig.AzblobSASToken = queryParamVals.Get("sas-token") } } u.RawQuery = "" return u.String(), accessConfig, nil } type ImportTables struct { Tables []ImportTable } func (r ImportTables) Value() (driver.Value, error) { b, err := json.Marshal(r) return string(b), err } func (r *ImportTables) Scan(input interface{}) error { return json.Unmarshal(input.([]byte), r) } type ImportIntoSubTask struct { DataBase string `json:"data_base"` TableName string `json:"table_name"` Result ImportTableResult `json:"result"` Message string `json:"message"` ImportStartedAt *time.Time `json:"import_started_at"` ImportFinishedAt *time.Time `json:"import_finished_at"` PostImportStartedAt *time.Time `json:"post_import_started_at"` PostImportFinishedAt *time.Time `json:"post_import_finished_at"` ImportIntoPhase string `json:"import_into_phase"` ImportIntoJobID uint64 `json:"import_into_job_id"` ImportIntoJobStatus string `json:"import_into_job_status"` ImportIntoFileType string `json:"import_into_file_type"` ImportIntoFileCompression string `json:"import_into_file_compression"` ImportIntoSchemaFile string `json:"import_into_schema_file"` ImportIntoFilesWildcard string `json:"import_into_files_wildcard"` ImportIntoTotalSize uint64 `json:"import_into_total_size"` ImportedRows int64 `json:"imported_rows"` // Deprecated, no need this field ExecutedNodes int `json:"executed_nodes"` CurrentStep string `json:"currentStep"` // The total size of the current step to be processed, in a human-readable format. CurrentStepTotalSize string `json:"current_step_total_size"` // The current step processing progress, as a percentage. CurrentStepProgressPercent int32 `json:"current_step_progress_percent"` // The ETA of the current step. CurrentStepETA string `json:"current_step_eta"` // The processed data size of the current step, in a human-readable format. CurrentStepProcessedDataSize string `json:"current_step_processed_data_size"` // The custom columns of the table to be imported. An empty list means matching the columns of the input file(s). Columns []string `json:"columns"` CurrentStepSpeed string `json:"current_step_speed"` } type ImportIntoSubTasks struct { // ControlPlaneFailed indicates the task failed in control-plane/workflow // handling rather than from a terminal IMPORT INTO SQL execution result. ControlPlaneFailed bool `json:"control_plane_failed,omitempty"` SubTasks []ImportIntoSubTask `json:"sub_tasks"` } func (r ImportIntoSubTasks) Value() (driver.Value, error) { b, err := json.Marshal(r) return string(b), err } func (r *ImportIntoSubTasks) Scan(input interface{}) error { return json.Unmarshal(input.([]byte), r) } type ImportIntoPrecheckTableMeta struct { Database string `json:"database"` Table string `json:"table"` FilesWildcard string `json:"files_wildcard"` } type ImportIntoPrecheckTableMetas []ImportIntoPrecheckTableMeta func (r ImportIntoPrecheckTableMetas) Value() (driver.Value, error) { b, err := json.Marshal(r) return string(b), err } func (r *ImportIntoPrecheckTableMetas) Scan(input interface{}) error { return json.Unmarshal(input.([]byte), r) }