Added JobErrorMsg.

Author: jgates108

src/ccontrol/MergingHandler.cc

@@ -70,8 +70,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;
@@ -84,7 +88,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

@@ -135,7 +135,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->qMetaSelect,
-                                                      sharedResources->czarId, userQueryId, resultDb);
+                                                      sharedResources->qMetaCzarId, userQueryId, resultDb);
     } catch (std::exception const& exc) {
         return std::make_shared<UserQueryInvalid>(exc.what());
     }

src/ccontrol/UserQueryFactory.cc

@@ -35,6 +35,7 @@
 // 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"
@@ -65,8 +66,7 @@ UserQueryProcessList::UserQueryProcessList(std::shared_ptr<query::SelectStmt> co
                                            std::shared_ptr<qmeta::QMetaSelect> const& qMetaSelect,
                                            qmeta::CzarId czarId, 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)),
@@ -93,8 +93,7 @@ UserQueryProcessList::UserQueryProcessList(std::shared_ptr<query::SelectStmt> co
 UserQueryProcessList::UserQueryProcessList(bool full, std::shared_ptr<qmeta::QMetaSelect> const& qMetaSelect,
                                            qmeta::CzarId czarId, 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)),

src/ccontrol/UserQueryProcessList.cc

@@ -28,6 +28,7 @@
 #include "cconfig/CzarConfig.h"
 #include "qmeta/QMeta.h"
 #include "qmeta/QProgress.h"
