Skip to content

TDengine
TDengine

taosdata / TDengine
大约 2 年 前同步成功

通知 1192
Star 22018
Fork 4786
TDengine
TDengine
提交 97f64a58 编写于 作者: D dapan
浏览文件
操作

Merge remote-tracking branch 'origin/feature/vnode_refact1' into feature/qnode

上级 005ff0ed 607d5c45
展开全部 隐藏空白更改
内联 并排
Showing with 3098 addition and 3301 deletion
include/util/tskiplist2.h → include/common/tdata.h
浏览文件 @ 97f64a58
......@@ -12,59 +12,34 @@
* 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_SKIPLIST2_H_
#define _TD_UTIL_SKIPLIST2_H_
#include "os.h"
#ifndef _TD_TDATA_H_
#define _TD_TDATA_H_
#ifdef __cplusplus
extern "C" {
#endif
#define SL_MAX_LEVEL 15
typedef struct SSkipList2 SSkipList2;
typedef struct SSLCursor SSLCursor;
typedef struct SSLCfg SSLCfg;
typedef struct SSLNode SSLNode;
typedef int32_t (*tslCmprFn)(const void *pKey1, int32_t nKey1, const void *pKey2, int32_t nKey2);
// SSkipList2
int32_t slOpen(const SSLCfg *pCfg, SSkipList2 **ppSl);
int32_t slClose(SSkipList2 *pSl);
int32_t slClear(SSkipList2 *pSl);
#include "os.h"
// SSLCursor
int32_t slcOpen(SSkipList2 *pSl, SSLCursor *pSlc);
int32_t slcClose(SSLCursor *pSlc);
int32_t slcMoveTo(SSLCursor *pSlc, const void *pKey, int32_t nKey);
int32_t slcMoveToNext(SSLCursor *pSlc);
int32_t slcMoveToPrev(SSLCursor *pSlc);
int32_t slcMoveToFirst(SSLCursor *pSlc);
int32_t slcMoveToLast(SSLCursor *pSlc);
int32_t slcPut(SSLCursor *pSlc, const void *pKey, int32_t nKey, const void *pData, int32_t nData);
int32_t slcGet(SSLCursor *pSlc, const void **ppKey, int32_t *nKey, const void **ppData, int32_t *nData);
int32_t slcDrop(SSLCursor *pSlc);
typedef struct STaosData TDATA, tdata_t;
// struct
struct SSLCfg {
int8_t maxLevel;
int32_t nKey;
int32_t nData;
tslCmprFn cmprFn;
void *pPool;
void *(*xMalloc)(void *, int32_t size);
void (*xFree)(void *, void *);
};
typedef enum {
TAOS_META_STABLE_DATA = 0, // super table meta
TAOS_META_TABLE_DATA, // non-super table meta
TAOS_TS_ROW_DATA, // row time-series data
TAOS_TS_COL_DATA, // col time-series data
TAOS_DATA_MAX
} ETaosDataT;
struct SSLCursor {
SSkipList2 *pSl;
SSLNode **forwards[SL_MAX_LEVEL];
struct STaosData {
ETaosDataT type;
uint32_t nPayload;
uint8_t *pPayload;
};
#ifdef __cplusplus
}
#endif
#endif /*_TD_UTIL_SKIPLIST2_H_*/
\ No newline at end of file
#endif /*_TD_TDATA_H_*/
\ No newline at end of file
include/common/tdataformat.h
浏览文件 @ 97f64a58
......@@ -150,29 +150,6 @@ int32_t tdAddColToSchema(STSchemaBuilder *pBuilder, int8_t type, int8_t flags,
STSchema *tdGetSchemaFromBuilder(STSchemaBuilder *pBuilder);
// ----------------- Semantic timestamp key definition
#ifdef TD_2_0
typedef uint64_t TKEY;
#define TKEY_INVALID UINT64_MAX
#define TKEY_NULL TKEY_INVALID
#define TKEY_NEGATIVE_FLAG (((TKEY)1) << 63)
#define TKEY_DELETE_FLAG (((TKEY)1) << 62)
#define TKEY_VALUE_FILTER (~(TKEY_NEGATIVE_FLAG | TKEY_DELETE_FLAG))
#define TKEY_IS_NEGATIVE(tkey) (((tkey)&TKEY_NEGATIVE_FLAG) != 0)
#define TKEY_IS_DELETED(tkey) (((tkey)&TKEY_DELETE_FLAG) != 0)
#define tdSetTKEYDeleted(tkey) ((tkey) | TKEY_DELETE_FLAG)
#define tdGetTKEY(key) (((TKEY)TABS(key)) | (TKEY_NEGATIVE_FLAG & (TKEY)(key)))
#define tdGetKey(tkey) (((TSKEY)((tkey)&TKEY_VALUE_FILTER)) * (TKEY_IS_NEGATIVE(tkey) ? -1 : 1))
#define MIN_TS_KEY ((TSKEY)0x8000000000000001)
#define MAX_TS_KEY ((TSKEY)0x3fffffffffffffff)
#define TD_TO_TKEY(key) tdGetTKEY(((key) < MIN_TS_KEY) ? MIN_TS_KEY : (((key) > MAX_TS_KEY) ? MAX_TS_KEY : key))
#else
// typedef uint64_t TKEY;
#define TKEY TSKEY
......@@ -192,8 +169,6 @@ typedef uint64_t TKEY;
#define TD_TO_TKEY(key) tdGetTKEY(((key) < MIN_TS_KEY) ? MIN_TS_KEY : (((key) > MAX_TS_KEY) ? MAX_TS_KEY : key))
#endif
static FORCE_INLINE TKEY keyToTkey(TSKEY key) {
TSKEY lkey = key;
if (key > MAX_TS_KEY) {
......@@ -218,157 +193,6 @@ static FORCE_INLINE int32_t tkeyComparFn(const void *tkey1, const void *tkey2) {
}
}
#if 0
// ----------------- Data row structure
/* A data row, the format is like below:
* |<------------------------------------------------ len ---------------------------------->|
* |<-- Head -->|<--------- flen -------------->| |
* +---------------------+---------------------------------+---------------------------------+
* | uint16_t | int16_t | | |
* +----------+----------+---------------------------------+---------------------------------+
* | len | sversion | First part | Second part |
* +----------+----------+---------------------------------+---------------------------------+
*
* NOTE: timestamp in this row structure is TKEY instead of TSKEY
*/
typedef void *SDataRow;
#define TD_DATA_ROW_HEAD_SIZE (sizeof(uint16_t) + sizeof(int16_t))
#define dataRowLen(r) (*(TDRowLenT *)(r)) // 0~65535
#define dataRowEnd(r) POINTER_SHIFT(r, dataRowLen(r))
#define dataRowVersion(r) (*(int16_t *)POINTER_SHIFT(r, sizeof(int16_t)))
#define dataRowTuple(r) POINTER_SHIFT(r, TD_DATA_ROW_HEAD_SIZE)
#define dataRowTKey(r) (*(TKEY *)(dataRowTuple(r)))
#define dataRowKey(r) tdGetKey(dataRowTKey(r))
#define dataRowSetLen(r, l) (dataRowLen(r) = (l))
#define dataRowSetVersion(r, v) (dataRowVersion(r) = (v))
#define dataRowCpy(dst, r) memcpy((dst), (r), dataRowLen(r))
#define dataRowMaxBytesFromSchema(s) (schemaTLen(s) + TD_DATA_ROW_HEAD_SIZE)
#define dataRowDeleted(r) TKEY_IS_DELETED(dataRowTKey(r))
SDataRow tdNewDataRowFromSchema(STSchema *pSchema);
void tdFreeDataRow(SDataRow row);
void tdInitDataRow(SDataRow row, STSchema *pSchema);
SDataRow tdDataRowDup(SDataRow row);
// offset here not include dataRow header length
static FORCE_INLINE int32_t tdAppendDataColVal(SDataRow row, const void *value, bool isCopyVarData, int8_t type,
int32_t offset) {
assert(value != NULL);
int32_t toffset = offset + TD_DATA_ROW_HEAD_SIZE;
if (IS_VAR_DATA_TYPE(type)) {
*(VarDataOffsetT *)POINTER_SHIFT(row, toffset) = dataRowLen(row);
if (isCopyVarData) {
memcpy(POINTER_SHIFT(row, dataRowLen(row)), value, varDataTLen(value));
}
dataRowLen(row) += varDataTLen(value);
} else {
if (offset == 0) {
assert(type == TSDB_DATA_TYPE_TIMESTAMP);
TKEY tvalue = tdGetTKEY(*(TSKEY *)value);
memcpy(POINTER_SHIFT(row, toffset), (const void *)(&tvalue), TYPE_BYTES[type]);
} else {
memcpy(POINTER_SHIFT(row, toffset), value, TYPE_BYTES[type]);
}
}
return 0;
}
// offset here not include dataRow header length
static FORCE_INLINE int32_t tdAppendColVal(SDataRow row, const void *value, int8_t type, int32_t offset) {
return tdAppendDataColVal(row, value, true, type, offset);
}
// NOTE: offset here including the header size
static FORCE_INLINE void *tdGetRowDataOfCol(SDataRow row, int8_t type, int32_t offset) {
if (IS_VAR_DATA_TYPE(type)) {
return POINTER_SHIFT(row, *(VarDataOffsetT *)POINTER_SHIFT(row, offset));
} else {
return POINTER_SHIFT(row, offset);
}
}
static FORCE_INLINE void *tdGetPtrToCol(SDataRow row, STSchema *pSchema, int32_t idx) {
return POINTER_SHIFT(row, TD_DATA_ROW_HEAD_SIZE + pSchema->columns[idx].offset);
}
static FORCE_INLINE void *tdGetColOfRowBySchema(SDataRow row, STSchema *pSchema, int32_t idx) {
int16_t offset = TD_DATA_ROW_HEAD_SIZE + pSchema->columns[idx].offset;
int8_t type = pSchema->columns[idx].type;
return tdGetRowDataOfCol(row, type, offset);
}
static FORCE_INLINE bool tdIsColOfRowNullBySchema(SDataRow row, STSchema *pSchema, int32_t idx) {
int16_t offset = TD_DATA_ROW_HEAD_SIZE + pSchema->columns[idx].offset;
int8_t type = pSchema->columns[idx].type;
return isNull(tdGetRowDataOfCol(row, type, offset), type);
}
static FORCE_INLINE void tdSetColOfRowNullBySchema(SDataRow row, STSchema *pSchema, int32_t idx) {
int16_t offset = TD_DATA_ROW_HEAD_SIZE + pSchema->columns[idx].offset;
int8_t type = pSchema->columns[idx].type;
int16_t bytes = pSchema->columns[idx].bytes;
setNull(tdGetRowDataOfCol(row, type, offset), type, bytes);
}
static FORCE_INLINE void tdCopyColOfRowBySchema(SDataRow dst, STSchema *pDstSchema, int32_t dstIdx, SDataRow src,
STSchema *pSrcSchema, int32_t srcIdx) {
int8_t type = pDstSchema->columns[dstIdx].type;
assert(type == pSrcSchema->columns[srcIdx].type);
void *pData = tdGetPtrToCol(dst, pDstSchema, dstIdx);
void *value = tdGetPtrToCol(src, pSrcSchema, srcIdx);
switch (type) {
case TSDB_DATA_TYPE_BINARY:
case TSDB_DATA_TYPE_NCHAR:
*(VarDataOffsetT *)pData = *(VarDataOffsetT *)value;
pData = POINTER_SHIFT(dst, *(VarDataOffsetT *)pData);
value = POINTER_SHIFT(src, *(VarDataOffsetT *)value);
memcpy(pData, value, varDataTLen(value));
break;
case TSDB_DATA_TYPE_NULL:
case TSDB_DATA_TYPE_BOOL:
case TSDB_DATA_TYPE_TINYINT:
case TSDB_DATA_TYPE_UTINYINT:
*(uint8_t *)pData = *(uint8_t *)value;
break;
case TSDB_DATA_TYPE_SMALLINT:
case TSDB_DATA_TYPE_USMALLINT:
*(uint16_t *)pData = *(uint16_t *)value;
break;
case TSDB_DATA_TYPE_INT:
case TSDB_DATA_TYPE_UINT:
*(uint32_t *)pData = *(uint32_t *)value;
break;
case TSDB_DATA_TYPE_BIGINT:
case TSDB_DATA_TYPE_UBIGINT:
*(uint64_t *)pData = *(uint64_t *)value;
break;
case TSDB_DATA_TYPE_FLOAT:
SET_FLOAT_PTR(pData, value);
break;
case TSDB_DATA_TYPE_DOUBLE:
SET_DOUBLE_PTR(pData, value);
break;
case TSDB_DATA_TYPE_TIMESTAMP:
if (pSrcSchema->columns[srcIdx].colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
*(TSKEY *)pData = tdGetKey(*(TKEY *)value);
} else {
*(TSKEY *)pData = *(TSKEY *)value;
}
break;
default:
memcpy(pData, value, pSrcSchema->columns[srcIdx].bytes);
}
}
#endif
// ----------------- Data column structure
// SDataCol arrangement: data => bitmap => dataOffset
typedef struct SDataCol {
......@@ -398,29 +222,6 @@ void *dataColSetOffset(SDataCol *pCol, int32_t nEle);
bool isNEleNull(SDataCol *pCol, int32_t nEle);
#if 0
// Get the data pointer from a column-wised data
static FORCE_INLINE const void *tdGetColDataOfRow(SDataCol *pCol, int32_t row) {
if (isAllRowsNull(pCol)) {
return getNullValue(pCol->type);
}
if (IS_VAR_DATA_TYPE(pCol->type)) {
return POINTER_SHIFT(pCol->pData, pCol->dataOff[row]);
} else {
return POINTER_SHIFT(pCol->pData, TYPE_BYTES[pCol->type] * row);
}
}
static FORCE_INLINE int32_t dataColGetNEleLen(SDataCol *pDataCol, int32_t rows) {
assert(rows > 0);
if (IS_VAR_DATA_TYPE(pDataCol->type)) {
return pDataCol->dataOff[rows - 1] + varDataTLen(tdGetColDataOfRow(pDataCol, rows - 1));
} else {
return TYPE_BYTES[pDataCol->type] * rows;
}
}
#endif
typedef struct {
col_id_t maxCols; // max number of columns
col_id_t numOfCols; // Total number of cols
......@@ -479,7 +280,8 @@ void tdResetDataCols(SDataCols *pCols);
int32_t tdInitDataCols(SDataCols *pCols, STSchema *pSchema);
SDataCols *tdDupDataCols(SDataCols *pCols, bool keepData);
SDataCols *tdFreeDataCols(SDataCols *pCols);
int32_t tdMergeDataCols(SDataCols *target, SDataCols *source, int32_t rowsToMerge, int32_t *pOffset, bool update, TDRowVerT maxVer);
int32_t tdMergeDataCols(SDataCols *target, SDataCols *source, int32_t rowsToMerge, int32_t *pOffset, bool update,
TDRowVerT maxVer);
// ----------------- K-V data row structure
/* |<-------------------------------------- len -------------------------------------------->|
......@@ -542,54 +344,6 @@ static FORCE_INLINE void *tdGetKVRowIdxOfCol(SKVRow row, int16_t colId) {
return taosbsearch(&colId, kvRowColIdx(row), kvRowNCols(row), sizeof(SColIdx), comparTagId, TD_EQ);
}
#if 0
// offset here not include kvRow header length
static FORCE_INLINE int32_t tdAppendKvColVal(SKVRow row, const void *value, bool isCopyValData, int16_t colId, int8_t type,
int32_t offset) {
assert(value != NULL);
int32_t toffset = offset + TD_KV_ROW_HEAD_SIZE;
SColIdx *pColIdx = (SColIdx *)POINTER_SHIFT(row, toffset);
char * ptr = (char *)POINTER_SHIFT(row, kvRowLen(row));
pColIdx->colId = colId;
pColIdx->offset = kvRowLen(row); // offset of pColIdx including the TD_KV_ROW_HEAD_SIZE
if (IS_VAR_DATA_TYPE(type)) {
if (isCopyValData) {
memcpy(ptr, value, varDataTLen(value));
}
kvRowLen(row) += varDataTLen(value);
} else {
if (offset == 0) {
assert(type == TSDB_DATA_TYPE_TIMESTAMP);
TKEY tvalue = tdGetTKEY(*(TSKEY *)value);
memcpy(ptr, (void *)(&tvalue), TYPE_BYTES[type]);
} else {
memcpy(ptr, value, TYPE_BYTES[type]);
}
kvRowLen(row) += TYPE_BYTES[type];
}
return 0;
}
// NOTE: offset here including the header size
static FORCE_INLINE void *tdGetKvRowDataOfCol(void *row, int32_t offset) { return POINTER_SHIFT(row, offset); }
static FORCE_INLINE void *tdGetKVRowValOfColEx(SKVRow row, int16_t colId, int32_t *nIdx) {
while (*nIdx < kvRowNCols(row)) {
SColIdx *pColIdx = kvRowColIdxAt(row, *nIdx);
if (pColIdx->colId == colId) {
++(*nIdx);
return tdGetKvRowDataOfCol(row, pColIdx->offset);
} else if (pColIdx->colId > colId) {
return NULL;
} else {
++(*nIdx);
}
}
return NULL;
}
#endif
// ----------------- K-V data row builder
typedef struct {
int16_t tCols;
......@@ -632,166 +386,6 @@ static FORCE_INLINE int32_t tdAddColToKVRow(SKVRowBuilder *pBuilder, col_id_t co
return 0;
}
#if 0
// ----------------- SMemRow appended with tuple row structure
/*
* |---------|------------------------------------------------- len ---------------------------------->|
* |<-------- Head ------>|<--------- flen -------------->| |
* |---------+---------------------+---------------------------------+---------------------------------+
* | uint8_t | uint16_t | int16_t | | |
* |---------+----------+----------+---------------------------------+---------------------------------+
* | flag | len | sversion | First part | Second part |
* +---------+----------+----------+---------------------------------+---------------------------------+
*
* NOTE: timestamp in this row structure is TKEY instead of TSKEY
*/
// ----------------- SMemRow appended with extended K-V data row structure
/* |--------------------|------------------------------------------------ len ---------------------------------->|
* |<------------- Head ------------>|<--------- flen -------------->| |
* |--------------------+----------+--------------------------------------------+---------------------------------+
* | uint8_t | int16_t | uint16_t | int16_t | | |
* |---------+----------+----------+----------+---------------------------------+---------------------------------+
* | flag | sversion | len | ncols | cols index | data part |
* |---------+----------+----------+----------+---------------------------------+---------------------------------+
*/
typedef void *SMemRow;
#define TD_MEM_ROW_TYPE_SIZE sizeof(uint8_t)
#define TD_MEM_ROW_KV_VER_SIZE sizeof(int16_t)
#define TD_MEM_ROW_KV_TYPE_VER_SIZE (TD_MEM_ROW_TYPE_SIZE + TD_MEM_ROW_KV_VER_SIZE)
#define TD_MEM_ROW_DATA_HEAD_SIZE (TD_MEM_ROW_TYPE_SIZE + TD_DATA_ROW_HEAD_SIZE)
#define TD_MEM_ROW_KV_HEAD_SIZE (TD_MEM_ROW_TYPE_SIZE + TD_MEM_ROW_KV_VER_SIZE + TD_KV_ROW_HEAD_SIZE)
#define SMEM_ROW_DATA 0x0U // SDataRow
#define SMEM_ROW_KV 0x01U // SKVRow
#define KVRatioConvert (0.9f)
#define memRowType(r) ((*(uint8_t *)(r)) & 0x01)
#define memRowSetType(r, t) ((*(uint8_t *)(r)) = (t)) // set the total byte in case of dirty memory
#define isDataRowT(t) (SMEM_ROW_DATA == (((uint8_t)(t)) & 0x01))
#define isDataRow(r) (SMEM_ROW_DATA == memRowType(r))
#define isKvRowT(t) (SMEM_ROW_KV == (((uint8_t)(t)) & 0x01))
#define isKvRow(r) (SMEM_ROW_KV == memRowType(r))
#define isUtilizeKVRow(k, d) ((k) < ((d)*KVRatioConvert))
#define memRowDataBody(r) POINTER_SHIFT(r, TD_MEM_ROW_TYPE_SIZE) // section after flag
#define memRowKvBody(r) \
POINTER_SHIFT(r, TD_MEM_ROW_KV_TYPE_VER_SIZE) // section after flag + sversion as to reuse SKVRow
#define memRowDataLen(r) (*(TDRowLenT *)memRowDataBody(r)) // 0~65535
#define memRowKvLen(r) (*(TDRowLenT *)memRowKvBody(r)) // 0~65535
#define memRowDataTLen(r) \
((TDRowLenT)(memRowDataLen(r) + TD_MEM_ROW_TYPE_SIZE)) // using uint32_t/int32_t to store the TLen
#define memRowKvTLen(r) ((TDRowLenT)(memRowKvLen(r) + TD_MEM_ROW_KV_TYPE_VER_SIZE))
#define memRowLen(r) (isDataRow(r) ? memRowDataLen(r) : memRowKvLen(r))
#define memRowTLen(r) (isDataRow(r) ? memRowDataTLen(r) : memRowKvTLen(r)) // using uint32_t/int32_t to store the TLen
static FORCE_INLINE char *memRowEnd(SMemRow row) {
if (isDataRow(row)) {
return (char *)dataRowEnd(memRowDataBody(row));
} else {
return (char *)kvRowEnd(memRowKvBody(row));
}
}
#define memRowDataVersion(r) dataRowVersion(memRowDataBody(r))
#define memRowKvVersion(r) (*(int16_t *)POINTER_SHIFT(r, TD_MEM_ROW_TYPE_SIZE))
#define memRowVersion(r) (isDataRow(r) ? memRowDataVersion(r) : memRowKvVersion(r)) // schema version
#define memRowSetKvVersion(r, v) (memRowKvVersion(r) = (v))
#define memRowTuple(r) (isDataRow(r) ? dataRowTuple(memRowDataBody(r)) : kvRowValues(memRowKvBody(r)))
#define memRowTKey(r) (isDataRow(r) ? dataRowTKey(memRowDataBody(r)) : kvRowTKey(memRowKvBody(r)))
#define memRowKey(r) (isDataRow(r) ? dataRowKey(memRowDataBody(r)) : kvRowKey(memRowKvBody(r)))
#define memRowKeys(r) (isDataRow(r) ? dataRowTuple(memRowDataBody(r)) : kvRowKeys(memRowKvBody(r)))
#define memRowSetTKey(r, k) \
do { \
if (isDataRow(r)) { \
dataRowTKey(memRowDataBody(r)) = (k); \
} else { \
kvRowTKey(memRowKvBody(r)) = (k); \
} \
} while (0)
#define memRowSetLen(r, l) (isDataRow(r) ? memRowDataLen(r) = (l) : memRowKvLen(r) = (l))
#define memRowSetVersion(r, v) (isDataRow(r) ? dataRowSetVersion(memRowDataBody(r), v) : memRowSetKvVersion(r, v))
#define memRowCpy(dst, r) memcpy((dst), (r), memRowTLen(r))
#define memRowMaxBytesFromSchema(s) (schemaTLen(s) + TD_MEM_ROW_DATA_HEAD_SIZE)
#define memRowDeleted(r) TKEY_IS_DELETED(memRowTKey(r))
SMemRow tdMemRowDup(SMemRow row);
void tdAppendMemRowToDataCol(SMemRow row, STSchema *pSchema, SDataCols *pCols, bool forceSetNull);
// NOTE: offset here including the header size
static FORCE_INLINE void *tdGetMemRowDataOfCol(void *row, int16_t colId, int8_t colType, uint16_t offset) {
if (isDataRow(row)) {
return tdGetRowDataOfCol(memRowDataBody(row), colType, offset);
} else {
return tdGetKVRowValOfCol(memRowKvBody(row), colId);
}
}
/**
* NOTE:
* 1. Applicable to scan columns one by one
* 2. offset here including the header size
*/
static FORCE_INLINE void *tdGetMemRowDataOfColEx(void *row, int16_t colId, int8_t colType, int32_t offset,
int32_t *kvNIdx) {
if (isDataRow(row)) {
return tdGetRowDataOfCol(memRowDataBody(row), colType, offset);
} else {
return tdGetKVRowValOfColEx(memRowKvBody(row), colId, kvNIdx);
}
}
static FORCE_INLINE int32_t tdAppendMemRowColVal(SMemRow row, const void *value, bool isCopyVarData, int16_t colId,
int8_t type, int32_t offset) {
if (isDataRow(row)) {
tdAppendDataColVal(memRowDataBody(row), value, isCopyVarData, type, offset);
} else {
tdAppendKvColVal(memRowKvBody(row), value, isCopyVarData, colId, type, offset);
}
return 0;
}
// make sure schema->flen appended for SDataRow
static FORCE_INLINE int32_t tdGetColAppendLen(uint8_t rowType, const void *value, int8_t colType) {
int32_t len = 0;
if (IS_VAR_DATA_TYPE(colType)) {
len += varDataTLen(value);
if (rowType == SMEM_ROW_KV) {
len += sizeof(SColIdx);
}
} else {
if (rowType == SMEM_ROW_KV) {
len += TYPE_BYTES[colType];
len += sizeof(SColIdx);
}
}
return len;
}
typedef struct {
int16_t colId;
uint8_t colType;
char * colVal;
} SColInfo;
static FORCE_INLINE void setSColInfo(SColInfo *colInfo, int16_t colId, uint8_t colType, char *colVal) {
colInfo->colId = colId;
colInfo->colType = colType;
colInfo->colVal = colVal;
}
SMemRow mergeTwoMemRows(void *buffer, SMemRow row1, SMemRow row2, STSchema *pSchema1, STSchema *pSchema2);
#endif
#ifdef __cplusplus
}
......
include/common/tmsg.h
浏览文件 @ 97f64a58
......@@ -252,21 +252,29 @@ STSRow* tGetSubmitBlkNext(SSubmitBlkIter* pIter);
int32_t tPrintFixedSchemaSubmitReq(const SSubmitReq* pReq, STSchema* pSchema);
typedef struct {
int32_t index; // index of failed block in submit blocks
int32_t vnode; // vnode index of failed block
int32_t sid; // table index of failed block
int32_t code; // errorcode while write data to vnode, such as not created, dropped, no space, invalid table
} SSubmitRspBlock;
typedef struct {
int32_t code; // 0-success, > 0 error code
int32_t numOfRows; // number of records the client is trying to write
int32_t affectedRows; // number of records actually written
int32_t failedRows; // number of failed records (exclude duplicate records)
int32_t numOfFailedBlocks;
SSubmitRspBlock failedBlocks[];
int8_t hashMeta;
int64_t uid;
union {
char* ename; // used for encode
const char* dname; // used for decode
};
int32_t numOfRows;
int32_t affectedRows;
} SSubmitBlkRsp;
typedef struct {
int32_t numOfRows;
int32_t affectedRows;
int32_t nBlocks;
union {
SArray* pArray;
SSubmitBlkRsp* pBlocks;
};
} SSubmitRsp;
int32_t tEncodeSSubmitRsp(SEncoder* pEncoder, const SSubmitRsp* pRsp);
int32_t tDecodeSSubmitRsp(SDecoder* pDecoder, SSubmitRsp* pRsp);
#define SCHEMA_SMA_ON 0x1
#define SCHEMA_IDX_ON 0x2
typedef struct SSchema {
......
include/common/tmsgcb.h
浏览文件 @ 97f64a58
......@@ -57,6 +57,7 @@ typedef struct {
RegisterBrokenLinkArgFp registerBrokenLinkArgFp;
ReleaseHandleFp releaseHandleFp;
ReportStartup reportStartupFp;
void* clientRpc;
} SMsgCb;
void tmsgSetDefaultMsgCb(const SMsgCb* pMsgCb);
......
include/common/tname.h
浏览文件 @ 97f64a58
......@@ -17,6 +17,7 @@
#define _TD_COMMON_NAME_H_
#include "tdef.h"
#include "tarray.h"
#ifdef __cplusplus
extern "C" {
......@@ -62,6 +63,18 @@ int32_t tNameSetAcctId(SName* dst, int32_t acctId);
bool tNameDBNameEqual(SName* left, SName* right);
typedef struct {
// input
SArray *tags; // element is SSmlKV
const char *sTableName; // super table name
uint8_t sTableNameLen; // the length of super table name
// output
char *childTableName; // must have size of TSDB_TABLE_NAME_LEN;
uint64_t uid; // child table uid, may be useful
} RandTableName;
void buildChildTableName(RandTableName *rName);
#ifdef __cplusplus
}
......
include/common/trow.h
浏览文件 @ 97f64a58
此差异已折叠。
include/libs/executor/executor.h
浏览文件 @ 97f64a58
......@@ -22,6 +22,7 @@ extern "C" {
#include "query.h"
#include "tcommon.h"
#include "tmsgcb.h"
typedef void* qTaskInfo_t;
typedef void* DataSinkHandle;
......@@ -29,11 +30,12 @@ struct SRpcMsg;
struct SSubplan;
typedef struct SReadHandle {
void* reader;
void* meta;
void* config;
void* vnode;
void* mnd;
void* reader;
void* meta;
void* config;
void* vnode;
void* mnd;
SMsgCb* pMsgCb;
} SReadHandle;
#define STREAM_DATA_TYPE_SUBMIT_BLOCK 0x1
......
include/libs/function/function.h
浏览文件 @ 97f64a58
......@@ -336,6 +336,17 @@ int32_t udfcOpen();
*/
int32_t udfcClose();
/**
* start udfd that serves udf function invocation under dnode startDnodeId
* @param startDnodeId
* @return
*/
int32_t udfStartUdfd(int32_t startDnodeId);
/**
* stop udfd
* @return
*/
int32_t udfStopUdfd();
#ifdef __cplusplus
}
#endif
......
include/libs/function/functionMgt.h
浏览文件 @ 97f64a58
......@@ -31,13 +31,13 @@ typedef enum EFunctionType {
FUNCTION_TYPE_ELAPSED,
FUNCTION_TYPE_IRATE,
FUNCTION_TYPE_LAST_ROW,
FUNCTION_TYPE_LEASTSQUARES,
FUNCTION_TYPE_MAX,
FUNCTION_TYPE_MIN,
FUNCTION_TYPE_MODE,
FUNCTION_TYPE_PERCENTILE,
FUNCTION_TYPE_SPREAD,
FUNCTION_TYPE_STDDEV,
FUNCTION_TYPE_LEASTSQUARES,
FUNCTION_TYPE_SUM,
FUNCTION_TYPE_TWA,
FUNCTION_TYPE_HISTOGRAM,
......@@ -154,6 +154,7 @@ bool fmIsWindowClauseFunc(int32_t funcId);
bool fmIsSpecialDataRequiredFunc(int32_t funcId);
bool fmIsDynamicScanOptimizedFunc(int32_t funcId);
bool fmIsMultiResFunc(int32_t funcId);
bool fmIsRepeatScanFunc(int32_t funcId);
bool fmIsUserDefinedFunc(int32_t funcId);
typedef enum EFuncDataRequired {
......@@ -170,7 +171,7 @@ int32_t fmGetScalarFuncExecFuncs(int32_t funcId, SScalarFuncExecFuncs* pFpSet);
int32_t fmGetUdafExecFuncs(int32_t funcId, SFuncExecFuncs* pFpSet);
int32_t fmSetInvertFunc(int32_t funcId, SFuncExecFuncs* pFpSet);
int32_t fmSetNormalFunc(int32_t funcId, SFuncExecFuncs* pFpSet);
bool fmIsInvertible(int32_t funcId);
bool fmIsInvertible(int32_t funcId);
#ifdef __cplusplus
}
......
include/libs/nodes/nodes.h
浏览文件 @ 97f64a58
......@@ -206,6 +206,7 @@ typedef enum ENodeType {
QUERY_NODE_PHYSICAL_PLAN_EXCHANGE,
QUERY_NODE_PHYSICAL_PLAN_SORT,
QUERY_NODE_PHYSICAL_PLAN_INTERVAL,
QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL,
QUERY_NODE_PHYSICAL_PLAN_FILL,
QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW,
QUERY_NODE_PHYSICAL_PLAN_STATE_WINDOW,
......
include/libs/nodes/plannodes.h
浏览文件 @ 97f64a58
......@@ -274,6 +274,8 @@ typedef struct SIntervalPhysiNode {
int8_t slidingUnit;
} SIntervalPhysiNode;
typedef SIntervalPhysiNode SStreamIntervalPhysiNode;
typedef struct SFillPhysiNode {
SPhysiNode node;
EFillMode mode;
......
include/libs/nodes/querynodes.h
浏览文件 @ 97f64a58
......@@ -233,6 +233,7 @@ typedef struct SSelectStmt {
uint8_t precision;
bool isEmptyResult;
bool hasAggFuncs;
bool hasRepeatScanFuncs;
bool isTimeOrderQuery;
} SSelectStmt;
......
include/libs/qcom/query.h
浏览文件 @ 97f64a58
......@@ -24,6 +24,7 @@ extern "C" {
#include "thash.h"
#include "tlog.h"
#include "tmsg.h"
#include "tmsgcb.h"
typedef enum {
JOB_TASK_STATUS_NULL = 0,
......@@ -149,7 +150,8 @@ int32_t cleanupTaskQueue();
*/
int32_t taosAsyncExec(__async_exec_fn_t execFn, void* execParam, int32_t* code);
int32_t asyncSendMsgToServerExt(void* pTransporter, SEpSet* epSet, int64_t* pTransporterId, const SMsgSendInfo* pInfo, bool persistHandle, void *ctx);
int32_t asyncSendMsgToServerExt(void* pTransporter, SEpSet* epSet, int64_t* pTransporterId, const SMsgSendInfo* pInfo,
bool persistHandle, void* ctx);
/**
* Asynchronously send message to server, after the response received, the callback will be incured.
......
source/client/src/clientSml.c
浏览文件 @ 97f64a58
......@@ -15,6 +15,7 @@
#include "tcommon.h"
#include "catalog.h"
#include "clientInt.h"
#include "tname.h"
//=================================================================================================
#define SPACE ' '
......@@ -97,6 +98,21 @@ typedef struct {
char *buf;
} SSmlMsgBuf;
typedef struct {
int32_t code;
int32_t lineNum;
int32_t numOfSTables;
int32_t numOfCTables;
int32_t numOfCreateSTables;
int64_t parseTime;
int64_t schemaTime;
int64_t insertBindTime;
int64_t insertRpcTime;
int64_t endTime;
} SSmlCostInfo;
typedef struct {
uint64_t id;
......@@ -114,6 +130,7 @@ typedef struct {
SRequestObj *pRequest;
SQuery *pQuery;
SSmlCostInfo cost;
int32_t affectedRows;
SSmlMsgBuf msgBuf;
SHashObj *dumplicateKey; // for dumplicate key
......@@ -147,45 +164,6 @@ static int32_t smlBuildInvalidDataMsg(SSmlMsgBuf* pBuf, const char *msg1, const
return TSDB_CODE_SML_INVALID_DATA;
}
static int smlCompareKv(const void* p1, const void* p2) {
SSmlKv* kv1 = *(SSmlKv**)p1;
SSmlKv* kv2 = *(SSmlKv**)p2;
int32_t kvLen1 = kv1->keyLen;
int32_t kvLen2 = kv2->keyLen;
int32_t res = strncasecmp(kv1->key, kv2->key, TMIN(kvLen1, kvLen2));
if (res != 0) {
return res;
} else {
return kvLen1-kvLen2;
}
}
static void smlBuildChildTableName(SSmlTableInfo *tags) {
int32_t size = taosArrayGetSize(tags->tags);
ASSERT(size > 0);
taosArraySort(tags->tags, smlCompareKv);
SStringBuilder sb = {0};
taosStringBuilderAppendStringLen(&sb, tags->sTableName, tags->sTableNameLen);
for (int j = 0; j < size; ++j) {
SSmlKv *tagKv = taosArrayGetP(tags->tags, j);
taosStringBuilderAppendStringLen(&sb, tagKv->key, tagKv->keyLen);
taosStringBuilderAppendStringLen(&sb, tagKv->value, tagKv->valueLen);
}
size_t len = 0;
char* keyJoined = taosStringBuilderGetResult(&sb, &len);
T_MD5_CTX context;
tMD5Init(&context);
tMD5Update(&context, (uint8_t *)keyJoined, (uint32_t)len);
tMD5Final(&context);
uint64_t digest1 = *(uint64_t*)(context.digest);
//uint64_t digest2 = *(uint64_t*)(context.digest + 8);
//snprintf(tags->childTableName, TSDB_TABLE_NAME_LEN, "t_%016"PRIx64"%016"PRIx64, digest1, digest2);
snprintf(tags->childTableName, TSDB_TABLE_NAME_LEN, "t_%016"PRIx64, digest1);
taosStringBuilderDestroy(&sb);
tags->uid = digest1;
}
static int32_t smlGenerateSchemaAction(SSchema* pointColField, SHashObj* dbAttrHash, SArray* dbAttrArray, bool isTag, char sTableName[],
SSchemaAction* action, bool* actionNeeded, SSmlHandle* info) {
// char fieldName[TSDB_COL_NAME_LEN] = {0};
......@@ -444,6 +422,7 @@ static int32_t smlModifyDBSchemas(SSmlHandle* info) {
uError("SML:0x%"PRIx64" catalogGetSTableMeta failed. super table name %s", info->id, schemaAction.createSTable.sTableName);
return code;
}
info->cost.numOfCreateSTables++;
}else if (code == TSDB_CODE_SUCCESS) {
} else {
uError("SML:0x%"PRIx64" load table meta error: %s", info->id, tstrerror(code));
......@@ -926,20 +905,6 @@ static bool smlParseValue(SSmlKv *pVal, SSmlMsgBuf *msg) {
return false;
}
static bool checkDuplicateKey(char *key, SHashObj *pHash, SSmlHandle* info) {
char *val = NULL;
val = taosHashGet(pHash, key, strlen(key));
if (val) {
uError("SML:0x%"PRIx64" Duplicate key detected:%s", info->id, key);
return true;
}
uint8_t dummy_val = 0;
taosHashPut(pHash, key, strlen(key), &dummy_val, sizeof(uint8_t));
return false;
}
static int32_t smlParseString(const char* sql, SSmlLineInfo *elements, SSmlMsgBuf *msg){
if(!sql) return TSDB_CODE_SML_INVALID_DATA;
while (*sql != '\0') { // jump the space at the begining
......@@ -1546,8 +1511,10 @@ static int32_t smlParseLine(SSmlHandle* info, const char* sql) {
tinfo->sTableName = elements.measure;
tinfo->sTableNameLen = elements.measureLen;
smlBuildChildTableName(tinfo);
uDebug("SML:0x%"PRIx64" child table name: %s", info->id, tinfo->childTableName);
RandTableName rName = {.tags=tinfo->tags, .sTableName=tinfo->sTableName, .sTableNameLen=tinfo->sTableNameLen,
.childTableName=tinfo->childTableName};
buildChildTableName(&rName);
tinfo->uid = rName.uid;
SSmlSTableMeta** tableMeta = taosHashGet(info->superTables, elements.measure, elements.measureLen);
if(tableMeta){ // update meta
......@@ -1604,7 +1571,7 @@ static void smlDestroyInfo(SSmlHandle* info){
static SSmlHandle* smlBuildSmlInfo(TAOS* taos, SRequestObj* request, SMLProtocolType protocol, int8_t precision, bool dataFormat){
int32_t code = TSDB_CODE_SUCCESS;
SSmlHandle* info = taosMemoryMalloc(sizeof(SSmlHandle));
SSmlHandle* info = taosMemoryCalloc(1, sizeof(SSmlHandle));
if (NULL == info) {
return NULL;
}
......@@ -1699,12 +1666,23 @@ static int32_t smlInsertData(SSmlHandle* info) {
}
smlBuildOutput(info->exec, info->pVgHash);
info->cost.insertRpcTime = taosGetTimestampUs();
launchQueryImpl(info->pRequest, info->pQuery, TSDB_CODE_SUCCESS, true);
info->affectedRows = taos_affected_rows(info->pRequest);
return info->pRequest->code;
}
static void smlPrintStatisticInfo(SSmlHandle *info){
uError("SML:0x%"PRIx64" smlInsertLines result, code:%d,lineNum:%d,stable num:%d,ctable num:%d,create stable num:%d \
parse cost:%"PRId64",schema cost:%"PRId64",bind cost:%"PRId64",rpc cost:%"PRId64",total cost:%"PRId64"", info->id, info->cost.code,
info->cost.lineNum, info->cost.numOfSTables, info->cost.numOfCTables, info->cost.numOfCreateSTables,
info->cost.schemaTime-info->cost.parseTime, info->cost.insertBindTime-info->cost.schemaTime,
info->cost.insertRpcTime-info->cost.insertBindTime, info->cost.endTime-info->cost.insertRpcTime,
info->cost.endTime-info->cost.parseTime);
}
static int smlInsertLines(SSmlHandle *info, char* lines[], int numLines) {
int32_t code = TSDB_CODE_SUCCESS;
......@@ -1714,6 +1692,7 @@ static int smlInsertLines(SSmlHandle *info, char* lines[], int numLines) {
goto cleanup;
}
info->cost.parseTime = taosGetTimestampUs();
for (int32_t i = 0; i < numLines; ++i) {
code = smlParseLine(info, lines[i]);
if (code != TSDB_CODE_SUCCESS) {
......@@ -1721,24 +1700,29 @@ static int smlInsertLines(SSmlHandle *info, char* lines[], int numLines) {
goto cleanup;
}
}
uDebug("SML:0x%"PRIx64" smlInsertLines parse success. tables %d", info->id, taosHashGetSize(info->childTables));
uDebug("SML:0x%"PRIx64" smlInsertLines parse success. super tables %d", info->id, taosHashGetSize(info->superTables));
info->cost.lineNum = numLines;
info->cost.numOfSTables = taosHashGetSize(info->superTables);
info->cost.numOfCTables = taosHashGetSize(info->childTables);
info->cost.schemaTime = taosGetTimestampUs();
code = smlModifyDBSchemas(info);
if (code != 0) {
uError("SML:0x%"PRIx64" smlModifyDBSchemas error : %s", info->id, tstrerror(code));
goto cleanup;
}
info->cost.insertBindTime = taosGetTimestampUs();
code = smlInsertData(info);
if (code != 0) {
uError("SML:0x%"PRIx64" smlInsertData error : %s", info->id, tstrerror(code));
goto cleanup;
}
uDebug("SML:0x%"PRIx64" smlInsertLines finish inserting %d lines.", info->id, numLines);
info->cost.endTime = taosGetTimestampUs();
cleanup:
info->cost.code = code;
smlPrintStatisticInfo(info);
return code;
}
......@@ -1790,7 +1774,6 @@ TAOS_RES* taos_schemaless_insert(TAOS* taos, char* lines[], int numLines, int pr
}
smlDestroyInfo(info);
end:
return (TAOS_RES*)request;
}
source/common/src/tdata.c 0 → 100644
浏览文件 @ 97f64a58
/*
* 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/>.
*/
\ No newline at end of file
source/common/src/tdataformat.c
浏览文件 @ 97f64a58
......@@ -129,50 +129,6 @@ void *tdDecodeSchema(void *buf, STSchema **pRSchema) {
return buf;
}
#if 0
int32_t tEncodeSTColumn(SCoder *pEncoder, const STColumn *pCol) {
if (tEncodeI16(pEncoder, pCol->colId) < 0) return -1;
if (tEncodeI8(pEncoder, pCol->type) < 0) return -1;
if (tEncodeI8(pEncoder, pCol->sma) < 0) return -1;
if (tEncodeI32(pEncoder, pCol->bytes) < 0) return -1;
if (tEncodeI32(pEncoder, pCol->offset) < 0) return -1;
return pEncoder->pos;
}
int32_t tDecodeSTColumn(SCoder *pDecoder, STColumn *pCol) {
if (tDecodeI16(pDecoder, &pCol->colId) < 0) return -1;
if (tDecodeI8(pDecoder, &pCol->type) < 0) return -1;
if (tDecodeI8(pDecoder, &pCol->sma) < 0) return -1;
if (tDecodeI32(pDecoder, &pCol->bytes) < 0) return -1;
if (tDecodeI32(pDecoder, &pCol->offset) < 0) return -1;
return 0;
}
int32_t tEncodeSchema(SCoder *pEncoder, const STSchema *pSchema) {
if (tEncodeI32(pEncoder, pSchema->numOfCols) < 0) return -1;
if (tEncodeI16(pEncoder, pSchema->version) < 0) return -1;
if (tEncodeU16(pEncoder, pSchema->flen) < 0) return -1;
if (tEncodeI32(pEncoder, pSchema->vlen) < 0) return -1;
if (tEncodeI32(pEncoder, pSchema->tlen) < 0) return -1;
for (int32_t i = 0; i < schemaNCols(pSchema); i++) {
const STColumn *pCol = schemaColAt(pSchema, i);
if (tEncodeSTColumn(pEncoder, pCol) < 0) return -1;
}
return 0;
}
int32_t tDecodeSchema(SCoder *pDecoder, STSchema *pSchema) {
if (tDecodeI32(pDecoder, &pSchema->numOfCols) < 0) return -1;
if (tDecodeI16(pDecoder, &pSchema->version) < 0) return -1;
if (tDecodeU16(pDecoder, &pSchema->flen) < 0) return -1;
if (tDecodeI32(pDecoder, &pSchema->vlen) < 0) return -1;
if (tDecodeI32(pDecoder, &pSchema->tlen) < 0) return -1;
return 0;
}
#endif
int tdInitTSchemaBuilder(STSchemaBuilder *pBuilder, schema_ver_t version) {
if (pBuilder == NULL) return -1;
......@@ -260,49 +216,6 @@ STSchema *tdGetSchemaFromBuilder(STSchemaBuilder *pBuilder) {
return pSchema;
}
#if 0
/**
* Initialize a data row
*/
void tdInitDataRow(SDataRow row, STSchema *pSchema) {
dataRowSetLen(row, TD_DATA_ROW_HEAD_SIZE + schemaFLen(pSchema));
dataRowSetVersion(row, schemaVersion(pSchema));
}
SDataRow tdNewDataRowFromSchema(STSchema *pSchema) {
int32_t size = dataRowMaxBytesFromSchema(pSchema);
SDataRow row = taosMemoryMalloc(size);
if (row == NULL) return NULL;
tdInitDataRow(row, pSchema);
return row;
}
/**
* Free the SDataRow object
*/
void tdFreeDataRow(SDataRow row) {
if (row) taosMemoryFree(row);
}
SDataRow tdDataRowDup(SDataRow row) {
SDataRow trow = taosMemoryMalloc(dataRowLen(row));
if (trow == NULL) return NULL;
dataRowCpy(trow, row);
return trow;
}
SMemRow tdMemRowDup(SMemRow row) {
SMemRow trow = taosMemoryMalloc(memRowTLen(row));
if (trow == NULL) return NULL;
memRowCpy(trow, row);
return trow;
}
#endif
void dataColInit(SDataCol *pDataCol, STColumn *pCol, int maxPoints) {
pDataCol->type = colType(pCol);
pDataCol->colId = colColId(pCol);
......@@ -312,39 +225,6 @@ void dataColInit(SDataCol *pDataCol, STColumn *pCol, int maxPoints) {
pDataCol->len = 0;
}
#if 0
// value from timestamp should be TKEY here instead of TSKEY
int dataColAppendVal(SDataCol *pCol, const void *value, int numOfRows, int maxPoints) {
ASSERT(pCol != NULL && value != NULL);
if (isAllRowsNull(pCol)) {
if (isNull(value, pCol->type)) {
// all null value yet, just return
return 0;
}
if (tdAllocMemForCol(pCol, maxPoints) < 0) return -1;
if (numOfRows > 0) {
// Find the first not null value, fill all previouse values as NULL
dataColSetNEleNull(pCol, numOfRows);
}
}
if (IS_VAR_DATA_TYPE(pCol->type)) {
// set offset
pCol->dataOff[numOfRows] = pCol->len;
// Copy data
memcpy(POINTER_SHIFT(pCol->pData, pCol->len), value, varDataTLen(value));
// Update the length
pCol->len += varDataTLen(value);
} else {
ASSERT(pCol->len == TYPE_BYTES[pCol->type] * numOfRows);
memcpy(POINTER_SHIFT(pCol->pData, pCol->len), value, pCol->bytes);
pCol->len += pCol->bytes;
}
return 0;
}
#endif
static FORCE_INLINE const void *tdGetColDataOfRowUnsafe(SDataCol *pCol, int row) {
if (IS_VAR_DATA_TYPE(pCol->type)) {
return POINTER_SHIFT(pCol->pData, pCol->dataOff[row]);
......@@ -361,31 +241,6 @@ bool isNEleNull(SDataCol *pCol, int nEle) {
return true;
}
#if 0
static FORCE_INLINE void dataColSetNullAt(SDataCol *pCol, int index) {
if (IS_VAR_DATA_TYPE(pCol->type)) {
pCol->dataOff[index] = pCol->len;
char *ptr = POINTER_SHIFT(pCol->pData, pCol->len);
setVardataNull(ptr, pCol->type);
pCol->len += varDataTLen(ptr);
} else {
setNull(POINTER_SHIFT(pCol->pData, TYPE_BYTES[pCol->type] * index), pCol->type, pCol->bytes);
pCol->len += TYPE_BYTES[pCol->type];
}
}
static void dataColSetNEleNull(SDataCol *pCol, int nEle, int8_t bitmapMode) {
if (IS_VAR_DATA_TYPE(pCol->type)) {
pCol->len = 0;
for (int i = 0; i < nEle; ++i) {
dataColSetNullAt(pCol, i);
}
} else {
setNullN(pCol->pData, pCol->type, pCol->bytes, nEle);
pCol->len = TYPE_BYTES[pCol->type] * nEle;
}
}
#endif
void *dataColSetOffset(SDataCol *pCol, int nEle) {
ASSERT(((pCol->type == TSDB_DATA_TYPE_BINARY) || (pCol->type == TSDB_DATA_TYPE_NCHAR)));
......@@ -483,42 +338,6 @@ SDataCols *tdFreeDataCols(SDataCols *pCols) {
return NULL;
}
#if 0
SDataCols *tdDupDataCols(SDataCols *pDataCols, bool keepData) {
SDataCols *pRet = tdNewDataCols(pDataCols->maxCols, pDataCols->maxPoints);
if (pRet == NULL) return NULL;
pRet->numOfCols = pDataCols->numOfCols;
pRet->sversion = pDataCols->sversion;
if (keepData) pRet->numOfRows = pDataCols->numOfRows;
for (int i = 0; i < pDataCols->numOfCols; i++) {
pRet->cols[i].type = pDataCols->cols[i].type;
pRet->cols[i].bitmap = pDataCols->cols[i].bitmap;
pRet->cols[i].colId = pDataCols->cols[i].colId;
pRet->cols[i].bytes = pDataCols->cols[i].bytes;
pRet->cols[i].offset = pDataCols->cols[i].offset;
if (keepData) {
if (pDataCols->cols[i].len > 0) {
if (tdAllocMemForCol(&pRet->cols[i], pRet->maxPoints) < 0) {
tdFreeDataCols(pRet);
return NULL;
}
pRet->cols[i].len = pDataCols->cols[i].len;
memcpy(pRet->cols[i].pData, pDataCols->cols[i].pData, pDataCols->cols[i].len);
if (IS_VAR_DATA_TYPE(pRet->cols[i].type)) {
int dataOffSize = sizeof(VarDataOffsetT) * pDataCols->maxPoints;
memcpy(pRet->cols[i].dataOff, pDataCols->cols[i].dataOff, dataOffSize);
}
}
}
}
return pRet;
}
#endif
void tdResetDataCols(SDataCols *pCols) {
if (pCols != NULL) {
pCols->numOfRows = 0;
......@@ -528,180 +347,6 @@ void tdResetDataCols(SDataCols *pCols) {
}
}
}
#if 0
static void tdAppendDataRowToDataCol(SDataRow row, STSchema *pSchema, SDataCols *pCols, bool forceSetNull) {
ASSERT(pCols->numOfRows == 0 || dataColsKeyLast(pCols) < dataRowKey(row));
int rcol = 0;
int dcol = 0;
while (dcol < pCols->numOfCols) {
bool setCol = 0;
SDataCol *pDataCol = &(pCols->cols[dcol]);
if (rcol >= schemaNCols(pSchema)) {
dataColAppendVal(pDataCol, getNullValue(pDataCol->type), pCols->numOfRows, pCols->maxPoints);
dcol++;
continue;
}
STColumn *pRowCol = schemaColAt(pSchema, rcol);
if (pRowCol->colId == pDataCol->colId) {
void *value = tdGetRowDataOfCol(row, pRowCol->type, pRowCol->offset + TD_DATA_ROW_HEAD_SIZE);
if(!isNull(value, pDataCol->type)) setCol = 1;
dataColAppendVal(pDataCol, value, pCols->numOfRows, pCols->maxPoints);
dcol++;
rcol++;
} else if (pRowCol->colId < pDataCol->colId) {
rcol++;
} else {
if(forceSetNull || setCol) {
dataColAppendVal(pDataCol, getNullValue(pDataCol->type), pCols->numOfRows, pCols->maxPoints);
}
dcol++;
}
}
pCols->numOfRows++;
}
static void tdAppendKVRowToDataCol(SKVRow row, STSchema *pSchema, SDataCols *pCols, bool forceSetNull) {
ASSERT(pCols->numOfRows == 0 || dataColsKeyLast(pCols) < kvRowKey(row));
int rcol = 0;
int dcol = 0;
int nRowCols = kvRowNCols(row);
while (dcol < pCols->numOfCols) {
bool setCol = 0;
SDataCol *pDataCol = &(pCols->cols[dcol]);
if (rcol >= nRowCols || rcol >= schemaNCols(pSchema)) {
dataColAppendVal(pDataCol, getNullValue(pDataCol->type), pCols->numOfRows, pCols->maxPoints);
++dcol;
continue;
}
SColIdx *colIdx = kvRowColIdxAt(row, rcol);
if (colIdx->colId == pDataCol->colId) {
void *value = tdGetKvRowDataOfCol(row, colIdx->offset);
if(!isNull(value, pDataCol->type)) setCol = 1;
dataColAppendVal(pDataCol, value, pCols->numOfRows, pCols->maxPoints);
++dcol;
++rcol;
} else if (colIdx->colId < pDataCol->colId) {
++rcol;
} else {
if(forceSetNull || setCol) {
dataColAppendVal(pDataCol, getNullValue(pDataCol->type), pCols->numOfRows, pCols->maxPoints);
}
++dcol;
}
}
pCols->numOfRows++;
}
void tdAppendMemRowToDataCol(SMemRow row, STSchema *pSchema, SDataCols *pCols, bool forceSetNull) {
if (isDataRow(row)) {
tdAppendDataRowToDataCol(memRowDataBody(row), pSchema, pCols, forceSetNull);
} else if (isKvRow(row)) {
tdAppendKVRowToDataCol(memRowKvBody(row), pSchema, pCols, forceSetNull);
} else {
ASSERT(0);
}
}
int tdMergeDataCols(SDataCols *target, SDataCols *source, int rowsToMerge, int *pOffset, bool forceSetNull) {
ASSERT(rowsToMerge > 0 && rowsToMerge <= source->numOfRows);
ASSERT(target->numOfCols == source->numOfCols);
int offset = 0;
if (pOffset == NULL) {
pOffset = &offset;
}
SDataCols *pTarget = NULL;
if ((target->numOfRows == 0) || (dataColsKeyLast(target) < dataColsKeyAtRow(source, *pOffset))) { // No overlap
ASSERT(target->numOfRows + rowsToMerge <= target->maxPoints);
for (int i = 0; i < rowsToMerge; i++) {
for (int j = 0; j < source->numOfCols; j++) {
if (source->cols[j].len > 0 || target->cols[j].len > 0) {
dataColAppendVal(target->cols + j, tdGetColDataOfRow(source->cols + j, i + (*pOffset)), target->numOfRows,
target->maxPoints);
}
}
target->numOfRows++;
}
(*pOffset) += rowsToMerge;
} else {
pTarget = tdDupDataCols(target, true);
if (pTarget == NULL) goto _err;
int iter1 = 0;
tdMergeTwoDataCols(target, pTarget, &iter1, pTarget->numOfRows, source, pOffset, source->numOfRows,
pTarget->numOfRows + rowsToMerge, forceSetNull);
}
tdFreeDataCols(pTarget);
return 0;
_err:
tdFreeDataCols(pTarget);
return -1;
}
// src2 data has more priority than src1
static void tdMergeTwoDataCols(SDataCols *target, SDataCols *src1, int *iter1, int limit1, SDataCols *src2, int *iter2,
int limit2, int tRows, bool forceSetNull) {
tdResetDataCols(target);
ASSERT(limit1 <= src1->numOfRows && limit2 <= src2->numOfRows);
while (target->numOfRows < tRows) {
if (*iter1 >= limit1 && *iter2 >= limit2) break;
TSKEY key1 = (*iter1 >= limit1) ? INT64_MAX : dataColsKeyAt(src1, *iter1);
TKEY tkey1 = (*iter1 >= limit1) ? TKEY_NULL : dataColsTKeyAt(src1, *iter1);
TSKEY key2 = (*iter2 >= limit2) ? INT64_MAX : dataColsKeyAt(src2, *iter2);
TKEY tkey2 = (*iter2 >= limit2) ? TKEY_NULL : dataColsTKeyAt(src2, *iter2);
ASSERT(tkey1 == TKEY_NULL || (!TKEY_IS_DELETED(tkey1)));
if (key1 < key2) {
for (int i = 0; i < src1->numOfCols; i++) {
ASSERT(target->cols[i].type == src1->cols[i].type);
if (src1->cols[i].len > 0 || target->cols[i].len > 0) {
dataColAppendVal(&(target->cols[i]), tdGetColDataOfRow(src1->cols + i, *iter1), target->numOfRows,
target->maxPoints);
}
}
target->numOfRows++;
(*iter1)++;
} else if (key1 >= key2) {
if ((key1 > key2) || (key1 == key2 && !TKEY_IS_DELETED(tkey2))) {
for (int i = 0; i < src2->numOfCols; i++) {
ASSERT(target->cols[i].type == src2->cols[i].type);
if (src2->cols[i].len > 0 && !isNull(src2->cols[i].pData, src2->cols[i].type)) {
dataColAppendVal(&(target->cols[i]), tdGetColDataOfRow(src2->cols + i, *iter2), target->numOfRows,
target->maxPoints);
} else if(!forceSetNull && key1 == key2 && src1->cols[i].len > 0) {
dataColAppendVal(&(target->cols[i]), tdGetColDataOfRow(src1->cols + i, *iter1), target->numOfRows,
target->maxPoints);
} else if(target->cols[i].len > 0) {
dataColSetNullAt(&target->cols[i], target->numOfRows);
}
}
target->numOfRows++;
}
(*iter2)++;
if (key1 == key2) (*iter1)++;
}
ASSERT(target->numOfRows <= target->maxPoints);
}
}
#endif
SKVRow tdKVRowDup(SKVRow row) {
SKVRow trow = taosMemoryMalloc(kvRowLen(row));
......@@ -859,98 +504,3 @@ SKVRow tdGetKVRowFromBuilder(SKVRowBuilder *pBuilder) {
return row;
}
#if 0
SMemRow mergeTwoMemRows(void *buffer, SMemRow row1, SMemRow row2, STSchema *pSchema1, STSchema *pSchema2) {
#if 0
ASSERT(memRowKey(row1) == memRowKey(row2));
ASSERT(schemaVersion(pSchema1) == memRowVersion(row1));
ASSERT(schemaVersion(pSchema2) == memRowVersion(row2));
ASSERT(schemaVersion(pSchema1) >= schemaVersion(pSchema2));
#endif
SArray *stashRow = taosArrayInit(pSchema1->numOfCols, sizeof(SColInfo));
if (stashRow == NULL) {
return NULL;
}
SMemRow pRow = buffer;
SDataRow dataRow = memRowDataBody(pRow);
memRowSetType(pRow, SMEM_ROW_DATA);
dataRowSetVersion(dataRow, schemaVersion(pSchema1)); // use latest schema version
dataRowSetLen(dataRow, (TDRowLenT)(TD_DATA_ROW_HEAD_SIZE + pSchema1->flen));
TDRowLenT dataLen = 0, kvLen = TD_MEM_ROW_KV_HEAD_SIZE;
int32_t i = 0; // row1
int32_t j = 0; // row2
int32_t nCols1 = schemaNCols(pSchema1);
int32_t nCols2 = schemaNCols(pSchema2);
SColInfo colInfo = {0};
int32_t kvIdx1 = 0, kvIdx2 = 0;
while (i < nCols1) {
STColumn *pCol = schemaColAt(pSchema1, i);
void * val1 = tdGetMemRowDataOfColEx(row1, pCol->colId, pCol->type, TD_DATA_ROW_HEAD_SIZE + pCol->offset, &kvIdx1);
// if val1 != NULL, use val1;
if (val1 != NULL && !isNull(val1, pCol->type)) {
tdAppendColVal(dataRow, val1, pCol->type, pCol->offset);
kvLen += tdGetColAppendLen(SMEM_ROW_KV, val1, pCol->type);
setSColInfo(&colInfo, pCol->colId, pCol->type, val1);
taosArrayPush(stashRow, &colInfo);
++i; // next col
continue;
}
void *val2 = NULL;
while (j < nCols2) {
STColumn *tCol = schemaColAt(pSchema2, j);
if (tCol->colId < pCol->colId) {
++j;
continue;
}
if (tCol->colId == pCol->colId) {
val2 = tdGetMemRowDataOfColEx(row2, tCol->colId, tCol->type, TD_DATA_ROW_HEAD_SIZE + tCol->offset, &kvIdx2);
} else if (tCol->colId > pCol->colId) {
// set NULL
}
break;
} // end of while(j<nCols2)
if (val2 == NULL) {
val2 = (void *)getNullValue(pCol->type);
}
tdAppendColVal(dataRow, val2, pCol->type, pCol->offset);
if (!isNull(val2, pCol->type)) {
kvLen += tdGetColAppendLen(SMEM_ROW_KV, val2, pCol->type);
setSColInfo(&colInfo, pCol->colId, pCol->type, val2);
taosArrayPush(stashRow, &colInfo);
}
++i; // next col
}
dataLen = memRowTLen(pRow);
if (kvLen < dataLen) {
// scan stashRow and generate SKVRow
memset(buffer, 0, sizeof(dataLen));
SMemRow tRow = buffer;
memRowSetType(tRow, SMEM_ROW_KV);
SKVRow kvRow = (SKVRow)memRowKvBody(tRow);
int16_t nKvNCols = (int16_t) taosArrayGetSize(stashRow);
kvRowSetLen(kvRow, (TDRowLenT)(TD_KV_ROW_HEAD_SIZE + sizeof(SColIdx) * nKvNCols));
kvRowSetNCols(kvRow, nKvNCols);
memRowSetKvVersion(tRow, pSchema1->version);
int32_t toffset = 0;
int16_t k;
for (k = 0; k < nKvNCols; ++k) {
SColInfo *pColInfo = taosArrayGet(stashRow, k);
tdAppendKvColVal(kvRow, pColInfo->colVal, true, pColInfo->colId, pColInfo->colType, toffset);
toffset += sizeof(SColIdx);
}
ASSERT(kvLen == memRowTLen(tRow));
}
taosArrayDestroy(stashRow);
return buffer;
}
#endif
source/common/src/tmsg.c
浏览文件 @ 97f64a58
......@@ -4015,3 +4015,65 @@ int32_t tDecodeSVSubmitReq(SDecoder *pCoder, SVSubmitReq *pReq) {
tEndDecode(pCoder);
return 0;
}
static int32_t tEncodeSSubmitBlkRsp(SEncoder *pEncoder, const SSubmitBlkRsp *pBlock) {
if (tStartEncode(pEncoder) < 0) return -1;
if (tEncodeI8(pEncoder, pBlock->hashMeta) < 0) return -1;
if (pBlock->hashMeta) {
if (tEncodeI64(pEncoder, pBlock->uid) < 0) return -1;
if (tEncodeCStr(pEncoder, pBlock->ename) < 0) return -1;
}
if (tEncodeI32v(pEncoder, pBlock->numOfRows) < 0) return -1;
if (tEncodeI32v(pEncoder, pBlock->affectedRows) < 0) return -1;
tEndEncode(pEncoder);
return 0;
}
static int32_t tDecodeSSubmitBlkRsp(SDecoder *pDecoder, SSubmitBlkRsp *pBlock) {
if (tStartDecode(pDecoder) < 0) return -1;
if (tDecodeI8(pDecoder, &pBlock->hashMeta) < 0) return -1;
if (pBlock->hashMeta) {
if (tDecodeI64(pDecoder, &pBlock->uid) < 0) return -1;
if (tDecodeCStr(pDecoder, &pBlock->dname) < 0) return -1;
}
if (tDecodeI32v(pDecoder, &pBlock->numOfRows) < 0) return -1;
if (tDecodeI32v(pDecoder, &pBlock->affectedRows) < 0) return -1;
tEndDecode(pDecoder);
return 0;
}
int32_t tEncodeSSubmitRsp(SEncoder *pEncoder, const SSubmitRsp *pRsp) {
int32_t nBlocks = taosArrayGetSize(pRsp->pArray);
if (tStartEncode(pEncoder) < 0) return -1;
if (tEncodeI32v(pEncoder, pRsp->numOfRows) < 0) return -1;
if (tEncodeI32v(pEncoder, pRsp->affectedRows) < 0) return -1;
if (tEncodeI32v(pEncoder, nBlocks) < 0) return -1;
for (int32_t iBlock = 0; iBlock < nBlocks; iBlock++) {
if (tEncodeSSubmitBlkRsp(pEncoder, (SSubmitBlkRsp *)taosArrayGet(pRsp->pArray, iBlock)) < 0) return -1;
}
tEndEncode(pEncoder);
return 0;
}
int32_t tDecodeSSubmitRsp(SDecoder *pDecoder, SSubmitRsp *pRsp) {
if (tStartDecode(pDecoder) < 0) return -1;
if (tDecodeI32v(pDecoder, &pRsp->numOfRows) < 0) return -1;
if (tDecodeI32v(pDecoder, &pRsp->affectedRows) < 0) return -1;
if (tDecodeI32v(pDecoder, &pRsp->nBlocks) < 0) return -1;
pRsp->pBlocks = tDecoderMalloc(pDecoder, sizeof(*pRsp->pBlocks) * pRsp->nBlocks);
if (pRsp->pBlocks == NULL) return -1;
for (int32_t iBlock = 0; iBlock < pRsp->nBlocks; iBlock++) {
if (tDecodeSSubmitBlkRsp(pDecoder, pRsp->pBlocks + iBlock) < 0) return -1;
}
tEndDecode(pDecoder);
return 0;
}
source/common/src/tname.c
浏览文件 @ 97f64a58
......@@ -15,6 +15,8 @@
#define _DEFAULT_SOURCE
#include "tname.h"
#include "tcommon.h"
#include "tstrbuild.h"
#define VALID_NAME_TYPE(x) ((x) == TSDB_DB_NAME_T || (x) == TSDB_TABLE_NAME_T)
......@@ -294,4 +296,43 @@ int32_t tNameFromString(SName* dst, const char* str, uint32_t type) {
return 0;
}
static int compareKv(const void* p1, const void* p2) {
SSmlKv* kv1 = *(SSmlKv**)p1;
SSmlKv* kv2 = *(SSmlKv**)p2;
int32_t kvLen1 = kv1->keyLen;
int32_t kvLen2 = kv2->keyLen;
int32_t res = strncasecmp(kv1->key, kv2->key, TMIN(kvLen1, kvLen2));
if (res != 0) {
return res;
} else {
return kvLen1-kvLen2;
}
}
/*
* use stable name and tags to grearate child table name
*/
void buildChildTableName(RandTableName *rName) {
int32_t size = taosArrayGetSize(rName->tags);
ASSERT(size > 0);
taosArraySort(rName->tags, compareKv);
SStringBuilder sb = {0};
taosStringBuilderAppendStringLen(&sb, rName->sTableName, rName->sTableNameLen);
for (int j = 0; j < size; ++j) {
SSmlKv *tagKv = taosArrayGetP(rName->tags, j);
taosStringBuilderAppendStringLen(&sb, tagKv->key, tagKv->keyLen);
taosStringBuilderAppendStringLen(&sb, tagKv->value, tagKv->valueLen);
}
size_t len = 0;
char* keyJoined = taosStringBuilderGetResult(&sb, &len);
T_MD5_CTX context;
tMD5Init(&context);
tMD5Update(&context, (uint8_t *)keyJoined, (uint32_t)len);
tMD5Final(&context);
uint64_t digest1 = *(uint64_t*)(context.digest);
uint64_t digest2 = *(uint64_t*)(context.digest + 8);
snprintf(rName->childTableName, TSDB_TABLE_NAME_LEN, "t_%016"PRIx64"%016"PRIx64, digest1, digest2);
taosStringBuilderDestroy(&sb);
rName->uid = digest1;
}
source/common/src/trow.c
浏览文件 @ 97f64a58
此差异已折叠。
source/dnode/mgmt/implement/inc/dmImp.h
浏览文件 @ 97f64a58
......@@ -26,8 +26,10 @@ int32_t dmOpenNode(SMgmtWrapper *pWrapper);
void dmCloseNode(SMgmtWrapper *pWrapper);
// dmTransport.c
int32_t dmInitTrans(SDnode *pDnode);
void dmCleanupTrans(SDnode *pDnode);
int32_t dmInitServer(SDnode *pDnode);
void dmCleanupServer(SDnode *pDnode);
int32_t dmInitClient(SDnode *pDnode);
void dmCleanupClient(SDnode *pDnode);
SProcCfg dmGenProcCfg(SMgmtWrapper *pWrapper);
SMsgCb dmGetMsgcb(SMgmtWrapper *pWrapper);
int32_t dmInitMsgHandle(SDnode *pDnode);
......
source/dnode/mgmt/implement/src/dmExec.c
浏览文件 @ 97f64a58
......@@ -213,10 +213,12 @@ static int32_t dmOpenNodes(SDnode *pDnode) {
}
pWrapper->procType = DND_PROC_CHILD;
if (dmInitClient(pDnode) != 0) {
return -1;
}
SMsgCb msgCb = pDnode->data.msgCb;
msgCb.pWrapper = pWrapper;
tmsgSetDefaultMsgCb(&msgCb);
pDnode->data.msgCb = dmGetMsgcb(pWrapper);
tmsgSetDefaultMsgCb(&pDnode->data.msgCb);
if (dmOpenNode(pWrapper) != 0) {
dError("node:%s, failed to open since %s", pWrapper->name, terrstr());
......@@ -234,6 +236,15 @@ static int32_t dmOpenNodes(SDnode *pDnode) {
pWrapper->procType = DND_PROC_SINGLE;
}
if (n == DNODE) {
if (dmInitClient(pDnode) != 0) {
return -1;
}
pDnode->data.msgCb = dmGetMsgcb(pWrapper);
tmsgSetDefaultMsgCb(&pDnode->data.msgCb);
}
if (dmOpenNode(pWrapper) != 0) {
dError("node:%s, failed to open since %s", pWrapper->name, terrstr());
return -1;
......@@ -281,21 +292,21 @@ static void dmProcessProcHandle(void *handle) {
}
static void dmWatchNodes(SDnode *pDnode) {
if (pDnode->ptype != DND_PROC_PARENT) return;
if (pDnode->ntype == NODE_END) return;
taosThreadMutexLock(&pDnode->mutex);
if (pDnode->ptype == DND_PROC_PARENT) {
for (EDndNodeType n = DNODE + 1; n < NODE_END; ++n) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
if (!pWrapper->required) continue;
if (pWrapper->procType != DND_PROC_PARENT) continue;
if (pDnode->ntype == NODE_END) continue;
for (EDndNodeType n = DNODE + 1; n < NODE_END; ++n) {
SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
if (!pWrapper->required) continue;
if (pWrapper->procType != DND_PROC_PARENT) continue;
if (pWrapper->procId <= 0 || !taosProcExist(pWrapper->procId)) {
dWarn("node:%s, process:%d is killed and needs to be restarted", pWrapper->name, pWrapper->procId);
if (pWrapper->procObj) {
taosProcCloseHandles(pWrapper->procObj, dmProcessProcHandle);
}
dmNewNodeProc(pWrapper, n);
if (pWrapper->procId <= 0 || !taosProcExist(pWrapper->procId)) {
dWarn("node:%s, process:%d is killed and needs to be restarted", pWrapper->name, pWrapper->procId);
if (pWrapper->procObj) {
taosProcCloseHandles(pWrapper->procObj, dmProcessProcHandle);
}
dmNewNodeProc(pWrapper, n);
}
}
taosThreadMutexUnlock(&pDnode->mutex);
......
source/dnode/mgmt/implement/src/dmHandle.c
浏览文件 @ 97f64a58
......@@ -217,145 +217,6 @@ static void dmStopMgmt(SMgmtWrapper *pWrapper) {
dmStopStatusThread(pWrapper->pDnode);
}
static int32_t dmSpawnUdfd(SDnode *pDnode);
void dmUdfdExit(uv_process_t *process, int64_t exitStatus, int termSignal) {
dInfo("udfd process exited with status %" PRId64 ", signal %d", exitStatus, termSignal);
SDnode *pDnode = process->data;
if (exitStatus == 0 && termSignal == 0 || atomic_load_32(&pDnode->udfdData.stopCalled)) {
dInfo("udfd process exit due to SIGINT or dnode-mgmt called stop");
} else {
dInfo("udfd process restart");
dmSpawnUdfd(pDnode);
}
}
static int32_t dmSpawnUdfd(SDnode *pDnode) {
dInfo("dnode start spawning udfd");
uv_process_options_t options = {0};
char path[PATH_MAX] = {0};
if (tsProcPath == NULL) {
path[0] = '.';
} else {
strncpy(path, tsProcPath, strlen(tsProcPath));
taosDirName(path);
}
#ifdef WINDOWS
strcat(path, "udfd.exe");
#else
strcat(path, "/udfd");
#endif
char* argsUdfd[] = {path, "-c", configDir, NULL};
options.args = argsUdfd;
options.file = path;
options.exit_cb = dmUdfdExit;
SUdfdData *pData = &pDnode->udfdData;
uv_pipe_init(&pData->loop, &pData->ctrlPipe, 1);
uv_stdio_container_t child_stdio[3];
child_stdio[0].flags = UV_CREATE_PIPE | UV_READABLE_PIPE;
child_stdio[0].data.stream = (uv_stream_t*) &pData->ctrlPipe;
child_stdio[1].flags = UV_IGNORE;
child_stdio[2].flags = UV_INHERIT_FD;
child_stdio[2].data.fd = 2;
options.stdio_count = 3;
options.stdio = child_stdio;
options.flags = UV_PROCESS_DETACHED;
char dnodeIdEnvItem[32] = {0};
char thrdPoolSizeEnvItem[32] = {0};
snprintf(dnodeIdEnvItem, 32, "%s=%d", "DNODE_ID", pDnode->data.dnodeId);
float numCpuCores = 4;
taosGetCpuCores(&numCpuCores);
snprintf(thrdPoolSizeEnvItem,32, "%s=%d", "UV_THREADPOOL_SIZE", (int)numCpuCores*2);
char* envUdfd[] = {dnodeIdEnvItem, thrdPoolSizeEnvItem, NULL};
options.env = envUdfd;
int err = uv_spawn(&pData->loop, &pData->process, &options);
pData->process.data = (void*)pDnode;
if (err != 0) {
dError("can not spawn udfd. path: %s, error: %s", path, uv_strerror(err));
}
return err;
}
static void dmUdfdCloseWalkCb(uv_handle_t* handle, void* arg) {
if (!uv_is_closing(handle)) {
uv_close(handle, NULL);
}
}
static void dmUdfdStopAsyncCb(uv_async_t *async) {
SDnode *pDnode = async->data;
SUdfdData *pData = &pDnode->udfdData;
uv_stop(&pData->loop);
}
static void dmWatchUdfd(void *args) {
SDnode *pDnode = args;
SUdfdData *pData = &pDnode->udfdData;
uv_loop_init(&pData->loop);
uv_async_init(&pData->loop, &pData->stopAsync, dmUdfdStopAsyncCb);
pData->stopAsync.data = pDnode;
int32_t err = dmSpawnUdfd(pDnode);
atomic_store_32(&pData->spawnErr, err);
uv_barrier_wait(&pData->barrier);
uv_run(&pData->loop, UV_RUN_DEFAULT);
uv_loop_close(&pData->loop);
uv_walk(&pData->loop, dmUdfdCloseWalkCb, NULL);
uv_run(&pData->loop, UV_RUN_DEFAULT);
uv_loop_close(&pData->loop);
return;
}
static int32_t dmStartUdfd(SDnode *pDnode) {
char dnodeId[8] = {0};
snprintf(dnodeId, sizeof(dnodeId), "%d", pDnode->data.dnodeId);
uv_os_setenv("DNODE_ID", dnodeId);
SUdfdData *pData = &pDnode->udfdData;
if (pData->startCalled) {
dInfo("dnode-mgmt start udfd already called");
return 0;
}
pData->startCalled = true;
uv_barrier_init(&pData->barrier, 2);
uv_thread_create(&pData->thread, dmWatchUdfd, pDnode);
uv_barrier_wait(&pData->barrier);
int32_t err = atomic_load_32(&pData->spawnErr);
if (err != 0) {
uv_barrier_destroy(&pData->barrier);
uv_async_send(&pData->stopAsync);
uv_thread_join(&pData->thread);
pData->needCleanUp = false;
dInfo("dnode-mgmt udfd cleaned up after spawn err");
} else {
pData->needCleanUp = true;
}
return err;
}
static int32_t dmStopUdfd(SDnode *pDnode) {
dInfo("dnode-mgmt to stop udfd. need cleanup: %d, spawn err: %d",
pDnode->udfdData.needCleanUp, pDnode->udfdData.spawnErr);
SUdfdData *pData = &pDnode->udfdData;
if (!pData->needCleanUp || atomic_load_32(&pData->stopCalled)) {
return 0;
}
atomic_store_32(&pData->stopCalled, 1);
pData->needCleanUp = false;
uv_barrier_destroy(&pData->barrier);
uv_async_send(&pData->stopAsync);
uv_thread_join(&pData->thread);
dInfo("dnode-mgmt udfd cleaned up");
return 0;
}
static int32_t dmInitMgmt(SMgmtWrapper *pWrapper) {
dInfo("dnode-mgmt start to init");
SDnode *pDnode = pWrapper->pDnode;
......@@ -381,13 +242,13 @@ static int32_t dmInitMgmt(SMgmtWrapper *pWrapper) {
return -1;
}
if (dmInitTrans(pDnode) != 0) {
if (dmInitServer(pDnode) != 0) {
dError("failed to init transport since %s", terrstr());
return -1;
}
dmReportStartup(pDnode, "dnode-transport", "initialized");
if (dmStartUdfd(pDnode) != 0) {
if (udfStartUdfd(pDnode->data.dnodeId) != 0) {
dError("failed to start udfd");
}
......@@ -398,7 +259,9 @@ static int32_t dmInitMgmt(SMgmtWrapper *pWrapper) {
static void dmCleanupMgmt(SMgmtWrapper *pWrapper) {
dInfo("dnode-mgmt start to clean up");
SDnode *pDnode = pWrapper->pDnode;
dmStopUdfd(pDnode);
udfStopUdfd();
dmStopWorker(pDnode);
taosWLockLatch(&pDnode->data.latch);
......@@ -412,7 +275,8 @@ static void dmCleanupMgmt(SMgmtWrapper *pWrapper) {
}
taosWUnLockLatch(&pDnode->data.latch);
dmCleanupTrans(pDnode);
dmCleanupClient(pDnode);
dmCleanupServer(pDnode);
dInfo("dnode-mgmt is cleaned up");
}
......
source/dnode/mgmt/implement/src/dmObj.c
浏览文件 @ 97f64a58
......@@ -124,9 +124,6 @@ SDnode *dmCreate(const SDnodeOpt *pOption) {
goto _OVER;
}
pDnode->data.msgCb = dmGetMsgcb(&pDnode->wrappers[DNODE]);
tmsgSetDefaultMsgCb(&pDnode->data.msgCb);
dInfo("dnode is created, data:%p", pDnode);
code = 0;
......
source/dnode/mgmt/implement/src/dmTransport.c
浏览文件 @ 97f64a58
......@@ -16,6 +16,8 @@
#define _DEFAULT_SOURCE
#include "dmImp.h"
#include "qworker.h"
#define INTERNAL_USER "_dnd"
#define INTERNAL_CKEY "_key"
#define INTERNAL_SECRET "_pwd"
......@@ -85,17 +87,14 @@ static void dmProcessRpcMsg(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, SEpSet *pEpSe
if ((pMsg = taosAllocateQitem(sizeof(SNodeMsg))) == NULL) goto _OVER;
if (dmBuildMsg(pMsg, pRpc) != 0) goto _OVER;
if (pWrapper->procType == DND_PROC_SINGLE) {
if (pWrapper->procType != DND_PROC_PARENT) {
dTrace("msg:%p, created, type:%s handle:%p user:%s", pMsg, TMSG_INFO(msgType), pRpc->handle, pMsg->user);
code = (*msgFp)(pWrapper, pMsg);
} else if (pWrapper->procType == DND_PROC_PARENT) {
} else {
dTrace("msg:%p, created and put into child queue, type:%s handle:%p code:0x%04x user:%s contLen:%d", pMsg,
TMSG_INFO(msgType), pRpc->handle, pMsg->rpcMsg.code & 0XFFFF, pMsg->user, pRpc->contLen);
code = taosProcPutToChildQ(pWrapper->procObj, pMsg, sizeof(SNodeMsg), pRpc->pCont, pRpc->contLen,
(isReq && (pMsg->rpcMsg.code == 0)) ? pRpc->handle : NULL, pRpc->refId, PROC_FUNC_REQ);
} else {
dTrace("msg:%p, should not processed in child process, handle:%p user:%s", pMsg, pRpc->handle, pMsg->user);
ASSERT(1);
}
_OVER:
......@@ -108,7 +107,7 @@ _OVER:
} else {
dError("msg:%p, type:%s handle:%p failed to process since 0x%04x:%s", pMsg, TMSG_INFO(msgType), pRpc->handle,
code & 0XFFFF, terrstr());
if (msgType & 1U) {
if (isReq) {
if (terrno != 0) code = terrno;
if (code == TSDB_CODE_NODE_NOT_DEPLOYED || code == TSDB_CODE_NODE_OFFLINE) {
if (msgType > TDMT_MND_MSG && msgType < TDMT_VND_MSG) {
......@@ -130,22 +129,27 @@ _OVER:
}
static void dmProcessMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
SDnodeTrans * pTrans = &pDnode->trans;
SDnodeTrans *pTrans = &pDnode->trans;
tmsg_t msgType = pMsg->msgType;
bool isReq = msgType & 1u;
SMsgHandle * pHandle = &pTrans->msgHandles[TMSG_INDEX(msgType)];
SMsgHandle *pHandle = &pTrans->msgHandles[TMSG_INDEX(msgType)];
SMgmtWrapper *pWrapper = pHandle->pNdWrapper;
if (msgType == TDMT_DND_SERVER_STATUS) {
dTrace("server status req will be processed, handle:%p, app:%p", pMsg->handle, pMsg->ahandle);
dmProcessServerStatusReq(pDnode, pMsg);
return;
}
if (msgType == TDMT_DND_NET_TEST) {
dTrace("net test req will be processed, handle:%p, app:%p", pMsg->handle, pMsg->ahandle);
dmProcessServerStatusReq(pDnode, pMsg);
return;
switch (msgType) {
case TDMT_DND_SERVER_STATUS:
dTrace("server status req will be processed, handle:%p, app:%p", pMsg->handle, pMsg->ahandle);
dmProcessServerStatusReq(pDnode, pMsg);
return;
case TDMT_DND_NET_TEST:
dTrace("net test req will be processed, handle:%p, app:%p", pMsg->handle, pMsg->ahandle);
dmProcessNetTestReq(pDnode, pMsg);
return;
case TDMT_MND_SYSTABLE_RETRIEVE_RSP:
case TDMT_VND_FETCH_RSP:
dTrace("retrieve rsp is received");
qWorkerProcessFetchRsp(NULL, NULL, pMsg);
pMsg->pCont = NULL; // already freed in qworker
return;
}
if (pDnode->status != DND_STAT_RUNNING) {
......@@ -233,16 +237,6 @@ int32_t dmInitMsgHandle(SDnode *pDnode) {
return 0;
}
static inline int32_t dmSendRpcReq(SDnode *pDnode, const SEpSet *pEpSet, SRpcMsg *pReq) {
if (pDnode->trans.clientRpc == NULL) {
terrno = TSDB_CODE_NODE_OFFLINE;
return -1;
}
rpcSendRequest(pDnode->trans.clientRpc, pEpSet, pReq, NULL);
return 0;
}
static void dmSendRpcRedirectRsp(SDnode *pDnode, const SRpcMsg *pReq) {
SEpSet epSet = {0};
dmGetMnodeEpSet(pDnode, &epSet);
......@@ -288,28 +282,20 @@ void dmSendToMnodeRecv(SDnode *pDnode, SRpcMsg *pReq, SRpcMsg *pRsp) {
}
static inline int32_t dmSendReq(SMgmtWrapper *pWrapper, const SEpSet *pEpSet, SRpcMsg *pReq) {
if (pWrapper->pDnode->status != DND_STAT_RUNNING) {
SDnode *pDnode = pWrapper->pDnode;
if (pDnode->status != DND_STAT_RUNNING) {
terrno = TSDB_CODE_NODE_OFFLINE;
dError("failed to send rpc msg since %s, handle:%p", terrstr(), pReq->handle);
return -1;
}
if (pWrapper->procType != DND_PROC_CHILD) {
return dmSendRpcReq(pWrapper->pDnode, pEpSet, pReq);
} else {
char *pHead = taosMemoryMalloc(sizeof(SRpcMsg) + sizeof(SEpSet));
if (pHead == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
memcpy(pHead, pReq, sizeof(SRpcMsg));
memcpy(pHead + sizeof(SRpcMsg), pEpSet, sizeof(SEpSet));
taosProcPutToParentQ(pWrapper->procObj, pHead, sizeof(SRpcMsg) + sizeof(SEpSet), pReq->pCont, pReq->contLen,
PROC_FUNC_REQ);
taosMemoryFree(pHead);
return 0;
if (pDnode->trans.clientRpc == NULL) {
terrno = TSDB_CODE_NODE_OFFLINE;
return -1;
}
rpcSendRequest(pDnode->trans.clientRpc, pEpSet, pReq, NULL);
return 0;
}
static inline void dmSendRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp) {
......@@ -396,9 +382,10 @@ static void dmConsumeParentQueue(SMgmtWrapper *pWrapper, SRpcMsg *pMsg, int16_t
pMsg->pCont = pCont;
if (ftype == PROC_FUNC_REQ) {
ASSERT(1);
dTrace("msg:%p, get from parent queue, send req:%s handle:%p code:0x%04x, app:%p", pMsg, TMSG_INFO(pMsg->msgType),
pMsg->handle, code, pMsg->ahandle);
dmSendRpcReq(pWrapper->pDnode, (SEpSet *)((char *)pMsg + sizeof(SRpcMsg)), pMsg);
dmSendReq(pWrapper, (SEpSet *)((char *)pMsg + sizeof(SRpcMsg)), pMsg);
} else if (ftype == PROC_FUNC_RSP) {
dTrace("msg:%p, get from parent queue, rsp handle:%p code:0x%04x, app:%p", pMsg, pMsg->handle, code, pMsg->ahandle);
pMsg->refId = taosProcRemoveHandle(pWrapper->procObj, pMsg->handle);
......@@ -421,23 +408,25 @@ static void dmConsumeParentQueue(SMgmtWrapper *pWrapper, SRpcMsg *pMsg, int16_t
}
SProcCfg dmGenProcCfg(SMgmtWrapper *pWrapper) {
SProcCfg cfg = {.childConsumeFp = (ProcConsumeFp)dmConsumeChildQueue,
.childMallocHeadFp = (ProcMallocFp)taosAllocateQitem,
.childFreeHeadFp = (ProcFreeFp)taosFreeQitem,
.childMallocBodyFp = (ProcMallocFp)rpcMallocCont,
.childFreeBodyFp = (ProcFreeFp)rpcFreeCont,
.parentConsumeFp = (ProcConsumeFp)dmConsumeParentQueue,
.parentMallocHeadFp = (ProcMallocFp)taosMemoryMalloc,
.parentFreeHeadFp = (ProcFreeFp)taosMemoryFree,
.parentMallocBodyFp = (ProcMallocFp)rpcMallocCont,
.parentFreeBodyFp = (ProcFreeFp)rpcFreeCont,
.shm = pWrapper->procShm,
.parent = pWrapper,
.name = pWrapper->name};
SProcCfg cfg = {
.childConsumeFp = (ProcConsumeFp)dmConsumeChildQueue,
.childMallocHeadFp = (ProcMallocFp)taosAllocateQitem,
.childFreeHeadFp = (ProcFreeFp)taosFreeQitem,
.childMallocBodyFp = (ProcMallocFp)rpcMallocCont,
.childFreeBodyFp = (ProcFreeFp)rpcFreeCont,
.parentConsumeFp = (ProcConsumeFp)dmConsumeParentQueue,
.parentMallocHeadFp = (ProcMallocFp)taosMemoryMalloc,
.parentFreeHeadFp = (ProcFreeFp)taosMemoryFree,
.parentMallocBodyFp = (ProcMallocFp)rpcMallocCont,
.parentFreeBodyFp = (ProcFreeFp)rpcFreeCont,
.shm = pWrapper->procShm,
.parent = pWrapper,
.name = pWrapper->name,
};
return cfg;
}
bool rpcRfp(int32_t code) {
static bool rpcRfp(int32_t code) {
if (code == TSDB_CODE_RPC_REDIRECT) {
return true;
} else {
......@@ -445,7 +434,7 @@ bool rpcRfp(int32_t code) {
}
}
static int32_t dmInitClient(SDnode *pDnode) {
int32_t dmInitClient(SDnode *pDnode) {
SDnodeTrans *pTrans = &pDnode->trans;
SRpcInit rpcInit = {0};
......@@ -471,11 +460,15 @@ static int32_t dmInitClient(SDnode *pDnode) {
return -1;
}
pDnode->data.msgCb = dmGetMsgcb(&pDnode->wrappers[DNODE]);
tmsgSetDefaultMsgCb(&pDnode->data.msgCb);
dDebug("dnode rpc client is initialized");
return 0;
}
static void dmCleanupClient(SDnode *pDnode) {
void dmCleanupClient(SDnode *pDnode) {
SDnodeTrans *pTrans = &pDnode->trans;
if (pTrans->clientRpc) {
rpcClose(pTrans->clientRpc);
......@@ -517,7 +510,7 @@ static inline int32_t dmRetrieveUserAuthInfo(SDnode *pDnode, char *user, char *s
SAuthReq authReq = {0};
tstrncpy(authReq.user, user, TSDB_USER_LEN);
int32_t contLen = tSerializeSAuthReq(NULL, 0, &authReq);
void * pReq = rpcMallocCont(contLen);
void *pReq = rpcMallocCont(contLen);
tSerializeSAuthReq(pReq, contLen, &authReq);
SRpcMsg rpcMsg = {.pCont = pReq, .contLen = contLen, .msgType = TDMT_MND_AUTH, .ahandle = (void *)9528};
......@@ -543,7 +536,7 @@ static inline int32_t dmRetrieveUserAuthInfo(SDnode *pDnode, char *user, char *s
return rpcRsp.code;
}
static int32_t dmInitServer(SDnode *pDnode) {
int32_t dmInitServer(SDnode *pDnode) {
SDnodeTrans *pTrans = &pDnode->trans;
SRpcInit rpcInit = {0};
......@@ -569,7 +562,7 @@ static int32_t dmInitServer(SDnode *pDnode) {
return 0;
}
static void dmCleanupServer(SDnode *pDnode) {
void dmCleanupServer(SDnode *pDnode) {
SDnodeTrans *pTrans = &pDnode->trans;
if (pTrans->serverRpc) {
rpcClose(pTrans->serverRpc);
......@@ -578,17 +571,6 @@ static void dmCleanupServer(SDnode *pDnode) {
}
}
int32_t dmInitTrans(SDnode *pDnode) {
if (dmInitServer(pDnode) != 0) return -1;
if (dmInitClient(pDnode) != 0) return -1;
return 0;
}
void dmCleanupTrans(SDnode *pDnode) {
dmCleanupServer(pDnode);
dmCleanupClient(pDnode);
}
SMsgCb dmGetMsgcb(SMgmtWrapper *pWrapper) {
SMsgCb msgCb = {
.sendReqFp = dmSendReq,
......@@ -598,6 +580,7 @@ SMsgCb dmGetMsgcb(SMgmtWrapper *pWrapper) {
.releaseHandleFp = dmReleaseHandle,
.reportStartupFp = dmReportStartupByWrapper,
.pWrapper = pWrapper,
.clientRpc = pWrapper->pDnode->trans.clientRpc,
};
return msgCb;
}
source/dnode/mgmt/interface/inc/dmDef.h
浏览文件 @ 97f64a58
......@@ -156,6 +156,8 @@ typedef struct SUdfdData {
uv_pipe_t ctrlPipe;
uv_async_t stopAsync;
int32_t stopCalled;
int32_t dnodeId;
} SUdfdData;
typedef struct SDnode {
......
source/dnode/mgmt/interface/inc/dmInt.h
浏览文件 @ 97f64a58
......@@ -37,7 +37,7 @@ void dmSetMsgHandle(SMgmtWrapper *pWrapper, tmsg_t msgType, NodeMsgFp nodeMsgF
void dmReportStartup(SDnode *pDnode, const char *pName, const char *pDesc);
void dmReportStartupByWrapper(SMgmtWrapper *pWrapper, const char *pName, const char *pDesc);
void dmProcessServerStatusReq(SDnode *pDnode, SRpcMsg *pMsg);
void dmProcessNettestReq(SDnode *pDnode, SRpcMsg *pMsg);
void dmProcessNetTestReq(SDnode *pDnode, SRpcMsg *pMsg);
void dmGetMonitorSysInfo(SMonSysInfo *pInfo);
// dmFile.c
......
source/dnode/mgmt/interface/src/dmInt.c
浏览文件 @ 97f64a58
......@@ -171,16 +171,17 @@ static void dmGetServerStatus(SDnode *pDnode, SServerStatusRsp *pStatus) {
}
}
void dmProcessNettestReq(SDnode *pDnode, SRpcMsg *pRpc) {
void dmProcessNetTestReq(SDnode *pDnode, SRpcMsg *pReq) {
dDebug("net test req is received");
SRpcMsg rsp = {.handle = pRpc->handle, .refId = pRpc->refId, .ahandle = pRpc->ahandle, .code = 0};
rsp.pCont = rpcMallocCont(pRpc->contLen);
SRpcMsg rsp = {.handle = pReq->handle, .refId = pReq->refId, .ahandle = pReq->ahandle, .code = 0};
rsp.pCont = rpcMallocCont(pReq->contLen);
if (rsp.pCont == NULL) {
rsp.code = TSDB_CODE_OUT_OF_MEMORY;
} else {
rsp.contLen = pRpc->contLen;
rsp.contLen = pReq->contLen;
}
rpcSendResponse(&rsp);
rpcFreeCont(pReq->pCont);
}
void dmProcessServerStatusReq(SDnode *pDnode, SRpcMsg *pReq) {
......@@ -208,6 +209,7 @@ void dmProcessServerStatusReq(SDnode *pDnode, SRpcMsg *pReq) {
_OVER:
rpcSendResponse(&rspMsg);
rpcFreeCont(pReq->pCont);
}
void dmGetMonitorSysInfo(SMonSysInfo *pInfo) {
......
source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
浏览文件 @ 97f64a58
......@@ -289,7 +289,6 @@ void vmInitMsgHandle(SMgmtWrapper *pWrapper) {
dmSetMsgHandle(pWrapper, TDMT_VND_QUERY, vmProcessQueryMsg, DEFAULT_HANDLE);
dmSetMsgHandle(pWrapper, TDMT_VND_QUERY_CONTINUE, vmProcessQueryMsg, DEFAULT_HANDLE);
dmSetMsgHandle(pWrapper, TDMT_VND_FETCH, vmProcessFetchMsg, DEFAULT_HANDLE);
dmSetMsgHandle(pWrapper, TDMT_VND_FETCH_RSP, vmProcessFetchMsg, DEFAULT_HANDLE);
dmSetMsgHandle(pWrapper, TDMT_VND_ALTER_TABLE, vmProcessWriteMsg, DEFAULT_HANDLE);
dmSetMsgHandle(pWrapper, TDMT_VND_UPDATE_TAG_VAL, vmProcessWriteMsg, DEFAULT_HANDLE);
dmSetMsgHandle(pWrapper, TDMT_VND_TABLE_META, vmProcessFetchMsg, DEFAULT_HANDLE);
......
source/dnode/mnode/impl/src/mndQuery.c
浏览文件 @ 97f64a58
......@@ -20,7 +20,7 @@
int32_t mndProcessQueryMsg(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
SReadHandle handle = {.mnd = pMnode};
SReadHandle handle = {.mnd = pMnode, .pMsgCb = &pMnode->msgCb};
mTrace("msg:%p, in query queue is processing", pReq);
switch (pReq->rpcMsg.msgType) {
......
source/dnode/qnode/src/qnode.c
浏览文件 @ 97f64a58
......@@ -51,7 +51,7 @@ int32_t qndGetLoad(SQnode *pQnode, SQnodeLoad *pLoad) { return 0; }
int32_t qndProcessQueryMsg(SQnode *pQnode, SRpcMsg *pMsg) {
qTrace("message in qnode query queue is processing");
SReadHandle handle = {0};
SReadHandle handle = {.pMsgCb = &pQnode->msgCb};
switch (pMsg->msgType) {
case TDMT_VND_QUERY: {
......
source/dnode/vnode/src/inc/vnodeInt.h
浏览文件 @ 97f64a58
......@@ -34,6 +34,7 @@
#include "tlockfree.h"
#include "tlosertree.h"
#include "tmallocator.h"
#include "tmsgcb.h"
#include "tskiplist.h"
#include "tstream.h"
#include "ttime.h"
......@@ -101,7 +102,7 @@ int32_t tsdbUpdateSmaWindow(STsdb* pTsdb, SSubmitReq* pMsg, int64_t version
int32_t tsdbCreateTSma(STsdb* pTsdb, char* pMsg);
int32_t tsdbInsertTSmaData(STsdb* pTsdb, int64_t indexUid, const char* msg);
int tsdbInsertData(STsdb* pTsdb, int64_t version, SSubmitReq* pMsg, SSubmitRsp* pRsp);
int tsdbInsertTableData(STsdb* pTsdb, SSubmitMsgIter* pMsgIter, SSubmitBlk* pBlock, int32_t* pAffectedRows);
int tsdbInsertTableData(STsdb* pTsdb, SSubmitMsgIter* pMsgIter, SSubmitBlk* pBlock, SSubmitBlkRsp* pRsp);
tsdbReaderT* tsdbQueryTables(SVnode* pVnode, SQueryTableDataCond* pCond, STableGroupInfo* groupList, uint64_t qId,
uint64_t taskId);
tsdbReaderT tsdbQueryCacheLastT(STsdb* tsdb, SQueryTableDataCond* pCond, STableGroupInfo* groupList, uint64_t qId,
......@@ -121,7 +122,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId);
// sma
int32_t tsdbRegisterRSma(STsdb* pTsdb, SMeta* pMeta, SVCreateStbReq* pReq);
int32_t tsdbRegisterRSma(STsdb* pTsdb, SMeta* pMeta, SVCreateStbReq* pReq, SMsgCb* pMsgCb);
int32_t tsdbFetchTbUidList(STsdb* pTsdb, STbUidStore** ppStore, tb_uid_t suid, tb_uid_t uid);
int32_t tsdbUpdateTbUidList(STsdb* pTsdb, STbUidStore* pUidStore);
void tsdbUidStoreDestory(STbUidStore* pStore);
......
source/dnode/vnode/src/tq/tq.c
浏览文件 @ 97f64a58
......@@ -378,6 +378,7 @@ int32_t tqDeserializeConsumer(STQ* pTq, const STqSerializedHead* pHead, STqConsu
SReadHandle handle = {
.reader = pReadHandle,
.meta = pTq->pVnode->pMeta,
.pMsgCb = &pTq->pVnode->msgCb,
};
pTopic->buffer.output[j].pReadHandle = pReadHandle;
pTopic->buffer.output[j].task = qCreateStreamExecTaskInfo(pTopic->qmsg, &handle);
......@@ -857,6 +858,7 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) {
SReadHandle handle = {
.reader = pExec->pExecReader[i],
.meta = pTq->pVnode->pMeta,
.pMsgCb = &pTq->pVnode->msgCb,
};
pExec->task[i] = qCreateStreamExecTaskInfo(pExec->qmsg, &handle);
ASSERT(pExec->task[i]);
......@@ -914,6 +916,7 @@ int32_t tqExpandTask(STQ* pTq, SStreamTask* pTask, int32_t parallel) {
SReadHandle handle = {
.reader = pStreamReader,
.meta = pTq->pVnode->pMeta,
.pMsgCb = &pTq->pVnode->msgCb,
};
pTask->exec.runners[i].inputHandle = pStreamReader;
pTask->exec.runners[i].executor = qCreateStreamExecTaskInfo(pTask->exec.qmsg, &handle);
......
source/dnode/vnode/src/tsdb/tsdbCommit.c
浏览文件 @ 97f64a58
......@@ -211,7 +211,7 @@ int tsdbCommit(STsdb *pRepo) {
void tsdbGetRtnSnap(STsdb *pRepo, SRtn *pRtn) {
STsdbKeepCfg *pCfg = REPO_KEEP_CFG(pRepo);
TSKEY minKey, midKey, maxKey, now;
TSKEY minKey, midKey, maxKey, now;
now = taosGetTimestamp(pCfg->precision);
minKey = now - pCfg->keep2 * tsTickPerDay[pCfg->precision];
......@@ -1386,34 +1386,7 @@ static void tsdbLoadAndMergeFromCache(SDataCols *pDataCols, int *iter, SCommitIt
tSkipListIterNext(pCommitIter->pIter);
} else {
#if 0
if (update != TD_ROW_OVERWRITE_UPDATE) {
// copy disk data
for (int i = 0; i < pDataCols->numOfCols; ++i) {
// TODO: dataColAppendVal may fail
SCellVal sVal = {0};
if (tdGetColDataOfRow(&sVal, pDataCols->cols + i, *iter, pDataCols->bitmapMode) < 0) {
TASSERT(0);
}
tdAppendValToDataCol(pTarget->cols + i, sVal.valType, sVal.val, pTarget->numOfRows, pTarget->maxPoints, pTarget->bitmapMode);
}
if (update == TD_ROW_DISCARD_UPDATE) pTarget->numOfRows++;
}
if (update != TD_ROW_DISCARD_UPDATE) {
// copy mem data
if (pSchema == NULL || schemaVersion(pSchema) != TD_ROW_SVER(row)) {
pSchema = tsdbGetTableSchemaImpl(pCommitIter->pTable, false, false, TD_ROW_SVER(row));
ASSERT(pSchema != NULL);
}
tdAppendSTSRowToDataCol(row, pSchema, pTarget, update == TD_ROW_OVERWRITE_UPDATE);
}
++(*iter);
tSkipListIterNext(pCommitIter->pIter);
#endif
if(lastKey != key1) {
if (lastKey != key1) {
lastKey = key1;
++pTarget->numOfRows;
}
......@@ -1484,29 +1457,4 @@ static bool tsdbCanAddSubBlock(SCommitH *pCommith, SBlock *pBlock, SMergeInfo *p
}
return false;
}
// int tsdbApplyRtn(STsdbRepo *pRepo) {
// SRtn rtn;
// SFSIter fsiter;
// STsdbFS * pfs = REPO_FS(pRepo);
// SDFileSet *pSet;
// // Get retention snapshot
// tsdbGetRtnSnap(pRepo, &rtn);
// tsdbFSIterInit(&fsiter, pfs, TSDB_FS_ITER_FORWARD);
// while ((pSet = tsdbFSIterNext(&fsiter))) {
// if (pSet->fid < rtn.minFid) {
// tsdbInfo("vgId:%d FSET %d at level %d disk id %d expires, remove it", REPO_ID(pRepo), pSet->fid,
// TSDB_FSET_LEVEL(pSet), TSDB_FSET_ID(pSet));
// continue;
// }
// if (tsdbApplyRtnOnFSet(pRepo, pSet, &rtn) < 0) {
// return -1;
// }
// }
// return 0;
// }
}
\ No newline at end of file
source/dnode/vnode/src/tsdb/tsdbMemTable.c
浏览文件 @ 97f64a58
......@@ -288,7 +288,7 @@ int tsdbLoadDataFromCache(STable *pTable, SSkipListIterator *pIter, TSKEY maxKey
return 0;
}
int tsdbInsertTableData(STsdb *pTsdb, SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlock, int32_t *pAffectedRows) {
int tsdbInsertTableData(STsdb *pTsdb, SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlock, SSubmitBlkRsp *pRsp) {
SSubmitBlkIter blkIter = {0};
STsdbMemTable *pMemTable = pTsdb->mem;
void *tptr;
......@@ -342,7 +342,8 @@ int tsdbInsertTableData(STsdb *pTsdb, SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlo
if (pMemTable->keyMin > keyMin) pMemTable->keyMin = keyMin;
if (pMemTable->keyMax < keyMax) pMemTable->keyMax = keyMax;
(*pAffectedRows) = pMsgIter->numOfRows;
pRsp->numOfRows = pRsp->numOfRows;
pRsp->affectedRows = pRsp->affectedRows;
return 0;
}
......
source/dnode/vnode/src/tsdb/tsdbMemTable2.c
浏览文件 @ 97f64a58
......@@ -21,16 +21,19 @@ typedef struct SMemSkipList SMemSkipList;
typedef struct SMemSkipListNode SMemSkipListNode;
typedef struct SMemSkipListCurosr SMemSkipListCurosr;
#define SL_MAX_LEVEL 5
struct SMemTable {
STsdb *pTsdb;
TSKEY minKey;
TSKEY maxKey;
int64_t minVer;
int64_t maxVer;
int64_t nRows;
int32_t nHash;
int32_t nBucket;
SMemData **pBuckets;
STsdb *pTsdb;
TSKEY minKey;
TSKEY maxKey;
int64_t minVer;
int64_t maxVer;
int64_t nRows;
int32_t nHash;
int32_t nBucket;
SMemData **pBuckets;
SMemSkipListCurosr *pSlc;
};
struct SMemSkipListNode {
......@@ -60,9 +63,15 @@ struct SMemData {
struct SMemSkipListCurosr {
SMemSkipList *pSl;
SMemSkipListNode *pNodeC;
SMemSkipListNode *pNodes[SL_MAX_LEVEL];
};
typedef struct {
int64_t version;
uint32_t szRow;
const STSRow *pRow;
} STsdbRow;
#define HASH_BUCKET(SUID, UID, NBUCKET) (TABS((SUID) + (UID)) % (NBUCKET))
#define SL_NODE_SIZE(l) (sizeof(SMemSkipListNode) + sizeof(SMemSkipListNode *) * (l)*2)
......@@ -76,7 +85,19 @@ struct SMemSkipListCurosr {
#define SL_HEAD_NODE_FORWARD(n, l) SL_NODE_FORWARD(n, l)
#define SL_TAIL_NODE_BACKWARD(n, l) SL_NODE_FORWARD(n, l)
static int8_t tsdbMemSkipListRandLevel(SMemSkipList *pSl);
static int8_t tsdbMemSkipListRandLevel(SMemSkipList *pSl);
static int32_t tsdbEncodeRow(SEncoder *pEncoder, const STsdbRow *pRow);
static int32_t tsdbDecodeRow(SDecoder *pDecoder, STsdbRow *pRow);
static int32_t tsdbMemSkipListCursorCreate(int8_t maxLevel, SMemSkipListCurosr **ppSlc);
static void tsdbMemSkipListCursorDestroy(SMemSkipListCurosr *pSlc);
static void tsdbMemSkipListCursorInit(SMemSkipListCurosr *pSlc, SMemSkipList *pSl);
static void tsdbMemSkipListCursorPut(SMemSkipListCurosr *pSlc, SMemSkipListNode *pNode);
static int32_t tsdbMemSkipListCursorMoveTo(SMemSkipListCurosr *pSlc, int64_t version, TSKEY ts, int32_t flags);
static void tsdbMemSkipListCursorMoveToFirst(SMemSkipListCurosr *pSlc);
static void tsdbMemSkipListCursorMoveToLast(SMemSkipListCurosr *pSlc);
static int32_t tsdbMemSkipListCursorMoveToNext(SMemSkipListCurosr *pSlc);
static int32_t tsdbMemSkipListCursorMoveToPrev(SMemSkipListCurosr *pSlc);
static SMemSkipListNode *tsdbMemSkipListNodeCreate(SVBufPool *pPool, SMemSkipList *pSl, const STsdbRow *pTRow);
// SMemTable
int32_t tsdbMemTableCreate2(STsdb *pTsdb, SMemTable **ppMemTb) {
......@@ -102,6 +123,11 @@ int32_t tsdbMemTableCreate2(STsdb *pTsdb, SMemTable **ppMemTb) {
taosMemoryFree(pMemTb);
return -1;
}
if (tsdbMemSkipListCursorCreate(pTsdb->pVnode->config.tsdbCfg.slLevel, &pMemTb->pSlc) < 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
taosMemoryFree(pMemTb->pBuckets);
taosMemoryFree(pMemTb);
}
*ppMemTb = pMemTb;
return 0;
......@@ -110,6 +136,7 @@ int32_t tsdbMemTableCreate2(STsdb *pTsdb, SMemTable **ppMemTb) {
int32_t tsdbMemTableDestroy2(STsdb *pTsdb, SMemTable *pMemTb) {
if (pMemTb) {
// loop to destroy the contents (todo)
tsdbMemSkipListCursorDestroy(pMemTb->pSlc);
taosMemoryFree(pMemTb->pBuckets);
taosMemoryFree(pMemTb);
}
......@@ -177,52 +204,42 @@ int32_t tsdbInsertData2(SMemTable *pMemTb, int64_t version, const SVSubmitBlk *p
}
// do insert data to SMemData
SMemSkipListCurosr slc = {0};
const STSRow *pRow;
uint32_t szRow;
SDecoder decoder = {0};
tDecoderInit(&decoder, pSubmitBlk->pData, pSubmitBlk->nData);
for (;;) {
if (tDecodeIsEnd(&decoder)) break;
if (tDecodeBinary(&decoder, (const uint8_t **)&pRow, &szRow) < 0) {
SMemSkipListNode *forwards[SL_MAX_LEVEL];
SMemSkipListNode *pNode;
int32_t iRow;
STsdbRow tRow = {.version = version};
SEncoder ec = {0};
SDecoder dc = {0};
tDecoderInit(&dc, pSubmitBlk->pData, pSubmitBlk->nData);
tsdbMemSkipListCursorInit(pMemTb->pSlc, &pMemData->sl);
for (iRow = 0;; iRow++) {
if (tDecodeIsEnd(&dc)) break;
// decode row
if (tDecodeBinary(&dc, (const uint8_t **)&tRow.pRow, &tRow.szRow) < 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
// check the row (todo)
// // move the cursor to position to write (todo)
// int32_t c;
// tsdbMemSkipListCursorMoveTo(&slc, pTSRow, version, &c);
// ASSERT(c);
// move cursor
tsdbMemSkipListCursorMoveTo(pMemTb->pSlc, version, tRow.pRow->ts, 0);
// encode row
int8_t level = tsdbMemSkipListRandLevel(&pMemData->sl);
int32_t tsize = SL_NODE_SIZE(level) + sizeof(version) + (0 /*todo*/);
SMemSkipListNode *pNode = vnodeBufPoolMalloc(pPool, tsize);
pNode = tsdbMemSkipListNodeCreate(pPool, &pMemData->sl, &tRow);
if (pNode == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
pNode->level = level;
// uint8_t *pData = SL_NODE_DATA(pSlNode);
// *(int64_t *)pData = version;
// pData += sizeof(version);
// memcpy(pData, pt, p - pt);
// // insert row
// tsdbMemSkipListCursorPut(&slc, pSlNode);
// put the node
tsdbMemSkipListCursorPut(pMemTb->pSlc, pNode);
// update status
if (pRow->ts < pMemData->minKey) pMemData->minKey = pRow->ts;
if (pRow->ts > pMemData->maxKey) pMemData->maxKey = pRow->ts;
if (tRow.pRow->ts < pMemData->minKey) pMemData->minKey = tRow.pRow->ts;
if (tRow.pRow->ts > pMemData->maxKey) pMemData->maxKey = tRow.pRow->ts;
}
tDecoderClear(&decoder);
// tsdbMemSkipListCursorClose(&slc);
tDecoderClear(&dc);
// update status
if (pMemData->minVer == -1) pMemData->minVer = version;
......@@ -236,17 +253,128 @@ int32_t tsdbInsertData2(SMemTable *pMemTb, int64_t version, const SVSubmitBlk *p
return 0;
}
static int8_t tsdbMemSkipListRandLevel(SMemSkipList *pSl) {
static FORCE_INLINE int8_t tsdbMemSkipListRandLevel(SMemSkipList *pSl) {
int8_t level = 1;
int8_t tlevel;
int8_t tlevel = TMIN(pSl->maxLevel, pSl->level + 1);
const uint32_t factor = 4;
if (pSl->size) {
tlevel = TMIN(pSl->maxLevel, pSl->level + 1);
while ((taosRandR(&pSl->seed) % factor) == 0 && level < tlevel) {
level++;
}
while ((taosRandR(&pSl->seed) % factor) == 0 && level < tlevel) {
level++;
}
return level;
}
static FORCE_INLINE int32_t tsdbEncodeRow(SEncoder *pEncoder, const STsdbRow *pRow) {
if (tEncodeI64(pEncoder, pRow->version) < 0) return -1;
if (tEncodeBinary(pEncoder, (const uint8_t *)pRow->pRow, pRow->szRow) < 0) return -1;
return 0;
}
static FORCE_INLINE int32_t tsdbDecodeRow(SDecoder *pDecoder, STsdbRow *pRow) {
if (tDecodeI64(pDecoder, &pRow->version) < 0) return -1;
if (tDecodeBinary(pDecoder, (const uint8_t **)&pRow->pRow, &pRow->szRow) < 0) return -1;
return 0;
}
static int32_t tsdbMemSkipListCursorCreate(int8_t maxLevel, SMemSkipListCurosr **ppSlc) {
*ppSlc = (SMemSkipListCurosr *)taosMemoryCalloc(1, sizeof(**ppSlc) + sizeof(SMemSkipListNode *) * maxLevel);
if (*ppSlc == NULL) {
return -1;
}
return 0;
}
static void tsdbMemSkipListCursorDestroy(SMemSkipListCurosr *pSlc) { taosMemoryFree(pSlc); }
static void tsdbMemSkipListCursorInit(SMemSkipListCurosr *pSlc, SMemSkipList *pSl) {
SMemSkipListNode *pHead = SL_HEAD_NODE(pSl);
pSlc->pSl = pSl;
// for (int8_t iLevel = 0; iLevel < pSl->maxLevel; iLevel++) {
// pSlc->forwards[iLevel] = pHead;
// }
}
static void tsdbMemSkipListCursorPut(SMemSkipListCurosr *pSlc, SMemSkipListNode *pNode) {
SMemSkipList *pSl = pSlc->pSl;
SMemSkipListNode *pNodeNext;
for (int8_t iLevel = 0; iLevel < pNode->level; iLevel++) {
// todo
ASSERT(0);
}
if (pSl->level < pNode->level) {
pSl->level = pNode->level;
}
pSl->size += 1;
}
static int32_t tsdbMemSkipListCursorMoveTo(SMemSkipListCurosr *pSlc, int64_t version, TSKEY ts, int32_t flags) {
SMemSkipListNode **pForwards = NULL;
SMemSkipList *pSl = pSlc->pSl;
int8_t maxLevel = pSl->maxLevel;
SMemSkipListNode *pHead = SL_HEAD_NODE(pSl);
SMemSkipListNode *pTail = SL_TAIL_NODE(pSl);
if (pSl->size == 0) {
for (int8_t iLevel = 0; iLevel < pSl->maxLevel; iLevel++) {
pForwards[iLevel] = pHead;
}
}
return 0;
}
static void tsdbMemSkipListCursorMoveToFirst(SMemSkipListCurosr *pSlc) {
SMemSkipList *pSl = pSlc->pSl;
SMemSkipListNode *pHead = SL_HEAD_NODE(pSl);
for (int8_t iLevel = 0; iLevel < pSl->maxLevel; iLevel++) {
pSlc->pNodes[iLevel] = pHead;
}
tsdbMemSkipListCursorMoveToNext(pSlc);
}
static void tsdbMemSkipListCursorMoveToLast(SMemSkipListCurosr *pSlc) {
SMemSkipList *pSl = pSlc->pSl;
SMemSkipListNode *pTail = SL_TAIL_NODE(pSl);
for (int8_t iLevel = 0; iLevel < pSl->maxLevel; iLevel++) {
pSlc->pNodes[iLevel] = pTail;
}
tsdbMemSkipListCursorMoveToPrev(pSlc);
}
static int32_t tsdbMemSkipListCursorMoveToNext(SMemSkipListCurosr *pSlc) {
// TODO
return 0;
}
static int32_t tsdbMemSkipListCursorMoveToPrev(SMemSkipListCurosr *pSlc) {
// TODO
return 0;
}
static SMemSkipListNode *tsdbMemSkipListNodeCreate(SVBufPool *pPool, SMemSkipList *pSl, const STsdbRow *pTRow) {
int32_t tsize;
int32_t ret;
int8_t level = tsdbMemSkipListRandLevel(pSl);
SMemSkipListNode *pNode = NULL;
SEncoder ec = {0};
tEncodeSize(tsdbEncodeRow, pTRow, tsize, ret);
pNode = vnodeBufPoolMalloc(pPool, tsize + SL_NODE_SIZE(level));
if (pNode) {
pNode->level = level;
tEncoderInit(&ec, (uint8_t *)SL_NODE_DATA(pNode), tsize);
tsdbEncodeRow(&ec, pTRow);
tEncoderClear(&ec);
}
return pNode;
}
\ No newline at end of file
source/dnode/vnode/src/tsdb/tsdbRead.c
浏览文件 @ 97f64a58
......@@ -351,35 +351,26 @@ static void setQueryTimewindow(STsdbReadHandle* pTsdbReadHandle, SQueryTableData
pTsdbReadHandle->window.ekey, pTsdbReadHandle->idStr);
}
}
#if 1
int nQUERY = 0;
#endif
static STsdb* getTsdbByRetentions(SVnode* pVnode, STsdbReadHandle* pReadHandle, TSKEY winSKey, SRetention* retentions) {
if (vnodeIsRollup(pVnode)) {
int level = 0;
#if 0
int level = 0;
int64_t now = taosGetTimestamp(pVnode->config.tsdbCfg.precision);
for (int i = 0; i < TSDB_RETENTION_MAX; ++i) {
SRetention* pRetention = retentions + i;
if (pRetention->keep <= 0 || (now - pRetention->keep) >= winSKey) {
SRetention* pRetention = retentions + level;
if (pRetention->keep <= 0) {
if (level > 0) {
--level;
}
break;
}
++level;
}
#endif
#if 1
switch ((nQUERY++) % 3) {
case 0:
level = 0;
break;
case 1:
level = 1;
break;
default:
level = 2;
if ((now - pRetention->keep) <= winSKey) {
break;
}
++level;
}
#endif
if (level == TSDB_RETENTION_L0) {
tsdbDebug("%p rsma level %d is selected to query\n", pReadHandle, level);
return VND_RSMA0(pVnode);
......@@ -391,7 +382,7 @@ static STsdb* getTsdbByRetentions(SVnode* pVnode, STsdbReadHandle* pReadHandle,
return VND_RSMA2(pVnode);
}
}
return pVnode->pTsdb;
return VND_TSDB(pVnode);
}
static STsdbReadHandle* tsdbQueryTablesImpl(SVnode* pVnode, SQueryTableDataCond* pCond, uint64_t qId, uint64_t taskId) {
......
source/dnode/vnode/src/tsdb/tsdbSma.c
浏览文件 @ 97f64a58
......@@ -1698,7 +1698,7 @@ int32_t tsdbDropTSma(STsdb *pTsdb, char *pMsg) {
* @param pReq
* @return int32_t
*/
int32_t tsdbRegisterRSma(STsdb *pTsdb, SMeta *pMeta, SVCreateStbReq *pReq) {
int32_t tsdbRegisterRSma(STsdb *pTsdb, SMeta *pMeta, SVCreateStbReq *pReq, SMsgCb *pMsgCb) {
if (!pReq->rollup) {
tsdbDebug("vgId:%d return directly since no rollup for stable %s %" PRIi64, REPO_ID(pTsdb), pReq->name, pReq->suid);
return TSDB_CODE_SUCCESS;
......@@ -1742,6 +1742,7 @@ int32_t tsdbRegisterRSma(STsdb *pTsdb, SMeta *pMeta, SVCreateStbReq *pReq) {
SReadHandle handle = {
.reader = pReadHandle,
.meta = pMeta,
.pMsgCb = pMsgCb,
};
if (param->qmsg1) {
......
source/dnode/vnode/src/tsdb/tsdbWrite.c
浏览文件 @ 97f64a58
......@@ -36,9 +36,10 @@ int tsdbInsertData(STsdb *pTsdb, int64_t version, SSubmitReq *pMsg, SSubmitRsp *
// loop to insert
tInitSubmitMsgIter(pMsg, &msgIter);
while (true) {
SSubmitBlkRsp r = {0};
tGetSubmitMsgNext(&msgIter, &pBlock);
if (pBlock == NULL) break;
if (tsdbInsertTableData(pTsdb, &msgIter, pBlock, &affectedrows) < 0) {
if (tsdbInsertTableData(pTsdb, &msgIter, pBlock, &r) < 0) {
return -1;
}
......@@ -46,8 +47,8 @@ int tsdbInsertData(STsdb *pTsdb, int64_t version, SSubmitReq *pMsg, SSubmitRsp *
}
if (pRsp != NULL) {
pRsp->affectedRows = affectedrows;
pRsp->numOfRows = numOfRows;
// pRsp->affectedRows = affectedrows;
// pRsp->numOfRows = numOfRows;
}
return 0;
......
source/libs/executor/inc/executorimpl.h
浏览文件 @ 97f64a58
......@@ -392,10 +392,7 @@ typedef struct SStreamBlockScanInfo {
} SStreamBlockScanInfo;
typedef struct SSysTableScanInfo {
union {
void* pTransporter;
SReadHandle readHandle;
};
SReadHandle readHandle;
SRetrieveMetaTableRsp* pRsp;
SRetrieveTableReq req;
......@@ -663,8 +660,6 @@ SResultRow* doSetResultOutBufByKey(SDiskbasedBuf* pResultBuf, SResultRowInfo* pR
char* pData, int16_t bytes, bool masterscan, uint64_t groupId,
SExecTaskInfo* pTaskInfo, bool isIntervalQuery, SAggSupporter* pSup);
SOperatorInfo* createExchangeOperatorInfo(const SNodeList* pSources, SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createTableScanOperatorInfo(void* pDataReader, SQueryTableDataCond* pCond, int32_t numOfOutput, int32_t dataLoadFlag, const uint8_t* scanInfo,
SArray* pColMatchInfo, SSDataBlock* pResBlock, SNode* pCondition, SInterval* pInterval, double sampleRatio, SExecTaskInfo* pTaskInfo);
......
source/libs/executor/src/groupoperator.c
浏览文件 @ 97f64a58
......@@ -262,6 +262,8 @@ static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator) {
return NULL;
}
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
SGroupbyOperatorInfo* pInfo = pOperator->info;
SSDataBlock* pRes = pInfo->binfo.pRes;
......@@ -289,7 +291,10 @@ static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator) {
// there is an scalar expression that needs to be calculated right before apply the group aggregation.
if (pInfo->pScalarExprInfo != NULL) {
projectApplyFunctions(pInfo->pScalarExprInfo, pBlock, pBlock, pInfo->pScalarFuncCtx, pInfo->numOfScalarExpr, NULL);
pTaskInfo->code = projectApplyFunctions(pInfo->pScalarExprInfo, pBlock, pBlock, pInfo->pScalarFuncCtx, pInfo->numOfScalarExpr, NULL);
if (pTaskInfo->code != TSDB_CODE_SUCCESS) {
longjmp(pTaskInfo->env, pTaskInfo->code);
}
}
// setTagValue(pOperator, pRuntimeEnv->current->pTable, pInfo->binfo.pCtx, pOperator->numOfExprs);
......
source/libs/executor/src/scanoperator.c
浏览文件 @ 97f64a58
......@@ -1057,7 +1057,8 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) {
pMsgSendInfo->fp = loadSysTableCallback;
int64_t transporterId = 0;
int32_t code = asyncSendMsgToServer(pInfo->pTransporter, &pInfo->epSet, &transporterId, pMsgSendInfo);
int32_t code =
asyncSendMsgToServer(pInfo->readHandle.pMsgCb->clientRpc, &pInfo->epSet, &transporterId, pMsgSendInfo);
tsem_wait(&pInfo->ready);
if (pTaskInfo->code) {
......@@ -1182,29 +1183,7 @@ SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSDataBlock* pRe
} else {
tsem_init(&pInfo->ready, 0, 0);
pInfo->epSet = epset;
#if 1
{ // todo refactor
SRpcInit rpcInit;
memset(&rpcInit, 0, sizeof(rpcInit));
rpcInit.localPort = 0;
rpcInit.label = "DB-META";
rpcInit.numOfThreads = 1;
rpcInit.cfp = qProcessFetchRsp;
rpcInit.sessions = tsMaxConnections;
rpcInit.connType = TAOS_CONN_CLIENT;
rpcInit.user = (char*)"root";
rpcInit.idleTime = tsShellActivityTimer * 1000;
rpcInit.ckey = "key";
rpcInit.spi = 1;
rpcInit.secret = (char*)"dcc5bed04851fec854c035b2e40263b6";
pInfo->pTransporter = rpcOpen(&rpcInit);
if (pInfo->pTransporter == NULL) {
return NULL; // todo
}
}
#endif
pInfo->readHandle = *(SReadHandle*)readHandle;
}
pOperator->name = "SysTableScanOperator";
......
source/libs/executor/src/sortoperator.c
浏览文件 @ 97f64a58
......@@ -114,7 +114,10 @@ void applyScalarFunction(SSDataBlock* pBlock, void* param) {
SOperatorInfo* pOperator = param;
SSortOperatorInfo* pSort = pOperator->info;
if (pOperator->pExpr != NULL) {
projectApplyFunctions(pOperator->pExpr, pBlock, pBlock, pSort->binfo.pCtx, pOperator->numOfExprs, NULL);
int32_t code = projectApplyFunctions(pOperator->pExpr, pBlock, pBlock, pSort->binfo.pCtx, pOperator->numOfExprs, NULL);
if (code != TSDB_CODE_SUCCESS) {
longjmp(pOperator->pTaskInfo->env, code);
}
}
}
......
source/libs/function/src/builtins.c
浏览文件 @ 97f64a58
此差异已折叠。
source/libs/function/src/tudf.c
浏览文件 @ 97f64a58
此差异已折叠。
source/libs/index/inc/indexComm.h
浏览文件 @ 97f64a58
此差异已折叠。
source/libs/index/src/indexCache.c
浏览文件 @ 97f64a58
此差异已折叠。
source/libs/index/src/indexComm.c
浏览文件 @ 97f64a58
此差异已折叠。
source/libs/index/src/indexTfile.c
浏览文件 @ 97f64a58
此差异已折叠。
source/libs/planner/test/planBasicTest.cpp
浏览文件 @ 97f64a58
......@@ -48,4 +48,6 @@ TEST_F(PlanBasicTest, func) {
useDb("root", "test");
run("SELECT DIFF(c1) FROM t1");
run("SELECT PERCENTILE(c1, 60) FROM t1");
}
source/libs/qcom/src/queryUtil.c
浏览文件 @ 97f64a58
此差异已折叠。
source/libs/sync/src/syncMain.c
浏览文件 @ 97f64a58
此差异已折叠。
source/libs/sync/src/syncRaftLog.c
浏览文件 @ 97f64a58
此差异已折叠。
source/util/src/tskiplist2.c 已删除 100644 → 0
浏览文件 @ 005ff0ed
此差异已折叠。
source/util/test/encodeTest.cpp
浏览文件 @ 97f64a58
此差异已折叠。
tests/script/sh/exec.sh
浏览文件 @ 97f64a58
此差异已折叠。
tests/script/test.sh
浏览文件 @ 97f64a58
此差异已折叠。
tests/script/tsim/insert/basic0.sim
浏览文件 @ 97f64a58
此差异已折叠。
tests/script/tsim/query/udf.sim
浏览文件 @ 97f64a58
此差异已折叠。
tests/system-test/2-query/diff.py
浏览文件 @ 97f64a58
此差异已折叠。
tests/system-test/fulltest.sh
浏览文件 @ 97f64a58
此差异已折叠。
tests/system-test/insert.json 0 → 100644
浏览文件 @ 97f64a58
此差异已折叠。
taos-tools @ 0ae9f872
比较 2f3dfddd...0ae9f872
此差异已折叠。
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或