Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
小楼昨夜-听风雨
TDengine
提交
3bf73df9
T
TDengine
项目概览
小楼昨夜-听风雨
/
TDengine
与 Fork 源项目一致
Fork自
taosdata / TDengine
通知
1
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
T
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
未验证
提交
3bf73df9
编写于
2月 03, 2021
作者:
S
Shengliang Guan
提交者:
GitHub
2月 03, 2021
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #5115 from taosdata/feature/TD-1925_new
Feature/td 1925 new
上级
92dfa470
829cf222
变更
19
隐藏空白更改
内联
并排
Showing
19 changed file
with
170 addition
and
79 deletion
+170
-79
deps/MsvcLibX/include/msvcUnistd.h
deps/MsvcLibX/include/msvcUnistd.h
+3
-2
src/common/src/tglobal.c
src/common/src/tglobal.c
+1
-1
src/os/inc/osDef.h
src/os/inc/osDef.h
+0
-4
src/os/inc/osWindows.h
src/os/inc/osWindows.h
+2
-10
src/os/src/detail/osFile.c
src/os/src/detail/osFile.c
+2
-0
src/os/src/windows/wString.c
src/os/src/windows/wString.c
+0
-12
src/os/src/windows/wWordexp.c
src/os/src/windows/wWordexp.c
+0
-4
src/tfs/src/tfs.c
src/tfs/src/tfs.c
+1
-1
src/tsdb/inc/tsdbFile.h
src/tsdb/inc/tsdbFile.h
+19
-1
src/tsdb/src/tsdbCommit.c
src/tsdb/src/tsdbCommit.c
+79
-20
src/tsdb/src/tsdbFS.c
src/tsdb/src/tsdbFS.c
+9
-1
src/tsdb/src/tsdbFile.c
src/tsdb/src/tsdbFile.c
+8
-2
src/tsdb/src/tsdbMemTable.c
src/tsdb/src/tsdbMemTable.c
+4
-2
src/tsdb/src/tsdbSync.c
src/tsdb/src/tsdbSync.c
+7
-2
src/util/src/tconfig.c
src/util/src/tconfig.c
+5
-0
src/vnode/src/vnodeCfg.c
src/vnode/src/vnodeCfg.c
+11
-0
tests/pytest/insert/retentionpolicy.py
tests/pytest/insert/retentionpolicy.py
+8
-6
tests/pytest/pytest_2.sh
tests/pytest/pytest_2.sh
+8
-8
tests/script/wtest.bat
tests/script/wtest.bat
+3
-3
未找到文件。
deps/MsvcLibX/include/msvcUnistd.h
浏览文件 @
3bf73df9
...
@@ -89,11 +89,12 @@ pid_t getppid(void); /* Get parent PID */
...
@@ -89,11 +89,12 @@ pid_t getppid(void); /* Get parent PID */
/* Path management */
/* Path management */
#if defined(_WIN32)
#if defined(_WIN32)
#if defined(_UTF8_SOURCE) || defined(_BSD_SOURCE) || defined(_GNU_SOURCE)
#define realpath realpathU
#define realpath realpathU
#if defined(_UTF8_SOURCE) || defined(_BSD_SOURCE) || defined(_GNU_SOURCE)
// #define realpath realpathU
#define CompactPath CompactPathU
#define CompactPath CompactPathU
#else
/* _ANSI_SOURCE */
#else
/* _ANSI_SOURCE */
#define realpath realpathA
//
#define realpath realpathA
#define CompactPath CompactPathA
#define CompactPath CompactPathA
#endif
#endif
#endif
/* defined(_WIN32) */
#endif
/* defined(_WIN32) */
...
...
src/common/src/tglobal.c
浏览文件 @
3bf73df9
...
@@ -59,7 +59,6 @@ char tsLocale[TSDB_LOCALE_LEN] = {0};
...
@@ -59,7 +59,6 @@ char tsLocale[TSDB_LOCALE_LEN] = {0};
char
tsCharset
[
TSDB_LOCALE_LEN
]
=
{
0
};
// default encode string
char
tsCharset
[
TSDB_LOCALE_LEN
]
=
{
0
};
// default encode string
int8_t
tsEnableCoreFile
=
0
;
int8_t
tsEnableCoreFile
=
0
;
int32_t
tsMaxBinaryDisplayWidth
=
30
;
int32_t
tsMaxBinaryDisplayWidth
=
30
;
char
tsTempDir
[
TSDB_FILENAME_LEN
]
=
"/tmp/"
;
/*
/*
* denote if the server needs to compress response message at the application layer to client, including query rsp,
* denote if the server needs to compress response message at the application layer to client, including query rsp,
...
@@ -182,6 +181,7 @@ char tsDnodeDir[TSDB_FILENAME_LEN] = {0};
...
@@ -182,6 +181,7 @@ char tsDnodeDir[TSDB_FILENAME_LEN] = {0};
char
tsMnodeDir
[
TSDB_FILENAME_LEN
]
=
{
0
};
char
tsMnodeDir
[
TSDB_FILENAME_LEN
]
=
{
0
};
char
tsDataDir
[
TSDB_FILENAME_LEN
]
=
{
0
};
char
tsDataDir
[
TSDB_FILENAME_LEN
]
=
{
0
};
char
tsScriptDir
[
TSDB_FILENAME_LEN
]
=
{
0
};
char
tsScriptDir
[
TSDB_FILENAME_LEN
]
=
{
0
};
char
tsTempDir
[
TSDB_FILENAME_LEN
]
=
"/tmp/"
;
int32_t
tsDiskCfgNum
=
0
;
int32_t
tsDiskCfgNum
=
0
;
...
...
src/os/inc/osDef.h
浏览文件 @
3bf73df9
...
@@ -26,10 +26,6 @@ extern "C" {
...
@@ -26,10 +26,6 @@ extern "C" {
#endif
#endif
#endif
#endif
#ifndef STDERR_FILENO
#define STDERR_FILENO (2)
#endif
#define FD_VALID(x) ((x) > STDERR_FILENO)
#define FD_VALID(x) ((x) > STDERR_FILENO)
#define FD_INITIALIZER ((int32_t)-1)
#define FD_INITIALIZER ((int32_t)-1)
...
...
src/os/inc/osWindows.h
浏览文件 @
3bf73df9
...
@@ -46,6 +46,8 @@
...
@@ -46,6 +46,8 @@
#include "msvcFcntl.h"
#include "msvcFcntl.h"
#include "msvcLibgen.h"
#include "msvcLibgen.h"
#include "msvcStdio.h"
#include "msvcStdio.h"
#include "msvcUnistd.h"
#include "msvcLibgen.h"
#include "sys/msvcStat.h"
#include "sys/msvcStat.h"
#include "sys/msvcTypes.h"
#include "sys/msvcTypes.h"
...
@@ -144,7 +146,6 @@ typedef int (*__compar_fn_t)(const void *, const void *);
...
@@ -144,7 +146,6 @@ typedef int (*__compar_fn_t)(const void *, const void *);
#define in_addr_t unsigned long
#define in_addr_t unsigned long
#define socklen_t int
#define socklen_t int
#define htobe64 htonll
#define htobe64 htonll
#define getpid _getpid
struct
tm
*
localtime_r
(
const
time_t
*
timep
,
struct
tm
*
result
);
struct
tm
*
localtime_r
(
const
time_t
*
timep
,
struct
tm
*
result
);
char
*
strptime
(
const
char
*
buf
,
const
char
*
fmt
,
struct
tm
*
tm
);
char
*
strptime
(
const
char
*
buf
,
const
char
*
fmt
,
struct
tm
*
tm
);
...
@@ -153,15 +154,8 @@ char * getpass(const char *prefix);
...
@@ -153,15 +154,8 @@ char * getpass(const char *prefix);
int
flock
(
int
fd
,
int
option
);
int
flock
(
int
fd
,
int
option
);
int
fsync
(
int
filedes
);
int
fsync
(
int
filedes
);
char
*
strndup
(
const
char
*
s
,
size_t
n
);
char
*
strndup
(
const
char
*
s
,
size_t
n
);
char
*
dirname
(
char
*
pszPathname
);
int
gettimeofday
(
struct
timeval
*
ptv
,
void
*
pTimeZone
);
int
gettimeofday
(
struct
timeval
*
ptv
,
void
*
pTimeZone
);
// for access function in io.h
#define F_OK 00 //Existence only
#define W_OK 02 //Write - only
#define R_OK 04 //Read - only
#define X_OK 06 //Read and write
// for send function in tsocket.c
// for send function in tsocket.c
#define MSG_NOSIGNAL 0
#define MSG_NOSIGNAL 0
#define SO_NO_CHECK 0x1234
#define SO_NO_CHECK 0x1234
...
@@ -208,8 +202,6 @@ typedef struct {
...
@@ -208,8 +202,6 @@ typedef struct {
int
wordexp
(
char
*
words
,
wordexp_t
*
pwordexp
,
int
flags
);
int
wordexp
(
char
*
words
,
wordexp_t
*
pwordexp
,
int
flags
);
void
wordfree
(
wordexp_t
*
pwordexp
);
void
wordfree
(
wordexp_t
*
pwordexp
);
char
*
realpath
(
char
*
path
,
char
*
resolved_path
);
#define openlog(a, b, c)
#define openlog(a, b, c)
#define closelog()
#define closelog()
#define LOG_ERR 0
#define LOG_ERR 0
...
...
src/os/src/detail/osFile.c
浏览文件 @
3bf73df9
...
@@ -142,6 +142,8 @@ int64_t taosCopy(char *from, char *to) {
...
@@ -142,6 +142,8 @@ int64_t taosCopy(char *from, char *to) {
if
(
bytes
<
sizeof
(
buffer
))
break
;
if
(
bytes
<
sizeof
(
buffer
))
break
;
}
}
fsync
(
fidto
);
close
(
fidfrom
);
close
(
fidfrom
);
close
(
fidto
);
close
(
fidto
);
return
size
;
return
size
;
...
...
src/os/src/windows/wString.c
浏览文件 @
3bf73df9
...
@@ -75,18 +75,6 @@ char *getpass(const char *prefix) {
...
@@ -75,18 +75,6 @@ char *getpass(const char *prefix) {
return
passwd
;
return
passwd
;
}
}
char
*
strndup
(
const
char
*
s
,
size_t
n
)
{
size_t
len
=
strlen
(
s
);
if
(
len
>=
n
)
{
len
=
n
;
}
char
*
r
=
calloc
(
len
+
1
,
1
);
memcpy
(
r
,
s
,
len
);
r
[
len
]
=
0
;
return
r
;
}
int
twcslen
(
const
wchar_t
*
wcs
)
{
int
twcslen
(
const
wchar_t
*
wcs
)
{
int
*
wstr
=
(
int
*
)
wcs
;
int
*
wstr
=
(
int
*
)
wcs
;
if
(
NULL
==
wstr
)
{
if
(
NULL
==
wstr
)
{
...
...
src/os/src/windows/wWordexp.c
浏览文件 @
3bf73df9
...
@@ -38,7 +38,3 @@ int wordexp(char *words, wordexp_t *pwordexp, int flags) {
...
@@ -38,7 +38,3 @@ int wordexp(char *words, wordexp_t *pwordexp, int flags) {
}
}
void
wordfree
(
wordexp_t
*
pwordexp
)
{}
void
wordfree
(
wordexp_t
*
pwordexp
)
{}
char
*
realpath
(
char
*
path
,
char
*
resolved_path
)
{
return
_fullpath
(
path
,
resolved_path
,
TSDB_FILENAME_LEN
-
1
);
}
\ No newline at end of file
src/tfs/src/tfs.c
浏览文件 @
3bf73df9
...
@@ -593,7 +593,7 @@ void taosGetDisk() {
...
@@ -593,7 +593,7 @@ void taosGetDisk() {
tsAvailLogDirGB
=
(
float
)(
diskSize
.
avail
/
unit
);
tsAvailLogDirGB
=
(
float
)(
diskSize
.
avail
/
unit
);
}
}
if
(
taosGetDiskSize
(
"/tmp"
,
&
diskSize
)
==
0
)
{
if
(
taosGetDiskSize
(
tsTempDir
,
&
diskSize
)
==
0
)
{
tsTotalTmpDirGB
=
(
float
)(
diskSize
.
tsize
/
unit
);
tsTotalTmpDirGB
=
(
float
)(
diskSize
.
tsize
/
unit
);
tsAvailTmpDirectorySpace
=
(
float
)(
diskSize
.
avail
/
unit
);
tsAvailTmpDirectorySpace
=
(
float
)(
diskSize
.
avail
/
unit
);
}
}
...
...
src/tsdb/inc/tsdbFile.h
浏览文件 @
3bf73df9
...
@@ -20,6 +20,8 @@
...
@@ -20,6 +20,8 @@
#define TSDB_FILE_DELIMITER 0xF00AFA0F
#define TSDB_FILE_DELIMITER 0xF00AFA0F
#define TSDB_FILE_INIT_MAGIC 0xFFFFFFFF
#define TSDB_FILE_INIT_MAGIC 0xFFFFFFFF
#define TSDB_IVLD_FID INT_MIN
#define TSDB_IVLD_FID INT_MIN
#define TSDB_FILE_STATE_OK 0
#define TSDB_FILE_STATE_BAD 1
#define TSDB_FILE_INFO(tf) (&((tf)->info))
#define TSDB_FILE_INFO(tf) (&((tf)->info))
#define TSDB_FILE_F(tf) (&((tf)->f))
#define TSDB_FILE_F(tf) (&((tf)->f))
...
@@ -31,6 +33,10 @@
...
@@ -31,6 +33,10 @@
#define TSDB_FILE_LEVEL(tf) TFILE_LEVEL(TSDB_FILE_F(tf))
#define TSDB_FILE_LEVEL(tf) TFILE_LEVEL(TSDB_FILE_F(tf))
#define TSDB_FILE_ID(tf) TFILE_ID(TSDB_FILE_F(tf))
#define TSDB_FILE_ID(tf) TFILE_ID(TSDB_FILE_F(tf))
#define TSDB_FILE_FSYNC(tf) fsync(TSDB_FILE_FD(tf))
#define TSDB_FILE_FSYNC(tf) fsync(TSDB_FILE_FD(tf))
#define TSDB_FILE_STATE(tf) ((tf)->state)
#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
;
typedef
enum
{
TSDB_FILE_HEAD
=
0
,
TSDB_FILE_DATA
,
TSDB_FILE_LAST
,
TSDB_FILE_MAX
,
TSDB_FILE_META
}
TSDB_FILE_T
;
...
@@ -47,10 +53,11 @@ typedef struct {
...
@@ -47,10 +53,11 @@ typedef struct {
SMFInfo
info
;
SMFInfo
info
;
TFILE
f
;
TFILE
f
;
int
fd
;
int
fd
;
uint8_t
state
;
}
SMFile
;
}
SMFile
;
void
tsdbInitMFile
(
SMFile
*
pMFile
,
SDiskID
did
,
int
vid
,
uint32_t
ver
);
void
tsdbInitMFile
(
SMFile
*
pMFile
,
SDiskID
did
,
int
vid
,
uint32_t
ver
);
void
tsdbInitMFileEx
(
SMFile
*
pMFile
,
SMFile
*
pOMFile
);
void
tsdbInitMFileEx
(
SMFile
*
pMFile
,
const
SMFile
*
pOMFile
);
int
tsdbEncodeSMFile
(
void
**
buf
,
SMFile
*
pMFile
);
int
tsdbEncodeSMFile
(
void
**
buf
,
SMFile
*
pMFile
);
void
*
tsdbDecodeSMFile
(
void
*
buf
,
SMFile
*
pMFile
);
void
*
tsdbDecodeSMFile
(
void
*
buf
,
SMFile
*
pMFile
);
int
tsdbEncodeSMFileEx
(
void
**
buf
,
SMFile
*
pMFile
);
int
tsdbEncodeSMFileEx
(
void
**
buf
,
SMFile
*
pMFile
);
...
@@ -165,6 +172,7 @@ typedef struct {
...
@@ -165,6 +172,7 @@ typedef struct {
SDFInfo
info
;
SDFInfo
info
;
TFILE
f
;
TFILE
f
;
int
fd
;
int
fd
;
uint8_t
state
;
}
SDFile
;
}
SDFile
;
void
tsdbInitDFile
(
SDFile
*
pDFile
,
SDiskID
did
,
int
vid
,
int
fid
,
uint32_t
ver
,
TSDB_FILE_T
ftype
);
void
tsdbInitDFile
(
SDFile
*
pDFile
,
SDiskID
did
,
int
vid
,
int
fid
,
uint32_t
ver
,
TSDB_FILE_T
ftype
);
...
@@ -346,4 +354,14 @@ static FORCE_INLINE void tsdbGetFidKeyRange(int days, int8_t precision, int fid,
...
@@ -346,4 +354,14 @@ static FORCE_INLINE void tsdbGetFidKeyRange(int days, int8_t precision, int fid,
*
maxKey
=
*
minKey
+
days
*
tsMsPerDay
[
precision
]
-
1
;
*
maxKey
=
*
minKey
+
days
*
tsMsPerDay
[
precision
]
-
1
;
}
}
static
FORCE_INLINE
bool
tsdbFSetIsOk
(
SDFileSet
*
pSet
)
{
for
(
TSDB_FILE_T
ftype
=
0
;
ftype
<
TSDB_FILE_MAX
;
ftype
++
)
{
if
(
TSDB_FILE_IS_BAD
(
TSDB_DFILE_IN_SET
(
pSet
,
ftype
)))
{
return
false
;
}
}
return
true
;
}
#endif
/* _TS_TSDB_FILE_H_ */
#endif
/* _TS_TSDB_FILE_H_ */
\ No newline at end of file
src/tsdb/src/tsdbCommit.c
浏览文件 @
3bf73df9
...
@@ -52,7 +52,7 @@ static int tsdbCommitMeta(STsdbRepo *pRepo);
...
@@ -52,7 +52,7 @@ static int tsdbCommitMeta(STsdbRepo *pRepo);
static
int
tsdbUpdateMetaRecord
(
STsdbFS
*
pfs
,
SMFile
*
pMFile
,
uint64_t
uid
,
void
*
cont
,
int
contLen
);
static
int
tsdbUpdateMetaRecord
(
STsdbFS
*
pfs
,
SMFile
*
pMFile
,
uint64_t
uid
,
void
*
cont
,
int
contLen
);
static
int
tsdbDropMetaRecord
(
STsdbFS
*
pfs
,
SMFile
*
pMFile
,
uint64_t
uid
);
static
int
tsdbDropMetaRecord
(
STsdbFS
*
pfs
,
SMFile
*
pMFile
,
uint64_t
uid
);
static
int
tsdbCommitTSData
(
STsdbRepo
*
pRepo
);
static
int
tsdbCommitTSData
(
STsdbRepo
*
pRepo
);
static
int
tsdbStartCommit
(
STsdbRepo
*
pRepo
);
static
void
tsdbStartCommit
(
STsdbRepo
*
pRepo
);
static
void
tsdbEndCommit
(
STsdbRepo
*
pRepo
,
int
eno
);
static
void
tsdbEndCommit
(
STsdbRepo
*
pRepo
,
int
eno
);
static
int
tsdbCommitToFile
(
SCommitH
*
pCommith
,
SDFileSet
*
pSet
,
int
fid
);
static
int
tsdbCommitToFile
(
SCommitH
*
pCommith
,
SDFileSet
*
pSet
,
int
fid
);
static
int
tsdbCreateCommitIters
(
SCommitH
*
pCommith
);
static
int
tsdbCreateCommitIters
(
SCommitH
*
pCommith
);
...
@@ -84,10 +84,7 @@ static int tsdbApplyRtn(STsdbRepo *pRepo);
...
@@ -84,10 +84,7 @@ static int tsdbApplyRtn(STsdbRepo *pRepo);
static
int
tsdbApplyRtnOnFSet
(
STsdbRepo
*
pRepo
,
SDFileSet
*
pSet
,
SRtn
*
pRtn
);
static
int
tsdbApplyRtnOnFSet
(
STsdbRepo
*
pRepo
,
SDFileSet
*
pSet
,
SRtn
*
pRtn
);
void
*
tsdbCommitData
(
STsdbRepo
*
pRepo
)
{
void
*
tsdbCommitData
(
STsdbRepo
*
pRepo
)
{
if
(
tsdbStartCommit
(
pRepo
)
<
0
)
{
tsdbStartCommit
(
pRepo
);
tsdbError
(
"vgId:%d failed to commit data while startting to commit since %s"
,
REPO_ID
(
pRepo
),
tstrerror
(
terrno
));
goto
_err
;
}
// Commit to update meta file
// Commit to update meta file
if
(
tsdbCommitMeta
(
pRepo
)
<
0
)
{
if
(
tsdbCommitMeta
(
pRepo
)
<
0
)
{
...
@@ -138,11 +135,15 @@ static int tsdbCommitMeta(STsdbRepo *pRepo) {
...
@@ -138,11 +135,15 @@ static int tsdbCommitMeta(STsdbRepo *pRepo) {
tsdbInitMFile
(
&
mf
,
did
,
REPO_ID
(
pRepo
),
FS_TXN_VERSION
(
REPO_FS
(
pRepo
)));
tsdbInitMFile
(
&
mf
,
did
,
REPO_ID
(
pRepo
),
FS_TXN_VERSION
(
REPO_FS
(
pRepo
)));
if
(
tsdbCreateMFile
(
&
mf
,
true
)
<
0
)
{
if
(
tsdbCreateMFile
(
&
mf
,
true
)
<
0
)
{
tsdbError
(
"vgId:%d failed to create META file since %s"
,
REPO_ID
(
pRepo
),
tstrerror
(
terrno
));
return
-
1
;
return
-
1
;
}
}
tsdbInfo
(
"vgId:%d meta file %s is created to commit"
,
REPO_ID
(
pRepo
),
TSDB_FILE_FULL_NAME
(
&
mf
));
}
else
{
}
else
{
tsdbInitMFileEx
(
&
mf
,
pOMFile
);
tsdbInitMFileEx
(
&
mf
,
pOMFile
);
if
(
tsdbOpenMFile
(
&
mf
,
O_WRONLY
)
<
0
)
{
if
(
tsdbOpenMFile
(
&
mf
,
O_WRONLY
)
<
0
)
{
tsdbError
(
"vgId:%d failed to open META file since %s"
,
REPO_ID
(
pRepo
),
tstrerror
(
terrno
));
return
-
1
;
return
-
1
;
}
}
}
}
...
@@ -154,12 +155,20 @@ static int tsdbCommitMeta(STsdbRepo *pRepo) {
...
@@ -154,12 +155,20 @@ static int tsdbCommitMeta(STsdbRepo *pRepo) {
if
(
pAct
->
act
==
TSDB_UPDATE_META
)
{
if
(
pAct
->
act
==
TSDB_UPDATE_META
)
{
pCont
=
(
SActCont
*
)
POINTER_SHIFT
(
pAct
,
sizeof
(
SActObj
));
pCont
=
(
SActCont
*
)
POINTER_SHIFT
(
pAct
,
sizeof
(
SActObj
));
if
(
tsdbUpdateMetaRecord
(
pfs
,
&
mf
,
pAct
->
uid
,
(
void
*
)(
pCont
->
cont
),
pCont
->
len
)
<
0
)
{
if
(
tsdbUpdateMetaRecord
(
pfs
,
&
mf
,
pAct
->
uid
,
(
void
*
)(
pCont
->
cont
),
pCont
->
len
)
<
0
)
{
tsdbError
(
"vgId:%d failed to update META record, uid %"
PRIu64
" since %s"
,
REPO_ID
(
pRepo
),
pAct
->
uid
,
tstrerror
(
terrno
));
tsdbCloseMFile
(
&
mf
);
tsdbCloseMFile
(
&
mf
);
tsdbApplyMFileChange
(
&
mf
,
pOMFile
);
// TODO: need to reload metaCache
return
-
1
;
return
-
1
;
}
}
}
else
if
(
pAct
->
act
==
TSDB_DROP_META
)
{
}
else
if
(
pAct
->
act
==
TSDB_DROP_META
)
{
if
(
tsdbDropMetaRecord
(
pfs
,
&
mf
,
pAct
->
uid
)
<
0
)
{
if
(
tsdbDropMetaRecord
(
pfs
,
&
mf
,
pAct
->
uid
)
<
0
)
{
tsdbError
(
"vgId:%d failed to drop META record, uid %"
PRIu64
" since %s"
,
REPO_ID
(
pRepo
),
pAct
->
uid
,
tstrerror
(
terrno
));
tsdbCloseMFile
(
&
mf
);
tsdbCloseMFile
(
&
mf
);
tsdbApplyMFileChange
(
&
mf
,
pOMFile
);
// TODO: need to reload metaCache
return
-
1
;
return
-
1
;
}
}
}
else
{
}
else
{
...
@@ -168,6 +177,9 @@ static int tsdbCommitMeta(STsdbRepo *pRepo) {
...
@@ -168,6 +177,9 @@ static int tsdbCommitMeta(STsdbRepo *pRepo) {
}
}
if
(
tsdbUpdateMFileHeader
(
&
mf
)
<
0
)
{
if
(
tsdbUpdateMFileHeader
(
&
mf
)
<
0
)
{
tsdbError
(
"vgId:%d failed to update META file header since %s, revert it"
,
REPO_ID
(
pRepo
),
tstrerror
(
terrno
));
tsdbApplyMFileChange
(
&
mf
,
pOMFile
);
// TODO: need to reload metaCache
return
-
1
;
return
-
1
;
}
}
...
@@ -208,6 +220,8 @@ void tsdbGetRtnSnap(STsdbRepo *pRepo, SRtn *pRtn) {
...
@@ -208,6 +220,8 @@ void tsdbGetRtnSnap(STsdbRepo *pRepo, SRtn *pRtn) {
pRtn
->
minFid
=
(
int
)(
TSDB_KEY_FID
(
minKey
,
pCfg
->
daysPerFile
,
pCfg
->
precision
));
pRtn
->
minFid
=
(
int
)(
TSDB_KEY_FID
(
minKey
,
pCfg
->
daysPerFile
,
pCfg
->
precision
));
pRtn
->
midFid
=
(
int
)(
TSDB_KEY_FID
(
midKey
,
pCfg
->
daysPerFile
,
pCfg
->
precision
));
pRtn
->
midFid
=
(
int
)(
TSDB_KEY_FID
(
midKey
,
pCfg
->
daysPerFile
,
pCfg
->
precision
));
pRtn
->
maxFid
=
(
int
)(
TSDB_KEY_FID
(
maxKey
,
pCfg
->
daysPerFile
,
pCfg
->
precision
));
pRtn
->
maxFid
=
(
int
)(
TSDB_KEY_FID
(
maxKey
,
pCfg
->
daysPerFile
,
pCfg
->
precision
));
tsdbDebug
(
"vgId:%d now:%"
PRId64
" minKey:%"
PRId64
" minFid:%d, midFid:%d, maxFid:%d"
,
REPO_ID
(
pRepo
),
now
,
minKey
,
pRtn
->
minFid
,
pRtn
->
midFid
,
pRtn
->
maxFid
);
}
}
static
int
tsdbUpdateMetaRecord
(
STsdbFS
*
pfs
,
SMFile
*
pMFile
,
uint64_t
uid
,
void
*
cont
,
int
contLen
)
{
static
int
tsdbUpdateMetaRecord
(
STsdbFS
*
pfs
,
SMFile
*
pMFile
,
uint64_t
uid
,
void
*
cont
,
int
contLen
)
{
...
@@ -238,7 +252,7 @@ static int tsdbUpdateMetaRecord(STsdbFS *pfs, SMFile *pMFile, uint64_t uid, void
...
@@ -238,7 +252,7 @@ static int tsdbUpdateMetaRecord(STsdbFS *pfs, SMFile *pMFile, uint64_t uid, void
tsdbUpdateMFileMagic
(
pMFile
,
POINTER_SHIFT
(
cont
,
contLen
-
sizeof
(
TSCKSUM
)));
tsdbUpdateMFileMagic
(
pMFile
,
POINTER_SHIFT
(
cont
,
contLen
-
sizeof
(
TSCKSUM
)));
SKVRecord
*
pRecord
=
taosHashGet
(
pfs
->
metaCache
,
(
void
*
)
&
uid
,
sizeof
(
uid
));
SKVRecord
*
pRecord
=
taosHashGet
(
pfs
->
metaCache
,
(
void
*
)
&
uid
,
sizeof
(
uid
));
if
(
pRecord
!=
NULL
)
{
if
(
pRecord
!=
NULL
)
{
pMFile
->
info
.
tombSize
+=
pRecord
->
size
;
pMFile
->
info
.
tombSize
+=
(
pRecord
->
size
+
sizeof
(
SKVRecord
))
;
}
else
{
}
else
{
pMFile
->
info
.
nRecords
++
;
pMFile
->
info
.
nRecords
++
;
}
}
...
@@ -253,7 +267,7 @@ static int tsdbDropMetaRecord(STsdbFS *pfs, SMFile *pMFile, uint64_t uid) {
...
@@ -253,7 +267,7 @@ static int tsdbDropMetaRecord(STsdbFS *pfs, SMFile *pMFile, uint64_t uid) {
SKVRecord
*
pRecord
=
taosHashGet
(
pfs
->
metaCache
,
(
void
*
)(
&
uid
),
sizeof
(
uid
));
SKVRecord
*
pRecord
=
taosHashGet
(
pfs
->
metaCache
,
(
void
*
)(
&
uid
),
sizeof
(
uid
));
if
(
pRecord
==
NULL
)
{
if
(
pRecord
==
NULL
)
{
tsdbError
(
"failed to drop
KV store
record with key %"
PRIu64
" since not find"
,
uid
);
tsdbError
(
"failed to drop
META
record with key %"
PRIu64
" since not find"
,
uid
);
return
-
1
;
return
-
1
;
}
}
...
@@ -264,11 +278,11 @@ static int tsdbDropMetaRecord(STsdbFS *pfs, SMFile *pMFile, uint64_t uid) {
...
@@ -264,11 +278,11 @@ static int tsdbDropMetaRecord(STsdbFS *pfs, SMFile *pMFile, uint64_t uid) {
void
*
pBuf
=
buf
;
void
*
pBuf
=
buf
;
tsdbEncodeKVRecord
(
&
pBuf
,
&
rInfo
);
tsdbEncodeKVRecord
(
&
pBuf
,
&
rInfo
);
if
(
tsdbAppendMFile
(
pMFile
,
buf
,
POINTER_DISTANCE
(
pBuf
,
buf
),
NULL
)
<
0
)
{
if
(
tsdbAppendMFile
(
pMFile
,
buf
,
sizeof
(
SKVRecord
),
NULL
)
<
0
)
{
return
-
1
;
return
-
1
;
}
}
pMFile
->
info
.
magic
=
taosCalcChecksum
(
pMFile
->
info
.
magic
,
(
uint8_t
*
)
buf
,
(
uint32_t
)
POINTER_DISTANCE
(
pBuf
,
buf
));
pMFile
->
info
.
magic
=
taosCalcChecksum
(
pMFile
->
info
.
magic
,
(
uint8_t
*
)
buf
,
sizeof
(
SKVRecord
));
pMFile
->
info
.
nDels
++
;
pMFile
->
info
.
nDels
++
;
pMFile
->
info
.
nRecords
--
;
pMFile
->
info
.
nRecords
--
;
pMFile
->
info
.
tombSize
+=
(
rInfo
.
size
+
sizeof
(
SKVRecord
)
*
2
);
pMFile
->
info
.
tombSize
+=
(
rInfo
.
size
+
sizeof
(
SKVRecord
)
*
2
);
...
@@ -302,7 +316,12 @@ static int tsdbCommitTSData(STsdbRepo *pRepo) {
...
@@ -302,7 +316,12 @@ static int tsdbCommitTSData(STsdbRepo *pRepo) {
// Skip expired memory data and expired FSET
// Skip expired memory data and expired FSET
tsdbSeekCommitIter
(
&
commith
,
commith
.
rtn
.
minKey
);
tsdbSeekCommitIter
(
&
commith
,
commith
.
rtn
.
minKey
);
while
((
pSet
=
tsdbFSIterNext
(
&
(
commith
.
fsIter
))))
{
while
((
pSet
=
tsdbFSIterNext
(
&
(
commith
.
fsIter
))))
{
if
(
pSet
->
fid
>=
commith
.
rtn
.
minFid
)
break
;
if
(
pSet
->
fid
<
commith
.
rtn
.
minFid
)
{
tsdbInfo
(
"vgId:%d FSET %d on level %d disk id %d expires, remove it"
,
REPO_ID
(
pRepo
),
pSet
->
fid
,
TSDB_FSET_LEVEL
(
pSet
),
TSDB_FSET_ID
(
pSet
));
}
else
{
break
;
}
}
}
// Loop to commit to each file
// Loop to commit to each file
...
@@ -349,7 +368,7 @@ static int tsdbCommitTSData(STsdbRepo *pRepo) {
...
@@ -349,7 +368,7 @@ static int tsdbCommitTSData(STsdbRepo *pRepo) {
return
0
;
return
0
;
}
}
static
int
tsdbStartCommit
(
STsdbRepo
*
pRepo
)
{
static
void
tsdbStartCommit
(
STsdbRepo
*
pRepo
)
{
SMemTable
*
pMem
=
pRepo
->
imem
;
SMemTable
*
pMem
=
pRepo
->
imem
;
ASSERT
(
pMem
->
numOfRows
>
0
||
listNEles
(
pMem
->
actList
)
>
0
);
ASSERT
(
pMem
->
numOfRows
>
0
||
listNEles
(
pMem
->
actList
)
>
0
);
...
@@ -360,7 +379,6 @@ static int tsdbStartCommit(STsdbRepo *pRepo) {
...
@@ -360,7 +379,6 @@ static int tsdbStartCommit(STsdbRepo *pRepo) {
tsdbStartFSTxn
(
pRepo
,
pMem
->
pointsAdd
,
pMem
->
storageAdd
);
tsdbStartFSTxn
(
pRepo
,
pMem
->
pointsAdd
,
pMem
->
storageAdd
);
pRepo
->
code
=
TSDB_CODE_SUCCESS
;
pRepo
->
code
=
TSDB_CODE_SUCCESS
;
return
0
;
}
}
static
void
tsdbEndCommit
(
STsdbRepo
*
pRepo
,
int
eno
)
{
static
void
tsdbEndCommit
(
STsdbRepo
*
pRepo
,
int
eno
)
{
...
@@ -413,14 +431,18 @@ static int tsdbCommitToFile(SCommitH *pCommith, SDFileSet *pSet, int fid) {
...
@@ -413,14 +431,18 @@ static int tsdbCommitToFile(SCommitH *pCommith, SDFileSet *pSet, int fid) {
if
(
pIter
->
pTable
==
NULL
)
continue
;
if
(
pIter
->
pTable
==
NULL
)
continue
;
if
(
tsdbCommitToTable
(
pCommith
,
tid
)
<
0
)
{
if
(
tsdbCommitToTable
(
pCommith
,
tid
)
<
0
)
{
// TODO: revert the file change
tsdbCloseCommitFile
(
pCommith
,
true
);
tsdbCloseCommitFile
(
pCommith
,
true
);
// revert the file change
tsdbApplyDFileSetChange
(
TSDB_COMMIT_WRITE_FSET
(
pCommith
),
pSet
);
return
-
1
;
return
-
1
;
}
}
}
}
if
(
tsdbWriteBlockIdx
(
pCommith
)
<
0
)
{
if
(
tsdbWriteBlockIdx
(
pCommith
)
<
0
)
{
tsdbError
(
"vgId:%d failed to write SBlockIdx part to FSET %d since %s"
,
REPO_ID
(
pRepo
),
fid
,
tstrerror
(
terrno
));
tsdbCloseCommitFile
(
pCommith
,
true
);
tsdbCloseCommitFile
(
pCommith
,
true
);
// revert the file change
tsdbApplyDFileSetChange
(
TSDB_COMMIT_WRITE_FSET
(
pCommith
),
pSet
);
return
-
1
;
return
-
1
;
}
}
...
@@ -674,7 +696,11 @@ static int tsdbCommitToTable(SCommitH *pCommith, int tid) {
...
@@ -674,7 +696,11 @@ static int tsdbCommitToTable(SCommitH *pCommith, int tid) {
TSDB_RUNLOCK_TABLE
(
pIter
->
pTable
);
TSDB_RUNLOCK_TABLE
(
pIter
->
pTable
);
if
(
tsdbWriteBlockInfo
(
pCommith
)
<
0
)
return
-
1
;
if
(
tsdbWriteBlockInfo
(
pCommith
)
<
0
)
{
tsdbError
(
"vgId:%d failed to write SBlockInfo part into file %s since %s"
,
TSDB_COMMIT_REPO_ID
(
pCommith
),
TSDB_FILE_FULL_NAME
(
TSDB_COMMIT_HEAD_FILE
(
pCommith
)),
tstrerror
(
terrno
));
return
-
1
;
}
return
0
;
return
0
;
}
}
...
@@ -926,6 +952,8 @@ static int tsdbWriteBlockIdx(SCommitH *pCommih) {
...
@@ -926,6 +952,8 @@ static int tsdbWriteBlockIdx(SCommitH *pCommih) {
if
(
nidx
<=
0
)
{
if
(
nidx
<=
0
)
{
// All data are deleted
// All data are deleted
pHeadf
->
info
.
offset
=
0
;
pHeadf
->
info
.
len
=
0
;
return
0
;
return
0
;
}
}
...
@@ -1227,7 +1255,6 @@ static void tsdbResetCommitFile(SCommitH *pCommith) {
...
@@ -1227,7 +1255,6 @@ static void tsdbResetCommitFile(SCommitH *pCommith) {
}
}
static
void
tsdbResetCommitTable
(
SCommitH
*
pCommith
)
{
static
void
tsdbResetCommitTable
(
SCommitH
*
pCommith
)
{
tdResetDataCols
(
pCommith
->
pDataCols
);
taosArrayClear
(
pCommith
->
aSubBlk
);
taosArrayClear
(
pCommith
->
aSubBlk
);
taosArrayClear
(
pCommith
->
aSupBlk
);
taosArrayClear
(
pCommith
->
aSupBlk
);
pCommith
->
pTable
=
NULL
;
pCommith
->
pTable
=
NULL
;
...
@@ -1256,6 +1283,9 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
...
@@ -1256,6 +1283,9 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
tsdbCloseAndUnsetFSet
(
&
(
pCommith
->
readh
));
tsdbCloseAndUnsetFSet
(
&
(
pCommith
->
readh
));
return
-
1
;
return
-
1
;
}
}
tsdbDebug
(
"vgId:%d FSET %d at level %d disk id %d is opened to read to commit"
,
REPO_ID
(
pRepo
),
TSDB_FSET_FID
(
pSet
),
TSDB_FSET_LEVEL
(
pSet
),
TSDB_FSET_ID
(
pSet
));
}
else
{
}
else
{
pCommith
->
isRFileSet
=
false
;
pCommith
->
isRFileSet
=
false
;
}
}
...
@@ -1266,6 +1296,8 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
...
@@ -1266,6 +1296,8 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
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
)));
if
(
tsdbCreateDFileSet
(
pWSet
,
true
)
<
0
)
{
if
(
tsdbCreateDFileSet
(
pWSet
,
true
)
<
0
)
{
tsdbError
(
"vgId:%d failed to create FSET %d at level %d disk id %d since %s"
,
REPO_ID
(
pRepo
),
TSDB_FSET_FID
(
pWSet
),
TSDB_FSET_LEVEL
(
pWSet
),
TSDB_FSET_ID
(
pWSet
),
tstrerror
(
terrno
));
if
(
pCommith
->
isRFileSet
)
{
if
(
pCommith
->
isRFileSet
)
{
tsdbCloseAndUnsetFSet
(
&
(
pCommith
->
readh
));
tsdbCloseAndUnsetFSet
(
&
(
pCommith
->
readh
));
}
}
...
@@ -1274,6 +1306,9 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
...
@@ -1274,6 +1306,9 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
pCommith
->
isDFileSame
=
false
;
pCommith
->
isDFileSame
=
false
;
pCommith
->
isLFileSame
=
false
;
pCommith
->
isLFileSame
=
false
;
tsdbDebug
(
"vgId:%d FSET %d at level %d disk id %d is created to commit"
,
REPO_ID
(
pRepo
),
TSDB_FSET_FID
(
pWSet
),
TSDB_FSET_LEVEL
(
pWSet
),
TSDB_FSET_ID
(
pWSet
));
}
else
{
}
else
{
did
.
level
=
TSDB_FSET_LEVEL
(
pSet
);
did
.
level
=
TSDB_FSET_LEVEL
(
pSet
);
did
.
id
=
TSDB_FSET_ID
(
pSet
);
did
.
id
=
TSDB_FSET_ID
(
pSet
);
...
@@ -1285,6 +1320,9 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
...
@@ -1285,6 +1320,9 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
SDFile
*
pWHeadf
=
TSDB_COMMIT_HEAD_FILE
(
pCommith
);
SDFile
*
pWHeadf
=
TSDB_COMMIT_HEAD_FILE
(
pCommith
);
tsdbInitDFile
(
pWHeadf
,
did
,
REPO_ID
(
pRepo
),
fid
,
FS_TXN_VERSION
(
REPO_FS
(
pRepo
)),
TSDB_FILE_HEAD
);
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
)
<
0
)
{
tsdbError
(
"vgId:%d failed to create file %s to commit since %s"
,
REPO_ID
(
pRepo
),
TSDB_FILE_FULL_NAME
(
pWHeadf
),
tstrerror
(
terrno
));
if
(
pCommith
->
isRFileSet
)
{
if
(
pCommith
->
isRFileSet
)
{
tsdbCloseAndUnsetFSet
(
&
(
pCommith
->
readh
));
tsdbCloseAndUnsetFSet
(
&
(
pCommith
->
readh
));
return
-
1
;
return
-
1
;
...
@@ -1296,7 +1334,10 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
...
@@ -1296,7 +1334,10 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
SDFile
*
pWDataf
=
TSDB_COMMIT_DATA_FILE
(
pCommith
);
SDFile
*
pWDataf
=
TSDB_COMMIT_DATA_FILE
(
pCommith
);
tsdbInitDFileEx
(
pWDataf
,
pRDataf
);
tsdbInitDFileEx
(
pWDataf
,
pRDataf
);
if
(
tsdbOpenDFile
(
pWDataf
,
O_WRONLY
)
<
0
)
{
if
(
tsdbOpenDFile
(
pWDataf
,
O_WRONLY
)
<
0
)
{
tsdbCloseDFile
(
pWHeadf
);
tsdbError
(
"vgId:%d failed to open file %s to commit since %s"
,
REPO_ID
(
pRepo
),
TSDB_FILE_FULL_NAME
(
pWDataf
),
tstrerror
(
terrno
));
tsdbCloseDFileSet
(
pWSet
);
tsdbRemoveDFile
(
pWHeadf
);
tsdbRemoveDFile
(
pWHeadf
);
if
(
pCommith
->
isRFileSet
)
{
if
(
pCommith
->
isRFileSet
)
{
tsdbCloseAndUnsetFSet
(
&
(
pCommith
->
readh
));
tsdbCloseAndUnsetFSet
(
&
(
pCommith
->
readh
));
...
@@ -1313,6 +1354,9 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
...
@@ -1313,6 +1354,9 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
pCommith
->
isLFileSame
=
true
;
pCommith
->
isLFileSame
=
true
;
if
(
tsdbOpenDFile
(
pWLastf
,
O_WRONLY
)
<
0
)
{
if
(
tsdbOpenDFile
(
pWLastf
,
O_WRONLY
)
<
0
)
{
tsdbError
(
"vgId:%d failed to open file %s to commit since %s"
,
REPO_ID
(
pRepo
),
TSDB_FILE_FULL_NAME
(
pWLastf
),
tstrerror
(
terrno
));
tsdbCloseDFileSet
(
pWSet
);
tsdbCloseDFileSet
(
pWSet
);
tsdbRemoveDFile
(
pWHeadf
);
tsdbRemoveDFile
(
pWHeadf
);
if
(
pCommith
->
isRFileSet
)
{
if
(
pCommith
->
isRFileSet
)
{
...
@@ -1325,6 +1369,9 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
...
@@ -1325,6 +1369,9 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
pCommith
->
isLFileSame
=
false
;
pCommith
->
isLFileSame
=
false
;
if
(
tsdbCreateDFile
(
pWLastf
,
true
)
<
0
)
{
if
(
tsdbCreateDFile
(
pWLastf
,
true
)
<
0
)
{
tsdbError
(
"vgId:%d failed to create file %s to commit since %s"
,
REPO_ID
(
pRepo
),
TSDB_FILE_FULL_NAME
(
pWLastf
),
tstrerror
(
terrno
));
tsdbCloseDFileSet
(
pWSet
);
tsdbCloseDFileSet
(
pWSet
);
tsdbRemoveDFile
(
pWHeadf
);
tsdbRemoveDFile
(
pWHeadf
);
if
(
pCommith
->
isRFileSet
)
{
if
(
pCommith
->
isRFileSet
)
{
...
@@ -1360,7 +1407,7 @@ static bool tsdbCanAddSubBlock(SCommitH *pCommith, SBlock *pBlock, SMergeInfo *p
...
@@ -1360,7 +1407,7 @@ static bool tsdbCanAddSubBlock(SCommitH *pCommith, SBlock *pBlock, SMergeInfo *p
if
(
pBlock
->
last
)
{
if
(
pBlock
->
last
)
{
if
(
pCommith
->
isLFileSame
&&
mergeRows
<
pCfg
->
minRowsPerFileBlock
)
return
true
;
if
(
pCommith
->
isLFileSame
&&
mergeRows
<
pCfg
->
minRowsPerFileBlock
)
return
true
;
}
else
{
}
else
{
if
(
mergeRows
<
pCfg
->
maxRowsPerFileBlock
)
return
true
;
if
(
pCommith
->
isDFileSame
&&
mergeRows
<=
pCfg
->
maxRowsPerFileBlock
)
return
true
;
}
}
}
}
...
@@ -1373,12 +1420,16 @@ static int tsdbApplyRtn(STsdbRepo *pRepo) {
...
@@ -1373,12 +1420,16 @@ static int tsdbApplyRtn(STsdbRepo *pRepo) {
STsdbFS
*
pfs
=
REPO_FS
(
pRepo
);
STsdbFS
*
pfs
=
REPO_FS
(
pRepo
);
SDFileSet
*
pSet
;
SDFileSet
*
pSet
;
// Get retention
i
snapshot
// Get retention snapshot
tsdbGetRtnSnap
(
pRepo
,
&
rtn
);
tsdbGetRtnSnap
(
pRepo
,
&
rtn
);
tsdbFSIterInit
(
&
fsiter
,
pfs
,
TSDB_FS_ITER_FORWARD
);
tsdbFSIterInit
(
&
fsiter
,
pfs
,
TSDB_FS_ITER_FORWARD
);
while
((
pSet
=
tsdbFSIterNext
(
&
fsiter
)))
{
while
((
pSet
=
tsdbFSIterNext
(
&
fsiter
)))
{
if
(
pSet
->
fid
<
rtn
.
minFid
)
continue
;
if
(
pSet
->
fid
<
rtn
.
minFid
)
{
tsdbInfo
(
"vgId:%d FSET %d at level %d disk id %d expires, remove it"
,
REPO_ID
(
pRepo
),
pSet
->
fid
,
TSDB_FSET_LEVEL
(
pSet
),
TSDB_FSET_ID
(
pSet
));
continue
;
}
if
(
tsdbApplyRtnOnFSet
(
pRepo
,
pSet
,
&
rtn
)
<
0
)
{
if
(
tsdbApplyRtnOnFSet
(
pRepo
,
pSet
,
&
rtn
)
<
0
)
{
return
-
1
;
return
-
1
;
...
@@ -1392,10 +1443,13 @@ static int tsdbApplyRtnOnFSet(STsdbRepo *pRepo, SDFileSet *pSet, SRtn *pRtn) {
...
@@ -1392,10 +1443,13 @@ static int tsdbApplyRtnOnFSet(STsdbRepo *pRepo, SDFileSet *pSet, SRtn *pRtn) {
SDiskID
did
;
SDiskID
did
;
SDFileSet
nSet
;
SDFileSet
nSet
;
STsdbFS
*
pfs
=
REPO_FS
(
pRepo
);
STsdbFS
*
pfs
=
REPO_FS
(
pRepo
);
int
level
;
ASSERT
(
pSet
->
fid
>=
pRtn
->
minFid
);
ASSERT
(
pSet
->
fid
>=
pRtn
->
minFid
);
tfsAllocDisk
(
tsdbGetFidLevel
(
pSet
->
fid
,
pRtn
),
&
(
did
.
level
),
&
(
did
.
id
));
level
=
tsdbGetFidLevel
(
pSet
->
fid
,
pRtn
);
tfsAllocDisk
(
level
,
&
(
did
.
level
),
&
(
did
.
id
));
if
(
did
.
level
==
TFS_UNDECIDED_LEVEL
)
{
if
(
did
.
level
==
TFS_UNDECIDED_LEVEL
)
{
terrno
=
TSDB_CODE_TDB_NO_AVAIL_DISK
;
terrno
=
TSDB_CODE_TDB_NO_AVAIL_DISK
;
return
-
1
;
return
-
1
;
...
@@ -1406,12 +1460,17 @@ static int tsdbApplyRtnOnFSet(STsdbRepo *pRepo, SDFileSet *pSet, SRtn *pRtn) {
...
@@ -1406,12 +1460,17 @@ static int tsdbApplyRtnOnFSet(STsdbRepo *pRepo, SDFileSet *pSet, SRtn *pRtn) {
tsdbInitDFileSet
(
&
nSet
,
did
,
REPO_ID
(
pRepo
),
pSet
->
fid
,
FS_TXN_VERSION
(
pfs
));
tsdbInitDFileSet
(
&
nSet
,
did
,
REPO_ID
(
pRepo
),
pSet
->
fid
,
FS_TXN_VERSION
(
pfs
));
if
(
tsdbCopyDFileSet
(
pSet
,
&
nSet
)
<
0
)
{
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
,
TSDB_FSET_LEVEL
(
pSet
),
did
.
level
,
tstrerror
(
terrno
));
return
-
1
;
return
-
1
;
}
}
if
(
tsdbUpdateDFileSet
(
pfs
,
&
nSet
)
<
0
)
{
if
(
tsdbUpdateDFileSet
(
pfs
,
&
nSet
)
<
0
)
{
return
-
1
;
return
-
1
;
}
}
tsdbInfo
(
"vgId:%d FSET %d is copied from level %d disk id %d to level %d disk id %d"
,
REPO_ID
(
pRepo
),
pSet
->
fid
,
TSDB_FSET_LEVEL
(
pSet
),
TSDB_FSET_ID
(
pSet
),
did
.
level
,
did
.
id
);
}
else
{
}
else
{
// On a correct level
// On a correct level
if
(
tsdbUpdateDFileSet
(
pfs
,
pSet
)
<
0
)
{
if
(
tsdbUpdateDFileSet
(
pfs
,
pSet
)
<
0
)
{
...
...
src/tsdb/src/tsdbFS.c
浏览文件 @
3bf73df9
...
@@ -701,6 +701,8 @@ int tsdbLoadMetaCache(STsdbRepo *pRepo, bool recoverMeta) {
...
@@ -701,6 +701,8 @@ int tsdbLoadMetaCache(STsdbRepo *pRepo, bool recoverMeta) {
int64_t
maxBufSize
=
0
;
int64_t
maxBufSize
=
0
;
SMFInfo
minfo
;
SMFInfo
minfo
;
taosHashEmpty
(
pfs
->
metaCache
);
// No meta file, just return
// No meta file, just return
if
(
pfs
->
cstatus
->
pmf
==
NULL
)
return
0
;
if
(
pfs
->
cstatus
->
pmf
==
NULL
)
return
0
;
...
@@ -718,6 +720,12 @@ int tsdbLoadMetaCache(STsdbRepo *pRepo, bool recoverMeta) {
...
@@ -718,6 +720,12 @@ int tsdbLoadMetaCache(STsdbRepo *pRepo, bool recoverMeta) {
while
(
true
)
{
while
(
true
)
{
int64_t
tsize
=
tsdbReadMFile
(
pMFile
,
tbuf
,
sizeof
(
SKVRecord
));
int64_t
tsize
=
tsdbReadMFile
(
pMFile
,
tbuf
,
sizeof
(
SKVRecord
));
if
(
tsize
==
0
)
break
;
if
(
tsize
==
0
)
break
;
if
(
tsize
<
0
)
{
tsdbError
(
"vgId:%d failed to read META file since %s"
,
REPO_ID
(
pRepo
),
tstrerror
(
terrno
));
return
-
1
;
}
if
(
tsize
<
sizeof
(
SKVRecord
))
{
if
(
tsize
<
sizeof
(
SKVRecord
))
{
tsdbError
(
"vgId:%d failed to read %"
PRIzu
" bytes from file %s"
,
REPO_ID
(
pRepo
),
sizeof
(
SKVRecord
),
tsdbError
(
"vgId:%d failed to read %"
PRIzu
" bytes from file %s"
,
REPO_ID
(
pRepo
),
sizeof
(
SKVRecord
),
TSDB_FILE_FULL_NAME
(
pMFile
));
TSDB_FILE_FULL_NAME
(
pMFile
));
...
@@ -840,7 +848,7 @@ static int tsdbScanRootDir(STsdbRepo *pRepo) {
...
@@ -840,7 +848,7 @@ static int tsdbScanRootDir(STsdbRepo *pRepo) {
continue
;
continue
;
}
}
if
(
tfsIsSameFile
(
pf
,
&
(
pfs
->
cstatus
->
pmf
->
f
)))
{
if
(
pfs
->
cstatus
->
pmf
&&
tfsIsSameFile
(
pf
,
&
(
pfs
->
cstatus
->
pmf
->
f
)))
{
continue
;
continue
;
}
}
...
...
src/tsdb/src/tsdbFile.c
浏览文件 @
3bf73df9
...
@@ -33,7 +33,7 @@ static int tsdbRollBackDFile(SDFile *pDFile);
...
@@ -33,7 +33,7 @@ static int tsdbRollBackDFile(SDFile *pDFile);
void
tsdbInitMFile
(
SMFile
*
pMFile
,
SDiskID
did
,
int
vid
,
uint32_t
ver
)
{
void
tsdbInitMFile
(
SMFile
*
pMFile
,
SDiskID
did
,
int
vid
,
uint32_t
ver
)
{
char
fname
[
TSDB_FILENAME_LEN
];
char
fname
[
TSDB_FILENAME_LEN
];
TSDB_FILE_SET_
CLOSED
(
pMFile
);
TSDB_FILE_SET_
STATE
(
pMFile
,
TSDB_FILE_STATE_OK
);
memset
(
&
(
pMFile
->
info
),
0
,
sizeof
(
pMFile
->
info
));
memset
(
&
(
pMFile
->
info
),
0
,
sizeof
(
pMFile
->
info
));
pMFile
->
info
.
magic
=
TSDB_FILE_INIT_MAGIC
;
pMFile
->
info
.
magic
=
TSDB_FILE_INIT_MAGIC
;
...
@@ -42,7 +42,7 @@ void tsdbInitMFile(SMFile *pMFile, SDiskID did, int vid, uint32_t ver) {
...
@@ -42,7 +42,7 @@ void tsdbInitMFile(SMFile *pMFile, SDiskID did, int vid, uint32_t ver) {
tfsInitFile
(
TSDB_FILE_F
(
pMFile
),
did
.
level
,
did
.
id
,
fname
);
tfsInitFile
(
TSDB_FILE_F
(
pMFile
),
did
.
level
,
did
.
id
,
fname
);
}
}
void
tsdbInitMFileEx
(
SMFile
*
pMFile
,
SMFile
*
pOMFile
)
{
void
tsdbInitMFileEx
(
SMFile
*
pMFile
,
const
SMFile
*
pOMFile
)
{
*
pMFile
=
*
pOMFile
;
*
pMFile
=
*
pOMFile
;
TSDB_FILE_SET_CLOSED
(
pMFile
);
TSDB_FILE_SET_CLOSED
(
pMFile
);
}
}
...
@@ -201,6 +201,7 @@ int tsdbScanAndTryFixMFile(STsdbRepo *pRepo) {
...
@@ -201,6 +201,7 @@ int tsdbScanAndTryFixMFile(STsdbRepo *pRepo) {
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 exit, report to upper layer to fix it"
,
REPO_ID
(
pRepo
),
TSDB_FILE_FULL_NAME
(
pMFile
));
TSDB_FILE_FULL_NAME
(
pMFile
));
pRepo
->
state
|=
TSDB_STATE_BAD_META
;
pRepo
->
state
|=
TSDB_STATE_BAD_META
;
TSDB_FILE_SET_STATE
(
pMFile
,
TSDB_FILE_STATE_BAD
);
return
0
;
return
0
;
}
}
...
@@ -232,6 +233,7 @@ int tsdbScanAndTryFixMFile(STsdbRepo *pRepo) {
...
@@ -232,6 +233,7 @@ int tsdbScanAndTryFixMFile(STsdbRepo *pRepo) {
tsdbError
(
"vgId:%d meta file %s has wrong size %"
PRId64
" expected %"
PRId64
", report to upper layer to fix it"
,
tsdbError
(
"vgId:%d meta file %s has wrong size %"
PRId64
" expected %"
PRId64
", report to upper layer to fix it"
,
REPO_ID
(
pRepo
),
TSDB_FILE_FULL_NAME
(
pMFile
),
mfstat
.
st_size
,
pMFile
->
info
.
size
);
REPO_ID
(
pRepo
),
TSDB_FILE_FULL_NAME
(
pMFile
),
mfstat
.
st_size
,
pMFile
->
info
.
size
);
pRepo
->
state
|=
TSDB_STATE_BAD_META
;
pRepo
->
state
|=
TSDB_STATE_BAD_META
;
TSDB_FILE_SET_STATE
(
pMFile
,
TSDB_FILE_STATE_BAD
);
terrno
=
TSDB_CODE_TDB_FILE_CORRUPTED
;
terrno
=
TSDB_CODE_TDB_FILE_CORRUPTED
;
return
0
;
return
0
;
}
else
{
}
else
{
...
@@ -293,6 +295,8 @@ static int tsdbRollBackMFile(SMFile *pMFile) {
...
@@ -293,6 +295,8 @@ static int tsdbRollBackMFile(SMFile *pMFile) {
void
tsdbInitDFile
(
SDFile
*
pDFile
,
SDiskID
did
,
int
vid
,
int
fid
,
uint32_t
ver
,
TSDB_FILE_T
ftype
)
{
void
tsdbInitDFile
(
SDFile
*
pDFile
,
SDiskID
did
,
int
vid
,
int
fid
,
uint32_t
ver
,
TSDB_FILE_T
ftype
)
{
char
fname
[
TSDB_FILENAME_LEN
];
char
fname
[
TSDB_FILENAME_LEN
];
TSDB_FILE_SET_STATE
(
pDFile
,
TSDB_FILE_STATE_OK
);
TSDB_FILE_SET_CLOSED
(
pDFile
);
TSDB_FILE_SET_CLOSED
(
pDFile
);
memset
(
&
(
pDFile
->
info
),
0
,
sizeof
(
pDFile
->
info
));
memset
(
&
(
pDFile
->
info
),
0
,
sizeof
(
pDFile
->
info
));
...
@@ -439,6 +443,7 @@ static int tsdbScanAndTryFixDFile(STsdbRepo *pRepo, SDFile *pDFile) {
...
@@ -439,6 +443,7 @@ static int tsdbScanAndTryFixDFile(STsdbRepo *pRepo, SDFile *pDFile) {
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 exit, report to upper layer to fix it"
,
REPO_ID
(
pRepo
),
TSDB_FILE_FULL_NAME
(
pDFile
));
TSDB_FILE_FULL_NAME
(
pDFile
));
pRepo
->
state
|=
TSDB_STATE_BAD_DATA
;
pRepo
->
state
|=
TSDB_STATE_BAD_DATA
;
TSDB_FILE_SET_STATE
(
pDFile
,
TSDB_FILE_STATE_BAD
);
return
0
;
return
0
;
}
}
...
@@ -470,6 +475,7 @@ static int tsdbScanAndTryFixDFile(STsdbRepo *pRepo, SDFile *pDFile) {
...
@@ -470,6 +475,7 @@ static int tsdbScanAndTryFixDFile(STsdbRepo *pRepo, SDFile *pDFile) {
tsdbError
(
"vgId:%d data file %s has wrong size %"
PRId64
" expected %"
PRId64
", report to upper layer to fix it"
,
tsdbError
(
"vgId:%d data file %s has wrong size %"
PRId64
" expected %"
PRId64
", report to upper layer to fix it"
,
REPO_ID
(
pRepo
),
TSDB_FILE_FULL_NAME
(
pDFile
),
dfstat
.
st_size
,
pDFile
->
info
.
size
);
REPO_ID
(
pRepo
),
TSDB_FILE_FULL_NAME
(
pDFile
),
dfstat
.
st_size
,
pDFile
->
info
.
size
);
pRepo
->
state
|=
TSDB_STATE_BAD_DATA
;
pRepo
->
state
|=
TSDB_STATE_BAD_DATA
;
TSDB_FILE_SET_STATE
(
pDFile
,
TSDB_FILE_STATE_BAD
);
terrno
=
TSDB_CODE_TDB_FILE_CORRUPTED
;
terrno
=
TSDB_CODE_TDB_FILE_CORRUPTED
;
return
0
;
return
0
;
}
else
{
}
else
{
...
...
src/tsdb/src/tsdbMemTable.c
浏览文件 @
3bf73df9
...
@@ -216,11 +216,13 @@ void *tsdbAllocBytes(STsdbRepo *pRepo, int bytes) {
...
@@ -216,11 +216,13 @@ void *tsdbAllocBytes(STsdbRepo *pRepo, int bytes) {
}
}
int
tsdbAsyncCommit
(
STsdbRepo
*
pRepo
)
{
int
tsdbAsyncCommit
(
STsdbRepo
*
pRepo
)
{
if
(
pRepo
->
mem
==
NULL
)
return
0
;
tsem_wait
(
&
(
pRepo
->
readyToCommit
));
tsem_wait
(
&
(
pRepo
->
readyToCommit
));
ASSERT
(
pRepo
->
imem
==
NULL
);
ASSERT
(
pRepo
->
imem
==
NULL
);
if
(
pRepo
->
mem
==
NULL
)
{
tsem_post
(
&
(
pRepo
->
readyToCommit
));
return
0
;
}
if
(
pRepo
->
code
!=
TSDB_CODE_SUCCESS
)
{
if
(
pRepo
->
code
!=
TSDB_CODE_SUCCESS
)
{
tsdbWarn
(
"vgId:%d try to commit when TSDB not in good state: %s"
,
REPO_ID
(
pRepo
),
tstrerror
(
terrno
));
tsdbWarn
(
"vgId:%d try to commit when TSDB not in good state: %s"
,
REPO_ID
(
pRepo
),
tstrerror
(
terrno
));
...
...
src/tsdb/src/tsdbSync.c
浏览文件 @
3bf73df9
...
@@ -85,6 +85,7 @@ int32_t tsdbSyncRecv(void *tsdb, SOCKET socketFd) {
...
@@ -85,6 +85,7 @@ int32_t tsdbSyncRecv(void *tsdb, SOCKET socketFd) {
pRepo
->
state
=
TSDB_STATE_OK
;
pRepo
->
state
=
TSDB_STATE_OK
;
tsdbInitSyncH
(
&
synch
,
pRepo
,
socketFd
);
tsdbInitSyncH
(
&
synch
,
pRepo
,
socketFd
);
tsem_wait
(
&
(
pRepo
->
readyToCommit
));
tsdbStartFSTxn
(
pRepo
,
0
,
0
);
tsdbStartFSTxn
(
pRepo
,
0
,
0
);
if
(
tsdbSyncRecvMeta
(
&
synch
)
<
0
)
{
if
(
tsdbSyncRecvMeta
(
&
synch
)
<
0
)
{
...
@@ -98,6 +99,7 @@ int32_t tsdbSyncRecv(void *tsdb, SOCKET socketFd) {
...
@@ -98,6 +99,7 @@ int32_t tsdbSyncRecv(void *tsdb, SOCKET socketFd) {
}
}
tsdbEndFSTxn
(
pRepo
);
tsdbEndFSTxn
(
pRepo
);
tsem_post
(
&
(
pRepo
->
readyToCommit
));
tsdbDestroySyncH
(
&
synch
);
tsdbDestroySyncH
(
&
synch
);
// Reload file change
// Reload file change
...
@@ -107,6 +109,7 @@ int32_t tsdbSyncRecv(void *tsdb, SOCKET socketFd) {
...
@@ -107,6 +109,7 @@ int32_t tsdbSyncRecv(void *tsdb, SOCKET socketFd) {
_err:
_err:
tsdbEndFSTxnWithError
(
REPO_FS
(
pRepo
));
tsdbEndFSTxnWithError
(
REPO_FS
(
pRepo
));
tsem_post
(
&
(
pRepo
->
readyToCommit
));
tsdbDestroySyncH
(
&
synch
);
tsdbDestroySyncH
(
&
synch
);
return
-
1
;
return
-
1
;
}
}
...
@@ -191,7 +194,8 @@ static int32_t tsdbSyncRecvMeta(SSyncH *pSynch) {
...
@@ -191,7 +194,8 @@ static int32_t tsdbSyncRecvMeta(SSyncH *pSynch) {
return
0
;
return
0
;
}
}
if
(
pLMFile
==
NULL
||
memcmp
(
&
(
pSynch
->
pmf
->
info
),
&
(
pLMFile
->
info
),
sizeof
(
SMFInfo
))
!=
0
)
{
if
(
pLMFile
==
NULL
||
memcmp
(
&
(
pSynch
->
pmf
->
info
),
&
(
pLMFile
->
info
),
sizeof
(
SMFInfo
))
!=
0
||
TSDB_FILE_IS_BAD
(
pLMFile
))
{
// Local has no meta file or has a different meta file, need to copy from remote
// Local has no meta file or has a different meta file, need to copy from remote
pSynch
->
mfChanged
=
true
;
pSynch
->
mfChanged
=
true
;
...
@@ -409,7 +413,8 @@ static int32_t tsdbSyncRecvDFileSetArray(SSyncH *pSynch) {
...
@@ -409,7 +413,8 @@ static int32_t tsdbSyncRecvDFileSetArray(SSyncH *pSynch) {
pSynch
->
pdf
!=
NULL
?
pSynch
->
pdf
->
fid
:
-
1
);
pSynch
->
pdf
!=
NULL
?
pSynch
->
pdf
->
fid
:
-
1
);
pLSet
=
tsdbFSIterNext
(
&
fsiter
);
pLSet
=
tsdbFSIterNext
(
&
fsiter
);
}
else
{
}
else
{
if
(
pLSet
&&
pSynch
->
pdf
&&
pLSet
->
fid
==
pSynch
->
pdf
->
fid
&&
tsdbIsTowFSetSame
(
pLSet
,
pSynch
->
pdf
))
{
if
(
pLSet
&&
pSynch
->
pdf
&&
pLSet
->
fid
==
pSynch
->
pdf
->
fid
&&
tsdbIsTowFSetSame
(
pLSet
,
pSynch
->
pdf
)
&&
tsdbFSetIsOk
(
pLSet
))
{
// Just keep local files and notify remote not to send
// Just keep local files and notify remote not to send
tsdbInfo
(
"vgId:%d, fileset:%d is same and no need to recv"
,
REPO_ID
(
pRepo
),
pLSet
->
fid
);
tsdbInfo
(
"vgId:%d, fileset:%d is same and no need to recv"
,
REPO_ID
(
pRepo
),
pLSet
->
fid
);
...
...
src/util/src/tconfig.c
浏览文件 @
3bf73df9
...
@@ -134,6 +134,11 @@ static bool taosReadDirectoryConfig(SGlobalCfg *cfg, char *input_value) {
...
@@ -134,6 +134,11 @@ static bool taosReadDirectoryConfig(SGlobalCfg *cfg, char *input_value) {
wordfree
(
&
full_path
);
wordfree
(
&
full_path
);
char
tmp
[
1025
]
=
{
0
};
if
(
realpath
(
option
,
tmp
)
!=
NULL
)
{
strcpy
(
option
,
tmp
);
}
int
code
=
taosMkDir
(
option
,
0755
);
int
code
=
taosMkDir
(
option
,
0755
);
if
(
code
!=
0
)
{
if
(
code
!=
0
)
{
terrno
=
TAOS_SYSTEM_ERROR
(
errno
);
terrno
=
TAOS_SYSTEM_ERROR
(
errno
);
...
...
src/vnode/src/vnodeCfg.c
浏览文件 @
3bf73df9
...
@@ -34,6 +34,7 @@ static void vnodeLoadCfg(SVnodeObj *pVnode, SCreateVnodeMsg* vnodeMsg) {
...
@@ -34,6 +34,7 @@ static void vnodeLoadCfg(SVnodeObj *pVnode, SCreateVnodeMsg* vnodeMsg) {
pVnode
->
tsdbCfg
.
maxRowsPerFileBlock
=
vnodeMsg
->
cfg
.
maxRowsPerFileBlock
;
pVnode
->
tsdbCfg
.
maxRowsPerFileBlock
=
vnodeMsg
->
cfg
.
maxRowsPerFileBlock
;
pVnode
->
tsdbCfg
.
precision
=
vnodeMsg
->
cfg
.
precision
;
pVnode
->
tsdbCfg
.
precision
=
vnodeMsg
->
cfg
.
precision
;
pVnode
->
tsdbCfg
.
compression
=
vnodeMsg
->
cfg
.
compression
;
pVnode
->
tsdbCfg
.
compression
=
vnodeMsg
->
cfg
.
compression
;
pVnode
->
tsdbCfg
.
update
=
vnodeMsg
->
cfg
.
update
;
pVnode
->
tsdbCfg
.
cacheLastRow
=
vnodeMsg
->
cfg
.
cacheLastRow
;
pVnode
->
tsdbCfg
.
cacheLastRow
=
vnodeMsg
->
cfg
.
cacheLastRow
;
pVnode
->
walCfg
.
walLevel
=
vnodeMsg
->
cfg
.
walLevel
;
pVnode
->
walCfg
.
walLevel
=
vnodeMsg
->
cfg
.
walLevel
;
pVnode
->
walCfg
.
fsyncPeriod
=
vnodeMsg
->
cfg
.
fsyncPeriod
;
pVnode
->
walCfg
.
fsyncPeriod
=
vnodeMsg
->
cfg
.
fsyncPeriod
;
...
@@ -227,6 +228,15 @@ int32_t vnodeReadCfg(SVnodeObj *pVnode) {
...
@@ -227,6 +228,15 @@ int32_t vnodeReadCfg(SVnodeObj *pVnode) {
}
}
vnodeMsg
.
cfg
.
quorum
=
(
int8_t
)
quorum
->
valueint
;
vnodeMsg
.
cfg
.
quorum
=
(
int8_t
)
quorum
->
valueint
;
cJSON
*
update
=
cJSON_GetObjectItem
(
root
,
"update"
);
if
(
!
update
||
update
->
type
!=
cJSON_Number
)
{
vError
(
"vgId: %d, failed to read %s, update not found"
,
pVnode
->
vgId
,
file
);
vnodeMsg
.
cfg
.
update
=
0
;
vnodeMsg
.
cfg
.
vgCfgVersion
=
0
;
}
else
{
vnodeMsg
.
cfg
.
update
=
(
int8_t
)
update
->
valueint
;
}
cJSON
*
cacheLastRow
=
cJSON_GetObjectItem
(
root
,
"cacheLastRow"
);
cJSON
*
cacheLastRow
=
cJSON_GetObjectItem
(
root
,
"cacheLastRow"
);
if
(
!
cacheLastRow
||
cacheLastRow
->
type
!=
cJSON_Number
)
{
if
(
!
cacheLastRow
||
cacheLastRow
->
type
!=
cJSON_Number
)
{
vError
(
"vgId: %d, failed to read %s, cacheLastRow not found"
,
pVnode
->
vgId
,
file
);
vError
(
"vgId: %d, failed to read %s, cacheLastRow not found"
,
pVnode
->
vgId
,
file
);
...
@@ -325,6 +335,7 @@ int32_t vnodeWriteCfg(SCreateVnodeMsg *pMsg) {
...
@@ -325,6 +335,7 @@ int32_t vnodeWriteCfg(SCreateVnodeMsg *pMsg) {
len
+=
snprintf
(
content
+
len
,
maxLen
-
len
,
"
\"
dbReplica
\"
: %d,
\n
"
,
pMsg
->
cfg
.
dbReplica
);
len
+=
snprintf
(
content
+
len
,
maxLen
-
len
,
"
\"
dbReplica
\"
: %d,
\n
"
,
pMsg
->
cfg
.
dbReplica
);
len
+=
snprintf
(
content
+
len
,
maxLen
-
len
,
"
\"
wals
\"
: %d,
\n
"
,
pMsg
->
cfg
.
wals
);
len
+=
snprintf
(
content
+
len
,
maxLen
-
len
,
"
\"
wals
\"
: %d,
\n
"
,
pMsg
->
cfg
.
wals
);
len
+=
snprintf
(
content
+
len
,
maxLen
-
len
,
"
\"
quorum
\"
: %d,
\n
"
,
pMsg
->
cfg
.
quorum
);
len
+=
snprintf
(
content
+
len
,
maxLen
-
len
,
"
\"
quorum
\"
: %d,
\n
"
,
pMsg
->
cfg
.
quorum
);
len
+=
snprintf
(
content
+
len
,
maxLen
-
len
,
"
\"
update
\"
: %d,
\n
"
,
pMsg
->
cfg
.
update
);
len
+=
snprintf
(
content
+
len
,
maxLen
-
len
,
"
\"
cacheLastRow
\"
: %d,
\n
"
,
pMsg
->
cfg
.
cacheLastRow
);
len
+=
snprintf
(
content
+
len
,
maxLen
-
len
,
"
\"
cacheLastRow
\"
: %d,
\n
"
,
pMsg
->
cfg
.
cacheLastRow
);
len
+=
snprintf
(
content
+
len
,
maxLen
-
len
,
"
\"
nodeInfos
\"
: [{
\n
"
);
len
+=
snprintf
(
content
+
len
,
maxLen
-
len
,
"
\"
nodeInfos
\"
: [{
\n
"
);
for
(
int32_t
i
=
0
;
i
<
pMsg
->
cfg
.
vgReplica
;
i
++
)
{
for
(
int32_t
i
=
0
;
i
<
pMsg
->
cfg
.
vgReplica
;
i
++
)
{
...
...
tests/pytest/insert/retentionpolicy.py
浏览文件 @
3bf73df9
...
@@ -44,7 +44,8 @@ class TDTestRetetion:
...
@@ -44,7 +44,8 @@ class TDTestRetetion:
caller
=
inspect
.
getframeinfo
(
inspect
.
stack
()[
1
][
0
])
caller
=
inspect
.
getframeinfo
(
inspect
.
stack
()[
1
][
0
])
args
=
(
caller
.
filename
,
caller
.
lineno
,
sql
,
self
.
queryRows
,
expectRows
)
args
=
(
caller
.
filename
,
caller
.
lineno
,
sql
,
self
.
queryRows
,
expectRows
)
os
.
system
(
"sudo timedatectl set-ntp true"
)
os
.
system
(
"sudo timedatectl set-ntp true"
)
time
.
sleep
(
40
)
os
.
system
(
"date -s '%s'"
%
(
datetime
.
datetime
.
now
()
+
datetime
.
timedelta
(
hours
=
1
)))
time
.
sleep
(
5
)
tdLog
.
exit
(
"%s(%d) failed: sql:%s, queryRows:%d != expect:%d"
%
args
)
tdLog
.
exit
(
"%s(%d) failed: sql:%s, queryRows:%d != expect:%d"
%
args
)
def
run
(
self
):
def
run
(
self
):
...
@@ -63,7 +64,7 @@ class TDTestRetetion:
...
@@ -63,7 +64,7 @@ class TDTestRetetion:
tdLog
.
info
(
"=============== step2"
)
tdLog
.
info
(
"=============== step2"
)
tdDnodes
.
stop
(
1
)
tdDnodes
.
stop
(
1
)
os
.
system
(
"sudo timedatectl set-ntp false"
)
os
.
system
(
"sudo timedatectl set-ntp false"
)
os
.
system
(
"
sudo date -s $(date -d
\"
${DATE} 2 days
\"
\"
+%Y%m%d
\"
)"
)
os
.
system
(
"
date -s '%s'"
%
(
datetime
.
datetime
.
now
()
+
datetime
.
timedelta
(
hours
=
48
))
)
tdDnodes
.
start
(
1
)
tdDnodes
.
start
(
1
)
cmd
=
'insert into test values(now,5);'
cmd
=
'insert into test values(now,5);'
tdDnodes
.
stop
(
1
)
tdDnodes
.
stop
(
1
)
...
@@ -79,7 +80,7 @@ class TDTestRetetion:
...
@@ -79,7 +80,7 @@ class TDTestRetetion:
self
.
checkRows
(
5
,
cmd
)
self
.
checkRows
(
5
,
cmd
)
tdLog
.
info
(
"=============== step3"
)
tdLog
.
info
(
"=============== step3"
)
tdDnodes
.
stop
(
1
)
tdDnodes
.
stop
(
1
)
os
.
system
(
"
sudo date -s $(date -d
\"
${DATE} 2 days
\"
\"
+%Y%m%d
\"
)"
)
os
.
system
(
"
date -s '%s'"
%
(
datetime
.
datetime
.
now
()
+
datetime
.
timedelta
(
hours
=
48
))
)
tdDnodes
.
start
(
1
)
tdDnodes
.
start
(
1
)
tdLog
.
info
(
cmd
)
tdLog
.
info
(
cmd
)
tdSql
.
execute
(
cmd
)
tdSql
.
execute
(
cmd
)
...
@@ -99,18 +100,19 @@ class TDTestRetetion:
...
@@ -99,18 +100,19 @@ class TDTestRetetion:
tdLog
.
info
(
cmd
)
tdLog
.
info
(
cmd
)
tdSql
.
execute
(
cmd
)
tdSql
.
execute
(
cmd
)
self
.
queryRows
=
tdSql
.
query
(
'select * from test'
)
self
.
queryRows
=
tdSql
.
query
(
'select * from test'
)
self
.
checkRows
(
7
,
cmd
)
self
.
checkRows
(
5
,
cmd
)
tdLog
.
info
(
"=============== step5"
)
tdLog
.
info
(
"=============== step5"
)
tdDnodes
.
stop
(
1
)
tdDnodes
.
stop
(
1
)
tdDnodes
.
start
(
1
)
tdDnodes
.
start
(
1
)
cmd
=
'select * from test where ts > now-1d'
cmd
=
'select * from test where ts > now-1d'
self
.
queryRows
=
tdSql
.
query
(
'select * from test where ts > now-1d'
)
self
.
queryRows
=
tdSql
.
query
(
'select * from test where ts > now-1d'
)
self
.
checkRows
(
1
,
cmd
)
self
.
checkRows
(
2
,
cmd
)
def
stop
(
self
):
def
stop
(
self
):
os
.
system
(
"sudo timedatectl set-ntp true"
)
os
.
system
(
"sudo timedatectl set-ntp true"
)
time
.
sleep
(
40
)
os
.
system
(
"date -s '%s'"
%
(
datetime
.
datetime
.
now
()
+
datetime
.
timedelta
(
hours
=
1
)))
time
.
sleep
(
5
)
tdSql
.
close
()
tdSql
.
close
()
tdLog
.
success
(
"%s successfully executed"
%
__file__
)
tdLog
.
success
(
"%s successfully executed"
%
__file__
)
...
...
tests/pytest/pytest_2.sh
浏览文件 @
3bf73df9
# update
# update
#
python3 ./test.py -f update/allow_update.py
python3 ./test.py
-f
update/allow_update.py
python3 ./test.py
-f
update/allow_update-0.py
python3 ./test.py
-f
update/allow_update-0.py
python3 ./test.py
-f
update/append_commit_data.py
python3 ./test.py
-f
update/append_commit_data.py
python3 ./test.py
-f
update/append_commit_last-0.py
python3 ./test.py
-f
update/append_commit_last-0.py
python3 ./test.py
-f
update/append_commit_last.py
python3 ./test.py
-f
update/append_commit_last.py
#
python3 ./test.py -f update/merge_commit_data.py
python3 ./test.py
-f
update/merge_commit_data.py
#
python3 ./test.py -f update/merge_commit_data-0.py
python3 ./test.py
-f
update/merge_commit_data-0.py
#
python3 ./test.py -f update/merge_commit_data2.py
python3 ./test.py
-f
update/merge_commit_data2.py
#
python3 ./test.py -f update/merge_commit_data2_update0.py
python3 ./test.py
-f
update/merge_commit_data2_update0.py
#
python3 ./test.py -f update/merge_commit_last-0.py
python3 ./test.py
-f
update/merge_commit_last-0.py
#
python3 ./test.py -f update/merge_commit_last.py
python3 ./test.py
-f
update/merge_commit_last.py
#
python3 ./test.py -f update/bug_td2279.py
python3 ./test.py
-f
update/bug_td2279.py
# wal
# wal
python3 ./test.py
-f
wal/addOldWalTest.py
python3 ./test.py
-f
wal/addOldWalTest.py
...
...
tests/script/wtest.bat
浏览文件 @
3bf73df9
...
@@ -44,10 +44,10 @@ echo serverPort 7100 >> %TAOS_CFG%
...
@@ -44,10 +44,10 @@ echo serverPort 7100 >> %TAOS_CFG%
echo
logDir
%LOG_DIR%
>>
%TAOS_CFG%
echo
logDir
%LOG_DIR%
>>
%TAOS_CFG%
echo
scriptDir
%SCRIPT_DIR%
>>
%TAOS_CFG%
echo
scriptDir
%SCRIPT_DIR%
>>
%TAOS_CFG%
echo
numOfLogLines
100000000
>>
%TAOS_CFG%
echo
numOfLogLines
100000000
>>
%TAOS_CFG%
echo
rpcDebugFlag
1
43
>>
%TAOS_CFG%
echo
rpcDebugFlag
1
35
>>
%TAOS_CFG%
echo
tmrDebugFlag
131
>>
%TAOS_CFG%
echo
tmrDebugFlag
131
>>
%TAOS_CFG%
echo
cDebugFlag
1
43
>>
%TAOS_CFG%
echo
cDebugFlag
1
35
>>
%TAOS_CFG%
echo
udebugFlag
1
43
>>
%TAOS_CFG%
echo
udebugFlag
1
35
>>
%TAOS_CFG%
echo
wal
0
>>
%TAOS_CFG%
echo
wal
0
>>
%TAOS_CFG%
echo
asyncLog
0
>>
%TAOS_CFG%
echo
asyncLog
0
>>
%TAOS_CFG%
echo
locale
en_US
.UTF
-
8
>>
%TAOS_CFG%
echo
locale
en_US
.UTF
-
8
>>
%TAOS_CFG%
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录