Support Wider Changelog Exclusion

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,23 +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 = make(diff.Changelog, len(fullChangelog))
-	copy(simplifiedChangelog, fullChangelog)
-
-	for _, path := range excludedChangelogPaths {
-		simplifiedChangelog = simplifiedChangelog.FilterOut(strings.Split(path, "."))
-	}
-
+	simplifiedChangelog = filterExcludedChangelog(fullChangelog, excludedChangelogPathSegments)
 	return fullChangelog, simplifiedChangelog, nil
 }
 

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))

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
+}

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)
+}

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
 	}