SWE / RE · Coding

KV Save/Restore:UTF-8长度前缀与限长文件分片

编码键值后任意字节分块,用manifest完整验证再恢复。

题目重点:编码键值后任意字节分块,用manifest完整验证再恢复。

题意与假设

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

  • 训练KV键值均string,UTF-8编码,uint32大端记录数/字节长度;不支持4GiB单字段。
  • save返回manifest/chunks模拟已有文件接口;所有chunks顺序稳定,restore先验证再产生新map,不原地半覆盖。

解题路线

  1. 序列化recordCount,然后每key/value写byteLength+rawBytes。分隔符、 、NUL、emoji都可包含,len应按UTF-8字节而非字符计。
  2. 整个blob按limit切分,header或UTF-8字符可跨chunk,因为restore按manifest顺序重连才decode。若要求不能跨文件记录,遇单记录>limit就需要新的fragment合同,本模型不混用。
  3. manifest含size/count/SHA256,恢复校验丢片/换序/截断,再检查每length、UTF-8、重复key与尾随字节。生产持久化须先写chunks/fsync,再原子发布manifest,旧版本保留到新版本可验证。

手动走一遍

键a|b和值中含换行都不影响边界;中文字符长度应由encode结果算。limit1仍可恢复但metadata/文件数巨大;丢一片即拒绝,不恢复部分map。

Python 参考实现

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

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

import struct,hashlib

def encode(values):
    parts=[struct.pack('>I',len(values))]
    for key,value in values.items():
        if type(key) is not str or type(value) is not str:raise ValueError("string KV")
        for string in (key,value):
            data=string.encode('utf-8');parts.extend((struct.pack('>I',len(data)),data))
    return b''.join(parts)

def decode(blob):
    position=0
    def number():
        nonlocal position
        if position+4>len(blob):raise ValueError("truncated length")
        result=struct.unpack_from('>I',blob,position)[0];position+=4;return result
    def string():
        nonlocal position
        length=number()
        if position+length>len(blob):raise ValueError("truncated text")
        result=blob[position:position+length].decode('utf-8');position+=length;return result
    count=number();result={}
    for _ in range(count):
        key,value=string(),string()
        if key in result:raise ValueError("duplicate key")
        result[key]=value
    if position!=len(blob):raise ValueError("trailing bytes")
    return result

def save(values,limit):
    if limit<=0:raise ValueError("positive chunk limit")
    blob=encode(values);chunks=[blob[i:i+limit] for i in range(0,len(blob),limit)]
    return {'size':len(blob),'count':len(chunks),'sha256':hashlib.sha256(blob).hexdigest()},chunks

def restore(manifest,chunks):
    blob=b''.join(chunks)
    if manifest['count']!=len(chunks) or manifest['size']!=len(blob) or manifest['sha256']!=hashlib.sha256(blob).hexdigest():raise ValueError("snapshot integrity")
    return decode(blob)

复杂度

总编码B,save/restore O(B+recordCount)时间/O(B)内存含blob/chunks;示例不是流式O(1)恢复。实际文件数量ceil(B/limit),打开/metadata开销不可忽略。

易错点与追问

容易失分的边界

  • 用str(map)或JSON却不声明字节格式,会与给定integer codec不兼容。
  • hash验证只检测完整性,不是对不可信文件的授权机制。

继续深挖

  • 流式跨chunk decoder怎样保留不完整length/text状态?
  • KV并发更新时如何取得一致snapshot并原子切换restore?
✓已复制微信号!