[BACKPORT 1.5.latest] Target path should be relative to project dir, rather than current working directory

.changes/unreleased/Fixes-20230525-165053.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Incorrect paths used for "target" and "state" directories
+time: 2023-05-25T16:50:53.718564-04:00
+custom:
+  Author: gshank
+  Issue: "7465"

core/dbt/cli/params.py

@@ -424,7 +424,7 @@
         dir_okay=True,
         file_okay=False,
         readable=True,
-        resolve_path=True,
+        resolve_path=False,
         path_type=Path,
     ),
 )

core/dbt/cli/requires.py

@@ -245,7 +245,7 @@ def wrapper(*args, **kwargs):
 
                 ctx.obj["manifest"] = manifest
                 if write and ctx.obj["flags"].write_json:
-                    write_manifest(manifest, ctx.obj["runtime_config"].target_path)
+                    write_manifest(manifest, ctx.obj["runtime_config"].project_target_path)
 
             return func(*args, **kwargs)
 

core/dbt/compilation.py

@@ -167,7 +167,7 @@ def __init__(self, config):
         self.config = config
 
     def initialize(self):
-        make_directory(self.config.target_path)
+        make_directory(self.config.project_target_path)
         make_directory(self.config.packages_install_path)
 
     # creates a ModelContext which is converted to
