more code

include/common/tdataformat.h

@@ -137,6 +137,7 @@ void    tColDataInit(SColData *pColData, int16_t cid, int8_t type, int8_t smaOn)
 void    tColDataClear(SColData *pColData);
 void    tColDataDeepClear(SColData *pColData);
 int32_t tColDataAppendValue(SColData *pColData, SColVal *pColVal);
+int32_t tColDataUpdateValue(SColData *pColData, SColVal *pColVal, int32_t flag);
 void    tColDataGetValue(SColData *pColData, int32_t iVal, SColVal *pColVal);
 uint8_t tColDataGetBitValue(const SColData *pColData, int32_t iVal);
 int32_t tColDataCopy(SColData *pColDataFrom, SColData *pColData, xMallocFn xMalloc, void *arg);
@@ -146,9 +147,9 @@ extern void (*tColDataCalcSMA[])(SColData *pColData, int64_t *sum, int64_t *max,
 int32_t tColDataAddValueByBind(SColData *pColData, TAOS_MULTI_BIND *pBind);
 void    tColDataSortMerge(SArray *colDataArr);
 
-//for raw block
+// for raw block
-int32_t tColDataAddValueByDataBlock(SColData *pColData, int8_t type, int32_t bytes,
+int32_t tColDataAddValueByDataBlock(SColData *pColData, int8_t type, int32_t bytes, int32_t nRows, char *lengthOrbitmap,
-                                    int32_t nRows, char* lengthOrbitmap, char *data);
+                                    char *data);
 // for encode/decode
 int32_t tPutColData(uint8_t *pBuf, SColData *pColData);
 int32_t tGetColData(uint8_t *pBuf, SColData *pColData);

source/common/src/tdataformat.c

@@ -1910,6 +1910,107 @@ int32_t tColDataAppendValue(SColData *pColData, SColVal *pColVal) {
       pColVal->value.nData);
 }
 
+static int32_t tColDataUpdateValue10(SColData *pColData, uint8_t *pData, uint32_t nData, int32_t flag) {
+  ASSERT(0);
+  return 0;
+}
+static int32_t tColDataUpdateValue11(SColData *pColData, uint8_t *pData, uint32_t nData, int32_t flag) {
+  ASSERT(0);
+  return 0;
+}
+static int32_t tColDataUpdateValue12(SColData *pColData, uint8_t *pData, uint32_t nData, int32_t flag) {
+  ASSERT(0);
+  return 0;
+}
+static int32_t tColDataUpdateValue20(SColData *pColData, uint8_t *pData, uint32_t nData, int32_t flag) {
+  ASSERT(0);
+  return 0;
+}
+static int32_t tColDataUpdateValue21(SColData *pColData, uint8_t *pData, uint32_t nData, int32_t flag) {
+  ASSERT(0);
+  return 0;
+}
+static int32_t tColDataUpdateValue22(SColData *pColData, uint8_t *pData, uint32_t nData, int32_t flag) {
+  ASSERT(0);
+  return 0;
+}
+static int32_t tColDataUpdateValue30(SColData *pColData, uint8_t *pData, uint32_t nData, int32_t flag) {
+  ASSERT(0);
+  return 0;
+}
+static int32_t tColDataUpdateValue31(SColData *pColData, uint8_t *pData, uint32_t nData, int32_t flag) {
+  ASSERT(0);
+  return 0;
+}
+static int32_t tColDataUpdateValue32(SColData *pColData, uint8_t *pData, uint32_t nData, int32_t flag) {
+  ASSERT(0);
+  return 0;
+}
+static int32_t tColDataUpdateValue40(SColData *pColData, uint8_t *pData, uint32_t nData, int32_t flag) {
+  ASSERT(0);
+  return 0;
+}
+static int32_t tColDataUpdateValue41(SColData *pColData, uint8_t *pData, uint32_t nData, int32_t flag) {
+  ASSERT(0);
+  return 0;
+}
+static int32_t tColDataUpdateValue42(SColData *pColData, uint8_t *pData, uint32_t nData, int32_t flag) {
+  ASSERT(0);
+  return 0;
+}
+static int32_t tColDataUpdateValue50(SColData *pColData, uint8_t *pData, uint32_t nData, int32_t flag) {
+  ASSERT(0);
+  return 0;
+}
+static int32_t tColDataUpdateValue51(SColData *pColData, uint8_t *pData, uint32_t nData, int32_t flag) {
+  ASSERT(0);
+  return 0;
+}
+static int32_t tColDataUpdateValue52(SColData *pColData, uint8_t *pData, uint32_t nData, int32_t flag) {
+  ASSERT(0);
+  return 0;
+}
+static int32_t tColDataUpdateValue60(SColData *pColData, uint8_t *pData, uint32_t nData, int32_t flag) {
+  ASSERT(0);
+  return 0;
+}
+static int32_t tColDataUpdateValue61(SColData *pColData, uint8_t *pData, uint32_t nData, int32_t flag) {
+  ASSERT(0);
+  return 0;
+}
+static int32_t tColDataUpdateValue62(SColData *pColData, uint8_t *pData, uint32_t nData, int32_t flag) {
+  ASSERT(0);
+  return 0;
+}
+static int32_t tColDataUpdateValue70(SColData *pColData, uint8_t *pData, uint32_t nData, int32_t flag) {
+  ASSERT(0);
+  return 0;
+}
+static int32_t tColDataUpdateValue71(SColData *pColData, uint8_t *pData, uint32_t nData, int32_t flag) {
+  ASSERT(0);
+  return 0;
+}
+static int32_t tColDataUpdateValue72(SColData *pColData, uint8_t *pData, uint32_t nData, int32_t flag) {
+  ASSERT(0);
+  return 0;
+}
+static int32_t (*tColDataUpdateValueImpl[8][3])(SColData *pColData, uint8_t *pData, uint32_t nData, int32_t flag) = {
+    {NULL, NULL, NULL},                                                     // 0
+    {tColDataUpdateValue10, tColDataUpdateValue11, tColDataUpdateValue12},  // HAS_NONE
+    {tColDataUpdateValue20, tColDataUpdateValue21, tColDataUpdateValue22},  // HAS_NULL
+    {tColDataUpdateValue30, tColDataUpdateValue31, tColDataUpdateValue32},  // HAS_NULL|HAS_NONE
+    {tColDataUpdateValue40, tColDataUpdateValue41, tColDataUpdateValue42},  // HAS_VALUE
+    {tColDataUpdateValue50, tColDataUpdateValue51, tColDataUpdateValue52},  // HAS_VALUE|HAS_NONE
+    {tColDataUpdateValue60, tColDataUpdateValue61, tColDataUpdateValue62},  // HAS_VALUE|HAS_NULL
+    {tColDataUpdateValue70, tColDataUpdateValue71, tColDataUpdateValue72},  // HAS_VALUE|HAS_NULL|HAS_NONE
+};
+int32_t tColDataUpdateValue(SColData *pColData, SColVal *pColVal, int32_t flag) {
+  ASSERT(pColData->cid == pColVal->cid && pColData->type == pColVal->type);
+  return tColDataUpdateValueImpl[pColData->flag][pColVal->flag](
+      pColData, IS_VAR_DATA_TYPE(pColData->type) ? pColVal->value.pData : (uint8_t *)&pColVal->value.val,
+      pColVal->value.nData, flag);
+}
+
 static FORCE_INLINE void tColDataGetValue1(SColData *pColData, int32_t iVal, SColVal *pColVal) {  // HAS_NONE
   *pColVal = COL_VAL_NONE(pColData->cid, pColData->type);
 }

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

