DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Improve migration source joining system

Browse Source
This commit is contained in:
Tomaž Jerman
2020-03-23 18:11:54 +01:00
parent db90a03c01
commit 457d5a1561
5 changed files with 183 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/migrate/README.adoc
+6 -12
View File
@@ -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"
}
]
}
pkg/migrate/join.go
+99 -47
View File
@@ -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
pkg/migrate/stream.go
+19 -40
View File
@@ -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,
})
}
pkg/migrate/types/migrateable.go
+7 -2
View File
@@ -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
}
)
pkg/migrate/types/node.go
+52 -35
View File
@@ -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 {