In-Memory Database ## The Challenge You need to build an in-memory key-field-value database. Think of this like a nested HashMap. Each "key" holds a collection of "field-value" pairs. You will b

In-Memory Database ## The Challenge You need to build an in-memory key-field-value database. Think of this like a nested HashMap. Each "key" holds a collection of "field-value" pairs. You will build this in four levels. Each level adds new features to the previous one.

Important Rule: Every operation gets a timestamp (a positive integer). This number always gets bigger. No two operations ever have the same timestamp. ## Level 1: Basic Operations ### Requirements Create a class called InMemoryDB. It needs to handle these basic tasks: 1.

Set: Save a value for a key and field. 2.

Get: Read a value. 3.

Delete: Remove a field. 4.

Compare and Set: Update a value only if it currently matches a specific value.

python class InMemoryDB: def __init__(self): """Start the database.""" pass def set(self, timestamp: int, key: str, field: str, value: str) -> None: """ Save a value. If the field exists, overwrite it. """ pass def get(self, timestamp: int, key: str, field: str) -> str: """ Read the value.

**Returns** "" if the key or field is missing. """ pass def delete(self, timestamp: int, key: str, field: str) -> bool: """ Remove a field.

**Returns** True if deleted, False if it wasn't there. """ pass def compare_and_set(self, timestamp: int, key: str, field: str, expected_value: str, new_value: str) -> bool: """ Update the value only if the current value is equal to expected_value.

**Returns** True if successful, False otherwise. """ pass

Example Usage

python db = InMemoryDB() db.set(1, "user1", "name", "Alice") db.set(2, "user1", "age", "30") db.get(3, "user1", "name") # "Alice" db.get(4, "user1", "email") # "" (field doesn't exist) db.get(5, "user2", "name") # "" (key doesn't exist) db.delete(6, "user1", "age") # True db.delete(7, "user1", "age") # False (already deleted) db.compare_and_set(8, "user1", "name", "Alice", "Bob") # True db.get(9, "user1", "name") # "Bob" db.compare_and_set(10, "user1", "name", "Alice", "Eve") # False (current value is "Bob") db.compare_and_set(11, "user1", "zip", "000", "111") # False (field doesn't exist)

• • • Level 1 Solution We use a dictionary of dictionaries. The outer dictionary holds the key, and the inner dictionary holds the field and value.

python class InMemoryDB: def __init__(self): self.data = {} # key -> {field -> value} def set(self, timestamp: int, key: str, field: str, value: str) -> None: if key not in self.data: self.data[key] = {} self.data[key][field] = value def get(self, timestamp: int, key: str, field: str) -> str: if key not in self.data or field not in self.data[key]:

**return** ""

**return** self.data[key][field] def delete(self, timestamp: int, key: str, field: str) -> bool: if key not in self.data or field not in self.data[key]:

**return** False del self.data[key][field] if not self.data[key]: del self.data[key]

**return** True def compare_and_set(self, timestamp: int, key: str, field: str, expected_value: str, new_value: str) -> bool: if key not in self.data or field not in self.data[key]:

**return** False if self.data[key][field] != expected_value:

**return** False self.data[key][field] = new_value return True

Complexity Analysis: | Method | Time | Space | | --- | --- | --- | | set | O(1) | O(1) per field | | get | O(1) | O(1) | | delete | O(1) | O(1) | | compare_and_set | O(1) | O(1) | ## Level 2: Search and Filter ### Requirements Now you need to list all fields for a key. You also need to filter these fields using a prefix (the start of the word). The results must be sorted alphabetically by the field name.

python def scan(self, timestamp: int, key: str) -> list: """

**Return** all field-value pairs for a key. Format: "field(value)" Order: Alphabetical by field name. """ pass def scan_with_prefix(self, timestamp: int, key: str, prefix: str) -> list: """

**Return** field-value pairs where the field starts with the prefix. Format: "field(value)" Order: Alphabetical by field name. """ pass

Example Usage

python db = InMemoryDB() db.set(1, "user1", "name", "Alice") db.set(2, "user1", "age", "30") db.set(3, "user1", "nickname", "Ali") db.scan(4, "user1") # ["age(30)", "name(Alice)", "nickname(Ali)"] db.scan_with_prefix(5, "user1", "na") # ["name(Alice)"] db.scan_with_prefix(6, "user1", "n") # ["name(Alice)", "nickname(Ali)"] db.scan_with_prefix(7, "user1", "x") # [] db.scan(8, "user999") # []

• • • Level 2 Solution We grab the fields for the key, sort them, and format them as strings. For the prefix search, we only include fields that start with the specific letters.

