Merge remote-tracking branch 'origin/develop' into feature/file

src/client/src/tscAsync.c

@@ -428,6 +428,7 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
       } else {
         assert(code == TSDB_CODE_SUCCESS);      
       }
+
       // param already freed by other routine and pSql in tscCache when ctrl + c 
       if (atomic_load_ptr(&pSql->param) == NULL) {
         return;
@@ -441,6 +442,20 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
       assert(pParObj->signature == pParObj && trs->subqueryIndex == pTableMetaInfo->vgroupIndex &&
           pTableMetaInfo->vgroupIndex >= 0 && pTableMetaInfo->vgroupList != NULL);
 
+      // tscProcessSql can add error into async res
+      tscProcessSql(pSql);
+      return;
+    } else if (TSDB_QUERY_HAS_TYPE(pQueryInfo->type, TSDB_QUERY_TYPE_TAG_FILTER_QUERY)) {
+      tscDebug("%p update table meta in local cache, continue to process sql and send corresponding tid_tag query", pSql);
+      STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
+      code = tscGetTableMeta(pSql, pTableMetaInfo);
+      if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
+        return;
+      } else {
+        assert(code == TSDB_CODE_SUCCESS);
+      }
+
+      assert((tscGetNumOfTags(pTableMetaInfo->pTableMeta) != 0));
       // tscProcessSql can add error into async res
       tscProcessSql(pSql);
       return;
@@ -465,7 +480,6 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
           tscResetSqlCmdObj(pCmd, false);
 
           code = tsParseSql(pSql, true);
