代码实现

继续接着前文，在经过前面一大堆的设计之后，我们终于可以开始着手代码的实现了。为了保证代码的可读性，接下来会使用面向对象的方式来组织代码。

首先，在最初的实现里直接写死了把所有的数据存放在 ./data.db 这个文件里，这种做法当然是不太行的，接下来我们改成由外部传入一个路径，如果没传的话才用这个作为默认路径。同时定义好涉及到文件结构相关的一些常量，例如文件头的大小，每个页的大小，尾部暂时先不管了：

export const DEFAULT_DB_FILE = './data.db';
export const FILE_HEADER_SIZE = 100;
export const MAGIC_HEADER = Buffer.from('my simpledb format\x00');
export const MAGIC_HEADER_SIZE = MAGIC_HEADER.length; // 19
export const PAGE_SIZE = 4096;

export const DEFAULT_DB_FILE = './data.db';
export const FILE_HEADER_SIZE = 100;
export const MAGIC_HEADER = Buffer.from('my simpledb format\x00');
export const MAGIC_HEADER_SIZE = MAGIC_HEADER.length; // 19
export const PAGE_SIZE = 4096;

其中头部，FILE_HEADER_SIZE，我们直接分配 100 个字节，这 100 个字节首先放一个自定的 Magic header string，这样先读出来就能知道这个文件是不是我们的数据库创建的，如果不是的话就直接报错。实际上对于一块二进制数据来说，分配一点其中的头部空间用来标识这块数据究竟是干嘛的是个非常常见的做法，尤其在处理文件的时候。接下来再分配 2 个字节，标识出来数据库每个页的大小，默认值是 4096，当然目前先不做让用户配置页大小的功能，以后有空了再说。

接下来首先定义一个 Database 类，程序启动时，会实例化一个 Database 对象出来，把文件路径传进去，然后对文件的头部进行一下检查，没有问题的话，就可以通过它暴露出的接口来进行读写操作了:

export class Database {
  constructor(filePath: string) {}
  public set(key: string, value: string) {}
  public get(key: string) {}
  public open() {} // throw if file header check fail
}

export class Database {
  constructor(filePath: string) {}
  public set(key: string, value: string) {}
  public get(key: string) {}
  public open() {} // throw if file header check fail
}

使用的方式是这样的：

// 使用命令行用户传入的路径参数或者默认的文件路径
const dbFile = process.argv[2] || DEFAULT_DB_FILE;
const db = new Database(dbFile);

db.open();

// start repl...
// db.get(...)
// db.set(...)

// 使用命令行用户传入的路径参数或者默认的文件路径
const dbFile = process.argv[2] || DEFAULT_DB_FILE;
const db = new Database(dbFile);

db.open();

// start repl...
// db.get(...)
// db.set(...)

在读写操作的过程中，每一个页都是一个 B 树节点，从根节点开始进行查找，每次从磁盘上读取一页到内存中，然后根据指针去读取下一页，因为涉及到频繁的从磁盘上读取页这种操作，因此我们会引入一个 Pager 对象，通过它提供的接口来进行，后续还可以在它其中加入缓存控制，以减少频繁的磁盘读取，从而优化性能。

export class Pager {}

export class Pager {}

针对头部的检查，也可以放在 Pager 里，同时 Pager 提供接口让我们通过指针来通过文件读取节点，考虑到每一个节点都是固定的 4096 字节大小，因此只需要给节点一个自增 id 作为指针就可以了，这里这个自增 id 是一个从 1 开始计数的数字。

export class Pager {
  public static getPageOffsetById(id: number) {
    return (id - 1) * PAGE_SIZE + FILE_HEADER_SIZE;
  }
  public verifyFileHeader(): boolean {}
  public readPageById(id: number): Buffer {}
  public writePageById(id: number, buf: Buffer) {}
}

export class Pager {
  public static getPageOffsetById(id: number) {
    return (id - 1) * PAGE_SIZE + FILE_HEADER_SIZE;
  }
  public verifyFileHeader(): boolean {}
  public readPageById(id: number): Buffer {}
  public writePageById(id: number, buf: Buffer) {}
}

