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

Merge pull request #1338 from taosdata/feature/2.0tsdb 

refactor code
This commit is contained in:
slguan 2020-03-10 16:54:27 +08:00 committed by GitHub
parent 23bed109d5 f909434a3e
commit ff5355b596
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
20 changed files with 242 additions and 420 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/CMakeLists.txt
View File

@ -14,6 +14,6 @@ ADD_SUBDIRECTORY(kit)
ADD_SUBDIRECTORY(plugins)
ADD_SUBDIRECTORY(sdb)
ADD_SUBDIRECTORY(mnode)
# ADD_SUBDIRECTORY(vnode)
ADD_SUBDIRECTORY(vnode)
# ADD_SUBDIRECTORY(dnode)
#ADD_SUBDIRECTORY(connector/jdbc)

42
src/common/inc/dataformat.h
View File

@ -19,11 +19,51 @@
#include <stdlib.h>
#include <string.h>
// #include "schema.h"
#include "taosdef.h"
#ifdef __cplusplus
extern "C" {
#endif
// ----------------- TSDB COLUMN DEFINITION
typedef struct {
int8_t type; // Column type
int16_t colId; // column ID
int32_t bytes; // column bytes
int32_t offset; // point offset in a row data
} STColumn;
#define colType(col) ((col)->type)
#define colColId(col) ((col)->colId)
#define colBytes(col) ((col)->bytes)
#define colOffset(col) ((col)->offset)
#define colSetType(col, t) (colType(col) = (t))
#define colSetColId(col, id) (colColId(col) = (id))
#define colSetBytes(col, b) (colBytes(col) = (b))
#define colSetOffset(col, o) (colOffset(col) = (o))
STColumn *tdNewCol(int8_t type, int16_t colId, int16_t bytes);
void tdFreeCol(STColumn *pCol);
void tdColCpy(STColumn *dst, STColumn *src);
void tdSetCol(STColumn *pCol, int8_t type, int16_t colId, int32_t bytes);
// ----------------- TSDB SCHEMA DEFINITION
typedef struct {
int32_t numOfCols;
int32_t padding; // TODO: replace the padding for useful variable
STColumn columns[];
} STSchema;
#define schemaNCols(s) ((s)->numOfCols)
#define schemaColAt(s, i) ((s)->columns + i)
STSchema *tdNewSchema(int32_t nCols);
STSchema *tdDupSchema(STSchema *pSchema);
void tdFreeSchema(STSchema *pSchema);
void tdUpdateSchema(STSchema *pSchema);
// ----------------- Data row structure
/* A data row, the format is like below:

130
src/common/src/dataformat.c
View File

@ -1,5 +1,135 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "dataformat.h"
/**
* Create a new STColumn object
* ASSUMPTIONS: VALID PARAMETERS
*
* @param type column type
* @param colId column ID
* @param bytes maximum bytes the col taken
*
* @return a STColumn object on success
* NULL for failure
*/
STColumn *tdNewCol(int8_t type, int16_t colId, int16_t bytes) {
if (!isValidDataType(type, 0)) return NULL;
STColumn *pCol = (STColumn *)calloc(1, sizeof(STColumn));
if (pCol == NULL) return NULL;
colSetType(pCol, type);
colSetColId(pCol, colId);
colSetOffset(pCol, -1);
switch (type) {
case TSDB_DATA_TYPE_BINARY:
case TSDB_DATA_TYPE_NCHAR:
colSetBytes(pCol, bytes);
break;
default:
colSetBytes(pCol, TYPE_BYTES[type]);
break;
}
return pCol;
}
/**
* Free a STColumn object CREATED with tdNewCol
*/
void tdFreeCol(STColumn *pCol) {
if (pCol) free(pCol);
}
/**
* Copy from source to destinition
*/
void tdColCpy(STColumn *dst, STColumn *src) { memcpy((void *)dst, (void *)src, sizeof(STColumn)); }
/**
* Set the column
*/
void tdSetCol(STColumn *pCol, int8_t type, int16_t colId, int32_t bytes) {
colSetType(pCol, type);
colSetColId(pCol, colId);
switch (type)
{
case TSDB_DATA_TYPE_BINARY:
case TSDB_DATA_TYPE_NCHAR:
colSetBytes(pCol, bytes);
break;
default:
colSetBytes(pCol, TYPE_BYTES[type]);
break;
}
}
/**
* Create a SSchema object with nCols columns
* ASSUMPTIONS: VALID PARAMETERS
*
* @param nCols number of columns the schema has
*
* @return a STSchema object for success
* NULL for failure
*/
STSchema *tdNewSchema(int32_t nCols) {
int32_t size = sizeof(STSchema) + sizeof(STColumn) * nCols;
STSchema *pSchema = (STSchema *)calloc(1, size);
if (pSchema == NULL) return NULL;
pSchema->numOfCols = nCols;
return pSchema;
}
/**
* Duplicate the schema and return a new object
*/
STSchema *tdDupSchema(STSchema *pSchema) {
STSchema *tSchema = tdNewSchema(schemaNCols(pSchema));
if (tSchema == NULL) return NULL;
int32_t size = sizeof(STSchema) + sizeof(STColumn) * schemaNCols(pSchema);
memcpy((void *)tSchema, (void *)pSchema, size);
return tSchema;
}
/**
* Free the SSchema object created by tdNewSchema or tdDupSchema
*/
void tdFreeSchema(STSchema *pSchema) {
if (pSchema == NULL) free(pSchema);
}
/**
* Function to update each columns's offset field in the schema.
* ASSUMPTIONS: VALID PARAMETERS
*/
void tdUpdateSchema(STSchema *pSchema) {
STColumn *pCol = NULL;
int32_t offset = 0;
for (int i = 0; i < schemaNCols(pSchema); i++) {
pCol = schemaColAt(pSchema, i);
colSetOffset(pCol, offset);
offset += TYPE_BYTES[pCol->type];
}
}
/**
* Create a data row with maximum row length bytes.
*

15
src/common/src/ttypes.c
View File

@ -16,7 +16,20 @@
#include "taosdef.h"
#include "ttokendef.h"
// #include "tutil.h"
const int32_t TYPE_BYTES[11] = {
-1, // TSDB_DATA_TYPE_NULL
sizeof(int8_t), // TSDB_DATA_TYPE_BOOL
sizeof(int8_t), // TSDB_DATA_TYPE_TINYINT
sizeof(int16_t), // TSDB_DATA_TYPE_SMALLINT
sizeof(int32_t), // TSDB_DATA_TYPE_INT
sizeof(int64_t), // TSDB_DATA_TYPE_BIGINT
sizeof(float), // TSDB_DATA_TYPE_FLOAT
sizeof(double), // TSDB_DATA_TYPE_DOUBLE
-1, // TSDB_DATA_TYPE_BINARY
sizeof(TSKEY), // TSDB_DATA_TYPE_TIMESTAMP
-1 // TSDB_DATA_TYPE_NCHAR
};
tDataTypeDescriptor tDataTypeDesc[11] = {
{TSDB_DATA_TYPE_NULL, 6, 1, "NOTYPE"},

2
src/inc/taosdef.h
View File

@ -45,6 +45,8 @@ extern "C" {
#define TSDB_DATA_TYPE_NCHAR 10 // unicode string
// Bytes for each type.
extern const int32_t TYPE_BYTES[11];
// TODO: replace and remove code below
#define CHAR_BYTES sizeof(char)
#define SHORT_BYTES sizeof(short)
#define INT_BYTES sizeof(int)

4
src/vnode/CMakeLists.txt
View File

@ -1,7 +1,5 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
PROJECT(TDengine)
# ADD_SUBDIRECTORY(common)
ADD_SUBDIRECTORY(wal)
ADD_SUBDIRECTORY(tsdb)
# ENABLE_TESTING()
# ADD_SUBDIRECTORY(tests)

13
src/vnode/common/CMakeLists.txt
View File

@ -1,13 +0,0 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
PROJECT(TDengine)
IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
INCLUDE_DIRECTORIES(${TD_OS_DIR}/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/util/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/dnode/inc)
INCLUDE_DIRECTORIES(inc)
AUX_SOURCE_DIRECTORY(src SRC)
ADD_LIBRARY(common ${SRC})
ENDIF ()

68
src/vnode/common/inc/schema.h
View File

@ -1,68 +0,0 @@
#ifndef _TD_SCHEMA_H_
#define _TD_SCHEMA_H_
#include <stdint.h>
#include <string.h>
#include "type.h"
#ifdef __cplusplus
extern "C" {
#endif
// ---- Column definition and operations
typedef struct {
int8_t type; // Column type
int16_t colId; // column ID
int16_t bytes; // column bytes
int32_t offset; // point offset in a row data
} SColumn;
#define colType(col) ((col)->type)
#define colColId(col) ((col)->colId)
#define colBytes(col) ((col)->bytes)
#define colOffset(col) ((col)->offset)
#define colSetType(col, t) (colType(col) = (t))
#define colSetColId(col, id) (colColId(col) = (id))
#define colSetBytes(col, b) (colBytes(col) = (b))
#define colSetOffset(col, o) (colOffset(col) = (o))
SColumn *tdNewCol(int8_t type, int16_t colId, int16_t bytes);
void tdFreeCol(SColumn *pCol);
void tdColCpy(SColumn *dst, SColumn *src);
// ---- Schema definition and operations
typedef struct {
int32_t numOfCols;
int32_t padding; // TODO: replace the padding for useful variable
SColumn columns[];
} SSchema;
#define schemaNCols(s) ((s)->numOfCols)
#define schemaColAt(s, i) ((s)->columns + i)
SSchema *tdNewSchema(int32_t nCols);
SSchema *tdDupSchema(SSchema *pSchema);
void tdFreeSchema(SSchema *pSchema);
void tdUpdateSchema(SSchema *pSchema);
int32_t tdMaxRowDataBytes(SSchema *pSchema);
// ---- Inline schema definition and operations
/* Inline schema definition
* +---------+---------+---------+-----+---------+-----------+-----+-----------+
* | int32_t | | | | | | | |
* +---------+---------+---------+-----+---------+-----------+-----+-----------+
* | len | SSchema | SColumn | ... | SColumn | col1_name | ... | colN_name |
* +---------+---------+---------+-----+---------+-----------+-----+-----------+
*/
typedef char *SISchema;
// TODO: add operations on SISchema
#ifdef __cplusplus
}
#endif
#endif // _TD_SCHEMA_H_

