MongoDB DBA 设计/图形架构师中级开发 .NET C#

使用 MongoDB 实现语义数据库 - 第一部分

Marc Clifton

5.00/5 (3投票s)

2016年2月12日

CPOL

19分钟阅读

20294

一个包含 3 个部分的系列文章的第一部分，介绍如何使用 MongoDB 实现语义数据库。

引言

正如我之前写过的

语义数据库是一种新型的数据库存储。由于尚不存在这样的数据库引擎，我们必须在现有引擎之上进行构建。在我之前的文章中，我讨论了如何在一个 SQL 数据库之上构建一个语义数据库。在本文中，我们将探讨如何在一个 NoSQL 数据库（特别是 MongoDB）之上构建一个语义数据库，并说明使用 NoSQL 数据库的一些优点。

随着 2015 年 12 月 MongoDB 3.2 版本的发布，这变得更加容易（也更现实），因为 3.2 版本支持使用 $lookup 聚合器进行左外连接。

本文涵盖了在 MongoDB 中为语义数据库创建基本的 CRUD 操作。

第二部分将涵盖一个实际的例子，无疑会“揭露”本文中的实现缺陷。第一部分涵盖了处理单个语义类型的关系层次结构，而第二部分将涵盖处理“语义类型之间”的关系。

第三部分将是惊喜！

代码在哪里？

代码可以在 GitHub 上找到：https://github.com/cliftonm/mongodbSemanticDatabase。

由于这是一项正在进行的努力，请从那里下载最新代码。例如，本文中使用的一些 Newtonsoft（尤其是 JObject）的使用已经在最新实现和测试中被 BsonDocument 替换。

语义数据库的基础概念

从根本上说，语义数据库捕捉关系。有两种主要的关系类型：

静态的、隐式的关系，它们定义了语义术语（符号）的结构（赋予意义）。这些通常用与面向对象编程中相同的术语来表达：“has a”（拥有）和“is a kind of”（是一种）。
静态或动态的显式关系，其中关系本身具有以语义术语表达的意义，并且动态关系可以随时间而变化。在编程中，这些关系通常在代码中隐式表达，例如，字典或其他键值对集合。动态关系通常有一个时间框架——开始和结束。

我已经在我的“面向关系编程”系列文章中写过关于显式动态关系的内容。

语义数据库实际上结合了语义和关系性，正如我们在这里将要演示的那样。

静态、隐式语义关系示例

人的名字是隐式语义关系的一个好例子。人们的名字（但在所有文化中并非如此，当然，即使在一种文化中也存在差异）由名字和姓氏组成。符号“名字”和“姓氏”本身就是符号“name”的特例。最后，“name”符号是（在此数据类型中）符号“string”的一个特例。这些都是“has a”和“is a kind of”关系。

显式关系示例

关系存在于专门的语义类型之间，我们可以将它们进一步分为静态关系和动态关系。

静态关系示例

固定关系是那些随时间永不改变的关系。例如，一个人总是与另外两个人有关系：“mother”（母亲）和“father”（父亲）。在一个人的一生中，这个人与其出生星座（狮子座、双子座等）之间存在静态关系。当然，构成该星座的星星实际上处于动态关系中。

动态关系示例

与静态关系相比，并非每个人都有儿子或女儿。这种关系可能会出现，并且当它出现时，它会与特定的时间戳相关联。一个人的名字可以是一种动态关系——考虑昵称和别名。所有权、居住地、婚姻、宗教信仰、犯罪记录、政党归属——这些都是潜在动态关系的例子。大多数关系都是动态的，发生在（可能重复发生）特定的时间间隔内。

显式关系本身就是语义的

像“mother”（母亲）、“daughter”（女儿）、“Republican”（共和党人）、“Buddhist”（佛教徒）这样的关系是语义的。它们表达了一种关系或一种状态，因此我们可以说

“[X] 是 [Z] 的 [Y]”——表达了关系，例如，“Marc 是 Elisabeth 的儿子”

或

“[X] 是一个 [Y]”——表达了一种关系状态，例如，“Marc 是一个人类学家”

或

“[X] [Y] [Z]”——表达了一种活动，例如，“Marc 拥有一只猫”

这些都是语义上（哈哈）描述关系。当我们说什么状态性的东西时，例如“[X] 是一个 [Y]”，我们可以将其转换为一个介词短语来描述关系，变成“[X] 与 [Y] 存在 [Z]”或类似的内容，尽管你可能需要发挥一点想象力。例如，“Marc 是一个男人”，作为一种存在状态，可以被关系性地描述为“Marc 是一个有男性性别的人类”，或者“Marc 拥有男性性别”。无论哪种情况，我们都做了一些有用的事情——我们识别了一个状态类别（性别），它允许我们明确地描述 Marc 和性别之间的关系。关于语言语义就说到这里！

结构关系和符号关系

语义数据库必须同时捕捉符号的层次结构以及符号与其他符号的关系。例如，如果我们有两个符号，一个代表人的名字，另一个代表地址，我们可以轻松地查询非典型关系：“给我列出所有名字也恰好是他们所居住城镇的街道名称的人。”

NoSQL 数据库的优势

虽然这是在 SQL 数据库中可以轻松想象的查询，但它假设查询要使用的模式已经存在。NoSQL 数据库的优势在于其模式本身是动态的。

隐式符号的结构会发生变化（想想名字和地址在不同文化中的差异）。
可以轻松添加新符号（只需添加一个新集合）。
可以轻松添加符号之间的新关系（只需添加一个包含两个字段的集合，关联两个集合的 ID）。

在 SQL 数据库中，这需要操作模式，而在 NoSQL 数据库中，这是不必要的（尽管出于性能考虑，我们仍然需要关注索引）。

使用 NoSQL 数据库消除了模拟真正语义数据库所需的大量工作，但如前所述，直到 $lookup 聚合器的出现，这都被留给了客户端实现，而不是由服务器端处理。在我们看来，在客户端处理不是一个可行的解决方案，因为需要大量内存和带宽才能拉取一个集合的所有记录，然后通过另一个集合的左连接来消除其中许多记录，而这个集合也必须完全加载到客户端内存中。幸运的是，MongoDB 的 $lookup 聚合器解决了这个问题。

NoSQL 数据库仍然不是语义数据库

仍然有大量工作需要完成（SQL 数据库也是如此），但这些工作实际上只涉及

创建集合
在运行时动态创建查询
维护规范化的数据库
引用计数
- 处理插入，避免数据重复
- 处理删除，确保被其他集合引用的数据在引用数为 0 之前不会被删除
- 处理更新，确保被其他集合引用的数据得到维护，并且只改变父引用

