diff options
| -rw-r--r-- | src/ingestor.cpp | 15 | ||||
| -rw-r--r-- | src/ingestor.hpp | 2 | ||||
| -rw-r--r-- | src/schema.sql | 3 | ||||
| -rw-r--r-- | src/sql/selectUninsertableLines.sql | 3 |
4 files changed, 14 insertions, 9 deletions
diff --git a/src/ingestor.cpp b/src/ingestor.cpp index 077f7d8..c06fd1c 100644 --- a/src/ingestor.cpp +++ b/src/ingestor.cpp @@ -345,12 +345,13 @@ namespace WebStat { auto deleteLine = dbconn->modify(SQL::DELETE_ENTITY, SQL::DELETE_ENTITY_OPTS); auto setEntityUnparsable = dbconn->modify(SQL::SET_ENTITY_TYPE, SQL::SET_ENTITY_TYPE_OPTS); setEntityUnparsable->bindParamS(0, "unparsable_line"); + auto insert = dbconn->modify(SQL::ACCESS_LOG_INSERT, SQL::ACCESS_LOG_INSERT_OPTS); unsigned int stored = 0; while (!terminated) { unsigned int batchSize = 0; DB::TransactionScope batchTx {*dbconn}; - for (auto [id, line] : lineSelect->as<EntityId, std::string>()) { + for (auto [id, line, originalHostnameId] : lineSelect->as<EntityId, std::string, EntityId>()) { batchSize += 1; try { DB::TransactionScope lineTx {*dbconn}; @@ -359,8 +360,9 @@ namespace WebStat { auto valuesEntities = entities(values); fillKnownEntities(valuesEntities); storeNewEntities(dbconn, valuesEntities); + insert->bindParam(0, originalHostnameId); + storeLogLine(insert.get(), values); existingEntities()->insert_range(valuesEntities | ENTITY_IDS); - storeLogLine(dbconn, values); deleteLine->bindParamI(0, id); deleteLine->execute(); @@ -411,6 +413,8 @@ namespace WebStat { Ingestor::ingestLogLines(DB::Connection * dbconn, const LinesView lines) { DB::TransactionScope batchTx {*dbconn}; + auto insert = dbconn->modify(SQL::ACCESS_LOG_INSERT, SQL::ACCESS_LOG_INSERT_OPTS); + insert->bindParam(0, hostnameId); for (const auto & line : lines) { if (auto result = scanLogLine(line)) { stats.linesParsed++; @@ -420,7 +424,7 @@ namespace WebStat { try { DB::TransactionScope lineTx {*dbconn}; storeNewEntities(dbconn, valuesEntities); - storeLogLine(dbconn, values); + storeLogLine(insert.get(), values); existingEntities()->insert_range(valuesEntities | ENTITY_IDS); } catch (const DB::Error & originalError) { @@ -678,11 +682,8 @@ namespace WebStat { template<typename... T> void - Ingestor::storeLogLine(DB::Connection * dbconn, const std::tuple<T...> & values) const + Ingestor::storeLogLine(DB::ModifyCommand * insert, const std::tuple<T...> & values) const { - auto insert = dbconn->modify(SQL::ACCESS_LOG_INSERT, SQL::ACCESS_LOG_INSERT_OPTS); - - insert->bindParam(0, hostnameId); std::apply( [&insert](auto &&... value) { unsigned int param = 1; diff --git a/src/ingestor.hpp b/src/ingestor.hpp index dacfdf7..1a47a1b 100644 --- a/src/ingestor.hpp +++ b/src/ingestor.hpp @@ -81,7 +81,7 @@ namespace WebStat { Job::Result jobStoreQueuedLines(); Job::Result jobRetryUninsertableLines(); - template<typename... T> void storeLogLine(DB::Connection *, const std::tuple<T...> &) const; + template<typename... T> void storeLogLine(DB::ModifyCommand * insert, const std::tuple<T...> &) const; IngestorSettings settings; diff --git a/src/schema.sql b/src/schema.sql index 40cab51..fd2f3c5 100644 --- a/src/schema.sql +++ b/src/schema.sql @@ -94,6 +94,9 @@ CREATE INDEX idx_entities_retryinsert ON bad_lines(id) WHERE type = 'uninsertable_line' AND detail ->> 'retriedAt' IS NULL; +ALTER TABLE bad_lines + ADD CONSTRAINT bad_lines_have_hostnameid CHECK (detail ? 'hostnameId'); + CREATE OR REPLACE FUNCTION entity(newValue text, newType entity) RETURNS TABLE( id integer, diff --git a/src/sql/selectUninsertableLines.sql b/src/sql/selectUninsertableLines.sql index 060e800..2a57683 100644 --- a/src/sql/selectUninsertableLines.sql +++ b/src/sql/selectUninsertableLines.sql @@ -1,6 +1,7 @@ SELECT id, - value + value, + cast(detail ->> 'hostnameId' AS int) AS hostnameId FROM entities WHERE |