Meta 内存数据库题详解：字段版本、TTL 与历史查询

题意与假设

这道练习题保留了面经中值得训练的核心约束，并将需要澄清的条件单独列出。现场拿到变体时，先和面试官确认这些条件。

• 每个 key 下有多个 field；field 的值是字符串，None 只表示不存在或删除。对同一 field 的写入时间非递减，读取可以查询历史时间。
• TTL 有效区间为 [写入时间, 写入时间+ttl)；新版本或删除会遮住旧版本，当前版本过期后也不会恢复更早版本。
• scan 返回指定 key 下仍可见的 field，按字段名字排序；prefix 只筛选字段名。

解题路线

1. 为每个 (key, field) 保存有序写入时间和对应版本。set 写入新版本，delete 写入墓碑，TTL 存入版本的过期时刻。
2. 历史读取用二分找到查询时刻之前最后一次写入，再检查该版本是否为墓碑或已过期。不能继续向前找旧版本。
3. compare_and_set 先读取指定时刻的可见值，仅在匹配非空预期值时写新版本；scan 枚举该 key 已出现过的字段并复用历史读取。

手动走一遍

t=1 写 user.name=A（TTL=3），在 t=3 可读 A，t=4 已过期。t=5 再写 B 后，t=6 可读 B；查询 t=2 仍得到 A。若 t=7 删除，t=8 读不到 B，也不会回退到 A。

Python 参考实现

下面的代码按上文假设实现，适合本地运行并追加自己的测试用例。

代码区可横向滚动；使用键盘时，先聚焦代码区再按方向键查看。

from bisect import bisect_right
from collections import defaultdict
from dataclasses import dataclass


@dataclass(frozen=True)
class Version:
    value: str | None
    expires_at: int | None = None


class MemoryDB:
    """One key/field history; timestamps of writes to a field are nondecreasing."""

    def __init__(self):
        self.times = {}
        self.versions = {}
        self.fields_by_key = defaultdict(set)

    def _write(self, time, key, field, value, ttl=None):
        if ttl is not None and ttl <= 0:
            raise ValueError("TTL must be positive")
        identity = (key, field)
        times = self.times.setdefault(identity, [])
        if times and time < times[-1]:
            raise ValueError("writes to one field must be time-ordered")
        times.append(time)
        self.fields_by_key[key].add(field)
        self.versions.setdefault(identity, []).append(
            Version(value, time + ttl if ttl is not None else None)
        )

    def set(self, time, key, field, value, ttl=None):
        if value is None:
            raise ValueError("None is reserved for deletion")
        self._write(time, key, field, value, ttl)

    def delete(self, time, key, field):
        if self.get_at(time, key, field) is None:
            return False
        self._write(time, key, field, None)
        return True

    def compare_and_set(self, time, key, field, expected, replacement, ttl=None):
        if expected is None or replacement is None or self.get_at(time, key, field) != expected:
            return False
        self._write(time, key, field, replacement, ttl)
        return True

    def compare_and_delete(self, time, key, field, expected):
        if expected is None or self.get_at(time, key, field) != expected:
            return False
        self._write(time, key, field, None)
        return True

    def get_at(self, time, key, field):
        identity = (key, field)
        times = self.times.get(identity, ())
        index = bisect_right(times, time) - 1
        if index < 0:
            return None
        version = self.versions[identity][index]
        if version.value is None or (version.expires_at is not None and time >= version.expires_at):
            return None
        return version.value

    def scan(self, time, key, prefix=""):
        fields = sorted(field for field in self.fields_by_key.get(key, ()) if field.startswith(prefix))
        return [(field, value) for field in fields if (value := self.get_at(time, key, field)) is not None]

复杂度

设某字段有 V 个版本、某 key 有 F 个历史字段。单次 get_at 是 O(log V)，写入摊销 O(1)，scan 排序并逐字段查询为 O(F log F + Σlog V_i)；空间与总写入版本数及历史字段数成正比。

易错点与追问

容易失分的边界

• TTL 过期后向前搜索旧版本会让已覆盖的数据意外复活；只检查查询时刻最后一次写入。
• 条件更新必须比较当前可见值。示例拒绝 expected=None，避免把不存在字段当成普通字符串值。
• 代码聚焦单线程语义；真实并发服务中 compare_and_set 的读与写需要同一把锁或原子事务。

继续深挖

• 如果查询保证时间单调、只查当前值，可以单独维护当前态索引；历史查询仍保留版本序列。
• 如果历史无限增长，需要定义数据保留期和压缩规则，不能在仍允许历史查询的范围内随意删除版本。