Repository: mlflow/mlflow
Pull request: Fix handling of empty directories in fs based artifact repositories.
Fixes #1881. 

## What changes are proposed in this pull request?
Distinguish between empty directories and files when downloading artifacts from filesystem based artifact repositories.
   
## How is this patch tested?
Enhanced unit test  for local artifact repository. 

## Release Notes
Distinguish between empty directories and files when downloading artifacts from filesystem based artifact repositories.


### Is this a user-facing change? 

- [ ] No. You can skip the rest of this section.
- [x] Yes. Give a description of this change to be included in the release notes for MLflow users.
  
### What component(s) does this PR affect?
 
- [ ] UI
- [ ] CLI 
- [ ] API 
- [ ] REST-API 
- [ ] Examples 
- [ ] Docs
- [ ] Tracking
- [ ] Projects 
- [x] Artifacts 
- [ ] Models 
- [ ] Scoring 
- [ ] Serving
- [ ] R
- [ ] Java
- [ ] Python

### How should the PR be classified in the release notes? Choose one:
 
- [ ] `rn/breaking-change` - The PR will be mentioned in the "Breaking Changes" section
- [x] `rn/none` - No description will be included. The PR will be mentioned only by the PR number in the "Small Bugfixes and Documentation Updates" section
- [ ] `rn/feature` - A new user-facing feature worth mentioning in the release notes
- [ ] `rn/bug-fix` - A user-facing bug fix worth mentioning in the release notes
- [ ] `rn/documentation` - A user-facing documentation change worth mentioning in the release notes


Problem statement:
[BUG] Spark UDF cannot load models with empty directories
### System information
- **Have I written custom code (as opposed to using a stock example script provided in MLflow)**: No
- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**: Linux
- **MLflow installed from (source or binary)**:  Binary
- **MLflow version (run ``mlflow --version``)**: 1.2.0
- **Python version**: 
- **npm version, if running the dev UI**:
- **Exact command to reproduce**:

### Describe the problem
Loading a MLflow model with empty directories as a Spark UDF raises an IsADirectoryError. I encountered this for TensorFlow models without variables (which is the case for models with all variables converted to constants), but it looks like it will be triggered for any MLflow model with an empty directory (untested).

```
IsADirectoryError: [Errno 21] Is a directory: 'path/to/model/tfmodel/variables'

/databricks/python/lib/python3.7/site-packages/mlflow/pyfunc/__init__.py in spark_udf(spark, model_uri, result_type)
    417     with TempDir() as local_tmpdir:
    418         local_model_path = _download_artifact_from_uri(
--> 419             artifact_uri=model_uri, output_path=local_tmpdir.path())
    420         archive_path = SparkModelCache.add_local_model(spark, local_model_path)
    421 

/databricks/python/lib/python3.7/site-packages/mlflow/tracking/artifact_utils.py in _download_artifact_from_uri(artifact_uri, output_path)
     64     root_uri = prefix + urllib.parse.urlunparse(parsed_uri)
     65     return get_artifact_repository(artifact_uri=root_uri).download_artifacts(
---> 66         artifact_path=artifact_path, dst_path=output_path)
```

### Code to reproduce issue
```
import tensorflow as tf
import mlflow
import mlflow.tensorflow
import mlflow.pyfunc

inputs = tf.placeholder(tf.float32, [None])
outputs = inputs/2 + 2
signature = (
    tf.saved_model.signature_def_utils.build_signature_def(
        inputs={"inputs": tf.saved_model.utils.build_tensor_info(inputs)},
        outputs={"outputs": tf.saved_model.utils.build_tensor_info(outputs)},
        method_name="predict"
    )
)

with tf.Session() as sess:
    builder = tf.saved_model.builder.SavedModelBuilder("saved_model")
    builder.add_meta_graph_and_variables(
        sess,
        [tf.saved_model.tag_constants.SERVING],
        signature_def_map={"predict": signature}
    )
    builder.save()

mlflow.tensorflow.save_model(tf_saved_model_dir="saved_model",
                             tf_meta_graph_tags=[tf.saved_model.tag_constants.SERVING],
                             tf_signature_def_key="predict",
                             path="model")

udf = mlflow.pyfunc.spark_udf(spark=spark, model_uri="model")
```

