| /** |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| |
| #include "Session.h" |
| #include "SessionDataSet.h" |
| #include "SessionImpl.h" |
| #include "ThriftConvert.h" |
| #include <algorithm> |
| #include <cstring> |
| #include <future> |
| #include <memory> |
| #include <stack> |
| #include <thrift/transport/TTransportException.h> |
| #include <time.h> |
| #include <unordered_set> |
| |
| using apache::thrift::transport::TTransportException; |
| |
| using namespace std; |
| |
| /** |
| * Timeout of query can be set by users. |
| * A negative number means using the default configuration of server. |
| * And value 0 will disable the function of query timeout. |
| */ |
| static const int64_t QUERY_TIMEOUT_MS = -1; |
| |
| TSDataType::TSDataType getTSDataTypeFromString(const string &str) { |
| // BOOLEAN, INT32, INT64, FLOAT, DOUBLE, TEXT, STRING, BLOB, TIMESTAMP, DATE, |
| // 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 == "TIMESTAMP") { |
| return TSDataType::TIMESTAMP; |
| } else if (str == "DATE") { |
| return TSDataType::DATE; |
| } else if (str == "BLOB") { |
| return TSDataType::BLOB; |
| } else if (str == "STRING") { |
| return TSDataType::STRING; |
| } else if (str == "OBJECT") { |
| return TSDataType::OBJECT; |
| } |
| return TSDataType::UNKNOWN; |
| } |
| |
| void Tablet::createColumns() { |
| for (size_t i = 0; i < schemas.size(); i++) { |
| TSDataType::TSDataType dataType = schemas[i].second; |
| switch (dataType) { |
| case TSDataType::BOOLEAN: |
| values[i] = new bool[maxRowNumber]; |
| break; |
| case TSDataType::DATE: |
| values[i] = new IoTDBDate[maxRowNumber]; |
| break; |
| case TSDataType::INT32: |
| values[i] = new int[maxRowNumber]; |
| break; |
| case TSDataType::TIMESTAMP: |
| case TSDataType::INT64: |
| values[i] = new int64_t[maxRowNumber]; |
| break; |
| case TSDataType::FLOAT: |
| values[i] = new float[maxRowNumber]; |
| break; |
| case TSDataType::DOUBLE: |
| values[i] = new double[maxRowNumber]; |
| break; |
| case TSDataType::STRING: |
| case TSDataType::BLOB: |
| case TSDataType::OBJECT: |
| case TSDataType::TEXT: |
| values[i] = new string[maxRowNumber]; |
| break; |
| default: |
| throw UnSupportedDataTypeException( |
| string("Data type ") + to_string(dataType) + " is not supported."); |
| } |
| } |
| } |
| |
| void Tablet::deleteColumns() { |
| for (size_t i = 0; i < schemas.size(); i++) { |
| if (!values[i]) |
| continue; |
| TSDataType::TSDataType dataType = schemas[i].second; |
| switch (dataType) { |
| case TSDataType::BOOLEAN: { |
| bool *valueBuf = (bool *)(values[i]); |
| delete[] valueBuf; |
| break; |
| } |
| case TSDataType::INT32: { |
| int *valueBuf = (int *)(values[i]); |
| delete[] valueBuf; |
| break; |
| } |
| case TSDataType::DATE: { |
| IoTDBDate *valueBuf = (IoTDBDate *)(values[i]); |
| delete[] valueBuf; |
| break; |
| } |
| case TSDataType::TIMESTAMP: |
| case TSDataType::INT64: { |
| int64_t *valueBuf = (int64_t *)(values[i]); |
| delete[] valueBuf; |
| break; |
| } |
| case TSDataType::FLOAT: { |
| float *valueBuf = (float *)(values[i]); |
| delete[] valueBuf; |
| break; |
| } |
| case TSDataType::DOUBLE: { |
| double *valueBuf = (double *)(values[i]); |
| delete[] valueBuf; |
| break; |
| } |
| case TSDataType::STRING: |
| case TSDataType::BLOB: |
| case TSDataType::OBJECT: |
| case TSDataType::TEXT: { |
| string *valueBuf = (string *)(values[i]); |
| delete[] valueBuf; |
| break; |
| } |
| default: |
| throw UnSupportedDataTypeException( |
| string("Data type ") + to_string(dataType) + " is not supported."); |
| } |
| values[i] = nullptr; |
| } |
| } |
| |
| void Tablet::deepCopyTabletColValue(void *const *srcPtr, void **destPtr, |
| TSDataType::TSDataType type, |
| int maxRowNumber) { |
| void *src = *srcPtr; |
| switch (type) { |
| case TSDataType::BOOLEAN: |
| *destPtr = new bool[maxRowNumber]; |
| memcpy(*destPtr, src, maxRowNumber * sizeof(bool)); |
| break; |
| case TSDataType::INT32: |
| *destPtr = new int32_t[maxRowNumber]; |
| memcpy(*destPtr, src, maxRowNumber * sizeof(int32_t)); |
| break; |
| case TSDataType::INT64: |
| case TSDataType::TIMESTAMP: |
| *destPtr = new int64_t[maxRowNumber]; |
| memcpy(*destPtr, src, maxRowNumber * sizeof(int64_t)); |
| break; |
| case TSDataType::FLOAT: |
| *destPtr = new float[maxRowNumber]; |
| memcpy(*destPtr, src, maxRowNumber * sizeof(float)); |
| break; |
| case TSDataType::DOUBLE: |
| *destPtr = new double[maxRowNumber]; |
| memcpy(*destPtr, src, maxRowNumber * sizeof(double)); |
| break; |
| case TSDataType::DATE: { |
| *destPtr = new IoTDBDate[maxRowNumber]; |
| IoTDBDate *srcDate = static_cast<IoTDBDate *>(src); |
| IoTDBDate *destDate = static_cast<IoTDBDate *>(*destPtr); |
| for (size_t j = 0; j < maxRowNumber; ++j) { |
| destDate[j] = srcDate[j]; |
| } |
| break; |
| } |
| case TSDataType::STRING: |
| case TSDataType::TEXT: |
| case TSDataType::OBJECT: |
| case TSDataType::BLOB: { |
| *destPtr = new std::string[maxRowNumber]; |
| std::string *srcStr = static_cast<std::string *>(src); |
| std::string *destStr = static_cast<std::string *>(*destPtr); |
| for (size_t j = 0; j < maxRowNumber; ++j) { |
| destStr[j] = srcStr[j]; |
| } |
| break; |
| } |
| default: |
| break; |
| } |
| } |
| |
| void Tablet::reset() { |
| rowSize = 0; |
| for (size_t i = 0; i < schemas.size(); i++) { |
| bitMaps[i].reset(); |
| } |
| } |
| |
| size_t Tablet::getTimeBytesSize() { return rowSize * 8; } |
| |
| size_t Tablet::getValueByteSize() { |
| size_t valueOccupation = 0; |
| for (size_t 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::DATE: |
| valueOccupation += rowSize * 4; |
| break; |
| case TSDataType::TIMESTAMP: |
| case TSDataType::INT64: |
| valueOccupation += rowSize * 8; |
| break; |
| case TSDataType::FLOAT: |
| valueOccupation += rowSize * 4; |
| break; |
| case TSDataType::DOUBLE: |
| valueOccupation += rowSize * 8; |
| break; |
| case TSDataType::STRING: |
| case TSDataType::BLOB: |
| case TSDataType::OBJECT: |
| case TSDataType::TEXT: { |
| valueOccupation += rowSize * 4; |
| string *valueBuf = (string *)(values[i]); |
| for (size_t j = 0; j < rowSize; j++) { |
| valueOccupation += valueBuf[j].size(); |
| } |
| break; |
| } |
| default: |
| throw UnSupportedDataTypeException(string("Data type ") + |
| to_string(schemas[i].second) + |
| " is not supported."); |
| } |
| } |
| return valueOccupation; |
| } |
| |
| void Tablet::setAligned(bool isAligned) { this->isAligned = 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.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]; |
| switch (dataType) { |
| case TSDataType::BOOLEAN: { |
| bool *valueBuf = (bool *)(tablet.values[i]); |
| for (size_t index = 0; index < tablet.rowSize; index++) { |
| if (!bitMap.isMarked(index)) { |
| valueBuffer.putBool(valueBuf[index]); |
| } else { |
| valueBuffer.putBool(false); |
| } |
| } |
| break; |
| } |
| case TSDataType::INT32: { |
| int *valueBuf = (int *)(tablet.values[i]); |
| for (size_t index = 0; index < tablet.rowSize; index++) { |
| if (!bitMap.isMarked(index)) { |
| valueBuffer.putInt(valueBuf[index]); |
| } else { |
| valueBuffer.putInt((numeric_limits<int32_t>::min)()); |
| } |
| } |
| break; |
| } |
| case TSDataType::DATE: { |
| IoTDBDate *valueBuf = (IoTDBDate *)(tablet.values[i]); |
| for (size_t index = 0; index < tablet.rowSize; index++) { |
| if (!bitMap.isMarked(index)) { |
| valueBuffer.putDate(valueBuf[index]); |
| } else { |
| valueBuffer.putInt(EMPTY_DATE_INT); |
| } |
| } |
| break; |
| } |
| case TSDataType::TIMESTAMP: |
| case TSDataType::INT64: { |
| int64_t *valueBuf = (int64_t *)(tablet.values[i]); |
| for (size_t index = 0; index < tablet.rowSize; index++) { |
| if (!bitMap.isMarked(index)) { |
| valueBuffer.putInt64(valueBuf[index]); |
| } else { |
| valueBuffer.putInt64((numeric_limits<int64_t>::min)()); |
| } |
| } |
| break; |
| } |
| case TSDataType::FLOAT: { |
| float *valueBuf = (float *)(tablet.values[i]); |
| for (size_t index = 0; index < tablet.rowSize; index++) { |
| if (!bitMap.isMarked(index)) { |
| valueBuffer.putFloat(valueBuf[index]); |
| } else { |
| valueBuffer.putFloat((numeric_limits<float>::min)()); |
| } |
| } |
| break; |
| } |
| case TSDataType::DOUBLE: { |
| double *valueBuf = (double *)(tablet.values[i]); |
| for (size_t index = 0; index < tablet.rowSize; index++) { |
| if (!bitMap.isMarked(index)) { |
| valueBuffer.putDouble(valueBuf[index]); |
| } else { |
| valueBuffer.putDouble((numeric_limits<double>::min)()); |
| } |
| } |
| break; |
| } |
| case TSDataType::STRING: |
| case TSDataType::BLOB: |
| case TSDataType::OBJECT: |
| case TSDataType::TEXT: { |
| string *valueBuf = (string *)(tablet.values[i]); |
| for (size_t index = 0; index < tablet.rowSize; index++) { |
| if (!bitMap.isMarked(index)) { |
| valueBuffer.putString(valueBuf[index]); |
| } else { |
| valueBuffer.putString(""); |
| } |
| } |
| break; |
| } |
| default: |
| throw UnSupportedDataTypeException( |
| string("Data type ") + to_string(dataType) + " is not supported."); |
| } |
| } |
| for (size_t i = 0; i < tablet.schemas.size(); i++) { |
| const BitMap &bitMap = tablet.bitMaps[i]; |
| bool columnHasNull = !bitMap.isAllUnmarked(); |
| valueBuffer.putChar(columnHasNull ? (char)1 : (char)0); |
| if (columnHasNull) { |
| const vector<char> &bytes = bitMap.getByteArray(); |
| for (size_t index = 0; index < tablet.rowSize / 8 + 1; index++) { |
| valueBuffer.putChar(bytes[index]); |
| } |
| } |
| } |
| return valueBuffer.str; |
| } |
| |
| string MeasurementNode::serialize() const { |
| MyStringBuffer buffer; |
| buffer.putString(getName()); |
| buffer.putChar(getDataType()); |
| buffer.putChar(getEncoding()); |
| buffer.putChar(getCompressionType()); |
| return buffer.str; |
| } |
| |
| string Template::serialize() const { |
| MyStringBuffer buffer; |
| std::stack<std::pair<std::string, std::shared_ptr<TemplateNode>>> nodeStack; |
| unordered_set<string> alignedPrefix; |
| buffer.putString(getName()); |
| buffer.putBool(isAligned()); |
| if (isAligned()) { |
| alignedPrefix.emplace(""); |
| } |
| |
| for (const auto &child : children_) { |
| nodeStack.push(make_pair("", child.second)); |
| } |
| |
| while (!nodeStack.empty()) { |
| auto cur = nodeStack.top(); |
| nodeStack.pop(); |
| |
| string prefix = cur.first; |
| shared_ptr<TemplateNode> cur_node_ptr = cur.second; |
| string fullPath(prefix); |
| |
| if (!cur_node_ptr->isMeasurement()) { |
| if (!prefix.empty()) { |
| fullPath.append("."); |
| } |
| fullPath.append(cur_node_ptr->getName()); |
| if (cur_node_ptr->isAligned()) { |
| alignedPrefix.emplace(fullPath); |
| } |
| for (const auto &child : cur_node_ptr->getChildren()) { |
| nodeStack.push(make_pair(fullPath, child.second)); |
| } |
| } else { |
| buffer.putString(prefix); |
| buffer.putBool(alignedPrefix.find(prefix) != alignedPrefix.end()); |
| buffer.concat(cur_node_ptr->serialize()); |
| } |
| } |
| |
| return buffer.str; |
| } |
| |
| Session::Session(const std::string &host, int rpcPort) : impl_(new Impl()) { |
| impl_->username_ = "root"; |
| impl_->password_ = "root"; |
| impl_->version = Version::V_1_0; |
| impl_->host_ = host; |
| impl_->rpcPort_ = rpcPort; |
| impl_->initZoneId(); |
| impl_->initNodesSupplier(); |
| } |
| |
| Session::Session(const std::vector<std::string> &nodeUrls, |
| const std::string &username, const std::string &password) |
| : impl_(new Impl()) { |
| impl_->nodeUrls_ = nodeUrls; |
| impl_->username_ = username; |
| impl_->password_ = password; |
| impl_->version = Version::V_1_0; |
| impl_->initZoneId(); |
| impl_->initNodesSupplier(impl_->nodeUrls_); |
| } |
| |
| Session::Session(const std::string &host, int rpcPort, |
| const std::string &username, const std::string &password) |
| : impl_(new Impl()) { |
| impl_->host_ = host; |
| impl_->rpcPort_ = rpcPort; |
| impl_->username_ = username; |
| impl_->password_ = password; |
| impl_->fetchSize_ = iotdb::session::DEFAULT_FETCH_SIZE; |
| impl_->version = Version::V_1_0; |
| impl_->initZoneId(); |
| impl_->initNodesSupplier(); |
| } |
| |
| Session::Session(const std::string &host, int rpcPort, |
| const std::string &username, const std::string &password, |
| const std::string &zoneId, int fetchSize) |
| : impl_(new Impl()) { |
| impl_->host_ = host; |
| impl_->rpcPort_ = rpcPort; |
| impl_->username_ = username; |
| impl_->password_ = password; |
| impl_->zoneId_ = zoneId; |
| impl_->fetchSize_ = fetchSize; |
| impl_->version = Version::V_1_0; |
| impl_->initZoneId(); |
| impl_->initNodesSupplier(); |
| } |
| |
| Session::Session(const std::string &host, const std::string &rpcPort, |
| const std::string &username, const std::string &password, |
| const std::string &zoneId, int fetchSize) |
| : impl_(new Impl()) { |
| impl_->host_ = host; |
| impl_->rpcPort_ = stoi(rpcPort); |
| impl_->username_ = username; |
| impl_->password_ = password; |
| impl_->zoneId_ = zoneId; |
| impl_->fetchSize_ = fetchSize; |
| impl_->version = Version::V_1_0; |
| impl_->initZoneId(); |
| impl_->initNodesSupplier(); |
| } |
| |
| Session::Session(AbstractSessionBuilder *builder) : impl_(new Impl()) { |
| impl_->host_ = builder->host; |
| impl_->rpcPort_ = builder->rpcPort; |
| impl_->username_ = builder->username; |
| impl_->password_ = builder->password; |
| impl_->zoneId_ = builder->zoneId; |
| impl_->fetchSize_ = builder->fetchSize; |
| impl_->version = Version::V_1_0; |
| impl_->sqlDialect_ = builder->sqlDialect; |
| impl_->database_ = builder->database; |
| impl_->enableAutoFetch_ = builder->enableAutoFetch; |
| impl_->enableRedirection_ = builder->enableRedirections; |
| impl_->connectTimeoutMs_ = builder->connectTimeoutMs; |
| impl_->nodeUrls_ = builder->nodeUrls; |
| impl_->useSSL_ = builder->useSSL; |
| impl_->trustCertFilePath_ = builder->trustCertFilePath; |
| impl_->initZoneId(); |
| impl_->initNodesSupplier(impl_->nodeUrls_); |
| } |
| |
| /** |
| * When delete variable, make sure release all resource. |
| */ |
| Session::~Session() { |
| try { |
| close(); |
| } catch (const exception &e) { |
| log_debug(e.what()); |
| } |
| } |
| |
| void Session::Impl::removeBrokenSessionConnection( |
| shared_ptr<SessionConnection> sessionConnection) { |
| if (enableRedirection_) { |
| this->endPointToSessionConnection.erase(sessionConnection->getEndPoint()); |
| } |
| |
| auto it1 = deviceIdToEndpoint.begin(); |
| while (it1 != deviceIdToEndpoint.end()) { |
| if (it1->second == sessionConnection->getEndPoint()) { |
| it1 = deviceIdToEndpoint.erase(it1); |
| } else { |
| ++it1; |
| } |
| } |
| } |
| |
| /** |
| * check whether the batch has been sorted |
| * |
| * @return whether the batch has been sorted |
| */ |
| bool Session::Impl::checkSorted(const Tablet &tablet) { |
| for (size_t i = 1; i < tablet.rowSize; i++) { |
| if (tablet.timestamps[i] < tablet.timestamps[i - 1]) { |
| return false; |
| } |
| } |
| return true; |
| } |
| |
| bool Session::Impl::checkSorted(const vector<int64_t> ×) { |
| for (size_t i = 1; i < times.size(); i++) { |
| if (times[i] < times[i - 1]) { |
| return false; |
| } |
| } |
| return true; |
| } |
| |
| template <typename T> |
| std::vector<T> sortList(const std::vector<T> &valueList, const int *index, |
| int indexLength) { |
| std::vector<T> sortedValues(valueList.size()); |
| for (int i = 0; i < indexLength; i++) { |
| sortedValues[i] = valueList[index[i]]; |
| } |
| return sortedValues; |
| } |
| |
| template <typename T> |
| void sortValuesList(T *valueList, const int *index, size_t indexLength) { |
| T *sortedValues = new T[indexLength]; |
| for (int i = 0; i < indexLength; i++) { |
| sortedValues[i] = valueList[index[i]]; |
| } |
| for (int i = 0; i < indexLength; i++) { |
| valueList[i] = sortedValues[i]; |
| } |
| delete[] sortedValues; |
| } |
| |
| void Session::Impl::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]; |
| for (size_t i = 0; i < tablet.rowSize; i++) { |
| index[i] = i; |
| } |
| |
| 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; |
| switch (dataType) { |
| case TSDataType::BOOLEAN: { |
| sortValuesList((bool *)(tablet.values[i]), index, tablet.rowSize); |
| break; |
| } |
| case TSDataType::INT32: { |
| sortValuesList((int *)(tablet.values[i]), index, tablet.rowSize); |
| break; |
| } |
| case TSDataType::DATE: { |
| sortValuesList((IoTDBDate *)(tablet.values[i]), index, tablet.rowSize); |
| break; |
| } |
| case TSDataType::TIMESTAMP: |
| case TSDataType::INT64: { |
| sortValuesList((int64_t *)(tablet.values[i]), index, tablet.rowSize); |
| break; |
| } |
| case TSDataType::FLOAT: { |
| sortValuesList((float *)(tablet.values[i]), index, tablet.rowSize); |
| break; |
| } |
| case TSDataType::DOUBLE: { |
| sortValuesList((double *)(tablet.values[i]), index, tablet.rowSize); |
| break; |
| } |
| case TSDataType::STRING: |
| case TSDataType::BLOB: |
| case TSDataType::OBJECT: |
| case TSDataType::TEXT: { |
| sortValuesList((string *)(tablet.values[i]), index, tablet.rowSize); |
| break; |
| } |
| default: |
| throw UnSupportedDataTypeException( |
| string("Data type ") + to_string(dataType) + " is not supported."); |
| } |
| } |
| |
| delete[] index; |
| } |
| |
| void Session::Impl::sortIndexByTimestamp(int *index, |
| std::vector<int64_t> ×tamps, |
| int length) { |
| if (length <= 1) { |
| return; |
| } |
| |
| TsCompare tsCompareObj(timestamps); |
| std::sort(&index[0], &index[length], tsCompareObj); |
| } |
| |
| /** |
| * Append value into buffer in Big Endian order to comply with IoTDB server |
| */ |
| void Session::Impl::appendValues(string &buffer, const char *value, int size) { |
| static bool hasCheckedEndianFlag = false; |
| static bool localCpuIsBigEndian = false; |
| if (!hasCheckedEndianFlag) { |
| hasCheckedEndianFlag = true; |
| int chk = |
| 0x0201; // used to distinguish CPU's type (BigEndian or LittleEndian) |
| localCpuIsBigEndian = (0x01 != *(char *)(&chk)); |
| } |
| |
| if (localCpuIsBigEndian) { |
| buffer.append(value, size); |
| } else { |
| for (int i = size - 1; i >= 0; i--) { |
| buffer.append(value + i, 1); |
| } |
| } |
| } |
| |
| void Session::Impl::putValuesIntoBuffer( |
| const vector<TSDataType::TSDataType> &types, const vector<char *> &values, |
| string &buf) { |
| int32_t date; |
| for (size_t i = 0; i < values.size(); i++) { |
| int8_t typeNum = getDataTypeNumber(types[i]); |
| buf.append((char *)(&typeNum), sizeof(int8_t)); |
| switch (types[i]) { |
| case TSDataType::BOOLEAN: |
| buf.append(values[i], 1); |
| break; |
| case TSDataType::INT32: |
| appendValues(buf, values[i], sizeof(int32_t)); |
| break; |
| case TSDataType::DATE: |
| date = parseDateExpressionToInt(*(IoTDBDate *)values[i]); |
| appendValues(buf, (char *)&date, sizeof(int32_t)); |
| break; |
| case TSDataType::TIMESTAMP: |
| case TSDataType::INT64: |
| appendValues(buf, values[i], sizeof(int64_t)); |
| break; |
| case TSDataType::FLOAT: |
| appendValues(buf, values[i], sizeof(float)); |
| break; |
| case TSDataType::DOUBLE: |
| appendValues(buf, values[i], sizeof(double)); |
| break; |
| case TSDataType::STRING: |
| case TSDataType::BLOB: |
| case TSDataType::OBJECT: |
| case TSDataType::TEXT: { |
| 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); |
| break; |
| } |
| default: |
| break; |
| } |
| } |
| } |
| |
| int8_t Session::Impl::getDataTypeNumber(TSDataType::TSDataType type) { |
| switch (type) { |
| case TSDataType::BOOLEAN: |
| return 0; |
| case TSDataType::INT32: |
| return 1; |
| case TSDataType::INT64: |
| return 2; |
| case TSDataType::FLOAT: |
| return 3; |
| case TSDataType::DOUBLE: |
| return 4; |
| case TSDataType::TEXT: |
| return 5; |
| case TSDataType::TIMESTAMP: |
| return 8; |
| case TSDataType::DATE: |
| return 9; |
| case TSDataType::BLOB: |
| return 10; |
| case TSDataType::STRING: |
| return 11; |
| case TSDataType::OBJECT: |
| return 12; |
| default: |
| return -1; |
| } |
| } |
| |
| string Session::Impl::getVersionString(Version::Version version) { |
| switch (version) { |
| case Version::V_0_12: |
| return "V_0_12"; |
| case Version::V_0_13: |
| return "V_0_13"; |
| case Version::V_1_0: |
| return "V_1_0"; |
| default: |
| return "V_0_12"; |
| } |
| } |
| |
| void Session::Impl::initZoneId() { |
| if (!zoneId_.empty()) { |
| return; |
| } |
| |
| time_t ts = 0; |
| struct tm tmv; |
| #if defined(_WIN64) || defined(WIN32) || defined(_WIN32) |
| localtime_s(&tmv, &ts); |
| #else |
| localtime_r(&ts, &tmv); |
| #endif |
| |
| char zoneStr[32]; |
| strftime(zoneStr, sizeof(zoneStr), "%z", &tmv); |
| zoneId_ = zoneStr; |
| } |
| |
| void Session::Impl::initNodesSupplier( |
| const std::vector<std::string> &nodeUrls) { |
| std::vector<TEndPoint> endPoints; |
| std::unordered_set<std::string> uniqueEndpoints; |
| |
| if (nodeUrls.empty() && host_.empty()) { |
| throw IoTDBException("No available nodes"); |
| } |
| |
| // Process provided node URLs |
| if (!nodeUrls.empty()) { |
| for (auto &url : nodeUrls) { |
| try { |
| TEndPoint endPoint = UrlUtils::parseTEndPointIpv4AndIpv6Url(url); |
| if (endPoint.port == 0) |
| continue; // Skip invalid endpoints |
| |
| std::string endpointKey = |
| endPoint.ip + ":" + std::to_string(endPoint.port); |
| if (uniqueEndpoints.find(endpointKey) == uniqueEndpoints.end()) { |
| endPoints.emplace_back(std::move(endPoint)); |
| uniqueEndpoints.insert(std::move(endpointKey)); |
| } |
| } catch (...) { |
| continue; // Skip malformed URLs |
| } |
| } |
| } |
| |
| // Fallback to local endpoint if no valid endpoints found |
| if (endPoints.empty()) { |
| if (host_.empty() || rpcPort_ == 0) { |
| throw IoTDBException("No valid endpoints available"); |
| } |
| TEndPoint endPoint; |
| endPoint.__set_ip(host_); |
| endPoint.__set_port(rpcPort_); |
| endPoints.emplace_back(std::move(endPoint)); |
| } |
| |
| if (enableAutoFetch_) { |
| nodesSupplier_ = NodesSupplier::create(endPoints, username_, password_, |
| useSSL_, trustCertFilePath_); |
| } else { |
| nodesSupplier_ = make_shared<StaticNodesSupplier>(endPoints); |
| } |
| } |
| |
| void Session::Impl::initDefaultSessionConnection() { |
| // Try all endpoints from supplier until a connection is established. |
| auto endpoints = nodesSupplier_->getEndPointList(); |
| bool connected = false; |
| |
| for (const auto &endpoint : endpoints) { |
| try { |
| host_ = endpoint.ip; |
| rpcPort_ = endpoint.port; |
| |
| defaultEndPoint_.__set_ip(host_); |
| defaultEndPoint_.__set_port(rpcPort_); |
| |
| defaultSessionConnection_ = std::make_shared<SessionConnection>( |
| this, defaultEndPoint_, zoneId_, nodesSupplier_, fetchSize_, 3, 500, |
| connectTimeoutMs_, sqlDialect_, database_); |
| |
| connected = true; |
| break; |
| } catch (const IoTDBException &e) { |
| log_debug(e.what()); |
| throw; |
| } catch (const std::exception &e) { |
| log_warn(e.what()); |
| } |
| } |
| |
| if (!connected) { |
| throw std::runtime_error( |
| "No available node to establish SessionConnection."); |
| } |
| } |
| |
| void Session::Impl::insertStringRecordsWithLeaderCache( |
| vector<string> deviceIds, vector<int64_t> times, |
| vector<vector<string>> measurementsList, vector<vector<string>> valuesList, |
| bool isAligned) { |
| std::unordered_map<std::shared_ptr<SessionConnection>, |
| TSInsertStringRecordsReq> |
| recordsGroup; |
| for (int i = 0; i < deviceIds.size(); i++) { |
| auto connection = getSessionConnection(deviceIds[i]); |
| if (recordsGroup.find(connection) == recordsGroup.end()) { |
| TSInsertStringRecordsReq request; |
| std::vector<std::string> emptyPrefixPaths; |
| std::vector<std::vector<std::string>> emptyMeasurementsList; |
| vector<vector<string>> emptyValuesList; |
| std::vector<int64_t> emptyTimestamps; |
| request.__set_isAligned(isAligned); |
| request.__set_prefixPaths(emptyPrefixPaths); |
| request.__set_timestamps(emptyTimestamps); |
| request.__set_measurementsList(emptyMeasurementsList); |
| request.__set_valuesList(emptyValuesList); |
| recordsGroup.insert(make_pair(connection, request)); |
| } |
| TSInsertStringRecordsReq &existingReq = recordsGroup[connection]; |
| existingReq.prefixPaths.emplace_back(deviceIds[i]); |
| existingReq.timestamps.emplace_back(times[i]); |
| existingReq.measurementsList.emplace_back(measurementsList[i]); |
| existingReq.valuesList.emplace_back(valuesList[i]); |
| } |
| std::function<void(std::shared_ptr<SessionConnection>, |
| const TSInsertStringRecordsReq &)> |
| consumer = |
| [](const std::shared_ptr<SessionConnection> &c, |
| const TSInsertStringRecordsReq &r) { c->insertStringRecords(r); }; |
| if (recordsGroup.size() == 1) { |
| insertOnce(recordsGroup, consumer); |
| } else { |
| insertByGroup(recordsGroup, consumer); |
| } |
| } |
| |
| void Session::Impl::insertRecordsWithLeaderCache( |
| vector<string> deviceIds, vector<int64_t> times, |
| vector<vector<string>> measurementsList, |
| const vector<vector<TSDataType::TSDataType>> &typesList, |
| vector<vector<char *>> valuesList, bool isAligned) { |
| std::unordered_map<std::shared_ptr<SessionConnection>, TSInsertRecordsReq> |
| recordsGroup; |
| for (int i = 0; i < deviceIds.size(); i++) { |
| auto connection = getSessionConnection(deviceIds[i]); |
| if (recordsGroup.find(connection) == recordsGroup.end()) { |
| TSInsertRecordsReq request; |
| std::vector<std::string> emptyPrefixPaths; |
| std::vector<std::vector<std::string>> emptyMeasurementsList; |
| std::vector<std::string> emptyValuesList; |
| std::vector<int64_t> emptyTimestamps; |
| request.__set_isAligned(isAligned); |
| request.__set_prefixPaths(emptyPrefixPaths); |
| request.__set_timestamps(emptyTimestamps); |
| request.__set_measurementsList(emptyMeasurementsList); |
| request.__set_valuesList(emptyValuesList); |
| recordsGroup.insert(make_pair(connection, request)); |
| } |
| TSInsertRecordsReq &existingReq = recordsGroup[connection]; |
| existingReq.prefixPaths.emplace_back(deviceIds[i]); |
| existingReq.timestamps.emplace_back(times[i]); |
| existingReq.measurementsList.emplace_back(measurementsList[i]); |
| vector<string> bufferList; |
| string buffer; |
| putValuesIntoBuffer(typesList[i], valuesList[i], buffer); |
| existingReq.valuesList.emplace_back(buffer); |
| recordsGroup[connection] = existingReq; |
| } |
| std::function<void(std::shared_ptr<SessionConnection>, |
| const TSInsertRecordsReq &)> |
| consumer = [](const std::shared_ptr<SessionConnection> &c, |
| const TSInsertRecordsReq &r) { c->insertRecords(r); }; |
| if (recordsGroup.size() == 1) { |
| insertOnce(recordsGroup, consumer); |
| } else { |
| insertByGroup(recordsGroup, consumer); |
| } |
| } |
| |
| void Session::Impl::insertTabletsWithLeaderCache( |
| unordered_map<string, Tablet *> &tablets, bool sorted, bool isAligned) { |
| std::unordered_map<std::shared_ptr<SessionConnection>, TSInsertTabletsReq> |
| tabletsGroup; |
| if (tablets.empty()) { |
| throw BatchExecutionException("No tablet is inserting!"); |
| } |
| for (const auto &item : tablets) { |
| if (isAligned != item.second->isAligned) { |
| throw BatchExecutionException( |
| "The tablets should be all aligned or non-aligned!"); |
| } |
| if (!checkSorted(*(item.second))) { |
| sortTablet(*(item.second)); |
| } |
| auto deviceId = item.first; |
| auto tablet = item.second; |
| auto connection = getSessionConnection(deviceId); |
| auto it = tabletsGroup.find(connection); |
| if (it == tabletsGroup.end()) { |
| TSInsertTabletsReq request; |
| tabletsGroup[connection] = request; |
| } |
| TSInsertTabletsReq &existingReq = tabletsGroup[connection]; |
| existingReq.prefixPaths.emplace_back(tablet->deviceId); |
| existingReq.timestampsList.emplace_back( |
| move(SessionUtils::getTime(*tablet))); |
| existingReq.valuesList.emplace_back(move(SessionUtils::getValue(*tablet))); |
| existingReq.sizeList.emplace_back(tablet->rowSize); |
| vector<int> dataTypes; |
| vector<string> measurements; |
| for (pair<string, TSDataType::TSDataType> schema : tablet->schemas) { |
| measurements.push_back(schema.first); |
| dataTypes.push_back(schema.second); |
| } |
| existingReq.measurementsList.emplace_back(measurements); |
| existingReq.typesList.emplace_back(dataTypes); |
| } |
| |
| std::function<void(std::shared_ptr<SessionConnection>, |
| const TSInsertTabletsReq &)> |
| consumer = [](const std::shared_ptr<SessionConnection> &c, |
| const TSInsertTabletsReq &r) { c->insertTablets(r); }; |
| if (tabletsGroup.size() == 1) { |
| insertOnce(tabletsGroup, consumer); |
| } else { |
| insertByGroup(tabletsGroup, consumer); |
| } |
| } |
| |
| void Session::open() { open(false, Impl::DEFAULT_TIMEOUT_MS); } |
| |
| void Session::open(bool enableRPCCompression) { |
| open(enableRPCCompression, Impl::DEFAULT_TIMEOUT_MS); |
| } |
| |
| void Session::open(bool enableRPCCompression, int connectionTimeoutInMs) { |
| if (!impl_->isClosed_) { |
| return; |
| } |
| |
| try { |
| impl_->initDefaultSessionConnection(); |
| } catch (const exception &e) { |
| log_debug(e.what()); |
| throw IoTDBException(e.what()); |
| } |
| impl_->zoneId_ = impl_->defaultSessionConnection_->zoneId; |
| |
| if (impl_->enableRedirection_) { |
| impl_->endPointToSessionConnection.insert( |
| make_pair(impl_->defaultEndPoint_, impl_->defaultSessionConnection_)); |
| } |
| |
| impl_->isClosed_ = false; |
| } |
| |
| void Session::close() { |
| if (impl_->isClosed_) { |
| return; |
| } |
| impl_->isClosed_ = true; |
| } |
| |
| void Session::insertRecord(const string &deviceId, int64_t time, |
| const vector<string> &measurements, |
| const vector<string> &values) { |
| TSInsertStringRecordReq req; |
| req.__set_prefixPath(deviceId); |
| req.__set_timestamp(time); |
| req.__set_measurements(measurements); |
| req.__set_values(values); |
| req.__set_isAligned(false); |
| try { |
| impl_->getSessionConnection(deviceId)->insertStringRecord(req); |
| } catch (RedirectException &e) { |
| impl_->handleRedirection(deviceId, endpointToThrift(e.endPoint)); |
| } catch (const IoTDBConnectionException &e) { |
| if (impl_->enableRedirection_ && impl_->deviceIdToEndpoint.find(deviceId) != |
| impl_->deviceIdToEndpoint.end()) { |
| impl_->deviceIdToEndpoint.erase(deviceId); |
| try { |
| impl_->defaultSessionConnection_->insertStringRecord(req); |
| } catch (RedirectException &e) { |
| } |
| } else { |
| throw; |
| } |
| } |
| } |
| |
| void Session::insertRecord(const string &deviceId, int64_t time, |
| const vector<string> &measurements, |
| const vector<TSDataType::TSDataType> &types, |
| const vector<char *> &values) { |
| TSInsertRecordReq req; |
| req.__set_prefixPath(deviceId); |
| req.__set_timestamp(time); |
| req.__set_measurements(measurements); |
| string buffer; |
| impl_->putValuesIntoBuffer(types, values, buffer); |
| req.__set_values(buffer); |
| req.__set_isAligned(false); |
| try { |
| impl_->getSessionConnection(deviceId)->insertRecord(req); |
| } catch (RedirectException &e) { |
| impl_->handleRedirection(deviceId, endpointToThrift(e.endPoint)); |
| } catch (const IoTDBConnectionException &e) { |
| if (impl_->enableRedirection_ && impl_->deviceIdToEndpoint.find(deviceId) != |
| impl_->deviceIdToEndpoint.end()) { |
| impl_->deviceIdToEndpoint.erase(deviceId); |
| try { |
| impl_->defaultSessionConnection_->insertRecord(req); |
| } catch (RedirectException &e) { |
| } |
| } else { |
| throw; |
| } |
| } |
| } |
| |
| void Session::insertAlignedRecord(const string &deviceId, int64_t time, |
| const vector<string> &measurements, |
| const vector<string> &values) { |
| TSInsertStringRecordReq req; |
| req.__set_prefixPath(deviceId); |
| req.__set_timestamp(time); |
| req.__set_measurements(measurements); |
| req.__set_values(values); |
| req.__set_isAligned(true); |
| try { |
| impl_->getSessionConnection(deviceId)->insertStringRecord(req); |
| } catch (RedirectException &e) { |
| impl_->handleRedirection(deviceId, endpointToThrift(e.endPoint)); |
| } catch (const IoTDBConnectionException &e) { |
| if (impl_->enableRedirection_ && impl_->deviceIdToEndpoint.find(deviceId) != |
| impl_->deviceIdToEndpoint.end()) { |
| impl_->deviceIdToEndpoint.erase(deviceId); |
| try { |
| impl_->defaultSessionConnection_->insertStringRecord(req); |
| } catch (RedirectException &e) { |
| } |
| } else { |
| throw; |
| } |
| } |
| } |
| |
| void Session::insertAlignedRecord(const string &deviceId, int64_t time, |
| const vector<string> &measurements, |
| const vector<TSDataType::TSDataType> &types, |
| const vector<char *> &values) { |
| TSInsertRecordReq req; |
| req.__set_prefixPath(deviceId); |
| req.__set_timestamp(time); |
| req.__set_measurements(measurements); |
| string buffer; |
| impl_->putValuesIntoBuffer(types, values, buffer); |
| req.__set_values(buffer); |
| req.__set_isAligned(false); |
| try { |
| impl_->getSessionConnection(deviceId)->insertRecord(req); |
| } catch (RedirectException &e) { |
| impl_->handleRedirection(deviceId, endpointToThrift(e.endPoint)); |
| } catch (const IoTDBConnectionException &e) { |
| if (impl_->enableRedirection_ && impl_->deviceIdToEndpoint.find(deviceId) != |
| impl_->deviceIdToEndpoint.end()) { |
| impl_->deviceIdToEndpoint.erase(deviceId); |
| try { |
| impl_->defaultSessionConnection_->insertRecord(req); |
| } catch (RedirectException &e) { |
| } |
| } else { |
| throw; |
| } |
| } |
| } |
| |
| void Session::insertRecords(const vector<string> &deviceIds, |
| const vector<int64_t> ×, |
| const vector<vector<string>> &measurementsList, |
| const vector<vector<string>> &valuesList) { |
| size_t 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); |
| } |
| |
| if (impl_->enableRedirection_) { |
| impl_->insertStringRecordsWithLeaderCache( |
| deviceIds, times, measurementsList, valuesList, false); |
| } else { |
| TSInsertStringRecordsReq request; |
| request.__set_prefixPaths(deviceIds); |
| request.__set_timestamps(times); |
| request.__set_measurementsList(measurementsList); |
| request.__set_valuesList(valuesList); |
| request.__set_isAligned(false); |
| try { |
| impl_->defaultSessionConnection_->insertStringRecords(request); |
| } catch (RedirectException &e) { |
| } |
| } |
| } |
| |
| void Session::insertRecords( |
| const vector<string> &deviceIds, const vector<int64_t> ×, |
| const vector<vector<string>> &measurementsList, |
| const vector<vector<TSDataType::TSDataType>> &typesList, |
| const vector<vector<char *>> &valuesList) { |
| size_t 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); |
| } |
| |
| if (impl_->enableRedirection_) { |
| impl_->insertRecordsWithLeaderCache(deviceIds, times, measurementsList, |
| typesList, valuesList, false); |
| } else { |
| TSInsertRecordsReq request; |
| request.__set_prefixPaths(deviceIds); |
| request.__set_timestamps(times); |
| request.__set_measurementsList(measurementsList); |
| vector<string> bufferList; |
| for (size_t i = 0; i < valuesList.size(); i++) { |
| string buffer; |
| impl_->putValuesIntoBuffer(typesList[i], valuesList[i], buffer); |
| bufferList.push_back(buffer); |
| } |
| request.__set_valuesList(bufferList); |
| request.__set_isAligned(false); |
| try { |
| impl_->defaultSessionConnection_->insertRecords(request); |
| } catch (RedirectException &e) { |
| } |
| } |
| } |
| |
| void Session::insertAlignedRecords( |
| const vector<string> &deviceIds, const vector<int64_t> ×, |
| const vector<vector<string>> &measurementsList, |
| const vector<vector<string>> &valuesList) { |
| size_t 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); |
| } |
| |
| if (impl_->enableRedirection_) { |
| impl_->insertStringRecordsWithLeaderCache( |
| deviceIds, times, measurementsList, valuesList, true); |
| } else { |
| TSInsertStringRecordsReq request; |
| request.__set_prefixPaths(deviceIds); |
| request.__set_timestamps(times); |
| request.__set_measurementsList(measurementsList); |
| request.__set_valuesList(valuesList); |
| request.__set_isAligned(true); |
| try { |
| impl_->defaultSessionConnection_->insertStringRecords(request); |
| } catch (RedirectException &e) { |
| } |
| } |
| } |
| |
| void Session::insertAlignedRecords( |
| const vector<string> &deviceIds, const vector<int64_t> ×, |
| const vector<vector<string>> &measurementsList, |
| const vector<vector<TSDataType::TSDataType>> &typesList, |
| const vector<vector<char *>> &valuesList) { |
| size_t 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); |
| } |
| |
| if (impl_->enableRedirection_) { |
| impl_->insertRecordsWithLeaderCache(deviceIds, times, measurementsList, |
| typesList, valuesList, true); |
| } else { |
| TSInsertRecordsReq request; |
| request.__set_prefixPaths(deviceIds); |
| request.__set_timestamps(times); |
| request.__set_measurementsList(measurementsList); |
| vector<string> bufferList; |
| for (size_t i = 0; i < valuesList.size(); i++) { |
| string buffer; |
| impl_->putValuesIntoBuffer(typesList[i], valuesList[i], buffer); |
| bufferList.push_back(buffer); |
| } |
| request.__set_valuesList(bufferList); |
| request.__set_isAligned(false); |
| try { |
| impl_->defaultSessionConnection_->insertRecords(request); |
| } catch (RedirectException &e) { |
| } |
| } |
| } |
| |
| void Session::insertRecordsOfOneDevice( |
| const 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( |
| const string &deviceId, vector<int64_t> ×, |
| vector<vector<string>> &measurementsList, |
| vector<vector<TSDataType::TSDataType>> &typesList, |
| vector<vector<char *>> &valuesList, bool sorted) { |
| if (!impl_->checkSorted(times)) { |
| int *index = new int[times.size()]; |
| for (size_t i = 0; i < times.size(); i++) { |
| index[i] = (int)i; |
| } |
| |
| impl_->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())); |
| valuesList = sortList(valuesList, index, (int)(times.size())); |
| delete[] index; |
| } |
| TSInsertRecordsOfOneDeviceReq request; |
| request.__set_prefixPath(deviceId); |
| request.__set_timestamps(times); |
| request.__set_measurementsList(measurementsList); |
| vector<string> bufferList; |
| for (size_t i = 0; i < valuesList.size(); i++) { |
| string buffer; |
| impl_->putValuesIntoBuffer(typesList[i], valuesList[i], buffer); |
| bufferList.push_back(buffer); |
| } |
| request.__set_valuesList(bufferList); |
| request.__set_isAligned(false); |
| TSStatus respStatus; |
| try { |
| impl_->getSessionConnection(deviceId)->insertRecordsOfOneDevice(request); |
| } catch (RedirectException &e) { |
| impl_->handleRedirection(deviceId, endpointToThrift(e.endPoint)); |
| } catch (const IoTDBConnectionException &e) { |
| if (impl_->enableRedirection_ && impl_->deviceIdToEndpoint.find(deviceId) != |
| impl_->deviceIdToEndpoint.end()) { |
| impl_->deviceIdToEndpoint.erase(deviceId); |
| try { |
| impl_->defaultSessionConnection_->insertRecordsOfOneDevice(request); |
| } catch (RedirectException &e) { |
| } |
| } else { |
| throw; |
| } |
| } |
| } |
| |
| void Session::insertAlignedRecordsOfOneDevice( |
| const string &deviceId, vector<int64_t> ×, |
| vector<vector<string>> &measurementsList, |
| vector<vector<TSDataType::TSDataType>> &typesList, |
| vector<vector<char *>> &valuesList) { |
| insertAlignedRecordsOfOneDevice(deviceId, times, measurementsList, typesList, |
| valuesList, false); |
| } |
| |
| void Session::insertAlignedRecordsOfOneDevice( |
| const string &deviceId, vector<int64_t> ×, |
| vector<vector<string>> &measurementsList, |
| vector<vector<TSDataType::TSDataType>> &typesList, |
| vector<vector<char *>> &valuesList, bool sorted) { |
| if (!impl_->checkSorted(times)) { |
| int *index = new int[times.size()]; |
| for (size_t i = 0; i < times.size(); i++) { |
| index[i] = (int)i; |
| } |
| |
| impl_->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())); |
| valuesList = sortList(valuesList, index, (int)(times.size())); |
| delete[] index; |
| } |
| TSInsertRecordsOfOneDeviceReq request; |
| request.__set_prefixPath(deviceId); |
| request.__set_timestamps(times); |
| request.__set_measurementsList(measurementsList); |
| vector<string> bufferList; |
| for (size_t i = 0; i < valuesList.size(); i++) { |
| string buffer; |
| impl_->putValuesIntoBuffer(typesList[i], valuesList[i], buffer); |
| bufferList.push_back(buffer); |
| } |
| request.__set_valuesList(bufferList); |
| request.__set_isAligned(true); |
| TSStatus respStatus; |
| try { |
| impl_->getSessionConnection(deviceId)->insertRecordsOfOneDevice(request); |
| } catch (RedirectException &e) { |
| impl_->handleRedirection(deviceId, endpointToThrift(e.endPoint)); |
| } catch (const IoTDBConnectionException &e) { |
| if (impl_->enableRedirection_ && impl_->deviceIdToEndpoint.find(deviceId) != |
| impl_->deviceIdToEndpoint.end()) { |
| impl_->deviceIdToEndpoint.erase(deviceId); |
| try { |
| impl_->defaultSessionConnection_->insertRecordsOfOneDevice(request); |
| } catch (RedirectException &e) { |
| } |
| } else { |
| throw; |
| } |
| } |
| } |
| |
| void Session::insertTablet(Tablet &tablet) { |
| try { |
| insertTablet(tablet, false); |
| } catch (const exception &e) { |
| log_debug(e.what()); |
| logic_error error(e.what()); |
| throw exception(error); |
| } |
| } |
| |
| void Session::Impl::buildInsertTabletReq(TSInsertTabletReq &request, |
| Tablet &tablet, bool sorted) { |
| if ((!sorted) && !checkSorted(tablet)) { |
| sortTablet(tablet); |
| } |
| |
| request.__set_prefixPath(tablet.deviceId); |
| |
| std::vector<std::string> reqMeasurements; |
| reqMeasurements.reserve(tablet.schemas.size()); |
| std::vector<int32_t> types; |
| types.reserve(tablet.schemas.size()); |
| for (pair<string, TSDataType::TSDataType> schema : tablet.schemas) { |
| reqMeasurements.push_back(schema.first); |
| types.push_back(schema.second); |
| } |
| request.__set_measurements(reqMeasurements); |
| request.__set_types(types); |
| request.__set_values(SessionUtils::getValue(tablet)); |
| request.__set_timestamps(SessionUtils::getTime(tablet)); |
| request.__set_size(tablet.rowSize); |
| request.__set_isAligned(tablet.isAligned); |
| } |
| |
| void Session::Impl::insertTablet(TSInsertTabletReq request) { |
| auto deviceId = request.prefixPath; |
| try { |
| getSessionConnection(deviceId)->insertTablet(request); |
| } catch (RedirectException &e) { |
| handleRedirection(deviceId, endpointToThrift(e.endPoint)); |
| } catch (const IoTDBConnectionException &e) { |
| if (enableRedirection_ && |
| deviceIdToEndpoint.find(deviceId) != deviceIdToEndpoint.end()) { |
| deviceIdToEndpoint.erase(deviceId); |
| try { |
| defaultSessionConnection_->insertTablet(request); |
| } catch (RedirectException &e) { |
| } |
| } else { |
| throw; |
| } |
| } |
| } |
| |
| void Session::insertTablet(Tablet &tablet, bool sorted) { |
| TSInsertTabletReq request; |
| impl_->buildInsertTabletReq(request, tablet, sorted); |
| impl_->insertTablet(request); |
| } |
| |
| void Session::insertAlignedTablet(Tablet &tablet) { |
| insertAlignedTablet(tablet, false); |
| } |
| |
| void Session::insertAlignedTablet(Tablet &tablet, bool sorted) { |
| tablet.setAligned(true); |
| try { |
| insertTablet(tablet, sorted); |
| } catch (const exception &e) { |
| log_debug(e.what()); |
| logic_error error(e.what()); |
| throw exception(error); |
| } |
| } |
| |
| void Session::insertTablets(unordered_map<string, Tablet *> &tablets) { |
| try { |
| insertTablets(tablets, false); |
| } catch (const exception &e) { |
| log_debug(e.what()); |
| logic_error error(e.what()); |
| throw exception(error); |
| } |
| } |
| |
| void Session::insertTablets(unordered_map<string, Tablet *> &tablets, |
| bool sorted) { |
| if (tablets.empty()) { |
| throw BatchExecutionException("No tablet is inserting!"); |
| } |
| auto beginIter = tablets.begin(); |
| bool isAligned = ((*beginIter).second)->isAligned; |
| if (impl_->enableRedirection_) { |
| impl_->insertTabletsWithLeaderCache(tablets, sorted, isAligned); |
| } else { |
| TSInsertTabletsReq request; |
| for (const auto &item : tablets) { |
| if (isAligned != item.second->isAligned) { |
| throw BatchExecutionException( |
| "The tablets should be all aligned or non-aligned!"); |
| } |
| if (!impl_->checkSorted(*(item.second))) { |
| impl_->sortTablet(*(item.second)); |
| } |
| request.prefixPaths.push_back(item.second->deviceId); |
| vector<string> measurements; |
| vector<int> dataTypes; |
| for (pair<string, TSDataType::TSDataType> schema : item.second->schemas) { |
| measurements.push_back(schema.first); |
| dataTypes.push_back(schema.second); |
| } |
| request.measurementsList.push_back(measurements); |
| request.typesList.push_back(dataTypes); |
| 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(isAligned); |
| try { |
| TSStatus respStatus; |
| impl_->defaultSessionConnection_->insertTablets(request); |
| RpcUtils::verifySuccess(respStatus); |
| } catch (RedirectException &e) { |
| } |
| } |
| } |
| |
| void Session::insertAlignedTablets(unordered_map<string, Tablet *> &tablets, |
| bool sorted) { |
| for (auto iter = tablets.begin(); iter != tablets.end(); iter++) { |
| iter->second->setAligned(true); |
| } |
| try { |
| insertTablets(tablets, sorted); |
| } catch (const exception &e) { |
| log_debug(e.what()); |
| logic_error error(e.what()); |
| throw exception(error); |
| } |
| } |
| |
| void Session::testInsertRecord(const string &deviceId, int64_t time, |
| const vector<string> &measurements, |
| const vector<string> &values) { |
| TSInsertStringRecordReq req; |
| req.__set_prefixPath(deviceId); |
| req.__set_timestamp(time); |
| req.__set_measurements(measurements); |
| req.__set_values(values); |
| TSStatus tsStatus; |
| try { |
| impl_->defaultSessionConnection_->testInsertStringRecord(req); |
| RpcUtils::verifySuccess(tsStatus); |
| } catch (const TTransportException &e) { |
| log_debug(e.what()); |
| throw IoTDBConnectionException(e.what()); |
| } catch (const IoTDBException &e) { |
| log_debug(e.what()); |
| throw; |
| } catch (const exception &e) { |
| log_debug(e.what()); |
| throw IoTDBException(e.what()); |
| } |
| } |
| |
| void Session::testInsertTablet(const Tablet &tablet) { |
| TSInsertTabletReq request; |
| request.prefixPath = tablet.deviceId; |
| for (pair<string, TSDataType::TSDataType> schema : tablet.schemas) { |
| request.measurements.push_back(schema.first); |
| request.types.push_back(schema.second); |
| } |
| request.__set_timestamps(move(SessionUtils::getTime(tablet))); |
| request.__set_values(move(SessionUtils::getValue(tablet))); |
| request.__set_size(tablet.rowSize); |
| try { |
| TSStatus tsStatus; |
| impl_->defaultSessionConnection_->testInsertTablet(request); |
| RpcUtils::verifySuccess(tsStatus); |
| } catch (const TTransportException &e) { |
| log_debug(e.what()); |
| throw IoTDBConnectionException(e.what()); |
| } catch (const IoTDBException &e) { |
| log_debug(e.what()); |
| throw; |
| } catch (const exception &e) { |
| log_debug(e.what()); |
| throw IoTDBException(e.what()); |
| } |
| } |
| |
| void Session::testInsertRecords(const vector<string> &deviceIds, |
| const vector<int64_t> ×, |
| const vector<vector<string>> &measurementsList, |
| const vector<vector<string>> &valuesList) { |
| size_t 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); |
| } |
| TSInsertStringRecordsReq request; |
| request.__set_prefixPaths(deviceIds); |
| request.__set_timestamps(times); |
| request.__set_measurementsList(measurementsList); |
| request.__set_valuesList(valuesList); |
| |
| try { |
| TSStatus tsStatus; |
| impl_->defaultSessionConnection_->getSessionClient()->insertStringRecords( |
| tsStatus, request); |
| RpcUtils::verifySuccess(tsStatus); |
| } catch (const TTransportException &e) { |
| log_debug(e.what()); |
| throw IoTDBConnectionException(e.what()); |
| } catch (const IoTDBException &e) { |
| log_debug(e.what()); |
| throw; |
| } catch (const exception &e) { |
| log_debug(e.what()); |
| throw IoTDBException(e.what()); |
| } |
| } |
| |
| void Session::deleteTimeseries(const string &path) { |
| vector<string> paths; |
| paths.push_back(path); |
| deleteTimeseries(paths); |
| } |
| |
| void Session::deleteTimeseries(const vector<string> &paths) { |
| impl_->defaultSessionConnection_->deleteTimeseries(paths); |
| } |
| |
| void Session::deleteData(const string &path, int64_t endTime) { |
| vector<string> paths; |
| paths.push_back(path); |
| deleteData(paths, LONG_LONG_MIN, endTime); |
| } |
| |
| void Session::deleteData(const vector<string> &paths, int64_t endTime) { |
| deleteData(paths, LONG_LONG_MIN, endTime); |
| } |
| |
| void Session::deleteData(const vector<string> &paths, int64_t startTime, |
| int64_t endTime) { |
| TSDeleteDataReq req; |
| req.__set_paths(paths); |
| req.__set_startTime(startTime); |
| req.__set_endTime(endTime); |
| impl_->defaultSessionConnection_->deleteData(req); |
| } |
| |
| void Session::setStorageGroup(const string &storageGroupId) { |
| impl_->defaultSessionConnection_->setStorageGroup(storageGroupId); |
| } |
| |
| void Session::deleteStorageGroup(const string &storageGroup) { |
| vector<string> storageGroups; |
| storageGroups.push_back(storageGroup); |
| deleteStorageGroups(storageGroups); |
| } |
| |
| void Session::deleteStorageGroups(const vector<string> &storageGroups) { |
| impl_->defaultSessionConnection_->deleteStorageGroups(storageGroups); |
| } |
| |
| void Session::createDatabase(const string &database) { |
| this->setStorageGroup(database); |
| } |
| |
| void Session::deleteDatabase(const string &database) { |
| this->deleteStorageGroups(vector<string>{database}); |
| } |
| |
| void Session::deleteDatabases(const vector<string> &databases) { |
| this->deleteStorageGroups(databases); |
| } |
| |
| void Session::createTimeseries(const string &path, |
| TSDataType::TSDataType dataType, |
| TSEncoding::TSEncoding encoding, |
| CompressionType::CompressionType compressor) { |
| try { |
| createTimeseries(path, dataType, encoding, compressor, nullptr, nullptr, |
| nullptr, ""); |
| } catch (const exception &e) { |
| log_debug(e.what()); |
| throw IoTDBException(e.what()); |
| } |
| } |
| |
| void Session::createTimeseries(const string &path, |
| TSDataType::TSDataType dataType, |
| TSEncoding::TSEncoding encoding, |
| CompressionType::CompressionType compressor, |
| map<string, string> *props, |
| map<string, string> *tags, |
| map<string, string> *attributes, |
| const string &measurementAlias) { |
| TSCreateTimeseriesReq req; |
| req.__set_path(path); |
| req.__set_dataType(dataType); |
| req.__set_encoding(encoding); |
| req.__set_compressor(compressor); |
| if (props != nullptr) { |
| req.__set_props(*props); |
| } |
| |
| if (tags != nullptr) { |
| req.__set_tags(*tags); |
| } |
| if (attributes != nullptr) { |
| req.__set_attributes(*attributes); |
| } |
| if (!measurementAlias.empty()) { |
| req.__set_measurementAlias(measurementAlias); |
| } |
| impl_->defaultSessionConnection_->createTimeseries(req); |
| } |
| |
| void Session::createMultiTimeseries( |
| const vector<string> &paths, |
| const vector<TSDataType::TSDataType> &dataTypes, |
| const vector<TSEncoding::TSEncoding> &encodings, |
| const vector<CompressionType::CompressionType> &compressors, |
| vector<map<string, string>> *propsList, |
| vector<map<string, string>> *tagsList, |
| vector<map<string, string>> *attributesList, |
| vector<string> *measurementAliasList) { |
| TSCreateMultiTimeseriesReq request; |
| request.__set_paths(paths); |
| |
| vector<int> dataTypesOrdinal; |
| dataTypesOrdinal.reserve(dataTypes.size()); |
| for (TSDataType::TSDataType dataType : dataTypes) { |
| dataTypesOrdinal.push_back(dataType); |
| } |
| request.__set_dataTypes(dataTypesOrdinal); |
| |
| vector<int> encodingsOrdinal; |
| encodingsOrdinal.reserve(encodings.size()); |
| for (TSEncoding::TSEncoding encoding : encodings) { |
| encodingsOrdinal.push_back(encoding); |
| } |
| request.__set_encodings(encodingsOrdinal); |
| |
| vector<int> compressorsOrdinal; |
| compressorsOrdinal.reserve(compressors.size()); |
| for (CompressionType::CompressionType compressor : compressors) { |
| compressorsOrdinal.push_back(compressor); |
| } |
| request.__set_compressors(compressorsOrdinal); |
| |
| if (propsList != nullptr) { |
| request.__set_propsList(*propsList); |
| } |
| |
| if (tagsList != nullptr) { |
| request.__set_tagsList(*tagsList); |
| } |
| if (attributesList != nullptr) { |
| request.__set_attributesList(*attributesList); |
| } |
| if (measurementAliasList != nullptr) { |
| request.__set_measurementAliasList(*measurementAliasList); |
| } |
| |
| impl_->defaultSessionConnection_->createMultiTimeseries(request); |
| } |
| |
| void Session::createAlignedTimeseries( |
| const std::string &deviceId, const std::vector<std::string> &measurements, |
| const std::vector<TSDataType::TSDataType> &dataTypes, |
| const std::vector<TSEncoding::TSEncoding> &encodings, |
| const std::vector<CompressionType::CompressionType> &compressors) { |
| TSCreateAlignedTimeseriesReq request; |
| request.__set_prefixPath(deviceId); |
| request.__set_measurements(measurements); |
| |
| vector<int> dataTypesOrdinal; |
| dataTypesOrdinal.reserve(dataTypes.size()); |
| for (TSDataType::TSDataType dataType : dataTypes) { |
| dataTypesOrdinal.push_back(dataType); |
| } |
| request.__set_dataTypes(dataTypesOrdinal); |
| |
| vector<int> encodingsOrdinal; |
| encodingsOrdinal.reserve(encodings.size()); |
| for (TSEncoding::TSEncoding encoding : encodings) { |
| encodingsOrdinal.push_back(encoding); |
| } |
| request.__set_encodings(encodingsOrdinal); |
| |
| vector<int> compressorsOrdinal; |
| compressorsOrdinal.reserve(compressors.size()); |
| for (CompressionType::CompressionType compressor : compressors) { |
| compressorsOrdinal.push_back(compressor); |
| } |
| request.__set_compressors(compressorsOrdinal); |
| |
| impl_->defaultSessionConnection_->createAlignedTimeseries(request); |
| } |
| |
| bool Session::checkTimeseriesExists(const string &path) { |
| try { |
| std::unique_ptr<SessionDataSet> dataset = |
| executeQueryStatement("SHOW TIMESERIES " + path); |
| if (dataset == nullptr) { |
| throw IoTDBException("executeQueryStatement failed"); |
| } |
| bool isExisted = dataset->hasNext(); |
| dataset->closeOperationHandle(); |
| return isExisted; |
| } catch (const exception &e) { |
| log_debug(e.what()); |
| throw IoTDBException(e.what()); |
| } |
| } |
| |
| shared_ptr<SessionConnection> Session::Impl::getQuerySessionConnection() { |
| auto endPoint = nodesSupplier_->getQueryEndPoint(); |
| if (!endPoint.is_initialized() || endPointToSessionConnection.empty()) { |
| return defaultSessionConnection_; |
| } |
| |
| auto it = endPointToSessionConnection.find(endPoint.value()); |
| if (it != endPointToSessionConnection.end()) { |
| return it->second; |
| } |
| |
| shared_ptr<SessionConnection> newConnection; |
| try { |
| newConnection = make_shared<SessionConnection>( |
| this, endPoint.value(), zoneId_, nodesSupplier_, fetchSize_, 60, 500, |
| connectTimeoutMs_, sqlDialect_, database_); |
| endPointToSessionConnection.emplace(endPoint.value(), newConnection); |
| return newConnection; |
| } catch (exception &e) { |
| log_debug("Session::Impl::getQuerySessionConnection() exception: " + |
| e.what()); |
| return newConnection; |
| } |
| } |
| |
| shared_ptr<SessionConnection> |
| Session::Impl::getSessionConnection(std::string deviceId) { |
| if (!enableRedirection_ || |
| deviceIdToEndpoint.find(deviceId) == deviceIdToEndpoint.end() || |
| endPointToSessionConnection.find(deviceIdToEndpoint[deviceId]) == |
| endPointToSessionConnection.end()) { |
| return defaultSessionConnection_; |
| } |
| return endPointToSessionConnection.find(deviceIdToEndpoint[deviceId])->second; |
| } |
| |
| string Session::getTimeZone() { |
| auto ret = impl_->defaultSessionConnection_->getTimeZone(); |
| return ret.timeZone; |
| } |
| |
| void Session::setTimeZone(const string &zoneId) { |
| TSSetTimeZoneReq req; |
| req.__set_sessionId(impl_->defaultSessionConnection_->sessionId); |
| req.__set_timeZone(zoneId); |
| impl_->defaultSessionConnection_->setTimeZone(req); |
| } |
| |
| unique_ptr<SessionDataSet> Session::executeQueryStatement(const string &sql) { |
| return executeQueryStatementMayRedirect(sql, QUERY_TIMEOUT_MS); |
| } |
| |
| unique_ptr<SessionDataSet> Session::executeQueryStatement(const string &sql, |
| int64_t timeoutInMs) { |
| return executeQueryStatementMayRedirect(sql, timeoutInMs); |
| } |
| |
| void Session::Impl::handleQueryRedirection(TEndPoint endPoint) { |
| if (!enableRedirection_) |
| return; |
| shared_ptr<SessionConnection> newConnection; |
| auto it = endPointToSessionConnection.find(endPoint); |
| if (it != endPointToSessionConnection.end()) { |
| newConnection = it->second; |
| } else { |
| try { |
| newConnection = make_shared<SessionConnection>( |
| this, endPoint, zoneId_, nodesSupplier_, fetchSize_, 60, 500, |
| connectTimeoutMs_, sqlDialect_, database_); |
| |
| endPointToSessionConnection.emplace(endPoint, newConnection); |
| } catch (exception &e) { |
| throw IoTDBConnectionException(e.what()); |
| } |
| } |
| defaultSessionConnection_ = newConnection; |
| } |
| |
| void Session::Impl::handleRedirection(const std::string &deviceId, |
| TEndPoint endPoint) { |
| if (!enableRedirection_) |
| return; |
| if (endPoint.ip == "0.0.0.0") |
| return; |
| deviceIdToEndpoint[deviceId] = endPoint; |
| |
| shared_ptr<SessionConnection> newConnection; |
| auto it = endPointToSessionConnection.find(endPoint); |
| if (it != endPointToSessionConnection.end()) { |
| newConnection = it->second; |
| } else { |
| try { |
| newConnection = make_shared<SessionConnection>( |
| this, endPoint, zoneId_, nodesSupplier_, fetchSize_, 60, 500, 1000, |
| sqlDialect_, database_); |
| endPointToSessionConnection.emplace(endPoint, newConnection); |
| } catch (exception &e) { |
| deviceIdToEndpoint.erase(deviceId); |
| throw IoTDBConnectionException(e.what()); |
| } |
| } |
| } |
| |
| std::unique_ptr<SessionDataSet> |
| Session::executeQueryStatementMayRedirect(const std::string &sql, |
| int64_t timeoutInMs) { |
| auto sessionConnection = impl_->getQuerySessionConnection(); |
| if (!sessionConnection) { |
| log_warn("Session connection not found"); |
| return nullptr; |
| } |
| try { |
| return sessionConnection->executeQueryStatement(sql, timeoutInMs); |
| } catch (RedirectException &e) { |
| log_warn("Session connection redirect exception: " + e.what()); |
| impl_->handleQueryRedirection(endpointToThrift(e.endPoint)); |
| try { |
| return impl_->defaultSessionConnection_->executeQueryStatement( |
| sql, timeoutInMs); |
| } catch (exception &e) { |
| log_error("Exception while executing redirected query statement: %s", |
| e.what()); |
| throw ExecutionException(e.what()); |
| } |
| } catch (exception &e) { |
| log_error("Exception while executing query statement: %s", e.what()); |
| throw; |
| } |
| } |
| |
| void Session::executeNonQueryStatement(const string &sql) { |
| try { |
| impl_->defaultSessionConnection_->executeNonQueryStatement(sql); |
| } catch (const exception &e) { |
| throw IoTDBException(e.what()); |
| } |
| } |
| |
| unique_ptr<SessionDataSet> |
| Session::executeRawDataQuery(const vector<string> &paths, int64_t startTime, |
| int64_t endTime) { |
| return impl_->defaultSessionConnection_->executeRawDataQuery(paths, startTime, |
| endTime); |
| } |
| |
| unique_ptr<SessionDataSet> |
| Session::executeLastDataQuery(const vector<string> &paths) { |
| return executeLastDataQuery(paths, LONG_LONG_MIN); |
| } |
| |
| unique_ptr<SessionDataSet> |
| Session::executeLastDataQuery(const vector<string> &paths, int64_t lastTime) { |
| return impl_->defaultSessionConnection_->executeLastDataQuery(paths, |
| lastTime); |
| } |
| |
| void Session::createSchemaTemplate(const Template &templ) { |
| TSCreateSchemaTemplateReq req; |
| req.__set_name(templ.getName()); |
| req.__set_serializedTemplate(templ.serialize()); |
| impl_->defaultSessionConnection_->createSchemaTemplate(req); |
| } |
| |
| void Session::setSchemaTemplate(const string &template_name, |
| const string &prefix_path) { |
| TSSetSchemaTemplateReq req; |
| req.__set_templateName(template_name); |
| req.__set_prefixPath(prefix_path); |
| impl_->defaultSessionConnection_->setSchemaTemplate(req); |
| } |
| |
| void Session::unsetSchemaTemplate(const string &prefix_path, |
| const string &template_name) { |
| TSUnsetSchemaTemplateReq req; |
| req.__set_templateName(template_name); |
| req.__set_prefixPath(prefix_path); |
| impl_->defaultSessionConnection_->unsetSchemaTemplate(req); |
| } |
| |
| void Session::addAlignedMeasurementsInTemplate( |
| const string &template_name, const vector<std::string> &measurements, |
| const vector<TSDataType::TSDataType> &dataTypes, |
| const vector<TSEncoding::TSEncoding> &encodings, |
| const vector<CompressionType::CompressionType> &compressors) { |
| TSAppendSchemaTemplateReq req; |
| req.__set_name(template_name); |
| req.__set_measurements(measurements); |
| req.__set_isAligned(true); |
| |
| vector<int> dataTypesOrdinal; |
| dataTypesOrdinal.reserve(dataTypes.size()); |
| for (TSDataType::TSDataType dataType : dataTypes) { |
| dataTypesOrdinal.push_back(dataType); |
| } |
| req.__set_dataTypes(dataTypesOrdinal); |
| |
| vector<int> encodingsOrdinal; |
| encodingsOrdinal.reserve(encodings.size()); |
| for (TSEncoding::TSEncoding encoding : encodings) { |
| encodingsOrdinal.push_back(encoding); |
| } |
| req.__set_encodings(encodingsOrdinal); |
| |
| vector<int> compressorsOrdinal; |
| compressorsOrdinal.reserve(compressors.size()); |
| for (CompressionType::CompressionType compressor : compressors) { |
| compressorsOrdinal.push_back(compressor); |
| } |
| req.__set_compressors(compressorsOrdinal); |
| |
| impl_->defaultSessionConnection_->appendSchemaTemplate(req); |
| } |
| |
| void Session::addAlignedMeasurementsInTemplate( |
| const string &template_name, const string &measurement, |
| TSDataType::TSDataType dataType, TSEncoding::TSEncoding encoding, |
| CompressionType::CompressionType compressor) { |
| vector<std::string> measurements(1, measurement); |
| vector<TSDataType::TSDataType> dataTypes(1, dataType); |
| vector<TSEncoding::TSEncoding> encodings(1, encoding); |
| vector<CompressionType::CompressionType> compressors(1, compressor); |
| addAlignedMeasurementsInTemplate(template_name, measurements, dataTypes, |
| encodings, compressors); |
| } |
| |
| void Session::addUnalignedMeasurementsInTemplate( |
| const string &template_name, const vector<std::string> &measurements, |
| const vector<TSDataType::TSDataType> &dataTypes, |
| const vector<TSEncoding::TSEncoding> &encodings, |
| const vector<CompressionType::CompressionType> &compressors) { |
| TSAppendSchemaTemplateReq req; |
| req.__set_name(template_name); |
| req.__set_measurements(measurements); |
| req.__set_isAligned(false); |
| |
| vector<int> dataTypesOrdinal; |
| dataTypesOrdinal.reserve(dataTypes.size()); |
| for (TSDataType::TSDataType dataType : dataTypes) { |
| dataTypesOrdinal.push_back(dataType); |
| } |
| req.__set_dataTypes(dataTypesOrdinal); |
| |
| vector<int> encodingsOrdinal; |
| encodingsOrdinal.reserve(encodings.size()); |
| for (TSEncoding::TSEncoding encoding : encodings) { |
| encodingsOrdinal.push_back(encoding); |
| } |
| req.__set_encodings(encodingsOrdinal); |
| |
| vector<int> compressorsOrdinal; |
| compressorsOrdinal.reserve(compressors.size()); |
| for (CompressionType::CompressionType compressor : compressors) { |
| compressorsOrdinal.push_back(compressor); |
| } |
| req.__set_compressors(compressorsOrdinal); |
| |
| impl_->defaultSessionConnection_->appendSchemaTemplate(req); |
| } |
| |
| void Session::addUnalignedMeasurementsInTemplate( |
| const string &template_name, const string &measurement, |
| TSDataType::TSDataType dataType, TSEncoding::TSEncoding encoding, |
| CompressionType::CompressionType compressor) { |
| vector<std::string> measurements(1, measurement); |
| vector<TSDataType::TSDataType> dataTypes(1, dataType); |
| vector<TSEncoding::TSEncoding> encodings(1, encoding); |
| vector<CompressionType::CompressionType> compressors(1, compressor); |
| addUnalignedMeasurementsInTemplate(template_name, measurements, dataTypes, |
| encodings, compressors); |
| } |
| |
| void Session::deleteNodeInTemplate(const string &template_name, |
| const string &path) { |
| TSPruneSchemaTemplateReq req; |
| req.__set_name(template_name); |
| req.__set_path(path); |
| impl_->defaultSessionConnection_->pruneSchemaTemplate(req); |
| } |
| |
| int Session::countMeasurementsInTemplate(const string &template_name) { |
| TSQueryTemplateReq req; |
| req.__set_name(template_name); |
| req.__set_queryType(TemplateQueryType::COUNT_MEASUREMENTS); |
| TSQueryTemplateResp resp = |
| impl_->defaultSessionConnection_->querySchemaTemplate(req); |
| return resp.count; |
| } |
| |
| bool Session::isMeasurementInTemplate(const string &template_name, |
| const string &path) { |
| TSQueryTemplateReq req; |
| req.__set_name(template_name); |
| req.__set_measurement(path); |
| req.__set_queryType(TemplateQueryType::IS_MEASUREMENT); |
| TSQueryTemplateResp resp = |
| impl_->defaultSessionConnection_->querySchemaTemplate(req); |
| return resp.result; |
| } |
| |
| bool Session::isPathExistInTemplate(const string &template_name, |
| const string &path) { |
| TSQueryTemplateReq req; |
| req.__set_name(template_name); |
| req.__set_measurement(path); |
| req.__set_queryType(TemplateQueryType::PATH_EXIST); |
| TSQueryTemplateResp resp = |
| impl_->defaultSessionConnection_->querySchemaTemplate(req); |
| return resp.result; |
| } |
| |
| std::vector<std::string> |
| Session::showMeasurementsInTemplate(const string &template_name) { |
| TSQueryTemplateReq req; |
| req.__set_name(template_name); |
| req.__set_measurement(""); |
| req.__set_queryType(TemplateQueryType::SHOW_MEASUREMENTS); |
| TSQueryTemplateResp resp = |
| impl_->defaultSessionConnection_->querySchemaTemplate(req); |
| return resp.measurements; |
| } |
| |
| std::vector<std::string> |
| Session::showMeasurementsInTemplate(const string &template_name, |
| const string &pattern) { |
| TSQueryTemplateReq req; |
| req.__set_name(template_name); |
| req.__set_measurement(pattern); |
| req.__set_queryType(TemplateQueryType::SHOW_MEASUREMENTS); |
| TSQueryTemplateResp resp = |
| impl_->defaultSessionConnection_->querySchemaTemplate(req); |
| return resp.measurements; |
| } |
| |
| bool Session::checkTemplateExists(const string &template_name) { |
| try { |
| std::unique_ptr<SessionDataSet> dataset = |
| executeQueryStatement("SHOW NODES IN DEVICE TEMPLATE " + template_name); |
| if (dataset == nullptr) { |
| throw IoTDBException("executeQueryStatement failed"); |
| } |
| bool isExisted = dataset->hasNext(); |
| dataset->closeOperationHandle(); |
| return isExisted; |
| } catch (const exception &e) { |
| if (strstr(e.what(), "does not exist") != NULL) { |
| return false; |
| } |
| log_debug(e.what()); |
| throw IoTDBException(e.what()); |
| } |
| } |