37
src/vnode/common/inc/type.h
View File

@ -1,37 +0,0 @@
#ifndef _TD_TYPE_H_
#define _TD_TYPE_H_
#include <stdint.h>
typedef enum {
TD_DATATYPE_BOOL = 0,
TD_DATATYPE_TINYINT,
TD_DATATYPE_SMALLINT,
TD_DATATYPE_INT,
TD_DATATYPE_BIGINT,
TD_DATATYPE_FLOAT,
TD_DATATYPE_DOUBLE,
TD_DATATYPE_TIMESTAMP,
TD_DATATYPE_VARCHAR,
TD_DATATYPE_NCHAR,
TD_DATATYPE_BINARY
} td_datatype_t;
extern const int32_t rowDataLen[];
// TODO: finish below
#define TD_DATATYPE_BOOL_NULL
#define TD_DATATYPE_TINYINT_NULL
#define TD_DATATYPE_SMALLINT_NULL
#define TD_DATATYPE_INT_NULL
#define TD_DATATYPE_BIGINT_NULL
#define TD_DATATYPE_FLOAT_NULL
#define TD_DATATYPE_DOUBLE_NULL
#define TD_DATATYPE_TIMESTAMP_NULL
#define TD_DATATYPE_VARCHAR_NULL
#define TD_DATATYPE_NCHAR_NULL
#define TD_DATATYPE_BINARY_NULL
#define TD_IS_VALID_DATATYPE(type) (((type) > TD_DATA_TYPE_INVLD) && ((type) <= TD_DATATYPE_BINARY))
#endif // _TD_TYPE_H_

136
src/vnode/common/src/schema.c
View File

@ -1,136 +0,0 @@
#include <stdlib.h>
#include "schema.h"
const int32_t rowDataLen[] = {
sizeof(int8_t), // TD_DATATYPE_BOOL,
sizeof(int8_t), // TD_DATATYPE_TINYINT,
sizeof(int16_t), // TD_DATATYPE_SMALLINT,
sizeof(int32_t), // TD_DATATYPE_INT,
sizeof(int64_t), // TD_DATATYPE_BIGINT,
sizeof(float), // TD_DATATYPE_FLOAT,
sizeof(double), // TD_DATATYPE_DOUBLE,
sizeof(int64_t), // TD_DATATYPE_TIMESTAMP
sizeof(int32_t), // TD_DATATYPE_VARCHAR,
sizeof(int32_t), // TD_DATATYPE_NCHAR,
sizeof(int32_t) // TD_DATATYPE_BINARY
};
/**
* Create a new SColumn object
* ASSUMPTIONS: VALID PARAMETERS
*
* @param type column type
* @param colId column ID
* @param bytes maximum bytes the col taken
*
* @return a SColumn object on success
* NULL for failure
*/
SColumn *tdNewCol(int8_t type, int16_t colId, int16_t bytes) {
SColumn *pCol = (SColumn *)calloc(1, sizeof(SColumn));
if (pCol == NULL) return NULL;
colSetType(pCol, type);
colSetColId(pCol, colId);
switch (type) {
case TD_DATATYPE_VARCHAR:
case TD_DATATYPE_NCHAR:
case TD_DATATYPE_BINARY:
colSetBytes(pCol, bytes);
break;
default:
colSetBytes(pCol, rowDataLen[type]);
break;
}
return pCol;
}
/**
* Free a SColumn object CREATED with tdNewCol
*/
void tdFreeCol(SColumn *pCol) {
if (pCol) free(pCol);
}
void tdColCpy(SColumn *dst, SColumn *src) { memcpy((void *)dst, (void *)src, sizeof(SColumn)); }
/**
* Create a SSchema object with nCols columns
* ASSUMPTIONS: VALID PARAMETERS
*
* @param nCols number of columns the schema has
*
* @return a SSchema object for success
* NULL for failure
*/
SSchema *tdNewSchema(int32_t nCols) {
int32_t size = sizeof(SSchema) + sizeof(SColumn) * nCols;
SSchema *pSchema = (SSchema *)calloc(1, size);
if (pSchema == NULL) return NULL;
pSchema->numOfCols = nCols;
return pSchema;
}
/**
* Free the SSchema object created by tdNewSchema or tdDupSchema
*/
void tdFreeSchema(SSchema *pSchema) {
if (pSchema == NULL) free(pSchema);
}
SSchema *tdDupSchema(SSchema *pSchema) {
SSchema *tSchema = tdNewSchema(schemaNCols(pSchema));
if (tSchema == NULL) return NULL;
int32_t size = sizeof(SSchema) + sizeof(SColumn) * schemaNCols(pSchema);
memcpy((void *)tSchema, (void *)pSchema, size);
return tSchema;
}
/**
* Function to update each columns's offset field in the schema.
* ASSUMPTIONS: VALID PARAMETERS
*/
void tdUpdateSchema(SSchema *pSchema) {
SColumn *pCol = NULL;
int32_t offset = 0;
for (int i = 0; i < schemaNCols(pSchema); i++) {
pCol = schemaColAt(pSchema, i);
colSetOffset(pCol, offset);
offset += rowDataLen[pCol->type];
}
}
/**
* Get the maximum size of a row data with the schema
*/
int32_t tdMaxRowDataBytes(SSchema *pSchema) {
int32_t size = 0;
SColumn *pCol = NULL;
for (int i = 0; i < schemaNCols(pSchema); i++) {
pCol = schemaColAt(pSchema, i);
size += rowDataLen[pCol->type];
switch (pCol->type) {
case TD_DATATYPE_VARCHAR:
size += (pCol->bytes + 1); // TODO: remove literal here
break;
case TD_DATATYPE_NCHAR:
size += (pCol->bytes + 4); // TODO: check and remove literal here
break;
case TD_DATATYPE_BINARY:
size += pCol->bytes;
break;
default:
break;
}
}
return size;
}

