This is an automated email from the ASF dual-hosted git repository.
hui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new fd60088adb [IOTDB-4351] Enhance client-cpp for performance (#7256)
fd60088adb is described below
commit fd60088adb38607219ad611b47d060bdb6a11414
Author: Jamber <[email protected]>
AuthorDate: Thu Sep 8 09:58:41 2022 +0800
[IOTDB-4351] Enhance client-cpp for performance (#7256)
---
client-cpp/src/main/CMakeLists.txt | 2 +-
client-cpp/src/main/Session.cpp | 57 +++++++++++++++++++++++++++-----------
client-cpp/src/main/Session.h | 27 ++++++++++++++----
3 files changed, 64 insertions(+), 22 deletions(-)
diff --git a/client-cpp/src/main/CMakeLists.txt
b/client-cpp/src/main/CMakeLists.txt
index b05cb26b7a..5a044a6383 100644
--- a/client-cpp/src/main/CMakeLists.txt
+++ b/client-cpp/src/main/CMakeLists.txt
@@ -21,7 +21,7 @@ PROJECT(iotdb_session CXX)
SET(CMAKE_CXX_STANDARD 11)
SET(CMAKE_CXX_STANDARD_REQUIRED ON)
SET(CMAKE_POSITION_INDEPENDENT_CODE ON)
-SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11")
+SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11 -Wall -g -O2 ")
SET(TOOLS_DIR "${CMAKE_SOURCE_DIR}/../../../../compile-tools")
# Add Thrift include directory
diff --git a/client-cpp/src/main/Session.cpp b/client-cpp/src/main/Session.cpp
index 4d3f276ccc..54840a1b69 100644
--- a/client-cpp/src/main/Session.cpp
+++ b/client-cpp/src/main/Session.cpp
@@ -274,14 +274,24 @@ void Tablet::setAligned(bool isAligned) {
string SessionUtils::getTime(const Tablet &tablet) {
MyStringBuffer timeBuffer;
+ unsigned int n = 8u * tablet.rowSize;
+ if (n > timeBuffer.str.capacity()) {
+ timeBuffer.reserve(n);
+ }
+
for (size_t i = 0; i < tablet.rowSize; i++) {
- timeBuffer.putLong(tablet.timestamps[i]);
+ timeBuffer.putInt64(tablet.timestamps[i]);
}
return timeBuffer.str;
}
string SessionUtils::getValue(const Tablet &tablet) {
MyStringBuffer valueBuffer;
+ unsigned int n = 8u * tablet.schemas.size() * tablet.rowSize;
+ if (n > valueBuffer.str.capacity()) {
+ valueBuffer.reserve(n);
+ }
+
for (size_t i = 0; i < tablet.schemas.size(); i++) {
TSDataType::TSDataType dataType = tablet.schemas[i].second;
const BitMap& bitMap = tablet.bitMaps[i];
@@ -314,10 +324,10 @@ string SessionUtils::getValue(const Tablet &tablet) {
int64_t* valueBuf = (int64_t*)(tablet.values[i]);
for (size_t index = 0; index < tablet.rowSize; index++) {
if (!bitMap.isMarked(index)) {
- valueBuffer.putLong(valueBuf[index]);
+ valueBuffer.putInt64(valueBuf[index]);
}
else {
- valueBuffer.putLong((numeric_limits<int64_t>::min)());
+ valueBuffer.putInt64((numeric_limits<int64_t>::min)());
}
}
break;
@@ -654,7 +664,7 @@ void Session::sortTablet(Tablet& tablet) {
index[i] = i;
}
- this->sortIndexByTimestamp(index, tablet.timestamps, tablet.rowSize);
+ sortIndexByTimestamp(index, tablet.timestamps, tablet.rowSize);
tablet.timestamps = sortList(tablet.timestamps, index, tablet.rowSize);
for (size_t i = 0; i < tablet.schemas.size(); i++) {
TSDataType::TSDataType dataType = tablet.schemas[i].second;
@@ -743,7 +753,7 @@ Session::putValuesIntoBuffer(const
vector<TSDataType::TSDataType> &types, const
appendValues(buf, values[i], sizeof(double));
break;
case TSDataType::TEXT: {
- uint32_t len = (uint32_t) strlen(values[i]);
+ int32_t len = (uint32_t) strlen(values[i]);
appendValues(buf, (char *) (&len), sizeof(uint32_t));
// no need to change the byte order of string value
buf.append(values[i], len);
@@ -1129,7 +1139,7 @@ void Session::insertRecordsOfOneDevice(const string
&deviceId,
index[i] = (int)i;
}
- this->sortIndexByTimestamp(index, times, (int)(times.size()));
+ sortIndexByTimestamp(index, times, (int)(times.size()));
times = sortList(times, index, (int)(times.size()));
measurementsList = sortList(measurementsList, index,
(int)(times.size()));
typesList = sortList(typesList, index, (int)(times.size()));
@@ -1181,7 +1191,7 @@ void Session::insertAlignedRecordsOfOneDevice(const
string &deviceId,
index[i] = (int)i;
}
- this->sortIndexByTimestamp(index, times, (int)(times.size()));
+ sortIndexByTimestamp(index, times, (int)(times.size()));
times = sortList(times, index, (int)(times.size()));
measurementsList = sortList(measurementsList, index,
(int)(times.size()));
typesList = sortList(typesList, index, (int)(times.size()));
@@ -1223,23 +1233,28 @@ void Session::insertTablet(Tablet &tablet) {
}
}
-void Session::insertTablet(Tablet &tablet, bool sorted) {
- if (!checkSorted(tablet)) {
+void Session::buildInsertTabletReq(TSInsertTabletReq &request, int64_t
sessionId, Tablet &tablet, bool sorted) {
+ if ((!sorted) && !checkSorted(tablet)) {
sortTablet(tablet);
}
- TSInsertTabletReq request;
request.__set_sessionId(sessionId);
request.prefixPath = tablet.deviceId;
+
+ request.measurements.reserve(tablet.schemas.size());
+ request.types.reserve(tablet.schemas.size());
for (pair<string, TSDataType::TSDataType> schema: tablet.schemas) {
request.measurements.push_back(schema.first);
request.types.push_back(schema.second);
}
- request.__set_timestamps(SessionUtils::getTime(tablet));
- request.__set_values(SessionUtils::getValue(tablet));
+
+ request.values = move(SessionUtils::getValue(tablet));
+ request.timestamps = move(SessionUtils::getTime(tablet));
request.__set_size(tablet.rowSize);
request.__set_isAligned(tablet.isAligned);
+}
+void Session::insertTablet(const TSInsertTabletReq &request){
try {
TSStatus respStatus;
client->insertTablet(respStatus, request);
@@ -1251,6 +1266,12 @@ void Session::insertTablet(Tablet &tablet, bool sorted) {
}
}
+void Session::insertTablet(Tablet &tablet, bool sorted) {
+ TSInsertTabletReq request;
+ buildInsertTabletReq(request, sessionId, tablet, sorted);
+ insertTablet(request);
+}
+
void Session::insertAlignedTablet(Tablet &tablet) {
insertAlignedTablet(tablet, false);
}
@@ -1302,8 +1323,8 @@ void Session::insertTablets(unordered_map<string, Tablet
*> &tablets, bool sorte
}
request.measurementsList.push_back(measurements);
request.typesList.push_back(dataTypes);
-
request.timestampsList.push_back(SessionUtils::getTime(*(item.second)));
- request.valuesList.push_back(SessionUtils::getValue(*(item.second)));
+
request.timestampsList.push_back(move(SessionUtils::getTime(*(item.second))));
+
request.valuesList.push_back(move(SessionUtils::getValue(*(item.second))));
request.sizeList.push_back(item.second->rowSize);
}
request.__set_isAligned(isFirstTabletAligned);
@@ -1360,8 +1381,8 @@ void Session::testInsertTablet(const Tablet &tablet) {
request.measurements.push_back(schema.first);
request.types.push_back(schema.second);
}
- request.__set_timestamps(SessionUtils::getTime(tablet));
- request.__set_values(SessionUtils::getValue(tablet));
+ request.__set_timestamps(move(SessionUtils::getTime(tablet)));
+ request.__set_values(move(SessionUtils::getValue(tablet)));
request.__set_size(tablet.rowSize);
try {
@@ -1639,6 +1660,10 @@ bool Session::checkTimeseriesExists(const string &path) {
}
}
+int64_t Session::getSessionId() {
+ return sessionId;
+}
+
string Session::getTimeZone() {
if (!zoneId.empty()) {
return zoneId;
diff --git a/client-cpp/src/main/Session.h b/client-cpp/src/main/Session.h
index 5adade396e..9d9dc665f5 100644
--- a/client-cpp/src/main/Session.h
+++ b/client-cpp/src/main/Session.h
@@ -276,6 +276,15 @@ public:
checkBigEndian();
}
+ void reserve(size_t n) {
+ str.reserve(n);
+ }
+
+ void clear() {
+ str.clear();
+ pos = 0;
+ }
+
bool hasRemaining() {
return pos < str.size();
}
@@ -315,7 +324,7 @@ public:
putOrderedByte((char *) &ins, 4);
}
- void putLong(int64_t ins) {
+ void putInt64(int64_t ins) {
putOrderedByte((char *) &ins, 8);
}
@@ -528,6 +537,8 @@ public:
size_t maxRowNumber; // the maximum number of rows for this tablet
bool isAligned; // whether this tablet store data of aligned timeseries
or not
+ Tablet() = default;
+
/**
* Return a tablet with default specified row number. This is the standard
* constructor (all Tablet should be the same size).
@@ -925,13 +936,13 @@ private:
const static int DEFAULT_TIMEOUT_MS = 0;
Version::Version version;
- bool checkSorted(const Tablet &tablet);
+ static bool checkSorted(const Tablet &tablet);
- bool checkSorted(const std::vector<int64_t> ×);
+ static bool checkSorted(const std::vector<int64_t> ×);
- void sortTablet(Tablet &tablet);
+ static void sortTablet(Tablet &tablet);
- void sortIndexByTimestamp(int *index, std::vector<int64_t> ×tamps,
int length);
+ static void sortIndexByTimestamp(int *index, std::vector<int64_t>
×tamps, int length);
std::string getTimeZone();
@@ -995,6 +1006,8 @@ public:
~Session();
+ int64_t getSessionId();
+
void open();
void open(bool enableRPCCompression);
@@ -1067,6 +1080,10 @@ public:
void insertTablet(Tablet &tablet, bool sorted);
+ static void buildInsertTabletReq(TSInsertTabletReq &request, int64_t
sessionId, Tablet &tablet, bool sorted);
+
+ void insertTablet(const TSInsertTabletReq &request);
+
void insertAlignedTablet(Tablet &tablet);
void insertAlignedTablet(Tablet &tablet, bool sorted);
