diff --git a/pkg/migrate/README.adoc b/pkg/migrate/README.adoc index 391615900..451490df6 100644 --- a/pkg/migrate/README.adoc +++ b/pkg/migrate/README.adoc @@ -93,16 +93,15 @@ For example; we want to populate a `User` module, that includes data from `User. .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. +`.join.json` files define how multiple migration nodes should join into a single 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. +The below example instructs, that the current module should be constructed from it self and `subMod`; based on the `SubModRef` and `subMod.Id` relation. +When creating a `.map.json` file, values from the join operation are available under the specified alias (`...->alias`). [source,json] ---- { - "joinModField": "joinMod.Id", - "Id": "subMod.baseModId" + "SubModRef->smod": "subMod.Id" } ---- @@ -123,13 +122,8 @@ Same applies for the other entry. }, { - "from": "joinModField.field1", - "to": "baseMod.joinedField1" - }, - - { - "from": "Id.name", - "to": "baseMod.reverseJoinNameField" + "from": "smod.field1", + "to": "baseMod.SubModField1" } ] } diff --git a/pkg/migrate/join.go b/pkg/migrate/join.go index 07a7e5e5f..cc799b2c2 100644 --- a/pkg/migrate/join.go +++ b/pkg/migrate/join.go @@ -3,6 +3,7 @@ package migrate import ( "encoding/csv" "encoding/json" + "errors" "io" "io/ioutil" "strings" @@ -11,10 +12,24 @@ import ( ) type ( - // temporary node structure + // mapLink helps us keep track between base nodes, joined nodes and the fields used + // for creating the link + mapLink struct { + jn *types.JoinedNode + // field from the base node used in the op. + baseField string + // alias to use for the base field; allows us to use the same field multiple times + baseFieldAlias string + // field from the joined node to use in the opp. + joinField string + } + + // temporary node for the join op. node struct { - mg *types.Migrateable - Joined []*types.JoinedNode + mg *types.Migrateable + // temporary migration node mapper based on aliases + mapper map[string]mapLink + aliasMap map[string]string } ) @@ -24,15 +39,26 @@ func sourceJoin(mm []types.Migrateable) ([]types.Migrateable, error) { var rr []*node joinedNodes := make(map[string]*types.JoinedNode) + // Algorithm outline: + // 1. determine all migration nodes that will be used as joined nodes + // 2. load entries for each join node + // 3. construct new output migration nodes + + // 1. determination for _, mg := range mm { + // this helps us avoid pesky pointer issues :) ww := mg - node := &node{mg: &ww} - rr = append(rr, node) + nd := &node{mg: &ww} + rr = append(rr, nd) if mg.Join == nil { continue } + // defer this, so we can do simple nil checks + nd.mapper = make(map[string]mapLink) + nd.aliasMap = make(map[string]string) + // join definition map defines how two sources are joined var joinDef map[string]string src, _ := ioutil.ReadAll(mg.Join) @@ -41,35 +67,49 @@ func sourceJoin(mm []types.Migrateable) ([]types.Migrateable, error) { 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] + // find all joined nodes for the given base node + for base, condition := range joinDef { + ptsB := strings.Split(base, "->") + baseField := ptsB[0] + baseFieldAlias := ptsB[1] + if _, ok := nd.aliasMap[baseFieldAlias]; ok { + return nil, errors.New("alias.used " + nd.mg.Name + " " + baseFieldAlias) + } + nd.aliasMap[baseFieldAlias] = baseField + + ptsC := strings.Split(condition, ".") + joinedModule := ptsC[0] + joinedField := ptsC[1] + + // register migration node as join node 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, + Mg: &ww, + Name: ww.Name, } } + // create a link between the base and joined node jn := joinedNodes[joinedModule] - node.Joined = append(node.Joined, jn) + nd.mapper[baseFieldAlias] = mapLink{ + jn: jn, + baseField: baseField, + baseFieldAlias: baseFieldAlias, + joinField: joinedField, + } break } } } } - // load each joined node's entries + // 2. load entries for _, jn := range joinedNodes { - jn.Entries = make(map[string][]*types.JoinEntry) + jn.Entries = make([]map[string]string, 0) reader := csv.NewReader(jn.Mg.Source) // header @@ -88,44 +128,56 @@ func sourceJoin(mm []types.Migrateable) ([]types.Migrateable, error) { return nil, err } - ee := make(types.JoinEntry) - var eId string + row := make(map[string]string) + jn.Entries = append(jn.Entries, row) for i, c := range record { - ee[header[i]] = c - - if header[i] == jn.JoinField { - eId = c - } + row[header[i]] = 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 + // 3. output 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 + for _, nd := range rr { + fnd := false + for _, jn := range joinedNodes { + if nd.mg.Name == jn.Name { + fnd = true + break } - 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) } + + // skip joined nodes + if fnd { + continue + } + + o := nd.mg + // skip nodes with no mappings + if nd.mapper == nil { + out = append(out, *o) + continue + } + + o.AliasMap = nd.aliasMap + + // create `alias.ID`: []entry mappings, that will be used when migrating + for alias, link := range nd.mapper { + if o.FieldMap == nil { + o.FieldMap = make(map[string]types.JoinedNodeRecords) + } + + for _, e := range link.jn.Entries { + kk := alias + "." + e[link.joinField] + if _, ok := o.FieldMap[kk]; !ok { + o.FieldMap[kk] = make(types.JoinedNodeRecords, 0) + } + + o.FieldMap[kk] = append(o.FieldMap[kk], e) + } + } + + out = append(out, *o) } return out, nil diff --git a/pkg/migrate/stream.go b/pkg/migrate/stream.go index 767276112..c6c4bc59e 100644 --- a/pkg/migrate/stream.go +++ b/pkg/migrate/stream.go @@ -122,56 +122,34 @@ func splitStream(m types.Migrateable) ([]types.Migrateable, error) { } } + val := record[hMap[from]] + // handle joins if strings.Contains(from, ".") { + // construct a `alias.joinOnID` value, so we can perform a simple map lookup pts := strings.Split(from, ".") - baseFIeld := pts[0] - joinedField := pts[1] + baseFieldAlias := pts[0] + originalOn := m.AliasMap[baseFieldAlias] + joinField := pts[1] + val = baseFieldAlias + "." + record[hMap[originalOn]] - // 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 + // modify header field to specify what joined node field to use + nmF += ":" + joinField } + bufs[nm].row = append(bufs[nm].row, val) if !bufs[nm].hasHeader { bufs[nm].header = append(bufs[nm].header, nmF) } } - } - // write csv rows - for _, v := range bufs { - v.writer.Write(v.row) - var nn []string - v.row = nn + // write csv rows + for _, v := range bufs { + v.writer.Write(v.row) + var nn []string + v.row = nn + } + break } } } @@ -182,7 +160,8 @@ func splitStream(m types.Migrateable) ([]types.Migrateable, error) { Name: v.name, Source: v.buffer, Header: &v.header, - FieldMap: v.joins, + FieldMap: m.FieldMap, + AliasMap: m.AliasMap, }) } diff --git a/pkg/migrate/types/migrateable.go b/pkg/migrate/types/migrateable.go index 581f7f29b..5dde2bb39 100644 --- a/pkg/migrate/types/migrateable.go +++ b/pkg/migrate/types/migrateable.go @@ -9,6 +9,9 @@ var ( ) type ( + JoinedNodeEntry map[string]string + JoinedNodeRecords []JoinedNodeEntry + Migrateable struct { Name string Path string @@ -22,7 +25,9 @@ type ( // join is used for source joining Join io.Reader Joins []*JoinedNode - // field: recordID: [value] - FieldMap map[string]map[string][]string + // alias.ID: [value] + FieldMap map[string]JoinedNodeRecords + // helps us determine what value field to use for linking + AliasMap map[string]string } ) diff --git a/pkg/migrate/types/node.go b/pkg/migrate/types/node.go index f99311a46..17785228f 100644 --- a/pkg/migrate/types/node.go +++ b/pkg/migrate/types/node.go @@ -16,16 +16,11 @@ import ( ) type ( - // field: value - JoinEntry map[string]string - JoinedNode struct { - Mg *Migrateable - Name string - BaseField string - JoinField string + Mg *Migrateable + Name string // id: field: value - Entries map[string][]*JoinEntry + Entries []map[string]string } // graph node @@ -61,7 +56,7 @@ type ( Lock *sync.Mutex // field: recordID: [value] - FieldMap map[string]map[string][]string + FieldMap map[string]JoinedNodeRecords } // map between migrated ID and Corteza ID @@ -70,6 +65,7 @@ type ( PostProc struct { Leafs []*Node Err error + Node *Node } ) @@ -113,6 +109,7 @@ func (n *Node) Migrate(repoRecord repository.RecordRepository, users map[string] ch <- PostProc{ Leafs: nil, Err: err, + Node: n, } return } @@ -122,6 +119,7 @@ func (n *Node) Migrate(repoRecord repository.RecordRepository, users map[string] ch <- PostProc{ Leafs: nil, Err: err, + Node: n, } return } @@ -451,6 +449,13 @@ func importNodeSource(n *Node, users map[string]uint64, repo repository.RecordRe break } } else { + joined := "" + if strings.Contains(h, ":") { + pts := strings.Split(h, ":") + h = pts[0] + joined = pts[1] + } + var f *types.ModuleField for _, ff := range n.Module.Fields { if ff.Name == h { @@ -463,42 +468,54 @@ func importNodeSource(n *Node, users map[string]uint64, repo repository.RecordRe continue } - // 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) - if !ok { - return nil, errors.New("moduleField.record.invalidRefFormat") - } + // simple hack to fully support multiple values from a joined node + vvs := make([]string, 0) - if mod, ok := n.mapping[ref]; ok { - if v, ok := mod[val]; ok { - val = v + // check if joinable + if joined != "" { + tmp := n.FieldMap[val] + for _, e := range tmp { + vvs = append(vvs, e[joined]) + } + } else { + vvs = []string{val} + } + + for _, val := range vvs { + if f.Options["moduleID"] != nil { + // spliced nodes should NOT manage their references + if !n.spliced { + ref, ok := f.Options["moduleID"].(string) + if !ok { + return nil, errors.New("moduleField.record.invalidRefFormat") + } + + if mod, ok := n.mapping[ref]; ok && val != "" { + if v, ok := mod[val]; ok && v != "" { + val = v + } else { + continue + } } else { continue } + } + values = append(values, val) + } else if f.Kind == "User" { + if u, ok := users[val]; ok { + val = fmt.Sprint(u) } else { continue } - } - values = []string{val} - } else if f.Kind == "User" { - if u, ok := users[val]; ok { - val = fmt.Sprint(u) + values = append(values, val) } else { - continue - } - values = []string{val} - } else { - val = strings.Map(fixUtf, val) + val = strings.Map(fixUtf, val) - if val == "" { - continue + if val == "" { + continue + } + values = append(values, val) } - values = []string{val} } for i, v := range values {