fix: meteor upsert replacement approach

core/asset/asset_test.go

@@ -469,6 +469,95 @@ func TestAssetPatch(t *testing.T) {
 				},
 			},
 		},
+		{
+			description: "should set data labels to null when patch value is null",
+			asset: asset.Asset{
+				Data: map[string]interface{}{
+					"labels": map[string]interface{}{
+						"usage_tag": "none",
+					},
+				},
+			},
+			patchData: map[string]interface{}{
+				"data": map[string]interface{}{
+					"labels": nil,
+				},
+			},
+			expected: asset.Asset{
+				Data: map[string]interface{}{
+					"labels": nil,
+				},
+			},
+		},
+		{
+			description: "should set data labels to null when patch value is empty map",
+			asset: asset.Asset{
+				Data: map[string]interface{}{
+					"labels": map[string]interface{}{
+						"usage_tag": "none",
+					},
+				},
+			},
+			patchData: map[string]interface{}{
+				"data": map[string]interface{}{
+					"labels": map[string]interface{}{},
+				},
+			},
+			expected: asset.Asset{
+				Data: map[string]interface{}{
+					"labels": nil,
+				},
+			},
+		},
+		{
+			description: "should replace data labels when patch has content",
+			asset: asset.Asset{
+				Data: map[string]interface{}{
+					"labels": map[string]interface{}{
+						"category_table_tag": "business-critical",
+						"usage_tag":          "none",
+					},
+				},
+			},
+			patchData: map[string]interface{}{
+				"data": map[string]interface{}{
+					"labels": map[string]interface{}{
+						"usage_tag": "none",
+					},
+				},
+			},
+			expected: asset.Asset{
+				Data: map[string]interface{}{
+					"labels": map[string]interface{}{
+						"usage_tag": "none",
+					},
+				},
+			},
+		},
+		{
+			description: "should ignore data labels when key is omitted from patch",
+			asset: asset.Asset{
+				Data: map[string]interface{}{
+					"labels": map[string]interface{}{
+						"usage_tag": "none",
+					},
+					"foo": "bar",
+				},
+			},
+			patchData: map[string]interface{}{
+				"data": map[string]interface{}{
+					"foo": "baz",
+				},
+			},
+			expected: asset.Asset{
+				Data: map[string]interface{}{
+					"labels": map[string]interface{}{
+						"usage_tag": "none",
+					},
+					"foo": "baz",
+				},
+			},
+		},
 	}
 	for _, tc := range testcases {
 		t.Run(tc.description, func(t *testing.T) {

core/asset/patch.go

@@ -5,6 +5,11 @@ import (
 	"github.com/goto/compass/pkg/mergemap"
 )
 
+const (
+	dataFieldLabels          = "labels"
+	dataFieldPartitionDetail = "partition_detail"
+)
+
 func (a *Asset) CopyForPatch() Asset {
 	cp := *a
 	cp.Data = deepCopyMap(a.Data)
@@ -131,6 +136,8 @@ func patchAssetData(a *Asset, data interface{}) {
 		return
 	}
 
+	patchReplaceableDataMapFields(a.Data, dataMap)
+
 	// Configure array merge behavior for data.columns
 	arrayMergeConfig := map[string]string{
 		"columns": "name",
@@ -139,6 +146,32 @@ func patchAssetData(a *Asset, data interface{}) {
 	a.Data = mergemap.Merge(a.Data, dataMap, arrayMergeConfig)
 }
 
+func patchReplaceableDataMapFields(dst, patch map[string]interface{}) {
+	for _, key := range []string{dataFieldLabels, dataFieldPartitionDetail} {
+		val, ok := patch[key]
+		if !ok {
+			continue
+		}
+
+		if val == nil || isEmptyStringMap(val) {
+			dst[key] = nil
+		} else {
+			dst[key] = deepCopyValue(val)
+		}
+
+		delete(patch, key)
+	}
+}
+
+func isEmptyStringMap(v interface{}) bool {
+	switch m := v.(type) {
+	case map[string]interface{}:
+		return len(m) == 0
+	default:
+		return false
+	}
+}
+
 func patchString(key string, data map[string]interface{}, defaultVal string) string {
 	_, exists := data[key]
 	if !exists {