[SDK]align display name generate logic with PFS (#274)

# Description

Please add an informative description that covers that changes made by
the pull request and link all relevant issues.

Refine run display name pattern

For run without upstream run (variant run), the pattern is
{client_run_display_name}-{variant_id}-{timestamp}
For run with upstream run (variant run), the pattern is
{upstream_run_display_name}-{client_run_display_name}-{timestamp}

# All Promptflow Contribution checklist:
- [ ] **The pull request does not introduce [breaking changes]**
- [ ] **CHANGELOG is updated for new features, bug fixes or other
significant changes.**
- [ ] **I have read the [contribution guidelines](../CONTRIBUTING.md).**

## General Guidelines and Best Practices
- [ ] Title of the pull request is clear and informative.
- [ ] 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
- [ ] Pull request includes test coverage for the included changes.

src/promptflow/promptflow/_sdk/entities/_run.py

@@ -126,9 +126,6 @@ def __init__(
         self._resources = kwargs.get("resources", None)
         # default run name: flow directory name + timestamp
         self.name = name or self._generate_run_name()
-        # default to use name if display_name is not provided
-        if not self.display_name:
-            self.display_name = self.name
 
     @property
     def created_on(self) -> str:
@@ -252,13 +249,15 @@ def _from_mt_service_entity(cls, run_entity) -> "Run":
         )
 
     def _to_orm_object(self) -> ORMRun:
+        """Convert current run entity to ORM object."""
+        display_name = self._format_display_name()
         return ORMRun(
             name=self.name,
             created_on=self.created_on,
             status=self.status,
             start_time=self._start_time.isoformat() if self._start_time else None,
             end_time=self._end_time.isoformat() if self._end_time else None,
-            display_name=self.display_name,
+            display_name=display_name,
             description=self.description,
             tags=json.dumps(self.tags) if self.tags else None,
             properties=json.dumps(self.properties),
@@ -365,11 +364,36 @@ def _generate_run_name(self) -> str:
         except Exception:
             return str(uuid.uuid4())
 
+    def _get_default_display_name(self) -> str:
+        display_name = self.display_name
+        if not display_name:
+            display_name = self._get_flow_dir().name
+        return display_name
+
+    def _format_display_name(self) -> str:
+        """
+        Format display name.
+            For run without upstream run (variant run)
+                the pattern is {client_run_display_name}-{variant_id}-{timestamp}
+            For run with upstream run (variant run)
+                the pattern is {upstream_run_display_name}-{client_run_display_name}-{timestamp}
+        """
+
+        display_name = self._get_default_display_name()
+        time_stamp = datetime.datetime.now().strftime("%Y%m%d%H%M")
+        if self.run:
+            display_name = f"{self.run.display_name}-{display_name}-{time_stamp}"
+        else:
+            variant = self.variant
+            variant = parse_variant(variant)[1] if variant else "default"
+            display_name = f"{display_name}-{variant}-{time_stamp}"
+        return display_name
+
     def _get_flow_dir(self) -> Path:
         flow = Path(self.flow)
         if flow.is_dir():
             return flow
-        return self.flow.parent
+        return flow.parent
 
     @classmethod
     def _get_schema_cls(self):
@@ -420,7 +444,8 @@ def _to_rest_object(self):
                 flow_definition_data_store_name=path_uri.datastore,
                 flow_definition_blob_path=path_uri.path,
                 run_id=self.name,
-                run_display_name=self.display_name,
+                # will use user provided display name since PFS will have special logic to update it.
+                run_display_name=self._get_default_display_name(),
                 description=self.description,
                 tags=self.tags,
                 node_variant=self.variant,
@@ -440,7 +465,7 @@ def _to_rest_object(self):
             return SubmitBulkRunRequest(
                 flow_definition_data_uri=str(self.flow),
                 run_id=self.name,
-                run_display_name=self.display_name,
+                run_display_name=self._get_default_display_name(),
                 description=self.description,
                 tags=self.tags,
                 node_variant=self.variant,

src/promptflow/promptflow/azure/_restclient/flow_service_caller.py

@@ -524,7 +524,7 @@ def create_flow_session(
             else:
                 action = "reset"
 
-            logger.info(f"Start polling until session {action} is complted...")
+            logger.info(f"Start polling until session {action} is completed...")
             # start polling status here.
             if "azure-asyncoperation" not in response.headers:
                 raise FlowRequestException(

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

@@ -221,7 +221,7 @@ def test_create_with_set(self, local_client):
             f"description={description}",
         )
         run = local_client.runs.get(run_id)
-        assert run.display_name == display_name
+        assert display_name in run.display_name
         assert run.tags == {"key": "val"}
         assert run.description == description
 
@@ -239,7 +239,7 @@ def test_create_with_set(self, local_client):
             f"description={description}",
             cwd=f"{RUNS_DIR}",
         )
-        assert run.display_name == display_name
+        assert display_name in run.display_name
         assert run.tags == {"key": "val"}
         assert run.description == description
 

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

@@ -345,7 +345,7 @@ def test_create_run_with_tags(self, pf):
             environment_variables={"API_BASE": "${azure_open_ai_connection.api_base}"},
         )
         assert run.name == name
-        assert run.display_name == display_name
+        assert f"{display_name}-default-" in run.display_name
         assert run.tags == tags
 
     def test_run_display_name(self, pf):
@@ -356,16 +356,27 @@ def test_run_display_name(self, pf):
                 environment_variables={"API_BASE": "${azure_open_ai_connection.api_base}"},
             )
         )
-        assert run.display_name == run.name
+        assert "print_env_var-default-" in run.display_name
+        base_run = pf.runs.create_or_update(
+            run=Run(
+                flow=Path(f"{FLOWS_DIR}/print_env_var"),
+                data=f"{DATAS_DIR}/env_var_names.jsonl",
+                environment_variables={"API_BASE": "${azure_open_ai_connection.api_base}"},
+                display_name="my_run",
+            )
+        )
+        assert "my_run-default-" in base_run.display_name
+
         run = pf.runs.create_or_update(
             run=Run(
                 flow=Path(f"{FLOWS_DIR}/print_env_var"),
                 data=f"{DATAS_DIR}/env_var_names.jsonl",
                 environment_variables={"API_BASE": "${azure_open_ai_connection.api_base}"},
                 display_name="my_run",
+                run=base_run,
             )
         )
-        assert run.display_name == "my_run"
+        assert f"{base_run.display_name}-my_run-" in run.display_name
 
     def test_run_dump(self, azure_open_ai_connection: AzureOpenAIConnection, pf: PFClient) -> None:
         data_path = f"{DATAS_DIR}/webClassification3.jsonl"