### Other info / logs
Include any logs or source code that would be helpful to diagnose the problem. If including tracebacks,
please include the full traceback. Large logs and files should be attached.

Hints from issue discussion:
Hey @dwyatte, thanks for posting! I have been able to reproduce this. The issue occurs when downloading artifacts which include empty directory.  
This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.

Commit message: fix

Patch to review:
diff --git a/mlflow/store/artifact/artifact_repo.py b/mlflow/store/artifact/artifact_repo.py
index b92ca185e232a..d72e3c1e7a4a1 100644
--- a/mlflow/store/artifact/artifact_repo.py
+++ b/mlflow/store/artifact/artifact_repo.py
@@ -57,6 +57,10 @@ def list_artifacts(self, path):
         """
         pass
 
+    def _is_directory(self, artifact_path):
+        listing = self.list_artifacts(artifact_path)
+        return len(listing) > 0
+
     def download_artifacts(self, artifact_path, dst_path=None):
         """
         Download an artifact file or directory to a local directory if applicable, and return a
@@ -75,28 +79,48 @@ def download_artifacts(self, artifact_path, dst_path=None):
 
         # TODO: Probably need to add a more efficient method to stream just a single artifact
         #       without downloading it, or to get a pre-signed URL for cloud storage.
-
-        def download_artifacts_into(artifact_path, dest_dir):
-            basename = posixpath.basename(artifact_path)
-            local_path = os.path.join(dest_dir, basename)
-            listing = self.list_artifacts(artifact_path)
-            if len(listing) > 0:
-                # Artifact_path is a directory, so make a directory for it and download everything
-                if not os.path.exists(local_path):
-                    os.mkdir(local_path)
-                for file_info in listing:
-                    # prevent an infinite loop (sometimes the current path is listed e.g. as ".")
-                    if file_info.path == "." or file_info.path == artifact_path:
-                        continue
-                    download_artifacts_into(artifact_path=file_info.path, dest_dir=local_path)
-            else:
-                self._download_file(remote_file_path=artifact_path, local_path=local_path)
-            return local_path
-
         if dst_path is None:
             dst_path = tempfile.mkdtemp()
         dst_path = os.path.abspath(dst_path)
+        if not os.path.exists(dst_path):
+            raise MlflowException(
+                message=(
+                    "The destination path for downloaded artifacts does not"
+                    " exist! Destination path: {dst_path}".format(dst_path=dst_path)),
+                error_code=RESOURCE_DOES_NOT_EXIST)
+        elif not os.path.isdir(dst_path):
+            raise MlflowException(
+                message=(
+                    "The destination path for downloaded artifacts must be a directory!"
+                    " Destination path: {dst_path}".format(dst_path=dst_path)),
+                error_code=INVALID_PARAMETER_VALUE)
 
