65.9K
CodeProject 正在变化。 阅读更多。
Home

使用 C++ 简化可扩展存储引擎 (ESE) API - 第一部分

Espen Harlinn
starIconstarIconstarIconstarIconstarIcon

5.00/5 (11投票s)

2020年8月27日

CPOL

14分钟阅读

viewsIcon

12068

downloadIcon

334

易于使用的 C++ 封装类， 用于 Extensible Storage Engine (ESE) C API

引言

相关文章

可扩展存储引擎的 C API 有点吓人，这可能就是为什么这个优秀的数据库引擎没有得到应有广泛使用的原因——如果它的 API 更易于访问的话。

使用 ESE 的主要原因是它

  • 随 Windows 一同安装，因此它会随操作系统的其余部分一起获得更新和错误修复。你很可能每天都在使用它，因为它被微软的关键技术所使用，例如
    • Microsoft Exchange
    • Active Directory
    • Windows 搜索
  • 它非常可靠。
  • 它相当快——每秒插入超过 100,000 条记录并不罕见。

在本文中，我将演示如何

  • 初始化数据库引擎并创建一个会话
  • 创建数据库
  • 打开一个现有数据库
  • 创建表
  • 打开一个现有表
  • 向表中添加列
  • 为表创建主索引
  • 向表中插入行
  • 在表中查找一行
  • 更新表中的一行
  • 使用事务
  • 根据搜索条件从表中选择一系列行
  • 在表或行范围内向前和向后迭代

测试代码位于 EseTests.cpp 文件中，已包含在源代码中，提供了演示如何存储和检索 ESE 支持的所有数据类型的示例。 

下载内容中还提供了一个演示 ESE 性能的示例

  • 在 5 分钟多一点的时间里插入了 44,640,000 行：每秒 140,160 行。
  • 在 5 分钟多一点的时间里更新了 44,640,000 行：每秒 138,427 行。
  • 以每秒超过 900,000 行的速度搜索并检索行。

虽然以上数字非常出色，但我还没有找到一种方法能让 ESE 在从表中删除行时表现优异。

要查看它在您自己系统上的运行情况，请构建并执行

ExampleCoreESE01.exe -c -r -t F:\<Path to database directory>\Database.edb

其中

  • -c 告诉程序创建一个新数据库
  • -r 告诉程序替换该位置上任何现有的数据库
  • -t 告诉程序执行性能测试 

这需要你的硬盘上有大约 10GB 的可用空间。  

该库通过提供使用 ESE C API 的 Unicode 版本和 ANSI 版本的重载，允许您同时使用基于 Unicode (UTF16) 和 char 的字符串，并在适用时根据参数的字符串/字符类型选择其中之一。

几年前，Artour Bakiev 写了一篇优秀的文章：可扩展存储引擎 (Extensible Storage Engine)，该文章演示了如何使用 ESE C API——它也展示了仅使用 Windows SDK 提供的 API 开始使用 ESE 所需付出的努力。

我想要的是一个简单方便的 C++ API，它能轻松访问引擎的功能，而不隐藏任何能力。创建引擎实例、启动会话、以及创建或打开数据库是一个常见的模式——它应该很简单

BOOST_AUTO_TEST_CASE( InstanceInitializationTest )
{
    Ese::Instance instance( "TestInstance" );
    auto session = instance.BeginSession( );
    auto database = session.CreateDatabase( DatabasePath, 
                                            Ese::CreateDatabaseFlags::OverwriteExisting );
}

如果出现任何问题，该库将抛出异常，因此不需要额外的错误检查——现在我们已经拥有了使用 ESE 数据库所需的一切。不过，这些异常必须在某个地方处理。

该库的大部分内容都使用了 Visual Studio 和 Doxygen 能理解的 XML 文档标签进行注释，这使得弄清楚一个函数的作用变得很容易。

构建和运行代码需要 boost C++ 库（https://boost.ac.cn/）。所提供的 Visual Studio 项目期望环境变量 BOOST_ROOT 指向您解压该发行版的目录，并且库文件可以在 $(BOOST_ROOT)\stage\lib 中找到。

Harlinn::Common::Core::Ese

要使用该库，只需 #include <HCCEse.h>，并链接 Harlinn.Common.Core.lib 库。

