diff --git a/iotdb-client/client-cpp/src/include/Session.h b/iotdb-client/client-cpp/src/include/Session.h index 054c8f6f7734f..0e8e1541ba00d 100644 --- a/iotdb-client/client-cpp/src/include/Session.h +++ b/iotdb-client/client-cpp/src/include/Session.h @@ -780,6 +780,11 @@ class Session { std::unique_ptr executeLastDataQuery(const std::vector& 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 + executeFastLastDataQueryForOnePrefixPath(const std::vector& prefixes); + void createSchemaTemplate(const Template& templ); void setSchemaTemplate(const std::string& template_name, const std::string& prefix_path); diff --git a/iotdb-client/client-cpp/src/rpc/SessionConnection.cpp b/iotdb-client/client-cpp/src/rpc/SessionConnection.cpp index 3eea30cb989d5..615dcc5deb44c 100644 --- a/iotdb-client/client-cpp/src/rpc/SessionConnection.cpp +++ b/iotdb-client/client-cpp/src/rpc/SessionConnection.cpp @@ -27,6 +27,7 @@ #include #include +#include #include #include "SessionDataSet.h" @@ -257,6 +258,43 @@ SessionConnection::executeLastDataQuery(const std::vector& paths, i zoneId, timeFactor, resp.columnIndex2TsBlockColumnIndexList); } +std::unique_ptr SessionConnection::executeFastLastDataQueryForOnePrefixPath( + const std::vector& 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( + [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); diff --git a/iotdb-client/client-cpp/src/rpc/SessionConnection.h b/iotdb-client/client-cpp/src/rpc/SessionConnection.h index c3c6f1a573560..b9d838b135baf 100644 --- a/iotdb-client/client-cpp/src/rpc/SessionConnection.h +++ b/iotdb-client/client-cpp/src/rpc/SessionConnection.h @@ -113,6 +113,9 @@ class SessionConnection : public std::enable_shared_from_this std::unique_ptr executeLastDataQuery(const std::vector& paths, int64_t lastTime); + std::unique_ptr + executeFastLastDataQueryForOnePrefixPath(const std::vector& prefixes); + void executeNonQueryStatement(const std::string& sql); std::unique_ptr executeQueryStatement(const std::string& sql, diff --git a/iotdb-client/client-cpp/src/session/Session.cpp b/iotdb-client/client-cpp/src/session/Session.cpp index e8cbc55d23cc1..b3a90cc49ede9 100644 --- a/iotdb-client/client-cpp/src/session/Session.cpp +++ b/iotdb-client/client-cpp/src/session/Session.cpp @@ -2312,6 +2312,11 @@ unique_ptr Session::executeLastDataQuery(const vector& p return impl_->getDefaultSessionConnection()->executeLastDataQuery(paths, lastTime); } +unique_ptr +Session::executeFastLastDataQueryForOnePrefixPath(const vector& prefixes) { + return impl_->getDefaultSessionConnection()->executeFastLastDataQueryForOnePrefixPath(prefixes); +} + void Session::createSchemaTemplate(const Template& templ) { TSCreateSchemaTemplateReq req; req.__set_name(templ.getName()); diff --git a/iotdb-client/client-cpp/test/cpp/sessionIT.cpp b/iotdb-client/client-cpp/test/cpp/sessionIT.cpp index 0c04d6bab68b5..2a0ec6cb7f82b 100644 --- a/iotdb-client/client-cpp/test/cpp/sessionIT.cpp +++ b/iotdb-client/client-cpp/test/cpp/sessionIT.cpp @@ -28,6 +28,7 @@ #include "common_types.h" #include +#include #include #include @@ -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 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 expected = {{paths[0], "30,3,INT64"}, + {paths[1], "20,2,INT64"}, + {paths[2], "40,4,INT64"}}; + const vector prefix = {"root", "test", "fast_last_cpp"}; + for (int query = 0; query < 2; ++query) { + auto dataSet = session->executeFastLastDataQueryForOnePrefixPath(prefix); + map 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) {