holo-client 2.5.4, update README

Author: zgdgod

holo-client-c/CMakeLists.txt

@@ -12,6 +12,9 @@ add_definitions(-DBUILD_VERSION=\"${GIT_VERSION}\")
 include(GNUInstallDirs)
 add_compile_options(-Wall)
 add_compile_options(-Werror)
+# 编译带符号
+# add_compile_options(-O0)
+# add_compile_options(-g3)
 # 当前使用的c编译器
 message("-- CMAKE_C_COMPILER: ${CMAKE_C_COMPILER_ID}")
 # 当前使用的c编译器的版本

holo-client-c/src/holo_client.c

@@ -4,6 +4,8 @@
 #include "logger_private.h"
 #include "unistd.h"
 
+#define INT32_STRING_MAX_LENGTH 12
+
 int check_mutation(HoloMutation mutation);
 int check_partition(HoloClient* client, HoloMutation mutation);
 int check_get(HoloGet get);
@@ -95,6 +97,7 @@ int holo_client_close_client(HoloClient* client) {
 }
 
 HoloTableSchema* holo_client_get_tableschema_by_tablename(HoloClient* client, HoloTableName name, bool withCache, char** errMsgAddr) {
+    LOG_DEBUG("get table schema by table name:%s", name.fullName);
     Meta meta = NULL;
     HoloTableSchema* schema = NULL;
     if(withCache) {
@@ -292,23 +295,38 @@ int check_mutation(HoloMutation mutation) {
     return HOLO_CLIENT_RET_OK;
 }
 
+static int get_int_partition_value_string(char* intValueStr, const char* intPartitionValue) {
+    char* copyValue = MALLOC(4, char);
+    memcpy(copyValue, intPartitionValue, 4);
+    endian_swap(copyValue, 4);
+    snprintf(intValueStr, INT32_STRING_MAX_LENGTH, "%d", *(int32_t*)copyValue);
+    endian_swap(copyValue, 4);
+    FREE(copyValue);
+    return HOLO_CLIENT_RET_OK;
+}
+
 void* find_partition_table_name(PGconn* conn, void* arg) {
     HoloRecord* record = arg;
     const char* findPartitionSql = "with inh as (SELECT i.inhrelid, i.inhparent FROM pg_catalog.pg_class c LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace LEFT JOIN pg_catalog.pg_inherits i on c.oid=i.inhparent where n.nspname = $1 and c.relname= $2) select n.nspname as schema_name, c.relname as table_name, inh.inhrelid, inh.inhparent, p.partstrat, pg_get_expr(c.relpartbound, c.oid, true) as part_expr, p.partdefid, p.partnatts, p.partattrs from inh join pg_catalog.pg_class c on inh.inhrelid = c.oid join pg_catalog.pg_namespace n on c.relnamespace = n.oid join pg_partitioned_table p on p.partrelid = inh.inhparent where pg_get_expr(c.relpartbound, c.oid, true) = $3 limit 1";
     const char* schemaName = record->schema->tableName->schemaName;
     const char* tableName = record->schema->tableName->tableName;
-    int length = strlen(record->values[record->schema->partitionColumn]) + 19;
-    char* partitionInfo = MALLOC(length, char);
-    const char* params[3] = {schemaName, tableName, partitionInfo};
+    char* partitionInfo = NULL;
     PGresult* res = NULL;
     HoloTableName* name = NULL;
-    Oid type = record->schema->columns[record->schema->partitionColumn].type;
-    if (type == HOLO_TYPE_TEXT || type == HOLO_TYPE_VARCHAR) {
+    char intValStr[INT32_STRING_MAX_LENGTH];
+    if (record->valueFormats[record->schema->partitionColumn] == 1) {
+        //int
+        get_int_partition_value_string(intValStr, record->values[record->schema->partitionColumn]);
+        int length = strlen(intValStr) + 17;
+        partitionInfo = MALLOC(length, char);
+        snprintf(partitionInfo, length, "FOR VALUES IN (%s)", intValStr);
+    } else {
         //text or varchar
+        int length = strlen(record->values[record->schema->partitionColumn]) + 19;
+        partitionInfo = MALLOC(length, char);
         snprintf(partitionInfo, length, "FOR VALUES IN ('%s')", record->values[record->schema->partitionColumn]);
-    } else {
-        snprintf(partitionInfo, length, "FOR VALUES IN (%s)", record->values[record->schema->partitionColumn]);
     }
+    const char* params[3] = {schemaName, tableName, partitionInfo};
     res = PQexecParams(conn, findPartitionSql, 3, NULL, params, NULL, NULL, 0);
     if (PQntuples(res) == 0) {
         PQclear(res);
@@ -328,13 +346,13 @@ void* retry_create_partition_child_table(PGconn* conn, void* arg) {
     int retry = 0;
     HoloRecord* record = arg;
     HoloTableSchema* parentSchema = record->schema;
-    char intVal[12];
-    char* partitionValue = record->values[record->schema->partitionColumn];
+    char* partitionValue = NULL;
+    char intValStr[INT32_STRING_MAX_LENGTH];
     if (record->valueFormats[parentSchema->partitionColumn] == 1) {
-        endian_swap(partitionValue, 4);
-        snprintf(intVal, 12, "%d", *(int32_t*)partitionValue);
-        endian_swap(partitionValue, 4);
-        partitionValue = intVal;
+        get_int_partition_value_string(intValStr, record->values[record->schema->partitionColumn]);
+        partitionValue = intValStr;
+    } else {
+        partitionValue = record->values[record->schema->partitionColumn];
     }
     int maxTableNameLength = strlen(parentSchema->tableName->tableName) + strlen(partitionValue) + 22;
     int maxSqlLength = 2 * strlen(parentSchema->tableName->schemaName) + strlen(parentSchema->tableName->tableName) + maxTableNameLength + strlen(partitionValue) + 57;
@@ -411,15 +429,14 @@ int check_partition(HoloClient* client, HoloMutation mutation) {
     HoloTableName* name = NULL;
     Sql sql = NULL;
     char* partition = NULL;
-    char intVal[12];
+    char intValStr[INT32_STRING_MAX_LENGTH];
     if (mutation->record->schema->partitionColumn == -1) return HOLO_CLIENT_RET_OK;
     parentSchema = mutation->record->schema;
-    partitionValue = mutation->record->values[parentSchema->partitionColumn];
     if (mutation->record->valueFormats[parentSchema->partitionColumn] == 1) {
-        endian_swap(partitionValue, 4);
-        snprintf(intVal, 12, "%d", *(int32_t*)partitionValue);
-        endian_swap(partitionValue, 4);
-        partitionValue = intVal;
+        get_int_partition_value_string(intValStr, mutation->record->values[parentSchema->partitionColumn]);
+        partitionValue = intValStr;
+    } else {
+        partitionValue = mutation->record->values[parentSchema->partitionColumn];
     }
     partitionSchema = meta_cache_find_partition(client->workerPool->metaCache, parentSchema, partitionValue);
     if (partitionSchema != NULL) {

holo-client-c/src/logger_log4c.c

@@ -42,24 +42,24 @@ void* holo_client_log_log4c(const int logLevel, const char* msg) {
     switch (logLevel)
     {
     case HOLO_LOG_LEVEL_DEBUG:
-        log4c_category_log(log_category, LOG4C_PRIORITY_DEBUG, "%s", msg);
+        log4c_category_log(log_category, LOG4C_PRIORITY_DEBUG, "[%ld] %s", (unsigned long)pthread_self(), msg);
         break;
     case HOLO_LOG_LEVEL_INFO:
-        log4c_category_log(log_category, LOG4C_PRIORITY_INFO, "%s", msg);
+        log4c_category_log(log_category, LOG4C_PRIORITY_INFO, "[%ld] %s", (unsigned long)pthread_self(), msg);
         break;
     case HOLO_LOG_LEVEL_WARNING:
-        log4c_category_log(log_category, LOG4C_PRIORITY_WARN, "%s", msg);
+        log4c_category_log(log_category, LOG4C_PRIORITY_WARN, "[%ld] %s", (unsigned long)pthread_self(), msg);
         break;
     case HOLO_LOG_LEVEL_ERROR:
-        log4c_category_log(log_category, LOG4C_PRIORITY_ERROR, "%s", msg);
+        log4c_category_log(log_category, LOG4C_PRIORITY_ERROR, "[%ld] %s", (unsigned long)pthread_self(), msg);
         break;
     case HOLO_LOG_LEVEL_FATAL:
-        log4c_category_log(log_category, LOG4C_PRIORITY_FATAL, "%s", msg);
+        log4c_category_log(log_category, LOG4C_PRIORITY_FATAL, "[%ld] %s", (unsigned long)pthread_self(), msg);
         break;
     case HOLO_LOG_LEVEL_NONE:
         break;
     default:
-        log4c_category_log(log_category, LOG4C_PRIORITY_INFO, "%s", msg);
+        log4c_category_log(log_category, LOG4C_PRIORITY_INFO, "[%ld] %s", (unsigned long)pthread_self(), msg);
         break;
     }
     return NULL;

holo-client-c/src/meta_cache.c

@@ -27,6 +27,7 @@ MetaCache* holo_client_new_metacache() {
     MetaCache* cache = MALLOC(1, MetaCache);
     dlist_init(&(cache->schemaList));
     dlist_init(&(cache->parentList));
+    dlist_init(&(cache->garbageList));
     cache->rwlock = MALLOC(1, pthread_rwlock_t);
     pthread_rwlock_init(cache->rwlock, NULL);
     return cache;
@@ -56,6 +57,13 @@ void clear_all_contents(MetaCache* cache) {
         holo_client_destroy_tableschema(schema);
         FREE(schemaItem);
     }
+    dlist_foreach_modify(miter, &(cache->garbageList)) {
+        schemaItem = dlist_container(SchemaItem, list_node, miter.cur);
+        schema = schemaItem->schema;
+        dlist_delete(miter.cur);
+        holo_client_destroy_tableschema(schema);
+        FREE(schemaItem);
+    }
 }
 
 void holo_client_destroy_metacache(MetaCache* cache) {
@@ -82,21 +90,53 @@ HoloTableSchema* find_tableschema_in_metacache(MetaCache* cache, HoloTableName n
     return schema;
 }
 
+void add_tableschema_to_garbage_list(MetaCache* cache, HoloTableSchema* schema) {
+    LOG_DEBUG("add table schema:%s to garbage list", schema->tableName->fullName);
+    SchemaItem* schemaItem = create_schema_item(schema);
+    dlist_push_tail(&(cache->garbageList), &(schemaItem->list_node));
+}
+
+bool should_update_table_schema_in_cache(HoloTableSchema* origin, HoloTableSchema* current) {
+    if (strcmp(origin->tableName->fullName, current->tableName->fullName) != 0) {
+        return false;
+    }
+    if (origin->nColumns != current->nColumns || origin->nPrimaryKeys != current->nPrimaryKeys || origin->nDistributionKeys != current->nDistributionKeys) {
+        return true;
+    }
+    for (int i = 0; i < current->nColumns; i++) {
+        HoloColumn originColumn = origin->columns[i];
+        HoloColumn currentColumn = current->columns[i];
+        if (!compare_strings(originColumn.name, currentColumn.name)
+            || !compare_strings(originColumn.defaultValue, currentColumn.defaultValue)
+            || originColumn.type != currentColumn.type
+            || originColumn.isPrimaryKey != currentColumn.isPrimaryKey
+            || originColumn.nullable != currentColumn.nullable) {
+            return true;
+        }
+    }
+
+    return false;
+}
+
 void add_tableschema_to_metacache(MetaCache* cache, HoloTableSchema* schema) {
     dlist_iter	iter;
     SchemaItem* schemaItem;
     bool updated = false;
     pthread_rwlock_wrlock(cache->rwlock); 
     dlist_foreach(iter, &cache->schemaList) {
         schemaItem = dlist_container(SchemaItem, list_node, iter.cur);
-        if (strcmp(schema->tableName->fullName, schemaItem->schema->tableName->fullName) == 0) {
-            holo_client_destroy_tableschema(schemaItem->schema);
+        if (should_update_table_schema_in_cache(schemaItem->schema, schema)) {
+            // 这里不能直接destory，还可能有mutation用着这个schema
+            // c没有引用计数，干脆先加到garbageList，最后一起清理
+            add_tableschema_to_garbage_list(cache, schemaItem->schema);
             schemaItem->schema = schema;
+            LOG_DEBUG("update table schema:%s in meta cache", schema->tableName->fullName);
             updated = true;
             break;
         }
     }
     if(!updated) {
+        LOG_DEBUG("add table schema:%s to meta cache", schema->tableName->fullName);
         schemaItem = create_schema_item(schema);
         dlist_push_tail(&(cache->schemaList), &(schemaItem->list_node));
         if (schema->partitionColumn > -1) {

holo-client-c/src/meta_cache.h

@@ -9,6 +9,7 @@
 typedef struct _MetaCache {
     dlist_head schemaList;
     dlist_head parentList;
+    dlist_head garbageList;
     pthread_rwlock_t* rwlock;
 } MetaCache;
 
@@ -38,5 +39,6 @@ HoloTableSchema* find_tableschema_in_metacache(MetaCache*, HoloTableName);
 void add_tableschema_to_metacache(MetaCache*, HoloTableSchema*);
 HoloTableSchema* meta_cache_find_partition(MetaCache*, HoloTableSchema*, char*);
 void meta_cache_add_partition(MetaCache*, HoloTableSchema*, HoloTableSchema*, char*);
+void add_tableschema_to_garbage_list(MetaCache*, HoloTableSchema*);
 
 #endif

holo-client-c/src/table_schema.c

@@ -56,6 +56,7 @@ HoloTableSchema* holo_client_new_tableschema() {
 }
 
 void holo_client_destroy_tableschema(HoloTableSchema* schema) {
+    LOG_DEBUG("destory table schema:%s", schema->tableName->fullName);
     holo_client_destroy_tablename(schema->tableName);
     FREE(schema->tableName);
     holo_client_destroy_columns(schema->columns, schema->nColumns);

holo-client-c/test/Makefile

@@ -39,7 +39,7 @@ COBJS:=$(SRCS:.c=.o)
 all:$(BIN)
 
 $(COBJS) : %.o: %.c
-	$(CC) -c $< -o $@ $(INCS) $(VERSION_FLAG) -std=c11 -D_XOPEN_SOURCE=600 -Wall -Werror
+	$(CC) -c $< -o $@ $(INCS) $(VERSION_FLAG) -std=c11 -D_XOPEN_SOURCE=600 -Wall -Werror -O0 -g3
 
 $(BIN):$(COBJS)
 	$(CC) -o $(BIN) $(COBJS) $(STATCI_LIBS) $(SHARED_LIBS) $(CFLAGS)