DRILL-6834: Introduce option to disable result set for DDL queries for JDBC connection

- Added session-scoped option `drill.exec.fetch_resultset_for_ddl` to control whether update count or result set should be returned for JDBC connection session. By default the option is set to `true` which ensures that result set is returned;
- Updated Drill JDBC: `DrillCursor` and `DrillStatement` to achieve desired behaviour.

closes #1549

4 files changed, 335 insertions, 113 deletions

diff --git a/protocol/src/main/java/org/apache/drill/exec/proto/SchemaUserBitShared.java b/protocol/src/main/java/org/apache/drill/exec/proto/SchemaUserBitShared.java
index 7b845a493..ba647626d 100644
--- a/protocol/src/main/java/org/apache/drill/exec/proto/SchemaUserBitShared.java
+++ b/protocol/src/main/java/org/apache/drill/exec/proto/SchemaUserBitShared.java
@@ -830,6 +830,8 @@ public final class SchemaUserBitShared
 
                 if(message.hasCarriesTwoByteSelectionVector())
                     output.writeBool(3, message.getCarriesTwoByteSelectionVector(), false);
+                if(message.hasAffectedRowsCount())
+                    output.writeInt32(4, message.getAffectedRowsCount(), false);
             }
             public boolean isInitialized(org.apache.drill.exec.proto.UserBitShared.RecordBatchDef message)
             {
@@ -879,6 +881,9 @@ public final class SchemaUserBitShared
                         case 3:
                             builder.setCarriesTwoByteSelectionVector(input.readBool());
                             break;
+                        case 4:
+                            builder.setAffectedRowsCount(input.readInt32());
+                            break;
                         default:
                             input.handleUnknownField(number, this);
                     }
@@ -922,6 +927,7 @@ public final class SchemaUserBitShared
                 case 1: return "recordCount";
                 case 2: return "field";
                 case 3: return "carriesTwoByteSelectionVector";
+                case 4: return "affectedRowsCount";
                 default: return null;
             }
         }
@@ -936,6 +942,7 @@ public final class SchemaUserBitShared
             fieldMap.put("recordCount", 1);
             fieldMap.put("field", 2);
             fieldMap.put("carriesTwoByteSelectionVector", 3);
+            fieldMap.put("affectedRowsCount", 4);
         }
     }
 
@@ -1484,6 +1491,8 @@ public final class SchemaUserBitShared
                 if(message.hasDef())
                     output.writeObject(3, message.getDef(), org.apache.drill.exec.proto.SchemaUserBitShared.RecordBatchDef.WRITE, false);
 
+                if(message.hasAffectedRowsCount())
+                    output.writeInt32(4, message.getAffectedRowsCount(), false);
             }
             public boolean isInitialized(org.apache.drill.exec.proto.UserBitShared.QueryData message)
             {
@@ -1534,6 +1543,9 @@ public final class SchemaUserBitShared
                             builder.setDef(input.mergeObject(org.apache.drill.exec.proto.UserBitShared.RecordBatchDef.newBuilder(), org.apache.drill.exec.proto.SchemaUserBitShared.RecordBatchDef.MERGE));
 
                             break;
+                        case 4:
+                            builder.setAffectedRowsCount(input.readInt32());
+                            break;
                         default:
                             input.handleUnknownField(number, this);
                     }
@@ -1577,6 +1589,7 @@ public final class SchemaUserBitShared
                 case 1: return "queryId";
                 case 2: return "rowCount";
                 case 3: return "def";
+                case 4: return "affectedRowsCount";
                 default: return null;
             }
         }
@@ -1591,6 +1604,7 @@ public final class SchemaUserBitShared
             fieldMap.put("queryId", 1);
             fieldMap.put("rowCount", 2);
             fieldMap.put("def", 3);
+            fieldMap.put("affectedRowsCount", 4);
         }
     }
 
