def upsert_run_metadata_fields(
        self,
        run_name: str,
        object_name: str,
        object_type: str,
        object_version: Optional[str] = None,
        entry_id: Optional[str] = None,
        entry_type: Optional[str] = None,
        **field_updates: Any,
    ) -> None:
        """
        Consolidated upsert method for run metadata entries.
        Supported entry types: "invocations", "computations", "metrics" and also entity level fields like "description", "run_status".

        This method retrieves the current run metadata, merges new fields with any existing
        entry (if present), and then pushes the updated run metadata via _update_run.

        For "invocations" and "computations", start_time_ms and end_time_ms are required by the backend.
        If these fields are not provided in the new update and the existing entry also does not have them,
        they default to 0. This allows start_time/end_time to be updated from separate calls without
        overwriting previously set values.

        Args:
            entry_type: One of "invocations", "computations", "metrics".
            entry_id: Unique identifier for the metadata entry.
            run_name: The name of the run.
            object_name: The name of the object (e.g. agent name).
            object_type: The type of the object.
            object_version: Optional version.
            **field_updates: The fields to upsert.
        """
        if entry_type and entry_type not in SUPPORTED_ENTRY_TYPES:
            raise ValueError(f"Unsupported entry type: {entry_type}")

        keys = list(field_updates.keys())
        for key in keys:
            if entry_type:
                field_updates[f"{entry_type}.{entry_id}.{key}"] = (
                    field_updates.pop(key)
                )

        # Retrieve the existing run.
        existing_run_metadata_df = self.get_run(
            run_name=run_name,
            object_name=object_name,
            object_type=object_type,
            object_version=object_version,
        )

        if existing_run_metadata_df.empty:
            raise ValueError(f"Run '{run_name}' does not exist.")

        existing_run = json.loads(
            list(
                existing_run_metadata_df.to_dict(orient="records")[0].values()
            )[0]
        )
        existing_run_metadata = existing_run.get("run_metadata", {})

        logger.debug(
            f"Existing run metadata before update: {existing_run_metadata}"
        )
        (
            updated_run_metadata,
            invocation_masks,
            metric_field_masks,
            computation_field_masks,
            non_map_field_masks,
        ) = self._update_run_metadata_field_masks(
            existing_run_metadata=existing_run_metadata,
            field_updates=field_updates,
        )

        logger.debug(
            f"invocation field masks: {invocation_masks}, "
            f"metrics field masks: {metric_field_masks}, "
            f"computation field masks: {computation_field_masks}, "
            f"non-map field masks: {non_map_field_masks}"
        )

        update_description = "description" in field_updates
        description = field_updates.get("description", None)

        update_run_status = "run_status" in field_updates
        run_status = field_updates.get("run_status", None)

        self._update_run(
            run_name=run_name,
            object_name=object_name,
            object_type=object_type,
            object_version=object_version,
            invocation_field_masks=invocation_masks,
            metric_field_masks=metric_field_masks,
            computation_field_masks=computation_field_masks,
            updated_run_metadata=updated_run_metadata,
            update_description=update_description,
            description=description,
            update_run_status=update_run_status,
            run_status=run_status,
        )

        Consolidated upsert method for run metadata entries.
        Supported entry types: "invocations", "computations", "metrics" and also entity level fields like "description", "run_status".

        This method retrieves the current run metadata, merges new fields with any existing
        entry (if present), and then pushes the updated run metadata via _update_run.

        For "invocations" and "computations", start_time_ms and end_time_ms are required by the backend.
        If these fields are not provided in the new update and the existing entry also does not have them,
        they default to 0. This allows start_time/end_time to be updated from separate calls without
        overwriting previously set values.

        Args:
            entry_type: One of "invocations", "computations", "metrics".
            entry_id: Unique identifier for the metadata entry.
            run_name: The name of the run.
            object_name: The name of the object (e.g. agent name).
            object_type: The type of the object.
            object_version: Optional version.
            **field_updates: The fields to upsert.
        
src/connectors/snowflake/trulens/connectors/snowflake/dao/run.py