[TD-2574]<enhance>: added general heapsort algorithm

src/query/src/qAggMain.c

@@ -2201,99 +2201,35 @@ static void valuePairAssign(tValuePair *dst, int16_t type, const char *val, int6
     memcpy((dst)->pTags, (src)->pTags, (size_t)(__l)); \
   } while (0)
 
-static void heapSwap(tValuePair *a, tValuePair *b, const int16_t tagLen) {
-  char        tag[32768];
-  tValuePair  temp;
+static int32_t topBotComparFn(const void *p1, const void *p2, const void *param)
+{
+  uint16_t     type = *(uint16_t *) param;
+  tValuePair  *val1 = *(tValuePair **) p1;
+  tValuePair  *val2 = *(tValuePair **) p2;
 
-  memset(tag, 0, sizeof(tag));
-  temp.pTags = tag;
-
-  VALUEPAIRASSIGN(&temp, a, tagLen);
-  VALUEPAIRASSIGN(a, b, tagLen);
-  VALUEPAIRASSIGN(b, &temp, tagLen);
-}
-
-static void heapAdjust(tValuePair **pList, uint16_t type, int16_t tagLen, int32_t start, int32_t end, bool minRoot) {
-  int32_t     parent = start;
-  int32_t     child = 2 * parent + 1;
-
-  while (child <= end) {
-    if (IS_SIGNED_NUMERIC_TYPE(type)) {
-      if (minRoot) {
-        if (child + 1 <= end && pList[child]->v.i64 < pList[child + 1]->v.i64) {
-          child++;
-        }
-
-        if (pList[parent]->v.i64 > pList[child]->v.i64) {
-          break;
-        }
-      } else {
-        if (child + 1 <= end && pList[child]->v.i64 >= pList[child + 1]->v.i64) {
-          child++;
-        }
-
-        if (pList[parent]->v.i64 <= pList[child]->v.i64) {
-          break;
-        }
-      }
-    } else if (IS_UNSIGNED_NUMERIC_TYPE(type)) {
-      if (minRoot) {
-        if (child + 1 <= end && pList[child]->v.u64 < pList[child + 1]->v.u64) {
-          child++;
-        }
-
-        if (pList[parent]->v.u64 > pList[child]->v.u64) {
-          break;
-        }
-      } else {
-        if (child + 1 <= end && pList[child]->v.u64 >= pList[child + 1]->v.u64) {
-          child++;
-        }
-
-        if (pList[parent]->v.u64 <= pList[child]->v.u64) {
-          break;
-        }
-      }
-    } else {
-      if (minRoot) {
-        if (child + 1 <= end && pList[child]->v.dKey < pList[child + 1]->v.dKey) {
-          child++;
-        }
-
-        if (pList[parent]->v.dKey > pList[child]->v.dKey) {
-          break;
-        }
-      } else {
-        if (child + 1 <= end && pList[child]->v.dKey >= pList[child + 1]->v.dKey) {
-          child++;
-        }
-
-        if (pList[parent]->v.dKey <= pList[child]->v.dKey) {
-          break;
-        }
-      }
-    }
-
-    heapSwap(pList[parent], pList[child], tagLen);
-
-    parent = child;
-    child = parent * 2 + 1;
+  if (IS_SIGNED_NUMERIC_TYPE(type)) {
+    return val1->v.i64 - val2->v.i64;
+  } else if (IS_UNSIGNED_NUMERIC_TYPE(type)) {
+    return val1->v.u64 - val2->v.u64;
   }
+
+  return val1->v.dKey - val2->v.dKey;
 }
 
-void heapSort(tValuePair **pList, uint16_t type, int16_t tagLen, int32_t len, bool minRoot) {
-  int32_t  i;
+static void topBotSwapFn(void *dst, void *src, const void *param)
+{
+  char         tag[32768];
+  tValuePair   temp;
+  uint16_t     tagLen = *(uint16_t *) param;
+  tValuePair  *vdst = *(tValuePair **) dst;
+  tValuePair  *vsrc = *(tValuePair **) src;
 
-  for (i = len / 2 - 1; i >= 0; i--) {
-    heapAdjust(pList, type, tagLen, i, len - 1, minRoot);
-  }
+  memset(tag, 0, sizeof(tag));
+  temp.pTags = tag;
 
-/*
-  for (i = len - 1; i > 0; i--) {
-    heapSwap(pList[0], pList[i], tagsLen);
-    heapAdjust(pList, type, 0, tagsLen, i - 1, minRoot);
-  }
-*/
+  VALUEPAIRASSIGN(&temp, vdst, tagLen);
+  VALUEPAIRASSIGN(vdst, vsrc, tagLen);
+  VALUEPAIRASSIGN(vsrc, &temp, tagLen);
 }
 
 static void do_top_function_add(STopBotInfo *pInfo, int32_t maxLen, void *pData, int64_t ts, uint16_t type,
@@ -2303,11 +2239,11 @@ static void do_top_function_add(STopBotInfo *pInfo, int32_t maxLen, void *pData,
   
   tValuePair **pList = pInfo->res;
   assert(pList != NULL);
-  
+
   if (pInfo->num < maxLen) {
     valuePairAssign(pList[pInfo->num], type, (const char *)&val.i64, ts, pTags, pTagInfo, stage);
 
-    heapSort(pList, type, pTagInfo->tagsLen, pInfo->num + 1, 0);
+    taosheapsort((void *) pList, sizeof(tValuePair **), pInfo->num + 1, (const void *) &type, topBotComparFn, (const void *) &pTagInfo->tagsLen, topBotSwapFn, 0);
  
     pInfo->num++;
   } else {
@@ -2315,7 +2251,7 @@ static void do_top_function_add(STopBotInfo *pInfo, int32_t maxLen, void *pData,
         (IS_UNSIGNED_NUMERIC_TYPE(type) && val.u64 > pList[0]->v.u64) ||
         (IS_FLOAT_TYPE(type) && val.dKey > pList[0]->v.dKey)) {
       valuePairAssign(pList[0], type, (const char *)&val.i64, ts, pTags, pTagInfo, stage);
-      heapAdjust(pList, type, pTagInfo->tagsLen, 0, maxLen - 1, 0);
+      taosheapadjust((void *) pList, sizeof(tValuePair **), 0, maxLen - 1, (const void *) &type, topBotComparFn, (const void *) &pTagInfo->tagsLen, topBotSwapFn, 0);
     }
   }
 }
@@ -2331,7 +2267,7 @@ static void do_bottom_function_add(STopBotInfo *pInfo, int32_t maxLen, void *pDa
   if (pInfo->num < maxLen) {
     valuePairAssign(pList[pInfo->num], type, (const char *)&val.i64, ts, pTags, pTagInfo, stage);
 
-    heapSort(pList, type, pTagInfo->tagsLen, pInfo->num + 1, 1);
+    taosheapsort((void *) pList, sizeof(tValuePair **), pInfo->num + 1, (const void *) &type, topBotComparFn, (const void *) &pTagInfo->tagsLen, topBotSwapFn, 1);
 
     pInfo->num++;
   } else {
@@ -2339,7 +2275,7 @@ static void do_bottom_function_add(STopBotInfo *pInfo, int32_t maxLen, void *pDa
         (IS_UNSIGNED_NUMERIC_TYPE(type) && val.u64 < pList[0]->v.u64) ||
         (IS_FLOAT_TYPE(type) && val.dKey < pList[0]->v.dKey)) {
       valuePairAssign(pList[0], type, (const char *)&val.i64, ts, pTags, pTagInfo, stage);
-      heapAdjust(pList, type, pTagInfo->tagsLen, 0, maxLen - 1, 1);
+      taosheapadjust((void *) pList, sizeof(tValuePair **), 0, maxLen - 1, (const void *) &type, topBotComparFn, (const void *) &pTagInfo->tagsLen, topBotSwapFn, 1);
     }
   }
 }

src/util/inc/talgo.h

@@ -34,6 +34,7 @@ typedef int (*__compar_fn_t) (const void *, const void *);
 #define elePtrAt(base, size, idx) (void *)((char *)(base) + (size) * (idx))
 
 typedef int32_t (*__ext_compar_fn_t)(const void *p1, const void *p2, const void *param);
+typedef void (*__ext_swap_fn_t)(void *p1, void *p2, const void *param);
 
 /**
  * quick sort, with the compare function requiring additional parameters support
@@ -59,6 +60,38 @@ void taosqsort(void *src, size_t numOfElem, size_t size, const void* param, __ex
  */
 void *taosbsearch(const void *key, const void *base, size_t nmemb, size_t size, __compar_fn_t fn, int flags);
 
+/**
+ * adjust heap
+ *
+ * @param base: the start address of array
+ * @param size: size of every item in array
+ * @param start: the first index
+ * @param end: the last index
+ * @param parcompar: parameters for compare function
+ * @param compar: user defined compare function
+ * @param parswap: parameters for swap function
+ * @param swap: user defined swap function, the default swap function doswap will be used if swap is NULL
+ * @param maxroot: if heap is max root heap
+ * @return
+ */
+void taosheapadjust(void *base, int32_t size, int32_t start, int32_t end, const void *parcompar, __ext_compar_fn_t compar, const void *parswap, __ext_swap_fn_t swap, bool maxroot);
+
+/**
+ * sort heap to make sure it is a max/min root heap
+ *
+ * @param base: the start address of array
+ * @param size: size of every item in array
+ * @param len: the length of array
+ * @param parcompar: parameters for compare function
+ * @param compar: user defined compare function
+ * @param parswap: parameters for swap function
+ * @param swap: user defined swap function, the default swap function doswap will be used if swap is NULL
+ * @param maxroot: if heap is max root heap
+ * @return
+ */
+void taosheapsort(void *base, int32_t size, int32_t len, const void *parcompar, __ext_compar_fn_t compar, const void *parswap, __ext_swap_fn_t swap, bool maxroot);
+
+
 #ifdef __cplusplus
 }
 #endif

src/util/src/talgo.c

@@ -225,3 +225,89 @@ void * taosbsearch(const void *key, const void *base, size_t nmemb, size_t size,
   
   return NULL;
 }
+
+void taosheapadjust(void *base, int32_t size, int32_t start, int32_t end, const void *parcompar, __ext_compar_fn_t compar, const void *parswap, __ext_swap_fn_t swap, bool maxroot)
+{
+  int32_t  parent;
+  int32_t  child;
+  char    *buf;
+
+  if (base && size > 0 && compar) {
+    parent = start;
+    child = 2 * parent + 1;
+
+    if (swap == NULL) {
+      buf = calloc(1, size);
+      if (buf == NULL) {
+        return;
+      }
+    }
+
+    if (maxroot) {
+      while (child <= end) {
+        if (child + 1 <= end && (*compar)(elePtrAt(base, size, child), elePtrAt(base, size, child + 1), parcompar) < 0) {
+          child++;
+        }
+
+        if ((*compar)(elePtrAt(base, size, parent), elePtrAt(base, size, child), parcompar) > 0) {
+          break;
+        }
+
+        if (swap == NULL) {
+          doswap(elePtrAt(base, size, parent), elePtrAt(base, size, child), size, buf);
+        } else {
+          (*swap)(elePtrAt(base, size, parent), elePtrAt(base, size, child), parswap);
+        }
+
+        parent = child;
+        child = 2 * parent + 1;
+      }
+    } else {
+      while (child <= end) {
+        if (child + 1 <= end && (*compar)(elePtrAt(base, size, child), elePtrAt(base, size, child + 1), parcompar) > 0) {
+          child++;
+        }
+
+        if ((*compar)(elePtrAt(base, size, parent), elePtrAt(base, size, child), parcompar) < 0) {
+          break;
+        }
+
+        if (swap == NULL) {
+          doswap(elePtrAt(base, size, parent), elePtrAt(base, size, child), size, buf);
+        } else {
+          (*swap)(elePtrAt(base, size, parent), elePtrAt(base, size, child), parswap);
+        }
+
+        parent = child;
+        child = 2 * parent + 1;
+      }
+    }
+
+    if (swap == NULL) {
+      tfree(buf);
+    }
+  }
+}
+
+void taosheapsort(void *base, int32_t size, int32_t len, const void *parcompar, __ext_compar_fn_t compar, const void *parswap, __ext_swap_fn_t swap, bool maxroot)
+{
+  int32_t  i;
+
+  if (base && size > 0) {
+    for (i = len / 2 - 1; i >= 0; i--) {
+      taosheapadjust(base, size, i, len - 1, parcompar, compar, parswap, swap, maxroot);
+    }
+  }
+
+/*
+  char *buf = calloc(1, size);
+
+  for (i = len - 1; i > 0; i--) {
+    doswap(elePtrAt(base, size, 0), elePtrAt(base, size, i));
+    taosheapadjust(base, size, 0, i - 1, parcompar, compar, parswap, swap, maxroot);
+  }
+
+  tfree(buf);
+*/
+}
+