对于文件头部，为了方便，也可以引入一个简单的抽象，其中存放了 Magic header string，页的大小，根节点的 id 等等，简单的设定一个格式：

0                             19                21                  25                    29            100
+------------------------------+-----------------+-------------------+--------------------+--------------+
| [buffer] magic_header_string | [int] page_size | [int] max_page_id | [int] root_page_id | unused_space |
+------------------------------+-----------------+-------------------+--------------------+--------------+

0                             19                21                  25                    29            100
+------------------------------+-----------------+-------------------+--------------------+--------------+
| [buffer] magic_header_string | [int] page_size | [int] max_page_id | [int] root_page_id | unused_space |
+------------------------------+-----------------+-------------------+--------------------+--------------+

每一次申请访问最大节点的 id 时，都需要进行一次自增的操作，以避免出现重复的情况：

export class FileHeader {
  public readonly buffer: Buffer; // 100 bytes
  public get maxPageId(): number {}

  public get rootPageId(): number {}
  public set rootPageId(n: number) {}

  public verify(): boolean {}
}

export class Pager {
  // ...
  private header: FileHeader | null = null;
}

export class FileHeader {
  public readonly buffer: Buffer; // 100 bytes
  public get maxPageId(): number {}

  public get rootPageId(): number {}
  public set rootPageId(n: number) {}

  public verify(): boolean {}
}

export class Pager {
  // ...
  private header: FileHeader | null = null;
}

TIP

我们最好以从磁盘中读取出来的 buffer 作为 single source of truth，将其加载进内存并初始化为我们抽象出来的各种对象时还是保留这个 buffer 作为内部的私有成员，之后不管是写入数据还是读取数据时，也都是直接通过这个 buffer。这点对于后面要实现的其他各种抽象也是一样的，避免多个数据源容易引起的不一致问题和管理的复杂度。这也是上面为什么会大量使用 getter/setter 模式来访问数据的原因。

而在每一个 B 树节点内部，像前一篇文章提到的，他们是由很多个 cell 组成的，因此我们也需要一个 Cell 对象来进行抽象。我们对这里的 B 树定一个约定，让所有的存储的用户数据都放在最底层的叶子节点，而所有的中间节点我们都只存放指针，这样方便之后进行范围查找，实际上这种做法已经有点像 B+ 树了，只需要让所有的叶子节点连起来变成一个双链表就行了，但为了简单我们先不这么处理，只是单纯约定好中间节点只存放指针。这样就存在两种类型的 Cell，存放指针的，和存放用户数据的：

export class BTree {}
export class BTreeNode {}
export class PointerCell {}
export class KeyValueCell {}

export class BTree {}
export class BTreeNode {}
export class PointerCell {}
export class KeyValueCell {}

另外，为了方便查询临近的节点，我们可以先返回到上一级的父节点，再通过父节点存放的指针去访问兄弟节点，这就要求在遍历时存储一下经过的节点，方便进行回退一步这类操作，因此我们可以引入一个 Cursor 的游标对象来进行管理，在查询时，像将每一步经过的节点留下一个面包屑记录，通过这个面包屑，它就像游标一样，可以前进后退，任意访问历史记录。

export class Cursor {}

export class Cursor {}

那么大概的抽象设计过程就是这样了，总结一下：

对于 `get` 流程

get 操作相对来说比较简单：

通过 Cursor 来查找到键所应该存放的叶子节点 BTreeNode；
每一次获取 BTreeNode 都是通过 Pager 读取文件来获得 buffer；
找到节点内对应存放值的 KeyValueCell；
这样就可以把值返回，然后反序列化展示给用户了；

对于 `set` 流程

set 是最复杂的：

通过 Cursor 来查找到键所应该存放在哪个叶子节点 BTreeNode；
检查一下这个节点能不能放下我们的数据；
如果可以的话，那就直接放进去；
如果不行的话，需要进行节点的分裂；
先分配一个新的节点；
把当前节点后半部分的数据 copy 过去，当前节点只留下前一半数据；
通过 Cursor 回退到父节点，在父节点中插入指向新节点的指针和分隔键；
父节点也要检查能不能放下这个数据，放不下的话再次重复之前的分裂过程；
如果根节点都放不下的话，我们将根节点分裂，然后创建一个新的节点作为新的根节点；
更新相关的元数据，例如根节点的 id 等等。

