You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by is...@apache.org on 2023/03/20 15:50:57 UTC
[ignite-3] branch ignite-17607 created (now 2c8373c1ed)
This is an automated email from the ASF dual-hosted git repository.
isapego pushed a change to branch ignite-17607
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
at 2c8373c1ed IGNITE-17607 Add test
This branch includes the following new commits:
new 9212e30294 IGNITE-17607 Compute stub
new 72c529be8e IGNITE-17607 execute_async stub
new 5a95982f7c IGNITE-17607 execute_async implementation
new f9188fa9be IGNITE-17607 Fix get cluster node
new f0aaea1e29 IGNITE-17607 get_cluster_nodes implemented
new d84db92eba IGNITE-17607 Add execute test, fix bug
new 64972d2429 IGNITE-17607 Add test
new 27c15ab6aa IGNITE-17607 Implement broadcast
new f140d16b30 IGNITE-17607 Add test
new 3d50b796c3 IGNITE-17607 Add nullptr to primitive
new 9432ffe877 IGNITE-17607 Add args test
new bd294cb770 IGNITE-17607 Add negative tests
new f36c368210 IGNITE-17607 Add all arg types test
new a661c39763 IGNITE-17607 Re-factor argument checks
new 0cea5d2779 IGNITE-17607 Execute colocated stub
new aca718633e IGNITE-17607 Execute colocated, untested
new 00dc41547f IGNITE-17607 Add test
new 32cb4532b1 IGNITE-17607 Add test
new 2c8373c1ed IGNITE-17607 Add test
The 19 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[ignite-3] 07/19: IGNITE-17607 Add test
Posted by is...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
isapego pushed a commit to branch ignite-17607
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 64972d2429176e610ca407f5a6cf31c06b22dd54
Author: Igor Sapego <is...@apache.org>
AuthorDate: Tue Mar 7 16:38:18 2023 +0300
IGNITE-17607 Add test
---
.../cpp/tests/client-test/compute_test.cpp | 24 ++++++++++++++++++++++
1 file changed, 24 insertions(+)
diff --git a/modules/platforms/cpp/tests/client-test/compute_test.cpp b/modules/platforms/cpp/tests/client-test/compute_test.cpp
index a90b4eac69..b09fd1b11f 100644
--- a/modules/platforms/cpp/tests/client-test/compute_test.cpp
+++ b/modules/platforms/cpp/tests/client-test/compute_test.cpp
@@ -43,6 +43,19 @@ protected:
// remove all
}
+ /**
+ * Get specific node.
+ * @param id Node id.
+ * @return Node.
+ */
+ cluster_node get_node(size_t id) {
+ auto nodes = m_client.get_cluster_nodes();
+ std::sort(nodes.begin(), nodes.end(), [] (const auto &n1, const auto &n2) {
+ return n1.get_name() < n2.get_name();
+ });
+ return nodes[id];
+ }
+
/** Ignite client. */
ignite_client m_client;
};
@@ -76,3 +89,14 @@ TEST_F(compute_test, execute_on_random_node) {
ASSERT_TRUE(result.has_value());
EXPECT_THAT(result.value().get<std::string>(), ::testing::StartsWith(PLATFORM_TEST_NODE_RUNNER));
}
+
+TEST_F(compute_test, execute_on_specific_node) {
+ auto res1 = m_client.get_compute().execute({get_node(0)}, NODE_NAME_JOB, {"-", 11});
+ auto res2 = m_client.get_compute().execute({get_node(1)}, NODE_NAME_JOB, {":", 22});
+
+ ASSERT_TRUE(res1.has_value());
+ ASSERT_TRUE(res2.has_value());
+
+ EXPECT_EQ(res1.value().get<std::string>(), PLATFORM_TEST_NODE_RUNNER + "-_11");
+ EXPECT_EQ(res2.value().get<std::string>(), PLATFORM_TEST_NODE_RUNNER + "_2:_22");
+}
[ignite-3] 01/19: IGNITE-17607 Compute stub
Posted by is...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
isapego pushed a commit to branch ignite-17607
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 9212e30294b90792fff56bed918715ccf3b4331b
Author: Igor Sapego <is...@apache.org>
AuthorDate: Tue Mar 7 03:00:44 2023 +0300
IGNITE-17607 Compute stub
---
modules/platforms/cpp/CMakeLists.txt | 2 +-
modules/platforms/cpp/cmake/ignite_test.cmake | 2 +-
modules/platforms/cpp/ignite/client/CMakeLists.txt | 3 +
.../cpp/ignite/client/compute/compute.cpp | 28 +++++++
.../platforms/cpp/ignite/client/compute/compute.h | 83 +++++++++++++++++++++
.../cpp/ignite/client/network/cluster_node.h | 87 ++++++++++++++++++++++
modules/platforms/cpp/ignite/common/CMakeLists.txt | 1 +
.../cpp/ignite/{network => common}/end_point.h | 2 +-
.../platforms/cpp/ignite/network/async_handler.h | 2 +-
.../network/detail/linux/connecting_context.h | 7 +-
.../network/detail/linux/linux_async_client.h | 8 +-
.../detail/linux/linux_async_worker_thread.h | 11 ++-
.../ignite/network/detail/win/win_async_client.h | 8 +-
13 files changed, 222 insertions(+), 22 deletions(-)
diff --git a/modules/platforms/cpp/CMakeLists.txt b/modules/platforms/cpp/CMakeLists.txt
index ca7bc3c401..a12077763f 100644
--- a/modules/platforms/cpp/CMakeLists.txt
+++ b/modules/platforms/cpp/CMakeLists.txt
@@ -15,7 +15,7 @@
# limitations under the License.
#
-cmake_minimum_required(VERSION 3.10)
+cmake_minimum_required(VERSION 3.18)
project(Ignite.C++ VERSION 3 LANGUAGES CXX)
set(CMAKE_CXX_STANDARD 17)
diff --git a/modules/platforms/cpp/cmake/ignite_test.cmake b/modules/platforms/cpp/cmake/ignite_test.cmake
index d80e604c7f..108fcf704e 100644
--- a/modules/platforms/cpp/cmake/ignite_test.cmake
+++ b/modules/platforms/cpp/cmake/ignite_test.cmake
@@ -38,5 +38,5 @@ function(ignite_test TEST_NAME TEST_SOURCE)
target_link_libraries(${TEST_NAME} ${IGNITE_TEST_LIBS} GTest::GTest GTest::Main)
endif()
- gtest_discover_tests(${TEST_NAME})
+ gtest_discover_tests(${TEST_NAME} XML_OUTPUT_DIR ${CMAKE_BINARY_DIR}/Testing/Result)
endfunction()
diff --git a/modules/platforms/cpp/ignite/client/CMakeLists.txt b/modules/platforms/cpp/ignite/client/CMakeLists.txt
index 53bf6486d8..e84687dce8 100644
--- a/modules/platforms/cpp/ignite/client/CMakeLists.txt
+++ b/modules/platforms/cpp/ignite/client/CMakeLists.txt
@@ -21,6 +21,7 @@ set(TARGET ${PROJECT_NAME})
set(SOURCES
ignite_client.cpp
+ compute/compute.cpp
sql/sql.cpp
sql/result_set.cpp
table/key_value_view.cpp
@@ -41,6 +42,8 @@ set(PUBLIC_HEADERS
ignite_client.h
ignite_client_configuration.h
ignite_logger.h
+ compute/compute.h
+ network/cluster_node.h
sql/sql.h
table/ignite_tuple.h
table/key_value_view.h
diff --git a/modules/platforms/cpp/ignite/client/compute/compute.cpp b/modules/platforms/cpp/ignite/client/compute/compute.cpp
new file mode 100644
index 0000000000..8476afbaab
--- /dev/null
+++ b/modules/platforms/cpp/ignite/client/compute/compute.cpp
@@ -0,0 +1,28 @@
+/*
+ * 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 "ignite/client/compute/compute.h"
+//#include "ignite/client/detail/compute/compute_impl.h"
+
+namespace ignite {
+
+void compute::execute_async(std::vector<cluster_node> nodes, std::string_view job_class_name,
+ std::vector<primitive> args, ignite_callback<std::optional<primitive>> callback) {
+ // TODO: Implement me
+}
+
+} // namespace ignite
diff --git a/modules/platforms/cpp/ignite/client/compute/compute.h b/modules/platforms/cpp/ignite/client/compute/compute.h
new file mode 100644
index 0000000000..8f142bb835
--- /dev/null
+++ b/modules/platforms/cpp/ignite/client/compute/compute.h
@@ -0,0 +1,83 @@
+/*
+ * 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/client/network/cluster_node.h"
+#include "ignite/client/primitive.h"
+#include "ignite/client/transaction/transaction.h"
+#include "ignite/common/config.h"
+#include "ignite/common/ignite_result.h"
+
+#include <memory>
+#include <utility>
+
+namespace ignite {
+
+namespace detail {
+class compute_impl;
+}
+
+/**
+ * Ignite Compute facade.
+ */
+class compute {
+ friend class ignite_client;
+
+public:
+ // Delete
+ compute() = delete;
+
+ /**
+ * Executes single SQL statement asynchronously and returns rows.
+ *
+ * @param tx Optional transaction. If nullptr implicit transaction for this single operation is used.
+ * @param statement Statement to execute.
+ * @param args Arguments for the statement.
+ * @param callback A callback called on operation completion with SQL result set.
+ */
+ IGNITE_API void execute_async(std::vector<cluster_node> nodes, std::string_view job_class_name, std::vector<primitive> args,
+ ignite_callback<std::optional<primitive>> callback);
+
+ /**
+ * Executes single SQL statement and returns rows.
+ *
+ * @param tx Optional transaction. If nullptr implicit transaction for this single operation is used.
+ * @param statement Statement to execute.
+ * @param args Arguments for the statement.
+ * @return SQL result set.
+ */
+ IGNITE_API std::optional<primitive> execute(std::vector<cluster_node> nodes, std::string_view job_class_name, std::vector<primitive> args) {
+ return sync<std::optional<primitive>>([this, nodes = std::move(nodes), job_class_name, args = std::move(args)](auto callback) mutable {
+ execute_async(std::move(nodes), job_class_name, std::move(args), std::move(callback));
+ });
+ }
+
+private:
+ /**
+ * Constructor
+ *
+ * @param impl Implementation
+ */
+ explicit compute(std::shared_ptr<detail::compute_impl> impl)
+ : m_impl(std::move(impl)) {}
+
+ /** Implementation. */
+ std::shared_ptr<detail::compute_impl> m_impl;
+};
+
+} // namespace ignite
diff --git a/modules/platforms/cpp/ignite/client/network/cluster_node.h b/modules/platforms/cpp/ignite/client/network/cluster_node.h
new file mode 100644
index 0000000000..1392b17d6d
--- /dev/null
+++ b/modules/platforms/cpp/ignite/client/network/cluster_node.h
@@ -0,0 +1,87 @@
+/*
+ * 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/common/end_point.h"
+
+#include <cstdint>
+#include <type_traits>
+#include <variant>
+#include <vector>
+
+namespace ignite {
+
+/**
+ * Ignite cluster node.
+ */
+class cluster_node {
+public:
+ // Default
+ cluster_node() = default;
+
+ /**
+ * Constructor.
+ *
+ * @param id Local ID.
+ * @param name Name.
+ * @param address Address.
+ */
+ cluster_node(std::string id, std::string name, network::end_point address)
+ : m_id(std::move(id))
+ , m_name(std::move(name))
+ , m_address(std::move(address)) {}
+
+ /**
+ * Gets the local node id. Changes after node restart.
+ *
+ * @return Local node id.
+ */
+ [[nodiscard]] const std::string& get_id() const {
+ return m_id;
+ }
+
+ /**
+ * Gets the unique name of the cluster member. Does not change after node restart.
+ *
+ * @return Unique name of the cluster member.
+ */
+ [[nodiscard]] const std::string& get_name() const {
+ return m_name;
+ }
+
+ /**
+ * Gets the node address.
+ *
+ * @return Node address.
+ */
+ [[nodiscard]] const network::end_point& get_address() const {
+ return m_address;
+ }
+
+private:
+ /** Local ID. */
+ std::string m_id{};
+
+ /** Name. */
+ std::string m_name{};
+
+ /** Address. */
+ network::end_point m_address{};
+};
+
+} // namespace ignite
diff --git a/modules/platforms/cpp/ignite/common/CMakeLists.txt b/modules/platforms/cpp/ignite/common/CMakeLists.txt
index 0f2af03db0..1c5a786c28 100644
--- a/modules/platforms/cpp/ignite/common/CMakeLists.txt
+++ b/modules/platforms/cpp/ignite/common/CMakeLists.txt
@@ -27,6 +27,7 @@ set(PUBLIC_HEADERS
bytes.h
bytes_view.h
config.h
+ end_point.h
ignite_date.h
ignite_date_time.h
ignite_error.h
diff --git a/modules/platforms/cpp/ignite/network/end_point.h b/modules/platforms/cpp/ignite/common/end_point.h
similarity index 99%
rename from modules/platforms/cpp/ignite/network/end_point.h
rename to modules/platforms/cpp/ignite/common/end_point.h
index 22b0ce67b0..3e895987ab 100644
--- a/modules/platforms/cpp/ignite/network/end_point.h
+++ b/modules/platforms/cpp/ignite/common/end_point.h
@@ -67,7 +67,7 @@ struct end_point {
std::string host;
/** TCP port. */
- uint16_t port = 0;
+ uint16_t port{0};
};
/**
diff --git a/modules/platforms/cpp/ignite/network/async_handler.h b/modules/platforms/cpp/ignite/network/async_handler.h
index 06d5b52fac..9a64c22035 100644
--- a/modules/platforms/cpp/ignite/network/async_handler.h
+++ b/modules/platforms/cpp/ignite/network/async_handler.h
@@ -17,9 +17,9 @@
#pragma once
+#include "ignite/common/end_point.h"
#include <ignite/common/ignite_error.h>
#include <ignite/network/data_buffer.h>
-#include <ignite/network/end_point.h>
#include <cstdint>
#include <optional>
diff --git a/modules/platforms/cpp/ignite/network/detail/linux/connecting_context.h b/modules/platforms/cpp/ignite/network/detail/linux/connecting_context.h
index b8ff6f640d..a1a6e28023 100644
--- a/modules/platforms/cpp/ignite/network/detail/linux/connecting_context.h
+++ b/modules/platforms/cpp/ignite/network/detail/linux/connecting_context.h
@@ -17,10 +17,9 @@
#pragma once
-#include "linux_async_client.h"
-
-#include <ignite/network/end_point.h>
-#include <ignite/network/tcp_range.h>
+#include "ignite/network/tcp_range.h"
+#include "ignite/network/detail/linux/linux_async_client.h"
+#include "ignite/common/end_point.h"
#include <cstdint>
#include <memory>
diff --git a/modules/platforms/cpp/ignite/network/detail/linux/linux_async_client.h b/modules/platforms/cpp/ignite/network/detail/linux/linux_async_client.h
index 6f100766b1..1bb5b9351c 100644
--- a/modules/platforms/cpp/ignite/network/detail/linux/linux_async_client.h
+++ b/modules/platforms/cpp/ignite/network/detail/linux/linux_async_client.h
@@ -19,10 +19,10 @@
#include "sockets.h"
-#include <ignite/network/async_handler.h>
-#include <ignite/network/codec.h>
-#include <ignite/network/end_point.h>
-#include <ignite/network/tcp_range.h>
+#include "ignite/network/async_handler.h"
+#include "ignite/network/codec.h"
+#include "ignite/network/tcp_range.h"
+#include "ignite/common/end_point.h"
#include <cstdint>
#include <deque>
diff --git a/modules/platforms/cpp/ignite/network/detail/linux/linux_async_worker_thread.h b/modules/platforms/cpp/ignite/network/detail/linux/linux_async_worker_thread.h
index df441581d0..71a264a068 100644
--- a/modules/platforms/cpp/ignite/network/detail/linux/linux_async_worker_thread.h
+++ b/modules/platforms/cpp/ignite/network/detail/linux/linux_async_worker_thread.h
@@ -17,12 +17,11 @@
#pragma once
-#include "connecting_context.h"
-#include "linux_async_client.h"
-
-#include <ignite/network/async_handler.h>
-#include <ignite/network/end_point.h>
-#include <ignite/network/tcp_range.h>
+#include "ignite/network/detail/linux/connecting_context.h"
+#include "ignite/network/detail/linux/linux_async_client.h"
+#include "ignite/network/async_handler.h"
+#include "ignite/network/tcp_range.h"
+#include "ignite/common/end_point.h"
#include <cstdint>
#include <ctime>
diff --git a/modules/platforms/cpp/ignite/network/detail/win/win_async_client.h b/modules/platforms/cpp/ignite/network/detail/win/win_async_client.h
index 10e4de0599..741c2a97a9 100644
--- a/modules/platforms/cpp/ignite/network/detail/win/win_async_client.h
+++ b/modules/platforms/cpp/ignite/network/detail/win/win_async_client.h
@@ -19,10 +19,10 @@
#include "sockets.h"
-#include <ignite/network/async_handler.h>
-#include <ignite/network/codec.h>
-#include <ignite/network/end_point.h>
-#include <ignite/network/tcp_range.h>
+#include "ignite/network/async_handler.h"
+#include "ignite/network/codec.h"
+#include "ignite/network/tcp_range.h"
+#include "ignite/common/end_point.h"
#include <cstdint>
#include <deque>
[ignite-3] 04/19: IGNITE-17607 Fix get cluster node
Posted by is...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
isapego pushed a commit to branch ignite-17607
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit f9188fa9bef4858e67f3837905ba3f52440cd07c
Author: Igor Sapego <is...@apache.org>
AuthorDate: Tue Mar 7 15:13:30 2023 +0300
IGNITE-17607 Fix get cluster node
---
.../client/handler/requests/cluster/ClientClusterGetNodesRequest.java | 2 ++
.../main/java/org/apache/ignite/internal/client/TcpIgniteClient.java | 3 +++
.../platforms/dotnet/Apache.Ignite/Internal/IgniteClientInternal.cs | 4 ++++
3 files changed, 9 insertions(+)
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/cluster/ClientClusterGetNodesRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/cluster/ClientClusterGetNodesRequest.java
index 20032f4fa5..4d3d137083 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/cluster/ClientClusterGetNodesRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/cluster/ClientClusterGetNodesRequest.java
@@ -42,6 +42,8 @@ public class ClientClusterGetNodesRequest {
out.packArrayHeader(nodes.size());
for (ClusterNode node : nodes) {
+ out.packArrayHeader(4);
+
out.packString(node.id());
out.packString(node.name());
out.packString(node.address().host());
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/TcpIgniteClient.java b/modules/client/src/main/java/org/apache/ignite/internal/client/TcpIgniteClient.java
index 8ab6beaafa..e5b644b250 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/TcpIgniteClient.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/TcpIgniteClient.java
@@ -160,6 +160,9 @@ public class TcpIgniteClient implements IgniteClient {
List<ClusterNode> res = new ArrayList<>(cnt);
for (int i = 0; i < cnt; i++) {
+ int fieldCnt = r.in().unpackArrayHeader();
+ assert fieldCnt == 4;
+
res.add(new ClusterNode(
r.in().unpackString(),
r.in().unpackString(),
diff --git a/modules/platforms/dotnet/Apache.Ignite/Internal/IgniteClientInternal.cs b/modules/platforms/dotnet/Apache.Ignite/Internal/IgniteClientInternal.cs
index d71548cf61..fe00808716 100644
--- a/modules/platforms/dotnet/Apache.Ignite/Internal/IgniteClientInternal.cs
+++ b/modules/platforms/dotnet/Apache.Ignite/Internal/IgniteClientInternal.cs
@@ -18,6 +18,7 @@
namespace Apache.Ignite.Internal
{
using System.Collections.Generic;
+ using System.Diagnostics;
using System.Net;
using System.Threading.Tasks;
using Ignite.Compute;
@@ -85,6 +86,9 @@ namespace Apache.Ignite.Internal
for (var i = 0; i < count; i++)
{
+ var fieldCount = r.ReadArrayHeader();
+ Debug.Assert(fieldCount == 4, "fieldCount == 4");
+
res.Add(new ClusterNode(
Id: r.ReadString(),
Name: r.ReadString(),
[ignite-3] 18/19: IGNITE-17607 Add test
Posted by is...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
isapego pushed a commit to branch ignite-17607
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 32cb4532b17f84ec48a2d34a198fc27e3d981d4a
Author: Igor Sapego <is...@apache.org>
AuthorDate: Mon Mar 20 18:09:21 2023 +0300
IGNITE-17607 Add test
---
modules/platforms/cpp/tests/client-test/compute_test.cpp | 15 +++++++++++++++
1 file changed, 15 insertions(+)
diff --git a/modules/platforms/cpp/tests/client-test/compute_test.cpp b/modules/platforms/cpp/tests/client-test/compute_test.cpp
index 318549b636..75bdb101fb 100644
--- a/modules/platforms/cpp/tests/client-test/compute_test.cpp
+++ b/modules/platforms/cpp/tests/client-test/compute_test.cpp
@@ -258,3 +258,18 @@ TEST_F(compute_test, execute_colocated) {
EXPECT_EQ(expectedNodeName, resNodeName);
}
}
+
+TEST_F(compute_test, execute_colocated_throws_when_table_does_not_exist) {
+ EXPECT_THROW(
+ {
+ try {
+ (void) m_client.get_compute().execute_colocated("unknownTable", get_tuple(42), ECHO_JOB, {});
+ } catch (const ignite_error &e) {
+ EXPECT_STREQ("Table does not exist: 'unknownTable'", e.what());
+ throw;
+ }
+ },
+ ignite_error);
+}
+
+
[ignite-3] 02/19: IGNITE-17607 execute_async stub
Posted by is...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
isapego pushed a commit to branch ignite-17607
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 72c529be8e77e27fc3f8234e23438eb878ea9bc7
Author: Igor Sapego <is...@apache.org>
AuthorDate: Tue Mar 7 12:59:32 2023 +0300
IGNITE-17607 execute_async stub
---
modules/platforms/cpp/ignite/client/CMakeLists.txt | 1 +
.../cpp/ignite/client/compute/compute.cpp | 38 ++++++++++++++--
.../platforms/cpp/ignite/client/compute/compute.h | 30 +++++++------
.../compute/compute_impl.cpp} | 13 +++---
.../compute.h => detail/compute/compute_impl.h} | 50 +++++++---------------
5 files changed, 74 insertions(+), 58 deletions(-)
diff --git a/modules/platforms/cpp/ignite/client/CMakeLists.txt b/modules/platforms/cpp/ignite/client/CMakeLists.txt
index e84687dce8..626b172a60 100644
--- a/modules/platforms/cpp/ignite/client/CMakeLists.txt
+++ b/modules/platforms/cpp/ignite/client/CMakeLists.txt
@@ -33,6 +33,7 @@ set(SOURCES
detail/cluster_connection.cpp
detail/utils.cpp
detail/node_connection.cpp
+ detail/compute/compute_impl.cpp
detail/sql/sql_impl.cpp
detail/table/table_impl.cpp
detail/table/tables_impl.cpp
diff --git a/modules/platforms/cpp/ignite/client/compute/compute.cpp b/modules/platforms/cpp/ignite/client/compute/compute.cpp
index 8476afbaab..1a4de991b7 100644
--- a/modules/platforms/cpp/ignite/client/compute/compute.cpp
+++ b/modules/platforms/cpp/ignite/client/compute/compute.cpp
@@ -16,13 +16,45 @@
*/
#include "ignite/client/compute/compute.h"
-//#include "ignite/client/detail/compute/compute_impl.h"
+#include "ignite/client/detail/compute/compute_impl.h"
+
+#include <random>
namespace ignite {
-void compute::execute_async(std::vector<cluster_node> nodes, std::string_view job_class_name,
+template<typename T>
+typename T::value_type get_random_element(const T &cont) {
+ // TODO: Move to utils
+ static std::mutex randomMutex;
+ static std::random_device rd;
+ static std::mt19937 gen(rd());
+
+ assert(!cont.empty());
+
+ std::uniform_int_distribution<size_t> distrib(0, cont.size() - 1);
+
+ std::lock_guard<std::mutex> lock(randomMutex);
+
+ return cont[distrib(gen)];
+}
+
+/**
+ * Check value argument.
+ *
+ * @param value Value tuple.
+ */
+template<typename T>
+void inline check_non_empty(const T &cont, const std::string& title) {
+ if (cont.empty())
+ throw ignite_error(title + " can not be empty");
+}
+
+void compute::execute_async(const std::vector<cluster_node>& nodes, std::string_view job_class_name,
std::vector<primitive> args, ignite_callback<std::optional<primitive>> callback) {
- // TODO: Implement me
+ check_non_empty(nodes, "Nodes container");
+ check_non_empty(job_class_name, "Job class name");
+
+ m_impl->execute_on_one_node(get_random_element(nodes), job_class_name, std::move(args), std::move(callback));
}
} // namespace ignite
diff --git a/modules/platforms/cpp/ignite/client/compute/compute.h b/modules/platforms/cpp/ignite/client/compute/compute.h
index 8f142bb835..d4d5f511dd 100644
--- a/modules/platforms/cpp/ignite/client/compute/compute.h
+++ b/modules/platforms/cpp/ignite/client/compute/compute.h
@@ -43,26 +43,28 @@ public:
compute() = delete;
/**
- * Executes single SQL statement asynchronously and returns rows.
+ * Executes a compute job represented by the given class on one of the specified nodes asynchronously.
*
- * @param tx Optional transaction. If nullptr implicit transaction for this single operation is used.
- * @param statement Statement to execute.
- * @param args Arguments for the statement.
- * @param callback A callback called on operation completion with SQL result set.
+ * @param nodes Nodes to use for the job execution.
+ * @param job_class_name Java class name of the job to execute.
+ * @param args Job arguments.
+ * @param callback A callback called on operation completion with job execution result.
*/
- IGNITE_API void execute_async(std::vector<cluster_node> nodes, std::string_view job_class_name, std::vector<primitive> args,
- ignite_callback<std::optional<primitive>> callback);
+ IGNITE_API void execute_async(const std::vector<cluster_node>& nodes, std::string_view job_class_name,
+ std::vector<primitive> args, ignite_callback<std::optional<primitive>> callback);
/**
- * Executes single SQL statement and returns rows.
+ * Executes a compute job represented by the given class on one of the specified nodes.
*
- * @param tx Optional transaction. If nullptr implicit transaction for this single operation is used.
- * @param statement Statement to execute.
- * @param args Arguments for the statement.
- * @return SQL result set.
+ * @param nodes Nodes to use for the job execution.
+ * @param job_class_name Java class name of the job to execute.
+ * @param args Job arguments.
+ * @return Job execution result.
*/
- IGNITE_API std::optional<primitive> execute(std::vector<cluster_node> nodes, std::string_view job_class_name, std::vector<primitive> args) {
- return sync<std::optional<primitive>>([this, nodes = std::move(nodes), job_class_name, args = std::move(args)](auto callback) mutable {
+ IGNITE_API std::optional<primitive> execute(std::vector<cluster_node> nodes, std::string_view job_class_name,
+ std::vector<primitive> args) {
+ return sync<std::optional<primitive>>(
+ [this, nodes = std::move(nodes), job_class_name, args = std::move(args)](auto callback) mutable {
execute_async(std::move(nodes), job_class_name, std::move(args), std::move(callback));
});
}
diff --git a/modules/platforms/cpp/ignite/client/compute/compute.cpp b/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.cpp
similarity index 69%
copy from modules/platforms/cpp/ignite/client/compute/compute.cpp
copy to modules/platforms/cpp/ignite/client/detail/compute/compute_impl.cpp
index 8476afbaab..fa719a66a3 100644
--- a/modules/platforms/cpp/ignite/client/compute/compute.cpp
+++ b/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.cpp
@@ -15,14 +15,13 @@
* limitations under the License.
*/
-#include "ignite/client/compute/compute.h"
-//#include "ignite/client/detail/compute/compute_impl.h"
+#include "ignite/client/detail/compute/compute_impl.h"
-namespace ignite {
+namespace ignite::detail {
-void compute::execute_async(std::vector<cluster_node> nodes, std::string_view job_class_name,
- std::vector<primitive> args, ignite_callback<std::optional<primitive>> callback) {
- // TODO: Implement me
+void compute_impl::execute_on_one_node(cluster_node nodes, std::string_view job_class_name, std::vector<primitive> args,
+ ignite_callback<std::optional<primitive>> callback) {
+ // TODO: Implement me.
}
-} // namespace ignite
+} // namespace ignite::detail
diff --git a/modules/platforms/cpp/ignite/client/compute/compute.h b/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.h
similarity index 55%
copy from modules/platforms/cpp/ignite/client/compute/compute.h
copy to modules/platforms/cpp/ignite/client/detail/compute/compute_impl.h
index 8f142bb835..a43167fa3c 100644
--- a/modules/platforms/cpp/ignite/client/compute/compute.h
+++ b/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.h
@@ -26,21 +26,17 @@
#include <memory>
#include <utility>
-namespace ignite {
-
-namespace detail {
-class compute_impl;
-}
+namespace ignite::detail {
/**
- * Ignite Compute facade.
+ * Ignite Compute implementation.
*/
-class compute {
+class compute_impl {
friend class ignite_client;
public:
// Delete
- compute() = delete;
+ compute_impl() = delete;
/**
* Executes single SQL statement asynchronously and returns rows.
@@ -50,34 +46,20 @@ public:
* @param args Arguments for the statement.
* @param callback A callback called on operation completion with SQL result set.
*/
- IGNITE_API void execute_async(std::vector<cluster_node> nodes, std::string_view job_class_name, std::vector<primitive> args,
+ void execute_on_one_node(cluster_node nodes, std::string_view job_class_name, std::vector<primitive> args,
ignite_callback<std::optional<primitive>> callback);
- /**
- * Executes single SQL statement and returns rows.
- *
- * @param tx Optional transaction. If nullptr implicit transaction for this single operation is used.
- * @param statement Statement to execute.
- * @param args Arguments for the statement.
- * @return SQL result set.
- */
- IGNITE_API std::optional<primitive> execute(std::vector<cluster_node> nodes, std::string_view job_class_name, std::vector<primitive> args) {
- return sync<std::optional<primitive>>([this, nodes = std::move(nodes), job_class_name, args = std::move(args)](auto callback) mutable {
- execute_async(std::move(nodes), job_class_name, std::move(args), std::move(callback));
- });
- }
-
private:
- /**
- * Constructor
- *
- * @param impl Implementation
- */
- explicit compute(std::shared_ptr<detail::compute_impl> impl)
- : m_impl(std::move(impl)) {}
-
- /** Implementation. */
- std::shared_ptr<detail::compute_impl> m_impl;
+// /**
+// * Constructor
+// *
+// * @param impl Implementation
+// */
+// explicit compute(std::shared_ptr<detail::compute_impl> impl)
+// : m_impl(std::move(impl)) {}
+//
+// /** Implementation. */
+// std::shared_ptr<detail::compute_impl> m_impl;
};
-} // namespace ignite
+} // namespace ignite::detail
[ignite-3] 13/19: IGNITE-17607 Add all arg types test
Posted by is...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
isapego pushed a commit to branch ignite-17607
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit f36c368210a4a349052454b36dbf066d4d25f037
Author: Igor Sapego <is...@apache.org>
AuthorDate: Wed Mar 15 02:32:50 2023 +0300
IGNITE-17607 Add all arg types test
---
.../platforms/cpp/ignite/client/detail/utils.cpp | 88 ++++++++++++++++++----
.../cpp/tests/client-test/compute_test.cpp | 69 +++++++++++++++++
2 files changed, 141 insertions(+), 16 deletions(-)
diff --git a/modules/platforms/cpp/ignite/client/detail/utils.cpp b/modules/platforms/cpp/ignite/client/detail/utils.cpp
index 5a21f1c37f..60a5aa9e79 100644
--- a/modules/platforms/cpp/ignite/client/detail/utils.cpp
+++ b/modules/platforms/cpp/ignite/client/detail/utils.cpp
@@ -106,16 +106,44 @@ void claim_primitive_with_type(binary_tuple_builder &builder, const primitive &v
builder.claim(ignite_type::BINARY, data);
break;
}
-
- case column_type::DECIMAL:
- case column_type::DATE:
- case column_type::TIME:
- case column_type::DATETIME:
- case column_type::TIMESTAMP:
- case column_type::BITMASK:
+ case column_type::DECIMAL: {
+ const auto& dec_value = value.get<big_decimal>();
+ claim_type_and_scale(builder, ignite_type::DECIMAL, dec_value.get_scale());
+ builder.claim_number(dec_value);
+ break;
+ }
+ case column_type::NUMBER: {
+ claim_type_and_scale(builder, ignite_type::NUMBER);
+ builder.claim_number(value.get<big_integer>());
+ break;
+ }
+ case column_type::DATE: {
+ claim_type_and_scale(builder, ignite_type::DATE);
+ builder.claim_date(value.get<ignite_date>());
+ break;
+ }
+ case column_type::TIME: {
+ claim_type_and_scale(builder, ignite_type::TIME);
+ builder.claim_time(value.get<ignite_time>());
+ break;
+ }
+ case column_type::DATETIME: {
+ claim_type_and_scale(builder, ignite_type::DATETIME);
+ builder.claim_date_time(value.get<ignite_date_time>());
+ break;
+ }
+ case column_type::TIMESTAMP: {
+ claim_type_and_scale(builder, ignite_type::TIMESTAMP);
+ builder.claim_timestamp(value.get<ignite_timestamp>());
+ break;
+ }
+ case column_type::BITMASK: {
+ claim_type_and_scale(builder, ignite_type::BITMASK);
+ builder.claim_bytes(value.get<bit_array>().get_raw());
+ break;
+ }
case column_type::PERIOD:
case column_type::DURATION:
- case column_type::NUMBER:
default:
throw ignite_error("Unsupported type: " + std::to_string(int(value.get_type())));
}
@@ -181,16 +209,44 @@ void append_primitive_with_type(binary_tuple_builder &builder, const primitive &
builder.append(ignite_type::BINARY, data);
break;
}
-
- case column_type::DECIMAL:
- case column_type::DATE:
- case column_type::TIME:
- case column_type::DATETIME:
- case column_type::TIMESTAMP:
- case column_type::BITMASK:
+ case column_type::DECIMAL: {
+ const auto& dec_value = value.get<big_decimal>();
+ append_type_and_scale(builder, ignite_type::DECIMAL, dec_value.get_scale());
+ builder.append_number(dec_value);
+ break;
+ }
+ case column_type::NUMBER: {
+ append_type_and_scale(builder, ignite_type::NUMBER);
+ builder.append_number(value.get<big_integer>());
+ break;
+ }
+ case column_type::DATE: {
+ append_type_and_scale(builder, ignite_type::DATE);
+ builder.append_date(value.get<ignite_date>());
+ break;
+ }
+ case column_type::TIME: {
+ append_type_and_scale(builder, ignite_type::TIME);
+ builder.append_time(value.get<ignite_time>());
+ break;
+ }
+ case column_type::DATETIME: {
+ append_type_and_scale(builder, ignite_type::DATETIME);
+ builder.append_date_time(value.get<ignite_date_time>());
+ break;
+ }
+ case column_type::TIMESTAMP: {
+ append_type_and_scale(builder, ignite_type::TIMESTAMP);
+ builder.append_timestamp(value.get<ignite_timestamp>());
+ break;
+ }
+ case column_type::BITMASK: {
+ append_type_and_scale(builder, ignite_type::BITMASK);
+ builder.append_bytes(value.get<bit_array>().get_raw());
+ break;
+ }
case column_type::PERIOD:
case column_type::DURATION:
- case column_type::NUMBER:
default:
throw ignite_error("Unsupported type: " + std::to_string(int(value.get_type())));
}
diff --git a/modules/platforms/cpp/tests/client-test/compute_test.cpp b/modules/platforms/cpp/tests/client-test/compute_test.cpp
index e6e2324970..4414e7ec3f 100644
--- a/modules/platforms/cpp/tests/client-test/compute_test.cpp
+++ b/modules/platforms/cpp/tests/client-test/compute_test.cpp
@@ -24,6 +24,8 @@
#include <gtest/gtest.h>
#include <chrono>
+#include <limits>
+#include <sstream>
using namespace ignite;
@@ -65,6 +67,33 @@ protected:
return {nodes.begin(), nodes.end()};
}
+ /**
+ * Check that passed argument returned in a specific string form.
+ *
+ * @tparam T Type of the argument.
+ * @param value Argument.
+ * @param expected_str Expected string form.
+ */
+ template<typename T>
+ void check_argument(T value, const std::string &expected_str) {
+ auto cluster_nodes = m_client.get_cluster_nodes();
+ auto result = m_client.get_compute().execute(cluster_nodes, ECHO_JOB, {value, expected_str});
+
+ ASSERT_TRUE(result.has_value());
+ EXPECT_EQ(result.value().template get<T>(), value);
+ }
+
+ /**
+ * Check that passed argument returned in an expected string form.
+ *
+ * @tparam T Type of the argument.
+ * @param value Argument.
+ */
+ template<typename T>
+ void check_argument(T value) {
+ check_argument(std::move(value), std::to_string(value));
+ }
+
/** Ignite client. */
ignite_client m_client;
};
@@ -172,6 +201,46 @@ TEST_F(compute_test, unknown_node_throws) {
ignite_error);
}
+TEST_F(compute_test, all_arg_types) {
+ check_argument<std::int8_t>(42);
+ check_argument<std::int8_t>(std::numeric_limits<std::int8_t>::max());
+ check_argument<std::int8_t>(std::numeric_limits<std::int8_t>::min());
+
+ check_argument<std::int16_t>(4242);
+ check_argument<std::int16_t>(std::numeric_limits<std::int16_t>::max());
+ check_argument<std::int16_t>(std::numeric_limits<std::int16_t>::min());
+
+ check_argument<std::int32_t>(424242);
+ check_argument<std::int32_t>(std::numeric_limits<std::int32_t>::max());
+ check_argument<std::int32_t>(std::numeric_limits<std::int32_t>::min());
+
+ check_argument<std::int64_t>(424242424242);
+ check_argument<std::int64_t>(std::numeric_limits<std::int64_t>::max());
+ check_argument<std::int64_t>(std::numeric_limits<std::int64_t>::min());
+
+ check_argument<float>(0.123456f);
+ check_argument<float>(std::numeric_limits<float>::max(), "3.4028235E38");
+ check_argument<float>(std::numeric_limits<float>::min(), "1.17549435E-38");
+
+ check_argument<double>(0.987654);
+ check_argument<double>(std::numeric_limits<double>::max(), "1.7976931348623157E308");
+ check_argument<double>(std::numeric_limits<double>::min(), "2.2250738585072014E-308");
+
+ check_argument<big_decimal>({123456, 3}, "123.456");
+ check_argument<big_decimal>({}, "0");
+ check_argument<big_decimal>({1, 0}, "1");
+
+ auto str_dec = "12345678909876543211234567890.987654321";
+ check_argument<big_decimal>(big_decimal(str_dec), str_dec);
+
+ check_argument<ignite_date>({2021, 11, 18}, "2021-11-18");
+ check_argument<ignite_time>({13, 8, 55, 266574889}, "13:08:55.266574889");
+ check_argument<ignite_date_time>({{2021, 11, 18}, {13, 8, 55, 266574889}}, "2021-11-18T13:08:55.266574889");
+
+ check_argument<uuid>({0, 0}, "00000000-0000-0000-0000-000000000000");
+ check_argument<uuid>({0x123e4567e89b12d3, 0x7456426614174000}, "123e4567-e89b-12d3-7456-426614174000");
+}
+
[ignite-3] 14/19: IGNITE-17607 Re-factor argument checks
Posted by is...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
isapego pushed a commit to branch ignite-17607
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit a661c397633d38d7507fc988395954ec750831fd
Author: Igor Sapego <is...@apache.org>
AuthorDate: Thu Mar 16 12:02:13 2023 +0300
IGNITE-17607 Re-factor argument checks
---
.../cpp/ignite/client/compute/compute.cpp | 31 +++++------
.../platforms/cpp/ignite/client/compute/compute.h | 14 +++++
.../ignite/client/detail/argument_check_utils.h | 65 ++++++++++++++++++++++
.../cpp/ignite/client/table/key_value_view.cpp | 59 +++++++-------------
.../cpp/ignite/client/table/record_view.cpp | 32 ++++-------
.../cpp/tests/client-test/compute_test.cpp | 5 --
6 files changed, 126 insertions(+), 80 deletions(-)
diff --git a/modules/platforms/cpp/ignite/client/compute/compute.cpp b/modules/platforms/cpp/ignite/client/compute/compute.cpp
index eab6b63092..77e803d306 100644
--- a/modules/platforms/cpp/ignite/client/compute/compute.cpp
+++ b/modules/platforms/cpp/ignite/client/compute/compute.cpp
@@ -16,6 +16,7 @@
*/
#include "ignite/client/compute/compute.h"
+#include "ignite/client/detail/argument_check_utils.h"
#include "ignite/client/detail/compute/compute_impl.h"
#include <random>
@@ -38,15 +39,12 @@ typename T::value_type get_random_element(const T &cont) {
return cont[distrib(gen)];
}
-/**
- * Check value argument.
- *
- * @param value Value tuple.
- */
-template<typename T>
-void inline check_non_empty(const T &cont, const std::string& title) {
- if (cont.empty())
- throw ignite_error(title + " can not be empty");
+void compute::execute_async(const std::vector<cluster_node>& nodes, std::string_view job_class_name,
+ const std::vector<primitive>& args, ignite_callback<std::optional<primitive>> callback) {
+ detail::arg_check::container_non_empty(nodes, "Nodes container");
+ detail::arg_check::container_non_empty(job_class_name, "Job class name");
+
+ m_impl->execute_on_one_node(get_random_element(nodes), job_class_name, args, std::move(callback));
}
void compute::broadcast_async(const std::set<cluster_node>& nodes, std::string_view job_class_name,
@@ -54,8 +52,8 @@ void compute::broadcast_async(const std::set<cluster_node>& nodes, std::string_v
ignite_callback<std::map<cluster_node, ignite_result<std::optional<primitive>>>> callback) {
typedef std::map<cluster_node, ignite_result<std::optional<primitive>>> result_type;
- check_non_empty(nodes, "Nodes set");
- check_non_empty(job_class_name, "Job class name");
+ detail::arg_check::container_non_empty(nodes, "Nodes set");
+ detail::arg_check::container_non_empty(job_class_name, "Job class name");
struct result_group {
explicit result_group(std::int32_t cnt, ignite_callback<result_type> &&cb) : m_cnt(cnt), m_callback(cb) {}
@@ -81,12 +79,13 @@ void compute::broadcast_async(const std::set<cluster_node>& nodes, std::string_v
}
}
-void compute::execute_async(const std::vector<cluster_node>& nodes, std::string_view job_class_name,
- const std::vector<primitive>& args, ignite_callback<std::optional<primitive>> callback) {
- check_non_empty(nodes, "Nodes container");
- check_non_empty(job_class_name, "Job class name");
+void compute::execute_colocated_async(const std::string &table_name, const ignite_tuple &key,
+ std::string_view job_class_name, const std::vector<primitive> &args,
+ ignite_callback<std::optional<primitive>> callback) {
+ detail::arg_check::container_non_empty(table_name, "Table name");
+ detail::arg_check::container_non_empty(job_class_name, "Job class name");
- m_impl->execute_on_one_node(get_random_element(nodes), job_class_name, args, std::move(callback));
+ // TODO: Implement me.
}
} // namespace ignite
diff --git a/modules/platforms/cpp/ignite/client/compute/compute.h b/modules/platforms/cpp/ignite/client/compute/compute.h
index 312bc04875..ff40355d8a 100644
--- a/modules/platforms/cpp/ignite/client/compute/compute.h
+++ b/modules/platforms/cpp/ignite/client/compute/compute.h
@@ -19,6 +19,7 @@
#include "ignite/client/network/cluster_node.h"
#include "ignite/client/primitive.h"
+#include "ignite/client/table/ignite_tuple.h"
#include "ignite/client/transaction/transaction.h"
#include "ignite/common/config.h"
#include "ignite/common/ignite_result.h"
@@ -99,6 +100,19 @@ public:
});
}
+ /**
+ * Executes a compute job represented by the given class on one of the specified nodes asynchronously.
+ *
+ * @param tableName Name of the table to be used with @c key to determine target node.
+ * @param key Table key to be used to determine the target node for job execution.
+ * @param job_class_name Java class name of the job to execute.
+ * @param args Job arguments.
+ * @param callback A callback called on operation completion with job execution result.
+ */
+ IGNITE_API void execute_colocated_async(const std::string &table_name, const ignite_tuple& key,
+ std::string_view job_class_name, const std::vector<primitive>& args,
+ ignite_callback<std::optional<primitive>> callback);
+
private:
/**
* Constructor
diff --git a/modules/platforms/cpp/ignite/client/detail/argument_check_utils.h b/modules/platforms/cpp/ignite/client/detail/argument_check_utils.h
new file mode 100644
index 0000000000..fac3b86e45
--- /dev/null
+++ b/modules/platforms/cpp/ignite/client/detail/argument_check_utils.h
@@ -0,0 +1,65 @@
+/*
+ * 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/client/table/ignite_tuple.h"
+
+#include <string>
+
+namespace ignite::detail::arg_check {
+
+/**
+ * Check key argument.
+ *
+ * @param key Key tuple.
+ */
+void inline tuple_non_empty(const ignite_tuple &value, const std::string& title) {
+ if (0 == value.column_count())
+ throw ignite_error(title + " can not be empty");
+}
+
+/**
+ * Check key argument.
+ *
+ * @param key Key tuple.
+ */
+void inline key_tuple_non_empty(const ignite_tuple &key) {
+ tuple_non_empty(key, "Key tuple");
+}
+
+/**
+ * Check value argument.
+ *
+ * @param value Value tuple.
+ */
+void inline value_tuple_non_empty(const ignite_tuple &value) {
+ tuple_non_empty(value, "Value tuple");
+}
+
+/**
+ * Check value argument.
+ *
+ * @param value Value tuple.
+ */
+template<typename T>
+void inline container_non_empty(const T &cont, const std::string& title) {
+ if (cont.empty())
+ throw ignite_error(title + " can not be empty");
+}
+
+} // namespace ignite::detail::arg_check
diff --git a/modules/platforms/cpp/ignite/client/table/key_value_view.cpp b/modules/platforms/cpp/ignite/client/table/key_value_view.cpp
index ccd229041b..bb54ac2992 100644
--- a/modules/platforms/cpp/ignite/client/table/key_value_view.cpp
+++ b/modules/platforms/cpp/ignite/client/table/key_value_view.cpp
@@ -16,30 +16,11 @@
*/
#include "ignite/client/table/key_value_view.h"
+#include "ignite/client/detail/argument_check_utils.h"
#include "ignite/client/detail/table/table_impl.h"
namespace ignite {
-/**
- * Check key argument.
- *
- * @param key Key tuple.
- */
-void inline check_key_argument(const ignite_tuple &key) {
- if (0 == key.column_count())
- throw ignite_error("Key tuple can not be empty");
-}
-
-/**
- * Check value argument.
- *
- * @param value Value tuple.
- */
-void inline check_value_argument(const ignite_tuple &value) {
- if (0 == value.column_count())
- throw ignite_error("Value tuple can not be empty");
-}
-
/**
* Process multiple kv pairs by uniting key and value part of the tuple
* to a single record.
@@ -58,15 +39,15 @@ std::vector<ignite_tuple> concat_records(const std::vector<std::pair<ignite_tupl
void key_value_view<ignite_tuple, ignite_tuple>::get_async(
transaction *tx, const ignite_tuple &key, ignite_callback<std::optional<value_type>> callback) {
- check_key_argument(key);
+ detail::arg_check::key_tuple_non_empty(key);
m_impl->get_async(tx, key, std::move(callback));
}
void key_value_view<ignite_tuple, ignite_tuple>::put_async(
transaction *tx, const key_type &key, const value_type &value, ignite_callback<void> callback) {
- check_key_argument(key);
- check_value_argument(value);
+ detail::arg_check::key_tuple_non_empty(key);
+ detail::arg_check::value_tuple_non_empty(value);
m_impl->upsert_async(tx, detail::concat(key, value), std::move(callback));
}
@@ -83,7 +64,7 @@ void key_value_view<ignite_tuple, ignite_tuple>::get_all_async(
void key_value_view<ignite_tuple, ignite_tuple>::contains_async(
transaction *tx, const ignite_tuple &key, ignite_callback<bool> callback) {
- check_key_argument(key);
+ detail::arg_check::key_tuple_non_empty(key);
m_impl->contains_async(tx, key, std::move(callback));
}
@@ -100,31 +81,31 @@ void key_value_view<ignite_tuple, ignite_tuple>::put_all_async(
void key_value_view<ignite_tuple, ignite_tuple>::get_and_put_async(transaction *tx, const key_type &key,
const value_type &value, ignite_callback<std::optional<value_type>> callback) {
- check_key_argument(key);
- check_value_argument(value);
+ detail::arg_check::key_tuple_non_empty(key);
+ detail::arg_check::value_tuple_non_empty(value);
m_impl->get_and_upsert_async(tx, detail::concat(key, value), std::move(callback));
}
void key_value_view<ignite_tuple, ignite_tuple>::put_if_absent_async(
transaction *tx, const key_type &key, const value_type &value, ignite_callback<bool> callback) {
- check_key_argument(key);
- check_value_argument(value);
+ detail::arg_check::key_tuple_non_empty(key);
+ detail::arg_check::value_tuple_non_empty(value);
m_impl->insert_async(tx, detail::concat(key, value), std::move(callback));
}
void key_value_view<ignite_tuple, ignite_tuple>::remove_async(
transaction *tx, const ignite_tuple &key, ignite_callback<bool> callback) {
- check_key_argument(key);
+ detail::arg_check::key_tuple_non_empty(key);
m_impl->remove_async(tx, key, std::move(callback));
}
void key_value_view<ignite_tuple, ignite_tuple>::remove_async(
transaction *tx, const key_type &key, const value_type &value, ignite_callback<bool> callback) {
- check_key_argument(key);
- check_value_argument(value);
+ detail::arg_check::key_tuple_non_empty(key);
+ detail::arg_check::value_tuple_non_empty(value);
m_impl->remove_exact_async(tx, detail::concat(key, value), std::move(callback));
}
@@ -151,32 +132,32 @@ void key_value_view<ignite_tuple, ignite_tuple>::remove_all_async(transaction *t
void key_value_view<ignite_tuple, ignite_tuple>::get_and_remove_async(
transaction *tx, const ignite_tuple &key, ignite_callback<std::optional<value_type>> callback) {
- check_key_argument(key);
+ detail::arg_check::key_tuple_non_empty(key);
m_impl->get_and_remove_async(tx, key, std::move(callback));
}
void key_value_view<ignite_tuple, ignite_tuple>::replace_async(
transaction *tx, const key_type &key, const value_type &value, ignite_callback<bool> callback) {
- check_key_argument(key);
- check_value_argument(value);
+ detail::arg_check::key_tuple_non_empty(key);
+ detail::arg_check::value_tuple_non_empty(value);
m_impl->replace_async(tx, detail::concat(key, value), std::move(callback));
}
void key_value_view<ignite_tuple, ignite_tuple>::replace_async(transaction *tx, const key_type &key,
const value_type &old_value, const value_type &new_value, ignite_callback<bool> callback) {
- check_key_argument(key);
- check_value_argument(old_value);
- check_value_argument(new_value);
+ detail::arg_check::key_tuple_non_empty(key);
+ detail::arg_check::value_tuple_non_empty(old_value);
+ detail::arg_check::value_tuple_non_empty(new_value);
m_impl->replace_async(tx, detail::concat(key, old_value), detail::concat(key, new_value), std::move(callback));
}
void key_value_view<ignite_tuple, ignite_tuple>::get_and_replace_async(transaction *tx, const key_type &key,
const value_type &value, ignite_callback<std::optional<value_type>> callback) {
- check_key_argument(key);
- check_value_argument(value);
+ detail::arg_check::key_tuple_non_empty(key);
+ detail::arg_check::value_tuple_non_empty(value);
m_impl->get_and_replace_async(tx, detail::concat(key, value), std::move(callback));
}
diff --git a/modules/platforms/cpp/ignite/client/table/record_view.cpp b/modules/platforms/cpp/ignite/client/table/record_view.cpp
index 392b115525..a21ffe01b7 100644
--- a/modules/platforms/cpp/ignite/client/table/record_view.cpp
+++ b/modules/platforms/cpp/ignite/client/table/record_view.cpp
@@ -17,21 +17,20 @@
#include "ignite/client/table/record_view.h"
#include "ignite/client/detail/table/table_impl.h"
+#include "ignite/client/detail/argument_check_utils.h"
namespace ignite {
void record_view<ignite_tuple>::get_async(
transaction *tx, const ignite_tuple &key, ignite_callback<std::optional<value_type>> callback) {
- if (0 == key.column_count())
- throw ignite_error("Tuple can not be empty");
+ detail::arg_check::tuple_non_empty(key, "Tuple");
m_impl->get_async(tx, key, std::move(callback));
}
void record_view<ignite_tuple>::upsert_async(
transaction *tx, const ignite_tuple &record, ignite_callback<void> callback) {
- if (0 == record.column_count())
- throw ignite_error("Tuple can not be empty");
+ detail::arg_check::tuple_non_empty(record, "Tuple");
m_impl->upsert_async(tx, record, std::move(callback));
}
@@ -58,16 +57,14 @@ void record_view<ignite_tuple>::upsert_all_async(
void record_view<ignite_tuple>::get_and_upsert_async(
transaction *tx, const ignite_tuple &record, ignite_callback<std::optional<value_type>> callback) {
- if (0 == record.column_count())
- throw ignite_error("Tuple can not be empty");
+ detail::arg_check::tuple_non_empty(record, "Tuple");
m_impl->get_and_upsert_async(tx, record, std::move(callback));
}
void record_view<ignite_tuple>::insert_async(
transaction *tx, const ignite_tuple &record, ignite_callback<bool> callback) {
- if (0 == record.column_count())
- throw ignite_error("Tuple can not be empty");
+ detail::arg_check::tuple_non_empty(record, "Tuple");
m_impl->insert_async(tx, record, std::move(callback));
}
@@ -84,47 +81,42 @@ void record_view<ignite_tuple>::insert_all_async(
void record_view<ignite_tuple>::replace_async(
transaction *tx, const ignite_tuple &record, ignite_callback<bool> callback) {
- if (0 == record.column_count())
- throw ignite_error("Tuple can not be empty");
+ detail::arg_check::tuple_non_empty(record, "Tuple");
m_impl->replace_async(tx, record, std::move(callback));
}
void record_view<ignite_tuple>::replace_async(
transaction *tx, const ignite_tuple &record, const ignite_tuple &new_record, ignite_callback<bool> callback) {
- if (0 == record.column_count() || 0 == new_record.column_count())
- throw ignite_error("Tuple can not be empty");
+ detail::arg_check::tuple_non_empty(record, "Tuple");
+ detail::arg_check::tuple_non_empty(new_record, "Tuple");
m_impl->replace_async(tx, record, new_record, std::move(callback));
}
void record_view<ignite_tuple>::get_and_replace_async(
transaction *tx, const ignite_tuple &record, ignite_callback<std::optional<value_type>> callback) {
- if (0 == record.column_count())
- throw ignite_error("Tuple can not be empty");
+ detail::arg_check::tuple_non_empty(record, "Tuple");
m_impl->get_and_replace_async(tx, record, std::move(callback));
}
void record_view<ignite_tuple>::remove_async(transaction *tx, const ignite_tuple &key, ignite_callback<bool> callback) {
- if (0 == key.column_count())
- throw ignite_error("Tuple can not be empty");
+ detail::arg_check::tuple_non_empty(key, "Tuple");
m_impl->remove_async(tx, key, std::move(callback));
}
void record_view<ignite_tuple>::remove_exact_async(
transaction *tx, const ignite_tuple &record, ignite_callback<bool> callback) {
- if (0 == record.column_count())
- throw ignite_error("Tuple can not be empty");
+ detail::arg_check::tuple_non_empty(record, "Tuple");
m_impl->remove_exact_async(tx, record, std::move(callback));
}
void record_view<ignite_tuple>::get_and_remove_async(
transaction *tx, const ignite_tuple &key, ignite_callback<std::optional<value_type>> callback) {
- if (0 == key.column_count())
- throw ignite_error("Tuple can not be empty");
+ detail::arg_check::tuple_non_empty(key, "Tuple");
m_impl->get_and_remove_async(tx, key, std::move(callback));
}
diff --git a/modules/platforms/cpp/tests/client-test/compute_test.cpp b/modules/platforms/cpp/tests/client-test/compute_test.cpp
index 4414e7ec3f..02004ee99d 100644
--- a/modules/platforms/cpp/tests/client-test/compute_test.cpp
+++ b/modules/platforms/cpp/tests/client-test/compute_test.cpp
@@ -25,7 +25,6 @@
#include <chrono>
#include <limits>
-#include <sstream>
using namespace ignite;
@@ -241,7 +240,3 @@ TEST_F(compute_test, all_arg_types) {
check_argument<uuid>({0x123e4567e89b12d3, 0x7456426614174000}, "123e4567-e89b-12d3-7456-426614174000");
}
-
-
-
-
[ignite-3] 12/19: IGNITE-17607 Add negative tests
Posted by is...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
isapego pushed a commit to branch ignite-17607
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit bd294cb7706df8600852b296fa46b75fe993baff
Author: Igor Sapego <is...@apache.org>
AuthorDate: Wed Mar 15 01:18:31 2023 +0300
IGNITE-17607 Add negative tests
---
.../cpp/tests/client-test/compute_test.cpp | 33 ++++++++++++++++++++++
.../cpp/tests/client-test/ignite_runner_suite.h | 2 ++
2 files changed, 35 insertions(+)
diff --git a/modules/platforms/cpp/tests/client-test/compute_test.cpp b/modules/platforms/cpp/tests/client-test/compute_test.cpp
index c35feba204..e6e2324970 100644
--- a/modules/platforms/cpp/tests/client-test/compute_test.cpp
+++ b/modules/platforms/cpp/tests/client-test/compute_test.cpp
@@ -139,6 +139,39 @@ TEST_F(compute_test, execute_with_args) {
EXPECT_EQ(result.value().get<std::string>(), "5.3_00000000-0000-0000-0000-000000000000_42_null");
}
+TEST_F(compute_test, job_error_propagates_to_client) {
+ auto cluster_nodes = m_client.get_cluster_nodes();
+
+ EXPECT_THROW(
+ {
+ try {
+ m_client.get_compute().execute(cluster_nodes, ERROR_JOB, {"unused"});
+ } catch (const ignite_error &e) {
+ EXPECT_THAT(e.what_str(), testing::HasSubstr("Custom job error"));
+ EXPECT_THAT(e.what_str(), testing::HasSubstr(
+ "org.apache.ignite.internal.runner.app.client.ItThinClientComputeTest$CustomException"));
+ EXPECT_THAT(e.what_str(), testing::HasSubstr("IGN-TBL-3"));
+ throw;
+ }
+ },
+ ignite_error);
+}
+
+TEST_F(compute_test, unknown_node_throws) {
+ auto unknown_node = cluster_node("some", "random", {"127.0.0.1", 1234});
+
+ EXPECT_THROW(
+ {
+ try {
+ m_client.get_compute().execute({unknown_node}, ECHO_JOB, {"unused"});
+ } catch (const ignite_error &e) {
+ EXPECT_THAT(e.what_str(), testing::HasSubstr("Specified node is not present in the cluster: random"));
+ throw;
+ }
+ },
+ ignite_error);
+}
+
diff --git a/modules/platforms/cpp/tests/client-test/ignite_runner_suite.h b/modules/platforms/cpp/tests/client-test/ignite_runner_suite.h
index e45c04c56a..288a0f2bae 100644
--- a/modules/platforms/cpp/tests/client-test/ignite_runner_suite.h
+++ b/modules/platforms/cpp/tests/client-test/ignite_runner_suite.h
@@ -51,6 +51,8 @@ public:
inline static const std::string NODE_NAME_JOB = IT_THIN_CLIENT_COMPUTE_TEST + "$NodeNameJob";
inline static const std::string CONCAT_JOB = IT_THIN_CLIENT_COMPUTE_TEST + "$ConcatJob";
+ inline static const std::string ERROR_JOB = IT_THIN_CLIENT_COMPUTE_TEST + "$IgniteExceptionJob";
+ inline static const std::string ECHO_JOB = IT_THIN_CLIENT_COMPUTE_TEST + "$EchoJob";
static constexpr const char *KEY_COLUMN = "key";
static constexpr const char *VAL_COLUMN = "val";
[ignite-3] 19/19: IGNITE-17607 Add test
Posted by is...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
isapego pushed a commit to branch ignite-17607
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 2c8373c1edb915519bf698cd9d54764257a10c81
Author: Igor Sapego <is...@apache.org>
AuthorDate: Mon Mar 20 18:22:33 2023 +0300
IGNITE-17607 Add test
---
.../cpp/tests/client-test/compute_test.cpp | 37 ++++++++++++++++++++++
1 file changed, 37 insertions(+)
diff --git a/modules/platforms/cpp/tests/client-test/compute_test.cpp b/modules/platforms/cpp/tests/client-test/compute_test.cpp
index 75bdb101fb..85623fd138 100644
--- a/modules/platforms/cpp/tests/client-test/compute_test.cpp
+++ b/modules/platforms/cpp/tests/client-test/compute_test.cpp
@@ -272,4 +272,41 @@ TEST_F(compute_test, execute_colocated_throws_when_table_does_not_exist) {
ignite_error);
}
+TEST_F(compute_test, execute_colocated_throws_when_key_column_is_missing) {
+ EXPECT_THROW(
+ {
+ try {
+ (void) m_client.get_compute().execute_colocated(TABLE_1, get_tuple("some"), ECHO_JOB, {});
+ } catch (const ignite_error &e) {
+ EXPECT_THAT(e.what_str(), ::testing::HasSubstr("Missed key column: KEY"));
+ throw;
+ }
+ },
+ ignite_error);
+}
+
+TEST_F(compute_test, execute_colocated_throws_when_key_is_empty) {
+ EXPECT_THROW(
+ {
+ try {
+ (void) m_client.get_compute().execute_colocated(TABLE_1, {}, ECHO_JOB, {});
+ } catch (const ignite_error &e) {
+ EXPECT_EQ("Key tuple can not be empty", e.what_str());
+ throw;
+ }
+ },
+ ignite_error);
+}
+TEST_F(compute_test, exception_in_server_job_propogates_to_client) {
+ EXPECT_THROW(
+ {
+ try {
+ (void) m_client.get_compute().execute_colocated(TABLE_1, {}, ECHO_JOB, {});
+ } catch (const ignite_error &e) {
+ EXPECT_EQ("Key tuple can not be empty", e.what_str());
+ throw;
+ }
+ },
+ ignite_error);
+}
[ignite-3] 05/19: IGNITE-17607 get_cluster_nodes implemented
Posted by is...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
isapego pushed a commit to branch ignite-17607
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit f0aaea1e296f89cb0523d3e447c58ccd0fda080e
Author: Igor Sapego <is...@apache.org>
AuthorDate: Tue Mar 7 15:49:11 2023 +0300
IGNITE-17607 get_cluster_nodes implemented
---
modules/platforms/cpp/ignite/client/CMakeLists.txt | 1 +
.../ignite/client/detail/ignite_client_impl.cpp | 49 +++++++++++++++
.../cpp/ignite/client/detail/ignite_client_impl.h | 19 ++++++
.../cpp/ignite/client/detail/node_connection.cpp | 2 +-
.../platforms/cpp/ignite/client/ignite_client.cpp | 14 +++++
.../platforms/cpp/ignite/client/ignite_client.h | 25 ++++++++
.../platforms/cpp/tests/client-test/CMakeLists.txt | 1 +
.../cpp/tests/client-test/compute_test.cpp | 70 ++++++++++++++++++++++
8 files changed, 180 insertions(+), 1 deletion(-)
diff --git a/modules/platforms/cpp/ignite/client/CMakeLists.txt b/modules/platforms/cpp/ignite/client/CMakeLists.txt
index 626b172a60..e77c9f1f0f 100644
--- a/modules/platforms/cpp/ignite/client/CMakeLists.txt
+++ b/modules/platforms/cpp/ignite/client/CMakeLists.txt
@@ -31,6 +31,7 @@ set(SOURCES
transaction/transaction.cpp
transaction/transactions.cpp
detail/cluster_connection.cpp
+ detail/ignite_client_impl.cpp
detail/utils.cpp
detail/node_connection.cpp
detail/compute/compute_impl.cpp
diff --git a/modules/platforms/cpp/ignite/client/detail/ignite_client_impl.cpp b/modules/platforms/cpp/ignite/client/detail/ignite_client_impl.cpp
new file mode 100644
index 0000000000..e5cffefcb9
--- /dev/null
+++ b/modules/platforms/cpp/ignite/client/detail/ignite_client_impl.cpp
@@ -0,0 +1,49 @@
+/*
+ * 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 "ignite/client/detail/ignite_client_impl.h"
+
+#include <ignite/protocol/utils.h>
+
+namespace ignite::detail {
+
+void ignite_client_impl::get_cluster_nodes_async(ignite_callback<std::vector<cluster_node>> callback) {
+ auto reader_func = [](protocol::reader &reader) -> std::vector<cluster_node> {
+ std::vector<cluster_node> nodes;
+ nodes.reserve(reader.read_array_size());
+
+ reader.read_array_raw([&nodes](auto, const msgpack_object &object) {
+ auto fields = object.via.array;
+ assert(fields.size >= 4);
+
+ auto id = protocol::unpack_object<std::string>(fields.ptr[0]);
+ auto name = protocol::unpack_object<std::string>(fields.ptr[1]);
+ auto host = protocol::unpack_object<std::string>(fields.ptr[2]);
+ auto port = protocol::unpack_object<std::int32_t>(fields.ptr[3]);
+
+ nodes.emplace_back(std::move(id), std::move(name),
+ network::end_point{std::move(host), std::uint16_t(port)});
+ });
+
+ return nodes;
+ };
+
+ m_connection->perform_request_rd<std::vector<cluster_node>>(
+ client_operation::CLUSTER_GET_NODES, std::move(reader_func), std::move(callback));
+}
+
+} // namespace ignite::detail
diff --git a/modules/platforms/cpp/ignite/client/detail/ignite_client_impl.h b/modules/platforms/cpp/ignite/client/detail/ignite_client_impl.h
index 0d23ce681e..8e9123332e 100644
--- a/modules/platforms/cpp/ignite/client/detail/ignite_client_impl.h
+++ b/modules/platforms/cpp/ignite/client/detail/ignite_client_impl.h
@@ -17,6 +17,7 @@
#pragma once
+#include <ignite/client/detail/compute/compute_impl.h>
#include <ignite/client/detail/cluster_connection.h>
#include <ignite/client/detail/sql/sql_impl.h>
#include <ignite/client/detail/table/tables_impl.h>
@@ -93,6 +94,13 @@ public:
*/
[[nodiscard]] std::shared_ptr<sql_impl> get_sql_impl() const { return m_sql; }
+ /**
+ * Get Compute management API implementation.
+ *
+ * @return Compute management API implementation.
+ */
+ [[nodiscard]] std::shared_ptr<compute_impl> get_compute_impl() const { return m_compute; }
+
/**
* Get transactions management API implementation.
*
@@ -100,6 +108,14 @@ public:
*/
[[nodiscard]] std::shared_ptr<transactions_impl> get_transactions_impl() const { return m_transactions; }
+ /**
+ * Gets the cluster nodes asynchronously.
+ * NOTE: Temporary API to enable Compute until we have proper Cluster API.
+ *
+ * @param callback Callback called with the list of cluster nodes upon success.
+ */
+ void get_cluster_nodes_async(ignite_callback<std::vector<cluster_node>> callback);
+
private:
/** Configuration. */
const ignite_client_configuration m_configuration;
@@ -113,6 +129,9 @@ private:
/** SQL. */
std::shared_ptr<sql_impl> m_sql;
+ /** Compute. */
+ std::shared_ptr<compute_impl> m_compute;
+
/** Transactions. */
std::shared_ptr<transactions_impl> m_transactions;
};
diff --git a/modules/platforms/cpp/ignite/client/detail/node_connection.cpp b/modules/platforms/cpp/ignite/client/detail/node_connection.cpp
index 5811331b6e..a17c2b944f 100644
--- a/modules/platforms/cpp/ignite/client/detail/node_connection.cpp
+++ b/modules/platforms/cpp/ignite/client/detail/node_connection.cpp
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-#include "node_connection.h"
+#include "ignite/client/detail/node_connection.h"
#include <ignite/protocol/utils.h>
diff --git a/modules/platforms/cpp/ignite/client/ignite_client.cpp b/modules/platforms/cpp/ignite/client/ignite_client.cpp
index 8c33215b60..09475940d0 100644
--- a/modules/platforms/cpp/ignite/client/ignite_client.cpp
+++ b/modules/platforms/cpp/ignite/client/ignite_client.cpp
@@ -74,10 +74,24 @@ sql ignite_client::get_sql() const noexcept {
return sql(impl().get_sql_impl());
}
+compute ignite_client::get_compute() const noexcept {
+ return compute(impl().get_compute_impl());
+}
+
transactions ignite_client::get_transactions() const noexcept {
return transactions(impl().get_transactions_impl());
}
+void ignite_client::get_cluster_nodes_async(ignite_callback<std::vector<cluster_node>> callback) {
+ return impl().get_cluster_nodes_async(std::move(callback));
+}
+
+std::vector<cluster_node> ignite_client::get_cluster_nodes() {
+ return sync<std::vector<cluster_node>>([this](auto callback) mutable {
+ get_cluster_nodes_async(std::move(callback));
+ });
+}
+
detail::ignite_client_impl &ignite_client::impl() noexcept {
return *((detail::ignite_client_impl *) (m_impl.get()));
}
diff --git a/modules/platforms/cpp/ignite/client/ignite_client.h b/modules/platforms/cpp/ignite/client/ignite_client.h
index acc02807bb..b20e10e8dd 100644
--- a/modules/platforms/cpp/ignite/client/ignite_client.h
+++ b/modules/platforms/cpp/ignite/client/ignite_client.h
@@ -17,7 +17,9 @@
#pragma once
+#include "ignite/client/compute/compute.h"
#include "ignite/client/ignite_client_configuration.h"
+#include "ignite/client/network/cluster_node.h"
#include "ignite/client/sql/sql.h"
#include "ignite/client/table/tables.h"
#include "ignite/client/transaction/transactions.h"
@@ -108,6 +110,13 @@ public:
*/
[[nodiscard]] IGNITE_API sql get_sql() const noexcept;
+ /**
+ * Gets the Compute API.
+ *
+ * @return Compute API.
+ */
+ [[nodiscard]] IGNITE_API compute get_compute() const noexcept;
+
/**
* Gets the Transactions API.
*
@@ -115,6 +124,22 @@ public:
*/
[[nodiscard]] IGNITE_API transactions get_transactions() const noexcept;
+ /**
+ * Gets the cluster nodes asynchronously.
+ * NOTE: Temporary API to enable Compute until we have proper Cluster API.
+ *
+ * @param callback Callback called with the list of cluster nodes upon success.
+ */
+ IGNITE_API void get_cluster_nodes_async(ignite_callback<std::vector<cluster_node>> callback);
+
+ /**
+ * Gets the cluster nodes.
+ * NOTE: Temporary API to enable Compute until we have proper Cluster API.
+ *
+ * @return The list of cluster nodes upon success.
+ */
+ [[nodiscard]] IGNITE_API std::vector<cluster_node> get_cluster_nodes();
+
private:
/**
* Constructor
diff --git a/modules/platforms/cpp/tests/client-test/CMakeLists.txt b/modules/platforms/cpp/tests/client-test/CMakeLists.txt
index 90b43f2ad3..2d9611905f 100644
--- a/modules/platforms/cpp/tests/client-test/CMakeLists.txt
+++ b/modules/platforms/cpp/tests/client-test/CMakeLists.txt
@@ -20,6 +20,7 @@ project(ignite-client-test)
set(TARGET ${PROJECT_NAME})
set(SOURCES
+ compute_test.cpp
gtest_logger.h
ignite_client_test.cpp
ignite_runner_suite.h
diff --git a/modules/platforms/cpp/tests/client-test/compute_test.cpp b/modules/platforms/cpp/tests/client-test/compute_test.cpp
new file mode 100644
index 0000000000..10b6e3aa1b
--- /dev/null
+++ b/modules/platforms/cpp/tests/client-test/compute_test.cpp
@@ -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.
+ */
+
+#include "ignite_runner_suite.h"
+
+#include "ignite/client/ignite_client.h"
+#include "ignite/client/ignite_client_configuration.h"
+
+#include <gmock/gmock-matchers.h>
+#include <gtest/gtest.h>
+
+#include <chrono>
+
+using namespace ignite;
+
+/**
+ * Test suite.
+ */
+class compute_test : public ignite_runner_suite {
+protected:
+ void SetUp() override {
+ ignite_client_configuration cfg{get_node_addrs()};
+ cfg.set_logger(get_logger());
+
+ m_client = ignite_client::start(cfg, std::chrono::seconds(30));
+ }
+
+ void TearDown() override {
+ // remove all
+ }
+
+ /** Ignite client. */
+ ignite_client m_client;
+};
+
+TEST_F(compute_test, get_cluster_nodes) {
+ auto cluster_nodes = m_client.get_cluster_nodes();
+
+ std::sort(cluster_nodes.begin(), cluster_nodes.end(), [] (const auto &n1, const auto &n2) {
+ return n1.get_name() < n2.get_name();
+ });
+
+ EXPECT_FALSE(cluster_nodes.empty());
+ EXPECT_EQ(2, cluster_nodes.size());
+
+ EXPECT_FALSE(cluster_nodes[0].get_id().empty());
+ EXPECT_FALSE(cluster_nodes[1].get_id().empty());
+
+ EXPECT_EQ(3344, cluster_nodes[0].get_address().port);
+ EXPECT_EQ(3345, cluster_nodes[1].get_address().port);
+
+ EXPECT_FALSE(cluster_nodes[0].get_address().host.empty());
+ EXPECT_FALSE(cluster_nodes[1].get_address().host.empty());
+
+ EXPECT_EQ(cluster_nodes[0].get_address().host, cluster_nodes[1].get_address().host);
+}
[ignite-3] 06/19: IGNITE-17607 Add execute test, fix bug
Posted by is...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
isapego pushed a commit to branch ignite-17607
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit d84db92eba3939abccc9acdb6d47e9f77855cb63
Author: Igor Sapego <is...@apache.org>
AuthorDate: Tue Mar 7 16:29:47 2023 +0300
IGNITE-17607 Add execute test, fix bug
---
.../platforms/cpp/ignite/client/detail/ignite_client_impl.h | 1 +
modules/platforms/cpp/tests/client-test/compute_test.cpp | 12 ++++++++++--
.../platforms/cpp/tests/client-test/ignite_runner_suite.h | 8 ++++++++
3 files changed, 19 insertions(+), 2 deletions(-)
diff --git a/modules/platforms/cpp/ignite/client/detail/ignite_client_impl.h b/modules/platforms/cpp/ignite/client/detail/ignite_client_impl.h
index 8e9123332e..55274053f5 100644
--- a/modules/platforms/cpp/ignite/client/detail/ignite_client_impl.h
+++ b/modules/platforms/cpp/ignite/client/detail/ignite_client_impl.h
@@ -53,6 +53,7 @@ public:
, m_connection(cluster_connection::create(m_configuration))
, m_tables(std::make_shared<tables_impl>(m_connection))
, m_sql(std::make_shared<sql_impl>(m_connection))
+ , m_compute(std::make_shared<compute_impl>(m_connection))
, m_transactions(std::make_shared<transactions_impl>(m_connection)) {}
/**
diff --git a/modules/platforms/cpp/tests/client-test/compute_test.cpp b/modules/platforms/cpp/tests/client-test/compute_test.cpp
index 10b6e3aa1b..a90b4eac69 100644
--- a/modules/platforms/cpp/tests/client-test/compute_test.cpp
+++ b/modules/platforms/cpp/tests/client-test/compute_test.cpp
@@ -54,8 +54,7 @@ TEST_F(compute_test, get_cluster_nodes) {
return n1.get_name() < n2.get_name();
});
- EXPECT_FALSE(cluster_nodes.empty());
- EXPECT_EQ(2, cluster_nodes.size());
+ ASSERT_EQ(2, cluster_nodes.size());
EXPECT_FALSE(cluster_nodes[0].get_id().empty());
EXPECT_FALSE(cluster_nodes[1].get_id().empty());
@@ -68,3 +67,12 @@ TEST_F(compute_test, get_cluster_nodes) {
EXPECT_EQ(cluster_nodes[0].get_address().host, cluster_nodes[1].get_address().host);
}
+
+TEST_F(compute_test, execute_on_random_node) {
+ auto cluster_nodes = m_client.get_cluster_nodes();
+
+ auto result = m_client.get_compute().execute(cluster_nodes, NODE_NAME_JOB, {});
+
+ ASSERT_TRUE(result.has_value());
+ EXPECT_THAT(result.value().get<std::string>(), ::testing::StartsWith(PLATFORM_TEST_NODE_RUNNER));
+}
diff --git a/modules/platforms/cpp/tests/client-test/ignite_runner_suite.h b/modules/platforms/cpp/tests/client-test/ignite_runner_suite.h
index eba5d5e926..0766823eed 100644
--- a/modules/platforms/cpp/tests/client-test/ignite_runner_suite.h
+++ b/modules/platforms/cpp/tests/client-test/ignite_runner_suite.h
@@ -43,6 +43,14 @@ public:
static constexpr std::string_view TABLE_1 = "tbl1"sv;
static constexpr std::string_view TABLE_NAME_ALL_COLUMNS = "tbl_all_columns"sv;
+ inline static const std::string PLATFORM_TEST_NODE_RUNNER =
+ "org.apache.ignite.internal.runner.app.PlatformTestNodeRunner";
+
+ inline static const std::string IT_THIN_CLIENT_COMPUTE_TEST =
+ "org.apache.ignite.internal.runner.app.client.ItThinClientComputeTest";
+
+ inline static const std::string NODE_NAME_JOB = IT_THIN_CLIENT_COMPUTE_TEST + "$NodeNameJob";
+
static constexpr const char *KEY_COLUMN = "key";
static constexpr const char *VAL_COLUMN = "val";
[ignite-3] 10/19: IGNITE-17607 Add nullptr to primitive
Posted by is...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
isapego pushed a commit to branch ignite-17607
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 3d50b796c363ceb3f155a784b54951b7d4dfd24e
Author: Igor Sapego <is...@apache.org>
AuthorDate: Wed Mar 15 01:02:23 2023 +0300
IGNITE-17607 Add nullptr to primitive
---
.../platforms/cpp/ignite/client/detail/utils.cpp | 14 +++++++++++
modules/platforms/cpp/ignite/client/primitive.h | 29 ++++++++++++++++++++--
.../platforms/cpp/ignite/client/primitive_test.cpp | 17 +++++++++++--
3 files changed, 56 insertions(+), 4 deletions(-)
diff --git a/modules/platforms/cpp/ignite/client/detail/utils.cpp b/modules/platforms/cpp/ignite/client/detail/utils.cpp
index 42a68ba7d9..5a21f1c37f 100644
--- a/modules/platforms/cpp/ignite/client/detail/utils.cpp
+++ b/modules/platforms/cpp/ignite/client/detail/utils.cpp
@@ -47,6 +47,13 @@ void append_type_and_scale(binary_tuple_builder &builder, ignite_type typ, std::
}
void claim_primitive_with_type(binary_tuple_builder &builder, const primitive &value) {
+ if (value.is_null()) {
+ builder.claim(std::nullopt); // Type.
+ builder.claim(std::nullopt); // Scale.
+ builder.claim(std::nullopt); // Value.
+ return;
+ }
+
switch (value.get_type()) {
case column_type::BOOLEAN: {
claim_type_and_scale(builder, ignite_type::INT8);
@@ -115,6 +122,13 @@ void claim_primitive_with_type(binary_tuple_builder &builder, const primitive &v
}
void append_primitive_with_type(binary_tuple_builder &builder, const primitive &value) {
+ if (value.is_null()) {
+ builder.append(std::nullopt); // Type.
+ builder.append(std::nullopt); // Scale.
+ builder.append(std::nullopt); // Value.
+ return;
+ }
+
switch (value.get_type()) {
case column_type::BOOLEAN: {
append_type_and_scale(builder, ignite_type::INT8);
diff --git a/modules/platforms/cpp/ignite/client/primitive.h b/modules/platforms/cpp/ignite/client/primitive.h
index 667fbb233f..a4a9f3b9b7 100644
--- a/modules/platforms/cpp/ignite/client/primitive.h
+++ b/modules/platforms/cpp/ignite/client/primitive.h
@@ -29,6 +29,7 @@
#include "ignite/common/uuid.h"
#include <cstdint>
+#include <optional>
#include <type_traits>
#include <variant>
#include <vector>
@@ -43,6 +44,16 @@ public:
// Default
primitive() = default;
+ /**
+ * Null constructor.
+ */
+ primitive(std::nullptr_t) {} // NOLINT(google-explicit-constructor)
+
+ /**
+ * Null option constructor.
+ */
+ primitive(std::nullopt_t) {} // NOLINT(google-explicit-constructor)
+
/**
* Constructor for boolean value.
*
@@ -237,12 +248,25 @@ public:
}
}
+ /**
+ * Check whether element is null.
+ *
+ * @return Value indicating whether element is null.
+ */
+ [[nodiscard]] bool is_null() const noexcept {
+ return m_value.index() == 0;
+ }
+
/**
* Get primitive type.
*
* @return Primitive type.
*/
- [[nodiscard]] column_type get_type() const { return static_cast<column_type>(m_value.index()); }
+ [[nodiscard]] column_type get_type() const noexcept {
+ if (is_null())
+ return column_type::UNDEFINED;
+ return static_cast<column_type>(m_value.index() - 1);
+ }
/**
* @brief Comparison operator.
@@ -271,7 +295,8 @@ private:
typedef void *unsupported_type;
/** Value type. */
- typedef std::variant<bool, // Bool = 0
+ typedef std::variant<std::nullptr_t,
+ bool, // Bool = 0
std::int8_t, // Int8 = 1
std::int16_t, // Int16 = 2
std::int32_t, // Int32 = 3
diff --git a/modules/platforms/cpp/ignite/client/primitive_test.cpp b/modules/platforms/cpp/ignite/client/primitive_test.cpp
index 46a549f832..b71b3033e3 100644
--- a/modules/platforms/cpp/ignite/client/primitive_test.cpp
+++ b/modules/platforms/cpp/ignite/client/primitive_test.cpp
@@ -23,11 +23,12 @@ using namespace ignite;
template<typename T>
void check_primitive_type(column_type expected) {
- primitive val_bool(T{});
- EXPECT_EQ(val_bool.get_type(), expected);
+ primitive val(T{});
+ EXPECT_EQ(val.get_type(), expected);
}
TEST(primitive, get_column_type) {
+ check_primitive_type<nullptr_t>(column_type::UNDEFINED);
check_primitive_type<bool>(column_type::BOOLEAN);
check_primitive_type<int8_t>(column_type::INT8);
check_primitive_type<int16_t>(column_type::INT16);
@@ -46,3 +47,15 @@ TEST(primitive, get_column_type) {
check_primitive_type<std::vector<std::byte>>(column_type::BYTE_ARRAY);
check_primitive_type<big_integer>(column_type::NUMBER);
}
+
+TEST(primitive, null_value_by_nullptr) {
+ primitive val(nullptr);
+ EXPECT_EQ(val.get_type(), column_type::UNDEFINED);
+ EXPECT_TRUE(val.is_null());
+}
+
+TEST(primitive, null_value_by_nullopt) {
+ primitive val(std::nullopt);
+ EXPECT_EQ(val.get_type(), column_type::UNDEFINED);
+ EXPECT_TRUE(val.is_null());
+}
[ignite-3] 15/19: IGNITE-17607 Execute colocated stub
Posted by is...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
isapego pushed a commit to branch ignite-17607
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 0cea5d2779158893162b2a6bbb3b1d6a3ed4f576
Author: Igor Sapego <is...@apache.org>
AuthorDate: Mon Mar 20 10:38:35 2023 +0300
IGNITE-17607 Execute colocated stub
---
.../cpp/ignite/client/compute/compute.cpp | 3 +-
.../platforms/cpp/ignite/client/compute/compute.h | 2 +-
.../ignite/client/detail/compute/compute_impl.cpp | 48 +++++
.../ignite/client/detail/compute/compute_impl.h | 23 ++-
.../cpp/ignite/client/detail/ignite_client_impl.h | 2 +-
.../cpp/ignite/client/detail/table/table_impl.cpp | 215 +--------------------
.../cpp/ignite/client/detail/table/table_impl.h | 21 ++
.../platforms/cpp/ignite/client/detail/utils.cpp | 194 +++++++++++++++++++
modules/platforms/cpp/ignite/client/detail/utils.h | 24 +++
modules/platforms/cpp/ignite/client/table/table.h | 1 +
10 files changed, 318 insertions(+), 215 deletions(-)
diff --git a/modules/platforms/cpp/ignite/client/compute/compute.cpp b/modules/platforms/cpp/ignite/client/compute/compute.cpp
index 77e803d306..7f710bffb6 100644
--- a/modules/platforms/cpp/ignite/client/compute/compute.cpp
+++ b/modules/platforms/cpp/ignite/client/compute/compute.cpp
@@ -83,9 +83,10 @@ void compute::execute_colocated_async(const std::string &table_name, const ignit
std::string_view job_class_name, const std::vector<primitive> &args,
ignite_callback<std::optional<primitive>> callback) {
detail::arg_check::container_non_empty(table_name, "Table name");
+ detail::arg_check::tuple_non_empty(key, "Key tuple");
detail::arg_check::container_non_empty(job_class_name, "Job class name");
- // TODO: Implement me.
+ m_impl->execute_colocated_async(table_name, key, job_class_name, args, std::move(callback));
}
} // namespace ignite
diff --git a/modules/platforms/cpp/ignite/client/compute/compute.h b/modules/platforms/cpp/ignite/client/compute/compute.h
index ff40355d8a..130ebb7abf 100644
--- a/modules/platforms/cpp/ignite/client/compute/compute.h
+++ b/modules/platforms/cpp/ignite/client/compute/compute.h
@@ -101,7 +101,7 @@ public:
}
/**
- * Executes a compute job represented by the given class on one of the specified nodes asynchronously.
+ * Asynchronously executes a job represented by the given class on one node where the given key is located.
*
* @param tableName Name of the table to be used with @c key to determine target node.
* @param key Table key to be used to determine the target node for job execution.
diff --git a/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.cpp b/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.cpp
index 4cd9a00335..ef845a3ff6 100644
--- a/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.cpp
+++ b/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.cpp
@@ -66,4 +66,52 @@ void compute_impl::execute_on_one_node(cluster_node node, std::string_view job_c
client_operation::COMPUTE_EXECUTE, writer_func, std::move(reader_func), std::move(callback));
}
+void compute_impl::execute_colocated_async(const std::string &table_name, const ignite_tuple &key,
+ std::string_view job, const std::vector<primitive> &args,
+ ignite_callback<std::optional<primitive>> callback) {
+ m_tables->get_table_async(table_name,
+ [table_name, callback = std::move(callback), key, job = std::string(job), args, conn = m_connection]
+ (auto &&res) mutable {
+ if (res.has_error()) {
+ callback({std::move(res.error())});
+ return;
+ }
+ auto &table_opt = res.value();
+ if (!table_opt) {
+ callback({ignite_error("Table does not exist: '" + table_name + "'")});
+ return;
+ }
+
+ auto table = table_impl::from_facade(*table_opt);
+ table->template with_latest_schema_async<std::optional<primitive>>(std::move(callback),
+ [table, key = std::move(key), job = std::move(job), args = std::move(args), conn] // NOLINT(performance-move-const-arg)
+ (const schema& sch, auto callback) mutable {
+ auto writer_func = [&key, &sch, &table, &job](protocol::writer &writer) {
+ writer.write(table->get_id());
+ writer.write(sch.version);
+ write_tuple(writer, sch, key, true);
+ writer.write(job);
+ // TODO: write arguments.
+ };
+
+ auto reader_func = [](protocol::reader &reader) -> std::optional<primitive> {
+ if (reader.try_read_nil())
+ return std::nullopt;
+
+ // TODO: Tuple to object.
+ auto tuple_data = reader.read_binary();
+ ignite_tuple res(3);
+ binary_tuple_parser parser(3, tuple_data);
+ auto typ = column_type(binary_tuple_parser::get_int32(*parser.get_next()));
+ auto scale = binary_tuple_parser::get_int32(*parser.get_next());
+
+ return read_next_column(parser, typ, scale);
+ };
+
+ conn->perform_request<std::optional<primitive>>(
+ client_operation::COMPUTE_EXECUTE_COLOCATED, writer_func, std::move(reader_func), std::move(callback));
+ });
+ });
+}
+
} // namespace ignite::detail
diff --git a/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.h b/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.h
index 7328d402f3..222c22481c 100644
--- a/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.h
+++ b/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.h
@@ -18,8 +18,10 @@
#pragma once
#include "ignite/client/detail/cluster_connection.h"
+#include "ignite/client/detail/table/tables_impl.h"
#include "ignite/client/network/cluster_node.h"
#include "ignite/client/primitive.h"
+#include "ignite/client/table/ignite_tuple.h"
#include "ignite/common/ignite_result.h"
#include <memory>
@@ -39,8 +41,9 @@ public:
*
* @param connection Connection.
*/
- explicit compute_impl(std::shared_ptr<cluster_connection> connection)
- : m_connection(std::move(connection)) {}
+ explicit compute_impl(std::shared_ptr<cluster_connection> connection, std::shared_ptr<tables_impl> tables)
+ : m_connection(std::move(connection))
+ , m_tables(std::move(tables)) {}
/**
* Executes a compute job represented by the given class on the specified node asynchronously.
@@ -53,9 +56,25 @@ public:
void execute_on_one_node(cluster_node node, std::string_view job_class_name, const std::vector<primitive>& args,
ignite_callback<std::optional<primitive>> callback);
+ /**
+ * Asynchronously executes a job represented by the given class on one node where the given key is located.
+ *
+ * @param tableName Name of the table to be used with @c key to determine target node.
+ * @param key Table key to be used to determine the target node for job execution.
+ * @param job_class_name Java class name of the job to execute.
+ * @param args Job arguments.
+ * @param callback A callback called on operation completion with job execution result.
+ */
+ void execute_colocated_async(const std::string &table_name, const ignite_tuple& key,
+ std::string_view job_class_name, const std::vector<primitive>& args,
+ ignite_callback<std::optional<primitive>> callback);
+
private:
/** Cluster connection. */
std::shared_ptr<cluster_connection> m_connection;
+
+ /** Tables. */
+ std::shared_ptr<tables_impl> m_tables;
};
} // namespace ignite::detail
diff --git a/modules/platforms/cpp/ignite/client/detail/ignite_client_impl.h b/modules/platforms/cpp/ignite/client/detail/ignite_client_impl.h
index 55274053f5..6357252e50 100644
--- a/modules/platforms/cpp/ignite/client/detail/ignite_client_impl.h
+++ b/modules/platforms/cpp/ignite/client/detail/ignite_client_impl.h
@@ -53,7 +53,7 @@ public:
, m_connection(cluster_connection::create(m_configuration))
, m_tables(std::make_shared<tables_impl>(m_connection))
, m_sql(std::make_shared<sql_impl>(m_connection))
- , m_compute(std::make_shared<compute_impl>(m_connection))
+ , m_compute(std::make_shared<compute_impl>(m_connection, m_tables))
, m_transactions(std::make_shared<transactions_impl>(m_connection)) {}
/**
diff --git a/modules/platforms/cpp/ignite/client/detail/table/table_impl.cpp b/modules/platforms/cpp/ignite/client/detail/table/table_impl.cpp
index 221aa2238d..eb6b3a4db9 100644
--- a/modules/platforms/cpp/ignite/client/detail/table/table_impl.cpp
+++ b/modules/platforms/cpp/ignite/client/detail/table/table_impl.cpp
@@ -18,226 +18,17 @@
#include "ignite/client/detail/table/table_impl.h"
#include "ignite/client/detail/transaction/transaction_impl.h"
#include "ignite/client/detail/utils.h"
+#include "ignite/client/table/table.h"
#include "ignite/common/bits.h"
#include "ignite/common/ignite_error.h"
#include "ignite/protocol/bitset_span.h"
#include "ignite/protocol/reader.h"
#include "ignite/protocol/writer.h"
-#include "ignite/schema/binary_tuple_builder.h"
#include "ignite/schema/binary_tuple_parser.h"
namespace ignite::detail {
-/**
- * Claim space for the column.
- *
- * @param builder Binary tuple builder.
- * @param typ Column type.
- * @param value Value.
- * @param scale Column scale.
- */
-void claim_column(binary_tuple_builder &builder, ignite_type typ, const primitive &value, std::int32_t scale) {
- switch (typ) {
- case ignite_type::INT8:
- builder.claim_int8(value.get<std::int8_t>());
- break;
- case ignite_type::INT16:
- builder.claim_int16(value.get<std::int16_t>());
- break;
- case ignite_type::INT32:
- builder.claim_int32(value.get<std::int32_t>());
- break;
- case ignite_type::INT64:
- builder.claim_int64(value.get<std::int64_t>());
- break;
- case ignite_type::FLOAT:
- builder.claim_float(value.get<float>());
- break;
- case ignite_type::DOUBLE:
- builder.claim_double(value.get<double>());
- break;
- case ignite_type::UUID:
- builder.claim_uuid(value.get<uuid>());
- break;
- case ignite_type::STRING:
- builder.claim_string(value.get<std::string>());
- break;
- case ignite_type::BINARY:
- builder.claim_bytes(value.get<std::vector<std::byte>>());
- break;
- case ignite_type::DECIMAL: {
- big_decimal to_write;
- value.get<big_decimal>().set_scale(scale, to_write);
- builder.claim_number(to_write);
- break;
- }
- case ignite_type::NUMBER:
- builder.claim_number(value.get<big_integer>());
- break;
- case ignite_type::DATE:
- builder.claim_date(value.get<ignite_date>());
- break;
- case ignite_type::TIME:
- builder.claim_time(value.get<ignite_time>());
- break;
- case ignite_type::DATETIME:
- builder.claim_date_time(value.get<ignite_date_time>());
- break;
- case ignite_type::TIMESTAMP:
- builder.claim_timestamp(value.get<ignite_timestamp>());
- break;
- case ignite_type::BITMASK:
- builder.claim_bytes(value.get<bit_array>().get_raw());
- break;
- default:
- throw ignite_error("Type with id " + std::to_string(int(typ)) + " is not yet supported");
- }
-}
-
-/**
- * Append column value to binary tuple.
- *
- * @param builder Binary tuple builder.
- * @param typ Column type.
- * @param value Value.
- * @param scale Column scale.
- */
-void append_column(binary_tuple_builder &builder, ignite_type typ, const primitive &value, std::int32_t scale) {
- switch (typ) {
- case ignite_type::INT8:
- builder.append_int8(value.get<std::int8_t>());
- break;
- case ignite_type::INT16:
- builder.append_int16(value.get<std::int16_t>());
- break;
- case ignite_type::INT32:
- builder.append_int32(value.get<std::int32_t>());
- break;
- case ignite_type::INT64:
- builder.append_int64(value.get<std::int64_t>());
- break;
- case ignite_type::FLOAT:
- builder.append_float(value.get<float>());
- break;
- case ignite_type::DOUBLE:
- builder.append_double(value.get<double>());
- break;
- case ignite_type::UUID:
- builder.append_uuid(value.get<uuid>());
- break;
- case ignite_type::STRING:
- builder.append_string(value.get<std::string>());
- break;
- case ignite_type::BINARY:
- builder.append_bytes(value.get<std::vector<std::byte>>());
- break;
- case ignite_type::DECIMAL: {
- big_decimal to_write;
- value.get<big_decimal>().set_scale(scale, to_write);
- builder.append_number(to_write);
- break;
- }
- case ignite_type::NUMBER:
- builder.append_number(value.get<big_integer>());
- break;
- case ignite_type::DATE:
- builder.append_date(value.get<ignite_date>());
- break;
- case ignite_type::TIME:
- builder.append_time(value.get<ignite_time>());
- break;
- case ignite_type::DATETIME:
- builder.append_date_time(value.get<ignite_date_time>());
- break;
- case ignite_type::TIMESTAMP:
- builder.append_timestamp(value.get<ignite_timestamp>());
- break;
- case ignite_type::BITMASK:
- builder.append_bytes(value.get<bit_array>().get_raw());
- break;
- default:
- throw ignite_error("Type with id " + std::to_string(int(typ)) + " is not yet supported");
- }
-}
-
-/**
- * Serialize tuple using table schema.
- *
- * @param sch Schema.
- * @param tuple Tuple.
- * @param key_only Should only key fields be serialized.
- * @param no_value No value bitset.
- * @return Serialized binary tuple.
- */
-std::vector<std::byte> pack_tuple(
- const schema &sch, const ignite_tuple &tuple, bool key_only, protocol::bitset_span &no_value) {
- auto count = std::int32_t(key_only ? sch.key_column_count : sch.columns.size());
- binary_tuple_builder builder{count};
-
- builder.start();
-
- for (std::int32_t i = 0; i < count; ++i) {
- const auto &col = sch.columns[i];
- auto col_idx = tuple.column_ordinal(col.name);
-
- if (col_idx >= 0)
- claim_column(builder, col.type, tuple.get(col_idx), col.scale);
- else
- builder.claim(std::nullopt);
- }
-
- builder.layout();
- for (std::int32_t i = 0; i < count; ++i) {
- const auto &col = sch.columns[i];
- auto col_idx = tuple.column_ordinal(col.name);
-
- if (col_idx >= 0)
- append_column(builder, col.type, tuple.get(col_idx), col.scale);
- else {
- builder.append(std::nullopt);
- no_value.set(std::size_t(i));
- }
- }
-
- return builder.build();
-}
-
-/**
- * Write tuple using table schema and writer.
- *
- * @param writer Writer.
- * @param sch Schema.
- * @param tuple Tuple.
- * @param key_only Should only key fields be written or not.
- */
-void write_tuple(protocol::writer &writer, const schema &sch, const ignite_tuple &tuple, bool key_only) {
- const std::size_t count = key_only ? sch.key_column_count : sch.columns.size();
- const std::size_t bytes_num = bytes_for_bits(count);
-
- auto no_value_bytes = reinterpret_cast<std::byte *>(alloca(bytes_num));
- protocol::bitset_span no_value(no_value_bytes, bytes_num);
-
- auto tuple_data = pack_tuple(sch, tuple, key_only, no_value);
-
- writer.write_bitset(no_value.data());
- writer.write_binary(tuple_data);
-}
-
-/**
- * Write tuples using table schema and writer.
- *
- * @param writer Writer.
- * @param sch Schema.
- * @param tuples Tuples.
- * @param key_only Should only key fields be written or not.
- */
-void write_tuples(protocol::writer &writer, const schema &sch, const std::vector<ignite_tuple> &tuples, bool key_only) {
- writer.write(std::int32_t(tuples.size()));
- for (auto &tuple : tuples)
- write_tuple(writer, sch, tuple, key_only);
-}
-
/**
* Write table operation header.
*
@@ -696,4 +487,8 @@ void table_impl::remove_all_exact_async(
});
}
+std::shared_ptr<table_impl> table_impl::from_facade(table &tb) {
+ return tb.m_impl;
+}
+
} // namespace ignite::detail
diff --git a/modules/platforms/cpp/ignite/client/detail/table/table_impl.h b/modules/platforms/cpp/ignite/client/detail/table/table_impl.h
index ed69979038..5b88f3b29b 100644
--- a/modules/platforms/cpp/ignite/client/detail/table/table_impl.h
+++ b/modules/platforms/cpp/ignite/client/detail/table/table_impl.h
@@ -27,6 +27,10 @@
#include <mutex>
#include <unordered_map>
+namespace ignite {
+class table;
+}
+
namespace ignite::detail {
/**
@@ -284,6 +288,23 @@ public:
void remove_all_exact_async(
transaction *tx, std::vector<ignite_tuple> records, ignite_callback<std::vector<ignite_tuple>> callback);
+ /**
+ * Extract implementation from facade.
+ *
+ * @param tb Table.
+ * @return Implementation.
+ */
+ [[nodiscard]] static std::shared_ptr<table_impl> from_facade(table& tb);
+
+ /**
+ * Get table ID.
+ *
+ * @return ID.
+ */
+ [[nodiscard]] uuid get_id() const {
+ return m_id;
+ }
+
private:
/**
* Load latest schema from server asynchronously.
diff --git a/modules/platforms/cpp/ignite/client/detail/utils.cpp b/modules/platforms/cpp/ignite/client/detail/utils.cpp
index 60a5aa9e79..7e66c97ab0 100644
--- a/modules/platforms/cpp/ignite/client/detail/utils.cpp
+++ b/modules/platforms/cpp/ignite/client/detail/utils.cpp
@@ -16,12 +16,187 @@
*/
#include "ignite/client/detail/utils.h"
+#include "ignite/common/bits.h"
#include "ignite/common/uuid.h"
#include <string>
namespace ignite::detail {
+/**
+ * Claim space for the column.
+ *
+ * @param builder Binary tuple builder.
+ * @param typ Column type.
+ * @param value Value.
+ * @param scale Column scale.
+ */
+void claim_column(binary_tuple_builder &builder, ignite_type typ, const primitive &value, std::int32_t scale) {
+ switch (typ) {
+ case ignite_type::INT8:
+ builder.claim_int8(value.get<std::int8_t>());
+ break;
+ case ignite_type::INT16:
+ builder.claim_int16(value.get<std::int16_t>());
+ break;
+ case ignite_type::INT32:
+ builder.claim_int32(value.get<std::int32_t>());
+ break;
+ case ignite_type::INT64:
+ builder.claim_int64(value.get<std::int64_t>());
+ break;
+ case ignite_type::FLOAT:
+ builder.claim_float(value.get<float>());
+ break;
+ case ignite_type::DOUBLE:
+ builder.claim_double(value.get<double>());
+ break;
+ case ignite_type::UUID:
+ builder.claim_uuid(value.get<uuid>());
+ break;
+ case ignite_type::STRING:
+ builder.claim_string(value.get<std::string>());
+ break;
+ case ignite_type::BINARY:
+ builder.claim_bytes(value.get<std::vector<std::byte>>());
+ break;
+ case ignite_type::DECIMAL: {
+ big_decimal to_write;
+ value.get<big_decimal>().set_scale(scale, to_write);
+ builder.claim_number(to_write);
+ break;
+ }
+ case ignite_type::NUMBER:
+ builder.claim_number(value.get<big_integer>());
+ break;
+ case ignite_type::DATE:
+ builder.claim_date(value.get<ignite_date>());
+ break;
+ case ignite_type::TIME:
+ builder.claim_time(value.get<ignite_time>());
+ break;
+ case ignite_type::DATETIME:
+ builder.claim_date_time(value.get<ignite_date_time>());
+ break;
+ case ignite_type::TIMESTAMP:
+ builder.claim_timestamp(value.get<ignite_timestamp>());
+ break;
+ case ignite_type::BITMASK:
+ builder.claim_bytes(value.get<bit_array>().get_raw());
+ break;
+ default:
+ throw ignite_error("Type with id " + std::to_string(int(typ)) + " is not yet supported");
+ }
+}
+
+/**
+ * Append column value to binary tuple.
+ *
+ * @param builder Binary tuple builder.
+ * @param typ Column type.
+ * @param value Value.
+ * @param scale Column scale.
+ */
+void append_column(binary_tuple_builder &builder, ignite_type typ, const primitive &value, std::int32_t scale) {
+ switch (typ) {
+ case ignite_type::INT8:
+ builder.append_int8(value.get<std::int8_t>());
+ break;
+ case ignite_type::INT16:
+ builder.append_int16(value.get<std::int16_t>());
+ break;
+ case ignite_type::INT32:
+ builder.append_int32(value.get<std::int32_t>());
+ break;
+ case ignite_type::INT64:
+ builder.append_int64(value.get<std::int64_t>());
+ break;
+ case ignite_type::FLOAT:
+ builder.append_float(value.get<float>());
+ break;
+ case ignite_type::DOUBLE:
+ builder.append_double(value.get<double>());
+ break;
+ case ignite_type::UUID:
+ builder.append_uuid(value.get<uuid>());
+ break;
+ case ignite_type::STRING:
+ builder.append_string(value.get<std::string>());
+ break;
+ case ignite_type::BINARY:
+ builder.append_bytes(value.get<std::vector<std::byte>>());
+ break;
+ case ignite_type::DECIMAL: {
+ big_decimal to_write;
+ value.get<big_decimal>().set_scale(scale, to_write);
+ builder.append_number(to_write);
+ break;
+ }
+ case ignite_type::NUMBER:
+ builder.append_number(value.get<big_integer>());
+ break;
+ case ignite_type::DATE:
+ builder.append_date(value.get<ignite_date>());
+ break;
+ case ignite_type::TIME:
+ builder.append_time(value.get<ignite_time>());
+ break;
+ case ignite_type::DATETIME:
+ builder.append_date_time(value.get<ignite_date_time>());
+ break;
+ case ignite_type::TIMESTAMP:
+ builder.append_timestamp(value.get<ignite_timestamp>());
+ break;
+ case ignite_type::BITMASK:
+ builder.append_bytes(value.get<bit_array>().get_raw());
+ break;
+ default:
+ throw ignite_error("Type with id " + std::to_string(int(typ)) + " is not yet supported");
+ }
+}
+
+/**
+ * Serialize tuple using table schema.
+ *
+ * @param sch Schema.
+ * @param tuple Tuple.
+ * @param key_only Should only key fields be serialized.
+ * @param no_value No value bitset.
+ * @return Serialized binary tuple.
+ */
+std::vector<std::byte> pack_tuple(
+ const schema &sch, const ignite_tuple &tuple, bool key_only, protocol::bitset_span &no_value) {
+ auto count = std::int32_t(key_only ? sch.key_column_count : sch.columns.size());
+ binary_tuple_builder builder{count};
+
+ builder.start();
+
+ for (std::int32_t i = 0; i < count; ++i) {
+ const auto &col = sch.columns[i];
+ auto col_idx = tuple.column_ordinal(col.name);
+
+ if (col_idx >= 0)
+ claim_column(builder, col.type, tuple.get(col_idx), col.scale);
+ else
+ builder.claim(std::nullopt);
+ }
+
+ builder.layout();
+ for (std::int32_t i = 0; i < count; ++i) {
+ const auto &col = sch.columns[i];
+ auto col_idx = tuple.column_ordinal(col.name);
+
+ if (col_idx >= 0)
+ append_column(builder, col.type, tuple.get(col_idx), col.scale);
+ else {
+ builder.append(std::nullopt);
+ no_value.set(std::size_t(i));
+ }
+ }
+
+ return builder.build();
+}
+
/**
* Claim type and scale header for a value written in binary tuple.
*
@@ -362,4 +537,23 @@ ignite_tuple concat(const ignite_tuple &left, const ignite_tuple &right) {
return res;
}
+void write_tuple(protocol::writer &writer, const schema &sch, const ignite_tuple &tuple, bool key_only) {
+ const std::size_t count = key_only ? sch.key_column_count : sch.columns.size();
+ const std::size_t bytes_num = bytes_for_bits(count);
+
+ auto no_value_bytes = reinterpret_cast<std::byte *>(alloca(bytes_num));
+ protocol::bitset_span no_value(no_value_bytes, bytes_num);
+
+ auto tuple_data = pack_tuple(sch, tuple, key_only, no_value);
+
+ writer.write_bitset(no_value.data());
+ writer.write_binary(tuple_data);
+}
+
+void write_tuples(protocol::writer &writer, const schema &sch, const std::vector<ignite_tuple> &tuples, bool key_only) {
+ writer.write(std::int32_t(tuples.size()));
+ for (auto &tuple : tuples)
+ write_tuple(writer, sch, tuple, key_only);
+}
+
} // namespace ignite::detail
diff --git a/modules/platforms/cpp/ignite/client/detail/utils.h b/modules/platforms/cpp/ignite/client/detail/utils.h
index fc248954a1..62f640d0e0 100644
--- a/modules/platforms/cpp/ignite/client/detail/utils.h
+++ b/modules/platforms/cpp/ignite/client/detail/utils.h
@@ -17,9 +17,12 @@
#pragma once
+#include "ignite/client/detail/table/schema.h"
#include "ignite/client/primitive.h"
#include "ignite/client/table/ignite_tuple.h"
#include "ignite/client/transaction/transaction.h"
+
+#include "ignite/protocol/writer.h"
#include "ignite/schema/binary_tuple_builder.h"
#include "ignite/schema/binary_tuple_parser.h"
#include "ignite/schema/ignite_type.h"
@@ -71,4 +74,25 @@ void append_primitive_with_type(binary_tuple_builder &builder, const primitive &
*/
[[nodiscard]] ignite_tuple concat(const ignite_tuple &left, const ignite_tuple &right);
+
+/**
+ * Write tuple using table schema and writer.
+ *
+ * @param writer Writer.
+ * @param sch Schema.
+ * @param tuple Tuple.
+ * @param key_only Should only key fields be written or not.
+ */
+void write_tuple(protocol::writer &writer, const schema &sch, const ignite_tuple &tuple, bool key_only);
+
+/**
+ * Write tuples using table schema and writer.
+ *
+ * @param writer Writer.
+ * @param sch Schema.
+ * @param tuples Tuples.
+ * @param key_only Should only key fields be written or not.
+ */
+void write_tuples(protocol::writer &writer, const schema &sch, const std::vector<ignite_tuple> &tuples, bool key_only);
+
} // namespace ignite::detail
diff --git a/modules/platforms/cpp/ignite/client/table/table.h b/modules/platforms/cpp/ignite/client/table/table.h
index 4d71cb94f6..d946d93fb0 100644
--- a/modules/platforms/cpp/ignite/client/table/table.h
+++ b/modules/platforms/cpp/ignite/client/table/table.h
@@ -38,6 +38,7 @@ class tables_impl;
* Table view.
*/
class table {
+ friend class detail::table_impl;
friend class detail::tables_impl;
public:
[ignite-3] 17/19: IGNITE-17607 Add test
Posted by is...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
isapego pushed a commit to branch ignite-17607
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 00dc41547f43cd25db2c5f57c529530359db3819
Author: Igor Sapego <is...@apache.org>
AuthorDate: Mon Mar 20 17:58:27 2023 +0300
IGNITE-17607 Add test
---
.../platforms/cpp/ignite/client/compute/compute.cpp | 2 +-
modules/platforms/cpp/ignite/client/compute/compute.h | 18 +++++++++++++++++-
.../cpp/ignite/client/detail/compute/compute_impl.cpp | 9 ++++-----
.../cpp/ignite/client/detail/compute/compute_impl.h | 2 +-
.../platforms/cpp/tests/client-test/compute_test.cpp | 18 ++++++++++++++++++
5 files changed, 41 insertions(+), 8 deletions(-)
diff --git a/modules/platforms/cpp/ignite/client/compute/compute.cpp b/modules/platforms/cpp/ignite/client/compute/compute.cpp
index 7f710bffb6..3bf267bbba 100644
--- a/modules/platforms/cpp/ignite/client/compute/compute.cpp
+++ b/modules/platforms/cpp/ignite/client/compute/compute.cpp
@@ -79,7 +79,7 @@ void compute::broadcast_async(const std::set<cluster_node>& nodes, std::string_v
}
}
-void compute::execute_colocated_async(const std::string &table_name, const ignite_tuple &key,
+void compute::execute_colocated_async(std::string_view table_name, const ignite_tuple &key,
std::string_view job_class_name, const std::vector<primitive> &args,
ignite_callback<std::optional<primitive>> callback) {
detail::arg_check::container_non_empty(table_name, "Table name");
diff --git a/modules/platforms/cpp/ignite/client/compute/compute.h b/modules/platforms/cpp/ignite/client/compute/compute.h
index 130ebb7abf..1ef28478b0 100644
--- a/modules/platforms/cpp/ignite/client/compute/compute.h
+++ b/modules/platforms/cpp/ignite/client/compute/compute.h
@@ -109,10 +109,26 @@ public:
* @param args Job arguments.
* @param callback A callback called on operation completion with job execution result.
*/
- IGNITE_API void execute_colocated_async(const std::string &table_name, const ignite_tuple& key,
+ IGNITE_API void execute_colocated_async(std::string_view table_name, const ignite_tuple& key,
std::string_view job_class_name, const std::vector<primitive>& args,
ignite_callback<std::optional<primitive>> callback);
+ /**
+ * Synchronously executes a job represented by the given class on one node where the given key is located.
+ *
+ * @param tableName Name of the table to be used with @c key to determine target node.
+ * @param key Table key to be used to determine the target node for job execution.
+ * @param job_class_name Java class name of the job to execute.
+ * @param args Job arguments.
+ * @return Job execution result.
+ */
+ IGNITE_API std::optional<primitive> execute_colocated(std::string_view table_name, const ignite_tuple& key,
+ std::string_view job_class_name, const std::vector<primitive>& args) {
+ return sync<std::optional<primitive>>([this, &table_name, &key, job_class_name, &args](auto callback) mutable {
+ execute_colocated_async(table_name, key, job_class_name, args, std::move(callback));
+ });
+ }
+
private:
/**
* Constructor
diff --git a/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.cpp b/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.cpp
index b6ebd4dc0b..eec73996a8 100644
--- a/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.cpp
+++ b/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.cpp
@@ -83,11 +83,10 @@ void compute_impl::execute_on_one_node(cluster_node node, std::string_view job_c
client_operation::COMPUTE_EXECUTE, writer_func, std::move(reader_func), std::move(callback));
}
-void compute_impl::execute_colocated_async(const std::string &table_name, const ignite_tuple &key,
- std::string_view job, const std::vector<primitive> &args,
- ignite_callback<std::optional<primitive>> callback) {
- m_tables->get_table_async(table_name,
- [table_name, callback = std::move(callback), key, job = std::string(job), args, conn = m_connection]
+void compute_impl::execute_colocated_async(std::string_view table_name, const ignite_tuple &key, std::string_view job,
+ const std::vector<primitive> &args, ignite_callback<std::optional<primitive>> callback) {
+ m_tables->get_table_async(table_name, [table_name = std::string(table_name), callback = std::move(callback),
+ key, job = std::string(job), args, conn = m_connection]
(auto &&res) mutable {
if (res.has_error()) {
callback({std::move(res.error())});
diff --git a/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.h b/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.h
index 222c22481c..f7a29200e9 100644
--- a/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.h
+++ b/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.h
@@ -65,7 +65,7 @@ public:
* @param args Job arguments.
* @param callback A callback called on operation completion with job execution result.
*/
- void execute_colocated_async(const std::string &table_name, const ignite_tuple& key,
+ void execute_colocated_async(std::string_view table_name, const ignite_tuple& key,
std::string_view job_class_name, const std::vector<primitive>& args,
ignite_callback<std::optional<primitive>> callback);
diff --git a/modules/platforms/cpp/tests/client-test/compute_test.cpp b/modules/platforms/cpp/tests/client-test/compute_test.cpp
index 02004ee99d..318549b636 100644
--- a/modules/platforms/cpp/tests/client-test/compute_test.cpp
+++ b/modules/platforms/cpp/tests/client-test/compute_test.cpp
@@ -240,3 +240,21 @@ TEST_F(compute_test, all_arg_types) {
check_argument<uuid>({0x123e4567e89b12d3, 0x7456426614174000}, "123e4567-e89b-12d3-7456-426614174000");
}
+TEST_F(compute_test, execute_colocated) {
+ std::map<std::int32_t, std::string> nodes_for_values = {
+ {1, ""},
+ {2, "_2"},
+ {3, ""},
+ {5, "_2"}
+ };
+
+ for (const auto &var : nodes_for_values) {
+ SCOPED_TRACE("key=" + std::to_string(var.first) + ", node=" + var.second);
+ auto key = get_tuple(var.first);
+
+ auto resNodeName = m_client.get_compute().execute_colocated(TABLE_1, key, NODE_NAME_JOB, {});
+ auto expectedNodeName = PLATFORM_TEST_NODE_RUNNER + var.second;
+
+ EXPECT_EQ(expectedNodeName, resNodeName);
+ }
+}
[ignite-3] 09/19: IGNITE-17607 Add test
Posted by is...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
isapego pushed a commit to branch ignite-17607
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit f140d16b30891c2460b7ce7beb8afed7ab4641a7
Author: Igor Sapego <is...@apache.org>
AuthorDate: Tue Mar 7 17:56:08 2023 +0300
IGNITE-17607 Add test
---
.../cpp/tests/client-test/compute_test.cpp | 32 ++++++++++++++++++++++
1 file changed, 32 insertions(+)
diff --git a/modules/platforms/cpp/tests/client-test/compute_test.cpp b/modules/platforms/cpp/tests/client-test/compute_test.cpp
index 7cbd15cc85..94f274f41e 100644
--- a/modules/platforms/cpp/tests/client-test/compute_test.cpp
+++ b/modules/platforms/cpp/tests/client-test/compute_test.cpp
@@ -56,6 +56,15 @@ protected:
return nodes[id];
}
+ /**
+ * Get nodes as set.
+ * @return Nodes in set.
+ */
+ std::set<cluster_node> get_node_set() {
+ auto nodes = m_client.get_cluster_nodes();
+ return {nodes.begin(), nodes.end()};
+ }
+
/** Ignite client. */
ignite_client m_client;
};
@@ -101,4 +110,27 @@ TEST_F(compute_test, execute_on_specific_node) {
EXPECT_EQ(res2.value().get<std::string>(), PLATFORM_TEST_NODE_RUNNER + "_2:_22");
}
+TEST_F(compute_test, execute_broadcast_one_node) {
+ auto res = m_client.get_compute().broadcast({get_node(1)}, NODE_NAME_JOB, {"42"});
+
+ ASSERT_EQ(res.size(), 1);
+
+ EXPECT_EQ(res.begin()->first, get_node(1));
+
+ ASSERT_TRUE(res.begin()->second.has_value());
+ EXPECT_EQ(res.begin()->second.value(), PLATFORM_TEST_NODE_RUNNER + "_242");
+}
+
+TEST_F(compute_test, execute_broadcast_all_nodes) {
+ auto res = m_client.get_compute().broadcast(get_node_set(), NODE_NAME_JOB, {"42"});
+
+ ASSERT_EQ(res.size(), 2);
+
+ EXPECT_EQ(res[get_node(0)].value(), get_node(0).get_name() + "42");
+ EXPECT_EQ(res[get_node(1)].value(), get_node(1).get_name() + "42");
+}
+
+
+
+
[ignite-3] 11/19: IGNITE-17607 Add args test
Posted by is...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
isapego pushed a commit to branch ignite-17607
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 9432ffe877f50d3ed5f548497082c95ca2935059
Author: Igor Sapego <is...@apache.org>
AuthorDate: Wed Mar 15 01:02:35 2023 +0300
IGNITE-17607 Add args test
---
modules/platforms/cpp/tests/client-test/compute_test.cpp | 9 +++++++++
modules/platforms/cpp/tests/client-test/ignite_runner_suite.h | 1 +
2 files changed, 10 insertions(+)
diff --git a/modules/platforms/cpp/tests/client-test/compute_test.cpp b/modules/platforms/cpp/tests/client-test/compute_test.cpp
index 94f274f41e..c35feba204 100644
--- a/modules/platforms/cpp/tests/client-test/compute_test.cpp
+++ b/modules/platforms/cpp/tests/client-test/compute_test.cpp
@@ -130,6 +130,15 @@ TEST_F(compute_test, execute_broadcast_all_nodes) {
EXPECT_EQ(res[get_node(1)].value(), get_node(1).get_name() + "42");
}
+TEST_F(compute_test, execute_with_args) {
+ auto cluster_nodes = m_client.get_cluster_nodes();
+
+ auto result = m_client.get_compute().execute(cluster_nodes, CONCAT_JOB, {5.3, uuid(), "42", nullptr});
+
+ ASSERT_TRUE(result.has_value());
+ EXPECT_EQ(result.value().get<std::string>(), "5.3_00000000-0000-0000-0000-000000000000_42_null");
+}
+
diff --git a/modules/platforms/cpp/tests/client-test/ignite_runner_suite.h b/modules/platforms/cpp/tests/client-test/ignite_runner_suite.h
index 0766823eed..e45c04c56a 100644
--- a/modules/platforms/cpp/tests/client-test/ignite_runner_suite.h
+++ b/modules/platforms/cpp/tests/client-test/ignite_runner_suite.h
@@ -50,6 +50,7 @@ public:
"org.apache.ignite.internal.runner.app.client.ItThinClientComputeTest";
inline static const std::string NODE_NAME_JOB = IT_THIN_CLIENT_COMPUTE_TEST + "$NodeNameJob";
+ inline static const std::string CONCAT_JOB = IT_THIN_CLIENT_COMPUTE_TEST + "$ConcatJob";
static constexpr const char *KEY_COLUMN = "key";
static constexpr const char *VAL_COLUMN = "val";
[ignite-3] 03/19: IGNITE-17607 execute_async implementation
Posted by is...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
isapego pushed a commit to branch ignite-17607
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 5a95982f7cf3e4227bc5bc45273d2254d12d166e
Author: Igor Sapego <is...@apache.org>
AuthorDate: Tue Mar 7 13:34:52 2023 +0300
IGNITE-17607 execute_async implementation
---
.../cpp/ignite/client/compute/compute.cpp | 4 +-
.../platforms/cpp/ignite/client/compute/compute.h | 8 ++--
.../cpp/ignite/client/detail/client_operation.h | 9 ++++
.../ignite/client/detail/compute/compute_impl.cpp | 48 ++++++++++++++++++++--
.../ignite/client/detail/compute/compute_impl.h | 36 ++++++++--------
5 files changed, 76 insertions(+), 29 deletions(-)
diff --git a/modules/platforms/cpp/ignite/client/compute/compute.cpp b/modules/platforms/cpp/ignite/client/compute/compute.cpp
index 1a4de991b7..2869346abd 100644
--- a/modules/platforms/cpp/ignite/client/compute/compute.cpp
+++ b/modules/platforms/cpp/ignite/client/compute/compute.cpp
@@ -50,11 +50,11 @@ void inline check_non_empty(const T &cont, const std::string& title) {
}
void compute::execute_async(const std::vector<cluster_node>& nodes, std::string_view job_class_name,
- std::vector<primitive> args, ignite_callback<std::optional<primitive>> callback) {
+ const std::vector<primitive>& args, ignite_callback<std::optional<primitive>> callback) {
check_non_empty(nodes, "Nodes container");
check_non_empty(job_class_name, "Job class name");
- m_impl->execute_on_one_node(get_random_element(nodes), job_class_name, std::move(args), std::move(callback));
+ m_impl->execute_on_one_node(get_random_element(nodes), job_class_name, args, std::move(callback));
}
} // namespace ignite
diff --git a/modules/platforms/cpp/ignite/client/compute/compute.h b/modules/platforms/cpp/ignite/client/compute/compute.h
index d4d5f511dd..5599ac3126 100644
--- a/modules/platforms/cpp/ignite/client/compute/compute.h
+++ b/modules/platforms/cpp/ignite/client/compute/compute.h
@@ -51,7 +51,7 @@ public:
* @param callback A callback called on operation completion with job execution result.
*/
IGNITE_API void execute_async(const std::vector<cluster_node>& nodes, std::string_view job_class_name,
- std::vector<primitive> args, ignite_callback<std::optional<primitive>> callback);
+ const std::vector<primitive>& args, ignite_callback<std::optional<primitive>> callback);
/**
* Executes a compute job represented by the given class on one of the specified nodes.
@@ -62,10 +62,10 @@ public:
* @return Job execution result.
*/
IGNITE_API std::optional<primitive> execute(std::vector<cluster_node> nodes, std::string_view job_class_name,
- std::vector<primitive> args) {
+ const std::vector<primitive>& args) {
return sync<std::optional<primitive>>(
- [this, nodes = std::move(nodes), job_class_name, args = std::move(args)](auto callback) mutable {
- execute_async(std::move(nodes), job_class_name, std::move(args), std::move(callback));
+ [this, nodes = std::move(nodes), job_class_name, args](auto callback) mutable {
+ execute_async(std::move(nodes), job_class_name, args, std::move(callback));
});
}
diff --git a/modules/platforms/cpp/ignite/client/detail/client_operation.h b/modules/platforms/cpp/ignite/client/detail/client_operation.h
index 52295c774f..dc75b16387 100644
--- a/modules/platforms/cpp/ignite/client/detail/client_operation.h
+++ b/modules/platforms/cpp/ignite/client/detail/client_operation.h
@@ -89,6 +89,15 @@ enum class client_operation {
/** Rollback transaction. */
TX_ROLLBACK = 45,
+ /** Execute compute job. */
+ COMPUTE_EXECUTE = 47,
+
+ /** Get cluster nodes. */
+ CLUSTER_GET_NODES = 48,
+
+ /** Execute compute job. */
+ COMPUTE_EXECUTE_COLOCATED = 49,
+
/** Execute SQL query. */
SQL_EXEC = 50,
diff --git a/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.cpp b/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.cpp
index fa719a66a3..4cd9a00335 100644
--- a/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.cpp
+++ b/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.cpp
@@ -16,12 +16,54 @@
*/
#include "ignite/client/detail/compute/compute_impl.h"
+#include "ignite/client/detail/utils.h"
+#include "ignite/schema/binary_tuple_builder.h"
namespace ignite::detail {
-void compute_impl::execute_on_one_node(cluster_node nodes, std::string_view job_class_name, std::vector<primitive> args,
- ignite_callback<std::optional<primitive>> callback) {
- // TODO: Implement me.
+void compute_impl::execute_on_one_node(cluster_node node, std::string_view job_class_name,
+ const std::vector<primitive>& args, ignite_callback<std::optional<primitive>> callback) {
+
+ auto writer_func = [&node, job_class_name, args](protocol::writer &writer) {
+ writer.write(node.get_name());
+ writer.write(job_class_name);
+
+ // TODO: Move to a separate function.
+ auto args_num = std::int32_t(args.size());
+
+ writer.write(args_num);
+
+ binary_tuple_builder args_builder{args_num * 3};
+
+ args_builder.start();
+ for (const auto &arg : args) {
+ claim_primitive_with_type(args_builder, arg);
+ }
+
+ args_builder.layout();
+ for (const auto &arg : args) {
+ append_primitive_with_type(args_builder, arg);
+ }
+
+ auto args_data = args_builder.build();
+ writer.write_binary(args_data);
+ };
+
+ auto reader_func = [](protocol::reader &reader) -> std::optional<primitive> {
+ if (reader.try_read_nil())
+ return std::nullopt;
+
+ auto tuple_data = reader.read_binary();
+ binary_tuple_parser parser(3, tuple_data);
+
+ // TODO: Move to separate func
+ auto typ = ignite_type(binary_tuple_parser::get_int32(parser.get_next().value()));
+ auto scale = binary_tuple_parser::get_int32(parser.get_next().value());
+ return read_next_column(parser, typ, scale);
+ };
+
+ m_connection->perform_request<std::optional<primitive>>(
+ client_operation::COMPUTE_EXECUTE, writer_func, std::move(reader_func), std::move(callback));
}
} // namespace ignite::detail
diff --git a/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.h b/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.h
index a43167fa3c..7328d402f3 100644
--- a/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.h
+++ b/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.h
@@ -17,10 +17,9 @@
#pragma once
+#include "ignite/client/detail/cluster_connection.h"
#include "ignite/client/network/cluster_node.h"
#include "ignite/client/primitive.h"
-#include "ignite/client/transaction/transaction.h"
-#include "ignite/common/config.h"
#include "ignite/common/ignite_result.h"
#include <memory>
@@ -35,31 +34,28 @@ class compute_impl {
friend class ignite_client;
public:
- // Delete
- compute_impl() = delete;
+ /**
+ * Constructor.
+ *
+ * @param connection Connection.
+ */
+ explicit compute_impl(std::shared_ptr<cluster_connection> connection)
+ : m_connection(std::move(connection)) {}
/**
- * Executes single SQL statement asynchronously and returns rows.
+ * Executes a compute job represented by the given class on the specified node asynchronously.
*
- * @param tx Optional transaction. If nullptr implicit transaction for this single operation is used.
- * @param statement Statement to execute.
- * @param args Arguments for the statement.
- * @param callback A callback called on operation completion with SQL result set.
+ * @param node Node to use for the job execution.
+ * @param job_class_name Java class name of the job to execute.
+ * @param args Job arguments.
+ * @param callback A callback called on operation completion with job execution result.
*/
- void execute_on_one_node(cluster_node nodes, std::string_view job_class_name, std::vector<primitive> args,
+ void execute_on_one_node(cluster_node node, std::string_view job_class_name, const std::vector<primitive>& args,
ignite_callback<std::optional<primitive>> callback);
private:
-// /**
-// * Constructor
-// *
-// * @param impl Implementation
-// */
-// explicit compute(std::shared_ptr<detail::compute_impl> impl)
-// : m_impl(std::move(impl)) {}
-//
-// /** Implementation. */
-// std::shared_ptr<detail::compute_impl> m_impl;
+ /** Cluster connection. */
+ std::shared_ptr<cluster_connection> m_connection;
};
} // namespace ignite::detail
[ignite-3] 08/19: IGNITE-17607 Implement broadcast
Posted by is...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
isapego pushed a commit to branch ignite-17607
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 27c15ab6aa1d9ad5ae2da5e82452403b1b543fa9
Author: Igor Sapego <is...@apache.org>
AuthorDate: Tue Mar 7 17:44:00 2023 +0300
IGNITE-17607 Implement broadcast
---
.../cpp/ignite/client/compute/compute.cpp | 32 ++++++++
.../platforms/cpp/ignite/client/compute/compute.h | 40 ++++++++--
.../cpp/ignite/client/network/cluster_node.h | 85 ++++++++++++++++++++++
.../cpp/tests/client-test/compute_test.cpp | 2 +
4 files changed, 154 insertions(+), 5 deletions(-)
diff --git a/modules/platforms/cpp/ignite/client/compute/compute.cpp b/modules/platforms/cpp/ignite/client/compute/compute.cpp
index 2869346abd..eab6b63092 100644
--- a/modules/platforms/cpp/ignite/client/compute/compute.cpp
+++ b/modules/platforms/cpp/ignite/client/compute/compute.cpp
@@ -49,6 +49,38 @@ void inline check_non_empty(const T &cont, const std::string& title) {
throw ignite_error(title + " can not be empty");
}
+void compute::broadcast_async(const std::set<cluster_node>& nodes, std::string_view job_class_name,
+ const std::vector<primitive>& args,
+ ignite_callback<std::map<cluster_node, ignite_result<std::optional<primitive>>>> callback) {
+ typedef std::map<cluster_node, ignite_result<std::optional<primitive>>> result_type;
+
+ check_non_empty(nodes, "Nodes set");
+ check_non_empty(job_class_name, "Job class name");
+
+ struct result_group {
+ explicit result_group(std::int32_t cnt, ignite_callback<result_type> &&cb) : m_cnt(cnt), m_callback(cb) {}
+
+ std::mutex m_mutex;
+ result_type m_res_map;
+ ignite_callback<result_type> m_callback;
+ std::int32_t m_cnt{0};
+ };
+
+ auto shared_res = std::make_shared<result_group>(std::int32_t(nodes.size()), std::move(callback));
+
+ for (const auto &node : nodes) {
+ m_impl->execute_on_one_node(node, job_class_name, args, [node, shared_res](auto &&res) {
+ auto &val = *shared_res;
+
+ std::lock_guard<std::mutex> lock(val.m_mutex);
+ val.m_res_map.emplace(node, res);
+ --val.m_cnt;
+ if (val.m_cnt == 0)
+ val.m_callback(std::move(val.m_res_map));
+ });
+ }
+}
+
void compute::execute_async(const std::vector<cluster_node>& nodes, std::string_view job_class_name,
const std::vector<primitive>& args, ignite_callback<std::optional<primitive>> callback) {
check_non_empty(nodes, "Nodes container");
diff --git a/modules/platforms/cpp/ignite/client/compute/compute.h b/modules/platforms/cpp/ignite/client/compute/compute.h
index 5599ac3126..312bc04875 100644
--- a/modules/platforms/cpp/ignite/client/compute/compute.h
+++ b/modules/platforms/cpp/ignite/client/compute/compute.h
@@ -25,6 +25,9 @@
#include <memory>
#include <utility>
+#include <vector>
+#include <map>
+#include <set>
namespace ignite {
@@ -61,14 +64,41 @@ public:
* @param args Job arguments.
* @return Job execution result.
*/
- IGNITE_API std::optional<primitive> execute(std::vector<cluster_node> nodes, std::string_view job_class_name,
- const std::vector<primitive>& args) {
- return sync<std::optional<primitive>>(
- [this, nodes = std::move(nodes), job_class_name, args](auto callback) mutable {
- execute_async(std::move(nodes), job_class_name, args, std::move(callback));
+ IGNITE_API std::optional<primitive> execute(
+ const std::vector<cluster_node>& nodes, std::string_view job_class_name, const std::vector<primitive>& args) {
+ return sync<std::optional<primitive>>([this, nodes, job_class_name, args](auto callback) mutable {
+ execute_async(nodes, job_class_name, args, std::move(callback));
});
}
+ /**
+ * Executes a compute job represented by the given class on all of the specified nodes asynchronously.
+ *
+ * @param nodes Nodes to use for the job execution.
+ * @param job_class_name Java class name of the job to execute.
+ * @param args Job arguments.
+ * @param callback A callback called on operation completion with jobs execution result.
+ */
+ IGNITE_API void broadcast_async(const std::set<cluster_node>& nodes, std::string_view job_class_name,
+ const std::vector<primitive>& args,
+ ignite_callback<std::map<cluster_node, ignite_result<std::optional<primitive>>>> callback);
+
+ /**
+ * Executes a compute job represented by the given class on one of the specified nodes.
+ *
+ * @param nodes Nodes to use for the job execution.
+ * @param job_class_name Java class name of the job to execute.
+ * @param args Job arguments.
+ * @return Job execution result.
+ */
+ IGNITE_API std::map<cluster_node, ignite_result<std::optional<primitive>>> broadcast(
+ const std::set<cluster_node>& nodes, std::string_view job_class_name, const std::vector<primitive>& args) {
+ return sync<std::map<cluster_node, ignite_result<std::optional<primitive>>>>(
+ [this, nodes, job_class_name, args](auto callback) mutable {
+ broadcast_async(nodes, job_class_name, args, std::move(callback));
+ });
+ }
+
private:
/**
* Constructor
diff --git a/modules/platforms/cpp/ignite/client/network/cluster_node.h b/modules/platforms/cpp/ignite/client/network/cluster_node.h
index 1392b17d6d..4e0a5f6e08 100644
--- a/modules/platforms/cpp/ignite/client/network/cluster_node.h
+++ b/modules/platforms/cpp/ignite/client/network/cluster_node.h
@@ -73,6 +73,25 @@ public:
return m_address;
}
+ /**
+ * Compare to another instance.
+ *
+ * @param other Another instance.
+ * @return Negative value if less, positive if larger and zero, if equals
+ * another instance.
+ */
+ [[nodiscard]] int compare(const cluster_node &other) const {
+ auto name_comp = m_name.compare(other.m_name);
+ if (name_comp)
+ return name_comp;
+
+ auto id_comp = m_id.compare(other.m_id);
+ if (id_comp)
+ return id_comp;
+
+ return m_address.compare(other.m_address);
+ }
+
private:
/** Local ID. */
std::string m_id{};
@@ -84,4 +103,70 @@ private:
network::end_point m_address{};
};
+/**
+ * Comparison operator.
+ *
+ * @param val1 First value.
+ * @param val2 Second value.
+ * @return True if equal.
+ */
+inline bool operator==(const cluster_node &val1, const cluster_node &val2) {
+ return val1.compare(val2) == 0;
+}
+
+/**
+ * Comparison operator.
+ *
+ * @param val1 First value.
+ * @param val2 Second value.
+ * @return True if not equal.
+ */
+inline bool operator!=(const cluster_node &val1, const cluster_node &val2) {
+ return !(val1 == val2);
+}
+
+/**
+ * Comparison operator.
+ *
+ * @param val1 First value.
+ * @param val2 Second value.
+ * @return True if less.
+ */
+inline bool operator<(const cluster_node &val1, const cluster_node &val2) {
+ return val1.compare(val2) < 0;
+}
+
+/**
+ * Comparison operator.
+ *
+ * @param val1 First value.
+ * @param val2 Second value.
+ * @return True if less or equal.
+ */
+inline bool operator<=(const cluster_node &val1, const cluster_node &val2) {
+ return val1.compare(val2) <= 0;
+}
+
+/**
+ * Comparison operator.
+ *
+ * @param val1 First value.
+ * @param val2 Second value.
+ * @return True if greater.
+ */
+inline bool operator>(const cluster_node &val1, const cluster_node &val2) {
+ return val1.compare(val2) > 0;
+}
+
+/**
+ * Comparison operator.
+ *
+ * @param val1 First value.
+ * @param val2 Second value.
+ * @return True if greater or equal.
+ */
+inline bool operator>=(const cluster_node &val1, const cluster_node &val2) {
+ return val1.compare(val2) >= 0;
+}
+
} // namespace ignite
diff --git a/modules/platforms/cpp/tests/client-test/compute_test.cpp b/modules/platforms/cpp/tests/client-test/compute_test.cpp
index b09fd1b11f..7cbd15cc85 100644
--- a/modules/platforms/cpp/tests/client-test/compute_test.cpp
+++ b/modules/platforms/cpp/tests/client-test/compute_test.cpp
@@ -100,3 +100,5 @@ TEST_F(compute_test, execute_on_specific_node) {
EXPECT_EQ(res1.value().get<std::string>(), PLATFORM_TEST_NODE_RUNNER + "-_11");
EXPECT_EQ(res2.value().get<std::string>(), PLATFORM_TEST_NODE_RUNNER + "_2:_22");
}
+
+
[ignite-3] 16/19: IGNITE-17607 Execute colocated, untested
Posted by is...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
isapego pushed a commit to branch ignite-17607
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit aca718633ececced935c2935e792aff67b4b928f
Author: Igor Sapego <is...@apache.org>
AuthorDate: Mon Mar 20 17:23:08 2023 +0300
IGNITE-17607 Execute colocated, untested
---
.../ignite/client/detail/compute/compute_impl.cpp | 82 ++++++++++++----------
1 file changed, 46 insertions(+), 36 deletions(-)
diff --git a/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.cpp b/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.cpp
index ef845a3ff6..b6ebd4dc0b 100644
--- a/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.cpp
+++ b/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.cpp
@@ -21,45 +21,62 @@
namespace ignite::detail {
-void compute_impl::execute_on_one_node(cluster_node node, std::string_view job_class_name,
- const std::vector<primitive>& args, ignite_callback<std::optional<primitive>> callback) {
+/**
+ * Write a collection of primitives as a binary tuple.
+ *
+ * @param writer Writer to use.
+ * @param args Arguments.
+ */
+void write_primitives_as_binary_tuple(protocol::writer &writer, const std::vector<primitive>& args) {
+ auto args_num = std::int32_t(args.size());
- auto writer_func = [&node, job_class_name, args](protocol::writer &writer) {
- writer.write(node.get_name());
- writer.write(job_class_name);
+ writer.write(args_num);
- // TODO: Move to a separate function.
- auto args_num = std::int32_t(args.size());
+ binary_tuple_builder args_builder{args_num * 3};
- writer.write(args_num);
+ args_builder.start();
+ for (const auto &arg : args) {
+ claim_primitive_with_type(args_builder, arg);
+ }
- binary_tuple_builder args_builder{args_num * 3};
+ args_builder.layout();
+ for (const auto &arg : args) {
+ append_primitive_with_type(args_builder, arg);
+ }
- args_builder.start();
- for (const auto &arg : args) {
- claim_primitive_with_type(args_builder, arg);
- }
+ auto args_data = args_builder.build();
+ writer.write_binary(args_data);
+}
- args_builder.layout();
- for (const auto &arg : args) {
- append_primitive_with_type(args_builder, arg);
- }
+/**
+ * Read primitive from a stream, which is encoded as a binary tuple.
+ *
+ * @param reader Reader.
+ * @return Value.
+ */
+std::optional<primitive> read_primitive_from_binary_tuple(protocol::reader &reader) {
+ auto tuple_data = reader.read_binary();
+ binary_tuple_parser parser(3, tuple_data);
+
+ auto typ = ignite_type(binary_tuple_parser::get_int32(parser.get_next().value()));
+ auto scale = binary_tuple_parser::get_int32(parser.get_next().value());
+ return read_next_column(parser, typ, scale);
+}
+
+void compute_impl::execute_on_one_node(cluster_node node, std::string_view job_class_name,
+ const std::vector<primitive>& args, ignite_callback<std::optional<primitive>> callback) {
- auto args_data = args_builder.build();
- writer.write_binary(args_data);
+ auto writer_func = [&node, job_class_name, args](protocol::writer &writer) {
+ writer.write(node.get_name());
+ writer.write(job_class_name);
+ write_primitives_as_binary_tuple(writer, args);
};
auto reader_func = [](protocol::reader &reader) -> std::optional<primitive> {
if (reader.try_read_nil())
return std::nullopt;
- auto tuple_data = reader.read_binary();
- binary_tuple_parser parser(3, tuple_data);
-
- // TODO: Move to separate func
- auto typ = ignite_type(binary_tuple_parser::get_int32(parser.get_next().value()));
- auto scale = binary_tuple_parser::get_int32(parser.get_next().value());
- return read_next_column(parser, typ, scale);
+ return read_primitive_from_binary_tuple(reader);
};
m_connection->perform_request<std::optional<primitive>>(
@@ -86,26 +103,19 @@ void compute_impl::execute_colocated_async(const std::string &table_name, const
table->template with_latest_schema_async<std::optional<primitive>>(std::move(callback),
[table, key = std::move(key), job = std::move(job), args = std::move(args), conn] // NOLINT(performance-move-const-arg)
(const schema& sch, auto callback) mutable {
- auto writer_func = [&key, &sch, &table, &job](protocol::writer &writer) {
+ auto writer_func = [&key, &sch, &table, &job, &args](protocol::writer &writer) {
writer.write(table->get_id());
writer.write(sch.version);
write_tuple(writer, sch, key, true);
writer.write(job);
- // TODO: write arguments.
+ write_primitives_as_binary_tuple(writer, args);
};
auto reader_func = [](protocol::reader &reader) -> std::optional<primitive> {
if (reader.try_read_nil())
return std::nullopt;
- // TODO: Tuple to object.
- auto tuple_data = reader.read_binary();
- ignite_tuple res(3);
- binary_tuple_parser parser(3, tuple_data);
- auto typ = column_type(binary_tuple_parser::get_int32(*parser.get_next()));
- auto scale = binary_tuple_parser::get_int32(*parser.get_next());
-
- return read_next_column(parser, typ, scale);
+ return read_primitive_from_binary_tuple(reader);
};
conn->perform_request<std::optional<primitive>>(