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 2025/01/22 11:36:54 UTC
(ignite-3) branch main updated: IGNITE-24200 C++ Unify Compute API (#5094)
This is an automated email from the ASF dual-hosted git repository.
isapego pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new 4e10fa47033 IGNITE-24200 C++ Unify Compute API (#5094)
4e10fa47033 is described below
commit 4e10fa4703354c639f632cf7e91ece7c6bb53586
Author: Igor Sapego <is...@apache.org>
AuthorDate: Wed Jan 22 12:36:49 2025 +0100
IGNITE-24200 C++ Unify Compute API (#5094)
---
modules/platforms/cpp/ignite/client/CMakeLists.txt | 5 +
.../{job_execution.cpp => broadcast_execution.h} | 55 +++++++----
.../ignite/client/compute/broadcast_job_target.cpp | 43 ++++++++
.../ignite/client/compute/broadcast_job_target.h | 64 ++++++++++++
.../cpp/ignite/client/compute/compute.cpp | 65 +++++++------
.../platforms/cpp/ignite/client/compute/compute.h | 80 +++++----------
.../cpp/ignite/client/compute/job_execution.cpp | 4 +
.../cpp/ignite/client/compute/job_execution.h | 24 +++--
.../cpp/ignite/client/compute/job_target.cpp | 51 ++++++++++
.../cpp/ignite/client/compute/job_target.h | 84 ++++++++++++++++
.../ignite/client/detail/argument_check_utils.h | 20 +++-
.../client/detail/compute/any_node_job_target.h | 55 +++++++++++
.../client/detail/compute/colocated_job_target.h | 68 +++++++++++++
.../ignite/client/detail/compute/compute_impl.cpp | 108 +++++++++++++--------
.../ignite/client/detail/compute/compute_impl.h | 16 ++-
.../client/detail/compute/job_execution_impl.h | 27 ++++--
.../ignite/client/detail/compute/job_target_type.h | 31 ++++++
.../detail/compute/nodes_broadcast_job_target.h | 52 ++++++++++
.../ignite/client/detail/ignite_client_impl.cpp | 14 +--
.../platforms/cpp/ignite/client/detail/utils.cpp | 14 +++
modules/platforms/cpp/ignite/client/detail/utils.h | 26 +++--
.../cpp/ignite/client/network/cluster_node.h | 4 +-
.../cpp/tests/client-test/compute_test.cpp | 95 +++++++++---------
.../client-test/schema_synchronization_test.cpp | 2 +-
.../cpp/tests/test-common/basic_auth_test_suite.h | 2 +-
25 files changed, 764 insertions(+), 245 deletions(-)
diff --git a/modules/platforms/cpp/ignite/client/CMakeLists.txt b/modules/platforms/cpp/ignite/client/CMakeLists.txt
index 83c55f32ad4..910338da055 100644
--- a/modules/platforms/cpp/ignite/client/CMakeLists.txt
+++ b/modules/platforms/cpp/ignite/client/CMakeLists.txt
@@ -21,8 +21,10 @@ set(TARGET ${PROJECT_NAME})
set(SOURCES
ignite_client.cpp
+ compute/broadcast_job_target.cpp
compute/compute.cpp
compute/job_execution.cpp
+ compute/job_target.cpp
sql/sql.cpp
sql/result_set.cpp
table/key_value_view.cpp
@@ -49,6 +51,8 @@ set(PUBLIC_HEADERS
ignite_client_configuration.h
ignite_logger.h
type_mapping.h
+ compute/broadcast_execution.h
+ compute/broadcast_job_target.h
compute/compute.h
compute/deployment_unit.h
compute/job_descriptor.h
@@ -56,6 +60,7 @@ set(PUBLIC_HEADERS
compute/job_execution_options.h
compute/job_state.h
compute/job_status.h
+ compute/job_target.h
detail/type_mapping_utils.h
network/cluster_node.h
sql/column_metadata.h
diff --git a/modules/platforms/cpp/ignite/client/compute/job_execution.cpp b/modules/platforms/cpp/ignite/client/compute/broadcast_execution.h
similarity index 52%
copy from modules/platforms/cpp/ignite/client/compute/job_execution.cpp
copy to modules/platforms/cpp/ignite/client/compute/broadcast_execution.h
index 05405441884..1fbcc5b738e 100644
--- a/modules/platforms/cpp/ignite/client/compute/job_execution.cpp
+++ b/modules/platforms/cpp/ignite/client/compute/broadcast_execution.h
@@ -15,30 +15,43 @@
* limitations under the License.
*/
-#include "ignite/client/compute/job_execution.h"
-#include "ignite/client/detail/compute/job_execution_impl.h"
-
-namespace ignite {
-
-uuid job_execution::get_id() const {
- return m_impl->get_id();
-}
+#pragma once
-void job_execution::get_state_async(ignite_callback<std::optional<job_state>> callback) {
- m_impl->get_state_async(std::move(callback));
-}
+#include "ignite/client/compute/job_execution.h"
+#include "ignite/common/ignite_result.h"
-void job_execution::get_result_async(ignite_callback<std::optional<binary_object>> callback) {
- m_impl->get_result_async(std::move(callback));
-}
+#include <vector>
-void job_execution::cancel_async(ignite_callback<job_execution::operation_result> callback) {
- m_impl->cancel_async(std::move(callback));
-}
+namespace ignite {
-void job_execution::change_priority_async(
- std::int32_t priority, ignite_callback<job_execution::operation_result> callback) {
- m_impl->change_priority_async(priority, std::move(callback));
-}
+/**
+ * Broadcast execution control object, provides information about the broadcast execution process and result.
+ */
+class broadcast_execution {
+public:
+ // Default
+ broadcast_execution() = default;
+
+ /**
+ * Constructor.
+ *
+ * @param executions Executions.
+ */
+ explicit broadcast_execution(std::vector<ignite_result<job_execution>> &&executions)
+ : m_executions(std::move(executions)) {}
+
+ /**
+ * Gets the job executions.
+ *
+ * @return Job executions.
+ */
+ [[nodiscard]] const std::vector<ignite_result<job_execution>> &get_job_executions() const {
+ return m_executions;
+ }
+
+private:
+ /** Executions. */
+ std::vector<ignite_result<job_execution>> m_executions;
+};
} // namespace ignite
diff --git a/modules/platforms/cpp/ignite/client/compute/broadcast_job_target.cpp b/modules/platforms/cpp/ignite/client/compute/broadcast_job_target.cpp
new file mode 100644
index 00000000000..11d71bb890f
--- /dev/null
+++ b/modules/platforms/cpp/ignite/client/compute/broadcast_job_target.cpp
@@ -0,0 +1,43 @@
+/*
+ * 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/broadcast_job_target.h"
+#include "ignite/client/table/ignite_tuple.h"
+
+#include "ignite/client/detail/argument_check_utils.h"
+#include "ignite/client/detail/compute/nodes_broadcast_job_target.h"
+
+namespace ignite {
+
+std::shared_ptr<broadcast_job_target> broadcast_job_target::node(cluster_node val) {
+ return std::shared_ptr<broadcast_job_target>{new detail::nodes_broadcast_job_target{{std::move(val)}}};
+}
+
+std::shared_ptr<broadcast_job_target> broadcast_job_target::nodes(std::set<cluster_node> vals) {
+ detail::arg_check::container_non_empty(vals, "Nodes set");
+
+ return std::shared_ptr<broadcast_job_target>{new detail::nodes_broadcast_job_target{std::move(vals)}};
+}
+
+std::shared_ptr<broadcast_job_target> broadcast_job_target::nodes(const std::vector<cluster_node> &vals) {
+ detail::arg_check::container_non_empty(vals, "Nodes set");
+
+ std::set<cluster_node> node_set(vals.begin(), vals.end());
+ return nodes(node_set);
+}
+
+} // namespace ignite
diff --git a/modules/platforms/cpp/ignite/client/compute/broadcast_job_target.h b/modules/platforms/cpp/ignite/client/compute/broadcast_job_target.h
new file mode 100644
index 00000000000..86f39e575e7
--- /dev/null
+++ b/modules/platforms/cpp/ignite/client/compute/broadcast_job_target.h
@@ -0,0 +1,64 @@
+/*
+ * 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/detail/config.h"
+#include "ignite/client/network/cluster_node.h"
+
+#include <set>
+#include <vector>
+#include <memory>
+
+namespace ignite {
+
+/**
+ * Job execution target.
+ */
+class broadcast_job_target {
+public:
+ // Default
+ virtual ~broadcast_job_target() = default;
+
+ /**
+ * Create a single node job target.
+ *
+ * @param val Node.
+ */
+ [[nodiscard]] IGNITE_API static std::shared_ptr<broadcast_job_target> node(cluster_node val);
+
+ /**
+ * Create a multiple node job target.
+ *
+ * @param vals Nodes.
+ */
+ [[nodiscard]] IGNITE_API static std::shared_ptr<broadcast_job_target> nodes(std::set<cluster_node> vals);
+
+ /**
+ * Create a multiple node job target.
+ *
+ * @param vals Nodes.
+ */
+ [[nodiscard]] IGNITE_API static std::shared_ptr<broadcast_job_target> nodes(const std::vector<cluster_node> &vals);
+
+protected:
+ // Default
+ broadcast_job_target() = default;
+};
+
+
+} // namespace ignite
diff --git a/modules/platforms/cpp/ignite/client/compute/compute.cpp b/modules/platforms/cpp/ignite/client/compute/compute.cpp
index 21d9ac31e0b..0836a2944d3 100644
--- a/modules/platforms/cpp/ignite/client/compute/compute.cpp
+++ b/modules/platforms/cpp/ignite/client/compute/compute.cpp
@@ -15,63 +15,74 @@
* limitations under the License.
*/
-#include "ignite/client/compute/compute.h"
#include "ignite/client/detail/argument_check_utils.h"
+#include "ignite/client/detail/compute/nodes_broadcast_job_target.h"
+#include <ignite/client/detail/compute/any_node_job_target.h>
+#include <ignite/client/detail/compute/colocated_job_target.h>
#include "ignite/client/detail/compute/compute_impl.h"
+#include "ignite/client/compute/compute.h"
+
namespace ignite {
-void compute::submit_async(const std::vector<cluster_node> &nodes, std::shared_ptr<job_descriptor> descriptor,
+void compute::submit_async(std::shared_ptr<job_target> target, std::shared_ptr<job_descriptor> descriptor,
const binary_object &arg, ignite_callback<job_execution> callback) {
- detail::arg_check::container_non_empty(nodes, "Nodes container");
+ detail::arg_check::pointer_valid(target, "Target");
detail::arg_check::container_non_empty(descriptor->get_job_class_name(), "Job class name");
- m_impl->submit_to_nodes(nodes, descriptor, arg, std::move(callback));
+ switch (target->get_type()) {
+ case detail::job_target_type::ANY_NODE: {
+ auto any_node_target = static_cast<detail::any_node_job_target*>(target.get());
+ m_impl->submit_to_nodes(any_node_target->get_nodes(), descriptor, arg, std::move(callback));
+ break;
+ }
+
+ case detail::job_target_type::COLOCATED: {
+ auto colocated_target = static_cast<detail::colocated_job_target*>(target.get());
+ m_impl->submit_colocated_async(*colocated_target, descriptor, arg, std::move(callback));
+ break;
+ }
+
+ default: {
+ assert(false);
+ }
+ }
+
}
-void compute::submit_broadcast_async(const std::set<cluster_node> &nodes, std::shared_ptr<job_descriptor> descriptor,
- const binary_object &arg,
- ignite_callback<std::map<cluster_node, ignite_result<job_execution>>> callback) {
- typedef std::map<cluster_node, ignite_result<job_execution>> result_type;
+void compute::submit_broadcast_async(std::shared_ptr<broadcast_job_target> target,
+ std::shared_ptr<job_descriptor> descriptor, const binary_object &arg,
+ ignite_callback<broadcast_execution> callback) {
- detail::arg_check::container_non_empty(nodes, "Nodes set");
+ detail::arg_check::pointer_valid(target, "Target pointer");
detail::arg_check::container_non_empty(descriptor->get_job_class_name(), "Job class name");
struct result_group {
- explicit result_group(std::int32_t cnt, ignite_callback<result_type> &&cb)
+ explicit result_group(std::int32_t cnt, ignite_callback<broadcast_execution> &&cb)
: m_cnt(cnt)
, m_callback(cb) {}
std::mutex m_mutex;
- result_type m_res_map;
+ std::vector<ignite_result<job_execution>> m_res_vector;
std::int32_t m_cnt{0};
- ignite_callback<result_type> m_callback;
+ ignite_callback<broadcast_execution> m_callback;
};
+ auto nodes = static_cast<detail::nodes_broadcast_job_target&>(*target).get_nodes();
auto shared_res = std::make_shared<result_group>(std::int32_t(nodes.size()), std::move(callback));
for (const auto &node : nodes) {
- std::vector<cluster_node> candidates = {node};
- m_impl->submit_to_nodes(candidates, descriptor, arg, [node, shared_res](auto &&res) {
+ std::set<cluster_node> candidates = {node};
+ m_impl->submit_to_nodes(candidates, descriptor, arg, [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_res_vector.emplace_back(std::move(res));
--val.m_cnt;
if (val.m_cnt == 0)
- val.m_callback(std::move(val.m_res_map));
+ val.m_callback(broadcast_execution(std::move(val.m_res_vector)));
});
}
}
-void compute::submit_colocated_async(std::string_view table_name, const ignite_tuple &key,
- std::shared_ptr<job_descriptor> descriptor, const binary_object &arg, ignite_callback<job_execution> 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(descriptor->get_job_class_name(), "Job class name");
-
- m_impl->submit_colocated_async(
- std::string(table_name), key, descriptor, arg, std::move(callback));
-}
-
-} // namespace ignite
+} // namespace ignite
\ No newline at end of file
diff --git a/modules/platforms/cpp/ignite/client/compute/compute.h b/modules/platforms/cpp/ignite/client/compute/compute.h
index 1c62618183f..38011f8b525 100644
--- a/modules/platforms/cpp/ignite/client/compute/compute.h
+++ b/modules/platforms/cpp/ignite/client/compute/compute.h
@@ -17,23 +17,19 @@
#pragma once
-#include "ignite/client/compute/deployment_unit.h"
+#include "ignite/client/compute/broadcast_execution.h"
+#include "ignite/client/compute/broadcast_job_target.h"
#include "ignite/client/compute/job_descriptor.h"
#include "ignite/client/compute/job_execution.h"
-#include "ignite/client/compute/job_execution_options.h"
+#include "ignite/client/compute/job_target.h"
#include "ignite/client/network/cluster_node.h"
-#include "ignite/client/table/ignite_tuple.h"
-#include "ignite/client/transaction/transaction.h"
#include "ignite/common/binary_object.h"
#include "ignite/common/detail/config.h"
#include "ignite/common/ignite_result.h"
-#include "ignite/common/primitive.h"
-#include <map>
#include <memory>
-#include <set>
#include <utility>
-#include <vector>
+
namespace ignite {
@@ -52,94 +48,64 @@ public:
compute() = delete;
/**
- * Submits for execution a compute job represented by the given class for an execution on one of the specified
- * nodes.
+ * Submits for execution a compute job represented by the given class for an execution on the specified target.
*
- * @param nodes Nodes to use for the job execution.
+ * @param target Job target.
* @param descriptor Descriptor.
* @param arg Job argument.
* @param callback A callback called on operation completion with job execution result.
*/
- IGNITE_API void submit_async(const std::vector<cluster_node> &nodes, std::shared_ptr<job_descriptor> descriptor,
+ IGNITE_API void submit_async(std::shared_ptr<job_target> target, std::shared_ptr<job_descriptor> descriptor,
const binary_object &arg, ignite_callback<job_execution> callback);
/**
- * Submits for execution a compute job represented by the given class on one of the specified nodes.
+ * Submits for execution a compute job represented by the given class on the specified target.
*
- * @param nodes Nodes to use for the job execution.
+ * @param target Job target.
* @param descriptor Descriptor.
* @param arg Job argument.
* @return Job execution result.
*/
- IGNITE_API job_execution submit(const std::vector<cluster_node> &nodes, std::shared_ptr<job_descriptor> descriptor,
+ IGNITE_API job_execution submit(std::shared_ptr<job_target> target, std::shared_ptr<job_descriptor> descriptor,
const binary_object &arg) {
return sync<job_execution>([&](auto callback) mutable {
- submit_async(nodes, descriptor, arg, std::move(callback));
+ submit_async(std::move(target), std::move(descriptor), arg, std::move(callback));
});
}
/**
- * Broadcast a compute job represented by the given class for an execution on all of the specified nodes.
+ * Broadcast a compute job represented by the given class for an execution on all the specified nodes.
*
- * @param nodes Nodes to use for the job execution.
+ * @param target Job target.
* @param descriptor Descriptor.
* @param arg Job argument.
* @param callback A callback called on operation completion with jobs execution result.
*/
- IGNITE_API void submit_broadcast_async(const std::set<cluster_node> &nodes,
+ IGNITE_API void submit_broadcast_async(std::shared_ptr<broadcast_job_target> target,
std::shared_ptr<job_descriptor> descriptor, const binary_object &arg,
- ignite_callback<std::map<cluster_node, ignite_result<job_execution>>> callback);
+ ignite_callback<broadcast_execution> callback);
/**
- * Broadcast a compute job represented by the given class on all of the specified nodes.
+ * Broadcast a compute job represented by the given class on all the specified nodes.
*
- * @param nodes Nodes to use for the job execution.
+ * @param target Job target.
* @param descriptor Descriptor.
* @param arg Job argument.
* @return Job execution result.
*/
- IGNITE_API std::map<cluster_node, ignite_result<job_execution>> submit_broadcast(
- const std::set<cluster_node> &nodes, std::shared_ptr<job_descriptor> descriptor,
+ IGNITE_API broadcast_execution submit_broadcast(
+ std::shared_ptr<broadcast_job_target> target, std::shared_ptr<job_descriptor> descriptor,
const binary_object &arg) {
- return sync<std::map<cluster_node, ignite_result<job_execution>>>([&](auto callback) mutable {
- submit_broadcast_async(nodes, descriptor, arg, std::move(callback));
- });
- }
-
- /**
- * Asynchronously executes a job represented by the given class on one node where the given key is located.
- *
- * @param table_name 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 descriptor Descriptor.
- * @param arg Job argument.
- * @param callback A callback called on operation completion with job execution result.
- */
- IGNITE_API void submit_colocated_async(std::string_view table_name, const ignite_tuple &key,
- std::shared_ptr<job_descriptor> descriptor, const binary_object &arg,
- ignite_callback<job_execution> callback);
-
- /**
- * Synchronously executes a job represented by the given class on one node where the given key is located.
- *
- * @param table_name 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 descriptor Descriptor.
- * @param arg Job argument.
- * @return Job execution result.
- */
- IGNITE_API job_execution submit_colocated(std::string_view table_name, const ignite_tuple &key,
- std::shared_ptr<job_descriptor> descriptor, const binary_object &arg) {
- return sync<job_execution>([&](auto callback) mutable {
- submit_colocated_async(table_name, key, descriptor, arg, std::move(callback));
+ return sync<broadcast_execution>([&](auto callback) mutable {
+ submit_broadcast_async(std::move(target), std::move(descriptor), arg, std::move(callback));
});
}
private:
/**
- * Constructor
+ * Constructor.
*
- * @param impl Implementation
+ * @param impl Implementation.
*/
explicit compute(std::shared_ptr<detail::compute_impl> impl)
: m_impl(std::move(impl)) {}
diff --git a/modules/platforms/cpp/ignite/client/compute/job_execution.cpp b/modules/platforms/cpp/ignite/client/compute/job_execution.cpp
index 05405441884..ffee6fb56f7 100644
--- a/modules/platforms/cpp/ignite/client/compute/job_execution.cpp
+++ b/modules/platforms/cpp/ignite/client/compute/job_execution.cpp
@@ -24,6 +24,10 @@ uuid job_execution::get_id() const {
return m_impl->get_id();
}
+const cluster_node &job_execution::get_node() const {
+ return m_impl->get_node();
+}
+
void job_execution::get_state_async(ignite_callback<std::optional<job_state>> callback) {
m_impl->get_state_async(std::move(callback));
}
diff --git a/modules/platforms/cpp/ignite/client/compute/job_execution.h b/modules/platforms/cpp/ignite/client/compute/job_execution.h
index 8ef30195fa2..bd7dfaa5b8d 100644
--- a/modules/platforms/cpp/ignite/client/compute/job_execution.h
+++ b/modules/platforms/cpp/ignite/client/compute/job_execution.h
@@ -18,6 +18,7 @@
#pragma once
#include "ignite/client/compute/job_state.h"
+#include "ignite/client/network/cluster_node.h"
#include "ignite/common/detail/config.h"
#include "ignite/common/ignite_result.h"
#include "ignite/common/binary_object.h"
@@ -66,14 +67,21 @@ public:
*
* @return Job ID.
*/
- [[nodiscard]] uuid get_id() const;
+ [[nodiscard]] IGNITE_API uuid get_id() const;
+
+ /**
+ * Gets the cluster node.
+ *
+ * @return Cluster node.
+ */
+ [[nodiscard]] IGNITE_API const cluster_node &get_node() const;
/**
* Gets the job execution state asynchronously. Can be @c nullopt if the job state no longer exists due to
* exceeding the retention time limit.
*
* @param callback Callback to be called when the operation is complete. Called with the job state.
- * Can be @c nullopt if the job state no longer exists due to exceeding the retention time limit.
+ * It Can be @c nullopt if the job state no longer exists due to exceeding the retention time limit.
*/
IGNITE_API void get_state_async(ignite_callback<std::optional<job_state>> callback);
@@ -92,7 +100,7 @@ public:
/**
* Gets the job execution result asynchronously.
*
- * Only one callback can be submitted for this operation at a time, which means you can not call this method in
+ * Only one callback can be submitted for this operation at a time, which means you cannot call this method in
* parallel.
* @param callback Callback to be called when the operation is complete. Called with the job execution result.
*/
@@ -101,7 +109,7 @@ public:
/**
* Gets the job execution result.
*
- * Only one thread can wait for result at a time, which means you can not call this method in parallel from
+ * Only one thread can wait for a result at a time, which means you cannot call this method in parallel from
* multiple threads.
* @return The job execution result.
*/
@@ -120,14 +128,14 @@ public:
/**
* Cancels the job execution.
*
- * @param return Result of the cancel operation.
+ * @return Result of the cancel operation.
*/
IGNITE_API operation_result cancel() {
return sync<operation_result>([this](auto callback) mutable { cancel_async(std::move(callback)); });
}
/**
- * Changes the job priority asynchronously. After priority change the job will be the last in the queue of jobs
+ * Changes the job priority asynchronously. After priority change, the job will be the last in the queue of jobs
* with the same priority.
*
* @param priority New priority.
@@ -136,11 +144,11 @@ public:
IGNITE_API void change_priority_async(std::int32_t priority, ignite_callback<operation_result> callback);
/**
- * Changes the job priority. After priority change the job will be the last in the queue of jobs with the same
+ * Changes the job priority. After priority change, the job will be the last in the queue of jobs with the same
* priority.
*
* @param priority New priority.
- * @param return Result of the operation.
+ * @return Result of the operation.
*/
IGNITE_API operation_result change_priority(std::int32_t priority) {
return sync<operation_result>(
diff --git a/modules/platforms/cpp/ignite/client/compute/job_target.cpp b/modules/platforms/cpp/ignite/client/compute/job_target.cpp
new file mode 100644
index 00000000000..08e47c0d074
--- /dev/null
+++ b/modules/platforms/cpp/ignite/client/compute/job_target.cpp
@@ -0,0 +1,51 @@
+/*
+ * 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/job_target.h"
+#include "ignite/client/table/ignite_tuple.h"
+
+#include "ignite/client/detail/argument_check_utils.h"
+#include "ignite/client/detail/compute/any_node_job_target.h"
+#include "ignite/client/detail/compute/colocated_job_target.h"
+
+namespace ignite {
+
+std::shared_ptr<job_target> job_target::node(cluster_node node) {
+ return std::shared_ptr<job_target>{new detail::any_node_job_target{{std::move(node)}}};
+}
+
+std::shared_ptr<job_target> job_target::any_node(std::set<cluster_node> nodes) {
+ detail::arg_check::container_non_empty(nodes, "Nodes set");
+
+ return std::shared_ptr<job_target>{new detail::any_node_job_target{std::move(nodes)}};
+}
+
+std::shared_ptr<job_target> job_target::any_node(const std::vector<cluster_node> &nodes) {
+ detail::arg_check::container_non_empty(nodes, "Nodes set");
+
+ std::set<cluster_node> node_set(nodes.begin(), nodes.end());
+ return any_node(node_set);
+}
+
+std::shared_ptr<job_target> job_target::colocated(std::string_view table_name, const ignite_tuple &key) {
+ detail::arg_check::container_non_empty(table_name, "Table name");
+ detail::arg_check::tuple_non_empty(key, "Key tuple");
+
+ return std::shared_ptr<job_target>{new detail::colocated_job_target{std::string{table_name}, key}};
+}
+
+} // namespace ignite
diff --git a/modules/platforms/cpp/ignite/client/compute/job_target.h b/modules/platforms/cpp/ignite/client/compute/job_target.h
new file mode 100644
index 00000000000..063c33a63cd
--- /dev/null
+++ b/modules/platforms/cpp/ignite/client/compute/job_target.h
@@ -0,0 +1,84 @@
+/*
+ * 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/detail/config.h"
+#include "ignite/client/network/cluster_node.h"
+#include "ignite/client/detail/compute/job_target_type.h"
+
+#include <set>
+#include <vector>
+#include <memory>
+
+namespace ignite {
+class ignite_tuple;
+class compute;
+class job_target;
+
+/**
+ * Job execution target.
+ */
+class job_target {
+ friend class compute;
+public:
+ // Default
+ virtual ~job_target() = default;
+
+ /**
+ * Create a single node job target.
+ *
+ * @param node Node.
+ */
+ [[nodiscard]] IGNITE_API static std::shared_ptr<job_target> node(cluster_node node);
+
+ /**
+ * Create a multiple node job target.
+ *
+ * @param nodes Nodes.
+ */
+ [[nodiscard]] IGNITE_API static std::shared_ptr<job_target> any_node(std::set<cluster_node> nodes);
+
+ /**
+ * Create a multiple node job target.
+ *
+ * @param nodes Nodes.
+ */
+ [[nodiscard]] IGNITE_API static std::shared_ptr<job_target> any_node(const std::vector<cluster_node> &nodes);
+
+ /**
+ * Creates a colocated job target for a specific table and key.
+ *
+ * @param table_name Table name.
+ * @param key Key.
+ */
+ [[nodiscard]] IGNITE_API static std::shared_ptr<job_target> colocated(std::string_view table_name, const ignite_tuple &key);
+
+protected:
+ // Default
+ job_target() = default;
+
+ /**
+ * Get the job type.
+ *
+ * @return Job type.
+ */
+ [[nodiscard]] virtual detail::job_target_type get_type() const = 0;
+};
+
+
+} // namespace ignite
diff --git a/modules/platforms/cpp/ignite/client/detail/argument_check_utils.h b/modules/platforms/cpp/ignite/client/detail/argument_check_utils.h
index 8c159ccfe74..accd496023e 100644
--- a/modules/platforms/cpp/ignite/client/detail/argument_check_utils.h
+++ b/modules/platforms/cpp/ignite/client/detail/argument_check_utils.h
@@ -26,7 +26,8 @@ namespace ignite::detail::arg_check {
/**
* Check key argument.
*
- * @param key Key tuple.
+ * @param value Value.
+ * @param title Title.
*/
void inline tuple_non_empty(const ignite_tuple &value, const std::string &title) {
if (0 == value.column_count())
@@ -52,9 +53,10 @@ void inline value_tuple_non_empty(const ignite_tuple &value) {
}
/**
- * Check value argument.
+ * Check container argument.
*
- * @param value Value tuple.
+ * @param cont Value tuple.
+ * @param title Title.
*/
template<typename T>
void inline container_non_empty(const T &cont, const std::string &title) {
@@ -62,4 +64,16 @@ void inline container_non_empty(const T &cont, const std::string &title) {
throw ignite_error(title + " can not be empty");
}
+/**
+ * Check pointer argument.
+ *
+ * @param ptr Pointer.
+ * @param title Title.
+ */
+template<typename T>
+void inline pointer_valid(const T &ptr, const std::string &title) {
+ if (!ptr)
+ throw ignite_error(title + " can not be nullptr");
+}
+
} // namespace ignite::detail::arg_check
diff --git a/modules/platforms/cpp/ignite/client/detail/compute/any_node_job_target.h b/modules/platforms/cpp/ignite/client/detail/compute/any_node_job_target.h
new file mode 100644
index 00000000000..c5a2d6b49b5
--- /dev/null
+++ b/modules/platforms/cpp/ignite/client/detail/compute/any_node_job_target.h
@@ -0,0 +1,55 @@
+/*
+ * 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/compute/job_target.h"
+#include "ignite/client/detail/compute/job_target_type.h"
+
+namespace ignite::detail {
+
+/**
+ * Job target represented by a set of nodes.
+ */
+class any_node_job_target : public job_target {
+public:
+ // Default
+ any_node_job_target() = default;
+
+ /**
+ * Constructor.
+ *
+ * @param nodes Nodes.
+ */
+ explicit any_node_job_target(std::set<cluster_node> &&nodes)
+ : m_nodes(std::move(nodes)) {}
+
+ /**
+ * Get nodes.
+ *
+ * @return Nodes.
+ */
+ [[nodiscard]] const std::set<cluster_node> &get_nodes() const { return m_nodes; }
+
+ [[nodiscard]] job_target_type get_type() const override { return job_target_type::ANY_NODE; }
+
+private:
+ /** Nodes. */
+ std::set<cluster_node> m_nodes;
+};
+
+} // namespace ignite::detail
diff --git a/modules/platforms/cpp/ignite/client/detail/compute/colocated_job_target.h b/modules/platforms/cpp/ignite/client/detail/compute/colocated_job_target.h
new file mode 100644
index 00000000000..703efee482b
--- /dev/null
+++ b/modules/platforms/cpp/ignite/client/detail/compute/colocated_job_target.h
@@ -0,0 +1,68 @@
+/*
+ * 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 "ignite/client/compute/job_target.h"
+#include "ignite/client/detail/compute/job_target_type.h"
+
+namespace ignite::detail {
+
+/**
+ * Job target defined by a table name and a key.
+ */
+class colocated_job_target : public job_target {
+public:
+ // Default
+ colocated_job_target() = default;
+
+ /**
+ * Constructor.
+ *
+ * @param table_name Name of the table to be used with a @c key to determine target node.
+ * @param key Table key to be used to determine the target node for job execution.
+ */
+ explicit colocated_job_target(std::string table_name, const ignite_tuple &key)
+ : m_table_name(std::move(table_name))
+ , m_key(key) {}
+
+ /**
+ * Get table name.
+ *
+ * @return Table name.
+ */
+ [[nodiscard]] const std::string & get_table_name() const { return m_table_name; }
+
+ /**
+ * Get key.
+ *
+ * @return Key.
+ */
+ [[nodiscard]] const ignite_tuple & get_key() const { return m_key; }
+
+ [[nodiscard]] job_target_type get_type() const override { return job_target_type::COLOCATED; }
+
+private:
+ /** Table name. */
+ std::string m_table_name;
+
+ /** Key. */
+ ignite_tuple m_key;
+};
+
+} // namespace ignite::detail
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 0ffc71d19d3..8cf2d6cfe98 100644
--- a/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.cpp
+++ b/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.cpp
@@ -16,6 +16,8 @@
*/
#include "ignite/client/detail/compute/compute_impl.h"
+
+#include "colocated_job_target.h"
#include "ignite/client/detail/argument_check_utils.h"
#include "ignite/client/detail/compute/job_execution_impl.h"
#include "ignite/client/detail/utils.h"
@@ -171,7 +173,7 @@ void write_units(protocol::writer &writer, const std::vector<deployment_unit> &u
}
/**
- * Response handler implementation for a compute.
+ * Response handler implementation for a job compute.
*/
class response_handler_compute final : public response_handler_adapter<job_execution> {
public:
@@ -181,7 +183,9 @@ public:
/**
* Constructor.
*
+ * @param compute Compute interface.
* @param callback Callback.
+ * @param skip_schema Indicates whether we should skip schema on read.
*/
explicit response_handler_compute(
const std::shared_ptr<compute_impl> &compute, ignite_callback<job_execution> callback, bool skip_schema)
@@ -207,6 +211,29 @@ public:
return res;
}
+ /**
+ * Processes a job execution result, submitting it to job_execution.
+ *
+ * @return Operation result.
+ */
+ ignite_result<void> process_job_result() {
+ assert(m_result_received);
+
+ this->m_handling_complete = true;
+ if (m_read_result.has_error()) {
+ m_execution->set_error(m_read_result.error());
+
+ return m_read_result;
+ }
+
+ auto handle_res = result_of_operation<void>([&]() {
+ m_execution->set_result(m_execution_result);
+ m_execution->set_final_state(m_final_state);
+ });
+
+ return handle_res;
+ }
+
/**
* Handle response.
*
@@ -214,54 +241,41 @@ public:
* @param flags Flags.
*/
[[nodiscard]] ignite_result<void> handle(
- std::shared_ptr<node_connection>, bytes_view msg, std::int32_t flags) final {
+ std::shared_ptr<node_connection>, bytes_view msg, std::int32_t flags) override final {
protocol::reader reader(msg);
if (!test_flag(flags, protocol::response_flag::NOTIFICATION_FLAG)) {
+ std::lock_guard<std::mutex> guard(m_state_mutex);
auto read_res = result_of_operation<job_execution>([&]() {
if (m_skip_schema)
reader.skip();
auto id = reader.read_uuid();
- m_execution = std::make_shared<job_execution_impl>(id, std::move(m_compute));
+ auto node = read_cluster_node(reader);
+ m_execution = std::make_shared<job_execution_impl>(id, std::move(node), std::move(m_compute));
return job_execution{m_execution};
});
auto handle_res = result_of_operation<void>([&]() { this->m_callback(std::move(read_res)); });
- if (read_res.has_error()) {
- return ignite_result<void>{std::move(read_res).error()};
- }
-
- return handle_res;
- } else {
- this->m_handling_complete = true;
-
- std::optional<primitive> res{};
- job_state state;
-
- auto read_res = result_of_operation<void>([&]() {
- res = unpack_compute_result(reader);
- state = read_job_state(reader);
- });
-
- if (!m_execution) {
- m_execution = std::make_shared<job_execution_impl>(state.id, std::move(m_compute));
- result_of_operation<void>([&]() { this->m_callback(job_execution{m_execution}); });
+ if (handle_res.has_error() || !m_result_received) {
+ return handle_res;
}
- if (read_res.has_error()) {
- m_execution->set_error(read_res.error());
-
- return read_res;
- }
+ return process_job_result();
+ }
- auto handle_res = result_of_operation<void>([&]() {
- m_execution->set_result(res);
- m_execution->set_final_state(state);
- });
+ std::lock_guard<std::mutex> guard(m_state_mutex);
+ m_read_result = result_of_operation<void>([&]() {
+ m_execution_result = unpack_compute_result(reader);
+ m_final_state = read_job_state(reader);
+ });
+ m_result_received = true;
- return handle_res;
+ if (!m_execution) {
+ return {};
}
+
+ return process_job_result();
}
private:
@@ -273,9 +287,24 @@ private:
/** Execution. */
std::shared_ptr<job_execution_impl> m_execution{};
+
+ /** Indicates that we've already got a result notification. */
+ bool m_result_received{false};
+
+ /** Execution read result. */
+ ignite_result<void> m_read_result{};
+
+ /** Execution result. */
+ std::optional<primitive> m_execution_result{};
+
+ /** Final job stat. */
+ job_state m_final_state;
+
+ /** State mutex. */
+ std::mutex m_state_mutex;
};
-void compute_impl::submit_to_nodes(const std::vector<cluster_node> &nodes, std::shared_ptr<job_descriptor> descriptor,
+void compute_impl::submit_to_nodes(const std::set<cluster_node> &nodes, std::shared_ptr<job_descriptor> descriptor,
const binary_object &arg, ignite_callback<job_execution> callback) {
auto writer_func = [&nodes, &descriptor, arg](protocol::writer &writer) {
@@ -299,25 +328,24 @@ void compute_impl::submit_to_nodes(const std::vector<cluster_node> &nodes, std::
protocol::client_operation::COMPUTE_EXECUTE, nullptr, writer_func, std::move(handler));
}
-void compute_impl::submit_colocated_async(const std::string &table_name, const ignite_tuple &key,
- std::shared_ptr<job_descriptor> descriptor, const binary_object &arg,
- ignite_callback<job_execution> callback) {
+void compute_impl::submit_colocated_async(const colocated_job_target &target,
+ std::shared_ptr<job_descriptor> descriptor, const binary_object &arg, ignite_callback<job_execution> callback) {
auto self = shared_from_this();
auto conn = m_connection;
- auto on_table_get = [self, table_name, key, descriptor, arg, conn, callback](auto &&res) mutable {
+ auto on_table_get = [self, target, descriptor, arg, conn, callback](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 + "'")});
+ callback({ignite_error("Table does not exist: '" + target.get_table_name() + "'")});
return;
}
auto table = table_impl::from_facade(*table_opt);
table->template with_proper_schema_async<job_execution>(
- callback, [self, table, key, descriptor, arg, conn](const schema &sch, auto callback) mutable {
+ callback, [self, table, key = target.get_key(), descriptor, arg, conn](const schema &sch, auto callback) mutable {
auto writer_func = [&key, &descriptor, &sch, &table, &arg](protocol::writer &writer) {
writer.write(table->get_id());
writer.write(sch.version);
@@ -338,7 +366,7 @@ void compute_impl::submit_colocated_async(const std::string &table_name, const i
});
};
- m_tables->get_table_async(table_name, std::move(on_table_get));
+ m_tables->get_table_async(target.get_table_name(), std::move(on_table_get));
}
void compute_impl::get_state_async(uuid id, ignite_callback<std::optional<job_state>> callback) {
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 9df6736e333..62f3565faca 100644
--- a/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.h
+++ b/modules/platforms/cpp/ignite/client/detail/compute/compute_impl.h
@@ -17,22 +17,20 @@
#pragma once
-#include "ignite/client/compute/deployment_unit.h"
#include "ignite/client/compute/job_descriptor.h"
#include "ignite/client/compute/job_execution.h"
-#include "ignite/client/compute/job_execution_options.h"
#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/table/ignite_tuple.h"
#include "ignite/common/binary_object.h"
#include "ignite/common/ignite_result.h"
-#include "ignite/common/primitive.h"
#include <memory>
+#include <set>
#include <utility>
namespace ignite::detail {
+class colocated_job_target;
/**
* Ignite Compute implementation.
@@ -59,22 +57,20 @@ public:
* @param arg Job argument.
* @param callback A callback called on operation completion with job execution result.
*/
- void submit_to_nodes(const std::vector<cluster_node> &nodes, std::shared_ptr<job_descriptor> descriptor,
+ void submit_to_nodes(const std::set<cluster_node> &nodes, std::shared_ptr<job_descriptor> descriptor,
const binary_object &arg, ignite_callback<job_execution> callback);
/**
* Submits a compute job represented by the given class for an execution on one of the nodes where the given key is
* located.
*
- * @param table_name 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 target Target.
* @param descriptor Descriptor.
* @param arg Job argument.
* @param callback A callback called on operation completion with job execution result.
*/
- void submit_colocated_async(const std::string &table_name, const ignite_tuple &key,
- std::shared_ptr<job_descriptor> descriptor, const binary_object &arg,
- ignite_callback<job_execution> callback);
+ void submit_colocated_async(const colocated_job_target &target, std::shared_ptr<job_descriptor> descriptor,
+ const binary_object &arg, ignite_callback<job_execution> callback);
/**
* Gets the job execution state. Can be @c nullopt if the job state no longer exists due to exceeding the
diff --git a/modules/platforms/cpp/ignite/client/detail/compute/job_execution_impl.h b/modules/platforms/cpp/ignite/client/detail/compute/job_execution_impl.h
index 7add7985cc6..59d0b7c0109 100644
--- a/modules/platforms/cpp/ignite/client/detail/compute/job_execution_impl.h
+++ b/modules/platforms/cpp/ignite/client/detail/compute/job_execution_impl.h
@@ -20,7 +20,6 @@
#include "ignite/client/compute/job_execution.h"
#include "ignite/client/compute/job_state.h"
#include "ignite/client/detail/cluster_connection.h"
-#include "ignite/common/detail/config.h"
#include "ignite/common/ignite_result.h"
#include "ignite/common/primitive.h"
#include "ignite/common/uuid.h"
@@ -41,10 +40,12 @@ public:
* Constructor
*
* @param id Job ID.
+ * @param node Cluster node.
* @param compute Compute.
*/
- explicit job_execution_impl(uuid id, std::shared_ptr<compute_impl> &&compute)
+ explicit job_execution_impl(uuid id, cluster_node node, std::shared_ptr<compute_impl> &&compute)
: m_id(id)
+ , m_node(std::move(node))
, m_compute(compute) {}
/**
@@ -54,10 +55,17 @@ public:
*/
[[nodiscard]] uuid get_id() const { return m_id; }
+ /**
+ * Gets the cluster node.
+ *
+ * @return Cluster node.
+ */
+ [[nodiscard]] const cluster_node &get_node() const { return m_node; }
+
/**
* Gets the job execution result asynchronously.
*
- * Only one callback can be submitted for this operation at a time, which means you can not call this method in
+ * Only one callback can be submitted for this operation at a time, which means you cannot call this method in
* parallel.
* @param callback Callback to be called when the operation is complete. Called with the job execution result.
*/
@@ -74,8 +82,8 @@ public:
* Gets the job execution state. Can be @c nullopt if the job state no longer exists due to exceeding the
* retention time limit.
*
- * @param callback Callback to be called when the operation is complete. Contains the job state. Can be @c nullopt
- * if the job state no longer exists due to exceeding the retention time limit.
+ * @param callback Callback to be called when the operation is complete. Contains the job state. It Can be
+ * @c nullopt if the job state no longer exists due to exceeding the retention time limit.
*/
void get_state_async(ignite_callback<std::optional<job_state>> callback);
@@ -96,16 +104,16 @@ public:
/**
* Cancels the job execution.
*
- * @param callback Callback to be called when the operation is complete. Contains cancel result.
+ * @param callback Callback to be called when the operation is complete. Contains a cancel result.
*/
void cancel_async(ignite_callback<job_execution::operation_result> callback);
/**
- * Changes the job priority. After priority change the job will be the last in the queue of jobs with the same
+ * Changes the job priority. After priority change, the job will be the last in the queue of jobs with the same
* priority.
*
* @param priority New priority.
- * @param callback Callback to be called when the operation is complete. Contains operation result.
+ * @param callback Callback to be called when the operation is complete. Contains an operation result.
*/
void change_priority_async(std::int32_t priority, ignite_callback<job_execution::operation_result> callback);
@@ -113,6 +121,9 @@ private:
/** Job ID. */
const uuid m_id;
+ /** Cluster node. */
+ const cluster_node m_node;
+
/** Compute. */
std::shared_ptr<compute_impl> m_compute;
diff --git a/modules/platforms/cpp/ignite/client/detail/compute/job_target_type.h b/modules/platforms/cpp/ignite/client/detail/compute/job_target_type.h
new file mode 100644
index 00000000000..1f1f7c7aecc
--- /dev/null
+++ b/modules/platforms/cpp/ignite/client/detail/compute/job_target_type.h
@@ -0,0 +1,31 @@
+/*
+ * 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
+
+namespace ignite::detail {
+
+/**
+ * Job target type.
+ */
+enum class job_target_type {
+ ANY_NODE,
+
+ COLOCATED,
+};
+
+} // namespace ignite::detail
diff --git a/modules/platforms/cpp/ignite/client/detail/compute/nodes_broadcast_job_target.h b/modules/platforms/cpp/ignite/client/detail/compute/nodes_broadcast_job_target.h
new file mode 100644
index 00000000000..689e522eb62
--- /dev/null
+++ b/modules/platforms/cpp/ignite/client/detail/compute/nodes_broadcast_job_target.h
@@ -0,0 +1,52 @@
+/*
+ * 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/compute/broadcast_job_target.h"
+
+namespace ignite::detail {
+
+/**
+ * Job target represented by a set of nodes.
+ */
+class nodes_broadcast_job_target : public broadcast_job_target {
+public:
+ // Default
+ nodes_broadcast_job_target() = default;
+
+ /**
+ * Constructor.
+ *
+ * @param nodes Nodes.
+ */
+ explicit nodes_broadcast_job_target(std::set<cluster_node> &&nodes)
+ : m_nodes(std::move(nodes)) {}
+
+ /**
+ * Get nodes.
+ *
+ * @return Nodes.
+ */
+ [[nodiscard]] const std::set<cluster_node> &get_nodes() const { return m_nodes; }
+
+private:
+ /** Nodes. */
+ std::set<cluster_node> m_nodes;
+};
+
+} // namespace ignite::detail
diff --git a/modules/platforms/cpp/ignite/client/detail/ignite_client_impl.cpp b/modules/platforms/cpp/ignite/client/detail/ignite_client_impl.cpp
index dfcb7b292ff..9d1254f7d94 100644
--- a/modules/platforms/cpp/ignite/client/detail/ignite_client_impl.cpp
+++ b/modules/platforms/cpp/ignite/client/detail/ignite_client_impl.cpp
@@ -16,11 +16,11 @@
*/
#include "ignite/client/detail/ignite_client_impl.h"
-
-#include <ignite/protocol/utils.h>
+#include "ignite/client/detail/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;
@@ -28,15 +28,7 @@ void ignite_client_impl::get_cluster_nodes_async(ignite_callback<std::vector<clu
nodes.reserve(std::size_t(size));
for (std::int32_t node_idx = 0; node_idx < size; ++node_idx) {
- auto fields_count = reader.read_int32();
- assert(fields_count >= 4);
-
- auto id = reader.read_uuid();
- auto name = reader.read_string();
- auto host = reader.read_string();
- auto port = reader.read_uint16();
-
- nodes.emplace_back(std::move(id), std::move(name), end_point{std::move(host), port});
+ nodes.emplace_back(read_cluster_node(reader));
}
return nodes;
diff --git a/modules/platforms/cpp/ignite/client/detail/utils.cpp b/modules/platforms/cpp/ignite/client/detail/utils.cpp
index e48e9fd6d24..273498e1304 100644
--- a/modules/platforms/cpp/ignite/client/detail/utils.cpp
+++ b/modules/platforms/cpp/ignite/client/detail/utils.cpp
@@ -339,4 +339,18 @@ std::vector<std::optional<ignite_tuple>> read_tuples_opt(protocol::reader &reade
return res;
}
+cluster_node read_cluster_node(protocol::reader &reader) {
+ auto fields_count = reader.read_int32();
+ assert(fields_count >= 4);
+
+ auto id = reader.read_uuid();
+ auto name = reader.read_string();
+ auto host = reader.read_string();
+ auto port = reader.read_uint16();
+
+ reader.skip(fields_count - 4);
+
+ return {std::move(id), std::move(name), end_point{std::move(host), port}};
+}
+
} // namespace ignite::detail
diff --git a/modules/platforms/cpp/ignite/client/detail/utils.h b/modules/platforms/cpp/ignite/client/detail/utils.h
index 808f490f0be..710c1e89c44 100644
--- a/modules/platforms/cpp/ignite/client/detail/utils.h
+++ b/modules/platforms/cpp/ignite/client/detail/utils.h
@@ -17,21 +17,19 @@
#pragma once
+#include "ignite/client/network/cluster_node.h"
#include "ignite/client/detail/table/schema.h"
#include "ignite/client/table/ignite_tuple.h"
#include "ignite/client/transaction/transaction.h"
-#include "ignite/common/primitive.h"
#include "ignite/protocol/writer.h"
-#include "ignite/tuple/binary_tuple_builder.h"
-#include "ignite/tuple/binary_tuple_parser.h"
namespace ignite::detail {
/**
* Tuple concatenation function.
*
- * @param left Left hand value.
- * @param right Right hand value.
+ * @param left Left-hand value.
+ * @param right Right-hand value.
* @return Resulting tuple.
*/
[[nodiscard]] ignite_tuple concat(const ignite_tuple &left, const ignite_tuple &right);
@@ -42,7 +40,7 @@ namespace ignite::detail {
* @param writer Writer.
* @param sch Schema.
* @param tuple Tuple.
- * @param key_only Should only key fields be written or not.
+ * @param key_only Indicates whether only key fields should be written or not.
*/
void write_tuple(protocol::writer &writer, const schema &sch, const ignite_tuple &tuple, bool key_only);
@@ -52,7 +50,7 @@ void write_tuple(protocol::writer &writer, const schema &sch, const ignite_tuple
* @param writer Writer.
* @param sch Schema.
* @param tuples Tuples.
- * @param key_only Should only key fields be written or not.
+ * @param key_only Indicates whether only key fields should be written or not.
*/
void write_tuples(protocol::writer &writer, const schema &sch, const std::vector<ignite_tuple> &tuples, bool key_only);
@@ -61,7 +59,7 @@ void write_tuples(protocol::writer &writer, const schema &sch, const std::vector
*
* @param reader Reader.
* @param sch Schema.
- * @param key_only Should only key fields be read or not.
+ * @param key_only Indicates whether only key fields should be written or not.
* @return Tuple.
*/
ignite_tuple read_tuple(protocol::reader &reader, const schema *sch, bool key_only);
@@ -80,7 +78,7 @@ std::optional<ignite_tuple> read_tuple_opt(protocol::reader &reader, const schem
*
* @param reader Reader.
* @param sch Schema.
- * @param key_only Should only key fields be read or not.
+ * @param key_only Indicates whether only key fields should be written or not.
* @return Tuples.
*/
std::vector<ignite_tuple> read_tuples(protocol::reader &reader, const schema *sch, bool key_only);
@@ -90,9 +88,17 @@ std::vector<ignite_tuple> read_tuples(protocol::reader &reader, const schema *sc
*
* @param reader Reader.
* @param sch Schema.
- * @param key_only Should only key fields be read or not.
+ * @param key_only Indicates whether only key fields should be written or not.
* @return Tuples.
*/
std::vector<std::optional<ignite_tuple>> read_tuples_opt(protocol::reader &reader, const schema *sch, bool key_only);
+/**
+ * Read cluster node.
+ *
+ * @param reader Reader.
+ * @return Cluster node.
+ */
+cluster_node read_cluster_node(protocol::reader &reader);
+
} // namespace ignite::detail
diff --git a/modules/platforms/cpp/ignite/client/network/cluster_node.h b/modules/platforms/cpp/ignite/client/network/cluster_node.h
index 1c4ffee887b..7ef1500f18f 100644
--- a/modules/platforms/cpp/ignite/client/network/cluster_node.h
+++ b/modules/platforms/cpp/ignite/client/network/cluster_node.h
@@ -18,11 +18,9 @@
#pragma once
#include "ignite/common/end_point.h"
+#include "ignite/common/uuid.h"
-#include <cstdint>
#include <type_traits>
-#include <variant>
-#include <vector>
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 79a24d38651..bb546403b80 100644
--- a/modules/platforms/cpp/tests/client-test/compute_test.cpp
+++ b/modules/platforms/cpp/tests/client-test/compute_test.cpp
@@ -71,13 +71,13 @@ protected:
template<typename T>
void check_argument(T value, const std::string &expected_str) {
auto cluster_nodes = m_client.get_cluster_nodes();
- auto execution = m_client.get_compute().submit(cluster_nodes, m_echo_job, {value});
+ auto execution = m_client.get_compute().submit(job_target::any_node(cluster_nodes), m_echo_job, {value});
auto result = execution.get_result();
ASSERT_TRUE(result.has_value());
EXPECT_EQ(result.value().get_primitive().template get<T>(), value);
- execution = m_client.get_compute().submit(cluster_nodes, m_to_string_job, {value});
+ execution = m_client.get_compute().submit(job_target::any_node(cluster_nodes), m_to_string_job, {value});
result = execution.get_result();
ASSERT_TRUE(result.has_value());
@@ -139,7 +139,7 @@ TEST_F(compute_test, get_cluster_nodes) {
TEST_F(compute_test, execute_on_random_node) {
auto cluster_nodes = m_client.get_cluster_nodes();
- auto execution = m_client.get_compute().submit(cluster_nodes, m_node_name_job, {});
+ auto execution = m_client.get_compute().submit(job_target::any_node(cluster_nodes), m_node_name_job, {});
auto result = execution.get_result();
ASSERT_TRUE(result.has_value());
@@ -147,8 +147,8 @@ TEST_F(compute_test, execute_on_random_node) {
}
TEST_F(compute_test, execute_on_specific_node) {
- auto execution1 = m_client.get_compute().submit({get_node(0)}, m_node_name_job, {"-11"});
- auto execution2 = m_client.get_compute().submit({get_node(1)}, m_node_name_job, {42});
+ auto execution1 = m_client.get_compute().submit(job_target::node(get_node(0)), m_node_name_job, {"-11"});
+ auto execution2 = m_client.get_compute().submit(job_target::node(get_node(1)), m_node_name_job, {42});
auto res1 = execution1.get_result();
auto res2 = execution2.get_result();
@@ -161,25 +161,29 @@ TEST_F(compute_test, execute_on_specific_node) {
}
TEST_F(compute_test, execute_broadcast_one_node) {
- auto res = m_client.get_compute().submit_broadcast({get_node(1)}, m_node_name_job, {"42"});
+ auto res = m_client.get_compute().submit_broadcast(broadcast_job_target::node(get_node(1)), m_node_name_job, {"42"});
+ auto execs = res.get_job_executions();
- ASSERT_EQ(res.size(), 1);
+ ASSERT_EQ(execs.size(), 1);
- EXPECT_EQ(res.begin()->first, get_node(1));
-
- ASSERT_TRUE(res.begin()->second.has_value());
- EXPECT_EQ(res.begin()->second.value().get_result()->get_primitive(), PLATFORM_TEST_NODE_RUNNER + "_242");
+ ASSERT_TRUE(execs.front().has_value());
+ EXPECT_EQ(execs.front().value().get_result()->get_primitive(), PLATFORM_TEST_NODE_RUNNER + "_242");
}
TEST_F(compute_test, execute_broadcast_all_nodes) {
- auto res = m_client.get_compute().submit_broadcast(get_node_set(), m_node_name_job, {"42"});
+ auto res = m_client.get_compute().submit_broadcast(broadcast_job_target::nodes(get_node_set()), m_node_name_job, {"42"});
+ auto execs = res.get_job_executions();
+
+ ASSERT_EQ(execs.size(), 4);
- ASSERT_EQ(res.size(), 4);
+ std::sort(execs.begin(), execs.end(), [] (auto &n1, auto &n2) {
+ return n1.value().get_node().get_name() < n2.value().get_node().get_name();
+ });
- EXPECT_EQ(res[get_node(0)].value().get_result()->get_primitive(), get_node(0).get_name() + "42");
- EXPECT_EQ(res[get_node(1)].value().get_result()->get_primitive(), get_node(1).get_name() + "42");
- EXPECT_EQ(res[get_node(2)].value().get_result()->get_primitive(), get_node(2).get_name() + "42");
- EXPECT_EQ(res[get_node(3)].value().get_result()->get_primitive(), get_node(3).get_name() + "42");
+ EXPECT_EQ(execs[0].value().get_result()->get_primitive().get<std::string>(), get_node(0).get_name() + "42");
+ EXPECT_EQ(execs[1].value().get_result()->get_primitive().get<std::string>(), get_node(1).get_name() + "42");
+ EXPECT_EQ(execs[2].value().get_result()->get_primitive().get<std::string>(), get_node(2).get_name() + "42");
+ EXPECT_EQ(execs[3].value().get_result()->get_primitive().get<std::string>(), get_node(3).get_name() + "42");
}
TEST_F(compute_test, job_error_propagates_to_client) {
@@ -188,7 +192,7 @@ TEST_F(compute_test, job_error_propagates_to_client) {
EXPECT_THROW(
{
try {
- m_client.get_compute().submit(cluster_nodes, m_error_job, {"unused"}).get_result();
+ m_client.get_compute().submit(job_target::any_node(cluster_nodes), m_error_job, {"unused"}).get_result();
} catch (const ignite_error &e) {
EXPECT_THAT(e.what_str(), testing::HasSubstr("Custom job error"));
// TODO https://issues.apache.org/jira/browse/IGNITE-19603
@@ -208,7 +212,7 @@ TEST_F(compute_test, unknown_node_execute_throws) {
EXPECT_THROW(
{
try {
- m_client.get_compute().submit({unknown_node}, m_echo_job, {"unused"});
+ m_client.get_compute().submit(job_target::node(unknown_node), m_echo_job, {"unused"});
} catch (const ignite_error &e) {
EXPECT_THAT(e.what_str(),
testing::HasSubstr("None of the specified nodes are present in the cluster: [random]"));
@@ -225,7 +229,7 @@ TEST_F(compute_test, DISABLED_unknown_node_broadcast_throws) {
EXPECT_THROW(
{
try {
- m_client.get_compute().submit_broadcast({unknown_node}, m_echo_job, {"unused"});
+ m_client.get_compute().submit_broadcast(broadcast_job_target::node(unknown_node), m_echo_job, {"unused"});
} catch (const ignite_error &e) {
EXPECT_THAT(e.what_str(),
testing::HasSubstr("None of the specified nodes are present in the cluster: [random]"));
@@ -286,7 +290,7 @@ TEST_F(compute_test, submit_colocated) {
SCOPED_TRACE("key=" + std::to_string(var.first) + ", node=" + var.second);
auto key = get_tuple(var.first);
- auto execution = m_client.get_compute().submit_colocated(TABLE_1, key, m_node_name_job, {});
+ auto execution = m_client.get_compute().submit(job_target::colocated(TABLE_1, key), m_node_name_job, {});
auto res_node_name = execution.get_result();
auto expected_node_name = PLATFORM_TEST_NODE_RUNNER + var.second;
@@ -298,7 +302,7 @@ TEST_F(compute_test, execute_colocated_throws_when_table_does_not_exist) {
EXPECT_THROW(
{
try {
- (void) m_client.get_compute().submit_colocated("unknownTable", get_tuple(42), m_echo_job, {});
+ (void) m_client.get_compute().submit(job_target::colocated("unknownTable", get_tuple(42)), m_echo_job, {});
} catch (const ignite_error &e) {
EXPECT_STREQ("Table does not exist: 'unknownTable'", e.what());
throw;
@@ -311,7 +315,7 @@ TEST_F(compute_test, execute_colocated_throws_when_key_column_is_missing) {
EXPECT_THROW(
{
try {
- (void) m_client.get_compute().submit_colocated(TABLE_1, get_tuple("some"), m_echo_job, {});
+ (void) m_client.get_compute().submit(job_target::colocated(TABLE_1, get_tuple("some")), m_echo_job, {});
} catch (const ignite_error &e) {
EXPECT_THAT(e.what_str(), ::testing::HasSubstr("Missed key column: KEY"));
throw;
@@ -324,7 +328,7 @@ TEST_F(compute_test, execute_colocated_throws_when_key_is_empty) {
EXPECT_THROW(
{
try {
- (void) m_client.get_compute().submit_colocated(TABLE_1, {}, m_echo_job, {});
+ (void) m_client.get_compute().submit(job_target::colocated(TABLE_1, {}), m_echo_job, {});
} catch (const ignite_error &e) {
EXPECT_EQ("Key tuple can not be empty", e.what_str());
throw;
@@ -342,7 +346,7 @@ TEST_F(compute_test, unknown_unit) {
.deployment_units({{"unknown"}})
.build();
- (void) m_client.get_compute().submit(cluster_nodes, job_desc, {});
+ (void) m_client.get_compute().submit(job_target::any_node(cluster_nodes), job_desc, {});
} catch (const ignite_error &e) {
EXPECT_THAT(e.what_str(), ::testing::HasSubstr("Deployment unit unknown:latest doesn't exist"));
throw;
@@ -360,7 +364,7 @@ TEST_F(compute_test, execute_unknown_unit_and_version) {
.deployment_units({{"unknown", "1.2.3"}})
.build();
- (void) m_client.get_compute().submit(cluster_nodes, job_desc, {});
+ (void) m_client.get_compute().submit(job_target::any_node(cluster_nodes), job_desc, {});
} catch (const ignite_error &e) {
EXPECT_THAT(e.what_str(), ::testing::HasSubstr("Deployment unit unknown:1.2.3 doesn't exist"));
throw;
@@ -378,7 +382,7 @@ TEST_F(compute_test, execute_colocated_unknown_unit_and_version) {
.deployment_units({{"unknown", "1.2.3"}})
.build();
- (void) comp.submit_colocated(TABLE_1, get_tuple(1), job_desc, {});
+ (void) comp.submit(job_target::colocated(TABLE_1, get_tuple(1)), job_desc, {});
} catch (const ignite_error &e) {
EXPECT_THAT(e.what_str(), ::testing::HasSubstr("Deployment unit unknown:1.2.3 doesn't exist"));
throw;
@@ -392,13 +396,14 @@ TEST_F(compute_test, broadcast_unknown_unit_and_version) {
.deployment_units({{"unknown", "1.2.3"}})
.build();
- auto res = m_client.get_compute().submit_broadcast({get_node(1)}, job_desc, {});
+ auto res = m_client.get_compute().submit_broadcast(broadcast_job_target::node(get_node(1)), job_desc, {});
+ auto execs = res.get_job_executions();
- ASSERT_EQ(res.size(), 1);
+ ASSERT_EQ(execs.size(), 1);
- auto &res1 = res[get_node(1)];
- ASSERT_TRUE(res1.has_error());
- EXPECT_THAT(res1.error().what_str(), ::testing::HasSubstr("Deployment unit unknown:1.2.3 doesn't exist"));
+ auto &exec1 = execs.front();
+ ASSERT_TRUE(exec1.has_error());
+ EXPECT_THAT(exec1.error().what_str(), ::testing::HasSubstr("Deployment unit unknown:1.2.3 doesn't exist"));
}
TEST_F(compute_test, execute_empty_unit_name) {
@@ -409,7 +414,7 @@ TEST_F(compute_test, execute_empty_unit_name) {
.deployment_units({{""}})
.build();
- (void) m_client.get_compute().submit({get_node(1)}, job_desc, {});
+ (void) m_client.get_compute().submit(job_target::node(get_node(1)), job_desc, {});
} catch (const ignite_error &e) {
EXPECT_EQ("Deployment unit name can not be empty", e.what_str());
throw;
@@ -426,7 +431,7 @@ TEST_F(compute_test, execute_empty_unit_version) {
.deployment_units({{"some", ""}})
.build();
- (void) m_client.get_compute().submit({get_node(1)}, job_desc, {});
+ (void) m_client.get_compute().submit(job_target::node(get_node(1)), job_desc, {});
} catch (const ignite_error &e) {
EXPECT_EQ("Deployment unit version can not be empty", e.what_str());
throw;
@@ -443,7 +448,7 @@ TEST_F(compute_test, broadcast_empty_unit_name) {
.deployment_units({{""}})
.build();
- (void) m_client.get_compute().submit_broadcast({get_node(1)}, job_desc, {});
+ (void) m_client.get_compute().submit_broadcast(broadcast_job_target::node(get_node(1)), job_desc, {});
} catch (const ignite_error &e) {
EXPECT_EQ("Deployment unit name can not be empty", e.what_str());
throw;
@@ -460,7 +465,7 @@ TEST_F(compute_test, broadcast_empty_unit_version) {
.deployment_units({{"some", ""}})
.build();
- (void) m_client.get_compute().submit_broadcast({get_node(1)}, job_desc, {});
+ (void) m_client.get_compute().submit_broadcast(broadcast_job_target::node(get_node(1)), job_desc, {});
} catch (const ignite_error &e) {
EXPECT_EQ("Deployment unit version can not be empty", e.what_str());
throw;
@@ -477,7 +482,7 @@ TEST_F(compute_test, execute_colocated_empty_unit_name) {
.deployment_units({{""}})
.build();
- (void) m_client.get_compute().submit_colocated(TABLE_1, get_tuple(1), job_desc, {});
+ (void) m_client.get_compute().submit(job_target::colocated(TABLE_1, get_tuple(1)), job_desc, {});
} catch (const ignite_error &e) {
EXPECT_EQ("Deployment unit name can not be empty", e.what_str());
throw;
@@ -495,7 +500,7 @@ TEST_F(compute_test, execute_colocated_empty_unit_version) {
.deployment_units({{"some", ""}})
.build();
- comp.submit_colocated(TABLE_1, get_tuple(1), job_desc, {});
+ comp.submit(job_target::colocated(TABLE_1, get_tuple(1)), job_desc, {});
} catch (const ignite_error &e) {
EXPECT_EQ("Deployment unit version can not be empty", e.what_str());
throw;
@@ -507,7 +512,7 @@ TEST_F(compute_test, execute_colocated_empty_unit_version) {
TEST_F(compute_test, job_execution_status_executing) {
const std::int32_t sleep_ms = 3000;
- auto execution = m_client.get_compute().submit({get_node(1)}, m_sleep_job, {sleep_ms});
+ auto execution = m_client.get_compute().submit(job_target::node(get_node(1)), m_sleep_job, {sleep_ms});
auto state = execution.get_state();
@@ -518,7 +523,7 @@ TEST_F(compute_test, job_execution_status_executing) {
TEST_F(compute_test, DISABLED_job_execution_status_completed) {
const std::int32_t sleep_ms = 1;
- auto execution = m_client.get_compute().submit({get_node(1)}, m_sleep_job, {sleep_ms});
+ auto execution = m_client.get_compute().submit(job_target::node(get_node(1)), m_sleep_job, {sleep_ms});
execution.get_result();
auto state = execution.get_state();
@@ -528,7 +533,7 @@ TEST_F(compute_test, DISABLED_job_execution_status_completed) {
}
TEST_F(compute_test, job_execution_status_failed) {
- auto execution = m_client.get_compute().submit({get_node(1)}, m_error_job, {"unused"});
+ auto execution = m_client.get_compute().submit(job_target::node(get_node(1)), m_error_job, {"unused"});
EXPECT_THROW(
{
@@ -548,9 +553,9 @@ TEST_F(compute_test, job_execution_status_failed) {
}
TEST_F(compute_test, job_execution_cancel) {
- const std::int32_t sleep_ms = 5000;
+ constexpr std::int32_t sleep_ms = 5000;
- auto execution = m_client.get_compute().submit({get_node(1)}, m_sleep_job, {sleep_ms});
+ auto execution = m_client.get_compute().submit(job_target::node(get_node(1)), m_sleep_job, {sleep_ms});
execution.cancel();
auto state = execution.get_state();
@@ -560,9 +565,9 @@ TEST_F(compute_test, job_execution_cancel) {
}
TEST_F(compute_test, job_execution_change_priority) {
- const std::int32_t sleep_ms = 5000;
+ constexpr std::int32_t sleep_ms = 5000;
- auto execution = m_client.get_compute().submit({get_node(1)}, m_sleep_job, {sleep_ms});
+ auto execution = m_client.get_compute().submit(job_target::node(get_node(1)), m_sleep_job, {sleep_ms});
auto res = execution.change_priority(123);
EXPECT_EQ(res, job_execution::operation_result::INVALID_STATE);
diff --git a/modules/platforms/cpp/tests/client-test/schema_synchronization_test.cpp b/modules/platforms/cpp/tests/client-test/schema_synchronization_test.cpp
index 690ef7df231..c5a1027d3e2 100644
--- a/modules/platforms/cpp/tests/client-test/schema_synchronization_test.cpp
+++ b/modules/platforms/cpp/tests/client-test/schema_synchronization_test.cpp
@@ -100,7 +100,7 @@ TEST_F(schema_synchronization_test, upsert_add_column_compute) {
m_client.get_sql().execute(nullptr, {"ALTER TABLE SCHEMA_SYN_TEST ADD COLUMN VAL2 INT"}, {});
auto descriptor = job_descriptor::builder(NODE_NAME_JOB).build();
- m_client.get_compute().submit_colocated("SCHEMA_SYN_TEST", {key}, descriptor, {}).get_result();
+ m_client.get_compute().submit(job_target::colocated("SCHEMA_SYN_TEST", {key}), descriptor, {}).get_result();
}
TEST_F(schema_synchronization_test, upsert_add_column_upsert_all) {
diff --git a/modules/platforms/cpp/tests/test-common/basic_auth_test_suite.h b/modules/platforms/cpp/tests/test-common/basic_auth_test_suite.h
index 3a039e3bf67..11f781c8787 100644
--- a/modules/platforms/cpp/tests/test-common/basic_auth_test_suite.h
+++ b/modules/platforms/cpp/tests/test-common/basic_auth_test_suite.h
@@ -99,7 +99,7 @@ public:
auto nodes = client.get_cluster_nodes();
auto descriptor = job_descriptor::builder(ENABLE_AUTHN_JOB).build();
- client.get_compute().submit(nodes, descriptor, {enable ? 1 : 0}).get_result();
+ client.get_compute().submit(job_target::any_node(nodes), descriptor, {enable ? 1 : 0}).get_result();
} catch (const ignite_error &) {
// Ignore.
// As a result of this call, the client may be disconnected from the server due to authn config change.