DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Add support for migration source joins

Browse Source
This commit is contained in:
Tomaž Jerman
2020-03-18 18:21:02 +01:00
parent 53288a00cd
commit 4b995099d2
7 changed files with 299 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
compose/commands/migrator.go
+14
View File
@@ -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
pkg/migrate/README.adoc
+61
View File
@@ -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"
}
]
}
]
----
pkg/migrate/join.go
+132
View File
@@ -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
}
pkg/migrate/main.go
+7
View File
@@ -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)
pkg/migrate/stream.go
+46 -3
View File
@@ -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,
})
}
pkg/migrate/types/migrateable.go
+6
View File
@@ -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
}
)
pkg/migrate/types/node.go
+33 -5
View File
@@ -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),
})
}
}
}