[enhancement](profile) profile counter should always gt than 0

Author: yiguolei

be/src/runtime/cache/result_node.h

@@ -93,7 +93,7 @@ class PartitionRowBatch {
 private:
     int64_t _partition_key;
     PCacheValue* _cache_value = nullptr;
-    size_t _data_size;
+    int64_t _data_size;
     CacheStat _cache_stat;
 };
 
@@ -175,7 +175,7 @@ class ResultNode {
     UniqueId _sql_key;
     ResultNode* _prev = nullptr;
     ResultNode* _next = nullptr;
-    size_t _data_size;
+    int64_t _data_size;
     PartitionRowBatchList _partition_list;
     PartitionRowBatchMap _partition_map;
 };

be/src/util/runtime_profile.h

@@ -190,15 +190,22 @@ class RuntimeProfile {
 
         virtual Counter* clone() const { return new Counter(type(), value(), _level); }
 
-        virtual void update(int64_t delta) { _value.fetch_add(delta, std::memory_order_relaxed); }
+        virtual void update(int64_t delta) {
+            _value.fetch_add(delta, std::memory_order_relaxed);
+            DCHECK_GT(_value.load(std::memory_order_seq_cst), -1L);
+        }
 
         void bit_or(int64_t delta) { _value.fetch_or(delta, std::memory_order_relaxed); }
 
-        virtual void set(int64_t value) { _value.store(value, std::memory_order_relaxed); }
+        virtual void set(int64_t value) {
+            _value.store(value, std::memory_order_relaxed);
+            DCHECK_GT(_value.load(std::memory_order_seq_cst), -1L);
+        }
 
         virtual void set(double value) {
             DCHECK_EQ(sizeof(value), sizeof(int64_t));
             _value.store(binary_cast<double, int64_t>(value), std::memory_order_relaxed);
+            DCHECK_GT(_value.load(std::memory_order_seq_cst), -1L);
         }
 
         virtual int64_t value() const { return _value.load(std::memory_order_relaxed); }
@@ -270,6 +277,7 @@ class RuntimeProfile {
             if (delta > 0) {
                 UpdateMax(current_value_);
             }
+            DCHECK_GT(current_value_.load(std::memory_order_seq_cst), -1L);
         }
         virtual void update(int64_t delta) override { add(delta); }
 
@@ -331,6 +339,7 @@ class RuntimeProfile {
         void set(int64_t v) override {
             current_value_.store(v, std::memory_order_relaxed);
             UpdateMax(v);
+            DCHECK_GT(_value.load(std::memory_order_seq_cst), -1L);
         }
 
         int64_t current_value() const { return current_value_.load(std::memory_order_relaxed); }

be/src/vec/common/sort/heap_sorter.h

@@ -50,7 +50,7 @@ class HeapSorter final : public Sorter {
 
 private:
     void _do_filter(MergeSortCursorImpl& block, size_t num_rows);
-    size_t _data_size = 0;
+    int64_t _data_size = 0;
     size_t _heap_size = 0;
     size_t _queue_row_num = 0;
     MergeSorterQueue _queue;