Return agate_table in dbt run-operation result

core/dbt/artifacts/schemas/run/v5/run.py

@@ -72,6 +72,9 @@ class RunResultOutput(BaseResult):
     compiled: Optional[bool]
     compiled_code: Optional[str]
     relation_name: Optional[str]
+    agate_table: Optional["agate.Table"] = field(
+        default=None, metadata={"serialize": lambda x: None, "deserialize": lambda x: None}
+    )
     batch_results: Optional[BatchResults] = None
 
 
@@ -88,6 +91,7 @@ def process_run_result(result: RunResult) -> RunResultOutput:
         message=result.message,
         adapter_response=result.adapter_response,
         failures=result.failures,
+        agate_table=result.agate_table,
         batch_results=result.batch_results,
         compiled=result.node.compiled if compiled else None,  # type:ignore
         compiled_code=result.node.compiled_code if compiled else None,  # type:ignore

core/dbt/task/run_operation.py

@@ -49,6 +49,9 @@ def _run_unsafe(self, package_name, macro_name) -> "agate.Table":
                 macro_name, project=package_name, kwargs=macro_kwargs, macro_resolver=self.manifest
             )
 
+        if isinstance(res, str):
+            return None
+
         return res
 
     def run(self) -> RunResultsArtifact:
@@ -61,10 +64,11 @@ def run(self) -> RunResultsArtifact:
 
         success = True
         package_name, macro_name = self._get_macro_parts()
+        execute_macro_result = None
 
         with collect_timing_info("execute", timing.append):
             try:
-                self._run_unsafe(package_name, macro_name)
+                execute_macro_result = self._run_unsafe(package_name, macro_name)
             except dbt_common.exceptions.DbtBaseException as exc:
                 fire_event(RunningOperationCaughtError(exc=str(exc)))
                 fire_event(LogDebugStackTrace(exc_info=traceback.format_exc()))
@@ -113,6 +117,7 @@ def run(self) -> RunResultsArtifact:
             ),
             thread_id=threading.current_thread().name,
             timing=timing,
+            agate_table=execute_macro_result,
             batch_results=None,
         )
 

tests/functional/adapter/dbt_run_operations/test_dbt_run_operations.py

@@ -23,7 +23,6 @@ def test_result_without_return(self, project):
         results = self.run_operation("select_something", name="world")
         assert results.results[0].agate_table is None
 
-    @pytest.mark.xfail
     def test_result_with_return(self, project):
         results = self.run_operation("select_something_with_return", name="world")
         assert len(results.results[0].agate_table) == 1