语义数据库中的规范化与非规范化数据

NoSQL 数据库可以轻松处理非规范化数据库——文档中的子文档。当然，在 SQL 数据库中创建非规范化模式也很容易，而这正是大多数遗留数据库的弊端。语义数据库的黄金法则是

所有数据都必须规范化！

语义数据库依赖于数据是规范化的这一事实——两个符号之间的关系是基于多对多关系集合的。它**不是**基于比较字段值构建的。以上面的人名和街道地址为例，典型的 SQL 查询会是这样的

select * from Person p
left join Address a on a.StreetName = p.FirstName

一个*语义*查询（在 SQL 中）看起来是这样的

select * from Name n                                  -- Name symbol
left join FirstName fn on fn.NameId = n.Id            -- n:m collection
left join PersonName pn on pn.FirstNameId = n.NameId  -- n:m collection
left join Person p on p.Id = pn.PersonId              -- Person symbol, having a 
                                                      -- Name association through "pn"
left join StreetName sn on sn.NameId = n.Id           -- n:m collection
left join AddressStreetName asn on asn.StreetNameId = sn.Id -- n:m collection
left join Address a on a.Id = asn.AddressId           -- Address symbol, having a 
                                                      -- StreetName association through "asn"

这看起来很糟糕（而且我不确定我是否写对了），但它表达了两个原本不相关的符号之间的语义结构和关系。

那么，重点是什么？

重点是，通过语义数据库，仍然可以查询两个符号之间不存在显式关系的关系。例如，可以向语义数据库提问

我能否通过一个人的名字将一个 Person 与一个 Address 联系起来？
除了显式关系之外，我还能通过哪些方式将一个 Person 与一个 Address 联系起来？

语义数据库可以递归地遍历语义结构，以发现（并验证）关联数据的新方法！一个非语义数据库（无论是 SQL 还是 NoSQL）都无法做到这一点，因为从“FirstName”到“StreetName”没有可发现的路径。

语义数据库在发现新关系方面的能力使其脱颖而出，这种关系不是基于显式定义的结构或动态关系，而是基于具有共享语义的关系。

应用

立即想到三个应用：

大数据
物联网
记录管理

大数据分析（“数据量巨大或复杂，以至于传统数据处理应用程序无法胜任”—— https://en.wikipedia.org/wiki/Big_data）在我不知情的情况下，仍然依赖于人类显式确定数据集之间的关系。语义数据库可以轻松查询以前未知关系，并促进将分散的数据集与显式关系含义相关联。

其次，与大数据一样，物联网将产生海量数据。在我看来，将这些数据关联成比初始数据集更有意义的东西的唯一方法是使其具有语义并将其存储在语义数据库中。

记录管理（如医疗、紧急情况、犯罪记录）已经是庞大、混乱且不相关的系统。此外，它一直在变化。

模式——人们想要追踪的信息
生命周期——疾病的发生（以及希望的消退），火被扑灭，人们犯下新的罪行，等等
关系——人与物之间的关系一直在变化

语义数据库是解决我们目前存储公共和非私密信息的各种、不兼容和有限的数据库的关键问题的解决方案。

实现

如前所述，在真正的语义数据库中，我们可以提出这些“我能否关联……”和“给出……的交集”的问题，并让服务器端完成繁重的工作。但是，由于真正的语义数据库不存在，我们必须自己完成这项繁重的工作。但首先，我们必须实现核心功能。第二部分将探讨我们可以用语义数据库做的更有趣的事情。

您需要什么

如果您以前没有将 MongoDB 与 C# 一起使用过，您需要

下载并安装 MongoDB 服务器
下载并安装 2.2.0 或更高版本的 MongoDB .NET 驱动程序
运行 MongoDB 服务器 mongod.exe，可以在控制台窗口或作为服务运行
- mongod.exe 的 3.2 64 位版本通常位于 C:\Program Files\MongoDB\Server\3.2\bin
可以选择下载并安装 RoboMongo，以便在一个漂亮的 GUI 中检查您的集合。

测试驱动开发

这是对测试驱动开发的一个很好的应用，因为我们可以说明很多关于我们对语义数据库的预期前置条件、函数和后置条件。当我们实现每个单元测试，添加新的行为要求，并实现使测试通过的方法时，您会注意到有时之前的测试以及通常的实现都会经常重构。这些是概念上我们想要执行的测试（这里的“集合”指的是 NoSQL 集合）。

创建具体的语义类型集合
创建分层的语义类型，演示非具体的特化（听起来像矛盾修辞法）如何产生具体集合和多对多关系集合
定义语义类型之间的关系，使用多对多关系集合
关系
1. 能够提问“一个语义类型的显式关系是什么？”
2. 能够提问“一个语义类型的可发现（隐式）关系是什么？”
3. 能够提问“这个语义类型的结构是什么？”
对于具体的和特化的语义类型
1. Insert：自动创建多对多关系实例文档和具体集合文档
2. 更新:
  1. 更新单例文档
  2. 将具有多个引用的文档解耦为两个独立的文档，并更新多对多引用
3. 删除:
  1. 删除单例文档及其层次结构
  2. 仅当一个文档被多个特化语义实例引用时才删除层次结构集合
4. 查询:
  1. 查询具体的语义类型
  2. 查询特化的具体类型，自动生成连接以解析到具体实例
  3. 查询两个或多个语义类型，自动生成连接以关联语义类型

命名约定

使用了以下命名约定：

集合和数据库名称为驼峰式（标识符的首字母小写，之后每个连接词的首字母大写）。
集合名称为单数。
字段名称为驼峰式。
多对多集合
- 单数名称
- 被引用的两个集合对象之间用下划线分隔
- 连接集合的 ID 格式为
  - [collectionName1]Id
  - [collectionName2]Id

创建具体的语义类型集合

我们将从创建一些具体的语义类型开始。具体语义类型通常描述单个具体类型，所以它相当基础。关于创建具体类型的极端情况，总有一个问题。例如，我们可以定义一个电话号码

PhoneNumber：string（甚至可以是数字）

或

phoneNumber
- countryCode：int
- areaCode：int
- exchange：int
- subscriberId：int

或

phoneNumber
- countryCode
  - number : int
- areaCode
  - number : int
- exchange
  - number : int
- subscriberId
  - number : int

不，我们不会将“number”分解成 0-32 的位！最后一个例子代表过度规范。我们可以通过问“我们是从具体回到一般，还是从一般到具体？”来判断何时过度规范化了语义类型。例如

电话号码到国家代码、区号、交换号和用户 ID 是从一般到具体的移动。
国家代码到“number”是从具体到一般的移动。

