Feature/sangshuduo/td 4959 taosdump support binarydump (#8568)

* [TD4959]<feature>: taosdump support dumping binary data.

* [TD-4959]<feature>: taosdump support avro format.

add avro as submodule.

* [TD-4959]<feature>: taosdump support avro format.

add avro as submodule.

* taosdump code refactoring.

* add avro argument interface.

* fix arm32 build issue.

* hugh refactoring.

* fix tag value rather than reuse note.

* redefine note/varible

* taosdump text-format dump reworked.

* use avro standard schema format.

* build avro-c on linux only.

* avro schema write works.

* support bool,tinyint,smallint,int,bigint,float,double,timestamp write

* support nchar dump

* support all codecs

* adjust options order.

* refactor code

* multithread dump in sql refactored.

* fix nchar quote mark.

* fix bash script error

* fix arg parse issue.

* add avro dump in stmt interfaces.

* basic dump in works.

* fix src/kit/taosdump/CMakeLists.txt for CentOS

* fix memory leak.

* no avro utils installed

* make cmake and install script supports avro.

* fix makerpm.sh mistake.

* fix CMakeLists.txt and partially add null support
Co-authored-by: NShuduo Sang <sdsang@taosdata.com>

cmake/define.inc

@@ -150,6 +150,8 @@ ENDIF ()
 
 IF ("${AVRO_SUPPORT}" MATCHES "true")
   SET(TD_AVRO_SUPPORT TRUE)
+ELSEIF ("${AVRO_SUPPORT}" MATCHES "false")
+  SET(TD_AVRO_SUPPORT FALSE)
 ENDIF ()
 
 IF (TD_AVRO_SUPPORT)

deps/CMakeLists.txt

@@ -25,7 +25,7 @@ IF (TD_DARWIN AND TD_MQTT)
   ADD_SUBDIRECTORY(MQTT-C)
 ENDIF ()
 
-IF (AVRO_SUPPORT)
+IF (TD_AVRO_SUPPORT)
   MESSAGE("")
   MESSAGE("${Green} ENABLE avro format support ${ColourReset}")
   MESSAGE("")

packaging/deb/DEBIAN/control

@@ -11,4 +11,3 @@ Maintainer: support@taosdata.com
 Provides: taosdata
 Homepage: http://taosdata.com
 Description: Big Data Platform Designed and Optimized for IoT.
-

packaging/deb/makedeb.sh

@@ -73,8 +73,19 @@ cp -r ${top_dir}/src/connector/go                   ${pkg_dir}${install_home_pat
 cp -r ${top_dir}/src/connector/nodejs               ${pkg_dir}${install_home_path}/connector
 cp ${compile_dir}/build/lib/taos-jdbcdriver*.*  ${pkg_dir}${install_home_path}/connector ||:
 
+install_user_local_path="/usr/local"
+
+if [ -f ${compile_dir}/build/lib/libavro.so.23.0.0 ]; then
+    mkdir -p ${pkg_dir}${install_user_local_path}/lib
+    cp ${compile_dir}/build/lib/libavro.so.23.0.0 ${pkg_dir}${install_user_local_path}/lib/
+    ln -sf libavro.so.23.0.0 ${pkg_dir}${install_user_local_path}/lib/libavro.so.23
+    ln -sf libavro.so.23 ${pkg_dir}${install_user_local_path}/lib/libavro.so
+fi
+if [ -f ${compile_dir}/build/lib/libavro.a ]; then
+    cp ${compile_dir}/build/lib/libavro.a ${pkg_dir}${install_user_local_path}/lib/
+fi
+
 if [ -f ${compile_dir}/build/bin/jemalloc-config ]; then
-    install_user_local_path="/usr/local"
     mkdir -p ${pkg_dir}${install_user_local_path}/{bin,lib,lib/pkgconfig,include/jemalloc,share/doc/jemalloc,share/man/man3}
     cp ${compile_dir}/build/bin/jemalloc-config ${pkg_dir}${install_user_local_path}/bin/
     if [ -f ${compile_dir}/build/bin/jemalloc.sh ]; then
@@ -114,6 +125,10 @@ chmod 755 ${pkg_dir}/DEBIAN/*
 debver="Version: "$tdengine_ver
 sed -i "2c$debver" ${pkg_dir}/DEBIAN/control
 
+if [ -f ${compile_dir}/build/lib/libavro.so.23.0.0 ]; then
+    sed -i.bak "s/#Depends: no/Depends: libjansson4, libsnappy1v5/g" ${pkg_dir}/DEBIAN/control
+fi
+
 #get taos version, then set deb name
 
 
@@ -145,4 +160,3 @@ cp ${pkg_dir}/*.deb ${output_dir}
 
 # clean tmep dir
 rm -rf ${pkg_dir}
-

packaging/rpm/makerpm.sh

@@ -32,20 +32,20 @@ if command -v sudo > /dev/null; then
 fi
 
 function cp_rpm_package() {
-local cur_dir
-cd $1
-cur_dir=$(pwd)
-
-for dirlist in "$(ls ${cur_dir})"; do
-  if test -d ${dirlist}; then
-    cd ${dirlist}
-    cp_rpm_package ${cur_dir}/${dirlist}
-    cd ..
-  fi
-  if test -e ${dirlist}; then
-    cp ${cur_dir}/${dirlist} ${output_dir}/TDengine-${tdengine_ver}.rpm
-  fi
-done
+    local cur_dir
+    cd $1
+    cur_dir=$(pwd)
+
+    for dirlist in "$(ls ${cur_dir})"; do
+        if test -d ${dirlist}; then
+            cd ${dirlist}
+            cp_rpm_package ${cur_dir}/${dirlist}
+            cd ..
+        fi
+        if test -e ${dirlist}; then
+            cp ${cur_dir}/${dirlist} ${output_dir}/TDengine-${tdengine_ver}.rpm
+        fi
+    done
 }
 
 if [ -d ${pkg_dir} ]; then
@@ -56,6 +56,10 @@ cd ${pkg_dir}
 
 ${csudo} mkdir -p BUILD BUILDROOT RPMS SOURCES SPECS SRPMS
 
+if [ -f ${compile_dir}/build/lib/libavro.so.23.0.0 ]; then
+    sed -i.bak 's/#Requires:/Requires: jansson snappy/g' ${spec_file}
+fi
+
 ${csudo} rpmbuild --define="_version ${tdengine_ver}" --define="_topdir ${pkg_dir}" --define="_compiledir ${compile_dir}" -bb ${spec_file}
 
 # copy rpm package to output_dir, and modify package name, then clean temp dir

packaging/rpm/tdengine.spec

@@ -79,6 +79,14 @@ cp -r %{_compiledir}/../src/connector/nodejs        %{buildroot}%{homepath}/conn
 cp %{_compiledir}/build/lib/taos-jdbcdriver*.*      %{buildroot}%{homepath}/connector ||:
 cp -r %{_compiledir}/../tests/examples/*            %{buildroot}%{homepath}/examples
 
+if [ -f %{_compiledir}/build/lib/libavro.so.23.0.0 ]; then
+    cp %{_compiledir}/build/lib/libavro.so.23.0.0 %{buildroot}%{homepath}/driver
+    ln -sf libavro.so.23.0.0 %{buildroot}%{homepath}/driver/libavro.so.23
+    ln -sf libavro.so.23 %{buildroot}%{homepath}/driver/libavro.so
+fi
+if [ -f %{_compiledir}/build/lib/libavro.a ]; then
+    cp %{_compiledir}/build/lib/libavro.a %{buildroot}%{homepath}/driver
+fi
 
 if [ -f %{_compiledir}/build/bin/jemalloc-config ]; then
     mkdir -p %{buildroot}%{userlocalpath}/bin

packaging/tools/post.sh

@@ -2,7 +2,7 @@
 #
 # This file is used to install tdengine rpm package on centos systems. The operating system
 # is required to use systemd to manage services at boot
-#set -x
+# set -x
 
 iplist=""
 serverFqdn=""
@@ -86,6 +86,24 @@ function install_include() {
     ${csudo} ln -s ${inc_dir}/taoserror.h ${inc_link_dir}/taoserror.h
 }
 
+function install_avro_lib() {
+    ${csudo} rm -f ${lib_link_dir}/libavro* || :
+    ${csudo} rm -f ${lib64_link_dir}/libavro* || :
+
+    if [[ -f ${lib_dir}/libavro.so.23.0.0 ]]; then
+        ${csudo} ln -s ${lib_dir}/libavro.so.23.0.0 ${lib_link_dir}/libavro.so.23.0.0
+        ${csudo} ln -s ${lib_link_dir}/libavro.so.23.0.0 ${lib_link_dir}/libavro.so.23
+        ${csudo} ln -s ${lib_link_dir}/libavro.so.23 ${lib_link_dir}/libavro.so
+
+        if [[ -d ${lib64_link_dir} && ! -e ${lib64_link_dir}/libavro.so ]]; then
+            ${csudo} ln -s ${lib_dir}/libavro.so.23.0.0 ${lib64_link_dir}/libavro.so.23.0.0           || :
+            ${csudo} ln -s ${lib64_link_dir}/libavro.so.23.0.0 ${lib64_link_dir}/libavro.so.23   || :
+            ${csudo} ln -s ${lib64_link_dir}/libavro.so.23 ${lib64_link_dir}/libavro.so   || :
+        fi
+    fi
+
+    ${csudo} ldconfig
+}
 function install_lib() {
     ${csudo} rm -f ${lib_link_dir}/libtaos* || :
     ${csudo} rm -f ${lib64_link_dir}/libtaos* || :
@@ -97,6 +115,8 @@ function install_lib() {
       ${csudo} ln -s ${lib_dir}/libtaos.* ${lib64_link_dir}/libtaos.so.1           || :
       ${csudo} ln -s ${lib64_link_dir}/libtaos.so.1 ${lib64_link_dir}/libtaos.so   || :
     fi
+
+    ${csudo} ldconfig
 }
 
 function install_bin() {
@@ -474,6 +494,7 @@ function install_TDengine() {
     # Install include, lib, binary and service
     install_include
     install_lib
+    install_avro_lib
     install_bin
     install_service
     install_config

src/kit/taosdump/CMakeLists.txt

@@ -67,13 +67,13 @@ LINK_DIRECTORIES(${CMAKE_BINARY_DIR}/build/lib ${CMAKE_BINARY_DIR}/build/lib64)
 IF (TD_LINUX)
   ADD_EXECUTABLE(taosdump ${SRC})
   IF (TD_SOMODE_STATIC)
-    IF (AVRO_SUPPORT)
+    IF (TD_AVRO_SUPPORT)
       TARGET_LINK_LIBRARIES(taosdump taos_static avro jansson)
     ELSE ()
       TARGET_LINK_LIBRARIES(taosdump taos_static)
     ENDIF()
   ELSE ()
-    IF (AVRO_SUPPORT)
+    IF (TD_AVRO_SUPPORT)
       TARGET_LINK_LIBRARIES(taosdump taos avro jansson)
     ELSE ()
       TARGET_LINK_LIBRARIES(taosdump taos)

src/kit/taosdump/taosdump.c

@@ -1977,45 +1977,87 @@ static int64_t writeResultToAvro(
             int len;
             switch (fields[col].type) {
                 case TSDB_DATA_TYPE_BOOL:
-                    avro_value_set_boolean(&value,
-                            ((((int32_t)(*((char *)row[col])))==1)?1:0));
+                    if (NULL == row[col]) {
+                        avro_value_set_int(&value, TSDB_DATA_BOOL_NULL);
+                    } else {
+                        avro_value_set_boolean(&value,
+                                ((((int32_t)(*((char *)row[col])))==1)?1:0));
+                    }
                     break;
 
                 case TSDB_DATA_TYPE_TINYINT:
-                    avro_value_set_int(&value, *((int8_t *)row[col]));
+                    if (NULL == row[col]) {
+                        avro_value_set_int(&value, TSDB_DATA_TINYINT_NULL);
+                    } else {
+                        avro_value_set_int(&value, *((int8_t *)row[col]));
+                    }
                     break;
 
                 case TSDB_DATA_TYPE_SMALLINT:
-                    avro_value_set_int(&value, *((int16_t *)row[col]));
+                    if (NULL == row[col]) {
+                        avro_value_set_int(&value, TSDB_DATA_SMALLINT_NULL);
+                    } else {
+                        avro_value_set_int(&value, *((int16_t *)row[col]));
+                    }
                     break;
 
                 case TSDB_DATA_TYPE_INT:
-                    avro_value_set_int(&value, *((int32_t *)row[col]));
+                    if (NULL == row[col]) {
+                        avro_value_set_int(&value, TSDB_DATA_INT_NULL);
+                    } else {
+                        avro_value_set_int(&value, *((int32_t *)row[col]));
+                    }
                     break;
 
                 case TSDB_DATA_TYPE_BIGINT:
-                    avro_value_set_long(&value, *((int64_t *)row[col]));
+                    if (NULL == row[col]) {
+                        avro_value_set_long(&value, TSDB_DATA_BIGINT_NULL);
+                    } else {
+                        avro_value_set_long(&value, *((int64_t *)row[col]));
+                    }
                     break;
 
                 case TSDB_DATA_TYPE_FLOAT:
-                    avro_value_set_float(&value, GET_FLOAT_VAL(row[col]));
+                    if (NULL == row[col]) {
+                        avro_value_set_float(&value, TSDB_DATA_FLOAT_NULL);
+                    } else {
+                        avro_value_set_float(&value, GET_FLOAT_VAL(row[col]));
+                    }
                     break;
 
                 case TSDB_DATA_TYPE_DOUBLE:
-                    avro_value_set_double(&value, GET_DOUBLE_VAL(row[col]));
+                    if (NULL == row[col]) {
+                        avro_value_set_double(&value, TSDB_DATA_DOUBLE_NULL);
+                    } else {
+                        avro_value_set_double(&value, GET_DOUBLE_VAL(row[col]));
+                    }
                     break;
 
                 case TSDB_DATA_TYPE_BINARY:
-                    avro_value_set_string(&value, (char *)row[col]);
+                    if (NULL == row[col]) {
+                        avro_value_set_string(&value,
+                                (char *)NULL);
+                    } else {
+                        avro_value_set_string(&value, (char *)row[col]);
+                    }
                     break;
 
                 case TSDB_DATA_TYPE_NCHAR:
-                    len = strlen((char*)row[col]);
-                    avro_value_set_bytes(&value, (void*)(row[col]),len);
+                    if (NULL == row[col]) {
+                        avro_value_set_bytes(&value,
+                                (void*)NULL,0);
+                    } else {
+                        len = strlen((char*)row[col]);
+                        avro_value_set_bytes(&value, (void*)(row[col]),len);
+                    }
                     break;
 
                 case TSDB_DATA_TYPE_TIMESTAMP:
-                    avro_value_set_long(&value, *((int64_t *)row[col]));
+                    if (NULL == row[col]) {
+                        avro_value_set_long(&value, TSDB_DATA_BIGINT_NULL);
+                    } else {
+                        avro_value_set_long(&value, *((int64_t *)row[col]));
+                    }
                     break;
 
                 default:
@@ -2199,6 +2241,8 @@ static int dumpInOneAvroFile(char* fcharset,
 
             FieldStruct *field = (FieldStruct *)(recordSchema->fields + sizeof(FieldStruct) * i);
 
+            bind->is_null = NULL;
+            int is_null = 1;
             if (0 == i) {
                 int64_t *ts = malloc(sizeof(int64_t));
                 assert(ts);
@@ -2210,9 +2254,9 @@ static int dumpInOneAvroFile(char* fcharset,
                 bind->buffer_length = sizeof(int64_t);
                 bind->buffer = ts;
                 bind->length = &bind->buffer_length;
-                bind->is_null = NULL;
             } else if (0 == avro_value_get_by_name(
                         &value, field->name, &field_value, NULL)) {
+
                 if (0 == strcasecmp(tableDes->cols[i].type, "int")) {
                     int32_t *n32 = malloc(sizeof(int32_t));
                     assert(n32);
@@ -2263,19 +2307,30 @@ static int dumpInOneAvroFile(char* fcharset,
                     assert(f);
 
                     avro_value_get_float(&field_value, f);
-                    debugPrint("%f | ", *f);
+                    if (TSDB_DATA_FLOAT_NULL == *f) {
+                        debugPrint("%s | ", "NULL");
+                        bind->is_null = &is_null;
+                    } else {
+                        debugPrint("%f | ", *f);
+                        bind->buffer = f;
+                    }
                     bind->buffer_type = TSDB_DATA_TYPE_FLOAT;
                     bind->buffer_length = sizeof(float);
-                    bind->buffer = f;
                 } else if (0 == strcasecmp(tableDes->cols[i].type, "double")) {
                     double *dbl = malloc(sizeof(double));
                     assert(dbl);
 
                     avro_value_get_double(&field_value, dbl);
-                    debugPrint("%f | ", *dbl);
+                    if (TSDB_DATA_DOUBLE_NULL == *dbl) {
+                        debugPrint("%s | ", "NULL");
+                        bind->is_null = &is_null;
+                    } else {
+                        debugPrint("%f | ", *dbl);
+                        bind->buffer = dbl;
+                    }
+                    bind->buffer = dbl;
                     bind->buffer_type = TSDB_DATA_TYPE_DOUBLE;
                     bind->buffer_length = sizeof(double);
-                    bind->buffer = dbl;
                 } else if (0 == strcasecmp(tableDes->cols[i].type, "binary")) {
                     size_t size;
 
@@ -2306,7 +2361,6 @@ static int dumpInOneAvroFile(char* fcharset,
                 }
 
                 bind->length = &bind->buffer_length;
-                bind->is_null = NULL;
             }
 
         }
@@ -2459,10 +2513,7 @@ static int64_t writeResultToSql(TAOS_RES *res, FILE *fp, char *dbName, char *tbN
 
     int32_t  sql_buf_len = g_args.max_sql_len;
     char* tmpBuffer = (char *)calloc(1, sql_buf_len + 128);
-    if (tmpBuffer == NULL) {
-        errorPrint("failed to allocate %d memory\n", sql_buf_len + 128);
-        return -1;
-    }
+    assert(tmpBuffer);
 
     char *pstr = tmpBuffer;