From 16a503f505dbc63a2d62ebecea35a9feaf144467 Mon Sep 17 00:00:00 2001
From: Burak Karakan <burak.karakan@gmail.com>
Date: Fri, 28 Feb 2025 21:44:08 +0300
Subject: [PATCH] do not parse the whole pipeline for asset parsing

---
 cmd/internal.go | 39 +++++++++++++++++++++++++++++++++++----
 1 file changed, 35 insertions(+), 4 deletions(-)

diff --git a/cmd/internal.go b/cmd/internal.go
index de1ae59b..cb5e9b56 100644
--- a/cmd/internal.go
+++ b/cmd/internal.go
@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"path/filepath"
 
 	"github.com/bruin-data/bruin/pkg/config"
 	"github.com/bruin-data/bruin/pkg/git"
@@ -256,26 +257,56 @@ func (r *ParseCommand) Run(assetPath string, lineage bool) error {
 		return cli.Exit("", 1)
 	}
 
-	pipelinePath, err := path.GetPipelineRootFromTask(assetPath, pipelineDefinitionFiles)
+	absoluteAssetPath, err := filepath.Abs(assetPath)
 	if err != nil {
 		printErrorJSON(err)
 		return cli.Exit("", 1)
 	}
 
-	repoRoot, err := git.FindRepoFromPath(assetPath)
+	pipelinePath, err := path.GetPipelineRootFromTask(absoluteAssetPath, pipelineDefinitionFiles)
 	if err != nil {
 		printErrorJSON(err)
 		return cli.Exit("", 1)
 	}
 
-	foundPipeline, err := DefaultPipelineBuilder.CreatePipelineFromPath(pipelinePath, true)
+	repoRoot, err := git.FindRepoFromPath(absoluteAssetPath)
+	if err != nil {
+		printErrorJSON(err)
+		return cli.Exit("", 1)
+	}
+
+	pipelineDefinitionPath, err := getPipelineDefinitionFullPath(pipelinePath)
+	if err != nil {
+		printErrorJSON(err)
+		return cli.Exit("", 1)
+	}
+
+	var foundPipeline *pipeline.Pipeline
+
+	// column-level lineage requires the whole pipeline to be parsed by nature, therefore we need to use the default pipeline builder.
+	// however, since the primary usecase of this command requires speed, we'll use a faster alternative if there's no lineage requested.
+	if lineage {
+		foundPipeline, err = DefaultPipelineBuilder.CreatePipelineFromPath(pipelineDefinitionPath, true)
+	} else {
+		foundPipeline, err = pipeline.PipelineFromPath(pipelineDefinitionPath, afero.NewOsFs())
+	}
+
 	if err != nil {
 		printErrorJSON(err)
+		return cli.Exit("", 1)
+	}
 
+	asset, err := DefaultPipelineBuilder.CreateAssetFromFile(absoluteAssetPath, nil)
+	if err != nil {
+		printErrorJSON(err)
 		return cli.Exit("", 1)
 	}
 
-	asset := foundPipeline.GetAssetByPath(assetPath)
+	asset, err = DefaultPipelineBuilder.MutateAsset(asset, foundPipeline)
+	if err != nil {
+		printErrorJSON(err)
+		return cli.Exit("", 1)
+	}
 
 	if lineage {
 		panics := lineageWg.WaitAndRecover()