+
 namespace lsst::qserv::ccontrol {
 
 UserQuerySharedResources::UserQuerySharedResources(

src/ccontrol/UserQueryResources.cc

@@ -582,11 +582,10 @@ QueryState UserQuerySelect::join() {
 
 /// Release resources held by the merger
 void UserQuerySelect::_discardMerger(std::lock_guard<std::mutex> const& lock) {
-    _infileMergerConfig.reset();
     if (_infileMerger && !_infileMerger->isFinished()) {
         throw UserQueryError(getQueryIdString() + " merger unfinished, cannot discard");
     }
-    _infileMerger.reset();
+    _infileMergerConfig.reset();
 }
 
 /// Release resources.
@@ -608,9 +607,8 @@ void UserQuerySelect::discard() {
         throw UserQueryError(getQueryIdString() + " Executive unfinished, cannot discard");
     }
 
+    // Deleting the executive may save some time if results were found early.
     _executive.reset();
-    _messageStore.reset();
-    _qSession.reset();
 
     try {
         _discardMerger(lock);

src/ccontrol/UserQuerySelect.cc

@@ -29,7 +29,8 @@
 // Qserv headers
 #include "cconfig/CzarConfig.h"
 #include "czar/Czar.h"
-#include "protojson/JobReadyMsg.h"
+#include "protojson/UberJobErrorMsg.h"
+#include "protojson/UberJobReadyMsg.h"
 #include "protojson/WorkerCzarComIssue.h"
 #include "qdisp/Executive.h"
 #include "qdisp/UberJob.h"
@@ -107,14 +108,14 @@ json HttpCzarWorkerModule::_handleJobError(string const& func) {
     // Parse and verify the json message and then kill the UberJob.
     json jsRet = {{"success", 0}, {"errortype", "unknown"}, {"note", "initialized"}};
     try {
-        // TODO:UJ see wbase::UberJobData::responseError for message construction
-        string const targetWorkerId = body().required<string>("workerid");
-        string const czarName = body().required<string>("czar");
-        qmeta::CzarId const czarId = body().required<qmeta::CzarId>("czarid");
-        QueryId const queryId = body().required<QueryId>("queryid");
-        UberJobId const uberJobId = body().required<UberJobId>("uberjobid");
-        int const errorCode = body().required<int>("errorCode");
-        string const errorMsg = body().required<string>("errorMsg");
+        string const& repliInstanceId = cconfig::CzarConfig::instance()->replicationInstanceId();
+        string const& repliAuthKey = cconfig::CzarConfig::instance()->replicationAuthKey();
+        auto const& jsReq = body().objJson;
+        auto jrMsg = protojson::UberJobErrorMsg::createFromJson(jsReq, repliInstanceId, repliAuthKey);
+
+        auto const queryId = jrMsg->getQueryId();
+        auto const czarId = jrMsg->getCzarId();
+        auto const uberJobId = jrMsg->getUberJobId();
 
         // Find UberJob
         qdisp::Executive::Ptr exec = czar::Czar::getCzar()->getExecutiveFromMap(queryId);
@@ -129,9 +130,8 @@ json HttpCzarWorkerModule::_handleJobError(string const& func) {
                                    " czar=" + to_string(czarId));
         }
 
-        auto importRes = uj->workerError(errorCode, errorMsg);
+        auto importRes = uj->workerError(jrMsg->getErrorCode(), jrMsg->getErrorMsg());
         jsRet = importRes;
-
     } catch (std::invalid_argument const& iaEx) {
         LOGS(_log, LOG_LVL_ERROR,
              "HttpCzarWorkerModule::_handleJobError received " << iaEx.what() << " js=" << body().objJson);
@@ -148,10 +148,8 @@ json HttpCzarWorkerModule::_handleJobReady(string const& func) {
     // Parse and verify the json message and then have the uberjob import the file.
     json jsRet = {{"success", 1}, {"errortype", "unknown"}, {"note", "initialized"}};
     try {
-        string const repliInstanceId = cconfig::CzarConfig::instance()->replicationInstanceId();
-        string const repliAuthKey = cconfig::CzarConfig::instance()->replicationAuthKey();
         auto const& jsReq = body().objJson;
-        auto jrMsg = protojson::JobReadyMsg::createFromJson(jsReq, repliInstanceId, repliAuthKey);
+        auto jrMsg = protojson::UberJobReadyMsg::createFromJson(jsReq);
 
         // Find UberJob
         auto queryId = jrMsg->getQueryId();
@@ -212,7 +210,7 @@ json HttpCzarWorkerModule::_handleWorkerCzarComIssue(string const& func) {
                 execPtr->killIncompleteUberJobsOnWorker(wId);
             }
         }
-        jsRet = wccIssue->serializeResponseJson();
+        jsRet = wccIssue->responseToJson();
         LOGS(_log, LOG_LVL_TRACE, "HttpCzarWorkerModule::_handleWorkerCzarComIssue jsRet=" << jsRet.dump());
 
     } catch (std::invalid_argument const& iaEx) {

src/czar/HttpCzarWorkerModule.cc

@@ -2,11 +2,11 @@ add_library(mysql SHARED)
 add_dependencies(mysql proto)
 
 target_sources(mysql PRIVATE
+    CsvBuffer.cc
     LocalInfile.cc
     MySqlConfig.cc
     MySqlConnection.cc
     MySqlUtils.cc
-    CsvBuffer.cc
     SchemaFactory.cc
 )
 

src/mysql/CMakeLists.txt

@@ -132,8 +132,6 @@ int LocalInfile::getError(char* buf, unsigned int bufLen) {
     return 0;
 }
 
-LocalInfile::Mgr::~Mgr() { LOGS(_log, LOG_LVL_TRACE, "LocalInfile::Mgr::~Mgr()"); }
-
 void LocalInfile::Mgr::attach(MYSQL* mysql) {
     mysql_set_local_infile_handler(mysql, local_infile_init, local_infile_read, local_infile_end,
                                    local_infile_error, this);
@@ -156,7 +154,11 @@ std::string LocalInfile::Mgr::prepareSrc(std::shared_ptr<CsvBuffer> const& csvBu
 int LocalInfile::Mgr::local_infile_init(void** ptr, const char* filename, void* userdata) {
     assert(userdata);
     LocalInfile::Mgr* m = static_cast<LocalInfile::Mgr*>(userdata);
+<<<<<<< HEAD
     auto csvBuffer = m->get(std::string(filename));
+=======
+    auto csvBuffer = m->getCsv(std::string(filename));
+>>>>>>> 24f5d37ea (Added JobErrorMsg.)
     assert(csvBuffer);
     LocalInfile* lf = new LocalInfile(filename, csvBuffer);
     *ptr = lf;
@@ -190,10 +192,17 @@ void LocalInfile::Mgr::setBuffer(std::string const& filename, std::shared_ptr<Cs
     }
 }
 
+<<<<<<< HEAD
 std::shared_ptr<CsvBuffer> LocalInfile::Mgr::get(std::string const& filename) {
     std::lock_guard<std::mutex> lock(_mapMutex);
     CsvBufferMap::iterator i = _map.find(filename);
     if (i == _map.end()) {
+=======
+std::shared_ptr<CsvBuffer> LocalInfile::Mgr::getCsv(std::string const& filename) {
+    std::lock_guard<std::mutex> lock(_mapMutex);
+    auto i = _mapCsv.find(filename);
+    if (i == _mapCsv.end()) {
+>>>>>>> 24f5d37ea (Added JobErrorMsg.)
         return std::shared_ptr<CsvBuffer>();
     }
     return i->second;
@@ -209,7 +218,11 @@ std::string LocalInfile::Mgr::_nextFilename() {
 
 bool LocalInfile::Mgr::_set(std::string const& filename, std::shared_ptr<CsvBuffer> const& csvBuffer) {
     std::lock_guard<std::mutex> lock(_mapMutex);
+<<<<<<< HEAD
     auto res = _map.insert(std::pair<std::string, std::shared_ptr<CsvBuffer>>(filename, csvBuffer));
+=======
+    auto res = _mapCsv.insert(std::pair<std::string, std::shared_ptr<CsvBuffer>>(filename, csvBuffer));
+>>>>>>> 24f5d37ea (Added JobErrorMsg.)
     return res.second;
 }
 

src/mysql/LocalInfile.cc

@@ -125,7 +125,7 @@ class LocalInfile::Mgr : boost::noncopyable {
 
     std::string insertBuffer(std::shared_ptr<CsvBuffer> const& csvBuffer);
     void setBuffer(std::string const& s, std::shared_ptr<CsvBuffer> const& csvBuffer);
-    std::shared_ptr<CsvBuffer> get(std::string const& filename);
+    std::shared_ptr<CsvBuffer> getCsv(std::string const& filename);
 
 private:
     /// @return next filename
@@ -135,7 +135,7 @@ class LocalInfile::Mgr : boost::noncopyable {
     bool _set(std::string const& filename, std::shared_ptr<CsvBuffer> const& csvBuffer);
 
     typedef std::map<std::string, std::shared_ptr<CsvBuffer>> CsvBufferMap;
-    CsvBufferMap _map;
+    CsvBufferMap _mapCsv;
     std::mutex _mapMutex;
 };