通常，将包含 Harlinn.Common.Core.lib 库的目录添加到 Visual Studio 中您 C++ 项目的“属性：链接器->常规”下的“附加库目录”条目中就足够了。

该库中有四个主要的类，它们封装了 ESE C API 的句柄类型。这些类是可移动赋值和移动构造的，但不可复制赋值和复制构造。这种设计旨在确保句柄的生命周期能以适当的方式方便地管理，确保当所属对象超出作用域时句柄总是被关闭。

实例

Instance 是根对象，所有对该库的使用都应该从创建此类型的实例开始。

Instance 对象持有并管理一个 ESE 实例句柄的生命周期。

Session

Session 对象持有并管理一个 ESE 会话句柄的生命周期。会话提供了 ESE 的事务上下文，所有 ESE 数据库操作都通过会话执行。

当一个实例被多个线程使用时，每个线程必须有自己的会话。

数据库

Database 对象持有并管理一个数据库句柄的生命周期。ESE 数据库句柄用于管理数据库的模式，并用于管理数据库内部的表。数据库句柄只能与创建它们的会话一起使用，该库有助于促进这一点。

表格

Table 对象持有并管理数据库游标句柄的生命周期。ESE 游标句柄用于读取行数据、搜索行；或创建、更新和删除行。它还用于定义表的列和索引。与数据库句柄一样，数据库游标句柄只能与创建它们的会话一起使用，该库也有助于促进这一点。

概念

该库的实现使用了模板，其中许多模板都通过少数几个概念（concepts）进行了约束。我发现使用概念比使用 std::enable_if<…> 以及我们以前必须处理的其他基于 SFINAE 的“技巧”更能显著提高代码的可读性。

DirectType 概念用于帮助编译器为在读写 ESE 时不需要任何修改的数据类型选择正确的模板。

template<typename T>
concept DirectType = ( ( std::is_integral_v<T> && !std::is_same_v<bool, T> ) ||
    std::is_floating_point_v<T> ||
    std::is_same_v<TimeSpan, T> ||
    std::is_same_v<Currency, T> ||
    std::is_same_v<Guid, T> );

这个概念用于为那些可以直接传递给 ESE C API 的数据类型选择模板，使用它消除了因使用 std::enable_if<…> 造成的混乱。

template<DirectType T>
void SetColumn( JET_COLUMNID columnid, const T& value, SetFlags flags = SetFlags::None ) const
{
    auto rc = SetColumn( columnid, &value, sizeof( std::decay_t<T> ), flags, nullptr );
    RequireSuccess( rc );
}

第一步

通常情况下，一个应用程序会有一个单独的 Ese::Instance 对象，以及多个会话，其中包含一个或多个打开的数据库，但为了简化本文的代码，我将它们捆绑到一个名为 Engine 的类中。

class Engine
{
public:
    Ese::Instance instance;
    Ese::Session session;
    Ese::Database database;

    Engine( bool createNewDatabase = true )
    {
        instance = Ese::Instance( "TestInstance" );
        instance.SetCreatePathIfNotExist( );
        instance.SetExceptionAction( Ese::ExceptionAction::None );
        instance.SetSystemPath( DatabaseSystemPath );
        instance.SetLogFilePath( DatabaseLogfilePath );

        session = instance.BeginSession( );
        if ( createNewDatabase )
        {
            database =
               session.CreateDatabase( DatabasePath, 
                                       Ese::CreateDatabaseFlags::OverwriteExisting );
        }
        else
        {
            session.AttachDatabase( DatabasePath );
            database = session.OpenDatabase( DatabasePath );
        }
    }
};

如果 createNewDatabase 为 false，Engine 类将打开一个现有的数据库。

调用 instance.SetCreatePathIfNotExist( ) 会告诉数据库引擎静默创建文件系统中任何缺失的必需目录，而无需开发者采取进一步行动。

instance.SetExceptionAction( Ese::ExceptionAction::None ) 用于告诉数据库引擎在发生错误时不要显示对话框。

instance.SetSystemPath( DatabaseSystemPath ) 用于设置将包含实例检查点文件的目录路径。