diff --git a/protocol/src/main/java/org/apache/drill/exec/proto/UserBitShared.java b/protocol/src/main/java/org/apache/drill/exec/proto/UserBitShared.java
index 2f5c3de21..f9696394f 100644
--- a/protocol/src/main/java/org/apache/drill/exec/proto/UserBitShared.java
+++ b/protocol/src/main/java/org/apache/drill/exec/proto/UserBitShared.java
@@ -6796,6 +6796,16 @@ public final class UserBitShared {
      * <code>optional bool carries_two_byte_selection_vector = 3;</code>
      */
     boolean getCarriesTwoByteSelectionVector();
+
+    // optional int32 affected_rows_count = 4;
+    /**
+     * <code>optional int32 affected_rows_count = 4;</code>
+     */
+    boolean hasAffectedRowsCount();
+    /**
+     * <code>optional int32 affected_rows_count = 4;</code>
+     */
+    int getAffectedRowsCount();
   }
   /**
    * Protobuf type {@code exec.shared.RecordBatchDef}
@@ -6866,6 +6876,11 @@ public final class UserBitShared {
               carriesTwoByteSelectionVector_ = input.readBool();
               break;
             }
+            case 32: {
+              bitField0_ |= 0x00000004;
+              affectedRowsCount_ = input.readInt32();
+              break;
+            }
           }
         }
       } catch (com.google.protobuf.InvalidProtocolBufferException e) {
@@ -6977,10 +6992,27 @@ public final class UserBitShared {
       return carriesTwoByteSelectionVector_;
     }
 
+    // optional int32 affected_rows_count = 4;
+    public static final int AFFECTED_ROWS_COUNT_FIELD_NUMBER = 4;
+    private int affectedRowsCount_;
+    /**
+     * <code>optional int32 affected_rows_count = 4;</code>
+     */
+    public boolean hasAffectedRowsCount() {
+      return ((bitField0_ & 0x00000004) == 0x00000004);
+    }
+    /**
+     * <code>optional int32 affected_rows_count = 4;</code>
+     */
+    public int getAffectedRowsCount() {
+      return affectedRowsCount_;
+    }
+
     private void initFields() {
       recordCount_ = 0;
       field_ = java.util.Collections.emptyList();
       carriesTwoByteSelectionVector_ = false;
+      affectedRowsCount_ = 0;
     }
     private byte memoizedIsInitialized = -1;
     public final boolean isInitialized() {
@@ -7003,6 +7035,9 @@ public final class UserBitShared {
       if (((bitField0_ & 0x00000002) == 0x00000002)) {
         output.writeBool(3, carriesTwoByteSelectionVector_);
       }
+      if (((bitField0_ & 0x00000004) == 0x00000004)) {
+        output.writeInt32(4, affectedRowsCount_);
+      }
       getUnknownFields().writeTo(output);
     }
 
@@ -7024,6 +7059,10 @@ public final class UserBitShared {
         size += com.google.protobuf.CodedOutputStream
           .computeBoolSize(3, carriesTwoByteSelectionVector_);
       }
+      if (((bitField0_ & 0x00000004) == 0x00000004)) {
+        size += com.google.protobuf.CodedOutputStream
+          .computeInt32Size(4, affectedRowsCount_);
+      }
       size += getUnknownFields().getSerializedSize();
       memoizedSerializedSize = size;
       return size;
@@ -7151,6 +7190,8 @@ public final class UserBitShared {
         }
         carriesTwoByteSelectionVector_ = false;
         bitField0_ = (bitField0_ & ~0x00000004);
+        affectedRowsCount_ = 0;
+        bitField0_ = (bitField0_ & ~0x00000008);
         return this;
       }
 
@@ -7196,6 +7237,10 @@ public final class UserBitShared {
           to_bitField0_ |= 0x00000002;
         }
         result.carriesTwoByteSelectionVector_ = carriesTwoByteSelectionVector_;
+        if (((from_bitField0_ & 0x00000008) == 0x00000008)) {
+          to_bitField0_ |= 0x00000004;
+        }
+        result.affectedRowsCount_ = affectedRowsCount_;
         result.bitField0_ = to_bitField0_;
         onBuilt();
         return result;
@@ -7244,6 +7289,9 @@ public final class UserBitShared {
         if (other.hasCarriesTwoByteSelectionVector()) {
           setCarriesTwoByteSelectionVector(other.getCarriesTwoByteSelectionVector());
         }
+        if (other.hasAffectedRowsCount()) {
+          setAffectedRowsCount(other.getAffectedRowsCount());
+        }
         this.mergeUnknownFields(other.getUnknownFields());
         return this;
       }
@@ -7577,6 +7625,39 @@ public final class UserBitShared {
         return this;
       }
 
+      // optional int32 affected_rows_count = 4;
+      private int affectedRowsCount_ ;
+      /**
+       * <code>optional int32 affected_rows_count = 4;</code>
+       */
+      public boolean hasAffectedRowsCount() {
+        return ((bitField0_ & 0x00000008) == 0x00000008);
+      }
+      /**
+       * <code>optional int32 affected_rows_count = 4;</code>
+       */
+      public int getAffectedRowsCount() {
+        return affectedRowsCount_;
+      }
+      /**
+       * <code>optional int32 affected_rows_count = 4;</code>
+       */
+      public Builder setAffectedRowsCount(int value) {
+        bitField0_ |= 0x00000008;
+        affectedRowsCount_ = value;
+        onChanged();
+        return this;
+      }
+      /**
+       * <code>optional int32 affected_rows_count = 4;</code>
+       */
+      public Builder clearAffectedRowsCount() {
+        bitField0_ = (bitField0_ & ~0x00000008);
+        affectedRowsCount_ = 0;
+        onChanged();
+        return this;
+      }
+
       // @@protoc_insertion_point(builder_scope:exec.shared.RecordBatchDef)
     }
 