+        def download_file(fullpath):
+            dirpath, _ = posixpath.split(fullpath)
+            local_dir_path = os.path.join(dst_path, dirpath)
+            local_file_path = os.path.join(dst_path, fullpath)
+            # Artifact_path is a directory, so make a directory for it and download everything
+            if not os.path.exists(local_dir_path):
+                os.makedirs(local_dir_path)
+            self._download_file(remote_file_path=fullpath, local_path=local_file_path)
+            return local_file_path
+
+        def download_artifact_dir(dir_path):
+            local_dir = os.path.join(dst_path, dir_path)
+            dir_content = self.list_artifacts(dir_path)
+            if not dir_content:  # empty dir
+                if not os.path.exists(local_dir):
+                    os.makedirs(local_dir)
+            else:
+                for file_info in dir_content:
+                    # prevent an infinite loop (sometimes the current path is listed e.g. as ".")
+                    if file_info.path == "." or file_info.path == dir_path:
+                        continue
+                    if file_info.is_dir:
+                        download_artifact_dir(dir_path=file_info.path)
+                    else:
+                        download_file(file_info.path)
+            return local_dir
         if not os.path.exists(dst_path):
             raise MlflowException(
                 message=(
@@ -110,7 +134,11 @@ def download_artifacts_into(artifact_path, dest_dir):
                     " Destination path: {dst_path}".format(dst_path=dst_path)),
                 error_code=INVALID_PARAMETER_VALUE)
 
-        return download_artifacts_into(artifact_path, dst_path)
+        # Check if the artifacts points to a directory
+        if self._is_directory(artifact_path):
+            return download_artifact_dir(artifact_path)
+        else:
+            return download_file(artifact_path)
 
     @abstractmethod
     def _download_file(self, remote_file_path, local_path):
diff --git a/mlflow/store/artifact/dbfs_artifact_repo.py b/mlflow/store/artifact/dbfs_artifact_repo.py
index 20c48c29530f8..ed6b2d791209a 100644
--- a/mlflow/store/artifact/dbfs_artifact_repo.py
+++ b/mlflow/store/artifact/dbfs_artifact_repo.py
@@ -52,6 +52,13 @@ def _dbfs_download(self, output_path, endpoint):
             finally:
                 response.close()
 
+    def _is_directory(self, artifact_path):
+        if artifact_path:
+            dbfs_path = self._get_dbfs_path(artifact_path)
+        else:
+            dbfs_path = self._get_dbfs_path('')
+        return self._dbfs_is_dir(dbfs_path)
+
     def _dbfs_is_dir(self, dbfs_path):
         response = self._databricks_api_request(
             endpoint=GET_STATUS_ENDPOINT, method='GET', json={'path': dbfs_path})
diff --git a/mlflow/store/artifact/ftp_artifact_repo.py b/mlflow/store/artifact/ftp_artifact_repo.py
index 325a12a022e07..4aed20bca2422 100644
--- a/mlflow/store/artifact/ftp_artifact_repo.py
+++ b/mlflow/store/artifact/ftp_artifact_repo.py
@@ -97,6 +97,12 @@ def log_artifacts(self, local_dir, artifact_path=None):
                 if os.path.isfile(os.path.join(root, f)):
                     self.log_artifact(os.path.join(root, f), upload_path)
 
+    def _is_directory(self, artifact_path):
+        artifact_dir = self.path
+        list_dir = posixpath.join(artifact_dir, artifact_path) if artifact_path else artifact_dir
+        with self.get_ftp_client() as ftp:
+            return self._is_dir(ftp, list_dir)
+
     def list_artifacts(self, path=None):
         with self.get_ftp_client() as ftp:
             artifact_dir = self.path
diff --git a/mlflow/store/artifact/local_artifact_repo.py b/mlflow/store/artifact/local_artifact_repo.py
index 83cf432b73bbd..3fb67a9c2ae42 100644
--- a/mlflow/store/artifact/local_artifact_repo.py
+++ b/mlflow/store/artifact/local_artifact_repo.py
@@ -31,6 +31,13 @@ def log_artifact(self, local_file, artifact_path=None):
             mkdir(artifact_dir)
         shutil.copy(local_file, artifact_dir)
 
+    def _is_directory(self, artifact_path):
+        # NOTE: The path is expected to be in posix format.
+        # Posix paths work fine on windows but just in case we normalize it here.
+        path = os.path.normpath(artifact_path) if artifact_path else ""
+        list_dir = os.path.join(self.artifact_dir, path) if path else self.artifact_dir
+        return os.path.isdir(list_dir)
+
     def log_artifacts(self, local_dir, artifact_path=None):
         verify_artifact_path(artifact_path)
         # NOTE: The artifact_path is expected to be in posix format.
