wirenboard / wb-mqtt-db / 96

Committed 31 Jul 2026 10:43AM UTC coverage: 78.543% (-1.8%) from 80.337%

Build # 96

Build Type

push

github

Committed by

web-flow

Commit Message

Use exported vars instead of MAKEFLAGS for coverage options (#66)

Coverage Stats

938 of 1102 branches covered (85.12%)

1563 of 1990 relevant lines covered (78.54%)

13.11 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

84.44

/src/sqlite_storage.cpp

#include "sqlite_storage.h"
#include "SQLiteCpp/SQLiteCpp.h"

#include <sqlite3.h>

#include <fstream>
#include <sys/stat.h>
#include <wblib/utils.h>

#include "db_migrations.h"
#include "log.h"
#include "utils.h"

using namespace std;
using namespace WBMQTT;
using namespace std::chrono;
using namespace Utils;

#define LOG(logger) ::logger.Log() << "[sqlite] "

namespace
{
    const char* DB_BACKUP_FILE_EXTENSION = ".backup";
    const int WB_DB_VERSION = 6;

    // const int UNDEFINED_ID = -1;
    const int CHANNEL_COLUMN = 1;

    string BackupFileName(const string& filename)
    {
        return filename + DB_BACKUP_FILE_EXTENSION;
    }

    // Function to optimize database for better performance
    void TuneDatabase(SQLite::Database& db)
    {
        // The WAL journaling mode uses a write-ahead log instead of a rollback
        // journal to implement transactions.
        // * WAL is significantly faster in most scenarios.
        // * WAL uses many fewer fsync() operations and is thus less vulnerable
        //   to problems on systems where the fsync() system call is broken.
        db.exec("PRAGMA journal_mode=WAL");

        // In WAL mode when synchronous is NORMAL, the WAL file is synchronized before
        // each checkpoint and the database file is synchronized after each completed
        // checkpoint and the WAL file header is synchronized when a WAL file begins
        // to be reused after a checkpoint, but no sync operations occur during most
        // transactions
        db.exec("PRAGMA synchronous=NORMAL");
    }

} // namespace

TSqliteStorage::TSqliteStorage(const string& dbFile)
{
    if (sqlite3_compileoption_used("ENABLE_UPDATE_DELETE_LIMIT") == 0) {
        wb_throw(TBaseException,
                 "libsqlite3 is built without SQLITE_ENABLE_UPDATE_DELETE_LIMIT "
                 "required for DELETE ... ORDER BY ... LIMIT statements");
    }

    bool isMemoryDb = (dbFile.find(":memory:") != string::npos);

    // check if backup file is present; if so, we should try to repair DB
    if (!isMemoryDb && CheckBackupFile(dbFile)) {
        LOG(Warn) << "Something went wrong last time, restoring old backup file";
        RestoreBackupFile(dbFile);
    }

    int flags = SQLite::OPEN_READWRITE | SQLite::OPEN_CREATE;
    if (isMemoryDb) {
        flags |= SQLite::OPEN_URI;
    }

    DB = std::make_unique<SQLite::Database>(dbFile, flags);

    TuneDatabase(*DB);

    if (!DB->tableExists("data")) {
        // new DB file created
        LOG(Info) << "Creating tables";
        CreateTables(WB_DB_VERSION);
    } else {
        int file_db_version = ReadDBVersion();
        if (file_db_version > WB_DB_VERSION) {
            wb_throw(TBaseException, "Database file is created by newer version of wb-mqtt-db");
        }
        if (file_db_version < WB_DB_VERSION) {
            LOG(Warn) << "Old database format found, trying to update...";
            CreateBackupFile(dbFile);
            UpdateDB(file_db_version);
        } else {
            LOG(Info) << "Creating tables if necessary";
            CreateTables(WB_DB_VERSION);
        }
    }

    LOG(Info) << "Create indices if necessary";
    CreateIndices();

    LOG(Info) << "Analyzing data table";
    DB->exec("ANALYZE data");
    DB->exec("ANALYZE sqlite_master");

    InsertRowQuery.reset(new SQLite::Statement(*DB,
                                               "INSERT INTO data (channel, value, min, max, retained, timestamp) "
                                               "VALUES (?, ?, ?, ?, ?, ?)"));

    CleanChannelQuery.reset(
        new SQLite::Statement(*DB, "DELETE FROM data WHERE channel = ? ORDER BY timestamp ASC LIMIT ?"));

    LOG(Info) << "DB initialization is done";

    if (!isMemoryDb && CheckBackupFile(dbFile)) {
        RemoveBackupFile(dbFile);
    }

    Load();
}

void TSqliteStorage::CreateTables(int dbVersion)
{
    LOG(Debug) << "Creating 'channels' table...";
    DB->exec("CREATE TABLE IF NOT EXISTS channels ( "
             "int_id INTEGER PRIMARY KEY AUTOINCREMENT, "
             "device VARCHAR(255), "
             "control VARCHAR(255), "
             "precision REAL, "
             "UNIQUE(device,control) "
             ")  ");

    LOG(Debug) << "Creating 'data' table...";
    DB->exec("CREATE TABLE IF NOT EXISTS data ("
             "uid INTEGER PRIMARY KEY AUTOINCREMENT, "
             "channel INTEGER,"
             "value VARCHAR(255),"
             "timestamp INTEGER DEFAULT(0),"
             "max VARCHAR(255),"
             "min VARCHAR(255),"
             "retained INTEGER"
             ")");

    LOG(Debug) << "Creating 'variables' table...";
    DB->exec("CREATE TABLE IF NOT EXISTS variables ("
             "name VARCHAR(255) PRIMARY KEY, "
             "value VARCHAR(255) )");

    {
        LOG(Debug) << "Updating database version variable...";
        SQLite::Statement query(*DB,
                                "INSERT OR REPLACE INTO variables (name, "
                                "value) VALUES ('db_version', ?)");
        query.bind(1, dbVersion);
        query.exec();
    }
}

void TSqliteStorage::CreateIndices()
{
    LOG(Debug) << "Creating 'data_topic' index on 'data' ('channel')";
    DB->exec("CREATE INDEX IF NOT EXISTS data_topic ON data (channel)");

    // NOTE: the following index is a "low quality" one according to sqlite
    // documentation. However, reversing the order of columns results in factor of
    // two decrease in SELECT performance. So we leave it here as it is.
    LOG(Debug) << "Creating 'data_topic_timestamp' index on 'data' ('channel', "
                  "'timestamp')";
    DB->exec("CREATE INDEX IF NOT EXISTS data_topic_timestamp ON data (channel, "
             "timestamp)");
}

void TSqliteStorage::Load()
{
    std::lock_guard<std::mutex> lg(Mutex);
    SQLite::Statement query(*DB, "SELECT int_id, device, control, precision FROM channels");
    SQLite::Statement rowCountQuery(*DB, "SELECT COUNT(uid), MAX(timestamp)/1000 FROM data WHERE channel=?");

    while (query.executeStep()) {
        rowCountQuery.reset();
        rowCountQuery.bind(1, query.getColumn(0).getInt64());
        rowCountQuery.executeStep();
        auto channel = CreateChannelPrivate(query.getColumn(0).getInt64(), query.getColumn(1), query.getColumn(2));
        SetRecordCount(*channel, rowCountQuery.getColumn(0));
        if (!rowCountQuery.getColumn(1).isNull()) {
            SetLastRecordTime(*channel, std::chrono::system_clock::from_time_t(rowCountQuery.getColumn(1).getInt64()));
        }
        if (!query.getColumn(3).isNull()) {
            SetPrecision(*channel, query.getColumn(3).getDouble());
        }
    }
}

int TSqliteStorage::ReadDBVersion()
{
    if (!DB->tableExists("variables")) {
        return 0;
    }

    SQLite::Statement query(*DB, "SELECT value FROM variables WHERE name = 'db_version'");
    while (query.executeStep()) {
        return query.getColumn(0).getInt();
    }

    return 0;
}

void TSqliteStorage::UpdateDB(int prev_version)
{
    auto migrations = GetMigrations();
    if (WB_DB_VERSION > migrations.size()) {
        wb_throw(TBaseException, "No migration to new DB version");
    }

    if (prev_version > WB_DB_VERSION) {
        wb_throw(TBaseException, "Unsupported DB version. Please consider deleting DB file.");
    }

    SQLite::Transaction transaction(*DB);
    for (; static_cast<unsigned int>(prev_version) < migrations.size(); ++prev_version) {
        LOG(Info) << "Convert database from version " << prev_version;
        migrations[prev_version](*DB);
    }
    transaction.commit();
    DB->exec("VACUUM");
}

/**
 * Check if DB backup file exists
 */
bool TSqliteStorage::CheckBackupFile(const string& dbFile)
{
    string backup_file = dbFile + DB_BACKUP_FILE_EXTENSION;
    struct stat buffer;

    if (stat(backup_file.c_str(), &buffer) < 0) {
        return false;
    }

    return S_ISREG(buffer.st_mode);
}

/**
 * Create DB backup file from existing
 */
void TSqliteStorage::CreateBackupFile(const string& dbFile)
{
    LOG(Info) << "Creating backup file for DB";
    CopyFile(dbFile, BackupFileName(dbFile));
}

/**
 * Restore backup file
 */
void TSqliteStorage::RestoreBackupFile(const string& dbFile)
{
    LOG(Info) << "Restoring detected backup file for DB";
    CopyFile(BackupFileName(dbFile), dbFile);
}

/**
 * Remove backup file
 */
void TSqliteStorage::RemoveBackupFile(const string& dbFile)
{
    LOG(Info) << "Removing backup file";
    std::remove(BackupFileName(dbFile).c_str());
}

void TSqliteStorage::WriteChannel(TChannelInfo& channelInfo,
                                  const std::string& value,
                                  const std::string& minimum,
                                  const std::string& maximum,
                                  bool retained,
                                  std::chrono::system_clock::time_point time)
{
    std::lock_guard<std::mutex> lg(Mutex);
    if (!Transaction) {
        Transaction.reset(new SQLite::Transaction(*DB));
    }

    LOG(Debug) << "Resulting channel ID for this request is " << channelInfo.GetId();

    InsertRowQuery->clearBindings();
    InsertRowQuery->bind(1, channelInfo.GetId());
    InsertRowQuery->bind(2, value);
    if (!minimum.empty()) {
        InsertRowQuery->bind(3, minimum);
    }
    if (!maximum.empty()) {
        InsertRowQuery->bind(4, maximum);
    }
    InsertRowQuery->bind(5, retained ? 1 : 0);
    InsertRowQuery->bind(6, std::chrono::duration_cast<std::chrono::milliseconds>(time.time_since_epoch()).count());
    InsertRowQuery->exec();
    InsertRowQuery->reset();

    SetRecordCount(channelInfo, channelInfo.GetRecordCount() + 1);
    SetLastRecordTime(channelInfo, time);
}

void TSqliteStorage::Commit()
{
    std::lock_guard<std::mutex> lg(Mutex);
    if (Transaction) {
        Transaction->commit();
        Transaction.reset();
    }
}

PChannelInfo TSqliteStorage::CreateChannel(const TChannelName& channelName)
{
    LOG(Info) << "Creating channel " << channelName.Device << "/" << channelName.Control;

    SQLite::Statement query(*DB, "INSERT INTO channels (device, control) VALUES (?, ?) ");
    query.bindNoCopy(1, channelName.Device);
    query.bindNoCopy(2, channelName.Control);
    query.exec();

    return CreateChannelPrivate(DB->getLastInsertRowid(), channelName.Device, channelName.Control);
}

/**
 * @brief Set channel's precision. One must call Commit to finalize writing to
 * storage.
 */
void TSqliteStorage::SetChannelPrecision(TChannelInfo& channelInfo, double precision)
{
    if (precision == channelInfo.GetPrecision()) {
        return;
    }

    LOG(Debug) << "Set channel's " << channelInfo.GetName() << " precision to " << precision;

    SQLite::Statement query(*DB, "UPDATE channels SET precision = ? WHERE int_id = ?");
    query.bind(1, precision);
    query.bind(2, channelInfo.GetId());
    query.exec();
    SetPrecision(channelInfo, precision);
}

void TSqliteStorage::GetRecordsWithAveragingInterval(IRecordsVisitor& visitor,
                                                     const std::vector<TChannelName>& channels,
                                                     std::chrono::system_clock::time_point startTime,
                                                     std::chrono::system_clock::time_point endTime,
                                                     int64_t startId,
                                                     uint32_t maxRecords,
                                                     std::chrono::milliseconds minInterval)
{
    if (minInterval.count() > 0) {
        GetRecordsWithAverage(visitor, channels, startTime, endTime, startId, maxRecords, minInterval);
    } else {
        GetRecordsWithoutAverage(visitor, channels, startTime, endTime, startId, maxRecords);
    }
}

int TSqliteStorage::BindParams(SQLite::Statement& query,
                               int param_num,
                               const std::vector<int64_t>& channelIds,
                               std::chrono::system_clock::time_point startTime,
                               std::chrono::system_clock::time_point endTime,
                               int64_t startId)
{
    for (auto id: channelIds) {
        query.bind(++param_num, id);
    }
    query.bind(++param_num, duration_cast<milliseconds>(startTime.time_since_epoch()).count());
    query.bind(++param_num, duration_cast<milliseconds>(endTime.time_since_epoch()).count());
    query.bind(++param_num, startId);
    return param_num;
}

void TSqliteStorage::GetRecordsWithoutAverage(IRecordsVisitor& visitor,
                                              const std::vector<TChannelName>& channels,
                                              std::chrono::system_clock::time_point startTime,
                                              std::chrono::system_clock::time_point endTime,
                                              int64_t startId,
                                              uint32_t maxRecords)
{
    auto channelIds = GetChannelIds(channels);
    string queryStr;
    AddWithoutAverageQuery(queryStr, channelIds.size());
    queryStr += " ORDER BY uid ASC LIMIT ?";

    std::lock_guard<std::mutex> lg(Mutex);

    SQLite::Statement query(*DB, queryStr);
    int param_num = BindParams(query, 0, channelIds, startTime, endTime, startId);
    query.bind(++param_num, maxRecords);

    ProcessGetRecordsResult(query, visitor);
}

void TSqliteStorage::GetRecordsWithAverage(IRecordsVisitor& visitor,
                                           const std::vector<TChannelName>& channels,
                                           std::chrono::system_clock::time_point startTime,
                                           std::chrono::system_clock::time_point endTime,
                                           int64_t startId,
                                           uint32_t maxRecords,
                                           std::chrono::milliseconds minInterval)
{
    auto channelIds = GetChannelIds(channels);
    string queryStr;
    AddWithAverageQuery(queryStr, channelIds.size());
    queryStr += " ORDER BY uid ASC LIMIT ?";

    std::lock_guard<std::mutex> lg(Mutex);

    SQLite::Statement query(*DB, queryStr);
    int param_num = BindParams(query, 0, channelIds, startTime, endTime, startId);
    LOG(Debug) << "day: fraction :" << minInterval.count();
    query.bind(++param_num, minInterval.count());
    query.bind(++param_num, maxRecords);

    ProcessGetRecordsResult(query, visitor);
}

std::vector<int64_t> TSqliteStorage::GetChannelIds(const std::vector<TChannelName>& channels) const
{
    std::vector<int64_t> res;
    for (const auto& channel: channels) {
        auto pChannel = FindChannel(channel);
        if (pChannel) {
            res.push_back(pChannel->GetId());
        }
    }
    return res;
}

void TSqliteStorage::GetRecordsWithLimit(IRecordsVisitor& visitor,
                                         const std::vector<TChannelName>& channels,
                                         std::chrono::system_clock::time_point startTime,
                                         std::chrono::system_clock::time_point endTime,
                                         int64_t startId,
                                         uint32_t maxRecords,
                                         size_t overallRecordsLimit)
{
    std::vector<int64_t> withAverage;
    std::vector<int64_t> withoutAverage;

    auto channelIds = GetChannelIds(channels);
    for (const auto& ch: GetRecordsCount(channelIds, startTime, endTime)) {
        if (overallRecordsLimit > 0 && ch.second > overallRecordsLimit) {
            withAverage.emplace_back(ch.first);
        } else {
            withoutAverage.emplace_back(ch.first);
        }
    }

    string queryStr;
    if (!withoutAverage.empty()) {
        AddWithoutAverageQuery(queryStr, withoutAverage.size());
    }
    if (!withAverage.empty()) {
        if (!queryStr.empty()) {
            queryStr += " UNION ALL ";
        }
        AddWithAverageQuery(queryStr, withAverage.size());
    }

    if (queryStr.empty()) {
        // No channels to select
        return;
    }
    queryStr += " ORDER BY uid ASC LIMIT ?";

    std::lock_guard<std::mutex> lg(Mutex);

    SQLite::Statement query(*DB, queryStr);

    int param_num = 0;
    if (!withoutAverage.empty()) {
        param_num = BindParams(query, param_num, withoutAverage, startTime, endTime, startId);
    }
    if (!withAverage.empty()) {
        param_num = BindParams(query, param_num, withAverage, startTime, endTime, startId);
        auto minInterval =
            std::chrono::duration_cast<std::chrono::milliseconds>(endTime - startTime) / overallRecordsLimit;
        query.bind(++param_num, static_cast<int64_t>(minInterval.count()));
    }
    query.bind(++param_num, maxRecords);

    ProcessGetRecordsResult(query, visitor);
}

void TSqliteStorage::ProcessGetRecordsResult(SQLite::Statement& query, IRecordsVisitor& visitor) const
{
    std::unordered_map<int, PChannelInfo> channelIdToNameMap;
    for (const auto& ch: GetChannelsPrivate()) {
        channelIdToNameMap.insert({ch.second->GetId(), ch.second});
    }
    while (query.executeStep()) {
        int channelId(query.getColumn(CHANNEL_COLUMN).getInt());
        if (!CallVisitor(visitor, query, true, *channelIdToNameMap[channelId])) {
            return;
        }
    }
}

void TSqliteStorage::GetChannels(IChannelVisitor& visitor)
{
    std::lock_guard<std::mutex> lg(Mutex);
    for (const auto& channel: GetChannelsPrivate()) {
        visitor.ProcessChannel(channel.second);
    }
}

void TSqliteStorage::DeleteRecords(TChannelInfo& channel, uint32_t count)
{
    std::lock_guard<std::mutex> lg(Mutex);
    CleanChannelQuery->bind(1, channel.GetId());
    CleanChannelQuery->bind(2, count);
    auto deletedRows = CleanChannelQuery->exec();
    CleanChannelQuery->reset();
    SetRecordCount(channel, channel.GetRecordCount() - deletedRows);
    LOG(Debug) << "Clear channel id = " << channel.GetId();
}

void TSqliteStorage::DeleteRecords(const std::vector<std::reference_wrapper<TChannelInfo>>& channels, uint32_t count)
{
    auto ids = Join(channels.cbegin(), channels.cend(), [](const TChannelInfo& ch) { return ch.GetId(); }, ",");
    std::unordered_map<uint64_t, int> deletedRows;
    {
        std::stringstream queryText;
        queryText << "SELECT count(), channel FROM "
                  << "(SELECT channel FROM data WHERE channel in (" << ids << ") ORDER BY timestamp ASC LIMIT " << count
                  << ") "
                  << "GROUP BY channel";
        SQLite::Statement query(*DB, queryText.str());
        while (query.executeStep()) {
            deletedRows[query.getColumn(1).getInt64()] = query.getColumn(0).getInt();
        }
    }
    {
        std::stringstream queryText;
        queryText << "DELETE FROM data WHERE channel in (" << ids << ") ORDER BY timestamp ASC LIMIT " << count;
        SQLite::Statement query(*DB, queryText.str());
        query.exec();
    }

    for (TChannelInfo& channel: channels) {
        auto it = deletedRows.find(channel.GetId());
        if (it != deletedRows.end()) {
            SetRecordCount(channel, channel.GetRecordCount() - it->second);
        }
    }
}

int TSqliteStorage::GetDBVersion()
{
    return WB_DB_VERSION;
}

std::unordered_map<int64_t, size_t> TSqliteStorage::GetRecordsCount(const std::vector<int64_t>& channelIds,
                                                                    std::chrono::system_clock::time_point startTime,
                                                                    std::chrono::system_clock::time_point endTime)
{
    string queryStr;

    queryStr = "SELECT COUNT(*), channel FROM data INDEXED BY "
               "data_topic_timestamp WHERE ";
    AddCommonWhereClause(queryStr, channelIds.size());
    queryStr += " GROUP BY channel";

    std::lock_guard<std::mutex> lg(Mutex);
    SQLite::Statement query(*DB, queryStr);

    std::unordered_map<int64_t, size_t> res;
    int param_num = 0;
    for (auto id: channelIds) {
        res[id] = 0;
        query.bind(++param_num, id);
    }
    query.bind(++param_num, duration_cast<milliseconds>(startTime.time_since_epoch()).count());
    query.bind(++param_num, duration_cast<milliseconds>(endTime.time_since_epoch()).count());

    while (query.executeStep()) {
        res[query.getColumn(CHANNEL_COLUMN).getInt()] = query.getColumn(0).getInt();
    }
    return res;
}

1	#include "sqlite_storage.h"
2	#include "SQLiteCpp/SQLiteCpp.h"
3
4	#include <sqlite3.h>
5
6	#include <fstream>
7	#include <sys/stat.h>
8	#include <wblib/utils.h>
9
10	#include "db_migrations.h"
11	#include "log.h"
12	#include "utils.h"
13
14	using namespace std;
15	using namespace WBMQTT;
16	using namespace std::chrono;
17	using namespace Utils;
18
19	#define LOG(logger) ::logger.Log() << "[sqlite] "
20
21	namespace
22	{
23	const char* DB_BACKUP_FILE_EXTENSION = ".backup";
24	const int WB_DB_VERSION = 6;
25
26	// const int UNDEFINED_ID = -1;
27	const int CHANNEL_COLUMN = 1;
28
29	string BackupFileName(const string& filename)	×
30	{
31	return filename + DB_BACKUP_FILE_EXTENSION;	×
32	}
33
34	// Function to optimize database for better performance
35	void TuneDatabase(SQLite::Database& db)	11✔
36	{
37	// The WAL journaling mode uses a write-ahead log instead of a rollback
38	// journal to implement transactions.
39	// * WAL is significantly faster in most scenarios.
40	// * WAL uses many fewer fsync() operations and is thus less vulnerable
41	// to problems on systems where the fsync() system call is broken.
42	db.exec("PRAGMA journal_mode=WAL");	11✔
43
44	// In WAL mode when synchronous is NORMAL, the WAL file is synchronized before
45	// each checkpoint and the database file is synchronized after each completed
46	// checkpoint and the WAL file header is synchronized when a WAL file begins
47	// to be reused after a checkpoint, but no sync operations occur during most
48	// transactions
49	db.exec("PRAGMA synchronous=NORMAL");	11✔
50	}	11✔
51
52	} // namespace
53
54	TSqliteStorage::TSqliteStorage(const string& dbFile)	11✔
55	{
56	if (sqlite3_compileoption_used("ENABLE_UPDATE_DELETE_LIMIT") == 0) {	11✔
57	wb_throw(TBaseException,	×
58	"libsqlite3 is built without SQLITE_ENABLE_UPDATE_DELETE_LIMIT "
59	"required for DELETE ... ORDER BY ... LIMIT statements");
60	}
61
62	bool isMemoryDb = (dbFile.find(":memory:") != string::npos);	11✔
63
64	// check if backup file is present; if so, we should try to repair DB
65	if (!isMemoryDb && CheckBackupFile(dbFile)) {	11✔
66	LOG(Warn) << "Something went wrong last time, restoring old backup file";	×
67	RestoreBackupFile(dbFile);	×
68	}
69
70	int flags = SQLite::OPEN_READWRITE \| SQLite::OPEN_CREATE;	11✔
71	if (isMemoryDb) {	11✔
72	flags \|= SQLite::OPEN_URI;	11✔
73	}
74
75	DB = std::make_unique<SQLite::Database>(dbFile, flags);	11✔
76
77	TuneDatabase(*DB);	11✔
78
79	if (!DB->tableExists("data")) {	11✔
80	// new DB file created
81	LOG(Info) << "Creating tables";	10✔
82	CreateTables(WB_DB_VERSION);	10✔
83	} else {
84	int file_db_version = ReadDBVersion();	1✔
85	if (file_db_version > WB_DB_VERSION) {	1✔
86	wb_throw(TBaseException, "Database file is created by newer version of wb-mqtt-db");	×
87	}
88	if (file_db_version < WB_DB_VERSION) {	1✔
89	LOG(Warn) << "Old database format found, trying to update...";	×
90	CreateBackupFile(dbFile);	×
91	UpdateDB(file_db_version);	×
92	} else {
93	LOG(Info) << "Creating tables if necessary";	1✔
94	CreateTables(WB_DB_VERSION);	1✔
95	}
96	}
97
98	LOG(Info) << "Create indices if necessary";	11✔
99	CreateIndices();	11✔
100
101	LOG(Info) << "Analyzing data table";	11✔
102	DB->exec("ANALYZE data");	11✔
103	DB->exec("ANALYZE sqlite_master");	11✔
104
105	InsertRowQuery.reset(new SQLite::Statement(*DB,	11✔
106	"INSERT INTO data (channel, value, min, max, retained, timestamp) "
107	"VALUES (?, ?, ?, ?, ?, ?)"));	11✔
108
109	CleanChannelQuery.reset(	11✔
110	new SQLite::Statement(*DB, "DELETE FROM data WHERE channel = ? ORDER BY timestamp ASC LIMIT ?"));	11✔
111
112	LOG(Info) << "DB initialization is done";	11✔
113
114	if (!isMemoryDb && CheckBackupFile(dbFile)) {	11✔
115	RemoveBackupFile(dbFile);	×
116	}
117
118	Load();	11✔
119	}	11✔
120
121	void TSqliteStorage::CreateTables(int dbVersion)	11✔
122	{
123	LOG(Debug) << "Creating 'channels' table...";	11✔
124	DB->exec("CREATE TABLE IF NOT EXISTS channels ( "	11✔
125	"int_id INTEGER PRIMARY KEY AUTOINCREMENT, "
126	"device VARCHAR(255), "
127	"control VARCHAR(255), "
128	"precision REAL, "
129	"UNIQUE(device,control) "
130	") ");
131
132	LOG(Debug) << "Creating 'data' table...";	11✔
133	DB->exec("CREATE TABLE IF NOT EXISTS data ("	11✔
134	"uid INTEGER PRIMARY KEY AUTOINCREMENT, "
135	"channel INTEGER,"
136	"value VARCHAR(255),"
137	"timestamp INTEGER DEFAULT(0),"
138	"max VARCHAR(255),"
139	"min VARCHAR(255),"
140	"retained INTEGER"
141	")");
142
143	LOG(Debug) << "Creating 'variables' table...";	11✔
144	DB->exec("CREATE TABLE IF NOT EXISTS variables ("	11✔
145	"name VARCHAR(255) PRIMARY KEY, "
146	"value VARCHAR(255) )");
147
148	{
149	LOG(Debug) << "Updating database version variable...";	11✔
150	SQLite::Statement query(*DB,	11✔
151	"INSERT OR REPLACE INTO variables (name, "
152	"value) VALUES ('db_version', ?)");	11✔
153	query.bind(1, dbVersion);	11✔
154	query.exec();	11✔
155	}	11✔
156	}	11✔
157
158	void TSqliteStorage::CreateIndices()	11✔
159	{
160	LOG(Debug) << "Creating 'data_topic' index on 'data' ('channel')";	11✔
161	DB->exec("CREATE INDEX IF NOT EXISTS data_topic ON data (channel)");	11✔
162
163	// NOTE: the following index is a "low quality" one according to sqlite
164	// documentation. However, reversing the order of columns results in factor of
165	// two decrease in SELECT performance. So we leave it here as it is.
166	LOG(Debug) << "Creating 'data_topic_timestamp' index on 'data' ('channel', "	22✔
167	"'timestamp')";	11✔
168	DB->exec("CREATE INDEX IF NOT EXISTS data_topic_timestamp ON data (channel, "	11✔
169	"timestamp)");
170	}	11✔
171
172	void TSqliteStorage::Load()	11✔
173	{
174	std::lock_guard<std::mutex> lg(Mutex);	11✔
175	SQLite::Statement query(*DB, "SELECT int_id, device, control, precision FROM channels");	11✔
176	SQLite::Statement rowCountQuery(*DB, "SELECT COUNT(uid), MAX(timestamp)/1000 FROM data WHERE channel=?");	11✔
177
178	while (query.executeStep()) {	13✔
179	rowCountQuery.reset();	2✔
180	rowCountQuery.bind(1, query.getColumn(0).getInt64());	2✔
181	rowCountQuery.executeStep();	2✔
182	auto channel = CreateChannelPrivate(query.getColumn(0).getInt64(), query.getColumn(1), query.getColumn(2));	2✔
183	SetRecordCount(*channel, rowCountQuery.getColumn(0));	2✔
184	if (!rowCountQuery.getColumn(1).isNull()) {	2✔
185	SetLastRecordTime(*channel, std::chrono::system_clock::from_time_t(rowCountQuery.getColumn(1).getInt64()));	2✔
186	}
187	if (!query.getColumn(3).isNull()) {	2✔
188	SetPrecision(*channel, query.getColumn(3).getDouble());	1✔
189	}
190	}	2✔
191	}	11✔
192
193	int TSqliteStorage::ReadDBVersion()	1✔
194	{
195	if (!DB->tableExists("variables")) {	1✔
196	return 0;	×
197	}
198
199	SQLite::Statement query(*DB, "SELECT value FROM variables WHERE name = 'db_version'");	1✔
200	while (query.executeStep()) {	1✔
201	return query.getColumn(0).getInt();	1✔
202	}
203
204	return 0;	×
205	}	1✔
206
207	void TSqliteStorage::UpdateDB(int prev_version)	×
208	{
209	auto migrations = GetMigrations();	×
210	if (WB_DB_VERSION > migrations.size()) {	×
211	wb_throw(TBaseException, "No migration to new DB version");	×
212	}
213
214	if (prev_version > WB_DB_VERSION) {	×
215	wb_throw(TBaseException, "Unsupported DB version. Please consider deleting DB file.");	×
216	}
217
218	SQLite::Transaction transaction(*DB);	×
219	for (; static_cast<unsigned int>(prev_version) < migrations.size(); ++prev_version) {	×
220	LOG(Info) << "Convert database from version " << prev_version;	×
221	migrations[prev_version](*DB);	×
222	}
223	transaction.commit();	×
224	DB->exec("VACUUM");	×
225	}	×
226
227	/**
228	* Check if DB backup file exists
229	*/
230	bool TSqliteStorage::CheckBackupFile(const string& dbFile)	×
231	{
232	string backup_file = dbFile + DB_BACKUP_FILE_EXTENSION;	×
233	struct stat buffer;
234
235	if (stat(backup_file.c_str(), &buffer) < 0) {	×
236	return false;	×
237	}
238
239	return S_ISREG(buffer.st_mode);	×
240	}	×
241
242	/**
243	* Create DB backup file from existing
244	*/
245	void TSqliteStorage::CreateBackupFile(const string& dbFile)	×
246	{
247	LOG(Info) << "Creating backup file for DB";	×
248	CopyFile(dbFile, BackupFileName(dbFile));	×
249	}	×
250
251	/**
252	* Restore backup file
253	*/
254	void TSqliteStorage::RestoreBackupFile(const string& dbFile)	×
255	{
256	LOG(Info) << "Restoring detected backup file for DB";	×
257	CopyFile(BackupFileName(dbFile), dbFile);	×
258	}	×
259
260	/**
261	* Remove backup file
262	*/
263	void TSqliteStorage::RemoveBackupFile(const string& dbFile)	×
264	{
265	LOG(Info) << "Removing backup file";	×
266	std::remove(BackupFileName(dbFile).c_str());	×
267	}	×
268
269	void TSqliteStorage::WriteChannel(TChannelInfo& channelInfo,	72✔
270	const std::string& value,
271	const std::string& minimum,
272	const std::string& maximum,
273	bool retained,
274	std::chrono::system_clock::time_point time)
275	{
276	std::lock_guard<std::mutex> lg(Mutex);	72✔
277	if (!Transaction) {	72✔
278	Transaction.reset(new SQLite::Transaction(*DB));	9✔
279	}
280
281	LOG(Debug) << "Resulting channel ID for this request is " << channelInfo.GetId();	72✔
282
283	InsertRowQuery->clearBindings();	72✔
284	InsertRowQuery->bind(1, channelInfo.GetId());	72✔
285	InsertRowQuery->bind(2, value);	72✔
286	if (!minimum.empty()) {	72✔
287	InsertRowQuery->bind(3, minimum);	11✔
288	}
289	if (!maximum.empty()) {	72✔
290	InsertRowQuery->bind(4, maximum);	11✔
291	}
292	InsertRowQuery->bind(5, retained ? 1 : 0);	72✔
293	InsertRowQuery->bind(6, std::chrono::duration_cast<std::chrono::milliseconds>(time.time_since_epoch()).count());	72✔
294	InsertRowQuery->exec();	72✔
295	InsertRowQuery->reset();	72✔
296
297	SetRecordCount(channelInfo, channelInfo.GetRecordCount() + 1);	72✔
298	SetLastRecordTime(channelInfo, time);	72✔
299	}	72✔
300
301	void TSqliteStorage::Commit()	2✔
302	{
303	std::lock_guard<std::mutex> lg(Mutex);	2✔
304	if (Transaction) {	2✔
305	Transaction->commit();	1✔
306	Transaction.reset();	1✔
307	}
308	}	2✔
309
310	PChannelInfo TSqliteStorage::CreateChannel(const TChannelName& channelName)	14✔
311	{
312	LOG(Info) << "Creating channel " << channelName.Device << "/" << channelName.Control;	14✔
313
314	SQLite::Statement query(*DB, "INSERT INTO channels (device, control) VALUES (?, ?) ");	14✔
315	query.bindNoCopy(1, channelName.Device);	14✔
316	query.bindNoCopy(2, channelName.Control);	14✔
317	query.exec();	14✔
318
319	return CreateChannelPrivate(DB->getLastInsertRowid(), channelName.Device, channelName.Control);	28✔
320	}	14✔
321
322	/**
323	* @brief Set channel's precision. One must call Commit to finalize writing to
324	* storage.
325	*/
326	void TSqliteStorage::SetChannelPrecision(TChannelInfo& channelInfo, double precision)	2✔
327	{
328	if (precision == channelInfo.GetPrecision()) {	2✔
329	return;	×
330	}
331
332	LOG(Debug) << "Set channel's " << channelInfo.GetName() << " precision to " << precision;	2✔
333
334	SQLite::Statement query(*DB, "UPDATE channels SET precision = ? WHERE int_id = ?");	2✔
335	query.bind(1, precision);	2✔
336	query.bind(2, channelInfo.GetId());	2✔
337	query.exec();	2✔
338	SetPrecision(channelInfo, precision);	2✔
339	}	2✔
340
341	void TSqliteStorage::GetRecordsWithAveragingInterval(IRecordsVisitor& visitor,	9✔
342	const std::vector<TChannelName>& channels,
343	std::chrono::system_clock::time_point startTime,
344	std::chrono::system_clock::time_point endTime,
345	int64_t startId,
346	uint32_t maxRecords,
347	std::chrono::milliseconds minInterval)
348	{
349	if (minInterval.count() > 0) {	9✔
350	GetRecordsWithAverage(visitor, channels, startTime, endTime, startId, maxRecords, minInterval);	2✔
351	} else {
352	GetRecordsWithoutAverage(visitor, channels, startTime, endTime, startId, maxRecords);	7✔
353	}
354	}	9✔
355
356	int TSqliteStorage::BindParams(SQLite::Statement& query,	15✔
357	int param_num,
358	const std::vector<int64_t>& channelIds,
359	std::chrono::system_clock::time_point startTime,
360	std::chrono::system_clock::time_point endTime,
361	int64_t startId)
362	{
363	for (auto id: channelIds) {	36✔
364	query.bind(++param_num, id);	21✔
365	}
366	query.bind(++param_num, duration_cast<milliseconds>(startTime.time_since_epoch()).count());	15✔
367	query.bind(++param_num, duration_cast<milliseconds>(endTime.time_since_epoch()).count());	15✔
368	query.bind(++param_num, startId);	15✔
369	return param_num;	15✔
370	}
371
372	void TSqliteStorage::GetRecordsWithoutAverage(IRecordsVisitor& visitor,	7✔
373	const std::vector<TChannelName>& channels,
374	std::chrono::system_clock::time_point startTime,
375	std::chrono::system_clock::time_point endTime,
376	int64_t startId,
377	uint32_t maxRecords)
378	{
379	auto channelIds = GetChannelIds(channels);	7✔
380	string queryStr;	7✔
381	AddWithoutAverageQuery(queryStr, channelIds.size());	7✔
382	queryStr += " ORDER BY uid ASC LIMIT ?";	7✔
383
384	std::lock_guard<std::mutex> lg(Mutex);	7✔
385
386	SQLite::Statement query(*DB, queryStr);	7✔
387	int param_num = BindParams(query, 0, channelIds, startTime, endTime, startId);	7✔
388	query.bind(++param_num, maxRecords);	7✔
389
390	ProcessGetRecordsResult(query, visitor);	7✔
391	}	7✔
392
393	void TSqliteStorage::GetRecordsWithAverage(IRecordsVisitor& visitor,	2✔
394	const std::vector<TChannelName>& channels,
395	std::chrono::system_clock::time_point startTime,
396	std::chrono::system_clock::time_point endTime,
397	int64_t startId,
398	uint32_t maxRecords,
399	std::chrono::milliseconds minInterval)
400	{
401	auto channelIds = GetChannelIds(channels);	2✔
402	string queryStr;	2✔
403	AddWithAverageQuery(queryStr, channelIds.size());	2✔
404	queryStr += " ORDER BY uid ASC LIMIT ?";	2✔
405
406	std::lock_guard<std::mutex> lg(Mutex);	2✔
407
408	SQLite::Statement query(*DB, queryStr);	2✔
409	int param_num = BindParams(query, 0, channelIds, startTime, endTime, startId);	2✔
410	LOG(Debug) << "day: fraction :" << minInterval.count();	2✔
411	query.bind(++param_num, minInterval.count());	2✔
412	query.bind(++param_num, maxRecords);	2✔
413
414	ProcessGetRecordsResult(query, visitor);	2✔
415	}	2✔
416
417	std::vector<int64_t> TSqliteStorage::GetChannelIds(const std::vector<TChannelName>& channels) const	16✔
418	{
419	std::vector<int64_t> res;	16✔
420	for (const auto& channel: channels) {	39✔
421	auto pChannel = FindChannel(channel);	23✔
422	if (pChannel) {	23✔
423	res.push_back(pChannel->GetId());	21✔
424	}
425	}	23✔
426	return res;	16✔
427	}	×
428
429	void TSqliteStorage::GetRecordsWithLimit(IRecordsVisitor& visitor,	7✔
430	const std::vector<TChannelName>& channels,
431	std::chrono::system_clock::time_point startTime,
432	std::chrono::system_clock::time_point endTime,
433	int64_t startId,
434	uint32_t maxRecords,
435	size_t overallRecordsLimit)
436	{
437	std::vector<int64_t> withAverage;	7✔
438	std::vector<int64_t> withoutAverage;	7✔
439
440	auto channelIds = GetChannelIds(channels);	7✔
441	for (const auto& ch: GetRecordsCount(channelIds, startTime, endTime)) {	16✔
442	if (overallRecordsLimit > 0 && ch.second > overallRecordsLimit) {	9✔
443	withAverage.emplace_back(ch.first);	3✔
444	} else {
445	withoutAverage.emplace_back(ch.first);	6✔
446	}
447	}	7✔
448
449	string queryStr;	7✔
450	if (!withoutAverage.empty()) {	7✔
451	AddWithoutAverageQuery(queryStr, withoutAverage.size());	4✔
452	}
453	if (!withAverage.empty()) {	7✔
454	if (!queryStr.empty()) {	2✔
455	queryStr += " UNION ALL ";	×
456	}
457	AddWithAverageQuery(queryStr, withAverage.size());	2✔
458	}
459
460	if (queryStr.empty()) {	7✔
461	// No channels to select
462	return;	1✔
463	}
464	queryStr += " ORDER BY uid ASC LIMIT ?";	6✔
465
466	std::lock_guard<std::mutex> lg(Mutex);	6✔
467
468	SQLite::Statement query(*DB, queryStr);	6✔
469
470	int param_num = 0;	6✔
471	if (!withoutAverage.empty()) {	6✔
472	param_num = BindParams(query, param_num, withoutAverage, startTime, endTime, startId);	4✔
473	}
474	if (!withAverage.empty()) {	6✔
475	param_num = BindParams(query, param_num, withAverage, startTime, endTime, startId);	2✔
476	auto minInterval =
477	std::chrono::duration_cast<std::chrono::milliseconds>(endTime - startTime) / overallRecordsLimit;	2✔
478	query.bind(++param_num, static_cast<int64_t>(minInterval.count()));	2✔
479	}
480	query.bind(++param_num, maxRecords);	6✔
481
482	ProcessGetRecordsResult(query, visitor);	6✔
483	}	10✔
484
485	void TSqliteStorage::ProcessGetRecordsResult(SQLite::Statement& query, IRecordsVisitor& visitor) const	15✔
486	{
487	std::unordered_map<int, PChannelInfo> channelIdToNameMap;	15✔
488	for (const auto& ch: GetChannelsPrivate()) {	37✔
489	channelIdToNameMap.insert({ch.second->GetId(), ch.second});	22✔
490	}
491	while (query.executeStep()) {	80✔
492	int channelId(query.getColumn(CHANNEL_COLUMN).getInt());	65✔
493	if (!CallVisitor(visitor, query, true, *channelIdToNameMap[channelId])) {	65✔
494	return;	×
495	}
496	}
497	}	15✔
498
499	void TSqliteStorage::GetChannels(IChannelVisitor& visitor)	1✔
500	{
501	std::lock_guard<std::mutex> lg(Mutex);	1✔
502	for (const auto& channel: GetChannelsPrivate()) {	3✔
503	visitor.ProcessChannel(channel.second);	2✔
504	}
505	}	1✔
506
507	void TSqliteStorage::DeleteRecords(TChannelInfo& channel, uint32_t count)	4✔
508	{
509	std::lock_guard<std::mutex> lg(Mutex);	4✔
510	CleanChannelQuery->bind(1, channel.GetId());	4✔
511	CleanChannelQuery->bind(2, count);	4✔
512	auto deletedRows = CleanChannelQuery->exec();	4✔
513	CleanChannelQuery->reset();	4✔
514	SetRecordCount(channel, channel.GetRecordCount() - deletedRows);	4✔
515	LOG(Debug) << "Clear channel id = " << channel.GetId();	4✔
516	}	4✔
517
518	void TSqliteStorage::DeleteRecords(const std::vector<std::reference_wrapper<TChannelInfo>>& channels, uint32_t count)	1✔
519	{
520	auto ids = Join(channels.cbegin(), channels.cend(), [](const TChannelInfo& ch) { return ch.GetId(); }, ",");	4✔
521	std::unordered_map<uint64_t, int> deletedRows;	1✔
522	{
523	std::stringstream queryText;	1✔
524	queryText << "SELECT count(), channel FROM "
525	<< "(SELECT channel FROM data WHERE channel in (" << ids << ") ORDER BY timestamp ASC LIMIT " << count	1✔
526	<< ") "
527	<< "GROUP BY channel";	1✔
528	SQLite::Statement query(*DB, queryText.str());	1✔
529	while (query.executeStep()) {	3✔
530	deletedRows[query.getColumn(1).getInt64()] = query.getColumn(0).getInt();	2✔
531	}
532	}	1✔
533	{
534	std::stringstream queryText;	1✔
535	queryText << "DELETE FROM data WHERE channel in (" << ids << ") ORDER BY timestamp ASC LIMIT " << count;	1✔
536	SQLite::Statement query(*DB, queryText.str());	1✔
537	query.exec();	1✔
538	}	1✔
539
540	for (TChannelInfo& channel: channels) {	3✔
541	auto it = deletedRows.find(channel.GetId());	2✔
542	if (it != deletedRows.end()) {	2✔
543	SetRecordCount(channel, channel.GetRecordCount() - it->second);	2✔
544	}
545	}
546	}	1✔
547
548	int TSqliteStorage::GetDBVersion()	1✔
549	{
550	return WB_DB_VERSION;	1✔
551	}
552
553	std::unordered_map<int64_t, size_t> TSqliteStorage::GetRecordsCount(const std::vector<int64_t>& channelIds,	7✔
554	std::chrono::system_clock::time_point startTime,
555	std::chrono::system_clock::time_point endTime)
556	{
557	string queryStr;	7✔
558
559	queryStr = "SELECT COUNT(*), channel FROM data INDEXED BY "
560	"data_topic_timestamp WHERE ";	7✔
561	AddCommonWhereClause(queryStr, channelIds.size());	7✔
562	queryStr += " GROUP BY channel";	7✔
563
564	std::lock_guard<std::mutex> lg(Mutex);	7✔
565	SQLite::Statement query(*DB, queryStr);	7✔
566
567	std::unordered_map<int64_t, size_t> res;	7✔
568	int param_num = 0;	7✔
569	for (auto id: channelIds) {	16✔
570	res[id] = 0;	9✔
571	query.bind(++param_num, id);	9✔
572	}
573	query.bind(++param_num, duration_cast<milliseconds>(startTime.time_since_epoch()).count());	7✔
574	query.bind(++param_num, duration_cast<milliseconds>(endTime.time_since_epoch()).count());	7✔
575
576	while (query.executeStep()) {	16✔
577	res[query.getColumn(CHANNEL_COLUMN).getInt()] = query.getColumn(0).getInt();	9✔
578	}
579	return res;	7✔
580	}	7✔

wirenboard / wb-mqtt-db / 96

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous