refact(cluster): node mgmt

cmake/cmake.define

 cmake_minimum_required(VERSION 3.16)
 
+if (NOT DEFINED TD_GRANT)
+  SET(TD_GRANT FALSE)
+endif()
+if (NOT DEFINED TD_USB_DONGLE)
+  SET(TD_USB_DONGLE FALSE)
+endif()
+IF (TD_GRANT)
+  ADD_DEFINITIONS(-D_GRANT)
+ENDIF ()
+
 IF ("${BUILD_TOOLS}" STREQUAL "")
   IF (TD_LINUX)
     IF (TD_ARM_32)

example/CMakeLists.txt

 add_executable(tmq "")
 add_executable(tstream "")
+add_executable(demoapi "")
 
 target_sources(tmq
     PRIVATE
@@ -10,6 +11,12 @@ target_sources(tstream
     PRIVATE
     "src/tstream.c"
 )
+
+target_sources(demoapi
+    PRIVATE
+    "src/demoapi.c"
+)
+
 target_link_libraries(tmq
     taos
 )
@@ -18,6 +25,10 @@ target_link_libraries(tstream
     taos
 )
 
+target_link_libraries(demoapi
+    taos
+)
+
 target_include_directories(tmq
     PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
 )
@@ -26,5 +37,11 @@ target_include_directories(tstream
     PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
 )
 
+target_include_directories(demoapi
+    PUBLIC "${TD_SOURCE_DIR}/include/client"
+    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
+)
+
 SET_TARGET_PROPERTIES(tmq PROPERTIES OUTPUT_NAME tmq)
 SET_TARGET_PROPERTIES(tstream PROPERTIES OUTPUT_NAME tstream)
+SET_TARGET_PROPERTIES(demoapi PROPERTIES OUTPUT_NAME demoapi)

example/src/demoapi.c

+// C api call sequence demo
+// to compile: gcc -o apidemo apidemo.c -ltaos
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+#include <inttypes.h>
+#include <argp.h>
+
+#include "taos.h"
+
+#define debugPrint(fmt, ...) \
+    do { if (g_args.debug_print || g_args.verbose_print) \
+      fprintf(stdout, "DEBG: "fmt, __VA_ARGS__); } while(0)
+
+#define warnPrint(fmt, ...) \
+    do { fprintf(stderr, "\033[33m"); \
+        fprintf(stderr, "WARN: "fmt, __VA_ARGS__); \
+        fprintf(stderr, "\033[0m"); } while(0)
+
+#define errorPrint(fmt, ...) \
+    do { fprintf(stderr, "\033[31m"); \
+        fprintf(stderr, "ERROR: "fmt, __VA_ARGS__); \
+        fprintf(stderr, "\033[0m"); } while(0)
+
+#define okPrint(fmt, ...) \
+    do { fprintf(stderr, "\033[32m"); \
+        fprintf(stderr, "OK: "fmt, __VA_ARGS__); \
+        fprintf(stderr, "\033[0m"); } while(0)
+
+int64_t g_num_of_tb = 2;
+int64_t g_num_of_rec = 2;
+
+static struct argp_option options[] = {
+    {"tables", 't', "NUMBER", 0, "Number of child tables, default is 10000."},
+    {"records", 'n', "NUMBER", 0,
+     "Number of records for each table, default is 10000."},
+    {0}};
+
+static error_t parse_opt(int key, char *arg, struct argp_state *state) {
+    switch (key) {
+        case 't':
+            g_num_of_tb = atoll(arg);
+            break;
+
+        case 'n':
+            g_num_of_rec = atoll(arg);
+            break;
+    }
+
+    return 0;
+}
+
+static struct argp argp = {options, parse_opt, "", ""};
+
+static void prepare_data(TAOS* taos) {
+    TAOS_RES *res;
+    res = taos_query(taos, "drop database if exists test;");
+    taos_free_result(res);
+    usleep(100000);
+
+    res = taos_query(taos, "create database test;");
+    taos_free_result(res);
+    usleep(100000);
+    taos_select_db(taos, "test");
+
+    res = taos_query(taos, "create table meters(ts timestamp, f float, n int, b binary(20)) tags(area int, localtion binary(20));");
+    taos_free_result(res);
+
+    char command[1024] = {0};
+    for (int64_t i = 0; i < g_num_of_tb; i ++) {
+        sprintf(command, "create table t%"PRId64" using meters tags(%"PRId64", '%s');",
+                i, i, (i%2)?"beijing":"shanghai");
+        res = taos_query(taos,  command);
+        taos_free_result(res);
+
+        int64_t j = 0;
+        int64_t total = 0;
+        int64_t affected;
+        for (; j < g_num_of_rec -1; j ++) {
+            sprintf(command, "insert into t%"PRId64" values(%" PRId64 ", %f, %"PRId64", '%c%d')",
+                    i, 1650000000000+j, (float)j, j, 'a'+(int)j%10, rand());
+            res = taos_query(taos,  command);
+            if ((res) && (0 == taos_errno(res))) {
+                affected = taos_affected_rows(res);
+                total += affected;
+            } else {
+                errorPrint("%s() LN%d: %s\n",
+                        __func__, __LINE__, taos_errstr(res));
+            }
+            taos_free_result(res);
+        }
+        sprintf(command, "insert into t%"PRId64" values(%" PRId64 ", NULL, NULL, NULL)",
+                i, 1650000000000+j+1);
+        res = taos_query(taos,  command);
+        if ((res) && (0 == taos_errno(res))) {
+            affected = taos_affected_rows(res);
+            total += affected;
+        } else {
+            errorPrint("%s() LN%d: %s\n",
+                    __func__, __LINE__, taos_errstr(res));
+        }
+        taos_free_result(res);
+
+        printf("insert %"PRId64" records into t%"PRId64", total affected rows: %"PRId64"\n", j, i, total);
+    }
+}
+
+static int print_result(TAOS_RES* res, int block) {
+    int64_t num_rows = 0;
+    TAOS_ROW    row = NULL;
+    int         num_fields = taos_num_fields(res);
+    TAOS_FIELD* fields = taos_fetch_fields(res);
+
+    if (block) {
+        warnPrint("%s() LN%d, call taos_fetch_block()\n", __func__, __LINE__);
+        int rows = 0;
+        while ((rows = taos_fetch_block(res, &row))) {
+            num_rows += rows;
+        }
+    } else {
+        warnPrint("%s() LN%d, call taos_fetch_rows()\n", __func__, __LINE__);
+        while ((row = taos_fetch_row(res))) {
+            char temp[256] = {0};
+            taos_print_row(temp, row, fields, num_fields);
+            puts(temp);
+            num_rows ++;
+        }
+    }
+
+    return num_rows;
+}
+
+static void verify_query(TAOS* taos) {
+    // TODO: select count(tbname) from stable once stable query work
+    char command[1024] = {0};
+
+    for (int64_t i = 0; i < g_num_of_tb; i++) {
+        sprintf(command, "select * from t%"PRId64"", i);
+        TAOS_RES* res = taos_query(taos, command);
+
+        if (res) {
+            if (0 == taos_errno(res)) {
+                int field_count = taos_field_count(res);
+                printf("field_count: %d\n", field_count);
+                int* lengths = taos_fetch_lengths(res);
+                if (lengths) {
+                for (int c = 0; c < field_count; c++) {
+                    printf("length of column %d is %d\n", c, lengths[c]);
+                }
+                } else {
+                    errorPrint("%s() LN%d: t%"PRId64"'s lengths is NULL\n",
+                            __func__, __LINE__, i);
+                }
+
+               int64_t rows = print_result(res, i % 2);
+               printf("rows is: %"PRId64"\n", rows);
+
+            } else {
+                errorPrint("%s() LN%d: %s\n",
+                        __func__, __LINE__, taos_errstr(res));
+            }
+        } else {
+            errorPrint("%s() LN%d: %s\n",
+                    __func__, __LINE__, taos_errstr(res));
+        }
+    }
+}
+
+int main(int argc, char *argv[]) {
+    const char* host = "127.0.0.1";
+    const char* user = "root";
+    const char* passwd = "taosdata";
+
+    argp_parse(&argp, argc, argv, 0, 0, NULL);
+    TAOS* taos = taos_connect(host, user, passwd, "", 0);
+    if (taos == NULL) {
+        printf("\033[31mfailed to connect to db, reason:%s\033[0m\n", taos_errstr(taos));
+        exit(1);
+    }
+
+    const char* info = taos_get_server_info(taos);
+    printf("server info: %s\n", info);
+    info = taos_get_client_info(taos);
+    printf("client info: %s\n", info);
+
+    prepare_data(taos);
+
+    verify_query(taos);
+
+    taos_close(taos);
+    printf("done\n");
+
+    return 0;
+}
+

example/src/tmq.c

@@ -163,12 +163,13 @@ void basic_consume_loop(tmq_t* tmq, tmq_list_t* topics) {
     printf("subscribe err\n");
     return;
   }
-  /*int32_t cnt = 0;*/
+  int32_t cnt = 0;
   /*clock_t startTime = clock();*/
   while (running) {
     tmq_message_t* tmqmessage = tmq_consumer_poll(tmq, 500);
     if (tmqmessage) {
-      /*cnt++;*/
+      cnt++;
+      printf("get data\n");
       msg_process(tmqmessage);
       tmq_message_destroy(tmqmessage);
       /*} else {*/

include/common/tcommon.h

@@ -222,7 +222,7 @@ typedef struct SFunctParam {
 // the structure for sql function in select clause
 typedef struct SResSchame {
   int8_t  type;
-  int32_t colId;
+  int32_t slotId;
   int32_t bytes;
   int32_t precision;
   int32_t scale;

include/common/tdatablock.h

@@ -136,7 +136,8 @@ static FORCE_INLINE void colDataAppendInt8(SColumnInfoData* pColumnInfoData, uin
 }
 
 static FORCE_INLINE void colDataAppendInt16(SColumnInfoData* pColumnInfoData, uint32_t currentRow, int16_t* v) {
-  ASSERT(pColumnInfoData->info.type == TSDB_DATA_TYPE_SMALLINT || pColumnInfoData->info.type == TSDB_DATA_TYPE_USMALLINT);
+  ASSERT(pColumnInfoData->info.type == TSDB_DATA_TYPE_SMALLINT ||
+         pColumnInfoData->info.type == TSDB_DATA_TYPE_USMALLINT);
   char* p = pColumnInfoData->pData + pColumnInfoData->info.bytes * currentRow;
   *(int16_t*)p = *(int16_t*)v;
 }
@@ -210,15 +211,19 @@ SSDataBlock* createOneDataBlock(const SSDataBlock* pDataBlock);
 
 void blockDebugShowData(const SArray* dataBlocks);
 
+static FORCE_INLINE int32_t blockEstimateEncodeSize(const SSDataBlock* pBlock) {
+  return blockDataGetSerialMetaSize(pBlock) + (int32_t)ceil(blockDataGetSerialRowSize(pBlock) * pBlock->info.rows);
+}
+
 static FORCE_INLINE int32_t blockCompressColData(SColumnInfoData* pColRes, int32_t numOfRows, char* data,
-                                            int8_t compressed) {
+                                                 int8_t compressed) {
   int32_t colSize = colDataGetLength(pColRes, numOfRows);
   return (*(tDataTypes[pColRes->info.type].compFunc))(pColRes->pData, colSize, numOfRows, data,
                                                       colSize + COMP_OVERFLOW_BYTES, compressed, NULL, 0);
 }
 
 static FORCE_INLINE void blockCompressEncode(const SSDataBlock* pBlock, char* data, int32_t* dataLen, int32_t numOfCols,
-                                       int8_t needCompress) {
+                                             int8_t needCompress) {
   int32_t* colSizes = (int32_t*)data;
 
   data += numOfCols * sizeof(int32_t);

include/common/tglobal.h

@@ -18,6 +18,7 @@
 
 #include "tarray.h"
 #include "tdef.h"
+#include "tconfig.h"
 
 #ifdef __cplusplus
 extern "C" {
@@ -129,6 +130,7 @@ void    taosCfgDynamicOptions(const char *option, const char *value);
 void    taosAddDataDir(int32_t index, char *v1, int32_t level, int32_t primary);
 
 struct SConfig *taosGetCfg();
+int32_t taosAddClientLogCfg(SConfig *pCfg);
 
 #ifdef __cplusplus
 }

include/common/tmsg.h

@@ -260,6 +260,7 @@ typedef struct {
 
 typedef struct SSchema {
   int8_t   type;
+  int8_t   index;  // default is 0, not index created
   col_id_t colId;
   int32_t  bytes;
   char     name[TSDB_COL_NAME_LEN];
@@ -2016,6 +2017,7 @@ typedef struct {
 static FORCE_INLINE int32_t taosEncodeSSchema(void** buf, const SSchema* pSchema) {
   int32_t tlen = 0;
   tlen += taosEncodeFixedI8(buf, pSchema->type);
+  tlen += taosEncodeFixedI8(buf, pSchema->index);
   tlen += taosEncodeFixedI32(buf, pSchema->bytes);
   tlen += taosEncodeFixedI16(buf, pSchema->colId);
   tlen += taosEncodeString(buf, pSchema->name);
@@ -2024,6 +2026,7 @@ static FORCE_INLINE int32_t taosEncodeSSchema(void** buf, const SSchema* pSchema
 
 static FORCE_INLINE void* taosDecodeSSchema(void* buf, SSchema* pSchema) {
   buf = taosDecodeFixedI8(buf, &pSchema->type);
+  buf = taosDecodeFixedI8(buf, &pSchema->index);
   buf = taosDecodeFixedI32(buf, &pSchema->bytes);
   buf = taosDecodeFixedI16(buf, &pSchema->colId);
   buf = taosDecodeStringTo(buf, pSchema->name);
@@ -2032,6 +2035,7 @@ static FORCE_INLINE void* taosDecodeSSchema(void* buf, SSchema* pSchema) {
 
 static FORCE_INLINE int32_t tEncodeSSchema(SCoder* pEncoder, const SSchema* pSchema) {
   if (tEncodeI8(pEncoder, pSchema->type) < 0) return -1;
+  if (tEncodeI8(pEncoder, pSchema->index) < 0) return -1;
   if (tEncodeI32(pEncoder, pSchema->bytes) < 0) return -1;
   if (tEncodeI16(pEncoder, pSchema->colId) < 0) return -1;
   if (tEncodeCStr(pEncoder, pSchema->name) < 0) return -1;
@@ -2040,6 +2044,7 @@ static FORCE_INLINE int32_t tEncodeSSchema(SCoder* pEncoder, const SSchema* pSch
 
 static FORCE_INLINE int32_t tDecodeSSchema(SCoder* pDecoder, SSchema* pSchema) {
   if (tDecodeI8(pDecoder, &pSchema->type) < 0) return -1;
+  if (tDecodeI8(pDecoder, &pSchema->index) < 0) return -1;
   if (tDecodeI32(pDecoder, &pSchema->bytes) < 0) return -1;
   if (tDecodeI16(pDecoder, &pSchema->colId) < 0) return -1;
   if (tDecodeCStrTo(pDecoder, pSchema->name) < 0) return -1;

include/common/ttime.h

@@ -64,6 +64,7 @@ char getPrecisionUnit(int32_t precision);
 
 int64_t convertTimePrecision(int64_t time, int32_t fromPrecision, int32_t toPrecision);
 int64_t convertTimeFromPrecisionToUnit(int64_t time, int32_t fromPrecision, char toUnit);
+int32_t convertStringToTimestamp(int16_t type, char *inputData, int64_t timePrec, int64_t *timeVal);
 
 void taosFormatUtcTime(char *buf, int32_t bufLen, int64_t time, int32_t precision);
 

include/common/ttokendef.h

@@ -140,88 +140,89 @@
 #define TK_APPS                           122
 #define TK_CONNECTIONS                    123
 #define TK_LICENCE                        124
-#define TK_QUERIES                        125
-#define TK_SCORES                         126
-#define TK_TOPICS                         127
-#define TK_VARIABLES                      128
-#define TK_BNODES                         129
-#define TK_SNODES                         130
-#define TK_LIKE                           131
-#define TK_INDEX                          132
-#define TK_FULLTEXT                       133
-#define TK_FUNCTION                       134
-#define TK_INTERVAL                       135
-#define TK_TOPIC                          136
-#define TK_AS                             137
-#define TK_DESC                           138
-#define TK_DESCRIBE                       139
-#define TK_RESET                          140
-#define TK_QUERY                          141
-#define TK_EXPLAIN                        142
-#define TK_ANALYZE                        143
-#define TK_VERBOSE                        144
-#define TK_NK_BOOL                        145
-#define TK_RATIO                          146
-#define TK_COMPACT                        147
-#define TK_VNODES                         148
-#define TK_IN                             149
-#define TK_OUTPUTTYPE                     150
-#define TK_AGGREGATE                      151
-#define TK_BUFSIZE                        152
-#define TK_STREAM                         153
-#define TK_INTO                           154
-#define TK_KILL                           155
-#define TK_CONNECTION                     156
-#define TK_MERGE                          157
-#define TK_VGROUP                         158
-#define TK_REDISTRIBUTE                   159
-#define TK_SPLIT                          160
-#define TK_SYNCDB                         161
-#define TK_NULL                           162
-#define TK_FIRST                          163
-#define TK_LAST                           164
-#define TK_NOW                            165
-#define TK_ROWTS                          166
-#define TK_TBNAME                         167
-#define TK_QSTARTTS                       168
-#define TK_QENDTS                         169
-#define TK_WSTARTTS                       170
-#define TK_WENDTS                         171
-#define TK_WDURATION                      172
-#define TK_BETWEEN                        173
-#define TK_IS                             174
-#define TK_NK_LT                          175
-#define TK_NK_GT                          176
-#define TK_NK_LE                          177
-#define TK_NK_GE                          178
-#define TK_NK_NE                          179
-#define TK_MATCH                          180
-#define TK_NMATCH                         181
-#define TK_JOIN                           182
-#define TK_INNER                          183
-#define TK_SELECT                         184
-#define TK_DISTINCT                       185
-#define TK_WHERE                          186
-#define TK_PARTITION                      187
-#define TK_BY                             188
-#define TK_SESSION                        189
-#define TK_STATE_WINDOW                   190
-#define TK_SLIDING                        191
-#define TK_FILL                           192
-#define TK_VALUE                          193
-#define TK_NONE                           194
-#define TK_PREV                           195
-#define TK_LINEAR                         196
-#define TK_NEXT                           197
-#define TK_GROUP                          198
-#define TK_HAVING                         199
-#define TK_ORDER                          200
-#define TK_SLIMIT                         201
-#define TK_SOFFSET                        202
-#define TK_LIMIT                          203
-#define TK_OFFSET                         204
-#define TK_ASC                            205
-#define TK_NULLS                          206
+#define TK_GRANTS                         125
+#define TK_QUERIES                        126
+#define TK_SCORES                         127
+#define TK_TOPICS                         128
+#define TK_VARIABLES                      129
+#define TK_BNODES                         130
+#define TK_SNODES                         131
+#define TK_LIKE                           132
+#define TK_INDEX                          133
+#define TK_FULLTEXT                       134
+#define TK_FUNCTION                       135
+#define TK_INTERVAL                       136
+#define TK_TOPIC                          137
+#define TK_AS                             138
+#define TK_DESC                           139
+#define TK_DESCRIBE                       140
+#define TK_RESET                          141
+#define TK_QUERY                          142
+#define TK_EXPLAIN                        143
+#define TK_ANALYZE                        144
+#define TK_VERBOSE                        145
+#define TK_NK_BOOL                        146
+#define TK_RATIO                          147
+#define TK_COMPACT                        148
+#define TK_VNODES                         149
+#define TK_IN                             150
+#define TK_OUTPUTTYPE                     151
+#define TK_AGGREGATE                      152
+#define TK_BUFSIZE                        153
+#define TK_STREAM                         154
+#define TK_INTO                           155
+#define TK_KILL                           156
+#define TK_CONNECTION                     157
+#define TK_MERGE                          158
+#define TK_VGROUP                         159
+#define TK_REDISTRIBUTE                   160
+#define TK_SPLIT                          161
+#define TK_SYNCDB                         162
+#define TK_NULL                           163
+#define TK_FIRST                          164
+#define TK_LAST                           165
+#define TK_NOW                            166
+#define TK_ROWTS                          167
+#define TK_TBNAME                         168
+#define TK_QSTARTTS                       169
+#define TK_QENDTS                         170
+#define TK_WSTARTTS                       171
+#define TK_WENDTS                         172
+#define TK_WDURATION                      173
+#define TK_BETWEEN                        174
+#define TK_IS                             175
+#define TK_NK_LT                          176
+#define TK_NK_GT                          177
+#define TK_NK_LE                          178
+#define TK_NK_GE                          179
+#define TK_NK_NE                          180
+#define TK_MATCH                          181
+#define TK_NMATCH                         182
+#define TK_JOIN                           183
+#define TK_INNER                          184
+#define TK_SELECT                         185
+#define TK_DISTINCT                       186
+#define TK_WHERE                          187
+#define TK_PARTITION                      188
+#define TK_BY                             189
+#define TK_SESSION                        190
+#define TK_STATE_WINDOW                   191
+#define TK_SLIDING                        192
+#define TK_FILL                           193
+#define TK_VALUE                          194
+#define TK_NONE                           195
+#define TK_PREV                           196
+#define TK_LINEAR                         197
+#define TK_NEXT                           198
+#define TK_GROUP                          199
+#define TK_HAVING                         200
+#define TK_ORDER                          201
+#define TK_SLIMIT                         202
+#define TK_SOFFSET                        203
+#define TK_LIMIT                          204
+#define TK_OFFSET                         205
+#define TK_ASC                            206
+#define TK_NULLS                          207
 
 #define TK_NK_SPACE                       300
 #define TK_NK_COMMENT                     301

include/common/ttypes.h

@@ -142,6 +142,43 @@ typedef struct {
     }                                          \
   } while (0)
 
+#define NUM_TO_STRING(_inputType, _input, _outputBytes, _output)     \
+  do {                                         \
+    switch (_inputType) {                      \
+      case TSDB_DATA_TYPE_TINYINT:             \
+        snprintf(_output, (int32_t)(_outputBytes), "%d", *(int8_t *)(_input));     \
+        break;                                 \
+      case TSDB_DATA_TYPE_UTINYINT:            \
+        snprintf(_output, (int32_t)(_outputBytes), "%d", *(uint8_t *)(_input));     \
+        break;                                 \
+      case TSDB_DATA_TYPE_SMALLINT:            \
+        snprintf(_output, (int32_t)(_outputBytes), "%d", *(int16_t *)(_input));     \
+        break;                                 \
+      case TSDB_DATA_TYPE_USMALLINT:           \
+        snprintf(_output, (int32_t)(_outputBytes), "%d", *(uint16_t *)(_input));     \
+        break;                                 \
+      case TSDB_DATA_TYPE_TIMESTAMP:           \
+      case TSDB_DATA_TYPE_BIGINT:              \
+        snprintf(_output, (int32_t)(_outputBytes), "%" PRId64, *(int64_t *)(_input));     \
+        break;                                 \
+      case TSDB_DATA_TYPE_UBIGINT:             \
+        snprintf(_output, (int32_t)(_outputBytes), "%" PRIu64, *(uint64_t *)(_input));     \
+        break;                                 \
+      case TSDB_DATA_TYPE_FLOAT:               \
+        snprintf(_output, (int32_t)(_outputBytes), "%f", *(float *)(_input));     \
+        break;                                 \
+      case TSDB_DATA_TYPE_DOUBLE:              \
+        snprintf(_output, (int32_t)(_outputBytes), "%f", *(double *)(_input));     \
+        break;                                 \
+      case TSDB_DATA_TYPE_UINT:                \
+        snprintf(_output, (int32_t)(_outputBytes), "%u", *(uint32_t *)(_input));     \
+        break;                                 \
+      default:                                 \
+        snprintf(_output, (int32_t)(_outputBytes), "%d", *(int32_t *)(_input));     \
+        break;                                 \
+    }                                          \
+  } while (0)
+
 #define IS_SIGNED_NUMERIC_TYPE(_t)   ((_t) >= TSDB_DATA_TYPE_TINYINT && (_t) <= TSDB_DATA_TYPE_BIGINT)
 #define IS_UNSIGNED_NUMERIC_TYPE(_t) ((_t) >= TSDB_DATA_TYPE_UTINYINT && (_t) <= TSDB_DATA_TYPE_UBIGINT)
 #define IS_FLOAT_TYPE(_t)            ((_t) == TSDB_DATA_TYPE_FLOAT || (_t) == TSDB_DATA_TYPE_DOUBLE)

include/libs/function/functionMgt.h

@@ -85,8 +85,8 @@ typedef enum EFunctionType {
   // conversion function
   FUNCTION_TYPE_CAST = 2000,
   FUNCTION_TYPE_TO_ISO8601,
+  FUNCTION_TYPE_TO_UNIXTIMESTAMP,
   FUNCTION_TYPE_TO_JSON,
-  FUNCTION_TYPE_UNIXTIMESTAMP,
 
   // date and time function
   FUNCTION_TYPE_NOW = 2500,
@@ -135,8 +135,17 @@ bool fmIsTimeorderFunc(int32_t funcId);
 bool fmIsPseudoColumnFunc(int32_t funcId);
 bool fmIsWindowPseudoColumnFunc(int32_t funcId);
 bool fmIsWindowClauseFunc(int32_t funcId);
+bool fmIsSpecialDataRequiredFunc(int32_t funcId);
+bool fmIsDynamicScanOptimizedFunc(int32_t funcId);
 
-int32_t fmFuncScanType(int32_t funcId);
+typedef enum EFuncDataRequired {
+  FUNC_DATA_REQUIRED_ALL_NEEDED = 1,
+  FUNC_DATA_REQUIRED_STATIS_NEEDED,
+  FUNC_DATA_REQUIRED_NO_NEEDED,
+  FUNC_DATA_REQUIRED_DISCARD
+} EFuncDataRequired;
+
+EFuncDataRequired fmFuncDataRequired(SFunctionNode* pFunc, STimeWindow* pTimeWindow);
 
 int32_t fmGetFuncExecFuncs(int32_t funcId, SFuncExecFuncs* pFpSet);
 int32_t fmGetScalarFuncExecFuncs(int32_t funcId, SScalarFuncExecFuncs* pFpSet);

include/libs/nodes/nodes.h

@@ -216,6 +216,7 @@ SNodeList* nodesMakeList();
 int32_t nodesListAppend(SNodeList* pList, SNodeptr pNode);
 int32_t nodesListStrictAppend(SNodeList* pList, SNodeptr pNode);
 int32_t nodesListMakeAppend(SNodeList** pList, SNodeptr pNode);
+int32_t nodesListMakeStrictAppend(SNodeList** pList, SNodeptr pNode);
 int32_t nodesListAppendList(SNodeList* pTarget, SNodeList* pSrc);
 int32_t nodesListStrictAppendList(SNodeList* pTarget, SNodeList* pSrc);
 int32_t nodesListPushFront(SNodeList* pList, SNodeptr pNode);

include/libs/nodes/plannodes.h

@@ -30,6 +30,7 @@ typedef struct SLogicNode {
   SNode* pConditions;
   SNodeList* pChildren;
   struct SLogicNode* pParent;
+  int32_t optimizedFlag;
 } SLogicNode;
 
 typedef enum EScanType {
@@ -50,6 +51,8 @@ typedef struct SScanLogicNode {
   SName tableName;
   bool showRewrite;
   double ratio;
+  SNodeList* pDynamicScanFuncs;
+  int32_t dataRequired;
 } SScanLogicNode;
 
 typedef struct SJoinLogicNode {
@@ -196,20 +199,13 @@ typedef struct SSystemTableScanPhysiNode {
   int32_t accountId;
 } SSystemTableScanPhysiNode;
 
-typedef enum EScanRequired {
-  SCAN_REQUIRED_DATA_NO_NEEDED = 1,
-  SCAN_REQUIRED_DATA_STATIS_NEEDED,
-  SCAN_REQUIRED_DATA_ALL_NEEDED,
-  SCAN_REQUIRED_DATA_DISCARD,
-} EScanRequired;
-
 typedef struct STableScanPhysiNode {
   SScanPhysiNode scan;
   uint8_t scanFlag;         // denotes reversed scan of data or not
   STimeWindow scanRange;
   double ratio;
-  EScanRequired scanRequired;
-  SNodeList* pScanReferFuncs;
+  int32_t dataRequired;
+  SNodeList* pDynamicScanFuncs;
 } STableScanPhysiNode;
 
 typedef STableScanPhysiNode STableSeqScanPhysiNode;

include/libs/scalar/filter.h

@@ -19,8 +19,8 @@
 extern "C" {
 #endif
 
-#include "tcommon.h"
 #include "nodes.h"
+#include "tcommon.h"
 
 typedef struct SFilterInfo SFilterInfo;
 typedef int32_t (*filer_get_col_from_id)(void *, int32_t, void **);
@@ -31,20 +31,20 @@ enum {
   FLT_OPTION_NEED_UNIQE = 4,
 };
 
-typedef struct SFilterColumnParam{
+typedef struct SFilterColumnParam {
   int32_t numOfCols;
-  SArray* pDataBlock;
+  SArray *pDataBlock;
 } SFilterColumnParam;
 
 extern int32_t filterInitFromNode(SNode *pNode, SFilterInfo **pinfo, uint32_t options);
-extern bool filterExecute(SFilterInfo *info, SSDataBlock *pSrc, int8_t** p, SColumnDataAgg *statis, int16_t numOfCols);
+extern bool filterExecute(SFilterInfo *info, SSDataBlock *pSrc, int8_t **p, SColumnDataAgg *statis, int16_t numOfCols);
 extern int32_t filterSetDataFromSlotId(SFilterInfo *info, void *param);
 extern int32_t filterSetDataFromColId(SFilterInfo *info, void *param);
 extern int32_t filterGetTimeRange(SNode *pNode, STimeWindow *win, bool *isStrict);
-extern int32_t filterConverNcharColumns(SFilterInfo* pFilterInfo, int32_t rows, bool *gotNchar);
-extern int32_t filterFreeNcharColumns(SFilterInfo* pFilterInfo);
-extern void filterFreeInfo(SFilterInfo *info);
-extern bool filterRangeExecute(SFilterInfo *info, SColumnDataAgg *pDataStatis, int32_t numOfCols, int32_t numOfRows);
+extern int32_t filterConverNcharColumns(SFilterInfo *pFilterInfo, int32_t rows, bool *gotNchar);
+extern int32_t filterFreeNcharColumns(SFilterInfo *pFilterInfo);
+extern void    filterFreeInfo(SFilterInfo *info);
+extern bool    filterRangeExecute(SFilterInfo *info, SColumnDataAgg *pDataStatis, int32_t numOfCols, int32_t numOfRows);
 
 #ifdef __cplusplus
 }

include/libs/scalar/scalar.h

@@ -70,6 +70,14 @@ int32_t ltrimFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOut
 int32_t rtrimFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
 int32_t substrFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
 
+/* Conversion functions */
+int32_t castFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
+
+/* Time related functions */
+int32_t toISO8601Function(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
+int32_t toUnixtimestampFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
+int32_t timeTruncateFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
+
 bool getTimePseudoFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 
 int32_t winStartTsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);

include/os/osDir.h

@@ -38,6 +38,7 @@ typedef struct TdDirEntry *TdDirEntryPtr;
 void    taosRemoveDir(const char *dirname);
 bool    taosDirExist(char *dirname);
 int32_t taosMkDir(const char *dirname);
+int32_t taosMulMkDir(const char *dirname);
 void    taosRemoveOldFiles(const char *dirname, int32_t keepDays);
 int32_t taosExpandDir(const char *dirname, char *outname, int32_t maxlen);
 int32_t taosRealPath(char *dirname, int32_t maxlen);

include/util/taoserror.h

@@ -597,6 +597,8 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_PAR_INVALID_ROLLUP_OPTION     TAOS_DEF_ERROR_CODE(0, 0x2622)
 #define TSDB_CODE_PAR_INVALID_RETENTIONS_OPTION TAOS_DEF_ERROR_CODE(0, 0x2623)
 #define TSDB_CODE_PAR_GROUPBY_WINDOW_COEXIST    TAOS_DEF_ERROR_CODE(0, 0x2624)
+#define TSDB_CODE_PAR_INVALID_OPTION_UNIT       TAOS_DEF_ERROR_CODE(0, 0x2625)
+#define TSDB_CODE_PAR_INVALID_KEEP_UNIT         TAOS_DEF_ERROR_CODE(0, 0x2626)
 
 //planner
 #define TSDB_CODE_PLAN_INTERNAL_ERROR              TAOS_DEF_ERROR_CODE(0, 0x2700)

include/util/tdef.h

@@ -94,6 +94,11 @@ extern const int32_t TYPE_BYTES[15];
 #define TSDB_TIME_PRECISION_MICRO_STR "us"
 #define TSDB_TIME_PRECISION_NANO_STR  "ns"
 
+#define TSDB_TIME_PRECISION_SEC_DIGITS   10
+#define TSDB_TIME_PRECISION_MILLI_DIGITS 13
+#define TSDB_TIME_PRECISION_MICRO_DIGITS 16
+#define TSDB_TIME_PRECISION_NANO_DIGITS  19
+
 #define TSDB_INFORMATION_SCHEMA_DB            "information_schema"
 #define TSDB_INS_TABLE_DNODES                 "dnodes"
 #define TSDB_INS_TABLE_MNODES                 "mnodes"

packaging/install.sh

@@ -157,7 +157,7 @@ function install_main_path() {
     ${csudo} mkdir -p ${install_main_dir}/cfg
     ${csudo} mkdir -p ${install_main_dir}/bin
     ${csudo} mkdir -p ${install_main_dir}/connector
-    ${csudo} mkdir -p ${install_main_dir}/driver
+    ${csudo} mkdir -p ${install_main_dir}/lib
     ${csudo} mkdir -p ${install_main_dir}/examples
     ${csudo} mkdir -p ${install_main_dir}/include
     ${csudo} mkdir -p ${install_main_dir}/init.d
@@ -199,6 +199,8 @@ function install_lib() {
     ${csudo} rm -f ${lib_link_dir}/libtaos.*         || :
     ${csudo} rm -f ${lib64_link_dir}/libtaos.*       || :
 
+    ${csudo} cp -rf ${script_dir}/lib/* ${install_main_dir}/lib && ${csudo} chmod 777 ${install_main_dir}/lib/*
+
     ${csudo} ln -s ${install_main_dir}/lib/libtaos.* ${lib_link_dir}/libtaos.so.1
     ${csudo} ln -s ${lib_link_dir}/libtaos.so.1 ${lib_link_dir}/libtaos.so
 

source/client/inc/clientInt.h

@@ -75,12 +75,12 @@ typedef int32_t (*FHbReqHandle)(SClientHbKey* connKey, void* param, SClientHbReq
 typedef struct {
   int8_t inited;
   // ctl
-  int8_t          threadStop;
-  TdThread       thread;
+  int8_t        threadStop;
+  TdThread      thread;
   TdThreadMutex lock;       // used when app init and cleanup
-  SArray*         appHbMgrs;  // SArray<SAppHbMgr*> one for each cluster
-  FHbReqHandle    reqHandle[HEARTBEAT_TYPE_MAX];
-  FHbRspHandle    rspHandle[HEARTBEAT_TYPE_MAX];
+  SArray*       appHbMgrs;  // SArray<SAppHbMgr*> one for each cluster
+  FHbReqHandle  reqHandle[HEARTBEAT_TYPE_MAX];
+  FHbRspHandle  rspHandle[HEARTBEAT_TYPE_MAX];
 } SClientHbMgr;
 
 typedef struct SQueryExecMetric {
@@ -118,42 +118,42 @@ struct SAppInstInfo {
 };
 
 typedef struct SAppInfo {
-  int64_t         startTime;
-  char            appName[TSDB_APP_NAME_LEN];
-  char*           ep;
-  int32_t         pid;
-  int32_t         numOfThreads;
-  SHashObj*       pInstMap;
+  int64_t       startTime;
+  char          appName[TSDB_APP_NAME_LEN];
+  char*         ep;
+  int32_t       pid;
+  int32_t       numOfThreads;
+  SHashObj*     pInstMap;
   TdThreadMutex mutex;
 } SAppInfo;
 
 typedef struct STscObj {
-  char            user[TSDB_USER_LEN];
-  char            pass[TSDB_PASSWORD_LEN];
-  char            db[TSDB_DB_FNAME_LEN];
-  char            ver[128];
-  int32_t         acctId;
-  uint32_t        connId;
-  int32_t         connType;
-  uint64_t        id;         // ref ID returned by taosAddRef
-  TdThreadMutex   mutex;      // used to protect the operation on db
-  int32_t         numOfReqs;  // number of sqlObj bound to this connection
-  SAppInstInfo*   pAppInfo;
+  char          user[TSDB_USER_LEN];
+  char          pass[TSDB_PASSWORD_LEN];
+  char          db[TSDB_DB_FNAME_LEN];
+  char          ver[128];
+  int32_t       acctId;
+  uint32_t      connId;
+  int32_t       connType;
+  uint64_t      id;         // ref ID returned by taosAddRef
+  TdThreadMutex mutex;      // used to protect the operation on db
+  int32_t       numOfReqs;  // number of sqlObj bound to this connection
+  SAppInstInfo* pAppInfo;
 } STscObj;
 
 typedef struct SResultColumn {
   union {
-    char*       nullbitmap;  // bitmap, one bit for each item in the list
-    int32_t*    offset;
+    char*    nullbitmap;  // bitmap, one bit for each item in the list
+    int32_t* offset;
   };
-  char*         pData;
+  char* pData;
 } SResultColumn;
 
 typedef struct SReqResultInfo {
   const char*    pRspMsg;
   const char*    pData;
-  TAOS_FIELD*    fields;     // todo, column names are not needed.
-  TAOS_FIELD*    userFields; // the fields info that return to user
+  TAOS_FIELD*    fields;      // todo, column names are not needed.
+  TAOS_FIELD*    userFields;  // the fields info that return to user
   uint32_t       numOfCols;
   int32_t*       length;
   char**         convertBuf;
@@ -180,13 +180,30 @@ typedef struct SRequestSendRecvBody {
   SShowReqInfo       showInfo;  // todo this attribute will be removed after the query framework being completed.
   SDataBuf           requestMsg;
   int64_t            queryJob;  // query job, created according to sql query DAG.
-  struct SQueryPlan* pDag;       // the query dag, generated according to the sql statement.
+  struct SQueryPlan* pDag;      // the query dag, generated according to the sql statement.
   SReqResultInfo     resInfo;
 } SRequestSendRecvBody;
 
 #define ERROR_MSG_BUF_DEFAULT_SIZE 512
 
+enum {
+  RES_TYPE__QUERY = 1,
+  RES_TYPE__TMQ,
+};
+
+#define TD_RES_QUERY(res) (*(int8_t*)res == RES_TYPE__QUERY)
+#define TD_RES_TMQ(res)   (*(int8_t*)res == RES_TYPE__TMQ)
+
+typedef struct SMqRspObj {
+  int8_t  resType;
+  char*   topic;
+  void*   vg;
+  SArray* res;  // SArray<SReqResultInfo>
+  int32_t resIter;
+} SMqRspObj;
+
 typedef struct SRequestObj {
+  int8_t               resType;  // query or tmq
   uint64_t             requestId;
   int32_t              type;  // request type
   STscObj*             pTscObj;
@@ -203,6 +220,25 @@ typedef struct SRequestObj {
   SRequestSendRecvBody body;
 } SRequestObj;
 
+static FORCE_INLINE SReqResultInfo* tmqGetCurResInfo(TAOS_RES* res) {
+  SMqRspObj* msg = (SMqRspObj*)res;
+  int32_t    resIter = msg->resIter == -1 ? 0 : msg->resIter;
+  return (SReqResultInfo*)taosArrayGet(msg->res, resIter);
+}
+
+static FORCE_INLINE SReqResultInfo* tmqGetNextResInfo(TAOS_RES* res) {
+  SMqRspObj* msg = (SMqRspObj*)res;
+  if (++msg->resIter < taosArrayGetSize(msg->res)) {
+    return (SReqResultInfo*)taosArrayGet(msg->res, msg->resIter);
+  }
+  return NULL;
+}
+
+static FORCE_INLINE SReqResultInfo* tscGetCurResInfo(TAOS_RES* res) {
+  if (TD_RES_QUERY(res)) return &(((SRequestObj*)res)->body.resInfo);
+  return tmqGetCurResInfo(res);
+}
+
 extern SAppInfo appInfo;
 extern int32_t  clientReqRefPool;
 extern int32_t  clientConnRefPool;
@@ -238,14 +274,17 @@ void initMsgHandleFp();
 TAOS* taos_connect_internal(const char* ip, const char* user, const char* pass, const char* auth, const char* db,
                             uint16_t port);
 
-void* doFetchRow(SRequestObj* pRequest, bool setupOneRowPtr, bool convertUcs4);
-
-int32_t setResultDataPtr(SReqResultInfo* pResultInfo, TAOS_FIELD* pFields, int32_t numOfCols, int32_t numOfRows, bool convertUcs4);
+int32_t parseSql(SRequestObj* pRequest, bool topicQuery, SQuery** pQuery);
+int32_t getPlan(SRequestObj* pRequest, SQuery* pQuery, SQueryPlan** pPlan, SArray* pNodeList);
 
 int32_t buildRequest(STscObj* pTscObj, const char* sql, int sqlLen, SRequestObj** pRequest);
 
-int32_t parseSql(SRequestObj* pRequest, bool topicQuery, SQuery** pQuery);
-int32_t getPlan(SRequestObj* pRequest, SQuery* pQuery, SQueryPlan** pPlan, SArray* pNodeList);
+void*   doFetchRow(SRequestObj* pRequest, bool setupOneRowPtr, bool convertUcs4);
+void    doSetOneRowPtr(SReqResultInfo* pResultInfo);
+int32_t setResultDataPtr(SReqResultInfo* pResultInfo, TAOS_FIELD* pFields, int32_t numOfCols, int32_t numOfRows,
+                         bool convertUcs4);
+void    setResSchemaInfo(SReqResultInfo* pResInfo, const SSchema* pSchema, int32_t numOfCols);
+int32_t setQueryResultFromRsp(SReqResultInfo* pResultInfo, const SRetrieveTableRsp* pRsp, bool convertUcs4);
 
 // --- heartbeat
 // global, called by mgmt

source/client/src/clientEnv.c

@@ -149,6 +149,7 @@ void *createRequest(STscObj *pObj, __taos_async_fn_t fp, void *param, int32_t ty
     return NULL;
   }
 
+  pRequest->resType = RES_TYPE__QUERY;
   pRequest->pDb = getDbOfConnection(pObj);
   pRequest->requestId = generateRequestId();
   pRequest->metric.start = taosGetTimestampUs();

source/client/src/clientImpl.c

@@ -13,7 +13,6 @@
 static int32_t       initEpSetFromCfg(const char* firstEp, const char* secondEp, SCorEpSet* pEpSet);
 static SMsgSendInfo* buildConnectMsg(SRequestObj* pRequest);
 static void          destroySendMsgInfo(SMsgSendInfo* pMsgBody);
-static int32_t       setQueryResultFromRsp(SReqResultInfo* pResultInfo, const SRetrieveTableRsp* pRsp, bool convertUcs4);
 
 static bool stringLengthCheck(const char* str, size_t maxsize) {
   if (str == NULL) {
@@ -42,7 +41,6 @@ static char* getClusterKey(const char* user, const char* auth, const char* ip, i
 
 static STscObj* taosConnectImpl(const char* user, const char* auth, const char* db, __taos_async_fn_t fp, void* param,
                                 SAppInstInfo* pAppInfo);
-static void     setResSchemaInfo(SReqResultInfo* pResInfo, const SSchema* pSchema, int32_t numOfCols);
 
 TAOS* taos_connect_internal(const char* ip, const char* user, const char* pass, const char* auth, const char* db,
                             uint16_t port) {
@@ -174,7 +172,7 @@ int32_t parseSql(SRequestObj* pRequest, bool topicQuery, SQuery** pQuery) {
 
 int32_t execLocalCmd(SRequestObj* pRequest, SQuery* pQuery) {
   SRetrieveTableRsp* pRsp = NULL;
-  int32_t code = qExecCommand(pQuery->pRoot, &pRsp);
+  int32_t            code = qExecCommand(pQuery->pRoot, &pRsp);
   if (TSDB_CODE_SUCCESS == code && NULL != pRsp) {
     code = setQueryResultFromRsp(&pRequest->body.resInfo, pRsp, false);
   }
@@ -190,7 +188,7 @@ int32_t execDdlQuery(SRequestObj* pRequest, SQuery* pQuery) {
   SCmdMsgInfo* pMsgInfo = pQuery->pCmdMsg;
   pRequest->type = pMsgInfo->msgType;
   pRequest->body.requestMsg = (SDataBuf){.pData = pMsgInfo->pMsg, .len = pMsgInfo->msgLen, .handle = NULL};
-  pMsgInfo->pMsg = NULL; // pMsg transferred to SMsgSendInfo management
+  pMsgInfo->pMsg = NULL;  // pMsg transferred to SMsgSendInfo management
 
   STscObj*      pTscObj = pRequest->pTscObj;
   SMsgSendInfo* pSendMsg = buildMsgInfoImpl(pRequest);
@@ -211,14 +209,12 @@ int32_t execDdlQuery(SRequestObj* pRequest, SQuery* pQuery) {
 
 int32_t getPlan(SRequestObj* pRequest, SQuery* pQuery, SQueryPlan** pPlan, SArray* pNodeList) {
   pRequest->type = pQuery->msgType;
-  SPlanContext cxt = {
-    .queryId = pRequest->requestId,
-    .acctId = pRequest->pTscObj->acctId,
-    .mgmtEpSet = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp),
-    .pAstRoot = pQuery->pRoot,
-    .showRewrite = pQuery->showRewrite
-  };
-  int32_t  code = qCreateQueryPlan(&cxt, pPlan, pNodeList);
+  SPlanContext cxt = {.queryId = pRequest->requestId,
+                      .acctId = pRequest->pTscObj->acctId,
+                      .mgmtEpSet = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp),
+                      .pAstRoot = pQuery->pRoot,
+                      .showRewrite = pQuery->showRewrite};
+  int32_t      code = qCreateQueryPlan(&cxt, pPlan, pNodeList);
   if (code != 0) {
     return code;
   }
@@ -234,10 +230,10 @@ void setResSchemaInfo(SReqResultInfo* pResInfo, const SSchema* pSchema, int32_t
 
   for (int32_t i = 0; i < pResInfo->numOfCols; ++i) {
     pResInfo->fields[i].bytes = pSchema[i].bytes;
-    pResInfo->fields[i].type  = pSchema[i].type;
+    pResInfo->fields[i].type = pSchema[i].type;
 
     pResInfo->userFields[i].bytes = pSchema[i].bytes;
-    pResInfo->userFields[i].type  = pSchema[i].type;
+    pResInfo->userFields[i].type = pSchema[i].type;
 
     if (pSchema[i].type == TSDB_DATA_TYPE_VARCHAR) {
       pResInfo->userFields[i].bytes -= VARSTR_HEADER_SIZE;
@@ -254,7 +250,8 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList
   void* pTransporter = pRequest->pTscObj->pAppInfo->pTransporter;
 
   SQueryResult res = {.code = 0, .numOfRows = 0, .msgSize = ERROR_MSG_BUF_DEFAULT_SIZE, .msg = pRequest->msgBuf};
-  int32_t      code = schedulerExecJob(pTransporter, pNodeList, pDag, &pRequest->body.queryJob, pRequest->sqlstr, pRequest->metric.start, &res);
+  int32_t      code = schedulerExecJob(pTransporter, pNodeList, pDag, &pRequest->body.queryJob, pRequest->sqlstr,
+                                       pRequest->metric.start, &res);
   if (code != TSDB_CODE_SUCCESS) {
     if (pRequest->body.queryJob != 0) {
       schedulerFreeJob(pRequest->body.queryJob);
@@ -274,14 +271,14 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList
   }
 
   pRequest->code = res.code;
-  terrno = res.code;  
+  terrno = res.code;
   return pRequest->code;
 }
 
 SRequestObj* execQueryImpl(STscObj* pTscObj, const char* sql, int sqlLen) {
   SRequestObj* pRequest = NULL;
-  SQuery* pQuery = NULL;
-  SArray* pNodeList = taosArrayInit(4, sizeof(struct SQueryNodeAddr));
+  SQuery*      pQuery = NULL;
+  SArray*      pNodeList = taosArrayInit(4, sizeof(struct SQueryNodeAddr));
 
   int32_t code = buildRequest(pTscObj, sql, sqlLen, &pRequest);
   if (TSDB_CODE_SUCCESS == code) {
@@ -320,15 +317,15 @@ SRequestObj* execQueryImpl(STscObj* pTscObj, const char* sql, int sqlLen) {
 }
 
 int32_t refreshMeta(STscObj* pTscObj, SRequestObj* pRequest) {
-  SCatalog *pCatalog = NULL;
-  int32_t code = 0;
-  int32_t dbNum = taosArrayGetSize(pRequest->dbList);
-  int32_t tblNum = taosArrayGetSize(pRequest->tableList);
+  SCatalog* pCatalog = NULL;
+  int32_t   code = 0;
+  int32_t   dbNum = taosArrayGetSize(pRequest->dbList);
+  int32_t   tblNum = taosArrayGetSize(pRequest->tableList);
 
   if (dbNum <= 0 && tblNum <= 0) {
     return TSDB_CODE_QRY_APP_ERROR;
   }
-  
+
   code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCatalog);
   if (code != TSDB_CODE_SUCCESS) {
     return code;
@@ -337,8 +334,8 @@ int32_t refreshMeta(STscObj* pTscObj, SRequestObj* pRequest) {
   SEpSet epset = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);
 
   for (int32_t i = 0; i < dbNum; ++i) {
-    char *dbFName = taosArrayGet(pRequest->dbList, i);
-    
+    char* dbFName = taosArrayGet(pRequest->dbList, i);
+
     code = catalogRefreshDBVgInfo(pCatalog, pTscObj->pAppInfo->pTransporter, &epset, dbFName);
     if (code != TSDB_CODE_SUCCESS) {
       return code;
@@ -346,7 +343,7 @@ int32_t refreshMeta(STscObj* pTscObj, SRequestObj* pRequest) {
   }
 
   for (int32_t i = 0; i < tblNum; ++i) {
-    SName *tableName = taosArrayGet(pRequest->tableList, i);
+    SName* tableName = taosArrayGet(pRequest->tableList, i);
 
     code = catalogRefreshTableMeta(pCatalog, pTscObj->pAppInfo->pTransporter, &epset, tableName, -1);
     if (code != TSDB_CODE_SUCCESS) {
@@ -357,11 +354,10 @@ int32_t refreshMeta(STscObj* pTscObj, SRequestObj* pRequest) {
   return code;
 }
 
-
 SRequestObj* execQuery(STscObj* pTscObj, const char* sql, int sqlLen) {
   SRequestObj* pRequest = NULL;
-  int32_t retryNum = 0;
-  int32_t code = 0;
+  int32_t      retryNum = 0;
+  int32_t      code = 0;
 
   while (retryNum++ < REQUEST_MAX_TRY_TIMES) {
     pRequest = execQueryImpl(pTscObj, sql, sqlLen);
@@ -377,7 +373,7 @@ SRequestObj* execQuery(STscObj* pTscObj, const char* sql, int sqlLen) {
 
     destroyRequest(pRequest);
   }
-  
+
   return pRequest;
 }
 
@@ -509,7 +505,8 @@ static void destroySendMsgInfo(SMsgSendInfo* pMsgBody) {
 }
 
 bool persistConnForSpecificMsg(void* parenct, tmsg_t msgType) {
-  return msgType == TDMT_VND_QUERY_RSP || msgType == TDMT_VND_FETCH_RSP || msgType == TDMT_VND_RES_READY_RSP || msgType == TDMT_VND_QUERY_HEARTBEAT_RSP;
+  return msgType == TDMT_VND_QUERY_RSP || msgType == TDMT_VND_FETCH_RSP || msgType == TDMT_VND_RES_READY_RSP ||
+         msgType == TDMT_VND_QUERY_HEARTBEAT_RSP;
 }
 
 void processMsgFromServer(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet) {
@@ -536,10 +533,10 @@ void processMsgFromServer(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet) {
     int32_t elapsed = pRequest->metric.rsp - pRequest->metric.start;
     if (pMsg->code == TSDB_CODE_SUCCESS) {
       tscDebug("0x%" PRIx64 " message:%s, code:%s rspLen:%d, elapsed:%d ms, reqId:0x%" PRIx64, pRequest->self,
-               TMSG_INFO(pMsg->msgType), tstrerror(pMsg->code), pMsg->contLen, elapsed/1000, pRequest->requestId);
+               TMSG_INFO(pMsg->msgType), tstrerror(pMsg->code), pMsg->contLen, elapsed / 1000, pRequest->requestId);
     } else {
       tscError("0x%" PRIx64 " SQL cmd:%s, code:%s rspLen:%d, elapsed time:%d ms, reqId:0x%" PRIx64, pRequest->self,
-               TMSG_INFO(pMsg->msgType), tstrerror(pMsg->code), pMsg->contLen, elapsed/1000, pRequest->requestId);
+               TMSG_INFO(pMsg->msgType), tstrerror(pMsg->code), pMsg->contLen, elapsed / 1000, pRequest->requestId);
     }
 
     taosReleaseRef(clientReqRefPool, pSendInfo->requestObjRefId);
@@ -590,7 +587,7 @@ TAOS* taos_connect_l(const char* ip, int ipLen, const char* user, int userLen, c
   return taos_connect(ipStr, userStr, passStr, dbStr, port);
 }
 
-static void doSetOneRowPtr(SReqResultInfo* pResultInfo) {
+void doSetOneRowPtr(SReqResultInfo* pResultInfo) {
   for (int32_t i = 0; i < pResultInfo->numOfCols; ++i) {
     SResultColumn* pCol = &pResultInfo->pCol[i];
 
@@ -722,8 +719,8 @@ _return:
 
 static int32_t doPrepareResPtr(SReqResultInfo* pResInfo) {
   if (pResInfo->row == NULL) {
-    pResInfo->row    = taosMemoryCalloc(pResInfo->numOfCols, POINTER_BYTES);
-    pResInfo->pCol   = taosMemoryCalloc(pResInfo->numOfCols, sizeof(SResultColumn));
+    pResInfo->row = taosMemoryCalloc(pResInfo->numOfCols, POINTER_BYTES);
+    pResInfo->pCol = taosMemoryCalloc(pResInfo->numOfCols, sizeof(SResultColumn));
     pResInfo->length = taosMemoryCalloc(pResInfo->numOfCols, sizeof(int32_t));
     pResInfo->convertBuf = taosMemoryCalloc(pResInfo->numOfCols, POINTER_BYTES);
 
@@ -740,7 +737,7 @@ static int32_t doConvertUCS4(SReqResultInfo* pResultInfo, int32_t numOfRows, int
     int32_t type = pResultInfo->fields[i].type;
     int32_t bytes = pResultInfo->fields[i].bytes;
 
-    if (type == TSDB_DATA_TYPE_NCHAR) {
+    if (type == TSDB_DATA_TYPE_NCHAR && colLength[i] > 0) {
       char* p = taosMemoryRealloc(pResultInfo->convertBuf[i], colLength[i]);
       if (p == NULL) {
         return TSDB_CODE_OUT_OF_MEMORY;
@@ -770,7 +767,8 @@ static int32_t doConvertUCS4(SReqResultInfo* pResultInfo, int32_t numOfRows, int
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t setResultDataPtr(SReqResultInfo* pResultInfo, TAOS_FIELD* pFields, int32_t numOfCols, int32_t numOfRows, bool convertUcs4) {
+int32_t setResultDataPtr(SReqResultInfo* pResultInfo, TAOS_FIELD* pFields, int32_t numOfCols, int32_t numOfRows,
+                         bool convertUcs4) {
   assert(numOfCols > 0 && pFields != NULL && pResultInfo != NULL);
   if (numOfRows == 0) {
     return TSDB_CODE_SUCCESS;
@@ -841,15 +839,16 @@ void resetConnectDB(STscObj* pTscObj) {
 int32_t setQueryResultFromRsp(SReqResultInfo* pResultInfo, const SRetrieveTableRsp* pRsp, bool convertUcs4) {
   assert(pResultInfo != NULL && pRsp != NULL);
 
-  pResultInfo->pRspMsg    = (const char*)pRsp;
-  pResultInfo->pData      = (void*)pRsp->data;
-  pResultInfo->numOfRows  = htonl(pRsp->numOfRows);
-  pResultInfo->current    = 0;
-  pResultInfo->completed  = (pRsp->completed == 1);
+  pResultInfo->pRspMsg = (const char*)pRsp;
+  pResultInfo->pData = (void*)pRsp->data;
+  pResultInfo->numOfRows = htonl(pRsp->numOfRows);
+  pResultInfo->current = 0;
+  pResultInfo->completed = (pRsp->completed == 1);
   pResultInfo->payloadLen = htonl(pRsp->compLen);
-  pResultInfo->precision  = pRsp->precision;
+  pResultInfo->precision = pRsp->precision;
 
   // TODO handle the compressed case
   pResultInfo->totalRows += pResultInfo->numOfRows;
-  return setResultDataPtr(pResultInfo, pResultInfo->fields, pResultInfo->numOfCols, pResultInfo->numOfRows, convertUcs4);
+  return setResultDataPtr(pResultInfo, pResultInfo->fields, pResultInfo->numOfCols, pResultInfo->numOfRows,
+                          convertUcs4);
 }

source/client/src/clientMain.c

@@ -71,7 +71,7 @@ void taos_cleanup(void) {
   tscInfo("all local resources released");
 }
 
-setConfRet   taos_set_config(const char *config) {
+setConfRet taos_set_config(const char *config) {
   // TODO
   setConfRet ret = {SET_CONF_RET_SUCC, {0}};
   return ret;
@@ -133,8 +133,7 @@ int taos_field_count(TAOS_RES *res) {
     return 0;
   }
 
-  SRequestObj    *pRequest = (SRequestObj *)res;
-  SReqResultInfo *pResInfo = &pRequest->body.resInfo;
+  SReqResultInfo *pResInfo = tscGetCurResInfo(res);
   return pResInfo->numOfCols;
 }
 
@@ -145,7 +144,7 @@ TAOS_FIELD *taos_fetch_fields(TAOS_RES *res) {
     return NULL;
   }
 
-  SReqResultInfo *pResInfo = &(((SRequestObj *)res)->body.resInfo);
+  SReqResultInfo *pResInfo = tscGetCurResInfo(res);
   return pResInfo->userFields;
 }
 
@@ -162,13 +161,36 @@ TAOS_ROW taos_fetch_row(TAOS_RES *res) {
     return NULL;
   }
 
-  SRequestObj *pRequest = (SRequestObj *)res;
-  if (pRequest->type == TSDB_SQL_RETRIEVE_EMPTY_RESULT || pRequest->type == TSDB_SQL_INSERT ||
-      pRequest->code != TSDB_CODE_SUCCESS || taos_num_fields(res) == 0) {
-    return NULL;
-  }
+  if (TD_RES_QUERY(res)) {
+    SRequestObj *pRequest = (SRequestObj *)res;
+    if (pRequest->type == TSDB_SQL_RETRIEVE_EMPTY_RESULT || pRequest->type == TSDB_SQL_INSERT ||
+        pRequest->code != TSDB_CODE_SUCCESS || taos_num_fields(res) == 0) {
+      return NULL;
+    }
+
+    return doFetchRow(pRequest, true, true);
+
+  } else if (TD_RES_TMQ(res)) {
+    SMqRspObj      *msg = ((SMqRspObj *)res);
+    SReqResultInfo *pResultInfo = taosArrayGet(msg->res, msg->resIter);
+
+    doSetOneRowPtr(pResultInfo);
+    pResultInfo->current += 1;
 
-  return doFetchRow(pRequest, true, true);
+    if (pResultInfo->row == NULL) {
+      msg->resIter++;
+      pResultInfo = taosArrayGet(msg->res, msg->resIter);
+      doSetOneRowPtr(pResultInfo);
+      pResultInfo->current += 1;
+    }
+
+    return pResultInfo->row;
+
+  } else {
+    // assert to avoid uninitialization error
+    ASSERT(0);
+  }
+  return NULL;
 }
 
 int taos_print_row(char *str, TAOS_ROW row, TAOS_FIELD *fields, int num_fields) {
@@ -260,12 +282,12 @@ int *taos_fetch_lengths(TAOS_RES *res) {
     return NULL;
   }
 
-  return ((SRequestObj *)res)->body.resInfo.length;
+  SReqResultInfo *pResInfo = tscGetCurResInfo(res);
+  return pResInfo->length;
 }
 
 TAOS_ROW *taos_result_block(TAOS_RES *res) {
-  SRequestObj* pRequest = (SRequestObj*) res;
-  if (pRequest == NULL) {
+  if (res == NULL) {
     terrno = TSDB_CODE_INVALID_PARA;
     return NULL;
   }
@@ -274,7 +296,8 @@ TAOS_ROW *taos_result_block(TAOS_RES *res) {
     return NULL;
   }
 
-  return &pRequest->body.resInfo.row;
+  SReqResultInfo *pResInfo = tscGetCurResInfo(res);
+  return &pResInfo->row;
 }
 
 // todo intergrate with tDataTypes
@@ -313,7 +336,7 @@ const char *taos_data_type(int type) {
 const char *taos_get_client_info() { return version; }
 
 int taos_affected_rows(TAOS_RES *res) {
-  if (res == NULL) {
+  if (res == NULL || TD_RES_TMQ(res)) {
     return 0;
   }
 
@@ -323,12 +346,17 @@ int taos_affected_rows(TAOS_RES *res) {
 }
 
 int taos_result_precision(TAOS_RES *res) {
-  SRequestObj* pRequest = (SRequestObj*) res;
-  if (pRequest == NULL) {
+  if (res == NULL) {
     return TSDB_TIME_PRECISION_MILLI;
   }
-
-  return pRequest->body.resInfo.precision;
+  if (TD_RES_QUERY(res)) {
+    SRequestObj *pRequest = (SRequestObj *)res;
+    return pRequest->body.resInfo.precision;
+  } else if (TD_RES_TMQ(res)) {
+    SReqResultInfo *info = tmqGetCurResInfo(res);
+    return info->precision;
+  }
+  return TSDB_TIME_PRECISION_MILLI;
 }
 
 int taos_select_db(TAOS *taos, const char *db) {
@@ -370,90 +398,115 @@ void taos_stop_query(TAOS_RES *res) {
 }
 
 bool taos_is_null(TAOS_RES *res, int32_t row, int32_t col) {
-  SRequestObj    *pRequestObj = res;
-  SReqResultInfo *pResultInfo = &pRequestObj->body.resInfo;
+  SReqResultInfo *pResultInfo = tscGetCurResInfo(res);
   if (col >= pResultInfo->numOfCols || col < 0 || row >= pResultInfo->numOfRows || row < 0) {
     return true;
   }
 
-  SResultColumn *pCol = &pRequestObj->body.resInfo.pCol[col];
+  SResultColumn *pCol = &pResultInfo->pCol[col];
   return colDataIsNull_f(pCol->nullbitmap, row);
 }
 
-bool taos_is_update_query(TAOS_RES *res) {
-  return taos_num_fields(res) == 0;
-}
+bool taos_is_update_query(TAOS_RES *res) { return taos_num_fields(res) == 0; }
 
 int taos_fetch_block(TAOS_RES *res, TAOS_ROW *rows) {
   int32_t numOfRows = 0;
-  /*int32_t code = */taos_fetch_block_s(res, &numOfRows, rows);
+  /*int32_t code = */ taos_fetch_block_s(res, &numOfRows, rows);
   return numOfRows;
 }
 
-int taos_fetch_block_s(TAOS_RES *res, int* numOfRows, TAOS_ROW *rows) {
-  SRequestObj *pRequest = (SRequestObj *)res;
-  if (pRequest == NULL) {
+int taos_fetch_block_s(TAOS_RES *res, int *numOfRows, TAOS_ROW *rows) {
+  if (res == NULL) {
     return 0;
   }
+  if (TD_RES_QUERY(res)) {
+    SRequestObj *pRequest = (SRequestObj *)res;
 
-  (*rows)      = NULL;
-  (*numOfRows) = 0;
+    (*rows) = NULL;
+    (*numOfRows) = 0;
 
-  if (pRequest->type == TSDB_SQL_RETRIEVE_EMPTY_RESULT || pRequest->type == TSDB_SQL_INSERT ||
-      pRequest->code != TSDB_CODE_SUCCESS || taos_num_fields(res) == 0) {
-    return 0;
-  }
+    if (pRequest->type == TSDB_SQL_RETRIEVE_EMPTY_RESULT || pRequest->type == TSDB_SQL_INSERT ||
+        pRequest->code != TSDB_CODE_SUCCESS || taos_num_fields(res) == 0) {
+      return 0;
+    }
 
-  doFetchRow(pRequest, false, true);
+    doFetchRow(pRequest, false, true);
 
-  // TODO refactor
-  SReqResultInfo *pResultInfo = &pRequest->body.resInfo;
-  pResultInfo->current = pResultInfo->numOfRows;
+    // TODO refactor
+    SReqResultInfo *pResultInfo = &pRequest->body.resInfo;
+    pResultInfo->current = pResultInfo->numOfRows;
 
-  (*rows)      = pResultInfo->row;
-  (*numOfRows) = pResultInfo->numOfRows;
-  return pRequest->code;
-}
+    (*rows) = pResultInfo->row;
+    (*numOfRows) = pResultInfo->numOfRows;
+    return pRequest->code;
+  } else if (TD_RES_TMQ(res)) {
+    SReqResultInfo *pResultInfo = tmqGetNextResInfo(res);
+    if (pResultInfo == NULL) return -1;
 
-int taos_fetch_raw_block(TAOS_RES *res, int* numOfRows, void** pData) {
-  SRequestObj *pRequest = (SRequestObj *)res;
-  if (pRequest == NULL) {
+    pResultInfo->current = pResultInfo->numOfRows;
+    (*rows) = pResultInfo->row;
+    (*numOfRows) = pResultInfo->numOfRows;
     return 0;
+  } else {
+    ASSERT(0);
+    return -1;
   }
+}
 
-  if (pRequest->type == TSDB_SQL_RETRIEVE_EMPTY_RESULT || pRequest->type == TSDB_SQL_INSERT ||
-      pRequest->code != TSDB_CODE_SUCCESS || taos_num_fields(res) == 0) {
+int taos_fetch_raw_block(TAOS_RES *res, int *numOfRows, void **pData) {
+  if (res == NULL) {
     return 0;
   }
+  if (TD_RES_QUERY(res)) {
+    SRequestObj *pRequest = (SRequestObj *)res;
 
-  doFetchRow(pRequest, false, false);
+    if (pRequest->type == TSDB_SQL_RETRIEVE_EMPTY_RESULT || pRequest->type == TSDB_SQL_INSERT ||
+        pRequest->code != TSDB_CODE_SUCCESS || taos_num_fields(res) == 0) {
+      return 0;
+    }
+
+    doFetchRow(pRequest, false, false);
+
+    SReqResultInfo *pResultInfo = &pRequest->body.resInfo;
 
-  SReqResultInfo *pResultInfo = &pRequest->body.resInfo;
+    pResultInfo->current = pResultInfo->numOfRows;
+    (*numOfRows) = pResultInfo->numOfRows;
+    (*pData) = (void *)pResultInfo->pData;
+
+    return 0;
 
-  pResultInfo->current = pResultInfo->numOfRows;
-  (*numOfRows) = pResultInfo->numOfRows;
-  (*pData) = (void*) pResultInfo->pData;
+  } else if (TD_RES_TMQ(res)) {
+    SReqResultInfo *pResultInfo = tmqGetNextResInfo(res);
+    if (pResultInfo == NULL) return -1;
 
-  return 0;
+    pResultInfo->current = pResultInfo->numOfRows;
+    (*numOfRows) = pResultInfo->numOfRows;
+    (*pData) = (void *)pResultInfo->pData;
+    return 0;
+
+  } else {
+    ASSERT(0);
+    return -1;
+  }
 }
 
 int *taos_get_column_data_offset(TAOS_RES *res, int columnIndex) {
-  SRequestObj *pRequest = (SRequestObj *)res;
-  if (pRequest == NULL) {
+  if (res == NULL) {
     return 0;
   }
 
-  int32_t numOfFields = taos_num_fields(pRequest);
+  int32_t numOfFields = taos_num_fields(res);
   if (columnIndex < 0 || columnIndex >= numOfFields || numOfFields == 0) {
     return 0;
   }
 
-  TAOS_FIELD* pField = &pRequest->body.resInfo.userFields[columnIndex];
+  SReqResultInfo *pResInfo = tscGetCurResInfo(res);
+  TAOS_FIELD     *pField = &pResInfo->userFields[columnIndex];
   if (!IS_VAR_DATA_TYPE(pField->type)) {
     return 0;
   }
 
-  return pRequest->body.resInfo.pCol[columnIndex].offset;
+  return pResInfo->pCol[columnIndex].offset;
 }
 
 int taos_validate_sql(TAOS *taos, const char *sql) { return true; }
@@ -483,18 +536,19 @@ void taos_fetch_rows_a(TAOS_RES *res, __taos_async_fn_t fp, void *param) {
   // TODO
 }
 
-TAOS_SUB *taos_subscribe(TAOS *taos, int restart, const char* topic, const char *sql, TAOS_SUBSCRIBE_CALLBACK fp, void *param, int interval) {
-    // TODO
-    return NULL;
+TAOS_SUB *taos_subscribe(TAOS *taos, int restart, const char *topic, const char *sql, TAOS_SUBSCRIBE_CALLBACK fp,
+                         void *param, int interval) {
+  // TODO
+  return NULL;
 }
 
 TAOS_RES *taos_consume(TAOS_SUB *tsub) {
-    // TODO
-    return NULL;
+  // TODO
+  return NULL;
 }
 
 void taos_unsubscribe(TAOS_SUB *tsub, int keepProgress) {
-    // TODO
+  // TODO
 }
 
 int taos_load_table_info(TAOS *taos, const char *tableNameList) {
@@ -553,26 +607,26 @@ int taos_stmt_set_tbname(TAOS_STMT *stmt, const char *name) {
 }
 
 int taos_stmt_is_insert(TAOS_STMT *stmt, int *insert) {
-    // TODO
-    return -1;
+  // TODO
+  return -1;
 }
 
 int taos_stmt_num_params(TAOS_STMT *stmt, int *nums) {
-    // TODO
-    return -1;
+  // TODO
+  return -1;
 }
 
-int taos_stmt_add_batch(TAOS_STMT* stmt) {
-    // TODO
-    return -1;
+int taos_stmt_add_batch(TAOS_STMT *stmt) {
+  // TODO
+  return -1;
 }
 
 TAOS_RES *taos_stmt_use_result(TAOS_STMT *stmt) {
-    // TODO
-    return NULL;
+  // TODO
+  return NULL;
 }
 
-int taos_stmt_bind_param_batch(TAOS_STMT* stmt, TAOS_MULTI_BIND* bind) {
-    // TODO
-    return -1;
+int taos_stmt_bind_param_batch(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind) {
+  // TODO
+  return -1;
 }

source/client/src/tmq.c

@@ -17,25 +17,19 @@
 #include "clientLog.h"
 #include "parser.h"
 #include "planner.h"
-#include "scheduler.h"
 #include "tdatablock.h"
 #include "tdef.h"
 #include "tglobal.h"
 #include "tmsgtype.h"
-#include "tpagedbuf.h"
 #include "tqueue.h"
 #include "tref.h"
 
-typedef struct {
-  int32_t curBlock;
-  int32_t curRow;
-  void**  uData;
-} SMqRowIter;
-
 struct tmq_message_t {
   SMqPollRsp msg;
+  char*      topic;
   void*      vg;
-  SMqRowIter iter;
+  SArray*    res;  // SArray<SReqResultInfo>
+  int32_t    resIter;
 };
 
 struct tmq_list_t {
@@ -849,7 +843,8 @@ int32_t tmqPollCb(void* param, const SDataBuf* pMsg, int32_t code) {
   if (msgEpoch < tmqEpoch) {
     /*printf("discard rsp epoch %d, current epoch %d\n", msgEpoch, tmqEpoch);*/
     /*tsem_post(&tmq->rspSem);*/
-    tscWarn("msg discard from vg %d since from earlier epoch, rsp epoch %d, current epoch %d", pParam->vgId, msgEpoch, tmqEpoch);
+    tscWarn("msg discard from vg %d since from earlier epoch, rsp epoch %d, current epoch %d", pParam->vgId, msgEpoch,
+            tmqEpoch);
     return 0;
   }
 
@@ -886,8 +881,8 @@ int32_t tmqPollCb(void* param, const SDataBuf* pMsg, int32_t code) {
   }
   memcpy(pRsp, pMsg->pData, sizeof(SMqRspHead));
   tDecodeSMqPollRsp(POINTER_SHIFT(pMsg->pData, sizeof(SMqRspHead)), &pRsp->msg);
-  pRsp->iter.curBlock = 0;
-  pRsp->iter.curRow = 0;
+  /*pRsp->iter.curBlock = 0;*/
+  /*pRsp->iter.curRow = 0;*/
   // TODO: alloc mem
   /*pRsp->*/
   /*printf("rsp commit off:%ld rsp off:%ld has data:%d\n", pRsp->committedOffset, pRsp->rspOffset, pRsp->numOfTopics);*/
@@ -899,8 +894,8 @@ int32_t tmqPollCb(void* param, const SDataBuf* pMsg, int32_t code) {
   }
 #endif
 
-  tscDebug("consumer %ld recv poll: vg %d, req offset %ld, rsp offset %ld", tmq->consumerId, pParam->pVg->vgId, pRsp->msg.reqOffset,
-           pRsp->msg.rspOffset);
+  tscDebug("consumer %ld recv poll: vg %d, req offset %ld, rsp offset %ld", tmq->consumerId, pParam->pVg->vgId,
+           pRsp->msg.reqOffset, pRsp->msg.rspOffset);
 
   pRsp->vg = pParam->pVg;
   taosWriteQitem(tmq->mqueue, pRsp);
@@ -921,7 +916,8 @@ bool tmqUpdateEp(tmq_t* tmq, int32_t epoch, SMqCMGetSubEpRsp* pRsp) {
   bool    set = false;
   int32_t topicNumGet = taosArrayGetSize(pRsp->topics);
   char    vgKey[TSDB_TOPIC_FNAME_LEN + 22];
-  tscDebug("consumer %ld update ep epoch %d to epoch %d, topic num: %d", tmq->consumerId, tmq->epoch, epoch, topicNumGet);
+  tscDebug("consumer %ld update ep epoch %d to epoch %d, topic num: %d", tmq->consumerId, tmq->epoch, epoch,
+           topicNumGet);
   SArray* newTopics = taosArrayInit(topicNumGet, sizeof(SMqClientTopic));
   if (newTopics == NULL) {
     return false;
@@ -1289,7 +1285,8 @@ int32_t tmqPollImpl(tmq_t* tmq, int64_t blockingTime) {
       int64_t transporterId = 0;
       /*printf("send poll\n");*/
       atomic_add_fetch_32(&tmq->waitingRequest, 1);
-      tscDebug("consumer %ld send poll to %s : vg %d, epoch %d, req offset %ld, reqId %lu", tmq->consumerId, pTopic->topicName, pVg->vgId, tmq->epoch, pVg->currentOffset, pReq->reqId);
+      tscDebug("consumer %ld send poll to %s : vg %d, epoch %d, req offset %ld, reqId %lu", tmq->consumerId,
+               pTopic->topicName, pVg->vgId, tmq->epoch, pVg->currentOffset, pReq->reqId);
       /*printf("send vg %d %ld\n", pVg->vgId, pVg->currentOffset);*/
       asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &pVg->epSet, &transporterId, sendInfo);
       pVg->pollCnt++;
@@ -1566,30 +1563,4 @@ const char* tmq_err2str(tmq_resp_err_t err) {
   return "fail";
 }
 
-TAOS_ROW tmq_get_row(tmq_message_t* message) {
-  SMqPollRsp* rsp = &message->msg;
-  while (1) {
-    if (message->iter.curBlock < taosArrayGetSize(rsp->pBlockData)) {
-      SSDataBlock* pBlock = taosArrayGet(rsp->pBlockData, message->iter.curBlock);
-      if (message->iter.curRow < pBlock->info.rows) {
-        for (int i = 0; i < pBlock->info.numOfCols; i++) {
-          SColumnInfoData* pData = taosArrayGet(pBlock->pDataBlock, i);
-          if (colDataIsNull_s(pData, message->iter.curRow))
-            message->iter.uData[i] = NULL;
-          else {
-            message->iter.uData[i] = colDataGetData(pData, message->iter.curRow);
-          }
-        }
-        message->iter.curRow++;
-        return message->iter.uData;
-      } else {
-        message->iter.curBlock++;
-        message->iter.curRow = 0;
-        continue;
-      }
-    }
-    return NULL;
-  }
-}
-
 char* tmq_get_topic_name(tmq_message_t* message) { return "not implemented yet"; }

source/common/src/tglobal.c

@@ -256,7 +256,7 @@ static int32_t taosLoadCfg(SConfig *pCfg, const char *inputCfgDir, const char *e
   return 0;
 }
 
-static int32_t taosAddClientLogCfg(SConfig *pCfg) {
+int32_t taosAddClientLogCfg(SConfig *pCfg) {
   if (cfgAddDir(pCfg, "configDir", configDir, 1) != 0) return -1;
   if (cfgAddDir(pCfg, "scriptDir", configDir, 1) != 0) return -1;
   if (cfgAddDir(pCfg, "logDir", tsLogDir, 1) != 0) return -1;
@@ -616,7 +616,7 @@ int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDi
 
   taosSetAllDebugFlag(cfgGetItem(pCfg, "debugFlag")->i32);
 
-  if (taosMkDir(tsLogDir) != 0) {
+  if (taosMulMkDir(tsLogDir) != 0) {
     uError("failed to create dir:%s since %s", tsLogDir, terrstr());
     cfgCleanup(pCfg);
     return -1;

source/common/src/tmsg.c

@@ -308,6 +308,7 @@ int32_t tSerializeSVCreateTbReq(void **buf, SVCreateTbReq *pReq) {
       tlen += taosEncodeFixedI16(buf, pReq->stbCfg.nTagCols);
       for (col_id_t i = 0; i < pReq->stbCfg.nTagCols; ++i) {
         tlen += taosEncodeFixedI8(buf, pReq->stbCfg.pTagSchema[i].type);
+        tlen += taosEncodeFixedI8(buf, pReq->stbCfg.pTagSchema[i].index);
         tlen += taosEncodeFixedI16(buf, pReq->stbCfg.pTagSchema[i].colId);
         tlen += taosEncodeFixedI32(buf, pReq->stbCfg.pTagSchema[i].bytes);
         tlen += taosEncodeString(buf, pReq->stbCfg.pTagSchema[i].name);
@@ -378,6 +379,7 @@ void *tDeserializeSVCreateTbReq(void *buf, SVCreateTbReq *pReq) {
       pReq->stbCfg.pTagSchema = (SSchema *)taosMemoryMalloc(pReq->stbCfg.nTagCols * sizeof(SSchema));
       for (col_id_t i = 0; i < pReq->stbCfg.nTagCols; ++i) {
         buf = taosDecodeFixedI8(buf, &(pReq->stbCfg.pTagSchema[i].type));
+        buf = taosDecodeFixedI8(buf, &(pReq->stbCfg.pTagSchema[i].index));
         buf = taosDecodeFixedI16(buf, &pReq->stbCfg.pTagSchema[i].colId);
         buf = taosDecodeFixedI32(buf, &pReq->stbCfg.pTagSchema[i].bytes);
         buf = taosDecodeStringTo(buf, pReq->stbCfg.pTagSchema[i].name);
@@ -1991,7 +1993,7 @@ void tFreeSUseDbBatchRsp(SUseDbBatchRsp *pRsp) {
   taosArrayDestroy(pRsp->pArray);
 }
 
-int32_t tSerializeSDbCfgReq(void* buf, int32_t bufLen, SDbCfgReq* pReq) {
+int32_t tSerializeSDbCfgReq(void *buf, int32_t bufLen, SDbCfgReq *pReq) {
   SCoder encoder = {0};
   tCoderInit(&encoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_ENCODER);
 
@@ -2004,7 +2006,7 @@ int32_t tSerializeSDbCfgReq(void* buf, int32_t bufLen, SDbCfgReq* pReq) {
   return tlen;
 }
 
-int32_t tDeserializeSDbCfgReq(void* buf, int32_t bufLen, SDbCfgReq* pReq) {
+int32_t tDeserializeSDbCfgReq(void *buf, int32_t bufLen, SDbCfgReq *pReq) {
   SCoder decoder = {0};
   tCoderInit(&decoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_DECODER);
 
@@ -2016,7 +2018,7 @@ int32_t tDeserializeSDbCfgReq(void* buf, int32_t bufLen, SDbCfgReq* pReq) {
   return 0;
 }
 
-int32_t tSerializeSDbCfgRsp(void* buf, int32_t bufLen, const SDbCfgRsp* pRsp) {
+int32_t tSerializeSDbCfgRsp(void *buf, int32_t bufLen, const SDbCfgRsp *pRsp) {
   SCoder encoder = {0};
   tCoderInit(&encoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_ENCODER);
 
@@ -2047,7 +2049,7 @@ int32_t tSerializeSDbCfgRsp(void* buf, int32_t bufLen, const SDbCfgRsp* pRsp) {
   return tlen;
 }
 
-int32_t tDeserializeSDbCfgRsp(void* buf, int32_t bufLen, SDbCfgRsp* pRsp) {
+int32_t tDeserializeSDbCfgRsp(void *buf, int32_t bufLen, SDbCfgRsp *pRsp) {
   SCoder decoder = {0};
   tCoderInit(&decoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_DECODER);
 
@@ -2078,7 +2080,7 @@ int32_t tDeserializeSDbCfgRsp(void* buf, int32_t bufLen, SDbCfgRsp* pRsp) {
   return 0;
 }
 
-int32_t tSerializeSUserIndexReq(void* buf, int32_t bufLen, SUserIndexReq* pReq) {
+int32_t tSerializeSUserIndexReq(void *buf, int32_t bufLen, SUserIndexReq *pReq) {
   SCoder encoder = {0};
   tCoderInit(&encoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_ENCODER);
 
@@ -2091,7 +2093,7 @@ int32_t tSerializeSUserIndexReq(void* buf, int32_t bufLen, SUserIndexReq* pReq)
   return tlen;
 }
 
-int32_t tDeserializeSUserIndexReq(void* buf, int32_t bufLen, SUserIndexReq* pReq) {
+int32_t tDeserializeSUserIndexReq(void *buf, int32_t bufLen, SUserIndexReq *pReq) {
   SCoder decoder = {0};
   tCoderInit(&decoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_DECODER);
 
@@ -2103,7 +2105,7 @@ int32_t tDeserializeSUserIndexReq(void* buf, int32_t bufLen, SUserIndexReq* pReq
   return 0;
 }
 
-int32_t tSerializeSUserIndexRsp(void* buf, int32_t bufLen, const SUserIndexRsp* pRsp) {
+int32_t tSerializeSUserIndexRsp(void *buf, int32_t bufLen, const SUserIndexRsp *pRsp) {
   SCoder encoder = {0};
   tCoderInit(&encoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_ENCODER);
 
@@ -2120,7 +2122,7 @@ int32_t tSerializeSUserIndexRsp(void* buf, int32_t bufLen, const SUserIndexRsp*
   return tlen;
 }
 
-int32_t tDeserializeSUserIndexRsp(void* buf, int32_t bufLen, SUserIndexRsp* pRsp) {
+int32_t tDeserializeSUserIndexRsp(void *buf, int32_t bufLen, SUserIndexRsp *pRsp) {
   SCoder decoder = {0};
   tCoderInit(&decoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_DECODER);
 
@@ -2136,7 +2138,6 @@ int32_t tDeserializeSUserIndexRsp(void* buf, int32_t bufLen, SUserIndexRsp* pRsp
   return 0;
 }
 
-
 int32_t tSerializeSShowReq(void *buf, int32_t bufLen, SShowReq *pReq) {
   SCoder encoder = {0};
   tCoderInit(&encoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_ENCODER);

source/common/src/ttime.c

@@ -406,7 +406,31 @@ int64_t convertTimeFromPrecisionToUnit(int64_t time, int32_t fromPrecision, char
     default: {
       return -1;
     }
-  }  
+  }
+}
+
+int32_t convertStringToTimestamp(int16_t type, char *inputData, int64_t timePrec, int64_t *timeVal) {
+  int32_t charLen = varDataLen(inputData);
+  char *newColData;
+  if (type == TSDB_DATA_TYPE_BINARY) {
+    newColData = taosMemoryCalloc(1,  charLen + 1);
+    memcpy(newColData, varDataVal(inputData), charLen);
+    taosParseTime(newColData, timeVal, charLen, (int32_t)timePrec, 0);
+    taosMemoryFree(newColData);
+  } else if (type == TSDB_DATA_TYPE_NCHAR) {
+    newColData = taosMemoryCalloc(1,  charLen / TSDB_NCHAR_SIZE + 1);
+    int len = taosUcs4ToMbs((TdUcs4 *)varDataVal(inputData), charLen, newColData);
+    if (len < 0){
+      taosMemoryFree(newColData);
+      return TSDB_CODE_FAILED;
+    }
+    newColData[len] = 0;
+    taosParseTime(newColData, timeVal, len + 1, (int32_t)timePrec, 0);
+    taosMemoryFree(newColData);
+  } else {
+    return TSDB_CODE_FAILED;
+  }
+  return TSDB_CODE_SUCCESS;
 }
 
 static int32_t getDuration(int64_t val, char unit, int64_t* result, int32_t timePrecision) {

source/common/src/tvariant.c

@@ -1028,13 +1028,18 @@ int32_t taosVariantTypeSetType(SVariant *pVariant, char type) {
 
 char * taosVariantGet(SVariant *pVar, int32_t type) {
   switch (type) {
-    case TSDB_DATA_TYPE_BOOL: 
+    case TSDB_DATA_TYPE_BOOL:
     case TSDB_DATA_TYPE_TINYINT:
     case TSDB_DATA_TYPE_SMALLINT:
+    case TSDB_DATA_TYPE_INT:
     case TSDB_DATA_TYPE_BIGINT:
-    case TSDB_DATA_TYPE_INT: 
     case TSDB_DATA_TYPE_TIMESTAMP:
       return (char *)&pVar->i;
+    case TSDB_DATA_TYPE_UTINYINT:
+    case TSDB_DATA_TYPE_USMALLINT:
+    case TSDB_DATA_TYPE_UINT:
+    case TSDB_DATA_TYPE_UBIGINT:
+      return (char *)&pVar->u;
     case TSDB_DATA_TYPE_DOUBLE:
     case TSDB_DATA_TYPE_FLOAT:
       return (char *)&pVar->d;
@@ -1042,7 +1047,7 @@ char * taosVariantGet(SVariant *pVar, int32_t type) {
       return (char *)pVar->pz;
     case TSDB_DATA_TYPE_NCHAR:
       return (char *)pVar->ucs4;
-    default: 
+    default:
       return NULL;
   }
 

source/dnode/mgmt/exe/dmMain.c

@@ -16,6 +16,7 @@
 #define _DEFAULT_SOURCE
 #include "dmImp.h"
 #include "tconfig.h"
+#include "tgrant.h"
 
 static struct {
   bool         dumpConfig;
@@ -89,10 +90,7 @@ static int32_t dmParseArgs(int32_t argc, char const *argv[]) {
   return 0;
 }
 
-static void dmGenerateGrant() {
-  // grantParseParameter();
-  printf("this feature is not implemented yet\n");
-}
+static void dmGenerateGrant() { grantParseParameter(); }
 
 static void dmPrintVersion() {
 #ifdef TD_ENTERPRISE

source/dnode/mgmt/implement/CMakeLists.txt

@@ -6,4 +6,12 @@ target_link_libraries(
 target_include_directories(
     dnode
     PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
-)
\ No newline at end of file
+)
+
+IF (TD_GRANT)
+  TARGET_LINK_LIBRARIES(dnode grant)
+ENDIF ()
+IF (TD_USB_DONGLE)
+  TARGET_LINK_LIBRARIES(dnode usb_dongle)
+else()
+ENDIF ()
\ No newline at end of file

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

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#define _DEFAULT_SOURCE
+#ifndef _GRANT
+#include "os.h"
+#include "taoserror.h"
+#include "tgrant.h"
+#include "mndInt.h"
+
+int32_t grantInit() { return TSDB_CODE_SUCCESS; }
+void    grantCleanUp() {}
+void    grantParseParameter() { mError("can't parsed parameter k"); }
+int32_t grantCheck(EGrantType grant) { return TSDB_CODE_SUCCESS; }
+void    grantReset(EGrantType grant, uint64_t value) {}
+void    grantAdd(EGrantType grant, uint64_t value) {}
+void    grantRestore(EGrantType grant, uint64_t value) {}
+
+#endif
\ No newline at end of file

source/dnode/vnode/CMakeLists.txt

@@ -55,6 +55,8 @@ target_include_directories(
     vnode
     PUBLIC "inc"
     PRIVATE "src/inc"
+    PUBLIC "${TD_SOURCE_DIR}/include/libs/scalar"
+    
 )
 target_link_libraries(
     vnode
@@ -69,6 +71,7 @@ target_link_libraries(
     PUBLIC scheduler
     PUBLIC tdb
     #PUBLIC bdb
+    #PUBLIC scalar
     PUBLIC transport
     PUBLIC stream
 )

source/dnode/vnode/src/meta/metaTDBImpl.c

@@ -25,18 +25,18 @@ typedef struct SPoolMem {
 static SPoolMem *openPool();
 static void      clearPool(SPoolMem *pPool);
 static void      closePool(SPoolMem *pPool);
-static void     *poolMalloc(void *arg, size_t size);
+static void *    poolMalloc(void *arg, size_t size);
 static void      poolFree(void *arg, void *ptr);
 
 struct SMetaDB {
   TXN       txn;
-  TENV     *pEnv;
-  TDB      *pTbDB;
-  TDB      *pSchemaDB;
-  TDB      *pNameIdx;
-  TDB      *pStbIdx;
-  TDB      *pNtbIdx;
-  TDB      *pCtbIdx;
+  TENV *    pEnv;
+  TDB *     pTbDB;
+  TDB *     pSchemaDB;
+  TDB *     pNameIdx;
+  TDB *     pStbIdx;
+  TDB *     pNtbIdx;
+  TDB *     pCtbIdx;
   SPoolMem *pPool;
 };
 
@@ -46,7 +46,7 @@ typedef struct __attribute__((__packed__)) {
 } SSchemaDbKey;
 
 typedef struct {
-  char    *name;
+  char *   name;
   tb_uid_t uid;
 } SNameIdxKey;
 
@@ -205,14 +205,14 @@ void metaCloseDB(SMeta *pMeta) {
 
 int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg) {
   tb_uid_t       uid;
-  SMetaDB       *pMetaDb;
-  void          *pKey;
-  void          *pVal;
+  SMetaDB *      pMetaDb;
+  void *         pKey;
+  void *         pVal;
   int            kLen;
   int            vLen;
   int            ret;
   char           buf[512];
-  void          *pBuf;
+  void *         pBuf;
   SCtbIdxKey     ctbIdxKey;
   SSchemaDbKey   schemaDbKey;
   SSchemaWrapper schemaWrapper;
@@ -329,11 +329,11 @@ int metaRemoveTableFromDb(SMeta *pMeta, tb_uid_t uid) {
 STbCfg *metaGetTbInfoByUid(SMeta *pMeta, tb_uid_t uid) {
   int      ret;
   SMetaDB *pMetaDb = pMeta->pDB;
-  void    *pKey;
-  void    *pVal;
+  void *   pKey;
+  void *   pVal;
   int      kLen;
   int      vLen;
-  STbCfg  *pTbCfg;
+  STbCfg * pTbCfg;
 
   // Fetch
   pKey = &uid;
@@ -385,14 +385,14 @@ SSchemaWrapper *metaGetTableSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, boo
 }
 
 static SSchemaWrapper *metaGetTableSchemaImpl(SMeta *pMeta, tb_uid_t uid, int32_t sver, bool isinline, bool isGetEx) {
-  void           *pKey;
-  void           *pVal;
+  void *          pKey;
+  void *          pVal;
   int             kLen;
   int             vLen;
   int             ret;
   SSchemaDbKey    schemaDbKey;
   SSchemaWrapper *pSchemaWrapper;
-  void           *pBuf;
+  void *          pBuf;
 
   // fetch
   schemaDbKey.uid = uid;
@@ -419,9 +419,9 @@ STSchema *metaGetTbTSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver) {
   tb_uid_t        quid;
   SSchemaWrapper *pSW;
   STSchemaBuilder sb;
-  SSchemaEx      *pSchema;
-  STSchema       *pTSchema;
-  STbCfg         *pTbCfg;
+  SSchemaEx *     pSchema;
+  STSchema *      pTSchema;
+  STbCfg *        pTbCfg;
 
   pTbCfg = metaGetTbInfoByUid(pMeta, uid);
   if (pTbCfg->type == META_CHILD_TABLE) {
@@ -452,7 +452,7 @@ struct SMTbCursor {
 
 SMTbCursor *metaOpenTbCursor(SMeta *pMeta) {
   SMTbCursor *pTbCur = NULL;
-  SMetaDB    *pDB = pMeta->pDB;
+  SMetaDB *   pDB = pMeta->pDB;
 
   pTbCur = (SMTbCursor *)taosMemoryCalloc(1, sizeof(*pTbCur));
   if (pTbCur == NULL) {
@@ -474,12 +474,12 @@ void metaCloseTbCursor(SMTbCursor *pTbCur) {
 }
 
 char *metaTbCursorNext(SMTbCursor *pTbCur) {
-  void  *pKey = NULL;
-  void  *pVal = NULL;
+  void * pKey = NULL;
+  void * pVal = NULL;
   int    kLen;
   int    vLen;
   int    ret;
-  void  *pBuf;
+  void * pBuf;
   STbCfg tbCfg;
 
   for (;;) {
@@ -503,17 +503,17 @@ char *metaTbCursorNext(SMTbCursor *pTbCur) {
 }
 
 struct SMCtbCursor {
-  TDBC    *pCur;
+  TDBC *   pCur;
   tb_uid_t suid;
-  void    *pKey;
-  void    *pVal;
+  void *   pKey;
+  void *   pVal;
   int      kLen;
   int      vLen;
 };
 
 SMCtbCursor *metaOpenCtbCursor(SMeta *pMeta, tb_uid_t uid) {
   SMCtbCursor *pCtbCur = NULL;
-  SMetaDB     *pDB = pMeta->pDB;
+  SMetaDB *    pDB = pMeta->pDB;
   int          ret;
 
   pCtbCur = (SMCtbCursor *)taosMemoryCalloc(1, sizeof(*pCtbCur));
@@ -621,6 +621,7 @@ static int metaEncodeSchema(void **buf, SSchemaWrapper *pSW) {
   for (int i = 0; i < pSW->nCols; i++) {
     pSchema = pSW->pSchema + i;
     tlen += taosEncodeFixedI8(buf, pSchema->type);
+    tlen += taosEncodeFixedI8(buf, pSchema->index);
     tlen += taosEncodeFixedI16(buf, pSchema->colId);
     tlen += taosEncodeFixedI32(buf, pSchema->bytes);
     tlen += taosEncodeString(buf, pSchema->name);
@@ -637,6 +638,7 @@ static void *metaDecodeSchema(void *buf, SSchemaWrapper *pSW) {
   for (int i = 0; i < pSW->nCols; i++) {
     pSchema = pSW->pSchema + i;
     buf = taosDecodeFixedI8(buf, &pSchema->type);
+    buf = taosSkipFixedLen(buf, sizeof(int8_t));
     buf = taosDecodeFixedI16(buf, &pSchema->colId);
     buf = taosDecodeFixedI32(buf, &pSchema->bytes);
     buf = taosDecodeStringTo(buf, pSchema->name);
@@ -781,7 +783,7 @@ static void closePool(SPoolMem *pPool) {
 }
 
 static void *poolMalloc(void *arg, size_t size) {
-  void     *ptr = NULL;
+  void *    ptr = NULL;
   SPoolMem *pPool = (SPoolMem *)arg;
   SPoolMem *pMem;
 

source/libs/executor/inc/executorimpl.h

@@ -411,7 +411,7 @@ typedef struct STableScanInfo {
   SResultRowInfo* pResultRowInfo;
   int32_t*        rowCellInfoOffset;
   SExprInfo*      pExpr;
-  SSDataBlock     block;
+  SSDataBlock*    pResBlock;
   SArray*         pColMatchInfo;
   int32_t         numOfOutput;
   int64_t         elapsedTime;
@@ -569,6 +569,9 @@ typedef struct SGroupbyOperatorInfo {
   int32_t        groupKeyLen;   // total group by column width
   SGroupResInfo  groupResInfo;
   SAggSupporter  aggSup;
+  SExprInfo*     pScalarExprInfo;
+  int32_t        numOfScalarExpr;// the number of scalar expression in group operator
+  SqlFunctionCtx*pScalarFuncCtx;
 } SGroupbyOperatorInfo;
 
 typedef struct SDataGroupInfo {
@@ -674,7 +677,7 @@ void operatorDummyCloseFn(void* param, int32_t numOfCols);
 int32_t appendDownstream(SOperatorInfo* p, SOperatorInfo** pDownstream, int32_t num);
 int32_t initAggInfo(SOptrBasicInfo* pBasicInfo, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols,
                     int32_t numOfRows, SSDataBlock* pResultBlock, size_t keyBufSize, const char* pkey);
-void toSDatablock(SGroupResInfo* pGroupResInfo, SDiskbasedBuf* pBuf, SSDataBlock* pBlock, int32_t rowCapacity, int32_t* rowCellOffset);
+void toSDatablock(SSDataBlock* pBlock, int32_t rowCapacity, SGroupResInfo* pGroupResInfo, SExprInfo* pExprInfo, SDiskbasedBuf* pBuf, int32_t* rowCellOffset);
 void finalizeMultiTupleQueryResult(SqlFunctionCtx* pCtx, int32_t numOfOutput, SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowInfo, int32_t* rowCellInfoOffset);
 void doApplyFunctions(SqlFunctionCtx* pCtx, STimeWindow* pWin, SColumnInfoData* pTimeWindowData, int32_t offset, int32_t forwardStep, TSKEY* tsCol, int32_t numOfTotal, int32_t numOfOutput, int32_t order);
 int32_t setGroupResultOutputBuf_rv(SOptrBasicInfo* binfo, int32_t numOfCols, char* pData, int16_t type,
@@ -685,12 +688,11 @@ int32_t setSDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLoadI
                                          uint64_t* total, SArray* pColList);
 void doSetOperatorCompleted(SOperatorInfo* pOperator);
 void doFilter(const SNode* pFilterNode, SSDataBlock* pBlock);
-SSDataBlock* getSortedBlockData(SSortHandle* pHandle, SSDataBlock* pDataBlock, int32_t capacity);
-SSDataBlock* loadNextDataBlock(void* param);
+SqlFunctionCtx* createSqlFunctionCtx(SExprInfo* pExprInfo, int32_t numOfOutput, int32_t** rowCellInfoOffset);
 
 SOperatorInfo* createExchangeOperatorInfo(const SNodeList* pSources, SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo);
 SOperatorInfo* createTableScanOperatorInfo(void* pTsdbReadHandle, int32_t order, int32_t numOfCols, int32_t repeatTime,
-                                           int32_t reverseTime, SArray* pColMatchInfo, SNode* pCondition, SExecTaskInfo* pTaskInfo);
+                                           int32_t reverseTime, SArray* pColMatchInfo, SSDataBlock* pResBlock, SNode* pCondition, SExecTaskInfo* pTaskInfo);
 SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock,
                                            SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo);
 SOperatorInfo* createMultiTableAggOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo);
@@ -704,8 +706,8 @@ SOperatorInfo* createSysTableScanOperatorInfo(void* pSysTableReadHandle, SSDataB
 SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlot,
                                           const STableGroupInfo* pTableGroupInfo, SExecTaskInfo* pTaskInfo);
 SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, int64_t gap, SExecTaskInfo* pTaskInfo);
-SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock,
-                                       SArray* pGroupColList, SNode* pCondition, SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo);
+SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock, SArray* pGroupColList,
+                                       SNode* pCondition, SExprInfo* pScalarExprInfo, int32_t numOfScalarExpr, SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo);
 SOperatorInfo* createDataBlockInfoScanOperator(void* dataReader, SExecTaskInfo* pTaskInfo);
 SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, SSDataBlock* pResBlock, SArray* pColList, SArray* pTableIdList, SExecTaskInfo* pTaskInfo);
 
@@ -729,6 +731,8 @@ SOperatorInfo* createJoinOperatorInfo(SOperatorInfo** pdownstream, int32_t numOf
                                       int32_t numOfOutput);
 #endif
 
+void projectApplyFunctions(SExprInfo* pExpr, SSDataBlock* pResult, SSDataBlock* pSrcBlock, SqlFunctionCtx* pCtx, int32_t numOfOutput, SArray* pPseudoList);
+
 void setInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t order);
 
 void finalizeQueryResult(SqlFunctionCtx* pCtx, int32_t numOfOutput);

source/libs/executor/src/dataDispatcher.c

@@ -92,9 +92,7 @@ static bool allocBuf(SDataDispatchHandle* pDispatcher, const SInputData* pInput,
     return false;
   }
 
-  // NOTE: there are four bytes of an integer more than the required buffer space.
-  // struct size + data payload + length for each column + bitmap length
-  pBuf->allocSize = sizeof(SRetrieveTableRsp) + blockDataGetSerialMetaSize(pInput->pData) + blockDataGetSize(pInput->pData);
+  pBuf->allocSize = sizeof(SRetrieveTableRsp) + blockEstimateEncodeSize(pInput->pData);
 
   pBuf->pData = taosMemoryMalloc(pBuf->allocSize);
   if (pBuf->pData == NULL) {

source/libs/executor/src/groupoperator.c

@@ -265,7 +265,7 @@ static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator, bool* newgrou
   SSDataBlock* pRes = pInfo->binfo.pRes;
 
   if (pOperator->status == OP_RES_TO_RETURN) {
-    toSDatablock(&pInfo->groupResInfo, pInfo->aggSup.pResultBuf, pRes, pInfo->binfo.capacity, pInfo->binfo.rowCellInfoOffset);
+    toSDatablock(pRes, pInfo->binfo.capacity, &pInfo->groupResInfo, pOperator->pExpr, pInfo->aggSup.pResultBuf, pInfo->binfo.rowCellInfoOffset);
     if (pRes->info.rows == 0 || !hasRemainDataInCurrentGroup(&pInfo->groupResInfo)) {
       pOperator->status = OP_EXEC_DONE;
     }
@@ -285,6 +285,12 @@ static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator, bool* newgrou
 
     // the pDataBlock are always the same one, no need to call this again
     setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, order);
+
+    // there is an scalar expression that needs to be calculated before apply the group aggregation.
+    if (pInfo->pScalarExprInfo != NULL) {
+      projectApplyFunctions(pInfo->pScalarExprInfo, pBlock, pBlock, pInfo->pScalarFuncCtx, pInfo->numOfScalarExpr, NULL);
+    }
+
     //    setTagValue(pOperator, pRuntimeEnv->current->pTable, pInfo->binfo.pCtx, pOperator->numOfOutput);
     doHashGroupbyAgg(pOperator, pBlock);
   }
@@ -305,7 +311,7 @@ static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator, bool* newgrou
   initGroupResInfo(&pInfo->groupResInfo, &pInfo->binfo.resultRowInfo);
 
   while(1) {
-    toSDatablock(&pInfo->groupResInfo, pInfo->aggSup.pResultBuf, pRes, pInfo->binfo.capacity, pInfo->binfo.rowCellInfoOffset);
+    toSDatablock(pRes, pInfo->binfo.capacity, &pInfo->groupResInfo, pOperator->pExpr, pInfo->aggSup.pResultBuf, pInfo->binfo.rowCellInfoOffset);
     doFilter(pInfo->pCondition, pRes);
 
     bool hasRemain = hasRemainDataInCurrentGroup(&pInfo->groupResInfo);
@@ -322,16 +328,21 @@ static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator, bool* newgrou
   return (pRes->info.rows == 0)? NULL:pRes;
 }
 
-SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock, SArray* pGroupColList, SNode* pCondition, SExecTaskInfo* pTaskInfo,
-                                       const STableGroupInfo* pTableGroupInfo) {
+SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock, SArray* pGroupColList,
+    SNode* pCondition, SExprInfo* pScalarExprInfo, int32_t numOfScalarExpr, SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo) {
   SGroupbyOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SGroupbyOperatorInfo));
   SOperatorInfo*        pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
   if (pInfo == NULL || pOperator == NULL) {
     goto _error;
   }
 
-  pInfo->pGroupCols = pGroupColList;
-  pInfo->pCondition = pCondition;
+  pInfo->pGroupCols      = pGroupColList;
+  pInfo->pCondition      = pCondition;
+
+  pInfo->pScalarExprInfo = pScalarExprInfo;
+  pInfo->numOfScalarExpr = numOfScalarExpr;
+  pInfo->pScalarFuncCtx = createSqlFunctionCtx(pExprInfo, numOfCols, &pInfo->binfo.rowCellInfoOffset);
+
 
   int32_t code = initGroupOptrInfo(&pInfo->pGroupColVals, &pInfo->groupKeyLen, &pInfo->keyBuf, pGroupColList);
   if (code != TSDB_CODE_SUCCESS) {

source/libs/executor/src/scanoperator.c

@@ -113,7 +113,7 @@ static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator, bool* newgroup) {
   STableScanInfo* pTableScanInfo = pOperator->info;
   SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;
 
-  SSDataBlock*     pBlock = &pTableScanInfo->block;
+  SSDataBlock*     pBlock = pTableScanInfo->pResBlock;
   STableGroupInfo* pTableGroupInfo = &pOperator->pTaskInfo->tableqinfoGroupInfo;
 
   *newgroup = false;
@@ -218,7 +218,7 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator, bool* newgroup) {
 }
 
 SOperatorInfo* createTableScanOperatorInfo(void* pTsdbReadHandle, int32_t order, int32_t numOfOutput,
-                                           int32_t repeatTime, int32_t reverseTime, SArray* pColMatchInfo,
+                                           int32_t repeatTime, int32_t reverseTime, SArray* pColMatchInfo, SSDataBlock* pResBlock,
                                            SNode* pCondition, SExecTaskInfo* pTaskInfo) {
   assert(repeatTime > 0);
 
@@ -232,12 +232,7 @@ SOperatorInfo* createTableScanOperatorInfo(void* pTsdbReadHandle, int32_t order,
     return NULL;
   }
 
-  pInfo->block.pDataBlock = taosArrayInit(numOfOutput, sizeof(SColumnInfoData));
-  for (int32_t i = 0; i < numOfOutput; ++i) {
-    SColumnInfoData idata = {0};
-    taosArrayPush(pInfo->block.pDataBlock, &idata);
-  }
-
+  pInfo->pResBlock        = pResBlock;
   pInfo->pFilterNode      = pCondition;
   pInfo->dataReader       = pTsdbReadHandle;
   pInfo->times            = repeatTime;
@@ -312,7 +307,7 @@ static SSDataBlock* doBlockInfoScan(SOperatorInfo* pOperator, bool* newgroup) {
   tsdbGetFileBlocksDistInfo(pTableScanInfo->dataReader, &tableBlockDist);
   tableBlockDist.numOfRowsInMemTable = (int32_t) tsdbGetNumOfRowsInMemTable(pTableScanInfo->dataReader);
 
-  SSDataBlock* pBlock = &pTableScanInfo->block;
+  SSDataBlock* pBlock = pTableScanInfo->pResBlock;
   pBlock->info.rows   = 1;
   pBlock->info.numOfCols = 1;
 
@@ -343,13 +338,13 @@ SOperatorInfo* createDataBlockInfoScanOperator(void* dataReader, SExecTaskInfo*
   }
 
   pInfo->dataReader       = dataReader;
-  pInfo->block.pDataBlock = taosArrayInit(1, sizeof(SColumnInfoData));
+//  pInfo->block.pDataBlock = taosArrayInit(1, sizeof(SColumnInfoData));
 
   SColumnInfoData infoData = {0};
   infoData.info.type  = TSDB_DATA_TYPE_BINARY;
   infoData.info.bytes = 1024;
   infoData.info.colId = 0;
-  taosArrayPush(pInfo->block.pDataBlock, &infoData);
+//  taosArrayPush(pInfo->block.pDataBlock, &infoData);
 
   pOperator->name          = "DataBlockInfoScanOperator";
   //  pOperator->operatorType = OP_TableBlockInfoScan;

source/libs/function/inc/builtins.h

@@ -22,7 +22,7 @@ extern "C" {
 
 #include "functionMgt.h"
 
-#define FUNCTION_NAME_MAX_LENGTH 16
+#define FUNCTION_NAME_MAX_LENGTH 32
 
 #define FUNC_MGT_FUNC_CLASSIFICATION_MASK(n)    (1 << n)
 
@@ -41,12 +41,14 @@ extern "C" {
 #define FUNC_MGT_TEST_MASK(val, mask) (((val) & (mask)) != 0)
 
 typedef int32_t (*FCheckAndGetResultType)(SFunctionNode* pFunc);
+typedef EFuncDataRequired (*FFuncDataRequired)(SFunctionNode* pFunc, STimeWindow* pTimeWindow);
 
 typedef struct SBuiltinFuncDefinition {
   char name[FUNCTION_NAME_MAX_LENGTH];
   EFunctionType type;
   uint64_t classification;
   FCheckAndGetResultType checkFunc;
+  FFuncDataRequired dataRequiredFunc;
   FExecGetEnv getEnvFunc;
   FExecInit initFunc;
   FExecProcess processFunc;

source/libs/function/inc/builtinsimpl.h

@@ -21,10 +21,12 @@ extern "C" {
 #endif
 
 #include "function.h"
+#include "functionMgt.h"
 
 bool functionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
 void functionFinalize(SqlFunctionCtx *pCtx);
 
+EFuncDataRequired countDataRequired(SFunctionNode* pFunc, STimeWindow* pTimeWindow);
 bool getCountFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 int32_t countFunction(SqlFunctionCtx *pCtx);
 

source/libs/function/src/builtinsimpl.c

@@ -55,6 +55,14 @@ void functionFinalize(SqlFunctionCtx *pCtx) {
   pResInfo->isNullRes = (pResInfo->numOfRes == 0)? 1:0;
 }
 
+EFuncDataRequired countDataRequired(SFunctionNode* pFunc, STimeWindow* pTimeWindow) {
+  SNode* pParam = nodesListGetNode(pFunc->pParameterList, 0);
+  if (QUERY_NODE_COLUMN == nodeType(pParam) && PRIMARYKEY_TIMESTAMP_COL_ID == ((SColumnNode*)pParam)->colId) {
+    return FUNC_DATA_REQUIRED_NO_NEEDED;
+  }
+  return FUNC_DATA_REQUIRED_STATIS_NEEDED;
+}
+
 bool getCountFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
   pEnv->calcMemSize = sizeof(int64_t);
   return true;

source/libs/function/src/functionMgt.c

@@ -76,6 +76,16 @@ int32_t fmGetFuncResultType(SFunctionNode* pFunc) {
   return funcMgtBuiltins[pFunc->funcId].checkFunc(pFunc);
 }
 
+EFuncDataRequired fmFuncDataRequired(SFunctionNode* pFunc, STimeWindow* pTimeWindow) {
+  if (pFunc->funcId < 0 || pFunc->funcId >= funcMgtBuiltinsNum) {
+    return FUNC_DATA_REQUIRED_ALL_NEEDED;
+  }
+  if (NULL == funcMgtBuiltins[pFunc->funcId].dataRequiredFunc) {
+    return FUNC_DATA_REQUIRED_ALL_NEEDED;
+  }
+  return funcMgtBuiltins[pFunc->funcId].dataRequiredFunc(pFunc, pTimeWindow);
+}
+
 int32_t fmGetFuncExecFuncs(int32_t funcId, SFuncExecFuncs* pFpSet) {
   if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
     return TSDB_CODE_FAILED;
@@ -120,6 +130,13 @@ bool fmIsNonstandardSQLFunc(int32_t funcId) {
   return isSpecificClassifyFunc(funcId, FUNC_MGT_NONSTANDARD_SQL_FUNC);
 }
 
+bool fmIsSpecialDataRequiredFunc(int32_t funcId) {
+  return isSpecificClassifyFunc(funcId, FUNC_MGT_SPECIAL_DATA_REQUIRED);
+}
+
+bool fmIsDynamicScanOptimizedFunc(int32_t funcId) {
+  return isSpecificClassifyFunc(funcId, FUNC_MGT_DYNAMIC_SCAN_OPTIMIZED);
+}
 
 void fmFuncMgtDestroy() {
   void* m = gFunMgtService.pFuncNameHashTable;

source/libs/function/src/tfill.c

@@ -541,7 +541,7 @@ struct SFillColInfo* createFillColInfo(SExprInfo* pExpr, int32_t numOfOutput, co
     pFillCol[i].col.bytes  = pExprInfo->base.resSchema.bytes;
     pFillCol[i].col.type   = (int8_t)pExprInfo->base.resSchema.type;
     pFillCol[i].col.offset = offset;
-    pFillCol[i].col.colId  = pExprInfo->base.resSchema.colId;
+    pFillCol[i].col.colId  = pExprInfo->base.resSchema.slotId;
     pFillCol[i].tagIndex   = -2;
     pFillCol[i].flag       = pExprInfo->base.pParam[0].pCol->flag;    // always be the normal column for table query
 //    pFillCol[i].functionId = pExprInfo->pExpr->_function.functionId;

taos-tools @ bf6c7669

-Subproject commit 33cdfe4f90a209f105c1b6091439798a9cde1e93
+Subproject commit bf6c766986c61ff4fc80421fdea682a8fd4b5b32