diff --git a/mlflow/store/artifact/runs_artifact_repo.py b/mlflow/store/artifact/runs_artifact_repo.py
index db34b80b40de6..602829581aa9d 100644
--- a/mlflow/store/artifact/runs_artifact_repo.py
+++ b/mlflow/store/artifact/runs_artifact_repo.py
@@ -84,6 +84,9 @@ def log_artifacts(self, local_dir, artifact_path=None):
         """
         self.repo.log_artifacts(local_dir, artifact_path)
 
+    def _is_directory(self, artifact_path):
+        return self.repo._is_directory(artifact_path)
+
     def list_artifacts(self, path):
         """
         Return all the artifacts for this run_id directly under path. If path is a file, returns
diff --git a/mlflow/store/artifact/sftp_artifact_repo.py b/mlflow/store/artifact/sftp_artifact_repo.py
index 981253925b69d..2143944998fa8 100644
--- a/mlflow/store/artifact/sftp_artifact_repo.py
+++ b/mlflow/store/artifact/sftp_artifact_repo.py
@@ -71,6 +71,11 @@ def log_artifacts(self, local_dir, artifact_path=None):
         self.sftp.makedirs(artifact_dir)
         self.sftp.put_r(local_dir, artifact_dir)
 
+    def _is_directory(self, artifact_path):
+        artifact_dir = self.path
+        path = posixpath.join(artifact_dir, artifact_path) if artifact_path else artifact_dir
+        return self.sftp.isdir(path)
+
     def list_artifacts(self, path=None):
         artifact_dir = self.path
         list_dir = posixpath.join(artifact_dir, path) if path else artifact_dir
diff --git a/tests/store/artifact/test_artifact_repo.py b/tests/store/artifact/test_artifact_repo.py
index 87e285d81182c..531051d06cf4c 100644
--- a/tests/store/artifact/test_artifact_repo.py
+++ b/tests/store/artifact/test_artifact_repo.py
@@ -1,3 +1,4 @@
+import posixpath
 import mock
 import pytest
 
@@ -17,7 +18,10 @@ def list_artifacts(self, path):
         raise NotImplementedError()
 
     def _download_file(self, remote_file_path, local_path):
-        print("download_file called with ", remote_file_path)
+        print("download_file called with '%s'" % remote_file_path)
+        if not remote_file_path.endswith("modelfile"):
+            raise Exception("haha")
+        assert remote_file_path.endswith("modelfile")
 
 
 @pytest.mark.parametrize("base_uri, download_arg, list_return_val", [
@@ -29,10 +33,30 @@ def _download_file(self, remote_file_path, local_path):
     ('', '12345/model', ['12345/model', '12345/model/modelfile']),
 ])
 def test_download_artifacts_does_not_infinitely_loop(base_uri, download_arg, list_return_val):
+    def list_artifacts(path):
+        fullpath = posixpath.join(base_uri, path)
+        if fullpath.endswith("model") or fullpath.endswith("model/"):
+            return [FileInfo(item, False, 123) for item in list_return_val]
+        elif fullpath.endswith("12345") or fullpath.endswith("12345/"):
+            return [FileInfo(posixpath.join(path, "model"), True, 0)]
+        else:
+            return []
 
+    with mock.patch.object(ArtifactRepositoryImpl, "list_artifacts") as list_artifacts_mock:
+        list_artifacts_mock.side_effect = list_artifacts
+        repo = ArtifactRepositoryImpl(base_uri)
+        repo.download_artifacts(download_arg)
+
+
+@pytest.mark.parametrize("base_uri, download_arg, list_return_val", [
+    ('', '12345/model', ['12345/model', '12345/model/modelfile', '12345/model/emptydir']),
+])
+def test_download_artifacts_handles_empty_dir(base_uri, download_arg, list_return_val):
     def list_artifacts(path):
         if path.endswith("model"):
