Pass down schema on table creation

dbt/include/dremio/macros/get_custom_name/get_custom_schema.sql

@@ -15,8 +15,6 @@ limitations under the License.*/
 {% macro dremio__generate_schema_name(custom_schema_name, node) -%}
   {%- set default_schema = target.schema if not is_datalake_node(node)
     else target.root_path -%}
-  {%- set custom_schema_name = custom_schema_name if not is_datalake_node(node)
-    else node.config.root_path -%}
   {{ generate_schema_name_impl(default_schema, custom_schema_name, node) }}
 {%- endmacro %}
 
@@ -25,9 +23,13 @@ limitations under the License.*/
 
       {{ default_schema }}
 
+  {%- elif default_schema == 'no_schema' -%}
+
+      {{ custom_schema_name}}
+
   {%- else -%}
 
-      {{ custom_schema_name }}
+      {{default_schema}}.{{ custom_schema_name }}
 
   {%- endif -%}
 {%- endmacro %}

dbt/include/dremio/macros/materializations/twin_strategy.sql

@@ -24,12 +24,13 @@ limitations under the License.*/
       {{ adapter.drop_relation(table_relation) }}
     {%- endif -%}
   {%- elif target_relation.type == 'table' -%}
-    {%- if twin_strategy in ['prevent', 'clone'] -%}
-      {%- set view_relation = api.Relation.create(
-          identifier=generate_alias_name_impl(model.name, config.get('alias', validator=validation.any[basestring]), model),
-          schema=generate_schema_name_impl(target.schema, config.get('schema', validator=validation.any[basestring]), model),
-          database=generate_database_name_impl(target.database, config.get('database', validator=validation.any[basestring]), model),
-          type='view') -%}
+    {%- set view_relation = api.Relation.create(
+        identifier=generate_alias_name_impl(model.name, config.get('alias', validator=validation.any[basestring]), model),
+        schema=generate_schema_name_impl(target.schema, config.get('schema', validator=validation.any[basestring]), model),
+        database=generate_database_name_impl(target.database, config.get('database', validator=validation.any[basestring]), model),
+        type='view') -%}
+    {%- set conflicting_view = adapter.get_relation(database=view_relation.database, schema=view_relation.schema, identifier=view_relation.identifier) -%}
+    {%- if conflicting_view is not none -%}
       {%- if twin_strategy == 'prevent' -%}
         {{ adapter.drop_relation(view_relation) }}
       {%- elif twin_strategy == 'clone' -%}

tests/functional/adapter/dremio_specific/test_nested_schema.py

