fix bug count(distinct field) (#87)

lgqss · liguoqiang · web-flow · commit 658ff42854ef · 2020-06-02T19:33:49.000+08:00
* fix bug count(distinct field)

* braft update to v1.1.1

* brpc update to 0.9.7

* fix bug mem row

Co-authored-by: liguoqiang &lt;liguoqiang@MacintoshdeMacBook-Pro.local&gt;
diff --git a/WORKSPACE b/WORKSPACE
@@ -137,7 +137,7 @@ bind(
 git_repository(
     name = "com_github_brpc_braft",
     remote = "https://github.com/baidu/braft.git",
-    tag = "v1.0.2",
+    tag = "v1.1.1",
 )
 
 bind(
@@ -148,7 +148,7 @@ bind(
 git_repository(
     name = "com_github_brpc_brpc",
     remote= "https://github.com/apache/incubator-brpc.git",
-    tag = "0.9.7-rc01",
+    tag = "0.9.7",
 )
 
 bind(
diff --git a/include/common/table_key.h b/include/common/table_key.h
@@ -155,6 +155,8 @@ using Reflection = google::protobuf::Reflection;
             const FieldInfo& field_info,
             int& pos) const;
 
+    int skip_field(const FieldInfo& field_info, int& pos) const;
+
 private:
     bool             _full;  //full key or just a prefix
     rocksdb::Slice   _data;
diff --git a/include/raft/my_raft_log_storage.h b/include/raft/my_raft_log_storage.h
@@ -97,10 +97,7 @@ class MyRaftLogStorage : public braft::LogStorage {
 
     // append entries to log, return append success number
     int append_entries(const std::vector<braft::LogEntry*>& entries 
-#ifdef BAIDU_INTERNAL
-            , braft::IOMetric* metric
-#endif
-            ) override;
+            , braft::IOMetric* metric) override;
 
     // delete logs from storage's head, [first_log_index, first_index_kept) will be discarded
     int truncate_prefix(const int64_t first_index_kept) override;
diff --git a/include/raft/my_raft_meta_storage.h b/include/raft/my_raft_meta_storage.h
@@ -66,7 +66,6 @@ class MyRaftMetaStorage : public RaftMetaStorage {
 
     // set term and peer_id
     virtual int set_term_and_votedfor(const int64_t term, const braft::PeerId& peer_id);
-#ifdef BAIDU_INTERNAL
     // init stable storage
     virtual butil::Status init();
     // set term and votedfor information
@@ -75,10 +74,6 @@ class MyRaftMetaStorage : public RaftMetaStorage {
     // get term and votedfor information
     virtual butil::Status get_term_and_votedfor(int64_t* term, braft::PeerId* peer_id, 
                                                    const braft::VersionedGroupId& group);
-#else
-    // init stable storage, check consistency and integrity
-    virtual int init();
-#endif
 
     RaftMetaStorage* new_instance(const std::string& uri) const override;
 
diff --git a/src/common/table_key.cpp b/src/common/table_key.cpp
@@ -144,4 +144,60 @@ int TableKey::decode_field(Message* message,
     return 0;
 }
 
+//TODO: secondary key
+int TableKey::skip_field(const FieldInfo& field_info, int& pos) const {
+    switch (field_info.type) {
+        case pb::INT8: {
+            pos += sizeof(int8_t);
+        } break;
+        case pb::INT16: {
+            pos += sizeof(int16_t);
+        } break;
+        case pb::TIME:
+        case pb::INT32: {
+            pos += sizeof(int32_t);
+        } break;
+        case pb::INT64: {
+            pos += sizeof(int64_t);
+        } break;
+        case pb::UINT8: {
+            pos += sizeof(uint8_t);
+        } break;
+        case pb::UINT16: {
+            pos += sizeof(uint16_t);
+        } break;
+        case pb::TIMESTAMP:
+        case pb::DATE:
+        case pb::UINT32: {
+            pos += sizeof(uint32_t);
+        } break;
+        case pb::DATETIME:
+        case pb::UINT64: {
+            pos += sizeof(uint64_t);
+        } break;
+        case pb::FLOAT: {
+            pos += sizeof(float);
+        } break;
+        case pb::DOUBLE: {
+            pos += sizeof(double);
+        } break;
+        case pb::BOOL: {
+            pos += sizeof(uint8_t);
+        } break;
+        case pb::STRING: {
+            //TODO no string pk-field is supported
+            if (pos >= (int)size()) {
+                DB_WARNING("string pos out of bound: %d %zu", pos, size());
+                return -2;
+            }
+            pos += (strlen(_data.data_) + 1);
+        } break;
+        default: {
+            DB_WARNING("un-supported field type: %d", field_info.type);
+            return -1;
+        } break;
+    }
+    return 0;
+}
+
 } // end of namespace baikaldb
diff --git a/src/common/tuple_record.cpp b/src/common/tuple_record.cpp
@@ -26,6 +26,7 @@ inline const google::protobuf::FieldDescriptor* get_field(
     } else {
         int32_t slot = (*field_slot)[field_info->id];
         if (slot == 0) {
+            DB_FATAL("field:%d, slot is 0, can't be here", field_info->id);
             return nullptr;
         }
         return (*mem_row)->get_field_by_slot(tuple_id, slot);
diff --git a/src/expr/agg_fn_call.cpp b/src/expr/agg_fn_call.cpp
@@ -173,6 +173,10 @@ struct AvgIntermediate {
 };
 
 bool AggFnCall::is_initialize(MemRow* dst) {
+    if (_is_distinct) {
+        return false;
+    }
+
     if (dst->get_value(_tuple_id, _intermediate_slot_id).is_null()) {
         return true;
     }
diff --git a/src/mem_row/mem_row.cpp b/src/mem_row/mem_row.cpp
@@ -80,23 +80,28 @@ int MemRow::decode_key(int32_t tuple_id, IndexInfo& index,
         pos += sizeof(uint8_t);
     }
     for (uint32_t idx = 0; idx < index.fields.size(); ++idx) {
+        // DB_WARNING("null_flag: %ld, %u, %d, %d, %s", 
+        //     index.id, null_flag, pos, index.fields[idx].can_null, 
+        //     key.data().ToString(true).c_str());
+        if (((null_flag >> (7 - idx)) & 0x01) && index.fields[idx].can_null) {
+            //DB_DEBUG("field is null: %d", idx);
+            continue;
+        }
         int32_t slot = field_slot[index.fields[idx].id];
         //说明不需要解析
+        //pos需要更新，容易出bug
         if (slot == 0) {
+            if (0 != key.skip_field(index.fields[idx], pos)) {
+                DB_WARNING("skip index field error");
+                return -1;
+            }
             continue;
         }
         const FieldDescriptor* field = descriptor->field(slot - 1);
         if (field == nullptr) {
             DB_WARNING("invalid field: %d slot: %d", index.fields[idx].id, slot);
             return -1;
         }
-        // DB_WARNING("null_flag: %ld, %u, %d, %d, %s", 
-        //     index.id, null_flag, pos, index.fields[idx].can_null, 
-        //     key.data().ToString(true).c_str());
-        if (((null_flag >> (7 - idx)) & 0x01) && index.fields[idx].can_null) {
-            //DB_DEBUG("field is null: %d", idx);
-            continue;
-        }
         if (0 != key.decode_field(tuple, reflection, field, index.fields[idx], pos)) {
             DB_WARNING("decode index field error");
             return -1;
@@ -121,7 +126,12 @@ int MemRow::decode_primary_key(int32_t tuple_id, IndexInfo& index, std::vector<i
     for (uint32_t idx = 0; idx < index.pk_fields.size(); ++idx) {
         int32_t slot = field_slot[index.pk_fields[idx].id];
         //说明不需要解析
+        //pos需要更新，容易出bug
         if (slot == 0) {
+            if (0 != key.skip_field(index.fields[idx], pos)) {
+                DB_WARNING("skip index field error");
+                return -1;
+            }
             continue;
         }
         const FieldDescriptor* field = descriptor->field(slot - 1);
diff --git a/src/raft/my_raft_log_storage.cpp b/src/raft/my_raft_log_storage.cpp
@@ -278,18 +278,11 @@ int64_t MyRaftLogStorage::get_term(const int64_t index) {
 int MyRaftLogStorage::append_entry(const braft::LogEntry* entry) {
     std::vector<braft::LogEntry*> entries;
     entries.push_back(const_cast<braft::LogEntry*>(entry));
-#ifdef BAIDU_INTERNAL
     return append_entries(entries, nullptr) == 1 ? 0 : -1;
-#else 
-    return append_entries(entries) == 1 ? 0 : -1;
-#endif
 }
 
 int MyRaftLogStorage::append_entries(const std::vector<braft::LogEntry*>& entries
-#ifdef BAIDU_INTERNAL
-        , braft::IOMetric* metric
-#endif
-        ) {
+        , braft::IOMetric* metric) {
     TimeCost time_cost;
     if (entries.empty()) {
         return 0;
diff --git a/src/raft/my_raft_meta_storage.cpp b/src/raft/my_raft_meta_storage.cpp
@@ -117,7 +117,6 @@ int MyRaftMetaStorage::set_term_and_votedfor(const int64_t term, const braft::Pe
     }
 }
 
-#ifdef BAIDU_INTERNAL
 butil::Status MyRaftMetaStorage::init() {
     butil::Status status;
     if (_is_inited) {
@@ -154,19 +153,6 @@ butil::Status MyRaftMetaStorage::get_term_and_votedfor(int64_t* term, braft::Pee
     status.set_error(EINVAL, "MyRaftMetaStorage is error, region_id: %ld", _region_id);
     return status;
 }
-#else
-int MyRaftMetaStorage::init() {
-    if (_is_inited) {
-        return 0;
-    }
-
-    int ret = load();
-    if (ret == 0) {
-        _is_inited = true;
-    }
-    return ret;
-}
-#endif
 
 int MyRaftMetaStorage::load() {
     braft::StablePBMeta meta;

Original file line number	Diff line number	Diff line change
`@@ -26,6 +26,7 @@ inline const google::protobuf::FieldDescriptor* get_field(`
`26`	`26`	`} else {`
`27`	`27`	`int32_t slot = (*field_slot)[field_info->id];`
`28`	`28`	`if (slot == 0) {`
	`29`	`+ DB_FATAL("field:%d, slot is 0, can't be here", field_info->id);`
`29`	`30`	`return nullptr;`
`30`	`31`	`}`
`31`	`32`	`return (*mem_row)->get_field_by_slot(tuple_id, slot);`