Merge branch 'dzxc' into t4335

.gitmodules

@@ -13,3 +13,6 @@
 [submodule "deps/jemalloc"]
 	path = deps/jemalloc
 	url = https://github.com/jemalloc/jemalloc
+[submodule "deps/TSZ"]
+	path = deps/TSZ
+	url = https://github.com/taosdata/TSZ.git

cmake/input.inc

@@ -91,3 +91,12 @@ SET(TD_MEMORY_SANITIZER FALSE)
 IF (${MEMORY_SANITIZER} MATCHES "true")
   SET(TD_MEMORY_SANITIZER TRUE)
 ENDIF ()
+
+IF (${TSZ_ENABLED} MATCHES "true")
+  # define add 
+  MESSAGE(STATUS "build with TSZ enabled")
+  ADD_DEFINITIONS(-DTD_TSZ)
+  set(VAR_TSZ "TSZ" CACHE INTERNAL "global variant tsz" )
+ELSE()
+  set(VAR_TSZ "" CACHE INTERNAL "global variant empty" )
+ENDIF()  

cmake/install.inc

@@ -32,7 +32,7 @@ ELSEIF (TD_WINDOWS)
   #INSTALL(TARGETS taos RUNTIME DESTINATION driver)
   #INSTALL(TARGETS shell RUNTIME DESTINATION .)
   IF (TD_MVN_INSTALLED)
-    INSTALL(FILES ${LIBRARY_OUTPUT_PATH}/taos-jdbcdriver-*-dist.jar DESTINATION connector/jdbc)
+    INSTALL(FILES ${LIBRARY_OUTPUT_PATH}/taos-jdbcdriver-2.0.33-dist.jar DESTINATION connector/jdbc)
   ENDIF ()
 ELSEIF (TD_DARWIN)
   SET(TD_MAKE_INSTALL_SH "${TD_COMMUNITY_DIR}/packaging/tools/make_install.sh")

cmake/version.inc

@@ -4,7 +4,7 @@ PROJECT(TDengine)
 IF (DEFINED VERNUMBER)
   SET(TD_VER_NUMBER ${VERNUMBER})
 ELSE ()
-  SET(TD_VER_NUMBER "2.1.4.1")
+  SET(TD_VER_NUMBER "2.1.5.0")
 ENDIF ()
 
 IF (DEFINED VERCOMPATIBLE)

deps/CMakeLists.txt

@@ -15,7 +15,6 @@ ADD_SUBDIRECTORY(cJson)
 ADD_SUBDIRECTORY(wepoll)
 ADD_SUBDIRECTORY(MsvcLibX)
 ADD_SUBDIRECTORY(rmonotonic)
-
 ADD_SUBDIRECTORY(lua)
 
 IF (TD_LINUX AND TD_MQTT)
@@ -38,3 +37,7 @@ IF (TD_LINUX_64 AND JEMALLOC_ENABLED)
     BUILD_COMMAND       ${MAKE}
   )
 ENDIF ()
+
+IF (${TSZ_ENABLED} MATCHES "true")
+  ADD_SUBDIRECTORY(TSZ)  
+ENDIF()
\ No newline at end of file

TSZ @ 0ca5b15a

+Subproject commit 0ca5b15a8eac40327dd737be52c926fa5675712c

documentation20/cn/02.getting-started/docs.md

@@ -126,7 +126,7 @@ taos> source <filename>;
 $ taosdemo
 ```
 
-该命令将在数据库 test 下面自动创建一张超级表 meters，该超级表下有 1 万张表，表名为 "t0" 到 "t9999"，每张表有 1 万条记录，每条记录有 (ts, current, voltage, phase) 四个字段，时间戳从 "2017-07-14 10:40:00 000" 到 "2017-07-14 10:40:09 999"，每张表带有标签 location 和 groupdId，groupdId 被设置为 1 到 10， location 被设置为 "beijing" 或者 "shanghai"。
+该命令将在数据库 test 下面自动创建一张超级表 meters，该超级表下有 1 万张表，表名为 "d0" 到 "d9999"，每张表有 1 万条记录，每条记录有 (ts, current, voltage, phase) 四个字段，时间戳从 "2017-07-14 10:40:00 000" 到 "2017-07-14 10:40:09 999"，每张表带有标签 location 和 groupdId，groupdId 被设置为 1 到 10， location 被设置为 "beijing" 或者 "shanghai"。
 
 执行这条命令大概需要几分钟，最后共插入 1 亿条记录。
 
@@ -156,10 +156,10 @@ taos> select count(*) from test.meters where location="beijing";
 taos> select avg(current), max(voltage), min(phase) from test.meters where groupdId=10;
 ```
 
-- 对表 t10 按 10s 进行平均值、最大值和最小值聚合统计：
+- 对表 d10 按 10s 进行平均值、最大值和最小值聚合统计：
 
 ```mysql
-taos> select avg(current), max(voltage), min(phase) from test.t10 interval(10s);
+taos> select avg(current), max(voltage), min(phase) from test.d10 interval(10s);
 ```
 
 **Note:** taosdemo 命令本身带有很多选项，配置表的数目、记录条数等等，请执行 `taosdemo --help` 详细列出。您可以设置不同参数进行体验。

documentation20/cn/12.taos-sql/docs.md

@@ -715,7 +715,7 @@ Query OK, 1 row(s) in set (0.001091s)
 2. 同时进行多个字段的范围过滤，需要使用关键词 AND 来连接不同的查询条件，暂不支持 OR 连接的不同列之间的查询过滤条件。
 3. 针对单一字段的过滤，如果是时间过滤条件，则一条语句中只支持设定一个；但针对其他的（普通）列或标签列，则可以使用 `OR` 关键字进行组合条件的查询过滤。例如： `((value > 20 AND value < 30) OR (value < 12))`。
 4. 从 2.0.17.0 版本开始，条件过滤开始支持 BETWEEN AND 语法，例如 `WHERE col2 BETWEEN 1.5 AND 3.25` 表示查询条件为“1.5 ≤ col2 ≤ 3.25”。
-5. 从 2.1.4.0 版本开始，条件过滤开始支持 IN 算子，例如 `WHERE city IN ('Beijing', 'Shanghai')`。说明：BOOL 类型写作 `{true, false}` 或 `{0, 1}` 均可，但不能写作 0、1 之外的整数；FLOAT 和 DOUBLE 类型会受到浮点数精度影响，集合内的值在精度范围内认为和数据行的值完全相等才能匹配成功。<!-- REPLACE_OPEN_TO_ENTERPRISE__IN_OPERATOR_AND_UNSIGNED_INTEGER -->
+5. 从 2.1.4.0 版本开始，条件过滤开始支持 IN 算子，例如 `WHERE city IN ('Beijing', 'Shanghai')`。说明：BOOL 类型写作 `{true, false}` 或 `{0, 1}` 均可，但不能写作 0、1 之外的整数；FLOAT 和 DOUBLE 类型会受到浮点数精度影响，集合内的值在精度范围内认为和数据行的值完全相等才能匹配成功；TIMESTAMP 类型支持非主键的列。<!-- REPLACE_OPEN_TO_ENTERPRISE__IN_OPERATOR_AND_UNSIGNED_INTEGER -->
 
 <!-- 
 <a class="anchor" id="having"></a>
@@ -1338,7 +1338,8 @@ SELECT function_list FROM stb_name
 - 查询过滤、聚合等操作按照每个切分窗口为独立的单位执行。聚合查询目前支持三种窗口的划分方式：
   1. 时间窗口：聚合时间段的窗口宽度由关键词 INTERVAL 指定，最短时间间隔 10 毫秒（10a）；并且支持偏移 offset（偏移必须小于间隔），也即时间窗口划分与“UTC 时刻 0”相比的偏移量。SLIDING 语句用于指定聚合时间段的前向增量，也即每次窗口向前滑动的时长。当 SLIDING 与 INTERVAL 取值相等的时候，滑动窗口即为翻转窗口。
     * 从 2.1.5.0 版本开始，INTERVAL 语句允许的最短时间间隔调整为 1 微秒（1u），当然如果所查询的 DATABASE 的时间精度设置为毫秒级，那么允许的最短时间间隔为 1 毫秒（1a）。
-  2. 状态窗口：使用整数（布尔值）或字符串来标识产生记录时设备的状态量，产生的记录如果具有相同的状态量取值则归属于同一个状态窗口，数值改变后该窗口关闭。状态量所对应的列作为 STATE_WINDOW 语句的参数来指定。
+    * **注意：**用到 INTERVAL 语句时，除非极特殊的情况，都要求把客户端和服务端的 taos.cfg 配置文件中的 timezone 参数配置为相同的取值，以避免时间处理函数频繁进行跨时区转换而导致的严重性能影响。
+  2. 状态窗口：使用整数或布尔值来标识产生记录时设备的状态量，产生的记录如果具有相同的状态量取值则归属于同一个状态窗口，数值改变后该窗口关闭。状态量所对应的列作为 STATE_WINDOW 语句的参数来指定。
   3. 会话窗口：时间戳所在的列由 SESSION 语句的 ts_col 参数指定，会话窗口根据相邻两条记录的时间戳差值来确定是否属于同一个会话——如果时间戳差异在 tol_val 以内，则认为记录仍属于同一个窗口；如果时间变化超过 tol_val，则自动开启下一个窗口。
 - WHERE 语句可以指定查询的起止时间和其他过滤条件。
 - FILL 语句指定某一窗口区间数据缺失的情况下的填充模式。填充模式包括以下几种：

documentation20/en/12.taos-sql/docs.md