@@ -387,7 +387,7 @@ def _compile_code(
 
     def write_graph_file(self, linker: Linker, manifest: Manifest):
         filename = graph_file_name
-        graph_path = os.path.join(self.config.target_path, filename)
+        graph_path = os.path.join(self.config.project_target_path, filename)
         flags = get_flags()
         if flags.WRITE_JSON:
             linker.write_graph(graph_path, manifest)
@@ -506,9 +506,8 @@ def _write_node(self, node: ManifestSQLNode) -> ManifestSQLNode:
         fire_event(WritingInjectedSQLForNode(node_info=get_node_info()))
 
         if node.compiled_code:
-            node.compiled_path = node.write_node(
-                self.config.target_path, "compiled", node.compiled_code
-            )
+            node.compiled_path = node.get_target_write_path(self.config.target_path, "compiled")
+            node.write_node(self.config.project_root, node.compiled_path, node.compiled_code)
         return node
 
     def compile_node(

core/dbt/config/project.py

@@ -700,3 +700,8 @@ def get_macro_search_order(self, macro_namespace: str):
             if dispatch_entry["macro_namespace"] == macro_namespace:
                 return dispatch_entry["search_order"]
         return None
+
+    @property
+    def project_target_path(self):
+        # If target_path is absolute, project_root will not be included
+        return os.path.join(self.project_root, self.target_path)

core/dbt/context/providers.py

@@ -806,7 +806,8 @@ def write(self, payload: str) -> str:
         # macros/source defs aren't 'writeable'.
         if isinstance(self.model, (Macro, SourceDefinition)):
             raise MacrosSourcesUnWriteableError(node=self.model)
-        self.model.build_path = self.model.write_node(self.config.target_path, "run", payload)
+        self.model.build_path = self.model.get_target_write_path(self.config.target_path, "run")
+        self.model.write_node(self.config.project_root, self.model.build_path, payload)
         return ""
 
     @contextmember

core/dbt/contracts/graph/nodes.py

@@ -323,17 +323,23 @@ class ParsedNode(NodeInfoMixin, ParsedNodeMandatory, SerializableType):
     relation_name: Optional[str] = None
     raw_code: str = ""
 
-    def write_node(self, target_path: str, subdirectory: str, payload: str):
+    def get_target_write_path(self, target_path: str, subdirectory: str):
+        # This is called for both the "compiled" subdirectory of "target" and the "run" subdirectory
         if os.path.basename(self.path) == os.path.basename(self.original_file_path):
             # One-to-one relationship of nodes to files.
             path = self.original_file_path
         else:
             #  Many-to-one relationship of nodes to files.
             path = os.path.join(self.original_file_path, self.path)
-        full_path = os.path.join(target_path, subdirectory, self.package_name, path)
+        target_write_path = os.path.join(target_path, subdirectory, self.package_name, path)
+        return target_write_path
 
-        write_file(full_path, payload)
-        return full_path
+    def write_node(self, project_root: str, compiled_path, compiled_code: str):
+        if os.path.isabs(compiled_path):
+            full_path = compiled_path
+        else:
+            full_path = os.path.join(project_root, compiled_path)
+        write_file(full_path, compiled_code)
 
     def _serialize(self):
         return self.to_dict()

core/dbt/contracts/state.py

@@ -7,31 +7,33 @@
 
 
 class PreviousState:
-    def __init__(self, path: Path, current_path: Path):
-        self.path: Path = path
-        self.current_path: Path = current_path
+    def __init__(self, state_path: Path, target_path: Path, project_root: Path):
+        self.state_path: Path = state_path
+        self.target_path: Path = target_path
+        self.project_root: Path = project_root
         self.manifest: Optional[WritableManifest] = None
         self.results: Optional[RunResultsArtifact] = None
         self.sources: Optional[FreshnessExecutionResultArtifact] = None
         self.sources_current: Optional[FreshnessExecutionResultArtifact] = None
 
-        manifest_path = self.path / "manifest.json"
+        # Note: if state_path is absolute, project_root will be ignored.
+        manifest_path = self.project_root / self.state_path / "manifest.json"
         if manifest_path.exists() and manifest_path.is_file():
             try:
                 self.manifest = WritableManifest.read_and_check_versions(str(manifest_path))
             except IncompatibleSchemaError as exc:
                 exc.add_filename(str(manifest_path))
                 raise
 
-        results_path = self.path / "run_results.json"
+        results_path = self.project_root / self.state_path / "run_results.json"
         if results_path.exists() and results_path.is_file():
             try:
                 self.results = RunResultsArtifact.read_and_check_versions(str(results_path))
             except IncompatibleSchemaError as exc:
                 exc.add_filename(str(results_path))
                 raise
 
-        sources_path = self.path / "sources.json"
+        sources_path = self.project_root / self.state_path / "sources.json"
         if sources_path.exists() and sources_path.is_file():
             try:
                 self.sources = FreshnessExecutionResultArtifact.read_and_check_versions(
@@ -41,7 +43,7 @@ def __init__(self, path: Path, current_path: Path):
                 exc.add_filename(str(sources_path))
                 raise
 
-        sources_current_path = self.current_path / "sources.json"
+        sources_current_path = self.project_root / self.target_path / "sources.json"
         if sources_current_path.exists() and sources_current_path.is_file():
             try:
                 self.sources_current = FreshnessExecutionResultArtifact.read_and_check_versions(

core/dbt/parser/manifest.py

@@ -238,7 +238,7 @@ def get_full_manifest(
             loader.track_project_load()
 
             if write_perf_info:
-                loader.write_perf_info(config.target_path)
+                loader.write_perf_info(config.project_target_path)
 
         return manifest
 
@@ -581,9 +581,7 @@ def macro_depends_on(self):
                     macro.depends_on.add_macro(dep_macro_id)  # will check for dupes
 
     def write_manifest_for_partial_parse(self):
-        path = os.path.join(
-            self.root_project.project_root, self.root_project.target_path, PARTIAL_PARSE_FILE_NAME
-        )
+        path = os.path.join(self.root_project.project_target_path, PARTIAL_PARSE_FILE_NAME)
         try:
             # This shouldn't be necessary, but we have gotten bug reports (#3757) of the
             # saved manifest not matching the code version.
@@ -688,9 +686,7 @@ def read_manifest_for_partial_parse(self) -> Optional[Manifest]:
         if not get_flags().PARTIAL_PARSE:
             fire_event(PartialParsingNotEnabled())
             return None
-        path = os.path.join(
-            self.root_project.project_root, self.root_project.target_path, PARTIAL_PARSE_FILE_NAME
-        )
+        path = os.path.join(self.root_project.project_target_path, PARTIAL_PARSE_FILE_NAME)
 
         reparse_reason = None
 

core/dbt/task/compile.py

@@ -125,7 +125,7 @@ def defer_to_manifest(self, adapter, selected_uids: AbstractSet[str]):
             favor_state=bool(self.args.favor_state),
         )
         # TODO: is it wrong to write the manifest here? I think it's right...
-        write_manifest(self.manifest, self.config.target_path)
+        write_manifest(self.manifest, self.config.project_target_path)
 
     def _runtime_initialize(self):
         if getattr(self.args, "inline", None):

core/dbt/task/freshness.py

@@ -160,7 +160,7 @@ def result_path(self):
         if self.args.output:
             return os.path.realpath(self.args.output)
         else:
-            return os.path.join(self.config.target_path, RESULT_FILE_NAME)
+            return os.path.join(self.config.project_target_path, RESULT_FILE_NAME)
 
     def raise_on_first_error(self):
         return False

core/dbt/task/generate.py

@@ -213,10 +213,12 @@ def run(self) -> CatalogArtifact:
                     compile_results=compile_results,
                 )
 
-        shutil.copyfile(DOCS_INDEX_FILE_PATH, os.path.join(self.config.target_path, "index.html"))
+        shutil.copyfile(
+            DOCS_INDEX_FILE_PATH, os.path.join(self.config.project_target_path, "index.html")
+        )
 
         for asset_path in self.config.asset_paths:
-            to_asset_path = os.path.join(self.config.target_path, asset_path)
+            to_asset_path = os.path.join(self.config.project_target_path, asset_path)
 
             if os.path.exists(to_asset_path):
                 shutil.rmtree(to_asset_path)
@@ -252,10 +254,10 @@ def run(self) -> CatalogArtifact:
             errors=errors,
         )
 
-        path = os.path.join(self.config.target_path, CATALOG_FILENAME)
+        path = os.path.join(self.config.project_target_path, CATALOG_FILENAME)
         results.write(path)
         if self.args.compile:
-            write_manifest(self.manifest, self.config.target_path)
+            write_manifest(self.manifest, self.config.project_target_path)
 
         if exceptions:
             fire_event(WriteCatalogFailure(num_exceptions=len(exceptions)))

core/dbt/task/runnable.py

@@ -79,7 +79,9 @@ def __init__(self, args, config, manifest):
     def set_previous_state(self):
         if self.args.state is not None:
             self.previous_state = PreviousState(
-                path=self.args.state, current_path=Path(self.config.target_path)
+                state_path=self.args.state,
+                target_path=Path(self.config.target_path),
+                project_root=Path(self.config.project_root),
             )
 
     def index_offset(self, value: int) -> int:
@@ -155,7 +157,7 @@ def get_runner_type(self, node):
         raise NotImplementedError("Not Implemented")
 
     def result_path(self):
-        return os.path.join(self.config.target_path, RESULT_FILE_NAME)
+        return os.path.join(self.config.project_target_path, RESULT_FILE_NAME)
 
     def get_runner(self, node):
         adapter = get_adapter(self.config)
@@ -453,7 +455,7 @@ def run(self):
             )
 
         if get_flags().WRITE_JSON:
-            write_manifest(self.manifest, self.config.target_path)
+            write_manifest(self.manifest, self.config.project_target_path)
             self.write_result(result)
 
         self.task_end_messages(result.results)

core/dbt/task/serve.py

@@ -12,7 +12,7 @@
 
 class ServeTask(ConfiguredTask):
     def run(self):
-        os.chdir(self.config.target_path)
+        os.chdir(self.config.project_target_path)
         shutil.copyfile(DOCS_INDEX_FILE_PATH, "index.html")
 
         port = self.args.port

test/unit/test_graph_selector_methods.py

@@ -1200,7 +1200,9 @@ def test_select_metric(manifest):
 def previous_state(manifest):
     writable = copy.deepcopy(manifest).writable_manifest()
     state = PreviousState(
-        path=Path("/path/does/not/exist"), current_path=Path("/path/does/not/exist")
+        state_path=Path("/path/does/not/exist"),
+        target_path=Path("/path/does/not/exist"),
+        project_root=Path("/path/does/not/exist"),
     )
     state.manifest = writable
     return state

tests/functional/configs/test_configs.py

@@ -58,11 +58,13 @@ def project_config_update(self):
         }
 
     def test_alternative_target_paths(self, project):