42
src/vnode/common/src/vnodePeer.c_del
View File

@ -1,42 +0,0 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#define _DEFAULT_SOURCE
#include "vnode.h"
#include "vnodeStatus.h"
int vnodeInitPeer(int numOfThreads) { return 0; }
void vnodeCleanUpPeer(int vnode) {}
int vnodeForwardToPeer(SMeterObj *pObj, char *cont, int contLen, char action, int sversion) { return 0; }
int vnodeRecoverFromPeer(SVnodeObj *pVnode, int fileId) { return -TSDB_CODE_FILE_CORRUPTED; }
void vnodeCloseAllSyncFds(int vnode) {}
void vnodeBroadcastStatusToUnsyncedPeer(SVnodeObj *pVnode) {}
int vnodeOpenPeerVnode(int vnode) {
SVnodeObj *pVnode = vnodeList + vnode;
pVnode->vnodeStatus = (pVnode->cfg.replications > 1) ? TSDB_VN_STATUS_UNSYNCED : TSDB_VN_STATUS_MASTER;
dPrint("vid:%d, status:%s numOfPeers:%d", vnode, taosGetVnodeStatusStr(pVnode->vnodeStatus), pVnode->cfg.replications - 1);
vnodeUpdateStreamRole(pVnode);
return 0;
}
void vnodeClosePeerVnode(int vnode) {}
void vnodeConfigVPeers(int vnode, int numOfPeers, SVPeerDesc peerDesc[]) {}

