Durable Key-Value Store Serialization

Implement a persistent key-value store that can serialize and deserialize data to/from a file system. The challenge is designing a custom binary encoding format — without JSON or pickle — that correctly round-trips keys and values containing arbitrary characters (newlines, emojis, null bytes, commas). A follow-up extends the problem to a chunked multi-file layout when each file is capped at 1 KB. The Challenge You need to build a permanent key-value store. This store must be able to save data to a file system and read it back later. You will get a "mock" (fake) file system and some tools to turn numbers and strings into bytes. The main difficulty is inventing your own format to save dictionaries. You cannot use easy tools like JSON or pickle. Your format must handle strings that contain strange characters. Important Context: You get a fake file system, not a real one. You get helper functions to change integers and strings into bytes. No JSON or pickle allowed. Keys and values can have any characters (like new lines, emojis, or commas). You must be able to load the data back exactly as it was saved. Part 1: Basic Store Implementation Tools You Are Given The interviewer gives you this mock code interface: class FileSystem: def save_blob(self, data: bytes) -> None: """Save bytes to the file system""" pass def get_blob(self) -> bytes: """Get bytes back from the file system""" pass # Helper functions provided to you def serialize_int(value: int) -> bytes: """Turn an integer into bytes""" pass def deserialize_int(data: bytes) -> int: """Turn bytes back into an integer""" pass def serialize_str(value: str) -> bytes: """Turn a string into bytes""" pass def deserialize_str(data: bytes) -> str: """Turn bytes back into a string""" pass What You Need to Write You need to write a KVStore class with these methods: class KVStore: def __init__(self, file_system: FileSystem): """Start with a file system instance""" pass def put(self, key: str, value: str) -> None: """Save a key-value pair in memory""" pass def get(self, key: str) -> str: """Find a value using a key""" pass def shutdown(self) -> None: """Turn the whole store into bytes and save it""" pass def restore(self) -> None: """Load bytes from the file and rebuild the store""" pass How It Should Work fs = FileSystem() kv_store = KVStore(fs) # Save some data kv_store.put("name", "John:Doe") kv_store.put("city", "New,York") kv_store.put("key\nwith\nnewlines", "value=with=equals") # Save to file system kv_store.shutdown() # Make a new instance and load data back new_kv_store = KVStore(fs) new_kv_store.restore() # The data should be exactly the same assert new_kv_store.get("name") == "John:Doe" assert new_kv_store.get("city") == "New,York" assert new_kv_store.get("key\nwith\nnewlines") == "value=with=equals" Hard Parts to Watch Out For Delimiter Conflicts: If you separate data with symbols like : or ,, it breaks if your data also contains those symbols. Example: If you save "key:value", but the key is "time:now", the computer gets confused. Escaping is Hard: Using backslashes (like \:) is messy to code and easy to break. Any Content Allowed: Keys and values might have: Symbols (:, ,, =) New lines (\n) Quotes Null bytes The Best Solution: Length-Prefixed Encoding The safest way to solve this is length-prefixed encoding. This means you write the length of the data before the data itself. The Concept: <length>:<data> Examples: String "a:b" (3 letters) → 3:a:b String "hello" (5 letters) → 5:hello For Key-Value Pairs: <keyLen>:<key><valueLen>:<value> Example: {"ab": "xyz"} → 2:ab3:xyz {"key:1": "val=ue"} → 5:key:16:val=ue Note: The examples above use text (like "5:") to explain the idea. In your real code, serialize_int() will likely create binary bytes (like 4 bytes of raw data). It won't look like human-readable text numbers. Why this is good: You know exactly how many bytes to read next. You don't need to scan for special characters or use escape codes. Real systems like Redis use this logic. Edge Cases to Test Empty dictionary Empty strings for keys or values Keys/values with symbols (:, ,, =) Keys/values with new lines Unicode characters (emojis, foreign languages) Very long strings Saving and loading multiple times Part 2: Handling File Size Limits Follow-up Question: What if each file can only hold 1KB (1024 bytes)? How do you change your code to save data that is bigger than that? New Rules Each file has a max size of 1KB. The user of your KVStore should not notice any difference. You must save and load everything correctly. You must handle data that needs many files. How to Solve It Metadata File: Create a special "metadata" file. This file simply counts how many chunks you have. Chunking: Chop your big data into small, fixed-size pieces (chunks). Reassembly: To load, read the metadata first. Then read all the chunks in order. Naming: Name the files clearly (e.g., chunk_0, chunk_1). Things to Discuss Where to cut? Cutting by byte count is easiest. Don't worry about cutting in the middle of a string; you will glue it back together before reading it. Metadata: Store the total number of chunks. Write the metadata file last (or first) to ensure safety. File Names: Use a simple pattern like chunk_0, chunk_1. Keep a separate name for metadata. New Tools The FileSystem interface changes to support multiple files: from typing import List class FileSystem: def save_blob(self, filename: str, data: bytes) -> None: """Save bytes to a specific filename""" pass def get_blob(self, filename: str) -> bytes: """Get bytes from a specific filename""" pass def list_files(self) -> List[str]: """See all files (helpful for debugging)""" pass Step-by-Step Plan To Save (Serialize): Turn the whole KV store into one big byte object (using the Part 1 method). Do the math: total_chunks = ceil(total_bytes / 1024). Write the total_chunks number to a metadata file. Slice the big byte object into 1KB pieces. Save each piece as a separate file. To Load (Deserialize): Read the metadata file to find out how many chunks exist. Read every chunk file in order (chunk_0, chunk_1, etc.). Glue all the chunks back together into one big byte object. Turn the big byte object back into a dictionary. Testing Part 2 Data smaller than 1KB (1 chunk) Data exactly 1KB Data slightly larger than 1KB (2 chunks) Huge data (many chunks) Empty store Check that chunks are put back together in the right order Coding Tips Tips for Part 1 Memory: Just use a Python dictionary {} to hold data while the program is running. Saving Logic: Loop through the dictionary. Turn the key into bytes. Get its length. Turn the value into bytes. Get its length. Combine them: length + key + length + value. Loading Logic: Use a pointer variable (like pos) to track where you are reading. Read the length (e.g., 4 bytes). Read that many bytes for the data. Move the pointer forward. Repeat. Helper Function: Write a function called read_length_and_data. It should read the size, grab the data, and return the new position. Tips for Part 2 Constants: Set CHUNK_SIZE = 1024. Math: Calculate chunks using (len(data) + CHUNK_SIZE - 1) // CHUNK_SIZE. This handles remainders correctly. Bytes: Keep everything as bytes. Only turn them into strings at the very end. Mistakes to Avoid Using JSON/Pickle: The interviewer wants custom code. Using Delimiters: Do not just put commas between items. It will break. Strings vs Bytes: Be careful. Make sure you encode strings to bytes before saving. Empty Strings: Handle cases where a key or value is empty (length is 0). Solution Code - Part 1 Here is a working example for Part 1: class KVStore: def __init__(self, file_system): self.fs = file_system self.store = {} def put(self, key: str, value: str) -> None: """Store a key-value pair""" self.store[key] = value def get(self, key: str) -> str: """Find value by key""" return self.store.get(key) def shutdown(self) -> None: """Turn dictionary to bytes and save to file""" serialized_bytes = self._serialize() self.fs.save_blob(serialized_bytes) def restore(self) -> None: """Load bytes and rebuild dictionary""" data_bytes = self.fs.get_blob() if data_bytes: self.store = self._deserialize(data_bytes) def _serialize(self) -> bytes: """ Convert dictionary to bytes. Format: <keyLen>:<key><valueLen>:<value> """ if not self.store: return b"" result = [] for key, value in self.store.items(): # Process key key_bytes = serialize_str(key) key_len_bytes = serialize_int(len(key_bytes)) # Process value value_bytes = serialize_str(value) value_len_bytes = serialize_int(len(value_bytes)) # Combine: len + key + len + value result.append(key_len_bytes) result.append(key_bytes) result.append(value_len_bytes) result.append(value_bytes) return b"".join(result) def _deserialize(self, data: bytes) -> dict: """ Turn bytes back into dictionary using the length format. """ store = {} pos = 0 while pos < len(data): # Read key size and key data key, pos = self._read_length_and_data(data, pos) # Read value size and value data value, pos = self._read_length_and_data(data, pos) store[key] = value return store def _read_length_and_data(self, data: bytes, pos: int) -> tuple: """ Helper to read one piece of data. Returns (string_data, new_position_pointer) """ # Read the length (assuming 4 bytes for an integer) length_bytes = data[pos:pos+4] length = deserialize_int(length_bytes) pos += 4 # Read the actual string data data_bytes = data[pos:pos+length] data_str = deserialize_str(data_bytes) pos += length return data_str, pos # Example usage if __name__ == "__main__": fs = FileSystem() kv = KVStore(fs) # Test tricky strings kv.put("key:with:colons", "value,with,commas") kv.put("key\nwith\nnewlines", "value=with=equals") kv.put("", "empty key") kv.put("empty value", "") # Save kv.shutdown() # Restore in a new object kv2 = KVStore(fs) kv2.restore() # Check results assert kv2.get("key:with:colons") == "value,with,commas" assert kv2.get("key\nwith\nnewlines") == "value=with=equals" assert kv2.get("") == "empty key" assert kv2.get("empty value") == "" print("All tests passed!") Solution Code - Part 2 Here is the solution using chunks (splitting files): class KVStoreChunked: CHUNK_SIZE = 1024 # Max 1KB per file METADATA_FILE = "_metadata" CHUNK_PREFIX = "chunk_" def __init__(self, file_system): self.fs = file_system self.store = {} def put(self, key: str, value: str) -> None: self.store[key] = value def get(self, key: str) -> str: return self.store.get(key) def shutdown(self) -> None: """Save data by splitting it into smaller files""" serialized_bytes = self._serialize() if not serialized_bytes: # Handle empty store metadata = serialize_int(0) # 0 chunks self.fs.save_blob(self.METADATA_FILE, metadata) return # Calculate how many chunks we need total_chunks = (len(serialized_bytes) + self.CHUNK_SIZE - 1) // self.CHUNK_SIZE # Save the count to the metadata file metadata = serialize_int(total_chunks) self.fs.save_blob(self.METADATA_FILE, metadata) # Save each chunk for i in range(total_chunks): start_idx = i * self.CHUNK_SIZE end_idx = min(start_idx + self.CHUNK_SIZE, len(serialized_bytes)) chunk_data = serialized_bytes[start_idx:end_idx] chunk_filename = f"{self.CHUNK_PREFIX}{i}" self.fs.save_blob(chunk_filename, chunk_data) def restore(self) -> None: """Load data by reading all chunks""" # Read metadata to find out how many files to load metadata_bytes = self.fs.get_blob(self.METADATA_FILE) total_chunks = deserialize_int(metadata_bytes) if total_chunks == 0: self.store = {} return # Read and glue all chunks together all_data = b"" for i in range(total_chunks): chunk_filename = f"{self.CHUNK_PREFIX}{i}" chunk_data = self.fs.get_blob(chunk_filename) all_data += chunk_data # Turn the full data back into a dictionary self.store = self._deserialize(all_data) def _serialize(self) -> bytes: """Same as Part 1""" if not self.store: return b"" result = [] for key, value in self.store.items(): key_bytes = serialize_str(key) key_len_bytes = serialize_int(len(key_bytes)) value_bytes = serialize_str(value) value_len_bytes = serialize_int(len(value_bytes)) result.append(key_len_bytes) result.append(key_bytes) result.append(value_len_bytes) result.append(value_bytes) return b"".join(result) def _deserialize(self, data: bytes) -> dict: """Same as Part 1""" store = {} pos = 0 while pos < len(data): key, pos = self._read_length_and_data(data, pos) value, pos = self._read_length_and_data(data, pos) store[key] = value return store def _read_length_and_data(self, data: bytes, pos: int) -> tuple: """Same as Part 1""" length_bytes = data[pos:pos+4] length = deserialize_int(length_bytes) pos += 4 data_bytes = data[pos:pos+length] data_str = deserialize_str(data_bytes) pos += length return data_str, pos # Example usage if __name__ == "__main__": fs = FileSystem() kv = KVStoreChunked(fs) # Make data big enough to need multiple chunks # Each item is about 182 bytes. 15 items ≈ 2730 bytes. # This will require 3 chunks (since max is 1024). for i in range(15): kv.put(f"key_{i}_" + "x" * 80, f"value_{i}_" + "y" * 80) # Save kv.shutdown() # Restore in new instance kv2 = KVStoreChunked(fs) kv2.restore() # Check if data is correct for i in range(15): expected_key = f"key_{i}_" + "x" * 80 expected_value = f"value_{i}_" + "y" * 80 assert kv2.get(expected_key) == expected_value print("Chunked storage tests passed!") Main Lessons Length-Prefixing: This is the best way to handle mixed strings without breaking your file format. Understand the API: Read the fake file system code carefully. Bytes vs Strings: Be careful when converting. You cannot write strings directly to the file system; they must be bytes. Chunking: Using a metadata file plus numbered chunk files is a standard way to solve size limits. Time Management: Don't spend too long on Part 1. You need time for the follow-up question. How to Pass the Interview Ask First: "Can I use JSON?" (The answer is usually no, but it shows you know standard tools). Clarify: Ask about the fake file system if you don't understand it. Helper Functions: Write small functions like read_length_and_data. It makes your code cleaner and easier for the interviewer to read. Test Edge Cases: Mention empty strings and weird symbols. Push for Part 2: A strong candidate needs to finish the chunking problem. Notes Append-only log recovery variant A current phone-screen variant phrases durability as rebuilding an in-memory key-value store from a log file after a system disconnection, rather than the fixed snapshot-and-chunking follow-up above. The implementation consumed the available coding time, leaving no time for explicit test cases or a follow-up. Budget tests early when this framing appears.