Rebase fixes.

Author: jgates108

src/CMakeLists.txt

@@ -62,7 +62,6 @@ add_subdirectory(css)
 add_subdirectory(czar)
 add_subdirectory(global)
 add_subdirectory(http)
-add_subdirectory(memman)
 add_subdirectory(mimic)
 add_subdirectory(mysql)
 add_subdirectory(parser)

src/admin/templates/xrootd/etc/xrdssi.cf.jinja

@@ -11,19 +11,6 @@ socket = {{ db_socket }}
 username = qsmaster
 password = CHANGEME2
 
-[memman]
-
-# MemMan class to use for managing memory for tables
-# can be "MemManReal", "MemManNone" or "MemManNoneRelaxed"
-# class = MemManReal
-
-# Memory available for locking tables, in MB
-# memory = 1000
-memory = 7900
-
-# Path to database tables
-location = /qserv/data/mysql
-
 [scheduler]
 
 # Thread pool size
@@ -79,13 +66,6 @@ maxsqlconn = 980
 # This value must be less than maxsqlconn. 
 reservedinteractivesqlconn = 930
 
-[transmits]
-# Maximum number of concurrent transmits to a czar.
-maxtransmits = 50
-# If more than this number of large transmits is happening at once, wait to
-# start more transmits until some are done.
-maxalreadytransmitting = 10
-
 [results]
 
 # The name of a folder where query results will be stored.

src/cconfig/CzarConfig.h

