Repository: skypilot-org/skypilot
Pull request: [Storage] Support multiple files in Storage
Closes #1070 and unblocks #1266.

Adds support for specifying multiple paths for upload to a Storage object using a list as a source:

```
  /dataset/:
    name: sky-multisource-storage\
    source: [~/mydir/myfile.txt, ~/datasets]
```

Tested:
- [x] `pytest test_smoke.py::TestStorageWithCredentials`
- [x] `sky launch examples/storage_demo.yaml`
- [x] Rendered docs locally 
- [x] All smoke tests 

Problem statement:
[Storage] Upload multiple files to the same bucket with different directories
It would be great if sky storage can support uploading multiple files into the same bucket but different folder in the bucket. Currently, a workaround is to write another program to upload the files first, before running the sky yaml.

```python
for l_path in local_paths:
  subprocess.run(f'gsutil -m cp {l_path} gs://bucket-name/run-{run_id}', shell=True)
```

In yaml:
```yaml
file_mounts:
  remote_path: gs://bucket-name/run-{run_id}
```

Hints from issue discussion:
Is it possible for the user to organize their files into a local hierarchy however they want, then simply specify the root of the hierarchy in the source? 

```bash
mkdir ~/mydir/run-1/
mv {run1_files_path} ~/mydir/run-1/
...
```

In YAML:
```yaml
file_mounts:
  remote_path: 
    source: ~/mydir/
```
> Is it possible for the user to organize their files into a local hierarchy however they want, then simply specify the root of the hierarchy in the source?
> 
> ```shell
> mkdir ~/mydir/run-1/
> mv {run1_files_path} ~/mydir/run-1/
> ...
> ```
> 
> In YAML:
> 
> ```yaml
> file_mounts:
>   remote_path: 
>     source: ~/mydir/
> ```

Yeah, I proposed that to them as well but seems like they do not really want to change the local hierarchy as they will start thousands of spot tasks, each of which will use some of the files. Moving the files into different folders for each task will make the local folder a bit messy.

Commit message: add validation

Patch to review:
diff --git a/docs/source/reference/storage.rst b/docs/source/reference/storage.rst
index 8ec6992f7fa..fda3083466f 100644
--- a/docs/source/reference/storage.rst
+++ b/docs/source/reference/storage.rst
@@ -156,6 +156,21 @@ and storage mounting:
         name: romil-output-bucket
         mode: MOUNT
 
+      # *** Uploading multiple files to the same Storage object ***
+      #
+      # The source field in a storage object can also be a list of local paths.
+      # This is useful when multiple files or directories need to be uploaded to the
+      # same bucket.
+      #
+      # Note: If the source list contains a directory, the entire directory is copied
+      # to the root of the bucket. For instance, in this example, the contents of
+      # ~/datasets are copied to s3://sky-multisource-storage/datasets/. ~/mydir/myfile.txt
+      # will appear at s3://sky-multisource-storage/myfile.txt.
+      /datasets-multisource-storage:
+        name: sky-multisource-storage2 # Make sure this name is unique or you own this bucket
+        source: [~/mydir/myfile.txt, ~/datasets]
+
+
     run: |
       pwd
       ls -la /
@@ -245,10 +260,10 @@ Storage YAML reference
 
       sky.Storage.source: str
         The source attribute specifies the local path that must be made available
