Minor refactoring of the result merging logic

Moved query completion/cancellation tests from the low-level context
of the IfileMerger class into the MergingHandler, where the logic
belongs to.

The change streamlines responsibilities of the affected classes and helps
code readers better understand the logic flow withing this corner of
the application.

Author: iagaponenko

src/ccontrol/MergingHandler.cc

@@ -49,6 +49,7 @@
 #include "proto/ProtoHeaderWrap.h"
 #include "proto/worker.pb.h"
 #include "qdisp/CzarStats.h"
+#include "qdisp/Executive.h"
 #include "qdisp/JobQuery.h"
 #include "rproc/InfileMerger.h"
 #include "util/Bug.h"
@@ -330,16 +331,27 @@ void MergingHandler::_initState() { _setError(0, ""); }
 bool MergingHandler::_merge(proto::ResponseSummary const& responseSummary,
                             proto::ResponseData const& responseData,
                             shared_ptr<qdisp::JobQuery> const& jobQuery) {
-    if (_flushed) {
-        throw util::Bug(ERR_LOC, "already flushed");
-    }
-    bool success = _infileMerger->merge(responseSummary, responseData, jobQuery);
-    if (!success) {
-        LOGS(_log, LOG_LVL_WARN, __func__ << " failed");
-        util::Error const& err = _infileMerger->getError();
-        _setError(ccontrol::MSG_RESULT_ERROR, err.getMsg());
+    if (_flushed) throw util::Bug(ERR_LOC, "already flushed");
+
+    // Nothing to do if size is zero.
+    if (responseData.row_size() == 0) return true;
+
+    // Do nothing if the query got cancelled for any reason.
+    if (jobQuery->isQueryCancelled()) return true;
+
+    // Check for other indicators that the query may have cancelled or finished.
+    auto executive = jobQuery->getExecutive();
+    if (executive == nullptr || executive->getCancelled() || executive->isLimitRowComplete()) {
+        return true;
     }
-    return success;
+
+    // Attempt the actual merge.
+    if (_infileMerger->merge(responseSummary, responseData)) return true;
+
+    LOGS(_log, LOG_LVL_WARN, __func__ << " failed");
+    util::Error const& err = _infileMerger->getError();
+    _setError(ccontrol::MSG_RESULT_ERROR, err.getMsg());
+    return false;
 }
 
 void MergingHandler::_setError(int code, std::string const& msg) {

src/rproc/InfileMerger.cc

@@ -59,8 +59,6 @@
 #include "proto/ProtoImporter.h"
 #include "proto/worker.pb.h"
 #include "qdisp/CzarStats.h"
-#include "qdisp/Executive.h"
-#include "qdisp/JobQuery.h"
 #include "qproc/DatabaseModels.h"
 #include "query/ColumnRef.h"
 #include "query/SelectStmt.h"
@@ -169,28 +167,13 @@ void InfileMerger::mergeCompleteFor(int jobId) {
 }
 
 bool InfileMerger::merge(proto::ResponseSummary const& responseSummary,
-                         proto::ResponseData const& responseData,
-                         std::shared_ptr<qdisp::JobQuery> const& jq) {
+                         proto::ResponseData const& responseData) {
     int const jobId = responseSummary.jobid();
     std::string queryIdJobStr = QueryIdHelper::makeIdStr(responseSummary.queryid(), jobId);
     if (!_queryIdStrSet) {
         _setQueryIdStr(QueryIdHelper::makeIdStr(responseSummary.queryid()));
     }
 
-    // Nothing to do if size is zero.
-    if (responseData.row_size() == 0) {
-        return true;
-    }
-
-    // Do nothing if the query got cancelled for any reason.
-    if (jq->isQueryCancelled()) {
-        return true;
-    }
-    auto executive = jq->getExecutive();
-    if (executive == nullptr || executive->getCancelled() || executive->isLimitRowComplete()) {
-        return true;
-    }
-
     TimeCountTracker<double>::CALLBACKFUNC cbf = [](TIMEPOINT start, TIMEPOINT end, double bytes,
                                                     bool success) {
         if (!success) return;

src/rproc/InfileMerger.h

@@ -53,7 +53,6 @@ class ResponseData;
 class ResponseSummary;
 }  // namespace proto
 namespace qdisp {
-class JobQuery;
 class MessageStore;
 }  // namespace qdisp
 namespace qproc {
@@ -103,10 +102,8 @@ class InfileMerger {
     ~InfileMerger() = default;
 
     /// Merge a worker response, which contains a single ResponseData message
-    /// Using job query info for early termination of the merge if needed.
     /// @return true if merge was successfully imported.
-    bool merge(proto::ResponseSummary const& responseSummary, proto::ResponseData const& responseData,
-               std::shared_ptr<qdisp::JobQuery> const& jq);
+    bool merge(proto::ResponseSummary const& responseSummary, proto::ResponseData const& responseData);
 
     /// Indicate the merge for the job is complete.
     void mergeCompleteFor(int jobId);