@@ -11586,6 +11667,16 @@ public final class UserBitShared {
      * <code>optional .exec.shared.RecordBatchDef def = 3;</code>
      */
     org.apache.drill.exec.proto.UserBitShared.RecordBatchDefOrBuilder getDefOrBuilder();
+
+    // optional int32 affected_rows_count = 4;
+    /**
+     * <code>optional int32 affected_rows_count = 4;</code>
+     */
+    boolean hasAffectedRowsCount();
+    /**
+     * <code>optional int32 affected_rows_count = 4;</code>
+     */
+    int getAffectedRowsCount();
   }
   /**
    * Protobuf type {@code exec.shared.QueryData}
@@ -11674,6 +11765,11 @@ public final class UserBitShared {
               bitField0_ |= 0x00000004;
               break;
             }
+            case 32: {
+              bitField0_ |= 0x00000008;
+              affectedRowsCount_ = input.readInt32();
+              break;
+            }
           }
         }
       } catch (com.google.protobuf.InvalidProtocolBufferException e) {
@@ -11774,10 +11870,27 @@ public final class UserBitShared {
       return def_;
     }
 
+    // optional int32 affected_rows_count = 4;
+    public static final int AFFECTED_ROWS_COUNT_FIELD_NUMBER = 4;
+    private int affectedRowsCount_;
+    /**
+     * <code>optional int32 affected_rows_count = 4;</code>
+     */
+    public boolean hasAffectedRowsCount() {
+      return ((bitField0_ & 0x00000008) == 0x00000008);
+    }
+    /**
+     * <code>optional int32 affected_rows_count = 4;</code>
+     */
+    public int getAffectedRowsCount() {
+      return affectedRowsCount_;
+    }
+
     private void initFields() {
       queryId_ = org.apache.drill.exec.proto.UserBitShared.QueryId.getDefaultInstance();
       rowCount_ = 0;
       def_ = org.apache.drill.exec.proto.UserBitShared.RecordBatchDef.getDefaultInstance();
+      affectedRowsCount_ = 0;
     }
     private byte memoizedIsInitialized = -1;
     public final boolean isInitialized() {
@@ -11800,6 +11913,9 @@ public final class UserBitShared {
       if (((bitField0_ & 0x00000004) == 0x00000004)) {
         output.writeMessage(3, def_);
       }
+      if (((bitField0_ & 0x00000008) == 0x00000008)) {
+        output.writeInt32(4, affectedRowsCount_);
+      }
       getUnknownFields().writeTo(output);
     }
 
@@ -11821,6 +11937,10 @@ public final class UserBitShared {
         size += com.google.protobuf.CodedOutputStream
           .computeMessageSize(3, def_);
       }
+      if (((bitField0_ & 0x00000008) == 0x00000008)) {
+        size += com.google.protobuf.CodedOutputStream
+          .computeInt32Size(4, affectedRowsCount_);
+      }
       size += getUnknownFields().getSerializedSize();
       memoizedSerializedSize = size;
       return size;
@@ -11958,6 +12078,8 @@ public final class UserBitShared {
           defBuilder_.clear();
         }
         bitField0_ = (bitField0_ & ~0x00000004);
+        affectedRowsCount_ = 0;
+        bitField0_ = (bitField0_ & ~0x00000008);
         return this;
       }
 
@@ -12006,6 +12128,10 @@ public final class UserBitShared {
         } else {
           result.def_ = defBuilder_.build();
         }
+        if (((from_bitField0_ & 0x00000008) == 0x00000008)) {
+          to_bitField0_ |= 0x00000008;
+        }
+        result.affectedRowsCount_ = affectedRowsCount_;
         result.bitField0_ = to_bitField0_;
         onBuilt();
         return result;
@@ -12031,6 +12157,9 @@ public final class UserBitShared {
         if (other.hasDef()) {
           mergeDef(other.getDef());
         }
+        if (other.hasAffectedRowsCount()) {
+          setAffectedRowsCount(other.getAffectedRowsCount());
+        }
         this.mergeUnknownFields(other.getUnknownFields());
         return this;
       }
@@ -12325,6 +12454,39 @@ public final class UserBitShared {
         return defBuilder_;
       }
 
+      // optional int32 affected_rows_count = 4;
+      private int affectedRowsCount_ ;
+      /**
+       * <code>optional int32 affected_rows_count = 4;</code>
+       */
+      public boolean hasAffectedRowsCount() {
+        return ((bitField0_ & 0x00000008) == 0x00000008);
+      }
+      /**
+       * <code>optional int32 affected_rows_count = 4;</code>
+       */
+      public int getAffectedRowsCount() {
+        return affectedRowsCount_;
+      }
+      /**
+       * <code>optional int32 affected_rows_count = 4;</code>
+       */
+      public Builder setAffectedRowsCount(int value) {
+        bitField0_ |= 0x00000008;
+        affectedRowsCount_ = value;
+        onChanged();
+        return this;
+      }
+      /**
+       * <code>optional int32 affected_rows_count = 4;</code>
+       */
+      public Builder clearAffectedRowsCount() {
+        bitField0_ = (bitField0_ & ~0x00000008);
+        affectedRowsCount_ = 0;
+        onChanged();
+        return this;
+      }
+
       // @@protoc_insertion_point(builder_scope:exec.shared.QueryData)
     }
 
