[td-225] merge develop branch

cmake/platform.inc

@@ -109,7 +109,7 @@ IF (TD_LINUX_64)
       IF (${CMAKE_CXX_COMPILER_ID} MATCHES "Clang")
         SET(COMMON_FLAGS "-std=gnu99 -Wall -Werror -Wno-missing-braces -fPIC -g3 -gdwarf-2 -msse4.2 -D_FILE_OFFSET_BITS=64 -D_LARGE_FILE")
       ELSE ()
-        SET(COMMON_FLAGS "-std=gnu99 -Wall -Werror -fPIC -malign-double -g3 -gdwarf-2 -malign-stringops -msse4.2 -D_FILE_OFFSET_BITS=64 -D_LARGE_FILE")
+        SET(COMMON_FLAGS "-std=gnu99 -Wall -pg -Werror -fPIC -malign-double -g3 -gdwarf-2 -malign-stringops -msse4.2 -D_FILE_OFFSET_BITS=64 -D_LARGE_FILE")
       ENDIF ()
     ELSE ()
       SET(COMMON_FLAGS "-std=gnu99 -Wall -Werror -fPIC -g -fsigned-char -fpack-struct=8 -D_FILE_OFFSET_BITS=64 -D_LARGE_FILE")

packaging/cfg/taos.cfg

@@ -51,18 +51,15 @@
 # number of threads per CPU core
 # numOfThreadsPerCore   1.0
 
-# number of vnodes per core in DNode
-# numOfVnodesPerCore    8
+# number of vgroups per db
+# maxVgroupsPerDb       0
+
+# max number of tables per vnode
+# maxTablesPerVnode     1000000
 
 # the ratio of threads responsible for querying in the total thread
 # ratioOfQueryThreads   0.5
 
-# number of total vnodes in DNode
-# numOfTotalVnodes      0
-
-# max number of tables per vnode
-# maxtablesPerVnode     1000
-
 # interval of check load balance when the management node is in normal operation
 # balanceInterval       300
 

src/client/src/tscAsync.c

