DRILL-2573: C++ Client - Separate QueryResult into QueryResult and QueryData

13 files changed, 1171 insertions, 1156 deletions

diff --git a/contrib/native/client/example/querySubmitter.cpp b/contrib/native/client/example/querySubmitter.cpp
index bef64bf20..85e89e0c2 100644
--- a/contrib/native/client/example/querySubmitter.cpp
+++ b/contrib/native/client/example/querySubmitter.cpp
@@ -71,15 +71,18 @@ Drill::status_t QueryResultsListener(void* ctx, Drill::RecordBatch* b, Drill::Dr
     // or
     // (received query state message passed by `err` and b is NULL)
     if(!err){
-        assert(b!=NULL);
-        b->print(std::cout, 0); // print all rows
-        std::cout << "DATA RECEIVED ..." << std::endl;
-        delete b; // we're done with this batch, we can delete it
-        if(bTestCancel){
-            return Drill::QRY_FAILURE;
+        if(b!=NULL){
+            b->print(std::cout, 0); // print all rows
+            std::cout << "DATA RECEIVED ..." << std::endl;
+            delete b; // we're done with this batch, we can delete it
+            if(bTestCancel){
+                return Drill::QRY_FAILURE;
+            }else{
+                return Drill::QRY_SUCCESS ;
+            }
         }else{
-            return Drill::QRY_SUCCESS ;
-        }
+            std::cout << "Query Complete." << std::endl;
+		}
     }else{
         assert(b==NULL);
         switch(err->status) {
@@ -392,6 +395,7 @@ int main(int argc, char* argv[]) {
                 }
                 client.freeQueryIterator(&pRecIter);
             }
+            client.waitForResults();
         }else{
             if(bSyncSend){
                 for(queryInpIter = queryInputs.begin(); queryInpIter != queryInputs.end(); queryInpIter++) {
diff --git a/contrib/native/client/src/clientlib/drillClientImpl.cpp b/contrib/native/client/src/clientlib/drillClientImpl.cpp
index 71f960e0f..dce5bdc30 100644
--- a/contrib/native/client/src/clientlib/drillClientImpl.cpp
+++ b/contrib/native/client/src/clientlib/drillClientImpl.cpp
@@ -53,7 +53,7 @@ static std::map<exec::shared::QueryResult_QueryState, status_t> QUERYSTATE_TO_ST
     (exec::shared::QueryResult_QueryState_COMPLETED, QRY_COMPLETED)
     (exec::shared::QueryResult_QueryState_CANCELED, QRY_CANCELED)
     (exec::shared::QueryResult_QueryState_FAILED, QRY_FAILED)
-    (exec::shared::QueryResult_QueryState_UNKNOWN_QUERY, QRY_UNKNOWN_QUERY);
+    ;
 
 RpcEncoder DrillClientImpl::s_encoder;
 RpcDecoder DrillClientImpl::s_decoder;
@@ -325,7 +325,7 @@ connectionStatus_t DrillClientImpl::validateHandshake(DrillUserProperties* prope
         return ret;
     }
     if(m_handshakeVersion != u2b.rpc_version()) {
-        DRILL_LOG(LOG_TRACE) << "Invalid rpc version.  Expected << "
+        DRILL_LOG(LOG_TRACE) << "Invalid rpc version.  Expected "
             << DRILL_RPC_VERSION << ", actual "<< m_handshakeVersion << "." << std::endl;
         return handleConnError(CONN_HANDSHAKE_FAILED,
                 getMessage(ERR_CONN_NOHSHAKE, DRILL_RPC_VERSION, m_handshakeVersion));
@@ -510,7 +510,59 @@ status_t DrillClientImpl::readMsg(ByteBuf_t _buf,
 status_t DrillClientImpl::processQueryResult(AllocatedBufferPtr  allocatedBuffer, InBoundRpcMessage& msg ){
     DrillClientQueryResult* pDrillClientQueryResult=NULL;
     status_t ret=QRY_SUCCESS;
-	exec::shared::QueryId qid;
+    exec::shared::QueryId qid;
+    sendAck(msg, true);
+    {
+        boost::lock_guard<boost::mutex> lock(this->m_dcMutex);
+        exec::shared::QueryResult qr;
+
+        DRILL_LOG(LOG_DEBUG) << "Processing Query Result " << std::endl;
+        qr.ParseFromArray(msg.m_pbody.data(), msg.m_pbody.size());
+        DRILL_LOG(LOG_TRACE) << qr.DebugString() << std::endl;
+        
+        qid.CopyFrom(qr.query_id());
+        
+        if (qr.has_query_state() &&
+                qr.query_state() != exec::shared::QueryResult_QueryState_RUNNING &&
+                qr.query_state() != exec::shared::QueryResult_QueryState_PENDING) {
+            pDrillClientQueryResult=findQueryResult(qid);
+            //Queries that have been cancelled or whose resources are freed before completion 
+            //do not have a DrillClientQueryResult object. We need not handle the terminal message 
+            //in that case since all it does is to free resources (and they have already been freed)
+            if(pDrillClientQueryResult!=NULL){
+                //Validate the RPC message
+                std::string valErr;
+                if( (ret=validateResultMessage(msg, qr, valErr)) != QRY_SUCCESS){
+                    delete allocatedBuffer;
+                    DRILL_LOG(LOG_TRACE) << "DrillClientImpl::processQueryResult: ERR_QRY_INVRPC." << std::endl;
+                    return handleQryError(ret, getMessage(ERR_QRY_INVRPC, valErr.c_str()), pDrillClientQueryResult);
+                }
+                ret=processQueryStatusResult(&qr, pDrillClientQueryResult);
+            }else{
+                // We've received the final message for a query that has been cancelled
+                // or for which the resources have been freed. We no longer need to listen
+                // for more incoming messages for such a query.
+                DRILL_LOG(LOG_TRACE) << "DrillClientImpl::processQueryResult:" << debugPrintQid(qid)<< " completed."<< std::endl;
+                m_pendingRequests--;
+                DRILL_LOG(LOG_TRACE) << "DrillClientImpl::processQueryResult: pending requests is " << m_pendingRequests<< std::endl;
+                ret=QRY_CANCELED;
+            }
+            delete allocatedBuffer;
+            return ret;
+        }else{
+            // Normal query results come back with query_state not set.
+            // Actually this is not strictly true. The query state is set to
+            // 0(i.e. PENDING), but protobuf thinks this means the value is not set.
+            DRILL_LOG(LOG_TRACE) << "DrillClientImpl::processQueryResult: Query State was not set.\n";
+        }
+    }
+    return ret;
+}
+
+status_t DrillClientImpl::processQueryData(AllocatedBufferPtr  allocatedBuffer, InBoundRpcMessage& msg ){
+    DrillClientQueryResult* pDrillClientQueryResult=NULL;
+    status_t ret=QRY_SUCCESS;
+    exec::shared::QueryId qid;
     // Be a good client and send ack as early as possible.
     // Drillbit pushed the query result to the client, the client should send ack
     // whenever it receives the message
@@ -518,62 +570,34 @@ status_t DrillClientImpl::processQueryResult(AllocatedBufferPtr  allocatedBuffer
     RecordBatch* pRecordBatch=NULL;
     {
         boost::lock_guard<boost::mutex> lock(this->m_dcMutex);
-        exec::shared::QueryResult* qr = new exec::shared::QueryResult; //Record Batch will own this object and free it up.
+        exec::shared::QueryData* qr = new exec::shared::QueryData; //Record Batch will own this object and free it up.
 
-        DRILL_LOG(LOG_DEBUG) << "Processing Query Result " << std::endl;
+        DRILL_LOG(LOG_DEBUG) << "Processing Query Data " << std::endl;
         qr->ParseFromArray(msg.m_pbody.data(), msg.m_pbody.size());
         DRILL_LOG(LOG_TRACE) << qr->DebugString() << std::endl;
 
-        DRILL_LOG(LOG_DEBUG) << "Searching for Query Id - " << debugPrintQid(qr->query_id()) << std::endl;
-
         qid.CopyFrom(qr->query_id());
-        std::map<exec::shared::QueryId*, DrillClientQueryResult*, compareQueryId>::iterator it;
-        DRILL_LOG(LOG_TRACE) << "DrillClientImpl::processQueryResult: m_queryResults size: " << m_queryResults.size() << std::endl;
-        if(m_queryResults.size() != 0){
-            for(it=m_queryResults.begin(); it!=m_queryResults.end(); it++){
-                DRILL_LOG(LOG_TRACE) << "DrillClientImpl::processQueryResult: m_QueryResult ids: [" << it->first->part1() << ":"
-                    << it->first->part2() << "]\n";
-            }
-        }
         if(qid.part1()==0){
-            DRILL_LOG(LOG_TRACE) << "DrillClientImpl::processQueryResult: QID=0. Ignore and return QRY_SUCCESS." << std::endl;
-            return QRY_SUCCESS;
-        }
-        it=this->m_queryResults.find(&qid);
-        if(it!=this->m_queryResults.end()){
-            pDrillClientQueryResult=(*it).second;
-        }else{
-            ret=processCancelledQueryResult(qid, qr);
-            DRILL_LOG(LOG_TRACE) << "Cleaning up resource allocated for canceled quyery." << std::endl;
-            delete qr;
+            DRILL_LOG(LOG_TRACE) << "DrillClientImpl::processQueryData: QID=0. Ignore and return QRY_SUCCESS." << std::endl;
             delete allocatedBuffer;
-            return ret;
+            return QRY_SUCCESS;
         }
-        DRILL_LOG(LOG_DEBUG) << "Drill Client Query Result Query Id - " <<
-            debugPrintQid(*pDrillClientQueryResult->m_pQueryId) << std::endl;
 
-        // Drillbit may send a query state change message which does not contain any
-        // record batch.
-        if (qr->has_query_state() &&
-                qr->query_state() != exec::shared::QueryResult_QueryState_RUNNING &&
-                qr->query_state() != exec::shared::QueryResult_QueryState_PENDING) {
-            ret=processQueryStatusResult(qr, pDrillClientQueryResult);
-            delete allocatedBuffer;
+        pDrillClientQueryResult=findQueryResult(qid);
+        if(pDrillClientQueryResult==NULL){
+            DRILL_LOG(LOG_TRACE) << "Cleaning up resources allocated for canceled query (" 
+                                 << debugPrintQid(qid) << ")." << std::endl;
             delete qr;
+            delete allocatedBuffer;
             return ret;
-        }else{
-            // Normal query results come back with query_state not set.
-            // Actually this is not strictly true. The query state is set to
-            // 0(i.e. PENDING), but protobuf thinks this means the value is not set.
-            DRILL_LOG(LOG_TRACE) << "DrillClientImpl::processQueryResult: Query State was not set.\n";
         }
-
+        
         //Validate the RPC message
         std::string valErr;
-        if( (ret=validateMessage(msg, *qr, valErr)) != QRY_SUCCESS){
+        if( (ret=validateDataMessage(msg, *qr, valErr)) != QRY_SUCCESS){
             delete allocatedBuffer;
             delete qr;
-            DRILL_LOG(LOG_TRACE) << "DrillClientImpl::processQueryResult: ERR_QRY_INVRPC.\n";
+            DRILL_LOG(LOG_TRACE) << "DrillClientImpl::processQueryData: ERR_QRY_INVRPC.\n";
             pDrillClientQueryResult->setQueryStatus(ret);
             return handleQryError(ret, getMessage(ERR_QRY_INVRPC, valErr.c_str()), pDrillClientQueryResult);
         }
@@ -590,16 +614,13 @@ status_t DrillClientImpl::processQueryResult(AllocatedBufferPtr  allocatedBuffer
             << pRecordBatch->getNumRecords()  << std::endl;
         DRILL_LOG(LOG_DEBUG) << debugPrintQid(qr->query_id())<<"recordBatch.numFields "
             << pRecordBatch->getNumFields()  << std::endl;
-        DRILL_LOG(LOG_DEBUG) << debugPrintQid(qr->query_id())<<"recordBatch.isLastChunk "
-            << pRecordBatch->isLastChunk()  << std::endl;
 
         ret=pDrillClientQueryResult->setupColumnDefs(qr);
         if(ret==QRY_SUCCESS_WITH_INFO){
             pRecordBatch->schemaChanged(true);
         }
 
-        pDrillClientQueryResult->m_bIsQueryPending=true;
-        pDrillClientQueryResult->m_bIsLastChunk=qr->is_last_chunk();
+        pDrillClientQueryResult->setIsQueryPending(true);
         pfnQueryResultsListener pResultsListener=pDrillClientQueryResult->m_pResultsListener;
         if(pDrillClientQueryResult->m_bIsLastChunk){
             DRILL_LOG(LOG_DEBUG) << debugPrintQid(*pDrillClientQueryResult->m_pQueryId)
@@ -619,7 +640,7 @@ status_t DrillClientImpl::processQueryResult(AllocatedBufferPtr  allocatedBuffer
         sendCancel(&qid);
         // Do not decrement pending requests here. We have sent a cancel and we may still receive results that are
         // pushed on the wire before the cancel is processed.
-        pDrillClientQueryResult->m_bIsQueryPending=false;
+        pDrillClientQueryResult->setIsQueryPending(false);
         DRILL_LOG(LOG_DEBUG) << "Client app cancelled query." << std::endl;
         pDrillClientQueryResult->setQueryStatus(ret);
         clearMapEntries(pDrillClientQueryResult);
@@ -628,42 +649,6 @@ status_t DrillClientImpl::processQueryResult(AllocatedBufferPtr  allocatedBuffer
     return ret;
 }
 
-status_t DrillClientImpl::processCancelledQueryResult(exec::shared::QueryId& qid, exec::shared::QueryResult* qr){
-    status_t ret=QRY_SUCCESS;
-    // look in cancelled queries
-    DRILL_LOG(LOG_DEBUG) << "Query Id - " << debugPrintQid(qr->query_id()) << " has been cancelled." << std::endl;
-    std::set<exec::shared::QueryId*, compareQueryId>::iterator it2;
-    exec::shared::QueryId* pQid=NULL;//
-    it2=this->m_cancelledQueries.find(&qid);
-    if(it2!=this->m_cancelledQueries.end()){
-        pQid=(*it2);
-        if(qr->has_query_state()){
-            ret = QUERYSTATE_TO_STATUS_MAP[qr->query_state()];
-            if(qr->query_state()==exec::shared::QueryResult_QueryState_COMPLETED
-                    || qr->query_state()==exec::shared::QueryResult_QueryState_CANCELED
-                    || qr->query_state()==exec::shared::QueryResult_QueryState_FAILED) {
-                this->m_pendingRequests--;
-                this->m_cancelledQueries.erase(it2);
-                delete pQid;
-                DRILL_LOG(LOG_DEBUG) << "Query Id - " << debugPrintQid(qr->query_id()) << " completed." << std::endl;
-                DRILL_LOG(LOG_DEBUG) << "Pending requests - " << this->m_pendingRequests << std::endl;
-            }
-        }
-    }else{
-        status_t ret=QRY_FAILED;
-        if(qr->has_query_state() && qr->query_state()==exec::shared::QueryResult_QueryState_COMPLETED){
-            ret = QUERYSTATE_TO_STATUS_MAP[qr->query_state()];
-        }else if(!qr->has_query_state() && qr->row_count()==0){
-            ret=QRY_SUCCESS;
-        }else{
-            //ret= handleQryError(QRY_INTERNAL_ERROR, getMessage(ERR_QRY_OUTOFORDER), NULL);
-            DRILL_LOG(LOG_DEBUG) << "Pending requests - " << getMessage(ERR_QRY_OUTOFORDER) << std::endl;
-            ret= QRY_SUCCESS;
-        }
-    }
-    return ret;
-}
-
 status_t DrillClientImpl::processQueryId(AllocatedBufferPtr allocatedBuffer, InBoundRpcMessage& msg ){
     DrillClientQueryResult* pDrillClientQueryResult=NULL;
     DRILL_LOG(LOG_DEBUG) << "Processing Query Handle with coordination id:" << msg.m_coord_id << std::endl;
@@ -698,51 +683,72 @@ status_t DrillClientImpl::processQueryId(AllocatedBufferPtr allocatedBuffer, InB
     return ret;
 }
 
+DrillClientQueryResult* DrillClientImpl::findQueryResult(exec::shared::QueryId& qid){
+    DrillClientQueryResult* pDrillClientQueryResult=NULL;
+    DRILL_LOG(LOG_DEBUG) << "Searching for Query Id - " << debugPrintQid(qid) << std::endl;
+    std::map<exec::shared::QueryId*, DrillClientQueryResult*, compareQueryId>::iterator it;
+    DRILL_LOG(LOG_TRACE) << "DrillClientImpl::processQueryResult: m_queryResults size: " << m_queryResults.size() << std::endl;
+    if(m_queryResults.size() != 0){
+        for(it=m_queryResults.begin(); it!=m_queryResults.end(); it++){
+            DRILL_LOG(LOG_TRACE) << "DrillClientImpl::findQueryResult: m_QueryResult ids: [" << it->first->part1() << ":"
+                << it->first->part2() << "]\n";
+        }
+    }
+    it=this->m_queryResults.find(&qid);
+    if(it!=this->m_queryResults.end()){
+        pDrillClientQueryResult=(*it).second;
+        DRILL_LOG(LOG_DEBUG) << "Drill Client Query Result Query Id - " <<
+            debugPrintQid(*pDrillClientQueryResult->m_pQueryId) << std::endl;
+    }
+    return pDrillClientQueryResult;
+}
+
 status_t DrillClientImpl::processQueryStatusResult(exec::shared::QueryResult* qr,
         DrillClientQueryResult* pDrillClientQueryResult){
-        status_t ret = QUERYSTATE_TO_STATUS_MAP[qr->query_state()];
+    status_t ret = QUERYSTATE_TO_STATUS_MAP[qr->query_state()];
+    if(pDrillClientQueryResult!=NULL){
         pDrillClientQueryResult->setQueryStatus(ret);
         pDrillClientQueryResult->setQueryState(qr->query_state());
-        switch(qr->query_state()) {
-            case exec::shared::QueryResult_QueryState_FAILED:
-            case exec::shared::QueryResult_QueryState_UNKNOWN_QUERY:
-                {
-                    // get the error message from protobuf and handle errors
-                    ret=handleQryError(ret, qr->error(0), pDrillClientQueryResult);
-                }
-                break;
-                // m_pendingRequests should be decremented when the query is
-                // completed
-            case exec::shared::QueryResult_QueryState_CANCELED:
-                {
-                    ret=handleTerminatedQryState(ret,
-                            getMessage(ERR_QRY_CANCELED),
-                            pDrillClientQueryResult);
-                    m_pendingRequests--;
-                }
-                break;
-            case exec::shared::QueryResult_QueryState_COMPLETED:
-                {
-                    //Not clean to call the handleTerminatedQryState method
-                    //because it signals an error to the listener.
-                    //The ODBC driver expects this though and the sync API
-                    //handles this (luckily).
-                    ret=handleTerminatedQryState(ret,
-                            getMessage(ERR_QRY_COMPLETED),
-                            pDrillClientQueryResult);
-                    m_pendingRequests--;
-                }
-                break;
-            default:
-                {
-                    DRILL_LOG(LOG_TRACE) << "DrillClientImpl::processQueryResult: Unknown Query State.\n";
-                    ret=handleQryError(QRY_INTERNAL_ERROR,
-                            getMessage(ERR_QRY_UNKQRYSTATE),
-                            pDrillClientQueryResult);
-                }
-                break;
-        }
-        return ret;
+    }
+    switch(qr->query_state()) {
+        case exec::shared::QueryResult_QueryState_FAILED:
+            {
+                // get the error message from protobuf and handle errors
+                ret=handleQryError(ret, qr->error(0), pDrillClientQueryResult);
+            }
+            break;
+            // m_pendingRequests should be decremented when the query is
+            // completed
+        case exec::shared::QueryResult_QueryState_CANCELED:
+            {
+                ret=handleTerminatedQryState(ret,
+                        getMessage(ERR_QRY_CANCELED),
+                        pDrillClientQueryResult);
+                m_pendingRequests--;
+            }
+            break;
+        case exec::shared::QueryResult_QueryState_COMPLETED:
+            {
+                //Not clean to call the handleTerminatedQryState method
+                //because it signals an error to the listener.
+                //The ODBC driver expects this though and the sync API
+                //handles this (luckily).
+                ret=handleTerminatedQryState(ret,
+                        getMessage(ERR_QRY_COMPLETED),
+                        pDrillClientQueryResult);
+                m_pendingRequests--;
+            }
+            break;
+        default:
+            {
+                DRILL_LOG(LOG_TRACE) << "DrillClientImpl::processQueryStatusResult: Unknown Query State.\n";
+                ret=handleQryError(QRY_INTERNAL_ERROR,
+                        getMessage(ERR_QRY_UNKQRYSTATE),
+                        pDrillClientQueryResult);
+            }
+            break;
+    }
+    return ret;
 }
 
 void DrillClientImpl::handleReadTimeout(const boost::system::error_code & err){
@@ -809,6 +815,14 @@ void DrillClientImpl::handleRead(ByteBuf_t _buf,
                 }
                 return;
             }
+        }else if(!error && msg.m_rpc_type==exec::user::QUERY_DATA){
+            if(processQueryData(allocatedBuffer, msg)!=QRY_SUCCESS){
+                if(m_pendingRequests!=0){
+                    boost::lock_guard<boost::mutex> lock(this->m_dcMutex);
+                    getNextResult();
+                }
+                return;
+            }
         }else if(!error && msg.m_rpc_type==exec::user::QUERY_HANDLE){
             if(processQueryId(allocatedBuffer, msg)!=QRY_SUCCESS){
                 if(m_pendingRequests!=0){
@@ -820,6 +834,7 @@ void DrillClientImpl::handleRead(ByteBuf_t _buf,
         }else if(!error && msg.m_rpc_type==exec::user::ACK){
             // Cancel requests will result in an ACK sent back.
             // Consume silently
+            delete allocatedBuffer;
             if(m_pendingRequests!=0){
                 boost::lock_guard<boost::mutex> lock(this->m_dcMutex);
                 getNextResult();
@@ -830,6 +845,7 @@ void DrillClientImpl::handleRead(ByteBuf_t _buf,
             if(error){
                 // We have a socket read error, but we do not know which query this is for.
                 // Signal ALL pending queries that they should stop waiting.
+                delete allocatedBuffer;
                 DRILL_LOG(LOG_TRACE) << "read error: " << error << std::endl;
                 handleQryError(QRY_COMM_ERROR, getMessage(ERR_QRY_COMMERR, error.message().c_str()), NULL);
                 return;
@@ -851,10 +867,11 @@ void DrillClientImpl::handleRead(ByteBuf_t _buf,
                         DRILL_LOG(LOG_TRACE) << "DrillClientImpl::handleRead: Handshake response from server. Ignore.\n";
                     }
                 }else{
-                DRILL_LOG(LOG_TRACE) << "DrillClientImpl::handleRead: ERR_QRY_INVRPCTYPE. "
+                    DRILL_LOG(LOG_TRACE) << "DrillClientImpl::handleRead: ERR_QRY_INVRPCTYPE. "
                         << "QueryResult returned " << msg.m_rpc_type << std::endl;
-                handleQryError(QRY_INTERNAL_ERROR, getMessage(ERR_QRY_INVRPCTYPE, msg.m_rpc_type), NULL);
+                    handleQryError(QRY_INTERNAL_ERROR, getMessage(ERR_QRY_INVRPCTYPE, msg.m_rpc_type), NULL);
                 }
+                delete allocatedBuffer;
                 return;
             }
         }
@@ -874,21 +891,25 @@ void DrillClientImpl::handleRead(ByteBuf_t _buf,
     return;
 }
 
-status_t DrillClientImpl::validateMessage(InBoundRpcMessage& msg, exec::shared::QueryResult& qr, std::string& valErr){
+status_t DrillClientImpl::validateDataMessage(InBoundRpcMessage& msg, exec::shared::QueryData& qd, std::string& valErr){
     if(msg.m_mode == exec::rpc::RESPONSE_FAILURE){
         valErr=getMessage(ERR_QRY_RESPFAIL);
         return QRY_FAILURE;
     }
-    if(qr.query_state()== exec::shared::QueryResult_QueryState_UNKNOWN_QUERY){
-        valErr=getMessage(ERR_QRY_UNKQRY);
+    if(qd.def().carries_two_byte_selection_vector() == true){
+        valErr=getMessage(ERR_QRY_SELVEC2);
         return QRY_FAILURE;
     }
-    if(qr.query_state()== exec::shared::QueryResult_QueryState_CANCELED){
-        valErr=getMessage(ERR_QRY_CANCELED);
+    return QRY_SUCCESS;
+}
+
+status_t DrillClientImpl::validateResultMessage(InBoundRpcMessage& msg, exec::shared::QueryResult& qr, std::string& valErr){
+    if(msg.m_mode == exec::rpc::RESPONSE_FAILURE){
+        valErr=getMessage(ERR_QRY_RESPFAIL);
         return QRY_FAILURE;
     }
-    if(qr.def().carries_two_byte_selection_vector() == true){
-        valErr=getMessage(ERR_QRY_SELVEC2);
+    if(qr.query_state()==exec::shared::QueryResult_QueryState_CANCELED){
+        valErr=getMessage(ERR_QRY_CANCELED);
         return QRY_FAILURE;
     }
     return QRY_SUCCESS;
@@ -948,7 +969,9 @@ status_t DrillClientImpl::handleTerminatedQryState(
         std::string msg,
         DrillClientQueryResult* pQueryResult){
     assert(pQueryResult!=NULL);
-    if(status!=QRY_COMPLETED){
+    if(status==QRY_COMPLETED){
+        pQueryResult->signalComplete();
+    }else{
         // set query error only if queries did not complete successfully
         DrillClientError* pErr = new DrillClientError(status, DrillClientError::QRY_ERROR_START+status, msg);
         pQueryResult->signalError(pErr);
@@ -957,18 +980,6 @@ status_t DrillClientImpl::handleTerminatedQryState(
 }
 
 
-void DrillClientImpl::clearCancelledEntries(){
-
-    std::map<int, DrillClientQueryResult*>::iterator iter;
-    boost::lock_guard<boost::mutex> lock(m_dcMutex);
-
-    if(!m_cancelledQueries.empty()){
-        std::set<exec::shared::QueryId*, compareQueryId>::iterator it;
-        m_cancelledQueries.erase(m_cancelledQueries.begin(), m_cancelledQueries.end());
-    }
-}
-
-
 void DrillClientImpl::clearMapEntries(DrillClientQueryResult* pQueryResult){
     std::map<int, DrillClientQueryResult*>::iterator iter;
     boost::lock_guard<boost::mutex> lock(m_dcMutex);
@@ -981,13 +992,6 @@ void DrillClientImpl::clearMapEntries(DrillClientQueryResult* pQueryResult){
         }
     }
     if(!m_queryResults.empty()){
-        // Save the query id and state and free when the query is complete
-        if(pQueryResult->getQueryState()!=exec::shared::QueryResult_QueryState_COMPLETED
-                && pQueryResult->getQueryState()!=exec::shared::QueryResult_QueryState_FAILED){
-            exec::shared::QueryId* pQueryId=new exec::shared::QueryId();
-            pQueryId->CopyFrom(pQueryResult->getQueryId());
-            m_cancelledQueries.insert(pQueryId);
-        }
         std::map<exec::shared::QueryId*, DrillClientQueryResult*, compareQueryId>::iterator it;
         for(it=m_queryResults.begin(); it!=m_queryResults.end(); it++) {
             if(pQueryResult==(DrillClientQueryResult*)it->second){
@@ -1017,7 +1021,7 @@ void DrillClientImpl::sendCancel(exec::shared::QueryId* pQueryId){
 
 // This COPIES the FieldMetadata definition for the record batch.  ColumnDefs held by this
 // class are used by the async callbacks.
-status_t DrillClientQueryResult::setupColumnDefs(exec::shared::QueryResult* pQueryResult) {
+status_t DrillClientQueryResult::setupColumnDefs(exec::shared::QueryData* pQueryData) {
     bool hasSchemaChanged=false;
     bool isFirstIter=false;
     boost::lock_guard<boost::mutex> schLock(this->m_schemaMutex);
@@ -1035,11 +1039,11 @@ status_t DrillClientQueryResult::setupColumnDefs(exec::shared::QueryResult* pQue
         }
     }
     m_columnDefs->clear();
-    size_t numFields=pQueryResult->def().field_size();
+    size_t numFields=pQueryData->def().field_size();
     if (numFields > 0){
         for(size_t i=0; i<numFields; i++){
             Drill::FieldMetadata* fmd= new Drill::FieldMetadata;
-            fmd->set(pQueryResult->def().field(i));
+            fmd->set(pQueryData->def().field(i));
             this->m_columnDefs->push_back(fmd);
 
             //Look for changes in the vector and trigger a Schema change event if necessary.
@@ -1077,19 +1081,21 @@ status_t DrillClientQueryResult::defaultQueryResultsListener(void* ctx,
     DRILL_LOG(LOG_TRACE) << "Query result listener called" << std::endl;
     //check if the query has been canceled. IF so then return FAILURE. Caller will send cancel to the server.
     if(this->m_bCancel){
-        delete b;
+        if(b!=NULL) delete b;
         return QRY_FAILURE;
     }
     if (!err) {
         // signal the cond var
         {
-            #ifdef DEBUG
-            DRILL_LOG(LOG_DEBUG)<<debugPrintQid(b->getQueryResult()->query_id())
-                << "Query result listener saved result to queue." << std::endl;
-            #endif
-            boost::lock_guard<boost::mutex> cvLock(this->m_cvMutex);
-            this->m_recordBatches.push(b);
-            this->m_bHasData=true;
+            if(b!=NULL){
+#ifdef DEBUG
+                DRILL_LOG(LOG_DEBUG)<<debugPrintQid(b->getQueryResult()->query_id())
+                    << "Query result listener saved result to queue." << std::endl;
+#endif
+                boost::lock_guard<boost::mutex> cvLock(this->m_cvMutex);
+                this->m_recordBatches.push(b);
+                this->m_bHasData=true;
+            }
         }
         m_cv.notify_one();
     }else{
@@ -1100,11 +1106,11 @@ status_t DrillClientQueryResult::defaultQueryResultsListener(void* ctx,
 
 RecordBatch*  DrillClientQueryResult::peekNext(){
     RecordBatch* pRecordBatch=NULL;
+    boost::unique_lock<boost::mutex> cvLock(this->m_cvMutex);
     //if no more data, return NULL;
     if(!m_bIsQueryPending) return NULL;
-    boost::unique_lock<boost::mutex> cvLock(this->m_cvMutex);
     DRILL_LOG(LOG_TRACE) << "Synchronous read waiting for data." << std::endl;
-    while(!this->m_bHasData && !m_bHasError) {
+    while(!this->m_bHasData && !m_bHasError && m_bIsQueryPending) {
         this->m_cv.wait(cvLock);
     }
     // READ but not remove first element from queue
@@ -1114,6 +1120,7 @@ RecordBatch*  DrillClientQueryResult::peekNext(){
 
 RecordBatch*  DrillClientQueryResult::getNext() {
     RecordBatch* pRecordBatch=NULL;
+    boost::unique_lock<boost::mutex> cvLock(this->m_cvMutex);
     //if no more data, return NULL;
     if(!m_bIsQueryPending){
         DRILL_LOG(LOG_TRACE) << "Query is done." << std::endl;
@@ -1123,9 +1130,8 @@ RecordBatch*  DrillClientQueryResult::getNext() {
         return NULL;
     }
 
-    boost::unique_lock<boost::mutex> cvLock(this->m_cvMutex);
     DRILL_LOG(LOG_TRACE) << "Synchronous read waiting for data." << std::endl;
-    while(!this->m_bHasData && !m_bHasError){
+    while(!this->m_bHasData && !m_bHasError && m_bIsQueryPending){
         this->m_cv.wait(cvLock);
     }
     // remove first element from queue
@@ -1133,16 +1139,16 @@ RecordBatch*  DrillClientQueryResult::getNext() {
     this->m_recordBatches.pop();
     this->m_bHasData=!this->m_recordBatches.empty();
     // if vector is empty, set m_bHasDataPending to false;
-    m_bIsQueryPending=!(this->m_recordBatches.empty()&&m_bIsLastChunk);
+    m_bIsQueryPending=!(this->m_recordBatches.empty()&&m_queryState==exec::shared::QueryResult_QueryState_COMPLETED);
     return pRecordBatch;
 }
 
 // Blocks until data is available
 void DrillClientQueryResult::waitForData() {
+    boost::unique_lock<boost::mutex> cvLock(this->m_cvMutex);
     //if no more data, return NULL;
     if(!m_bIsQueryPending) return;
-    boost::unique_lock<boost::mutex> cvLock(this->m_cvMutex);
-    while(!this->m_bHasData && !m_bHasError) {
+    while(!this->m_bHasData && !m_bHasError && m_bIsQueryPending) {
         this->m_cv.wait(cvLock);
     }
 }
@@ -1164,9 +1170,9 @@ void DrillClientQueryResult::signalError(DrillClientError* pErr){
         }else{
             defaultQueryResultsListener(this, NULL, pErr);
         }
-        m_bIsQueryPending=false;
         {
             boost::lock_guard<boost::mutex> cvLock(this->m_cvMutex);
+            m_bIsQueryPending=false;
             m_bHasData=false;
             m_bHasError=true;
         }
@@ -1176,6 +1182,24 @@ void DrillClientQueryResult::signalError(DrillClientError* pErr){
     return;
 }
 
+void DrillClientQueryResult::signalComplete(){
+    pfnQueryResultsListener pResultsListener=this->m_pResultsListener;
+    if(pResultsListener!=NULL){
+        pResultsListener(this, NULL, NULL);
+    }else{
+        defaultQueryResultsListener(this, NULL, NULL);
+    }
+    {
+        boost::lock_guard<boost::mutex> cvLock(this->m_cvMutex);
+        m_bIsQueryPending=false;
+        m_bIsQueryPending=!(this->m_recordBatches.empty()&&m_queryState==exec::shared::QueryResult_QueryState_COMPLETED);
+        m_bHasError=false;
+    }
+    //Signal the cv in case there is a client waiting for data already.
+    m_cv.notify_one();
+    return;
+}
+
 void DrillClientQueryResult::clearAndDestroy(){
     //free memory allocated for FieldMetadata objects saved in m_columnDefs;
     if(!m_columnDefs->empty()){
@@ -1207,7 +1231,7 @@ void DrillClientQueryResult::clearAndDestroy(){
     }
     if(m_pError!=NULL){
         delete m_pError; m_pError=NULL;
-}
+    }
 }
 
 char ZookeeperImpl::s_drillRoot[]="/drill/";
diff --git a/contrib/native/client/src/clientlib/drillClientImpl.hpp b/contrib/native/client/src/clientlib/drillClientImpl.hpp
index 33f81dbbf..95fe92254 100644
--- a/contrib/native/client/src/clientlib/drillClientImpl.hpp
+++ b/contrib/native/client/src/clientlib/drillClientImpl.hpp
@@ -127,14 +127,19 @@ class DrillClientQueryResult{
 
     void setQueryState(exec::shared::QueryResult_QueryState s){ m_queryState = s;}
     exec::shared::QueryResult_QueryState getQueryState(){ return m_queryState;}
+    void setIsQueryPending(bool isPending){
+        boost::lock_guard<boost::mutex> cvLock(this->m_cvMutex);
+        m_bIsQueryPending=isPending;
+    }
 
     private:
-    status_t setupColumnDefs(exec::shared::QueryResult* pQueryResult);
+    status_t setupColumnDefs(exec::shared::QueryData* pQueryData);
     status_t defaultQueryResultsListener(void* ctx, RecordBatch* b, DrillClientError* err);
     // Construct a DrillClientError object, set the appropriate state and signal any listeners, condition variables.
     // Also used when a query is cancelled or when a query completed response is received.
     // Error object is now owned by the DrillClientQueryResult object.
     void signalError(DrillClientError* pErr);
+    void signalComplete();
     void clearAndDestroy();
 
 
@@ -212,7 +217,6 @@ class DrillClientImpl{
                 this->m_pWork = NULL;
             }
 
-            clearCancelledEntries();
             m_deadlineTimer.cancel();
             m_io_service.stop();
             boost::system::error_code ignorederr;
@@ -272,13 +276,16 @@ class DrillClientImpl{
                 InBoundRpcMessage& msg,
                 boost::system::error_code& error);
         status_t processQueryResult(AllocatedBufferPtr allocatedBuffer, InBoundRpcMessage& msg);
+        status_t processQueryData(AllocatedBufferPtr allocatedBuffer, InBoundRpcMessage& msg);
         status_t processCancelledQueryResult( exec::shared::QueryId& qid, exec::shared::QueryResult* qr);
         status_t processQueryId(AllocatedBufferPtr allocatedBuffer, InBoundRpcMessage& msg );
+        DrillClientQueryResult* findQueryResult(exec::shared::QueryId& qid);
         status_t processQueryStatusResult( exec::shared::QueryResult* qr,
                 DrillClientQueryResult* pDrillClientQueryResult);
         void handleReadTimeout(const boost::system::error_code & err);
         void handleRead(ByteBuf_t _buf, const boost::system::error_code & err, size_t bytes_transferred) ;
-        status_t validateMessage(InBoundRpcMessage& msg, exec::shared::QueryResult& qr, std::string& valError);
+        status_t validateDataMessage(InBoundRpcMessage& msg, exec::shared::QueryData& qd, std::string& valError);
+        status_t validateResultMessage(InBoundRpcMessage& msg, exec::shared::QueryResult& qr, std::string& valError);
         connectionStatus_t handleConnError(connectionStatus_t status, std::string msg);
         status_t handleQryError(status_t status, std::string msg, DrillClientQueryResult* pQueryResult);
         status_t handleQryError(status_t status,
@@ -291,7 +298,6 @@ class DrillClientImpl{
                 DrillClientQueryResult* pQueryResult);
         void broadcastError(DrillClientError* pErr);
         void clearMapEntries(DrillClientQueryResult* pQueryResult);
-        void clearCancelledEntries();
         void sendAck(InBoundRpcMessage& msg, bool isOk);
         void sendCancel(exec::shared::QueryId* pQueryId);
 
@@ -335,11 +341,6 @@ class DrillClientImpl{
 
         // Map of query id to query result for currently executing queries
         std::map<exec::shared::QueryId*, DrillClientQueryResult*, compareQueryId> m_queryResults;
-        //
-        // State for every Query id whose queries have result data pending but which
-        // have been cancelled and whose resources have been released by the client application.
-        // The entry is cleared when the state changes to completed or failed.
-        std::set<exec::shared::QueryId*, compareQueryId> m_cancelledQueries;
 
 };
 
diff --git a/contrib/native/client/src/clientlib/recordBatch.cpp b/contrib/native/client/src/clientlib/recordBatch.cpp
index 44140b253..c6c033b79 100644
--- a/contrib/native/client/src/clientlib/recordBatch.cpp
+++ b/contrib/native/client/src/clientlib/recordBatch.cpp
@@ -306,7 +306,7 @@ ret_t FieldBatch::loadNull(size_t nRecords){
     return RET_SUCCESS;
 }
     
-RecordBatch::RecordBatch(exec::shared::QueryResult* pResult, AllocatedBufferPtr r, ByteBuf_t b)
+RecordBatch::RecordBatch(exec::shared::QueryData* pResult, AllocatedBufferPtr r, ByteBuf_t b)
     :m_fieldDefs(new(std::vector<Drill::FieldMetadata*>)){
         m_pQueryResult=pResult;
         m_pRecordBatchDef=&pResult->def();
@@ -398,7 +398,7 @@ size_t RecordBatch::getNumFields(){
 }
 
 bool RecordBatch::isLastChunk(){
-    return m_pQueryResult->is_last_chunk(); 
+    return false;
 }
 
 
diff --git a/contrib/native/client/src/include/drill/common.hpp b/contrib/native/client/src/include/drill/common.hpp
index 656069271..72b9a9873 100644
--- a/contrib/native/client/src/include/drill/common.hpp
+++ b/contrib/native/client/src/include/drill/common.hpp
@@ -34,7 +34,7 @@
 #include <vector>
 #include <boost/shared_ptr.hpp>
 
-#define DRILL_RPC_VERSION 3
+#define DRILL_RPC_VERSION 4
 
 #define LENGTH_PREFIX_MAX_LENGTH 5
 #define LEN_PREFIX_BUFLEN LENGTH_PREFIX_MAX_LENGTH
diff --git a/contrib/native/client/src/include/drill/drillClient.hpp b/contrib/native/client/src/include/drill/drillClient.hpp
index 9289df3c7..c288c7073 100644
--- a/contrib/native/client/src/include/drill/drillClient.hpp
+++ b/contrib/native/client/src/include/drill/drillClient.hpp
@@ -167,7 +167,9 @@ typedef void* QueryHandle_t;
  * Query Results listener callback. This function is called for every record batch after it has
  * been received and decoded. The listener function should return a status.
  * If the listener returns failure, the query will be canceled.
- *
+ * The listener is also called one last time when the query is completed or gets an error. In that
+ * case the RecordBatch Parameter is NULL. The DrillClientError parameter is NULL is there was no
+ * error oterwise it will have a valid DrillClientError object.
  * DrillClientQueryResult will hold a listener & listener contxt for the call back function
  */
 typedef status_t (*pfnQueryResultsListener)(QueryHandle_t ctx, RecordBatch* b, DrillClientError* err);
diff --git a/contrib/native/client/src/include/drill/recordBatch.hpp b/contrib/native/client/src/include/drill/recordBatch.hpp
index 92a4c3ad6..12cbad46d 100644
--- a/contrib/native/client/src/include/drill/recordBatch.hpp
+++ b/contrib/native/client/src/include/drill/recordBatch.hpp
@@ -56,6 +56,7 @@ namespace exec{
         class SerializedField;
         class RecordBatchDef;
         class QueryResult;
+        class QueryData;
     };
 };
 
@@ -863,7 +864,7 @@ class DECLSPEC_DRILL_CLIENT RecordBatch{
         //m_allocatedBuffer is the memory block allocated to hold the incoming RPC message. Record Batches operate on
         //slices of the allocated buffer. The first slice (the first Field Batch), begins at m_buffer. Data in the
         //allocated buffer before m_buffer is mostly the RPC header, and the QueryResult object.
-        RecordBatch(exec::shared::QueryResult* pResult, AllocatedBufferPtr r, ByteBuf_t b);
+        RecordBatch(exec::shared::QueryData* pResult, AllocatedBufferPtr r, ByteBuf_t b);
 
         ~RecordBatch();
 
@@ -876,7 +877,7 @@ class DECLSPEC_DRILL_CLIENT RecordBatch{
         size_t getNumRecords(){ return m_numRecords;}
         std::vector<FieldBatch*>& getFields(){ return m_fields;}
         size_t getNumFields();
-        bool isLastChunk();
+        DEPRECATED bool isLastChunk();
 
         boost::shared_ptr<std::vector<Drill::FieldMetadata*> > getColumnDefs(){ return m_fieldDefs;}
 
@@ -902,10 +903,10 @@ class DECLSPEC_DRILL_CLIENT RecordBatch{
         bool hasSchemaChanged(){ return m_bHasSchemaChanged;}
 
         #ifdef DEBUG
-        const exec::shared::QueryResult* getQueryResult(){ return this->m_pQueryResult;}
+        const exec::shared::QueryData* getQueryResult(){ return this->m_pQueryResult;}
         #endif
     private:
-        const exec::shared::QueryResult* m_pQueryResult;
+        const exec::shared::QueryData* m_pQueryResult;
         const exec::shared::RecordBatchDef* m_pRecordBatchDef;
         AllocatedBufferPtr m_allocatedBuffer;
         ByteBuf_t m_buffer;
diff --git a/contrib/native/client/src/protobuf/BitData.pb.cc b/contrib/native/client/src/protobuf/BitData.pb.cc
index ef4f99d84..57bceff82 100644
--- a/contrib/native/client/src/protobuf/BitData.pb.cc
+++ b/contrib/native/client/src/protobuf/BitData.pb.cc
@@ -43,10 +43,9 @@ void protobuf_AssignDesc_BitData_2eproto() {
       "BitData.proto");
   GOOGLE_CHECK(file != NULL);
   BitClientHandshake_descriptor_ = file->message_type(0);
-  static const int BitClientHandshake_offsets_[3] = {
+  static const int BitClientHandshake_offsets_[2] = {
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(BitClientHandshake, rpc_version_),
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(BitClientHandshake, channel_),
-    GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(BitClientHandshake, handle_),
   };
   BitClientHandshake_reflection_ =
     new ::google::protobuf::internal::GeneratedMessageReflection(
@@ -75,8 +74,10 @@ void protobuf_AssignDesc_BitData_2eproto() {
       ::google::protobuf::MessageFactory::generated_factory(),
       sizeof(BitServerHandshake));
   FragmentRecordBatch_descriptor_ = file->message_type(2);
-  static const int FragmentRecordBatch_offsets_[6] = {
-    GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(FragmentRecordBatch, handle_),
+  static const int FragmentRecordBatch_offsets_[8] = {
+    GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(FragmentRecordBatch, query_id_),
+    GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(FragmentRecordBatch, receiving_major_fragment_id_),
+    GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(FragmentRecordBatch, receiving_minor_fragment_id_),
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(FragmentRecordBatch, sending_major_fragment_id_),
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(FragmentRecordBatch, sending_minor_fragment_id_),
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(FragmentRecordBatch, def_),
@@ -138,20 +139,21 @@ void protobuf_AddDesc_BitData_2eproto() {
   ::google::protobuf::DescriptorPool::InternalAddGeneratedFile(
     "\n\rBitData.proto\022\rexec.bit.data\032\025Executio"
     "nProtos.proto\032\022Coordination.proto\032\023UserB"
-    "itShared.proto\"\207\001\n\022BitClientHandshake\022\023\n"
-    "\013rpc_version\030\001 \001(\005\0222\n\007channel\030\002 \001(\0162\027.ex"
-    "ec.shared.RpcChannel:\010BIT_DATA\022(\n\006handle"
-    "\030\003 \001(\0132\030.exec.bit.FragmentHandle\")\n\022BitS"
-    "erverHandshake\022\023\n\013rpc_version\030\001 \001(\005\"\342\001\n\023"
-    "FragmentRecordBatch\022(\n\006handle\030\001 \001(\0132\030.ex"
-    "ec.bit.FragmentHandle\022!\n\031sending_major_f"
-    "ragment_id\030\002 \001(\005\022!\n\031sending_minor_fragme"
-    "nt_id\030\003 \001(\005\022(\n\003def\030\004 \001(\0132\033.exec.shared.R"
-    "ecordBatchDef\022\023\n\013isLastBatch\030\005 \001(\010\022\034\n\ris"
-    "OutOfMemory\030\006 \001(\010:\005false*D\n\007RpcType\022\r\n\tH"
-    "ANDSHAKE\020\000\022\007\n\003ACK\020\001\022\013\n\007GOODBYE\020\002\022\024\n\020REQ_"
-    "RECORD_BATCH\020\003B(\n\033org.apache.drill.exec."
-    "protoB\007BitDataH\001", 616);
+    "itShared.proto\"]\n\022BitClientHandshake\022\023\n\013"
+    "rpc_version\030\001 \001(\005\0222\n\007channel\030\002 \001(\0162\027.exe"
+    "c.shared.RpcChannel:\010BIT_DATA\")\n\022BitServ"
+    "erHandshake\022\023\n\013rpc_version\030\001 \001(\005\"\252\002\n\023Fra"
+    "gmentRecordBatch\022&\n\010query_id\030\001 \001(\0132\024.exe"
+    "c.shared.QueryId\022#\n\033receiving_major_frag"
+    "ment_id\030\002 \001(\005\022#\n\033receiving_minor_fragmen"
+    "t_id\030\003 \003(\005\022!\n\031sending_major_fragment_id\030"
+    "\004 \001(\005\022!\n\031sending_minor_fragment_id\030\005 \001(\005"
+    "\022(\n\003def\030\006 \001(\0132\033.exec.shared.RecordBatchD"
+    "ef\022\023\n\013isLastBatch\030\007 \001(\010\022\034\n\risOutOfMemory"
+    "\030\010 \001(\010:\005false*D\n\007RpcType\022\r\n\tHANDSHAKE\020\000\022"
+    "\007\n\003ACK\020\001\022\013\n\007GOODBYE\020\002\022\024\n\020REQ_RECORD_BATC"
+    "H\020\003B(\n\033org.apache.drill.exec.protoB\007BitD"
+    "ataH\001", 645);
   ::google::protobuf::MessageFactory::InternalRegisterGeneratedFile(
     "BitData.proto", &protobuf_RegisterTypes);
   BitClientHandshake::default_instance_ = new BitClientHandshake();
@@ -191,7 +193,6 @@ bool RpcType_IsValid(int value) {
 #ifndef _MSC_VER
 const int BitClientHandshake::kRpcVersionFieldNumber;
 const int BitClientHandshake::kChannelFieldNumber;
-const int BitClientHandshake::kHandleFieldNumber;
 #endif  // !_MSC_VER
 
 BitClientHandshake::BitClientHandshake()
@@ -200,7 +201,6 @@ BitClientHandshake::BitClientHandshake()
 }
 
 void BitClientHandshake::InitAsDefaultInstance() {
-  handle_ = const_cast< ::exec::bit::FragmentHandle*>(&::exec::bit::FragmentHandle::default_instance());
 }
 
 BitClientHandshake::BitClientHandshake(const BitClientHandshake& from)
@@ -213,7 +213,6 @@ void BitClientHandshake::SharedCtor() {
   _cached_size_ = 0;
   rpc_version_ = 0;
   channel_ = 1;
-  handle_ = NULL;
   ::memset(_has_bits_, 0, sizeof(_has_bits_));
 }
 
@@ -223,7 +222,6 @@ BitClientHandshake::~BitClientHandshake() {
 
 void BitClientHandshake::SharedDtor() {
   if (this != default_instance_) {
-    delete handle_;
   }
 }
 
@@ -252,9 +250,6 @@ void BitClientHandshake::Clear() {
   if (_has_bits_[0 / 32] & (0xffu << (0 % 32))) {
     rpc_version_ = 0;
     channel_ = 1;
-    if (has_handle()) {
-      if (handle_ != NULL) handle_->::exec::bit::FragmentHandle::Clear();
-    }
   }
   ::memset(_has_bits_, 0, sizeof(_has_bits_));
   mutable_unknown_fields()->Clear();
@@ -298,20 +293,6 @@ bool BitClientHandshake::MergePartialFromCodedStream(
         } else {
           goto handle_uninterpreted;
         }
-        if (input->ExpectTag(26)) goto parse_handle;
-        break;
-      }
-
-      // optional .exec.bit.FragmentHandle handle = 3;
-      case 3: {
-        if (::google::protobuf::internal::WireFormatLite::GetTagWireType(tag) ==
-            ::google::protobuf::internal::WireFormatLite::WIRETYPE_LENGTH_DELIMITED) {
-         parse_handle:
-          DO_(::google::protobuf::internal::WireFormatLite::ReadMessageNoVirtual(
-               input, mutable_handle()));
-        } else {
-          goto handle_uninterpreted;
-        }
         if (input->ExpectAtEnd()) return true;
         break;
       }
@@ -345,12 +326,6 @@ void BitClientHandshake::SerializeWithCachedSizes(
       2, this->channel(), output);
   }
 
-  // optional .exec.bit.FragmentHandle handle = 3;
-  if (has_handle()) {
-    ::google::protobuf::internal::WireFormatLite::WriteMessageMaybeToArray(
-      3, this->handle(), output);
-  }
-
   if (!unknown_fields().empty()) {
     ::google::protobuf::internal::WireFormat::SerializeUnknownFields(
         unknown_fields(), output);
@@ -370,13 +345,6 @@ void BitClientHandshake::SerializeWithCachedSizes(
       2, this->channel(), target);
   }
 
-  // optional .exec.bit.FragmentHandle handle = 3;
-  if (has_handle()) {
-    target = ::google::protobuf::internal::WireFormatLite::
-      WriteMessageNoVirtualToArray(
-        3, this->handle(), target);
-  }
-
   if (!unknown_fields().empty()) {
     target = ::google::protobuf::internal::WireFormat::SerializeUnknownFieldsToArray(
         unknown_fields(), target);
@@ -401,13 +369,6 @@ int BitClientHandshake::ByteSize() const {
         ::google::protobuf::internal::WireFormatLite::EnumSize(this->channel());
     }
 
-    // optional .exec.bit.FragmentHandle handle = 3;
-    if (has_handle()) {
-      total_size += 1 +
-        ::google::protobuf::internal::WireFormatLite::MessageSizeNoVirtual(
-          this->handle());
-    }
-
   }
   if (!unknown_fields().empty()) {
     total_size +=
@@ -441,9 +402,6 @@ void BitClientHandshake::MergeFrom(const BitClientHandshake& from) {
     if (from.has_channel()) {
       set_channel(from.channel());
     }
-    if (from.has_handle()) {
-      mutable_handle()->::exec::bit::FragmentHandle::MergeFrom(from.handle());
-    }
   }
   mutable_unknown_fields()->MergeFrom(from.unknown_fields());
 }
@@ -469,7 +427,6 @@ void BitClientHandshake::Swap(BitClientHandshake* other) {
   if (other != this) {
     std::swap(rpc_version_, other->rpc_version_);
     std::swap(channel_, other->channel_);
-    std::swap(handle_, other->handle_);
     std::swap(_has_bits_[0], other->_has_bits_[0]);
     _unknown_fields_.Swap(&other->_unknown_fields_);
     std::swap(_cached_size_, other->_cached_size_);
@@ -696,7 +653,9 @@ void BitServerHandshake::Swap(BitServerHandshake* other) {
 // ===================================================================
 
 #ifndef _MSC_VER
-const int FragmentRecordBatch::kHandleFieldNumber;
+const int FragmentRecordBatch::kQueryIdFieldNumber;
+const int FragmentRecordBatch::kReceivingMajorFragmentIdFieldNumber;
+const int FragmentRecordBatch::kReceivingMinorFragmentIdFieldNumber;
 const int FragmentRecordBatch::kSendingMajorFragmentIdFieldNumber;
 const int FragmentRecordBatch::kSendingMinorFragmentIdFieldNumber;
 const int FragmentRecordBatch::kDefFieldNumber;
@@ -710,7 +669,7 @@ FragmentRecordBatch::FragmentRecordBatch()
 }
 
 void FragmentRecordBatch::InitAsDefaultInstance() {
-  handle_ = const_cast< ::exec::bit::FragmentHandle*>(&::exec::bit::FragmentHandle::default_instance());
+  query_id_ = const_cast< ::exec::shared::QueryId*>(&::exec::shared::QueryId::default_instance());
   def_ = const_cast< ::exec::shared::RecordBatchDef*>(&::exec::shared::RecordBatchDef::default_instance());
 }
 
@@ -722,7 +681,8 @@ FragmentRecordBatch::FragmentRecordBatch(const FragmentRecordBatch& from)
 
 void FragmentRecordBatch::SharedCtor() {
   _cached_size_ = 0;
-  handle_ = NULL;
+  query_id_ = NULL;
+  receiving_major_fragment_id_ = 0;
   sending_major_fragment_id_ = 0;
   sending_minor_fragment_id_ = 0;
   def_ = NULL;
@@ -737,7 +697,7 @@ FragmentRecordBatch::~FragmentRecordBatch() {
 
 void FragmentRecordBatch::SharedDtor() {
   if (this != default_instance_) {
-    delete handle_;
+    delete query_id_;
     delete def_;
   }
 }
@@ -765,9 +725,10 @@ FragmentRecordBatch* FragmentRecordBatch::New() const {
 
 void FragmentRecordBatch::Clear() {
   if (_has_bits_[0 / 32] & (0xffu << (0 % 32))) {
-    if (has_handle()) {
-      if (handle_ != NULL) handle_->::exec::bit::FragmentHandle::Clear();
+    if (has_query_id()) {
+      if (query_id_ != NULL) query_id_->::exec::shared::QueryId::Clear();
     }
+    receiving_major_fragment_id_ = 0;
     sending_major_fragment_id_ = 0;
     sending_minor_fragment_id_ = 0;
     if (has_def()) {
@@ -776,6 +737,7 @@ void FragmentRecordBatch::Clear() {
     islastbatch_ = false;
     isoutofmemory_ = false;
   }
+  receiving_minor_fragment_id_.Clear();
   ::memset(_has_bits_, 0, sizeof(_has_bits_));
   mutable_unknown_fields()->Clear();
 }
@@ -786,23 +748,61 @@ bool FragmentRecordBatch::MergePartialFromCodedStream(
   ::google::protobuf::uint32 tag;
   while ((tag = input->ReadTag()) != 0) {
     switch (::google::protobuf::internal::WireFormatLite::GetTagFieldNumber(tag)) {
-      // optional .exec.bit.FragmentHandle handle = 1;
+      // optional .exec.shared.QueryId query_id = 1;
       case 1: {
         if (::google::protobuf::internal::WireFormatLite::GetTagWireType(tag) ==
             ::google::protobuf::internal::WireFormatLite::WIRETYPE_LENGTH_DELIMITED) {
           DO_(::google::protobuf::internal::WireFormatLite::ReadMessageNoVirtual(
-               input, mutable_handle()));
+               input, mutable_query_id()));
         } else {
           goto handle_uninterpreted;
         }
-        if (input->ExpectTag(16)) goto parse_sending_major_fragment_id;
+        if (input->ExpectTag(16)) goto parse_receiving_major_fragment_id;
         break;
       }
 
-      // optional int32 sending_major_fragment_id = 2;
+      // optional int32 receiving_major_fragment_id = 2;
       case 2: {
         if (::google::protobuf::internal::WireFormatLite::GetTagWireType(tag) ==
             ::google::protobuf::internal::WireFormatLite::WIRETYPE_VARINT) {
+         parse_receiving_major_fragment_id:
+          DO_((::google::protobuf::internal::WireFormatLite::ReadPrimitive<
+                   ::google::protobuf::int32, ::google::protobuf::internal::WireFormatLite::TYPE_INT32>(
+                 input, &receiving_major_fragment_id_)));
+          set_has_receiving_major_fragment_id();
+        } else {
+          goto handle_uninterpreted;
+        }
+        if (input->ExpectTag(24)) goto parse_receiving_minor_fragment_id;
+        break;
+      }
+
+      // repeated int32 receiving_minor_fragment_id = 3;
+      case 3: {
+        if (::google::protobuf::internal::WireFormatLite::GetTagWireType(tag) ==
+            ::google::protobuf::internal::WireFormatLite::WIRETYPE_VARINT) {
+         parse_receiving_minor_fragment_id:
+          DO_((::google::protobuf::internal::WireFormatLite::ReadRepeatedPrimitive<
+                   ::google::protobuf::int32, ::google::protobuf::internal::WireFormatLite::TYPE_INT32>(
+                 1, 24, input, this->mutable_receiving_minor_fragment_id())));
+        } else if (::google::protobuf::internal::WireFormatLite::GetTagWireType(tag)
+                   == ::google::protobuf::internal::WireFormatLite::
+                      WIRETYPE_LENGTH_DELIMITED) {
+          DO_((::google::protobuf::internal::WireFormatLite::ReadPackedPrimitiveNoInline<
+                   ::google::protobuf::int32, ::google::protobuf::internal::WireFormatLite::TYPE_INT32>(
+                 input, this->mutable_receiving_minor_fragment_id())));
+        } else {
+          goto handle_uninterpreted;
+        }
+        if (input->ExpectTag(24)) goto parse_receiving_minor_fragment_id;
+        if (input->ExpectTag(32)) goto parse_sending_major_fragment_id;
+        break;
+      }
+
+      // optional int32 sending_major_fragment_id = 4;
+      case 4: {
+        if (::google::protobuf::internal::WireFormatLite::GetTagWireType(tag) ==
+            ::google::protobuf::internal::WireFormatLite::WIRETYPE_VARINT) {
          parse_sending_major_fragment_id:
           DO_((::google::protobuf::internal::WireFormatLite::ReadPrimitive<
                    ::google::protobuf::int32, ::google::protobuf::internal::WireFormatLite::TYPE_INT32>(
@@ -811,12 +811,12 @@ bool FragmentRecordBatch::MergePartialFromCodedStream(
         } else {
           goto handle_uninterpreted;
         }
-        if (input->ExpectTag(24)) goto parse_sending_minor_fragment_id;
+        if (input->ExpectTag(40)) goto parse_sending_minor_fragment_id;
         break;
       }
 
-      // optional int32 sending_minor_fragment_id = 3;
-      case 3: {
+      // optional int32 sending_minor_fragment_id = 5;
+      case 5: {
         if (::google::protobuf::internal::WireFormatLite::GetTagWireType(tag) ==
             ::google::protobuf::internal::WireFormatLite::WIRETYPE_VARINT) {
          parse_sending_minor_fragment_id:
@@ -827,12 +827,12 @@ bool FragmentRecordBatch::MergePartialFromCodedStream(
         } else {
           goto handle_uninterpreted;
         }
-        if (input->ExpectTag(34)) goto parse_def;
+        if (input->ExpectTag(50)) goto parse_def;
         break;
       }
 
-      // optional .exec.shared.RecordBatchDef def = 4;
-      case 4: {
+      // optional .exec.shared.RecordBatchDef def = 6;
+      case 6: {
         if (::google::protobuf::internal::WireFormatLite::GetTagWireType(tag) ==
             ::google::protobuf::internal::WireFormatLite::WIRETYPE_LENGTH_DELIMITED) {
          parse_def:
@@ -841,12 +841,12 @@ bool FragmentRecordBatch::MergePartialFromCodedStream(
         } else {
           goto handle_uninterpreted;
         }
-        if (input->ExpectTag(40)) goto parse_isLastBatch;
+        if (input->ExpectTag(56)) goto parse_isLastBatch;
         break;
       }
 
-      // optional bool isLastBatch = 5;
-      case 5: {
+      // optional bool isLastBatch = 7;
+      case 7: {
         if (::google::protobuf::internal::WireFormatLite::GetTagWireType(tag) ==
             ::google::protobuf::internal::WireFormatLite::WIRETYPE_VARINT) {
          parse_isLastBatch:
@@ -857,12 +857,12 @@ bool FragmentRecordBatch::MergePartialFromCodedStream(
         } else {
           goto handle_uninterpreted;
         }
-        if (input->ExpectTag(48)) goto parse_isOutOfMemory;
+        if (input->ExpectTag(64)) goto parse_isOutOfMemory;
         break;
       }
 
-      // optional bool isOutOfMemory = 6 [default = false];
-      case 6: {
+      // optional bool isOutOfMemory = 8 [default = false];
+      case 8: {
         if (::google::protobuf::internal::WireFormatLite::GetTagWireType(tag) ==
             ::google::protobuf::internal::WireFormatLite::WIRETYPE_VARINT) {
          parse_isOutOfMemory:
@@ -895,36 +895,47 @@ bool FragmentRecordBatch::MergePartialFromCodedStream(
 
 void FragmentRecordBatch::SerializeWithCachedSizes(
     ::google::protobuf::io::CodedOutputStream* output) const {
-  // optional .exec.bit.FragmentHandle handle = 1;
-  if (has_handle()) {
+  // optional .exec.shared.QueryId query_id = 1;
+  if (has_query_id()) {
     ::google::protobuf::internal::WireFormatLite::WriteMessageMaybeToArray(
-      1, this->handle(), output);
+      1, this->query_id(), output);
   }
 
-  // optional int32 sending_major_fragment_id = 2;
+  // optional int32 receiving_major_fragment_id = 2;
+  if (has_receiving_major_fragment_id()) {
+    ::google::protobuf::internal::WireFormatLite::WriteInt32(2, this->receiving_major_fragment_id(), output);
+  }
+
+  // repeated int32 receiving_minor_fragment_id = 3;
+  for (int i = 0; i < this->receiving_minor_fragment_id_size(); i++) {
+    ::google::protobuf::internal::WireFormatLite::WriteInt32(
+      3, this->receiving_minor_fragment_id(i), output);
+  }
+
+  // optional int32 sending_major_fragment_id = 4;
   if (has_sending_major_fragment_id()) {
-    ::google::protobuf::internal::WireFormatLite::WriteInt32(2, this->sending_major_fragment_id(), output);
+    ::google::protobuf::internal::WireFormatLite::WriteInt32(4, this->sending_major_fragment_id(), output);
   }
 
-  // optional int32 sending_minor_fragment_id = 3;
+  // optional int32 sending_minor_fragment_id = 5;
   if (has_sending_minor_fragment_id()) {
-    ::google::protobuf::internal::WireFormatLite::WriteInt32(3, this->sending_minor_fragment_id(), output);
+    ::google::protobuf::internal::WireFormatLite::WriteInt32(5, this->sending_minor_fragment_id(), output);
   }
 
-  // optional .exec.shared.RecordBatchDef def = 4;
+  // optional .exec.shared.RecordBatchDef def = 6;
   if (has_def()) {
     ::google::protobuf::internal::WireFormatLite::WriteMessageMaybeToArray(
-      4, this->def(), output);
+      6, this->def(), output);
   }
 
-  // optional bool isLastBatch = 5;
+  // optional bool isLastBatch = 7;
   if (has_islastbatch()) {
-    ::google::protobuf::internal::WireFormatLite::WriteBool(5, this->islastbatch(), output);
+    ::google::protobuf::internal::WireFormatLite::WriteBool(7, this->islastbatch(), output);
   }
 
-  // optional bool isOutOfMemory = 6 [default = false];
+  // optional bool isOutOfMemory = 8 [default = false];
   if (has_isoutofmemory()) {
-    ::google::protobuf::internal::WireFormatLite::WriteBool(6, this->isoutofmemory(), output);
+    ::google::protobuf::internal::WireFormatLite::WriteBool(8, this->isoutofmemory(), output);
   }
 
   if (!unknown_fields().empty()) {
@@ -935,38 +946,49 @@ void FragmentRecordBatch::SerializeWithCachedSizes(
 
 ::google::protobuf::uint8* FragmentRecordBatch::SerializeWithCachedSizesToArray(
     ::google::protobuf::uint8* target) const {
-  // optional .exec.bit.FragmentHandle handle = 1;
-  if (has_handle()) {
+  // optional .exec.shared.QueryId query_id = 1;
+  if (has_query_id()) {
     target = ::google::protobuf::internal::WireFormatLite::
       WriteMessageNoVirtualToArray(
-        1, this->handle(), target);
+        1, this->query_id(), target);
+  }
+
+  // optional int32 receiving_major_fragment_id = 2;
+  if (has_receiving_major_fragment_id()) {
+    target = ::google::protobuf::internal::WireFormatLite::WriteInt32ToArray(2, this->receiving_major_fragment_id(), target);
+  }
+
+  // repeated int32 receiving_minor_fragment_id = 3;
+  for (int i = 0; i < this->receiving_minor_fragment_id_size(); i++) {
+    target = ::google::protobuf::internal::WireFormatLite::
+      WriteInt32ToArray(3, this->receiving_minor_fragment_id(i), target);
   }
 
-  // optional int32 sending_major_fragment_id = 2;
+  // optional int32 sending_major_fragment_id = 4;
   if (has_sending_major_fragment_id()) {
-    target = ::google::protobuf::internal::WireFormatLite::WriteInt32ToArray(2, this->sending_major_fragment_id(), target);
+    target = ::google::protobuf::internal::WireFormatLite::WriteInt32ToArray(4, this->sending_major_fragment_id(), target);
   }
 
-  // optional int32 sending_minor_fragment_id = 3;
+  // optional int32 sending_minor_fragment_id = 5;
   if (has_sending_minor_fragment_id()) {
-    target = ::google::protobuf::internal::WireFormatLite::WriteInt32ToArray(3, this->sending_minor_fragment_id(), target);
+    target = ::google::protobuf::internal::WireFormatLite::WriteInt32ToArray(5, this->sending_minor_fragment_id(), target);
   }
 
-  // optional .exec.shared.RecordBatchDef def = 4;
+  // optional .exec.shared.RecordBatchDef def = 6;
   if (has_def()) {
     target = ::google::protobuf::internal::WireFormatLite::
       WriteMessageNoVirtualToArray(
-        4, this->def(), target);
+        6, this->def(), target);
   }
 
-  // optional bool isLastBatch = 5;
+  // optional bool isLastBatch = 7;
   if (has_islastbatch()) {
-    target = ::google::protobuf::internal::WireFormatLite::WriteBoolToArray(5, this->islastbatch(), target);
+    target = ::google::protobuf::internal::WireFormatLite::WriteBoolToArray(7, this->islastbatch(), target);
   }
 
-  // optional bool isOutOfMemory = 6 [default = false];
+  // optional bool isOutOfMemory = 8 [default = false];
   if (has_isoutofmemory()) {
-    target = ::google::protobuf::internal::WireFormatLite::WriteBoolToArray(6, this->isoutofmemory(), target);
+    target = ::google::protobuf::internal::WireFormatLite::WriteBoolToArray(8, this->isoutofmemory(), target);
   }
 
   if (!unknown_fields().empty()) {
@@ -980,45 +1002,62 @@ int FragmentRecordBatch::ByteSize() const {
   int total_size = 0;
 
   if (_has_bits_[0 / 32] & (0xffu << (0 % 32))) {
-    // optional .exec.bit.FragmentHandle handle = 1;
-    if (has_handle()) {
+    // optional .exec.shared.QueryId query_id = 1;
+    if (has_query_id()) {
       total_size += 1 +
         ::google::protobuf::internal::WireFormatLite::MessageSizeNoVirtual(
-          this->handle());
+          this->query_id());
     }
 
-    // optional int32 sending_major_fragment_id = 2;
+    // optional int32 receiving_major_fragment_id = 2;
+    if (has_receiving_major_fragment_id()) {
+      total_size += 1 +
+        ::google::protobuf::internal::WireFormatLite::Int32Size(
+          this->receiving_major_fragment_id());
+    }
+
+    // optional int32 sending_major_fragment_id = 4;
     if (has_sending_major_fragment_id()) {
       total_size += 1 +
         ::google::protobuf::internal::WireFormatLite::Int32Size(
           this->sending_major_fragment_id());
     }
 
-    // optional int32 sending_minor_fragment_id = 3;
+    // optional int32 sending_minor_fragment_id = 5;
     if (has_sending_minor_fragment_id()) {
       total_size += 1 +
         ::google::protobuf::internal::WireFormatLite::Int32Size(
           this->sending_minor_fragment_id());
     }
 
-    // optional .exec.shared.RecordBatchDef def = 4;
+    // optional .exec.shared.RecordBatchDef def = 6;
     if (has_def()) {
       total_size += 1 +
         ::google::protobuf::internal::WireFormatLite::MessageSizeNoVirtual(
           this->def());
     }
 
-    // optional bool isLastBatch = 5;
+    // optional bool isLastBatch = 7;
     if (has_islastbatch()) {
       total_size += 1 + 1;
     }
 
-    // optional bool isOutOfMemory = 6 [default = false];
+    // optional bool isOutOfMemory = 8 [default = false];
     if (has_isoutofmemory()) {
       total_size += 1 + 1;
     }
 
   }
+  // repeated int32 receiving_minor_fragment_id = 3;
+  {
+    int data_size = 0;
+    for (int i = 0; i < this->receiving_minor_fragment_id_size(); i++) {
+      data_size += ::google::protobuf::internal::WireFormatLite::
+        Int32Size(this->receiving_minor_fragment_id(i));
+    }
+    total_size += 1 * this->receiving_minor_fragment_id_size() + data_size;
+  }
+
   if (!unknown_fields().empty()) {
     total_size +=
       ::google::protobuf::internal::WireFormat::ComputeUnknownFieldsSize(
@@ -1044,9 +1083,13 @@ void FragmentRecordBatch::MergeFrom(const ::google::protobuf::Message& from) {
 
 void FragmentRecordBatch::MergeFrom(const FragmentRecordBatch& from) {
   GOOGLE_CHECK_NE(&from, this);
+  receiving_minor_fragment_id_.MergeFrom(from.receiving_minor_fragment_id_);
   if (from._has_bits_[0 / 32] & (0xffu << (0 % 32))) {
-    if (from.has_handle()) {
-      mutable_handle()->::exec::bit::FragmentHandle::MergeFrom(from.handle());
+    if (from.has_query_id()) {
+      mutable_query_id()->::exec::shared::QueryId::MergeFrom(from.query_id());
+    }
+    if (from.has_receiving_major_fragment_id()) {
+      set_receiving_major_fragment_id(from.receiving_major_fragment_id());
     }
     if (from.has_sending_major_fragment_id()) {
       set_sending_major_fragment_id(from.sending_major_fragment_id());
@@ -1086,7 +1129,9 @@ bool FragmentRecordBatch::IsInitialized() const {
 
 void FragmentRecordBatch::Swap(FragmentRecordBatch* other) {
   if (other != this) {
-    std::swap(handle_, other->handle_);
+    std::swap(query_id_, other->query_id_);
+    std::swap(receiving_major_fragment_id_, other->receiving_major_fragment_id_);
+    receiving_minor_fragment_id_.Swap(&other->receiving_minor_fragment_id_);
     std::swap(sending_major_fragment_id_, other->sending_major_fragment_id_);
     std::swap(sending_minor_fragment_id_, other->sending_minor_fragment_id_);
     std::swap(def_, other->def_);
diff --git a/contrib/native/client/src/protobuf/BitData.pb.h b/contrib/native/client/src/protobuf/BitData.pb.h
index f1f935395..806d7f78f 100644
--- a/contrib/native/client/src/protobuf/BitData.pb.h
+++ b/contrib/native/client/src/protobuf/BitData.pb.h
@@ -134,32 +134,20 @@ class BitClientHandshake : public ::google::protobuf::Message {
   inline ::exec::shared::RpcChannel channel() const;
   inline void set_channel(::exec::shared::RpcChannel value);
 
-  // optional .exec.bit.FragmentHandle handle = 3;
-  inline bool has_handle() const;
-  inline void clear_handle();
-  static const int kHandleFieldNumber = 3;
-  inline const ::exec::bit::FragmentHandle& handle() const;
-  inline ::exec::bit::FragmentHandle* mutable_handle();
-  inline ::exec::bit::FragmentHandle* release_handle();
-  inline void set_allocated_handle(::exec::bit::FragmentHandle* handle);
-
   // @@protoc_insertion_point(class_scope:exec.bit.data.BitClientHandshake)
  private:
   inline void set_has_rpc_version();
   inline void clear_has_rpc_version();
   inline void set_has_channel();
   inline void clear_has_channel();
-  inline void set_has_handle();
-  inline void clear_has_handle();
 
   ::google::protobuf::UnknownFieldSet _unknown_fields_;
 
   ::google::protobuf::int32 rpc_version_;
   int channel_;
-  ::exec::bit::FragmentHandle* handle_;
 
   mutable int _cached_size_;
-  ::google::protobuf::uint32 _has_bits_[(3 + 31) / 32];
+  ::google::protobuf::uint32 _has_bits_[(2 + 31) / 32];
 
   friend void  protobuf_AddDesc_BitData_2eproto();
   friend void protobuf_AssignDesc_BitData_2eproto();
@@ -306,56 +294,77 @@ class FragmentRecordBatch : public ::google::protobuf::Message {
 
   // accessors -------------------------------------------------------
 
-  // optional .exec.bit.FragmentHandle handle = 1;
-  inline bool has_handle() const;
-  inline void clear_handle();
-  static const int kHandleFieldNumber = 1;
-  inline const ::exec::bit::FragmentHandle& handle() const;
-  inline ::exec::bit::FragmentHandle* mutable_handle();
-  inline ::exec::bit::FragmentHandle* release_handle();
-  inline void set_allocated_handle(::exec::bit::FragmentHandle* handle);
-
-  // optional int32 sending_major_fragment_id = 2;
+  // optional .exec.shared.QueryId query_id = 1;
+  inline bool has_query_id() const;
+  inline void clear_query_id();
+  static const int kQueryIdFieldNumber = 1;
+  inline const ::exec::shared::QueryId& query_id() const;
+  inline ::exec::shared::QueryId* mutable_query_id();
+  inline ::exec::shared::QueryId* release_query_id();
+  inline void set_allocated_query_id(::exec::shared::QueryId* query_id);
+
+  // optional int32 receiving_major_fragment_id = 2;
+  inline bool has_receiving_major_fragment_id() const;
+  inline void clear_receiving_major_fragment_id();
+  static const int kReceivingMajorFragmentIdFieldNumber = 2;
+  inline ::google::protobuf::int32 receiving_major_fragment_id() const;
+  inline void set_receiving_major_fragment_id(::google::protobuf::int32 value);
+
+  // repeated int32 receiving_minor_fragment_id = 3;
+  inline int receiving_minor_fragment_id_size() const;
+  inline void clear_receiving_minor_fragment_id();
+  static const int kReceivingMinorFragmentIdFieldNumber = 3;
+  inline ::google::protobuf::int32 receiving_minor_fragment_id(int index) const;
+  inline void set_receiving_minor_fragment_id(int index, ::google::protobuf::int32 value);
+  inline void add_receiving_minor_fragment_id(::google::protobuf::int32 value);
+  inline const ::google::protobuf::RepeatedField< ::google::protobuf::int32 >&
+      receiving_minor_fragment_id() const;
+  inline ::google::protobuf::RepeatedField< ::google::protobuf::int32 >*
+      mutable_receiving_minor_fragment_id();
+
+  // optional int32 sending_major_fragment_id = 4;
   inline bool has_sending_major_fragment_id() const;
   inline void clear_sending_major_fragment_id();
-  static const int kSendingMajorFragmentIdFieldNumber = 2;
+  static const int kSendingMajorFragmentIdFieldNumber = 4;
   inline ::google::protobuf::int32 sending_major_fragment_id() const;
   inline void set_sending_major_fragment_id(::google::protobuf::int32 value);
 
-  // optional int32 sending_minor_fragment_id = 3;
+  // optional int32 sending_minor_fragment_id = 5;
   inline bool has_sending_minor_fragment_id() const;
   inline void clear_sending_minor_fragment_id();
-  static const int kSendingMinorFragmentIdFieldNumber = 3;
+  static const int kSendingMinorFragmentIdFieldNumber = 5;
   inline ::google::protobuf::int32 sending_minor_fragment_id() const;
   inline void set_sending_minor_fragment_id(::google::protobuf::int32 value);
 
-  // optional .exec.shared.RecordBatchDef def = 4;
+  // optional .exec.shared.RecordBatchDef def = 6;
   inline bool has_def() const;
   inline void clear_def();
-  static const int kDefFieldNumber = 4;
+  static const int kDefFieldNumber = 6;
   inline const ::exec::shared::RecordBatchDef& def() const;
   inline ::exec::shared::RecordBatchDef* mutable_def();
   inline ::exec::shared::RecordBatchDef* release_def();
   inline void set_allocated_def(::exec::shared::RecordBatchDef* def);
 
-  // optional bool isLastBatch = 5;
+  // optional bool isLastBatch = 7;
   inline bool has_islastbatch() const;
   inline void clear_islastbatch();
-  static const int kIsLastBatchFieldNumber = 5;
+  static const int kIsLastBatchFieldNumber = 7;
   inline bool islastbatch() const;
   inline void set_islastbatch(bool value);
 
-  // optional bool isOutOfMemory = 6 [default = false];
+  // optional bool isOutOfMemory = 8 [default = false];
   inline bool has_isoutofmemory() const;
   inline void clear_isoutofmemory();
-  static const int kIsOutOfMemoryFieldNumber = 6;
+  static const int kIsOutOfMemoryFieldNumber = 8;
   inline bool isoutofmemory() const;
   inline void set_isoutofmemory(bool value);
 
   // @@protoc_insertion_point(class_scope:exec.bit.data.FragmentRecordBatch)
  private:
-  inline void set_has_handle();
-  inline void clear_has_handle();
+  inline void set_has_query_id();
+  inline void clear_has_query_id();
+  inline void set_has_receiving_major_fragment_id();
+  inline void clear_has_receiving_major_fragment_id();
   inline void set_has_sending_major_fragment_id();
   inline void clear_has_sending_major_fragment_id();
   inline void set_has_sending_minor_fragment_id();
@@ -369,15 +378,17 @@ class FragmentRecordBatch : public ::google::protobuf::Message {
 
   ::google::protobuf::UnknownFieldSet _unknown_fields_;
 
-  ::exec::bit::FragmentHandle* handle_;
+  ::exec::shared::QueryId* query_id_;
+  ::google::protobuf::RepeatedField< ::google::protobuf::int32 > receiving_minor_fragment_id_;
+  ::google::protobuf::int32 receiving_major_fragment_id_;
   ::google::protobuf::int32 sending_major_fragment_id_;
-  ::google::protobuf::int32 sending_minor_fragment_id_;
   ::exec::shared::RecordBatchDef* def_;
+  ::google::protobuf::int32 sending_minor_fragment_id_;
   bool islastbatch_;
   bool isoutofmemory_;
 
   mutable int _cached_size_;
-  ::google::protobuf::uint32 _has_bits_[(6 + 31) / 32];
+  ::google::protobuf::uint32 _has_bits_[(8 + 31) / 32];
 
   friend void  protobuf_AddDesc_BitData_2eproto();
   friend void protobuf_AssignDesc_BitData_2eproto();
@@ -438,44 +449,6 @@ inline void BitClientHandshake::set_channel(::exec::shared::RpcChannel value) {
   channel_ = value;
 }
 
-// optional .exec.bit.FragmentHandle handle = 3;
-inline bool BitClientHandshake::has_handle() const {
-  return (_has_bits_[0] & 0x00000004u) != 0;
-}
-inline void BitClientHandshake::set_has_handle() {
-  _has_bits_[0] |= 0x00000004u;
-}
-inline void BitClientHandshake::clear_has_handle() {
-  _has_bits_[0] &= ~0x00000004u;
-}
-inline void BitClientHandshake::clear_handle() {
-  if (handle_ != NULL) handle_->::exec::bit::FragmentHandle::Clear();
-  clear_has_handle();
-}
-inline const ::exec::bit::FragmentHandle& BitClientHandshake::handle() const {
-  return handle_ != NULL ? *handle_ : *default_instance_->handle_;
-}
-inline ::exec::bit::FragmentHandle* BitClientHandshake::mutable_handle() {
-  set_has_handle();
-  if (handle_ == NULL) handle_ = new ::exec::bit::FragmentHandle;
-  return handle_;
-}
-inline ::exec::bit::FragmentHandle* BitClientHandshake::release_handle() {
-  clear_has_handle();
-  ::exec::bit::FragmentHandle* temp = handle_;
-  handle_ = NULL;
-  return temp;
-}
-inline void BitClientHandshake::set_allocated_handle(::exec::bit::FragmentHandle* handle) {
-  delete handle_;
-  handle_ = handle;
-  if (handle) {
-    set_has_handle();
-  } else {
-    clear_has_handle();
-  }
-}
-
 // -------------------------------------------------------------------
 
 // BitServerHandshake
@@ -506,54 +479,101 @@ inline void BitServerHandshake::set_rpc_version(::google::protobuf::int32 value)
 
 // FragmentRecordBatch
 
-// optional .exec.bit.FragmentHandle handle = 1;
-inline bool FragmentRecordBatch::has_handle() const {
+// optional .exec.shared.QueryId query_id = 1;
+inline bool FragmentRecordBatch::has_query_id() const {
   return (_has_bits_[0] & 0x00000001u) != 0;
 }
-inline void FragmentRecordBatch::set_has_handle() {
+inline void FragmentRecordBatch::set_has_query_id() {
   _has_bits_[0] |= 0x00000001u;
 }
-inline void FragmentRecordBatch::clear_has_handle() {
+inline void FragmentRecordBatch::clear_has_query_id() {
   _has_bits_[0] &= ~0x00000001u;
 }
-inline void FragmentRecordBatch::clear_handle() {
-  if (handle_ != NULL) handle_->::exec::bit::FragmentHandle::Clear();
-  clear_has_handle();
+inline void FragmentRecordBatch::clear_query_id() {
+  if (query_id_ != NULL) query_id_->::exec::shared::QueryId::Clear();
+  clear_has_query_id();
 }
-inline const ::exec::bit::FragmentHandle& FragmentRecordBatch::handle() const {
-  return handle_ != NULL ? *handle_ : *default_instance_->handle_;
+inline const ::exec::shared::QueryId& FragmentRecordBatch::query_id() const {
+  return query_id_ != NULL ? *query_id_ : *default_instance_->query_id_;
 }
-inline ::exec::bit::FragmentHandle* FragmentRecordBatch::mutable_handle() {
-  set_has_handle();
-  if (handle_ == NULL) handle_ = new ::exec::bit::FragmentHandle;
-  return handle_;
+inline ::exec::shared::QueryId* FragmentRecordBatch::mutable_query_id() {
+  set_has_query_id();
+  if (query_id_ == NULL) query_id_ = new ::exec::shared::QueryId;
+  return query_id_;
 }
-inline ::exec::bit::FragmentHandle* FragmentRecordBatch::release_handle() {
-  clear_has_handle();
-  ::exec::bit::FragmentHandle* temp = handle_;
-  handle_ = NULL;
+inline ::exec::shared::QueryId* FragmentRecordBatch::release_query_id() {
+  clear_has_query_id();
+  ::exec::shared::QueryId* temp = query_id_;
+  query_id_ = NULL;
   return temp;
 }
-inline void FragmentRecordBatch::set_allocated_handle(::exec::bit::FragmentHandle* handle) {
-  delete handle_;
-  handle_ = handle;
-  if (handle) {
-    set_has_handle();
+inline void FragmentRecordBatch::set_allocated_query_id(::exec::shared::QueryId* query_id) {
+  delete query_id_;
+  query_id_ = query_id;
+  if (query_id) {
+    set_has_query_id();
   } else {
-    clear_has_handle();
+    clear_has_query_id();
   }
 }
 
-// optional int32 sending_major_fragment_id = 2;
-inline bool FragmentRecordBatch::has_sending_major_fragment_id() const {
+// optional int32 receiving_major_fragment_id = 2;
+inline bool FragmentRecordBatch::has_receiving_major_fragment_id() const {
   return (_has_bits_[0] & 0x00000002u) != 0;
 }
-inline void FragmentRecordBatch::set_has_sending_major_fragment_id() {
+inline void FragmentRecordBatch::set_has_receiving_major_fragment_id() {
   _has_bits_[0] |= 0x00000002u;
 }
-inline void FragmentRecordBatch::clear_has_sending_major_fragment_id() {
+inline void FragmentRecordBatch::clear_has_receiving_major_fragment_id() {
   _has_bits_[0] &= ~0x00000002u;
 }
+inline void FragmentRecordBatch::clear_receiving_major_fragment_id() {
+  receiving_major_fragment_id_ = 0;
+  clear_has_receiving_major_fragment_id();
+}
+inline ::google::protobuf::int32 FragmentRecordBatch::receiving_major_fragment_id() const {
+  return receiving_major_fragment_id_;
+}
+inline void FragmentRecordBatch::set_receiving_major_fragment_id(::google::protobuf::int32 value) {
+  set_has_receiving_major_fragment_id();
+  receiving_major_fragment_id_ = value;
+}
+
+// repeated int32 receiving_minor_fragment_id = 3;
+inline int FragmentRecordBatch::receiving_minor_fragment_id_size() const {
+  return receiving_minor_fragment_id_.size();
+}
+inline void FragmentRecordBatch::clear_receiving_minor_fragment_id() {
+  receiving_minor_fragment_id_.Clear();
+}
+inline ::google::protobuf::int32 FragmentRecordBatch::receiving_minor_fragment_id(int index) const {
+  return receiving_minor_fragment_id_.Get(index);
+}
+inline void FragmentRecordBatch::set_receiving_minor_fragment_id(int index, ::google::protobuf::int32 value) {
+  receiving_minor_fragment_id_.Set(index, value);
+}
+inline void FragmentRecordBatch::add_receiving_minor_fragment_id(::google::protobuf::int32 value) {
+  receiving_minor_fragment_id_.Add(value);
+}
+inline const ::google::protobuf::RepeatedField< ::google::protobuf::int32 >&
+FragmentRecordBatch::receiving_minor_fragment_id() const {
+  return receiving_minor_fragment_id_;
+}
+inline ::google::protobuf::RepeatedField< ::google::protobuf::int32 >*
+FragmentRecordBatch::mutable_receiving_minor_fragment_id() {
+  return &receiving_minor_fragment_id_;
+}
+
+// optional int32 sending_major_fragment_id = 4;
+inline bool FragmentRecordBatch::has_sending_major_fragment_id() const {
+  return (_has_bits_[0] & 0x00000008u) != 0;
+}
+inline void FragmentRecordBatch::set_has_sending_major_fragment_id() {
+  _has_bits_[0] |= 0x00000008u;
+}
+inline void FragmentRecordBatch::clear_has_sending_major_fragment_id() {
+  _has_bits_[0] &= ~0x00000008u;
+}
 inline void FragmentRecordBatch::clear_sending_major_fragment_id() {
   sending_major_fragment_id_ = 0;
   clear_has_sending_major_fragment_id();
@@ -566,15 +586,15 @@ inline void FragmentRecordBatch::set_sending_major_fragment_id(::google::protobu
   sending_major_fragment_id_ = value;
 }
 
-// optional int32 sending_minor_fragment_id = 3;
+// optional int32 sending_minor_fragment_id = 5;
 inline bool FragmentRecordBatch::has_sending_minor_fragment_id() const {
-  return (_has_bits_[0] & 0x00000004u) != 0;
+  return (_has_bits_[0] & 0x00000010u) != 0;
 }
 inline void FragmentRecordBatch::set_has_sending_minor_fragment_id() {
-  _has_bits_[0] |= 0x00000004u;
+  _has_bits_[0] |= 0x00000010u;
 }
 inline void FragmentRecordBatch::clear_has_sending_minor_fragment_id() {
-  _has_bits_[0] &= ~0x00000004u;
+  _has_bits_[0] &= ~0x00000010u;
 }
 inline void FragmentRecordBatch::clear_sending_minor_fragment_id() {
   sending_minor_fragment_id_ = 0;
@@ -588,15 +608,15 @@ inline void FragmentRecordBatch::set_sending_minor_fragment_id(::google::protobu
   sending_minor_fragment_id_ = value;
 }
 
-// optional .exec.shared.RecordBatchDef def = 4;
+// optional .exec.shared.RecordBatchDef def = 6;
 inline bool FragmentRecordBatch::has_def() const {
-  return (_has_bits_[0] & 0x00000008u) != 0;
+  return (_has_bits_[0] & 0x00000020u) != 0;
 }
 inline void FragmentRecordBatch::set_has_def() {
-  _has_bits_[0] |= 0x00000008u;
+  _has_bits_[0] |= 0x00000020u;
 }
 inline void FragmentRecordBatch::clear_has_def() {
-  _has_bits_[0] &= ~0x00000008u;
+  _has_bits_[0] &= ~0x00000020u;
 }
 inline void FragmentRecordBatch::clear_def() {
   if (def_ != NULL) def_->::exec::shared::RecordBatchDef::Clear();
@@ -626,15 +646,15 @@ inline void FragmentRecordBatch::set_allocated_def(::exec::shared::RecordBatchDe
   }
 }
 
-// optional bool isLastBatch = 5;
+// optional bool isLastBatch = 7;
 inline bool FragmentRecordBatch::has_islastbatch() const {
-  return (_has_bits_[0] & 0x00000010u) != 0;
+  return (_has_bits_[0] & 0x00000040u) != 0;
 }
 inline void FragmentRecordBatch::set_has_islastbatch() {
-  _has_bits_[0] |= 0x00000010u;
+  _has_bits_[0] |= 0x00000040u;
 }
 inline void FragmentRecordBatch::clear_has_islastbatch() {
-  _has_bits_[0] &= ~0x00000010u;
+  _has_bits_[0] &= ~0x00000040u;
 }
 inline void FragmentRecordBatch::clear_islastbatch() {
   islastbatch_ = false;
@@ -648,15 +668,15 @@ inline void FragmentRecordBatch::set_islastbatch(bool value) {
   islastbatch_ = value;
 }
 
-// optional bool isOutOfMemory = 6 [default = false];
+// optional bool isOutOfMemory = 8 [default = false];
 inline bool FragmentRecordBatch::has_isoutofmemory() const {
-  return (_has_bits_[0] & 0x00000020u) != 0;
+  return (_has_bits_[0] & 0x00000080u) != 0;
 }
 inline void FragmentRecordBatch::set_has_isoutofmemory() {
-  _has_bits_[0] |= 0x00000020u;
+  _has_bits_[0] |= 0x00000080u;
 }
 inline void FragmentRecordBatch::clear_has_isoutofmemory() {
-  _has_bits_[0] &= ~0x00000020u;
+  _has_bits_[0] &= ~0x00000080u;
 }
 inline void FragmentRecordBatch::clear_isoutofmemory() {
   isoutofmemory_ = false;
diff --git a/contrib/native/client/src/protobuf/User.pb.cc b/contrib/native/client/src/protobuf/User.pb.cc
index d85c81bd6..360becb4c 100644
--- a/contrib/native/client/src/protobuf/User.pb.cc
+++ b/contrib/native/client/src/protobuf/User.pb.cc
@@ -220,14 +220,14 @@ void protobuf_AddDesc_User_2eproto() {
     "lts_mode\030\001 \001(\0162\033.exec.user.QueryResultsM"
     "ode\022$\n\004type\030\002 \001(\0162\026.exec.shared.QueryTyp"
     "e\022\014\n\004plan\030\003 \001(\t\")\n\022BitToUserHandshake\022\023\n"
-    "\013rpc_version\030\002 \001(\005*\270\001\n\007RpcType\022\r\n\tHANDSH"
+    "\013rpc_version\030\002 \001(\005*\310\001\n\007RpcType\022\r\n\tHANDSH"
     "AKE\020\000\022\007\n\003ACK\020\001\022\013\n\007GOODBYE\020\002\022\r\n\tRUN_QUERY"
     "\020\003\022\020\n\014CANCEL_QUERY\020\004\022\023\n\017REQUEST_RESULTS\020"
-    "\005\022\020\n\014QUERY_RESULT\020\006\022\020\n\014QUERY_HANDLE\020\007\022\026\n"
-    "\022REQ_META_FUNCTIONS\020\010\022\026\n\022RESP_FUNCTION_L"
-    "IST\020\t*#\n\020QueryResultsMode\022\017\n\013STREAM_FULL"
-    "\020\001B+\n\033org.apache.drill.exec.protoB\nUserP"
-    "rotosH\001", 927);
+    "\005\022\016\n\nQUERY_DATA\020\006\022\020\n\014QUERY_HANDLE\020\007\022\026\n\022R"
+    "EQ_META_FUNCTIONS\020\010\022\026\n\022RESP_FUNCTION_LIS"
+    "T\020\t\022\020\n\014QUERY_RESULT\020\n*#\n\020QueryResultsMod"
+    "e\022\017\n\013STREAM_FULL\020\001B+\n\033org.apache.drill.e"
+    "xec.protoB\nUserProtosH\001", 943);
   ::google::protobuf::MessageFactory::InternalRegisterGeneratedFile(
     "User.proto", &protobuf_RegisterTypes);
   Property::default_instance_ = new Property();
@@ -267,6 +267,7 @@ bool RpcType_IsValid(int value) {
     case 7:
     case 8:
     case 9:
+    case 10:
       return true;
     default:
       return false;
diff --git a/contrib/native/client/src/protobuf/User.pb.h b/contrib/native/client/src/protobuf/User.pb.h
index eca199dc0..69daf50d2 100644
--- a/contrib/native/client/src/protobuf/User.pb.h
+++ b/contrib/native/client/src/protobuf/User.pb.h
@@ -51,14 +51,15 @@ enum RpcType {
   RUN_QUERY = 3,
   CANCEL_QUERY = 4,
   REQUEST_RESULTS = 5,
-  QUERY_RESULT = 6,
+  QUERY_DATA = 6,
   QUERY_HANDLE = 7,
   REQ_META_FUNCTIONS = 8,
-  RESP_FUNCTION_LIST = 9
+  RESP_FUNCTION_LIST = 9,
+  QUERY_RESULT = 10
 };
 bool RpcType_IsValid(int value);
 const RpcType RpcType_MIN = HANDSHAKE;
-const RpcType RpcType_MAX = RESP_FUNCTION_LIST;
+const RpcType RpcType_MAX = QUERY_RESULT;
 const int RpcType_ARRAYSIZE = RpcType_MAX + 1;
 
 const ::google::protobuf::EnumDescriptor* RpcType_descriptor();
diff --git a/contrib/native/client/src/protobuf/UserBitShared.pb.cc b/contrib/native/client/src/protobuf/UserBitShared.pb.cc
index 7c237f636..b07ecda41 100644
--- a/contrib/native/client/src/protobuf/UserBitShared.pb.cc
+++ b/contrib/native/client/src/protobuf/UserBitShared.pb.cc
@@ -56,6 +56,9 @@ const ::google::protobuf::Descriptor* QueryResult_descriptor_ = NULL;
 const ::google::protobuf::internal::GeneratedMessageReflection*
   QueryResult_reflection_ = NULL;
 const ::google::protobuf::EnumDescriptor* QueryResult_QueryState_descriptor_ = NULL;
+const ::google::protobuf::Descriptor* QueryData_descriptor_ = NULL;
+const ::google::protobuf::internal::GeneratedMessageReflection*
+  QueryData_reflection_ = NULL;
 const ::google::protobuf::Descriptor* QueryInfo_descriptor_ = NULL;
 const ::google::protobuf::internal::GeneratedMessageReflection*
   QueryInfo_reflection_ = NULL;
@@ -270,18 +273,10 @@ void protobuf_AssignDesc_UserBitShared_2eproto() {
       ::google::protobuf::MessageFactory::generated_factory(),
       sizeof(NodeStatus));
   QueryResult_descriptor_ = file->message_type(10);
-  static const int QueryResult_offsets_[11] = {
+  static const int QueryResult_offsets_[3] = {
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(QueryResult, query_state_),
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(QueryResult, query_id_),
-    GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(QueryResult, is_last_chunk_),
-    GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(QueryResult, row_count_),
-    GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(QueryResult, records_scan_),
-    GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(QueryResult, records_error_),
-    GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(QueryResult, submission_time_),
-    GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(QueryResult, node_status_),
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(QueryResult, error_),
-    GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(QueryResult, def_),
-    GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(QueryResult, schema_changed_),
   };
   QueryResult_reflection_ =
     new ::google::protobuf::internal::GeneratedMessageReflection(
@@ -295,7 +290,24 @@ void protobuf_AssignDesc_UserBitShared_2eproto() {
       ::google::protobuf::MessageFactory::generated_factory(),
       sizeof(QueryResult));
   QueryResult_QueryState_descriptor_ = QueryResult_descriptor_->enum_type(0);
-  QueryInfo_descriptor_ = file->message_type(11);
+  QueryData_descriptor_ = file->message_type(11);
+  static const int QueryData_offsets_[3] = {
+    GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(QueryData, query_id_),
+    GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(QueryData, row_count_),
+    GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(QueryData, def_),
+  };
+  QueryData_reflection_ =
+    new ::google::protobuf::internal::GeneratedMessageReflection(
+      QueryData_descriptor_,
+      QueryData::default_instance_,
+      QueryData_offsets_,
+      GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(QueryData, _has_bits_[0]),
+      GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(QueryData, _unknown_fields_),
+      -1,
+      ::google::protobuf::DescriptorPool::generated_pool(),
+      ::google::protobuf::MessageFactory::generated_factory(),
+      sizeof(QueryData));
+  QueryInfo_descriptor_ = file->message_type(12);
   static const int QueryInfo_offsets_[5] = {
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(QueryInfo, query_),
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(QueryInfo, start_),
@@ -314,7 +326,7 @@ void protobuf_AssignDesc_UserBitShared_2eproto() {
       ::google::protobuf::DescriptorPool::generated_pool(),
       ::google::protobuf::MessageFactory::generated_factory(),
       sizeof(QueryInfo));
-  QueryProfile_descriptor_ = file->message_type(12);
+  QueryProfile_descriptor_ = file->message_type(13);
   static const int QueryProfile_offsets_[11] = {
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(QueryProfile, id_),
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(QueryProfile, type_),
@@ -339,7 +351,7 @@ void protobuf_AssignDesc_UserBitShared_2eproto() {
       ::google::protobuf::DescriptorPool::generated_pool(),
       ::google::protobuf::MessageFactory::generated_factory(),
       sizeof(QueryProfile));
-  MajorFragmentProfile_descriptor_ = file->message_type(13);
+  MajorFragmentProfile_descriptor_ = file->message_type(14);
   static const int MajorFragmentProfile_offsets_[2] = {
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(MajorFragmentProfile, major_fragment_id_),
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(MajorFragmentProfile, minor_fragment_profile_),
@@ -355,7 +367,7 @@ void protobuf_AssignDesc_UserBitShared_2eproto() {
       ::google::protobuf::DescriptorPool::generated_pool(),
       ::google::protobuf::MessageFactory::generated_factory(),
       sizeof(MajorFragmentProfile));
-  MinorFragmentProfile_descriptor_ = file->message_type(14);
+  MinorFragmentProfile_descriptor_ = file->message_type(15);
   static const int MinorFragmentProfile_offsets_[9] = {
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(MinorFragmentProfile, state_),
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(MinorFragmentProfile, error_),
@@ -378,14 +390,14 @@ void protobuf_AssignDesc_UserBitShared_2eproto() {
       ::google::protobuf::DescriptorPool::generated_pool(),
       ::google::protobuf::MessageFactory::generated_factory(),
       sizeof(MinorFragmentProfile));
-  OperatorProfile_descriptor_ = file->message_type(15);
+  OperatorProfile_descriptor_ = file->message_type(16);
   static const int OperatorProfile_offsets_[8] = {
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(OperatorProfile, input_profile_),
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(OperatorProfile, operator_id_),
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(OperatorProfile, operator_type_),
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(OperatorProfile, setup_nanos_),
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(OperatorProfile, process_nanos_),
-    GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(OperatorProfile, local_memory_allocated_),
+    GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(OperatorProfile, peak_local_memory_allocated_),
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(OperatorProfile, metric_),
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(OperatorProfile, wait_nanos_),
   };
@@ -400,7 +412,7 @@ void protobuf_AssignDesc_UserBitShared_2eproto() {
       ::google::protobuf::DescriptorPool::generated_pool(),
       ::google::protobuf::MessageFactory::generated_factory(),
       sizeof(OperatorProfile));
-  StreamProfile_descriptor_ = file->message_type(16);
+  StreamProfile_descriptor_ = file->message_type(17);
   static const int StreamProfile_offsets_[3] = {
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(StreamProfile, records_),
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(StreamProfile, batches_),
@@ -417,7 +429,7 @@ void protobuf_AssignDesc_UserBitShared_2eproto() {
       ::google::protobuf::DescriptorPool::generated_pool(),
       ::google::protobuf::MessageFactory::generated_factory(),
       sizeof(StreamProfile));
-  MetricValue_descriptor_ = file->message_type(17);
+  MetricValue_descriptor_ = file->message_type(18);
   static const int MetricValue_offsets_[3] = {
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(MetricValue, metric_id_),
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(MetricValue, long_value_),
@@ -473,6 +485,8 @@ void protobuf_RegisterTypes(const ::std::string&) {
   ::google::protobuf::MessageFactory::InternalRegisterGeneratedMessage(
     QueryResult_descriptor_, &QueryResult::default_instance());
   ::google::protobuf::MessageFactory::InternalRegisterGeneratedMessage(
+    QueryData_descriptor_, &QueryData::default_instance());
+  ::google::protobuf::MessageFactory::InternalRegisterGeneratedMessage(
     QueryInfo_descriptor_, &QueryInfo::default_instance());
   ::google::protobuf::MessageFactory::InternalRegisterGeneratedMessage(
     QueryProfile_descriptor_, &QueryProfile::default_instance());
@@ -513,6 +527,8 @@ void protobuf_ShutdownFile_UserBitShared_2eproto() {
   delete NodeStatus_reflection_;
   delete QueryResult::default_instance_;
   delete QueryResult_reflection_;
+  delete QueryData::default_instance_;
+  delete QueryData_reflection_;
   delete QueryInfo::default_instance_;
   delete QueryInfo_reflection_;
   delete QueryProfile::default_instance_;
@@ -572,76 +588,74 @@ void protobuf_AddDesc_UserBitShared_2eproto() {
     "t\030\004 \001(\005\022\027\n\017var_byte_length\030\005 \001(\005\022\023\n\013grou"
     "p_count\030\006 \001(\005\022\025\n\rbuffer_length\030\007 \001(\005\"7\n\n"
     "NodeStatus\022\017\n\007node_id\030\001 \001(\005\022\030\n\020memory_fo"
-    "otprint\030\002 \001(\003\"\335\003\n\013QueryResult\0228\n\013query_s"
+    "otprint\030\002 \001(\003\"\206\002\n\013QueryResult\0228\n\013query_s"
     "tate\030\001 \001(\0162#.exec.shared.QueryResult.Que"
     "ryState\022&\n\010query_id\030\002 \001(\0132\024.exec.shared."
-    "QueryId\022\025\n\ris_last_chunk\030\003 \001(\010\022\021\n\trow_co"
-    "unt\030\004 \001(\005\022\024\n\014records_scan\030\005 \001(\003\022\025\n\rrecor"
-    "ds_error\030\006 \001(\003\022\027\n\017submission_time\030\007 \001(\003\022"
-    ",\n\013node_status\030\010 \003(\0132\027.exec.shared.NodeS"
-    "tatus\022(\n\005error\030\t \003(\0132\031.exec.shared.Drill"
-    "PBError\022(\n\003def\030\n \001(\0132\033.exec.shared.Recor"
-    "dBatchDef\022\026\n\016schema_changed\030\013 \001(\010\"b\n\nQue"
-    "ryState\022\013\n\007PENDING\020\000\022\013\n\007RUNNING\020\001\022\r\n\tCOM"
-    "PLETED\020\002\022\014\n\010CANCELED\020\003\022\n\n\006FAILED\020\004\022\021\n\rUN"
-    "KNOWN_QUERY\020\005\"\224\001\n\tQueryInfo\022\r\n\005query\030\001 \001"
-    "(\t\022\r\n\005start\030\002 \001(\003\0222\n\005state\030\003 \001(\0162#.exec."
-    "shared.QueryResult.QueryState\022\014\n\004user\030\004 "
-    "\001(\t\022\'\n\007foreman\030\005 \001(\0132\026.exec.DrillbitEndp"
-    "oint\"\336\002\n\014QueryProfile\022 \n\002id\030\001 \001(\0132\024.exec"
-    ".shared.QueryId\022$\n\004type\030\002 \001(\0162\026.exec.sha"
-    "red.QueryType\022\r\n\005start\030\003 \001(\003\022\013\n\003end\030\004 \001("
-    "\003\022\r\n\005query\030\005 \001(\t\022\014\n\004plan\030\006 \001(\t\022\'\n\007forema"
-    "n\030\007 \001(\0132\026.exec.DrillbitEndpoint\0222\n\005state"
-    "\030\010 \001(\0162#.exec.shared.QueryResult.QuerySt"
-    "ate\022\027\n\017total_fragments\030\t \001(\005\022\032\n\022finished"
-    "_fragments\030\n \001(\005\022;\n\020fragment_profile\030\013 \003"
-    "(\0132!.exec.shared.MajorFragmentProfile\"t\n"
-    "\024MajorFragmentProfile\022\031\n\021major_fragment_"
-    "id\030\001 \001(\005\022A\n\026minor_fragment_profile\030\002 \003(\013"
-    "2!.exec.shared.MinorFragmentProfile\"\274\002\n\024"
-    "MinorFragmentProfile\022)\n\005state\030\001 \001(\0162\032.ex"
-    "ec.shared.FragmentState\022(\n\005error\030\002 \001(\0132\031"
-    ".exec.shared.DrillPBError\022\031\n\021minor_fragm"
-    "ent_id\030\003 \001(\005\0226\n\020operator_profile\030\004 \003(\0132\034"
-    ".exec.shared.OperatorProfile\022\022\n\nstart_ti"
-    "me\030\005 \001(\003\022\020\n\010end_time\030\006 \001(\003\022\023\n\013memory_use"
-    "d\030\007 \001(\003\022\027\n\017max_memory_used\030\010 \001(\003\022(\n\010endp"
-    "oint\030\t \001(\0132\026.exec.DrillbitEndpoint\"\372\001\n\017O"
-    "peratorProfile\0221\n\rinput_profile\030\001 \003(\0132\032."
-    "exec.shared.StreamProfile\022\023\n\013operator_id"
-    "\030\003 \001(\005\022\025\n\roperator_type\030\004 \001(\005\022\023\n\013setup_n"
-    "anos\030\005 \001(\003\022\025\n\rprocess_nanos\030\006 \001(\003\022\036\n\026loc"
-    "al_memory_allocated\030\007 \001(\003\022(\n\006metric\030\010 \003("
-    "\0132\030.exec.shared.MetricValue\022\022\n\nwait_nano"
-    "s\030\t \001(\003\"B\n\rStreamProfile\022\017\n\007records\030\001 \001("
-    "\003\022\017\n\007batches\030\002 \001(\003\022\017\n\007schemas\030\003 \001(\003\"J\n\013M"
-    "etricValue\022\021\n\tmetric_id\030\001 \001(\005\022\022\n\nlong_va"
-    "lue\030\002 \001(\003\022\024\n\014double_value\030\003 \001(\001*5\n\nRpcCh"
-    "annel\022\017\n\013BIT_CONTROL\020\000\022\014\n\010BIT_DATA\020\001\022\010\n\004"
-    "USER\020\002*/\n\tQueryType\022\007\n\003SQL\020\001\022\013\n\007LOGICAL\020"
-    "\002\022\014\n\010PHYSICAL\020\003*k\n\rFragmentState\022\013\n\007SEND"
-    "ING\020\000\022\027\n\023AWAITING_ALLOCATION\020\001\022\013\n\007RUNNIN"
-    "G\020\002\022\014\n\010FINISHED\020\003\022\r\n\tCANCELLED\020\004\022\n\n\006FAIL"
-    "ED\020\005*\264\005\n\020CoreOperatorType\022\021\n\rSINGLE_SEND"
-    "ER\020\000\022\024\n\020BROADCAST_SENDER\020\001\022\n\n\006FILTER\020\002\022\022"
-    "\n\016HASH_AGGREGATE\020\003\022\r\n\tHASH_JOIN\020\004\022\016\n\nMER"
-    "GE_JOIN\020\005\022\031\n\025HASH_PARTITION_SENDER\020\006\022\t\n\005"
-    "LIMIT\020\007\022\024\n\020MERGING_RECEIVER\020\010\022\034\n\030ORDERED"
-    "_PARTITION_SENDER\020\t\022\013\n\007PROJECT\020\n\022\026\n\022UNOR"
-    "DERED_RECEIVER\020\013\022\020\n\014RANGE_SENDER\020\014\022\n\n\006SC"
-    "REEN\020\r\022\034\n\030SELECTION_VECTOR_REMOVER\020\016\022\027\n\023"
-    "STREAMING_AGGREGATE\020\017\022\016\n\nTOP_N_SORT\020\020\022\021\n"
-    "\rEXTERNAL_SORT\020\021\022\t\n\005TRACE\020\022\022\t\n\005UNION\020\023\022\014"
-    "\n\010OLD_SORT\020\024\022\032\n\026PARQUET_ROW_GROUP_SCAN\020\025"
-    "\022\021\n\rHIVE_SUB_SCAN\020\026\022\025\n\021SYSTEM_TABLE_SCAN"
-    "\020\027\022\021\n\rMOCK_SUB_SCAN\020\030\022\022\n\016PARQUET_WRITER\020"
-    "\031\022\023\n\017DIRECT_SUB_SCAN\020\032\022\017\n\013TEXT_WRITER\020\033\022"
-    "\021\n\rTEXT_SUB_SCAN\020\034\022\021\n\rJSON_SUB_SCAN\020\035\022\030\n"
-    "\024INFO_SCHEMA_SUB_SCAN\020\036\022\023\n\017COMPLEX_TO_JS"
-    "ON\020\037\022\025\n\021PRODUCER_CONSUMER\020 \022\022\n\016HBASE_SUB"
-    "_SCAN\020!\022\n\n\006WINDOW\020\"B.\n\033org.apache.drill."
-    "exec.protoB\rUserBitSharedH\001", 4107);
+    "QueryId\022(\n\005error\030\003 \003(\0132\031.exec.shared.Dri"
+    "llPBError\"k\n\nQueryState\022\013\n\007PENDING\020\000\022\013\n\007"
+    "RUNNING\020\001\022\r\n\tCOMPLETED\020\002\022\014\n\010CANCELED\020\003\022\n"
+    "\n\006FAILED\020\004\022\032\n\026CANCELLATION_REQUESTED\020\005\"p"
+    "\n\tQueryData\022&\n\010query_id\030\001 \001(\0132\024.exec.sha"
+    "red.QueryId\022\021\n\trow_count\030\002 \001(\005\022(\n\003def\030\003 "
+    "\001(\0132\033.exec.shared.RecordBatchDef\"\224\001\n\tQue"
+    "ryInfo\022\r\n\005query\030\001 \001(\t\022\r\n\005start\030\002 \001(\003\0222\n\005"
+    "state\030\003 \001(\0162#.exec.shared.QueryResult.Qu"
+    "eryState\022\014\n\004user\030\004 \001(\t\022\'\n\007foreman\030\005 \001(\0132"
+    "\026.exec.DrillbitEndpoint\"\336\002\n\014QueryProfile"
+    "\022 \n\002id\030\001 \001(\0132\024.exec.shared.QueryId\022$\n\004ty"
+    "pe\030\002 \001(\0162\026.exec.shared.QueryType\022\r\n\005star"
+    "t\030\003 \001(\003\022\013\n\003end\030\004 \001(\003\022\r\n\005query\030\005 \001(\t\022\014\n\004p"
+    "lan\030\006 \001(\t\022\'\n\007foreman\030\007 \001(\0132\026.exec.Drillb"
+    "itEndpoint\0222\n\005state\030\010 \001(\0162#.exec.shared."
+    "QueryResult.QueryState\022\027\n\017total_fragment"
+    "s\030\t \001(\005\022\032\n\022finished_fragments\030\n \001(\005\022;\n\020f"
+    "ragment_profile\030\013 \003(\0132!.exec.shared.Majo"
+    "rFragmentProfile\"t\n\024MajorFragmentProfile"
+    "\022\031\n\021major_fragment_id\030\001 \001(\005\022A\n\026minor_fra"
+    "gment_profile\030\002 \003(\0132!.exec.shared.MinorF"
+    "ragmentProfile\"\274\002\n\024MinorFragmentProfile\022"
+    ")\n\005state\030\001 \001(\0162\032.exec.shared.FragmentSta"
+    "te\022(\n\005error\030\002 \001(\0132\031.exec.shared.DrillPBE"
+    "rror\022\031\n\021minor_fragment_id\030\003 \001(\005\0226\n\020opera"
+    "tor_profile\030\004 \003(\0132\034.exec.shared.Operator"
+    "Profile\022\022\n\nstart_time\030\005 \001(\003\022\020\n\010end_time\030"
+    "\006 \001(\003\022\023\n\013memory_used\030\007 \001(\003\022\027\n\017max_memory"
+    "_used\030\010 \001(\003\022(\n\010endpoint\030\t \001(\0132\026.exec.Dri"
+    "llbitEndpoint\"\377\001\n\017OperatorProfile\0221\n\rinp"
+    "ut_profile\030\001 \003(\0132\032.exec.shared.StreamPro"
+    "file\022\023\n\013operator_id\030\003 \001(\005\022\025\n\roperator_ty"
+    "pe\030\004 \001(\005\022\023\n\013setup_nanos\030\005 \001(\003\022\025\n\rprocess"
+    "_nanos\030\006 \001(\003\022#\n\033peak_local_memory_alloca"
+    "ted\030\007 \001(\003\022(\n\006metric\030\010 \003(\0132\030.exec.shared."
+    "MetricValue\022\022\n\nwait_nanos\030\t \001(\003\"B\n\rStrea"
+    "mProfile\022\017\n\007records\030\001 \001(\003\022\017\n\007batches\030\002 \001"
+    "(\003\022\017\n\007schemas\030\003 \001(\003\"J\n\013MetricValue\022\021\n\tme"
+    "tric_id\030\001 \001(\005\022\022\n\nlong_value\030\002 \001(\003\022\024\n\014dou"
+    "ble_value\030\003 \001(\001*5\n\nRpcChannel\022\017\n\013BIT_CON"
+    "TROL\020\000\022\014\n\010BIT_DATA\020\001\022\010\n\004USER\020\002*/\n\tQueryT"
+    "ype\022\007\n\003SQL\020\001\022\013\n\007LOGICAL\020\002\022\014\n\010PHYSICAL\020\003*"
+    "k\n\rFragmentState\022\013\n\007SENDING\020\000\022\027\n\023AWAITIN"
+    "G_ALLOCATION\020\001\022\013\n\007RUNNING\020\002\022\014\n\010FINISHED\020"
+    "\003\022\r\n\tCANCELLED\020\004\022\n\n\006FAILED\020\005*\264\005\n\020CoreOpe"
+    "ratorType\022\021\n\rSINGLE_SENDER\020\000\022\024\n\020BROADCAS"
+    "T_SENDER\020\001\022\n\n\006FILTER\020\002\022\022\n\016HASH_AGGREGATE"
+    "\020\003\022\r\n\tHASH_JOIN\020\004\022\016\n\nMERGE_JOIN\020\005\022\031\n\025HAS"
+    "H_PARTITION_SENDER\020\006\022\t\n\005LIMIT\020\007\022\024\n\020MERGI"
+    "NG_RECEIVER\020\010\022\034\n\030ORDERED_PARTITION_SENDE"
+    "R\020\t\022\013\n\007PROJECT\020\n\022\026\n\022UNORDERED_RECEIVER\020\013"
+    "\022\020\n\014RANGE_SENDER\020\014\022\n\n\006SCREEN\020\r\022\034\n\030SELECT"
+    "ION_VECTOR_REMOVER\020\016\022\027\n\023STREAMING_AGGREG"
+    "ATE\020\017\022\016\n\nTOP_N_SORT\020\020\022\021\n\rEXTERNAL_SORT\020\021"
+    "\022\t\n\005TRACE\020\022\022\t\n\005UNION\020\023\022\014\n\010OLD_SORT\020\024\022\032\n\026"
+    "PARQUET_ROW_GROUP_SCAN\020\025\022\021\n\rHIVE_SUB_SCA"
+    "N\020\026\022\025\n\021SYSTEM_TABLE_SCAN\020\027\022\021\n\rMOCK_SUB_S"
+    "CAN\020\030\022\022\n\016PARQUET_WRITER\020\031\022\023\n\017DIRECT_SUB_"
+    "SCAN\020\032\022\017\n\013TEXT_WRITER\020\033\022\021\n\rTEXT_SUB_SCAN"
+    "\020\034\022\021\n\rJSON_SUB_SCAN\020\035\022\030\n\024INFO_SCHEMA_SUB"
+    "_SCAN\020\036\022\023\n\017COMPLEX_TO_JSON\020\037\022\025\n\021PRODUCER"
+    "_CONSUMER\020 \022\022\n\016HBASE_SUB_SCAN\020!\022\n\n\006WINDO"
+    "W\020\"B.\n\033org.apache.drill.exec.protoB\rUser"
+    "BitSharedH\001", 4011);
   ::google::protobuf::MessageFactory::InternalRegisterGeneratedFile(
     "UserBitShared.proto", &protobuf_RegisterTypes);
   UserCredentials::default_instance_ = new UserCredentials();
@@ -655,6 +669,7 @@ void protobuf_AddDesc_UserBitShared_2eproto() {
   SerializedField::default_instance_ = new SerializedField();
   NodeStatus::default_instance_ = new NodeStatus();
   QueryResult::default_instance_ = new QueryResult();
+  QueryData::default_instance_ = new QueryData();
   QueryInfo::default_instance_ = new QueryInfo();
   QueryProfile::default_instance_ = new QueryProfile();
   MajorFragmentProfile::default_instance_ = new MajorFragmentProfile();
@@ -673,6 +688,7 @@ void protobuf_AddDesc_UserBitShared_2eproto() {
   SerializedField::default_instance_->InitAsDefaultInstance();
   NodeStatus::default_instance_->InitAsDefaultInstance();
   QueryResult::default_instance_->InitAsDefaultInstance();
+  QueryData::default_instance_->InitAsDefaultInstance();
   QueryInfo::default_instance_->InitAsDefaultInstance();
   QueryProfile::default_instance_->InitAsDefaultInstance();
   MajorFragmentProfile::default_instance_->InitAsDefaultInstance();
@@ -4173,7 +4189,7 @@ const QueryResult_QueryState QueryResult::RUNNING;
 const QueryResult_QueryState QueryResult::COMPLETED;
 const QueryResult_QueryState QueryResult::CANCELED;
 const QueryResult_QueryState QueryResult::FAILED;
-const QueryResult_QueryState QueryResult::UNKNOWN_QUERY;
+const QueryResult_QueryState QueryResult::CANCELLATION_REQUESTED;
 const QueryResult_QueryState QueryResult::QueryState_MIN;
 const QueryResult_QueryState QueryResult::QueryState_MAX;
 const int QueryResult::QueryState_ARRAYSIZE;
@@ -4181,15 +4197,7 @@ const int QueryResult::QueryState_ARRAYSIZE;
 #ifndef _MSC_VER
 const int QueryResult::kQueryStateFieldNumber;
 const int QueryResult::kQueryIdFieldNumber;
-const int QueryResult::kIsLastChunkFieldNumber;
-const int QueryResult::kRowCountFieldNumber;
-const int QueryResult::kRecordsScanFieldNumber;
-const int QueryResult::kRecordsErrorFieldNumber;
-const int QueryResult::kSubmissionTimeFieldNumber;
-const int QueryResult::kNodeStatusFieldNumber;
 const int QueryResult::kErrorFieldNumber;
-const int QueryResult::kDefFieldNumber;
-const int QueryResult::kSchemaChangedFieldNumber;
 #endif  // !_MSC_VER
 
 QueryResult::QueryResult()
@@ -4199,7 +4207,6 @@ QueryResult::QueryResult()
 
 void QueryResult::InitAsDefaultInstance() {
   query_id_ = const_cast< ::exec::shared::QueryId*>(&::exec::shared::QueryId::default_instance());
-  def_ = const_cast< ::exec::shared::RecordBatchDef*>(&::exec::shared::RecordBatchDef::default_instance());
 }
 
 QueryResult::QueryResult(const QueryResult& from)
@@ -4212,13 +4219,6 @@ void QueryResult::SharedCtor() {
   _cached_size_ = 0;
   query_state_ = 0;
   query_id_ = NULL;
-  is_last_chunk_ = false;
-  row_count_ = 0;
-  records_scan_ = GOOGLE_LONGLONG(0);
-  records_error_ = GOOGLE_LONGLONG(0);
-  submission_time_ = GOOGLE_LONGLONG(0);
-  def_ = NULL;
-  schema_changed_ = false;
   ::memset(_has_bits_, 0, sizeof(_has_bits_));
 }
 
@@ -4229,7 +4229,6 @@ QueryResult::~QueryResult() {
 void QueryResult::SharedDtor() {
   if (this != default_instance_) {
     delete query_id_;
-    delete def_;
   }
 }
 
@@ -4260,19 +4259,7 @@ void QueryResult::Clear() {
     if (has_query_id()) {
       if (query_id_ != NULL) query_id_->::exec::shared::QueryId::Clear();
     }
-    is_last_chunk_ = false;
-    row_count_ = 0;
-    records_scan_ = GOOGLE_LONGLONG(0);
-    records_error_ = GOOGLE_LONGLONG(0);
-    submission_time_ = GOOGLE_LONGLONG(0);
   }
-  if (_has_bits_[9 / 32] & (0xffu << (9 % 32))) {
-    if (has_def()) {
-      if (def_ != NULL) def_->::exec::shared::RecordBatchDef::Clear();
-    }
-    schema_changed_ = false;
-  }
-  node_status_.Clear();
   error_.Clear();
   ::memset(_has_bits_, 0, sizeof(_has_bits_));
   mutable_unknown_fields()->Clear();
@@ -4314,122 +4301,311 @@ bool QueryResult::MergePartialFromCodedStream(
         } else {
           goto handle_uninterpreted;
         }
-        if (input->ExpectTag(24)) goto parse_is_last_chunk;
+        if (input->ExpectTag(26)) goto parse_error;
         break;
       }
 
-      // optional bool is_last_chunk = 3;
+      // repeated .exec.shared.DrillPBError error = 3;
       case 3: {
         if (::google::protobuf::internal::WireFormatLite::GetTagWireType(tag) ==
-            ::google::protobuf::internal::WireFormatLite::WIRETYPE_VARINT) {
-         parse_is_last_chunk:
-          DO_((::google::protobuf::internal::WireFormatLite::ReadPrimitive<
-                   bool, ::google::protobuf::internal::WireFormatLite::TYPE_BOOL>(
-                 input, &is_last_chunk_)));
-          set_has_is_last_chunk();
+            ::google::protobuf::internal::WireFormatLite::WIRETYPE_LENGTH_DELIMITED) {
+         parse_error:
+          DO_(::google::protobuf::internal::WireFormatLite::ReadMessageNoVirtual(
+                input, add_error()));
         } else {
           goto handle_uninterpreted;
         }
-        if (input->ExpectTag(32)) goto parse_row_count;
+        if (input->ExpectTag(26)) goto parse_error;
+        if (input->ExpectAtEnd()) return true;
         break;
       }
 
-      // optional int32 row_count = 4;
-      case 4: {
+      default: {
+      handle_uninterpreted:
         if (::google::protobuf::internal::WireFormatLite::GetTagWireType(tag) ==
-            ::google::protobuf::internal::WireFormatLite::WIRETYPE_VARINT) {
-         parse_row_count:
-          DO_((::google::protobuf::internal::WireFormatLite::ReadPrimitive<
-                   ::google::protobuf::int32, ::google::protobuf::internal::WireFormatLite::TYPE_INT32>(
-                 input, &row_count_)));
-          set_has_row_count();
-        } else {
-          goto handle_uninterpreted;
+            ::google::protobuf::internal::WireFormatLite::WIRETYPE_END_GROUP) {
+          return true;
         }
-        if (input->ExpectTag(40)) goto parse_records_scan;
+        DO_(::google::protobuf::internal::WireFormat::SkipField(
+              input, tag, mutable_unknown_fields()));
         break;
       }
+    }
+  }
+  return true;
+#undef DO_
+}
 
-      // optional int64 records_scan = 5;
-      case 5: {
-        if (::google::protobuf::internal::WireFormatLite::GetTagWireType(tag) ==
-            ::google::protobuf::internal::WireFormatLite::WIRETYPE_VARINT) {
-         parse_records_scan:
-          DO_((::google::protobuf::internal::WireFormatLite::ReadPrimitive<
-                   ::google::protobuf::int64, ::google::protobuf::internal::WireFormatLite::TYPE_INT64>(
-                 input, &records_scan_)));
-          set_has_records_scan();
-        } else {
-          goto handle_uninterpreted;
-        }
-        if (input->ExpectTag(48)) goto parse_records_error;
-        break;
-      }
+void QueryResult::SerializeWithCachedSizes(
+    ::google::protobuf::io::CodedOutputStream* output) const {
+  // optional .exec.shared.QueryResult.QueryState query_state = 1;
+  if (has_query_state()) {
+    ::google::protobuf::internal::WireFormatLite::WriteEnum(
+      1, this->query_state(), output);
+  }
 
-      // optional int64 records_error = 6;
-      case 6: {
-        if (::google::protobuf::internal::WireFormatLite::GetTagWireType(tag) ==
-            ::google::protobuf::internal::WireFormatLite::WIRETYPE_VARINT) {
-         parse_records_error:
-          DO_((::google::protobuf::internal::WireFormatLite::ReadPrimitive<
-                   ::google::protobuf::int64, ::google::protobuf::internal::WireFormatLite::TYPE_INT64>(
-                 input, &records_error_)));
-          set_has_records_error();
-        } else {
-          goto handle_uninterpreted;
-        }
-        if (input->ExpectTag(56)) goto parse_submission_time;
-        break;
-      }
+  // optional .exec.shared.QueryId query_id = 2;
+  if (has_query_id()) {
+    ::google::protobuf::internal::WireFormatLite::WriteMessageMaybeToArray(
+      2, this->query_id(), output);
+  }
 
-      // optional int64 submission_time = 7;
-      case 7: {
-        if (::google::protobuf::internal::WireFormatLite::GetTagWireType(tag) ==
-            ::google::protobuf::internal::WireFormatLite::WIRETYPE_VARINT) {
-         parse_submission_time:
-          DO_((::google::protobuf::internal::WireFormatLite::ReadPrimitive<
-                   ::google::protobuf::int64, ::google::protobuf::internal::WireFormatLite::TYPE_INT64>(
-                 input, &submission_time_)));
-          set_has_submission_time();
-        } else {
-          goto handle_uninterpreted;
-        }
-        if (input->ExpectTag(66)) goto parse_node_status;
-        break;
-      }
+  // repeated .exec.shared.DrillPBError error = 3;
+  for (int i = 0; i < this->error_size(); i++) {
+    ::google::protobuf::internal::WireFormatLite::WriteMessageMaybeToArray(
+      3, this->error(i), output);
+  }
 
-      // repeated .exec.shared.NodeStatus node_status = 8;
-      case 8: {
+  if (!unknown_fields().empty()) {
+    ::google::protobuf::internal::WireFormat::SerializeUnknownFields(
+        unknown_fields(), output);
+  }
+}
+
+::google::protobuf::uint8* QueryResult::SerializeWithCachedSizesToArray(
+    ::google::protobuf::uint8* target) const {
+  // optional .exec.shared.QueryResult.QueryState query_state = 1;
+  if (has_query_state()) {
+    target = ::google::protobuf::internal::WireFormatLite::WriteEnumToArray(
+      1, this->query_state(), target);
+  }
+
+  // optional .exec.shared.QueryId query_id = 2;
+  if (has_query_id()) {
+    target = ::google::protobuf::internal::WireFormatLite::
+      WriteMessageNoVirtualToArray(
+        2, this->query_id(), target);
+  }
+
+  // repeated .exec.shared.DrillPBError error = 3;
+  for (int i = 0; i < this->error_size(); i++) {
+    target = ::google::protobuf::internal::WireFormatLite::
+      WriteMessageNoVirtualToArray(
+        3, this->error(i), target);
+  }
+
+  if (!unknown_fields().empty()) {
+    target = ::google::protobuf::internal::WireFormat::SerializeUnknownFieldsToArray(
+        unknown_fields(), target);
+  }
+  return target;
+}
+
+int QueryResult::ByteSize() const {
+  int total_size = 0;
+
+  if (_has_bits_[0 / 32] & (0xffu << (0 % 32))) {
+    // optional .exec.shared.QueryResult.QueryState query_state = 1;
+    if (has_query_state()) {
+      total_size += 1 +
+        ::google::protobuf::internal::WireFormatLite::EnumSize(this->query_state());
+    }
+
+    // optional .exec.shared.QueryId query_id = 2;
+    if (has_query_id()) {
+      total_size += 1 +
+        ::google::protobuf::internal::WireFormatLite::MessageSizeNoVirtual(
+          this->query_id());
+    }
+
+  }
+  // repeated .exec.shared.DrillPBError error = 3;
+  total_size += 1 * this->error_size();
+  for (int i = 0; i < this->error_size(); i++) {
+    total_size +=
+      ::google::protobuf::internal::WireFormatLite::MessageSizeNoVirtual(
+        this->error(i));
+  }
+
+  if (!unknown_fields().empty()) {
+    total_size +=
+      ::google::protobuf::internal::WireFormat::ComputeUnknownFieldsSize(
+        unknown_fields());
+  }
+  GOOGLE_SAFE_CONCURRENT_WRITES_BEGIN();
+  _cached_size_ = total_size;
+  GOOGLE_SAFE_CONCURRENT_WRITES_END();
+  return total_size;
+}
+
+void QueryResult::MergeFrom(const ::google::protobuf::Message& from) {
+  GOOGLE_CHECK_NE(&from, this);
+  const QueryResult* source =
+    ::google::protobuf::internal::dynamic_cast_if_available<const QueryResult*>(
+      &from);
+  if (source == NULL) {
+    ::google::protobuf::internal::ReflectionOps::Merge(from, this);
+  } else {
+    MergeFrom(*source);
+  }
+}
+
+void QueryResult::MergeFrom(const QueryResult& from) {
+  GOOGLE_CHECK_NE(&from, this);
+  error_.MergeFrom(from.error_);
+  if (from._has_bits_[0 / 32] & (0xffu << (0 % 32))) {
+    if (from.has_query_state()) {
+      set_query_state(from.query_state());
+    }
+    if (from.has_query_id()) {
+      mutable_query_id()->::exec::shared::QueryId::MergeFrom(from.query_id());
+    }
+  }
+  mutable_unknown_fields()->MergeFrom(from.unknown_fields());
+}
+
+void QueryResult::CopyFrom(const ::google::protobuf::Message& from) {
+  if (&from == this) return;
+  Clear();
+  MergeFrom(from);
+}
+
+void QueryResult::CopyFrom(const QueryResult& from) {
+  if (&from == this) return;
+  Clear();
+  MergeFrom(from);
+}
+
+bool QueryResult::IsInitialized() const {
+
+  return true;
+}
+
+void QueryResult::Swap(QueryResult* other) {
+  if (other != this) {
+    std::swap(query_state_, other->query_state_);
+    std::swap(query_id_, other->query_id_);
+    error_.Swap(&other->error_);
+    std::swap(_has_bits_[0], other->_has_bits_[0]);
+    _unknown_fields_.Swap(&other->_unknown_fields_);
+    std::swap(_cached_size_, other->_cached_size_);
+  }
+}
+
+::google::protobuf::Metadata QueryResult::GetMetadata() const {
+  protobuf_AssignDescriptorsOnce();
+  ::google::protobuf::Metadata metadata;
+  metadata.descriptor = QueryResult_descriptor_;
+  metadata.reflection = QueryResult_reflection_;
+  return metadata;
+}
+
+
+// ===================================================================
+
+#ifndef _MSC_VER
+const int QueryData::kQueryIdFieldNumber;
+const int QueryData::kRowCountFieldNumber;
+const int QueryData::kDefFieldNumber;
+#endif  // !_MSC_VER
+
+QueryData::QueryData()
+  : ::google::protobuf::Message() {
+  SharedCtor();
+}
+
+void QueryData::InitAsDefaultInstance() {
+  query_id_ = const_cast< ::exec::shared::QueryId*>(&::exec::shared::QueryId::default_instance());
+  def_ = const_cast< ::exec::shared::RecordBatchDef*>(&::exec::shared::RecordBatchDef::default_instance());
+}
+
+QueryData::QueryData(const QueryData& from)
+  : ::google::protobuf::Message() {
+  SharedCtor();
+  MergeFrom(from);
+}
+
+void QueryData::SharedCtor() {
+  _cached_size_ = 0;
+  query_id_ = NULL;
+  row_count_ = 0;
+  def_ = NULL;
+  ::memset(_has_bits_, 0, sizeof(_has_bits_));
+}
+
+QueryData::~QueryData() {
+  SharedDtor();
+}
+
+void QueryData::SharedDtor() {
+  if (this != default_instance_) {
+    delete query_id_;
+    delete def_;
+  }
+}
+
+void QueryData::SetCachedSize(int size) const {
+  GOOGLE_SAFE_CONCURRENT_WRITES_BEGIN();
+  _cached_size_ = size;
+  GOOGLE_SAFE_CONCURRENT_WRITES_END();
+}
+const ::google::protobuf::Descriptor* QueryData::descriptor() {
+  protobuf_AssignDescriptorsOnce();
+  return QueryData_descriptor_;
+}
+
+const QueryData& QueryData::default_instance() {
+  if (default_instance_ == NULL) protobuf_AddDesc_UserBitShared_2eproto();
+  return *default_instance_;
+}
+
+QueryData* QueryData::default_instance_ = NULL;
+
+QueryData* QueryData::New() const {
+  return new QueryData;
+}
+
+void QueryData::Clear() {
+  if (_has_bits_[0 / 32] & (0xffu << (0 % 32))) {
+    if (has_query_id()) {
+      if (query_id_ != NULL) query_id_->::exec::shared::QueryId::Clear();
+    }
+    row_count_ = 0;
+    if (has_def()) {
+      if (def_ != NULL) def_->::exec::shared::RecordBatchDef::Clear();
+    }
+  }
+  ::memset(_has_bits_, 0, sizeof(_has_bits_));
+  mutable_unknown_fields()->Clear();
+}
+
+bool QueryData::MergePartialFromCodedStream(
+    ::google::protobuf::io::CodedInputStream* input) {
+#define DO_(EXPRESSION) if (!(EXPRESSION)) return false
+  ::google::protobuf::uint32 tag;
+  while ((tag = input->ReadTag()) != 0) {
+    switch (::google::protobuf::internal::WireFormatLite::GetTagFieldNumber(tag)) {
+      // optional .exec.shared.QueryId query_id = 1;
+      case 1: {
         if (::google::protobuf::internal::WireFormatLite::GetTagWireType(tag) ==
             ::google::protobuf::internal::WireFormatLite::WIRETYPE_LENGTH_DELIMITED) {
-         parse_node_status:
           DO_(::google::protobuf::internal::WireFormatLite::ReadMessageNoVirtual(
-                input, add_node_status()));
+               input, mutable_query_id()));
         } else {
           goto handle_uninterpreted;
         }
-        if (input->ExpectTag(66)) goto parse_node_status;
-        if (input->ExpectTag(74)) goto parse_error;
+        if (input->ExpectTag(16)) goto parse_row_count;
         break;
       }
 
-      // repeated .exec.shared.DrillPBError error = 9;
-      case 9: {
+      // optional int32 row_count = 2;
+      case 2: {
         if (::google::protobuf::internal::WireFormatLite::GetTagWireType(tag) ==
-            ::google::protobuf::internal::WireFormatLite::WIRETYPE_LENGTH_DELIMITED) {
-         parse_error:
-          DO_(::google::protobuf::internal::WireFormatLite::ReadMessageNoVirtual(
-                input, add_error()));
+            ::google::protobuf::internal::WireFormatLite::WIRETYPE_VARINT) {
+         parse_row_count:
+          DO_((::google::protobuf::internal::WireFormatLite::ReadPrimitive<
+                   ::google::protobuf::int32, ::google::protobuf::internal::WireFormatLite::TYPE_INT32>(
+                 input, &row_count_)));
+          set_has_row_count();
         } else {
           goto handle_uninterpreted;
         }
-        if (input->ExpectTag(74)) goto parse_error;
-        if (input->ExpectTag(82)) goto parse_def;
+        if (input->ExpectTag(26)) goto parse_def;
         break;
       }
 
-      // optional .exec.shared.RecordBatchDef def = 10;
-      case 10: {
+      // optional .exec.shared.RecordBatchDef def = 3;
+      case 3: {
         if (::google::protobuf::internal::WireFormatLite::GetTagWireType(tag) ==
             ::google::protobuf::internal::WireFormatLite::WIRETYPE_LENGTH_DELIMITED) {
          parse_def:
@@ -4438,22 +4614,6 @@ bool QueryResult::MergePartialFromCodedStream(
         } else {
           goto handle_uninterpreted;
         }
-        if (input->ExpectTag(88)) goto parse_schema_changed;
-        break;
-      }
-
-      // optional bool schema_changed = 11;
-      case 11: {
-        if (::google::protobuf::internal::WireFormatLite::GetTagWireType(tag) ==
-            ::google::protobuf::internal::WireFormatLite::WIRETYPE_VARINT) {
-         parse_schema_changed:
-          DO_((::google::protobuf::internal::WireFormatLite::ReadPrimitive<
-                   bool, ::google::protobuf::internal::WireFormatLite::TYPE_BOOL>(
-                 input, &schema_changed_)));
-          set_has_schema_changed();
-        } else {
-          goto handle_uninterpreted;
-        }
         if (input->ExpectAtEnd()) return true;
         break;
       }
@@ -4474,66 +4634,23 @@ bool QueryResult::MergePartialFromCodedStream(
 #undef DO_
 }
 
-void QueryResult::SerializeWithCachedSizes(
+void QueryData::SerializeWithCachedSizes(
     ::google::protobuf::io::CodedOutputStream* output) const {
-  // optional .exec.shared.QueryResult.QueryState query_state = 1;
-  if (has_query_state()) {
-    ::google::protobuf::internal::WireFormatLite::WriteEnum(
-      1, this->query_state(), output);
-  }
-
-  // optional .exec.shared.QueryId query_id = 2;
+  // optional .exec.shared.QueryId query_id = 1;
   if (has_query_id()) {
     ::google::protobuf::internal::WireFormatLite::WriteMessageMaybeToArray(
-      2, this->query_id(), output);
-  }
-
-  // optional bool is_last_chunk = 3;
-  if (has_is_last_chunk()) {
-    ::google::protobuf::internal::WireFormatLite::WriteBool(3, this->is_last_chunk(), output);
+      1, this->query_id(), output);
   }
 
-  // optional int32 row_count = 4;
+  // optional int32 row_count = 2;
   if (has_row_count()) {
-    ::google::protobuf::internal::WireFormatLite::WriteInt32(4, this->row_count(), output);
-  }
-
-  // optional int64 records_scan = 5;
-  if (has_records_scan()) {
-    ::google::protobuf::internal::WireFormatLite::WriteInt64(5, this->records_scan(), output);
-  }
-
-  // optional int64 records_error = 6;
-  if (has_records_error()) {
-    ::google::protobuf::internal::WireFormatLite::WriteInt64(6, this->records_error(), output);
+    ::google::protobuf::internal::WireFormatLite::WriteInt32(2, this->row_count(), output);
   }
 
-  // optional int64 submission_time = 7;
-  if (has_submission_time()) {
-    ::google::protobuf::internal::WireFormatLite::WriteInt64(7, this->submission_time(), output);
-  }
-
-  // repeated .exec.shared.NodeStatus node_status = 8;
-  for (int i = 0; i < this->node_status_size(); i++) {
-    ::google::protobuf::internal::WireFormatLite::WriteMessageMaybeToArray(
-      8, this->node_status(i), output);
-  }
-
-  // repeated .exec.shared.DrillPBError error = 9;
-  for (int i = 0; i < this->error_size(); i++) {
-    ::google::protobuf::internal::WireFormatLite::WriteMessageMaybeToArray(
-      9, this->error(i), output);
-  }
-
-  // optional .exec.shared.RecordBatchDef def = 10;
+  // optional .exec.shared.RecordBatchDef def = 3;
   if (has_def()) {
     ::google::protobuf::internal::WireFormatLite::WriteMessageMaybeToArray(
-      10, this->def(), output);
-  }
-
-  // optional bool schema_changed = 11;
-  if (has_schema_changed()) {
-    ::google::protobuf::internal::WireFormatLite::WriteBool(11, this->schema_changed(), output);
+      3, this->def(), output);
   }
 
   if (!unknown_fields().empty()) {
@@ -4542,70 +4659,25 @@ void QueryResult::SerializeWithCachedSizes(
   }
 }
 
-::google::protobuf::uint8* QueryResult::SerializeWithCachedSizesToArray(
+::google::protobuf::uint8* QueryData::SerializeWithCachedSizesToArray(
     ::google::protobuf::uint8* target) const {
-  // optional .exec.shared.QueryResult.QueryState query_state = 1;
-  if (has_query_state()) {
-    target = ::google::protobuf::internal::WireFormatLite::WriteEnumToArray(
-      1, this->query_state(), target);
-  }
-
-  // optional .exec.shared.QueryId query_id = 2;
+  // optional .exec.shared.QueryId query_id = 1;
   if (has_query_id()) {
     target = ::google::protobuf::internal::WireFormatLite::
       WriteMessageNoVirtualToArray(
-        2, this->query_id(), target);
+        1, this->query_id(), target);
   }
 
-  // optional bool is_last_chunk = 3;
-  if (has_is_last_chunk()) {
-    target = ::google::protobuf::internal::WireFormatLite::WriteBoolToArray(3, this->is_last_chunk(), target);
-  }
-
-  // optional int32 row_count = 4;
+  // optional int32 row_count = 2;
   if (has_row_count()) {
-    target = ::google::protobuf::internal::WireFormatLite::WriteInt32ToArray(4, this->row_count(), target);
-  }
-
-  // optional int64 records_scan = 5;
-  if (has_records_scan()) {
-    target = ::google::protobuf::internal::WireFormatLite::WriteInt64ToArray(5, this->records_scan(), target);
-  }
-
-  // optional int64 records_error = 6;
-  if (has_records_error()) {
-    target = ::google::protobuf::internal::WireFormatLite::WriteInt64ToArray(6, this->records_error(), target);
-  }
-
-  // optional int64 submission_time = 7;
-  if (has_submission_time()) {
-    target = ::google::protobuf::internal::WireFormatLite::WriteInt64ToArray(7, this->submission_time(), target);
+    target = ::google::protobuf::internal::WireFormatLite::WriteInt32ToArray(2, this->row_count(), target);
   }
 
-  // repeated .exec.shared.NodeStatus node_status = 8;
-  for (int i = 0; i < this->node_status_size(); i++) {
-    target = ::google::protobuf::internal::WireFormatLite::
-      WriteMessageNoVirtualToArray(
-        8, this->node_status(i), target);
-  }
-
-  // repeated .exec.shared.DrillPBError error = 9;
-  for (int i = 0; i < this->error_size(); i++) {
-    target = ::google::protobuf::internal::WireFormatLite::
-      WriteMessageNoVirtualToArray(
-        9, this->error(i), target);
-  }
-
-  // optional .exec.shared.RecordBatchDef def = 10;
+  // optional .exec.shared.RecordBatchDef def = 3;
   if (has_def()) {
     target = ::google::protobuf::internal::WireFormatLite::
       WriteMessageNoVirtualToArray(
-        10, this->def(), target);
-  }
-
-  // optional bool schema_changed = 11;
-  if (has_schema_changed()) {
-    target = ::google::protobuf::internal::WireFormatLite::WriteBoolToArray(11, this->schema_changed(), target);
+        3, this->def(), target);
   }
 
   if (!unknown_fields().empty()) {
@@ -4615,87 +4687,32 @@ void QueryResult::SerializeWithCachedSizes(
   return target;
 }
 
-int QueryResult::ByteSize() const {
+int QueryData::ByteSize() const {
   int total_size = 0;
 
   if (_has_bits_[0 / 32] & (0xffu << (0 % 32))) {
-    // optional .exec.shared.QueryResult.QueryState query_state = 1;
-    if (has_query_state()) {
-      total_size += 1 +
-        ::google::protobuf::internal::WireFormatLite::EnumSize(this->query_state());
-    }
-
-    // optional .exec.shared.QueryId query_id = 2;
+    // optional .exec.shared.QueryId query_id = 1;
     if (has_query_id()) {
       total_size += 1 +
         ::google::protobuf::internal::WireFormatLite::MessageSizeNoVirtual(
           this->query_id());
     }
 
-    // optional bool is_last_chunk = 3;
-    if (has_is_last_chunk()) {
-      total_size += 1 + 1;
-    }
-
-    // optional int32 row_count = 4;
+    // optional int32 row_count = 2;
     if (has_row_count()) {
       total_size += 1 +
         ::google::protobuf::internal::WireFormatLite::Int32Size(
           this->row_count());
     }
 
-    // optional int64 records_scan = 5;
-    if (has_records_scan()) {
-      total_size += 1 +
-        ::google::protobuf::internal::WireFormatLite::Int64Size(
-          this->records_scan());
-    }
-
-    // optional int64 records_error = 6;
-    if (has_records_error()) {
-      total_size += 1 +
-        ::google::protobuf::internal::WireFormatLite::Int64Size(
-          this->records_error());
-    }
-
-    // optional int64 submission_time = 7;
-    if (has_submission_time()) {
-      total_size += 1 +
-        ::google::protobuf::internal::WireFormatLite::Int64Size(
-          this->submission_time());
-    }
-
-  }
-  if (_has_bits_[9 / 32] & (0xffu << (9 % 32))) {
-    // optional .exec.shared.RecordBatchDef def = 10;
+    // optional .exec.shared.RecordBatchDef def = 3;
     if (has_def()) {
       total_size += 1 +
         ::google::protobuf::internal::WireFormatLite::MessageSizeNoVirtual(
           this->def());
     }
 
-    // optional bool schema_changed = 11;
-    if (has_schema_changed()) {
-      total_size += 1 + 1;
-    }
-
   }
-  // repeated .exec.shared.NodeStatus node_status = 8;
-  total_size += 1 * this->node_status_size();
-  for (int i = 0; i < this->node_status_size(); i++) {
-    total_size +=
-      ::google::protobuf::internal::WireFormatLite::MessageSizeNoVirtual(
-        this->node_status(i));
-  }
-
-  // repeated .exec.shared.DrillPBError error = 9;
-  total_size += 1 * this->error_size();
-  for (int i = 0; i < this->error_size(); i++) {
-    total_size +=
-      ::google::protobuf::internal::WireFormatLite::MessageSizeNoVirtual(
-        this->error(i));
-  }
-
   if (!unknown_fields().empty()) {
     total_size +=
       ::google::protobuf::internal::WireFormat::ComputeUnknownFieldsSize(
@@ -4707,10 +4724,10 @@ int QueryResult::ByteSize() const {
   return total_size;
 }
 
-void QueryResult::MergeFrom(const ::google::protobuf::Message& from) {
+void QueryData::MergeFrom(const ::google::protobuf::Message& from) {
   GOOGLE_CHECK_NE(&from, this);
-  const QueryResult* source =
-    ::google::protobuf::internal::dynamic_cast_if_available<const QueryResult*>(
+  const QueryData* source =
+    ::google::protobuf::internal::dynamic_cast_if_available<const QueryData*>(
       &from);
   if (source == NULL) {
     ::google::protobuf::internal::ReflectionOps::Merge(from, this);
@@ -4719,85 +4736,55 @@ void QueryResult::MergeFrom(const ::google::protobuf::Message& from) {
   }
 }
 
-void QueryResult::MergeFrom(const QueryResult& from) {
+void QueryData::MergeFrom(const QueryData& from) {
   GOOGLE_CHECK_NE(&from, this);
-  node_status_.MergeFrom(from.node_status_);
-  error_.MergeFrom(from.error_);
   if (from._has_bits_[0 / 32] & (0xffu << (0 % 32))) {
-    if (from.has_query_state()) {
-      set_query_state(from.query_state());
-    }
     if (from.has_query_id()) {
       mutable_query_id()->::exec::shared::QueryId::MergeFrom(from.query_id());
     }
-    if (from.has_is_last_chunk()) {
-      set_is_last_chunk(from.is_last_chunk());
-    }
     if (from.has_row_count()) {
       set_row_count(from.row_count());
     }
-    if (from.has_records_scan()) {
-      set_records_scan(from.records_scan());
-    }
-    if (from.has_records_error()) {
-      set_records_error(from.records_error());
-    }
-    if (from.has_submission_time()) {
-      set_submission_time(from.submission_time());
-    }
-  }
-  if (from._has_bits_[9 / 32] & (0xffu << (9 % 32))) {
     if (from.has_def()) {
       mutable_def()->::exec::shared::RecordBatchDef::MergeFrom(from.def());
     }
-    if (from.has_schema_changed()) {
-      set_schema_changed(from.schema_changed());
-    }
   }
   mutable_unknown_fields()->MergeFrom(from.unknown_fields());
 }
 
-void QueryResult::CopyFrom(const ::google::protobuf::Message& from) {
+void QueryData::CopyFrom(const ::google::protobuf::Message& from) {
   if (&from == this) return;
   Clear();
   MergeFrom(from);
 }
 
-void QueryResult::CopyFrom(const QueryResult& from) {
+void QueryData::CopyFrom(const QueryData& from) {
   if (&from == this) return;
   Clear();
   MergeFrom(from);
 }
 
-bool QueryResult::IsInitialized() const {
+bool QueryData::IsInitialized() const {
 
   return true;
 }
 
-void QueryResult::Swap(QueryResult* other) {
+void QueryData::Swap(QueryData* other) {
   if (other != this) {
-    std::swap(query_state_, other->query_state_);
     std::swap(query_id_, other->query_id_);
-    std::swap(is_last_chunk_, other->is_last_chunk_);
     std::swap(row_count_, other->row_count_);
-    std::swap(records_scan_, other->records_scan_);
-    std::swap(records_error_, other->records_error_);
-    std::swap(submission_time_, other->submission_time_);
-    node_status_.Swap(&other->node_status_);
-    error_.Swap(&other->error_);
     std::swap(def_, other->def_);
-    std::swap(schema_changed_, other->schema_changed_);
     std::swap(_has_bits_[0], other->_has_bits_[0]);
     _unknown_fields_.Swap(&other->_unknown_fields_);
     std::swap(_cached_size_, other->_cached_size_);
   }
 }
 
-::google::protobuf::Metadata QueryResult::GetMetadata() const {
+::google::protobuf::Metadata QueryData::GetMetadata() const {
   protobuf_AssignDescriptorsOnce();
   ::google::protobuf::Metadata metadata;
-  metadata.descriptor = QueryResult_descriptor_;
-  metadata.reflection = QueryResult_reflection_;
+  metadata.descriptor = QueryData_descriptor_;
+  metadata.reflection = QueryData_reflection_;
   return metadata;
 }
 
@@ -6691,7 +6678,7 @@ const int OperatorProfile::kOperatorIdFieldNumber;
 const int OperatorProfile::kOperatorTypeFieldNumber;
 const int OperatorProfile::kSetupNanosFieldNumber;
 const int OperatorProfile::kProcessNanosFieldNumber;
-const int OperatorProfile::kLocalMemoryAllocatedFieldNumber;
+const int OperatorProfile::kPeakLocalMemoryAllocatedFieldNumber;
 const int OperatorProfile::kMetricFieldNumber;
 const int OperatorProfile::kWaitNanosFieldNumber;
 #endif  // !_MSC_VER
@@ -6716,7 +6703,7 @@ void OperatorProfile::SharedCtor() {
   operator_type_ = 0;
   setup_nanos_ = GOOGLE_LONGLONG(0);
   process_nanos_ = GOOGLE_LONGLONG(0);
-  local_memory_allocated_ = GOOGLE_LONGLONG(0);
+  peak_local_memory_allocated_ = GOOGLE_LONGLONG(0);
   wait_nanos_ = GOOGLE_LONGLONG(0);
   ::memset(_has_bits_, 0, sizeof(_has_bits_));
 }
@@ -6757,7 +6744,7 @@ void OperatorProfile::Clear() {
     operator_type_ = 0;
     setup_nanos_ = GOOGLE_LONGLONG(0);
     process_nanos_ = GOOGLE_LONGLONG(0);
-    local_memory_allocated_ = GOOGLE_LONGLONG(0);
+    peak_local_memory_allocated_ = GOOGLE_LONGLONG(0);
     wait_nanos_ = GOOGLE_LONGLONG(0);
   }
   input_profile_.Clear();
@@ -6847,19 +6834,19 @@ bool OperatorProfile::MergePartialFromCodedStream(
         } else {
           goto handle_uninterpreted;
         }
-        if (input->ExpectTag(56)) goto parse_local_memory_allocated;
+        if (input->ExpectTag(56)) goto parse_peak_local_memory_allocated;
         break;
       }
 
-      // optional int64 local_memory_allocated = 7;
+      // optional int64 peak_local_memory_allocated = 7;
       case 7: {
         if (::google::protobuf::internal::WireFormatLite::GetTagWireType(tag) ==
             ::google::protobuf::internal::WireFormatLite::WIRETYPE_VARINT) {
-         parse_local_memory_allocated:
+         parse_peak_local_memory_allocated:
           DO_((::google::protobuf::internal::WireFormatLite::ReadPrimitive<
                    ::google::protobuf::int64, ::google::protobuf::internal::WireFormatLite::TYPE_INT64>(
-                 input, &local_memory_allocated_)));
-          set_has_local_memory_allocated();
+                 input, &peak_local_memory_allocated_)));
+          set_has_peak_local_memory_allocated();
         } else {
           goto handle_uninterpreted;
         }
@@ -6942,9 +6929,9 @@ void OperatorProfile::SerializeWithCachedSizes(
     ::google::protobuf::internal::WireFormatLite::WriteInt64(6, this->process_nanos(), output);
   }
 
-  // optional int64 local_memory_allocated = 7;
-  if (has_local_memory_allocated()) {
-    ::google::protobuf::internal::WireFormatLite::WriteInt64(7, this->local_memory_allocated(), output);
+  // optional int64 peak_local_memory_allocated = 7;
+  if (has_peak_local_memory_allocated()) {
+    ::google::protobuf::internal::WireFormatLite::WriteInt64(7, this->peak_local_memory_allocated(), output);
   }
 
   // repeated .exec.shared.MetricValue metric = 8;
@@ -6993,9 +6980,9 @@ void OperatorProfile::SerializeWithCachedSizes(
     target = ::google::protobuf::internal::WireFormatLite::WriteInt64ToArray(6, this->process_nanos(), target);
   }
 
-  // optional int64 local_memory_allocated = 7;
-  if (has_local_memory_allocated()) {
-    target = ::google::protobuf::internal::WireFormatLite::WriteInt64ToArray(7, this->local_memory_allocated(), target);
+  // optional int64 peak_local_memory_allocated = 7;
+  if (has_peak_local_memory_allocated()) {
+    target = ::google::protobuf::internal::WireFormatLite::WriteInt64ToArray(7, this->peak_local_memory_allocated(), target);
   }
 
   // repeated .exec.shared.MetricValue metric = 8;
@@ -7049,11 +7036,11 @@ int OperatorProfile::ByteSize() const {
           this->process_nanos());
     }
 
-    // optional int64 local_memory_allocated = 7;
-    if (has_local_memory_allocated()) {
+    // optional int64 peak_local_memory_allocated = 7;
+    if (has_peak_local_memory_allocated()) {
       total_size += 1 +
         ::google::protobuf::internal::WireFormatLite::Int64Size(
-          this->local_memory_allocated());
+          this->peak_local_memory_allocated());
     }
 
     // optional int64 wait_nanos = 9;
@@ -7120,8 +7107,8 @@ void OperatorProfile::MergeFrom(const OperatorProfile& from) {
     if (from.has_process_nanos()) {
       set_process_nanos(from.process_nanos());
     }
-    if (from.has_local_memory_allocated()) {
-      set_local_memory_allocated(from.local_memory_allocated());
+    if (from.has_peak_local_memory_allocated()) {
+      set_peak_local_memory_allocated(from.peak_local_memory_allocated());
     }
     if (from.has_wait_nanos()) {
       set_wait_nanos(from.wait_nanos());
@@ -7154,7 +7141,7 @@ void OperatorProfile::Swap(OperatorProfile* other) {
     std::swap(operator_type_, other->operator_type_);
     std::swap(setup_nanos_, other->setup_nanos_);
     std::swap(process_nanos_, other->process_nanos_);
-    std::swap(local_memory_allocated_, other->local_memory_allocated_);
+    std::swap(peak_local_memory_allocated_, other->peak_local_memory_allocated_);
     metric_.Swap(&other->metric_);
     std::swap(wait_nanos_, other->wait_nanos_);
     std::swap(_has_bits_[0], other->_has_bits_[0]);
diff --git a/contrib/native/client/src/protobuf/UserBitShared.pb.h b/contrib/native/client/src/protobuf/UserBitShared.pb.h
index bbf3fdcf0..e2f5fd0fa 100644
--- a/contrib/native/client/src/protobuf/UserBitShared.pb.h
+++ b/contrib/native/client/src/protobuf/UserBitShared.pb.h
@@ -49,6 +49,7 @@ class NamePart;
 class SerializedField;
 class NodeStatus;
 class QueryResult;
+class QueryData;
 class QueryInfo;
 class QueryProfile;
 class MajorFragmentProfile;
@@ -82,11 +83,11 @@ enum QueryResult_QueryState {
   QueryResult_QueryState_COMPLETED = 2,
   QueryResult_QueryState_CANCELED = 3,
   QueryResult_QueryState_FAILED = 4,
-  QueryResult_QueryState_UNKNOWN_QUERY = 5
+  QueryResult_QueryState_CANCELLATION_REQUESTED = 5
 };
 bool QueryResult_QueryState_IsValid(int value);
 const QueryResult_QueryState QueryResult_QueryState_QueryState_MIN = QueryResult_QueryState_PENDING;
-const QueryResult_QueryState QueryResult_QueryState_QueryState_MAX = QueryResult_QueryState_UNKNOWN_QUERY;
+const QueryResult_QueryState QueryResult_QueryState_QueryState_MAX = QueryResult_QueryState_CANCELLATION_REQUESTED;
 const int QueryResult_QueryState_QueryState_ARRAYSIZE = QueryResult_QueryState_QueryState_MAX + 1;
 
 const ::google::protobuf::EnumDescriptor* QueryResult_QueryState_descriptor();
@@ -1457,7 +1458,7 @@ class QueryResult : public ::google::protobuf::Message {
   static const QueryState COMPLETED = QueryResult_QueryState_COMPLETED;
   static const QueryState CANCELED = QueryResult_QueryState_CANCELED;
   static const QueryState FAILED = QueryResult_QueryState_FAILED;
-  static const QueryState UNKNOWN_QUERY = QueryResult_QueryState_UNKNOWN_QUERY;
+  static const QueryState CANCELLATION_REQUESTED = QueryResult_QueryState_CANCELLATION_REQUESTED;
   static inline bool QueryState_IsValid(int value) {
     return QueryResult_QueryState_IsValid(value);
   }
@@ -1497,57 +1498,10 @@ class QueryResult : public ::google::protobuf::Message {
   inline ::exec::shared::QueryId* release_query_id();
   inline void set_allocated_query_id(::exec::shared::QueryId* query_id);
 
-  // optional bool is_last_chunk = 3;
-  inline bool has_is_last_chunk() const;
-  inline void clear_is_last_chunk();
-  static const int kIsLastChunkFieldNumber = 3;
-  inline bool is_last_chunk() const;
-  inline void set_is_last_chunk(bool value);
-
-  // optional int32 row_count = 4;
-  inline bool has_row_count() const;
-  inline void clear_row_count();
-  static const int kRowCountFieldNumber = 4;
-  inline ::google::protobuf::int32 row_count() const;
-  inline void set_row_count(::google::protobuf::int32 value);
-
-  // optional int64 records_scan = 5;
-  inline bool has_records_scan() const;
-  inline void clear_records_scan();
-  static const int kRecordsScanFieldNumber = 5;
-  inline ::google::protobuf::int64 records_scan() const;
-  inline void set_records_scan(::google::protobuf::int64 value);
-
-  // optional int64 records_error = 6;
-  inline bool has_records_error() const;
-  inline void clear_records_error();
-  static const int kRecordsErrorFieldNumber = 6;
-  inline ::google::protobuf::int64 records_error() const;
-  inline void set_records_error(::google::protobuf::int64 value);
-
-  // optional int64 submission_time = 7;
-  inline bool has_submission_time() const;
-  inline void clear_submission_time();
-  static const int kSubmissionTimeFieldNumber = 7;
-  inline ::google::protobuf::int64 submission_time() const;
-  inline void set_submission_time(::google::protobuf::int64 value);
-
-  // repeated .exec.shared.NodeStatus node_status = 8;
-  inline int node_status_size() const;
-  inline void clear_node_status();
-  static const int kNodeStatusFieldNumber = 8;
-  inline const ::exec::shared::NodeStatus& node_status(int index) const;
-  inline ::exec::shared::NodeStatus* mutable_node_status(int index);
-  inline ::exec::shared::NodeStatus* add_node_status();
-  inline const ::google::protobuf::RepeatedPtrField< ::exec::shared::NodeStatus >&
-      node_status() const;
-  inline ::google::protobuf::RepeatedPtrField< ::exec::shared::NodeStatus >*
-      mutable_node_status();
-
-  // repeated .exec.shared.DrillPBError error = 9;
+  // repeated .exec.shared.DrillPBError error = 3;
   inline int error_size() const;
   inline void clear_error();
-  static const int kErrorFieldNumber = 9;
+  static const int kErrorFieldNumber = 3;
   inline const ::exec::shared::DrillPBError& error(int index) const;
   inline ::exec::shared::DrillPBError* mutable_error(int index);
   inline ::exec::shared::DrillPBError* add_error();
@@ -1556,66 +1510,134 @@ class QueryResult : public ::google::protobuf::Message {
   inline ::google::protobuf::RepeatedPtrField< ::exec::shared::DrillPBError >*
       mutable_error();
 
-  // optional .exec.shared.RecordBatchDef def = 10;
+  // @@protoc_insertion_point(class_scope:exec.shared.QueryResult)
+ private:
+  inline void set_has_query_state();
+  inline void clear_has_query_state();
+  inline void set_has_query_id();
+  inline void clear_has_query_id();
+
+  ::google::protobuf::UnknownFieldSet _unknown_fields_;
+
+  ::exec::shared::QueryId* query_id_;
+  ::google::protobuf::RepeatedPtrField< ::exec::shared::DrillPBError > error_;
+  int query_state_;
+
+  mutable int _cached_size_;
+  ::google::protobuf::uint32 _has_bits_[(3 + 31) / 32];
+
+  friend void  protobuf_AddDesc_UserBitShared_2eproto();
+  friend void protobuf_AssignDesc_UserBitShared_2eproto();
+  friend void protobuf_ShutdownFile_UserBitShared_2eproto();
+
+  void InitAsDefaultInstance();
+  static QueryResult* default_instance_;
+};
+// -------------------------------------------------------------------
+
+class QueryData : public ::google::protobuf::Message {
+ public:
+  QueryData();
+  virtual ~QueryData();
+
+  QueryData(const QueryData& from);
+
+  inline QueryData& operator=(const QueryData& from) {
+    CopyFrom(from);
+    return *this;
+  }
+
+  inline const ::google::protobuf::UnknownFieldSet& unknown_fields() const {
+    return _unknown_fields_;
+  }
+
+  inline ::google::protobuf::UnknownFieldSet* mutable_unknown_fields() {
+    return &_unknown_fields_;
+  }
+
+  static const ::google::protobuf::Descriptor* descriptor();
+  static const QueryData& default_instance();
+
+  void Swap(QueryData* other);
+
+  // implements Message ----------------------------------------------
+
+  QueryData* New() const;
+  void CopyFrom(const ::google::protobuf::Message& from);
+  void MergeFrom(const ::google::protobuf::Message& from);
+  void CopyFrom(const QueryData& from);
+  void MergeFrom(const QueryData& from);
+  void Clear();
+  bool IsInitialized() const;
+
+  int ByteSize() const;
+  bool MergePartialFromCodedStream(
+      ::google::protobuf::io::CodedInputStream* input);
+  void SerializeWithCachedSizes(
+      ::google::protobuf::io::CodedOutputStream* output) const;
+  ::google::protobuf::uint8* SerializeWithCachedSizesToArray(::google::protobuf::uint8* output) const;
+  int GetCachedSize() const { return _cached_size_; }
+  private:
+  void SharedCtor();
+  void SharedDtor();
+  void SetCachedSize(int size) const;
+  public:
+
+  ::google::protobuf::Metadata GetMetadata() const;
+
+  // nested types ----------------------------------------------------
+
+  // accessors -------------------------------------------------------
+
+  // optional .exec.shared.QueryId query_id = 1;
+  inline bool has_query_id() const;
+  inline void clear_query_id();
+  static const int kQueryIdFieldNumber = 1;
+  inline const ::exec::shared::QueryId& query_id() const;
+  inline ::exec::shared::QueryId* mutable_query_id();
+  inline ::exec::shared::QueryId* release_query_id();
+  inline void set_allocated_query_id(::exec::shared::QueryId* query_id);
+
+  // optional int32 row_count = 2;
+  inline bool has_row_count() const;
+  inline void clear_row_count();
+  static const int kRowCountFieldNumber = 2;
+  inline ::google::protobuf::int32 row_count() const;
+  inline void set_row_count(::google::protobuf::int32 value);
+
+  // optional .exec.shared.RecordBatchDef def = 3;
   inline bool has_def() const;
   inline void clear_def();
-  static const int kDefFieldNumber = 10;
+  static const int kDefFieldNumber = 3;
   inline const ::exec::shared::RecordBatchDef& def() const;
   inline ::exec::shared::RecordBatchDef* mutable_def();
   inline ::exec::shared::RecordBatchDef* release_def();
   inline void set_allocated_def(::exec::shared::RecordBatchDef* def);
 
-  // optional bool schema_changed = 11;
-  inline bool has_schema_changed() const;
-  inline void clear_schema_changed();
-  static const int kSchemaChangedFieldNumber = 11;
-  inline bool schema_changed() const;
-  inline void set_schema_changed(bool value);
-
-  // @@protoc_insertion_point(class_scope:exec.shared.QueryResult)
+  // @@protoc_insertion_point(class_scope:exec.shared.QueryData)
  private:
-  inline void set_has_query_state();
-  inline void clear_has_query_state();
   inline void set_has_query_id();
   inline void clear_has_query_id();
-  inline void set_has_is_last_chunk();
-  inline void clear_has_is_last_chunk();
   inline void set_has_row_count();
   inline void clear_has_row_count();
-  inline void set_has_records_scan();
-  inline void clear_has_records_scan();
-  inline void set_has_records_error();
-  inline void clear_has_records_error();
-  inline void set_has_submission_time();
-  inline void clear_has_submission_time();
   inline void set_has_def();
   inline void clear_has_def();
-  inline void set_has_schema_changed();
-  inline void clear_has_schema_changed();
 
   ::google::protobuf::UnknownFieldSet _unknown_fields_;
 
   ::exec::shared::QueryId* query_id_;
-  int query_state_;
-  ::google::protobuf::int32 row_count_;
-  ::google::protobuf::int64 records_scan_;
-  ::google::protobuf::int64 records_error_;
-  ::google::protobuf::int64 submission_time_;
-  ::google::protobuf::RepeatedPtrField< ::exec::shared::NodeStatus > node_status_;
-  ::google::protobuf::RepeatedPtrField< ::exec::shared::DrillPBError > error_;
   ::exec::shared::RecordBatchDef* def_;
-  bool is_last_chunk_;
-  bool schema_changed_;
+  ::google::protobuf::int32 row_count_;
 
   mutable int _cached_size_;
-  ::google::protobuf::uint32 _has_bits_[(11 + 31) / 32];
+  ::google::protobuf::uint32 _has_bits_[(3 + 31) / 32];
 
   friend void  protobuf_AddDesc_UserBitShared_2eproto();
   friend void protobuf_AssignDesc_UserBitShared_2eproto();
   friend void protobuf_ShutdownFile_UserBitShared_2eproto();
 
   void InitAsDefaultInstance();
-  static QueryResult* default_instance_;
+  static QueryData* default_instance_;
 };
 // -------------------------------------------------------------------
 
@@ -2310,12 +2332,12 @@ class OperatorProfile : public ::google::protobuf::Message {
   inline ::google::protobuf::int64 process_nanos() const;
   inline void set_process_nanos(::google::protobuf::int64 value);
 
-  // optional int64 local_memory_allocated = 7;
-  inline bool has_local_memory_allocated() const;
-  inline void clear_local_memory_allocated();
-  static const int kLocalMemoryAllocatedFieldNumber = 7;
-  inline ::google::protobuf::int64 local_memory_allocated() const;
-  inline void set_local_memory_allocated(::google::protobuf::int64 value);
+  // optional int64 peak_local_memory_allocated = 7;
+  inline bool has_peak_local_memory_allocated() const;
+  inline void clear_peak_local_memory_allocated();
+  static const int kPeakLocalMemoryAllocatedFieldNumber = 7;
+  inline ::google::protobuf::int64 peak_local_memory_allocated() const;
+  inline void set_peak_local_memory_allocated(::google::protobuf::int64 value);
 
   // repeated .exec.shared.MetricValue metric = 8;
   inline int metric_size() const;
@@ -2346,8 +2368,8 @@ class OperatorProfile : public ::google::protobuf::Message {
   inline void clear_has_setup_nanos();
   inline void set_has_process_nanos();
   inline void clear_has_process_nanos();
-  inline void set_has_local_memory_allocated();
-  inline void clear_has_local_memory_allocated();
+  inline void set_has_peak_local_memory_allocated();
+  inline void clear_has_peak_local_memory_allocated();
   inline void set_has_wait_nanos();
   inline void clear_has_wait_nanos();
 
@@ -2358,7 +2380,7 @@ class OperatorProfile : public ::google::protobuf::Message {
   ::google::protobuf::int32 operator_type_;
   ::google::protobuf::int64 setup_nanos_;
   ::google::protobuf::int64 process_nanos_;
-  ::google::protobuf::int64 local_memory_allocated_;
+  ::google::protobuf::int64 peak_local_memory_allocated_;
   ::google::protobuf::RepeatedPtrField< ::exec::shared::MetricValue > metric_;
   ::google::protobuf::int64 wait_nanos_;
 
@@ -4039,195 +4061,124 @@ inline void QueryResult::set_allocated_query_id(::exec::shared::QueryId* query_i
   }
 }
 
-// optional bool is_last_chunk = 3;
-inline bool QueryResult::has_is_last_chunk() const {
-  return (_has_bits_[0] & 0x00000004u) != 0;
-}
-inline void QueryResult::set_has_is_last_chunk() {
-  _has_bits_[0] |= 0x00000004u;
-}
-inline void QueryResult::clear_has_is_last_chunk() {
-  _has_bits_[0] &= ~0x00000004u;
-}
-inline void QueryResult::clear_is_last_chunk() {
-  is_last_chunk_ = false;
-  clear_has_is_last_chunk();
-}
-inline bool QueryResult::is_last_chunk() const {
-  return is_last_chunk_;
-}
-inline void QueryResult::set_is_last_chunk(bool value) {
-  set_has_is_last_chunk();
-  is_last_chunk_ = value;
+// repeated .exec.shared.DrillPBError error = 3;
+inline int QueryResult::error_size() const {
+  return error_.size();
 }
-
-// optional int32 row_count = 4;
-inline bool QueryResult::has_row_count() const {
-  return (_has_bits_[0] & 0x00000008u) != 0;
+inline void QueryResult::clear_error() {
+  error_.Clear();
 }
-inline void QueryResult::set_has_row_count() {
-  _has_bits_[0] |= 0x00000008u;
+inline const ::exec::shared::DrillPBError& QueryResult::error(int index) const {
+  return error_.Get(index);
 }
-inline void QueryResult::clear_has_row_count() {
-  _has_bits_[0] &= ~0x00000008u;
+inline ::exec::shared::DrillPBError* QueryResult::mutable_error(int index) {
+  return error_.Mutable(index);
 }
-inline void QueryResult::clear_row_count() {
-  row_count_ = 0;
-  clear_has_row_count();
+inline ::exec::shared::DrillPBError* QueryResult::add_error() {
+  return error_.Add();
 }
-inline ::google::protobuf::int32 QueryResult::row_count() const {
-  return row_count_;
+inline const ::google::protobuf::RepeatedPtrField< ::exec::shared::DrillPBError >&
+QueryResult::error() const {
+  return error_;
 }
-inline void QueryResult::set_row_count(::google::protobuf::int32 value) {
-  set_has_row_count();
-  row_count_ = value;
+inline ::google::protobuf::RepeatedPtrField< ::exec::shared::DrillPBError >*
+QueryResult::mutable_error() {
+  return &error_;
 }
 
-// optional int64 records_scan = 5;
-inline bool QueryResult::has_records_scan() const {
-  return (_has_bits_[0] & 0x00000010u) != 0;
-}
-inline void QueryResult::set_has_records_scan() {
-  _has_bits_[0] |= 0x00000010u;
-}
-inline void QueryResult::clear_has_records_scan() {
-  _has_bits_[0] &= ~0x00000010u;
-}
-inline void QueryResult::clear_records_scan() {
-  records_scan_ = GOOGLE_LONGLONG(0);
-  clear_has_records_scan();
-}
-inline ::google::protobuf::int64 QueryResult::records_scan() const {
-  return records_scan_;
-}
-inline void QueryResult::set_records_scan(::google::protobuf::int64 value) {
-  set_has_records_scan();
-  records_scan_ = value;
-}
+// -------------------------------------------------------------------
 
-// optional int64 records_error = 6;
-inline bool QueryResult::has_records_error() const {
-  return (_has_bits_[0] & 0x00000020u) != 0;
-}
-inline void QueryResult::set_has_records_error() {
-  _has_bits_[0] |= 0x00000020u;
-}
-inline void QueryResult::clear_has_records_error() {
-  _has_bits_[0] &= ~0x00000020u;
-}
-inline void QueryResult::clear_records_error() {
-  records_error_ = GOOGLE_LONGLONG(0);
-  clear_has_records_error();
-}
-inline ::google::protobuf::int64 QueryResult::records_error() const {
-  return records_error_;
-}
-inline void QueryResult::set_records_error(::google::protobuf::int64 value) {
-  set_has_records_error();
-  records_error_ = value;
-}
+// QueryData
 
-// optional int64 submission_time = 7;
-inline bool QueryResult::has_submission_time() const {
-  return (_has_bits_[0] & 0x00000040u) != 0;
-}
-inline void QueryResult::set_has_submission_time() {
-  _has_bits_[0] |= 0x00000040u;
-}
-inline void QueryResult::clear_has_submission_time() {
-  _has_bits_[0] &= ~0x00000040u;
-}
-inline void QueryResult::clear_submission_time() {
-  submission_time_ = GOOGLE_LONGLONG(0);
-  clear_has_submission_time();
-}
-inline ::google::protobuf::int64 QueryResult::submission_time() const {
-  return submission_time_;
-}
-inline void QueryResult::set_submission_time(::google::protobuf::int64 value) {
-  set_has_submission_time();
-  submission_time_ = value;
+// optional .exec.shared.QueryId query_id = 1;
+inline bool QueryData::has_query_id() const {
+  return (_has_bits_[0] & 0x00000001u) != 0;
 }
-
-// repeated .exec.shared.NodeStatus node_status = 8;
-inline int QueryResult::node_status_size() const {
-  return node_status_.size();
+inline void QueryData::set_has_query_id() {
+  _has_bits_[0] |= 0x00000001u;
 }
-inline void QueryResult::clear_node_status() {
-  node_status_.Clear();
+inline void QueryData::clear_has_query_id() {
+  _has_bits_[0] &= ~0x00000001u;
 }
-inline const ::exec::shared::NodeStatus& QueryResult::node_status(int index) const {
-  return node_status_.Get(index);
+inline void QueryData::clear_query_id() {
+  if (query_id_ != NULL) query_id_->::exec::shared::QueryId::Clear();
+  clear_has_query_id();
 }
-inline ::exec::shared::NodeStatus* QueryResult::mutable_node_status(int index) {
-  return node_status_.Mutable(index);
+inline const ::exec::shared::QueryId& QueryData::query_id() const {
+  return query_id_ != NULL ? *query_id_ : *default_instance_->query_id_;
 }
-inline ::exec::shared::NodeStatus* QueryResult::add_node_status() {
-  return node_status_.Add();
+inline ::exec::shared::QueryId* QueryData::mutable_query_id() {
+  set_has_query_id();
+  if (query_id_ == NULL) query_id_ = new ::exec::shared::QueryId;
+  return query_id_;
 }
-inline const ::google::protobuf::RepeatedPtrField< ::exec::shared::NodeStatus >&
-QueryResult::node_status() const {
-  return node_status_;
+inline ::exec::shared::QueryId* QueryData::release_query_id() {
+  clear_has_query_id();
+  ::exec::shared::QueryId* temp = query_id_;
+  query_id_ = NULL;
+  return temp;
 }
-inline ::google::protobuf::RepeatedPtrField< ::exec::shared::NodeStatus >*
-QueryResult::mutable_node_status() {
-  return &node_status_;
+inline void QueryData::set_allocated_query_id(::exec::shared::QueryId* query_id) {
+  delete query_id_;
+  query_id_ = query_id;
+  if (query_id) {
+    set_has_query_id();
+  } else {
+    clear_has_query_id();
+  }
 }
 
-// repeated .exec.shared.DrillPBError error = 9;
-inline int QueryResult::error_size() const {
-  return error_.size();
-}
-inline void QueryResult::clear_error() {
-  error_.Clear();
+// optional int32 row_count = 2;
+inline bool QueryData::has_row_count() const {
+  return (_has_bits_[0] & 0x00000002u) != 0;
 }
-inline const ::exec::shared::DrillPBError& QueryResult::error(int index) const {
-  return error_.Get(index);
+inline void QueryData::set_has_row_count() {
+  _has_bits_[0] |= 0x00000002u;
 }
-inline ::exec::shared::DrillPBError* QueryResult::mutable_error(int index) {
-  return error_.Mutable(index);
+inline void QueryData::clear_has_row_count() {
+  _has_bits_[0] &= ~0x00000002u;
 }
-inline ::exec::shared::DrillPBError* QueryResult::add_error() {
-  return error_.Add();
+inline void QueryData::clear_row_count() {
+  row_count_ = 0;
+  clear_has_row_count();
 }
-inline const ::google::protobuf::RepeatedPtrField< ::exec::shared::DrillPBError >&
-QueryResult::error() const {
-  return error_;
+inline ::google::protobuf::int32 QueryData::row_count() const {
+  return row_count_;
 }
-inline ::google::protobuf::RepeatedPtrField< ::exec::shared::DrillPBError >*
-QueryResult::mutable_error() {
-  return &error_;
+inline void QueryData::set_row_count(::google::protobuf::int32 value) {
+  set_has_row_count();
+  row_count_ = value;
 }
 
-// optional .exec.shared.RecordBatchDef def = 10;
-inline bool QueryResult::has_def() const {
-  return (_has_bits_[0] & 0x00000200u) != 0;
+// optional .exec.shared.RecordBatchDef def = 3;
+inline bool QueryData::has_def() const {
+  return (_has_bits_[0] & 0x00000004u) != 0;
 }
-inline void QueryResult::set_has_def() {
-  _has_bits_[0] |= 0x00000200u;
+inline void QueryData::set_has_def() {
+  _has_bits_[0] |= 0x00000004u;
 }
-inline void QueryResult::clear_has_def() {
-  _has_bits_[0] &= ~0x00000200u;
+inline void QueryData::clear_has_def() {
+  _has_bits_[0] &= ~0x00000004u;
 }
-inline void QueryResult::clear_def() {
+inline void QueryData::clear_def() {
   if (def_ != NULL) def_->::exec::shared::RecordBatchDef::Clear();
   clear_has_def();
 }
-inline const ::exec::shared::RecordBatchDef& QueryResult::def() const {
+inline const ::exec::shared::RecordBatchDef& QueryData::def() const {
   return def_ != NULL ? *def_ : *default_instance_->def_;
 }
-inline ::exec::shared::RecordBatchDef* QueryResult::mutable_def() {
+inline ::exec::shared::RecordBatchDef* QueryData::mutable_def() {
   set_has_def();
   if (def_ == NULL) def_ = new ::exec::shared::RecordBatchDef;
   return def_;
 }
-inline ::exec::shared::RecordBatchDef* QueryResult::release_def() {
+inline ::exec::shared::RecordBatchDef* QueryData::release_def() {
   clear_has_def();
   ::exec::shared::RecordBatchDef* temp = def_;
   def_ = NULL;
   return temp;
 }
-inline void QueryResult::set_allocated_def(::exec::shared::RecordBatchDef* def) {
+inline void QueryData::set_allocated_def(::exec::shared::RecordBatchDef* def) {
   delete def_;
   def_ = def;
   if (def) {
@@ -4237,28 +4188,6 @@ inline void QueryResult::set_allocated_def(::exec::shared::RecordBatchDef* def)
   }
 }
 
-// optional bool schema_changed = 11;
-inline bool QueryResult::has_schema_changed() const {
-  return (_has_bits_[0] & 0x00000400u) != 0;
-}
-inline void QueryResult::set_has_schema_changed() {
-  _has_bits_[0] |= 0x00000400u;
-}
-inline void QueryResult::clear_has_schema_changed() {
-  _has_bits_[0] &= ~0x00000400u;
-}
-inline void QueryResult::clear_schema_changed() {
-  schema_changed_ = false;
-  clear_has_schema_changed();
-}
-inline bool QueryResult::schema_changed() const {
-  return schema_changed_;
-}
-inline void QueryResult::set_schema_changed(bool value) {
-  set_has_schema_changed();
-  schema_changed_ = value;
-}
-
 // -------------------------------------------------------------------
 
 // QueryInfo
@@ -5271,26 +5200,26 @@ inline void OperatorProfile::set_process_nanos(::google::protobuf::int64 value)
   process_nanos_ = value;
 }
 
-// optional int64 local_memory_allocated = 7;
-inline bool OperatorProfile::has_local_memory_allocated() const {
+// optional int64 peak_local_memory_allocated = 7;
+inline bool OperatorProfile::has_peak_local_memory_allocated() const {
   return (_has_bits_[0] & 0x00000020u) != 0;
 }
-inline void OperatorProfile::set_has_local_memory_allocated() {
+inline void OperatorProfile::set_has_peak_local_memory_allocated() {
   _has_bits_[0] |= 0x00000020u;
 }
-inline void OperatorProfile::clear_has_local_memory_allocated() {
+inline void OperatorProfile::clear_has_peak_local_memory_allocated() {
   _has_bits_[0] &= ~0x00000020u;
 }
-inline void OperatorProfile::clear_local_memory_allocated() {
-  local_memory_allocated_ = GOOGLE_LONGLONG(0);
-  clear_has_local_memory_allocated();
+inline void OperatorProfile::clear_peak_local_memory_allocated() {
+  peak_local_memory_allocated_ = GOOGLE_LONGLONG(0);
+  clear_has_peak_local_memory_allocated();
 }
-inline ::google::protobuf::int64 OperatorProfile::local_memory_allocated() const {
-  return local_memory_allocated_;
+inline ::google::protobuf::int64 OperatorProfile::peak_local_memory_allocated() const {
+  return peak_local_memory_allocated_;
 }
-inline void OperatorProfile::set_local_memory_allocated(::google::protobuf::int64 value) {
-  set_has_local_memory_allocated();
-  local_memory_allocated_ = value;
+inline void OperatorProfile::set_peak_local_memory_allocated(::google::protobuf::int64 value) {
+  set_has_peak_local_memory_allocated();
+  peak_local_memory_allocated_ = value;
 }
 
 // repeated .exec.shared.MetricValue metric = 8;