[promptflow][BugFix] Print error message when run status is failed (#1051)

# Description

If run status is failed, print error message; otherwise, user cannot
know why the run is failed in SDK experience.

SDK experience:

<img width="1112" alt="image"
src="https://github.com/microsoft/promptflow/assets/38847871/8441b759-da46-4268-aadf-26a72ea8c670">

CLI experience:

<img width="1115" alt="image"
src="https://github.com/microsoft/promptflow/assets/38847871/702a1596-0a0b-45b1-b40e-549d0614b927">

# All Promptflow Contribution checklist:
- [x] **The pull request does not introduce [breaking changes].**
- [ ] **CHANGELOG is updated for new features, bug fixes or other
significant changes.**
- [x] **I have read the [contribution guidelines](../CONTRIBUTING.md).**
- [ ] **Create an issue and link to the pull request to get dedicated
review from promptflow team. Learn more: [suggested
workflow](../CONTRIBUTING.md#suggested-workflow).**

## General Guidelines and Best Practices
- [x] Title of the pull request is clear and informative.
- [x] There are a small number of commits, each of which have an
informative message. This means that previously merged commits do not
appear in the history of the PR. For more information on cleaning up the
commits in your PR, [see this
page](https://github.com/Azure/azure-powershell/blob/master/documentation/development-docs/cleaning-up-commits.md).

### Testing Guidelines
- [x] Pull request includes test coverage for the included changes.

Author: zhengfeiwang

src/promptflow/promptflow/_sdk/operations/_local_storage_operations.py

@@ -430,13 +430,13 @@ def _prepare_folder(path: Union[str, Path]) -> Path:
 
     @staticmethod
     def _outputs_padding(df: pd.DataFrame, expected_rows: int) -> pd.DataFrame:
+        if len(df) == expected_rows:
+            return df
         missing_lines = []
         lines_set = set(df[LINE_NUMBER].values)
         for i in range(expected_rows):
             if i not in lines_set:
                 missing_lines.append({LINE_NUMBER: i})
-        if len(missing_lines) == 0:
-            return df
         df_to_append = pd.DataFrame(missing_lines)
         res = pd.concat([df, df_to_append], ignore_index=True)
         res = res.sort_values(by=LINE_NUMBER, ascending=True)

src/promptflow/promptflow/_sdk/operations/_run_operations.py

@@ -22,7 +22,7 @@
 )
 from promptflow._sdk._errors import InvalidRunStatusError, RunExistsError, RunNotFoundError, RunOperationParameterError
 from promptflow._sdk._orm import RunInfo as ORMRun
-from promptflow._sdk._utils import incremental_print, safe_parse_object_list
+from promptflow._sdk._utils import incremental_print, print_red_error, safe_parse_object_list
 from promptflow._sdk._visualize_functions import dump_html, generate_html_string
 from promptflow._sdk.entities import Run
 from promptflow._sdk.operations._local_storage_operations import LocalStorageOperations
@@ -140,7 +140,10 @@ def stream(self, name: Union[str, Run]) -> Run:
             available_logs = local_storage.logger.get_logs()
             incremental_print(available_logs, printed, file_handler)
             self._print_run_summary(run)
-            # won't print error here, put it in run dict
+            # print error message when run is failed
+            if run.status == RunStatus.FAILED:
+                error_message = local_storage.load_exception()["message"]
+                print_red_error(error_message)
         except KeyboardInterrupt:
             error_message = "The output streaming for the run was interrupted, but the run is still executing."
             print(error_message)

src/promptflow/tests/sdk_cli_test/e2etests/test_flow_run.py

@@ -58,6 +58,23 @@ def create_run_against_run(client, run: Run) -> Run:
     )
 
 
+def assert_run_with_invalid_column_mapping(client: PFClient, run: Run, capfd: pytest.CaptureFixture) -> None:
+    assert run.status == RunStatus.FAILED
+
+    expected_error_message = "The input for batch run is incorrect. Couldn't find these mapping relations"
+
+    client.stream(run.name)
+    out, _ = capfd.readouterr()
+    assert expected_error_message in out
+
+    local_storage = LocalStorageOperations(run)
+    assert os.path.exists(local_storage._exception_path)
+
+    exception = local_storage.load_exception()
+    assert expected_error_message in exception["message"]
+    assert exception["code"] == "BulkRunException"
+
+
 @pytest.mark.usefixtures("use_secrets_config_file", "setup_local_connection", "install_custom_tool_pkg")
 @pytest.mark.sdk_test
 @pytest.mark.e2etest
@@ -321,7 +338,7 @@ def test_run_reference_failed_run(self, pf):
         with pytest.raises(RunNotFoundError):
             pf.runs.get(name=run_name)
 
-    def test_referenced_output_not_exist(self, pf):
+    def test_referenced_output_not_exist(self, pf: PFClient, capfd: pytest.CaptureFixture) -> None:
         # failed run won't generate output
         failed_run = pf.run(
             flow=f"{FLOWS_DIR}/failed_flow",
@@ -336,13 +353,7 @@ def test_referenced_output_not_exist(self, pf):
             flow=f"{FLOWS_DIR}/failed_flow",
             column_mapping={"text": "${run.outputs.text}"},
         )
-
-        local_storage = LocalStorageOperations(run)
-        assert os.path.exists(local_storage._exception_path)
-
-        exception = local_storage.load_exception()
-        assert "The input for batch run is incorrect. Couldn't find these mapping relations" in exception["message"]
-        assert exception["code"] == "BulkRunException"
+        assert_run_with_invalid_column_mapping(pf, run, capfd)
 
     def test_connection_overwrite_file(self, local_client, local_aoai_connection):
         run = create_yaml_run(
@@ -650,7 +661,12 @@ def test_flow_bulk_run_with_additional_includes(self, azure_open_ai_connection:
         additional_includes = _get_additional_includes(snapshot_path / "flow.dag.yaml")
         assert not additional_includes
 
-    def test_input_mapping_source_not_found_error(self, azure_open_ai_connection: AzureOpenAIConnection, pf):
+    def test_input_mapping_source_not_found_error(
+        self,
+        azure_open_ai_connection: AzureOpenAIConnection,
+        pf: PFClient,
+        capfd: pytest.CaptureFixture,
+    ):
         # input_mapping source not found error won't create run
         name = str(uuid.uuid4())
         data_path = f"{DATAS_DIR}/webClassification3.jsonl"
@@ -660,13 +676,7 @@ def test_input_mapping_source_not_found_error(self, azure_open_ai_connection: Az
             column_mapping={"not_exist": "${data.not_exist_key}"},
             name=name,
         )
-
-        local_storage = LocalStorageOperations(run)
-        assert os.path.exists(local_storage._exception_path)
-
-        exception = local_storage.load_exception()
-        assert "The input for batch run is incorrect. Couldn't find these mapping relations" in exception["message"]
-        assert exception["code"] == "BulkRunException"
+        assert_run_with_invalid_column_mapping(pf, run, capfd)
 
     def test_input_mapping_with_dict(self, azure_open_ai_connection: AzureOpenAIConnection, pf):
         data_path = f"{DATAS_DIR}/webClassification3.jsonl"