DRILL-1352: C++ Client. Update the decoding of nullable value vectors to read a byte for every nullable bit.

8 files changed, 516 insertions, 45 deletions

diff --git a/contrib/native/client/src/include/drill/common.hpp b/contrib/native/client/src/include/drill/common.hpp
index dfb04e83f..151d698b5 100644
--- a/contrib/native/client/src/include/drill/common.hpp
+++ b/contrib/native/client/src/include/drill/common.hpp
@@ -33,7 +33,7 @@
 #include <vector>
 #include <boost/shared_ptr.hpp>
 
-#define DRILL_RPC_VERSION 1
+#define DRILL_RPC_VERSION 2
 
 #define LENGTH_PREFIX_MAX_LENGTH 5
 #define LEN_PREFIX_BUFLEN LENGTH_PREFIX_MAX_LENGTH
diff --git a/contrib/native/client/src/include/drill/protobuf/Types.pb.h b/contrib/native/client/src/include/drill/protobuf/Types.pb.h
index 853415317..b26177c44 100644
--- a/contrib/native/client/src/include/drill/protobuf/Types.pb.h
+++ b/contrib/native/client/src/include/drill/protobuf/Types.pb.h
@@ -72,11 +72,12 @@ enum MinorType {
   DM_UNKNOWN = 37,
   INTERVALYEAR = 38,
   INTERVALDAY = 39,
-  LIST = 40
+  LIST = 40,
+  GENERIC_OBJECT = 41
 };
 bool MinorType_IsValid(int value);
 const MinorType MinorType_MIN = LATE;
-const MinorType MinorType_MAX = LIST;
+const MinorType MinorType_MAX = GENERIC_OBJECT;
 const int MinorType_ARRAYSIZE = MinorType_MAX + 1;
 
 const ::google::protobuf::EnumDescriptor* MinorType_descriptor();
diff --git a/contrib/native/client/src/include/drill/recordBatch.hpp b/contrib/native/client/src/include/drill/recordBatch.hpp
index 9a3df2b6f..61db88c8c 100644
--- a/contrib/native/client/src/include/drill/recordBatch.hpp
+++ b/contrib/native/client/src/include/drill/recordBatch.hpp
@@ -386,7 +386,7 @@ template <typename VALUE_TYPE>
 {
     public:
         NullableValueVectorFixed(SlicedByteBuf *b, size_t rowCount):ValueVectorBase(b, rowCount){
-            size_t offsetEnd = (size_t)ceil(rowCount/8.0);
+            size_t offsetEnd = (size_t)rowCount;
             this->m_pBitmap= new SlicedByteBuf(*b, 0, offsetEnd);
             this->m_pData= new SlicedByteBuf(*b, offsetEnd, b->getLength());
             // TODO: testing boundary case(null columns)
@@ -399,7 +399,7 @@ template <typename VALUE_TYPE>
 
         // test whether the value is null in the position index
         bool isNull(size_t index) const {
-            return (m_pBitmap->getBit(index)==0);
+            return (m_pBitmap->getByte(index)==0);
         }
 
         VALUE_TYPE get(size_t index) const {
@@ -584,14 +584,14 @@ template <class VALUEHOLDER_CLASS_TYPE, class VALUE_VECTOR_TYPE>
         public:
 
             NullableValueVectorTyped(SlicedByteBuf *b, size_t rowCount):ValueVectorBase(b, rowCount){
-                size_t offsetEnd = (size_t)ceil(rowCount/8.0);
+                size_t offsetEnd = (size_t)rowCount;
                 this->m_pBitmap= new SlicedByteBuf(*b, 0, offsetEnd);
                 this->m_pData= new SlicedByteBuf(*b, offsetEnd, b->getLength()-offsetEnd);
                 this->m_pVector= new VALUE_VECTOR_TYPE(m_pData, rowCount);
             }
             // Specialized for Decimal Types
             NullableValueVectorTyped(SlicedByteBuf *b, size_t rowCount, int32_t scale):ValueVectorBase(b, rowCount){
-                size_t offsetEnd = (size_t)ceil(rowCount/8.0);
+                size_t offsetEnd = (size_t)rowCount;
                 this->m_pBitmap= new SlicedByteBuf(*b, 0, offsetEnd);
                 this->m_pData= new SlicedByteBuf(*b, offsetEnd, b->getLength()-offsetEnd);
                 this->m_pVector= new VALUE_VECTOR_TYPE(m_pData, rowCount, scale);
@@ -604,7 +604,7 @@ template <class VALUEHOLDER_CLASS_TYPE, class VALUE_VECTOR_TYPE>
             }
 
             bool isNull(size_t index) const{
-                return (m_pBitmap->getBit(index)==0);
+                return (m_pBitmap->getByte(index)==0);
             }
 
             VALUEHOLDER_CLASS_TYPE get(size_t index) const {
diff --git a/contrib/native/client/src/protobuf/BitControl.pb.cc b/contrib/native/client/src/protobuf/BitControl.pb.cc
index 4f8749cee..a7623861e 100644
--- a/contrib/native/client/src/protobuf/BitControl.pb.cc
+++ b/contrib/native/client/src/protobuf/BitControl.pb.cc
@@ -37,6 +37,9 @@ const ::google::protobuf::internal::GeneratedMessageReflection*
 const ::google::protobuf::Descriptor* WorkQueueStatus_descriptor_ = NULL;
 const ::google::protobuf::internal::GeneratedMessageReflection*
   WorkQueueStatus_reflection_ = NULL;
+const ::google::protobuf::Descriptor* FinishedReceiver_descriptor_ = NULL;
+const ::google::protobuf::internal::GeneratedMessageReflection*
+  FinishedReceiver_reflection_ = NULL;
 const ::google::protobuf::EnumDescriptor* RpcType_descriptor_ = NULL;
 
 }  // namespace
@@ -142,6 +145,22 @@ void protobuf_AssignDesc_BitControl_2eproto() {
       ::google::protobuf::DescriptorPool::generated_pool(),
       ::google::protobuf::MessageFactory::generated_factory(),
       sizeof(WorkQueueStatus));
+  FinishedReceiver_descriptor_ = file->message_type(5);
+  static const int FinishedReceiver_offsets_[2] = {
+    GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(FinishedReceiver, receiver_),
+    GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(FinishedReceiver, sender_),
+  };
+  FinishedReceiver_reflection_ =
+    new ::google::protobuf::internal::GeneratedMessageReflection(
+      FinishedReceiver_descriptor_,
+      FinishedReceiver::default_instance_,
+      FinishedReceiver_offsets_,
+      GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(FinishedReceiver, _has_bits_[0]),
+      GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(FinishedReceiver, _unknown_fields_),
+      -1,
+      ::google::protobuf::DescriptorPool::generated_pool(),
+      ::google::protobuf::MessageFactory::generated_factory(),
+      sizeof(FinishedReceiver));
   RpcType_descriptor_ = file->enum_type(0);
 }
 
@@ -165,6 +184,8 @@ void protobuf_RegisterTypes(const ::std::string&) {
     PlanFragment_descriptor_, &PlanFragment::default_instance());
   ::google::protobuf::MessageFactory::InternalRegisterGeneratedMessage(
     WorkQueueStatus_descriptor_, &WorkQueueStatus::default_instance());
+  ::google::protobuf::MessageFactory::InternalRegisterGeneratedMessage(
+    FinishedReceiver_descriptor_, &FinishedReceiver::default_instance());
 }
 
 }  // namespace