@@ -124,7 +124,6 @@ class CzarConfig {
     /// Getters for result aggregation options.
     int getMaxTableSizeMB() const { return _maxTableSizeMB->getVal(); }
     int getMaxSqlConnectionAttempts() const { return _maxSqlConnectionAttempts->getVal(); }
-    int getResultMaxConnections() const { return _resultMaxConnections->getVal(); }
 
     /// The size of the TCP connection pool witin the client API that is used
     /// by the merger to pool result files from workers via the HTTP protocol.
@@ -285,8 +284,6 @@ class CzarConfig {
             util::ConfigValTInt::create(_configValMap, "resultdb", "maxtablesize_mb", notReq, 5001);
     CVTIntPtr _maxSqlConnectionAttempts =
             util::ConfigValTInt::create(_configValMap, "resultdb", "maxsqlconnectionattempts", notReq, 10);
-    CVTIntPtr _resultMaxConnections =
-            util::ConfigValTInt::create(_configValMap, "resultdb", "maxconnections", notReq, 40);
     CVTIntPtr _resultMaxHttpConnections =
             util::ConfigValTInt::create(_configValMap, "resultdb", "maxhttpconnections", notReq, 2000);
     CVTIntPtr _oldestResultKeptDays =

src/ccontrol/CMakeLists.txt

@@ -12,9 +12,7 @@ target_sources(ccontrol PRIVATE
     ParseRunner.cc
     QueryState.cc
     UserQueryAsyncResult.cc
-    UserQueryDrop.cc
     UserQueryFactory.cc
-    UserQueryFlushChunksCache.cc
     UserQueryProcessList.cc
     UserQueryQueries.cc
     UserQuerySelectCountStar.cc

src/ccontrol/MergingHandler.cc

@@ -74,8 +74,12 @@ class MergingHandler : public qdisp::ResponseHandler {
 
     /// @see ResponseHandler::flushHttp
     /// @see MerginHandler::_mergeHttp
-    std::tuple<bool, bool> flushHttp(std::string const& fileUrl, uint64_t expectedRows,
-                                     uint64_t& resultRows) override;
+    std::tuple<bool, bool> flushHttp(
+            std::string const& fileUrl, uint64_t fileSize, uint64_t expectedRows,
+            /* &&&
+                std::tuple<bool, bool> flushHttp(std::string const& fileUrl, uint64_t expectedRows,
+            >>>>>>> a27525c04017db9a30061fa0bb4b5228c0c5d1b2 */
+            uint64_t& resultRows) override;
 
     /// @see ResponseHandler::flushHttpError
     void flushHttpError(int errorCode, std::string const& errorMsg, int status) override;
@@ -88,7 +92,9 @@ class MergingHandler : public qdisp::ResponseHandler {
 
 private:
     /// Call InfileMerger to do the work of merging this data to the result.
-    bool _mergeHttp(std::shared_ptr<qdisp::UberJob> const& uberJob, proto::ResponseData const& responseData);
+
+    bool _mergeHttp(std::shared_ptr<qdisp::UberJob> const& uberJob, std::string const& fileUrl,
+                    uint64_t fileSize);
 
     /// Set error code and string.
     void _setError(int code, std::string const& msg, int errorState);

src/ccontrol/MergingHandler.h

@@ -31,11 +31,13 @@
 #include "lsst/log/Log.h"
 
 // Qserv headers
+#include "cconfig/CzarConfig.h"
 #include "qmeta/Exceptions.h"
 #include "qmeta/JobStatus.h"
 #include "qmeta/QMeta.h"
 #include "qmeta/MessageStore.h"
 #include "sql/SqlConnection.h"
+#include "sql/SqlConnectionFactory.h"
 #include "sql/SqlResults.h"
 
 namespace {
@@ -46,13 +48,11 @@ namespace lsst::qserv::ccontrol {
 
 // Constructors
 UserQueryAsyncResult::UserQueryAsyncResult(QueryId queryId, qmeta::CzarId qMetaCzarId,
-                                           std::shared_ptr<qmeta::QMeta> const& qMeta,
-                                           sql::SqlConnection* resultDbConn)
+                                           std::shared_ptr<qmeta::QMeta> const& qMeta)
         : UserQuery(),
           _queryId(queryId),
           _qMetaCzarId(qMetaCzarId),
           _qMeta(qMeta),
-          _resultDbConn(resultDbConn),
           _messageStore(std::make_shared<qmeta::MessageStore>()) {
     LOGS(_log, LOG_LVL_DEBUG, "UserQueryAsyncResult: QID=" << queryId);
 
@@ -122,9 +122,11 @@ void UserQueryAsyncResult::submit() {
     std::string const resultTableName = _qInfo.resultLocation().substr(6);
 
     // check that message and result tables exist
+    auto const czarConfig = cconfig::CzarConfig::instance();
+    auto const resultDbConn = sql::SqlConnectionFactory::make(czarConfig->getMySqlResultConfig());
     sql::SqlErrorObject sqlErrObj;
-    if (!_resultDbConn->tableExists(_qInfo.msgTableName(), sqlErrObj) or
-        !_resultDbConn->tableExists(resultTableName, sqlErrObj)) {
+    if (!resultDbConn->tableExists(_qInfo.msgTableName(), sqlErrObj) or
+        !resultDbConn->tableExists(resultTableName, sqlErrObj)) {
         std::string message = "Result or message table does not exist, result is likely expired.";
         LOGS(_log, LOG_LVL_DEBUG, message);
         _messageStore->addErrorMessage("SYSTEM", message);
@@ -135,7 +137,7 @@ void UserQueryAsyncResult::submit() {
     // into the message store, at this point original result table must be unlocked
     std::string query = "SELECT chunkId, code, message, severity, timeStamp FROM " + _qInfo.msgTableName();
     sql::SqlResults sqlResults;
-    if (!_resultDbConn->runQuery(query, sqlResults, sqlErrObj)) {
+    if (!resultDbConn->runQuery(query, sqlResults, sqlErrObj)) {
         LOGS(_log, LOG_LVL_ERROR, "Failed to retrieve message table data: " << sqlErrObj.errMsg());
         std::string message = "Failed to retrieve message table data.";
         _messageStore->addErrorMessage("SYSTEM_SQL", message);
@@ -170,7 +172,7 @@ void UserQueryAsyncResult::submit() {
     // of results I'm going to drop this table now, meaning result can be only
     // retrieved once.
     query = "DROP TABLE " + _qInfo.msgTableName();
-    if (!_resultDbConn->runQuery(query, sqlErrObj)) {
+    if (!resultDbConn->runQuery(query, sqlErrObj)) {
         LOGS(_log, LOG_LVL_ERROR, "Failed to drop message table: " << sqlErrObj.errMsg());
         // Users do not care about this error, so don't send it upstream.
     } else {

src/ccontrol/UserQueryAsyncResult.cc

@@ -39,10 +39,6 @@ namespace lsst::qserv::qmeta {
 class QMeta;
 }
 
-namespace lsst::qserv::sql {
-class SqlConnection;
-}
-
 namespace lsst::qserv::ccontrol {
 
 /// @addtogroup ccontrol
@@ -61,10 +57,9 @@ class UserQueryAsyncResult : public UserQuery {
      *  @param queryId:       Query ID for which to return result
      *  @param qMetaCzarId:   ID for current czar
      *  @param qMetaSelect:   QMetaSelect instance
-     *  @param resultDbConn:  Connection to results database
      */
     UserQueryAsyncResult(QueryId queryId, qmeta::CzarId qMetaCzarId,
-                         std::shared_ptr<qmeta::QMeta> const& qMeta, sql::SqlConnection* resultDbConn);
+                         std::shared_ptr<qmeta::QMeta> const& qMeta);
 
     // Destructor
     ~UserQueryAsyncResult();
@@ -111,7 +106,6 @@ class UserQueryAsyncResult : public UserQuery {
     QueryId _queryId;
     qmeta::CzarId _qMetaCzarId;
     std::shared_ptr<qmeta::QMeta> _qMeta;
-    sql::SqlConnection* _resultDbConn;
     qmeta::QInfo _qInfo;
     std::shared_ptr<qmeta::MessageStore> _messageStore;
     QueryState _qState = UNKNOWN;

src/ccontrol/UserQueryAsyncResult.h

@@ -41,8 +41,6 @@
 #include "ccontrol/ConfigMap.h"
 #include "ccontrol/ParseRunner.h"
 #include "ccontrol/UserQueryAsyncResult.h"
-#include "ccontrol/UserQueryDrop.h"
-#include "ccontrol/UserQueryFlushChunksCache.h"
 #include "ccontrol/UserQueryInvalid.h"
 #include "ccontrol/UserQueryProcessList.h"
 #include "ccontrol/UserQueryQueries.h"
@@ -135,8 +133,7 @@ std::shared_ptr<UserQuery> _makeUserQueryProcessList(query::SelectStmt::Ptr& stm
     }
     LOGS(_log, LOG_LVL_DEBUG, "SELECT query is a PROCESSLIST");
     try {
-        return std::make_shared<UserQueryProcessList>(stmt, sharedResources->resultDbConn.get(),
-                                                      sharedResources->qMetaSelect,
+        return std::make_shared<UserQueryProcessList>(stmt, sharedResources->qMetaSelect,
                                                       sharedResources->qMetaCzarId, userQueryId, resultDb);
     } catch (std::exception const& exc) {
         return std::make_shared<UserQueryInvalid>(exc.what());
@@ -164,9 +161,8 @@ std::shared_ptr<UserQuery> _makeUserQueryQueries(query::SelectStmt::Ptr& stmt,
     }
     LOGS(_log, LOG_LVL_DEBUG, "SELECT query is a QUERIES");
     try {
-        return std::make_shared<UserQueryQueries>(stmt, sharedResources->resultDbConn.get(),
-                                                  sharedResources->qMetaSelect, sharedResources->qMetaCzarId,
-                                                  userQueryId, resultDb);
+        return std::make_shared<UserQueryQueries>(stmt, sharedResources->qMetaSelect,
+                                                  sharedResources->qMetaCzarId, userQueryId, resultDb);
     } catch (std::exception const& exc) {
         return std::make_shared<UserQueryInvalid>(exc.what());
     }
@@ -219,8 +215,7 @@ std::shared_ptr<UserQuerySharedResources> makeUserQuerySharedResources(
             std::make_shared<qmeta::QMetaMysql>(czarConfig->getMySqlQmetaConfig(),
                                                 czarConfig->getMaxMsgSourceStore()),
             std::make_shared<qmeta::QStatusMysql>(czarConfig->getMySqlQStatusDataConfig()),
-            std::make_shared<qmeta::QMetaSelect>(czarConfig->getMySqlQmetaConfig()),
-            sql::SqlConnectionFactory::make(czarConfig->getMySqlResultConfig()), dbModels, czarName,
+            std::make_shared<qmeta::QMetaSelect>(czarConfig->getMySqlQmetaConfig()), dbModels, czarName,
             czarConfig->getInteractiveChunkLimit());
 }
 
@@ -325,9 +320,9 @@ UserQuery::Ptr UserQueryFactory::newUserQuery(std::string const& aQuery, std::st
             qmetaHasDataForSelectCountStarQuery(stmt, _userQuerySharedResources, defaultDb, rowsTable)) {
             LOGS(_log, LOG_LVL_DEBUG, "make UserQuerySelectCountStar");
             auto uq = std::make_shared<UserQuerySelectCountStar>(
-                    query, _userQuerySharedResources->resultDbConn, _userQuerySharedResources->qMetaSelect,
-                    _userQuerySharedResources->queryMetadata, userQueryId, rowsTable, resultDb, countSpelling,
-                    _userQuerySharedResources->qMetaCzarId, async);
+                    query, _userQuerySharedResources->qMetaSelect, _userQuerySharedResources->queryMetadata,
+                    userQueryId, rowsTable, resultDb, countSpelling, _userQuerySharedResources->qMetaCzarId,
+                    async);
             uq->qMetaRegister(resultLocation, msgTableName);
             return uq;
         }
@@ -372,8 +367,8 @@ UserQuery::Ptr UserQueryFactory::newUserQuery(std::string const& aQuery, std::st
         auto uq = std::make_shared<UserQuerySelect>(
                 qs, messageStore, executive, _userQuerySharedResources->databaseModels, infileMergerConfig,
                 _userQuerySharedResources->secondaryIndex, _userQuerySharedResources->queryMetadata,
-                _userQuerySharedResources->queryStatsData, _userQuerySharedResources->semaMgrConnections,
-                _userQuerySharedResources->qMetaCzarId, errorExtra, async, resultDb, uberJobMaxChunks);
+                _userQuerySharedResources->queryStatsData, _userQuerySharedResources->qMetaCzarId, errorExtra,
+                async, resultDb, uberJobMaxChunks);
         if (sessionValid) {
             uq->qMetaRegister(resultLocation, msgTableName);
             uq->setupMerger();
@@ -383,39 +378,13 @@ UserQuery::Ptr UserQueryFactory::newUserQuery(std::string const& aQuery, std::st
         return uq;
     } else if (UserQueryType::isSelectResult(query, userJobId)) {
         auto uq = std::make_shared<UserQueryAsyncResult>(userJobId, _userQuerySharedResources->qMetaCzarId,
-                                                         _userQuerySharedResources->queryMetadata,
-                                                         _userQuerySharedResources->resultDbConn.get());
+                                                         _userQuerySharedResources->queryMetadata);
         LOGS(_log, LOG_LVL_DEBUG, "make UserQueryAsyncResult: userJobId=" << userJobId);
         return uq;
-    } else if (UserQueryType::isDropTable(query, dbName, tableName)) {
-        // processing DROP TABLE
-        if (dbName.empty()) {
-            dbName = defaultDb;
-        }
-        auto uq = std::make_shared<UserQueryDrop>(_userQuerySharedResources->css, dbName, tableName,
-                                                  _userQuerySharedResources->resultDbConn.get(),
-                                                  _userQuerySharedResources->queryMetadata,
-                                                  _userQuerySharedResources->qMetaCzarId);
-        LOGS(_log, LOG_LVL_DEBUG, "make UserQueryDrop: " << dbName << "." << tableName);
-        return uq;
-    } else if (UserQueryType::isDropDb(query, dbName)) {
-        // processing DROP DATABASE
-        auto uq = std::make_shared<UserQueryDrop>(_userQuerySharedResources->css, dbName, std::string(),
-                                                  _userQuerySharedResources->resultDbConn.get(),
-                                                  _userQuerySharedResources->queryMetadata,
-                                                  _userQuerySharedResources->qMetaCzarId);
-        LOGS(_log, LOG_LVL_DEBUG, "make UserQueryDrop: db=" << dbName);
-        return uq;
-    } else if (UserQueryType::isFlushChunksCache(query, dbName)) {
-        auto uq = std::make_shared<UserQueryFlushChunksCache>(_userQuerySharedResources->css, dbName,
-                                                              _userQuerySharedResources->resultDbConn.get());
-        LOGS(_log, LOG_LVL_DEBUG, "make UserQueryFlushChunksCache: " << dbName);
-        return uq;
     } else if (UserQueryType::isShowProcessList(query, full)) {
         LOGS(_log, LOG_LVL_DEBUG, "make UserQueryProcessList: full=" << (full ? 'y' : 'n'));
         try {
-            return std::make_shared<UserQueryProcessList>(full, _userQuerySharedResources->resultDbConn.get(),
-                                                          _userQuerySharedResources->qMetaSelect,
+            return std::make_shared<UserQueryProcessList>(full, _userQuerySharedResources->qMetaSelect,
                                                           _userQuerySharedResources->qMetaCzarId, userQueryId,
                                                           resultDb);
         } catch (std::exception const& exc) {

src/ccontrol/UserQueryFactory.cc

@@ -35,12 +35,14 @@
 // Qserv headers
 #include "css/CssAccess.h"
 #include "css/CssError.h"
+#include "cconfig/CzarConfig.h"
 #include "qmeta/MessageStore.h"
 #include "qmeta/Exceptions.h"
 #include "qmeta/QMetaSelect.h"
 #include "query/FromList.h"
 #include "query/SelectStmt.h"
 #include "sql/SqlConnection.h"
+#include "sql/SqlConnectionFactory.h"
 #include "sql/SqlErrorObject.h"
 #include "sql/SqlBulkInsert.h"
 #include "sql/statement.h"
@@ -61,12 +63,10 @@ namespace lsst::qserv::ccontrol {
 
 // Constructor
 UserQueryProcessList::UserQueryProcessList(std::shared_ptr<query::SelectStmt> const& statement,
-                                           sql::SqlConnection* resultDbConn,
                                            std::shared_ptr<qmeta::QMetaSelect> const& qMetaSelect,
                                            qmeta::CzarId qMetaCzarId, std::string const& userQueryId,
                                            std::string const& resultDb)
-        : _resultDbConn(resultDbConn),
-          _qMetaSelect(qMetaSelect),
+        : _qMetaSelect(qMetaSelect),
           _qMetaCzarId(qMetaCzarId),
           _messageStore(std::make_shared<qmeta::MessageStore>()),
           _resultTableName(::g_nextResultTableId(userQueryId)),
@@ -90,12 +90,10 @@ UserQueryProcessList::UserQueryProcessList(std::shared_ptr<query::SelectStmt> co
     }
 }
 
-UserQueryProcessList::UserQueryProcessList(bool full, sql::SqlConnection* resultDbConn,
-                                           std::shared_ptr<qmeta::QMetaSelect> const& qMetaSelect,
+UserQueryProcessList::UserQueryProcessList(bool full, std::shared_ptr<qmeta::QMetaSelect> const& qMetaSelect,
                                            qmeta::CzarId qMetaCzarId, std::string const& userQueryId,
                                            std::string const& resultDb)
-        : _resultDbConn(resultDbConn),
-          _qMetaSelect(qMetaSelect),
+        : _qMetaSelect(qMetaSelect),
           _qMetaCzarId(qMetaCzarId),
           _messageStore(std::make_shared<qmeta::MessageStore>()),
           _resultTableName(::g_nextResultTableId(userQueryId)),
@@ -156,8 +154,11 @@ void UserQueryProcessList::submit() {
         if (col.colType.sqlType == "TIMESTAMP") createTable += " NULL";
     }
     createTable += ')';
+
     LOGS(_log, LOG_LVL_DEBUG, "creating result table: " << createTable);
-    if (!_resultDbConn->runQuery(createTable, errObj)) {
+    auto const czarConfig = cconfig::CzarConfig::instance();
+    auto const resultDbConn = sql::SqlConnectionFactory::make(czarConfig->getMySqlResultConfig());
+    if (!resultDbConn->runQuery(createTable, errObj)) {
         LOGS(_log, LOG_LVL_ERROR, "failed to create result table: " << errObj.errMsg());
         std::string message = "Internal failure, failed to create result table: " + errObj.errMsg();
         _messageStore->addMessage(-1, "PROCESSLIST", 1051, message, MessageSeverity::MSG_ERROR);
@@ -172,7 +173,7 @@ void UserQueryProcessList::submit() {
     }
 
     // copy stuff over to result table
-    sql::SqlBulkInsert bulkInsert(_resultDbConn, _resultTableName, resColumns);
+    sql::SqlBulkInsert bulkInsert(resultDbConn.get(), _resultTableName, resColumns);
     for (auto& row : *results) {
         std::vector<std::string> values;
         for (unsigned i = 0; i != row.size(); ++i) {
@@ -191,7 +192,7 @@ void UserQueryProcessList::submit() {
                 values.push_back(std::string(ptr, ptr + len));
             } else {
                 // everything else should be quoted
-                values.push_back("'" + _resultDbConn->escapeString(std::string(ptr, ptr + len)) + "'");
+                values.push_back("'" + resultDbConn->escapeString(std::string(ptr, ptr + len)) + "'");
             }
         }