diff --git a/.changes/unreleased/Features-20230223-145508.yaml b/.changes/unreleased/Features-20230223-145508.yaml
new file mode 100644
index 000000000..ade28e2b5
--- /dev/null
+++ b/.changes/unreleased/Features-20230223-145508.yaml
@@ -0,0 +1,6 @@
+kind: Features
+body: 'implement get_column_schema_from_query '
+time: 2023-02-23T14:55:08.186645-05:00
+custom:
+  Author: michelleark
+  Issue: "529"
diff --git a/dbt/adapters/bigquery/impl.py b/dbt/adapters/bigquery/impl.py
index bf7eabeda..af59e2666 100644
--- a/dbt/adapters/bigquery/impl.py
+++ b/dbt/adapters/bigquery/impl.py
@@ -465,6 +465,17 @@ def copy_table(self, source, destination, materialization):
 
         return "COPY TABLE with materialization: {}".format(materialization)
 
+    @available.parse(lambda *a, **k: [])
+    def get_column_schema_from_query(self, sql: str) -> List[BigQueryColumn]:
+        """Get a list of the column names and data types from the given sql.
+
+        :param str sql: The sql to execute.
+        :return: List[BigQueryColumn]
+        """
+        _, iterator = self.connections.raw_execute(sql)
+        columns = [self.Column.create_from_field(field) for field in iterator.schema]
+        return columns
+
     @available.parse(lambda *a, **k: False)
     def get_columns_in_select_sql(self, select_sql: str) -> List[BigQueryColumn]:
         try:
diff --git a/dbt/include/bigquery/macros/utils/get_columns_spec_ddl.sql b/dbt/include/bigquery/macros/utils/get_columns_spec_ddl.sql
index 4f2720b7e..f074f1bee 100644
--- a/dbt/include/bigquery/macros/utils/get_columns_spec_ddl.sql
+++ b/dbt/include/bigquery/macros/utils/get_columns_spec_ddl.sql
@@ -14,3 +14,7 @@
       {{exceptions.warn("We noticed you have `constraints_check` configs, these are NOT compatible with BigQuery and will be ignored")}}
   {%- endif %}
 {% endmacro %}
+
+{% macro bigquery__format_column(column) -%}
+  {{ return(column.column.lower() ~ " " ~ column.data_type) }}
+{%- endmacro -%}
diff --git a/tests/functional/adapter/test_constraints.py b/tests/functional/adapter/test_constraints.py
index b9a271b8c..85abc7acc 100644
--- a/tests/functional/adapter/test_constraints.py
+++ b/tests/functional/adapter/test_constraints.py
@@ -31,6 +31,7 @@
 # - raises an explicit error, if you try to set a primary key constraint, because it's not enforced
 constraints_yml = model_schema_yml.replace("text", "string").replace("primary key", "")
 
+
 class TestBigQueryConstraintsColumnsEqual(BaseConstraintsColumnsEqual):
     @pytest.fixture(scope="class")
     def models(self):
@@ -40,6 +41,30 @@ def models(self):
             "constraints_schema.yml": constraints_yml,
         }
 
+    @pytest.fixture
+    def string_type(self):
+        return "STRING"
+
+    @pytest.fixture
+    def int_type(self):
+        return "INT64"
+
+    @pytest.fixture
+    def data_types(self, int_type, string_type):
+        # sql_column_value, schema_data_type, error_data_type
+        return [
+            ['1', int_type, int_type],
+            ["'1'", string_type, string_type],
+            ["cast('2019-01-01' as date)", 'date', 'DATE'],
+            ["true", 'bool', 'BOOL'],
+            ["cast('2013-11-03 00:00:00-07' as TIMESTAMP)", 'timestamp', 'TIMESTAMP'],
+            ["['a','b','c']", f'ARRAY<{string_type}>', f'ARRAY<{string_type}>'],
+            ["[1,2,3]", f'ARRAY<{int_type}>', f'ARRAY<{int_type}>'],
+            ["cast(1 as NUMERIC)", 'numeric', 'NUMERIC'],
+            ["""JSON '{"name": "Cooper", "forname": "Alice"}'""", 'json', 'JSON'],
+            ['STRUCT("Rudisha" AS name, [23.4, 26.3, 26.4, 26.1] AS laps)', 'STRUCT<name STRING, laps ARRAY<FLOAT64>>', 'STRUCT<name STRING, laps ARRAY<FLOAT64>>']
+        ]
+
 
 class TestBigQueryConstraintsRuntimeEnforcement(BaseConstraintsRuntimeEnforcement):
     @pytest.fixture(scope="class")
@@ -48,7 +73,7 @@ def models(self):
             "my_model.sql": my_model_sql,
             "constraints_schema.yml": constraints_yml,
         }
-    
+
     @pytest.fixture(scope="class")
     def expected_sql(self, project):
         relation = relation_from_name(project.adapter, "my_model")