wirenboard / wb-mqtt-db / 1

Committed 09 Oct 2026 07:06AM UTC coverage: 79.788% (-2.2%) from 82.03%

Build # 1

Build Type

Pull #70

github

Specific Base 4957c0

Committed by

sikmir

Commit Message

Fix boolean value aggregation

Pull Request Pull Request #70: Fix boolean value aggregation

Coverage Stats

2465 of 6264 branches covered (39.35%)

87 of 87 new or added lines in 2 files covered. (100.0%)

146 existing lines in 8 files now uncovered.

1654 of 2073 relevant lines covered (79.79%)

20.31 hits per line

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

84.77

/src/sqlite_storage.cpp

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

#include <sqlite3.h> // for sqlite3_compileoption_used

#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> // for sqlite3_compileoption_used
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	{
UNCOV 31	return filename + DB_BACKUP_FILE_EXTENSION;	×
32	}
33
34	// Function to optimize database for better performance
35	void TuneDatabase(SQLite::Database& db)	19 ✔
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");	19 ✔
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");	19 ✔
50	}	19 ✔
51
52	} // namespace
53
54	TSqliteStorage::TSqliteStorage(const string& dbFile)	19 ✔
55	{
56	if (sqlite3_compileoption_used("ENABLE_UPDATE_DELETE_LIMIT") == 0) {	19 !
UNCOV 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);	19 ✔
63
64	// check if backup file is present; if so, we should try to repair DB
65	if (!isMemoryDb && CheckBackupFile(dbFile)) {	19 !
UNCOV 66	LOG(Warn) << "Something went wrong last time, restoring old backup file";	×
UNCOV 67	RestoreBackupFile(dbFile);	×
68	}
69
70	int flags = SQLite::OPEN_READWRITE \| SQLite::OPEN_CREATE;	19 ✔
71	if (isMemoryDb) {	19 !
72	flags \|= SQLite::OPEN_URI;	19 ✔
73	}
74
75	DB = std::make_unique<SQLite::Database>(dbFile, flags);	19 !
76
77	TuneDatabase(*DB);	19 !
78
79	if (!DB->tableExists("data")) {	19 !
80	// new DB file created
81	LOG(Info) << "Creating tables";	18 !
82	CreateTables(WB_DB_VERSION);	18 !
83	} else {
84	int file_db_version = ReadDBVersion();	1 !
85	if (file_db_version > WB_DB_VERSION) {	1 !
UNCOV 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 !
UNCOV 89	LOG(Warn) << "Old database format found, trying to update...";	×
UNCOV 90	CreateBackupFile(dbFile);	×
UNCOV 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";	19 !
99	CreateIndices();	19 !
100
101	LOG(Info) << "Analyzing data table";	19 !
102	DB->exec("ANALYZE data");	19 !
103	DB->exec("ANALYZE sqlite_master");	19 !
104
105	InsertRowQuery.reset(new SQLite::Statement(*DB,	19 !
106	"INSERT INTO data (channel, value, min, max, retained, timestamp) "
107	"VALUES (?, ?, ?, ?, ?, ?)"));	19 !
108
109	CleanChannelQuery.reset(	19 !
110	new SQLite::Statement(*DB, "DELETE FROM data WHERE channel = ? ORDER BY timestamp ASC LIMIT ?"));	19 !
111
112	LOG(Info) << "DB initialization is done";	19 !
113
114	if (!isMemoryDb && CheckBackupFile(dbFile)) {	19 !
UNCOV 115	RemoveBackupFile(dbFile);	×
116	}
117
118	Load();	19 !
119	}	19 ✔
120
121	void TSqliteStorage::CreateTables(int dbVersion)	19 ✔
122	{
123	LOG(Debug) << "Creating 'channels' table...";	19 !
124	DB->exec("CREATE TABLE IF NOT EXISTS channels ( "	19 ✔
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...";	19 !
133	DB->exec("CREATE TABLE IF NOT EXISTS data ("	19 ✔
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...";	19 !
144	DB->exec("CREATE TABLE IF NOT EXISTS variables ("	19 ✔
145	"name VARCHAR(255) PRIMARY KEY, "
146	"value VARCHAR(255) )");
147
148	{
149	LOG(Debug) << "Updating database version variable...";	19 !
150	SQLite::Statement query(*DB,	19 ✔
151	"INSERT OR REPLACE INTO variables (name, "
152	"value) VALUES ('db_version', ?)");	19 !
153	query.bind(1, dbVersion);	19 !
154	query.exec();	19 !
155	}	19 ✔
156	}	19 ✔
157
158	void TSqliteStorage::CreateIndices()	19 ✔
159	{
160	LOG(Debug) << "Creating 'data_topic' index on 'data' ('channel')";	19 !
161	DB->exec("CREATE INDEX IF NOT EXISTS data_topic ON data (channel)");	19 ✔
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', "	38 !
167	"'timestamp')";	19 !
168	DB->exec("CREATE INDEX IF NOT EXISTS data_topic_timestamp ON data (channel, "	19 ✔
169	"timestamp)");
170	}	19 ✔
171
172	void TSqliteStorage::Load()	19 ✔
173	{
174	std::lock_guard<std::mutex> lg(Mutex);	19 !
175	SQLite::Statement query(*DB, "SELECT int_id, device, control, precision FROM channels");	19 !
176	SQLite::Statement rowCountQuery(*DB, "SELECT COUNT(uid), MAX(timestamp)/1000 FROM data WHERE channel=?");	19 !
177
178	while (query.executeStep()) {	21 !
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	}	19 ✔
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
UNCOV 204	return 0;	×
205	}	1 ✔
206
207	void TSqliteStorage::UpdateDB(int prev_version)	×
208	{
UNCOV 209	auto migrations = GetMigrations();	×
210	if (WB_DB_VERSION > migrations.size()) {	×
211	wb_throw(TBaseException, "No migration to new DB version");	×
212	}
213
UNCOV 214	if (prev_version > WB_DB_VERSION) {	×
215	wb_throw(TBaseException, "Unsupported DB version. Please consider deleting DB file.");	×
216	}
217
UNCOV 218	SQLite::Transaction transaction(*DB);	×
UNCOV 219	for (; static_cast<unsigned int>(prev_version) < migrations.size(); ++prev_version) {	×
UNCOV 220	LOG(Info) << "Convert database from version " << prev_version;	×
UNCOV 221	migrations[prev_version](*DB);	×
222	}
UNCOV 223	transaction.commit();	×
224	DB->exec("VACUUM");	×
UNCOV 225	}	×
226
227	/**
228	* Check if DB backup file exists
229	*/
UNCOV 230	bool TSqliteStorage::CheckBackupFile(const string& dbFile)	×
231	{
UNCOV 232	string backup_file = dbFile + DB_BACKUP_FILE_EXTENSION;	×
233	struct stat buffer;
234
UNCOV 235	if (stat(backup_file.c_str(), &buffer) < 0) {	×
UNCOV 236	return false;	×
237	}
238
239	return S_ISREG(buffer.st_mode);	×
240	}	×
241
242	/**
243	* Create DB backup file from existing
244	*/
UNCOV 245	void TSqliteStorage::CreateBackupFile(const string& dbFile)	×
246	{
UNCOV 247	LOG(Info) << "Creating backup file for DB";	×
248	CopyFile(dbFile, BackupFileName(dbFile));	×
249	}	×
250
251	/**
252	* Restore backup file
253	*/
UNCOV 254	void TSqliteStorage::RestoreBackupFile(const string& dbFile)	×
255	{
UNCOV 256	LOG(Info) << "Restoring detected backup file for DB";	×
257	CopyFile(BackupFileName(dbFile), dbFile);	×
258	}	×
259
260	/**
261	* Remove backup file
262	*/
UNCOV 263	void TSqliteStorage::RemoveBackupFile(const string& dbFile)	×
264	{
UNCOV 265	LOG(Info) << "Removing backup file";	×
UNCOV 266	std::remove(BackupFileName(dbFile).c_str());	×
UNCOV 267	}	×
268
269	void TSqliteStorage::WriteChannel(TChannelInfo& channelInfo,	124 ✔
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);	124 !
277	if (!Transaction) {	124 ✔
278	Transaction.reset(new SQLite::Transaction(*DB));	23 !
279	}
280
281	LOG(Debug) << "Resulting channel ID for this request is " << channelInfo.GetId();	124 !
282
283	InsertRowQuery->clearBindings();	124 !
284	InsertRowQuery->bind(1, channelInfo.GetId());	124 !
285	InsertRowQuery->bind(2, value);	124 !
286	if (!minimum.empty()) {	124 ✔
287	InsertRowQuery->bind(3, minimum);	11 !
288	}
289	if (!maximum.empty()) {	124 ✔
290	InsertRowQuery->bind(4, maximum);	11 !
291	}
292	InsertRowQuery->bind(5, retained ? 1 : 0);	124 !
293	InsertRowQuery->bind(6, std::chrono::duration_cast<std::chrono::milliseconds>(time.time_since_epoch()).count());	124 !
294	InsertRowQuery->exec();	124 !
295	InsertRowQuery->reset();	124 !
296
297	SetRecordCount(channelInfo, channelInfo.GetRecordCount() + 1);	124 !
298	SetLastRecordTime(channelInfo, time);	124 !
299	}	124 ✔
300
301	void TSqliteStorage::Commit()	28 ✔
302	{
303	std::lock_guard<std::mutex> lg(Mutex);	28 !
304	if (Transaction) {	28 ✔
305	Transaction->commit();	15 !
306	Transaction.reset();	15 ✔
307	}
308	}	28 ✔
309
310	PChannelInfo TSqliteStorage::CreateChannel(const TChannelName& channelName)	36 ✔
311	{
312	LOG(Info) << "Creating channel " << channelName.Device << "/" << channelName.Control;	36 !
313
314	SQLite::Statement query(*DB, "INSERT INTO channels (device, control) VALUES (?, ?) ");	36 !
315	query.bindNoCopy(1, channelName.Device);	36 !
316	query.bindNoCopy(2, channelName.Control);	36 !
317	query.exec();	36 !
318
319	return CreateChannelPrivate(DB->getLastInsertRowid(), channelName.Device, channelName.Control);	72 !
320	}	36 ✔
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)	54 ✔
327	{
328	if (precision == channelInfo.GetPrecision()) {	54 !
329	return;	30 ✔
330	}
331
332	LOG(Debug) << "Set channel's " << channelInfo.GetName() << " precision to " << precision;	24 !
333
334	SQLite::Statement query(*DB, "UPDATE channels SET precision = ? WHERE int_id = ?");	24 !
335	query.bind(1, precision);	24 !
336	query.bind(2, channelInfo.GetId());	24 !
337	query.exec();	24 !
338	SetPrecision(channelInfo, precision);	24 !
339	}	24 ✔
340
341	void TSqliteStorage::GetRecordsWithAveragingInterval(IRecordsVisitor& visitor,	15 ✔
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) {	15 ✔
350	GetRecordsWithAverage(visitor, channels, startTime, endTime, startId, maxRecords, minInterval);	2 ✔
351	} else {
352	GetRecordsWithoutAverage(visitor, channels, startTime, endTime, startId, maxRecords);	13 ✔
353	}
354	}	15 ✔
355
356	int TSqliteStorage::BindParams(SQLite::Statement& query,	21 ✔
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) {	48 ✔
364	query.bind(++param_num, id);	27 !
365	}
366	query.bind(++param_num, duration_cast<milliseconds>(startTime.time_since_epoch()).count());	21 !
367	query.bind(++param_num, duration_cast<milliseconds>(endTime.time_since_epoch()).count());	21 !
368	query.bind(++param_num, startId);	21 ✔
369	return param_num;	21 ✔
370	}
371
372	void TSqliteStorage::GetRecordsWithoutAverage(IRecordsVisitor& visitor,	13 ✔
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);	13 !
380	string queryStr;	13 ✔
381	AddWithoutAverageQuery(queryStr, channelIds.size());	13 !
382	queryStr += " ORDER BY uid ASC LIMIT ?";	13 !
383
384	std::lock_guard<std::mutex> lg(Mutex);	13 !
385
386	SQLite::Statement query(*DB, queryStr);	13 !
387	int param_num = BindParams(query, 0, channelIds, startTime, endTime, startId);	13 !
388	query.bind(++param_num, maxRecords);	13 !
389
390	ProcessGetRecordsResult(query, visitor);	13 !
391	}	13 ✔
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	22 ✔
418	{
419	std::vector<int64_t> res;	22 ✔
420	for (const auto& channel: channels) {	51 ✔
421	auto pChannel = FindChannel(channel);	29 !
422	if (pChannel) {	29 ✔
423	res.push_back(pChannel->GetId());	27 !
424	}
425	}	29 ✔
426	return res;	22 ✔
UNCOV 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 !
UNCOV 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	21 ✔
486	{
487	std::unordered_map<int, PChannelInfo> channelIdToNameMap;	21 ✔
488	for (const auto& ch: GetChannelsPrivate()) {	49 !
489	channelIdToNameMap.insert({ch.second->GetId(), ch.second});	28 !
490	}
491	while (query.executeStep()) {	116 !
492	int channelId(query.getColumn(CHANNEL_COLUMN).getInt());	95 !
493	if (!CallVisitor(visitor, query, true, *channelIdToNameMap[channelId])) {	95 !
UNCOV 494	return;	×
495	}
496	}
497	}	21 !
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)	3 ✔
519	{
520	auto ids = Join(channels.cbegin(), channels.cend(), [](const TChannelInfo& ch) { return ch.GetId(); }, ",");	24 !
521	std::unordered_map<uint64_t, int> deletedRows;	3 ✔
522	{
523	std::stringstream queryText;	3 !
524	queryText << "SELECT count(), channel FROM "
525	<< "(SELECT channel FROM data WHERE channel in (" << ids << ") ORDER BY timestamp ASC LIMIT " << count	3 !
526	<< ") "
527	<< "GROUP BY channel";	3 !
528	SQLite::Statement query(*DB, queryText.str());	3 !
529	while (query.executeStep()) {	15 !
530	deletedRows[query.getColumn(1).getInt64()] = query.getColumn(0).getInt();	12 !
531	}
532	}	3 ✔
533	{
534	std::stringstream queryText;	3 !
535	queryText << "DELETE FROM data WHERE channel in (" << ids << ") ORDER BY timestamp ASC LIMIT " << count;	3 !
536	SQLite::Statement query(*DB, queryText.str());	3 !
537	query.exec();	3 !
538	}	3 ✔
539
540	for (TChannelInfo& channel: channels) {	21 ✔
541	auto it = deletedRows.find(channel.GetId());	18 !
542	if (it != deletedRows.end()) {	18 ✔
543	SetRecordCount(channel, channel.GetRecordCount() - it->second);	12 !
544	}
545	}
546	}	3 ✔
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 / 1

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