Google GitGoogle Git
Sign in
apache/tsfile/refs/heads/experiment-sqlite/./cpp/src/sqlite/tsfile_sqlite.cc
blob: befaa13a7c0602b9072a5ca8cad5499521051123 [file]
/*
* 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>
#ifndef __APPLE__
#include <linux/fs.h>
#include <sys/syscall.h>
#endif
#include <algorithm>
#include <cctype>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <limits>
#include <map>
#include <memory>
#include <set>
#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 Connection {
std::map<std::string, HybridTable*> tables;
bool managing = false;
};
struct HybridCursor : sqlite3_vtab_cursor {
HybridTable* table = nullptr;
std::vector<Row> rows;
size_t pos = 0;
};
struct HybridTable : sqlite3_vtab {
sqlite3* db = nullptr;
Connection* connection = nullptr;
std::string db_name;
std::string table_name;
std::string directory;
std::string precision;
std::string source_file;
std::string source_table;
std::string source_identity;
bool readonly = false;
bool exhausted = false;
bool source_has_rows = false;
int64_t source_max = kNoWatermark;
std::vector<Column> columns;
int time_index = -1;
std::vector<int> tag_indexes;
int64_t watermark = kNoWatermark;
std::vector<PendingFile> pending;
std::map<int, size_t> savepoint_marks;
uint64_t file_counter = 0;
bool new_directory_owner = false;
bool uncommitted_create = false;
std::vector<std::string> created_directories;
};
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;
}
std::string trim(const std::string& value) {
size_t a = value.find_first_not_of(" \t\r\n");
if (a == std::string::npos) return "";
return value.substr(a, value.find_last_not_of(" \t\r\n") - a + 1);
}
bool sql_token(const std::string& input, size_t& pos, std::string& token) {
while (pos < input.size() &&
std::isspace(static_cast<unsigned char>(input[pos])))
++pos;
token.clear();
if (pos == input.size()) return false;
char quote = input[pos];
if (quote == '\'' || quote == '"' || quote == '`' || quote == '[') {
char end = quote == '[' ? ']' : quote;
++pos;
while (pos < input.size()) {
char c = input[pos++];
if (c == end) {
if (pos < input.size() && input[pos] == end) {
token += end;
++pos;
} else
return true;
} else
token += c;
}
return false;
}
size_t start = pos;
while (pos < input.size() &&
!std::isspace(static_cast<unsigned char>(input[pos])) &&
input[pos] != '=')
++pos;
token = input.substr(start, pos - start);
return !token.empty();
}
bool parse_key_value(const char* arg, std::string& key, std::string& value) {
std::string input = arg == nullptr ? "" : arg;
size_t pos = 0;
if (!sql_token(input, pos, key)) return false;
while (pos < input.size() &&
std::isspace(static_cast<unsigned char>(input[pos])))
++pos;
if (pos == input.size() || input[pos++] != '=') return false;
if (!sql_token(input, pos, value)) return false;
return trim(input.substr(pos)).empty();
}
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 + "_tsfile$" +
(std::string(suffix) == "_data" ? "hot" : suffix + 1));
}
std::string hot_rowid(const HybridTable* table) {
std::string name = "tsfile$rowid";
for (;;) {
bool collision = false;
for (const auto& column : table->columns)
if (lower(column.name) == name) {
collision = true;
break;
}
if (!collision) return quote_id(name);
name += '$';
}
}
std::string schema_signature(const HybridTable* table) {
std::ostringstream out;
for (const auto& column : table->columns)
out << column.name.size() << ':' << column.name << ':'
<< static_cast<int>(column.type) << ':'
<< static_cast<int>(column.category) << ';';
return out.str();
}
bool restore_source_schema(HybridTable* table, std::string& error) {
sqlite3_stmt* stmt = nullptr;
std::string sql =
"SELECT schema,precision,source_max,source_file,source_table,mode "
"FROM " +
shadow_name(table, "_config") + " WHERE id=1";
int rc = sqlite3_prepare_v2(table->db, sql.c_str(), -1, &stmt, nullptr);
if (rc != SQLITE_OK || sqlite3_step(stmt) != SQLITE_ROW) {
sqlite3_finalize(stmt);
error = "stored TsFile configuration is unavailable";
return false;
}
auto text = [stmt](int i) {
const unsigned char* value = sqlite3_column_text(stmt, i);
return value ? std::string(reinterpret_cast<const char*>(value),
sqlite3_column_bytes(stmt, i))
: std::string();
};
std::string signature = text(0), precision = text(1);
bool valid = text(3) == table->source_file &&
text(4) == table->source_table &&
text(5) == (table->readonly ? "readonly" : "writable") &&
(table->precision.empty() || table->precision == precision);
table->precision = precision;
table->source_has_rows = sqlite3_column_type(stmt, 2) != SQLITE_NULL;
table->source_max = sqlite3_column_int64(stmt, 2);
sqlite3_finalize(stmt);
std::istringstream input(signature);
while (valid && input.peek() != std::char_traits<char>::eof()) {
size_t length = 0;
char delimiter = 0;
int type = 0, category = 0;
if (!(input >> length >> delimiter) || delimiter != ':' ||
length > signature.size()) {
valid = false;
break;
}
std::string name(length, '\0');
if (!input.read(&name[0], length) || !(input >> delimiter) ||
delimiter != ':' || !(input >> type >> delimiter) ||
delimiter != ':' || !(input >> category >> delimiter) ||
delimiter != ';') {
valid = false;
break;
}
if (category != static_cast<int>(ColumnCategory::TIME) &&
category != static_cast<int>(ColumnCategory::TAG) &&
category != static_cast<int>(ColumnCategory::FIELD)) {
valid = false;
break;
}
table->columns.push_back({name, static_cast<TSDataType>(type),
static_cast<ColumnCategory>(category)});
}
if (!valid || table->columns.empty()) {
error =
"stored TsFile schema is invalid or conflicts with creation "
"options";
return false;
}
return true;
}
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 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 publish_without_replacing(const std::string& from, const std::string& to) {
#ifdef __APPLE__
return renamex_np(from.c_str(), to.c_str(), RENAME_EXCL) == 0
? SQLITE_OK
: SQLITE_IOERR;
#else
return syscall(SYS_renameat2, AT_FDCWD, from.c_str(), AT_FDCWD, to.c_str(),
RENAME_NOREPLACE) == 0
? SQLITE_OK
: SQLITE_IOERR;
#endif
}
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_int64(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;
}
std::string file_identity(const std::string& path) {
int fd = open(path.c_str(), O_RDONLY);
if (fd < 0) return "";
struct stat st {};
if (fstat(fd, &st) != 0 || !S_ISREG(st.st_mode)) {
close(fd);
return "";
}
uint64_t hash = 14695981039346656037ULL;
unsigned char buffer[16384];
ssize_t n;
while ((n = read(fd, buffer, sizeof(buffer))) > 0)
for (ssize_t i = 0; i < n; ++i) {
hash ^= buffer[i];
hash *= 1099511628211ULL;
}
close(fd);
if (n < 0) return "";
return std::to_string(st.st_size) + ":" + std::to_string(hash);
}
std::shared_ptr<TableSchema> source_schema(TsFileReader& reader,
const std::string& name) {
for (const auto& schema : reader.get_all_table_schemas())
if (lower(schema->get_table_name()) == lower(name)) return schema;
return nullptr;
}
bool infer_source(HybridTable* table, std::string& error) {
table->source_identity = file_identity(table->source_file);
TsFileReader reader;
if (table->source_identity.empty() ||
reader.open(table->source_file) != common::E_OK) {
error = "cannot read source file: " + table->source_file;
return false;
}
auto schema = source_schema(reader, table->source_table);
if (!schema) {
error = "source table not found: " + table->source_table;
reader.close();
return false;
}
auto properties = reader.get_tsfile_properties();
std::string time_name = "time";
auto time_property = properties.find("tsfile_sqlite.time_column");
if (time_property != properties.end() && !time_property->second.is_null)
time_name.assign(time_property->second.value.begin(),
time_property->second.value.end());
table->columns.push_back(
{time_name, common::TIMESTAMP, ColumnCategory::TIME});
auto names = schema->get_measurement_names();
auto fields = schema->get_measurement_schemas();
auto categories = schema->get_column_categories();
for (size_t i = 0; i < fields.size(); ++i)
table->columns.push_back(
{names[i], fields[i]->data_type_, categories[i]});
auto precision = properties.find("tsfile_sqlite.timestamp_precision");
if (precision != properties.end()) {
std::string known(precision->second.value.begin(),
precision->second.value.end());
if (precision->second.is_null ||
(known != "ms" && known != "us" && known != "ns") ||
(!table->precision.empty() && table->precision != known)) {
error = "invalid or conflicting timestamp_precision: " +
table->source_file;
reader.close();
return false;
}
table->precision = known;
}
if (table->precision.empty()) table->precision = "unknown";
if (!reader.get_table_schema(table->source_table)) {
reader.close();
return true;
}
ResultSet* result = nullptr;
int rc = reader.query(table->source_table, names, kNoWatermark,
std::numeric_limits<int64_t>::max(), result);
if (rc != common::E_OK) {
error = "cannot query source table";
reader.close();
return false;
}
bool next = false;
while ((rc = result->next(next)) == common::E_OK && next) {
table->source_has_rows = true;
table->source_max =
std::max(table->source_max, result->get_value<int64_t>(1));
}
reader.destroy_query_data_set(result);
reader.close();
if (rc != common::E_OK ||
file_identity(table->source_file) != table->source_identity) {
error = "source file unreadable or changed during creation";
return false;
}
if (!table->readonly && table->source_has_rows) {
table->exhausted =
table->source_max == std::numeric_limits<int64_t>::max();
if (!table->exhausted) table->watermark = table->source_max + 1;
}
return true;
}
int create_shadow_tables(HybridTable* table, bool insert_config) {
int rc = SQLITE_OK;
if (!table->readonly) {
std::ostringstream data;
data << "CREATE TABLE " << shadow_name(table, "_data") << " ("
<< hot_rowid(table) << " INTEGER PRIMARY KEY";
for (size_t i = 0; i < table->columns.size(); ++i) {
data << ',';
data << quote_id(table->columns[i].name) << ' '
<< sqlite_type(table->columns[i].type);
if (table->columns[i].category == ColumnCategory::TIME)
data << " NOT NULL";
}
data << ')';
rc = exec_sql(table->db, data.str(), table);
if (rc != SQLITE_OK) return rc;
std::ostringstream unique;
unique << "CREATE UNIQUE INDEX " << quote_id(table->db_name) << '.'
<< quote_id(table->table_name + "_tsfile$key") << " ON "
<< quote_id(table->table_name + "_tsfile$hot") << '(';
for (int index : table->tag_indexes) {
std::string name = quote_id(table->columns[index].name);
unique << '(' << name << " IS NULL),coalesce(" << name << ",''),";
}
unique << quote_id(table->columns[table->time_index].name) << ')';
rc = exec_sql(table->db, unique.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, source_table TEXT NOT NULL, identity TEXT "
"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, precision "
"TEXT NOT NULL, "
"directory TEXT NOT NULL, schema TEXT NOT NULL, mode TEXT NOT NULL, "
"source_file TEXT NOT NULL, source_table TEXT NOT NULL, source_max "
"INTEGER)";
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")
<< " VALUES(1,"
<< (table->readonly || table->exhausted
? "NULL"
: std::to_string(table->watermark))
<< ',' << quote_sql(table->precision) << ','
<< quote_sql(table->directory) << ','
<< quote_sql(schema_signature(table)) << ','
<< quote_sql(table->readonly ? "readonly" : "writable") << ','
<< quote_sql(table->source_file) << ','
<< quote_sql(table->source_table) << ','
<< (table->source_has_rows ? std::to_string(table->source_max)
: "NULL")
<< ')';
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->exhausted =
!table->readonly && sqlite3_column_type(stmt, 0) == SQLITE_NULL;
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)
sql << " NOT NULL";
}
sql << ")";
return sqlite3_declare_vtab(table->db, sql.str().c_str());
}
bool parse_args(HybridTable* table, int argc, const char* const* argv,
std::string& error, bool create) {
std::set<std::string> options;
for (int i = 3; i < argc; ++i) {
std::string key, value;
if (!parse_key_value(argv[i], key, value)) {
std::string input = argv[i], name, type, category;
size_t pos = 0;
if (!sql_token(input, pos, name) || !sql_token(input, pos, type) ||
!sql_token(input, pos, category) ||
!trim(input.substr(pos)).empty()) {
error = "column must be name TYPE CATEGORY";
return false;
}
Column column;
if (!parse_column("placeholder:" + type + ":" + category, column,
error))
return false;
column.name = name;
table->columns.push_back(column);
continue;
}
key = lower(key);
if (!options.insert(key).second) {
error = "duplicate option: " + key;
return false;
}
if (key == "directory")
table->directory = value;
else if (key == "file")
table->source_file = value;
else if (key == "source_table")
table->source_table = value;
else if (key == "timestamp_precision")
table->precision = lower(value);
else {
error = "unknown tsfile_hybrid option: " + key;
return false;
}
}
if (options.count("timestamp_precision") && table->precision != "ms" &&
table->precision != "us" && table->precision != "ns") {
error = "timestamp_precision must be ms, us, or ns";
return false;
}
table->readonly = options.count("file") && !options.count("directory");
if (options.count("file")) {
if (table->source_file.empty() || table->source_file[0] != '/' ||
table->source_table.empty() || !table->columns.empty()) {
error =
"file requires an absolute path, source_table, and inferred "
"columns";
return false;
}
if (create ? !infer_source(table, error)
: !restore_source_schema(table, error))
return false;
} else if (options.count("source_table")) {
error = "source_table requires file";
return false;
}
if (!table->readonly &&
(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" &&
!(table->readonly && table->precision == "unknown")) {
error =
"timestamp_precision must be ms, us, or ns for a writable table";
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() ||
sqlite_type(table->columns[i].type).empty()) {
error = "column name or type is unsupported";
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;
}
return true;
}
std::string directory_owner(const HybridTable* table) {
const char* filename =
sqlite3_db_filename(table->db, table->db_name.c_str());
std::string database;
if (filename && filename[0]) {
char* resolved = realpath(filename, nullptr);
database = resolved ? resolved : filename;
free(resolved);
} else {
database =
"connection:" +
std::to_string(reinterpret_cast<uintptr_t>(table->connection)) +
':' + table->db_name;
}
return "tsfile-sqlite:1\n" + database + '\n' + table->table_name + '\n';
}
int check_directory_owner(HybridTable* table) {
if (table->readonly) return SQLITE_OK;
int fd = open((table->directory + "/.tsfile-owner").c_str(), O_RDONLY);
if (fd < 0) {
set_error(table, "TsFile directory ownership record is missing");
return SQLITE_CANTOPEN;
}
const std::string expected = directory_owner(table);
std::string actual(expected.size() + 1, '\0');
ssize_t bytes = read(fd, &actual[0], actual.size());
close(fd);
if (bytes != static_cast<ssize_t>(expected.size()) ||
actual.compare(0, expected.size(), expected) != 0) {
set_error(
table,
"TsFile directory belongs to a different SQLite database or table");
return SQLITE_CONSTRAINT;
}
return SQLITE_OK;
}
void release_new_directory(HybridTable* table) {
if (table->new_directory_owner)
unlink((table->directory + "/.tsfile-owner").c_str());
table->new_directory_owner = false;
for (auto it = table->created_directories.rbegin();
it != table->created_directories.rend(); ++it)
rmdir(it->c_str());
table->created_directories.clear();
}
int acquire_directory(HybridTable* table) {
if (table->readonly) return SQLITE_OK;
// Resolve existing ancestors before testing ownership so aliases and '..'
// cannot evade a parent's exclusive directory reservation.
std::string path = table->directory;
std::vector<std::string> missing;
char* resolved = nullptr;
while (!(resolved = realpath(path.c_str(), nullptr))) {
size_t slash = path.find_last_of('/');
if (slash == std::string::npos || path.empty()) return SQLITE_CANTOPEN;
std::string component = path.substr(slash + 1);
if (component.empty() || component == "." || component == "..")
return SQLITE_CANTOPEN;
missing.push_back(component);
path = slash == 0 ? "/" : path.substr(0, slash);
}
std::string existing(resolved);
free(resolved);
for (std::string ancestor = existing; !ancestor.empty();) {
if (access((ancestor + "/.tsfile-owner").c_str(), F_OK) == 0) {
set_error(table, "directory is already owned by a TsFile table: " +
ancestor);
return SQLITE_CONSTRAINT;
}
size_t slash = ancestor.find_last_of('/');
if (slash == std::string::npos || ancestor == "/") break;
ancestor = slash == 0 ? "/" : ancestor.substr(0, slash);
}
path = existing;
for (auto it = missing.rbegin(); it != missing.rend(); ++it) {
path += (path == "/" ? "" : "/") + *it;
if (mkdir(path.c_str(), 0755) != 0) return SQLITE_CANTOPEN;
table->created_directories.push_back(path);
}
DIR* dir = opendir(path.c_str());
if (!dir) return SQLITE_CANTOPEN;
bool empty = true;
while (dirent* entry = readdir(dir))
if (std::strcmp(entry->d_name, ".") &&
std::strcmp(entry->d_name, "..")) {
empty = false;
break;
}
closedir(dir);
if (!empty) {
set_error(table, "new table directory must be empty: " + path);
return SQLITE_CONSTRAINT;
}
int fd = open((path + "/.tsfile-owner").c_str(),
O_WRONLY | O_CREAT | O_EXCL, 0600);
if (fd < 0) return SQLITE_CANTOPEN;
// This marker reserves the directory across connections and databases;
// it is never inferred from a filename suffix or another table's files.
std::string owner = directory_owner(table);
bool ok = write(fd, owner.data(), owner.size()) ==
static_cast<ssize_t>(owner.size());
ok = fsync(fd) == 0 && ok;
close(fd);
table->new_directory_owner = true;
if (!ok || sync_directory(path) != SQLITE_OK) return SQLITE_IOERR_FSYNC;
return SQLITE_OK;
}
int init_table(HybridTable* table, int argc, const char* const* argv,
bool create, char** error_message) {
table->uncommitted_create = create;
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, create)) {
if (error_message)
*error_message = sqlite3_mprintf("%s", error.c_str());
return SQLITE_ERROR;
}
if (create) {
int rc = acquire_directory(table);
if (rc != SQLITE_OK) {
if (error_message)
*error_message = sqlite3_mprintf(
"%s", table->zErrMsg ? table->zErrMsg
: "cannot acquire TsFile directory");
return rc;
}
}
if (!create && !table->readonly && !directory_exists(table->directory)) {
if (error_message)
*error_message = sqlite3_mprintf("TsFile directory does not exist");
return SQLITE_CANTOPEN;
}
if (!create) {
int rc = check_directory_owner(table);
if (rc != SQLITE_OK) {
if (error_message)
*error_message = sqlite3_mprintf(
"%s", table->zErrMsg ? table->zErrMsg
: "invalid directory owner");
return rc;
}
}
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;
if (!table->source_file.empty()) {
rc = exec_sql(table->db,
"INSERT INTO " + shadow_name(table, "_segments") +
" VALUES(" + quote_sql(table->source_file) +
",0,0," + quote_sql(table->source_table) + "," +
quote_sql(table->source_identity) + ")",
table);
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) {
std::unique_ptr<HybridTable> table(new HybridTable());
table->db = db;
table->connection = static_cast<Connection*>(aux);
int rc = init_table(table.get(), argc, argv, create, error_message);
if (rc != SQLITE_OK) {
release_new_directory(table.get());
return rc;
}
table->connection
->tables[lower(table->db_name) + "." + lower(table->table_name)] =
table.get();
*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);
table->connection->tables.erase(lower(table->db_name) + "." +
lower(table->table_name));
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);
release_new_directory(table);
table->connection->tables.erase(lower(table->db_name) + "." +
lower(table->table_name));
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;
if (table->readonly) return SQLITE_OK;
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 << ',' << hot_rowid(table) << " 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 " << hot_rowid(table);
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,source_table,identity 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;
std::string read_path = reinterpret_cast<const char*>(path);
std::string selected =
reinterpret_cast<const char*>(sqlite3_column_text(stmt, 1));
std::string identity =
reinterpret_cast<const char*>(sqlite3_column_text(stmt, 2));
for (const auto& pending : table->pending)
if (pending.final_path == read_path && !pending.temporary.empty() &&
!pending.renamed)
read_path = pending.temporary;
if (file_identity(read_path) != identity) {
set_error(table, "missing or changed file for " +
table->table_name + ": " + read_path);
sqlite3_finalize(stmt);
return SQLITE_IOERR;
}
TsFileReader reader;
int reader_rc = reader.open(read_path);
if (reader_rc != common::E_OK) {
sqlite3_finalize(stmt);
return SQLITE_IOERR;
}
if (!reader.get_table_schema(selected) &&
source_schema(reader, selected)) {
reader.close();
++segment_ordinal;
continue;
}
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(selected, 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 (table->exhausted || time.is_null || time.type != common::TIMESTAMP ||
time.i64 < table->watermark)
return SQLITE_CONSTRAINT;
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) {
set_error(table, "TsFile cold rows are immutable");
return SQLITE_READONLY;
}
std::string sql = "DELETE FROM " + shadow_name(table, "_data") + " WHERE " +
hot_rowid(table) + "=?";
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) {
set_error(table, "TsFile cold rows are immutable");
return SQLITE_READONLY;
}
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 " << hot_rowid(table) << "=?";
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_rows(HybridTable* table, std::vector<Row> rows,
const std::string& path, bool* created) {
*created = false;
std::stable_sort(rows.begin(), rows.end(),
[table](const Row& a, const Row& b) {
for (int index : table->tag_indexes) {
const Value& x = a.values[index];
const Value& y = b.values[index];
if (x.is_null != y.is_null) return x.is_null;
if (!x.is_null && x.bytes != y.bytes)
return x.bytes < y.bytes;
}
return a.values[0].i64 < b.values[0].i64;
});
WriteFile write_file;
if (write_file.create(path, O_WRONLY | O_CREAT | O_EXCL, 0644) !=
common::E_OK)
return SQLITE_CANTOPEN;
*created = true;
auto 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);
uint32_t tablet_rows = 0;
for (const Row& row : rows) {
tablet.add_timestamp(tablet_rows, row.values[0].i64);
for (size_t i = 1; i < table->columns.size(); ++i) {
const Column& column = table->columns[i];
const Value& value = row.values[i];
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) {
return SQLITE_ERROR;
}
}
++tablet_rows;
if (tablet_rows == max_rows) {
if (writer.write_table(tablet) != common::E_OK) return SQLITE_IOERR;
tablet.reset();
tablet_rows = 0;
}
}
if (tablet_rows && writer.write_table(tablet) != common::E_OK)
return SQLITE_IOERR;
if (table->precision != "unknown") {
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;
}
std::vector<uint8_t> time_name(table->columns[0].name.begin(),
table->columns[0].name.end());
if (writer.add_tsfile_property("tsfile_sqlite.time_column", time_name) !=
common::E_OK)
return SQLITE_IOERR;
if (writer.flush() != common::E_OK) return SQLITE_IOERR;
return close_writer_and_sync(write_file, writer);
}
int write_segment(HybridTable* table, int64_t cutoff, PendingFile& file,
sqlite3_int64& row_count, bool all = false) {
HybridCursor cursor;
cursor.table = table;
std::vector<int> projection;
for (size_t i = 0; i < table->columns.size(); ++i) projection.push_back(i);
int rc = read_hot(&cursor, {}, nullptr, table->watermark, true, cutoff, all,
{}, projection);
if (rc != SQLITE_OK) return rc;
row_count = cursor.rows.size();
if (row_count == 0) return SQLITE_OK;
if (!directory_exists(table->directory)) return SQLITE_CANTOPEN;
struct stat st {};
do {
std::string base = table->directory + '/' +
file_stem(table->table_name) + '-' +
std::to_string(getpid()) + '-' +
std::to_string(table->file_counter++);
file.temporary = base + ".tmp";
file.final_path = base + ".tsfile";
} while (lstat(file.temporary.c_str(), &st) == 0 ||
lstat(file.final_path.c_str(), &st) == 0);
bool created = false;
rc = write_rows(table, std::move(cursor.rows), file.temporary, &created);
// An O_EXCL failure can mean another directory entry appeared after the
// check. Cleanup must not unlink a path this operation never created.
if (!created) file.temporary.clear();
return rc;
}
int seal(HybridTable* table, int64_t cutoff, sqlite3_int64* sealed = nullptr,
bool all = false) {
if (cutoff < table->watermark) return SQLITE_CONSTRAINT;
if (table->readonly) return SQLITE_READONLY;
if (table->exhausted) return SQLITE_CONSTRAINT;
PendingFile file;
sqlite3_int64 row_count = 0;
int rc = write_segment(table, cutoff, file, row_count, all);
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,source_table,identity) 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);
sqlite3_bind_text(stmt, 4, table->table_name.c_str(), -1,
SQLITE_TRANSIENT);
std::string identity = file_identity(file.temporary);
sqlite3_bind_text(stmt, 5, identity.c_str(), -1, SQLITE_TRANSIENT);
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) +
(all ? " <= ?" : " < ?");
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) {
if (all && cutoff == std::numeric_limits<int64_t>::max())
sqlite3_bind_null(stmt, 1);
else
sqlite3_bind_int64(stmt, 1, cutoff);
rc = sqlite3_step(stmt);
}
sqlite3_finalize(stmt);
if (rc != SQLITE_DONE) return rc;
table->exhausted = all && cutoff == std::numeric_limits<int64_t>::max();
table->watermark = cutoff;
if (sealed) *sealed = row_count;
return SQLITE_OK;
}
int xUpdate(sqlite3_vtab* vtab, int argc, sqlite3_value** argv,
sqlite3_int64* rowid) {
HybridTable* table = static_cast<HybridTable*>(vtab);
if (table->readonly) {
set_error(table, "table is readonly");
return SQLITE_READONLY;
}
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 + 2) return SQLITE_ERROR;
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();
int rc = check_directory_owner(table);
return rc == SQLITE_OK ? load_config(table) : rc;
}
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 (publish_without_replacing(file.temporary, file.final_path) !=
SQLITE_OK)
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->new_directory_owner = false;
table->uncommitted_create = false;
table->created_directories.clear();
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();
release_new_directory(table);
load_config(table);
return SQLITE_OK;
}
int xSavepoint(sqlite3_vtab* vtab, int id) {
HybridTable* table = static_cast<HybridTable*>(vtab);
table->savepoint_marks[id] = table->pending.size();
return SQLITE_OK;
}
int xRelease(sqlite3_vtab* vtab, int id) {
HybridTable* table = static_cast<HybridTable*>(vtab);
table->savepoint_marks.erase(table->savepoint_marks.lower_bound(id),
table->savepoint_marks.end());
return SQLITE_OK;
}
int xRollbackTo(sqlite3_vtab* vtab, int id) {
HybridTable* table = static_cast<HybridTable*>(vtab);
auto it = table->savepoint_marks.find(id);
size_t mark = it == table->savepoint_marks.end() ? 0 : it->second;
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();
}
table->savepoint_marks.erase(table->savepoint_marks.upper_bound(id),
table->savepoint_marks.end());
int rc = load_config(table);
if (rc != SQLITE_OK && table->uncommitted_create) {
// SQLite has already undone this table's creation. Its shadow config
// no longer exists; the vtab is about to be disconnected.
release_new_directory(table);
return SQLITE_OK;
}
return rc;
}
int xRename(sqlite3_vtab*, const char*) { return SQLITE_CONSTRAINT; }
int xShadowName(const char* name) {
if (name == nullptr) return 0;
const char* suffixes[] = {"tsfile$hot", "tsfile$segments", "tsfile$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};
#include "tsfile_sqlite_management.inc"
} // 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;
}
Connection* connection = new Connection();
int rc = sqlite3_create_module_v2(
db, "tsfile_hybrid", &kModule, connection,
[](void* p) { delete static_cast<Connection*>(p); });
if (rc != SQLITE_OK) return rc;
return register_management(db, connection);
}