[td-11818] optimize the sort performance.

include/common/common.h

@@ -74,6 +74,7 @@ typedef struct SVarColAttr {
 // pBlockAgg->numOfNull == 0, no data are null.
 typedef struct SColumnInfoData {
   SColumnInfo info;   // TODO filter info needs to be removed
+  bool        hasNull;// if current column data has null value.
   char       *pData;  // the corresponding block data in memory
   union {
     char       *nullbitmap;  // bitmap, one bit for each item in the list

include/common/tep.h

@@ -27,12 +27,50 @@ bool isEpsetEqual(const SEpSet *s1, const SEpSet *s2);
 void   updateEpSet_s(SCorEpSet *pEpSet, SEpSet *pNewEpSet);
 SEpSet getEpSet_s(SCorEpSet *pEpSet);
 
-bool colDataIsNull_f(const char* bitmap, uint32_t row);
+#define BitPos(_n)        ((_n) & ((1<<NBIT) - 1))
+#define NBIT (3u)
+
+static FORCE_INLINE bool colDataIsNull_f(const char* bitmap, uint32_t row) {
+  return (bitmap[row>>3u] & (1u<<(7u - BitPos(row)))) == (1u<<(7u - BitPos(row)));
+}
+
 void colDataSetNull_f(char* bitmap, uint32_t row);
 
-bool colDataIsNull(const SColumnInfoData* pColumnInfoData, uint32_t totalRows, uint32_t row, SColumnDataAgg* pColAgg);
+static FORCE_INLINE bool colDataIsNull(const SColumnInfoData* pColumnInfoData, uint32_t totalRows, uint32_t row, SColumnDataAgg* pColAgg) {
+  if (!pColumnInfoData->hasNull) {
+    return false;
+  }
+
+  if (pColAgg != NULL) {
+    if (pColAgg->numOfNull == totalRows) {
+      ASSERT(pColumnInfoData->nullbitmap == NULL);
+      return true;
+    } else if (pColAgg->numOfNull == 0) {
+      ASSERT(pColumnInfoData->nullbitmap == NULL);
+      return false;
+    }
+  }
+
+  if (IS_VAR_DATA_TYPE(pColumnInfoData->info.type)) {
+    return pColumnInfoData->varmeta.offset[row] == -1;
+  } else {
+    if (pColumnInfoData->nullbitmap == NULL) {
+      return false;
+    }
+
+    return colDataIsNull_f(pColumnInfoData->nullbitmap, row);
+  }
+}
+
+static FORCE_INLINE char* colDataGet(SColumnInfoData* pColumnInfoData, uint32_t row) {
+  char* p = pColumnInfoData->pData;
+  if (IS_VAR_DATA_TYPE(pColumnInfoData->info.type)) {
+    return p + pColumnInfoData->varmeta.offset[row];
+  } else {
+    return p + (row * pColumnInfoData->info.bytes);
+  }
+}
 
-char* colDataGet(SColumnInfoData* pColumnInfoData, uint32_t row);
 int32_t colDataAppend(SColumnInfoData* pColumnInfoData, uint32_t currentRow, const char* pData, bool isNull);
 int32_t colDataMergeCol(SColumnInfoData* pColumnInfoData, uint32_t numOfRow1, const SColumnInfoData* pSource, uint32_t numOfRow2);
 int32_t colDataUpdateTsWindow(SSDataBlock* pDataBlock);

source/common/src/tep.c

@@ -60,49 +60,12 @@ SEpSet getEpSet_s(SCorEpSet *pEpSet) {
   return ep;
 }
 
-bool colDataIsNull(const SColumnInfoData* pColumnInfoData, uint32_t totalRows, uint32_t row, SColumnDataAgg* pColAgg) {
-  if (pColAgg != NULL) {
-    if (pColAgg->numOfNull == totalRows) {
-      ASSERT(pColumnInfoData->nullbitmap == NULL);
-      return true;
-    } else if (pColAgg->numOfNull == 0) {
-      ASSERT(pColumnInfoData->nullbitmap == NULL);
-      return false;
-    }
-  }
-
-  if (IS_VAR_DATA_TYPE(pColumnInfoData->info.type)) {
-    return pColumnInfoData->varmeta.offset[row] == -1;
-  } else {
-    if (pColumnInfoData->nullbitmap == NULL) {
-      return false;
-    }
-
-    return colDataIsNull_f(pColumnInfoData->nullbitmap, row);
-  }
-}
-
-#define NBIT (3u)
 #define BitmapLen(_n)     (((_n) + ((1<<NBIT)-1)) >> NBIT)
-#define BitPos(_n)        ((_n) & ((1<<NBIT) - 1))
-
-bool colDataIsNull_f(const char* bitmap, uint32_t row) {
-  return (bitmap[row>>3u] & (1u<<(7u - BitPos(row)))) == (1u<<(7u - BitPos(row)));
-}
 
 void colDataSetNull_f(char* bitmap, uint32_t row) {
   bitmap[row>>3u] |= (1u << (7u - BitPos(row)));
 }
 
-char* colDataGet(SColumnInfoData* pColumnInfoData, uint32_t row) {
-  char* p = pColumnInfoData->pData;
-  if (IS_VAR_DATA_TYPE(pColumnInfoData->info.type)) {
-    return p + pColumnInfoData->varmeta.offset[row];
-  } else {
-    return p + (row * pColumnInfoData->info.bytes);
-  }
-}
-
 static int32_t ensureBitmapSize(SColumnInfoData* pColumnInfoData, uint32_t size) {
 #if 0
   ASSERT(pColumnInfoData != NULL);
@@ -579,37 +542,42 @@ int32_t dataBlockCompar(const void* p1, const void* p2, const void* param) {
   int32_t* right = (int32_t*) p2;
 
   SArray* pInfo = pHelper->orderInfo;
-  size_t num = taosArrayGetSize(pInfo);
+
-  for(int32_t i = 0; i < num; ++i) {
+  for(int32_t i = 0; i < pInfo->size; ++i) {
     SBlockOrderInfo* pOrder = taosArrayGet(pInfo, i);
-    SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, pOrder->colIndex);
+    SColumnInfoData* pColInfoData = TARRAY_GET_ELEM(pDataBlock->pDataBlock, pOrder->colIndex);
 
-    bool leftNull  = colDataIsNull(pColInfoData, pDataBlock->info.rows, *left, pDataBlock->pBlockAgg);
+    if (pColInfoData->hasNull) {
-    bool rightNull = colDataIsNull(pColInfoData, pDataBlock->info.rows, *right, pDataBlock->pBlockAgg);
+      bool leftNull  = colDataIsNull(pColInfoData, pDataBlock->info.rows, *left, pDataBlock->pBlockAgg);
-    if (leftNull && rightNull) {
+      bool rightNull = colDataIsNull(pColInfoData, pDataBlock->info.rows, *right, pDataBlock->pBlockAgg);
-      continue; // continue to next slot
+      if (leftNull && rightNull) {
+        continue; // continue to next slot
+      }
+
+      if (rightNull) {
+        return pHelper->nullFirst? 1:-1;
+      }
+
+      if (leftNull) {
+        return pHelper->nullFirst? -1:1;
+      }
     }
 
-    if (rightNull) {
+    void* left1  = colDataGet(pColInfoData, *left);
-      return pHelper->nullFirst? 1:-1;
-    }
-
-    if (leftNull) {
-      return pHelper->nullFirst? -1:1;
-    }
-
-    void* left1 = colDataGet(pColInfoData, *left);
     void* right1 = colDataGet(pColInfoData, *right);
 
     switch(pColInfoData->info.type) {
       case TSDB_DATA_TYPE_INT: {
-        if (*(int32_t*) left1 == *(int32_t*) right1) {
+        int32_t leftx  = *(int32_t*) left1;
+        int32_t rightx = *(int32_t*) right1;
+
+        if (leftx == rightx) {
           break;
         } else {
           if (pOrder->order == TSDB_ORDER_ASC) {
-            return (*(int32_t*) left1 <= *(int32_t*) right1)? -1:1;
+            return (leftx <= rightx)? -1:1;
           } else {
-            return (*(int32_t*) left1 <= *(int32_t*) right1)? 1:-1;
+            return (leftx <= rightx)? 1:-1;
           }
         }
       }