Fix get columns in relation

CHANGELOG.md

@@ -7,6 +7,7 @@
 ### Fixes
 
 - Fix column-level `persist_docs` on Delta tables, add tests ([#180](https://github.com/fishtown-analytics/dbt-spark/pull/180))
+- Fix `get_columns_in_relation` when called on models created in the same run ([#197](https://github.com/dbt-labs/dbt-spark/pull/197))
 
 ## dbt-spark 0.20.0rc1 (June 8, 2021)
 

dbt/adapters/spark/impl.py

@@ -212,7 +212,7 @@ def get_columns_in_relation(self, relation: Relation) -> List[SparkColumn]:
                                 for cached_relation in cached_relations
                                 if str(cached_relation) == str(relation)),
                                None)
-        if cached_relation is None:
+        if cached_relation is None or cached_relation.information is None:
             rows: List[agate.Row] = super().get_columns_in_relation(relation)
             columns = self.parse_describe_extended(relation, rows)
         else:

test/custom/get_columns_in_relation/models/child.sql

@@ -0,0 +1 @@
+SELECT 1

test/custom/get_columns_in_relation/models/get_columns_from_child.sql

@@ -0,0 +1,6 @@
+SELECT 
+  {% set cols = adapter.get_columns_in_relation(ref('child')) %}
+  {% for col in cols %}
+    {{ adapter.quote(col.column) }}{%- if not loop.last %},{{ '\n ' }}{% endif %}
+  {% endfor %}
+FROM {{ ref('child') }}

test/custom/get_columns_in_relation/test_get_columns_in_relation.py

@@ -0,0 +1,27 @@
+from test.custom.base import DBTSparkIntegrationTest, use_profile
+
+
+class TestGetColumnInRelationInSameRun(DBTSparkIntegrationTest):
+    @property
+    def schema(self):
+        return "get_columns_in_relation"
+
+    @property
+    def models(self):
+        return "models"
+
+    def run_and_test(self):
+        self.run_dbt(["run"])
+        self.assertTablesEqual("child", "get_columns_from_child")
+
+    @use_profile("apache_spark")
+    def test_get_columns_in_relation_in_same_run_apache_spark(self):
+        self.run_and_test()
+
+    @use_profile("databricks_cluster")
+    def test_get_columns_in_relation_in_same_run_databricks_cluster(self):
+        self.run_and_test()
+
+    @use_profile("databricks_sql_endpoint")
+    def test_get_columns_in_relation_in_same_run_databricks_sql_endpoint(self):
+        self.run_and_test()