Merge remote-tracking branch 'origin/3.0' into feature/3.0_wxy

include/util/tskiplist2.h → include/common/tdata.h

@@ -12,59 +12,34 @@
  * You should have received a copy of the GNU Affero General Public License
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
-#ifndef _TD_UTIL_SKIPLIST2_H_
-#define _TD_UTIL_SKIPLIST2_H_
 
-#include "os.h"
+#ifndef _TD_TDATA_H_
+#define _TD_TDATA_H_
 
 #ifdef __cplusplus
 extern "C" {
 #endif
 
-#define SL_MAX_LEVEL 15
-
-typedef struct SSkipList2 SSkipList2;
-typedef struct SSLCursor  SSLCursor;
-typedef struct SSLCfg     SSLCfg;
-typedef struct SSLNode    SSLNode;
-
-typedef int32_t (*tslCmprFn)(const void *pKey1, int32_t nKey1, const void *pKey2, int32_t nKey2);
-
-// SSkipList2
-int32_t slOpen(const SSLCfg *pCfg, SSkipList2 **ppSl);
-int32_t slClose(SSkipList2 *pSl);
-int32_t slClear(SSkipList2 *pSl);
+#include "os.h"
 
-// SSLCursor
-int32_t slcOpen(SSkipList2 *pSl, SSLCursor *pSlc);
-int32_t slcClose(SSLCursor *pSlc);
-int32_t slcMoveTo(SSLCursor *pSlc, const void *pKey, int32_t nKey);
-int32_t slcMoveToNext(SSLCursor *pSlc);
-int32_t slcMoveToPrev(SSLCursor *pSlc);
-int32_t slcMoveToFirst(SSLCursor *pSlc);
-int32_t slcMoveToLast(SSLCursor *pSlc);
-int32_t slcPut(SSLCursor *pSlc, const void *pKey, int32_t nKey, const void *pData, int32_t nData);
-int32_t slcGet(SSLCursor *pSlc, const void **ppKey, int32_t *nKey, const void **ppData, int32_t *nData);
-int32_t slcDrop(SSLCursor *pSlc);
+typedef struct STaosData TDATA, tdata_t;
 
-// struct
-struct SSLCfg {
-  int8_t    maxLevel;
-  int32_t   nKey;
-  int32_t   nData;
-  tslCmprFn cmprFn;
-  void     *pPool;
-  void *(*xMalloc)(void *, int32_t size);
-  void (*xFree)(void *, void *);
-};
+typedef enum {
+  TAOS_META_STABLE_DATA = 0,  // super table meta
+  TAOS_META_TABLE_DATA,       // non-super table meta
+  TAOS_TS_ROW_DATA,           // row time-series data
+  TAOS_TS_COL_DATA,           // col time-series data
+  TAOS_DATA_MAX
+} ETaosDataT;
 
-struct SSLCursor {
-  SSkipList2 *pSl;
-  SSLNode   **forwards[SL_MAX_LEVEL];
+struct STaosData {
+  ETaosDataT type;
+  uint32_t   nPayload;
+  uint8_t   *pPayload;
 };
 
 #ifdef __cplusplus
 }
 #endif
 
-#endif /*_TD_UTIL_SKIPLIST2_H_*/
+#endif /*_TD_TDATA_H_*/
\ No newline at end of file

include/common/tdataformat.h

@@ -150,29 +150,6 @@ int32_t   tdAddColToSchema(STSchemaBuilder *pBuilder, int8_t type, int8_t flags,
 STSchema *tdGetSchemaFromBuilder(STSchemaBuilder *pBuilder);
 
 // ----------------- Semantic timestamp key definition
-#ifdef TD_2_0
-
-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)TABS(key)) | (TKEY_NEGATIVE_FLAG & (TKEY)(key)))
-#define tdGetKey(tkey)         (((TSKEY)((tkey)&TKEY_VALUE_FILTER)) * (TKEY_IS_NEGATIVE(tkey) ? -1 : 1))
-
-#define MIN_TS_KEY ((TSKEY)0x8000000000000001)
-#define MAX_TS_KEY ((TSKEY)0x3fffffffffffffff)
-
-#define TD_TO_TKEY(key) tdGetTKEY(((key) < MIN_TS_KEY) ? MIN_TS_KEY : (((key) > MAX_TS_KEY) ? MAX_TS_KEY : key))
-
-#else
-
 // typedef uint64_t TKEY;
 #define TKEY TSKEY
 
@@ -192,8 +169,6 @@ typedef uint64_t TKEY;
 
 #define TD_TO_TKEY(key) tdGetTKEY(((key) < MIN_TS_KEY) ? MIN_TS_KEY : (((key) > MAX_TS_KEY) ? MAX_TS_KEY : key))
 
