From e0b6faddbe0012b7a3e130bc863fabcfd4dfc385 Mon Sep 17 00:00:00 2001
From: Lennart Kats <lennart.kats@databricks.com>
Date: Fri, 20 Dec 2024 09:39:30 +0100
Subject: [PATCH] Add a recommendation for pipeline catalogs

---
 .../mutator/apply_presets_catalog_schema.go   | 29 ++++++++++++++++++-
 .../apply_presets_catalog_schema_test.go      | 12 +++++++-
 2 files changed, 39 insertions(+), 2 deletions(-)

diff --git a/bundle/config/mutator/apply_presets_catalog_schema.go b/bundle/config/mutator/apply_presets_catalog_schema.go
index 6276d7c52..269cfd47b 100644
--- a/bundle/config/mutator/apply_presets_catalog_schema.go
+++ b/bundle/config/mutator/apply_presets_catalog_schema.go
@@ -65,13 +65,24 @@ func (m *applyPresetsCatalogSchema) Apply(ctx context.Context, b *bundle.Bundle)
 	}
 
 	// Pipelines
-	for _, pl := range r.Pipelines {
+	allSameCatalog := allPipelinesSameCatalog(&r)
+	for key, pl := range r.Pipelines {
 		if pl.PipelineSpec == nil {
 			continue
 		}
 		if pl.Catalog == "" {
 			pl.Catalog = p.Catalog
 		}
+		if allSameCatalog && pl.Catalog == p.Catalog {
+			// Just for the common case where all pipelines have the same catalog,
+			// we show a recommendation to leave it out and rely on presets.
+			// This can happen when using the original default template.
+			diags = diags.Extend(diag.Diagnostics{{
+				Summary:   "Omit the catalog field since it will be automatically populated from presets.catalog",
+				Severity:  diag.Recommendation,
+				Locations: b.Config.GetLocations("resources.pipelines." + key + ".catalog"),
+			}})
+		}
 		if pl.GatewayDefinition != nil {
 			if pl.GatewayDefinition.GatewayStorageCatalog == "" {
 				pl.GatewayDefinition.GatewayStorageCatalog = p.Catalog
@@ -347,3 +358,19 @@ func fileIncludesPattern(ctx context.Context, filePath string, expected string)
 	}
 	return matched
 }
+
+func allPipelinesSameCatalog(r *config.Resources) bool {
+	var firstCatalog string
+
+	for _, pl := range r.Pipelines {
+		if pl.PipelineSpec == nil || pl.PipelineSpec.Catalog == "" {
+			return false
+		}
+		if firstCatalog == "" {
+			firstCatalog = pl.PipelineSpec.Catalog
+		} else if pl.PipelineSpec.Catalog != firstCatalog {
+			return false
+		}
+	}
+	return firstCatalog != ""
+}
diff --git a/bundle/config/mutator/apply_presets_catalog_schema_test.go b/bundle/config/mutator/apply_presets_catalog_schema_test.go
index df4efeb8a..d0f9975fc 100644
--- a/bundle/config/mutator/apply_presets_catalog_schema_test.go
+++ b/bundle/config/mutator/apply_presets_catalog_schema_test.go
@@ -179,7 +179,7 @@ func TestApplyPresetsCatalogSchemaWhenAlreadySet(t *testing.T) {
 	b := mockPresetsCatalogSchema()
 	recordedFields := recordPlaceholderFields(t, b)
 
-	diags := bundle.Apply(context.Background(), b, mutator.ApplyPresets())
+	diags := bundle.Apply(context.Background(), b, mutator.ApplyPresetsCatalogSchema())
 	require.NoError(t, diags.Error())
 
 	for _, f := range recordedFields {
@@ -190,6 +190,16 @@ func TestApplyPresetsCatalogSchemaWhenAlreadySet(t *testing.T) {
 	}
 }
 
+func TestApplyPresetsCatalogSchemaRecommmendRemovingCatalog(t *testing.T) {
+	b := mockPresetsCatalogSchema()
+	b.Config.Resources.Jobs["key"].Parameters = nil // avoid warnings about the job parameters
+	b.Config.Resources.Pipelines["key"].Catalog = "my_catalog"
+
+	diags := bundle.Apply(context.Background(), b, mutator.ApplyPresetsCatalogSchema())
+	require.Equal(t, 1, len(diags))
+	require.Equal(t, "Omit the catalog field since it will be automatically populated from presets.catalog", diags[0].Summary)
+}
+
 func TestApplyPresetsCatalogSchemaWhenNotSet(t *testing.T) {
 	b := mockPresetsCatalogSchema()
 	recordedFields := recordPlaceholderFields(t, b)