You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "Erixonich (via GitHub)" <gi...@apache.org> on 2026/04/09 18:51:04 UTC
[PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Erixonich opened a new pull request, #7966:
URL: https://github.com/apache/ignite-3/pull/7966
https://issues.apache.org/jira/browse/IGNITE-22141
Thank you for submitting the pull request.
To streamline the review process of the patch and ensure better code quality
we ask both an author and a reviewer to verify the following:
### The Review Checklist
- [ ] **Formal criteria:** TC status, codestyle, mandatory documentation. Also make sure to complete the following:
\- There is a single JIRA ticket related to the pull request.
\- The web-link to the pull request is attached to the JIRA ticket.
\- The JIRA ticket has the Patch Available state.
\- The description of the JIRA ticket explains WHAT was made, WHY and HOW.
\- The pull request title is treated as the final commit message. The following pattern must be used: IGNITE-XXXX Change summary where XXXX - number of JIRA issue.
- [ ] **Design:** new code conforms with the design principles of the components it is added to.
- [ ] **Patch quality:** patch cannot be split into smaller pieces, its size must be reasonable.
- [ ] **Code quality:** code is clean and readable, necessary developer documentation is added if needed.
- [ ] **Tests code quality:** test set covers positive/negative scenarios, happy/edge cases. Tests are effective in terms of execution time and resources.
### Notes
- [Apache Ignite Coding Guidelines](https://cwiki.apache.org/confluence/display/IGNITE/Java+Code+Style+Guide)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "Erixonich (via GitHub)" <gi...@apache.org>.
Erixonich commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3079107483
##########
modules/platforms/cpp/ignite/protocol/messages.h:
##########
@@ -66,10 +69,16 @@ struct handshake_response {
protocol_context context{};
/** Observable timestamp. */
- std::int64_t observable_timestamp;
+ std::int64_t observable_timestamp{};
/** Server idle timeout in ms. */
- std::int64_t idle_timeout_ms;
+ std::int64_t idle_timeout_ms{};
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "isapego (via GitHub)" <gi...@apache.org>.
isapego merged PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "Erixonich (via GitHub)" <gi...@apache.org>.
Erixonich commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3079358497
##########
modules/platforms/cpp/ignite/client/detail/cluster_connection.h:
##########
@@ -289,6 +319,22 @@ class cluster_connection : public std::enable_shared_from_this<cluster_connectio
*/
std::shared_ptr<node_connection> get_random_connected_channel();
+ /**
+ * Get a preferred node connection if partition mapping is provided
+ * otherwise returns random node connection.
+ * @param preferred_node_name Name of preferred node.
+ * @return Node connection.
+ */
+ std::shared_ptr<node_connection> get_preferred_channel(const std::string& preferred_node_name);
+
+ /**
+ * Get connection according to provided partition mapping
+ * otherwise returns random node connection.
+ * @param pm Partition mapping.
+ * @return Node connection.
+ */
+ std::shared_ptr<node_connection> get_connected_channel(const std::optional<std::string> &preferred_node_name);
Review Comment:
done, removed get_preferred_node.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "isapego (via GitHub)" <gi...@apache.org>.
isapego commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3080281956
##########
modules/platforms/cpp/ignite/common/uuid_test.cpp:
##########
@@ -64,3 +64,54 @@ TEST(uuid, stream) {
EXPECT_EQ(uuidString, uuidString2);
}
+
+TEST(uuid, incorrect_string_representation) {
+ auto uuidText = "foo";
+
+ auto resOpt = ignite::uuid::from_string(uuidText);
+
+ EXPECT_FALSE(resOpt.has_value());
+}
+
+class uuid_string_presentation_fixture: public ::testing::TestWithParam<std::tuple<std::string, ignite::uuid>> {};
+
+TEST_P(uuid_string_presentation_fixture, from_string) {
+ auto [uuidText, uuid] = GetParam();
+
+ auto resOpt = ignite::uuid::from_string(uuidText);
+
+ EXPECT_TRUE(resOpt.has_value());
+
+ EXPECT_EQ(uuid.get_most_significant_bits(), resOpt->get_most_significant_bits());
+ EXPECT_EQ(uuid.get_least_significant_bits(), resOpt->get_least_significant_bits());
+}
+
+// We check that two representation are consistent both ways.
+TEST_P(uuid_string_presentation_fixture, to_string) {
+ auto [uuidText, uuid] = GetParam();
+
+
+ std::stringstream ss;
+ ss << uuid;
+
+ EXPECT_EQ(uuidText, ss.str());
+}
Review Comment:
Let's go here uuid -> string -> uuid.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "isapego (via GitHub)" <gi...@apache.org>.
isapego commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3078172196
##########
modules/platforms/cpp/ignite/client/detail/table/table_impl.cpp:
##########
@@ -472,4 +584,68 @@ std::shared_ptr<table_impl> table_impl::from_facade(table &tb) {
return tb.m_impl;
}
+void table_impl::update_partition_assignment() {
+ ignite_callback<std::shared_ptr<protocol::partition_assignment>> callback = [self=shared_from_this()](auto pa) {
+ if (pa.has_error()) {
+ self->m_connection->get_logger()->log_error("Error while updating partition assignment for table"
+ + self->get_name() + "error " + pa.error().what_str());
+
+ return;
+ }
+
+ std::lock_guard lock(self->m_partitions_mutex);
+ self->m_partition_assignment = std::move(pa).value();
+ };
+
+ load_partition_assignment_async(std::move(callback));
+}
Review Comment:
Let's just create a ticket for that and put a TODO here. The solution is functional, it's merely an optimization.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "Erixonich (via GitHub)" <gi...@apache.org>.
Erixonich commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3070809063
##########
modules/platforms/cpp/ignite/client/detail/table/table_impl.cpp:
##########
@@ -472,4 +584,68 @@ std::shared_ptr<table_impl> table_impl::from_facade(table &tb) {
return tb.m_impl;
}
+void table_impl::update_partition_assignment() {
+ ignite_callback<std::shared_ptr<protocol::partition_assignment>> callback = [self=shared_from_this()](auto pa) {
+ if (pa.has_error()) {
+ self->m_connection->get_logger()->log_error("Error while updating partition assignment for table"
+ + self->get_name() + "error " + pa.error().what_str());
+
+ return;
+ }
+
+ std::lock_guard lock(self->m_partitions_mutex);
+ self->m_partition_assignment = std::move(pa).value();
+ };
+
+ load_partition_assignment_async(std::move(callback));
+}
Review Comment:
@isapego I believe we have similar issue with other types of request, would you like to discuss it in details?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3067154184
##########
modules/platforms/cpp/ignite/common/uuid.h:
##########
@@ -44,7 +47,65 @@ class uuid {
*/
constexpr uuid(std::int64_t most, std::int64_t least) noexcept
: most(most)
- , least(least) {}
+ , least(least) { }
+
+ /**
+ * Parses string-encoded uuid.
+ * Example of possible input value: ff979603-fb56-49e9-bc79-7c4487bbbafd.
+ * @param text String-encoded uuid.
+ * @return @c uuid object if parsing was successful, otherwise empty optional.
+ */
+ static std::optional<uuid> from_string(const std::string& text) {
+ if (text.length() != 36) {
+ return {};
+ }
+
+ auto str = text.c_str();
+
+ auto parse_chunk = [str](size_t beg, size_t end, int64_t& out) -> bool {
+ char* p;
+
+ if (errno != 0) {
+ errno = 0;
+ }
+
+ out = std::strtoll(str + beg, &p, 16);
+
+ if (p != str + end || (*p != '-' && *p != '\0') || errno == ERANGE) {
+ return false;
+ }
+
+ return true;
+ };
+
+ int64_t msb1, msb2, msb3;
+ int64_t lsb1, lsb2;
+
+ if (!parse_chunk(0, 8, msb1)) {
+ return {};
+ }
+
+ if (!parse_chunk(9, 13, msb2)) {
+ return {};
+ }
+
+ if (!parse_chunk(14, 18, msb3)) {
+ return {};
+ }
+
+ if (!parse_chunk(19, 23, lsb1)) {
+ return {};
+ }
+
+ if (!parse_chunk(24, 36, lsb2)) {
+ return {};
+ }
+
+ int64_t msb = msb1 << 32 | msb2 << 16 | msb3;
+ int64_t lsb = lsb1 << 48 | lsb2;
+
+ return uuid{msb, lsb};
Review Comment:
`uuid::from_string` builds the 64-bit halves using left shifts on signed `int64_t` values (`msb1 << 32`, `lsb1 << 48`). Left-shifting into the sign bit is undefined behavior and can yield different results across compilers/optimizations. Use `uint64_t` intermediates for the bit assembly (as the stream operator already does) and only cast to `int64_t` at the end.
```suggestion
std::uint64_t msb =
(static_cast<std::uint64_t>(msb1) << 32) | (static_cast<std::uint64_t>(msb2) << 16) | static_cast<std::uint64_t>(msb3);
std::uint64_t lsb = (static_cast<std::uint64_t>(lsb1) << 48) | static_cast<std::uint64_t>(lsb2);
return uuid{static_cast<std::int64_t>(msb), static_cast<std::int64_t>(lsb)};
```
##########
modules/platforms/cpp/tests/test-common/proxy/message_listener.h:
##########
@@ -0,0 +1,176 @@
+// 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.
+//
+
+#pragma once
+
+#include "ignite/network/length_prefix_codec.h"
+#include "ignite/protocol/client_operation.h"
+#include "ignite/protocol/reader.h"
+
+#include <queue>
+#include <shared_mutex>
+#include <utility>
+
+namespace ignite::proxy {
+
+using raw_message = std::vector<char>;
+
+class structured_message {
+public:
+ structured_message() = default;
+
+ structured_message(const structured_message& structured_message) = default;
+
+ explicit structured_message(std::vector<std::byte> data)
+ :m_data(std::move(data)) { }
+
+ [[nodiscard]] protocol::reader payload_reader() const {
+ return protocol::reader{{m_data.data() + m_payload_pos, m_data.size() - m_payload_pos}};
+ }
+protected:
+ std::vector<std::byte> m_data{};
+ size_t m_payload_pos{};
+};
+
+/**
+ * Message which clients sends to server.
+ */
+class client_message: public structured_message{
+public:
+ client_message() = default;
+
+ client_message(const client_message& msg) = default;
+
+ explicit client_message(std::vector<std::byte> data)
+ : structured_message(std::move(data))
+ {
+ protocol::reader rd{{m_data.data(), m_data.size()}};
+
+ m_op = static_cast<protocol::client_operation>(rd.read_int32());
+ m_req_id = rd.read_int64();
+
+ m_payload_pos = rd.position();
+
+ assert(m_payload_pos == 0 || m_payload_pos < m_data.size());
Review Comment:
`assert(m_payload_pos == 0 || m_payload_pos < m_data.size())` rejects the valid case where the payload is empty (`m_payload_pos == m_data.size()`). This can trip assertions for operations/responses with no payload. Allow `<= m_data.size()` (and ideally validate that the header read didn't run past the end).
```suggestion
assert(m_payload_pos <= m_data.size());
assert(m_payload_pos == 0 || m_payload_pos <= m_data.size());
```
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/PlatformTestNodeRunner.java:
##########
@@ -57,6 +57,7 @@
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
Review Comment:
This import appears unused in the file. If Checkstyle/IDE inspections are enforced in CI, unused imports can fail the build or at least create noise. Remove the `HashMap` import if it's not needed.
```suggestion
```
##########
modules/platforms/cpp/tests/client-test/partition_awareness_test.cpp:
##########
@@ -0,0 +1,623 @@
+// 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 "detail/string_extensions.h"
+#include "ignite_runner_suite.h"
+#include "tests/test-common/test_utils.h"
+
+#include "ignite/client/detail/utils.h"
+#include "ignite/client/ignite_client.h"
+#include "ignite/client/ignite_client_configuration.h"
+
+#include "proxy/asio_proxy.h"
+#include "proxy/message_listener.h"
+
+#include <gtest/gtest.h>
+
+namespace ignite {
+
+using namespace std::chrono_literals;
+
+using part_id_type = std::size_t;
+using key_type = std::int64_t;
+
+constexpr key_type MIN_KEY = 0;
+constexpr key_type MAX_KEY = 1000;
+
+static const std::map<std::string, std::string> NODES_TO_ADDR = {
+ {"org.apache.ignite.internal.runner.app.PlatformTestNodeRunner", "127.0.0.1:10942"},
+ {"org.apache.ignite.internal.runner.app.PlatformTestNodeRunner_2", "127.0.0.1:10943"}
+};
+
+/**
+ * Parses string-encoded partition distribution.
+ * Example of string:
+ * ff979603-fb56-49e9-bc79-7c4487bbbafd=[4];2e14bb82-5cb5-4283-aafd-f2d8552a842a=[5, 8, 9];
+ * @param encoded String-encoded distribution
+ * @return Conveniently structured distribution.
+ */
+std::map<int64_t, uuid> parse_partition_distribution(const std::string& encoded) {
+ std::map<int64_t, uuid> res;
+
+ size_t lhs = 0;
+ size_t rhs = std::string::npos;
+ while ((rhs = encoded.find(';', lhs)) != std::string::npos) {
+ std::string chunk= encoded.substr(lhs, rhs - lhs);
+
+ auto eq_pos = chunk.find('=');
+
+ if (eq_pos == std::string::npos) {
+ throw std::runtime_error("Incorrect partition distribution text:" + encoded);
+ }
+
+ auto node_id = uuid::from_string(chunk.substr(0, eq_pos));
+
+ if (!node_id.has_value()) {
+ throw std::runtime_error("can't parse partition distribution, incorrect input:" + encoded);
+ }
+
+ std::string num_list = chunk.substr(eq_pos + 1); // e.g. [1,2,3]
+
+ size_t lo = 1;
+ size_t hi = std::string::npos;
+ while ((hi = num_list.find_first_of(",]", lo)) != std::string::npos) {
+ int64_t part_id = std::stoll(num_list.substr(lo, hi - lo));
+ res[part_id] = *node_id;
+ lo = hi + 1;
+ }
+ lhs = rhs + 1;
+ }
+
+ return res;
+}
+
+class partition_awareness_test : public ignite_runner_suite {
+protected:
+ void SetUp() override {
+ ignite_client_configuration client_cfg;
+ client_cfg.set_endpoints(get_node_addrs());
+
+ m_direct_client = ignite_client::start(client_cfg, 5s);
+
+ auto nodes = m_direct_client.get_cluster_nodes();
+
+
+ for (auto& node: nodes) {
+ if (NODES_TO_ADDR.count(node.get_name())) {
+ m_endpoint_to_node_id[NODES_TO_ADDR.at(node.get_name())] = node.get_id();
+ }
+ }
+
+ ASSERT_EQ(m_endpoint_to_node_id.size(), get_node_addrs().size()) << "Incorrect node names or address!"
+ << "This test should be able to connect to all non-ssl nodes";
+
+ collect_partition_distribution();
+
+ setup_proxy();
+ }
+
+ static ignite_tuple key_tup(key_type key) {
+ return get_tuple(key);
+ }
+
+ static ignite_tuple prim_val(key_type key) {
+ return get_tuple(std::to_string(key));
+ }
+
+ static ignite_tuple alt_val(key_type key) {
+ return get_tuple(std::to_string(key + MAX_KEY));
+ }
+
+ static ignite_tuple prim_rec(key_type key) {
+ return get_tuple(key, std::to_string(key));
+ }
+
+ static ignite_tuple alt_rec(key_type key) {
+ return get_tuple(key, std::to_string(key + MAX_KEY));
+ }
+
+ void collect_partition_distribution() {
+ std::map<key_type, int64_t> key_to_part;
+
+ populate_table();
+
+ std::stringstream sql;
+ sql << "select " << KEY_COLUMN << ", " << PART_PSEUDOCOLUMN << " from " << TABLE_1 << " order by " << KEY_COLUMN;
+
+ auto rs = m_direct_client.get_sql().execute(nullptr, nullptr, {sql.str()}, {});
+ do {
+ auto page = rs.current_page();
+
+ for (auto &rec : page) {
+ auto key = rec.get(KEY_COLUMN).get<key_type>();
+ auto part_id = rec.get(PART_PSEUDOCOLUMN).get<int64_t>();
+
+ key_to_part[key] = part_id;
+ }
+
+ if (rs.has_more_pages())
+ rs.fetch_next_page();
+ else
+ break;
+ } while (true);
+
+
+ auto job_exec = m_direct_client.get_compute().submit(
+ job_target::any_node(m_direct_client.get_cluster_nodes()),
+ job_descriptor::builder{GET_PART_DISTRIBUTION_JOB}.build(),
+ {TABLE_1}
+ );
+
+ auto res = job_exec.get_result()->get_primitive().get<std::string>();
+
+ auto part_to_node = parse_partition_distribution(res);
+
+ for (auto [key, part_id] : key_to_part) {
+ m_key_distribution[key] = part_to_node[part_id];
+
+ auto node_id = part_to_node[part_id];
+ std::string endpoint;
+ for (auto& [ep, id] : m_endpoint_to_node_id) {
+ if (node_id == id) {
+ endpoint = ep;
+ }
+ }
Review Comment:
`endpoint` is assigned but never used. With `-Werror` enabled (see top-level CMake `WARNINGS_AS_ERRORS`), this can fail the build. Remove the variable or use it for an assertion (e.g., verify every `node_id` maps to a known endpoint).
```suggestion
bool endpoint_found = false;
for (auto& [ep, id] : m_endpoint_to_node_id) {
if (node_id == id) {
endpoint_found = true;
break;
}
}
ASSERT_TRUE(endpoint_found);
```
##########
modules/platforms/cpp/tests/test-common/proxy/message_listener.h:
##########
@@ -0,0 +1,176 @@
+// 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.
+//
+
+#pragma once
+
+#include "ignite/network/length_prefix_codec.h"
+#include "ignite/protocol/client_operation.h"
+#include "ignite/protocol/reader.h"
+
Review Comment:
This header relies on transitive includes for `assert` and `std::atomic_bool` (no `<cassert>` / `<atomic>` included here). That makes the header non-self-contained and can break builds if include order changes or it is included directly elsewhere. Add the missing standard headers explicitly.
```suggestion
#include <atomic>
#include <cassert>
```
##########
modules/platforms/cpp/ignite/protocol/messages.cpp:
##########
@@ -100,4 +100,39 @@ handshake_response parse_handshake_response(bytes_view message) {
return res;
}
+void write_partition_assignment_request(writer &writer, std::int32_t table_id, std::int64_t timestamp) {
+ writer.write(table_id);
+ writer.write(timestamp);
+}
+
+std::shared_ptr<partition_assignment> read_partition_assignment_response(reader &reader, std::int64_t timestamp) {
+ auto cnt = reader.read_int32();
+ if (cnt < 0)
+ throw ignite_error("Invalid partition count: " + std::to_string(cnt));
+
+ std::vector<std::optional<std::string>> partitions;
+ partitions.reserve(cnt);
Review Comment:
`read_partition_assignment_response` accepts `cnt == 0` as valid and proceeds, but server/client implementations treat non-positive partition counts as invalid. Consider validating `cnt <= 0` and failing fast with a clear error to avoid silently creating an empty assignment and disabling partition awareness.
##########
modules/platforms/cpp/ignite/client/detail/cluster_connection.h:
##########
@@ -395,6 +448,9 @@ class cluster_connection : public std::enable_shared_from_this<cluster_connectio
/** Node connections. */
std::unordered_map<uint64_t, std::shared_ptr<node_connection>> m_connections;
+ /** Node consistent ID to node instance mapping. */
+ std::unordered_map<std::string, node_connection*> m_consistent_id_mapping;
+
Review Comment:
`m_consistent_id_mapping` is added but not used anywhere (no reads/writes in the codebase). If this was intended to optimize preferred-node lookups, please either wire it up (update on connect/disconnect and use it in `get_preferred_channel`) or remove it to avoid dead state and confusion.
```suggestion
```
##########
modules/platforms/cpp/ignite/protocol/partition_assignment.h:
##########
@@ -0,0 +1,61 @@
+/*
+* 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
Review Comment:
The Apache license header formatting is inconsistent (`* Licensed...` line is missing the usual leading space). Some repos have automated checks that require the exact header template, so this can break CI. Align this header with the standard `/*\n * Licensed ...` format used elsewhere in the C++ sources.
##########
modules/platforms/cpp/ignite/client/detail/table/table_impl.cpp:
##########
@@ -472,4 +584,68 @@ std::shared_ptr<table_impl> table_impl::from_facade(table &tb) {
return tb.m_impl;
}
+void table_impl::update_partition_assignment() {
+ ignite_callback<std::shared_ptr<protocol::partition_assignment>> callback = [self=shared_from_this()](auto pa) {
+ if (pa.has_error()) {
+ self->m_connection->get_logger()->log_error("Error while updating partition assignment for table"
+ + self->get_name() + "error " + pa.error().what_str());
+
+ return;
+ }
+
+ std::lock_guard lock(self->m_partitions_mutex);
+ self->m_partition_assignment = std::move(pa).value();
+ };
+
+ load_partition_assignment_async(std::move(callback));
+}
Review Comment:
`update_partition_assignment()` triggers `load_partition_assignment_async()` on every operation and there is no de-duplication of in-flight assignment requests. Under concurrent load (or when the assignment is outdated), multiple threads can send many `PARTITION_ASSIGNMENT_GET` requests for the same table/timestamp. Consider caching the in-flight future/result per table+timestamp (similar to Java `ClientTable#getPartitionAssignment`) or adding a simple in-progress flag guarded by `m_partitions_mutex`.
##########
modules/platforms/cpp/ignite/client/detail/table/table_impl.cpp:
##########
@@ -472,4 +584,68 @@ std::shared_ptr<table_impl> table_impl::from_facade(table &tb) {
return tb.m_impl;
}
+void table_impl::update_partition_assignment() {
+ ignite_callback<std::shared_ptr<protocol::partition_assignment>> callback = [self=shared_from_this()](auto pa) {
+ if (pa.has_error()) {
+ self->m_connection->get_logger()->log_error("Error while updating partition assignment for table"
+ + self->get_name() + "error " + pa.error().what_str());
+
+ return;
+ }
+
+ std::lock_guard lock(self->m_partitions_mutex);
+ self->m_partition_assignment = std::move(pa).value();
+ };
+
+ load_partition_assignment_async(std::move(callback));
+}
+
+void table_impl::load_partition_assignment_async(ignite_callback<std::shared_ptr<protocol::partition_assignment>> callback) {
+ std::int64_t timestamp = m_connection->get_assignment_timestamp();
+
+ auto pa = get_partition_assignment();
+ if (pa && !pa->is_outdated(timestamp)) {
+ m_connection->get_logger()->log_debug("Partition assignment for table " + get_name() + " is up to date.");
+
+ callback(std::move(pa));
+ return;
+ }
+
+ auto writer_func = [id = m_id, timestamp](protocol::writer &writer, auto&) {
+ protocol::write_partition_assignment_request(writer, id, timestamp);
+ };
+
+ auto reader_func = [timestamp](protocol::reader &reader) -> std::shared_ptr<protocol::partition_assignment> {
+ return protocol::read_partition_assignment_response(reader, timestamp);
+ };
+
+ m_connection->perform_request<std::shared_ptr<protocol::partition_assignment>>(
+ protocol::client_operation::PARTITION_ASSIGNMENT_GET,
+ nullptr,
+ writer_func,
+ std::move(reader_func),
+ std::move(callback));
+}
+
+std::optional<std::string> table_impl::get_preferred_node_name(const ignite_tuple &key_or_rec, const schema &sch) {
+ auto pa = get_partition_assignment();
+
+ if (!pa || pa->get_partitions().empty()) {
+ m_connection->get_logger()->log_debug("No partition distribution available, random node will called");
+ return {};
+ }
+
+ hash_calculator hc;
+ for (auto column : sch.collocated_columns) {
+ auto val = key_or_rec.get(column->key_index);
Review Comment:
`get_preferred_node_name` reads tuple values by `column->key_index` (tuple ordinal), but tuples are serialized by column *name* (see `pack_tuple`: `tuple.column_ordinal(col.name)`). For record tuples (and even key tuples built with a different insertion order), `key_index` is not a stable ordinal, so the hash (and partition selection) can be wrong. Use column names (or schema column indices) to fetch values from `ignite_tuple` when computing the colocated hash.
```suggestion
auto val = key_or_rec.get(column->name);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "Erixonich (via GitHub)" <gi...@apache.org>.
Erixonich commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3080368879
##########
modules/platforms/cpp/ignite/protocol/partition_assignment.h:
##########
@@ -0,0 +1,70 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <optional>
+#include <string>
+#include <vector>
+#include <cstdint>
+
+namespace ignite::protocol {
+
+/**
+ * Partition assignment.
+ */
+class partition_assignment {
+public:
+ partition_assignment() = default;
+
+ /**
+ * Constructor.
+ *
+ * @param timestamp Partition assignment data.
+ * @param partitions Partition assignment encode into vector.
+ */
+ partition_assignment(std::int64_t timestamp, std::vector<std::optional<std::string>> partitions)
+ : m_timestamp(timestamp)
+ , m_partitions(std::move(partitions)) {}
+
+ /**
+ * Check whether the assignment is outdated.
+ *
+ * @param actual_timestamp Timestamp.
+ * @return @c true if assignment is outdated.
+ */
+ [[nodiscard]] bool is_outdated(std::int64_t actual_timestamp) const { return m_timestamp < actual_timestamp; }
+
+ /**
+ * Partitions. Vector is decoded as following: size of collection is total number of partitions,
+ * i-th value means that partition with id == i resides on (primary replica of that partition belongs to)
+ * node with provided consisted id (AKA node name).
+ *
+ * @return Vector with partition assignments.
+ */
+ [[nodiscard]] const std::vector<std::optional<std::string>>& get_partitions() const {
+ return m_partitions;
+ }
+private:
+ /** Assignment timestamp. */
+ std::int64_t m_timestamp{0};
+
+
+ std::vector<std::optional<std::string>> m_partitions;
Review Comment:
somehow I've missed that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "Erixonich (via GitHub)" <gi...@apache.org>.
Erixonich commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3079046756
##########
modules/platforms/cpp/ignite/client/detail/table/schema.h:
##########
@@ -118,30 +125,52 @@ struct schema {
* @return A new schema instance.
*/
static std::shared_ptr<schema> create_instance(std::int32_t version, std::vector<column> &&cols) {
- std::int32_t key_columns_cnt = 0;
+ size_t key_columns_cnt = 0;
+ size_t collocated_columns_cnt = 0;
Review Comment:
done
##########
modules/platforms/cpp/tests/test-common/proxy/message_listener.h:
##########
@@ -0,0 +1,178 @@
+// 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.
+//
+
+#pragma once
+
+#include "ignite/network/length_prefix_codec.h"
+#include "ignite/protocol/client_operation.h"
+#include "ignite/protocol/reader.h"
+
+#include <atomic>
+#include <cassert>
+#include <queue>
+#include <shared_mutex>
+#include <utility>
+
+namespace ignite::proxy {
+
+using raw_message = std::vector<char>;
+
+class structured_message {
+public:
+ structured_message() = default;
+
+ structured_message(const structured_message& structured_message) = default;
+
+ explicit structured_message(std::vector<std::byte> data)
+ :m_data(std::move(data)) { }
+
+ [[nodiscard]] protocol::reader payload_reader() const {
+ return protocol::reader{{m_data.data() + m_payload_pos, m_data.size() - m_payload_pos}};
+ }
+protected:
+ std::vector<std::byte> m_data{};
+ size_t m_payload_pos{};
+};
+
+/**
+ * Message which clients sends to server.
+ */
+class client_message: public structured_message{
+public:
+ client_message() = default;
+
+ client_message(const client_message& msg) = default;
+
+ explicit client_message(std::vector<std::byte> data)
+ : structured_message(std::move(data))
+ {
+ protocol::reader rd{{m_data.data(), m_data.size()}};
+
+ m_op = static_cast<protocol::client_operation>(rd.read_int32());
+ m_req_id = rd.read_int64();
+
+ m_payload_pos = rd.position();
+
+ assert(m_payload_pos <= m_data.size());
+ }
+
+ [[nodiscard]] protocol::client_operation get_op() const { return m_op; }
+
+ [[nodiscard]] int64_t get_req_id() const { return m_req_id; }
+
+private:
+ protocol::client_operation m_op{};
+ int64_t m_req_id{};
+};
+
+/**
+ * Message which server sends to client.
+ */
+class server_message : public structured_message {
+public:
+ server_message() = default;
+
+ server_message(const server_message&) = default;
+
+ explicit server_message(std::vector<std::byte> data)
+ : structured_message(std::move(data))
+ {
+ protocol::reader rd{{m_data.data(), m_data.size()}};
+
+ m_req_id = rd.read_int64();
+ m_flags = rd.read_int32();
+ m_obs_ts = rd.read_int64();
+
+ m_payload_pos = rd.position();
+
+ assert(m_payload_pos <= m_data.size());
+ }
+
+ [[nodiscard]] int64_t get_req_id() const { return m_req_id; }
+
+ [[nodiscard]] int32_t get_flags() const { return m_flags; }
+
+ [[nodiscard]] int64_t get_obs_ts() const { return m_obs_ts; }
+
+private:
+ int64_t m_req_id{};
+ int32_t m_flags{};
+ int64_t m_obs_ts{};
+};
+
+/**
+ * Intercepts messages which go through @c asio_proxy.
+ */
+class message_listener {
+public:
+ void register_message(raw_message msg) {
+ if (enable_message_registration.load()) {
+ std::unique_lock lock(m_mutex);
+ m_queue.push(std::move(msg));
+ }
+ }
+
+ [[nodiscard]] std::queue<raw_message> get_msg_queue() const {
+ std::shared_lock lock(m_mutex);
+ return m_queue;
+ }
+
+ template<typename MESSAGE_TYPE>
+ std::vector<MESSAGE_TYPE> get_next() {
+ std::unique_lock lock(m_mutex);
+
+ std::vector<MESSAGE_TYPE> res;
+
+ while (!m_queue.empty() && res.empty()) {
+ auto& chunk = m_queue.front();
+
+ network::data_buffer_ref buf{{chunk.data(), chunk.size()}};
+
+ while (true) {
+ auto out = codec.decode(buf);
+
+ if (out.empty()) {
+ break;
+ }
+
+ auto out_bv = out.get_bytes_view();
+ std::vector<std::byte> data{out_bv.begin(), out_bv.end()};
+ res.emplace_back(std::move(data));
+ }
+ m_queue.pop();
+ }
+
+ return res;
+ }
+
+ /**
+ * Enable/disables message registration for message listeners;
+ * @param enable @c True if registration enabled, otherwise disabled.
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "isapego (via GitHub)" <gi...@apache.org>.
isapego commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3080305950
##########
modules/platforms/cpp/ignite/client/detail/cluster_connection.h:
##########
@@ -289,6 +319,22 @@ class cluster_connection : public std::enable_shared_from_this<cluster_connectio
*/
std::shared_ptr<node_connection> get_random_connected_channel();
+ /**
+ * Get a preferred node connection if partition mapping is provided
+ * otherwise returns random node connection.
+ * @param preferred_node_name Name of preferred node.
+ * @return Node connection.
+ */
+ std::shared_ptr<node_connection> get_preferred_channel(const std::string& preferred_node_name);
+
+ /**
+ * Get connection according to provided partition mapping
+ * otherwise returns random node connection.
+ * @param pm Partition mapping.
+ * @return Node connection.
+ */
+ std::shared_ptr<node_connection> get_connected_channel(const std::optional<std::string> &preferred_node_name);
Review Comment:
I still see both methods, and from the name it's hard to understand which does what. We actually only need one.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "Erixonich (via GitHub)" <gi...@apache.org>.
Erixonich commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3079329875
##########
modules/platforms/cpp/ignite/client/detail/node_connection.h:
##########
@@ -232,7 +232,14 @@ class node_connection : public std::enable_shared_from_this<node_connection> {
*/
std::shared_ptr<ignite_logger> get_logger() const { return m_logger; }
- void handle_timeouts();
+ /**
+ * Cancels waiting for over-due responses.
+ */
+ void handle_timeouts();
+
+ const std::string& get_node_name() const {
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "Erixonich (via GitHub)" <gi...@apache.org>.
Erixonich commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3079117270
##########
modules/platforms/cpp/ignite/protocol/messages.cpp:
##########
@@ -100,4 +100,39 @@ handshake_response parse_handshake_response(bytes_view message) {
return res;
}
+void write_partition_assignment_request(writer &writer, std::int32_t table_id, std::int64_t timestamp) {
+ writer.write(table_id);
+ writer.write(timestamp);
+}
+
+std::shared_ptr<partition_assignment> read_partition_assignment_response(reader &reader, std::int64_t timestamp) {
+ auto cnt = reader.read_int32();
+ if (cnt <= 0)
+ throw ignite_error("Invalid partition count: " + std::to_string(cnt));
Review Comment:
I've synchronized with what java client does:
```
int cnt = r.in().unpackInt();
if (cnt <= 0) {
throw new IgniteException(INTERNAL_ERR, "Invalid partition count returned by the server: " + cnt);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "Erixonich (via GitHub)" <gi...@apache.org>.
Erixonich commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3079140886
##########
modules/platforms/cpp/ignite/protocol/messages.cpp:
##########
@@ -100,4 +100,39 @@ handshake_response parse_handshake_response(bytes_view message) {
return res;
}
+void write_partition_assignment_request(writer &writer, std::int32_t table_id, std::int64_t timestamp) {
+ writer.write(table_id);
+ writer.write(timestamp);
+}
+
+std::shared_ptr<partition_assignment> read_partition_assignment_response(reader &reader, std::int64_t timestamp) {
+ auto cnt = reader.read_int32();
+ if (cnt <= 0)
+ throw ignite_error("Invalid partition count: " + std::to_string(cnt));
Review Comment:
Regarding bool flag (assignment_available) in this case server should return correct number of partitions without values for vector itself.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "Erixonich (via GitHub)" <gi...@apache.org>.
Erixonich commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3079117270
##########
modules/platforms/cpp/ignite/protocol/messages.cpp:
##########
@@ -100,4 +100,39 @@ handshake_response parse_handshake_response(bytes_view message) {
return res;
}
+void write_partition_assignment_request(writer &writer, std::int32_t table_id, std::int64_t timestamp) {
+ writer.write(table_id);
+ writer.write(timestamp);
+}
+
+std::shared_ptr<partition_assignment> read_partition_assignment_response(reader &reader, std::int64_t timestamp) {
+ auto cnt = reader.read_int32();
+ if (cnt <= 0)
+ throw ignite_error("Invalid partition count: " + std::to_string(cnt));
Review Comment:
I've synchronized with what java client does:
```
int cnt = r.in().unpackInt();
if (cnt <= 0) {
throw new IgniteException(INTERNAL_ERR, "Invalid partition count returned by the server: " + cnt);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "Erixonich (via GitHub)" <gi...@apache.org>.
Erixonich commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3079333889
##########
modules/platforms/cpp/ignite/client/detail/cluster_connection.h:
##########
@@ -289,6 +319,22 @@ class cluster_connection : public std::enable_shared_from_this<cluster_connectio
*/
std::shared_ptr<node_connection> get_random_connected_channel();
+ /**
+ * Get a preferred node connection if partition mapping is provided
+ * otherwise returns random node connection.
+ * @param preferred_node_name Name of preferred node.
+ * @return Node connection.
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "isapego (via GitHub)" <gi...@apache.org>.
isapego commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3080245315
##########
modules/platforms/cpp/ignite/protocol/messages.cpp:
##########
@@ -100,4 +100,39 @@ handshake_response parse_handshake_response(bytes_view message) {
return res;
}
+void write_partition_assignment_request(writer &writer, std::int32_t table_id, std::int64_t timestamp) {
+ writer.write(table_id);
+ writer.write(timestamp);
+}
+
+std::shared_ptr<partition_assignment> read_partition_assignment_response(reader &reader, std::int64_t timestamp) {
+ auto cnt = reader.read_int32();
+ if (cnt <= 0)
+ throw ignite_error("Invalid partition count: " + std::to_string(cnt));
+
+ std::vector<std::optional<std::string>> partitions;
+ partitions.reserve(cnt);
+
+ bool assignment_available = reader.read_bool();
+ if (!assignment_available) {
+ // Invalidate the current assignment so that we can retry on the next call.
+ // Return an empty array so that per-partition batches can be initialized.
+ // We'll get the actual assignment on the next call.
+ partitions.insert(partitions.end(), cnt, std::nullopt);
+ return std::make_shared<partition_assignment>(0, std::move(partitions));
+ }
+
+ // Returned timestamp can be newer than requested.
+ std::int64_t ts = reader.read_int64();
+ if (ts < timestamp)
+ throw ignite_error("Returned timestamp is older than requested: " + std::to_string(ts) + " < "
+ + std::to_string(timestamp));
Review Comment:
Assertion in Java Client is dangerous actually. Maybe we should fix it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "Erixonich (via GitHub)" <gi...@apache.org>.
Erixonich commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3070809063
##########
modules/platforms/cpp/ignite/client/detail/table/table_impl.cpp:
##########
@@ -472,4 +584,68 @@ std::shared_ptr<table_impl> table_impl::from_facade(table &tb) {
return tb.m_impl;
}
+void table_impl::update_partition_assignment() {
+ ignite_callback<std::shared_ptr<protocol::partition_assignment>> callback = [self=shared_from_this()](auto pa) {
+ if (pa.has_error()) {
+ self->m_connection->get_logger()->log_error("Error while updating partition assignment for table"
+ + self->get_name() + "error " + pa.error().what_str());
+
+ return;
+ }
+
+ std::lock_guard lock(self->m_partitions_mutex);
+ self->m_partition_assignment = std::move(pa).value();
+ };
+
+ load_partition_assignment_async(std::move(callback));
+}
Review Comment:
@isapego I believe we have similar issue with other types of request, would you like to discuss it in details?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "Erixonich (via GitHub)" <gi...@apache.org>.
Erixonich commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3079310526
##########
modules/platforms/cpp/ignite/common/uuid_test.cpp:
##########
@@ -64,3 +64,13 @@ TEST(uuid, stream) {
EXPECT_EQ(uuidString, uuidString2);
}
+
+TEST(uuid, from_string) {
+ std::string uuidText = "4b62e46a-d380-460f-94ea-9b4320752634";
+
+ auto opt = ignite::uuid::from_string(uuidText);
+
+ EXPECT_TRUE(opt.has_value());
+ EXPECT_EQ(5432155248028304911LL, opt->get_most_significant_bits());
+ EXPECT_EQ(-7716184298936261068LL, opt->get_least_significant_bits());
+}
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "isapego (via GitHub)" <gi...@apache.org>.
isapego commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3078159084
##########
modules/platforms/cpp/ignite/client/detail/table/table_impl.cpp:
##########
@@ -472,4 +584,68 @@ std::shared_ptr<table_impl> table_impl::from_facade(table &tb) {
return tb.m_impl;
}
+void table_impl::update_partition_assignment() {
+ ignite_callback<std::shared_ptr<protocol::partition_assignment>> callback = [self=shared_from_this()](auto pa) {
+ if (pa.has_error()) {
+ self->m_connection->get_logger()->log_error("Error while updating partition assignment for table"
+ + self->get_name() + "error " + pa.error().what_str());
+
+ return;
+ }
+
+ std::lock_guard lock(self->m_partitions_mutex);
+ self->m_partition_assignment = std::move(pa).value();
+ };
+
+ load_partition_assignment_async(std::move(callback));
+}
+
+void table_impl::load_partition_assignment_async(ignite_callback<std::shared_ptr<protocol::partition_assignment>> callback) {
+ std::int64_t timestamp = m_connection->get_assignment_timestamp();
+
+ auto pa = get_partition_assignment();
+ if (pa && !pa->is_outdated(timestamp)) {
+ m_connection->get_logger()->log_debug("Partition assignment for table " + get_name() + " is up to date.");
+
+ callback(std::move(pa));
+ return;
+ }
+
+ auto writer_func = [id = m_id, timestamp](protocol::writer &writer, auto&) {
+ protocol::write_partition_assignment_request(writer, id, timestamp);
+ };
+
+ auto reader_func = [timestamp](protocol::reader &reader) -> std::shared_ptr<protocol::partition_assignment> {
+ return protocol::read_partition_assignment_response(reader, timestamp);
+ };
+
+ m_connection->perform_request<std::shared_ptr<protocol::partition_assignment>>(
+ protocol::client_operation::PARTITION_ASSIGNMENT_GET,
+ nullptr,
+ writer_func,
+ std::move(reader_func),
+ std::move(callback));
+}
+
+std::optional<std::string> table_impl::get_preferred_node_name(const ignite_tuple &key_or_rec, const schema &sch) {
+ auto pa = get_partition_assignment();
+
+ if (!pa || pa->get_partitions().empty()) {
+ m_connection->get_logger()->log_debug("No partition distribution available, random node will called");
+ return {};
+ }
+
+ hash_calculator hc;
+ for (auto column : sch.collocated_columns) {
+ auto val = key_or_rec.get(column->key_index);
Review Comment:
Agree, the collocated columns order is what matters here.
##########
modules/platforms/cpp/ignite/client/detail/table/table_impl.cpp:
##########
@@ -472,4 +584,68 @@ std::shared_ptr<table_impl> table_impl::from_facade(table &tb) {
return tb.m_impl;
}
+void table_impl::update_partition_assignment() {
+ ignite_callback<std::shared_ptr<protocol::partition_assignment>> callback = [self=shared_from_this()](auto pa) {
+ if (pa.has_error()) {
+ self->m_connection->get_logger()->log_error("Error while updating partition assignment for table"
+ + self->get_name() + "error " + pa.error().what_str());
+
+ return;
+ }
+
+ std::lock_guard lock(self->m_partitions_mutex);
+ self->m_partition_assignment = std::move(pa).value();
+ };
+
+ load_partition_assignment_async(std::move(callback));
+}
+
+void table_impl::load_partition_assignment_async(ignite_callback<std::shared_ptr<protocol::partition_assignment>> callback) {
+ std::int64_t timestamp = m_connection->get_assignment_timestamp();
+
+ auto pa = get_partition_assignment();
+ if (pa && !pa->is_outdated(timestamp)) {
+ m_connection->get_logger()->log_debug("Partition assignment for table " + get_name() + " is up to date.");
+
+ callback(std::move(pa));
+ return;
+ }
+
+ auto writer_func = [id = m_id, timestamp](protocol::writer &writer, auto&) {
+ protocol::write_partition_assignment_request(writer, id, timestamp);
+ };
+
+ auto reader_func = [timestamp](protocol::reader &reader) -> std::shared_ptr<protocol::partition_assignment> {
+ return protocol::read_partition_assignment_response(reader, timestamp);
+ };
+
+ m_connection->perform_request<std::shared_ptr<protocol::partition_assignment>>(
+ protocol::client_operation::PARTITION_ASSIGNMENT_GET,
+ nullptr,
+ writer_func,
+ std::move(reader_func),
+ std::move(callback));
+}
+
+std::optional<std::string> table_impl::get_preferred_node_name(const ignite_tuple &key_or_rec, const schema &sch) {
+ auto pa = get_partition_assignment();
+
+ if (!pa || pa->get_partitions().empty()) {
+ m_connection->get_logger()->log_debug("No partition distribution available, random node will called");
+ return {};
+ }
+
+ hash_calculator hc;
+ for (auto column : sch.collocated_columns) {
+ auto val = key_or_rec.get(column->key_index);
Review Comment:
Agreed, the collocated columns order is what matters here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "Erixonich (via GitHub)" <gi...@apache.org>.
Erixonich commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3079092999
##########
modules/platforms/cpp/ignite/protocol/partition_assignment.h:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <optional>
+#include <string>
+#include <vector>
+#include <cstdint>
+
+namespace ignite::protocol {
+
+/**
+ * Partition assignment.
+ */
+class partition_assignment {
+public:
+ partition_assignment() = default;
+ partition_assignment(const partition_assignment&) = default;
+ partition_assignment(partition_assignment&&) noexcept = default;
+ partition_assignment &operator=(const partition_assignment &other) = default;
+ partition_assignment &operator=(partition_assignment &&other) noexcept = default;
+
+ partition_assignment(std::int64_t timestamp, std::vector<std::optional<std::string>> partitions)
+ : timestamp(timestamp)
+ , partitions(std::move(partitions)) {}
+
+ /**
+ * Check whether the assignment is outdated.
+ *
+ * @param actual_timestamp Timestamp.
+ * @return @c true if assignment is outdated.
+ */
+ [[nodiscard]] bool is_outdated(std::int64_t actual_timestamp) const { return timestamp < actual_timestamp; }
+
+ [[nodiscard]] const std::vector<std::optional<std::string>>& get_partitions() const {
+ return partitions;
+ }
+private:
+ /** Assignment timestamp. */
+ std::int64_t timestamp{0};
+
+ /** Partitions. */
+ std::vector<std::optional<std::string>> partitions;
Review Comment:
fixed docs and for getter as well
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "isapego (via GitHub)" <gi...@apache.org>.
isapego commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3080232389
##########
modules/platforms/cpp/ignite/protocol/partition_assignment.h:
##########
@@ -0,0 +1,70 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <optional>
+#include <string>
+#include <vector>
+#include <cstdint>
+
+namespace ignite::protocol {
+
+/**
+ * Partition assignment.
+ */
+class partition_assignment {
+public:
+ partition_assignment() = default;
+
+ /**
+ * Constructor.
+ *
+ * @param timestamp Partition assignment data.
+ * @param partitions Partition assignment encode into vector.
+ */
+ partition_assignment(std::int64_t timestamp, std::vector<std::optional<std::string>> partitions)
+ : m_timestamp(timestamp)
+ , m_partitions(std::move(partitions)) {}
+
+ /**
+ * Check whether the assignment is outdated.
+ *
+ * @param actual_timestamp Timestamp.
+ * @return @c true if assignment is outdated.
+ */
+ [[nodiscard]] bool is_outdated(std::int64_t actual_timestamp) const { return m_timestamp < actual_timestamp; }
+
+ /**
+ * Partitions. Vector is decoded as following: size of collection is total number of partitions,
+ * i-th value means that partition with id == i resides on (primary replica of that partition belongs to)
+ * node with provided consisted id (AKA node name).
+ *
+ * @return Vector with partition assignments.
+ */
+ [[nodiscard]] const std::vector<std::optional<std::string>>& get_partitions() const {
+ return m_partitions;
+ }
+private:
+ /** Assignment timestamp. */
+ std::int64_t m_timestamp{0};
+
+
+ std::vector<std::optional<std::string>> m_partitions;
Review Comment:
Still no documentation
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "isapego (via GitHub)" <gi...@apache.org>.
isapego commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3080280028
##########
modules/platforms/cpp/ignite/common/uuid_test.cpp:
##########
@@ -64,3 +64,54 @@ TEST(uuid, stream) {
EXPECT_EQ(uuidString, uuidString2);
}
+
+TEST(uuid, incorrect_string_representation) {
+ auto uuidText = "foo";
+
+ auto resOpt = ignite::uuid::from_string(uuidText);
+
+ EXPECT_FALSE(resOpt.has_value());
+}
+
+class uuid_string_presentation_fixture: public ::testing::TestWithParam<std::tuple<std::string, ignite::uuid>> {};
+
+TEST_P(uuid_string_presentation_fixture, from_string) {
+ auto [uuidText, uuid] = GetParam();
+
+ auto resOpt = ignite::uuid::from_string(uuidText);
+
+ EXPECT_TRUE(resOpt.has_value());
+
+ EXPECT_EQ(uuid.get_most_significant_bits(), resOpt->get_most_significant_bits());
+ EXPECT_EQ(uuid.get_least_significant_bits(), resOpt->get_least_significant_bits());
Review Comment:
Let's go here string -> uuid -> string.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "Erixonich (via GitHub)" <gi...@apache.org>.
Erixonich commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3080449065
##########
modules/platforms/cpp/ignite/common/uuid_test.cpp:
##########
@@ -64,3 +64,54 @@ TEST(uuid, stream) {
EXPECT_EQ(uuidString, uuidString2);
}
+
+TEST(uuid, incorrect_string_representation) {
+ auto uuidText = "foo";
+
+ auto resOpt = ignite::uuid::from_string(uuidText);
+
+ EXPECT_FALSE(resOpt.has_value());
+}
+
+class uuid_string_presentation_fixture: public ::testing::TestWithParam<std::tuple<std::string, ignite::uuid>> {};
+
+TEST_P(uuid_string_presentation_fixture, from_string) {
+ auto [uuidText, uuid] = GetParam();
+
+ auto resOpt = ignite::uuid::from_string(uuidText);
+
+ EXPECT_TRUE(resOpt.has_value());
+
+ EXPECT_EQ(uuid.get_most_significant_bits(), resOpt->get_most_significant_bits());
+ EXPECT_EQ(uuid.get_least_significant_bits(), resOpt->get_least_significant_bits());
Review Comment:
added new test
##########
modules/platforms/cpp/ignite/common/uuid_test.cpp:
##########
@@ -64,3 +64,54 @@ TEST(uuid, stream) {
EXPECT_EQ(uuidString, uuidString2);
}
+
+TEST(uuid, incorrect_string_representation) {
+ auto uuidText = "foo";
+
+ auto resOpt = ignite::uuid::from_string(uuidText);
+
+ EXPECT_FALSE(resOpt.has_value());
+}
+
+class uuid_string_presentation_fixture: public ::testing::TestWithParam<std::tuple<std::string, ignite::uuid>> {};
+
+TEST_P(uuid_string_presentation_fixture, from_string) {
+ auto [uuidText, uuid] = GetParam();
+
+ auto resOpt = ignite::uuid::from_string(uuidText);
+
+ EXPECT_TRUE(resOpt.has_value());
+
+ EXPECT_EQ(uuid.get_most_significant_bits(), resOpt->get_most_significant_bits());
+ EXPECT_EQ(uuid.get_least_significant_bits(), resOpt->get_least_significant_bits());
+}
+
+// We check that two representation are consistent both ways.
+TEST_P(uuid_string_presentation_fixture, to_string) {
+ auto [uuidText, uuid] = GetParam();
+
+
+ std::stringstream ss;
+ ss << uuid;
+
+ EXPECT_EQ(uuidText, ss.str());
+}
Review Comment:
added new test
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3060488404
##########
modules/platforms/cpp/ignite/protocol/partition_assignment.h:
##########
@@ -0,0 +1,38 @@
+/*
+ * Copyright (C) GridGain Systems. All Rights Reserved.
+ * _________ _____ __________________ _____
+ * __ ____/___________(_)______ /__ ____/______ ____(_)_______
+ * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \
+ * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / /
+ * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/
Review Comment:
The new header uses a proprietary GridGain copyright banner instead of the standard ASF Apache 2.0 header used elsewhere in this repository. This is a licensing/compliance issue; please replace it with the standard ASF license header (matching other C++ headers under modules/platforms/cpp/ignite/...).
```suggestion
* 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.
```
##########
modules/platforms/cpp/ignite/common/uuid.h:
##########
@@ -44,7 +46,60 @@ class uuid {
*/
constexpr uuid(std::int64_t most, std::int64_t least) noexcept
: most(most)
- , least(least) {}
+ , least(least) { }
+
+ /**
+ * Parses string-encoded uuid.
+ * Example of possible input value: ff979603-fb56-49e9-bc79-7c4487bbbafd.
+ * @param text String-encoded uuid.
+ * @return @c uuid object if parsing was successful, otherwise empty optional.
+ */
+ static std::optional<uuid> from_string(const std::string& text) {
+ if (text.length() != 36) {
+ return {};
+ }
+
+ auto str = text.c_str();
+
+ auto parse_chunk = [str](size_t beg, size_t end, int64_t& out) -> bool {
+ char* p;
+ out = std::strtoll(str + beg, &p, 16);
+
+ if (p != str + end || (*p != '-' && *p != '\0') || errno == ERANGE) {
+ return false;
+ }
Review Comment:
`uuid::from_string` uses `errno`/`ERANGE` but this header does not include `<cerrno>`, and `errno` is not reset before calling `strtoll`, so a previous error can cause false negatives. Include `<cerrno>` and set `errno = 0` before each `strtoll` (and/or avoid relying on `errno` unless it’s explicitly cleared).
##########
modules/platforms/cpp/ignite/protocol/partition_assignment.h:
##########
@@ -0,0 +1,38 @@
+/*
+ * Copyright (C) GridGain Systems. All Rights Reserved.
+ * _________ _____ __________________ _____
+ * __ ____/___________(_)______ /__ ____/______ ____(_)_______
+ * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \
+ * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / /
+ * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/
+ */
+
+#pragma once
+
+#include <optional>
+#include <string>
+#include <vector>
+#include <cstdint>
+
+namespace ignite::protocol {
+
+/**
+ * Partition assignment.
+ */
+struct partition_assignment {
+ /** Assignment timestamp. */
+ volatile std::int64_t timestamp{0};
+
Review Comment:
`volatile` does not provide thread-safety and is generally not appropriate for cross-thread synchronization in C++. If this timestamp is accessed concurrently, use `std::atomic<std::int64_t>` (or keep it as a plain `std::int64_t` and ensure all access is externally synchronized).
##########
modules/platforms/cpp/tests/test-common/proxy/message_listener.h:
##########
@@ -0,0 +1,164 @@
+// 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.
+//
+
+#pragma once
+
+#include "ignite/network/length_prefix_codec.h"
+#include "ignite/protocol/client_operation.h"
+#include "ignite/protocol/reader.h"
+
+#include <queue>
+#include <shared_mutex>
+#include <utility>
+
+namespace ignite::proxy {
+
+using raw_message = std::vector<char>;
+
+class structured_message {
+public:
+ structured_message() = default;
+
+ structured_message(const structured_message& structured_message) = default;
+
+ explicit structured_message(std::vector<std::byte> data)
+ :m_data(std::move(data)) { }
+
+ [[nodiscard]] protocol::reader payload_reader() const {
+ return protocol::reader{{m_data.data() + m_payload_pos, m_data.size() - m_payload_pos}};
+ }
+protected:
+ std::vector<std::byte> m_data{};
+ size_t m_payload_pos{};
+};
+
+/**
+ * Message which clients sends to server.
+ */
+class client_message: public structured_message{
+public:
+ client_message() = default;
+
+ client_message(const client_message& msg) = default;
+
+ explicit client_message(std::vector<std::byte> data)
+ : structured_message(std::move(data))
+ {
+ protocol::reader rd{{m_data.data(), m_data.size()}};
+
+ m_op = static_cast<protocol::client_operation>(rd.read_int32());
+ m_req_id = rd.read_int64();
+
+ m_payload_pos = rd.position();
+
+ assert(m_payload_pos < m_data.size());
+ }
Review Comment:
`assert(m_payload_pos < m_data.size())` rejects valid messages with an empty payload (where `m_payload_pos == m_data.size()`). This can trigger in real protocol messages that have only headers. The assert should allow `==` (and ideally also validate that the header fields were fully read before using `m_payload_pos`).
##########
modules/platforms/cpp/ignite/client/detail/cluster_connection.cpp:
##########
@@ -262,9 +274,36 @@ std::shared_ptr<node_connection> cluster_connection::get_random_connected_channe
return std::next(m_connections.begin(), idx)->second;
}
+std::shared_ptr<node_connection> cluster_connection::get_preferred_channel(const std::string& preferred_node_name) {
+ std::unique_lock lock(m_connections_mutex);
+ for (auto& [id, conn] : m_connections) {
+ if (conn->m_node_name == preferred_node_name) {
+ return conn;
+ }
+ }
Review Comment:
`cluster_connection` accesses `conn->m_node_name`, but `m_node_name` is a private member of `node_connection` (declared in node_connection.h). This will not compile without a friend declaration. Expose a public accessor on `node_connection` (e.g., `get_node_name()` / `node_name()`) or make `cluster_connection` a friend if that’s acceptable.
##########
modules/platforms/cpp/ignite/client/detail/table/table_impl.cpp:
##########
@@ -472,4 +482,63 @@ std::shared_ptr<table_impl> table_impl::from_facade(table &tb) {
return tb.m_impl;
}
+void table_impl::update_partition_assignment() {
+ ignite_callback<std::shared_ptr<protocol::partition_assignment>> callback = [self=shared_from_this()](auto pa) {
+ self->m_partition_assignment = pa.value();
+ };
+
+ load_partition_assignment_async(std::move(callback));
+}
+
+void table_impl::load_partition_assignment_async(ignite_callback<std::shared_ptr<protocol::partition_assignment>> callback) {
+ std::int64_t timestamp = m_connection->get_assignment_timestamp();
+
+ {
+ auto pa = get_partition_assignment();
+ if (pa && !pa->is_outdated(timestamp)) {
+ m_connection->get_logger()->log_debug("Partition assignment for table " + get_name() + " is up to date.");
+
+ callback(std::move(pa));
+ return;
+ }
+ }
+
+ auto writer_func = [id = m_id, timestamp](protocol::writer &writer, auto&) {
+ protocol::write_partition_assignment_request(writer, id, timestamp);
+ };
+
+ auto reader_func = [timestamp](protocol::reader &reader) -> std::shared_ptr<protocol::partition_assignment> {
+ return protocol::read_partition_assignment_response(reader, timestamp);
+ };
+
+ m_connection->perform_request<std::shared_ptr<protocol::partition_assignment>>(
+ protocol::client_operation::PARTITION_ASSIGNMENT_GET,
+ nullptr,
+ writer_func,
+ std::move(reader_func),
+ std::move(callback));
+}
+
+std::optional<std::string> table_impl::get_preferred_node_name(const ignite_tuple &key, const schema &sch) {
+ auto pa = get_partition_assignment();
+
+ if (!pa) {
+ return {};
+ }
+
+ hash_calculator hc;
+
+ for (auto column : sch.collocated_columns) {
+ auto val = key.get(column->key_index);
+ hc.append(val, column->scale, column->precision);
+ }
+
+ auto hash = hc.result_hash();
+
+
+ auto part_id = std::abs(hash % static_cast<int32_t>(pa->partitions.size()));
+
+ return pa->partitions[part_id];
Review Comment:
`get_preferred_node_name` can divide by zero when `pa->partitions.size() == 0` (e.g., when the server returns `cnt == 0`, or if an empty assignment is present). This makes `hash % size` undefined and will crash. Add a guard for empty `partitions` before computing the modulo and return `std::nullopt` (falling back to random channel) when mapping is unavailable.
##########
modules/platforms/cpp/tests/test-common/proxy/message_listener.h:
##########
@@ -0,0 +1,164 @@
+// 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.
+//
+
+#pragma once
+
+#include "ignite/network/length_prefix_codec.h"
+#include "ignite/protocol/client_operation.h"
+#include "ignite/protocol/reader.h"
+
+#include <queue>
+#include <shared_mutex>
+#include <utility>
+
+namespace ignite::proxy {
+
+using raw_message = std::vector<char>;
+
+class structured_message {
+public:
+ structured_message() = default;
+
+ structured_message(const structured_message& structured_message) = default;
+
+ explicit structured_message(std::vector<std::byte> data)
+ :m_data(std::move(data)) { }
+
+ [[nodiscard]] protocol::reader payload_reader() const {
+ return protocol::reader{{m_data.data() + m_payload_pos, m_data.size() - m_payload_pos}};
+ }
+protected:
+ std::vector<std::byte> m_data{};
+ size_t m_payload_pos{};
+};
+
+/**
+ * Message which clients sends to server.
+ */
+class client_message: public structured_message{
+public:
+ client_message() = default;
+
+ client_message(const client_message& msg) = default;
+
+ explicit client_message(std::vector<std::byte> data)
+ : structured_message(std::move(data))
+ {
+ protocol::reader rd{{m_data.data(), m_data.size()}};
+
+ m_op = static_cast<protocol::client_operation>(rd.read_int32());
+ m_req_id = rd.read_int64();
+
+ m_payload_pos = rd.position();
+
+ assert(m_payload_pos < m_data.size());
+ }
+
+ [[nodiscard]] protocol::client_operation get_op() const { return m_op; }
+
+ [[nodiscard]] int64_t get_req_id() const { return m_req_id; }
+
+private:
+ protocol::client_operation m_op{};
+ int64_t m_req_id{};
+};
+
+/**
+ * Message which server sends to client.
+ */
+class server_message : public structured_message {
+public:
+ server_message() = default;
+
+ server_message(const server_message&) = default;
+
+ explicit server_message(std::vector<std::byte> data)
+ : structured_message(std::move(data))
+ {
+ protocol::reader rd{{m_data.data(), m_data.size()}};
+
+ m_req_id = rd.read_int64();
+ m_flags = rd.read_int32();
+ m_obs_ts = rd.read_int64();
+
+ m_payload_pos = rd.position();
+
+ assert(m_payload_pos < m_data.size());
+ }
+
+ [[nodiscard]] int64_t get_req_id() const { return m_req_id; }
+
+ [[nodiscard]] int32_t get_flags() const { return m_flags; }
+
+ [[nodiscard]] int64_t get_obs_ts() const { return m_obs_ts; }
+
+private:
+ int64_t m_req_id{};
+ int32_t m_flags{};
+ int64_t m_obs_ts{};
+};
+
+/**
+ * Intercepts messages which go through @c asio_proxy.
+ */
+class message_listener {
+public:
+ void register_message(raw_message msg) {
+ std::unique_lock lock(m_mutex);
+ m_queue.push(std::move(msg));
+ }
+
+ [[nodiscard]] std::queue<raw_message> get_msg_queue() const {
+ std::shared_lock lock(m_mutex);
+ return m_queue;
+ }
+
+ template<typename MESSAGE_TYPE>
+ std::vector<MESSAGE_TYPE> get_next() {
+ std::unique_lock lock(m_mutex);
+
+ std::vector<MESSAGE_TYPE> res;
+
+ while (!m_queue.empty() && res.empty()) {
+ auto chunk = m_queue.front();
+ m_queue.pop();
+
+ network::data_buffer_ref buf{{chunk.data(), chunk.size()}};
Review Comment:
`get_next()` copies the queued buffer (`auto chunk = m_queue.front();`) and keeps the mutex locked while decoding/parsing messages. This is avoidable overhead and can block producers unnecessarily. Move the chunk out of the queue (e.g., `std::move(m_queue.front())`) and consider releasing the lock before decoding to minimize contention.
##########
modules/platforms/cpp/tests/test-common/proxy/message_listener.h:
##########
@@ -0,0 +1,164 @@
+// 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.
+//
+
+#pragma once
+
+#include "ignite/network/length_prefix_codec.h"
+#include "ignite/protocol/client_operation.h"
+#include "ignite/protocol/reader.h"
+
+#include <queue>
+#include <shared_mutex>
+#include <utility>
+
+namespace ignite::proxy {
+
+using raw_message = std::vector<char>;
+
+class structured_message {
+public:
+ structured_message() = default;
+
+ structured_message(const structured_message& structured_message) = default;
+
+ explicit structured_message(std::vector<std::byte> data)
+ :m_data(std::move(data)) { }
+
+ [[nodiscard]] protocol::reader payload_reader() const {
+ return protocol::reader{{m_data.data() + m_payload_pos, m_data.size() - m_payload_pos}};
+ }
+protected:
+ std::vector<std::byte> m_data{};
+ size_t m_payload_pos{};
+};
+
+/**
+ * Message which clients sends to server.
+ */
+class client_message: public structured_message{
+public:
+ client_message() = default;
+
+ client_message(const client_message& msg) = default;
+
+ explicit client_message(std::vector<std::byte> data)
+ : structured_message(std::move(data))
+ {
+ protocol::reader rd{{m_data.data(), m_data.size()}};
+
+ m_op = static_cast<protocol::client_operation>(rd.read_int32());
+ m_req_id = rd.read_int64();
+
+ m_payload_pos = rd.position();
+
+ assert(m_payload_pos < m_data.size());
+ }
+
+ [[nodiscard]] protocol::client_operation get_op() const { return m_op; }
+
+ [[nodiscard]] int64_t get_req_id() const { return m_req_id; }
+
+private:
+ protocol::client_operation m_op{};
+ int64_t m_req_id{};
+};
+
+/**
+ * Message which server sends to client.
+ */
+class server_message : public structured_message {
+public:
+ server_message() = default;
+
+ server_message(const server_message&) = default;
+
+ explicit server_message(std::vector<std::byte> data)
+ : structured_message(std::move(data))
+ {
+ protocol::reader rd{{m_data.data(), m_data.size()}};
+
+ m_req_id = rd.read_int64();
+ m_flags = rd.read_int32();
+ m_obs_ts = rd.read_int64();
+
+ m_payload_pos = rd.position();
+
+ assert(m_payload_pos < m_data.size());
+ }
Review Comment:
Same issue as in `client_message`: `assert(m_payload_pos < m_data.size())` will fail for responses with no payload (payload position equals buffer size). This should allow empty payloads (use `<=`) so the proxy can inspect any operation safely.
##########
modules/platforms/cpp/tests/client-test/partition_awareness_test.cpp:
##########
@@ -0,0 +1,296 @@
+// 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 "detail/string_extensions.h"
+#include "ignite_runner_suite.h"
+#include "tests/test-common/test_utils.h"
+
+#include "ignite/client/detail/utils.h"
+#include "ignite/client/ignite_client.h"
+#include "ignite/client/ignite_client_configuration.h"
+
+#include "proxy/asio_proxy.h"
+#include "proxy/message_listener.h"
+
+#include <gtest/gtest.h>
+
+namespace ignite {
+
+using namespace std::chrono_literals;
+
+using part_id_type = std::size_t;
+using key_type = std::int64_t;
+
+constexpr key_type MIN_KEY = 0;
+constexpr key_type MAX_KEY = 1000;
+
+static const std::map<std::string, std::string> NODES_TO_ADDR = {
+ {"org.apache.ignite.internal.runner.app.PlatformTestNodeRunner", "127.0.0.1:10942"},
+ {"org.apache.ignite.internal.runner.app.PlatformTestNodeRunner_2", "127.0.0.1:10943"}
+};
+
+/**
+ * Parses string-encoded partition distribution.
+ * Example of string:
+ * ff979603-fb56-49e9-bc79-7c4487bbbafd=[4];2e14bb82-5cb5-4283-aafd-f2d8552a842a=[5, 8, 9];
+ * @param encoded String-encoded distribution
+ * @return Conveniently structured distribution.
+ */
+std::map<int64_t, uuid> parse_partition_distribution(const std::string& encoded) {
+ std::map<int64_t, uuid> res;
+
+ size_t lhs = 0;
+ size_t rhs = std::string::npos;
+ while ((rhs = encoded.find(';', lhs)) != std::string::npos) {
+ std::string chunk= encoded.substr(lhs, rhs - lhs);
+
+ char eq_pos = chunk.find('=');
+
Review Comment:
`chunk.find('=')` returns `size_t`, but the result is stored in `char eq_pos`. This truncates the position (and turns `npos` into an arbitrary small value), causing incorrect parsing and potentially out-of-range substrings. Use `size_t` for `eq_pos` and explicitly handle the `npos` case.
```suggestion
std::string::size_type eq_pos = chunk.find('=');
if (eq_pos == std::string::npos) {
throw std::runtime_error("can't parse partition distribution, incorrect input:" + encoded);
}
```
##########
modules/platforms/cpp/ignite/client/detail/node_connection.h:
##########
@@ -157,6 +157,8 @@ class node_connection : public std::enable_shared_from_this<node_connection> {
bool sent = m_pool->send(m_id, std::move(message));
if (!sent) {
get_and_remove_handler(req_id);
+ /*m_logger->log_error(*/auto str= "Connection id=" + std::to_string(id()) + " req_id=" + std::to_string(req_id) + " failed";//);
+ std::cout << str << std::endl;
Review Comment:
This introduces direct `std::cout` printing (and commented-out logger code) in a core request path. Besides being noisy in production/tests, it can break builds (header doesn’t include `<iostream>`) and bypasses the existing logging/error handling conventions. Please remove the `std::cout` usage and report the send failure via `m_logger` and/or by propagating an `ignite_error` to the handler/callback.
```suggestion
m_logger->log_error(
"Connection id=" + std::to_string(id()) + " req_id=" + std::to_string(req_id) + " failed");
```
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/PlatformTestNodeRunner.java:
##########
@@ -119,6 +120,7 @@
import org.apache.ignite.table.Table;
import org.apache.ignite.table.Tuple;
import org.apache.ignite.table.partition.Partition;
+import org.apache.ignite.table.partition.PartitionDistribution;
import org.jetbrains.annotations.Nullable;
Review Comment:
Unused imports were added (`java.util.HashMap` and `org.apache.ignite.table.partition.PartitionDistribution`) but are not referenced anywhere in this file. This will typically fail compilation/checkstyle with unused-import rules; please remove them or use them as intended.
##########
modules/platforms/cpp/ignite/client/detail/table/table_impl.cpp:
##########
@@ -472,4 +482,63 @@ std::shared_ptr<table_impl> table_impl::from_facade(table &tb) {
return tb.m_impl;
}
+void table_impl::update_partition_assignment() {
+ ignite_callback<std::shared_ptr<protocol::partition_assignment>> callback = [self=shared_from_this()](auto pa) {
Review Comment:
`update_partition_assignment()` callback assumes success (`pa.value()`) and writes `m_partition_assignment` without taking `m_partitions_mutex`. This can both crash on errors (value() when `has_error()`) and cause a data race with `get_partition_assignment()` which *does* lock the mutex. Handle the error case (propagate/log) and update `m_partition_assignment` under `m_partitions_mutex`.
```suggestion
ignite_callback<std::shared_ptr<protocol::partition_assignment>> callback = [self = shared_from_this()](auto pa) {
if (pa.has_error()) {
self->m_connection->get_logger()->log_error(
"Failed to update partition assignment for table " + self->get_name() + ".");
return;
}
std::lock_guard lock(self->m_partitions_mutex);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "Erixonich (via GitHub)" <gi...@apache.org>.
Erixonich commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3079094331
##########
modules/platforms/cpp/ignite/protocol/partition_assignment.h:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <optional>
+#include <string>
+#include <vector>
+#include <cstdint>
+
+namespace ignite::protocol {
+
+/**
+ * Partition assignment.
+ */
+class partition_assignment {
+public:
+ partition_assignment() = default;
+ partition_assignment(const partition_assignment&) = default;
+ partition_assignment(partition_assignment&&) noexcept = default;
+ partition_assignment &operator=(const partition_assignment &other) = default;
+ partition_assignment &operator=(partition_assignment &&other) noexcept = default;
+
+ partition_assignment(std::int64_t timestamp, std::vector<std::optional<std::string>> partitions)
+ : timestamp(timestamp)
+ , partitions(std::move(partitions)) {}
+
+ /**
+ * Check whether the assignment is outdated.
+ *
+ * @param actual_timestamp Timestamp.
+ * @return @c true if assignment is outdated.
+ */
+ [[nodiscard]] bool is_outdated(std::int64_t actual_timestamp) const { return timestamp < actual_timestamp; }
+
+ [[nodiscard]] const std::vector<std::optional<std::string>>& get_partitions() const {
Review Comment:
done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "Erixonich (via GitHub)" <gi...@apache.org>.
Erixonich commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3079117270
##########
modules/platforms/cpp/ignite/protocol/messages.cpp:
##########
@@ -100,4 +100,39 @@ handshake_response parse_handshake_response(bytes_view message) {
return res;
}
+void write_partition_assignment_request(writer &writer, std::int32_t table_id, std::int64_t timestamp) {
+ writer.write(table_id);
+ writer.write(timestamp);
+}
+
+std::shared_ptr<partition_assignment> read_partition_assignment_response(reader &reader, std::int64_t timestamp) {
+ auto cnt = reader.read_int32();
+ if (cnt <= 0)
+ throw ignite_error("Invalid partition count: " + std::to_string(cnt));
Review Comment:
I've synchronized with what java client does:
```
int cnt = r.in().unpackInt();
if (cnt <= 0) {
throw new IgniteException(INTERNAL_ERR, "Invalid partition count returned by the server: " + cnt);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "Erixonich (via GitHub)" <gi...@apache.org>.
Erixonich commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3079157098
##########
modules/platforms/cpp/ignite/protocol/messages.cpp:
##########
@@ -100,4 +100,39 @@ handshake_response parse_handshake_response(bytes_view message) {
return res;
}
+void write_partition_assignment_request(writer &writer, std::int32_t table_id, std::int64_t timestamp) {
+ writer.write(table_id);
+ writer.write(timestamp);
+}
+
+std::shared_ptr<partition_assignment> read_partition_assignment_response(reader &reader, std::int64_t timestamp) {
+ auto cnt = reader.read_int32();
+ if (cnt <= 0)
+ throw ignite_error("Invalid partition count: " + std::to_string(cnt));
+
+ std::vector<std::optional<std::string>> partitions;
+ partitions.reserve(cnt);
+
+ bool assignment_available = reader.read_bool();
+ if (!assignment_available) {
+ // Invalidate the current assignment so that we can retry on the next call.
+ // Return an empty array so that per-partition batches can be initialized.
+ // We'll get the actual assignment on the next call.
+ partitions.insert(partitions.end(), cnt, std::nullopt);
+ return std::make_shared<partition_assignment>(0, std::move(partitions));
+ }
+
+ // Returned timestamp can be newer than requested.
+ std::int64_t ts = reader.read_int64();
+ if (ts < timestamp)
+ throw ignite_error("Returned timestamp is older than requested: " + std::to_string(ts) + " < "
+ + std::to_string(timestamp));
Review Comment:
this is to discuss, but reference java client will halt in this case (assertion is used in this case).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "Erixonich (via GitHub)" <gi...@apache.org>.
Erixonich commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3079314728
##########
modules/platforms/cpp/ignite/common/uuid.h:
##########
@@ -46,6 +50,64 @@ class uuid {
: most(most)
, least(least) {}
+ /**
+ * Parses string-encoded uuid.
+ * Example of possible input value: ff979603-fb56-49e9-bc79-7c4487bbbafd.
+ * @param text String-encoded uuid.
+ * @return @c uuid object if parsing was successful, otherwise empty optional.
+ */
+ static std::optional<uuid> from_string(const std::string& text) {
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "isapego (via GitHub)" <gi...@apache.org>.
isapego commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3080264807
##########
modules/platforms/cpp/ignite/common/uuid_test.cpp:
##########
@@ -64,3 +64,54 @@ TEST(uuid, stream) {
EXPECT_EQ(uuidString, uuidString2);
}
+
+TEST(uuid, incorrect_string_representation) {
+ auto uuidText = "foo";
+
+ auto resOpt = ignite::uuid::from_string(uuidText);
+
+ EXPECT_FALSE(resOpt.has_value());
+}
+
+class uuid_string_presentation_fixture: public ::testing::TestWithParam<std::tuple<std::string, ignite::uuid>> {};
+
+TEST_P(uuid_string_presentation_fixture, from_string) {
+ auto [uuidText, uuid] = GetParam();
+
+ auto resOpt = ignite::uuid::from_string(uuidText);
+
+ EXPECT_TRUE(resOpt.has_value());
+
+ EXPECT_EQ(uuid.get_most_significant_bits(), resOpt->get_most_significant_bits());
+ EXPECT_EQ(uuid.get_least_significant_bits(), resOpt->get_least_significant_bits());
+}
+
+// We check that two representation are consistent both ways.
+TEST_P(uuid_string_presentation_fixture, to_string) {
+ auto [uuidText, uuid] = GetParam();
+
+
+ std::stringstream ss;
+ ss << uuid;
+
+ EXPECT_EQ(uuidText, ss.str());
+}
+
+// Values has taken randomly from java program, as this implementation should have same behaviour as java.lang.UUID for compatibility reasons.
+INSTANTIATE_TEST_SUITE_P(
+ various_uuids,
+ uuid_string_presentation_fixture,
+ ::testing::Values(
+ std::tuple{"4b62e46a-d380-460f-94ea-9b4320752634", ignite::uuid{5432155248028304911LL, -7716184298936261068LL}},
+ std::tuple{"cabf7114-7763-4ad5-b312-5ffadcbf48c8", ignite::uuid{-3837224024780092715, -5543262660289673016}},
+ std::tuple{"8ee62c79-5636-44b9-adc4-c9b4272a2ba6", ignite::uuid{-8149777576031271751, -5925389434124358746}},
+ std::tuple{"3e69a946-fc2a-4da1-9938-d1f082dd36ce", ignite::uuid{4497311825249586593, -7405938756292888882}},
+ std::tuple{"7cc9f130-821b-4580-bfda-955072540643", ignite::uuid{8991983321665455488, -4622217894794361277}},
+ std::tuple{"1f86b0b1-dfab-4108-8b7e-8b3e4177ceb0", ignite::uuid{2271697340063236360, -8395119555869421904}},
+ std::tuple{"88cc69d9-d191-482d-b295-aa458cf99167", ignite::uuid{-8589374005057599443, -5578365347733859993}},
+ std::tuple{"a8b6298c-126e-4c4d-a521-de3989a31492", ignite::uuid{-6289794147994940339, -6547708044516322158}},
+ std::tuple{"03b210a8-5a6a-4cae-b4fc-f30df9f1eebd", ignite::uuid{266293643225746606, -5405178211397931331}},
+ std::tuple{"f145e277-bb58-4f89-adfa-3166a975b71b", ignite::uuid{-1061193133303771255, -5910357243970865381}},
+ std::tuple{"dc8b5857-067f-486e-aa5b-52cbdb26dd9c", ignite::uuid{-2554851232808220562, -6171247828872536676}}
Review Comment:
Why do we pass uuid values here? Let's get them using `from_string`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "Erixonich (via GitHub)" <gi...@apache.org>.
Erixonich commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3080369894
##########
modules/platforms/cpp/ignite/protocol/partition_assignment.h:
##########
@@ -0,0 +1,70 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <optional>
+#include <string>
+#include <vector>
+#include <cstdint>
+
+namespace ignite::protocol {
+
+/**
+ * Partition assignment.
+ */
+class partition_assignment {
+public:
+ partition_assignment() = default;
+
+ /**
+ * Constructor.
+ *
+ * @param timestamp Partition assignment data.
+ * @param partitions Partition assignment encode into vector.
+ */
+ partition_assignment(std::int64_t timestamp, std::vector<std::optional<std::string>> partitions)
+ : m_timestamp(timestamp)
+ , m_partitions(std::move(partitions)) {}
+
+ /**
+ * Check whether the assignment is outdated.
+ *
+ * @param actual_timestamp Timestamp.
+ * @return @c true if assignment is outdated.
+ */
+ [[nodiscard]] bool is_outdated(std::int64_t actual_timestamp) const { return m_timestamp < actual_timestamp; }
+
+ /**
+ * Partitions. Vector is decoded as following: size of collection is total number of partitions,
+ * i-th value means that partition with id == i resides on (primary replica of that partition belongs to)
+ * node with provided consisted id (AKA node name).
+ *
+ * @return Vector with partition assignments.
+ */
+ [[nodiscard]] const std::vector<std::optional<std::string>>& get_partitions() const {
+ return m_partitions;
+ }
+private:
+ /** Assignment timestamp. */
+ std::int64_t m_timestamp{0};
+
+
+ std::vector<std::optional<std::string>> m_partitions;
Review Comment:
fixed for sure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "Erixonich (via GitHub)" <gi...@apache.org>.
Erixonich commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3070563437
##########
modules/platforms/cpp/ignite/client/detail/table/table_impl.cpp:
##########
@@ -472,4 +584,68 @@ std::shared_ptr<table_impl> table_impl::from_facade(table &tb) {
return tb.m_impl;
}
+void table_impl::update_partition_assignment() {
+ ignite_callback<std::shared_ptr<protocol::partition_assignment>> callback = [self=shared_from_this()](auto pa) {
+ if (pa.has_error()) {
+ self->m_connection->get_logger()->log_error("Error while updating partition assignment for table"
+ + self->get_name() + "error " + pa.error().what_str());
+
+ return;
+ }
+
+ std::lock_guard lock(self->m_partitions_mutex);
+ self->m_partition_assignment = std::move(pa).value();
+ };
+
+ load_partition_assignment_async(std::move(callback));
+}
+
+void table_impl::load_partition_assignment_async(ignite_callback<std::shared_ptr<protocol::partition_assignment>> callback) {
+ std::int64_t timestamp = m_connection->get_assignment_timestamp();
+
+ auto pa = get_partition_assignment();
+ if (pa && !pa->is_outdated(timestamp)) {
+ m_connection->get_logger()->log_debug("Partition assignment for table " + get_name() + " is up to date.");
+
+ callback(std::move(pa));
+ return;
+ }
+
+ auto writer_func = [id = m_id, timestamp](protocol::writer &writer, auto&) {
+ protocol::write_partition_assignment_request(writer, id, timestamp);
+ };
+
+ auto reader_func = [timestamp](protocol::reader &reader) -> std::shared_ptr<protocol::partition_assignment> {
+ return protocol::read_partition_assignment_response(reader, timestamp);
+ };
+
+ m_connection->perform_request<std::shared_ptr<protocol::partition_assignment>>(
+ protocol::client_operation::PARTITION_ASSIGNMENT_GET,
+ nullptr,
+ writer_func,
+ std::move(reader_func),
+ std::move(callback));
+}
+
+std::optional<std::string> table_impl::get_preferred_node_name(const ignite_tuple &key_or_rec, const schema &sch) {
+ auto pa = get_partition_assignment();
+
+ if (!pa || pa->get_partitions().empty()) {
+ m_connection->get_logger()->log_debug("No partition distribution available, random node will called");
+ return {};
+ }
+
+ hash_calculator hc;
+ for (auto column : sch.collocated_columns) {
+ auto val = key_or_rec.get(column->key_index);
Review Comment:
It is doesnt matter, order is defined by order of schema::collocated_columns. Besides when calling ignite_tuple::get(std::string_view) it internaly does same but before calculation for for idx value.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3070841779
##########
modules/platforms/cpp/tests/client-test/partition_awareness_test.cpp:
##########
@@ -0,0 +1,737 @@
+// 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 "detail/string_extensions.h"
+#include "ignite_runner_suite.h"
+#include "tests/test-common/test_utils.h"
+
+#include "ignite/client/detail/utils.h"
+#include "ignite/client/ignite_client.h"
+#include "ignite/client/ignite_client_configuration.h"
+
+#include "proxy/asio_proxy.h"
+#include "proxy/message_listener.h"
+
+#include <gtest/gtest.h>
+
+namespace ignite {
+
+using namespace std::chrono_literals;
+
+using part_id_type = std::size_t;
+using id_type = std::int64_t;
+
+constexpr id_type MIN_ID = 0;
+constexpr id_type MAX_ID = 1000;
+
+static const std::map<std::string, std::string> NODES_TO_ADDR = {
+ {"org.apache.ignite.internal.runner.app.PlatformTestNodeRunner", "127.0.0.1:10942"},
+ {"org.apache.ignite.internal.runner.app.PlatformTestNodeRunner_2", "127.0.0.1:10943"}
+};
+
+/**
+ * Parses string-encoded partition distribution.
+ * Example of string:
+ * ff979603-fb56-49e9-bc79-7c4487bbbafd=[4];2e14bb82-5cb5-4283-aafd-f2d8552a842a=[5, 8, 9];
+ * @param encoded String-encoded distribution
+ * @return Conveniently structured distribution.
+ */
+std::map<int64_t, uuid> parse_partition_distribution(const std::string& encoded) {
+ std::map<int64_t, uuid> res;
+
+ size_t lhs = 0;
+ size_t rhs = std::string::npos;
+ while ((rhs = encoded.find(';', lhs)) != std::string::npos) {
+ std::string chunk = encoded.substr(lhs, rhs - lhs);
+
+ auto eq_pos = chunk.find('=');
+
+ if (eq_pos == std::string::npos) {
+ throw std::runtime_error("Incorrect partition distribution text:" + encoded);
+ }
+
+ auto node_id = uuid::from_string(chunk.substr(0, eq_pos));
+
+ if (!node_id.has_value()) {
+ throw std::runtime_error("can't parse partition distribution, incorrect input:" + encoded);
+ }
+
+ std::string num_list = chunk.substr(eq_pos + 1); // e.g. [1,2,3]
+
+ size_t lo = 1;
+ size_t hi = std::string::npos;
+ while ((hi = num_list.find_first_of(",]", lo)) != std::string::npos) {
+ int64_t part_id = std::stoll(num_list.substr(lo, hi - lo));
+ res[part_id] = *node_id;
+ lo = hi + 1;
+ }
+ lhs = rhs + 1;
+ }
+
+ return res;
+}
+
+template<typename T>
+class partition_awareness_test : public ignite_runner_suite {
+private:
+ T tab_info{};
+protected:
+ void SetUp() override {
+ ignite_client_configuration client_cfg;
+ client_cfg.set_endpoints(get_node_addrs());
+
+ m_direct_client = ignite_client::start(client_cfg, 5s);
+
+ auto nodes = m_direct_client.get_cluster_nodes();
+
+ for (auto& node : nodes) {
+ if (NODES_TO_ADDR.count(node.get_name())) {
+ m_endpoint_to_node_id[NODES_TO_ADDR.at(node.get_name())] = node.get_id();
+ }
+ }
+
+ ASSERT_EQ(m_endpoint_to_node_id.size(), get_node_addrs().size())
+ << "Incorrect node names or address!"
+ << "This test should be able to connect to all non-ssl nodes";
+
+ drop_table_if_exits();
+
+ create_table();
+
+ collect_partition_distribution();
+
+ setup_proxy();
+ }
+
+ void TearDown() override {
+ drop_table_if_exits();
+ }
+
+ ignite_tuple key_tup(id_type key) {
+ return get_tuple(tab_info.key_column_names, tab_info.get_pk_vals(key));
+ }
+
+ ignite_tuple main_val_tup(id_type key) {
+ return get_tuple(tab_info.non_key_column_names, tab_info.get_main_vals(key));
+ }
+
+ ignite_tuple alt_val_tup(id_type key) {
+ return get_tuple(tab_info.non_key_column_names, tab_info.get_alt_vals(key));
+ }
+
+ ignite_tuple main_rec(id_type key) {
+ return get_tuple(tab_info.all_column_names, tab_info.get_main_rec(key));
+ }
+
+ ignite_tuple alt_rec(id_type key) {
+ return get_tuple(tab_info.all_column_names, tab_info.get_main_rec(key));
Review Comment:
`alt_rec` returns `get_main_rec` instead of `get_alt_rec`, so tests that rely on an alternate record (e.g. replace / upsert paths) end up using the main record and lose coverage (and can mask bugs).
```suggestion
return get_tuple(tab_info.all_column_names, tab_info.get_alt_rec(key));
```
##########
modules/platforms/cpp/ignite/common/uuid.h:
##########
@@ -17,9 +17,12 @@
#pragma once
+#include <cerrno>
#include <cstdint>
+#include <cstdlib>
#include <iomanip>
#include <istream>
+#include <optional>
#include <ostream>
Review Comment:
`uuid::from_string` uses `std::string`, but this header doesn't include `<string>`. Relying on indirect includes can break compilation depending on standard library / build flags; include `<string>` directly here to make the header self-contained.
```suggestion
#include <ostream>
#include <string>
```
##########
modules/platforms/cpp/ignite/client/detail/table/table_impl.cpp:
##########
@@ -472,4 +584,68 @@ std::shared_ptr<table_impl> table_impl::from_facade(table &tb) {
return tb.m_impl;
}
+void table_impl::update_partition_assignment() {
+ ignite_callback<std::shared_ptr<protocol::partition_assignment>> callback = [self=shared_from_this()](auto pa) {
+ if (pa.has_error()) {
+ self->m_connection->get_logger()->log_error("Error while updating partition assignment for table"
+ + self->get_name() + "error " + pa.error().what_str());
Review Comment:
Log message concatenation is missing spaces and reads awkwardly ("... for table<name>error ..."). This makes logs harder to scan and grep; include separators and consider standardizing the wording.
```suggestion
self->m_connection->get_logger()->log_error(
"Error while updating partition assignment for table '" + self->get_name() + "': "
+ pa.error().what_str());
```
##########
modules/platforms/cpp/tests/client-test/partition_awareness_test.cpp:
##########
@@ -0,0 +1,737 @@
+// 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 "detail/string_extensions.h"
+#include "ignite_runner_suite.h"
+#include "tests/test-common/test_utils.h"
+
+#include "ignite/client/detail/utils.h"
+#include "ignite/client/ignite_client.h"
+#include "ignite/client/ignite_client_configuration.h"
+
+#include "proxy/asio_proxy.h"
+#include "proxy/message_listener.h"
+
+#include <gtest/gtest.h>
+
+namespace ignite {
+
+using namespace std::chrono_literals;
+
+using part_id_type = std::size_t;
+using id_type = std::int64_t;
+
+constexpr id_type MIN_ID = 0;
+constexpr id_type MAX_ID = 1000;
+
+static const std::map<std::string, std::string> NODES_TO_ADDR = {
+ {"org.apache.ignite.internal.runner.app.PlatformTestNodeRunner", "127.0.0.1:10942"},
+ {"org.apache.ignite.internal.runner.app.PlatformTestNodeRunner_2", "127.0.0.1:10943"}
+};
+
+/**
+ * Parses string-encoded partition distribution.
+ * Example of string:
+ * ff979603-fb56-49e9-bc79-7c4487bbbafd=[4];2e14bb82-5cb5-4283-aafd-f2d8552a842a=[5, 8, 9];
+ * @param encoded String-encoded distribution
+ * @return Conveniently structured distribution.
+ */
+std::map<int64_t, uuid> parse_partition_distribution(const std::string& encoded) {
+ std::map<int64_t, uuid> res;
+
+ size_t lhs = 0;
+ size_t rhs = std::string::npos;
+ while ((rhs = encoded.find(';', lhs)) != std::string::npos) {
+ std::string chunk = encoded.substr(lhs, rhs - lhs);
+
+ auto eq_pos = chunk.find('=');
+
+ if (eq_pos == std::string::npos) {
+ throw std::runtime_error("Incorrect partition distribution text:" + encoded);
+ }
+
+ auto node_id = uuid::from_string(chunk.substr(0, eq_pos));
+
+ if (!node_id.has_value()) {
+ throw std::runtime_error("can't parse partition distribution, incorrect input:" + encoded);
+ }
+
+ std::string num_list = chunk.substr(eq_pos + 1); // e.g. [1,2,3]
+
+ size_t lo = 1;
+ size_t hi = std::string::npos;
+ while ((hi = num_list.find_first_of(",]", lo)) != std::string::npos) {
+ int64_t part_id = std::stoll(num_list.substr(lo, hi - lo));
+ res[part_id] = *node_id;
+ lo = hi + 1;
+ }
+ lhs = rhs + 1;
+ }
+
+ return res;
+}
+
+template<typename T>
+class partition_awareness_test : public ignite_runner_suite {
+private:
+ T tab_info{};
+protected:
+ void SetUp() override {
+ ignite_client_configuration client_cfg;
+ client_cfg.set_endpoints(get_node_addrs());
+
+ m_direct_client = ignite_client::start(client_cfg, 5s);
+
+ auto nodes = m_direct_client.get_cluster_nodes();
+
+ for (auto& node : nodes) {
+ if (NODES_TO_ADDR.count(node.get_name())) {
+ m_endpoint_to_node_id[NODES_TO_ADDR.at(node.get_name())] = node.get_id();
+ }
+ }
+
+ ASSERT_EQ(m_endpoint_to_node_id.size(), get_node_addrs().size())
+ << "Incorrect node names or address!"
+ << "This test should be able to connect to all non-ssl nodes";
+
+ drop_table_if_exits();
+
+ create_table();
+
+ collect_partition_distribution();
+
+ setup_proxy();
+ }
+
+ void TearDown() override {
+ drop_table_if_exits();
+ }
+
+ ignite_tuple key_tup(id_type key) {
+ return get_tuple(tab_info.key_column_names, tab_info.get_pk_vals(key));
+ }
+
+ ignite_tuple main_val_tup(id_type key) {
+ return get_tuple(tab_info.non_key_column_names, tab_info.get_main_vals(key));
+ }
+
+ ignite_tuple alt_val_tup(id_type key) {
+ return get_tuple(tab_info.non_key_column_names, tab_info.get_alt_vals(key));
+ }
+
+ ignite_tuple main_rec(id_type key) {
+ return get_tuple(tab_info.all_column_names, tab_info.get_main_rec(key));
+ }
+
+ ignite_tuple alt_rec(id_type key) {
+ return get_tuple(tab_info.all_column_names, tab_info.get_main_rec(key));
+ }
+
+ void create_table() {
+ m_direct_client.get_sql().execute(nullptr, nullptr, {tab_info.create_table_ddl()}, {});
+ }
+
+ void drop_table_if_exits() {
Review Comment:
Method name `drop_table_if_exits` is misspelled ("exits" vs "exists"), which makes the test fixture harder to read/search. Consider renaming to `drop_table_if_exists` (and update call sites).
```suggestion
void drop_table_if_exists() {
```
##########
modules/platforms/cpp/ignite/client/detail/table/table_impl.cpp:
##########
@@ -472,4 +584,68 @@ std::shared_ptr<table_impl> table_impl::from_facade(table &tb) {
return tb.m_impl;
}
+void table_impl::update_partition_assignment() {
+ ignite_callback<std::shared_ptr<protocol::partition_assignment>> callback = [self=shared_from_this()](auto pa) {
+ if (pa.has_error()) {
+ self->m_connection->get_logger()->log_error("Error while updating partition assignment for table"
+ + self->get_name() + "error " + pa.error().what_str());
+
+ return;
+ }
+
+ std::lock_guard lock(self->m_partitions_mutex);
+ self->m_partition_assignment = std::move(pa).value();
+ };
+
+ load_partition_assignment_async(std::move(callback));
+}
+
+void table_impl::load_partition_assignment_async(ignite_callback<std::shared_ptr<protocol::partition_assignment>> callback) {
+ std::int64_t timestamp = m_connection->get_assignment_timestamp();
+
+ auto pa = get_partition_assignment();
+ if (pa && !pa->is_outdated(timestamp)) {
+ m_connection->get_logger()->log_debug("Partition assignment for table " + get_name() + " is up to date.");
+
+ callback(std::move(pa));
+ return;
+ }
+
+ auto writer_func = [id = m_id, timestamp](protocol::writer &writer, auto&) {
+ protocol::write_partition_assignment_request(writer, id, timestamp);
+ };
+
+ auto reader_func = [timestamp](protocol::reader &reader) -> std::shared_ptr<protocol::partition_assignment> {
+ return protocol::read_partition_assignment_response(reader, timestamp);
+ };
+
+ m_connection->perform_request<std::shared_ptr<protocol::partition_assignment>>(
+ protocol::client_operation::PARTITION_ASSIGNMENT_GET,
+ nullptr,
+ writer_func,
+ std::move(reader_func),
+ std::move(callback));
+}
+
+std::optional<std::string> table_impl::get_preferred_node_name(const ignite_tuple &key_or_rec, const schema &sch) {
+ auto pa = get_partition_assignment();
+
+ if (!pa || pa->get_partitions().empty()) {
+ m_connection->get_logger()->log_debug("No partition distribution available, random node will called");
Review Comment:
Log message grammar: "random node will called" is ungrammatical and also missing a word ("be"). Fixing this improves log clarity when partition assignment isn't available.
```suggestion
m_connection->get_logger()->log_debug("No partition distribution available, a random node will be called");
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "isapego (via GitHub)" <gi...@apache.org>.
isapego commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3077787797
##########
modules/platforms/cpp/tests/test-common/proxy/message_listener.h:
##########
@@ -0,0 +1,178 @@
+// 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.
+//
+
+#pragma once
+
+#include "ignite/network/length_prefix_codec.h"
+#include "ignite/protocol/client_operation.h"
+#include "ignite/protocol/reader.h"
+
+#include <atomic>
+#include <cassert>
+#include <queue>
+#include <shared_mutex>
+#include <utility>
+
+namespace ignite::proxy {
+
+using raw_message = std::vector<char>;
+
+class structured_message {
+public:
+ structured_message() = default;
+
+ structured_message(const structured_message& structured_message) = default;
+
+ explicit structured_message(std::vector<std::byte> data)
+ :m_data(std::move(data)) { }
+
+ [[nodiscard]] protocol::reader payload_reader() const {
+ return protocol::reader{{m_data.data() + m_payload_pos, m_data.size() - m_payload_pos}};
+ }
+protected:
+ std::vector<std::byte> m_data{};
+ size_t m_payload_pos{};
+};
+
+/**
+ * Message which clients sends to server.
+ */
+class client_message: public structured_message{
+public:
+ client_message() = default;
+
+ client_message(const client_message& msg) = default;
+
+ explicit client_message(std::vector<std::byte> data)
+ : structured_message(std::move(data))
+ {
+ protocol::reader rd{{m_data.data(), m_data.size()}};
+
+ m_op = static_cast<protocol::client_operation>(rd.read_int32());
+ m_req_id = rd.read_int64();
+
+ m_payload_pos = rd.position();
+
+ assert(m_payload_pos <= m_data.size());
+ }
+
+ [[nodiscard]] protocol::client_operation get_op() const { return m_op; }
+
+ [[nodiscard]] int64_t get_req_id() const { return m_req_id; }
+
+private:
+ protocol::client_operation m_op{};
+ int64_t m_req_id{};
+};
+
+/**
+ * Message which server sends to client.
+ */
+class server_message : public structured_message {
+public:
+ server_message() = default;
+
+ server_message(const server_message&) = default;
+
+ explicit server_message(std::vector<std::byte> data)
+ : structured_message(std::move(data))
+ {
+ protocol::reader rd{{m_data.data(), m_data.size()}};
+
+ m_req_id = rd.read_int64();
+ m_flags = rd.read_int32();
+ m_obs_ts = rd.read_int64();
+
+ m_payload_pos = rd.position();
+
+ assert(m_payload_pos <= m_data.size());
+ }
+
+ [[nodiscard]] int64_t get_req_id() const { return m_req_id; }
+
+ [[nodiscard]] int32_t get_flags() const { return m_flags; }
+
+ [[nodiscard]] int64_t get_obs_ts() const { return m_obs_ts; }
+
+private:
+ int64_t m_req_id{};
+ int32_t m_flags{};
+ int64_t m_obs_ts{};
+};
+
+/**
+ * Intercepts messages which go through @c asio_proxy.
+ */
+class message_listener {
+public:
+ void register_message(raw_message msg) {
+ if (enable_message_registration.load()) {
+ std::unique_lock lock(m_mutex);
+ m_queue.push(std::move(msg));
+ }
+ }
+
+ [[nodiscard]] std::queue<raw_message> get_msg_queue() const {
+ std::shared_lock lock(m_mutex);
+ return m_queue;
+ }
+
+ template<typename MESSAGE_TYPE>
+ std::vector<MESSAGE_TYPE> get_next() {
+ std::unique_lock lock(m_mutex);
+
+ std::vector<MESSAGE_TYPE> res;
+
+ while (!m_queue.empty() && res.empty()) {
+ auto& chunk = m_queue.front();
+
+ network::data_buffer_ref buf{{chunk.data(), chunk.size()}};
+
+ while (true) {
+ auto out = codec.decode(buf);
+
+ if (out.empty()) {
+ break;
+ }
+
+ auto out_bv = out.get_bytes_view();
+ std::vector<std::byte> data{out_bv.begin(), out_bv.end()};
+ res.emplace_back(std::move(data));
+ }
+ m_queue.pop();
+ }
+
+ return res;
+ }
+
+ /**
+ * Enable/disables message registration for message listeners;
+ * @param enable @c True if registration enabled, otherwise disabled.
Review Comment:
```suggestion
* Enable/disable message registration for message listeners;
* @param enable @c True if registration enabled, otherwise disabled.
```
##########
modules/platforms/cpp/ignite/client/detail/table/schema.h:
##########
@@ -118,30 +125,52 @@ struct schema {
* @return A new schema instance.
*/
static std::shared_ptr<schema> create_instance(std::int32_t version, std::vector<column> &&cols) {
- std::int32_t key_columns_cnt = 0;
+ size_t key_columns_cnt = 0;
+ size_t collocated_columns_cnt = 0;
Review Comment:
Minor: Here and in other places:
```suggestion
std::size_t key_columns_cnt = 0;
std::size_t collocated_columns_cnt = 0;
```
##########
modules/platforms/cpp/ignite/protocol/messages.h:
##########
@@ -66,10 +69,16 @@ struct handshake_response {
protocol_context context{};
/** Observable timestamp. */
- std::int64_t observable_timestamp;
+ std::int64_t observable_timestamp{};
/** Server idle timeout in ms. */
- std::int64_t idle_timeout_ms;
+ std::int64_t idle_timeout_ms{};
Review Comment:
Let's init it to explicit value if you do anyway. Reads better, requires less implicit knowledge.
```suggestion
std::int64_t observable_timestamp{0};
/** Server idle timeout in ms. */
std::int64_t idle_timeout_ms{0};
```
##########
modules/platforms/cpp/ignite/protocol/messages.cpp:
##########
@@ -100,4 +100,39 @@ handshake_response parse_handshake_response(bytes_view message) {
return res;
}
+void write_partition_assignment_request(writer &writer, std::int32_t table_id, std::int64_t timestamp) {
+ writer.write(table_id);
+ writer.write(timestamp);
+}
+
+std::shared_ptr<partition_assignment> read_partition_assignment_response(reader &reader, std::int64_t timestamp) {
+ auto cnt = reader.read_int32();
+ if (cnt <= 0)
+ throw ignite_error("Invalid partition count: " + std::to_string(cnt));
Review Comment:
Are we sure partition count can never be zero in response? What about the case when the next field `assignment_available` is `false`?
##########
modules/platforms/cpp/ignite/common/uuid_test.cpp:
##########
@@ -64,3 +64,13 @@ TEST(uuid, stream) {
EXPECT_EQ(uuidString, uuidString2);
}
+
+TEST(uuid, from_string) {
+ std::string uuidText = "4b62e46a-d380-460f-94ea-9b4320752634";
+
+ auto opt = ignite::uuid::from_string(uuidText);
+
+ EXPECT_TRUE(opt.has_value());
+ EXPECT_EQ(5432155248028304911LL, opt->get_most_significant_bits());
+ EXPECT_EQ(-7716184298936261068LL, opt->get_least_significant_bits());
+}
Review Comment:
Let's add more tests. This is a public interface change, this function will be exposed to a user, we have to test it properly. We should definitely check that conversation from string and back to string produces the same result at least.
##########
modules/platforms/cpp/ignite/common/uuid.h:
##########
@@ -46,6 +50,64 @@ class uuid {
: most(most)
, least(least) {}
+ /**
+ * Parses string-encoded uuid.
+ * Example of possible input value: ff979603-fb56-49e9-bc79-7c4487bbbafd.
+ * @param text String-encoded uuid.
+ * @return @c uuid object if parsing was successful, otherwise empty optional.
+ */
+ static std::optional<uuid> from_string(const std::string& text) {
+ if (text.length() != 36) {
+ return {};
+ }
+
+ auto str = text.c_str();
+
+ auto parse_chunk = [str](size_t beg, size_t end, uint64_t& out) -> bool {
+ char* p;
+
+ if (errno != 0) {
+ errno = 0;
+ }
+
+ out = std::strtoull(str + beg, &p, 16);
+
+ if (p != str + end || (*p != '-' && *p != '\0') || errno == ERANGE) {
+ return false;
+ }
+
+ return true;
+ };
+
+ uint64_t msb1, msb2, msb3;
+ uint64_t lsb1, lsb2;
Review Comment:
Minor: here and in other places:
```suggestion
std::uint64_t msb1, msb2, msb3;
std::uint64_t lsb1, lsb2;
```
##########
modules/platforms/cpp/ignite/protocol/partition_assignment.h:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <optional>
+#include <string>
+#include <vector>
+#include <cstdint>
+
+namespace ignite::protocol {
+
+/**
+ * Partition assignment.
+ */
+class partition_assignment {
+public:
+ partition_assignment() = default;
+ partition_assignment(const partition_assignment&) = default;
+ partition_assignment(partition_assignment&&) noexcept = default;
+ partition_assignment &operator=(const partition_assignment &other) = default;
+ partition_assignment &operator=(partition_assignment &&other) noexcept = default;
+
+ partition_assignment(std::int64_t timestamp, std::vector<std::optional<std::string>> partitions)
+ : timestamp(timestamp)
+ , partitions(std::move(partitions)) {}
+
+ /**
+ * Check whether the assignment is outdated.
+ *
+ * @param actual_timestamp Timestamp.
+ * @return @c true if assignment is outdated.
+ */
+ [[nodiscard]] bool is_outdated(std::int64_t actual_timestamp) const { return timestamp < actual_timestamp; }
+
+ [[nodiscard]] const std::vector<std::optional<std::string>>& get_partitions() const {
+ return partitions;
+ }
+private:
+ /** Assignment timestamp. */
+ std::int64_t timestamp{0};
+
+ /** Partitions. */
+ std::vector<std::optional<std::string>> partitions;
Review Comment:
It's a class, so fields should start with `m_` prefix. Also, it's not clear what `partitions` here - is it a mapping [index (=partition) -> node name]? Let's specify it explicitly.
```suggestion
/** Assignment timestamp. */
std::int64_t m_timestamp{0};
/** Partitions. */
std::vector<std::optional<std::string>> m_partitions;
```
##########
modules/platforms/cpp/ignite/protocol/partition_assignment.h:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <optional>
+#include <string>
+#include <vector>
+#include <cstdint>
+
+namespace ignite::protocol {
+
+/**
+ * Partition assignment.
+ */
+class partition_assignment {
+public:
+ partition_assignment() = default;
+ partition_assignment(const partition_assignment&) = default;
+ partition_assignment(partition_assignment&&) noexcept = default;
+ partition_assignment &operator=(const partition_assignment &other) = default;
+ partition_assignment &operator=(partition_assignment &&other) noexcept = default;
+
+ partition_assignment(std::int64_t timestamp, std::vector<std::optional<std::string>> partitions)
+ : timestamp(timestamp)
+ , partitions(std::move(partitions)) {}
+
+ /**
+ * Check whether the assignment is outdated.
+ *
+ * @param actual_timestamp Timestamp.
+ * @return @c true if assignment is outdated.
+ */
+ [[nodiscard]] bool is_outdated(std::int64_t actual_timestamp) const { return timestamp < actual_timestamp; }
+
+ [[nodiscard]] const std::vector<std::optional<std::string>>& get_partitions() const {
Review Comment:
Documentation is missing.
##########
modules/platforms/cpp/ignite/protocol/partition_assignment.h:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <optional>
+#include <string>
+#include <vector>
+#include <cstdint>
+
+namespace ignite::protocol {
+
+/**
+ * Partition assignment.
+ */
+class partition_assignment {
+public:
+ partition_assignment() = default;
+ partition_assignment(const partition_assignment&) = default;
+ partition_assignment(partition_assignment&&) noexcept = default;
+ partition_assignment &operator=(const partition_assignment &other) = default;
+ partition_assignment &operator=(partition_assignment &&other) noexcept = default;
+
+ partition_assignment(std::int64_t timestamp, std::vector<std::optional<std::string>> partitions)
Review Comment:
Documentation is missing.
##########
modules/platforms/cpp/ignite/protocol/partition_assignment.h:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <optional>
+#include <string>
+#include <vector>
+#include <cstdint>
+
+namespace ignite::protocol {
+
+/**
+ * Partition assignment.
+ */
+class partition_assignment {
+public:
+ partition_assignment() = default;
+ partition_assignment(const partition_assignment&) = default;
+ partition_assignment(partition_assignment&&) noexcept = default;
+ partition_assignment &operator=(const partition_assignment &other) = default;
+ partition_assignment &operator=(partition_assignment &&other) noexcept = default;
Review Comment:
Only default constructor is needed (see [core guidelines](https://isocpp.github.io/CppCoreGuidelines/CppCoreGuidelines#rc-zero))
```suggestion
partition_assignment() = default;
```
##########
modules/platforms/cpp/ignite/client/detail/cluster_connection.h:
##########
@@ -289,6 +319,22 @@ class cluster_connection : public std::enable_shared_from_this<cluster_connectio
*/
std::shared_ptr<node_connection> get_random_connected_channel();
+ /**
+ * Get a preferred node connection if partition mapping is provided
+ * otherwise returns random node connection.
+ * @param preferred_node_name Name of preferred node.
+ * @return Node connection.
Review Comment:
Here and in other places: spacing, and don't wrap too short (Allowed line length is 120)
```suggestion
/**
* Get a preferred node connection if partition mapping is provided otherwise returns random node connection.
*
* @param preferred_node_name Name of preferred node.
* @return Node connection.
```
##########
modules/platforms/cpp/tests/test-common/proxy/message_listener.h:
##########
@@ -0,0 +1,178 @@
+// 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.
+//
Review Comment:
Here and in other new files: our standard header is different
##########
modules/platforms/cpp/ignite/common/uuid.h:
##########
@@ -46,6 +50,64 @@ class uuid {
: most(most)
, least(least) {}
+ /**
+ * Parses string-encoded uuid.
+ * Example of possible input value: ff979603-fb56-49e9-bc79-7c4487bbbafd.
+ * @param text String-encoded uuid.
+ * @return @c uuid object if parsing was successful, otherwise empty optional.
+ */
+ static std::optional<uuid> from_string(const std::string& text) {
Review Comment:
Function definition should go to cpp - it's a public header, we're hiding the implementation.
##########
modules/platforms/cpp/ignite/client/detail/cluster_connection.h:
##########
@@ -289,6 +319,22 @@ class cluster_connection : public std::enable_shared_from_this<cluster_connectio
*/
std::shared_ptr<node_connection> get_random_connected_channel();
+ /**
+ * Get a preferred node connection if partition mapping is provided
+ * otherwise returns random node connection.
+ * @param preferred_node_name Name of preferred node.
+ * @return Node connection.
+ */
+ std::shared_ptr<node_connection> get_preferred_channel(const std::string& preferred_node_name);
+
+ /**
+ * Get connection according to provided partition mapping
+ * otherwise returns random node connection.
+ * @param pm Partition mapping.
+ * @return Node connection.
+ */
+ std::shared_ptr<node_connection> get_connected_channel(const std::optional<std::string> &preferred_node_name);
Review Comment:
This is a bit confusing. Let's just leave `get_preferred_channel`, which acts as `get_connected_channel`. You don't use `get_preferred_channel` on itself now anywhere anyway.
##########
modules/platforms/cpp/ignite/protocol/messages.cpp:
##########
@@ -100,4 +100,39 @@ handshake_response parse_handshake_response(bytes_view message) {
return res;
}
+void write_partition_assignment_request(writer &writer, std::int32_t table_id, std::int64_t timestamp) {
+ writer.write(table_id);
+ writer.write(timestamp);
+}
+
+std::shared_ptr<partition_assignment> read_partition_assignment_response(reader &reader, std::int64_t timestamp) {
+ auto cnt = reader.read_int32();
+ if (cnt <= 0)
+ throw ignite_error("Invalid partition count: " + std::to_string(cnt));
+
+ std::vector<std::optional<std::string>> partitions;
+ partitions.reserve(cnt);
+
+ bool assignment_available = reader.read_bool();
+ if (!assignment_available) {
+ // Invalidate the current assignment so that we can retry on the next call.
+ // Return an empty array so that per-partition batches can be initialized.
+ // We'll get the actual assignment on the next call.
+ partitions.insert(partitions.end(), cnt, std::nullopt);
+ return std::make_shared<partition_assignment>(0, std::move(partitions));
+ }
+
+ // Returned timestamp can be newer than requested.
+ std::int64_t ts = reader.read_int64();
+ if (ts < timestamp)
+ throw ignite_error("Returned timestamp is older than requested: " + std::to_string(ts) + " < "
+ + std::to_string(timestamp));
Review Comment:
Should it really be an exception though? It seems like a normal situation that can happen, not some kind of error.
##########
modules/platforms/cpp/ignite/client/detail/node_connection.h:
##########
@@ -232,7 +232,14 @@ class node_connection : public std::enable_shared_from_this<node_connection> {
*/
std::shared_ptr<ignite_logger> get_logger() const { return m_logger; }
- void handle_timeouts();
+ /**
+ * Cancels waiting for over-due responses.
+ */
+ void handle_timeouts();
+
+ const std::string& get_node_name() const {
Review Comment:
The documentation missing.
##########
modules/platforms/cpp/ignite/common/uuid.h:
##########
@@ -46,6 +50,64 @@ class uuid {
: most(most)
, least(least) {}
+ /**
+ * Parses string-encoded uuid.
+ * Example of possible input value: ff979603-fb56-49e9-bc79-7c4487bbbafd.
Review Comment:
```suggestion
* Parse string-encoded uuid.
*
* Example of possible input value: ff979603-fb56-49e9-bc79-7c4487bbbafd.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "Erixonich (via GitHub)" <gi...@apache.org>.
Erixonich commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3079050824
##########
modules/platforms/cpp/tests/test-common/proxy/message_listener.h:
##########
@@ -0,0 +1,178 @@
+// 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.
+//
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "Erixonich (via GitHub)" <gi...@apache.org>.
Erixonich commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3079101614
##########
modules/platforms/cpp/ignite/protocol/partition_assignment.h:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <optional>
+#include <string>
+#include <vector>
+#include <cstdint>
+
+namespace ignite::protocol {
+
+/**
+ * Partition assignment.
+ */
+class partition_assignment {
+public:
+ partition_assignment() = default;
+ partition_assignment(const partition_assignment&) = default;
+ partition_assignment(partition_assignment&&) noexcept = default;
+ partition_assignment &operator=(const partition_assignment &other) = default;
+ partition_assignment &operator=(partition_assignment &&other) noexcept = default;
+
+ partition_assignment(std::int64_t timestamp, std::vector<std::optional<std::string>> partitions)
Review Comment:
done
##########
modules/platforms/cpp/ignite/protocol/partition_assignment.h:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <optional>
+#include <string>
+#include <vector>
+#include <cstdint>
+
+namespace ignite::protocol {
+
+/**
+ * Partition assignment.
+ */
+class partition_assignment {
+public:
+ partition_assignment() = default;
+ partition_assignment(const partition_assignment&) = default;
+ partition_assignment(partition_assignment&&) noexcept = default;
+ partition_assignment &operator=(const partition_assignment &other) = default;
+ partition_assignment &operator=(partition_assignment &&other) noexcept = default;
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "isapego (via GitHub)" <gi...@apache.org>.
isapego commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3080264807
##########
modules/platforms/cpp/ignite/common/uuid_test.cpp:
##########
@@ -64,3 +64,54 @@ TEST(uuid, stream) {
EXPECT_EQ(uuidString, uuidString2);
}
+
+TEST(uuid, incorrect_string_representation) {
+ auto uuidText = "foo";
+
+ auto resOpt = ignite::uuid::from_string(uuidText);
+
+ EXPECT_FALSE(resOpt.has_value());
+}
+
+class uuid_string_presentation_fixture: public ::testing::TestWithParam<std::tuple<std::string, ignite::uuid>> {};
+
+TEST_P(uuid_string_presentation_fixture, from_string) {
+ auto [uuidText, uuid] = GetParam();
+
+ auto resOpt = ignite::uuid::from_string(uuidText);
+
+ EXPECT_TRUE(resOpt.has_value());
+
+ EXPECT_EQ(uuid.get_most_significant_bits(), resOpt->get_most_significant_bits());
+ EXPECT_EQ(uuid.get_least_significant_bits(), resOpt->get_least_significant_bits());
+}
+
+// We check that two representation are consistent both ways.
+TEST_P(uuid_string_presentation_fixture, to_string) {
+ auto [uuidText, uuid] = GetParam();
+
+
+ std::stringstream ss;
+ ss << uuid;
+
+ EXPECT_EQ(uuidText, ss.str());
+}
+
+// Values has taken randomly from java program, as this implementation should have same behaviour as java.lang.UUID for compatibility reasons.
+INSTANTIATE_TEST_SUITE_P(
+ various_uuids,
+ uuid_string_presentation_fixture,
+ ::testing::Values(
+ std::tuple{"4b62e46a-d380-460f-94ea-9b4320752634", ignite::uuid{5432155248028304911LL, -7716184298936261068LL}},
+ std::tuple{"cabf7114-7763-4ad5-b312-5ffadcbf48c8", ignite::uuid{-3837224024780092715, -5543262660289673016}},
+ std::tuple{"8ee62c79-5636-44b9-adc4-c9b4272a2ba6", ignite::uuid{-8149777576031271751, -5925389434124358746}},
+ std::tuple{"3e69a946-fc2a-4da1-9938-d1f082dd36ce", ignite::uuid{4497311825249586593, -7405938756292888882}},
+ std::tuple{"7cc9f130-821b-4580-bfda-955072540643", ignite::uuid{8991983321665455488, -4622217894794361277}},
+ std::tuple{"1f86b0b1-dfab-4108-8b7e-8b3e4177ceb0", ignite::uuid{2271697340063236360, -8395119555869421904}},
+ std::tuple{"88cc69d9-d191-482d-b295-aa458cf99167", ignite::uuid{-8589374005057599443, -5578365347733859993}},
+ std::tuple{"a8b6298c-126e-4c4d-a521-de3989a31492", ignite::uuid{-6289794147994940339, -6547708044516322158}},
+ std::tuple{"03b210a8-5a6a-4cae-b4fc-f30df9f1eebd", ignite::uuid{266293643225746606, -5405178211397931331}},
+ std::tuple{"f145e277-bb58-4f89-adfa-3166a975b71b", ignite::uuid{-1061193133303771255, -5910357243970865381}},
+ std::tuple{"dc8b5857-067f-486e-aa5b-52cbdb26dd9c", ignite::uuid{-2554851232808220562, -6171247828872536676}}
Review Comment:
Why do we pass uuid values here? Let's get them using `from_string`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "isapego (via GitHub)" <gi...@apache.org>.
isapego commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3078159084
##########
modules/platforms/cpp/ignite/client/detail/table/table_impl.cpp:
##########
@@ -472,4 +584,68 @@ std::shared_ptr<table_impl> table_impl::from_facade(table &tb) {
return tb.m_impl;
}
+void table_impl::update_partition_assignment() {
+ ignite_callback<std::shared_ptr<protocol::partition_assignment>> callback = [self=shared_from_this()](auto pa) {
+ if (pa.has_error()) {
+ self->m_connection->get_logger()->log_error("Error while updating partition assignment for table"
+ + self->get_name() + "error " + pa.error().what_str());
+
+ return;
+ }
+
+ std::lock_guard lock(self->m_partitions_mutex);
+ self->m_partition_assignment = std::move(pa).value();
+ };
+
+ load_partition_assignment_async(std::move(callback));
+}
+
+void table_impl::load_partition_assignment_async(ignite_callback<std::shared_ptr<protocol::partition_assignment>> callback) {
+ std::int64_t timestamp = m_connection->get_assignment_timestamp();
+
+ auto pa = get_partition_assignment();
+ if (pa && !pa->is_outdated(timestamp)) {
+ m_connection->get_logger()->log_debug("Partition assignment for table " + get_name() + " is up to date.");
+
+ callback(std::move(pa));
+ return;
+ }
+
+ auto writer_func = [id = m_id, timestamp](protocol::writer &writer, auto&) {
+ protocol::write_partition_assignment_request(writer, id, timestamp);
+ };
+
+ auto reader_func = [timestamp](protocol::reader &reader) -> std::shared_ptr<protocol::partition_assignment> {
+ return protocol::read_partition_assignment_response(reader, timestamp);
+ };
+
+ m_connection->perform_request<std::shared_ptr<protocol::partition_assignment>>(
+ protocol::client_operation::PARTITION_ASSIGNMENT_GET,
+ nullptr,
+ writer_func,
+ std::move(reader_func),
+ std::move(callback));
+}
+
+std::optional<std::string> table_impl::get_preferred_node_name(const ignite_tuple &key_or_rec, const schema &sch) {
+ auto pa = get_partition_assignment();
+
+ if (!pa || pa->get_partitions().empty()) {
+ m_connection->get_logger()->log_debug("No partition distribution available, random node will called");
+ return {};
+ }
+
+ hash_calculator hc;
+ for (auto column : sch.collocated_columns) {
+ auto val = key_or_rec.get(column->key_index);
Review Comment:
Agreed, the collocated columns order is what matters here. But I'm kind of impressed by copilot here anyway :D
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "Erixonich (via GitHub)" <gi...@apache.org>.
Erixonich commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3079319881
##########
modules/platforms/cpp/ignite/common/uuid.h:
##########
@@ -46,6 +50,64 @@ class uuid {
: most(most)
, least(least) {}
+ /**
+ * Parses string-encoded uuid.
+ * Example of possible input value: ff979603-fb56-49e9-bc79-7c4487bbbafd.
+ * @param text String-encoded uuid.
+ * @return @c uuid object if parsing was successful, otherwise empty optional.
+ */
+ static std::optional<uuid> from_string(const std::string& text) {
+ if (text.length() != 36) {
+ return {};
+ }
+
+ auto str = text.c_str();
+
+ auto parse_chunk = [str](size_t beg, size_t end, uint64_t& out) -> bool {
+ char* p;
+
+ if (errno != 0) {
+ errno = 0;
+ }
+
+ out = std::strtoull(str + beg, &p, 16);
+
+ if (p != str + end || (*p != '-' && *p != '\0') || errno == ERANGE) {
+ return false;
+ }
+
+ return true;
+ };
+
+ uint64_t msb1, msb2, msb3;
+ uint64_t lsb1, lsb2;
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22141 C++ 3.0: Implement Partition Awareness for C++ client [ignite-3]
Posted by "Erixonich (via GitHub)" <gi...@apache.org>.
Erixonich commented on code in PR #7966:
URL: https://github.com/apache/ignite-3/pull/7966#discussion_r3079316863
##########
modules/platforms/cpp/ignite/common/uuid.h:
##########
@@ -46,6 +50,64 @@ class uuid {
: most(most)
, least(least) {}
+ /**
+ * Parses string-encoded uuid.
+ * Example of possible input value: ff979603-fb56-49e9-bc79-7c4487bbbafd.
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org