-#endif
-
 static FORCE_INLINE TKEY keyToTkey(TSKEY key) {
   TSKEY lkey = key;
   if (key > MAX_TS_KEY) {
@@ -218,157 +193,6 @@ static FORCE_INLINE int32_t tkeyComparFn(const void *tkey1, const void *tkey2) {
   }
 }
 
-#if 0
-// ----------------- Data row structure
-
-/* A data row, the format is like below:
- * |<------------------------------------------------ len ---------------------------------->|
- * |<--     Head      -->|<---------   flen -------------->|                                 |
- * +---------------------+---------------------------------+---------------------------------+
- * | uint16_t |  int16_t |                                 |                                 |
- * +----------+----------+---------------------------------+---------------------------------+
- * |   len    | sversion |           First part            |             Second part         |
- * +----------+----------+---------------------------------+---------------------------------+
- *
- * NOTE: timestamp in this row structure is TKEY instead of TSKEY
- */
-typedef void *SDataRow;
-
-#define TD_DATA_ROW_HEAD_SIZE (sizeof(uint16_t) + sizeof(int16_t))
-
-#define dataRowLen(r)                (*(TDRowLenT *)(r))  // 0~65535
-#define dataRowEnd(r)                POINTER_SHIFT(r, dataRowLen(r))
-#define dataRowVersion(r)            (*(int16_t *)POINTER_SHIFT(r, sizeof(int16_t)))
-#define dataRowTuple(r)              POINTER_SHIFT(r, TD_DATA_ROW_HEAD_SIZE)
-#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);
-void     tdInitDataRow(SDataRow row, STSchema *pSchema);
-SDataRow tdDataRowDup(SDataRow row);
-
-// offset here not include dataRow header length
-static FORCE_INLINE int32_t tdAppendDataColVal(SDataRow row, const void *value, bool isCopyVarData, int8_t type,
-                                           int32_t offset) {
-  assert(value != NULL);
-  int32_t toffset = offset + TD_DATA_ROW_HEAD_SIZE;
-
-  if (IS_VAR_DATA_TYPE(type)) {
-    *(VarDataOffsetT *)POINTER_SHIFT(row, toffset) = dataRowLen(row);
-    if (isCopyVarData) {
-      memcpy(POINTER_SHIFT(row, dataRowLen(row)), 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), (const void *)(&tvalue), TYPE_BYTES[type]);
-    } else {
-      memcpy(POINTER_SHIFT(row, toffset), value, TYPE_BYTES[type]);
-    }
-  }
-
-  return 0;
-}
-
-// offset here not include dataRow header length
-static FORCE_INLINE int32_t tdAppendColVal(SDataRow row, const void *value, int8_t type, int32_t offset) {
-  return tdAppendDataColVal(row, value, true, type, offset);
-}
-
-// NOTE: offset here including the header size
-static FORCE_INLINE void *tdGetRowDataOfCol(SDataRow row, int8_t type, int32_t offset) {
-  if (IS_VAR_DATA_TYPE(type)) {
-    return POINTER_SHIFT(row, *(VarDataOffsetT *)POINTER_SHIFT(row, offset));
-  } else {
-    return POINTER_SHIFT(row, offset);
-  }
-}
-
-static FORCE_INLINE void *tdGetPtrToCol(SDataRow row, STSchema *pSchema, int32_t idx) {
-  return POINTER_SHIFT(row, TD_DATA_ROW_HEAD_SIZE + pSchema->columns[idx].offset);
-}
-
-static FORCE_INLINE void *tdGetColOfRowBySchema(SDataRow row, STSchema *pSchema, int32_t idx) {
-  int16_t offset = TD_DATA_ROW_HEAD_SIZE + pSchema->columns[idx].offset;
-  int8_t  type = pSchema->columns[idx].type;
-
-  return tdGetRowDataOfCol(row, type, offset);
-}
-
-static FORCE_INLINE bool tdIsColOfRowNullBySchema(SDataRow row, STSchema *pSchema, int32_t idx) {
-  int16_t offset = TD_DATA_ROW_HEAD_SIZE + pSchema->columns[idx].offset;
-  int8_t  type = pSchema->columns[idx].type;
-
-  return isNull(tdGetRowDataOfCol(row, type, offset), type);
-}
-
-static FORCE_INLINE void tdSetColOfRowNullBySchema(SDataRow row, STSchema *pSchema, int32_t idx) {
-  int16_t offset = TD_DATA_ROW_HEAD_SIZE + pSchema->columns[idx].offset;
-  int8_t  type = pSchema->columns[idx].type;
-  int16_t bytes = pSchema->columns[idx].bytes;
-
-  setNull(tdGetRowDataOfCol(row, type, offset), type, bytes);
-}
-
-static FORCE_INLINE void tdCopyColOfRowBySchema(SDataRow dst, STSchema *pDstSchema, int32_t dstIdx, SDataRow src,
-                                                STSchema *pSrcSchema, int32_t srcIdx) {
-  int8_t type = pDstSchema->columns[dstIdx].type;
-  assert(type == pSrcSchema->columns[srcIdx].type);
-  void *pData = tdGetPtrToCol(dst, pDstSchema, dstIdx);
-  void *value = tdGetPtrToCol(src, pSrcSchema, srcIdx);
-
-  switch (type) {
-    case TSDB_DATA_TYPE_BINARY:
-    case TSDB_DATA_TYPE_NCHAR:
-      *(VarDataOffsetT *)pData = *(VarDataOffsetT *)value;
-      pData = POINTER_SHIFT(dst, *(VarDataOffsetT *)pData);
-      value = POINTER_SHIFT(src, *(VarDataOffsetT *)value);
-      memcpy(pData, value, varDataTLen(value));
-      break;
-    case TSDB_DATA_TYPE_NULL:
-    case TSDB_DATA_TYPE_BOOL:
-    case TSDB_DATA_TYPE_TINYINT:
-    case TSDB_DATA_TYPE_UTINYINT:
-      *(uint8_t *)pData = *(uint8_t *)value;
-      break;
-    case TSDB_DATA_TYPE_SMALLINT:
-    case TSDB_DATA_TYPE_USMALLINT:
-      *(uint16_t *)pData = *(uint16_t *)value;
-      break;
-    case TSDB_DATA_TYPE_INT:
-    case TSDB_DATA_TYPE_UINT:
-      *(uint32_t *)pData = *(uint32_t *)value;
-      break;
-    case TSDB_DATA_TYPE_BIGINT:
-    case TSDB_DATA_TYPE_UBIGINT:
-      *(uint64_t *)pData = *(uint64_t *)value;
-      break;
-    case TSDB_DATA_TYPE_FLOAT:
-      SET_FLOAT_PTR(pData, value);
-      break;
-    case TSDB_DATA_TYPE_DOUBLE:
-      SET_DOUBLE_PTR(pData, value);
-      break;
-    case TSDB_DATA_TYPE_TIMESTAMP:
-      if (pSrcSchema->columns[srcIdx].colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
-        *(TSKEY *)pData = tdGetKey(*(TKEY *)value);
-      } else {
-        *(TSKEY *)pData = *(TSKEY *)value;
-      }
-      break;
-    default:
-      memcpy(pData, value, pSrcSchema->columns[srcIdx].bytes);
-  }
-}
-#endif
 // ----------------- Data column structure
 // SDataCol arrangement: data => bitmap => dataOffset
 typedef struct SDataCol {
@@ -398,29 +222,6 @@ void   *dataColSetOffset(SDataCol *pCol, int32_t nEle);
 
 bool isNEleNull(SDataCol *pCol, int32_t nEle);
 
-#if 0
-// Get the data pointer from a column-wised data
-static FORCE_INLINE const void *tdGetColDataOfRow(SDataCol *pCol, int32_t row) {
-  if (isAllRowsNull(pCol)) {
-    return getNullValue(pCol->type);
-  }
-  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, int32_t rows) {
-  assert(rows > 0);
-
-  if (IS_VAR_DATA_TYPE(pDataCol->type)) {
-    return pDataCol->dataOff[rows - 1] + varDataTLen(tdGetColDataOfRow(pDataCol, rows - 1));
-  } else {
-    return TYPE_BYTES[pDataCol->type] * rows;
-  }
-}
-#endif
 typedef struct {
   col_id_t  maxCols;    // max number of columns
   col_id_t  numOfCols;  // Total number of cols
@@ -479,7 +280,8 @@ void       tdResetDataCols(SDataCols *pCols);
 int32_t    tdInitDataCols(SDataCols *pCols, STSchema *pSchema);
 SDataCols *tdDupDataCols(SDataCols *pCols, bool keepData);
 SDataCols *tdFreeDataCols(SDataCols *pCols);
-int32_t tdMergeDataCols(SDataCols *target, SDataCols *source, int32_t rowsToMerge, int32_t *pOffset, bool update, TDRowVerT maxVer);
+int32_t    tdMergeDataCols(SDataCols *target, SDataCols *source, int32_t rowsToMerge, int32_t *pOffset, bool update,
+                           TDRowVerT maxVer);
 
 // ----------------- K-V data row structure
 /* |<-------------------------------------- len -------------------------------------------->|
@@ -542,54 +344,6 @@ static FORCE_INLINE void *tdGetKVRowIdxOfCol(SKVRow row, int16_t colId) {
   return taosbsearch(&colId, kvRowColIdx(row), kvRowNCols(row), sizeof(SColIdx), comparTagId, TD_EQ);
 }
 
-#if 0
-// offset here not include kvRow header length
-static FORCE_INLINE int32_t tdAppendKvColVal(SKVRow row, const void *value, bool isCopyValData, int16_t colId, int8_t type,
-                                         int32_t offset) {
-  assert(value != NULL);
-  int32_t  toffset = offset + TD_KV_ROW_HEAD_SIZE;
-  SColIdx *pColIdx = (SColIdx *)POINTER_SHIFT(row, toffset);
-  char *   ptr = (char *)POINTER_SHIFT(row, kvRowLen(row));
-
-  pColIdx->colId = colId;
-  pColIdx->offset = kvRowLen(row);  // offset of pColIdx including the TD_KV_ROW_HEAD_SIZE
-
-  if (IS_VAR_DATA_TYPE(type)) {
-    if (isCopyValData) {
-      memcpy(ptr, value, varDataTLen(value));
-    }
-    kvRowLen(row) += varDataTLen(value);
-  } else {
-    if (offset == 0) {
-      assert(type == TSDB_DATA_TYPE_TIMESTAMP);
-      TKEY tvalue = tdGetTKEY(*(TSKEY *)value);
-      memcpy(ptr, (void *)(&tvalue), TYPE_BYTES[type]);
-    } else {
-      memcpy(ptr, value, TYPE_BYTES[type]);
-    }
-    kvRowLen(row) += TYPE_BYTES[type];
-  }
-
-  return 0;
-}
-// NOTE: offset here including the header size
-static FORCE_INLINE void *tdGetKvRowDataOfCol(void *row, int32_t offset) { return POINTER_SHIFT(row, offset); }
-
-static FORCE_INLINE void *tdGetKVRowValOfColEx(SKVRow row, int16_t colId, int32_t *nIdx) {
-  while (*nIdx < kvRowNCols(row)) {
-    SColIdx *pColIdx = kvRowColIdxAt(row, *nIdx);
-    if (pColIdx->colId == colId) {
-      ++(*nIdx);
-      return tdGetKvRowDataOfCol(row, pColIdx->offset);
-    } else if (pColIdx->colId > colId) {
-      return NULL;
-    } else {
-      ++(*nIdx);
-    }
-  }
-  return NULL;
-}
-#endif
 // ----------------- K-V data row builder
 typedef struct {
   int16_t  tCols;
@@ -632,166 +386,6 @@ static FORCE_INLINE int32_t tdAddColToKVRow(SKVRowBuilder *pBuilder, col_id_t co
 
   return 0;
 }
-#if 0
-// ----------------- SMemRow appended with tuple row structure
-/*
- * |---------|------------------------------------------------- len ---------------------------------->|
- * |<--------     Head      ------>|<---------   flen -------------->|                                 |
- * |---------+---------------------+---------------------------------+---------------------------------+
- * | uint8_t | uint16_t |  int16_t |                                 |                                 |
- * |---------+----------+----------+---------------------------------+---------------------------------+
- * |  flag   |   len    | sversion |           First part            |             Second part         |
- * +---------+----------+----------+---------------------------------+---------------------------------+
- *
- * NOTE: timestamp in this row structure is TKEY instead of TSKEY
- */
-
-// ----------------- SMemRow appended with extended K-V data row structure
-/* |--------------------|------------------------------------------------  len ---------------------------------->|
- * |<-------------     Head      ------------>|<---------   flen -------------->|                                 |
- * |--------------------+----------+--------------------------------------------+---------------------------------+
- * | uint8_t | int16_t  | uint16_t |  int16_t |                                 |                                 |
- * |---------+----------+----------+----------+---------------------------------+---------------------------------+
- * |   flag  | sversion |   len    |   ncols  |           cols index            |             data part           |
- * |---------+----------+----------+----------+---------------------------------+---------------------------------+
- */
-
-typedef void *SMemRow;
-
-#define TD_MEM_ROW_TYPE_SIZE        sizeof(uint8_t)
-#define TD_MEM_ROW_KV_VER_SIZE      sizeof(int16_t)
-#define TD_MEM_ROW_KV_TYPE_VER_SIZE (TD_MEM_ROW_TYPE_SIZE + TD_MEM_ROW_KV_VER_SIZE)
-#define TD_MEM_ROW_DATA_HEAD_SIZE   (TD_MEM_ROW_TYPE_SIZE + TD_DATA_ROW_HEAD_SIZE)
-#define TD_MEM_ROW_KV_HEAD_SIZE     (TD_MEM_ROW_TYPE_SIZE + TD_MEM_ROW_KV_VER_SIZE + TD_KV_ROW_HEAD_SIZE)
-
-#define SMEM_ROW_DATA 0x0U   // SDataRow
-#define SMEM_ROW_KV   0x01U  // SKVRow
-
-#define KVRatioConvert (0.9f)
-
-#define memRowType(r) ((*(uint8_t *)(r)) & 0x01)
-
-#define memRowSetType(r, t)  ((*(uint8_t *)(r)) = (t))  // set the total byte in case of dirty memory
-#define isDataRowT(t)        (SMEM_ROW_DATA == (((uint8_t)(t)) & 0x01))
-#define isDataRow(r)         (SMEM_ROW_DATA == memRowType(r))
-#define isKvRowT(t)          (SMEM_ROW_KV == (((uint8_t)(t)) & 0x01))
-#define isKvRow(r)           (SMEM_ROW_KV == memRowType(r))
-#define isUtilizeKVRow(k, d) ((k) < ((d)*KVRatioConvert))
-
-#define memRowDataBody(r) POINTER_SHIFT(r, TD_MEM_ROW_TYPE_SIZE)  // section after flag
-#define memRowKvBody(r) \
-  POINTER_SHIFT(r, TD_MEM_ROW_KV_TYPE_VER_SIZE)  // section after flag + sversion as to reuse SKVRow
-
-#define memRowDataLen(r) (*(TDRowLenT *)memRowDataBody(r))  //  0~65535
-#define memRowKvLen(r)   (*(TDRowLenT *)memRowKvBody(r))    //  0~65535
-
-#define memRowDataTLen(r) \
-  ((TDRowLenT)(memRowDataLen(r) + TD_MEM_ROW_TYPE_SIZE))  // using uint32_t/int32_t to store the TLen
-
-#define memRowKvTLen(r) ((TDRowLenT)(memRowKvLen(r) + TD_MEM_ROW_KV_TYPE_VER_SIZE))
-
-#define memRowLen(r)  (isDataRow(r) ? memRowDataLen(r) : memRowKvLen(r))
-#define memRowTLen(r) (isDataRow(r) ? memRowDataTLen(r) : memRowKvTLen(r))  // using uint32_t/int32_t to store the TLen
-
-static FORCE_INLINE char *memRowEnd(SMemRow row) {
-  if (isDataRow(row)) {
-    return (char *)dataRowEnd(memRowDataBody(row));
-  } else {
-    return (char *)kvRowEnd(memRowKvBody(row));
-  }
-}
-
-#define memRowDataVersion(r)     dataRowVersion(memRowDataBody(r))
-#define memRowKvVersion(r)       (*(int16_t *)POINTER_SHIFT(r, TD_MEM_ROW_TYPE_SIZE))
-#define memRowVersion(r)         (isDataRow(r) ? memRowDataVersion(r) : memRowKvVersion(r))  // schema version
-#define memRowSetKvVersion(r, v) (memRowKvVersion(r) = (v))
-#define memRowTuple(r)           (isDataRow(r) ? dataRowTuple(memRowDataBody(r)) : kvRowValues(memRowKvBody(r)))
-
-#define memRowTKey(r) (isDataRow(r) ? dataRowTKey(memRowDataBody(r)) : kvRowTKey(memRowKvBody(r)))
-#define memRowKey(r)  (isDataRow(r) ? dataRowKey(memRowDataBody(r)) : kvRowKey(memRowKvBody(r)))
-#define memRowKeys(r) (isDataRow(r) ? dataRowTuple(memRowDataBody(r)) : kvRowKeys(memRowKvBody(r)))
-#define memRowSetTKey(r, k)                 \
-  do {                                      \
-    if (isDataRow(r)) {                     \
-      dataRowTKey(memRowDataBody(r)) = (k); \
-    } else {                                \
-      kvRowTKey(memRowKvBody(r)) = (k);     \
-    }                                       \
-  } while (0)
-
-#define memRowSetLen(r, l)          (isDataRow(r) ? memRowDataLen(r) = (l) : memRowKvLen(r) = (l))
-#define memRowSetVersion(r, v)      (isDataRow(r) ? dataRowSetVersion(memRowDataBody(r), v) : memRowSetKvVersion(r, v))
-#define memRowCpy(dst, r)           memcpy((dst), (r), memRowTLen(r))
-#define memRowMaxBytesFromSchema(s) (schemaTLen(s) + TD_MEM_ROW_DATA_HEAD_SIZE)
-#define memRowDeleted(r)            TKEY_IS_DELETED(memRowTKey(r))
-
-SMemRow tdMemRowDup(SMemRow row);
-void    tdAppendMemRowToDataCol(SMemRow row, STSchema *pSchema, SDataCols *pCols, bool forceSetNull);
-
-// NOTE: offset here including the header size
-static FORCE_INLINE void *tdGetMemRowDataOfCol(void *row, int16_t colId, int8_t colType, uint16_t offset) {
-  if (isDataRow(row)) {
-    return tdGetRowDataOfCol(memRowDataBody(row), colType, offset);
-  } else {
-    return tdGetKVRowValOfCol(memRowKvBody(row), colId);
-  }
-}
-
-/**
- * NOTE:
- *  1. Applicable to scan columns one by one
- *  2. offset here including the header size
- */
-static FORCE_INLINE void *tdGetMemRowDataOfColEx(void *row, int16_t colId, int8_t colType, int32_t offset,
-                                                 int32_t *kvNIdx) {
-  if (isDataRow(row)) {
-    return tdGetRowDataOfCol(memRowDataBody(row), colType, offset);
-  } else {
-    return tdGetKVRowValOfColEx(memRowKvBody(row), colId, kvNIdx);
-  }
-}
-
-static FORCE_INLINE int32_t tdAppendMemRowColVal(SMemRow row, const void *value, bool isCopyVarData, int16_t colId,
-                                             int8_t type, int32_t offset) {
-  if (isDataRow(row)) {
-    tdAppendDataColVal(memRowDataBody(row), value, isCopyVarData, type, offset);
-  } else {
-    tdAppendKvColVal(memRowKvBody(row), value, isCopyVarData, colId, type, offset);
-  }
-  return 0;
-}
-
-// make sure schema->flen appended for SDataRow
-static FORCE_INLINE int32_t tdGetColAppendLen(uint8_t rowType, const void *value, int8_t colType) {
-  int32_t len = 0;
-  if (IS_VAR_DATA_TYPE(colType)) {
-    len += varDataTLen(value);
-    if (rowType == SMEM_ROW_KV) {
-      len += sizeof(SColIdx);
-    }
-  } else {
-    if (rowType == SMEM_ROW_KV) {
-      len += TYPE_BYTES[colType];
-      len += sizeof(SColIdx);
-    }
-  }
-  return len;
-}
-
-typedef struct {
-  int16_t colId;
-  uint8_t colType;
-  char *  colVal;
-} SColInfo;
-
-static FORCE_INLINE void setSColInfo(SColInfo *colInfo, int16_t colId, uint8_t colType, char *colVal) {
-  colInfo->colId = colId;
-  colInfo->colType = colType;
-  colInfo->colVal = colVal;
-}
-
-SMemRow mergeTwoMemRows(void *buffer, SMemRow row1, SMemRow row2, STSchema *pSchema1, STSchema *pSchema2);
-#endif
 
 #ifdef __cplusplus
 }

include/common/tmsg.h

@@ -1458,7 +1458,7 @@ typedef struct {
 static FORCE_INLINE SMqRebInfo* tNewSMqRebSubscribe(const char* key) {
   SMqRebInfo* pRebInfo = (SMqRebInfo*)taosMemoryCalloc(1, sizeof(SMqRebInfo));
   if (pRebInfo == NULL) {
-    goto _err;
+    return NULL;
   }
   strcpy(pRebInfo->key, key);
   pRebInfo->lostConsumers = taosArrayInit(0, sizeof(int64_t));

include/common/tname.h

@@ -16,8 +16,8 @@
 #ifndef _TD_COMMON_NAME_H_
 #define _TD_COMMON_NAME_H_
 
-#include "tdef.h"
 #include "tarray.h"
+#include "tdef.h"
 
 #ifdef __cplusplus
 extern "C" {
@@ -65,19 +65,19 @@ bool tNameDBNameEqual(SName* left, SName* right);
 
 typedef struct {
   // input
-  SArray         *tags;         // element is SSmlKV
-  const char     *sTableName;   // super table name
-  uint8_t        sTableNameLen; // the length of super table name
+  SArray*     tags;           // element is SSmlKv
+  const char* sTableName;     // super table name
+  uint8_t     sTableNameLen;  // the length of super table name
 
   // output
-  char           *childTableName;   // must have size of TSDB_TABLE_NAME_LEN;
-  uint64_t       uid;               // child table uid, may be useful
+  char*    childTableName;  // must have size of TSDB_TABLE_NAME_LEN;
+  uint64_t uid;             // child table uid, may be useful
 } RandTableName;
 
-void buildChildTableName(RandTableName *rName);
+void buildChildTableName(RandTableName* rName);
 
 #ifdef __cplusplus
 }
 #endif
 
-#endif  /*_TD_COMMON_NAME_H_*/
+#endif /*_TD_COMMON_NAME_H_*/

include/libs/function/functionMgt.h

@@ -31,13 +31,13 @@ typedef enum EFunctionType {
   FUNCTION_TYPE_ELAPSED,
   FUNCTION_TYPE_IRATE,
   FUNCTION_TYPE_LAST_ROW,
-  FUNCTION_TYPE_LEASTSQUARES,
   FUNCTION_TYPE_MAX,
   FUNCTION_TYPE_MIN,
   FUNCTION_TYPE_MODE,
   FUNCTION_TYPE_PERCENTILE,
   FUNCTION_TYPE_SPREAD,
   FUNCTION_TYPE_STDDEV,
+  FUNCTION_TYPE_LEASTSQUARES,
   FUNCTION_TYPE_SUM,
   FUNCTION_TYPE_TWA,
   FUNCTION_TYPE_HISTOGRAM,

include/libs/qcom/query.h

@@ -184,7 +184,7 @@ extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t
 #define SET_META_TYPE_TABLE(t)      (t) = META_TYPE_TABLE
 #define SET_META_TYPE_BOTH_TABLE(t) (t) = META_TYPE_BOTH_TABLE
 
-#define NEED_CLIENT_RM_TBLMETA_ERROR(_code) ((_code) == TSDB_CODE_TDB_INVALID_TABLE_ID || (_code) == TSDB_CODE_VND_TB_NOT_EXIST)
+#define NEED_CLIENT_RM_TBLMETA_ERROR(_code) ((_code) == TSDB_CODE_PAR_TABLE_NOT_EXIST || (_code) == TSDB_CODE_VND_TB_NOT_EXIST)
 #define NEED_CLIENT_REFRESH_VG_ERROR(_code) ((_code) == TSDB_CODE_VND_HASH_MISMATCH || (_code) == TSDB_CODE_VND_INVALID_VGROUP_ID)
 #define NEED_CLIENT_REFRESH_TBLMETA_ERROR(_code) ((_code) == TSDB_CODE_TDB_TABLE_RECREATED)
 #define NEED_CLIENT_HANDLE_ERROR(_code) (NEED_CLIENT_RM_TBLMETA_ERROR(_code) || NEED_CLIENT_REFRESH_VG_ERROR(_code) || NEED_CLIENT_REFRESH_TBLMETA_ERROR(_code))

include/libs/stream/tstream.h

@@ -16,6 +16,7 @@
 #include "tdatablock.h"
 #include "tmsg.h"
 #include "tmsgcb.h"
+#include "tqueue.h"
 #include "trpc.h"
 
 #ifdef __cplusplus
@@ -154,6 +155,10 @@ struct SStreamTask {
     STaskDispatcherShuffle shuffleDispatcher;
   };
 
+  // msg buffer
+  int32_t     memUsed;
+  STaosQueue* inputQ;
+
   // application storage
   void* ahandle;
 };
@@ -194,6 +199,8 @@ typedef struct {
   SArray* res;  // SArray<SSDataBlock>
 } SStreamSinkReq;
 
+int32_t streamEnqueueData(SStreamTask* pTask, const void* input, int32_t inputType);
+
 int32_t streamExecTask(SStreamTask* pTask, SMsgCb* pMsgCb, const void* input, int32_t inputType, int32_t workId);
 
 #ifdef __cplusplus

include/util/taoserror.h

@@ -88,6 +88,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_CFG_NOT_FOUND                 TAOS_DEF_ERROR_CODE(0, 0x0114)
 #define TSDB_CODE_REPEAT_INIT                   TAOS_DEF_ERROR_CODE(0, 0x0115)
 #define TSDB_CODE_DUP_KEY                       TAOS_DEF_ERROR_CODE(0, 0x0116)
+#define TSDB_CODE_NEED_RETRY                    TAOS_DEF_ERROR_CODE(0, 0x0117)
 
 #define TSDB_CODE_REF_NO_MEMORY                 TAOS_DEF_ERROR_CODE(0, 0x0140)
 #define TSDB_CODE_REF_FULL                      TAOS_DEF_ERROR_CODE(0, 0x0141)

source/client/inc/clientInt.h

@@ -310,6 +310,7 @@ void hbMgrInitMqHbRspHandle();
 SRequestObj* launchQueryImpl(SRequestObj* pRequest, SQuery* pQuery, int32_t code, bool keepQuery);
 int32_t      getQueryPlan(SRequestObj* pRequest, SQuery* pQuery, SArray** pNodeList);
 int32_t      scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList);
+int32_t refreshMeta(STscObj* pTscObj, SRequestObj* pRequest);
 
 #ifdef __cplusplus
 }

source/client/inc/clientStmt.h

@@ -60,6 +60,7 @@ typedef struct SStmtBindInfo {
   int32_t      sBindRowNum;
   int32_t      sBindLastIdx;
   int8_t       tbType;
+  bool         tagsCached;
   void*        boundTags;  
   char*        tbName;
   SName        sname;

source/client/src/clientSml.c

@@ -405,7 +405,7 @@ static int32_t smlModifyDBSchemas(SSmlHandle* info) {
 
     code = catalogGetSTableMeta(info->pCatalog, info->taos->pAppInfo->pTransporter, &ep, &pName, &pTableMeta);
 
-    if (code == TSDB_CODE_TDB_INVALID_TABLE_ID || code == TSDB_CODE_MND_INVALID_STB) {
+    if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST || code == TSDB_CODE_MND_INVALID_STB) {
       SSchemaAction schemaAction = {0};
       schemaAction.action = SCHEMA_ACTION_CREATE_STABLE;
       memcpy(schemaAction.createSTable.sTableName, superTable, superTableLen);

source/client/src/clientStmt.c

@@ -29,6 +29,11 @@ int32_t stmtSwitchStatus(STscStmt* pStmt, STMT_STATUS newStatus) {
       if (STMT_STATUS_EQ(INIT) || STMT_STATUS_EQ(BIND_COL)) {
         code = TSDB_CODE_TSC_STMT_API_ERROR;
       }
+/*      
+      if ((pStmt->sql.type == STMT_TYPE_MULTI_INSERT) && ()) {
+        code = TSDB_CODE_TSC_STMT_API_ERROR;
+      }
+*/
       break;
     case STMT_BIND_COL:
       if (STMT_STATUS_EQ(INIT) || STMT_STATUS_EQ(BIND)) {
@@ -123,6 +128,7 @@ int32_t stmtSetBindInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags) {
   pStmt->bInfo.tbSuid = pTableMeta->suid;
   pStmt->bInfo.tbType = pTableMeta->tableType;
   pStmt->bInfo.boundTags = tags;
+  pStmt->bInfo.tagsCached = false;
 
   return TSDB_CODE_SUCCESS;
 }
@@ -207,8 +213,6 @@ int32_t stmtParseSql(STscStmt* pStmt) {
       STMT_ERR_RET(TSDB_CODE_TSC_STMT_CLAUSE_ERROR);
   }
 
-  STMT_ERR_RET(stmtCacheBlock(pStmt));
-
   return TSDB_CODE_SUCCESS;
 }
 
@@ -219,8 +223,10 @@ int32_t stmtCleanBindInfo(STscStmt* pStmt) {
   pStmt->bInfo.needParse = true;
 
   taosMemoryFreeClear(pStmt->bInfo.tbName);
-  destroyBoundColumnInfo(pStmt->bInfo.boundTags);
-  taosMemoryFreeClear(pStmt->bInfo.boundTags);
+  if (!pStmt->bInfo.tagsCached) {
+    destroyBoundColumnInfo(pStmt->bInfo.boundTags);
+    taosMemoryFreeClear(pStmt->bInfo.boundTags);
+  }
 
   return TSDB_CODE_SUCCESS;
 }
@@ -275,6 +281,7 @@ int32_t stmtCleanSQLInfo(STscStmt* pStmt) {
 
     qDestroyStmtDataBlock(pCache->pDataBlock);
     destroyBoundColumnInfo(pCache->boundTags);
+    taosMemoryFreeClear(pCache->boundTags);
     
     pIter = taosHashIterate(pStmt->sql.pTableCache, pIter);
   }
@@ -302,7 +309,15 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
 
   STableMeta *pTableMeta = NULL;
   SEpSet ep = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
-  STMT_ERR_RET(catalogGetTableMeta(pStmt->pCatalog, pStmt->taos->pAppInfo->pTransporter, &ep, &pStmt->bInfo.sname, &pTableMeta));
+  int32_t code = catalogGetTableMeta(pStmt->pCatalog, pStmt->taos->pAppInfo->pTransporter, &ep, &pStmt->bInfo.sname, &pTableMeta);
+  if (TSDB_CODE_PAR_TABLE_NOT_EXIST == code) {
+    STMT_ERR_RET(stmtCleanBindInfo(pStmt));
+    
+    return TSDB_CODE_SUCCESS;
+  }
+
+  STMT_ERR_RET(code);
+  
   uint64_t uid = pTableMeta->uid;
   uint64_t suid = pTableMeta->suid;
   int8_t tableType = pTableMeta->tableType;
@@ -328,6 +343,7 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
     pStmt->bInfo.tbSuid = suid;
     pStmt->bInfo.tbType = tableType;
     pStmt->bInfo.boundTags = pCache->boundTags;
+    pStmt->bInfo.tagsCached = true;
 
     return TSDB_CODE_SUCCESS;
   }
@@ -340,6 +356,7 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
     pStmt->bInfo.tbSuid = suid;
     pStmt->bInfo.tbType = tableType;
     pStmt->bInfo.boundTags = pCache->boundTags;
+    pStmt->bInfo.tagsCached = true;
 
     STableDataBlocks* pNewBlock = NULL;
     STMT_ERR_RET(qRebuildStmtDataBlock(&pNewBlock, pCache->pDataBlock));
@@ -448,10 +465,12 @@ int stmtSetTbTags(TAOS_STMT *stmt, TAOS_MULTI_BIND *tags) {
 
   STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTAGS));
 
-  if (pStmt->bInfo.needParse) {
-    STMT_ERR_RET(stmtParseSql(pStmt));
+  if (!pStmt->bInfo.needParse) {
+    return TSDB_CODE_SUCCESS;
   }
 
+  STMT_ERR_RET(stmtParseSql(pStmt));
+
   STableDataBlocks **pDataBlock = (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, (const char*)&pStmt->bInfo.tbUid, sizeof(pStmt->bInfo.tbUid));
   if (NULL == pDataBlock) {
     tscError("table uid %" PRIx64 "not found in exec blockHash", pStmt->bInfo.tbUid);
@@ -501,8 +520,6 @@ int32_t stmtFetchColFields(STscStmt* pStmt, int32_t *fieldNum, TAOS_FIELD** fiel
 int stmtBindBatch(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind, int32_t colIdx) {
   STscStmt* pStmt = (STscStmt*)stmt;
 
-  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_BIND));
-
   if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 && STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
     pStmt->bInfo.needParse = false;
   }
@@ -520,6 +537,8 @@ int stmtBindBatch(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind, int32_t colIdx) {
     STMT_ERR_RET(stmtParseSql(pStmt));
   }
 
+  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_BIND));
+
   if (STMT_TYPE_QUERY == pStmt->sql.type) {
     if (NULL == pStmt->sql.pQueryPlan) {
       STMT_ERR_RET(getQueryPlan(pStmt->exec.pRequest, pStmt->sql.pQuery, &pStmt->sql.nodeList));
@@ -586,6 +605,16 @@ int stmtExec(TAOS_STMT *stmt) {
     STMT_ERR_RET(qBuildStmtOutput(pStmt->sql.pQuery, pStmt->exec.pVgHash, pStmt->exec.pBlockHash));
     launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, TSDB_CODE_SUCCESS, true);
   }
+
+  if (pStmt->exec.pRequest->code && NEED_CLIENT_HANDLE_ERROR(pStmt->exec.pRequest->code)) {
+    code = refreshMeta(pStmt->exec.pRequest->pTscObj, pStmt->exec.pRequest);
+    if (code) {
+      pStmt->exec.pRequest->code = code;
+    } else {
+      STMT_ERR_RET(stmtResetStmt(pStmt));
+      STMT_ERR_RET(TSDB_CODE_NEED_RETRY);
+    }
+  }
   
   STMT_ERR_JRET(pStmt->exec.pRequest->code);
 
@@ -613,13 +642,11 @@ int stmtClose(TAOS_STMT *stmt) {
 const char *stmtErrstr(TAOS_STMT *stmt) {
   STscStmt* pStmt = (STscStmt*)stmt;
 
-  if (stmt == NULL) {
+  if (stmt == NULL || NULL == pStmt->exec.pRequest) {
     return (char*) tstrerror(terrno);
   }
 
-  if (pStmt->exec.pRequest) {
-    pStmt->exec.pRequest->code = terrno;
-  }
+  pStmt->exec.pRequest->code = terrno;
 
   return taos_errstr(pStmt->exec.pRequest);
 }

source/client/src/tmq.c

@@ -187,7 +187,7 @@ typedef struct {
 
 tmq_conf_t* tmq_conf_new() {
   tmq_conf_t* conf = taosMemoryCalloc(1, sizeof(tmq_conf_t));
-  conf->autoCommit = false;
+  conf->autoCommit = true;
   conf->autoCommitInterval = 5000;
   conf->resetOffset = TMQ_CONF__RESET_OFFSET__EARLIEAST;
   return conf;

source/common/src/tdata.c

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
\ No newline at end of file

source/common/src/tdatablock.c

@@ -493,7 +493,7 @@ SSDataBlock* blockDataExtractBlock(SSDataBlock* pBlock, int32_t startIndex, int3
     for (int32_t j = startIndex; j < (startIndex + rowCount); ++j) {
       bool isNull = false;
       if (pBlock->pBlockAgg == NULL) {
-        isNull = colDataIsNull_s(pColData, pBlock->info.rows);
+        isNull = colDataIsNull_s(pColData, j);
       } else {
         isNull = colDataIsNull(pColData, pBlock->info.rows, j, pBlock->pBlockAgg[i]);
       }

source/common/src/tdataformat.c

@@ -129,50 +129,6 @@ void *tdDecodeSchema(void *buf, STSchema **pRSchema) {
   return buf;
 }
 
-#if 0
-int32_t tEncodeSTColumn(SCoder *pEncoder, const STColumn *pCol) {
-  if (tEncodeI16(pEncoder, pCol->colId) < 0) return -1;
-  if (tEncodeI8(pEncoder, pCol->type) < 0) return -1;
-  if (tEncodeI8(pEncoder, pCol->sma) < 0) return -1;
-  if (tEncodeI32(pEncoder, pCol->bytes) < 0) return -1;
-  if (tEncodeI32(pEncoder, pCol->offset) < 0) return -1;
-  return pEncoder->pos;
-}
-
-int32_t tDecodeSTColumn(SCoder *pDecoder, STColumn *pCol) {
-  if (tDecodeI16(pDecoder, &pCol->colId) < 0) return -1;
-  if (tDecodeI8(pDecoder, &pCol->type) < 0) return -1;
-  if (tDecodeI8(pDecoder, &pCol->sma) < 0) return -1;
-  if (tDecodeI32(pDecoder, &pCol->bytes) < 0) return -1;
-  if (tDecodeI32(pDecoder, &pCol->offset) < 0) return -1;
-  return 0;
-}
-
-int32_t tEncodeSchema(SCoder *pEncoder, const STSchema *pSchema) {
-  if (tEncodeI32(pEncoder, pSchema->numOfCols) < 0) return -1;
-  if (tEncodeI16(pEncoder, pSchema->version) < 0) return -1;
-  if (tEncodeU16(pEncoder, pSchema->flen) < 0) return -1;
-  if (tEncodeI32(pEncoder, pSchema->vlen) < 0) return -1;
-  if (tEncodeI32(pEncoder, pSchema->tlen) < 0) return -1;
-
-  for (int32_t i = 0; i < schemaNCols(pSchema); i++) {
-    const STColumn *pCol = schemaColAt(pSchema, i);
-    if (tEncodeSTColumn(pEncoder, pCol) < 0) return -1;
-  }
-  return 0;
-}
-
-int32_t tDecodeSchema(SCoder *pDecoder, STSchema *pSchema) {
-  if (tDecodeI32(pDecoder, &pSchema->numOfCols) < 0) return -1;
-  if (tDecodeI16(pDecoder, &pSchema->version) < 0) return -1;
-  if (tDecodeU16(pDecoder, &pSchema->flen) < 0) return -1;
-  if (tDecodeI32(pDecoder, &pSchema->vlen) < 0) return -1;
-  if (tDecodeI32(pDecoder, &pSchema->tlen) < 0) return -1;
-
-  return 0;
-}
-#endif
-
 int tdInitTSchemaBuilder(STSchemaBuilder *pBuilder, schema_ver_t version) {
   if (pBuilder == NULL) return -1;
 
@@ -260,49 +216,6 @@ STSchema *tdGetSchemaFromBuilder(STSchemaBuilder *pBuilder) {
   return pSchema;
 }
 
-#if 0
-/**
- * Initialize a data row
- */
-void tdInitDataRow(SDataRow row, STSchema *pSchema) {
-  dataRowSetLen(row, TD_DATA_ROW_HEAD_SIZE + schemaFLen(pSchema));
-  dataRowSetVersion(row, schemaVersion(pSchema));
-}
-
-SDataRow tdNewDataRowFromSchema(STSchema *pSchema) {
-  int32_t size = dataRowMaxBytesFromSchema(pSchema);
-
-  SDataRow row = taosMemoryMalloc(size);
-  if (row == NULL) return NULL;
-
-  tdInitDataRow(row, pSchema);
-  return row;
-}
-
-/**
- * Free the SDataRow object
- */
-void tdFreeDataRow(SDataRow row) {
-  if (row) taosMemoryFree(row);
-}
-
-SDataRow tdDataRowDup(SDataRow row) {
-  SDataRow trow = taosMemoryMalloc(dataRowLen(row));
-  if (trow == NULL) return NULL;
-
-  dataRowCpy(trow, row);
-  return trow;
-}
-
-SMemRow tdMemRowDup(SMemRow row) {
-  SMemRow trow = taosMemoryMalloc(memRowTLen(row));
-  if (trow == NULL) return NULL;
-
-  memRowCpy(trow, row);
-  return trow;
-}
-#endif
-
 void dataColInit(SDataCol *pDataCol, STColumn *pCol, int maxPoints) {
   pDataCol->type = colType(pCol);
   pDataCol->colId = colColId(pCol);
@@ -312,39 +225,6 @@ void dataColInit(SDataCol *pDataCol, STColumn *pCol, int maxPoints) {
   pDataCol->len = 0;
 }
 
-#if 0
-// value from timestamp should be TKEY here instead of TSKEY
-int dataColAppendVal(SDataCol *pCol, const void *value, int numOfRows, int maxPoints) {
-  ASSERT(pCol != NULL && value != NULL);
-
-  if (isAllRowsNull(pCol)) {
-    if (isNull(value, pCol->type)) {
-      // all null value yet, just return
-      return 0;
-    }
-
-    if (tdAllocMemForCol(pCol, maxPoints) < 0) return -1;
-    if (numOfRows > 0) {
-      // Find the first not null value, fill all previouse values as NULL
-      dataColSetNEleNull(pCol, numOfRows);
-    }
-  }
-
-  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);
-    memcpy(POINTER_SHIFT(pCol->pData, pCol->len), value, pCol->bytes);
-    pCol->len += pCol->bytes;
-  }
-  return 0;
-}
-#endif
 static FORCE_INLINE const void *tdGetColDataOfRowUnsafe(SDataCol *pCol, int row) {
   if (IS_VAR_DATA_TYPE(pCol->type)) {
     return POINTER_SHIFT(pCol->pData, pCol->dataOff[row]);
@@ -361,31 +241,6 @@ bool isNEleNull(SDataCol *pCol, int nEle) {
   return true;
 }
 
-#if 0
-static FORCE_INLINE void dataColSetNullAt(SDataCol *pCol, int index) {
-  if (IS_VAR_DATA_TYPE(pCol->type)) {
-    pCol->dataOff[index] = pCol->len;
-    char *ptr = POINTER_SHIFT(pCol->pData, pCol->len);
-    setVardataNull(ptr, pCol->type);
-    pCol->len += varDataTLen(ptr);
-  } else {
-    setNull(POINTER_SHIFT(pCol->pData, TYPE_BYTES[pCol->type] * index), pCol->type, pCol->bytes);
-    pCol->len += TYPE_BYTES[pCol->type];
-  }
-}
-
-static void dataColSetNEleNull(SDataCol *pCol, int nEle, int8_t bitmapMode) {
-  if (IS_VAR_DATA_TYPE(pCol->type)) {
-    pCol->len = 0;
-    for (int i = 0; i < nEle; ++i) {
-      dataColSetNullAt(pCol, i);
-    }
-  } else {
-    setNullN(pCol->pData, pCol->type, pCol->bytes, nEle);
-    pCol->len = TYPE_BYTES[pCol->type] * nEle;
-  }
-}
-#endif
 void *dataColSetOffset(SDataCol *pCol, int nEle) {
   ASSERT(((pCol->type == TSDB_DATA_TYPE_BINARY) || (pCol->type == TSDB_DATA_TYPE_NCHAR)));
 
@@ -483,42 +338,6 @@ SDataCols *tdFreeDataCols(SDataCols *pCols) {
   return NULL;
 }
 
-#if 0
-SDataCols *tdDupDataCols(SDataCols *pDataCols, bool keepData) {
-  SDataCols *pRet = tdNewDataCols(pDataCols->maxCols, pDataCols->maxPoints);
-  if (pRet == NULL) return NULL;
-
-  pRet->numOfCols = pDataCols->numOfCols;
-  pRet->sversion = pDataCols->sversion;
-  if (keepData) pRet->numOfRows = pDataCols->numOfRows;
-
-  for (int i = 0; i < pDataCols->numOfCols; i++) {
-    pRet->cols[i].type = pDataCols->cols[i].type;
-    pRet->cols[i].bitmap = pDataCols->cols[i].bitmap;
-    pRet->cols[i].colId = pDataCols->cols[i].colId;
-    pRet->cols[i].bytes = pDataCols->cols[i].bytes;
-    pRet->cols[i].offset = pDataCols->cols[i].offset;
-
-    if (keepData) {
-      if (pDataCols->cols[i].len > 0) {
-        if (tdAllocMemForCol(&pRet->cols[i], pRet->maxPoints) < 0) {
-          tdFreeDataCols(pRet);
-          return NULL;
-        }
-        pRet->cols[i].len = pDataCols->cols[i].len;
-        memcpy(pRet->cols[i].pData, pDataCols->cols[i].pData, pDataCols->cols[i].len);
-        if (IS_VAR_DATA_TYPE(pRet->cols[i].type)) {
-          int dataOffSize = sizeof(VarDataOffsetT) * pDataCols->maxPoints;
-          memcpy(pRet->cols[i].dataOff, pDataCols->cols[i].dataOff, dataOffSize);
-        }
-      }
-    }
-  }
-
-  return pRet;
-}
-#endif
-
 void tdResetDataCols(SDataCols *pCols) {
   if (pCols != NULL) {
     pCols->numOfRows = 0;
@@ -528,180 +347,6 @@ void tdResetDataCols(SDataCols *pCols) {
     }
   }
 }
-#if 0
-static void tdAppendDataRowToDataCol(SDataRow row, STSchema *pSchema, SDataCols *pCols, bool forceSetNull) {
-  ASSERT(pCols->numOfRows == 0 || dataColsKeyLast(pCols) < dataRowKey(row));
-
-  int rcol = 0;
-  int dcol = 0;
-
-  while (dcol < pCols->numOfCols) {
-    bool setCol = 0;
-    SDataCol *pDataCol = &(pCols->cols[dcol]);
-    if (rcol >= schemaNCols(pSchema)) {
-      dataColAppendVal(pDataCol, getNullValue(pDataCol->type), pCols->numOfRows, pCols->maxPoints);
-      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);
-      if(!isNull(value, pDataCol->type)) setCol = 1;
-      dataColAppendVal(pDataCol, value, pCols->numOfRows, pCols->maxPoints);
-      dcol++;
-      rcol++;
-    } else if (pRowCol->colId < pDataCol->colId) {
-      rcol++;
-    } else {
-      if(forceSetNull || setCol) {
-        dataColAppendVal(pDataCol, getNullValue(pDataCol->type), pCols->numOfRows, pCols->maxPoints);
-      }
-      dcol++;
-    }
-  }
-  pCols->numOfRows++;
-}
-
-static void tdAppendKVRowToDataCol(SKVRow row, STSchema *pSchema, SDataCols *pCols, bool forceSetNull) {
-  ASSERT(pCols->numOfRows == 0 || dataColsKeyLast(pCols) < kvRowKey(row));
-
-  int rcol = 0;
-  int dcol = 0;
-
-  int nRowCols = kvRowNCols(row);
-
-  while (dcol < pCols->numOfCols) {
-    bool setCol = 0;
-    SDataCol *pDataCol = &(pCols->cols[dcol]);
-    if (rcol >= nRowCols || rcol >= schemaNCols(pSchema)) {
-      dataColAppendVal(pDataCol, getNullValue(pDataCol->type), pCols->numOfRows, pCols->maxPoints);
-      ++dcol;
-      continue;
-    }
-
-    SColIdx *colIdx = kvRowColIdxAt(row, rcol);
-
-    if (colIdx->colId == pDataCol->colId) {
-      void *value = tdGetKvRowDataOfCol(row, colIdx->offset);
-      if(!isNull(value, pDataCol->type)) setCol = 1;
-      dataColAppendVal(pDataCol, value, pCols->numOfRows, pCols->maxPoints);
-      ++dcol;
-      ++rcol;
-    } else if (colIdx->colId < pDataCol->colId) {
-      ++rcol;
-    } else {
-      if(forceSetNull || setCol) {
-        dataColAppendVal(pDataCol, getNullValue(pDataCol->type), pCols->numOfRows, pCols->maxPoints);
-      }
-      ++dcol;
-    }
-  }
-  pCols->numOfRows++;
-}
-
-void tdAppendMemRowToDataCol(SMemRow row, STSchema *pSchema, SDataCols *pCols, bool forceSetNull) {
-  if (isDataRow(row)) {
-    tdAppendDataRowToDataCol(memRowDataBody(row), pSchema, pCols, forceSetNull);
-  } else if (isKvRow(row)) {
-    tdAppendKVRowToDataCol(memRowKvBody(row), pSchema, pCols, forceSetNull);
-  } else {
-    ASSERT(0);
-  }
-}
-
-int tdMergeDataCols(SDataCols *target, SDataCols *source, int rowsToMerge, int *pOffset, bool forceSetNull) {
-  ASSERT(rowsToMerge > 0 && rowsToMerge <= source->numOfRows);
-  ASSERT(target->numOfCols == source->numOfCols);
-  int offset = 0;
-
-  if (pOffset == NULL) {
-    pOffset = &offset;
-  }
-
-  SDataCols *pTarget = NULL;
-
-  if ((target->numOfRows == 0) || (dataColsKeyLast(target) < dataColsKeyAtRow(source, *pOffset))) {  // 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 || target->cols[j].len > 0) {
-          dataColAppendVal(target->cols + j, tdGetColDataOfRow(source->cols + j, i + (*pOffset)), target->numOfRows,
-                           target->maxPoints);
-        }
-      }
-      target->numOfRows++;
-    }
-    (*pOffset) += rowsToMerge;
-  } else {
-    pTarget = tdDupDataCols(target, true);
-    if (pTarget == NULL) goto _err;
-
-    int iter1 = 0;
-    tdMergeTwoDataCols(target, pTarget, &iter1, pTarget->numOfRows, source, pOffset, source->numOfRows,
-                       pTarget->numOfRows + rowsToMerge, forceSetNull);
-  }
-
-  tdFreeDataCols(pTarget);
-  return 0;
-
-_err:
-  tdFreeDataCols(pTarget);
-  return -1;
-}
-
-// 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, bool forceSetNull) {
-  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 : 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);
-
-    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 || target->cols[i].len > 0) {
-          dataColAppendVal(&(target->cols[i]), tdGetColDataOfRow(src1->cols + i, *iter1), target->numOfRows,
-                           target->maxPoints);
-        }
-      }
-
-      target->numOfRows++;
-      (*iter1)++;
-    } 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 && !isNull(src2->cols[i].pData, src2->cols[i].type)) {
-            dataColAppendVal(&(target->cols[i]), tdGetColDataOfRow(src2->cols + i, *iter2), target->numOfRows,
-                             target->maxPoints);
-          } else if(!forceSetNull && key1 == key2 && src1->cols[i].len > 0) {
-            dataColAppendVal(&(target->cols[i]), tdGetColDataOfRow(src1->cols + i, *iter1), target->numOfRows,
-                             target->maxPoints);
-          } else if(target->cols[i].len > 0) {
-            dataColSetNullAt(&target->cols[i], target->numOfRows);
-          }
-        }
-        target->numOfRows++;
-      }
-
-      (*iter2)++;
-      if (key1 == key2) (*iter1)++;
-    }
-
-    ASSERT(target->numOfRows <= target->maxPoints);
-  }
-}
-#endif
 
 SKVRow tdKVRowDup(SKVRow row) {
   SKVRow trow = taosMemoryMalloc(kvRowLen(row));
@@ -859,98 +504,3 @@ SKVRow tdGetKVRowFromBuilder(SKVRowBuilder *pBuilder) {
 
   return row;
 }
-#if 0
-SMemRow mergeTwoMemRows(void *buffer, SMemRow row1, SMemRow row2, STSchema *pSchema1, STSchema *pSchema2) {
-#if 0
-  ASSERT(memRowKey(row1) == memRowKey(row2));
-  ASSERT(schemaVersion(pSchema1) == memRowVersion(row1));
-  ASSERT(schemaVersion(pSchema2) == memRowVersion(row2));
-  ASSERT(schemaVersion(pSchema1) >= schemaVersion(pSchema2));
-#endif
-
-  SArray *stashRow = taosArrayInit(pSchema1->numOfCols, sizeof(SColInfo));
-  if (stashRow == NULL) {
-    return NULL;
-  }
-
-  SMemRow  pRow = buffer;
-  SDataRow dataRow = memRowDataBody(pRow);
-  memRowSetType(pRow, SMEM_ROW_DATA);
-  dataRowSetVersion(dataRow, schemaVersion(pSchema1));  // use latest schema version
-  dataRowSetLen(dataRow, (TDRowLenT)(TD_DATA_ROW_HEAD_SIZE + pSchema1->flen));
-
-  TDRowLenT dataLen = 0, kvLen = TD_MEM_ROW_KV_HEAD_SIZE;
-
-  int32_t  i = 0;  // row1
-  int32_t  j = 0;  // row2
-  int32_t  nCols1 = schemaNCols(pSchema1);
-  int32_t  nCols2 = schemaNCols(pSchema2);
-  SColInfo colInfo = {0};
-  int32_t  kvIdx1 = 0, kvIdx2 = 0;
-
-  while (i < nCols1) {
-    STColumn *pCol = schemaColAt(pSchema1, i);
-    void *    val1 = tdGetMemRowDataOfColEx(row1, pCol->colId, pCol->type, TD_DATA_ROW_HEAD_SIZE + pCol->offset, &kvIdx1);
-    // if val1 != NULL, use val1;
-    if (val1 != NULL && !isNull(val1, pCol->type)) {
-      tdAppendColVal(dataRow, val1, pCol->type, pCol->offset);
-      kvLen += tdGetColAppendLen(SMEM_ROW_KV, val1, pCol->type);
-      setSColInfo(&colInfo, pCol->colId, pCol->type, val1);
-      taosArrayPush(stashRow, &colInfo);
-      ++i;  // next col
-      continue;
-    }
-
-    void *val2 = NULL;
-    while (j < nCols2) {
-      STColumn *tCol = schemaColAt(pSchema2, j);
-      if (tCol->colId < pCol->colId) {
-        ++j;
-        continue;
-      }
-      if (tCol->colId == pCol->colId) {
-        val2 = tdGetMemRowDataOfColEx(row2, tCol->colId, tCol->type, TD_DATA_ROW_HEAD_SIZE + tCol->offset, &kvIdx2);
-      } else if (tCol->colId > pCol->colId) {
-        // set NULL
-      }
-      break;
-    }  // end of while(j<nCols2)
-    if (val2 == NULL) {
-      val2 = (void *)getNullValue(pCol->type);
-    }
-    tdAppendColVal(dataRow, val2, pCol->type, pCol->offset);
-    if (!isNull(val2, pCol->type)) {
-      kvLen += tdGetColAppendLen(SMEM_ROW_KV, val2, pCol->type);
-      setSColInfo(&colInfo, pCol->colId, pCol->type, val2);
-      taosArrayPush(stashRow, &colInfo);
-    }
-
-    ++i;  // next col
-  }
-
-  dataLen = memRowTLen(pRow);
-
-  if (kvLen < dataLen) {
-    // scan stashRow and generate SKVRow
-    memset(buffer, 0, sizeof(dataLen));
-    SMemRow tRow = buffer;
-    memRowSetType(tRow, SMEM_ROW_KV);
-    SKVRow kvRow = (SKVRow)memRowKvBody(tRow);
-    int16_t nKvNCols = (int16_t) taosArrayGetSize(stashRow);
-    kvRowSetLen(kvRow, (TDRowLenT)(TD_KV_ROW_HEAD_SIZE + sizeof(SColIdx) * nKvNCols));
-    kvRowSetNCols(kvRow, nKvNCols);
-    memRowSetKvVersion(tRow, pSchema1->version);
-
-    int32_t toffset = 0;
-    int16_t k;
-    for (k = 0; k < nKvNCols; ++k) {
-      SColInfo *pColInfo = taosArrayGet(stashRow, k);
-      tdAppendKvColVal(kvRow, pColInfo->colVal, true, pColInfo->colId, pColInfo->colType, toffset);
-      toffset += sizeof(SColIdx);
-    }
-    ASSERT(kvLen == memRowTLen(tRow));
-  }
-  taosArrayDestroy(stashRow);
-  return buffer;
-}
-#endif

source/common/src/tmsg.c

@@ -1246,21 +1246,21 @@ int32_t tDeserializeSGetUserAuthRspImpl(SDecoder *pDecoder, SGetUserAuthRsp *pRs
   for (int32_t i = 0; i < numOfCreatedDbs; ++i) {
     char db[TSDB_DB_FNAME_LEN] = {0};
     if (tDecodeCStrTo(pDecoder, db) < 0) return -1;
-    int32_t len = strlen(db) + 1;
+    int32_t len = strlen(db);
     taosHashPut(pRsp->createdDbs, db, len, db, len);
   }
 
   for (int32_t i = 0; i < numOfReadDbs; ++i) {
     char db[TSDB_DB_FNAME_LEN] = {0};
     if (tDecodeCStrTo(pDecoder, db) < 0) return -1;
-    int32_t len = strlen(db) + 1;
+    int32_t len = strlen(db);
     taosHashPut(pRsp->readDbs, db, len, db, len);
   }
 
   for (int32_t i = 0; i < numOfWriteDbs; ++i) {
     char db[TSDB_DB_FNAME_LEN] = {0};
     if (tDecodeCStrTo(pDecoder, db) < 0) return -1;
-    int32_t len = strlen(db) + 1;
+    int32_t len = strlen(db);
     taosHashPut(pRsp->writeDbs, db, len, db, len);
   }
 

source/common/src/tname.c

@@ -18,11 +18,9 @@
 #include "tcommon.h"
 #include "tstrbuild.h"
 
-#define VALID_NAME_TYPE(x)  ((x) == TSDB_DB_NAME_T || (x) == TSDB_TABLE_NAME_T)
+#define VALID_NAME_TYPE(x) ((x) == TSDB_DB_NAME_T || (x) == TSDB_TABLE_NAME_T)
 
-bool tscValidateTableNameLength(size_t len) {
-  return len < TSDB_TABLE_NAME_LEN;
-}
+bool tscValidateTableNameLength(size_t len) { return len < TSDB_TABLE_NAME_LEN; }
 
 #if 0
 // TODO refactor
@@ -95,12 +93,12 @@ int64_t taosGetIntervalStartTimestamp(int64_t startTime, int64_t slidingTime, in
       * but in case of DST, the start time of one day need to be dynamically decided.
       */
       // todo refactor to extract function that is available for Linux/Windows/Mac platform
-  #if defined(WINDOWS) && _MSC_VER >= 1900
+#if defined(WINDOWS) && _MSC_VER >= 1900
       // see https://docs.microsoft.com/en-us/cpp/c-runtime-library/daylight-dstbias-timezone-and-tzname?view=vs-2019
       int64_t timezone = _timezone;
       int32_t daylight = _daylight;
       char**  tzname = _tzname;
-  #endif
+#endif
 
       int64_t t = (precision == TSDB_TIME_PRECISION_MILLI) ? MILLISECOND_PER_SECOND : MILLISECOND_PER_SECOND * 1000L;
       start += timezone * t;
@@ -142,10 +140,10 @@ int32_t tNameExtractFullName(const SName* name, char* dst) {
 int32_t tNameLen(const SName* name) {
   assert(name != NULL);
 
-  char tmp[12] = {0};
+  char    tmp[12] = {0};
   int32_t len = sprintf(tmp, "%d", name->acctId);
-  int32_t len1 = (int32_t) strlen(name->dbname);
-  int32_t len2 = (int32_t) strlen(name->tname);
+  int32_t len1 = (int32_t)strlen(name->dbname);
+  int32_t len2 = (int32_t)strlen(name->tname);
 
   if (name->type == TSDB_DB_NAME_T) {
     assert(len2 == 0);
@@ -200,9 +198,7 @@ const char* tNameGetTableName(const SName* name) {
   return &name->tname[0];
 }
 
-void tNameAssign(SName* dst, const SName* src) {
-  memcpy(dst, src, sizeof(SName));
-}
+void tNameAssign(SName* dst, const SName* src) { memcpy(dst, src, sizeof(SName)); }
 
 int32_t tNameSetDbName(SName* dst, int32_t acct, const char* dbName, size_t nameLen) {
   assert(dst != NULL && dbName != NULL && nameLen > 0);
@@ -244,7 +240,6 @@ bool tNameDBNameEqual(SName* left, SName* right) {
   return (0 == strcmp(left->dbname, right->dbname));
 }
 
-
 int32_t tNameFromString(SName* dst, const char* str, uint32_t type) {
   assert(dst != NULL && str != NULL && strlen(str) > 0);
 
@@ -260,14 +255,14 @@ int32_t tNameFromString(SName* dst, const char* str, uint32_t type) {
 
   if ((type & T_NAME_DB) == T_NAME_DB) {
     dst->type = TSDB_DB_NAME_T;
-    char* start = (char*)((p == NULL)? str:(p+1));
+    char* start = (char*)((p == NULL) ? str : (p + 1));
 
     int32_t len = 0;
     p = strstr(start, TS_PATH_DELIMITER);
     if (p == NULL) {
-      len = (int32_t) strlen(start);
+      len = (int32_t)strlen(start);
     } else {
-      len = (int32_t) (p - start);
+      len = (int32_t)(p - start);
     }
 
     // too long account id or too long db name
@@ -275,21 +270,21 @@ int32_t tNameFromString(SName* dst, const char* str, uint32_t type) {
       return -1;
     }
 
-    memcpy (dst->dbname, start, len);
+    memcpy(dst->dbname, start, len);
     dst->dbname[len] = 0;
   }
 
   if ((type & T_NAME_TABLE) == T_NAME_TABLE) {
     dst->type = TSDB_TABLE_NAME_T;
-    char* start = (char*) ((p == NULL)? str: (p+1));
+    char* start = (char*)((p == NULL) ? str : (p + 1));
 
     // too long account id or too long db name
-    int32_t len = (int32_t) strlen(start);
+    int32_t len = (int32_t)strlen(start);
     if ((len >= tListLen(dst->tname)) || (len <= 0)) {
       return -1;
     }
 
-    memcpy (dst->tname, start, len);
+    memcpy(dst->tname, start, len);
     dst->tname[len] = 0;
   }
 
@@ -305,14 +300,14 @@ static int compareKv(const void* p1, const void* p2) {
   if (res != 0) {
     return res;
   } else {
-    return kvLen1-kvLen2;
+    return kvLen1 - kvLen2;
   }
 }
 
 /*
  * use stable name and tags to grearate child table name
  */
-void buildChildTableName(RandTableName *rName) {
+void buildChildTableName(RandTableName* rName) {
   int32_t size = taosArrayGetSize(rName->tags);
   ASSERT(size > 0);
   taosArraySort(rName->tags, compareKv);
@@ -320,19 +315,19 @@ void buildChildTableName(RandTableName *rName) {
   SStringBuilder sb = {0};
   taosStringBuilderAppendStringLen(&sb, rName->sTableName, rName->sTableNameLen);
   for (int j = 0; j < size; ++j) {
-    SSmlKv *tagKv = taosArrayGetP(rName->tags, j);
+    SSmlKv* tagKv = taosArrayGetP(rName->tags, j);
     taosStringBuilderAppendStringLen(&sb, tagKv->key, tagKv->keyLen);
     taosStringBuilderAppendStringLen(&sb, tagKv->value, tagKv->valueLen);
   }
-  size_t len = 0;
-  char* keyJoined = taosStringBuilderGetResult(&sb, &len);
+  size_t    len = 0;
+  char*     keyJoined = taosStringBuilderGetResult(&sb, &len);
   T_MD5_CTX context;
   tMD5Init(&context);
-  tMD5Update(&context, (uint8_t *)keyJoined, (uint32_t)len);
+  tMD5Update(&context, (uint8_t*)keyJoined, (uint32_t)len);
   tMD5Final(&context);
   uint64_t digest1 = *(uint64_t*)(context.digest);
   uint64_t digest2 = *(uint64_t*)(context.digest + 8);
-  snprintf(rName->childTableName, TSDB_TABLE_NAME_LEN, "t_%016"PRIx64"%016"PRIx64, digest1, digest2);
+  snprintf(rName->childTableName, TSDB_TABLE_NAME_LEN, "t_%016" PRIx64 "%016" PRIx64, digest1, digest2);
   taosStringBuilderDestroy(&sb);
   rName->uid = digest1;
 }

source/dnode/mnode/impl/src/mndDb.c

@@ -1550,10 +1550,8 @@ static void dumpDbInfoData(SSDataBlock *pBlock, SDbObj *pDb, SShowObj *pShow, in
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
     colDataAppend(pColInfo, rows, (const char *)t, false);
 
-    //  single stable model
-    int8_t m = 0;
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
-    colDataAppend(pColInfo, rows, (const char *)&m, false);
+    colDataAppend(pColInfo, rows, (const char *)&pDb->cfg.numOfStables, false);
 
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols);
     colDataAppend(pColInfo, rows, (const char *)b, false);

source/dnode/mnode/impl/src/mndDef.c

@@ -196,7 +196,9 @@ SMqVgEp *tCloneSMqVgEp(const SMqVgEp *pVgEp) {
   return pVgEpNew;
 }
 
-void tDeleteSMqVgEp(SMqVgEp *pVgEp) { taosMemoryFree(pVgEp->qmsg); }
+void tDeleteSMqVgEp(SMqVgEp *pVgEp) {
+  if (pVgEp->qmsg) taosMemoryFree(pVgEp->qmsg);
+}
 
 int32_t tEncodeSMqVgEp(void **buf, const SMqVgEp *pVgEp) {
   int32_t tlen = 0;

source/dnode/mnode/impl/src/mndTopic.c

@@ -298,6 +298,8 @@ static int32_t mndCreateTopic(SMnode *pMnode, SNodeMsg *pReq, SCMCreateTopicReq
 
     SNode *pAst = NULL;
     if (nodesStringToNode(pCreate->ast, &pAst) != 0) {
+      taosMemoryFree(topicObj.ast);
+      taosMemoryFree(topicObj.sql);
       mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
       return -1;
     }
@@ -307,16 +309,22 @@ static int32_t mndCreateTopic(SMnode *pMnode, SNodeMsg *pReq, SCMCreateTopicReq
     SPlanContext cxt = {.pAstRoot = pAst, .topicQuery = true};
     if (qCreateQueryPlan(&cxt, &pPlan, NULL) != 0) {
       mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
+      taosMemoryFree(topicObj.ast);
+      taosMemoryFree(topicObj.sql);
       return -1;
     }
 
     if (qExtractResultSchema(pAst, &topicObj.schema.nCols, &topicObj.schema.pSchema) != 0) {
       mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
+      taosMemoryFree(topicObj.ast);
+      taosMemoryFree(topicObj.sql);
       return -1;
     }
 
     if (nodesNodeToString(pPlan, false, &topicObj.physicalPlan, NULL) != 0) {
       mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
+      taosMemoryFree(topicObj.ast);
+      taosMemoryFree(topicObj.sql);
       return -1;
     }
   } else {
@@ -331,6 +339,8 @@ static int32_t mndCreateTopic(SMnode *pMnode, SNodeMsg *pReq, SCMCreateTopicReq
   STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_CREATE_TOPIC, &pReq->rpcMsg);
   if (pTrans == NULL) {
     mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
+    taosMemoryFreeClear(topicObj.ast);
+    taosMemoryFreeClear(topicObj.sql);
     taosMemoryFreeClear(topicObj.physicalPlan);
     return -1;
   }

source/dnode/mnode/impl/src/mndUser.c

@@ -459,7 +459,7 @@ static int32_t mndProcessAlterUserReq(SNodeMsg *pReq) {
   }
 
   if (alterReq.alterType == TSDB_ALTER_USER_ADD_READ_DB || alterReq.alterType == TSDB_ALTER_USER_ADD_ALL_DB) {
-    if (strcmp(alterReq.dbname, "*") != 0) {
+    if (strcmp(alterReq.dbname, "1.*") != 0) {
       int32_t len = strlen(alterReq.dbname) + 1;
       SDbObj *pDb = mndAcquireDb(pMnode, alterReq.dbname);
       if (pDb == NULL) {
@@ -483,7 +483,7 @@ static int32_t mndProcessAlterUserReq(SNodeMsg *pReq) {
   }
 
   if (alterReq.alterType == TSDB_ALTER_USER_ADD_WRITE_DB || alterReq.alterType == TSDB_ALTER_USER_ADD_ALL_DB) {
-    if (strcmp(alterReq.dbname, "*") != 0) {
+    if (strcmp(alterReq.dbname, "1.*") != 0) {
       int32_t len = strlen(alterReq.dbname) + 1;
       SDbObj *pDb = mndAcquireDb(pMnode, alterReq.dbname);
       if (pDb == NULL) {

source/dnode/mnode/impl/test/user/user.cpp

@@ -367,8 +367,8 @@ TEST_F(MndTestUser, 03_Alter_User) {
     EXPECT_EQ(numOfReadDbs, 1);
     EXPECT_EQ(numOfWriteDbs, 0);
 
-    char* dbname = (char*)taosHashGet(authRsp.readDbs, "1.d2", 5);
-    EXPECT_STREQ(dbname, "1.d2");
+    char* dbname = (char*)taosHashGet(authRsp.readDbs, "1.d2", 4);
+    EXPECT_TRUE(dbname != NULL);
 
     taosHashCleanup(authRsp.readDbs);
     taosHashCleanup(authRsp.writeDbs);

source/dnode/vnode/src/tq/tq.c

@@ -233,16 +233,18 @@ int32_t tqPushMsgNew(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_
 int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver) {
   if (msgType != TDMT_VND_SUBMIT) return 0;
 
-  void* data = taosMemoryMalloc(msgLen);
-  if (data == NULL) {
+  // make sure msgType == TDMT_VND_SUBMIT
+  if (tsdbUpdateSmaWindow(pTq->pVnode->pTsdb, msg, ver) != 0) {
     return -1;
   }
-  memcpy(data, msg, msgLen);
 
-  // make sure msgType == TDMT_VND_SUBMIT
-  if (tsdbUpdateSmaWindow(pTq->pVnode->pTsdb, msg, ver) != 0) {
+  if (taosHashGetSize(pTq->pStreamTasks) == 0) return 0;
+
+  void* data = taosMemoryMalloc(msgLen);
+  if (data == NULL) {
     return -1;
   }
+  memcpy(data, msg, msgLen);
 
   SRpcMsg req = {
       .msgType = TDMT_VND_STREAM_TRIGGER,

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

@@ -211,7 +211,7 @@ int tsdbCommit(STsdb *pRepo) {
 
 void tsdbGetRtnSnap(STsdb *pRepo, SRtn *pRtn) {
   STsdbKeepCfg *pCfg = REPO_KEEP_CFG(pRepo);
-  TSKEY     minKey, midKey, maxKey, now;
+  TSKEY         minKey, midKey, maxKey, now;
 
   now = taosGetTimestamp(pCfg->precision);
   minKey = now - pCfg->keep2 * tsTickPerDay[pCfg->precision];
@@ -1386,34 +1386,7 @@ static void tsdbLoadAndMergeFromCache(SDataCols *pDataCols, int *iter, SCommitIt
 
       tSkipListIterNext(pCommitIter->pIter);
     } else {
-#if 0
-      if (update != TD_ROW_OVERWRITE_UPDATE) {
-        // copy disk data
-        for (int i = 0; i < pDataCols->numOfCols; ++i) {
-          // TODO: dataColAppendVal may fail
-          SCellVal sVal = {0};
-          if (tdGetColDataOfRow(&sVal, pDataCols->cols + i, *iter, pDataCols->bitmapMode) < 0) {
-            TASSERT(0);
-          }
-          tdAppendValToDataCol(pTarget->cols + i, sVal.valType, sVal.val, pTarget->numOfRows, pTarget->maxPoints, pTarget->bitmapMode);
-        }
-
-        if (update == TD_ROW_DISCARD_UPDATE) pTarget->numOfRows++;
-      }
-      if (update != TD_ROW_DISCARD_UPDATE) {
-        // copy mem data
-        if (pSchema == NULL || schemaVersion(pSchema) != TD_ROW_SVER(row)) {
-          pSchema = tsdbGetTableSchemaImpl(pCommitIter->pTable, false, false, TD_ROW_SVER(row));
-          ASSERT(pSchema != NULL);
-        }
-
-        tdAppendSTSRowToDataCol(row, pSchema, pTarget, update == TD_ROW_OVERWRITE_UPDATE);
-      }
-      ++(*iter);
-      tSkipListIterNext(pCommitIter->pIter);
-#endif
-
-      if(lastKey != key1) {
+      if (lastKey != key1) {
         lastKey = key1;
         ++pTarget->numOfRows;
       }
@@ -1484,29 +1457,4 @@ static bool tsdbCanAddSubBlock(SCommitH *pCommith, SBlock *pBlock, SMergeInfo *p
   }
 
   return false;
-}
-
+}
\ No newline at end of file
-// int tsdbApplyRtn(STsdbRepo *pRepo) {
-//   SRtn       rtn;
-//   SFSIter    fsiter;
-//   STsdbFS *  pfs = REPO_FS(pRepo);
-//   SDFileSet *pSet;
-
-//   // Get retention snapshot
-//   tsdbGetRtnSnap(pRepo, &rtn);
-
-//   tsdbFSIterInit(&fsiter, pfs, TSDB_FS_ITER_FORWARD);
-//   while ((pSet = tsdbFSIterNext(&fsiter))) {
-//     if (pSet->fid < rtn.minFid) {
-//       tsdbInfo("vgId:%d FSET %d at level %d disk id %d expires, remove it", REPO_ID(pRepo), pSet->fid,
-//                TSDB_FSET_LEVEL(pSet), TSDB_FSET_ID(pSet));
-//       continue;
-//     }
-
-//     if (tsdbApplyRtnOnFSet(pRepo, pSet, &rtn) < 0) {
-//       return -1;
-//     }
-//   }
-
-//   return 0;
-// }

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

@@ -248,11 +248,13 @@ int tsdbLoadDataFromCache(STable *pTable, SSkipListIterator *pIter, TSKEY maxKey
             pMergeInfo->nOperations++;
             pMergeInfo->keyFirst = TMIN(pMergeInfo->keyFirst, rowKey);
             pMergeInfo->keyLast = TMAX(pMergeInfo->keyLast, rowKey);
-            lastKey = rowKey;
             if (pCols) {
-              ++pCols->numOfRows;
+              if (lastKey != TSKEY_INITIAL_VAL) {
+                ++pCols->numOfRows;
+              }
               tsdbAppendTableRowToCols(pTable, pCols, &pSchema, row, false);
             }
+            lastKey = rowKey;
           } else {
             tsdbAppendTableRowToCols(pTable, pCols, &pSchema, row, true);
           }
@@ -326,6 +328,11 @@ int tsdbInsertTableData(STsdb *pTsdb, SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlo
 
   tSkipListPutBatchByIter(pTbData->pData, &blkIter, (iter_next_fn_t)tGetSubmitBlkNext);
 
+#ifdef TD_DEBUG_PRINT_ROW
+  printf("!!! %s:%d table %" PRIi64 " has %d rows in skiplist\n\n", __func__, __LINE__, pTbData->uid,
+         SL_SIZE(pTbData->pData));
+#endif
+
   // Set statistics
   keyMax = TD_ROW_KEY(blkIter.row);
 

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

@@ -21,16 +21,19 @@ typedef struct SMemSkipList       SMemSkipList;
 typedef struct SMemSkipListNode   SMemSkipListNode;
 typedef struct SMemSkipListCurosr SMemSkipListCurosr;
 
+#define SL_MAX_LEVEL 5
+
 struct SMemTable {
-  STsdb     *pTsdb;
-  TSKEY      minKey;
-  TSKEY      maxKey;
-  int64_t    minVer;
-  int64_t    maxVer;
-  int64_t    nRows;
-  int32_t    nHash;
-  int32_t    nBucket;
-  SMemData **pBuckets;
+  STsdb              *pTsdb;
+  TSKEY               minKey;
+  TSKEY               maxKey;
+  int64_t             minVer;
+  int64_t             maxVer;
+  int64_t             nRows;
+  int32_t             nHash;
+  int32_t             nBucket;
+  SMemData          **pBuckets;
+  SMemSkipListCurosr *pSlc;
 };
 
 struct SMemSkipListNode {
@@ -60,9 +63,15 @@ struct SMemData {
 
 struct SMemSkipListCurosr {
   SMemSkipList     *pSl;
-  SMemSkipListNode *pNodeC;
+  SMemSkipListNode *pNodes[SL_MAX_LEVEL];
 };
 
+typedef struct {
+  int64_t       version;
+  uint32_t      szRow;
+  const STSRow *pRow;
+} STsdbRow;
+
 #define HASH_BUCKET(SUID, UID, NBUCKET) (TABS((SUID) + (UID)) % (NBUCKET))
 
 #define SL_NODE_SIZE(l)        (sizeof(SMemSkipListNode) + sizeof(SMemSkipListNode *) * (l)*2)
@@ -76,7 +85,19 @@ struct SMemSkipListCurosr {
 #define SL_HEAD_NODE_FORWARD(n, l)  SL_NODE_FORWARD(n, l)
 #define SL_TAIL_NODE_BACKWARD(n, l) SL_NODE_FORWARD(n, l)
 
-static int8_t tsdbMemSkipListRandLevel(SMemSkipList *pSl);
+static int8_t  tsdbMemSkipListRandLevel(SMemSkipList *pSl);
+static int32_t tsdbEncodeRow(SEncoder *pEncoder, const STsdbRow *pRow);
+static int32_t tsdbDecodeRow(SDecoder *pDecoder, STsdbRow *pRow);
+static int32_t tsdbMemSkipListCursorCreate(int8_t maxLevel, SMemSkipListCurosr **ppSlc);
+static void    tsdbMemSkipListCursorDestroy(SMemSkipListCurosr *pSlc);
+static void    tsdbMemSkipListCursorInit(SMemSkipListCurosr *pSlc, SMemSkipList *pSl);
+static void    tsdbMemSkipListCursorPut(SMemSkipListCurosr *pSlc, SMemSkipListNode *pNode);
+static int32_t tsdbMemSkipListCursorMoveTo(SMemSkipListCurosr *pSlc, int64_t version, TSKEY ts, int32_t flags);
+static void    tsdbMemSkipListCursorMoveToFirst(SMemSkipListCurosr *pSlc);
+static void    tsdbMemSkipListCursorMoveToLast(SMemSkipListCurosr *pSlc);
+static int32_t tsdbMemSkipListCursorMoveToNext(SMemSkipListCurosr *pSlc);
+static int32_t tsdbMemSkipListCursorMoveToPrev(SMemSkipListCurosr *pSlc);
+static SMemSkipListNode *tsdbMemSkipListNodeCreate(SVBufPool *pPool, SMemSkipList *pSl, const STsdbRow *pTRow);
 
 // SMemTable
 int32_t tsdbMemTableCreate2(STsdb *pTsdb, SMemTable **ppMemTb) {
@@ -102,6 +123,11 @@ int32_t tsdbMemTableCreate2(STsdb *pTsdb, SMemTable **ppMemTb) {
     taosMemoryFree(pMemTb);
     return -1;
   }
+  if (tsdbMemSkipListCursorCreate(pTsdb->pVnode->config.tsdbCfg.slLevel, &pMemTb->pSlc) < 0) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    taosMemoryFree(pMemTb->pBuckets);
+    taosMemoryFree(pMemTb);
+  }
 
   *ppMemTb = pMemTb;
   return 0;
@@ -110,6 +136,7 @@ int32_t tsdbMemTableCreate2(STsdb *pTsdb, SMemTable **ppMemTb) {
 int32_t tsdbMemTableDestroy2(STsdb *pTsdb, SMemTable *pMemTb) {
   if (pMemTb) {
     // loop to destroy the contents (todo)
+    tsdbMemSkipListCursorDestroy(pMemTb->pSlc);
     taosMemoryFree(pMemTb->pBuckets);
     taosMemoryFree(pMemTb);
   }
@@ -177,52 +204,42 @@ int32_t tsdbInsertData2(SMemTable *pMemTb, int64_t version, const SVSubmitBlk *p
   }
 
   // do insert data to SMemData
-  SMemSkipListCurosr slc = {0};
-  const STSRow      *pRow;
-  uint32_t           szRow;
-  SDecoder           decoder = {0};
-
-  tDecoderInit(&decoder, pSubmitBlk->pData, pSubmitBlk->nData);
-  for (;;) {
-    if (tDecodeIsEnd(&decoder)) break;
-
-    if (tDecodeBinary(&decoder, (const uint8_t **)&pRow, &szRow) < 0) {
+  SMemSkipListNode *forwards[SL_MAX_LEVEL];
+  SMemSkipListNode *pNode;
+  int32_t           iRow;
+  STsdbRow          tRow = {.version = version};
+  SEncoder          ec = {0};
+  SDecoder          dc = {0};
+
+  tDecoderInit(&dc, pSubmitBlk->pData, pSubmitBlk->nData);
+  tsdbMemSkipListCursorInit(pMemTb->pSlc, &pMemData->sl);
+  for (iRow = 0;; iRow++) {
+    if (tDecodeIsEnd(&dc)) break;
+
+    // decode row
+    if (tDecodeBinary(&dc, (const uint8_t **)&tRow.pRow, &tRow.szRow) < 0) {
       terrno = TSDB_CODE_INVALID_MSG;
       return -1;
     }
 
-    // check the row (todo)
-
-    //   // move the cursor to position to write (todo)
-    //   int32_t c;
-    //   tsdbMemSkipListCursorMoveTo(&slc, pTSRow, version, &c);
-    //   ASSERT(c);
+    // move cursor
+    tsdbMemSkipListCursorMoveTo(pMemTb->pSlc, version, tRow.pRow->ts, 0);
 
     // encode row
-    int8_t            level = tsdbMemSkipListRandLevel(&pMemData->sl);
-    int32_t           tsize = SL_NODE_SIZE(level) + sizeof(version) + (0 /*todo*/);
-    SMemSkipListNode *pNode = vnodeBufPoolMalloc(pPool, tsize);
+    pNode = tsdbMemSkipListNodeCreate(pPool, &pMemData->sl, &tRow);
     if (pNode == NULL) {
       terrno = TSDB_CODE_OUT_OF_MEMORY;
       return -1;
     }
 
-    pNode->level = level;
-
-    //   uint8_t *pData = SL_NODE_DATA(pSlNode);
-    //   *(int64_t *)pData = version;
-    //   pData += sizeof(version);
-    //   memcpy(pData, pt, p - pt);
-
-    //   // insert row
-    //   tsdbMemSkipListCursorPut(&slc, pSlNode);
+    // put the node
+    tsdbMemSkipListCursorPut(pMemTb->pSlc, pNode);
 
     // update status
-    if (pRow->ts < pMemData->minKey) pMemData->minKey = pRow->ts;
-    if (pRow->ts > pMemData->maxKey) pMemData->maxKey = pRow->ts;
+    if (tRow.pRow->ts < pMemData->minKey) pMemData->minKey = tRow.pRow->ts;
+    if (tRow.pRow->ts > pMemData->maxKey) pMemData->maxKey = tRow.pRow->ts;
   }
-  tDecoderClear(&decoder);
-  // tsdbMemSkipListCursorClose(&slc);
+  tDecoderClear(&dc);
 
   // update status
   if (pMemData->minVer == -1) pMemData->minVer = version;
@@ -236,17 +253,128 @@ int32_t tsdbInsertData2(SMemTable *pMemTb, int64_t version, const SVSubmitBlk *p
   return 0;
 }
 
-static int8_t tsdbMemSkipListRandLevel(SMemSkipList *pSl) {
+static FORCE_INLINE int8_t tsdbMemSkipListRandLevel(SMemSkipList *pSl) {
   int8_t         level = 1;
-  int8_t         tlevel;
+  int8_t         tlevel = TMIN(pSl->maxLevel, pSl->level + 1);
   const uint32_t factor = 4;
 
-  if (pSl->size) {
-    tlevel = TMIN(pSl->maxLevel, pSl->level + 1);
-    while ((taosRandR(&pSl->seed) % factor) == 0 && level < tlevel) {
-      level++;
-    }
+  while ((taosRandR(&pSl->seed) % factor) == 0 && level < tlevel) {
+    level++;
   }
 
   return level;
+}
+
+static FORCE_INLINE int32_t tsdbEncodeRow(SEncoder *pEncoder, const STsdbRow *pRow) {
+  if (tEncodeI64(pEncoder, pRow->version) < 0) return -1;
+  if (tEncodeBinary(pEncoder, (const uint8_t *)pRow->pRow, pRow->szRow) < 0) return -1;
+  return 0;
+}
+
+static FORCE_INLINE int32_t tsdbDecodeRow(SDecoder *pDecoder, STsdbRow *pRow) {
+  if (tDecodeI64(pDecoder, &pRow->version) < 0) return -1;
+  if (tDecodeBinary(pDecoder, (const uint8_t **)&pRow->pRow, &pRow->szRow) < 0) return -1;
+  return 0;
+}
+
+static int32_t tsdbMemSkipListCursorCreate(int8_t maxLevel, SMemSkipListCurosr **ppSlc) {
+  *ppSlc = (SMemSkipListCurosr *)taosMemoryCalloc(1, sizeof(**ppSlc) + sizeof(SMemSkipListNode *) * maxLevel);
+  if (*ppSlc == NULL) {
+    return -1;
+  }
+  return 0;
+}
+
+static void tsdbMemSkipListCursorDestroy(SMemSkipListCurosr *pSlc) { taosMemoryFree(pSlc); }
+
+static void tsdbMemSkipListCursorInit(SMemSkipListCurosr *pSlc, SMemSkipList *pSl) {
+  SMemSkipListNode *pHead = SL_HEAD_NODE(pSl);
+  pSlc->pSl = pSl;
+  // for (int8_t iLevel = 0; iLevel < pSl->maxLevel; iLevel++) {
+  //   pSlc->forwards[iLevel] = pHead;
+  // }
+}
+
+static void tsdbMemSkipListCursorPut(SMemSkipListCurosr *pSlc, SMemSkipListNode *pNode) {
+  SMemSkipList     *pSl = pSlc->pSl;
+  SMemSkipListNode *pNodeNext;
+
+  for (int8_t iLevel = 0; iLevel < pNode->level; iLevel++) {
+    // todo
+
+    ASSERT(0);
+  }
+
+  if (pSl->level < pNode->level) {
+    pSl->level = pNode->level;
+  }
+
+  pSl->size += 1;
+}
+
+static int32_t tsdbMemSkipListCursorMoveTo(SMemSkipListCurosr *pSlc, int64_t version, TSKEY ts, int32_t flags) {
+  SMemSkipListNode **pForwards = NULL;
+  SMemSkipList      *pSl = pSlc->pSl;
+  int8_t             maxLevel = pSl->maxLevel;
+  SMemSkipListNode  *pHead = SL_HEAD_NODE(pSl);
+  SMemSkipListNode  *pTail = SL_TAIL_NODE(pSl);
+
+  if (pSl->size == 0) {
+    for (int8_t iLevel = 0; iLevel < pSl->maxLevel; iLevel++) {
+      pForwards[iLevel] = pHead;
+    }
+  }
+
+  return 0;
+}
+
+static void tsdbMemSkipListCursorMoveToFirst(SMemSkipListCurosr *pSlc) {
+  SMemSkipList     *pSl = pSlc->pSl;
+  SMemSkipListNode *pHead = SL_HEAD_NODE(pSl);
+
+  for (int8_t iLevel = 0; iLevel < pSl->maxLevel; iLevel++) {
+    pSlc->pNodes[iLevel] = pHead;
+  }
+
+  tsdbMemSkipListCursorMoveToNext(pSlc);
+}
+
+static void tsdbMemSkipListCursorMoveToLast(SMemSkipListCurosr *pSlc) {
+  SMemSkipList     *pSl = pSlc->pSl;
+  SMemSkipListNode *pTail = SL_TAIL_NODE(pSl);
+
+  for (int8_t iLevel = 0; iLevel < pSl->maxLevel; iLevel++) {
+    pSlc->pNodes[iLevel] = pTail;
+  }
+
+  tsdbMemSkipListCursorMoveToPrev(pSlc);
+}
+
+static int32_t tsdbMemSkipListCursorMoveToNext(SMemSkipListCurosr *pSlc) {
+  // TODO
+  return 0;
+}
+
+static int32_t tsdbMemSkipListCursorMoveToPrev(SMemSkipListCurosr *pSlc) {
+  // TODO
+  return 0;
+}
+
+static SMemSkipListNode *tsdbMemSkipListNodeCreate(SVBufPool *pPool, SMemSkipList *pSl, const STsdbRow *pTRow) {
+  int32_t           tsize;
+  int32_t           ret;
+  int8_t            level = tsdbMemSkipListRandLevel(pSl);
+  SMemSkipListNode *pNode = NULL;
+  SEncoder          ec = {0};
+
+  tEncodeSize(tsdbEncodeRow, pTRow, tsize, ret);
+  pNode = vnodeBufPoolMalloc(pPool, tsize + SL_NODE_SIZE(level));
+  if (pNode) {
+    pNode->level = level;
+    tEncoderInit(&ec, (uint8_t *)SL_NODE_DATA(pNode), tsize);
+    tsdbEncodeRow(&ec, pTRow);
+    tEncoderClear(&ec);
+  }
+
+  return pNode;
 }
\ No newline at end of file

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

@@ -351,35 +351,26 @@ static void setQueryTimewindow(STsdbReadHandle* pTsdbReadHandle, SQueryTableData
               pTsdbReadHandle->window.ekey, pTsdbReadHandle->idStr);
   }
 }
-#if 1
-int nQUERY = 0;
-#endif
+
 static STsdb* getTsdbByRetentions(SVnode* pVnode, STsdbReadHandle* pReadHandle, TSKEY winSKey, SRetention* retentions) {
   if (vnodeIsRollup(pVnode)) {
-    int level = 0;
-#if 0
+    int     level = 0;
     int64_t now = taosGetTimestamp(pVnode->config.tsdbCfg.precision);
+
     for (int i = 0; i < TSDB_RETENTION_MAX; ++i) {
-      SRetention* pRetention = retentions + i;
-      if (pRetention->keep <= 0 || (now - pRetention->keep) >= winSKey) {
+      SRetention* pRetention = retentions + level;
+      if (pRetention->keep <= 0) {
+        if (level > 0) {
+          --level;
+        }
         break;
       }
-      ++level;
-    }
-#endif
-#if 1
-    switch ((nQUERY++) % 3) {
-      case 0:
-        level = 0;
-        break;
-      case 1:
-        level = 1;
-        break;
-      default:
-        level = 2;
+      if ((now - pRetention->keep) <= winSKey) {
         break;
+      }
+      ++level;
     }
-#endif
+
     if (level == TSDB_RETENTION_L0) {
       tsdbDebug("%p rsma level %d is selected to query\n", pReadHandle, level);
       return VND_RSMA0(pVnode);
@@ -391,7 +382,7 @@ static STsdb* getTsdbByRetentions(SVnode* pVnode, STsdbReadHandle* pReadHandle,
       return VND_RSMA2(pVnode);
     }
   }
-  return pVnode->pTsdb;
+  return VND_TSDB(pVnode);
 }
 
 static STsdbReadHandle* tsdbQueryTablesImpl(SVnode* pVnode, SQueryTableDataCond* pCond, uint64_t qId, uint64_t taskId) {
@@ -3879,7 +3870,7 @@ int32_t tsdbQuerySTableByTagCond(void* pMeta, uint64_t uid, TSKEY skey, const ch
 
   if (metaGetTableEntryByUid(&mr, uid) < 0) {
     tsdbError("%p failed to get stable, uid:%" PRIu64 ", TID:0x%" PRIx64 " QID:0x%" PRIx64, pMeta, uid, taskId, reqId);
-    terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;
+    terrno = TSDB_CODE_PAR_TABLE_NOT_EXIST;
     goto _error;
   } else {
     tsdbDebug("%p succeed to get stable, uid:%" PRIu64 ", TID:0x%" PRIx64 " QID:0x%" PRIx64, pMeta, uid, taskId, reqId);
@@ -3949,7 +3940,7 @@ int32_t tsdbGetOneTableGroup(void* pMeta, uint64_t uid, TSKEY startKey, STableGr
   metaReaderInit(&mr, (SMeta*)pMeta, 0);
 
   if (metaGetTableEntryByUid(&mr, uid) < 0) {
-    terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;
+    terrno = TSDB_CODE_PAR_TABLE_NOT_EXIST;
     goto _error;
   }
 

source/dnode/vnode/src/vnd/vnodeSvr.c

@@ -497,6 +497,47 @@ _exit:
   return 0;
 }
 
+static int vnodeDebugPrintSubmitMsg(SVnode *pVnode, SSubmitReq *pMsg, const char* tags) {
+  ASSERT(pMsg != NULL);
+  SSubmitMsgIter msgIter = {0};
+  SMeta         *pMeta = pVnode->pMeta;
+  SSubmitBlk    *pBlock = NULL;
+  SSubmitBlkIter blkIter = {0};
+  STSRow        *row = NULL;
+  STSchema      *pSchema = NULL;
+  tb_uid_t       suid = 0;
+
+  if (tInitSubmitMsgIter(pMsg, &msgIter) < 0) return -1;
+  while (true) {
+    if (tGetSubmitMsgNext(&msgIter, &pBlock) < 0) return -1;
+    if (pBlock == NULL) break;
+    tInitSubmitBlkIter(&msgIter, pBlock, &blkIter);
+    if (blkIter.row == NULL) continue;
+    if (!pSchema || (suid != msgIter.suid)) {
+      if (pSchema) {
+        taosMemoryFreeClear(pSchema);
+      }
+      pSchema = metaGetTbTSchema(pMeta, msgIter.suid, 0);  // TODO: use the real schema
+      if(pSchema) {
+        suid = msgIter.suid;
+      }
+    }
+    if(!pSchema) {
+      printf("%s:%d no valid schema\n", tags, __LINE__);
+      continue;
+    }
+    char __tags[128] = {0};
+    snprintf(__tags, 128, "%s: uid %" PRIi64 " ", tags, msgIter.uid);
+    while ((row = tGetSubmitBlkNext(&blkIter))) {
+      tdSRowPrint(row, pSchema, __tags);
+    }
+  }
+
+  taosMemoryFreeClear(pSchema);
+
+  return 0;
+}
+
 static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
   SSubmitReq    *pSubmitReq = (SSubmitReq *)pReq;
   SSubmitMsgIter msgIter = {0};
@@ -508,6 +549,10 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
 
   pRsp->code = 0;
 
+#ifdef TD_DEBUG_PRINT_ROW
+  vnodeDebugPrintSubmitMsg(pVnode, pReq, __func__);
+#endif
+
   // handle the request
   if (tInitSubmitMsgIter(pSubmitReq, &msgIter) < 0) {
     pRsp->code = TSDB_CODE_INVALID_MSG;
@@ -550,7 +595,8 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
       goto _exit;
     }
 
-    rsp.numOfRows += nRows;
+    rsp.affectedRows += nRows;
+    
   }
 
 _exit:

source/libs/catalog/inc/catalogInt.h

@@ -36,7 +36,7 @@ extern "C" {
 
 #define CTG_DEFAULT_INVALID_VERSION (-1)
 
-#define CTG_ERR_CODE_TABLE_NOT_EXIST TSDB_CODE_TDB_INVALID_TABLE_ID
+#define CTG_ERR_CODE_TABLE_NOT_EXIST TSDB_CODE_PAR_TABLE_NOT_EXIST
 
 enum {
   CTG_READ = 1,

source/libs/executor/src/executorimpl.c

@@ -4199,7 +4199,7 @@ static void destroyOperatorInfo(SOperatorInfo* pOperator) {
     }
   }
 
-  taosMemoryFree(pOperator->pExpr);
+  taosMemoryFreeClear(pOperator->pExpr);
   taosMemoryFreeClear(pOperator->info);
   taosMemoryFreeClear(pOperator);
 }
@@ -4384,6 +4384,9 @@ void destroySFillOperatorInfo(void* param, int32_t numOfOutput) {
 }
 
 static void destroyProjectOperatorInfo(void* param, int32_t numOfOutput) {
+  if (NULL == param) {
+    return;
+  }
   SProjectOperatorInfo* pInfo = (SProjectOperatorInfo*)param;
   doDestroyBasicInfo(&pInfo->binfo, numOfOutput);
   cleanupAggSup(&pInfo->aggSup);

source/libs/function/inc/builtinsimpl.h

@@ -55,6 +55,12 @@ int32_t stddevFunction(SqlFunctionCtx* pCtx);
 int32_t stddevFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
 int32_t stddevInvertFunction(SqlFunctionCtx* pCtx);
 
+bool getLeastSQRFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
+bool leastSQRFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
+int32_t leastSQRFunction(SqlFunctionCtx* pCtx);
+int32_t leastSQRFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
+int32_t leastSQRInvertFunction(SqlFunctionCtx* pCtx);
+
 bool getPercentileFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 bool percentileFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
 int32_t percentileFunction(SqlFunctionCtx *pCtx);

source/libs/function/src/builtins.c

@@ -226,6 +226,23 @@ static int32_t translateSpread(SFunctionNode* pFunc, char* pErrBuf, int32_t len)
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t translateLeastSQR(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
+  int32_t numOfParams = LIST_LENGTH(pFunc->pParameterList);
+  if (3 != numOfParams) {
+    return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName);
+  }
+
+  for (int32_t i = 0; i < numOfParams; ++i) {
+    uint8_t colType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, i))->resType.type;
+    if (!IS_NUMERIC_TYPE(colType)) {
+      return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
+    }
+  }
+
+  pFunc->node.resType = (SDataType) { .bytes = 64, .type = TSDB_DATA_TYPE_BINARY };
+  return TSDB_CODE_SUCCESS;
+}
+
 static int32_t translateHistogram(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
   if (4 != LIST_LENGTH(pFunc->pParameterList)) {
     return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName);
@@ -537,6 +554,17 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
     .finalizeFunc = stddevFinalize,
     .invertFunc   = stddevInvertFunction
   },
+  {
+    .name = "leastsquares",
+    .type = FUNCTION_TYPE_LEASTSQUARES,
+    .classification = FUNC_MGT_AGG_FUNC,
+    .translateFunc = translateLeastSQR,
+    .getEnvFunc   = getLeastSQRFuncEnv,
+    .initFunc     = leastSQRFunctionSetup,
+    .processFunc  = leastSQRFunction,
+    .finalizeFunc = leastSQRFinalize,
+    .invertFunc   = leastSQRInvertFunction
+  },
   {
     .name = "avg",
     .type = FUNCTION_TYPE_AVG,

source/libs/function/src/builtinsimpl.c

@@ -65,6 +65,13 @@ typedef struct SStddevRes {
   };
 } SStddevRes;
 
+typedef struct SLeastSQRInfo {
+  double matrix[2][3];
+  double startVal;
+  double stepVal;
+  int64_t num;
+} SLeastSQRInfo;
+
 typedef struct SPercentileInfo {
   double      result;
   tMemBucket* pMemBucket;
@@ -105,6 +112,7 @@ typedef struct SHistoFuncBin {
 
 typedef struct SHistoFuncInfo {
   int32_t numOfBins;
+  int32_t totalCount;
   bool    normalized;
   SHistoFuncBin bins[];
 } SHistoFuncInfo;
@@ -1404,6 +1412,181 @@ int32_t stddevFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   return functionFinalize(pCtx, pBlock);
 }
 
+bool getLeastSQRFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
+  pEnv->calcMemSize = sizeof(SLeastSQRInfo);
+  return true;
+}
+
+bool leastSQRFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo) {
+  if (!functionSetup(pCtx, pResultInfo)) {
+    return false;
+  }
+
+  SLeastSQRInfo* pInfo = GET_ROWCELL_INTERBUF(pResultInfo);
+
+  pInfo->startVal = IS_FLOAT_TYPE(pCtx->param[1].param.nType) ? pCtx->param[1].param.d :
+                                                                (double)pCtx->param[1].param.i;
+  pInfo->stepVal = IS_FLOAT_TYPE(pCtx->param[1].param.nType) ? pCtx->param[2].param.d :
+                                                                (double)pCtx->param[1].param.i;
+  return true;
+}
+
+#define LEASTSQR_CAL(p, x, y, index, step) \
+  do {                                     \
+    (p)[0][0] += (double)(x) * (x);        \
+    (p)[0][1] += (double)(x);              \
+    (p)[0][2] += (double)(x) * (y)[index]; \
+    (p)[1][2] += (y)[index];               \
+    (x) += step;                           \
+  } while (0)
+
+int32_t leastSQRFunction(SqlFunctionCtx* pCtx) {
+  int32_t numOfElem = 0;
+
+  SInputColumnInfoData* pInput = &pCtx->input;
+  int32_t               type = pInput->pData[0]->info.type;
+
+  SLeastSQRInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
+
+  SColumnInfoData* pCol = pInput->pData[0];
+
+  double(*param)[3] = pInfo->matrix;
+  double x = pInfo->startVal;
+
+  int32_t start = pInput->startRowIndex;
+  int32_t numOfRows = pInput->numOfRows;
+
+  switch (type) {
+    case TSDB_DATA_TYPE_TINYINT: {
+      int8_t* plist = (int8_t*)pCol->pData;
+      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
+        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
+          continue;
+        }
+        numOfElem++;
+        LEASTSQR_CAL(param, x, plist, i, pInfo->stepVal);
+
+        break;
+      }
+    }
+    case TSDB_DATA_TYPE_SMALLINT: {
+      int16_t* plist = (int16_t*)pCol->pData;
+      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
+        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
+          continue;
+        }
+
+        numOfElem++;
+        LEASTSQR_CAL(param, x, plist, i, pInfo->stepVal);
+      }
+      break;
+    }
+
+    case TSDB_DATA_TYPE_INT: {
+      int32_t* plist = (int32_t*)pCol->pData;
+      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
+        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
+          continue;
+        }
+
+        numOfElem++;
+        LEASTSQR_CAL(param, x, plist, i, pInfo->stepVal);
+      }
+
+      break;
+    }
+
+    case TSDB_DATA_TYPE_BIGINT: {
+      int64_t* plist = (int64_t*)pCol->pData;
+      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
+        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
+          continue;
+        }
+
+        numOfElem++;
+        LEASTSQR_CAL(param, x, plist, i, pInfo->stepVal);
+      }
+      break;
+    }
+
+    case TSDB_DATA_TYPE_FLOAT: {
+      float* plist = (float*)pCol->pData;
+      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
+        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
+          continue;
+        }
+
+        numOfElem++;
+        LEASTSQR_CAL(param, x, plist, i, pInfo->stepVal);
+      }
+      break;
+    }
+
+    case TSDB_DATA_TYPE_DOUBLE: {
+      double* plist = (double*)pCol->pData;
+      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
+        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
+          continue;
+        }
+
+        numOfElem++;
+        LEASTSQR_CAL(param, x, plist, i, pInfo->stepVal);
+      }
+      break;
+    }
+
+    default:
+      break;
+  }
+
+  pInfo->startVal = x;
+  pInfo->num += numOfElem;
+
+  SET_VAL(GET_RES_INFO(pCtx), numOfElem, 1);
+
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t leastSQRFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
+  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
+  SLeastSQRInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
+  int32_t        slotId = pCtx->pExpr->base.resSchema.slotId;
+  SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);
+
+  int32_t currentRow = pBlock->info.rows;
+
+  if (0 == pInfo->num) {
+    return 0;
+  }
+
+  double(*param)[3] = pInfo->matrix;
+
+  param[1][1] = (double)pInfo->num;
+  param[1][0] = param[0][1];
+
+  param[0][0] -= param[1][0] * (param[0][1] / param[1][1]);
+  param[0][2] -= param[1][2] * (param[0][1] / param[1][1]);
+  param[0][1] = 0;
+  param[1][2] -= param[0][2] * (param[1][0] / param[0][0]);
+  param[1][0] = 0;
+  param[0][2] /= param[0][0];
+
+  param[1][2] /= param[1][1];
+
+  char buf[64] = {0};
+  size_t len = snprintf(varDataVal(buf), sizeof(buf) - VARSTR_HEADER_SIZE, "{slop:%.6lf, intercept:%.6lf}", param[0][2], param[1][2]);
+  varDataSetLen(buf, len);
+
+  colDataAppend(pCol, currentRow, buf, false);
+
+  return pResInfo->numOfRes;
+}
+
+int32_t leastSQRInvertFunction(SqlFunctionCtx* pCtx) {
+  //TODO
+  return TSDB_CODE_SUCCESS;
+}
+
 bool getPercentileFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
   pEnv->calcMemSize = sizeof(SPercentileInfo);
   return true;
@@ -2398,6 +2581,9 @@ bool histogramFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo *pResultIn
   }
 
   SHistoFuncInfo *pInfo = GET_ROWCELL_INTERBUF(pResultInfo);
+  pInfo->numOfBins = 0;
+  pInfo->totalCount = 0;
+  pInfo->normalized = 0;
 
   int8_t binType = getHistogramBinType(varDataVal(pCtx->param[1].param.pz));
   if (binType == UNKNOWN_BIN) {
@@ -2427,7 +2613,6 @@ int32_t histogramFunction(SqlFunctionCtx *pCtx) {
   int32_t numOfRows = pInput->numOfRows;
 
   int32_t numOfElems = 0;
-  int32_t totalElems = 0;
   for (int32_t i = start; i < numOfRows + start; ++i) {
     if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
       continue;
@@ -2442,23 +2627,13 @@ int32_t histogramFunction(SqlFunctionCtx *pCtx) {
     for (int32_t k = 0; k < pInfo->numOfBins; ++k) {
       if (v > pInfo->bins[k].lower && v <= pInfo->bins[k].upper) {
         pInfo->bins[k].count++;
-        totalElems++;
+        pInfo->totalCount++;
         break;
       }
     }
 
   }
 
-  if (pInfo->normalized) {
-    for (int32_t k = 0; k < pInfo->numOfBins; ++k) {
-      if(totalElems != 0) {
-        pInfo->bins[k].percentage = pInfo->bins[k].count / (double)totalElems;
-      } else {
-        pInfo->bins[k].percentage = 0;
-      }
-    }
-  }
-
   SET_VAL(GET_RES_INFO(pCtx), numOfElems, pInfo->numOfBins);
   return TSDB_CODE_SUCCESS;
 }
@@ -2471,14 +2646,24 @@ int32_t histogramFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
 
   int32_t currentRow = pBlock->info.rows;
 
+  if (pInfo->normalized) {
+    for (int32_t k = 0; k < pResInfo->numOfRes; ++k) {
+      if(pInfo->totalCount != 0) {
+        pInfo->bins[k].percentage = pInfo->bins[k].count / (double)pInfo->totalCount;
+      } else {
+        pInfo->bins[k].percentage = 0;
+      }
+    }
+  }
+
   for (int32_t i = 0; i < pResInfo->numOfRes; ++i) {
     int32_t len;
     char buf[512] = {0};
     if (!pInfo->normalized) {
-      len = sprintf(buf + VARSTR_HEADER_SIZE, "{\"lower_bin\":%g, \"upper_bin\":%g, \"count\":%"PRId64"}",
+      len = sprintf(varDataVal(buf), "{\"lower_bin\":%g, \"upper_bin\":%g, \"count\":%"PRId64"}",
                    pInfo->bins[i].lower, pInfo->bins[i].upper, pInfo->bins[i].count);
     } else {
-      len = sprintf(buf + VARSTR_HEADER_SIZE, "{\"lower_bin\":%g, \"upper_bin\":%g, \"count\":%lf}",
+      len = sprintf(varDataVal(buf), "{\"lower_bin\":%g, \"upper_bin\":%g, \"count\":%lf}",
                    pInfo->bins[i].lower, pInfo->bins[i].upper, pInfo->bins[i].percentage);
     }
     varDataSetLen(buf, len);

source/libs/function/src/udfd.c

@@ -409,7 +409,7 @@ void udfdPipeCloseCb(uv_handle_t *pipe) {
 void udfdUvHandleError(SUdfdUvConn *conn) { uv_close((uv_handle_t *)conn->client, udfdPipeCloseCb); }
 
 void udfdPipeRead(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf) {
-  fnDebug("udf read %zu bytes from client", nread);
+  fnDebug("udf read %zd bytes from client", nread);
   if (nread == 0) return;
 
   SUdfdUvConn *conn = client->data;

source/libs/index/inc/indexComm.h

@@ -20,11 +20,23 @@
 extern "C" {
 #endif
 
+#include "indexInt.h"
+#include "tcompare.h"
+
 extern char JSON_COLUMN[];
 extern char JSON_VALUE_DELIM;
 
 char* indexPackJsonData(SIndexTerm* itm);
 char* indexPackJsonDataPrefix(SIndexTerm* itm, int32_t* skip);
+
+typedef enum { MATCH, CONTINUE, BREAK } TExeCond;
+
+typedef TExeCond (*_cache_range_compare)(void* a, void* b, int8_t type);
+
+TExeCond tDoCommpare(__compar_fn_t func, int8_t comType, void* a, void* b);
+
+_cache_range_compare indexGetCompare(RangeType ty);
+
 #ifdef __cplusplus
 }
 #endif

source/libs/index/src/indexCache.c

@@ -60,50 +60,6 @@ static int32_t cacheSearchRange_JSON(void* cache, SIndexTerm* ct, SIdxTempResult
 static int32_t cacheSearchCompareFunc_JSON(void* cache, SIndexTerm* term, SIdxTempResult* tr, STermValueType* s,
                                            RangeType type);
 
-typedef enum { MATCH, CONTINUE, BREAK } TExeCond;
-typedef TExeCond (*_cache_range_compare)(void* a, void* b, int8_t type);
-
-static TExeCond tDoCommpare(__compar_fn_t func, int8_t comType, void* a, void* b) {
-  // optime later
-  int32_t ret = func(a, b);
-  switch (comType) {
-    case QUERY_LESS_THAN: {
-      if (ret < 0) return MATCH;
-    } break;
-    case QUERY_LESS_EQUAL: {
-      if (ret <= 0) return MATCH;
-      break;
-    }
-    case QUERY_GREATER_THAN: {
-      if (ret > 0) return MATCH;
-      break;
-    }
-    case QUERY_GREATER_EQUAL: {
-      if (ret >= 0) return MATCH;
-    }
-  }
-  return CONTINUE;
-}
-static TExeCond tCompareLessThan(void* a, void* b, int8_t type) {
-  __compar_fn_t func = getComparFunc(type, 0);
-  return tDoCommpare(func, QUERY_LESS_THAN, a, b);
-}
-static TExeCond tCompareLessEqual(void* a, void* b, int8_t type) {
-  __compar_fn_t func = getComparFunc(type, 0);
-  return tDoCommpare(func, QUERY_LESS_EQUAL, a, b);
-}
-static TExeCond tCompareGreaterThan(void* a, void* b, int8_t type) {
-  __compar_fn_t func = getComparFunc(type, 0);
-  return tDoCommpare(func, QUERY_GREATER_THAN, a, b);
-}
-static TExeCond tCompareGreaterEqual(void* a, void* b, int8_t type) {
-  __compar_fn_t func = getComparFunc(type, 0);
-  return tDoCommpare(func, QUERY_GREATER_EQUAL, a, b);
-}
-
-static TExeCond (*rangeCompare[])(void* a, void* b, int8_t type) = {tCompareLessThan, tCompareLessEqual,
-                                                                    tCompareGreaterThan, tCompareGreaterEqual};
-
 static int32_t (*cacheSearch[][QUERY_MAX])(void* cache, SIndexTerm* ct, SIdxTempResult* tr, STermValueType* s) = {
     {cacheSearchTerm, cacheSearchPrefix, cacheSearchSuffix, cacheSearchRegex, cacheSearchLessThan, cacheSearchLessEqual,
      cacheSearchGreaterThan, cacheSearchGreaterEqual, cacheSearchRange},
@@ -169,7 +125,7 @@ static int32_t cacheSearchCompareFunc(void* cache, SIndexTerm* term, SIdxTempRes
     return 0;
   }
 
-  _cache_range_compare cmpFn = rangeCompare[type];
+  _cache_range_compare cmpFn = indexGetCompare(type);
 
   MemTable*   mem = cache;
   IndexCache* pCache = mem->pCache;
@@ -295,7 +251,7 @@ static int32_t cacheSearchCompareFunc_JSON(void* cache, SIndexTerm* term, SIdxTe
   if (cache == NULL) {
     return 0;
   }
-  _cache_range_compare cmpFn = rangeCompare[type];
+  _cache_range_compare cmpFn = indexGetCompare(type);
 
   MemTable*   mem = cache;
   IndexCache* pCache = mem->pCache;

source/libs/index/src/indexComm.c

@@ -13,12 +13,58 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#include "indexComm.h"
 #include "index.h"
 #include "indexInt.h"
+#include "tcompare.h"
 
 char JSON_COLUMN[] = "JSON";
 char JSON_VALUE_DELIM = '&';
 
+static TExeCond tCompareLessThan(void* a, void* b, int8_t type) {
+  __compar_fn_t func = getComparFunc(type, 0);
+  return tDoCommpare(func, QUERY_LESS_THAN, a, b);
+}
+static TExeCond tCompareLessEqual(void* a, void* b, int8_t type) {
+  __compar_fn_t func = getComparFunc(type, 0);
+  return tDoCommpare(func, QUERY_LESS_EQUAL, a, b);
+}
+static TExeCond tCompareGreaterThan(void* a, void* b, int8_t type) {
+  __compar_fn_t func = getComparFunc(type, 0);
+  return tDoCommpare(func, QUERY_GREATER_THAN, a, b);
+}
+static TExeCond tCompareGreaterEqual(void* a, void* b, int8_t type) {
+  __compar_fn_t func = getComparFunc(type, 0);
+  return tDoCommpare(func, QUERY_GREATER_EQUAL, a, b);
+}
+
+TExeCond tDoCommpare(__compar_fn_t func, int8_t comType, void* a, void* b) {
+  // optime later
+  int32_t ret = func(a, b);
+  switch (comType) {
+    case QUERY_LESS_THAN: {
+      if (ret < 0) return MATCH;
+    } break;
+    case QUERY_LESS_EQUAL: {
+      if (ret <= 0) return MATCH;
+      break;
+    }
+    case QUERY_GREATER_THAN: {
+      if (ret > 0) return MATCH;
+      break;
+    }
+    case QUERY_GREATER_EQUAL: {
+      if (ret >= 0) return MATCH;
+    }
+  }
+  return CONTINUE;
+}
+
+static TExeCond (*rangeCompare[])(void* a, void* b, int8_t type) = {tCompareLessThan, tCompareLessEqual,
+                                                                    tCompareGreaterThan, tCompareGreaterEqual};
+
+_cache_range_compare indexGetCompare(RangeType ty) { return rangeCompare[ty]; }
+
 char* indexPackJsonData(SIndexTerm* itm) {
   /*
    * |<-----colname---->|<-----dataType---->|<--------colVal---------->|
@@ -46,6 +92,7 @@ char* indexPackJsonData(SIndexTerm* itm) {
 
   return buf;
 }
+
 char* indexPackJsonDataPrefix(SIndexTerm* itm, int32_t* skip) {
   /*
    * |<-----colname---->|<-----dataType---->|<--------colVal---------->|

source/libs/parser/src/parInsert.c

@@ -241,6 +241,15 @@ static int32_t getTableMetaImpl(SInsertParseContext* pCxt, SToken* pTname, bool
   SParseContext* pBasicCtx = pCxt->pComCxt;
   SName          name = {0};
   createSName(&name, pTname, pBasicCtx->acctId, pBasicCtx->db, &pCxt->msg);
+
+  char dbFname[TSDB_DB_FNAME_LEN] = {0};
+  tNameGetFullDbName(&name, dbFname);
+
+  bool pass = false;
+  CHECK_CODE(catalogChkAuth(pBasicCtx->pCatalog, pBasicCtx->pTransporter, &pBasicCtx->mgmtEpSet, pBasicCtx->pUser, dbFname, AUTH_TYPE_WRITE, &pass));
+  if (!pass) {
+    return TSDB_CODE_PAR_PERMISSION_DENIED;
+  }
   if (isStb) {
     CHECK_CODE(catalogGetSTableMeta(pBasicCtx->pCatalog, pBasicCtx->pTransporter, &pBasicCtx->mgmtEpSet, &name,
                                     &pCxt->pTableMeta));
@@ -1151,6 +1160,7 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) {
     (*pCxt->pStmtCb->setExecInfoFn)(pCxt->pStmtCb->pStmt, pCxt->pVgroupsHashObj, pCxt->pTableBlockHashObj);
     pCxt->pVgroupsHashObj = NULL;
     pCxt->pTableBlockHashObj = NULL;
+    pCxt->pTableMeta = NULL;
 
     return TSDB_CODE_SUCCESS;
   }
@@ -1276,7 +1286,7 @@ int32_t qBindStmtTagsValue(void *pBlock, void *boundTags, int64_t suid, char *tN
     return TSDB_CODE_TSC_OUT_OF_MEMORY;
   }
 
-  SSchema* pSchema = getTableTagSchema(pDataBlock->pTableMeta);
+  SSchema* pSchema = pDataBlock->pTableMeta->schema;
   SKvParam param = {.builder = &tagBuilder};
 
   for (int c = 0; c < tags->numOfBound; ++c) {

source/libs/parser/src/parInsertData.c

@@ -469,7 +469,7 @@ int32_t mergeTableDataBlocks(SHashObj* pHashObj, uint8_t payloadType, SArray** p
       // the maximum expanded size in byte when a row-wise data is converted to SDataRow format
       int32_t expandSize = isRawPayload ? getRowExpandSize(pOneTableBlock->pTableMeta) : 0;
       int64_t destSize = dataBuf->size + pOneTableBlock->size + pBlocks->numOfRows * expandSize +
-                         sizeof(STColumn) * getNumOfColumns(pOneTableBlock->pTableMeta);
+                         sizeof(STColumn) * getNumOfColumns(pOneTableBlock->pTableMeta) + pOneTableBlock->createTbReqLen;
 
       if (dataBuf->nAllocSize < destSize) {
         dataBuf->nAllocSize = (uint32_t)(destSize * 1.5);
@@ -601,6 +601,7 @@ int32_t qResetStmtDataBlock(void* block, bool keepBuf) {
   pBlock->numOfTables = 1;
   pBlock->nAllocSize = TSDB_PAYLOAD_SIZE;
   pBlock->headerSize = pBlock->size;
+  pBlock->createTbReqLen = 0;
 
   memset(&pBlock->rowBuilder, 0, sizeof(pBlock->rowBuilder));
 

source/libs/parser/src/parTranslater.c

@@ -2588,7 +2588,7 @@ static int32_t translateDropTable(STranslateContext* pCxt, SDropTableStmt* pStmt
   SName       tableName;
   int32_t     code = getTableMetaImpl(
           pCxt, toName(pCxt->pParseCxt->acctId, pClause->dbName, pClause->tableName, &tableName), &pTableMeta);
-  if ((TSDB_CODE_TDB_INVALID_TABLE_ID == code || TSDB_CODE_VND_TB_NOT_EXIST == code) && pClause->ignoreNotExists) {
+  if ((TSDB_CODE_PAR_TABLE_NOT_EXIST == code || TSDB_CODE_VND_TB_NOT_EXIST == code) && pClause->ignoreNotExists) {
     return TSDB_CODE_SUCCESS;
   }
   if (TSDB_CODE_SUCCESS == code) {
@@ -3200,7 +3200,7 @@ static int32_t translateGrant(STranslateContext* pCxt, SGrantStmt* pStmt) {
     req.alterType = TSDB_ALTER_USER_ADD_WRITE_DB;
   }
   strcpy(req.user, pStmt->userName);
-  strcpy(req.dbname, pStmt->dbName);
+  sprintf(req.dbname, "%d.%s", pCxt->pParseCxt->acctId, pStmt->dbName);
   return buildCmdMsg(pCxt, TDMT_MND_ALTER_USER, (FSerializeFunc)tSerializeSAlterUserReq, &req);
 }
 

source/libs/scalar/src/filter.c

@@ -3577,6 +3577,22 @@ EDealRes fltReviseRewriter(SNode** pNode, void* pContext) {
   }
   
   if (QUERY_NODE_NODE_LIST == nodeType(*pNode)) {
+    SNodeListNode *listNode = (SNodeListNode *)*pNode;
+    if (QUERY_NODE_VALUE != nodeType(listNode->pNodeList->pHead->pNode)) {
+      stat->scalarMode = true;
+      return DEAL_RES_CONTINUE;
+    }
+
+    SValueNode *valueNode = (SValueNode *)listNode->pNodeList->pHead->pNode;
+    uint8_t type = valueNode->node.resType.type;
+    SNode *node = NULL;
+    FOREACH(node, listNode->pNodeList) {
+      if (type != ((SValueNode *)node)->node.resType.type) {
+        stat->scalarMode = true;
+        return DEAL_RES_CONTINUE;
+      }
+    }
+
     return DEAL_RES_CONTINUE;
   }
 

source/libs/tdb/src/db/tdbBtree.c

@@ -113,7 +113,9 @@ int tdbBtreeOpen(int keyLen, int valLen, SPager *pPager, tdb_cmpr_fn_t kcmpr, SB
 }
 
 int tdbBtreeClose(SBTree *pBt) {
-  // TODO
+  if (pBt) {
+    tdbOsFree(pBt);
+  }
   return 0;
 }
 

source/libs/tdb/src/db/tdbDb.c

@@ -66,7 +66,10 @@ int tdbDbOpen(const char *fname, int keyLen, int valLen, tdb_cmpr_fn_t keyCmprFn
 }
 
 int tdbDbClose(TDB *pDb) {
-  // TODO
+  if (pDb) {
+    tdbBtreeClose(pDb->pBt);
+    tdbOsFree(pDb);
+  }
   return 0;
 }