-        in the storage object. It can either be a local path, in which case data
-        is uploaded to the cloud to an appropriate object store (s3 or gcs), or it
-        can be a remote path (s3://, gs://), in which case it is copied or mounted
-        directly (see mode flag below).
+        in the storage object. It can either be a local path or a list of local
+        paths, in which case data is uploaded to the cloud to an appropriate
+        object store (s3 or gcs), or it can be a remote path (s3://, gs://),
+        in which case it is copied or mounted directly (see mode flag below).
 
       sky.Storage.store: str; either of 's3' or 'gcs'
         If you wish to force sky.Storage to be backed by a specific cloud object
diff --git a/examples/storage_demo.yaml b/examples/storage_demo.yaml
index b6d0480648a..250b88ea2d7 100644
--- a/examples/storage_demo.yaml
+++ b/examples/storage_demo.yaml
@@ -8,9 +8,6 @@
 
 name: storage-demo
 
-resources:
-  cloud: aws
-
 ###############################################
 #  SkyPilot storage #
 ###############################################
@@ -40,7 +37,7 @@ resources:
 # 3. You want to have a write-able path to directly write files to S3 buckets -
 #    specify a name (to create a bucket if it doesn't exist) and set the mode
 #    to MOUNT. This is useful for writing code outputs, such as checkpoints or
-#    logs directly to a S3 bucket.
+#    logs directly to a S3/GCS bucket.
 #
 # 4. You want to have a shared file-system across workers running on different
 #    nodes - specify a name (to create a bucket if it doesn't exist) and set
@@ -94,7 +91,7 @@ file_mounts:
   # When the VM is initialized, the contents of the bucket are copied to
   # /datasets-storage. If the bucket already exists, it is fetched and re-used.
   /datasets-storage:
-    name: sky-dataset-romilzz # Make sure this name is unique or you own this bucket
+    name: sky-dataset-mybucket # Make sure this name is unique or you own this bucket
     source: ~/datasets
     store: s3 # Could be either of [s3, gcs]; default: None
     persistent: True  # Defaults to True, can be set to false.
@@ -110,7 +107,7 @@ file_mounts:
   # other storage mounts using the same bucket with MOUNT mode. Note that the
   # source is synced with the remote bucket everytime this task is run.
   /dataset-storage-mount:
-    name: sky-dataset-romilzz
+    name: sky-dataset-mybucket
     source: ~/datasets
     mode: MOUNT
 
@@ -137,9 +134,23 @@ file_mounts:
   # this approach can also be used to create a shared filesystem across workers.
   # See examples/storage/pingpong.yaml for an example.
   /outputs-mount:
-    name: romil-output-bucketzz
+    name: sky-output-bucket # Make sure this name is unique or you own this bucket
     mode: MOUNT
 
+  # *** Uploading multiple files to the same Storage object ***
+  #
+  # The source field in a storage object can also be a list of local paths.
+  # This is useful when multiple files or directories need to be uploaded to the
+  # same bucket.
+  #
+  # Note: If the source list contains a directory, the entire directory is copied
+  # to the root of the bucket. For instance, in this example, the contents of
+  # ~/datasets are copied to s3://sky-multisource-storage/datasets/. ~/mydir/myfile.txt
+  # will appear at s3://sky-multisource-storage/myfile.txt.
+  /datasets-multisource-storage:
+    name: sky-multisource-storage2 # Make sure this name is unique or you own this bucket
+    source: [~/mydir/myfile.txt, ~/datasets]
+
 run: |
   pwd
   ls -la /
diff --git a/sky/data/storage.py b/sky/data/storage.py
index b5a853de167..1a8fd623363 100644
--- a/sky/data/storage.py
+++ b/sky/data/storage.py
@@ -3,7 +3,7 @@
 import os
 import subprocess
 import time
-from typing import Any, Dict, Optional, Tuple, Union
+from typing import Any, Dict, Optional, Tuple, Union, List
 import urllib.parse
 
 from sky import clouds
@@ -216,8 +216,19 @@ def get_handle(self) -> StorageHandle:
         """
         raise NotImplementedError
 
-    def sync_local_dir(self) -> None:
-        """Syncs a local directory to a Store bucket."""
+    def upload_local_path(self,
+                          local_path: str,
+                          create_dir: bool = False) -> None:
+        """Syncs a local directory to a Store bucket.
+
+        Args:
+            local_path: str; Path to local file or directory
+            create_dir: If the local_path is a directory and this is set to
+                False, the contents of the directory are directly uploaded to
+                root of the bucket. If the local_path is a directory and this is
+                set to True, the directory is created in the bucket root and
+                contents are uploaded to it.
+        """
         raise NotImplementedError
 
     def download_remote_dir(self, local_path: str) -> None:
@@ -334,7 +345,7 @@ def remove_store(self, store: AbstractStore) -> None:
 
     def __init__(self,
                  name: Optional[str] = None,
-                 source: Optional[Path] = None,
+                 source: Optional[Union[Path, List[Path]]] = None,
                  stores: Optional[Dict[StoreType, AbstractStore]] = None,
                  persistent: Optional[bool] = True,
                  mode: StorageMode = StorageMode.MOUNT,
@@ -364,9 +375,9 @@ def __init__(self,
         Args:
           name: str; Name of the storage object. Typically used as the
             bucket name in backing object stores.
-          source: str; File path where the data is initially stored. Can be a
-            local path or a cloud URI (s3://, gs://, etc.). Local paths do not
-            need to be absolute.
+          source: str, List[str]; File path where the data is initially stored.
+            Can be a single local path, a list of local paths, or a cloud URI
+            (s3://, gs://, etc.). Local paths do not need to be absolute.
           stores: Optional; Specify pre-initialized stores (S3Store, GcsStore).
           persistent: bool; Whether to persist across sky launches.
           mode: StorageMode; Specify how the storage object is manifested on
@@ -420,9 +431,11 @@ def __init__(self,
             # syncing to file_mount stage..
             if self.sync_on_reconstruction:
                 msg = ''
-                if self.source and \
-                        not data_utils.is_cloud_store_url(self.source):
-                    msg = ' and uploading from source'
+                if self.source:
+                    if isinstance(self.source,
+                                  list) or not data_utils.is_cloud_store_url(
+                                      self.source):
+                        msg = ' and uploading from source'
                 logger.info(f'Verifying bucket{msg} for storage {self.name}')
                 self.sync_all_stores()
 
@@ -440,10 +453,11 @@ def __init__(self,
             if self.source is not None:
                 # If source is a pre-existing bucket, connect to the bucket
                 # If the bucket does not exist, this will error out
-                if self.source.startswith('s3://'):
-                    self.add_store(StoreType.S3)
-                elif self.source.startswith('gs://'):
-                    self.add_store(StoreType.GCS)
+                if isinstance(self.source, str):
+                    if self.source.startswith('s3://'):
+                        self.add_store(StoreType.S3)
+                    elif self.source.startswith('gs://'):
+                        self.add_store(StoreType.GCS)
 
     @staticmethod
     def _validate_source(source: str, mode: StorageMode,
@@ -461,57 +475,72 @@ def _validate_source(source: str, mode: StorageMode,
           source: str; The source path.
           is_local_path: bool; Whether the source is a local path. False if URI.
         """
-        # Check if source is a valid local/remote URL
-        split_path = urllib.parse.urlsplit(source)
-        if split_path.scheme == '':
-            if source.endswith('/'):
+
+        def _validate_local_source(local_source):
+            if local_source.endswith('/'):
                 with ux_utils.print_exception_no_traceback():
                     raise exceptions.StorageSourceError(
                         'Storage source paths cannot end with a slash '
                         '(try "/mydir: /mydir" or "/myfile: /myfile"). '
-                        f'Found source={source}')
+                        f'Found source={local_source}')
             # Local path, check if it exists
-            full_src = os.path.abspath(os.path.expanduser(source))
+            full_src = os.path.abspath(os.path.expanduser(local_source))
             # Only check if local source exists if it is synced to the bucket
             if not os.path.exists(full_src) and sync_on_reconstruction:
                 with ux_utils.print_exception_no_traceback():
                     raise exceptions.StorageSourceError(
                         'Local source path does not'
-                        f' exist: {source}')
-            # Check if source is a file - throw error if it is
-            if os.path.isfile(full_src):
-                with ux_utils.print_exception_no_traceback():
-                    raise exceptions.StorageSourceError(
-                        'Storage source path cannot be a file - only '
-                        'directories are supported as a source. '
-                        'To upload a single file, either\n'
-                        '1) Use regular file mounts by writing '
-                        f'<destination_path>: {source} in the file_mounts '
-                        'section of your YAML, or\n2) Place the file in a '
-                        'directory and specify the directory as the source.')
+                        f' exist: {local_source}')
             # Raise warning if user's path is a symlink
             elif os.path.islink(full_src):
                 logger.warning(f'Source path {source} is a symlink. '
                                'Referenced contents are uploaded, matching '
                                'the default behavior for S3 and GCS syncing.')
