0

In-memory file system with files and folders

by · about 3 hours ago· ⑂ 0 remixes

Question: In-memory file system with files and folders

Companies: Harvey, Perplexity

Difficulty: Medium

Problem statement

Implement an in-memory file system with path-based file and folder access, plus follow-ups on per-directory limits, duplicate names and identical contents. Covers the bank's vault directory, trie-backed and API variants. Reported on 1point3acres' Harvey bank; reconstructed in our own words.

Clarify the setup

Build an in-memory file system addressed by absolute paths like /matters/acme/nda.txt:

  • mkdir(path): create a folder and any missing parents.
  • write(path, content, append=False): create or overwrite a file (creating parent folders), or append to it.
  • read(path) -> str.
  • ls(path) -> list[str]: a folder's entries sorted by name; for a file, just its name.
  • delete(path, recursive=False): deleting a non-empty folder needs recursive=True.
  • move(src, dst): rename or move a file or folder.

Follow-ups the interviewer is likely to add (part 2): a maximum number of entries per folder, automatically renaming colliding uploads (nda.txt → nda (1).txt), and finding files with identical contents. Ask about errors: raise FileNotFoundError, FileExistsError, IsADirectoryError, NotADirectoryError like the operating system does; it reads naturally and tests can check the type.

Approach

A tree of nodes. A folder node holds children: name -> node; a file node holds its content. Resolving a path walks from the root, one dictionary lookup per component, so every operation is O(depth) plus the work on the node itself (sorting for ls, copying content for read).

Keep one helper, _walk(parts, create), that returns the folder at a path (optionally creating missing ones) and raises the right error when a component is a file. Every operation is then a few lines on top of it, and all the error cases live in one place.

Move needs two checks people forget: the destination's parent must exist (or be created), and a folder cannot be moved into itself or its own descendant (/a into /a/b), which would detach it from the tree.

Part 2

  • Folder limits: check len(children) before adding a new name (overwriting an existing file does not count).
  • Unique names: if name exists, try stem (1).ext, stem (2).ext, … The first free one wins. Splitting at the last dot keeps report.final.pdf → report.final (1).pdf.
  • Identical contents: maintain an index content hash -> set of file paths, updated on every write, delete and move. Then "find duplicates" is a scan over groups with more than one path, with no reading or hashing at query time. Hash with SHA-256 so different contents never collide in practice.

Code

import hashlib
from typing import Dict, List, Optional, Set, Tuple


class _Dir:
    __slots__ = ("children",)

    def __init__(self) -> None:
        self.children: Dict[str, object] = {}


class _File:
    __slots__ = ("content",)

    def __init__(self, content: str = "") -> None:
        self.content = content


def _split(path: str) -> List[str]:
    if not path.startswith("/"):
        raise ValueError(f"absolute path required: {path!r}")
    return [p for p in path.split("/") if p]


