Merge branch '3.0' into feature/tdb

.gitignore

@@ -24,7 +24,6 @@ mac/
 *.orig
 src/connector/nodejs/node_modules/
 src/connector/nodejs/out/
-tests/test/
 tests/taoshebei/
 tests/taoscsv/
 tests/taosdalipu/
@@ -88,7 +87,9 @@ tests/comparisonTest/opentsdb/opentsdbtest/.settings/
 tests/examples/JDBC/JDBCDemo/.classpath
 tests/examples/JDBC/JDBCDemo/.project
 tests/examples/JDBC/JDBCDemo/.settings/
-source/libs/parser/inc/new_sql.*
+source/libs/parser/inc/sql.*
+tests/script/tmqResult.txt
+tests/tmqResult.txt
 
 # Emacs
 # -*- mode: gitignore; -*-

.gitmodules

@@ -10,10 +10,9 @@
 [submodule "deps/TSZ"]
 	path = deps/TSZ
 	url = https://github.com/taosdata/TSZ.git
-[submodule "tests"]
-	path = tests
-	url = https://github.com/taosdata/tests
-	branch = 3.0
 [submodule "examples/rust"]
 	path = examples/rust
 	url = https://github.com/songtianyi/tdengine-rust-bindings.git
+[submodule "tools/taos-tools"]
+	path = tools/taos-tools
+	url = https://github.com/taosdata/taos-tools

2.0/src/client/inc/tsclient.h

