holo-clent-c, holo-client-go v1.3.7

Author: zgdgod

README.md

@@ -101,8 +101,8 @@ Connectors for Hologres
 ```xml
 <dependency>
     <groupId>com.alibaba.hologres</groupId>
-    <artifactId>hologres-connector-flink-1.13</artifactId>
-    <version>1.4.0</version>
+    <artifactId>hologres-connector-flink-1.17</artifactId>
+    <version>1.4.2</version>
     <classifier>jar-with-dependencies</classifier>
 </dependency>
 ```

holo-client-c/CMakeLists.txt

@@ -1,6 +1,8 @@
 cmake_minimum_required(VERSION 3.9)
 project(holo-client VERSION 1.1.0 DESCRIPTION "holo-client")
-execute_process(COMMAND git tag --sort=-taggerdate | head -n 1
+execute_process(
+	COMMAND git tag --sort=-taggerdate
+	COMMAND head -n 1
 	TIMEOUT 10
 	OUTPUT_VARIABLE GIT_VERSION
 	OUTPUT_STRIP_TRAILING_WHITESPACE
@@ -27,6 +29,7 @@ set(MY_PUBLIC_HEADERS
 	include/defs.h
     include/holo_client.h
     include/holo_config.h
+	include/logger.h
     include/record.h
     include/request.h
     include/table_schema.h
@@ -48,7 +51,9 @@ add_library(holo-client SHARED
 	src/get_collector.c
 	src/holo_client.c
 	src/holo_config.c
+	src/keywords.c
 	src/logger.c
+	src/logger_log4c.c
 	src/lp_map.c
 	src/meta_cache.c
 	src/metrics.c
@@ -82,7 +87,9 @@ add_library(${STATIC_LIB_NAME} STATIC
 	src/get_collector.c
 	src/holo_client.c
 	src/holo_config.c
+	src/keywords.c
 	src/logger.c
+	src/logger_log4c.c
 	src/lp_map.c
 	src/meta_cache.c
 	src/metrics.c

holo-client-c/Readme.md

@@ -54,7 +54,6 @@ find_path (LIBPQ_INCLUDE_DIRS NAMES libpq-fe.h HINTS /usr/include/postgresql NO_
 
 ### 写入普通表
 ```c
-holo_client_logger_open();
 char* connInfo = "host=xxxx.hologres.aliyuncs.com port=xxx dbname=xxxxx user=xxxxxxxxxx password=xxxxxxxxxx";
 HoloConfig config = holo_client_new_config(connInfo);
 
@@ -76,13 +75,11 @@ holo_client_submit(client, mutation);
 holo_client_flush_client(client);
 
 holo_client_close_client(client);
-holo_client_logger_close();
 ```
 
 ### 写入分区表
 若分区已存在，不论dynamicPartition为何值，写入数据都将插入到正确的分区表中；若分区不存在，dynamicPartition设置为true时，将会自动创建不存在的分区，否则插入失败
 ```c
-holo_client_logger_open();
 char* connInfo = "host=xxxx.hologres.aliyuncs.com port=xxx dbname=xxxxx user=xxxxxxxxxx password=xxxxxxxxxx";
 HoloConfig config = holo_client_new_config(connInfo);
 config.dynamicPartition = true;
@@ -105,12 +102,10 @@ holo_client_submit(client, mutation);
 holo_client_flush_client(client);
 
 holo_client_close_client(client);
-holo_client_logger_close();
 ```
 
 ### 写入含主键表
 ```c
-holo_client_logger_open();
 char* connInfo = "host=xxxx.hologres.aliyuncs.com port=xxx dbname=xxxxx user=xxxxxxxxxx password=xxxxxxxxxx";
 HoloConfig config = holo_client_new_config(connInfo);
 config.writeMode = INSERT_OR_UPDATE;//配置主键冲突时策略
@@ -133,11 +128,9 @@ holo_client_submit(client, mutation);
 holo_client_flush_client(client);
 
 holo_client_close_client(client);
-holo_client_logger_close();
 ```
 ### 根据主键删除
 ```c
-holo_client_logger_open();
 char* connInfo = "host=xxxx.hologres.aliyuncs.com port=xxx dbname=xxxxx user=xxxxxxxxxx password=xxxxxxxxxx";
 HoloConfig config = holo_client_new_config(connInfo);
 
@@ -157,12 +150,10 @@ holo_client_submit(client, mutation);
 holo_client_flush_client(client);
 
 holo_client_close_client(client);
-holo_client_logger_close();
 ```
 
 ### 写入失败的处理
 ```c
-holo_client_logger_open();
 char* connInfo = "host=xxxx.hologres.aliyuncs.com port=xxx dbname=xxxxx user=xxxxxxxxxx password=xxxxxxxxxx";
 HoloConfig config = holo_client_new_config(connInfo);
 config.exceptionHandler = handle_failed_record; //每条插入失败的record都会调用这个函数
@@ -180,12 +171,38 @@ holo_client_submit(client, mutation);
 holo_client_flush_client(client);
 
 holo_client_close_client(client);
-holo_client_logger_close();
+```
+
+v1.3.7之后的版本，增加了 holo_client_flush_client_with_errmsg(HoloClient*, char**) 接口，可以获取一次flush中引擎返回的第一个报错的message
+```c
+char* connInfo = "host=xxxx.hologres.aliyuncs.com port=xxx dbname=xxxxx user=xxxxxxxxxx password=xxxxxxxxxx";
+HoloConfig config = holo_client_new_config(connInfo);
+HoloClient* client = holo_client_new_client(config);
+
+HoloTableSchema* schema = holo_client_get_tableschema(client, "schema_name", "table_name", true);
+HoloMutation mutation = holo_client_new_mutation_request(schema);
+holo_client_set_req_int32_val_by_colindex(mutation, 0, 0);
+holo_client_submit(client, mutation); 
+
+mutation = holo_client_new_mutation_request(schema);
+holo_client_set_req_int32_val_by_colindex(mutation, 0, 1);
+holo_client_submit(client, mutation);
+
+char* errMsg = NULL;
+int ret = holo_client_flush_client_with_errmsg(client, &errMsg); //传入参数为errMsg的地址，即char**
+if (ret != HOLO_CLIENT_RET_OK) {
+    printf("holo client flush failed!\n");
+    if (errMsg != NULL) {
+        printf("errMsg: %s\n", errMsg); //如果flush后errMsg不为NULL，则返回了flush中引擎返回的第一个报错的message
+        free(errMsg); //message需要用户手动free，否则会发生内存泄漏
+    }
+}
+
+holo_client_close_client(client);
 ```
 
 ### 根据主键点查
 ```c
-holo_client_logger_open();
 char* connInfo = "host=xxxx.hologres.aliyuncs.com port=xxx dbname=xxxxx user=xxxxxxxxxx password=xxxxxxxxxx";
 HoloConfig config = holo_client_new_config(connInfo);
 
@@ -213,7 +230,6 @@ holo_client_destroy_get_request(get); //得到结果后需要手动释放
 holo_client_flush_client(client);
 
 holo_client_close_client(client);
-holo_client_logger_close();
 ```
 
 ## 附录
@@ -222,14 +238,16 @@ holo_client_logger_close();
 | --- | --- | --- |
 | connInfo| 无 | 必填, 格式:“host=xxxxxxxxx port=xxx dbname=xxxxx user=xxxxxxxxxx password=xxxxxxxxxx”| 
 | dynamicPartition | false | 若为true，当分区不存在时自动创建分区 |
+| unnestMode | false | 若为true，则会采用unnest模式写入，当batchSize不稳定时，不会产生多个PreparedStatement，节省服务端内存，但是会影响写入速度 |
 | useFixedFe | false | 若为true，写入将交给holo的FixedFE执行，会降低写入时的并发连接数 |
 | writeMode | INSERT_OR_REPLACE | 当INSERT目标表为有主键的表时采用不同策略<br>INSERT_OR_IGNORE 当主键冲突时，不写入<br>INSERT_OR_UPDATE 当主键冲突时，更新相应列<br>INSERT_OR_REPLACE当主键冲突时，更新所有列|
+| autoFlush | true | 是否自动触发批量提交，若为false，当合并后的插入数量达到batchSize，或合并后的插入数据字节数达到writeBatchByteSize，或距离上次提交超过writeMaxIntervalMs时，不会触发批量提交，此时再次插入会报错 |
 | batchSize | 512 | 每个写入线程的最大批次大小，在经过WriteMode合并后的插入数量达到batchSize时进行一次批量提交 |
 | threadSize | 1 | 写入并发线程数（每个并发占用1个数据库连接） |
 | shardCollectorSize | 2 * threadSize | 每个表的缓冲区个数，建议shardCollectorSize大于threadSize |
 | writeBatchByteSize | 2MB | 每个写入线程的最大批次bytes大小，在经过WriteMode合并后的插入数据字节数达到writeBatchByteSize时进行一次批量提交 |
 | writeBatchTotalByteSize | writeBatchByteSize * shardCollectorSize | 所有表所有缓冲区所有数据总和最大bytes大小，达到writeBatchTotalByteSize时全部提交 |
-| writeMaxIntervalMs | 10000 ms | 距离上次提交超过writeMaxIntervalMs会触发一次批量提交 |
+| writeMaxIntervalMs | 10000 ms | 距离上次提交超过writeMaxIntervalMs会触发一次批量提交，若小于等于0表示不检查writeMaxIntervalMs |
 | exceptionHandler | handle_exception_by_doing_nothing | 对于写入失败的数据的处理函数 |
 | retryCount | 3 | 当连接故障时的重试次数 |
 | retrySleepInitMs | 1000 ms | 每次重试的等待时间 = retrySleepInitMs + retry * retrySleepStepMs |
@@ -238,15 +256,12 @@ holo_client_logger_close();
 
 
 ### LOG说明
-holo client用的是log库是log4c
-
-在所有跟holo client有关的代码开始前需要调用 holo_client_logger_open();
-结束后需要调用 holo_client_logger_close();
+holo client默认用的是log库是log4c，用户可以自己实现自己的log函数，并使用holo_client_setup_logger设置log level和自定义log函数（在调用holo_client_new_client之前），参考logger.h。
 
 *注意
-
-holo_client_logger_open()和holo_client_logger_close()在整个代码里分别只能调用一次，建议在所有和holo client有关的代码的开始前和结束后调用。
-将log4crc文件复制到最终生成的可执行文件的同一文件夹下。如果要修改log的level和打印到的地点，就修改log4crc
+默认的log4c需要在所有跟holo client有关的代码开始前调用holo_client_logger_open()，结束后调用holo_client_logger_close();
+holo_client_logger_open()和holo_client_logger_close()在整个代码里分别只能调用一次。
+log4c库需要配置log4crc文件，并将log4crc文件复制到最终生成的可执行文件的同一文件夹下。如果要修改log的level和打印到的地点，就修改log4crc
 
 log4crc example1(打印到stdout):
 ```xml
@@ -383,7 +398,6 @@ import "C"
 import "unsafe"
 
 func main() {
-	C.holo_client_logger_open()
 	connInfo := "host=xxxx port=xx dbname=xxx user=xxx password=xxxxx"
 	cInfo := C.CString(connInfo)
 	config := C.holo_client_new_config(cInfo)
@@ -409,7 +423,6 @@ func main() {
 
 	C.holo_client_flush_client(client)
 	C.holo_client_close_client(client)
-	C.holo_client_logger_close()
 	C.free(unsafe.Pointer(cInfo))
 	C.free(unsafe.Pointer(cTableName))
 }
@@ -436,7 +449,6 @@ import (
 )
 
 func main() {
-	C.holo_client_logger_open()
 	connInfo := "host=xxxxx port=xxx dbname=xxxxx user=xxxxx password=xxxxx"
 	cInfo := C.CString(connInfo)
 	config := C.holo_client_new_config(cInfo)
@@ -465,7 +477,6 @@ func main() {
 	}
 	C.holo_client_flush_client(client)
 	C.holo_client_close_client(client)
-	C.holo_client_logger_close()
 	C.free(unsafe.Pointer(cInfo))
 	C.free(unsafe.Pointer(cTableName))
 	C.free(unsafe.Pointer(cSchemaName))

holo-client-c/include/defs.h

@@ -9,6 +9,17 @@
 # define __HOLO_CLIENT_END_DECLS
 #endif
 
+#define HOLO_MAX_LOG_LEN        1024
+
+#define HOLO_CLIENT_MAX_THREAD_SIZE 64
+
+#define HOLO_LOG_LEVEL_DEBUG    200
+#define HOLO_LOG_LEVEL_INFO     300
+#define HOLO_LOG_LEVEL_WARNING  400
+#define HOLO_LOG_LEVEL_ERROR    500
+#define HOLO_LOG_LEVEL_FATAL    600
+#define HOLO_LOG_LEVEL_NONE     10000
+
 /*
  * Oids of holo-client-c support types
  * copy from pg_type_d.h
@@ -45,4 +56,34 @@
 #define HOLO_TYPE_DATE_ARRAY    1182
 #define HOLO_TYPE_TIMESTAMP_ARRAY   1115
 #define HOLO_TYPE_TIMESTAMPTZ_ARRAY 1185
+
+/**
+ * Errcode for holo-client-c
+ */
+#define HOLO_CLIENT_RET_OK                  0
+#define HOLO_CLIENT_RET_FAIL                -1
+#define HOLO_CLIENT_FLUSH_FAIL              -100
+
+#define HOLO_CLIENT_INVALID_PARAM           -200
+#define HOLO_CLIENT_INVALID_COL_IDX         -201
+#define HOLO_CLIENT_INVALID_COL_NAME        -202
+#define HOLO_CLIENT_TYPE_NOT_MATCH          -203
+#define HOLO_CLIENT_COL_ALREADY_SET         -204
+#define HOLO_CLIENT_NOT_NULL_BUT_SET_NULL   -205
+#define HOLO_CLIENT_COL_NOT_PK              -206
+
+#define HOLO_CLIENT_CHECK_CONSTRAINT_FAIL   -300
+#define HOLO_CLIENT_PK_IS_NULL              -301
+#define HOLO_CLIENT_PARTITION_IS_NULL       -302
+#define HOLO_CLIENT_DELETE_NO_PK_TABLE      -303
+#define HOLO_CLIENT_PARTITION_NOT_EXIST     -304
+#define HOLO_CLIENT_PK_NOT_SET_IN_DELETE    -305
+#define HOLO_CLIENT_NOT_NULL_BUT_NOT_SET    -306
+#define HOLO_CLIENT_PARTITION_META_CHANGE   -307
+
+#define HOLO_CLIENT_EXCEED_MAX_BYTE         -400
+#define HOLO_CLIENT_EXCEED_MAX_NUM          -401
+#define HOLO_CLIENT_EXCEED_MAX_INTERVAL     -402
+#define HOLO_CLIENT_EXCEED_MAX_TOTAL_BYTE   -403
+
 #endif

holo-client-c/include/holo_client.h

@@ -2,6 +2,7 @@
 #define _HOLO_CLIENT_H_
 
 #include "defs.h"
+#include "logger.h"
 #include "request.h"
 #include "worker_pool.h"
 #include "record.h"
@@ -15,18 +16,22 @@ HoloClient* holo_client_new_client(HoloConfig);
 HoloClient* holo_client_new_client_with_workerpool(HoloConfig, HoloWorkerPool*);
 
 int holo_client_flush_client(HoloClient*);
+int holo_client_flush_client_with_errmsg(HoloClient*, char**);
 int holo_client_close_client(HoloClient*);
 
 //Schema name可以为NULL，会被设为"public"
 HoloTableSchema* holo_client_get_tableschema(HoloClient*, const char*, const char*, bool);
+HoloTableSchema* holo_client_get_tableschema_with_errmsg(HoloClient*, const char*, const char*, bool, char**);
 
 int holo_client_submit(HoloClient*, HoloMutation);
+int holo_client_submit_with_errmsg(HoloClient*, HoloMutation, char**);
+int holo_client_submit_with_attachments(HoloClient*, HoloMutation, int64_t, int64_t);
 int holo_client_get(HoloClient*, HoloGet);
 HoloRecord* holo_client_get_record(const HoloGet);
 char* holo_client_get_record_val(const HoloRecord* record, int colIndex);
 
-void holo_client_logger_open();
-void holo_client_logger_close();
+//通过接口返回的error code，可以获取相应错误的error message（字符串形式）
+const char* holo_client_get_errmsg_with_errcode(int errCode);
 
 __HOLO_CLIENT_END_DECLS
 

holo-client-c/include/holo_config.h

@@ -35,6 +35,8 @@ typedef struct _HoloConfig {
     bool dynamicPartition; //false
     bool useFixedFe; //false
     int connectionSizeWhenUseFixedFe; //1
+    bool autoFlush; //true
+    bool unnestMode; //false
 
     /* 
      * HoloExceptionHandler callback, and pointer which will be passed to the actual callback as parameter

holo-client-c/include/logger.h

@@ -0,0 +1,25 @@
+#ifndef _LOGGER_H_
+#define _LOGGER_H_
+
+#include "defs.h"
+
+__HOLO_CLIENT_BEGIN_DECLS
+
+/**
+ * Definition of HoloLogger callback, the default logger is log4c
+ * you can make your own implement of HoloLogger, and use holo_client_setup_logger to setup
+ * the first arg is log level, the second is message that will be logged
+ * if you don't want any log, set logger to holo_client_log_do_nothing, which will reduce overhead
+ */
+typedef void* (*HoloLogger)(const int, const char*);
+
+void* holo_client_log_do_nothing(const int logLevel, const char* msg);
+
+void holo_client_setup_logger(HoloLogger logger, int loglevel);
+
+void holo_client_logger_open();
+void holo_client_logger_close();
+
+__HOLO_CLIENT_END_DECLS
+
+#endif

holo-client-c/include/table_schema.h

@@ -34,6 +34,9 @@ typedef struct _HoloTableSchema {
     int partitionColumn;
 } HoloTableSchema;
 
+const char* holo_client_get_column_name(HoloTableSchema* schema, int colIndex);
+const char* holo_client_get_column_type_name(HoloTableSchema* schema, int colIndex);
+
 //通过HoloColumn的type字段（无符号整型数字），可以获取HoloColumn的类型（字符串）
 const char* holo_client_get_type_name_with_type_oid(unsigned int typeOid);
 

holo-client-c/src/action.c

@@ -1,6 +1,6 @@
 #include "action.h"
 #include "utils.h"
-#include "logger.h"
+#include "logger_private.h"
 
 ActionItem* create_action_item(Action* action) {
     ActionItem* item = MALLOC(1, ActionItem);