实现

接下来就开始逐步开始实现我们刚才定义的类，抽象的构成大约是：

Database
Pager
Cursor
BTree
BTreeNode
Cell

先从最下面的 Cell 开始，这个类是用来存储数据和指针的，它的二进制格式，像前文提到的：

对于 Pointer 类型，需要有：

cell 类型
分隔键大小
子页的指针（其实就是子页的编号）
分隔键

对于存放键值对的 KeyValue 类型的 cell，需要有：

cell 类型
分隔键大小（也就是键的大小）
值的大小
键值对

实际上因为我们可以知道节点的类型，那么对 cell 的类型是并不需要持久化的，这样可以设计出一个非常简单的格式，对于中间节点中存放指针的 cell：

0                4               8            +key_size
+----------------+---------------+-------------+
| [int] key_size | [int] page_id | [bytes] key |
+----------------+---------------+-------------+

0                4               8            +key_size
+----------------+---------------+-------------+
| [int] key_size | [int] page_id | [bytes] key |
+----------------+---------------+-------------+

然后我们根据他们的格式定义出对应的读取数据的方法：

export class PointerCell {
  public static calcSize(keySize: number) {
    return 8 + keySize;
  }

  public static create(
    key: Buffer,
    childPageId: number,
    offset: number
  ): PointerCell {
    const buf = Buffer.alloc(8);
    buf.writeInt32BE(key.length, 0);
    buf.writeInt32BE(childPageId, 4);
    return new PointerCell(
      Buffer.concat([buf, key], buf.length + key.length),
      offset
    );
  }

  public readonly type = CellType.Pointer;

  public get keySize(): number {
    return this.buffer.readInt32BE(0); // 4 bytes
  }

  public get key(): Buffer {
    return this.buffer.slice(8, 8 + this.keySize);
  }

  public get childPageId(): number {
    return this.buffer.readInt32BE(4); // 4 bytes
  }

  public get size(): number {
    return this.buffer.length;
  }

  public readonly buffer: Buffer;
  public readonly offset: number;

  constructor(rawBuffer: Buffer, offset: number) {
    this.buffer = rawBuffer;
    this.offset = offset;
  }
}

export class PointerCell {
  public static calcSize(keySize: number) {
    return 8 + keySize;
  }

  public static create(
    key: Buffer,
    childPageId: number,
    offset: number
  ): PointerCell {
    const buf = Buffer.alloc(8);
    buf.writeInt32BE(key.length, 0);
    buf.writeInt32BE(childPageId, 4);
    return new PointerCell(
      Buffer.concat([buf, key], buf.length + key.length),
      offset
    );
  }

  public readonly type = CellType.Pointer;

  public get keySize(): number {
    return this.buffer.readInt32BE(0); // 4 bytes
  }

  public get key(): Buffer {
    return this.buffer.slice(8, 8 + this.keySize);
  }

  public get childPageId(): number {
    return this.buffer.readInt32BE(4); // 4 bytes
  }

  public get size(): number {
    return this.buffer.length;
  }

  public readonly buffer: Buffer;
  public readonly offset: number;

  constructor(rawBuffer: Buffer, offset: number) {
    this.buffer = rawBuffer;
    this.offset = offset;
  }
}

而对于叶子结点中存放数据的是这样子的：

0                4                  8           +key_size        +value_size
+----------------+------------------+-------------+----------------+
| [int] key_size | [int] value_size | [bytes] key | [bytes] value  |
+----------------+------------------+-------------+----------------+

0                4                  8           +key_size        +value_size
+----------------+------------------+-------------+----------------+
| [int] key_size | [int] value_size | [bytes] key | [bytes] value  |
+----------------+------------------+-------------+----------------+

它的实现如下：

export class KeyValueCell {
  public static calcSize(keySize: number, valueSize: number) {
    return 8 + keySize + valueSize;
  }

