You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2021/08/04 03:47:33 UTC
[iotdb] branch query-memory-leak created (now 8d5d616)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a change to branch query-memory-leak
in repository https://gitbox.apache.org/repos/asf/iotdb.git.
at 8d5d616 merge 0.12 and fix conflicts
This branch includes the following new commits:
new 0cb9ecf fix leak in sessionIdToStatementId & statementIdToQueryId
new a728ec3 fix leak in cpp client: statementId not set in TSCloseOperationReq
new 83c34ff fix wrong return status
new 24e74f9 fix format
new 8d5d616 merge 0.12 and fix conflicts
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[iotdb] 03/05: fix wrong return status
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch query-memory-leak
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 83c34ff2ebfdd96e1a7d7b371b34ba4d1b9595fe
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Mon Aug 2 12:37:41 2021 +0800
fix wrong return status
---
server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
index 7073d98..38aa59e 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
@@ -310,10 +310,10 @@ public class TSServiceImpl implements TSIService.Iface {
} else {
sessionManager.closeStatement(req.sessionId, req.statementId);
}
+ return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
+ } else {
return RpcUtils.getStatus(
TSStatusCode.CLOSE_OPERATION_ERROR, "statement id not set by client.");
- } else {
- return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
}
} catch (Exception e) {
return onNPEOrUnexpectedException(
[iotdb] 02/05: fix leak in cpp client: statementId not set in
TSCloseOperationReq
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch query-memory-leak
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit a728ec3a70d1ddd3680c9cacb87f3e6f592d3233
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Mon Aug 2 12:31:27 2021 +0800
fix leak in cpp client: statementId not set in TSCloseOperationReq
---
client-cpp/src/main/Session.cpp | 5 +++--
client-cpp/src/main/Session.h | 4 +++-
2 files changed, 6 insertions(+), 3 deletions(-)
diff --git a/client-cpp/src/main/Session.cpp b/client-cpp/src/main/Session.cpp
index 8cc696b..286e3c9 100644
--- a/client-cpp/src/main/Session.cpp
+++ b/client-cpp/src/main/Session.cpp
@@ -339,6 +339,7 @@ void SessionDataSet::closeOperationHandle()
{
shared_ptr<TSCloseOperationReq> closeReq(new TSCloseOperationReq());
closeReq->__set_sessionId(sessionId);
+ closeReq->__set_statementId(statementId);
closeReq->__set_queryId(queryId);
shared_ptr<TSStatus> closeResp(new TSStatus());
try
@@ -576,7 +577,7 @@ void Session::close()
}
shared_ptr<TSCloseSessionReq> req(new TSCloseSessionReq());
req->__set_sessionId(sessionId);
- try
+ try
{
shared_ptr<TSStatus> resp(new TSStatus());
client->closeSession(*resp,*req);
@@ -1146,7 +1147,7 @@ unique_ptr<SessionDataSet> Session::executeQueryStatement(string sql)
}
shared_ptr<TSQueryDataSet> queryDataSet(new TSQueryDataSet(resp->queryDataSet));
return unique_ptr<SessionDataSet>(new SessionDataSet(
- sql, resp->columns, resp->dataTypeList, resp->queryId, client, sessionId, queryDataSet));
+ sql, resp->columns, resp->dataTypeList, resp->queryId, statementId, client, sessionId, queryDataSet));
}
void Session::executeNonQueryStatement(string sql)
diff --git a/client-cpp/src/main/Session.h b/client-cpp/src/main/Session.h
index 62995a4..f2c0ea7 100644
--- a/client-cpp/src/main/Session.h
+++ b/client-cpp/src/main/Session.h
@@ -505,6 +505,7 @@ private:
bool hasCachedRecord = false;
std::string sql;
int64_t queryId;
+ int64_t statementId;
int64_t sessionId;
std::shared_ptr<TSIServiceIf> client;
int batchSize = 1024;
@@ -528,12 +529,13 @@ private:
public:
SessionDataSet(){}
- SessionDataSet(std::string sql, std::vector<std::string>& columnNameList, std::vector<std::string>& columnTypeList, int64_t queryId,
+ SessionDataSet(std::string sql, std::vector<std::string>& columnNameList, std::vector<std::string>& columnTypeList, int64_t queryId, int64_t statementId,
std::shared_ptr<TSIServiceIf> client, int64_t sessionId, std::shared_ptr<TSQueryDataSet> queryDataSet) : tsQueryDataSetTimeBuffer(queryDataSet->time)
{
this->sessionId = sessionId;
this->sql = sql;
this->queryId = queryId;
+ this->statementId = statementId;
this->client = client;
this->columnNameList = columnNameList;
this->currentBitmap = new char[columnNameList.size()];
[iotdb] 04/05: fix format
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch query-memory-leak
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 24e74f91ea62a39703939b996a684bac065b1143
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Wed Aug 4 10:45:33 2021 +0800
fix format
---
client-cpp/src/main/Session.cpp | 735 ++++++++++++++++++----------------------
client-cpp/src/main/Session.h | 541 ++++++++++++++++-------------
2 files changed, 645 insertions(+), 631 deletions(-)
diff --git a/client-cpp/src/main/Session.cpp b/client-cpp/src/main/Session.cpp
index 286e3c9..6f4b240 100644
--- a/client-cpp/src/main/Session.cpp
+++ b/client-cpp/src/main/Session.cpp
@@ -23,17 +23,17 @@ using namespace std;
TSDataType::TSDataType getTSDataTypeFromString(string str) {
// BOOLEAN, INT32, INT64, FLOAT, DOUBLE, TEXT, NULLTYPE
- if (str == "BOOLEAN") return TSDataType::BOOLEAN;
- else if(str == "INT32") return TSDataType::INT32;
- else if(str == "INT64") return TSDataType::INT64;
- else if(str == "FLOAT") return TSDataType::FLOAT;
- else if(str == "DOUBLE") return TSDataType::DOUBLE;
- else if(str == "TEXT") return TSDataType::TEXT;
- else if(str == "NULLTYPE") return TSDataType::NULLTYPE;
+ if (str == "BOOLEAN") return TSDataType::BOOLEAN;
+ else if (str == "INT32") return TSDataType::INT32;
+ else if (str == "INT64") return TSDataType::INT64;
+ else if (str == "FLOAT") return TSDataType::FLOAT;
+ else if (str == "DOUBLE") return TSDataType::DOUBLE;
+ else if (str == "TEXT") return TSDataType::TEXT;
+ else if (str == "NULLTYPE") return TSDataType::NULLTYPE;
return TSDataType::TEXT;
}
-void RpcUtils::verifySuccess(TSStatus& status) {
+void RpcUtils::verifySuccess(TSStatus &status) {
if (status.code == TSStatusCode::MULTIPLE_ERROR) {
verifySuccess(status.subStatus);
return;
@@ -44,7 +44,8 @@ void RpcUtils::verifySuccess(TSStatus& status) {
throw IoTDBConnectionException(buf);
}
}
-void RpcUtils::verifySuccess(vector<TSStatus>& statuses) {
+
+void RpcUtils::verifySuccess(vector <TSStatus> &statuses) {
for (TSStatus status : statuses) {
if (status.code != TSStatusCode::SUCCESS_STATUS) {
char buf[111];
@@ -59,36 +60,45 @@ TSStatus RpcUtils::getStatus(TSStatusCode::TSStatusCode tsStatusCode) {
tmpTSStatus.__set_code(tsStatusCode);
return tmpTSStatus;
}
+
TSStatus RpcUtils::getStatus(int code, string message) {
TSStatus status = TSStatus();
status.__set_code(code);
status.__set_message(message);
return status;
}
-shared_ptr<TSExecuteStatementResp> RpcUtils::getTSExecuteStatementResp(TSStatusCode::TSStatusCode tsStatusCode) {
+
+shared_ptr <TSExecuteStatementResp> RpcUtils::getTSExecuteStatementResp(TSStatusCode::TSStatusCode tsStatusCode) {
TSStatus status = getStatus(tsStatusCode);
return getTSExecuteStatementResp(status);
}
-shared_ptr<TSExecuteStatementResp> RpcUtils::getTSExecuteStatementResp(TSStatusCode::TSStatusCode tsStatusCode, string message) {
+
+shared_ptr <TSExecuteStatementResp>
+RpcUtils::getTSExecuteStatementResp(TSStatusCode::TSStatusCode tsStatusCode, string message) {
TSStatus status = getStatus(tsStatusCode, message);
return getTSExecuteStatementResp(status);
}
-shared_ptr<TSExecuteStatementResp> RpcUtils::getTSExecuteStatementResp(TSStatus& status) {
- shared_ptr<TSExecuteStatementResp> resp(new TSExecuteStatementResp());
+
+shared_ptr <TSExecuteStatementResp> RpcUtils::getTSExecuteStatementResp(TSStatus &status) {
+ shared_ptr <TSExecuteStatementResp> resp(new TSExecuteStatementResp());
TSStatus tsStatus(status);
resp->status = status;
return resp;
}
-shared_ptr<TSFetchResultsResp> RpcUtils::getTSFetchResultsResp(TSStatusCode::TSStatusCode tsStatusCode) {
+
+shared_ptr <TSFetchResultsResp> RpcUtils::getTSFetchResultsResp(TSStatusCode::TSStatusCode tsStatusCode) {
TSStatus status = getStatus(tsStatusCode);
return getTSFetchResultsResp(status);
}
-shared_ptr<TSFetchResultsResp> RpcUtils::getTSFetchResultsResp(TSStatusCode::TSStatusCode tsStatusCode, string appendMessage) {
+
+shared_ptr <TSFetchResultsResp>
+RpcUtils::getTSFetchResultsResp(TSStatusCode::TSStatusCode tsStatusCode, string appendMessage) {
TSStatus status = getStatus(tsStatusCode, appendMessage);
return getTSFetchResultsResp(status);
}
-shared_ptr<TSFetchResultsResp> RpcUtils::getTSFetchResultsResp(TSStatus& status) {
- shared_ptr<TSFetchResultsResp> resp(new TSFetchResultsResp());
+
+shared_ptr <TSFetchResultsResp> RpcUtils::getTSFetchResultsResp(TSStatus &status) {
+ shared_ptr <TSFetchResultsResp> resp(new TSFetchResultsResp());
TSStatus tsStatus(status);
resp->__set_status(tsStatus);
return resp;
@@ -116,37 +126,37 @@ int Tablet::getValueByteSize() {
int valueOccupation = 0;
for (int i = 0; i < schemas.size(); i++) {
switch (schemas[i].second) {
- case TSDataType::BOOLEAN:
- valueOccupation += rowSize;
- break;
- case TSDataType::INT32:
- valueOccupation += rowSize * 4;
- break;
- case TSDataType::INT64:
- valueOccupation += rowSize * 8;
- break;
- case TSDataType::FLOAT:
- valueOccupation += rowSize * 4;
- break;
- case TSDataType::DOUBLE:
- valueOccupation += rowSize * 8;
- break;
- case TSDataType::TEXT:
- valueOccupation += rowSize * 4;
- for (string value : values[i]) {
- valueOccupation += value.size();
- }
- break;
- default:
- char buf[111];
- sprintf(buf, "Data type %d is not supported.", schemas[i].second);
- throw UnSupportedDataTypeException(buf);
+ case TSDataType::BOOLEAN:
+ valueOccupation += rowSize;
+ break;
+ case TSDataType::INT32:
+ valueOccupation += rowSize * 4;
+ break;
+ case TSDataType::INT64:
+ valueOccupation += rowSize * 8;
+ break;
+ case TSDataType::FLOAT:
+ valueOccupation += rowSize * 4;
+ break;
+ case TSDataType::DOUBLE:
+ valueOccupation += rowSize * 8;
+ break;
+ case TSDataType::TEXT:
+ valueOccupation += rowSize * 4;
+ for (string value : values[i]) {
+ valueOccupation += value.size();
+ }
+ break;
+ default:
+ char buf[111];
+ sprintf(buf, "Data type %d is not supported.", schemas[i].second);
+ throw UnSupportedDataTypeException(buf);
}
}
return valueOccupation;
}
-string SessionUtils::getTime(Tablet& tablet) {
+string SessionUtils::getTime(Tablet &tablet) {
MyStringBuffer timeBuffer;
for (int i = 0; i < tablet.rowSize; i++) {
timeBuffer.putLong(tablet.timestamps[i]);
@@ -154,78 +164,74 @@ string SessionUtils::getTime(Tablet& tablet) {
return timeBuffer.str;
}
-string SessionUtils::getValue(Tablet& tablet) {
+string SessionUtils::getValue(Tablet &tablet) {
MyStringBuffer valueBuffer;
for (int i = 0; i < tablet.schemas.size(); i++) {
TSDataType::TSDataType dataType = tablet.schemas[i].second;
- switch (dataType)
- {
- case TSDataType::BOOLEAN:
- for (int index = 0; index < tablet.rowSize; index++) {
- valueBuffer.putBool(tablet.values[i][index] == "true" ? true : false);
- }
- break;
- case TSDataType::INT32:
- for (int index = 0; index < tablet.rowSize; index++) {
- valueBuffer.putInt(stoi(tablet.values[i][index]));
- }
- break;
- case TSDataType::INT64:
- for (int index = 0; index < tablet.rowSize; index++) {
- valueBuffer.putLong(stol(tablet.values[i][index]));
- }
- break;
- case TSDataType::FLOAT:
- for (int index = 0; index < tablet.rowSize; index++) {
- valueBuffer.putFloat(stof(tablet.values[i][index]));
- }
- break;
- case TSDataType::DOUBLE:
- for (int index = 0; index < tablet.rowSize; index++) {
- valueBuffer.putDouble(stod(tablet.values[i][index]));
- }
- break;
- case TSDataType::TEXT:
- for (int index = 0; index < tablet.rowSize; index++) {
- valueBuffer.putString(tablet.values[i][index]);
- }
- break;
- default:
- char buf[111];
- sprintf(buf, "Data type %d is not supported.", dataType);
- throw UnSupportedDataTypeException(buf);
- break;
+ switch (dataType) {
+ case TSDataType::BOOLEAN:
+ for (int index = 0; index < tablet.rowSize; index++) {
+ valueBuffer.putBool(tablet.values[i][index] == "true" ? true : false);
+ }
+ break;
+ case TSDataType::INT32:
+ for (int index = 0; index < tablet.rowSize; index++) {
+ valueBuffer.putInt(stoi(tablet.values[i][index]));
+ }
+ break;
+ case TSDataType::INT64:
+ for (int index = 0; index < tablet.rowSize; index++) {
+ valueBuffer.putLong(stol(tablet.values[i][index]));
+ }
+ break;
+ case TSDataType::FLOAT:
+ for (int index = 0; index < tablet.rowSize; index++) {
+ valueBuffer.putFloat(stof(tablet.values[i][index]));
+ }
+ break;
+ case TSDataType::DOUBLE:
+ for (int index = 0; index < tablet.rowSize; index++) {
+ valueBuffer.putDouble(stod(tablet.values[i][index]));
+ }
+ break;
+ case TSDataType::TEXT:
+ for (int index = 0; index < tablet.rowSize; index++) {
+ valueBuffer.putString(tablet.values[i][index]);
+ }
+ break;
+ default:
+ char buf[111];
+ sprintf(buf, "Data type %d is not supported.", dataType);
+ throw UnSupportedDataTypeException(buf);
+ break;
}
}
return valueBuffer.str;
}
-int SessionDataSet::getBatchSize()
-{
+int SessionDataSet::getBatchSize() {
return batchSize;
}
-void SessionDataSet::setBatchSize(int batchSize)
-{
+void SessionDataSet::setBatchSize(int batchSize) {
this->batchSize = batchSize;
}
-vector<string> SessionDataSet::getColumnNames() { return this->columnNameList; }
+vector <string> SessionDataSet::getColumnNames() { return this->columnNameList; }
-bool SessionDataSet::hasNext()
-{
+bool SessionDataSet::hasNext() {
if (hasCachedRecord) {
return true;
}
if (!tsQueryDataSetTimeBuffer.hasRemaining()) {
- shared_ptr<TSFetchResultsReq> req(new TSFetchResultsReq());
+ shared_ptr <TSFetchResultsReq> req(new TSFetchResultsReq());
req->__set_sessionId(sessionId);
req->__set_statement(sql);
req->__set_fetchSize(batchSize);
req->__set_queryId(queryId);
req->__set_isAlign(true);
try {
- shared_ptr<TSFetchResultsResp> resp(new TSFetchResultsResp());
+ shared_ptr <TSFetchResultsResp> resp(new TSFetchResultsResp());
client->fetchResults(*resp, *req);
RpcUtils::verifySuccess(resp->status);
@@ -237,8 +243,7 @@ bool SessionDataSet::hasNext()
rowsIndex = 0;
}
}
- catch (IoTDBConnectionException e)
- {
+ catch (IoTDBConnectionException e) {
char buf[111];
sprintf(buf, "Cannot fetch result from server, because of network connection: %s", e.what());
throw IoTDBConnectionException(buf);
@@ -251,7 +256,7 @@ bool SessionDataSet::hasNext()
}
void SessionDataSet::constructOneRow() {
- vector<Field> outFields;
+ vector <Field> outFields;
int loc = 0;
for (int i = 0; i < columnSize; i++) {
Field field;
@@ -269,41 +274,41 @@ void SessionDataSet::constructOneRow() {
TSDataType::TSDataType dataType = getTSDataTypeFromString(columnTypeDeduplicatedList[loc]);
field.dataType = dataType;
switch (dataType) {
- case TSDataType::BOOLEAN: {
- bool booleanValue = valueBuffer->getBool();
- field.boolV = booleanValue;
- break;
- }
- case TSDataType::INT32: {
- int intValue = valueBuffer->getInt();
- field.intV = intValue;
- break;
- }
- case TSDataType::INT64: {
- int64_t longValue = valueBuffer->getLong();
- field.longV = longValue;
- break;
- }
- case TSDataType::FLOAT: {
- float floatValue = valueBuffer->getFloat();
- field.floatV = floatValue;
- break;
- }
- case TSDataType::DOUBLE: {
- double doubleValue = valueBuffer->getDouble();
- field.doubleV = doubleValue;
- break;
- }
- case TSDataType::TEXT: {
- string stringValue = valueBuffer->getString();
- field.stringV = stringValue;
- break;
- }
- default: {
- char buf[111];
- sprintf(buf, "Data type %s is not supported.", columnTypeDeduplicatedList[i].c_str());
- throw UnSupportedDataTypeException(buf);
- }
+ case TSDataType::BOOLEAN: {
+ bool booleanValue = valueBuffer->getBool();
+ field.boolV = booleanValue;
+ break;
+ }
+ case TSDataType::INT32: {
+ int intValue = valueBuffer->getInt();
+ field.intV = intValue;
+ break;
+ }
+ case TSDataType::INT64: {
+ int64_t longValue = valueBuffer->getLong();
+ field.longV = longValue;
+ break;
+ }
+ case TSDataType::FLOAT: {
+ float floatValue = valueBuffer->getFloat();
+ field.floatV = floatValue;
+ break;
+ }
+ case TSDataType::DOUBLE: {
+ double doubleValue = valueBuffer->getDouble();
+ field.doubleV = doubleValue;
+ break;
+ }
+ case TSDataType::TEXT: {
+ string stringValue = valueBuffer->getString();
+ field.stringV = stringValue;
+ break;
+ }
+ default: {
+ char buf[111];
+ sprintf(buf, "Data type %s is not supported.", columnTypeDeduplicatedList[i].c_str());
+ throw UnSupportedDataTypeException(buf);
+ }
}
} else {
field.dataType = TSDataType::NULLTYPE;
@@ -323,8 +328,7 @@ bool SessionDataSet::isNull(int index, int rowNum) {
return ((flag >> shift) & bitmap) == 0;
}
-RowRecord* SessionDataSet::next()
-{
+RowRecord *SessionDataSet::next() {
if (!hasCachedRecord) {
if (!hasNext()) {
return NULL;
@@ -335,22 +339,19 @@ RowRecord* SessionDataSet::next()
return &rowRecord;
}
-void SessionDataSet::closeOperationHandle()
-{
- shared_ptr<TSCloseOperationReq> closeReq(new TSCloseOperationReq());
+void SessionDataSet::closeOperationHandle() {
+ shared_ptr <TSCloseOperationReq> closeReq(new TSCloseOperationReq());
closeReq->__set_sessionId(sessionId);
closeReq->__set_statementId(statementId);
closeReq->__set_queryId(queryId);
- shared_ptr<TSStatus> closeResp(new TSStatus());
- try
- {
- client->closeOperation(*closeResp,*closeReq);
+ shared_ptr <TSStatus> closeResp(new TSStatus());
+ try {
+ client->closeOperation(*closeResp, *closeReq);
RpcUtils::verifySuccess(*closeResp);
}
- catch (IoTDBConnectionException e)
- {
+ catch (IoTDBConnectionException e) {
char buf[111];
- sprintf(buf,"Error occurs when connecting to server for close operation, because: %s",e.what());
+ sprintf(buf, "Error occurs when connecting to server for close operation, because: %s", e.what());
throw IoTDBConnectionException(buf);
}
}
@@ -361,7 +362,7 @@ void SessionDataSet::closeOperationHandle()
*
* @return whether the batch has been sorted
*/
-bool Session::checkSorted(Tablet& tablet) {
+bool Session::checkSorted(Tablet &tablet) {
for (int i = 1; i < tablet.rowSize; i++) {
if (tablet.timestamps[i] < tablet.timestamps[i - 1]) {
return false;
@@ -370,7 +371,7 @@ bool Session::checkSorted(Tablet& tablet) {
return true;
}
-bool Session::checkSorted(vector<int64_t>& times) {
+bool Session::checkSorted(vector <int64_t> ×) {
for (int i = 1; i < times.size(); i++) {
if (times[i] < times[i - 1]) {
return false;
@@ -379,13 +380,13 @@ bool Session::checkSorted(vector<int64_t>& times) {
return true;
}
-void Session::sortTablet(Tablet& tablet) {
+void Session::sortTablet(Tablet &tablet) {
/*
* following part of code sort the batch data by time,
* so we can insert continuous data in value list to get a better performance
*/
- // sort to get index, and use index to sort value list
- int* index = new int[tablet.rowSize];
+ // sort to get index, and use index to sort value list
+ int *index = new int[tablet.rowSize];
for (int i = 0; i < tablet.rowSize; i++) {
index[i] = i;
}
@@ -399,7 +400,7 @@ void Session::sortTablet(Tablet& tablet) {
delete[] index;
}
-void Session::sortIndexByTimestamp(int* index, std::vector<int64_t>& timestamps, int length) {
+void Session::sortIndexByTimestamp(int *index, std::vector <int64_t> ×tamps, int length) {
// Use Insert Sort Algorithm
if (length >= 2) {
for (int i = 1; i < length; i++) {
@@ -422,16 +423,16 @@ void Session::sortIndexByTimestamp(int* index, std::vector<int64_t>& timestamps,
/**
* Append value into buffer in Big Endian order to comply with IoTDB server
*/
-void Session::appendValues(string &buffer, char* value, int size) {
+void Session::appendValues(string &buffer, char *value, int size) {
for (int i = size - 1; i >= 0; i--) {
buffer.append(value + i, 1);
}
}
-void Session::putValuesIntoBuffer(vector<TSDataType::TSDataType>& types, vector<char*>& values, string& buf) {
+void Session::putValuesIntoBuffer(vector <TSDataType::TSDataType> &types, vector<char *> &values, string &buf) {
for (int i = 0; i < values.size(); i++) {
int8_t typeNum = getDataTypeNumber(types[i]);
- buf.append((char*)(&typeNum), sizeof(int8_t));
+ buf.append((char *) (&typeNum), sizeof(int8_t));
switch (types[i]) {
case TSDataType::BOOLEAN:
buf.append(values[i], 1);
@@ -451,7 +452,7 @@ void Session::putValuesIntoBuffer(vector<TSDataType::TSDataType>& types, vector<
case TSDataType::TEXT:
string str(values[i]);
int len = str.length();
- appendValues(buf, (char*)(&len), sizeof(int));
+ appendValues(buf, (char *) (&len), sizeof(int));
// no need to change the byte order of string value
buf.append(values[i], len);
break;
@@ -478,73 +479,61 @@ int8_t Session::getDataTypeNumber(TSDataType::TSDataType type) {
}
}
-void Session::open()
-{
- try
- {
+void Session::open() {
+ try {
open(false, DEFAULT_TIMEOUT_MS);
}
- catch (IoTDBConnectionException e)
- {
+ catch (IoTDBConnectionException e) {
throw IoTDBConnectionException(e.what());
}
}
void Session::open(bool enableRPCCompression) {
- try
- {
+ try {
open(enableRPCCompression, DEFAULT_TIMEOUT_MS);
}
- catch (IoTDBConnectionException e)
- {
+ catch (IoTDBConnectionException e) {
throw IoTDBConnectionException(e.what());
}
}
-void Session::open(bool enableRPCCompression, int connectionTimeoutInMs)
-{
- if (!isClosed)
- {
+void Session::open(bool enableRPCCompression, int connectionTimeoutInMs) {
+ if (!isClosed) {
return;
}
- shared_ptr<TSocket> socket(new TSocket(host, rpcPort));
- shared_ptr<TTransport> transport(new TFramedTransport(socket));
+ shared_ptr <TSocket> socket(new TSocket(host, rpcPort));
+ shared_ptr <TTransport> transport(new TFramedTransport(socket));
socket->setConnTimeout(connectionTimeoutInMs);
- if (!transport->isOpen())
- {
- try
- {
+ if (!transport->isOpen()) {
+ try {
transport->open();
- }
- catch (TTransportException e)
- {
+ }
+ catch (TTransportException e) {
throw IoTDBConnectionException(e.what());
}
}
- if (enableRPCCompression)
- {
- shared_ptr<TCompactProtocol> protocol(new TCompactProtocol(transport));
- shared_ptr<TSIServiceIf> client_instance(new TSIServiceClient(protocol));
+ if (enableRPCCompression) {
+ shared_ptr <TCompactProtocol> protocol(new TCompactProtocol(transport));
+ shared_ptr <TSIServiceIf> client_instance(new TSIServiceClient(protocol));
client = client_instance;
} else {
- shared_ptr<TBinaryProtocol> protocol(new TBinaryProtocol(transport));
- shared_ptr<TSIServiceIf> client_instance(new TSIServiceClient(protocol));
+ shared_ptr <TBinaryProtocol> protocol(new TBinaryProtocol(transport));
+ shared_ptr <TSIServiceIf> client_instance(new TSIServiceClient(protocol));
client = client_instance;
}
- shared_ptr<TSOpenSessionReq> openReq(new TSOpenSessionReq());
+ shared_ptr <TSOpenSessionReq> openReq(new TSOpenSessionReq());
openReq->__set_username(username);
openReq->__set_password(password);
openReq->__set_zoneId(zoneId);
- try
- {
- shared_ptr<TSOpenSessionResp> openResp(new TSOpenSessionResp());
- client->openSession(*openResp,*openReq);
+ try {
+ shared_ptr <TSOpenSessionResp> openResp(new TSOpenSessionResp());
+ client->openSession(*openResp, *openReq);
RpcUtils::verifySuccess(openResp->status);
- if (protocolVersion != openResp->serverProtocolVersion)
- {
+ if (protocolVersion != openResp->serverProtocolVersion) {
if (openResp->serverProtocolVersion == 0) {// less than 0.10
char buf[111];
- sprintf(buf, "Protocol not supported, Client version is %d, but Server version is %d", protocolVersion, openResp->serverProtocolVersion);
+ sprintf(buf, "Protocol not supported, Client version is %d, but Server version is %d", protocolVersion,
+ openResp->serverProtocolVersion);
logic_error e(buf);
throw exception(e);
}
@@ -552,15 +541,14 @@ void Session::open(bool enableRPCCompression, int connectionTimeoutInMs)
sessionId = openResp->sessionId;
statementId = client->requestStatementId(sessionId);
-
+
if (zoneId != "") {
setTimeZone(zoneId);
} else {
zoneId = getTimeZone();
}
}
- catch (exception e)
- {
+ catch (exception e) {
transport->close();
throw IoTDBConnectionException(e.what());
}
@@ -568,59 +556,48 @@ void Session::open(bool enableRPCCompression, int connectionTimeoutInMs)
}
-
-void Session::close()
-{
- if (isClosed)
- {
+void Session::close() {
+ if (isClosed) {
return;
}
- shared_ptr<TSCloseSessionReq> req(new TSCloseSessionReq());
+ shared_ptr <TSCloseSessionReq> req(new TSCloseSessionReq());
req->__set_sessionId(sessionId);
- try
- {
- shared_ptr<TSStatus> resp(new TSStatus());
- client->closeSession(*resp,*req);
+ try {
+ shared_ptr <TSStatus> resp(new TSStatus());
+ client->closeSession(*resp, *req);
}
- catch (exception e)
- {
+ catch (exception e) {
char buf[111];
- sprintf(buf,"Error occurs when closing session at server. Maybe server is down. %s",e.what());
+ sprintf(buf, "Error occurs when closing session at server. Maybe server is down. %s", e.what());
throw IoTDBConnectionException(buf);
- }
+ }
isClosed = true;
- if (transport != NULL)
- {
+ if (transport != NULL) {
transport->close();
}
}
-
-void Session::insertRecord(string deviceId, int64_t time, vector<string>& measurements, vector<string>& values)
-{
- shared_ptr<TSInsertStringRecordReq> req(new TSInsertStringRecordReq());
+void Session::insertRecord(string deviceId, int64_t time, vector <string> &measurements, vector <string> &values) {
+ shared_ptr <TSInsertStringRecordReq> req(new TSInsertStringRecordReq());
req->__set_sessionId(sessionId);
req->__set_deviceId(deviceId);
req->__set_timestamp(time);
req->__set_measurements(measurements);
req->__set_values(values);
- shared_ptr<TSStatus> resp(new TSStatus());
- try
- {
- client->insertStringRecord(*resp,*req);
+ shared_ptr <TSStatus> resp(new TSStatus());
+ try {
+ client->insertStringRecord(*resp, *req);
RpcUtils::verifySuccess(*resp);
}
- catch (IoTDBConnectionException& e)
- {
+ catch (IoTDBConnectionException &e) {
throw IoTDBConnectionException(e.what());
}
}
-void Session::insertRecord(string deviceId, int64_t time, vector<string>& measurements,
- vector<TSDataType::TSDataType>& types, vector<char*>& values)
-{
- shared_ptr<TSInsertRecordReq> req(new TSInsertRecordReq());
+void Session::insertRecord(string deviceId, int64_t time, vector <string> &measurements,
+ vector <TSDataType::TSDataType> &types, vector<char *> &values) {
+ shared_ptr <TSInsertRecordReq> req(new TSInsertRecordReq());
req->__set_sessionId(sessionId);
req->__set_deviceId(deviceId);
req->__set_timestamp(time);
@@ -628,54 +605,55 @@ void Session::insertRecord(string deviceId, int64_t time, vector<string>& measu
string buffer;
putValuesIntoBuffer(types, values, buffer);
req->__set_values(buffer);
- shared_ptr<TSStatus> resp(new TSStatus());
+ shared_ptr <TSStatus> resp(new TSStatus());
try {
- client->insertRecord(*resp,*req);
+ client->insertRecord(*resp, *req);
RpcUtils::verifySuccess(*resp);
- } catch (IoTDBConnectionException& e) {
+ } catch (IoTDBConnectionException &e) {
throw IoTDBConnectionException(e.what());
}
}
-void Session::insertRecords(vector<string>& deviceIds, vector<int64_t>& times, vector<vector<string>>& measurementsList, vector<vector<string>>& valuesList) {
+void
+Session::insertRecords(vector <string> &deviceIds, vector <int64_t> ×, vector <vector<string>> &measurementsList,
+ vector <vector<string>> &valuesList) {
int len = deviceIds.size();
if (len != times.size() || len != measurementsList.size() || len != valuesList.size()) {
logic_error e("deviceIds, times, measurementsList and valuesList's size should be equal");
throw exception(e);
}
- shared_ptr<TSInsertStringRecordsReq> request(new TSInsertStringRecordsReq());
+ shared_ptr <TSInsertStringRecordsReq> request(new TSInsertStringRecordsReq());
request->__set_sessionId(sessionId);
request->__set_deviceIds(deviceIds);
request->__set_timestamps(times);
request->__set_measurementsList(measurementsList);
request->__set_valuesList(valuesList);
- try
- {
- shared_ptr<TSStatus> resp(new TSStatus());
+ try {
+ shared_ptr <TSStatus> resp(new TSStatus());
client->insertStringRecords(*resp, *request);
RpcUtils::verifySuccess(*resp);
}
- catch (IoTDBConnectionException& e)
- {
+ catch (IoTDBConnectionException &e) {
throw IoTDBConnectionException(e.what());
}
}
-void Session::insertRecords(vector<string>& deviceIds, vector<int64_t>& times,
- vector<vector<string>>& measurementsList, vector<vector<TSDataType::TSDataType>> typesList,
- vector<vector<char*>>& valuesList) {
+void Session::insertRecords(vector <string> &deviceIds, vector <int64_t> ×,
+ vector <vector<string>> &measurementsList,
+ vector <vector<TSDataType::TSDataType>> typesList,
+ vector <vector<char *>> &valuesList) {
int len = deviceIds.size();
if (len != times.size() || len != measurementsList.size() || len != valuesList.size()) {
logic_error e("deviceIds, times, measurementsList and valuesList's size should be equal");
throw exception(e);
}
- shared_ptr<TSInsertRecordsReq> request(new TSInsertRecordsReq());
+ shared_ptr <TSInsertRecordsReq> request(new TSInsertRecordsReq());
request->__set_sessionId(sessionId);
request->__set_deviceIds(deviceIds);
request->__set_timestamps(times);
request->__set_measurementsList(measurementsList);
- vector<string> bufferList;
+ vector <string> bufferList;
for (int i = 0; i < valuesList.size(); i++) {
string buffer;
putValuesIntoBuffer(typesList[i], valuesList[i], buffer);
@@ -684,30 +662,32 @@ void Session::insertRecords(vector<string>& deviceIds, vector<int64_t>& times,
request->__set_valuesList(bufferList);
try {
- shared_ptr<TSStatus> resp(new TSStatus());
+ shared_ptr <TSStatus> resp(new TSStatus());
client->insertRecords(*resp, *request);
RpcUtils::verifySuccess(*resp);
- } catch (IoTDBConnectionException& e) {
+ } catch (IoTDBConnectionException &e) {
throw IoTDBConnectionException(e.what());
}
}
-void Session::insertRecordsOfOneDevice(string deviceId, vector<int64_t>& times,
- vector<vector<string>> measurementsList, vector<vector<TSDataType::TSDataType>> typesList,
- vector<vector<char*>>& valuesList) {
+void Session::insertRecordsOfOneDevice(string deviceId, vector <int64_t> ×,
+ vector <vector<string>> measurementsList,
+ vector <vector<TSDataType::TSDataType>> typesList,
+ vector <vector<char *>> &valuesList) {
insertRecordsOfOneDevice(deviceId, times, measurementsList, typesList, valuesList, false);
}
-void Session::insertRecordsOfOneDevice(string deviceId, vector<int64_t>& times,
- vector<vector<string>> measurementsList, vector<vector<TSDataType::TSDataType>> typesList,
- vector<vector<char*>>& valuesList, bool sorted) {
+void Session::insertRecordsOfOneDevice(string deviceId, vector <int64_t> ×,
+ vector <vector<string>> measurementsList,
+ vector <vector<TSDataType::TSDataType>> typesList,
+ vector <vector<char *>> &valuesList, bool sorted) {
if (sorted) {
if (!checkSorted(times)) {
throw BatchExecutionException("Times in InsertOneDeviceRecords are not in ascending order");
}
} else {
- int* index = new int[times.size()];
+ int *index = new int[times.size()];
for (int i = 0; i < times.size(); i++) {
index[i] = i;
}
@@ -719,12 +699,12 @@ void Session::insertRecordsOfOneDevice(string deviceId, vector<int64_t>& times,
valuesList = sortList(valuesList, index, times.size());
delete[] index;
}
- unique_ptr<TSInsertRecordsOfOneDeviceReq> request(new TSInsertRecordsOfOneDeviceReq());
+ unique_ptr <TSInsertRecordsOfOneDeviceReq> request(new TSInsertRecordsOfOneDeviceReq());
request->__set_sessionId(sessionId);
request->__set_deviceId(deviceId);
request->__set_timestamps(times);
request->__set_measurementsList(measurementsList);
- vector<string> bufferList;
+ vector <string> bufferList;
for (int i = 0; i < valuesList.size(); i++) {
string buffer;
putValuesIntoBuffer(typesList[i], valuesList[i], buffer);
@@ -733,27 +713,25 @@ void Session::insertRecordsOfOneDevice(string deviceId, vector<int64_t>& times,
request->__set_valuesList(bufferList);
try {
- unique_ptr<TSStatus> resp(new TSStatus());
+ unique_ptr <TSStatus> resp(new TSStatus());
client->insertRecordsOfOneDevice(*resp, *request);
RpcUtils::verifySuccess(*resp);
- } catch (const exception& e) {
+ } catch (const exception &e) {
throw IoTDBConnectionException(e.what());
}
}
-void Session::insertTablet(Tablet& tablet) {
- try
- {
+void Session::insertTablet(Tablet &tablet) {
+ try {
insertTablet(tablet, false);
}
- catch (const exception& e)
- {
+ catch (const exception &e) {
logic_error error(e.what());
throw exception(error);
}
}
-void Session::insertTablet(Tablet& tablet, bool sorted) {
+void Session::insertTablet(Tablet &tablet, bool sorted) {
if (sorted) {
if (!checkSorted(tablet)) {
throw BatchExecutionException("Times in Tablet are not in ascending order");
@@ -762,10 +740,10 @@ void Session::insertTablet(Tablet& tablet, bool sorted) {
sortTablet(tablet);
}
- shared_ptr<TSInsertTabletReq> request(new TSInsertTabletReq());
+ shared_ptr <TSInsertTabletReq> request(new TSInsertTabletReq());
request->__set_sessionId(sessionId);
request->deviceId = tablet.deviceId;
- for (pair<string, TSDataType::TSDataType> schema : tablet.schemas) {
+ for (pair <string, TSDataType::TSDataType> schema : tablet.schemas) {
request->measurements.push_back(schema.first);
request->types.push_back(schema.second);
}
@@ -773,32 +751,28 @@ void Session::insertTablet(Tablet& tablet, bool sorted) {
request->__set_values(SessionUtils::getValue(tablet));
request->__set_size(tablet.rowSize);
- try
- {
- shared_ptr<TSStatus> resp(new TSStatus());
+ try {
+ shared_ptr <TSStatus> resp(new TSStatus());
client->insertTablet(*resp, *request);
RpcUtils::verifySuccess(*resp);
}
- catch (IoTDBConnectionException& e)
- {
+ catch (IoTDBConnectionException &e) {
throw new IoTDBConnectionException(e.what());
}
}
-void Session::insertTablets(map<string, Tablet*>& tablets) {
- try
- {
+void Session::insertTablets(map<string, Tablet *> &tablets) {
+ try {
insertTablets(tablets, false);
}
- catch (const exception& e)
- {
+ catch (const exception &e) {
logic_error error(e.what());
throw exception(error);
}
}
-void Session::insertTablets(map<string, Tablet*>& tablets, bool sorted) {
- shared_ptr<TSInsertTabletsReq> request(new TSInsertTabletsReq());
+void Session::insertTablets(map<string, Tablet *> &tablets, bool sorted) {
+ shared_ptr <TSInsertTabletsReq> request(new TSInsertTabletsReq());
request->__set_sessionId(sessionId);
for (auto &item : tablets) {
@@ -811,9 +785,9 @@ void Session::insertTablets(map<string, Tablet*>& tablets, bool sorted) {
}
request->deviceIds.push_back(item.second->deviceId);
- vector<string> measurements;
+ vector <string> measurements;
vector<int> dataTypes;
- for (pair<string, TSDataType::TSDataType> schema : item.second->schemas) {
+ for (pair <string, TSDataType::TSDataType> schema : item.second->schemas) {
measurements.push_back(schema.first);
dataTypes.push_back(schema.second);
}
@@ -823,43 +797,39 @@ void Session::insertTablets(map<string, Tablet*>& tablets, bool sorted) {
request->valuesList.push_back(SessionUtils::getValue(*(item.second)));
request->sizeList.push_back(item.second->rowSize);
- try
- {
- shared_ptr<TSStatus> resp(new TSStatus());
+ try {
+ shared_ptr <TSStatus> resp(new TSStatus());
client->insertTablets(*resp, *request);
RpcUtils::verifySuccess(*resp);
}
- catch (const exception& e)
- {
+ catch (const exception &e) {
throw IoTDBConnectionException(e.what());
}
}
}
-void Session::testInsertRecord(string deviceId, int64_t time, vector<string>& measurements, vector<string>& values) {
- shared_ptr<TSInsertStringRecordReq> req(new TSInsertStringRecordReq());
+void Session::testInsertRecord(string deviceId, int64_t time, vector <string> &measurements, vector <string> &values) {
+ shared_ptr <TSInsertStringRecordReq> req(new TSInsertStringRecordReq());
req->__set_sessionId(sessionId);
req->__set_deviceId(deviceId);
req->__set_timestamp(time);
req->__set_measurements(measurements);
req->__set_values(values);
- shared_ptr<TSStatus> resp(new TSStatus());
- try
- {
+ shared_ptr <TSStatus> resp(new TSStatus());
+ try {
client->insertStringRecord(*resp, *req);
RpcUtils::verifySuccess(*resp);
}
- catch (IoTDBConnectionException e)
- {
+ catch (IoTDBConnectionException e) {
throw IoTDBConnectionException(e.what());
}
}
-void Session::testInsertTablet(Tablet& tablet) {
- shared_ptr<TSInsertTabletReq> request(new TSInsertTabletReq());
+void Session::testInsertTablet(Tablet &tablet) {
+ shared_ptr <TSInsertTabletReq> request(new TSInsertTabletReq());
request->__set_sessionId(sessionId);
request->deviceId = tablet.deviceId;
- for (pair<string, TSDataType::TSDataType> schema : tablet.schemas) {
+ for (pair <string, TSDataType::TSDataType> schema : tablet.schemas) {
request->measurements.push_back(schema.first);
request->types.push_back(schema.second);
}
@@ -867,139 +837,121 @@ void Session::testInsertTablet(Tablet& tablet) {
request->__set_values(SessionUtils::getValue(tablet));
request->__set_size(tablet.rowSize);
- try
- {
- shared_ptr<TSStatus> resp(new TSStatus());
+ try {
+ shared_ptr <TSStatus> resp(new TSStatus());
client->testInsertTablet(*resp, *request);
RpcUtils::verifySuccess(*resp);
}
- catch (IoTDBConnectionException& e)
- {
+ catch (IoTDBConnectionException &e) {
throw new IoTDBConnectionException(e.what());
}
}
-void Session::testInsertRecords(vector<string>& deviceIds, vector<int64_t>& times, vector<vector<string>>& measurementsList, vector<vector<string>>& valuesList) {
+void Session::testInsertRecords(vector <string> &deviceIds, vector <int64_t> ×,
+ vector <vector<string>> &measurementsList, vector <vector<string>> &valuesList) {
int len = deviceIds.size();
if (len != times.size() || len != measurementsList.size() || len != valuesList.size()) {
logic_error error("deviceIds, times, measurementsList and valuesList's size should be equal");
throw exception(error);
}
- shared_ptr<TSInsertStringRecordsReq> request(new TSInsertStringRecordsReq());
+ shared_ptr <TSInsertStringRecordsReq> request(new TSInsertStringRecordsReq());
request->__set_sessionId(sessionId);
request->__set_deviceIds(deviceIds);
request->__set_timestamps(times);
request->__set_measurementsList(measurementsList);
request->__set_valuesList(valuesList);
- try
- {
- shared_ptr<TSStatus> resp(new TSStatus());
+ try {
+ shared_ptr <TSStatus> resp(new TSStatus());
client->insertStringRecords(*resp, *request);
RpcUtils::verifySuccess(*resp);
}
- catch (IoTDBConnectionException& e)
- {
+ catch (IoTDBConnectionException &e) {
throw IoTDBConnectionException(e.what());
}
}
-void Session::deleteTimeseries(string path)
-{
- vector<string> paths;
+void Session::deleteTimeseries(string path) {
+ vector <string> paths;
paths.push_back(path);
deleteTimeseries(paths);
}
-void Session::deleteTimeseries(vector<string>& paths)
-{
- shared_ptr<TSStatus> resp(new TSStatus());
- try
- {
+void Session::deleteTimeseries(vector <string> &paths) {
+ shared_ptr <TSStatus> resp(new TSStatus());
+ try {
client->deleteTimeseries(*resp, sessionId, paths);
RpcUtils::verifySuccess(*resp);
}
- catch (IoTDBConnectionException& e)
- {
+ catch (IoTDBConnectionException &e) {
throw IoTDBConnectionException(e.what());
}
}
-void Session::deleteData(string path, int64_t time)
-{
- vector<string> paths;
+void Session::deleteData(string path, int64_t time) {
+ vector <string> paths;
paths.push_back(path);
deleteData(paths, time);
}
-void Session::deleteData(vector<string>& deviceId, int64_t time)
-{
- shared_ptr<TSDeleteDataReq> req(new TSDeleteDataReq());
+void Session::deleteData(vector <string> &deviceId, int64_t time) {
+ shared_ptr <TSDeleteDataReq> req(new TSDeleteDataReq());
req->__set_sessionId(sessionId);
req->__set_paths(deviceId);
req->__set_endTime(time);
- shared_ptr<TSStatus> resp(new TSStatus());
- try
- {
- client->deleteData(*resp,*req);
+ shared_ptr <TSStatus> resp(new TSStatus());
+ try {
+ client->deleteData(*resp, *req);
RpcUtils::verifySuccess(*resp);
}
- catch (exception& e)
- {
+ catch (exception &e) {
throw IoTDBConnectionException(e.what());
}
}
-void Session::setStorageGroup(string storageGroupId)
-{
- shared_ptr<TSStatus> resp(new TSStatus());
- try
- {
- client->setStorageGroup(*resp,sessionId, storageGroupId);
+void Session::setStorageGroup(string storageGroupId) {
+ shared_ptr <TSStatus> resp(new TSStatus());
+ try {
+ client->setStorageGroup(*resp, sessionId, storageGroupId);
RpcUtils::verifySuccess(*resp);
}
- catch (IoTDBConnectionException& e)
- {
+ catch (IoTDBConnectionException &e) {
throw IoTDBConnectionException(e.what());
}
}
-void Session::deleteStorageGroup(string storageGroup)
-{
- vector<string> storageGroups;
+void Session::deleteStorageGroup(string storageGroup) {
+ vector <string> storageGroups;
storageGroups.push_back(storageGroup);
deleteStorageGroups(storageGroups);
}
-void Session::deleteStorageGroups(vector<string>& storageGroups)
-{
- shared_ptr<TSStatus> resp(new TSStatus());
- try
- {
+void Session::deleteStorageGroups(vector <string> &storageGroups) {
+ shared_ptr <TSStatus> resp(new TSStatus());
+ try {
client->deleteStorageGroups(*resp, sessionId, storageGroups);
RpcUtils::verifySuccess(*resp);
}
- catch (IoTDBConnectionException& e)
- {
+ catch (IoTDBConnectionException &e) {
throw IoTDBConnectionException(e.what());
}
}
-void Session::createTimeseries(string path, TSDataType::TSDataType dataType, TSEncoding::TSEncoding encoding, CompressionType::CompressionType compressor) {
- try
- {
+void Session::createTimeseries(string path, TSDataType::TSDataType dataType, TSEncoding::TSEncoding encoding,
+ CompressionType::CompressionType compressor) {
+ try {
createTimeseries(path, dataType, encoding, compressor, NULL, NULL, NULL, "");
}
- catch (IoTDBConnectionException& e)
- {
+ catch (IoTDBConnectionException &e) {
throw IoTDBConnectionException(e.what());
}
}
-void Session::createTimeseries(string path, TSDataType::TSDataType dataType, TSEncoding::TSEncoding encoding, CompressionType::CompressionType compressor,
- map<string, string>* props, map<string, string>* tags, map<string, string>* attributes, string measurementAlias)
-{
- shared_ptr<TSCreateTimeseriesReq> req(new TSCreateTimeseriesReq());
+void Session::createTimeseries(string path, TSDataType::TSDataType dataType, TSEncoding::TSEncoding encoding,
+ CompressionType::CompressionType compressor,
+ map <string, string> *props, map <string, string> *tags,
+ map <string, string> *attributes, string measurementAlias) {
+ shared_ptr <TSCreateTimeseriesReq> req(new TSCreateTimeseriesReq());
req->__set_sessionId(sessionId);
req->__set_path(path);
req->__set_dataType(dataType);
@@ -1019,21 +971,23 @@ void Session::createTimeseries(string path, TSDataType::TSDataType dataType, TSE
req->__set_measurementAlias(measurementAlias);
}
- shared_ptr<TSStatus> resp(new TSStatus());
- try
- {
- client->createTimeseries(*resp,*req);
+ shared_ptr <TSStatus> resp(new TSStatus());
+ try {
+ client->createTimeseries(*resp, *req);
RpcUtils::verifySuccess(*resp);
}
- catch (IoTDBConnectionException& e)
- {
+ catch (IoTDBConnectionException &e) {
throw IoTDBConnectionException(e.what());
}
}
-void Session::createMultiTimeseries(vector<string> paths, vector<TSDataType::TSDataType> dataTypes, vector<TSEncoding::TSEncoding> encodings, vector<CompressionType::CompressionType> compressors,
- vector<map<string, string>>* propsList, vector<map<string, string>>* tagsList, vector<map<string, string>>* attributesList, vector<string>* measurementAliasList) {
- shared_ptr<TSCreateMultiTimeseriesReq> request(new TSCreateMultiTimeseriesReq());
+void Session::createMultiTimeseries(vector <string> paths, vector <TSDataType::TSDataType> dataTypes,
+ vector <TSEncoding::TSEncoding> encodings,
+ vector <CompressionType::CompressionType> compressors,
+ vector <map<string, string>> *propsList, vector <map<string, string>> *tagsList,
+ vector <map<string, string>> *attributesList,
+ vector <string> *measurementAliasList) {
+ shared_ptr <TSCreateMultiTimeseriesReq> request(new TSCreateMultiTimeseriesReq());
request->__set_sessionId(sessionId);
request->__set_paths(paths);
@@ -1069,14 +1023,12 @@ void Session::createMultiTimeseries(vector<string> paths, vector<TSDataType::TSD
request->__set_measurementAliasList(*measurementAliasList);
}
- try
- {
- shared_ptr<TSStatus> resp(new TSStatus());
+ try {
+ shared_ptr <TSStatus> resp(new TSStatus());
client->createMultiTimeseries(*resp, *request);
RpcUtils::verifySuccess(*resp);
}
- catch (IoTDBConnectionException& e)
- {
+ catch (IoTDBConnectionException &e) {
throw IoTDBConnectionException(e.what());
}
}
@@ -1091,79 +1043,66 @@ bool Session::checkTimeseriesExists(string path) {
}
}
-string Session::getTimeZone()
-{
- if (zoneId != "")
- {
+string Session::getTimeZone() {
+ if (zoneId != "") {
return zoneId;
}
- shared_ptr<TSGetTimeZoneResp> resp(new TSGetTimeZoneResp());
- try
- {
+ shared_ptr <TSGetTimeZoneResp> resp(new TSGetTimeZoneResp());
+ try {
client->getTimeZone(*resp, sessionId);
RpcUtils::verifySuccess(resp->status);
}
- catch (IoTDBConnectionException& e)
- {
+ catch (IoTDBConnectionException &e) {
throw IoTDBConnectionException(e.what());
}
return resp->timeZone;
}
-void Session::setTimeZone(string zoneId)
-{
- shared_ptr<TSSetTimeZoneReq> req(new TSSetTimeZoneReq());
+void Session::setTimeZone(string zoneId) {
+ shared_ptr <TSSetTimeZoneReq> req(new TSSetTimeZoneReq());
req->__set_sessionId(sessionId);
req->__set_timeZone(zoneId);
- shared_ptr<TSStatus> resp(new TSStatus());
- try
- {
- client->setTimeZone(*resp,*req);
+ shared_ptr <TSStatus> resp(new TSStatus());
+ try {
+ client->setTimeZone(*resp, *req);
}
- catch (IoTDBConnectionException& e)
- {
+ catch (IoTDBConnectionException &e) {
throw IoTDBConnectionException(e.what());
}
RpcUtils::verifySuccess(*resp);
this->zoneId = zoneId;
}
-unique_ptr<SessionDataSet> Session::executeQueryStatement(string sql)
-{
- shared_ptr<TSExecuteStatementReq> req(new TSExecuteStatementReq());
+unique_ptr <SessionDataSet> Session::executeQueryStatement(string sql) {
+ shared_ptr <TSExecuteStatementReq> req(new TSExecuteStatementReq());
req->__set_sessionId(sessionId);
req->__set_statementId(statementId);
req->__set_statement(sql);
req->__set_fetchSize(fetchSize);
- shared_ptr<TSExecuteStatementResp> resp(new TSExecuteStatementResp());
- try
- {
- client->executeStatement(*resp,*req);
+ shared_ptr <TSExecuteStatementResp> resp(new TSExecuteStatementResp());
+ try {
+ client->executeStatement(*resp, *req);
RpcUtils::verifySuccess(resp->status);
}
- catch (IoTDBConnectionException e)
- {
+ catch (IoTDBConnectionException e) {
throw IoTDBConnectionException(e.what());
}
- shared_ptr<TSQueryDataSet> queryDataSet(new TSQueryDataSet(resp->queryDataSet));
+ shared_ptr <TSQueryDataSet> queryDataSet(new TSQueryDataSet(resp->queryDataSet));
return unique_ptr<SessionDataSet>(new SessionDataSet(
- sql, resp->columns, resp->dataTypeList, resp->queryId, statementId, client, sessionId, queryDataSet));
+ sql, resp->columns, resp->dataTypeList, resp->queryId, statementId, client, sessionId, queryDataSet));
}
-void Session::executeNonQueryStatement(string sql)
-{
- shared_ptr<TSExecuteStatementReq> req(new TSExecuteStatementReq());
+void Session::executeNonQueryStatement(string sql) {
+ shared_ptr <TSExecuteStatementReq> req(new TSExecuteStatementReq());
req->__set_sessionId(sessionId);
req->__set_statementId(statementId);
req->__set_statement(sql);
- shared_ptr<TSExecuteStatementResp> resp(new TSExecuteStatementResp());
- try
- {
- client->executeUpdateStatement(*resp,*req);
+ shared_ptr <TSExecuteStatementResp> resp(new TSExecuteStatementResp());
+ try {
+ client->executeUpdateStatement(*resp, *req);
RpcUtils::verifySuccess(resp->status);
}
- catch (IoTDBConnectionException e)
- {
+ catch (IoTDBConnectionException e) {
throw IoTDBConnectionException(e.what());
}
}
diff --git a/client-cpp/src/main/Session.h b/client-cpp/src/main/Session.h
index f2c0ea7..00021a7 100644
--- a/client-cpp/src/main/Session.h
+++ b/client-cpp/src/main/Session.h
@@ -46,59 +46,63 @@ using ::apache::thrift::transport::TBufferedTransport;
using ::apache::thrift::transport::TFramedTransport;
using ::apache::thrift::TException;
-class IoTDBConnectionException : public std::exception
-{
- public:
- IoTDBConnectionException() : message() {}
- IoTDBConnectionException(const char* m) : message(m) {}
- IoTDBConnectionException(std::string m) : message(m) {}
- virtual const char* what() const throw ()
- {
- return message.c_str();
- }
+class IoTDBConnectionException : public std::exception {
+public:
+ IoTDBConnectionException() : message() {}
+
+ IoTDBConnectionException(const char *m) : message(m) {}
+
+ IoTDBConnectionException(std::string m) : message(m) {}
+
+ virtual const char *what() const throw() {
+ return message.c_str();
+ }
- private:
- std::string message;
+private:
+ std::string message;
};
-class BatchExecutionException : public std::exception
-{
+class BatchExecutionException : public std::exception {
public:
BatchExecutionException() : message() {}
- BatchExecutionException(const char* m) : message(m) {}
+
+ BatchExecutionException(const char *m) : message(m) {}
+
BatchExecutionException(std::string m) : message(m) {}
- BatchExecutionException(std::vector<TSStatus> statusList) : message(), statusList(statusList) {}
- BatchExecutionException(std::vector<TSStatus> statusList, std::string m) : message(m), statusList(statusList) {}
- virtual const char* what() const throw ()
- {
+
+ BatchExecutionException(std::vector <TSStatus> statusList) : message(), statusList(statusList) {}
+
+ BatchExecutionException(std::vector <TSStatus> statusList, std::string m) : message(m), statusList(statusList) {}
+
+ virtual const char *what() const throw() {
return message.c_str();
}
- std::vector<TSStatus> statusList;
+
+ std::vector <TSStatus> statusList;
private:
std::string message;
};
-class UnSupportedDataTypeException : public std::exception
-{
+class UnSupportedDataTypeException : public std::exception {
private:
std::string message;
public:
UnSupportedDataTypeException() : message() {}
- UnSupportedDataTypeException(const char* m) : message(m) {}
+
+ UnSupportedDataTypeException(const char *m) : message(m) {}
+
UnSupportedDataTypeException(std::string m) : message("UnSupported dataType: " + m) {}
};
-namespace CompressionType{
+namespace CompressionType {
- enum CompressionType
- {
+ enum CompressionType {
UNCOMPRESSED, SNAPPY, GZIP, LZO, SDT, PAA, PLA, LZ4
};
}
-namespace TSDataType{
- enum TSDataType
- {
+namespace TSDataType {
+ enum TSDataType {
BOOLEAN, INT32, INT64, FLOAT, DOUBLE, TEXT, NULLTYPE
};
}
@@ -170,48 +174,59 @@ namespace TSStatusCode {
TIME_OUT = 701,
NO_LEADER = 702,
UNSUPPORTED_OPERATION = 703,
- NODE_READ_ONLY= 704,
+ NODE_READ_ONLY = 704,
INCOMPATIBLE_VERSION = 203,
};
}
-class RpcUtils
-{
+class RpcUtils {
public:
- std::shared_ptr<TSStatus> SUCCESS_STATUS;
+ std::shared_ptr <TSStatus> SUCCESS_STATUS;
+
RpcUtils() {
SUCCESS_STATUS = std::make_shared<TSStatus>();
SUCCESS_STATUS->__set_code(TSStatusCode::SUCCESS_STATUS);
}
- static void verifySuccess(TSStatus& status);
- static void verifySuccess(std::vector<TSStatus>& statuses);
+
+ static void verifySuccess(TSStatus &status);
+
+ static void verifySuccess(std::vector <TSStatus> &statuses);
+
static TSStatus getStatus(TSStatusCode::TSStatusCode tsStatusCode);
+
static TSStatus getStatus(int code, std::string message);
- static std::shared_ptr<TSExecuteStatementResp> getTSExecuteStatementResp(TSStatusCode::TSStatusCode tsStatusCode);
- static std::shared_ptr<TSExecuteStatementResp> getTSExecuteStatementResp(TSStatusCode::TSStatusCode tsStatusCode, std::string message);
- static std::shared_ptr<TSExecuteStatementResp> getTSExecuteStatementResp(TSStatus& status);
- static std::shared_ptr<TSFetchResultsResp> getTSFetchResultsResp(TSStatusCode::TSStatusCode tsStatusCode);
- static std::shared_ptr<TSFetchResultsResp> getTSFetchResultsResp(TSStatusCode::TSStatusCode tsStatusCode, std::string appendMessage);
- static std::shared_ptr<TSFetchResultsResp> getTSFetchResultsResp(TSStatus& status);
+
+ static std::shared_ptr <TSExecuteStatementResp> getTSExecuteStatementResp(TSStatusCode::TSStatusCode tsStatusCode);
+
+ static std::shared_ptr <TSExecuteStatementResp>
+ getTSExecuteStatementResp(TSStatusCode::TSStatusCode tsStatusCode, std::string message);
+
+ static std::shared_ptr <TSExecuteStatementResp> getTSExecuteStatementResp(TSStatus &status);
+
+ static std::shared_ptr <TSFetchResultsResp> getTSFetchResultsResp(TSStatusCode::TSStatusCode tsStatusCode);
+
+ static std::shared_ptr <TSFetchResultsResp>
+ getTSFetchResultsResp(TSStatusCode::TSStatusCode tsStatusCode, std::string appendMessage);
+
+ static std::shared_ptr <TSFetchResultsResp> getTSFetchResultsResp(TSStatus &status);
};
// Simulate the ByteBuffer class in Java
class MyStringBuffer {
private:
- char* getchar(int len)
- {
- char* ret = new char[len];
+ char *getchar(int len) {
+ char *ret = new char[len];
for (int i = pos; i < pos + len; i++)
ret[pos + len - 1 - i] = str[i];
pos += len;
return ret;
}
- void putchar(int len, char* ins)
- {
+ void putchar(int len, char *ins) {
for (int i = len - 1; i > -1; i--)
str += ins[i];
}
+
public:
std::string str;
int pos;
@@ -232,107 +247,92 @@ public:
// return (byte)tmpChar;
//}
- int getInt()
- {
- char* data = getchar(4);
- int ret = *(int*)data;
+ int getInt() {
+ char *data = getchar(4);
+ int ret = *(int *) data;
delete[]data;
return ret;
}
- int64_t getLong()
- {
- char* data = getchar(8);
- int64_t ret = *(int64_t*)data;
+ int64_t getLong() {
+ char *data = getchar(8);
+ int64_t ret = *(int64_t *) data;
delete[]data;
return ret;
}
- float getFloat()
- {
- char* data = getchar(4);
- float ret = *(float*)data;
+ float getFloat() {
+ char *data = getchar(4);
+ float ret = *(float *) data;
delete[]data;
return ret;
}
- double getDouble()
- {
- char* data = getchar(8);
- double ret = *(double*)data;
+ double getDouble() {
+ char *data = getchar(8);
+ double ret = *(double *) data;
delete[]data;
return ret;
}
- char getChar()
- {
- char* data = getchar(1);
- char ret = *(char*)data;
+ char getChar() {
+ char *data = getchar(1);
+ char ret = *(char *) data;
delete[]data;
return ret;
}
- bool getBool()
- {
+ bool getBool() {
char bo = getChar();
return bo == 1;
}
- std::string getString()
- {
+ std::string getString() {
int len = getInt();
std::string ret;
for (int i = 0; i < len; i++) ret.append(1, getChar());
return ret;
}
- void putInt(int ins)
- {
- char* data = (char*)&ins;
+ void putInt(int ins) {
+ char *data = (char *) &ins;
putchar(4, data);
}
- void putLong(int64_t ins)
- {
- char* data = (char*)&ins;
+ void putLong(int64_t ins) {
+ char *data = (char *) &ins;
putchar(8, data);
}
- void putFloat(float ins)
- {
- char* data = (char*)&ins;
+ void putFloat(float ins) {
+ char *data = (char *) &ins;
putchar(4, data);
}
- void putDouble(double ins)
- {
- char* data = (char*)&ins;
+ void putDouble(double ins) {
+ char *data = (char *) &ins;
putchar(8, data);
}
- void putChar(char ins)
- {
- char* data = (char*)&ins;
+ void putChar(char ins) {
+ char *data = (char *) &ins;
putchar(1, data);
}
- void putBool(bool ins)
- {
+ void putBool(bool ins) {
char tmp = 0;
if (ins) tmp = 1;
putChar(tmp);
}
- void putString(std::string ins)
- {
+ void putString(std::string ins) {
int len = ins.size();
putInt(len);
for (int i = 0; i < len; i++) putChar(ins[i]);
}
};
-class Field
-{
+class Field {
public:
TSDataType::TSDataType dataType;
bool boolV;
@@ -341,11 +341,12 @@ public:
float floatV;
double doubleV;
std::string stringV;
- Field(TSDataType::TSDataType a)
- {
+
+ Field(TSDataType::TSDataType a) {
dataType = a;
}
- Field(){}
+
+ Field() {}
};
/*
@@ -367,13 +368,14 @@ private:
static const int DEFAULT_SIZE = 1024;
public:
std::string deviceId; // deviceId of this tablet
- std::vector<std::pair<std::string, TSDataType::TSDataType>> schemas; // the list of measurement schemas for creating the tablet
+ std::vector <std::pair<std::string, TSDataType::TSDataType>> schemas; // the list of measurement schemas for creating the tablet
std::vector <int64_t> timestamps; //timestamps in this tablet
- std::vector<std::vector<std::string>> values;
+ std::vector <std::vector<std::string>> values;
int rowSize; //the number of rows to include in this tablet
int maxRowNumber; // the maximum number of rows for this tablet
- Tablet(){}
+ Tablet() {}
+
/**
* Return a tablet with default specified row number. This is the standard
* constructor (all Tablet should be the same size).
@@ -381,7 +383,7 @@ public:
* @param deviceId the name of the device specified to be written in
* @param timeseries the list of measurement schemas for creating the tablet
*/
- Tablet(std::string deviceId, std::vector<std::pair<std::string, TSDataType::TSDataType>>& timeseries) {
+ Tablet(std::string deviceId, std::vector <std::pair<std::string, TSDataType::TSDataType>> ×eries) {
Tablet(deviceId, timeseries, DEFAULT_SIZE);
}
@@ -395,7 +397,8 @@ public:
* batch
* @param maxRowNumber the maximum number of rows for this tablet
*/
- Tablet(std::string deviceId, std::vector<std::pair<std::string, TSDataType::TSDataType>>& schemas, int maxRowNumber) {
+ Tablet(std::string deviceId, std::vector <std::pair<std::string, TSDataType::TSDataType>> &schemas,
+ int maxRowNumber) {
this->deviceId = deviceId;
this->schemas = schemas;
this->maxRowNumber = maxRowNumber;
@@ -413,31 +416,34 @@ public:
void reset(); // Reset Tablet to the default state - set the rowSize to 0
void createColumns();
+
int getTimeBytesSize();
+
int getValueByteSize(); // total byte size that values occupies
};
class SessionUtils {
public:
- static std::string getTime(Tablet& tablet);
- static std::string getValue(Tablet& tablet);
+ static std::string getTime(Tablet &tablet);
+
+ static std::string getValue(Tablet &tablet);
};
-class RowRecord
-{
+class RowRecord {
public:
int64_t timestamp;
- std::vector<Field> fields;
- RowRecord(int64_t timestamp)
- {
+ std::vector <Field> fields;
+
+ RowRecord(int64_t timestamp) {
this->timestamp = timestamp;
}
- RowRecord(int64_t timestamp, std::vector<Field> &fields) {
+
+ RowRecord(int64_t timestamp, std::vector <Field> &fields) {
this->timestamp = timestamp;
this->fields = fields;
}
- RowRecord()
- {
+
+ RowRecord() {
this->timestamp = -1;
}
@@ -445,43 +451,40 @@ public:
this->fields.push_back(f);
}
- std::string toString()
- {
+ std::string toString() {
char buf[111];
- sprintf(buf,"%lld",timestamp);
+ sprintf(buf, "%lld", timestamp);
std::string ret = buf;
- for (int i = 0; i < fields.size(); i++)
- {
+ for (int i = 0; i < fields.size(); i++) {
ret.append("\t");
TSDataType::TSDataType dataType = fields[i].dataType;
- switch (dataType)
- {
- case TSDataType::BOOLEAN:{
+ switch (dataType) {
+ case TSDataType::BOOLEAN: {
if (fields[i].boolV) ret.append("true");
else ret.append("false");
break;
}
- case TSDataType::INT32:{
+ case TSDataType::INT32: {
char buf[111];
- sprintf(buf,"%d",fields[i].intV);
+ sprintf(buf, "%d", fields[i].intV);
ret.append(buf);
break;
}
case TSDataType::INT64: {
char buf[111];
- sprintf(buf,"%lld",fields[i].longV);
+ sprintf(buf, "%lld", fields[i].longV);
ret.append(buf);
break;
}
- case TSDataType::FLOAT:{
+ case TSDataType::FLOAT: {
char buf[111];
- sprintf(buf,"%f",fields[i].floatV);
+ sprintf(buf, "%f", fields[i].floatV);
ret.append(buf);
break;
}
- case TSDataType::DOUBLE:{
+ case TSDataType::DOUBLE: {
char buf[111];
- sprintf(buf,"%lf",fields[i].doubleV);
+ sprintf(buf, "%lf", fields[i].doubleV);
ret.append(buf);
break;
}
@@ -489,7 +492,7 @@ public:
ret.append(fields[i].stringV);
break;
}
- case TSDataType::NULLTYPE:{
+ case TSDataType::NULLTYPE: {
ret.append("NULL");
}
}
@@ -499,18 +502,17 @@ public:
}
};
-class SessionDataSet
-{
+class SessionDataSet {
private:
bool hasCachedRecord = false;
std::string sql;
int64_t queryId;
int64_t statementId;
int64_t sessionId;
- std::shared_ptr<TSIServiceIf> client;
+ std::shared_ptr <TSIServiceIf> client;
int batchSize = 1024;
- std::vector<std::string> columnNameList;
- std::vector<std::string> columnTypeDeduplicatedList;
+ std::vector <std::string> columnNameList;
+ std::vector <std::string> columnTypeDeduplicatedList;
// duplicated column index -> origin index
std::map<int, int> duplicateLocation;
// column name -> column location
@@ -519,19 +521,21 @@ private:
int columnSize = 0;
int rowsIndex = 0; // used to record the row index in current TSQueryDataSet
- std::shared_ptr<TSQueryDataSet> tsQueryDataSet;
+ std::shared_ptr <TSQueryDataSet> tsQueryDataSet;
MyStringBuffer tsQueryDataSetTimeBuffer;
- std::vector<std::unique_ptr<MyStringBuffer>> valueBuffers;
- std::vector<std::unique_ptr<MyStringBuffer>> bitmapBuffers;
+ std::vector <std::unique_ptr<MyStringBuffer>> valueBuffers;
+ std::vector <std::unique_ptr<MyStringBuffer>> bitmapBuffers;
RowRecord rowRecord;
- char* currentBitmap; // used to cache the current bitmap for every column
+ char *currentBitmap; // used to cache the current bitmap for every column
static const int flag = 0x80; // used to do `or` operation with bitmap to judge whether the value is null
public:
- SessionDataSet(){}
- SessionDataSet(std::string sql, std::vector<std::string>& columnNameList, std::vector<std::string>& columnTypeList, int64_t queryId, int64_t statementId,
- std::shared_ptr<TSIServiceIf> client, int64_t sessionId, std::shared_ptr<TSQueryDataSet> queryDataSet) : tsQueryDataSetTimeBuffer(queryDataSet->time)
- {
+ SessionDataSet() {}
+
+ SessionDataSet(std::string sql, std::vector <std::string> &columnNameList,
+ std::vector <std::string> &columnTypeList, int64_t queryId, int64_t statementId,
+ std::shared_ptr <TSIServiceIf> client, int64_t sessionId,
+ std::shared_ptr <TSQueryDataSet> queryDataSet) : tsQueryDataSetTimeBuffer(queryDataSet->time) {
this->sessionId = sessionId;
this->sql = sql;
this->queryId = queryId;
@@ -550,122 +554,193 @@ public:
this->columnMap[name] = i;
this->columnTypeDeduplicatedList.push_back(columnTypeList[i]);
}
- this->valueBuffers.push_back(std::unique_ptr<MyStringBuffer>(new MyStringBuffer(queryDataSet->valueList[i])));
- this->bitmapBuffers.push_back(std::unique_ptr<MyStringBuffer>(new MyStringBuffer(queryDataSet->bitmapList[i])));
+ this->valueBuffers.push_back(
+ std::unique_ptr<MyStringBuffer>(new MyStringBuffer(queryDataSet->valueList[i])));
+ this->bitmapBuffers.push_back(
+ std::unique_ptr<MyStringBuffer>(new MyStringBuffer(queryDataSet->bitmapList[i])));
}
this->tsQueryDataSet = queryDataSet;
}
int getBatchSize();
+
void setBatchSize(int batchSize);
- std::vector<std::string> getColumnNames();
+
+ std::vector <std::string> getColumnNames();
+
bool hasNext();
+
void constructOneRow();
+
bool isNull(int index, int rowNum);
- RowRecord* next();
+
+ RowRecord *next();
+
void closeOperationHandle();
};
template<typename T>
-std::vector<T> sortList(std::vector<T>& valueList, int* index, int indexLength) {
- std::vector<T> sortedValues(valueList.size());
+std::vector <T> sortList(std::vector <T> &valueList, int *index, int indexLength) {
+ std::vector <T> sortedValues(valueList.size());
for (int i = 0; i < indexLength; i++) {
sortedValues[i] = valueList[index[i]];
}
return sortedValues;
}
-class Session
-{
- private:
- std::string host;
- int rpcPort;
- std::string username;
- std::string password;
- TSProtocolVersion::type protocolVersion = TSProtocolVersion::IOTDB_SERVICE_PROTOCOL_V3;
- std::shared_ptr<TSIServiceIf> client;
- std::shared_ptr<apache::thrift::transport::TSocket> transport;
- bool isClosed = true;
- int64_t sessionId;
- int64_t statementId;
- std::string zoneId;
- int fetchSize;
- const static int DEFAULT_FETCH_SIZE = 10000;
- const static int DEFAULT_TIMEOUT_MS = 0;
-
- bool checkSorted(Tablet& tablet);
- bool checkSorted(std::vector<int64_t>& times);
- void sortTablet(Tablet& tablet);
- void sortIndexByTimestamp(int *index, std::vector<int64_t>& timestamps, int length);
- std::string getTimeZone();
- void setTimeZone(std::string zoneId);
- void appendValues(std::string &buffer, char* value, int size);
- void putValuesIntoBuffer(std::vector<TSDataType::TSDataType>& types, std::vector<char*>& values, std::string& buf);
- int8_t getDataTypeNumber(TSDataType::TSDataType type);
- public:
- Session(std::string host, int rpcPort) : username("user"), password("password") {
- this->host = host;
- this->rpcPort = rpcPort;
- }
+class Session {
+private:
+ std::string host;
+ int rpcPort;
+ std::string username;
+ std::string password;
+ TSProtocolVersion::type protocolVersion = TSProtocolVersion::IOTDB_SERVICE_PROTOCOL_V3;
+ std::shared_ptr <TSIServiceIf> client;
+ std::shared_ptr <apache::thrift::transport::TSocket> transport;
+ bool isClosed = true;
+ int64_t sessionId;
+ int64_t statementId;
+ std::string zoneId;
+ int fetchSize;
+ const static int DEFAULT_FETCH_SIZE = 10000;
+ const static int DEFAULT_TIMEOUT_MS = 0;
- Session(std::string host, int rpcPort, std::string username, std::string password)
- : fetchSize(10000) {
- this->host = host;
- this->rpcPort = rpcPort;
- this->username = username;
- this->password = password;
- this->zoneId = "UTC+08:00";
- }
+ bool checkSorted(Tablet &tablet);
- Session(std::string host, int rpcPort, std::string username, std::string password, int fetchSize) {
- this->host = host;
- this->rpcPort = rpcPort;
- this->username = username;
- this->password = password;
- this->fetchSize = fetchSize;
- this->zoneId = "UTC+08:00";
- }
+ bool checkSorted(std::vector <int64_t> ×);
- Session(std::string host, std::string rpcPort, std::string username = "user",
- std::string password = "password", int fetchSize = 10000) {
- this->host = host;
- this->rpcPort = stoi(rpcPort);
- this->username = username;
- this->password = password;
- this->fetchSize = fetchSize;
- this->zoneId = "UTC+08:00";
- }
+ void sortTablet(Tablet &tablet);
+
+ void sortIndexByTimestamp(int *index, std::vector <int64_t> ×tamps, int length);
+
+ std::string getTimeZone();
+
+ void setTimeZone(std::string zoneId);
+
+ void appendValues(std::string &buffer, char *value, int size);
+
+ void
+ putValuesIntoBuffer(std::vector <TSDataType::TSDataType> &types, std::vector<char *> &values, std::string &buf);
+
+ int8_t getDataTypeNumber(TSDataType::TSDataType type);
+
+public:
+ Session(std::string host, int rpcPort) : username("user"), password("password") {
+ this->host = host;
+ this->rpcPort = rpcPort;
+ }
+
+ Session(std::string host, int rpcPort, std::string username, std::string password)
+ : fetchSize(10000) {
+ this->host = host;
+ this->rpcPort = rpcPort;
+ this->username = username;
+ this->password = password;
+ this->zoneId = "UTC+08:00";
+ }
+
+ Session(std::string host, int rpcPort, std::string username, std::string password, int fetchSize) {
+ this->host = host;
+ this->rpcPort = rpcPort;
+ this->username = username;
+ this->password = password;
+ this->fetchSize = fetchSize;
+ this->zoneId = "UTC+08:00";
+ }
+
+ Session(std::string host, std::string rpcPort, std::string username = "user",
+ std::string password = "password", int fetchSize = 10000) {
+ this->host = host;
+ this->rpcPort = stoi(rpcPort);
+ this->username = username;
+ this->password = password;
+ this->fetchSize = fetchSize;
+ this->zoneId = "UTC+08:00";
+ }
+
+ void open();
+
+ void open(bool enableRPCCompression);
+
+ void open(bool enableRPCCompression, int connectionTimeoutInMs);
+
+ void close();
+
+ void insertRecord(std::string deviceId, int64_t time, std::vector <std::string> &measurements,
+ std::vector <std::string> &values);
+
+ void insertRecord(std::string deviceId, int64_t time, std::vector <std::string> &measurements,
+ std::vector <TSDataType::TSDataType> &types, std::vector<char *> &values);
+
+ void insertRecords(std::vector <std::string> &deviceIds, std::vector <int64_t> ×,
+ std::vector <std::vector<std::string>> &measurementsList,
+ std::vector <std::vector<std::string>> &valuesList);
+
+ void insertRecords(std::vector <std::string> &deviceIds, std::vector <int64_t> ×,
+ std::vector <std::vector<std::string>> &measurementsList,
+ std::vector <std::vector<TSDataType::TSDataType>> typesList,
+ std::vector <std::vector<char *>> &valuesList);
+
+ void insertRecordsOfOneDevice(std::string deviceId, std::vector <int64_t> ×,
+ std::vector <std::vector<std::string>> measurementsList,
+ std::vector <std::vector<TSDataType::TSDataType>> typesList,
+ std::vector <std::vector<char *>> &valuesList);
+
+ void insertRecordsOfOneDevice(std::string deviceId, std::vector <int64_t> ×,
+ std::vector <std::vector<std::string>> measurementsList,
+ std::vector <std::vector<TSDataType::TSDataType>> typesList,
+ std::vector <std::vector<char *>> &valuesList, bool sorted);
+
+ void insertTablet(Tablet &tablet);
+
+ void insertTablet(Tablet &tablet, bool sorted);
+
+ void insertTablets(std::map<std::string, Tablet *> &tablets);
+
+ void insertTablets(std::map<std::string, Tablet *> &tablets, bool sorted);
+
+ void testInsertRecord(std::string deviceId, int64_t time, std::vector <std::string> &measurements,
+ std::vector <std::string> &values);
+
+ void testInsertTablet(Tablet &tablet);
+
+ void testInsertRecords(std::vector <std::string> &deviceIds, std::vector <int64_t> ×,
+ std::vector <std::vector<std::string>> &measurementsList,
+ std::vector <std::vector<std::string>> &valuesList);
+
+ void deleteTimeseries(std::string path);
+
+ void deleteTimeseries(std::vector <std::string> &paths);
+
+ void deleteData(std::string path, int64_t time);
+
+ void deleteData(std::vector <std::string> &deviceId, int64_t time);
+
+ void setStorageGroup(std::string storageGroupId);
+
+ void deleteStorageGroup(std::string storageGroup);
+
+ void deleteStorageGroups(std::vector <std::string> &storageGroups);
+
+ void createTimeseries(std::string path, TSDataType::TSDataType dataType, TSEncoding::TSEncoding encoding,
+ CompressionType::CompressionType compressor);
+
+ void createTimeseries(std::string path, TSDataType::TSDataType dataType, TSEncoding::TSEncoding encoding,
+ CompressionType::CompressionType compressor,
+ std::map <std::string, std::string> *props, std::map <std::string, std::string> *tags,
+ std::map <std::string, std::string> *attributes, std::string measurementAlias);
+
+ void createMultiTimeseries(std::vector <std::string> paths, std::vector <TSDataType::TSDataType> dataTypes,
+ std::vector <TSEncoding::TSEncoding> encodings,
+ std::vector <CompressionType::CompressionType> compressors,
+ std::vector <std::map<std::string, std::string>> *propsList,
+ std::vector <std::map<std::string, std::string>> *tagsList,
+ std::vector <std::map<std::string, std::string>> *attributesList,
+ std::vector <std::string> *measurementAliasList);
+
+ bool checkTimeseriesExists(std::string path);
+
+ std::unique_ptr <SessionDataSet> executeQueryStatement(std::string sql);
- void open();
- void open(bool enableRPCCompression);
- void open(bool enableRPCCompression, int connectionTimeoutInMs);
- void close();
- void insertRecord(std::string deviceId, int64_t time, std::vector<std::string>& measurements, std::vector<std::string>& values);
- void insertRecord(std::string deviceId, int64_t time, std::vector<std::string>& measurements, std::vector<TSDataType::TSDataType>& types, std::vector<char*>& values);
- void insertRecords(std::vector<std::string>& deviceIds, std::vector<int64_t>& times, std::vector<std::vector<std::string>>& measurementsList, std::vector<std::vector<std::string>>& valuesList);
- void insertRecords(std::vector<std::string>& deviceIds, std::vector<int64_t>& times, std::vector<std::vector<std::string>>& measurementsList, std::vector<std::vector<TSDataType::TSDataType>> typesList, std::vector<std::vector<char*>>& valuesList);
- void insertRecordsOfOneDevice(std::string deviceId, std::vector<int64_t>& times, std::vector<std::vector<std::string>> measurementsList, std::vector<std::vector<TSDataType::TSDataType>> typesList, std::vector<std::vector<char*>>& valuesList);
- void insertRecordsOfOneDevice(std::string deviceId, std::vector<int64_t>& times, std::vector<std::vector<std::string>> measurementsList, std::vector<std::vector<TSDataType::TSDataType>> typesList, std::vector<std::vector<char*>>& valuesList, bool sorted);
- void insertTablet(Tablet& tablet);
- void insertTablet(Tablet& tablet, bool sorted);
- void insertTablets(std::map<std::string, Tablet*>& tablets);
- void insertTablets(std::map<std::string, Tablet*>& tablets, bool sorted);
- void testInsertRecord(std::string deviceId, int64_t time, std::vector<std::string>& measurements, std::vector<std::string>& values);
- void testInsertTablet(Tablet& tablet);
- void testInsertRecords(std::vector<std::string>& deviceIds, std::vector<int64_t>& times, std::vector<std::vector<std::string>>& measurementsList, std::vector<std::vector<std::string>>& valuesList);
- void deleteTimeseries(std::string path);
- void deleteTimeseries(std::vector<std::string>& paths);
- void deleteData(std::string path, int64_t time);
- void deleteData(std::vector<std::string>& deviceId, int64_t time);
- void setStorageGroup(std::string storageGroupId);
- void deleteStorageGroup(std::string storageGroup);
- void deleteStorageGroups(std::vector<std::string>& storageGroups);
- void createTimeseries(std::string path, TSDataType::TSDataType dataType, TSEncoding::TSEncoding encoding, CompressionType::CompressionType compressor);
- void createTimeseries(std::string path, TSDataType::TSDataType dataType, TSEncoding::TSEncoding encoding, CompressionType::CompressionType compressor,
- std::map<std::string, std::string>* props, std::map<std::string, std::string>* tags, std::map<std::string, std::string>* attributes, std::string measurementAlias);
- void createMultiTimeseries(std::vector<std::string> paths, std::vector<TSDataType::TSDataType> dataTypes, std::vector<TSEncoding::TSEncoding> encodings, std::vector<CompressionType::CompressionType> compressors,
- std::vector<std::map<std::string, std::string>>* propsList, std::vector<std::map<std::string, std::string>>* tagsList, std::vector<std::map<std::string, std::string>>* attributesList, std::vector<std::string>* measurementAliasList);
- bool checkTimeseriesExists(std::string path);
- std::unique_ptr<SessionDataSet> executeQueryStatement(std::string sql);
- void executeNonQueryStatement(std::string sql);
+ void executeNonQueryStatement(std::string sql);
};
[iotdb] 01/05: fix leak in sessionIdToStatementId &
statementIdToQueryId
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch query-memory-leak
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 0cb9ecf6b32756b2e4571516d612c6652742214d
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Mon Aug 2 11:53:00 2021 +0800
fix leak in sessionIdToStatementId & statementIdToQueryId
---
.../apache/iotdb/db/query/control/SessionManager.java | 17 +++++++++++------
.../java/org/apache/iotdb/db/service/TSServiceImpl.java | 13 +++++++++----
2 files changed, 20 insertions(+), 10 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/query/control/SessionManager.java b/server/src/main/java/org/apache/iotdb/db/query/control/SessionManager.java
index 799f8a6..c3257fb 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/control/SessionManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/control/SessionManager.java
@@ -26,7 +26,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.ZoneId;
-import java.util.Collections;
import java.util.Map;
import java.util.Set;
import java.util.TimeZone;
@@ -78,6 +77,7 @@ public class SessionManager {
public long requestSessionId(String username, String zoneId) {
long sessionId = sessionIdGenerator.incrementAndGet();
+
currSessionId.set(sessionId);
sessionIdToUsername.put(sessionId, username);
sessionIdToZoneId.put(sessionId, ZoneId.of(zoneId));
@@ -88,10 +88,15 @@ public class SessionManager {
public boolean releaseSessionResource(long sessionId) {
sessionIdToZoneId.remove(sessionId);
- for (long statementId :
- sessionIdToStatementId.getOrDefault(sessionId, Collections.emptySet())) {
- for (long queryId : statementIdToQueryId.getOrDefault(statementId, Collections.emptySet())) {
- releaseQueryResourceNoExceptions(queryId);
+ Set<Long> statementIdSet = sessionIdToStatementId.remove(sessionId);
+ if (statementIdSet != null) {
+ for (Long statementId : statementIdSet) {
+ Set<Long> queryIdSet = statementIdToQueryId.remove(statementId);
+ if (queryIdSet != null) {
+ for (Long queryId : queryIdSet) {
+ releaseQueryResourceNoExceptions(queryId);
+ }
+ }
}
}
@@ -109,7 +114,7 @@ public class SessionManager {
public void closeStatement(long sessionId, long statementId) {
Set<Long> queryIdSet = statementIdToQueryId.remove(statementId);
if (queryIdSet != null) {
- for (long queryId : queryIdSet) {
+ for (Long queryId : queryIdSet) {
releaseQueryResourceNoExceptions(queryId);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
index d1bf262..7073d98 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
@@ -304,12 +304,17 @@ public class TSServiceImpl implements TSIService.Iface {
}
try {
- if (req.isSetStatementId() && req.isSetQueryId()) {
- sessionManager.closeDataset(req.statementId, req.queryId);
+ if (req.isSetStatementId()) {
+ if (req.isSetQueryId()) {
+ sessionManager.closeDataset(req.statementId, req.queryId);
+ } else {
+ sessionManager.closeStatement(req.sessionId, req.statementId);
+ }
+ return RpcUtils.getStatus(
+ TSStatusCode.CLOSE_OPERATION_ERROR, "statement id not set by client.");
} else {
- sessionManager.closeStatement(req.sessionId, req.statementId);
+ return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
}
- return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
} catch (Exception e) {
return onNPEOrUnexpectedException(
e, "executing closeOperation", TSStatusCode.CLOSE_OPERATION_ERROR);
[iotdb] 05/05: merge 0.12 and fix conflicts
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch query-memory-leak
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 8d5d616f18be4d7f6e9531a6d022bd8f3ae258c0
Merge: 24e74f9 f26bc84
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Wed Aug 4 10:52:22 2021 +0800
merge 0.12 and fix conflicts
client-cpp/src/main/Session.cpp | 47 +++---
client-cpp/src/main/Session.h | 41 +++---
compile-tools/thrift/pom.xml | 3 +-
docs/SystemDesign/SchemaManager/SchemaManager.md | 2 +-
.../zh/SystemDesign/SchemaManager/SchemaManager.md | 2 +-
.../org/apache/iotdb/db/metadata/MManager.java | 33 +----
.../java/org/apache/iotdb/db/metadata/MTree.java | 36 +++++
.../iotdb/db/qp/strategy/PhysicalGenerator.java | 24 +++-
.../iotdb/db/query/reader/series/SeriesReader.java | 2 +
.../iotdb/db/integration/IoTDBAlignByDeviceIT.java | 54 +++++++
.../iotdb/db/metadata/MManagerBasicTest.java | 160 ++++++++++-----------
11 files changed, 226 insertions(+), 178 deletions(-)