instance.SetLogFilePath( DatabaseLogfilePath ) 用于设置将包含实例事务日志的目录路径。

这些选项通常会来自配置文件或注册表中的某个键。

数据库引擎可以在全局、实例和会话级别上设置大量的参数。为这些参数提供一个类型安全的 API 并为其编写文档是一项持续的工作。检索参数的函数命名为 QueryParameterName，设置参数的函数命名为 SetParameterName。

Engine 类不是库的一部分；其目的是简化测试用例的编写。有了它，创建一张表就变得非常简单

BOOST_AUTO_TEST_CASE( CreateTableTest1 )
{
    Engine engine;
    auto& database = engine.database;
    auto table1 = database.CreateTable( "table1" );
    BOOST_TEST( table1.IsValid( ) );
}

CreateTable(...) 的实现看起来非常简单，但它是一个模板函数，允许您在需要时指定自己的 Table 实现。

template<TableType T = Table, CharType C>
[[nodiscard]] T CreateTable(const C* tablename,
                            unsigned long initialNumberOfPages = 1,
                            unsigned long density = 0) const
{
    JET_TABLEID tableId = 0;
    auto rc = CreateTable( tablename, initialNumberOfPages, density, &tableId );
    RequireSuccess( rc );
    T result( sessionId_, tableId );
    result.OnTableCreated( );
    return result;
}

如你所见，我们仍然可以完全访问 JetCreateTableA 提供的所有功能，但通常情况下，我们对默认值很满意。

现在我们有了一张表，是时候添加一列了

auto columnId = table1.AddText( "fieldName" );

以上代码创建了一个可变大小的文本列，最大长度设置为 127 个字符。这意味着即使我们存储 UTF16 编码的文本，它也可以参与索引。插入一行只需要几个步骤

std::string ValueToInsert( "Text to store" );
table1.Insert( );
table1.SetColumn( columnId, ValueToInsert );
table1.Update( );

table1.Insert( ) 准备游标以插入一条新记录，并将所有列初始化为默认状态。如果表有自增列，则无论更新是否完成，都会为该记录分配一个新值。

table1.SetColumn(…) 为库支持的所有数据类型提供了合适的重载。

为了能够向数据库写入长度为 0 的字符串，SetColumn 会将标志 SetFlags::ZeroLength 添加到 flags 参数中。

template<StringType T>
void SetColumn( JET_COLUMNID columnId, const T& text, SetFlags flags = SetFlags::None ) const
{
    using CharT = typename T::value_type;
    DWORD length = static_cast<unsigned long>( text.length( ) * sizeof( CharT ) );
    if ( !length )
    {
        flags |= SetFlags::ZeroLength;
    }
    auto rc = SetColumn( columnId, text.c_str( ), length, flags, nullptr );
    RequireSuccess( rc );
}

如果没有这个标志，指定长度为 0 会告诉数据库引擎将 NULL 赋给该列。

这个 C++ 模板提供了一个对 std::string 和 std::wstring 都适用的实现。

该库提供了一个单独的函数

table1.SetNull( columnId );

用于将 NULL 赋给列。

现在我们已经向数据库写入了一个值，我们可以使用以下方式读回它

auto value1 = table1.As<std::string>( columnId );

As<…>(columnId) 提供了一种便捷的方式，既可以读取列值，又可以判断是否存储了 NULL 值，因为它返回一个 std::optional<T>。

template<typename T>
std::optional<T> As( JET_COLUMNID columnId, RetrieveFlags flags = RetrieveFlags::None ) const
{
    T data;
    if ( Read( columnId, data, flags ) )
    {
        return std::optional<T>( std::move( data ) );
    }
    else
    {
        return std::optional<T>( );
    }
}

如果 value1.has_value( ) 返回 false，则该列为 NULL。另一个方便的选择是使用 table1.Read(…)，如果该列为 NULL，它将返回 false。Read(…) 函数为库支持的所有数据类型都提供了重载。

template<DirectType T>
bool Read( JET_COLUMNID columnId, T& value, RetrieveFlags retrieveFlags = RetrieveFlags::None ) const
{
    constexpr unsigned long DataSize = sizeof( std::decay_t<T> );
    unsigned long actualDataSize;
    auto rc = RetrieveColumn( columnId, &value, DataSize, &actualDataSize, retrieveFlags, nullptr );
    return CheckReadResult( rc );
}