class FileSystem:
    def __init__(self, max_entries: Optional[int] = None) -> None:
        self.root = _Dir()
        self.max_entries = max_entries
        self.by_hash: Dict[str, Set[str]] = {}

    # ---- core operations ----------------------------------------------------------
    def mkdir(self, path: str) -> None:
        self._walk(_split(path), create=True)

    def write(self, path: str, content: str, append: bool = False) -> None:
        parts = _split(path)
        if not parts:
            raise IsADirectoryError(path)
        parent = self._walk(parts[:-1], create=True)
        node = parent.children.get(parts[-1])
        if isinstance(node, _Dir):
            raise IsADirectoryError(path)
        if node is None:
            self._check_room(parent)
            node = parent.children[parts[-1]] = _File()
        else:
            self._unindex(path, node)
        node.content = node.content + content if append else content
        self._index(path, node)

    def read(self, path: str) -> str:
        node = self._node(path)
        if isinstance(node, _Dir):
            raise IsADirectoryError(path)
        return node.content

    def ls(self, path: str) -> List[str]:
        node = self._node(path)
        if isinstance(node, _File):
            return [_split(path)[-1]]
        return sorted(node.children)

    def delete(self, path: str, recursive: bool = False) -> None:
        parts = _split(path)
        if not parts:
            raise PermissionError("cannot delete the root")
        parent = self._walk(parts[:-1], create=False)
        node = parent.children.get(parts[-1])
        if node is None:
            raise FileNotFoundError(path)
        if isinstance(node, _Dir) and node.children and not recursive:
            raise OSError(f"folder not empty: {path}")
        for file_path, f in self._files_under(path, node):
            self._unindex(file_path, f)
        del parent.children[parts[-1]]

    def move(self, src: str, dst: str) -> None:
        s, d = _split(src), _split(dst)
        if not s:
            raise PermissionError("cannot move the root")
        if d[:len(s)] == s:
            raise ValueError("cannot move a folder into itself")
        src_parent = self._walk(s[:-1], create=False)
        node = src_parent.children.get(s[-1])
        if node is None:
            raise FileNotFoundError(src)
        dst_parent = self._walk(d[:-1], create=True)
        if d[-1] in dst_parent.children:
            raise FileExistsError(dst)
        self._check_room(dst_parent)
        moved = list(self._files_under(src, node))
        for p, f in moved:
            self._unindex(p, f)
        del src_parent.children[s[-1]]
        dst_parent.children[d[-1]] = node
        for p, f in moved:
            self._index(dst + p[len(src):], f)

    # ---- part 2 -------------------------------------------------------------------
    def unique_name(self, folder: str, name: str) -> str:
        node = self._node(folder)
        if not isinstance(node, _Dir):
            raise NotADirectoryError(folder)
        if name not in node.children:
            return name
        stem, dot, ext = name.rpartition(".")
        if not dot or not stem:                      # no extension, or a dotfile
            stem, ext = name, ""
        n = 1
        while True:
            candidate = f"{stem} ({n}).{ext}" if ext else f"{stem} ({n})"
            if candidate not in node.children:
                return candidate
            n += 1

    def upload(self, folder: str, name: str, content: str) -> str:
        """Write without overwriting: returns the path actually used."""
        self.mkdir(folder)
        final = self.unique_name(folder, name)
        path = folder.rstrip("/") + "/" + final
        self.write(path, content)
        return path

    def duplicates(self) -> List[List[str]]:
        return sorted(sorted(paths) for paths in self.by_hash.values() if len(paths) > 1)

    # ---- helpers ------------------------------------------------------------------
    def _walk(self, parts: List[str], create: bool) -> _Dir:
        node = self.root
        for i, name in enumerate(parts):
            child = node.children.get(name)
            if child is None:
                if not create:
                    raise FileNotFoundError("/" + "/".join(parts[:i + 1]))
                self._check_room(node)
                child = node.children[name] = _Dir()
            if isinstance(child, _File):
                raise NotADirectoryError("/" + "/".join(parts[:i + 1]))
            node = child
        return node

    def _node(self, path: str) -> object:
        parts = _split(path)
        if not parts:
            return self.root
        parent = self._walk(parts[:-1], create=False)
        node = parent.children.get(parts[-1])
        if node is None:
            raise FileNotFoundError(path)
        return node

    def _check_room(self, folder: _Dir) -> None:
        if self.max_entries is not None and len(folder.children) >= self.max_entries:
            raise OSError("folder is full")

    def _files_under(self, path: str, node: object) -> List[Tuple[str, _File]]:
        if isinstance(node, _File):
            return [(path, node)]
        out: List[Tuple[str, _File]] = []
        for name, child in node.children.items():
            out += self._files_under(path.rstrip("/") + "/" + name, child)
        return out

    @staticmethod
    def _digest(f: _File) -> str:
        return hashlib.sha256(f.content.encode()).hexdigest()

    def _index(self, path: str, f: _File) -> None:
        self.by_hash.setdefault(self._digest(f), set()).add(path)

    def _unindex(self, path: str, f: _File) -> None:
        h = self._digest(f)
        paths = self.by_hash.get(h)
        if paths:
            paths.discard(path)
            if not paths:
                del self.by_hash[h]

Rehashing on unindex keeps the code short; storing each file's digest on the node avoids it and makes large files cheap to move.

Complexity

  • Path resolution: O(depth). read and write: plus O(content) for hashing and copying.
  • ls: O(k log k) for k entries (sorting). Keep children in a sorted structure if listings are frequent and folders huge.
  • delete and move of a folder: O(files under it) to keep the duplicate index right.
  • duplicates: O(number of distinct contents).

Test cases

def _expect(exc, fn):
    try:
        fn()
    except exc:
        return
    assert False, f"expected {exc.__name__}"


def test_basic_files_and_folders():
    fs = FileSystem()
    fs.write("/matters/acme/nda.txt", "secret")
    fs.write("/matters/acme/nda.txt", " terms", append=True)
    assert fs.read("/matters/acme/nda.txt") == "secret terms"
    fs.mkdir("/matters/zeta")
    assert fs.ls("/matters") == ["acme", "zeta"]
    assert fs.ls("/matters/acme/nda.txt") == ["nda.txt"]
    assert fs.ls("/") == ["matters"]