@@ -198,6 +198,10 @@ void taos_fetch_rows_a(TAOS_RES *taosa, void (*fp)(void *, TAOS_RES *, int), voi
   SSqlRes *pRes = &pSql->res;
   SSqlCmd *pCmd = &pSql->cmd;
 
+  // user-defined callback function is stored in fetchFp
+  pSql->fetchFp = fp;
+  pSql->fp = tscAsyncFetchRowsProxy;
+
   if (pRes->qhandle == 0) {
     tscError("qhandle is NULL");
     pRes->code = TSDB_CODE_TSC_INVALID_QHANDLE;
@@ -205,10 +209,6 @@ void taos_fetch_rows_a(TAOS_RES *taosa, void (*fp)(void *, TAOS_RES *, int), voi
     return;
   }
 
-  // user-defined callback function is stored in fetchFp
-  pSql->fetchFp = fp;
-  pSql->fp = tscAsyncFetchRowsProxy;
-
   pSql->param = param;
   tscResetForNextRetrieve(pRes);
   

src/client/src/tscSQLParser.c

@@ -4962,6 +4962,7 @@ static void setCreateDBOption(SCMCreateDbMsg* pMsg, SCreateDBInfo* pCreateDb) {
   pMsg->commitTime = htonl(pCreateDb->commitTime);
   pMsg->minRowsPerFileBlock = htonl(pCreateDb->minRowsPerBlock);
   pMsg->maxRowsPerFileBlock = htonl(pCreateDb->maxRowsPerBlock);
+  pMsg->fsyncPeriod = htonl(pCreateDb->fsyncPeriod);
   pMsg->compression = pCreateDb->compressionLevel;
   pMsg->walLevel = (char)pCreateDb->walLevel;
   pMsg->replications = pCreateDb->replica;
@@ -5529,6 +5530,13 @@ int32_t tscCheckCreateDbParams(SSqlCmd* pCmd, SCMCreateDbMsg* pCreate) {
     return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg);
   }
 
+  val = htonl(pCreate->fsyncPeriod);
+  if (val != -1 && (val < TSDB_MIN_FSYNC_PERIOD || val > TSDB_MAX_FSYNC_PERIOD)) {
+    snprintf(msg, tListLen(msg), "invalid db option fsyncPeriod: %d valid range: [%d, %d]", val,
+             TSDB_MIN_FSYNC_PERIOD, TSDB_MAX_FSYNC_PERIOD);
+    return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg);
+  }
+
   if (pCreate->compression != -1 &&
       (pCreate->compression < TSDB_MIN_COMP_LEVEL || pCreate->compression > TSDB_MAX_COMP_LEVEL)) {
     snprintf(msg, tListLen(msg), "invalid db option compression: %d valid range: [%d, %d]", pCreate->compression,

src/common/inc/tglobal.h

@@ -68,7 +68,9 @@ extern int64_t tsMaxRetentWindow;
 // db parameters in client
 extern int32_t tsCacheBlockSize;
 extern int32_t tsBlocksPerVnode;
+extern int32_t tsMinTablePerVnode;
 extern int32_t tsMaxTablePerVnode;
+extern int32_t tsTableIncStepPerVnode;
 extern int32_t tsMaxVgroupsPerDb;
 extern int16_t tsDaysPerFile;
 extern int32_t tsDaysToKeep;
@@ -78,6 +80,7 @@ extern int16_t tsCommitTime;  // seconds
 extern int32_t tsTimePrecision;
 extern int16_t tsCompression;
 extern int16_t tsWAL;
+extern int32_t tsFsyncPeriod;
 extern int32_t tsReplications;
 
 // balance

src/common/src/tglobal.c

@@ -38,7 +38,7 @@ uint16_t tsDnodeShellPort = 6030;  // udp[6035-6039] tcp[6035]
 uint16_t tsDnodeDnodePort = 6035;  // udp/tcp
 uint16_t tsSyncPort = 6040;
 int32_t  tsStatusInterval = 1;  // second
-int16_t  tsNumOfVnodesPerCore = 8;
+int16_t  tsNumOfVnodesPerCore = 32;
 int16_t  tsNumOfTotalVnodes = TSDB_INVALID_VNODE_NUM;
 int32_t  tsNumOfMnodes = 3;
 int32_t  tsEnableVnodeBak = 1;
@@ -110,9 +110,13 @@ int16_t tsCommitTime    = TSDB_DEFAULT_COMMIT_TIME;  // seconds
 int32_t tsTimePrecision = TSDB_DEFAULT_PRECISION;
 int16_t tsCompression   = TSDB_DEFAULT_COMP_LEVEL;
 int16_t tsWAL           = TSDB_DEFAULT_WAL_LEVEL;
+int32_t tsFsyncPeriod   = TSDB_DEFAULT_FSYNC_PERIOD;
 int32_t tsReplications  = TSDB_DEFAULT_DB_REPLICA_OPTION;
 int32_t tsMaxVgroupsPerDb  = 0;
+int32_t tsMinTablePerVnode = 100;
 int32_t tsMaxTablePerVnode = TSDB_DEFAULT_TABLES;
+int32_t tsTableIncStepPerVnode = TSDB_TABLES_STEP;
+
 // balance
 int32_t tsEnableBalance = 1;
 int32_t tsAlternativeRole = 0;
@@ -389,16 +393,6 @@ static void doInitGlobalConfig() {
   cfg.unitType = TAOS_CFG_UTYPE_NONE;
   taosInitConfigOption(cfg);
 
-  cfg.option = "numOfVnodesPerCore";
-  cfg.ptr = &tsNumOfVnodesPerCore;
-  cfg.valType = TAOS_CFG_VTYPE_INT16;
-  cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW;
-  cfg.minValue = 1;
-  cfg.maxValue = 64;
-  cfg.ptrLength = 0;
-  cfg.unitType = TAOS_CFG_UTYPE_NONE;
-  taosInitConfigOption(cfg);
-
   cfg.option = "numOfTotalVnodes";
   cfg.ptr = &tsNumOfTotalVnodes;
   cfg.valType = TAOS_CFG_VTYPE_INT16;
@@ -622,6 +616,26 @@ static void doInitGlobalConfig() {
   cfg.unitType = TAOS_CFG_UTYPE_NONE;
   taosInitConfigOption(cfg);
 
+  cfg.option = "minTablesPerVnode";
+  cfg.ptr = &tsMinTablePerVnode;
+  cfg.valType = TAOS_CFG_VTYPE_INT32;
+  cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW;
+  cfg.minValue = TSDB_MIN_TABLES;
+  cfg.maxValue = TSDB_MAX_TABLES;
+  cfg.ptrLength = 0;
+  cfg.unitType = TAOS_CFG_UTYPE_NONE;
+  taosInitConfigOption(cfg);
+
+  cfg.option = "tableIncStepPerVnode";
+  cfg.ptr = &tsTableIncStepPerVnode;
+  cfg.valType = TAOS_CFG_VTYPE_INT32;
+  cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW;
+  cfg.minValue = TSDB_MIN_TABLES;
+  cfg.maxValue = TSDB_MAX_TABLES;
+  cfg.ptrLength = 0;
+  cfg.unitType = TAOS_CFG_UTYPE_NONE;
+  taosInitConfigOption(cfg);
+
   cfg.option = "cache";
   cfg.ptr = &tsCacheBlockSize;
   cfg.valType = TAOS_CFG_VTYPE_INT32;
@@ -702,6 +716,16 @@ static void doInitGlobalConfig() {
   cfg.unitType = TAOS_CFG_UTYPE_NONE;
   taosInitConfigOption(cfg);
 
+  cfg.option = "fsync";
+  cfg.ptr = &tsFsyncPeriod;
+  cfg.valType = TAOS_CFG_VTYPE_INT32;
+  cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW;
+  cfg.minValue = TSDB_MIN_FSYNC_PERIOD;
+  cfg.maxValue = TSDB_MAX_FSYNC_PERIOD;
+  cfg.ptrLength = 0;
+  cfg.unitType = TAOS_CFG_UTYPE_NONE;
+  taosInitConfigOption(cfg);
+
   cfg.option = "replica";
   cfg.ptr = &tsReplications;
   cfg.valType = TAOS_CFG_VTYPE_INT32;

src/dnode/src/dnodeMgmt.c

@@ -401,6 +401,7 @@ static int32_t dnodeProcessCreateVnodeMsg(SRpcMsg *rpcMsg) {
   pCreate->cfg.daysToKeep          = htonl(pCreate->cfg.daysToKeep);
   pCreate->cfg.minRowsPerFileBlock = htonl(pCreate->cfg.minRowsPerFileBlock);
   pCreate->cfg.maxRowsPerFileBlock = htonl(pCreate->cfg.maxRowsPerFileBlock);
+  pCreate->cfg.fsyncPeriod         = htonl(pCreate->cfg.fsyncPeriod);
   pCreate->cfg.commitTime          = htonl(pCreate->cfg.commitTime);
 
   for (int32_t j = 0; j < pCreate->cfg.replications; ++j) {

src/dnode/src/dnodePeer.c

@@ -60,7 +60,7 @@ int32_t dnodeInitServer() {
   rpcInit.label        = "DND-S";
   rpcInit.numOfThreads = 1;
   rpcInit.cfp          = dnodeProcessReqMsgFromDnode;
-  rpcInit.sessions     = 100;
+  rpcInit.sessions     = TSDB_MAX_VNODES;
   rpcInit.connType     = TAOS_CONN_SERVER;
   rpcInit.idleTime     = tsShellActivityTimer * 1000;
 
@@ -122,7 +122,7 @@ int32_t dnodeInitClient() {
   rpcInit.label        = "DND-C";
   rpcInit.numOfThreads = 1;
   rpcInit.cfp          = dnodeProcessRspFromDnode;
-  rpcInit.sessions     = 100;
+  rpcInit.sessions     = TSDB_MAX_VNODES;
   rpcInit.connType     = TAOS_CONN_CLIENT;
   rpcInit.idleTime     = tsShellActivityTimer * 1000;
   rpcInit.user         = "t";

src/dnode/src/dnodeSystem.c

@@ -70,7 +70,13 @@ int32_t main(int32_t argc, char *argv[]) {
     }
 #endif
 #ifdef TAOS_RANDOM_FILE_FAIL
-    else if (strcmp(argv[i], "--random-file-fail-factor") == 0) {
+    else if (strcmp(argv[i], "--random-file-fail-output") == 0) {
+      if ((i < argc - 1) && (argv[i + 1][0] != '-')) {
+        taosSetRandomFileFailOutput(argv[++i]);
+      } else {
+        taosSetRandomFileFailOutput(NULL);
+      }
+    } else if (strcmp(argv[i], "--random-file-fail-factor") == 0) {
       if ( (i+1) < argc ) {
         int factor = atoi(argv[i+1]);
         printf("The factor of random failure is %d\n", factor);

src/inc/taosdef.h

@@ -274,8 +274,8 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size);
 #define TSDB_DEFAULT_PAYLOAD_SIZE 5120   // default payload size, greater than PATH_MAX value
 #define TSDB_EXTRA_PAYLOAD_SIZE   128    // extra bytes for auth
 #define TSDB_CQ_SQL_SIZE          1024
-#define TSDB_MAX_VNODES           256
-#define TSDB_MIN_VNODES           50
+#define TSDB_MAX_VNODES           2048
+#define TSDB_MIN_VNODES           256
 #define TSDB_INVALID_VNODE_NUM    0
 
 #define TSDB_DNODE_ROLE_ANY       0
@@ -296,9 +296,9 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size);
 #define TSDB_DEFAULT_TOTAL_BLOCKS       4
 
 #define TSDB_MIN_TABLES                 4
-#define TSDB_MAX_TABLES                 5000000
-#define TSDB_DEFAULT_TABLES             200000
-#define TSDB_TABLES_STEP                10000
+#define TSDB_MAX_TABLES                 10000000
+#define TSDB_DEFAULT_TABLES             1000000
+#define TSDB_TABLES_STEP                1000
 
 #define TSDB_MIN_DAYS_PER_FILE          1
 #define TSDB_MAX_DAYS_PER_FILE          3650 
@@ -332,6 +332,10 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size);
 #define TSDB_MAX_WAL_LEVEL              2
 #define TSDB_DEFAULT_WAL_LEVEL          1
 
+#define TSDB_MIN_FSYNC_PERIOD           0
+#define TSDB_MAX_FSYNC_PERIOD           180000   // millisecond
+#define TSDB_DEFAULT_FSYNC_PERIOD       3000     // three second
+
 #define TSDB_MIN_DB_REPLICA_OPTION      1
 #define TSDB_MAX_DB_REPLICA_OPTION      3
 #define TSDB_DEFAULT_DB_REPLICA_OPTION  1

src/inc/taoserror.h

@@ -209,6 +209,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_QRY_NO_DISKSPACE,             0, 0x0702, "query no d
 TAOS_DEFINE_ERROR(TSDB_CODE_QRY_OUT_OF_MEMORY,            0, 0x0703, "query out of memory")
 TAOS_DEFINE_ERROR(TSDB_CODE_QRY_APP_ERROR,                0, 0x0704, "query app error")
 TAOS_DEFINE_ERROR(TSDB_CODE_QRY_DUP_JOIN_KEY,             0, 0x0705, "query duplicated join key")
+TAOS_DEFINE_ERROR(TSDB_CODE_QRY_EXCEED_TAGS_LIMIT,        0, 0x0706, "query tag conditon too many")
 
 // grant
 TAOS_DEFINE_ERROR(TSDB_CODE_GRANT_EXPIRED,                0, 0x0800, "grant expired")

src/inc/taosmsg.h

@@ -515,6 +515,7 @@ typedef struct {
   int32_t  minRowsPerFileBlock;
   int32_t  maxRowsPerFileBlock;
   int32_t  commitTime;
+  int32_t  fsyncPeriod;
   uint8_t  precision;   // time resolution
   int8_t   compression;
   int8_t   walLevel;
@@ -608,6 +609,7 @@ typedef struct {
   int32_t  minRowsPerFileBlock;
   int32_t  maxRowsPerFileBlock;
   int32_t  commitTime;
+  int32_t  fsyncPeriod;
   int8_t   precision;
   int8_t   compression;
   int8_t   walLevel;

src/inc/ttokendef.h

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

src/inc/twal.h

@@ -35,6 +35,7 @@ typedef struct {
 
 typedef struct {
   int8_t    walLevel;  // wal level
+  int32_t   fsyncPeriod; // millisecond
   int8_t    wals;      // number of WAL files;
   int8_t    keep;      // keep the wal file when closed
 } SWalCfg;

src/kit/taosmigrate/taosmigrateVnodeCfg.c

@@ -48,6 +48,7 @@ static int32_t saveVnodeCfg(SVnodeObj *pVnode, char* cfgFile)
   len += snprintf(content + len, maxLen - len, "  \"precision\": %d,\n", pVnode->tsdbCfg.precision);
   len += snprintf(content + len, maxLen - len, "  \"compression\": %d,\n", pVnode->tsdbCfg.compression);
   len += snprintf(content + len, maxLen - len, "  \"walLevel\": %d,\n", pVnode->walCfg.walLevel);
+  len += snprintf(content + len, maxLen - len, "  \"fsync\": %d,\n", pVnode->walCfg.fsyncPeriod);
   len += snprintf(content + len, maxLen - len, "  \"replica\": %d,\n", pVnode->syncCfg.replica);
   len += snprintf(content + len, maxLen - len, "  \"wals\": %d,\n", pVnode->walCfg.wals);
   len += snprintf(content + len, maxLen - len, "  \"quorum\": %d,\n", pVnode->syncCfg.quorum);
@@ -212,6 +213,13 @@ static int32_t readVnodeCfg(SVnodeObj *pVnode, char* cfgFile)
   }
   pVnode->walCfg.walLevel = (int8_t) walLevel->valueint;
 
+  cJSON *fsyncPeriod = cJSON_GetObjectItem(root, "fsync");
+  if (!fsyncPeriod || fsyncPeriod->type != cJSON_Number) {
+    printf("vgId:%d, failed to read vnode cfg, fsyncPeriod not found\n", pVnode->vgId);
+    goto PARSE_OVER;
+  }
+  pVnode->walCfg.fsyncPeriod = fsyncPeriod->valueint;
+
   cJSON *wals = cJSON_GetObjectItem(root, "wals");
   if (!wals || wals->type != cJSON_Number) {
     printf("vgId:%d, failed to read vnode cfg, wals not found\n", pVnode->vgId);

src/mnode/inc/mnodeDef.h

@@ -160,6 +160,7 @@ typedef struct {
   int32_t minRowsPerFileBlock;
   int32_t maxRowsPerFileBlock;
   int32_t commitTime;
+  int32_t fsyncPeriod;
   int8_t  precision;
   int8_t  compression;
   int8_t  walLevel;

src/mnode/src/mnodeDb.c

@@ -287,14 +287,14 @@ static int32_t mnodeCheckDbCfg(SDbCfg *pCfg) {
     return TSDB_CODE_MND_INVALID_DB_OPTION;
   }
 
-  if (pCfg->replications < TSDB_MIN_DB_REPLICA_OPTION || pCfg->replications > TSDB_MAX_DB_REPLICA_OPTION) {
-    mError("invalid db option replications:%d valid range: [%d, %d]", pCfg->replications, TSDB_MIN_DB_REPLICA_OPTION,
-           TSDB_MAX_DB_REPLICA_OPTION);
+  if (pCfg->fsyncPeriod < TSDB_MIN_FSYNC_PERIOD || pCfg->fsyncPeriod > TSDB_MAX_FSYNC_PERIOD) {
+    mError("invalid db option fsyncPeriod:%d, valid range: [%d, %d]", pCfg->fsyncPeriod, TSDB_MIN_FSYNC_PERIOD, TSDB_MAX_FSYNC_PERIOD);
     return TSDB_CODE_MND_INVALID_DB_OPTION;
   }
 
-  if (pCfg->walLevel < TSDB_MIN_WAL_LEVEL) {
-    mError("invalid db option walLevel:%d must be greater than 0", pCfg->walLevel);
+  if (pCfg->replications < TSDB_MIN_DB_REPLICA_OPTION || pCfg->replications > TSDB_MAX_DB_REPLICA_OPTION) {
+    mError("invalid db option replications:%d valid range: [%d, %d]", pCfg->replications, TSDB_MIN_DB_REPLICA_OPTION,
+           TSDB_MAX_DB_REPLICA_OPTION);
     return TSDB_CODE_MND_INVALID_DB_OPTION;
   }
 
@@ -318,6 +318,7 @@ static void mnodeSetDefaultDbCfg(SDbCfg *pCfg) {
   if (pCfg->daysToKeep2 < 0) pCfg->daysToKeep2 = pCfg->daysToKeep;
   if (pCfg->minRowsPerFileBlock < 0) pCfg->minRowsPerFileBlock = tsMinRowsInFileBlock;
   if (pCfg->maxRowsPerFileBlock < 0) pCfg->maxRowsPerFileBlock = tsMaxRowsInFileBlock;
+  if (pCfg->fsyncPeriod <0) pCfg->fsyncPeriod = tsFsyncPeriod;
   if (pCfg->commitTime < 0) pCfg->commitTime = tsCommitTime;
   if (pCfg->precision < 0) pCfg->precision = tsTimePrecision;
   if (pCfg->compression < 0) pCfg->compression = tsCompression;
@@ -367,6 +368,7 @@ static int32_t mnodeCreateDb(SAcctObj *pAcct, SCMCreateDbMsg *pCreate, void *pMs
     .daysToKeep2         = pCreate->daysToKeep2,
     .minRowsPerFileBlock = pCreate->minRowsPerFileBlock,
     .maxRowsPerFileBlock = pCreate->maxRowsPerFileBlock,
+    .fsyncPeriod         = pCreate->fsyncPeriod,
     .commitTime          = pCreate->commitTime,
     .precision           = pCreate->precision,
     .compression         = pCreate->compression,
@@ -559,6 +561,12 @@ static int32_t mnodeGetDbMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn
     pSchema[cols].bytes = htons(pShow->bytes[cols]);
     cols++;
 
+    pShow->bytes[cols] = 4;
+    pSchema[cols].type = TSDB_DATA_TYPE_INT;
+    strcpy(pSchema[cols].name, "fsync");
+    pSchema[cols].bytes = htons(pShow->bytes[cols]);
+    cols++;
+
     pShow->bytes[cols] = 1;
     pSchema[cols].type = TSDB_DATA_TYPE_TINYINT;
     strcpy(pSchema[cols].name, "comp");
@@ -682,6 +690,10 @@ static int32_t mnodeRetrieveDbs(SShowObj *pShow, char *data, int32_t rows, void
       *(int8_t *)pWrite = pDb->cfg.walLevel;
       cols++;
 
+      pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
+      *(int32_t *)pWrite = pDb->cfg.fsyncPeriod;
+      cols++;
+
       pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
       *(int8_t *)pWrite = pDb->cfg.compression;
       cols++;
@@ -758,6 +770,7 @@ static int32_t mnodeProcessCreateDbMsg(SMnodeMsg *pMsg) {
   pCreate->daysToKeep1     = htonl(pCreate->daysToKeep1);
   pCreate->daysToKeep2     = htonl(pCreate->daysToKeep2);
   pCreate->commitTime      = htonl(pCreate->commitTime);
+  pCreate->fsyncPeriod     = htonl(pCreate->fsyncPeriod);
   pCreate->minRowsPerFileBlock = htonl(pCreate->minRowsPerFileBlock);
   pCreate->maxRowsPerFileBlock = htonl(pCreate->maxRowsPerFileBlock);
   
@@ -785,6 +798,7 @@ static SDbCfg mnodeGetAlterDbOption(SDbObj *pDb, SCMAlterDbMsg *pAlter) {
   int32_t minRows        = htonl(pAlter->minRowsPerFileBlock);
   int32_t maxRows        = htonl(pAlter->maxRowsPerFileBlock);
   int32_t commitTime     = htonl(pAlter->commitTime);
+  int32_t fsyncPeriod    = htonl(pAlter->fsyncPeriod);
   int8_t  compression    = pAlter->compression;
   int8_t  walLevel       = pAlter->walLevel;
   int8_t  replications   = pAlter->replications;
@@ -861,6 +875,11 @@ static SDbCfg mnodeGetAlterDbOption(SDbObj *pDb, SCMAlterDbMsg *pAlter) {
     terrno = TSDB_CODE_MND_INVALID_DB_OPTION;
   }
 
+  if (fsyncPeriod >= 0 && fsyncPeriod != pDb->cfg.fsyncPeriod) {
+    mError("db:%s, can't alter fsyncPeriod option", pDb->name);
+    terrno = TSDB_CODE_MND_INVALID_DB_OPTION;
+  }
+
   if (replications > 0 && replications != pDb->cfg.replications) {
     mDebug("db:%s, replications:%d change to %d", pDb->name, pDb->cfg.replications, replications);
     newCfg.replications = replications;

src/mnode/src/mnodeProfile.c

@@ -115,7 +115,7 @@ SConnObj *mnodeAccquireConn(int32_t connId, char *user, uint32_t ip, uint16_t po
   uint64_t expireTime = CONN_KEEP_TIME * 1000 + (uint64_t)taosGetTimestampMs();
   SConnObj *pConn = taosCacheUpdateExpireTimeByName(tsMnodeConnCache, &connId, sizeof(int32_t), expireTime);
   if (pConn == NULL) {
-    mError("connId:%d, is already destroyed, user:%s ip:%s:%u", connId, user, taosIpStr(ip), port);
+    mDebug("connId:%d, is already destroyed, user:%s ip:%s:%u", connId, user, taosIpStr(ip), port);
     return NULL;
   }
 

src/mnode/src/mnodeSdb.c

@@ -170,7 +170,7 @@ static void *sdbGetTableFromId(int32_t tableId) {
 }
 
 static int32_t sdbInitWal() {
-  SWalCfg walCfg = {.walLevel = 2, .wals = 2, .keep = 1};
+  SWalCfg walCfg = {.walLevel = 2, .wals = 2, .keep = 1, .fsyncPeriod = 0};
   char temp[TSDB_FILENAME_LEN];
   sprintf(temp, "%s/wal", tsMnodeDir);
   tsSdbObj.wal = walOpen(temp, &walCfg);

src/mnode/src/mnodeTable.c

@@ -1740,11 +1740,13 @@ static int32_t mnodeProcessCreateChildTableMsg(SMnodeMsg *pMsg) {
         return code;
       }
 
-      if (pMsg->pVgroup == NULL) {
-        pMsg->pVgroup = pVgroup;
-        mnodeIncVgroupRef(pVgroup);
+      if (pMsg->pVgroup != NULL) {
+        mnodeDecVgroupRef(pMsg->pVgroup);
       }
 
+      pMsg->pVgroup = pVgroup;
+      mnodeIncVgroupRef(pVgroup);
+
       mDebug("app:%p:%p, table:%s, allocated in vgroup, vgId:%d sid:%d", pMsg->rpcMsg.ahandle, pMsg, pCreate->tableId,
              pVgroup->vgId, sid);
 

src/mnode/src/mnodeVgroup.c

@@ -323,7 +323,7 @@ static int32_t mnodeAllocVgroupIdPool(SVgObj *pInputVgroup) {
   if (pDb == NULL) return TSDB_CODE_MND_APP_ERROR;
 
   int32_t minIdPoolSize = TSDB_MAX_TABLES;
-  int32_t maxIdPoolSize = TSDB_MIN_TABLES;
+  int32_t maxIdPoolSize = tsMinTablePerVnode;
   for (int32_t v = 0; v < pDb->numOfVgroups; ++v) {
     SVgObj *pVgroup = pDb->vgList[v];
     if (pVgroup == NULL) continue;
@@ -347,10 +347,10 @@ static int32_t mnodeAllocVgroupIdPool(SVgObj *pInputVgroup) {
 
   // realloc all vgroups in db
   int32_t newIdPoolSize;
-  if (minIdPoolSize * 4 < TSDB_TABLES_STEP) {
+  if (minIdPoolSize * 4 < tsTableIncStepPerVnode) {
     newIdPoolSize = minIdPoolSize * 4;
   } else {
-    newIdPoolSize = ((minIdPoolSize / TSDB_TABLES_STEP) + 1) * TSDB_TABLES_STEP;
+    newIdPoolSize = ((minIdPoolSize / tsTableIncStepPerVnode) + 1) * tsTableIncStepPerVnode;
   }
 
   if (newIdPoolSize > tsMaxTablePerVnode) {
@@ -471,6 +471,8 @@ static int32_t mnodeCreateVgroupCb(SMnodeMsg *pMsg, int32_t code) {
   }
 
   pMsg->expected = pVgroup->numOfVnodes;
+  pMsg->successed = 0;
+  pMsg->received = 0;
   mnodeSendCreateVgroupMsg(pVgroup, pMsg);
 
   return TSDB_CODE_MND_ACTION_IN_PROGRESS;
@@ -755,6 +757,7 @@ SMDCreateVnodeMsg *mnodeBuildCreateVnodeMsg(SVgObj *pVgroup) {
   pCfg->daysToKeep2         = htonl(pDb->cfg.daysToKeep2);  
   pCfg->minRowsPerFileBlock = htonl(pDb->cfg.minRowsPerFileBlock);
   pCfg->maxRowsPerFileBlock = htonl(pDb->cfg.maxRowsPerFileBlock);
+  pCfg->fsyncPeriod         = htonl(pDb->cfg.fsyncPeriod);
   pCfg->commitTime          = htonl(pDb->cfg.commitTime);
   pCfg->precision           = pDb->cfg.precision;
   pCfg->compression         = pDb->cfg.compression;
@@ -836,6 +839,8 @@ static void mnodeProcessCreateVnodeRsp(SRpcMsg *rpcMsg) {
          pVgroup->vgId, tstrerror(rpcMsg->code), mnodeMsg->received, mnodeMsg->successed, mnodeMsg->expected,
          mnodeMsg->rpcMsg.handle, rpcMsg->ahandle);
 
+  assert(mnodeMsg->received <= mnodeMsg->expected);
+
   if (mnodeMsg->received != mnodeMsg->expected) return;
 
   if (mnodeMsg->received == mnodeMsg->successed) {

src/plugins/http/src/httpContext.c

@@ -137,7 +137,7 @@ void httpReleaseContext(HttpContext *pContext) {
   assert(refCount >= 0);
 
   HttpContext **ppContext = pContext->ppContext;
-  httpDebug("context:%p, is releasd, data:%p refCount:%d", pContext, ppContext, refCount);
+  httpDebug("context:%p, is released, data:%p refCount:%d", pContext, ppContext, refCount);
 
   if (tsHttpServer.contextCache != NULL) {
     taosCacheRelease(tsHttpServer.contextCache, (void **)(&ppContext), false);

src/plugins/http/src/httpSql.c

@@ -47,6 +47,10 @@ void httpProcessMultiSqlRetrieveCallBack(void *param, TAOS_RES *result, int numO
     }
   }
 
+  // if (tscResultsetFetchCompleted(result)) {
+  //   isContinue = false;
+  // }
+
   if (isContinue) {
     // retrieve next batch of rows
     httpDebug("context:%p, fd:%d, ip:%s, user:%s, process pos:%d, continue retrieve, numOfRows:%d, sql:%s",
@@ -75,7 +79,8 @@ void httpProcessMultiSqlCallBack(void *param, TAOS_RES *result, int code) {
   HttpContext *pContext = (HttpContext *)param;
   if (pContext == NULL) return;
 
-  HttpSqlCmds *     multiCmds = pContext->multiCmds;
+  code = taos_errno(result);  
+  HttpSqlCmds *multiCmds = pContext->multiCmds;
   HttpEncodeMethod *encode = pContext->encodeMethod;
 
   HttpSqlCmd *singleCmd = multiCmds->cmds + multiCmds->pos;
@@ -109,8 +114,8 @@ void httpProcessMultiSqlCallBack(void *param, TAOS_RES *result, int code) {
     return;
   }
 
-  int num_fields = taos_field_count(result);
-  if (num_fields == 0) {
+  bool isUpdate = tscIsUpdateQuery(result);
+  if (isUpdate) {
     // not select or show commands
     int affectRows = taos_affected_rows(result);
     httpDebug("context:%p, fd:%d, ip:%s, user:%s, process pos:%d, affect rows:%d, sql:%s",
@@ -221,9 +226,9 @@ void httpProcessSingleSqlRetrieveCallBack(void *param, TAOS_RES *result, int num
     if (numOfRows < 0) {
       httpError("context:%p, fd:%d, ip:%s, user:%s, retrieve failed, code:%s", pContext, pContext->fd, pContext->ipstr,
                 pContext->user, tstrerror(numOfRows));
-    } 
-    
-    taos_free_result(result);    
+    }
+
+    taos_free_result(result);
 
     if (encode->stopJsonFp) {
       (encode->stopJsonFp)(pContext, &pContext->singleCmd);
@@ -238,6 +243,7 @@ void httpProcessSingleSqlCallBack(void *param, TAOS_RES *result, int unUsedCode)
   if (pContext == NULL) return;
 
   int32_t code = taos_errno(result);
+
   HttpEncodeMethod *encode = pContext->encodeMethod;
 
   if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {

src/query/inc/qExecutor.h

@@ -121,7 +121,6 @@ typedef struct SQueryCostInfo {
   uint32_t loadBlockStatis;
   uint32_t discardBlocks;
   uint64_t elapsedTime;
-  uint64_t ioTime;
   uint64_t computTime;
 } SQueryCostInfo;
 
@@ -201,7 +200,7 @@ typedef struct SQInfo {
    */
   int32_t          tableIndex;
   int32_t          numOfGroupResultPages;
-  void*     pBuf; // allocated buffer for STableQueryInfo, sizeof(STableQueryInfo)*numOfTables;
+  void*            pBuf; // allocated buffer for STableQueryInfo, sizeof(STableQueryInfo)*numOfTables;
 
 } SQInfo;
 

src/query/inc/qUtil.h

@@ -23,7 +23,7 @@ void copyTimeWindowResBuf(SQueryRuntimeEnv* pRuntimeEnv, SWindowResult* dst, con
 int32_t initWindowResInfo(SWindowResInfo* pWindowResInfo, SQueryRuntimeEnv* pRuntimeEnv, int32_t size,
                           int32_t threshold, int16_t type);
 
-void    cleanupTimeWindowInfo(SWindowResInfo* pWindowResInfo, int32_t numOfCols);
+void    cleanupTimeWindowInfo(SWindowResInfo* pWindowResInfo);
 void    resetTimeWindowInfo(SQueryRuntimeEnv* pRuntimeEnv, SWindowResInfo* pWindowResInfo);
 void    clearFirstNTimeWindow(SQueryRuntimeEnv *pRuntimeEnv, int32_t num);
 
@@ -43,9 +43,7 @@ static FORCE_INLINE SWindowResult *getWindowResult(SWindowResInfo *pWindowResInf
 
 bool isWindowResClosed(SWindowResInfo *pWindowResInfo, int32_t slot);
 
-void createQueryResultInfo(SQuery *pQuery, SWindowResult *pResultRow, bool isSTableQuery, size_t interBufSize);
-
-//char *getPosInResultPage(SQueryRuntimeEnv *pRuntimeEnv, int32_t columnIndex, SWindowResult *pResult);
+int32_t createQueryResultInfo(SQuery *pQuery, SWindowResult *pResultRow, bool isSTableQuery, size_t interBufSize);
 
 static FORCE_INLINE char *getPosInResultPage(SQueryRuntimeEnv *pRuntimeEnv, int32_t columnIndex, SWindowResult *pResult) {
   assert(pResult != NULL && pRuntimeEnv != NULL);

src/query/inc/qsqlparser.h

@@ -116,6 +116,7 @@ typedef struct SCreateDBInfo {
   int32_t   daysPerFile;
   int32_t   minRowsPerBlock;
   int32_t   maxRowsPerBlock;
+  int32_t   fsyncPeriod;
   int64_t   commitTime;
   int32_t   walLevel;
   int32_t   compressionLevel;

src/query/inc/sql.y

@@ -221,6 +221,7 @@ maxrows(Y) ::= MAXROWS INTEGER(X).            { Y = X; }
 blocks(Y)  ::= BLOCKS INTEGER(X).             { Y = X; }
 ctime(Y)   ::= CTIME INTEGER(X).              { Y = X; }
 wal(Y)     ::= WAL INTEGER(X).                { Y = X; }
+fsync(Y)   ::= FSYNC INTEGER(X).              { Y = X; }
 comp(Y)    ::= COMP INTEGER(X).               { Y = X; }
 prec(Y)    ::= PRECISION STRING(X).           { Y = X; }
 
@@ -236,6 +237,7 @@ db_optr(Y) ::= db_optr(Z) maxrows(X).        { Y = Z; Y.maxRowsPerBlock = strtod
 db_optr(Y) ::= db_optr(Z) blocks(X).         { Y = Z; Y.numOfBlocks = strtol(X.z, NULL, 10); }
 db_optr(Y) ::= db_optr(Z) ctime(X).          { Y = Z; Y.commitTime = strtol(X.z, NULL, 10); }
 db_optr(Y) ::= db_optr(Z) wal(X).            { Y = Z; Y.walLevel = strtol(X.z, NULL, 10); }
+db_optr(Y) ::= db_optr(Z) fsync(X).          { Y = Z; Y.fsyncPeriod = strtol(X.z, NULL, 10); }
 db_optr(Y) ::= db_optr(Z) comp(X).           { Y = Z; Y.compressionLevel = strtol(X.z, NULL, 10); }
 db_optr(Y) ::= db_optr(Z) prec(X).           { Y = Z; Y.precision = X; }
 db_optr(Y) ::= db_optr(Z) keep(X).           { Y = Z; Y.keep = X; }
@@ -249,6 +251,7 @@ alter_db_optr(Y) ::= alter_db_optr(Z) keep(X).        { Y = Z; Y.keep = X; }
 alter_db_optr(Y) ::= alter_db_optr(Z) blocks(X).      { Y = Z; Y.numOfBlocks = strtol(X.z, NULL, 10); }
 alter_db_optr(Y) ::= alter_db_optr(Z) comp(X).        { Y = Z; Y.compressionLevel = strtol(X.z, NULL, 10); }
 alter_db_optr(Y) ::= alter_db_optr(Z) wal(X).         { Y = Z; Y.walLevel = strtol(X.z, NULL, 10); }
+alter_db_optr(Y) ::= alter_db_optr(Z) fsync(X).       { Y = Z; Y.fsyncPeriod = strtod(X.z, NULL, 10); }
 
 %type typename {TAOS_FIELD}
 typename(A) ::= ids(X). { 

src/query/src/qExecutor.c

@@ -144,7 +144,7 @@ static void setExecParams(SQuery *pQuery, SQLFunctionCtx *pCtx, void* inputData,
                           SDataStatis *pStatis, void *param, int32_t colIndex);
 
 static void initCtxOutputBuf(SQueryRuntimeEnv *pRuntimeEnv);
-static void destroyTableQueryInfo(STableQueryInfo *pTableQueryInfo, int32_t numOfCols);
+static void destroyTableQueryInfo(STableQueryInfo *pTableQueryInfo);
 static void resetCtxOutputBuf(SQueryRuntimeEnv *pRuntimeEnv);
 static bool hasMainOutput(SQuery *pQuery);
 static void buildTagQueryResult(SQInfo *pQInfo);
@@ -1583,7 +1583,7 @@ static void teardownQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv) {
   SQInfo* pQInfo = (SQInfo*) GET_QINFO_ADDR(pRuntimeEnv);
 
   qDebug("QInfo:%p teardown runtime env", pQInfo);
-  cleanupTimeWindowInfo(&pRuntimeEnv->windowResInfo, pQuery->numOfOutput);
+  cleanupTimeWindowInfo(&pRuntimeEnv->windowResInfo);
 
   if (pRuntimeEnv->pCtx != NULL) {
     for (int32_t i = 0; i < pQuery->numOfOutput; ++i) {
@@ -1921,17 +1921,6 @@ static FORCE_INLINE int32_t getNumOfRowsInResultPage(SQuery *pQuery, bool topBot
   return (DEFAULT_INTERN_BUF_PAGE_SIZE - sizeof(tFilePage)) / rowSize;
 }
 
-//char *getPosInResultPage(SQueryRuntimeEnv *pRuntimeEnv, int32_t columnIndex, SWindowResult *pResult) {
-//  assert(pResult != NULL && pRuntimeEnv != NULL);
-//
-//  SQuery    *pQuery = pRuntimeEnv->pQuery;
-//  tFilePage *page = GET_RES_BUF_PAGE_BY_ID(pRuntimeEnv->pResultBuf, pResult->pos.pageId);
-//  int32_t realRowId = pResult->pos.rowId * GET_ROW_PARAM_FOR_MULTIOUTPUT(pQuery, pRuntimeEnv->topBotQuery, pRuntimeEnv->stableQuery);
-//
-//  return ((char *)page->data) + pRuntimeEnv->offset[columnIndex] * pRuntimeEnv->numOfRowsPerPage +
-//         pQuery->pSelectExpr[columnIndex].bytes * realRowId;
-//}
-
 #define IS_PREFILTER_TYPE(_t) ((_t) != TSDB_DATA_TYPE_BINARY && (_t) != TSDB_DATA_TYPE_NCHAR)
 
 static bool needToLoadDataBlock(SQueryRuntimeEnv* pRuntimeEnv, SDataStatis *pDataStatis, SQLFunctionCtx *pCtx,
@@ -2543,11 +2532,7 @@ void UNUSED_FUNC displayInterResult(tFilePage **pdata, SQueryRuntimeEnv* pRuntim
       
       switch (pQuery->pSelectExpr[i].type) {
         case TSDB_DATA_TYPE_BINARY: {
-//          int32_t colIndex = pQuery->pSelectExpr[i].base.colInfo.colIndex;
           int32_t type = pQuery->pSelectExpr[i].type;
-//          } else {
-//            type = pMeterObj->schema[colIndex].type;
-//          }
           printBinaryData(pQuery->pSelectExpr[i].base.functionId, pdata[i]->data + pQuery->pSelectExpr[i].bytes * j,
                           type);
           break;
@@ -3016,17 +3001,22 @@ void switchCtxOrder(SQueryRuntimeEnv *pRuntimeEnv) {
   }
 }
 
-void createQueryResultInfo(SQuery *pQuery, SWindowResult *pResultRow, bool isSTableQuery, size_t interBufSize) {
+int32_t createQueryResultInfo(SQuery *pQuery, SWindowResult *pResultRow, bool isSTableQuery, size_t interBufSize) {
   int32_t numOfCols = pQuery->numOfOutput;
 
   size_t size = numOfCols * sizeof(SResultInfo) + interBufSize;
   pResultRow->resultInfo = calloc(1, size);
+  if (pResultRow->resultInfo == NULL) {
+    return TSDB_CODE_QRY_OUT_OF_MEMORY;
+  }
+
   pResultRow->pos = (SPosInfo) {-1, -1};
 
   char* buf = (char*) pResultRow->resultInfo + numOfCols * sizeof(SResultInfo);
 
   // set the intermediate result output buffer
   setWindowResultInfo(pResultRow->resultInfo, pQuery, isSTableQuery, buf);
+  return TSDB_CODE_SUCCESS;
 }
 
 void resetCtxOutputBuf(SQueryRuntimeEnv *pRuntimeEnv) {
@@ -3407,19 +3397,22 @@ static STableQueryInfo *createTableQueryInfo(SQueryRuntimeEnv *pRuntimeEnv, void
   if (QUERY_IS_INTERVAL_QUERY(pQuery) || pRuntimeEnv->groupbyNormalCol) {
     int32_t initialSize = 16;
     int32_t initialThreshold = 100;
-    initWindowResInfo(&pTableQueryInfo->windowResInfo, pRuntimeEnv, initialSize, initialThreshold, TSDB_DATA_TYPE_INT);
+    int32_t code = initWindowResInfo(&pTableQueryInfo->windowResInfo, pRuntimeEnv, initialSize, initialThreshold, TSDB_DATA_TYPE_INT);
+    if (code != TSDB_CODE_SUCCESS) {
+      return NULL;
+    }
   } else { // in other aggregate query, do not initialize the windowResInfo
   }
 
   return pTableQueryInfo;
 }
 
-void destroyTableQueryInfo(STableQueryInfo *pTableQueryInfo, int32_t numOfCols) {
+void destroyTableQueryInfo(STableQueryInfo *pTableQueryInfo) {
   if (pTableQueryInfo == NULL) {
     return;
   }
 
-  cleanupTimeWindowInfo(&pTableQueryInfo->windowResInfo, numOfCols);
+  cleanupTimeWindowInfo(&pTableQueryInfo->windowResInfo);
 }
 
 #define CHECK_QUERY_TIME_RANGE(_q, _tableInfo)                                              \
@@ -3878,45 +3871,11 @@ int32_t doFillGapsInResults(SQueryRuntimeEnv* pRuntimeEnv, tFilePage **pDst, int
 static void queryCostStatis(SQInfo *pQInfo) {
   SQueryRuntimeEnv *pRuntimeEnv = &pQInfo->runtimeEnv;
   SQueryCostInfo *pSummary = &pRuntimeEnv->summary;
-//  if (pRuntimeEnv->pResultBuf == NULL) {
-////    pSummary->tmpBufferInDisk = 0;
-//  } else {
-////    pSummary->tmpBufferInDisk = getResBufSize(pRuntimeEnv->pResultBuf);
-//  }
-//
-//  qDebug("QInfo:%p cost: comp blocks:%d, size:%d Bytes, elapsed time:%.2f ms", pQInfo, pSummary->readCompInfo,
-//         pSummary->totalCompInfoSize, pSummary->loadCompInfoUs / 1000.0);
-//
-//  qDebug("QInfo:%p cost: field info: %d, size:%d Bytes, avg size:%.2f Bytes, elapsed time:%.2f ms", pQInfo,
-//         pSummary->readField, pSummary->totalFieldSize, (double)pSummary->totalFieldSize / pSummary->readField,
-//         pSummary->loadFieldUs / 1000.0);
-//
-//  qDebug(
-//      "QInfo:%p cost: file blocks:%d, size:%d Bytes, elapsed time:%.2f ms, skipped:%d, in-memory gen null:%d Bytes",
-//      pQInfo, pSummary->readDiskBlocks, pSummary->totalBlockSize, pSummary->loadBlocksUs / 1000.0,
-//      pSummary->skippedFileBlocks, pSummary->totalGenData);
-  
-  qDebug("QInfo:%p :cost summary: elapsed time:%"PRId64" us, io time:%"PRId64" us, total blocks:%d, load block statis:%d,"
+
+  qDebug("QInfo:%p :cost summary: elapsed time:%"PRId64" us, total blocks:%d, load block statis:%d,"
          " load data block:%d, total rows:%"PRId64 ", check rows:%"PRId64,
-         pQInfo, pSummary->elapsedTime, pSummary->ioTime, pSummary->totalBlocks, pSummary->loadBlockStatis,
+         pQInfo, pSummary->elapsedTime, pSummary->totalBlocks, pSummary->loadBlockStatis,
          pSummary->loadBlocks, pSummary->totalRows, pSummary->totalCheckedRows);
-
-//  qDebug("QInfo:%p cost: temp file:%d Bytes", pQInfo, pSummary->tmpBufferInDisk);
-//
-//  qDebug("QInfo:%p cost: file:%d, table:%d", pQInfo, pSummary->numOfFiles, pSummary->numOfTables);
-//  qDebug("QInfo:%p cost: seek ops:%d", pQInfo, pSummary->numOfSeek);
-//
-//  double total = pSummary->fileTimeUs + pSummary->cacheTimeUs;
-//  double io = pSummary->loadCompInfoUs + pSummary->loadBlocksUs + pSummary->loadFieldUs;
-  
-//  double computing = total - io;
-//
-//  qDebug(
-//      "QInfo:%p cost: total elapsed time:%.2f ms, file:%.2f ms(%.2f%), cache:%.2f ms(%.2f%). io:%.2f ms(%.2f%),"
-//      "comput:%.2fms(%.2f%)",
-//      pQInfo, total / 1000.0, pSummary->fileTimeUs / 1000.0, pSummary->fileTimeUs * 100 / total,
-//      pSummary->cacheTimeUs / 1000.0, pSummary->cacheTimeUs * 100 / total, io / 1000.0, io * 100 / total,
-//      computing / 1000.0, computing * 100 / total);
 }
 
 static void updateOffsetVal(SQueryRuntimeEnv *pRuntimeEnv, SDataBlockInfo *pBlockInfo) {
@@ -4227,7 +4186,10 @@ int32_t doInitQInfo(SQInfo *pQInfo, STSBuf *pTsBuf, void *tsdb, int32_t vgId, bo
         type = TSDB_DATA_TYPE_INT;  // group id
       }
 
-      initWindowResInfo(&pRuntimeEnv->windowResInfo, pRuntimeEnv, 32, 4096, type);
+      code = initWindowResInfo(&pRuntimeEnv->windowResInfo, pRuntimeEnv, 32, 4096, type);
+      if (code != TSDB_CODE_SUCCESS) {
+        return code;
+      }
     }
 
   } else if (pRuntimeEnv->groupbyNormalCol || QUERY_IS_INTERVAL_QUERY(pQuery)) {
@@ -4244,7 +4206,10 @@ int32_t doInitQInfo(SQInfo *pQInfo, STSBuf *pTsBuf, void *tsdb, int32_t vgId, bo
       type = TSDB_DATA_TYPE_TIMESTAMP;
     }
 
-    initWindowResInfo(&pRuntimeEnv->windowResInfo, pRuntimeEnv, rows, 4096, type);
+    code = initWindowResInfo(&pRuntimeEnv->windowResInfo, pRuntimeEnv, rows, 4096, type);
+    if (code != TSDB_CODE_SUCCESS) {
+      return code;
+    }
   }
 
   if (pQuery->fillType != TSDB_FILL_NONE && !isPointInterpoQuery(pQuery)) {
@@ -4733,8 +4698,6 @@ static void doRestoreContext(SQInfo *pQInfo) {
 static void doCloseAllTimeWindowAfterScan(SQInfo *pQInfo) {
   SQuery *pQuery = pQInfo->runtimeEnv.pQuery;
 
-//  int32_t step = GET_FORWARD_DIRECTION_FACTOR(pQuery->order.order);
-
   if (QUERY_IS_INTERVAL_QUERY(pQuery)) {
     size_t numOfGroup = GET_NUM_OF_TABLEGROUP(pQInfo);
     for (int32_t i = 0; i < numOfGroup; ++i) {
@@ -4744,7 +4707,6 @@ static void doCloseAllTimeWindowAfterScan(SQInfo *pQInfo) {
       for (int32_t j = 0; j < num; ++j) {
         STableQueryInfo* item = taosArrayGetP(group, j);
         closeAllTimeWindow(&item->windowResInfo);
-//        removeRedundantWindow(&item->windowResInfo, item->lastKey - step, step);
       }
     }
   } else {  // close results for group result
@@ -5417,10 +5379,11 @@ static int32_t buildAirthmeticExprFromMsg(SExprInfo *pArithExprInfo, SQueryTable
   qDebug("qmsg:%p create arithmetic expr from binary string: %s", pQueryMsg, pArithExprInfo->base.arg[0].argValue.pz);
 
   tExprNode* pExprNode = NULL;
-  TRY(32) {
+  TRY(TSDB_MAX_TAGS) {
     pExprNode = exprTreeFromBinary(pArithExprInfo->base.arg[0].argValue.pz, pArithExprInfo->base.arg[0].argBytes);
   } CATCH( code ) {
     CLEANUP_EXECUTE();
+    qError("qmsg:%p failed to create arithmetic expression string from:%s, reason: %s", pQueryMsg, pArithExprInfo->base.arg[0].argValue.pz, tstrerror(code));
     return code;
   } END_TRY
 
@@ -5788,6 +5751,9 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SArray* pTableIdList,
 
     size_t s = taosArrayGetSize(pa);
     SArray* p1 = taosArrayInit(s, POINTER_BYTES);
+    if (p1 == NULL) {
+      goto _cleanup;
+    }
 
     for(int32_t j = 0; j < s; ++j) {
       void* pTable = taosArrayGetP(pa, j);
@@ -5802,6 +5768,9 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SArray* pTableIdList,
 
       void* buf = pQInfo->pBuf + index * sizeof(STableQueryInfo);
       STableQueryInfo* item = createTableQueryInfo(&pQInfo->runtimeEnv, pTable, window, buf);
+      if (item == NULL) {
+        goto _cleanup;
+      }
       item->groupIndex = i;
       taosArrayPush(p1, &item);
       taosHashPut(pQInfo->tableqinfoGroupInfo.map, &id->tid, sizeof(id->tid), &item, POINTER_BYTES);
@@ -5968,9 +5937,7 @@ static void freeQInfo(SQInfo *pQInfo) {
       size_t num = taosArrayGetSize(p);
       for(int32_t j = 0; j < num; ++j) {
         STableQueryInfo* item = taosArrayGetP(p, j);
-        if (item != NULL) {
-          destroyTableQueryInfo(item, pQuery->numOfOutput);
-        }
+        destroyTableQueryInfo(item);
       }
 
       taosArrayDestroy(p);
@@ -6155,6 +6122,7 @@ int32_t qCreateQueryInfo(void* tsdb, int32_t vgId, SQueryTableMsg* pQueryMsg, vo
       code = tsdbQuerySTableByTagCond(tsdb, id->uid, tagCond, pQueryMsg->tagCondLen, pQueryMsg->tagNameRelType, tbnameCond, &tableGroupInfo, pGroupColIndex,
                                           numOfGroupByCols);
       if (code != TSDB_CODE_SUCCESS) {
+        qError("qmsg:%p failed to QueryStable, reason: %s", pQueryMsg, tstrerror(code));
         goto _over;
       }
     } else {
@@ -6314,7 +6282,6 @@ bool qHasMoreResultsToRetrieve(qinfo_t qinfo) {
   }
 
   if (ret) {
-//    T_REF_INC(pQInfo);
     qDebug("QInfo:%p has more results waits for client retrieve", pQInfo);
   }
 
@@ -6370,6 +6337,7 @@ int32_t qKillQuery(qinfo_t qinfo) {
     return TSDB_CODE_QRY_INVALID_QHANDLE;
   }
 
+  sem_post(&pQInfo->dataReady);
   setQueryKilled(pQInfo);
   return TSDB_CODE_SUCCESS;
 }
@@ -6582,13 +6550,14 @@ void** qRegisterQInfo(void* pMgmt, uint64_t qInfo) {
 
   SQueryMgmt *pQueryMgmt = pMgmt;
   if (pQueryMgmt->qinfoPool == NULL) {
+    qError("QInfo:%p failed to add qhandle into qMgmt, since qMgmt is closed", (void *)qInfo);
     return NULL;
   }
 
   pthread_mutex_lock(&pQueryMgmt->lock);
   if (pQueryMgmt->closed) {
     pthread_mutex_unlock(&pQueryMgmt->lock);
-
+    qError("QInfo:%p failed to add qhandle into cache, since qMgmt is colsing", (void *)qInfo);
     return NULL;
   } else {
     uint64_t handleVal = (uint64_t) qInfo;

src/query/src/qUtil.c

@@ -50,8 +50,15 @@ int32_t initWindowResInfo(SWindowResInfo *pWindowResInfo, SQueryRuntimeEnv *pRun
   
   // use the pointer arraylist
   pWindowResInfo->pResult = calloc(threshold, sizeof(SWindowResult));
+  if (pWindowResInfo->pResult == NULL) {
+    return TSDB_CODE_QRY_OUT_OF_MEMORY;
+  }
+
   for (int32_t i = 0; i < pWindowResInfo->capacity; ++i) {
-    createQueryResultInfo(pRuntimeEnv->pQuery, &pWindowResInfo->pResult[i], pRuntimeEnv->stableQuery, pRuntimeEnv->interBufSize);
+    int32_t code = createQueryResultInfo(pRuntimeEnv->pQuery, &pWindowResInfo->pResult[i], pRuntimeEnv->stableQuery, pRuntimeEnv->interBufSize);
+    if (code != TSDB_CODE_SUCCESS) {
+      return code;
+    }
   }
   
   return TSDB_CODE_SUCCESS;
@@ -65,7 +72,7 @@ void destroyTimeWindowRes(SWindowResult *pWindowRes) {
   free(pWindowRes->resultInfo);
 }
 
-void cleanupTimeWindowInfo(SWindowResInfo *pWindowResInfo, int32_t numOfCols) {
+void cleanupTimeWindowInfo(SWindowResInfo *pWindowResInfo) {
   if (pWindowResInfo == NULL) {
     return;
   }
@@ -74,9 +81,10 @@ void cleanupTimeWindowInfo(SWindowResInfo *pWindowResInfo, int32_t numOfCols) {
     return;
   }
   
-  for (int32_t i = 0; i < pWindowResInfo->capacity; ++i) {
-    SWindowResult *pResult = &pWindowResInfo->pResult[i];
-    destroyTimeWindowRes(pResult);
+  if (pWindowResInfo->pResult != NULL) {
+    for (int32_t i = 0; i < pWindowResInfo->capacity; ++i) {
+      destroyTimeWindowRes(&pWindowResInfo->pResult[i]);
+    }
   }
   
   taosHashCleanup(pWindowResInfo->hashList);

src/query/src/qast.c

@@ -962,10 +962,13 @@ static UNUSED_FUNC char* exception_strdup(const char* str) {
 
 static tExprNode* exprTreeFromBinaryImpl(SBufferReader* br) {
   int32_t anchor = CLEANUP_GET_ANCHOR();
+  if (CLEANUP_EXCEED_LIMIT()) {
+    THROW(TSDB_CODE_QRY_EXCEED_TAGS_LIMIT);
+    return NULL;
+  }
 
   tExprNode* pExpr = exception_calloc(1, sizeof(tExprNode));
   CLEANUP_PUSH_VOID_PTR_PTR(true, tExprNodeDestroy, pExpr, NULL);
-
   pExpr->nodeType = tbufReadUint8(br);
   
   if (pExpr->nodeType == TSQL_NODE_VALUE) {
@@ -995,7 +998,6 @@ static tExprNode* exprTreeFromBinaryImpl(SBufferReader* br) {
     pExpr->_node.hasPK = tbufReadUint8(br);
     pExpr->_node.pLeft = exprTreeFromBinaryImpl(br);
     pExpr->_node.pRight = exprTreeFromBinaryImpl(br);
-    
     assert(pExpr->_node.pLeft != NULL && pExpr->_node.pRight != NULL);
   }
   

src/query/src/qparserImpl.c

@@ -896,6 +896,7 @@ void setDefaultCreateDbOption(SCreateDBInfo *pDBInfo) {
   pDBInfo->compressionLevel = -1;
 
   pDBInfo->walLevel = -1;
+  pDBInfo->fsyncPeriod = -1;
   pDBInfo->commitTime = -1;
   pDBInfo->maxTablesPerVnode = -1;
 

src/query/src/qresultBuf.c

@@ -26,27 +26,24 @@ int32_t createDiskbasedResultBuffer(SDiskbasedResultBuf** pResultBuf, int32_t si
   pResBuf->path = strdup(path);
 
   pResBuf->fd = open(pResBuf->path, O_CREAT | O_RDWR, 0666);
-  
-  memset(path, 0, tListLen(path));
-
   if (!FD_VALID(pResBuf->fd)) {
     qError("failed to create tmp file: %s on disk. %s", pResBuf->path, strerror(errno));
-    return TSDB_CODE_QRY_NO_DISKSPACE;
+    return TAOS_SYSTEM_ERROR(errno);
   }
 
   int32_t ret = ftruncate(pResBuf->fd, pResBuf->numOfPages * DEFAULT_INTERN_BUF_PAGE_SIZE);
   if (ret != TSDB_CODE_SUCCESS) {
     qError("failed to create tmp file: %s on disk. %s", pResBuf->path, strerror(errno));
-    return TSDB_CODE_QRY_NO_DISKSPACE;
+    return TAOS_SYSTEM_ERROR(errno);
   }
 
   pResBuf->pBuf = mmap(NULL, pResBuf->totalBufSize, PROT_READ | PROT_WRITE, MAP_SHARED, pResBuf->fd, 0);
   if (pResBuf->pBuf == MAP_FAILED) {
     qError("QInfo:%p failed to map temp file: %s. %s", handle, pResBuf->path, strerror(errno));
-    return TSDB_CODE_QRY_OUT_OF_MEMORY; // todo change error code
+    return TAOS_SYSTEM_ERROR(errno);
   }
 
-  qDebug("QInfo:%p create tmp file for output result: %s, %" PRId64 "bytes", handle, pResBuf->path,
+  qDebug("QInfo:%p create tmp file for output result:%s, %" PRId64 "bytes", handle, pResBuf->path,
       pResBuf->totalBufSize);
   
   return TSDB_CODE_SUCCESS;

src/query/src/qtokenizer.c

@@ -124,6 +124,7 @@ static SKeyword keywordTable[] = {
     {"CACHE",        TK_CACHE},
     {"CTIME",        TK_CTIME},
     {"WAL",          TK_WAL},
+    {"FSYNC",        TK_FSYNC},
     {"COMP",         TK_COMP},
     {"PRECISION",    TK_PRECISION},
     {"LP",           TK_LP},

src/tsdb/inc/tsdbMain.h

@@ -96,6 +96,11 @@ typedef struct {
 } STsdbBufPool;
 
 // ------------------ tsdbMemTable.c
+typedef struct {
+  STable *           pTable;
+  SSkipListIterator *pIter;
+} SCommitIter;
+
 typedef struct {
   uint64_t   uid;
   TSKEY      keyFirst;
@@ -206,10 +211,10 @@ typedef struct {
   int64_t offset : 63;
   int32_t algorithm : 8;
   int32_t numOfRows : 24;
-  int32_t sversion;
   int32_t len;
+  int32_t keyLen;     // key column length, keyOffset = offset+sizeof(SCompData)+sizeof(SCompCol)*numOfCols
   int16_t numOfSubBlocks;
-  int16_t numOfCols;
+  int16_t numOfCols; // not including timestamp column
   TSKEY   keyFirst;
   TSKEY   keyLast;
 } SCompBlock;
@@ -377,6 +382,24 @@ int   tsdbUnRefMemTable(STsdbRepo* pRepo, SMemTable* pMemTable);
 int   tsdbTakeMemSnapshot(STsdbRepo* pRepo, SMemTable** pMem, SMemTable** pIMem);
 void* tsdbAllocBytes(STsdbRepo* pRepo, int bytes);
 int   tsdbAsyncCommit(STsdbRepo* pRepo);
+int   tsdbLoadDataFromCache(STable* pTable, SSkipListIterator* pIter, TSKEY maxKey, int maxRowsToRead, SDataCols* pCols,
+                            TSKEY* filterKeys, int nFilterKeys);
+
+static FORCE_INLINE SDataRow tsdbNextIterRow(SSkipListIterator* pIter) {
+  if (pIter == NULL) return NULL;
+
+  SSkipListNode* node = tSkipListIterGet(pIter);
+  if (node == NULL) return NULL;
+
+  return SL_GET_NODE_DATA(node);
+}
+
+static FORCE_INLINE TSKEY tsdbNextIterKey(SSkipListIterator* pIter) {
+  SDataRow row = tsdbNextIterRow(pIter);
+  if (row == NULL) return -1;
+
+  return dataRowKey(row);
+}
 
 // ------------------ tsdbFile.c
 #define TSDB_KEY_FILEID(key, daysPerFile, precision) ((key) / tsMsPerDay[(precision)] / (daysPerFile))
@@ -421,25 +444,36 @@ void        tsdbRemoveFileGroup(STsdbRepo* pRepo, SFileGroup* pFGroup);
 #define helperType(h) (h)->type
 #define helperRepo(h) (h)->pRepo
 #define helperState(h) (h)->state
-
-int   tsdbInitReadHelper(SRWHelper* pHelper, STsdbRepo* pRepo);
-int   tsdbInitWriteHelper(SRWHelper* pHelper, STsdbRepo* pRepo);
-void  tsdbDestroyHelper(SRWHelper* pHelper);
-void  tsdbResetHelper(SRWHelper* pHelper);
-int   tsdbSetAndOpenHelperFile(SRWHelper* pHelper, SFileGroup* pGroup);
-int   tsdbCloseHelperFile(SRWHelper* pHelper, bool hasError);
-void  tsdbSetHelperTable(SRWHelper* pHelper, STable* pTable, STsdbRepo* pRepo);
-int   tsdbWriteDataBlock(SRWHelper* pHelper, SDataCols* pDataCols);
-int   tsdbMoveLastBlockIfNeccessary(SRWHelper* pHelper);
-int   tsdbWriteCompInfo(SRWHelper* pHelper);
-int   tsdbWriteCompIdx(SRWHelper* pHelper);
-int   tsdbLoadCompIdx(SRWHelper* pHelper, void* target);
-int   tsdbLoadCompInfo(SRWHelper* pHelper, void* target);
-int   tsdbLoadCompData(SRWHelper* phelper, SCompBlock* pcompblock, void* target);
-void  tsdbGetDataStatis(SRWHelper* pHelper, SDataStatis* pStatis, int numOfCols);
-int   tsdbLoadBlockDataCols(SRWHelper* pHelper, SCompBlock* pCompBlock, SCompInfo* pCompInfo, int16_t* colIds,
-                            int numOfColIds);
-int   tsdbLoadBlockData(SRWHelper* pHelper, SCompBlock* pCompBlock, SCompInfo* pCompInfo);
+#define TSDB_NLAST_FILE_OPENED(h) ((h)->files.nLastF.fd > 0)
+
+int  tsdbInitReadHelper(SRWHelper* pHelper, STsdbRepo* pRepo);
+int  tsdbInitWriteHelper(SRWHelper* pHelper, STsdbRepo* pRepo);
+void tsdbDestroyHelper(SRWHelper* pHelper);
+void tsdbResetHelper(SRWHelper* pHelper);
+int  tsdbSetAndOpenHelperFile(SRWHelper* pHelper, SFileGroup* pGroup);
+int  tsdbCloseHelperFile(SRWHelper* pHelper, bool hasError);
+void tsdbSetHelperTable(SRWHelper* pHelper, STable* pTable, STsdbRepo* pRepo);
+int  tsdbCommitTableData(SRWHelper* pHelper, SCommitIter* pCommitIter, SDataCols* pDataCols, TSKEY maxKey);
+int  tsdbMoveLastBlockIfNeccessary(SRWHelper* pHelper);
+int  tsdbWriteCompInfo(SRWHelper* pHelper);
+int  tsdbWriteCompIdx(SRWHelper* pHelper);
+int  tsdbLoadCompIdx(SRWHelper* pHelper, void* target);
+int  tsdbLoadCompInfo(SRWHelper* pHelper, void* target);
+int  tsdbLoadCompData(SRWHelper* phelper, SCompBlock* pcompblock, void* target);
+void tsdbGetDataStatis(SRWHelper* pHelper, SDataStatis* pStatis, int numOfCols);
+int  tsdbLoadBlockDataCols(SRWHelper* pHelper, SCompBlock* pCompBlock, SCompInfo* pCompInfo, int16_t* colIds,
+                           int numOfColIds);
+int  tsdbLoadBlockData(SRWHelper* pHelper, SCompBlock* pCompBlock, SCompInfo* pCompInfo);
+
+static FORCE_INLINE int compTSKEY(const void* key1, const void* key2) {
+  if (*(TSKEY*)key1 > *(TSKEY*)key2) {
+    return 1;
+  } else if (*(TSKEY*)key1 == *(TSKEY*)key2) {
+    return 0;
+  } else {
+    return -1;
+  }
+}
 
 // ------------------ tsdbMain.c
 #define REPO_ID(r) (r)->config.tsdbId

src/tsdb/src/tsdbMemTable.c

@@ -18,11 +18,6 @@
 
 #define TSDB_DATA_SKIPLIST_LEVEL 5
 
-typedef struct {
-  STable *           pTable;
-  SSkipListIterator *pIter;
-} SCommitIter;
-
 static FORCE_INLINE STsdbBufBlock *tsdbGetCurrBufBlock(STsdbRepo *pRepo);
 
 static void        tsdbFreeBytes(STsdbRepo *pRepo, void *ptr, int bytes);
@@ -34,14 +29,11 @@ static char *      tsdbGetTsTupleKey(const void *data);
 static void *      tsdbCommitData(void *arg);
 static int         tsdbCommitMeta(STsdbRepo *pRepo);
 static void        tsdbEndCommit(STsdbRepo *pRepo);
-static TSKEY       tsdbNextIterKey(SCommitIter *pIter);
 static int         tsdbHasDataToCommit(SCommitIter *iters, int nIters, TSKEY minKey, TSKEY maxKey);
 static int  tsdbCommitToFile(STsdbRepo *pRepo, int fid, SCommitIter *iters, SRWHelper *pHelper, SDataCols *pDataCols);
 static void tsdbGetFidKeyRange(int daysPerFile, int8_t precision, int fileId, TSKEY *minKey, TSKEY *maxKey);
-static SCommitIter *tsdbCreateTableIters(STsdbRepo *pRepo);
-static void         tsdbDestroyTableIters(SCommitIter *iters, int maxTables);
-static int          tsdbReadRowsFromCache(STsdbMeta *pMeta, STable *pTable, SSkipListIterator *pIter, TSKEY maxKey,
-                                          int maxRowsToRead, SDataCols *pCols);
+static SCommitIter *tsdbCreateCommitIters(STsdbRepo *pRepo);
+static void         tsdbDestroyCommitIters(SCommitIter *iters, int maxTables);
 
 // ---------------- INTERNAL FUNCTIONS ----------------
 int tsdbInsertRowToMem(STsdbRepo *pRepo, SDataRow row, STable *pTable) {
@@ -252,6 +244,66 @@ int tsdbAsyncCommit(STsdbRepo *pRepo) {
   return 0;
 }
 
+int tsdbLoadDataFromCache(STable *pTable, SSkipListIterator *pIter, TSKEY maxKey, int maxRowsToRead, SDataCols *pCols,
+                          TSKEY *filterKeys, int nFilterKeys) {
+  ASSERT(maxRowsToRead > 0 && nFilterKeys >= 0);
+  if (pIter == NULL) return 0;
+  STSchema *pSchema = NULL;
+  int       numOfRows = 0;
+  TSKEY     keyNext = 0;
+  int       filterIter = 0;
+
+  if (nFilterKeys != 0) { // for filter purpose
+    ASSERT(filterKeys != NULL);
+    keyNext = tsdbNextIterKey(pIter);
+    if (keyNext < 0 || keyNext > maxKey) return numOfRows;
+    void *ptr = taosbsearch((void *)(&keyNext), (void *)filterKeys, nFilterKeys, sizeof(TSKEY), compTSKEY, TD_GE);
+    filterIter = (ptr == NULL) ? nFilterKeys : (POINTER_DISTANCE(ptr, filterKeys) / sizeof(TSKEY));
+  }
+
+  do {
+    if (numOfRows >= maxRowsToRead) break;
+
+    SDataRow row = tsdbNextIterRow(pIter);
+    if (row == NULL) break;
+
+    keyNext = dataRowKey(row);
+    if (keyNext < 0 || keyNext > maxKey) break;
+
+    bool keyFiltered = false;
+    if (nFilterKeys != 0) {
+      while (true) {
+        if (filterIter >= nFilterKeys) break;
+        if (keyNext == filterKeys[filterIter]) {
+          keyFiltered = true;
+          filterIter++;
+          break;
+        } else if (keyNext < filterKeys[filterIter]) {
+          break;
+        } else {
+          filterIter++;
+        }
+      }
+    }
+
+    if (!keyFiltered) {
+      if (pCols) {
+        if (pSchema == NULL || schemaVersion(pSchema) != dataRowVersion(row)) {
+          pSchema = tsdbGetTableSchemaImpl(pTable, false, false, dataRowVersion(row));
+          if (pSchema == NULL) {
+            ASSERT(0);
+          }
+        }
+
+        tdAppendDataRowToDataCol(row, pSchema, pCols);
+      }
+      numOfRows++;
+    }
+  } while (tSkipListIterNext(pIter));
+
+  return numOfRows;
+}
+
 // ---------------- LOCAL FUNCTIONS ----------------
 static FORCE_INLINE STsdbBufBlock *tsdbGetCurrBufBlock(STsdbRepo *pRepo) {
   ASSERT(pRepo != NULL);
@@ -378,7 +430,7 @@ static void *tsdbCommitData(void *arg) {
 
   // Create the iterator to read from cache
   if (pMem->numOfRows > 0) {
-    iters = tsdbCreateTableIters(pRepo);
+    iters = tsdbCreateCommitIters(pRepo);
     if (iters == NULL) {
       tsdbError("vgId:%d failed to create commit iterator since %s", REPO_ID(pRepo), tstrerror(terrno));
       goto _exit;
@@ -418,7 +470,7 @@ static void *tsdbCommitData(void *arg) {
 
 _exit:
   tdFreeDataCols(pDataCols);
-  tsdbDestroyTableIters(iters, pCfg->maxTables);
+  tsdbDestroyCommitIters(iters, pCfg->maxTables);
   tsdbDestroyHelper(&whelper);
   tsdbEndCommit(pRepo);
   tsdbInfo("vgId:%d commit over", pRepo->config.tsdbId);
@@ -479,19 +531,9 @@ static void tsdbEndCommit(STsdbRepo *pRepo) {
   if (pRepo->appH.notifyStatus) pRepo->appH.notifyStatus(pRepo->appH.appH, TSDB_STATUS_COMMIT_OVER);
 }
 
-static TSKEY tsdbNextIterKey(SCommitIter *pIter) {
-  if (pIter == NULL) return -1;
-
-  SSkipListNode *node = tSkipListIterGet(pIter->pIter);
-  if (node == NULL) return -1;
-
-  SDataRow row = SL_GET_NODE_DATA(node);
-  return dataRowKey(row);
-}
-
 static int tsdbHasDataToCommit(SCommitIter *iters, int nIters, TSKEY minKey, TSKEY maxKey) {
   for (int i = 0; i < nIters; i++) {
-    TSKEY nextKey = tsdbNextIterKey(iters + i);
+    TSKEY nextKey = tsdbNextIterKey((iters + i)->pIter);
     if (nextKey > 0 && (nextKey >= minKey && nextKey <= maxKey)) return 1;
   }
   return 0;
@@ -504,7 +546,6 @@ static void tsdbGetFidKeyRange(int daysPerFile, int8_t precision, int fileId, TS
 
 static int tsdbCommitToFile(STsdbRepo *pRepo, int fid, SCommitIter *iters, SRWHelper *pHelper, SDataCols *pDataCols) {
   char *      dataDir = NULL;
-  STsdbMeta * pMeta = pRepo->tsdbMeta;
   STsdbCfg *  pCfg = &pRepo->config;
   STsdbFileH *pFileH = pRepo->tsdbFileH;
   SFileGroup *pGroup = NULL;
@@ -549,33 +590,13 @@ static int tsdbCommitToFile(STsdbRepo *pRepo, int fid, SCommitIter *iters, SRWHe
     if (pIter->pIter != NULL) {
       tdInitDataCols(pDataCols, tsdbGetTableSchemaImpl(pIter->pTable, false, false, -1));
 
-      int maxRowsToRead = pCfg->maxRowsPerFileBlock * 4 / 5;
-      int nLoop = 0;
-      while (true) {
-        int rowsRead = tsdbReadRowsFromCache(pMeta, pIter->pTable, pIter->pIter, maxKey, maxRowsToRead, pDataCols);
-        ASSERT(rowsRead >= 0);
-        if (pDataCols->numOfRows == 0) break;
-        nLoop++;
-
-        ASSERT(dataColsKeyFirst(pDataCols) >= minKey && dataColsKeyFirst(pDataCols) <= maxKey);
-        ASSERT(dataColsKeyLast(pDataCols) >= minKey && dataColsKeyLast(pDataCols) <= maxKey);
-
-        int rowsWritten = tsdbWriteDataBlock(pHelper, pDataCols);
-        ASSERT(rowsWritten != 0);
-        if (rowsWritten < 0) {
-          taosRUnLockLatch(&(pIter->pTable->latch));
-          tsdbError("vgId:%d failed to write data block to table %s tid %d uid %" PRIu64 " since %s", REPO_ID(pRepo),
-                    TABLE_CHAR_NAME(pIter->pTable), TABLE_TID(pIter->pTable), TABLE_UID(pIter->pTable),
-                    tstrerror(terrno));
-          goto _err;
-        }
-        ASSERT(rowsWritten <= pDataCols->numOfRows);
-
-        tdPopDataColsPoints(pDataCols, rowsWritten);
-        maxRowsToRead = pCfg->maxRowsPerFileBlock * 4 / 5 - pDataCols->numOfRows;
+      if (tsdbCommitTableData(pHelper, pIter, pDataCols, maxKey) < 0) {
+        taosRUnLockLatch(&(pIter->pTable->latch));
+        tsdbError("vgId:%d failed to write data of table %s tid %d uid %" PRIu64 " since %s", REPO_ID(pRepo),
+                  TABLE_CHAR_NAME(pIter->pTable), TABLE_TID(pIter->pTable), TABLE_UID(pIter->pTable),
+                  tstrerror(terrno));
+        goto _err;
       }
-
-      ASSERT(pDataCols->numOfRows == 0);
     }
 
     taosRUnLockLatch(&(pIter->pTable->latch));
@@ -615,7 +636,7 @@ _err:
   return -1;
 }
 
-static SCommitIter *tsdbCreateTableIters(STsdbRepo *pRepo) {
+static SCommitIter *tsdbCreateCommitIters(STsdbRepo *pRepo) {
   STsdbCfg * pCfg = &(pRepo->config);
   SMemTable *pMem = pRepo->imem;
   STsdbMeta *pMeta = pRepo->tsdbMeta;
@@ -645,21 +666,18 @@ static SCommitIter *tsdbCreateTableIters(STsdbRepo *pRepo) {
         goto _err;
       }
 
-      if (!tSkipListIterNext(iters[i].pIter)) {
-        terrno = TSDB_CODE_TDB_NO_TABLE_DATA_IN_MEM;
-        goto _err;
-      }
+      tSkipListIterNext(iters[i].pIter);
     }
   }
 
   return iters;
 
 _err:
-  tsdbDestroyTableIters(iters, pCfg->maxTables);
+  tsdbDestroyCommitIters(iters, pCfg->maxTables);
   return NULL;
 }
 
-static void tsdbDestroyTableIters(SCommitIter *iters, int maxTables) {
+static void tsdbDestroyCommitIters(SCommitIter *iters, int maxTables) {
   if (iters == NULL) return;
 
   for (int i = 1; i < maxTables; i++) {
@@ -670,35 +688,4 @@ static void tsdbDestroyTableIters(SCommitIter *iters, int maxTables) {
   }
 
   free(iters);
-}
-
-static int tsdbReadRowsFromCache(STsdbMeta *pMeta, STable *pTable, SSkipListIterator *pIter, TSKEY maxKey, int maxRowsToRead, SDataCols *pCols) {
-  ASSERT(maxRowsToRead > 0);
-  if (pIter == NULL) return 0;
-  STSchema *pSchema = NULL;
-
-  int numOfRows = 0;
-
-  do {
-    if (numOfRows >= maxRowsToRead) break;
-
-    SSkipListNode *node = tSkipListIterGet(pIter);
-    if (node == NULL) break;
-
-    SDataRow row = SL_GET_NODE_DATA(node);
-    if (dataRowKey(row) > maxKey) break;
-
-    if (pSchema == NULL || schemaVersion(pSchema) != dataRowVersion(row)) {
-      pSchema = tsdbGetTableSchemaImpl(pTable, true, false, dataRowVersion(row));
-      if (pSchema == NULL) {
-        // TODO: deal with the error here
-        ASSERT(0);
-      }
-    }
-
-    tdAppendDataRowToDataCol(row, pSchema, pCols);
-    numOfRows++;
-  } while (tSkipListIterNext(pIter));
-
-  return numOfRows;
 }
\ No newline at end of file

src/tsdb/src/tsdbMeta.c

@@ -727,7 +727,7 @@ static STable *tsdbNewTable(STableCfg *pCfg, bool isSuper) {
 
   T_REF_INC(pTable);
 
-  tsdbDebug("table %s tid %d uid %" PRIu64 " is created", TABLE_CHAR_NAME(pTable), TABLE_TID(pTable),
+  tsdbTrace("table %s tid %d uid %" PRIu64 " is created", TABLE_CHAR_NAME(pTable), TABLE_TID(pTable),
             TABLE_UID(pTable));
 
   return pTable;
@@ -740,7 +740,7 @@ _err:
 static void tsdbFreeTable(STable *pTable) {
   if (pTable) {
     if (pTable->name != NULL)
-      tsdbDebug("table %s tid %d uid %" PRIu64 " is destroyed", TABLE_CHAR_NAME(pTable), TABLE_TID(pTable),
+      tsdbTrace("table %s tid %d uid %" PRIu64 " is freed", TABLE_CHAR_NAME(pTable), TABLE_TID(pTable),
                 TABLE_UID(pTable));
     tfree(TABLE_NAME(pTable));
     if (TABLE_TYPE(pTable) != TSDB_CHILD_TABLE) {

src/tsdb/src/tsdbRead.c

@@ -93,7 +93,7 @@ typedef struct SBlockOrderSupporter {
 typedef struct SIOCostSummary {
   int64_t blockLoadTime;
   int64_t statisInfoLoadTime;
-  int64_t blockMergeTime;
+  int64_t checkForNextTime;
 } SIOCostSummary;
 
 typedef struct STsdbQueryHandle {
@@ -388,7 +388,7 @@ SDataRow getSDataRowInTableMem(STableCheckInfo* pCheckInfo) {
   return NULL;
 }
 
-bool moveToNextRow(STableCheckInfo* pCheckInfo) {
+static bool moveToNextRow(STableCheckInfo* pCheckInfo) {
   bool hasNext = false;
   if (pCheckInfo->chosen == 0) {
     if (pCheckInfo->iter != NULL) {
@@ -587,22 +587,13 @@ static int32_t getFileCompInfo(STsdbQueryHandle* pQueryHandle, int32_t* numOfBlo
                     .uid = (_checkInfo)->tableId.uid})
 
 
-
 static bool doLoadFileDataBlock(STsdbQueryHandle* pQueryHandle, SCompBlock* pBlock, STableCheckInfo* pCheckInfo) {
   STsdbRepo *pRepo = pQueryHandle->pTsdb;
-
-  // TODO refactor
-  SCompData* data = calloc(1, sizeof(SCompData) + sizeof(SCompCol) * pBlock->numOfCols);
-
-  data->numOfCols = pBlock->numOfCols;
-  data->uid = pCheckInfo->pTableObj->tableId.uid;
-
-  bool    blockLoaded = false;
-  int64_t st = taosGetTimestampUs();
+  bool       blockLoaded = false;
+  int64_t    st = taosGetTimestampUs();
 
   if (pCheckInfo->pDataCols == NULL) {
     STsdbMeta* pMeta = tsdbGetMeta(pRepo);
-    // TODO
     pCheckInfo->pDataCols = tdNewDataCols(pMeta->maxRowBytes, pMeta->maxCols, pRepo->config.maxRowsPerFileBlock);
   }
 
@@ -611,9 +602,10 @@ static bool doLoadFileDataBlock(STsdbQueryHandle* pQueryHandle, SCompBlock* pBlo
   tdInitDataCols(pQueryHandle->rhelper.pDataCols[0], pSchema);
   tdInitDataCols(pQueryHandle->rhelper.pDataCols[1], pSchema);
 
-//  int16_t* colIds = pQueryHandle->defaultLoadColumn->pData;
-//  int32_t ret = tsdbLoadBlockDataCols(&(pQueryHandle->rhelper), pBlock, pCheckInfo->pCompInfo, colIds, QH_GET_NUM_OF_COLS(pQueryHandle));
-  if (tsdbLoadBlockData(&(pQueryHandle->rhelper), pBlock, pCheckInfo->pCompInfo) == 0) {
+  int16_t* colIds = pQueryHandle->defaultLoadColumn->pData;
+  int32_t ret = tsdbLoadBlockDataCols(&(pQueryHandle->rhelper), pBlock, pCheckInfo->pCompInfo, colIds, QH_GET_NUM_OF_COLS(pQueryHandle));
+//  int32_t ret = tsdbLoadBlockData(&(pQueryHandle->rhelper), pBlock, pCheckInfo->pCompInfo);
+    if (ret == TSDB_CODE_SUCCESS) {
     SDataBlockLoadInfo* pBlockLoadInfo = &pQueryHandle->dataBlockLoadInfo;
 
     pBlockLoadInfo->fileGroup = pQueryHandle->pFileGroup;
@@ -627,12 +619,10 @@ static bool doLoadFileDataBlock(STsdbQueryHandle* pQueryHandle, SCompBlock* pBlo
   assert(pCols->numOfRows != 0 && pCols->numOfRows <= pBlock->numOfRows);
 
   pBlock->numOfRows = pCols->numOfRows;
-  tfree(data);
-
-  int64_t et = taosGetTimestampUs() - st;
 
-  pQueryHandle->cost.blockLoadTime += et;
-  tsdbDebug("%p load file block into buffer, elapsed time:%"PRId64 " us", pQueryHandle, et);
+  int64_t elapsedTime = (taosGetTimestampUs() - st);
+  pQueryHandle->cost.blockLoadTime += elapsedTime;
+  tsdbDebug("%p load file block into buffer, elapsed time:%"PRId64 " us", pQueryHandle, elapsedTime);
 
   return blockLoaded;
 }
@@ -1492,7 +1482,10 @@ static bool doHasDataInBuffer(STsdbQueryHandle* pQueryHandle) {
 // handle data in cache situation
 bool tsdbNextDataBlock(TsdbQueryHandleT* pHandle) {
   STsdbQueryHandle* pQueryHandle = (STsdbQueryHandle*) pHandle;
-  
+
+  int64_t stime = taosGetTimestampUs();
+  int64_t elapsedTime = stime;
+
   size_t numOfTables = taosArrayGetSize(pQueryHandle->pTableCheckInfo);
   assert(numOfTables > 0);
 
@@ -1609,6 +1602,8 @@ bool tsdbNextDataBlock(TsdbQueryHandleT* pHandle) {
     }
 
     if (exists) {
+      elapsedTime = taosGetTimestampUs() - stime;
+      pQueryHandle->cost.checkForNextTime += elapsedTime;
       return exists;
     }
   
@@ -1617,7 +1612,11 @@ bool tsdbNextDataBlock(TsdbQueryHandleT* pHandle) {
   }
   
   // TODO: opt by consider the scan order
-  return doHasDataInBuffer(pQueryHandle);
+  bool ret = doHasDataInBuffer(pQueryHandle);
+
+  elapsedTime = taosGetTimestampUs() - stime;
+  pQueryHandle->cost.checkForNextTime += elapsedTime;
+  return ret;
 }
 
 void changeQueryHandleForLastrowQuery(TsdbQueryHandleT pqHandle) {
@@ -1895,8 +1894,6 @@ SArray* tsdbRetrieveDataBlock(TsdbQueryHandleT* pQueryHandle, SArray* pIdList) {
   }
 }
 
-SArray* tsdbRetrieveDataRow(TsdbQueryHandleT* pQueryHandle, SArray* pIdList, SQueryRowCond* pCond) { return NULL; }
-
 static int32_t getAllTableList(STable* pSuperTable, SArray* list) {
   SSkipListIterator* iter = tSkipListCreateIter(pSuperTable->pIndex);
   while (tSkipListIterNext(iter)) {
@@ -2214,7 +2211,7 @@ int32_t tsdbQuerySTableByTagCond(TSDB_REPO_T* tsdb, uint64_t uid, const char* pT
   int32_t ret = TSDB_CODE_SUCCESS;
   tExprNode* expr = NULL;
 
-  TRY(32) {
+  TRY(TSDB_MAX_TAGS) {
     expr = exprTreeFromTableName(tbnameCond);
     if (expr == NULL) {
       expr = exprTreeFromBinary(pTagCond, len);
@@ -2238,7 +2235,8 @@ int32_t tsdbQuerySTableByTagCond(TSDB_REPO_T* tsdb, uint64_t uid, const char* pT
 
   } CATCH( code ) {
     CLEANUP_EXECUTE();
-    ret = code;
+    terrno = code;
+    goto _error; 
     // TODO: more error handling
   } END_TRY
 
@@ -2364,8 +2362,9 @@ void tsdbCleanupQueryHandle(TsdbQueryHandleT queryHandle) {
 
   tsdbDestroyHelper(&pQueryHandle->rhelper);
 
-  tsdbDebug(":io-cost summary: statis-info time:%"PRId64"us, datablock time:%" PRId64"us ,%p", pQueryHandle->cost.statisInfoLoadTime,
-      pQueryHandle->cost.blockLoadTime, pQueryHandle->qinfo);
+  SIOCostSummary* pCost = &pQueryHandle->cost;
+  tsdbDebug(":io-cost summary: statis-info:%"PRId64"us, datablock:%" PRId64"us, check data:%"PRId64"us, %p",
+      pCost->statisInfoLoadTime, pCost->blockLoadTime, pCost->checkForNextTime, pQueryHandle->qinfo);
 
   tfree(pQueryHandle);
 }

src/util/inc/exception.h

@@ -78,6 +78,7 @@ void cleanupPush_int_ptr        ( bool failOnly, void* func, void* arg );
 int32_t cleanupGetActionCount();
 void cleanupExecuteTo( int32_t anchor, bool failed );
 void cleanupExecute( SExceptionNode* node, bool failed );
+bool cleanupExceedLimit();
 
 #define CLEANUP_PUSH_VOID_PTR_PTR( failOnly, func, arg1, arg2 )  cleanupPush_void_ptr_ptr( (failOnly), (void*)(func), (void*)(arg1), (void*)(arg2) )
 #define CLEANUP_PUSH_VOID_PTR_BOOL( failOnly, func, arg1, arg2 ) cleanupPush_void_ptr_bool( (failOnly), (void*)(func), (void*)(arg1), (bool)(arg2) )
@@ -91,7 +92,7 @@ void cleanupExecute( SExceptionNode* node, bool failed );
 
 #define CLEANUP_GET_ANCHOR()          cleanupGetActionCount()
 #define CLEANUP_EXECUTE_TO( anchor, failed )  cleanupExecuteTo( (anchor), (failed) )
-
+#define CLEANUP_EXCEED_LIMIT()        cleanupExceedLimit() 
 
 ////////////////////////////////////////////////////////////////////////////////
 // functions & macros for exception handling

src/util/inc/tfile.h

@@ -19,13 +19,14 @@
 #ifdef TAOS_RANDOM_FILE_FAIL
 
 void taosSetRandomFileFailFactor(int factor);
-ssize_t taos_tread(int fd, void *buf, size_t count);
-ssize_t taos_twrite(int fd, void *buf, size_t count);
-off_t taos_lseek(int fd, off_t offset, int whence);
+void taosSetRandomFileFailOutput(const char *path);
+ssize_t taos_tread(int fd, void *buf, size_t count, const char *file, uint32_t line);
+ssize_t taos_twrite(int fd, void *buf, size_t count, const char *file, uint32_t line);
+off_t taos_lseek(int fd, off_t offset, int whence, const char *file, uint32_t line);
 
-#define tread(fd, buf, count)  taos_tread(fd, buf, count)
-#define twrite(fd, buf, count)  taos_twrite(fd, buf, count)
-#define lseek(fd, offset, whence)  taos_lseek(fd, offset, whence)
+#define tread(fd, buf, count)  taos_tread(fd, buf, count, __FILE__, __LINE__)
+#define twrite(fd, buf, count)  taos_twrite(fd, buf, count, __FILE__, __LINE__)
+#define lseek(fd, offset, whence)  taos_lseek(fd, offset, whence, __FILE__, __LINE__)
 
 #endif  // TAOS_RANDOM_FILE_FAIL
 

src/util/src/exception.c

@@ -147,3 +147,6 @@ void cleanupExecuteTo( int32_t anchor, bool failed ) {
 void cleanupExecute( SExceptionNode* node, bool failed ) {
     doExecuteCleanup( node, 0, failed );
 }
+bool cleanupExceedLimit() {
+  return expList->numCleanupAction >= expList->maxCleanupAction;
+}

src/util/src/tcache.c

@@ -294,7 +294,7 @@ void *taosCachePut(SCacheObj *pCacheObj, const void *key, size_t keyLen, const v
     }
   } else {  // old data exists, update the node
     pNode = taosUpdateCacheImpl(pCacheObj, pOld, key, keyLen, pData, dataSize, duration * 1000L);
-    uDebug("cache:%s, key:%p, %p exist in cache, updated", pCacheObj->name, key, pNode->data);
+    uDebug("cache:%s, key:%p, %p exist in cache, updated old:%p", pCacheObj->name, key, pNode->data, pOld);
   }
 
   __cache_unlock(pCacheObj);
@@ -307,26 +307,30 @@ void *taosCacheAcquireByKey(SCacheObj *pCacheObj, const void *key, size_t keyLen
     return NULL;
   }
 
+  void *pData = NULL;
+
   __cache_rd_lock(pCacheObj);
-  
+
   SCacheDataNode **ptNode = (SCacheDataNode **)taosHashGet(pCacheObj->pHashTable, key, keyLen);
 
   int32_t ref = 0;
   if (ptNode != NULL) {
     ref = T_REF_INC(*ptNode);
+    pData = (*ptNode)->data;
   }
+
   __cache_unlock(pCacheObj);
-  
-  if (ptNode != NULL) {
+
+  if (pData != NULL) {
     atomic_add_fetch_32(&pCacheObj->statistics.hitCount, 1);
-    uDebug("cache:%s, key:%p, %p is retrieved from cache, refcnt:%d", pCacheObj->name, key, (*ptNode)->data, ref);
+    uDebug("cache:%s, key:%p, %p is retrieved from cache, refcnt:%d", pCacheObj->name, key, pData, ref);
   } else {
     atomic_add_fetch_32(&pCacheObj->statistics.missCount, 1);
     uDebug("cache:%s, key:%p, not in cache, retrieved failed", pCacheObj->name, key);
   }
-  
+
   atomic_add_fetch_32(&pCacheObj->statistics.totalAccess, 1);
-  return (ptNode != NULL) ? (*ptNode)->data : NULL;
+  return pData;
 }
 
 void* taosCacheUpdateExpireTimeByName(SCacheObj *pCacheObj, void *key, size_t keyLen, uint64_t expireTime) {
@@ -453,21 +457,20 @@ void taosCacheRelease(SCacheObj *pCacheObj, void **data, bool _remove) {
   } else {
     uDebug("cache:%s, key:%p, %p is released, refcnt:%d", pCacheObj->name, pNode->key, pNode->data, T_REF_VAL_GET(pNode) - 1);
 
+    __cache_wr_lock(pCacheObj);
+
     // NOTE: once refcount is decrease, pNode may be freed by other thread immediately.
     int32_t ref = T_REF_DEC(pNode);
 
-    if (inTrashCan) {
+    if (inTrashCan && (ref == 0)) {
       // Remove it if the ref count is 0.
       // The ref count does not need to load and check again after lock acquired, since ref count can not be increased when
       // the node is in trashcan.
-      if (ref == 0) {
-        __cache_wr_lock(pCacheObj);
-        assert(pNode->pTNodeHeader->pData == pNode);
-        taosRemoveFromTrashCan(pCacheObj, pNode->pTNodeHeader);
-        __cache_unlock(pCacheObj);
-      }
-
+      assert(pNode->pTNodeHeader->pData == pNode);
+      taosRemoveFromTrashCan(pCacheObj, pNode->pTNodeHeader);
     }
+
+    __cache_unlock(pCacheObj);
   }
 
 //   else {

src/util/src/tfile.c

@@ -29,14 +29,55 @@
 #ifdef TAOS_RANDOM_FILE_FAIL
 
 static int random_file_fail_factor = 20;
+static FILE *fpRandomFileFailOutput = NULL;
 
 void taosSetRandomFileFailFactor(int factor)
 {
   random_file_fail_factor = factor;
 }
+
+static void close_random_file_fail_output()
+{
+  if (fpRandomFileFailOutput != NULL) {
+    if (fpRandomFileFailOutput != stdout) {
+      fclose(fpRandomFileFailOutput);
+    }
+    fpRandomFileFailOutput = NULL;
+  }
+}
+
+static void random_file_fail_output_sig(int sig)
+{
+  fprintf(fpRandomFileFailOutput, "signal %d received.\n", sig);
+
+  struct sigaction act = {0};
+  act.sa_handler = SIG_DFL;
+  sigaction(sig, &act, NULL);
+
+  close_random_file_fail_output();
+  exit(EXIT_FAILURE);
+}
+
+void taosSetRandomFileFailOutput(const char *path)
+{
+  if (path == NULL) {
+    fpRandomFileFailOutput = stdout;
+  } else if ((fpRandomFileFailOutput = fopen(path, "w")) != NULL) {
+    atexit(close_random_file_fail_output);
+  } else {
+    printf("failed to open random file fail log file '%s', errno=%d\n", path, errno);
+    return;
+  }
+
+  struct sigaction act = {0};
+  act.sa_handler = random_file_fail_output_sig;
+  sigaction(SIGFPE, &act, NULL);
+  sigaction(SIGSEGV, &act, NULL);
+  sigaction(SIGILL, &act, NULL);
+}
 #endif
 
-ssize_t taos_tread(int fd, void *buf, size_t count)
+ssize_t taos_tread(int fd, void *buf, size_t count, const char *file, uint32_t line)
 {
 #ifdef TAOS_RANDOM_FILE_FAIL
   if (random_file_fail_factor > 0) {
@@ -49,7 +90,7 @@ ssize_t taos_tread(int fd, void *buf, size_t count)
   return tread(fd, buf, count);
 }
 
-ssize_t taos_twrite(int fd, void *buf, size_t count)
+ssize_t taos_twrite(int fd, void *buf, size_t count, const char *file, uint32_t line)
 {
 #ifdef TAOS_RANDOM_FILE_FAIL
   if (random_file_fail_factor > 0) {
@@ -62,7 +103,7 @@ ssize_t taos_twrite(int fd, void *buf, size_t count)
   return twrite(fd, buf, count);
 }
 
-off_t taos_lseek(int fd, off_t offset, int whence)
+off_t taos_lseek(int fd, off_t offset, int whence, const char *file, uint32_t line)
 {
 #ifdef TAOS_RANDOM_FILE_FAIL
   if (random_file_fail_factor > 0) {

src/vnode/src/vnodeMain.c

@@ -59,7 +59,7 @@ int32_t vnodeInitResources() {
   vnodeInitWriteFp();
   vnodeInitReadFp();
 
-  tsDnodeVnodesHash = taosHashInit(TSDB_MAX_VNODES, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true);
+  tsDnodeVnodesHash = taosHashInit(TSDB_MIN_VNODES, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true);
   if (tsDnodeVnodesHash == NULL) {
     vError("failed to init vnode list");
     return TSDB_CODE_VND_OUT_OF_MEMORY;
@@ -69,6 +69,7 @@ int32_t vnodeInitResources() {
 }
 
 void vnodeCleanupResources() {
+
   if (tsDnodeVnodesHash != NULL) {
     taosHashCleanup(tsDnodeVnodesHash);
     tsDnodeVnodesHash = NULL;
@@ -137,7 +138,7 @@ int32_t vnodeCreate(SMDCreateVnodeMsg *pVnodeCfg) {
     return TSDB_CODE_VND_INIT_FAILED;
   }
 
-  vInfo("vgId:%d, vnode is created, clog:%d", pVnodeCfg->cfg.vgId, pVnodeCfg->cfg.walLevel);
+  vInfo("vgId:%d, vnode is created, walLevel:%d fsyncPeriod:%d", pVnodeCfg->cfg.vgId, pVnodeCfg->cfg.walLevel, pVnodeCfg->cfg.fsyncPeriod);
   code = vnodeOpen(pVnodeCfg->cfg.vgId, rootDir);
 
   return code;
@@ -618,6 +619,7 @@ static int32_t vnodeSaveCfg(SMDCreateVnodeMsg *pVnodeCfg) {
   len += snprintf(content + len, maxLen - len, "  \"precision\": %d,\n", pVnodeCfg->cfg.precision);
   len += snprintf(content + len, maxLen - len, "  \"compression\": %d,\n", pVnodeCfg->cfg.compression);
   len += snprintf(content + len, maxLen - len, "  \"walLevel\": %d,\n", pVnodeCfg->cfg.walLevel);
+  len += snprintf(content + len, maxLen - len, "  \"fsync\": %d,\n", pVnodeCfg->cfg.fsyncPeriod);
   len += snprintf(content + len, maxLen - len, "  \"replica\": %d,\n", pVnodeCfg->cfg.replications);
   len += snprintf(content + len, maxLen - len, "  \"wals\": %d,\n", pVnodeCfg->cfg.wals);
   len += snprintf(content + len, maxLen - len, "  \"quorum\": %d,\n", pVnodeCfg->cfg.quorum);
@@ -782,6 +784,13 @@ static int32_t vnodeReadCfg(SVnodeObj *pVnode) {
   }
   pVnode->walCfg.walLevel = (int8_t) walLevel->valueint;
 
+  cJSON *fsyncPeriod = cJSON_GetObjectItem(root, "fsync");
+  if (!walLevel || walLevel->type != cJSON_Number) {
+    vError("vgId:%d, failed to read vnode cfg, fsyncPeriod not found", pVnode->vgId);
+    goto PARSE_OVER;
+  }
+  pVnode->walCfg.fsyncPeriod = fsyncPeriod->valueint;
+
   cJSON *wals = cJSON_GetObjectItem(root, "wals");
   if (!wals || wals->type != cJSON_Number) {
     vError("vgId:%d, failed to read vnode cfg, wals not found", pVnode->vgId);

src/vnode/src/vnodeRead.c

@@ -108,9 +108,10 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) {
     if (code == TSDB_CODE_SUCCESS) {
       handle = qRegisterQInfo(pVnode->qMgmt, (uint64_t) pQInfo);
       if (handle == NULL) {  // failed to register qhandle
+        vError("vgId:%d QInfo:%p register qhandle failed, return to app, code:%s", pVnode->vgId, (void *)pQInfo,
+               tstrerror(pRsp->code));
         pRsp->code = TSDB_CODE_QRY_INVALID_QHANDLE;
         qDestroyQueryInfo(pQInfo);  // destroy it directly
-        vError("vgId:%d QInfo:%p register qhandle failed, return to app, code:%s", pVnode->vgId, (void*) pQInfo, tstrerror(pRsp->code));
       } else {
         assert(*handle == pQInfo);
         pRsp->qhandle = htobe64((uint64_t) pQInfo);

src/wal/src/walMain.c

@@ -25,6 +25,7 @@
 #include "tlog.h"
 #include "tchecksum.h"
 #include "tutil.h"
+#include "ttimer.h"
 #include "taoserror.h"
 #include "twal.h"
 #include "tqueue.h"
@@ -44,6 +45,9 @@ typedef struct {
   int      fd;
   int      keep;
   int      level;
+  int32_t  fsyncPeriod;
+  void    *timer;
+  void    *signature;
   int      max;  // maximum number of wal files
   uint32_t id;   // increase continuously
   int      num;  // number of wal files
@@ -52,10 +56,23 @@ typedef struct {
   pthread_mutex_t mutex;
 } SWal;
 
+static void    *walTmrCtrl = NULL;
+static int     tsWalNum = 0;
+static pthread_once_t walModuleInit = PTHREAD_ONCE_INIT;
 static uint32_t walSignature = 0xFAFBFDFE;
-static int walHandleExistingFiles(const char *path);
-static int walRestoreWalFile(SWal *pWal, void *pVnode, FWalWrite writeFp);
-static int walRemoveWalFiles(const char *path);
+static int  walHandleExistingFiles(const char *path);
+static int  walRestoreWalFile(SWal *pWal, void *pVnode, FWalWrite writeFp);
+static int  walRemoveWalFiles(const char *path);
+static void walProcessFsyncTimer(void *param, void *tmrId);
+static void walRelease(SWal *pWal);
+
+static void walModuleInitFunc() {
+  walTmrCtrl = taosTmrInit(1000, 100, 300000, "WAL");
+  if (walTmrCtrl == NULL) 
+    walModuleInit = PTHREAD_ONCE_INIT;
+  else
+    wDebug("WAL module is initialized");
+}
 
 void *walOpen(const char *path, const SWalCfg *pCfg) {
   SWal *pWal = calloc(sizeof(SWal), 1);
@@ -64,20 +81,38 @@ void *walOpen(const char *path, const SWalCfg *pCfg) {
     return NULL;
   }
 
+  pthread_once(&walModuleInit, walModuleInitFunc);
+  if (walTmrCtrl == NULL) {
+    free(pWal);
+    terrno = TAOS_SYSTEM_ERROR(errno);
+    return NULL;
+  }
+
+  atomic_add_fetch_32(&tsWalNum, 1);    
   pWal->fd = -1;
   pWal->max = pCfg->wals;
   pWal->id = 0;
   pWal->num = 0;
   pWal->level = pCfg->walLevel;
   pWal->keep = pCfg->keep;
+  pWal->fsyncPeriod = pCfg->fsyncPeriod;
+  pWal->signature = pWal;
   tstrncpy(pWal->path, path, sizeof(pWal->path));
   pthread_mutex_init(&pWal->mutex, NULL);
 
+  if (pWal->fsyncPeriod > 0  && pWal->level == TAOS_WAL_FSYNC) {
+    pWal->timer = taosTmrStart(walProcessFsyncTimer, pWal->fsyncPeriod, pWal, walTmrCtrl);
+    if (pWal->timer == NULL) {
+      terrno = TAOS_SYSTEM_ERROR(errno);
+      walRelease(pWal);
+      return NULL;
+    }
+  }
+
   if (tmkdir(path, 0755) != 0) {
     terrno = TAOS_SYSTEM_ERROR(errno);
     wError("wal:%s, failed to create directory(%s)", path, strerror(errno));
-    pthread_mutex_destroy(&pWal->mutex);
-    free(pWal);
+    walRelease(pWal);
     pWal = NULL;
   }
      
@@ -89,12 +124,11 @@ void *walOpen(const char *path, const SWalCfg *pCfg) {
   if (pWal && pWal->fd <0) {
     terrno = TAOS_SYSTEM_ERROR(errno);
     wError("wal:%s, failed to open(%s)", path, strerror(errno));
-    pthread_mutex_destroy(&pWal->mutex);
-    free(pWal);
+    walRelease(pWal);
     pWal = NULL;
   } 
 
-  if (pWal) wDebug("wal:%s, it is open, level:%d", path, pWal->level);
+  if (pWal) wDebug("wal:%s, it is open, level:%d fsyncPeriod:%d", path, pWal->level, pWal->fsyncPeriod);
   return pWal;
 }
 
@@ -102,7 +136,8 @@ void walClose(void *handle) {
   if (handle == NULL) return;
   
   SWal *pWal = handle;  
-  close(pWal->fd);
+  tclose(pWal->fd);
+  if (pWal->timer) taosTmrStopA(&pWal->timer);
 
   if (pWal->keep == 0) {
     // remove all files in the directory
@@ -118,9 +153,7 @@ void walClose(void *handle) {
     wDebug("wal:%s, it is closed and kept", pWal->name);
   }
 
-  pthread_mutex_destroy(&pWal->mutex);
-
-  free(pWal);
+  walRelease(pWal);
 }
 
 int walRenew(void *handle) {
@@ -194,9 +227,9 @@ int walWrite(void *handle, SWalHead *pHead) {
 void walFsync(void *handle) {
 
   SWal *pWal = handle;
-  if (pWal == NULL) return;
+  if (pWal == NULL || pWal->level != TAOS_WAL_FSYNC || pWal->fd < 0) return;
 
-  if (pWal->level == TAOS_WAL_FSYNC && pWal->fd >=0) {
+  if (pWal->fsyncPeriod == 0) {
     if (fsync(pWal->fd) < 0) {
       wError("wal:%s, fsync failed(%s)", pWal->name, strerror(errno));
     }
@@ -303,6 +336,20 @@ int walGetWalFile(void *handle, char *name, uint32_t *index) {
   return code;
 }  
 
+static void walRelease(SWal *pWal) {
+
+  pthread_mutex_destroy(&pWal->mutex);
+  pWal->signature = NULL;
+  free(pWal);
+
+  if (atomic_sub_fetch_32(&tsWalNum, 1) == 0) {
+    if (walTmrCtrl) taosTmrCleanUp(walTmrCtrl);
+    walTmrCtrl = NULL;
+    walModuleInit = PTHREAD_ONCE_INIT;
+    wDebug("WAL module is cleaned up");
+  }
+}
+
 static int walRestoreWalFile(SWal *pWal, void *pVnode, FWalWrite writeFp) {
   char *name = pWal->name;
 
@@ -433,3 +480,15 @@ static int walRemoveWalFiles(const char *path) {
   return terrno;
 }
 
+static void walProcessFsyncTimer(void *param, void *tmrId) {
+  SWal *pWal = param;
+
+  if (pWal->signature != pWal) return;
+  if (pWal->fd < 0) return;
+
+  if (fsync(pWal->fd) < 0) {
+    wError("wal:%s, fsync failed(%s)", pWal->name, strerror(errno));
+  }
+  
+  pWal->timer = taosTmrStart(walProcessFsyncTimer, pWal->fsyncPeriod, pWal, walTmrCtrl);
+}

tests/CMakeLists.txt

@@ -12,3 +12,4 @@ SET(CMAKE_VERBOSE_MAKEFILE ON)
 ADD_SUBDIRECTORY(examples/c)
 ADD_SUBDIRECTORY(tsim)
 ADD_SUBDIRECTORY(test/c)
+ADD_SUBDIRECTORY(comparisonTest/tdengine)

tests/comparisonTest/tdengine/CMakeLists.txt

+CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
+PROJECT(TDengine)
+
+INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/inc)
+INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/util/inc)
+INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/common/inc)
+INCLUDE_DIRECTORIES(${TD_OS_DIR}/inc)
+
+IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
+  add_executable(tdengineTest tdengineTest.c)
+  target_link_libraries(tdengineTest taos_static tutil common pthread)
+ENDIF()

tests/comparisonTest/tdengine/tdengineTest.c

@@ -7,6 +7,7 @@
 #include <time.h>
 #include <pthread.h>
 #include <sys/time.h>
+#include <inttypes.h>
 
 typedef struct {
   char sql[256];
@@ -123,19 +124,21 @@ void writeDataImp(void *param) {
   if (taos == NULL)
     taos_error(taos);
 
-  int code = taos_query(taos, "use db");
+  TAOS_RES* result = taos_query(taos, "use db");
+  int32_t code = taos_errno(result);
   if (code != 0) {
     taos_error(taos);
   }
+  taos_free_result(result);
 
-  char sql[65000];
+  char *sql = calloc(1, 8*1024*1024);
   int sqlLen = 0;
   int lastMachineid = 0;
   int counter = 0;
   int totalRecords = 0;
 
   for (int j = pThread->sID; j <= pThread->eID; j++) {
-    char fileName[256];
+    char fileName[300];
     sprintf(fileName, "%s/testdata%d.csv", arguments.dataDir, j);
 
     FILE *fp = fopen(fileName, "r");
@@ -162,7 +165,7 @@ void writeDataImp(void *param) {
       int64_t timestamp;
       int temperature;
       float humidity;
-      sscanf(line, "%d%s%d%lld%d%f", &machineid, machinename, &machinegroup, &timestamp, &temperature, &humidity);
+      sscanf(line, "%d%s%d%" PRId64 "%d%f", &machineid, machinename, &machinegroup, &timestamp, &temperature, &humidity);
 
       if (counter == 0) {
         sqlLen = sprintf(sql, "insert into");
@@ -174,14 +177,16 @@ void writeDataImp(void *param) {
                           machineid, machineid, machinename, machinegroup);
       }
 
-      sqlLen += sprintf(sql + sqlLen, "(%lld,%d,%f)", timestamp, temperature, humidity);
+      sqlLen += sprintf(sql + sqlLen, "(%" PRId64 ",%d,%f)", timestamp, temperature, humidity);
       counter++;
 
       if (counter >= arguments.rowsPerRequest) {
-        int code = taos_query(taos, sql);
+        TAOS_RES *result = taos_query(taos, sql);
+        int32_t   code = taos_errno(result);
         if (code != 0) {
-          printf("thread:%d error:%d reason:%s\n", pThread->pid, code, taos_errstr(taos));
+          printf("thread:%d error:%d reason:%s\n", pThread->threadId, code, taos_errstr(taos));
         }
+        taos_free_result(result);
 
         totalRecords += counter;
         counter = 0;
@@ -194,15 +199,18 @@ void writeDataImp(void *param) {
   }
 
   if (counter > 0) {
-    int code = taos_query(taos, sql);
+    TAOS_RES *result = taos_query(taos, sql);
+    int32_t   code = taos_errno(result);
     if (code != 0) {
-      printf("thread:%d error:%d reason:%s\n", pThread->pid, code, taos_errstr(taos));
+      printf("thread:%d error:%d reason:%s\n", pThread->threadId, code, taos_errstr(taos));
     }
+    taos_free_result(result);
 
     totalRecords += counter;
   }
 
   __sync_fetch_and_add(&statis.totalRows, totalRecords);
+  free(sql);
 }
 
 void writeData() {
@@ -215,19 +223,23 @@ void writeData() {
   taos_init();
 
   void *taos = taos_connect("127.0.0.1", "root", "taosdata", NULL, 0);
-  if (taos == NULL)
-    taos_error(taos);
+  if (taos == NULL) taos_error(taos);
 
-  int code = taos_query(taos, "create database if not exists db");
+  TAOS_RES *result = taos_query(taos, "create database if not exists db");
+  int32_t   code = taos_errno(result);
   if (code != 0) {
     taos_error(taos);
   }
+  taos_free_result(result);
 
-  code = taos_query(taos, "create table if not exists db.devices(ts timestamp, temperature int, humidity float) "
-    "tags(devid int, devname binary(16), devgroup int)");
+  result = taos_query(taos,
+                      "create table if not exists db.devices(ts timestamp, temperature int, humidity float) "
+                      "tags(devid int, devname binary(16), devgroup int)");
+  code = taos_errno(result);
   if (code != 0) {
     taos_error(taos);
   }
+  taos_free_result(result);
 
   int64_t st = getTimeStampMs();
 
@@ -292,17 +304,12 @@ void readData() {
 
     int64_t st = getTimeStampMs();
 
-    int code = taos_query(taos, line);
+    TAOS_RES *result = taos_query(taos, line);
+    int32_t   code = taos_errno(result);
     if (code != 0) {
       taos_error(taos);
     }
 
-    void *result = taos_use_result(taos);
-    if (result == NULL) {
-      printf("failed to get result, reason:%s\n", taos_errstr(taos));
-      exit(1);
-    }
-
     TAOS_ROW row;
     int rows = 0;
     //int num_fields = taos_field_count(taos);

tests/pytest/util/dnodes.py

@@ -28,7 +28,9 @@ class TDSimClient:
             "locale": "en_US.UTF-8",
             "charset": "UTF-8",
             "asyncLog": "0",
-            "maxTablesPerVnode": "4",
+            "minTablesPerVnode": "4",
+            "maxTablesPerVnode": "1000",
+            "tableIncStepPerVnode": "10000",
             "maxVgroupsPerDb": "1000",
             "sdbDebugFlag": "143",
             "rpcDebugFlag": "135",

tests/script/general/http/restful_full.sim

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

tests/script/general/parser/testSuite.sim

@@ -26,7 +26,7 @@ sleep 2000
 run general/parser/fill.sim
 sleep 2000
 run general/parser/fill_stb.sim
-sleep 2000                                    
+sleep 2000
 #run general/parser/fill_us.sim               #
 sleep 2000
 run general/parser/first_last.sim
@@ -91,13 +91,11 @@ run general/parser/select_with_tags.sim
 sleep 2000
 run general/parser/groupby.sim
 sleep 2000
+run general/parser/tags_filter.sim
+sleep 2000
 run general/parser/union.sim
 sleep 2000
 run general/parser/sliding.sim
-sleep 2000
-run general/parser/fill_us.sim
-sleep 2000
-run general/parser/tags_filter.sim
 
 #sleep 2000
 #run general/parser/repeatStream.sim

tests/script/sh/deploy.sh

@@ -111,38 +111,40 @@ echo "serverPort          ${NODE}"          >> $TAOS_CFG
 echo "dataDir             $DATA_DIR"      >> $TAOS_CFG
 echo "logDir              $LOG_DIR"       >> $TAOS_CFG
 echo "debugFlag           131"            >> $TAOS_CFG
-echo "mDebugFlag          135"            >> $TAOS_CFG
-echo "sdbDebugFlag        135"            >> $TAOS_CFG
-echo "dDebugFlag          135"            >> $TAOS_CFG
-echo "vDebugFlag          135"            >> $TAOS_CFG
-echo "tsdbDebugFlag       135"            >> $TAOS_CFG
+echo "mDebugFlag          131"            >> $TAOS_CFG
+echo "sdbDebugFlag        131"            >> $TAOS_CFG
+echo "dDebugFlag          131"            >> $TAOS_CFG
+echo "vDebugFlag          131"            >> $TAOS_CFG
+echo "tsdbDebugFlag       131"            >> $TAOS_CFG
 echo "cDebugFlag          135"            >> $TAOS_CFG
-echo "jnidebugFlag        135"            >> $TAOS_CFG
-echo "odbcdebugFlag       135"            >> $TAOS_CFG
-echo "httpDebugFlag       135"            >> $TAOS_CFG
+echo "jnidebugFlag        131"            >> $TAOS_CFG
+echo "odbcdebugFlag       131"            >> $TAOS_CFG
+echo "httpDebugFlag       143"            >> $TAOS_CFG
 echo "monitorDebugFlag    131"            >> $TAOS_CFG
 echo "mqttDebugFlag       131"            >> $TAOS_CFG
-echo "qdebugFlag          135"            >> $TAOS_CFG
-echo "rpcDebugFlag        135"            >> $TAOS_CFG
+echo "qdebugFlag          131"            >> $TAOS_CFG
+echo "rpcDebugFlag        131"            >> $TAOS_CFG
 echo "tmrDebugFlag        131"            >> $TAOS_CFG
-echo "udebugFlag          135"            >> $TAOS_CFG
-echo "sdebugFlag          135"            >> $TAOS_CFG
-echo "wdebugFlag          135"            >> $TAOS_CFG
+echo "udebugFlag          131"            >> $TAOS_CFG
+echo "sdebugFlag          131"            >> $TAOS_CFG
+echo "wdebugFlag          131"            >> $TAOS_CFG
 echo "monitor             0"              >> $TAOS_CFG
 echo "monitorInterval     1"              >> $TAOS_CFG
 echo "http                0"              >> $TAOS_CFG
 echo "numOfThreadsPerCore 2.0"            >> $TAOS_CFG
 echo "defaultPass         taosdata"       >> $TAOS_CFG
-echo "numOfLogLines       10000000"       >> $TAOS_CFG
+echo "numOfLogLines       20000000"       >> $TAOS_CFG
 echo "mnodeEqualVnodeNum  0"              >> $TAOS_CFG
 echo "clog                2"              >> $TAOS_CFG
 echo "statusInterval      1"              >> $TAOS_CFG
 echo "numOfTotalVnodes    4"              >> $TAOS_CFG
 echo "maxVgroupsPerDb     4"              >> $TAOS_CFG
+echo "minTablesPerVnode   4"              >> $TAOS_CFG
 echo "maxTablesPerVnode   1000"           >> $TAOS_CFG
+echo "tableIncStepPerVnode 10000"         >> $TAOS_CFG
 echo "asyncLog            0"              >> $TAOS_CFG
 echo "numOfMnodes         1"              >> $TAOS_CFG
 echo "locale    en_US.UTF-8"              >> $TAOS_CFG
-echo "anyIp               0"              >> $TAOS_CFG
+echo "fsync               0"              >> $TAOS_CFG
 
 

tests/script/tmp/prepare.sim

@@ -25,10 +25,15 @@ system sh/cfg.sh -n dnode2 -c numOfTotalVnodes -v 20
 system sh/cfg.sh -n dnode3 -c numOfTotalVnodes -v 20
 system sh/cfg.sh -n dnode4 -c numOfTotalVnodes -v 20
 
-system sh/cfg.sh -n dnode1 -c maxVgroupsPerDb -v 20
-system sh/cfg.sh -n dnode2 -c maxVgroupsPerDb -v 20
-system sh/cfg.sh -n dnode3 -c maxVgroupsPerDb -v 20
-system sh/cfg.sh -n dnode4 -c maxVgroupsPerDb -v 20
+system sh/cfg.sh -n dnode1 -c maxVgroupsPerDb -v 2
+system sh/cfg.sh -n dnode2 -c maxVgroupsPerDb -v 2
+system sh/cfg.sh -n dnode3 -c maxVgroupsPerDb -v 2
+system sh/cfg.sh -n dnode4 -c maxVgroupsPerDb -v 2
+
+system sh/cfg.sh -n dnode1 -c maxTablesPerVnode -v 100000
+system sh/cfg.sh -n dnode2 -c maxTablesPerVnode -v 100000
+system sh/cfg.sh -n dnode3 -c maxTablesPerVnode -v 100000
+system sh/cfg.sh -n dnode4 -c maxTablesPerVnode -v 100000
 
 system sh/cfg.sh -n dnode1 -c http -v 1
 system sh/cfg.sh -n dnode2 -c http -v 1

tests/script/unique/arbitrator/sync_replica3_dnodeChang_DropAddAlterTableDropDb.sim

@@ -29,7 +29,7 @@ system sh/cfg.sh -n dnode2 -c alternativeRole -v 2
 system sh/cfg.sh -n dnode3 -c alternativeRole -v 2
 system sh/cfg.sh -n dnode4 -c alternativeRole -v 2
 
-$totalTableNum = 10
+$totalTableNum = 40
 system sh/cfg.sh -n dnode1 -c maxtablesPerVnode -v $totalTableNum
 system sh/cfg.sh -n dnode2 -c maxtablesPerVnode -v $totalTableNum
 system sh/cfg.sh -n dnode3 -c maxtablesPerVnode -v $totalTableNum
@@ -65,9 +65,10 @@ sleep 3000
 $totalTableNum = 20
 $sleepTimer = 3000
 
+$maxTables = $totalTableNum * 2 
 $db = db
-print create database $db replica 3 maxTables $totalTableNum
-sql create database $db replica 3 maxTables $totalTableNum
+print create database $db replica 3 maxTables $maxTables
+sql create database $db replica 3 maxTables $maxTables
 sql use $db
 
 # create table , insert data

tests/script/unique/cluster/cluster_main1.sim

@@ -80,6 +80,7 @@ run_back unique/cluster/main1_client1_2.sim
 run_back unique/cluster/main1_client1_3.sim
 run_back unique/cluster/client3.sim
 run_back unique/cluster/client4.sim
+run_back unique/cluster/client5.sim
 
 sleep 20000
 

tests/script/unique/cluster/cluster_main2.sim

@@ -84,6 +84,7 @@ run_back unique/cluster/main2_client2_2.sim
 run_back unique/cluster/main2_client2_3.sim
 run_back unique/cluster/client3.sim
 run_back unique/cluster/client4.sim
+run_back unique/cluster/client5.sim
 
 sleep 20000