Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
bcd3929e
T
TDengine
项目概览
taosdata
/
TDengine
大约 1 年 前同步成功
通知
1184
Star
22015
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看板
体验新版 GitCode,发现更多精彩内容 >>
未验证
提交
bcd3929e
编写于
5月 06, 2022
作者:
H
Hongze Cheng
提交者:
GitHub
5月 06, 2022
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #12145 from taosdata/feature/vnode_refact1
fix: drop table if exists
上级
b99f18e6
26362113
变更
12
显示空白变更内容
内联
并排
Showing
12 changed file
with
277 addition
and
64 deletion
+277
-64
include/common/tmsg.h
include/common/tmsg.h
+22
-0
source/common/src/tmsg.c
source/common/src/tmsg.c
+62
-0
source/dnode/vnode/CMakeLists.txt
source/dnode/vnode/CMakeLists.txt
+1
-0
source/dnode/vnode/src/inc/tsdb.h
source/dnode/vnode/src/inc/tsdb.h
+0
-1
source/dnode/vnode/src/inc/vnodeInt.h
source/dnode/vnode/src/inc/vnodeInt.h
+1
-1
source/dnode/vnode/src/tsdb/tsdbCommit.c
source/dnode/vnode/src/tsdb/tsdbCommit.c
+0
-4
source/dnode/vnode/src/tsdb/tsdbMemTable.c
source/dnode/vnode/src/tsdb/tsdbMemTable.c
+4
-6
source/dnode/vnode/src/tsdb/tsdbMemTable2.c
source/dnode/vnode/src/tsdb/tsdbMemTable2.c
+93
-0
source/dnode/vnode/src/tsdb/tsdbSma.c
source/dnode/vnode/src/tsdb/tsdbSma.c
+15
-0
source/dnode/vnode/src/vnd/vnodeSvr.c
source/dnode/vnode/src/vnd/vnodeSvr.c
+49
-25
source/libs/parser/src/parInsert.c
source/libs/parser/src/parInsert.c
+30
-22
source/libs/parser/src/parInsertData.c
source/libs/parser/src/parInsertData.c
+0
-5
未找到文件。
include/common/tmsg.h
浏览文件 @
bcd3929e
...
...
@@ -2578,6 +2578,28 @@ static FORCE_INLINE void tDeleteSMqAskEpRsp(SMqAskEpRsp* pRsp) {
taosArrayDestroyEx
(
pRsp
->
topics
,
(
void
(
*
)(
void
*
))
tDeleteSMqSubTopicEp
);
}
#define TD_AUTO_CREATE_TABLE 0x1
typedef
struct
{
int64_t
suid
;
int64_t
uid
;
int32_t
sver
;
uint64_t
nData
;
const
void
*
pData
;
SVCreateTbReq
cTbReq
;
}
SVSubmitBlk
;
typedef
struct
{
int32_t
flags
;
int32_t
nBlocks
;
union
{
SArray
*
pArray
;
SVSubmitBlk
*
pBlocks
;
};
}
SVSubmitReq
;
int32_t
tEncodeSVSubmitReq
(
SCoder
*
pCoder
,
const
SVSubmitReq
*
pReq
);
int32_t
tDecodeSVSubmitReq
(
SCoder
*
pCoder
,
SVSubmitReq
*
pReq
);
#pragma pack(pop)
#ifdef __cplusplus
...
...
source/common/src/tmsg.c
浏览文件 @
bcd3929e
...
...
@@ -3860,3 +3860,65 @@ int32_t tDecodeSVDropStbReq(SCoder *pCoder, SVDropStbReq *pReq) {
tEndDecode
(
pCoder
);
return
0
;
}
static
int32_t
tEncodeSVSubmitBlk
(
SCoder
*
pCoder
,
const
SVSubmitBlk
*
pBlock
,
int32_t
flags
)
{
if
(
tStartEncode
(
pCoder
)
<
0
)
return
-
1
;
if
(
tEncodeI64
(
pCoder
,
pBlock
->
suid
)
<
0
)
return
-
1
;
if
(
tEncodeI64
(
pCoder
,
pBlock
->
uid
)
<
0
)
return
-
1
;
if
(
tEncodeI32v
(
pCoder
,
pBlock
->
sver
)
<
0
)
return
-
1
;
if
(
tEncodeBinary
(
pCoder
,
pBlock
->
pData
,
pBlock
->
nData
)
<
0
)
return
-
1
;
if
(
flags
&
TD_AUTO_CREATE_TABLE
)
{
if
(
tEncodeSVCreateTbReq
(
pCoder
,
&
pBlock
->
cTbReq
)
<
0
)
return
-
1
;
}
tEndEncode
(
pCoder
);
return
0
;
}
static
int32_t
tDecodeSVSubmitBlk
(
SCoder
*
pCoder
,
SVSubmitBlk
*
pBlock
,
int32_t
flags
)
{
if
(
tStartDecode
(
pCoder
)
<
0
)
return
-
1
;
if
(
tDecodeI64
(
pCoder
,
&
pBlock
->
suid
)
<
0
)
return
-
1
;
if
(
tDecodeI64
(
pCoder
,
&
pBlock
->
uid
)
<
0
)
return
-
1
;
if
(
tDecodeI32v
(
pCoder
,
&
pBlock
->
sver
)
<
0
)
return
-
1
;
if
(
tDecodeBinary
(
pCoder
,
&
pBlock
->
pData
,
&
pBlock
->
nData
)
<
0
)
return
-
1
;
if
(
flags
&
TD_AUTO_CREATE_TABLE
)
{
if
(
tDecodeSVCreateTbReq
(
pCoder
,
&
pBlock
->
cTbReq
)
<
0
)
return
-
1
;
}
tEndDecode
(
pCoder
);
return
0
;
}
int32_t
tEncodeSVSubmitReq
(
SCoder
*
pCoder
,
const
SVSubmitReq
*
pReq
)
{
int32_t
nBlocks
=
taosArrayGetSize
(
pReq
->
pArray
);
if
(
tStartEncode
(
pCoder
)
<
0
)
return
-
1
;
if
(
tEncodeI32v
(
pCoder
,
pReq
->
flags
)
<
0
)
return
-
1
;
if
(
tEncodeI32v
(
pCoder
,
nBlocks
)
<
0
)
return
-
1
;
for
(
int32_t
iBlock
=
0
;
iBlock
<
nBlocks
;
iBlock
++
)
{
if
(
tEncodeSVSubmitBlk
(
pCoder
,
(
SVSubmitBlk
*
)
taosArrayGet
(
pReq
->
pArray
,
iBlock
),
pReq
->
flags
)
<
0
)
return
-
1
;
}
tEndEncode
(
pCoder
);
return
0
;
}
int32_t
tDecodeSVSubmitReq
(
SCoder
*
pCoder
,
SVSubmitReq
*
pReq
)
{
if
(
tStartDecode
(
pCoder
)
<
0
)
return
-
1
;
if
(
tDecodeI32v
(
pCoder
,
&
pReq
->
flags
)
<
0
)
return
-
1
;
if
(
tDecodeI32v
(
pCoder
,
&
pReq
->
nBlocks
)
<
0
)
return
-
1
;
pReq
->
pBlocks
=
tCoderMalloc
(
pCoder
,
sizeof
(
SVSubmitBlk
)
*
pReq
->
nBlocks
);
if
(
pReq
->
pBlocks
==
NULL
)
return
-
1
;
for
(
int32_t
iBlock
=
0
;
iBlock
<
pReq
->
nBlocks
;
iBlock
++
)
{
if
(
tDecodeSVSubmitBlk
(
pCoder
,
pReq
->
pBlocks
+
iBlock
,
pReq
->
flags
)
<
0
)
return
-
1
;
}
tEndDecode
(
pCoder
);
return
0
;
}
\ No newline at end of file
source/dnode/vnode/CMakeLists.txt
浏览文件 @
bcd3929e
...
...
@@ -30,6 +30,7 @@ target_sources(
"src/tsdb/tsdbFS.c"
"src/tsdb/tsdbOpen.c"
"src/tsdb/tsdbMemTable.c"
"src/tsdb/tsdbMemTable2.c"
"src/tsdb/tsdbRead.c"
"src/tsdb/tsdbReadImpl.c"
"src/tsdb/tsdbSma.c"
...
...
source/dnode/vnode/src/inc/tsdb.h
浏览文件 @
bcd3929e
...
...
@@ -40,7 +40,6 @@ typedef struct STable STable;
int
tsdbMemTableCreate
(
STsdb
*
pTsdb
,
STsdbMemTable
**
ppMemTable
);
void
tsdbMemTableDestroy
(
STsdb
*
pTsdb
,
STsdbMemTable
*
pMemTable
);
int
tsdbInsertTableData
(
STsdb
*
pTsdb
,
SSubmitMsgIter
*
pMsgIter
,
SSubmitBlk
*
pBlock
,
int32_t
*
pAffectedRows
);
int
tsdbLoadDataFromCache
(
STable
*
pTable
,
SSkipListIterator
*
pIter
,
TSKEY
maxKey
,
int
maxRowsToRead
,
SDataCols
*
pCols
,
TKEY
*
filterKeys
,
int
nFilterKeys
,
bool
keepDup
,
SMergeInfo
*
pMergeInfo
);
...
...
source/dnode/vnode/src/inc/vnodeInt.h
浏览文件 @
bcd3929e
...
...
@@ -100,6 +100,7 @@ int32_t tsdbUpdateSmaWindow(STsdb* pTsdb, SSubmitReq* pMsg, int64_t version
int32_t
tsdbCreateTSma
(
STsdb
*
pTsdb
,
char
*
pMsg
);
int32_t
tsdbInsertTSmaData
(
STsdb
*
pTsdb
,
int64_t
indexUid
,
const
char
*
msg
);
int
tsdbInsertData
(
STsdb
*
pTsdb
,
int64_t
version
,
SSubmitReq
*
pMsg
,
SSubmitRsp
*
pRsp
);
int
tsdbInsertTableData
(
STsdb
*
pTsdb
,
SSubmitMsgIter
*
pMsgIter
,
SSubmitBlk
*
pBlock
,
int32_t
*
pAffectedRows
);
tsdbReaderT
*
tsdbQueryTables
(
SVnode
*
pVnode
,
SQueryTableDataCond
*
pCond
,
STableGroupInfo
*
groupList
,
uint64_t
qId
,
uint64_t
taskId
);
tsdbReaderT
tsdbQueryCacheLastT
(
STsdb
*
tsdb
,
SQueryTableDataCond
*
pCond
,
STableGroupInfo
*
groupList
,
uint64_t
qId
,
...
...
@@ -125,7 +126,6 @@ int32_t tsdbUpdateTbUidList(STsdb* pTsdb, STbUidStore* pUidStore);
void
tsdbUidStoreDestory
(
STbUidStore
*
pStore
);
void
*
tsdbUidStoreFree
(
STbUidStore
*
pStore
);
int32_t
tsdbTriggerRSma
(
STsdb
*
pTsdb
,
void
*
pMsg
,
int32_t
inputType
);
int32_t
tsdbProcessSubmitReq
(
STsdb
*
pTsdb
,
int64_t
version
,
void
*
pReq
);
typedef
struct
{
int8_t
streamType
;
// sma or other
...
...
source/dnode/vnode/src/tsdb/tsdbCommit.c
浏览文件 @
bcd3929e
...
...
@@ -69,10 +69,6 @@ static void tsdbDestroyCommitIters(SCommitH *pCommith);
static
int
tsdbCommitToFile
(
SCommitH
*
pCommith
,
SDFileSet
*
pSet
,
int
fid
);
static
void
tsdbResetCommitFile
(
SCommitH
*
pCommith
);
static
int
tsdbSetAndOpenCommitFile
(
SCommitH
*
pCommith
,
SDFileSet
*
pSet
,
int
fid
);
// static int tsdbCommitMeta(STsdbRepo *pRepo);
// static int tsdbUpdateMetaRecord(STsdbFS *pfs, SMFile *pMFile, uint64_t uid, void *cont, int contLen, bool compact);
// static int tsdbDropMetaRecord(STsdbFS *pfs, SMFile *pMFile, uint64_t uid);
// static int tsdbCompactMetaFile(STsdbRepo *pRepo, STsdbFS *pfs, SMFile *pMFile);
static
int
tsdbCommitToTable
(
SCommitH
*
pCommith
,
int
tid
);
static
int
tsdbSetCommitTable
(
SCommitH
*
pCommith
,
STable
*
pTable
);
static
int
tsdbComparKeyBlock
(
const
void
*
arg1
,
const
void
*
arg2
);
...
...
source/dnode/vnode/src/tsdb/tsdbMemTable.c
浏览文件 @
bcd3929e
...
...
@@ -191,9 +191,6 @@ int tsdbLoadDataFromCache(STable *pTable, SSkipListIterator *pIter, TSKEY maxKey
}
int
tsdbInsertTableData
(
STsdb
*
pTsdb
,
SSubmitMsgIter
*
pMsgIter
,
SSubmitBlk
*
pBlock
,
int32_t
*
pAffectedRows
)
{
// STsdbMeta *pMeta = pRepo->tsdbMeta;
// int32_t points = 0;
// STable *pTable = NULL;
SSubmitBlkIter
blkIter
=
{
0
};
STsdbMemTable
*
pMemTable
=
pTsdb
->
mem
;
void
*
tptr
;
...
...
@@ -221,8 +218,9 @@ int tsdbInsertTableData(STsdb *pTsdb, SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlo
}
// copy data to buffer pool
pBlkCopy
=
(
SSubmitBlk
*
)
vnodeBufPoolMalloc
(
pTsdb
->
mem
->
pPool
,
pMsgIter
->
dataLen
+
sizeof
(
*
pBlock
));
memcpy
(
pBlkCopy
,
pBlock
,
pMsgIter
->
dataLen
+
sizeof
(
*
pBlock
));
int32_t
tlen
=
pMsgIter
->
dataLen
+
pMsgIter
->
schemaLen
+
sizeof
(
*
pBlock
);
pBlkCopy
=
(
SSubmitBlk
*
)
vnodeBufPoolMalloc
(
pTsdb
->
mem
->
pPool
,
tlen
);
memcpy
(
pBlkCopy
,
pBlock
,
tlen
);
tInitSubmitBlkIter
(
pMsgIter
,
pBlkCopy
,
&
blkIter
);
if
(
blkIter
.
row
==
NULL
)
return
0
;
...
...
@@ -241,7 +239,7 @@ int tsdbInsertTableData(STsdb *pTsdb, SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlo
if
(
pMemTable
->
keyMin
>
keyMin
)
pMemTable
->
keyMin
=
keyMin
;
if
(
pMemTable
->
keyMax
<
keyMax
)
pMemTable
->
keyMax
=
keyMax
;
(
*
pAffectedRows
)
+
=
pMsgIter
->
numOfRows
;
(
*
pAffectedRows
)
=
pMsgIter
->
numOfRows
;
return
0
;
}
...
...
source/dnode/vnode/src/tsdb/tsdbMemTable2.c
0 → 100644
浏览文件 @
bcd3929e
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "tsdb.h"
typedef
struct
SMemTable
SMemTable
;
typedef
struct
SMemData
SMemData
;
typedef
struct
SMemSkipList
SMemSkipList
;
typedef
struct
SMemSkipListCfg
SMemSkipListCfg
;
struct
SMemTable
{
STsdb
*
pTsdb
;
TSKEY
minKey
;
TSKEY
maxKey
;
int64_t
minVer
;
int64_t
maxVer
;
int64_t
nRows
;
int32_t
nHash
;
int32_t
nBucket
;
SMemData
**
pBuckets
;
};
struct
SMemSkipListCfg
{
int8_t
maxLevel
;
int32_t
nKey
;
int32_t
nData
;
};
struct
SMemSkipList
{
int8_t
level
;
uint32_t
seed
;
};
struct
SMemData
{
SMemData
*
pHashNext
;
tb_uid_t
suid
;
tb_uid_t
uid
;
TSKEY
minKey
;
TSKEY
maxKey
;
int64_t
minVer
;
int64_t
maxVer
;
int64_t
nRows
;
SMemSkipList
sl
;
};
// SMemTable
int32_t
tsdbMemTableCreate2
(
STsdb
*
pTsdb
,
SMemTable
**
ppMemTb
)
{
SMemTable
*
pMemTb
=
NULL
;
pMemTb
=
taosMemoryCalloc
(
1
,
sizeof
(
*
pMemTb
));
if
(
pMemTb
==
NULL
)
{
terrno
=
TSDB_CODE_OUT_OF_MEMORY
;
return
-
1
;
}
pMemTb
->
pTsdb
=
pTsdb
;
pMemTb
->
minKey
=
TSKEY_MAX
;
pMemTb
->
maxKey
=
TSKEY_MIN
;
pMemTb
->
minVer
=
-
1
;
pMemTb
->
maxVer
=
-
1
;
pMemTb
->
nRows
=
0
;
pMemTb
->
nHash
=
0
;
pMemTb
->
nBucket
=
1024
;
pMemTb
->
pBuckets
=
taosMemoryCalloc
(
pMemTb
->
nBucket
,
sizeof
(
*
pMemTb
->
pBuckets
));
if
(
pMemTb
->
pBuckets
==
NULL
)
{
terrno
=
TSDB_CODE_OUT_OF_MEMORY
;
return
-
1
;
}
*
ppMemTb
=
pMemTb
;
return
0
;
}
int32_t
tsdbMemTableDestroy2
(
STsdb
*
pTsdb
,
SMemTable
*
pMT
)
{
// TODO
return
0
;
}
// SMemData
// SMemSkipList
\ No newline at end of file
source/dnode/vnode/src/tsdb/tsdbSma.c
浏览文件 @
bcd3929e
...
...
@@ -1962,6 +1962,21 @@ int32_t tsdbUpdateTbUidList(STsdb *pTsdb, STbUidStore *pStore) {
return
TSDB_CODE_SUCCESS
;
}
static
int32_t
tsdbProcessSubmitReq
(
STsdb
*
pTsdb
,
int64_t
version
,
void
*
pReq
)
{
if
(
!
pReq
)
{
terrno
=
TSDB_CODE_INVALID_PTR
;
return
TSDB_CODE_FAILED
;
}
SSubmitReq
*
pSubmitReq
=
(
SSubmitReq
*
)
pReq
;
if
(
tsdbInsertData
(
pTsdb
,
version
,
pSubmitReq
,
NULL
)
<
0
)
{
return
TSDB_CODE_FAILED
;
}
return
TSDB_CODE_SUCCESS
;
}
static
int32_t
tsdbFetchSubmitReqSuids
(
SSubmitReq
*
pMsg
,
STbUidStore
*
pStore
)
{
ASSERT
(
pMsg
!=
NULL
);
SSubmitMsgIter
msgIter
=
{
0
};
...
...
source/dnode/vnode/src/vnd/vnodeSvr.c
浏览文件 @
bcd3929e
...
...
@@ -450,7 +450,7 @@ static int vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq, in
SCoder
coder
=
{
0
};
int
ret
;
pRsp
->
msgType
=
TDMT_VND_
CREATE_STB
_RSP
;
pRsp
->
msgType
=
TDMT_VND_
DROP_TABLE
_RSP
;
pRsp
->
pCont
=
NULL
;
pRsp
->
contLen
=
0
;
pRsp
->
code
=
TSDB_CODE_SUCCESS
;
...
...
@@ -473,10 +473,14 @@ static int vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq, in
/* code */
ret
=
metaDropTable
(
pVnode
->
pMeta
,
version
,
pDropTbReq
);
if
(
ret
<
0
)
{
if
(
pDropTbReq
->
igNotExists
&&
terrno
==
TSDB_CODE_VND_TABLE_NOT_EXIST
)
{
dropTbRsp
.
code
=
TSDB_CODE_SUCCESS
;
}
else
{
dropTbRsp
.
code
=
terrno
;
}
}
else
{
dropTbRsp
.
code
=
TSDB_CODE_SUCCESS
;
}
taosArrayPush
(
rsp
.
pArray
,
&
dropTbRsp
);
}
...
...
@@ -489,19 +493,54 @@ _exit:
static
int
vnodeProcessSubmitReq
(
SVnode
*
pVnode
,
int64_t
version
,
void
*
pReq
,
int32_t
len
,
SRpcMsg
*
pRsp
)
{
SSubmitReq
*
pSubmitReq
=
(
SSubmitReq
*
)
pReq
;
SSubmitMsgIter
msgIter
=
{
0
};
SSubmitBlk
*
pBlock
;
SSubmitRsp
rsp
=
{
0
};
SVCreateTbReq
createTbReq
=
{
0
};
SCoder
coder
=
{
0
};
int32_t
nRows
;
pRsp
->
code
=
0
;
// handle the request
if
(
tsdbInsertData
(
pVnode
->
pTsdb
,
version
,
pSubmitReq
,
&
rsp
)
<
0
)
{
if
(
tInitSubmitMsgIter
(
pSubmitReq
,
&
msgIter
)
<
0
)
{
pRsp
->
code
=
TSDB_CODE_INVALID_MSG
;
goto
_exit
;
}
for
(;;)
{
tGetSubmitMsgNext
(
&
msgIter
,
&
pBlock
);
if
(
pBlock
==
NULL
)
break
;
// create table for auto create table mode
if
(
msgIter
.
schemaLen
>
0
)
{
tCoderInit
(
&
coder
,
TD_LITTLE_ENDIAN
,
pBlock
->
data
,
msgIter
.
schemaLen
,
TD_DECODER
);
if
(
tDecodeSVCreateTbReq
(
&
coder
,
&
createTbReq
)
<
0
)
{
pRsp
->
code
=
TSDB_CODE_INVALID_MSG
;
tCoderClear
(
&
coder
);
goto
_exit
;
}
if
(
metaCreateTable
(
pVnode
->
pMeta
,
version
,
&
createTbReq
)
<
0
)
{
if
(
terrno
!=
TSDB_CODE_TDB_TABLE_ALREADY_EXIST
)
{
pRsp
->
code
=
terrno
;
return
-
1
;
tCoderClear
(
&
coder
);
goto
_exit
;
}
}
// pRsp->msgType = TDMT_VND_SUBMIT_RSP
;
// vnodeProcessSubmitReq(pVnode, ptr, pRsp);
tCoderClear
(
&
coder
)
;
}
if
(
tsdbInsertTableData
(
pVnode
->
pTsdb
,
&
msgIter
,
pBlock
,
&
nRows
)
<
0
)
{
pRsp
->
code
=
terrno
;
goto
_exit
;
}
rsp
.
numOfRows
+=
nRows
;
}
_exit:
// encode the response (TODO)
pRsp
->
pCont
=
rpcMallocCont
(
sizeof
(
SSubmitRsp
));
memcpy
(
pRsp
->
pCont
,
&
rsp
,
sizeof
(
rsp
));
...
...
@@ -511,18 +550,3 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
return
0
;
}
int32_t
tsdbProcessSubmitReq
(
STsdb
*
pTsdb
,
int64_t
version
,
void
*
pReq
)
{
if
(
!
pReq
)
{
terrno
=
TSDB_CODE_INVALID_PTR
;
return
TSDB_CODE_FAILED
;
}
SSubmitReq
*
pSubmitReq
=
(
SSubmitReq
*
)
pReq
;
if
(
tsdbInsertData
(
pTsdb
,
version
,
pSubmitReq
,
NULL
)
<
0
)
{
return
TSDB_CODE_FAILED
;
}
return
TSDB_CODE_SUCCESS
;
}
source/libs/parser/src/parInsert.c
浏览文件 @
bcd3929e
...
...
@@ -22,7 +22,6 @@
#include "ttime.h"
#include "ttypes.h"
// clang-format off
#define NEXT_TOKEN(pSql, sToken) \
do { \
int32_t index = 0; \
...
...
@@ -248,12 +247,11 @@ static int32_t getTableMetaImpl(SInsertParseContext* pCxt, SToken* pTname, bool
}
else
{
CHECK_CODE
(
catalogGetTableMeta
(
pBasicCtx
->
pCatalog
,
pBasicCtx
->
pTransporter
,
&
pBasicCtx
->
mgmtEpSet
,
&
name
,
&
pCxt
->
pTableMeta
));
}
SVgroupInfo
vg
;
CHECK_CODE
(
catalogGetTableHashVgroup
(
pBasicCtx
->
pCatalog
,
pBasicCtx
->
pTransporter
,
&
pBasicCtx
->
mgmtEpSet
,
&
name
,
&
vg
));
CHECK_CODE
(
taosHashPut
(
pCxt
->
pVgroupsHashObj
,
(
const
char
*
)
&
vg
.
vgId
,
sizeof
(
vg
.
vgId
),
(
char
*
)
&
vg
,
sizeof
(
vg
)));
}
return
TSDB_CODE_SUCCESS
;
}
...
...
@@ -828,12 +826,21 @@ static int32_t cloneTableMeta(STableMeta* pSrc, STableMeta** pDst) {
return
TSDB_CODE_SUCCESS
;
}
static
int32_t
storeTableMeta
(
SHashObj
*
pHash
,
const
char
*
pName
,
int32_t
len
,
STableMeta
*
pMeta
)
{
static
int32_t
storeTableMeta
(
SInsertParseContext
*
pCxt
,
SHashObj
*
pHash
,
SName
*
pTableName
,
const
char
*
pName
,
int32_t
len
,
STableMeta
*
pMeta
)
{
SVgroupInfo
vg
;
SParseContext
*
pBasicCtx
=
pCxt
->
pComCxt
;
CHECK_CODE
(
catalogGetTableHashVgroup
(
pBasicCtx
->
pCatalog
,
pBasicCtx
->
pTransporter
,
&
pBasicCtx
->
mgmtEpSet
,
pTableName
,
&
vg
));
CHECK_CODE
(
taosHashPut
(
pCxt
->
pVgroupsHashObj
,
(
const
char
*
)
&
vg
.
vgId
,
sizeof
(
vg
.
vgId
),
(
char
*
)
&
vg
,
sizeof
(
vg
)));
pMeta
->
uid
=
tGenIdPI64
();
pMeta
->
vgId
=
vg
.
vgId
;
STableMeta
*
pBackup
=
NULL
;
if
(
TSDB_CODE_SUCCESS
!=
cloneTableMeta
(
pMeta
,
&
pBackup
))
{
return
TSDB_CODE_TSC_OUT_OF_MEMORY
;
}
pBackup
->
uid
=
tGenIdPI64
();
return
taosHashPut
(
pHash
,
pName
,
len
,
&
pBackup
,
POINTER_BYTES
);
}
...
...
@@ -856,7 +863,7 @@ static int32_t parseUsingClause(SInsertParseContext* pCxt, SToken* pTbnameToken)
if
(
TSDB_SUPER_TABLE
!=
pCxt
->
pTableMeta
->
tableType
)
{
return
buildInvalidOperationMsg
(
&
pCxt
->
msg
,
"create table only from super table is allowed"
);
}
CHECK_CODE
(
storeTableMeta
(
pCxt
->
pSubTableHashObj
,
tbFName
,
len
,
pCxt
->
pTableMeta
));
CHECK_CODE
(
storeTableMeta
(
pCxt
,
pCxt
->
pSubTableHashObj
,
&
name
,
tbFName
,
len
,
pCxt
->
pTableMeta
));
SSchema
*
pTagsSchema
=
getTableTagSchema
(
pCxt
->
pTableMeta
);
setBoundColumnInfo
(
&
pCxt
->
tags
,
pTagsSchema
,
getNumOfTags
(
pCxt
->
pTableMeta
));
...
...
@@ -1262,8 +1269,9 @@ int32_t qBuildStmtOutput(SQuery* pQuery, SHashObj* pVgHash, SHashObj* pBlockHash
return
TSDB_CODE_SUCCESS
;
}
int32_t
qBindStmtTagsValue
(
void
*
pBlock
,
void
*
boundTags
,
int64_t
suid
,
SName
*
pName
,
TAOS_MULTI_BIND
*
bind
,
char
*
msgBuf
,
int32_t
msgBufLen
){
STableDataBlocks
*
pDataBlock
=
(
STableDataBlocks
*
)
pBlock
;
int32_t
qBindStmtTagsValue
(
void
*
pBlock
,
void
*
boundTags
,
int64_t
suid
,
SName
*
pName
,
TAOS_MULTI_BIND
*
bind
,
char
*
msgBuf
,
int32_t
msgBufLen
)
{
STableDataBlocks
*
pDataBlock
=
(
STableDataBlocks
*
)
pBlock
;
SMsgBuf
pBuf
=
{.
buf
=
msgBuf
,
.
len
=
msgBufLen
};
SParsedDataColInfo
*
tags
=
(
SParsedDataColInfo
*
)
boundTags
;
if
(
NULL
==
tags
)
{
...
...
@@ -1312,9 +1320,8 @@ int32_t qBindStmtTagsValue(void *pBlock, void *boundTags, int64_t suid, SName *p
return
TSDB_CODE_SUCCESS
;
}
int32_t
qBindStmtColsValue
(
void
*
pBlock
,
TAOS_MULTI_BIND
*
bind
,
char
*
msgBuf
,
int32_t
msgBufLen
)
{
STableDataBlocks
*
pDataBlock
=
(
STableDataBlocks
*
)
pBlock
;
int32_t
qBindStmtColsValue
(
void
*
pBlock
,
TAOS_MULTI_BIND
*
bind
,
char
*
msgBuf
,
int32_t
msgBufLen
)
{
STableDataBlocks
*
pDataBlock
=
(
STableDataBlocks
*
)
pBlock
;
SSchema
*
pSchema
=
getTableColumnSchema
(
pDataBlock
->
pTableMeta
);
int32_t
extendedRowSize
=
getExtendedRowSize
(
pDataBlock
);
SParsedDataColInfo
*
spd
=
&
pDataBlock
->
boundColumnInfo
;
...
...
@@ -1391,8 +1398,9 @@ int32_t qBindStmtColsValue(void *pBlock, TAOS_MULTI_BIND *bind, char *msgBuf, in
return
TSDB_CODE_SUCCESS
;
}
int32_t
qBindStmtSingleColValue
(
void
*
pBlock
,
TAOS_MULTI_BIND
*
bind
,
char
*
msgBuf
,
int32_t
msgBufLen
,
int32_t
colIdx
,
int32_t
rowNum
)
{
STableDataBlocks
*
pDataBlock
=
(
STableDataBlocks
*
)
pBlock
;
int32_t
qBindStmtSingleColValue
(
void
*
pBlock
,
TAOS_MULTI_BIND
*
bind
,
char
*
msgBuf
,
int32_t
msgBufLen
,
int32_t
colIdx
,
int32_t
rowNum
)
{
STableDataBlocks
*
pDataBlock
=
(
STableDataBlocks
*
)
pBlock
;
SSchema
*
pSchema
=
getTableColumnSchema
(
pDataBlock
->
pTableMeta
);
int32_t
extendedRowSize
=
getExtendedRowSize
(
pDataBlock
);
SParsedDataColInfo
*
spd
=
&
pDataBlock
->
boundColumnInfo
;
...
...
@@ -1459,7 +1467,7 @@ int32_t qBindStmtSingleColValue(void *pBlock, TAOS_MULTI_BIND *bind, char *msgBu
}
#ifdef TD_DEBUG_PRINT_ROW
if
(
rowEnd
)
{
if
(
rowEnd
)
{
STSchema
*
pSTSchema
=
tdGetSTSChemaFromSSChema
(
&
pSchema
,
spd
->
numOfCols
);
tdSRowPrint
(
row
,
pSTSchema
,
__func__
);
taosMemoryFree
(
pSTSchema
);
...
...
source/libs/parser/src/parInsertData.c
浏览文件 @
bcd3929e
...
...
@@ -498,14 +498,9 @@ int32_t mergeTableDataBlocks(SHashObj* pHashObj, uint8_t payloadType, SArray** p
ASSERT
(
blkKeyInfo
.
pKeyTuple
!=
NULL
&&
pBlocks
->
numOfRows
>
0
);
}
int32_t
len
=
pBlocks
->
numOfRows
*
(
isRawPayload
?
(
pOneTableBlock
->
rowSize
+
expandSize
)
:
getExtendedRowSize
(
pOneTableBlock
))
+
sizeof
(
STColumn
)
*
getNumOfColumns
(
pOneTableBlock
->
pTableMeta
);
// erase the empty space reserved for binary data
int32_t
finalLen
=
trimDataBlock
(
dataBuf
->
pData
+
dataBuf
->
size
,
pOneTableBlock
,
blkKeyInfo
.
pKeyTuple
,
isRawPayload
);
assert
(
finalLen
<=
len
);
dataBuf
->
size
+=
(
finalLen
+
sizeof
(
SSubmitBlk
));
assert
(
dataBuf
->
size
<=
dataBuf
->
nAllocSize
);
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录