Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions iotdb-client/client-cpp/src/include/Session.h
Original file line number Diff line number Diff line change
Expand Up @@ -780,6 +780,11 @@ class Session {
std::unique_ptr<SessionDataSet> executeLastDataQuery(const std::vector<std::string>& paths,
int64_t lastTime);

// Query the latest value of each series below a tree-model prefix in a 1C1D deployment.
// Pass path nodes without wildcards; result rows are not ordered.
std::unique_ptr<SessionDataSet>
executeFastLastDataQueryForOnePrefixPath(const std::vector<std::string>& prefixes);

void createSchemaTemplate(const Template& templ);

void setSchemaTemplate(const std::string& template_name, const std::string& prefix_path);
Expand Down
38 changes: 38 additions & 0 deletions iotdb-client/client-cpp/src/rpc/SessionConnection.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
#include <thrift/transport/TSocket.h>
#include <thrift/transport/TTransportException.h>

#include <stdexcept>
#include <utility>

#include "SessionDataSet.h"
Expand Down Expand Up @@ -257,6 +258,43 @@ SessionConnection::executeLastDataQuery(const std::vector<std::string>& paths, i
zoneId, timeFactor, resp.columnIndex2TsBlockColumnIndexList);
}

std::unique_ptr<SessionDataSet> SessionConnection::executeFastLastDataQueryForOnePrefixPath(
const std::vector<std::string>& prefixes) {
for (const auto& prefix : prefixes) {
if (prefix.find('*') != std::string::npos) {
throw std::invalid_argument("Fast Last Query prefix must not contain wildcards");
}
}

TSFastLastDataQueryForOnePrefixPathReq req;
req.__set_prefixes(prefixes);
req.__set_fetchSize(fetchSize);
req.__set_enableRedirectQuery(session->enableRedirection_);

auto result = callWithRetryAndReconnect<TSExecuteStatementResp>(
[this, &req]() {
req.__set_sessionId(sessionId);
req.__set_statementId(statementId);
TSExecuteStatementResp resp;
client->executeFastLastDataQueryForOnePrefixPath(resp, req);
return resp;
},
[](const TSExecuteStatementResp& resp) { return resp.status; });
if (result.getException()) {
throw IoTDBConnectionException(extractExceptionMessage(result.getException()));
}
TSExecuteStatementResp resp = result.getResult();
if (result.getRetryAttempts() == 0) {
RpcUtils::verifySuccessWithRedirection(resp.status);
} else {
RpcUtils::verifySuccess(resp.status);
}
return createSessionDataSet("", resp.columns, resp.dataTypeList, resp.columnNameIndexMap,
resp.queryId, statementId, client, sessionId, resp.queryResult,
resp.ignoreTimeStamp, connectionTimeoutInMs, resp.moreData, fetchSize,
zoneId, timeFactor, resp.columnIndex2TsBlockColumnIndexList);
}

void SessionConnection::executeNonQueryStatement(const string& sql) {
TSExecuteStatementReq req;
req.__set_sessionId(sessionId);
Expand Down
3 changes: 3 additions & 0 deletions iotdb-client/client-cpp/src/rpc/SessionConnection.h
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,9 @@ class SessionConnection : public std::enable_shared_from_this<SessionConnection>
std::unique_ptr<SessionDataSet> executeLastDataQuery(const std::vector<std::string>& paths,
int64_t lastTime);

std::unique_ptr<SessionDataSet>
executeFastLastDataQueryForOnePrefixPath(const std::vector<std::string>& prefixes);

void executeNonQueryStatement(const std::string& sql);

std::unique_ptr<SessionDataSet> executeQueryStatement(const std::string& sql,
Expand Down
5 changes: 5 additions & 0 deletions iotdb-client/client-cpp/src/session/Session.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -2312,6 +2312,11 @@ unique_ptr<SessionDataSet> Session::executeLastDataQuery(const vector<string>& p
return impl_->getDefaultSessionConnection()->executeLastDataQuery(paths, lastTime);
}

unique_ptr<SessionDataSet>
Session::executeFastLastDataQueryForOnePrefixPath(const vector<string>& prefixes) {
return impl_->getDefaultSessionConnection()->executeFastLastDataQueryForOnePrefixPath(prefixes);
}

void Session::createSchemaTemplate(const Template& templ) {
TSCreateSchemaTemplateReq req;
req.__set_name(templ.getName());
Expand Down
47 changes: 47 additions & 0 deletions iotdb-client/client-cpp/test/cpp/sessionIT.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
#include "common_types.h"

#include <atomic>
#include <map>
#include <stdexcept>
#include <thread>

Expand Down Expand Up @@ -844,6 +845,52 @@ TEST_CASE("Test executeLastDataQuery ", "[testExecuteLastDataQuery]") {
REQUIRE(sessionDataSet->hasNext() == false);
}

TEST_CASE("Fast Last Query returns all series under a tree prefix", "[fastLastQuery]") {
const vector<string> paths = {"root.test.fast_last_cpp.d1.s1", "root.test.fast_last_cpp.d1.s2",
"root.test.fast_last_cpp.d2.s1"};
for (const auto& path : paths) {
if (session->checkTimeseriesExists(path)) {
session->deleteTimeseries(path);
}
session->createTimeseries(path, TSDataType::INT64, TSEncoding::RLE,
CompressionType::SNAPPY);
}
session->insertRecord("root.test.fast_last_cpp.d1", 20, {"s1", "s2"}, {"1", "2"});
session->insertRecord("root.test.fast_last_cpp.d1", 30, {"s1"}, {"3"});
session->insertRecord("root.test.fast_last_cpp.d2", 40, {"s1"}, {"4"});

const map<string, string> expected = {{paths[0], "30,3,INT64"},
{paths[1], "20,2,INT64"},
{paths[2], "40,4,INT64"}};
const vector<string> prefix = {"root", "test", "fast_last_cpp"};
for (int query = 0; query < 2; ++query) {
auto dataSet = session->executeFastLastDataQueryForOnePrefixPath(prefix);
map<string, string> actual;
while (dataSet->hasNext()) {
auto row = dataSet->next();
REQUIRE(row->fields.size() == 3);
const auto& fields = row->fields;
auto inserted = actual.emplace(fields[0].stringV.value(),
to_string(row->timestamp) + "," +
fields[1].stringV.value() + "," + fields[2].stringV.value());
REQUIRE(inserted.second);
}
dataSet->closeOperationHandle();
REQUIRE(actual == expected);
}

auto empty = session->executeFastLastDataQueryForOnePrefixPath(
{"root", "test", "fast_last_cpp", "missing"});
REQUIRE_FALSE(empty->hasNext());
empty->closeOperationHandle();
REQUIRE_THROWS_AS(session->executeFastLastDataQueryForOnePrefixPath({"root", "test", "*"}),
std::invalid_argument);

for (const auto& path : paths) {
session->deleteTimeseries(path);
}
}

// Helper function for comparing TEndPoint with detailed error message
void assertTEndPointEqual(const TEndPoint& actual, const std::string& expectedIp, int expectedPort,
const char* file, int line) {
Expand Down
Loading