这为我们提供了一个明确定义的方法来决定何时停止类型层次结构。

规则/验证

国家代码、区号、交换号——它们都有特定的规则，所以即使它们是整数，也有格式规则、最小和最大长度限制等等。我们最终（但不是在这篇文章中）希望将一个具体类型与一个规则表中的规则集合关联起来。

我们定义的语义类型越具体，我们就越可能遇到表示冲突，尤其是文化冲突。并非所有国家都遵守北美拨号计划——显然只有 24 个国家遵守。

因此，我们的第一个语义类型是

countryCode
areaCode
exchange
subscriberId

我们观察到一个语义类型总是以一个我们称为“value”的字段结尾，此外，我们注意到一个具体的语义类型只有一个值字段。如果它有多个，那么它将是一个复合类型，我们应该将其分解为构成它的子类型。所以这里我们有另一个规则：

一个具体的语义类型通常只有一个值字段。

乍一看，这可能非常奇怪、违反直觉且效率低下，但它使我们能够完全以语义方式表达数据。查找集合（键值对）是此规则的一个例外。

由语义类型产生的第二件事是，通常存在没有具体值的占位符集合。我称之为“abstract”集合，因为它们只包含对子集合的引用。

[TestMethod]
public void CreateConcreteCollection()
{
  SemanticDatabase sd = Helpers.CreateCleanDatabase();
  Assert.IsTrue(sd.GetCollections().Count == 0, "Collection should be 0 length.");
  Schema schema = Helpers.InstantiateSchema("{name: 'countryCode'}");
  sd.InstantiateSchema(schema);
  List<string> collections = sd.GetCollections();
  Assert.IsTrue(collections.Count == 1, "Expected 1 collection.");
  Assert.IsTrue(collections[0] == "countryCode", "Collection does not match expected name");
}

此实现的核心工作是由于 2.2.0 版本中的一个错误而产生的荒谬的解决方法（这种错误在开源项目中真的让人感觉很糟糕）。

public void CreateCollection(string collectionName)
{
  // This throws a NullReferenceException!
  // See here: https://jira.mongodb.org/browse/CSHARP-1524
  // This is apparently fixed in version 2.2.1
  // db.CreateCollection(collectionName);

  // For now, we use this workaround:

  // As per the documentation: MongoDB creates collections automatically when 
  // they are first used, so you only need to call this method if you want 
  // to provide non-default options.
  // What we do here is create a collection with a single entry, 
  // then delete that item, thus resulting in an empty collection. 
  // While I get the "don't create it until you need it" concept, 
  // there are reasons (like my unit tests) for why I want the collection 
  // actually physically created.
  var data = new BsonDocument(collectionName, "{Value : 0}");
  var collection = db.GetCollection<BsonDocument>(collectionName);
  collection.InsertOne(data);
  var result = collection.DeleteOne(new BsonDocument
              ("_id", data.Elements.Single(el => el.Name == "_id").Value));
}}

创建分层的语义类型

在这里，我们测试从层次结构创建集合。

[TestMethod]
public void CreateSpecializedCollection()
{
  SemanticDatabase sd = Helpers.CreateCleanDatabase();
  Assert.IsTrue(sd.GetCollections().Count == 0, "Collection should be 0 length.");
  Schema schema = Helpers.InstantiateSchema(@"
  {
    name: 'phoneNumber', 
    subtypes:
    [
      {name: 'countryCode'},
      {name: 'areaCode'},
      {name: 'exchange'},
      {name: 'subscriberId'},
    ]
  }");
  sd.InstantiateSchema(schema);
  List<string> collections = sd.GetCollections();
  Assert.IsTrue(collections.Count == 5, "Expected 5 collections.");
  Assert.IsTrue(collections.Contains("phoneNumber"));
  Assert.IsTrue(collections.Contains("countryCode"));
  Assert.IsTrue(collections.Contains("areaCode"));
  Assert.IsTrue(collections.Contains("exchange"));
  Assert.IsTrue(collections.Contains("subscriberId"));
}}

同样，这里没有什么神奇之处——创建的是空集合。

由于这些是无模式的文档集合，所以当然没有字段定义。

插入具体语义类型

在这里，我们测试向具体语义类型插入记录。底层实现非常基础，因为我们仍然需要编写更复杂、更有趣的语义层次结构插入。但首先是基本测试。