@@ -1132,7 +1132,7 @@ TDengine supports aggregations over data, they are listed below:
     ```
     Function: Return the difference between the max value and the min value of a column in statistics /STable.
     
-    Return Data Type: Same as applicable fields.
+    Return Data Type: Double.
     
     Applicable Fields: All types except binary, nchar, bool.
     

packaging/docker/dockerManifest.sh

@@ -44,6 +44,7 @@ echo "version=${version}"
 #docker manifest rm tdengine/tdengine
 #docker manifest rm tdengine/tdengine:${version}
 if [ "$verType" == "beta" ]; then
+  docker manifest rm tdengine/tdengine:latest
   docker manifest create -a tdengine/tdengine-beta:${version} tdengine/tdengine-amd64-beta:${version} tdengine/tdengine-aarch64-beta:${version} tdengine/tdengine-aarch32-beta:${version}
   docker manifest create -a tdengine/tdengine-beta:latest tdengine/tdengine-amd64-beta:latest tdengine/tdengine-aarch64-beta:latest tdengine/tdengine-aarch32-beta:latest
   docker login -u tdengine -p ${passWord}  #replace the docker registry username and password
@@ -51,6 +52,7 @@ if [ "$verType" == "beta" ]; then
   docker manifest push tdengine/tdengine-beta:${version}
 
 elif [ "$verType" == "stable" ]; then
+  docker manifest rm tdengine/tdengine:latest
   docker manifest create -a tdengine/tdengine:${version} tdengine/tdengine-amd64:${version} tdengine/tdengine-aarch64:${version} tdengine/tdengine-aarch32:${version}
   docker manifest create -a tdengine/tdengine:latest tdengine/tdengine-amd64:latest tdengine/tdengine-aarch64:latest tdengine/tdengine-aarch32:latest
   docker login -u tdengine -p ${passWord}  #replace the docker registry username and password

packaging/docker/dockerbuild.sh

@@ -67,8 +67,10 @@ done
 
 if [ "$verType" == "beta" ]; then
   dockername=${cpuType}-${verType}
+  dirName=${pkgFile%-beta*}
 elif [ "$verType" == "stable" ]; then
   dockername=${cpuType}
+  dirName=${pkgFile%-Linux*}
 else
   echo "unknow verType, nor stabel or beta"
   exit 1
@@ -84,11 +86,10 @@ comunityArchiveDir=/nas/TDengine/v$version/community   # community version’pac
 cd ${scriptDir}
 cp -f ${comunityArchiveDir}/${pkgFile}  .
 
-dirName=${pkgFile%-Linux*}
 echo "dirName=${dirName}"
 
 
-docker build --rm -f "Dockerfile" -t tdengine/tdengine-${dockername}:${version} "." --build-arg pkgFile=${pkgFile} --build-arg dirName=${dirName}
+docker build --rm -f "Dockerfile"  --network=host -t tdengine/tdengine-${dockername}:${version} "." --build-arg pkgFile=${pkgFile} --build-arg dirName=${dirName}
 docker login -u tdengine -p ${passWord}  #replace the docker registry username and password
 docker push tdengine/tdengine-${dockername}:${version}
 

packaging/tools/install_client.sh

@@ -182,7 +182,13 @@ function install_jemalloc() {
             ${csudo} /usr/bin/install -c -d /usr/local/share/man/man3
             ${csudo} /usr/bin/install -c -m 644 ${jemalloc_dir}/share/man/man3/jemalloc.3 /usr/local/share/man/man3
         fi
+
+        if [ -d /etc/ld.so.conf.d ]; then
+            ${csudo} echo "/usr/local/lib" > /etc/ld.so.conf.d/jemalloc.conf
             ${csudo} ldconfig
+        else
+            echo "/etc/ld.so.conf.d not found!"
+        fi
     fi
 }
 

packaging/tools/make_install.sh

@@ -421,7 +421,7 @@ function install_service() {
 }
 
 function update_TDengine() {
-    echo -e "${GREEN}Start to update TDEngine...${NC}"
+    echo -e "${GREEN}Start to update TDengine...${NC}"
     # Stop the service if running
 
     if [ "$osType" != "Darwin" ]; then

packaging/tools/makearbi.sh

@@ -59,7 +59,7 @@ pkg_name=${install_dir}-${osType}-${cpuType}
 #   exit 1
 # fi
 
-if [ "$verType" == "beta" ]; then
+if [[ "$verType" == "beta" ]] || [[ "$verType" == "preRelease" ]]; then
   pkg_name=${install_dir}-${verType}-${osType}-${cpuType} 
 elif [ "$verType" == "stable" ]; then
   pkg_name=${pkg_name}

packaging/tools/makeclient.sh

@@ -182,7 +182,7 @@ pkg_name=${install_dir}-${osType}-${cpuType}
 #   exit 1
 # fi
 
-if [ "$verType" == "beta" ]; then
+if [[ "$verType" == "beta" ]] || [[ "$verType" == "preRelease" ]]; then
   pkg_name=${install_dir}-${verType}-${osType}-${cpuType} 
 elif [ "$verType" == "stable" ]; then
   pkg_name=${pkg_name}

packaging/tools/makeclient_power.sh

@@ -67,6 +67,39 @@ mkdir -p ${install_dir}
 mkdir -p ${install_dir}/inc && cp ${header_files} ${install_dir}/inc
 mkdir -p ${install_dir}/cfg && cp ${cfg_dir}/taos.cfg ${install_dir}/cfg/taos.cfg
 
+if [ -f ${build_dir}/bin/jemalloc-config ]; then
+    mkdir -p ${install_dir}/jemalloc/{bin,lib,lib/pkgconfig,include/jemalloc,share/doc/jemalloc,share/man/man3}
+    cp ${build_dir}/bin/jemalloc-config ${install_dir}/jemalloc/bin
+    if [ -f ${build_dir}/bin/jemalloc.sh ]; then
+        cp ${build_dir}/bin/jemalloc.sh ${install_dir}/jemalloc/bin
+    fi
+    if [ -f ${build_dir}/bin/jeprof ]; then
+        cp ${build_dir}/bin/jeprof ${install_dir}/jemalloc/bin
+    fi
+    if [ -f ${build_dir}/include/jemalloc/jemalloc.h ]; then
+        cp ${build_dir}/include/jemalloc/jemalloc.h ${install_dir}/jemalloc/include/jemalloc
+    fi
+    if [ -f ${build_dir}/lib/libjemalloc.so.2 ]; then
+        cp ${build_dir}/lib/libjemalloc.so.2 ${install_dir}/jemalloc/lib
+        ln -sf libjemalloc.so.2 ${install_dir}/jemalloc/lib/libjemalloc.so
+    fi
+    if [ -f ${build_dir}/lib/libjemalloc.a ]; then
+        cp ${build_dir}/lib/libjemalloc.a ${install_dir}/jemalloc/lib
+    fi
+    if [ -f ${build_dir}/lib/libjemalloc_pic.a ]; then
+        cp ${build_dir}/lib/libjemalloc_pic.a ${install_dir}/jemalloc/lib
+    fi
+    if [ -f ${build_dir}/lib/pkgconfig/jemalloc.pc ]; then
+        cp ${build_dir}/lib/pkgconfig/jemalloc.pc ${install_dir}/jemalloc/lib/pkgconfig
+    fi
+    if [ -f ${build_dir}/share/doc/jemalloc/jemalloc.html ]; then
+        cp ${build_dir}/share/doc/jemalloc/jemalloc.html ${install_dir}/jemalloc/share/doc/jemalloc
+    fi
+    if [ -f ${build_dir}/share/man/man3/jemalloc.3 ]; then
+        cp ${build_dir}/share/man/man3/jemalloc.3 ${install_dir}/jemalloc/share/man/man3
+    fi
+fi
+
 sed -i '/dataDir/ {s/taos/power/g}'  ${install_dir}/cfg/taos.cfg
 sed -i '/logDir/  {s/taos/power/g}'  ${install_dir}/cfg/taos.cfg
 sed -i "s/TDengine/PowerDB/g"        ${install_dir}/cfg/taos.cfg

packaging/tools/makepkg.sh

@@ -215,7 +215,7 @@ pkg_name=${install_dir}-${osType}-${cpuType}
 #   exit 1
 # fi
 
-if [ "$verType" == "beta" ]; then
+if [[ "$verType" == "beta" ]] || [[ "$verType" == "preRelease" ]]; then
   pkg_name=${install_dir}-${verType}-${osType}-${cpuType} 
 elif [ "$verType" == "stable" ]; then
   pkg_name=${pkg_name}

snap/snapcraft.yaml

 name: tdengine
 base: core18
-version: '2.1.4.1'
+version: '2.1.5.0'
 icon: snap/gui/t-dengine.svg
 summary: an open-source big data platform designed and optimized for IoT. 
 description: |
@@ -72,7 +72,7 @@ parts:
       - usr/bin/taosd
       - usr/bin/taos
       - usr/bin/taosdemo
-      - usr/lib/libtaos.so.2.1.4.1
+      - usr/lib/libtaos.so.2.1.5.0
       - usr/lib/libtaos.so.1
       - usr/lib/libtaos.so
 

src/balance/src/bnThread.c

@@ -23,6 +23,8 @@
 static SBnThread tsBnThread;
 
 static void *bnThreadFunc(void *arg) {
+  setThreadName("bnThreadd");
+
   while (1) {
     pthread_mutex_lock(&tsBnThread.mutex);
     if (tsBnThread.stop) {

src/client/CMakeLists.txt

@@ -11,7 +11,7 @@ IF (TD_LINUX)
 
   # set the static lib name
   ADD_LIBRARY(taos_static STATIC ${SRC})
-  TARGET_LINK_LIBRARIES(taos_static common query trpc tutil pthread m rt)
+  TARGET_LINK_LIBRARIES(taos_static common query trpc tutil pthread m rt ${VAR_TSZ})
   SET_TARGET_PROPERTIES(taos_static PROPERTIES OUTPUT_NAME "taos_static")
   SET_TARGET_PROPERTIES(taos_static PROPERTIES CLEAN_DIRECT_OUTPUT 1)
 

src/client/inc/tscUtil.h

@@ -94,11 +94,23 @@ typedef struct SVgroupTableInfo {
   SArray     *itemList;   // SArray<STableIdInfo>
 } SVgroupTableInfo;
 
+typedef struct SBlockKeyTuple {
+  TSKEY skey;
+  void* payloadAddr;
+} SBlockKeyTuple;
+
+typedef struct SBlockKeyInfo {
+  int32_t         maxBytesAlloc;
+  SBlockKeyTuple* pKeyTuple;
+} SBlockKeyInfo;
+
 int32_t converToStr(char *str, int type, void *buf, int32_t bufSize, int32_t *len);
 
 int32_t tscCreateDataBlock(size_t initialSize, int32_t rowSize, int32_t startOffset, SName* name, STableMeta* pTableMeta, STableDataBlocks** dataBlocks);
 void tscDestroyDataBlock(STableDataBlocks* pDataBlock, bool removeMeta);
-void tscSortRemoveDataBlockDupRows(STableDataBlocks* dataBuf);
+void    tscSortRemoveDataBlockDupRowsRaw(STableDataBlocks* dataBuf);
+int     tscSortRemoveDataBlockDupRows(STableDataBlocks* dataBuf, SBlockKeyInfo* pBlkKeyInfo);
+int32_t tsSetBlockInfo(SSubmitBlk *pBlocks, const STableMeta *pTableMeta, int32_t numOfRows);
 
 void tscDestroyBoundColumnInfo(SParsedDataColInfo* pColInfo);
 void doRetrieveSubqueryData(SSchedMsg *pMsg);
@@ -261,7 +273,7 @@ void tscVgroupTableCopy(SVgroupTableInfo* info, SVgroupTableInfo* pInfo);
 
 int  tscGetSTableVgroupInfo(SSqlObj* pSql, SQueryInfo* pQueryInfo);
 int  tscGetTableMeta(SSqlObj* pSql, STableMetaInfo* pTableMetaInfo);
-int  tscGetTableMetaEx(SSqlObj* pSql, STableMetaInfo* pTableMetaInfo, bool createIfNotExists);
+int  tscGetTableMetaEx(SSqlObj *pSql, STableMetaInfo *pTableMetaInfo, bool createIfNotExists, bool onlyLocal);
 int32_t tscGetUdfFromNode(SSqlObj *pSql, SQueryInfo* pQueryInfo);
 
 void tscResetForNextRetrieve(SSqlRes* pRes);

src/client/inc/tsclient.h

@@ -88,13 +88,43 @@ typedef struct SBoundColumn {
   int32_t offset;  // all column offset value
 } SBoundColumn;
 
+typedef struct {
+  uint16_t schemaColIdx;
+  uint16_t boundIdx;
+  uint16_t finalIdx;
+} SBoundIdxInfo;
+
+typedef enum _COL_ORDER_STATUS {
+  ORDER_STATUS_UNKNOWN = 0,
+  ORDER_STATUS_ORDERED = 1,
+  ORDER_STATUS_DISORDERED = 2,
+} EOrderStatus;
+
 typedef struct SParsedDataColInfo {
   int16_t        numOfCols;
   int16_t        numOfBound;
-  int32_t        *boundedColumns;
-  SBoundColumn   *cols;
+  int32_t *      boundedColumns;  // bounded column idx according to schema
+  SBoundColumn * cols;
+  SBoundIdxInfo *colIdxInfo;
+  int8_t         orderStatus;  // bounded columns: 
 } SParsedDataColInfo;
 
+#define IS_DATA_COL_ORDERED(s) ((s) == (int8_t)ORDER_STATUS_ORDERED)
+
+typedef struct {
+  SSchema *   pSchema;
+  int16_t     sversion;
+  int32_t     flen;
+  uint16_t    nCols;
+  void *      buf;
+  void *      pDataBlock;
+  SSubmitBlk *pSubmitBlk;
+} SMemRowBuilder;
+
+typedef struct {
+  TDRowLenT allNullLen;
+} SMemRowHelper;
+
 typedef struct STableDataBlocks {
   SName       tableName;
   int8_t      tsSource;     // where does the UNIX timestamp come from, server or client
@@ -108,13 +138,15 @@ typedef struct STableDataBlocks {
   uint32_t    size;
   STableMeta *pTableMeta;   // the tableMeta of current table, the table meta will be used during submit, keep a ref to avoid to be removed from cache
   char       *pData;
+  bool        cloned;
   
   SParsedDataColInfo boundColumnInfo;
 
   // for parameter ('?') binding
   uint32_t       numOfAllocedParams;
   uint32_t       numOfParams;
-  SParamInfo *params;
+  SParamInfo *   params;
+  SMemRowHelper  rowHelper;
 } STableDataBlocks;
 
 typedef struct {
@@ -128,6 +160,7 @@ typedef struct SInsertStatementParam {
   SHashObj    *pTableBlockHashList;     // data block for each table
   SArray      *pDataBlocks;             // SArray<STableDataBlocks*>. Merged submit block for each vgroup
   int8_t       schemaAttached;          // denote if submit block is built with table schema or not
+  uint8_t      payloadType;             // EPayloadType. 0: K-V payload for non-prepare insert, 1: rawPayload for prepare insert
   STagData     tagData;                 // NOTE: pTagData->data is used as a variant length array
 
   int32_t      batchSize;               // for parameter ('?') binding and batch processing
@@ -139,6 +172,14 @@ typedef struct SInsertStatementParam {
   char        *sql;                     // current sql statement position
 } SInsertStatementParam;
 
+typedef enum {
+  PAYLOAD_TYPE_KV = 0,
+  PAYLOAD_TYPE_RAW = 1,
+} EPayloadType;
+
+#define IS_RAW_PAYLOAD(t) \
+  (((int)(t)) == PAYLOAD_TYPE_RAW)  // 0: K-V payload for non-prepare insert, 1: rawPayload for prepare insert
+
 // TODO extract sql parser supporter
 typedef struct {
   int     command;
@@ -243,6 +284,7 @@ typedef struct SSqlObj {
   void *           pStream;
   void *           pSubscription;
   char *           sqlstr;
+  void *           pBuf;  // table meta buffer
   char             parseRetry;
   char             retry;
   char             maxRetry;
@@ -370,6 +412,7 @@ int32_t tscSQLSyntaxErrMsg(char* msg, const char* additionalInfo,  const char* s
 
 int32_t tscValidateSqlInfo(SSqlObj *pSql, struct SSqlInfo *pInfo);
 
+int32_t tsSetBlockInfo(SSubmitBlk *pBlocks, const STableMeta *pTableMeta, int32_t numOfRows);
 extern int32_t    sentinel;
 extern SHashObj  *tscVgroupMap;
 extern SHashObj  *tscTableMetaInfo;
@@ -386,6 +429,11 @@ extern int (*tscBuildMsg[TSDB_SQL_MAX])(SSqlObj *pSql, SSqlInfo *pInfo);
 void tscBuildVgroupTableInfo(SSqlObj* pSql, STableMetaInfo* pTableMetaInfo, SArray* tables);
 int16_t getNewResColId(SSqlCmd* pCmd);
 
+int32_t schemaIdxCompar(const void *lhs, const void *rhs);
+int32_t boundIdxCompar(const void *lhs, const void *rhs);
+int     initSMemRowHelper(SMemRowHelper *pHelper, SSchema *pSSchema, uint16_t nCols, uint16_t allNullColsLen);
+int32_t getExtendedRowSize(STableComInfo *tinfo);
+
 #ifdef __cplusplus
 }
 #endif

src/client/src/TSDBJNIConnector.c

@@ -946,3 +946,34 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_setTableNameTagsI
 
   return JNI_SUCCESS;
 }
+
+JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_insertLinesImp(JNIEnv *env, jobject jobj,
+                                                                                jobjectArray lines, jlong conn) {
+  TAOS *taos = (TAOS *)conn;
+  if (taos == NULL) {
+    jniError("jobj:%p, connection already closed", jobj);
+    return JNI_CONNECTION_NULL;
+  }
+
+  int numLines = (*env)->GetArrayLength(env, lines);
+  char** c_lines = calloc(numLines, sizeof(char*));
+
+  for (int i = 0; i < numLines; ++i) {
+    jstring line = (jstring) ((*env)->GetObjectArrayElement(env, lines, i));
+    c_lines[i] = (char*)(*env)->GetStringUTFChars(env, line, 0);
+  }
+
+  int code = taos_insert_lines(taos, c_lines, numLines);
+
+  for (int i = 0; i < numLines; ++i) {
+    jstring line = (jstring) ((*env)->GetObjectArrayElement(env, lines, i));
+    (*env)->ReleaseStringUTFChars(env, line, c_lines[i]);
+  }
+
+  if (code != TSDB_CODE_SUCCESS) {
+    jniError("jobj:%p, conn:%p, code:%s", jobj, taos, tstrerror(code));
+    return JNI_TDENGINE_ERROR;
+  }
+
+  return code;
+}
\ No newline at end of file

src/client/src/taos.def

@@ -7,11 +7,16 @@ taos_connect_auth
 taos_close
 taos_stmt_init
 taos_stmt_prepare
+taos_stmt_set_tbname_tags
+taos_stmt_set_tbname
+taos_stmt_is_insert
+taos_stmt_num_params
 taos_stmt_bind_param
 taos_stmt_add_batch
 taos_stmt_execute
 taos_stmt_use_result
 taos_stmt_close
+taos_stmt_errstr
 taos_query
 taos_fetch_row
 taos_result_precision
@@ -37,6 +42,4 @@ taos_consume
 taos_unsubscribe
 taos_open_stream
 taos_close_stream
-taos_fetch_block
 taos_load_table_info
-

src/client/src/tscPrepare.c

@@ -47,6 +47,7 @@ typedef struct SNormalStmt {
 typedef struct SMultiTbStmt {
   bool      nameSet;
   bool      tagSet;
+  bool      subSet;
   uint64_t  currentUid;
   char     *sqlstr;
   uint32_t  tbNum;
@@ -54,6 +55,7 @@ typedef struct SMultiTbStmt {
   SStrToken stbname;
   SStrToken values;
   SArray   *tags;
+  STableDataBlocks *lastBlock;
   SHashObj *pTableHash;
   SHashObj *pTableBlockHashList;     // data block for each table
 } SMultiTbStmt;
@@ -347,7 +349,7 @@ int32_t fillTablesColumnsNull(SSqlObj* pSql) {
 
 ////////////////////////////////////////////////////////////////////////////////
 // functions for insertion statement preparation
-static int doBindParam(STableDataBlocks* pBlock, char* data, SParamInfo* param, TAOS_BIND* bind, int32_t colNum) {
+static FORCE_INLINE int doBindParam(STableDataBlocks* pBlock, char* data, SParamInfo* param, TAOS_BIND* bind, int32_t colNum) {
     if (bind->is_null != NULL && *(bind->is_null)) {
       setNull(data + param->offset, param->type, param->bytes);
       return TSDB_CODE_SUCCESS;
@@ -746,25 +748,25 @@ static int doBindParam(STableDataBlocks* pBlock, char* data, SParamInfo* param,
     case TSDB_DATA_TYPE_BOOL:
     case TSDB_DATA_TYPE_TINYINT:
     case TSDB_DATA_TYPE_UTINYINT:
-      size = 1;
+      *(uint8_t *)(data + param->offset) = *(uint8_t *)bind->buffer;
       break;
 
     case TSDB_DATA_TYPE_SMALLINT:
     case TSDB_DATA_TYPE_USMALLINT:
-      size = 2;
+      *(uint16_t *)(data + param->offset) = *(uint16_t *)bind->buffer;
       break;
 
     case TSDB_DATA_TYPE_INT:
     case TSDB_DATA_TYPE_UINT:
     case TSDB_DATA_TYPE_FLOAT:
-      size = 4;
+      *(uint32_t *)(data + param->offset) = *(uint32_t *)bind->buffer;
       break;
 
     case TSDB_DATA_TYPE_BIGINT:
     case TSDB_DATA_TYPE_UBIGINT:
     case TSDB_DATA_TYPE_DOUBLE:
     case TSDB_DATA_TYPE_TIMESTAMP:
-      size = 8;
+      *(uint64_t *)(data + param->offset) = *(uint64_t *)bind->buffer;
       break;
 
     case TSDB_DATA_TYPE_BINARY:
@@ -790,7 +792,6 @@ static int doBindParam(STableDataBlocks* pBlock, char* data, SParamInfo* param,
       return TSDB_CODE_TSC_INVALID_VALUE;
   }
 
-  memcpy(data + param->offset, bind->buffer, size);
   if (param->offset == 0) {
     if (tsCheckTimestamp(pBlock, data + param->offset) != TSDB_CODE_SUCCESS) {
       tscError("invalid timestamp");
@@ -801,6 +802,58 @@ static int doBindParam(STableDataBlocks* pBlock, char* data, SParamInfo* param,
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t insertStmtGenLastBlock(STableDataBlocks** lastBlock, STableDataBlocks* pBlock) {
+  *lastBlock = (STableDataBlocks*)malloc(sizeof(STableDataBlocks));
+  memcpy(*lastBlock, pBlock, sizeof(STableDataBlocks));
+  (*lastBlock)->cloned = true;
+  
+  (*lastBlock)->pData    = NULL;
+  (*lastBlock)->ordered  = true;
+  (*lastBlock)->prevTS   = INT64_MIN;
+  (*lastBlock)->size     = sizeof(SSubmitBlk);
+  (*lastBlock)->tsSource = -1;
+
+  return TSDB_CODE_SUCCESS;
+}
+
+
+static int32_t insertStmtGenBlock(STscStmt* pStmt, STableDataBlocks** pBlock, STableMeta* pTableMeta, SName* name) {
+  int32_t code = 0;
+  
+  if (pStmt->mtb.lastBlock == NULL) {
+    tscError("no previous data block");
+    return TSDB_CODE_TSC_APP_ERROR;
+  }
+
+  int32_t msize = tscGetTableMetaSize(pTableMeta);
+  int32_t tsize = sizeof(STableDataBlocks) + msize;
+
+  void *t = malloc(tsize);
+  *pBlock = t;
+
+  memcpy(*pBlock, pStmt->mtb.lastBlock, sizeof(STableDataBlocks));
+
+  t = (char *)t + sizeof(STableDataBlocks);
+  (*pBlock)->pTableMeta = t;
+  memcpy((*pBlock)->pTableMeta, pTableMeta, msize);
+
+  (*pBlock)->pData = malloc((*pBlock)->nAllocSize);
+
+  (*pBlock)->vgId     = (*pBlock)->pTableMeta->vgId;
+
+  tNameAssign(&(*pBlock)->tableName, name);
+  
+  SSubmitBlk* blk = (SSubmitBlk*)(*pBlock)->pData;
+  memset(blk, 0, sizeof(*blk));
+  
+  code = tsSetBlockInfo(blk, pTableMeta, 0);
+  if (code != TSDB_CODE_SUCCESS) {
+    STMT_RET(code);
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
 
 static int doBindBatchParam(STableDataBlocks* pBlock, SParamInfo* param, TAOS_MULTI_BIND* bind, int32_t rowNum) {
   if (bind->buffer_type != param->type || !isValidDataType(param->type)) {
@@ -1227,11 +1280,11 @@ int stmtParseInsertTbTags(SSqlObj* pSql, STscStmt* pStmt) {
     pStmt->mtb.tbname = sToken;
     pStmt->mtb.nameSet = false;
     if (pStmt->mtb.pTableHash == NULL) {
-      pStmt->mtb.pTableHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, false);
+      pStmt->mtb.pTableHash = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, false);
     }
 
     if (pStmt->mtb.pTableBlockHashList == NULL) {
-      pStmt->mtb.pTableBlockHashList = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, false);
+      pStmt->mtb.pTableBlockHashList = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, false);
     }
 
     pStmt->mtb.tagSet = true;
@@ -1522,6 +1575,7 @@ int taos_stmt_prepare(TAOS_STMT* stmt, const char* sql, unsigned long length) {
 
 int taos_stmt_set_tbname_tags(TAOS_STMT* stmt, const char* name, TAOS_BIND* tags) {
   STscStmt* pStmt = (STscStmt*)stmt;
+  int32_t code = 0;
 
   if (stmt == NULL || pStmt->pSql == NULL || pStmt->taos == NULL) {
     STMT_RET(TSDB_CODE_TSC_DISCONNECTED);
@@ -1559,6 +1613,11 @@ int taos_stmt_set_tbname_tags(TAOS_STMT* stmt, const char* name, TAOS_BIND* tags
 
     SSubmitBlk* pBlk = (SSubmitBlk*) (*t1)->pData;
     pCmd->batchSize = pBlk->numOfRows;
+    if (pBlk->numOfRows == 0) {
+      (*t1)->prevTS = INT64_MIN;
+    }
+
+    tsSetBlockInfo(pBlk, (*t1)->pTableMeta, pBlk->numOfRows);
 
     taosHashPut(pCmd->insertParam.pTableBlockHashList, (void *)&pStmt->mtb.currentUid, sizeof(pStmt->mtb.currentUid), (void*)t1, POINTER_BYTES);
 
@@ -1566,6 +1625,51 @@ int taos_stmt_set_tbname_tags(TAOS_STMT* stmt, const char* name, TAOS_BIND* tags
     STMT_RET(TSDB_CODE_SUCCESS);
   }
 
+  if (pStmt->mtb.subSet && taosHashGetSize(pStmt->mtb.pTableHash) > 0) {
+    STableMetaInfo* pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, 0);
+    STableMeta* pTableMeta = pTableMetaInfo->pTableMeta;
+    char sTableName[TSDB_TABLE_FNAME_LEN];
+    strncpy(sTableName, pTableMeta->sTableName, sizeof(sTableName));
+
+    SStrToken tname = {0};
+    tname.type = TK_STRING;
+    tname.z = (char *)name;
+    tname.n = (uint32_t)strlen(name);
+    SName fullname = {0};
+    tscSetTableFullName(&fullname, &tname, pSql);
+
+    memcpy(&pTableMetaInfo->name, &fullname, sizeof(fullname));
+
+    code = tscGetTableMetaEx(pSql, pTableMetaInfo, false, true);
+    if (code != TSDB_CODE_SUCCESS) {
+      STMT_RET(code);
+    }
+
+    pTableMeta = pTableMetaInfo->pTableMeta;
+
+    if (strcmp(sTableName, pTableMeta->sTableName)) {
+      tscError("0x%"PRIx64" only tables belongs to one stable is allowed", pSql->self);
+      STMT_RET(TSDB_CODE_TSC_APP_ERROR);
+    }
+
+    STableDataBlocks* pBlock = NULL;
+
+    insertStmtGenBlock(pStmt, &pBlock, pTableMeta, &pTableMetaInfo->name);
+
+    pCmd->batchSize = 0;
+    
+    pStmt->mtb.currentUid = pTableMeta->id.uid;
+    pStmt->mtb.tbNum++;
+    
+    taosHashPut(pCmd->insertParam.pTableBlockHashList, (void *)&pStmt->mtb.currentUid, sizeof(pStmt->mtb.currentUid), (void*)&pBlock, POINTER_BYTES);
+    taosHashPut(pStmt->mtb.pTableBlockHashList, (void *)&pStmt->mtb.currentUid, sizeof(pStmt->mtb.currentUid), (void*)&pBlock, POINTER_BYTES);
+    taosHashPut(pStmt->mtb.pTableHash, name, strlen(name), (char*) &pTableMeta->id.uid, sizeof(pTableMeta->id.uid));
+
+    tscDebug("0x%"PRIx64" table:%s is prepared, uid:%" PRIx64, pSql->self, name, pStmt->mtb.currentUid);
+    
+    STMT_RET(TSDB_CODE_SUCCESS);
+  }
+
   if (pStmt->mtb.tagSet) {
     pStmt->mtb.tbname = tscReplaceStrToken(&pSql->sqlstr, &pStmt->mtb.tbname, name);
   } else {
@@ -1594,7 +1698,7 @@ int taos_stmt_set_tbname_tags(TAOS_STMT* stmt, const char* name, TAOS_BIND* tags
     pCmd->insertParam.pTableBlockHashList = hashList;
   }
 
-  int32_t code = tsParseSql(pStmt->pSql, true);
+  code = tsParseSql(pStmt->pSql, true);
   if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
     // wait for the callback function to post the semaphore
     tsem_wait(&pStmt->pSql->rspSem);
@@ -1622,6 +1726,10 @@ int taos_stmt_set_tbname_tags(TAOS_STMT* stmt, const char* name, TAOS_BIND* tags
     taosHashPut(pStmt->mtb.pTableBlockHashList, (void *)&pStmt->mtb.currentUid, sizeof(pStmt->mtb.currentUid), (void*)&pBlock, POINTER_BYTES);
     taosHashPut(pStmt->mtb.pTableHash, name, strlen(name), (char*) &pTableMeta->id.uid, sizeof(pTableMeta->id.uid));
 
+    if (pStmt->mtb.lastBlock == NULL) {
+      insertStmtGenLastBlock(&pStmt->mtb.lastBlock, pBlock);
+    }
+
     tscDebug("0x%"PRIx64" table:%s is prepared, uid:%" PRIx64, pSql->self, name, pStmt->mtb.currentUid);
   }
 
@@ -1629,7 +1737,17 @@ int taos_stmt_set_tbname_tags(TAOS_STMT* stmt, const char* name, TAOS_BIND* tags
 }
 
 
+int taos_stmt_set_sub_tbname(TAOS_STMT* stmt, const char* name) {
+  STscStmt* pStmt = (STscStmt*)stmt;
+  pStmt->mtb.subSet = true;
+  return taos_stmt_set_tbname_tags(stmt, name, NULL);
+}
+
+
+
 int taos_stmt_set_tbname(TAOS_STMT* stmt, const char* name) {
+  STscStmt* pStmt = (STscStmt*)stmt;
+  pStmt->mtb.subSet = false;
   return taos_stmt_set_tbname_tags(stmt, name, NULL);
 }
 
@@ -1653,6 +1771,7 @@ int taos_stmt_close(TAOS_STMT* stmt) {
       if (pStmt->pSql && pStmt->pSql->res.code != 0) {
         rmMeta = true;
       }
+      tscDestroyDataBlock(pStmt->mtb.lastBlock, rmMeta);
       pStmt->mtb.pTableBlockHashList = tscDestroyBlockHashTable(pStmt->mtb.pTableBlockHashList, rmMeta);
       taosHashCleanup(pStmt->pSql->cmd.insertParam.pTableBlockHashList);
       pStmt->pSql->cmd.insertParam.pTableBlockHashList = NULL;
@@ -1687,6 +1806,8 @@ int taos_stmt_bind_param(TAOS_STMT* stmt, TAOS_BIND* bind) {
 
     pStmt->last = STMT_BIND;
 
+    tscDebug("tableId:%" PRIu64 ", try to bind one row", pStmt->mtb.currentUid);
+
     STMT_RET(insertStmtBindParam(pStmt, bind));
   } else {
     STMT_RET(normalStmtBindParam(pStmt, bind));
@@ -1806,6 +1927,7 @@ int taos_stmt_execute(TAOS_STMT* stmt) {
 
     pStmt->last = STMT_EXECUTE;
 
+    pStmt->pSql->cmd.insertParam.payloadType = PAYLOAD_TYPE_RAW;
     if (pStmt->multiTbInsert) {
       ret = insertBatchStmtExecute(pStmt);
     } else {

src/client/src/tscProfile.c

@@ -16,6 +16,7 @@
 #include "os.h"
 #include "tscLog.h"
 #include "tsclient.h"
+#include "tsocket.h"
 #include "ttimer.h"
 #include "tutil.h"
 #include "taosmsg.h"
@@ -252,6 +253,16 @@ int tscBuildQueryStreamDesc(void *pMsg, STscObj *pObj) {
     //pQdesc->useconds = htobe64(pSql->res.useconds);
     pQdesc->useconds = htobe64(now - pSql->stime);
     pQdesc->qId = htobe64(pSql->res.qId);
+    pQdesc->sqlObjId = htobe64(pSql->self);
+    pQdesc->pid = pHeartbeat->pid;
+    if (pSql->cmd.pQueryInfo->stableQuery == true) {
+      pQdesc->numOfSub = pSql->subState.numOfSub;
+    } else {
+      pQdesc->numOfSub = 1;
+    }
+    pQdesc->numOfSub = htonl(pQdesc->numOfSub);
+
+    taosGetFqdn(pQdesc->fqdn);
 
     pHeartbeat->numOfQueries++;
     pQdesc++;

src/client/src/tscSQLParser.c

@@ -1396,12 +1396,16 @@ static bool validateTableColumnInfo(SArray* pFieldList, SSqlCmd* pCmd) {
   const char* msg4 = "invalid data type";
   const char* msg5 = "invalid binary/nchar column length";
   const char* msg6 = "invalid column name";
+  const char* msg7 = "too many columns";
 
   // number of fields no less than 2
   size_t numOfCols = taosArrayGetSize(pFieldList);
-  if (numOfCols <= 1 || numOfCols > TSDB_MAX_COLUMNS) {
+  if (numOfCols <= 1 ) {
     invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg);
     return false;
+  } else if (numOfCols > TSDB_MAX_COLUMNS) {
+    invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg7);
+    return false;
   }
 
   // first column must be timestamp
@@ -1536,6 +1540,7 @@ bool validateOneTags(SSqlCmd* pCmd, TAOS_FIELD* pTagField) {
   const char* msg4 = "invalid tag name";
   const char* msg5 = "invalid binary/nchar tag length";
   const char* msg6 = "invalid data type in tags";
+  const char* msg7 = "too many columns";
 
   STableMetaInfo* pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd,  0);
   STableMeta*     pTableMeta = pTableMetaInfo->pTableMeta;
@@ -1543,6 +1548,12 @@ bool validateOneTags(SSqlCmd* pCmd, TAOS_FIELD* pTagField) {
   int32_t numOfTags = tscGetNumOfTags(pTableMeta);
   int32_t numOfCols = tscGetNumOfColumns(pTableMeta);
 
+  // no more max columns
+  if (numOfTags + numOfCols >= TSDB_MAX_COLUMNS) {
+    invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg7);
+    return false;
+  }
+
   // no more than 6 tags
   if (numOfTags == TSDB_MAX_TAGS) {
     char msg[128] = {0};
@@ -2012,6 +2023,11 @@ static int32_t checkForUdf(SSqlObj* pSql, SQueryInfo* pQueryInfo, SArray* pSelec
 */
 
 static SUdfInfo* isValidUdf(SArray* pUdfInfo, const char* name, int32_t len) {
+  if(pUdfInfo == NULL){
+    tscError("udfinfo is null");
+    return NULL;
+  }
+  
   size_t t = taosArrayGetSize(pUdfInfo);
   for(int32_t i = 0; i < t; ++i) {
     SUdfInfo* pUdf = taosArrayGet(pUdfInfo, i);
@@ -5999,6 +6015,16 @@ int32_t setAlterTableInfo(SSqlObj* pSql, struct SSqlInfo* pInfo) {
       return invalidOperationMsg(pMsg, msg22);
     }
 
+    SSchema* pSchema = (SSchema*) pTableMetaInfo->pTableMeta->schema;
+    int16_t numOfColumns = pTableMetaInfo->pTableMeta->tableInfo.numOfColumns;
+    int16_t i;
+    uint32_t nLen = 0;
+    for (i = 0; i < numOfColumns; ++i) {
+      nLen += (i != columnIndex.columnIndex) ? pSchema[i].bytes : pItem->bytes;
+    }
+    if (nLen >= TSDB_MAX_BYTES_PER_ROW) {
+      return invalidOperationMsg(pMsg, msg24);
+    }
     TAOS_FIELD f = tscCreateField(pColSchema->type, name.z, pItem->bytes);
     tscFieldInfoAppend(&pQueryInfo->fieldsInfo, &f);
   }else if (pAlterSQL->type == TSDB_ALTER_TABLE_MODIFY_TAG_COLUMN) {
@@ -6040,6 +6066,17 @@ int32_t setAlterTableInfo(SSqlObj* pSql, struct SSqlInfo* pInfo) {
       return invalidOperationMsg(pMsg, msg22);
     }
 
+    SSchema* pSchema = tscGetTableTagSchema(pTableMetaInfo->pTableMeta);
+    int16_t numOfTags = tscGetNumOfTags(pTableMetaInfo->pTableMeta);
+    int16_t i;
+    uint32_t nLen = 0;
+    for (i = 0; i < numOfTags; ++i) {
+      nLen += (i != columnIndex.columnIndex) ? pSchema[i].bytes : pItem->bytes;
+    }
+    if (nLen >= TSDB_MAX_TAGS_LEN) {
+      return invalidOperationMsg(pMsg, msg24);
+    }
+
     TAOS_FIELD f = tscCreateField(pColSchema->type, name.z, pItem->bytes);
     tscFieldInfoAppend(&pQueryInfo->fieldsInfo, &f);
   }
@@ -8005,6 +8042,28 @@ static void freeElem(void* p) {
   tfree(*(char**)p);
 }
 
+int32_t tnameComparFn(const void* p1, const void* p2) {
+  SName* pn1 = (SName*)p1;
+  SName* pn2 = (SName*)p2;
+
+  int32_t ret = strncmp(pn1->acctId, pn2->acctId, tListLen(pn1->acctId));
+  if (ret != 0) {
+    return ret > 0? 1:-1;
+  } else {
+    ret = strncmp(pn1->dbname, pn2->dbname, tListLen(pn1->dbname));
+    if (ret != 0) {
+      return ret > 0? 1:-1;
+    } else {
+      ret = strncmp(pn1->tname, pn2->tname, tListLen(pn1->tname));
+      if (ret != 0) {
+        return ret > 0? 1:-1;
+      } else {
+        return 0;
+      }
+    }
+  }
+}
+
 int32_t loadAllTableMeta(SSqlObj* pSql, struct SSqlInfo* pInfo) {
   SSqlCmd* pCmd = &pSql->cmd;
 
@@ -8048,13 +8107,20 @@ int32_t loadAllTableMeta(SSqlObj* pSql, struct SSqlInfo* pInfo) {
   uint32_t maxSize = tscGetTableMetaMaxSize();
   char     name[TSDB_TABLE_FNAME_LEN] = {0};
 
-  char buf[80 * 1024] = {0};
-  assert(maxSize < 80 * 1024);
+  assert(maxSize < 80 * TSDB_MAX_COLUMNS);
+  if (!pSql->pBuf) {
+    if (NULL == (pSql->pBuf = tcalloc(1, 80 * TSDB_MAX_COLUMNS))) {
+      return TSDB_CODE_TSC_OUT_OF_MEMORY;
+    }
+  }
   pTableMeta = calloc(1, maxSize);
 
   plist = taosArrayInit(4, POINTER_BYTES);
   pVgroupList = taosArrayInit(4, POINTER_BYTES);
 
+  taosArraySort(tableNameList, tnameComparFn);
+  taosArrayRemoveDuplicate(tableNameList, tnameComparFn, NULL);
+
   size_t numOfTables = taosArrayGetSize(tableNameList);
   for (int32_t i = 0; i < numOfTables; ++i) {
     SName* pname = taosArrayGet(tableNameList, i);
@@ -8066,7 +8132,7 @@ int32_t loadAllTableMeta(SSqlObj* pSql, struct SSqlInfo* pInfo) {
 
     if (pTableMeta->id.uid > 0) {
       if (pTableMeta->tableType == TSDB_CHILD_TABLE) {
-        code = tscCreateTableMetaFromSTableMeta(pTableMeta, name, buf);
+        code = tscCreateTableMetaFromSTableMeta(pTableMeta, name, pSql->pBuf);
 
         // create the child table meta from super table failed, try load it from mnode
         if (code != TSDB_CODE_SUCCESS) {
@@ -8075,8 +8141,7 @@ int32_t loadAllTableMeta(SSqlObj* pSql, struct SSqlInfo* pInfo) {
           continue;
         }
       } else if (pTableMeta->tableType == TSDB_SUPER_TABLE) {
-        // the vgroup list of a super table is not kept in local buffer, so here need retrieve it
-        // from the mnode each time
+        // the vgroup list of super table is not kept in local buffer, so here need retrieve it from the mnode each time
         char* t = strdup(name);
         taosArrayPush(pVgroupList, &t);
       }
@@ -8103,6 +8168,7 @@ int32_t loadAllTableMeta(SSqlObj* pSql, struct SSqlInfo* pInfo) {
   if (pInfo->funcs) {
     funcSize = taosArrayGetSize(pInfo->funcs);
   }
+
   if (funcSize > 0) {
     for (size_t i = 0; i < funcSize; ++i) {
       SStrToken* t = taosArrayGet(pInfo->funcs, i);

src/client/src/tscServer.c

@@ -2776,7 +2776,7 @@ int32_t getMultiTableMetaFromMnode(SSqlObj *pSql, SArray* pNameList, SArray* pVg
   return code;
 }
 
-int32_t tscGetTableMetaImpl(SSqlObj* pSql, STableMetaInfo *pTableMetaInfo, bool autocreate) {
+int32_t tscGetTableMetaImpl(SSqlObj* pSql, STableMetaInfo *pTableMetaInfo, bool autocreate, bool onlyLocal) {
   assert(tIsValidName(&pTableMetaInfo->name));
 
   uint32_t size = tscGetTableMetaMaxSize();
@@ -2804,31 +2804,38 @@ int32_t tscGetTableMetaImpl(SSqlObj* pSql, STableMetaInfo *pTableMetaInfo, bool
   taosHashGetClone(tscTableMetaInfo, name, len, NULL, pTableMetaInfo->pTableMeta, -1);
 
   // TODO resize the tableMeta
-  char buf[80*1024] = {0};
-  assert(size < 80*1024);
+  assert(size < 80 * TSDB_MAX_COLUMNS);
+  if (!pSql->pBuf) {
+    if (NULL == (pSql->pBuf = tcalloc(1, 80 * TSDB_MAX_COLUMNS))) {
+      return TSDB_CODE_TSC_OUT_OF_MEMORY;
+    }
+  }
 
   STableMeta* pMeta = pTableMetaInfo->pTableMeta;
   if (pMeta->id.uid > 0) {
     // in case of child table, here only get the
     if (pMeta->tableType == TSDB_CHILD_TABLE) {
-      int32_t code = tscCreateTableMetaFromSTableMeta(pTableMetaInfo->pTableMeta, name, buf);
+      int32_t code = tscCreateTableMetaFromSTableMeta(pTableMetaInfo->pTableMeta, name, pSql->pBuf);
       if (code != TSDB_CODE_SUCCESS) {
         return getTableMetaFromMnode(pSql, pTableMetaInfo, autocreate);
       }
     }
-
     return TSDB_CODE_SUCCESS;
   }
 
+  if (onlyLocal) {
+    return TSDB_CODE_TSC_NO_META_CACHED;
+  }
+  
   return getTableMetaFromMnode(pSql, pTableMetaInfo, autocreate);
 }
 
 int32_t tscGetTableMeta(SSqlObj *pSql, STableMetaInfo *pTableMetaInfo) {
-  return tscGetTableMetaImpl(pSql, pTableMetaInfo, false);
+  return tscGetTableMetaImpl(pSql, pTableMetaInfo, false, false);
 }
 
-int tscGetTableMetaEx(SSqlObj *pSql, STableMetaInfo *pTableMetaInfo, bool createIfNotExists) {
-  return tscGetTableMetaImpl(pSql, pTableMetaInfo, createIfNotExists);
+int tscGetTableMetaEx(SSqlObj *pSql, STableMetaInfo *pTableMetaInfo, bool createIfNotExists, bool onlyLocal) {
+  return tscGetTableMetaImpl(pSql, pTableMetaInfo, createIfNotExists, onlyLocal);
 }
 
 int32_t tscGetUdfFromNode(SSqlObj *pSql, SQueryInfo* pQueryInfo) {
@@ -2934,7 +2941,6 @@ int tscGetSTableVgroupInfo(SSqlObj *pSql, SQueryInfo* pQueryInfo) {
   if (allVgroupInfoRetrieved(pQueryInfo)) {
     return TSDB_CODE_SUCCESS;
   }
-
   SSqlObj *pNew = calloc(1, sizeof(SSqlObj));
   pNew->pTscObj = pSql->pTscObj;
   pNew->signature = pNew;

src/client/src/tscSubquery.c

@@ -712,6 +712,15 @@ static void updateQueryTimeRange(SQueryInfo* pQueryInfo, STimeWindow* win) {
   pQueryInfo->window = *win;
 }
 
+int32_t tagValCompar(const void* p1, const void* p2) {
+  const STidTags* t1 = (const STidTags*) varDataVal(p1);
+  const STidTags* t2 = (const STidTags*) varDataVal(p2);
+
+  __compar_fn_t func = getComparFunc(t1->padding, 0);
+
+  return func(t1->tag, t2->tag);
+}
+
 int32_t tidTagsCompar(const void* p1, const void* p2) {
   const STidTags* t1 = (const STidTags*) (p1);
   const STidTags* t2 = (const STidTags*) (p2);
@@ -720,28 +729,7 @@ int32_t tidTagsCompar(const void* p1, const void* p2) {
     return (t1->vgId > t2->vgId) ? 1 : -1;
   }
 
-  tstr* tag1 = (tstr*) t1->tag;
-  tstr* tag2 = (tstr*) t2->tag;
-
-  if (tag1->len != tag2->len) {
-    return (tag1->len > tag2->len)? 1: -1;
-  }
-
-  return strncmp(tag1->data, tag2->data, tag1->len);
-}
-
-int32_t tagValCompar(const void* p1, const void* p2) {
-  const STidTags* t1 = (const STidTags*) varDataVal(p1);
-  const STidTags* t2 = (const STidTags*) varDataVal(p2);
-
-  tstr* tag1 = (tstr*) t1->tag;
-  tstr* tag2 = (tstr*) t2->tag;
-
-  if (tag1->len != tag2->len) {
-    return (tag1->len > tag2->len)? 1: -1;
-  }
-
-  return memcmp(tag1->data, tag2->data, tag1->len);
+  return tagValCompar(p1, p2);
 }
 
 void tscBuildVgroupTableInfo(SSqlObj* pSql, STableMetaInfo* pTableMetaInfo, SArray* tables) {
@@ -871,6 +859,12 @@ static bool checkForDuplicateTagVal(SSchema* pColSchema, SJoinSupporter* p1, SSq
   return true;
 }
 
+static void setTidTagType(SJoinSupporter* p, uint8_t type) {
+  for (int32_t i = 0; i < p->num; ++i) {
+    STidTags * tag = (STidTags*) varDataVal(p->pIdTagList + i * p->tagSize);
+    tag->padding = type;
+  }
+}
 
 static int32_t getIntersectionOfTableTuple(SQueryInfo* pQueryInfo, SSqlObj* pParentSql, SArray* resList) {
   int16_t joinNum = pParentSql->subState.numOfSub;
@@ -890,6 +884,8 @@ static int32_t getIntersectionOfTableTuple(SQueryInfo* pQueryInfo, SSqlObj* pPar
   for (int32_t i = 0; i < joinNum; i++) {
     SJoinSupporter* p = pParentSql->pSubs[i]->param;
 
+    setTidTagType(p, pColSchema->type);
+
     ctxlist[i].p = p;
     ctxlist[i].res = taosArrayInit(p->num, size);
 
@@ -2441,7 +2437,7 @@ int32_t tscHandleMasterSTableQuery(SSqlObj *pSql) {
 
   pRes->qId = 0x1;  // hack the qhandle check
 
-  const uint32_t nBufferSize = (1u << 16u);  // 64KB
+  const uint32_t nBufferSize = (1u << 18u);  // 256KB
 
   SQueryInfo     *pQueryInfo = tscGetQueryInfo(pCmd);
   STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);

src/client/src/tscUtil.c

@@ -1315,8 +1315,7 @@ void tscFreeQueryInfo(SSqlCmd* pCmd, bool removeMeta) {
     return;
   }
 
-  SQueryInfo* pQueryInfo = tscGetQueryInfo(pCmd);
-
+  SQueryInfo* pQueryInfo = pCmd->pQueryInfo;
   while(pQueryInfo != NULL) {
     SQueryInfo* p = pQueryInfo->sibling;
 
@@ -1488,6 +1487,7 @@ void tscFreeSqlObj(SSqlObj* pSql) {
   pSql->signature = NULL;
   pSql->fp = NULL;
   tfree(pSql->sqlstr);
+  tfree(pSql->pBuf);
 
   tfree(pSql->pSubs);
   pSql->subState.numOfSub = 0;
@@ -1508,6 +1508,7 @@ void tscFreeSqlObj(SSqlObj* pSql) {
 void tscDestroyBoundColumnInfo(SParsedDataColInfo* pColInfo) {
   tfree(pColInfo->boundedColumns);
   tfree(pColInfo->cols);
+  tfree(pColInfo->colIdxInfo);
 }
 
 void tscDestroyDataBlock(STableDataBlocks* pDataBlock, bool removeMeta) {
@@ -1516,12 +1517,6 @@ void tscDestroyDataBlock(STableDataBlocks* pDataBlock, bool removeMeta) {
   }
 
   tfree(pDataBlock->pData);
-  tfree(pDataBlock->params);
-
-  // free the refcount for metermeta
-  if (pDataBlock->pTableMeta != NULL) {
-    tfree(pDataBlock->pTableMeta);
-  }
 
   if (removeMeta) {
     char name[TSDB_TABLE_FNAME_LEN] = {0};
@@ -1530,7 +1525,17 @@ void tscDestroyDataBlock(STableDataBlocks* pDataBlock, bool removeMeta) {
     taosHashRemove(tscTableMetaInfo, name, strnlen(name, TSDB_TABLE_FNAME_LEN));
   }
 
+  if (!pDataBlock->cloned) {
+    tfree(pDataBlock->params);
+
+    // free the refcount for metermeta
+    if (pDataBlock->pTableMeta != NULL) {
+      tfree(pDataBlock->pTableMeta);
+    }
+
     tscDestroyBoundColumnInfo(&pDataBlock->boundColumnInfo);
+  }
+
   tfree(pDataBlock);
 }
 
@@ -1596,7 +1601,7 @@ void freeUdfInfo(SUdfInfo* pUdfInfo) {
   taosCloseDll(pUdfInfo->handle);
 }
 
-
+// todo refactor
 void*  tscDestroyUdfArrayList(SArray* pUdfList) {
   if (pUdfList == NULL) {
     return NULL;
@@ -1709,12 +1714,14 @@ int32_t tscCreateDataBlock(size_t defaultSize, int32_t rowSize, int32_t startOff
     dataBuf->nAllocSize = dataBuf->headerSize * 2;
   }
   
-  dataBuf->pData = calloc(1, dataBuf->nAllocSize);
+  //dataBuf->pData = calloc(1, dataBuf->nAllocSize);
+  dataBuf->pData = malloc(dataBuf->nAllocSize);
   if (dataBuf->pData == NULL) {
     tscError("failed to allocated memory, reason:%s", strerror(errno));
     tfree(dataBuf);
     return TSDB_CODE_TSC_OUT_OF_MEMORY;
   }
+  memset(dataBuf->pData, 0, sizeof(SSubmitBlk));
 
   //Here we keep the tableMeta to avoid it to be remove by other threads.
   dataBuf->pTableMeta = tscTableMetaDup(pTableMeta);
@@ -1762,7 +1769,104 @@ int32_t tscGetDataBlockFromList(SHashObj* pHashList, int64_t id, int32_t size, i
   return TSDB_CODE_SUCCESS;
 }
 
-static int trimDataBlock(void* pDataBlock, STableDataBlocks* pTableDataBlock, bool includeSchema) {
+static SMemRow tdGenMemRowFromBuilder(SMemRowBuilder* pBuilder) {
+  SSchema* pSchema = pBuilder->pSchema;
+  char*    p = (char*)pBuilder->buf;
+  int      toffset = 0;
+  uint16_t nCols = pBuilder->nCols;
+
+  uint8_t  memRowType = payloadType(p);
+  uint16_t nColsBound = payloadNCols(p);
+  if (pBuilder->nCols <= 0 || nColsBound <= 0) {
+    return NULL;
+  }
+  char*    pVals = POINTER_SHIFT(p, payloadValuesOffset(p));
+  SMemRow* memRow = (SMemRow)pBuilder->pDataBlock;
+  memRowSetType(memRow, memRowType);
+
+  // ----------------- Raw payload structure for row:
+  /* |<------------ Head ------------->|<----------- body of column data tuple ------------------->|
+   * |                                 |<----------------- flen ------------->|<--- value part --->|
+   * |SMemRowType| dataTLen |  nCols   |  colId  | colType | offset   |  ...  | value |...|...|... |
+   * +-----------+----------+----------+--------------------------------------|--------------------|
+   * | uint8_t   | uint32_t | uint16_t | int16_t | uint8_t | uint16_t |  ...  |.......|...|...|... |
+   * +-----------+----------+----------+--------------------------------------+--------------------|
+   *  1. offset in column data tuple starts from the value part in case of uint16_t overflow.
+   *  2. dataTLen: total length including the header and body.
+   */
+
+  if (memRowType == SMEM_ROW_DATA) {
+    SDataRow trow = (SDataRow)memRowDataBody(memRow);
+    dataRowSetLen(trow, (TDRowLenT)(TD_DATA_ROW_HEAD_SIZE + pBuilder->flen));
+    dataRowSetVersion(trow, pBuilder->sversion);
+
+    p = (char*)payloadBody(pBuilder->buf);
+    uint16_t i = 0, j = 0;
+    while (j < nCols) {
+      if (i >= nColsBound) {
+        break;
+      }
+      int16_t colId = payloadColId(p);
+      if (colId == pSchema[j].colId) {
+        // ASSERT(payloadColType(p) == pSchema[j].type);
+        tdAppendColVal(trow, POINTER_SHIFT(pVals, payloadColOffset(p)), pSchema[j].type, toffset);
+        toffset += TYPE_BYTES[pSchema[j].type];
+        p = payloadNextCol(p);
+        ++i;
+        ++j;
+      } else if (colId < pSchema[j].colId) {
+        p = payloadNextCol(p);
+        ++i;
+      } else {
+        tdAppendColVal(trow, getNullValue(pSchema[j].type), pSchema[j].type, toffset);
+        toffset += TYPE_BYTES[pSchema[j].type];
+        ++j;
+      }
+    }
+
+    while (j < nCols) {
+      tdAppendColVal(trow, getNullValue(pSchema[j].type), pSchema[j].type, toffset);
+      toffset += TYPE_BYTES[pSchema[j].type];
+      ++j;
+    }
+    
+    #if 0 // no need anymore
+    while (i < nColsBound) {
+      p = payloadNextCol(p);
+      ++i;
+    }
+    #endif
+
+  } else if (memRowType == SMEM_ROW_KV) {
+    SKVRow   kvRow = (SKVRow)memRowKvBody(memRow);
+    kvRowSetLen(kvRow, (TDRowLenT)(TD_KV_ROW_HEAD_SIZE + sizeof(SColIdx) * nColsBound));
+    kvRowSetNCols(kvRow, nColsBound);
+    memRowSetKvVersion(memRow, pBuilder->sversion);
+
+    p = (char*)payloadBody(pBuilder->buf);
+    int i = 0;
+    while (i < nColsBound) {
+      int16_t colId = payloadColId(p);
+      uint8_t colType = payloadColType(p);
+      tdAppendKvColVal(kvRow, POINTER_SHIFT(pVals,payloadColOffset(p)), colId, colType, toffset);
+      toffset += sizeof(SColIdx);
+      p = payloadNextCol(p);
+      ++i;
+    }
+
+  } else {
+    ASSERT(0);
+  }
+  int32_t rowTLen = memRowTLen(memRow);
+  pBuilder->pDataBlock = (char*)pBuilder->pDataBlock + rowTLen;  // next row
+  pBuilder->pSubmitBlk->dataLen += rowTLen;
+
+  return memRow;
+}
+
+// Erase the empty space reserved for binary data
+static int trimDataBlock(void* pDataBlock, STableDataBlocks* pTableDataBlock, SInsertStatementParam* insertParam,
+                         SBlockKeyTuple* blkKeyTuple) {
   // TODO: optimize this function, handle the case while binary is not presented
   STableMeta*     pTableMeta = pTableDataBlock->pTableMeta;
   STableComInfo   tinfo = tscGetTableInfo(pTableMeta);
@@ -1775,7 +1879,7 @@ static int trimDataBlock(void* pDataBlock, STableDataBlocks* pTableDataBlock, bo
   int32_t flen = 0;  // original total length of row
 
   // schema needs to be included into the submit data block
-  if (includeSchema) {
+  if (insertParam->schemaAttached) {
     int32_t numOfCols = tscGetNumOfColumns(pTableDataBlock->pTableMeta);
     for(int32_t j = 0; j < numOfCols; ++j) {
       STColumn* pCol = (STColumn*) pDataBlock;
@@ -1802,20 +1906,38 @@ static int trimDataBlock(void* pDataBlock, STableDataBlocks* pTableDataBlock, bo
   pBlock->dataLen = 0;
   int32_t numOfRows = htons(pBlock->numOfRows);
 
+  if (IS_RAW_PAYLOAD(insertParam->payloadType)) {
     for (int32_t i = 0; i < numOfRows; ++i) {
-    SDataRow trow = (SDataRow) pDataBlock;
+      SMemRow memRow = (SMemRow)pDataBlock;
+      memRowSetType(memRow, SMEM_ROW_DATA);
+      SDataRow trow = memRowDataBody(memRow);
       dataRowSetLen(trow, (uint16_t)(TD_DATA_ROW_HEAD_SIZE + flen));
       dataRowSetVersion(trow, pTableMeta->sversion);
 
       int toffset = 0;
       for (int32_t j = 0; j < tinfo.numOfColumns; j++) {
-      tdAppendColVal(trow, p, pSchema[j].type, pSchema[j].bytes, toffset);
+        tdAppendColVal(trow, p, pSchema[j].type, toffset);
         toffset += TYPE_BYTES[pSchema[j].type];
         p += pSchema[j].bytes;
       }
 
-    pDataBlock = (char*)pDataBlock + dataRowLen(trow);
-    pBlock->dataLen += dataRowLen(trow);
+      pDataBlock = (char*)pDataBlock + memRowTLen(memRow);
+      pBlock->dataLen += memRowTLen(memRow);
+    }
+  } else {
+    SMemRowBuilder rowBuilder;
+    rowBuilder.pSchema = pSchema;
+    rowBuilder.sversion = pTableMeta->sversion;
+    rowBuilder.flen = flen;
+    rowBuilder.nCols = tinfo.numOfColumns;
+    rowBuilder.pDataBlock = pDataBlock;
+    rowBuilder.pSubmitBlk = pBlock;
+    rowBuilder.buf = p;
+
+    for (int32_t i = 0; i < numOfRows; ++i) {
+      rowBuilder.buf = (blkKeyTuple + i)->payloadAddr;
+      tdGenMemRowFromBuilder(&rowBuilder);
+    }
   }
 
   int32_t len = pBlock->dataLen + pBlock->schemaLen;
@@ -1826,7 +1948,7 @@ static int trimDataBlock(void* pDataBlock, STableDataBlocks* pTableDataBlock, bo
 }
 
 static int32_t getRowExpandSize(STableMeta* pTableMeta) {
-  int32_t result = TD_DATA_ROW_HEAD_SIZE;
+  int32_t  result = TD_MEM_ROW_DATA_HEAD_SIZE;
   int32_t columns = tscGetNumOfColumns(pTableMeta);
   SSchema* pSchema = tscGetTableSchema(pTableMeta);
   for(int32_t i = 0; i < columns; i++) {
@@ -1840,16 +1962,14 @@ static int32_t getRowExpandSize(STableMeta* pTableMeta) {
 static void extractTableNameList(SInsertStatementParam *pInsertParam, bool freeBlockMap) {
   pInsertParam->numOfTables = (int32_t) taosHashGetSize(pInsertParam->pTableBlockHashList);
   if (pInsertParam->pTableNameList == NULL) {
-    pInsertParam->pTableNameList = calloc(pInsertParam->numOfTables, POINTER_BYTES);
-  } else {
-    memset(pInsertParam->pTableNameList, 0, pInsertParam->numOfTables * POINTER_BYTES);
+    pInsertParam->pTableNameList = malloc(pInsertParam->numOfTables * POINTER_BYTES);
   }
 
   STableDataBlocks **p1 = taosHashIterate(pInsertParam->pTableBlockHashList, NULL);
   int32_t i = 0;
   while(p1) {
     STableDataBlocks* pBlocks = *p1;
-    tfree(pInsertParam->pTableNameList[i]);
+    //tfree(pInsertParam->pTableNameList[i]);
 
     pInsertParam->pTableNameList[i++] = tNameDup(&pBlocks->tableName);
     p1 = taosHashIterate(pInsertParam->pTableBlockHashList, p1);
@@ -1862,13 +1982,17 @@ static void extractTableNameList(SInsertStatementParam *pInsertParam, bool freeB
 
 int32_t tscMergeTableDataBlocks(SInsertStatementParam *pInsertParam, bool freeBlockMap) {
   const int INSERT_HEAD_SIZE = sizeof(SMsgDesc) + sizeof(SSubmitMsg);
-
+  int       code = 0;
+  bool      isRawPayload = IS_RAW_PAYLOAD(pInsertParam->payloadType);
   void*     pVnodeDataBlockHashList = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, false);
   SArray*   pVnodeDataBlockList = taosArrayInit(8, POINTER_BYTES);
 
   STableDataBlocks** p = taosHashIterate(pInsertParam->pTableBlockHashList, NULL);
 
   STableDataBlocks* pOneTableBlock = *p;
+
+  SBlockKeyInfo blkKeyInfo = {0};  // share by pOneTableBlock
+
   while(pOneTableBlock) {
     SSubmitBlk* pBlocks = (SSubmitBlk*) pOneTableBlock->pData;
     if (pBlocks->numOfRows > 0) {
@@ -1882,36 +2006,53 @@ int32_t tscMergeTableDataBlocks(SInsertStatementParam *pInsertParam, bool freeBl
         tscError("0x%"PRIx64" failed to prepare the data block buffer for merging table data, code:%d", pInsertParam->objectId, ret);
         taosHashCleanup(pVnodeDataBlockHashList);
         tscDestroyBlockArrayList(pVnodeDataBlockList);
+        tfree(blkKeyInfo.pKeyTuple);
         return ret;
       }
 
       int64_t destSize = dataBuf->size + pOneTableBlock->size + pBlocks->numOfRows * expandSize + sizeof(STColumn) * tscGetNumOfColumns(pOneTableBlock->pTableMeta);
 
       if (dataBuf->nAllocSize < destSize) {
-        while (dataBuf->nAllocSize < destSize) {
-          dataBuf->nAllocSize = (uint32_t)(dataBuf->nAllocSize * 1.5);
-        }
+        dataBuf->nAllocSize = (uint32_t)(destSize * 1.5);
 
         char* tmp = realloc(dataBuf->pData, dataBuf->nAllocSize);
         if (tmp != NULL) {
           dataBuf->pData = tmp;
-          memset(dataBuf->pData + dataBuf->size, 0, dataBuf->nAllocSize - dataBuf->size);
+          //memset(dataBuf->pData + dataBuf->size, 0, dataBuf->nAllocSize - dataBuf->size);
         } else {  // failed to allocate memory, free already allocated memory and return error code
           tscError("0x%"PRIx64" failed to allocate memory for merging submit block, size:%d", pInsertParam->objectId, dataBuf->nAllocSize);
 
           taosHashCleanup(pVnodeDataBlockHashList);
           tscDestroyBlockArrayList(pVnodeDataBlockList);
           tfree(dataBuf->pData);
+          tfree(blkKeyInfo.pKeyTuple);
 
           return TSDB_CODE_TSC_OUT_OF_MEMORY;
         }
       }
 
-      tscSortRemoveDataBlockDupRows(pOneTableBlock);
-      char* ekey = (char*)pBlocks->data + pOneTableBlock->rowSize*(pBlocks->numOfRows-1);
+      if (isRawPayload) {
+        tscSortRemoveDataBlockDupRowsRaw(pOneTableBlock);
+        char* ekey = (char*)pBlocks->data + pOneTableBlock->rowSize * (pBlocks->numOfRows - 1);
 
-      tscDebug("0x%"PRIx64" name:%s, tid:%d rows:%d sversion:%d skey:%" PRId64 ", ekey:%" PRId64, pInsertParam->objectId, tNameGetTableName(&pOneTableBlock->tableName),
-          pBlocks->tid, pBlocks->numOfRows, pBlocks->sversion, GET_INT64_VAL(pBlocks->data), GET_INT64_VAL(ekey));
+        tscDebug("0x%" PRIx64 " name:%s, tid:%d rows:%d sversion:%d skey:%" PRId64 ", ekey:%" PRId64,
+                 pInsertParam->objectId, tNameGetTableName(&pOneTableBlock->tableName), pBlocks->tid,
+                 pBlocks->numOfRows, pBlocks->sversion, GET_INT64_VAL(pBlocks->data), GET_INT64_VAL(ekey));
+      } else {
+        if ((code = tscSortRemoveDataBlockDupRows(pOneTableBlock, &blkKeyInfo)) != 0) {
+          taosHashCleanup(pVnodeDataBlockHashList);
+          tscDestroyBlockArrayList(pVnodeDataBlockList);
+          tfree(dataBuf->pData);
+          tfree(blkKeyInfo.pKeyTuple);
+          return code;
+        }
+        ASSERT(blkKeyInfo.pKeyTuple != NULL && pBlocks->numOfRows > 0);
+
+        SBlockKeyTuple* pLastKeyTuple = blkKeyInfo.pKeyTuple + pBlocks->numOfRows - 1;
+        tscDebug("0x%" PRIx64 " name:%s, tid:%d rows:%d sversion:%d skey:%" PRId64 ", ekey:%" PRId64,
+                 pInsertParam->objectId, tNameGetTableName(&pOneTableBlock->tableName), pBlocks->tid,
+                 pBlocks->numOfRows, pBlocks->sversion, blkKeyInfo.pKeyTuple->skey, pLastKeyTuple->skey);
+      }
       
       int32_t len = pBlocks->numOfRows * (pOneTableBlock->rowSize + expandSize) + sizeof(STColumn) * tscGetNumOfColumns(pOneTableBlock->pTableMeta);
 
@@ -1922,7 +2063,7 @@ int32_t tscMergeTableDataBlocks(SInsertStatementParam *pInsertParam, bool freeBl
       pBlocks->schemaLen = 0;
 
       // erase the empty space reserved for binary data
-      int32_t finalLen = trimDataBlock(dataBuf->pData + dataBuf->size, pOneTableBlock, pInsertParam->schemaAttached);
+      int32_t finalLen = trimDataBlock(dataBuf->pData + dataBuf->size, pOneTableBlock, pInsertParam, blkKeyInfo.pKeyTuple);
       assert(finalLen <= len);
 
       dataBuf->size += (finalLen + sizeof(SSubmitBlk));
@@ -1950,6 +2091,7 @@ int32_t tscMergeTableDataBlocks(SInsertStatementParam *pInsertParam, bool freeBl
   // free the table data blocks;
   pInsertParam->pDataBlocks = pVnodeDataBlockList;
   taosHashCleanup(pVnodeDataBlockHashList);
+  tfree(blkKeyInfo.pKeyTuple);
 
   return TSDB_CODE_SUCCESS;
 }
@@ -2986,6 +3128,7 @@ void tscInitQueryInfo(SQueryInfo* pQueryInfo) {
 int32_t tscAddQueryInfo(SSqlCmd* pCmd) {
   assert(pCmd != NULL);
   SQueryInfo* pQueryInfo = calloc(1, sizeof(SQueryInfo));
+
   if (pQueryInfo == NULL) {
     return TSDB_CODE_TSC_OUT_OF_MEMORY;
   }
@@ -3418,6 +3561,7 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, __async_cb_func_t
   pNew->pTscObj   = pSql->pTscObj;
   pNew->signature = pNew;
   pNew->sqlstr    = strdup(pSql->sqlstr);
+  tsem_init(&pNew->rspSem, 0, 0);
 
   SSqlCmd* pnCmd  = &pNew->cmd;
   memcpy(pnCmd, pCmd, sizeof(SSqlCmd));
@@ -3778,7 +3922,7 @@ bool tscIsUpdateQuery(SSqlObj* pSql) {
   }
 
   SSqlCmd* pCmd = &pSql->cmd;
-  return ((pCmd->command >= TSDB_SQL_INSERT && pCmd->command <= TSDB_SQL_DROP_DNODE) || TSDB_SQL_USE_DB == pCmd->command);
+  return ((pCmd->command >= TSDB_SQL_INSERT && pCmd->command <= TSDB_SQL_DROP_DNODE) || TSDB_SQL_RESET_CACHE == pCmd->command || TSDB_SQL_USE_DB == pCmd->command);
 }
 
 char* tscGetSqlStr(SSqlObj* pSql) {
@@ -3968,17 +4112,21 @@ void tscTryQueryNextClause(SSqlObj* pSql, __async_cb_func_t fp) {
   //backup the total number of result first
   int64_t num = pRes->numOfTotal + pRes->numOfClauseTotal;
 
-
   // DON't free final since it may be recoreded and used later in APP
   TAOS_FIELD* finalBk = pRes->final;
   pRes->final = NULL;
   tscFreeSqlResult(pSql);
-  pRes->final = finalBk;
 
+  pRes->final = finalBk;
   pRes->numOfTotal = num;
 
+  for(int32_t i = 0; i < pSql->subState.numOfSub; ++i) {
+    taos_free_result(pSql->pSubs[i]);
+  }
+
   tfree(pSql->pSubs);
   pSql->subState.numOfSub = 0;
+
   pSql->fp = fp;
 
   tscDebug("0x%"PRIx64" try data in the next subclause", pSql->self);
@@ -4239,7 +4387,7 @@ STableMeta* tscTableMetaDup(STableMeta* pTableMeta) {
   assert(pTableMeta != NULL);
   size_t size = tscGetTableMetaSize(pTableMeta);
 
-  STableMeta* p = calloc(1, size);
+  STableMeta* p = malloc(size);
   memcpy(p, pTableMeta, size);
   return p;
 }
@@ -4618,6 +4766,11 @@ int32_t nameComparFn(const void* n1, const void* n2) {
   }
 }
 
+static void freeContent(void* p) {
+  char* ptr = *(char**)p;
+  tfree(ptr);
+}
+
 int tscTransferTableNameList(SSqlObj *pSql, const char *pNameList, int32_t length, SArray* pNameArray) {
   SSqlCmd *pCmd = &pSql->cmd;
 
@@ -4665,32 +4818,7 @@ int tscTransferTableNameList(SSqlObj *pSql, const char *pNameList, int32_t lengt
   }
 
   taosArraySort(pNameArray, nameComparFn);
-
-  int32_t pos = 0;
-  for(int32_t i = 1; i < len; ++i) {
-    char** p1 = taosArrayGet(pNameArray, pos);
-    char** p2 = taosArrayGet(pNameArray, i);
-
-    if (strcmp(*p1, *p2) == 0) {
-      // do nothing
-    } else {
-      if (pos + 1 != i) {
-        char* p = taosArrayGetP(pNameArray, pos + 1);
-        tfree(p);
-        taosArraySet(pNameArray, pos + 1, p2);
-        pos += 1;
-      } else {
-        pos += 1;
-      }
-    }
-  }
-
-  for(int32_t i = pos + 1; i < pNameArray->size; ++i) {
-    char* p = taosArrayGetP(pNameArray, i);
-    tfree(p);
-  }
-
-  pNameArray->size = pos + 1;
+  taosArrayRemoveDuplicate(pNameArray, nameComparFn, freeContent);
   return TSDB_CODE_SUCCESS;
 }
 

src/common/inc/tdataformat.h

@@ -55,8 +55,8 @@ extern "C" {
 typedef struct {
   int8_t   type;    // Column type
   int16_t  colId;   // column ID
-  int16_t bytes;   // column bytes
-  int16_t offset;  // point offset in SDataRow after the header part
+  uint16_t bytes;   // column bytes
+  uint16_t offset;  // point offset in SDataRow after the header part.
 } STColumn;
 
 #define colType(col) ((col)->type)
@@ -167,10 +167,11 @@ static FORCE_INLINE int tkeyComparFn(const void *tkey1, const void *tkey2) {
     return 0;
   }
 }
+
 // ----------------- Data row structure
 
 /* A data row, the format is like below:
- * |<--------------------+--------------------------- len ---------------------------------->|
+ * |<------------------------------------------------ len ---------------------------------->|
  * |<--     Head      -->|<---------   flen -------------->|                                 |
  * +---------------------+---------------------------------+---------------------------------+
  * | uint16_t |  int16_t |                                 |                                 |
@@ -184,8 +185,8 @@ typedef void *SDataRow;
 
 #define TD_DATA_ROW_HEAD_SIZE (sizeof(uint16_t) + sizeof(int16_t))
 
-#define dataRowLen(r) (*(uint16_t *)(r))
-#define dataRowVersion(r) *(int16_t *)POINTER_SHIFT(r, sizeof(int16_t))
+#define dataRowLen(r) (*(TDRowLenT *)(r))  // 0~65535
+#define dataRowVersion(r) (*(int16_t *)POINTER_SHIFT(r, sizeof(int16_t)))
 #define dataRowTuple(r) POINTER_SHIFT(r, TD_DATA_ROW_HEAD_SIZE)
 #define dataRowTKey(r) (*(TKEY *)(dataRowTuple(r)))
 #define dataRowKey(r) tdGetKey(dataRowTKey(r))
@@ -201,20 +202,19 @@ void     tdInitDataRow(SDataRow row, STSchema *pSchema);
 SDataRow tdDataRowDup(SDataRow row);
 
 // offset here not include dataRow header length
-static FORCE_INLINE int tdAppendColVal(SDataRow row, void *value, int8_t type, int32_t bytes, int32_t offset) {
+static FORCE_INLINE int tdAppendColVal(SDataRow row, const void *value, int8_t type, int32_t offset) {
   ASSERT(value != NULL);
   int32_t toffset = offset + TD_DATA_ROW_HEAD_SIZE;
-  char *  ptr = (char *)POINTER_SHIFT(row, dataRowLen(row));
 
   if (IS_VAR_DATA_TYPE(type)) {
     *(VarDataOffsetT *)POINTER_SHIFT(row, toffset) = dataRowLen(row);
-    memcpy(ptr, value, varDataTLen(value));
+    memcpy(POINTER_SHIFT(row, dataRowLen(row)), value, varDataTLen(value));
     dataRowLen(row) += varDataTLen(value);
   } else {
     if (offset == 0) {
       ASSERT(type == TSDB_DATA_TYPE_TIMESTAMP);
       TKEY tvalue = tdGetTKEY(*(TSKEY *)value);
-      memcpy(POINTER_SHIFT(row, toffset), (void *)(&tvalue), TYPE_BYTES[type]);
+      memcpy(POINTER_SHIFT(row, toffset), (const void *)(&tvalue), TYPE_BYTES[type]);
     } else {
       memcpy(POINTER_SHIFT(row, toffset), value, TYPE_BYTES[type]);
     }
@@ -245,17 +245,21 @@ typedef struct SDataCol {
   TSKEY           ts;         // only used in last NULL column
 } SDataCol;
 
+#define isAllRowsNull(pCol) ((pCol)->len == 0)
 static FORCE_INLINE void dataColReset(SDataCol *pDataCol) { pDataCol->len = 0; }
 
 void dataColInit(SDataCol *pDataCol, STColumn *pCol, void **pBuf, int maxPoints);
-void dataColAppendVal(SDataCol *pCol, void *value, int numOfRows, int maxPoints);
+void dataColAppendVal(SDataCol *pCol, const void *value, int numOfRows, int maxPoints);
 void dataColSetOffset(SDataCol *pCol, int nEle);
 
 bool isNEleNull(SDataCol *pCol, int nEle);
 void dataColSetNEleNull(SDataCol *pCol, int nEle, int maxPoints);
 
 // Get the data pointer from a column-wised data
-static FORCE_INLINE void *tdGetColDataOfRow(SDataCol *pCol, int row) {
+static FORCE_INLINE const void *tdGetColDataOfRow(SDataCol *pCol, int row) {
+  if (isAllRowsNull(pCol)) {
+    return getNullValue(pCol->type);
+  }
   if (IS_VAR_DATA_TYPE(pCol->type)) {
     return POINTER_SHIFT(pCol->pData, pCol->dataOff[row]);
   } else {
@@ -287,7 +291,7 @@ typedef struct {
 } SDataCols;
 
 #define keyCol(pCols) (&((pCols)->cols[0]))  // Key column
-#define dataColsTKeyAt(pCols, idx) ((TKEY *)(keyCol(pCols)->pData))[(idx)]
+#define dataColsTKeyAt(pCols, idx) ((TKEY *)(keyCol(pCols)->pData))[(idx)]  // the idx row of column-wised data
 #define dataColsKeyAt(pCols, idx) tdGetKey(dataColsTKeyAt(pCols, idx))
 static FORCE_INLINE TKEY dataColsTKeyFirst(SDataCols *pCols) {
   if (pCols->numOfRows) {
@@ -331,13 +335,13 @@ void       tdResetDataCols(SDataCols *pCols);
 int        tdInitDataCols(SDataCols *pCols, STSchema *pSchema);
 SDataCols *tdDupDataCols(SDataCols *pCols, bool keepData);
 SDataCols *tdFreeDataCols(SDataCols *pCols);
-void       tdAppendDataRowToDataCol(SDataRow row, STSchema *pSchema, SDataCols *pCols);
 int        tdMergeDataCols(SDataCols *target, SDataCols *source, int rowsToMerge, int *pOffset);
 
 // ----------------- K-V data row structure
-/*
+/* |<-------------------------------------- len -------------------------------------------->|
+ * |<----- header  ----->|<--------------------------- body -------------------------------->|
  * +----------+----------+---------------------------------+---------------------------------+
- * |  int16_t |  int16_t |                                 |                                 |
+ * | uint16_t |  int16_t |                                 |                                 |
  * +----------+----------+---------------------------------+---------------------------------+
  * |    len   |   ncols  |           cols index            |             data part           |
  * +----------+----------+---------------------------------+---------------------------------+
@@ -346,13 +350,13 @@ typedef void *SKVRow;
 
 typedef struct {
   int16_t  colId;
-  int16_t offset;
+  uint16_t offset;
 } SColIdx;
 
-#define TD_KV_ROW_HEAD_SIZE (2 * sizeof(int16_t))
+#define TD_KV_ROW_HEAD_SIZE (sizeof(uint16_t) + sizeof(int16_t))
 
-#define kvRowLen(r) (*(int16_t *)(r))
-#define kvRowNCols(r) (*(int16_t *)POINTER_SHIFT(r, sizeof(int16_t)))
+#define kvRowLen(r) (*(TDRowLenT *)(r))
+#define kvRowNCols(r) (*(int16_t *)POINTER_SHIFT(r, sizeof(uint16_t)))
 #define kvRowSetLen(r, len) kvRowLen(r) = (len)
 #define kvRowSetNCols(r, n) kvRowNCols(r) = (n)
 #define kvRowColIdx(r) (SColIdx *)POINTER_SHIFT(r, TD_KV_ROW_HEAD_SIZE)
@@ -362,6 +366,9 @@ typedef struct {
 #define kvRowColIdxAt(r, i) (kvRowColIdx(r) + (i))
 #define kvRowFree(r) tfree(r)
 #define kvRowEnd(r) POINTER_SHIFT(r, kvRowLen(r))
+#define kvRowTKey(r) (*(TKEY *)(kvRowValues(r)))
+#define kvRowKey(r) tdGetKey(kvRowTKey(r))
+#define kvRowDeleted(r) TKEY_IS_DELETED(kvRowTKey(r))
 
 SKVRow tdKVRowDup(SKVRow row);
 int    tdSetKVRowDataOfCol(SKVRow *orow, int16_t colId, int8_t type, void *value);
@@ -385,13 +392,44 @@ static FORCE_INLINE void *tdGetKVRowValOfCol(SKVRow row, int16_t colId) {
   return kvRowColVal(row, (SColIdx *)ret);
 }
 
+static FORCE_INLINE void *tdGetKVRowIdxOfCol(SKVRow row, int16_t colId) {
+  return taosbsearch(&colId, kvRowColIdx(row), kvRowNCols(row), sizeof(SColIdx), comparTagId, TD_EQ);
+}
+
+// offset here not include kvRow header length
+static FORCE_INLINE int tdAppendKvColVal(SKVRow row, const void *value, int16_t colId, int8_t type, int32_t offset) {
+  ASSERT(value != NULL);
+  int32_t  toffset = offset + TD_KV_ROW_HEAD_SIZE;
+  SColIdx *pColIdx = (SColIdx *)POINTER_SHIFT(row, toffset);
+  char *   ptr = (char *)POINTER_SHIFT(row, kvRowLen(row));
+
+  pColIdx->colId = colId;
+  pColIdx->offset = kvRowLen(row);  // offset of pColIdx including the TD_KV_ROW_HEAD_SIZE
+
+  if (IS_VAR_DATA_TYPE(type)) {
+    memcpy(ptr, value, varDataTLen(value));
+    kvRowLen(row) += varDataTLen(value);
+  } else {
+    if (offset == 0) {
+      ASSERT(type == TSDB_DATA_TYPE_TIMESTAMP);
+      TKEY tvalue = tdGetTKEY(*(TSKEY *)value);
+      memcpy(ptr, (void *)(&tvalue), TYPE_BYTES[type]);
+    } else {
+      memcpy(ptr, value, TYPE_BYTES[type]);
+    }
+    kvRowLen(row) += TYPE_BYTES[type];
+  }
+
+  return 0;
+}
+
 // ----------------- K-V data row builder
 typedef struct {
   int16_t  tCols;
   int16_t  nCols;
   SColIdx *pColIdx;
-  int16_t  alloc;
-  int16_t  size;
+  uint16_t alloc;
+  uint16_t size;
   void *   buf;
 } SKVRowBuilder;
 
@@ -427,6 +465,144 @@ static FORCE_INLINE int tdAddColToKVRow(SKVRowBuilder *pBuilder, int16_t colId,
   return 0;
 }
 
+// ----------------- SMemRow appended with sequential data row structure
+/*
+ * |---------|------------------------------------------------- len ---------------------------------->|
+ * |<--------     Head      ------>|<---------   flen -------------->|                                 |
+ * |---------+---------------------+---------------------------------+---------------------------------+
+ * | uint8_t | uint16_t |  int16_t |                                 |                                 |
+ * |---------+----------+----------+---------------------------------+---------------------------------+
+ * |  flag   |   len    | sversion |           First part            |             Second part         |
+ * +---------+----------+----------+---------------------------------+---------------------------------+
+ *
+ * NOTE: timestamp in this row structure is TKEY instead of TSKEY
+ */
+
+// ----------------- SMemRow appended with extended K-V data row structure
+/* |--------------------|------------------------------------------------  len ---------------------------------->|
+ * |<-------------     Head      ------------>|<---------   flen -------------->|                                 |
+ * |--------------------+----------+--------------------------------------------+---------------------------------+
+ * | uint8_t | int16_t  | uint16_t |  int16_t |                                 |                                 |
+ * |---------+----------+----------+----------+---------------------------------+---------------------------------+
+ * |   flag  | sversion |   len    |   ncols  |           cols index            |             data part           |
+ * |---------+----------+----------+----------+---------------------------------+---------------------------------+
+ */
+
+typedef void *SMemRow;
+
+#define TD_MEM_ROW_TYPE_SIZE sizeof(uint8_t)
+#define TD_MEM_ROW_KV_VER_SIZE sizeof(int16_t)
+#define TD_MEM_ROW_KV_TYPE_VER_SIZE (TD_MEM_ROW_TYPE_SIZE + TD_MEM_ROW_KV_VER_SIZE)
+#define TD_MEM_ROW_DATA_HEAD_SIZE (TD_MEM_ROW_TYPE_SIZE + TD_DATA_ROW_HEAD_SIZE)
+// #define TD_MEM_ROW_KV_HEAD_SIZE (TD_MEM_ROW_TYPE_SIZE + TD_MEM_ROW_KV_VER_SIZE + TD_KV_ROW_HEAD_SIZE)
+
+#define SMEM_ROW_DATA 0U  // SDataRow
+#define SMEM_ROW_KV 1U    // SKVRow
+
+#define memRowType(r) (*(uint8_t *)(r))
+#define isDataRow(r) (SMEM_ROW_DATA == memRowType(r))
+#define isKvRow(r) (SMEM_ROW_KV == memRowType(r))
+
+#define memRowDataBody(r) POINTER_SHIFT(r, TD_MEM_ROW_TYPE_SIZE)  // section after flag
+#define memRowKvBody(r) \
+  POINTER_SHIFT(r, TD_MEM_ROW_KV_TYPE_VER_SIZE)  // section after flag + sversion as to reuse SKVRow
+
+#define memRowDataLen(r) (*(TDRowLenT *)memRowDataBody(r))  //  0~65535
+#define memRowKvLen(r) (*(TDRowLenT *)memRowKvBody(r))      //  0~65535
+
+#define memRowDataTLen(r) \
+  ((TDRowTLenT)(memRowDataLen(r) + TD_MEM_ROW_TYPE_SIZE))  // using uint32_t/int32_t to store the TLen
+
+#define memRowKvTLen(r) ((TDRowTLenT)(memRowKvLen(r) + TD_MEM_ROW_KV_TYPE_VER_SIZE))
+
+#define memRowLen(r) (isDataRow(r) ? memRowDataLen(r) : memRowKvLen(r))
+#define memRowTLen(r) (isDataRow(r) ? memRowDataTLen(r) : memRowKvTLen(r))  // using uint32_t/int32_t to store the TLen
+
+#define memRowDataVersion(r) dataRowVersion(memRowDataBody(r))
+#define memRowKvVersion(r) (*(int16_t *)POINTER_SHIFT(r, TD_MEM_ROW_TYPE_SIZE))
+#define memRowVersion(r) (isDataRow(r) ? memRowDataVersion(r) : memRowKvVersion(r))  // schema version
+#define memRowSetKvVersion(r, v) (memRowKvVersion(r) = (v))
+#define memRowTuple(r) (isDataRow(r) ? dataRowTuple(memRowDataBody(r)) : kvRowValues(memRowKvBody(r)))
+
+#define memRowTKey(r) (isDataRow(r) ? dataRowTKey(memRowDataBody(r)) : kvRowTKey(memRowKvBody(r)))
+#define memRowKey(r) (isDataRow(r) ? dataRowKey(memRowDataBody(r)) : kvRowKey(memRowKvBody(r)))
+#define memRowSetTKey(r, k)                 \
+  do {                                      \
+    if (isDataRow(r)) {                     \
+      dataRowTKey(memRowDataBody(r)) = (k); \
+    } else {                                \
+      kvRowTKey(memRowKvBody(r)) = (k);     \
+    }                                       \
+  } while (0)
+
+#define memRowSetType(r, t) (memRowType(r) = (t))
+#define memRowSetLen(r, l) (isDataRow(r) ? memRowDataLen(r) = (l) : memRowKvLen(r) = (l))
+#define memRowSetVersion(r, v) (isDataRow(r) ? dataRowSetVersion(memRowDataBody(r), v) : memRowKvSetVersion(r, v))
+#define memRowCpy(dst, r) memcpy((dst), (r), memRowTLen(r))
+#define memRowMaxBytesFromSchema(s) (schemaTLen(s) + TD_MEM_ROW_DATA_HEAD_SIZE)
+#define memRowDeleted(r) TKEY_IS_DELETED(memRowTKey(r))
+
+SMemRow tdMemRowDup(SMemRow row);
+void    tdAppendMemRowToDataCol(SMemRow row, STSchema *pSchema, SDataCols *pCols);
+// NOTE: offset here including the header size
+static FORCE_INLINE void *tdGetKvRowDataOfCol(void *row, int32_t offset) { return POINTER_SHIFT(row, offset); }
+// NOTE: offset here including the header size
+static FORCE_INLINE void *tdGetMemRowDataOfCol(void *row, int8_t type, int32_t offset) {
+  if (isDataRow(row)) {
+    return tdGetRowDataOfCol(row, type, offset);
+  } else if (isKvRow(row)) {
+    return tdGetKvRowDataOfCol(row, offset);
+  } else {
+    ASSERT(0);
+  }
+  return NULL;
+}
+
+// ----------------- Raw payload structure for row:
+/* |<------------ Head ------------->|<----------- body of column data tuple ------------------->|
+ * |                                 |<----------------- flen ------------->|<--- value part --->|
+ * |SMemRowType| dataTLen |  nCols   |  colId  | colType | offset   |  ...  | value |...|...|... |
+ * +-----------+----------+----------+--------------------------------------|--------------------|
+ * | uint8_t   | uint32_t | uint16_t | int16_t | uint8_t | uint16_t |  ...  |.......|...|...|... |
+ * +-----------+----------+----------+--------------------------------------+--------------------|
+ *  1. offset in column data tuple starts from the value part in case of uint16_t overflow.
+ *  2. dataTLen: total length including the header and body.
+ */
+
+#define PAYLOAD_NCOLS_LEN sizeof(uint16_t)
+#define PAYLOAD_NCOLS_OFFSET (sizeof(uint8_t) + sizeof(TDRowTLenT))
+#define PAYLOAD_HEADER_LEN (PAYLOAD_NCOLS_OFFSET + PAYLOAD_NCOLS_LEN)
+#define PAYLOAD_ID_LEN sizeof(int16_t)
+#define PAYLOAD_ID_TYPE_LEN (sizeof(int16_t) + sizeof(uint8_t))
+#define PAYLOAD_COL_HEAD_LEN (PAYLOAD_ID_TYPE_LEN + sizeof(uint16_t))
+#define PAYLOAD_PRIMARY_COL_LEN (PAYLOAD_ID_TYPE_LEN + sizeof(TSKEY))
+
+#define payloadBody(r) POINTER_SHIFT(r, PAYLOAD_HEADER_LEN)
+#define payloadType(r) (*(uint8_t *)(r))
+#define payloadSetType(r, t) (payloadType(r) = (t))
+#define payloadTLen(r) (*(TDRowTLenT *)POINTER_SHIFT(r, TD_MEM_ROW_TYPE_SIZE))  // including total header
+#define payloadSetTLen(r, l) (payloadTLen(r) = (l))
+#define payloadNCols(r) (*(TDRowLenT *)POINTER_SHIFT(r, PAYLOAD_NCOLS_OFFSET))
+#define payloadSetNCols(r, n) (payloadNCols(r) = (n))
+#define payloadValuesOffset(r) \
+  (PAYLOAD_HEADER_LEN + payloadNCols(r) * PAYLOAD_COL_HEAD_LEN)    // avoid using the macro in loop
+#define payloadValues(r) POINTER_SHIFT(r, payloadValuesOffset(r))  // avoid using the macro in loop
+#define payloadColId(c) (*(int16_t *)(c))
+#define payloadColType(c) (*(uint8_t *)POINTER_SHIFT(c, PAYLOAD_ID_LEN))
+#define payloadColOffset(c) (*(uint16_t *)POINTER_SHIFT(c, PAYLOAD_ID_TYPE_LEN))
+#define payloadColValue(c) POINTER_SHIFT(c, payloadColOffset(c))
+
+#define payloadColSetId(c, i) (payloadColId(c) = (i))
+#define payloadColSetType(c, t) (payloadColType(c) = (t))
+#define payloadColSetOffset(c, o) (payloadColOffset(c) = (o))
+
+#define payloadTSKey(r) (*(TSKEY *)POINTER_SHIFT(r, payloadValuesOffset(r)))
+#define payloadTKey(r) (*(TKEY *)POINTER_SHIFT(r, payloadValuesOffset(r)))
+#define payloadKey(r) tdGetKey(payloadTKey(r))
+
+
+static FORCE_INLINE char *payloadNextCol(char *pCol) { return (char *)POINTER_SHIFT(pCol, PAYLOAD_COL_HEAD_LEN); }
+
 #ifdef __cplusplus
 }
 #endif

src/common/inc/tglobal.h

@@ -205,6 +205,16 @@ extern int32_t wDebugFlag;
 extern int32_t cqDebugFlag;
 extern int32_t debugFlag;
 
+#ifdef TD_TSZ
+// lossy 
+extern char lossyColumns[];
+extern double fPrecision;
+extern double dPrecision;
+extern uint32_t maxRange;
+extern uint32_t curRange;
+extern char Compressor[];
+#endif
+
 typedef struct {
   char dir[TSDB_FILENAME_LEN];
   int  level;

src/common/src/tdataformat.c

@@ -198,6 +198,14 @@ SDataRow tdDataRowDup(SDataRow row) {
   return trow;
 }
 
+SMemRow tdMemRowDup(SMemRow row) {
+  SMemRow trow = malloc(memRowTLen(row));
+  if (trow == NULL) return NULL;
+
+  memRowCpy(trow, row);
+  return trow;
+}
+
 void dataColInit(SDataCol *pDataCol, STColumn *pCol, void **pBuf, int maxPoints) {
   pDataCol->type = colType(pCol);
   pDataCol->colId = colColId(pCol);
@@ -217,11 +225,22 @@ void dataColInit(SDataCol *pDataCol, STColumn *pCol, void **pBuf, int maxPoints)
     *pBuf = POINTER_SHIFT(*pBuf, pDataCol->spaceSize);
   }
 }
-
 // value from timestamp should be TKEY here instead of TSKEY
-void dataColAppendVal(SDataCol *pCol, void *value, int numOfRows, int maxPoints) {
+void dataColAppendVal(SDataCol *pCol, const void *value, int numOfRows, int maxPoints) {
   ASSERT(pCol != NULL && value != NULL);
 
+  if (isAllRowsNull(pCol)) {
+    if (isNull(value, pCol->type)) {
+      // all null value yet, just return
+      return;
+    }
+
+    if (numOfRows > 0) {
+      // Find the first not null value, fill all previouse values as NULL
+      dataColSetNEleNull(pCol, numOfRows, maxPoints);
+    }
+  }
+
   if (IS_VAR_DATA_TYPE(pCol->type)) {
     // set offset
     pCol->dataOff[numOfRows] = pCol->len;
@@ -243,7 +262,7 @@ bool isNEleNull(SDataCol *pCol, int nEle) {
   return true;
 }
 
-void dataColSetNullAt(SDataCol *pCol, int index) {
+FORCE_INLINE void dataColSetNullAt(SDataCol *pCol, int index) {
   if (IS_VAR_DATA_TYPE(pCol->type)) {
     pCol->dataOff[index] = pCol->len;
     char *ptr = POINTER_SHIFT(pCol->pData, pCol->len);
@@ -399,8 +418,7 @@ void tdResetDataCols(SDataCols *pCols) {
     }
   }
 }
-
-void tdAppendDataRowToDataCol(SDataRow row, STSchema *pSchema, SDataCols *pCols) {
+static void tdAppendDataRowToDataCol(SDataRow row, STSchema *pSchema, SDataCols *pCols) {
   ASSERT(pCols->numOfRows == 0 || dataColsKeyLast(pCols) < dataRowKey(row));
 
   int rcol = 0;
@@ -419,7 +437,8 @@ void tdAppendDataRowToDataCol(SDataRow row, STSchema *pSchema, SDataCols *pCols)
     while (dcol < pCols->numOfCols) {
       SDataCol *pDataCol = &(pCols->cols[dcol]);
       if (rcol >= schemaNCols(pSchema)) {
-        dataColSetNullAt(pDataCol, pCols->numOfRows);
+        // dataColSetNullAt(pDataCol, pCols->numOfRows);
+        dataColAppendVal(pDataCol, getNullValue(pDataCol->type), pCols->numOfRows, pCols->maxPoints);
         dcol++;
         continue;
       }
@@ -433,7 +452,8 @@ void tdAppendDataRowToDataCol(SDataRow row, STSchema *pSchema, SDataCols *pCols)
       } else if (pRowCol->colId < pDataCol->colId) {
         rcol++;
       } else {
-        dataColSetNullAt(pDataCol, pCols->numOfRows);
+        // dataColSetNullAt(pDataCol, pCols->numOfRows);
+        dataColAppendVal(pDataCol, getNullValue(pDataCol->type), pCols->numOfRows, pCols->maxPoints);
         dcol++;
       }
     }
@@ -441,6 +461,62 @@ void tdAppendDataRowToDataCol(SDataRow row, STSchema *pSchema, SDataCols *pCols)
   pCols->numOfRows++;
 }
 
+static void tdAppendKvRowToDataCol(SKVRow row, STSchema *pSchema, SDataCols *pCols) {
+  ASSERT(pCols->numOfRows == 0 || dataColsKeyLast(pCols) < kvRowKey(row));
+
+  int rcol = 0;
+  int dcol = 0;
+
+  if (kvRowDeleted(row)) {
+    for (; dcol < pCols->numOfCols; dcol++) {
+      SDataCol *pDataCol = &(pCols->cols[dcol]);
+      if (dcol == 0) {
+        dataColAppendVal(pDataCol, kvRowValues(row), pCols->numOfRows, pCols->maxPoints);
+      } else {
+        dataColSetNullAt(pDataCol, pCols->numOfRows);
+      }
+    }
+  } else {
+    int nRowCols = kvRowNCols(row);
+
+    while (dcol < pCols->numOfCols) {
+      SDataCol *pDataCol = &(pCols->cols[dcol]);
+      if (rcol >= nRowCols || rcol >= schemaNCols(pSchema)) {
+        // dataColSetNullAt(pDataCol, pCols->numOfRows);
+        dataColAppendVal(pDataCol, getNullValue(pDataCol->type), pCols->numOfRows, pCols->maxPoints);
+        ++dcol;
+        continue;
+      }
+
+      SColIdx *colIdx = kvRowColIdxAt(row, rcol);
+
+      if (colIdx->colId == pDataCol->colId) {
+        void *value = tdGetKvRowDataOfCol(row, colIdx->offset);
+        dataColAppendVal(pDataCol, value, pCols->numOfRows, pCols->maxPoints);
+        ++dcol;
+        ++rcol;
+      } else if (colIdx->colId < pDataCol->colId) {
+        ++rcol;
+      } else {
+        // dataColSetNullAt(pDataCol, pCols->numOfRows);
+        dataColAppendVal(pDataCol, getNullValue(pDataCol->type), pCols->numOfRows, pCols->maxPoints);
+        ++dcol;
+      }
+    }
+  }
+  pCols->numOfRows++;
+}
+
+void tdAppendMemRowToDataCol(SMemRow row, STSchema *pSchema, SDataCols *pCols) {
+  if (isDataRow(row)) {
+    tdAppendDataRowToDataCol(memRowDataBody(row), pSchema, pCols);
+  } else if (isKvRow(row)) {
+    tdAppendKvRowToDataCol(memRowKvBody(row), pSchema, pCols);
+  } else {
+    ASSERT(0);
+  }
+}
+
 int tdMergeDataCols(SDataCols *target, SDataCols *source, int rowsToMerge, int *pOffset) {
   ASSERT(rowsToMerge > 0 && rowsToMerge <= source->numOfRows);
   ASSERT(target->numOfCols == source->numOfCols);
@@ -563,7 +639,7 @@ int tdSetKVRowDataOfCol(SKVRow *orow, int16_t colId, int8_t type, void *value) {
     nrow = malloc(kvRowLen(row) + sizeof(SColIdx) + diff);
     if (nrow == NULL) return -1;
 
-    kvRowSetLen(nrow, kvRowLen(row) + (int16_t)sizeof(SColIdx) + diff);
+    kvRowSetLen(nrow, kvRowLen(row) + (uint16_t)sizeof(SColIdx) + diff);
     kvRowSetNCols(nrow, kvRowNCols(row) + 1);
 
     if (ptr == NULL) {
@@ -605,8 +681,8 @@ int tdSetKVRowDataOfCol(SKVRow *orow, int16_t colId, int8_t type, void *value) {
       if (varDataTLen(value) == varDataTLen(pOldVal)) { // just update the column value in place
         memcpy(pOldVal, value, varDataTLen(value));
       } else { // need to reallocate the memory
-        int16_t diff = varDataTLen(value) - varDataTLen(pOldVal);
-        int16_t nlen = kvRowLen(row) + diff;
+        uint16_t diff = varDataTLen(value) - varDataTLen(pOldVal);
+        uint16_t nlen = kvRowLen(row) + diff;
         ASSERT(nlen > 0);
         nrow = malloc(nlen);
         if (nrow == NULL) return -1;

src/common/src/tglobal.c

@@ -244,6 +244,19 @@ int32_t tsdbDebugFlag = 131;
 int32_t cqDebugFlag = 131;
 int32_t fsDebugFlag = 135;
 
+#ifdef TD_TSZ
+//
+// lossy compress 6
+//
+char lossyColumns[32] = "";  // "float|double" means all float and double columns can be lossy compressed.  set empty can close lossy compress.
+// below option can take effect when tsLossyColumns not empty 
+double   fPrecision   = 1E-8;   // float column precision
+double   dPrecision   = 1E-16;  // double column precision
+uint32_t maxRange     = 500;    // max range
+uint32_t curRange     = 100;    // range
+char     Compressor[32] = "ZSTD_COMPRESSOR"; // ZSTD_COMPRESSOR or GZIP_COMPRESSOR 
+#endif
+
 int32_t (*monStartSystemFp)() = NULL;
 void (*monStopSystemFp)() = NULL;
 void (*monExecuteSQLFp)(char *sql) = NULL;
@@ -1517,6 +1530,62 @@ static void doInitGlobalConfig(void) {
   cfg.ptrLength = tListLen(tsTempDir);
   cfg.unitType = TAOS_CFG_UTYPE_NONE;
   taosInitConfigOption(cfg);
+
+#ifdef TD_TSZ
+  // lossy compress
+  cfg.option = "lossyColumns";
+  cfg.ptr = lossyColumns;
+  cfg.valType = TAOS_CFG_VTYPE_STRING;
+  cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG;
+  cfg.minValue = 0;
+  cfg.maxValue = 0;
+  cfg.ptrLength = tListLen(lossyColumns);
+  cfg.unitType = TAOS_CFG_UTYPE_NONE;
+  taosInitConfigOption(cfg);
+
+  cfg.option = "fPrecision";
+  cfg.ptr = &fPrecision;
+  cfg.valType = TAOS_CFG_VTYPE_DOUBLE;
+  cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG;
+  cfg.minValue = MIN_FLOAT;
+  cfg.maxValue = MAX_FLOAT;
+  cfg.ptrLength = 0;
+  cfg.unitType = TAOS_CFG_UTYPE_NONE;
+
+  
+  taosInitConfigOption(cfg);
+
+  cfg.option = "dPrecision";
+  cfg.ptr = &dPrecision;
+  cfg.valType = TAOS_CFG_VTYPE_DOUBLE;
+  cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG;
+  cfg.minValue = MIN_FLOAT;
+  cfg.maxValue = MAX_FLOAT;
+  cfg.ptrLength = 0;
+  cfg.unitType = TAOS_CFG_UTYPE_NONE;
+  taosInitConfigOption(cfg);
+
+  cfg.option = "maxRange";
+  cfg.ptr = &maxRange;
+  cfg.valType = TAOS_CFG_VTYPE_INT32;
+  cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG;
+  cfg.minValue = 0;
+  cfg.maxValue = 65536;
+  cfg.ptrLength = 0;
+  cfg.unitType = TAOS_CFG_UTYPE_NONE;
+  taosInitConfigOption(cfg);
+
+  cfg.option = "range";
+  cfg.ptr = &curRange;
+  cfg.valType = TAOS_CFG_VTYPE_INT32;
+  cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG;
+  cfg.minValue = 0;
+  cfg.maxValue = 65536;
+  cfg.ptrLength = 0;
+  cfg.unitType = TAOS_CFG_UTYPE_NONE;
+  taosInitConfigOption(cfg);
+#endif
+
 }
 
 void taosInitGlobalCfg() {

src/common/src/tname.c

@@ -306,7 +306,7 @@ bool tIsValidName(const SName* name) {
 SName* tNameDup(const SName* name) {
   assert(name != NULL);
 
-  SName* p = calloc(1, sizeof(SName));
+  SName* p = malloc(sizeof(SName));
   memcpy(p, name, sizeof(SName));
   return p;
 }

src/common/src/ttypes.c

@@ -38,7 +38,11 @@ const int32_t TYPE_BYTES[15] = {
 
 #define DO_STATICS(__sum, __min, __max, __minIndex, __maxIndex, _list, _index) \
   do {                                                                         \
+    if (_list[(_index)] >= (INT64_MAX - (__sum))) {                            \
+      __sum = INT64_MAX;                                                       \
+    } else {                                                                   \
       (__sum) += (_list)[(_index)];                                            \
+    }                                                                          \
     if ((__min) > (_list)[(_index)]) {                                         \
       (__min) = (_list)[(_index)];                                             \
       (__minIndex) = (_index);                                                 \
@@ -503,19 +507,21 @@ static uint8_t   nullTinyIntu = TSDB_DATA_UTINYINT_NULL;
 static uint16_t     nullSmallIntu = TSDB_DATA_USMALLINT_NULL;
 static uint32_t     nullIntu = TSDB_DATA_UINT_NULL;
 static uint64_t     nullBigIntu = TSDB_DATA_UBIGINT_NULL;
+static SBinaryNullT nullBinary = {1, TSDB_DATA_BINARY_NULL};
+static SNCharNullT  nullNchar = {4, TSDB_DATA_NCHAR_NULL};
 
-static union {
-  tstr str;
-  char pad[sizeof(tstr) + 4];
-} nullBinary = {.str = {.len = 1}}, nullNchar = {.str = {.len = 4}};
+// static union {
+//   tstr str;
+//   char pad[sizeof(tstr) + 4];
+// } nullBinary = {.str = {.len = 1}}, nullNchar = {.str = {.len = 4}};
 
-static void *nullValues[] = {
+static const void *nullValues[] = {
     &nullBool,     &nullTinyInt,   &nullSmallInt, &nullInt,    &nullBigInt,
     &nullFloat,    &nullDouble,    &nullBinary,   &nullBigInt, &nullNchar,
     &nullTinyIntu, &nullSmallIntu, &nullIntu,     &nullBigIntu,
 };
 
-void *getNullValue(int32_t type) {
+const void *getNullValue(int32_t type) {
   assert(type >= TSDB_DATA_TYPE_BOOL && type <= TSDB_DATA_TYPE_UBIGINT);
   return nullValues[type - 1];
 }

grafanaplugin @ 4a4d7909

-Subproject commit 3530c6df097134a410bacec6b3cd013ef38a61aa
+Subproject commit 4a4d79099b076b8ff12d5b4fdbcba54049a6866d

src/connector/jdbc/CMakeLists.txt

@@ -8,7 +8,7 @@ IF (TD_MVN_INSTALLED)
   ADD_CUSTOM_COMMAND(OUTPUT ${JDBC_CMD_NAME}
     POST_BUILD
     COMMAND mvn -Dmaven.test.skip=true install -f ${CMAKE_CURRENT_SOURCE_DIR}/pom.xml
-    COMMAND ${CMAKE_COMMAND} -E copy ${CMAKE_CURRENT_SOURCE_DIR}/target/taos-jdbcdriver-*-dist.jar ${LIBRARY_OUTPUT_PATH}
+    COMMAND ${CMAKE_COMMAND} -E copy ${CMAKE_CURRENT_SOURCE_DIR}/target/taos-jdbcdriver-2.0.33-dist.jar ${LIBRARY_OUTPUT_PATH}
     COMMAND mvn -Dmaven.test.skip=true clean -f ${CMAKE_CURRENT_SOURCE_DIR}/pom.xml
     COMMENT "build jdbc driver")
   ADD_CUSTOM_TARGET(${JDBC_TARGET_NAME} ALL WORKING_DIRECTORY ${EXECUTABLE_OUTPUT_PATH} DEPENDS ${JDBC_CMD_NAME})

src/connector/jdbc/deploy-pom.xml

@@ -5,7 +5,7 @@
 
   <groupId>com.taosdata.jdbc</groupId>
   <artifactId>taos-jdbcdriver</artifactId>
-  <version>2.0.32</version>
+  <version>2.0.33</version>
   <packaging>jar</packaging>
 
   <name>JDBCDriver</name>

src/connector/jdbc/pom.xml

@@ -37,13 +37,6 @@
         <maven.test.jvmargs></maven.test.jvmargs>
     </properties>
     <dependencies>
-        <dependency>
-            <groupId>junit</groupId>
-            <artifactId>junit</artifactId>
-            <version>4.13.1</version>
-            <scope>test</scope>
-        </dependency>
-        <!-- for restful -->
         <dependency>
             <groupId>org.apache.httpcomponents</groupId>
             <artifactId>httpclient</artifactId>
@@ -52,12 +45,18 @@
         <dependency>
             <groupId>com.alibaba</groupId>
             <artifactId>fastjson</artifactId>
-            <version>1.2.58</version>
+            <version>1.2.76</version>
         </dependency>
         <dependency>
             <groupId>com.google.guava</groupId>
             <artifactId>guava</artifactId>
-            <version>30.0-jre</version>
+            <version>30.1.1-jre</version>
+        </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>4.13.1</version>
+            <scope>test</scope>
         </dependency>
     </dependencies>
 

src/connector/jdbc/src/main/java/com/taosdata/jdbc/AbstractConnection.java

@@ -171,11 +171,7 @@ public abstract class AbstractConnection extends WrapperImpl implements Connecti
         // do nothing
     }
 
-    @Override
-    public Statement createStatement(int resultSetType, int resultSetConcurrency) throws SQLException {
-        if (isClosed())
-            throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_CONNECTION_CLOSED);
-
+    private void checkResultSetTypeAndResultSetConcurrency(int resultSetType, int resultSetConcurrency) throws SQLException {
         switch (resultSetType) {
             case ResultSet.TYPE_FORWARD_ONLY:
                 break;
@@ -194,7 +190,14 @@ public abstract class AbstractConnection extends WrapperImpl implements Connecti
             default:
                 throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_INVALID_VARIABLE);
         }
+    }
 
+    @Override
+    public Statement createStatement(int resultSetType, int resultSetConcurrency) throws SQLException {
+        if (isClosed())
+            throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_CONNECTION_CLOSED);
+
+        checkResultSetTypeAndResultSetConcurrency(resultSetType, resultSetConcurrency);
         return createStatement();
     }
 
@@ -203,24 +206,7 @@ public abstract class AbstractConnection extends WrapperImpl implements Connecti
         if (isClosed())
             throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_CONNECTION_CLOSED);
 
-        switch (resultSetType) {
-            case ResultSet.TYPE_FORWARD_ONLY:
-                break;
-            case ResultSet.TYPE_SCROLL_INSENSITIVE:
-            case ResultSet.TYPE_SCROLL_SENSITIVE:
-                throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_UNSUPPORTED_METHOD);
-            default:
-                throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_INVALID_VARIABLE);
-        }
-
-        switch (resultSetConcurrency) {
-            case ResultSet.CONCUR_READ_ONLY:
-                break;
-            case ResultSet.CONCUR_UPDATABLE:
-                throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_UNSUPPORTED_METHOD);
-            default:
-                throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_INVALID_VARIABLE);
-        }
+        checkResultSetTypeAndResultSetConcurrency(resultSetType, resultSetConcurrency);
         return prepareStatement(sql);
     }
 

src/connector/jdbc/src/main/java/com/taosdata/jdbc/EmptyResultSet.java

@@ -326,7 +326,7 @@ public class EmptyResultSet implements ResultSet {
 
     @Override
     public int getFetchDirection() throws SQLException {
-        return 0;
+        return ResultSet.FETCH_FORWARD;
     }
 
     @Override
@@ -341,12 +341,12 @@ public class EmptyResultSet implements ResultSet {
 
     @Override
     public int getType() throws SQLException {
-        return 0;
+        return ResultSet.TYPE_FORWARD_ONLY;
     }
 
     @Override
     public int getConcurrency() throws SQLException {
-        return 0;
+        return ResultSet.CONCUR_READ_ONLY;
     }
 
     @Override
@@ -746,7 +746,7 @@ public class EmptyResultSet implements ResultSet {
 
     @Override
     public int getHoldability() throws SQLException {
-        return 0;
+        return ResultSet.CLOSE_CURSORS_AT_COMMIT;
     }
 
     @Override

src/connector/jdbc/src/main/java/com/taosdata/jdbc/TSDBErrorNumbers.java

@@ -40,7 +40,7 @@ public class TSDBErrorNumbers {
     public static final int ERROR_JNI_FETCH_END = 0x2358;       // fetch to the end of resultSet
     public static final int ERROR_JNI_OUT_OF_MEMORY = 0x2359;   // JNI alloc memory failed
 
-    private static final Set<Integer> errorNumbers = new HashSet();
+    private static final Set<Integer> errorNumbers = new HashSet<>();
 
     static {
         errorNumbers.add(ERROR_CONNECTION_CLOSED);

src/connector/jdbc/src/main/java/com/taosdata/jdbc/TSDBJNIConnector.java

@@ -348,4 +348,13 @@ public class TSDBJNIConnector {
     }
 
     private native int closeStmt(long stmt, long con);
+
+    public void insertLines(String[] lines) throws SQLException {
+        int code = insertLinesImp(lines, this.taos);
+        if (code != TSDBConstants.JNI_SUCCESS) {
+            throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_UNKNOWN, "failed to insertLines");
+        }
+    }
+
+    private native int insertLinesImp(String[] lines, long conn);
 }

src/connector/jdbc/src/main/java/com/taosdata/jdbc/TSDBResultSetRowData.java

@@ -19,6 +19,7 @@ import com.taosdata.jdbc.utils.NullType;
 import java.math.BigDecimal;
 import java.sql.SQLException;
 import java.sql.Timestamp;
+import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Collections;
 
@@ -463,6 +464,25 @@ public class TSDBResultSetRowData {
         data.set(col, tsObj);
     }
 
+    /**
+     * this implementation is used for TDengine old version
+     */
+    public void setTimestamp(int col, long ts) {
+        //TODO: this implementation contains logical error
+        // when precision is us the (long ts) is 16 digital number
+        // when precision is ms, the (long ts) is 13 digital number
+        // we need a JNI function like this:
+        //      public void setTimestamp(int col, long epochSecond, long nanoAdjustment)
+        if (ts < 1_0000_0000_0000_0L) {
+            data.set(col, new Timestamp(ts));
+        } else {
+            long epochSec = ts / 1000_000L;
+            long nanoAdjustment = ts % 1000_000L * 1000L;
+            Timestamp timestamp = Timestamp.from(Instant.ofEpochSecond(epochSec, nanoAdjustment));
+            data.set(col, timestamp);
+        }
+    }
+
     public Timestamp getTimestamp(int col, int nativeType) {
         Object obj = data.get(col - 1);
         if (obj == null)

src/connector/jdbc/src/main/java/com/taosdata/jdbc/rs/RestfulResultSet.java

@@ -61,7 +61,7 @@ public class RestfulResultSet extends AbstractResultSet implements ResultSet {
             return;
         // parse row data
         for (int rowIndex = 0; rowIndex < data.size(); rowIndex++) {
-            List<Object> row = new ArrayList();
+            List<Object> row = new ArrayList<>();
             JSONArray jsonRow = data.getJSONArray(rowIndex);
             for (int colIndex = 0; colIndex < this.metaData.getColumnCount(); colIndex++) {
                 row.add(parseColumnData(jsonRow, colIndex, columns.get(colIndex).taos_type));

src/connector/jdbc/src/main/java/com/taosdata/jdbc/utils/HttpClientPoolUtil.java

@@ -9,6 +9,7 @@ import org.apache.http.client.protocol.HttpClientContext;
 import org.apache.http.conn.ConnectionKeepAliveStrategy;
 import org.apache.http.entity.StringEntity;
 import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.DefaultHttpRequestRetryHandler;
 import org.apache.http.impl.client.HttpClients;
 import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
 import org.apache.http.message.BasicHeaderElementIterator;
@@ -34,7 +35,11 @@ public class HttpClientPoolUtil {
             PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager();
             connectionManager.setDefaultMaxPerRoute(DEFAULT_MAX_PER_ROUTE);
             connectionManager.setMaxTotal(DEFAULT_MAX_TOTAL);
-            httpClient = HttpClients.custom().setKeepAliveStrategy(DEFAULT_KEEP_ALIVE_STRATEGY).setConnectionManager(connectionManager).build();
+            httpClient = HttpClients.custom()
+                    .setKeepAliveStrategy(DEFAULT_KEEP_ALIVE_STRATEGY)
+                    .setConnectionManager(connectionManager)
+                    .setRetryHandler(new DefaultHttpRequestRetryHandler(3, true))
+                    .build();
         }
     }
 

src/connector/jdbc/src/main/java/com/taosdata/jdbc/utils/OSUtils.java

@@ -4,14 +4,14 @@ public class OSUtils {
     private static final String OS = System.getProperty("os.name").toLowerCase();
 
     public static boolean isWindows() {
-        return OS.indexOf("win") >= 0;
+        return OS.contains("win");
     }
 
     public static boolean isMac() {
-        return OS.indexOf("mac") >= 0;
+        return OS.contains("mac");
     }
 
     public static boolean isLinux() {
-        return OS.indexOf("nux") >= 0;
+        return OS.contains("nux");
     }
 }

src/connector/jdbc/src/main/java/com/taosdata/jdbc/utils/SqlSyntaxValidator.java

@@ -16,7 +16,7 @@ package com.taosdata.jdbc.utils;
 
 public class SqlSyntaxValidator {
 
-    private static final String[] SQL = {"select", "insert", "import", "create", "use", "alter", "drop", "set", "show", "describe"};
+    private static final String[] SQL = {"select", "insert", "import", "create", "use", "alter", "drop", "set", "show", "describe", "reset"};
     private static final String[] updateSQL = {"insert", "import", "create", "use", "alter", "drop", "set"};
     private static final String[] querySQL = {"select", "show", "describe"};
 
@@ -61,29 +61,11 @@ public class SqlSyntaxValidator {
 
     public static boolean isUseSql(String sql) {
         return sql.trim().toLowerCase().startsWith("use");
-//                || sql.trim().toLowerCase().matches("create\\s*database.*") || sql.toLowerCase().toLowerCase().matches("drop\\s*database.*");
-    }
-
-    public static boolean isShowSql(String sql) {
-        return sql.trim().toLowerCase().startsWith("show");
-    }
-
-    public static boolean isDescribeSql(String sql) {
-        return sql.trim().toLowerCase().startsWith("describe");
-    }
-
-
-    public static boolean isInsertSql(String sql) {
-        return sql.trim().toLowerCase().startsWith("insert") || sql.trim().toLowerCase().startsWith("import");
     }
 
     public static boolean isSelectSql(String sql) {
         return sql.trim().toLowerCase().startsWith("select");
     }
 
-    public static boolean isShowDatabaseSql(String sql) {
-        return sql.trim().toLowerCase().matches("show\\s*databases");
-    }
-
 
 }

src/connector/jdbc/src/main/java/com/taosdata/jdbc/utils/TaosInfo.java

@@ -7,9 +7,9 @@ import java.util.concurrent.atomic.AtomicLong;
 public class TaosInfo implements TaosInfoMBean {
 
     private static volatile TaosInfo instance;
-    private AtomicLong connect_open = new AtomicLong();
-    private AtomicLong connect_close = new AtomicLong();
-    private AtomicLong statement_count = new AtomicLong();
+    private final AtomicLong connect_open = new AtomicLong();
+    private final AtomicLong connect_close = new AtomicLong();
+    private final AtomicLong statement_count = new AtomicLong();
 
     static {
         try {

src/connector/jdbc/src/test/java/com/taosdata/jdbc/TSDBConnectionTest.java

@@ -5,7 +5,6 @@ import org.junit.Assert;
 import org.junit.BeforeClass;
 import org.junit.Test;
 
-import javax.management.OperationsException;
 import java.sql.*;
 import java.util.Properties;
 

src/connector/jdbc/src/test/java/com/taosdata/jdbc/TSDBParameterMetaDataTest.java

 package com.taosdata.jdbc;
 
-import com.taosdata.jdbc.rs.RestfulParameterMetaData;
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.BeforeClass;

src/connector/jdbc/src/test/java/com/taosdata/jdbc/cases/DriverAutoloadTest.java

@@ -13,7 +13,7 @@ import java.util.Properties;
 public class DriverAutoloadTest {
 
     private Properties properties;
-    private String host = "127.0.0.1";
+    private final String host = "127.0.0.1";
 
     @Test
     public void testRestful() throws SQLException {