@@ -116,7 +116,7 @@ typedef struct SParsedDataColInfo {
   uint16_t       allNullLen;  // TODO: get from STSchema(base on SDataRow)
   uint16_t       extendedVarLen;
   uint16_t       boundNullLen;    // bound column len with all NULL value(without VarDataOffsetT/SColIdx part)
-  int32_t *      boundedColumns;  // bound column idx according to schema
+  int32_t       *boundColumns;    // bound column idx according to schema
   SBoundColumn * cols;
   SBoundIdxInfo *colIdxInfo;
   int8_t         orderStatus;  // bound columns
@@ -125,7 +125,7 @@ typedef struct SParsedDataColInfo {
 #define IS_DATA_COL_ORDERED(spd) ((spd->orderStatus) == (int8_t)ORDER_STATUS_ORDERED)
 
 typedef struct {
-  uint8_t memRowType;  // default is 0, that is SDataRow
+  uint8_t rowType;  // default is 0, that is SDataRow
   int32_t rowSize;
 } SMemRowBuilder;
 
@@ -137,17 +137,17 @@ void destroyMemRowBuilder(SMemRowBuilder *pBuilder);
 /**
  * @brief
  *
- * @param memRowType
+ * @param rowType
  * @param spd
  * @param idx   the absolute bound index of columns
  * @return FORCE_INLINE
  */
-static FORCE_INLINE void tscGetMemRowAppendInfo(SSchema *pSchema, uint8_t memRowType, SParsedDataColInfo *spd,
-                                                int32_t idx, int32_t *toffset, int16_t *colId) {
+static FORCE_INLINE void tscGetSTSRowAppendInfo(SSchema *pSchema, uint8_t rowType, SParsedDataColInfo *spd, int32_t idx,
+                                                int32_t *toffset, int16_t *colId) {
   int32_t schemaIdx = 0;
   if (IS_DATA_COL_ORDERED(spd)) {
-    schemaIdx = spd->boundedColumns[idx];
-    if (isDataRowT(memRowType)) {
+    schemaIdx = spd->boundColumns[idx];
+    if (isDataRowT(rowType)) {
       *toffset = (spd->cols + schemaIdx)->toffset;  // the offset of firstPart
     } else {
       *toffset = idx * sizeof(SColIdx);  // the offset of SColIdx
@@ -155,7 +155,7 @@ static FORCE_INLINE void tscGetMemRowAppendInfo(SSchema *pSchema, uint8_t memRow
   } else {
     ASSERT(idx == (spd->colIdxInfo + idx)->boundIdx);
     schemaIdx = (spd->colIdxInfo + idx)->schemaColIdx;
-    if (isDataRowT(memRowType)) {
+    if (isDataRowT(rowType)) {
       *toffset = (spd->cols + schemaIdx)->toffset;
     } else {
       *toffset = ((spd->colIdxInfo + idx)->finalIdx) * sizeof(SColIdx);

2.0/src/client/src/tscParseInsert.c

@@ -428,7 +428,7 @@ int tsParseOneRow(char **str, STableDataBlocks *pDataBlocks, int16_t timePrec, i
   // 1. set the parsed value from sql string
   for (int i = 0; i < spd->numOfBound; ++i) {
     // the start position in data block buffer of current value in sql
-    int32_t colIndex = spd->boundedColumns[i];
+    int32_t colIndex = spd->boundColumns[i];
 
     char *start = row + spd->cols[colIndex].offset;
 
@@ -495,7 +495,7 @@ int tsParseOneRow(char **str, STableDataBlocks *pDataBlocks, int16_t timePrec, i
     bool    isPrimaryKey = (colIndex == PRIMARYKEY_TIMESTAMP_COL_INDEX);
     int32_t toffset = -1;
     int16_t colId = -1;
-    tscGetMemRowAppendInfo(schema, pBuilder->memRowType, spd, i, &toffset, &colId);
+    tscGetSTSRowAppendInfo(schema, pBuilder->memRowType, spd, i, &toffset, &colId);
 
     int32_t ret =
         tsParseOneColumnKV(pSchema, &sToken, row, pInsertParam->msg, str, isPrimaryKey, timePrec, toffset, colId);
@@ -630,7 +630,7 @@ void tscSetBoundColumnInfo(SParsedDataColInfo *pColInfo, SSchema *pSchema, int32
   pColInfo->numOfCols = numOfCols;
   pColInfo->numOfBound = numOfCols;
   pColInfo->orderStatus = ORDER_STATUS_ORDERED;  // default is ORDERED for non-bound mode
-  pColInfo->boundedColumns = calloc(pColInfo->numOfCols, sizeof(int32_t));
+  pColInfo->boundColumns = calloc(pColInfo->numOfCols, sizeof(int32_t));
   pColInfo->cols = calloc(pColInfo->numOfCols, sizeof(SBoundColumn));
   pColInfo->colIdxInfo = NULL;
   pColInfo->flen = 0;
@@ -656,7 +656,7 @@ void tscSetBoundColumnInfo(SParsedDataColInfo *pColInfo, SSchema *pSchema, int32
       default:
         break;
     }
-    pColInfo->boundedColumns[i] = i;
+    pColInfo->boundColumns[i] = i;
   }
   pColInfo->allNullLen += pColInfo->flen;
   pColInfo->boundNullLen = pColInfo->allNullLen;  // default set allNullLen
@@ -991,7 +991,7 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql, char** boundC
     }
 
     for (int i = 0; i < spd.numOfBound; ++i) {
-      SSchema* pSchema = &pTagSchema[spd.boundedColumns[i]];
+      SSchema *pSchema = &pTagSchema[spd.boundColumns[i]];
 
       index = 0;
       sToken = tStrGetToken(sql, &index, true);
@@ -1158,7 +1158,7 @@ static int32_t parseBoundColumns(SInsertStatementParam *pInsertParam, SParsedDat
 
   pColInfo->numOfBound = 0;
   pColInfo->boundNullLen = 0;
-  memset(pColInfo->boundedColumns, 0, sizeof(int32_t) * nCols);
+  memset(pColInfo->boundColumns, 0, sizeof(int32_t) * nCols);
   for (int32_t i = 0; i < nCols; ++i) {
     pColInfo->cols[i].valStat = VAL_STAT_NONE;
   }
@@ -1205,7 +1205,7 @@ static int32_t parseBoundColumns(SInsertStatementParam *pInsertParam, SParsedDat
         }
 
         pColInfo->cols[t].valStat = VAL_STAT_HAS;
-        pColInfo->boundedColumns[pColInfo->numOfBound] = t;
+        pColInfo->boundColumns[pColInfo->numOfBound] = t;
         ++pColInfo->numOfBound;
         switch (pSchema[t].type) {
           case TSDB_DATA_TYPE_BINARY:
@@ -1239,7 +1239,7 @@ static int32_t parseBoundColumns(SInsertStatementParam *pInsertParam, SParsedDat
           }
 
           pColInfo->cols[t].valStat = VAL_STAT_HAS;
-          pColInfo->boundedColumns[pColInfo->numOfBound] = t;
+          pColInfo->boundColumns[pColInfo->numOfBound] = t;
           ++pColInfo->numOfBound;
           switch (pSchema[t].type) {
             case TSDB_DATA_TYPE_BINARY:
@@ -1279,7 +1279,7 @@ static int32_t parseBoundColumns(SInsertStatementParam *pInsertParam, SParsedDat
     }
     SBoundIdxInfo *pColIdx = pColInfo->colIdxInfo;
     for (uint16_t i = 0; i < pColInfo->numOfBound; ++i) {
-      pColIdx[i].schemaColIdx = (uint16_t)pColInfo->boundedColumns[i];
+      pColIdx[i].schemaColIdx = (uint16_t)pColInfo->boundColumns[i];
       pColIdx[i].boundIdx = i;
     }
     qsort(pColIdx, pColInfo->numOfBound, sizeof(SBoundIdxInfo), schemaIdxCompar);
@@ -1289,7 +1289,7 @@ static int32_t parseBoundColumns(SInsertStatementParam *pInsertParam, SParsedDat
     qsort(pColIdx, pColInfo->numOfBound, sizeof(SBoundIdxInfo), boundIdxCompar);
   }
 
-  memset(&pColInfo->boundedColumns[pColInfo->numOfBound], 0,
+  memset(&pColInfo->boundColumns[pColInfo->numOfBound], 0,
          sizeof(int32_t) * (pColInfo->numOfCols - pColInfo->numOfBound));
 
   return TSDB_CODE_SUCCESS;

2.0/src/client/src/tscUtil.c

@@ -1554,7 +1554,7 @@ void tscFreeSqlObj(SSqlObj* pSql) {
 }
 
 void tscDestroyBoundColumnInfo(SParsedDataColInfo* pColInfo) {
-  tfree(pColInfo->boundedColumns);
+  tfree(pColInfo->boundColumns);
   tfree(pColInfo->cols);
   tfree(pColInfo->colIdxInfo);
 }

2.0/src/query/inc/qExecutor.h

@@ -589,7 +589,7 @@ SOperatorInfo* createTableSeqScanOperatorInfo(void* pTsdbQueryHandle, SQueryRunt
 SOperatorInfo* createAggregateOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput);
 SOperatorInfo* createProjectOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput);
 SOperatorInfo* createLimitOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream);
-SOperatorInfo* createTimeIntervalOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput);
+SOperatorInfo* createIntervalOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput);
 SOperatorInfo* createAllTimeIntervalOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput);
 SOperatorInfo* createSWindowOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput);
 SOperatorInfo* createFillOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput, bool multigroupResult);

2.0/src/query/src/qExecutor.c

@@ -2166,7 +2166,7 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int32_t numOf
       }
       case OP_TimeWindow: {
         pRuntimeEnv->proot =
-            createTimeIntervalOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pQueryAttr->pExpr1, pQueryAttr->numOfOutput);
+            createIntervalOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pQueryAttr->pExpr1, pQueryAttr->numOfOutput);
         int32_t opType = pRuntimeEnv->proot->upstream[0]->operatorType;
         if (opType != OP_DummyInput && opType != OP_Join) {
           setTableScanFilterOperatorInfo(pRuntimeEnv->proot->upstream[0]->info, pRuntimeEnv->proot);
@@ -6756,7 +6756,7 @@ SOperatorInfo* createLimitOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorI
   return pOperator;
 }
 
-SOperatorInfo* createTimeIntervalOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput) {
+SOperatorInfo* createIntervalOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput) {
   STableIntervalOperatorInfo* pInfo = calloc(1, sizeof(STableIntervalOperatorInfo));
 
   pInfo->pCtx = createSQLFunctionCtx(pRuntimeEnv, pExpr, numOfOutput, &pInfo->rowCellInfoOffset);

CMakeLists.txt

@@ -8,10 +8,11 @@ project(
 
 set(CMAKE_SUPPORT_DIR "${CMAKE_SOURCE_DIR}/cmake")
 set(CMAKE_CONTRIB_DIR "${CMAKE_SOURCE_DIR}/contrib")
-include(${CMAKE_SUPPORT_DIR}/cmake.options)
 
-SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -Werror -fPIC -gdwarf-2 -msse4.2 -mfma -g3")
-SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Werror -fPIC -gdwarf-2 -msse4.2 -mfma -g3")
+include(${CMAKE_SUPPORT_DIR}/cmake.platform)
+include(${CMAKE_SUPPORT_DIR}/cmake.define)
+include(${CMAKE_SUPPORT_DIR}/cmake.options)
+include(${CMAKE_SUPPORT_DIR}/cmake.version)
 
 # contrib
 add_subdirectory(contrib)
@@ -25,6 +26,7 @@ if(${BUILD_TEST})
     include(CTest)
     enable_testing()
 endif(${BUILD_TEST})
+
 add_subdirectory(source)
 add_subdirectory(tools)
 add_subdirectory(tests)

Jenkinsfile2

@@ -35,6 +35,7 @@ def abort_previous(){
 def pre_test(){
     sh'hostname'
     sh '''
+    date
     sudo rmtaos || echo "taosd has not installed"
     '''
     sh '''
@@ -60,8 +61,9 @@ def pre_test(){
         sh '''
         cd ${WKC}
         git checkout 3.0
+        [ -d contrib/bdb ] && cd contrib/bdb && git clean -fxd && cd ../..
         '''
-      } 
+      }
       else{
         sh '''
         cd ${WKC}
@@ -74,37 +76,9 @@ def pre_test(){
     git pull >/dev/null
     git fetch origin +refs/pull/${CHANGE_ID}/merge
     git checkout -qf FETCH_HEAD
-    git submodule update --init --recursive --remote
+    git submodule update --init --recursive
     '''
-    script {
-      if (env.CHANGE_TARGET == 'master') {
-        sh '''
-        cd ${WKCT}
-        git checkout master
-        '''
-        }
-      else if(env.CHANGE_TARGET == '2.0'){
-        sh '''
-        cd ${WKCT}
-        git checkout 2.0
-        '''
-      }
-      else if(env.CHANGE_TARGET == '3.0'){
-        sh '''
-        cd ${WKCT}
-        git checkout 3.0
-        '''
-      } 
-      else{
-        sh '''
-        cd ${WKCT}
-        git checkout develop
-        '''
-      }
-    }
     sh'''
-    cd ${WKCT}
-    git pull >/dev/null
     cd ${WKC}
     export TZ=Asia/Harbin
     date
@@ -123,11 +97,10 @@ pipeline {
   environment{
       WK = '/var/lib/jenkins/workspace/TDinternal'
       WKC= '/var/lib/jenkins/workspace/TDengine'
-      WKCT= '/var/lib/jenkins/workspace/TDengine/tests'
   }
   stages {
       stage('pre_build'){
-          agent{label 'slave3_0'}
+          agent{label " slave3_0 || slave15 || slave16 || slave17 "}
           options { skipDefaultCheckout() } 
           when {
               changeRequest()

cmake/cmake.define

+cmake_minimum_required(VERSION 3.16)
+
+IF ("${BUILD_TOOLS}" STREQUAL "")
+  IF (TD_LINUX)
+    IF (TD_ARM_32)
+      SET(BUILD_TOOLS "false")
+    ELSEIF (TD_ARM_64)
+      SET(BUILD_TOOLS "false")
+    ELSE ()
+      SET(BUILD_TOOLS "false")
+    ENDIF ()
+  ELSEIF (TD_DARWIN)
+    SET(BUILD_TOOLS "false")
+  ELSE ()
+    SET(BUILD_TOOLS "false")
+  ENDIF ()
+ENDIF ()
+
+IF ("${BUILD_TOOLS}" MATCHES "false")
+    MESSAGE("${Yellow} Will _not_ build taos_tools! ${ColourReset}")
+    SET(TD_TAOS_TOOLS FALSE)
+ELSE ()
+    MESSAGE("")
+    MESSAGE("${Green} Will build taos_tools! ${ColourReset}")
+    MESSAGE("")
+    SET(TD_TAOS_TOOLS TRUE)
+ENDIF ()
+
+IF (TD_WINDOWS)
+    MESSAGE("${Yellow} set compiler flag for Windows! ${ColourReset}")
+    SET(CMAKE_GENERATOR "NMake Makefiles" CACHE INTERNAL "" FORCE)
+    SET(COMMON_FLAGS "/nologo /WX /wd4018 /wd4999 /Oi /Oy- /Gm- /EHsc /MT /GS /Gy /fp:precise /Zc:wchar_t /Zc:forScope /Gd /errorReport:prompt /analyze-")
+
+    IF (MSVC AND (MSVC_VERSION GREATER_EQUAL 1900))
+      SET(COMMON_FLAGS "${COMMON_FLAGS} /Wv:18")
+    ENDIF ()
+
+ELSE ()
+    SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -Werror -fPIC -gdwarf-2 -g3")
+    SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Werror -fPIC -gdwarf-2 -g3")
+
+MESSAGE("System processor ID: ${CMAKE_SYSTEM_PROCESSOR}")
+IF (${CMAKE_SYSTEM_PROCESSOR} MATCHES "arm64")
+    ADD_DEFINITIONS("-D_TD_ARM_")
+ELSE ()
+    ADD_DEFINITIONS("-msse4.2 -mfma")
+ENDIF ()
+
+ENDIF ()

cmake/cmake.options

 # =========================================================
 # Deps options
 # =========================================================
+
+IF(${TD_WINDOWS})
+
+    MESSAGE("build pthread Win32")
+    option(
+            BUILD_PTHREAD
+            "If build pthread on Windows"
+            ON
+          )
+
+    MESSAGE("build gnu regex for Windows")
+    option(
+            BUILD_GNUREGEX
+            "If build gnu regex on Windows"
+            ON
+          )
+
+    MESSAGE("build iconv Win32")
+    option(
+            BUILD_WITH_ICONV
+            "If build iconv on Windows"
+            ON
+          )
+
+ENDIF ()
+
+IF(${TD_LINUX} MATCHES TRUE)
+
 option(
     BUILD_TEST 
     "If build unit tests using googletest" 
     ON
 )
 
+ENDIF ()
+
 option(
     BUILD_WITH_LEVELDB 
     "If build with leveldb" 
@@ -25,11 +55,16 @@ option(
     OFF
 )
 
-option(
-    BUILD_WITH_BDB 
-    "If build with BerkleyDB" 
-    ON
-)
+IF(${TD_WINDOWS})
+    MESSAGE("Not build BDB on Windows")
+ELSE ()
+    option(
+            BUILD_WITH_BDB
+            "If build with BerkleyDB"
+            ON
+          )
+
+ENDIF ()
 
 option(
     BUILD_WITH_LUCENE 
@@ -68,12 +103,16 @@ option(
     OFF
 )
 
+IF(${TD_LINUX} MATCHES TRUE)
+
 option(
     BUILD_DEPENDENCY_TESTS 
     "If build dependency tests" 
     ON
 )
 
+ENDIF ()
+
 option(
     BUILD_DOCS
     "If use doxygen build documents" 

cmake/cmake.platform

+cmake_minimum_required(VERSION 3.16)
+
+MESSAGE("Current system is ${CMAKE_SYSTEM_NAME}")
+
+# init
+SET(TD_LINUX FALSE)
+SET(TD_WINDOWS FALSE)
+SET(TD_DARWIN FALSE)
+
+MESSAGE("Compiler ID: ${CMAKE_CXX_COMPILER_ID}")
+if(CMAKE_COMPILER_IS_GNUCXX MATCHES 1)
+    set(CXX_COMPILER_IS_GNU TRUE)
+else()
+    set(CXX_COMPILER_IS_GNU FALSE)
+endif()
+
+MESSAGE("Current system name is ${CMAKE_SYSTEM_NAME}.")
+
+IF (${CMAKE_SYSTEM_NAME} MATCHES "Linux" OR ${CMAKE_SYSTEM_NAME} MATCHES "Darwin")
+
+    IF (${CXX_COMPILER_IS_GNU})
+        SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -std=gnu99")
+    ELSE ()
+        ADD_DEFINITIONS("-Wno-tautological-constant-out-of-range-compare -Wno-pointer-sign -Wno-unknown-warning-option")
+        set(CMAKE_SHARED_LIBRARY_CREATE_C_FLAGS "${CMAKE_SHARED_LIBRARY_CREATE_C_FLAGS} -undefined dynamic_lookup")
+        set(CMAKE_SHARED_LIBRARY_CREATE_CXX_FLAGS "${CMAKE_SHARED_LIBRARY_CREATE_CXX_FLAGS} -undefined dynamic_lookup")
+    ENDIF ()
+
+    IF (${CMAKE_SYSTEM_NAME} MATCHES "Linux")
+
+        SET(TD_LINUX TRUE)
+        SET(OSTYPE "Linux")
+        ADD_DEFINITIONS("-DLINUX")
+
+        IF (${CMAKE_SIZEOF_VOID_P} MATCHES 8)
+            SET(TD_LINUX_64 TRUE)
+        ELSE ()
+            SET(TD_LINUX_32 TRUE)
+        ENDIF ()
+
+    ELSEIF (${CMAKE_SYSTEM_NAME} MATCHES "Darwin")
+
+        SET(TD_DARWIN TRUE)
+        SET(OSTYPE "macOS")
+        ADD_DEFINITIONS("-DDARWIN -Wno-tautological-pointer-compare -Wno-return-type")
+
+        MESSAGE("Current system processor is ${CMAKE_SYSTEM_PROCESSOR}.")
+        IF (${CMAKE_SYSTEM_PROCESSOR} MATCHES "arm64" OR ${CMAKE_SYSTEM_PROCESSOR} MATCHES "x86_64")
+            MESSAGE("Current system arch is arm64")
+            SET(TD_DARWIN_64 TRUE)
+            ADD_DEFINITIONS("-D_TD_DARWIN_64")
+        ENDIF ()
+
+        ADD_DEFINITIONS("-DHAVE_UNISTD_H")
+    ENDIF ()
+
+ELSEIF (${CMAKE_SYSTEM_NAME} MATCHES "Windows")
+
+    SET(TD_WINDOWS TRUE)
+    SET(OSTYPE "Windows")
+    ADD_DEFINITIONS("-DWINDOWS")
+
+    IF (${CMAKE_SIZEOF_VOID_P} MATCHES 8)
+        SET(TD_WINDOWS_64 TRUE)
+    	ADD_DEFINITIONS("-D_TD_WINDOWS_64")
+    ELSE ()
+        SET(TD_WINDOWS_32 TRUE)
+    	ADD_DEFINITIONS("-D_TD_WINDOWS_32")
+    ENDIF ()
+
+ENDIF()
+
+MESSAGE("C Compiler ID: ${CMAKE_C_COMPILER_ID}")
+MESSAGE("CXX Compiler ID: ${CMAKE_CXX_COMPILER_ID}")

cmake/cmake.version

+
+IF (DEFINED VERNUMBER)
+  SET(TD_VER_NUMBER ${VERNUMBER})
+ELSE ()
+  SET(TD_VER_NUMBER "3.0.0.0")
+ENDIF ()
+
+IF (DEFINED VERCOMPATIBLE)
+  SET(TD_VER_COMPATIBLE ${VERCOMPATIBLE})
+ELSE ()
+  SET(TD_VER_COMPATIBLE "3.0.0.0")
+ENDIF ()
+
+find_program(HAVE_GIT NAMES git)
+
+IF (DEFINED GITINFO)
+  SET(TD_VER_GIT ${GITINFO})
+ELSEIF (HAVE_GIT)
+  execute_process(COMMAND git log -1 --format=%H WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR} OUTPUT_VARIABLE GIT_COMMITID)
+  #message(STATUS "git log result:${GIT_COMMITID}")
+  IF (GIT_COMMITID)
+    string (REGEX REPLACE "[\n\t\r]" "" GIT_COMMITID ${GIT_COMMITID})
+    SET(TD_VER_GIT ${GIT_COMMITID})
+  ELSE ()
+    message(STATUS "not a git repository")
+    SET(TD_VER_GIT "no git commit id")
+  ENDIF ()
+ELSE ()
+  message(STATUS "no git cmd")
+  SET(TD_VER_GIT "no git commit id")
+ENDIF ()
+
+IF (DEFINED VERDATE)
+  SET(TD_VER_DATE ${VERDATE})
+ELSE ()
+  STRING(TIMESTAMP TD_VER_DATE "%Y-%m-%d %H:%M:%S") 
+ENDIF ()
+
+IF (DEFINED VERTYPE)
+  SET(TD_VER_VERTYPE ${VERTYPE})
+ELSE ()
+  SET(TD_VER_VERTYPE "stable")
+ENDIF ()
+
+IF (DEFINED CPUTYPE)
+  SET(TD_VER_CPUTYPE ${CPUTYPE})
+ELSE ()
+  IF (TD_WINDOWS_32)
+    SET(TD_VER_CPUTYPE "x86")
+  ELSEIF (TD_LINUX_32)
+    SET(TD_VER_CPUTYPE "x86")  
+  ELSEIF (TD_ARM_32)
+    SET(TD_VER_CPUTYPE "x86")  
+  ELSEIF (TD_MIPS_32)
+    SET(TD_VER_CPUTYPE "x86")  
+  ELSE ()
+    SET(TD_VER_CPUTYPE "x64")
+  ENDIF ()
+ENDIF ()
+
+IF (DEFINED OSTYPE)
+  SET(TD_VER_OSTYPE ${OSTYPE})
+ELSE ()
+  SET(TD_VER_OSTYPE "Linux")
+ENDIF ()
+
+MESSAGE(STATUS "============= compile version parameter information start ============= ")
+MESSAGE(STATUS "ver number:" ${TD_VER_NUMBER})
+MESSAGE(STATUS "compatible ver number:" ${TD_VER_COMPATIBLE})
+MESSAGE(STATUS "communit commit id:" ${TD_VER_GIT})
+MESSAGE(STATUS "build date:" ${TD_VER_DATE})
+MESSAGE(STATUS "ver type:" ${TD_VER_VERTYPE})
+MESSAGE(STATUS "ver cpu:" ${TD_VER_CPUTYPE})
+MESSAGE(STATUS "os type:" ${TD_VER_OSTYPE})
+MESSAGE(STATUS "============= compile version parameter information end  ============= ")
+
+STRING(REPLACE "." "_" TD_LIB_VER_NUMBER ${TD_VER_NUMBER})

cmake/gnuregex_CMakeLists.txt.in

+
+# gnuregex
+ExternalProject_Add(gnuregex
+        URL https://launchpad.net/gnuregex/trunk/2.9/+download/libgnurx-src-2.9.zip
+        DOWNLOAD_NAME libgnurx-src.zip
+        SOURCE_DIR "${CMAKE_CONTRIB_DIR}/gnuregex"
+        BINARY_DIR ""
+        #BUILD_IN_SOURCE TRUE
+        CONFIGURE_COMMAND ""
+        BUILD_COMMAND ""
+        INSTALL_COMMAND ""
+        TEST_COMMAND ""
+)

cmake/iconv_CMakeLists.txt.in

+
+# iconv
+ExternalProject_Add(iconv
+        GIT_REPOSITORY https://github.com/win-iconv/win-iconv.git
+        GIT_TAG v0.0.8
+        SOURCE_DIR "${CMAKE_CONTRIB_DIR}/iconv"
+        BINARY_DIR ""
+        CONFIGURE_COMMAND ""
+        BUILD_COMMAND ""
+        INSTALL_COMMAND ""
+        TEST_COMMAND ""
+        )
\ No newline at end of file

cmake/pthread_CMakeLists.txt.in

+
+# pthread
+ExternalProject_Add(pthread
+        GIT_REPOSITORY https://github.com/GerHobbelt/pthread-win32
+        GIT_TAG v3.0.3.1
+        SOURCE_DIR "${CMAKE_CONTRIB_DIR}/pthread-win32"
+        BINARY_DIR ""
+        #BUILD_IN_SOURCE TRUE
+        CONFIGURE_COMMAND ""
+        BUILD_COMMAND ""
+        INSTALL_COMMAND ""
+        TEST_COMMAND ""
+)

contrib/CMakeLists.txt

@@ -9,6 +9,16 @@ endfunction(cat IN_FILE OUT_FILE)
 set(CONTRIB_TMP_FILE "${CMAKE_BINARY_DIR}/deps_tmp_CMakeLists.txt.in")
 configure_file("${CMAKE_SUPPORT_DIR}/deps_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
 
+# pthread
+if(${BUILD_PTHREAD})
+    cat("${CMAKE_SUPPORT_DIR}/pthread_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
+endif()
+
+# gnu regex
+if(${BUILD_GNUREGEX})
+    cat("${CMAKE_SUPPORT_DIR}/gnuregex_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
+endif()
+
 # googletest
 if(${BUILD_TEST})
     cat("${CMAKE_SUPPORT_DIR}/gtest_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
@@ -117,6 +127,11 @@ target_include_directories(
 # zlib
 set(CMAKE_PROJECT_INCLUDE_BEFORE "${CMAKE_SUPPORT_DIR}/EnableCMP0048.txt.in")
 add_subdirectory(zlib)
+target_include_directories(
+    zlibstatic
+    PUBLIC ${CMAKE_CURRENT_BINARY_DIR}/zlib
+    PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/zlib
+)
 target_include_directories(
     zlib
     PUBLIC ${CMAKE_CURRENT_BINARY_DIR}/zlib
@@ -193,6 +208,10 @@ endif(${BUILD_WITH_TRAFT})
 
 # LIBUV
 if(${BUILD_WITH_UV})
+    if (NOT ${CMAKE_SYSTEM_NAME} MATCHES "Windows")
+        MESSAGE("Windows need set no-sign-compare")
+        add_compile_options(-Wno-sign-compare)
+    endif ()
     add_subdirectory(libuv)
 endif(${BUILD_WITH_UV})
 
@@ -223,6 +242,7 @@ if(${BUILD_WITH_SQLITE})
     )
 endif(${BUILD_WITH_SQLITE})
 
+# pthread
 
 
 # ================================================================================================

contrib/test/craft/raftMain.c

@@ -10,7 +10,7 @@
 #include <raft.h>
 #include <raft/uv.h>
 #include "raftServer.h"
-#include "common.h"
+#include "tcommon.h"
 
 const char *exe_name;
 
@@ -377,7 +377,7 @@ void printConf(SRaftServerConfig *pConf) {
 
 
 int main(int argc, char **argv) { 
-	srand(time(NULL));
+	taosSeedRand(time(NULL));
 	int32_t ret;
 
 	exe_name = argv[0];

contrib/test/craft/raftServer.c

 #include <stdlib.h>
-#include "common.h"
+#include "tcommon.h"
 #include "raftServer.h"
 
 char *keys;

contrib/test/craft/raftServer.h

@@ -11,7 +11,7 @@ extern "C" {
 #include <string.h>
 #include "raft.h"
 #include "raft/uv.h"
-#include "common.h"
+#include "tcommon.h"
 
 
 // simulate a db store, just for test

contrib/test/craft/simulate_vnode.c

@@ -132,7 +132,7 @@ static void proposeValue(struct raft *r) {
     buf.base = raft_malloc(buf.len);
 
     // mock ts value
-    int vid = rand() % VNODE_COUNT;
+    int vid = taosRand() % VNODE_COUNT;
     snprintf(buf.base, buf.len, "%d:value_%ld", vid, time(NULL));
 
 	printf("propose value: %s \n", (char*)buf.base);
@@ -174,7 +174,7 @@ void usage() {
 }
 
 int main(int argc, char **argv) {
-	srand(time(NULL));
+	taosSeedRand(time(NULL));
 
 	exe_name = argv[0];
 	if (argc < 2) {

example/CMakeLists.txt

-aux_source_directory(src TMQ_DEMO_SRC)
+add_executable(tmq "")
+add_executable(tstream "")
 
-add_executable(tmq ${TMQ_DEMO_SRC})
-target_link_libraries(
-    tmq
-    PUBLIC taos
-    #PUBLIC util
-    #PUBLIC common
-    #PUBLIC os
+target_sources(tmq
+    PRIVATE
+    "src/tmq.c"
 )
-target_include_directories(
-    tmq
+
+target_sources(tstream
+    PRIVATE
+    "src/tstream.c"
+)
+target_link_libraries(tmq
+    taos
+)
+
+target_link_libraries(tstream
+    taos
+)
+
+target_include_directories(tmq
+    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
+)
+
+target_include_directories(tstream
     PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
 )
 
 SET_TARGET_PROPERTIES(tmq PROPERTIES OUTPUT_NAME tmq)
+SET_TARGET_PROPERTIES(tstream PROPERTIES OUTPUT_NAME tstream)

example/src/tmq.c

@@ -44,7 +44,7 @@ int32_t init_env() {
 
   pRes = taos_query(pConn, "create stable if not exists st1 (ts timestamp, k int) tags(a int)");
   if (taos_errno(pRes) != 0) {
-    printf("failed to create super table 123_$^), reason:%s\n", taos_errstr(pRes));
+    printf("failed to create super table st1, reason:%s\n", taos_errstr(pRes));
     return -1;
   }
   taos_free_result(pRes);
@@ -66,7 +66,7 @@ int32_t init_env() {
 }
 
 int32_t create_topic() {
-  printf("create topic");
+  printf("create topic\n");
   TAOS_RES* pRes;
   TAOS*     pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
   if (pConn == NULL) {
@@ -91,6 +91,10 @@ int32_t create_topic() {
   return 0;
 }
 
+void tmq_commit_cb_print(tmq_t* tmq, tmq_resp_err_t resp, tmq_topic_vgroup_list_t* offsets, void* param) {
+  printf("commit %d\n", resp);
+}
+
 tmq_t* build_consumer() {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
   assert(pConn != NULL);
@@ -103,6 +107,7 @@ tmq_t* build_consumer() {
 
   tmq_conf_t* conf = tmq_conf_new();
   tmq_conf_set(conf, "group.id", "tg2");
+  tmq_conf_set_offset_commit_cb(conf, tmq_commit_cb_print);
   tmq_t* tmq = tmq_consumer_new(pConn, conf, NULL, 0);
   return tmq;
 }
@@ -144,7 +149,7 @@ void basic_consume_loop(tmq_t* tmq, tmq_list_t* topics) {
 }
 
 void sync_consume_loop(tmq_t* tmq, tmq_list_t* topics) {
-  static const int MIN_COMMIT_COUNT = 1000;
+  static const int MIN_COMMIT_COUNT = 1;
 
   int            msg_count = 0;
   tmq_resp_err_t err;
@@ -155,7 +160,7 @@ void sync_consume_loop(tmq_t* tmq, tmq_list_t* topics) {
   }
 
   while (running) {
-    tmq_message_t* tmqmessage = tmq_consumer_poll(tmq, 500);
+    tmq_message_t* tmqmessage = tmq_consumer_poll(tmq, 1000);
     if (tmqmessage) {
       msg_process(tmqmessage);
       tmq_message_destroy(tmqmessage);
@@ -214,6 +219,6 @@ int main(int argc, char* argv[]) {
   tmq_t*      tmq = build_consumer();
   tmq_list_t* topic_list = build_topic_list();
   /*perf_loop(tmq, topic_list);*/
-  basic_consume_loop(tmq, topic_list);
-  /*sync_consume_loop(tmq, topic_list);*/
+  /*basic_consume_loop(tmq, topic_list);*/
+  sync_consume_loop(tmq, topic_list);
 }

example/src/tstream.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/>.
+ */
+
+#include <assert.h>
+#include <stdio.h>
+#include <string.h>
+#include <time.h>
+#include "taos.h"
+
+int32_t init_env() {
+  TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
+  if (pConn == NULL) {
+    return -1;
+  }
+
+  TAOS_RES* pRes = taos_query(pConn, "create database if not exists abc1 vgroups 1");
+  if (taos_errno(pRes) != 0) {
+    printf("error in create db, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
+  pRes = taos_query(pConn, "use abc1");
+  if (taos_errno(pRes) != 0) {
+    printf("error in use db, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
+  pRes = taos_query(pConn, "create stable if not exists st1 (ts timestamp, k int) tags(a int)");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to create super table st1, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
+  pRes = taos_query(pConn, "create table if not exists tu1 using st1 tags(1)");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to create child table tu1, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
+  pRes = taos_query(pConn, "create table if not exists tu2 using st1 tags(2)");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to create child table tu2, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+  return 0;
+}
+
+int32_t create_stream() {
+  printf("create stream\n");
+  TAOS_RES* pRes;
+  TAOS*     pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
+  if (pConn == NULL) {
+    return -1;
+  }
+
+  pRes = taos_query(pConn, "use abc1");
+  if (taos_errno(pRes) != 0) {
+    printf("error in use db, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
+  /*const char* sql = "select min(k), max(k), sum(k) from tu1";*/
+  const char* sql = "select min(k), max(k), sum(k) as sum_of_k from st1";
+  /*const char* sql = "select sum(k) from tu1 interval(10m)";*/
+  pRes = tmq_create_stream(pConn, "stream1", "out1", sql);
+  if (taos_errno(pRes) != 0) {
+    printf("failed to create stream out1, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+  taos_close(pConn);
+  return 0;
+}
+
+int main(int argc, char* argv[]) {
+  int code;
+  if (argc > 1) {
+    printf("env init\n");
+    code = init_env();
+  }
+  create_stream();
+#if 0
+  tmq_t*      tmq = build_consumer();
+  tmq_list_t* topic_list = build_topic_list();
+  /*perf_loop(tmq, topic_list);*/
+  /*basic_consume_loop(tmq, topic_list);*/
+  sync_consume_loop(tmq, topic_list);
+#endif
+}

examples/c/schemaless.c

@@ -19,7 +19,7 @@ void shuffle(char**lines, size_t n)
     size_t i;
     for (i = 0; i < n - 1; i++)
     {
-      size_t j = i + rand() / (RAND_MAX / (n - i) + 1);
+      size_t j = i + taosRand() / (RAND_MAX / (n - i) + 1);
       char* t = lines[j];
       lines[j] = lines[i];
       lines[i] = t;

include/client/taos.h

@@ -31,26 +31,27 @@ typedef void   TAOS_SUB;
 typedef void **TAOS_ROW;
 
 // Data type definition
-#define TSDB_DATA_TYPE_NULL      0   // 1 bytes
-#define TSDB_DATA_TYPE_BOOL      1   // 1 bytes
-#define TSDB_DATA_TYPE_TINYINT   2   // 1 byte
-#define TSDB_DATA_TYPE_SMALLINT  3   // 2 bytes
-#define TSDB_DATA_TYPE_INT       4   // 4 bytes
-#define TSDB_DATA_TYPE_BIGINT    5   // 8 bytes
-#define TSDB_DATA_TYPE_FLOAT     6   // 4 bytes
-#define TSDB_DATA_TYPE_DOUBLE    7   // 8 bytes
-#define TSDB_DATA_TYPE_BINARY    8   // string, alias for varchar
-#define TSDB_DATA_TYPE_TIMESTAMP 9   // 8 bytes
-#define TSDB_DATA_TYPE_NCHAR     10  // unicode string
-#define TSDB_DATA_TYPE_UTINYINT  11  // 1 byte
-#define TSDB_DATA_TYPE_USMALLINT 12  // 2 bytes
-#define TSDB_DATA_TYPE_UINT      13  // 4 bytes
-#define TSDB_DATA_TYPE_UBIGINT   14  // 8 bytes
-#define TSDB_DATA_TYPE_VARCHAR   15  // string
-#define TSDB_DATA_TYPE_VARBINARY 16  // binary
-#define TSDB_DATA_TYPE_JSON      17  // json
-#define TSDB_DATA_TYPE_DECIMAL   18  // decimal
-#define TSDB_DATA_TYPE_BLOB      19  // binary
+#define TSDB_DATA_TYPE_NULL       0   // 1 bytes
+#define TSDB_DATA_TYPE_BOOL       1   // 1 bytes
+#define TSDB_DATA_TYPE_TINYINT    2   // 1 byte
+#define TSDB_DATA_TYPE_SMALLINT   3   // 2 bytes
+#define TSDB_DATA_TYPE_INT        4   // 4 bytes
+#define TSDB_DATA_TYPE_BIGINT     5   // 8 bytes
+#define TSDB_DATA_TYPE_FLOAT      6   // 4 bytes
+#define TSDB_DATA_TYPE_DOUBLE     7   // 8 bytes
+#define TSDB_DATA_TYPE_VARCHAR    8   // string, alias for varchar
+#define TSDB_DATA_TYPE_TIMESTAMP  9   // 8 bytes
+#define TSDB_DATA_TYPE_NCHAR      10  // unicode string
+#define TSDB_DATA_TYPE_UTINYINT   11  // 1 byte
+#define TSDB_DATA_TYPE_USMALLINT  12  // 2 bytes
+#define TSDB_DATA_TYPE_UINT       13  // 4 bytes
+#define TSDB_DATA_TYPE_UBIGINT    14  // 8 bytes
+#define TSDB_DATA_TYPE_JSON       15  // json string
+#define TSDB_DATA_TYPE_VARBINARY  16  // binary
+#define TSDB_DATA_TYPE_DECIMAL    17  // decimal
+#define TSDB_DATA_TYPE_BLOB       18  // binary
+#define TSDB_DATA_TYPE_MEDIUMBLOB 19
+#define TSDB_DATA_TYPE_BINARY     TSDB_DATA_TYPE_VARCHAR  // string
 
 typedef enum {
   TSDB_OPTION_LOCALE,
@@ -84,11 +85,7 @@ typedef struct taosField {
   int32_t bytes;
 } TAOS_FIELD;
 
-#ifdef _TD_GO_DLL_
-#define DLL_EXPORT __declspec(dllexport)
-#else
 #define DLL_EXPORT
-#endif
 
 typedef void (*__taos_async_fn_t)(void *param, TAOS_RES *, int code);
 
@@ -198,8 +195,8 @@ DLL_EXPORT TAOS_RES *taos_schemaless_insert(TAOS *taos, char *lines[], int numLi
 /* --------------------------TMQ INTERFACE------------------------------- */
 
 enum tmq_resp_err_t {
+  TMQ_RESP_ERR__FAIL = -1,
   TMQ_RESP_ERR__SUCCESS = 0,
-  TMQ_RESP_ERR__FAIL = 1,
 };
 
 typedef enum tmq_resp_err_t tmq_resp_err_t;
@@ -216,18 +213,17 @@ typedef void(tmq_commit_cb(tmq_t *, tmq_resp_err_t, tmq_topic_vgroup_list_t *, v
 
 DLL_EXPORT tmq_list_t *tmq_list_new();
 DLL_EXPORT int32_t     tmq_list_append(tmq_list_t *, const char *);
+DLL_EXPORT void        tmq_list_destroy(tmq_list_t *);
 
-DLL_EXPORT TAOS_RES   *tmq_create_topic(TAOS *taos, const char *name, const char *sql, int sqlLen);
 DLL_EXPORT tmq_t      *tmq_consumer_new(void *conn, tmq_conf_t *conf, char *errstr, int32_t errstrLen);
+DLL_EXPORT tmq_t      *tmq_consumer_new1(tmq_conf_t *conf, char *errstr, int32_t errstrLen);
 DLL_EXPORT void        tmq_message_destroy(tmq_message_t *tmq_message);
 DLL_EXPORT const char *tmq_err2str(tmq_resp_err_t);
 
 /* ------------------------TMQ CONSUMER INTERFACE------------------------ */
 DLL_EXPORT tmq_resp_err_t tmq_subscribe(tmq_t *tmq, tmq_list_t *topic_list);
-#if 0
-DLL_EXPORT tmq_resp_err_t tmq_unsubscribe(tmq_t* tmq);
-DLL_EXPORT tmq_resp_err_t tmq_subscription(tmq_t* tmq, tmq_topic_vgroup_list_t** topics);
-#endif
+DLL_EXPORT tmq_resp_err_t tmq_unsubscribe(tmq_t *tmq);
+DLL_EXPORT tmq_resp_err_t tmq_subscription(tmq_t *tmq, tmq_list_t **topics);
 DLL_EXPORT tmq_message_t *tmq_consumer_poll(tmq_t *tmq, int64_t blocking_time);
 DLL_EXPORT tmq_resp_err_t tmq_consumer_close(tmq_t *tmq);
 #if 0
@@ -238,6 +234,7 @@ DLL_EXPORT tmq_resp_err_t tmq_commit(tmq_t *tmq, const tmq_topic_vgroup_list_t *
 #if 0
 DLL_EXPORT tmq_resp_err_t tmq_commit_message(tmq_t* tmq, const tmq_message_t* tmqmessage, int32_t async);
 #endif
+DLL_EXPORT tmq_resp_err_t tmq_seek(tmq_t *tmq, const tmq_topic_vgroup_t *offset);
 /* ----------------------TMQ CONFIGURATION INTERFACE---------------------- */
 
 enum tmq_conf_res_t {
@@ -249,14 +246,30 @@ enum tmq_conf_res_t {
 typedef enum tmq_conf_res_t tmq_conf_res_t;
 
 DLL_EXPORT tmq_conf_t    *tmq_conf_new();
-DLL_EXPORT void           tmq_conf_destroy(tmq_conf_t *conf);
 DLL_EXPORT tmq_conf_res_t tmq_conf_set(tmq_conf_t *conf, const char *key, const char *value);
+DLL_EXPORT void           tmq_conf_destroy(tmq_conf_t *conf);
 DLL_EXPORT void           tmq_conf_set_offset_commit_cb(tmq_conf_t *conf, tmq_commit_cb *cb);
 
 // temporary used function for demo only
 void    tmqShowMsg(tmq_message_t *tmq_message);
 int32_t tmqGetSkipLogNum(tmq_message_t *tmq_message);
 
+/* -------------------------TMQ MSG HANDLE INTERFACE---------------------- */
+
+DLL_EXPORT TAOS_ROW tmq_get_row(tmq_message_t *message);
+DLL_EXPORT char    *tmq_get_topic_name(tmq_message_t *message);
+DLL_EXPORT char    *tmq_get_topic_schema(tmq_t *tmq, const char *topic);
+
+/* --------------------TMPORARY INTERFACE FOR TESTING--------------------- */
+DLL_EXPORT TAOS_RES *tmq_create_topic(TAOS *taos, const char *name, const char *sql, int sqlLen);
+
+DLL_EXPORT TAOS_RES *tmq_create_stream(TAOS *taos, const char *streamName, const char *tbName, const char *sql);
+
+/* -------------------------------- OTHER -------------------------------- */
+typedef void (*TAOS_SUBSCRIBE_CALLBACK)(TAOS_SUB *tsub, TAOS_RES *res, void *param, int code);
+
+DLL_EXPORT int taos_stmt_affected_rows(TAOS_STMT *stmt);
+
 #ifdef __cplusplus
 }
 #endif

include/common/common.h

-/*
- * 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/>.
- */
-
-#ifndef TDENGINE_COMMON_H
-#define TDENGINE_COMMON_H
-
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-#include "taosdef.h"
-#include "tarray.h"
-#include "tmsg.h"
-#include "tvariant.h"
-// typedef struct STimeWindow {
-//   TSKEY skey;
-//   TSKEY ekey;
-// } STimeWindow;
-
-// typedef struct {
-//   int32_t dataLen;
-//   char    name[TSDB_TABLE_FNAME_LEN];
-//   char   *data;
-// } STagData;
-
-// typedef struct SSchema {
-//   uint8_t type;
-//   char    name[TSDB_COL_NAME_LEN];
-//   int16_t colId;
-//   int16_t bytes;
-// } SSchema;
-
-#define TMQ_REQ_TYPE_COMMIT_ONLY 0
-#define TMQ_REQ_TYPE_CONSUME_ONLY 1
-#define TMQ_REQ_TYPE_CONSUME_AND_COMMIT 2
-
-typedef struct {
-  uint32_t  numOfTables;
-  SArray   *pGroupList;
-  SHashObj *map;  // speedup acquire the tableQueryInfo by table uid
-} STableGroupInfo;
-
-typedef struct SColumnDataAgg {
-  int16_t colId;
-  int64_t sum;
-  int64_t max;
-  int64_t min;
-  int16_t maxIndex;
-  int16_t minIndex;
-  int16_t numOfNull;
-} SColumnDataAgg;
-
-typedef struct SDataBlockInfo {
-  STimeWindow window;
-  int32_t     rows;
-  int32_t     numOfCols;
-  int64_t     uid;
-} SDataBlockInfo;
-
-typedef struct SConstantItem {
-  SColumnInfo info;
-  int32_t     startRow;  // run-length-encoding to save the space for multiple rows
-  int32_t     endRow;
-  SVariant    value;
-} SConstantItem;
-
-// info.numOfCols = taosArrayGetSize(pDataBlock) + taosArrayGetSize(pConstantList);
-typedef struct SSDataBlock {
-  SColumnDataAgg *pBlockAgg;
-  SArray         *pDataBlock;    // SArray<SColumnInfoData>
-  SArray         *pConstantList; // SArray<SConstantItem>, it is a constant/tags value of the corresponding result value.
-  SDataBlockInfo  info;
-} SSDataBlock;
-
-typedef struct SVarColAttr {
-  int32_t *offset;    // start position for each entry in the list
-  uint32_t length;    // used buffer size that contain the valid data
-  uint32_t allocLen;  // allocated buffer size
-} SVarColAttr;
-
-// pBlockAgg->numOfNull == info.rows, all data are null
-// pBlockAgg->numOfNull == 0, no data are null.
-typedef struct SColumnInfoData {
-  SColumnInfo info;   // TODO filter info needs to be removed
-  bool        hasNull;// if current column data has null value.
-  char       *pData;  // the corresponding block data in memory
-  union {
-    char       *nullbitmap;  // bitmap, one bit for each item in the list
-    SVarColAttr varmeta;
-  };
-} SColumnInfoData;
-
-static FORCE_INLINE int32_t tEncodeDataBlock(void** buf, const SSDataBlock* pBlock) {
-  int64_t tbUid = pBlock->info.uid;
-  int32_t numOfCols = pBlock->info.numOfCols;
-  int32_t rows = pBlock->info.rows;
-  int32_t sz = taosArrayGetSize(pBlock->pDataBlock);
-
-  int32_t tlen = 0;
-  tlen += taosEncodeFixedI64(buf, tbUid);
-  tlen += taosEncodeFixedI32(buf, numOfCols);
-  tlen += taosEncodeFixedI32(buf, rows);
-  tlen += taosEncodeFixedI32(buf, sz);
-  for (int32_t i = 0; i < sz; i++) {
-    SColumnInfoData* pColData = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, i);
-    tlen += taosEncodeFixedI16(buf, pColData->info.colId);
-    tlen += taosEncodeFixedI16(buf, pColData->info.type);
-    tlen += taosEncodeFixedI16(buf, pColData->info.bytes);
-    int32_t colSz = rows * pColData->info.bytes;
-    tlen += taosEncodeBinary(buf, pColData->pData, colSz);
-  }
-  return tlen;
-}
-
-static FORCE_INLINE void* tDecodeDataBlock(const void* buf, SSDataBlock* pBlock) {
-  int32_t sz;
-
-  buf = taosDecodeFixedI64(buf, &pBlock->info.uid);
-  buf = taosDecodeFixedI32(buf, &pBlock->info.numOfCols);
-  buf = taosDecodeFixedI32(buf, &pBlock->info.rows);
-  buf = taosDecodeFixedI32(buf, &sz);
-  pBlock->pDataBlock = taosArrayInit(sz, sizeof(SColumnInfoData));
-  for (int32_t i = 0; i < sz; i++) {
-    SColumnInfoData data = {0};
-    buf = taosDecodeFixedI16(buf, &data.info.colId);
-    buf = taosDecodeFixedI16(buf, &data.info.type);
-    buf = taosDecodeFixedI16(buf, &data.info.bytes);
-    int32_t colSz = pBlock->info.rows * data.info.bytes;
-    buf = taosDecodeBinary(buf, (void**)&data.pData, colSz);
-    taosArrayPush(pBlock->pDataBlock, &data);
-  }
-  return (void*)buf;
-}
-
-static FORCE_INLINE int32_t tEncodeSMqConsumeRsp(void** buf, const SMqConsumeRsp* pRsp) {
-  int32_t tlen = 0;
-  int32_t sz = 0;
-  tlen += taosEncodeFixedI64(buf, pRsp->consumerId);
-  tlen += taosEncodeFixedI64(buf, pRsp->committedOffset);
-  tlen += taosEncodeFixedI64(buf, pRsp->reqOffset);
-  tlen += taosEncodeFixedI64(buf, pRsp->rspOffset);
-  tlen += taosEncodeFixedI32(buf, pRsp->skipLogNum);
-  tlen += taosEncodeFixedI32(buf, pRsp->numOfTopics);
-  if (pRsp->numOfTopics == 0) return tlen;
-  tlen += tEncodeSSchemaWrapper(buf, pRsp->schemas);
-  if (pRsp->pBlockData) {
-    sz = taosArrayGetSize(pRsp->pBlockData);
-  }
-  tlen += taosEncodeFixedI32(buf, sz);
-  for (int32_t i = 0; i < sz; i++) {
-    SSDataBlock* pBlock = (SSDataBlock*)taosArrayGet(pRsp->pBlockData, i);
-    tlen += tEncodeDataBlock(buf, pBlock);
-  }
-  return tlen;
-}
-
-static FORCE_INLINE void* tDecodeSMqConsumeRsp(void* buf, SMqConsumeRsp* pRsp) {
-  int32_t sz;
-  buf = taosDecodeFixedI64(buf, &pRsp->consumerId);
-  buf = taosDecodeFixedI64(buf, &pRsp->committedOffset);
-  buf = taosDecodeFixedI64(buf, &pRsp->reqOffset);
-  buf = taosDecodeFixedI64(buf, &pRsp->rspOffset);
-  buf = taosDecodeFixedI32(buf, &pRsp->skipLogNum);
-  buf = taosDecodeFixedI32(buf, &pRsp->numOfTopics);
-  if (pRsp->numOfTopics == 0) return buf;
-  pRsp->schemas = (SSchemaWrapper*)calloc(1, sizeof(SSchemaWrapper));
-  if (pRsp->schemas == NULL) return NULL;
-  buf = tDecodeSSchemaWrapper(buf, pRsp->schemas);
-  buf = taosDecodeFixedI32(buf, &sz);
-  pRsp->pBlockData = taosArrayInit(sz, sizeof(SSDataBlock));
-  for (int32_t i = 0; i < sz; i++) {
-    SSDataBlock block = {0};
-    tDecodeDataBlock(buf, &block);
-    taosArrayPush(pRsp->pBlockData, &block);
-  }
-  return buf;
-}
-
-static FORCE_INLINE void tDeleteSSDataBlock(SSDataBlock* pBlock) {
-  if (pBlock == NULL) {
-    return;
-  }
-
-  // int32_t numOfOutput = pBlock->info.numOfCols;
-  int32_t sz = taosArrayGetSize(pBlock->pDataBlock);
-  for (int32_t i = 0; i < sz; ++i) {
-    SColumnInfoData* pColInfoData = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, i);
-    tfree(pColInfoData->pData);
-  }
-
-  taosArrayDestroy(pBlock->pDataBlock);
-  tfree(pBlock->pBlockAgg);
-  // tfree(pBlock);
-}
-
-static FORCE_INLINE void tDeleteSMqConsumeRsp(SMqConsumeRsp* pRsp) {
-  if (pRsp->schemas) {
-    if (pRsp->schemas->nCols) {
-      tfree(pRsp->schemas->pSchema);
-    }
-    free(pRsp->schemas);
-  }
-  taosArrayDestroyEx(pRsp->pBlockData, (void (*)(void*))tDeleteSSDataBlock);
-  pRsp->pBlockData = NULL;
-  // for (int i = 0; i < taosArrayGetSize(pRsp->pBlockData); i++) {
-  // SSDataBlock* pDataBlock = (SSDataBlock*)taosArrayGet(pRsp->pBlockData, i);
-  // tDeleteSSDataBlock(pDataBlock);
-  //}
-}
-
-//======================================================================================================================
-// the following structure shared by parser and executor
-typedef struct SColumn {
-  uint64_t    uid;
-  char        name[TSDB_COL_NAME_LEN];
-  int8_t      flag;  // column type: normal column, tag, or user-input column (integer/float/string)
-  SColumnInfo info;
-} SColumn;
-
-typedef struct SLimit {
-  int64_t limit;
-  int64_t offset;
-} SLimit;
-
-typedef struct SOrder {
-  uint32_t order;
-  SColumn  col;
-} SOrder;
-
-typedef struct SGroupbyExpr {
-  SArray* columnInfo;  // SArray<SColIndex>, group by columns information
-  bool    groupbyTag;  // group by tag or column
-} SGroupbyExpr;
-
-// the structure for sql function in select clause
-typedef struct SSqlExpr {
-  char    token[TSDB_COL_NAME_LEN];  // original token
-  SSchema resSchema;
-
-  int32_t   numOfCols;
-  SColumn*  pColumns;       // data columns that are required by query
-  int32_t   interBytes;     // inter result buffer size
-  int16_t   numOfParams;    // argument value of each function
-  SVariant  param[3];       // parameters are not more than 3
-} SSqlExpr;
-
-typedef struct SExprInfo {
-  struct SSqlExpr   base;
-  struct tExprNode* pExpr;
-} SExprInfo;
-
-typedef struct SStateWindow {
-  SColumn col;
-} SStateWindow;
-
-typedef struct SSessionWindow {
-  int64_t gap;  // gap between two session window(in microseconds)
-  SColumn col;
-} SSessionWindow;
-
-#define QUERY_ASC_FORWARD_STEP 1
-#define QUERY_DESC_FORWARD_STEP -1
-
-#define GET_FORWARD_DIRECTION_FACTOR(ord) (((ord) == TSDB_ORDER_ASC) ? QUERY_ASC_FORWARD_STEP : QUERY_DESC_FORWARD_STEP)
-
-#ifdef __cplusplus
-}
-#endif
-
-#endif  // TDENGINE_COMMON_H

include/common/taosdef.h

@@ -16,44 +16,29 @@
 #ifndef _TD_COMMON_TAOS_DEF_H_
 #define _TD_COMMON_TAOS_DEF_H_
 
+#include "taos.h"
+#include "tdef.h"
+
 #ifdef __cplusplus
 extern "C" {
 #endif
 
-#include "taos.h"
-#include "tdef.h"
-
 typedef int64_t tb_uid_t;
 
-#define TSWINDOW_INITIALIZER ((STimeWindow){INT64_MIN, INT64_MAX})
-#define TSWINDOW_DESC_INITIALIZER ((STimeWindow){INT64_MAX, INT64_MIN})
+#define TSWINDOW_INITIALIZER       ((STimeWindow){INT64_MIN, INT64_MAX})
+#define TSWINDOW_DESC_INITIALIZER  ((STimeWindow){INT64_MAX, INT64_MIN})
 #define IS_TSWINDOW_SPECIFIED(win) (((win).skey != INT64_MIN) || ((win).ekey != INT64_MAX))
 
 typedef enum {
-  TAOS_QTYPE_RPC = 1,
-  TAOS_QTYPE_FWD = 2,
-  TAOS_QTYPE_WAL = 3,
-  TAOS_QTYPE_CQ = 4,
-  TAOS_QTYPE_QUERY = 5
-} EQType;
-
-typedef enum {
-  TSDB_SUPER_TABLE = 1,   // super table
-  TSDB_CHILD_TABLE = 2,   // table created from super table
-  TSDB_NORMAL_TABLE = 3,  // ordinary table
-  TSDB_STREAM_TABLE = 4,  // table created from stream computing
-  TSDB_TEMP_TABLE = 5,    // temp table created by nest query
-  TSDB_TABLE_MAX = 6
+  TSDB_SUPER_TABLE  = 1,    // super table
+  TSDB_CHILD_TABLE  = 2,    // table created from super table
+  TSDB_NORMAL_TABLE = 3,    // ordinary table
+  TSDB_STREAM_TABLE = 4,    // table created from stream computing
+  TSDB_TEMP_TABLE   = 5,    // temp table created by nest query
+  TSDB_SYSTEM_TABLE = 6,
+  TSDB_TABLE_MAX    = 7
 } ETableType;
 
-typedef enum {
-  TSDB_MOD_MNODE = 1,
-  TSDB_MOD_HTTP = 2,
-  TSDB_MOD_MONITOR = 3,
-  TSDB_MOD_MQTT = 4,
-  TSDB_MOD_MAX = 5
-} EModuleType;
-
 typedef enum {
   TSDB_CHECK_ITEM_NETWORK,
   TSDB_CHECK_ITEM_MEM,
@@ -66,10 +51,35 @@ typedef enum {
   TSDB_CHECK_ITEM_MAX
 } ECheckItemType;
 
-typedef enum { TD_ROW_DISCARD_UPDATE = 0, TD_ROW_OVERWRITE_UPDATE = 1, TD_ROW_PARTIAL_UPDATE = 2 } TDUpdateConfig;
+typedef enum {
+  TD_ROW_DISCARD_UPDATE = 0,
+  TD_ROW_OVERWRITE_UPDATE = 1,
+  TD_ROW_PARTIAL_UPDATE = 2,
+} TDUpdateConfig;
+
+typedef enum {
+  TSDB_STATIS_OK = 0,    // statis part exist and load successfully
+  TSDB_STATIS_NONE = 1,  // statis part not exist
+} ETsdbStatisStatus;
+
+typedef enum {
+  TSDB_SMA_STAT_UNKNOWN = -1,  // unknown
+  TSDB_SMA_STAT_OK = 0,        // ready to provide service
+  TSDB_SMA_STAT_EXPIRED = 1,   // not ready or expired
+  TSDB_SMA_STAT_DROPPED = 2,   // sma dropped
+} ETsdbSmaStat;                // bit operation
+
+
+typedef enum {
+  TSDB_SMA_TYPE_BLOCK = 0,       // Block-wise SMA
+  TSDB_SMA_TYPE_TIME_RANGE = 1,  // Time-range-wise SMA
+  TSDB_SMA_TYPE_ROLLUP = 2,      // Rollup SMA
+} ETsdbSmaType;
 
 extern char *qtypeStr[];
 
+#define TSDB_PORT_HTTP      11
+
 #ifdef __cplusplus
 }
 #endif

include/common/tcfg.h

-/*
- * 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/>.
- */
-
-#ifndef _TD_COMMON_CFG_H_
-#define _TD_COMMON_CFG_H_
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-#include "tdef.h"
-
-typedef struct {
-  char dir[TSDB_FILENAME_LEN];
-  int  level;
-  int  primary;
-} SDiskCfg;
-
-#ifdef __cplusplus
-}
-#endif
-
-#endif /*_TD_COMMON_CFG_H_*/

include/common/tcommon.h

+/*
+ * 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/>.
+ */
+
+#ifndef _TD_COMMON_DEF_H_
+#define _TD_COMMON_DEF_H_
+
+#include "taosdef.h"
+#include "tarray.h"
+#include "tmsg.h"
+#include "tvariant.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+enum {
+  TMQ_CONF__RESET_OFFSET__LATEST = -1,
+  TMQ_CONF__RESET_OFFSET__EARLIEAST = -2,
+  TMQ_CONF__RESET_OFFSET__NONE = -3,
+};
+
+enum {
+  TMQ_MSG_TYPE__DUMMY = 0,
+  TMQ_MSG_TYPE__POLL_RSP,
+  TMQ_MSG_TYPE__EP_RSP,
+};
+
+typedef struct {
+  uint32_t  numOfTables;
+  SArray*   pGroupList;
+  SHashObj* map;  // speedup acquire the tableQueryInfo by table uid
+} STableGroupInfo;
+
+typedef struct SColumnDataAgg {
+  int16_t colId;
+  int64_t sum;
+  int64_t max;
+  int64_t min;
+  int16_t maxIndex;
+  int16_t minIndex;
+  int16_t numOfNull;
+} SColumnDataAgg;
+
+typedef struct SDataBlockInfo {
+  STimeWindow window;
+  int32_t     rows;
+  int32_t     rowSize;
+  int16_t     numOfCols;
+  int16_t     hasVarCol;
+  union {
+    int64_t uid;
+    int64_t blockId;
+  };
+  int64_t groupId;  // no need to serialize
+} SDataBlockInfo;
+
+typedef struct SSDataBlock {
+  SColumnDataAgg* pBlockAgg;
+  SArray*         pDataBlock;  // SArray<SColumnInfoData>
+  SDataBlockInfo  info;
+} SSDataBlock;
+
+typedef struct SVarColAttr {
+  int32_t* offset;    // start position for each entry in the list
+  uint32_t length;    // used buffer size that contain the valid data
+  uint32_t allocLen;  // allocated buffer size
+} SVarColAttr;
+
+// pBlockAgg->numOfNull == info.rows, all data are null
+// pBlockAgg->numOfNull == 0, no data are null.
+typedef struct SColumnInfoData {
+  SColumnInfo info;     // TODO filter info needs to be removed
+  bool        hasNull;  // if current column data has null value.
+  char*       pData;    // the corresponding block data in memory
+  union {
+    char*       nullbitmap;  // bitmap, one bit for each item in the list
+    SVarColAttr varmeta;
+  };
+} SColumnInfoData;
+
+void*   blockDataDestroy(SSDataBlock* pBlock);
+int32_t tEncodeDataBlock(void** buf, const SSDataBlock* pBlock);
+void*   tDecodeDataBlock(const void* buf, SSDataBlock* pBlock);
+
+int32_t tEncodeDataBlocks(void** buf, const SArray* blocks);
+void*   tDecodeDataBlocks(const void* buf, SArray** blocks);
+
+static FORCE_INLINE void blockDestroyInner(SSDataBlock* pBlock) {
+  // WARNING: do not use info.numOfCols,
+  // sometimes info.numOfCols != array size
+  int32_t numOfOutput = taosArrayGetSize(pBlock->pDataBlock);
+  for (int32_t i = 0; i < numOfOutput; ++i) {
+    SColumnInfoData* pColInfoData = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, i);
+    if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) {
+      taosMemoryFreeClear(pColInfoData->varmeta.offset);
+    } else {
+      taosMemoryFreeClear(pColInfoData->nullbitmap);
+    }
+
+    taosMemoryFreeClear(pColInfoData->pData);
+  }
+
+  taosArrayDestroy(pBlock->pDataBlock);
+  taosMemoryFreeClear(pBlock->pBlockAgg);
+}
+
+static FORCE_INLINE void tDeleteSSDataBlock(SSDataBlock* pBlock) { blockDestroyInner(pBlock); }
+
+static FORCE_INLINE int32_t tEncodeSMqPollRsp(void** buf, const SMqPollRsp* pRsp) {
+  int32_t tlen = 0;
+  int32_t sz = 0;
+  // tlen += taosEncodeFixedI64(buf, pRsp->consumerId);
+  tlen += taosEncodeFixedI64(buf, pRsp->reqOffset);
+  tlen += taosEncodeFixedI64(buf, pRsp->rspOffset);
+  tlen += taosEncodeFixedI32(buf, pRsp->skipLogNum);
+  tlen += taosEncodeFixedI32(buf, pRsp->numOfTopics);
+  if (pRsp->numOfTopics == 0) return tlen;
+  tlen += taosEncodeSSchemaWrapper(buf, pRsp->schema);
+  if (pRsp->pBlockData) {
+    sz = taosArrayGetSize(pRsp->pBlockData);
+  }
+  tlen += taosEncodeFixedI32(buf, sz);
+  for (int32_t i = 0; i < sz; i++) {
+    SSDataBlock* pBlock = (SSDataBlock*)taosArrayGet(pRsp->pBlockData, i);
+    tlen += tEncodeDataBlock(buf, pBlock);
+  }
+  return tlen;
+}
+
+static FORCE_INLINE void* tDecodeSMqPollRsp(void* buf, SMqPollRsp* pRsp) {
+  int32_t sz;
+  // buf = taosDecodeFixedI64(buf, &pRsp->consumerId);
+  buf = taosDecodeFixedI64(buf, &pRsp->reqOffset);
+  buf = taosDecodeFixedI64(buf, &pRsp->rspOffset);
+  buf = taosDecodeFixedI32(buf, &pRsp->skipLogNum);
+  buf = taosDecodeFixedI32(buf, &pRsp->numOfTopics);
+  if (pRsp->numOfTopics == 0) return buf;
+  pRsp->schema = (SSchemaWrapper*)taosMemoryCalloc(1, sizeof(SSchemaWrapper));
+  if (pRsp->schema == NULL) return NULL;
+  buf = taosDecodeSSchemaWrapper(buf, pRsp->schema);
+  buf = taosDecodeFixedI32(buf, &sz);
+  pRsp->pBlockData = taosArrayInit(sz, sizeof(SSDataBlock));
+  for (int32_t i = 0; i < sz; i++) {
+    SSDataBlock block = {0};
+    tDecodeDataBlock(buf, &block);
+    taosArrayPush(pRsp->pBlockData, &block);
+  }
+  return buf;
+}
+
+static FORCE_INLINE void tDeleteSMqConsumeRsp(SMqPollRsp* pRsp) {
+  if (pRsp->schema) {
+    if (pRsp->schema->nCols) {
+      taosMemoryFreeClear(pRsp->schema->pSchema);
+    }
+    taosMemoryFree(pRsp->schema);
+  }
+  taosArrayDestroyEx(pRsp->pBlockData, (void (*)(void*))blockDestroyInner);
+  pRsp->pBlockData = NULL;
+}
+
+//======================================================================================================================
+// the following structure shared by parser and executor
+typedef struct SColumn {
+  union {
+    uint64_t uid;
+    int64_t  dataBlockId;
+  };
+
+  int16_t colId;
+  int16_t slotId;
+
+  char    name[TSDB_COL_NAME_LEN];
+  int8_t  flag;  // column type: normal column, tag, or user-input column (integer/float/string)
+  int16_t type;
+  int32_t bytes;
+  uint8_t precision;
+  uint8_t scale;
+} SColumn;
+
+typedef struct SLimit {
+  int64_t limit;
+  int64_t offset;
+} SLimit;
+
+typedef struct SOrder {
+  uint32_t order;
+  SColumn  col;
+} SOrder;
+
+typedef struct SGroupbyExpr {
+  SArray* columnInfo;  // SArray<SColIndex>, group by columns information
+  bool    groupbyTag;  // group by tag or column
+} SGroupbyExpr;
+
+typedef struct SFunctParam {
+  int32_t  type;
+  SColumn* pCol;
+  SVariant param;
+} SFunctParam;
+
+// the structure for sql function in select clause
+typedef struct SResSchame {
+  int8_t  type;
+  int32_t colId;
+  int32_t bytes;
+  int32_t precision;
+  int32_t scale;
+  char    name[TSDB_COL_NAME_LEN];
+} SResSchema;
+
+// TODO move away to executor.h
+typedef struct SExprBasicInfo {
+  SResSchema   resSchema;
+  int16_t      numOfParams;  // argument value of each function
+  SFunctParam* pParam;
+} SExprBasicInfo;
+
+typedef struct SExprInfo {
+  struct SExprBasicInfo base;
+  struct tExprNode*     pExpr;
+} SExprInfo;
+
+typedef struct SStateWindow {
+  SColumn col;
+} SStateWindow;
+
+typedef struct SSessionWindow {
+  int64_t gap;  // gap between two session window(in microseconds)
+  SColumn col;
+} SSessionWindow;
+
+#define QUERY_ASC_FORWARD_STEP  1
+#define QUERY_DESC_FORWARD_STEP -1
+
+#define GET_FORWARD_DIRECTION_FACTOR(ord) (((ord) == TSDB_ORDER_ASC) ? QUERY_ASC_FORWARD_STEP : QUERY_DESC_FORWARD_STEP)
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TD_COMMON_DEF_H_*/

include/common/tcompare.h

-/*
- * 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/>.
- */
-
-#ifndef _TD_TCOMPARE_H_
-#define _TD_TCOMPARE_H_
-
-#include "compare.h"
-#include "ttypes.h"
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-int32_t compareStrPatternMatch(const void* pLeft, const void* pRight);
-int32_t compareStrPatternNotMatch(const void* pLeft, const void* pRight);
-
-int32_t compareWStrPatternMatch(const void* pLeft, const void* pRight);
-int32_t compareWStrPatternNotMatch(const void* pLeft, const void* pRight);
-
-__compar_fn_t getComparFunc(int32_t type, int32_t optr);
-__compar_fn_t getKeyComparFunc(int32_t keyType, int32_t order);
-int32_t       doCompare(const char* a, const char* b, int32_t type, size_t size);
-
-#ifdef __cplusplus
-}
-#endif
-
-#endif /*_TD_TCOMPARE_H_*/

include/common/tdatablock.h

+/*
+ * 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/>.
+ */
+
+#ifndef _TD_COMMON_EP_H_
+#define _TD_COMMON_EP_H_
+
+#include "tcommon.h"
+#include "tmsg.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+typedef struct SCorEpSet {
+  int32_t version;
+  SEpSet  epSet;
+} SCorEpSet;
+
+typedef struct SBlockOrderInfo {
+  int32_t          order;
+  int32_t          colIndex;
+  SColumnInfoData* pColData;
+} SBlockOrderInfo;
+
+int32_t taosGetFqdnPortFromEp(const char* ep, SEp* pEp);
+void    addEpIntoEpSet(SEpSet* pEpSet, const char* fqdn, uint16_t port);
+
+bool isEpsetEqual(const SEpSet* s1, const SEpSet* s2);
+
+void   updateEpSet_s(SCorEpSet* pEpSet, SEpSet* pNewEpSet);
+SEpSet getEpSet_s(SCorEpSet* pEpSet);
+
+#define NBIT                     (3u)
+#define BitPos(_n)               ((_n) & ((1 << NBIT) - 1))
+#define BMCharPos(bm_, r_)       ((bm_)[(r_) >> NBIT])
+#define colDataIsNull_f(bm_, r_) ((BMCharPos(bm_, r_) & (1u << (7u - BitPos(r_)))) == (1u << (7u - BitPos(r_))))
+
+#define colDataSetNull_f(bm_, r_)                    \
+  do {                                               \
+    BMCharPos(bm_, r_) |= (1u << (7u - BitPos(r_))); \
+  } while (0)
+
+static FORCE_INLINE bool colDataIsNull_s(const SColumnInfoData* pColumnInfoData, uint32_t row) {
+  if (!pColumnInfoData->hasNull) {
+    return false;
+  }
+  if (IS_VAR_DATA_TYPE(pColumnInfoData->info.type)) {
+    return pColumnInfoData->varmeta.offset[row] == -1;
+  } else {
+    if (pColumnInfoData->nullbitmap == NULL) {
+      return false;
+    }
+
+    return colDataIsNull_f(pColumnInfoData->nullbitmap, row);
+  }
+}
+
+static FORCE_INLINE bool colDataIsNull(const SColumnInfoData* pColumnInfoData, uint32_t totalRows, uint32_t row,
+                                       SColumnDataAgg* pColAgg) {
+  if (!pColumnInfoData->hasNull) {
+    return false;
+  }
+
+  if (pColAgg != NULL) {
+    if (pColAgg->numOfNull == totalRows) {
+      ASSERT(pColumnInfoData->nullbitmap == NULL);
+      return true;
+    } else if (pColAgg->numOfNull == 0) {
+      ASSERT(pColumnInfoData->nullbitmap == NULL);
+      return false;
+    }
+  }
+
+  if (IS_VAR_DATA_TYPE(pColumnInfoData->info.type)) {
+    return pColumnInfoData->varmeta.offset[row] == -1;
+  } else {
+    if (pColumnInfoData->nullbitmap == NULL) {
+      return false;
+    }
+
+    return colDataIsNull_f(pColumnInfoData->nullbitmap, row);
+  }
+}
+
+#define BitmapLen(_n) (((_n) + ((1 << NBIT) - 1)) >> NBIT)
+
+// SColumnInfoData, rowNumber
+#define colDataGetData(p1_, r_)                                                        \
+  ((IS_VAR_DATA_TYPE((p1_)->info.type)) ? ((p1_)->pData + (p1_)->varmeta.offset[(r_)]) \
+                                        : ((p1_)->pData + ((r_) * (p1_)->info.bytes)))
+
+int32_t colDataAppend(SColumnInfoData* pColumnInfoData, uint32_t currentRow, const char* pData, bool isNull);
+int32_t colDataMergeCol(SColumnInfoData* pColumnInfoData, uint32_t numOfRow1, const SColumnInfoData* pSource,
+                        uint32_t numOfRow2);
+int32_t colDataAssign(SColumnInfoData* pColumnInfoData, const SColumnInfoData* pSource, int32_t numOfRows);
+int32_t blockDataUpdateTsWindow(SSDataBlock* pDataBlock);
+
+int32_t colDataGetLength(const SColumnInfoData* pColumnInfoData, int32_t numOfRows);
+void    colDataTrim(SColumnInfoData* pColumnInfoData);
+
+size_t blockDataGetNumOfCols(const SSDataBlock* pBlock);
+size_t blockDataGetNumOfRows(const SSDataBlock* pBlock);
+
+int32_t blockDataMerge(SSDataBlock* pDest, const SSDataBlock* pSrc);
+int32_t blockDataSplitRows(SSDataBlock* pBlock, bool hasVarCol, int32_t startIndex, int32_t* stopIndex,
+                           int32_t pageSize);
+int32_t blockDataToBuf(char* buf, const SSDataBlock* pBlock);
+int32_t blockDataFromBuf(SSDataBlock* pBlock, const char* buf);
+
+SSDataBlock* blockDataExtractBlock(SSDataBlock* pBlock, int32_t startIndex, int32_t rowCount);
+
+size_t blockDataGetSize(const SSDataBlock* pBlock);
+size_t blockDataGetRowSize(SSDataBlock* pBlock);
+double blockDataGetSerialRowSize(const SSDataBlock* pBlock);
+size_t blockDataGetSerialMetaSize(const SSDataBlock* pBlock);
+
+SSchema* blockDataExtractSchema(const SSDataBlock* pBlock, int32_t* numOfCols);
+
+int32_t blockDataSort(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullFirst);
+int32_t blockDataSort_rv(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullFirst);
+
+int32_t      blockDataEnsureColumnCapacity(SColumnInfoData* pColumn, uint32_t numOfRows);
+int32_t      blockDataEnsureCapacity(SSDataBlock* pDataBlock, uint32_t numOfRows);
+void         blockDataCleanup(SSDataBlock* pDataBlock);
+SSDataBlock* createOneDataBlock(const SSDataBlock* pDataBlock);
+size_t       blockDataGetCapacityInRow(const SSDataBlock* pBlock, size_t pageSize);
+void*        blockDataDestroy(SSDataBlock* pBlock);
+
+void blockDebugShowData(const SArray* dataBlocks);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TD_COMMON_EP_H_*/

include/common/tep.h

-#ifndef TDENGINE_TEP_H
-#define TDENGINE_TEP_H
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-#include "os.h"
-#include "tmsg.h"
-#include "common.h"
-
-typedef struct SCorEpSet {
-  int32_t version;
-  SEpSet  epSet;
-} SCorEpSet;
-
-typedef struct SBlockOrderInfo {
-  int32_t          order;
-  int32_t          colIndex;
-  SColumnInfoData *pColData;
-//  int32_t          type;
-//  int32_t          bytes;
-//  bool             hasNull;
-} SBlockOrderInfo;
-
-int  taosGetFqdnPortFromEp(const char *ep, SEp *pEp);
-void addEpIntoEpSet(SEpSet *pEpSet, const char *fqdn, uint16_t port);
-
-bool isEpsetEqual(const SEpSet *s1, const SEpSet *s2);
-
-void   updateEpSet_s(SCorEpSet *pEpSet, SEpSet *pNewEpSet);
-SEpSet getEpSet_s(SCorEpSet *pEpSet);
-
-#define NBIT                     (3u)
-#define BitPos(_n)               ((_n) & ((1 << NBIT) - 1))
-#define BMCharPos(bm_, r_)       ((bm_)[(r_) >> NBIT])
-#define colDataIsNull_f(bm_, r_) ((BMCharPos(bm_, r_) & (1u << (7u - BitPos(r_)))) == (1u << (7u - BitPos(r_))))
-
-#define colDataSetNull_f(bm_, r_)                    \
-  do {                                               \
-    BMCharPos(bm_, r_) |= (1u << (7u - BitPos(r_))); \
-  } while (0)
-
-static FORCE_INLINE bool colDataIsNull(const SColumnInfoData* pColumnInfoData, uint32_t totalRows, uint32_t row, SColumnDataAgg* pColAgg) {
-  if (!pColumnInfoData->hasNull) {
-    return false;
-  }
-
-  if (pColAgg != NULL) {
-    if (pColAgg->numOfNull == totalRows) {
-      ASSERT(pColumnInfoData->nullbitmap == NULL);
-      return true;
-    } else if (pColAgg->numOfNull == 0) {
-      ASSERT(pColumnInfoData->nullbitmap == NULL);
-      return false;
-    }
-  }
-
-  if (IS_VAR_DATA_TYPE(pColumnInfoData->info.type)) {
-    return pColumnInfoData->varmeta.offset[row] == -1;
-  } else {
-    if (pColumnInfoData->nullbitmap == NULL) {
-      return false;
-    }
-
-    return colDataIsNull_f(pColumnInfoData->nullbitmap, row);
-  }
-}
-
-#define colDataGet(p1_, r_)                                                          \
-  ((IS_VAR_DATA_TYPE((p1_)->info.type)) ? (p1_)->pData + (p1_)->varmeta.offset[(r_)] \
-                                        : (p1_)->pData + ((r_) * (p1_)->info.bytes));
-
-int32_t colDataAppend(SColumnInfoData* pColumnInfoData, uint32_t currentRow, const char* pData, bool isNull);
-int32_t colDataMergeCol(SColumnInfoData* pColumnInfoData, uint32_t numOfRow1, const SColumnInfoData* pSource, uint32_t numOfRow2);
-int32_t blockDataUpdateTsWindow(SSDataBlock* pDataBlock);
-
-int32_t colDataGetSize(const SColumnInfoData* pColumnInfoData, int32_t numOfRows);
-void colDataTrim(SColumnInfoData* pColumnInfoData);
-
-size_t colDataGetNumOfCols(const SSDataBlock* pBlock);
-size_t colDataGetNumOfRows(const SSDataBlock* pBlock);
-
-int32_t blockDataMerge(SSDataBlock* pDest, const SSDataBlock* pSrc);
-int32_t blockDataSplitRows(SSDataBlock* pBlock, bool hasVarCol, int32_t startIndex, int32_t* stopIndex, int32_t pageSize);
-SSDataBlock* blockDataExtractBlock(SSDataBlock* pBlock, int32_t startIndex, int32_t rowCount);
-
-int32_t blockDataToBuf(char* buf, const SSDataBlock* pBlock);
-int32_t blockDataFromBuf(SSDataBlock* pBlock, const char* buf);
-
-size_t blockDataGetSize(const SSDataBlock* pBlock);
-size_t blockDataGetRowSize(const SSDataBlock* pBlock);
-double blockDataGetSerialRowSize(const SSDataBlock* pBlock);
-size_t blockDataGetSerialMetaSize(const SSDataBlock* pBlock);
-
-size_t blockDataNumOfRowsForSerialize(const SSDataBlock* pBlock, int32_t blockSize);
-
-int32_t blockDataSort(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullFirst);
-int32_t blockDataSort_rv(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullFirst);
-
-int32_t blockDataEnsureCapacity(SSDataBlock* pDataBlock, uint32_t numOfRows);
-void    blockDataClearup(SSDataBlock* pDataBlock, bool hasVarCol);
-void   *blockDataDestroy(SSDataBlock *pBlock);
-
-#ifdef __cplusplus
-}
-#endif
-
-#endif  // TDENGINE_TEP_H

include/common/tglobal.h

@@ -16,57 +16,61 @@
 #ifndef _TD_COMMON_GLOBAL_H_
 #define _TD_COMMON_GLOBAL_H_
 
+#include "tarray.h"
+#include "tdef.h"
+
 #ifdef __cplusplus
 extern "C" {
 #endif
 
-#include "tdef.h"
-#include "tcfg.h"
-
 // cluster
 extern char     tsFirst[];
 extern char     tsSecond[];
 extern char     tsLocalFqdn[];
 extern char     tsLocalEp[];
 extern uint16_t tsServerPort;
+extern int32_t  tsVersion;
 extern int32_t  tsStatusInterval;
-extern int8_t   tsEnableTelemetryReporting;
-extern int32_t  tsNumOfSupportVnodes;
+extern bool     tsEnableTelemetryReporting;
 
 // common
-extern int      tsRpcTimer;
-extern int      tsRpcMaxTime;
-extern int      tsRpcForceTcp;  // all commands go to tcp protocol if this is enabled
-extern int32_t  tsMaxConnections;
-extern int32_t  tsMaxShellConns;
-extern int32_t  tsShellActivityTimer;
-extern uint32_t tsMaxTmrCtrl;
-extern float    tsNumOfThreadsPerCore;
-extern int32_t  tsNumOfCommitThreads;
-extern float    tsRatioOfQueryCores;
-extern int8_t   tsDaylight;
-extern int8_t   tsEnableCoreFile;
-extern int32_t  tsCompressMsgSize;
-extern int32_t  tsCompressColData;
-extern int32_t  tsMaxNumOfDistinctResults;
-extern char     tsTempDir[];
-extern int      tsCompatibleModel;  // 2.0 compatible model
-extern int8_t   tsEnableSlaveQuery;
-extern int8_t   tsEnableAdjustMaster;
-extern int8_t   tsPrintAuth;
-extern int64_t  tsTickPerDay[3];
+extern int32_t tsRpcTimer;
+extern int32_t tsRpcMaxTime;
+extern bool    tsRpcForceTcp;  // all commands go to tcp protocol if this is enabled
+extern int32_t tsMaxConnections;
+extern int32_t tsMaxShellConns;
+extern int32_t tsShellActivityTimer;
+extern int32_t tsMaxTmrCtrl;
+extern float   tsNumOfThreadsPerCore;
+extern int32_t tsNumOfCommitThreads;
+extern float   tsRatioOfQueryCores;
+extern int32_t tsCompressMsgSize;
+extern int32_t tsCompressColData;
+extern int32_t tsMaxNumOfDistinctResults;
+extern int32_t tsCompatibleModel;
+extern bool    tsEnableSlaveQuery;
+extern bool    tsPrintAuth;
+extern int64_t tsTickPerDay[3];
+extern int32_t tsMultiProcess;
+
+// monitor
+extern bool     tsEnableMonitor;
+extern int32_t  tsMonitorInterval;
+extern char     tsMonitorFqdn[];
+extern uint16_t tsMonitorPort;
+extern int32_t  tsMonitorMaxLogs;
+extern bool     tsMonitorComp;
 
 // query buffer management
 extern int32_t tsQueryBufferSize;  // maximum allowed usage buffer size in MB for each data node during query processing
 extern int64_t tsQueryBufferSizeBytes;   // maximum allowed usage buffer size in byte for each data node
-extern int32_t tsRetrieveBlockingModel;  // retrieve threads will be blocked
-extern int8_t  tsKeepOriginalColumnName;
-extern int8_t  tsDeadLockKillQuery;
+extern bool    tsRetrieveBlockingModel;  // retrieve threads will be blocked
+extern bool    tsKeepOriginalColumnName;
+extern bool    tsDeadLockKillQuery;
 
 // client
 extern int32_t tsMaxWildCardsLen;
 extern int32_t tsMaxRegexStringLen;
-extern int8_t  tsTscEnableRecordSql;
 extern int32_t tsMaxNumOfOrderedResults;
 extern int32_t tsMinSlidingTime;
 extern int32_t tsMinIntervalTime;
@@ -77,24 +81,10 @@ extern float   tsStreamComputDelayRatio;  // the delayed computing ration of the
 extern int32_t tsProjectExecInterval;
 extern int64_t tsMaxRetentWindow;
 
-// system info
-extern float    tsTotalLogDirGB;
-extern float    tsTotalTmpDirGB;
-extern float    tsTotalDataDirGB;
-extern float    tsAvailLogDirGB;
-extern float    tsAvailTmpDirectorySpace;
-extern float    tsAvailDataDirGB;
-extern float    tsUsedDataDirGB;
-extern float    tsMinimalLogDirGB;
-extern float    tsReservedTmpDirectorySpace;
-extern float    tsMinimalDataDirGB;
-extern uint32_t tsVersion;
-
 // build info
 extern char version[];
 extern char compatible_version[];
 extern char gitinfo[];
-extern char gitinfoOfInternal[];
 extern char buildinfo[];
 
 // lossy
@@ -105,18 +95,19 @@ extern uint32_t tsMaxRange;
 extern uint32_t tsCurRange;
 extern char     tsCompressor[];
 
+// tfs
 extern int32_t  tsDiskCfgNum;
 extern SDiskCfg tsDiskCfg[];
 
 #define NEEDTO_COMPRESSS_MSG(size) (tsCompressMsgSize != -1 && (size) > tsCompressMsgSize)
 
-void    taosInitGlobalCfg();
-int32_t taosCheckAndPrintCfg();
-int32_t taosCfgDynamicOptions(char *msg);
-bool    taosCheckBalanceCfgOptions(const char *option, int32_t *vnodeId, int32_t *dnodeId);
-void    taosAddDataDir(int index, char *v1, int level, int primary);
-void    taosReadDataDirCfg(char *v1, char *v2, char *v3);
-void    taosPrintDataDirCfg();
+int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDir, const char *envFile,
+                      const char *apolloUrl, SArray *pArgs, bool tsc);
+int32_t taosInitCfg(const char *cfgDir, const char *envFile, const char *apolloUrl, SArray *pArgs, bool tsc);
+void    taosCleanupCfg();
+void    taosCfgDynamicOptions(const char *option, const char *value);
+
+struct SConfig *taosGetCfg();
 
 #ifdef __cplusplus
 }

include/common/tlocale.h

-/*
- * 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/>.
- */
-
-#ifndef _TD_COMMON_LOCALE_H_
-#define _TD_COMMON_LOCALE_H_
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-void tsSetLocale();
-
-#ifdef __cplusplus
-}
-#endif /*_TD_COMMON_LOCALE_H_*/
-
-#endif

include/common/tmsgcb.h

+/*
+ * 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/>.
+ */
+
+#ifndef _TD_COMMON_MSG_CB_H_
+#define _TD_COMMON_MSG_CB_H_
+
+#include "os.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+typedef struct SRpcMsg      SRpcMsg;
+typedef struct SEpSet       SEpSet;
+typedef struct SMgmtWrapper SMgmtWrapper;
+typedef enum {
+  QUERY_QUEUE,
+  FETCH_QUEUE,
+  READ_QUEUE,
+  WRITE_QUEUE,
+  APPLY_QUEUE,
+  SYNC_QUEUE,
+  MERGE_QUEUE,
+  QUEUE_MAX,
+} EQueueType;
+
+typedef int32_t (*PutToQueueFp)(SMgmtWrapper* pWrapper, SRpcMsg* pReq);
+typedef int32_t (*GetQueueSizeFp)(SMgmtWrapper* pWrapper, int32_t vgId, EQueueType qtype);
+typedef int32_t (*SendReqFp)(SMgmtWrapper* pWrapper, SEpSet* epSet, SRpcMsg* pReq);
+typedef int32_t (*SendMnodeReqFp)(SMgmtWrapper* pWrapper, SRpcMsg* pReq);
+typedef void (*SendRspFp)(SMgmtWrapper* pWrapper, SRpcMsg* pRsp);
+
+typedef struct {
+  SMgmtWrapper*  pWrapper;
+  PutToQueueFp   queueFps[QUEUE_MAX];
+  GetQueueSizeFp qsizeFp;
+  SendReqFp      sendReqFp;
+  SendMnodeReqFp sendMnodeReqFp;
+  SendRspFp      sendRspFp;
+} SMsgCb;
+
+int32_t tmsgPutToQueue(const SMsgCb* pMsgCb, EQueueType qtype, SRpcMsg* pReq);
+int32_t tmsgGetQueueSize(const SMsgCb* pMsgCb, int32_t vgId, EQueueType qtype);
+int32_t tmsgSendReq(const SMsgCb* pMsgCb, SEpSet* epSet, SRpcMsg* pReq);
+int32_t tmsgSendMnodeReq(const SMsgCb* pMsgCb, SRpcMsg* pReq);
+void    tmsgSendRsp(const SMsgCb* pMsgCb, SRpcMsg* pRsp);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TD_COMMON_MSG_CB_H_*/

include/common/tmsgdef.h

@@ -47,7 +47,7 @@ enum {
 #define TD_NEW_MSG_SEG(TYPE) TYPE##_NUM,
 #define TD_DEF_MSG_TYPE(TYPE, MSG, REQ, RSP)
 
-int tMsgDict[] = {
+int32_t tMsgDict[] = {
 
 #elif defined(TD_MSG_SEG_CODE_)
 
@@ -127,26 +127,34 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_MND_CREATE_STB, "mnode-create-stb", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_ALTER_STB, "mnode-alter-stb", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_DROP_STB, "mnode-drop-stb", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_STB_META, "mnode-stb-meta", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_SMA, "mnode-create-sma", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_DROP_SMA, "mnode-drop-sma", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_TABLE_META, "mnode-table-meta", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_VGROUP_LIST, "mnode-vgroup-list", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_QNODE_LIST, "mnode-qnode-list", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_KILL_QUERY, "mnode-kill-query", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_KILL_CONN, "mnode-kill-conn", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_HEARTBEAT, "mnode-heartbeat", SClientHbBatchReq, SClientHbBatchRsp)
   TD_DEF_MSG_TYPE(TDMT_MND_SHOW, "mnode-show", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_SHOW_RETRIEVE, "mnode-retrieve", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_SYSTABLE_RETRIEVE, "mnode-systable-retrieve", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_STATUS, "mnode-status", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_TRANS, "mnode-trans", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_TRANS_TIMER, "mnode-trans-tmr", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_KILL_TRANS, "mnode-kill-trans", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_TELEM_TIMER, "mnode-telem-tmr", SMTimerReq, SMTimerReq)
   TD_DEF_MSG_TYPE(TDMT_MND_GRANT, "mnode-grant", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_AUTH, "mnode-auth", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_CREATE_TOPIC, "mnode-create-topic", SMCreateTopicReq, SMCreateTopicRsp)
   TD_DEF_MSG_TYPE(TDMT_MND_ALTER_TOPIC, "mnode-alter-topic", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_DROP_TOPIC, "mnode-drop-topic", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_SUBSCRIBE, "mnode-subscribe", SCMSubscribeReq, SCMSubscribeRsp)
-  TD_DEF_MSG_TYPE(TDMT_MND_RESET_OFFSET, "mnode-reset-offset", SMqCMResetOffsetReq, SMqCMResetOffsetRsp)
   TD_DEF_MSG_TYPE(TDMT_MND_GET_SUB_EP, "mnode-get-sub-ep", SMqCMGetSubEpReq, SMqCMGetSubEpRsp)
-  TD_DEF_MSG_TYPE(TDMT_MND_MQ_TIMER, "mnode-mq-timer", SMTimerReq, SMTimerReq)
+  TD_DEF_MSG_TYPE(TDMT_MND_MQ_TIMER, "mnode-mq-tmr", SMTimerReq, SMTimerReq)
   TD_DEF_MSG_TYPE(TDMT_MND_MQ_DO_REBALANCE, "mnode-mq-do-rebalance", SMqDoRebalanceMsg, SMqDoRebalanceMsg)
+  TD_DEF_MSG_TYPE(TDMT_MND_MQ_COMMIT_OFFSET, "mnode-mq-commit-offset", SMqCMCommitOffsetReq, SMqCMCommitOffsetRsp)
+  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_STREAM, "mnode-create-stream", SCMCreateStreamReq, SCMCreateStreamRsp)
+  TD_DEF_MSG_TYPE(TDMT_MND_ALTER_STREAM, "mnode-alter-stream", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_DROP_STREAM, "mnode-drop-stream", NULL, NULL)
 
   // Requests handled by VNODE
   TD_NEW_MSG_SEG(TDMT_VND_MSG)
@@ -179,19 +187,36 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_VND_SHOW_TABLES, "vnode-show-tables", SVShowTablesReq, SVShowTablesRsp)
   TD_DEF_MSG_TYPE(TDMT_VND_SHOW_TABLES_FETCH, "vnode-show-tables-fetch", SVShowTablesFetchReq, SVShowTablesFetchRsp)
   TD_DEF_MSG_TYPE(TDMT_VND_QUERY_CONTINUE, "vnode-query-continue", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_QUERY_HEARTBEAT, "vnode-query-heartbeat", NULL, NULL)
 
   TD_DEF_MSG_TYPE(TDMT_VND_SUBSCRIBE, "vnode-subscribe", SMVSubscribeReq, SMVSubscribeRsp)
   TD_DEF_MSG_TYPE(TDMT_VND_CONSUME, "vnode-consume", SMqCVConsumeReq, SMqCVConsumeRsp)
+  TD_DEF_MSG_TYPE(TDMT_VND_TASK_DEPLOY, "vnode-task-deploy", SStreamTaskDeployReq, SStreamTaskDeployRsp)
+  TD_DEF_MSG_TYPE(TDMT_VND_TASK_EXEC, "vnode-task-exec", SStreamTaskExecReq, SStreamTaskExecRsp)
+  TD_DEF_MSG_TYPE(TDMT_VND_TASK_PIPE_EXEC, "vnode-task-pipe-exec", SStreamTaskExecReq, SStreamTaskExecRsp)
+  TD_DEF_MSG_TYPE(TDMT_VND_TASK_MERGE_EXEC, "vnode-task-merge-exec", SStreamTaskExecReq, SStreamTaskExecRsp)
+  TD_DEF_MSG_TYPE(TDMT_VND_TASK_WRITE_EXEC, "vnode-task-write-exec", SStreamTaskExecReq, SStreamTaskExecRsp)
+  TD_DEF_MSG_TYPE(TDMT_VND_STREAM_TRIGGER, "vnode-stream-trigger", NULL, NULL)
 
+  TD_DEF_MSG_TYPE(TDMT_VND_CREATE_SMA, "vnode-create-sma", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_CANCEL_SMA, "vnode-cancel-sma", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_DROP_SMA, "vnode-drop-sma", NULL, NULL)
 
   // Requests handled by QNODE
   TD_NEW_MSG_SEG(TDMT_QND_MSG)
 
   // Requests handled by SNODE
   TD_NEW_MSG_SEG(TDMT_SND_MSG)
-
+  TD_DEF_MSG_TYPE(TDMT_SND_TASK_DEPLOY, "snode-task-deploy", SStreamTaskDeployReq, SStreamTaskDeployRsp)
+  TD_DEF_MSG_TYPE(TDMT_SND_TASK_EXEC, "snode-task-exec", SStreamTaskExecReq, SStreamTaskExecRsp)
+  TD_DEF_MSG_TYPE(TDMT_SND_TASK_PIPE_EXEC, "snode-task-pipe-exec", SStreamTaskExecReq, SStreamTaskExecRsp)
+  TD_DEF_MSG_TYPE(TDMT_SND_TASK_MERGE_EXEC, "snode-task-merge-exec", SStreamTaskExecReq, SStreamTaskExecRsp)
+
+  // Requests handled by SCHEDULER
+  TD_NEW_MSG_SEG(TDMT_SCH_MSG)
+  TD_DEF_MSG_TYPE(TDMT_SCH_LINK_BROKEN, "scheduler-link-broken", NULL, NULL)
+  
 #if defined(TD_MSG_NUMBER_)
   TDMT_MAX
 #endif
 };
-// clang-format on

include/common/tname.h

@@ -13,26 +13,24 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#ifndef TDENGINE_TNAME_H
-#define TDENGINE_TNAME_H
+#ifndef _TD_COMMON_NAME_H_
+#define _TD_COMMON_NAME_H_
+
+#include "tdef.h"
 
 #ifdef __cplusplus
 extern "C" {
 #endif
 
+#define TSDB_DB_NAME_T    1
+#define TSDB_TABLE_NAME_T 2
 
-#include "tdef.h"
-#include "tmsg.h"
-
-#define TSDB_DB_NAME_T     1
-#define TSDB_TABLE_NAME_T  2
-
-#define T_NAME_ACCT        0x1u
-#define T_NAME_DB          0x2u
-#define T_NAME_TABLE       0x4u
+#define T_NAME_ACCT  0x1u
+#define T_NAME_DB    0x2u
+#define T_NAME_TABLE 0x4u
 
 typedef struct SName {
-  uint8_t type;  //db_name_t, table_name_t
+  uint8_t type;  // db_name_t, table_name_t
   int32_t acctId;
   char    dbname[TSDB_DB_NAME_LEN];
   char    tname[TSDB_TABLE_NAME_LEN];
@@ -62,11 +60,11 @@ int32_t tNameFromString(SName* dst, const char* str, uint32_t type);
 
 int32_t tNameSetAcctId(SName* dst, int32_t acctId);
 
-SSchema createSchema(uint8_t type, int32_t bytes, int32_t colId, const char* name);
+bool tNameDBNameEqual(SName* left, SName* right);
+
 
 #ifdef __cplusplus
 }
 #endif
 
-
-#endif  // TDENGINE_TNAME_H
+#endif  /*_TD_COMMON_NAME_H_*/

include/common/trequest.h

@@ -13,8 +13,8 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#ifndef _TD_TREQUEST_H_
-#define _TD_TREQUEST_H_
+#ifndef _TD_COMMON_REQUEST_H_
+#define _TD_COMMON_REQUEST_H_
 
 #ifdef __cplusplus
 extern "C" {
@@ -36,7 +36,7 @@ void            tdClearRBIter(SReqBatchIter *pIter);
 
 /* ------------------------ TYPES DEFINITION ------------------------ */
 struct SReqBatchIter {
-  int        iReq;
+  int32_t    iReq;
   SReqBatch *pReqBatch;
 };
 
@@ -44,4 +44,4 @@ struct SReqBatchIter {
 }
 #endif
 
-#endif /*_TD_TREQUEST_H_*/
+#endif /*_TD_COMMON_REQUEST_H_*/
\ No newline at end of file

include/common/trow.h

@@ -18,11 +18,11 @@
 
 #include "os.h"
 #include "talgo.h"
+#include "taosdef.h"
 #include "taoserror.h"
 #include "tbuffer.h"
-#include "tdef.h"
-#include "taosdef.h"
 #include "tdataformat.h"
+#include "tdef.h"
 #include "tschema.h"
 #include "ttypes.h"
 #include "tutil.h"
@@ -47,21 +47,21 @@ extern "C" {
 #define TD_VTYPE_NONE 0x0U   // none or unknown/undefined
 #define TD_VTYPE_NULL 0x01U  // null val
 #define TD_VTYPE_NORM 0x02U  // normal val: not none, not null
-#define TD_VTYPE_MAX 0x03U   //
+#define TD_VTYPE_MAX  0x03U  //
 
 #define TD_VTYPE_NONE_BYTE 0x0U
 #define TD_VTYPE_NULL_BYTE 0x55U
 #define TD_VTYPE_NORM_BYTE 0xAAU
 
-#define TD_ROWS_ALL_NORM 0x01U
+#define TD_ROWS_ALL_NORM  0x01U
 #define TD_ROWS_NULL_NORM 0x0U
 
-#define TD_COL_ROWS_NORM(c) ((c)->bitmap == TD_ROWS_ALL_NORM)  // all rows of SDataCol/SBlockCol is NORM
+#define TD_COL_ROWS_NORM(c)          ((c)->bitmap == TD_ROWS_ALL_NORM)  // all rows of SDataCol/SBlockCol is NORM
 #define TD_SET_COL_ROWS_BTIMAP(c, v) ((c)->bitmap = (v))
-#define TD_SET_COL_ROWS_NORM(c) TD_SET_COL_ROWS_BTIMAP((c), TD_ROWS_ALL_NORM)
-#define TD_SET_COL_ROWS_MISC(c) TD_SET_COL_ROWS_BTIMAP((c), TD_ROWS_NULL_NORM)
+#define TD_SET_COL_ROWS_NORM(c)      TD_SET_COL_ROWS_BTIMAP((c), TD_ROWS_ALL_NORM)
+#define TD_SET_COL_ROWS_MISC(c)      TD_SET_COL_ROWS_BTIMAP((c), TD_ROWS_NULL_NORM)
 
-#define KvConvertRatio (0.9f)
+#define KvConvertRatio            (0.9f)
 #define isSelectKVRow(klen, tlen) ((klen) < ((tlen)*KvConvertRatio))
 
 #ifdef TD_SUPPORT_BITMAP
@@ -103,7 +103,8 @@ typedef struct {
 
 typedef struct {
   // TODO
-} STpRow;  // tuple
+  int tmp;  // TODO: to avoid compile error
+} STpRow;   // tuple
 
 #pragma pack(push, 1)
 typedef struct {
@@ -118,6 +119,8 @@ typedef struct {
 } SKvRow;
 
 typedef struct {
+  /// timestamp
+  TSKEY ts;
   union {
     /// union field for encode and decode
     uint32_t info;
@@ -138,8 +141,6 @@ typedef struct {
   uint32_t len;
   /// row version
   uint64_t ver;
-  /// timestamp
-  TSKEY ts;
   /// the inline data, maybe a tuple or a k-v tuple
   char data[];
 } STSRow;
@@ -162,36 +163,38 @@ typedef struct {
   int32_t extendedRowSize;
 } SRowBuilder;
 
-#define TD_ROW_HEAD_LEN (sizeof(STSRow))
+#define TD_ROW_HEAD_LEN  (sizeof(STSRow))
 #define TD_ROW_NCOLS_LEN (sizeof(col_id_t))
 
-#define TD_ROW_TYPE(r) ((r)->type)
-#define TD_ROW_DELETE(r) ((r)->del)
-#define TD_ROW_ENDIAN(r) ((r)->endian)
-#define TD_ROW_SVER(r) ((r)->sver)
-#define TD_ROW_NCOLS(r) ((r)->data)  // only valid for SKvRow
-#define TD_ROW_DATA(r) ((r)->data)
-#define TD_ROW_LEN(r) ((r)->len)
-#define TD_ROW_KEY(r) ((r)->ts)
-#define TD_ROW_KEY_ADDR(r) POINTER_SHIFT((r), 16)
+#define TD_ROW_INFO(r)     ((r)->info)
+#define TD_ROW_TYPE(r)     ((r)->type)
+#define TD_ROW_DELETE(r)   ((r)->del)
+#define TD_ROW_ENDIAN(r)   ((r)->endian)
+#define TD_ROW_SVER(r)     ((r)->sver)
+#define TD_ROW_NCOLS(r)    ((r)->data)  // only valid for SKvRow
+#define TD_ROW_DATA(r)     ((r)->data)
+#define TD_ROW_LEN(r)      ((r)->len)
+#define TD_ROW_KEY(r)      ((r)->ts)
+#define TD_ROW_KEY_ADDR(r) (r)
 
 // N.B. If without STSchema, getExtendedRowSize() is used to get the rowMaxBytes and
-// (int)ceil((double)nCols/TD_VTYPE_PARTS) should be added if TD_SUPPORT_BITMAP defined.
+// (int32_t)ceil((double)nCols/TD_VTYPE_PARTS) should be added if TD_SUPPORT_BITMAP defined.
 #define TD_ROW_MAX_BYTES_FROM_SCHEMA(s) (schemaTLen(s) + TD_ROW_HEAD_LEN)
 
-#define TD_ROW_SET_TYPE(r, t) (TD_ROW_TYPE(r) = (t))
-#define TD_ROW_SET_DELETE(r) (TD_ROW_DELETE(r) = 1)
-#define TD_ROW_SET_SVER(r, v) (TD_ROW_SVER(r) = (v))
-#define TD_ROW_SET_LEN(r, l) (TD_ROW_LEN(r) = (l))
+#define TD_ROW_SET_INFO(r, i)  (TD_ROW_INFO(r) = (i))
+#define TD_ROW_SET_TYPE(r, t)  (TD_ROW_TYPE(r) = (t))
+#define TD_ROW_SET_DELETE(r)   (TD_ROW_DELETE(r) = 1)
+#define TD_ROW_SET_SVER(r, v)  (TD_ROW_SVER(r) = (v))
+#define TD_ROW_SET_LEN(r, l)   (TD_ROW_LEN(r) = (l))
 #define TD_ROW_SET_NCOLS(r, n) (*(col_id_t *)TD_ROW_NCOLS(r) = (n))
 
 #define TD_ROW_IS_DELETED(r) (TD_ROW_DELETE(r) == 1)
-#define TD_IS_TP_ROW(r) (TD_ROW_TYPE(r) == TD_ROW_TP)
-#define TD_IS_KV_ROW(r) (TD_ROW_TYPE(r) == TD_ROW_KV)
-#define TD_IS_TP_ROW_T(t) ((t) == TD_ROW_TP)
-#define TD_IS_KV_ROW_T(t) ((t) == TD_ROW_KV)
+#define TD_IS_TP_ROW(r)      (TD_ROW_TYPE(r) == TD_ROW_TP)
+#define TD_IS_KV_ROW(r)      (TD_ROW_TYPE(r) == TD_ROW_KV)
+#define TD_IS_TP_ROW_T(t)    ((t) == TD_ROW_TP)
+#define TD_IS_KV_ROW_T(t)    ((t) == TD_ROW_KV)
 
-#define TD_BOOL_STR(b) ((b) ? "true" : "false")
+#define TD_BOOL_STR(b)       ((b) ? "true" : "false")
 #define isUtilizeKVRow(k, d) ((k) < ((d)*KVRatioConvert))
 
 #define TD_ROW_COL_IDX(r) POINTER_SHIFT(TD_ROW_DATA(r), sizeof(col_id_t))
@@ -217,7 +220,7 @@ static FORCE_INLINE void    tdRowCopy(void *dst, STSRow *row) { memcpy(dst, row,
 static FORCE_INLINE int32_t tdSetBitmapValType(void *pBitmap, int16_t colIdx, TDRowValT valType);
 int32_t                     tdSetBitmapValTypeN(void *pBitmap, int16_t nEle, TDRowValT valType);
 static FORCE_INLINE int32_t tdGetBitmapValType(void *pBitmap, int16_t colIdx, TDRowValT *pValType);
-int tdAppendValToDataCol(SDataCol *pCol, TDRowValT valType, const void *val, int numOfRows, int maxPoints);
+int32_t tdAppendValToDataCol(SDataCol *pCol, TDRowValT valType, const void *val, int32_t numOfRows, int32_t maxPoints);
 static FORCE_INLINE int32_t tdAppendColValToTpRow(SRowBuilder *pBuilder, TDRowValT valType, const void *val,
                                                   bool isCopyVarData, int8_t colType, int16_t colIdx, int32_t offset);
 static FORCE_INLINE int32_t tdAppendColValToKvRow(SRowBuilder *pBuilder, TDRowValT valType, const void *val,
@@ -355,7 +358,6 @@ static FORCE_INLINE int32_t tdGetBitmapValType(void *pBitmap, int16_t colIdx, TD
  *
  */
 
-
 /**
  * @brief
  *
@@ -432,7 +434,6 @@ static FORCE_INLINE int32_t tdSRowSetExtendedInfo(SRowBuilder *pBuilder, int32_t
       pBuilder->rowType = TD_ROW_TP;
     }
 
-
   } else {
     pBuilder->rowType = TD_ROW_TP;
   }
@@ -474,6 +475,7 @@ static int32_t tdSRowResetBuf(SRowBuilder *pBuilder, void *pBuf) {
     return terrno;
   }
 
+  TD_ROW_SET_INFO(pBuilder->pBuf, 0);
   TD_ROW_SET_TYPE(pBuilder->pBuf, pBuilder->rowType);
 
   uint32_t len = 0;
@@ -669,8 +671,9 @@ static FORCE_INLINE int32_t tdAppendColValToKvRow(SRowBuilder *pBuilder, TDRowVa
  * @param colIdx sorted column index, start from 0
  * @return FORCE_INLINE
  */
-static FORCE_INLINE int32_t tdAppendColValToRow(SRowBuilder *pBuilder, int16_t colId, int8_t colType, TDRowValT valType,
-                                                const void *val, bool isCopyVarData, int32_t offset, int16_t colIdx) {
+static FORCE_INLINE int32_t tdAppendColValToRow(SRowBuilder *pBuilder, col_id_t colId, int8_t colType,
+                                                TDRowValT valType, const void *val, bool isCopyVarData, int32_t offset,
+                                                col_id_t colIdx) {
   STSRow *pRow = pBuilder->pBuf;
   if (!val) {
 #ifdef TD_SUPPORT_BITMAP
@@ -698,9 +701,9 @@ static FORCE_INLINE int32_t tdAppendColValToRow(SRowBuilder *pBuilder, int16_t c
   }
   // TODO:  We can avoid the type judegement by FP, but would prevent the inline scheme.
   if (TD_IS_TP_ROW(pRow)) {
-    tdAppendColValToTpRow(pBuilder, valType, val, true, colType, colIdx, offset);
+    tdAppendColValToTpRow(pBuilder, valType, val, isCopyVarData, colType, colIdx, offset);
   } else {
-    tdAppendColValToKvRow(pBuilder, valType, val, true, colType, colIdx, offset, colId);
+    tdAppendColValToKvRow(pBuilder, valType, val, isCopyVarData, colType, colIdx, offset, colId);
   }
   return TSDB_CODE_SUCCESS;
 }
@@ -731,7 +734,7 @@ static FORCE_INLINE int32_t tdGetTpRowValOfCol(SCellVal *output, STSRow *pRow, v
   return TSDB_CODE_SUCCESS;
 }
 
-static FORCE_INLINE int compareKvRowColId(const void *key1, const void *key2) {
+static FORCE_INLINE int32_t compareKvRowColId(const void *key1, const void *key2) {
   if (*(int16_t *)key1 > ((SColIdx *)key2)->colId) {
     return 1;
   } else if (*(int16_t *)key1 < ((SColIdx *)key2)->colId) {
@@ -793,8 +796,8 @@ static FORCE_INLINE void tdSTSRowIterInit(STSRowIter *pIter, STSchema *pSchema)
   pIter->maxColId = pSchema->columns[pSchema->numOfCols - 1].colId;
 }
 
-static int tdCompareColId(const void *arg1, const void *arg2) {
-  int       colId = *(int *)arg1;
+static int32_t tdCompareColId(const void *arg1, const void *arg2) {
+  int32_t   colId = *(int32_t *)arg1;
   STColumn *pCol = (STColumn *)arg2;
 
   if (colId < pCol->colId) {
@@ -959,20 +962,24 @@ static FORCE_INLINE bool tdSTSRowIterNext(STSRowIter *pIter, col_id_t colId, col
 STSRow *mergeTwoRows(void *buffer, STSRow *row1, STSRow *row2, STSchema *pSchema1, STSchema *pSchema2);
 
 // Get the data pointer from a column-wised data
-static FORCE_INLINE int32_t tdGetColDataOfRow(SCellVal *pVal, SDataCol *pCol, int row) {
+static FORCE_INLINE int32_t tdGetColDataOfRow(SCellVal *pVal, SDataCol *pCol, int32_t row) {
   if (isAllRowsNone(pCol)) {
     pVal->valType = TD_VTYPE_NULL;
 #ifdef TD_SUPPORT_READ2
-    pVal->val = (void*)getNullValue(pCol->type);
+    pVal->val = (void *)getNullValue(pCol->type);
 #else
     pVal->val = NULL;
 #endif
     return TSDB_CODE_SUCCESS;
   }
-  if (tdGetBitmapValType(pCol->pBitmap, row, &(pVal->valType)) < 0) {
+
+  if (TD_COL_ROWS_NORM(pCol)) {
+    pVal->valType = TD_VTYPE_NORM;
+  } else if (tdGetBitmapValType(pCol->pBitmap, row, &(pVal->valType)) < 0) {
     return terrno;
   }
-  if (TD_COL_ROWS_NORM(pCol) || tdValTypeIsNorm(pVal->valType)) {
+
+  if (tdValTypeIsNorm(pVal->valType)) {
     if (IS_VAR_DATA_TYPE(pCol->type)) {
       pVal->val = POINTER_SHIFT(pCol->pData, pCol->dataOff[row]);
     } else {
@@ -981,7 +988,7 @@ static FORCE_INLINE int32_t tdGetColDataOfRow(SCellVal *pVal, SDataCol *pCol, in
   } else {
     pVal->valType = TD_VTYPE_NULL;
 #ifdef TD_SUPPORT_READ2
-    pVal->val = (void*)getNullValue(pCol->type);
+    pVal->val = (void *)getNullValue(pCol->type);
 #else
     pVal->val = NULL;
 #endif
@@ -1000,7 +1007,8 @@ static FORCE_INLINE bool tdSTpRowGetVal(STSRow *pRow, col_id_t colId, col_type_t
   return true;
 }
 
-static FORCE_INLINE bool tdSKvRowGetVal(STSRow *pRow, col_id_t colId, uint32_t offset, col_id_t colIdx, SCellVal *pVal) {
+static FORCE_INLINE bool tdSKvRowGetVal(STSRow *pRow, col_id_t colId, uint32_t offset, col_id_t colIdx,
+                                        SCellVal *pVal) {
   if (colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
     tdRowSetVal(pVal, TD_VTYPE_NORM, TD_ROW_KEY_ADDR(pRow));
     return true;
@@ -1010,7 +1018,7 @@ static FORCE_INLINE bool tdSKvRowGetVal(STSRow *pRow, col_id_t colId, uint32_t o
   return true;
 }
 
-static FORCE_INLINE int32_t dataColGetNEleLen(SDataCol *pDataCol, int rows) {
+static FORCE_INLINE int32_t dataColGetNEleLen(SDataCol *pDataCol, int32_t rows) {
   ASSERT(rows > 0);
   int32_t result = 0;
 
@@ -1079,15 +1087,15 @@ typedef struct {
 // STSRowBuilder
 #define trbInit(rt, allocator, endian, target, size) \
   { .type = (rt), .bw = tbufInitWriter(allocator, endian), .pRow = (target) }
-void trbSetRowInfo(STSRowBuilder *pRB, bool del, uint16_t sver);
-void trbSetRowVersion(STSRowBuilder *pRB, uint64_t ver);
-void trbSetRowTS(STSRowBuilder *pRB, TSKEY ts);
-int  trbWriteCol(STSRowBuilder *pRB, void *pData, col_id_t cid);
+void    trbSetRowInfo(STSRowBuilder *pRB, bool del, uint16_t sver);
+void    trbSetRowVersion(STSRowBuilder *pRB, uint64_t ver);
+void    trbSetRowTS(STSRowBuilder *pRB, TSKEY ts);
+int32_t trbWriteCol(STSRowBuilder *pRB, void *pData, col_id_t cid);
 
 // STSRowReader
 #define tRowReaderInit(schema, row) \
   { .schema = (schema), .row = (row) }
-int tRowReaderRead(STSRowReader *pRowReader, col_id_t cid, void *target, uint64_t size);
+int32_t tRowReaderRead(STSRowReader *pRowReader, col_id_t cid, void *target, uint64_t size);
 
 // STSRowBatchIter
 #define tRowBatchIterInit(pRB) \
@@ -1099,4 +1107,4 @@ const STSRow *tRowBatchIterNext(STSRowBatchIter *pRowBatchIter);
 }
 #endif
 
-#endif /*_TD_COMMON_ROW_H_*/
+#endif /*_TD_COMMON_ROW_H_*/
\ No newline at end of file

include/common/tschema.h

@@ -70,7 +70,7 @@ typedef struct {
   { .size = (capacity), .pSchema = (target) }
 void tSchemaBuilderSetSver(STShemaBuilder *pSchemaBuilder, uint16_t sver);
 void tSchemaBuilderSetSMA(bool sma, SArray *smaArray);
-int  tSchemaBuilderPutColumn(char *cname, bool sma, uint8_t type, col_id_t cid, uint32_t bytes, char *comment);
+int32_t  tSchemaBuilderPutColumn(char *cname, bool sma, uint8_t type, col_id_t cid, uint32_t bytes, char *comment);
 
 #endif
 

include/common/ttime.h

@@ -16,15 +16,26 @@
 #ifndef _TD_COMMON_TIME_H_
 #define _TD_COMMON_TIME_H_
 
+#include "taosdef.h"
+#include "tmsg.h"
+
 #ifdef __cplusplus
 extern "C" {
 #endif
 
-#include "taosdef.h"
-#include "tmsg.h"
-
 #define TIME_IS_VAR_DURATION(_t) ((_t) == 'n' || (_t) == 'y' || (_t) == 'N' || (_t) == 'Y')
 
+#define TIME_UNIT_NANOSECOND  'b'
+#define TIME_UNIT_MICROSECOND 'u'
+#define TIME_UNIT_MILLISECOND 'a'
+#define TIME_UNIT_SECOND      's'
+#define TIME_UNIT_MINUTE      'm'
+#define TIME_UNIT_HOUR        'h'
+#define TIME_UNIT_DAY         'd'
+#define TIME_UNIT_WEEK        'w'
+#define TIME_UNIT_MONTH       'n'
+#define TIME_UNIT_YEAR        'y'
+
 /*
  * @return timestamp decided by global conf variable, tsTimePrecision
  * if precision == TSDB_TIME_PRECISION_MICRO, it returns timestamp in microsecond.
@@ -35,7 +46,7 @@ static FORCE_INLINE int64_t taosGetTimestamp(int32_t precision) {
     return taosGetTimestampUs();
   } else if (precision == TSDB_TIME_PRECISION_NANO) {
     return taosGetTimestampNs();
-  }else {
+  } else {
     return taosGetTimestampMs();
   }
 }
@@ -52,8 +63,10 @@ void    deltaToUtcInitOnce();
 
 int64_t convertTimePrecision(int64_t time, int32_t fromPrecision, int32_t toPrecision);
 
+void taosFormatUtcTime(char *buf, int32_t bufLen, int64_t time, int32_t precision);
+
 #ifdef __cplusplus
 }
 #endif
 
-#endif  /*_TD_COMMON_TIME_H_*/
+#endif /*_TD_COMMON_TIME_H_*/

include/common/ttimezone.h

-/*
- * 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/>.
- */
-
-#ifndef _TD_COMMON_TIMEZONE_H_
-#define _TD_COMMON_TIMEZONE_H_
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-void tsSetTimeZone();
-
-#ifdef __cplusplus
-}
-#endif
-
-#endif /*_TD_COMMON_TIMEZONE_H_*/

include/common/ttszip.h

@@ -13,19 +13,19 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#ifndef TDENGINE_TTSZIP_H
-#define TDENGINE_TTSZIP_H
-
-#ifdef __cplusplus
-extern "C" {
-#endif
+#ifndef _TD_COMMON_TTSZIP_H_
+#define _TD_COMMON_TTSZIP_H_
 
 #include "os.h"
 #include "tdef.h"
 #include "tvariant.h"
 
-#define MEM_BUF_SIZE (1 << 20)
-#define TS_COMP_FILE_MAGIC 0x87F5EC4C
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define MEM_BUF_SIZE           (1 << 20)
+#define TS_COMP_FILE_MAGIC     0x87F5EC4C
 #define TS_COMP_FILE_GROUP_MAX 512
 
 typedef struct STSList {
@@ -61,10 +61,10 @@ typedef struct STSBlock {
  * and the offset of int32_t type is enough
  */
 typedef struct STSGroupBlockInfo {
-  int32_t id;          // group id
-  int32_t offset;      // offset set value in file
-  int32_t numOfBlocks; // number of total blocks
-  int32_t compLen;     // compressed size
+  int32_t id;           // group id
+  int32_t offset;       // offset set value in file
+  int32_t numOfBlocks;  // number of total blocks
+  int32_t compLen;      // compressed size
 } STSGroupBlockInfo;
 
 typedef struct STSGroupBlockInfoEx {
@@ -73,9 +73,9 @@ typedef struct STSGroupBlockInfoEx {
 } STSGroupBlockInfoEx;
 
 typedef struct STSBuf {
-  FILE*    f;
-  char     path[PATH_MAX];
-  uint32_t fileSize;
+  TdFilePtr pFile;
+  char      path[PATH_MAX];
+  uint32_t  fileSize;
 
   // todo use array
   STSGroupBlockInfoEx* pData;
@@ -112,9 +112,9 @@ STSBuf* tsBufClone(STSBuf* pTSBuf);
 
 STSGroupBlockInfo* tsBufGetGroupBlockInfo(STSBuf* pTSBuf, int32_t id);
 
-void    tsBufFlush(STSBuf* pTSBuf);
-void    tsBufResetPos(STSBuf* pTSBuf);
-bool    tsBufNextPos(STSBuf* pTSBuf);
+void tsBufFlush(STSBuf* pTSBuf);
+void tsBufResetPos(STSBuf* pTSBuf);
+bool tsBufNextPos(STSBuf* pTSBuf);
 
 STSElem tsBufGetElem(STSBuf* pTSBuf);
 STSElem tsBufGetElemStartPos(STSBuf* pTSBuf, int32_t id, SVariant* tag);
@@ -144,4 +144,4 @@ bool tsBufIsValidElem(STSElem* pElem);
 }
 #endif
 
-#endif  // TDENGINE_TTSZIP_H
+#endif  /*_TD_COMMON_TTSZIP_H_*/

include/common/ttypes.h

-#ifndef TDENGINE_TTYPE_H
-#define TDENGINE_TTYPE_H
+/*
+ * 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/>.
+ */
+
+#ifndef _TD_COMMON_TTYPE_H_
+#define _TD_COMMON_TTYPE_H_
 
 #ifdef __cplusplus
 extern "C" {
@@ -12,13 +27,9 @@ extern "C" {
 typedef int32_t  VarDataOffsetT;
 typedef uint32_t TDRowLenT;
 typedef uint8_t  TDRowValT;
-typedef uint16_t col_id_t;
+typedef int16_t  col_id_t;
 typedef int8_t   col_type_t;
-
-typedef struct tstr {
-  VarDataLenT len;
-  char        data[];
-} tstr;
+typedef int32_t  col_bytes_t;
 
 #pragma pack(push, 1)
 typedef struct {
@@ -32,19 +43,18 @@ typedef struct {
 } SNCharNullT;
 #pragma pack(pop)
 
-#define varDataTLen(v)      (sizeof(VarDataLenT) + varDataLen(v))
-#define varDataCopy(dst, v) memcpy((dst), (void*) (v), varDataTLen(v))
-#define varDataLenByData(v) (*(VarDataLenT *)(((char*)(v)) - VARSTR_HEADER_SIZE))
-#define varDataSetLen(v, _len) (((VarDataLenT *)(v))[0] = (VarDataLenT) (_len))
-#define IS_VAR_DATA_TYPE(t) (((t) == TSDB_DATA_TYPE_BINARY) || ((t) == TSDB_DATA_TYPE_NCHAR))
-
-#define varDataNetLen(v)       (htons(((VarDataLenT *)(v))[0]))
-#define varDataNetTLen(v)      (sizeof(VarDataLenT) + varDataNetLen(v))
+#define varDataTLen(v)         (sizeof(VarDataLenT) + varDataLen(v))
+#define varDataCopy(dst, v)    memcpy((dst), (void *)(v), varDataTLen(v))
+#define varDataLenByData(v)    (*(VarDataLenT *)(((char *)(v)) - VARSTR_HEADER_SIZE))
+#define varDataSetLen(v, _len) (((VarDataLenT *)(v))[0] = (VarDataLenT)(_len))
+#define IS_VAR_DATA_TYPE(t)    (((t) == TSDB_DATA_TYPE_VARCHAR) || ((t) == TSDB_DATA_TYPE_NCHAR))
 
+#define varDataNetLen(v)  (htons(((VarDataLenT *)(v))[0]))
+#define varDataNetTLen(v) (sizeof(VarDataLenT) + varDataNetLen(v))
 
 // this data type is internally used only in 'in' query to hold the values
-#define TSDB_DATA_TYPE_POINTER_ARRAY      (1000)
-#define TSDB_DATA_TYPE_VALUE_ARRAY      (1001)
+#define TSDB_DATA_TYPE_POINTER_ARRAY (1000)
+#define TSDB_DATA_TYPE_VALUE_ARRAY   (1001)
 
 #define GET_TYPED_DATA(_v, _finalType, _type, _data) \
   do {                                               \
@@ -62,7 +72,7 @@ typedef struct {
       case TSDB_DATA_TYPE_USMALLINT:                 \
         (_v) = (_finalType)GET_UINT16_VAL(_data);    \
         break;                                       \
-      case TSDB_DATA_TYPE_TIMESTAMP:\
+      case TSDB_DATA_TYPE_TIMESTAMP:                 \
       case TSDB_DATA_TYPE_BIGINT:                    \
         (_v) = (_finalType)(GET_INT64_VAL(_data));   \
         break;                                       \
@@ -91,6 +101,8 @@ typedef struct {
   do {                                         \
     switch (_type) {                           \
       case TSDB_DATA_TYPE_BOOL:                \
+        *(bool *)(_v) = (bool)(_data);         \
+        break;                                 \
       case TSDB_DATA_TYPE_TINYINT:             \
         *(int8_t *)(_v) = (int8_t)(_data);     \
         break;                                 \
@@ -104,6 +116,7 @@ typedef struct {
         *(uint16_t *)(_v) = (uint16_t)(_data); \
         break;                                 \
       case TSDB_DATA_TYPE_BIGINT:              \
+      case TSDB_DATA_TYPE_TIMESTAMP:           \
         *(int64_t *)(_v) = (int64_t)(_data);   \
         break;                                 \
       case TSDB_DATA_TYPE_UBIGINT:             \
@@ -132,20 +145,22 @@ typedef struct {
 
 #define IS_NUMERIC_TYPE(_t) ((IS_SIGNED_NUMERIC_TYPE(_t)) || (IS_UNSIGNED_NUMERIC_TYPE(_t)) || (IS_FLOAT_TYPE(_t)))
 
-#define IS_VALID_TINYINT(_t)    ((_t) > INT8_MIN  && (_t) <= INT8_MAX)
-#define IS_VALID_SMALLINT(_t)   ((_t) > INT16_MIN && (_t) <= INT16_MAX)
-#define IS_VALID_INT(_t)        ((_t) > INT32_MIN && (_t) <= INT32_MAX)
-#define IS_VALID_BIGINT(_t)     ((_t) > INT64_MIN && (_t) <= INT64_MAX)
-#define IS_VALID_UTINYINT(_t)   ((_t) >= 0 && (_t) < UINT8_MAX)
-#define IS_VALID_USMALLINT(_t)  ((_t) >= 0 && (_t) < UINT16_MAX)
-#define IS_VALID_UINT(_t)       ((_t) >= 0 && (_t) < UINT32_MAX)
-#define IS_VALID_UBIGINT(_t)    ((_t) >= 0 && (_t) < UINT64_MAX)
-#define IS_VALID_FLOAT(_t)      ((_t) >= -FLT_MAX && (_t) <= FLT_MAX)
-#define IS_VALID_DOUBLE(_t)     ((_t) >= -DBL_MAX && (_t) <= DBL_MAX)
-
-#define IS_CONVERT_AS_SIGNED(_t) (IS_SIGNED_NUMERIC_TYPE(_t) || (_t) == (TSDB_DATA_TYPE_BOOL) || (_t) == (TSDB_DATA_TYPE_TIMESTAMP))
+#define IS_VALID_TINYINT(_t)   ((_t) > INT8_MIN && (_t) <= INT8_MAX)
+#define IS_VALID_SMALLINT(_t)  ((_t) > INT16_MIN && (_t) <= INT16_MAX)
+#define IS_VALID_INT(_t)       ((_t) > INT32_MIN && (_t) <= INT32_MAX)
+#define IS_VALID_BIGINT(_t)    ((_t) > INT64_MIN && (_t) <= INT64_MAX)
+#define IS_VALID_UTINYINT(_t)  ((_t) >= 0 && (_t) < UINT8_MAX)
+#define IS_VALID_USMALLINT(_t) ((_t) >= 0 && (_t) < UINT16_MAX)
+#define IS_VALID_UINT(_t)      ((_t) >= 0 && (_t) < UINT32_MAX)
+#define IS_VALID_UBIGINT(_t)   ((_t) >= 0 && (_t) < UINT64_MAX)
+#define IS_VALID_FLOAT(_t)     ((_t) >= -FLT_MAX && (_t) <= FLT_MAX)
+#define IS_VALID_DOUBLE(_t)    ((_t) >= -DBL_MAX && (_t) <= DBL_MAX)
+
+#define IS_CONVERT_AS_SIGNED(_t) \
+  (IS_SIGNED_NUMERIC_TYPE(_t) || (_t) == (TSDB_DATA_TYPE_BOOL) || (_t) == (TSDB_DATA_TYPE_TIMESTAMP))
 #define IS_CONVERT_AS_UNSIGNED(_t) (IS_UNSIGNED_NUMERIC_TYPE(_t) || (_t) == (TSDB_DATA_TYPE_BOOL))
 
+// TODO remove this function
 static FORCE_INLINE bool isNull(const void *val, int32_t type) {
   switch (type) {
     case TSDB_DATA_TYPE_BOOL:
@@ -164,17 +179,17 @@ static FORCE_INLINE bool isNull(const void *val, int32_t type) {
     case TSDB_DATA_TYPE_DOUBLE:
       return *(uint64_t *)val == TSDB_DATA_DOUBLE_NULL;
     case TSDB_DATA_TYPE_NCHAR:
-      return varDataLen(val) == sizeof(int32_t) && *(uint32_t*) varDataVal(val) == TSDB_DATA_NCHAR_NULL;
+      return varDataLen(val) == sizeof(int32_t) && *(uint32_t *)varDataVal(val) == TSDB_DATA_NCHAR_NULL;
     case TSDB_DATA_TYPE_BINARY:
-      return varDataLen(val) == sizeof(int8_t) && *(uint8_t *) varDataVal(val) == TSDB_DATA_BINARY_NULL;
+      return varDataLen(val) == sizeof(int8_t) && *(uint8_t *)varDataVal(val) == TSDB_DATA_BINARY_NULL;
     case TSDB_DATA_TYPE_UTINYINT:
-      return *(uint8_t*) val == TSDB_DATA_UTINYINT_NULL;
+      return *(uint8_t *)val == TSDB_DATA_UTINYINT_NULL;
     case TSDB_DATA_TYPE_USMALLINT:
-      return *(uint16_t*) val == TSDB_DATA_USMALLINT_NULL;
+      return *(uint16_t *)val == TSDB_DATA_USMALLINT_NULL;
     case TSDB_DATA_TYPE_UINT:
-      return *(uint32_t*) val == TSDB_DATA_UINT_NULL;
+      return *(uint32_t *)val == TSDB_DATA_UINT_NULL;
     case TSDB_DATA_TYPE_UBIGINT:
-      return *(uint64_t*) val == TSDB_DATA_UBIGINT_NULL;
+      return *(uint64_t *)val == TSDB_DATA_UBIGINT_NULL;
 
     default:
       return false;
@@ -185,32 +200,31 @@ typedef struct tDataTypeDescriptor {
   int16_t type;
   int16_t nameLen;
   int32_t bytes;
-  char *  name;
+  char   *name;
   int64_t minValue;
   int64_t maxValue;
-  int (*compFunc)(const char *const input, int inputSize, const int nelements, char *const output, int outputSize,
-                  char algorithm, char *const buffer, int bufferSize);
-  int (*decompFunc)(const char *const input, int compressedSize, const int nelements, char *const output,
-                    int outputSize, char algorithm, char *const buffer, int bufferSize);
-  void (*statisFunc)(const void *pData, int32_t numofrow, int64_t *min, int64_t *max, int64_t *sum,
-                        int16_t *minindex, int16_t *maxindex, int16_t *numofnull);
+  int32_t (*compFunc)(const char *const input, int32_t inputSize, const int32_t nelements, char *const output,
+                      int32_t outputSize, char algorithm, char *const buffer, int32_t bufferSize);
+  int32_t (*decompFunc)(const char *const input, int32_t compressedSize, const int32_t nelements, char *const output,
+                        int32_t outputSize, char algorithm, char *const buffer, int32_t bufferSize);
+  void (*statisFunc)(const void *pData, int32_t numofrow, int64_t *min, int64_t *max, int64_t *sum, int16_t *minindex,
+                     int16_t *maxindex, int16_t *numofnull);
 } tDataTypeDescriptor;
 
 extern tDataTypeDescriptor tDataTypes[15];
 
 bool isValidDataType(int32_t type);
 
-void  setVardataNull(void* val, int32_t type);
-void  setNull(void *val, int32_t type, int32_t bytes);
-void  setNullN(void *val, int32_t type, int32_t bytes, int32_t numOfElems);
+void        setVardataNull(void *val, int32_t type);
+void        setNull(void *val, int32_t type, int32_t bytes);
+void        setNullN(void *val, int32_t type, int32_t bytes, int32_t numOfElems);
 const void *getNullValue(int32_t type);
 
-void assignVal(char *val, const char *src, int32_t len, int32_t type);
-void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size, void* buf);
-void operateVal(void *dst, void *s1, void *s2, int32_t optr, int32_t type);
-void* getDataMin(int32_t type);
-void* getDataMax(int32_t type);
-
+void  assignVal(char *val, const char *src, int32_t len, int32_t type);
+void  tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size, void *buf);
+void  operateVal(void *dst, void *s1, void *s2, int32_t optr, int32_t type);
+void *getDataMin(int32_t type);
+void *getDataMax(int32_t type);
 
 #define SET_DOUBLE_NULL(v) (*(uint64_t *)(v) = TSDB_DATA_DOUBLE_NULL)
 #define SET_BIGINT_NULL(v) (*(uint64_t *)(v) = TSDB_DATA_BIGINT_NULL)
@@ -219,4 +233,4 @@ void* getDataMax(int32_t type);
 }
 #endif
 
-#endif  // TDENGINE_TTYPE_H
+#endif  /*_TD_COMMON_TTYPE_H_*/

include/common/tvariant.h

@@ -13,34 +13,34 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#ifndef TDENGINE_TVARIANT_H
-#define TDENGINE_TVARIANT_H
+#ifndef _TD_COMMON_VARIANT_H_
+#define _TD_COMMON_VARIANT_H_
+
+#include "tarray.h"
 
 #ifdef __cplusplus
 extern "C" {
 #endif
 
-#include "tarray.h"
-
 // variant, each number/string/field_id has a corresponding struct during parsing sql
 typedef struct SVariant {
-  uint32_t   nType;
-  int32_t    nLen;  // only used for string, for number, it is useless
+  uint32_t nType;
+  int32_t  nLen;  // only used for string, for number, it is useless
   union {
     int64_t  i;
     uint64_t u;
     double   d;
     char    *pz;
-    wchar_t *wpz;
-    SArray  *arr; // only for 'in' query to hold value list, not value for a field
+    TdUcs4  *ucs4;
+    SArray  *arr;  // only for 'in' query to hold value list, not value for a field
   };
 } SVariant;
 
-int32_t toInteger(const char* z, int32_t n, int32_t base, int64_t* value, bool* issigned);
+int32_t toInteger(const char *z, int32_t n, int32_t base, int64_t *value, bool *issigned);
 
 bool taosVariantIsValid(SVariant *pVar);
 
-void taosVariantCreate(SVariant *pVar, const char* z, int32_t n, int32_t type);
+void taosVariantCreate(SVariant *pVar, const char *z, int32_t n, int32_t type);
 
 void taosVariantCreateFromBinary(SVariant *pVar, const char *pz, size_t len, uint32_t type);
 
@@ -48,7 +48,7 @@ void taosVariantDestroy(SVariant *pV);
 
 void taosVariantAssign(SVariant *pDst, const SVariant *pSrc);
 
-int32_t taosVariantCompare(const SVariant* p1, const SVariant* p2);
+int32_t taosVariantCompare(const SVariant *p1, const SVariant *p2);
 
 int32_t taosVariantToString(SVariant *pVar, char *dst);
 
@@ -64,4 +64,4 @@ int32_t taosVariantTypeSetType(SVariant *pVariant, char type);
 }
 #endif
 
-#endif  // TDENGINE_TVARIANT_H
+#endif  /*_TD_COMMON_VARIANT_H_*/