@@ -24354,117 +24516,119 @@ public final class UserBitShared {
       "\013method_name\030\004 \001(\t\022\030\n\020is_native_method\030\005" +
       " \001(\010\"\\\n\014ParsingError\022\024\n\014start_column\030\002 \001" +
       "(\005\022\021\n\tstart_row\030\003 \001(\005\022\022\n\nend_column\030\004 \001(" +
-      "\005\022\017\n\007end_row\030\005 \001(\005\"~\n\016RecordBatchDef\022\024\n\014" +
-      "record_count\030\001 \001(\005\022+\n\005field\030\002 \003(\0132\034.exec" +
-      ".shared.SerializedField\022)\n!carries_two_b" +
-      "yte_selection_vector\030\003 \001(\010\"\205\001\n\010NamePart\022",
-      "(\n\004type\030\001 \001(\0162\032.exec.shared.NamePart.Typ" +
-      "e\022\014\n\004name\030\002 \001(\t\022$\n\005child\030\003 \001(\0132\025.exec.sh" +
-      "ared.NamePart\"\033\n\004Type\022\010\n\004NAME\020\000\022\t\n\005ARRAY" +
-      "\020\001\"\324\001\n\017SerializedField\022%\n\nmajor_type\030\001 \001" +
-      "(\0132\021.common.MajorType\022(\n\tname_part\030\002 \001(\013" +
-      "2\025.exec.shared.NamePart\022+\n\005child\030\003 \003(\0132\034" +
-      ".exec.shared.SerializedField\022\023\n\013value_co" +
-      "unt\030\004 \001(\005\022\027\n\017var_byte_length\030\005 \001(\005\022\025\n\rbu" +
-      "ffer_length\030\007 \001(\005\"7\n\nNodeStatus\022\017\n\007node_" +
-      "id\030\001 \001(\005\022\030\n\020memory_footprint\030\002 \001(\003\"\263\002\n\013Q",
-      "ueryResult\0228\n\013query_state\030\001 \001(\0162#.exec.s" +
-      "hared.QueryResult.QueryState\022&\n\010query_id" +
-      "\030\002 \001(\0132\024.exec.shared.QueryId\022(\n\005error\030\003 " +
-      "\003(\0132\031.exec.shared.DrillPBError\"\227\001\n\nQuery" +
-      "State\022\014\n\010STARTING\020\000\022\013\n\007RUNNING\020\001\022\r\n\tCOMP" +
-      "LETED\020\002\022\014\n\010CANCELED\020\003\022\n\n\006FAILED\020\004\022\032\n\026CAN" +
-      "CELLATION_REQUESTED\020\005\022\014\n\010ENQUEUED\020\006\022\r\n\tP" +
-      "REPARING\020\007\022\014\n\010PLANNING\020\010\"p\n\tQueryData\022&\n" +
-      "\010query_id\030\001 \001(\0132\024.exec.shared.QueryId\022\021\n" +
-      "\trow_count\030\002 \001(\005\022(\n\003def\030\003 \001(\0132\033.exec.sha",
-      "red.RecordBatchDef\"\330\001\n\tQueryInfo\022\r\n\005quer" +
-      "y\030\001 \001(\t\022\r\n\005start\030\002 \001(\003\0222\n\005state\030\003 \001(\0162#." +
-      "exec.shared.QueryResult.QueryState\022\017\n\004us" +
-      "er\030\004 \001(\t:\001-\022\'\n\007foreman\030\005 \001(\0132\026.exec.Dril" +
-      "lbitEndpoint\022\024\n\014options_json\030\006 \001(\t\022\022\n\nto" +
-      "tal_cost\030\007 \001(\001\022\025\n\nqueue_name\030\010 \001(\t:\001-\"\263\004" +
-      "\n\014QueryProfile\022 \n\002id\030\001 \001(\0132\024.exec.shared" +
-      ".QueryId\022$\n\004type\030\002 \001(\0162\026.exec.shared.Que" +
-      "ryType\022\r\n\005start\030\003 \001(\003\022\013\n\003end\030\004 \001(\003\022\r\n\005qu" +
-      "ery\030\005 \001(\t\022\014\n\004plan\030\006 \001(\t\022\'\n\007foreman\030\007 \001(\013",
-      "2\026.exec.DrillbitEndpoint\0222\n\005state\030\010 \001(\0162" +
-      "#.exec.shared.QueryResult.QueryState\022\027\n\017" +
-      "total_fragments\030\t \001(\005\022\032\n\022finished_fragme" +
-      "nts\030\n \001(\005\022;\n\020fragment_profile\030\013 \003(\0132!.ex" +
-      "ec.shared.MajorFragmentProfile\022\017\n\004user\030\014" +
-      " \001(\t:\001-\022\r\n\005error\030\r \001(\t\022\024\n\014verboseError\030\016" +
-      " \001(\t\022\020\n\010error_id\030\017 \001(\t\022\022\n\nerror_node\030\020 \001" +
-      "(\t\022\024\n\014options_json\030\021 \001(\t\022\017\n\007planEnd\030\022 \001(" +
-      "\003\022\024\n\014queueWaitEnd\030\023 \001(\003\022\022\n\ntotal_cost\030\024 " +
-      "\001(\001\022\025\n\nqueue_name\030\025 \001(\t:\001-\022\017\n\007queryId\030\026 ",
-      "\001(\t\"t\n\024MajorFragmentProfile\022\031\n\021major_fra" +
-      "gment_id\030\001 \001(\005\022A\n\026minor_fragment_profile" +
-      "\030\002 \003(\0132!.exec.shared.MinorFragmentProfil" +
-      "e\"\350\002\n\024MinorFragmentProfile\022)\n\005state\030\001 \001(" +
-      "\0162\032.exec.shared.FragmentState\022(\n\005error\030\002" +
-      " \001(\0132\031.exec.shared.DrillPBError\022\031\n\021minor" +
-      "_fragment_id\030\003 \001(\005\0226\n\020operator_profile\030\004" +
-      " \003(\0132\034.exec.shared.OperatorProfile\022\022\n\nst" +
-      "art_time\030\005 \001(\003\022\020\n\010end_time\030\006 \001(\003\022\023\n\013memo" +
-      "ry_used\030\007 \001(\003\022\027\n\017max_memory_used\030\010 \001(\003\022(",
-      "\n\010endpoint\030\t \001(\0132\026.exec.DrillbitEndpoint" +
-      "\022\023\n\013last_update\030\n \001(\003\022\025\n\rlast_progress\030\013" +
-      " \001(\003\"\377\001\n\017OperatorProfile\0221\n\rinput_profil" +
-      "e\030\001 \003(\0132\032.exec.shared.StreamProfile\022\023\n\013o" +
-      "perator_id\030\003 \001(\005\022\025\n\roperator_type\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_allocated\030\007 \001(\003" +
-      "\022(\n\006metric\030\010 \003(\0132\030.exec.shared.MetricVal" +
-      "ue\022\022\n\nwait_nanos\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\007sch",
-      "emas\030\003 \001(\003\"J\n\013MetricValue\022\021\n\tmetric_id\030\001" +
-      " \001(\005\022\022\n\nlong_value\030\002 \001(\003\022\024\n\014double_value" +
-      "\030\003 \001(\001\")\n\010Registry\022\035\n\003jar\030\001 \003(\0132\020.exec.s" +
-      "hared.Jar\"/\n\003Jar\022\014\n\004name\030\001 \001(\t\022\032\n\022functi" +
-      "on_signature\030\002 \003(\t\"W\n\013SaslMessage\022\021\n\tmec" +
-      "hanism\030\001 \001(\t\022\014\n\004data\030\002 \001(\014\022\'\n\006status\030\003 \001" +
-      "(\0162\027.exec.shared.SaslStatus*5\n\nRpcChanne" +
-      "l\022\017\n\013BIT_CONTROL\020\000\022\014\n\010BIT_DATA\020\001\022\010\n\004USER" +
-      "\020\002*V\n\tQueryType\022\007\n\003SQL\020\001\022\013\n\007LOGICAL\020\002\022\014\n" +
-      "\010PHYSICAL\020\003\022\r\n\tEXECUTION\020\004\022\026\n\022PREPARED_S",
-      "TATEMENT\020\005*\207\001\n\rFragmentState\022\013\n\007SENDING\020" +
-      "\000\022\027\n\023AWAITING_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" +
-      "\022\032\n\026CANCELLATION_REQUESTED\020\006*\222\t\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\032\n\026RANGE_PARTITION_SENDER\020\014\022\n\n\006SCREEN\020\r" +
-      "\022\034\n\030SELECTION_VECTOR_REMOVER\020\016\022\027\n\023STREAM" +
-      "ING_AGGREGATE\020\017\022\016\n\nTOP_N_SORT\020\020\022\021\n\rEXTER" +
-      "NAL_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\rHI" +
-      "VE_SUB_SCAN\020\026\022\025\n\021SYSTEM_TABLE_SCAN\020\027\022\021\n\r" +
-      "MOCK_SUB_SCAN\020\030\022\022\n\016PARQUET_WRITER\020\031\022\023\n\017D" +
-      "IRECT_SUB_SCAN\020\032\022\017\n\013TEXT_WRITER\020\033\022\021\n\rTEX" +
-      "T_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\006WINDOW\020\"\022\024\n\020NESTED_LOOP_JOIN\020#\022\021\n\rA" +
-      "VRO_SUB_SCAN\020$\022\021\n\rPCAP_SUB_SCAN\020%\022\022\n\016KAF" +
-      "KA_SUB_SCAN\020&\022\021\n\rKUDU_SUB_SCAN\020\'\022\013\n\007FLAT" +
-      "TEN\020(\022\020\n\014LATERAL_JOIN\020)\022\n\n\006UNNEST\020*\022,\n(H" +
-      "IVE_DRILL_NATIVE_PARQUET_ROW_GROUP_SCAN\020" +
-      "+\022\r\n\tJDBC_SCAN\020,\022\022\n\016REGEX_SUB_SCAN\020-\022\023\n\017" +
-      "MAPRDB_SUB_SCAN\020.\022\022\n\016MONGO_SUB_SCAN\020/\022\017\n" +
-      "\013KUDU_WRITER\0200\022\026\n\022OPEN_TSDB_SUB_SCAN\0201\022\017" +
-      "\n\013JSON_WRITER\0202\022\026\n\022HTPPD_LOG_SUB_SCAN\0203\022",
-      "\022\n\016IMAGE_SUB_SCAN\0204\022\025\n\021SEQUENCE_SUB_SCAN" +
-      "\0205\022\023\n\017PARTITION_LIMIT\0206\022\023\n\017PCAPNG_SUB_SC" +
-      "AN\0207\022\022\n\016RUNTIME_FILTER\0208\022\017\n\013ROWKEY_JOIN\020" +
-      "9*g\n\nSaslStatus\022\020\n\014SASL_UNKNOWN\020\000\022\016\n\nSAS" +
-      "L_START\020\001\022\024\n\020SASL_IN_PROGRESS\020\002\022\020\n\014SASL_" +
-      "SUCCESS\020\003\022\017\n\013SASL_FAILED\020\004B.\n\033org.apache" +
-      ".drill.exec.protoB\rUserBitSharedH\001"
+      "\005\022\017\n\007end_row\030\005 \001(\005\"\233\001\n\016RecordBatchDef\022\024\n" +
+      "\014record_count\030\001 \001(\005\022+\n\005field\030\002 \003(\0132\034.exe" +
+      "c.shared.SerializedField\022)\n!carries_two_" +
+      "byte_selection_vector\030\003 \001(\010\022\033\n\023affected_",
+      "rows_count\030\004 \001(\005\"\205\001\n\010NamePart\022(\n\004type\030\001 " +
+      "\001(\0162\032.exec.shared.NamePart.Type\022\014\n\004name\030" +
+      "\002 \001(\t\022$\n\005child\030\003 \001(\0132\025.exec.shared.NameP" +
+      "art\"\033\n\004Type\022\010\n\004NAME\020\000\022\t\n\005ARRAY\020\001\"\324\001\n\017Ser" +
+      "ializedField\022%\n\nmajor_type\030\001 \001(\0132\021.commo" +
+      "n.MajorType\022(\n\tname_part\030\002 \001(\0132\025.exec.sh" +
+      "ared.NamePart\022+\n\005child\030\003 \003(\0132\034.exec.shar" +
+      "ed.SerializedField\022\023\n\013value_count\030\004 \001(\005\022" +
+      "\027\n\017var_byte_length\030\005 \001(\005\022\025\n\rbuffer_lengt" +
+      "h\030\007 \001(\005\"7\n\nNodeStatus\022\017\n\007node_id\030\001 \001(\005\022\030",
+      "\n\020memory_footprint\030\002 \001(\003\"\263\002\n\013QueryResult" +
+      "\0228\n\013query_state\030\001 \001(\0162#.exec.shared.Quer" +
+      "yResult.QueryState\022&\n\010query_id\030\002 \001(\0132\024.e" +
+      "xec.shared.QueryId\022(\n\005error\030\003 \003(\0132\031.exec" +
+      ".shared.DrillPBError\"\227\001\n\nQueryState\022\014\n\010S" +
+      "TARTING\020\000\022\013\n\007RUNNING\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\022\014\n\010ENQUEUED\020\006\022\r\n\tPREPARING\020\007" +
+      "\022\014\n\010PLANNING\020\010\"\215\001\n\tQueryData\022&\n\010query_id" +
+      "\030\001 \001(\0132\024.exec.shared.QueryId\022\021\n\trow_coun",
+      "t\030\002 \001(\005\022(\n\003def\030\003 \001(\0132\033.exec.shared.Recor" +
+      "dBatchDef\022\033\n\023affected_rows_count\030\004 \001(\005\"\330" +
+      "\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.QueryRes" +
+      "ult.QueryState\022\017\n\004user\030\004 \001(\t:\001-\022\'\n\007forem" +
+      "an\030\005 \001(\0132\026.exec.DrillbitEndpoint\022\024\n\014opti" +
+      "ons_json\030\006 \001(\t\022\022\n\ntotal_cost\030\007 \001(\001\022\025\n\nqu" +
+      "eue_name\030\010 \001(\t:\001-\"\263\004\n\014QueryProfile\022 \n\002id" +
+      "\030\001 \001(\0132\024.exec.shared.QueryId\022$\n\004type\030\002 \001" +
+      "(\0162\026.exec.shared.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\007foreman\030\007 \001(\0132\026.exec.DrillbitEndp" +
+      "oint\0222\n\005state\030\010 \001(\0162#.exec.shared.QueryR" +
+      "esult.QueryState\022\027\n\017total_fragments\030\t \001(" +
+      "\005\022\032\n\022finished_fragments\030\n \001(\005\022;\n\020fragmen" +
+      "t_profile\030\013 \003(\0132!.exec.shared.MajorFragm" +
+      "entProfile\022\017\n\004user\030\014 \001(\t:\001-\022\r\n\005error\030\r \001" +
+      "(\t\022\024\n\014verboseError\030\016 \001(\t\022\020\n\010error_id\030\017 \001" +
+      "(\t\022\022\n\nerror_node\030\020 \001(\t\022\024\n\014options_json\030\021" +
+      " \001(\t\022\017\n\007planEnd\030\022 \001(\003\022\024\n\014queueWaitEnd\030\023 ",
+      "\001(\003\022\022\n\ntotal_cost\030\024 \001(\001\022\025\n\nqueue_name\030\025 " +
+      "\001(\t:\001-\022\017\n\007queryId\030\026 \001(\t\"t\n\024MajorFragment" +
+      "Profile\022\031\n\021major_fragment_id\030\001 \001(\005\022A\n\026mi" +
+      "nor_fragment_profile\030\002 \003(\0132!.exec.shared" +
+      ".MinorFragmentProfile\"\350\002\n\024MinorFragmentP" +
+      "rofile\022)\n\005state\030\001 \001(\0162\032.exec.shared.Frag" +
+      "mentState\022(\n\005error\030\002 \001(\0132\031.exec.shared.D" +
+      "rillPBError\022\031\n\021minor_fragment_id\030\003 \001(\005\0226" +
+      "\n\020operator_profile\030\004 \003(\0132\034.exec.shared.O" +
+      "peratorProfile\022\022\n\nstart_time\030\005 \001(\003\022\020\n\010en",
+      "d_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.e" +
+      "xec.DrillbitEndpoint\022\023\n\013last_update\030\n \001(" +
+      "\003\022\025\n\rlast_progress\030\013 \001(\003\"\377\001\n\017OperatorPro" +
+      "file\0221\n\rinput_profile\030\001 \003(\0132\032.exec.share" +
+      "d.StreamProfile\022\023\n\013operator_id\030\003 \001(\005\022\025\n\r" +
+      "operator_type\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_me" +
+      "mory_allocated\030\007 \001(\003\022(\n\006metric\030\010 \003(\0132\030.e" +
+      "xec.shared.MetricValue\022\022\n\nwait_nanos\030\t \001",
+      "(\003\"B\n\rStreamProfile\022\017\n\007records\030\001 \001(\003\022\017\n\007" +
+      "batches\030\002 \001(\003\022\017\n\007schemas\030\003 \001(\003\"J\n\013Metric" +
+      "Value\022\021\n\tmetric_id\030\001 \001(\005\022\022\n\nlong_value\030\002" +
+      " \001(\003\022\024\n\014double_value\030\003 \001(\001\")\n\010Registry\022\035" +
+      "\n\003jar\030\001 \003(\0132\020.exec.shared.Jar\"/\n\003Jar\022\014\n\004" +
+      "name\030\001 \001(\t\022\032\n\022function_signature\030\002 \003(\t\"W" +
+      "\n\013SaslMessage\022\021\n\tmechanism\030\001 \001(\t\022\014\n\004data" +
+      "\030\002 \001(\014\022\'\n\006status\030\003 \001(\0162\027.exec.shared.Sas" +
+      "lStatus*5\n\nRpcChannel\022\017\n\013BIT_CONTROL\020\000\022\014" +
+      "\n\010BIT_DATA\020\001\022\010\n\004USER\020\002*V\n\tQueryType\022\007\n\003S",
+      "QL\020\001\022\013\n\007LOGICAL\020\002\022\014\n\010PHYSICAL\020\003\022\r\n\tEXECU" +
+      "TION\020\004\022\026\n\022PREPARED_STATEMENT\020\005*\207\001\n\rFragm" +
+      "entState\022\013\n\007SENDING\020\000\022\027\n\023AWAITING_ALLOCA" +
+      "TION\020\001\022\013\n\007RUNNING\020\002\022\014\n\010FINISHED\020\003\022\r\n\tCAN" +
+      "CELLED\020\004\022\n\n\006FAILED\020\005\022\032\n\026CANCELLATION_REQ" +
+      "UESTED\020\006*\222\t\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_PARTITION_SENDER\020\006" +
+      "\022\t\n\005LIMIT\020\007\022\024\n\020MERGING_RECEIVER\020\010\022\034\n\030ORD",
+      "ERED_PARTITION_SENDER\020\t\022\013\n\007PROJECT\020\n\022\026\n\022" +
+      "UNORDERED_RECEIVER\020\013\022\032\n\026RANGE_PARTITION_" +
+      "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" +
+      " \022\022\n\016HBASE_SUB_SCAN\020!\022\n\n\006WINDOW\020\"\022\024\n\020NES" +
+      "TED_LOOP_JOIN\020#\022\021\n\rAVRO_SUB_SCAN\020$\022\021\n\rPC" +
+      "AP_SUB_SCAN\020%\022\022\n\016KAFKA_SUB_SCAN\020&\022\021\n\rKUD" +
+      "U_SUB_SCAN\020\'\022\013\n\007FLATTEN\020(\022\020\n\014LATERAL_JOI" +
+      "N\020)\022\n\n\006UNNEST\020*\022,\n(HIVE_DRILL_NATIVE_PAR" +
+      "QUET_ROW_GROUP_SCAN\020+\022\r\n\tJDBC_SCAN\020,\022\022\n\016" +
+      "REGEX_SUB_SCAN\020-\022\023\n\017MAPRDB_SUB_SCAN\020.\022\022\n" +
+      "\016MONGO_SUB_SCAN\020/\022\017\n\013KUDU_WRITER\0200\022\026\n\022OP",
+      "EN_TSDB_SUB_SCAN\0201\022\017\n\013JSON_WRITER\0202\022\026\n\022H" +
+      "TPPD_LOG_SUB_SCAN\0203\022\022\n\016IMAGE_SUB_SCAN\0204\022" +
+      "\025\n\021SEQUENCE_SUB_SCAN\0205\022\023\n\017PARTITION_LIMI" +
+      "T\0206\022\023\n\017PCAPNG_SUB_SCAN\0207\022\022\n\016RUNTIME_FILT" +
+      "ER\0208\022\017\n\013ROWKEY_JOIN\0209*g\n\nSaslStatus\022\020\n\014S" +
+      "ASL_UNKNOWN\020\000\022\016\n\nSASL_START\020\001\022\024\n\020SASL_IN" +
+      "_PROGRESS\020\002\022\020\n\014SASL_SUCCESS\020\003\022\017\n\013SASL_FA" +
+      "ILED\020\004B.\n\033org.apache.drill.exec.protoB\rU" +
+      "serBitSharedH\001"
     };
     com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
       new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() {
@@ -24512,7 +24676,7 @@ public final class UserBitShared {
           internal_static_exec_shared_RecordBatchDef_fieldAccessorTable = new
             com.google.protobuf.GeneratedMessage.FieldAccessorTable(
               internal_static_exec_shared_RecordBatchDef_descriptor,
-              new java.lang.String[] { "RecordCount", "Field", "CarriesTwoByteSelectionVector", });
+              new java.lang.String[] { "RecordCount", "Field", "CarriesTwoByteSelectionVector", "AffectedRowsCount", });
           internal_static_exec_shared_NamePart_descriptor =
             getDescriptor().getMessageTypes().get(7);
           internal_static_exec_shared_NamePart_fieldAccessorTable = new
@@ -24542,7 +24706,7 @@ public final class UserBitShared {
           internal_static_exec_shared_QueryData_fieldAccessorTable = new
             com.google.protobuf.GeneratedMessage.FieldAccessorTable(
               internal_static_exec_shared_QueryData_descriptor,
-              new java.lang.String[] { "QueryId", "RowCount", "Def", });
+              new java.lang.String[] { "QueryId", "RowCount", "Def", "AffectedRowsCount", });
           internal_static_exec_shared_QueryInfo_descriptor =
             getDescriptor().getMessageTypes().get(12);
           internal_static_exec_shared_QueryInfo_fieldAccessorTable = new
diff --git a/protocol/src/main/java/org/apache/drill/exec/proto/beans/QueryData.java b/protocol/src/main/java/org/apache/drill/exec/proto/beans/QueryData.java
index 36f27477b..774484856 100644
--- a/protocol/src/main/java/org/apache/drill/exec/proto/beans/QueryData.java
+++ b/protocol/src/main/java/org/apache/drill/exec/proto/beans/QueryData.java
@@ -50,6 +50,7 @@ public final class QueryData implements Externalizable, Message<QueryData>, Sche
     private QueryId queryId;
     private int rowCount;
     private RecordBatchDef def;
+    private int affectedRowsCount;
 
     public QueryData()
     {
@@ -97,6 +98,19 @@ public final class QueryData implements Externalizable, Message<QueryData>, Sche
         return this;
     }
 
+    // affectedRowsCount
+
+    public int getAffectedRowsCount()
+    {
+        return affectedRowsCount;
+    }
+
+    public QueryData setAffectedRowsCount(int affectedRowsCount)
+    {
+        this.affectedRowsCount = affectedRowsCount;
+        return this;
+    }
+
     // java serialization
 
     public void readExternal(ObjectInput in) throws IOException
@@ -162,6 +176,9 @@ public final class QueryData implements Externalizable, Message<QueryData>, Sche
                     message.def = input.mergeObject(message.def, RecordBatchDef.getSchema());
                     break;
 
+                case 4:
+                    message.affectedRowsCount = input.readInt32();
+                    break;
                 default:
                     input.handleUnknownField(number, this);
             }   