@@ -0,0 +1,117 @@
+import pytest
+from dbt.tests.util import run_dbt, get_connection
+from tests.utils.util import relation_from_name
+
+
+custom_schema_table_no_schema_model = """
+{{ config(
+    materialized='table'
+) }}
+select 1 as id
+"""
+
+custom_schema_table_model = """
+{{ config(
+    materialized='table',
+    schema='schema'
+) }}
+select 1 as id
+"""
+
+custom_schema_table_nested_model = """
+{{ config(
+    materialized='table',
+    schema='nested.schema'
+) }}
+select 1 as id
+"""
+
+custom_schema_view_no_schema_model = """
+{{ config(
+    materialized='view'
+) }}
+select 1 as id
+"""
+
+custom_schema_view_model = """
+{{ config(
+    materialized='view',
+    schema='schema'
+) }}
+select 1 as id
+"""
+
+custom_schema_view_nested_model = """
+{{ config(
+    materialized='view',
+    schema='nested.schema'
+) }}
+select 1 as id
+"""
+
+
+class TestGetCustomSchema:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "custom_schema_table_no_schema.sql": custom_schema_table_no_schema_model,
+            "custom_schema_table.sql": custom_schema_table_model,
+            "custom_schema_table_nested.sql": custom_schema_table_nested_model,
+            "custom_schema_view_no_schema.sql": custom_schema_view_no_schema_model,
+            "custom_schema_view.sql": custom_schema_view_model,
+            "custom_schema_view_nested.sql": custom_schema_view_nested_model,
+        }
+
+    def test_custom_schema_table_no_schema(self, project):
+        run_dbt(["run", "--select", "custom_schema_table_no_schema"])
+        table_relation = relation_from_name(
+            project.adapter, "custom_schema_table_no_schema")
+        with get_connection(project.adapter):
+            columns = project.adapter.get_columns_in_relation(table_relation)
+        assert len(columns) == 1
+        assert columns[0].name == "id"
+
+    def test_custom_schema_table(self, project):
+        run_dbt(["run", "--select", "custom_schema_table"])
+        table_relation = relation_from_name(
+            project.adapter, "schema.custom_schema_table")
+        with get_connection(project.adapter):
+            columns = project.adapter.get_columns_in_relation(table_relation)
+        assert len(columns) == 1
+        assert columns[0].name == "id"
+
+    def test_custom_schema_table_nested(self, project):
+        run_dbt(["run", "--select", "custom_schema_table_nested"])
+        table_relation = relation_from_name(
+            project.adapter, "nested.schema.custom_schema_table_nested")
+        with get_connection(project.adapter):
+            columns = project.adapter.get_columns_in_relation(table_relation)
+        assert len(columns) == 1
+        assert columns[0].name == "id"
+
+    def test_custom_schema_view_no_schema(self, project):
+        run_dbt(["run", "--select", "custom_schema_view_no_schema"])
+        view_relation = relation_from_name(
+            project.adapter, "custom_schema_view_no_schema")
+        with get_connection(project.adapter):
+            columns = project.adapter.get_columns_in_relation(view_relation)
+        assert len(columns) == 1
+        assert columns[0].name == "id"
+
+    def test_custom_schema_view(self, project):
+        run_dbt(["run", "--select", "custom_schema_view"])
+        view_relation = relation_from_name(
+            project.adapter, "schema.custom_schema_view")
+        with get_connection(project.adapter):
+            columns = project.adapter.get_columns_in_relation(view_relation)
+        assert len(columns) == 1
+        assert columns[0].name == "id"
+
+    def test_custom_schema_view_nested(self, project):
+        run_dbt(["run", "--select", "custom_schema_view_nested"])
+        view_relation = relation_from_name(
+            project.adapter, "nested.schema.custom_schema_view_nested")
+        with get_connection(project.adapter):
+            columns = project.adapter.get_columns_in_relation(view_relation)
+        assert len(columns) == 1
+        assert columns[0].name == "id"

tests/utils/util.py

@@ -42,26 +42,27 @@ def relation_from_name(adapter, name: str, materialization=""):
     quote_policy = cls.get_default_quote_policy().to_dict()
     include_policy = cls.get_default_include_policy().to_dict()
 
+    relation_parts = name.split(".")
+
     # Make sure we have database/schema/identifier parts, even if
     # only identifier was supplied.
-    relation_parts = name.split(".")
-    if len(relation_parts) == 1:
-        if materialization == "view" or "view" in name:
-            relation_parts.insert(0, credentials.schema)
-        else:
-            relation_parts.insert(0, credentials.root_path)
-    if len(relation_parts) == 2:
-        # if the relation is a view then use database
-        if materialization == "view" or "view" in name:
-            relation_parts.insert(0, credentials.database)
-            relation_type = "view"
-        else:
-            relation_parts.insert(0, credentials.datalake)
-            relation_type = "table"
+
+    # if the relation is a view then use database
+    if materialization == "view" or "view" in name:
+        relation_parts.insert(0, credentials.database)
+        relation_parts.insert(1, credentials.schema)
+    else:
+        relation_parts.insert(0, credentials.datalake)
+        if credentials.root_path not in [None, "no_schema"]:
+            relation_parts.insert(1, credentials.root_path)
+
+
+    relation_type = "table" if materialization != "view" and "view" not in name else "view"
+
     kwargs = {
         "database": relation_parts[0],
-        "schema": relation_parts[1],
-        "identifier": relation_parts[2],
+        "schema": ".".join(relation_parts[1:-1]),
+        "identifier": relation_parts[-1],
         "type": relation_type,
     }