Fixed variable override in target with full variable syntax (#1749)

## Changes
This PR makes sure that both of this override syntax for variables work
correctly
```
targets:
  dev:
    variables:
      cluster1:
        spark_version: "14.2.x-scala2.11"
        node_type_id: "Standard_DS3_v2"
        num_workers: 4
        spark_conf:
          spark.speculation: false
          spark.databricks.delta.retentionDurationCheck.enabled: false
      cluster2:
        default:
          spark_version: "14.2.x-scala2.11"
          node_type_id: "Standard_DS3_v2"
          num_workers: 4
          spark_conf:
            spark.speculation: false
            spark.databricks.delta.retentionDurationCheck.enabled: false
```
## Tests
Added regression test

---------

Co-authored-by: Pieter Noordhuis <pieter.noordhuis@databricks.com>

bundle/config/root.go

@@ -406,6 +406,30 @@ func (r *Root) MergeTargetOverrides(name string) error {
 	return r.updateWithDynamicValue(root)
 }
 
+var variableKeywords = []string{"default", "lookup"}
+
+// isFullVariableOverrideDef checks if the given value is a full syntax varaible override.
+// A full syntax variable override is a map with only one of the following
+// keys: "default", "lookup".
+func isFullVariableOverrideDef(v dyn.Value) bool {
+	mv, ok := v.AsMap()
+	if !ok {
+		return false
+	}
+
+	if mv.Len() != 1 {
+		return false
+	}
+
+	for _, keyword := range variableKeywords {
+		if _, ok := mv.GetByString(keyword); ok {
+			return true
+		}
+	}
+
+	return false
+}
+
 // rewriteShorthands performs lightweight rewriting of the configuration
 // tree where we allow users to write a shorthand and must rewrite to the full form.
 func rewriteShorthands(v dyn.Value) (dyn.Value, error) {
@@ -433,30 +457,27 @@ func rewriteShorthands(v dyn.Value) (dyn.Value, error) {
 				}, variable.Locations()), nil
 
 			case dyn.KindMap, dyn.KindSequence:
-				lookup, err := dyn.Get(variable, "lookup")
+				// If it's a full variable definition, leave it as is.
-				// If lookup is set, we don't want to rewrite the variable and return it as is.
+				if isFullVariableOverrideDef(variable) {
-				if err == nil && lookup.Kind() != dyn.KindInvalid {
 					return variable, nil
 				}
 
 				// Check if the original definition of variable has a type field.
+				// If it has a type field, it means the shorthand is a value of a complex type.
 				// Type might not be found if the variable overriden in a separate file
 				// and configuration is not merged yet.
 				typeV, err := dyn.GetByPath(v, p.Append(dyn.Key("type")))
-				if err != nil {
+				if err == nil && typeV.MustString() == "complex" {
-					return dyn.NewValue(map[string]dyn.Value{
-						"default": variable,
-					}, variable.Locations()), nil
-				}
-
-				if typeV.MustString() == "complex" {
 					return dyn.NewValue(map[string]dyn.Value{
 						"type":    typeV,
 						"default": variable,
 					}, variable.Locations()), nil
 				}
 
-				return variable, nil
+				// If it's a shorthand, rewrite it to a full variable definition.
+				return dyn.NewValue(map[string]dyn.Value{
+					"default": variable,
+				}, variable.Locations()), nil
 
 			default:
 				return variable, nil

bundle/tests/complex_variables_test.go

@@ -81,9 +81,10 @@ func TestComplexVariablesOverrideWithMultipleFiles(t *testing.T) {
 		),
 	))
 	require.NoError(t, diags.Error())
-
+	for _, cluster := range b.Config.Resources.Jobs["my_job"].JobClusters {
-	require.Equal(t, "14.2.x-scala2.11", b.Config.Resources.Jobs["my_job"].JobClusters[0].NewCluster.SparkVersion)
+		require.Equalf(t, "14.2.x-scala2.11", cluster.NewCluster.SparkVersion, "cluster: %v", cluster.JobClusterKey)
-	require.Equal(t, "Standard_DS3_v2", b.Config.Resources.Jobs["my_job"].JobClusters[0].NewCluster.NodeTypeId)
+		require.Equalf(t, "Standard_DS3_v2", cluster.NewCluster.NodeTypeId, "cluster: %v", cluster.JobClusterKey)
-	require.Equal(t, 4, b.Config.Resources.Jobs["my_job"].JobClusters[0].NewCluster.NumWorkers)
+		require.Equalf(t, 4, cluster.NewCluster.NumWorkers, "cluster: %v", cluster.JobClusterKey)
-	require.Equal(t, "false", b.Config.Resources.Jobs["my_job"].JobClusters[0].NewCluster.SparkConf["spark.speculation"])
+		require.Equalf(t, "false", cluster.NewCluster.SparkConf["spark.speculation"], "cluster: %v", cluster.JobClusterKey)
+	}
 }

bundle/tests/variables/complex_multiple_files/databricks.yml

@@ -5,13 +5,48 @@ resources:
   jobs:
     my_job:
       job_clusters:
-        - job_cluster_key: key
+        - job_cluster_key: key1
-          new_cluster: ${var.cluster}
+          new_cluster: ${var.cluster1}
-
+        - job_cluster_key: key2
+          new_cluster: ${var.cluster2}
+        - job_cluster_key: key3
+          new_cluster: ${var.cluster3}
+        - job_cluster_key: key4
+          new_cluster: ${var.cluster4}
 variables:
-  cluster:
+  cluster1:
+    type: complex
+    description: "A cluster definition"
+  cluster2:
+    type: complex
+    description: "A cluster definition"
+  cluster3:
+    type: complex
+    description: "A cluster definition"
+  cluster4:
     type: complex
     description: "A cluster definition"
 
 include:
   - ./variables/*.yml
+
+
+targets:
+  default:
+  dev:
+    variables:
+      cluster3:
+        spark_version: "14.2.x-scala2.11"
+        node_type_id: "Standard_DS3_v2"
+        num_workers: 4
+        spark_conf:
+          spark.speculation: false
+          spark.databricks.delta.retentionDurationCheck.enabled: false
+      cluster4:
+        default:
+          spark_version: "14.2.x-scala2.11"
+          node_type_id: "Standard_DS3_v2"
+          num_workers: 4
+          spark_conf:
+            spark.speculation: false
+            spark.databricks.delta.retentionDurationCheck.enabled: false

bundle/tests/variables/complex_multiple_files/variables/clusters.yml

@@ -2,10 +2,18 @@ targets:
   default:
   dev:
     variables:
-      cluster:
+      cluster1:
         spark_version: "14.2.x-scala2.11"
         node_type_id: "Standard_DS3_v2"
         num_workers: 4
         spark_conf:
           spark.speculation: false
           spark.databricks.delta.retentionDurationCheck.enabled: false
+      cluster2:
+        default:
+          spark_version: "14.2.x-scala2.11"
+          node_type_id: "Standard_DS3_v2"
+          num_workers: 4
+          spark_conf:
+            spark.speculation: false
+            spark.databricks.delta.retentionDurationCheck.enabled: false