Improve dbt errors

cmd/lakectl/cmd/dbt.go

@@ -135,8 +135,9 @@ func CopyModels(ctx context.Context, models []model, continueOnError bool, fromS
 				return err
 			}
 			fmt.Printf("copy %s.%s -> %s.%s  failed: %s\n", m.Schema, m.Alias, toSchema, m.Alias, err)
+		} else {
+			fmt.Printf("copied %s.%s -> %s.%s\n", m.Schema, m.Alias, toSchema, m.Alias)
 		}
-		fmt.Printf("copied %s.%s -> %s.%s\n", m.Schema, m.Alias, toSchema, m.Alias)
 	}
 	return nil
 }
@@ -146,7 +147,7 @@ func getDbtTables(projectRoot string) ([]model, error) {
 	dbtCmd.Dir = projectRoot
 	output, err := dbtCmd.Output()
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("failed to run dbt debug with output:\n %s\n and error: %s", output, err)
 	}
 	models := make([]model, 0)
 	scan := bufio.NewScanner(bytes.NewReader(output))
@@ -155,7 +156,7 @@ func getDbtTables(projectRoot string) ([]model, error) {
 		var m model
 		err = json.Unmarshal(line, &m)
 		if err != nil {
-			return nil, err
+			return nil, fmt.Errorf("failed to unmarshal json: %w", err)
 		}
 		models = append(models, m)
 	}
@@ -168,7 +169,7 @@ func dbtDebug(projectRoot string) string {
 	dbtCmd.Dir = projectRoot
 	output, err := dbtCmd.Output()
 	if err != nil {
-		DieErr(err)
+		DieFmt("failed to run dbt debug with output:\n %s\n and error: %s", output, err)
 	}
 	submatch := schemaRegex.FindSubmatch(output)
 	if submatch == nil || len(submatch) < 2 {

pkg/metastore/ms_client.go

@@ -44,7 +44,11 @@ type Client interface {
 func CopyOrMerge(ctx context.Context, fromClient, toClient Client, fromDB, fromTable, toDB, toTable, toBranch, serde string, partition []string, fixSparkPlaceHolder bool, dbfsLocation string) error {
 	transformLocation := func(location string) (string, error) {
 		location = HandleDBFSLocation(location, dbfsLocation)
-		return ReplaceBranchName(location, toBranch)
+		transformedLocation, err := ReplaceBranchName(location, toBranch)
+		if err != nil {
+			return "", fmt.Errorf("failed to replace branch name with location: %s and branch: %s : %w", location, toBranch, err)
+		}
+		return transformedLocation, nil
 	}
 	return copyOrMergeWithTransformLocation(ctx, fromClient, toClient, fromDB, fromTable, toDB, toTable, serde, false, partition, transformLocation, fixSparkPlaceHolder)
 }
@@ -55,22 +59,30 @@ func CopyDB(ctx context.Context, fromClient, toClient Client, fromDB, toDB, toBr
 			return "", nil
 		}
 		location = HandleDBFSLocation(location, dbfsLocation)
-		return ReplaceBranchName(location, toBranch)
+		transformedLocation, err := ReplaceBranchName(location, toBranch)
+		if err != nil {
+			return "", fmt.Errorf("failed to replace branch name with location: %s and branch: %s : %w", location, toBranch, err)
+		}
+		return transformedLocation, nil
 	}
 	return copyDBWithTransformLocation(ctx, fromClient, toClient, fromDB, toDB, transformLocation)
 }
 
 func copyDBWithTransformLocation(ctx context.Context, fromClient, toClient Client, fromDB string, toDB string, transformLocation func(location string) (string, error)) error {
 	schema, err := fromClient.GetDatabase(ctx, fromDB)
 	if err != nil {
-		return err
+		return fmt.Errorf("failed to get database on copy: %w", err)
 	}
 	schema.Name = toDB
 	schema.LocationURI, err = transformLocation(schema.LocationURI)
 	if err != nil {
 		return err
 	}
-	return toClient.CreateDatabase(ctx, schema)
+	err = toClient.CreateDatabase(ctx, schema)
+	if err != nil {
+		return fmt.Errorf("failed to create database with name %s and location %s   :%w", schema.Name, schema.LocationURI, err)
+	}
+	return nil
 }
 
 func copyOrMergeWithTransformLocation(ctx context.Context, fromClient, toClient Client, fromDB, fromTable, toDB, toTable, serde string, setSymlink bool, partition []string, transformLocation func(location string) (string, error), fixSparkPlaceHolder bool) error {
@@ -146,10 +158,12 @@ func CopyOrMergeAll(ctx context.Context, fromClient, toClient Client, schemaFilt
 
 // HandleDBFSLocation translates Data Bricks File system path to the S3 path using the dbfsLocation
 func HandleDBFSLocation(location string, dbfsLocation string) string {
+	l := location
 	if dbfsLocation != "" && strings.HasPrefix(location, dbfsPrefix) {
-		location = strings.Replace(location, dbfsPrefix, dbfsLocation, 1)
+		l = strings.Replace(location, dbfsPrefix, dbfsLocation, 1)
 	}
-	return location
+	logging.Default().WithFields(logging.Fields{"dbfsLocation": dbfsLocation, "location": location, "new_location": l}).Info("translate databricks file system path to s3 path")
+	return l
 }
 
 func ImportAll(ctx context.Context, fromClient, toClient Client, schemaFilter, tableFilter, repo, toBranch string, continueOnError, fixSparkPlaceHolder bool, dbfsLocation string) error {