floraachy
/
console
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[migration] cleanup unused target config

This commit is contained in:
Kassian Sun 2023-04-17 12:41:34 +08:00
parent 81c9df8ef4
commit eced674817
2 changed files with 1 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
plugin/migration/model.go
View File

@ -153,7 +153,6 @@ type IndexMigrationBulkConfig struct {
type IndexMigrationTargetConfig struct { type IndexMigrationTargetConfig struct {
ClusterId string `json:"cluster_id"` ClusterId string `json:"cluster_id"`
Bulk IndexMigrationBulkConfig `json:"bulk"` Bulk IndexMigrationBulkConfig `json:"bulk"`
QueryDSL util.MapStr `json:"query_dsl,omitempty"`
} }
type PipelineTaskLoggingConfig struct { type PipelineTaskLoggingConfig struct {

31
plugin/migration/pipeline.go
View File

@ -1014,34 +1014,6 @@ func (p *DispatcherProcessor) splitMajorMigrationTask(taskItem *task2.Task) erro
partitionSource.PartitionId = partitionID partitionSource.PartitionId = partitionID
partitionSource.QueryDSL = partition.Filter partitionSource.QueryDSL = partition.Filter
partitionSource.QueryString = "" partitionSource.QueryString = ""
var must []interface{}
if partition.Other {
must = append(must, partition.Filter)
} else {
must = append(must, util.MapStr{
"range": util.MapStr{
index.Partition.FieldName: util.MapStr{
"gte": partition.Start,
"lt": partition.End,
},
},
})
}
if targetMust != nil {
must = append(must, targetMust...)
}
partitionTarget := target
if len(must) > 0 {
partitionTarget.QueryDSL = util.MapStr{
"query": util.MapStr{
"bool": util.MapStr{
"must": must,
},
},
}
}
partitionMigrationTask := task2.Task{ partitionMigrationTask := task2.Task{
ParentId: []string{taskItem.ID}, ParentId: []string{taskItem.ID},
@ -1060,13 +1032,12 @@ func (p *DispatcherProcessor) splitMajorMigrationTask(taskItem *task2.Task) erro
}, },
ConfigString: util.MustToJSON(IndexMigrationTaskConfig{ ConfigString: util.MustToJSON(IndexMigrationTaskConfig{
Source: partitionSource, Source: partitionSource,
Target: partitionTarget, Target: target,
Execution: clusterMigrationTask.Settings.Execution, Execution: clusterMigrationTask.Settings.Execution,
}), }),
} }
partitionMigrationTask.ID = util.GetUUID() partitionMigrationTask.ID = util.GetUUID()
err = orm.Create(nil, &partitionMigrationTask) err = orm.Create(nil, &partitionMigrationTask)
target.QueryDSL = nil
if err != nil { if err != nil {
return fmt.Errorf("store index migration task(partition) error: %w", err) return fmt.Errorf("store index migration task(partition) error: %w", err)
} }