通过利用 C++ 模板，上述 Read(...) 的实现允许我们检索以下列类型的数据：char、signed char、unsigned char、short、unsigned short、long、unsigned long、long long、unsigned long long、float、double、TimeSpan、Currency 和 Guid。

该库扩展了可与 ESE 一起使用的数据类型数量，它会尝试将 C++ 类型映射到最“合理”的 ESE 支持类型。TimeSpan 是一种与 .Net TimeSpan 值类型非常相似的类型，其中时间段以“ticks”为单位进行测量。该库将 TimeSpan 类型映射为 long long。DateTime 内部也以 long long 存储数据，但它会被转换为 double 类型，因为这是 ESE 用于 datetime 的原生类型。该库支持以下数据类型与 ESE 一起使用

  • bool
  • char/signed char/unsigned char
  • short/unsigned short
  • long/unsigned long
  • long long/unsigned long long
  • float
  • double
  • 货币
  • 日期时间
  • TimeSpan
  • Guid
  • std::string/std::wstring
  • std::vector<char>/std::vector<signed char>/std::vector<unsigned char>
  • Core::IO::MemoryStream

一个列是文本列还是长文本列，取决于创建该列时其最大长度的设定，而在读写数据库时，无论是哪种类型都是透明的。二进制列和长二进制列的处理方式类似。在为表定义列时，您可以使用以下函数

  • AddBoolean
  • AddSByte
  • AddByte
  • AddInt16
  • AddUInt16
  • AddInt32
  • AddUInt32
  • AddInt64
  • AddUInt64
  • AddSingle
  • AddDouble
  • AddCurrency
  • AddTimeSpan
  • AddDateTime
  • AddGuid
  • AddText
  • AddBinary

这些都是小型的包装函数，可以让 Visual Studio 中的代码补全功能为您服务，这样您就不必总是查阅文档。通过提供将 ESE C API 中相关的宏定义组合在一起的 enum class 类型，也支持代码补全。

是时候来点更复杂的东西了。假设我们想要存储由 Guid 标识的传感器的计量数据，并且我们有以下结构来保存数据

struct SensorValue
{
    Guid Sensor;
    DateTime Timestamp;
    Int64 Flags = 0;
    Double Value = 0.0;
};

本着保持简单的精神，我们创建一个从 Engine 派生的新类 SensorEngine

class SensorEngine : public Engine
{
public:
    ...
};

该示例的完整源代码在 EseTests.cpp 中提供，要运行示例代码，请执行

Harlinn.Common.Core.Tests.exe --run_test=EseTests/InsertSearchAndUpdateSensorValueTableTest1

从命令行。

SessionEngine 有几个变量

class SensorEngine : public Engine
{
public:
    ...
    JET_COLUMNID SensorColumnId;
    JET_COLUMNID TimestampColumnId;
    JET_COLUMNID FlagsColumnId;
    JET_COLUMNID ValueColumnId;
    Ese::Table SensorValues;
    ...
};

各个 JET_COLUMNID 成员将保存表中四列的列ID，而 Ese::Table SensorValues 则持有该表的数据库游标句柄。

创建这张表很简单

void CreateSensorValueTable( )
{
    session.BeginTransaction( );

    SensorValues = database.CreateTable( SensorValueTableName );
    SensorColumnId = SensorValues.AddGuid( SensorColumnName );
    TimestampColumnId = SensorValues.AddDateTime( TimestampColumnName );
    FlagsColumnId = SensorValues.AddUInt64( FlagsColumnName );
    ValueColumnId = SensorValues.AddDouble( ValueColumnName );
    SensorValues.CreateIndex( SensorValueIndexName, Ese::IndexFlags::Primary, L"+S\0+T\0", 6 );

    session.CommitTransaction( );
}

对 SensorValues.CreateIndex(…) 的调用用于为表创建一个主索引。第三个参数指定了索引的字段和顺序。‘+’表示升序，而‘-’则表示降序。‘S’是传感器ID的Guid列名，‘T’是 SensorValue 结构中 Timestamp 成员的 DateTime 列名。键规范的每个部分都以‘\0’结尾，而编译器在字符串末尾生成的隐式‘\0’提供了终止规范所需的两个连续的‘\0’值。在它们自己的事务中执行数据定义操作也是一个好习惯。

