Exporter: Add UC support for DLT pipelines (#3494)

Now `catalog` and `target` fields are now resolved into actual references to UC
objects (catalogs & schemas)

Author: alexott

exporter/exporter_test.go

@@ -2472,7 +2472,10 @@ func TestIncrementalDLTAndMLflowWebhooks(t *testing.T) {
 					PipelineID:   "def",
 					Name:         "def",
 					LastModified: 1690156900000,
-					Spec:         &pipelines.PipelineSpec{},
+					Spec: &pipelines.PipelineSpec{
+						Target:  "default",
+						Catalog: "main",
+					},
 				},
 				ReuseRequest: true,
 			},

exporter/importables.go

@@ -1919,6 +1919,12 @@ var resourcesMap map[string]importable = map[string]importable{
 			var pipeline pipelines.PipelineSpec
 			s := ic.Resources["databricks_pipeline"].Schema
 			common.DataToStructPointer(r.Data, s, &pipeline)
+			if pipeline.Catalog != "" && pipeline.Target != "" {
+				ic.Emit(&resource{
+					Resource: "databricks_schema",
+					ID:       pipeline.Catalog + "." + pipeline.Target,
+				})
+			}
 			for _, lib := range pipeline.Libraries {
 				if lib.Notebook != nil {
 					ic.emitNotebookOrRepo(lib.Notebook.Path)
@@ -1983,6 +1989,9 @@ var resourcesMap map[string]importable = map[string]importable{
 			return numLibraries == 0
 		},
 		Depends: []reference{
+			{Path: "catalog", Resource: "databricks_catalog"},
+			{Path: "target", Resource: "databricks_schema", Match: "name",
+				IsValidApproximation: dltIsMatchingCatalogAndSchema, SkipDirectLookup: true},
 			{Path: "cluster.aws_attributes.instance_profile_arn", Resource: "databricks_instance_profile"},
 			{Path: "cluster.init_scripts.dbfs.destination", Resource: "databricks_dbfs_file", Match: "dbfs_path"},
 			{Path: "cluster.init_scripts.volumes.destination", Resource: "databricks_file"},

exporter/util.go

@@ -1477,3 +1477,21 @@ func (ic *importContext) emitWorkspaceObjectParentDirectory(r *resource) {
 		r.AddExtraData(ParentDirectoryExtraKey, directoryPath)
 	}
 }
+
+func dltIsMatchingCatalogAndSchema(ic *importContext, res *resource, ra *resourceApproximation, origPath string) bool {
+	res_catalog_name := res.Data.Get("catalog").(string)
+	if res_catalog_name == "" {
+		return false
+	}
+	res_schema_name := res.Data.Get("target").(string)
+	ra_catalog_name, cat_found := ra.Get("catalog_name")
+	ra_schema_name, schema_found := ra.Get("name")
+	if !cat_found || !schema_found {
+		log.Printf("[WARN] Can't find attributes in approximation: %s %s, catalog='%v' (found? %v) schema='%v' (found? %v). Resource: %s, catalog='%s', schema='%s'",
+			ra.Type, ra.Name, ra_catalog_name, cat_found, ra_schema_name, schema_found, res.Resource, res_catalog_name, res_schema_name)
+		return true
+	}
+
+	result := ra_catalog_name.(string) == res_catalog_name && ra_schema_name.(string) == res_schema_name
+	return result
+}