def test_errors_match_the_os():
    fs = FileSystem()
    fs.write("/a/f.txt", "x")
    _expect(FileNotFoundError, lambda: fs.read("/a/missing"))
    _expect(IsADirectoryError, lambda: fs.read("/a"))
    _expect(NotADirectoryError, lambda: fs.write("/a/f.txt/g", "y"))
    _expect(OSError, lambda: fs.delete("/a"))
    fs.delete("/a", recursive=True)
    _expect(FileNotFoundError, lambda: fs.ls("/a"))


def test_move_rename_and_into_itself():
    fs = FileSystem()
    fs.write("/a/b/c.txt", "c")
    fs.move("/a/b", "/x/y")
    assert fs.read("/x/y/c.txt") == "c" and fs.ls("/a") == []
    _expect(ValueError, lambda: fs.move("/x", "/x/y/z"))
    fs.write("/x/other.txt", "o")
    _expect(FileExistsError, lambda: fs.move("/x/other.txt", "/x/y/c.txt"))
def test_folder_limit():
    fs = FileSystem(max_entries=2)
    fs.write("/d/a", "1")
    fs.write("/d/b", "2")
    fs.write("/d/a", "overwrite is fine")
    _expect(OSError, lambda: fs.write("/d/c", "3"))


def test_unique_names_on_upload():
    fs = FileSystem()
    assert fs.upload("/inbox", "nda.txt", "v1") == "/inbox/nda.txt"
    assert fs.upload("/inbox", "nda.txt", "v2") == "/inbox/nda (1).txt"
    assert fs.upload("/inbox", "nda.txt", "v3") == "/inbox/nda (2).txt"
    assert fs.upload("/inbox", "report.final.pdf", "p") == "/inbox/report.final.pdf"
    assert fs.upload("/inbox", "report.final.pdf", "p") == "/inbox/report.final (1).pdf"
    assert fs.upload("/inbox", ".env", "e") == "/inbox/.env"
    assert fs.upload("/inbox", ".env", "e") == "/inbox/.env (1)"
    assert fs.read("/inbox/nda (1).txt") == "v2"


def test_duplicate_contents_tracked_through_changes():
    fs = FileSystem()
    fs.write("/a/1.txt", "same")
    fs.write("/b/2.txt", "same")
    fs.write("/c/3.txt", "different")
    assert fs.duplicates() == [["/a/1.txt", "/b/2.txt"]]
    fs.move("/b", "/archive/b")
    assert fs.duplicates() == [["/a/1.txt", "/archive/b/2.txt"]]
    fs.write("/c/3.txt", "same")
    assert fs.duplicates() == [["/a/1.txt", "/archive/b/2.txt", "/c/3.txt"]]
    fs.delete("/archive", recursive=True)
    fs.write("/a/1.txt", "changed")
    assert fs.duplicates() == []

Follow-ups

  1. Large files. Store content in chunks (say 4 MB) addressed by hash. Identical chunks are stored once across files (deduplication), appends only add chunks, and the file's hash becomes a hash of its chunk hashes.
  2. Concurrency. A single lock is simplest. For more parallelism, lock per folder and acquire locks in path order for move to avoid deadlocks.
  3. Permissions. Folder-level access for each client matter: store an ACL on folders and check it during path resolution, inheriting from parents unless overridden.
  4. Versions. Keep previous contents per file (a list of content hashes with timestamps) so a document can be restored; with content addressing, versions cost only the chunks that changed.
  5. Persistence. Write an append-only log of operations and replay it on start, with periodic snapshots of the tree.

Interviewer expectations

Mid-level (L4)

  • A clean tree of folders and files with path resolution, and correct mkdir -p, write, append, read and sorted ls.
  • Sensible errors for missing paths and files used as folders.
  • Tests for nested paths.

Senior (L5)

  • One resolution helper that centralizes errors; delete with the non-empty check; move that refuses moving a folder into itself.
  • Folder limits, unique-name generation with extensions and dotfiles handled, and an incrementally maintained duplicate index.
  • Keeps the index correct through overwrite, move and delete, with tests for each.

Staff (L6+)

  • Chunked, content-addressed storage for large files, deduplication and versions.
  • Concurrency with per-folder locks and ordered acquisition, folder ACLs for client matters, persistence via a log and snapshots.
  • Chooses data structures from the expected sizes (huge folders, deep trees, frequent listings).

Sign in to comment, vote or remix.