[TestMethod]
public void InsertConcreteTypeTest()
{
  SemanticDatabase sd = Helpers.CreateCleanDatabase();
  Assert.IsTrue(sd.GetCollections().Count == 0, "Collection should be 0 length.");
  Schema schema = Helpers.InstantiateSchema(@"
  {
    name: 'countryCodeLookup', 
    concreteTypes:
    {
      value: 'System.Int32',
      name: 'System.String'
    }
  }");

  sd.InstantiateSchema(schema);
  Assert.IsTrue(sd.GetCollections().Count == 1, "Collection should be length of 1.");

  sd.Insert(schema, "{value: 1, name: 'United States'}");
  sd.Insert(schema, "{value: 20, name: 'Egypt'}");
  sd.Insert(schema, "{value: 30, name: 'Greece'}");

  List<string> json = sd.GetAll("countryCode");

  Assert.IsTrue(json[0].Contains("{ \"value\" : 1, \"name\" : \"United States\" }"));;
  Assert.IsTrue(json[1].Contains("{ \"value\" : 20, \"name\" : \"Egypt\" }"));
  Assert.IsTrue(json[2].Contains("{ \"value\" : 30, \"name\" : \"Greece\" }"));
}

以及基本实现（请注意，它完全忽略了模式）。

public void Insert(Schema schema, string json)
{
  db.GetCollection<BsonDocument>(schema.Name).InsertOne(BsonDocument.Parse(json));
}

插入语义层次结构

现在，关于上述实现，敏锐的读者可能会说，“嗯，国家不也是一个语义类型吗？”确实如此，所以让我们创建一个合适的语义层次结构并编写一个插入层次结构的测试。

Name Collection（名称集合）

“name”集合是一种奇怪的生物。它是一种语义泛化，但它也有一个特定的含义——它是事物的名称。我们人类就是靠命名事物的，亚瑟·C·克拉克甚至写了一个短篇故事，说人类的创造完全是为了列出上帝的所有名字，之后宇宙就结束了——《上帝的九十亿个名字》。因此，“name”是语义数据库中的一个特殊集合，以便实体可以被命名。

请注意，我们仍然在插入一个扁平化的层次结构。稍后我们将讨论解析重复字段名和分层插入。

[TestMethod]
public void InsertHierarchyTest()
{
  SemanticDatabase sd = Helpers.CreateCleanDatabase();
  Assert.IsTrue(sd.GetCollections().Count == 0, "Collection should be 0 length.");
  Schema schema = Helpers.InstantiateSchema(@"
  {
    name: 'countryCode', 
    concreteTypes:
    {
      value: 'System.Int32',
    },
    subtypes: 
    [
      {
        name: 'countryName', 
        subtypes: 
        [
          {
            name: 'name', 
            concreteTypes:
            {
              name: 'System.String'
            }
          }
        ]
      }
    ]
  }");
  sd.InstantiateSchema(schema);
  Assert.IsTrue(sd.GetCollections().Count == 3, "Collection should be length of 3.");
  sd.Insert(schema, JObject.Parse("{value: 1, name: 'United States'}"));
  sd.Insert(schema, JObject.Parse("{value: 20, name: 'Egypt'}"));
  sd.Insert(schema, JObject.Parse("{value: 30, name: 'Greece'}"));


  Assert.IsTrue(json[0].Contains("{ \"name\" : \"United States\" }"));
  Assert.IsTrue(json[1].Contains("{ \"name\" : \"Egypt\" }"));
  Assert.IsTrue(json[2].Contains("{ \"name\" : \"Greece\" }"));

  json = sd.GetAll("countryName");
  Assert.IsTrue(json.Count==3);

  json = sd.GetAll("countryCode");
  Assert.IsTrue(json[0].Contains("{ \"value\" : 1"));;
  Assert.IsTrue(json[1].Contains("{ \"value\" : 20"));
  Assert.IsTrue(json[2].Contains("{ \"value\" : 30"));}}

注意我们正在创建的层次结构，并且中间语义类型“countryName”没有任何具体值。为什么要这样做？因为从语义上讲，它允许我们查询两件不同的事情：

给我列出所有有名字的事物的名字。
给我列出所有国家的名字。

注意，顶层实现已经变得更加复杂，递归地进入模式结构，在递归时移除具体类型，并将引用 ID 添加到父 JSON 对象中。

public string nsert(Schema schema, JObject jobj)
{
  string id = null;

  if (schema.IsConcreteType)
  {
    id = Insert(schema.Name, jobj);
  }
  else
  {
    JObject currentObject = GetConcreteObjects(schema, jobj);
    JObject subjobj = RemoveCurrentConcreteObjects(schema, jobj);
    RecurseIntoSubtypes(schema, currentObject, subjobj);
    id = Insert(schema.Name, currentObject);
  }}

  return id;
}

插入重复项测试

接下来我们深入探讨一些更复杂的场景：在语义数据库中，我们绝不会复制记录，无论它在层次结构中的哪个位置。相反，我们增加引用计数。

在语义数据库中，这完全规范化了与特定语义类型相关联的值，因此我们永远不需要比较两个语义层次结构之间的值，因为我们依赖于语义模式中数据的规范化来创建不同语义结构之间的“连接”。当数据库的整个模式设计良好时，这效果很好，但它不能阻止我们在两个数据库之间进行值比较。

