fix:add test case for raw_block interface

source/libs/parser/src/parInsertUtil.c

@@ -621,12 +621,12 @@ int rawBlockBindData(SQuery* query, STableMeta* pTableMeta, void* data, SVCreate
   SSchema*       pSchema = getTableColumnSchema(pTableCxt->pMeta);
   SBoundColInfo* boundInfo = &pTableCxt->boundColsInfo;
 
-  if (numFields != numOfCols) {
+  if (tFields != NULL && numFields != numOfCols) {
     uError("numFields:%d != numOfCols:%d", numFields, numOfCols);
     ret = TSDB_CODE_INVALID_PARA;
     goto end;
   }
-  if (numFields > boundInfo->numOfBound) {
+  if (tFields != NULL && numFields > boundInfo->numOfBound) {
     uError("numFields:%d > boundInfo->numOfBound:%d", numFields, boundInfo->numOfBound);
     ret = TSDB_CODE_INVALID_PARA;
     goto end;
@@ -634,7 +634,7 @@ int rawBlockBindData(SQuery* query, STableMeta* pTableMeta, void* data, SVCreate
   for (int c = 0; c < boundInfo->numOfBound; ++c) {
     SSchema*  pColSchema = &pSchema[c];
     SColData* pCol = taosArrayGet(pTableCxt->pData->aCol, c);
-    if(findFileds(pColSchema, tFields, numFields)){
+    if(tFields == NULL || findFileds(pColSchema, tFields, numFields)){
       if (*fields != pColSchema->type && *(int32_t*)(fields + sizeof(int8_t)) != pColSchema->bytes) {
         uError("type or bytes not equal");
         ret = TSDB_CODE_INVALID_PARA;

tests/parallel_test/cases.task

@@ -93,6 +93,7 @@
 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/dataFromTsdbNWal.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/dataFromTsdbNWal-multiCtb.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmq_taosx.py
+,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/raw_block_interface_test.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/stbTagFilter-multiCtb.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqSubscribeStb-r3.py -N 5
 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmq3mnodeSwitch.py -N 6 -M 3 

tests/system-test/7-tmq/raw_block_interface_test.py

+
+import taos
+import sys
+import time
+import socket
+import os
+import threading
+
+from util.log import *
+from util.sql import *
+from util.cases import *
+from util.dnodes import *
+from util.common import *
+sys.path.append("./7-tmq")
+from tmqCommon import *
+
+class TDTestCase:
+    def init(self, conn, logSql, replicaVar=1):
+        self.replicaVar = int(replicaVar)
+        tdLog.debug(f"start to excute {__file__}")
+        tdSql.init(conn.cursor())
+
+    def checkData(self):
+        tdSql.execute('use db_raw')
+        tdSql.query("select * from d1")
+        tdSql.checkRows(1)
+        tdSql.checkData(0, 1, 120)
+
+        tdSql.query("select * from d2")
+        tdSql.checkRows(1)
+        tdSql.checkData(0, 1, None)
+
+        return
+
+    def check(self):
+        buildPath = tdCom.getBuildPath()
+        cmdStr = '%s/build/bin/write_raw_block_test'%(buildPath)
+        tdLog.info(cmdStr)
+        os.system(cmdStr)
+
+        self.checkData()
+
+        return
+
+    def run(self):
+        tdSql.prepare()
+        self.check()
+
+    def stop(self):
+        tdSql.close()
+        tdLog.success(f"{__file__} successfully executed")
+
+tdCases.addLinux(__file__, TDTestCase())
+tdCases.addWindows(__file__, TDTestCase())

utils/test/c/CMakeLists.txt

@@ -3,6 +3,7 @@ add_dependencies(tmq_demo taos)
 add_executable(tmq_sim tmqSim.c)
 add_executable(create_table createTable.c)
 add_executable(tmq_taosx_ci tmq_taosx_ci.c)
+add_executable(write_raw_block_test write_raw_block_test.c)
 add_executable(sml_test sml_test.c)
 add_executable(get_db_name_test get_db_name_test.c)
 target_link_libraries(
@@ -34,6 +35,14 @@ target_link_libraries(
     PUBLIC os
 )
 
+target_link_libraries(
+    write_raw_block_test
+    PUBLIC taos_static
+    PUBLIC util
+    PUBLIC common
+    PUBLIC os
+)
+
 target_link_libraries(
     sml_test
     PUBLIC taos_static

utils/test/c/write_raw_block_test.c

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include <assert.h>
+#include <stdio.h>
+#include <time.h>
+#include "taos.h"
+#include "types.h"
+
+int buildStable(TAOS* pConn, TAOS_RES* pRes) {
+  pRes = taos_query(pConn,
+                    "CREATE STABLE `meters` (`ts` TIMESTAMP, `current` INT, `voltage` INT, `phase` FLOAT) TAGS "
+                    "(`groupid` INT, `location` VARCHAR(16))");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to create super table meters, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
+  pRes = taos_query(pConn, "create table d0 using meters tags(1, 'San Francisco')");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to create child table d0, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
+  pRes = taos_query(pConn, "insert into d0 (ts, current) values (now, 120)");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to insert into table d0, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
+  pRes = taos_query(pConn, "create table d1 using meters tags(2, 'San Francisco')");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to create child table d1, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
+  pRes = taos_query(pConn, "create table d2 using meters tags(3, 'San Francisco')");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to create child table d1, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
+  return 0;
+}
+
+int32_t init_env() {
+  TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
+  if (pConn == NULL) {
+    return -1;
+  }
+
+  TAOS_RES* pRes = taos_query(pConn, "drop database if exists db_raw");
+  if (taos_errno(pRes) != 0) {
+    printf("error in drop db_taosx, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
+  pRes = taos_query(pConn, "create database if not exists db_raw vgroups 2");
+  if (taos_errno(pRes) != 0) {
+    printf("error in create db_taosx, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
+  pRes = taos_query(pConn, "use db_raw");
+  if (taos_errno(pRes) != 0) {
+    printf("error in create db_taosx, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
+  buildStable(pConn, pRes);
+
+  pRes = taos_query(pConn, "select * from d0");
+  if (taos_errno(pRes) != 0) {
+    printf("error in drop db_taosx, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  void   *data = NULL;
+  int32_t numOfRows = 0;
+  int     error_code = taos_fetch_raw_block(pRes, &numOfRows, &data);
+  ASSERT(error_code == 0);
+  ASSERT(numOfRows == 1);
+
+  taos_write_raw_block(pConn, numOfRows, data, "d1");
+
+  pRes = taos_query(pConn, "select ts,phase from d0");
+  if (taos_errno(pRes) != 0) {
+    printf("error in drop db_taosx, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  error_code = taos_fetch_raw_block(pRes, &numOfRows, &data);
+  ASSERT(error_code == 0);
+  ASSERT(numOfRows == 1);
+
+  int         numFields = taos_num_fields(pRes);
+  TAOS_FIELD *fields = taos_fetch_fields(pRes);
+  taos_write_raw_block_with_fields(pConn, numOfRows, data, "d2", fields, numFields);
+
+  taos_close(pConn);
+  return 0;
+}
+
+int main(int argc, char* argv[]) {
+  if (init_env() < 0) {
+    return -1;
+  }
+}