  public static create(
    key: Buffer,
    value: Buffer,
    offset: number
  ): KeyValueCell {
    const buf = Buffer.alloc(8);
    buf.writeInt32BE(key.length, 0);
    buf.writeInt32BE(value.length, 4);
    return new KeyValueCell(
      Buffer.concat([buf, key, value], buf.length + key.length + value.length),
      offset
    );
  }

  public readonly type = CellType.KeyValue;

  public get keySize(): number {
    return this.buffer.readInt32BE(0); // 4 bytes
  }

  public get valueSize(): number {
    return this.buffer.readInt32BE(4); // 4 bytes
  }

  public get key(): Buffer {
    return this.buffer.slice(8, 8 + this.keySize);
  }

  public get value(): Buffer {
    return this.buffer.slice(
      8 + this.keySize,
      8 + this.keySize + this.valueSize
    );
  }

  public get size(): number {
    return this.buffer.length;
  }

  public readonly buffer: Buffer;
  public readonly offset: number;

  constructor(rawBuffer: Buffer, offset: number) {
    this.buffer = rawBuffer;
    this.offset = offset;
  }
}

export class KeyValueCell {
  public static calcSize(keySize: number, valueSize: number) {
    return 8 + keySize + valueSize;
  }

  public static create(
    key: Buffer,
    value: Buffer,
    offset: number
  ): KeyValueCell {
    const buf = Buffer.alloc(8);
    buf.writeInt32BE(key.length, 0);
    buf.writeInt32BE(value.length, 4);
    return new KeyValueCell(
      Buffer.concat([buf, key, value], buf.length + key.length + value.length),
      offset
    );
  }

  public readonly type = CellType.KeyValue;

  public get keySize(): number {
    return this.buffer.readInt32BE(0); // 4 bytes
  }

  public get valueSize(): number {
    return this.buffer.readInt32BE(4); // 4 bytes
  }

  public get key(): Buffer {
    return this.buffer.slice(8, 8 + this.keySize);
  }

  public get value(): Buffer {
    return this.buffer.slice(
      8 + this.keySize,
      8 + this.keySize + this.valueSize
    );
  }

  public get size(): number {
    return this.buffer.length;
  }

  public readonly buffer: Buffer;
  public readonly offset: number;

  constructor(rawBuffer: Buffer, offset: number) {
    this.buffer = rawBuffer;
    this.offset = offset;
  }
}

我们直接使用了用户提交的键值对中的键作为分隔键，而他们又都会被存储为二进制的 buffer，因此直接使用 Buffer.compare 来进行比较大小从而排序。它的接口很简单的，就是传入两个 buffer，然后返回一个数字。这个数字可能值为 -1，0 或者 1，分别表示小于，等于或者大于。

所有的 Cell 都会存储在 BTreeNode 中，使用前面提到的 Slotted Pages 技术，首先给每一个节点分配一点空间作为头部，例如这里使用了固定的 8 个字节，存放一些元信息，然后剩下的部分先摆放一个个有序的，存储着 cell 位置的指针（占用 2 个字节的 16 位整数），然后 cell 按照写入顺序从页的尾部开始，从后往前摆放：

0        8            freeStart   cellAreaStart          4096
+--------+---------------+------------+-------------------+
| header | cell_pointers | free_space | cell_content_area |
+--------+---------------+------------+-------------------+

0        8            freeStart   cellAreaStart          4096
+--------+---------------+------------+-------------------+
| header | cell_pointers | free_space | cell_content_area |
+--------+---------------+------------+-------------------+

而在它的头部中，就记录着 freeStart 和 cellAreaStart 的位置，随着数据的增多，freeStart 会逐渐向后移动，而 cellAreaStart 会逐渐向前移动：

0                  1                  3                       5           8
+------------------+------------------+-----------------------+-----------+
| [int] page_type  | [int] free_start | [int] cell_area_start | reserved  |
+------------------+------------------+-----------------------+-----------+

0                  1                  3                       5           8
+------------------+------------------+-----------------------+-----------+
| [int] page_type  | [int] free_start | [int] cell_area_start | reserved  |
+------------------+------------------+-----------------------+-----------+

这样 BTreeNode 的结构就定义好了：

