Merge branch '3.0' of https://github.com/taosdata/TDengine into feature/vnode_refact1

include/common/trow.h

@@ -307,8 +307,8 @@ static FORCE_INLINE int32_t tdSetBitmapValTypeII(void *pBitmap, int16_t colIdx,
   // use literal value directly and not use formula to simplify the codes
   switch (nOffset) {
     case 0:
-      *pDestByte = ((*pDestByte) & 0x3F) | (valType << 6);
       // set the value and clear other partitions for offset 0
+      *pDestByte = (valType << 6);
       // *pDestByte |= (valType << 6);
       break;
     case 1:
@@ -417,8 +417,8 @@ static FORCE_INLINE int32_t tdSetBitmapValTypeI(void *pBitmap, int16_t colIdx, T
   // use literal value directly and not use formula to simplify the codes
   switch (nOffset) {
     case 0:
-      *pDestByte = ((*pDestByte) & 0x7F) | (valType << 7);
       // set the value and clear other partitions for offset 0
+      *pDestByte = (valType << 7);  
       // *pDestByte |= (valType << 7);
       break;
     case 1:

source/dnode/vnode/src/tsdb/tsdbRead.c

@@ -1410,30 +1410,33 @@ static int32_t doCopyRowsFromFileBlock(STsdbReadHandle* pTsdbReadHandle, int32_t
     if (!isAllRowsNull(src) && pColInfo->info.colId == src->colId) {
       if (!IS_VAR_DATA_TYPE(pColInfo->info.type)) {  // todo opt performance
         //        memmove(pData, (char*)src->pData + bytes * start, bytes * num);
-        for (int32_t k = start; k < num + start; ++k) {
+        int32_t rowIndex = numOfRows;
+        for (int32_t k = start; k <= end; ++k, ++rowIndex) {
           SCellVal sVal = {0};
           if (tdGetColDataOfRow(&sVal, src, k, pCols->bitmapMode) < 0) {
             TASSERT(0);
           }
 
           if (sVal.valType == TD_VTYPE_NULL) {
-            colDataAppendNULL(pColInfo, k);
+            colDataAppendNULL(pColInfo, rowIndex);
           } else {
-            colDataAppend(pColInfo, k, sVal.val, false);
+            colDataAppend(pColInfo, rowIndex, sVal.val, false);
           }
         }
       } else {  // handle the var-string
+        int32_t rowIndex = numOfRows;
+
         // todo refactor, only copy one-by-one
-        for (int32_t k = start; k < num + start; ++k) {
+        for (int32_t k = start; k < num + start; ++k, ++rowIndex) {
           SCellVal sVal = {0};
           if (tdGetColDataOfRow(&sVal, src, k, pCols->bitmapMode) < 0) {
             TASSERT(0);
           }
 
           if (sVal.valType == TD_VTYPE_NULL) {
-            colDataAppendNULL(pColInfo, k);
+            colDataAppendNULL(pColInfo, rowIndex);
           } else {
-            colDataAppend(pColInfo, k, sVal.val, false);
+            colDataAppend(pColInfo, rowIndex, sVal.val, false);
           }
         }
       }
@@ -1441,8 +1444,9 @@ static int32_t doCopyRowsFromFileBlock(STsdbReadHandle* pTsdbReadHandle, int32_t
       j++;
       i++;
     } else {                                           // pColInfo->info.colId < src->colId, it is a NULL data
-      for (int32_t k = start; k < num + start; ++k) {  // TODO opt performance
-        colDataAppend(pColInfo, k, NULL, true);
+      int32_t rowIndex = numOfRows;
+      for (int32_t k = start; k < num + start; ++k, ++rowIndex) {  // TODO opt performance
+        colDataAppend(pColInfo, rowIndex, NULL, true);
       }
       i++;
     }
@@ -1450,8 +1454,10 @@ static int32_t doCopyRowsFromFileBlock(STsdbReadHandle* pTsdbReadHandle, int32_t
 
   while (i < requiredNumOfCols) {  // the remain columns are all null data
     SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
-    for (int32_t k = start; k < num + start; ++k) {
-      colDataAppend(pColInfo, k, NULL, true);  // TODO add a fast version to set a number of consecutive NULL value.
+    int32_t rowIndex = numOfRows;
+
+    for (int32_t k = start; k < num + start; ++k, ++rowIndex) {
+      colDataAppend(pColInfo, rowIndex, NULL, true);  // TODO add a fast version to set a number of consecutive NULL value.
     }
     i++;
   }
@@ -1749,7 +1755,7 @@ int32_t getEndPosInDataBlock(STsdbReadHandle* pTsdbReadHandle, SDataBlockInfo* p
 // be included in the query time window will be discarded
 static void doMergeTwoLevelData(STsdbReadHandle* pTsdbReadHandle, STableCheckInfo* pCheckInfo, SBlock* pBlock) {
   SQueryFilePos* cur = &pTsdbReadHandle->cur;
-  SDataBlockInfo blockInfo = {0};  // GET_FILE_DATA_BLOCK_INFO(pCheckInfo, pBlock);
+  SDataBlockInfo blockInfo = GET_FILE_DATA_BLOCK_INFO(pCheckInfo, pBlock);
   STsdbCfg*      pCfg = &pTsdbReadHandle->pTsdb->config;
 
   initTableMemIterator(pTsdbReadHandle, pCheckInfo);
@@ -1771,9 +1777,7 @@ static void doMergeTwoLevelData(STsdbReadHandle* pTsdbReadHandle, STableCheckInf
   STable* pTable = NULL;
   int32_t endPos = getEndPosInDataBlock(pTsdbReadHandle, &blockInfo);
 
-  tsdbDebug("%p uid:%" PRIu64 " start merge data block, file block range:%" PRIu64 "-%" PRIu64
-            " rows:%d, start:%d,"
-            "end:%d, %s",
+  tsdbDebug("%p uid:%" PRIu64 " start merge data block, file block range:%" PRIu64 "-%" PRIu64 " rows:%d, start:%d, end:%d, %s",
             pTsdbReadHandle, pCheckInfo->tableId, blockInfo.window.skey, blockInfo.window.ekey, blockInfo.rows,
             cur->pos, endPos, pTsdbReadHandle->idStr);
 

source/libs/scalar/src/sclfunc.c

@@ -304,7 +304,7 @@ static int32_t doLengthFunction(SScalarParam *pInput, int32_t inputNum, SScalarP
       continue;
     }
 
-    char *in = pInputData->pData + pInputData->varmeta.offset[i];
+    char *in = colDataGetData(pInputData, i);
     out[i] = lenFn(in, type);
   }
 
@@ -395,11 +395,8 @@ int32_t concatFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOu
 
     int16_t dataLen = 0;
     for (int32_t i = 0; i < inputNum; ++i) {
-      if (pInput[i].numOfRows == 1) {
-        input[i] = pInputData[i]->pData + pInputData[i]->varmeta.offset[0];
-      } else {
-        input[i] = pInputData[i]->pData + pInputData[i]->varmeta.offset[k];
-      }
+      int32_t rowIdx = (pInput[i].numOfRows == 1) ? 0 : k;
+      input[i] = colDataGetData(pInputData[i], rowIdx);
 
       ret = concatCopyHelper(input[i], output, hasNchar, GET_PARAM_TYPE(&pInput[i]), &dataLen);
       if (ret != TSDB_CODE_SUCCESS) {
@@ -473,11 +470,8 @@ int32_t concatWsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *p
         break;
       }
 
-      if (pInput[i].numOfRows == 1) {
-        input[i] = pInputData[i]->pData + pInputData[i]->varmeta.offset[0];
-      } else {
-        input[i] = pInputData[i]->pData + pInputData[i]->varmeta.offset[k];
-      }
+      int32_t rowIdx = (pInput[i].numOfRows == 1) ? 0 : k;
+      input[i] = colDataGetData(pInputData[i], rowIdx);
 
       ret = concatCopyHelper(input[i], output, hasNchar, GET_PARAM_TYPE(&pInput[i]), &dataLen);
       if (ret != TSDB_CODE_SUCCESS) {
@@ -534,7 +528,7 @@ static int32_t doCaseConvFunction(SScalarParam *pInput, int32_t inputNum, SScala
       continue;
     }
 
-    char *input = pInputData->pData + pInputData->varmeta.offset[i];
+    char *input = colDataGetData(pInput[0].columnData, i);
     int32_t len = varDataLen(input);
     if (type == TSDB_DATA_TYPE_VARCHAR) {
       for (int32_t j = 0; j < len; ++j) {
@@ -575,8 +569,8 @@ static int32_t doTrimFunction(SScalarParam *pInput, int32_t inputNum, SScalarPar
       colDataAppendNULL(pOutputData, i);
       continue;
     }
-    char *input  = pInputData->pData + pInputData->varmeta.offset[i];
 
+    char *input = colDataGetData(pInput[0].columnData, i);
     int32_t len = varDataLen(input);
     int32_t charLen = (type == TSDB_DATA_TYPE_VARCHAR) ? len : len / TSDB_NCHAR_SIZE;
     trimFn(input, output, type, charLen);
@@ -615,19 +609,16 @@ int32_t substrFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOu
   SColumnInfoData *pInputData  = pInput->columnData;
   SColumnInfoData *pOutputData = pOutput->columnData;
 
-  char *input  = pInputData->pData + pInputData->varmeta.offset[0];
-  char *output = NULL;
-
   int32_t outputLen = pInputData->varmeta.length * pInput->numOfRows;
   char *outputBuf = taosMemoryCalloc(outputLen, 1);
-  output = outputBuf;
+  char *output = outputBuf;
 
   for (int32_t i = 0; i < pInput->numOfRows; ++i) {
     if (colDataIsNull_s(pInputData, i)) {
       colDataAppendNULL(pOutputData, i);
       continue;
     }
-
+    char *input = colDataGetData(pInput[0].columnData, i);
     int32_t len = varDataLen(input);
     int32_t startPosBytes;
 
@@ -646,7 +637,6 @@ int32_t substrFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOu
 
     varDataSetLen(output, resLen);
     colDataAppend(pOutputData, i , output, false);
-    input += varDataTLen(input);
     output += varDataTLen(output);
   }
 
@@ -799,13 +789,13 @@ int32_t castFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutp
 int32_t toISO8601Function(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
   int32_t type = GET_PARAM_TYPE(pInput);
 
-  char *input  = pInput[0].columnData->pData;
   for (int32_t i = 0; i < pInput[0].numOfRows; ++i) {
     if (colDataIsNull_s(pInput[0].columnData, i)) {
       colDataAppendNULL(pOutput->columnData, i);
       continue;
     }
 
+    char *input = colDataGetData(pInput[0].columnData, i);
     char fraction[20] = {0};
     bool hasFraction = false;
     NUM_TO_STRING(type, input, sizeof(fraction), fraction);
@@ -822,7 +812,8 @@ int32_t toISO8601Function(SScalarParam *pInput, int32_t inputNum, SScalarParam *
       } else if (tsDigits == TSDB_TIME_PRECISION_NANO_DIGITS) {
         timeVal = timeVal / (1000 * 1000 * 1000);
       } else {
-        assert(0);
+        colDataAppendNULL(pOutput->columnData, i);
+        continue;
       }
       hasFraction = true;
       memmove(fraction, fraction + TSDB_TIME_PRECISION_SEC_DIGITS, TSDB_TIME_PRECISION_SEC_DIGITS);
@@ -852,7 +843,6 @@ int32_t toISO8601Function(SScalarParam *pInput, int32_t inputNum, SScalarParam *
     varDataSetLen(buf, len);
 
     colDataAppend(pOutput->columnData, i, buf, false);
-    input   += tDataTypes[type].bytes;
   }
 
   pOutput->numOfRows = pInput->numOfRows;
@@ -864,12 +854,12 @@ int32_t toUnixtimestampFunction(SScalarParam *pInput, int32_t inputNum, SScalarP
   int32_t type = GET_PARAM_TYPE(pInput);
   int32_t timePrec = GET_PARAM_PRECISON(pInput);
 
-  char *input = pInput[0].columnData->pData + pInput[0].columnData->varmeta.offset[0];
   for (int32_t i = 0; i < pInput[0].numOfRows; ++i) {
     if (colDataIsNull_s(pInput[0].columnData, i)) {
       colDataAppendNULL(pOutput->columnData, i);
       continue;
     }
+    char *input = colDataGetData(pInput[0].columnData, i);
 
     int64_t timeVal = 0;
     int32_t ret = convertStringToTimestamp(type, input, timePrec, &timeVal);
@@ -879,7 +869,6 @@ int32_t toUnixtimestampFunction(SScalarParam *pInput, int32_t inputNum, SScalarP
     }
 
     colDataAppend(pOutput->columnData, i, (char *)&timeVal, false);
-    input += varDataTLen(input);
   }
 
   pOutput->numOfRows = pInput->numOfRows;
@@ -897,19 +886,14 @@ int32_t timeTruncateFunction(SScalarParam *pInput, int32_t inputNum, SScalarPara
   int64_t factor = (timePrec == TSDB_TIME_PRECISION_MILLI) ? 1000 :
                    (timePrec == TSDB_TIME_PRECISION_MICRO ? 1000000 : 1000000000);
 
-  char *input = NULL;
-  if (IS_VAR_DATA_TYPE(type)) {
-    input = pInput[0].columnData->pData + pInput[0].columnData->varmeta.offset[0];
-  } else {
-    input = pInput[0].columnData->pData;
-  }
-
   for (int32_t i = 0; i < pInput[0].numOfRows; ++i) {
     if (colDataIsNull_s(pInput[0].columnData, i)) {
       colDataAppendNULL(pOutput->columnData, i);
       continue;
     }
 
+    char *input = colDataGetData(pInput[0].columnData, i);
+
     if (IS_VAR_DATA_TYPE(type)) { /* datetime format strings */
       int32_t ret = convertStringToTimestamp(type, input, TSDB_TIME_PRECISION_NANO, &timeVal);
       if (ret != TSDB_CODE_SUCCESS) {
@@ -959,7 +943,8 @@ int32_t timeTruncateFunction(SScalarParam *pInput, int32_t inputNum, SScalarPara
         } else if (tsDigits <= TSDB_TIME_PRECISION_SEC_DIGITS){
           timeVal = timeVal * factor;
         } else {
-          assert(0);
+          colDataAppendNULL(pOutput->columnData, i);
+          continue;
         }
         break;
       }
@@ -973,7 +958,8 @@ int32_t timeTruncateFunction(SScalarParam *pInput, int32_t inputNum, SScalarPara
         } else if (tsDigits <= TSDB_TIME_PRECISION_SEC_DIGITS) {
           timeVal = timeVal * factor;
         } else {
-          assert(0);
+          colDataAppendNULL(pOutput->columnData, i);
+          continue;
         }
         break;
       }
@@ -987,7 +973,8 @@ int32_t timeTruncateFunction(SScalarParam *pInput, int32_t inputNum, SScalarPara
         } else if (tsDigits <= TSDB_TIME_PRECISION_SEC_DIGITS) {
           timeVal = timeVal * factor / factor / 60 * 60 * factor;
         } else {
-          assert(0);
+          colDataAppendNULL(pOutput->columnData, i);
+          continue;
         }
         break;
       }
@@ -1001,7 +988,8 @@ int32_t timeTruncateFunction(SScalarParam *pInput, int32_t inputNum, SScalarPara
         } else if (tsDigits <= TSDB_TIME_PRECISION_SEC_DIGITS) {
           timeVal = timeVal * factor / factor / 3600 * 3600 * factor;
         } else {
-          assert(0);
+          colDataAppendNULL(pOutput->columnData, i);
+          continue;
         }
         break;
       }
@@ -1015,7 +1003,8 @@ int32_t timeTruncateFunction(SScalarParam *pInput, int32_t inputNum, SScalarPara
         } else if (tsDigits <= TSDB_TIME_PRECISION_SEC_DIGITS) {
           timeVal = timeVal * factor / factor / 86400* 86400 * factor;
         } else {
-          assert(0);
+          colDataAppendNULL(pOutput->columnData, i);
+          continue;
         }
         break;
       }
@@ -1029,7 +1018,8 @@ int32_t timeTruncateFunction(SScalarParam *pInput, int32_t inputNum, SScalarPara
         } else if (tsDigits <= TSDB_TIME_PRECISION_SEC_DIGITS) {
           timeVal = timeVal * factor / factor / 604800 * 604800* factor;
         } else {
-          assert(0);
+          colDataAppendNULL(pOutput->columnData, i);
+          continue;
         }
         break;
       }
@@ -1068,11 +1058,6 @@ int32_t timeTruncateFunction(SScalarParam *pInput, int32_t inputNum, SScalarPara
     }
 
     colDataAppend(pOutput->columnData, i, (char *)&timeVal, false);
-    if (IS_VAR_DATA_TYPE(type)) {
-      input += varDataTLen(input);
-    } else {
-      input += tDataTypes[type].bytes;
-    }
   }
 
   pOutput->numOfRows = pInput->numOfRows;
@@ -1094,12 +1079,6 @@ int32_t timeDiffFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *p
         type != TSDB_DATA_TYPE_BINARY && type != TSDB_DATA_TYPE_NCHAR) {
       return TSDB_CODE_FAILED;
     }
-
-    if (IS_VAR_DATA_TYPE(type)) {
-      input[k] = pInput[k].columnData->pData + pInput[k].columnData->varmeta.offset[0];
-    } else {
-      input[k] = pInput[k].columnData->pData;
-    }
   }
 
   for (int32_t i = 0; i < pInput[0].numOfRows; ++i) {
@@ -1109,6 +1088,9 @@ int32_t timeDiffFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *p
         continue;
       }
 
+      int32_t rowIdx = (pInput[k].numOfRows == 1) ? 0 : i;
+      input[k] = colDataGetData(pInput[k].columnData, rowIdx);
+
       int32_t type = GET_PARAM_TYPE(&pInput[k]);
       if (IS_VAR_DATA_TYPE(type)) { /* datetime format strings */
         int32_t ret = convertStringToTimestamp(type, input[k], TSDB_TIME_PRECISION_NANO, &timeVal[k]);
@@ -1138,14 +1120,9 @@ int32_t timeDiffFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *p
           timeVal[k] = timeVal[k] * 1000;
         } else if (tsDigits == TSDB_TIME_PRECISION_NANO_DIGITS) {
           timeVal[k] = timeVal[k];
-        }
-      }
-
-      if (pInput[k].numOfRows != 1) {
-        if (IS_VAR_DATA_TYPE(type)) {
-          input[k] += varDataTLen(input[k]);
         } else {
-          input[k] += tDataTypes[type].bytes;
+          colDataAppendNULL(pOutput->columnData, i);
+          continue;
         }
       }
     }