物联网点位管理引擎

在物联网（Internet of Things，简称 IoT）中，“点位”通常指被监控或管理的物理对象上的具体信息点或数据采集点。这些点位可以是传感器、执行器、控制器等物联网设备上的具体接口或参数，也可以是设备本身作为一个整体被监控的标识。其中数据对象包括静态属性（ID，名称等）和动态属性（温度，压力，湿度，状态等）。

DolphinDB 在 3.00.2 版本中推出的基于 TSDB 引擎自研的、针对物联网场景中海量点位数据进行高效管理的存储引擎。

值类型可变的列，能够在单值模型的字段里存储不同类型的属性值。

系统实时更新最新值缓存，并直接从缓存而非磁盘中读取数据，创建以分区为单位的最新值缓存表。最新值缓存功能可在 create 或 createPartitionedTable 中通过 latestKeyCache 参数开启。功能说明小节将详细介绍其原理。

对 sortColumns 中指定的除最后时间列外的其它列进行映射。将其映射为内部 ID。在后续的存储和缓存操作中，系统将通过内部 ID 来代指某个测点，以达到节省存储空间的目的。该表与 SymbolBase 类似，以分区为单位。该映射会存储于静态表中，用户无感知。

IOTANY 列即值类型可变的列，建表时最多可以创建一个 IOTANY 列。IOTANY 列类型只能用在 create 语句中，不能单独建一个该类型的向量或者表。DolphinDB 在存储 IOTANY 列数据时， 将在 TSDB Level File 层级单独存储不同类型的值。

关于 TSDB 存储引擎，物联网点位管理引擎：

• 规定将唯一标识一组测点的多个列（一般为 id 列+各种 tag 列）+时间列设置为 sortColumns。

• 支持三种去重策略：ALL（保留所有数据，为默认值）， LAST（仅保留最新数据），FIRST（仅保留第一条数据）。

• 支持 HashMappingFunction。

• 暂不支持 softDelete。

• 最新值查询功能暂不支持 snapshot read。

• 暂不支持 level 4 compaction。

• 暂不支持向量数据库等。

• 暂不支持添加 IOTANY 列。

• 不支持 sortColumns 中包含 DECIMAL 类型。

• 不支持 upsert! 时设置 ignoreNull=true。

• 不支持 upsert! 时传入 IOTANY 向量。

推荐通过 create 语句创建含 IOTANY 列的点位管理表。

dbName = "dfs://db"
if (existsDatabase(dbName)) {
    dropDatabase(dbName)
}
// 创建数据库，存储引擎为IOTDB
create database "dfs://db" partitioned by HASH([INT, 20]),VALUE(2017.08.07..2017.08.11), engine='IOTDB'
// 创建点位表，其中 firstSortKey 为 deviceId 和 location，lastSortKey(时间戳列)为 timestamp
create table "dfs://db"."pt" (
        deviceId INT,
        location SYMBOL,
        timestamp TIMESTAMP,
        value IOTANY
)
partitioned by deviceId, timestamp,
sortColumns = [`deviceId, `location, `timestamp],
latestKeyCache = true

对于每个测点，第一次写入时会确定其 IOTANY 列的类型，如果之后对该测点写入不同的类型，会报错。

在非 KEEP_LAST 的情况下，可以使用 update 语句修改测点 IOTANY 列的类型，但是 where 条件里只能包含firstSortKey，这样能保证每次更新所有该测点的记录，以确保同一个测点的 IOTANY 列的类型一直统一。

// 测点[1, `loc1`] 第一次写入
pt = loadTable("dfs://db", "pt")
t = table([1] as deviceId,
  [`loc1] as location,
  [now()] as timestamp,
  [int(233)] as value)
pt.append!(t) 

// 测点[1, `loc1`] 第二次写入，value类型为double
t = table([1] as deviceId,
  [`loc1] as location,
  [now()] as timestamp,
  [double(233)] as value)
pt.append!(t) 

// 修改测点类型为double
// 修改失败，where条件只能包含firstSortKey
update pt set value=double(233) where deviceId = 1 and location=`loc1 and timestamp >= 2017.08.07 
// 修改成功
update pt set value=double(233) where deviceId = 1 and location=`loc1 

此时再次执行第二次写入的脚本，结果为写入成功。

在非 KEEP_LAST 的设置情况下，支持调用 udpate 时变更某个点位 IOTANY 列的类型。

在写入示例的最后也涉及到了更新操作，此处聚焦后再次讲解。

t = table([`dev1] as deviceId,
  [now()] as timestamp,
  [`loc1] as location,
  [int(233)] as value)
pt.append!(t) //该测点的类型为INT
update pt set value = 2.33 where deviceId = `dev1 and location = `loc1
// 将测点值更新为double
update pt set value = 2.33 where deviceId = `dev1
// ok
update pt set value = 2.33 where deviceId = `dev1 and location = `loc1 and timestamp > xxx
// 不行，一次需要更新该点位所有的记录

前文介绍，最新值查询的优化有最新值缓存表和最新值查询优化两种方式。在满足不同条件时会用到不同的优化策略。本例为方便展示，将创建一个新的点位管理表。

dbName = "dfs://test";
if(existsDatabase(dbName)){
	dropDatabase(dbName)
}
// 创建数据库，存储引擎为IOTDB
create database "dfs://test" 
partitioned by HASH([LONG,1]),HASH([DATE, 1]),
engine = 'IOTDB'
dummy = table(1000000:0, `date`deviceId`key1`val_any, [DATE, LONG, INT, DOUBLE])
// 创建点位表，其中 firstSortKey 为 deviceId 和 key1，lastSortKey(时间戳列)为 date
create table "dfs://test"."pt1"(
        date DATE,
        deviceId LONG,
        key1 INT,
        val_any DOUBLE
)
partitioned by deviceId, date
sortColumns = [`deviceId,`key1,`date]
keepDuplicates = ALL
latestKeyCache = true
pt1 = loadTable(dbName, "pt1")
// 导入测试数据
n = 200000
date = rand(2012.01.01..2012.12.31, n)
key1 = take(200199..(200199+n), n)
deviceId = take(200..(200+n), n)
val = rand(double(n), n)
t = table(date, deviceId, key1, val)
pt1.append!(t)

select [HINT_EXPLAIN] * from pt1 where deviceId in [200, 201, 202]
context by deviceId, key1 csort date
limit -1
// HINT_EXPLAIN 输出
{
    .......
    "lastQuery": {
        "optimizationMethod": LatestKeyCache
    },
    ......
}

select [HINT_EXPLAIN] * from pt1 where deviceId = 200 and key1 = 200199 
and date > 2010.12.31
context by deviceId, key1 csort date
limit -1
// HINT_EXPLAIN 输出
{
    .......
    "lastQuery": {
        "optimizationMethod": LatestKeyQuery
    },
    ......
}

select [HINT_EXPLAIN] * from pt1 where deviceId = 200
and val_any = 123
context by deviceId, key1 csort date
limit -1
// HINT_EXPLAIN 输出
{
    .......
    "lastQuery": {
        "optimizationMethod": None
    },
    ......
}

相关文档：

• create
• createPartitionedTable
• schema
• getSessionMemoryStat
• 参数配置