-
           if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
             return;
           } else if (code != TSDB_CODE_SUCCESS) {

src/client/src/tscLocalMerge.c

@@ -97,6 +97,9 @@ static void tscInitSqlContext(SSqlCmd *pCmd, SLocalReducer *pReducer, tOrderDesc
       pCtx->param[2].i64Key = pQueryInfo->order.order;
       pCtx->param[2].nType  = TSDB_DATA_TYPE_BIGINT;
       pCtx->param[1].i64Key = pQueryInfo->order.orderColId;
+    } else if (functionId == TSDB_FUNC_APERCT) {
+      pCtx->param[0].i64Key = pExpr->param[0].i64Key;
+      pCtx->param[0].nType  = pExpr->param[0].nType;
     }
 
     pCtx->interBufBytes = pExpr->interBytes;

src/client/src/tscParseInsert.c

@@ -1309,7 +1309,7 @@ int tsParseSql(SSqlObj *pSql, bool initial) {
   if ((!pCmd->parseFinished) && (!initial)) {
     tscDebug("%p resume to parse sql: %s", pSql, pCmd->curSql);
   }
-  
+
   ret = tscAllocPayload(&pSql->cmd, TSDB_DEFAULT_PAYLOAD_SIZE);
   if (TSDB_CODE_SUCCESS != ret) {
     return ret;

src/client/src/tscSQLParser.c

@@ -5311,6 +5311,7 @@ static void setCreateDBOption(SCreateDbMsg* pMsg, SCreateDBInfo* pCreateDb) {
   pMsg->replications = pCreateDb->replica;
   pMsg->quorum = pCreateDb->quorum;
   pMsg->ignoreExist = pCreateDb->ignoreExists;
+  pMsg->update = pCreateDb->update;
 }
 
 int32_t parseCreateDBOptions(SSqlCmd* pCmd, SCreateDBInfo* pCreateDbSql) {

src/client/src/tscSubquery.c

@@ -523,7 +523,7 @@ static void quitAllSubquery(SSqlObj* pSqlObj, SJoinSupporter* pSupporter) {
   assert(pSqlObj->subState.numOfRemain > 0);
 
   if (atomic_sub_fetch_32(&pSqlObj->subState.numOfRemain, 1) <= 0) {
-    tscError("%p all subquery return and query failed, global code:%d", pSqlObj, pSqlObj->res.code);
+    tscError("%p all subquery return and query failed, global code:%s", pSqlObj, tstrerror(pSqlObj->res.code));
     freeJoinSubqueryObj(pSqlObj);
   }
 }
@@ -565,7 +565,7 @@ int32_t tagValCompar(const void* p1, const void* p2) {
     return (tag1->len > tag2->len)? 1: -1;
   }
 
-  return strncmp(tag1->data, tag2->data, tag1->len);
+  return memcmp(tag1->data, tag2->data, tag1->len);
 }
 
 void tscBuildVgroupTableInfo(SSqlObj* pSql, STableMetaInfo* pTableMetaInfo, SArray* tables) {

src/common/inc/tdataformat.h

@@ -119,6 +119,33 @@ void      tdResetTSchemaBuilder(STSchemaBuilder *pBuilder, int32_t version);
 int       tdAddColToSchema(STSchemaBuilder *pBuilder, int8_t type, int16_t colId, int16_t bytes);
 STSchema *tdGetSchemaFromBuilder(STSchemaBuilder *pBuilder);
 
+// ----------------- Semantic timestamp key definition
+typedef uint64_t TKEY;
+
+#define TKEY_INVALID UINT64_MAX
+#define TKEY_NULL TKEY_INVALID
+#define TKEY_NEGATIVE_FLAG (((TKEY)1) << 63)
+#define TKEY_DELETE_FLAG (((TKEY)1) << 62)
+#define TKEY_VALUE_FILTER (~(TKEY_NEGATIVE_FLAG | TKEY_DELETE_FLAG))
+
+#define TKEY_IS_NEGATIVE(tkey) (((tkey)&TKEY_NEGATIVE_FLAG) != 0)
+#define TKEY_IS_DELETED(tkey) (((tkey)&TKEY_DELETE_FLAG) != 0)
+#define tdSetTKEYDeleted(tkey) ((tkey) | TKEY_DELETE_FLAG)
+#define tdGetTKEY(key) (((TKEY)ABS(key)) | (TKEY_NEGATIVE_FLAG & (TKEY)(key)))
+#define tdGetKey(tkey) (((TSKEY)((tkey)&TKEY_VALUE_FILTER)) * (TKEY_IS_NEGATIVE(tkey) ? -1 : 1))
+
+static FORCE_INLINE int tkeyComparFn(const void *tkey1, const void *tkey2) {
+  TSKEY key1 = tdGetKey(*(TKEY *)tkey1);
+  TSKEY key2 = tdGetKey(*(TKEY *)tkey2);
+
+  if (key1 < key2) {
+    return -1;
+  } else if (key1 > key2) {
+    return 1;
+  } else {
+    return 0;
+  }
+}
 // ----------------- Data row structure
 
 /* A data row, the format is like below:
@@ -129,6 +156,8 @@ STSchema *tdGetSchemaFromBuilder(STSchemaBuilder *pBuilder);
  * +----------+----------+---------------------------------+---------------------------------+
  * |   len    | sversion |           First part            |             Second part         |
  * +----------+----------+---------------------------------+---------------------------------+
+ * 
+ * NOTE: timestamp in this row structure is TKEY instead of TSKEY
  */
 typedef void *SDataRow;
 
@@ -137,11 +166,13 @@ typedef void *SDataRow;
 #define dataRowLen(r) (*(uint16_t *)(r))
 #define dataRowVersion(r) *(int16_t *)POINTER_SHIFT(r, sizeof(int16_t))
 #define dataRowTuple(r) POINTER_SHIFT(r, TD_DATA_ROW_HEAD_SIZE)
-#define dataRowKey(r) (*(TSKEY *)(dataRowTuple(r)))
+#define dataRowTKey(r) (*(TKEY *)(dataRowTuple(r)))
+#define dataRowKey(r) tdGetKey(dataRowTKey(r))
 #define dataRowSetLen(r, l) (dataRowLen(r) = (l))
 #define dataRowSetVersion(r, v) (dataRowVersion(r) = (v))
 #define dataRowCpy(dst, r) memcpy((dst), (r), dataRowLen(r))
 #define dataRowMaxBytesFromSchema(s) (schemaTLen(s) + TD_DATA_ROW_HEAD_SIZE)
+#define dataRowDeleted(r) TKEY_IS_DELETED(dataRowTKey(r))
 
 SDataRow tdNewDataRowFromSchema(STSchema *pSchema);
 void     tdFreeDataRow(SDataRow row);
@@ -154,16 +185,18 @@ static FORCE_INLINE int tdAppendColVal(SDataRow row, void *value, int8_t type, i
   int32_t toffset = offset + TD_DATA_ROW_HEAD_SIZE;
   char *  ptr = (char *)POINTER_SHIFT(row, dataRowLen(row));
 
-  switch (type) {
-    case TSDB_DATA_TYPE_BINARY:
-    case TSDB_DATA_TYPE_NCHAR:
-      *(VarDataOffsetT *)POINTER_SHIFT(row, toffset) = dataRowLen(row);
-      memcpy(ptr, value, varDataTLen(value));
-      dataRowLen(row) += varDataTLen(value);
-      break;
-    default:
+  if (IS_VAR_DATA_TYPE(type)) {
+    *(VarDataOffsetT *)POINTER_SHIFT(row, toffset) = dataRowLen(row);
+    memcpy(ptr, value, varDataTLen(value));
+    dataRowLen(row) += varDataTLen(value);
+  } else {
+    if (offset == 0) {
+      ASSERT(type == TSDB_DATA_TYPE_TIMESTAMP);
+      TKEY tvalue = tdGetTKEY(*(TSKEY *)value);
+      memcpy(POINTER_SHIFT(row, toffset), (void *)(&tvalue), TYPE_BYTES[type]);
+    } else {
       memcpy(POINTER_SHIFT(row, toffset), value, TYPE_BYTES[type]);
-      break;
+    }
   }
 
   return 0;
@@ -171,12 +204,10 @@ static FORCE_INLINE int tdAppendColVal(SDataRow row, void *value, int8_t type, i
 
 // NOTE: offset here including the header size
 static FORCE_INLINE void *tdGetRowDataOfCol(SDataRow row, int8_t type, int32_t offset) {
-  switch (type) {
-    case TSDB_DATA_TYPE_BINARY:
-    case TSDB_DATA_TYPE_NCHAR:
-      return POINTER_SHIFT(row, *(VarDataOffsetT *)POINTER_SHIFT(row, offset));
-    default:
-      return POINTER_SHIFT(row, offset);
+  if (IS_VAR_DATA_TYPE(type)) {
+    return POINTER_SHIFT(row, *(VarDataOffsetT *)POINTER_SHIFT(row, offset));
+  } else {
+    return POINTER_SHIFT(row, offset);
   }
 }
 
@@ -196,7 +227,6 @@ static FORCE_INLINE void dataColReset(SDataCol *pDataCol) { pDataCol->len = 0; }
 
 void dataColInit(SDataCol *pDataCol, STColumn *pCol, void **pBuf, int maxPoints);
 void dataColAppendVal(SDataCol *pCol, void *value, int numOfRows, int maxPoints);
-void dataColPopPoints(SDataCol *pCol, int pointsToPop, int numOfRows);
 void dataColSetOffset(SDataCol *pCol, int nEle);
 
 bool isNEleNull(SDataCol *pCol, int nEle);
@@ -204,28 +234,20 @@ void dataColSetNEleNull(SDataCol *pCol, int nEle, int maxPoints);
 
 // Get the data pointer from a column-wised data
 static FORCE_INLINE void *tdGetColDataOfRow(SDataCol *pCol, int row) {
-  switch (pCol->type) {
-    case TSDB_DATA_TYPE_BINARY:
-    case TSDB_DATA_TYPE_NCHAR:
-      return POINTER_SHIFT(pCol->pData, pCol->dataOff[row]);
-      break;
-
-    default:
-      return POINTER_SHIFT(pCol->pData, TYPE_BYTES[pCol->type] * row);
-      break;
+  if (IS_VAR_DATA_TYPE(pCol->type)) {
+    return POINTER_SHIFT(pCol->pData, pCol->dataOff[row]);
+  } else {
+    return POINTER_SHIFT(pCol->pData, TYPE_BYTES[pCol->type] * row);
   }
 }
 
 static FORCE_INLINE int32_t dataColGetNEleLen(SDataCol *pDataCol, int rows) {
   ASSERT(rows > 0);
 
-  switch (pDataCol->type) {
-    case TSDB_DATA_TYPE_BINARY:
-    case TSDB_DATA_TYPE_NCHAR:
-      return pDataCol->dataOff[rows - 1] + varDataTLen(tdGetColDataOfRow(pDataCol, rows - 1));
-      break;
-    default:
-      return TYPE_BYTES[pDataCol->type] * rows;
+  if (IS_VAR_DATA_TYPE(pDataCol->type)) {
+    return pDataCol->dataOff[rows - 1] + varDataTLen(tdGetColDataOfRow(pDataCol, rows - 1));
+  } else {
+    return TYPE_BYTES[pDataCol->type] * rows;
   }
 }
 
@@ -243,9 +265,14 @@ typedef struct {
 } SDataCols;
 
 #define keyCol(pCols) (&((pCols)->cols[0]))  // Key column
-#define dataColsKeyAt(pCols, idx) ((TSKEY *)(keyCol(pCols)->pData))[(idx)]
-#define dataColsKeyFirst(pCols) dataColsKeyAt(pCols, 0)
-#define dataColsKeyLast(pCols) ((pCols->numOfRows == 0) ? 0 : dataColsKeyAt(pCols, (pCols)->numOfRows - 1))
+#define dataColsTKeyAt(pCols, idx) ((TKEY *)(keyCol(pCols)->pData))[(idx)]
+#define dataColsKeyAt(pCols, idx) tdGetKey(dataColsTKeyAt(pCols, idx))
+#define dataColsTKeyFirst(pCols) (((pCols)->numOfRows == 0) ? TKEY_INVALID : dataColsTKeyAt(pCols, 0))
+#define dataColsKeyFirst(pCols) (((pCols)->numOfRows == 0) ? TSDB_DATA_TIMESTAMP_NULL : dataColsKeyAt(pCols, 0))
+#define dataColsTKeyLast(pCols) \
+  (((pCols)->numOfRows == 0) ? TKEY_INVALID : dataColsTKeyAt(pCols, (pCols)->numOfRows - 1))
+#define dataColsKeyLast(pCols) \
+  (((pCols)->numOfRows == 0) ? TSDB_DATA_TIMESTAMP_NULL : dataColsKeyAt(pCols, (pCols)->numOfRows - 1))
 
 SDataCols *tdNewDataCols(int maxRowSize, int maxCols, int maxRows);
 void       tdResetDataCols(SDataCols *pCols);
@@ -253,10 +280,7 @@ int        tdInitDataCols(SDataCols *pCols, STSchema *pSchema);
 SDataCols *tdDupDataCols(SDataCols *pCols, bool keepData);
 void       tdFreeDataCols(SDataCols *pCols);
 void       tdAppendDataRowToDataCol(SDataRow row, STSchema *pSchema, SDataCols *pCols);
-void       tdPopDataColsPoints(SDataCols *pCols, int pointsToPop);  //!!!!
 int        tdMergeDataCols(SDataCols *target, SDataCols *src, int rowsToMerge);
-void       tdMergeTwoDataCols(SDataCols *target, SDataCols *src1, int *iter1, int limit1, SDataCols *src2, int *iter2,
-                              int limit2, int tRows);
 
 // ----------------- K-V data row structure
 /*

src/common/inc/tglobal.h

@@ -88,6 +88,7 @@ extern int16_t tsWAL;
 extern int32_t tsFsyncPeriod;
 extern int32_t tsReplications;
 extern int32_t tsQuorum;
+extern int32_t tsUpdate;
 
 // balance
 extern int32_t tsEnableBalance;

src/common/src/tdataformat.c

@@ -18,6 +18,9 @@
 #include "tcoding.h"
 #include "wchar.h"
 
+static void tdMergeTwoDataCols(SDataCols *target, SDataCols *src1, int *iter1, int limit1, SDataCols *src2, int *iter2,
+                               int limit2, int tRows);
+
 /**
  * Duplicate the schema and return a new object
  */
@@ -202,7 +205,7 @@ void dataColInit(SDataCol *pDataCol, STColumn *pCol, void **pBuf, int maxPoints)
   pDataCol->offset = colOffset(pCol) + TD_DATA_ROW_HEAD_SIZE;
 
   pDataCol->len = 0;
-  if (pDataCol->type == TSDB_DATA_TYPE_BINARY || pDataCol->type == TSDB_DATA_TYPE_NCHAR) {
+  if (IS_VAR_DATA_TYPE(pDataCol->type)) {
     pDataCol->dataOff = (VarDataOffsetT *)(*pBuf);
     pDataCol->pData = POINTER_SHIFT(*pBuf, sizeof(VarDataOffsetT) * maxPoints);
     pDataCol->spaceSize = pDataCol->bytes * maxPoints;
@@ -215,60 +218,29 @@ void dataColInit(SDataCol *pDataCol, STColumn *pCol, void **pBuf, int maxPoints)
   }
 }
 
+// value from timestamp should be TKEY here instead of TSKEY
 void dataColAppendVal(SDataCol *pCol, void *value, int numOfRows, int maxPoints) {
   ASSERT(pCol != NULL && value != NULL);
 
-  switch (pCol->type) {
-    case TSDB_DATA_TYPE_BINARY:
-    case TSDB_DATA_TYPE_NCHAR:
-      // set offset
-      pCol->dataOff[numOfRows] = pCol->len;
-      // Copy data
-      memcpy(POINTER_SHIFT(pCol->pData, pCol->len), value, varDataTLen(value));
-      // Update the length
-      pCol->len += varDataTLen(value);
-      break;
-    default:
-      ASSERT(pCol->len == TYPE_BYTES[pCol->type] * numOfRows);
-      memcpy(POINTER_SHIFT(pCol->pData, pCol->len), value, pCol->bytes);
-      pCol->len += pCol->bytes;
-      break;
-  }
-}
-
-void dataColPopPoints(SDataCol *pCol, int pointsToPop, int numOfRows) {
-  int pointsLeft = numOfRows - pointsToPop;
-
-  ASSERT(pointsLeft > 0);
-
-  if (pCol->type == TSDB_DATA_TYPE_BINARY || pCol->type == TSDB_DATA_TYPE_NCHAR) {
-    ASSERT(pCol->len > 0);
-    VarDataOffsetT toffset = pCol->dataOff[pointsToPop];
-    pCol->len = pCol->len - toffset;
-    ASSERT(pCol->len > 0);
-    memmove(pCol->pData, POINTER_SHIFT(pCol->pData, toffset), pCol->len);
-    dataColSetOffset(pCol, pointsLeft);
+  if (IS_VAR_DATA_TYPE(pCol->type)) {
+    // set offset
+    pCol->dataOff[numOfRows] = pCol->len;
+    // Copy data
+    memcpy(POINTER_SHIFT(pCol->pData, pCol->len), value, varDataTLen(value));
+    // Update the length
+    pCol->len += varDataTLen(value);
   } else {
     ASSERT(pCol->len == TYPE_BYTES[pCol->type] * numOfRows);
-    pCol->len = TYPE_BYTES[pCol->type] * pointsLeft;
-    memmove(pCol->pData, POINTER_SHIFT(pCol->pData, TYPE_BYTES[pCol->type] * pointsToPop), pCol->len);
+    memcpy(POINTER_SHIFT(pCol->pData, pCol->len), value, pCol->bytes);
+    pCol->len += pCol->bytes;
   }
 }
 
 bool isNEleNull(SDataCol *pCol, int nEle) {
-  switch (pCol->type) {
-    case TSDB_DATA_TYPE_BINARY:
-    case TSDB_DATA_TYPE_NCHAR:
-      for (int i = 0; i < nEle; i++) {
-        if (!isNull(tdGetColDataOfRow(pCol, i), pCol->type)) return false;
-      }
-      return true;
-    default:
-      for (int i = 0; i < nEle; i++) {
-        if (!isNull(tdGetColDataOfRow(pCol, i), pCol->type)) return false;
-      }
-      return true;
+  for (int i = 0; i < nEle; i++) {
+    if (!isNull(tdGetColDataOfRow(pCol, i), pCol->type)) return false;
   }
+  return true;
 }
 
 void dataColSetNullAt(SDataCol *pCol, int index) {
@@ -390,7 +362,7 @@ SDataCols *tdDupDataCols(SDataCols *pDataCols, bool keepData) {
     pRet->cols[i].spaceSize = pDataCols->cols[i].spaceSize;
     pRet->cols[i].pData = (void *)((char *)pRet->buf + ((char *)(pDataCols->cols[i].pData) - (char *)(pDataCols->buf)));
 
-    if (pRet->cols[i].type == TSDB_DATA_TYPE_BINARY || pRet->cols[i].type == TSDB_DATA_TYPE_NCHAR) {
+    if (IS_VAR_DATA_TYPE(pRet->cols[i].type)) {
       ASSERT(pDataCols->cols[i].dataOff != NULL);
       pRet->cols[i].dataOff =
           (int32_t *)((char *)pRet->buf + ((char *)(pDataCols->cols[i].dataOff) - (char *)(pDataCols->buf)));
@@ -400,7 +372,7 @@ SDataCols *tdDupDataCols(SDataCols *pDataCols, bool keepData) {
       pRet->cols[i].len = pDataCols->cols[i].len;
       if (pDataCols->cols[i].len > 0) {
         memcpy(pRet->cols[i].pData, pDataCols->cols[i].pData, pDataCols->cols[i].len);
-        if (pRet->cols[i].type == TSDB_DATA_TYPE_BINARY || pRet->cols[i].type == TSDB_DATA_TYPE_NCHAR) {
+        if (IS_VAR_DATA_TYPE(pRet->cols[i].type)) {
           memcpy(pRet->cols[i].dataOff, pDataCols->cols[i].dataOff, sizeof(VarDataOffsetT) * pDataCols->maxPoints);
         }
       }
@@ -420,58 +392,54 @@ void tdResetDataCols(SDataCols *pCols) {
 }
 
 void tdAppendDataRowToDataCol(SDataRow row, STSchema *pSchema, SDataCols *pCols) {
-  ASSERT(dataColsKeyLast(pCols) < dataRowKey(row));
+  ASSERT(pCols->numOfRows == 0 || dataColsKeyLast(pCols) < dataRowKey(row));
 
   int rcol = 0;
   int dcol = 0;
 
-  while (dcol < pCols->numOfCols) {
-    SDataCol *pDataCol = &(pCols->cols[dcol]);
-    if (rcol >= schemaNCols(pSchema)) {
-      dataColSetNullAt(pDataCol, pCols->numOfRows);
-      dcol++;
-      continue;
+  if (dataRowDeleted(row)) {
+    for (; dcol < pCols->numOfCols; dcol++) {
+      SDataCol *pDataCol = &(pCols->cols[dcol]);
+      if (dcol == 0) {
+        dataColAppendVal(pDataCol, dataRowTuple(row), pCols->numOfRows, pCols->maxPoints);
+      } else {
+        dataColSetNullAt(pDataCol, pCols->numOfRows);
+      }
     }
+  } else {
+    while (dcol < pCols->numOfCols) {
+      SDataCol *pDataCol = &(pCols->cols[dcol]);
+      if (rcol >= schemaNCols(pSchema)) {
+        dataColSetNullAt(pDataCol, pCols->numOfRows);
+        dcol++;
+        continue;
+      }
 
-    STColumn *pRowCol = schemaColAt(pSchema, rcol);
-    if (pRowCol->colId == pDataCol->colId) {
-      void *value = tdGetRowDataOfCol(row, pRowCol->type, pRowCol->offset+TD_DATA_ROW_HEAD_SIZE);
-      dataColAppendVal(pDataCol, value, pCols->numOfRows, pCols->maxPoints);
-      dcol++;
-      rcol++;
-    } else if (pRowCol->colId < pDataCol->colId) {
-      rcol++;
-    } else {
-      dataColSetNullAt(pDataCol, pCols->numOfRows);
-      dcol++;
+      STColumn *pRowCol = schemaColAt(pSchema, rcol);
+      if (pRowCol->colId == pDataCol->colId) {
+        void *value = tdGetRowDataOfCol(row, pRowCol->type, pRowCol->offset + TD_DATA_ROW_HEAD_SIZE);
+        dataColAppendVal(pDataCol, value, pCols->numOfRows, pCols->maxPoints);
+        dcol++;
+        rcol++;
+      } else if (pRowCol->colId < pDataCol->colId) {
+        rcol++;
+      } else {
+        dataColSetNullAt(pDataCol, pCols->numOfRows);
+        dcol++;
+      }
     }
   }
   pCols->numOfRows++;
 }
 
-// Pop pointsToPop points from the SDataCols
-void tdPopDataColsPoints(SDataCols *pCols, int pointsToPop) {
-  int pointsLeft = pCols->numOfRows - pointsToPop;
-  if (pointsLeft <= 0) {
-    tdResetDataCols(pCols);
-    return;
-  }
-
-  for (int iCol = 0; iCol < pCols->numOfCols; iCol++) {
-    SDataCol *pCol = pCols->cols + iCol;
-    dataColPopPoints(pCol, pointsToPop, pCols->numOfRows);
-  }
-  pCols->numOfRows = pointsLeft;
-}
-
 int tdMergeDataCols(SDataCols *target, SDataCols *source, int rowsToMerge) {
   ASSERT(rowsToMerge > 0 && rowsToMerge <= source->numOfRows);
-  ASSERT(target->numOfRows + rowsToMerge <= target->maxPoints);
   ASSERT(target->numOfCols == source->numOfCols);
 
   SDataCols *pTarget = NULL;
 
   if (dataColsKeyLast(target) < dataColsKeyFirst(source)) {  // No overlap
+    ASSERT(target->numOfRows + rowsToMerge <= target->maxPoints);
     for (int i = 0; i < rowsToMerge; i++) {
       for (int j = 0; j < source->numOfCols; j++) {
         if (source->cols[j].len > 0) {
@@ -499,17 +467,23 @@ _err:
   return -1;
 }
 
-void tdMergeTwoDataCols(SDataCols *target, SDataCols *src1, int *iter1, int limit1, SDataCols *src2, int *iter2, int limit2, int tRows) {
+// src2 data has more priority than src1
+static void tdMergeTwoDataCols(SDataCols *target, SDataCols *src1, int *iter1, int limit1, SDataCols *src2, int *iter2,
+                               int limit2, int tRows) {
   tdResetDataCols(target);
   ASSERT(limit1 <= src1->numOfRows && limit2 <= src2->numOfRows);
 
   while (target->numOfRows < tRows) {
     if (*iter1 >= limit1 && *iter2 >= limit2) break;
 
-    TSKEY key1 = (*iter1 >= limit1) ? INT64_MAX : ((TSKEY *)(src1->cols[0].pData))[*iter1];
-    TSKEY key2 = (*iter2 >= limit2) ? INT64_MAX : ((TSKEY *)(src2->cols[0].pData))[*iter2];
+    TSKEY key1 = (*iter1 >= limit1) ? INT64_MAX : dataColsKeyAt(src1, *iter1);
+    TKEY  tkey1 = (*iter1 >= limit1) ? TKEY_NULL : dataColsTKeyAt(src1, *iter1);
+    TSKEY key2 = (*iter2 >= limit2) ? INT64_MAX : dataColsKeyAt(src2, *iter2);
+    TKEY  tkey2 = (*iter2 >= limit2) ? TKEY_NULL : dataColsTKeyAt(src2, *iter2);
 
-    if (key1 <= key2) {
+    ASSERT(tkey1 == TKEY_NULL || (!TKEY_IS_DELETED(tkey1)));
+
+    if (key1 < key2) {
       for (int i = 0; i < src1->numOfCols; i++) {
         ASSERT(target->cols[i].type == src1->cols[i].type);
         if (src1->cols[i].len > 0) {
@@ -520,19 +494,23 @@ void tdMergeTwoDataCols(SDataCols *target, SDataCols *src1, int *iter1, int limi
 
       target->numOfRows++;
       (*iter1)++;
-      if (key1 == key2) (*iter2)++;
-    } else {
-      for (int i = 0; i < src2->numOfCols; i++) {
-        ASSERT(target->cols[i].type == src2->cols[i].type);
-        if (src2->cols[i].len > 0) {
-          dataColAppendVal(&(target->cols[i]), tdGetColDataOfRow(src2->cols + i, *iter2), target->numOfRows,
-                           target->maxPoints);
+    } else if (key1 >= key2) {
+      if ((key1 > key2) || (key1 == key2 && !TKEY_IS_DELETED(tkey2))) {
+        for (int i = 0; i < src2->numOfCols; i++) {
+          ASSERT(target->cols[i].type == src2->cols[i].type);
+          if (src2->cols[i].len > 0) {
+            dataColAppendVal(&(target->cols[i]), tdGetColDataOfRow(src2->cols + i, *iter2), target->numOfRows,
+                             target->maxPoints);
+          }
         }
+        target->numOfRows++;
       }
 
-      target->numOfRows++;
       (*iter2)++;
+      if (key1 == key2) (*iter1)++;
     }
+
+    ASSERT(target->numOfRows <= target->maxPoints);
   }
 }
 

src/common/src/tglobal.c

@@ -120,6 +120,7 @@ int16_t tsWAL           = TSDB_DEFAULT_WAL_LEVEL;
 int32_t tsFsyncPeriod   = TSDB_DEFAULT_FSYNC_PERIOD;
 int32_t tsReplications  = TSDB_DEFAULT_DB_REPLICA_OPTION;
 int32_t tsQuorum        = TSDB_DEFAULT_DB_QUORUM_OPTION;
+int32_t tsUpdate        = TSDB_DEFAULT_DB_UPDATE_OPTION;
 int32_t tsMaxVgroupsPerDb  = 0;
 int32_t tsMinTablePerVnode = TSDB_TABLES_STEP;
 int32_t tsMaxTablePerVnode = TSDB_DEFAULT_TABLES;
@@ -786,6 +787,16 @@ static void doInitGlobalConfig(void) {
   cfg.unitType = TAOS_CFG_UTYPE_NONE;
   taosInitConfigOption(cfg);
 
+  cfg.option = "update";
+  cfg.ptr = &tsUpdate;
+  cfg.valType = TAOS_CFG_VTYPE_INT32;
+  cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW;
+  cfg.minValue = TSDB_MIN_DB_UPDATE;
+  cfg.maxValue = TSDB_MAX_DB_UPDATE;
+  cfg.ptrLength = 0;
+  cfg.unitType = TAOS_CFG_UTYPE_NONE;
+  taosInitConfigOption(cfg);
+
   cfg.option = "mqttHostName";
   cfg.ptr = tsMqttHostName;
   cfg.valType = TAOS_CFG_VTYPE_STRING;

src/dnode/src/dnodeVRead.c

@@ -92,33 +92,23 @@ void dnodeDispatchToVReadQueue(SRpcMsg *pMsg) {
     pHead->vgId = htonl(pHead->vgId);
     pHead->contLen = htonl(pHead->contLen);
 
-    taos_queue queue = vnodeAcquireRqueue(pHead->vgId);
-
-    if (queue == NULL) {
-      leftLen -= pHead->contLen;
-      pCont -= pHead->contLen;
-      continue;
+    void *pVnode = vnodeAcquire(pHead->vgId);
+    if (pVnode != NULL) {
+      int32_t code = vnodeWriteToRQueue(pVnode, pCont, pHead->contLen, TAOS_QTYPE_RPC, pMsg);
+      if (code == TSDB_CODE_SUCCESS) queuedMsgNum++;
+      vnodeRelease(pVnode);
     }
 
-    // put message into queue
-    SVReadMsg *pRead = taosAllocateQitem(sizeof(SVReadMsg));
-    pRead->rpcMsg = *pMsg;
-    pRead->pCont = pCont;
-    pRead->contLen = pHead->contLen;
-
-    // next vnode
     leftLen -= pHead->contLen;
     pCont -= pHead->contLen;
-    queuedMsgNum++;
-
-    taosWriteQitem(queue, TAOS_QTYPE_RPC, pRead);
   }
 
   if (queuedMsgNum == 0) {
     SRpcMsg rpcRsp = {.handle = pMsg->handle, .code = TSDB_CODE_VND_INVALID_VGROUP_ID};
     rpcSendResponse(&rpcRsp);
-    rpcFreeCont(pMsg->pCont);
   }
+
+  rpcFreeCont(pMsg->pCont);
 }
 
 void *dnodeAllocVReadQueue(void *pVnode) {
@@ -162,50 +152,48 @@ void dnodeFreeVReadQueue(void *rqueue) {
 
 void dnodeSendRpcVReadRsp(void *pVnode, SVReadMsg *pRead, int32_t code) {
   SRpcMsg rpcRsp = {
-    .handle  = pRead->rpcMsg.handle,
+    .handle  = pRead->rpcHandle,
     .pCont   = pRead->rspRet.rsp,
     .contLen = pRead->rspRet.len,
     .code    = code,
   };
 
   rpcSendResponse(&rpcRsp);
-  rpcFreeCont(pRead->rpcMsg.pCont);
   vnodeRelease(pVnode);
 }
 
 void dnodeDispatchNonRspMsg(void *pVnode, SVReadMsg *pRead, int32_t code) {
-  rpcFreeCont(pRead->rpcMsg.pCont);
   vnodeRelease(pVnode);
 }
 
 static void *dnodeProcessReadQueue(void *param) {
-  SVReadMsg *pReadMsg;
+  SVReadMsg *pRead;
   int32_t    qtype;
   void *     pVnode;
 
   while (1) {
-    if (taosReadQitemFromQset(tsVReadQset, &qtype, (void **)&pReadMsg, &pVnode) == 0) {
+    if (taosReadQitemFromQset(tsVReadQset, &qtype, (void **)&pRead, &pVnode) == 0) {
       dDebug("qset:%p dnode vread got no message from qset, exiting", tsVReadQset);
       break;
     }
 
-    dDebug("%p, msg:%s will be processed in vread queue, qtype:%d, msg:%p", pReadMsg->rpcMsg.ahandle,
-           taosMsg[pReadMsg->rpcMsg.msgType], qtype, pReadMsg);
+    dDebug("%p, msg:%p:%s will be processed in vread queue, qtype:%d", pRead->rpcAhandle, pRead,
+           taosMsg[pRead->msgType], qtype);
 
-    int32_t code = vnodeProcessRead(pVnode, pReadMsg);
+    int32_t code = vnodeProcessRead(pVnode, pRead);
 
     if (qtype == TAOS_QTYPE_RPC && code != TSDB_CODE_QRY_NOT_READY) {
-      dnodeSendRpcVReadRsp(pVnode, pReadMsg, code);
+      dnodeSendRpcVReadRsp(pVnode, pRead, code);
     } else {
       if (code == TSDB_CODE_QRY_HAS_RSP) {
-        dnodeSendRpcVReadRsp(pVnode, pReadMsg, pReadMsg->rpcMsg.code);
+        dnodeSendRpcVReadRsp(pVnode, pRead, pRead->code);
       } else {  // code == TSDB_CODE_QRY_NOT_READY, do not return msg to client
-        assert(pReadMsg->rpcMsg.handle == NULL || (pReadMsg->rpcMsg.handle != NULL && pReadMsg->rpcMsg.msgType == 5));
-        dnodeDispatchNonRspMsg(pVnode, pReadMsg, code);
+        assert(pRead->rpcHandle == NULL || (pRead->rpcHandle != NULL && pRead->msgType == 5));
+        dnodeDispatchNonRspMsg(pVnode, pRead, code);
       }
     }
 
-    taosFreeQitem(pReadMsg);
+    taosFreeQitem(pRead);
   }
 
   return NULL;

src/inc/taosdef.h

@@ -75,6 +75,7 @@ extern const int32_t TYPE_BYTES[11];
 #define TSDB_DATA_SMALLINT_NULL         0x8000
 #define TSDB_DATA_INT_NULL              0x80000000L
 #define TSDB_DATA_BIGINT_NULL           0x8000000000000000L
+#define TSDB_DATA_TIMESTAMP_NULL        TSDB_DATA_BIGINT_NULL
 
 #define TSDB_DATA_FLOAT_NULL            0x7FF00000              // it is an NAN
 #define TSDB_DATA_DOUBLE_NULL           0x7FFFFF0000000000L     // an NAN
@@ -363,6 +364,10 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size, void* buf
 #define TSDB_MAX_WAL_LEVEL              2
 #define TSDB_DEFAULT_WAL_LEVEL          1
 
+#define TSDB_MIN_DB_UPDATE              0
+#define TSDB_MAX_DB_UPDATE              1
+#define TSDB_DEFAULT_DB_UPDATE_OPTION   0
+
 #define TSDB_MIN_FSYNC_PERIOD           0
 #define TSDB_MAX_FSYNC_PERIOD           180000   // millisecond
 #define TSDB_DEFAULT_FSYNC_PERIOD       3000     // three second

src/inc/taosmsg.h

@@ -544,6 +544,8 @@ typedef struct {
   int8_t   replications;
   int8_t   quorum;
   int8_t   ignoreExist;
+  int8_t   update;
+  int8_t   reserve[9];
 } SCreateDbMsg, SAlterDbMsg;
 
 typedef struct {
@@ -654,7 +656,8 @@ typedef struct {
   int8_t   replications;
   int8_t   wals;
   int8_t   quorum;
-  int8_t   reserved[16];
+  int8_t   update;
+  int8_t   reserved[15];
 } SVnodeCfg;
 
 typedef struct {

src/inc/tsdb.h

@@ -65,6 +65,7 @@ typedef struct {
   int32_t maxRowsPerFileBlock;  // maximum rows per file block
   int8_t  precision;
   int8_t  compression;
+  int8_t  update;
 } STsdbCfg;
 
 // --------- TSDB REPOSITORY USAGE STATISTICS

src/inc/ttokendef.h

@@ -114,114 +114,115 @@
 #define TK_FSYNC                           95
 #define TK_COMP                            96
 #define TK_PRECISION                       97
-#define TK_LP                              98
-#define TK_RP                              99
-#define TK_TAGS                           100
-#define TK_USING                          101
-#define TK_AS                             102
-#define TK_COMMA                          103
-#define TK_NULL                           104
-#define TK_SELECT                         105
-#define TK_UNION                          106
-#define TK_ALL                            107
-#define TK_FROM                           108
-#define TK_VARIABLE                       109
-#define TK_INTERVAL                       110
-#define TK_FILL                           111
-#define TK_SLIDING                        112
-#define TK_ORDER                          113
-#define TK_BY                             114
-#define TK_ASC                            115
-#define TK_DESC                           116
-#define TK_GROUP                          117
-#define TK_HAVING                         118
-#define TK_LIMIT                          119
-#define TK_OFFSET                         120
-#define TK_SLIMIT                         121
-#define TK_SOFFSET                        122
-#define TK_WHERE                          123
-#define TK_NOW                            124
-#define TK_RESET                          125
-#define TK_QUERY                          126
-#define TK_ADD                            127
-#define TK_COLUMN                         128
-#define TK_TAG                            129
-#define TK_CHANGE                         130
-#define TK_SET                            131
-#define TK_KILL                           132
-#define TK_CONNECTION                     133
-#define TK_STREAM                         134
-#define TK_COLON                          135
-#define TK_ABORT                          136
-#define TK_AFTER                          137
-#define TK_ATTACH                         138
-#define TK_BEFORE                         139
-#define TK_BEGIN                          140
-#define TK_CASCADE                        141
-#define TK_CLUSTER                        142
-#define TK_CONFLICT                       143
-#define TK_COPY                           144
-#define TK_DEFERRED                       145
-#define TK_DELIMITERS                     146
-#define TK_DETACH                         147
-#define TK_EACH                           148
-#define TK_END                            149
-#define TK_EXPLAIN                        150
-#define TK_FAIL                           151
-#define TK_FOR                            152
-#define TK_IGNORE                         153
-#define TK_IMMEDIATE                      154
-#define TK_INITIALLY                      155
-#define TK_INSTEAD                        156
-#define TK_MATCH                          157
-#define TK_KEY                            158
-#define TK_OF                             159
-#define TK_RAISE                          160
-#define TK_REPLACE                        161
-#define TK_RESTRICT                       162
-#define TK_ROW                            163
-#define TK_STATEMENT                      164
-#define TK_TRIGGER                        165
-#define TK_VIEW                           166
-#define TK_COUNT                          167
-#define TK_SUM                            168
-#define TK_AVG                            169
-#define TK_MIN                            170
-#define TK_MAX                            171
-#define TK_FIRST                          172
-#define TK_LAST                           173
-#define TK_TOP                            174
-#define TK_BOTTOM                         175
-#define TK_STDDEV                         176
-#define TK_PERCENTILE                     177
-#define TK_APERCENTILE                    178
-#define TK_LEASTSQUARES                   179
-#define TK_HISTOGRAM                      180
-#define TK_DIFF                           181
-#define TK_SPREAD                         182
-#define TK_TWA                            183
-#define TK_INTERP                         184
-#define TK_LAST_ROW                       185
-#define TK_RATE                           186
-#define TK_IRATE                          187
-#define TK_SUM_RATE                       188
-#define TK_SUM_IRATE                      189
-#define TK_AVG_RATE                       190
-#define TK_AVG_IRATE                      191
-#define TK_TBID                           192
-#define TK_SEMI                           193
-#define TK_NONE                           194
-#define TK_PREV                           195
-#define TK_LINEAR                         196
-#define TK_IMPORT                         197
-#define TK_METRIC                         198
-#define TK_TBNAME                         199
-#define TK_JOIN                           200
-#define TK_METRICS                        201
-#define TK_STABLE                         202
-#define TK_INSERT                         203
-#define TK_INTO                           204
-#define TK_VALUES                         205
+#define TK_UPDATE                          98
+#define TK_LP                              99
+#define TK_RP                             100
+#define TK_TAGS                           101
+#define TK_USING                          102
+#define TK_AS                             103
+#define TK_COMMA                          104
+#define TK_NULL                           105
+#define TK_SELECT                         106
+#define TK_UNION                          107
+#define TK_ALL                            108
+#define TK_FROM                           109
+#define TK_VARIABLE                       110
+#define TK_INTERVAL                       111
+#define TK_FILL                           112
+#define TK_SLIDING                        113
+#define TK_ORDER                          114
+#define TK_BY                             115
+#define TK_ASC                            116
+#define TK_DESC                           117
+#define TK_GROUP                          118
+#define TK_HAVING                         119
+#define TK_LIMIT                          120
+#define TK_OFFSET                         121
+#define TK_SLIMIT                         122
+#define TK_SOFFSET                        123
+#define TK_WHERE                          124
+#define TK_NOW                            125
+#define TK_RESET                          126
+#define TK_QUERY                          127
+#define TK_ADD                            128
+#define TK_COLUMN                         129
+#define TK_TAG                            130
+#define TK_CHANGE                         131
+#define TK_SET                            132
+#define TK_KILL                           133
+#define TK_CONNECTION                     134
+#define TK_STREAM                         135
+#define TK_COLON                          136
+#define TK_ABORT                          137
+#define TK_AFTER                          138
+#define TK_ATTACH                         139
+#define TK_BEFORE                         140
+#define TK_BEGIN                          141
+#define TK_CASCADE                        142
+#define TK_CLUSTER                        143
+#define TK_CONFLICT                       144
+#define TK_COPY                           145
+#define TK_DEFERRED                       146
+#define TK_DELIMITERS                     147
+#define TK_DETACH                         148
+#define TK_EACH                           149
+#define TK_END                            150
+#define TK_EXPLAIN                        151
+#define TK_FAIL                           152
+#define TK_FOR                            153
+#define TK_IGNORE                         154
+#define TK_IMMEDIATE                      155
+#define TK_INITIALLY                      156
+#define TK_INSTEAD                        157
+#define TK_MATCH                          158
+#define TK_KEY                            159
+#define TK_OF                             160
+#define TK_RAISE                          161
+#define TK_REPLACE                        162
+#define TK_RESTRICT                       163
+#define TK_ROW                            164
+#define TK_STATEMENT                      165
+#define TK_TRIGGER                        166
+#define TK_VIEW                           167
+#define TK_COUNT                          168
+#define TK_SUM                            169
+#define TK_AVG                            170
+#define TK_MIN                            171
+#define TK_MAX                            172
+#define TK_FIRST                          173
+#define TK_LAST                           174
+#define TK_TOP                            175
+#define TK_BOTTOM                         176
+#define TK_STDDEV                         177
+#define TK_PERCENTILE                     178
+#define TK_APERCENTILE                    179
+#define TK_LEASTSQUARES                   180
+#define TK_HISTOGRAM                      181
+#define TK_DIFF                           182
+#define TK_SPREAD                         183
+#define TK_TWA                            184
+#define TK_INTERP                         185
+#define TK_LAST_ROW                       186
+#define TK_RATE                           187
+#define TK_IRATE                          188
+#define TK_SUM_RATE                       189
+#define TK_SUM_IRATE                      190
+#define TK_AVG_RATE                       191
+#define TK_AVG_IRATE                      192
+#define TK_TBID                           193
+#define TK_SEMI                           194
+#define TK_NONE                           195
+#define TK_PREV                           196
+#define TK_LINEAR                         197
+#define TK_IMPORT                         198
+#define TK_METRIC                         199
+#define TK_TBNAME                         200
+#define TK_JOIN                           201
+#define TK_METRICS                        202
+#define TK_STABLE                         203
+#define TK_INSERT                         204
+#define TK_INTO                           205
+#define TK_VALUES                         206
 
 #define TK_SPACE                          300
 #define TK_COMMENT                        301

src/inc/vnode.h

@@ -37,10 +37,15 @@ typedef struct {
 } SRspRet;
 
 typedef struct {
-  SRspRet rspRet;
-  void *  pCont;
+  int32_t code;
   int32_t contLen;
-  SRpcMsg rpcMsg;
+  void *  rpcHandle;
+  void *  rpcAhandle;
+  void *  qhandle;
+  int8_t  qtype;
+  int8_t  msgType;
+  SRspRet rspRet;
+  char    pCont[];
 } SVReadMsg;
 
 typedef struct {
@@ -62,13 +67,11 @@ int32_t vnodeAlter(void *pVnode, SCreateVnodeMsg *pVnodeCfg);
 int32_t vnodeClose(int32_t vgId);
 
 void*   vnodeAcquire(int32_t vgId);        // add refcount
-void*   vnodeAcquireRqueue(int32_t vgId);  // add refCount, get read queue 
 void    vnodeRelease(void *pVnode);        // dec refCount
 void*   vnodeGetWal(void *pVnode);
 
 int32_t vnodeWriteToWQueue(void *vparam, void *wparam, int32_t qtype, void *rparam);
 int32_t vnodeProcessWrite(void *vparam, void *wparam, int32_t qtype, void *rparam);
-int32_t vnodeCheckWrite(void *pVnode);
 int32_t vnodeGetVnodeList(int32_t vnodeList[], int32_t *numOfVnodes);
 void    vnodeBuildStatusMsg(void *param);
 void    vnodeConfirmForward(void *param, uint64_t version, int32_t code);
@@ -77,8 +80,8 @@ void    vnodeSetAccess(SVgroupAccess *pAccess, int32_t numOfVnodes);
 int32_t vnodeInitResources();
 void    vnodeCleanupResources();
 
-int32_t vnodeProcessRead(void *pVnode, SVReadMsg *pReadMsg);
-int32_t vnodeCheckRead(void *pVnode);
+int32_t vnodeWriteToRQueue(void *vparam, void *pCont, int32_t contLen, int8_t qtype, void *rparam);
+int32_t vnodeProcessRead(void *pVnode, SVReadMsg *pRead);
 
 #ifdef __cplusplus
 }

src/mnode/inc/mnodeDef.h

@@ -172,7 +172,8 @@ typedef struct {
   int8_t  walLevel;
   int8_t  replications;
   int8_t  quorum;
-  int8_t  reserved[12];
+  int8_t  update;
+  int8_t  reserved[11];
 } SDbCfg;
 
 typedef struct SDbObj {

src/mnode/src/mnodeDb.c

@@ -319,6 +319,11 @@ static int32_t mnodeCheckDbCfg(SDbCfg *pCfg) {
   }
 #endif
 
+  if (pCfg->update < TSDB_MIN_DB_UPDATE || pCfg->update > TSDB_MAX_DB_UPDATE) {
+    mError("invalid db option update:%d valid range: [%d, %d]", pCfg->update, TSDB_MIN_DB_UPDATE, TSDB_MAX_DB_UPDATE);
+    return TSDB_CODE_MND_INVALID_DB_OPTION;
+  }
+
   return TSDB_CODE_SUCCESS;
 }
 
@@ -339,6 +344,7 @@ static void mnodeSetDefaultDbCfg(SDbCfg *pCfg) {
   if (pCfg->walLevel < 0) pCfg->walLevel = tsWAL;
   if (pCfg->replications < 0) pCfg->replications = tsReplications;
   if (pCfg->quorum < 0) pCfg->quorum = tsQuorum;
+  if (pCfg->update < 0) pCfg->update = tsUpdate;
 }
 
 static int32_t mnodeCreateDbCb(SMnodeMsg *pMsg, int32_t code) {
@@ -391,7 +397,8 @@ static int32_t mnodeCreateDb(SAcctObj *pAcct, SCreateDbMsg *pCreate, SMnodeMsg *
     .compression         = pCreate->compression,
     .walLevel            = pCreate->walLevel,
     .replications        = pCreate->replications,
-    .quorum              = pCreate->quorum
+    .quorum              = pCreate->quorum,
+    .update              = pCreate->update
   };
 
   mnodeSetDefaultDbCfg(&pDb->cfg);
@@ -610,6 +617,12 @@ static int32_t mnodeGetDbMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn
   pSchema[cols].bytes = htons(pShow->bytes[cols]);
   cols++;
 
+  pShow->bytes[cols] = 1;
+  pSchema[cols].type = TSDB_DATA_TYPE_TINYINT;
+  strcpy(pSchema[cols].name, "update");
+  pSchema[cols].bytes = htons(pShow->bytes[cols]);
+  cols++;
+
   pShow->bytes[cols] = 10 + VARSTR_HEADER_SIZE;
   pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
   strcpy(pSchema[cols].name, "status");
@@ -749,6 +762,10 @@ static int32_t mnodeRetrieveDbs(SShowObj *pShow, char *data, int32_t rows, void
     STR_WITH_SIZE_TO_VARSTR(pWrite, prec, 2);
     cols++;
 
+    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
+    *(int8_t *)pWrite = pDb->cfg.update;
+    cols++;
+
     pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
     if (pDb->status == TSDB_DB_STATUS_READY) {
       const char *src = "ready";
@@ -848,6 +865,7 @@ static SDbCfg mnodeGetAlterDbOption(SDbObj *pDb, SAlterDbMsg *pAlter) {
   int8_t  replications   = pAlter->replications;
   int8_t  quorum         = pAlter->quorum;
   int8_t  precision      = pAlter->precision;
+  int8_t  update         = pAlter->update;
   
   terrno = TSDB_CODE_SUCCESS;
 
@@ -950,6 +968,16 @@ static SDbCfg mnodeGetAlterDbOption(SDbObj *pDb, SAlterDbMsg *pAlter) {
     newCfg.quorum = quorum;
   }
 
+  if (update >= 0 && update != pDb->cfg.update) {
+#if 0
+    mDebug("db:%s, update:%d change to %d", pDb->name, pDb->cfg.update, update);
+    newCfg.update = update;
+#else
+    mError("db:%s, can't alter update option", pDb->name);
+    terrno = TSDB_CODE_MND_INVALID_DB_OPTION;
+#endif
+  }
+
   return newCfg;
 }
 

src/mnode/src/mnodeVgroup.c

@@ -850,6 +850,7 @@ static SCreateVnodeMsg *mnodeBuildVnodeMsg(SVgObj *pVgroup) {
   pCfg->replications        = (int8_t) pVgroup->numOfVnodes;
   pCfg->wals                = 3;
   pCfg->quorum              = pDb->cfg.quorum;
+  pCfg->update              = pDb->cfg.update;
   
   SVnodeDesc *pNodes = pVnode->nodes;
   for (int32_t j = 0; j < pVgroup->numOfVnodes; ++j) {

src/query/inc/qSqlparser.h

@@ -129,6 +129,7 @@ typedef struct SCreateDBInfo {
   int32_t   compressionLevel;
   SStrToken precision;
   bool      ignoreExists;
+  int8_t    update; 
   
   tVariantList *keep;
 } SCreateDBInfo;

src/query/inc/sql.y

@@ -239,6 +239,7 @@ wal(Y)     ::= WAL INTEGER(X).                { Y = X; }
 fsync(Y)   ::= FSYNC INTEGER(X).              { Y = X; }
 comp(Y)    ::= COMP INTEGER(X).               { Y = X; }
 prec(Y)    ::= PRECISION STRING(X).           { Y = X; }
+update(Y)  ::= UPDATE INTEGER(X).             { Y = X; }     
 
 %type db_optr {SCreateDBInfo}
 db_optr(Y) ::= . {setDefaultCreateDbOption(&Y);}
@@ -256,6 +257,7 @@ db_optr(Y) ::= db_optr(Z) fsync(X).          { Y = Z; Y.fsyncPeriod = strtol(X.z
 db_optr(Y) ::= db_optr(Z) comp(X).           { Y = Z; Y.compressionLevel = strtol(X.z, NULL, 10); }
 db_optr(Y) ::= db_optr(Z) prec(X).           { Y = Z; Y.precision = X; }
 db_optr(Y) ::= db_optr(Z) keep(X).           { Y = Z; Y.keep = X; }
+db_optr(Y) ::= db_optr(Z) update(X).         { Y = Z; Y.update = strtol(X.z, NULL, 10); }
 
 %type alter_db_optr {SCreateDBInfo}
 alter_db_optr(Y) ::= . { setDefaultCreateDbOption(&Y);}
@@ -267,6 +269,7 @@ alter_db_optr(Y) ::= alter_db_optr(Z) blocks(X).      { Y = Z; Y.numOfBlocks = s
 alter_db_optr(Y) ::= alter_db_optr(Z) comp(X).        { Y = Z; Y.compressionLevel = strtol(X.z, NULL, 10); }
 alter_db_optr(Y) ::= alter_db_optr(Z) wal(X).         { Y = Z; Y.walLevel = strtol(X.z, NULL, 10); }
 alter_db_optr(Y) ::= alter_db_optr(Z) fsync(X).       { Y = Z; Y.fsyncPeriod = strtol(X.z, NULL, 10); }
+alter_db_optr(Y) ::= alter_db_optr(Z) update(X).       { Y = Z; Y.update = strtol(X.z, NULL, 10); }
 
 %type typename {TAOS_FIELD}
 typename(A) ::= ids(X). { 

src/query/inc/tsqlfunction.h

@@ -149,7 +149,7 @@ typedef struct SResultRowCellInfo {
   int8_t   hasResult;       // result generated, not NULL value
   bool     initialized;     // output buffer has been initialized
   bool     complete;        // query has completed
-  uint16_t numOfRes;        // num of output result in current buffer
+  uint32_t numOfRes;        // num of output result in current buffer
 } SResultRowCellInfo;
 
 #define GET_ROWCELL_INTERBUF(_c) ((void*) ((char*)(_c) + sizeof(SResultRowCellInfo)))

src/query/src/qAst.c

@@ -32,7 +32,6 @@
 #include "tstoken.h"
 #include "ttokendef.h"
 #include "tulog.h"
-#include "tutil.h"
 
 /*
  *
@@ -415,9 +414,9 @@ static void tQueryIndexColumn(SSkipList* pSkipList, tQueryInfo* pQueryInfo, SArr
   }
 
   if (cond.start != NULL) {
-    iter = tSkipListCreateIterFromVal(pSkipList, (char*) cond.start->v, pSkipList->keyInfo.type, TSDB_ORDER_ASC);
+    iter = tSkipListCreateIterFromVal(pSkipList, (char*) cond.start->v, pSkipList->type, TSDB_ORDER_ASC);
   } else {
-    iter = tSkipListCreateIterFromVal(pSkipList, (char*)(cond.end ? cond.end->v: NULL), pSkipList->keyInfo.type, TSDB_ORDER_DESC);
+    iter = tSkipListCreateIterFromVal(pSkipList, (char*)(cond.end ? cond.end->v: NULL), pSkipList->type, TSDB_ORDER_DESC);
   }
   
   if (cond.start != NULL) {
@@ -432,7 +431,7 @@ static void tQueryIndexColumn(SSkipList* pSkipList, tQueryInfo* pQueryInfo, SArr
           break;
         }
 
-        STableKeyInfo info = {.pTable = *(void**)SL_GET_NODE_DATA(pNode), .lastKey = TSKEY_INITIAL_VAL};
+        STableKeyInfo info = {.pTable = (void*)SL_GET_NODE_DATA(pNode), .lastKey = TSKEY_INITIAL_VAL};
         taosArrayPush(result, &info);
       }
     } else if (optr == TSDB_RELATION_GREATER || optr == TSDB_RELATION_GREATER_EQUAL) { // greater equal
@@ -450,7 +449,7 @@ static void tQueryIndexColumn(SSkipList* pSkipList, tQueryInfo* pQueryInfo, SArr
         if (ret == 0 && optr == TSDB_RELATION_GREATER) {
           continue;
         } else {
-          STableKeyInfo info = {.pTable = *(void**)SL_GET_NODE_DATA(pNode), .lastKey = TSKEY_INITIAL_VAL};
+          STableKeyInfo info = {.pTable = (void*)SL_GET_NODE_DATA(pNode), .lastKey = TSKEY_INITIAL_VAL};
           taosArrayPush(result, &info);
           comp = false;
         }
@@ -465,22 +464,22 @@ static void tQueryIndexColumn(SSkipList* pSkipList, tQueryInfo* pQueryInfo, SArr
           continue;
         }
 
-        STableKeyInfo info = {.pTable = *(void**)SL_GET_NODE_DATA(pNode), .lastKey = TSKEY_INITIAL_VAL};
+        STableKeyInfo info = {.pTable = (void*)SL_GET_NODE_DATA(pNode), .lastKey = TSKEY_INITIAL_VAL};
         taosArrayPush(result, &info);
       }
       
       tSkipListDestroyIter(iter);
   
       comp = true;
-      iter = tSkipListCreateIterFromVal(pSkipList, (char*) cond.start->v, pSkipList->keyInfo.type, TSDB_ORDER_DESC);
+      iter = tSkipListCreateIterFromVal(pSkipList, (char*) cond.start->v, pSkipList->type, TSDB_ORDER_DESC);
       while(tSkipListIterNext(iter)) {
         SSkipListNode* pNode = tSkipListIterGet(iter);
         comp = comp && (pQueryInfo->compare(SL_GET_NODE_KEY(pSkipList, pNode), cond.start->v) == 0);
-        if (comp) {
+      if (comp) {
           continue;
         }
 
-        STableKeyInfo info = {.pTable = *(void**)SL_GET_NODE_DATA(pNode), .lastKey = TSKEY_INITIAL_VAL};
+        STableKeyInfo info = {.pTable = (void*)SL_GET_NODE_DATA(pNode), .lastKey = TSKEY_INITIAL_VAL};
         taosArrayPush(result, &info);
       }
   
@@ -504,7 +503,7 @@ static void tQueryIndexColumn(SSkipList* pSkipList, tQueryInfo* pQueryInfo, SArr
         if (ret == 0 && optr == TSDB_RELATION_LESS) {
           continue;
         } else {
-          STableKeyInfo info = {.pTable = *(void **)SL_GET_NODE_DATA(pNode), .lastKey = TSKEY_INITIAL_VAL};
+          STableKeyInfo info = {.pTable = (void *)SL_GET_NODE_DATA(pNode), .lastKey = TSKEY_INITIAL_VAL};
           taosArrayPush(result, &info);
           comp = false;  // no need to compare anymore
         }
@@ -518,7 +517,7 @@ static void tQueryIndexColumn(SSkipList* pSkipList, tQueryInfo* pQueryInfo, SArr
         bool isnull = isNull(SL_GET_NODE_KEY(pSkipList, pNode), pQueryInfo->sch.type);
         if ((pQueryInfo->optr == TSDB_RELATION_ISNULL && isnull) ||
             (pQueryInfo->optr == TSDB_RELATION_NOTNULL && (!isnull))) {
-          STableKeyInfo info = {.pTable = *(void **)SL_GET_NODE_DATA(pNode), .lastKey = TSKEY_INITIAL_VAL};
+          STableKeyInfo info = {.pTable = (void *)SL_GET_NODE_DATA(pNode), .lastKey = TSKEY_INITIAL_VAL};
           taosArrayPush(result, &info);
         }
       }
@@ -682,7 +681,7 @@ static void tSQLBinaryTraverseOnSkipList(tExprNode *pExpr, SArray *pResult, SSki
   while (tSkipListIterNext(iter)) {
     SSkipListNode *pNode = tSkipListIterGet(iter);
     if (filterItem(pExpr, pNode, param)) {
-      taosArrayPush(pResult, SL_GET_NODE_DATA(pNode));
+      taosArrayPush(pResult, &(SL_GET_NODE_DATA(pNode)));
     }
   }
   tSkipListDestroyIter(iter);
@@ -697,7 +696,7 @@ static void tQueryIndexlessColumn(SSkipList* pSkipList, tQueryInfo* pQueryInfo,
     SSkipListNode *pNode = tSkipListIterGet(iter);
     char *         pData = SL_GET_NODE_DATA(pNode);
 
-    tstr *name = (tstr*) tsdbGetTableName(*(void**) pData);
+    tstr *name = (tstr*) tsdbGetTableName((void*) pData);
 
     // todo speed up by using hash
     if (pQueryInfo->sch.colId == TSDB_TBNAME_COLUMN_INDEX) {
@@ -711,7 +710,7 @@ static void tQueryIndexlessColumn(SSkipList* pSkipList, tQueryInfo* pQueryInfo,
     }
 
     if (addToResult) {
-      STableKeyInfo info = {.pTable = *(void**)pData, .lastKey = TSKEY_INITIAL_VAL};
+      STableKeyInfo info = {.pTable = (void*)pData, .lastKey = TSKEY_INITIAL_VAL};
       taosArrayPush(res, &info);
     }
   }

src/query/src/qExecutor.c

@@ -1609,7 +1609,7 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int16_t order
 
   pRuntimeEnv->pCtx = (SQLFunctionCtx *)calloc(pQuery->numOfOutput, sizeof(SQLFunctionCtx));
   pRuntimeEnv->rowCellInfoOffset = calloc(pQuery->numOfOutput, sizeof(int32_t));
-  pRuntimeEnv->pResultRow = getNewResultRow(pRuntimeEnv->pool);//calloc(1, sizeof(SResultRow));
+  pRuntimeEnv->pResultRow = getNewResultRow(pRuntimeEnv->pool);
 
   if (pRuntimeEnv->pResultRow == NULL || pRuntimeEnv->pCtx == NULL || pRuntimeEnv->rowCellInfoOffset == NULL) {
     goto _clean;
@@ -1745,6 +1745,7 @@ static void teardownQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv) {
 
   pRuntimeEnv->pTSBuf = tsBufDestroy(pRuntimeEnv->pTSBuf);
   taosTFree(pRuntimeEnv->keyBuf);
+  taosTFree(pRuntimeEnv->rowCellInfoOffset);
 
   taosHashCleanup(pRuntimeEnv->pResultRowHashTable);
   pRuntimeEnv->pResultRowHashTable = NULL;

src/query/src/qFill.c

@@ -170,7 +170,7 @@ int64_t getFilledNumOfRes(SFillInfo* pFillInfo, TSKEY ekey, int32_t maxNumOfRows
   int32_t numOfRows = taosNumOfRemainRows(pFillInfo);
 
   TSKEY ekey1 = ekey;
-  if (pFillInfo->order != TSDB_ORDER_ASC) {
+  if (!FILL_IS_ASC_FILL(pFillInfo)) {
     pFillInfo->endKey = taosTimeTruncate(ekey, &pFillInfo->interval, pFillInfo->precision);
   }
 

src/query/src/qHistogram.c

@@ -168,7 +168,7 @@ int32_t tHistogramAdd(SHistogramInfo** pHisto, double val) {
       (*pHisto)->numOfEntries += 1;
     }
   } else { /* insert a new slot */
-    if ((*pHisto)->numOfElems > 1 && idx < (*pHisto)->numOfEntries) {
+    if ((*pHisto)->numOfElems >= 1 && idx < (*pHisto)->numOfEntries) {
       if (idx > 0) {
         assert((*pHisto)->elems[idx - 1].val <= val);
       }
@@ -661,4 +661,4 @@ SHistogramInfo* tHistogramMerge(SHistogramInfo* pHisto1, SHistogramInfo* pHisto2
 
   free(pHistoBins);
   return pResHistogram;
-}
\ No newline at end of file
+}

src/query/src/qParserImpl.c

@@ -872,5 +872,6 @@ void setDefaultCreateDbOption(SCreateDBInfo *pDBInfo) {
   pDBInfo->quorum = -1;
   pDBInfo->keep = NULL;
 
+  pDBInfo->update = -1;
   memset(&pDBInfo->precision, 0, sizeof(SStrToken));
 }

src/query/src/qTokenizer.c

@@ -155,6 +155,7 @@ static SKeyword keywordTable[] = {
     {"INSERT",       TK_INSERT},
     {"INTO",         TK_INTO},
     {"VALUES",       TK_VALUES},
+    {"UPDATE",       TK_UPDATE},
     {"RESET",        TK_RESET},
     {"QUERY",        TK_QUERY},
     {"ADD",          TK_ADD},
@@ -664,4 +665,4 @@ void taosCleanupKeywordsTable() {
   if (m != NULL && atomic_val_compare_exchange_ptr(&keywordHashTable, m, 0) == m) {
     taosHashCleanup(m);
   }
-}
\ No newline at end of file
+}

src/tsdb/inc/tsdbMain.h

@@ -320,6 +320,15 @@ typedef struct {
   void*      compBuffer;  // Buffer for temperary compress/decompress purpose
 } SRWHelper;
 
+typedef struct {
+  int   rowsInserted;
+  int   rowsUpdated;
+  int   rowsDeleteSucceed;
+  int   rowsDeleteFailed;
+  int   nOperations;
+  TSKEY keyFirst;
+  TSKEY keyLast;
+} SMergeInfo;
 // ------------------ tsdbScan.c
 typedef struct {
   SFileGroup fGroup;
@@ -422,7 +431,7 @@ void          tsdbCloseBufPool(STsdbRepo* pRepo);
 SListNode*    tsdbAllocBufBlockFromPool(STsdbRepo* pRepo);
 
 // ------------------ tsdbMemTable.c
-int   tsdbInsertRowToMem(STsdbRepo* pRepo, SDataRow row, STable* pTable);
+int   tsdbUpdateRowInMem(STsdbRepo* pRepo, SDataRow row, STable* pTable);
 int   tsdbRefMemTable(STsdbRepo* pRepo, SMemTable* pMemTable);
 int   tsdbUnRefMemTable(STsdbRepo* pRepo, SMemTable* pMemTable);
 int   tsdbTakeMemSnapshot(STsdbRepo* pRepo, SMemTable** pMem, SMemTable** pIMem);
@@ -430,7 +439,7 @@ void  tsdbUnTakeMemSnapShot(STsdbRepo* pRepo, SMemTable* pMem, SMemTable* pIMem)
 void* tsdbAllocBytes(STsdbRepo* pRepo, int bytes);
 int   tsdbAsyncCommit(STsdbRepo* pRepo);
 int   tsdbLoadDataFromCache(STable* pTable, SSkipListIterator* pIter, TSKEY maxKey, int maxRowsToRead, SDataCols* pCols,
-                            TSKEY* filterKeys, int nFilterKeys);
+                            TKEY* filterKeys, int nFilterKeys, bool keepDup, SMergeInfo* pMergeInfo);
 
 static FORCE_INLINE SDataRow tsdbNextIterRow(SSkipListIterator* pIter) {
   if (pIter == NULL) return NULL;
@@ -438,16 +447,23 @@ static FORCE_INLINE SDataRow tsdbNextIterRow(SSkipListIterator* pIter) {
   SSkipListNode* node = tSkipListIterGet(pIter);
   if (node == NULL) return NULL;
 
-  return *(SDataRow *)SL_GET_NODE_DATA(node);
+  return (SDataRow)SL_GET_NODE_DATA(node);
 }
 
 static FORCE_INLINE TSKEY tsdbNextIterKey(SSkipListIterator* pIter) {
   SDataRow row = tsdbNextIterRow(pIter);
-  if (row == NULL) return -1;
+  if (row == NULL) return TSDB_DATA_TIMESTAMP_NULL;
 
   return dataRowKey(row);
 }
 
+static FORCE_INLINE TKEY tsdbNextIterTKey(SSkipListIterator* pIter) {
+  SDataRow row = tsdbNextIterRow(pIter);
+  if (row == NULL) return TKEY_NULL;
+
+  return dataRowTKey(row);
+}
+
 static FORCE_INLINE STsdbBufBlock* tsdbGetCurrBufBlock(STsdbRepo* pRepo) {
   ASSERT(pRepo != NULL);
   if (pRepo->mem == NULL) return NULL;

src/tsdb/src/tsdbMain.c

@@ -512,6 +512,9 @@ static int32_t tsdbCheckAndSetDefaultCfg(STsdbCfg *pCfg) {
     }
   }
 
+  // update check
+  if (pCfg->update != 0) pCfg->update = 1;
+
   return 0;
 
 _err:
@@ -762,7 +765,7 @@ static int32_t tsdbInsertDataToTable(STsdbRepo *pRepo, SSubmitBlk *pBlock, TSKEY
       return -1;
     }
 
-    if (tsdbInsertRowToMem(pRepo, row, pTable) < 0) return -1;
+    if (tsdbUpdateRowInMem(pRepo, row, pTable) < 0) return -1;
 
     (*affectedrows)++;
     points++;
@@ -923,6 +926,7 @@ static int tsdbEncodeCfg(void **buf, STsdbCfg *pCfg) {
   tlen += taosEncodeVariantI32(buf, pCfg->maxRowsPerFileBlock);
   tlen += taosEncodeFixedI8(buf, pCfg->precision);
   tlen += taosEncodeFixedI8(buf, pCfg->compression);
+  tlen += taosEncodeFixedI8(buf, pCfg->update);
 
   return tlen;
 }
@@ -939,6 +943,7 @@ static void *tsdbDecodeCfg(void *buf, STsdbCfg *pCfg) {
   buf = taosDecodeVariantI32(buf, &(pCfg->maxRowsPerFileBlock));
   buf = taosDecodeFixedI8(buf, &(pCfg->precision));
   buf = taosDecodeFixedI8(buf, &(pCfg->compression));
+  buf = taosDecodeFixedI8(buf, &(pCfg->update));
 
   return buf;
 }

src/tsdb/src/tsdbMemTable.c

@@ -32,43 +32,40 @@ static int tsdbCommitToFile(STsdbRepo *pRepo, int fid, SCommitIter *iters, SRWHe
 static SCommitIter *tsdbCreateCommitIters(STsdbRepo *pRepo);
 static void         tsdbDestroyCommitIters(SCommitIter *iters, int maxTables);
 static int          tsdbAdjustMemMaxTables(SMemTable *pMemTable, int maxTables);
+static int          tsdbAppendTableRowToCols(STable *pTable, SDataCols *pCols, STSchema **ppSchema, SDataRow row);
 
 // ---------------- INTERNAL FUNCTIONS ----------------
-int tsdbInsertRowToMem(STsdbRepo *pRepo, SDataRow row, STable *pTable) {
+int tsdbUpdateRowInMem(STsdbRepo *pRepo, SDataRow row, STable *pTable) {
   STsdbCfg *  pCfg = &pRepo->config;
   STsdbMeta * pMeta = pRepo->tsdbMeta;
-  int32_t     level = 0;
-  int32_t     headSize = 0;
+  TKEY        tkey = dataRowTKey(row);
   TSKEY       key = dataRowKey(row);
   SMemTable * pMemTable = pRepo->mem;
   STableData *pTableData = NULL;
-  SSkipList * pSList = NULL;
+  bool        isRowDelete = TKEY_IS_DELETED(tkey);
 
-  if (pMemTable != NULL && TABLE_TID(pTable) < pMemTable->maxTables && pMemTable->tData[TABLE_TID(pTable)] != NULL &&
-      pMemTable->tData[TABLE_TID(pTable)]->uid == TABLE_UID(pTable)) {
-    pTableData = pMemTable->tData[TABLE_TID(pTable)];
-    pSList = pTableData->pData;
-  }
-
-  tSkipListNewNodeInfo(pSList, &level, &headSize);
+  if (isRowDelete) {
+    if (!pCfg->update) {
+      tsdbWarn("vgId:%d vnode is not allowed to update but try to delete a data row", REPO_ID(pRepo));
+      terrno = TSDB_CODE_TDB_INVALID_ACTION;
+      return -1;
+    }
 
-  SSkipListNode *pNode = (SSkipListNode *)malloc(headSize + sizeof(SDataRow *));
-  if (pNode == NULL) {
-    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
-    return -1;
+    if (key > TABLE_LASTKEY(pTable)) {
+      tsdbTrace("vgId:%d skip to delete row key %" PRId64 " which is larger than table lastKey %" PRId64,
+                REPO_ID(pRepo), key, TABLE_LASTKEY(pTable));
+      return 0;
+    }
   }
 
   void *pRow = tsdbAllocBytes(pRepo, dataRowLen(row));
   if (pRow == NULL) {
     tsdbError("vgId:%d failed to insert row with key %" PRId64 " to table %s while allocate %d bytes since %s",
               REPO_ID(pRepo), key, TABLE_CHAR_NAME(pTable), dataRowLen(row), tstrerror(terrno));
-    free(pNode);
     return -1;
   }
 
-  pNode->level = level;
   dataRowCpy(pRow, row);
-  *(SDataRow *)SL_GET_NODE_DATA(pNode) = pRow;
 
   // Operations above may change pRepo->mem, retake those values
   ASSERT(pRepo->mem != NULL);
@@ -77,7 +74,6 @@ int tsdbInsertRowToMem(STsdbRepo *pRepo, SDataRow row, STable *pTable) {
   if (TABLE_TID(pTable) >= pMemTable->maxTables) {
     if (tsdbAdjustMemMaxTables(pMemTable, pMeta->maxTables) < 0) {
       tsdbFreeBytes(pRepo, pRow, dataRowLen(row));
-      free(pNode);
       return -1;
     }
   }
@@ -97,7 +93,6 @@ int tsdbInsertRowToMem(STsdbRepo *pRepo, SDataRow row, STable *pTable) {
                 " to table %s while create new table data object since %s",
                 REPO_ID(pRepo), key, TABLE_CHAR_NAME(pTable), tstrerror(terrno));
       tsdbFreeBytes(pRepo, (void *)pRow, dataRowLen(row));
-      free(pNode);
       return -1;
     }
 
@@ -106,24 +101,31 @@ int tsdbInsertRowToMem(STsdbRepo *pRepo, SDataRow row, STable *pTable) {
 
   ASSERT((pTableData != NULL) && pTableData->uid == TABLE_UID(pTable));
 
-  if (tSkipListPut(pTableData->pData, pNode) == NULL) {
+  int64_t oldSize = SL_SIZE(pTableData->pData);
+  if (tSkipListPut(pTableData->pData, pRow) == NULL) {
     tsdbFreeBytes(pRepo, (void *)pRow, dataRowLen(row));
-    free(pNode);
   } else {
-    if (TABLE_LASTKEY(pTable) < key) TABLE_LASTKEY(pTable) = key;
+    int64_t deltaSize = SL_SIZE(pTableData->pData) - oldSize;
+    if (isRowDelete) {
+      if (TABLE_LASTKEY(pTable) == key) {
+        // TODO: need to update table last key here (may from file)
+      }
+    } else {
+      if (TABLE_LASTKEY(pTable) < key) TABLE_LASTKEY(pTable) = key;
+    }
+
     if (pMemTable->keyFirst > key) pMemTable->keyFirst = key;
     if (pMemTable->keyLast < key) pMemTable->keyLast = key;
-    pMemTable->numOfRows++;
+    pMemTable->numOfRows += deltaSize;
 
     if (pTableData->keyFirst > key) pTableData->keyFirst = key;
     if (pTableData->keyLast < key) pTableData->keyLast = key;
-    pTableData->numOfRows++;
-
-    ASSERT(pTableData->numOfRows == tSkipListGetSize(pTableData->pData));
+    pTableData->numOfRows += deltaSize;
   }
 
-  tsdbTrace("vgId:%d a row is inserted to table %s tid %d uid %" PRIu64 " key %" PRIu64, REPO_ID(pRepo),
-            TABLE_CHAR_NAME(pTable), TABLE_TID(pTable), TABLE_UID(pTable), key);
+  tsdbTrace("vgId:%d a row is %s table %s tid %d uid %" PRIu64 " key %" PRIu64, REPO_ID(pRepo),
+            isRowDelete ? "deleted from" : "updated in", TABLE_CHAR_NAME(pTable), TABLE_TID(pTable), TABLE_UID(pTable),
+            key);
 
   return 0;
 }
@@ -295,63 +297,124 @@ int tsdbAsyncCommit(STsdbRepo *pRepo) {
   return 0;
 }
 
+/**
+ * This is an important function to load data or try to load data from memory skiplist iterator.
+ * 
+ * This function load memory data until:
+ * 1. iterator ends
+ * 2. data key exceeds maxKey
+ * 3. rowsIncreased = rowsInserted - rowsDeleteSucceed >= maxRowsToRead
+ * 4. operations in pCols not exceeds its max capacity if pCols is given
+ * 
+ * The function tries to procceed AS MUSH AS POSSIBLE.
+ */
 int tsdbLoadDataFromCache(STable *pTable, SSkipListIterator *pIter, TSKEY maxKey, int maxRowsToRead, SDataCols *pCols,
-                          TSKEY *filterKeys, int nFilterKeys) {
-  ASSERT(maxRowsToRead > 0 && nFilterKeys >= 0);
+                          TKEY *filterKeys, int nFilterKeys, bool keepDup, SMergeInfo *pMergeInfo) {
+  ASSERT(maxRowsToRead > 0 && nFilterKeys >= 0 && pMergeInfo != NULL);
   if (pIter == NULL) return 0;
   STSchema *pSchema = NULL;
-  int       numOfRows = 0;
-  TSKEY     keyNext = 0;
+  TSKEY     rowKey = 0;
+  TSKEY     fKey = 0;
+  bool      isRowDel = false;
   int       filterIter = 0;
+  SDataRow  row = NULL;
+
+  memset(pMergeInfo, 0, sizeof(*pMergeInfo));
+  pMergeInfo->keyFirst = INT64_MAX;
+  pMergeInfo->keyLast = INT64_MIN;
+  if (pCols) tdResetDataCols(pCols);
+
+  row = tsdbNextIterRow(pIter);
+  if (row == NULL || dataRowKey(row) > maxKey) {
+    rowKey = INT64_MAX;
+    isRowDel = false;
+  } else {
+    rowKey = dataRowKey(row);
+    isRowDel = dataRowDeleted(row);
+  }
 
-  if (nFilterKeys != 0) { // for filter purpose
-    ASSERT(filterKeys != NULL);
-    keyNext = tsdbNextIterKey(pIter);
-    if (keyNext < 0 || keyNext > maxKey) return numOfRows;
-    void *ptr = taosbsearch((void *)(&keyNext), (void *)filterKeys, nFilterKeys, sizeof(TSKEY), compTSKEY, TD_GE);
-    filterIter = (ptr == NULL) ? nFilterKeys : (int)((POINTER_DISTANCE(ptr, filterKeys) / sizeof(TSKEY)));
-  }
-
-  do {
-    SDataRow row = tsdbNextIterRow(pIter);
-    if (row == NULL) break;
-
-    keyNext = dataRowKey(row);
-    if (keyNext > maxKey) break;
-
-    bool keyFiltered = false;
-    if (nFilterKeys != 0) {
-      while (true) {
-        if (filterIter >= nFilterKeys) break;
-        if (keyNext == filterKeys[filterIter]) {
-          keyFiltered = true;
-          filterIter++;
-          break;
-        } else if (keyNext < filterKeys[filterIter]) {
-          break;
+  if (filterIter >= nFilterKeys) {
+    fKey = INT64_MAX;
+  } else {
+    fKey = tdGetKey(filterKeys[filterIter]);
+  }
+
+  while (true) {
+    if (fKey == INT64_MAX && rowKey == INT64_MAX) break;
+
+    if (fKey < rowKey) {
+      pMergeInfo->keyFirst = MIN(pMergeInfo->keyFirst, fKey);
+      pMergeInfo->keyLast = MAX(pMergeInfo->keyLast, fKey);
+
+      filterIter++;
+      if (filterIter >= nFilterKeys) {
+        fKey = INT64_MAX;
+      } else {
+        fKey = tdGetKey(filterKeys[filterIter]);
+      }
+    } else if (fKey > rowKey) {
+      if (isRowDel) {
+        pMergeInfo->rowsDeleteFailed++;
+      } else {
+        if (pMergeInfo->rowsInserted - pMergeInfo->rowsDeleteSucceed >= maxRowsToRead) break;
+        if (pCols && pMergeInfo->nOperations >= pCols->maxPoints) break;
+        pMergeInfo->rowsInserted++;
+        pMergeInfo->nOperations++;
+        pMergeInfo->keyFirst = MIN(pMergeInfo->keyFirst, rowKey);
+        pMergeInfo->keyLast = MAX(pMergeInfo->keyLast, rowKey);
+        tsdbAppendTableRowToCols(pTable, pCols, &pSchema, row);
+      }
+
+      tSkipListIterNext(pIter);
+      row = tsdbNextIterRow(pIter);
+      if (row == NULL || dataRowKey(row) > maxKey) {
+        rowKey = INT64_MAX;
+        isRowDel = false;
+      } else {
+        rowKey = dataRowKey(row);
+        isRowDel = dataRowDeleted(row);
+      }
+    } else {
+      if (isRowDel) {
+        ASSERT(!keepDup);
+        if (pCols && pMergeInfo->nOperations >= pCols->maxPoints) break;
+        pMergeInfo->rowsDeleteSucceed++;
+        pMergeInfo->nOperations++;
+        tsdbAppendTableRowToCols(pTable, pCols, &pSchema, row);
+      } else {
+        if (keepDup) {
+          if (pCols && pMergeInfo->nOperations >= pCols->maxPoints) break;
+          pMergeInfo->rowsUpdated++;
+          pMergeInfo->nOperations++;
+          pMergeInfo->keyFirst = MIN(pMergeInfo->keyFirst, rowKey);
+          pMergeInfo->keyLast = MAX(pMergeInfo->keyLast, rowKey);
+          tsdbAppendTableRowToCols(pTable, pCols, &pSchema, row);
         } else {
-          filterIter++;
+          pMergeInfo->keyFirst = MIN(pMergeInfo->keyFirst, fKey);
+          pMergeInfo->keyLast = MAX(pMergeInfo->keyLast, fKey);
         }
       }
-    }
 
-    if (!keyFiltered) {
-      if (numOfRows >= maxRowsToRead) break;
-      if (pCols) {
-        if (pSchema == NULL || schemaVersion(pSchema) != dataRowVersion(row)) {
-          pSchema = tsdbGetTableSchemaImpl(pTable, false, false, dataRowVersion(row));
-          if (pSchema == NULL) {
-            ASSERT(0);
-          }
-        }
+      tSkipListIterNext(pIter);
+      row = tsdbNextIterRow(pIter);
+      if (row == NULL || dataRowKey(row) > maxKey) {
+        rowKey = INT64_MAX;
+        isRowDel = false;
+      } else {
+        rowKey = dataRowKey(row);
+        isRowDel = dataRowDeleted(row);
+      }
 
-        tdAppendDataRowToDataCol(row, pSchema, pCols);
+      filterIter++;
+      if (filterIter >= nFilterKeys) {
+        fKey = INT64_MAX;
+      } else {
+        fKey = tdGetKey(filterKeys[filterIter]);
       }
-      numOfRows++;
     }
-  } while (tSkipListIterNext(pIter));
+  }
 
-  return numOfRows;
+  return 0;
 }
 
 // ---------------- LOCAL FUNCTIONS ----------------
@@ -440,8 +503,9 @@ static STableData *tsdbNewTableData(STsdbCfg *pCfg, STable *pTable) {
   pTableData->keyLast = 0;
   pTableData->numOfRows = 0;
 
-  pTableData->pData = tSkipListCreate(TSDB_DATA_SKIPLIST_LEVEL, TSDB_DATA_TYPE_TIMESTAMP,
-                                      TYPE_BYTES[TSDB_DATA_TYPE_TIMESTAMP], 0, 0, 1, tsdbGetTsTupleKey);
+  pTableData->pData =
+      tSkipListCreate(TSDB_DATA_SKIPLIST_LEVEL, TSDB_DATA_TYPE_TIMESTAMP, TYPE_BYTES[TSDB_DATA_TYPE_TIMESTAMP],
+                      tkeyComparFn, pCfg->update ? SL_UPDATE_DUP_KEY : SL_DISCARD_DUP_KEY, tsdbGetTsTupleKey);
   if (pTableData->pData == NULL) {
     terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
     goto _err;
@@ -461,7 +525,7 @@ static void tsdbFreeTableData(STableData *pTableData) {
   }
 }
 
-static char *tsdbGetTsTupleKey(const void *data) { return dataRowTuple(*(SDataRow *)data); }
+static char *tsdbGetTsTupleKey(const void *data) { return dataRowTuple((SDataRow)data); }
 
 static void *tsdbCommitData(void *arg) {
   STsdbRepo *  pRepo = (STsdbRepo *)arg;
@@ -583,7 +647,7 @@ static void tsdbEndCommit(STsdbRepo *pRepo) {
 static int tsdbHasDataToCommit(SCommitIter *iters, int nIters, TSKEY minKey, TSKEY maxKey) {
   for (int i = 0; i < nIters; i++) {
     TSKEY nextKey = tsdbNextIterKey((iters + i)->pIter);
-    if (nextKey > 0 && (nextKey >= minKey && nextKey <= maxKey)) return 1;
+    if (nextKey != TSDB_DATA_TIMESTAMP_NULL && (nextKey >= minKey && nextKey <= maxKey)) return 1;
   }
   return 0;
 }
@@ -779,5 +843,21 @@ static int tsdbAdjustMemMaxTables(SMemTable *pMemTable, int maxTables) {
 
   taosTFree(tData);
 
+  return 0;
+}
+
+static int tsdbAppendTableRowToCols(STable *pTable, SDataCols *pCols, STSchema **ppSchema, SDataRow row) {
+  if (pCols) {
+    if (*ppSchema == NULL || schemaVersion(*ppSchema) != dataRowVersion(row)) {
+      *ppSchema = tsdbGetTableSchemaImpl(pTable, false, false, dataRowVersion(row));
+      if (*ppSchema == NULL) {
+        ASSERT(false);
+        return -1;
+      }
+    }
+
+    tdAppendDataRowToDataCol(row, *ppSchema, pCols);
+  }
+
   return 0;
 }
\ No newline at end of file

src/tsdb/src/tsdbMeta.c

@@ -86,7 +86,8 @@ int tsdbCreateTable(TSDB_REPO_T *repo, STableCfg *pCfg) {
   if (pTable != NULL) {
     tsdbError("vgId:%d table %s already exists, tid %d uid %" PRId64, REPO_ID(pRepo), TABLE_CHAR_NAME(pTable),
               TABLE_TID(pTable), TABLE_UID(pTable));
-    return TSDB_CODE_TDB_TABLE_ALREADY_EXIST;
+    terrno = TSDB_CODE_TDB_TABLE_ALREADY_EXIST;
+    goto _err;
   }
 
   if (pCfg->type == TSDB_CHILD_TABLE) {
@@ -643,7 +644,7 @@ static void tsdbOrgMeta(void *pHandle) {
 }
 
 static char *getTagIndexKey(const void *pData) {
-  STable *pTable = *(STable **)pData;
+  STable *pTable = (STable *)pData;
 
   STSchema *pSchema = tsdbGetTableTagSchema(pTable);
   STColumn *pCol = schemaColAt(pSchema, DEFAULT_TAG_INDEX_COLUMN);
@@ -700,7 +701,7 @@ static STable *tsdbCreateTableFromCfg(STableCfg *pCfg, bool isSuper) {
     }
     pTable->tagVal = NULL;
     STColumn *pCol = schemaColAt(pTable->tagSchema, DEFAULT_TAG_INDEX_COLUMN);
-    pTable->pIndex = tSkipListCreate(TSDB_SUPER_TABLE_SL_LEVEL, colType(pCol), (uint8_t)(colBytes(pCol)), 1, 0, 1, getTagIndexKey);
+    pTable->pIndex = tSkipListCreate(TSDB_SUPER_TABLE_SL_LEVEL, colType(pCol), (uint8_t)(colBytes(pCol)), NULL, SL_ALLOW_DUP_KEY, getTagIndexKey);
     if (pTable->pIndex == NULL) {
       terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
       goto _err;
@@ -900,23 +901,8 @@ static int tsdbAddTableIntoIndex(STsdbMeta *pMeta, STable *pTable, bool refSuper
 
   pTable->pSuper = pSTable;
 
-  int32_t level = 0;
-  int32_t headSize = 0;
-
-  tSkipListNewNodeInfo(pSTable->pIndex, &level, &headSize);
-
-  // NOTE: do not allocate the space for key, since in each skip list node, only keep the pointer to pTable, not the
-  // actual key value, and the key value will be retrieved during query through the pTable and getTagIndexKey function
-  SSkipListNode *pNode = calloc(1, headSize + sizeof(STable *));
-  if (pNode == NULL) {
-    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
-    return -1;
-  }
-  pNode->level = level;
-
-  memcpy(SL_GET_NODE_DATA(pNode), &pTable, sizeof(STable *));
+  tSkipListPut(pSTable->pIndex, (void *)pTable);
 
-  tSkipListPut(pSTable->pIndex, pNode);
   if (refSuper) T_REF_INC(pSTable);
   return 0;
 }
@@ -940,7 +926,7 @@ static int tsdbRemoveTableFromIndex(STsdbMeta *pMeta, STable *pTable) {
     SSkipListNode *pNode = taosArrayGetP(res, i);
 
     // STableIndexElem* pElem = (STableIndexElem*) SL_GET_NODE_DATA(pNode);
-    if (*(STable **)SL_GET_NODE_DATA(pNode) == pTable) {  // this is the exact what we need
+    if ((STable *)SL_GET_NODE_DATA(pNode) == pTable) {  // this is the exact what we need
       tSkipListRemoveNode(pSTable->pIndex, pNode);
     }
   }
@@ -1170,8 +1156,8 @@ static void *tsdbDecodeTable(void *buf, STable **pRTable) {
     if (TABLE_TYPE(pTable) == TSDB_SUPER_TABLE) {
       buf = tdDecodeSchema(buf, &(pTable->tagSchema));
       STColumn *pCol = schemaColAt(pTable->tagSchema, DEFAULT_TAG_INDEX_COLUMN);
-      pTable->pIndex =
-          tSkipListCreate(TSDB_SUPER_TABLE_SL_LEVEL, colType(pCol), (uint8_t)(colBytes(pCol)), 1, 0, 1, getTagIndexKey);
+      pTable->pIndex = tSkipListCreate(TSDB_SUPER_TABLE_SL_LEVEL, colType(pCol), (uint8_t)(colBytes(pCol)), NULL,
+                                       SL_ALLOW_DUP_KEY, getTagIndexKey);
       if (pTable->pIndex == NULL) {
         terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
         tsdbFreeTable(pTable);
@@ -1197,7 +1183,7 @@ static int tsdbGetTableEncodeSize(int8_t act, STable *pTable) {
     tlen = sizeof(SListNode) + sizeof(SActObj) + sizeof(SActCont) + tsdbEncodeTable(NULL, pTable) + sizeof(TSCKSUM);
   } else {
     if (TABLE_TYPE(pTable) == TSDB_SUPER_TABLE) {
-      tlen = (int)((sizeof(SListNode) + sizeof(SActObj)) * (tSkipListGetSize(pTable->pIndex) + 1));
+      tlen = (int)((sizeof(SListNode) + sizeof(SActObj)) * (SL_SIZE(pTable->pIndex) + 1));
     } else {
       tlen = sizeof(SListNode) + sizeof(SActObj);
     }
@@ -1244,7 +1230,7 @@ static int tsdbRemoveTableFromStore(STsdbRepo *pRepo, STable *pTable) {
     }
 
     while (tSkipListIterNext(pIter)) {
-      STable *tTable = *(STable **)SL_GET_NODE_DATA(tSkipListIterGet(pIter));
+      STable *tTable = (STable *)SL_GET_NODE_DATA(tSkipListIterGet(pIter));
       ASSERT(TABLE_TYPE(tTable) == TSDB_CHILD_TABLE);
       pBuf = tsdbInsertTableAct(pRepo, TSDB_DROP_META, pBuf, tTable);
     }
@@ -1269,7 +1255,7 @@ static int tsdbRmTableFromMeta(STsdbRepo *pRepo, STable *pTable) {
     tsdbWLockRepoMeta(pRepo);
 
     while (tSkipListIterNext(pIter)) {
-      STable *tTable = *(STable **)SL_GET_NODE_DATA(tSkipListIterGet(pIter));
+      STable *tTable = (STable *)SL_GET_NODE_DATA(tSkipListIterGet(pIter));
       tsdbRemoveTableFromMeta(pRepo, tTable, false, false);
     }
 

src/util/CMakeLists.txt

@@ -8,7 +8,7 @@ TARGET_LINK_LIBRARIES(tutil pthread osdetail lz4 z)
   
 IF (TD_LINUX)
   TARGET_LINK_LIBRARIES(tutil m rt)
-  ADD_SUBDIRECTORY(tests)
+  # ADD_SUBDIRECTORY(tests)
 
   FIND_PATH(ICONV_INCLUDE_EXIST iconv.h /usr/include/ /usr/local/include/)
   IF (ICONV_INCLUDE_EXIST)

src/util/tests/skiplistTest.cpp

@@ -247,7 +247,7 @@ void skiplistPerformanceTest() {
   printf("total:%" PRIu64 " ms, avg:%f\n", e - s, (e - s) / (double)size);
   printf("max level of skiplist:%d, actually level:%d\n ", pSkipList->maxLevel, pSkipList->level);
 
-  assert(tSkipListGetSize(pSkipList) == size);
+  assert(SL_GET_SIZE(pSkipList) == size);
 
   //  printf("the level of skiplist is:\n");
   //
@@ -273,7 +273,7 @@ void skiplistPerformanceTest() {
 
   int64_t et = taosGetTimestampMs();
   printf("delete %d data from skiplist, elapased time:%" PRIu64 "ms\n", 10000, et - st);
-  assert(tSkipListGetSize(pSkipList) == size);
+  assert(SL_GET_SIZE(pSkipList) == size);
 
   tSkipListDestroy(pSkipList);
   taosTFree(total);

src/vnode/src/vnodeWrite.c

@@ -97,7 +97,7 @@ int32_t vnodeProcessWrite(void *vparam, void *wparam, int32_t qtype, void *rpara
   return syncCode;
 }
 
-int32_t vnodeCheckWrite(void *param) {
+static int32_t vnodeCheckWrite(void *param) {
   SVnodeObj *pVnode = param;
   if (!(pVnode->accessState & TSDB_VN_WRITE_ACCCESS)) {
     vDebug("vgId:%d, no write auth, recCount:%d pVnode:%p", pVnode->vgId, pVnode->refCount, pVnode);

tests/pytest/crash_gen_bootstrap.py

@@ -11,7 +11,7 @@
 ###################################################################
 
 import sys
-from crash_gen.crash_gen import MainExec
+from crash_gen.crash_gen_main import MainExec
 
 if __name__ == "__main__":
     

tests/script/general/http/restful_full.sim

@@ -81,7 +81,7 @@ print ===============  step2 - no db
 #11
 system_content curl -H 'Authorization: Taosd /KfeAzX/f9na8qdtNZmtONryp201ma04bEl8LcvLUd7a8qdtNZmtONryp201ma04'  -d 'show databases' 127.0.0.1:7111/rest/sql
 print 11-> $system_content
-if $system_content != @{"status":"succ","head":["name","created_time","ntables","vgroups","replica","quorum","days","keep1,keep2,keep(D)","cache(MB)","blocks","minrows","maxrows","wallevel","fsync","comp","precision","status"],"data":[],"rows":0}@ then
+if $system_content != @{"status":"succ","head":["name","created_time","ntables","vgroups","replica","quorum","days","keep1,keep2,keep(D)","cache(MB)","blocks","minrows","maxrows","wallevel","fsync","comp","precision","update","status"],"data":[],"rows":0}@ then
   return -1
 endi