40
src/vnode/common/tstr/inc/tstring.h
View File

@ -1,40 +0,0 @@
/* A dynamic string library
*/
#if !defined(_TD_TSTRING_H_)
#define _TD_TSTRING_H_
#include <stdint.h>
#include <string.h>
#define TD_TSTRING_INIT_SIZE 16
typedef char* tstring_t;
// The string header
typedef struct {
int32_t space; // Allocated data space
char data[];
} STStrHdr;
// Get the data length of the string
#define TSTRLEN(pstr) strlen((char *)pstr)
// Get the real allocated string length
#define TSTRSPACE(pstr) (*(int32_t *)((char *)pstr - sizeof(STStrHdr)))
// Get the available space
#define TSTAVAIL(pstr) (TSTRSPACE(pstr) - TSTRLEN(pstr))
// Create an empty tstring with default size
tstring_t tdNewTString();
// Create an empty tstring with size
tstring_t tdNewTStringWithSize(uint32_t size);
// Create a tstring with a init value
tstring_t tdNewTStringWithValue(char *value);
// Create a tstring with a init value & size
tstring_t tdNewTStringWithValueSize(char *value, uint32_t size);
tstring_t tstrcat(tstring_t dest, tstring_t src);
int32_t tstrcmp(tstring_t str1, tstring_t str2);
int32_t tstrncmp(tstring_t str1, tstring_t str2, int32_t n);
#endif // _TD_TSTRING_H_

2
src/vnode/tsdb/CMakeLists.txt
View File

@ -14,5 +14,5 @@ IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
TARGET_LINK_LIBRARIES(tsdb common tutil)
# Someone has no gtest directory, so comment it
#ADD_SUBDIRECTORY(tests)
ADD_SUBDIRECTORY(tests)
ENDIF ()

4
src/vnode/tsdb/inc/tsdb.h
View File

