xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request #9473 from taosdata/feature/vnode 

make create table OK
This commit is contained in:
Shengliang Guan 2021-12-29 14:36:11 +08:00 committed by GitHub
parent ea3f05f667 3b3566e3b4
commit 094962e74a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
18 changed files with 977 additions and 503 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.devcontainer/devcontainer.json
View File

@ -28,5 +28,5 @@
// "postCreateCommand": "gcc -v", // "postCreateCommand": "gcc -v",
// Comment out connect as root instead. More info: https://aka.ms/vscode-remote/containers/non-root. // Comment out connect as root instead. More info: https://aka.ms/vscode-remote/containers/non-root.
"remoteUser": "vscode" "remoteUser": "root"
} }

48
contrib/test/tdev/src/main.c
View File

@ -14,6 +14,10 @@
#define tPutB(buf, val) \ #define tPutB(buf, val) \
({ \ ({ \
((uint8_t *)buf)[7] = ((val) >> 56) & 0xff; \
((uint8_t *)buf)[6] = ((val) >> 48) & 0xff; \
((uint8_t *)buf)[5] = ((val) >> 40) & 0xff; \
((uint8_t *)buf)[4] = ((val) >> 32) & 0xff; \
((uint8_t *)buf)[3] = ((val) >> 24) & 0xff; \ ((uint8_t *)buf)[3] = ((val) >> 24) & 0xff; \
((uint8_t *)buf)[2] = ((val) >> 16) & 0xff; \ ((uint8_t *)buf)[2] = ((val) >> 16) & 0xff; \
((uint8_t *)buf)[1] = ((val) >> 8) & 0xff; \ ((uint8_t *)buf)[1] = ((val) >> 8) & 0xff; \
@ -23,11 +27,31 @@
#define tPutC(buf, val) \ #define tPutC(buf, val) \
({ \ ({ \
if (buf) { \
((uint64_t *)buf)[0] = (val); \ ((uint64_t *)buf)[0] = (val); \
POINTER_SHIFT(buf, sizeof(val)); \ POINTER_SHIFT(buf, sizeof(val)); \
} \
NULL; \
})
#define tPutD(buf, val) \
({ \
uint64_t tmp = val; \
for (size_t i = 0; i < sizeof(val); i++) { \
((uint8_t *)buf)[i] = tmp & 0xff; \
tmp >>= 8; \
} \
POINTER_SHIFT(buf, sizeof(val)); \
}) })
typedef enum { A, B, C } T; static inline void tPutE(void **buf, uint64_t val) {
if (buf) {
((uint64_t *)(*buf))[0] = val;
*buf = POINTER_SHIFT(*buf, sizeof(val));
}
}
typedef enum { A, B, C, D, E } T;
static void func(T t) { static void func(T t) {
uint64_t val = 198; uint64_t val = 198;
@ -59,6 +83,22 @@ static void func(T t) {
} }
} }
break; break;
case D:
for (size_t i = 0; i < 10 * 1024l * 1024l * 1024l; i++) {
pBuf = tPutD(pBuf, val);
if (POINTER_DISTANCE(buf, pBuf) == 1024) {
pBuf = buf;
}
}
break;
case E:
for (size_t i = 0; i < 10 * 1024l * 1024l * 1024l; i++) {
tPutE(&pBuf, val);
if (POINTER_DISTANCE(buf, pBuf) == 1024) {
pBuf = buf;
}
}
break;
default: default:
break; break;
@ -83,5 +123,11 @@ int main(int argc, char const *argv[]) {
func(C); func(C);
uint64_t t4 = now(); uint64_t t4 = now();
printf("C: %ld\n", t4 - t3); printf("C: %ld\n", t4 - t3);
func(D);
uint64_t t5 = now();
printf("D: %ld\n", t5 - t4);
func(E);
uint64_t t6 = now();
printf("E: %ld\n", t6 - t5);
return 0; return 0;
} }

1
include/common/tmsg.h
View File

@ -20,6 +20,7 @@
extern "C" { extern "C" {
#endif #endif
#include "encode.h"
#include "taosdef.h" #include "taosdef.h"
#include "taoserror.h" #include "taoserror.h"
#include "tcoding.h" #include "tcoding.h"

30
include/dnode/vnode/meta/meta.h
View File

@ -18,8 +18,8 @@
#include "mallocator.h" #include "mallocator.h"
#include "os.h" #include "os.h"
#include "trow.h"
#include "tmsg.h" #include "tmsg.h"
#include "trow.h"
#ifdef __cplusplus #ifdef __cplusplus
extern "C" { extern "C" {
@ -47,35 +47,13 @@ int metaCreateTable(SMeta *pMeta, STbCfg *pTbCfg);
int metaDropTable(SMeta *pMeta, tb_uid_t uid); int metaDropTable(SMeta *pMeta, tb_uid_t uid);
int metaCommit(SMeta *pMeta); int metaCommit(SMeta *pMeta);
// For Query
int metaGetTableInfo(SMeta *pMeta, char *tbname, STableMetaMsg **ppMsg);
// Options // Options
void metaOptionsInit(SMetaCfg *pMetaCfg); void metaOptionsInit(SMetaCfg *pMetaCfg);
void metaOptionsClear(SMetaCfg *pMetaCfg); void metaOptionsClear(SMetaCfg *pMetaCfg);
// STbCfg
#define META_INIT_STB_CFG(NAME, TTL, KEEP, SUID, PSCHEMA, PTAGSCHEMA) \
{ \
.name = (NAME), .ttl = (TTL), .keep = (KEEP), .type = META_SUPER_TABLE, .stbCfg = { \
.suid = (SUID), \
.pSchema = (PSCHEMA), \
.pTagSchema = (PTAGSCHEMA) \
} \
}
#define META_INIT_CTB_CFG(NAME, TTL, KEEP, SUID, PTAG) \
{ \
.name = (NAME), .ttl = (TTL), .keep = (KEEP), .type = META_CHILD_TABLE, .ctbCfg = {.suid = (SUID), .pTag = PTAG } \
}
#define META_INIT_NTB_CFG(NAME, TTL, KEEP, SUID, PSCHEMA) \
{ \
.name = (NAME), .ttl = (TTL), .keep = (KEEP), .type = META_NORMAL_TABLE, .ntbCfg = {.pSchema = (PSCHEMA) } \
}
#define META_CLEAR_TB_CFG(pTbCfg)
int metaEncodeTbCfg(void **pBuf, STbCfg *pTbCfg);
void *metaDecodeTbCfg(void *pBuf, STbCfg *pTbCfg);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

62
include/dnode/vnode/vnode.h
View File

@ -187,68 +187,6 @@ void vnodeOptionsInit(SVnodeCfg *pOptions);
*/ */
void vnodeOptionsClear(SVnodeCfg *pOptions); void vnodeOptionsClear(SVnodeCfg *pOptions);
/* ------------------------ REQUESTS ------------------------ */
typedef STbCfg SVCreateTableReq;
typedef struct {
tb_uid_t uid;
} SVDropTableReq;
typedef struct {
// TODO
} SVSubmitReq;
typedef struct {
uint64_t ver;
union {
SVCreateTableReq ctReq;
SVDropTableReq dtReq;
};
} SVnodeReq;
typedef struct {
int err;
char info[];
} SVnodeRsp;
static FORCE_INLINE void vnodeSetCreateStbReq(SVnodeReq *pReq, char *name, uint32_t ttl, uint32_t keep, tb_uid_t suid,
STSchema *pSchema, STSchema *pTagSchema) {
pReq->ver = 0;
pReq->ctReq.name = name;
pReq->ctReq.ttl = ttl;
pReq->ctReq.keep = keep;
pReq->ctReq.type = META_SUPER_TABLE;
pReq->ctReq.stbCfg.suid = suid;
pReq->ctReq.stbCfg.pSchema = pSchema;
pReq->ctReq.stbCfg.pTagSchema = pTagSchema;
}
static FORCE_INLINE void vnodeSetCreateCtbReq(SVnodeReq *pReq, char *name, uint32_t ttl, uint32_t keep, tb_uid_t suid,
SKVRow pTag) {
pReq->ver = 0;
pReq->ctReq.name = name;
pReq->ctReq.ttl = ttl;
pReq->ctReq.keep = keep;
pReq->ctReq.type = META_CHILD_TABLE;
pReq->ctReq.ctbCfg.suid = suid;
pReq->ctReq.ctbCfg.pTag = pTag;
}
static FORCE_INLINE void vnodeSetCreateNtbReq(SVnodeReq *pReq, char *name, uint32_t ttl, uint32_t keep,
STSchema *pSchema) {
pReq->ver = 0;
pReq->ctReq.name = name;
pReq->ctReq.ttl = ttl;
pReq->ctReq.keep = keep;
pReq->ctReq.type = META_NORMAL_TABLE;
pReq->ctReq.ntbCfg.pSchema = pSchema;
}
int vnodeBuildReq(void **buf, const SVnodeReq *pReq, tmsg_t type);
void *vnodeParseReq(void *buf, SVnodeReq *pReq, tmsg_t type);
/* ------------------------ FOR COMPILE ------------------------ */ /* ------------------------ FOR COMPILE ------------------------ */
int32_t vnodeAlter(SVnode *pVnode, const SVnodeCfg *pCfg); int32_t vnodeAlter(SVnode *pVnode, const SVnodeCfg *pCfg);

3
include/os/osEndian.h
View File

@ -20,8 +20,11 @@
extern "C" { extern "C" {
#endif #endif
typedef enum { TD_LITTLE_ENDIAN = 0, TD_BIG_ENDIAN } td_endian_t;
static const int32_t endian_test_var = 1; static const int32_t endian_test_var = 1;
#define IS_LITTLE_ENDIAN() (*(uint8_t *)(&endian_test_var) != 0) #define IS_LITTLE_ENDIAN() (*(uint8_t *)(&endian_test_var) != 0)
#define TD_RT_ENDIAN() (IS_LITTLE_ENDIAN() ? TD_LITTLE_ENDIAN : TD_BIG_ENDIAN)
#ifdef __cplusplus #ifdef __cplusplus
} }

476
include/util/encode.h Normal file
View File

@ -0,0 +1,476 @@
/*
* 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_UTIL_ENCODE_H_
#define _TD_UTIL_ENCODE_H_
#include "tcoding.h"
#include "tmacro.h"
#ifdef __cplusplus
extern "C" {
#endif
typedef struct {
td_endian_t endian;
uint8_t* data;
int64_t size;
int64_t pos;
} SEncoder, SDecoder;
#define tPut(TYPE, BUF, VAL) ((TYPE*)(BUF))[0] = (VAL)
#define tGet(TYPE, BUF, VAL) (VAL) = ((TYPE*)(BUF))[0]
#define tRPut16(PDEST, PSRC) \
((uint8_t*)(PDEST))[0] = ((uint8_t*)(PSRC))[1]; \
((uint8_t*)(PDEST))[1] = ((uint8_t*)(PSRC))[0];
#define tRPut32(PDEST, PSRC) \
((uint8_t*)(PDEST))[0] = ((uint8_t*)(PSRC))[3]; \
((uint8_t*)(PDEST))[1] = ((uint8_t*)(PSRC))[2]; \
((uint8_t*)(PDEST))[2] = ((uint8_t*)(PSRC))[1]; \
((uint8_t*)(PDEST))[3] = ((uint8_t*)(PSRC))[0];
#define tRPut64(PDEST, PSRC) \
((uint8_t*)(PDEST))[0] = ((uint8_t*)(PSRC))[7]; \
((uint8_t*)(PDEST))[1] = ((uint8_t*)(PSRC))[6]; \
((uint8_t*)(PDEST))[2] = ((uint8_t*)(PSRC))[5]; \
((uint8_t*)(PDEST))[3] = ((uint8_t*)(PSRC))[4]; \
((uint8_t*)(PDEST))[4] = ((uint8_t*)(PSRC))[3]; \
((uint8_t*)(PDEST))[5] = ((uint8_t*)(PSRC))[2]; \
((uint8_t*)(PDEST))[6] = ((uint8_t*)(PSRC))[1]; \
((uint8_t*)(PDEST))[7] = ((uint8_t*)(PSRC))[0];
#define tRGet16 tRPut16
#define tRGet32 tRPut32
#define tRGet64 tRPut64
#define TD_CODER_CURRENT(CODER) ((CODER)->data + (CODER)->pos)
#define TD_CODER_MOVE_POS(CODER, MOVE) ((CODER)->pos += (MOVE))
#define TD_CHECK_CODER_CAPACITY_FAILED(CODER, EXPSIZE) (((CODER)->size - (CODER)->pos) < (EXPSIZE))
/* ------------------------ FOR ENCODER ------------------------ */
static FORCE_INLINE void tInitEncoder(SEncoder* pEncoder, td_endian_t endian, uint8_t* data, int64_t size) {
pEncoder->endian = endian;
pEncoder->data = data;
pEncoder->size = (data) ? size : 0;
pEncoder->pos = 0;
}
// 8
static FORCE_INLINE int tEncodeU8(SEncoder* pEncoder, uint8_t val) {
if (pEncoder->data) {
if (TD_CHECK_CODER_CAPACITY_FAILED(pEncoder, sizeof(val))) return -1;
tPut(uint8_t, TD_CODER_CURRENT(pEncoder), val);
}
TD_CODER_MOVE_POS(pEncoder, sizeof(val));
return 0;
}
static FORCE_INLINE int tEncodeI8(SEncoder* pEncoder, int8_t val) {
if (pEncoder->data) {
if (TD_CHECK_CODER_CAPACITY_FAILED(pEncoder, sizeof(val))) return -1;
tPut(int8_t, TD_CODER_CURRENT(pEncoder), val);
}
TD_CODER_MOVE_POS(pEncoder, sizeof(val));
return 0;
}
// 16
static FORCE_INLINE int tEncodeU16(SEncoder* pEncoder, uint16_t val) {
if (pEncoder->data) {
if (TD_CHECK_CODER_CAPACITY_FAILED(pEncoder, sizeof(val))) return -1;
if (TD_RT_ENDIAN() == pEncoder->endian) {
tPut(uint16_t, TD_CODER_CURRENT(pEncoder), val);
} else {
tRPut16(TD_CODER_CURRENT(pEncoder), &val);
}
}
TD_CODER_MOVE_POS(pEncoder, sizeof(val));
return 0;
}
static FORCE_INLINE int tEncodeI16(SEncoder* pEncoder, int16_t val) {
if (pEncoder->data) {
if (TD_CHECK_CODER_CAPACITY_FAILED(pEncoder, sizeof(val))) return -1;
if (TD_RT_ENDIAN() == pEncoder->endian) {
tPut(int16_t, TD_CODER_CURRENT(pEncoder), val);
} else {
tRPut16(TD_CODER_CURRENT(pEncoder), &val);
}
}
TD_CODER_MOVE_POS(pEncoder, sizeof(val));
return 0;
}
// 32
static FORCE_INLINE int tEncodeU32(SEncoder* pEncoder, uint32_t val) {
if (pEncoder->data) {
if (TD_CHECK_CODER_CAPACITY_FAILED(pEncoder, sizeof(val))) return -1;
if (TD_RT_ENDIAN() == pEncoder->endian) {
tPut(uint32_t, TD_CODER_CURRENT(pEncoder), val);
} else {
tRPut32(TD_CODER_CURRENT(pEncoder), &val);
}
}
TD_CODER_MOVE_POS(pEncoder, sizeof(val));
return 0;
}
static FORCE_INLINE int tEncodeI32(SEncoder* pEncoder, int32_t val) {
if (pEncoder->data) {
if (TD_CHECK_CODER_CAPACITY_FAILED(pEncoder, sizeof(val))) return -1;
if (TD_RT_ENDIAN() == pEncoder->endian) {
tPut(int32_t, TD_CODER_CURRENT(pEncoder), val);
} else {
tRPut32(TD_CODER_CURRENT(pEncoder), &val);
}
}
TD_CODER_MOVE_POS(pEncoder, sizeof(val));
return 0;
}
// 64
static FORCE_INLINE int tEncodeU64(SEncoder* pEncoder, uint64_t val) {
if (pEncoder->data) {
if (TD_CHECK_CODER_CAPACITY_FAILED(pEncoder, sizeof(val))) return -1;
if (TD_RT_ENDIAN() == pEncoder->endian) {
tPut(uint64_t, TD_CODER_CURRENT(pEncoder), val);
} else {
tRPut64(TD_CODER_CURRENT(pEncoder), &val);
}
}
TD_CODER_MOVE_POS(pEncoder, sizeof(val));
return 0;
}
static FORCE_INLINE int tEncodeI64(SEncoder* pEncoder, int64_t val) {
if (pEncoder->data) {
if (TD_CHECK_CODER_CAPACITY_FAILED(pEncoder, sizeof(val))) return -1;
if (TD_RT_ENDIAN() == pEncoder->endian) {
tPut(int64_t, TD_CODER_CURRENT(pEncoder), val);
} else {
tRPut64(TD_CODER_CURRENT(pEncoder), &val);
}
}
TD_CODER_MOVE_POS(pEncoder, sizeof(val));
return 0;
}
// 16v
static FORCE_INLINE int tEncodeU16v(SEncoder* pEncoder, uint16_t val) {
int64_t i = 0;
while (val >= ENCODE_LIMIT) {
if (pEncoder->data) {
if (TD_CHECK_CODER_CAPACITY_FAILED(pEncoder, 1)) return -1;
TD_CODER_CURRENT(pEncoder)[i] = (val | ENCODE_LIMIT) & 0xff;
}
val >>= 7;
i++;
}
if (pEncoder->data) {
if (TD_CHECK_CODER_CAPACITY_FAILED(pEncoder, 1)) return -1;
TD_CODER_CURRENT(pEncoder)[i] = (uint8_t)val;
}
TD_CODER_MOVE_POS(pEncoder, i + 1);
return 0;
}
static FORCE_INLINE int tEncodeI16v(SEncoder* pEncoder, int16_t val) {
return tEncodeU16v(pEncoder, ZIGZAGE(int16_t, val));
}
// 32v
static FORCE_INLINE int tEncodeU32v(SEncoder* pEncoder, uint32_t val) {
int64_t i = 0;
while (val >= ENCODE_LIMIT) {
if (pEncoder->data) {
if (TD_CHECK_CODER_CAPACITY_FAILED(pEncoder, 1)) return -1;
TD_CODER_CURRENT(pEncoder)[i] = (val | ENCODE_LIMIT) & 0xff;
}
val >>= 7;
i++;
}
if (pEncoder->data) {
if (TD_CHECK_CODER_CAPACITY_FAILED(pEncoder, 1)) return -1;
TD_CODER_CURRENT(pEncoder)[i] = (uint8_t)val;
}
TD_CODER_MOVE_POS(pEncoder, i + 1);
return 0;
}
static FORCE_INLINE int tEncodeI32v(SEncoder* pEncoder, int32_t val) {
return tEncodeU32v(pEncoder, ZIGZAGE(int32_t, val));
}
// 64v
static FORCE_INLINE int tEncodeU64v(SEncoder* pEncoder, uint64_t val) {
int64_t i = 0;
while (val >= ENCODE_LIMIT) {
if (pEncoder->data) {
if (TD_CHECK_CODER_CAPACITY_FAILED(pEncoder, 1)) return -1;
TD_CODER_CURRENT(pEncoder)[i] = (val | ENCODE_LIMIT) & 0xff;
}
val >>= 7;
i++;
}
if (pEncoder->data) {
if (TD_CHECK_CODER_CAPACITY_FAILED(pEncoder, 1)) return -1;
TD_CODER_CURRENT(pEncoder)[i] = (uint8_t)val;
}
TD_CODER_MOVE_POS(pEncoder, i + 1);
return 0;
}
static FORCE_INLINE int tEncodeI64v(SEncoder* pEncoder, int64_t val) {
return tEncodeU64v(pEncoder, ZIGZAGE(int64_t, val));
}
static FORCE_INLINE int tEncodeFloat(SEncoder* pEncoder, float val) {
// TODO
return 0;
}
static FORCE_INLINE int tEncodeDouble(SEncoder* pEncoder, double val) {
// TODO
return 0;
}
static FORCE_INLINE int tEncodeCStr(SEncoder* pEncoder, const char* val) {
// TODO
return 0;
}
/* ------------------------ FOR DECODER ------------------------ */
static FORCE_INLINE void tInitDecoder(SDecoder* pDecoder, td_endian_t endian, uint8_t* data, int64_t size) {
ASSERT(!TD_IS_NULL(data));
pDecoder->endian = endian;
pDecoder->data = data;
pDecoder->size = size;
pDecoder->pos = 0;
}
// 8
static FORCE_INLINE int tDecodeU8(SDecoder* pDecoder, uint8_t* val) {
if (TD_CHECK_CODER_CAPACITY_FAILED(pDecoder, sizeof(*val))) return -1;
tGet(uint8_t, TD_CODER_CURRENT(pDecoder), *val);
TD_CODER_MOVE_POS(pDecoder, sizeof(*val));
return 0;
}
static FORCE_INLINE int tDecodeI8(SDecoder* pDecoder, int8_t* val) {
if (TD_CHECK_CODER_CAPACITY_FAILED(pDecoder, sizeof(*val))) return -1;
tGet(int8_t, TD_CODER_CURRENT(pDecoder), *val);
TD_CODER_MOVE_POS(pDecoder, sizeof(*val));
return 0;
}
// 16
static FORCE_INLINE int tDecodeU16(SDecoder* pDecoder, uint16_t* val) {
if (TD_CHECK_CODER_CAPACITY_FAILED(pDecoder, sizeof(*val))) return -1;
if (TD_RT_ENDIAN() == pDecoder->endian) {
tGet(uint16_t, TD_CODER_CURRENT(pDecoder), *val);
} else {
tRGet16(val, TD_CODER_CURRENT(pDecoder));
}
TD_CODER_MOVE_POS(pDecoder, sizeof(*val));
return 0;
}
static FORCE_INLINE int tDecodeI16(SDecoder* pDecoder, int16_t* val) {
if (TD_CHECK_CODER_CAPACITY_FAILED(pDecoder, sizeof(*val))) return -1;
if (TD_RT_ENDIAN() == pDecoder->endian) {
tGet(int16_t, TD_CODER_CURRENT(pDecoder), *val);
} else {
tRGet16(val, TD_CODER_CURRENT(pDecoder));
}
TD_CODER_MOVE_POS(pDecoder, sizeof(*val));
return 0;
}
// 32
static FORCE_INLINE int tDecodeU32(SDecoder* pDecoder, uint32_t* val) {
if (TD_CHECK_CODER_CAPACITY_FAILED(pDecoder, sizeof(*val))) return -1;
if (TD_RT_ENDIAN() == pDecoder->endian) {
tGet(uint32_t, TD_CODER_CURRENT(pDecoder), *val);
} else {
tRGet32(val, TD_CODER_CURRENT(pDecoder));
}
TD_CODER_MOVE_POS(pDecoder, sizeof(*val));
return 0;
}
static FORCE_INLINE int tDecodeI32(SDecoder* pDecoder, int32_t* val) {
if (TD_CHECK_CODER_CAPACITY_FAILED(pDecoder, sizeof(*val))) return -1;
if (TD_RT_ENDIAN() == pDecoder->endian) {
tGet(int32_t, TD_CODER_CURRENT(pDecoder), *val);
} else {
tRGet32(val, TD_CODER_CURRENT(pDecoder));
}
TD_CODER_MOVE_POS(pDecoder, sizeof(*val));
return 0;
}
// 64
static FORCE_INLINE int tDecodeU64(SDecoder* pDecoder, uint64_t* val) {
if (TD_CHECK_CODER_CAPACITY_FAILED(pDecoder, sizeof(*val))) return -1;
if (TD_RT_ENDIAN() == pDecoder->endian) {
tGet(uint64_t, TD_CODER_CURRENT(pDecoder), *val);
} else {
tRGet64(val, TD_CODER_CURRENT(pDecoder));
}
TD_CODER_MOVE_POS(pDecoder, sizeof(*val));
return 0;
}
static FORCE_INLINE int tDecodeI64(SDecoder* pDecoder, int64_t* val) {
if (TD_CHECK_CODER_CAPACITY_FAILED(pDecoder, sizeof(*val))) return -1;
if (TD_RT_ENDIAN() == pDecoder->endian) {
tGet(int64_t, TD_CODER_CURRENT(pDecoder), *val);
} else {
tRGet64(val, TD_CODER_CURRENT(pDecoder));
}
TD_CODER_MOVE_POS(pDecoder, sizeof(*val));
return 0;
}
// 16v
static FORCE_INLINE int tDecodeU16v(SDecoder* pDecoder, uint16_t* val) {
int64_t i = 0;
*val = 0;
for (;;) {
if (TD_CHECK_CODER_CAPACITY_FAILED(pDecoder, 1)) return -1;
uint16_t tval = TD_CODER_CURRENT(pDecoder)[i];
if (tval < ENCODE_LIMIT) {
(*val) |= (tval << (7 * i));
break;
} else {
(*val) |= (((tval) & (ENCODE_LIMIT - 1)) << (7 * i));
i++;
}
}
TD_CODER_MOVE_POS(pDecoder, i);
return 0;
}
static FORCE_INLINE int tDecodeI16v(SDecoder* pDecoder, int16_t* val) {
uint16_t tval;
if (tDecodeU16v(pDecoder, &tval) < 0) {
return -1;
}
*val = ZIGZAGD(int16_t, tval);
return 0;
}
// 32v
static FORCE_INLINE int tDecodeU32v(SDecoder* pDecoder, uint32_t* val) {
int64_t i = 0;
*val = 0;
for (;;) {
if (TD_CHECK_CODER_CAPACITY_FAILED(pDecoder, 1)) return -1;
uint32_t tval = TD_CODER_CURRENT(pDecoder)[i];
if (tval < ENCODE_LIMIT) {
(*val) |= (tval << (7 * i));
break;
} else {
(*val) |= (((tval) & (ENCODE_LIMIT - 1)) << (7 * i));
i++;
}
}
TD_CODER_MOVE_POS(pDecoder, i);
return 0;
}
static FORCE_INLINE int tDecodeI32v(SDecoder* pDecoder, int32_t* val) {
uint32_t tval;
if (tDecodeU32v(pDecoder, &tval) < 0) {
return -1;
}
*val = ZIGZAGD(int32_t, tval);
return 0;
}
// 64v
static FORCE_INLINE int tDecodeU64v(SDecoder* pDecoder, uint64_t* val) {
int64_t i = 0;
*val = 0;
for (;;) {
if (TD_CHECK_CODER_CAPACITY_FAILED(pDecoder, 1)) return -1;
uint64_t tval = TD_CODER_CURRENT(pDecoder)[i];
if (tval < ENCODE_LIMIT) {
(*val) |= (tval << (7 * i));
break;
} else {
(*val) |= (((tval) & (ENCODE_LIMIT - 1)) << (7 * i));
i++;
}
}
TD_CODER_MOVE_POS(pDecoder, i);
return 0;
}
static FORCE_INLINE int tDecodeI64v(SDecoder* pDecoder, int64_t* val) {
uint64_t tval;
if (tDecodeU64v(pDecoder, &tval) < 0) {
return -1;
}
*val = ZIGZAGD(int64_t, tval);
return 0;
}
static FORCE_INLINE int tDecodeFloat(SDecoder* pDecoder, float* val) {
// TODO
return 0;
}
static FORCE_INLINE int tDecodeDouble(SDecoder* pDecoder, double* val) {
// TODO
return 0;
}
static FORCE_INLINE int tDecodeCStr(SDecoder* pEncoder, const char** val) {
// TODO
return 0;
}
#ifdef __cplusplus
}
#endif
#endif /*_TD_UTIL_ENCODE_H_*/

67
include/util/tbuffer.h
View File

@ -16,6 +16,8 @@
#ifndef _TD_UTIL_BUFFER_H #ifndef _TD_UTIL_BUFFER_H
#define _TD_UTIL_BUFFER_H #define _TD_UTIL_BUFFER_H
#include "os.h"
#ifdef __cplusplus #ifdef __cplusplus
extern "C" { extern "C" {
#endif #endif
@ -90,48 +92,17 @@ typedef struct SBufferWriter {
#define tbufTell(buf) ((buf)->pos) #define tbufTell(buf) ((buf)->pos)
//////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
// reader functions & macros /* ------------------------ BUFFER WRITER FUNCTIONS AND MACROS ------------------------ */
// *Endian*, if true, reader functions of primitive types will do 'ntoh' automatically
#define tbufInitReader( Data, Size, Endian ) {.endian = (Endian), .data = (Data), .pos = 0, .size = ((Data) == NULL ? 0 :(Size))}
size_t tbufSkip( SBufferReader* buf, size_t size );
const char* tbufRead( SBufferReader* buf, size_t size );
void tbufReadToBuffer( SBufferReader* buf, void* dst, size_t size );
const char* tbufReadString( SBufferReader* buf, size_t* len );
size_t tbufReadToString( SBufferReader* buf, char* dst, size_t size );
const char* tbufReadBinary( SBufferReader* buf, size_t *len );
size_t tbufReadToBinary( SBufferReader* buf, void* dst, size_t size );
bool tbufReadBool( SBufferReader* buf );
char tbufReadChar( SBufferReader* buf );
int8_t tbufReadInt8( SBufferReader* buf );
uint8_t tbufReadUint8( SBufferReader* buf );
int16_t tbufReadInt16( SBufferReader* buf );
uint16_t tbufReadUint16( SBufferReader* buf );
int32_t tbufReadInt32( SBufferReader* buf );
uint32_t tbufReadUint32( SBufferReader* buf );
int64_t tbufReadInt64( SBufferReader* buf );
uint64_t tbufReadUint64( SBufferReader* buf );
float tbufReadFloat( SBufferReader* buf );
double tbufReadDouble( SBufferReader* buf );
////////////////////////////////////////////////////////////////////////////////
// writer functions & macros
// *Allocator*, function to allocate memory, will use 'realloc' if NULL // *Allocator*, function to allocate memory, will use 'realloc' if NULL
// *Endian*, if true, writer functions of primitive types will do 'hton' automatically // *Endian*, if true, writer functions of primitive types will do 'hton' automatically
#define tbufInitWriter( Allocator, Endian ) {.endian = (Endian), .data = NULL, .pos = 0, .size = 0, .allocator = ((Allocator) == NULL ? realloc : (Allocator))} #define tbufInitWriter(Allocator, Endian) \
void tbufCloseWriter( SBufferWriter* buf ); { .endian = (Endian), .data = NULL, .pos = 0, .size = 0, .allocator = ((Allocator) == NULL ? realloc : (Allocator)) }
void tbufCloseWriter(SBufferWriter* buf);
void tbufEnsureCapacity(SBufferWriter* buf, size_t size); void tbufEnsureCapacity(SBufferWriter* buf, size_t size);
size_t tbufReserve(SBufferWriter* buf, size_t size); size_t tbufReserve(SBufferWriter* buf, size_t size);
char* tbufGetData(SBufferWriter* buf, bool takeOver); char* tbufGetData(SBufferWriter* buf, bool takeOver);
void tbufWrite(SBufferWriter* buf, const void* data, size_t size); void tbufWrite(SBufferWriter* buf, const void* data, size_t size);
void tbufWriteAt(SBufferWriter* buf, size_t pos, const void* data, size_t size); void tbufWriteAt(SBufferWriter* buf, size_t pos, const void* data, size_t size);
void tbufWriteStringLen(SBufferWriter* buf, const char* str, size_t len); void tbufWriteStringLen(SBufferWriter* buf, const char* str, size_t len);
@ -142,7 +113,6 @@ void tbufWriteString( SBufferWriter* buf, const char* str );
// size before read. Write only write the data itself, which means the reader // size before read. Write only write the data itself, which means the reader
// need to know data size before read. // need to know data size before read.
void tbufWriteBinary(SBufferWriter* buf, const void* data, size_t len); void tbufWriteBinary(SBufferWriter* buf, const void* data, size_t len);
void tbufWriteBool(SBufferWriter* buf, bool data); void tbufWriteBool(SBufferWriter* buf, bool data);
void tbufWriteBoolAt(SBufferWriter* buf, size_t pos, bool data); void tbufWriteBoolAt(SBufferWriter* buf, size_t pos, bool data);
void tbufWriteChar(SBufferWriter* buf, char data); void tbufWriteChar(SBufferWriter* buf, char data);
@ -168,6 +138,31 @@ void tbufWriteFloatAt( SBufferWriter* buf, size_t pos, float data );
void tbufWriteDouble(SBufferWriter* buf, double data); void tbufWriteDouble(SBufferWriter* buf, double data);
void tbufWriteDoubleAt(SBufferWriter* buf, size_t pos, double data); void tbufWriteDoubleAt(SBufferWriter* buf, size_t pos, double data);
/* ------------------------ BUFFER READER FUNCTIONS AND MACROS ------------------------ */
// *Endian*, if true, reader functions of primitive types will do 'ntoh' automatically
#define tbufInitReader(Data, Size, Endian) \
{ .endian = (Endian), .data = (Data), .pos = 0, .size = ((Data) == NULL ? 0 : (Size)) }
size_t tbufSkip(SBufferReader* buf, size_t size);
const char* tbufRead(SBufferReader* buf, size_t size);
void tbufReadToBuffer(SBufferReader* buf, void* dst, size_t size);
const char* tbufReadString(SBufferReader* buf, size_t* len);
size_t tbufReadToString(SBufferReader* buf, char* dst, size_t size);
const char* tbufReadBinary(SBufferReader* buf, size_t* len);
size_t tbufReadToBinary(SBufferReader* buf, void* dst, size_t size);
bool tbufReadBool(SBufferReader* buf);
char tbufReadChar(SBufferReader* buf);
int8_t tbufReadInt8(SBufferReader* buf);
uint8_t tbufReadUint8(SBufferReader* buf);
int16_t tbufReadInt16(SBufferReader* buf);
uint16_t tbufReadUint16(SBufferReader* buf);
int32_t tbufReadInt32(SBufferReader* buf);
uint32_t tbufReadUint32(SBufferReader* buf);
int64_t tbufReadInt64(SBufferReader* buf);
uint64_t tbufReadUint64(SBufferReader* buf);
float tbufReadFloat(SBufferReader* buf);
double tbufReadDouble(SBufferReader* buf);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

4
include/util/tcoding.h
View File

@ -25,6 +25,8 @@ extern "C" {
#define ZIGZAGE(T, v) ((u##T)((v) >> (sizeof(T) * 8 - 1))) ^ (((u##T)(v)) << 1) // zigzag encode #define ZIGZAGE(T, v) ((u##T)((v) >> (sizeof(T) * 8 - 1))) ^ (((u##T)(v)) << 1) // zigzag encode
#define ZIGZAGD(T, v) ((v) >> 1) ^ -((T)((v)&1)) // zigzag decode #define ZIGZAGD(T, v) ((v) >> 1) ^ -((T)((v)&1)) // zigzag decode
/* ------------------------ LEGACY CODES ------------------------ */
#if 1
// ---- Fixed U8 // ---- Fixed U8
static FORCE_INLINE int taosEncodeFixedU8(void **buf, uint8_t value) { static FORCE_INLINE int taosEncodeFixedU8(void **buf, uint8_t value) {
if (buf != NULL) { if (buf != NULL) {
@ -368,6 +370,8 @@ static FORCE_INLINE void *taosDecodeStringTo(void *buf, char *value) {
return POINTER_SHIFT(buf, size); return POINTER_SHIFT(buf, size);
} }
#endif
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

2
include/util/tmacro.h
View File

@ -35,6 +35,8 @@ typedef int8_t td_mode_flag_t;
#define TD_CHECK_AND_SET_MOD_CLEAR(FLAG) atomic_val_compare_exchange_8((FLAG), TD_MOD_UNCLEARD, TD_MOD_CLEARD) #define TD_CHECK_AND_SET_MOD_CLEAR(FLAG) atomic_val_compare_exchange_8((FLAG), TD_MOD_UNCLEARD, TD_MOD_CLEARD)
#define TD_IS_NULL(PTR) ((PTR) == NULL)
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

4
source/dnode/vnode/impl/inc/vnodeRequest.h
View File

@ -23,8 +23,8 @@ extern "C" {
#endif #endif
// SVDropTableReq // SVDropTableReq
int vnodeBuildDropTableReq(void **buf, const SVDropTableReq *pReq); // int vnodeBuildDropTableReq(void **buf, const SVDropTableReq *pReq);
void *vnodeParseDropTableReq(void *buf, SVDropTableReq *pReq); // void *vnodeParseDropTableReq(void *buf, SVDropTableReq *pReq);
#ifdef __cplusplus #ifdef __cplusplus
} }

26
source/dnode/vnode/impl/src/vnodeQuery.c
View File

@ -13,12 +13,10 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#include "vnodeDef.h"
#include "vnodeQuery.h" #include "vnodeQuery.h"
#include "vnodeDef.h"
int vnodeQueryOpen(SVnode *pVnode) { int vnodeQueryOpen(SVnode *pVnode) { return qWorkerInit(NULL, &pVnode->pQuery); }
return qWorkerInit(NULL, &pVnode->pQuery);
}
int vnodeProcessQueryReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) { int vnodeProcessQueryReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
vInfo("query message is processed"); vInfo("query message is processed");
@ -46,4 +44,24 @@ int vnodeProcessFetchReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
return 0; return 0;
} }
static int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
STableInfoMsg *pReq = (STableInfoMsg *)(pMsg->pCont);
STableMetaMsg *pRspMsg;
int ret;
if (metaGetTableInfo(pVnode->pMeta, pReq->tableFname, &pRspMsg) < 0) {
return -1;
}
*pRsp = malloc(sizeof(SRpcMsg));
if (TD_IS_NULL(*pRsp)) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
free(pMsg);
return -1;
}
// TODO
(*pRsp)->pCont = pRspMsg;
return 0;
}

3
source/dnode/vnode/impl/src/vnodeRequest.c
View File

@ -15,6 +15,8 @@
#include "vnodeDef.h" #include "vnodeDef.h"
#if 0
static int vnodeBuildCreateTableReq(void **buf, const SVCreateTableReq *pReq); static int vnodeBuildCreateTableReq(void **buf, const SVCreateTableReq *pReq);
static void *vnodeParseCreateTableReq(void *buf, SVCreateTableReq *pReq); static void *vnodeParseCreateTableReq(void *buf, SVCreateTableReq *pReq);
@ -114,3 +116,4 @@ int vnodeBuildDropTableReq(void **buf, const SVDropTableReq *pReq) {
void *vnodeParseDropTableReq(void *buf, SVDropTableReq *pReq) { void *vnodeParseDropTableReq(void *buf, SVDropTableReq *pReq) {
// TODO // TODO
} }
#endif

9
source/dnode/vnode/impl/src/vnodeWrite.c
View File

@ -28,7 +28,6 @@ int vnodeProcessNoWalWMsgs(SVnode *pVnode, SRpcMsg *pMsg) {
int vnodeProcessWMsgs(SVnode *pVnode, SArray *pMsgs) { int vnodeProcessWMsgs(SVnode *pVnode, SArray *pMsgs) {
SRpcMsg * pMsg; SRpcMsg * pMsg;
SVnodeReq *pVnodeReq;
for (int i = 0; i < taosArrayGetSize(pMsgs); i++) { for (int i = 0; i < taosArrayGetSize(pMsgs); i++) {
pMsg = *(SRpcMsg **)taosArrayGet(pMsgs, i); pMsg = *(SRpcMsg **)taosArrayGet(pMsgs, i);
@ -51,7 +50,6 @@ int vnodeProcessWMsgs(SVnode *pVnode, SArray *pMsgs) {
} }
int vnodeApplyWMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) { int vnodeApplyWMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
SVnodeReq vReq;
SVCreateTbReq vCreateTbReq; SVCreateTbReq vCreateTbReq;
void * ptr = vnodeMalloc(pVnode, pMsg->contLen); void * ptr = vnodeMalloc(pVnode, pMsg->contLen);
if (ptr == NULL) { if (ptr == NULL) {
@ -70,6 +68,7 @@ int vnodeApplyWMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
switch (pMsg->msgType) { switch (pMsg->msgType) {
case TDMT_VND_CREATE_STB: case TDMT_VND_CREATE_STB:
case TDMT_VND_CREATE_TABLE:
tDeserializeSVCreateTbReq(POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)), &vCreateTbReq); tDeserializeSVCreateTbReq(POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)), &vCreateTbReq);
if (metaCreateTable(pVnode->pMeta, &(vCreateTbReq)) < 0) { if (metaCreateTable(pVnode->pMeta, &(vCreateTbReq)) < 0) {
// TODO: handle error // TODO: handle error
@ -79,9 +78,9 @@ int vnodeApplyWMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
break; break;
case TDMT_VND_DROP_STB: case TDMT_VND_DROP_STB:
case TDMT_VND_DROP_TABLE: case TDMT_VND_DROP_TABLE:
if (metaDropTable(pVnode->pMeta, vReq.dtReq.uid) < 0) { // if (metaDropTable(pVnode->pMeta, vReq.dtReq.uid) < 0) {
// TODO: handle error // // TODO: handle error
} // }
break; break;
case TDMT_VND_SUBMIT: case TDMT_VND_SUBMIT:
if (tsdbInsertData(pVnode->pTsdb, (SSubmitMsg *)ptr) < 0) { if (tsdbInsertData(pVnode->pTsdb, (SSubmitMsg *)ptr) < 0) {

84
source/dnode/vnode/meta/src/metaBDBImpl.c
View File

@ -351,7 +351,7 @@ static int metaCtbIdxCb(DB *pIdx, const DBT *pKey, const DBT *pValue, DBT *pSKey
pDbt[0].size = sizeof(pTbCfg->ctbCfg.suid); pDbt[0].size = sizeof(pTbCfg->ctbCfg.suid);
// Second key is the first tag // Second key is the first tag
void *pTagVal = tdGetKVRowValOfCol(pTbCfg->ctbCfg.pTag, 0); void *pTagVal = tdGetKVRowValOfCol(pTbCfg->ctbCfg.pTag, (kvRowColIdx(pTbCfg->ctbCfg.pTag))[0].colId);
pDbt[1].data = varDataVal(pTagVal); pDbt[1].data = varDataVal(pTagVal);
pDbt[1].size = varDataLen(pTagVal); pDbt[1].size = varDataLen(pTagVal);
@ -403,10 +403,10 @@ static void *metaDecodeTbInfo(void *buf, STbCfg *pTbCfg) {
buf = taosDecodeFixedU8(buf, &(pTbCfg->type)); buf = taosDecodeFixedU8(buf, &(pTbCfg->type));
if (pTbCfg->type == META_SUPER_TABLE) { if (pTbCfg->type == META_SUPER_TABLE) {
buf = taosDecodeVariantU32(buf, pTbCfg->stbCfg.nTagCols); buf = taosDecodeVariantU32(buf, &(pTbCfg->stbCfg.nTagCols));
pTbCfg->stbCfg.pTagSchema = (SSchema *)malloc(sizeof(SSchema) * pTbCfg->stbCfg.nTagCols); pTbCfg->stbCfg.pTagSchema = (SSchema *)malloc(sizeof(SSchema) * pTbCfg->stbCfg.nTagCols);
for (uint32_t i = 0; i < pTbCfg->stbCfg.nTagCols; i++) { for (uint32_t i = 0; i < pTbCfg->stbCfg.nTagCols; i++) {
buf = taosDecodeFixedI8(buf, &pTbCfg->stbCfg.pSchema[i].type); buf = taosDecodeFixedI8(buf, &(pTbCfg->stbCfg.pSchema[i].type));
buf = taosDecodeFixedI32(buf, &pTbCfg->stbCfg.pSchema[i].colId); buf = taosDecodeFixedI32(buf, &pTbCfg->stbCfg.pSchema[i].colId);
buf = taosDecodeFixedI32(buf, &pTbCfg->stbCfg.pSchema[i].bytes); buf = taosDecodeFixedI32(buf, &pTbCfg->stbCfg.pSchema[i].bytes);
buf = taosDecodeStringTo(buf, pTbCfg->stbCfg.pSchema[i].name); buf = taosDecodeStringTo(buf, pTbCfg->stbCfg.pSchema[i].name);
@ -429,3 +429,81 @@ static void metaClearTbCfg(STbCfg *pTbCfg) {
tfree(pTbCfg->ctbCfg.pTag); tfree(pTbCfg->ctbCfg.pTag);
} }
} }
/* ------------------------ FOR QUERY ------------------------ */
int metaGetTableInfo(SMeta *pMeta, char *tbname, STableMetaMsg **ppMsg) {
DBT key = {0};
DBT value = {0};
SMetaDB * pMetaDB = pMeta->pDB;
int ret;
STbCfg tbCfg;
SSchemaKey schemaKey;
DBT key1 = {0};
DBT value1 = {0};
uint32_t ncols;
void * pBuf;
int tlen;
STableMetaMsg *pMsg;
key.data = tbname;
key.size = strlen(tbname) + 1;
ret = pMetaDB->pNameIdx->get(pMetaDB->pNameIdx, NULL, &key, &value, 0);
if (ret != 0) {
// TODO
return -1;
}
metaDecodeTbInfo(value.data, &tbCfg);
switch (tbCfg.type) {
case META_SUPER_TABLE:
schemaKey.uid = tbCfg.stbCfg.suid;
schemaKey.sver = 0;
key1.data = &schemaKey;
key1.size = sizeof(schemaKey);
ret = pMetaDB->pSchemaDB->get(pMetaDB->pSchemaDB, &key1, &value1, NULL, 0);
if (ret != 0) {
// TODO
return -1;
}
pBuf = value1.data;
pBuf = taosDecodeFixedU32(pBuf, &ncols);
tlen = sizeof(STableMetaMsg) + (tbCfg.stbCfg.nTagCols + ncols) * sizeof(SSchema);
pMsg = calloc(1, tlen);
if (pMsg == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
strcpy(pMsg->tbFname, tbCfg.name);
pMsg->numOfTags = tbCfg.stbCfg.nTagCols;
pMsg->numOfColumns = ncols;
pMsg->tableType = tbCfg.type;
pMsg->sversion = 0;
pMsg->tversion = 0;
pMsg->suid = tbCfg.stbCfg.suid;
pMsg->tuid = tbCfg.stbCfg.suid;
for (size_t i = 0; i < tbCfg.stbCfg.nTagCols; i++) {
}
break;
case META_CHILD_TABLE:
ASSERT(0);
break;
case META_NORMAL_TABLE:
ASSERT(0);
break;
default:
ASSERT(0);
break;
}
*ppMsg = pMsg;
return 0;
}

54
source/dnode/vnode/meta/src/metaTbCfg.c
View File

@ -47,57 +47,3 @@ size_t metaEncodeTbObjFromTbOptions(const STbCfg *pTbOptions, void *pBuf, size_t
return tlen; return tlen;
} }
int metaEncodeTbCfg(void **pBuf, STbCfg *pTbCfg) {
int tsize = 0;
tsize += taosEncodeString(pBuf, pTbCfg->name);
tsize += taosEncodeFixedU32(pBuf, pTbCfg->ttl);
tsize += taosEncodeFixedU32(pBuf, pTbCfg->keep);
tsize += taosEncodeFixedU8(pBuf, pTbCfg->type);
switch (pTbCfg->type) {
case META_SUPER_TABLE:
tsize += taosEncodeFixedU64(pBuf, pTbCfg->stbCfg.suid);
tsize += tdEncodeSchema(pBuf, pTbCfg->stbCfg.pSchema);
tsize += tdEncodeSchema(pBuf, pTbCfg->stbCfg.pTagSchema);
break;
case META_CHILD_TABLE:
tsize += taosEncodeFixedU64(pBuf, pTbCfg->ctbCfg.suid);
tsize += tdEncodeKVRow(pBuf, pTbCfg->ctbCfg.pTag);
break;
case META_NORMAL_TABLE:
tsize += tdEncodeSchema(pBuf, pTbCfg->ntbCfg.pSchema);
break;
default:
break;
}
return tsize;
}
void *metaDecodeTbCfg(void *pBuf, STbCfg *pTbCfg) {
pBuf = taosDecodeString(pBuf, &(pTbCfg->name));
pBuf = taosDecodeFixedU32(pBuf, &(pTbCfg->ttl));
pBuf = taosDecodeFixedU32(pBuf, &(pTbCfg->keep));
pBuf = taosDecodeFixedU8(pBuf, &(pTbCfg->type));
switch (pTbCfg->type) {
case META_SUPER_TABLE:
pBuf = taosDecodeFixedU64(pBuf, &(pTbCfg->stbCfg.suid));
pBuf = tdDecodeSchema(pBuf, &(pTbCfg->stbCfg.pSchema));
pBuf = tdDecodeSchema(pBuf, &(pTbCfg->stbCfg.pTagSchema));
break;
case META_CHILD_TABLE:
pBuf = taosDecodeFixedU64(pBuf, &(pTbCfg->ctbCfg.suid));
pBuf = tdDecodeKVRow(pBuf, &(pTbCfg->ctbCfg.pTag));
break;
case META_NORMAL_TABLE:
pBuf = tdDecodeSchema(pBuf, &(pTbCfg->ntbCfg.pSchema));
break;
default:
break;
}
return pBuf;
}

33
source/libs/parser/src/dCDAstProcess.c
View File

@ -17,7 +17,8 @@ static bool has(SArray* pFieldList, int32_t startIndex, const char* name) {
return false; return false;
} }
static int32_t setShowInfo(SShowInfo* pShowInfo, SParseBasicCtx *pCtx, void** output, int32_t* outputLen, SMsgBuf* pMsgBuf) { static int32_t setShowInfo(SShowInfo* pShowInfo, SParseBasicCtx* pCtx, void** output, int32_t* outputLen,
SMsgBuf* pMsgBuf) {
const char* msg1 = "invalid name"; const char* msg1 = "invalid name";
const char* msg2 = "wildcard string should be less than %d characters"; const char* msg2 = "wildcard string should be less than %d characters";
const char* msg3 = "database name too long"; const char* msg3 = "database name too long";
@ -116,8 +117,8 @@ static int32_t doCheckDbOptions(SCreateDbMsg* pCreate, SMsgBuf* pMsgBuf) {
int32_t val = htonl(pCreate->daysPerFile); int32_t val = htonl(pCreate->daysPerFile);
if (val != -1 && (val < TSDB_MIN_DAYS_PER_FILE || val > TSDB_MAX_DAYS_PER_FILE)) { if (val != -1 && (val < TSDB_MIN_DAYS_PER_FILE || val > TSDB_MAX_DAYS_PER_FILE)) {
snprintf(msg, tListLen(msg), "invalid db option daysPerFile: %d valid range: [%d, %d]", val, snprintf(msg, tListLen(msg), "invalid db option daysPerFile: %d valid range: [%d, %d]", val, TSDB_MIN_DAYS_PER_FILE,
TSDB_MIN_DAYS_PER_FILE, TSDB_MAX_DAYS_PER_FILE); TSDB_MAX_DAYS_PER_FILE);
return buildInvalidOperationMsg(pMsgBuf, msg); return buildInvalidOperationMsg(pMsgBuf, msg);
} }
@ -137,15 +138,15 @@ static int32_t doCheckDbOptions(SCreateDbMsg* pCreate, SMsgBuf* pMsgBuf) {
val = htonl(pCreate->commitTime); val = htonl(pCreate->commitTime);
if (val != -1 && (val < TSDB_MIN_COMMIT_TIME || val > TSDB_MAX_COMMIT_TIME)) { if (val != -1 && (val < TSDB_MIN_COMMIT_TIME || val > TSDB_MAX_COMMIT_TIME)) {
snprintf(msg, tListLen(msg), "invalid db option commitTime: %d valid range: [%d, %d]", val, snprintf(msg, tListLen(msg), "invalid db option commitTime: %d valid range: [%d, %d]", val, TSDB_MIN_COMMIT_TIME,
TSDB_MIN_COMMIT_TIME, TSDB_MAX_COMMIT_TIME); TSDB_MAX_COMMIT_TIME);
return buildInvalidOperationMsg(pMsgBuf, msg); return buildInvalidOperationMsg(pMsgBuf, msg);
} }
val = htonl(pCreate->fsyncPeriod); val = htonl(pCreate->fsyncPeriod);
if (val != -1 && (val < TSDB_MIN_FSYNC_PERIOD || val > TSDB_MAX_FSYNC_PERIOD)) { if (val != -1 && (val < TSDB_MIN_FSYNC_PERIOD || val > TSDB_MAX_FSYNC_PERIOD)) {
snprintf(msg, tListLen(msg), "invalid db option fsyncPeriod: %d valid range: [%d, %d]", val, snprintf(msg, tListLen(msg), "invalid db option fsyncPeriod: %d valid range: [%d, %d]", val, TSDB_MIN_FSYNC_PERIOD,
TSDB_MIN_FSYNC_PERIOD, TSDB_MAX_FSYNC_PERIOD); TSDB_MAX_FSYNC_PERIOD);
return buildInvalidOperationMsg(pMsgBuf, msg); return buildInvalidOperationMsg(pMsgBuf, msg);
} }
@ -284,7 +285,8 @@ int32_t doCheckForCreateTable(SSqlInfo* pInfo, SMsgBuf* pMsgBuf) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t doCheckForCreateCTable(SSqlInfo* pInfo, SParseBasicCtx *pCtx, SMsgBuf* pMsgBuf, char** pOutput, int32_t* len, SEpSet* pEpSet) { int32_t doCheckForCreateCTable(SSqlInfo* pInfo, SParseBasicCtx* pCtx, SMsgBuf* pMsgBuf, char** pOutput, int32_t* len,
SEpSet* pEpSet) {
const char* msg1 = "invalid table name"; const char* msg1 = "invalid table name";
const char* msg2 = "tags number not matched"; const char* msg2 = "tags number not matched";
const char* msg3 = "tag value too long"; const char* msg3 = "tag value too long";
@ -478,11 +480,11 @@ int32_t doCheckForCreateCTable(SSqlInfo* pInfo, SParseBasicCtx *pCtx, SMsgBuf* p
req.ctbCfg.suid = pSuperTableMeta->suid; req.ctbCfg.suid = pSuperTableMeta->suid;
req.ctbCfg.pTag = row; req.ctbCfg.pTag = row;
int32_t serLen = tSerializeSVCreateTbReq(NULL, &req); int32_t serLen = sizeof(SMsgHead) + tSerializeSVCreateTbReq(NULL, &req);
char* buf1 = calloc(1, serLen); char* buf1 = calloc(1, serLen);
char* p = buf1; *pOutput = buf1;
buf1 += sizeof(SMsgHead);
tSerializeSVCreateTbReq((void*)&buf1, &req); tSerializeSVCreateTbReq((void*)&buf1, &req);
*pOutput = p;
*len = serLen; *len = serLen;
SVgroupInfo info = {0}; SVgroupInfo info = {0};
@ -494,12 +496,15 @@ int32_t doCheckForCreateCTable(SSqlInfo* pInfo, SParseBasicCtx *pCtx, SMsgBuf* p
pEpSet->port[i] = info.epAddr[i].port; pEpSet->port[i] = info.epAddr[i].port;
tstrncpy(pEpSet->fqdn[i], info.epAddr[i].fqdn, tListLen(pEpSet->fqdn[i])); tstrncpy(pEpSet->fqdn[i], info.epAddr[i].fqdn, tListLen(pEpSet->fqdn[i]));
} }
((SMsgHead*)(*pOutput))->vgId = htonl(info.vgId);
((SMsgHead*)(*pOutput))->contLen = htonl(serLen);
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t qParserValidateDclSqlNode(SSqlInfo* pInfo, SParseBasicCtx* pCtx, SDclStmtInfo* pDcl, char* msgBuf, int32_t msgBufLen) { int32_t qParserValidateDclSqlNode(SSqlInfo* pInfo, SParseBasicCtx* pCtx, SDclStmtInfo* pDcl, char* msgBuf,
int32_t msgBufLen) {
int32_t code = 0; int32_t code = 0;
SMsgBuf m = {.buf = msgBuf, .len = msgBufLen}; SMsgBuf m = {.buf = msgBuf, .len = msgBufLen};
@ -690,7 +695,8 @@ int32_t qParserValidateDclSqlNode(SSqlInfo* pInfo, SParseBasicCtx* pCtx, SDclStm
pDcl->pMsg = (char*)buildCreateTableMsg(pCreateTable, &pDcl->msgLen, pCtx, pMsgBuf); pDcl->pMsg = (char*)buildCreateTableMsg(pCreateTable, &pDcl->msgLen, pCtx, pMsgBuf);
pDcl->msgType = (pCreateTable->type == TSQL_CREATE_TABLE) ? TDMT_VND_CREATE_TABLE : TDMT_MND_CREATE_STB; pDcl->msgType = (pCreateTable->type == TSQL_CREATE_TABLE) ? TDMT_VND_CREATE_TABLE : TDMT_MND_CREATE_STB;
} else if (pCreateTable->type == TSQL_CREATE_CTABLE) { } else if (pCreateTable->type == TSQL_CREATE_CTABLE) {
if ((code = doCheckForCreateCTable(pInfo, pCtx, pMsgBuf, &pDcl->pMsg, &pDcl->msgLen, &pDcl->epSet)) != TSDB_CODE_SUCCESS) { if ((code = doCheckForCreateCTable(pInfo, pCtx, pMsgBuf, &pDcl->pMsg, &pDcl->msgLen, &pDcl->epSet)) !=
TSDB_CODE_SUCCESS) {
return code; return code;
} }
@ -739,4 +745,3 @@ int32_t qParserValidateDclSqlNode(SSqlInfo* pInfo, SParseBasicCtx* pCtx, SDclStm
return code; return code;
} }

38
source/util/src/tbuffer.c
View File

@ -13,9 +13,9 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#include "os.h"
#include "tbuffer.h" #include "tbuffer.h"
#include "exception.h" #include "exception.h"
#include "os.h"
//#include "taoserror.h" //#include "taoserror.h"
typedef union Un4B { typedef union Un4B {
@ -281,46 +281,28 @@ void tbufWriteStringLen( SBufferWriter* buf, const char* str, size_t len ) {
tbufWriteChar(buf, '\0'); tbufWriteChar(buf, '\0');
} }
void tbufWriteString( SBufferWriter* buf, const char* str ) { void tbufWriteString(SBufferWriter* buf, const char* str) { tbufWriteStringLen(buf, str, strlen(str)); }
tbufWriteStringLen( buf, str, strlen(str) );
}
void tbufWriteBinary(SBufferWriter* buf, const void* data, size_t len) { void tbufWriteBinary(SBufferWriter* buf, const void* data, size_t len) {
tbufWriteLength(buf, len); tbufWriteLength(buf, len);
tbufWrite(buf, data, len); tbufWrite(buf, data, len);
} }
void tbufWriteBool( SBufferWriter* buf, bool data ) { void tbufWriteBool(SBufferWriter* buf, bool data) { tbufWrite(buf, &data, sizeof(data)); }
tbufWrite( buf, &data, sizeof(data) );
}
void tbufWriteBoolAt( SBufferWriter* buf, size_t pos, bool data ) { void tbufWriteBoolAt(SBufferWriter* buf, size_t pos, bool data) { tbufWriteAt(buf, pos, &data, sizeof(data)); }
tbufWriteAt( buf, pos, &data, sizeof(data) );
}
void tbufWriteChar( SBufferWriter* buf, char data ) { void tbufWriteChar(SBufferWriter* buf, char data) { tbufWrite(buf, &data, sizeof(data)); }
tbufWrite( buf, &data, sizeof(data) );
}
void tbufWriteCharAt( SBufferWriter* buf, size_t pos, char data ) { void tbufWriteCharAt(SBufferWriter* buf, size_t pos, char data) { tbufWriteAt(buf, pos, &data, sizeof(data)); }
tbufWriteAt( buf, pos, &data, sizeof(data) );
}
void tbufWriteInt8( SBufferWriter* buf, int8_t data ) { void tbufWriteInt8(SBufferWriter* buf, int8_t data) { tbufWrite(buf, &data, sizeof(data)); }
tbufWrite( buf, &data, sizeof(data) );
}
void tbufWriteInt8At( SBufferWriter* buf, size_t pos, int8_t data ) { void tbufWriteInt8At(SBufferWriter* buf, size_t pos, int8_t data) { tbufWriteAt(buf, pos, &data, sizeof(data)); }
tbufWriteAt( buf, pos, &data, sizeof(data) );
}
void tbufWriteUint8( SBufferWriter* buf, uint8_t data ) { void tbufWriteUint8(SBufferWriter* buf, uint8_t data) { tbufWrite(buf, &data, sizeof(data)); }
tbufWrite( buf, &data, sizeof(data) );
}
void tbufWriteUint8At( SBufferWriter* buf, size_t pos, uint8_t data ) { void tbufWriteUint8At(SBufferWriter* buf, size_t pos, uint8_t data) { tbufWriteAt(buf, pos, &data, sizeof(data)); }
tbufWriteAt( buf, pos, &data, sizeof(data) );
}
void tbufWriteInt16(SBufferWriter* buf, int16_t data) { void tbufWriteInt16(SBufferWriter* buf, int16_t data) {
if (buf->endian) { if (buf->endian) {