+
+        # Check if source is a list of paths
+        if isinstance(source, list):
+            # Validate each path
+            for local_source in source:
+                _validate_local_source(local_source)
             is_local_source = True
-        elif split_path.scheme in ['s3', 'gs']:
-            is_local_source = False
-            # Storage mounting does not support mounting specific files from
-            # cloud store - ensure path points to only a directory
-            if mode == StorageMode.MOUNT:
-                if split_path.path.strip('/') != '':
-                    with ux_utils.print_exception_no_traceback():
-                        raise exceptions.StorageModeError(
-                            'MOUNT mode does not support'
-                            ' mounting specific files from cloud'
-                            ' storage. Please use COPY mode or'
-                            ' specify only the bucket name as'
-                            ' the source.')
         else:
-            with ux_utils.print_exception_no_traceback():
-                raise exceptions.StorageSourceError(
-                    f'Supported paths: local, s3://, gs://. Got: {source}')
+            # Check if str source is a valid local/remote URL
+            split_path = urllib.parse.urlsplit(source)
+            if split_path.scheme == '':
+                _validate_local_source(source)
+                # Check if source is a file - throw error if it is
+                full_src = os.path.abspath(os.path.expanduser(source))
+                if os.path.isfile(full_src):
+                    with ux_utils.print_exception_no_traceback():
+                        raise exceptions.StorageSourceError(
+                            'Storage source path cannot be a file - only'
+                            ' directories are supported as a source. '
+                            'To upload a single file, specify it in a list '
+                            f'by writing <destination_path>: [{source}]. Note '
+                            'that the file will be uploaded to the root of the '
+                            'bucket and will appear at <destination_path>/'
+                            f'{os.path.basename(source)}. Alternatively, you '
+                            'can directly upload the file to the VM without '
+                            'using a bucket by writing <destination_path>: '
+                            f'{source} in the file_mounts section of your YAML')
+                is_local_source = True
+            elif split_path.scheme in ['s3', 'gs']:
+                is_local_source = False
+                # Storage mounting does not support mounting specific files from
+                # cloud store - ensure path points to only a directory
+                if mode == StorageMode.MOUNT:
+                    if split_path.path.strip('/') != '':
+                        with ux_utils.print_exception_no_traceback():
+                            raise exceptions.StorageModeError(
+                                'MOUNT mode does not support'
+                                ' mounting specific files from cloud'
+                                ' storage. Please use COPY mode or'
+                                ' specify only the bucket name as'
+                                ' the source.')
+            else:
+                with ux_utils.print_exception_no_traceback():
+                    raise exceptions.StorageSourceError(
+                        f'Supported paths: local, s3://, gs://. Got: {source}')
         return source, is_local_source
 
     def _validate_storage_spec(self) -> None:
