diff --git a/internal/core/asset/export_test.go b/internal/core/asset/export_test.go new file mode 100644 index 00000000..8e10b727 --- /dev/null +++ b/internal/core/asset/export_test.go @@ -0,0 +1,8 @@ +package asset + +// The tests that need a database live in asset_test: pgtest reaches this package through the +// plugin registry, so importing it from here would close a cycle. +var ( + MustLoadProfile = mustLoadProfile + ValidCreate = validCreate +) diff --git a/internal/core/asset/links.go b/internal/core/asset/links.go index 78da99fb..a0fdc69d 100644 --- a/internal/core/asset/links.go +++ b/internal/core/asset/links.go @@ -2,6 +2,7 @@ package asset import ( "context" + "errors" "fmt" "slices" "strings" @@ -61,6 +62,81 @@ func LinkIDs(value any) []string { return nil } +// mrnPrefix marks a link written by something that only knows the target's MRN, such as a +// connector: IDs are minted here, so a source cannot name them. +const mrnPrefix = "mrn://" + +// HasLinkMRNs reports whether metadata holds an MRN anywhere a link could be. +func HasLinkMRNs(value any) bool { + switch v := value.(type) { + case string: + return strings.HasPrefix(v, mrnPrefix) + case []any: + return slices.ContainsFunc(v, HasLinkMRNs) + case []string: + return slices.ContainsFunc(v, func(item string) bool { return strings.HasPrefix(item, mrnPrefix) }) + case map[string]any: + for _, item := range v { + if HasLinkMRNs(item) { + return true + } + } + } + return false +} + +// resolveLinkMRNs replaces, in the asset-control fields, each MRN with the ID of the asset it +// names, on a copy of the metadata. An MRN that names nothing is left out; with deferred, its whole field is, so that a run +// which creates both ends can fill the field in one go on its second pass instead of leaving a +// partial value that would then count as curated. +func (s *service) resolveLinkMRNs(ctx context.Context, asset *Asset, deferred bool) error { + if !HasLinkMRNs(asset.Metadata) { + return nil + } + // The caller keeps its map as it sent it: a run sends the same metadata again on its second pass. + metadata, err := cloneMetadata(asset.Metadata) + if err != nil { + return err + } + asset.Metadata = metadata + for _, f := range LinkFields(s.registry()) { + value, present := metamodel.ValueAt(metadata, f.Storage) + if !present || !HasLinkMRNs(value) { + continue + } + raw := LinkIDs(value) + ids := make([]any, 0, len(raw)) + missing := false + for _, item := range raw { + if !strings.HasPrefix(item, mrnPrefix) { + ids = append(ids, item) + continue + } + target, err := s.repo.GetByMRN(ctx, item) + if errors.Is(err, ErrNotFound) || errors.Is(err, ErrAssetNotFound) { + missing = true + continue + } + if err != nil { + return fmt.Errorf("resolving linked asset %s: %w", item, err) + } + ids = append(ids, target.ID) + } + var resolved any = ids + if _, single := value.(string); single && len(ids) > 0 { + resolved = ids[0] + } + if len(ids) == 0 || (missing && deferred) { + resolved = nil + } + parts := strings.Split(strings.TrimPrefix(f.Storage, "metadata."), ".") + if err := setMetadataValue(metadata, parts, resolved); err != nil { + return fmt.Errorf("field %q: %w", f.ID, err) + } + } + return nil +} + // checkLinks rejects asset-control values naming the asset itself, or adding an asset that does // not exist or whose asset_type the field's targetAssetTypes leaves out. Links the asset already had are kept even once their target is deleted, so an // unrelated edit or a re-ingest never fails on them. diff --git a/internal/core/asset/links_test.go b/internal/core/asset/links_test.go index f927a91d..5f84aae3 100644 --- a/internal/core/asset/links_test.go +++ b/internal/core/asset/links_test.go @@ -145,3 +145,41 @@ fields: } } } + +func TestAnMRNLinkResolvesOnceItsTargetExists(t *testing.T) { + svc := newLinkService(t) + ctx := context.Background() + first := validCreate("source") + mrn := "mrn://table/test/source" + first.MRN = &mrn + first.Metadata["example"] = map[string]any{"applies_to": []any{"mrn://table/test/later"}} + first.DeferLinks = true + source, err := svc.Create(ctx, first) + if err != nil { + t.Fatal(err) + } + if value, present := metamodel.ValueAt(source.Metadata, "metadata.example.applies_to"); present { + t.Fatalf("a deferred field with an MRN that names nothing yet is not stored: %v", value) + } + later := validCreate("later") + laterMRN := "mrn://table/test/later" + later.MRN = &laterMRN + target, err := svc.Create(ctx, later) + if err != nil { + t.Fatal(err) + } + updated, err := svc.Update(ctx, source.ID, UpdateInput{ + Metadata: map[string]any{"example": map[string]any{"applies_to": []any{"mrn://table/test/later"}}}, + FromSync: true, + }) + if err != nil { + t.Fatal(err) + } + value, _ := metamodel.ValueAt(updated.Metadata, "metadata.example.applies_to") + if got := LinkIDs(value); len(got) != 1 || got[0] != target.ID { + t.Fatalf("the link must hold the target's ID once it exists: %v", got) + } + if !HasLinkMRNs(map[string]any{"a": []any{"x", "mrn://t/p/n"}}) || HasLinkMRNs(map[string]any{"a": "x"}) { + t.Fatal("HasLinkMRNs finds an MRN anywhere in the metadata") + } +} diff --git a/internal/core/asset/metamodel_postgres_test.go b/internal/core/asset/metamodel_postgres_test.go index 5210e217..df53691e 100644 --- a/internal/core/asset/metamodel_postgres_test.go +++ b/internal/core/asset/metamodel_postgres_test.go @@ -1,4 +1,4 @@ -package asset +package asset_test import ( "context" @@ -6,6 +6,7 @@ import ( "testing" "time" + "github.com/marmotdata/marmot/internal/core/asset" "github.com/marmotdata/marmot/internal/core/metamodel" "github.com/marmotdata/marmot/internal/metrics" "github.com/marmotdata/marmot/internal/store/postgres/pgtest" @@ -18,9 +19,9 @@ func (dbRecorder) RecordDBQuery(context.Context, string, time.Duration, bool) {} func TestMetamodelPostgresWrites(t *testing.T) { pool := pgtest.TempDB(t) ctx := context.Background() - repo := NewPostgresRepository(pool, dbRecorder{}) - svc := NewService(repo, WithMetamodel(mustLoadProfile(t))) - input := validCreate("postgres-metamodel") + repo := asset.NewPostgresRepository(pool, dbRecorder{}) + svc := asset.NewService(repo, asset.WithMetamodel(asset.MustLoadProfile(t))) + input := asset.ValidCreate("postgres-metamodel") if err := pool.QueryRow(ctx, "SELECT id FROM users WHERE username = 'admin'").Scan(&input.CreatedBy); err != nil { t.Fatal(err) } @@ -46,7 +47,7 @@ func TestMetamodelPostgresWrites(t *testing.T) { if _, err := svc.PatchFields(ctx, created.ID, 2, map[string]any{"retention": -1}); err == nil { t.Fatal("invalid value persisted") } - if _, err := svc.PatchFields(ctx, created.ID, 1, map[string]any{"retention": 1}); !errors.Is(err, ErrVersionConflict) { + if _, err := svc.PatchFields(ctx, created.ID, 1, map[string]any{"retention": 1}); !errors.Is(err, asset.ErrVersionConflict) { t.Fatalf("stale patch: %v", err) } @@ -57,7 +58,7 @@ func TestMetamodelPostgresWrites(t *testing.T) { } start := make(chan struct{}) results := make(chan error, 2) - for _, a := range []*Asset{stored, other} { + for _, a := range []*asset.Asset{stored, other} { go func() { <-start; results <- repo.Update(ctx, a) }() } close(start) @@ -67,7 +68,7 @@ func TestMetamodelPostgresWrites(t *testing.T) { switch { case err == nil: successes++ - case errors.Is(err, ErrVersionConflict): + case errors.Is(err, asset.ErrVersionConflict): conflicts++ default: t.Fatal(err) diff --git a/internal/core/asset/service.go b/internal/core/asset/service.go index 17221c6d..a0e46036 100644 --- a/internal/core/asset/service.go +++ b/internal/core/asset/service.go @@ -68,15 +68,18 @@ type Environment struct { } // @name Environment type CreateInput struct { - Name *string `json:"name" validate:"required"` - MRN *string `json:"mrn" validate:"required"` - Type string `json:"type" validate:"required"` - Providers []string `json:"providers" validate:"required"` - Description *string `json:"description"` - Metadata map[string]interface{} `json:"metadata"` - Schema map[string]string `json:"schema"` - Tags []string `json:"tags"` - CreatedBy string `json:"created_by" validate:"required"` + Name *string `json:"name" validate:"required"` + MRN *string `json:"mrn" validate:"required"` + Type string `json:"type" validate:"required"` + Providers []string `json:"providers" validate:"required"` + Description *string `json:"description"` + Metadata map[string]interface{} `json:"metadata"` + Schema map[string]string `json:"schema"` + Tags []string `json:"tags"` + CreatedBy string `json:"created_by" validate:"required"` + // DeferLinks is set by a discovery run on its first pass: a link field naming, by MRN, an asset + // that does not exist yet is left out whole, for the run to send again once its batch is in. + DeferLinks bool `json:"-"` Sources []AssetSource `json:"sources"` Environments map[string]Environment `json:"environments"` ExternalLinks []ExternalLink `json:"external_links"` @@ -109,6 +112,8 @@ type UpdateInput struct { // FromSync marks a discovery run: the source fills governed fields that are still // empty but never overwrites a value already set, so it needs no expected version. FromSync bool `json:"-"` + // DeferLinks has the meaning it has in CreateInput. + DeferLinks bool `json:"-"` } type Filter struct { @@ -498,6 +503,9 @@ func (s *service) Create(ctx context.Context, input CreateInput) (*Asset, error) if err := s.validateAsset(asset); err != nil { return nil, err } + if err := s.resolveLinkMRNs(ctx, asset, input.DeferLinks); err != nil { + return nil, err + } if err := s.checkLinks(ctx, asset.ID, asset.Metadata, nil); err != nil { return nil, err } @@ -700,6 +708,9 @@ func (s *service) Update(ctx context.Context, id string, input UpdateInput) (*As return nil, err } } + if err := s.resolveLinkMRNs(ctx, asset, input.DeferLinks); err != nil { + return nil, err + } if err := s.checkLinks(ctx, asset.ID, asset.Metadata, oldAsset.Metadata); err != nil { return nil, err } diff --git a/internal/core/asset/terms_pg_test.go b/internal/core/asset/terms_pg_test.go index 7df87063..7d13e2cb 100644 --- a/internal/core/asset/terms_pg_test.go +++ b/internal/core/asset/terms_pg_test.go @@ -1,10 +1,11 @@ -package asset +package asset_test import ( "context" "testing" "time" + "github.com/marmotdata/marmot/internal/core/asset" "github.com/marmotdata/marmot/internal/metrics" "github.com/marmotdata/marmot/internal/store/postgres/pgtest" ) @@ -31,7 +32,7 @@ func TestAssetsByTermListsTheTermAssets(t *testing.T) { t.Fatal(err) } - assets, total, err := NewPostgresRepository(pool, noopRecorder{}).GetAssetsByTerm(ctx, termID, 10, 0) + assets, total, err := asset.NewPostgresRepository(pool, noopRecorder{}).GetAssetsByTerm(ctx, termID, 10, 0) if err != nil { t.Fatal(err) } diff --git a/internal/core/runs/link_mrns_test.go b/internal/core/runs/link_mrns_test.go new file mode 100644 index 00000000..181daad2 --- /dev/null +++ b/internal/core/runs/link_mrns_test.go @@ -0,0 +1,52 @@ +package runs + +import ( + "context" + "strings" + "testing" + + "github.com/marmotdata/marmot/internal/core/asset" + "github.com/marmotdata/marmot/internal/core/lineage" + "github.com/marmotdata/marmot/internal/core/metamodel" + "github.com/marmotdata/marmot/internal/store/postgres/pgtest" + "github.com/stretchr/testify/require" +) + +const linkProfile = `formatVersion: 1 +id: example +version: 1 +defaultLocale: en +fields: + - id: parent_area + type: list + itemType: string + core: true + nullable: true + storage: metadata.example.parent_area + presentation: + labelKey: example.parent_area.label + control: asset +` + +func TestProcessEntities_LinksByMRNAnAssetCreatedLaterInTheSameRun(t *testing.T) { + pool := pgtest.TempDB(t) + ctx := context.Background() + registry, err := metamodel.Load(strings.NewReader(linkProfile)) + require.NoError(t, err) + assets := asset.NewService(asset.NewPostgresRepository(pool, recorderStub{}), asset.WithMetamodel(registry)) + service := NewService(NewPostgresRepository(pool), assets, lineage.NewService(lineage.NewPostgresRepository(pool), assets), nil, recorderStub{}) + + child := tableInput("ventas") + child.Metadata = map[string]interface{}{ + "example": map[string]interface{}{"parent_area": []interface{}{"mrn://table/postgresql/corporacion", "mrn://table/postgresql/nowhere"}}, + } + // The child comes first: its parent does not exist when it is created. + process(t, service, "areas", []CreateAssetInput{child, tableInput("corporacion")}, nil) + + parent, err := assets.GetByMRN(ctx, "mrn://table/postgresql/corporacion") + require.NoError(t, err) + stored, err := assets.GetByMRN(ctx, "mrn://table/postgresql/ventas") + require.NoError(t, err) + value, _ := metamodel.ValueAt(stored.Metadata, "metadata.example.parent_area") + require.Equal(t, []string{parent.ID}, asset.LinkIDs(value), "the MRN becomes the parent's ID; one that names nothing is dropped") +} diff --git a/internal/core/runs/service.go b/internal/core/runs/service.go index a15866a8..57a23b21 100644 --- a/internal/core/runs/service.go +++ b/internal/core/runs/service.go @@ -174,6 +174,12 @@ type GlossarySyncer interface { SyncTerms(ctx context.Context, source string, inputs []glossary.TermInput) (*glossary.SyncResult, error) } +// linkedAsset is an asset whose metadata names others by MRN. +type linkedAsset struct { + id, mrn string + metadata map[string]interface{} +} + type service struct { repo Repository assetService asset.Service @@ -291,6 +297,7 @@ func (s *service) ProcessEntities(ctx context.Context, runID string, assets []Cr return nil, fmt.Errorf("getting last run checkpoints: %w", err) } + var relink []linkedAsset response := &ProcessAssetsResponse{ Assets: make([]AssetResult, 0, len(assets)), Lineage: make([]LineageResult, 0, len(lineage)), @@ -350,6 +357,7 @@ func (s *service) ProcessEntities(ctx context.Context, runID string, assets []Cr QueryLanguage: ast.QueryLanguage, Sources: ast.Sources, CreatedBy: run.CreatedBy, + DeferLinks: true, } created, err := s.assetService.Create(ctx, createInput) if created != nil { @@ -388,6 +396,7 @@ func (s *service) ProcessEntities(ctx context.Context, runID string, assets []Cr Sources: ast.Sources, SkipNotification: true, FromSync: true, + DeferLinks: true, } // Not cached when another writer created the asset after the fetch above. existingAsset := existingAssets[assetMRN] @@ -425,6 +434,10 @@ func (s *service) ProcessEntities(ctx context.Context, runID string, assets []Cr } } + if status != StatusFailed && assetID != "" && asset.HasLinkMRNs(ast.Metadata) { + relink = append(relink, linkedAsset{id: assetID, mrn: assetMRN, metadata: ast.Metadata}) + } + result := AssetResult{ Name: ast.Name, Type: ast.Type, @@ -454,6 +467,17 @@ func (s *service) ProcessEntities(ctx context.Context, runID string, assets []Cr } } + // A link by MRN may name an asset this same batch created after the one that holds it, so the + // fields left out on the way in are written now that every target exists. + // ponytail: an asset the run found unchanged is not revisited; a target that another source + // adds later is linked the next time this asset changes. + for _, linked := range relink { + input := asset.UpdateInput{Metadata: linked.metadata, SkipNotification: true, FromSync: true} + if _, err := s.assetService.Update(ctx, linked.id, input); err != nil { + log.Warn().Err(err).Str("asset_mrn", linked.mrn).Msg("Failed to link assets by MRN") + } + } + staleEntities := s.GetStaleEntities(ctx, lastCheckpoints, currentMRNs) for _, staleMRN := range staleEntities { if err := s.assetService.DeleteByMRN(ctx, staleMRN); err != nil {