@@ -181,6 +198,9 @@ public final class QueryData implements Externalizable, Message<QueryData>, Sche
         if(message.def != null)
              output.writeObject(3, message.def, RecordBatchDef.getSchema(), false);
 
+
+        if(message.affectedRowsCount != 0)
+            output.writeInt32(4, message.affectedRowsCount, false);
     }
 
     public String getFieldName(int number)
@@ -190,6 +210,7 @@ public final class QueryData implements Externalizable, Message<QueryData>, Sche
             case 1: return "queryId";
             case 2: return "rowCount";
             case 3: return "def";
+            case 4: return "affectedRowsCount";
             default: return null;
         }
     }
@@ -206,6 +227,7 @@ public final class QueryData implements Externalizable, Message<QueryData>, Sche
         __fieldMap.put("queryId", 1);
         __fieldMap.put("rowCount", 2);
         __fieldMap.put("def", 3);
+        __fieldMap.put("affectedRowsCount", 4);
     }
     
 }
diff --git a/protocol/src/main/java/org/apache/drill/exec/proto/beans/RecordBatchDef.java b/protocol/src/main/java/org/apache/drill/exec/proto/beans/RecordBatchDef.java
index 53bfc91f5..83760133c 100644
--- a/protocol/src/main/java/org/apache/drill/exec/proto/beans/RecordBatchDef.java
+++ b/protocol/src/main/java/org/apache/drill/exec/proto/beans/RecordBatchDef.java
@@ -52,6 +52,7 @@ public final class RecordBatchDef implements Externalizable, Message<RecordBatch
     private int recordCount;
     private List<SerializedField> field;
     private Boolean carriesTwoByteSelectionVector;
