diff --git a/server/compose/envoy/record_datasource.go b/server/compose/envoy/record_datasource.go index 9193cc7f1..db555fc64 100644 --- a/server/compose/envoy/record_datasource.go +++ b/server/compose/envoy/record_datasource.go @@ -17,6 +17,8 @@ type ( Mapping envoyx.DatasourceMapping Provider envoyx.Provider + CheckExisting func(ctx context.Context, ref ...[]string) ([]uint64, error) + currentIndex int // Reusable buffer for reading records @@ -25,6 +27,8 @@ type ( // Index to map from ref to ID // @todo we might need to flush these to the disc in case a huge dataset is passed in refToID map[string]uint64 + // @todo might be worth putting both into one map; not sure how much space we'd save up + existingIDs map[uint64]bool } // iteratorProvider is a wrapper around the dal.Iterator to conform to the diff --git a/server/compose/envoy/store_decode.go b/server/compose/envoy/store_decode.go index 7251ab2e2..38ee7ee37 100644 --- a/server/compose/envoy/store_decode.go +++ b/server/compose/envoy/store_decode.go @@ -240,8 +240,9 @@ func (d StoreDecoder) decodeRecordDatasource(ctx context.Context, s store.Storer } ou := &RecordDatasource{ - Provider: &iteratorProvider{iter: iter}, - refToID: make(map[string]uint64), + Provider: &iteratorProvider{iter: iter}, + refToID: make(map[string]uint64), + existingIDs: make(map[uint64]bool), // @todo consider providing defaults from the outside Mapping: envoyx.DatasourceMapping{ KeyField: []string{"id"}, diff --git a/server/compose/envoy/store_encode.go b/server/compose/envoy/store_encode.go index 7f0b14f76..3e9fddba8 100644 --- a/server/compose/envoy/store_encode.go +++ b/server/compose/envoy/store_encode.go @@ -153,9 +153,14 @@ func (e StoreEncoder) validateRecord(*types.Record) (err error) { } func (e StoreEncoder) prepare(ctx context.Context, p envoyx.EncodeParams, s store.Storer, rt string, nn envoyx.NodeSet) (err error) { + dl, err := e.grabDal(p) + if err != nil { + return + } + switch rt { case ComposeRecordDatasourceAuxType: - return e.prepareRecordDatasource(ctx, p, s, nn) + return e.prepareRecordDatasource(ctx, p, s, dl, nn) } return diff --git a/server/compose/envoy/store_encode_datasource.go b/server/compose/envoy/store_encode_datasource.go index 89203c776..7146ac7d7 100644 --- a/server/compose/envoy/store_encode_datasource.go +++ b/server/compose/envoy/store_encode_datasource.go @@ -28,7 +28,7 @@ var ( rvFormatter = values.Formatter() ) -func (e StoreEncoder) prepareRecordDatasource(ctx context.Context, p envoyx.EncodeParams, s store.Storer, nn envoyx.NodeSet) (err error) { +func (e StoreEncoder) prepareRecordDatasource(ctx context.Context, p envoyx.EncodeParams, s store.Storer, dl dal.FullService, nn envoyx.NodeSet) (err error) { // @todo match existing records; for now use just the ID like V1 for _, n := range nn { @@ -41,7 +41,7 @@ func (e StoreEncoder) prepareRecordDatasource(ctx context.Context, p envoyx.Enco panic("unexpected datasource type: node expecting type of RecordDatasource") } - err = e.prepareRecords(ctx, p, s, ds) + err = e.prepareRecords(ctx, p, s, dl, ds, nn) if err != nil { return } @@ -54,7 +54,7 @@ func (e StoreEncoder) prepareRecordDatasource(ctx context.Context, p envoyx.Enco return } -func (e StoreEncoder) prepareRecords(ctx context.Context, p envoyx.EncodeParams, s store.Storer, ds *RecordDatasource) (err error) { +func (e StoreEncoder) prepareRecords(ctx context.Context, p envoyx.EncodeParams, s store.Storer, dl dal.FullService, ds *RecordDatasource, nn envoyx.NodeSet) (err error) { var ( aux = make(map[string]string) more bool @@ -62,6 +62,15 @@ func (e StoreEncoder) prepareRecords(ctx context.Context, p envoyx.EncodeParams, ) ds.refToID = make(map[string]uint64) + ds.existingIDs = make(map[uint64]bool) + + // Just so we don't need to do a branch later down the line + cex := ds.CheckExisting + if cex == nil || len(ds.Mapping.KeyField) == 0 { + cex = func(ctx context.Context, ref ...[]string) ([]uint64, error) { + return make([]uint64, len(ref)), nil + } + } for { ident, more, err = ds.Next(ctx, aux) @@ -69,7 +78,21 @@ func (e StoreEncoder) prepareRecords(ctx context.Context, p envoyx.EncodeParams, return } - ds.AddRef(id.Next(), ident...) + // @todo we'll need to batch these up + existing, err := cex(ctx, ident) + if err != nil { + return err + } + + if existing[0] != 0 { + ds.existingIDs[existing[0]] = true + } + + if existing[0] == 0 { + existing[0] = id.Next() + } + + ds.AddRef(existing[0], ident...) // Construct a simple record for basic validation/preprocessing rec := types.Record{} @@ -143,7 +166,8 @@ func (e StoreEncoder) encodeRecordDatasource(ctx context.Context, p envoyx.Encod var ( rec types.Record - records types.RecordSet + creates types.RecordSet + updates types.RecordSet rve *types.RecordValueErrorSet ) for { @@ -185,15 +209,29 @@ func (e StoreEncoder) encodeRecordDatasource(ctx context.Context, p envoyx.Encod } ax := rec - records = append(records, &ax) - if len(records) > recordBatchMaxChunk { - err = dalutils.ComposeRecordCreate(ctx, dl, mod, records...) + if ds.existingIDs[rec.ID] { + updates = append(updates, &ax) + } else { + creates = append(creates, &ax) + } + + if len(creates) > recordBatchMaxChunk { + err = dalutils.ComposeRecordCreate(ctx, dl, mod, creates...) if err != nil { return } - records = make(types.RecordSet, 0, recordBatchMaxChunk/2) + creates = make(types.RecordSet, 0, recordBatchMaxChunk/2) + } + + if len(updates) > recordBatchMaxChunk { + err = dalutils.ComposeRecordUpdate(ctx, dl, mod, updates...) + if err != nil { + return + } + + updates = make(types.RecordSet, 0, recordBatchMaxChunk/2) } return }() @@ -213,8 +251,12 @@ func (e StoreEncoder) encodeRecordDatasource(ctx context.Context, p envoyx.Encod } } - if len(records) > 0 { - err = dalutils.ComposeRecordCreate(ctx, dl, mod, records...) + if len(creates) > 0 { + err = dalutils.ComposeRecordCreate(ctx, dl, mod, creates...) + } + + if len(updates) > 0 { + err = dalutils.ComposeRecordUpdate(ctx, dl, mod, updates...) } return } diff --git a/server/compose/rest/record.go b/server/compose/rest/record.go index 786b11467..52e84b435 100644 --- a/server/compose/rest/record.go +++ b/server/compose/rest/record.go @@ -11,6 +11,7 @@ import ( "strings" "time" + "github.com/cortezaproject/corteza/server/compose/dalutils" composeEnvoy "github.com/cortezaproject/corteza/server/compose/envoy" "github.com/cortezaproject/corteza/server/compose/rest/request" "github.com/cortezaproject/corteza/server/compose/service" @@ -26,6 +27,7 @@ import ( "github.com/cortezaproject/corteza/server/pkg/filter" "github.com/cortezaproject/corteza/server/pkg/revisions" "github.com/cortezaproject/corteza/server/store" + "github.com/spf13/cast" ) type ( @@ -530,6 +532,17 @@ func (ctrl *Record) ImportRun(ctx context.Context, r *request.RecordImportRun) ( Placeholder: true, } + keyField := []string{} + + // Check if we're mapping the ID field even + for _, v := range importSession.Fields { + auxf := strings.ToLower(v) + + if auxf == "id" || auxf == "recordid" || auxf == "record_id" { + keyField = []string{importSession.Key} + } + } + node = &envoyx.Node{ Datasource: &composeEnvoy.RecordDatasource{ Mapping: envoyx.DatasourceMapping{ @@ -537,10 +550,62 @@ func (ctrl *Record) ImportRun(ctx context.Context, r *request.RecordImportRun) ( References: map[string]string{}, Scope: map[string]string{}, Defaultable: false, + KeyField: keyField, Mapping: envoyx.FieldMapping{ Map: fieldMapping, }, }, + + CheckExisting: func(ctx context.Context, idents ...[]string) (out []uint64, err error) { + qp := make([]string, 0, len(idents)) + for _, ident := range idents { + if len(ident) != 1 { + continue + } + + rid := cast.ToUint64(ident[0]) + if rid == 0 { + continue + } + + qp = append(qp, fmt.Sprintf("recordID='%d'", rid)) + } + + bong, _, err := dalutils.ComposeRecordsList(ctx, dal.Service(), mod, types.RecordFilter{ + ModuleID: mod.ID, + NamespaceID: mod.NamespaceID, + Query: strings.Join(qp, " OR "), + }) + if err != nil { + return + } + + for _, ident := range idents { + if len(ident) != 1 { + out = append(out, 0) + continue + } + + rid := cast.ToUint64(ident[0]) + if rid == 0 { + out = append(out, 0) + continue + } + + // Find the correct one in the fetched slice + got := uint64(0) + for _, r := range bong { + if r.ID == rid { + got = r.ID + break + } + } + + out = append(out, got) + } + + return + }, }, ResourceType: composeEnvoy.ComposeRecordDatasourceAuxType, Identifiers: envoyx.MakeIdentifiers(importSession.Name), @@ -573,6 +638,8 @@ func (ctrl *Record) ImportRun(ctx context.Context, r *request.RecordImportRun) ( return } + // panic("AAAAAA") + { // @todo this is temporary because the service's logic is a bit flawed for this case err = storeEncoder.Prepare(ctx, encodeParams, composeEnvoy.ComposeRecordDatasourceAuxType, envoyx.NodeSet{node}) diff --git a/server/compose/service/import_session.go b/server/compose/service/import_session.go index ac2a6e3bb..a824a4a63 100644 --- a/server/compose/service/import_session.go +++ b/server/compose/service/import_session.go @@ -101,7 +101,7 @@ func (svc *importSession) Create(ctx context.Context, f io.ReadSeeker, name, con // @todo improve this bit k := prepKey(f) - if k == "id" || k == "recordid" { + if k == "id" || k == "recordid" || k == "record_id" { sh.Key = f } }