python class InMemoryDB: def __init__(self): self.data = {} # key -> {field -> value} def set(self, timestamp: int, key: str, field: str, value: str) -> None: if key not in self.data: self.data[key] = {} self.data[key][field] = value def get(self, timestamp: int, key: str, field: str) -> str: if key not in self.data or field not in self.data[key]:

**return** ""

**return** self.data[key][field] def delete(self, timestamp: int, key: str, field: str) -> bool: if key not in self.data or field not in self.data[key]:

**return** False del self.data[key][field] if not self.data[key]: del self.data[key]

**return** True def compare_and_set(self, timestamp: int, key: str, field: str, expected_value: str, new_value: str) -> bool: if key not in self.data or field not in self.data[key]:

**return** False if self.data[key][field] != expected_value:

**return** False self.data[key][field] = new_value return True def scan(self, timestamp: int, key: str) -> list: if key not in self.data:

**return** []

**return** [f"{field}({value})" for field, value in sorted(self.data[key].items())] def scan_with_prefix(self, timestamp: int, key: str, prefix: str) -> list: if key not in self.data:

**return** []

**return** [f"{field}({value})" for field, value in sorted(self.data[key].items()) if field.startswith(prefix)]

Complexity Analysis: | Method | Time | Space | | --- | --- | --- | | scan | O(F log F) | O(F) | | scan_with_prefix | O(F log F) | O(F) | Here, F is the number of fields. The sorting step causes the O(F log F) time complexity. This is usually okay because read/write operations (O(1)) happen more often than scanning. ## Level 3: Expiration Times (TTL) ### Requirements Now, records can expire. You will add a "Time-To-Live" (TTL). *

Expiration: A record expires when current_time >= timestamp + ttl. *

Behavior: If a record is expired, the database should act like it does not exist. *

Clock: The timestamp

passed to the function is the current time.

python def set_with_ttl(self, timestamp: int, key: str, field: str, value: str, ttl: int) -> None: """ Save a value that expires at time = timestamp + ttl. It overwrites any existing value and TTL. """ pass def compare_and_set_with_ttl(self, timestamp: int, key: str, field: str, expected_value: str, new_value: str, ttl: int) -> bool: """ Update a value and set a new TTL only if the current value matches. """ pass

Example Usage

python db = InMemoryDB() db.set(1, "user1", "name", "Alice") # No TTL — never expires db.set_with_ttl(2, "user1", "session", "abc", 10) # Expires at time 12 db.get(3, "user1", "session") # "abc" db.get(11, "user1", "session") # "abc" (11 < 12, still good) db.get(12, "user1", "session") # "" (12 >= 12, expired) db.get(13, "user1", "name") # "Alice" (no TTL, still valid) db.set_with_ttl(14, "user2", "token", "xyz", 5) # Expires at time 19 db.scan(15, "user2") # ["token(xyz)"] db.scan(19, "user2") # [] (expired) db.set_with_ttl(20, "user3", "code", "123", 10) # Expires at time 30 db.compare_and_set_with_ttl(21, "user3", "code", "123", "456", 5) # True, new expiry at 26 db.get(25, "user3", "code") # "456" db.get(26, "user3", "code") # "" (expired)

• • • Level 3 Solution We will use two dictionaries: 1. self.data: Stores the actual values. 2. self.expiry: Stores the expiration time for each field. We use

Lazy Expiration. We do not delete old data immediately. Instead, whenever someone asks for a record (via get or scan), we check if it is expired. If it is, we delete it then.

python class InMemoryDB: def __init__(self): self.data = {} # key -> {field -> value} self.expiry = {} # key -> {field -> expiry_time} def _is_expired(self, key: str, field: str, timestamp: int) -> bool: """Check if time is up for this field.""" if key in self.expiry and field in self.expiry[key]:

**return** timestamp >= self.expiry[key][field]

**return** False def _clean_field(self, key: str, field: str): """Remove a field and its expiry time.""" if key in self.data and field in self.data[key]: del self.data[key][field] if not self.data[key]: del self.data[key] if key in self.expiry and field in self.expiry[key]: del self.expiry[key][field] if not self.expiry[key]: del self.expiry[key] def set(self, timestamp: int, key: str, field: str, value: str) -> None: if key not in self.data: self.data[key] = {} self.data[key][field] = value # Clear any existing TTL because set() makes it permanent if key in self.expiry and field in self.expiry[key]: del self.expiry[key][field] def set_with_ttl(self, timestamp: int, key: str, field: str, value: str, ttl: int) -> None: if key not in self.data: self.data[key] = {} self.data[key][field] = value if key not in self.expiry: self.expiry[key] = {} self.expiry[key][field] = timestamp + ttl def get(self, timestamp: int, key: str, field: str) -> str: if key not in self.data or field not in self.data[key]:

