Repository: dbt-labs/dbt-snowflake (python)

Issue:
[Feature] Iceberg Custom base_location
### Is this your first time submitting a feature request?

- [X] I have read the [expectations for open source contributors](https://docs.getdbt.com/docs/contributing/oss-expectations)
- [X] I have searched the existing issues, and I could not find an existing issue for this feature
- [X] I am requesting a straightforward extension of existing dbt-snowflake functionality, rather than a Big Idea better suited to a discussion

### Describe the feature

Allow users to overwrite the default `base_location`. Not just add a `base_location_subpath`

### Describe alternatives you've considered

_No response_

### Who will this benefit?

Teams using dbt Snowflake Iceberg models who need to change the storage location in S3

### Are you interested in contributing this feature?

Happy to!

### Anything else?

Docs on current configuration: https://docs.getdbt.com/reference/resource-configs/snowflake-configs#base-location-subpath

Pull request:
Feature: Custom Iceberg `base_location_root`
resolves #1284 
[Related docs](https://docs.getdbt.com/reference/resource-configs/snowflake-configs#base-location)

<!---
  Include the number of the issue addressed by this PR above if applicable.
  PRs for code changes without an associated issue *will not be merged*.
  See CONTRIBUTING.md for more information.

  Include the number of the docs issue that was opened for this PR. If
  this change has no user-facing implications, "N/A" suffices instead. New
  docs tickets can be created by clicking the link above or by going to
  https://github.com/dbt-labs/docs.getdbt.com/issues/new/choose.
-->

### Problem
<!---
  Describe the problem this PR is solving. What is the application state
  before this PR is merged?
-->
Currently, we only allow users to append a `base_location_subpath` to a `base_location` when creating a Snowflake Iceberg table. We've hard coded in the `base_location` root to be `_dbt/<schema-name>/<table-name>`.

Some customers have requested we allow overrides of the top-level `_dbt/` directory. After discussing with product (@amychen1776), we've decided to add support for this.

----

### Solution
<!---
  Describe the way this PR solves the above problem. Add as much detail as you
  can to help reviewers understand your changes. Include any alternatives and
  tradeoffs you considered.
-->
Add a new **optional** config for Snowflake called `base_location_root`. If this is provided, it will override the default dbt `base_location` value (`_dbt/<schema_name>/<table_name>/`) to `base_location_root/<schema_name>/<table_name>/`)

### Examples from test runs:
Configs for `_MODEL_BASIC_ICEBERG_MODEL`: `base_location_subpath="subpath",` (no `base_location_root`). Snowflake ddl generated:
```
create or replace iceberg table analytics.test17370768401216806666_test_table_basic.iceberg_table
 ...
        base_location = '_dbt/test17370768401216806666_test_table_basic/iceberg_table/subpath'
...
```

Configs for `_MODEL_BASIC_ICEBERG_MODEL_WITH_PATH`: ` base_location_root="root_path",` (no `base_location_subpath`). Snowflake ddl generated:
```
create or replace iceberg table analytics.test17370768401216806666_test_table_basic.iceberg_tableb
...
      base_location = 'root_path/test17370768401216806666_test_table_basic/iceberg_tableb'
....
```

Configs for `_MODEL_BASIC_ICEBERG_MODEL_WITH_PATH_SUBPATH`: ` base_location_root="root_path", base_location_subpath="subpath",`. Snowflake ddl generated:
```
create or replace iceberg table analytics.test17370768401216806666_test_table_basic.iceberg_tablec
...
     base_location = 'root_path/test17370768401216806666_test_table_basic/iceberg_tablec/subpath'
...
```

### Validation of Tests
See passing functional & unit tests in CI

---
### Checklist

- [x] I have read [the contributing guide](https://github.com/dbt-labs/dbt-core/blob/main/CONTRIBUTING.md) and understand what's expected of me
- [x] I have run this code in development and it appears to resolve the stated issue
- [x] This PR includes tests, or tests are not required/relevant for this PR
- [x] This PR has already received feedback and approval from Product or DX
@amychen1776 and I have discussed and aligned on these changes.


Patch:
diff --git a/.changes/unreleased/Features-20250113-133414.yaml b/.changes/unreleased/Features-20250113-133414.yaml
new file mode 100644
index 0000000..869ed6b
--- /dev/null
+++ b/.changes/unreleased/Features-20250113-133414.yaml
@@ -0,0 +1,6 @@
+kind: Features
+body: Added support for custom iceberg base_location_root
+time: 2025-01-13T13:34:14.326047-08:00
+custom:
+    Author: LProcopi15
+    Issue: "1284"
diff --git a/dbt/adapters/snowflake/impl.py b/dbt/adapters/snowflake/impl.py
index 7ccff9f..6ae8ef1 100644
--- a/dbt/adapters/snowflake/impl.py
+++ b/dbt/adapters/snowflake/impl.py
@@ -54,6 +54,7 @@ class SnowflakeConfig(AdapterConfig):
     # extended formats
     table_format: Optional[str] = None
     external_volume: Optional[str] = None
+    base_location_root: Optional[str] = None
     base_location_subpath: Optional[str] = None
 
 
diff --git a/dbt/adapters/snowflake/relation.py b/dbt/adapters/snowflake/relation.py
index 54db219..f3ee3e5 100644
--- a/dbt/adapters/snowflake/relation.py
+++ b/dbt/adapters/snowflake/relation.py
@@ -204,7 +204,10 @@ class SnowflakeRelation(BaseRelation):
             return ""
 
     def get_iceberg_ddl_options(self, config: RelationConfig) -> str:
-        base_location: str = f"_dbt/{self.schema}/{self.name}"
+        # If the base_location_root config is supplied, overwrite the default value ("_dbt/")
+        base_location: str = (
+            f"{config.get('base_location_root', '_dbt')}/{self.schema}/{self.name}"
+        )
 
         if subpath := config.get("base_location_subpath"):
             base_location += f"/{subpath}"


Test patch:
diff --git a/tests/functional/iceberg/models.py b/tests/functional/iceberg/models.py
index 6433f74..e6da6ac 100644
--- a/tests/functional/iceberg/models.py
+++ b/tests/functional/iceberg/models.py
@@ -23,6 +23,37 @@ _MODEL_BASIC_ICEBERG_MODEL = """
 select * from {{ ref('first_table') }}
 """
 
+_MODEL_BASIC_ICEBERG_MODEL_WITH_PATH = """
+{{
+  config(
+    transient = "true",
+    materialized = "table",
+    cluster_by=['id'],
+    table_format="iceberg",
+    external_volume="s3_iceberg_snow",
+    base_location_root="root_path",
+  )
+}}
+
+select * from {{ ref('first_table') }}
+"""
+
+_MODEL_BASIC_ICEBERG_MODEL_WITH_PATH_SUBPATH = """
+{{
+  config(
+    transient = "true",
+    materialized = "table",
+    cluster_by=['id'],
+    table_format="iceberg",
+    external_volume="s3_iceberg_snow",
+    base_location_root="root_path",
+    base_location_subpath="subpath",
+  )
+}}
+
+select * from {{ ref('first_table') }}
+"""
+
 _MODEL_BASIC_DYNAMIC_TABLE_MODEL = """
 {{ config(
     materialized='dynamic_table',
@@ -36,6 +67,38 @@ _MODEL_BASIC_DYNAMIC_TABLE_MODEL = """
 select * from {{ ref('first_table') }}
 """
 
+_MODEL_BASIC_DYNAMIC_TABLE_MODEL_WITH_PATH = """
+{{
+  config(
+    transient = "transient",
+    materialized = "dynamic_table",
+    cluster_by=['id'],
+    table_format="iceberg",
+    external_volume="s3_iceberg_snow",
+    base_location_root="root_path",
+  )
+}}
+
+select * from {{ ref('first_table') }}
+"""
+
+_MODEL_BASIC_DYNAMIC_TABLE_MODEL_WITH_PATH_SUBPATH = """
+{{
+  config(
+    transient = "true",
+    materialized = "dynamic_table",
+    cluster_by=['id'],
+    table_format="iceberg",
+    external_volume="s3_iceberg_snow",
+    base_location_root="root_path",
+    base_location_subpath='subpath',
+  )
+}}
+
+select * from {{ ref('first_table') }}
+"""
+
+
 _MODEL_BASIC_DYNAMIC_TABLE_MODEL_WITH_SUBPATH = """
 {{ config(
     materialized='dynamic_table',
diff --git a/tests/functional/iceberg/test_table_basic.py b/tests/functional/iceberg/test_table_basic.py
index e835a5f..faf4b34 100644
--- a/tests/functional/iceberg/test_table_basic.py
+++ b/tests/functional/iceberg/test_table_basic.py
@@ -7,7 +7,11 @@ from dbt.tests.util import run_dbt, rm_file, write_file
 from tests.functional.iceberg.models import (
     _MODEL_BASIC_TABLE_MODEL,
     _MODEL_BASIC_ICEBERG_MODEL,
+    _MODEL_BASIC_ICEBERG_MODEL_WITH_PATH,
+    _MODEL_BASIC_ICEBERG_MODEL_WITH_PATH_SUBPATH,
     _MODEL_BASIC_DYNAMIC_TABLE_MODEL,
+    _MODEL_BASIC_DYNAMIC_TABLE_MODEL_WITH_PATH,
+    _MODEL_BASIC_DYNAMIC_TABLE_MODEL_WITH_PATH_SUBPATH,
     _MODEL_BASIC_DYNAMIC_TABLE_MODEL_WITH_SUBPATH,
     _MODEL_BUILT_ON_ICEBERG_TABLE,
     _MODEL_TABLE_BEFORE_SWAP,
@@ -26,14 +30,18 @@ class TestIcebergTableBuilds:
         return {
             "first_table.sql": _MODEL_BASIC_TABLE_MODEL,
             "iceberg_table.sql": _MODEL_BASIC_ICEBERG_MODEL,
+            "iceberg_tableb.sql": _MODEL_BASIC_ICEBERG_MODEL_WITH_PATH,
+            "iceberg_tablec.sql": _MODEL_BASIC_ICEBERG_MODEL_WITH_PATH_SUBPATH,
             "table_built_on_iceberg_table.sql": _MODEL_BUILT_ON_ICEBERG_TABLE,
             "dynamic_table.sql": _MODEL_BASIC_DYNAMIC_TABLE_MODEL,
-            "dynamic_tableb.sql": _MODEL_BASIC_DYNAMIC_TABLE_MODEL_WITH_SUBPATH,
+            "dynamic_tableb.sql": _MODEL_BASIC_DYNAMIC_TABLE_MODEL_WITH_PATH,
+            "dynamic_tablec.sql": _MODEL_BASIC_DYNAMIC_TABLE_MODEL_WITH_PATH_SUBPATH,
+            "dynamic_tabled.sql": _MODEL_BASIC_DYNAMIC_TABLE_MODEL_WITH_SUBPATH,
         }
 
     def test_iceberg_tables_build_and_can_be_referred(self, project):
         run_results = run_dbt()
-        assert len(run_results) == 5
+        assert len(run_results) == 9
 
 
 class TestIcebergTableTypeBuildsOnExistingTable:
diff --git a/tests/unit/test_iceberg_location.py b/tests/unit/test_iceberg_location.py
new file mode 100644
index 0000000..dca82b4
--- /dev/null
+++ b/tests/unit/test_iceberg_location.py
@@ -0,0 +1,79 @@
+import pytest
+from dbt.adapters.snowflake.relation import SnowflakeRelation
+
+
+@pytest.fixture
+def iceberg_config() -> dict:
+    """Fixture providing standard Iceberg configuration."""
+    return {
+        "schema": "my_schema",
+        "identifier": "my_table",
+        "external_volume": "s3_iceberg_snow",
+        "base_location_root": "root_path",
+        "base_location_subpath": "subpath",
+    }
+
+
+def get_actual_base_location(config: dict[str, str]) -> str:
+    """Get the actual base location from the configuration by parsing the DDL predicates."""
+
+    relation = SnowflakeRelation.create(
+        schema=config["schema"],
+        identifier=config["identifier"],
+    )
+
+    actual_ddl_predicates = relation.get_iceberg_ddl_options(config).strip()
+    actual_base_location = actual_ddl_predicates.split("base_location = ")[1]
+
+    return actual_base_location
+
+
+def test_iceberg_path_and_subpath(iceberg_config: dict[str, str]):
+    """Test when base_location_root and base_location_subpath are provided"""
+    expected_base_location = (
+        f"'{iceberg_config['base_location_root']}/"
+        f"{iceberg_config['schema']}/"
+        f"{iceberg_config['identifier']}/"
+        f"{iceberg_config['base_location_subpath']}'"
+    ).strip()
+
+    assert get_actual_base_location(iceberg_config) == expected_base_location
+
+
+def test_iceberg_only_subpath(iceberg_config: dict[str, str]):
+    """Test when only base_location_subpath is provided"""
+    del iceberg_config["base_location_root"]
+
+    expected_base_location = (
+        f"'_dbt/"
+        f"{iceberg_config['schema']}/"
+        f"{iceberg_config['identifier']}/"
+        f"{iceberg_config['base_location_subpath']}'"
+    ).strip()
+
+    assert get_actual_base_location(iceberg_config) == expected_base_location
+
+
+def test_iceberg_only_path(iceberg_config: dict[str, str]):
+    """Test when only base_location_root is provided"""
+    del iceberg_config["base_location_subpath"]
+
+    expected_base_location = (
+        f"'{iceberg_config['base_location_root']}/"
+        f"{iceberg_config['schema']}/"
+        f"{iceberg_config['identifier']}'"
+    ).strip()
+
+    assert get_actual_base_location(iceberg_config) == expected_base_location
+
+
+def test_iceberg_no_path(iceberg_config: dict[str, str]):
+    """Test when no base_location_root or is base_location_subpath provided"""
+    del iceberg_config["base_location_root"]
+    del iceberg_config["base_location_subpath"]
+
+    expected_base_location = (
+        f"'_dbt/" f"{iceberg_config['schema']}/" f"{iceberg_config['identifier']}'"
+    ).strip()
+
+    assert get_actual_base_location(iceberg_config) == expected_base_location