@@ -744,8 +773,9 @@ def add_if_not_none(key, value):
                 config[key] = value
 
         name = self.name
-        if self.source is not None and data_utils.is_cloud_store_url(
-                self.source):
+        if self.source is not None and isinstance(
+                self.source, str) and data_utils.is_cloud_store_url(
+                    self.source):
             # Remove name if source is a cloud store URL
             name = None
         add_if_not_none('name', name)
@@ -778,17 +808,22 @@ def __init__(self,
 
     def _validate(self):
         if self.source is not None:
-            if self.source.startswith('s3://'):
-                assert self.name == data_utils.split_s3_path(self.source)[0], (
-                    'S3 Bucket is specified as path, the name should be the '
-                    'same as S3 bucket.')
-            elif self.source.startswith('gs://'):
-                assert self.name == data_utils.split_gcs_path(self.source)[0], (
-                    'GCS Bucket is specified as path, the name should be the '
-                    'same as GCS bucket.')
-                assert data_utils.verify_gcs_bucket(self.name), (
-                    f'Source specified as {self.source}, a GCS bucket. ',
-                    'GCS Bucket should exist.')
+            if isinstance(self.source, str):
+                if self.source.startswith('s3://'):
+                    assert self.name == data_utils.split_s3_path(
+                        self.source
+                    )[0], (
+                        'S3 Bucket is specified as path, the name should be the'
+                        ' same as S3 bucket.')
+                elif self.source.startswith('gs://'):
+                    assert self.name == data_utils.split_gcs_path(
+                        self.source
+                    )[0], (
+                        'GCS Bucket is specified as path, the name should be '
+                        'the same as GCS bucket.')
+                    assert data_utils.verify_gcs_bucket(self.name), (
+                        f'Source specified as {self.source}, a GCS bucket. ',
+                        'GCS Bucket should exist.')
 
     def initialize(self):
         """Initializes the S3 store object on the cloud.
@@ -820,13 +855,23 @@ def upload(self):
             StorageUploadError: if upload fails.
         """
         try:
-            if self.source is not None:
+            if isinstance(self.source, list):
+                for source_path in self.source:
+                    # We upload each file in the list as individually because
+                    # aws s3 sync does not support a list of files. Symlinking
+                    # all files into the same dir would also not work since
+                    # we need to exclude any symlinks within dirs with the
+                    # --no-follow-symlinks flag.
+                    self.upload_local_path(source_path,
+                                           create_dir=os.path.isdir(
+                                               os.path.expanduser(source_path)))
+            elif self.source is not None:
                 if self.source.startswith('s3://'):
                     pass
                 elif self.source.startswith('gs://'):
                     self._transfer_to_s3()
                 else:
-                    self.sync_local_dir()
+                    self.upload_local_path(self.source)
         except exceptions.StorageUploadError:
             raise
         except Exception as e:
@@ -840,19 +885,33 @@ def delete(self) -> None:
     def get_handle(self) -> StorageHandle:
         return aws.resource('s3').Bucket(self.name)
 
-    def sync_local_dir(self) -> None:
-        """Syncs a local directory to a S3 bucket.
+    def upload_local_path(self,
+                          local_path: str,
+                          create_dir: bool = False) -> None:
+        """Invokes aws cli to upload a local path to a S3 bucket.
 
         AWS Sync by default uses 10 threads to upload files to the bucket.  To
         increase parallelism, modify max_concurrent_requests in your aws config
         file (Default path: ~/.aws/config).
         """
-        source = os.path.abspath(os.path.expanduser(self.source))
-        sync_command = ('aws s3 sync --no-follow-symlinks '
-                        f'{source} s3://{self.name}/')
+        full_local_path = os.path.abspath(os.path.expanduser(local_path))
+        if os.path.isfile(full_local_path):
+            file_name = os.path.basename(full_local_path)
+            dir_path = os.path.dirname(full_local_path)
+            sync_command = ('aws s3 sync --no-follow-symlinks --exclude="*" '
+                            f'--include="{file_name}" {dir_path} '
+                            f's3://{self.name}')
+        else:
+            if create_dir:
+                dest_dir_name = os.path.basename(full_local_path)
+            else:
+                dest_dir_name = ''
+            sync_command = ('aws s3 sync --no-follow-symlinks '
+                            f'{full_local_path} '
+                            f's3://{self.name}/{dest_dir_name}')
         with backend_utils.safe_console_status(
                 f'[bold cyan]Syncing '
-                f'[green]{self.source}[/] to [green]s3://{self.name}/[/]'):
+                f'[green]{local_path}[/] to [green]s3://{self.name}/[/]'):
             # TODO(zhwu): Use log_lib.run_with_log() and redirect the output
             # to a log file.
             with subprocess.Popen(sync_command.split(' '),
@@ -880,7 +939,7 @@ def sync_local_dir(self) -> None:
                         logger.error(stderr)
                         raise exceptions.StorageUploadError(
                             f'Upload to S3 failed for store {self.name} and '
-                            f'source {self.source}. Please check the logs.')
+                            f'source {local_path}. Please check the logs.')
 
     def _transfer_to_s3(self) -> None:
         if self.source.startswith('gs://'):
@@ -919,7 +978,7 @@ def _get_bucket(self) -> Tuple[StorageHandle, bool]:
                         _BUCKET_FAIL_TO_CONNECT_MESSAGE.format(name=self.name) +
                         f' To debug, consider using {command}.') from e
 
-        if self.source is not None and self.source.startswith('s3://'):
+        if isinstance(self.source, str) and self.source.startswith('s3://'):
             with ux_utils.print_exception_no_traceback():
                 raise exceptions.StorageBucketGetError(
                     'Attempted to connect to a non-existent bucket: '
@@ -1032,17 +1091,22 @@ def __init__(self,
 
     def _validate(self):
         if self.source is not None:
-            if self.source.startswith('s3://'):
-                assert self.name == data_utils.split_s3_path(self.source)[0], (
-                    'S3 Bucket is specified as path, the name should be the '
-                    'same as S3 bucket.')
-                assert data_utils.verify_s3_bucket(self.name), (
-                    f'Source specified as {self.source}, an S3 bucket. ',
-                    'S3 Bucket should exist.')
-            elif self.source.startswith('gs://'):
-                assert self.name == data_utils.split_gcs_path(self.source)[0], (
-                    'GCS Bucket is specified as path, the name should be the '
-                    'same as GCS bucket.')
+            if isinstance(self.source, str):
+                if self.source.startswith('s3://'):
+                    assert self.name == data_utils.split_s3_path(
+                        self.source
+                    )[0], (
+                        'S3 Bucket is specified as path, the name should be the'
+                        ' same as S3 bucket.')
+                    assert data_utils.verify_s3_bucket(self.name), (
+                        f'Source specified as {self.source}, an S3 bucket. ',
+                        'S3 Bucket should exist.')
+                elif self.source.startswith('gs://'):
+                    assert self.name == data_utils.split_gcs_path(
+                        self.source
+                    )[0], (
+                        'GCS Bucket is specified as path, the name should be '
+                        'the same as GCS bucket.')
 
     def initialize(self):
         """Initializes the GCS store object on the cloud.
@@ -1074,13 +1138,22 @@ def upload(self):
             StorageUploadError: if upload fails.
         """
         try:
-            if self.source is not None:
+            if isinstance(self.source, list):
+                for source_path in self.source:
+                    # We upload each file in the list as individually because
+                    # gsutil rsync does not support a list of files. Symlinking
+                    # all files into the same dir would also not work since
+                    # gsutil does not support following dir symlinks.
+                    self.upload_local_path(source_path,
+                                           create_dir=os.path.isdir(
+                                               os.path.expanduser(source_path)))
+            elif self.source is not None:
                 if self.source.startswith('gs://'):
                     pass
                 elif self.source.startswith('s3://'):
                     self._transfer_to_gcs()
                 else:
-                    self.sync_local_dir()
+                    self.upload_local_path(self.source)
         except exceptions.StorageUploadError:
             raise
         except Exception as e:
@@ -1094,15 +1167,31 @@ def delete(self) -> None:
     def get_handle(self) -> StorageHandle:
         return self.client.get_bucket(self.name)
 
-    def sync_local_dir(self) -> None:
-        """Syncs a local directory to a GCS bucket."""
-        source = os.path.abspath(os.path.expanduser(self.source))
+    def upload_local_path(self,
+                          local_path: str,
+                          create_dir: bool = False) -> None:
+        """Invokes gsutil to upload a local path to a GCS bucket."""
+        full_local_path = os.path.abspath(os.path.expanduser(local_path))
         # TODO(zhwu): Speed up the upload process by providing a list of files
         # so that we can utilize the parallelism.
-        sync_command = f'gsutil -m rsync -d -r {source} gs://{self.name}/'
+        if os.path.isfile(full_local_path):
+            # If source is a file, use negative lookahead regex to upload only
+            # specified file since gsutil does not support including only
+            # specific files in rsync.
+            file_name = os.path.basename(full_local_path)
+            dir_path = os.path.dirname(full_local_path)
+            sync_command = (f'gsutil -m rsync -x "(?!^{file_name}$)" '
+                            f'{dir_path} gs://{self.name}')
+        else:
+            if create_dir:
+                dest_dir_name = os.path.basename(full_local_path)
+            else:
+                dest_dir_name = ''
+            sync_command = (f'gsutil -m rsync -r {full_local_path} '
+                            f'gs://{self.name}/{dest_dir_name}')
         with backend_utils.safe_console_status(
                 f'[bold cyan]Syncing '
-                f'[green]{self.source}[/] to [green]gs://{self.name}/[/]'):
+                f'[green]{local_path}[/] to [green]gs://{self.name}/[/]'):
             with subprocess.Popen(sync_command.split(' '),
                                   stderr=subprocess.PIPE,
                                   stdout=subprocess.DEVNULL) as process:
@@ -1128,7 +1217,7 @@ def sync_local_dir(self) -> None:
                         logger.error(stderr)
                         raise exceptions.StorageUploadError(
                             f'Upload to GCS failed for store {self.name} and '
-                            f'source {self.source}. Please check logs.')
+                            f'source {local_path}. Please check logs.')
 
     def _transfer_to_gcs(self) -> None:
         if self.source.startswith('s3://'):
@@ -1150,7 +1239,7 @@ def _get_bucket(self) -> Tuple[StorageHandle, bool]:
             bucket = self.client.get_bucket(self.name)
             return bucket, False
         except gcp.not_found_exception() as e:
-            if self.source is not None and self.source.startswith('gs://'):
+            if isinstance(self.source, str) and self.source.startswith('gs://'):
                 with ux_utils.print_exception_no_traceback():
                     raise exceptions.StorageBucketGetError(
                         'Attempted to connect to a non-existent bucket: '
diff --git a/sky/utils/schemas.py b/sky/utils/schemas.py
index 3493ebaff7e..06e7047effd 100644
--- a/sky/utils/schemas.py
+++ b/sky/utils/schemas.py
@@ -83,7 +83,10 @@ def get_storage_schema():
                 'type': 'string',
             },
             'source': {
-                'type': 'string',
+                'type': ['string', 'array'],
+                'items': {
+                    'type': 'string'
+                }
             },
             'store': {
                 'type': 'string',
diff --git a/tests/test_smoke.py b/tests/test_smoke.py
index 51a0861c802..113fd84622d 100644
--- a/tests/test_smoke.py
+++ b/tests/test_smoke.py
@@ -901,6 +901,15 @@ def tmp_local_storage_obj(self, tmp_bucket_name, tmp_source):
         yield from self.yield_storage_object(name=tmp_bucket_name,
                                              source=tmp_source)
 
+    @pytest.fixture
+    def tmp_local_list_storage_obj(self, tmp_bucket_name, tmp_source):
+        # Creates a temp storage object which uses a list of paths as source.
+        # Stores must be added in the test. After upload, the bucket should
+        # have two files - /tmp-file and /tmp-source/tmp-file
+        list_source = [tmp_source, tmp_source + '/tmp-file']
+        yield from self.yield_storage_object(name=tmp_bucket_name,
+                                             source=list_source)
+
     @pytest.fixture
     def tmp_copy_mnt_existing_storage_obj(self, tmp_scratch_storage_obj):
         # Creates a copy mount storage which reuses an existing storage object.
@@ -1029,11 +1038,11 @@ def test_private_bucket(self, private_bucket):
             storage_obj = storage_lib.Storage(source=private_bucket)
 
     @staticmethod
-    def cli_ls_cmd(store_type, bucket_name):
+    def cli_ls_cmd(store_type, bucket_name, suffix=''):
         if store_type == storage_lib.StoreType.S3:
-            return ['aws', 's3', 'ls', f's3://{bucket_name}']
+            return ['aws', 's3', 'ls', f's3://{bucket_name}{suffix}']
         if store_type == storage_lib.StoreType.GCS:
-            return ['gsutil', 'ls', f'gs://{bucket_name}']
+            return ['gsutil', 'ls', f'gs://{bucket_name}{suffix}']
 
     @pytest.mark.parametrize('ext_bucket_fixture, store_type',
                              [('tmp_awscli_bucket', storage_lib.StoreType.S3),
@@ -1076,6 +1085,28 @@ def test_copy_mount_existing_storage(self,
         out = subprocess.check_output(['sky', 'storage', 'ls']).decode('utf-8')
         assert storage_name in out, f'Storage {storage_name} not found in sky storage ls.'
 
+    @pytest.mark.parametrize(
+        'store_type', [storage_lib.StoreType.S3, storage_lib.StoreType.GCS])
+    def test_list_source(self, tmp_local_list_storage_obj, store_type):
+        # Uses a list in the source field to specify a file and a directory to
+        # be uploaded to the storage object.
+        tmp_local_list_storage_obj.add_store(store_type)
+
+        # Check if tmp-file exists in the bucket root using cli
+        out = subprocess.check_output(
+            self.cli_ls_cmd(store_type, tmp_local_list_storage_obj.name))
+        assert 'tmp-file' in out.decode('utf-8'), \
+            'File not found in bucket - output was : {}'.format(out.decode
+                                                                ('utf-8'))
+
+        # Check if tmp-file exists in the bucket/tmp-source using cli
+        out = subprocess.check_output(
+            self.cli_ls_cmd(store_type, tmp_local_list_storage_obj.name,
+                            '/tmp-source/'))
+        assert 'tmp-file' in out.decode('utf-8'), \
+            'File not found in bucket - output was : {}'.format(out.decode
+                                                                ('utf-8'))
+
 
 # ---------- Testing YAML Specs ----------
 # Our sky storage requires credentials to check the bucket existance when


Code review comments:
diff_hunk: @@ -1074,13 +1138,22 @@ def upload(self):
             StorageUploadError: if upload fails.
         """
         try:
-            if self.source is not None:
+            if isinstance(self.source, list):
+                for source_path in self.source:
+                    # We upload each file in the list as individually because
+                    # gsutil rsync does not support a list of files. Symlinking
+                    # all files into the same dir would also not work since
+                    # gsutil does not support following dir symlinks.
path: sky/data/storage.py
text: @user1:
Out of curiosity, why we are using `gsutil rsync` instead of `gsutil cp`? It seems that `gsutil cp` supports the following, which will significantly improve the uploading speed when multiple files are being uploaded. https://cloud.google.com/storage/docs/gsutil/commands/cp
```
cat filelist | gsutil -m cp -I gs://my-bucket
```

@author:
Great question - `gsutil cp` will force upload a file even if it already exists on the bucket every time the task is launched. `gsutil rsync` uploads only files which either don't exist on the bucket or have been modified locally. This reduces upload time and number of operations required to sync the storage object. 

@user1:
Thanks for the explanation!
The reason I mentioned this is that for the salk users, we did observe that looping through all the files to be uploaded with `gsutil rsync` or `gsutil cp` was much slower than `cat filelist | gsutil -m cp -I gs://my-bucket/subdir`(10 min vs 3 min for more than 200 files). That is to say `sky spot launch` will be blocked by the uploading of the files.
It seems `gsutil cp` supports a `-n` option that avoids overwriting the files, but if that does not have the same semantic as `gsutil rsync`, I think it should be fine to keep the current way.

@author:
Oh that's a great point. It seems `gsutil cp -n` should do what we need. Let me try it out and update the PR if it works

diff_hunk: @@ -461,57 +475,72 @@ def _validate_source(source: str, mode: StorageMode,
           source: str; The source path.
           is_local_path: bool; Whether the source is a local path. False if URI.
         """
-        # Check if source is a valid local/remote URL
-        split_path = urllib.parse.urlsplit(source)
-        if split_path.scheme == '':
-            if source.endswith('/'):
+
+        def _validate_local_source(local_source):
+            if local_source.endswith('/'):
                 with ux_utils.print_exception_no_traceback():
                     raise exceptions.StorageSourceError(
                         'Storage source paths cannot end with a slash '
                         '(try "/mydir: /mydir" or "/myfile: /myfile"). '
-                        f'Found source={source}')
+                        f'Found source={local_source}')
             # Local path, check if it exists
-            full_src = os.path.abspath(os.path.expanduser(source))
+            full_src = os.path.abspath(os.path.expanduser(local_source))
             # Only check if local source exists if it is synced to the bucket
             if not os.path.exists(full_src) and sync_on_reconstruction:
                 with ux_utils.print_exception_no_traceback():
                     raise exceptions.StorageSourceError(
                         'Local source path does not'
-                        f' exist: {source}')
-            # Check if source is a file - throw error if it is
-            if os.path.isfile(full_src):
-                with ux_utils.print_exception_no_traceback():
-                    raise exceptions.StorageSourceError(
-                        'Storage source path cannot be a file - only '
-                        'directories are supported as a source. '
-                        'To upload a single file, either\n'
-                        '1) Use regular file mounts by writing '
-                        f'<destination_path>: {source} in the file_mounts '
-                        'section of your YAML, or\n2) Place the file in a '
-                        'directory and specify the directory as the source.')
+                        f' exist: {local_source}')
             # Raise warning if user's path is a symlink
             elif os.path.islink(full_src):
                 logger.warning(f'Source path {source} is a symlink. '
                                'Referenced contents are uploaded, matching '
                                'the default behavior for S3 and GCS syncing.')
+
+        # Check if source is a list of paths
+        if isinstance(source, list):
+            # Validate each path
+            for local_source in source:
+                _validate_local_source(local_source)
             is_local_source = True
-        elif split_path.scheme in ['s3', 'gs']:
-            is_local_source = False
-            # Storage mounting does not support mounting specific files from
-            # cloud store - ensure path points to only a directory
-            if mode == StorageMode.MOUNT:
-                if split_path.path.strip('/') != '':
-                    with ux_utils.print_exception_no_traceback():
-                        raise exceptions.StorageModeError(
-                            'MOUNT mode does not support'
-                            ' mounting specific files from cloud'
-                            ' storage. Please use COPY mode or'
-                            ' specify only the bucket name as'
-                            ' the source.')
         else:
-            with ux_utils.print_exception_no_traceback():
-                raise exceptions.StorageSourceError(
-                    f'Supported paths: local, s3://, gs://. Got: {source}')
+            # Check if str source is a valid local/remote URL
+            split_path = urllib.parse.urlsplit(source)
+            if split_path.scheme == '':
+                _validate_local_source(source)
+                # Check if source is a file - throw error if it is
+                full_src = os.path.abspath(os.path.expanduser(source))
+                if os.path.isfile(full_src):
+                    with ux_utils.print_exception_no_traceback():
+                        raise exceptions.StorageSourceError(
+                            'Storage source path cannot be a file - only'
+                            ' directories are supported as a source. '
+                            'To upload a single file, specify it in a list '
+                            f'by writing <destination_path>: [{source}]. Note '
path: sky/data/storage.py
text: @user1:
Should this be `source: [{source}]` instead, as `<destination_path>: [{source}]` is not a legal schema in our YAML? Or we can have the full specification:
```
<destination_path>:
    source: [{source}]
    name: some-bucket
```

@author:
Ahh good point - fixed!

diff_hunk: @@ -334,7 +345,7 @@ def remove_store(self, store: AbstractStore) -> None:
 
     def __init__(self,
                  name: Optional[str] = None,
-                 source: Optional[Path] = None,
+                 source: Optional[Union[Path, List[Path]]] = None,
path: sky/data/storage.py
text: @user1:
very small nit: `Union[Path, List[Path], None]`
https://google.github.io/styleguide/pyguide.html#3195-nonetype

@author:
good catch!