From 4b995099d2ca83ab4731de7d6e4c0e83d226b2a8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Toma=C5=BE=20Jerman?= Date: Wed, 18 Mar 2020 18:21:02 +0100 Subject: [PATCH] Add support for migration source joins --- compose/commands/migrator.go | 14 ++++ pkg/migrate/README.adoc | 61 ++++++++++++++ pkg/migrate/join.go | 132 +++++++++++++++++++++++++++++++ pkg/migrate/main.go | 7 ++ pkg/migrate/stream.go | 49 +++++++++++- pkg/migrate/types/migrateable.go | 6 ++ pkg/migrate/types/node.go | 38 +++++++-- 7 files changed, 299 insertions(+), 8 deletions(-) create mode 100644 pkg/migrate/join.go diff --git a/compose/commands/migrator.go b/compose/commands/migrator.go index 225184838..f3bfcf5ef 100644 --- a/compose/commands/migrator.go +++ b/compose/commands/migrator.go @@ -84,6 +84,20 @@ func Migrator() *cobra.Command { mm.Name = name mm.Map = file + mg = migrateableAdd(mg, mm) + } else if strings.HasSuffix(info.Name(), ".join.json") { + file, err := os.Open(path) + if err != nil { + log.Fatal(err) + } + + ext := filepath.Ext(info.Name()) + // @todo improve this!! + name := info.Name()[0 : len(info.Name())-len(ext)-5] + mm := migrateableSource(mg, name) + mm.Name = name + mm.Join = file + mg = migrateableAdd(mg, mm) } return nil diff --git a/pkg/migrate/README.adoc b/pkg/migrate/README.adoc index b70c9e21e..391615900 100644 --- a/pkg/migrate/README.adoc +++ b/pkg/migrate/README.adoc @@ -74,3 +74,64 @@ Currently only simple conditions, such as `type=specialType` are supported. } ] ---- + +== Joining Migration Sources + +An important feature is the system's ability to construct a migration map from multiple migration sources. +For example; we want to populate a `User` module, that includes data from `User.csv` and `SysUser.csv`. + +=== Algorithrm + +* unmarshal the given `.join.json` +* for each migration node that defines a `.join.json`: +** determine all "joined" migration nodes that will be used in this join operation, +** create `{ field: { id: [ value, ... ] } }` object for each base migration node, based on joined nodes, +** when processing the migration node, respect the above mentioned object and include the specified data. + + +=== Example + +.source.join.json + +`.join.json` files tell the system what fields in the base module will be joined with what fields in the sub module. + +The below example instructs, that the current module should be joined with `joinMod` based on the `Id` field. +Same applies for the other entry. + +[source,json] +---- +{ + "joinModField": "joinMod.Id", + "Id": "subMod.baseModId" +} +---- + +.source.map.json +[source,json] +---- +[ + { + "map": [ + { + "from": "Id", + "to": "baseMod.Id" + }, + + { + "from": "baseField1", + "to": "baseMod.baseField1" + }, + + { + "from": "joinModField.field1", + "to": "baseMod.joinedField1" + }, + + { + "from": "Id.name", + "to": "baseMod.reverseJoinNameField" + } + ] + } +] +---- diff --git a/pkg/migrate/join.go b/pkg/migrate/join.go new file mode 100644 index 000000000..07a7e5e5f --- /dev/null +++ b/pkg/migrate/join.go @@ -0,0 +1,132 @@ +package migrate + +import ( + "encoding/csv" + "encoding/json" + "io" + "io/ioutil" + "strings" + + "github.com/cortezaproject/corteza-server/pkg/migrate/types" +) + +type ( + // temporary node structure + node struct { + mg *types.Migrateable + Joined []*types.JoinedNode + } +) + +// Creates JoinNodes for each Migrateable node included in a source join process +// See readme for more +func sourceJoin(mm []types.Migrateable) ([]types.Migrateable, error) { + var rr []*node + joinedNodes := make(map[string]*types.JoinedNode) + + for _, mg := range mm { + ww := mg + node := &node{mg: &ww} + rr = append(rr, node) + + if mg.Join == nil { + continue + } + + // join definition map defines how two sources are joined + var joinDef map[string]string + src, _ := ioutil.ReadAll(mg.Join) + err := json.Unmarshal(src, &joinDef) + if err != nil { + return nil, err + } + + // find all joined nodes for the given migration node + for baseField, condition := range joinDef { + pts := strings.Split(condition, ".") + joinedModule := pts[0] + joinedByField := pts[1] + + for _, m := range mm { + if m.Name == joinedModule { + if _, ok := joinedNodes[joinedModule]; !ok { + ww := m + joinedNodes[joinedModule] = &types.JoinedNode{ + Mg: &ww, + Name: ww.Name, + BaseField: baseField, + JoinField: joinedByField, + } + } + + jn := joinedNodes[joinedModule] + node.Joined = append(node.Joined, jn) + break + } + } + } + } + + // load each joined node's entries + for _, jn := range joinedNodes { + jn.Entries = make(map[string][]*types.JoinEntry) + reader := csv.NewReader(jn.Mg.Source) + + // header + header, err := reader.Read() + if err == io.EOF { + break + } + + for { + record, err := reader.Read() + if err == io.EOF { + break + } + + if err != nil { + return nil, err + } + + ee := make(types.JoinEntry) + var eId string + for i, c := range record { + ee[header[i]] = c + + if header[i] == jn.JoinField { + eId = c + } + } + + if jn.Entries[eId] == nil { + jn.Entries[eId] = make([]*types.JoinEntry, 0) + } + jn.Entries[eId] = append(jn.Entries[eId], &ee) + } + } + + // construct new migration nodes + // they should include context for the join operation + out := make([]types.Migrateable, 0) + for _, r := range rr { + // include only base migration nodes; exclude joined nodes + if _, ok := joinedNodes[r.mg.Name]; !ok { + jn := r.Joined + for _, s := range jn { + // no need for it further + s.Mg = nil + } + mgg := types.Migrateable{ + Name: r.mg.Name, + Header: r.mg.Header, + Path: r.mg.Path, + Source: r.mg.Source, + Map: r.mg.Map, + Joins: jn, + } + out = append(out, mgg) + } + } + + return out, nil +} diff --git a/pkg/migrate/main.go b/pkg/migrate/main.go index d0f18e012..3abf05daa 100644 --- a/pkg/migrate/main.go +++ b/pkg/migrate/main.go @@ -72,6 +72,12 @@ func Migrate(mg []types.Migrateable, ns *cct.Namespace, ctx context.Context) err } } + // Handle source joins + mg, err = sourceJoin(mg) + if err != nil { + return err + } + // 2. prepare and link migration nodes for _, mgR := range mg { ss, err := splitStream(mgR) @@ -110,6 +116,7 @@ func Migrate(mg []types.Migrateable, ns *cct.Namespace, ctx context.Context) err Reader: r, Header: header, Lock: &sync.Mutex{}, + FieldMap: m.FieldMap, } n = mig.AddNode(n) diff --git a/pkg/migrate/stream.go b/pkg/migrate/stream.go index f932960bb..446679ff8 100644 --- a/pkg/migrate/stream.go +++ b/pkg/migrate/stream.go @@ -19,6 +19,9 @@ type ( row []string header []string writer *csv.Writer + + // field: masterID: [value] + joins map[string]map[string][]string } ) @@ -117,6 +120,45 @@ func splitStream(m types.Migrateable) ([]types.Migrateable, error) { } } + // handle joins + if strings.Contains(from, ".") { + pts := strings.Split(from, ".") + baseFIeld := pts[0] + joinedField := pts[1] + + // find slave + var jn *types.JoinedNode + for _, j := range m.Joins { + if j.BaseField == baseFIeld { + jn = j + break + } + } + if jn == nil { + return nil, errors.New("joinedNode.missing " + from) + } + + if bufs[nm].joins == nil { + bufs[nm].joins = make(map[string]map[string][]string) + } + + if bufs[nm].joins[nmF] == nil { + bufs[nm].joins[nmF] = make(map[string][]string) + } + + vals := make([]string, 0) + for _, e := range jn.Entries[record[hMap[baseFIeld]]] { + for k, v := range *e { + if k == joinedField { + vals = append(vals, v) + } + } + } + + bufs[nm].joins[nmF][record[hMap[baseFIeld]]] = vals + from = baseFIeld + } + bufs[nm].row = append(bufs[nm].row, record[hMap[from]]) if i == 0 { bufs[nm].header = append(bufs[nm].header, nmF) @@ -136,9 +178,10 @@ func splitStream(m types.Migrateable) ([]types.Migrateable, error) { // make migrateable nodes from the generated streams for _, v := range bufs { rr = append(rr, types.Migrateable{ - Name: v.name, - Source: v.buffer, - Header: &v.header, + Name: v.name, + Source: v.buffer, + Header: &v.header, + FieldMap: v.joins, }) } diff --git a/pkg/migrate/types/migrateable.go b/pkg/migrate/types/migrateable.go index 81421e088..581f7f29b 100644 --- a/pkg/migrate/types/migrateable.go +++ b/pkg/migrate/types/migrateable.go @@ -18,5 +18,11 @@ type ( Source io.Reader // map is used for stream splitting Map io.Reader + + // join is used for source joining + Join io.Reader + Joins []*JoinedNode + // field: recordID: [value] + FieldMap map[string]map[string][]string } ) diff --git a/pkg/migrate/types/node.go b/pkg/migrate/types/node.go index 1333cf51e..f99311a46 100644 --- a/pkg/migrate/types/node.go +++ b/pkg/migrate/types/node.go @@ -16,6 +16,18 @@ import ( ) type ( + // field: value + JoinEntry map[string]string + + JoinedNode struct { + Mg *Migrateable + Name string + BaseField string + JoinField string + // id: field: value + Entries map[string][]*JoinEntry + } + // graph node Node struct { // unique node name @@ -47,6 +59,9 @@ type ( Visited bool Lock *sync.Mutex + + // field: recordID: [value] + FieldMap map[string]map[string][]string } // map between migrated ID and Corteza ID @@ -157,6 +172,9 @@ func (n *Node) Merge(nn *Node) { if nn.Header != nil { n.Header = nn.Header } + if nn.FieldMap != nil { + n.FieldMap = nn.FieldMap + } } // link the two nodes @@ -389,6 +407,7 @@ func importNodeSource(n *Node, users map[string]uint64, repo repository.RecordRe vals := types.RecordValueSet{} for i, h := range n.Header { + var values []string val := record[i] if sysField(h) { @@ -444,7 +463,10 @@ func importNodeSource(n *Node, users map[string]uint64, repo repository.RecordRe continue } - if f.Options["moduleID"] != nil { + // check if joinable + if _, ok := n.FieldMap[h]; ok { + values = n.FieldMap[h][val] + } else if f.Options["moduleID"] != nil { // spliced nodes should NOT manage their references if !n.spliced { ref, ok := f.Options["moduleID"].(string) @@ -462,24 +484,30 @@ func importNodeSource(n *Node, users map[string]uint64, repo repository.RecordRe continue } } + values = []string{val} } else if f.Kind == "User" { if u, ok := users[val]; ok { val = fmt.Sprint(u) } else { continue } + values = []string{val} } else { val = strings.Map(fixUtf, val) if val == "" { continue } + values = []string{val} } - vals = append(vals, &types.RecordValue{ - Name: h, - Value: val, - }) + for i, v := range values { + vals = append(vals, &types.RecordValue{ + Name: h, + Value: v, + Place: uint(i), + }) + } } }