未验证 提交 b723eb2b 编写于 作者: C Cary Xu 提交者: GitHub

Merge branch 'develop' into enhance/TS-385

......@@ -129,6 +129,17 @@ mkdir debug && cd debug
cmake .. && cmake --build .
```
Note TDengine 2.3.0.0 and later use a component named 'blm3' to play http daemon role by default instead of the http daemon embedded in the early version of TDengine. The blm3 is programmed by go language. If you pull TDengine source code to the latest from an existing codebase, please execute 'git submodule update --init --recursive' to pull blm3 source code. Please install go language 1.14 or above for compiling blm3. If you meet difficulties regarding 'go mod', especially you are from China, you can use a proxy to solve the problem.
```
go env -w GO111MODULE=on
go env -w GOPROXY=https://goproxy.cn,direct
```
Or you can use the following command to choose to embed old httpd too.
```
cmake .. -DBUILD_HTTP=true
```
You can use Jemalloc as memory allocator instead of glibc:
```
apt install autoconf
......
......@@ -4,7 +4,7 @@ PROJECT(TDengine)
IF (DEFINED VERNUMBER)
SET(TD_VER_NUMBER ${VERNUMBER})
ELSE ()
SET(TD_VER_NUMBER "2.2.0.5")
SET(TD_VER_NUMBER "2.3.0.0")
ENDIF ()
IF (DEFINED VERCOMPATIBLE)
......
......@@ -1280,7 +1280,7 @@ TDengine支持针对数据的聚合查询。提供支持的聚合和选择函数
适用于:**表、超级表**。
说明:*P*值有效取值范围0≤P≤100,为 0 的时候等同于 MIN,为 100 的时候等同于MAX;*algo_type*的有效输入:**default** 和 **t-digest**。 用于指定计算近似分位数的算法。可不提供第三个参数的输入,此时将使用 default 的算法进行计算,即 apercentile(column_name, 50, "default") 与 apercentile(column_name, 50) 等价。当使用“t-digest”参数的时候,将使用t-digest方式采样计算近似分位数。但该参数指定计算算法的功能从2.2.0.x版本开始支持,2.2.0.0之前的版本不支持指定使用算法的功能。
说明:<br/>**P**值有效取值范围0≤P≤100,为 0 的时候等同于 MIN,为 100 的时候等同于MAX;<br/>**algo_type**的有效输入:**default** 和 **t-digest**。 用于指定计算近似分位数的算法。可不提供第三个参数的输入,此时将使用 default 的算法进行计算,即 apercentile(column_name, 50, "default") 与 apercentile(column_name, 50) 等价。当使用“t-digest”参数的时候,将使用t-digest方式采样计算近似分位数。但该参数指定计算算法的功能从2.2.0.x版本开始支持,2.2.0.0之前的版本不支持指定使用算法的功能。<br/>
嵌套子查询支持:适用于内层查询和外层查询。
......@@ -1319,76 +1319,72 @@ TDengine支持针对数据的聚合查询。提供支持的聚合和选择函数
限制:LAST_ROW() 不能与 INTERVAL 一起使用。
说明:在用于超级表时,时间戳完全一样且同为最大的数据行可能有多个,那么会从中随机返回一条,而并不保证多次运行所挑选的数据行必然一致。
示例:
```mysql
说明:在用于超级表时,时间戳完全一样且同为最大的数据行可能有多个,那么会从中随机返回一条,而并不保证多次运行所挑选的数据行必然一致。<br/>
<br/>示例:
```mysql
taos> SELECT LAST_ROW(current) FROM meters;
last_row(current) |
=======================
12.30000 |
Query OK, 1 row(s) in set (0.001238s)
Query OK, 1 row(s) in set (0.001238s)
taos> SELECT LAST_ROW(current) FROM d1002;
last_row(current) |
=======================
10.30000 |
Query OK, 1 row(s) in set (0.001042s)
```
```
- **INTERP**
- **INTERP**
```mysql
SELECT INTERP(field_name) FROM { tb_name | stb_name } WHERE ts='timestamp' [FILL ({ VALUE | PREV | NULL | LINEAR | NEXT})];
```
功能说明:返回表/超级表的指定时间截面、指定字段的记录。
返回结果数据类型:同字段类型。
应用字段:数值型字段。
适用于:**表、超级表**。
说明:(从 2.0.15.0 版本开始新增此函数)
功能说明:返回表/超级表的指定时间截面、指定字段的记录。
1)INTERP 必须指定时间断面,如果该时间断面不存在直接对应的数据,那么会根据 FILL 参数的设定进行插值。此外,条件语句里面可附带筛选条件,例如标签、tbname
返回结果数据类型:同字段类型
2)INTERP 查询要求查询的时间区间必须位于数据集合(表)的所有记录的时间范围之内。如果给定的时间戳位于时间范围之外,即使有插值指令,仍然不返回结果
应用字段:数值型字段
3)单个 INTERP 函数查询只能够针对一个时间点进行查询,如果需要返回等时间间隔的断面数据,可以通过 INTERP 配合 EVERY 的方式来进行查询处理(而不是使用 INTERVAL),其含义是每隔固定长度的时间进行插值
适用于:**表、超级表**
说明:(从 2.0.15.0 版本开始新增此函数) <br/>1)INTERP 必须指定时间断面,如果该时间断面不存在直接对应的数据,那么会根据 FILL 参数的设定进行插值。此外,条件语句里面可附带筛选条件,例如标签、tbname。<br/>2)INTERP 查询要求查询的时间区间必须位于数据集合(表)的所有记录的时间范围之内。如果给定的时间戳位于时间范围之外,即使有插值指令,仍然不返回结果。<br/>3)单个 INTERP 函数查询只能够针对一个时间点进行查询,如果需要返回等时间间隔的断面数据,可以通过 INTERP 配合 EVERY 的方式来进行查询处理(而不是使用 INTERVAL),其含义是每隔固定长度的时间进行插值。<br/>
示例:
```sql
```mysql
taos> SELECT INTERP(*) FROM meters WHERE ts='2017-7-14 18:40:00.004';
interp(ts) | interp(current) | interp(voltage) | interp(phase) |
==========================================================================================
2017-07-14 18:40:00.004 | 9.84020 | 216 | 0.32222 |
Query OK, 1 row(s) in set (0.002652s)
```
如果给定的时间戳无对应的数据,在不指定插值生成策略的情况下,不会返回结果,如果指定了插值策略,会根据插值策略返回结果。
```sql
```
如果给定的时间戳无对应的数据,在不指定插值生成策略的情况下,不会返回结果,如果指定了插值策略,会根据插值策略返回结果。
```mysql
taos> SELECT INTERP(*) FROM meters WHERE tbname IN ('d636') AND ts='2017-7-14 18:40:00.005';
Query OK, 0 row(s) in set (0.004022s)
taos> SELECT INTERP(*) FROM meters WHERE tbname IN ('d636') AND ts='2017-7-14 18:40:00.005' FILL(PREV);;
taos> SELECT INTERP(*) FROM meters WHERE tbname IN ('d636') AND ts='2017-7-14 18:40:00.005' FILL(PREV);
interp(ts) | interp(current) | interp(voltage) | interp(phase) |
==========================================================================================
2017-07-14 18:40:00.005 | 9.88150 | 217 | 0.32500 |
Query OK, 1 row(s) in set (0.003056s)
```
```
如下所示代码表示在时间区间 `['2017-7-14 18:40:00', '2017-7-14 18:40:00.014']` 中每隔 5 毫秒 进行一次断面计算。
如下所示代码表示在时间区间 `['2017-7-14 18:40:00', '2017-7-14 18:40:00.014']` 中每隔 5 毫秒 进行一次断面计算。
```sql
```mysql
taos> SELECT INTERP(current) FROM d636 WHERE ts>='2017-7-14 18:40:00' AND ts<='2017-7-14 18:40:00.014' EVERY(5a);
ts | interp(current) |
=================================================
2017-07-14 18:40:00.000 | 10.04179 |
2017-07-14 18:40:00.010 | 10.16123 |
Query OK, 2 row(s) in set (0.003487s)
```
```
### 计算函数
......
......@@ -213,7 +213,7 @@ else
exit 1
fi
make
make -j8
cd ${curr_dir}
......
......@@ -254,7 +254,10 @@ function install_lib() {
fi
if [ "$osType" != "Darwin" ]; then
${csudo} cp ${binary_dir}/build/lib/libtaos.so.${verNumber} ${install_main_dir}/driver && ${csudo} chmod 777 ${install_main_dir}/driver/*
${csudo} cp ${binary_dir}/build/lib/libtaos.so.${verNumber} \
${install_main_dir}/driver \
&& ${csudo} chmod 777 ${install_main_dir}/driver/*
${csudo} ln -sf ${install_main_dir}/driver/libtaos.* ${lib_link_dir}/libtaos.so.1
${csudo} ln -sf ${lib_link_dir}/libtaos.so.1 ${lib_link_dir}/libtaos.so
......@@ -263,11 +266,28 @@ function install_lib() {
${csudo} ln -sf ${lib64_link_dir}/libtaos.so.1 ${lib64_link_dir}/libtaos.so
fi
else
${csudo} cp -Rf ${binary_dir}/build/lib/libtaos.${verNumber}.dylib ${install_main_dir}/driver || ${csudo} cp -Rf ${binary_dir}/build/lib/libtaos.${verNumber}.dylib ${install_main_2_dir}/driver && ${csudo} chmod 777 ${install_main_dir}/driver/* || ${csudo} chmod 777 ${install_main_2_dir}/driver/*
${csudo} cp -Rf ${binary_dir}/build/lib/libtaos.${verNumber}.dylib \
${install_main_dir}/driver \
|| ${csudo} cp -Rf ${binary_dir}/build/lib/libtaos.${verNumber}.dylib \
${install_main_2_dir}/driver \
&& ${csudo} chmod 777 ${install_main_dir}/driver/* \
|| ${csudo} chmod 777 ${install_main_2_dir}/driver/*
${csudo} ln -sf ${install_main_dir}/driver/libtaos.* \
${install_main_dir}/driver/libtaos.1.dylib \
|| ${csudo} ln -sf ${install_main_2_dir}/driver/libtaos.* \
${install_main_2_dir}/driver/libtaos.1.dylib || :
${csudo} ln -sf ${install_main_dir}/driver/libtaos.1.dylib \
${install_main_dir}/driver/libtaos.dylib \
|| ${csudo} ln -sf ${install_main_2_dir}/driver/libtaos.1.dylib \
${install_main_2_dir}/driver/libtaos.dylib || :
${csudo} ln -sf ${install_main_dir}/driver/libtaos.${verNumber}.dylib \
${lib_link_dir}/libtaos.1.dylib \
|| ${csudo} ln -sf ${install_main_2_dir}/driver/libtaos.${verNumber}.dylib \
${lib_link_dir}/libtaos.1.dylib || :
${csudo} ln -sf ${install_main_dir}/driver/libtaos.* ${install_main_dir}/driver/libtaos.1.dylib || ${csudo} ln -sf ${install_main_2_dir}/driver/libtaos.* ${install_main_2_dir}/driver/libtaos.1.dylib || :
${csudo} ln -sf ${install_main_dir}/driver/libtaos.1.dylib ${install_main_dir}/driver/libtaos.dylib || ${csudo} ln -sf ${install_main_2_dir}/driver/libtaos.1.dylib ${install_main_2_dir}/driver/libtaos.dylib || :
${csudo} ln -sf ${install_main_dir}/driver/libtaos.${verNumber}.dylib ${lib_link_dir}/libtaos.1.dylib || ${csudo} ln -sf ${install_main_2_dir}/driver/libtaos.${verNumber}.dylib ${lib_link_dir}/libtaos.1.dylib || :
${csudo} ln -sf ${lib_link_dir}/libtaos.1.dylib ${lib_link_dir}/libtaos.dylib || :
fi
......@@ -282,11 +302,17 @@ function install_header() {
if [ "$osType" != "Darwin" ]; then
${csudo} rm -f ${inc_link_dir}/taos.h ${inc_link_dir}/taoserror.h || :
${csudo} cp -f ${source_dir}/src/inc/taos.h ${source_dir}/src/inc/taoserror.h ${install_main_dir}/include && ${csudo} chmod 644 ${install_main_dir}/include/*
${csudo} cp -f ${source_dir}/src/inc/taos.h ${source_dir}/src/inc/taoserror.h \
${install_main_dir}/include && ${csudo} chmod 644 ${install_main_dir}/include/*
${csudo} ln -s ${install_main_dir}/include/taos.h ${inc_link_dir}/taos.h
${csudo} ln -s ${install_main_dir}/include/taoserror.h ${inc_link_dir}/taoserror.h
else
${csudo} cp -f ${source_dir}/src/inc/taos.h ${source_dir}/src/inc/taoserror.h ${install_main_dir}/include || ${csudo} cp -f ${source_dir}/src/inc/taos.h ${source_dir}/src/inc/taoserror.h ${install_main_2_dir}/include && ${csudo} chmod 644 ${install_main_dir}/include/* || ${csudo} chmod 644 ${install_main_2_dir}/include/*
${csudo} cp -f ${source_dir}/src/inc/taos.h ${source_dir}/src/inc/taoserror.h \
${install_main_dir}/include \
|| ${csudo} cp -f ${source_dir}/src/inc/taos.h ${source_dir}/src/inc/taoserror.h \
${install_main_2_dir}/include \
&& ${csudo} chmod 644 ${install_main_dir}/include/* \
|| ${csudo} chmod 644 ${install_main_2_dir}/include/*
fi
}
......@@ -301,7 +327,12 @@ function install_config() {
${csudo} cp -f ${script_dir}/../cfg/taos.cfg ${install_main_dir}/cfg/taos.cfg.org
${csudo} ln -s ${cfg_install_dir}/taos.cfg ${install_main_dir}/cfg/taos.cfg
else
${csudo} cp -f ${script_dir}/../cfg/taos.cfg ${install_main_dir}/cfg/taos.cfg.org || ${csudo} cp -f ${script_dir}/../cfg/taos.cfg ${install_main_2_dir}/cfg/taos.cfg.org
if [ "$osType" != "Darwin" ]; then
${csudo} cp -f ${script_dir}/../cfg/taos.cfg ${install_main_dir}/cfg/taos.cfg.org
else
${csudo} cp -f ${script_dir}/../cfg/taos.cfg ${install_main_dir}/cfg/taos.cfg.org\
|| ${csudo} cp -f ${script_dir}/../cfg/taos.cfg ${install_main_2_dir}/cfg/taos.cfg.org
fi
fi
}
......@@ -312,14 +343,21 @@ function install_blm3_config() {
${csudo} cp ${binary_dir}/test/cfg/blm.toml ${cfg_install_dir}
[ -f ${cfg_install_dir}/blm.toml ] &&
${csudo} chmod 644 ${cfg_install_dir}/blm.toml
[ -f ${binary_dir}/test/cfg//blm.toml ] &&
[ -f ${binary_dir}/test/cfg/blm.toml ] &&
${csudo} cp -f ${binary_dir}/test/cfg/blm.toml ${install_main_dir}/cfg/blm.toml.org
[ -f ${cfg_install_dir}/blm.toml ] &&
${csudo} ln -s ${cfg_install_dir}/blm.toml ${install_main_dir}/cfg/blm.toml
else
[ -f ${binary_dir}/test/cfg//blm.toml ] &&
${csudo} cp -f ${binary_dir}/test/cfg/blm.toml ${install_main_dir}/cfg/blm.toml.org \
|| ${csudo} cp -f ${binary_dir}/test/cfg/blm.toml ${install_main_2_dir}/cfg/blm.toml.org
if [ -f "${binary_dir}/test/cfg/blm.toml" ]; then
if [ "$osType" != "Darwin" ]; then
${csudo} cp -f ${binary_dir}/test/cfg/blm.toml \
${install_main_dir}/cfg/blm.toml.org
else
${csudo} cp -f ${binary_dir}/test/cfg/blm.toml ${install_main_dir}/cfg/blm.toml.org \
|| ${csudo} cp -f ${binary_dir}/test/cfg/blm.toml \
${install_main_2_dir}/cfg/blm.toml.org
fi
fi
fi
}
......
name: tdengine
base: core18
version: '2.2.0.5'
version: '2.3.0.0'
icon: snap/gui/t-dengine.svg
summary: an open-source big data platform designed and optimized for IoT.
description: |
......@@ -72,7 +72,7 @@ parts:
- usr/bin/taosd
- usr/bin/taos
- usr/bin/taosdemo
- usr/lib/libtaos.so.2.2.0.5
- usr/lib/libtaos.so.2.3.0.0
- usr/lib/libtaos.so.1
- usr/lib/libtaos.so
......
......@@ -2924,7 +2924,9 @@ int32_t getMultiTableMetaFromMnode(SSqlObj *pSql, SArray* pNameList, SArray* pVg
}
int32_t tscGetTableMetaImpl(SSqlObj* pSql, STableMetaInfo *pTableMetaInfo, bool autocreate, bool onlyLocal) {
assert(tIsValidName(&pTableMetaInfo->name));
if (!tIsValidName(&pTableMetaInfo->name)) {
return TSDB_CODE_TSC_APP_ERROR;
}
char name[TSDB_TABLE_FNAME_LEN] = {0};
tNameExtractFullName(&pTableMetaInfo->name, name);
......
......@@ -110,6 +110,7 @@ extern int8_t tsCacheLastRow;
//tsdb
extern bool tsdbForceKeepFile;
extern bool tsdbForceCompactFile;
extern int32_t tsdbWalFlushSize;
// balance
......
......@@ -156,6 +156,7 @@ int32_t tsTsdbMetaCompactRatio = TSDB_META_COMPACT_RATIO;
// tsdb config
// For backward compatibility
bool tsdbForceKeepFile = false;
bool tsdbForceCompactFile = false; // compact TSDB fileset forcibly
int32_t tsdbWalFlushSize = TSDB_DEFAULT_WAL_FLUSH_SIZE; // MB
// balance
......
......@@ -392,7 +392,7 @@ public abstract class AbstractConnection extends WrapperImpl implements Connecti
//true if the connection is valid, false otherwise
if (isClosed())
return false;
if (timeout < 0) //SQLException - if the value supplied for timeout is less then 0
if (timeout < 0) //SQLException - if the value supplied for timeout is less than 0
throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_INVALID_VARIABLE);
ExecutorService executor = Executors.newCachedThreadPool();
......
......@@ -42,6 +42,8 @@ int32_t main(int32_t argc, char *argv[]) {
}
} else if (strcmp(argv[i], "-C") == 0) {
dump_config = 1;
} else if (strcmp(argv[i], "--force-compact-file") == 0) {
tsdbForceCompactFile = true;
} else if (strcmp(argv[i], "--force-keep-file") == 0) {
tsdbForceKeepFile = true;
} else if (strcmp(argv[i], "--compact-mnode-wal") == 0) {
......
......@@ -457,6 +457,11 @@ typedef enum {
TD_ROW_PARTIAL_UPDATE = 2
} TDUpdateConfig;
typedef enum {
TSDB_STATIS_OK = 0, // statis part exist and load successfully
TSDB_STATIS_NONE = 1, // statis part not exist
} ETsdbStatisStatus;
extern char *qtypeStr[];
#ifdef __cplusplus
......
......@@ -271,6 +271,7 @@ int32_t* taosGetErrno();
#define TSDB_CODE_TDB_MESSED_MSG TAOS_DEF_ERROR_CODE(0, 0x0614) //"TSDB messed message")
#define TSDB_CODE_TDB_IVLD_TAG_VAL TAOS_DEF_ERROR_CODE(0, 0x0615) //"TSDB invalid tag value")
#define TSDB_CODE_TDB_NO_CACHE_LAST_ROW TAOS_DEF_ERROR_CODE(0, 0x0616) //"TSDB no cache last row data")
#define TSDB_CODE_TDB_INCOMPLETE_DFILESET TAOS_DEF_ERROR_CODE(0, 0x0617) //"TSDB incomplete DFileSet")
// query
#define TSDB_CODE_QRY_INVALID_QHANDLE TAOS_DEF_ERROR_CODE(0, 0x0700) //"Invalid handle")
......@@ -286,8 +287,8 @@ int32_t* taosGetErrno();
#define TSDB_CODE_QRY_TOO_MANY_TIMEWINDOW TAOS_DEF_ERROR_CODE(0, 0x070A) //"Too many time window in query")
#define TSDB_CODE_QRY_NOT_ENOUGH_BUFFER TAOS_DEF_ERROR_CODE(0, 0x070B) //"Query buffer limit has reached")
#define TSDB_CODE_QRY_INCONSISTAN TAOS_DEF_ERROR_CODE(0, 0x070C) //"File inconsistency in replica")
#define TSDB_CODE_QRY_INVALID_TIME_CONDITION TAOS_DEF_ERROR_CODE(0, 0x070D) //"invalid time condition")
#define TSDB_CODE_QRY_SYS_ERROR TAOS_DEF_ERROR_CODE(0, 0x070E) //"System error")
#define TSDB_CODE_QRY_SYS_ERROR TAOS_DEF_ERROR_CODE(0, 0x070D) //"System error")
#define TSDB_CODE_QRY_INVALID_TIME_CONDITION TAOS_DEF_ERROR_CODE(0, 0x070E) //"invalid time condition")
// grant
......
......@@ -225,6 +225,7 @@ typedef struct SArguments_S {
char * database;
int replica;
char * tb_prefix;
bool escapeChar;
char * sqlFile;
bool use_metric;
bool drop_database;
......@@ -298,6 +299,7 @@ typedef struct SSuperTable_S {
StrColumn tags[TSDB_MAX_TAGS];
char* childTblName;
bool escapeChar;
char* colsOfCreateChildTable;
uint64_t lenOfOneRow;
uint64_t lenOfTagOfOneRow;
......@@ -502,6 +504,7 @@ typedef struct SThreadInfo_S {
uint64_t querySeq; // sequence number of sql command
TAOS_SUB* tsub;
int sockfd;
} threadInfo;
#ifdef WINDOWS
......@@ -581,8 +584,7 @@ static void prompt();
static int createDatabasesAndStables();
static void createChildTables();
static int queryDbExec(TAOS *taos, char *command, QUERY_TYPE type, bool quiet);
static int postProceSql(char *host, struct sockaddr_in *pServAddr,
uint16_t port, char* sqlstr, threadInfo *pThreadInfo);
static int postProceSql(char *host, uint16_t port, char* sqlstr, threadInfo *pThreadInfo);
static int64_t getTSRandTail(int64_t timeStampStep, int32_t seq,
int disorderRatio, int disorderRange);
static bool getInfoFromJsonFile(char* file);
......@@ -638,6 +640,7 @@ SArguments g_args = {
"test", // database
1, // replica
"d", // tb_prefix
false, // escapeChar
NULL, // sqlFile
true, // use_metric
true, // drop_database
......@@ -1777,6 +1780,9 @@ static void parse_args(int argc, char *argv[], SArguments *arguments) {
errorUnrecognized(argv[0], argv[i]);
exit(EXIT_FAILURE);
}
} else if ((0 == strncmp(argv[i], "-E", strlen("-E")))
|| (0 == strncmp(argv[i], "--escape-character", strlen("--escape-character")))) {
arguments->escapeChar = true;
} else if ((strcmp(argv[i], "-N") == 0)
|| (0 == strcmp(argv[i], "--normal-table"))) {
arguments->demo_mode = false;
......@@ -2212,7 +2218,7 @@ static void selectAndGetResult(
} else if (0 == strncasecmp(g_queryInfo.queryMode, "rest", strlen("rest"))) {
int retCode = postProceSql(
g_queryInfo.host, &(g_queryInfo.serv_addr), g_queryInfo.port,
g_queryInfo.host, g_queryInfo.port,
command,
pThreadInfo);
if (0 != retCode) {
......@@ -3391,7 +3397,7 @@ static void printfQuerySystemInfo(TAOS * taos) {
free(dbInfos);
}
static int postProceSql(char *host, struct sockaddr_in *pServAddr, uint16_t port,
static int postProceSql(char *host, uint16_t port,
char* sqlstr, threadInfo *pThreadInfo)
{
char *req_fmt = "POST %s HTTP/1.1\r\nHost: %s:%d\r\nAccept: */*\r\nAuthorization: Basic %s\r\nContent-Length: %d\r\nContent-Type: application/x-www-form-urlencoded\r\n\r\n%s";
......@@ -3429,29 +3435,6 @@ static int postProceSql(char *host, struct sockaddr_in *pServAddr, uint16_t port
size_t encoded_len = 4 * ((userpass_buf_len +2) / 3);
char base64_buf[INPUT_BUF_LEN];
#ifdef WINDOWS
WSADATA wsaData;
WSAStartup(MAKEWORD(2, 2), &wsaData);
SOCKET sockfd;
#else
int sockfd;
#endif
sockfd = socket(AF_INET, SOCK_STREAM, 0);
if (sockfd < 0) {
#ifdef WINDOWS
errorPrint( "Could not create socket : %d" , WSAGetLastError());
#endif
debugPrint("%s() LN%d, sockfd=%d\n", __func__, __LINE__, sockfd);
free(request_buf);
ERROR_EXIT("opening socket");
}
int retConn = connect(sockfd, (struct sockaddr *)pServAddr, sizeof(struct sockaddr));
debugPrint("%s() LN%d connect() return %d\n", __func__, __LINE__, retConn);
if (retConn < 0) {
free(request_buf);
ERROR_EXIT("connecting");
}
memset(base64_buf, 0, INPUT_BUF_LEN);
......@@ -3491,9 +3474,9 @@ static int postProceSql(char *host, struct sockaddr_in *pServAddr, uint16_t port
sent = 0;
do {
#ifdef WINDOWS
bytes = send(sockfd, request_buf + sent, req_str_len - sent, 0);
bytes = send(pThreadInfo->sockfd, request_buf + sent, req_str_len - sent, 0);
#else
bytes = write(sockfd, request_buf + sent, req_str_len - sent);
bytes = write(pThreadInfo->sockfd, request_buf + sent, req_str_len - sent);
#endif
if (bytes < 0)
ERROR_EXIT("writing message to socket");
......@@ -3505,12 +3488,18 @@ static int postProceSql(char *host, struct sockaddr_in *pServAddr, uint16_t port
memset(response_buf, 0, RESP_BUF_LEN);
resp_len = sizeof(response_buf) - 1;
received = 0;
char resEncodingChunk[] = "Encoding: chunked";
char resHttp[] = "HTTP/1.1 ";
char resHttpOk[] = "HTTP/1.1 200 OK";
do {
#ifdef WINDOWS
bytes = recv(sockfd, response_buf + received, resp_len - received, 0);
bytes = recv(pThreadInfo->sockfds, response_buf + received, resp_len - received, 0);
#else
bytes = read(sockfd, response_buf + received, resp_len - received);
bytes = read(pThreadInfo->sockfd, response_buf + received, resp_len - received);
#endif
verbosePrint("%s() LN%d: bytes:%d\n", __func__, __LINE__, bytes);
if (bytes < 0) {
free(request_buf);
ERROR_EXIT("reading response from socket");
......@@ -3518,6 +3507,19 @@ static int postProceSql(char *host, struct sockaddr_in *pServAddr, uint16_t port
if (bytes == 0)
break;
received += bytes;
verbosePrint("%s() LN%d: received:%d resp_len:%d, response_buf:\n%s\n",
__func__, __LINE__, received, resp_len, response_buf);
if (((NULL != strstr(response_buf, resEncodingChunk))
&& (NULL != strstr(response_buf, resHttp)))
|| ((NULL != strstr(response_buf, resHttpOk))
&& (NULL != strstr(response_buf, "\"status\":")))) {
debugPrint(
"%s() LN%d: received:%d resp_len:%d, response_buf:\n%s\n",
__func__, __LINE__, received, resp_len, response_buf);
break;
}
} while(received < resp_len);
if (received == resp_len) {
......@@ -3526,20 +3528,18 @@ static int postProceSql(char *host, struct sockaddr_in *pServAddr, uint16_t port
}
response_buf[RESP_BUF_LEN - 1] = '\0';
printf("Response:\n%s\n", response_buf);
if (strlen(pThreadInfo->filePath) > 0) {
appendResultBufToFile(response_buf, pThreadInfo);
}
free(request_buf);
#ifdef WINDOWS
closesocket(sockfd);
WSACleanup();
#else
close(sockfd);
#endif
if (NULL == strstr(response_buf, resHttpOk)) {
errorPrint("%s() LN%d, Response:\n%s\n",
__func__, __LINE__, response_buf);
return -1;
}
return 0;
}
......@@ -4513,6 +4513,8 @@ static void* createTable(void *sarg)
i <= pThreadInfo->end_table_to; i++) {
if (0 == g_Dbs.use_metric) {
snprintf(pThreadInfo->buffer, buff_len,
g_args.escapeChar ?
"CREATE TABLE IF NOT EXISTS %s.`%s%"PRIu64"` %s;" :
"CREATE TABLE IF NOT EXISTS %s.%s%"PRIu64" %s;",
pThreadInfo->db_name,
g_args.tb_prefix, i,
......@@ -4550,7 +4552,8 @@ static void* createTable(void *sarg)
ERROR_EXIT("use metric, but tag buffer is NULL\n");
}
len += snprintf(pThreadInfo->buffer + len,
buff_len - len,
buff_len - len, stbInfo->escapeChar ?
"if not exists %s.`%s%"PRIu64"` using %s.`%s` tags %s " :
"if not exists %s.%s%"PRIu64" using %s.%s tags %s ",
pThreadInfo->db_name, stbInfo->childTblPrefix,
i, pThreadInfo->db_name,
......@@ -4574,7 +4577,6 @@ static void* createTable(void *sarg)
return NULL;
}
pThreadInfo->tables_created += batchNum;
uint64_t currentPrintTime = taosGetTimestampMs();
if (currentPrintTime - lastPrintTime > 30*1000) {
printf("thread[%d] already create %"PRIu64" - %"PRIu64" tables\n",
......@@ -4588,8 +4590,8 @@ static void* createTable(void *sarg)
NO_INSERT_TYPE, false)) {
errorPrint2("queryDbExec() failed. buffer:\n%s\n", pThreadInfo->buffer);
}
pThreadInfo->tables_created += batchNum;
}
free(pThreadInfo->buffer);
return NULL;
}
......@@ -5531,6 +5533,24 @@ static bool getMetaFromInsertJsonFile(cJSON* root) {
tstrncpy(g_Dbs.db[i].superTbls[j].childTblPrefix, prefix->valuestring,
TBNAME_PREFIX_LEN);
cJSON *escapeChar = cJSON_GetObjectItem(stbInfo, "escape_character");
if (escapeChar
&& escapeChar->type == cJSON_String
&& escapeChar->valuestring != NULL) {
if ((0 == strncasecmp(escapeChar->valuestring, "yes", 3))) {
g_Dbs.db[i].superTbls[j].escapeChar = true;
} else if (0 == strncasecmp(escapeChar->valuestring, "no", 2)) {
g_Dbs.db[i].superTbls[j].escapeChar = false;
} else {
g_Dbs.db[i].superTbls[j].escapeChar = false;
}
} else if (!escapeChar) {
g_Dbs.db[i].superTbls[j].escapeChar = false;
} else {
errorPrint("%s", "failed to read json, escape_character not found\n");
goto PARSE_OVER;
}
cJSON *autoCreateTbl = cJSON_GetObjectItem(stbInfo, "auto_create_table");
if (autoCreateTbl
&& autoCreateTbl->type == cJSON_String
......@@ -6982,7 +7002,7 @@ static int32_t execInsert(threadInfo *pThreadInfo, uint32_t k)
verbosePrint("[%d] %s() LN%d %s\n", pThreadInfo->threadID,
__func__, __LINE__, pThreadInfo->buffer);
if (0 != postProceSql(g_Dbs.host, &g_Dbs.serv_addr, g_Dbs.port,
if (0 != postProceSql(g_Dbs.host, g_Dbs.port,
pThreadInfo->buffer, pThreadInfo)) {
affectedRows = -1;
printf("========restful return fail, threadID[%d]\n",
......@@ -7021,7 +7041,8 @@ static void getTableName(char *pTblName,
if (stbInfo) {
if (AUTO_CREATE_SUBTBL != stbInfo->autoCreateTable) {
if (stbInfo->childTblLimit > 0) {
snprintf(pTblName, TSDB_TABLE_NAME_LEN, "%s",
snprintf(pTblName, TSDB_TABLE_NAME_LEN,
stbInfo->escapeChar ? "`%s`" : "%s",
stbInfo->childTblName +
(tableSeq - stbInfo->childTblOffset) * TSDB_TABLE_NAME_LEN);
} else {
......@@ -7029,15 +7050,17 @@ static void getTableName(char *pTblName,
pThreadInfo->threadID, __func__, __LINE__,
pThreadInfo->start_table_from,
pThreadInfo->ntables, tableSeq);
snprintf(pTblName, TSDB_TABLE_NAME_LEN, "%s",
snprintf(pTblName, TSDB_TABLE_NAME_LEN, stbInfo->escapeChar ? "`%s`" : "%s",
stbInfo->childTblName + tableSeq * TSDB_TABLE_NAME_LEN);
}
} else {
snprintf(pTblName, TSDB_TABLE_NAME_LEN, "%s%"PRIu64"",
snprintf(pTblName, TSDB_TABLE_NAME_LEN,
stbInfo->escapeChar ? "`%s%"PRIu64"`" : "%s%"PRIu64"",
stbInfo->childTblPrefix, tableSeq);
}
} else {
snprintf(pTblName, TSDB_TABLE_NAME_LEN, "%s%"PRIu64"",
snprintf(pTblName, TSDB_TABLE_NAME_LEN,
g_args.escapeChar ? "`%s%"PRIu64"`" : "%s%"PRIu64"",
g_args.tb_prefix, tableSeq);
}
}
......@@ -10521,6 +10544,33 @@ static void startMultiThreadInsertData(int threads, char* db_name,
pThreadInfo->start_time = pThreadInfo->start_time + rand_int() % 10000 - rand_tinyint();
}
*/
if (g_args.iface == REST_IFACE || ((stbInfo) && (stbInfo->iface == REST_IFACE))) {
#ifdef WINDOWS
WSADATA wsaData;
WSAStartup(MAKEWORD(2, 2), &wsaData);
SOCKET sockfd;
#else
int sockfd;
#endif
sockfd = socket(AF_INET, SOCK_STREAM, 0);
if (sockfd < 0) {
#ifdef WINDOWS
errorPrint( "Could not create socket : %d" , WSAGetLastError());
#endif
debugPrint("%s() LN%d, sockfd=%d\n", __func__, __LINE__, sockfd);
ERROR_EXIT("opening socket");
}
int retConn = connect(sockfd, (struct sockaddr *)&(g_Dbs.serv_addr), sizeof(struct sockaddr));
debugPrint("%s() LN%d connect() return %d\n", __func__, __LINE__, retConn);
if (retConn < 0) {
ERROR_EXIT("connecting");
}
pThreadInfo->sockfd = sockfd;
}
tsem_init(&(pThreadInfo->lock_sem), 0, 0);
if (ASYNC_MODE == g_Dbs.asyncMode) {
pthread_create(pids + i, NULL, asyncWrite, pThreadInfo);
......@@ -10558,6 +10608,14 @@ static void startMultiThreadInsertData(int threads, char* db_name,
tmfree((char *)pThreadInfo->bind_ts_array);
tmfree(pThreadInfo->bindParams);
tmfree(pThreadInfo->is_null);
if (g_args.iface == REST_IFACE || ((stbInfo) && (stbInfo->iface == REST_IFACE))) {
#ifdef WINDOWS
closesocket(pThreadInfo->sockfd);
WSACleanup();
#else
close(pThreadInfo->sockfd);
#endif
}
#else
if (pThreadInfo->sampleBindArray) {
for (int k = 0; k < MAX_SAMPLES; k++) {
......@@ -11220,6 +11278,31 @@ static int queryTestProcess() {
}
}
if (0 == strncasecmp(g_queryInfo.queryMode, "rest", 4)) {
#ifdef WINDOWS
WSADATA wsaData;
WSAStartup(MAKEWORD(2, 2), &wsaData);
SOCKET sockfd;
#else
int sockfd;
#endif
sockfd = socket(AF_INET, SOCK_STREAM, 0);
if (sockfd < 0) {
#ifdef WINDOWS
errorPrint( "Could not create socket : %d" , WSAGetLastError());
#endif
debugPrint("%s() LN%d, sockfd=%d\n", __func__, __LINE__, sockfd);
ERROR_EXIT("opening socket");
}
int retConn = connect(sockfd, (struct sockaddr *)&(g_queryInfo.serv_addr),
sizeof(struct sockaddr));
debugPrint("%s() LN%d connect() return %d\n", __func__, __LINE__, retConn);
if (retConn < 0) {
ERROR_EXIT("connecting");
}
pThreadInfo->sockfd = sockfd;
}
pThreadInfo->taos = NULL;// workaround to use separate taos connection;
pthread_create(pids + seq, NULL, specifiedTableQuery,
......@@ -11271,6 +11354,31 @@ static int queryTestProcess() {
pThreadInfo->end_table_to = i < b ? tableFrom + a : tableFrom + a - 1;
tableFrom = pThreadInfo->end_table_to + 1;
pThreadInfo->taos = NULL; // workaround to use separate taos connection;
if (0 == strncasecmp(g_queryInfo.queryMode, "rest", 4)) {
#ifdef WINDOWS
WSADATA wsaData;
WSAStartup(MAKEWORD(2, 2), &wsaData);
SOCKET sockfd;
#else
int sockfd;
#endif
sockfd = socket(AF_INET, SOCK_STREAM, 0);
if (sockfd < 0) {
#ifdef WINDOWS
errorPrint( "Could not create socket : %d" , WSAGetLastError());
#endif
debugPrint("%s() LN%d, sockfd=%d\n", __func__, __LINE__, sockfd);
ERROR_EXIT("opening socket");
}
int retConn = connect(sockfd, (struct sockaddr *)&(g_queryInfo.serv_addr),
sizeof(struct sockaddr));
debugPrint("%s() LN%d connect() return %d\n", __func__, __LINE__, retConn);
if (retConn < 0) {
ERROR_EXIT("connecting");
}
pThreadInfo->sockfd = sockfd;
}
pthread_create(pidsOfSub + i, NULL, superTableQuery, pThreadInfo);
}
......@@ -11283,6 +11391,15 @@ static int queryTestProcess() {
for (int i = 0; i < nConcurrent; i++) {
for (int j = 0; j < nSqlCount; j++) {
pthread_join(pids[i * nSqlCount + j], NULL);
if (0 == strncasecmp(g_queryInfo.queryMode, "rest", 4)) {
threadInfo *pThreadInfo = infos + i * nSqlCount + j;
#ifdef WINDOWS
closesocket(pThreadInfo->sockfd);
WSACleanup();
#else
close(pThreadInfo->sockfd);
#endif
}
}
}
}
......@@ -11292,6 +11409,15 @@ static int queryTestProcess() {
for (int i = 0; i < g_queryInfo.superQueryInfo.threadCnt; i++) {
pthread_join(pidsOfSub[i], NULL);
if (0 == strncasecmp(g_queryInfo.queryMode, "rest", 4)) {
threadInfo *pThreadInfo = infosOfSub + i;
#ifdef WINDOWS
closesocket(pThreadInfo->sockfd);
WSACleanup();
#else
close(pThreadInfo->sockfd);
#endif
}
}
tmfree((char*)pidsOfSub);
......@@ -11868,6 +11994,7 @@ static void setParaFromArg() {
g_Dbs.db[0].superTblCount = 1;
tstrncpy(g_Dbs.db[0].superTbls[0].stbName, "meters", TSDB_TABLE_NAME_LEN);
g_Dbs.db[0].superTbls[0].childTblCount = g_args.ntables;
g_Dbs.db[0].superTbls[0].escapeChar = g_args.escapeChar;
g_Dbs.threadCount = g_args.nthreads;
g_Dbs.threadCountForCreateTbl = g_args.nthreads;
g_Dbs.asyncMode = g_args.async_mode;
......@@ -11966,7 +12093,6 @@ static int regexMatch(const char *s, const char *reg, int cflags) {
printf("Regex match failed: %s\n", msgbuf);
exit(EXIT_FAILURE);
}
return 0;
}
......
......@@ -1335,11 +1335,16 @@ static void *dumpNormalTablesOfStb(void *arg) {
char tmpBuf[4096] = {0};
if (g_args.outpath[0] != 0) {
sprintf(tmpBuf, "%s/%s.%d.sql",
g_args.outpath, pThreadInfo->dbName, pThreadInfo->threadIndex);
sprintf(tmpBuf, "%s/%s.%s.%d.sql",
g_args.outpath,
pThreadInfo->dbName,
pThreadInfo->stbName,
pThreadInfo->threadIndex);
} else {
sprintf(tmpBuf, "%s.%d.sql",
pThreadInfo->dbName, pThreadInfo->threadIndex);
sprintf(tmpBuf, "%s.%s.%d.sql",
pThreadInfo->dbName,
pThreadInfo->stbName,
pThreadInfo->threadIndex);
}
fp = fopen(tmpBuf, "w");
......
......@@ -38,8 +38,8 @@ void *tsdbCommitData(STsdbRepo *pRepo);
int tsdbApplyRtnOnFSet(STsdbRepo *pRepo, SDFileSet *pSet, SRtn *pRtn);
int tsdbWriteBlockInfoImpl(SDFile *pHeadf, STable *pTable, SArray *pSupA, SArray *pSubA, void **ppBuf, SBlockIdx *pIdx);
int tsdbWriteBlockIdx(SDFile *pHeadf, SArray *pIdxA, void **ppBuf);
int tsdbWriteBlockImpl(STsdbRepo *pRepo, STable *pTable, SDFile *pDFile, SDataCols *pDataCols, SBlock *pBlock,
bool isLast, bool isSuper, void **ppBuf, void **ppCBuf);
int tsdbWriteBlockImpl(STsdbRepo *pRepo, STable *pTable, SDFile *pDFile, SDFile *pDFileAggr, SDataCols *pDataCols,
SBlock *pBlock, bool isLast, bool isSuper, void **ppBuf, void **ppCBuf, void **ppExBuf);
int tsdbApplyRtn(STsdbRepo *pRepo);
static FORCE_INLINE int tsdbGetFidLevel(int fid, SRtn *pRtn) {
......
......@@ -16,7 +16,29 @@
#ifndef _TD_TSDB_FS_H_
#define _TD_TSDB_FS_H_
#define TSDB_FS_VERSION 0
/**
* 1. The fileset .head/.data/.last use the same fver 0 before 2021.10.10.
* 2. .head fver is 1 when extract aggregate block data from .data/.last file and save to separate .smad/.smal file
* since 2021.10.10
* // TODO update date and add release version.
*/
typedef enum {
TSDB_FS_VER_0 = 0,
TSDB_FS_VER_1,
} ETsdbFsVer;
#define TSDB_FVER_TYPE uint32_t
#define TSDB_LATEST_FVER TSDB_FS_VER_1 // latest version for DFile
#define TSDB_LATEST_SFS_VER TSDB_FS_VER_1 // latest version for 'current' file
static FORCE_INLINE uint32_t tsdbGetDFSVersion(TSDB_FILE_T fType) { // latest version for DFile
switch (fType) {
case TSDB_FILE_HEAD:
return TSDB_FS_VER_1;
default:
return TSDB_FS_VER_0;
}
}
// ================== TSDB global config
extern bool tsdbForceKeepFile;
......
......@@ -37,8 +37,22 @@
#define TSDB_FILE_SET_STATE(tf, s) ((tf)->state = (s))
#define TSDB_FILE_IS_OK(tf) (TSDB_FILE_STATE(tf) == TSDB_FILE_STATE_OK)
#define TSDB_FILE_IS_BAD(tf) (TSDB_FILE_STATE(tf) == TSDB_FILE_STATE_BAD)
typedef enum { TSDB_FILE_HEAD = 0, TSDB_FILE_DATA, TSDB_FILE_LAST, TSDB_FILE_MAX, TSDB_FILE_META } TSDB_FILE_T;
#define ASSERT_TSDB_FSET_NFILES_VALID(s) \
do { \
uint8_t nDFiles = tsdbGetNFiles(s); \
ASSERT((nDFiles >= TSDB_FILE_MIN) && (nDFiles <= TSDB_FILE_MAX)); \
} while (0)
typedef enum {
TSDB_FILE_HEAD = 0,
TSDB_FILE_DATA,
TSDB_FILE_LAST,
TSDB_FILE_SMAD, // sma for .data
TSDB_FILE_SMAL, // sma for .last
TSDB_FILE_MAX,
TSDB_FILE_META
} TSDB_FILE_T;
#define TSDB_FILE_MIN 3U // min valid number of files in one DFileSet(.head/.data/.last)
// =============== SMFile
typedef struct {
......@@ -166,6 +180,7 @@ typedef struct {
uint32_t offset;
uint64_t size;
uint64_t tombSize;
uint32_t fver;
} SDFInfo;
typedef struct {
......@@ -178,8 +193,8 @@ typedef struct {
void tsdbInitDFile(SDFile* pDFile, SDiskID did, int vid, int fid, uint32_t ver, TSDB_FILE_T ftype);
void tsdbInitDFileEx(SDFile* pDFile, SDFile* pODFile);
int tsdbEncodeSDFile(void** buf, SDFile* pDFile);
void* tsdbDecodeSDFile(void* buf, SDFile* pDFile);
int tsdbCreateDFile(SDFile* pDFile, bool updateHeader);
void* tsdbDecodeSDFile(void* buf, SDFile* pDFile, uint32_t sfver);
int tsdbCreateDFile(SDFile* pDFile, bool updateHeader, TSDB_FILE_T ftype);
int tsdbUpdateDFileHeader(SDFile* pDFile);
int tsdbLoadDFileHeader(SDFile* pDFile, SDFInfo* pInfo);
int tsdbParseDFilename(const char* fname, int* vid, int* fid, TSDB_FILE_T* ftype, uint32_t* version);
......@@ -283,11 +298,29 @@ static FORCE_INLINE int tsdbCopyDFile(SDFile* pSrc, SDFile* pDest) {
// =============== SDFileSet
typedef struct {
int fid;
int state;
SDFile files[TSDB_FILE_MAX];
int fid;
int state;
uint16_t ver; // fset version
SDFile files[TSDB_FILE_MAX];
} SDFileSet;
typedef enum {
TSDB_FSET_VER_0 = 0, // .head/.data/.last
TSDB_FSET_VER_1, // .head/.data/.last/.smad/.smal
} ETsdbFSetVer;
#define TSDB_LATEST_FSET_VER TSDB_FSET_VER_1
// get nDFiles in SDFileSet
static FORCE_INLINE uint8_t tsdbGetNFiles(SDFileSet* pSet) {
switch (pSet->ver) {
case TSDB_FSET_VER_0:
return TSDB_FILE_MIN;
case TSDB_FSET_VER_1:
default:
return TSDB_FILE_MAX;
}
}
#define TSDB_FSET_FID(s) ((s)->fid)
#define TSDB_DFILE_IN_SET(s, t) ((s)->files + (t))
#define TSDB_FSET_LEVEL(s) TSDB_FILE_LEVEL(TSDB_DFILE_IN_SET(s, 0))
......@@ -298,17 +331,17 @@ typedef struct {
TSDB_FILE_SET_CLOSED(TSDB_DFILE_IN_SET(s, ftype)); \
} \
} while (0);
#define TSDB_FSET_FSYNC(s) \
do { \
for (TSDB_FILE_T ftype = TSDB_FILE_HEAD; ftype < TSDB_FILE_MAX; ftype++) { \
TSDB_FILE_FSYNC(TSDB_DFILE_IN_SET(s, ftype)); \
} \
#define TSDB_FSET_FSYNC(s) \
do { \
for (TSDB_FILE_T ftype = TSDB_FILE_HEAD; ftype < tsdbGetNFiles(s); ftype++) { \
TSDB_FILE_FSYNC(TSDB_DFILE_IN_SET(s, ftype)); \
} \
} while (0);
void tsdbInitDFileSet(SDFileSet* pSet, SDiskID did, int vid, int fid, uint32_t ver);
void tsdbInitDFileSet(SDFileSet* pSet, SDiskID did, int vid, int fid, uint32_t ver, uint16_t fsetVer);
void tsdbInitDFileSetEx(SDFileSet* pSet, SDFileSet* pOSet);
int tsdbEncodeDFileSet(void** buf, SDFileSet* pSet);
void* tsdbDecodeDFileSet(void* buf, SDFileSet* pSet);
void* tsdbDecodeDFileSet(void* buf, SDFileSet* pSet, uint32_t sfver);
int tsdbEncodeDFileSetEx(void** buf, SDFileSet* pSet);
void* tsdbDecodeDFileSetEx(void* buf, SDFileSet* pSet);
int tsdbApplyDFileSetChange(SDFileSet* from, SDFileSet* to);
......@@ -317,13 +350,15 @@ int tsdbUpdateDFileSetHeader(SDFileSet* pSet);
int tsdbScanAndTryFixDFileSet(STsdbRepo *pRepo, SDFileSet* pSet);
static FORCE_INLINE void tsdbCloseDFileSet(SDFileSet* pSet) {
for (TSDB_FILE_T ftype = 0; ftype < TSDB_FILE_MAX; ftype++) {
ASSERT_TSDB_FSET_NFILES_VALID(pSet);
for (TSDB_FILE_T ftype = 0; ftype < tsdbGetNFiles(pSet); ftype++) {
tsdbCloseDFile(TSDB_DFILE_IN_SET(pSet, ftype));
}
}
static FORCE_INLINE int tsdbOpenDFileSet(SDFileSet* pSet, int flags) {
for (TSDB_FILE_T ftype = 0; ftype < TSDB_FILE_MAX; ftype++) {
ASSERT_TSDB_FSET_NFILES_VALID(pSet);
for (TSDB_FILE_T ftype = 0; ftype < tsdbGetNFiles(pSet); ftype++) {
if (tsdbOpenDFile(TSDB_DFILE_IN_SET(pSet, ftype), flags) < 0) {
tsdbCloseDFileSet(pSet);
return -1;
......@@ -333,13 +368,15 @@ static FORCE_INLINE int tsdbOpenDFileSet(SDFileSet* pSet, int flags) {
}
static FORCE_INLINE void tsdbRemoveDFileSet(SDFileSet* pSet) {
for (TSDB_FILE_T ftype = 0; ftype < TSDB_FILE_MAX; ftype++) {
ASSERT_TSDB_FSET_NFILES_VALID(pSet);
for (TSDB_FILE_T ftype = 0; ftype < tsdbGetNFiles(pSet); ftype++) {
(void)tsdbRemoveDFile(TSDB_DFILE_IN_SET(pSet, ftype));
}
}
static FORCE_INLINE int tsdbCopyDFileSet(SDFileSet* pSrc, SDFileSet* pDest) {
for (TSDB_FILE_T ftype = 0; ftype < TSDB_FILE_MAX; ftype++) {
ASSERT_TSDB_FSET_NFILES_VALID(pSrc);
for (TSDB_FILE_T ftype = 0; ftype < tsdbGetNFiles(pSrc); ftype++) {
if (tsdbCopyDFile(TSDB_DFILE_IN_SET(pSrc, ftype), TSDB_DFILE_IN_SET(pDest, ftype)) < 0) {
tsdbRemoveDFileSet(pDest);
return -1;
......
......@@ -35,6 +35,7 @@ typedef struct {
TSKEY maxKey;
} SBlockIdx;
#if 0
typedef struct {
int64_t last : 1;
int64_t offset : 63;
......@@ -46,8 +47,55 @@ typedef struct {
int16_t numOfCols; // not including timestamp column
TSKEY keyFirst;
TSKEY keyLast;
} SBlock;
} SBlock;
#endif
/**
* keyLen; // key column length, keyOffset = offset+sizeof(SBlockData)+sizeof(SBlockCol)*numOfCols
* numOfCols; // not including timestamp column
*/
#define SBlockFieldsP0 \
int64_t last : 1; \
int64_t offset : 63; \
int32_t algorithm : 8; \
int32_t numOfRows : 24; \
int32_t len; \
int32_t keyLen; \
int16_t numOfSubBlocks; \
int16_t numOfCols; \
TSKEY keyFirst; \
TSKEY keyLast
/**
* aggrStat; // only valid when blkVer > 0. 0 - no aggr part in .data/.last/.smad/.smal, 1 - has aggr in .smad/.smal
* blkVer; // 0 - original block, 1 - block since importing .smad/.smal
* aggrOffset; // only valid when blkVer > 0 and aggrStat > 0
*/
#define SBlockFieldsP1 \
uint64_t aggrStat : 3; \
uint64_t blkVer : 5; \
uint64_t aggrOffset : 56; \
uint32_t aggrLen
typedef struct {
SBlockFieldsP0;
} SBlockV0;
typedef struct {
SBlockFieldsP0;
SBlockFieldsP1;
} SBlockV1;
typedef enum {
TSDB_SBLK_VER_0 = 0,
TSDB_SBLK_VER_1,
} ESBlockVer;
#define SBlockVerLatest TSDB_SBLK_VER_1
#define SBlock SBlockV1 // latest SBlock definition
// lastest SBlockInfo definition
typedef struct {
int32_t delimiter; // For recovery usage
int32_t tid;
......@@ -68,7 +116,31 @@ typedef struct {
int16_t numOfNull;
uint8_t offsetH;
char padding[1];
} SBlockCol;
} SBlockColV0;
typedef struct {
int16_t colId;
uint8_t offsetH;
uint8_t reserved; // reserved field, not used
int32_t len;
uint32_t type : 8;
uint32_t offset : 24;
// char padding[];
} SBlockColV1;
#define SBlockCol SBlockColV1 // latest SBlockCol definition
typedef struct {
int16_t colId;
int16_t maxIndex;
int16_t minIndex;
int16_t numOfNull;
int64_t sum;
int64_t max;
int64_t min;
} SAggrBlkColV1;
#define SAggrBlkCol SAggrBlkColV1 // latest SAggrBlkCol definition
// Code here just for back-ward compatibility
static FORCE_INLINE void tsdbSetBlockColOffset(SBlockCol *pBlockCol, uint32_t offset) {
......@@ -88,6 +160,10 @@ typedef struct {
uint64_t uid; // For recovery usage
SBlockCol cols[];
} SBlockData;
typedef struct {
int32_t numOfCols; // For recovery usage
SAggrBlkCol cols[];
} SAggrBlkData;
struct SReadH {
STsdbRepo * pRepo;
......@@ -96,11 +172,13 @@ struct SReadH {
STable * pTable; // table to read
SBlockIdx * pBlkIdx; // current reading table SBlockIdx
int cidx;
SBlockInfo *pBlkInfo;
SBlockInfo * pBlkInfo; // SBlockInfoV#
SBlockData *pBlkData; // Block info
SAggrBlkData *pAggrBlkData; // Aggregate Block info
SDataCols * pDCols[2];
void * pBuf; // buffer
void * pCBuf; // compression buffer
void * pExBuf; // extra buffer
};
#define TSDB_READ_REPO(rh) ((rh)->pRepo)
......@@ -110,10 +188,38 @@ struct SReadH {
#define TSDB_READ_HEAD_FILE(rh) TSDB_DFILE_IN_SET(TSDB_READ_FSET(rh), TSDB_FILE_HEAD)
#define TSDB_READ_DATA_FILE(rh) TSDB_DFILE_IN_SET(TSDB_READ_FSET(rh), TSDB_FILE_DATA)
#define TSDB_READ_LAST_FILE(rh) TSDB_DFILE_IN_SET(TSDB_READ_FSET(rh), TSDB_FILE_LAST)
#define TSDB_READ_SMAD_FILE(rh) TSDB_DFILE_IN_SET(TSDB_READ_FSET(rh), TSDB_FILE_SMAD)
#define TSDB_READ_SMAL_FILE(rh) TSDB_DFILE_IN_SET(TSDB_READ_FSET(rh), TSDB_FILE_SMAL)
#define TSDB_READ_BUF(rh) ((rh)->pBuf)
#define TSDB_READ_COMP_BUF(rh) ((rh)->pCBuf)
#define TSDB_READ_EXBUF(rh) ((rh)->pExBuf)
#define TSDB_BLOCK_STATIS_SIZE(ncols, blkVer) \
(sizeof(SBlockData) + sizeof(SBlockColV##blkVer) * (ncols) + sizeof(TSCKSUM))
#define TSDB_BLOCK_STATIS_SIZE(ncols) (sizeof(SBlockData) + sizeof(SBlockCol) * (ncols) + sizeof(TSCKSUM))
static FORCE_INLINE size_t tsdbBlockStatisSize(int nCols, uint32_t blkVer) {
switch (blkVer) {
case TSDB_SBLK_VER_0:
return TSDB_BLOCK_STATIS_SIZE(nCols, 0);
case TSDB_SBLK_VER_1:
default:
return TSDB_BLOCK_STATIS_SIZE(nCols, 1);
}
}
#define TSDB_BLOCK_AGGR_SIZE(ncols, blkVer) \
(sizeof(SAggrBlkData) + sizeof(SAggrBlkColV##blkVer) * (ncols) + sizeof(TSCKSUM))
static FORCE_INLINE size_t tsdbBlockAggrSize(int nCols, uint32_t blkVer) {
switch (blkVer) {
case TSDB_SBLK_VER_0:
ASSERT(false);
return 0;
case TSDB_SBLK_VER_1:
default:
return TSDB_BLOCK_AGGR_SIZE(nCols, 1);
}
}
int tsdbInitReadH(SReadH *pReadh, STsdbRepo *pRepo);
void tsdbDestroyReadH(SReadH *pReadh);
......@@ -121,13 +227,14 @@ int tsdbSetAndOpenReadFSet(SReadH *pReadh, SDFileSet *pSet);
void tsdbCloseAndUnsetFSet(SReadH *pReadh);
int tsdbLoadBlockIdx(SReadH *pReadh);
int tsdbSetReadTable(SReadH *pReadh, STable *pTable);
int tsdbLoadBlockInfo(SReadH *pReadh, void *pTarget);
int tsdbLoadBlockInfo(SReadH *pReadh, void **pTarget, uint32_t *extendedLen);
int tsdbLoadBlockData(SReadH *pReadh, SBlock *pBlock, SBlockInfo *pBlockInfo);
int tsdbLoadBlockDataCols(SReadH *pReadh, SBlock *pBlock, SBlockInfo *pBlkInfo, int16_t *colIds, int numOfColsIds);
int tsdbLoadBlockStatis(SReadH *pReadh, SBlock *pBlock);
int tsdbLoadBlockOffset(SReadH *pReadh, SBlock *pBlock);
int tsdbEncodeSBlockIdx(void **buf, SBlockIdx *pIdx);
void *tsdbDecodeSBlockIdx(void *buf, SBlockIdx *pIdx);
void tsdbGetBlockStatis(SReadH *pReadh, SDataStatis *pStatis, int numOfCols);
void tsdbGetBlockStatis(SReadH *pReadh, SDataStatis *pStatis, int numOfCols, SBlock *pBlock);
static FORCE_INLINE int tsdbMakeRoom(void **ppBuf, size_t size) {
void * pBuf = *ppBuf;
......@@ -150,4 +257,21 @@ static FORCE_INLINE int tsdbMakeRoom(void **ppBuf, size_t size) {
return 0;
}
static FORCE_INLINE SBlockCol *tsdbGetSBlockCol(SBlock *pBlock, SBlockCol **pDestBlkCol, SBlockCol *pBlkCols,
int colIdx) {
if (pBlock->blkVer == SBlockVerLatest) {
*pDestBlkCol = pBlkCols + colIdx;
return *pDestBlkCol;
}
if (pBlock->blkVer == TSDB_SBLK_VER_0) {
SBlockColV0 *pBlkCol = (SBlockColV0 *)pBlkCols + colIdx;
(*pDestBlkCol)->colId = pBlkCol->colId;
(*pDestBlkCol)->len = pBlkCol->len;
(*pDestBlkCol)->type = pBlkCol->type;
(*pDestBlkCol)->offset = pBlkCol->offset;
(*pDestBlkCol)->offsetH = pBlkCol->offsetH;
}
return *pDestBlkCol;
}
#endif /*_TD_TSDB_READ_IMPL_H_*/
......@@ -51,8 +51,11 @@ typedef struct {
#define TSDB_COMMIT_HEAD_FILE(ch) TSDB_DFILE_IN_SET(TSDB_COMMIT_WRITE_FSET(ch), TSDB_FILE_HEAD)
#define TSDB_COMMIT_DATA_FILE(ch) TSDB_DFILE_IN_SET(TSDB_COMMIT_WRITE_FSET(ch), TSDB_FILE_DATA)
#define TSDB_COMMIT_LAST_FILE(ch) TSDB_DFILE_IN_SET(TSDB_COMMIT_WRITE_FSET(ch), TSDB_FILE_LAST)
#define TSDB_COMMIT_SMAD_FILE(ch) TSDB_DFILE_IN_SET(TSDB_COMMIT_WRITE_FSET(ch), TSDB_FILE_SMAD)
#define TSDB_COMMIT_SMAL_FILE(ch) TSDB_DFILE_IN_SET(TSDB_COMMIT_WRITE_FSET(ch), TSDB_FILE_SMAL)
#define TSDB_COMMIT_BUF(ch) TSDB_READ_BUF(&((ch)->readh))
#define TSDB_COMMIT_COMP_BUF(ch) TSDB_READ_COMP_BUF(&((ch)->readh))
#define TSDB_COMMIT_EXBUF(ch) TSDB_READ_EXBUF(&((ch)->readh))
#define TSDB_COMMIT_DEFAULT_ROWS(ch) TSDB_DEFAULT_BLOCK_ROWS(TSDB_COMMIT_REPO(ch)->config.maxRowsPerFileBlock)
#define TSDB_COMMIT_TXN_VERSION(ch) FS_TXN_VERSION(REPO_FS(TSDB_COMMIT_REPO(ch)))
......@@ -136,7 +139,7 @@ int tsdbApplyRtnOnFSet(STsdbRepo *pRepo, SDFileSet *pSet, SRtn *pRtn) {
if (did.level > TSDB_FSET_LEVEL(pSet)) {
// Need to move the FSET to higher level
tsdbInitDFileSet(&nSet, did, REPO_ID(pRepo), pSet->fid, FS_TXN_VERSION(pfs));
tsdbInitDFileSet(&nSet, did, REPO_ID(pRepo), pSet->fid, FS_TXN_VERSION(pfs), pSet->ver);
if (tsdbCopyDFileSet(pSet, &nSet) < 0) {
tsdbError("vgId:%d failed to copy FSET %d from level %d to level %d since %s", REPO_ID(pRepo), pSet->fid,
......@@ -912,7 +915,7 @@ static int tsdbNextCommitFid(SCommitH *pCommith) {
} else {
int tfid = (int)(TSDB_KEY_FID(nextKey, pCfg->daysPerFile, pCfg->precision));
if (fid == TSDB_IVLD_FID || fid > tfid) {
fid = tfid;
fid = tfid; // find the least fid
}
}
}
......@@ -946,7 +949,7 @@ static int tsdbCommitToTable(SCommitH *pCommith, int tid) {
SBlock *pBlock;
if (pCommith->readh.pBlkIdx) {
if (tsdbLoadBlockInfo(&(pCommith->readh), NULL) < 0) {
if (tsdbLoadBlockInfo(&(pCommith->readh), NULL, NULL) < 0) {
TSDB_RUNLOCK_TABLE(pIter->pTable);
return -1;
}
......@@ -1053,40 +1056,57 @@ static int tsdbComparKeyBlock(const void *arg1, const void *arg2) {
}
}
int tsdbWriteBlockImpl(STsdbRepo *pRepo, STable *pTable, SDFile *pDFile, SDataCols *pDataCols, SBlock *pBlock,
bool isLast, bool isSuper, void **ppBuf, void **ppCBuf) {
int tsdbWriteBlockImpl(STsdbRepo *pRepo, STable *pTable, SDFile *pDFile, SDFile *pDFileAggr, SDataCols *pDataCols,
SBlock *pBlock, bool isLast, bool isSuper, void **ppBuf, void **ppCBuf, void **ppExBuf) {
STsdbCfg * pCfg = REPO_CFG(pRepo);
SBlockData *pBlockData;
int64_t offset = 0;
SAggrBlkData *pAggrBlkData = NULL;
int64_t offset = 0, offsetAggr = 0;
int rowsToWrite = pDataCols->numOfRows;
ASSERT(rowsToWrite > 0 && rowsToWrite <= pCfg->maxRowsPerFileBlock);
ASSERT((!isLast) || rowsToWrite < pCfg->minRowsPerFileBlock);
// Make buffer space
if (tsdbMakeRoom(ppBuf, TSDB_BLOCK_STATIS_SIZE(pDataCols->numOfCols)) < 0) {
if (tsdbMakeRoom(ppBuf, tsdbBlockStatisSize(pDataCols->numOfCols, SBlockVerLatest)) < 0) {
return -1;
}
pBlockData = (SBlockData *)(*ppBuf);
if (tsdbMakeRoom(ppExBuf, tsdbBlockAggrSize(pDataCols->numOfCols, SBlockVerLatest)) < 0) {
return -1;
}
pAggrBlkData = (SAggrBlkData *)(*ppExBuf);
// Get # of cols not all NULL(not including key column)
int nColsNotAllNull = 0;
int nAggrCols = 0;
for (int ncol = 1; ncol < pDataCols->numOfCols; ncol++) { // ncol from 1, we skip the timestamp column
SDataCol * pDataCol = pDataCols->cols + ncol;
SBlockCol *pBlockCol = pBlockData->cols + nColsNotAllNull;
SDataCol * pDataCol = pDataCols->cols + ncol;
SBlockCol * pBlockCol = pBlockData->cols + nColsNotAllNull;
SAggrBlkCol *pAggrBlkCol = pAggrBlkData->cols + nColsNotAllNull;
if (isAllRowsNull(pDataCol)) { // all data to commit are NULL, just ignore it
continue;
}
memset(pBlockCol, 0, sizeof(*pBlockCol));
memset(pAggrBlkCol, 0, sizeof(*pAggrBlkCol));
pBlockCol->colId = pDataCol->colId;
pBlockCol->type = pDataCol->type;
pAggrBlkCol->colId = pDataCol->colId;
if (tDataTypes[pDataCol->type].statisFunc) {
#if 0
(*tDataTypes[pDataCol->type].statisFunc)(pDataCol->pData, rowsToWrite, &(pBlockCol->min), &(pBlockCol->max),
&(pBlockCol->sum), &(pBlockCol->minIndex), &(pBlockCol->maxIndex),
&(pBlockCol->numOfNull));
#endif
(*tDataTypes[pDataCol->type].statisFunc)(pDataCol->pData, rowsToWrite, &(pAggrBlkCol->min), &(pAggrBlkCol->max),
&(pAggrBlkCol->sum), &(pAggrBlkCol->minIndex), &(pAggrBlkCol->maxIndex),
&(pAggrBlkCol->numOfNull));
++nAggrCols;
}
nColsNotAllNull++;
}
......@@ -1096,9 +1116,12 @@ int tsdbWriteBlockImpl(STsdbRepo *pRepo, STable *pTable, SDFile *pDFile, SDataCo
// Compress the data if neccessary
int tcol = 0; // counter of not all NULL and written columns
uint32_t toffset = 0;
int32_t tsize = TSDB_BLOCK_STATIS_SIZE(nColsNotAllNull);
int32_t tsize = (int32_t)tsdbBlockStatisSize(nColsNotAllNull, SBlockVerLatest);
int32_t lsize = tsize;
int32_t keyLen = 0;
uint32_t tsizeAggr = (uint32_t)tsdbBlockAggrSize(nColsNotAllNull, SBlockVerLatest);
for (int ncol = 0; ncol < pDataCols->numOfCols; ncol++) {
// All not NULL columns finish
if (ncol != 0 && tcol >= nColsNotAllNull) break;
......@@ -1165,7 +1188,20 @@ int tsdbWriteBlockImpl(STsdbRepo *pRepo, STable *pTable, SDFile *pDFile, SDataCo
return -1;
}
// Update pBlock membership vairables
uint32_t aggrStatus = ((nAggrCols > 0) && (rowsToWrite > 8)) ? 1 : 0; // TODO: How to make the decision?
if (aggrStatus > 0) {
pAggrBlkData->numOfCols = nColsNotAllNull;
taosCalcChecksumAppend(0, (uint8_t *)pAggrBlkData, tsizeAggr);
tsdbUpdateDFileMagic(pDFileAggr, POINTER_SHIFT(pAggrBlkData, tsizeAggr - sizeof(TSCKSUM)));
// Write the whole block to file
if (tsdbAppendDFile(pDFileAggr, (void *)pAggrBlkData, tsizeAggr, &offsetAggr) < tsizeAggr) {
return -1;
}
}
// Update pBlock membership variables
pBlock->last = isLast;
pBlock->offset = offset;
pBlock->algorithm = pCfg->compression;
......@@ -1176,6 +1212,11 @@ int tsdbWriteBlockImpl(STsdbRepo *pRepo, STable *pTable, SDFile *pDFile, SDataCo
pBlock->numOfCols = nColsNotAllNull;
pBlock->keyFirst = dataColsKeyFirst(pDataCols);
pBlock->keyLast = dataColsKeyLast(pDataCols);
// since blkVer1
pBlock->aggrStat = aggrStatus;
pBlock->blkVer = SBlockVerLatest;
pBlock->aggrOffset = (uint64_t)offsetAggr;
pBlock->aggrLen = tsizeAggr;
tsdbDebug("vgId:%d tid:%d a block of data is written to file %s, offset %" PRId64
" numOfRows %d len %d numOfCols %" PRId16 " keyFirst %" PRId64 " keyLast %" PRId64,
......@@ -1187,12 +1228,12 @@ int tsdbWriteBlockImpl(STsdbRepo *pRepo, STable *pTable, SDFile *pDFile, SDataCo
static int tsdbWriteBlock(SCommitH *pCommith, SDFile *pDFile, SDataCols *pDataCols, SBlock *pBlock, bool isLast,
bool isSuper) {
return tsdbWriteBlockImpl(TSDB_COMMIT_REPO(pCommith), TSDB_COMMIT_TABLE(pCommith), pDFile, pDataCols, pBlock, isLast,
isSuper, (void **)(&(TSDB_COMMIT_BUF(pCommith))),
(void **)(&(TSDB_COMMIT_COMP_BUF(pCommith))));
return tsdbWriteBlockImpl(TSDB_COMMIT_REPO(pCommith), TSDB_COMMIT_TABLE(pCommith), pDFile,
isLast ? TSDB_COMMIT_SMAL_FILE(pCommith) : TSDB_COMMIT_SMAD_FILE(pCommith), pDataCols,
pBlock, isLast, isSuper, (void **)(&(TSDB_COMMIT_BUF(pCommith))),
(void **)(&(TSDB_COMMIT_COMP_BUF(pCommith))), (void **)(&(TSDB_COMMIT_EXBUF(pCommith))));
}
static int tsdbWriteBlockInfo(SCommitH *pCommih) {
SDFile * pHeadf = TSDB_COMMIT_HEAD_FILE(pCommih);
SBlockIdx blkIdx;
......@@ -1521,7 +1562,7 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
// Set and open commit FSET
if (pSet == NULL || did.level > TSDB_FSET_LEVEL(pSet)) {
// Create a new FSET to write data
tsdbInitDFileSet(pWSet, did, REPO_ID(pRepo), fid, FS_TXN_VERSION(REPO_FS(pRepo)));
tsdbInitDFileSet(pWSet, did, REPO_ID(pRepo), fid, FS_TXN_VERSION(REPO_FS(pRepo)), TSDB_LATEST_FSET_VER);
if (tsdbCreateDFileSet(pWSet, true) < 0) {
tsdbError("vgId:%d failed to create FSET %d at level %d disk id %d since %s", REPO_ID(pRepo),
......@@ -1543,11 +1584,12 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
pCommith->wSet.fid = fid;
pCommith->wSet.state = 0;
pCommith->wSet.ver = TSDB_LATEST_FSET_VER;
// TSDB_FILE_HEAD
SDFile *pWHeadf = TSDB_COMMIT_HEAD_FILE(pCommith);
tsdbInitDFile(pWHeadf, did, REPO_ID(pRepo), fid, FS_TXN_VERSION(REPO_FS(pRepo)), TSDB_FILE_HEAD);
if (tsdbCreateDFile(pWHeadf, true) < 0) {
if (tsdbCreateDFile(pWHeadf, true, TSDB_FILE_HEAD) < 0) {
tsdbError("vgId:%d failed to create file %s to commit since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pWHeadf),
tstrerror(terrno));
......@@ -1596,7 +1638,7 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
tsdbInitDFile(pWLastf, did, REPO_ID(pRepo), fid, FS_TXN_VERSION(REPO_FS(pRepo)), TSDB_FILE_LAST);
pCommith->isLFileSame = false;
if (tsdbCreateDFile(pWLastf, true) < 0) {
if (tsdbCreateDFile(pWLastf, true, TSDB_FILE_LAST) < 0) {
tsdbError("vgId:%d failed to create file %s to commit since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pWLastf),
tstrerror(terrno));
......@@ -1608,6 +1650,75 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
}
}
}
// TSDB_FILE_SMAD
SDFile *pRSmadF = TSDB_READ_SMAD_FILE(&(pCommith->readh));
SDFile *pWSmadF = TSDB_COMMIT_SMAD_FILE(pCommith);
if (access(TSDB_FILE_FULL_NAME(pRSmadF), F_OK) != 0) {
tsdbDebug("vgId:%d create data file %s as not exist", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pRSmadF));
tsdbInitDFile(pWSmadF, did, REPO_ID(pRepo), fid, FS_TXN_VERSION(REPO_FS(pRepo)), TSDB_FILE_SMAD);
if (tsdbCreateDFile(pWSmadF, true, TSDB_FILE_SMAD) < 0) {
tsdbError("vgId:%d failed to create file %s to commit since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pWSmadF),
tstrerror(terrno));
tsdbCloseDFileSet(pWSet);
(void)tsdbRemoveDFile(pWHeadf);
if (pCommith->isRFileSet) {
tsdbCloseAndUnsetFSet(&(pCommith->readh));
return -1;
}
}
} else {
tsdbInitDFileEx(pWSmadF, pRSmadF);
if (tsdbOpenDFile(pWSmadF, O_RDWR) < 0) {
tsdbError("vgId:%d failed to open file %s to commit since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pWSmadF),
tstrerror(terrno));
tsdbCloseDFileSet(pWSet);
tsdbRemoveDFile(pWHeadf);
if (pCommith->isRFileSet) {
tsdbCloseAndUnsetFSet(&(pCommith->readh));
return -1;
}
}
}
// TSDB_FILE_SMAL
ASSERT(tsdbGetNFiles(pWSet) >= TSDB_FILE_SMAL);
SDFile *pRSmalF = TSDB_READ_SMAL_FILE(&(pCommith->readh));
SDFile *pWSmalF = TSDB_COMMIT_SMAL_FILE(pCommith);
if ((pCommith->isLFileSame) && access(TSDB_FILE_FULL_NAME(pRSmalF), F_OK) == 0) {
tsdbInitDFileEx(pWSmalF, pRSmalF);
if (tsdbOpenDFile(pWSmalF, O_RDWR) < 0) {
tsdbError("vgId:%d failed to open file %s to commit since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pWSmalF),
tstrerror(terrno));
tsdbCloseDFileSet(pWSet);
tsdbRemoveDFile(pWHeadf);
if (pCommith->isRFileSet) {
tsdbCloseAndUnsetFSet(&(pCommith->readh));
return -1;
}
}
} else {
tsdbDebug("vgId:%d create data file %s as not exist", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pRSmalF));
tsdbInitDFile(pWSmalF, did, REPO_ID(pRepo), fid, FS_TXN_VERSION(REPO_FS(pRepo)), TSDB_FILE_SMAL);
if (tsdbCreateDFile(pWSmalF, true, TSDB_FILE_SMAL) < 0) {
tsdbError("vgId:%d failed to create file %s to commit since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pWSmalF),
tstrerror(terrno));
tsdbCloseDFileSet(pWSet);
(void)tsdbRemoveDFile(pWHeadf);
if (pCommith->isRFileSet) {
tsdbCloseAndUnsetFSet(&(pCommith->readh));
return -1;
}
}
}
}
return 0;
......
......@@ -37,8 +37,11 @@ typedef struct {
#define TSDB_COMPACT_HEAD_FILE(pComph) TSDB_DFILE_IN_SET(TSDB_COMPACT_WSET(pComph), TSDB_FILE_HEAD)
#define TSDB_COMPACT_DATA_FILE(pComph) TSDB_DFILE_IN_SET(TSDB_COMPACT_WSET(pComph), TSDB_FILE_DATA)
#define TSDB_COMPACT_LAST_FILE(pComph) TSDB_DFILE_IN_SET(TSDB_COMPACT_WSET(pComph), TSDB_FILE_LAST)
#define TSDB_COMPACT_SMAD_FILE(pComph) TSDB_DFILE_IN_SET(TSDB_COMPACT_WSET(pComph), TSDB_FILE_SMAD)
#define TSDB_COMPACT_SMAL_FILE(pComph) TSDB_DFILE_IN_SET(TSDB_COMPACT_WSET(pComph), TSDB_FILE_SMAL)
#define TSDB_COMPACT_BUF(pComph) TSDB_READ_BUF(&((pComph)->readh))
#define TSDB_COMPACT_COMP_BUF(pComph) TSDB_READ_COMP_BUF(&((pComph)->readh))
#define TSDB_COMPACT_EXBUF(pComph) TSDB_READ_EXBUF(&((pComph)->readh))
static int tsdbAsyncCompact(STsdbRepo *pRepo);
static void tsdbStartCompact(STsdbRepo *pRepo);
......@@ -56,7 +59,7 @@ static int tsdbCompactFSetInit(SCompactH *pComph, SDFileSet *pSet);
static void tsdbCompactFSetEnd(SCompactH *pComph);
static int tsdbCompactFSetImpl(SCompactH *pComph);
static int tsdbWriteBlockToRightFile(SCompactH *pComph, STable *pTable, SDataCols *pDataCols, void **ppBuf,
void **ppCBuf);
void **ppCBuf, void **ppExBuf);
enum { TSDB_NO_COMPACT, TSDB_IN_COMPACT, TSDB_WAITING_COMPACT};
int tsdbCompact(STsdbRepo *pRepo) { return tsdbAsyncCompact(pRepo); }
......@@ -194,7 +197,7 @@ static int tsdbCompactMeta(STsdbRepo *pRepo) {
}
tsdbInitDFileSet(TSDB_COMPACT_WSET(pComph), did, REPO_ID(pRepo), TSDB_FSET_FID(pSet),
FS_TXN_VERSION(REPO_FS(pRepo)));
FS_TXN_VERSION(REPO_FS(pRepo)), TSDB_LATEST_FSET_VER);
if (tsdbCreateDFileSet(TSDB_COMPACT_WSET(pComph), true) < 0) {
tsdbError("vgId:%d failed to compact FSET %d since %s", REPO_ID(pRepo), pSet->fid, tstrerror(terrno));
tsdbCompactFSetEnd(pComph);
......@@ -218,6 +221,9 @@ static int tsdbCompactMeta(STsdbRepo *pRepo) {
}
static bool tsdbShouldCompact(SCompactH *pComph) {
if (tsdbForceCompactFile) {
return true;
}
STsdbRepo * pRepo = TSDB_COMPACT_REPO(pComph);
STsdbCfg * pCfg = REPO_CFG(pRepo);
SReadH * pReadh = &(pComph->readh);
......@@ -358,7 +364,8 @@ static int tsdbCompactMeta(STsdbRepo *pRepo) {
tsdbUnRefTable(pTh->pTable);
}
pTh->pInfo = taosTZfree(pTh->pInfo);
// pTh->pInfo = taosTZfree(pTh->pInfo);
tfree(pTh->pInfo);
}
pComph->tbArray = taosArrayDestroy(pComph->tbArray);
......@@ -384,11 +391,8 @@ static int tsdbCompactMeta(STsdbRepo *pRepo) {
pTh->bindex = *(pReadH->pBlkIdx);
pTh->pBlkIdx = &(pTh->bindex);
if (tsdbMakeRoom((void **)(&(pTh->pInfo)), pTh->pBlkIdx->len) < 0) {
return -1;
}
if (tsdbLoadBlockInfo(pReadH, (void *)(pTh->pInfo)) < 0) {
uint32_t originLen = 0;
if (tsdbLoadBlockInfo(pReadH, (void **)(&(pTh->pInfo)), &originLen) < 0) {
return -1;
}
}
......@@ -421,6 +425,7 @@ static int tsdbCompactMeta(STsdbRepo *pRepo) {
SBlockIdx blkIdx;
void ** ppBuf = &(TSDB_COMPACT_BUF(pComph));
void ** ppCBuf = &(TSDB_COMPACT_COMP_BUF(pComph));
void ** ppExBuf = &(TSDB_COMPACT_EXBUF(pComph));
int defaultRows = TSDB_DEFAULT_BLOCK_ROWS(pCfg->maxRowsPerFileBlock);
taosArrayClear(pComph->aBlkIdx);
......@@ -451,7 +456,7 @@ static int tsdbCompactMeta(STsdbRepo *pRepo) {
// Merge pComph->pDataCols and pReadh->pDCols[0] and write data to file
if (pComph->pDataCols->numOfRows == 0 && pBlock->numOfRows >= defaultRows) {
if (tsdbWriteBlockToRightFile(pComph, pTh->pTable, pReadh->pDCols[0], ppBuf, ppCBuf) < 0) {
if (tsdbWriteBlockToRightFile(pComph, pTh->pTable, pReadh->pDCols[0], ppBuf, ppCBuf, ppExBuf) < 0) {
return -1;
}
} else {
......@@ -467,7 +472,7 @@ static int tsdbCompactMeta(STsdbRepo *pRepo) {
break;
}
if (tsdbWriteBlockToRightFile(pComph, pTh->pTable, pComph->pDataCols, ppBuf, ppCBuf) < 0) {
if (tsdbWriteBlockToRightFile(pComph, pTh->pTable, pComph->pDataCols, ppBuf, ppCBuf, ppExBuf) < 0) {
return -1;
}
tdResetDataCols(pComph->pDataCols);
......@@ -476,7 +481,7 @@ static int tsdbCompactMeta(STsdbRepo *pRepo) {
}
if (pComph->pDataCols->numOfRows > 0 &&
tsdbWriteBlockToRightFile(pComph, pTh->pTable, pComph->pDataCols, ppBuf, ppCBuf) < 0) {
tsdbWriteBlockToRightFile(pComph, pTh->pTable, pComph->pDataCols, ppBuf, ppCBuf, ppExBuf) < 0) {
return -1;
}
......@@ -499,7 +504,7 @@ static int tsdbCompactMeta(STsdbRepo *pRepo) {
}
static int tsdbWriteBlockToRightFile(SCompactH *pComph, STable *pTable, SDataCols *pDataCols, void **ppBuf,
void **ppCBuf) {
void **ppCBuf, void **ppExBuf) {
STsdbRepo *pRepo = TSDB_COMPACT_REPO(pComph);
STsdbCfg * pCfg = REPO_CFG(pRepo);
SDFile * pDFile;
......@@ -516,7 +521,9 @@ static int tsdbCompactMeta(STsdbRepo *pRepo) {
isLast = false;
}
if (tsdbWriteBlockImpl(pRepo, pTable, pDFile, pDataCols, &block, isLast, true, ppBuf, ppCBuf) < 0) {
if (tsdbWriteBlockImpl(pRepo, pTable, pDFile,
isLast ? TSDB_COMPACT_SMAL_FILE(pComph) : TSDB_COMPACT_SMAD_FILE(pComph), pDataCols, &block,
isLast, true, ppBuf, ppCBuf, ppExBuf) < 0) {
return -1;
}
......@@ -526,5 +533,5 @@ static int tsdbCompactMeta(STsdbRepo *pRepo) {
}
return 0;
}
}
......@@ -36,6 +36,7 @@ static int tsdbComparTFILE(const void *arg1, const void *arg2);
static void tsdbScanAndTryFixDFilesHeader(STsdbRepo *pRepo, int32_t *nExpired);
static int tsdbProcessExpiredFS(STsdbRepo *pRepo);
static int tsdbCreateMeta(STsdbRepo *pRepo);
static int tsdbFetchTFileSet(STsdbRepo *pRepo, SArray **fArray);
// For backward compatibility
// ================== CURRENT file header info
......@@ -89,18 +90,33 @@ static int tsdbEncodeDFileSetArray(void **buf, SArray *pArray) {
return tlen;
}
static void *tsdbDecodeDFileSetArray(void *buf, SArray *pArray) {
static int tsdbDecodeDFileSetArray(void **originBuf, void *buf, SArray *pArray, SFSHeader *pSFSHeader) {
uint64_t nset;
SDFileSet dset;
dset.ver = TSDB_FSET_VER_0; // default value
taosArrayClear(pArray);
buf = taosDecodeFixedU64(buf, &nset);
if (pSFSHeader->version == TSDB_FS_VER_0) {
// record fver in new version of 'current' file
uint64_t extendedSize = pSFSHeader->len + nset * TSDB_FILE_MAX * sizeof(TSDB_FVER_TYPE);
if (taosTSizeof(*originBuf) < extendedSize) {
size_t ptrDistance = POINTER_DISTANCE(buf, *originBuf);
if (tsdbMakeRoom(originBuf, (size_t)extendedSize) < 0) {
terrno = TSDB_CODE_FS_OUT_OF_MEMORY;
return -1;
}
buf = POINTER_SHIFT(*originBuf, ptrDistance);
}
}
for (size_t i = 0; i < nset; i++) {
buf = tsdbDecodeDFileSet(buf, &dset);
buf = tsdbDecodeDFileSet(buf, &dset, pSFSHeader->version);
taosArrayPush(pArray, (void *)(&dset));
}
return buf;
return TSDB_CODE_SUCCESS;
}
static int tsdbEncodeFSStatus(void **buf, SFSStatus *pStatus) {
......@@ -114,15 +130,12 @@ static int tsdbEncodeFSStatus(void **buf, SFSStatus *pStatus) {
return tlen;
}
static void *tsdbDecodeFSStatus(void *buf, SFSStatus *pStatus) {
static int tsdbDecodeFSStatus(void **originBuf, void *buf, SFSStatus *pStatus, SFSHeader *pSFSHeader) {
tsdbResetFSStatus(pStatus);
pStatus->pmf = &(pStatus->mf);
buf = tsdbDecodeSMFile(buf, pStatus->pmf);
buf = tsdbDecodeDFileSetArray(buf, pStatus->df);
return buf;
return tsdbDecodeDFileSetArray(originBuf, buf, pStatus->df, pSFSHeader);
}
static SFSStatus *tsdbNewFSStatus(int maxFSet) {
......@@ -414,7 +427,7 @@ static int tsdbSaveFSStatus(SFSStatus *pStatus, int vid) {
return -1;
}
fsheader.version = TSDB_FS_VERSION;
fsheader.version = TSDB_LATEST_SFS_VER;
if (pStatus->pmf == NULL) {
ASSERT(taosArrayGetSize(pStatus->df) == 0);
fsheader.len = 0;
......@@ -689,7 +702,7 @@ static int tsdbOpenFSFromCurrent(STsdbRepo *pRepo) {
ptr = tsdbDecodeFSHeader(ptr, &fsheader);
ptr = tsdbDecodeFSMeta(ptr, &(pStatus->meta));
if (fsheader.version != TSDB_FS_VERSION) {
if (fsheader.version != TSDB_FS_VER_0) {
// TODO: handle file version change
}
......@@ -718,7 +731,9 @@ static int tsdbOpenFSFromCurrent(STsdbRepo *pRepo) {
}
ptr = buffer;
ptr = tsdbDecodeFSStatus(ptr, pStatus);
if (tsdbDecodeFSStatus(&buffer, ptr, pStatus, &fsheader) < 0) {
goto _err;
}
} else {
tsdbResetFSStatus(pStatus);
}
......@@ -752,7 +767,7 @@ static int tsdbScanAndTryFixFS(STsdbRepo *pRepo) {
SDFileSet *pSet = (SDFileSet *)taosArrayGet(pStatus->df, i);
if (tsdbScanAndTryFixDFileSet(pRepo, pSet) < 0) {
tsdbError("vgId:%d failed to fix MFile since %s", REPO_ID(pRepo), tstrerror(terrno));
tsdbError("vgId:%d failed to fix DFileSet since %s", REPO_ID(pRepo), tstrerror(terrno));
return -1;
}
}
......@@ -966,7 +981,7 @@ static bool tsdbIsTFileInFS(STsdbFS *pfs, const TFILE *pf) {
SDFileSet *pSet;
while ((pSet = tsdbFSIterNext(&fsiter))) {
for (TSDB_FILE_T ftype = 0; ftype < TSDB_FILE_MAX; ftype++) {
for (TSDB_FILE_T ftype = 0; ftype < tsdbGetNFiles(pSet); ftype++) {
SDFile *pDFile = TSDB_DFILE_IN_SET(pSet, ftype);
if (tfsIsSameFile(pf, TSDB_FILE_F(pDFile))) {
return true;
......@@ -1098,25 +1113,23 @@ static int tsdbRestoreMeta(STsdbRepo *pRepo) {
return 0;
}
static int tsdbRestoreDFileSet(STsdbRepo *pRepo) {
static int tsdbFetchTFileSet(STsdbRepo *pRepo, SArray **fArray) {
char dataDir[TSDB_FILENAME_LEN];
char bname[TSDB_FILENAME_LEN];
TDIR * tdir = NULL;
const TFILE *pf = NULL;
const char * pattern = "^v[0-9]+f[0-9]+\\.(head|data|last)(-ver[0-9]+)?$";
SArray * fArray = NULL;
const char * pattern = "^v[0-9]+f[0-9]+\\.(head|data|last|smad|smal)(-ver[0-9]+)?$";
regex_t regex;
STsdbFS * pfs = REPO_FS(pRepo);
tsdbGetDataDir(REPO_ID(pRepo), dataDir);
// Resource allocation and init
regcomp(&regex, pattern, REG_EXTENDED);
fArray = taosArrayInit(1024, sizeof(TFILE));
if (fArray == NULL) {
*fArray = taosArrayInit(1024, sizeof(TFILE));
if (*fArray == NULL) {
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
tsdbError("vgId:%d failed to restore DFileSet while open directory %s since %s", REPO_ID(pRepo), dataDir,
tsdbError("vgId:%d failed to fetch TFileSet while open directory %s since %s", REPO_ID(pRepo), dataDir,
tstrerror(terrno));
regfree(&regex);
return -1;
......@@ -1124,9 +1137,9 @@ static int tsdbRestoreDFileSet(STsdbRepo *pRepo) {
tdir = tfsOpendir(dataDir);
if (tdir == NULL) {
tsdbError("vgId:%d failed to restore DFileSet while open directory %s since %s", REPO_ID(pRepo), dataDir,
tsdbError("vgId:%d failed to fetch TFileSet while open directory %s since %s", REPO_ID(pRepo), dataDir,
tstrerror(terrno));
taosArrayDestroy(fArray);
taosArrayDestroy(*fArray);
regfree(&regex);
return -1;
}
......@@ -1136,10 +1149,10 @@ static int tsdbRestoreDFileSet(STsdbRepo *pRepo) {
int code = regexec(&regex, bname, 0, NULL, 0);
if (code == 0) {
if (taosArrayPush(fArray, (void *)pf) == NULL) {
if (taosArrayPush(*fArray, (void *)pf) == NULL) {
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
tfsClosedir(tdir);
taosArrayDestroy(fArray);
taosArrayDestroy(*fArray);
regfree(&regex);
return -1;
}
......@@ -1150,10 +1163,10 @@ static int tsdbRestoreDFileSet(STsdbRepo *pRepo) {
continue;
} else {
// Has other error
tsdbError("vgId:%d failed to restore DFileSet Array while run regexec since %s", REPO_ID(pRepo), strerror(code));
tsdbError("vgId:%d failed to fetch TFileSet Array while run regexec since %s", REPO_ID(pRepo), strerror(code));
terrno = TAOS_SYSTEM_ERROR(code);
tfsClosedir(tdir);
taosArrayDestroy(fArray);
taosArrayDestroy(*fArray);
regfree(&regex);
return -1;
}
......@@ -1163,101 +1176,173 @@ static int tsdbRestoreDFileSet(STsdbRepo *pRepo) {
regfree(&regex);
// Sort the array according to file name
taosArraySort(fArray, tsdbComparTFILE);
size_t index = 0;
// Loop to recover each file set
for (;;) {
if (index >= taosArrayGetSize(fArray)) {
break;
}
SDFileSet fset = {0};
taosArraySort(*fArray, tsdbComparTFILE);
return 0;
}
TSDB_FSET_SET_CLOSED(&fset);
// update the function if the DFileSet definition updates
static bool tsdbIsDFileSetValid(int nFiles) {
switch (nFiles) {
case TSDB_FILE_MIN:
case TSDB_FILE_MAX:
return true;
default:
return false;
}
}
// Loop to recover ONE fset
for (TSDB_FILE_T ftype = 0; ftype < TSDB_FILE_MAX; ftype++) {
SDFile *pDFile = TSDB_DFILE_IN_SET(&fset, ftype);
static int tsdbRestoreDFileSet(STsdbRepo *pRepo) {
const TFILE *pf = NULL;
SArray * fArray = NULL;
STsdbFS * pfs = REPO_FS(pRepo);
char dataDir[TSDB_FILENAME_LEN] = "\0";
size_t fArraySize = 0;
if (index >= taosArrayGetSize(fArray)) {
tsdbError("vgId:%d incomplete DFileSet, fid:%d", REPO_ID(pRepo), fset.fid);
taosArrayDestroy(fArray);
return -1;
}
tsdbGetDataDir(REPO_ID(pRepo), dataDir);
pf = taosArrayGet(fArray, index);
if (tsdbFetchTFileSet(pRepo, &fArray) < 0) {
tsdbError("vgId:%d failed to fetch TFileSet from %s to restore since %s", REPO_ID(pRepo), dataDir,
tstrerror(terrno));
return -1;
}
int tvid, tfid;
TSDB_FILE_T ttype;
uint32_t tversion;
char _bname[TSDB_FILENAME_LEN];
if ((fArraySize = taosArrayGetSize(fArray)) <= 0) {
taosArrayDestroy(fArray);
tsdbInfo("vgId:%d size of DFileSet from %s is %" PRIu32, REPO_ID(pRepo), dataDir, (uint32_t)fArraySize);
return 0;
}
tfsbasename(pf, _bname);
tsdbParseDFilename(_bname, &tvid, &tfid, &ttype, &tversion);
// Loop to recover each file set
SDFileSet fset = {0};
uint8_t nDFiles = 0;
bool isOneFSetFinish = true;
int lastFType = -1;
// one fileset ends when (1) the array ends or (2) encounter different fid
for (size_t index = 0; index < fArraySize; ++index) {
int tvid = -1, tfid = -1;
TSDB_FILE_T ttype = TSDB_FILE_MAX;
uint32_t tversion = -1;
char bname[TSDB_FILENAME_LEN] = "\0";
pf = taosArrayGet(fArray, index);
tfsbasename(pf, bname);
tsdbParseDFilename(bname, &tvid, &tfid, &ttype, &tversion);
ASSERT(tvid == REPO_ID(pRepo));
SDFile *pDFile = TSDB_DFILE_IN_SET(&fset, ttype);
if (tfid < pRepo->rtn.minFid) { // skip the file expired
continue;
}
if ((isOneFSetFinish == false) && (lastFType == ttype)) { // only fetch the 1st file with same fid and type.
continue;
}
ASSERT(tvid == REPO_ID(pRepo));
lastFType = ttype;
if (tfid < pRepo->rtn.minFid) { // skip file expired
++index;
continue;
}
if (ftype == 0) {
fset.fid = tfid;
if (index == 0) {
memset(&fset, 0, sizeof(SDFileSet));
TSDB_FSET_SET_CLOSED(&fset);
nDFiles = 1;
fset.fid = tfid;
pDFile->f = *pf;
isOneFSetFinish = false;
} else {
if (fset.fid == tfid) {
++nDFiles;
pDFile->f = *pf;
// (1) the array ends
if (index == fArraySize - 1) {
if (tsdbIsDFileSetValid(nDFiles)) {
tsdbInfo("vgId:%d DFileSet %d is fetched, nDFiles=%" PRIu8, REPO_ID(pRepo), fset.fid, nDFiles);
isOneFSetFinish = true;
} else {
// return error in case of removing uncomplete DFileSets
terrno = TSDB_CODE_TDB_INCOMPLETE_DFILESET;
tsdbError("vgId:%d incomplete DFileSet, fid:%d, nDFiles=%" PRIu8, REPO_ID(pRepo), fset.fid, nDFiles);
taosArrayDestroy(fArray);
return -1;
}
}
} else {
if (tfid != fset.fid) {
tsdbError("vgId:%d incomplete dFileSet, fid:%d", REPO_ID(pRepo), fset.fid);
// (2) encounter different fid
if (tsdbIsDFileSetValid(nDFiles)) {
tsdbInfo("vgId:%d DFileSet %d is fetched, nDFiles=%" PRIu8, REPO_ID(pRepo), fset.fid, nDFiles);
isOneFSetFinish = true;
} else {
// return error in case of removing uncomplete DFileSets
terrno = TSDB_CODE_TDB_INCOMPLETE_DFILESET;
tsdbError("vgId:%d incomplete DFileSet, fid:%d, nDFiles=%" PRIu8, REPO_ID(pRepo), fset.fid, nDFiles);
taosArrayDestroy(fArray);
return -1;
#if 0
// next FSet
memset(&fset, 0, sizeof(SDFileSet));
TSDB_FSET_SET_CLOSED(&fset);
nDFiles = 1;
fset.fid = tfid;
pDFile->f = *pf;
isOneFSetFinish = false;
continue;
#endif
}
}
}
if (ttype != ftype) {
tsdbError("vgId:%d incomplete dFileSet, fid:%d", REPO_ID(pRepo), fset.fid);
taosArrayDestroy(fArray);
return -1;
}
pDFile->f = *pf;
if (tsdbOpenDFile(pDFile, O_RDONLY) < 0) {
tsdbError("vgId:%d failed to open DFile %s since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pDFile), tstrerror(terrno));
taosArrayDestroy(fArray);
return -1;
}
if (tsdbLoadDFileHeader(pDFile, &(pDFile->info)) < 0) {
tsdbError("vgId:%d failed to load DFile %s header since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pDFile),
tstrerror(terrno));
taosArrayDestroy(fArray);
return -1;
}
if (tsdbForceKeepFile) {
struct stat tfstat;
if (isOneFSetFinish) {
for (TSDB_FILE_T ftype = 0; ftype < nDFiles; ++ftype) {
SDFile * pDFile1 = TSDB_DFILE_IN_SET(&fset, ftype);
if (tsdbOpenDFile(pDFile1, O_RDONLY) < 0) {
tsdbError("vgId:%d failed to open DFile %s since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pDFile1),
tstrerror(terrno));
taosArrayDestroy(fArray);
return -1;
}
// Get real file size
if (fstat(pDFile->fd, &tfstat) < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
if (tsdbLoadDFileHeader(pDFile1, &(pDFile1->info)) < 0) {
tsdbError("vgId:%d failed to load DFile %s header since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pDFile1),
tstrerror(terrno));
taosArrayDestroy(fArray);
return -1;
}
if (pDFile->info.size != tfstat.st_size) {
int64_t tfsize = pDFile->info.size;
pDFile->info.size = tfstat.st_size;
tsdbInfo("vgId:%d file %s header size is changed from %" PRId64 " to %" PRId64, REPO_ID(pRepo),
TSDB_FILE_FULL_NAME(pDFile), tfsize, pDFile->info.size);
if (tsdbForceKeepFile) {
struct stat tfstat;
// Get real file size
if (fstat(pDFile1->fd, &tfstat) < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
taosArrayDestroy(fArray);
return -1;
}
if (pDFile1->info.size != tfstat.st_size) {
int64_t tfsize = pDFile1->info.size;
pDFile1->info.size = tfstat.st_size;
tsdbInfo("vgId:%d file %s header size is changed from %" PRId64 " to %" PRId64, REPO_ID(pRepo),
TSDB_FILE_FULL_NAME(pDFile1), tfsize, pDFile1->info.size);
}
}
tsdbCloseDFile(pDFile1);
}
tsdbInfo("vgId:%d FSET %d is restored", REPO_ID(pRepo), fset.fid);
tsdbCloseDFile(pDFile);
index++;
}
// TODO: update the logic when TSDB_FSET_VER definition update.
if (nDFiles == TSDB_FILE_MIN) {
fset.ver = TSDB_FSET_VER_0;
} else {
fset.ver = TSDB_LATEST_FSET_VER;
}
taosArrayPush(pfs->cstatus->df, &fset);
tsdbInfo("vgId:%d FSET %d is restored", REPO_ID(pRepo), fset.fid);
taosArrayPush(pfs->cstatus->df, &fset);
// next FSet
memset(&fset, 0, sizeof(SDFileSet));
TSDB_FSET_SET_CLOSED(&fset);
nDFiles = 1;
fset.fid = tfid;
pDFile->f = *pf;
isOneFSetFinish = false;
}
}
// Resource release
......@@ -1312,7 +1397,13 @@ static int tsdbComparTFILE(const void *arg1, const void *arg2) {
} else if (ftype1 > ftype2) {
return 1;
} else {
return 0;
if (version1 < version2) {
return -1;
} else if (version1 > version2) {
return 1;
} else {
return 0;
}
}
}
}
......@@ -1335,7 +1426,7 @@ static void tsdbScanAndTryFixDFilesHeader(STsdbRepo *pRepo, int32_t *nExpired) {
continue;
}
for (TSDB_FILE_T ftype = 0; ftype < TSDB_FILE_MAX; ftype++) {
for (TSDB_FILE_T ftype = 0; ftype < tsdbGetNFiles(&fset); ftype++) {
SDFile *pDFile = TSDB_DFILE_IN_SET(&fset, ftype);
if ((tsdbLoadDFileHeader(pDFile, &info) < 0) || pDFile->info.size != info.size ||
......
......@@ -16,17 +16,19 @@
#include "tsdbint.h"
static const char *TSDB_FNAME_SUFFIX[] = {
"head", // TSDB_FILE_HEAD
"data", // TSDB_FILE_DATA
"last", // TSDB_FILE_LAST
"", // TSDB_FILE_MAX
"meta", // TSDB_FILE_META
"head", // TSDB_FILE_HEAD
"data", // TSDB_FILE_DATA
"last", // TSDB_FILE_LAST
"smad", // TSDB_FILE_SMA_DATA(Small Materialized Aggregate for .data File)
"smal", // TSDB_FILE_SMA_LAST(Small Materialized Aggregate for .last File)
"", // TSDB_FILE_MAX
"meta", // TSDB_FILE_META
};
static void tsdbGetFilename(int vid, int fid, uint32_t ver, TSDB_FILE_T ftype, char *fname);
static int tsdbRollBackMFile(SMFile *pMFile);
static int tsdbEncodeDFInfo(void **buf, SDFInfo *pInfo);
static void *tsdbDecodeDFInfo(void *buf, SDFInfo *pInfo);
static void *tsdbDecodeDFInfo(void *buf, SDFInfo *pInfo, TSDB_FVER_TYPE sfver);
static int tsdbRollBackDFile(SDFile *pDFile);
// ============== SMFile
......@@ -198,7 +200,7 @@ int tsdbScanAndTryFixMFile(STsdbRepo *pRepo) {
tsdbInitMFileEx(&mf, pMFile);
if (access(TSDB_FILE_FULL_NAME(pMFile), F_OK) != 0) {
tsdbError("vgId:%d meta file %s not exit, report to upper layer to fix it", REPO_ID(pRepo),
tsdbError("vgId:%d meta file %s not exist, report to upper layer to fix it", REPO_ID(pRepo),
TSDB_FILE_FULL_NAME(pMFile));
pRepo->state |= TSDB_STATE_BAD_META;
TSDB_FILE_SET_STATE(pMFile, TSDB_FILE_STATE_BAD);
......@@ -301,6 +303,7 @@ void tsdbInitDFile(SDFile *pDFile, SDiskID did, int vid, int fid, uint32_t ver,
memset(&(pDFile->info), 0, sizeof(pDFile->info));
pDFile->info.magic = TSDB_FILE_INIT_MAGIC;
pDFile->info.fver = tsdbGetDFSVersion(ftype);
tsdbGetFilename(vid, fid, ver, ftype, fname);
tfsInitFile(&(pDFile->f), did.level, did.id, fname);
......@@ -320,8 +323,8 @@ int tsdbEncodeSDFile(void **buf, SDFile *pDFile) {
return tlen;
}
void *tsdbDecodeSDFile(void *buf, SDFile *pDFile) {
buf = tsdbDecodeDFInfo(buf, &(pDFile->info));
void *tsdbDecodeSDFile(void *buf, SDFile *pDFile, uint32_t sfver) {
buf = tsdbDecodeDFInfo(buf, &(pDFile->info), sfver);
buf = tfsDecodeFile(buf, &(pDFile->f));
TSDB_FILE_SET_CLOSED(pDFile);
......@@ -339,8 +342,8 @@ static int tsdbEncodeSDFileEx(void **buf, SDFile *pDFile) {
static void *tsdbDecodeSDFileEx(void *buf, SDFile *pDFile) {
char *aname;
buf = tsdbDecodeDFInfo(buf, &(pDFile->info));
// The sync module would send DFileSet with latest verion.
buf = tsdbDecodeDFInfo(buf, &(pDFile->info), TSDB_LATEST_SFS_VER);
buf = taosDecodeString(buf, &aname);
strncpy(TSDB_FILE_FULL_NAME(pDFile), aname, TSDB_FILENAME_LEN);
TSDB_FILE_SET_CLOSED(pDFile);
......@@ -349,7 +352,7 @@ static void *tsdbDecodeSDFileEx(void *buf, SDFile *pDFile) {
return buf;
}
int tsdbCreateDFile(SDFile *pDFile, bool updateHeader) {
int tsdbCreateDFile(SDFile *pDFile, bool updateHeader, TSDB_FILE_T fType) {
ASSERT(pDFile->info.size == 0 && pDFile->info.magic == TSDB_FILE_INIT_MAGIC);
pDFile->fd = open(TSDB_FILE_FULL_NAME(pDFile), O_WRONLY | O_CREAT | O_TRUNC | O_BINARY, 0755);
......@@ -379,6 +382,7 @@ int tsdbCreateDFile(SDFile *pDFile, bool updateHeader) {
}
pDFile->info.size += TSDB_FILE_HEAD_SIZE;
pDFile->info.fver = tsdbGetDFSVersion(fType);
if (tsdbUpdateDFileHeader(pDFile) < 0) {
tsdbCloseDFile(pDFile);
......@@ -397,7 +401,6 @@ int tsdbUpdateDFileHeader(SDFile *pDFile) {
}
void *ptr = buf;
taosEncodeFixedU32(&ptr, TSDB_FS_VERSION);
tsdbEncodeDFInfo(&ptr, &(pDFile->info));
taosCalcChecksumAppend(0, (uint8_t *)buf, TSDB_FILE_HEAD_SIZE);
......@@ -410,7 +413,7 @@ int tsdbUpdateDFileHeader(SDFile *pDFile) {
int tsdbLoadDFileHeader(SDFile *pDFile, SDFInfo *pInfo) {
char buf[TSDB_FILE_HEAD_SIZE] = "\0";
uint32_t _version;
// uint32_t _version;
ASSERT(TSDB_FILE_OPENED(pDFile));
......@@ -428,8 +431,7 @@ int tsdbLoadDFileHeader(SDFile *pDFile, SDFInfo *pInfo) {
}
void *pBuf = buf;
pBuf = taosDecodeFixedU32(pBuf, &_version);
pBuf = tsdbDecodeDFInfo(pBuf, pInfo);
pBuf = tsdbDecodeDFInfo(pBuf, pInfo, TSDB_LATEST_FVER); // only make sure the parameter sfver > 0
return 0;
}
......@@ -440,7 +442,7 @@ static int tsdbScanAndTryFixDFile(STsdbRepo *pRepo, SDFile *pDFile) {
tsdbInitDFileEx(&df, pDFile);
if (access(TSDB_FILE_FULL_NAME(pDFile), F_OK) != 0) {
tsdbError("vgId:%d data file %s not exit, report to upper layer to fix it", REPO_ID(pRepo),
tsdbError("vgId:%d data file %s not exist, report to upper layer to fix it", REPO_ID(pRepo),
TSDB_FILE_FULL_NAME(pDFile));
pRepo->state |= TSDB_STATE_BAD_DATA;
TSDB_FILE_SET_STATE(pDFile, TSDB_FILE_STATE_BAD);
......@@ -487,7 +489,7 @@ static int tsdbScanAndTryFixDFile(STsdbRepo *pRepo, SDFile *pDFile) {
static int tsdbEncodeDFInfo(void **buf, SDFInfo *pInfo) {
int tlen = 0;
tlen += taosEncodeFixedU32(buf, pInfo->fver);
tlen += taosEncodeFixedU32(buf, pInfo->magic);
tlen += taosEncodeFixedU32(buf, pInfo->len);
tlen += taosEncodeFixedU32(buf, pInfo->totalBlocks);
......@@ -499,7 +501,12 @@ static int tsdbEncodeDFInfo(void **buf, SDFInfo *pInfo) {
return tlen;
}
static void *tsdbDecodeDFInfo(void *buf, SDFInfo *pInfo) {
static void *tsdbDecodeDFInfo(void *buf, SDFInfo *pInfo, TSDB_FVER_TYPE sfver) {
if (sfver > TSDB_FS_VER_0) {
buf = taosDecodeFixedU32(buf, &(pInfo->fver));
} else {
pInfo->fver = TSDB_FS_VER_0; // default value
}
buf = taosDecodeFixedU32(buf, &(pInfo->magic));
buf = taosDecodeFixedU32(buf, &(pInfo->len));
buf = taosDecodeFixedU32(buf, &(pInfo->totalBlocks));
......@@ -556,19 +563,22 @@ static int tsdbRollBackDFile(SDFile *pDFile) {
}
// ============== Operations on SDFileSet
void tsdbInitDFileSet(SDFileSet *pSet, SDiskID did, int vid, int fid, uint32_t ver) {
void tsdbInitDFileSet(SDFileSet *pSet, SDiskID did, int vid, int fid, uint32_t ver, uint16_t fsetVer) {
pSet->fid = fid;
pSet->state = 0;
pSet->ver = fsetVer;
for (TSDB_FILE_T ftype = 0; ftype < TSDB_FILE_MAX; ftype++) {
for (TSDB_FILE_T ftype = 0; ftype < tsdbGetNFiles(pSet); ftype++) {
SDFile *pDFile = TSDB_DFILE_IN_SET(pSet, ftype);
tsdbInitDFile(pDFile, did, vid, fid, ver, ftype);
}
}
void tsdbInitDFileSetEx(SDFileSet *pSet, SDFileSet *pOSet) {
ASSERT_TSDB_FSET_NFILES_VALID(pOSet);
pSet->fid = pOSet->fid;
for (TSDB_FILE_T ftype = 0; ftype < TSDB_FILE_MAX; ftype++) {
pSet->ver = pOSet->ver;
for (TSDB_FILE_T ftype = 0; ftype < tsdbGetNFiles(pSet); ftype++) {
tsdbInitDFileEx(TSDB_DFILE_IN_SET(pSet, ftype), TSDB_DFILE_IN_SET(pOSet, ftype));
}
}
......@@ -577,21 +587,28 @@ int tsdbEncodeDFileSet(void **buf, SDFileSet *pSet) {
int tlen = 0;
tlen += taosEncodeFixedI32(buf, pSet->fid);
for (TSDB_FILE_T ftype = 0; ftype < TSDB_FILE_MAX; ftype++) {
tlen += taosEncodeFixedU16(buf, pSet->ver);
for (TSDB_FILE_T ftype = 0; ftype < tsdbGetNFiles(pSet); ftype++) {
tlen += tsdbEncodeSDFile(buf, TSDB_DFILE_IN_SET(pSet, ftype));
}
return tlen;
}
void *tsdbDecodeDFileSet(void *buf, SDFileSet *pSet) {
void *tsdbDecodeDFileSet(void *buf, SDFileSet *pSet, uint32_t sfver) {
int32_t fid;
buf = taosDecodeFixedI32(buf, &(fid));
pSet->state = 0;
pSet->fid = fid;
for (TSDB_FILE_T ftype = 0; ftype < TSDB_FILE_MAX; ftype++) {
buf = tsdbDecodeSDFile(buf, TSDB_DFILE_IN_SET(pSet, ftype));
if (sfver > TSDB_FS_VER_0) {
buf = taosDecodeFixedU16(buf, &(pSet->ver));
}
ASSERT_TSDB_FSET_NFILES_VALID(pSet);
for (TSDB_FILE_T ftype = 0; ftype < tsdbGetNFiles(pSet); ftype++) {
buf = tsdbDecodeSDFile(buf, TSDB_DFILE_IN_SET(pSet, ftype), sfver);
}
return buf;
}
......@@ -600,7 +617,8 @@ int tsdbEncodeDFileSetEx(void **buf, SDFileSet *pSet) {
int tlen = 0;
tlen += taosEncodeFixedI32(buf, pSet->fid);
for (TSDB_FILE_T ftype = 0; ftype < TSDB_FILE_MAX; ftype++) {
tlen += taosEncodeFixedU16(buf, pSet->ver);
for (TSDB_FILE_T ftype = 0; ftype < tsdbGetNFiles(pSet); ftype++) {
tlen += tsdbEncodeSDFileEx(buf, TSDB_DFILE_IN_SET(pSet, ftype));
}
......@@ -611,17 +629,20 @@ void *tsdbDecodeDFileSetEx(void *buf, SDFileSet *pSet) {
int32_t fid;
buf = taosDecodeFixedI32(buf, &(fid));
buf = taosDecodeFixedU16(buf, &(pSet->ver));
pSet->fid = fid;
for (TSDB_FILE_T ftype = 0; ftype < TSDB_FILE_MAX; ftype++) {
for (TSDB_FILE_T ftype = 0; ftype < tsdbGetNFiles(pSet); ftype++) {
buf = tsdbDecodeSDFileEx(buf, TSDB_DFILE_IN_SET(pSet, ftype));
}
return buf;
}
int tsdbApplyDFileSetChange(SDFileSet *from, SDFileSet *to) {
for (TSDB_FILE_T ftype = 0; ftype < TSDB_FILE_MAX; ftype++) {
SDFile *pDFileFrom = (from) ? TSDB_DFILE_IN_SET(from, ftype) : NULL;
SDFile *pDFileTo = (to) ? TSDB_DFILE_IN_SET(to, ftype) : NULL;
uint8_t nFilesFrom = from ? tsdbGetNFiles(from) : 0;
uint8_t nFilesTo = to ? tsdbGetNFiles(to) : 0;
for (TSDB_FILE_T ftype = 0; ftype < MAX(nFilesFrom, nFilesTo); ftype++) {
SDFile *pDFileFrom = ftype < nFilesFrom ? TSDB_DFILE_IN_SET(from, ftype) : NULL;
SDFile *pDFileTo = ftype < nFilesTo ? TSDB_DFILE_IN_SET(to, ftype) : NULL;
if (tsdbApplyDFileChange(pDFileFrom, pDFileTo) < 0) {
return -1;
}
......@@ -631,8 +652,8 @@ int tsdbApplyDFileSetChange(SDFileSet *from, SDFileSet *to) {
}
int tsdbCreateDFileSet(SDFileSet *pSet, bool updateHeader) {
for (TSDB_FILE_T ftype = 0; ftype < TSDB_FILE_MAX; ftype++) {
if (tsdbCreateDFile(TSDB_DFILE_IN_SET(pSet, ftype), updateHeader) < 0) {
for (TSDB_FILE_T ftype = 0; ftype < tsdbGetNFiles(pSet); ftype++) {
if (tsdbCreateDFile(TSDB_DFILE_IN_SET(pSet, ftype), updateHeader, ftype) < 0) {
tsdbCloseDFileSet(pSet);
tsdbRemoveDFileSet(pSet);
return -1;
......@@ -643,7 +664,7 @@ int tsdbCreateDFileSet(SDFileSet *pSet, bool updateHeader) {
}
int tsdbUpdateDFileSetHeader(SDFileSet *pSet) {
for (TSDB_FILE_T ftype = 0; ftype < TSDB_FILE_MAX; ftype++) {
for (TSDB_FILE_T ftype = 0; ftype < tsdbGetNFiles(pSet); ftype++) {
if (tsdbUpdateDFileHeader(TSDB_DFILE_IN_SET(pSet, ftype)) < 0) {
return -1;
}
......@@ -652,7 +673,8 @@ int tsdbUpdateDFileSetHeader(SDFileSet *pSet) {
}
int tsdbScanAndTryFixDFileSet(STsdbRepo *pRepo, SDFileSet *pSet) {
for (TSDB_FILE_T ftype = 0; ftype < TSDB_FILE_MAX; ftype++) {
ASSERT_TSDB_FSET_NFILES_VALID(pSet);
for (TSDB_FILE_T ftype = 0; ftype < tsdbGetNFiles(pSet); ftype++) {
if (tsdbScanAndTryFixDFile(pRepo, TSDB_DFILE_IN_SET(pSet, ftype)) < 0) {
return -1;
}
......
......@@ -696,7 +696,7 @@ static int tsdbRestoreLastColumns(STsdbRepo *pRepo, STable *pTable, SReadH* pRea
tdInitDataRow(memRowDataBody(row), pSchema);
// first load block index info
if (tsdbLoadBlockInfo(pReadh, NULL) < 0) {
if (tsdbLoadBlockInfo(pReadh, NULL, NULL) < 0) {
err = -1;
goto out;
}
......@@ -730,9 +730,10 @@ static int tsdbRestoreLastColumns(STsdbRepo *pRepo, STable *pTable, SReadH* pRea
// file block with sub-blocks has no statistics data
if (pBlock->numOfSubBlocks <= 1) {
tsdbLoadBlockStatis(pReadh, pBlock);
tsdbGetBlockStatis(pReadh, pBlockStatis, (int)numColumns);
loadStatisData = true;
if (tsdbLoadBlockStatis(pReadh, pBlock) == TSDB_STATIS_OK) {
tsdbGetBlockStatis(pReadh, pBlockStatis, (int)numColumns, pBlock);
loadStatisData = true;
}
}
for (int16_t i = 0; i < numColumns && numColumns > pTable->restoreColumnNum; ++i) {
......@@ -798,7 +799,7 @@ out:
static int tsdbRestoreLastRow(STsdbRepo *pRepo, STable *pTable, SReadH* pReadh, SBlockIdx *pIdx) {
ASSERT(pTable->lastRow == NULL);
if (tsdbLoadBlockInfo(pReadh, NULL) < 0) {
if (tsdbLoadBlockInfo(pReadh, NULL, NULL) < 0) {
return -1;
}
......
......@@ -1054,21 +1054,10 @@ static int32_t loadBlockInfo(STsdbQueryHandle * pQueryHandle, int32_t index, int
return 0; // no data blocks in the file belongs to pCheckInfo->pTable
}
if (pCheckInfo->compSize < (int32_t)compIndex->len) {
assert(compIndex->len > 0);
assert(compIndex->len > 0);
char* t = realloc(pCheckInfo->pCompInfo, compIndex->len);
if (t == NULL) {
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
code = TSDB_CODE_TDB_OUT_OF_MEMORY;
return code;
}
pCheckInfo->pCompInfo = (SBlockInfo*)t;
pCheckInfo->compSize = compIndex->len;
}
if (tsdbLoadBlockInfo(&(pQueryHandle->rhelper), (void*)(pCheckInfo->pCompInfo)) < 0) {
if (tsdbLoadBlockInfo(&(pQueryHandle->rhelper), (void**)(&pCheckInfo->pCompInfo),
(uint32_t*)(&pCheckInfo->compSize)) < 0) {
return terrno;
}
SBlockInfo* pCompInfo = pCheckInfo->pCompInfo;
......@@ -3318,8 +3307,12 @@ int32_t tsdbRetrieveDataBlockStatisInfo(TsdbQueryHandleT* pQueryHandle, SDataSta
}
int64_t stime = taosGetTimestampUs();
if (tsdbLoadBlockStatis(&pHandle->rhelper, pBlockInfo->compBlock) < 0) {
int statisStatus = tsdbLoadBlockStatis(&pHandle->rhelper, pBlockInfo->compBlock);
if (statisStatus < TSDB_STATIS_OK) {
return terrno;
} else if (statisStatus > TSDB_STATIS_OK) {
*pBlockStatis = NULL;
return TSDB_CODE_SUCCESS;
}
int16_t* colIds = pHandle->defaultLoadColumn->pData;
......@@ -3330,7 +3323,7 @@ int32_t tsdbRetrieveDataBlockStatisInfo(TsdbQueryHandleT* pQueryHandle, SDataSta
pHandle->statis[i].colId = colIds[i];
}
tsdbGetBlockStatis(&pHandle->rhelper, pHandle->statis, (int)numOfCols);
tsdbGetBlockStatis(&pHandle->rhelper, pHandle->statis, (int)numOfCols, pBlockInfo->compBlock);
// always load the first primary timestamp column data
SDataStatis* pPrimaryColStatis = &pHandle->statis[0];
......
......@@ -25,6 +25,8 @@ static int tsdbCheckAndDecodeColumnData(SDataCol *pDataCol, void *content, int3
static int tsdbLoadBlockDataColsImpl(SReadH *pReadh, SBlock *pBlock, SDataCols *pDataCols, int16_t *colIds,
int numOfColIds);
static int tsdbLoadColData(SReadH *pReadh, SDFile *pDFile, SBlock *pBlock, SBlockCol *pBlockCol, SDataCol *pDataCol);
static int tsdbLoadBlockStatisFromDFile(SReadH *pReadh, SBlock *pBlock);
static int tsdbLoadBlockStatisFromAggr(SReadH *pReadh, SBlock *pBlock);
int tsdbInitReadH(SReadH *pReadh, STsdbRepo *pRepo) {
ASSERT(pReadh != NULL && pRepo != NULL);
......@@ -61,11 +63,12 @@ int tsdbInitReadH(SReadH *pReadh, STsdbRepo *pRepo) {
void tsdbDestroyReadH(SReadH *pReadh) {
if (pReadh == NULL) return;
pReadh->pExBuf = taosTZfree(pReadh->pExBuf);
pReadh->pCBuf = taosTZfree(pReadh->pCBuf);
pReadh->pBuf = taosTZfree(pReadh->pBuf);
pReadh->pDCols[0] = tdFreeDataCols(pReadh->pDCols[0]);
pReadh->pDCols[1] = tdFreeDataCols(pReadh->pDCols[1]);
pReadh->pAggrBlkData = taosTZfree(pReadh->pAggrBlkData);
pReadh->pBlkData = taosTZfree(pReadh->pBlkData);
pReadh->pBlkInfo = taosTZfree(pReadh->pBlkInfo);
pReadh->cidx = 0;
......@@ -198,6 +201,7 @@ int tsdbSetReadTable(SReadH *pReadh, STable *pTable) {
return 0;
}
#if 0
int tsdbLoadBlockInfo(SReadH *pReadh, void *pTarget) {
ASSERT(pReadh->pBlkIdx != NULL);
......@@ -241,6 +245,129 @@ int tsdbLoadBlockInfo(SReadH *pReadh, void *pTarget) {
return 0;
}
#endif
static FORCE_INLINE int32_t tsdbGetSBlockVer(int32_t fver) {
switch (fver) {
case TSDB_FS_VER_0:
return TSDB_SBLK_VER_0;
case TSDB_FS_VER_1:
return TSDB_SBLK_VER_1;
default:
return SBlockVerLatest;
}
}
static FORCE_INLINE size_t tsdbSizeOfSBlock(int32_t sBlkVer) {
switch (sBlkVer) {
case TSDB_SBLK_VER_0:
return sizeof(SBlockV0);
case TSDB_SBLK_VER_1:
return sizeof(SBlockV1);
default:
return sizeof(SBlock);
}
}
static int tsdbSBlkInfoRefactor(SDFile *pHeadf, SBlockInfo **pDstBlkInfo, SBlockIdx *pBlkIdx, uint32_t *dstBlkInfoLen) {
int sBlkVer = tsdbGetSBlockVer(pHeadf->info.fver);
if (sBlkVer > TSDB_SBLK_VER_0) {
*dstBlkInfoLen = pBlkIdx->len;
return TSDB_CODE_SUCCESS;
}
size_t originBlkSize = tsdbSizeOfSBlock(sBlkVer);
size_t nBlks = (pBlkIdx->len - sizeof(SBlockInfo)) / originBlkSize;
*dstBlkInfoLen = (uint32_t)(sizeof(SBlockInfo) + nBlks * sizeof(SBlock));
if (pBlkIdx->len == *dstBlkInfoLen) {
return TSDB_CODE_SUCCESS;
}
ASSERT(*dstBlkInfoLen >= pBlkIdx->len);
SBlockInfo *tmpBlkInfo = NULL;
if (tsdbMakeRoom((void **)(&tmpBlkInfo), *dstBlkInfoLen) < 0) return -1;
memset(tmpBlkInfo, 0, *dstBlkInfoLen); // the blkVer is set to 0
memcpy(tmpBlkInfo, *pDstBlkInfo, sizeof(SBlockInfo)); // copy header
uint32_t nSubBlks = 0;
for (int i = 0; i < nBlks; ++i) {
SBlock *tmpBlk = tmpBlkInfo->blocks + i;
memcpy(tmpBlk, POINTER_SHIFT((*pDstBlkInfo)->blocks, i * originBlkSize), originBlkSize);
if (i < pBlkIdx->numOfBlocks) { // super blocks
if (tmpBlk->numOfSubBlocks > 1) { // has sub blocks
tmpBlk->offset = sizeof(SBlockInfo) + (pBlkIdx->numOfBlocks + nSubBlks) * sizeof(SBlock);
nSubBlks += tmpBlk->numOfSubBlocks;
}
}
// TODO: update the fields if the SBlock definition change later
}
taosTZfree(*pDstBlkInfo);
*pDstBlkInfo = tmpBlkInfo;
return TSDB_CODE_SUCCESS;
}
int tsdbLoadBlockInfo(SReadH *pReadh, void **pTarget, uint32_t *extendedLen) {
ASSERT(pReadh->pBlkIdx != NULL);
SDFile * pHeadf = TSDB_READ_HEAD_FILE(pReadh);
SBlockIdx * pBlkIdx = pReadh->pBlkIdx;
if (tsdbSeekDFile(pHeadf, pBlkIdx->offset, SEEK_SET) < 0) {
tsdbError("vgId:%d failed to load SBlockInfo part while seek file %s since %s, offset:%u len:%u",
TSDB_READ_REPO_ID(pReadh), TSDB_FILE_FULL_NAME(pHeadf), tstrerror(terrno), pBlkIdx->offset, pBlkIdx->len);
return -1;
}
if (tsdbMakeRoom((void **)(&pReadh->pBlkInfo), pBlkIdx->len) < 0) return -1;
int64_t nread = tsdbReadDFile(pHeadf, (void *)(pReadh->pBlkInfo), pBlkIdx->len);
if (nread < 0) {
tsdbError("vgId:%d failed to load SBlockInfo part while read file %s since %s, offset:%u len :%u",
TSDB_READ_REPO_ID(pReadh), TSDB_FILE_FULL_NAME(pHeadf), tstrerror(terrno), pBlkIdx->offset, pBlkIdx->len);
return -1;
}
if (nread < pBlkIdx->len) {
terrno = TSDB_CODE_TDB_FILE_CORRUPTED;
tsdbError("vgId:%d SBlockInfo part in file %s is corrupted, offset:%u expected bytes:%u read bytes:%" PRId64,
TSDB_READ_REPO_ID(pReadh), TSDB_FILE_FULL_NAME(pHeadf), pBlkIdx->offset, pBlkIdx->len, nread);
return -1;
}
if (!taosCheckChecksumWhole((uint8_t *)(pReadh->pBlkInfo), pBlkIdx->len)) {
terrno = TSDB_CODE_TDB_FILE_CORRUPTED;
tsdbError("vgId:%d SBlockInfo part in file %s is corrupted since wrong checksum, offset:%u len :%u",
TSDB_READ_REPO_ID(pReadh), TSDB_FILE_FULL_NAME(pHeadf), pBlkIdx->offset, pBlkIdx->len);
return -1;
}
ASSERT(pBlkIdx->tid == pReadh->pBlkInfo->tid && pBlkIdx->uid == pReadh->pBlkInfo->uid);
uint32_t dstBlkInfoLen = 0;
if (tsdbSBlkInfoRefactor(pHeadf, &(pReadh->pBlkInfo), pBlkIdx, &dstBlkInfoLen) < 0) {
return -1;
}
if (extendedLen != NULL) {
if (pTarget != NULL) {
if (*extendedLen < dstBlkInfoLen) {
char *t = realloc(*pTarget, dstBlkInfoLen);
if (t == NULL) {
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
return -1;
}
*pTarget = t;
}
memcpy(*pTarget, (void *)(pReadh->pBlkInfo), dstBlkInfoLen);
}
*extendedLen = dstBlkInfoLen;
}
return TSDB_CODE_SUCCESS;
}
int tsdbLoadBlockData(SReadH *pReadh, SBlock *pBlock, SBlockInfo *pBlkInfo) {
ASSERT(pBlock->numOfSubBlocks > 0);
......@@ -296,18 +423,15 @@ int tsdbLoadBlockDataCols(SReadH *pReadh, SBlock *pBlock, SBlockInfo *pBlkInfo,
return 0;
}
int tsdbLoadBlockStatis(SReadH *pReadh, SBlock *pBlock) {
ASSERT(pBlock->numOfSubBlocks <= 1);
static int tsdbLoadBlockStatisFromDFile(SReadH *pReadh, SBlock *pBlock) {
SDFile *pDFile = (pBlock->last) ? TSDB_READ_LAST_FILE(pReadh) : TSDB_READ_DATA_FILE(pReadh);
if (tsdbSeekDFile(pDFile, pBlock->offset, SEEK_SET) < 0) {
tsdbError("vgId:%d failed to load block statis part while seek file %s to offset %" PRId64 " since %s",
TSDB_READ_REPO_ID(pReadh), TSDB_FILE_FULL_NAME(pDFile), (int64_t)pBlock->offset, tstrerror(terrno));
return -1;
}
size_t size = TSDB_BLOCK_STATIS_SIZE(pBlock->numOfCols);
size_t size = tsdbBlockStatisSize(pBlock->numOfCols, (uint32_t)pBlock->blkVer);
if (tsdbMakeRoom((void **)(&(pReadh->pBlkData)), size) < 0) return -1;
int64_t nread = tsdbReadDFile(pDFile, (void *)(pReadh->pBlkData), size);
......@@ -331,10 +455,66 @@ int tsdbLoadBlockStatis(SReadH *pReadh, SBlock *pBlock) {
TSDB_READ_REPO_ID(pReadh), TSDB_FILE_FULL_NAME(pDFile), (int64_t)pBlock->offset, size);
return -1;
}
return 0;
}
static int tsdbLoadBlockStatisFromAggr(SReadH *pReadh, SBlock *pBlock) {
ASSERT((pBlock->blkVer > TSDB_SBLK_VER_0) && (pBlock->aggrStat)); // TODO: remove after pass all the test
SDFile *pDFileAggr = pBlock->last ? TSDB_READ_SMAL_FILE(pReadh) : TSDB_READ_SMAD_FILE(pReadh);
if (tsdbSeekDFile(pDFileAggr, pBlock->aggrOffset, SEEK_SET) < 0) {
tsdbError("vgId:%d failed to load block aggr part while seek file %s to offset %" PRIu64 " since %s",
TSDB_READ_REPO_ID(pReadh), TSDB_FILE_FULL_NAME(pDFileAggr), (uint64_t)pBlock->aggrOffset,
tstrerror(terrno));
return -1;
}
size_t sizeAggr = tsdbBlockAggrSize(pBlock->numOfCols, (uint32_t)pBlock->blkVer);
if (tsdbMakeRoom((void **)(&(pReadh->pAggrBlkData)), sizeAggr) < 0) return -1;
int64_t nreadAggr = tsdbReadDFile(pDFileAggr, (void *)(pReadh->pAggrBlkData), sizeAggr);
if (nreadAggr < 0) {
tsdbError("vgId:%d failed to load block aggr part while read file %s since %s, offset:%" PRIu64 " len :%" PRIzu,
TSDB_READ_REPO_ID(pReadh), TSDB_FILE_FULL_NAME(pDFileAggr), tstrerror(terrno),
(uint64_t)pBlock->aggrOffset, sizeAggr);
return -1;
}
if (nreadAggr < sizeAggr) {
terrno = TSDB_CODE_TDB_FILE_CORRUPTED;
tsdbError("vgId:%d block aggr part in file %s is corrupted, offset:%" PRIu64 " expected bytes:%" PRIzu
" read bytes: %" PRId64,
TSDB_READ_REPO_ID(pReadh), TSDB_FILE_FULL_NAME(pDFileAggr), (uint64_t)pBlock->aggrOffset, sizeAggr,
nreadAggr);
return -1;
}
if (!taosCheckChecksumWhole((uint8_t *)(pReadh->pAggrBlkData), (uint32_t)sizeAggr)) {
terrno = TSDB_CODE_TDB_FILE_CORRUPTED;
tsdbError("vgId:%d block aggr part in file %s is corrupted since wrong checksum, offset:%" PRIu64 " len :%" PRIzu,
TSDB_READ_REPO_ID(pReadh), TSDB_FILE_FULL_NAME(pDFileAggr), (uint64_t)pBlock->aggrOffset, sizeAggr);
return -1;
}
return 0;
}
int tsdbLoadBlockStatis(SReadH *pReadh, SBlock *pBlock) {
ASSERT(pBlock->numOfSubBlocks <= 1);
if (pBlock->blkVer > TSDB_SBLK_VER_0) {
if (pBlock->aggrStat) {
return tsdbLoadBlockStatisFromAggr(pReadh, pBlock);
}
return TSDB_STATIS_NONE;
}
return tsdbLoadBlockStatisFromDFile(pReadh, pBlock);
}
int tsdbLoadBlockOffset(SReadH *pReadh, SBlock *pBlock) {
ASSERT(pBlock->numOfSubBlocks <= 1);
return tsdbLoadBlockStatisFromDFile(pReadh, pBlock);
}
int tsdbEncodeSBlockIdx(void **buf, SBlockIdx *pIdx) {
int tlen = 0;
......@@ -369,30 +549,58 @@ void *tsdbDecodeSBlockIdx(void *buf, SBlockIdx *pIdx) {
return buf;
}
void tsdbGetBlockStatis(SReadH *pReadh, SDataStatis *pStatis, int numOfCols) {
SBlockData *pBlockData = pReadh->pBlkData;
void tsdbGetBlockStatis(SReadH *pReadh, SDataStatis *pStatis, int numOfCols, SBlock *pBlock) {
if (pBlock->blkVer == TSDB_SBLK_VER_0) {
SBlockData *pBlockData = pReadh->pBlkData;
for (int i = 0, j = 0; i < numOfCols;) {
if (j >= pBlockData->numOfCols) {
pStatis[i].numOfNull = -1;
i++;
continue;
for (int i = 0, j = 0; i < numOfCols;) {
if (j >= pBlockData->numOfCols) {
pStatis[i].numOfNull = -1;
i++;
continue;
}
SBlockColV0 *pSBlkCol = ((SBlockColV0 *)(pBlockData->cols)) + j;
if (pStatis[i].colId == pSBlkCol->colId) {
pStatis[i].sum = pSBlkCol->sum;
pStatis[i].max = pSBlkCol->max;
pStatis[i].min = pSBlkCol->min;
pStatis[i].maxIndex = pSBlkCol->maxIndex;
pStatis[i].minIndex = pSBlkCol->minIndex;
pStatis[i].numOfNull = pSBlkCol->numOfNull;
i++;
j++;
} else if (pStatis[i].colId < pSBlkCol->colId) {
pStatis[i].numOfNull = -1;
i++;
} else {
j++;
}
}
} else if (pBlock->aggrStat) {
SAggrBlkData *pAggrBlkData = pReadh->pAggrBlkData;
if (pStatis[i].colId == pBlockData->cols[j].colId) {
pStatis[i].sum = pBlockData->cols[j].sum;
pStatis[i].max = pBlockData->cols[j].max;
pStatis[i].min = pBlockData->cols[j].min;
pStatis[i].maxIndex = pBlockData->cols[j].maxIndex;
pStatis[i].minIndex = pBlockData->cols[j].minIndex;
pStatis[i].numOfNull = pBlockData->cols[j].numOfNull;
i++;
j++;
} else if (pStatis[i].colId < pBlockData->cols[j].colId) {
pStatis[i].numOfNull = -1;
i++;
} else {
j++;
for (int i = 0, j = 0; i < numOfCols;) {
if (j >= pAggrBlkData->numOfCols) {
pStatis[i].numOfNull = -1;
i++;
continue;
}
SAggrBlkCol *pAggrBlkCol = ((SAggrBlkCol *)(pAggrBlkData->cols)) + j;
if (pStatis[i].colId == pAggrBlkCol->colId) {
pStatis[i].sum = pAggrBlkCol->sum;
pStatis[i].max = pAggrBlkCol->max;
pStatis[i].min = pAggrBlkCol->min;
pStatis[i].maxIndex = pAggrBlkCol->maxIndex;
pStatis[i].minIndex = pAggrBlkCol->minIndex;
pStatis[i].numOfNull = pAggrBlkCol->numOfNull;
i++;
j++;
} else if (pStatis[i].colId < pAggrBlkCol->colId) {
pStatis[i].numOfNull = -1;
i++;
} else {
j++;
}
}
}
}
......@@ -443,7 +651,7 @@ static int tsdbLoadBlockDataImpl(SReadH *pReadh, SBlock *pBlock, SDataCols *pDat
return -1;
}
int32_t tsize = TSDB_BLOCK_STATIS_SIZE(pBlock->numOfCols);
int32_t tsize = (int32_t)tsdbBlockStatisSize(pBlock->numOfCols, (uint32_t)pBlock->blkVer);
if (!taosCheckChecksumWhole((uint8_t *)TSDB_READ_BUF(pReadh), tsize)) {
terrno = TSDB_CODE_TDB_FILE_CORRUPTED;
tsdbError("vgId:%d block statis part in file %s is corrupted since wrong checksum, offset:%" PRId64 " len :%d",
......@@ -459,6 +667,8 @@ static int tsdbLoadBlockDataImpl(SReadH *pReadh, SBlock *pBlock, SDataCols *pDat
// Recover the data
int ccol = 0; // loop iter for SBlockCol object
int dcol = 0; // loop iter for SDataCols object
SBlockCol blockCol = {0};
SBlockCol *pBlockCol = &blockCol;
while (dcol < pDataCols->numOfCols) {
SDataCol *pDataCol = &(pDataCols->cols[dcol]);
if (dcol != 0 && ccol >= pBlockData->numOfCols) {
......@@ -472,8 +682,9 @@ static int tsdbLoadBlockDataImpl(SReadH *pReadh, SBlock *pBlock, SDataCols *pDat
uint32_t toffset = TSDB_KEY_COL_OFFSET;
int32_t tlen = pBlock->keyLen;
if (dcol != 0) {
SBlockCol *pBlockCol = &(pBlockData->cols[ccol]);
tsdbGetSBlockCol(pBlock, &pBlockCol, pBlockData->cols, ccol);
tcolId = pBlockCol->colId;
toffset = tsdbGetBlockColOffset(pBlockCol);
tlen = pBlockCol->len;
......@@ -555,7 +766,7 @@ static int tsdbLoadBlockDataColsImpl(SReadH *pReadh, SBlock *pBlock, SDataCols *
tdResetDataCols(pDataCols);
// If only load timestamp column, no need to load SBlockData part
if (numOfColIds > 1 && tsdbLoadBlockStatis(pReadh, pBlock) < 0) return -1;
if (numOfColIds > 1 && tsdbLoadBlockOffset(pReadh, pBlock) < 0) return -1;
pDataCols->numOfRows = pBlock->numOfRows;
......@@ -597,7 +808,9 @@ static int tsdbLoadBlockDataColsImpl(SReadH *pReadh, SBlock *pBlock, SDataCols *
break;
}
pBlockCol = &(pReadh->pBlkData->cols[ccol]);
pBlockCol = &blockCol;
tsdbGetSBlockCol(pBlock, &pBlockCol, pReadh->pBlkData->cols, ccol);
if (pBlockCol->colId > colId) {
pBlockCol = NULL;
break;
......@@ -631,7 +844,8 @@ static int tsdbLoadColData(SReadH *pReadh, SDFile *pDFile, SBlock *pBlock, SBloc
if (tsdbMakeRoom((void **)(&TSDB_READ_BUF(pReadh)), pBlockCol->len) < 0) return -1;
if (tsdbMakeRoom((void **)(&TSDB_READ_COMP_BUF(pReadh)), tsize) < 0) return -1;
int64_t offset = pBlock->offset + TSDB_BLOCK_STATIS_SIZE(pBlock->numOfCols) + tsdbGetBlockColOffset(pBlockCol);
int64_t offset = pBlock->offset + tsdbBlockStatisSize(pBlock->numOfCols, (uint32_t)pBlock->blkVer) +
tsdbGetBlockColOffset(pBlockCol);
if (tsdbSeekDFile(pDFile, offset, SEEK_SET) < 0) {
tsdbError("vgId:%d failed to load block column data while seek file %s to offset %" PRId64 " since %s",
TSDB_READ_REPO_ID(pReadh), TSDB_FILE_FULL_NAME(pDFile), offset, tstrerror(terrno));
......
......@@ -466,7 +466,7 @@ static int32_t tsdbSyncRecvDFileSetArray(SSyncH *pSynch) {
return -1;
}
tsdbInitDFileSet(&fset, did, REPO_ID(pRepo), pSynch->pdf->fid, FS_TXN_VERSION(pfs));
tsdbInitDFileSet(&fset, did, REPO_ID(pRepo), pSynch->pdf->fid, FS_TXN_VERSION(pfs), pSynch->pdf->ver);
// Create new FSET
if (tsdbCreateDFileSet(&fset, false) < 0) {
......@@ -474,7 +474,7 @@ static int32_t tsdbSyncRecvDFileSetArray(SSyncH *pSynch) {
return -1;
}
for (TSDB_FILE_T ftype = 0; ftype < TSDB_FILE_MAX; ftype++) {
for (TSDB_FILE_T ftype = 0; ftype < tsdbGetNFiles(pSynch->pdf); ftype++) {
SDFile *pDFile = TSDB_DFILE_IN_SET(&fset, ftype); // local file
SDFile *pRDFile = TSDB_DFILE_IN_SET(pSynch->pdf, ftype); // remote file
......@@ -550,7 +550,10 @@ static int32_t tsdbSyncRecvDFileSetArray(SSyncH *pSynch) {
}
static bool tsdbIsTowFSetSame(SDFileSet *pSet1, SDFileSet *pSet2) {
for (TSDB_FILE_T ftype = 0; ftype < TSDB_FILE_MAX; ftype++) {
if (pSet1->ver != pSet2->ver) {
return false;
}
for (TSDB_FILE_T ftype = 0; ftype < tsdbGetNFiles(pSet1); ftype++) {
SDFile *pDFile1 = TSDB_DFILE_IN_SET(pSet1, ftype);
SDFile *pDFile2 = TSDB_DFILE_IN_SET(pSet2, ftype);
......@@ -592,7 +595,7 @@ static int32_t tsdbSyncSendDFileSet(SSyncH *pSynch, SDFileSet *pSet) {
if (toSend) {
tsdbInfo("vgId:%d, fileset:%d will be sent", REPO_ID(pRepo), pSet->fid);
for (TSDB_FILE_T ftype = 0; ftype < TSDB_FILE_MAX; ftype++) {
for (TSDB_FILE_T ftype = 0; ftype < tsdbGetNFiles(pSet); ftype++) {
SDFile df = *TSDB_DFILE_IN_SET(pSet, ftype);
if (tsdbOpenDFile(&df, O_RDONLY) < 0) {
......
......@@ -278,6 +278,8 @@ TAOS_DEFINE_ERROR(TSDB_CODE_TDB_IVD_CREATE_TABLE_INFO, "Invalid information t
TAOS_DEFINE_ERROR(TSDB_CODE_TDB_NO_AVAIL_DISK, "No available disk")
TAOS_DEFINE_ERROR(TSDB_CODE_TDB_MESSED_MSG, "TSDB messed message")
TAOS_DEFINE_ERROR(TSDB_CODE_TDB_IVLD_TAG_VAL, "TSDB invalid tag value")
TAOS_DEFINE_ERROR(TSDB_CODE_TDB_NO_CACHE_LAST_ROW, "TSDB no cache last row data")
TAOS_DEFINE_ERROR(TSDB_CODE_TDB_INCOMPLETE_DFILESET, "Incomplete DFileSet")
// query
TAOS_DEFINE_ERROR(TSDB_CODE_QRY_INVALID_QHANDLE, "Invalid handle")
......
......@@ -95,7 +95,7 @@ int main(int argc, char *argv[]) {
exit(1);
}
printf("presss any key to exit\n");
printf("press any key to exit\n");
getchar();
taos_close_stream(pStream);
......
......@@ -24,7 +24,7 @@ class TDTestCase:
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), logSql)
self.ts = 1500000000000
self.ts = 1600000000000
self.num = 10
def run(self):
......@@ -84,6 +84,26 @@ class TDTestCase:
tdSql.query("select 'dcs' as options from stb where col > 200 limit 1 union all select 'aaa' as options from stb limit 10")
tdSql.checkData(0, 0, 'aaa')
# https://jira.taosdata.com:18080/browse/TS-444
tdLog.info("test case for TS-444")
tdSql.query("select count(*) as count, loc from st where ts between 1600000000000 and 1600000000010 group by loc")
tdSql.checkRows(6)
tdSql.query("select count(*) as count, loc from st where ts between 1600000000020 and 1600000000030 group by loc")
tdSql.checkRows(0)
tdSql.query(''' select count(*) as count, loc from st where ts between 1600000000000 and 1600000000010 group by loc
union all
select count(*) as count, loc from st where ts between 1600000000020 and 1600000000030 group by loc''')
tdSql.checkRows(6)
tdSql.query(''' select count(*) as count, loc from st where ts between 1600000000020 and 1600000000030 group by loc
union all
select count(*) as count, loc from st where ts between 1600000000000 and 1600000000010 group by loc''')
tdSql.checkRows(6)
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
......
......@@ -52,7 +52,7 @@ class TDTestCase:
tdLog.info("taosd found in %s" % buildPath)
binPath = buildPath+ "/build/bin/"
os.system("rm -rf table/create1.py.sql")
os.system("rm -rf table/create.py.sql")
tdSql.prepare()
print("==============step1")
......
......@@ -17,7 +17,7 @@ typedef struct SUdfInit{
void abs_max(char* data, short itype, short ibytes, int numOfRows, long long* ts, char* dataOutput, char* interBuf, char* tsOutput,
int* numOfOutput, short otype, short obytes, SUdfInit* buf) {
int i;
int r = 0;
long r = 0;
printf("abs_max input data:%p, type:%d, rows:%d, ts:%p,%lld, dataoutput:%p, tsOutput:%p, numOfOutput:%p, buf:%p\n", data, itype, numOfRows, ts, *ts, dataOutput, tsOutput, numOfOutput, buf);
if (itype == 5) {
r=*(long *)dataOutput;
......@@ -29,7 +29,7 @@ void abs_max(char* data, short itype, short ibytes, int numOfRows, long long* ts
}
*numOfOutput=1;
long v = abs(*((long *)data + i));
long v = labs(*((long *)data + i));
if (v > r) {
r = v;
}
......
......@@ -170,8 +170,8 @@ if $system_content != 0 then
endi
system_content ls ../../../sim/dnode3/data/vnode/vnode2/tsdb/data/ -l | grep "^-" | wc -l | tr -d '\n'
print ---->dnode3 data files: $system_content expect: 3
if $system_content != 3 then
print ---->dnode3 data files: $system_content expect: 5
if $system_content != 5 then
return -1
endi
......@@ -408,27 +408,4 @@ sql select count(*) from $stb
print data00 $data00
if $data00 != $totalRows then
return -1
endi
endi
\ No newline at end of file
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册