add error for empty test result set

core/dbt/node_runners.py

@@ -546,12 +546,13 @@ def execute_test(self, test):
             fetch=True)
 
         num_rows = len(table.rows)
-        if num_rows > 1:
+        if num_rows != 1:
             num_cols = len(table.columns)
-            raise RuntimeError(
-                "Bad test {name}: Returned {rows} rows and {cols} cols"
-                .format(name=test.name, rows=num_rows, cols=num_cols))
-
+            dbt.exceptions.raise_compiler_error(
+                f"Bad test {test.test_metadata.name}: "
+                f"Returned {num_rows} rows "
+                f"and {num_cols} cols but expected 1 row and 1 column"
+            )
         return table[0][0]
 
     def before_execute(self):

test/integration/008_schema_tests_test/macros-v2/malformed/tests.sql

@@ -0,0 +1,6 @@
+-- tries to build a macro that always returns an empty set
+{% macro test_empty_aggregation(model, column_name) %}
+
+SELECT * from {{ model }} WHERE 1=0
+
+{% endmacro %}

test/integration/008_schema_tests_test/models-v2/custom-bad-test-macro/schema.yml

@@ -0,0 +1,12 @@
+version: 2
+
+models:
+  - name: table_copy
+    description: "A copy of the table"
+    columns:
+      - name: email
+        tests:
+          - not_null
+    # should throw a runtime error
+    tests:
+      - empty_aggregation

test/integration/008_schema_tests_test/models-v2/custom-bad-test-macro/table_copy.sql

@@ -0,0 +1,8 @@
+
+{{
+    config(
+        materialized='table'
+    )
+}}
+
+select * from {{ this.schema }}.seed

test/integration/008_schema_tests_test/test_schema_v2_tests.py

@@ -85,6 +85,51 @@ def test_postgres_malformed_schema_strict_will_break_run(self):
             self.run_dbt(strict=False)
 
 
+class TestMalformedMacroTests(DBTIntegrationTest):
+
+    def setUp(self):
+        DBTIntegrationTest.setUp(self)
+        self.run_sql_file("seed.sql")
+
+    @property
+    def schema(self):
+        return "schema_tests_008"
+
+    @property
+    def models(self):
+        return "models-v2/custom-bad-test-macro"
+
+    @property
+    def project_config(self):
+        return {
+            "macro-paths": ["macros-v2/malformed"],
+        }
+
+    def run_schema_validations(self):
+        args = FakeArgs()
+        test_task = TestTask(args, self.config)
+        return test_task.run()
+
+    @use_profile('postgres')
+    def test_postgres_malformed_macro_reports_error(self):
+        self.run_dbt(["deps"])
+        self.run_dbt()
+        expected_failure = 'not_null'
+
+        test_results = self.run_schema_validations()
+
+        self.assertEqual(len(test_results), 2)
+
+        for result in test_results:
+            self.assertTrue(result.error is not None or result.fail)
+            # Assert that error is thrown for empty schema test
+            if result.error is not None:
+                self.assertIn("Returned 0 rows", result.error)
+            # Assert that failure occurs for normal schema test
+            elif result.fail:
+                self.assertIn(expected_failure, result.node.name)
+
+
 class TestHooksInTests(DBTIntegrationTest):
 
     @property
@@ -148,7 +193,7 @@ def project_config(self):
         # dbt-integration-project contains a schema.yml file
         # both should work!
         return {
-            "macro-paths": ["macros-v2"],
+            "macro-paths": ["macros-v2/macros"],
         }
 
     @property