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:17 UTC
(tsfile) branch experiment-sqlite created (now 9871a42e3)
This is an automated email from the ASF dual-hosted git repository.
ColinLeeo pushed a change to branch experiment-sqlite
in repository https://gitbox.apache.org/repos/asf/tsfile.git
at 9871a42e3 feat(sqlite): add experimental TsFile hybrid extension
This branch includes the following new commits:
new 9871a42e3 feat(sqlite): add experimental TsFile hybrid extension
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(tsfile) 01/01: feat(sqlite): add experimental TsFile hybrid extension
Posted by co...@apache.org.
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