要打开一个已经存在的表，你可以这样做

void OpenSensorValueTable( )
{
    SensorValues = database.OpenTable( SensorValueTableName );

    SensorColumnId = SensorValues.GetColumnId( SensorColumnName );
    TimestampColumnId = SensorValues.GetColumnId( TimestampColumnName );
    FlagsColumnId = SensorValues.GetColumnId( FlagsColumnName );
    ValueColumnId = SensorValues.GetColumnId( ValueColumnName );

    SensorValues.SetCurrentIndex( SensorValueIndexName );
}

既然我们有了一张表，我们需要在该表中插入和更新行

void Insert( const SensorValue& value )
{
    SensorValues.Insert( );

    SensorValues.SetColumn( SensorColumnId, value.Sensor );
    SensorValues.SetColumn( TimestampColumnId, value.Timestamp );
    SensorValues.SetColumn( FlagsColumnId, value.Flags );
    SensorValues.SetColumn( ValueColumnId, value.Value );

    SensorValues.Update( );
}

void Update( const SensorValue& value )
{
    if ( MoveTo( value.Sensor, value.Timestamp ) )
    {
        SensorValues.Replace( );

        SensorValues.SetColumn( FlagsColumnId, value.Flags );
        SensorValues.SetColumn( ValueColumnId, value.Value );

        SensorValues.Update( );
    }
}

我们已经了解了在 ESE 中向表中插入行的机制，更新行也相当简单。SensorValues.Replace( ) 准备游标进行一次更新，这次更新不会修改属于主键的列。

然后我们设置列值，再告诉数据库引擎我们已经完成了对记录的修改。

MoveTo(…) 的这个实现会为由 sensorId 标识的传感器找到一个时间戳小于或等于参数 timestamp 的现有行。

bool MoveTo( const Guid& sensorId, const DateTime& timestamp ) const
{
    SensorValues.MakeKey( sensorId, Ese::KeyFlags::NewKey );
    SensorValues.MakeKey( timestamp );

    auto rc = SensorValues.Seek( Ese::SeekFlags::LessOrEqual );
    return rc >= Ese::Result::Success;
}

通过传递 Ese::KeyFlags::NewKey 标志，我们告诉引擎正在为数据库游标创建一个新的搜索键。连续调用不带此标志的 MakeKey(…) 会按照用于创建当前索引的键字符串指定的顺序，向当前搜索键中添加字段。

使用 Ese::SeekFlags::LessOrEqual 调用 SensorValues.Seek(…) 会告诉引擎将游标定位在至少有一个字段匹配的行上，而其余字段的列数据可以比较为小于或等于。在这种情况下，我们有了一个函数，它可以找到一个时间戳与参数匹配的记录，或者是时间戳**小于**参数的**最大**时间戳的记录。

还有一个 MoveTo(...) 的重载

bool MoveTo( const Guid& sensorId ) const
{
    SensorValues.MakeKey( sensorId, Ese::KeyFlags::NewKey );
    auto rc = SensorValues.Seek( Ese::SeekFlags::GreaterOrEqual );
    return rc >= Ese::Result::Success;
}

它将游标定位到由 sensorId 标识的传感器的第一行。这可以用来遍历为该 sensorId 存储的所有行

if ( engine.MoveTo( firstSensor ) )
{
    size_t rowCount = 0;
    double sum = 0.0;
    do
    {
        auto sensorId = engine.Sensor( );
        if ( sensorId != firstSensor )
        {
            break;
        }
        sum += engine.Value( );
        rowCount++;
    } while ( sensorValues.MoveNext( ) );
    printf( "MoveTo - Count: %zu, sum: %f\n", rowCount, sum );
}

在这种情况下，我们必须检查游标是否超出了为所请求传感器存储的数据范围。实现这一点的更好方法是在主键索引上设置一个筛选器。