+        # chdir to a different directory to test creation of target directory under project_root
+        os.chdir(project.profiles_dir)
         run_dbt(["seed"])
 
         target_path = ""
-        for d in os.listdir("."):
-            if os.path.isdir(d) and d.startswith("target_"):
+        for d in os.listdir(project.project_root):
+            if os.path.isdir(os.path.join(project.project_root, d)) and d.startswith("target_"):
                 target_path = d
         assert os.path.exists(os.path.join(project.project_root, target_path, "manifest.json"))
 

tests/functional/defer_state/test_defer_state.py

@@ -77,12 +77,15 @@ def profiles_config_update(self, dbt_profile_target, unique_schema, other_schema
         outputs["otherschema"]["schema"] = other_schema
         return {"test": {"outputs": outputs, "target": "default"}}
 
-    def copy_state(self):
-        if not os.path.exists("state"):
-            os.makedirs("state")
-        shutil.copyfile("target/manifest.json", "state/manifest.json")
+    def copy_state(self, project_root):
+        state_path = os.path.join(project_root, "state")
+        if not os.path.exists(state_path):
+            os.makedirs(state_path)
+        shutil.copyfile(
+            f"{project_root}/target/manifest.json", f"{project_root}/state/manifest.json"
+        )
 
-    def run_and_save_state(self):
+    def run_and_save_state(self, project_root):
         results = run_dbt(["seed"])
         assert len(results) == 1
         assert not any(r.node.deferred for r in results)
@@ -93,7 +96,7 @@ def run_and_save_state(self):
         assert len(results) == 2
 
         # copy files
-        self.copy_state()
+        self.copy_state(project_root)
 
 
 class TestDeferStateUnsupportedCommands(BaseDeferState):
@@ -110,21 +113,24 @@ def test_no_state(self, project):
 
 class TestRunCompileState(BaseDeferState):
     def test_run_and_compile_defer(self, project):
-        self.run_and_save_state()
+        self.run_and_save_state(project.project_root)
 
         # defer test, it succeeds
+        # Change directory to ensure that state directory is underneath
+        # project directory.
+        os.chdir(project.profiles_dir)
         results = run_dbt(["compile", "--state", "state", "--defer"])
         assert len(results.results) == 6
         assert results.results[0].node.name == "seed"
 
 
 class TestSnapshotState(BaseDeferState):
     def test_snapshot_state_defer(self, project):
-        self.run_and_save_state()
+        self.run_and_save_state(project.project_root)
         # snapshot succeeds without --defer
         run_dbt(["snapshot"])
         # copy files
-        self.copy_state()
+        self.copy_state(project.project_root)
         # defer test, it succeeds
         run_dbt(["snapshot", "--state", "state", "--defer"])
         # favor_state test, it succeeds
@@ -134,7 +140,7 @@ def test_snapshot_state_defer(self, project):
 class TestRunDeferState(BaseDeferState):
     def test_run_and_defer(self, project, unique_schema, other_schema):
         project.create_test_schema(other_schema)
-        self.run_and_save_state()
+        self.run_and_save_state(project.project_root)
 
         # test tests first, because run will change things
         # no state, wrong schema, failure.
@@ -186,7 +192,7 @@ def test_run_and_defer(self, project, unique_schema, other_schema):
 
 class TestRunDeferStateChangedModel(BaseDeferState):
     def test_run_defer_state_changed_model(self, project):
-        self.run_and_save_state()
+        self.run_and_save_state(project.project_root)
 
         # change "view_model"
         write_file(changed_view_model_sql, "models", "view_model.sql")
@@ -215,7 +221,7 @@ def test_run_defer_state_changed_model(self, project):
 class TestRunDeferStateIFFNotExists(BaseDeferState):
     def test_run_defer_iff_not_exists(self, project, unique_schema, other_schema):
         project.create_test_schema(other_schema)
-        self.run_and_save_state()
+        self.run_and_save_state(project.project_root)
 
         results = run_dbt(["seed", "--target", "otherschema"])
         assert len(results) == 1
@@ -238,7 +244,7 @@ def test_run_defer_iff_not_exists(self, project, unique_schema, other_schema):
 class TestDeferStateDeletedUpstream(BaseDeferState):
     def test_run_defer_deleted_upstream(self, project, unique_schema, other_schema):
         project.create_test_schema(other_schema)
-        self.run_and_save_state()
+        self.run_and_save_state(project.project_root)
 
         # remove "ephemeral_model" + change "table_model"
         rm_file("models", "ephemeral_model.sql")