-            return [FileInfo(item, False, 123) for item in list_return_val]
+            return [FileInfo(item, item.endswith("emptydir"), 123) for item in list_return_val]
+        elif path.endswith("12345") or path.endswith("12345/"):
+            return [FileInfo("12345/model", True, 0)]
         else:
             return []
 
diff --git a/tests/store/artifact/test_dbfs_rest_artifact_repo.py b/tests/store/artifact/test_dbfs_rest_artifact_repo.py
index 88ef112dfedbd..7bb42f9fd3020 100644
--- a/tests/store/artifact/test_dbfs_rest_artifact_repo.py
+++ b/tests/store/artifact/test_dbfs_rest_artifact_repo.py
@@ -208,15 +208,13 @@ def test_download_artifacts(self, dbfs_artifact_repo):
                 Mock(text='{}')   # this call is for listing `/dir/a.txt`.
             ]
             dbfs_artifact_repo.download_artifacts('/')
-            assert list_mock.call_count == 3
-            assert download_mock.call_count == 2
+            assert list_mock.call_count == 2
+            assert download_mock.call_count == 1
             chronological_download_calls = list(download_mock.call_args_list)
             # Calls are in reverse chronological order by default
             chronological_download_calls.reverse()
-            _, kwargs_call_1 = chronological_download_calls[0]
-            _, kwargs_call_2 = chronological_download_calls[1]
-            assert kwargs_call_1['endpoint'] == '/dbfs/test/dir'
-            assert kwargs_call_2['endpoint'] == '/dbfs/test/a.txt'
+            _, kwargs_call = chronological_download_calls[0]
+            assert kwargs_call['endpoint'] == '/dbfs/test/a.txt'
 
 
 def test_get_host_creds_from_default_store_file_store():
diff --git a/tests/store/artifact/test_ftp_artifact_repo.py b/tests/store/artifact/test_ftp_artifact_repo.py
index 55242e29228ba..97fd414f34bf2 100644
--- a/tests/store/artifact/test_ftp_artifact_repo.py
+++ b/tests/store/artifact/test_ftp_artifact_repo.py
@@ -225,6 +225,7 @@ def test_download_artifacts(ftp_mock):
     # mocked file structure
     #  |- model
     #     |- model.pb
+    #     |- empty_dir
     #     |- variables
     #        |- test.txt
     dir_path = posixpath.join(artifact_root_path, 'model')
@@ -232,6 +233,8 @@ def test_download_artifacts(ftp_mock):
     # list artifacts at sub directory level
     model_file_path_sub = 'model.pb'
     model_file_path_full = posixpath.join(dir_path, model_file_path_sub)
+    empty_dir_name = "empty_dir"
+    empty_dir_path = posixpath.join(dir_path, empty_dir_name)
     subdir_name = 'variables'
     subdir_path_full = posixpath.join(dir_path, subdir_name)
     subfile_name = 'test.txt'
@@ -239,29 +242,41 @@ def test_download_artifacts(ftp_mock):
 
     is_dir_mapping = {
         dir_path: True,
+        empty_dir_path: True,
         model_file_path_full: False,
         subdir_path_full: True,
         subfile_path_full: False,
     }
 
     is_dir_call_args = [
-        dir_path, model_file_path_full, subdir_path_full,
+        dir_path, model_file_path_full, empty_dir_path, subdir_path_full,
         model_file_path_full,
         subdir_path_full, subfile_path_full,
         subfile_path_full
     ]
 
-    cwd_side_effect = [
-        None if is_dir_mapping[call_arg] else ftplib.error_perm for call_arg in is_dir_call_args
-    ]
+    def cwd_side_effect(call_arg):
+        if not is_dir_mapping[call_arg]:
+            raise ftplib.error_perm
     ftp_mock.cwd = MagicMock(side_effect=cwd_side_effect)