**return** "" if self._is_expired(key, field, timestamp): self._clean_field(key, field)

**return** ""

**return** self.data[key][field] def delete(self, timestamp: int, key: str, field: str) -> bool: if key not in self.data or field not in self.data[key]:

**return** False if self._is_expired(key, field, timestamp): self._clean_field(key, field)

**return** False self._clean_field(key, field)

**return** True def compare_and_set(self, timestamp: int, key: str, field: str, expected_value: str, new_value: str) -> bool: if key not in self.data or field not in self.data[key]:

**return** False if self._is_expired(key, field, timestamp): self._clean_field(key, field)

**return** False if self.data[key][field] != expected_value:

**return** False self.data[key][field] = new_value return True def compare_and_set_with_ttl(self, timestamp: int, key: str, field: str, expected_value: str, new_value: str, ttl: int) -> bool: if key not in self.data or field not in self.data[key]:

**return** False if self._is_expired(key, field, timestamp): self._clean_field(key, field)

**return** False if self.data[key][field] != expected_value:

**return** False self.data[key][field] = new_value if key not in self.expiry: self.expiry[key] = {} self.expiry[key][field] = timestamp + ttl return True def scan(self, timestamp: int, key: str) -> list: if key not in self.data:

**return** [] result = [] for field, value in sorted(self.data[key].items()): if not self._is_expired(key, field, timestamp): result.append(f"{field}({value})")

**return** result def scan_with_prefix(self, timestamp: int, key: str, prefix: str) -> list: if key not in self.data:

**return** [] result = [] for field, value in sorted(self.data[key].items()): if field.startswith(prefix) and not self._is_expired(key, field, timestamp): result.append(f"{field}({value})")

**return** result

Complexity Analysis: | Method | Time | Space | | --- | --- | --- | | set_with_ttl | O(1) | O(1) | | get | O(1) | O(1) | | scan | O(F log F) | O(F) | This approach is efficient because we only do work when needed (lazy). We don't need a background process constantly checking for old data. ## Level 4: Save and Restore ### Requirements You need to save the database state at a specific time and restore it later. 1.

Backup: Save everything (values and TTLs). Do not save records that are already expired. 2.

Restore: Reset the database to match a previous backup. * If you ask to restore to a time that doesn't have an exact backup, use the latest backup before that time. * After restoring, the TTLs should work exactly as they did in the backup.

python def backup(self, timestamp: int) -> None: """ Save the current state of the database. Do not include expired records. """ pass def restore(self, timestamp: int, backup_timestamp: int) -> None: """ Restore the database from a backup. Find the latest backup where time <= backup_timestamp. """ pass

Example Usage

python db = InMemoryDB() db.set(1, "app", "version", "1.0") db.set_with_ttl(2, "app", "cache", "data1", 20) # Expires at time 22 db.backup(3) db.set(4, "app", "version", "2.0") db.delete(6, "app", "cache") db.restore(9, 3) # Go back to time 3 db.get(10, "app", "version") # "1.0" (restored) db.get(12, "app", "cache") # "data1" (restored, TTL still active) db.get(22, "app", "cache") # "" (expired normally)

• • • Level 4 Solution We store backups in a list and a dictionary. * self.backups: Maps a timestamp to a snapshot of the data. * self.backup_timestamps: A sorted list of backup times. This helps us find the closest backup quickly using binary search. We use copy.deepcopy to make sure the backup is independent of the live database.

python import copy import bisect class InMemoryDB: def __init__(self): self.data = {} # key -> {field -> value} self.expiry = {} # key -> {field -> expiry_time} self.backup_timestamps = [] # sorted list of backup times self.backups = {} # timestamp -> (data_snapshot, expiry_snapshot) def _is_expired(self, key: str, field: str, timestamp: int) -> bool: if key in self.expiry and field in self.expiry[key]:

**return** timestamp >= self.expiry[key][field]

**return** False def _clean_field(self, key: str, field: str): if key in self.data and field in self.data[key]: del self.data[key][field] if not self.data[key]: del self.data[key] if key in self.expiry and field in self.expiry[key]: del self.expiry[key][field] if not self.expiry[key]: del self.expiry[key] def set(self, timestamp: int, key: str, field: str, value: str) -> None: if key not in self.data: self.data[key] = {} self.data[key][field] = value if key in self.expiry and field in self.expiry[key]: del self.expiry[key][field] def set_with_ttl(self, timestamp: int, key: str, field: str, value: str, ttl: int) -> None: if key not in self.data: self.data[key] = {} self.data[key][field] = value if key not in self.expiry: self.expiry[key] = {} self.expiry[key][field] = timestamp + ttl def get(self, timestamp: int, key: str, field: str) -> str: if key not in self.data or field not in self.data[key]:

