From ecea7f571d014cc3d306af32d7716647df060237 Mon Sep 17 00:00:00 2001 From: Frederick Wijayadisusilo Date: Fri, 2 Oct 2026 11:06:15 +0700 Subject: [PATCH 1/2] feat: reduce memory usage for compass upsert --- core/asset/asset.go | 4 +-- core/asset/patch.go | 35 +++++++++++++++++++++ go.mod | 1 - go.sum | 2 -- internal/store/postgres/asset_repository.go | 31 +++++++++++++----- 5 files changed, 59 insertions(+), 14 deletions(-) diff --git a/core/asset/asset.go b/core/asset/asset.go index 302c2d76..95c73b34 100644 --- a/core/asset/asset.go +++ b/core/asset/asset.go @@ -82,9 +82,7 @@ func (a *Asset) Diff(otherAsset *Asset, excludedChangelogPaths []string) (fullCh return fullChangelog, fullChangelog, nil } - simplifiedChangelog = make(diff.Changelog, len(fullChangelog)) - copy(simplifiedChangelog, fullChangelog) - + simplifiedChangelog = fullChangelog for _, path := range excludedChangelogPaths { simplifiedChangelog = simplifiedChangelog.FilterOut(strings.Split(path, ".")) } diff --git a/core/asset/patch.go b/core/asset/patch.go index 97816775..b4f85b47 100644 --- a/core/asset/patch.go +++ b/core/asset/patch.go @@ -5,6 +5,41 @@ import ( "github.com/goto/compass/pkg/mergemap" ) +func (a *Asset) CopyForPatch() Asset { + cp := *a + cp.Data = deepCopyMap(a.Data) + + return cp +} + +func deepCopyMap(m map[string]interface{}) map[string]interface{} { + if m == nil { + return nil + } + + cp := make(map[string]interface{}, len(m)) + for key, value := range m { + cp[key] = deepCopyValue(value) + } + + return cp +} + +func deepCopyValue(value interface{}) interface{} { + switch v := value.(type) { + case map[string]interface{}: + return deepCopyMap(v) + case []interface{}: + cp := make([]interface{}, len(v)) + for i, item := range v { + cp[i] = deepCopyValue(item) + } + return cp + default: + return value + } +} + // patch appends asset with data from map. It mutates the asset itself. func patchAsset(a *Asset, patchData map[string]interface{}) { a.URN = patchString("urn", patchData, a.URN) diff --git a/go.mod b/go.mod index 80cb77ac..c9dd3283 100644 --- a/go.mod +++ b/go.mod @@ -22,7 +22,6 @@ require ( github.com/jackc/pgconn v1.14.3 github.com/jackc/pgerrcode v0.0.0-20201024163028-a0d42d470451 github.com/jackc/pgx/v4 v4.18.2 - github.com/jinzhu/copier v0.4.0 github.com/jmoiron/sqlx v1.3.5 github.com/lib/pq v1.10.2 github.com/newrelic/go-agent/v3 v3.15.2 diff --git a/go.sum b/go.sum index 392100ee..02a8d838 100644 --- a/go.sum +++ b/go.sum @@ -875,8 +875,6 @@ github.com/jackc/puddle v1.1.1/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dv github.com/jackc/puddle v1.1.3/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk= github.com/jeremywohl/flatten v1.0.1 h1:LrsxmB3hfwJuE+ptGOijix1PIfOoKLJ3Uee/mzbgtrs= github.com/jeremywohl/flatten v1.0.1/go.mod h1:4AmD/VxjWcI5SRB0n6szE2A6s2fsNHDLO0nAlMHgfLQ= -github.com/jinzhu/copier v0.4.0 h1:w3ciUoD19shMCRargcpm0cm91ytaBhDvuRpz1ODO/U8= -github.com/jinzhu/copier v0.4.0/go.mod h1:DfbEm0FYsaqBcKcFuvmOZb218JkPGtvSHsKg8S8hyyg= github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc= github.com/jinzhu/now v1.1.1/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8= github.com/jmespath/go-jmespath v0.0.0-20160202185014-0b12d6b521d8/go.mod h1:Nht3zPeWKUH0NzdCt2Blrr5ys8VGpn0CEB0cQHVjt7k= diff --git a/internal/store/postgres/asset_repository.go b/internal/store/postgres/asset_repository.go index dbf802ed..e0e1a8b1 100644 --- a/internal/store/postgres/asset_repository.go +++ b/internal/store/postgres/asset_repository.go @@ -17,7 +17,6 @@ import ( "github.com/goto/compass/pkg/generichelper" "github.com/goto/compass/pkg/queryexpr" "github.com/goto/salt/log" - "github.com/jinzhu/copier" "github.com/jmoiron/sqlx" "github.com/r3labs/diff/v2" ) @@ -481,7 +480,7 @@ func (r *AssetRepository) Upsert( capturedChangelog = fullChangelog resolvedSQLInitialized = asset.BumpOptimusQueryVersions(fetchedAsset.Data, ast.Data, fullChangelog) - upsertedAsset, err = r.update(ctx, tx, ast, &fetchedAsset, simplifiedChangelog) + upsertedAsset, err = r.update(ctx, tx, ast, &fetchedAsset, simplifiedChangelog, fullChangelog) if err != nil { return fmt.Errorf("error updating asset to DB: %w", err) } @@ -542,10 +541,7 @@ func (r *AssetRepository) UpsertPatch( //nolint:gocognit } // update flow - var newAsset asset.Asset - if err := copier.CopyWithOption(&newAsset, fetchedAsset, copier.Option{DeepCopy: true}); err != nil { - return err - } + newAsset := fetchedAsset.CopyForPatch() newAsset.Patch(patchData) newAsset.RefreshedAt = ast.RefreshedAt @@ -563,7 +559,7 @@ func (r *AssetRepository) UpsertPatch( //nolint:gocognit capturedChangelog = fullChangelog resolvedSQLInitialized = asset.BumpOptimusQueryVersions(fetchedAsset.Data, newAsset.Data, fullChangelog) - upsertedAsset, err = r.update(ctx, tx, &newAsset, &fetchedAsset, simplifiedChangelog) + upsertedAsset, err = r.update(ctx, tx, &newAsset, &fetchedAsset, simplifiedChangelog, fullChangelog) if err != nil { return fmt.Errorf("error updating asset to DB: %w", err) } @@ -1196,7 +1192,15 @@ func (r *AssetRepository) insert(ctx context.Context, tx *sqlx.Tx, ast *asset.As return &insertedAsset, nil } -func (r *AssetRepository) update(ctx context.Context, tx *sqlx.Tx, newAsset, oldAsset *asset.Asset, clog diff.Changelog) (*asset.Asset, error) { +// update persists newAsset. clog is the changelog stored against the new +// version; fullChangelog is the unfiltered one, which is what decides whether +// the asset changed at all. +func (r *AssetRepository) update( + ctx context.Context, + tx *sqlx.Tx, + newAsset, oldAsset *asset.Asset, + clog, fullChangelog diff.Changelog, +) (*asset.Asset, error) { assetID := oldAsset.ID if !isValidUUID(assetID) { return nil, asset.InvalidError{AssetID: assetID} @@ -1209,6 +1213,17 @@ func (r *AssetRepository) update(ctx context.Context, tx *sqlx.Tx, newAsset, old currentTime = *newAsset.RefreshedAt } + // Nothing about the asset changed, so only record that it was seen again + // rather than rewriting the whole row and its data blob. + if len(fullChangelog) == 0 { + refreshedAsset, err := r.updateAssetRefreshedAt(ctx, tx, assetID, currentTime) + if err != nil { + return nil, err + } + + return &refreshedAsset, nil + } + // managing owners newAssetOwners, err := r.createOrFetchUsers(ctx, tx, newAsset.Owners) if err != nil { From 623533b3a6c78995c297db184ffe0f8c43ce9c38 Mon Sep 17 00:00:00 2001 From: Frederick Wijayadisusilo Date: Tue, 6 Oct 2026 18:01:34 +0700 Subject: [PATCH 2/2] feat: support wider changelog exclusion --- core/asset/asset.go | 13 +- core/asset/asset_test.go | 4 +- core/asset/changelog.go | 102 ++++++++++ core/asset/changelog_test.go | 205 ++++++++++++++++++++ core/asset/config.go | 10 + core/asset/service.go | 4 +- internal/store/postgres/asset_repository.go | 89 ++++++--- 7 files changed, 380 insertions(+), 47 deletions(-) create mode 100644 core/asset/changelog.go create mode 100644 core/asset/changelog_test.go diff --git a/core/asset/asset.go b/core/asset/asset.go index 95c73b34..5cc5d985 100644 --- a/core/asset/asset.go +++ b/core/asset/asset.go @@ -3,7 +3,6 @@ package asset //go:generate mockery --name=Repository -r --case underscore --with-expecter --structname AssetRepository --filename asset_repository.go --output=./mocks import ( "context" - "strings" "time" "github.com/goto/compass/core/user" @@ -72,21 +71,13 @@ type SoftDeleteAssetParams struct { // Diff returns nil changelog with nil error if equal // returns wrapped r3labs/diff Changelog struct with nil error if not equal -func (a *Asset) Diff(otherAsset *Asset, excludedChangelogPaths []string) (fullChangelog, simplifiedChangelog diff.Changelog, err error) { +func (a *Asset) Diff(otherAsset *Asset, excludedChangelogPathSegments [][]string) (fullChangelog, simplifiedChangelog diff.Changelog, err error) { fullChangelog, err = diff.Diff(a, otherAsset, diff.DiscardComplexOrigin(), diff.AllowTypeMismatch(true)) if err != nil { return nil, nil, err } - if len(excludedChangelogPaths) == 0 { - return fullChangelog, fullChangelog, nil - } - - simplifiedChangelog = fullChangelog - for _, path := range excludedChangelogPaths { - simplifiedChangelog = simplifiedChangelog.FilterOut(strings.Split(path, ".")) - } - + simplifiedChangelog = filterExcludedChangelog(fullChangelog, excludedChangelogPathSegments) return fullChangelog, simplifiedChangelog, nil } diff --git a/core/asset/asset_test.go b/core/asset/asset_test.go index f5530faf..7954b870 100644 --- a/core/asset/asset_test.go +++ b/core/asset/asset_test.go @@ -83,7 +83,7 @@ func TestDiffTopLevel(t *testing.T) { t.Fatal(err) } - _, cl, err := sourceAsset.Diff(&targetAsset, []string{}) + _, cl, err := sourceAsset.Diff(&targetAsset, nil) assert.Equal(t, tc.Error, err) require.Equal(t, len(tc.Changelog), len(cl)) @@ -247,7 +247,7 @@ func TestDiffData(t *testing.T) { t.Fatal(err) } - _, cl, err := sourceAsset.Diff(&targetAsset, []string{"update_time"}) + _, cl, err := sourceAsset.Diff(&targetAsset, asset.ParseExcludedChangelogPathSegments([]string{"update_time"})) assert.Equal(t, tc.Error, err) require.Equal(t, len(tc.Changelog), len(cl)) diff --git a/core/asset/changelog.go b/core/asset/changelog.go new file mode 100644 index 00000000..9ce87a97 --- /dev/null +++ b/core/asset/changelog.go @@ -0,0 +1,102 @@ +package asset + +import ( + "strings" + + "github.com/r3labs/diff/v2" +) + +// ParseExcludedChangelogPathSegments splits dotted exclusion paths for diff filtering. +func ParseExcludedChangelogPathSegments(excludedPaths []string) [][]string { + if len(excludedPaths) == 0 { + return nil + } + + segments := make([][]string, len(excludedPaths)) + for i, path := range excludedPaths { + segments[i] = strings.Split(path, ".") + } + + return segments +} + +func changelogPathMatchesExcluded(changePath, excludedSegments []string) bool { + changeIdx := 0 + for excludedIdx := range excludedSegments { + changeIdx = skipExcludedChangelogPathPrefix(changePath, changeIdx, excludedSegments, excludedIdx) + if changeIdx >= len(changePath) || changePath[changeIdx] != excludedSegments[excludedIdx] { + return false + } + changeIdx++ + } + + return true +} + +func skipExcludedChangelogPathPrefix(changePath []string, changeIdx int, excludedSegments []string, excludedIdx int) int { + for changeIdx < len(changePath) && isSkippableExcludedChangelogSegment(changePath, changeIdx, excludedSegments, excludedIdx) { + changeIdx++ + } + + return changeIdx +} + +func isSkippableExcludedChangelogSegment( + changePath []string, + changeIdx int, + excludedSegments []string, + excludedIdx int, +) bool { + if isChangelogArrayIndex(changePath[changeIdx]) { + return true + } + + return excludedIdx > 0 && changePath[changeIdx] == excludedSegments[excludedIdx-1] +} + +func changelogChangeExcluded(changePath []string, excludedSegments [][]string) bool { + for _, segments := range excludedSegments { + if changelogPathMatchesExcluded(changePath, segments) { + return true + } + } + + return false +} + +func isChangelogArrayIndex(segment string) bool { + if segment == "" { + return false + } + for _, r := range segment { + if r < '0' || r > '9' { + return false + } + } + + return true +} + +func filterExcludedChangelog(changelog diff.Changelog, excludedPathSegments [][]string) diff.Changelog { + if len(excludedPathSegments) == 0 || len(changelog) == 0 { + return changelog + } + + var filtered diff.Changelog + for _, change := range changelog { + if changelogChangeExcluded(change.Path, excludedPathSegments) { + continue + } + filtered = append(filtered, change) + } + + if len(filtered) == len(changelog) { + return changelog + } + + if len(filtered) == 0 { + return nil + } + + return filtered +} diff --git a/core/asset/changelog_test.go b/core/asset/changelog_test.go new file mode 100644 index 00000000..0c8cddc7 --- /dev/null +++ b/core/asset/changelog_test.go @@ -0,0 +1,205 @@ +package asset_test + +import ( + "encoding/json" + "testing" + + "github.com/goto/compass/core/asset" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func excludedSegments(paths ...string) [][]string { + return asset.ParseExcludedChangelogPathSegments(paths) +} + +func TestExcludedChangelogPathsInsideArrays(t *testing.T) { + source := `{ + "data": { + "columns": [ + { + "name": "jakarta_data_date", + "description": "old description" + } + ] + } + }` + target := `{ + "data": { + "columns": [ + { + "name": "jakarta_data_date", + "description": "new description" + } + ] + } + }` + + var sourceAsset, targetAsset asset.Asset + require.NoError(t, json.Unmarshal([]byte(source), &sourceAsset)) + require.NoError(t, json.Unmarshal([]byte(target), &targetAsset)) + + full, simplified, err := sourceAsset.Diff(&targetAsset, excludedSegments("data.columns.description")) + require.NoError(t, err) + require.Len(t, full, 1) + assert.Equal(t, []string{"data", "columns", "0", "description"}, full[0].Path) + assert.Empty(t, simplified) +} + +func TestExcludedChangelogPathsNestedColumnArrays(t *testing.T) { + source := `{ + "data": { + "columns": [ + { + "name": "parent", + "columns": [ + { + "name": "child", + "description": "old" + } + ] + } + ] + } + }` + target := `{ + "data": { + "columns": [ + { + "name": "parent", + "columns": [ + { + "name": "child", + "description": "new" + } + ] + } + ] + } + }` + + var sourceAsset, targetAsset asset.Asset + require.NoError(t, json.Unmarshal([]byte(source), &sourceAsset)) + require.NoError(t, json.Unmarshal([]byte(target), &targetAsset)) + + _, simplified, err := sourceAsset.Diff(&targetAsset, excludedSegments("data.columns.description")) + require.NoError(t, err) + assert.Empty(t, simplified) +} + +func TestExcludedChangelogPathsMultipleTopLevelColumns(t *testing.T) { + source := `{ + "data": { + "columns": [ + {"name": "a", "description": "old-a"}, + {"name": "b", "description": "old-b"} + ] + } + }` + target := `{ + "data": { + "columns": [ + {"name": "a", "description": "new-a"}, + {"name": "b", "description": "new-b"} + ] + } + }` + + var sourceAsset, targetAsset asset.Asset + require.NoError(t, json.Unmarshal([]byte(source), &sourceAsset)) + require.NoError(t, json.Unmarshal([]byte(target), &targetAsset)) + + full, simplified, err := sourceAsset.Diff(&targetAsset, excludedSegments("data.columns.description")) + require.NoError(t, err) + require.Len(t, full, 2) + assert.Empty(t, simplified) +} + +func TestExcludedChangelogPathsMixedExcludedAndTrackedColumnChanges(t *testing.T) { + source := `{ + "data": { + "columns": [ + {"name": "a", "description": "old-a"}, + {"name": "old-b", "description": "old-llm-b"} + ] + } + }` + target := `{ + "data": { + "columns": [ + {"name": "a", "description": "new-a"}, + {"name": "new-b", "description": "new-llm-b"} + ] + } + }` + + var sourceAsset, targetAsset asset.Asset + require.NoError(t, json.Unmarshal([]byte(source), &sourceAsset)) + require.NoError(t, json.Unmarshal([]byte(target), &targetAsset)) + + _, simplified, err := sourceAsset.Diff(&targetAsset, excludedSegments("data.columns.description")) + require.NoError(t, err) + require.Len(t, simplified, 1) + assert.Equal(t, []string{"data", "columns", "1", "name"}, simplified[0].Path) +} + +func TestExcludedChangelogPathsKeepsOtherColumnFields(t *testing.T) { + source := `{ + "data": { + "columns": [ + { + "name": "old_name", + "description": "same" + } + ] + } + }` + target := `{ + "data": { + "columns": [ + { + "name": "new_name", + "description": "same" + } + ] + } + }` + + var sourceAsset, targetAsset asset.Asset + require.NoError(t, json.Unmarshal([]byte(source), &sourceAsset)) + require.NoError(t, json.Unmarshal([]byte(target), &targetAsset)) + + _, simplified, err := sourceAsset.Diff(&targetAsset, excludedSegments("data.columns.description")) + require.NoError(t, err) + require.Len(t, simplified, 1) + assert.Equal(t, []string{"data", "columns", "0", "name"}, simplified[0].Path) +} + +func TestFilterExcludedChangelogReusesChangelogWhenNothingRemoved(t *testing.T) { + source := `{"data": {"title": "old"}}` + target := `{"data": {"title": "new"}}` + + var sourceAsset, targetAsset asset.Asset + require.NoError(t, json.Unmarshal([]byte(source), &sourceAsset)) + require.NoError(t, json.Unmarshal([]byte(target), &targetAsset)) + + full, simplified, err := sourceAsset.Diff(&targetAsset, excludedSegments("data.columns.description")) + require.NoError(t, err) + require.NotEmpty(t, full) + assert.Equal(t, full, simplified) +} + +func TestExcludedChangelogPathsStillFiltersTopLevelDataFields(t *testing.T) { + source := `{"data": {"update_time": "2024-01-01T00:00:00Z", "title": "old"}}` + target := `{"data": {"update_time": "2024-01-02T00:00:00Z", "title": "new"}}` + + var sourceAsset, targetAsset asset.Asset + require.NoError(t, json.Unmarshal([]byte(source), &sourceAsset)) + require.NoError(t, json.Unmarshal([]byte(target), &targetAsset)) + + full, simplified, err := sourceAsset.Diff(&targetAsset, excludedSegments("data.update_time")) + require.NoError(t, err) + require.Len(t, full, 2) + require.Len(t, simplified, 1) + assert.Equal(t, []string{"data", "title"}, simplified[0].Path) +} diff --git a/core/asset/config.go b/core/asset/config.go index 9bd61ab7..25447181 100644 --- a/core/asset/config.go +++ b/core/asset/config.go @@ -11,11 +11,21 @@ type Config struct { AdditionalTypes []string `mapstructure:"additional_types"` DeleteAssetsTimeout time.Duration `mapstructure:"delete_assets_timeout" default:"5m"` ExcludedChangelogPaths []string `mapstructure:"excluded_changelog_paths"` + ExcludedChangelogPathSegments [][]string `mapstructure:"-"` ColumnLineageHost string `mapstructure:"column_lineage_host"` ColumnLineageChangeIdentifier string `mapstructure:"column_lineage_change_identifier"` } +func (c *Config) prepareExcludedChangelog() { + if c.ExcludedChangelogPaths == nil { + c.ExcludedChangelogPaths = []string{} + } + c.ExcludedChangelogPathSegments = ParseExcludedChangelogPathSegments(c.ExcludedChangelogPaths) +} + func (c *Config) Validate() error { + c.prepareExcludedChangelog() + if c.DeleteAssetsTimeout == 0 { return errDeleteAssetsTimeoutIsZero } diff --git a/core/asset/service.go b/core/asset/service.go index 464ad669..1fe39a31 100644 --- a/core/asset/service.go +++ b/core/asset/service.go @@ -72,9 +72,7 @@ func NewService(deps ServiceDeps) (service *Service, cancel func()) { shutdownCtx: shutdownCtx, shutdownCancel: shutdownCancel, } - if newService.config.ExcludedChangelogPaths == nil { - newService.config.ExcludedChangelogPaths = []string{} - } + newService.config.prepareExcludedChangelog() return newService, func() { shutdownCancel() diff --git a/internal/store/postgres/asset_repository.go b/internal/store/postgres/asset_repository.go index e0e1a8b1..4b0fbbff 100644 --- a/internal/store/postgres/asset_repository.go +++ b/internal/store/postgres/asset_repository.go @@ -454,7 +454,7 @@ func (r *AssetRepository) Upsert( } resolvedSQLInitialized = asset.InitOptimusQueryVersions(ast.Data) - _, simplifiedChangelog, err := new(asset.Asset).Diff(ast, assetConfig.ExcludedChangelogPaths) + _, simplifiedChangelog, err := new(asset.Asset).Diff(ast, assetConfig.ExcludedChangelogPathSegments) if err != nil { return fmt.Errorf("error diffing two assets: %w", err) } @@ -473,14 +473,17 @@ func (r *AssetRepository) Upsert( // reset IsDeleted flag if asset is resync'd ast.IsDeleted = false - fullChangelog, simplifiedChangelog, err := fetchedAsset.Diff(ast, assetConfig.ExcludedChangelogPaths) + fullChangelog, simplifiedChangelog, err := fetchedAsset.Diff(ast, assetConfig.ExcludedChangelogPathSegments) if err != nil { return fmt.Errorf("error diffing two assets: %w", err) } capturedChangelog = fullChangelog resolvedSQLInitialized = asset.BumpOptimusQueryVersions(fetchedAsset.Data, ast.Data, fullChangelog) - upsertedAsset, err = r.update(ctx, tx, ast, &fetchedAsset, simplifiedChangelog, fullChangelog) + upsertedAsset, err = r.update(ctx, tx, ast, &fetchedAsset, assetUpdateChangelogs{ + simplified: simplifiedChangelog, + full: fullChangelog, + }) if err != nil { return fmt.Errorf("error updating asset to DB: %w", err) } @@ -524,7 +527,7 @@ func (r *AssetRepository) UpsertPatch( //nolint:gocognit } resolvedSQLInitialized = asset.InitOptimusQueryVersions(ast.Data) - fullChangelog, simplifiedChangelog, err := new(asset.Asset).Diff(ast, assetConfig.ExcludedChangelogPaths) + fullChangelog, simplifiedChangelog, err := new(asset.Asset).Diff(ast, assetConfig.ExcludedChangelogPathSegments) if err != nil { return fmt.Errorf("error diffing two assets: %w", err) } @@ -552,14 +555,17 @@ func (r *AssetRepository) UpsertPatch( //nolint:gocognit return err } - fullChangelog, simplifiedChangelog, err := fetchedAsset.Diff(&newAsset, assetConfig.ExcludedChangelogPaths) + fullChangelog, simplifiedChangelog, err := fetchedAsset.Diff(&newAsset, assetConfig.ExcludedChangelogPathSegments) if err != nil { return fmt.Errorf("error diffing two assets: %w", err) } capturedChangelog = fullChangelog resolvedSQLInitialized = asset.BumpOptimusQueryVersions(fetchedAsset.Data, newAsset.Data, fullChangelog) - upsertedAsset, err = r.update(ctx, tx, &newAsset, &fetchedAsset, simplifiedChangelog, fullChangelog) + upsertedAsset, err = r.update(ctx, tx, &newAsset, &fetchedAsset, assetUpdateChangelogs{ + simplified: simplifiedChangelog, + full: fullChangelog, + }) if err != nil { return fmt.Errorf("error updating asset to DB: %w", err) } @@ -1192,14 +1198,16 @@ func (r *AssetRepository) insert(ctx context.Context, tx *sqlx.Tx, ast *asset.As return &insertedAsset, nil } -// update persists newAsset. clog is the changelog stored against the new -// version; fullChangelog is the unfiltered one, which is what decides whether -// the asset changed at all. +type assetUpdateChangelogs struct { + simplified diff.Changelog + full diff.Changelog +} + func (r *AssetRepository) update( ctx context.Context, tx *sqlx.Tx, newAsset, oldAsset *asset.Asset, - clog, fullChangelog diff.Changelog, + changelogs assetUpdateChangelogs, ) (*asset.Asset, error) { assetID := oldAsset.ID if !isValidUUID(assetID) { @@ -1213,9 +1221,7 @@ func (r *AssetRepository) update( currentTime = *newAsset.RefreshedAt } - // Nothing about the asset changed, so only record that it was seen again - // rather than rewriting the whole row and its data blob. - if len(fullChangelog) == 0 { + if len(changelogs.full) == 0 { refreshedAsset, err := r.updateAssetRefreshedAt(ctx, tx, assetID, currentTime) if err != nil { return nil, err @@ -1224,21 +1230,11 @@ func (r *AssetRepository) update( return &refreshedAsset, nil } - // managing owners - newAssetOwners, err := r.createOrFetchUsers(ctx, tx, newAsset.Owners) - if err != nil { - return nil, fmt.Errorf("error creating and fetching owners: %w", err) - } - toInserts, toRemoves := r.compareOwners(oldAsset.Owners, newAssetOwners) - if err := r.insertOwners(ctx, tx, assetID, toInserts); err != nil { - return nil, fmt.Errorf("error inserting asset's new owners: %w", err) - } - if err := r.removeOwners(ctx, tx, assetID, toRemoves); err != nil { - return nil, fmt.Errorf("error removing asset's old owners: %w", err) + if err := r.replaceAssetOwners(ctx, tx, assetID, oldAsset, newAsset); err != nil { + return nil, err } - // update assets - if len(clog) != 0 { + if len(changelogs.simplified) != 0 { newVersion, err := asset.IncreaseMinorVersion(oldAsset.Version) if err != nil { return nil, err @@ -1254,16 +1250,47 @@ func (r *AssetRepository) update( return nil, err } - if len(clog) != 0 { - // insert versions - if err := r.insertAssetVersion(ctx, tx, newAsset, clog); err != nil { - return nil, err - } + if err := r.insertAssetVersionIfNeeded(ctx, tx, newAsset, changelogs.simplified); err != nil { + return nil, err } return &updatedAsset, nil } +func (r *AssetRepository) replaceAssetOwners( + ctx context.Context, + tx *sqlx.Tx, + assetID string, + oldAsset, newAsset *asset.Asset, +) error { + newAssetOwners, err := r.createOrFetchUsers(ctx, tx, newAsset.Owners) + if err != nil { + return fmt.Errorf("error creating and fetching owners: %w", err) + } + toInserts, toRemoves := r.compareOwners(oldAsset.Owners, newAssetOwners) + if err := r.insertOwners(ctx, tx, assetID, toInserts); err != nil { + return fmt.Errorf("error inserting asset's new owners: %w", err) + } + if err := r.removeOwners(ctx, tx, assetID, toRemoves); err != nil { + return fmt.Errorf("error removing asset's old owners: %w", err) + } + + return nil +} + +func (r *AssetRepository) insertAssetVersionIfNeeded( + ctx context.Context, + tx *sqlx.Tx, + newAsset *asset.Asset, + versionChangelog diff.Changelog, +) error { + if len(versionChangelog) == 0 { + return nil + } + + return r.insertAssetVersion(ctx, tx, newAsset, versionChangelog) +} + func (r *AssetRepository) updateAsset(ctx context.Context, tx *sqlx.Tx, assetID string, newAsset *asset.Asset) (asset.Asset, error) { updateCTE := sq.Update("assets"). Set("urn", newAsset.URN).