Merge pull request #11491 from taosdata/feature/tq

enh(query): fetch raw block fail back to query type

source/client/src/clientMain.c

@@ -463,25 +463,7 @@ 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;
-
-    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;
-    }
-
-    doFetchRows(pRequest, false, false);
-
-    SReqResultInfo *pResultInfo = &pRequest->body.resInfo;
-
-    pResultInfo->current = pResultInfo->numOfRows;
-    (*numOfRows) = pResultInfo->numOfRows;
-    (*pData) = (void *)pResultInfo->pData;
-
-    return 0;
-
-  } else if (TD_RES_TMQ(res)) {
+  if (TD_RES_TMQ(res)) {
     SReqResultInfo *pResultInfo = tmqGetNextResInfo(res);
     if (pResultInfo == NULL) return -1;
 
@@ -489,11 +471,24 @@ int taos_fetch_raw_block(TAOS_RES *res, int *numOfRows, void **pData) {
     (*numOfRows) = pResultInfo->numOfRows;
     (*pData) = (void *)pResultInfo->pData;
     return 0;
+  }
 
-  } else {
-    ASSERT(0);
-    return -1;
+  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 0;
   }
+
+  doFetchRows(pRequest, false, false);
+
+  SReqResultInfo *pResultInfo = &pRequest->body.resInfo;
+
+  pResultInfo->current = pResultInfo->numOfRows;
+  (*numOfRows) = pResultInfo->numOfRows;
+  (*pData) = (void *)pResultInfo->pData;
+
+  return 0;
 }
 
 int *taos_get_column_data_offset(TAOS_RES *res, int columnIndex) {

tests/test/c/tmqDemo.c

@@ -13,8 +13,6 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#define ALLOW_FORBID_FUNC
-
 #include <assert.h>
 #include <dirent.h>
 #include <stdio.h>
@@ -36,7 +34,12 @@
 #define MAX_SQL_STR_LEN (1024 * 1024)
 #define MAX_ROW_STR_LEN (16 * 1024)
 
-enum _RUN_MODE { TMQ_RUN_INSERT_AND_CONSUME, TMQ_RUN_ONLY_INSERT, TMQ_RUN_ONLY_CONSUME, TMQ_RUN_MODE_BUTT };
+enum _RUN_MODE {
+  TMQ_RUN_INSERT_AND_CONSUME,
+  TMQ_RUN_ONLY_INSERT,
+  TMQ_RUN_ONLY_CONSUME,
+  TMQ_RUN_MODE_BUTT,
+};
 
 typedef struct {
   char    dbName[32];
@@ -338,6 +341,7 @@ tmq_t* build_consumer() {
   tmq_conf_set(conf, "td.connect.pass", "taosdata");
   tmq_conf_set(conf, "td.connect.db", g_stConfInfo.dbName);
   tmq_t* tmq = tmq_consumer_new1(conf, NULL, 0);
+  assert(tmq);
   tmq_conf_destroy(conf);
   return tmq;
 }
@@ -596,7 +600,7 @@ void printParaIntoFile() {
   g_fp = pFile;
 
   time_t    tTime = taosGetTimestampSec();
-  struct tm tm = *localtime(&tTime);
+  struct tm tm = *taosLocalTime(&tTime, NULL);
 
   taosFprintfFile(pFile, "###################################################################\n");
   taosFprintfFile(pFile, "# configDir:                %s\n", configDir);