**return** "" if self._is_expired(key, field, timestamp): self._clean_field(key, field)

**return** ""

**return** self.data[key][field] def delete(self, timestamp: int, key: str, field: str) -> bool: if key not in self.data or field not in self.data[key]:

**return** False if self._is_expired(key, field, timestamp): self._clean_field(key, field)

**return** False self._clean_field(key, field)

**return** True def compare_and_set(self, timestamp: int, key: str, field: str, expected_value: str, new_value: str) -> bool: if key not in self.data or field not in self.data[key]:

**return** False if self._is_expired(key, field, timestamp): self._clean_field(key, field)

**return** False if self.data[key][field] != expected_value:

**return** False self.data[key][field] = new_value return True def compare_and_set_with_ttl(self, timestamp: int, key: str, field: str, expected_value: str, new_value: str, ttl: int) -> bool: if key not in self.data or field not in self.data[key]:

**return** False if self._is_expired(key, field, timestamp): self._clean_field(key, field)

**return** False if self.data[key][field] != expected_value:

**return** False self.data[key][field] = new_value if key not in self.expiry: self.expiry[key] = {} self.expiry[key][field] = timestamp + ttl return True def scan(self, timestamp: int, key: str) -> list: if key not in self.data:

**return** [] result = [] for field, value in sorted(self.data[key].items()): if not self._is_expired(key, field, timestamp): result.append(f"{field}({value})")

**return** result def scan_with_prefix(self, timestamp: int, key: str, prefix: str) -> list: if key not in self.data:

**return** [] result = [] for field, value in sorted(self.data[key].items()): if field.startswith(prefix) and not self._is_expired(key, field, timestamp): result.append(f"{field}({value})")

**return** result def backup(self, timestamp: int) -> None: # Create a clean snapshot without expired items data_snapshot = {} expiry_snapshot = {} for key, fields in self.data.items(): clean_fields = {} clean_expiry = {} for field, value in fields.items(): if not self._is_expired(key, field, timestamp): clean_fields[field] = value if key in self.expiry and field in self.expiry[key]: clean_expiry[field] = self.expiry[key][field] if clean_fields: data_snapshot[key] = clean_fields if clean_expiry: expiry_snapshot[key] = clean_expiry self.backups[timestamp] = (data_snapshot, expiry_snapshot) bisect.insort(self.backup_timestamps, timestamp) def restore(self, timestamp: int, backup_timestamp: int) -> None: # Find the correct backup timestamp idx = bisect.bisect_right(self.backup_timestamps, backup_timestamp) - 1 if idx < 0:

**return** # No valid backup found actual_ts = self.backup_timestamps[idx] data_snapshot, expiry_snapshot = self.backups[actual_ts] # Use deepcopy so we don't mess up the backup if we edit data later self.data = copy.deepcopy(data_snapshot) self.expiry = copy.deepcopy(expiry_snapshot)

Complexity Analysis: | Method | Time | Space | | --- | --- | --- | | backup | O(N) | O(N) | | restore | O(N + log B) | O(N) | Where N is the amount of data and B is the number of backups. restore uses binary search (log B) to find the backup, and then copies the data (N). ## Common Interview Questions 1.

Why use a HashMap instead of a Tree? *

HashMap: Very fast for getting or setting single items (O(1)). *

Tree/SortedMap: Better if you need to scan fields often (O(F)). But since we set/get more often than we scan, the HashMap is the better choice here. 2.

How else can we handle Expiration (TTL)? *

Lazy (used here): Check only when asked. Simple and uses no extra CPU when idle. *

Eager: Have a background process that constantly deletes old keys. This saves memory but is harder to code and uses CPU even when no one is using the database. 3.

How can we improve Backups? *

Full Snapshot (used here): Easy to implement, but uses a lot of memory because we copy everything. *

Incremental: Only save the changes since the last backup. This saves space but makes restoring slower (you have to replay all the changes). 4.

What if multiple people use the DB at the same time? * compare_and_set helps prevent overwriting changes accidentally. * Real databases use "locks" to ensure two people don't write to the same spot at the exact same instant. ## Complexity Summary | Method | Time | Space | | --- | --- | --- | | set | O(1) | O(1) | | get | O(1) | O(1) | | scan | O(F log F) | O(F) | | backup | O(N) | O(N) | | restore | O(N) | O(N) |