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

refact: tdata

This commit is contained in:
Hongze Cheng 2022-05-09 09:43:15 +00:00
parent 37c122f0a7
commit db998d29b6
7 changed files with 96 additions and 989 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
include/common/tdata.h Normal file
View File

@ -0,0 +1,45 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_TDATA_H_
#define _TD_TDATA_H_
#ifdef __cplusplus
extern "C" {
#endif
#include "os.h"
typedef struct STaosData TDATA, tdata_t;
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 STaosData {
ETaosDataT type;
uint32_t nPayload;
uint8_t *pPayload;
};
#ifdef __cplusplus
}
#endif
#endif /*_TD_TDATA_H_*/

410
include/common/tdataformat.h
View File

@ -150,29 +150,6 @@ int32_t tdAddColToSchema(STSchemaBuilder *pBuilder, int8_t type, int8_t flags,
STSchema *tdGetSchemaFromBuilder(STSchemaBuilder *pBuilder); STSchema *tdGetSchemaFromBuilder(STSchemaBuilder *pBuilder);
// ----------------- Semantic timestamp key definition // ----------------- 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; // typedef uint64_t TKEY;
#define TKEY TSKEY #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)) #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) { static FORCE_INLINE TKEY keyToTkey(TSKEY key) {
TSKEY lkey = key; TSKEY lkey = key;
if (key > MAX_TS_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 // ----------------- Data column structure
// SDataCol arrangement: data => bitmap => dataOffset // SDataCol arrangement: data => bitmap => dataOffset
typedef struct SDataCol { typedef struct SDataCol {
@ -398,29 +222,6 @@ void *dataColSetOffset(SDataCol *pCol, int32_t nEle);
bool isNEleNull(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 { typedef struct {
col_id_t maxCols; // max number of columns col_id_t maxCols; // max number of columns
col_id_t numOfCols; // Total number of cols col_id_t numOfCols; // Total number of cols
@ -479,7 +280,8 @@ void tdResetDataCols(SDataCols *pCols);
int32_t tdInitDataCols(SDataCols *pCols, STSchema *pSchema); int32_t tdInitDataCols(SDataCols *pCols, STSchema *pSchema);
SDataCols *tdDupDataCols(SDataCols *pCols, bool keepData); SDataCols *tdDupDataCols(SDataCols *pCols, bool keepData);
SDataCols *tdFreeDataCols(SDataCols *pCols); 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 // ----------------- K-V data row structure
/* |<-------------------------------------- len -------------------------------------------->| /* |<-------------------------------------- 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); 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 // ----------------- K-V data row builder
typedef struct { typedef struct {
int16_t tCols; int16_t tCols;
@ -632,166 +386,6 @@ static FORCE_INLINE int32_t tdAddColToKVRow(SKVRowBuilder *pBuilder, col_id_t co
return 0; 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 #ifdef __cplusplus
} }

76
include/common/trow.h
View File

@ -165,15 +165,15 @@ typedef struct {
#define TD_ROW_HEAD_LEN (sizeof(STSRow)) #define TD_ROW_HEAD_LEN (sizeof(STSRow))
#define TD_ROW_NCOLS_LEN (sizeof(col_id_t)) #define TD_ROW_NCOLS_LEN (sizeof(col_id_t))
#define TD_ROW_INFO(r) ((r)->info) #define TD_ROW_INFO(r) ((r)->info)
#define TD_ROW_TYPE(r) ((r)->type) #define TD_ROW_TYPE(r) ((r)->type)
#define TD_ROW_DELETE(r) ((r)->del) #define TD_ROW_DELETE(r) ((r)->del)
#define TD_ROW_ENDIAN(r) ((r)->endian) #define TD_ROW_ENDIAN(r) ((r)->endian)
#define TD_ROW_SVER(r) ((r)->sver) #define TD_ROW_SVER(r) ((r)->sver)
#define TD_ROW_NCOLS(r) ((r)->data) // only valid for SKvRow #define TD_ROW_NCOLS(r) ((r)->data) // only valid for SKvRow
#define TD_ROW_DATA(r) ((r)->data) #define TD_ROW_DATA(r) ((r)->data)
#define TD_ROW_LEN(r) ((r)->len) #define TD_ROW_LEN(r) ((r)->len)
#define TD_ROW_KEY(r) ((r)->ts) #define TD_ROW_KEY(r) ((r)->ts)
// #define TD_ROW_VER(r) ((r)->ver) // #define TD_ROW_VER(r) ((r)->ver)
#define TD_ROW_KEY_ADDR(r) (r) #define TD_ROW_KEY_ADDR(r) (r)
@ -1410,64 +1410,6 @@ static void tdSRowPrint(STSRow *row, STSchema *pSchema, const char *tag) {
} }
printf("\n"); printf("\n");
} }
#ifdef TROW_ORIGIN_HZ
typedef struct {
uint32_t nRows;
char rows[];
} STSRowBatch;
static void tdSRowPrint(STSRow *row) {
printf("type:%d, del:%d, sver:%d\n", row->type, row->del, row->sver);
printf("isDeleted:%s, isTpRow:%s, isKvRow:%s\n", TD_BOOL_STR(TD_ROW_IS_DELETED(row)), TD_BOOL_STR(TD_IS_TP_ROW(row)),
TD_BOOL_STR(TD_IS_KV_ROW(row)));
}
typedef enum {
/// tuple row builder
TD_TP_ROW_BUILDER = 0,
/// kv row builder
TD_KV_ROW_BUILDER,
/// self-determined row builder
TD_SD_ROW_BUILDER
} ERowBbuilderT;
typedef struct {
/// row builder type
ERowBbuilderT type;
/// buffer writer
SBufferWriter bw;
/// target row
STSRow *pRow;
} STSRowBuilder;
typedef struct {
STSchema *pSchema;
STSRow *pRow;
} STSRowReader;
typedef struct {
uint32_t it;
STSRowBatch *pRowBatch;
} STSRowBatchIter;
// STSRowBuilder
#define trbInit(rt, allocator, endian, target, size) \
{ .type = (rt), .bw = tbufInitWriter(allocator, endian), .pRow = (target) }
void trbSetRowInfo(STSRowBuilder *pRB, bool del, uint16_t sver);
void trbSetRowVersion(STSRowBuilder *pRB, uint64_t ver);
void trbSetRowTS(STSRowBuilder *pRB, TSKEY ts);
int32_t trbWriteCol(STSRowBuilder *pRB, void *pData, col_id_t cid);
// STSRowReader
#define tRowReaderInit(schema, row) \
{ .schema = (schema), .row = (row) }
int32_t tRowReaderRead(STSRowReader *pRowReader, col_id_t cid, void *target, uint64_t size);
// STSRowBatchIter
#define tRowBatchIterInit(pRB) \
{ .it = 0, .pRowBatch = (pRB) }
const STSRow *tRowBatchIterNext(STSRowBatchIter *pRowBatchIter);
#endif
#ifdef __cplusplus #ifdef __cplusplus
} }

14
source/common/src/tdata.c Normal file
View File

@ -0,0 +1,14 @@
/*
* 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/>.
*/

450
source/common/src/tdataformat.c
View File

@ -129,50 +129,6 @@ void *tdDecodeSchema(void *buf, STSchema **pRSchema) {
return buf; 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) { int tdInitTSchemaBuilder(STSchemaBuilder *pBuilder, schema_ver_t version) {
if (pBuilder == NULL) return -1; if (pBuilder == NULL) return -1;
@ -260,49 +216,6 @@ STSchema *tdGetSchemaFromBuilder(STSchemaBuilder *pBuilder) {
return pSchema; 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) { void dataColInit(SDataCol *pDataCol, STColumn *pCol, int maxPoints) {
pDataCol->type = colType(pCol); pDataCol->type = colType(pCol);
pDataCol->colId = colColId(pCol); pDataCol->colId = colColId(pCol);
@ -312,39 +225,6 @@ void dataColInit(SDataCol *pDataCol, STColumn *pCol, int maxPoints) {
pDataCol->len = 0; 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) { static FORCE_INLINE const void *tdGetColDataOfRowUnsafe(SDataCol *pCol, int row) {
if (IS_VAR_DATA_TYPE(pCol->type)) { if (IS_VAR_DATA_TYPE(pCol->type)) {
return POINTER_SHIFT(pCol->pData, pCol->dataOff[row]); return POINTER_SHIFT(pCol->pData, pCol->dataOff[row]);
@ -361,31 +241,6 @@ bool isNEleNull(SDataCol *pCol, int nEle) {
return true; 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) { void *dataColSetOffset(SDataCol *pCol, int nEle) {
ASSERT(((pCol->type == TSDB_DATA_TYPE_BINARY) || (pCol->type == TSDB_DATA_TYPE_NCHAR))); ASSERT(((pCol->type == TSDB_DATA_TYPE_BINARY) || (pCol->type == TSDB_DATA_TYPE_NCHAR)));
@ -483,42 +338,6 @@ SDataCols *tdFreeDataCols(SDataCols *pCols) {
return NULL; 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) { void tdResetDataCols(SDataCols *pCols) {
if (pCols != NULL) { if (pCols != NULL) {
pCols->numOfRows = 0; 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 tdKVRowDup(SKVRow row) {
SKVRow trow = taosMemoryMalloc(kvRowLen(row)); SKVRow trow = taosMemoryMalloc(kvRowLen(row));
@ -859,98 +504,3 @@ SKVRow tdGetKVRowFromBuilder(SKVRowBuilder *pBuilder) {
return row; 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

58
source/dnode/vnode/src/tsdb/tsdbCommit.c
View File

@ -211,7 +211,7 @@ int tsdbCommit(STsdb *pRepo) {
void tsdbGetRtnSnap(STsdb *pRepo, SRtn *pRtn) { void tsdbGetRtnSnap(STsdb *pRepo, SRtn *pRtn) {
STsdbKeepCfg *pCfg = REPO_KEEP_CFG(pRepo); STsdbKeepCfg *pCfg = REPO_KEEP_CFG(pRepo);
TSKEY minKey, midKey, maxKey, now; TSKEY minKey, midKey, maxKey, now;
now = taosGetTimestamp(pCfg->precision); now = taosGetTimestamp(pCfg->precision);
minKey = now - pCfg->keep2 * tsTickPerDay[pCfg->precision]; minKey = now - pCfg->keep2 * tsTickPerDay[pCfg->precision];
@ -1386,34 +1386,7 @@ static void tsdbLoadAndMergeFromCache(SDataCols *pDataCols, int *iter, SCommitIt
tSkipListIterNext(pCommitIter->pIter); tSkipListIterNext(pCommitIter->pIter);
} else { } else {
#if 0 if (lastKey != key1) {
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) {
lastKey = key1; lastKey = key1;
++pTarget->numOfRows; ++pTarget->numOfRows;
} }
@ -1484,29 +1457,4 @@ static bool tsdbCanAddSubBlock(SCommitH *pCommith, SBlock *pBlock, SMergeInfo *p
} }
return false; 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;
// }

32
source/dnode/vnode/src/tsdb/tsdbMemTable2.c
View File

@ -21,6 +21,8 @@ typedef struct SMemSkipList SMemSkipList;
typedef struct SMemSkipListNode SMemSkipListNode; typedef struct SMemSkipListNode SMemSkipListNode;
typedef struct SMemSkipListCurosr SMemSkipListCurosr; typedef struct SMemSkipListCurosr SMemSkipListCurosr;
#define SL_MAX_LEVEL 5
struct SMemTable { struct SMemTable {
STsdb *pTsdb; STsdb *pTsdb;
TSKEY minKey; TSKEY minKey;
@ -61,7 +63,7 @@ struct SMemData {
struct SMemSkipListCurosr { struct SMemSkipListCurosr {
SMemSkipList *pSl; SMemSkipList *pSl;
SMemSkipListNode *pNodeC; SMemSkipListNode *pNodes[SL_MAX_LEVEL];
}; };
typedef struct { typedef struct {
@ -70,8 +72,6 @@ typedef struct {
const STSRow *pRow; const STSRow *pRow;
} STsdbRow; } STsdbRow;
#define SL_MAX_LEVEL 15
#define HASH_BUCKET(SUID, UID, NBUCKET) (TABS((SUID) + (UID)) % (NBUCKET)) #define HASH_BUCKET(SUID, UID, NBUCKET) (TABS((SUID) + (UID)) % (NBUCKET))
#define SL_NODE_SIZE(l) (sizeof(SMemSkipListNode) + sizeof(SMemSkipListNode *) * (l)*2) #define SL_NODE_SIZE(l) (sizeof(SMemSkipListNode) + sizeof(SMemSkipListNode *) * (l)*2)
@ -93,6 +93,8 @@ static void tsdbMemSkipListCursorDestroy(SMemSkipListCurosr *pSlc);
static void tsdbMemSkipListCursorInit(SMemSkipListCurosr *pSlc, SMemSkipList *pSl); static void tsdbMemSkipListCursorInit(SMemSkipListCurosr *pSlc, SMemSkipList *pSl);
static void tsdbMemSkipListCursorPut(SMemSkipListCurosr *pSlc, SMemSkipListNode *pNode); static void tsdbMemSkipListCursorPut(SMemSkipListCurosr *pSlc, SMemSkipListNode *pNode);
static int32_t tsdbMemSkipListCursorMoveTo(SMemSkipListCurosr *pSlc, int64_t version, TSKEY ts, int32_t flags); 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 tsdbMemSkipListCursorMoveToNext(SMemSkipListCurosr *pSlc);
static int32_t tsdbMemSkipListCursorMoveToPrev(SMemSkipListCurosr *pSlc); static int32_t tsdbMemSkipListCursorMoveToPrev(SMemSkipListCurosr *pSlc);
static SMemSkipListNode *tsdbMemSkipListNodeCreate(SVBufPool *pPool, SMemSkipList *pSl, const STsdbRow *pTRow); static SMemSkipListNode *tsdbMemSkipListNodeCreate(SVBufPool *pPool, SMemSkipList *pSl, const STsdbRow *pTRow);
@ -326,14 +328,26 @@ static int32_t tsdbMemSkipListCursorMoveTo(SMemSkipListCurosr *pSlc, int64_t ver
return 0; return 0;
} }
static int32_t tsdbMemSkipListCursorMoveToFirst(SMemSkipListCurosr *pSlc) { static void tsdbMemSkipListCursorMoveToFirst(SMemSkipListCurosr *pSlc) {
// TODO SMemSkipList *pSl = pSlc->pSl;
return 0; SMemSkipListNode *pHead = SL_HEAD_NODE(pSl);
for (int8_t iLevel = 0; iLevel < pSl->maxLevel; iLevel++) {
pSlc->pNodes[iLevel] = pHead;
}
tsdbMemSkipListCursorMoveToNext(pSlc);
} }
static int32_t tsdbMemSkipListCursorMoveToLast(SMemSkipListCurosr *pSlc) { static void tsdbMemSkipListCursorMoveToLast(SMemSkipListCurosr *pSlc) {
// TODO SMemSkipList *pSl = pSlc->pSl;
return 0; 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) { static int32_t tsdbMemSkipListCursorMoveToNext(SMemSkipListCurosr *pSlc) {