@@ -204,6 +204,11 @@ def __init__(
         )
 
 
+class SavedDatasetLocationAlreadyExists(Exception):
+    def __init__(self, location: str):
+        super().__init__(f"Saved dataset location {location} already exists.")
+
+
 class FeastOfflineStoreInvalidName(Exception):
     def __init__(self, offline_store_class_name: str):
         super().__init__(
 
@@ -1155,6 +1155,7 @@ def create_saved_dataset(
         storage: SavedDatasetStorage,
         tags: Optional[Dict[str, str]] = None,
         feature_service: Optional[FeatureService] = None,
+        allow_overwrite: bool = False,
     ) -> SavedDataset:
         """
         Execute provided retrieval job and persist its outcome in given storage.
@@ -1163,6 +1164,14 @@ def create_saved_dataset(
         Name for the saved dataset should be unique within project, since it's possible to overwrite previously stored dataset
         with the same name.
 
+        Args:
+            from_: The retrieval job whose result should be persisted.
+            name: The name of the saved dataset.
+            storage: The saved dataset storage object indicating where the result should be persisted.
+            tags (optional): A dictionary of key-value pairs to store arbitrary metadata.
+            feature_service (optional): The feature service that should be associated with this saved dataset.
+            allow_overwrite (optional): If True, the persisted result can overwrite an existing table or file.
+
         Returns:
             SavedDataset object with attached RetrievalJob
 
@@ -1195,7 +1204,7 @@ def create_saved_dataset(
         dataset.min_event_timestamp = from_.metadata.min_event_timestamp
         dataset.max_event_timestamp = from_.metadata.max_event_timestamp
 
-        from_.persist(storage)
+        from_.persist(storage=storage, allow_overwrite=allow_overwrite)
 
         dataset = dataset.with_retrieval_job(
             self._get_provider().retrieve_saved_dataset(
 
@@ -493,7 +493,7 @@ def _execute_query(
         block_until_done(client=self.client, bq_job=bq_job, timeout=timeout)
         return bq_job
 
-    def persist(self, storage: SavedDatasetStorage):
+    def persist(self, storage: SavedDatasetStorage, allow_overwrite: bool = False):
         assert isinstance(storage, SavedDatasetBigQueryStorage)
 
         self.to_bigquery(
 
@@ -402,7 +402,7 @@ def _to_arrow_internal(self) -> pa.Table:
     def metadata(self) -> Optional[RetrievalMetadata]:
         return self._metadata
 
-    def persist(self, storage: SavedDatasetStorage):
+    def persist(self, storage: SavedDatasetStorage, allow_overwrite: bool = False):
         assert isinstance(storage, SavedDatasetAthenaStorage)
         self.to_athena(table_name=storage.athena_options.table)
 
 
@@ -297,7 +297,7 @@ def _to_arrow_internal(self) -> pa.Table:
     def metadata(self) -> Optional[RetrievalMetadata]:
         return self._metadata
 
-    def persist(self, storage: SavedDatasetStorage):
+    def persist(self, storage: SavedDatasetStorage, allow_overwrite: bool = False):
         assert isinstance(storage, SavedDatasetPostgreSQLStorage)
 
         df_to_postgres_table(
 
@@ -337,7 +337,7 @@ def _to_arrow_internal(self) -> pyarrow.Table:
             self.to_spark_df().write.parquet(temp_dir, mode="overwrite")
             return pq.read_table(temp_dir)
 
-    def persist(self, storage: SavedDatasetStorage):
+    def persist(self, storage: SavedDatasetStorage, allow_overwrite: bool = False):
         """
         Run the retrieval and persist the results in the same offline store used for read.
         Please note the persisting is done only within the scope of the spark session.
 
@@ -126,7 +126,7 @@ def to_trino(
         self._client.execute_query(query_text=query)
         return destination_table
 
-    def persist(self, storage: SavedDatasetStorage):
+    def persist(self, storage: SavedDatasetStorage, allow_overwrite: bool = False):
         """
         Run the retrieval and persist the results in the same offline store used for read.
         """
 
@@ -1,3 +1,4 @@
+import os
 import uuid
 from datetime import datetime
 from pathlib import Path
@@ -11,13 +12,16 @@
 import pytz
 from pydantic.typing import Literal
 
-from feast import FileSource, OnDemandFeatureView
 from feast.data_source import DataSource
-from feast.errors import FeastJoinKeysDuringMaterialization
+from feast.errors import (
+    FeastJoinKeysDuringMaterialization,
+    SavedDatasetLocationAlreadyExists,
+)
 from feast.feature_logging import LoggingConfig, LoggingSource
 from feast.feature_view import DUMMY_ENTITY_ID, DUMMY_ENTITY_VAL, FeatureView
 from feast.infra.offline_stores.file_source import (
     FileLoggingDestination,
+    FileSource,
     SavedDatasetFileStorage,
 )
 from feast.infra.offline_stores.offline_store import (
@@ -30,6 +34,7 @@
     get_pyarrow_schema_from_batch_source,
 )
 from feast.infra.registry.base_registry import BaseRegistry
+from feast.on_demand_feature_view import OnDemandFeatureView
 from feast.repo_config import FeastConfigBaseModel, RepoConfig
 from feast.saved_dataset import SavedDatasetStorage
 from feast.usage import log_exceptions_and_usage
@@ -83,8 +88,13 @@ def _to_arrow_internal(self):
         df = self.evaluation_function().compute()
         return pyarrow.Table.from_pandas(df)
 
-    def persist(self, storage: SavedDatasetStorage):
+    def persist(self, storage: SavedDatasetStorage, allow_overwrite: bool = False):
         assert isinstance(storage, SavedDatasetFileStorage)
+
+        # Check if the specified location already exists.
+        if not allow_overwrite and os.path.exists(storage.file_options.uri):
+            raise SavedDatasetLocationAlreadyExists(location=storage.file_options.uri)
+
         filesystem, path = FileSource.create_filesystem_and_path(
             storage.file_options.uri,
             storage.file_options.s3_endpoint_override,
 
@@ -96,12 +96,20 @@ def __eq__(self, other):
         )
 
     @property
-    def path(self):
-        """
-        Returns the path of this file data source.
-        """
+    def path(self) -> str:
+        """Returns the path of this file data source."""
         return self.file_options.uri
 
+    @property
+    def file_format(self) -> Optional[FileFormat]:
+        """Returns the file format of this file data source."""
+        return self.file_options.file_format
+
+    @property
+    def s3_endpoint_override(self) -> Optional[str]:
+        """Returns the s3 endpoint override of this file data source."""
+        return self.file_options.s3_endpoint_override
+
     @staticmethod
     def from_proto(data_source: DataSourceProto):
         return FileSource(
@@ -177,24 +185,33 @@ def get_table_query_string(self) -> str:
 class FileOptions:
     """
     Configuration options for a file data source.
+
+    Attributes:
+        uri: File source url, e.g. s3:// or local file.
+        s3_endpoint_override: Custom s3 endpoint (used only with s3 uri).
+        file_format: File source format, e.g. parquet.
     """
 
+    uri: str
+    file_format: Optional[FileFormat]
+    s3_endpoint_override: str
+
     def __init__(
         self,
+        uri: str,
         file_format: Optional[FileFormat],
         s3_endpoint_override: Optional[str],
-        uri: Optional[str],
     ):
         """
         Initializes a FileOptions object.
 
         Args:
+            uri: File source url, e.g. s3:// or local file.
             file_format (optional): File source format, e.g. parquet.
             s3_endpoint_override (optional): Custom s3 endpoint (used only with s3 uri).
-            uri (optional): File source url, e.g. s3:// or local file.
         """
+        self.uri = uri
         self.file_format = file_format
-        self.uri = uri or ""
         self.s3_endpoint_override = s3_endpoint_override or ""
 
     @classmethod
@@ -269,6 +286,17 @@ def to_data_source(self) -> DataSource:
             s3_endpoint_override=self.file_options.s3_endpoint_override,
         )
 
+    @staticmethod
+    def from_data_source(data_source: DataSource) -> "SavedDatasetStorage":
+        assert isinstance(data_source, FileSource)
+        return SavedDatasetFileStorage(
+            path=data_source.path,
+            file_format=data_source.file_format
+            if data_source.file_format
+            else ParquetFormat(),
+            s3_endpoint_override=data_source.s3_endpoint_override,
+        )
+
 
 class FileLoggingDestination(LoggingDestination):
     _proto_kind = "file_destination"
 
@@ -179,8 +179,16 @@ def on_demand_feature_views(self) -> List[OnDemandFeatureView]:
         pass
 
     @abstractmethod
-    def persist(self, storage: SavedDatasetStorage):
-        """Synchronously executes the underlying query and persists the result in the same offline store."""
+    def persist(self, storage: SavedDatasetStorage, allow_overwrite: bool = False):
+        """
+        Synchronously executes the underlying query and persists the result in the same offline store
+        at the specified destination.
+
+        Args:
+            storage: The saved dataset storage object specifying where the result should be persisted.
+            allow_overwrite: If True, a pre-existing location (e.g. table or file) can be overwritten.
+                Currently not all individual offline store implementations make use of this parameter.
+        """
         pass
 
     @property