+    private int affectedRowsCount;
 
     public RecordBatchDef()
     {
@@ -99,6 +100,19 @@ public final class RecordBatchDef implements Externalizable, Message<RecordBatch
         return this;
     }
 
+    // affectedRowsCount
+
+    public int getAffectedRowsCount()
+    {
+        return affectedRowsCount;
+    }
+
+    public RecordBatchDef setAffectedRowsCount(int affectedRowsCount)
+    {
+        this.affectedRowsCount = affectedRowsCount;
+        return this;
+    }
+
     // java serialization
 
     public void readExternal(ObjectInput in) throws IOException
@@ -165,6 +179,9 @@ public final class RecordBatchDef implements Externalizable, Message<RecordBatch
                 case 3:
                     message.carriesTwoByteSelectionVector = input.readBool();
                     break;
+                case 4:
+                    message.affectedRowsCount = input.readInt32();
+                    break;
                 default:
                     input.handleUnknownField(number, this);
             }   
@@ -189,6 +206,9 @@ public final class RecordBatchDef implements Externalizable, Message<RecordBatch
 
         if(message.carriesTwoByteSelectionVector != null)
             output.writeBool(3, message.carriesTwoByteSelectionVector, false);
+
+        if(message.affectedRowsCount != 0)
+            output.writeInt32(4, message.affectedRowsCount, false);
     }
 
     public String getFieldName(int number)
@@ -198,6 +218,7 @@ public final class RecordBatchDef implements Externalizable, Message<RecordBatch
             case 1: return "recordCount";
             case 2: return "field";
             case 3: return "carriesTwoByteSelectionVector";
+            case 4: return "affectedRowsCount";
             default: return null;
         }
     }
@@ -214,6 +235,7 @@ public final class RecordBatchDef implements Externalizable, Message<RecordBatch
         __fieldMap.put("recordCount", 1);
         __fieldMap.put("field", 2);
         __fieldMap.put("carriesTwoByteSelectionVector", 3);
+        __fieldMap.put("affectedRowsCount", 4);
     }
     
 }