-    ftp_mock.nlst = MagicMock(side_effect=[[model_file_path_sub, subdir_name], [subfile_name]])
 
+    def nlst_side_effect(call_arg):
+        if call_arg == dir_path:
+            return [model_file_path_sub, subdir_name, empty_dir_name]
+        elif call_arg == subdir_path_full:
+            return [subfile_name]
+        elif call_arg == empty_dir_path:
+            return []
+        else:
+            raise Exception("should never call nlst for non-directories {}".format(call_arg))
+
+    ftp_mock.nlst = MagicMock(side_effect=nlst_side_effect)
     repo.download_artifacts("model")
 
     cwd_call_args = [arg_entry[0][0] for arg_entry in ftp_mock.cwd.call_args_list]
-    assert cwd_call_args == is_dir_call_args
-    assert ftp_mock.nlst.call_count == 2
+
+    assert set(cwd_call_args) == set(is_dir_call_args)
+    assert ftp_mock.nlst.call_count == 3
     assert ftp_mock.retrbinary.call_args_list[0][0][0] == 'RETR ' + model_file_path_full
     assert ftp_mock.retrbinary.call_args_list[1][0][0] == 'RETR ' + subfile_path_full
 
diff --git a/tests/store/artifact/test_local_artifact_repo.py b/tests/store/artifact/test_local_artifact_repo.py
index da531fd3fd6e0..98d0d70238e2e 100644
--- a/tests/store/artifact/test_local_artifact_repo.py
+++ b/tests/store/artifact/test_local_artifact_repo.py
@@ -41,7 +41,7 @@ def test_log_artifacts(local_artifact_repo, local_artifact_root):
 
     artifacts_list = local_artifact_repo.list_artifacts()
     assert len(artifacts_list) == 1
-    assert artifacts_list[0].path == artifact_rel_path
+    assert (artifacts_list[0].path == artifact_rel_path)
 
     artifact_dst_path = os.path.join(local_artifact_root, artifact_rel_path)
     assert os.path.exists(artifact_dst_path)
@@ -52,13 +52,19 @@ def test_log_artifacts(local_artifact_repo, local_artifact_root):
 def test_download_artifacts(local_artifact_repo):
     artifact_rel_path = "test.txt"
     artifact_text = "hello world!"
+    empty_dir_path = "empty_dir"
     with TempDir(chdr=True) as local_dir:
         artifact_src_path = local_dir.path(artifact_rel_path)
+        os.mkdir(local_dir.path(empty_dir_path))
         with open(artifact_src_path, "w") as f:
             f.write(artifact_text)
-        local_artifact_repo.log_artifact(artifact_src_path)
+        local_artifact_repo.log_artifacts(local_dir.path())
         dst_path = local_artifact_repo.download_artifacts(artifact_path=artifact_rel_path)
         assert open(dst_path).read() == artifact_text
+        dst_dir = local_artifact_repo.download_artifacts(artifact_path="")
+        empty_dir_dst_path = os.path.join(dst_dir, empty_dir_path)
+        assert os.path.isdir(empty_dir_dst_path)
+        assert len(os.listdir(empty_dir_dst_path)) == 0
 
 
 def test_download_artifacts_does_not_copy(local_artifact_repo):


Code review comments:
diff_hunk: @@ -17,7 +18,10 @@ def list_artifacts(self, path):
         raise NotImplementedError()
 
     def _download_file(self, remote_file_path, local_path):
-        print("download_file called with ", remote_file_path)
+        print("download_file called with '%s'" % remote_file_path)
+        if not remote_file_path.endswith("modelfile"):
+            raise Exception("haha")
path: tests/store/artifact/test_artifact_repo.py
text: @user1:
lol, maybe remove this? :P (since we assert in the following line anyways)

@author:
it's removed, thanks for the catch  :)