You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tsfile.apache.org by co...@apache.org on 2026/09/02 07:51:18 UTC
(tsfile) 01/01: feat(sqlite): add experimental TsFile hybrid extension
This is an automated email from the ASF dual-hosted git repository.
ColinLeeo pushed a commit to branch experiment-sqlite
in repository https://gitbox.apache.org/repos/asf/tsfile.git
commit 9871a42e38515a0a548a5e035f67231f15be1770
Author: colin <sh...@163.com>
AuthorDate: Wed Sep 2 15:30:11 2026 +0800
feat(sqlite): add experimental TsFile hybrid extension
---
cpp/CMakeLists.txt | 19 +
cpp/src/CMakeLists.txt | 4 +
cpp/src/sqlite/CMakeLists.txt | 44 +
cpp/src/sqlite/README.md | 74 ++
cpp/src/sqlite/TECHNICAL_GUIDE.md | 443 +++++++++
cpp/src/sqlite/USER_GUIDE.md | 390 ++++++++
cpp/src/sqlite/tsfile_sqlite.cc | 1692 +++++++++++++++++++++++++++++++++
cpp/test/CMakeLists.txt | 27 +-
cpp/test/sqlite/tsfile_sqlite_test.cc | 186 ++++
9 files changed, 2878 insertions(+), 1 deletion(-)
diff --git a/cpp/CMakeLists.txt b/cpp/CMakeLists.txt
index 7ff07a786..ad586fc98 100755
--- a/cpp/CMakeLists.txt
+++ b/cpp/CMakeLists.txt
@@ -229,6 +229,25 @@ endif ()
option(BUILD_TOOLS "Build the tsfile command-line tools" ON)
message("cmake using: BUILD_TOOLS=${BUILD_TOOLS}")
+option(BUILD_SQLITE_EXTENSION
+ "Build the experimental SQLite + TsFile loadable extension"
+ OFF)
+message("cmake using: BUILD_SQLITE_EXTENSION=${BUILD_SQLITE_EXTENSION}")
+
+if (BUILD_SQLITE_EXTENSION AND WIN32)
+ message(FATAL_ERROR
+ "BUILD_SQLITE_EXTENSION currently supports Linux and macOS only")
+endif ()
+
+if (BUILD_SQLITE_EXTENSION AND NOT TSFILE_BUILD_SHARED)
+ message(FATAL_ERROR
+ "BUILD_SQLITE_EXTENSION requires TSFILE_BUILD_SHARED=ON")
+endif ()
+
+if (BUILD_SQLITE_EXTENSION)
+ find_package(SQLite3 3.31 REQUIRED)
+endif ()
+
option(ENABLE_ANTLR4 "Enable ANTLR4 runtime" ON)
message("cmake using: ENABLE_ANTLR4=${ENABLE_ANTLR4}")
diff --git a/cpp/src/CMakeLists.txt b/cpp/src/CMakeLists.txt
index 2ab2dd5d6..fa421dfa2 100644
--- a/cpp/src/CMakeLists.txt
+++ b/cpp/src/CMakeLists.txt
@@ -157,3 +157,7 @@ install(TARGETS tsfile
RUNTIME DESTINATION ${LIBRARY_OUTPUT_PATH}
LIBRARY DESTINATION ${LIBRARY_OUTPUT_PATH}
ARCHIVE DESTINATION ${LIBRARY_OUTPUT_PATH})
+
+if (BUILD_SQLITE_EXTENSION)
+ add_subdirectory(sqlite)
+endif ()
diff --git a/cpp/src/sqlite/CMakeLists.txt b/cpp/src/sqlite/CMakeLists.txt
new file mode 100644
index 000000000..b8c6c16fb
--- /dev/null
+++ b/cpp/src/sqlite/CMakeLists.txt
@@ -0,0 +1,44 @@
+#[[
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ https://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+]]
+
+find_package(SQLite3 3.31 REQUIRED)
+
+add_library(tsfile_sqlite MODULE tsfile_sqlite.cc)
+target_include_directories(tsfile_sqlite PRIVATE
+ ${CMAKE_SOURCE_DIR}/src
+ ${SQLite3_INCLUDE_DIRS})
+target_link_libraries(tsfile_sqlite PRIVATE tsfile)
+set_target_properties(tsfile_sqlite PROPERTIES
+ PREFIX ""
+ OUTPUT_NAME "tsfile_sqlite"
+ LIBRARY_OUTPUT_DIRECTORY ${LIBRARY_OUTPUT_PATH})
+
+if (APPLE)
+ set_target_properties(tsfile_sqlite PROPERTIES
+ SUFFIX ".dylib"
+ BUILD_RPATH "@loader_path"
+ INSTALL_RPATH "@loader_path")
+else ()
+ set_target_properties(tsfile_sqlite PROPERTIES
+ BUILD_RPATH "$ORIGIN"
+ INSTALL_RPATH "$ORIGIN")
+endif ()
+
+install(TARGETS tsfile_sqlite
+ LIBRARY DESTINATION ${LIBRARY_OUTPUT_PATH})
diff --git a/cpp/src/sqlite/README.md b/cpp/src/sqlite/README.md
new file mode 100644
index 000000000..81c043c0e
--- /dev/null
+++ b/cpp/src/sqlite/README.md
@@ -0,0 +1,74 @@
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+-->
+
+# SQLite + TsFile extension
+
+`tsfile_sqlite` is an experimental SQLite loadable extension that keeps recent
+rows in SQLite and seals older rows into immutable TsFile table-model segments.
+
+Documentation:
+
+- [User guide](USER_GUIDE.md): build, load, configure, query, seal, deploy, and
+ operate the extension.
+- [Technical guide](TECHNICAL_GUIDE.md): architecture, virtual-table callbacks,
+ storage layout, query planning, and transaction/crash consistency.
+
+Build it with:
+
+```bash
+cmake -S cpp -B cpp/build/sqlite \
+ -DBUILD_SQLITE_EXTENSION=ON -DTSFILE_BUILD_SHARED=ON -DBUILD_TEST=ON
+cmake --build cpp/build/sqlite --target tsfile_sqlite
+```
+
+The extension is emitted next to `libtsfile` in the build `lib` directory.
+Load it and create a hybrid table with:
+
+```sql
+.load ./tsfile_sqlite
+
+CREATE VIRTUAL TABLE sensor USING tsfile_hybrid(
+ directory='/absolute/path/to/segments',
+ timestamp_precision='ms',
+ column='time:TIMESTAMP:TIME',
+ column='device:STRING:TAG',
+ column='temperature:DOUBLE:FIELD'
+);
+```
+
+Seal the half-open historical interval ending at `cutoff` with:
+
+```sql
+INSERT INTO sensor(_tsfile_command, _tsfile_cutoff)
+VALUES ('seal', 1700000000000);
+```
+
+The generated `<table>_data`, `<table>_segments`, and `<table>_config` tables
+are SQLite shadow tables. A seal is synchronous and participates in the
+surrounding SQLite transaction. Data at or above the watermark remains mutable;
+attempts to insert or modify older data return `SQLITE_CONSTRAINT`.
+
+The supported field types are `BOOLEAN`, `INT32`, `INT64`, `FLOAT`, `DOUBLE`,
+`TEXT`, `STRING`, `BLOB`, `DATE`, and `TIMESTAMP`. The first column must be a
+`TIMESTAMP:TIME`, TAG columns must be non-null `STRING`, and the TAG columns
+together with time form the unique key. The directory is created on table
+creation and must be dedicated to that logical table; dropping the virtual
+table does not delete already exported segment files.
diff --git a/cpp/src/sqlite/TECHNICAL_GUIDE.md b/cpp/src/sqlite/TECHNICAL_GUIDE.md
new file mode 100644
index 000000000..e746ca416
--- /dev/null
+++ b/cpp/src/sqlite/TECHNICAL_GUIDE.md
@@ -0,0 +1,443 @@
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+-->
+
+# tsfile_sqlite 技术手册
+
+本文说明 SQLite + TsFile Hybrid MVP 的设计边界、内部数据结构、SQLite
+virtual table 回调、冷热读写路径以及事务和崩溃一致性。
+
+实现入口是 `tsfile_sqlite.cc`，对 SQLite 注册的模块名为 `tsfile_hybrid`。
+
+## 1. 目标和非目标
+
+### 1.1 目标
+
+- 让应用通过一张 SQLite 逻辑表查询热数据和 TsFile 历史数据；
+- 让近期可变数据保留完整 SQLite 事务和 CRUD 语义；
+- 通过显式时间水位把历史数据导出为标准、不可变 TsFile；
+- 使文件生成、SQLite manifest 更新和热数据删除具备事务一致性；
+- 复用 libtsfile 的 table model、Tablet writer 和 C++ ResultSet reader。
+
+### 1.2 非目标
+
+- 不修改 SQLite pager、B-tree、WAL、VFS 或 SQL parser；
+- 不让 TsFile 成为普通 SQLite database file；
+- 不提供历史 correction/tombstone；
+- 不实现自动 seal、compaction、分布式 manifest 或在线备份协议；
+- 不在 MVP 中承诺跨数据源的物理输出顺序。
+
+因此当前实现是 SQLite loadable extension，而不是 SQLite 内核 fork。若应用不
+希望执行 `.load`，未来可以把相同模块静态链接到定制 SQLite，并通过
+`sqlite3_auto_extension()` 注册；这不会改变本文的数据模型。
+
+## 2. 总体架构
+
+```text
+ SQLite connection
+ |
+ SQL / transaction / planner
+ |
+ tsfile_hybrid virtual table
+ / \
+ SQLite shadow hot table TsFile segment files
+ CRUD, WAL, constraints immutable table model
+ \ /
+ merged virtual cursor
+```
+
+主要组件职责：
+
+| 组件 | 职责 |
+| --- | --- |
+| SQLite | SQL 执行、事务、WAL/journal、锁、残余谓词、排序和聚合 |
+| `tsfile_sqlite` | virtual table 适配、schema 校验、冷热路由、seal、manifest |
+| `libtsfile` | TsFile table model 编码、文件 footer、查询和 ResultSet |
+
+扩展使用 `sqlite3ext.h` 和 `SQLITE_EXTENSION_INIT1/2` 获取宿主 SQLite API，
+导出标准入口 `sqlite3_extension_init`。入口首次执行时初始化 libtsfile，然后
+通过 `sqlite3_create_module_v2()` 注册 `tsfile_hybrid`。
+
+## 3. 构建集成
+
+根 CMake 增加默认关闭的选项：
+
+```text
+BUILD_SQLITE_EXTENSION=OFF
+```
+
+启用时：
+
+- 拒绝 Windows；
+- 要求 `TSFILE_BUILD_SHARED=ON`；
+- 通过 `find_package(SQLite3 3.31 REQUIRED)` 校验 SQLite；
+- 构建 `MODULE` library `tsfile_sqlite`；
+- 链接同一构建中的 `tsfile` target；
+- Linux 使用 `$ORIGIN`，macOS 使用 `@loader_path` 查找同目录 `libtsfile`；
+- 不给模块添加 Unix 默认的 `lib` 前缀。
+
+测试 target `TsFile_Sqlite_Test` 动态加载刚构建的扩展，避免测试到系统中其他
+版本的插件。
+
+## 4. Virtual table schema
+
+扩展解析重复的 `column=name:TYPE:CATEGORY` 参数，并向 SQLite 声明：
+
+```sql
+CREATE TABLE x(
+ <public columns>,
+ _tsfile_command TEXT HIDDEN,
+ _tsfile_cutoff INTEGER HIDDEN
+);
+```
+
+两个隐藏列只用于把控制命令送入 `xUpdate`，普通 `SELECT *` 不会返回它们。
+
+初始化阶段验证：
+
+- 目录是绝对路径；
+- 时间精度属于 `ms/us/ns`；
+- 第一列且唯一 TIME 列是 `TIMESTAMP:TIME`；
+- 至少一个 `STRING:TAG`；
+- 列名非空、大小写不重复且不占用隐藏列名称。
+
+模块调用：
+
+```c
+sqlite3_vtab_config(db, SQLITE_VTAB_DIRECTONLY);
+sqlite3_vtab_config(db, SQLITE_VTAB_CONSTRAINT_SUPPORT, 1);
+```
+
+`DIRECTONLY` 限制从 view、trigger 等间接 schema 对象调用虚拟表，减少恶意
+数据库文件在连接打开时触发外部文件访问的风险。约束支持标志允许 `xUpdate`
+向 SQLite 返回精确的约束错误。
+
+## 5. Shadow tables
+
+假设逻辑表名为 `sensor`，`xCreate` 在逻辑表所属 schema 中创建：
+
+### 5.1 `sensor_data`
+
+```sql
+CREATE TABLE sensor_data(
+ time INTEGER NOT NULL,
+ device TEXT NOT NULL,
+ temperature REAL,
+ UNIQUE(device, time)
+);
+```
+
+真实列会根据用户 schema 展开。所有 TAG 加 TIME 形成复合唯一键。它是唯一的
+可变数据存储，rowid 由 SQLite 分配。
+
+### 5.2 `sensor_segments`
+
+```sql
+CREATE TABLE sensor_segments(
+ path TEXT PRIMARY KEY,
+ cutoff INTEGER NOT NULL,
+ row_count INTEGER NOT NULL
+);
+```
+
+这是冷段 manifest。`path` 是绝对文件路径，`cutoff` 是生成该段时的新水位，
+`row_count` 用于诊断和后续优化。
+
+### 5.3 `sensor_config`
+
+```sql
+CREATE TABLE sensor_config(
+ id INTEGER PRIMARY KEY CHECK(id = 1),
+ watermark INTEGER NOT NULL,
+ precision TEXT NOT NULL,
+ directory TEXT NOT NULL,
+ schema TEXT NOT NULL
+);
+```
+
+初始 watermark 是 `INT64_MIN`。schema 以列名、TsFile 类型枚举和类别枚举组成
+签名。`xConnect` 重新打开逻辑表时验证 precision、directory 和 schema 签名，
+不一致返回 `SQLITE_CORRUPT`。
+
+`xShadowName` 将 `_data`、`_segments` 和 `_config` 后缀报告给 SQLite。名称通过
+schema 限定和 identifier quoting 生成，因此 attached database 的 shadow
+table 不会错误创建到 `main`。
+
+`xDestroy` 删除三个 shadow table，但刻意不删除 TsFile 文件，避免 `DROP TABLE`
+隐式执行不可恢复的外部文件删除。
+
+## 6. 类型转换
+
+扩展内部用 `Value` 保存精确类型、NULL 标志、数值或带长度的字节串。
+
+写入方向：
+
+```text
+sqlite3_value -> Value -> SQLite hot binding / Tablet value
+```
+
+读取方向：
+
+```text
+SQLite column / C++ ResultSet -> Value -> sqlite3_result_*
+```
+
+TEXT、STRING 和 BLOB 都使用显式指针加长度复制，不把内容当作零结尾 C 字符串，
+因此空字符串、嵌入 `\0` 的 BLOB 和 NULL 可以区分。冷读直接使用 C++
+`ResultSet`，不依赖当前 C wrapper 的字符串接口。
+
+SQLite 是动态类型系统，扩展在 `xUpdate` 中做严格运行时检查。INT32/DATE 会
+额外检查 int32 上下界。BOOLEAN 接受 INTEGER，并归一化为 0 或 1。
+
+## 7. 热区 DML 路径
+
+所有 `INSERT/UPDATE/DELETE` 通过 `xUpdate` 路由：
+
+### INSERT
+
+1. 将 SQLite values 转成 schema 指定的 `Value`；
+2. 验证 TIME 和 TAG 非 NULL；
+3. 验证 `time >= watermark`；
+4. 参数化插入 `_data`；
+5. 由 shadow table 的 UNIQUE 约束检查 `(TAG..., TIME)`。
+
+### UPDATE
+
+1. 冷行的合成 rowid 为负，直接返回 `SQLITE_CONSTRAINT`；
+2. 不允许改变 SQLite rowid；
+3. 校验完整的新行值和新时间；
+4. 按正 rowid 更新 `_data`。
+
+### DELETE
+
+- 负 rowid 返回 `SQLITE_CONSTRAINT`；
+- 正 rowid 从 `_data` 删除。
+
+因为 DML 最终是同一连接上的 SQLite shadow table DML，所以 journal/WAL、事务
+隔离、写锁和回滚仍由 SQLite 提供。
+
+## 8. 查询规划和执行
+
+### 8.1 `xBestIndex`
+
+当前接受以下可用约束：
+
+- TIME 的 `>`、`>=`、`<`、`<=`；
+- 使用 BINARY collation 的 TAG `=`。
+
+对于接受的约束，`argvIndex` 被编码进 `idxStr`。`omit` 保持为 0，要求 SQLite
+对返回行再次执行原 SQL 条件。这样即使 libtsfile 与 SQLite 在边界、类型或
+collation 上存在差异，也不会产生错误结果。
+
+`sqlite3_index_info.colUsed` 用于计算投影列。TIME 和所有下推约束涉及的列会被
+强制加入内部投影，以便执行边界检查和 SQLite 残余检查。
+
+MVP 不设置 `orderByConsumed`，也不下推跨冷热来源的 `LIMIT/OFFSET`。
+
+### 8.2 `xFilter`
+
+`xFilter` 解码约束和投影，然后依次调用：
+
+```text
+read_hot() -> read_cold() -> HybridCursor.rows
+```
+
+当前 cursor 会物化本次查询的所有候选行。`xNext/xEof/xColumn/xRowid` 在该
+内存数组上实现 SQLite cursor 接口。这使 MVP 的合并逻辑简单，但不适合无界
+大结果集；后续应改造成 hot/cold 流式 cursor 和 k-way merge。
+
+### 8.3 热读取
+
+`read_hot` 动态生成参数化 SQL，只选择投影需要的列，并把接受的 TIME/TAG
+约束写入 `_data` 查询。结果按 hot rowid 读取，然后执行一次内部边界和 TAG
+字节比较；返回 SQLite 后仍有 SQLite 的最终残余检查。
+
+### 8.4 冷读取
+
+`read_cold` 从 manifest 按 path 遍历每个 TsFile：
+
+1. 为当前文件创建 `TsFileReader`；
+2. 将 TAG 等值条件组合成 `TagFilterBuilder` AND filter；
+3. 将开闭时间范围转换为 TsFileReader 的闭区间；
+4. 只请求投影所需的 value columns；
+5. 从 C++ ResultSet 恢复精确类型、NULL 和字节长度；
+6. 关闭 ResultSet 和 reader。
+
+当前 manifest 的 cutoff 尚未用于跳过不相交的段；reader 会对每个段应用时间
+范围。FIELD 谓词不进入 TsFile reader，由 SQLite 在结果返回后复核。
+
+### 8.5 Rowid
+
+热行直接暴露 SQLite `_data.rowid`，通常为正值。冷行按
+`(segment ordinal, row ordinal)` 编码为负 int64。负号同时充当不可变标志，
+使 `xUpdate` 可以拒绝冷行修改。
+
+冷 rowid 只对一次遍历有意义，不属于持久存储协议。
+
+## 9. Seal 写路径
+
+控制语句：
+
+```sql
+INSERT INTO sensor(_tsfile_command, _tsfile_cutoff)
+VALUES ('seal', ?);
+```
+
+在 `xUpdate` 中识别命令，cutoff 必须是 SQLite INTEGER 且不能小于当前
+watermark。
+
+### 9.1 段生成
+
+`write_segment()` 完成：
+
+1. 确保绝对目录存在；
+2. 使用表名、进程 ID 和单调计数器生成不冲突的 `.tmp`/`.tsfile` 路径；
+3. 以 `O_CREAT | O_EXCL` 创建临时文件；
+4. 从 `_data` 读取 `[watermark, cutoff)`；
+5. 按所有 TAG、TIME 排序；
+6. 每 1024 行填充一个 table-model `Tablet`；
+7. 通过 `TsFileTableWriter` 写入；
+8. 写入时间精度 property、flush、footer 并 fsync 文件。
+
+若区间没有行，writer 仍会被正确关闭，随后删除空临时文件。调用方仍更新
+watermark，但不插入 manifest 记录。
+
+### 9.2 SQLite 状态变更
+
+非空段写完临时文件后，`seal()` 在当前 SQLite 事务中：
+
+1. 向 `_segments` 插入最终文件路径、cutoff 和 row count；
+2. 从 `_data` 删除 `[watermark, cutoff)`；
+3. 更新 `_config.watermark`；
+4. 把文件记录到当前事务的 `pending` 数组。
+
+这些 shadow table 修改与调用 seal 的外层 SQL 事务相同，不单独提交。
+
+## 10. 事务回调和文件原子性
+
+SQLite 数据库事务不能直接回滚外部文件，因此扩展通过 virtual table transaction
+callbacks 协调两者。
+
+```text
+xBegin
+ |
+xUpdate(seal): write .tmp + update transactional shadow state
+ |
+xSync: validate .tmp -> atomic rename -> fsync directory
+ |
+SQLite commits database/WAL
+ |
+xCommit: forget pending state
+```
+
+### 10.1 `xSync`
+
+SQLite 准备提交时，扩展对每个 pending 文件执行：
+
+1. 使用独立 `TsFileReader` 重新打开临时文件，验证 footer 可读；
+2. 确认最终路径不存在；
+3. 在同一目录执行原子 `rename()`；
+4. `fsync()` 目录，持久化目录项。
+
+只有全部成功，SQLite 才继续提交数据库事务。
+
+### 10.2 `xRollback`
+
+回滚时删除本事务仍存在的 `.tmp`，也删除已经 rename 但数据库事务未提交的
+`.tsfile`，然后从 `_config` 重新加载 watermark。
+
+### 10.3 Savepoint
+
+`xSavepoint` 记录 pending 数组长度。`xRollbackTo` 删除 savepoint 之后创建的
+文件并重新加载事务可见的配置，`xRelease` 移除相应标记。
+
+### 10.4 崩溃窗口
+
+| 崩溃位置 | SQLite 恢复结果 | 文件结果与恢复方式 |
+| --- | --- | --- |
+| 写临时文件过程中 | 热数据和旧 manifest 保留 | 遗留 `.tmp`，下次 seal 清理 |
+| 临时文件完成、rename 前 | 热数据和旧 manifest 保留 | 遗留 `.tmp`，下次 seal 清理 |
+| rename 后、SQLite COMMIT 前 | 热数据和旧 manifest 保留 | 遗留未引用 `.tsfile`，下次 seal 清理 |
+| SQLite COMMIT 后 | 新 manifest、水位和热表删除持久化 | manifest 引用的 `.tsfile` 已完成 rename 和目录 fsync |
+
+`cleanup_orphans()` 在 seal 已取得 SQLite 写事务的上下文中运行。它读取 manifest，
+删除独占目录内未被引用且不属于当前 pending 集合的 `.tmp` 和 `.tsfile`。这也是
+为什么目录独占不是建议，而是正确性前提。
+
+## 11. 并发模型
+
+- 热 DML、manifest 和配置依赖 SQLite 的连接事务与写锁；
+- seal 是同步写操作，不会在后台线程与 SQLite 事务脱离运行；
+- 读事务看到由 SQLite 快照决定的 shadow state；
+- manifest 只在 SQLite 事务提交后对其他连接可见；
+- TsFile 发布使用同目录原子 rename，读取者不会观察到半写文件。
+
+当前设计不提供跨进程独立修改 TsFile 目录的协调机制。目录必须只由拥有 SQLite
+manifest 的扩展实例管理。
+
+## 12. 安全和运维边界
+
+- `DIRECTONLY` 限制间接调用，但加载 native extension 本身仍等价于加载本地
+ 代码，应用应只加载可信二进制；
+- 配置要求绝对路径，避免工作目录变化导致数据库重新打开到不同数据集；
+- SQL identifier 和字符串均经过 quoting，运行时值使用 prepared statement；
+- TsFile 文件权限当前为 `0644`，目录创建权限当前为 `0755`，最终仍受进程
+ `umask` 影响；
+- `DROP TABLE` 不清理外部文件；
+- SQLite backup API 不会自动包含 TsFile，备份协议必须覆盖数据库和目录。
+
+## 13. 测试覆盖
+
+`cpp/test/sqlite/tsfile_sqlite_test.cc` 当前覆盖：
+
+- 热数据 CRUD；
+- seal 的 commit 和 rollback；
+- watermark 对旧时间写入的约束；
+- INT32 越界；
+- 冷行 UPDATE/DELETE 拒绝；
+- NULL、空 TEXT 和包含零字节的 BLOB 往返；
+- attached database 的 shadow table 隔离。
+
+实现开发过程中还应持续补充：
+
+- 多连接 WAL 并发；
+- 多次 seal 和多 TAG 查询；
+- 独立 TsFileReader 验证 property/schema/row count；
+- write/close/fsync/rename/commit 各故障点注入；
+- 重启和孤儿文件清理；
+- SQLite 3.45.3 主版本与 3.31 最低版本矩阵；
+- Linux/macOS CI。
+
+## 14. 当前限制与后续方向
+
+当前 MVP 的主要技术债务：
+
+1. 把查询候选行全部物化到内存，应改为流式 merge cursor；
+2. manifest 仅保存 cutoff，尚未记录每段 min/max time、TAG 统计和校验信息；
+3. 没有跨段 compaction，段数增加后每次查询都要打开更多文件；
+4. 没有 correction/tombstone，无法修改冷历史；
+5. 没有后台 seal 调度和限流；
+6. 没有冷段删除与 retention policy；
+7. schema 演进需要新建逻辑表；
+8. 没有一体化在线备份、迁移和目录重定位工具；
+9. 尚未实现 Windows 文件同步和原子发布语义。
+
+若继续演进，建议优先处理流式查询、segment pruning、故障注入测试和一致性备份
+协议，再考虑自动 seal、compaction 和历史修正。
diff --git a/cpp/src/sqlite/USER_GUIDE.md b/cpp/src/sqlite/USER_GUIDE.md
new file mode 100644
index 000000000..4c5075efa
--- /dev/null
+++ b/cpp/src/sqlite/USER_GUIDE.md
@@ -0,0 +1,390 @@
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+-->
+
+# tsfile_sqlite 用户手册
+
+`tsfile_sqlite` 是一个实验性的 SQLite loadable extension。它提供
+`tsfile_hybrid` 虚拟表，让一张逻辑表同时使用两种物理存储：
+
+- 尚未封存的近期数据保存在 SQLite shadow table 中，支持事务和 CRUD；
+- 已封存的历史数据保存在不可变的 TsFile 段文件中；
+- 应用继续对同一张虚拟表执行 SQL，扩展自动合并冷热数据。
+
+它适合以追加为主、近期数据偶尔需要修正、历史数据可以冻结的时序场景。
+它不是 SQLite 通用表的替代存储引擎，也不会自动把已有 SQLite 表转换为
+TsFile。
+
+## 1. 环境要求
+
+- Linux 或 macOS；
+- SQLite 3.31 或更高版本；
+- 支持加载扩展的 SQLite 构建；
+- CMake 构建时启用共享版 `libtsfile`；
+- TsFile 目录必须使用绝对路径，并由一张逻辑表独占。
+
+当前 MVP 不支持 Windows。
+
+## 2. 构建
+
+在仓库根目录执行：
+
+```bash
+cmake -S cpp -B cpp/build/sqlite \
+ -DBUILD_SQLITE_EXTENSION=ON \
+ -DTSFILE_BUILD_SHARED=ON \
+ -DBUILD_TEST=ON
+
+cmake --build cpp/build/sqlite --target tsfile_sqlite -j
+```
+
+产物位于构建目录的 `lib` 子目录：
+
+- Linux：`cpp/build/sqlite/lib/tsfile_sqlite.so`
+- macOS：`cpp/build/sqlite/lib/tsfile_sqlite.dylib`
+
+扩展依赖同一次构建产生的 `libtsfile`。默认 RPATH 会从扩展所在目录寻找
+`libtsfile`，部署时建议把二者放在同一目录。
+
+运行扩展测试：
+
+```bash
+cmake --build cpp/build/sqlite --target TsFile_Sqlite_Test -j
+ctest --test-dir cpp/build/sqlite/test -R TsFileSqliteTest \
+ --output-on-failure
+```
+
+## 3. 加载扩展
+
+### 3.1 SQLite CLI
+
+```sql
+.load /absolute/path/to/tsfile_sqlite
+```
+
+SQLite CLI 通常会根据平台自动补全 `.so` 或 `.dylib` 后缀。也可以传入完整
+文件名。
+
+### 3.2 C/C++ 应用
+
+```c
+sqlite3_enable_load_extension(db, 1);
+
+char *error = NULL;
+int rc = sqlite3_load_extension(
+ db, "/absolute/path/to/tsfile_sqlite", NULL, &error);
+
+sqlite3_enable_load_extension(db, 0);
+```
+
+应用应在打开数据库连接后、访问 hybrid 表之前加载扩展。生产环境建议加载
+完成后立即关闭动态扩展加载能力。
+
+## 4. 创建逻辑表
+
+```sql
+CREATE VIRTUAL TABLE sensor USING tsfile_hybrid(
+ directory='/var/lib/example/sensor',
+ timestamp_precision='ms',
+ column='time:TIMESTAMP:TIME',
+ column='device:STRING:TAG',
+ column='region:STRING:TAG',
+ column='temperature:DOUBLE:FIELD',
+ column='status:STRING:FIELD',
+ column='payload:BLOB:FIELD'
+);
+```
+
+模块参数如下：
+
+| 参数 | 要求 |
+| --- | --- |
+| `directory` | 必填、绝对路径、由当前逻辑表独占 |
+| `timestamp_precision` | 必填，只能是 `ms`、`us` 或 `ns` |
+| `column` | 可重复，格式为 `名称:类型:类别` |
+
+列定义必须满足：
+
+- 恰好一个 `TIME` 列；
+- `TIME` 必须是第一列，类型必须为 `TIMESTAMP`；
+- 至少一个 `TAG` 列；
+- `TAG` 类型必须为 `STRING`，值不能为 `NULL`；
+- `FIELD` 可以为 `NULL`；
+- 列名大小写不敏感地唯一；
+- `_tsfile_command` 和 `_tsfile_cutoff` 是保留名称。
+
+所有 TAG 列与 TIME 列共同组成唯一键。例如上表的唯一键是：
+
+```text
+(device, region, time)
+```
+
+### 4.1 数据类型映射
+
+| TsFile 类型 | SQLite 表现 | 写入要求 |
+| --- | --- | --- |
+| `BOOLEAN` | INTEGER | 必须传 SQLite INTEGER；0 为假，非 0 为真 |
+| `INT32` | INTEGER | 必须在 int32 范围内 |
+| `INT64` | INTEGER | SQLite int64 |
+| `FLOAT` | REAL | INTEGER 或 REAL |
+| `DOUBLE` | REAL | INTEGER 或 REAL |
+| `TEXT` | TEXT | SQLite TEXT |
+| `STRING` | TEXT | SQLite TEXT |
+| `BLOB` | BLOB | SQLite BLOB，保留长度和二进制零字节 |
+| `DATE` | INTEGER | 必须在 int32 范围内 |
+| `TIMESTAMP` | INTEGER | SQLite int64 |
+
+扩展不会换算时间戳。`timestamp_precision` 仅声明整数时间戳的单位，并写入
+非空 TsFile 段的 `tsfile_sqlite.timestamp_precision` property。
+
+### 4.2 创建后的固定配置
+
+schema、目录和时间精度会记录在配置 shadow table 中。数据库重新打开时，
+扩展会验证这些信息是否与 `CREATE VIRTUAL TABLE` 中保存的参数一致。
+
+当前版本不支持修改 schema、目录或时间精度。需要变更时，应创建一张新的
+逻辑表并迁移数据。
+
+## 5. 写入和修改热数据
+
+普通 DML 的用法与 SQLite 表一致：
+
+```sql
+INSERT INTO sensor(time, device, region, temperature, status)
+VALUES (1700000000000, 'device-1', 'cn-east', 21.5, 'ok');
+
+UPDATE sensor
+SET temperature = 22.0
+WHERE device = 'device-1'
+ AND region = 'cn-east'
+ AND time = 1700000000000;
+
+DELETE FROM sensor
+WHERE device = 'device-1'
+ AND region = 'cn-east'
+ AND time = 1700000000000;
+```
+
+热数据实际写入 `<虚拟表名>_data` shadow table，因此自动使用 SQLite 的
+rollback journal/WAL、锁、唯一约束、事务和 savepoint。
+
+```sql
+BEGIN;
+
+INSERT INTO sensor(time, device, region, temperature)
+VALUES (1700000001000, 'device-1', 'cn-east', 22.1);
+
+UPDATE sensor
+SET status = 'checked'
+WHERE device = 'device-1' AND time = 1700000001000;
+
+COMMIT;
+```
+
+## 6. 查询冷热数据
+
+无论数据位于 SQLite 还是 TsFile，都查询同一张虚拟表：
+
+```sql
+SELECT time, device, temperature, status
+FROM sensor
+WHERE device = 'device-1'
+ AND time >= 1700000000000
+ AND time < 1700086400000
+ORDER BY time;
+```
+
+常规 SQLite SQL 仍然可用，包括 FIELD 条件、表达式、聚合、排序和分页：
+
+```sql
+SELECT device, avg(temperature)
+FROM sensor
+WHERE time >= 1700000000000
+ AND temperature IS NOT NULL
+GROUP BY device;
+```
+
+为了得到更好的 TsFile 扫描效率，查询应尽量包含：
+
+- 整数时间范围；
+- 使用 `BINARY` collation 的 TAG 等值条件；
+- 只选择需要的列。
+
+扩展会把这些条件和投影下推到冷热读取路径。FIELD 条件、排序、聚合、
+`LIMIT/OFFSET` 由 SQLite 在合并结果上处理。扩展不会声称原始输出已经满足
+`ORDER BY`，因此需要稳定顺序时必须显式写出 `ORDER BY`。
+
+所有已下推的约束仍由 SQLite 二次检查，以保证 SQL 结果正确。
+
+## 7. 封存历史数据
+
+使用两个隐藏列发送 `seal` 命令：
+
+```sql
+INSERT INTO sensor(_tsfile_command, _tsfile_cutoff)
+VALUES ('seal', 1700086400000);
+```
+
+`cutoff` 是不包含的上界。上述操作封存：
+
+```text
+旧 watermark <= time < 1700086400000
+```
+
+其中 `time == 1700086400000` 的行仍在热区，可以继续修改。
+
+一次非空 seal 会：
+
+1. 按全部 TAG、TIME 排序读取待封存热数据；
+2. 使用 Tablet 批量写入临时 TsFile；
+3. 校验并将临时文件原子改名为 `.tsfile`；
+4. 在 manifest 中登记文件；
+5. 从热表删除已封存行；
+6. 将 watermark 推进到 cutoff。
+
+如果区间内没有数据，不生成 TsFile，但仍会推进 watermark。cutoff 不能小于
+当前 watermark。
+
+seal 是同步写操作，在完成期间会占用 SQLite 写事务。可以显式把它放入事务：
+
+```sql
+BEGIN IMMEDIATE;
+
+INSERT INTO sensor(_tsfile_command, _tsfile_cutoff)
+VALUES ('seal', 1700086400000);
+
+COMMIT;
+```
+
+如果事务回滚，manifest、watermark 和热数据删除都会回滚，扩展也会删除本次
+事务产生的临时文件或已经改名的段文件。
+
+## 8. Watermark 和冷数据不可变性
+
+watermark 把逻辑时间轴分成两部分：
+
+```text
+time < watermark 冷区：不可修改
+time >= watermark 热区：允许 INSERT/UPDATE/DELETE
+```
+
+以下操作会返回约束错误：
+
+- 插入 `time < watermark` 的行；
+- 把热行的时间更新到 watermark 之前；
+- 更新或删除已经位于 TsFile 的冷行；
+- 写入重复的 `(所有 TAG, TIME)` 唯一键。
+
+当前版本没有 correction 或 tombstone。如果业务必须修正历史数据，需要重建
+逻辑表或在业务层保留单独的修正数据。
+
+## 9. 内部状态与诊断
+
+每张名为 `sensor` 的 hybrid 表拥有三个 SQLite shadow table：
+
+| 表 | 内容 |
+| --- | --- |
+| `sensor_data` | 可变热数据 |
+| `sensor_segments` | TsFile 路径、cutoff 和行数 manifest |
+| `sensor_config` | watermark、精度、目录和 schema 签名 |
+
+可以只读查看它们进行诊断：
+
+```sql
+SELECT watermark, precision, directory FROM sensor_config;
+
+SELECT path, cutoff, row_count
+FROM sensor_segments
+ORDER BY cutoff;
+
+SELECT count(*) AS hot_rows FROM sensor_data;
+```
+
+不要直接修改 shadow table。绕过虚拟表写入会破坏 watermark、manifest 和文件
+之间的一致性。
+
+热行使用 SQLite 的正 rowid；冷行使用扩展生成的负 rowid。冷 rowid 是内部
+实现标识，不应作为跨查询或跨版本稳定的业务主键。
+
+## 10. 数据目录和生命周期
+
+每张逻辑表必须使用独占目录。扩展会在 seal 前清理目录中未被 manifest 引用的
+`.tmp` 和 `.tsfile` 文件，用来恢复进程崩溃后遗留的孤儿文件。因此不要把手工
+创建的 TsFile、其他表的段文件或任何同后缀文件放进该目录。
+
+执行：
+
+```sql
+DROP TABLE sensor;
+```
+
+会删除虚拟表及其三个 shadow table，但不会删除已经导出的 TsFile。删除或归档
+这些文件需要由运维流程显式完成。
+
+## 11. 备份与恢复
+
+完整数据由两部分共同组成：
+
+1. SQLite 数据库文件及其 WAL/journal；
+2. manifest 引用的 TsFile 目录。
+
+只备份 SQLite 文件会丢失冷数据，只备份 TsFile 目录会丢失热数据和 manifest。
+备份工具必须同时捕获两部分的一致视图。MVP 尚未提供在线快照命令，推荐在停止
+写入后备份，或者在应用层协调 SQLite checkpoint/事务与目录快照。
+
+恢复时必须保持配置中记录的绝对目录可用；如果恢复到不同路径，应创建新的逻辑
+表并执行受控迁移，而不是手工修改 `sensor_config`。
+
+## 12. 常见问题
+
+### 加载时报 `not authorized` 或扩展加载被禁用
+
+确认 SQLite 构建允许 loadable extension，并在应用连接上调用
+`sqlite3_enable_load_extension()`。CLI 使用 `.load` 即可。
+
+### 加载时报找不到 `libtsfile`
+
+把当前构建对应的 `libtsfile` 与 `tsfile_sqlite` 放到同一目录，避免混用不同
+版本的库。必要时检查平台动态链接器的搜索路径。
+
+### 创建表时报目录错误
+
+`directory` 必须是绝对路径，父目录必须可创建或可写。连接已有数据库时，该
+目录必须仍然存在。
+
+### seal 后无法更新某些行
+
+这是 watermark 的预期行为。任何 `time < watermark` 的数据已经进入不可变
+冷区。
+
+### 查询没有固定顺序
+
+虚拟表会合并多个来源，但不承诺自然顺序。需要顺序时使用显式 `ORDER BY`。
+
+## 13. MVP 限制
+
+- seal 只能显式、同步执行；
+- 冷数据不可更新或删除；
+- 不支持后台自动封存、compaction 和冷段删除；
+- 不支持原地 schema 演进；
+- 查询游标当前会在内存中汇集冷热结果；
+- 不跨冷热来源下推排序和 `LIMIT/OFFSET`；
+- Windows 尚未支持；
+- 这仍是实验性扩展，接口和内部格式可能继续演进。
diff --git a/cpp/src/sqlite/tsfile_sqlite.cc b/cpp/src/sqlite/tsfile_sqlite.cc
new file mode 100644
index 000000000..8dd5f0c65
--- /dev/null
+++ b/cpp/src/sqlite/tsfile_sqlite.cc
@@ -0,0 +1,1692 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "sqlite3ext.h"
+SQLITE_EXTENSION_INIT1
+
+#include <dirent.h>
+#include <errno.h>
+#include <fcntl.h>
+#include <sys/stat.h>
+#include <sys/types.h>
+#include <unistd.h>
+
+#include <algorithm>
+#include <cctype>
+#include <cstdio>
+#include <cstdlib>
+#include <cstring>
+#include <limits>
+#include <memory>
+#include <sstream>
+#include <string>
+#include <utility>
+#include <vector>
+
+#ifndef O_DIRECTORY
+#define O_DIRECTORY 0
+#endif
+
+#include "common/schema.h"
+#include "common/tablet.h"
+#include "common/tsfile_common.h"
+#include "file/write_file.h"
+#include "reader/filter/tag_filter.h"
+#include "reader/result_set.h"
+#include "reader/tsfile_reader.h"
+#include "utils/db_utils.h"
+#include "writer/tsfile_table_writer.h"
+
+namespace {
+
+using common::ColumnCategory;
+using common::ColumnSchema;
+using common::TSDataType;
+using storage::ResultSet;
+using storage::TableSchema;
+using storage::Tablet;
+using storage::TsFileReader;
+using storage::TsFileTableWriter;
+using storage::WriteFile;
+
+const int64_t kNoWatermark = std::numeric_limits<int64_t>::min();
+
+struct Column {
+ std::string name;
+ TSDataType type;
+ ColumnCategory category;
+};
+
+struct Value {
+ TSDataType type = common::NULL_TYPE;
+ bool is_null = true;
+ bool b = false;
+ int32_t i32 = 0;
+ int64_t i64 = 0;
+ float f = 0;
+ double d = 0;
+ std::string bytes;
+};
+
+struct Row {
+ sqlite3_int64 rowid = 0;
+ std::vector<Value> values;
+};
+
+struct PendingFile {
+ std::string temporary;
+ std::string final_path;
+ bool renamed = false;
+};
+
+struct HybridTable;
+
+struct HybridCursor : sqlite3_vtab_cursor {
+ HybridTable* table = nullptr;
+ std::vector<Row> rows;
+ size_t pos = 0;
+};
+
+struct HybridTable : sqlite3_vtab {
+ sqlite3* db = nullptr;
+ std::string db_name;
+ std::string table_name;
+ std::string directory;
+ std::string precision;
+ std::vector<Column> columns;
+ int time_index = -1;
+ std::vector<int> tag_indexes;
+ int64_t watermark = kNoWatermark;
+ std::vector<PendingFile> pending;
+ std::vector<size_t> savepoint_marks;
+ uint64_t file_counter = 0;
+};
+
+struct ConstraintSpec {
+ int column;
+ int op;
+ int argv_index;
+ ConstraintSpec(int column_value, int op_value, int argv_value)
+ : column(column_value), op(op_value), argv_index(argv_value) {}
+};
+
+std::string lower(std::string value) {
+ for (char& c : value)
+ c = static_cast<char>(std::tolower(static_cast<unsigned char>(c)));
+ return value;
+}
+
+std::string quote_id(const std::string& value) {
+ std::string out = "\"";
+ for (char c : value) {
+ if (c == '"')
+ out += "\"\"";
+ else
+ out += c;
+ }
+ out += '"';
+ return out;
+}
+
+std::string quote_sql(const std::string& value) {
+ std::string out = "'";
+ for (char c : value) {
+ if (c == '\'')
+ out += "''";
+ else
+ out += c;
+ }
+ out += '\'';
+ return out;
+}
+
+void copy_bytes(std::string& destination, const void* data, int length) {
+ destination.clear();
+ if (data != nullptr && length > 0)
+ destination.assign(static_cast<const char*>(data),
+ static_cast<size_t>(length));
+}
+
+void set_error(sqlite3_vtab* vtab, const std::string& message) {
+ if (vtab == nullptr) return;
+ if (vtab->zErrMsg != nullptr) sqlite3_free(vtab->zErrMsg);
+ vtab->zErrMsg = sqlite3_mprintf("%s", message.c_str());
+}
+
+int exec_sql(sqlite3* db, const std::string& sql,
+ sqlite3_vtab* vtab = nullptr) {
+ char* error = nullptr;
+ int rc = sqlite3_exec(db, sql.c_str(), nullptr, nullptr, &error);
+ if (rc != SQLITE_OK && vtab != nullptr) {
+ set_error(vtab, error == nullptr ? "SQLite error" : error);
+ }
+ sqlite3_free(error);
+ return rc;
+}
+
+bool parse_key_value(const char* arg, std::string& key, std::string& value) {
+ if (arg == nullptr) return false;
+ const char* equal = std::strchr(arg, '=');
+ if (equal == nullptr) return false;
+ key.assign(arg, static_cast<size_t>(equal - arg));
+ value.assign(equal + 1);
+ if (value.size() >= 2 && ((value.front() == '\'' && value.back() == '\'') ||
+ (value.front() == '"' && value.back() == '"'))) {
+ value = value.substr(1, value.size() - 2);
+ }
+ return true;
+}
+
+bool parse_column(const std::string& spec, Column& column, std::string& error) {
+ std::vector<std::string> parts;
+ size_t start = 0;
+ while (start <= spec.size()) {
+ size_t end = spec.find(':', start);
+ parts.push_back(spec.substr(
+ start, end == std::string::npos ? std::string::npos : end - start));
+ if (end == std::string::npos) break;
+ start = end + 1;
+ }
+ if (parts.size() != 3 || parts[0].empty()) {
+ error = "column must be name:TYPE:CATEGORY";
+ return false;
+ }
+ TSDataType type;
+ if (!common::parse_data_type_name(parts[1], type)) {
+ error = "unsupported TsFile column type: " + parts[1];
+ return false;
+ }
+ std::string category = lower(parts[2]);
+ ColumnCategory col_category;
+ if (category == "time")
+ col_category = ColumnCategory::TIME;
+ else if (category == "tag")
+ col_category = ColumnCategory::TAG;
+ else if (category == "field")
+ col_category = ColumnCategory::FIELD;
+ else {
+ error = "column category must be TIME, TAG, or FIELD";
+ return false;
+ }
+ column = {parts[0], type, col_category};
+ return true;
+}
+
+std::string sqlite_type(TSDataType type) {
+ switch (type) {
+ case common::BOOLEAN:
+ return "INTEGER";
+ case common::INT32:
+ case common::INT64:
+ case common::DATE:
+ case common::TIMESTAMP:
+ return "INTEGER";
+ case common::FLOAT:
+ case common::DOUBLE:
+ return "REAL";
+ case common::TEXT:
+ case common::STRING:
+ return "TEXT";
+ case common::BLOB:
+ return "BLOB";
+ default:
+ return "";
+ }
+}
+
+std::string shadow_name(const HybridTable* table, const char* suffix) {
+ return quote_id(table->db_name) + "." +
+ quote_id(table->table_name + suffix);
+}
+
+std::string schema_signature(const HybridTable* table) {
+ std::ostringstream out;
+ for (size_t i = 0; i < table->columns.size(); ++i) {
+ if (i) out << ';';
+ out << table->columns[i].name << ':'
+ << static_cast<int>(table->columns[i].type) << ':'
+ << static_cast<int>(table->columns[i].category);
+ }
+ return out.str();
+}
+
+std::string file_stem(const std::string& table_name) {
+ std::string stem = "tsfile";
+ for (char c : table_name) {
+ if (std::isalnum(static_cast<unsigned char>(c)) || c == '_' || c == '-')
+ stem += c;
+ else
+ stem += '_';
+ }
+ return stem;
+}
+
+bool ensure_directory(const std::string& path) {
+ if (path.empty() || path[0] != '/') return false;
+ size_t pos = 1;
+ while (pos <= path.size()) {
+ pos = path.find('/', pos);
+ std::string part =
+ path.substr(0, pos == std::string::npos ? path.size() : pos);
+ if (!part.empty() && mkdir(part.c_str(), 0755) != 0 && errno != EEXIST)
+ return false;
+ if (pos == std::string::npos) break;
+ ++pos;
+ }
+ struct stat st {};
+ return stat(path.c_str(), &st) == 0 && S_ISDIR(st.st_mode);
+}
+
+bool directory_exists(const std::string& path) {
+ struct stat st {};
+ return stat(path.c_str(), &st) == 0 && S_ISDIR(st.st_mode);
+}
+
+int sync_directory(const std::string& path) {
+ int fd = open(path.c_str(), O_RDONLY | O_DIRECTORY);
+ if (fd < 0) return SQLITE_IOERR_FSYNC;
+ int rc = fsync(fd) == 0 ? SQLITE_OK : SQLITE_IOERR_FSYNC;
+ close(fd);
+ return rc;
+}
+
+int close_writer_and_sync(WriteFile& write_file, TsFileTableWriter& writer) {
+ // Keep a duplicate descriptor because TsFileTableWriter::close() writes
+ // the footer and closes the original descriptor. This explicit fsync is
+ // independent of libtsfile's process-global sync_on_close setting.
+ int sync_fd = dup(write_file.get_fd());
+ if (sync_fd < 0) return SQLITE_IOERR_FSYNC;
+ int close_rc = writer.close();
+ int sync_rc = (close_rc == common::E_OK && fsync(sync_fd) == 0)
+ ? SQLITE_OK
+ : SQLITE_IOERR_FSYNC;
+ int close_fd_rc = close(sync_fd) == 0 ? SQLITE_OK : SQLITE_IOERR;
+ if (close_rc != common::E_OK) return SQLITE_IOERR;
+ if (sync_rc != SQLITE_OK) return sync_rc;
+ return close_fd_rc;
+}
+
+bool value_from_sqlite(sqlite3_value* value, const Column& column, Value& out,
+ std::string& error) {
+ int type = sqlite3_value_type(value);
+ if (type == SQLITE_NULL) {
+ out.type = column.type;
+ out.is_null = true;
+ return true;
+ }
+ out.type = column.type;
+ out.is_null = false;
+ switch (column.type) {
+ case common::BOOLEAN:
+ if (type != SQLITE_INTEGER) {
+ error = "BOOLEAN requires INTEGER";
+ return false;
+ }
+ out.b = sqlite3_value_int(value) != 0;
+ break;
+ case common::INT32:
+ case common::DATE:
+ if (type != SQLITE_INTEGER) {
+ error = "INT32/DATE requires INTEGER";
+ return false;
+ }
+ {
+ sqlite3_int64 number = sqlite3_value_int64(value);
+ if (number < std::numeric_limits<int32_t>::min() ||
+ number > std::numeric_limits<int32_t>::max()) {
+ error = "INT32/DATE value is out of range";
+ return false;
+ }
+ out.i32 = static_cast<int32_t>(number);
+ }
+ break;
+ case common::INT64:
+ case common::TIMESTAMP:
+ if (type != SQLITE_INTEGER) {
+ error = "INT64/TIMESTAMP requires INTEGER";
+ return false;
+ }
+ out.i64 = sqlite3_value_int64(value);
+ break;
+ case common::FLOAT:
+ if (type != SQLITE_INTEGER && type != SQLITE_FLOAT) {
+ error = "FLOAT requires numeric value";
+ return false;
+ }
+ out.f = static_cast<float>(sqlite3_value_double(value));
+ break;
+ case common::DOUBLE:
+ if (type != SQLITE_INTEGER && type != SQLITE_FLOAT) {
+ error = "DOUBLE requires numeric value";
+ return false;
+ }
+ out.d = sqlite3_value_double(value);
+ break;
+ case common::TEXT:
+ case common::STRING: {
+ if (type != SQLITE_TEXT) {
+ error = "TEXT/STRING requires TEXT";
+ return false;
+ }
+ const unsigned char* text = sqlite3_value_text(value);
+ copy_bytes(out.bytes, text, sqlite3_value_bytes(value));
+ break;
+ }
+ case common::BLOB: {
+ if (type != SQLITE_BLOB) {
+ error = "BLOB requires BLOB";
+ return false;
+ }
+ const void* blob = sqlite3_value_blob(value);
+ copy_bytes(out.bytes, blob, sqlite3_value_bytes(value));
+ break;
+ }
+ default:
+ error = "unsupported column type";
+ return false;
+ }
+ return true;
+}
+
+void bind_value(sqlite3_stmt* stmt, int index, const Value& value) {
+ if (value.is_null) {
+ sqlite3_bind_null(stmt, index);
+ return;
+ }
+ switch (value.type) {
+ case common::BOOLEAN:
+ sqlite3_bind_int(stmt, index, value.b ? 1 : 0);
+ break;
+ case common::INT32:
+ case common::DATE:
+ sqlite3_bind_int(stmt, index, value.i32);
+ break;
+ case common::INT64:
+ case common::TIMESTAMP:
+ sqlite3_bind_int64(stmt, index, value.i64);
+ break;
+ case common::FLOAT:
+ sqlite3_bind_double(stmt, index, value.f);
+ break;
+ case common::DOUBLE:
+ sqlite3_bind_double(stmt, index, value.d);
+ break;
+ case common::TEXT:
+ case common::STRING:
+ sqlite3_bind_text(stmt, index, value.bytes.data(),
+ static_cast<int>(value.bytes.size()),
+ SQLITE_TRANSIENT);
+ break;
+ case common::BLOB:
+ sqlite3_bind_blob(stmt, index, value.bytes.data(),
+ static_cast<int>(value.bytes.size()),
+ SQLITE_TRANSIENT);
+ break;
+ default:
+ sqlite3_bind_null(stmt, index);
+ break;
+ }
+}
+
+bool row_value_from_sqlite(sqlite3_stmt* stmt, int index, const Column& column,
+ Value& out) {
+ int type = sqlite3_column_type(stmt, index);
+ out.type = column.type;
+ out.is_null = type == SQLITE_NULL;
+ if (out.is_null) return true;
+ switch (column.type) {
+ case common::BOOLEAN:
+ out.b = sqlite3_column_int(stmt, index) != 0;
+ break;
+ case common::INT32:
+ case common::DATE:
+ out.i32 = sqlite3_column_int(stmt, index);
+ break;
+ case common::INT64:
+ case common::TIMESTAMP:
+ out.i64 = sqlite3_column_int64(stmt, index);
+ break;
+ case common::FLOAT:
+ out.f = static_cast<float>(sqlite3_column_double(stmt, index));
+ break;
+ case common::DOUBLE:
+ out.d = sqlite3_column_double(stmt, index);
+ break;
+ case common::TEXT:
+ case common::STRING: {
+ const unsigned char* text = sqlite3_column_text(stmt, index);
+ copy_bytes(out.bytes, text, sqlite3_column_bytes(stmt, index));
+ break;
+ }
+ case common::BLOB: {
+ const void* blob = sqlite3_column_blob(stmt, index);
+ copy_bytes(out.bytes, blob, sqlite3_column_bytes(stmt, index));
+ break;
+ }
+ default:
+ return false;
+ }
+ return true;
+}
+
+int create_shadow_tables(HybridTable* table, bool insert_config) {
+ std::ostringstream data;
+ data << "CREATE TABLE " << shadow_name(table, "_data") << " (";
+ for (size_t i = 0; i < table->columns.size(); ++i) {
+ if (i) data << ',';
+ data << quote_id(table->columns[i].name) << ' '
+ << sqlite_type(table->columns[i].type);
+ if (table->columns[i].category == ColumnCategory::TIME ||
+ table->columns[i].category == ColumnCategory::TAG)
+ data << " NOT NULL";
+ }
+ data << ", UNIQUE(";
+ bool first = true;
+ for (int index : table->tag_indexes) {
+ if (!first) data << ',';
+ first = false;
+ data << quote_id(table->columns[index].name);
+ }
+ data << ',' << quote_id(table->columns[table->time_index].name) << "))";
+ int rc = exec_sql(table->db, data.str(), table);
+ if (rc != SQLITE_OK) return rc;
+
+ std::ostringstream segments;
+ segments << "CREATE TABLE " << shadow_name(table, "_segments")
+ << " (path TEXT PRIMARY KEY, cutoff INTEGER NOT NULL, row_count "
+ "INTEGER NOT NULL)";
+ rc = exec_sql(table->db, segments.str(), table);
+ if (rc != SQLITE_OK) return rc;
+
+ std::ostringstream config;
+ config << "CREATE TABLE " << shadow_name(table, "_config")
+ << " (id INTEGER PRIMARY KEY CHECK(id=1), watermark INTEGER NOT "
+ "NULL, precision TEXT NOT NULL, directory TEXT NOT NULL, schema "
+ "TEXT NOT NULL)";
+ rc = exec_sql(table->db, config.str(), table);
+ if (rc != SQLITE_OK) return rc;
+ if (insert_config) {
+ std::ostringstream insert;
+ insert << "INSERT INTO " << shadow_name(table, "_config")
+ << "(id,watermark,precision,directory,schema) VALUES(1,"
+ << kNoWatermark << ',' << quote_sql(table->precision) << ','
+ << quote_sql(table->directory) << ','
+ << quote_sql(schema_signature(table)) << ')';
+ rc = exec_sql(table->db, insert.str(), table);
+ if (rc != SQLITE_OK) return rc;
+ }
+ return SQLITE_OK;
+}
+
+int load_config(HybridTable* table) {
+ std::string sql = "SELECT watermark,precision,directory,schema FROM " +
+ shadow_name(table, "_config") + " WHERE id=1";
+ sqlite3_stmt* stmt = nullptr;
+ int rc = sqlite3_prepare_v2(table->db, sql.c_str(), -1, &stmt, nullptr);
+ if (rc != SQLITE_OK) return rc;
+ rc = sqlite3_step(stmt);
+ if (rc == SQLITE_ROW) {
+ table->watermark = sqlite3_column_int64(stmt, 0);
+ const unsigned char* precision = sqlite3_column_text(stmt, 1);
+ std::string stored_precision =
+ precision == nullptr ? ""
+ : reinterpret_cast<const char*>(precision);
+ const unsigned char* directory = sqlite3_column_text(stmt, 2);
+ const unsigned char* schema = sqlite3_column_text(stmt, 3);
+ std::string stored_directory =
+ directory == nullptr ? ""
+ : reinterpret_cast<const char*>(directory);
+ std::string stored_schema =
+ schema == nullptr ? "" : reinterpret_cast<const char*>(schema);
+ if (stored_precision != table->precision ||
+ stored_directory != table->directory ||
+ stored_schema != schema_signature(table)) {
+ rc = SQLITE_CORRUPT;
+ } else {
+ table->precision = stored_precision;
+ rc = SQLITE_OK;
+ }
+ } else if (rc == SQLITE_DONE) {
+ rc = SQLITE_CORRUPT;
+ }
+ sqlite3_finalize(stmt);
+ return rc;
+}
+
+int declare_table(HybridTable* table) {
+ std::ostringstream sql;
+ sql << "CREATE TABLE x(";
+ for (size_t i = 0; i < table->columns.size(); ++i) {
+ if (i) sql << ',';
+ sql << quote_id(table->columns[i].name) << ' '
+ << sqlite_type(table->columns[i].type);
+ if (table->columns[i].category == ColumnCategory::TIME ||
+ table->columns[i].category == ColumnCategory::TAG)
+ sql << " NOT NULL";
+ }
+ sql << ",_tsfile_command TEXT HIDDEN,_tsfile_cutoff INTEGER HIDDEN)";
+ return sqlite3_declare_vtab(table->db, sql.str().c_str());
+}
+
+bool parse_args(HybridTable* table, int argc, const char* const* argv,
+ std::string& error) {
+ for (int i = 3; i < argc; ++i) {
+ std::string key, value;
+ if (!parse_key_value(argv[i], key, value)) {
+ error = "module arguments must be key=value";
+ return false;
+ }
+ key = lower(key);
+ if (key == "directory")
+ table->directory = value;
+ else if (key == "timestamp_precision")
+ table->precision = lower(value);
+ else if (key == "column") {
+ Column column;
+ if (!parse_column(value, column, error)) return false;
+ table->columns.push_back(column);
+ } else {
+ error = "unknown tsfile_hybrid option: " + key;
+ return false;
+ }
+ }
+ if (table->directory.empty() || table->directory[0] != '/') {
+ error = "directory must be an absolute path";
+ return false;
+ }
+ if (table->precision != "ms" && table->precision != "us" &&
+ table->precision != "ns") {
+ error = "timestamp_precision must be ms, us, or ns";
+ return false;
+ }
+ if (table->columns.empty() ||
+ table->columns[0].category != ColumnCategory::TIME ||
+ table->columns[0].type != common::TIMESTAMP) {
+ error = "the first column must be TIMESTAMP:TIME";
+ return false;
+ }
+ for (size_t i = 0; i < table->columns.size(); ++i) {
+ if (table->columns[i].name.empty()) {
+ error = "column name cannot be empty";
+ return false;
+ }
+ if (lower(table->columns[i].name) == "_tsfile_command" ||
+ lower(table->columns[i].name) == "_tsfile_cutoff") {
+ error = "column name is reserved by tsfile_hybrid";
+ return false;
+ }
+ for (size_t j = 0; j < i; ++j)
+ if (lower(table->columns[i].name) ==
+ lower(table->columns[j].name)) {
+ error = "duplicate column name";
+ return false;
+ }
+ if (table->columns[i].category == ColumnCategory::TIME) {
+ if (table->time_index != -1) {
+ error = "exactly one TIME column is required";
+ return false;
+ }
+ table->time_index = static_cast<int>(i);
+ } else if (table->columns[i].category == ColumnCategory::TAG) {
+ if (table->columns[i].type != common::STRING) {
+ error = "TAG columns must use STRING";
+ return false;
+ }
+ table->tag_indexes.push_back(static_cast<int>(i));
+ }
+ }
+ if (table->time_index != 0) {
+ error = "TIME column must be the first column";
+ return false;
+ }
+ if (table->tag_indexes.empty()) {
+ error = "at least one TAG column is required";
+ return false;
+ }
+ return true;
+}
+
+int init_table(HybridTable* table, int argc, const char* const* argv,
+ bool create, char** error_message) {
+ table->db_name = argv[1] == nullptr ? "main" : argv[1];
+ table->table_name = argv[2] == nullptr ? "" : argv[2];
+ std::string error;
+ if (!parse_args(table, argc, argv, error)) {
+ if (error_message)
+ *error_message = sqlite3_mprintf("%s", error.c_str());
+ return SQLITE_ERROR;
+ }
+ if (create && !ensure_directory(table->directory)) {
+ if (error_message)
+ *error_message = sqlite3_mprintf("cannot create TsFile directory");
+ return SQLITE_CANTOPEN;
+ }
+ if (!create && !directory_exists(table->directory)) {
+ if (error_message)
+ *error_message = sqlite3_mprintf("TsFile directory does not exist");
+ return SQLITE_CANTOPEN;
+ }
+ if (declare_table(table) != SQLITE_OK) return SQLITE_ERROR;
+ sqlite3_vtab_config(table->db, SQLITE_VTAB_DIRECTONLY);
+ sqlite3_vtab_config(table->db, SQLITE_VTAB_CONSTRAINT_SUPPORT, 1);
+ if (create) {
+ int rc = create_shadow_tables(table, true);
+ if (rc != SQLITE_OK) return rc;
+ } else {
+ int rc = load_config(table);
+ if (rc != SQLITE_OK) return rc;
+ }
+ return SQLITE_OK;
+}
+
+int create_or_connect(sqlite3* db, void* aux, int argc, const char* const* argv,
+ sqlite3_vtab** vtab, char** error_message, bool create) {
+ (void)aux;
+ std::unique_ptr<HybridTable> table(new HybridTable());
+ table->db = db;
+ int rc = init_table(table.get(), argc, argv, create, error_message);
+ if (rc != SQLITE_OK) return rc;
+ *vtab = table.release();
+ return SQLITE_OK;
+}
+
+int xCreate(sqlite3* db, void* aux, int argc, const char* const* argv,
+ sqlite3_vtab** vtab, char** error) {
+ return create_or_connect(db, aux, argc, argv, vtab, error, true);
+}
+int xConnect(sqlite3* db, void* aux, int argc, const char* const* argv,
+ sqlite3_vtab** vtab, char** error) {
+ return create_or_connect(db, aux, argc, argv, vtab, error, false);
+}
+
+int xDisconnect(sqlite3_vtab* vtab) {
+ HybridTable* table = static_cast<HybridTable*>(vtab);
+ delete table;
+ return SQLITE_OK;
+}
+
+int xDestroy(sqlite3_vtab* vtab) {
+ HybridTable* table = static_cast<HybridTable*>(vtab);
+ int rc =
+ exec_sql(table->db,
+ "DROP TABLE IF EXISTS " + shadow_name(table, "_data"), table);
+ if (rc == SQLITE_OK)
+ rc = exec_sql(table->db,
+ "DROP TABLE IF EXISTS " + shadow_name(table, "_segments"),
+ table);
+ if (rc == SQLITE_OK)
+ rc = exec_sql(table->db,
+ "DROP TABLE IF EXISTS " + shadow_name(table, "_config"),
+ table);
+ delete table;
+ return rc;
+}
+
+int xBestIndex(sqlite3_vtab* vtab, sqlite3_index_info* info) {
+ HybridTable* table = static_cast<HybridTable*>(vtab);
+ std::vector<ConstraintSpec> specs;
+ int next_arg = 1;
+ for (int i = 0; i < info->nConstraint; ++i) {
+ const auto& constraint = info->aConstraint[i];
+ if (!constraint.usable || constraint.iColumn < 0 ||
+ constraint.iColumn >= static_cast<int>(table->columns.size()))
+ continue;
+ bool accepted = constraint.iColumn == table->time_index &&
+ (constraint.op == SQLITE_INDEX_CONSTRAINT_GE ||
+ constraint.op == SQLITE_INDEX_CONSTRAINT_GT ||
+ constraint.op == SQLITE_INDEX_CONSTRAINT_LE ||
+ constraint.op == SQLITE_INDEX_CONSTRAINT_LT);
+ accepted =
+ accepted ||
+ (constraint.op == SQLITE_INDEX_CONSTRAINT_EQ &&
+ std::find(table->tag_indexes.begin(), table->tag_indexes.end(),
+ constraint.iColumn) != table->tag_indexes.end() &&
+ (sqlite3_vtab_collation(info, i) == nullptr ||
+ lower(sqlite3_vtab_collation(info, i)) == "binary"));
+ if (!accepted) continue;
+ info->aConstraintUsage[i].argvIndex = next_arg++;
+ info->aConstraintUsage[i].omit = 0;
+ specs.push_back({constraint.iColumn, constraint.op,
+ info->aConstraintUsage[i].argvIndex});
+ }
+ std::vector<int> projection;
+ const sqlite3_uint64 used_columns = info->colUsed;
+ for (size_t i = 0; i < table->columns.size(); ++i) {
+ const bool used = i >= 64 || (used_columns & (1ULL << i)) != 0;
+ if (used) projection.push_back(static_cast<int>(i));
+ }
+ auto add_projection_column = [&projection](int column) {
+ if (std::find(projection.begin(), projection.end(), column) ==
+ projection.end())
+ projection.push_back(column);
+ };
+ // The time column is needed to apply the residual half-open range check;
+ // accepted constraints are also kept available for the second check.
+ add_projection_column(table->time_index);
+ for (const ConstraintSpec& spec : specs) add_projection_column(spec.column);
+ std::sort(projection.begin(), projection.end());
+ if (specs.empty())
+ info->estimatedCost = 1000000.0;
+ else
+ info->estimatedCost = 1000.0;
+ std::ostringstream encoded;
+ for (size_t i = 0; i < specs.size(); ++i) {
+ if (i) encoded << ';';
+ encoded << specs[i].column << ':' << specs[i].op << ':'
+ << specs[i].argv_index;
+ }
+ encoded << "|p:";
+ for (size_t i = 0; i < projection.size(); ++i) {
+ if (i) encoded << ',';
+ encoded << projection[i];
+ }
+ std::string text = encoded.str();
+ if (!text.empty()) {
+ info->idxStr = static_cast<char*>(sqlite3_malloc(text.size() + 1));
+ if (info->idxStr == nullptr) return SQLITE_NOMEM;
+ std::memcpy(info->idxStr, text.c_str(), text.size() + 1);
+ info->needToFreeIdxStr = 1;
+ }
+ return SQLITE_OK;
+}
+
+int parse_specs(const char* encoded, std::vector<ConstraintSpec>& specs,
+ std::vector<int>& projection) {
+ if (encoded == nullptr || *encoded == '\0') return SQLITE_OK;
+ std::string plan(encoded);
+ const size_t projection_marker = plan.find("|p:");
+ std::string spec_text = projection_marker == std::string::npos
+ ? plan
+ : plan.substr(0, projection_marker);
+ if (projection_marker != std::string::npos) {
+ std::stringstream projection_input(plan.substr(projection_marker + 3));
+ std::string part;
+ while (std::getline(projection_input, part, ',')) {
+ if (part.empty()) return SQLITE_ERROR;
+ projection.push_back(std::atoi(part.c_str()));
+ }
+ }
+ std::stringstream input(spec_text);
+ std::string token;
+ while (std::getline(input, token, ';')) {
+ std::stringstream item(token);
+ std::string part;
+ std::vector<int> values;
+ while (std::getline(item, part, ':'))
+ values.push_back(std::atoi(part.c_str()));
+ if (values.size() != 3) return SQLITE_ERROR;
+ specs.push_back({values[0], values[1], values[2]});
+ }
+ return SQLITE_OK;
+}
+
+void apply_constraint_bounds(const std::vector<ConstraintSpec>& specs,
+ sqlite3_value** argv, int64_t& lower_bound,
+ bool& lower_inclusive, int64_t& upper_bound,
+ bool& upper_inclusive,
+ std::vector<std::pair<int, std::string>>& tag_eq) {
+ lower_bound = kNoWatermark;
+ upper_bound = std::numeric_limits<int64_t>::max();
+ lower_inclusive = true;
+ upper_inclusive = true;
+ for (const ConstraintSpec& spec : specs) {
+ sqlite3_value* value = argv[spec.argv_index - 1];
+ if (spec.column == 0 && sqlite3_value_type(value) == SQLITE_INTEGER) {
+ int64_t number = sqlite3_value_int64(value);
+ if (spec.op == SQLITE_INDEX_CONSTRAINT_GE ||
+ spec.op == SQLITE_INDEX_CONSTRAINT_GT) {
+ if (number > lower_bound ||
+ (number == lower_bound &&
+ spec.op == SQLITE_INDEX_CONSTRAINT_GE)) {
+ lower_bound = number;
+ lower_inclusive = spec.op == SQLITE_INDEX_CONSTRAINT_GE;
+ }
+ } else if (spec.op == SQLITE_INDEX_CONSTRAINT_LE ||
+ spec.op == SQLITE_INDEX_CONSTRAINT_LT) {
+ if (number < upper_bound ||
+ (number == upper_bound &&
+ spec.op == SQLITE_INDEX_CONSTRAINT_LE)) {
+ upper_bound = number;
+ upper_inclusive = spec.op == SQLITE_INDEX_CONSTRAINT_LE;
+ }
+ }
+ } else if (spec.op == SQLITE_INDEX_CONSTRAINT_EQ &&
+ sqlite3_value_type(value) == SQLITE_TEXT) {
+ tag_eq.emplace_back(
+ spec.column,
+ std::string(
+ reinterpret_cast<const char*>(sqlite3_value_text(value)),
+ static_cast<size_t>(sqlite3_value_bytes(value))));
+ }
+ }
+}
+
+bool row_matches(const Row& row, const HybridTable* table, int64_t lower,
+ bool lower_inclusive, int64_t upper, bool upper_inclusive,
+ const std::vector<std::pair<int, std::string>>& tag_eq) {
+ const Value& time = row.values[table->time_index];
+ int64_t timestamp = time.type == common::TIMESTAMP ? time.i64 : 0;
+ if (timestamp < lower || (!lower_inclusive && timestamp == lower))
+ return false;
+ if (timestamp > upper || (!upper_inclusive && timestamp == upper))
+ return false;
+ for (const auto& condition : tag_eq) {
+ const Value& value = row.values[condition.first];
+ if (value.is_null || value.bytes != condition.second) return false;
+ }
+ return true;
+}
+
+int read_hot(HybridCursor* cursor, const std::vector<ConstraintSpec>& specs,
+ sqlite3_value** argv, int64_t lower, bool lower_inclusive,
+ int64_t upper, bool upper_inclusive,
+ const std::vector<std::pair<int, std::string>>& tag_eq,
+ const std::vector<int>& projection) {
+ HybridTable* table = cursor->table;
+ std::ostringstream sql;
+ sql << "SELECT ";
+ for (size_t i = 0; i < projection.size(); ++i) {
+ const int column = projection[i];
+ if (i) sql << ',';
+ sql << quote_id(table->columns[column].name);
+ }
+ sql << ",rowid FROM " << shadow_name(table, "_data");
+ if (!specs.empty()) {
+ sql << " WHERE ";
+ for (size_t i = 0; i < specs.size(); ++i) {
+ if (i) sql << " AND ";
+ sql << quote_id(table->columns[specs[i].column].name);
+ switch (specs[i].op) {
+ case SQLITE_INDEX_CONSTRAINT_EQ:
+ sql << "=?";
+ break;
+ case SQLITE_INDEX_CONSTRAINT_GE:
+ sql << ">=?";
+ break;
+ case SQLITE_INDEX_CONSTRAINT_GT:
+ sql << ">?";
+ break;
+ case SQLITE_INDEX_CONSTRAINT_LE:
+ sql << "<=?";
+ break;
+ case SQLITE_INDEX_CONSTRAINT_LT:
+ sql << "<?";
+ break;
+ default:
+ sql << "=?";
+ break;
+ }
+ }
+ }
+ sql << " ORDER BY rowid";
+ sqlite3_stmt* stmt = nullptr;
+ int rc =
+ sqlite3_prepare_v2(table->db, sql.str().c_str(), -1, &stmt, nullptr);
+ if (rc != SQLITE_OK) return rc;
+ for (const ConstraintSpec& spec : specs)
+ sqlite3_bind_value(stmt, spec.argv_index, argv[spec.argv_index - 1]);
+ while ((rc = sqlite3_step(stmt)) == SQLITE_ROW) {
+ Row row;
+ row.rowid =
+ sqlite3_column_int64(stmt, static_cast<int>(projection.size()));
+ row.values.resize(table->columns.size());
+ bool valid = true;
+ for (size_t i = 0; i < projection.size(); ++i) {
+ const int column = projection[i];
+ valid = valid && row_value_from_sqlite(stmt, static_cast<int>(i),
+ table->columns[column],
+ row.values[column]);
+ }
+ if (valid && row_matches(row, table, lower, lower_inclusive, upper,
+ upper_inclusive, tag_eq))
+ cursor->rows.push_back(std::move(row));
+ }
+ sqlite3_finalize(stmt);
+ return rc == SQLITE_DONE ? SQLITE_OK : rc;
+}
+
+std::shared_ptr<TableSchema> build_tsfile_schema(const HybridTable* table) {
+ std::vector<ColumnSchema> columns;
+ for (size_t i = 1; i < table->columns.size(); ++i) {
+ const Column& column = table->columns[i];
+ columns.emplace_back(column.name, column.type, column.category);
+ }
+ return std::make_shared<TableSchema>(table->table_name, columns);
+}
+
+sqlite3_int64 cold_rowid(uint64_t segment_ordinal, uint64_t row_ordinal) {
+ // Keep cold rowids negative and deterministic for a segment traversal.
+ // The high/low split leaves enough room for ordinary segment and row
+ // counts while avoiding collisions with SQLite's positive hot rowids.
+ const uint64_t max_code =
+ static_cast<uint64_t>(std::numeric_limits<sqlite3_int64>::max()) - 1;
+ uint64_t code = (segment_ordinal << 32) | (row_ordinal & 0xffffffffULL);
+ if (code > max_code) code = max_code;
+ return -static_cast<sqlite3_int64>(code + 1);
+}
+
+int append_result_row(HybridCursor* cursor, ResultSet* result,
+ const std::vector<int>& value_indexes,
+ sqlite3_int64 rowid) {
+ HybridTable* table = cursor->table;
+ Row row;
+ row.rowid = rowid;
+ row.values.resize(table->columns.size());
+ row.values[0].type = common::TIMESTAMP;
+ row.values[0].is_null = result->is_null(1);
+ if (!row.values[0].is_null)
+ row.values[0].i64 = result->get_value<int64_t>(1);
+ for (size_t i = 0; i < value_indexes.size(); ++i) {
+ int table_index = value_indexes[i];
+ uint32_t result_index = static_cast<uint32_t>(i + 2);
+ Value& value = row.values[table_index];
+ value.type = table->columns[table_index].type;
+ value.is_null = result->is_null(result_index);
+ if (value.is_null) continue;
+ switch (value.type) {
+ case common::BOOLEAN:
+ value.b = result->get_value<bool>(result_index);
+ break;
+ case common::INT32:
+ value.i32 = result->get_value<int32_t>(result_index);
+ break;
+ case common::DATE:
+ value.i32 = result->get_value<int32_t>(result_index);
+ break;
+ case common::INT64:
+ case common::TIMESTAMP:
+ value.i64 = result->get_value<int64_t>(result_index);
+ break;
+ case common::FLOAT:
+ value.f = result->get_value<float>(result_index);
+ break;
+ case common::DOUBLE:
+ value.d = result->get_value<double>(result_index);
+ break;
+ case common::TEXT:
+ case common::STRING:
+ case common::BLOB: {
+ common::String* string =
+ result->get_value<common::String*>(result_index);
+ if (string != nullptr)
+ copy_bytes(value.bytes, string->buf_, string->len_);
+ break;
+ }
+ default:
+ return SQLITE_ERROR;
+ }
+ }
+ if (row_matches(row, table, kNoWatermark, true,
+ std::numeric_limits<int64_t>::max(), true, {}))
+ cursor->rows.push_back(std::move(row));
+ return SQLITE_OK;
+}
+
+int read_cold(HybridCursor* cursor, const std::vector<ConstraintSpec>& specs,
+ sqlite3_value** argv, int64_t lower, bool lower_inclusive,
+ int64_t upper, bool upper_inclusive,
+ const std::vector<std::pair<int, std::string>>& tag_eq,
+ const std::vector<int>& projection) {
+ HybridTable* table = cursor->table;
+ std::string sql = "SELECT path FROM " + shadow_name(table, "_segments") +
+ " ORDER BY path";
+ sqlite3_stmt* stmt = nullptr;
+ int rc = sqlite3_prepare_v2(table->db, sql.c_str(), -1, &stmt, nullptr);
+ if (rc != SQLITE_OK) return rc;
+ std::shared_ptr<TableSchema> schema = build_tsfile_schema(table);
+ std::vector<std::string> value_columns;
+ std::vector<int> value_indexes;
+ for (int column : projection) {
+ if (column == table->time_index) continue;
+ value_columns.push_back(table->columns[column].name);
+ value_indexes.push_back(column);
+ }
+ uint64_t segment_ordinal = 0;
+ while ((rc = sqlite3_step(stmt)) == SQLITE_ROW) {
+ const unsigned char* path = sqlite3_column_text(stmt, 0);
+ if (path == nullptr) continue;
+ TsFileReader reader;
+ int reader_rc = reader.open(reinterpret_cast<const char*>(path));
+ if (reader_rc != common::E_OK) {
+ sqlite3_finalize(stmt);
+ return SQLITE_IOERR;
+ }
+ storage::Filter* tag_filter = nullptr;
+ storage::TagFilterBuilder tag_builder(schema.get());
+ for (const auto& condition : tag_eq) {
+ storage::Filter* next = tag_builder.eq(
+ table->columns[condition.first].name, condition.second);
+ if (next == nullptr) continue;
+ if (tag_filter == nullptr)
+ tag_filter = next;
+ else
+ tag_filter =
+ storage::TagFilterBuilder::and_filter(tag_filter, next);
+ }
+ ResultSet* result = nullptr;
+ int64_t query_lower =
+ lower == kNoWatermark ? std::numeric_limits<int64_t>::min() : lower;
+ if (!lower_inclusive &&
+ query_lower != std::numeric_limits<int64_t>::max())
+ ++query_lower;
+ int64_t query_upper = upper;
+ if (!upper_inclusive &&
+ query_upper != std::numeric_limits<int64_t>::min())
+ --query_upper;
+ if (query_lower > query_upper) {
+ delete tag_filter;
+ reader.close();
+ continue;
+ }
+ int query_rc =
+ reader.query(table->table_name, value_columns, query_lower,
+ query_upper, result, tag_filter);
+ if (query_rc != common::E_OK) {
+ delete tag_filter;
+ reader.close();
+ sqlite3_finalize(stmt);
+ return SQLITE_IOERR;
+ }
+ bool has_next = false;
+ int next_rc = common::E_OK;
+ uint64_t row_ordinal = 0;
+ while ((next_rc = result->next(has_next)) == common::E_OK && has_next) {
+ if (append_result_row(cursor, result, value_indexes,
+ cold_rowid(segment_ordinal, row_ordinal++)) !=
+ SQLITE_OK) {
+ reader.destroy_query_data_set(result);
+ delete tag_filter;
+ reader.close();
+ sqlite3_finalize(stmt);
+ return SQLITE_ERROR;
+ }
+ }
+ reader.destroy_query_data_set(result);
+ delete tag_filter;
+ reader.close();
+ if (next_rc != common::E_OK) {
+ sqlite3_finalize(stmt);
+ return SQLITE_IOERR;
+ }
+ ++segment_ordinal;
+ }
+ sqlite3_finalize(stmt);
+ return rc == SQLITE_DONE ? SQLITE_OK : rc;
+}
+
+int xOpen(sqlite3_vtab* vtab, sqlite3_vtab_cursor** cursor) {
+ std::unique_ptr<HybridCursor> result(new HybridCursor());
+ result->table = static_cast<HybridTable*>(vtab);
+ *cursor = result.release();
+ return SQLITE_OK;
+}
+int xClose(sqlite3_vtab_cursor* cursor) {
+ delete static_cast<HybridCursor*>(cursor);
+ return SQLITE_OK;
+}
+
+int xFilter(sqlite3_vtab_cursor* cursor_base, int idx_num, const char* idx_str,
+ int argc, sqlite3_value** argv) {
+ (void)idx_num;
+ (void)argc;
+ HybridCursor* cursor = static_cast<HybridCursor*>(cursor_base);
+ cursor->rows.clear();
+ cursor->pos = 0;
+ std::vector<ConstraintSpec> specs;
+ std::vector<int> projection;
+ int rc = parse_specs(idx_str, specs, projection);
+ if (rc != SQLITE_OK) return rc;
+ if (projection.empty()) {
+ for (size_t i = 0; i < cursor->table->columns.size(); ++i)
+ projection.push_back(static_cast<int>(i));
+ }
+ int64_t lower, upper;
+ bool lower_inclusive, upper_inclusive;
+ std::vector<std::pair<int, std::string>> tag_eq;
+ apply_constraint_bounds(specs, argv, lower, lower_inclusive, upper,
+ upper_inclusive, tag_eq);
+ rc = read_hot(cursor, specs, argv, lower, lower_inclusive, upper,
+ upper_inclusive, tag_eq, projection);
+ if (rc == SQLITE_OK)
+ rc = read_cold(cursor, specs, argv, lower, lower_inclusive, upper,
+ upper_inclusive, tag_eq, projection);
+ return rc;
+}
+int xNext(sqlite3_vtab_cursor* cursor_base) {
+ ++static_cast<HybridCursor*>(cursor_base)->pos;
+ return SQLITE_OK;
+}
+int xEof(sqlite3_vtab_cursor* cursor_base) {
+ HybridCursor* cursor = static_cast<HybridCursor*>(cursor_base);
+ return cursor->pos >= cursor->rows.size();
+}
+
+int xColumn(sqlite3_vtab_cursor* cursor_base, sqlite3_context* context,
+ int column) {
+ HybridCursor* cursor = static_cast<HybridCursor*>(cursor_base);
+ if (column >= static_cast<int>(cursor->table->columns.size())) {
+ sqlite3_result_null(context);
+ return SQLITE_OK;
+ }
+ const Value& value = cursor->rows[cursor->pos].values[column];
+ if (value.is_null) {
+ sqlite3_result_null(context);
+ return SQLITE_OK;
+ }
+ switch (value.type) {
+ case common::BOOLEAN:
+ sqlite3_result_int(context, value.b ? 1 : 0);
+ break;
+ case common::INT32:
+ case common::DATE:
+ sqlite3_result_int(context, value.i32);
+ break;
+ case common::INT64:
+ case common::TIMESTAMP:
+ sqlite3_result_int64(context, value.i64);
+ break;
+ case common::FLOAT:
+ sqlite3_result_double(context, value.f);
+ break;
+ case common::DOUBLE:
+ sqlite3_result_double(context, value.d);
+ break;
+ case common::TEXT:
+ case common::STRING:
+ sqlite3_result_text(context, value.bytes.data(),
+ static_cast<int>(value.bytes.size()),
+ SQLITE_TRANSIENT);
+ break;
+ case common::BLOB:
+ sqlite3_result_blob(context, value.bytes.data(),
+ static_cast<int>(value.bytes.size()),
+ SQLITE_TRANSIENT);
+ break;
+ default:
+ sqlite3_result_null(context);
+ break;
+ }
+ return SQLITE_OK;
+}
+int xRowid(sqlite3_vtab_cursor* cursor_base, sqlite3_int64* rowid) {
+ HybridCursor* cursor = static_cast<HybridCursor*>(cursor_base);
+ *rowid = cursor->rows[cursor->pos].rowid;
+ return SQLITE_OK;
+}
+
+int check_mutable(HybridTable* table, const std::vector<Value>& values) {
+ const Value& time = values[table->time_index];
+ if (time.is_null || time.type != common::TIMESTAMP ||
+ time.i64 < table->watermark)
+ return SQLITE_CONSTRAINT;
+ for (int index : table->tag_indexes)
+ if (values[index].is_null) return SQLITE_CONSTRAINT_NOTNULL;
+ return SQLITE_OK;
+}
+
+int insert_hot(HybridTable* table, int argc, sqlite3_value** argv,
+ sqlite3_int64* rowid) {
+ std::vector<Value> values(table->columns.size());
+ std::string error;
+ for (size_t i = 0; i < table->columns.size(); ++i) {
+ if (!value_from_sqlite(argv[i + 2], table->columns[i], values[i],
+ error)) {
+ set_error(table, error);
+ return SQLITE_MISMATCH;
+ }
+ }
+ int rc = check_mutable(table, values);
+ if (rc != SQLITE_OK) {
+ set_error(table,
+ "row is outside the mutable watermark or violates NOT NULL");
+ return rc;
+ }
+ std::ostringstream sql;
+ sql << "INSERT INTO " << shadow_name(table, "_data") << '(';
+ for (size_t i = 0; i < table->columns.size(); ++i) {
+ if (i) sql << ',';
+ sql << quote_id(table->columns[i].name);
+ }
+ sql << ") VALUES(";
+ for (size_t i = 0; i < table->columns.size(); ++i) {
+ if (i) sql << ',';
+ sql << '?';
+ }
+ sql << ')';
+ sqlite3_stmt* stmt = nullptr;
+ rc = sqlite3_prepare_v2(table->db, sql.str().c_str(), -1, &stmt, nullptr);
+ if (rc == SQLITE_OK) {
+ for (size_t i = 0; i < values.size(); ++i)
+ bind_value(stmt, static_cast<int>(i + 1), values[i]);
+ rc = sqlite3_step(stmt);
+ }
+ if (rc == SQLITE_DONE) {
+ *rowid = sqlite3_last_insert_rowid(table->db);
+ rc = SQLITE_OK;
+ }
+ sqlite3_finalize(stmt);
+ return rc;
+}
+
+int delete_hot(HybridTable* table, sqlite3_int64 rowid) {
+ if (rowid < 0) return SQLITE_CONSTRAINT;
+ std::string sql =
+ "DELETE FROM " + shadow_name(table, "_data") + " WHERE rowid=?";
+ sqlite3_stmt* stmt = nullptr;
+ int rc = sqlite3_prepare_v2(table->db, sql.c_str(), -1, &stmt, nullptr);
+ if (rc == SQLITE_OK) {
+ sqlite3_bind_int64(stmt, 1, rowid);
+ rc = sqlite3_step(stmt);
+ }
+ sqlite3_finalize(stmt);
+ return rc == SQLITE_DONE ? SQLITE_OK : rc;
+}
+
+int update_hot(HybridTable* table, int argc, sqlite3_value** argv,
+ sqlite3_int64 rowid) {
+ if (rowid < 0) return SQLITE_CONSTRAINT;
+ std::vector<Value> values(table->columns.size());
+ std::string error;
+ for (size_t i = 0; i < table->columns.size(); ++i) {
+ if (!value_from_sqlite(argv[i + 2], table->columns[i], values[i],
+ error)) {
+ set_error(table, error);
+ return SQLITE_MISMATCH;
+ }
+ }
+ int rc = check_mutable(table, values);
+ if (rc != SQLITE_OK) return rc;
+ std::ostringstream sql;
+ sql << "UPDATE " << shadow_name(table, "_data") << " SET ";
+ for (size_t i = 0; i < table->columns.size(); ++i) {
+ if (i) sql << ',';
+ sql << quote_id(table->columns[i].name) << "=?";
+ }
+ sql << " WHERE rowid=?";
+ sqlite3_stmt* stmt = nullptr;
+ rc = sqlite3_prepare_v2(table->db, sql.str().c_str(), -1, &stmt, nullptr);
+ if (rc == SQLITE_OK) {
+ for (size_t i = 0; i < values.size(); ++i)
+ bind_value(stmt, static_cast<int>(i + 1), values[i]);
+ sqlite3_bind_int64(stmt, static_cast<int>(values.size() + 1), rowid);
+ rc = sqlite3_step(stmt);
+ }
+ sqlite3_finalize(stmt);
+ if (rc == SQLITE_DONE) return SQLITE_OK;
+ return rc;
+}
+
+int write_segment(HybridTable* table, int64_t cutoff, PendingFile& file,
+ sqlite3_int64& row_count) {
+ if (!ensure_directory(table->directory)) return SQLITE_CANTOPEN;
+ std::ostringstream base;
+ do {
+ base.str("");
+ base.clear();
+ base << table->directory << '/' << file_stem(table->table_name) << '-'
+ << static_cast<long long>(getpid()) << '-'
+ << table->file_counter++;
+ file.temporary = base.str() + ".tmp";
+ file.final_path = base.str() + ".tsfile";
+ } while (access(file.temporary.c_str(), F_OK) == 0 ||
+ access(file.final_path.c_str(), F_OK) == 0);
+ WriteFile write_file;
+ int flags = O_WRONLY | O_CREAT | O_EXCL | O_TRUNC;
+ int rc = write_file.create(file.temporary, flags, 0644);
+ if (rc != common::E_OK) return SQLITE_CANTOPEN;
+ std::shared_ptr<TableSchema> schema = build_tsfile_schema(table);
+ TsFileTableWriter writer(&write_file, schema.get());
+ std::vector<std::string> names;
+ std::vector<TSDataType> types;
+ std::vector<ColumnCategory> categories;
+ for (size_t i = 1; i < table->columns.size(); ++i) {
+ names.push_back(table->columns[i].name);
+ types.push_back(table->columns[i].type);
+ categories.push_back(table->columns[i].category);
+ }
+ const int max_rows = 1024;
+ Tablet tablet(table->table_name, names, types, categories, max_rows);
+ std::string sql = "SELECT " + quote_id(table->columns[0].name);
+ for (size_t i = 1; i < table->columns.size(); ++i)
+ sql += "," + quote_id(table->columns[i].name);
+ sql += " FROM " + shadow_name(table, "_data") + " WHERE " +
+ quote_id(table->columns[0].name) + " >= ? AND " +
+ quote_id(table->columns[0].name) + " < ? ORDER BY ";
+ bool first = true;
+ for (int index : table->tag_indexes) {
+ if (!first) sql += ',';
+ first = false;
+ sql += quote_id(table->columns[index].name);
+ }
+ sql += ',' + quote_id(table->columns[0].name);
+ sqlite3_stmt* stmt = nullptr;
+ rc = sqlite3_prepare_v2(table->db, sql.c_str(), -1, &stmt, nullptr);
+ if (rc != SQLITE_OK) return rc;
+ sqlite3_bind_int64(stmt, 1, table->watermark);
+ sqlite3_bind_int64(stmt, 2, cutoff);
+ uint32_t tablet_rows = 0;
+ while ((rc = sqlite3_step(stmt)) == SQLITE_ROW) {
+ int64_t timestamp = sqlite3_column_int64(stmt, 0);
+ tablet.add_timestamp(tablet_rows, timestamp);
+ for (size_t i = 1; i < table->columns.size(); ++i) {
+ const Column& column = table->columns[i];
+ Value value;
+ if (!row_value_from_sqlite(stmt, static_cast<int>(i), column,
+ value)) {
+ sqlite3_finalize(stmt);
+ return SQLITE_ERROR;
+ }
+ if (value.is_null) continue;
+ int add_rc = common::E_OK;
+ switch (column.type) {
+ case common::BOOLEAN:
+ add_rc = tablet.add_value(
+ tablet_rows, static_cast<uint32_t>(i - 1), value.b);
+ break;
+ case common::INT32:
+ case common::DATE:
+ add_rc = tablet.add_value(
+ tablet_rows, static_cast<uint32_t>(i - 1), value.i32);
+ break;
+ case common::INT64:
+ case common::TIMESTAMP:
+ add_rc = tablet.add_value(
+ tablet_rows, static_cast<uint32_t>(i - 1), value.i64);
+ break;
+ case common::FLOAT:
+ add_rc = tablet.add_value(
+ tablet_rows, static_cast<uint32_t>(i - 1), value.f);
+ break;
+ case common::DOUBLE:
+ add_rc = tablet.add_value(
+ tablet_rows, static_cast<uint32_t>(i - 1), value.d);
+ break;
+ case common::TEXT:
+ case common::STRING:
+ case common::BLOB:
+ add_rc = tablet.add_value(tablet_rows,
+ static_cast<uint32_t>(i - 1),
+ common::String(value.bytes));
+ break;
+ default:
+ add_rc = common::E_TYPE_NOT_SUPPORTED;
+ }
+ if (add_rc != common::E_OK) {
+ sqlite3_finalize(stmt);
+ return SQLITE_ERROR;
+ }
+ }
+ ++tablet_rows;
+ ++row_count;
+ if (tablet_rows == max_rows) {
+ if (writer.write_table(tablet) != common::E_OK) {
+ sqlite3_finalize(stmt);
+ return SQLITE_IOERR;
+ }
+ tablet.reset();
+ tablet_rows = 0;
+ }
+ }
+ sqlite3_finalize(stmt);
+ if (rc != SQLITE_DONE) return rc;
+ if (tablet_rows > 0 && writer.write_table(tablet) != common::E_OK)
+ return SQLITE_IOERR;
+ if (row_count == 0) return close_writer_and_sync(write_file, writer);
+ std::vector<uint8_t> precision(table->precision.begin(),
+ table->precision.end());
+ if (writer.add_tsfile_property("tsfile_sqlite.timestamp_precision",
+ precision) != common::E_OK)
+ return SQLITE_IOERR;
+ if (writer.flush() != common::E_OK) return SQLITE_IOERR;
+ return close_writer_and_sync(write_file, writer);
+}
+
+void cleanup_orphans(HybridTable* table) {
+ std::vector<std::string> referenced;
+ std::string sql = "SELECT path FROM " + shadow_name(table, "_segments");
+ sqlite3_stmt* stmt = nullptr;
+ if (sqlite3_prepare_v2(table->db, sql.c_str(), -1, &stmt, nullptr) ==
+ SQLITE_OK) {
+ while (sqlite3_step(stmt) == SQLITE_ROW) {
+ const unsigned char* path = sqlite3_column_text(stmt, 0);
+ if (path != nullptr)
+ referenced.emplace_back(reinterpret_cast<const char*>(path));
+ }
+ }
+ sqlite3_finalize(stmt);
+ DIR* directory = opendir(table->directory.c_str());
+ if (directory == nullptr) return;
+ while (dirent* entry = readdir(directory)) {
+ std::string name(entry->d_name);
+ bool candidate =
+ name.size() > 7 && name.substr(name.size() - 7) == ".tsfile";
+ candidate = candidate ||
+ (name.size() > 4 && name.substr(name.size() - 4) == ".tmp");
+ if (!candidate) continue;
+ std::string path = table->directory + "/" + name;
+ bool pending = false;
+ for (const PendingFile& file : table->pending) {
+ if (path == file.temporary || path == file.final_path) {
+ pending = true;
+ break;
+ }
+ }
+ if (pending) continue;
+ if (std::find(referenced.begin(), referenced.end(), path) ==
+ referenced.end())
+ unlink(path.c_str());
+ }
+ closedir(directory);
+}
+
+int seal(HybridTable* table, int64_t cutoff) {
+ if (cutoff < table->watermark) return SQLITE_CONSTRAINT;
+ cleanup_orphans(table);
+ PendingFile file;
+ sqlite3_int64 row_count = 0;
+ int rc = write_segment(table, cutoff, file, row_count);
+ if (rc != SQLITE_OK) {
+ unlink(file.temporary.c_str());
+ return rc;
+ }
+ if (row_count > 0) {
+ std::string insert = "INSERT INTO " + shadow_name(table, "_segments") +
+ "(path,cutoff,row_count) VALUES(?,?,?)";
+ sqlite3_stmt* stmt = nullptr;
+ rc = sqlite3_prepare_v2(table->db, insert.c_str(), -1, &stmt, nullptr);
+ if (rc == SQLITE_OK) {
+ sqlite3_bind_text(stmt, 1, file.final_path.c_str(), -1,
+ SQLITE_TRANSIENT);
+ sqlite3_bind_int64(stmt, 2, cutoff);
+ sqlite3_bind_int64(stmt, 3, row_count);
+ rc = sqlite3_step(stmt);
+ }
+ sqlite3_finalize(stmt);
+ if (rc != SQLITE_DONE) {
+ unlink(file.temporary.c_str());
+ return rc;
+ }
+ std::string del = "DELETE FROM " + shadow_name(table, "_data") +
+ " WHERE " + quote_id(table->columns[0].name) +
+ " >= ? AND " + quote_id(table->columns[0].name) +
+ " < ?";
+ stmt = nullptr;
+ rc = sqlite3_prepare_v2(table->db, del.c_str(), -1, &stmt, nullptr);
+ if (rc == SQLITE_OK) {
+ sqlite3_bind_int64(stmt, 1, table->watermark);
+ sqlite3_bind_int64(stmt, 2, cutoff);
+ rc = sqlite3_step(stmt);
+ }
+ sqlite3_finalize(stmt);
+ if (rc != SQLITE_DONE) {
+ unlink(file.temporary.c_str());
+ return rc;
+ }
+ table->pending.push_back(std::move(file));
+ } else {
+ unlink(file.temporary.c_str());
+ }
+ std::string update = "UPDATE " + shadow_name(table, "_config") +
+ " SET watermark=? WHERE id=1";
+ sqlite3_stmt* stmt = nullptr;
+ rc = sqlite3_prepare_v2(table->db, update.c_str(), -1, &stmt, nullptr);
+ if (rc == SQLITE_OK) {
+ sqlite3_bind_int64(stmt, 1, cutoff);
+ rc = sqlite3_step(stmt);
+ }
+ sqlite3_finalize(stmt);
+ if (rc != SQLITE_DONE) return rc;
+ table->watermark = cutoff;
+ return SQLITE_OK;
+}
+
+int xUpdate(sqlite3_vtab* vtab, int argc, sqlite3_value** argv,
+ sqlite3_int64* rowid) {
+ HybridTable* table = static_cast<HybridTable*>(vtab);
+ int public_count = static_cast<int>(table->columns.size());
+ if (argc == 1) return delete_hot(table, sqlite3_value_int64(argv[0]));
+ if (argc != public_count + 4) return SQLITE_ERROR;
+ sqlite3_value* command = argv[2 + public_count];
+ if (sqlite3_value_type(command) != SQLITE_NULL) {
+ const unsigned char* text = sqlite3_value_text(command);
+ if (text == nullptr ||
+ lower(reinterpret_cast<const char*>(text)) != "seal") {
+ set_error(vtab, "unknown _tsfile_command");
+ return SQLITE_ERROR;
+ }
+ sqlite3_value* cutoff = argv[3 + public_count];
+ if (sqlite3_value_type(cutoff) != SQLITE_INTEGER) {
+ set_error(vtab, "_tsfile_cutoff must be INTEGER");
+ return SQLITE_MISMATCH;
+ }
+ return seal(table, sqlite3_value_int64(cutoff));
+ }
+ if (sqlite3_value_type(argv[0]) == SQLITE_NULL)
+ return insert_hot(table, argc, argv, rowid);
+ sqlite3_int64 old_rowid = sqlite3_value_int64(argv[0]);
+ if (sqlite3_value_type(argv[1]) == SQLITE_NULL)
+ return delete_hot(table, old_rowid);
+ sqlite3_int64 new_rowid = sqlite3_value_int64(argv[1]);
+ if (new_rowid != old_rowid) return SQLITE_CONSTRAINT;
+ return update_hot(table, argc, argv, old_rowid);
+}
+
+int xBegin(sqlite3_vtab* vtab) {
+ HybridTable* table = static_cast<HybridTable*>(vtab);
+ table->pending.clear();
+ table->savepoint_marks.clear();
+ return SQLITE_OK;
+}
+int xSync(sqlite3_vtab* vtab) {
+ HybridTable* table = static_cast<HybridTable*>(vtab);
+ for (PendingFile& file : table->pending) {
+ if (file.temporary.empty()) continue;
+ TsFileReader reader;
+ int open_rc = reader.open(file.temporary);
+ if (open_rc != common::E_OK) return SQLITE_IOERR;
+ int close_rc = reader.close();
+ if (close_rc != common::E_OK) return SQLITE_IOERR;
+ if (access(file.final_path.c_str(), F_OK) == 0) return SQLITE_IOERR;
+ if (rename(file.temporary.c_str(), file.final_path.c_str()) != 0)
+ return SQLITE_IOERR;
+ file.renamed = true;
+ if (sync_directory(table->directory) != SQLITE_OK)
+ return SQLITE_IOERR_FSYNC;
+ }
+ for (PendingFile& file : table->pending) file.temporary.clear();
+ return SQLITE_OK;
+}
+int xCommit(sqlite3_vtab* vtab) {
+ HybridTable* table = static_cast<HybridTable*>(vtab);
+ table->pending.clear();
+ table->savepoint_marks.clear();
+ return SQLITE_OK;
+}
+int xRollback(sqlite3_vtab* vtab) {
+ HybridTable* table = static_cast<HybridTable*>(vtab);
+ for (const PendingFile& file : table->pending) {
+ if (!file.temporary.empty()) unlink(file.temporary.c_str());
+ if (file.renamed) unlink(file.final_path.c_str());
+ }
+ table->pending.clear();
+ table->savepoint_marks.clear();
+ load_config(table);
+ return SQLITE_OK;
+}
+int xSavepoint(sqlite3_vtab* vtab, int) {
+ HybridTable* table = static_cast<HybridTable*>(vtab);
+ table->savepoint_marks.push_back(table->pending.size());
+ return SQLITE_OK;
+}
+int xRelease(sqlite3_vtab* vtab, int) {
+ HybridTable* table = static_cast<HybridTable*>(vtab);
+ if (!table->savepoint_marks.empty()) table->savepoint_marks.pop_back();
+ return SQLITE_OK;
+}
+int xRollbackTo(sqlite3_vtab* vtab, int) {
+ HybridTable* table = static_cast<HybridTable*>(vtab);
+ if (table->savepoint_marks.empty()) return SQLITE_OK;
+ size_t mark = table->savepoint_marks.back();
+ while (table->pending.size() > mark) {
+ const PendingFile& file = table->pending.back();
+ if (!file.temporary.empty()) unlink(file.temporary.c_str());
+ if (file.renamed) unlink(file.final_path.c_str());
+ table->pending.pop_back();
+ }
+ load_config(table);
+ return SQLITE_OK;
+}
+int xRename(sqlite3_vtab*, const char*) { return SQLITE_CONSTRAINT; }
+int xShadowName(const char* name) {
+ if (name == nullptr) return 0;
+ const char* suffixes[] = {"_data", "_segments", "_config"};
+ for (const char* suffix : suffixes) {
+ size_t length = std::strlen(name), suffix_length = std::strlen(suffix);
+ if (length > suffix_length &&
+ std::strcmp(name + length - suffix_length, suffix) == 0)
+ return 1;
+ }
+ return 0;
+}
+
+const sqlite3_module kModule = {
+ 3, xCreate, xConnect, xBestIndex, xDisconnect,
+ xDestroy, xOpen, xClose, xFilter, xNext,
+ xEof, xColumn, xRowid, xUpdate, xBegin,
+ xSync, xCommit, xRollback, nullptr, xRename,
+ xSavepoint, xRelease, xRollbackTo, xShadowName, nullptr};
+
+} // namespace
+
+extern "C" int sqlite3_extension_init(sqlite3* db, char** error_message,
+ const sqlite3_api_routines* api) {
+ SQLITE_EXTENSION_INIT2(api);
+ static bool initialized = false;
+ if (!initialized) {
+ if (storage::libtsfile_init() != common::E_OK) {
+ if (error_message)
+ *error_message =
+ sqlite3_mprintf("libtsfile initialization failed");
+ return SQLITE_ERROR;
+ }
+ initialized = true;
+ }
+ return sqlite3_create_module_v2(db, "tsfile_hybrid", &kModule, nullptr,
+ nullptr);
+}
diff --git a/cpp/test/CMakeLists.txt b/cpp/test/CMakeLists.txt
index f2d671998..5a843dddb 100644
--- a/cpp/test/CMakeLists.txt
+++ b/cpp/test/CMakeLists.txt
@@ -235,7 +235,7 @@ file(GLOB_RECURSE TEST_SRCS
if (ENABLE_ANTLR4)
file(GLOB_RECURSE PARSER_TEST_SRCS "parser/*_test.cc")
list(APPEND TEST_SRCS ${PARSER_TEST_SRCS})
-endif()
+endif ()
if (ENABLE_SNAPPY)
file(GLOB_RECURSE SNAPPY_TEST_SRCS "compress/*snappy*_test.cc")
@@ -326,6 +326,31 @@ endif ()
set_target_properties(TsFile_Test PROPERTIES RUNTIME_OUTPUT_DIRECTORY ${LIB_TSFILE_SDK_DIR})
+if (BUILD_SQLITE_EXTENSION)
+ add_executable(TsFile_Sqlite_Test sqlite/tsfile_sqlite_test.cc)
+ target_include_directories(TsFile_Sqlite_Test PRIVATE
+ ${VENDORED_GTEST_INCLUDE_DIRS}
+ ${SQLite3_INCLUDE_DIRS})
+ target_link_libraries(TsFile_Sqlite_Test
+ GTest::gtest_main
+ tsfile
+ ${SQLite3_LIBRARIES})
+ add_dependencies(TsFile_Sqlite_Test tsfile_sqlite)
+ target_compile_definitions(TsFile_Sqlite_Test PRIVATE
+ TSFILE_SQLITE_EXTENSION_PATH="$<TARGET_FILE:tsfile_sqlite>")
+ set_target_properties(TsFile_Sqlite_Test PROPERTIES
+ RUNTIME_OUTPUT_DIRECTORY ${LIB_TSFILE_SDK_DIR}
+ SKIP_BUILD_RPATH TRUE)
+ add_test(NAME TsFileSqliteTest COMMAND TsFile_Sqlite_Test)
+ if (APPLE)
+ set_tests_properties(TsFileSqliteTest PROPERTIES
+ ENVIRONMENT "DYLD_LIBRARY_PATH=$<TARGET_FILE_DIR:tsfile>")
+ else ()
+ set_tests_properties(TsFileSqliteTest PROPERTIES
+ ENVIRONMENT "LD_LIBRARY_PATH=$<TARGET_FILE_DIR:tsfile>")
+ endif ()
+endif()
+
# On Windows, copy tsfile DLL next to the test exe so it can load at runtime
# (and when gtest_discover_tests runs the exe). Use TARGET_FILE so the path
# is correct for the current build config (e.g. Release).
diff --git a/cpp/test/sqlite/tsfile_sqlite_test.cc b/cpp/test/sqlite/tsfile_sqlite_test.cc
new file mode 100644
index 000000000..87d0cda2f
--- /dev/null
+++ b/cpp/test/sqlite/tsfile_sqlite_test.cc
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include <gtest/gtest.h>
+#include <sqlite3.h>
+#include <unistd.h>
+
+#include <cstdio>
+#include <string>
+
+namespace {
+
+class TsFileSqliteTest : public ::testing::Test {
+ protected:
+ void SetUp() override {
+ directory_ = "/tmp/tsfile-sqlite-test-" +
+ std::to_string(static_cast<long long>(getpid()));
+ std::string sql = "SELECT load_extension(" +
+ quote(TSFILE_SQLITE_EXTENSION_PATH) + ")";
+ ASSERT_EQ(SQLITE_OK, sqlite3_open(":memory:", &db_));
+ char* error = nullptr;
+ ASSERT_EQ(SQLITE_OK, sqlite3_enable_load_extension(db_, 1));
+ ASSERT_EQ(SQLITE_OK,
+ sqlite3_exec(db_, sql.c_str(), nullptr, nullptr, &error))
+ << (error == nullptr ? "" : error);
+ sqlite3_free(error);
+ ASSERT_EQ(SQLITE_OK,
+ exec("CREATE VIRTUAL TABLE sensor USING tsfile_hybrid("
+ "directory='" +
+ directory_ +
+ "',"
+ "timestamp_precision='ms',"
+ "column='time:TIMESTAMP:TIME',"
+ "column='device:STRING:TAG',"
+ "column='temperature:DOUBLE:FIELD')"));
+ }
+
+ void TearDown() override {
+ if (db_ != nullptr) sqlite3_close(db_);
+ db_ = nullptr;
+ }
+
+ int exec(const std::string& sql) {
+ char* error = nullptr;
+ int rc = sqlite3_exec(db_, sql.c_str(), nullptr, nullptr, &error);
+ if (rc != SQLITE_OK) {
+ ADD_FAILURE() << (error == nullptr ? "" : error);
+ }
+ sqlite3_free(error);
+ return rc;
+ }
+
+ int exec_raw(const std::string& sql) {
+ char* error = nullptr;
+ int rc = sqlite3_exec(db_, sql.c_str(), nullptr, nullptr, &error);
+ sqlite3_free(error);
+ return rc;
+ }
+
+ int count(const std::string& sql) {
+ sqlite3_stmt* stmt = nullptr;
+ EXPECT_EQ(SQLITE_OK,
+ sqlite3_prepare_v2(db_, sql.c_str(), -1, &stmt, nullptr));
+ int result = -1;
+ if (stmt != nullptr && sqlite3_step(stmt) == SQLITE_ROW) {
+ result = sqlite3_column_int(stmt, 0);
+ }
+ sqlite3_finalize(stmt);
+ return result;
+ }
+
+ static std::string quote(const std::string& value) {
+ std::string result = "'";
+ for (char c : value) result += c == '\'' ? "''" : std::string(1, c);
+ result += '\'';
+ return result;
+ }
+
+ sqlite3* db_ = nullptr;
+ std::string directory_;
+};
+
+TEST_F(TsFileSqliteTest, HotCrudAndSealRollback) {
+ ASSERT_EQ(SQLITE_OK, exec("INSERT INTO sensor VALUES(1,'d0',1.5)"));
+ ASSERT_EQ(SQLITE_OK, exec("INSERT INTO sensor VALUES(2,'d0',2.5)"));
+ ASSERT_EQ(2, count("SELECT count(*) FROM sensor"));
+ ASSERT_EQ(SQLITE_OK, exec("BEGIN"));
+ ASSERT_EQ(SQLITE_OK,
+ exec("INSERT INTO sensor(_tsfile_command,_tsfile_cutoff) "
+ "VALUES('seal',2)"));
+ ASSERT_EQ(SQLITE_OK, exec("ROLLBACK"));
+ ASSERT_EQ(2, count("SELECT count(*) FROM sensor"));
+ ASSERT_EQ(0, count("SELECT count(*) FROM sensor_segments"));
+
+ ASSERT_EQ(SQLITE_OK,
+ exec("INSERT INTO sensor(_tsfile_command,_tsfile_cutoff) "
+ "VALUES('seal',2)"));
+ ASSERT_EQ(2, count("SELECT count(*) FROM sensor"));
+ ASSERT_EQ(1, count("SELECT count(*) FROM sensor_segments"));
+ ASSERT_EQ(1, count("SELECT count(*) FROM sensor WHERE time=1"));
+ ASSERT_EQ(1, count("SELECT count(*) FROM sensor WHERE time=2"));
+ char* error = nullptr;
+ int rc = sqlite3_exec(db_, "INSERT INTO sensor VALUES(1,'d0',9.0)", nullptr,
+ nullptr, &error);
+ sqlite3_free(error);
+ EXPECT_EQ(SQLITE_CONSTRAINT, rc);
+}
+
+TEST_F(TsFileSqliteTest, Int32RangeAndColdRowsAreImmutable) {
+ ASSERT_EQ(SQLITE_OK, exec("CREATE VIRTUAL TABLE ints USING tsfile_hybrid("
+ "directory='" +
+ directory_ +
+ "/ints',timestamp_precision='ms',"
+ "column='time:TIMESTAMP:TIME',"
+ "column='device:STRING:TAG',"
+ "column='reading:INT32:FIELD')"));
+ EXPECT_EQ(SQLITE_MISMATCH,
+ exec_raw("INSERT INTO ints VALUES(1,'d0',2147483648)"));
+ ASSERT_EQ(SQLITE_OK, exec("INSERT INTO ints VALUES(1,'d0',42)"));
+ ASSERT_EQ(SQLITE_OK,
+ exec("INSERT INTO ints(_tsfile_command,_tsfile_cutoff) "
+ "VALUES('seal',2)"));
+ EXPECT_EQ(SQLITE_CONSTRAINT,
+ exec_raw("UPDATE ints SET reading=43 WHERE time=1"));
+ EXPECT_EQ(SQLITE_CONSTRAINT, exec_raw("DELETE FROM ints WHERE time=1"));
+ EXPECT_EQ(1, count("SELECT count(*) FROM ints "
+ "WHERE time=1 AND reading=42"));
+}
+
+TEST_F(TsFileSqliteTest, NullAndBlobSurviveSeal) {
+ ASSERT_EQ(SQLITE_OK,
+ exec("CREATE VIRTUAL TABLE payload USING tsfile_hybrid("
+ "directory='" +
+ directory_ +
+ "/payload',timestamp_precision='ms',"
+ "column='time:TIMESTAMP:TIME',"
+ "column='device:STRING:TAG',"
+ "column='payload:BLOB:FIELD',"
+ "column='note:TEXT:FIELD')"));
+ ASSERT_EQ(SQLITE_OK,
+ exec("INSERT INTO payload VALUES(1,'d0',X'000102','')"));
+ ASSERT_EQ(SQLITE_OK, exec("INSERT INTO payload VALUES(2,'d0',NULL,NULL)"));
+ ASSERT_EQ(SQLITE_OK,
+ exec("INSERT INTO payload(_tsfile_command,_tsfile_cutoff) "
+ "VALUES('seal',3)"));
+ EXPECT_EQ(1, count("SELECT count(*) FROM payload "
+ "WHERE time=1 AND hex(payload)='000102'"));
+ EXPECT_EQ(1, count("SELECT count(*) FROM payload "
+ "WHERE time=2 AND payload IS NULL AND note IS NULL"));
+ EXPECT_EQ(1, count("SELECT count(*) FROM payload "
+ "WHERE time=1 AND length(note)=0"));
+}
+
+TEST_F(TsFileSqliteTest, AttachedDatabaseUsesItsOwnShadowTables) {
+ ASSERT_EQ(SQLITE_OK, exec("ATTACH ':memory:' AS aux"));
+ ASSERT_EQ(SQLITE_OK,
+ exec("CREATE VIRTUAL TABLE aux.attached USING tsfile_hybrid("
+ "directory='" +
+ directory_ +
+ "/attached',timestamp_precision='ms',"
+ "column='time:TIMESTAMP:TIME',"
+ "column='device:STRING:TAG',"
+ "column='reading:INT32:FIELD')"));
+ ASSERT_EQ(SQLITE_OK, exec("INSERT INTO aux.attached VALUES(1,'d0',7)"));
+ EXPECT_EQ(1, count("SELECT count(*) FROM aux.attached_data"));
+ EXPECT_EQ(0, count("SELECT count(*) FROM main.sqlite_master "
+ "WHERE name='attached_data'"));
+}
+
+} // namespace