@@ -180,6 +201,8 @@ void protobuf_ShutdownFile_BitControl_2eproto() {
   delete PlanFragment_reflection_;
   delete WorkQueueStatus::default_instance_;
   delete WorkQueueStatus_reflection_;
+  delete FinishedReceiver::default_instance_;
+  delete FinishedReceiver_reflection_;
 }
 
 void protobuf_AddDesc_BitControl_2eproto() {
@@ -216,15 +239,18 @@ void protobuf_AddDesc_BitControl_2eproto() {
     "e_zone\030\020 \001(\005\022\024\n\014options_json\030\021 \001(\t\"f\n\017Wo"
     "rkQueueStatus\022(\n\010endpoint\030\001 \001(\0132\026.exec.D"
     "rillbitEndpoint\022\024\n\014queue_length\030\002 \001(\005\022\023\n"
-    "\013report_time\030\003 \001(\003*\207\002\n\007RpcType\022\r\n\tHANDSH"
-    "AKE\020\000\022\007\n\003ACK\020\001\022\013\n\007GOODBYE\020\002\022\033\n\027REQ_INIAT"
-    "ILIZE_FRAGMENT\020\003\022\027\n\023REQ_CANCEL_FRAGMENT\020"
-    "\006\022\027\n\023REQ_FRAGMENT_STATUS\020\007\022\022\n\016REQ_BIT_ST"
-    "ATUS\020\010\022\024\n\020REQ_QUERY_STATUS\020\t\022\030\n\024RESP_FRA"
-    "GMENT_HANDLE\020\n\022\030\n\024RESP_FRAGMENT_STATUS\020\013"
-    "\022\023\n\017RESP_BIT_STATUS\020\014\022\025\n\021RESP_QUERY_STAT"
-    "US\020\rB+\n\033org.apache.drill.exec.protoB\nBit"
-    "ControlH\001", 1289);
+    "\013report_time\030\003 \001(\003\"h\n\020FinishedReceiver\022*"
+    "\n\010receiver\030\001 \001(\0132\030.exec.bit.FragmentHand"
+    "le\022(\n\006sender\030\002 \001(\0132\030.exec.bit.FragmentHa"
+    "ndle*\242\002\n\007RpcType\022\r\n\tHANDSHAKE\020\000\022\007\n\003ACK\020\001"
+    "\022\013\n\007GOODBYE\020\002\022\033\n\027REQ_INIATILIZE_FRAGMENT"
+    "\020\003\022\027\n\023REQ_CANCEL_FRAGMENT\020\006\022\031\n\025REQ_RECEI"
+    "VER_FINISHED\020\007\022\027\n\023REQ_FRAGMENT_STATUS\020\010\022"
+    "\022\n\016REQ_BIT_STATUS\020\t\022\024\n\020REQ_QUERY_STATUS\020"
+    "\n\022\030\n\024RESP_FRAGMENT_HANDLE\020\013\022\030\n\024RESP_FRAG"
+    "MENT_STATUS\020\014\022\023\n\017RESP_BIT_STATUS\020\r\022\025\n\021RE"
+    "SP_QUERY_STATUS\020\016B+\n\033org.apache.drill.ex"
+    "ec.protoB\nBitControlH\001", 1422);
   ::google::protobuf::MessageFactory::InternalRegisterGeneratedFile(
     "BitControl.proto", &protobuf_RegisterTypes);
   BitControlHandshake::default_instance_ = new BitControlHandshake();
