Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
267e450a
T
TDengine
项目概览
taosdata
/
TDengine
1 年多 前同步成功
通知
1185
Star
22016
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
T
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
提交
267e450a
编写于
7月 04, 2022
作者:
H
Hongze Cheng
浏览文件
操作
浏览文件
下载
差异文件
Merge branch '3.0' of
https://github.com/taosdata/TDengine
into feat/tsdb_refact
上级
af1b6f99
8e71d539
变更
19
隐藏空白更改
内联
并排
Showing
19 changed file
with
257 addition
and
180 deletion
+257
-180
include/util/taoserror.h
include/util/taoserror.h
+2
-0
source/libs/index/inc/indexFst.h
source/libs/index/inc/indexFst.h
+7
-7
source/libs/index/inc/indexFstFile.h
source/libs/index/inc/indexFstFile.h
+96
-0
source/libs/index/inc/indexFstNode.h
source/libs/index/inc/indexFstNode.h
+4
-4
source/libs/index/inc/indexTfile.h
source/libs/index/inc/indexTfile.h
+5
-5
source/libs/index/src/index.c
source/libs/index/src/index.c
+2
-2
source/libs/index/src/indexFst.c
source/libs/index/src/indexFst.c
+30
-32
source/libs/index/src/indexFstDfa.c
source/libs/index/src/indexFstDfa.c
+2
-1
source/libs/index/src/indexFstFile.c
source/libs/index/src/indexFstFile.c
+38
-46
source/libs/index/src/indexFstNode.c
source/libs/index/src/indexFstNode.c
+1
-1
source/libs/index/src/indexFstUtil.c
source/libs/index/src/indexFstUtil.c
+0
-14
source/libs/index/src/indexTfile.c
source/libs/index/src/indexTfile.c
+17
-16
source/libs/index/test/fstTest.cc
source/libs/index/test/fstTest.cc
+13
-14
source/libs/index/test/fstUT.cc
source/libs/index/test/fstUT.cc
+13
-14
source/libs/index/test/indexTests.cc
source/libs/index/test/indexTests.cc
+15
-16
source/libs/index/test/jsonUT.cc
source/libs/index/test/jsonUT.cc
+0
-1
source/libs/index/test/utilUT.cc
source/libs/index/test/utilUT.cc
+0
-1
source/libs/transport/src/transComm.c
source/libs/transport/src/transComm.c
+11
-6
source/util/src/terror.c
source/util/src/terror.c
+1
-0
未找到文件。
include/util/taoserror.h
浏览文件 @
267e450a
...
...
@@ -628,6 +628,8 @@ int32_t* taosGetErrno();
//index
#define TSDB_CODE_INDEX_REBUILDING TAOS_DEF_ERROR_CODE(0, 0x3200)
#define TSDB_CODE_INDEX_INVALID_FILE TAOS_DEF_ERROR_CODE(0, 0x3201)
//tmq
#define TSDB_CODE_TMQ_INVALID_MSG TAOS_DEF_ERROR_CODE(0, 0x4000)
...
...
source/libs/index/inc/indexFst.h
浏览文件 @
267e450a
...
...
@@ -21,7 +21,7 @@ extern "C" {
#endif
#include "indexFstAutomation.h"
#include "indexFst
CountingWriter
.h"
#include "indexFst
File
.h"
#include "indexFstNode.h"
#include "indexFstRegistry.h"
#include "indexFstUtil.h"
...
...
@@ -90,8 +90,8 @@ FstBuilderNode* fstUnFinishedNodesPopEmpty(FstUnFinishedNodes* nodes);
uint64_t
fstUnFinishedNodesFindCommPrefixAndSetOutput
(
FstUnFinishedNodes
*
node
,
FstSlice
bs
,
Output
in
,
Output
*
out
);
typedef
struct
FstBuilder
{
FstCountingWriter
*
wrt
;
// The FST raw data is written directly to `wtr`.
FstUnFinishedNodes
*
unfinished
;
// The stack of unfinished nodes
IdxFstFile
*
wrt
;
// The FST raw data is written directly to `wtr`.
FstUnFinishedNodes
*
unfinished
;
// The stack of unfinished
nodes
FstRegistry
*
registry
;
// A map of finished nodes.
FstSlice
last
;
// The last word added
CompiledAddr
lastAddr
;
// The address of the last compiled node
...
...
@@ -125,9 +125,9 @@ FstState fstStateCreateFrom(FstSlice* data, CompiledAddr addr);
FstState
fstStateCreate
(
State
state
);
// compile
void
fstStateCompileForOneTransNext
(
FstCountingWriter
*
w
,
CompiledAddr
addr
,
uint8_t
inp
);
void
fstStateCompileForOneTrans
(
FstCountingWriter
*
w
,
CompiledAddr
addr
,
FstTransition
*
trn
);
void
fstStateCompileForAnyTrans
(
FstCountingWriter
*
w
,
CompiledAddr
addr
,
FstBuilderNode
*
node
);
void
fstStateCompileForOneTransNext
(
IdxFstFile
*
w
,
CompiledAddr
addr
,
uint8_t
inp
);
void
fstStateCompileForOneTrans
(
IdxFstFile
*
w
,
CompiledAddr
addr
,
FstTransition
*
trn
);
void
fstStateCompileForAnyTrans
(
IdxFstFile
*
w
,
CompiledAddr
addr
,
FstBuilderNode
*
node
);
// set_comm_input
void
fstStateSetCommInput
(
FstState
*
state
,
uint8_t
inp
);
...
...
@@ -282,7 +282,7 @@ FStmSt* stmBuilderIntoStm(FStmBuilder* sb);
bool
fstVerify
(
Fst
*
fst
);
// refactor this function
bool
fstBuilderNodeCompileTo
(
FstBuilderNode
*
b
,
FstCountingWriter
*
wrt
,
CompiledAddr
lastAddr
,
CompiledAddr
startAddr
);
bool
fstBuilderNodeCompileTo
(
FstBuilderNode
*
b
,
IdxFstFile
*
wrt
,
CompiledAddr
lastAddr
,
CompiledAddr
startAddr
);
typedef
struct
StreamState
{
FstNode
*
node
;
...
...
source/libs/index/inc/indexFst
CountingWriter
.h
→
source/libs/index/inc/indexFst
File
.h
浏览文件 @
267e450a
...
...
@@ -13,8 +13,8 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef __INDEX_FST_
COUNTING_WRITER
_H__
#define __INDEX_FST_
COUNTING_WRITER
_H__
#ifndef __INDEX_FST_
FILE
_H__
#define __INDEX_FST_
FILE
_H__
#include "indexInt.h"
...
...
@@ -29,65 +29,65 @@ extern "C" {
static
char
tmpFile
[]
=
"./index"
;
typedef
enum
WriterType
{
TMemory
,
TFile
}
WriterType
;
typedef
struct
Writer
Ctx
{
int
(
*
write
)(
struct
Writer
Ctx
*
ctx
,
uint8_t
*
buf
,
int
len
);
int
(
*
read
)(
struct
Writer
Ctx
*
ctx
,
uint8_t
*
buf
,
int
len
);
int
(
*
flush
)(
struct
Writer
Ctx
*
ctx
);
int
(
*
readFrom
)(
struct
Writer
Ctx
*
ctx
,
uint8_t
*
buf
,
int
len
,
int32_t
offset
);
int
(
*
size
)(
struct
Writer
Ctx
*
ctx
);
typedef
struct
IFile
Ctx
{
int
(
*
write
)(
struct
IFile
Ctx
*
ctx
,
uint8_t
*
buf
,
int
len
);
int
(
*
read
)(
struct
IFile
Ctx
*
ctx
,
uint8_t
*
buf
,
int
len
);
int
(
*
flush
)(
struct
IFile
Ctx
*
ctx
);
int
(
*
readFrom
)(
struct
IFile
Ctx
*
ctx
,
uint8_t
*
buf
,
int
len
,
int32_t
offset
);
int
(
*
size
)(
struct
IFile
Ctx
*
ctx
);
WriterType
type
;
union
{
struct
{
TdFilePtr
pFile
;
bool
readOnly
;
char
buf
[
256
];
int
size
;
bool
readOnly
;
char
buf
[
256
];
int
64_t
size
;
#ifdef USE_MMAP
char
*
ptr
;
#endif
}
file
;
struct
{
int32_t
cap
a
;
int32_t
cap
;
char
*
buf
;
}
mem
;
};
int32_t
offset
;
int32_t
limit
;
}
Writer
Ctx
;
}
IFile
Ctx
;
static
int
writeCtxDoWrite
(
Writer
Ctx
*
ctx
,
uint8_t
*
buf
,
int
len
);
static
int
writeCtxDoRead
(
Writer
Ctx
*
ctx
,
uint8_t
*
buf
,
int
len
);
static
int
writeCtxDoReadFrom
(
Writer
Ctx
*
ctx
,
uint8_t
*
buf
,
int
len
,
int32_t
offset
);
static
int
writeCtxDoFlush
(
Writer
Ctx
*
ctx
);
static
int
idxFileCtxDoWrite
(
IFile
Ctx
*
ctx
,
uint8_t
*
buf
,
int
len
);
static
int
idxFileCtxDoRead
(
IFile
Ctx
*
ctx
,
uint8_t
*
buf
,
int
len
);
static
int
idxFileCtxDoReadFrom
(
IFile
Ctx
*
ctx
,
uint8_t
*
buf
,
int
len
,
int32_t
offset
);
static
int
idxFileCtxDoFlush
(
IFile
Ctx
*
ctx
);
WriterCtx
*
writer
CtxCreate
(
WriterType
type
,
const
char
*
path
,
bool
readOnly
,
int32_t
capacity
);
void
writerCtxDestroy
(
Writer
Ctx
*
w
,
bool
remove
);
IFileCtx
*
idxFile
CtxCreate
(
WriterType
type
,
const
char
*
path
,
bool
readOnly
,
int32_t
capacity
);
void
idxFileCtxDestroy
(
IFile
Ctx
*
w
,
bool
remove
);
typedef
uint32_t
CheckSummer
;
typedef
struct
FstCountingWriter
{
typedef
struct
IdxFstFile
{
void
*
wrt
;
// wrap any writer that counts and checksum bytes written
uint64_t
count
;
CheckSummer
summer
;
}
FstCountingWriter
;
}
IdxFstFile
;
int
fstCountingWriterWrite
(
FstCountingWriter
*
write
,
uint8_t
*
buf
,
uint32_t
len
);
int
idxFileWrite
(
IdxFstFile
*
write
,
uint8_t
*
buf
,
uint32_t
len
);
int
fstCountingWriterRead
(
FstCountingWriter
*
write
,
uint8_t
*
buf
,
uint32_t
len
);
int
idxFileRead
(
IdxFstFile
*
write
,
uint8_t
*
buf
,
uint32_t
len
);
int
fstCountingWriterFlush
(
FstCountingWriter
*
write
);
int
idxFileFlush
(
IdxFstFile
*
write
);
uint32_t
fstCountingWriterMaskedCheckSum
(
FstCountingWriter
*
write
);
uint32_t
idxFileMaskedCheckSum
(
IdxFstFile
*
write
);
FstCountingWriter
*
fstCountingWriter
Create
(
void
*
wtr
);
void
fstCountingWriterDestroy
(
FstCountingWriter
*
w
);
IdxFstFile
*
idxFile
Create
(
void
*
wtr
);
void
idxFileDestroy
(
IdxFstFile
*
w
);
void
fstCountingWriterPackUintIn
(
FstCountingWriter
*
writer
,
uint64_t
n
,
uint8_t
nBytes
);
uint8_t
fstCountingWriterPackUint
(
FstCountingWriter
*
writer
,
uint64_t
n
);
void
idxFilePackUintIn
(
IdxFstFile
*
writer
,
uint64_t
n
,
uint8_t
nBytes
);
uint8_t
idxFilePackUint
(
IdxFstFile
*
writer
,
uint64_t
n
);
#define FST_WRITER_COUNT(writer) (writer->count)
#define FST_WRITER_COUNT(writer)
(writer->count)
#define FST_WRITER_INTER_WRITER(writer) (writer->wtr)
#define FST_WRITE_CHECK_SUMMER(writer) (writer->summer)
#define FST_WRITE_CHECK_SUMMER(writer)
(writer->summer)
#ifdef __cplusplus
}
...
...
source/libs/index/inc/indexFstNode.h
浏览文件 @
267e450a
...
...
@@ -20,12 +20,12 @@
extern
"C"
{
#endif
#include "indexFst
CountingWriter
.h"
#include "indexFst
File
.h"
#include "indexFstUtil.h"
#include "indexInt.h"
#define FST_BUILDER_NODE_IS_FINAL(bn) (bn->isFinal)
#define FST_BUILDER_NODE_TRANS_ISEMPTY(bn) (taosArrayGetSize(bn->trans) == 0)
#define FST_BUILDER_NODE_IS_FINAL(bn)
(bn->isFinal)
#define FST_BUILDER_NODE_TRANS_ISEMPTY(bn)
(taosArrayGetSize(bn->trans) == 0)
#define FST_BUILDER_NODE_FINALOUTPUT_ISZERO(bn) (bn->finalOutput == 0)
typedef
struct
FstTransition
{
...
...
@@ -46,7 +46,7 @@ FstBuilderNode* fstBuilderNodeClone(FstBuilderNode* src);
void
fstBuilderNodeCloneFrom
(
FstBuilderNode
*
dst
,
FstBuilderNode
*
src
);
// bool fstBuilderNodeCompileTo(FstBuilderNode *b,
FstCountingWriter
*wrt,
// bool fstBuilderNodeCompileTo(FstBuilderNode *b,
IdxFile'
*wrt,
// CompiledAddr lastAddr, CompiledAddr startAddr);
bool
fstBuilderNodeEqual
(
FstBuilderNode
*
n1
,
FstBuilderNode
*
n2
);
...
...
source/libs/index/inc/indexTfile.h
浏览文件 @
267e450a
...
...
@@ -16,7 +16,7 @@
#define __INDEX_TFILE_H__
#include "indexFst.h"
#include "indexFst
CountingWriter
.h"
#include "indexFst
File
.h"
#include "indexInt.h"
#include "indexTfile.h"
#include "indexUtil.h"
...
...
@@ -59,7 +59,7 @@ typedef struct TFileCache {
typedef
struct
TFileWriter
{
FstBuilder
*
fb
;
WriterCtx
*
ctx
;
IFileCtx
*
ctx
;
TFileHeader
header
;
uint32_t
offset
;
}
TFileWriter
;
...
...
@@ -68,7 +68,7 @@ typedef struct TFileWriter {
typedef
struct
TFileReader
{
T_REF_DECLARE
()
Fst
*
fst
;
WriterCtx
*
ctx
;
IFileCtx
*
ctx
;
TFileHeader
header
;
bool
remove
;
}
TFileReader
;
...
...
@@ -103,7 +103,7 @@ void tfileCachePut(TFileCache* tcache, ICacheKey* key, TFileReader* read
TFileReader
*
tfileGetReaderByCol
(
IndexTFile
*
tf
,
uint64_t
suid
,
char
*
colName
);
TFileReader
*
tfileReaderOpen
(
char
*
path
,
uint64_t
suid
,
int64_t
version
,
const
char
*
colName
);
TFileReader
*
tfileReaderCreate
(
Writer
Ctx
*
ctx
);
TFileReader
*
tfileReaderCreate
(
IFile
Ctx
*
ctx
);
void
tfileReaderDestroy
(
TFileReader
*
reader
);
int
tfileReaderSearch
(
TFileReader
*
reader
,
SIndexTermQuery
*
query
,
SIdxTRslt
*
tr
);
void
tfileReaderRef
(
TFileReader
*
reader
);
...
...
@@ -111,7 +111,7 @@ void tfileReaderUnRef(TFileReader* reader);
TFileWriter
*
tfileWriterOpen
(
char
*
path
,
uint64_t
suid
,
int64_t
version
,
const
char
*
colName
,
uint8_t
type
);
void
tfileWriterClose
(
TFileWriter
*
tw
);
TFileWriter
*
tfileWriterCreate
(
Writer
Ctx
*
ctx
,
TFileHeader
*
header
);
TFileWriter
*
tfileWriterCreate
(
IFile
Ctx
*
ctx
,
TFileHeader
*
header
);
void
tfileWriterDestroy
(
TFileWriter
*
tw
);
int
tfileWriterPut
(
TFileWriter
*
tw
,
void
*
data
,
bool
order
);
int
tfileWriterFinish
(
TFileWriter
*
tw
);
...
...
source/libs/index/src/index.c
浏览文件 @
267e450a
...
...
@@ -39,7 +39,7 @@
#define INDEX_DATA_BIGINT_NULL 0x8000000000000000LL
#define INDEX_DATA_TIMESTAMP_NULL TSDB_DATA_BIGINT_NULL
#define INDEX_DATA_FLOAT_NULL 0x7FF00000 // it is an NAN
#define INDEX_DATA_FLOAT_NULL 0x7FF00000
// it is an NAN
#define INDEX_DATA_DOUBLE_NULL 0x7FFFFF0000000000LL // an NAN
#define INDEX_DATA_NCHAR_NULL 0xFFFFFFFF
#define INDEX_DATA_BINARY_NULL 0xFF
...
...
@@ -614,7 +614,7 @@ static int idxGenTFile(SIndex* sIdx, IndexCache* cache, SArray* batch) {
return
ret
;
END:
if
(
tw
!=
NULL
)
{
writer
CtxDestroy
(
tw
->
ctx
,
true
);
idxFile
CtxDestroy
(
tw
->
ctx
,
true
);
taosMemoryFree
(
tw
);
}
return
-
1
;
...
...
source/libs/index/src/indexFst.c
浏览文件 @
267e450a
...
...
@@ -19,11 +19,11 @@
#include "tchecksum.h"
#include "tcoding.h"
static
void
fstPackDeltaIn
(
FstCountingWriter
*
wrt
,
CompiledAddr
nodeAddr
,
CompiledAddr
transAddr
,
uint8_t
nBytes
)
{
static
void
fstPackDeltaIn
(
IdxFstFile
*
wrt
,
CompiledAddr
nodeAddr
,
CompiledAddr
transAddr
,
uint8_t
nBytes
)
{
CompiledAddr
deltaAddr
=
(
transAddr
==
EMPTY_ADDRESS
)
?
EMPTY_ADDRESS
:
nodeAddr
-
transAddr
;
fstCountingWriter
PackUintIn
(
wrt
,
deltaAddr
,
nBytes
);
idxFile
PackUintIn
(
wrt
,
deltaAddr
,
nBytes
);
}
static
uint8_t
fstPackDetla
(
FstCountingWriter
*
wrt
,
CompiledAddr
nodeAddr
,
CompiledAddr
transAddr
)
{
static
uint8_t
fstPackDetla
(
IdxFstFile
*
wrt
,
CompiledAddr
nodeAddr
,
CompiledAddr
transAddr
)
{
uint8_t
nBytes
=
packDeltaSize
(
nodeAddr
,
transAddr
);
fstPackDeltaIn
(
wrt
,
nodeAddr
,
transAddr
,
nBytes
);
return
nBytes
;
...
...
@@ -208,7 +208,7 @@ FstState fstStateCreate(State state) {
return
fstStateDict
[
idx
];
}
// compile
void
fstStateCompileForOneTransNext
(
FstCountingWriter
*
w
,
CompiledAddr
addr
,
uint8_t
inp
)
{
void
fstStateCompileForOneTransNext
(
IdxFstFile
*
w
,
CompiledAddr
addr
,
uint8_t
inp
)
{
FstState
s
=
fstStateCreate
(
OneTransNext
);
fstStateSetCommInput
(
&
s
,
inp
);
...
...
@@ -216,21 +216,21 @@ void fstStateCompileForOneTransNext(FstCountingWriter* w, CompiledAddr addr, uin
uint8_t
v
=
fstStateCommInput
(
&
s
,
&
null
);
if
(
null
)
{
// w->write_all(&[inp])
fstCountingWriter
Write
(
w
,
&
inp
,
1
);
idxFile
Write
(
w
,
&
inp
,
1
);
}
fstCountingWriter
Write
(
w
,
&
(
s
.
val
),
1
);
idxFile
Write
(
w
,
&
(
s
.
val
),
1
);
// w->write_all(&[s.val])
return
;
}
void
fstStateCompileForOneTrans
(
FstCountingWriter
*
w
,
CompiledAddr
addr
,
FstTransition
*
trn
)
{
void
fstStateCompileForOneTrans
(
IdxFstFile
*
w
,
CompiledAddr
addr
,
FstTransition
*
trn
)
{
Output
out
=
trn
->
out
;
uint8_t
outPackSize
=
(
out
==
0
?
0
:
fstCountingWriter
PackUint
(
w
,
out
));
uint8_t
outPackSize
=
(
out
==
0
?
0
:
idxFile
PackUint
(
w
,
out
));
uint8_t
transPackSize
=
fstPackDetla
(
w
,
addr
,
trn
->
addr
);
PackSizes
packSizes
=
0
;
FST_SET_OUTPUT_PACK_SIZE
(
packSizes
,
outPackSize
);
FST_SET_TRANSITION_PACK_SIZE
(
packSizes
,
transPackSize
);
fstCountingWriter
Write
(
w
,
(
char
*
)
&
packSizes
,
sizeof
(
packSizes
));
idxFile
Write
(
w
,
(
char
*
)
&
packSizes
,
sizeof
(
packSizes
));
FstState
st
=
fstStateCreate
(
OneTrans
);
...
...
@@ -239,12 +239,12 @@ void fstStateCompileForOneTrans(FstCountingWriter* w, CompiledAddr addr, FstTran
bool
null
=
false
;
uint8_t
inp
=
fstStateCommInput
(
&
st
,
&
null
);
if
(
null
==
true
)
{
fstCountingWriter
Write
(
w
,
(
char
*
)
&
trn
->
inp
,
sizeof
(
trn
->
inp
));
idxFile
Write
(
w
,
(
char
*
)
&
trn
->
inp
,
sizeof
(
trn
->
inp
));
}
fstCountingWriter
Write
(
w
,
(
char
*
)(
&
(
st
.
val
)),
sizeof
(
st
.
val
));
idxFile
Write
(
w
,
(
char
*
)(
&
(
st
.
val
)),
sizeof
(
st
.
val
));
return
;
}
void
fstStateCompileForAnyTrans
(
FstCountingWriter
*
w
,
CompiledAddr
addr
,
FstBuilderNode
*
node
)
{
void
fstStateCompileForAnyTrans
(
IdxFstFile
*
w
,
CompiledAddr
addr
,
FstBuilderNode
*
node
)
{
int32_t
sz
=
taosArrayGetSize
(
node
->
trans
);
assert
(
sz
<=
256
);
...
...
@@ -275,11 +275,11 @@ void fstStateCompileForAnyTrans(FstCountingWriter* w, CompiledAddr addr, FstBuil
if
(
anyOuts
)
{
if
(
FST_BUILDER_NODE_IS_FINAL
(
node
))
{
fstCountingWriter
PackUintIn
(
w
,
node
->
finalOutput
,
oSize
);
idxFile
PackUintIn
(
w
,
node
->
finalOutput
,
oSize
);
}
for
(
int32_t
i
=
sz
-
1
;
i
>=
0
;
i
--
)
{
FstTransition
*
t
=
taosArrayGet
(
node
->
trans
,
i
);
fstCountingWriter
PackUintIn
(
w
,
t
->
out
,
oSize
);
idxFile
PackUintIn
(
w
,
t
->
out
,
oSize
);
}
}
for
(
int32_t
i
=
sz
-
1
;
i
>=
0
;
i
--
)
{
...
...
@@ -288,7 +288,7 @@ void fstStateCompileForAnyTrans(FstCountingWriter* w, CompiledAddr addr, FstBuil
}
for
(
int32_t
i
=
sz
-
1
;
i
>=
0
;
i
--
)
{
FstTransition
*
t
=
taosArrayGet
(
node
->
trans
,
i
);
fstCountingWriter
Write
(
w
,
(
char
*
)
&
t
->
inp
,
1
);
idxFile
Write
(
w
,
(
char
*
)
&
t
->
inp
,
1
);
// fstPackDeltaIn(w, addr, t->addr, tSize);
}
if
(
sz
>
TRANS_INDEX_THRESHOLD
)
{
...
...
@@ -306,10 +306,10 @@ void fstStateCompileForAnyTrans(FstCountingWriter* w, CompiledAddr addr, FstBuil
index
[
t
->
inp
]
=
i
;
// fstPackDeltaIn(w, addr, t->addr, tSize);
}
fstCountingWriter
Write
(
w
,
(
char
*
)
index
,
256
);
idxFile
Write
(
w
,
(
char
*
)
index
,
256
);
taosMemoryFree
(
index
);
}
fstCountingWriter
Write
(
w
,
(
char
*
)
&
packSizes
,
1
);
idxFile
Write
(
w
,
(
char
*
)
&
packSizes
,
1
);
bool
null
=
false
;
fstStateStateNtrans
(
&
st
,
&
null
);
if
(
null
==
true
)
{
...
...
@@ -318,12 +318,12 @@ void fstStateCompileForAnyTrans(FstCountingWriter* w, CompiledAddr addr, FstBuil
// encoded in the state byte.
uint8_t
v
=
1
;
if
(
sz
==
256
)
{
fstCountingWriter
Write
(
w
,
(
char
*
)
&
v
,
1
);
idxFile
Write
(
w
,
(
char
*
)
&
v
,
1
);
}
else
{
fstCountingWriter
Write
(
w
,
(
char
*
)
&
sz
,
1
);
idxFile
Write
(
w
,
(
char
*
)
&
sz
,
1
);
}
}
fstCountingWriter
Write
(
w
,
(
char
*
)(
&
(
st
.
val
)),
1
);
idxFile
Write
(
w
,
(
char
*
)(
&
(
st
.
val
)),
1
);
return
;
}
...
...
@@ -753,7 +753,7 @@ bool fstNodeCompile(FstNode* node, void* w, CompiledAddr lastAddr, CompiledAddr
return
true
;
}
bool
fstBuilderNodeCompileTo
(
FstBuilderNode
*
b
,
FstCountingWriter
*
wrt
,
CompiledAddr
lastAddr
,
CompiledAddr
startAddr
)
{
bool
fstBuilderNodeCompileTo
(
FstBuilderNode
*
b
,
IdxFstFile
*
wrt
,
CompiledAddr
lastAddr
,
CompiledAddr
startAddr
)
{
return
fstNodeCompile
(
NULL
,
wrt
,
lastAddr
,
startAddr
,
b
);
}
...
...
@@ -763,7 +763,7 @@ FstBuilder* fstBuilderCreate(void* w, FstType ty) {
return
b
;
}
b
->
wrt
=
fstCountingWriter
Create
(
w
);
b
->
wrt
=
idxFile
Create
(
w
);
b
->
unfinished
=
fstUnFinishedNodesCreate
();
b
->
registry
=
fstRegistryCreate
(
10000
,
2
);
b
->
last
=
fstSliceCreate
(
NULL
,
0
);
...
...
@@ -773,12 +773,12 @@ FstBuilder* fstBuilderCreate(void* w, FstType ty) {
char
buf64
[
8
]
=
{
0
};
void
*
pBuf64
=
buf64
;
taosEncodeFixedU64
(
&
pBuf64
,
VERSION
);
fstCountingWriter
Write
(
b
->
wrt
,
buf64
,
sizeof
(
buf64
));
idxFile
Write
(
b
->
wrt
,
buf64
,
sizeof
(
buf64
));
pBuf64
=
buf64
;
memset
(
buf64
,
0
,
sizeof
(
buf64
));
taosEncodeFixedU64
(
&
pBuf64
,
ty
);
fstCountingWriter
Write
(
b
->
wrt
,
buf64
,
sizeof
(
buf64
));
idxFile
Write
(
b
->
wrt
,
buf64
,
sizeof
(
buf64
));
return
b
;
}
...
...
@@ -787,7 +787,7 @@ void fstBuilderDestroy(FstBuilder* b) {
return
;
}
fstCountingWriter
Destroy
(
b
->
wrt
);
idxFile
Destroy
(
b
->
wrt
);
fstUnFinishedNodesDestroy
(
b
->
unfinished
);
fstRegistryDestroy
(
b
->
registry
);
fstSliceDestroy
(
&
b
->
last
);
...
...
@@ -905,21 +905,19 @@ void* fstBuilderInsertInner(FstBuilder* b) {
void
*
pBuf64
=
buf64
;
taosEncodeFixedU64
(
&
pBuf64
,
b
->
len
);
fstCountingWriter
Write
(
b
->
wrt
,
buf64
,
sizeof
(
buf64
));
idxFile
Write
(
b
->
wrt
,
buf64
,
sizeof
(
buf64
));
pBuf64
=
buf64
;
taosEncodeFixedU64
(
&
pBuf64
,
rootAddr
);
fstCountingWriter
Write
(
b
->
wrt
,
buf64
,
sizeof
(
buf64
));
idxFile
Write
(
b
->
wrt
,
buf64
,
sizeof
(
buf64
));
char
buf32
[
4
]
=
{
0
};
void
*
pBuf32
=
buf32
;
uint32_t
sum
=
fstCountingWriter
MaskedCheckSum
(
b
->
wrt
);
uint32_t
sum
=
idxFile
MaskedCheckSum
(
b
->
wrt
);
taosEncodeFixedU32
(
&
pBuf32
,
sum
);
fstCountingWriter
Write
(
b
->
wrt
,
buf32
,
sizeof
(
buf32
));
idxFile
Write
(
b
->
wrt
,
buf32
,
sizeof
(
buf32
));
fstCountingWriterFlush
(
b
->
wrt
);
// fstCountingWriterDestroy(b->wrt);
// b->wrt = NULL;
idxFileFlush
(
b
->
wrt
);
return
b
->
wrt
;
}
void
fstBuilderFinish
(
FstBuilder
*
b
)
{
fstBuilderInsertInner
(
b
);
}
...
...
source/libs/index/src/indexFstDfa.c
浏览文件 @
267e450a
...
...
@@ -61,9 +61,10 @@ void dfaBuilderDestroy(FstDfaBuilder *builder) {
pIter
=
taosHashIterate
(
builder
->
cache
,
pIter
);
}
taosHashCleanup
(
builder
->
cache
);
taosMemoryFree
(
builder
);
}
FstDfa
*
dfaBuilder
Build
(
FstDfaBuilder
*
builder
)
{
FstDfa
*
dfaBuilder
(
FstDfaBuilder
*
builder
)
{
uint32_t
sz
=
taosArrayGetSize
(
builder
->
dfa
->
insts
);
FstSparseSet
*
cur
=
sparSetCreate
(
sz
);
FstSparseSet
*
nxt
=
sparSetCreate
(
sz
);
...
...
source/libs/index/src/indexFst
CountingWriter
.c
→
source/libs/index/src/indexFst
File
.c
浏览文件 @
267e450a
...
...
@@ -13,13 +13,13 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "indexFst
CountingWriter
.h"
#include "indexFst
File
.h"
#include "indexFstUtil.h"
#include "indexInt.h"
#include "os.h"
#include "tutil.h"
static
int
writeCtxDoWrite
(
Writer
Ctx
*
ctx
,
uint8_t
*
buf
,
int
len
)
{
static
int
idxFileCtxDoWrite
(
IFile
Ctx
*
ctx
,
uint8_t
*
buf
,
int
len
)
{
if
(
ctx
->
type
==
TFile
)
{
assert
(
len
==
taosWriteFile
(
ctx
->
file
.
pFile
,
buf
,
len
));
}
else
{
...
...
@@ -28,7 +28,7 @@ static int writeCtxDoWrite(WriterCtx* ctx, uint8_t* buf, int len) {
ctx
->
offset
+=
len
;
return
len
;
}
static
int
writeCtxDoRead
(
Writer
Ctx
*
ctx
,
uint8_t
*
buf
,
int
len
)
{
static
int
idxFileCtxDoRead
(
IFile
Ctx
*
ctx
,
uint8_t
*
buf
,
int
len
)
{
int
nRead
=
0
;
if
(
ctx
->
type
==
TFile
)
{
#ifdef USE_MMAP
...
...
@@ -44,7 +44,7 @@ static int writeCtxDoRead(WriterCtx* ctx, uint8_t* buf, int len) {
return
nRead
;
}
static
int
writeCtxDoReadFrom
(
Writer
Ctx
*
ctx
,
uint8_t
*
buf
,
int
len
,
int32_t
offset
)
{
static
int
idxFileCtxDoReadFrom
(
IFile
Ctx
*
ctx
,
uint8_t
*
buf
,
int
len
,
int32_t
offset
)
{
int
nRead
=
0
;
if
(
ctx
->
type
==
TFile
)
{
// tfLseek(ctx->file.pFile, offset, 0);
...
...
@@ -61,7 +61,7 @@ static int writeCtxDoReadFrom(WriterCtx* ctx, uint8_t* buf, int len, int32_t off
}
return
nRead
;
}
static
int
writeCtxGetSize
(
Writer
Ctx
*
ctx
)
{
static
int
idxFileCtxGetSize
(
IFile
Ctx
*
ctx
)
{
if
(
ctx
->
type
==
TFile
)
{
int64_t
file_size
=
0
;
taosStatFile
(
ctx
->
file
.
buf
,
&
file_size
,
NULL
);
...
...
@@ -69,7 +69,7 @@ static int writeCtxGetSize(WriterCtx* ctx) {
}
return
0
;
}
static
int
writeCtxDoFlush
(
Writer
Ctx
*
ctx
)
{
static
int
idxFileCtxDoFlush
(
IFile
Ctx
*
ctx
)
{
if
(
ctx
->
type
==
TFile
)
{
// taosFsyncFile(ctx->file.pFile);
taosFsyncFile
(
ctx
->
file
.
pFile
);
...
...
@@ -80,8 +80,8 @@ static int writeCtxDoFlush(WriterCtx* ctx) {
return
1
;
}
WriterCtx
*
writer
CtxCreate
(
WriterType
type
,
const
char
*
path
,
bool
readOnly
,
int32_t
capacity
)
{
WriterCtx
*
ctx
=
taosMemoryCalloc
(
1
,
sizeof
(
Writer
Ctx
));
IFileCtx
*
idxFile
CtxCreate
(
WriterType
type
,
const
char
*
path
,
bool
readOnly
,
int32_t
capacity
)
{
IFileCtx
*
ctx
=
taosMemoryCalloc
(
1
,
sizeof
(
IFile
Ctx
));
if
(
ctx
==
NULL
)
{
return
NULL
;
}
...
...
@@ -90,39 +90,36 @@ WriterCtx* writerCtxCreate(WriterType type, const char* path, bool readOnly, int
if
(
ctx
->
type
==
TFile
)
{
// ugly code, refactor later
ctx
->
file
.
readOnly
=
readOnly
;
memcpy
(
ctx
->
file
.
buf
,
path
,
strlen
(
path
));
if
(
readOnly
==
false
)
{
// ctx->file.pFile = open(path, O_WRONLY | O_CREAT | O_APPEND, S_IRWXU | S_IRWXG | S_IRWXO);
ctx
->
file
.
pFile
=
taosOpenFile
(
path
,
TD_FILE_CREATE
|
TD_FILE_WRITE
|
TD_FILE_APPEND
);
taosFtruncateFile
(
ctx
->
file
.
pFile
,
0
);
int64_t
file_size
;
taosStatFile
(
path
,
&
file_size
,
NULL
);
ctx
->
file
.
size
=
(
int
)
file_size
;
taosStatFile
(
path
,
&
ctx
->
file
.
size
,
NULL
);
// ctx->file.size = (int)size;
}
else
{
// ctx->file.pFile = open(path, O_RDONLY, S_IRWXU | S_IRWXG | S_IRWXO);
ctx
->
file
.
pFile
=
taosOpenFile
(
path
,
TD_FILE_READ
);
int64_t
file_
size
=
0
;
taosFStatFile
(
ctx
->
file
.
pFile
,
&
file_
size
,
NULL
);
ctx
->
file
.
size
=
(
int
)
file_
size
;
int64_t
size
=
0
;
taosFStatFile
(
ctx
->
file
.
pFile
,
&
ctx
->
file
.
size
,
NULL
);
ctx
->
file
.
size
=
(
int
)
size
;
#ifdef USE_MMAP
ctx
->
file
.
ptr
=
(
char
*
)
tfMmapReadOnly
(
ctx
->
file
.
pFile
,
ctx
->
file
.
size
);
#endif
}
memcpy
(
ctx
->
file
.
buf
,
path
,
strlen
(
path
));
if
(
ctx
->
file
.
pFile
==
NULL
)
{
indexError
(
"failed to open file, error %d"
,
errno
);
goto
END
;
}
}
else
if
(
ctx
->
type
==
TMemory
)
{
ctx
->
mem
.
buf
=
taosMemoryCalloc
(
1
,
sizeof
(
char
)
*
capacity
);
ctx
->
mem
.
cap
a
=
capacity
;
ctx
->
mem
.
cap
=
capacity
;
}
ctx
->
write
=
writ
eCtxDoWrite
;
ctx
->
read
=
writ
eCtxDoRead
;
ctx
->
flush
=
writ
eCtxDoFlush
;
ctx
->
readFrom
=
writ
eCtxDoReadFrom
;
ctx
->
size
=
writ
eCtxGetSize
;
ctx
->
write
=
idxFil
eCtxDoWrite
;
ctx
->
read
=
idxFil
eCtxDoRead
;
ctx
->
flush
=
idxFil
eCtxDoFlush
;
ctx
->
readFrom
=
idxFil
eCtxDoReadFrom
;
ctx
->
size
=
idxFil
eCtxGetSize
;
ctx
->
offset
=
0
;
ctx
->
limit
=
capacity
;
...
...
@@ -135,7 +132,7 @@ END:
taosMemoryFree
(
ctx
);
return
NULL
;
}
void
writerCtxDestroy
(
Writer
Ctx
*
ctx
,
bool
remove
)
{
void
idxFileCtxDestroy
(
IFile
Ctx
*
ctx
,
bool
remove
)
{
if
(
ctx
->
type
==
TMemory
)
{
taosMemoryFree
(
ctx
->
mem
.
buf
);
}
else
{
...
...
@@ -149,9 +146,6 @@ void writerCtxDestroy(WriterCtx* ctx, bool remove) {
if
(
ctx
->
file
.
readOnly
==
false
)
{
int64_t
file_size
=
0
;
taosStatFile
(
ctx
->
file
.
buf
,
&
file_size
,
NULL
);
// struct stat fstat;
// stat(ctx->file.buf, &fstat);
// indexError("write file size: %d", (int)(fstat.st_size));
}
if
(
remove
)
{
unlink
(
ctx
->
file
.
buf
);
...
...
@@ -160,30 +154,29 @@ void writerCtxDestroy(WriterCtx* ctx, bool remove) {
taosMemoryFree
(
ctx
);
}
FstCountingWriter
*
fstCountingWriter
Create
(
void
*
wrt
)
{
FstCountingWriter
*
cw
=
taosMemoryCalloc
(
1
,
sizeof
(
FstCountingWriter
));
IdxFstFile
*
idxFile
Create
(
void
*
wrt
)
{
IdxFstFile
*
cw
=
taosMemoryCalloc
(
1
,
sizeof
(
IdxFstFile
));
if
(
cw
==
NULL
)
{
return
NULL
;
}
cw
->
wrt
=
wrt
;
//(void *)(writerCtxCreate(TFile, readOnly));
return
cw
;
}
void
fstCountingWriterDestroy
(
FstCountingWriter
*
cw
)
{
void
idxFileDestroy
(
IdxFstFile
*
cw
)
{
// free wrt object: close fd or free mem
fstCountingWriter
Flush
(
cw
);
//
writerCtxDestroy((Writer
Ctx *)(cw->wrt));
idxFile
Flush
(
cw
);
//
idxFileCtxDestroy((IFile
Ctx *)(cw->wrt));
taosMemoryFree
(
cw
);
}
int
fstCountingWriterWrite
(
FstCountingWriter
*
write
,
uint8_t
*
buf
,
uint32_t
len
)
{
int
idxFileWrite
(
IdxFstFile
*
write
,
uint8_t
*
buf
,
uint32_t
len
)
{
if
(
write
==
NULL
)
{
return
0
;
}
// update checksum
// write data to file/socket or mem
Writer
Ctx
*
ctx
=
write
->
wrt
;
IFile
Ctx
*
ctx
=
write
->
wrt
;
int
nWrite
=
ctx
->
write
(
ctx
,
buf
,
len
);
assert
(
nWrite
==
len
);
...
...
@@ -192,42 +185,41 @@ int fstCountingWriterWrite(FstCountingWriter* write, uint8_t* buf, uint32_t len)
write
->
summer
=
taosCalcChecksum
(
write
->
summer
,
buf
,
len
);
return
len
;
}
int
fstCountingWriterRead
(
FstCountingWriter
*
write
,
uint8_t
*
buf
,
uint32_t
len
)
{
int
idxFileRead
(
IdxFstFile
*
write
,
uint8_t
*
buf
,
uint32_t
len
)
{
if
(
write
==
NULL
)
{
return
0
;
}
Writer
Ctx
*
ctx
=
write
->
wrt
;
int
nRead
=
ctx
->
read
(
ctx
,
buf
,
len
);
IFile
Ctx
*
ctx
=
write
->
wrt
;
int
nRead
=
ctx
->
read
(
ctx
,
buf
,
len
);
// assert(nRead == len);
return
nRead
;
}
uint32_t
fstCountingWriterMaskedCheckSum
(
FstCountingWriter
*
write
)
{
uint32_t
idxFileMaskedCheckSum
(
IdxFstFile
*
write
)
{
// opt
return
write
->
summer
;
}
int
fstCountingWriterFlush
(
FstCountingWriter
*
write
)
{
Writer
Ctx
*
ctx
=
write
->
wrt
;
int
idxFileFlush
(
IdxFstFile
*
write
)
{
IFile
Ctx
*
ctx
=
write
->
wrt
;
ctx
->
flush
(
ctx
);
// write->wtr->flush
return
1
;
}
void
fstCountingWriterPackUintIn
(
FstCountingWriter
*
writer
,
uint64_t
n
,
uint8_t
nBytes
)
{
void
idxFilePackUintIn
(
IdxFstFile
*
writer
,
uint64_t
n
,
uint8_t
nBytes
)
{
assert
(
1
<=
nBytes
&&
nBytes
<=
8
);
uint8_t
*
buf
=
taosMemoryCalloc
(
8
,
sizeof
(
uint8_t
));
for
(
uint8_t
i
=
0
;
i
<
nBytes
;
i
++
)
{
buf
[
i
]
=
(
uint8_t
)
n
;
n
=
n
>>
8
;
}
fstCountingWriter
Write
(
writer
,
buf
,
nBytes
);
idxFile
Write
(
writer
,
buf
,
nBytes
);
taosMemoryFree
(
buf
);
return
;
}
uint8_t
fstCountingWriterPackUint
(
FstCountingWriter
*
writer
,
uint64_t
n
)
{
uint8_t
idxFilePackUint
(
IdxFstFile
*
writer
,
uint64_t
n
)
{
uint8_t
nBytes
=
packSize
(
n
);
fstCountingWriter
PackUintIn
(
writer
,
n
,
nBytes
);
idxFile
PackUintIn
(
writer
,
n
,
nBytes
);
return
nBytes
;
}
source/libs/index/src/indexFstNode.c
浏览文件 @
267e450a
...
...
@@ -95,7 +95,7 @@ void fstBuilderNodeCloneFrom(FstBuilderNode* dst, FstBuilderNode* src) {
}
}
// bool fstBuilderNodeCompileTo(FstBuilderNode *b,
FstCountingWriter
*wrt, CompiledAddr lastAddr, CompiledAddr
// bool fstBuilderNodeCompileTo(FstBuilderNode *b,
IdxFile
*wrt, CompiledAddr lastAddr, CompiledAddr
// startAddr) {
// size_t sz = taosArrayGetSize(b->trans);
...
...
source/libs/index/src/indexFstUtil.c
浏览文件 @
267e450a
...
...
@@ -75,7 +75,6 @@ CompiledAddr unpackDelta(char* data, uint64_t len, uint64_t nodeAddr) {
}
// fst slice func
//
FstSlice
fstSliceCreate
(
uint8_t
*
data
,
uint64_t
len
)
{
FstString
*
str
=
(
FstString
*
)
taosMemoryMalloc
(
sizeof
(
FstString
));
...
...
@@ -164,16 +163,3 @@ int fstSliceCompare(FstSlice* a, FstSlice* b) {
return
0
;
}
}
// FstStack* fstStackCreate(size_t elemSize, StackFreeElem freeFn) {
// FstStack *s = taosMemoryCalloc(1, sizeof(FstStack));
// if (s == NULL) { return NULL; }
// s->
// s->freeFn
//
//}
// void *fstStackPush(FstStack *s, void *elem);
// void *fstStackTop(FstStack *s);
// size_t fstStackLen(FstStack *s);
// void *fstStackGetAt(FstStack *s, size_t i);
// void fstStackDestory(FstStack *);
source/libs/index/src/indexTfile.c
浏览文件 @
267e450a
...
...
@@ -16,7 +16,7 @@
#include "index.h"
#include "indexComm.h"
#include "indexFst.h"
#include "indexFst
CountingWriter
.h"
#include "indexFst
File
.h"
#include "indexUtil.h"
#include "taosdef.h"
#include "taoserror.h"
...
...
@@ -103,7 +103,7 @@ TFileCache* tfileCacheCreate(const char* path) {
for
(
size_t
i
=
0
;
i
<
taosArrayGetSize
(
files
);
i
++
)
{
char
*
file
=
taosArrayGetP
(
files
,
i
);
WriterCtx
*
wc
=
writer
CtxCreate
(
TFile
,
file
,
true
,
1024
*
1024
*
64
);
IFileCtx
*
wc
=
idxFile
CtxCreate
(
TFile
,
file
,
true
,
1024
*
1024
*
64
);
if
(
wc
==
NULL
)
{
indexError
(
"failed to open index:%s"
,
file
);
goto
End
;
...
...
@@ -175,7 +175,7 @@ void tfileCachePut(TFileCache* tcache, ICacheKey* key, TFileReader* reader) {
tfileReaderRef
(
reader
);
return
;
}
TFileReader
*
tfileReaderCreate
(
Writer
Ctx
*
ctx
)
{
TFileReader
*
tfileReaderCreate
(
IFile
Ctx
*
ctx
)
{
TFileReader
*
reader
=
taosMemoryCalloc
(
1
,
sizeof
(
TFileReader
));
if
(
reader
==
NULL
)
{
return
NULL
;
...
...
@@ -216,7 +216,7 @@ void tfileReaderDestroy(TFileReader* reader) {
}
else
{
indexInfo
(
"%s is not removed"
,
reader
->
ctx
->
file
.
buf
);
}
writer
CtxDestroy
(
reader
->
ctx
,
reader
->
remove
);
idxFile
CtxDestroy
(
reader
->
ctx
,
reader
->
remove
);
taosMemoryFree
(
reader
);
}
...
...
@@ -490,7 +490,7 @@ TFileWriter* tfileWriterOpen(char* path, uint64_t suid, int64_t version, const c
char
fullname
[
256
]
=
{
0
};
tfileGenFileFullName
(
fullname
,
path
,
suid
,
colName
,
version
);
// indexInfo("open write file name %s", fullname);
WriterCtx
*
wcx
=
writer
CtxCreate
(
TFile
,
fullname
,
false
,
1024
*
1024
*
64
);
IFileCtx
*
wcx
=
idxFile
CtxCreate
(
TFile
,
fullname
,
false
,
1024
*
1024
*
64
);
if
(
wcx
==
NULL
)
{
return
NULL
;
}
...
...
@@ -507,18 +507,18 @@ TFileReader* tfileReaderOpen(char* path, uint64_t suid, int64_t version, const c
char
fullname
[
256
]
=
{
0
};
tfileGenFileFullName
(
fullname
,
path
,
suid
,
colName
,
version
);
WriterCtx
*
wc
=
writer
CtxCreate
(
TFile
,
fullname
,
true
,
1024
*
1024
*
1024
);
IFileCtx
*
wc
=
idxFile
CtxCreate
(
TFile
,
fullname
,
true
,
1024
*
1024
*
1024
);
if
(
wc
==
NULL
)
{
terrno
=
TAOS_SYSTEM_ERROR
(
errno
);
indexError
(
"failed to open readonly file: %s, reason: %s"
,
fullname
,
terrstr
());
return
NULL
;
}
indexTrace
(
"open read file name:%s, file size: %
d
"
,
wc
->
file
.
buf
,
wc
->
file
.
size
);
indexTrace
(
"open read file name:%s, file size: %
"
PRId64
"
"
,
wc
->
file
.
buf
,
wc
->
file
.
size
);
TFileReader
*
reader
=
tfileReaderCreate
(
wc
);
return
reader
;
}
TFileWriter
*
tfileWriterCreate
(
Writer
Ctx
*
ctx
,
TFileHeader
*
header
)
{
TFileWriter
*
tfileWriterCreate
(
IFile
Ctx
*
ctx
,
TFileHeader
*
header
)
{
TFileWriter
*
tw
=
taosMemoryCalloc
(
1
,
sizeof
(
TFileWriter
));
if
(
tw
==
NULL
)
{
indexError
(
"index: %"
PRIu64
" failed to alloc TFilerWriter"
,
header
->
suid
);
...
...
@@ -609,14 +609,14 @@ void tfileWriterClose(TFileWriter* tw) {
if
(
tw
==
NULL
)
{
return
;
}
writer
CtxDestroy
(
tw
->
ctx
,
false
);
idxFile
CtxDestroy
(
tw
->
ctx
,
false
);
taosMemoryFree
(
tw
);
}
void
tfileWriterDestroy
(
TFileWriter
*
tw
)
{
if
(
tw
==
NULL
)
{
return
;
}
writer
CtxDestroy
(
tw
->
ctx
,
false
);
idxFile
CtxDestroy
(
tw
->
ctx
,
false
);
taosMemoryFree
(
tw
);
}
...
...
@@ -892,8 +892,8 @@ static int tfileReaderLoadHeader(TFileReader* reader) {
return
0
;
}
static
int
tfileReaderLoadFst
(
TFileReader
*
reader
)
{
Writer
Ctx
*
ctx
=
reader
->
ctx
;
int
size
=
ctx
->
size
(
ctx
);
IFile
Ctx
*
ctx
=
reader
->
ctx
;
int
size
=
ctx
->
size
(
ctx
);
// current load fst into memory, refactor it later
int
fstSize
=
size
-
reader
->
header
.
fstOffset
-
sizeof
(
tfileMagicNumber
);
...
...
@@ -905,8 +905,9 @@ static int tfileReaderLoadFst(TFileReader* reader) {
int64_t
ts
=
taosGetTimestampUs
();
int32_t
nread
=
ctx
->
readFrom
(
ctx
,
buf
,
fstSize
,
reader
->
header
.
fstOffset
);
int64_t
cost
=
taosGetTimestampUs
()
-
ts
;
indexInfo
(
"nread = %d, and fst offset=%d, fst size: %d, filename: %s, file size: %d, time cost: %"
PRId64
"us"
,
nread
,
reader
->
header
.
fstOffset
,
fstSize
,
ctx
->
file
.
buf
,
ctx
->
file
.
size
,
cost
);
indexInfo
(
"nread = %d, and fst offset=%d, fst size: %d, filename: %s, file size: %"
PRId64
", time cost: %"
PRId64
"us"
,
nread
,
reader
->
header
.
fstOffset
,
fstSize
,
ctx
->
file
.
buf
,
ctx
->
file
.
size
,
cost
);
// we assuse fst size less than FST_MAX_SIZE
assert
(
nread
>
0
&&
nread
<=
fstSize
);
...
...
@@ -919,7 +920,7 @@ static int tfileReaderLoadFst(TFileReader* reader) {
}
static
int
tfileReaderLoadTableIds
(
TFileReader
*
reader
,
int32_t
offset
,
SArray
*
result
)
{
// TODO(yihao): opt later
Writer
Ctx
*
ctx
=
reader
->
ctx
;
IFile
Ctx
*
ctx
=
reader
->
ctx
;
// add block cache
char
block
[
4096
]
=
{
0
};
int32_t
nread
=
ctx
->
readFrom
(
ctx
,
block
,
sizeof
(
block
),
offset
);
...
...
@@ -952,7 +953,7 @@ static int tfileReaderLoadTableIds(TFileReader* reader, int32_t offset, SArray*
}
static
int
tfileReaderVerify
(
TFileReader
*
reader
)
{
// just validate header and Footer, file corrupted also shuild be verified later
Writer
Ctx
*
ctx
=
reader
->
ctx
;
IFile
Ctx
*
ctx
=
reader
->
ctx
;
uint64_t
tMagicNumber
=
0
;
...
...
source/libs/index/test/fstTest.cc
浏览文件 @
267e450a
...
...
@@ -7,7 +7,6 @@
#include "index.h"
#include "indexCache.h"
#include "indexFst.h"
#include "indexFstCountingWriter.h"
#include "indexFstUtil.h"
#include "indexInt.h"
#include "indexTfile.h"
...
...
@@ -20,7 +19,7 @@ class FstWriter {
public:
FstWriter
()
{
taosRemoveFile
(
fileName
.
c_str
());
_wc
=
writer
CtxCreate
(
TFile
,
fileName
.
c_str
(),
false
,
64
*
1024
*
1024
);
_wc
=
idxFile
CtxCreate
(
TFile
,
fileName
.
c_str
(),
false
,
64
*
1024
*
1024
);
_b
=
fstBuilderCreate
(
_wc
,
0
);
}
bool
Put
(
const
std
::
string
&
key
,
uint64_t
val
)
{
...
...
@@ -38,25 +37,25 @@ class FstWriter {
fstBuilderFinish
(
_b
);
fstBuilderDestroy
(
_b
);
writer
CtxDestroy
(
_wc
,
false
);
idxFile
CtxDestroy
(
_wc
,
false
);
}
private:
FstBuilder
*
_b
;
WriterCtx
*
_wc
;
IFileCtx
*
_wc
;
};
class
FstReadMemory
{
public:
FstReadMemory
(
int32_t
size
,
const
std
::
string
&
fileName
=
TD_TMP_DIR_PATH
"tindex.tindex"
)
{
_wc
=
writer
CtxCreate
(
TFile
,
fileName
.
c_str
(),
true
,
64
*
1024
);
_w
=
fstCountingWriter
Create
(
_wc
);
_wc
=
idxFile
CtxCreate
(
TFile
,
fileName
.
c_str
(),
true
,
64
*
1024
);
_w
=
idxFile
Create
(
_wc
);
_size
=
size
;
memset
((
void
*
)
&
_s
,
0
,
sizeof
(
_s
));
}
bool
init
()
{
char
*
buf
=
(
char
*
)
taosMemoryCalloc
(
1
,
sizeof
(
char
)
*
_size
);
int
nRead
=
fstCountingWriter
Read
(
_w
,
(
uint8_t
*
)
buf
,
_size
);
int
nRead
=
idxFile
Read
(
_w
,
(
uint8_t
*
)
buf
,
_size
);
if
(
nRead
<=
0
)
{
return
false
;
}
...
...
@@ -141,18 +140,18 @@ class FstReadMemory {
}
~
FstReadMemory
()
{
fstCountingWriter
Destroy
(
_w
);
idxFile
Destroy
(
_w
);
fstDestroy
(
_fst
);
fstSliceDestroy
(
&
_s
);
writer
CtxDestroy
(
_wc
,
false
);
idxFile
CtxDestroy
(
_wc
,
false
);
}
private:
FstCountingWriter
*
_w
;
Fst
*
_fst
;
FstSlice
_s
;
WriterCtx
*
_wc
;
int32_t
_size
;
IdxFstFile
*
_w
;
Fst
*
_fst
;
FstSlice
_s
;
IFileCtx
*
_wc
;
int32_t
_size
;
};
#define L 100
...
...
source/libs/index/test/fstUT.cc
浏览文件 @
267e450a
...
...
@@ -8,7 +8,6 @@
#include "index.h"
#include "indexCache.h"
#include "indexFst.h"
#include "indexFstCountingWriter.h"
#include "indexFstUtil.h"
#include "indexInt.h"
#include "indexTfile.h"
...
...
@@ -40,7 +39,7 @@ static void EnvCleanup() {}
class
FstWriter
{
public:
FstWriter
()
{
_wc
=
writer
CtxCreate
(
TFile
,
tindex
,
false
,
64
*
1024
*
1024
);
_wc
=
idxFile
CtxCreate
(
TFile
,
tindex
,
false
,
64
*
1024
*
1024
);
_b
=
fstBuilderCreate
(
_wc
,
0
);
}
bool
Put
(
const
std
::
string
&
key
,
uint64_t
val
)
{
...
...
@@ -58,25 +57,25 @@ class FstWriter {
fstBuilderFinish
(
_b
);
fstBuilderDestroy
(
_b
);
writer
CtxDestroy
(
_wc
,
false
);
idxFile
CtxDestroy
(
_wc
,
false
);
}
private:
FstBuilder
*
_b
;
WriterCtx
*
_wc
;
IFileCtx
*
_wc
;
};
class
FstReadMemory
{
public:
FstReadMemory
(
size_t
size
)
{
_wc
=
writer
CtxCreate
(
TFile
,
tindex
,
true
,
64
*
1024
);
_w
=
fstCountingWriter
Create
(
_wc
);
_wc
=
idxFile
CtxCreate
(
TFile
,
tindex
,
true
,
64
*
1024
);
_w
=
idxFile
Create
(
_wc
);
_size
=
size
;
memset
((
void
*
)
&
_s
,
0
,
sizeof
(
_s
));
}
bool
init
()
{
char
*
buf
=
(
char
*
)
taosMemoryCalloc
(
1
,
sizeof
(
char
)
*
_size
);
int
nRead
=
fstCountingWriter
Read
(
_w
,
(
uint8_t
*
)
buf
,
_size
);
int
nRead
=
idxFile
Read
(
_w
,
(
uint8_t
*
)
buf
,
_size
);
if
(
nRead
<=
0
)
{
return
false
;
}
...
...
@@ -130,18 +129,18 @@ class FstReadMemory {
}
~
FstReadMemory
()
{
fstCountingWriter
Destroy
(
_w
);
idxFile
Destroy
(
_w
);
fstDestroy
(
_fst
);
fstSliceDestroy
(
&
_s
);
writer
CtxDestroy
(
_wc
,
false
);
idxFile
CtxDestroy
(
_wc
,
false
);
}
private:
FstCountingWriter
*
_w
;
Fst
*
_fst
;
FstSlice
_s
;
WriterCtx
*
_wc
;
size_t
_size
;
IdxFstFile
*
_w
;
Fst
*
_fst
;
FstSlice
_s
;
IFileCtx
*
_wc
;
size_t
_size
;
};
class
FstWriterEnv
:
public
::
testing
::
Test
{
...
...
source/libs/index/test/indexTests.cc
浏览文件 @
267e450a
...
...
@@ -20,7 +20,6 @@
#include "index.h"
#include "indexCache.h"
#include "indexFst.h"
#include "indexFstCountingWriter.h"
#include "indexFstUtil.h"
#include "indexInt.h"
#include "indexTfile.h"
...
...
@@ -51,7 +50,7 @@ class DebugInfo {
class
FstWriter
{
public:
FstWriter
()
{
_wc
=
writer
CtxCreate
(
TFile
,
TD_TMP_DIR_PATH
"tindex"
,
false
,
64
*
1024
*
1024
);
_wc
=
idxFile
CtxCreate
(
TFile
,
TD_TMP_DIR_PATH
"tindex"
,
false
,
64
*
1024
*
1024
);
_b
=
fstBuilderCreate
(
NULL
,
0
);
}
bool
Put
(
const
std
::
string
&
key
,
uint64_t
val
)
{
...
...
@@ -64,25 +63,25 @@ class FstWriter {
fstBuilderFinish
(
_b
);
fstBuilderDestroy
(
_b
);
writer
CtxDestroy
(
_wc
,
false
);
idxFile
CtxDestroy
(
_wc
,
false
);
}
private:
FstBuilder
*
_b
;
WriterCtx
*
_wc
;
IFileCtx
*
_wc
;
};
class
FstReadMemory
{
public:
FstReadMemory
(
size_t
size
)
{
_wc
=
writer
CtxCreate
(
TFile
,
TD_TMP_DIR_PATH
"tindex"
,
true
,
64
*
1024
);
_w
=
fstCountingWriter
Create
(
_wc
);
_wc
=
idxFile
CtxCreate
(
TFile
,
TD_TMP_DIR_PATH
"tindex"
,
true
,
64
*
1024
);
_w
=
idxFile
Create
(
_wc
);
_size
=
size
;
memset
((
void
*
)
&
_s
,
0
,
sizeof
(
_s
));
}
bool
init
()
{
char
*
buf
=
(
char
*
)
taosMemoryCalloc
(
1
,
sizeof
(
char
)
*
_size
);
int
nRead
=
fstCountingWriter
Read
(
_w
,
(
uint8_t
*
)
buf
,
_size
);
int
nRead
=
idxFile
Read
(
_w
,
(
uint8_t
*
)
buf
,
_size
);
if
(
nRead
<=
0
)
{
return
false
;
}
...
...
@@ -124,18 +123,18 @@ class FstReadMemory {
}
~
FstReadMemory
()
{
fstCountingWriter
Destroy
(
_w
);
idxFile
Destroy
(
_w
);
fstDestroy
(
_fst
);
fstSliceDestroy
(
&
_s
);
writer
CtxDestroy
(
_wc
,
true
);
idxFile
CtxDestroy
(
_wc
,
true
);
}
private:
FstCountingWriter
*
_w
;
Fst
*
_fst
;
FstSlice
_s
;
WriterCtx
*
_wc
;
size_t
_size
;
IdxFstFile
*
_w
;
Fst
*
_fst
;
FstSlice
_s
;
IFileCtx
*
_wc
;
size_t
_size
;
};
#define L 100
...
...
@@ -392,13 +391,13 @@ class TFileObj {
fileName_
=
path
;
WriterCtx
*
ctx
=
writer
CtxCreate
(
TFile
,
path
.
c_str
(),
false
,
64
*
1024
*
1024
);
IFileCtx
*
ctx
=
idxFile
CtxCreate
(
TFile
,
path
.
c_str
(),
false
,
64
*
1024
*
1024
);
writer_
=
tfileWriterCreate
(
ctx
,
&
header
);
return
writer_
!=
NULL
?
true
:
false
;
}
bool
InitReader
()
{
WriterCtx
*
ctx
=
writer
CtxCreate
(
TFile
,
fileName_
.
c_str
(),
true
,
64
*
1024
*
1024
);
IFileCtx
*
ctx
=
idxFile
CtxCreate
(
TFile
,
fileName_
.
c_str
(),
true
,
64
*
1024
*
1024
);
reader_
=
tfileReaderCreate
(
ctx
);
return
reader_
!=
NULL
?
true
:
false
;
}
...
...
source/libs/index/test/jsonUT.cc
浏览文件 @
267e450a
...
...
@@ -7,7 +7,6 @@
#include "index.h"
#include "indexCache.h"
#include "indexFst.h"
#include "indexFstCountingWriter.h"
#include "indexFstUtil.h"
#include "indexInt.h"
#include "indexTfile.h"
...
...
source/libs/index/test/utilUT.cc
浏览文件 @
267e450a
...
...
@@ -8,7 +8,6 @@
#include "indexCache.h"
#include "indexComm.h"
#include "indexFst.h"
#include "indexFstCountingWriter.h"
#include "indexFstUtil.h"
#include "indexInt.h"
#include "indexTfile.h"
...
...
source/libs/transport/src/transComm.c
浏览文件 @
267e450a
...
...
@@ -479,6 +479,10 @@ bool transEpSetIsEqual(SEpSet* a, SEpSet* b) {
}
return
true
;
}
static
int32_t
transGetRefMgt
()
{
//
return
refMgt
;
}
static
void
transInitEnv
()
{
refMgt
=
transOpenExHandleMgt
(
50000
);
...
...
@@ -486,8 +490,9 @@ static void transInitEnv() {
}
static
void
transDestroyEnv
()
{
// close ref
transCloseExHandleMgt
(
refMgt
);
transCloseExHandleMgt
();
}
void
transInit
()
{
// init env
taosThreadOnce
(
&
transModuleInit
,
transInitEnv
);
...
...
@@ -502,25 +507,25 @@ int32_t transOpenExHandleMgt(int size) {
}
void
transCloseExHandleMgt
()
{
// close ref
taosCloseRef
(
refMgt
);
taosCloseRef
(
transGetRefMgt
()
);
}
int64_t
transAddExHandle
(
void
*
p
)
{
// acquire extern handle
return
taosAddRef
(
refMgt
,
p
);
return
taosAddRef
(
transGetRefMgt
()
,
p
);
}
int32_t
transRemoveExHandle
(
int64_t
refId
)
{
// acquire extern handle
return
taosRemoveRef
(
refMgt
,
refId
);
return
taosRemoveRef
(
transGetRefMgt
()
,
refId
);
}
SExHandle
*
transAcquireExHandle
(
int64_t
refId
)
{
// acquire extern handle
return
(
SExHandle
*
)
taosAcquireRef
(
refMgt
,
refId
);
return
(
SExHandle
*
)
taosAcquireRef
(
transGetRefMgt
()
,
refId
);
}
int32_t
transReleaseExHandle
(
int64_t
refId
)
{
// release extern handle
return
taosReleaseRef
(
refMgt
,
refId
);
return
taosReleaseRef
(
transGetRefMgt
()
,
refId
);
}
void
transDestoryExHandle
(
void
*
handle
)
{
if
(
handle
==
NULL
)
{
...
...
source/util/src/terror.c
浏览文件 @
267e450a
...
...
@@ -599,6 +599,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_TQ_NO_COMMITTED_OFFSET, "No committed offset
TAOS_DEFINE_ERROR
(
TSDB_CODE_INDEX_REBUILDING
,
"Index is rebuilding"
)
TAOS_DEFINE_ERROR
(
TSDB_CODE_INDEX_REBUILDING
,
"Invalid index file"
)
TAOS_DEFINE_ERROR
(
TSDB_CODE_TMQ_INVALID_MSG
,
"Invalid message"
)
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录