Skip to content

Commit aa365aa

Browse files
authored
[Client] Add Fast Last Query API to C++ session (#18860)
1 parent 330b0c7 commit aa365aa

5 files changed

Lines changed: 98 additions & 0 deletions

File tree

‎iotdb-client/client-cpp/src/include/Session.h‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -780,6 +780,11 @@ class Session {
780780
std::unique_ptr<SessionDataSet> executeLastDataQuery(const std::vector<std::string>& paths,
781781
int64_t lastTime);
782782

783+
// Query the latest value of each series below a tree-model prefix in a 1C1D deployment.
784+
// Pass path nodes without wildcards; result rows are not ordered.
785+
std::unique_ptr<SessionDataSet>
786+
executeFastLastDataQueryForOnePrefixPath(const std::vector<std::string>& prefixes);
787+
783788
void createSchemaTemplate(const Template& templ);
784789

785790
void setSchemaTemplate(const std::string& template_name, const std::string& prefix_path);

‎iotdb-client/client-cpp/src/rpc/SessionConnection.cpp‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
#include <thrift/transport/TSocket.h>
2828
#include <thrift/transport/TTransportException.h>
2929

30+
#include <stdexcept>
3031
#include <utility>
3132

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

261+
std::unique_ptr<SessionDataSet> SessionConnection::executeFastLastDataQueryForOnePrefixPath(
262+
const std::vector<std::string>& prefixes) {
263+
for (const auto& prefix : prefixes) {
264+
if (prefix.find('*') != std::string::npos) {
265+
throw std::invalid_argument("Fast Last Query prefix must not contain wildcards");
266+
}
267+
}
268+
269+
TSFastLastDataQueryForOnePrefixPathReq req;
270+
req.__set_prefixes(prefixes);
271+
req.__set_fetchSize(fetchSize);
272+
req.__set_enableRedirectQuery(session->enableRedirection_);
273+
274+
auto result = callWithRetryAndReconnect<TSExecuteStatementResp>(
275+
[this, &req]() {
276+
req.__set_sessionId(sessionId);
277+
req.__set_statementId(statementId);
278+
TSExecuteStatementResp resp;
279+
client->executeFastLastDataQueryForOnePrefixPath(resp, req);
280+
return resp;
281+
},
282+
[](const TSExecuteStatementResp& resp) { return resp.status; });
283+
if (result.getException()) {
284+
throw IoTDBConnectionException(extractExceptionMessage(result.getException()));
285+
}
286+
TSExecuteStatementResp resp = result.getResult();
287+
if (result.getRetryAttempts() == 0) {
288+
RpcUtils::verifySuccessWithRedirection(resp.status);
289+
} else {
290+
RpcUtils::verifySuccess(resp.status);
291+
}
292+
return createSessionDataSet("", resp.columns, resp.dataTypeList, resp.columnNameIndexMap,
293+
resp.queryId, statementId, client, sessionId, resp.queryResult,
294+
resp.ignoreTimeStamp, connectionTimeoutInMs, resp.moreData, fetchSize,
295+
zoneId, timeFactor, resp.columnIndex2TsBlockColumnIndexList);
296+
}
297+
260298
void SessionConnection::executeNonQueryStatement(const string& sql) {
261299
TSExecuteStatementReq req;
262300
req.__set_sessionId(sessionId);

‎iotdb-client/client-cpp/src/rpc/SessionConnection.h‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -113,6 +113,9 @@ class SessionConnection : public std::enable_shared_from_this<SessionConnection>
113113
std::unique_ptr<SessionDataSet> executeLastDataQuery(const std::vector<std::string>& paths,
114114
int64_t lastTime);
115115

116+
std::unique_ptr<SessionDataSet>
117+
executeFastLastDataQueryForOnePrefixPath(const std::vector<std::string>& prefixes);
118+
116119
void executeNonQueryStatement(const std::string& sql);
117120

118121
std::unique_ptr<SessionDataSet> executeQueryStatement(const std::string& sql,

‎iotdb-client/client-cpp/src/session/Session.cpp‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2312,6 +2312,11 @@ unique_ptr<SessionDataSet> Session::executeLastDataQuery(const vector<string>& p
23122312
return impl_->getDefaultSessionConnection()->executeLastDataQuery(paths, lastTime);
23132313
}
23142314

2315+
unique_ptr<SessionDataSet>
2316+
Session::executeFastLastDataQueryForOnePrefixPath(const vector<string>& prefixes) {
2317+
return impl_->getDefaultSessionConnection()->executeFastLastDataQueryForOnePrefixPath(prefixes);
2318+
}
2319+
23152320
void Session::createSchemaTemplate(const Template& templ) {
23162321
TSCreateSchemaTemplateReq req;
23172322
req.__set_name(templ.getName());

‎iotdb-client/client-cpp/test/cpp/sessionIT.cpp‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
#include "common_types.h"
2929

3030
#include <atomic>
31+
#include <map>
3132
#include <stdexcept>
3233
#include <thread>
3334

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

848+
TEST_CASE("Fast Last Query returns all series under a tree prefix", "[fastLastQuery]") {
849+
const vector<string> paths = {"root.test.fast_last_cpp.d1.s1", "root.test.fast_last_cpp.d1.s2",
850+
"root.test.fast_last_cpp.d2.s1"};
851+
for (const auto& path : paths) {
852+
if (session->checkTimeseriesExists(path)) {
853+
session->deleteTimeseries(path);
854+
}
855+
session->createTimeseries(path, TSDataType::INT64, TSEncoding::RLE,
856+
CompressionType::SNAPPY);
857+
}
858+
session->insertRecord("root.test.fast_last_cpp.d1", 20, {"s1", "s2"}, {"1", "2"});
859+
session->insertRecord("root.test.fast_last_cpp.d1", 30, {"s1"}, {"3"});
860+
session->insertRecord("root.test.fast_last_cpp.d2", 40, {"s1"}, {"4"});
861+
862+
const map<string, string> expected = {{paths[0], "30,3,INT64"},
863+
{paths[1], "20,2,INT64"},
864+
{paths[2], "40,4,INT64"}};
865+
const vector<string> prefix = {"root", "test", "fast_last_cpp"};
866+
for (int query = 0; query < 2; ++query) {
867+
auto dataSet = session->executeFastLastDataQueryForOnePrefixPath(prefix);
868+
map<string, string> actual;
869+
while (dataSet->hasNext()) {
870+
auto row = dataSet->next();
871+
REQUIRE(row->fields.size() == 3);
872+
const auto& fields = row->fields;
873+
auto inserted = actual.emplace(fields[0].stringV.value(),
874+
to_string(row->timestamp) + "," +
875+
fields[1].stringV.value() + "," + fields[2].stringV.value());
876+
REQUIRE(inserted.second);
877+
}
878+
dataSet->closeOperationHandle();
879+
REQUIRE(actual == expected);
880+
}
881+
882+
auto empty = session->executeFastLastDataQueryForOnePrefixPath(
883+
{"root", "test", "fast_last_cpp", "missing"});
884+
REQUIRE_FALSE(empty->hasNext());
885+
empty->closeOperationHandle();
886+
REQUIRE_THROWS_AS(session->executeFastLastDataQueryForOnePrefixPath({"root", "test", "*"}),
887+
std::invalid_argument);
888+
889+
for (const auto& path : paths) {
890+
session->deleteTimeseries(path);
891+
}
892+
}
893+
847894
// Helper function for comparing TEndPoint with detailed error message
848895
void assertTEndPointEqual(const TEndPoint& actual, const std::string& expectedIp, int expectedPort,
849896
const char* file, int line) {

0 commit comments

Comments
 (0)