export const enum PageType {
  EMPTY = 0x00,
  LEAF = 0x0d,
  INTERNAL = 0x05,
}

export class BTreeNode {
  public static isEqualKey(a: Buffer, b: Buffer) {
    return Buffer.compare(a, b) === 0;
  }

  public readonly id: number;
  public readonly buffer: Buffer;

  // header
  private get pageType(): PageType {
    return this.buffer.readInt8(0);
  }

  private set pageType(t: PageType) {
    this.buffer.writeUInt8(t);
  }

  private get freeStart(): number {
    return this.buffer.readInt16BE(1);
  }

  private set freeStart(n: number) {
    this.buffer.writeUInt16BE(n, 1);
  }

  private get cellAreaStart(): number {
    return this.buffer.readInt16BE(3);
  }

  private set cellAreaStart(n: number) {
    this.buffer.writeUInt16BE(n, 3);
  }

  constructor(id: number, rawBuffer: Buffer) {
    this.id = id;
    this.buffer = rawBuffer;
  }
}

export const enum PageType {
  EMPTY = 0x00,
  LEAF = 0x0d,
  INTERNAL = 0x05,
}

export class BTreeNode {
  public static isEqualKey(a: Buffer, b: Buffer) {
    return Buffer.compare(a, b) === 0;
  }

  public readonly id: number;
  public readonly buffer: Buffer;

  // header
  private get pageType(): PageType {
    return this.buffer.readInt8(0);
  }

  private set pageType(t: PageType) {
    this.buffer.writeUInt8(t);
  }

  private get freeStart(): number {
    return this.buffer.readInt16BE(1);
  }

  private set freeStart(n: number) {
    this.buffer.writeUInt16BE(n, 1);
  }

  private get cellAreaStart(): number {
    return this.buffer.readInt16BE(3);
  }

  private set cellAreaStart(n: number) {
    this.buffer.writeUInt16BE(n, 3);
  }

  constructor(id: number, rawBuffer: Buffer) {
    this.id = id;
    this.buffer = rawBuffer;
  }
}

注意我们实际上并不会在节点的头部存储它的 id，因为每一个节点都是从根节点开始通过 id 作为指针寻址找到的，文件的头部可以存放根节点的 id，而根节点内又存放了子节点的 id，自上往下，因此子节点并不需要浪费空间来存储自身的 id。

我们还可以定义一些辅助方法：

export class BTreeNode {
  //...
  public isEmptyNode() {
    return this.pageType === PageType.EMPTY;
  }

  public isLeafNode() {
    return this.pageType === PageType.LEAF;
  }

  public isInternalNode() {
    return this.pageType === PageType.INTERNAL;
  }
}

export class BTreeNode {
  //...
  public isEmptyNode() {
    return this.pageType === PageType.EMPTY;
  }

  public isLeafNode() {
    return this.pageType === PageType.LEAF;
  }

  public isInternalNode() {
    return this.pageType === PageType.INTERNAL;
  }
}

而 BTree ，主要就是提供 find 和 insert 的接口，同时它依赖着 Pager 和 Cursor:

export class BTree {
  public root: BTreeNode | null = null;
  public readonly pager: Pager;
  public readonly cursor: Cursor;

  constructor(pager: Pager, cursor: Cursor) {
    this.pager = pager;
    this.cursor = cursor;
    this.root = this.cursor.getRoot();
  }

  public find(key: Buffer): Buffer | null {}
  public insert(key: Buffer, value: Buffer) {}
}

export class BTree {
  public root: BTreeNode | null = null;
  public readonly pager: Pager;
  public readonly cursor: Cursor;

  constructor(pager: Pager, cursor: Cursor) {
    this.pager = pager;
    this.cursor = cursor;
    this.root = this.cursor.getRoot();
  }

  public find(key: Buffer): Buffer | null {}
  public insert(key: Buffer, value: Buffer) {}
}

下面我们就继续实现完整的查询逻辑！

Build a Simple Database

Build a Simple Database

代码实现

对于 `get` 流程

对于 `set` 流程

实现

Build a Simple Database

Build a Simple Database

代码实现

对于 get 流程

对于 set 流程

实现

对于 `get` 流程

对于 `set` 流程