bool Filter( const Guid& sensorId ) const
{
    SensorValues.MakeKey( sensorId, Ese::KeyFlags::NewKey );
    auto rc = SensorValues.Seek( Ese::SeekFlags::GreaterOrEqual );
    if ( rc >= Ese::Result::Success )
    {
        SensorValues.MakeKey( sensorId, Ese::KeyFlags::NewKey );
        SensorValues.MakeKey( DateTime::MaxValue() );
        SensorValues.SetIndexRange( Ese::IndexRengeFlags::UpperLimit );
        return true;
    }
    else
    {
        return false;
    }
}

该实现与前面 MoveTo(…) 函数的实现开始时完全相同，我们只需要设置一个新的包含以下内容的搜索键：

  1. 传感器 ID
  2. 一个大于或等于可能存储在 timestamp 列中最大值的值。

然后调用 SensorValues.SetIndexRange(…)，告诉数据库引擎当前键现在代表了我们想要迭代的范围的上限。

if ( engine.Filter( firstSensor ) )
{
    size_t rowCount = 0;
    double sum = 0.0;
    do
    {
        sum += engine.Value( );
        rowCount++;
    } while ( sensorValues.MoveNext( ) );
    printf( "Filter - Count: %zu, sum: %f\n", rowCount, sum );
}

将表的数据库游标移动到某个传感器的最后一个条目也相当容易。

bool MoveToLast( const Guid& sensorId ) const
{
    SensorValues.MakeKey( sensorId, Ese::KeyFlags::NewKey );
    SensorValues.MakeKey( DateTime::MaxValue() );
    auto rc = SensorValues.Seek( Ese::SeekFlags::LessOrEqual );
    return rc >= Ese::Result::Success;
}

这个搜索键的逻辑与上面为 Filter(…) 函数创建的第二个搜索键相同。现在我们可以向后遍历该传感器的行

if ( engine.MoveToLast( secondSensor ) )
{
    size_t rowCount = 0;
    double sum = 0.0;
    do
    {
        auto sensorId = engine.Sensor( );
        if ( sensorId != secondSensor )
        {
            break;
        }
        sum += engine.Value( );
        rowCount++;
    } while ( sensorValues.MovePrevious( ) );
    printf( "MoveToLast - Count: %zu, sum: %f\n", rowCount, sum );
}

我们仍然需要检查我们没有越过范围的起始点，和之前一样，我们可以设置一个筛选器。

bool ReverseFilter( const Guid& sensorId ) const
{
    SensorValues.MakeKey( sensorId, Ese::KeyFlags::NewKey );
    SensorValues.MakeKey( DateTime::MaxValue( ) );

    auto rc = SensorValues.Seek( Ese::SeekFlags::LessOrEqual );
    if ( rc >= Ese::Result::Success )
    {
        SensorValues.MakeKey( sensorId, Ese::KeyFlags::NewKey );
        SensorValues.SetIndexRange( );
        return true;
    }
    else
    {
        return false;
    }
}

使用这个过滤器可以显著简化反向循环

if ( engine.ReverseFilter( secondSensor ) )
{
    size_t rowCount = 0;
    double sum = 0.0;
    do
    {
        sum += engine.Value( );
        rowCount++;
    } while ( sensorValues.MovePrevious( ) );
    printf( "ReverseFilter - Count: %zu, sum: %f\n", rowCount, sum );
}

 

现在我希望我已经说服您在下一个需要快速可靠存储引擎的 C++ 项目中尝试使用可扩展存储引擎。我已在多个项目中使用 ESE，发现它确实非常稳健。在调试过程中，我可能在更新/事务中途杀死了进程数千次——数据库从未因此而损坏。

在生产环境中，我有一些解决方案在 ESE 中存储了超过 10 TB 的数据，引擎能够高效可靠地处理这些数据。

与其他 NoSQL 引擎相比，我非常欣赏 ESE 允许我在一张表上创建多个索引，因为这是你经常需要使用其他存储引擎提供的简单键/值 API 自己实现的功能。

历史

  • 2020年8月28日 - 首次发布。
  • 2020年8月31日 - 代码清理，使用概念（concepts）消除了几个 requires 子句。 
  • 2020年10月6日 - 错误修复，清理了大部分单元测试。
  • 2020年10月7日 - 为 Harlinn.Common.Core 库添加了更多单元测试。
© . All rights reserved.