@@ -232,11 +258,13 @@ void protobuf_AddDesc_BitControl_2eproto() {
   FragmentStatus::default_instance_ = new FragmentStatus();
   PlanFragment::default_instance_ = new PlanFragment();
   WorkQueueStatus::default_instance_ = new WorkQueueStatus();
+  FinishedReceiver::default_instance_ = new FinishedReceiver();
   BitControlHandshake::default_instance_->InitAsDefaultInstance();
   BitStatus::default_instance_->InitAsDefaultInstance();
   FragmentStatus::default_instance_->InitAsDefaultInstance();
   PlanFragment::default_instance_->InitAsDefaultInstance();
   WorkQueueStatus::default_instance_->InitAsDefaultInstance();
+  FinishedReceiver::default_instance_->InitAsDefaultInstance();
   ::google::protobuf::internal::OnShutdown(&protobuf_ShutdownFile_BitControl_2eproto);
 }
 
@@ -264,6 +292,7 @@ bool RpcType_IsValid(int value) {
     case 11:
     case 12:
     case 13:
+    case 14:
       return true;
     default:
       return false;
@@ -2142,6 +2171,264 @@ void WorkQueueStatus::Swap(WorkQueueStatus* other) {
 }
 
 
+// ===================================================================
+
+#ifndef _MSC_VER
+const int FinishedReceiver::kReceiverFieldNumber;
+const int FinishedReceiver::kSenderFieldNumber;
+#endif  // !_MSC_VER
+
+FinishedReceiver::FinishedReceiver()
+  : ::google::protobuf::Message() {
+  SharedCtor();
+}
+
+void FinishedReceiver::InitAsDefaultInstance() {
+  receiver_ = const_cast< ::exec::bit::FragmentHandle*>(&::exec::bit::FragmentHandle::default_instance());
+  sender_ = const_cast< ::exec::bit::FragmentHandle*>(&::exec::bit::FragmentHandle::default_instance());
+}
+
+FinishedReceiver::FinishedReceiver(const FinishedReceiver& from)
+  : ::google::protobuf::Message() {
+  SharedCtor();
+  MergeFrom(from);
+}
+
+void FinishedReceiver::SharedCtor() {
+  _cached_size_ = 0;
+  receiver_ = NULL;
+  sender_ = NULL;
+  ::memset(_has_bits_, 0, sizeof(_has_bits_));
+}
+
+FinishedReceiver::~FinishedReceiver() {
+  SharedDtor();
+}
+
+void FinishedReceiver::SharedDtor() {
+  if (this != default_instance_) {
+    delete receiver_;
+    delete sender_;
+  }
+}
+
+void FinishedReceiver::SetCachedSize(int size) const {
+  GOOGLE_SAFE_CONCURRENT_WRITES_BEGIN();
+  _cached_size_ = size;
+  GOOGLE_SAFE_CONCURRENT_WRITES_END();
+}
+const ::google::protobuf::Descriptor* FinishedReceiver::descriptor() {
+  protobuf_AssignDescriptorsOnce();
+  return FinishedReceiver_descriptor_;
+}
+
+const FinishedReceiver& FinishedReceiver::default_instance() {
+  if (default_instance_ == NULL) protobuf_AddDesc_BitControl_2eproto();
+  return *default_instance_;
+}
+
+FinishedReceiver* FinishedReceiver::default_instance_ = NULL;
+
+FinishedReceiver* FinishedReceiver::New() const {
+  return new FinishedReceiver;
+}
+
+void FinishedReceiver::Clear() {
+  if (_has_bits_[0 / 32] & (0xffu << (0 % 32))) {
+    if (has_receiver()) {
+      if (receiver_ != NULL) receiver_->::exec::bit::FragmentHandle::Clear();
+    }
+    if (has_sender()) {
+      if (sender_ != NULL) sender_->::exec::bit::FragmentHandle::Clear();
+    }
+  }
+  ::memset(_has_bits_, 0, sizeof(_has_bits_));
+  mutable_unknown_fields()->Clear();
+}
+
+bool FinishedReceiver::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.bit.FragmentHandle receiver = 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_receiver()));
+        } else {
+          goto handle_uninterpreted;
+        }
+        if (input->ExpectTag(18)) goto parse_sender;
+        break;
+      }
+
+      // optional .exec.bit.FragmentHandle sender = 2;
+      case 2: {
+        if (::google::protobuf::internal::WireFormatLite::GetTagWireType(tag) ==
+            ::google::protobuf::internal::WireFormatLite::WIRETYPE_LENGTH_DELIMITED) {
+         parse_sender:
+          DO_(::google::protobuf::internal::WireFormatLite::ReadMessageNoVirtual(
+               input, mutable_sender()));
+        } else {
+          goto handle_uninterpreted;
+        }
+        if (input->ExpectAtEnd()) return true;
+        break;
+      }
+
+      default: {
+      handle_uninterpreted:
+        if (::google::protobuf::internal::WireFormatLite::GetTagWireType(tag) ==
+            ::google::protobuf::internal::WireFormatLite::WIRETYPE_END_GROUP) {
+          return true;
+        }
+        DO_(::google::protobuf::internal::WireFormat::SkipField(
+              input, tag, mutable_unknown_fields()));
+        break;
+      }
+    }
+  }
+  return true;
+#undef DO_
+}
+
+void FinishedReceiver::SerializeWithCachedSizes(
+    ::google::protobuf::io::CodedOutputStream* output) const {
+  // optional .exec.bit.FragmentHandle receiver = 1;
+  if (has_receiver()) {
+    ::google::protobuf::internal::WireFormatLite::WriteMessageMaybeToArray(
+      1, this->receiver(), output);
+  }
+
+  // optional .exec.bit.FragmentHandle sender = 2;
+  if (has_sender()) {
+    ::google::protobuf::internal::WireFormatLite::WriteMessageMaybeToArray(
+      2, this->sender(), output);
+  }
+
+  if (!unknown_fields().empty()) {
+    ::google::protobuf::internal::WireFormat::SerializeUnknownFields(
+        unknown_fields(), output);
+  }
+}
+
+::google::protobuf::uint8* FinishedReceiver::SerializeWithCachedSizesToArray(
+    ::google::protobuf::uint8* target) const {
+  // optional .exec.bit.FragmentHandle receiver = 1;
+  if (has_receiver()) {
+    target = ::google::protobuf::internal::WireFormatLite::
+      WriteMessageNoVirtualToArray(
+        1, this->receiver(), target);
+  }
+
+  // optional .exec.bit.FragmentHandle sender = 2;
+  if (has_sender()) {
+    target = ::google::protobuf::internal::WireFormatLite::
+      WriteMessageNoVirtualToArray(
+        2, this->sender(), target);
+  }
+
+  if (!unknown_fields().empty()) {
+    target = ::google::protobuf::internal::WireFormat::SerializeUnknownFieldsToArray(
+        unknown_fields(), target);
+  }
+  return target;
+}
+
+int FinishedReceiver::ByteSize() const {
+  int total_size = 0;
+
+  if (_has_bits_[0 / 32] & (0xffu << (0 % 32))) {
+    // optional .exec.bit.FragmentHandle receiver = 1;
+    if (has_receiver()) {
+      total_size += 1 +
+        ::google::protobuf::internal::WireFormatLite::MessageSizeNoVirtual(
+          this->receiver());
+    }
+
+    // optional .exec.bit.FragmentHandle sender = 2;
+    if (has_sender()) {
+      total_size += 1 +
+        ::google::protobuf::internal::WireFormatLite::MessageSizeNoVirtual(
+          this->sender());
+    }
+
+  }
+  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 FinishedReceiver::MergeFrom(const ::google::protobuf::Message& from) {
+  GOOGLE_CHECK_NE(&from, this);
+  const FinishedReceiver* source =
+    ::google::protobuf::internal::dynamic_cast_if_available<const FinishedReceiver*>(
+      &from);
+  if (source == NULL) {
+    ::google::protobuf::internal::ReflectionOps::Merge(from, this);
+  } else {
+    MergeFrom(*source);
+  }
+}
+
+void FinishedReceiver::MergeFrom(const FinishedReceiver& from) {
+  GOOGLE_CHECK_NE(&from, this);
+  if (from._has_bits_[0 / 32] & (0xffu << (0 % 32))) {
+    if (from.has_receiver()) {
+      mutable_receiver()->::exec::bit::FragmentHandle::MergeFrom(from.receiver());
+    }
+    if (from.has_sender()) {
+      mutable_sender()->::exec::bit::FragmentHandle::MergeFrom(from.sender());
+    }
+  }
+  mutable_unknown_fields()->MergeFrom(from.unknown_fields());
+}
+
+void FinishedReceiver::CopyFrom(const ::google::protobuf::Message& from) {
+  if (&from == this) return;
+  Clear();
+  MergeFrom(from);
+}
+
+void FinishedReceiver::CopyFrom(const FinishedReceiver& from) {
+  if (&from == this) return;
+  Clear();
+  MergeFrom(from);
+}
+
+bool FinishedReceiver::IsInitialized() const {
+
+  return true;
+}
+
+void FinishedReceiver::Swap(FinishedReceiver* other) {
+  if (other != this) {
+    std::swap(receiver_, other->receiver_);
+    std::swap(sender_, other->sender_);
+    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 FinishedReceiver::GetMetadata() const {
+  protobuf_AssignDescriptorsOnce();
+  ::google::protobuf::Metadata metadata;
+  metadata.descriptor = FinishedReceiver_descriptor_;
+  metadata.reflection = FinishedReceiver_reflection_;
+  return metadata;
+}
+
+
 // @@protoc_insertion_point(namespace_scope)
 
 }  // namespace control
diff --git a/contrib/native/client/src/protobuf/BitControl.pb.h b/contrib/native/client/src/protobuf/BitControl.pb.h
index 06af929d8..011f8a8c9 100644
--- a/contrib/native/client/src/protobuf/BitControl.pb.h
+++ b/contrib/native/client/src/protobuf/BitControl.pb.h
@@ -44,6 +44,7 @@ class BitStatus;
 class FragmentStatus;
 class PlanFragment;
 class WorkQueueStatus;
+class FinishedReceiver;
 
 enum RpcType {
   HANDSHAKE = 0,
@@ -51,13 +52,14 @@ enum RpcType {
   GOODBYE = 2,
   REQ_INIATILIZE_FRAGMENT = 3,
   REQ_CANCEL_FRAGMENT = 6,
-  REQ_FRAGMENT_STATUS = 7,
-  REQ_BIT_STATUS = 8,
-  REQ_QUERY_STATUS = 9,
-  RESP_FRAGMENT_HANDLE = 10,
-  RESP_FRAGMENT_STATUS = 11,
-  RESP_BIT_STATUS = 12,
-  RESP_QUERY_STATUS = 13
+  REQ_RECEIVER_FINISHED = 7,
+  REQ_FRAGMENT_STATUS = 8,
+  REQ_BIT_STATUS = 9,
+  REQ_QUERY_STATUS = 10,
+  RESP_FRAGMENT_HANDLE = 11,
+  RESP_FRAGMENT_STATUS = 12,
+  RESP_BIT_STATUS = 13,
+  RESP_QUERY_STATUS = 14
 };
 bool RpcType_IsValid(int value);
 const RpcType RpcType_MIN = HANDSHAKE;
@@ -582,9 +584,9 @@ class PlanFragment : public ::google::protobuf::Message {
   ::exec::DrillbitEndpoint* assignment_;
   ::exec::DrillbitEndpoint* foreman_;
   ::google::protobuf::int64 mem_initial_;
-  ::google::protobuf::int64 mem_max_;
   bool leaf_fragment_;
   ::google::protobuf::int32 time_zone_;
+  ::google::protobuf::int64 mem_max_;
   ::google::protobuf::int64 query_start_time_;
   ::exec::shared::UserCredentials* credentials_;
   ::std::string* options_json_;
@@ -703,6 +705,102 @@ class WorkQueueStatus : public ::google::protobuf::Message {
   void InitAsDefaultInstance();
   static WorkQueueStatus* default_instance_;
 };
+// -------------------------------------------------------------------
+
+class FinishedReceiver : public ::google::protobuf::Message {
+ public:
+  FinishedReceiver();
+  virtual ~FinishedReceiver();
+
+  FinishedReceiver(const FinishedReceiver& from);
+
+  inline FinishedReceiver& operator=(const FinishedReceiver& 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 FinishedReceiver& default_instance();
+
+  void Swap(FinishedReceiver* other);
+
+  // implements Message ----------------------------------------------
+
+  FinishedReceiver* New() const;
+  void CopyFrom(const ::google::protobuf::Message& from);
+  void MergeFrom(const ::google::protobuf::Message& from);
+  void CopyFrom(const FinishedReceiver& from);
+  void MergeFrom(const FinishedReceiver& 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.bit.FragmentHandle receiver = 1;
+  inline bool has_receiver() const;
+  inline void clear_receiver();
+  static const int kReceiverFieldNumber = 1;
+  inline const ::exec::bit::FragmentHandle& receiver() const;
+  inline ::exec::bit::FragmentHandle* mutable_receiver();
+  inline ::exec::bit::FragmentHandle* release_receiver();
+  inline void set_allocated_receiver(::exec::bit::FragmentHandle* receiver);
+
+  // optional .exec.bit.FragmentHandle sender = 2;
+  inline bool has_sender() const;
+  inline void clear_sender();
+  static const int kSenderFieldNumber = 2;
+  inline const ::exec::bit::FragmentHandle& sender() const;
+  inline ::exec::bit::FragmentHandle* mutable_sender();
+  inline ::exec::bit::FragmentHandle* release_sender();
+  inline void set_allocated_sender(::exec::bit::FragmentHandle* sender);
+
+  // @@protoc_insertion_point(class_scope:exec.bit.control.FinishedReceiver)
+ private:
+  inline void set_has_receiver();
+  inline void clear_has_receiver();
+  inline void set_has_sender();
+  inline void clear_has_sender();
+
+  ::google::protobuf::UnknownFieldSet _unknown_fields_;
+
+  ::exec::bit::FragmentHandle* receiver_;
+  ::exec::bit::FragmentHandle* sender_;
+
+  mutable int _cached_size_;
+  ::google::protobuf::uint32 _has_bits_[(2 + 31) / 32];
+
+  friend void  protobuf_AddDesc_BitControl_2eproto();
+  friend void protobuf_AssignDesc_BitControl_2eproto();
+  friend void protobuf_ShutdownFile_BitControl_2eproto();
+
+  void InitAsDefaultInstance();
+  static FinishedReceiver* default_instance_;
+};
 // ===================================================================
 
 
@@ -1482,6 +1580,86 @@ inline void WorkQueueStatus::set_report_time(::google::protobuf::int64 value) {
   report_time_ = value;
 }
 
+// -------------------------------------------------------------------
+
+// FinishedReceiver
+
+// optional .exec.bit.FragmentHandle receiver = 1;
+inline bool FinishedReceiver::has_receiver() const {
+  return (_has_bits_[0] & 0x00000001u) != 0;
+}
+inline void FinishedReceiver::set_has_receiver() {
+  _has_bits_[0] |= 0x00000001u;
+}
+inline void FinishedReceiver::clear_has_receiver() {
+  _has_bits_[0] &= ~0x00000001u;
+}
+inline void FinishedReceiver::clear_receiver() {
+  if (receiver_ != NULL) receiver_->::exec::bit::FragmentHandle::Clear();
+  clear_has_receiver();
+}
+inline const ::exec::bit::FragmentHandle& FinishedReceiver::receiver() const {
+  return receiver_ != NULL ? *receiver_ : *default_instance_->receiver_;
+}
+inline ::exec::bit::FragmentHandle* FinishedReceiver::mutable_receiver() {
+  set_has_receiver();
+  if (receiver_ == NULL) receiver_ = new ::exec::bit::FragmentHandle;
+  return receiver_;
+}
+inline ::exec::bit::FragmentHandle* FinishedReceiver::release_receiver() {
+  clear_has_receiver();
+  ::exec::bit::FragmentHandle* temp = receiver_;
+  receiver_ = NULL;
+  return temp;
+}
+inline void FinishedReceiver::set_allocated_receiver(::exec::bit::FragmentHandle* receiver) {
+  delete receiver_;
+  receiver_ = receiver;
+  if (receiver) {
+    set_has_receiver();
+  } else {
+    clear_has_receiver();
+  }
+}
+
+// optional .exec.bit.FragmentHandle sender = 2;
+inline bool FinishedReceiver::has_sender() const {
+  return (_has_bits_[0] & 0x00000002u) != 0;
+}
+inline void FinishedReceiver::set_has_sender() {
+  _has_bits_[0] |= 0x00000002u;
+}
+inline void FinishedReceiver::clear_has_sender() {
+  _has_bits_[0] &= ~0x00000002u;
+}
+inline void FinishedReceiver::clear_sender() {
+  if (sender_ != NULL) sender_->::exec::bit::FragmentHandle::Clear();
+  clear_has_sender();
+}
+inline const ::exec::bit::FragmentHandle& FinishedReceiver::sender() const {
+  return sender_ != NULL ? *sender_ : *default_instance_->sender_;
+}
+inline ::exec::bit::FragmentHandle* FinishedReceiver::mutable_sender() {
+  set_has_sender();
+  if (sender_ == NULL) sender_ = new ::exec::bit::FragmentHandle;
+  return sender_;
+}
+inline ::exec::bit::FragmentHandle* FinishedReceiver::release_sender() {
+  clear_has_sender();
+  ::exec::bit::FragmentHandle* temp = sender_;
+  sender_ = NULL;
+  return temp;
+}
+inline void FinishedReceiver::set_allocated_sender(::exec::bit::FragmentHandle* sender) {
+  delete sender_;
+  sender_ = sender;
+  if (sender) {
+    set_has_sender();
+  } else {
+    clear_has_sender();
+  }
+}
+
 
 // @@protoc_insertion_point(namespace_scope)
 
diff --git a/contrib/native/client/src/protobuf/Types.pb.cc b/contrib/native/client/src/protobuf/Types.pb.cc
index c7fa2eaaf..a14f40453 100644
--- a/contrib/native/client/src/protobuf/Types.pb.cc
+++ b/contrib/native/client/src/protobuf/Types.pb.cc
@@ -91,7 +91,7 @@ void protobuf_AddDesc_Types_2eproto() {
     "inor_type\030\001 \001(\0162\021.common.MinorType\022\036\n\004mo"
     "de\030\002 \001(\0162\020.common.DataMode\022\r\n\005width\030\003 \001("
     "\005\022\021\n\tprecision\030\004 \001(\005\022\r\n\005scale\030\005 \001(\005\022\020\n\010t"
-    "imeZone\030\006 \001(\005*\374\003\n\tMinorType\022\010\n\004LATE\020\000\022\007\n"
+    "imeZone\030\006 \001(\005*\220\004\n\tMinorType\022\010\n\004LATE\020\000\022\007\n"
     "\003MAP\020\001\022\013\n\007TINYINT\020\003\022\014\n\010SMALLINT\020\004\022\007\n\003INT"
     "\020\005\022\n\n\006BIGINT\020\006\022\014\n\010DECIMAL9\020\007\022\r\n\tDECIMAL1"
     "8\020\010\022\023\n\017DECIMAL28SPARSE\020\t\022\023\n\017DECIMAL38SPA"
@@ -104,9 +104,10 @@ void protobuf_AddDesc_Types_2eproto() {
     "\036\022\t\n\005UINT4\020\037\022\t\n\005UINT8\020 \022\022\n\016DECIMAL28DENS"
     "E\020!\022\022\n\016DECIMAL38DENSE\020\"\022\016\n\nDM_UNKNOWN\020%\022"
     "\020\n\014INTERVALYEAR\020&\022\017\n\013INTERVALDAY\020\'\022\010\n\004LI"
-    "ST\020(*=\n\010DataMode\022\017\n\013DM_OPTIONAL\020\000\022\017\n\013DM_"
-    "REQUIRED\020\001\022\017\n\013DM_REPEATED\020\002B-\n\035org.apach"
-    "e.drill.common.typesB\nTypeProtosH\001", 794);
+    "ST\020(\022\022\n\016GENERIC_OBJECT\020)*=\n\010DataMode\022\017\n\013"
+    "DM_OPTIONAL\020\000\022\017\n\013DM_REQUIRED\020\001\022\017\n\013DM_REP"
+    "EATED\020\002B-\n\035org.apache.drill.common.types"
+    "B\nTypeProtosH\001", 814);
   ::google::protobuf::MessageFactory::InternalRegisterGeneratedFile(
     "Types.proto", &protobuf_RegisterTypes);
   MajorType::default_instance_ = new MajorType();
@@ -162,6 +163,7 @@ bool MinorType_IsValid(int value) {
     case 38:
     case 39:
     case 40:
+    case 41:
       return true;
     default:
       return false;
diff --git a/contrib/native/client/src/protobuf/UserBitShared.pb.cc b/contrib/native/client/src/protobuf/UserBitShared.pb.cc
index 88507e5a3..0e7649f27 100644
--- a/contrib/native/client/src/protobuf/UserBitShared.pb.cc
+++ b/contrib/native/client/src/protobuf/UserBitShared.pb.cc
@@ -532,24 +532,25 @@ void protobuf_AddDesc_UserBitShared_2eproto() {
     "\020\001\022\013\n\007LOGICAL\020\002\022\014\n\010PHYSICAL\020\003*k\n\rFragmen"
     "tState\022\013\n\007SENDING\020\000\022\027\n\023AWAITING_ALLOCATI"
     "ON\020\001\022\013\n\007RUNNING\020\002\022\014\n\010FINISHED\020\003\022\r\n\tCANCE"
-    "LLED\020\004\022\n\n\006FAILED\020\005*\372\004\n\020CoreOperatorType\022"
+    "LLED\020\004\022\n\n\006FAILED\020\005*\224\005\n\020CoreOperatorType\022"
     "\021\n\rSINGLE_SENDER\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\nMERGE_JOIN\020\005\022\031\n\025HASH_PARTITIO"
     "N_SENDER\020\006\022\t\n\005LIMIT\020\007\022\024\n\020MERGING_RECEIVE"
     "R\020\010\022\034\n\030ORDERED_PARTITION_SENDER\020\t\022\013\n\007PRO"
-    "JECT\020\n\022\023\n\017RANDOM_RECEIVER\020\013\022\020\n\014RANGE_SEN"
-    "DER\020\014\022\n\n\006SCREEN\020\r\022\034\n\030SELECTION_VECTOR_RE"
-    "MOVER\020\016\022\027\n\023STREAMING_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_G"
-    "ROUP_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\016PARQ"
-    "UET_WRITER\020\031\022\023\n\017DIRECT_SUB_SCAN\020\032\022\017\n\013TEX"
-    "T_WRITER\020\033\022\021\n\rTEXT_SUB_SCAN\020\034\022\021\n\rJSON_SU"
-    "B_SCAN\020\035\022\030\n\024INFO_SCHEMA_SUB_SCAN\020\036\022\023\n\017CO"
-    "MPLEX_TO_JSON\020\037B.\n\033org.apache.drill.exec"
-    ".protoB\rUserBitSharedH\001", 3543);
+    "JECT\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\030SELECTION_VECTOR"
+    "_REMOVER\020\016\022\027\n\023STREAMING_AGGREGATE\020\017\022\016\n\nT"
+    "OP_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_RO"
+    "W_GROUP_SCAN\020\025\022\021\n\rHIVE_SUB_SCAN\020\026\022\025\n\021SYS"
+    "TEM_TABLE_SCAN\020\027\022\021\n\rMOCK_SUB_SCAN\020\030\022\022\n\016P"
+    "ARQUET_WRITER\020\031\022\023\n\017DIRECT_SUB_SCAN\020\032\022\017\n\013"
+    "TEXT_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"
+    " B.\n\033org.apache.drill.exec.protoB\rUserBi"
+    "tSharedH\001", 3569);
   ::google::protobuf::MessageFactory::InternalRegisterGeneratedFile(
     "UserBitShared.proto", &protobuf_RegisterTypes);
   UserCredentials::default_instance_ = new UserCredentials();
@@ -677,6 +678,7 @@ bool CoreOperatorType_IsValid(int value) {
     case 29:
     case 30:
     case 31:
+    case 32:
       return true;
     default:
       return false;
diff --git a/contrib/native/client/src/protobuf/UserBitShared.pb.h b/contrib/native/client/src/protobuf/UserBitShared.pb.h
index 6ef17c38f..e7d40b19d 100644
--- a/contrib/native/client/src/protobuf/UserBitShared.pb.h
+++ b/contrib/native/client/src/protobuf/UserBitShared.pb.h
@@ -171,7 +171,7 @@ enum CoreOperatorType {
   MERGING_RECEIVER = 8,
   ORDERED_PARTITION_SENDER = 9,
   PROJECT = 10,
-  RANDOM_RECEIVER = 11,
+  UNORDERED_RECEIVER = 11,
   RANGE_SENDER = 12,
   SCREEN = 13,
   SELECTION_VECTOR_REMOVER = 14,
@@ -191,11 +191,12 @@ enum CoreOperatorType {
   TEXT_SUB_SCAN = 28,
   JSON_SUB_SCAN = 29,
   INFO_SCHEMA_SUB_SCAN = 30,
-  COMPLEX_TO_JSON = 31
+  COMPLEX_TO_JSON = 31,
+  PRODUCER_CONSUMER = 32
 };
 bool CoreOperatorType_IsValid(int value);
 const CoreOperatorType CoreOperatorType_MIN = SINGLE_SENDER;
-const CoreOperatorType CoreOperatorType_MAX = COMPLEX_TO_JSON;
+const CoreOperatorType CoreOperatorType_MAX = PRODUCER_CONSUMER;
 const int CoreOperatorType_ARRAYSIZE = CoreOperatorType_MAX + 1;
 
 const ::google::protobuf::EnumDescriptor* CoreOperatorType_descriptor();