dvcfs: handle new workspace subdirectories in walk (#6811)

1. add try-catch to catch KeyError exception while walking dvc folder
2. Fix #6695

Co-authored-by: Zhenyu Gao <[email protected]>

dvc/fs/dvc.py

@@ -166,16 +166,19 @@ class DvcFileSystem(BaseFileSystem):  # pylint:disable=abstract-method
             self._add_dir(trie, out, **kwargs)
 
         root_len = len(root.parts)
-        for key, out in trie.iteritems(prefix=root.parts):  # noqa: B301
-            if key == root.parts:
-                continue
+        try:
+            for key, out in trie.iteritems(prefix=root.parts):  # noqa: B301
+                if key == root.parts:
+                    continue
 
-            name = key[root_len]
-            if len(key) > root_len + 1 or (out and out.is_dir_checksum):
-                dirs.add(name)
-                continue
+                name = key[root_len]
+                if len(key) > root_len + 1 or (out and out.is_dir_checksum):
+                    dirs.add(name)
+                    continue
 
-            files.append(name)
+                files.append(name)
+        except KeyError:
+            pass
 
         assert topdown
         dirs = list(dirs)

tests/func/test_ls.py

@@ -117,6 +117,18 @@ def test_ls_repo_dvc_only_recursive(tmp_dir, dvc, scm):
     )
 
 
+def test_ls_repo_with_new_path_dir(tmp_dir, dvc, scm):
+    tmp_dir.scm_gen(FS_STRUCTURE, commit="init")
+    tmp_dir.dvc_gen({"mysub": {}}, commit="dvc")
+    tmp_dir.gen({"mysub/sub": {"foo": "content"}})
+
+    files = Repo.ls(os.fspath(tmp_dir), path="mysub/sub")
+    match_files(
+        files,
+        ((("foo",), False),),
+    )
+
+
 def test_ls_repo_with_path_dir(tmp_dir, dvc, scm):
     tmp_dir.scm_gen(FS_STRUCTURE, commit="init")
     tmp_dir.dvc_gen(DVC_STRUCTURE, commit="dvc")