提交 8fe7c183 编写于 作者: H Haojun Liao

Merge remote-tracking branch 'origin/3.0' into feature/3.0_liaohj

......@@ -124,8 +124,25 @@ typedef struct TAOS_MULTI_BIND {
int num;
} TAOS_MULTI_BIND;
typedef enum {
SET_CONF_RET_SUCC = 0,
SET_CONF_RET_ERR_PART = -1,
SET_CONF_RET_ERR_INNER = -2,
SET_CONF_RET_ERR_JSON_INVALID = -3,
SET_CONF_RET_ERR_JSON_PARSE = -4,
SET_CONF_RET_ERR_ONLY_ONCE = -5,
SET_CONF_RET_ERR_TOO_LONG = -6
} SET_CONF_RET_CODE;
#define RET_MSG_LENGTH 1024
typedef struct setConfRet {
SET_CONF_RET_CODE retCode;
char retMsg[RET_MSG_LENGTH];
} setConfRet;
DLL_EXPORT void taos_cleanup(void);
DLL_EXPORT int taos_options(TSDB_OPTION option, const void *arg, ...);
DLL_EXPORT setConfRet taos_set_config(const char *config);
DLL_EXPORT TAOS *taos_connect(const char *ip, const char *user, const char *pass, const char *db, uint16_t port);
DLL_EXPORT TAOS *taos_connect_l(const char *ip, int ipLen, const char *user, int userLen, const char *pass, int passLen,
const char *db, int dbLen, uint16_t port);
......@@ -168,10 +185,13 @@ DLL_EXPORT int taos_select_db(TAOS *taos, const char *db);
DLL_EXPORT int taos_print_row(char *str, TAOS_ROW row, TAOS_FIELD *fields, int num_fields);
DLL_EXPORT void taos_stop_query(TAOS_RES *res);
DLL_EXPORT bool taos_is_null(TAOS_RES *res, int32_t row, int32_t col);
DLL_EXPORT bool taos_is_update_query(TAOS_RES *res);
DLL_EXPORT int taos_fetch_block(TAOS_RES *res, TAOS_ROW *rows);
DLL_EXPORT int taos_validate_sql(TAOS *taos, const char *sql);
DLL_EXPORT void taos_reset_current_db(TAOS *taos);
DLL_EXPORT int *taos_fetch_lengths(TAOS_RES *res);
DLL_EXPORT TAOS_ROW *taos_result_block(TAOS_RES *res);
DLL_EXPORT const char *taos_get_server_info(TAOS *taos);
DLL_EXPORT const char *taos_get_client_info();
......
......@@ -51,7 +51,7 @@ extern int32_t tsCompatibleModel;
extern bool tsEnableSlaveQuery;
extern bool tsPrintAuth;
extern int64_t tsTickPerDay[3];
extern int32_t tsMultiProcess;
extern bool tsMultiProcess;
// monitor
extern bool tsEnableMonitor;
......
......@@ -2016,7 +2016,6 @@ typedef struct {
int8_t slidingUnit; // MACRO: TIME_UNIT_XXX
int8_t timezoneInt; // sma data expired if timezone changes.
char indexName[TSDB_INDEX_NAME_LEN];
char timezone[TD_TIMEZONE_LEN];
int32_t exprLen;
int32_t tagsFilterLen;
int64_t indexUid;
......@@ -2054,32 +2053,6 @@ void* tDeserializeSVCreateTSmaReq(void* buf, SVCreateTSmaReq* pReq);
int32_t tSerializeSVDropTSmaReq(void** buf, SVDropTSmaReq* pReq);
void* tDeserializeSVDropTSmaReq(void* buf, SVDropTSmaReq* pReq);
typedef struct {
col_id_t colId;
uint16_t blockSize; // sma data block size
char data[];
} STSmaColData;
typedef struct {
tb_uid_t tableUid; // super/child/normal table uid
int32_t dataLen; // not including head
char data[];
} STSmaTbData;
typedef struct {
int64_t indexUid;
TSKEY skey; // startKey of one interval/sliding window
int64_t interval;
int32_t dataLen; // not including head
int8_t intervalUnit;
char data[];
} STSmaDataWrapper; // sma data for a interval/sliding window
// interval/sliding => window
// => window->table->colId
// => 当一个window下所有的表均计算完成时,流计算告知tsdb清除window的过期标记
// RSma: Rollup SMA
typedef struct {
int64_t interval;
......
......@@ -138,55 +138,59 @@
#define TK_INTERVAL 120
#define TK_TOPIC 121
#define TK_AS 122
#define TK_NK_BOOL 123
#define TK_NULL 124
#define TK_NK_VARIABLE 125
#define TK_NK_UNDERLINE 126
#define TK_ROWTS 127
#define TK_TBNAME 128
#define TK_QSTARTTS 129
#define TK_QENDTS 130
#define TK_WSTARTTS 131
#define TK_WENDTS 132
#define TK_WDURATION 133
#define TK_BETWEEN 134
#define TK_IS 135
#define TK_NK_LT 136
#define TK_NK_GT 137
#define TK_NK_LE 138
#define TK_NK_GE 139
#define TK_NK_NE 140
#define TK_MATCH 141
#define TK_NMATCH 142
#define TK_IN 143
#define TK_JOIN 144
#define TK_INNER 145
#define TK_SELECT 146
#define TK_DISTINCT 147
#define TK_WHERE 148
#define TK_PARTITION 149
#define TK_BY 150
#define TK_SESSION 151
#define TK_STATE_WINDOW 152
#define TK_SLIDING 153
#define TK_FILL 154
#define TK_VALUE 155
#define TK_NONE 156
#define TK_PREV 157
#define TK_LINEAR 158
#define TK_NEXT 159
#define TK_GROUP 160
#define TK_HAVING 161
#define TK_ORDER 162
#define TK_SLIMIT 163
#define TK_SOFFSET 164
#define TK_LIMIT 165
#define TK_OFFSET 166
#define TK_ASC 167
#define TK_DESC 168
#define TK_NULLS 169
#define TK_FIRST 170
#define TK_LAST 171
#define TK_EXPLAIN 123
#define TK_ANALYZE 124
#define TK_VERBOSE 125
#define TK_NK_BOOL 126
#define TK_RATIO 127
#define TK_NULL 128
#define TK_NK_VARIABLE 129
#define TK_NK_UNDERLINE 130
#define TK_ROWTS 131
#define TK_TBNAME 132
#define TK_QSTARTTS 133
#define TK_QENDTS 134
#define TK_WSTARTTS 135
#define TK_WENDTS 136
#define TK_WDURATION 137
#define TK_BETWEEN 138
#define TK_IS 139
#define TK_NK_LT 140
#define TK_NK_GT 141
#define TK_NK_LE 142
#define TK_NK_GE 143
#define TK_NK_NE 144
#define TK_MATCH 145
#define TK_NMATCH 146
#define TK_IN 147
#define TK_JOIN 148
#define TK_INNER 149
#define TK_SELECT 150
#define TK_DISTINCT 151
#define TK_WHERE 152
#define TK_PARTITION 153
#define TK_BY 154
#define TK_SESSION 155
#define TK_STATE_WINDOW 156
#define TK_SLIDING 157
#define TK_FILL 158
#define TK_VALUE 159
#define TK_NONE 160
#define TK_PREV 161
#define TK_LINEAR 162
#define TK_NEXT 163
#define TK_GROUP 164
#define TK_HAVING 165
#define TK_ORDER 166
#define TK_SLIMIT 167
#define TK_SOFFSET 168
#define TK_LIMIT 169
#define TK_OFFSET 170
#define TK_ASC 171
#define TK_DESC 172
#define TK_NULLS 173
#define TK_FIRST 174
#define TK_LAST 175
#define TK_NK_SPACE 300
#define TK_NK_COMMENT 301
......
......@@ -46,11 +46,12 @@ typedef struct {
char localFqdn[TSDB_FQDN_LEN];
char firstEp[TSDB_EP_LEN];
char secondEp[TSDB_EP_LEN];
SDiskCfg *pDisks;
SDiskCfg *disks;
int32_t numOfDisks;
int8_t ntype;
} SDnodeOpt;
typedef enum { DND_EVENT_START, DND_EVENT_STOP = 1, DND_EVENT_RELOAD } EDndEvent;
typedef enum { DND_EVENT_START, DND_EVENT_STOP = 1, DND_EVENT_CHILD } EDndEvent;
/**
* @brief Initialize and start the dnode.
......
......@@ -70,6 +70,7 @@ typedef enum ENodeType {
QUERY_NODE_DATABASE_OPTIONS,
QUERY_NODE_TABLE_OPTIONS,
QUERY_NODE_INDEX_OPTIONS,
QUERY_NODE_EXPLAIN_OPTIONS,
// Statement nodes are used in parser and planner module.
QUERY_NODE_SET_OPERATOR,
......@@ -99,6 +100,7 @@ typedef enum ENodeType {
QUERY_NODE_CREATE_TOPIC_STMT,
QUERY_NODE_DROP_TOPIC_STMT,
QUERY_NODE_ALTER_LOCAL_STMT,
QUERY_NODE_EXPLAIN_STMT,
QUERY_NODE_SHOW_DATABASES_STMT,
QUERY_NODE_SHOW_TABLES_STMT,
QUERY_NODE_SHOW_STABLES_STMT,
......
......@@ -49,6 +49,7 @@ typedef struct SScanLogicNode {
STimeWindow scanRange;
SName tableName;
bool showRewrite;
double ratio;
} SScanLogicNode;
typedef struct SJoinLogicNode {
......@@ -197,7 +198,7 @@ typedef struct STableScanPhysiNode {
SScanPhysiNode scan;
uint8_t scanFlag; // denotes reversed scan of data or not
STimeWindow scanRange;
SNode* pScanConditions;
double ratio;
} STableScanPhysiNode;
typedef STableScanPhysiNode STableSeqScanPhysiNode;
......@@ -252,6 +253,7 @@ typedef struct SIntervalPhysiNode {
int64_t sliding;
int8_t intervalUnit;
int8_t slidingUnit;
uint8_t precision;
SFillNode* pFill;
} SIntervalPhysiNode;
......@@ -297,18 +299,23 @@ typedef struct SSubplan {
SDataSinkNode* pDataSink; // data of the subplan flow into the datasink
} SSubplan;
typedef enum EQueryMode {
QUERY_MODE_NORMAL = 1,
QUERY_MODE_EXPLAIN,
QUERY_MODE_EXPLAIN_AN
} EQueryMode;
typedef enum EExplainMode {
EXPLAIN_MODE_DISABLE = 1,
EXPLAIN_MODE_STATIC,
EXPLAIN_MODE_ANALYZE
} EExplainMode;
typedef struct SExplainInfo {
EExplainMode mode;
bool verbose;
} SExplainInfo;
typedef struct SQueryPlan {
ENodeType type;
uint64_t queryId;
int32_t numOfSubplans;
SNodeList* pSubplans; // Element is SNodeListNode. The execution level of subplan, starting from 0.
SExplainInfo explainInfo;
} SQueryPlan;
#ifdef __cplusplus
......
......@@ -131,6 +131,7 @@ typedef struct SRealTableNode {
struct STableMeta* pMeta;
SVgroupsInfo* pVgroupList;
char useDbName[TSDB_DB_NAME_LEN];
double ratio;
} SRealTableNode;
typedef struct STempTableNode {
......@@ -282,6 +283,19 @@ typedef struct SVnodeModifOpStmt {
const char* sql; // current sql statement position
} SVnodeModifOpStmt;
typedef struct SExplainOptions {
ENodeType type;
bool verbose;
double ratio;
} SExplainOptions;
typedef struct SExplainStmt {
ENodeType type;
bool analyze;
SExplainOptions* pOptions;
SNode* pQuery;
} SExplainStmt;
void nodesWalkSelectStmt(SSelectStmt* pSelect, ESqlClause clause, FNodeWalker walker, void* pContext);
void nodesRewriteSelectStmt(SSelectStmt* pSelect, ESqlClause clause, FNodeRewriter rewriter, void* pContext);
......
......@@ -42,6 +42,9 @@ extern "C" {
#include <sys/types.h>
#include <termios.h>
#include <sys/statvfs.h>
#include <sys/prctl.h>
#include <sys/shm.h>
#include <sys/wait.h>
#if defined(DARWIN)
#else
......@@ -70,7 +73,6 @@ extern "C" {
#include <wchar.h>
#include <wctype.h>
#include "osAtomic.h"
#include "osDef.h"
#include "osDir.h"
......@@ -84,6 +86,7 @@ extern "C" {
#include "osThread.h"
#include "osSemaphore.h"
#include "osSignal.h"
#include "osShm.h"
#include "osSleep.h"
#include "osSocket.h"
#include "osString.h"
......
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_OS_SHM_H_
#define _TD_OS_SHM_H_
#ifdef __cplusplus
extern "C" {
#endif
typedef struct {
int32_t id;
int32_t size;
void* ptr;
} SShm;
int32_t taosCreateShm(SShm *pShm, int32_t shmsize) ;
void taosDropShm(SShm *pShm);
int32_t taosAttachShm(SShm *pShm);
void taosDetachShm(SShm *pShm);
#ifdef __cplusplus
}
#endif
#endif /*_TD_OS_SHM_H_*/
......@@ -32,6 +32,9 @@ typedef pthread_once_t TdThreadOnce;
typedef pthread_rwlockattr_t TdThreadRwlockAttr;
typedef pthread_cond_t TdThreadCond;
typedef pthread_condattr_t TdThreadCondAttr;
typedef pthread_key_t TdThreadKey;
typedef pthread_barrier_t TdThreadBarrier;
typedef pthread_barrierattr_t TdThreadBarrierAttr;
#define taosThreadCleanupPush pthread_cleanup_push
#define taosThreadCleanupPop pthread_cleanup_pop
......@@ -39,78 +42,190 @@ typedef pthread_condattr_t TdThreadCondAttr;
// If the error is in a third-party library, place this header file under the third-party library header file.
// When you want to use this feature, you should find or add the same function in the following section.
#ifndef ALLOW_FORBID_FUNC
#define pthread_t PTHREAD_T_TYPE_TAOS_FORBID
#define pthread_spinlock_t PTHREAD_SPINLOCK_T_TYPE_TAOS_FORBID
#define pthread_mutex_t PTHREAD_MUTEX_T_TYPE_TAOS_FORBID
#define pthread_mutexattr_t PTHREAD_MUTEXATTR_T_TYPE_TAOS_FORBID
#define pthread_rwlock_t PTHREAD_RWLOCK_T_TYPE_TAOS_FORBID
#define pthread_attr_t PTHREAD_ATTR_T_TYPE_TAOS_FORBID
#define pthread_once_t PTHREAD_ONCE_T_TYPE_TAOS_FORBID
#define pthread_rwlockattr_t PTHREAD_RWLOCKATTR_T_TYPE_TAOS_FORBID
#define pthread_cond_t PTHREAD_COND_T_TYPE_TAOS_FORBID
#define pthread_condattr_t PTHREAD_CONDATTR_T_TYPE_TAOS_FORBID
#define pthread_spin_init PTHREAD_SPIN_INIT_FUNC_TAOS_FORBID
#define pthread_mutex_init PTHREAD_MUTEX_INIT_FUNC_TAOS_FORBID
#define pthread_spin_destroy PTHREAD_SPIN_DESTROY_FUNC_TAOS_FORBID
#define pthread_mutex_destroy PTHREAD_MUTEX_DESTROY_FUNC_TAOS_FORBID
#define pthread_spin_lock PTHREAD_SPIN_LOCK_FUNC_TAOS_FORBID
#define pthread_mutex_lock PTHREAD_MUTEX_LOCK_FUNC_TAOS_FORBID
#define pthread_spin_unlock PTHREAD_SPIN_UNLOCK_FUNC_TAOS_FORBID
#define pthread_mutex_unlock PTHREAD_MUTEX_UNLOCK_FUNC_TAOS_FORBID
#define pthread_rwlock_rdlock PTHREAD_RWLOCK_RDLOCK_FUNC_TAOS_FORBID
#define pthread_rwlock_wrlock PTHREAD_RWLOCK_WRLOCK_FUNC_TAOS_FORBID
#define pthread_rwlock_unlock PTHREAD_RWLOCK_UNLOCK_FUNC_TAOS_FORBID
#define pthread_testcancel PTHREAD_TESTCANCEL_FUNC_TAOS_FORBID
#define pthread_attr_init PTHREAD_ATTR_INIT_FUNC_TAOS_FORBID
#define pthread_create PTHREAD_CREATE_FUNC_TAOS_FORBID
#define pthread_once PTHREAD_ONCE_FUNC_TAOS_FORBID
#define pthread_attr_setdetachstate PTHREAD_ATTR_SETDETACHSTATE_FUNC_TAOS_FORBID
#define pthread_attr_destroy PTHREAD_ATTR_DESTROY_FUNC_TAOS_FORBID
#define pthread_join PTHREAD_JOIN_FUNC_TAOS_FORBID
#define pthread_rwlock_init PTHREAD_RWLOCK_INIT_FUNC_TAOS_FORBID
#define pthread_rwlock_destroy PTHREAD_RWLOCK_DESTROY_FUNC_TAOS_FORBID
#define pthread_cond_signal PTHREAD_COND_SIGNAL_FUNC_TAOS_FORBID
#define pthread_cond_init PTHREAD_COND_INIT_FUNC_TAOS_FORBID
#define pthread_cond_broadcast PTHREAD_COND_BROADCAST_FUNC_TAOS_FORBID
#define pthread_cond_destroy PTHREAD_COND_DESTROY_FUNC_TAOS_FORBID
#define pthread_cond_wait PTHREAD_COND_WAIT_FUNC_TAOS_FORBID
#define pthread_self PTHREAD_SELF_FUNC_TAOS_FORBID
#define pthread_equal PTHREAD_EQUAL_FUNC_TAOS_FORBID
#define pthread_sigmask PTHREAD_SIGMASK_FUNC_TAOS_FORBID
#define pthread_cancel PTHREAD_CANCEL_FUNC_TAOS_FORBID
#define pthread_kill PTHREAD_KILL_FUNC_TAOS_FORBID
// #define pthread_t PTHREAD_T_TYPE_TAOS_FORBID
// #define pthread_spinlock_t PTHREAD_SPINLOCK_T_TYPE_TAOS_FORBID
// #define pthread_mutex_t PTHREAD_MUTEX_T_TYPE_TAOS_FORBID
// #define pthread_mutexattr_t PTHREAD_MUTEXATTR_T_TYPE_TAOS_FORBID
// #define pthread_rwlock_t PTHREAD_RWLOCK_T_TYPE_TAOS_FORBID
// #define pthread_attr_t PTHREAD_ATTR_T_TYPE_TAOS_FORBID
// #define pthread_once_t PTHREAD_ONCE_T_TYPE_TAOS_FORBID
// #define pthread_rwlockattr_t PTHREAD_RWLOCKATTR_T_TYPE_TAOS_FORBID
// #define pthread_cond_t PTHREAD_COND_T_TYPE_TAOS_FORBID
// #define pthread_condattr_t PTHREAD_CONDATTR_T_TYPE_TAOS_FORBID
// #define pthread_key_t PTHREAD_KEY_T_TYPE_TAOS_FORBID
// #define pthread_barrier_t PTHREAD_BARRIER_T_TYPE_TAOS_FORBID
// #define pthread_barrierattr_t PTHREAD_BARRIERATTR_T_TYPE_TAOS_FORBID
// #define pthread_create PTHREAD_CREATE_FUNC_TAOS_FORBID
// #define pthread_attr_destroy PTHREAD_ATTR_DESTROY_FUNC_TAOS_FORBID
// #define pthread_attr_getdetachstate PTHREAD_ATTR_GETDETACHSTATE_FUNC_TAOS_FORBID
// #define pthread_attr_getinheritsched PTHREAD_ATTR_GETINHERITSCHED_FUNC_TAOS_FORBID
// #define pthread_attr_getschedparam PTHREAD_ATTR_GETSCHEDPARAM_FUNC_TAOS_FORBID
// #define pthread_attr_getschedpolicy PTHREAD_ATTR_GETSCHEDPOLICY_FUNC_TAOS_FORBID
// #define pthread_attr_getscope PTHREAD_ATTR_GETSCOPE_FUNC_TAOS_FORBID
// #define pthread_attr_getstacksize PTHREAD_ATTR_GETSTACKSIZE_FUNC_TAOS_FORBID
// #define pthread_attr_init PTHREAD_ATTR_INIT_FUNC_TAOS_FORBID
// #define pthread_attr_setdetachstate PTHREAD_ATTR_SETDETACHSTATE_FUNC_TAOS_FORBID
// #define pthread_attr_setinheritsched PTHREAD_ATTR_SETINHERITSCHED_FUNC_TAOS_FORBID
// #define pthread_attr_setschedparam PTHREAD_ATTR_SETSCHEDPARAM_FUNC_TAOS_FORBID
// #define pthread_attr_setschedpolicy PTHREAD_ATTR_SETSCHEDPOLICY_FUNC_TAOS_FORBID
// #define pthread_attr_setscope PTHREAD_ATTR_SETSCOPE_FUNC_TAOS_FORBID
// #define pthread_attr_setstacksize PTHREAD_ATTR_SETSTACKSIZE_FUNC_TAOS_FORBID
// #define pthread_barrier_destroy PTHREAD_BARRIER_DESTROY_FUNC_TAOS_FORBID
// #define pthread_barrier_init PTHREAD_BARRIER_INIT_FUNC_TAOS_FORBID
// #define pthread_barrier_wait PTHREAD_BARRIER_WAIT_FUNC_TAOS_FORBID
// #define pthread_barrierattr_destroy PTHREAD_BARRIERATTR_DESTROY_FUNC_TAOS_FORBID
// #define pthread_barrierattr_getpshared PTHREAD_BARRIERATTR_GETPSHARED_FUNC_TAOS_FORBID
// #define pthread_barrierattr_init PTHREAD_BARRIERATTR_INIT_FUNC_TAOS_FORBID
// #define pthread_barrierattr_setpshared PTHREAD_BARRIERATTR_SETPSHARED_FUNC_TAOS_FORBID
// #define pthread_cancel PTHREAD_CANCEL_FUNC_TAOS_FORBID
// #define pthread_cond_destroy PTHREAD_COND_DESTROY_FUNC_TAOS_FORBID
// #define pthread_cond_init PTHREAD_COND_INIT_FUNC_TAOS_FORBID
// #define pthread_cond_signal PTHREAD_COND_SIGNAL_FUNC_TAOS_FORBID
// #define pthread_cond_broadcast PTHREAD_COND_BROADCAST_FUNC_TAOS_FORBID
// #define pthread_cond_wait PTHREAD_COND_WAIT_FUNC_TAOS_FORBID
// #define pthread_cond_timedwait PTHREAD_COND_TIMEDWAIT_FUNC_TAOS_FORBID
// #define pthread_condattr_destroy PTHREAD_CONDATTR_DESTROY_FUNC_TAOS_FORBID
// #define pthread_condattr_getpshared PTHREAD_CONDATTR_GETPSHARED_FUNC_TAOS_FORBID
// #define pthread_condattr_init PTHREAD_CONDATTR_INIT_FUNC_TAOS_FORBID
// #define pthread_condattr_setpshared PTHREAD_CONDATTR_SETPSHARED_FUNC_TAOS_FORBID
// #define pthread_detach PTHREAD_DETACH_FUNC_TAOS_FORBID
// #define pthread_equal PTHREAD_EQUAL_FUNC_TAOS_FORBID
// #define pthread_exit PTHREAD_EXIT_FUNC_TAOS_FORBID
// #define pthread_getschedparam PTHREAD_GETSCHEDPARAM_FUNC_TAOS_FORBID
// #define pthread_getspecific PTHREAD_GETSPECIFIC_FUNC_TAOS_FORBID
// #define pthread_join PTHREAD_JOIN_FUNC_TAOS_FORBID
// #define pthread_key_create PTHREAD_KEY_CREATE_FUNC_TAOS_FORBID
// #define pthread_key_delete PTHREAD_KEY_DELETE_FUNC_TAOS_FORBID
// #define pthread_kill PTHREAD_KILL_FUNC_TAOS_FORBID
// #define pthread_mutex_consistent PTHREAD_MUTEX_CONSISTENT_FUNC_TAOS_FORBID
// #define pthread_mutex_destroy PTHREAD_MUTEX_DESTROY_FUNC_TAOS_FORBID
// #define pthread_mutex_init PTHREAD_MUTEX_INIT_FUNC_TAOS_FORBID
// #define pthread_mutex_lock PTHREAD_MUTEX_LOCK_FUNC_TAOS_FORBID
// #define pthread_mutex_timedlock PTHREAD_MUTEX_TIMEDLOCK_FUNC_TAOS_FORBID
// #define pthread_mutex_trylock PTHREAD_MUTEX_TRYLOCK_FUNC_TAOS_FORBID
// #define pthread_mutex_unlock PTHREAD_MUTEX_UNLOCK_FUNC_TAOS_FORBID
// #define pthread_mutexattr_destroy PTHREAD_MUTEXATTR_DESTROY_FUNC_TAOS_FORBID
// #define pthread_mutexattr_getpshared PTHREAD_MUTEXATTR_GETPSHARED_FUNC_TAOS_FORBID
// #define pthread_mutexattr_getrobust PTHREAD_MUTEXATTR_GETROBUST_FUNC_TAOS_FORBID
// #define pthread_mutexattr_gettype PTHREAD_MUTEXATTR_GETTYPE_FUNC_TAOS_FORBID
// #define pthread_mutexattr_init PTHREAD_MUTEXATTR_INIT_FUNC_TAOS_FORBID
// #define pthread_mutexattr_setpshared PTHREAD_MUTEXATTR_SETPSHARED_FUNC_TAOS_FORBID
// #define pthread_mutexattr_setrobust PTHREAD_MUTEXATTR_SETROBUST_FUNC_TAOS_FORBID
// #define pthread_mutexattr_settype PTHREAD_MUTEXATTR_SETTYPE_FUNC_TAOS_FORBID
// #define pthread_once PTHREAD_ONCE_FUNC_TAOS_FORBID
// #define pthread_rwlock_destroy PTHREAD_RWLOCK_DESTROY_FUNC_TAOS_FORBID
// #define pthread_rwlock_init PTHREAD_RWLOCK_INIT_FUNC_TAOS_FORBID
// #define pthread_rwlock_rdlock PTHREAD_RWLOCK_RDLOCK_FUNC_TAOS_FORBID
// #define pthread_rwlock_timedrdlock PTHREAD_RWLOCK_TIMEDRDLOCK_FUNC_TAOS_FORBID
// #define pthread_rwlock_timedwrlock PTHREAD_RWLOCK_TIMEDWRLOCK_FUNC_TAOS_FORBID
// #define pthread_rwlock_tryrdlock PTHREAD_RWLOCK_TRYRDLOCK_FUNC_TAOS_FORBID
// #define pthread_rwlock_trywrlock PTHREAD_RWLOCK_TRYWRLOCK_FUNC_TAOS_FORBID
// #define pthread_rwlock_unlock PTHREAD_RWLOCK_UNLOCK_FUNC_TAOS_FORBID
// #define pthread_rwlock_wrlock PTHREAD_RWLOCK_WRLOCK_FUNC_TAOS_FORBID
// #define pthread_rwlockattr_destroy PTHREAD_RWLOCKATTR_DESTROY_FUNC_TAOS_FORBID
// #define pthread_rwlockattr_getpshared PTHREAD_RWLOCKATTR_GETPSHARED_FUNC_TAOS_FORBID
// #define pthread_rwlockattr_init PTHREAD_RWLOCKATTR_INIT_FUNC_TAOS_FORBID
// #define pthread_rwlockattr_setpshared PTHREAD_RWLOCKATTR_SETPSHARED_FUNC_TAOS_FORBID
// #define pthread_self PTHREAD_SELF_FUNC_TAOS_FORBID
// #define pthread_setcancelstate PTHREAD_SETCANCELSTATE_FUNC_TAOS_FORBID
// #define pthread_setcanceltype PTHREAD_SETCANCELTYPE_FUNC_TAOS_FORBID
// #define pthread_setschedparam PTHREAD_SETSCHEDPARAM_FUNC_TAOS_FORBID
// #define pthread_setspecific PTHREAD_SETSPECIFIC_FUNC_TAOS_FORBID
// #define pthread_spin_destroy PTHREAD_SPIN_DESTROY_FUNC_TAOS_FORBID
// #define pthread_spin_init PTHREAD_SPIN_INIT_FUNC_TAOS_FORBID
// #define pthread_spin_lock PTHREAD_SPIN_LOCK_FUNC_TAOS_FORBID
// #define pthread_spin_trylock PTHREAD_SPIN_TRYLOCK_FUNC_TAOS_FORBID
// #define pthread_spin_unlock PTHREAD_SPIN_UNLOCK_FUNC_TAOS_FORBID
// #define pthread_testcancel PTHREAD_TESTCANCEL_FUNC_TAOS_FORBID
// #define pthread_sigmask PTHREAD_SIGMASK_FUNC_TAOS_FORBID
// #define sigwait SIGWAIT_FUNC_TAOS_FORBID
#endif
int32_t taosThreadSpinInit(TdThreadSpinlock *lock, int pshared);
int32_t taosThreadMutexInit(TdThreadMutex *mutex, const TdThreadMutexAttr *attr);
int32_t taosThreadSpinDestroy(TdThreadSpinlock *lock);
int32_t taosThreadCreate(TdThread * tid, const TdThreadAttr * attr, void *(*start)(void *), void *arg);
int32_t taosThreadAttrDestroy(TdThreadAttr * attr);
int32_t taosThreadAttrGetDetachState(const TdThreadAttr * attr, int32_t *detachstate);
int32_t taosThreadAttrGetInheritSched(const TdThreadAttr * attr, int32_t *inheritsched);
int32_t taosThreadAttrGetSchedParam(const TdThreadAttr * attr, struct sched_param *param);
int32_t taosThreadAttrGetSchedPolicy(const TdThreadAttr * attr, int32_t *policy);
int32_t taosThreadAttrGetScope(const TdThreadAttr * attr, int32_t *contentionscope);
int32_t taosThreadAttrGetStackSize(const TdThreadAttr * attr, size_t * stacksize);
int32_t taosThreadAttrInit(TdThreadAttr * attr);
int32_t taosThreadAttrSetDetachState(TdThreadAttr * attr, int32_t detachstate);
int32_t taosThreadAttrSetInheritSched(TdThreadAttr * attr, int32_t inheritsched);
int32_t taosThreadAttrSetSchedParam(TdThreadAttr * attr, const struct sched_param *param);
int32_t taosThreadAttrSetSchedPolicy(TdThreadAttr * attr, int32_t policy);
int32_t taosThreadAttrSetScope(TdThreadAttr * attr, int32_t contentionscope);
int32_t taosThreadAttrSetStackSize(TdThreadAttr * attr, size_t stacksize);
int32_t taosThreadBarrierDestroy(TdThreadBarrier * barrier);
int32_t taosThreadBarrierInit(TdThreadBarrier * barrier, const TdThreadBarrierAttr * attr, uint32_t count);
int32_t taosThreadBarrierWait(TdThreadBarrier * barrier);
int32_t taosThreadBarrierAttrDestroy(TdThreadBarrierAttr * attr);
int32_t taosThreadBarrierAttrGetPshared(const TdThreadBarrierAttr * attr, int32_t *pshared);
int32_t taosThreadBarrierAttrInit(TdThreadBarrierAttr * attr);
int32_t taosThreadBarrierAttrSetPshared(TdThreadBarrierAttr * attr, int32_t pshared);
int32_t taosThreadCancel(TdThread thread);
int32_t taosThreadCondDestroy(TdThreadCond * cond);
int32_t taosThreadCondInit(TdThreadCond * cond, const TdThreadCondAttr * attr);
int32_t taosThreadCondSignal(TdThreadCond * cond);
int32_t taosThreadCondBroadcast(TdThreadCond * cond);
int32_t taosThreadCondWait(TdThreadCond * cond, TdThreadMutex * mutex);
int32_t taosThreadCondTimedWait(TdThreadCond * cond, TdThreadMutex * mutex, const struct timespec *abstime);
int32_t taosThreadCondAttrDestroy(TdThreadCondAttr * attr);
int32_t taosThreadCondAttrGetPshared(const TdThreadCondAttr * attr, int32_t *pshared);
int32_t taosThreadCondAttrInit(TdThreadCondAttr * attr);
int32_t taosThreadCondAttrSetPshared(TdThreadCondAttr * attr, int32_t pshared);
int32_t taosThreadDetach(TdThread thread);
int32_t taosThreadEqual(TdThread t1, TdThread t2);
void taosThreadExit(void *valuePtr);
int32_t taosThreadGetSchedParam(TdThread thread, int32_t *policy, struct sched_param *param);
void *taosThreadGetSpecific(TdThreadKey key);
int32_t taosThreadJoin(TdThread thread, void **valuePtr);
int32_t taosThreadKeyCreate(TdThreadKey * key, void(*destructor)(void *));
int32_t taosThreadKeyDelete(TdThreadKey key);
int32_t taosThreadKill(TdThread thread, int32_t sig);
int32_t taosThreadMutexConsistent(TdThreadMutex* mutex);
int32_t taosThreadMutexDestroy(TdThreadMutex * mutex);
int32_t taosThreadSpinLock(TdThreadSpinlock *lock);
int32_t taosThreadMutexLock(TdThreadMutex *mutex);
int32_t taosThreadRwlockRdlock(TdThreadRwlock *rwlock);
int32_t taosThreadSpinUnlock(TdThreadSpinlock *lock);
int32_t taosThreadMutexUnlock(TdThreadMutex *mutex);
int32_t taosThreadRwlockWrlock(TdThreadRwlock *rwlock);
int32_t taosThreadRwlockUnlock(TdThreadRwlock *rwlock);
void taosThreadTestCancel(void);
int32_t taosThreadAttrInit(TdThreadAttr *attr);
int32_t taosThreadCreate(TdThread *tid, const TdThreadAttr *attr, void*(*start)(void*), void *arg);
int32_t taosThreadOnce(TdThreadOnce *onceControl, void(*initRoutine)(void));
int32_t taosThreadAttrSetDetachState(TdThreadAttr *attr, int32_t detachState);
int32_t taosThreadAttrDestroy(TdThreadAttr *attr);
int32_t taosThreadJoin(TdThread thread, void **pValue);
int32_t taosThreadRwlockInit(TdThreadRwlock *rwlock, const TdThreadRwlockAttr *attr);
int32_t taosThreadRwlockDestroy(TdThreadRwlock *rwlock);
int32_t taosThreadCondSignal(TdThreadCond *cond);
int32_t taosThreadCondInit(TdThreadCond *cond, const TdThreadCondAttr *attr);
int32_t taosThreadCondBroadcast(TdThreadCond *cond);
int32_t taosThreadCondDestroy(TdThreadCond *cond);
int32_t taosThreadCondWait(TdThreadCond *cond, TdThreadMutex *mutex);
int32_t taosThreadMutexInit(TdThreadMutex * mutex, const TdThreadMutexAttr * attr);
int32_t taosThreadMutexLock(TdThreadMutex * mutex);
int32_t taosThreadMutexTimedLock(TdThreadMutex * mutex, const struct timespec *abstime);
int32_t taosThreadMutexTryLock(TdThreadMutex * mutex);
int32_t taosThreadMutexUnlock(TdThreadMutex * mutex);
int32_t taosThreadMutexAttrDestroy(TdThreadMutexAttr * attr);
int32_t taosThreadMutexAttrGetPshared(const TdThreadMutexAttr * attr, int32_t *pshared);
int32_t taosThreadMutexAttrGetRobust(const TdThreadMutexAttr * attr, int32_t * robust);
int32_t taosThreadMutexAttrGetType(const TdThreadMutexAttr * attr, int32_t *kind);
int32_t taosThreadMutexAttrInit(TdThreadMutexAttr * attr);
int32_t taosThreadMutexAttrSetPshared(TdThreadMutexAttr * attr, int32_t pshared);
int32_t taosThreadMutexAttrSetRobust(TdThreadMutexAttr * attr, int32_t robust);
int32_t taosThreadMutexAttrSetType(TdThreadMutexAttr * attr, int32_t kind);
int32_t taosThreadOnce(TdThreadOnce * onceControl, void(*initRoutine)(void));
int32_t taosThreadRwlockDestroy(TdThreadRwlock * rwlock);
int32_t taosThreadRwlockInit(TdThreadRwlock * rwlock, const TdThreadRwlockAttr * attr);
int32_t taosThreadRwlockRdlock(TdThreadRwlock * rwlock);
int32_t taosThreadRwlockTimedRdlock(TdThreadRwlock * rwlock, const struct timespec *abstime);
int32_t taosThreadRwlockTimedWrlock(TdThreadRwlock * rwlock, const struct timespec *abstime);
int32_t taosThreadRwlockTryRdlock(TdThreadRwlock * rwlock);
int32_t taosThreadRwlockTryWrlock(TdThreadRwlock * rwlock);
int32_t taosThreadRwlockUnlock(TdThreadRwlock * rwlock);
int32_t taosThreadRwlockWrlock(TdThreadRwlock * rwlock);
int32_t taosThreadRwlockAttrDestroy(TdThreadRwlockAttr * attr);
int32_t taosThreadRwlockAttrGetPshared(const TdThreadRwlockAttr * attr, int32_t *pshared);
int32_t taosThreadRwlockAttrInit(TdThreadRwlockAttr * attr);
int32_t taosThreadRwlockAttrSetPshared(TdThreadRwlockAttr * attr, int32_t pshared);
TdThread taosThreadSelf(void);
int32_t taosThreadEqual(TdThread t1, TdThread t2);
int32_t taosThreadSigmask(int how, sigset_t const *set, sigset_t *oset);
int32_t taosThreadCancel(TdThread thread);
int32_t taosThreadKill(TdThread thread, int sig);
int32_t taosThreadSetCancelState(int32_t state, int32_t *oldstate);
int32_t taosThreadSetCancelType(int32_t type, int32_t *oldtype);
int32_t taosThreadSetSchedParam(TdThread thread, int32_t policy, const struct sched_param *param);
int32_t taosThreadSetSpecific(TdThreadKey key, const void *value);
int32_t taosThreadSpinDestroy(TdThreadSpinlock * lock);
int32_t taosThreadSpinInit(TdThreadSpinlock * lock, int32_t pshared);
int32_t taosThreadSpinLock(TdThreadSpinlock * lock);
int32_t taosThreadSpinTrylock(TdThreadSpinlock * lock);
int32_t taosThreadSpinUnlock(TdThreadSpinlock * lock);
void taosThreadTestCancel(void);
int32_t taosThreadSigMask(int32_t how, sigset_t const *set, sigset_t * oset);
int32_t taosThreadSigWait(const sigset_t * set, int32_t *sig);
#ifdef __cplusplus
}
#endif
......
......@@ -20,6 +20,8 @@
extern "C" {
#endif
#include <time.h>
// If the error is in a third-party library, place this header file under the third-party library header file.
// When you want to use this feature, you should find or add the same function in the following section.
#ifndef ALLOW_FORBID_FUNC
......
......@@ -16,7 +16,7 @@
#ifndef _TD_UTIL_TAOS_ERROR_H_
#define _TD_UTIL_TAOS_ERROR_H_
#include "os.h"
#include <stdint.h>
#ifdef __cplusplus
extern "C" {
......
......@@ -381,6 +381,14 @@ typedef enum ELogicConditionType {
#define TSDB_MAX_DB_DELAY 10
#define TSDB_DEFAULT_DB_DELAY 2
#define TSDB_DEFAULT_EXPLAIN_VERBOSE false
#define TSDB_MIN_EXPLAIN_RATIO 0
#define TSDB_MAX_EXPLAIN_RATIO 1
#define TSDB_DEFAULT_EXPLAIN_RATIO 0.001
#define TSDB_EXPLAIN_RESULT_ROW_SIZE 1024
#define TSDB_MAX_JOIN_TABLE_NUM 10
#define TSDB_MAX_UNION_CLAUSE 5
......
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "consumer.h"
......@@ -200,6 +200,9 @@ static void doDestroyRequest(void *p) {
taosArrayDestroy(pRequest->body.showInfo.pArray);
}
taosArrayDestroy(pRequest->tableList);
taosArrayDestroy(pRequest->dbList);
deregisterRequest(pRequest);
taosMemoryFreeClear(pRequest);
}
......
......@@ -71,6 +71,12 @@ void taos_cleanup(void) {
tscInfo("all local resources released");
}
setConfRet taos_set_config(const char *config) {
// TODO
setConfRet ret = {SET_CONF_RET_SUCC, {0}};
return ret;
}
TAOS *taos_connect(const char *ip, const char *user, const char *pass, const char *db, uint16_t port) {
tscDebug("try to connect to %s:%u, user:%s db:%s", ip, port, user, db);
if (user == NULL) {
......@@ -257,6 +263,11 @@ int *taos_fetch_lengths(TAOS_RES *res) {
return ((SRequestObj *)res)->body.resInfo.length;
}
TAOS_ROW *taos_result_block(TAOS_RES *res) {
// TODO
return NULL;
}
// todo intergrate with tDataTypes
const char *taos_data_type(int type) {
switch (type) {
......@@ -353,6 +364,11 @@ bool taos_is_null(TAOS_RES *res, int32_t row, int32_t col) {
return colDataIsNull_f(pCol->nullbitmap, row);
}
bool taos_is_update_query(TAOS_RES *res) {
// TODO
return true;
}
int taos_fetch_block(TAOS_RES *res, TAOS_ROW *rows) {
if (res == NULL) {
return 0;
......@@ -376,6 +392,11 @@ int taos_fetch_block(TAOS_RES *res, TAOS_ROW *rows) {
int taos_validate_sql(TAOS *taos, const char *sql) { return true; }
void taos_reset_current_db(TAOS *taos) {
// TODO
return;
}
const char *taos_get_server_info(TAOS *taos) {
if (taos == NULL) {
return NULL;
......@@ -407,6 +428,11 @@ void taos_unsubscribe(TAOS_SUB *tsub, int keepProgress) {
// TODO
}
int taos_load_table_info(TAOS *taos, const char *tableNameList) {
// TODO
return -1;
}
TAOS_STMT *taos_stmt_init(TAOS *taos) {
// TODO
return NULL;
......
......@@ -241,6 +241,10 @@ void tmq_list_destroy(tmq_list_t* list) {
taosArrayDestroyEx(container, (void (*)(void*))taosMemoryFree);
}
static int32_t tmqMakeTopicVgKey(char* dst, const char* topicName, int32_t vg) {
return sprintf(dst, "%s:%d", topicName, vg);
}
void tmqClearUnhandleMsg(tmq_t* tmq) {
tmq_message_t* msg = NULL;
while (1) {
......@@ -827,7 +831,7 @@ int32_t tmqPollCb(void* param, const SDataBuf* pMsg, int32_t code) {
SMqClientVg* pVg = pParam->pVg;
tmq_t* tmq = pParam->tmq;
if (code != 0) {
printf("msg discard, code:%x\n", code);
tscWarn("msg discard, code:%x", code);
goto WRITE_QUEUE_FAIL;
}
......@@ -835,12 +839,12 @@ int32_t tmqPollCb(void* param, const SDataBuf* pMsg, int32_t code) {
int32_t tmqEpoch = atomic_load_32(&tmq->epoch);
if (msgEpoch < tmqEpoch) {
tsem_post(&tmq->rspSem);
printf("discard rsp epoch %d, current epoch %d\n", msgEpoch, tmqEpoch);
tscWarn("discard rsp epoch %d, current epoch %d", msgEpoch, tmqEpoch);
return 0;
}
if (msgEpoch != tmqEpoch) {
printf("mismatch rsp epoch %d, current epoch %d\n", msgEpoch, tmqEpoch);
tscWarn("mismatch rsp epoch %d, current epoch %d", msgEpoch, tmqEpoch);
} else {
atomic_sub_fetch_32(&tmq->waitingRequest, 1);
}
......@@ -899,19 +903,54 @@ WRITE_QUEUE_FAIL:
bool tmqUpdateEp(tmq_t* tmq, int32_t epoch, SMqCMGetSubEpRsp* pRsp) {
/*printf("call update ep %d\n", epoch);*/
bool set = false;
int32_t sz = taosArrayGetSize(pRsp->topics);
SArray* newTopics = taosArrayInit(sz, sizeof(SMqClientTopic));
for (int32_t i = 0; i < sz; i++) {
int32_t topicNumGet = taosArrayGetSize(pRsp->topics);
char vgKey[TSDB_TOPIC_FNAME_LEN + 22];
SArray* newTopics = taosArrayInit(topicNumGet, sizeof(SMqClientTopic));
if (newTopics == NULL) {
return false;
}
SHashObj* pHash = taosHashInit(64, MurmurHash3_32, false, HASH_NO_LOCK);
if (pHash == NULL) {
taosArrayDestroy(newTopics);
return false;
}
// find topic, build hash
for (int32_t i = 0; i < topicNumGet; i++) {
SMqClientTopic topic = {0};
SMqSubTopicEp* pTopicEp = taosArrayGet(pRsp->topics, i);
taosHashClear(pHash);
topic.topicName = strdup(pTopicEp->topic);
int32_t vgSz = taosArrayGetSize(pTopicEp->vgs);
topic.vgs = taosArrayInit(vgSz, sizeof(SMqClientVg));
for (int32_t j = 0; j < vgSz; j++) {
int32_t topicNumCur = taosArrayGetSize(tmq->clientTopics);
for (int32_t j = 0; j < topicNumCur; j++) {
// find old topic
SMqClientTopic* pTopicCur = taosArrayGet(tmq->clientTopics, j);
if (pTopicCur->vgs && strcmp(pTopicCur->topicName, pTopicEp->topic) == 0) {
int32_t vgNumCur = taosArrayGetSize(pTopicCur->vgs);
if (vgNumCur == 0) break;
for (int32_t k = 0; k < vgNumCur; k++) {
SMqClientVg* pVgCur = taosArrayGet(pTopicCur->vgs, k);
sprintf(vgKey, "%s:%d", topic.topicName, pVgCur->vgId);
taosHashPut(pHash, vgKey, strlen(vgKey), &pVgCur->currentOffset, sizeof(int64_t));
}
break;
}
}
int32_t vgNumGet = taosArrayGetSize(pTopicEp->vgs);
topic.vgs = taosArrayInit(vgNumGet, sizeof(SMqClientVg));
for (int32_t j = 0; j < vgNumGet; j++) {
SMqSubVgEp* pVgEp = taosArrayGet(pTopicEp->vgs, j);
sprintf(vgKey, "%s:%d", topic.topicName, pVgEp->vgId);
int64_t* pOffset = taosHashGet(pHash, vgKey, strlen(vgKey));
int64_t offset = pVgEp->offset;
if (pOffset != NULL) {
offset = *pOffset;
}
SMqClientVg clientVg = {
.pollCnt = 0,
.currentOffset = pVgEp->offset,
.currentOffset = offset,
.vgId = pVgEp->vgId,
.epSet = pVgEp->epSet,
.vgStatus = TMQ_VG_STATUS__IDLE,
......@@ -922,6 +961,7 @@ bool tmqUpdateEp(tmq_t* tmq, int32_t epoch, SMqCMGetSubEpRsp* pRsp) {
taosArrayPush(newTopics, &topic);
}
if (tmq->clientTopics) taosArrayDestroy(tmq->clientTopics);
taosHashCleanup(pHash);
tmq->clientTopics = newTopics;
atomic_store_32(&tmq->epoch, epoch);
return set;
......@@ -931,7 +971,7 @@ int32_t tmqAskEpCb(void* param, const SDataBuf* pMsg, int32_t code) {
SMqAskEpCbParam* pParam = (SMqAskEpCbParam*)param;
tmq_t* tmq = pParam->tmq;
if (code != 0) {
printf("get topic endpoint error, not ready, wait:%d\n", pParam->sync);
tscError("get topic endpoint error, not ready, wait:%d\n", pParam->sync);
goto END;
}
......@@ -1302,6 +1342,7 @@ tmq_message_t* tmq_consumer_poll(tmq_t* tmq, int64_t blocking_time) {
while (1) {
/*printf("cycle\n");*/
tmqAskEp(tmq, false);
tmqPollImpl(tmq, blocking_time);
tsem_wait(&tmq->rspSem);
......
......@@ -45,7 +45,7 @@ float tsRatioOfQueryCores = 1.0f;
int32_t tsMaxBinaryDisplayWidth = 30;
bool tsEnableSlaveQuery = 1;
bool tsPrintAuth = 0;
int32_t tsMultiProcess = 0;
bool tsMultiProcess = 0;
// monitor
bool tsEnableMonitor = 1;
......@@ -347,7 +347,7 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
if (cfgAddBool(pCfg, "printAuth", tsPrintAuth, 0) != 0) return -1;
if (cfgAddBool(pCfg, "slaveQuery", tsEnableSlaveQuery, 0) != 0) return -1;
if (cfgAddBool(pCfg, "deadLockKillQuery", tsDeadLockKillQuery, 0) != 0) return -1;
if (cfgAddInt32(pCfg, "multiProcess", tsMultiProcess, 0, 2, 0) != 0) return -1;
if (cfgAddBool(pCfg, "multiProcess", tsMultiProcess, 0) != 0) return -1;
if (cfgAddBool(pCfg, "monitor", tsEnableMonitor, 0) != 0) return -1;
if (cfgAddInt32(pCfg, "monitorInterval", tsMonitorInterval, 1, 360000, 0) != 0) return -1;
......@@ -466,7 +466,7 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
tsPrintAuth = cfgGetItem(pCfg, "printAuth")->bval;
tsEnableSlaveQuery = cfgGetItem(pCfg, "slaveQuery")->bval;
tsDeadLockKillQuery = cfgGetItem(pCfg, "deadLockKillQuery")->bval;
tsMultiProcess = cfgGetItem(pCfg, "multiProcess")->i32;
tsMultiProcess = cfgGetItem(pCfg, "multiProcess")->bval;
tsEnableMonitor = cfgGetItem(pCfg, "monitor")->bval;
tsMonitorInterval = cfgGetItem(pCfg, "monitorInterval")->i32;
......
aux_source_directory(src DNODE_SRC)
aux_source_directory(dnode/src DNODE_SRC)
aux_source_directory(qnode/src DNODE_SRC)
aux_source_directory(bnode/src DNODE_SRC)
aux_source_directory(snode/src DNODE_SRC)
aux_source_directory(vnode/src DNODE_SRC)
aux_source_directory(mnode/src DNODE_SRC)
aux_source_directory(container/src DNODE_SRC)
aux_source_directory(dm/src DNODE_SRC)
aux_source_directory(qm/src DNODE_SRC)
aux_source_directory(bm/src DNODE_SRC)
aux_source_directory(sm/src DNODE_SRC)
aux_source_directory(vm/src DNODE_SRC)
aux_source_directory(mm/src DNODE_SRC)
aux_source_directory(main/src DNODE_SRC)
add_library(dnode STATIC ${DNODE_SRC})
target_link_libraries(
dnode cjson mnode vnode qnode snode bnode wal sync taos tfs monitor
......@@ -14,16 +12,22 @@ target_link_libraries(
target_include_directories(
dnode
PUBLIC "${CMAKE_SOURCE_DIR}/include/dnode/mgmt"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/dnode/inc"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/qnode/inc"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/bnode/inc"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/snode/inc"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/vnode/inc"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/mnode/inc"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/container/inc"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/dm/inc"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/qm/inc"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/bm/inc"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/sm/inc"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/vm/inc"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/mm/inc"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/main/inc"
)
add_subdirectory(main)
aux_source_directory(main/exe EXEC_SRC)
add_executable(taosd ${EXEC_SRC})
target_include_directories(
taosd
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/main/inc"
)
target_link_libraries(taosd dnode)
if(${BUILD_TEST})
add_subdirectory(test)
......
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http:www.gnu.org/licenses/>.
*/
#define _DEFAULT_SOURCE
#include "dndInt.h"
#define MAXLEN 1024
int32_t dndReadFile(SMgmtWrapper *pWrapper, bool *pDeployed) {
int32_t code = TSDB_CODE_NODE_PARSE_FILE_ERROR;
int32_t len = 0;
const int32_t maxLen = MAXLEN;
char content[MAXLEN + 1] = {0};
cJSON *root = NULL;
char file[PATH_MAX];
TdFilePtr pFile = NULL;
snprintf(file, sizeof(file), "%s%s%s.json", pWrapper->path, TD_DIRSEP, pWrapper->name);
pFile = taosOpenFile(file, TD_FILE_READ);
if (pFile == NULL) {
dDebug("file %s not exist", file);
code = 0;
goto _OVER;
}
len = (int32_t)taosReadFile(pFile, content, maxLen);
if (len <= 0) {
dError("failed to read %s since content is null", file);
goto _OVER;
}
content[len] = 0;
root = cJSON_Parse(content);
if (root == NULL) {
dError("failed to read %s since invalid json format", file);
goto _OVER;
}
cJSON *deployed = cJSON_GetObjectItem(root, "deployed");
if (!deployed || deployed->type != cJSON_Number) {
dError("failed to read %s since deployed not found", file);
goto _OVER;
}
*pDeployed = deployed->valueint != 0;
code = 0;
dDebug("succcessed to read file %s, deployed:%d", file, *pDeployed);
_OVER:
if (root != NULL) cJSON_Delete(root);
if (pFile != NULL) taosCloseFile(&pFile);
terrno = code;
return code;
}
int32_t dndWriteFile(SMgmtWrapper *pWrapper, bool deployed) {
char file[PATH_MAX] = {0};
snprintf(file, sizeof(file), "%s%s%s.json", pWrapper->path, TD_DIRSEP, pWrapper->name);
TdFilePtr pFile = taosOpenFile(file, TD_FILE_CTEATE | TD_FILE_WRITE | TD_FILE_TRUNC);
if (pFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to write %s since %s", file, terrstr());
return -1;
}
int32_t len = 0;
const int32_t maxLen = MAXLEN;
char content[MAXLEN + 1] = {0};
len += snprintf(content + len, maxLen - len, "{\n");
len += snprintf(content + len, maxLen - len, " \"deployed\": %d\n", deployed);
len += snprintf(content + len, maxLen - len, "}\n");
taosWriteFile(pFile, content, len);
taosFsyncFile(pFile);
taosCloseFile(&pFile);
char realfile[PATH_MAX] = {0};
snprintf(realfile, sizeof(realfile), "%s%s%s.json", pWrapper->path, TD_DIRSEP, pWrapper->name);
if (taosRenameFile(file, realfile) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to rename %s since %s", file, terrstr());
return -1;
}
dInfo("successed to write %s, deployed:%d", realfile, deployed);
return 0;
}
aux_source_directory(src EXEC_SRC)
add_executable(taosd ${EXEC_SRC})
target_include_directories(
taosd
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
)
target_link_libraries(taosd dnode)
......@@ -14,19 +14,22 @@
*/
#define _DEFAULT_SOURCE
#include "dndMain.h"
#include "dnd.h"
#include "tconfig.h"
static struct {
bool dumpConfig;
bool generateGrant;
bool printAuth;
bool printVersion;
char envFile[PATH_MAX];
char apolloUrl[PATH_MAX];
SDnode *pDnode;
bool dumpConfig;
bool generateGrant;
bool printAuth;
bool printVersion;
char envFile[PATH_MAX];
char apolloUrl[PATH_MAX];
SArray *pArgs; // SConfigPair
SDnode *pDnode;
ENodeType ntype;
} global = {0};
static void dndSigintHandle(int signum, void *info, void *ctx) {
static void dndStopDnode(int signum, void *info, void *ctx) {
dInfo("signal:%d is received", signum);
SDnode *pDnode = atomic_val_compare_exchange_ptr(&global.pDnode, 0, global.pDnode);
if (pDnode != NULL) {
......@@ -34,15 +37,32 @@ static void dndSigintHandle(int signum, void *info, void *ctx) {
}
}
static void dndHandleChild(int signum, void *info, void *ctx) {
dInfo("signal:%d is received", signum);
dndHandleEvent(global.pDnode, DND_EVENT_CHILD);
}
static void dndSetSignalHandle() {
taosSetSignal(SIGTERM, dndSigintHandle);
taosSetSignal(SIGHUP, dndSigintHandle);
taosSetSignal(SIGINT, dndSigintHandle);
taosSetSignal(SIGABRT, dndSigintHandle);
taosSetSignal(SIGBREAK, dndSigintHandle);
taosSetSignal(SIGTERM, dndStopDnode);
taosSetSignal(SIGHUP, dndStopDnode);
taosSetSignal(SIGINT, dndStopDnode);
taosSetSignal(SIGABRT, dndStopDnode);
taosSetSignal(SIGBREAK, dndStopDnode);
if (!tsMultiProcess) {
// Set the single process signal
} else if (global.ntype == DNODE) {
// Set the parent process signal
// When the child process exits, the parent process receives a signal
taosSetSignal(SIGCHLD, dndHandleChild);
} else {
// Set child process signal
// When the parent process exits, the child process will receive the SIGKILL signal
prctl(PR_SET_PDEATHSIG, SIGKILL);
}
}
static int32_t dndParseOption(int32_t argc, char const *argv[]) {
static int32_t dndParseArgs(int32_t argc, char const *argv[]) {
for (int32_t i = 1; i < argc; ++i) {
if (strcmp(argv[i], "-c") == 0) {
if (i < argc - 1) {
......@@ -55,10 +75,16 @@ static int32_t dndParseOption(int32_t argc, char const *argv[]) {
printf("'-c' requires a parameter, default is %s\n", configDir);
return -1;
}
} else if (strcmp(argv[i], "-C") == 0) {
global.dumpConfig = true;
} else if (strcmp(argv[i], "-a") == 0) {
tstrncpy(global.apolloUrl, argv[++i], PATH_MAX);
} else if (strcmp(argv[i], "-e") == 0) {
tstrncpy(global.envFile, argv[++i], PATH_MAX);
} else if (strcmp(argv[i], "-k") == 0) {
global.generateGrant = true;
} else if (strcmp(argv[i], "-n") == 0) {
global.ntype = atoi(argv[++i]);
} else if (strcmp(argv[i], "-C") == 0) {
global.dumpConfig = true;
} else if (strcmp(argv[i], "-V") == 0) {
global.printVersion = true;
} else {
......@@ -68,15 +94,66 @@ static int32_t dndParseOption(int32_t argc, char const *argv[]) {
return 0;
}
static void dndGenerateGrant() {
// grantParseParameter();
printf("this feature is not implemented yet\n");
}
static void dndPrintVersion() {
#ifdef TD_ENTERPRISE
char *releaseName = "enterprise";
#else
char *releaseName = "community";
#endif
printf("%s version: %s compatible_version: %s\n", releaseName, version, compatible_version);
printf("gitinfo: %s\n", gitinfo);
printf("buildInfo: %s\n", buildinfo);
}
static void dndDumpCfg() {
SConfig *pCfg = taosGetCfg();
cfgDumpCfg(pCfg, 0, 1);
}
static SDnodeOpt dndGetOpt() {
SConfig *pCfg = taosGetCfg();
SDnodeOpt option = {0};
option.numOfSupportVnodes = cfgGetItem(pCfg, "supportVnodes")->i32;
tstrncpy(option.dataDir, tsDataDir, sizeof(option.dataDir));
tstrncpy(option.firstEp, tsFirst, sizeof(option.firstEp));
tstrncpy(option.secondEp, tsSecond, sizeof(option.firstEp));
option.serverPort = tsServerPort;
tstrncpy(option.localFqdn, tsLocalFqdn, sizeof(option.localFqdn));
snprintf(option.localEp, sizeof(option.localEp), "%s:%u", option.localFqdn, option.serverPort);
option.disks = tsDiskCfg;
option.numOfDisks = tsDiskCfgNum;
option.ntype = global.ntype;
return option;
}
static int32_t dndInitLog() {
char logName[12] = {0};
snprintf(logName, sizeof(logName), "%slog", dndNodeLogStr(global.ntype));
return taosCreateLog(logName, 1, configDir, global.envFile, global.apolloUrl, global.pArgs, 0);
}
static void dndSetProcName(char **argv) {
if (global.ntype != DNODE) {
const char *name = dndNodeProcStr(global.ntype);
prctl(PR_SET_NAME, name);
strcpy(argv[0], name);
}
}
static int32_t dndRunDnode() {
if (dndInit() != 0) {
dInfo("failed to initialize dnode environment since %s", terrstr());
dError("failed to initialize environment since %s", terrstr());
return -1;
}
SDnodeOpt option = dndGetOpt();
SDnode *pDnode = dndCreate(&option);
SDnode *pDnode = dndCreate(&option);
if (pDnode == NULL) {
dError("failed to to create dnode object since %s", terrstr());
return -1;
......@@ -85,9 +162,9 @@ static int32_t dndRunDnode() {
dndSetSignalHandle();
}
dInfo("start the TDengine service");
dInfo("start the service");
int32_t code = dndRun(pDnode);
dInfo("start shutting down the TDengine service");
dInfo("start shutting down the service");
global.pDnode = NULL;
dndClose(pDnode);
......@@ -99,11 +176,12 @@ static int32_t dndRunDnode() {
int main(int argc, char const *argv[]) {
if (!taosCheckSystemIsSmallEnd()) {
dError("failed to start TDengine since on non-small-end machines");
printf("failed to start since on non-small-end machines\n");
return -1;
}
if (dndParseOption(argc, argv) != 0) {
if (dndParseArgs(argc, argv) != 0) {
printf("failed to start since parse args error\n");
return -1;
}
......@@ -117,13 +195,13 @@ int main(int argc, char const *argv[]) {
return 0;
}
if (taosCreateLog("taosdlog", 1, configDir, global.envFile, global.apolloUrl, NULL, 0) != 0) {
dError("failed to start TDengine since read log config error");
if (dndInitLog() != 0) {
printf("failed to start since init log error\n");
return -1;
}
if (taosInitCfg(configDir, global.envFile, global.apolloUrl, NULL, 0) != 0) {
dError("failed to start TDengine since read config error");
if (taosInitCfg(configDir, global.envFile, global.apolloUrl, global.pArgs, 0) != 0) {
dError("failed to start since read config error");
return -1;
}
......@@ -134,5 +212,6 @@ int main(int argc, char const *argv[]) {
return 0;
}
dndSetProcName((char **)argv);
return dndRunDnode();
}
......@@ -96,10 +96,11 @@ typedef struct SMgmtWrapper {
bool required;
EProcType procType;
SProcObj *pProc;
SShm shm;
void *pMgmt;
SDnode *pDnode;
NodeMsgFp msgFps[TDMT_MAX];
int32_t msgVgIds[TDMT_MAX]; // Handle the case where the same message type is distributed to qnode or vnode
int8_t msgVgIds[TDMT_MAX]; // Handle the case where the same message type is distributed to qnode or vnode
SMgmtFp fp;
} SMgmtWrapper;
......@@ -119,23 +120,26 @@ typedef struct SDnode {
char *firstEp;
char *secondEp;
char *dataDir;
SDiskCfg *pDisks;
SDiskCfg *disks;
int32_t numOfDisks;
uint16_t serverPort;
bool dropped;
ENodeType ntype;
EDndStatus status;
EDndEvent event;
SStartupReq startup;
TdFilePtr pLockFile;
TdFilePtr runtimeFile;
STransMgmt trans;
SMgmtWrapper wrappers[NODE_MAX];
} SDnode;
EDndStatus dndGetStatus(SDnode *pDnode);
void dndSetStatus(SDnode *pDnode, EDndStatus stat);
void dndSetMsgHandle(SMgmtWrapper *pWrapper, int32_t msgType, NodeMsgFp nodeMsgFp, int32_t vgId);
void dndReportStartup(SDnode *pDnode, const char *pName, const char *pDesc);
void dndSendMonitorReport(SDnode *pDnode);
const char *dndNodeLogStr(ENodeType ntype);
const char *dndNodeProcStr(ENodeType ntype);
EDndStatus dndGetStatus(SDnode *pDnode);
void dndSetStatus(SDnode *pDnode, EDndStatus stat);
void dndSetMsgHandle(SMgmtWrapper *pWrapper, tmsg_t msgType, NodeMsgFp nodeMsgFp, int8_t vgId);
void dndReportStartup(SDnode *pDnode, const char *pName, const char *pDesc);
void dndSendMonitorReport(SDnode *pDnode);
int32_t dndSendReqToMnode(SMgmtWrapper *pWrapper, SRpcMsg *pMsg);
int32_t dndSendReqToDnode(SMgmtWrapper *pWrapper, const SEpSet *pEpSet, SRpcMsg *pMsg);
......
......@@ -34,7 +34,6 @@ int32_t dndInit();
void dndCleanup();
const char *dndStatStr(EDndStatus stat);
void dndGetStartup(SDnode *pDnode, SStartupReq *pStartup);
TdFilePtr dndCheckRunning(const char *dataDir);
void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pMsg);
// dndMsg.c
......@@ -58,6 +57,11 @@ void dndCleanupClient(SDnode *pDnode);
int32_t dndInitMsgHandle(SDnode *pDnode);
void dndSendRpcRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp);
// dndFile.c
int32_t dndOpenRuntimeFile(SDnode *pDnode);
int32_t dndWriteRuntimeFile(SDnode *pDnode);
void dndCloseRuntimeFile(SDnode *pDnode);
#ifdef __cplusplus
}
#endif
......
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_DND_MAIN_H_
#define _TD_DND_MAIN_H_
#include "dnode.h"
#include "taoserror.h"
#include "tconfig.h"
#include "tglobal.h"
#include "tlog.h"
#include "version.h"
#ifdef __cplusplus
extern "C" {
#endif
#define dFatal(...) { if (dDebugFlag & DEBUG_FATAL) { taosPrintLog("DND FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); }}
#define dError(...) { if (dDebugFlag & DEBUG_ERROR) { taosPrintLog("DND ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); }}
#define dWarn(...) { if (dDebugFlag & DEBUG_WARN) { taosPrintLog("DND WARN ", DEBUG_WARN, 255, __VA_ARGS__); }}
#define dInfo(...) { if (dDebugFlag & DEBUG_INFO) { taosPrintLog("DND ", DEBUG_INFO, 255, __VA_ARGS__); }}
#define dDebug(...) { if (dDebugFlag & DEBUG_DEBUG) { taosPrintLog("DND ", DEBUG_DEBUG, dDebugFlag, __VA_ARGS__); }}
#define dTrace(...) { if (dDebugFlag & DEBUG_TRACE) { taosPrintLog("DND ", DEBUG_TRACE, dDebugFlag, __VA_ARGS__); }}
void dndDumpCfg();
void dndPrintVersion();
void dndGenerateGrant();
SDnodeOpt dndGetOpt();
#ifdef __cplusplus
}
#endif
#endif /*_TD_DND_MAIN_H_*/
......@@ -259,7 +259,7 @@ static int32_t dndRunInMultiProcess(SDnode *pDnode) {
}
int32_t dndRun(SDnode *pDnode) {
if (tsMultiProcess == 0) {
if (!tsMultiProcess) {
if (dndRunInSingleProcess(pDnode) != 0) {
dError("failed to run dnode in single process mode since %s", terrstr());
return -1;
......
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http:www.gnu.org/licenses/>.
*/
#define _DEFAULT_SOURCE
#include "dndInt.h"
#define MAXLEN 1024
int32_t dndReadFile(SMgmtWrapper *pWrapper, bool *pDeployed) {
int32_t code = TSDB_CODE_NODE_PARSE_FILE_ERROR;
int64_t len = 0;
char content[MAXLEN + 1] = {0};
cJSON *root = NULL;
char file[PATH_MAX];
TdFilePtr pFile = NULL;
snprintf(file, sizeof(file), "%s%s%s.json", pWrapper->path, TD_DIRSEP, pWrapper->name);
pFile = taosOpenFile(file, TD_FILE_READ);
if (pFile == NULL) {
dDebug("file %s not exist", file);
code = 0;
goto _OVER;
}
len = taosReadFile(pFile, content, MAXLEN);
if (len <= 0) {
dError("failed to read %s since content is null", file);
goto _OVER;
}
root = cJSON_Parse(content);
if (root == NULL) {
dError("failed to read %s since invalid json format", file);
goto _OVER;
}
cJSON *deployed = cJSON_GetObjectItem(root, "deployed");
if (!deployed || deployed->type != cJSON_Number) {
dError("failed to read %s since deployed not found", file);
goto _OVER;
}
*pDeployed = deployed->valueint != 0;
dDebug("succcessed to read file %s, deployed:%d", file, *pDeployed);
code = 0;
_OVER:
if (root != NULL) cJSON_Delete(root);
if (pFile != NULL) taosCloseFile(&pFile);
terrno = code;
return code;
}
int32_t dndWriteFile(SMgmtWrapper *pWrapper, bool deployed) {
int32_t code = -1;
int32_t len = 0;
char content[MAXLEN + 1] = {0};
char file[PATH_MAX] = {0};
char realfile[PATH_MAX] = {0};
TdFilePtr pFile = NULL;
snprintf(file, sizeof(file), "%s%s%s.json", pWrapper->path, TD_DIRSEP, pWrapper->name);
snprintf(realfile, sizeof(realfile), "%s%s%s.json", pWrapper->path, TD_DIRSEP, pWrapper->name);
pFile = taosOpenFile(file, TD_FILE_CTEATE | TD_FILE_WRITE | TD_FILE_TRUNC);
if (pFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to write %s since %s", file, terrstr());
goto _OVER;
}
len += snprintf(content + len, MAXLEN - len, "{\n");
len += snprintf(content + len, MAXLEN - len, " \"deployed\": %d\n", deployed);
len += snprintf(content + len, MAXLEN - len, "}\n");
if (taosWriteFile(pFile, content, len) != len) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to write file:%s since %s", file, terrstr());
goto _OVER;
}
if (taosFsyncFile(pFile) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to fsync file:%s since %s", file, terrstr());
goto _OVER;
}
taosCloseFile(&pFile);
if (taosRenameFile(file, realfile) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to rename %s since %s", file, terrstr());
return -1;
}
dInfo("successed to write %s, deployed:%d", realfile, deployed);
code = 0;
_OVER:
if (pFile != NULL) {
taosCloseFile(&pFile);
}
return code;
}
int32_t dndOpenRuntimeFile(SDnode *pDnode) {
int32_t code = -1;
char itemName[24] = {0};
char content[MAXLEN + 1] = {0};
char file[PATH_MAX] = {0};
cJSON *root = NULL;
TdFilePtr pFile = NULL;
snprintf(file, sizeof(file), "%s%s.running", pDnode->dataDir, TD_DIRSEP);
pFile = taosOpenFile(file, TD_FILE_CTEATE | TD_FILE_WRITE | TD_FILE_TRUNC);
if (pFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to open file:%s since %s", file, terrstr());
goto _OVER;
}
if (taosLockFile(pFile) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to lock file:%s since %s", file, terrstr());
goto _OVER;
}
if (taosReadFile(pFile, content, MAXLEN) > 0) {
root = cJSON_Parse(content);
if (root == NULL) {
terrno = TSDB_CODE_NODE_PARSE_FILE_ERROR;
dError("failed to read %s since invalid json format", file);
goto _OVER;
}
for (ENodeType ntype = DNODE + 1; ntype < NODE_MAX; ++ntype) {
snprintf(itemName, sizeof(itemName), "%s_shmid", dndNodeProcStr(ntype));
cJSON *shmid = cJSON_GetObjectItem(root, itemName);
if (shmid && shmid->type == cJSON_Number) {
pDnode->wrappers[ntype].shm.id = shmid->valueint;
}
snprintf(itemName, sizeof(itemName), "%s_shmsize", dndNodeProcStr(ntype));
cJSON *shmsize = cJSON_GetObjectItem(root, itemName);
if (shmsize && shmsize->type == cJSON_Number) {
pDnode->wrappers[ntype].shm.size = shmsize->valueint;
}
}
}
if (tsMultiProcess || pDnode->ntype == DNODE) {
for (ENodeType ntype = DNODE; ntype < NODE_MAX; ++ntype) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[pDnode->ntype];
if (pWrapper->shm.id > 0) {
dDebug("shmid:%d, is closed, size:%d", pWrapper->shm.id, pWrapper->shm.size);
taosDropShm(&pWrapper->shm);
}
}
} else {
SMgmtWrapper *pWrapper = &pDnode->wrappers[pDnode->ntype];
if (taosAttachShm(&pWrapper->shm) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("shmid:%d, failed to attach since %s", pWrapper->shm.id, terrstr());
goto _OVER;
}
dDebug("shmid:%d, is attached, size:%d", pWrapper->shm.id, pWrapper->shm.size);
}
dDebug("successed to open %s", file);
code = 0;
_OVER:
if (root != NULL) cJSON_Delete(root);
if (code != 0) {
if (pFile != NULL) taosCloseFile(&pFile);
} else {
pDnode->runtimeFile = pFile;
}
return code;
}
int32_t dndWriteRuntimeFile(SDnode *pDnode) {
int32_t code = -1;
int32_t len = 0;
char content[MAXLEN + 1] = {0};
char file[PATH_MAX] = {0};
char realfile[PATH_MAX] = {0};
TdFilePtr pFile = NULL;
snprintf(file, sizeof(file), "%s%s.running.bak", pDnode->dataDir, TD_DIRSEP);
snprintf(realfile, sizeof(realfile), "%s%s.running", pDnode->dataDir, TD_DIRSEP);
pFile = taosOpenFile(file, TD_FILE_CTEATE | TD_FILE_WRITE | TD_FILE_TRUNC);
if (pFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to open file:%s since %s", file, terrstr());
goto _OVER;
}
len += snprintf(content + len, MAXLEN - len, "{\n");
for (ENodeType ntype = DNODE + 1; ntype < NODE_MAX; ++ntype) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[pDnode->ntype];
len += snprintf(content + len, MAXLEN - len, " \"%s_shmid\": %d,\n", dndNodeProcStr(ntype), pWrapper->shm.id);
if (ntype == NODE_MAX - 1) {
len += snprintf(content + len, MAXLEN - len, " \"%s_shmsize\": %d\n", dndNodeProcStr(ntype), pWrapper->shm.size);
} else {
len += snprintf(content + len, MAXLEN - len, " \"%s_shmsize\": %d,\n", dndNodeProcStr(ntype), pWrapper->shm.size);
}
}
len += snprintf(content + len, MAXLEN - len, "}\n");
if (taosWriteFile(pFile, content, len) != len) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to write file:%s since %s", file, terrstr());
goto _OVER;
}
if (taosFsyncFile(pFile) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to fsync file:%s since %s", file, terrstr());
goto _OVER;
}
taosCloseFile(&pFile);
if (taosRenameFile(file, realfile) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to rename %s to %s since %s", file, realfile, terrstr());
return -1;
}
dDebug("successed to write %s", realfile);
code = 0;
_OVER:
if (pFile != NULL) {
taosCloseFile(&pFile);
}
return code;
}
void dndCloseRuntimeFile(SDnode *pDnode) {
if (pDnode->runtimeFile) {
taosUnLockFile(pDnode->runtimeFile);
taosCloseFile(&pDnode->runtimeFile);
pDnode->runtimeFile = NULL;
}
}
\ No newline at end of file
......@@ -20,7 +20,7 @@
static int8_t once = DND_ENV_INIT;
int32_t dndInit() {
dInfo("start to init dnode env");
dDebug("start to init dnode env");
if (atomic_val_compare_exchange_8(&once, DND_ENV_INIT, DND_ENV_READY) != DND_ENV_INIT) {
terrno = TSDB_CODE_REPEAT_INIT;
dError("failed to init dnode env since %s", terrstr());
......@@ -31,12 +31,6 @@ int32_t dndInit() {
taosBlockSIGPIPE();
taosResolveCRC();
if (rpcInit() != 0) {
dError("failed to init rpc since %s", terrstr());
dndCleanup();
return -1;
}
SMonCfg monCfg = {0};
monCfg.maxLogs = tsMonitorMaxLogs;
monCfg.port = tsMonitorPort;
......@@ -44,29 +38,27 @@ int32_t dndInit() {
monCfg.comp = tsMonitorComp;
if (monInit(&monCfg) != 0) {
dError("failed to init monitor since %s", terrstr());
dndCleanup();
return -1;
}
dInfo("dnode env is initialized");
dDebug("dnode env is initialized");
return 0;
}
void dndCleanup() {
dInfo("start to cleanup dnode env");
dDebug("start to cleanup dnode env");
if (atomic_val_compare_exchange_8(&once, DND_ENV_READY, DND_ENV_CLEANUP) != DND_ENV_READY) {
dError("dnode env is already cleaned up");
return;
}
monCleanup();
rpcCleanup();
walCleanUp();
taosStopCacheRefreshWorker();
dInfo("dnode env is cleaned up");
dDebug("dnode env is cleaned up");
}
void dndSetMsgHandle(SMgmtWrapper *pWrapper, int32_t msgType, NodeMsgFp nodeMsgFp, int32_t vgId) {
void dndSetMsgHandle(SMgmtWrapper *pWrapper, tmsg_t msgType, NodeMsgFp nodeMsgFp, int8_t vgId) {
pWrapper->msgFps[TMSG_INDEX(msgType)] = nodeMsgFp;
pWrapper->msgVgIds[TMSG_INDEX(msgType)] = vgId;
}
......@@ -80,19 +72,6 @@ void dndSetStatus(SDnode *pDnode, EDndStatus status) {
}
}
const char *dndStatStr(EDndStatus status) {
switch (status) {
case DND_STAT_INIT:
return "init";
case DND_STAT_RUNNING:
return "running";
case DND_STAT_STOPPED:
return "stopped";
default:
return "unknown";
}
}
void dndReportStartup(SDnode *pDnode, const char *pName, const char *pDesc) {
SStartupReq *pStartup = &pDnode->startup;
tstrncpy(pStartup->name, pName, TSDB_STEP_NAME_LEN);
......@@ -105,29 +84,6 @@ void dndGetStartup(SDnode *pDnode, SStartupReq *pStartup) {
pStartup->finished = (dndGetStatus(pDnode) == DND_STAT_RUNNING);
}
TdFilePtr dndCheckRunning(const char *dataDir) {
char filepath[PATH_MAX] = {0};
snprintf(filepath, sizeof(filepath), "%s/.running", dataDir);
TdFilePtr pFile = taosOpenFile(filepath, TD_FILE_CTEATE | TD_FILE_WRITE | TD_FILE_TRUNC);
if (pFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to lock file:%s since %s", filepath, terrstr());
return NULL;
}
int32_t ret = taosLockFile(pFile);
if (ret != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to lock file:%s since %s", filepath, terrstr());
taosCloseFile(&pFile);
return NULL;
}
dDebug("file:%s is locked", filepath);
return pFile;
}
void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pReq) {
dDebug("startup req is received");
SStartupReq *pStartup = rpcMallocCont(sizeof(SStartupReq));
......
......@@ -16,7 +16,7 @@
#define _DEFAULT_SOURCE
#include "dndInt.h"
static int32_t dndInitMemory(SDnode *pDnode, const SDnodeOpt *pOption) {
static int32_t dndInitVars(SDnode *pDnode, const SDnodeOpt *pOption) {
pDnode->numOfSupportVnodes = pOption->numOfSupportVnodes;
pDnode->serverPort = pOption->serverPort;
pDnode->dataDir = strdup(pOption->dataDir);
......@@ -24,8 +24,9 @@ static int32_t dndInitMemory(SDnode *pDnode, const SDnodeOpt *pOption) {
pDnode->localFqdn = strdup(pOption->localFqdn);
pDnode->firstEp = strdup(pOption->firstEp);
pDnode->secondEp = strdup(pOption->secondEp);
pDnode->pDisks = pOption->pDisks;
pDnode->disks = pOption->disks;
pDnode->numOfDisks = pOption->numOfDisks;
pDnode->ntype = pOption->ntype;
pDnode->rebootTime = taosGetTimestampMs();
if (pDnode->dataDir == NULL || pDnode->localEp == NULL || pDnode->localFqdn == NULL || pDnode->firstEp == NULL ||
......@@ -36,16 +37,12 @@ static int32_t dndInitMemory(SDnode *pDnode, const SDnodeOpt *pOption) {
return 0;
}
static void dndClearMemory(SDnode *pDnode) {
static void dndClearVars(SDnode *pDnode) {
for (ENodeType n = 0; n < NODE_MAX; ++n) {
SMgmtWrapper *pMgmt = &pDnode->wrappers[n];
taosMemoryFreeClear(pMgmt->path);
}
if (pDnode->pLockFile != NULL) {
taosUnLockFile(pDnode->pLockFile);
taosCloseFile(&pDnode->pLockFile);
pDnode->pLockFile = NULL;
}
dndCloseRuntimeFile(pDnode);
taosMemoryFreeClear(pDnode->localEp);
taosMemoryFreeClear(pDnode->localFqdn);
taosMemoryFreeClear(pDnode->firstEp);
......@@ -67,13 +64,21 @@ SDnode *dndCreate(const SDnodeOpt *pOption) {
goto _OVER;
}
if (dndInitMemory(pDnode, pOption) != 0) {
if (dndInitVars(pDnode, pOption) != 0) {
dError("failed to init variables since %s", terrstr());
goto _OVER;
}
dndSetStatus(pDnode, DND_STAT_INIT);
pDnode->pLockFile = dndCheckRunning(pDnode->dataDir);
if (pDnode->pLockFile == NULL) {
dmGetMgmtFp(&pDnode->wrappers[DNODE]);
mmGetMgmtFp(&pDnode->wrappers[MNODE]);
vmGetMgmtFp(&pDnode->wrappers[VNODES]);
qmGetMgmtFp(&pDnode->wrappers[QNODE]);
smGetMgmtFp(&pDnode->wrappers[SNODE]);
bmGetMgmtFp(&pDnode->wrappers[BNODE]);
if (dndOpenRuntimeFile(pDnode) != 0) {
dError("failed to open runtime file since %s", terrstr());
goto _OVER;
}
......@@ -87,13 +92,6 @@ SDnode *dndCreate(const SDnodeOpt *pOption) {
goto _OVER;
}
dmGetMgmtFp(&pDnode->wrappers[DNODE]);
mmGetMgmtFp(&pDnode->wrappers[MNODE]);
vmGetMgmtFp(&pDnode->wrappers[VNODES]);
qmGetMgmtFp(&pDnode->wrappers[QNODE]);
smGetMgmtFp(&pDnode->wrappers[SNODE]);
bmGetMgmtFp(&pDnode->wrappers[BNODE]);
if (dndInitMsgHandle(pDnode) != 0) {
goto _OVER;
}
......@@ -116,7 +114,7 @@ SDnode *dndCreate(const SDnodeOpt *pOption) {
_OVER:
if (code != 0 && pDnode) {
dndClearMemory(pDnode);
dndClearVars(pDnode);
pDnode = NULL;
dError("failed to create dnode object since %s", terrstr());
} else {
......@@ -145,7 +143,7 @@ void dndClose(SDnode *pDnode) {
dndCloseNode(pWrapper);
}
dndClearMemory(pDnode);
dndClearVars(pDnode);
dInfo("dnode object is closed, data:%p", pDnode);
}
......
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#define _DEFAULT_SOURCE
#include "dndInt.h"
const char *dndStatStr(EDndStatus status) {
switch (status) {
case DND_STAT_INIT:
return "init";
case DND_STAT_RUNNING:
return "running";
case DND_STAT_STOPPED:
return "stopped";
default:
return "unknown";
}
}
const char *dndNodeLogStr(ENodeType ntype) {
switch (ntype) {
case VNODES:
return "vnode";
case QNODE:
return "qnode";
case SNODE:
return "snode";
case MNODE:
return "mnode";
case BNODE:
return "bnode";
default:
return "taosd";
}
}
const char *dndNodeProcStr(ENodeType ntype) {
switch (ntype) {
case VNODES:
return "taosv";
case QNODE:
return "taosq";
case SNODE:
return "taoss";
case MNODE:
return "taosm";
case BNODE:
return "taosb";
default:
return "taosd";
}
}
......@@ -285,7 +285,7 @@ static int32_t vmInit(SMgmtWrapper *pWrapper) {
tstrncpy(dCfg.dir, pDnode->dataDir, TSDB_FILENAME_LEN);
dCfg.level = 0;
dCfg.primary = 1;
SDiskCfg *pDisks = pDnode->pDisks;
SDiskCfg *pDisks = pDnode->disks;
int32_t numOfDisks = pDnode->numOfDisks;
if (numOfDisks <= 0 || pDisks == NULL) {
pDisks = &dCfg;
......
......@@ -76,7 +76,7 @@ static void vmProcessFetchQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
SVnodeObj *pVnode = pInfo->ahandle;
dTrace("msg:%p, will be processed in vnode-fetch queue", pMsg);
int32_t code = vnodeProcessFetchMsg(pVnode->pImpl, &pMsg->rpcMsg);
int32_t code = vnodeProcessFetchMsg(pVnode->pImpl, &pMsg->rpcMsg, pInfo);
if (code != 0) {
vmSendRsp(pVnode->pWrapper, pMsg, code);
dTrace("msg:%p, is freed, result:0x%04x:%s", pMsg, code & 0XFFFF, tstrerror(code));
......@@ -168,7 +168,7 @@ static void vmProcessMergeQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
taosGetQitem(qall, (void **)&pMsg);
dTrace("msg:%p, will be processed in vnode-merge queue", pMsg);
int32_t code = vnodeProcessFetchMsg(pVnode->pImpl, &pMsg->rpcMsg);
int32_t code = vnodeProcessFetchMsg(pVnode->pImpl, &pMsg->rpcMsg, pInfo);
if (code != 0) {
vmSendRsp(pVnode->pWrapper, pMsg, code);
dTrace("msg:%p, is freed, result:0x%04x:%s", pMsg, code & 0XFFFF, tstrerror(code));
......@@ -414,8 +414,7 @@ int32_t vmStartWorker(SVnodesMgmt *pMgmt) {
pWPool->max = maxMergeThreads;
if (tWWorkerInit(pWPool) != 0) return -1;
SSingleWorkerCfg cfg = {
.min = 1, .max = 1, .name = "vnode-mgmt", .fp = (FItem)vmProcessMgmtQueue, .param = pMgmt};
SSingleWorkerCfg cfg = {.min = 1, .max = 1, .name = "vnode-mgmt", .fp = (FItem)vmProcessMgmtQueue, .param = pMgmt};
if (tSingleWorkerInit(&pMgmt->mgmtWorker, &cfg) != 0) {
dError("failed to start vnode-mgmt worker since %s", terrstr());
return -1;
......
......@@ -26,7 +26,7 @@
#include "parser.h"
#include "tname.h"
#define MND_TOPIC_VER_NUMBER 1
#define MND_TOPIC_VER_NUMBER 1
#define MND_TOPIC_RESERVE_SIZE 64
static int32_t mndTopicActionInsert(SSdb *pSdb, SMqTopicObj *pTopic);
......
......@@ -46,14 +46,11 @@ target_link_libraries(
PUBLIC stream
PUBLIC qworker
PUBLIC sync
# TODO: get rid of BDB
PUBLIC bdb
PUBLIC tdb
)
if(${META_DB_IMPL} STREQUAL "BDB")
target_link_libraries(vnode PUBLIC bdb)
elseif(${META_DB_IMPL} STREQUAL "TDB")
target_link_libraries(vnode PUBLIC tdb)
endif()
if(${BUILD_TEST})
add_subdirectory(test)
endif(${BUILD_TEST})
......@@ -100,10 +100,11 @@ int32_t tsdbUpdateSmaWindow(STsdb *pTsdb, const char *msg);
* @brief Insert tSma(Time-range-wise SMA) data from stream computing engine
*
* @param pTsdb
* @param indexUid
* @param msg
* @return int32_t
*/
int32_t tsdbInsertTSmaData(STsdb *pTsdb, char *msg);
int32_t tsdbInsertTSmaData(STsdb *pTsdb, int64_t indexUid, const char *msg);
/**
* @brief Drop tSma data and local cache.
......@@ -130,16 +131,11 @@ int32_t tsdbInsertRSmaData(STsdb *pTsdb, char *msg);
* @param pTsdb
* @param pData
* @param indexUid
* @param interval
* @param intervalUnit
* @param tableUid
* @param colId
* @param querySKey
* @param nMaxResult
* @return int32_t
*/
int32_t tsdbGetTSmaData(STsdb *pTsdb, STSmaDataWrapper *pData, int64_t indexUid, int64_t interval, int8_t intervalUnit,
tb_uid_t tableUid, col_id_t colId, TSKEY querySKey, int32_t nMaxResult);
int32_t tsdbGetTSmaData(STsdb *pTsdb, char *pData, int64_t indexUid, TSKEY querySKey, int32_t nMaxResult);
// STsdbCfg
int tsdbOptionsInit(STsdbCfg *);
......
......@@ -17,8 +17,9 @@
#define _TD_VNODE_H_
#include "os.h"
#include "trpc.h"
#include "tmsgcb.h"
#include "tqueue.h"
#include "trpc.h"
#include "meta.h"
#include "tarray.h"
......@@ -166,7 +167,7 @@ int vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg);
* @param pMsg The request message
* @return int 0 for success, -1 for failure
*/
int vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg);
int vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo);
/* ------------------------ SVnodeCfg ------------------------ */
/**
......@@ -185,7 +186,6 @@ void vnodeOptionsClear(SVnodeCfg *pOptions);
int vnodeValidateTableHash(SVnodeCfg *pVnodeOptions, char *tableFName);
/* ------------------------ FOR COMPILE ------------------------ */
int32_t vnodeAlter(SVnode *pVnode, const SVnodeCfg *pCfg);
......
......@@ -44,11 +44,10 @@ int32_t tsdbRemoveTSmaData(STsdb *pTsdb, STSma *param, STimeWindow *pWin);
#endif
// internal func
static FORCE_INLINE int32_t tsdbEncodeTSmaKey(tb_uid_t tableUid, col_id_t colId, TSKEY tsKey, void **pData) {
static FORCE_INLINE int32_t tsdbEncodeTSmaKey(int64_t groupId, TSKEY tsKey, void **pData) {
int32_t len = 0;
len += taosEncodeFixedI64(pData, tableUid);
len += taosEncodeFixedU16(pData, colId);
len += taosEncodeFixedI64(pData, tsKey);
len += taosEncodeFixedI64(pData, groupId);
return len;
}
......
......@@ -197,9 +197,9 @@ int tqCommit(STQ*);
int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg);
int32_t tqProcessSetConnReq(STQ* pTq, char* msg);
int32_t tqProcessRebReq(STQ* pTq, char* msg);
int32_t tqProcessTaskExec(STQ* pTq, char* msg, int32_t msgLen);
int32_t tqProcessTaskExec(STQ* pTq, char* msg, int32_t msgLen, int32_t workerId);
int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen);
int32_t tqProcessStreamTrigger(STQ* pTq, void* data, int32_t dataLen);
int32_t tqProcessStreamTrigger(STQ* pTq, void* data, int32_t dataLen, int32_t workerId);
// sma
void smaHandleRes(void* pVnode, int64_t smaId, const SArray* data);
......
......@@ -15,131 +15,637 @@
#include "metaDef.h"
#include "tdb.h"
#include "tdbInt.h"
struct SMetaDB {
TENV *pEnv;
TDB * pTbDB;
TDB * pSchemaDB;
TDB * pNameIdx;
TDB * pStbIdx;
TDB * pNtbIdx;
TDB * pCtbIdx;
// tag index hash table
// suid+colid --> TDB *
struct {
} tagIdxHt;
TDB *pTbDB;
TDB *pSchemaDB;
TDB *pNameIdx;
TDB *pStbIdx;
TDB *pNtbIdx;
TDB *pCtbIdx;
};
#define A(op, flag) \
do { \
if ((ret = op) != 0) goto flag; \
} while (0)
typedef struct __attribute__((__packed__)) {
tb_uid_t uid;
int32_t sver;
} SSchemaDbKey;
typedef struct {
char *name;
tb_uid_t uid;
} SNameIdxKey;
typedef struct {
tb_uid_t suid;
tb_uid_t uid;
} SCtbIdxKey;
static int metaEncodeTbInfo(void **buf, STbCfg *pTbCfg);
static void *metaDecodeTbInfo(void *buf, STbCfg *pTbCfg);
static int metaEncodeSchema(void **buf, SSchemaWrapper *pSW);
static void *metaDecodeSchema(void *buf, SSchemaWrapper *pSW);
static inline int metaUidCmpr(const void *arg1, int len1, const void *arg2, int len2) {
tb_uid_t uid1, uid2;
ASSERT(len1 == sizeof(tb_uid_t));
ASSERT(len2 == sizeof(tb_uid_t));
uid1 = ((tb_uid_t *)arg1)[0];
uid2 = ((tb_uid_t *)arg2)[0];
if (uid1 < uid2) {
return -1;
}
if (uid1 == uid2) {
return 0;
} else {
return 1;
}
}
static inline int metaSchemaKeyCmpr(const void *arg1, int len1, const void *arg2, int len2) {
int c;
SSchemaDbKey *pKey1 = (SSchemaDbKey *)arg1;
SSchemaDbKey *pKey2 = (SSchemaDbKey *)arg2;
c = metaUidCmpr(arg1, sizeof(tb_uid_t), arg2, sizeof(tb_uid_t));
if (c) return c;
if (pKey1->sver > pKey2->sver) {
return 1;
} else if (pKey1->sver == pKey2->sver) {
return 0;
} else {
return -1;
}
}
static inline int metaNameIdxCmpr(const void *arg1, int len1, const void *arg2, int len2) {
return strcmp((char *)arg1, (char *)arg2);
}
static inline int metaCtbIdxCmpr(const void *arg1, int len1, const void *arg2, int len2) {
int c;
SCtbIdxKey *pKey1 = (SCtbIdxKey *)arg1;
SCtbIdxKey *pKey2 = (SCtbIdxKey *)arg2;
c = metaUidCmpr(arg1, sizeof(tb_uid_t), arg2, sizeof(tb_uid_t));
if (c) return c;
return metaUidCmpr(&pKey1->uid, sizeof(tb_uid_t), &pKey2->uid, sizeof(tb_uid_t));
}
int metaOpenDB(SMeta *pMeta) {
SMetaDB *pDb;
TENV * pEnv;
TDB * pTbDB;
TDB * pSchemaDB;
TDB * pNameIdx;
TDB * pStbIdx;
TDB * pNtbIdx;
TDB * pCtbIdx;
SMetaDB *pMetaDb;
int ret;
pDb = (SMetaDB *)taosMemoryCalloc(1, sizeof(*pDb));
if (pDb == NULL) {
// allocate DB handle
pMetaDb = taosMemoryCalloc(1, sizeof(*pMetaDb));
if (pMetaDb == NULL) {
// TODO
ASSERT(0);
return -1;
}
// Create and open the ENV
A((tdbEnvCreate(&pEnv)), _err);
#if 0
// Set options of the environment
A(tdbEnvSetPageSize(pEnv, 8192), _err);
A(tdbEnvSetCacheSize(pEnv, 16 * 1024 * 1024), _err);
#endif
A((tdbEnvOpen(&pEnv)), _err);
// open the ENV
ret = tdbEnvOpen(pMeta->path, 4096, 256, &(pMetaDb->pEnv));
if (ret < 0) {
// TODO
ASSERT(0);
return -1;
}
// Create and open each DB
A(tdbCreate(&pTbDB), _err);
A(tdbOpen(&pTbDB, "table.db", NULL, pEnv), _err);
// open table DB
ret = tdbDbOpen("table.db", sizeof(tb_uid_t), TDB_VARIANT_LEN, metaUidCmpr, pMetaDb->pEnv, &(pMetaDb->pTbDB));
if (ret < 0) {
// TODO
ASSERT(0);
return -1;
}
A(tdbCreate(&pSchemaDB), _err);
A(tdbOpen(&pSchemaDB, "schema.db", NULL, pEnv), _err);
// open schema DB
ret = tdbDbOpen("schema.db", sizeof(SSchemaDbKey), TDB_VARIANT_LEN, metaSchemaKeyCmpr, pMetaDb->pEnv,
&(pMetaDb->pSchemaDB));
if (ret < 0) {
// TODO
ASSERT(0);
return -1;
}
A(tdbCreate(&pNameIdx), _err);
A(tdbOpen(&pNameIdx, "name.db", NULL, pEnv), _err);
// tdbAssociate();
ret = tdbDbOpen("name.idx", TDB_VARIANT_LEN, 0, metaNameIdxCmpr, pMetaDb->pEnv, &(pMetaDb->pNameIdx));
if (ret < 0) {
// TODO
ASSERT(0);
return -1;
}
pDb->pEnv = pEnv;
pDb->pTbDB = pTbDB;
pDb->pSchemaDB = pSchemaDB;
pMeta->pDB = pDb;
return 0;
ret = tdbDbOpen("stb.idx", sizeof(tb_uid_t), 0, metaUidCmpr, pMetaDb->pEnv, &(pMetaDb->pStbIdx));
if (ret < 0) {
// TODO
ASSERT(0);
return -1;
}
ret = tdbDbOpen("ntb.idx", sizeof(tb_uid_t), 0, metaUidCmpr, pMetaDb->pEnv, &(pMetaDb->pNtbIdx));
if (ret < 0) {
// TODO
ASSERT(0);
return -1;
}
_err:
return -1;
ret = tdbDbOpen("ctb.idx", sizeof(SCtbIdxKey), 0, metaCtbIdxCmpr, pMetaDb->pEnv, &(pMetaDb->pCtbIdx));
if (ret < 0) {
// TODO
ASSERT(0);
return -1;
}
pMeta->pDB = pMetaDb;
return 0;
}
void metaCloseDB(SMeta *pMeta) {
// TODO
if (pMeta->pDB) {
tdbDbClose(pMeta->pDB->pCtbIdx);
tdbDbClose(pMeta->pDB->pNtbIdx);
tdbDbClose(pMeta->pDB->pStbIdx);
tdbDbClose(pMeta->pDB->pNameIdx);
tdbDbClose(pMeta->pDB->pSchemaDB);
tdbDbClose(pMeta->pDB->pTbDB);
taosMemoryFree(pMeta->pDB);
}
}
int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg) {
// TODO
tb_uid_t uid;
SMetaDB *pMetaDb;
void *pKey;
void *pVal;
int kLen;
int vLen;
int ret;
char buf[512];
void *pBuf;
SCtbIdxKey ctbIdxKey;
SSchemaDbKey schemaDbKey;
SSchemaWrapper schemaWrapper;
pMetaDb = pMeta->pDB;
// TODO: make this operation pre-process
if (pTbCfg->type == META_SUPER_TABLE) {
uid = pTbCfg->stbCfg.suid;
} else {
uid = metaGenerateUid(pMeta);
}
// save to table.db
pKey = &uid;
kLen = sizeof(uid);
pVal = pBuf = buf;
metaEncodeTbInfo(&pBuf, pTbCfg);
vLen = POINTER_DISTANCE(pBuf, buf);
ret = tdbDbInsert(pMetaDb->pTbDB, pKey, kLen, pVal, vLen);
if (ret < 0) {
return -1;
}
// save to schema.db for META_SUPER_TABLE and META_NORMAL_TABLE
if (pTbCfg->type != META_CHILD_TABLE) {
schemaDbKey.uid = uid;
schemaDbKey.sver = 0; // TODO
pKey = &schemaDbKey;
kLen = sizeof(schemaDbKey);
if (pTbCfg->type == META_SUPER_TABLE) {
schemaWrapper.nCols = pTbCfg->stbCfg.nCols;
schemaWrapper.pSchema = pTbCfg->stbCfg.pSchema;
} else {
schemaWrapper.nCols = pTbCfg->ntbCfg.nCols;
schemaWrapper.pSchema = pTbCfg->ntbCfg.pSchema;
}
pVal = pBuf = buf;
metaEncodeSchema(&pBuf, &schemaWrapper);
vLen = POINTER_DISTANCE(pBuf, buf);
ret = tdbDbInsert(pMetaDb->pSchemaDB, pKey, kLen, pVal, vLen);
if (ret < 0) {
return -1;
}
}
// update name.idx
int nameLen = strlen(pTbCfg->name);
memcpy(buf, pTbCfg->name, nameLen + 1);
((tb_uid_t *)(buf + nameLen + 1))[0] = uid;
pKey = buf;
kLen = nameLen + 1 + sizeof(uid);
pVal = NULL;
vLen = 0;
ret = tdbDbInsert(pMetaDb->pNameIdx, pKey, kLen, pVal, vLen);
if (ret < 0) {
return -1;
}
// update other index
if (pTbCfg->type == META_SUPER_TABLE) {
pKey = &uid;
kLen = sizeof(uid);
pVal = NULL;
vLen = 0;
ret = tdbDbInsert(pMetaDb->pStbIdx, pKey, kLen, pVal, vLen);
if (ret < 0) {
return -1;
}
} else if (pTbCfg->type == META_CHILD_TABLE) {
ctbIdxKey.suid = pTbCfg->ctbCfg.suid;
ctbIdxKey.uid = uid;
pKey = &ctbIdxKey;
kLen = sizeof(ctbIdxKey);
pVal = NULL;
vLen = 0;
ret = tdbDbInsert(pMetaDb->pCtbIdx, pKey, kLen, pVal, vLen);
if (ret < 0) {
return -1;
}
} else if (pTbCfg->type == META_NORMAL_TABLE) {
pKey = &uid;
kLen = sizeof(uid);
pVal = NULL;
vLen = 0;
ret = tdbDbInsert(pMetaDb->pNtbIdx, pKey, kLen, pVal, vLen);
if (ret < 0) {
return -1;
}
}
return 0;
}
int metaRemoveTableFromDb(SMeta *pMeta, tb_uid_t uid) {
// TODO
ASSERT(0);
return 0;
}
STbCfg *metaGetTbInfoByUid(SMeta *pMeta, tb_uid_t uid) {
// TODO
return NULL;
int ret;
SMetaDB *pMetaDb = pMeta->pDB;
void *pKey;
void *pVal;
int kLen;
int vLen;
STbCfg *pTbCfg;
// Fetch
pKey = &uid;
kLen = sizeof(uid);
pVal = NULL;
ret = tdbDbGet(pMetaDb->pTbDB, pKey, kLen, &pVal, &vLen);
if (ret < 0) {
return NULL;
}
// Decode
pTbCfg = taosMemoryMalloc(sizeof(*pTbCfg));
metaDecodeTbInfo(pVal, pTbCfg);
TDB_FREE(pVal);
return pTbCfg;
}
STbCfg *metaGetTbInfoByName(SMeta *pMeta, char *tbname, tb_uid_t *uid) {
// TODO
return NULL;
void *pKey;
void *pVal;
void *ppKey;
int pkLen;
int kLen;
int vLen;
int ret;
pKey = tbname;
kLen = strlen(tbname) + 1;
pVal = NULL;
ppKey = NULL;
ret = tdbDbPGet(pMeta->pDB->pNameIdx, pKey, kLen, &ppKey, &pkLen, &pVal, &vLen);
if (ret < 0) {
return NULL;
}
ASSERT(pkLen == kLen + sizeof(uid));
*uid = *(tb_uid_t *)POINTER_SHIFT(ppKey, kLen);
TDB_FREE(ppKey);
TDB_FREE(pVal);
return metaGetTbInfoByUid(pMeta, *uid);
}
SSchemaWrapper *metaGetTableSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, bool isinline) {
// TODO
return NULL;
void *pKey;
void *pVal;
int kLen;
int vLen;
int ret;
SSchemaDbKey schemaDbKey;
SSchemaWrapper *pSchemaWrapper;
void *pBuf;
// fetch
schemaDbKey.uid = uid;
schemaDbKey.sver = sver;
pKey = &schemaDbKey;
kLen = sizeof(schemaDbKey);
pVal = NULL;
ret = tdbDbGet(pMeta->pDB->pSchemaDB, pKey, kLen, &pVal, &vLen);
if (ret < 0) {
return NULL;
}
// decode
pBuf = pVal;
pSchemaWrapper = taosMemoryMalloc(sizeof(*pSchemaWrapper));
metaDecodeSchema(pBuf, pSchemaWrapper);
TDB_FREE(pVal);
return pSchemaWrapper;
}
STSchema *metaGetTbTSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver) {
// TODO
return NULL;
tb_uid_t quid;
SSchemaWrapper *pSW;
STSchemaBuilder sb;
SSchema *pSchema;
STSchema *pTSchema;
STbCfg *pTbCfg;
pTbCfg = metaGetTbInfoByUid(pMeta, uid);
if (pTbCfg->type == META_CHILD_TABLE) {
quid = pTbCfg->ctbCfg.suid;
} else {
quid = uid;
}
pSW = metaGetTableSchema(pMeta, quid, sver, true);
if (pSW == NULL) {
return NULL;
}
tdInitTSchemaBuilder(&sb, 0);
for (int i = 0; i < pSW->nCols; i++) {
pSchema = pSW->pSchema + i;
tdAddColToSchema(&sb, pSchema->type, pSchema->colId, pSchema->bytes);
}
pTSchema = tdGetSchemaFromBuilder(&sb);
tdDestroyTSchemaBuilder(&sb);
return pTSchema;
}
struct SMTbCursor {
TDBC *pDbc;
};
SMTbCursor *metaOpenTbCursor(SMeta *pMeta) {
// TODO
return NULL;
SMTbCursor *pTbCur = NULL;
SMetaDB *pDB = pMeta->pDB;
pTbCur = (SMTbCursor *)taosMemoryCalloc(1, sizeof(*pTbCur));
if (pTbCur == NULL) {
return NULL;
}
tdbDbcOpen(pDB->pTbDB, &pTbCur->pDbc);
return pTbCur;
}
void metaCloseTbCursor(SMTbCursor *pTbCur) {
// TODO
if (pTbCur) {
if (pTbCur->pDbc) {
tdbDbcClose(pTbCur->pDbc);
}
taosMemoryFree(pTbCur);
}
}
char *metaTbCursorNext(SMTbCursor *pTbCur) {
// TODO
void *pKey = NULL;
void *pVal = NULL;
int kLen;
int vLen;
int ret;
void *pBuf;
STbCfg tbCfg;
for (;;) {
ret = tdbDbNext(pTbCur->pDbc, &pKey, &kLen, &pVal, &vLen);
if (ret < 0) break;
pBuf = pVal;
metaDecodeTbInfo(pBuf, &tbCfg);
if (tbCfg.type == META_SUPER_TABLE) {
taosMemoryFree(tbCfg.name);
taosMemoryFree(tbCfg.stbCfg.pTagSchema);
continue;
;
} else if (tbCfg.type == META_CHILD_TABLE) {
kvRowFree(tbCfg.ctbCfg.pTag);
}
return tbCfg.name;
}
return NULL;
}
struct SMCtbCursor {
TDBC *pCur;
tb_uid_t suid;
void *pKey;
void *pVal;
int kLen;
int vLen;
};
SMCtbCursor *metaOpenCtbCursor(SMeta *pMeta, tb_uid_t uid) {
SMCtbCursor *pCtbCur = NULL;
SMetaDB *pDB = pMeta->pDB;
int ret;
pCtbCur = (SMCtbCursor *)taosMemoryCalloc(1, sizeof(*pCtbCur));
if (pCtbCur == NULL) {
return NULL;
}
pCtbCur->suid = uid;
ret = tdbDbcOpen(pDB->pCtbIdx, &pCtbCur->pCur);
if (ret < 0) {
taosMemoryFree(pCtbCur);
return NULL;
}
// TODO: move the cursor to the suid there
return pCtbCur;
}
void metaCloseCtbCurosr(SMCtbCursor *pCtbCur) {
if (pCtbCur) {
if (pCtbCur->pCur) {
tdbDbcClose(pCtbCur->pCur);
TDB_FREE(pCtbCur->pKey);
TDB_FREE(pCtbCur->pVal);
}
taosMemoryFree(pCtbCur);
}
}
tb_uid_t metaCtbCursorNext(SMCtbCursor *pCtbCur) {
int ret;
SCtbIdxKey *pCtbIdxKey;
ret = tdbDbNext(pCtbCur->pCur, &pCtbCur->pKey, &pCtbCur->kLen, &pCtbCur->pVal, &pCtbCur->vLen);
if (ret < 0) {
return 0;
}
pCtbIdxKey = pCtbCur->pVal;
return pCtbIdxKey->uid;
}
int metaGetTbNum(SMeta *pMeta) {
// TODO
// ASSERT(0);
return 0;
}
STSmaWrapper *metaGetSmaInfoByTable(SMeta *pMeta, tb_uid_t uid) {
// TODO
ASSERT(0);
return NULL;
}
void metaCloseCtbCurosr(SMCtbCursor *pCtbCur) {
int metaRemoveSmaFromDb(SMeta *pMeta, int64_t indexUid) {
// TODO
ASSERT(0);
return 0;
}
tb_uid_t metaCtbCursorNext(SMCtbCursor *pCtbCur) {
int metaSaveSmaToDB(SMeta *pMeta, STSma *pSmaCfg) {
// TODO
ASSERT(0);
return 0;
}
\ No newline at end of file
}
STSma *metaGetSmaInfoByIndex(SMeta *pMeta, int64_t indexUid) {
// TODO
ASSERT(0);
return NULL;
}
const char *metaSmaCursorNext(SMSmaCursor *pCur) {
// TODO
ASSERT(0);
return NULL;
}
void metaCloseSmaCurosr(SMSmaCursor *pCur) {
// TODO
ASSERT(0);
}
SArray *metaGetSmaTbUids(SMeta *pMeta, bool isDup) {
// TODO
ASSERT(0);
return NULL;
}
SMSmaCursor *metaOpenSmaCursor(SMeta *pMeta, tb_uid_t uid) {
// TODO
ASSERT(0);
return NULL;
}
static int metaEncodeSchema(void **buf, SSchemaWrapper *pSW) {
int tlen = 0;
SSchema *pSchema;
tlen += taosEncodeFixedU32(buf, pSW->nCols);
for (int i = 0; i < pSW->nCols; i++) {
pSchema = pSW->pSchema + i;
tlen += taosEncodeFixedI8(buf, pSchema->type);
tlen += taosEncodeFixedI16(buf, pSchema->colId);
tlen += taosEncodeFixedI32(buf, pSchema->bytes);
tlen += taosEncodeString(buf, pSchema->name);
}
return tlen;
}
static void *metaDecodeSchema(void *buf, SSchemaWrapper *pSW) {
SSchema *pSchema;
buf = taosDecodeFixedU32(buf, &pSW->nCols);
pSW->pSchema = (SSchema *)taosMemoryMalloc(sizeof(SSchema) * pSW->nCols);
for (int i = 0; i < pSW->nCols; i++) {
pSchema = pSW->pSchema + i;
buf = taosDecodeFixedI8(buf, &pSchema->type);
buf = taosDecodeFixedI16(buf, &pSchema->colId);
buf = taosDecodeFixedI32(buf, &pSchema->bytes);
buf = taosDecodeStringTo(buf, pSchema->name);
}
return buf;
}
static int metaEncodeTbInfo(void **buf, STbCfg *pTbCfg) {
int tsize = 0;
tsize += taosEncodeString(buf, pTbCfg->name);
tsize += taosEncodeFixedU32(buf, pTbCfg->ttl);
tsize += taosEncodeFixedU32(buf, pTbCfg->keep);
tsize += taosEncodeFixedU8(buf, pTbCfg->info);
if (pTbCfg->type == META_SUPER_TABLE) {
SSchemaWrapper sw = {.nCols = pTbCfg->stbCfg.nTagCols, .pSchema = pTbCfg->stbCfg.pTagSchema};
tsize += metaEncodeSchema(buf, &sw);
} else if (pTbCfg->type == META_CHILD_TABLE) {
tsize += taosEncodeFixedU64(buf, pTbCfg->ctbCfg.suid);
tsize += tdEncodeKVRow(buf, pTbCfg->ctbCfg.pTag);
} else if (pTbCfg->type == META_NORMAL_TABLE) {
// TODO
} else {
ASSERT(0);
}
return tsize;
}
static void *metaDecodeTbInfo(void *buf, STbCfg *pTbCfg) {
buf = taosDecodeString(buf, &(pTbCfg->name));
buf = taosDecodeFixedU32(buf, &(pTbCfg->ttl));
buf = taosDecodeFixedU32(buf, &(pTbCfg->keep));
buf = taosDecodeFixedU8(buf, &(pTbCfg->info));
if (pTbCfg->type == META_SUPER_TABLE) {
SSchemaWrapper sw;
buf = metaDecodeSchema(buf, &sw);
pTbCfg->stbCfg.nTagCols = sw.nCols;
pTbCfg->stbCfg.pTagSchema = sw.pSchema;
} else if (pTbCfg->type == META_CHILD_TABLE) {
buf = taosDecodeFixedU64(buf, &(pTbCfg->ctbCfg.suid));
buf = tdDecodeKVRow(buf, &(pTbCfg->ctbCfg.pTag));
} else if (pTbCfg->type == META_NORMAL_TABLE) {
// TODO
} else {
ASSERT(0);
}
return buf;
}
......@@ -356,7 +356,6 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
void* buf = rpcMallocCont(tlen);
if (buf == NULL) {
pMsg->code = -1;
ASSERT(0);
return -1;
}
((SMqRspHead*)buf)->mqMsgType = TMQ_MSG_TYPE__POLL_RSP;
......@@ -490,7 +489,7 @@ int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen) {
return 0;
}
int32_t tqProcessStreamTrigger(STQ* pTq, void* data, int32_t dataLen) {
int32_t tqProcessStreamTrigger(STQ* pTq, void* data, int32_t dataLen, int32_t workerId) {
void* pIter = NULL;
while (1) {
......@@ -498,14 +497,14 @@ int32_t tqProcessStreamTrigger(STQ* pTq, void* data, int32_t dataLen) {
if (pIter == NULL) break;
SStreamTask* pTask = (SStreamTask*)pIter;
if (streamExecTask(pTask, &pTq->pVnode->msgCb, data, STREAM_DATA_TYPE_SUBMIT_BLOCK, 0) < 0) {
if (streamExecTask(pTask, &pTq->pVnode->msgCb, data, STREAM_DATA_TYPE_SUBMIT_BLOCK, workerId) < 0) {
// TODO
}
}
return 0;
}
int32_t tqProcessTaskExec(STQ* pTq, char* msg, int32_t msgLen) {
int32_t tqProcessTaskExec(STQ* pTq, char* msg, int32_t msgLen, int32_t workerId) {
SStreamTaskExecReq req;
tDecodeSStreamTaskExecReq(msg, &req);
......@@ -515,7 +514,7 @@ int32_t tqProcessTaskExec(STQ* pTq, char* msg, int32_t msgLen) {
SStreamTask* pTask = taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t));
ASSERT(pTask);
if (streamExecTask(pTask, &pTq->pVnode->msgCb, req.data, STREAM_DATA_TYPE_SSDATA_BLOCK, 0) < 0) {
if (streamExecTask(pTask, &pTq->pVnode->msgCb, req.data, STREAM_DATA_TYPE_SSDATA_BLOCK, workerId) < 0) {
// TODO
}
return 0;
......
......@@ -167,8 +167,10 @@ SArray* tqRetrieveDataBlock(STqReadHandle* pHandle) {
if (!tdSTSRowIterNext(&iter, pColData->info.colId, pColData->info.type, &sVal)) {
break;
}
// TODO handle null
colDataAppend(pColData, curRow, sVal.val, sVal.valType == TD_VTYPE_NULL);
if (colDataAppend(pColData, curRow, sVal.val, sVal.valType == TD_VTYPE_NULL) < 0) {
taosArrayDestroyEx(pArray, (void (*)(void*))tDeleteSSDataBlock);
return NULL;
}
}
curRow++;
}
......
......@@ -41,7 +41,7 @@ int vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg) {
}
}
int vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg) {
int vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
vTrace("message in fetch queue is processing");
char *msgstr = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
int32_t msgLen = pMsg->contLen - sizeof(SMsgHead);
......@@ -69,9 +69,9 @@ int vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg) {
return tqProcessPollReq(pVnode->pTq, pMsg);
case TDMT_VND_TASK_PIPE_EXEC:
case TDMT_VND_TASK_MERGE_EXEC:
return tqProcessTaskExec(pVnode->pTq, msgstr, msgLen);
return tqProcessTaskExec(pVnode->pTq, msgstr, msgLen, pInfo->workerId);
case TDMT_VND_STREAM_TRIGGER:
return tqProcessStreamTrigger(pVnode->pTq, pMsg->pCont, pMsg->contLen);
return tqProcessStreamTrigger(pVnode->pTq, pMsg->pCont, pMsg->contLen, pInfo->workerId);
case TDMT_VND_QUERY_HEARTBEAT:
return qWorkerProcessHbMsg(pVnode, pVnode->pQuery, pMsg);
default:
......
......@@ -17,7 +17,9 @@
void smaHandleRes(void *pVnode, int64_t smaId, const SArray *data) {
// TODO
blockDebugShowData(data);
tsdbInsertTSmaData(((SVnode *)pVnode)->pTsdb, smaId, (const char *)data);
}
void vnodeProcessWMsgs(SVnode *pVnode, SArray *pMsgs) {
......@@ -184,24 +186,25 @@ int vnodeApplyWMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
}
} break;
case TDMT_VND_TASK_WRITE_EXEC: {
if (tqProcessTaskExec(pVnode->pTq, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)),
pMsg->contLen - sizeof(SMsgHead)) < 0) {
if (tqProcessTaskExec(pVnode->pTq, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)), pMsg->contLen - sizeof(SMsgHead),
0) < 0) {
}
} break;
case TDMT_VND_CREATE_SMA: { // timeRangeSMA
#if 1
SSmaCfg vCreateSmaReq = {0};
if (tDeserializeSVCreateTSmaReq(POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)), &vCreateSmaReq) == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
vWarn("vgId%d: TDMT_VND_CREATE_SMA received but deserialize failed since %s", pVnode->config.vgId, terrstr(terrno));
vWarn("vgId%d: TDMT_VND_CREATE_SMA received but deserialize failed since %s", pVnode->config.vgId,
terrstr(terrno));
return -1;
}
vWarn("vgId%d: TDMT_VND_CREATE_SMA received for %s:%" PRIi64, pVnode->config.vgId, vCreateSmaReq.tSma.indexName,
vCreateSmaReq.tSma.indexUid);
// record current timezone of server side
tstrncpy(vCreateSmaReq.tSma.timezone, tsTimezoneStr, TD_TIMEZONE_LEN);
vCreateSmaReq.tSma.timezoneInt = tsTimezone;
if (metaCreateTSma(pVnode->pMeta, &vCreateSmaReq) < 0) {
// TODO: handle error
......
......@@ -15,6 +15,7 @@
#include <gtest/gtest.h>
#include <tsdbDef.h>
#include <taoserror.h>
#include <tglobal.h>
#include <iostream>
......@@ -280,7 +281,7 @@ TEST(testCase, tSma_metaDB_Put_Get_Del_Test) {
}
#endif
#if 0
#if 1
TEST(testCase, tSma_Data_Insert_Query_Test) {
// step 1: prepare meta
const char *smaIndexName1 = "sma_index_test_1";
......@@ -299,9 +300,9 @@ TEST(testCase, tSma_Data_Insert_Query_Test) {
// encode
STSma tSma = {0};
tSma.version = 0;
tSma.intervalUnit = TIME_UNIT_DAY;
tSma.intervalUnit = TIME_UNIT_MINUTE;
tSma.interval = 1;
tSma.slidingUnit = TIME_UNIT_HOUR;
tSma.slidingUnit = TIME_UNIT_MINUTE;
tSma.sliding = 1; // sliding = interval when it's convert window
tSma.indexUid = indexUid1;
tstrncpy(tSma.indexName, smaIndexName1, TSDB_INDEX_NAME_LEN);
......@@ -330,8 +331,8 @@ TEST(testCase, tSma_Data_Insert_Query_Test) {
ASSERT_EQ(metaSaveSmaToDB(pMeta, pSmaCfg), 0);
// step 2: insert data
STsdb *pTsdb = (STsdb *)taosMemoryCalloc(1, sizeof(STsdb));
STsdbCfg *pCfg = &pTsdb->config;
STsdb *pTsdb = (STsdb *)taosMemoryCalloc(1, sizeof(STsdb));
STsdbCfg *pCfg = &pTsdb->config;
pTsdb->pMeta = pMeta;
pTsdb->vgId = 2;
......@@ -405,15 +406,94 @@ TEST(testCase, tSma_Data_Insert_Query_Test) {
ASSERT_EQ(tsdbUpdateSmaWindow(pTsdb, (const char *)pMsg), 0);
// init
int32_t allocCnt = 0;
int32_t allocStep = 16384;
int32_t buffer = 1024;
void *buf = NULL;
ASSERT_EQ(tsdbMakeRoom(&buf, allocStep), 0);
int32_t bufSize = taosTSizeof(buf);
int32_t numOfTables = 10;
col_id_t numOfCols = 4096;
ASSERT_GT(numOfCols, 0);
const int32_t tSmaGroupSize = 4;
const int32_t tSmaNumOfTags = 2;
const int64_t tSmaGroupId = 12345670;
const col_id_t tSmaNumOfCols = 9; // binary/nchar/varbinary/varchar are only used for tags for group by conditions.
const int32_t tSmaNumOfRows = 2;
SArray *pDataBlocks = taosArrayInit(tSmaGroupSize, sizeof(SSDataBlock *));
ASSERT_NE(pDataBlocks, nullptr);
int32_t tSmaTypeArray[tSmaNumOfCols] = {TSDB_DATA_TYPE_TIMESTAMP, TSDB_DATA_TYPE_BOOL, TSDB_DATA_TYPE_INT,
TSDB_DATA_TYPE_UBIGINT, TSDB_DATA_TYPE_SMALLINT, TSDB_DATA_TYPE_FLOAT,
TSDB_DATA_TYPE_DOUBLE, TSDB_DATA_TYPE_VARCHAR, TSDB_DATA_TYPE_NCHAR};
// last 2 columns for group by tags
// int32_t tSmaTypeArray[tSmaNumOfCols] = {TSDB_DATA_TYPE_TIMESTAMP, TSDB_DATA_TYPE_BOOL};
const char *tSmaGroupbyTags[tSmaGroupSize * tSmaNumOfTags] = {"BeiJing", "HaiDian", "BeiJing", "ChaoYang",
"ShangHai", "PuDong", "ShangHai", "MinHang"};
TSKEY tSmaSKeyMs = (int64_t)1648535332 * 1000;
int64_t tSmaIntervalMs = tSma.interval * 60 * 1000;
int64_t tSmaInitVal = 0;
for (int32_t g = 0; g < tSmaGroupSize; ++g) {
SSDataBlock *pDataBlock = (SSDataBlock *)taosMemoryCalloc(1, sizeof(SSDataBlock));
ASSERT_NE(pDataBlock, nullptr);
pDataBlock->pBlockAgg = NULL;
pDataBlock->info.numOfCols = tSmaNumOfCols;
pDataBlock->info.rows = tSmaNumOfRows;
pDataBlock->info.groupId = tSmaGroupId + g;
pDataBlock->pDataBlock = taosArrayInit(tSmaNumOfCols, sizeof(SColumnInfoData *));
ASSERT_NE(pDataBlock->pDataBlock, nullptr);
for (int32_t c = 0; c < tSmaNumOfCols; ++c) {
SColumnInfoData *pColInfoData = (SColumnInfoData *)taosMemoryCalloc(1, sizeof(SColumnInfoData));
ASSERT_NE(pColInfoData, nullptr);
pColInfoData->info.type = tSmaTypeArray[c];
if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) {
pColInfoData->info.bytes = 100; // update accordingly
} else {
pColInfoData->info.bytes = TYPE_BYTES[pColInfoData->info.type];
}
pColInfoData->pData = (char *)taosMemoryCalloc(1, tSmaNumOfRows * pColInfoData->info.bytes);
for (int32_t r = 0; r < tSmaNumOfRows; ++r) {
void *pCellData = pColInfoData->pData + r * pColInfoData->info.bytes;
switch (pColInfoData->info.type) {
case TSDB_DATA_TYPE_TIMESTAMP:
*(TSKEY *)pCellData = tSmaSKeyMs + tSmaIntervalMs * r;
break;
case TSDB_DATA_TYPE_BOOL:
*(bool *)pCellData = (bool)tSmaInitVal++;
break;
case TSDB_DATA_TYPE_INT:
*(int *)pCellData = (int)tSmaInitVal++;
break;
case TSDB_DATA_TYPE_UBIGINT:
*(uint64_t *)pCellData = (uint64_t)tSmaInitVal++;
break;
case TSDB_DATA_TYPE_SMALLINT:
*(int16_t *)pCellData = (int16_t)tSmaInitVal++;
break;
case TSDB_DATA_TYPE_FLOAT:
*(float *)pCellData = (float)tSmaInitVal++;
break;
case TSDB_DATA_TYPE_DOUBLE:
*(double *)pCellData = (double)tSmaInitVal++;
break;
case TSDB_DATA_TYPE_VARCHAR: // city
varDataSetLen(pCellData, strlen(tSmaGroupbyTags[g * 2]));
memcpy(varDataVal(pCellData), tSmaGroupbyTags[g * 2], varDataLen(pCellData));
break;
case TSDB_DATA_TYPE_NCHAR: // district
varDataSetLen(pCellData, strlen(tSmaGroupbyTags[g * 2 + 1]));
memcpy(varDataVal(pCellData), tSmaGroupbyTags[g * 2 + 1], varDataLen(pCellData));
break;
default:
ASSERT_EQ(0, 1); // add definition
break;
}
}
// push SColumnInfoData
taosArrayPush(pDataBlock->pDataBlock, &pColInfoData);
}
// push SSDataBlock
taosArrayPush(pDataBlocks, &pDataBlock);
}
// execute
ASSERT_EQ(tsdbInsertTSmaData(pTsdb, tSma.indexUid, (const char *)pDataBlocks), TSDB_CODE_SUCCESS);
#if 0
STSmaDataWrapper *pSmaData = NULL;
......@@ -464,26 +544,30 @@ TEST(testCase, tSma_Data_Insert_Query_Test) {
ASSERT_EQ(tsdbInsertTSmaData(pTsdb, (char *)pSmaData), TSDB_CODE_SUCCESS);
#endif
SSDataBlock *pSmaData = (SSDataBlock *)taosMemoryCalloc(1, sizeof(SSDataBlock));
// step 3: query
uint32_t checkDataCnt = 0;
for (int32_t t = 0; t < numOfTables; ++t) {
for (col_id_t c = 0; c < numOfCols; ++c) {
ASSERT_EQ(tsdbGetTSmaData(pTsdb, NULL, indexUid1, interval1, intervalUnit1, tbUid + t,
c + PRIMARYKEY_TIMESTAMP_COL_ID, skey1, 1),
TSDB_CODE_SUCCESS);
++checkDataCnt;
}
}
ASSERT_EQ(tsdbGetTSmaData(pTsdb, NULL, indexUid1, skey1, 1), TSDB_CODE_SUCCESS);
++checkDataCnt;
printf("%s:%d The sma data check count for insert and query is %" PRIu32 "\n", __FILE__, __LINE__, checkDataCnt);
// release data
taosMemoryFreeClear(pMsg);
taosTZfree(buf);
for (int32_t i = 0; i < taosArrayGetSize(pDataBlocks); ++i) {
SSDataBlock *pDataBlock = (SSDataBlock *)taosArrayGet(pDataBlocks, i);
int32_t numOfOutput = taosArrayGetSize(pDataBlock->pDataBlock);
for (int32_t j = 0; j < numOfOutput; ++j) {
SColumnInfoData *pColInfoData = (SColumnInfoData *)taosArrayGet(pDataBlock->pDataBlock, j);
colDataDestroy(pColInfoData);
}
taosArrayDestroy(pDataBlock->pDataBlock);
taosMemoryFreeClear(pDataBlock->pBlockAgg);
taosMemoryFreeClear(pDataBlock);
}
taosArrayDestroy(pDataBlocks);
// release meta
tdDestroyTSma(&tSma);
tfsClose(pTsdb->pTfs);
......
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com> *
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
......@@ -16,7 +15,7 @@
#define __INDEX_CACHE_H__
#include "indexInt.h"
#include "index_util.h"
#include "indexUtil.h"
#include "tskiplist.h"
// ----------------- key structure in skiplist ---------------------
......
......@@ -20,12 +20,12 @@
extern "C" {
#endif
#include "indexFstAutomation.h"
#include "indexFstCountingWriter.h"
#include "indexFstNode.h"
#include "indexFstRegistry.h"
#include "indexFstUtil.h"
#include "indexInt.h"
#include "index_fst_automation.h"
#include "index_fst_counting_writer.h"
#include "index_fst_node.h"
#include "index_fst_registry.h"
#include "index_fst_util.h"
#define OUTPUT_PREFIX(a, b) ((a) > (b) ? (b) : (a)
......
......@@ -19,8 +19,8 @@
extern "C" {
#endif
#include "indexFstUtil.h"
#include "indexInt.h"
#include "index_fst_util.h"
typedef struct AutomationCtx AutomationCtx;
......
......@@ -20,9 +20,9 @@
extern "C" {
#endif
#include "indexFstCountingWriter.h"
#include "indexFstUtil.h"
#include "indexInt.h"
#include "index_fst_counting_writer.h"
#include "index_fst_util.h"
#define FST_BUILDER_NODE_IS_FINAL(bn) (bn->isFinal)
#define FST_BUILDER_NODE_TRANS_ISEMPTY(bn) (taosArrayGetSize(bn->trans) == 0)
......
......@@ -19,9 +19,9 @@
extern "C" {
#endif
#include "indexFstNode.h"
#include "indexFstUtil.h"
#include "indexInt.h"
#include "index_fst_node.h"
#include "index_fst_util.h"
typedef struct FstRegistryCell {
CompiledAddr addr;
......
......@@ -20,8 +20,8 @@
extern "C" {
#endif
#include "indexFstCommon.h"
#include "indexInt.h"
#include "index_fst_common.h"
typedef uint64_t FstType;
typedef uint64_t CompiledAddr;
......
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_INDEX_SPARSE_H_
#define _TD_INDEX_SPARSE_H_
#include "tarray.h"
#ifdef __cplusplus
extern "C" {
#endif
typedef struct FstSparseSet {
SArray *dense;
SArray *sparse;
int32_t size;
} FstSparseSet;
FstSparseSet *sparSetCreate(int32_t sz);
void sparSetDestroy(FstSparseSet *s);
uint32_t sparSetLen(FstSparseSet *ss);
uint32_t sparSetAdd(FstSparseSet *ss, uint32_t ip);
uint32_t sparSetGet(FstSparseSet *ss, uint32_t i);
bool sparSetContains(FstSparseSet *ss, uint32_t ip);
void sparSetClear(FstSparseSet *ss);
#ifdef __cplusplus
}
#endif
#endif
......@@ -15,11 +15,11 @@
#ifndef __INDEX_TFILE_H__
#define __INDEX_TFILE_H__
#include "indexFst.h"
#include "indexFstCountingWriter.h"
#include "indexInt.h"
#include "index_fst.h"
#include "index_fst_counting_writer.h"
#include "index_tfile.h"
#include "index_util.h"
#include "indexTfile.h"
#include "indexUtil.h"
#include "tlockfree.h"
#ifdef __cplusplus
......
......@@ -14,11 +14,11 @@
*/
#include "index.h"
#include "indexCache.h"
#include "indexComm.h"
#include "indexInt.h"
#include "index_cache.h"
#include "index_comm.h"
#include "index_tfile.h"
#include "index_util.h"
#include "indexTfile.h"
#include "indexUtil.h"
#include "tdef.h"
#include "tsched.h"
......@@ -236,11 +236,11 @@ int indexDelete(SIndex* index, SIndexMultiTermQuery* query) {
return 1;
}
int indexRebuild(SIndex* index, SIndexOpts* opts){
int indexRebuild(SIndex* index, SIndexOpts* opts) {
#ifdef USE_INVERTED_INDEX
#endif
return 0;
return 0;
}
SIndexOpts* indexOptsCreate() {
......
......@@ -13,9 +13,9 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "index_cache.h"
#include "index_comm.h"
#include "index_util.h"
#include "indexCache.h"
#include "indexComm.h"
#include "indexUtil.h"
#include "tcompare.h"
#include "tsched.h"
......
......@@ -13,9 +13,9 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "index_fst.h"
#include "indexFst.h"
#include "indexFstAutomation.h"
#include "indexInt.h"
#include "index_fst_automation.h"
#include "tchecksum.h"
#include "tcoding.h"
......
......@@ -13,7 +13,7 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "index_fst_automation.h"
#include "indexFstAutomation.h"
StartWithStateValue* startWithStateValueCreate(StartWithStateKind kind, ValueType ty, void* val) {
StartWithStateValue* sv = taosMemoryCalloc(1, sizeof(StartWithStateValue));
......
......@@ -12,7 +12,7 @@
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "index_fst_common.h"
#include "indexFstCommon.h"
const uint8_t COMMON_INPUTS[] = {
84, // '\x00'
......@@ -274,20 +274,23 @@ const uint8_t COMMON_INPUTS[] = {
};
const char COMMON_INPUTS_INV[] = {
't', 'e', '/', 'o', 'a', 's', 'r', 'i', 'p', 'c', 'n', 'w', '.', 'h', 'l', 'm',
'-', 'd', 'u', '0', '1', '2', 'g', '=', ':', 'b', 'f', '3', 'y', '5', '&', '_',
'4', 'v', '9', '6', '7', '8', 'k', '%', '?', 'x', 'C', 'D', 'A', 'S', 'F', 'I',
'B', 'E', 'j', 'P', 'T', 'z', 'R', 'N', 'M', '+', 'L', 'O', 'q', 'H', 'G', 'W',
'U', 'V', ',', 'Y', 'K', 'J', 'Z', 'X', 'Q', ';', ')', '(', '~', '[', ']', '$',
'!', '\'', '*', '@', '\x00', '\x01', '\x02', '\x03', '\x04', '\x05', '\x06', '\x07', '\x08', '\t', '\n', '\x0b',
'\x0c', '\r', '\x0e', '\x0f', '\x10', '\x11', '\x12', '\x13', '\x14', '\x15', '\x16', '\x17', '\x18', '\x19', '\x1a', '\x1b',
'\x1c', '\x1d', '\x1e', '\x1f', ' ', '"', '#', '<', '>', '\\', '^', '`', '{', '|', '}', '\x7f',
'\x80', '\x81', '\x82', '\x83', '\x84', '\x85', '\x86', '\x87', '\x88', '\x89', '\x8a', '\x8b', '\x8c', '\x8d', '\x8e', '\x8f',
'\x90', '\x91', '\x92', '\x93', '\x94', '\x95', '\x96', '\x97', '\x98', '\x99', '\x9a', '\x9b', '\x9c', '\x9d', '\x9e', '\x9f',
'\xa0', '\xa1', '\xa2', '\xa3', '\xa4', '\xa5', '\xa6', '\xa7', '\xa8', '\xa9', '\xaa', '\xab', '\xac', '\xad', '\xae', '\xaf',
'\xb0', '\xb1', '\xb2', '\xb3', '\xb4', '\xb5', '\xb6', '\xb7', '\xb8', '\xb9', '\xba', '\xbb', '\xbc', '\xbd', '\xbe', '\xbf',
'\xc0', '\xc1', '\xc2', '\xc3', '\xc4', '\xc5', '\xc6', '\xc7', '\xc8', '\xc9', '\xca', '\xcb', '\xcc', '\xcd', '\xce', '\xcf',
'\xd0', '\xd1', '\xd2', '\xd3', '\xd4', '\xd5', '\xd6', '\xd7', '\xd8', '\xd9', '\xda', '\xdb', '\xdc', '\xdd', '\xde', '\xdf',
'\xe0', '\xe1', '\xe2', '\xe3', '\xe4', '\xe5', '\xe6', '\xe7', '\xe8', '\xe9', '\xea', '\xeb', '\xec', '\xed', '\xee', '\xef',
'\xf0', '\xf1', '\xf2', '\xf3', '\xf4', '\xf5', '\xf6', '\xf7', '\xf8', '\xf9', '\xfa', '\xfb', '\xfc', '\xfd', '\xfe', '\xff',
't', 'e', '/', 'o', 'a', 's', 'r', 'i', 'p', 'c', 'n', 'w', '.', 'h',
'l', 'm', '-', 'd', 'u', '0', '1', '2', 'g', '=', ':', 'b', 'f', '3',
'y', '5', '&', '_', '4', 'v', '9', '6', '7', '8', 'k', '%', '?', 'x',
'C', 'D', 'A', 'S', 'F', 'I', 'B', 'E', 'j', 'P', 'T', 'z', 'R', 'N',
'M', '+', 'L', 'O', 'q', 'H', 'G', 'W', 'U', 'V', ',', 'Y', 'K', 'J',
'Z', 'X', 'Q', ';', ')', '(', '~', '[', ']', '$', '!', '\'', '*', '@',
'\x00', '\x01', '\x02', '\x03', '\x04', '\x05', '\x06', '\x07', '\x08', '\t', '\n', '\x0b', '\x0c', '\r',
'\x0e', '\x0f', '\x10', '\x11', '\x12', '\x13', '\x14', '\x15', '\x16', '\x17', '\x18', '\x19', '\x1a', '\x1b',
'\x1c', '\x1d', '\x1e', '\x1f', ' ', '"', '#', '<', '>', '\\', '^', '`', '{', '|',
'}', '\x7f', '\x80', '\x81', '\x82', '\x83', '\x84', '\x85', '\x86', '\x87', '\x88', '\x89', '\x8a', '\x8b',
'\x8c', '\x8d', '\x8e', '\x8f', '\x90', '\x91', '\x92', '\x93', '\x94', '\x95', '\x96', '\x97', '\x98', '\x99',
'\x9a', '\x9b', '\x9c', '\x9d', '\x9e', '\x9f', '\xa0', '\xa1', '\xa2', '\xa3', '\xa4', '\xa5', '\xa6', '\xa7',
'\xa8', '\xa9', '\xaa', '\xab', '\xac', '\xad', '\xae', '\xaf', '\xb0', '\xb1', '\xb2', '\xb3', '\xb4', '\xb5',
'\xb6', '\xb7', '\xb8', '\xb9', '\xba', '\xbb', '\xbc', '\xbd', '\xbe', '\xbf', '\xc0', '\xc1', '\xc2', '\xc3',
'\xc4', '\xc5', '\xc6', '\xc7', '\xc8', '\xc9', '\xca', '\xcb', '\xcc', '\xcd', '\xce', '\xcf', '\xd0', '\xd1',
'\xd2', '\xd3', '\xd4', '\xd5', '\xd6', '\xd7', '\xd8', '\xd9', '\xda', '\xdb', '\xdc', '\xdd', '\xde', '\xdf',
'\xe0', '\xe1', '\xe2', '\xe3', '\xe4', '\xe5', '\xe6', '\xe7', '\xe8', '\xe9', '\xea', '\xeb', '\xec', '\xed',
'\xee', '\xef', '\xf0', '\xf1', '\xf2', '\xf3', '\xf4', '\xf5', '\xf6', '\xf7', '\xf8', '\xf9', '\xfa', '\xfb',
'\xfc', '\xfd', '\xfe', '\xff',
};
......@@ -13,10 +13,10 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "os.h"
#include "index_fst_counting_writer.h"
#include "indexFstCountingWriter.h"
#include "indexFstUtil.h"
#include "indexInt.h"
#include "index_fst_util.h"
#include "os.h"
#include "tutil.h"
static int writeCtxDoWrite(WriterCtx* ctx, uint8_t* buf, int len) {
......@@ -82,7 +82,9 @@ static int writeCtxDoFlush(WriterCtx* ctx) {
WriterCtx* writerCtxCreate(WriterType type, const char* path, bool readOnly, int32_t capacity) {
WriterCtx* ctx = taosMemoryCalloc(1, sizeof(WriterCtx));
if (ctx == NULL) { return NULL; }
if (ctx == NULL) {
return NULL;
}
ctx->type = type;
if (ctx->type == TFile) {
......@@ -126,7 +128,9 @@ WriterCtx* writerCtxCreate(WriterType type, const char* path, bool readOnly, int
return ctx;
END:
if (ctx->type == TMemory) { taosMemoryFree(ctx->mem.buf); }
if (ctx->type == TMemory) {
taosMemoryFree(ctx->mem.buf);
}
taosMemoryFree(ctx);
return NULL;
}
......@@ -148,14 +152,18 @@ void writerCtxDestroy(WriterCtx* ctx, bool remove) {
// stat(ctx->file.buf, &fstat);
// indexError("write file size: %d", (int)(fstat.st_size));
}
if (remove) { unlink(ctx->file.buf); }
if (remove) {
unlink(ctx->file.buf);
}
}
taosMemoryFree(ctx);
}
FstCountingWriter* fstCountingWriterCreate(void* wrt) {
FstCountingWriter* cw = taosMemoryCalloc(1, sizeof(FstCountingWriter));
if (cw == NULL) { return NULL; }
if (cw == NULL) {
return NULL;
}
cw->wrt = wrt;
//(void *)(writerCtxCreate(TFile, readOnly));
......@@ -169,7 +177,9 @@ void fstCountingWriterDestroy(FstCountingWriter* cw) {
}
int fstCountingWriterWrite(FstCountingWriter* write, uint8_t* buf, uint32_t len) {
if (write == NULL) { return 0; }
if (write == NULL) {
return 0;
}
// update checksum
// write data to file/socket or mem
WriterCtx* ctx = write->wrt;
......@@ -182,7 +192,9 @@ int fstCountingWriterWrite(FstCountingWriter* write, uint8_t* buf, uint32_t len)
return len;
}
int fstCountingWriterRead(FstCountingWriter* write, uint8_t* buf, uint32_t len) {
if (write == NULL) { return 0; }
if (write == NULL) {
return 0;
}
WriterCtx* ctx = write->wrt;
int nRead = ctx->read(ctx, buf, len);
// assert(nRead == len);
......
......@@ -12,7 +12,7 @@
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "index_fst_node.h"
#include "indexFstNode.h"
FstBuilderNode* fstBuilderNodeDefault() {
FstBuilderNode* bn = taosMemoryMalloc(sizeof(FstBuilderNode));
......@@ -22,31 +22,45 @@ FstBuilderNode* fstBuilderNodeDefault() {
return bn;
}
void fstBuilderNodeDestroy(FstBuilderNode* node) {
if (node == NULL) { return; }
if (node == NULL) {
return;
}
taosArrayDestroy(node->trans);
taosMemoryFree(node);
}
bool fstBuilderNodeEqual(FstBuilderNode* n1, FstBuilderNode* n2) {
if (n1 == n2) { return true; }
if (n1 == NULL || n2 == NULL) { return false; }
if (n1 == n2) {
return true;
}
if (n1 == NULL || n2 == NULL) {
return false;
}
if (n1->isFinal != n2->isFinal || n1->finalOutput != n2->finalOutput) { return false; }
if (n1->isFinal != n2->isFinal || n1->finalOutput != n2->finalOutput) {
return false;
}
size_t s1 = n1->trans ? taosArrayGetSize(n1->trans) : 0;
size_t s2 = n2->trans ? taosArrayGetSize(n2->trans) : 0;
if (s1 != s2) { return false; }
if (s1 != s2) {
return false;
}
for (size_t i = 0; i < s1; i++) {
FstTransition* t1 = taosArrayGet(n1->trans, i);
FstTransition* t2 = taosArrayGet(n2->trans, i);
if (t1->inp != t2->inp || t1->out != t2->out || t1->addr != t2->addr) { return false; }
if (t1->inp != t2->inp || t1->out != t2->out || t1->addr != t2->addr) {
return false;
}
}
return true;
}
FstBuilderNode* fstBuilderNodeClone(FstBuilderNode* src) {
FstBuilderNode* node = taosMemoryMalloc(sizeof(FstBuilderNode));
if (node == NULL) { return NULL; }
if (node == NULL) {
return NULL;
}
//
size_t sz = taosArrayGetSize(src->trans);
......@@ -64,7 +78,9 @@ FstBuilderNode* fstBuilderNodeClone(FstBuilderNode* src) {
}
// not destroy src, User's bussiness
void fstBuilderNodeCloneFrom(FstBuilderNode* dst, FstBuilderNode* src) {
if (dst == NULL || src == NULL) { return; }
if (dst == NULL || src == NULL) {
return;
}
dst->isFinal = src->isFinal;
dst->finalOutput = src->finalOutput;
......
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册