@ -19,8 +19,8 @@
#include <stdbool.h>
#include <stdint.h>
#include "taosdef.h"
#include "dataformat.h"
#include "schema.h"
#ifdef __cplusplus
extern "C" {
@ -83,7 +83,7 @@ typedef struct {
int64_t createdTime;
int32_t numOfCols; // number of columns. For table form super table, not includes the tag schema
SSchema *schema; // If numOfCols == schema_->numOfCols, it is a normal table, stableName = NULL
STSchema *schema; // If numOfCols == schema_->numOfCols, it is a normal table, stableName = NULL
// If numOfCols < schema->numOfCols, it is a table created from super table
// assert(numOfCols <= schema->numOfCols);

4
src/vnode/tsdb/inc/tsdbMeta.h
View File

@ -53,7 +53,7 @@ typedef struct STable {
// For TSDB_SUPER_TABLE, it is the schema including tags
// For TSDB_NTABLE, it is only the schema, not including tags
// For TSDB_STABLE, it is NULL
SSchema *pSchema;
STSchema *pSchema;
// Tag value for this table
// For TSDB_SUPER_TABLE and TSDB_NTABLE, it is NULL
@ -97,7 +97,7 @@ typedef struct {
#define TSDB_TABLE_CACHE_DATA(pTable) ((pTable)->content.pData)
#define TSDB_SUPER_TABLE_INDEX(pTable) ((pTable)->content.pIndex)
SSchema *tsdbGetTableSchema(STable *pTable);
STSchema *tsdbGetTableSchema(STable *pTable);
// ---- Operation on SMetaHandle
#define TSDB_NUM_OF_TABLES(pHandle) ((pHandle)->numOfTables)

5
src/vnode/tsdb/src/tsdbMain.c
View File

@ -261,14 +261,13 @@ STableInfo *tsdbGetTableInfo(tsdb_repo_t *pRepo, STableId tableId) {
// TODO: need to return the number of data inserted
int32_t tsdbInsertData(tsdb_repo_t *repo, SSubmitMsg *pMsg) {
STsdbRepo * pRepo = (STsdbRepo *)repo;
SSubmitBlock *pBlock = pMsg->data;
SSubmitBlock *pBlock = (SSubmitBlock *)pMsg->data;
for (int i = 0; i < pMsg->numOfTables; i++) { // Loop to deal with the submit message
if (tsdbInsertDataToTable(repo, pBlock) < 0) {
return -1;
}
pBlock = ((char *)pBlock) + sizeof(SSubmitBlock) + pBlock->len;
pBlock = (SSubmitBlock *)(((char *)pBlock) + sizeof(SSubmitBlock) + pBlock->len);
}
return 0;

63
src/vnode/tsdb/tests/tsdbTests.cpp
View File

@ -2,9 +2,10 @@
#include <stdlib.h>
#include "tsdb.h"
#include "dataformat.h"
#include "tsdbMeta.h"
TEST(TsdbTest, DISABLED_createTable) {
TEST(TsdbTest, createTable) {
STsdbMeta *pMeta = tsdbCreateMeta(100);
ASSERT_NE(pMeta, nullptr);
@ -14,7 +15,7 @@ TEST(TsdbTest, DISABLED_createTable) {
config.numOfCols = 5;
config.schema = tdNewSchema(config.numOfCols);
for (int i = 0; i < schemaNCols(config.schema); i++) {
SColumn *pCol = tdNewCol(TD_DATATYPE_BIGINT, i, 0);
STColumn *pCol = tdNewCol(TSDB_DATA_TYPE_BIGINT, i, 0);
tdColCpy(schemaColAt(config.schema, i), pCol);
tdFreeCol(pCol);
}
@ -40,11 +41,11 @@ TEST(TsdbTest, createRepo) {
config.tableId.uid = 98868728187539L;
config.numOfCols = 5;
config.schema = tdNewSchema(config.numOfCols);
SColumn *pCol = tdNewCol(TD_DATATYPE_TIMESTAMP, 0, 0);
STColumn *pCol = tdNewCol(TSDB_DATA_TYPE_TIMESTAMP, 0, 0);
tdColCpy(schemaColAt(config.schema, 0), pCol);
tdFreeCol(pCol);
for (int i = 1; i < schemaNCols(config.schema); i++) {
pCol = tdNewCol(TD_DATATYPE_BIGINT, i, 0);
pCol = tdNewCol(TSDB_DATA_TYPE_BIGINT, i, 0);
tdColCpy(schemaColAt(config.schema, i), pCol);
tdFreeCol(pCol);
}
@ -52,41 +53,41 @@ TEST(TsdbTest, createRepo) {
tsdbCreateTable(pRepo, &config);
// Write some data
int32_t size = sizeof(SSubmitMsg) + sizeof(SSubmitBlock) + tdMaxRowDataBytes(config.schema) * 10 + sizeof(int32_t);
// int32_t size = sizeof(SSubmitMsg) + sizeof(SSubmitBlock) + tdMaxRowDataBytes(config.schema) * 10 + sizeof(int32_t);
tdUpdateSchema(config.schema);
// tdUpdateSchema(config.schema);
SSubmitMsg *pMsg = (SSubmitMsg *)malloc(size);
pMsg->numOfTables = 1; // TODO: use api
// SSubmitMsg *pMsg = (SSubmitMsg *)malloc(size);
// pMsg->numOfTables = 1; // TODO: use api
SSubmitBlock *pBlock = (SSubmitBlock *)pMsg->data;
pBlock->tableId = {.uid = 98868728187539L, .tid = 0};
pBlock->sversion = 0;
pBlock->len = sizeof(SSubmitBlock);
// SSubmitBlock *pBlock = (SSubmitBlock *)pMsg->data;
// pBlock->tableId = {.uid = 98868728187539L, .tid = 0};
// pBlock->sversion = 0;
// pBlock->len = sizeof(SSubmitBlock);
SDataRows rows = pBlock->data;
dataRowsInit(rows);
// SDataRows rows = pBlock->data;
// dataRowsInit(rows);
SDataRow row = tdNewDataRow(tdMaxRowDataBytes(config.schema));
int64_t ttime = 1583508800000;
for (int i = 0; i < 10; i++) { // loop over rows
ttime += (10000 * i);
tdDataRowReset(row);
for (int j = 0; j < schemaNCols(config.schema); j++) {
if (j == 0) { // set time stamp
tdAppendColVal(row, (void *)(&ttime), schemaColAt(config.schema, j), 40);
} else { // set other fields
int32_t val = 10;
tdAppendColVal(row, (void *)(&val), schemaColAt(config.schema, j), 40);
}
}
// SDataRow row = tdNewDataRow(tdMaxRowDataBytes(config.schema));
// int64_t ttime = 1583508800000;
// for (int i = 0; i < 10; i++) { // loop over rows
// ttime += (10000 * i);
// tdDataRowReset(row);
// for (int j = 0; j < schemaNCols(config.schema); j++) {
// if (j == 0) { // set time stamp
// tdAppendColVal(row, (void *)(&ttime), schemaColAt(config.schema, j), 40);
// } else { // set other fields
// int32_t val = 10;
// tdAppendColVal(row, (void *)(&val), schemaColAt(config.schema, j), 40);
// }
// }
tdDataRowsAppendRow(rows, row);
}
// tdDataRowsAppendRow(rows, row);
// }
tsdbInsertData(pRepo, pMsg);
// tsdbInsertData(pRepo, pMsg);
tdFreeDataRow(row);
// tdFreeDataRow(row);
tdFreeSchema(config.schema);
tsdbDropRepo(pRepo);

4
src/vnode/wal/CMakeLists.txt
View File

@ -0,0 +1,4 @@
AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR}/src SRC)
ADD_LIBRARY(wal ${SRC})
TARGET_INCLUDE_DIRECTORIES(wal PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/inc)

0
src/vnode/wal/inc/vnodewal.h → src/vnode/wal/inc/vnodeWal.h
View File

49
src/vnode/common/inc/vnodePeer.h → src/vnode/wal/src/vnodeWal.c
View File

@ -12,45 +12,16 @@
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <stdlib.h>
#ifndef TDENGINE_VNODE_PEER_H
#define TDENGINE_VNODEPEER_H
#include "vnodeWal.h"
#ifdef __cplusplus
extern "C" {
#endif
typedef struct {
/* TODO */
} SWal;
#include <stdint.h>
#include <stdbool.h>
#include "taosdef.h"
/*
* Initialize the resources
*/
int32_t vnodeInitPeers(int numOfThreads);
/*
* Free the resources
*/
void vnodeCleanUpPeers();
/*
* Start a vnode synchronization process
*/
int32_t vnodeOpenPeer(int32_t vnode);
/*
* Update the peerinfo of vnode
*/
int32_t vnodeConfigPeer(SVpeerDescArray msg);
/*
* Close a vnode synchronization process
*/
void vnodeCleanUpPeer(int32_t vnode);
#ifdef __cplusplus
}
#endif
#endif // TDENGINE_VNODEPEER_H
walh *vnodeOpenWal(int vnode, uint8_t op) { return NULL; }
int vnodeCloseWal(walh *pWal) { return 0; }
int vnodeRenewWal(walh *pWal) { return 0; }
int vnodeWriteWal(walh *pWal, void *cont, int contLen) { return 0; }
int vnodeSyncWal(walh *pWal) { return 0; }