@@ -530,6 +530,8 @@ static bool tsdbCompactRowIsDeleted(STsdbCompactor *pCompactor, TSDBROW *pRow) {
     int64_t version;
     if (pCompactor->iKey < taosArrayGetSize(pCompactor->aSkyLine)) {
       version = TMAX(pCompactor->sKey.version, pCompactor->aTSDBKEY[pCompactor->iKey].version);
+    } else {
+      version = pCompactor->sKey.version;
     }
 
     if (tKey.version > version) {
@@ -858,15 +860,19 @@ int32_t tsdbCompact(STsdb *pTsdb, int32_t flag) {
     if (pCompactor->pDFileSet == NULL) break;
 
     // loop to merge row by row
-    SRowInfo *pRowInfo = NULL;
+    SRowInfo *pRowInfo;
-    STSchema *pTSchema = NULL;
+    STSchema *pTSchema;
     int64_t   nRow = 0;
-    for (;;) {
+
-      code = tsdbCompactGetRow(pCompactor, &pRowInfo, &pTSchema);
+    code = tsdbCompactGetRow(pCompactor, &pRowInfo, &pTSchema);
+    TSDB_CHECK_CODE(code, lino, _exit);
+
+    if (pRowInfo && (code = tBlockDataInit(&pCompactor->bData, &(TABLEID){.suid = pRowInfo->suid, .uid = pRowInfo->uid},
+                                           pTSchema, NULL, 0))) {
       TSDB_CHECK_CODE(code, lino, _exit);
+    }
 
-      if (pRowInfo == NULL) break;
+    while (pRowInfo) {
-
       nRow++;
 
       if (pCompactor->bData.suid == 0 && pCompactor->bData.uid == 0) {  // init the block data if not initialized yet
@@ -907,6 +913,9 @@ int32_t tsdbCompact(STsdb *pTsdb, int32_t flag) {
       // iterate to next row
       code = tsdbCompactNextRow(pCompactor);
       TSDB_CHECK_CODE(code, lino, _exit);
+
+      code = tsdbCompactGetRow(pCompactor, &pRowInfo, &pTSchema);
+      TSDB_CHECK_CODE(code, lino, _exit);
     }
 
     code = tsdbCompactWriteBlockData(pCompactor);