[TestMethod]
public void InsertDuplicateHierarchyTest()
{
  SemanticDatabase sd = Helpers.CreateCleanDatabase();
  Assert.IsTrue(sd.GetCollections().Count == 0, "Collection should be 0 length.");
  Schema schema = Helpers.InstantiateSchema(@"
  {
    name: 'countryCode', 
    concreteTypes:
    {
      value: 'System.Int32',
    },
    subtypes: 
    [
      {
        name: 'countryName', 
        subtypes: 
        [
          {
            name: 'name', 
            concreteTypes:
            {
              name: 'System.String'
            }
          }
        ]
      }
    ]
  }");

  sd.InstantiateSchema(schema);
  Assert.IsTrue(sd.GetCollections().Count == 3, "Collection should be length of 3.");
  sd.Insert(schema, JObject.Parse("{value: 1, name: 'United States'}"));

  List<string> json;

  json = sd.GetAll("name");
  Assert.IsTrue(json.Count == 1);
  Assert.IsTrue(json[0].Contains("\"name\" : \"United States\""));

  json = sd.GetAll("countryName");
  Assert.IsTrue(json.Count == 1);

  json = sd.GetAll("countryCode");
  Assert.IsTrue(json.Count == 1);
  Assert.IsTrue(json[0].Contains("\"value\" : 1"));

  // Duplicate insert:
  sd.Insert(schema, JObject.Parse("{value: 1, name: 'United States'}"));
  json = sd.GetAll("name");
  Assert.IsTrue(json.Count == 1);
  Assert.IsTrue(json[0].Contains("\"_ref\" : 2 }"));

  json = sd.GetAll("countryName");
  Assert.IsTrue(json.Count == 1);
  Assert.IsTrue(json[0].Contains("\"_ref\" : 2 }"));

  json = sd.GetAll("countryCode");
  Assert.IsTrue(json.Count == 1);
  Assert.IsTrue(json[0].Contains("\"_ref\" : 2 }"));
}

为了使此测试通过，insert 方法再次被重构，但您应该注意到具体语义实例插入和层次结构插入之间存在一个模式。

public string Insert(Schema schema, JObject jobj)
{
  string id = null;

  if (schema.IsConcreteType)
  {
    int refCount;

    if (IsDuplicate(schema.Name, jobj, out id, out refCount))
    {
      IncrementRefCount(schema.Name, id, refCount);
    }
    else
    {
      JObject withRef = AddRef1(jobj);
      id = Insert(schema.Name, withRef);
    }
  }
  else
  {
    JObject currentObject = GetConcreteObjects(schema, jobj);
    JObject subjobj = RemoveCurrentConcreteObjects(schema, jobj);
    RecurseIntoSubtypes(schema, currentObject, subjobj);
    int refCount;

    if (IsDuplicate(schema.Name, currentObject, out id, out refCount))
    {
      IncrementRefCount(schema.Name, id, refCount);
    }
    else
    {
      JObject withRef = AddRef1(currentObject);
      id = Insert(schema.Name, withRef);
    }
  }

  return id;
}

具体语义查询测试

这个针对具体语义类型的 Query 调用实际上只是一个对 GetAll() 方法的调用，但我们仍然在此进行测试。

[TestClass]
public class QueryTests
{
  [TestMethod]
  public void ConcreteQueryTest()
  {
    SemanticDatabase sd = Helpers.CreateCleanDatabase();
    Assert.IsTrue(sd.GetCollections().Count == 0, "Collection should be 0 length.");
    Schema schema = Helpers.InstantiateSchema(@"
    {
      name: 'countryCodeLookup', 
      concreteTypes:
      {
        value: 'System.Int32',
        name: 'System.String'
      }
    }");

    sd.InstantiateSchema(schema);
    Assert.IsTrue(sd.GetCollections().Count == 1, "Collection should be length of 1.");
  
    sd.Insert(schema, JObject.Parse("{value: 1, name: 'United States'}"));
    List<JObject> records = sd.Query(schema);

    Assert.IsTrue(records.Count == 1);
    Assert.IsTrue(records[0].Contains("{\"value\":1,\"name\":\"United States\""));
  }
}

分层语义查询测试

客户端

此查询返回指定层次结构中语义实例的扁平化记录（文档数据库也就如此了！）。

[TestMethod]
public void HierarchicalQueryTest()
{
  SemanticDatabase sd = Helpers.CreateCleanDatabase();
  Assert.IsTrue(sd.GetCollections().Count == 0, "Collection should be 0 length.");
  Schema schema = Helpers.InstantiateSchema(@"
  {
    name: 'countryCode', 
    concreteTypes:
    {
      value: 'System.Int32',
    },
    subtypes: 
    [
      {
        name: 'countryName', 
        subtypes: 
        [
          {
            name: 'name', 
            concreteTypes:
            {
              name: 'System.String'
            }
          }
        ]
      }
    ]
  }");

  sd.InstantiateSchema(schema);
  Assert.IsTrue(sd.GetCollections().Count == 3, "Collection should be length of 3.");
  sd.Insert(schema, JObject.Parse("{value: 1, name: 'United States'}"));
  sd.Insert(schema, JObject.Parse("{value: 20, name: 'Egypt'}"));
  sd.Insert(schema, JObject.Parse("{value: 30, name: 'Greece'}"));

  List<BsonDocument> json;

  json = sd.Query(schema);
  Assert.IsTrue(json.Count == 3);

  Assert.IsTrue(json[0].ToString().Contains("\"value\" : 1, \"name\" : \"United States\""));
  Assert.IsTrue(json[1].ToString().Contains("\"value\" : 20, \"name\" : \"Egypt\""));
  Assert.IsTrue(json[2].ToString().Contains("\"value\" : 30, \"name\" : \"Greece\""));
}

我们可以为客户端实现重构查询。

public List<BsonDocument> Query(Schema schema, string id = null)
{
  List<BsonDocument> records = new List<BsonDocument>();

  records = GetAll(schema.Name, id);

  foreach (BsonDocument record in records)
  {
    record.Remove("_ref");

    foreach (Schema subtype in schema.Subtypes)
    {
      string childIdName = subtype.Name + "Id";
      // Remove the FK ID, as we don't want it in the final recordset
      string childId = record[childIdName].ToString();
      record.Remove(childIdName);
      List<BsonDocument> childRecords = Query(subtype, childId);

      // TODO: Assert that childRecords <= 1, and we know only one child record exists 
      // because we don't allow duplicates.
      if (childRecords.Count == 1)
      {
        childRecords[0].Elements.ForEach(p => record.Add(p.Name, childRecords[0][p.Name]));
      }
    }
  }

  return records;
}

服务器端

相反，我们可以构建一个 MongoDB 查询在服务器端运行。在 Mongo 控制台中，它看起来像这样：

db.countryCode.aggregate(
  {$lookup: {from: "countryName", localField:"countryNameId", 
  foreignField: "_id", as: "countryName"} },
  {$unwind: "$countryName"},
  {$lookup: {from: "name", localField:"countryName.nameId", foreignField: "_id", as: "name"} },
  {$unwind: "$name"},
  {$project: {"value": "$value", "name": "$name.name", "_id":0} }
)

请参阅我的文章《使用 MongoDB 的 $lookup 聚合器》，了解如何使用 $lookup 来连接集合。

我们将扩展我们的测试，以同时测试服务器端。

json = sd.QueryServerSide(schema);
Assert.IsTrue(json.Count == 3);

Assert.IsTrue(json[0].ToString().Contains("\"value\" : 1, \"name\" : \"United States\""));
Assert.IsTrue(json[1].ToString().Contains("\"value\" : 20, \"name\" : \"Egypt\""));
Assert.IsTrue(json[2].ToString().Contains("\"value\" : 30, \"name\" : \"Greece\""));

我们执行一个服务器端查询，通过检查模式来构建聚合器。

public List<BsonDocument> QueryServerSide(Schema schema, string id = null)
{
  var collection = db.GetCollection<BsonDocument>(schema.Name);
  List<string> projections = new List<string>();
  List<string> pipeline = BuildQueryPipeline(schema, String.Empty, projections);
  pipeline.Add(String.Format
             ("{{$project: {{{0}, '_id':0}} }}", String.Join(",", projections)));
  var aggr = collection.Aggregate();
  pipeline.ForEach(s => aggr = aggr.AppendStage<BsonDocument>(s));
  List<BsonDocument> records = aggr.ToList();

  return records;
}

和

protected List<string> BuildQueryPipeline
(Schema schema, string parentName, List<string> projections)
{
  List<string> pipeline = new List<string>();

  schema.ConcreteTypes.ForEach(kvp => projections.Add
 (String.Format("'{0}':'${1}'", kvp.Key, parentName + kvp.Key)));

  foreach (Schema subtype in schema.Subtypes)
  {
    pipeline.Add(String.Format("{{$lookup: {{from: '{0}', localField:'{2}{1}', 
   foreignField: '_id', as: '{0}'}} }},", subtype.Name, subtype.Name + "Id", parentName));
    pipeline.Add(String.Format("{{$unwind: '${0}'}}", subtype.Name));
    List<string> subpipeline = BuildQueryPipeline(subtype, subtype.Name + ".", projections);

    if (subpipeline.Count > 0)
    {
      pipeline[pipeline.Count - 1] = pipeline.Last() + ",";
      pipeline.AddRange(subpipeline);
    }
  }

  return pipeline;
}

删除测试

删除具体实例测试

在这里，我们测试删除一个具体实例。

[TestMethod]
public void DeleteSingleInstanceTest()
{
  SemanticDatabase sd = Helpers.CreateCleanDatabase();
  Assert.IsTrue(sd.GetCollections().Count == 0, "Collection should be 0 length.");
  Schema schema = Helpers.GetSimpleTestSchema();

  sd.InstantiateSchema(schema);
  Assert.IsTrue(sd.GetCollections().Count == 1, "Collection should be length of 1.");

  sd.Insert(schema, JObject.Parse("{value: 1, name: 'United States'}"));
  sd.Insert(schema, JObject.Parse("{value: 20, name: 'Egypt'}"));
  sd.Insert(schema, JObject.Parse("{value: 30, name: 'Greece'}"));

  List<BsonDocument> bson = sd.GetAll("countryCodeLookup");
  Assert.IsTrue(bson.Count == 3);

  sd.Delete(schema, JObject.Parse("{value: 1, name: 'United States'}"));

  bson = sd.GetAll("countryCodeLookup");
  Assert.IsTrue(bson.Count == 2);

  Assert.IsTrue(bson[0].ToString().Contains("\"value\" : 20, \"name\" : \"Egypt\""));
  Assert.IsTrue(bson[1].ToString().Contains("\"value\" : 30, \"name\" : \"Greece\""));
}

删除的实现与 insert 非常相似，我们递减引用计数（如果大于 1），否则删除集合记录。

protected string Delete(Schema schema, BsonDocument doc)
{
  string id = null;

  if (schema.IsConcreteType)
  {
    int refCount = GetRefCount(schema.Name, doc, out id);

    if (refCount == 1)
    {
      Delete(schema.Name, id);
    }
    else
    {
      DecrementRefCount(schema.Name, id, refCount);
    }
  }
  else
  {
    BsonDocument currentObject = GetConcreteObjects(schema, doc);
    BsonDocument subjobj = RemoveCurrentConcreteObjects(schema, doc);
    DeleteRecurseIntoSubtypes(schema, currentObject, subjobj);
    int refCount = GetRefCount(schema.Name, currentObject, out id);

    if (refCount == 1)
    {
      Delete(schema.Name, id);
    }
    else
    {
      DecrementRefCount(schema.Name, id, refCount);
    }
  }

  return id;
}

由于上面的内容与插入过程非常相似，所以我决定先编写完整的实现，然后再编写附加测试！

删除多重引用测试

[TestMethod]
public void DeleteMultipleReferenceTest()
{
  SemanticDatabase sd = Helpers.CreateCleanDatabase();
  Assert.IsTrue(sd.GetCollections().Count == 0, "Collection should be 0 length.");
  Schema schema = Helpers.GetSimpleTestSchema();

  sd.InstantiateSchema(schema);
  Assert.IsTrue(sd.GetCollections().Count == 1, "Collection should be length of 1.");

  sd.Insert(schema, JObject.Parse("{value: 1, name: 'United States'}"));
  sd.Insert(schema, JObject.Parse("{value: 20, name: 'Egypt'}"));
  sd.Insert(schema, JObject.Parse("{value: 30, name: 'Greece'}"));

  // second reference:
  sd.Insert(schema, JObject.Parse("{value: 1, name: 'United States'}"));

  List<BsonDocument> bson = sd.GetAll("countryCodeLookup");
  Assert.IsTrue(bson.Count == 3);

  // First delete:
  sd.Delete(schema, JObject.Parse("{value: 1, name: 'United States'}"));

  bson = sd.GetAll("countryCodeLookup");
  Assert.IsTrue(bson.Count == 3);

  Assert.IsTrue(bson[0].ToString().Contains("\"value\" : 1, \"name\" : \"United States\""));
  Assert.IsTrue(bson[1].ToString().Contains("\"value\" : 20, \"name\" : \"Egypt\""));
  Assert.IsTrue(bson[2].ToString().Contains("\"value\" : 30, \"name\" : \"Greece\""));

  // Second delete:
  sd.Delete(schema, JObject.Parse("{value: 1, name: 'United States'}"));

  bson = sd.GetAll("countryCodeLookup");
  Assert.IsTrue(bson.Count == 2);

  Assert.IsTrue(bson[0].ToString().Contains("\"value\" : 20, \"name\" : \"Egypt\""));
  Assert.IsTrue(bson[1].ToString().Contains("\"value\" : 30, \"name\" : \"Greece\""));
}

删除层次结构测试

[TestMethod]
public void DeleteHierarchyTest()
{
  SemanticDatabase sd = Helpers.CreateCleanDatabase();
  Assert.IsTrue(sd.GetCollections().Count == 0, "Collection should be 0 length.");
  Schema schema = Helpers.GetTestHierarchySchema();

  sd.InstantiateSchema(schema);
  Assert.IsTrue(sd.GetCollections().Count == 3, "Collection should be length of 3.");
  sd.Insert(schema, JObject.Parse("{value: 1, name: 'United States'}"));
  sd.Insert(schema, JObject.Parse("{value: 20, name: 'Egypt'}"));
  sd.Insert(schema, JObject.Parse("{value: 30, name: 'Greece'}"));

  List<BsonDocument> bson;

  bson = sd.Query(schema);
  Assert.IsTrue(bson.Count == 3);

  sd.Delete(schema, JObject.Parse("{value: 1, name: 'United States'}"));

  bson = sd.Query(schema);
  Assert.IsTrue(bson.Count == 2);

  Assert.IsTrue(bson[0].ToString().Contains("\"value\" : 20, \"name\" : \"Egypt\""));
  Assert.IsTrue(bson[1].ToString().Contains("\"value\" : 30, \"name\" : \"Greece\""));
}

删除多重引用层次结构测试

[TestMethod]
public void DeleteMultipleReferenceHierarchyTest()
{
  SemanticDatabase sd = Helpers.CreateCleanDatabase();
  Assert.IsTrue(sd.GetCollections().Count == 0, "Collection should be 0 length.");
  Schema schema = Helpers.GetTestHierarchySchema();

  sd.InstantiateSchema(schema);
  Assert.IsTrue(sd.GetCollections().Count == 3, "Collection should be length of 3.");
  sd.Insert(schema, JObject.Parse("{value: 1, name: 'United States'}"));
  sd.Insert(schema, JObject.Parse("{value: 20, name: 'Egypt'}"));
  sd.Insert(schema, JObject.Parse("{value: 30, name: 'Greece'}"));

  // Insert a record that re-uses the country name.
  sd.Insert(schema, JObject.Parse("{value: 2, name: 'United States'}"));

  List<BsonDocument> bson;

  bson = sd.Query(schema);
  Assert.IsTrue(bson.Count == 4);

  // Delete just the re-use high-level type.
  sd.Delete(schema, JObject.Parse("{value: 2, name: 'United States'}"));

  bson = sd.Query(schema);
  Assert.IsTrue(bson.Count == 3);

  Assert.IsTrue(bson[0].ToString().Contains("\"value\" : 1, \"name\" : \"United States\""));
  Assert.IsTrue(bson[1].ToString().Contains("\"value\" : 20, \"name\" : \"Egypt\""));
  Assert.IsTrue(bson[2].ToString().Contains("\"value\" : 30, \"name\" : \"Greece\""));
}

遗漏的测试

有一个重要的测试被遗漏了。

// TODO: We should not be able to delete a sub-type if it is referenced by
// a super-type. We need a master schema 
// to know whether a sub-type has a super-type somewhere,
// or we need to ask the DB for fields of the form "[subtype]Id", 
// which would indicate that the
// subtype is an FK in a supertype.

就“主模式”而言，由于 NoSQL 数据库是无模式的，因此不一定能轻松地从数据库本身提取它。但是，像 Variety 这样的模式分析工具看起来是一个很好的起点。当然，要确定模式，该工具必须实际检查每个集合中的*记录*。理想情况下，我们应该有一个独立的主模式，但在我撰写本文时，我还没有实现它。

更新测试

更新是最复杂/最有趣的。

必须提供原始语义类型的完整值集以及新值——我们实际上不能仅基于某个主键来更新值。
如果没有对该语义类型的其他引用，则可以直接更新具体类型。
如果存在其他引用：
1. 必须递减当前类型的引用计数。
2. 必须插入该类型的新实例。
3. 必须更新超类型的“外键”引用。
4. 此过程需要向上递归遍历层次结构。

第一点是语义数据库与典型关系数据库相比最显著的特征之一。

更新具体语义类型测试

这是最简单的测试，其中一个具体语义类型（没有子类型的）被更新。

[TestMethod]
public void UpdateConcreteTypeTest()
{
  SemanticDatabase sd = Helpers.CreateCleanDatabase();
  Assert.IsTrue(sd.GetCollections().Count == 0, "Collection should be 0 length.");
  Schema schema = Helpers.GetSimpleTestSchema();

  sd.InstantiateSchema(schema);
  Assert.IsTrue(sd.GetCollections().Count == 1, "Collection should be length of 1.");

  sd.Insert(schema, BsonDocument.Parse("{value: 1, name: 'United States'}"));
  sd.Insert(schema, BsonDocument.Parse("{value: 20, name: 'Egypt'}"));
  sd.Insert(schema, BsonDocument.Parse("{value: 30, name: 'Greece'}"));

  List<BsonDocument> bson = sd.GetAll("countryCodeLookup");
  Assert.IsTrue(bson.Count == 3);

  sd.Update(schema, BsonDocument.Parse("{value: 1, name: 'United States'}"), 
  BsonDocument.Parse("{value: 1, name: 'United States of America'}"));
  
  bson = sd.GetAll("countryCodeLookup");
  Assert.IsTrue(bson.Count == 3);

  Assert.IsTrue(bson[0].ToString().Contains
  ("\"value\" : 1, \"name\" : \"United States of America\""));
  Assert.IsTrue(bson[1].ToString().Contains("\"value\" : 20, \"name\" : \"Egypt\""));
  Assert.IsTrue(bson[2].ToString().Contains("\"value\" : 30, \"name\" : \"Greece\""));
}

更新语义层次结构的底部测试

在这里，我们进行一个与上面非常相似的测试，除了这次我们测试更新层次结构的底部元素。

/// <summary>
/// Test changing a concrete value at the bottom of the hierarcy.
/// </summary>
[TestMethod]
public void UpdateBottomHierarchySingleReferenceTest()
{
  SemanticDatabase sd = Helpers.CreateCleanDatabase();
  Assert.IsTrue(sd.GetCollections().Count == 0, "Collection should be 0 length.");
  Schema schema = Helpers.GetTestHierarchySchema();

  sd.InstantiateSchema(schema);
  Assert.IsTrue(sd.GetCollections().Count == 3, "Collection should be length of 3.");
  sd.Insert(schema, BsonDocument.Parse("{value: 1, name: 'United States'}"));
  sd.Insert(schema, BsonDocument.Parse("{value: 20, name: 'Egypt'}"));
  sd.Insert(schema, BsonDocument.Parse("{value: 30, name: 'Greece'}"));

  List<BsonDocument> bson;

  bson = sd.Query(schema);
  Assert.IsTrue(bson.Count == 3);

  // This tests updating the bottom of the hierarchy, 
  // and since there are no other references, we can update the only instance.
  sd.Update(schema, BsonDocument.Parse("{value: 1, name: 'United States'}"), 
  BsonDocument.Parse("{value: 1, name: 'United States of America'}"));
  bson = sd.Query(schema);
  Assert.IsTrue(bson.Count == 3);

  Assert.IsTrue(bson[0].ToString().Contains
  ("\"value\" : 1, \"name\" : \"United States of America\""));
  Assert.IsTrue(bson[1].ToString().Contains("\"value\" : 20, \"name\" : \"Egypt\""));
  Assert.IsTrue(bson[2].ToString().Contains("\"value\" : 30, \"name\" : \"Greece\""));
}

更新层次结构顶部测试

在这里，我们测试更新语义类型层次结构顶部的具体值。

/// <summary>
/// Test changing the country code (the value field), which is at the top of the hierarchy.
/// </summary>
[TestMethod]
public void UpdateTopHierarchySingleReferenceTest()
{
  SemanticDatabase sd = Helpers.CreateCleanDatabase();
  Assert.IsTrue(sd.GetCollections().Count == 0, "Collection should be 0 length.");
  Schema schema = Helpers.GetTestHierarchySchema();

  sd.InstantiateSchema(schema);
  Assert.IsTrue(sd.GetCollections().Count == 3, "Collection should be length of 3.");
  sd.Insert(schema, BsonDocument.Parse("{value: 1, name: 'United States'}"));
  sd.Insert(schema, BsonDocument.Parse("{value: 20, name: 'Egypt'}"));
  sd.Insert(schema, BsonDocument.Parse("{value: 30, name: 'Greece'}"));

  List<BsonDocument> bson;

  bson = sd.Query(schema);
  Assert.IsTrue(bson.Count == 3);

  // This tests updating the top of the hierachy, changing the country code from 1 to 3.
  sd.Update(schema, BsonDocument.Parse("{value: 1}"), BsonDocument.Parse("{value: 3}"));
  bson = sd.Query(schema);
  Assert.IsTrue(bson.Count == 3);

  Assert.IsTrue(bson[0].ToString().Contains("\"value\" : 3, \"name\" : \"United States\""));
  Assert.IsTrue(bson[1].ToString().Contains("\"value\" : 20, \"name\" : \"Egypt\""));
  Assert.IsTrue(bson[2].ToString().Contains("\"value\" : 30, \"name\" : \"Greece\""));
}

更新多重引用测试

在这里，我们测试更新层次结构的底部，当一个具体值被两个单独的语义实例引用时。

/// <summary>
/// Test changing the country name when there are two references to the same country.
/// </summary>
[TestMethod]
public void UpdateBottomHierarchyMultipleReferenceTest()
{
  SemanticDatabase sd = Helpers.CreateCleanDatabase();
  Assert.IsTrue(sd.GetCollections().Count == 0, "Collection should be 0 length.");
  Schema schema = Helpers.GetTestHierarchySchema();

  sd.InstantiateSchema(schema);
  Assert.IsTrue(sd.GetCollections().Count == 3, "Collection should be length of 3.");
  sd.Insert(schema, BsonDocument.Parse("{value: 1, name: 'United States'}"));
  sd.Insert(schema, BsonDocument.Parse("{value: 20, name: 'Egypt'}"));
  sd.Insert(schema, BsonDocument.Parse("{value: 30, name: 'Greece'}"));
  sd.Insert(schema, BsonDocument.Parse
 ("{value: 40, name: 'United States'}")); // The country name is in error.

  List<BsonDocument> bson;

  bson = sd.Query(schema);
  Assert.IsTrue(sd.GetAll("name").Count == 3); // "United States" has a ref count of 2
  Assert.IsTrue(bson.Count == 4);

  // Fix the country name:
  sd.Update(schema, BsonDocument.Parse("{value: 40, name: 'United States'}"), 
           BsonDocument.Parse("{value: 40, name: 'Romania'}"));
  bson = sd.Query(schema);
  Assert.IsTrue(sd.GetAll("name").Count == 4); // Now we should have four unique country names
  Assert.IsTrue(bson.Count == 4);

  Assert.IsTrue(bson[0].ToString().Contains("\"value\" : 1, \"name\" : \"United States\""));
  Assert.IsTrue(bson[1].ToString().Contains("\"value\" : 20, \"name\" : \"Egypt\""));
  Assert.IsTrue(bson[2].ToString().Contains("\"value\" : 30, \"name\" : \"Greece\""));
  Assert.IsTrue(bson[3].ToString().Contains("\"value\" : 40, \"name\" : \"Romania\""));
}

实现

高级实现是一个递归过程，我们确定记录的字段是否可以更新（只有一个引用）或者是否需要递减引用计数并创建新实例，这个过程会向上回溯到层次结构。我们还有一个特殊情况处理器用于部分语义类型——即扁平化数据未延伸到最低子类型的情况，如上面“更新层次结构顶部”测试所示。请注意此操作的复杂性以及在深入层次结构时对每个记录的 _id 的依赖，这对于在递归展开时确定是否需要更新超类型的引用至关重要。

protected string Update(Schema schema, BsonDocument docOriginal, 
                       BsonDocument docNew, string schemaId)
{
  string id = null;

  if (schema.IsConcreteType)
  {
    int refCount = GetRefCount(schema.Name, docOriginal, out id);

    if (refCount == 1)
    {
      Update(schema.Name, id, docOriginal, docNew);
    }
    else
    {
      // We never have 0 references, because this would have meant decrementing from 1, 
     // which would instead trigger and update above.
      DecrementRefCount(schema.Name, id, refCount);
      id = InternalInsert(schema, docNew);
    }
  }
  else
  {
    BsonDocument currentOriginalObject = GetConcreteObjects(schema, docOriginal);
    BsonDocument record = null;

    if (schemaId == null)
    {
      // We must use the concrete objects to determine the record.
      // If there are no concrete objects, we have an error.
      // There should be a single unique record for the concrete object.
      if (currentOriginalObject.Elements.Count() == 0)
      {
        throw new SemanticDatabaseException("Cannot update the a semantic type 
                        starting with the abstract type " + schema.Name);
      }

      record = GetRecord(schema.Name, currentOriginalObject);

      if (record == null)
      {
        throw new SemanticDatabaseException("The original record for the semantic type " + 
       schema.Name + " cannot be found.\r\nData: " + currentOriginalObject.ToString());
      }
    }
    else
    {
      // We use the subtype id to get the record.
      record = GetRecord(schema.Name, new BsonDocument("_id", new ObjectId(schemaId)));

      if (record == null)
      {
        throw new SemanticDatabaseException("An instance of " + 
        schema.Name + " with _id = " + schemaId + " does not exist!");
      }
    }

    BsonDocument subOriginalJobj = RemoveCurrentConcreteObjects(schema, docOriginal);

    if (subOriginalJobj.Elements.Count() == 0)
    {
      // There is nothing further to do, 
     // as we're not changing anything further in the hierarchy.
      // Update the current concrete types.
      id = record.Elements.Single(el => el.Name == "_id").Value.ToString();
      int refCount = record.Elements.Single(el => el.Name == "_ref").Value.ToInt32();

      if (refCount == 1)
      {
        BsonDocument currentNewObject = GetConcreteObjects(schema, docNew);
        Update(schema.Name, id, record, currentNewObject);
      }
      else
      {
        // TODO: THIS CODE PATH IS NOT TESTED!
        // Now we have a problem -- something else is referencing this record 
        // other than our current hierarch, 
        // but we don't know what. But we're updating this particular type instance. 
        // Do all the super-types reflect this change in the subtype? We'll assume no.
        // Only this hierarchy gets updated.
        DecrementRefCount(schema.Name, id, refCount);
        id = InternalInsert(schema, docNew);

        // Otherwise:
        // All supertypes referencing this hierarchy get updated.
        //BsonDocument currentNewObject = GetConcreteObjects(schema, docNew);
        //Update(schema.Name, id, record, currentNewObject);
      }
    }
    else
    {
      BsonDocument currentNewObject = GetConcreteObjects(schema, docNew);
      BsonDocument subNewJobj = RemoveCurrentConcreteObjects(schema, docNew);
      UpdateRecurseIntoSubtypes(schema, record, currentOriginalObject, 
                               subOriginalJobj, currentNewObject, subNewJobj);
      id = record.Elements.Single(el => el.Name == "_id").Value.ToString();
      int refCount = record.Elements.Single(el => el.Name == "_ref").Value.ToInt32();

      if (refCount == 1)
      {
        Update(schema.Name, id, record, currentNewObject);
      }
      else
      {
        // We never have 0 references, 
        // because this would have meant decrementing from 1, 
        // which would instead trigger and update above.
        DecrementRefCount(schema.Name, id, refCount);
        id = InternalInsert(schema, docNew);
      }
    }
  }

  return id;
}

结论

到目前为止一切顺利——尽管在路径测试中存在一些 TODO 和空白，但这足以继续进行第二部分。我意识到这里提出的想法可能很疯狂，而且这篇文章有点无聊，因为它基本上只是单元测试。然而，第二部分应该会变得更有趣一些，第三部分肯定会！

历史

2016 年 2 月 12 日：初始版本