Repository: dagster-io/dagster
Issue: UPathIOManager truncates paths when partition key has a "." in it

### Dagster version

dagster, version 1.1.17

### What's the issue?

I'm using a custom PartitionsDefinition which uses static decimal partitioning keys, which happen to make sense for my data. It provides partition keys which include a "." character. For example, `0.94-to-0.95` is a partition key I use.

I also use a UPathIOManager - specifically, one that works with Google Cloud Storage.

The IO manager is storing `asset` outputs in paths like `"gs://redacted/redacted/redacted/0.94-to-0"`

### What did you expect to happen?

I would expect the UPathIOManager's generated paths to be unique per partition key, but since they truncate the partition key if it has a "." character, this is not the case. Data gets overwritten.

### How to reproduce?

Make a static partition definition using keys ["0.0", "1.0"]. Make assets using that partition definition, and use a UPathIOManager. Observe the paths where the assets are stored.

### Deployment type

Dagster Helm chart

### Deployment details

_No response_

### Additional information

_No response_

### Message from the maintainers

Impacted by this issue? Give it a 👍! We factor engagement into prioritization.

Changed files:
python_modules/dagster/dagster_tests/storage_tests/test_upath_io_manager.py, python_modules/dagster/dagster/_core/storage/upath_io_manager.py

diff --git a/python_modules/dagster/dagster/_core/storage/upath_io_manager.py b/python_modules/dagster/dagster/_core/storage/upath_io_manager.py
index c00b6f28b5..931bf3a169 100644
--- a/python_modules/dagster/dagster/_core/storage/upath_io_manager.py
+++ b/python_modules/dagster/dagster/_core/storage/upath_io_manager.py
@@ -2,6 +2,7 @@
 
 import inspect
 from abc import abstractmethod
+from pathlib import Path
 from typing import Any, Dict, Union
 
 from upath import UPath
@@ -75,7 +76,7 @@ def _get_path(self, context: Union[InputContext, OutputContext]) -> UPath:
         Returns the I/O path for a given context.
         Should not be used with partitions (use `_get_paths_for_partitions` instead).
         """
-        return self._get_path_without_extension(context).with_suffix(self.extension)
+        return _append_suffix(self._get_path_without_extension(context), self.extension)
 
     def _get_paths_for_partitions(
         self, context: Union[InputContext, OutputContext]
@@ -83,16 +84,12 @@ def _get_paths_for_partitions(
         """
         Returns a dict of partition_keys into I/O paths for a given context.
         """
-        if not context.has_asset_partitions:
-            raise TypeError(
-                f"Detected {context.dagster_type.typing_type} input type "
-                "but the asset is not partitioned"
-            )
+        check.invariant(context.has_asset_partitions, "The asset is not partitioned")
 
         partition_keys = context.asset_partition_keys
         asset_path = self._get_path_without_extension(context)
         return {
-            partition_key: (asset_path / partition_key).with_suffix(self.extension)
+            partition_key: _append_suffix(asset_path / partition_key, self.extension)
             for partition_key in partition_keys
         }
 
@@ -230,3 +227,7 @@ def handle_output(self, context: OutputContext, obj: Any):
         metadata.update(custom_metadata)  # type: ignore
 
         context.add_output_metadata(metadata)
+
+
+def _append_suffix(path: Path, suffix: str) -> Path:
+    return path.with_suffix(path.suffix + suffix)
diff --git a/python_modules/dagster/dagster_tests/storage_tests/test_upath_io_manager.py b/python_modules/dagster/dagster_tests/storage_tests/test_upath_io_manager.py
index 700ffa3f7e..e82b27beb9 100644
--- a/python_modules/dagster/dagster_tests/storage_tests/test_upath_io_manager.py
+++ b/python_modules/dagster/dagster_tests/storage_tests/test_upath_io_manager.py
@@ -1,13 +1,15 @@
 import json
 import pickle
 from datetime import datetime
-from pathlib import Path
+from pathlib import Path, PosixPath
 from typing import Any, Dict, List, cast
+from unittest.mock import MagicMock
 
 import pytest
 from dagster import (
     AllPartitionMapping,
     AssetIn,
+    AssetKey,
     DagsterType,
     DailyPartitionsDefinition,
     Field,
@@ -347,3 +349,35 @@ def my_asset() -> Any:
     ].entry_data.value == get_length(
         json_data
     )
+
+
+def test_get_path():
+    class MyIOManager(UPathIOManager):
+        def dump_to_path(self, context: OutputContext, obj: Any, path: UPath):
+            ...
+
+        def load_from_path(self, context: InputContext, path: UPath) -> Any:
+            ...
+
+    io_manager = MyIOManager(UPath("base_path"))
+
+    asset_output_context = MagicMock(asset_key=AssetKey(["foo", "bar"]))
+    assert io_manager._get_path(asset_output_context) == PosixPath("base_path/foo/bar")
+
+    partitioned_asset_output_context = MagicMock(
+        asset_key=AssetKey(["foo", "bar"]),
+        has_asset_partitions=True,
+        asset_partition_keys=["baz"],
+    )
+    assert io_manager._get_paths_for_partitions(partitioned_asset_output_context) == {
+        "baz": PosixPath("base_path/foo/bar/baz")
+    }
+
+    period_partitioned_asset_output_context = MagicMock(
+        asset_key=AssetKey(["foo", "bar"]),
+        has_asset_partitions=True,
+        asset_partition_keys=["apple.banana.orange"],
+    )
+    assert io_manager._get_paths_for_partitions(period_partitioned_asset_output_context) == {
+        "apple.banana.orange": PosixPath("base_path/foo/bar/apple.banana.orange")
+    }