IGNITE-24200 C++ Unify Compute API

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,13 +51,16 @@ 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
     compute/job_execution.h
     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

modules/platforms/cpp/ignite/client/compute/broadcast_execution.h

@@ -0,0 +1,57 @@
+/*
+ * 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_execution.h"
+#include "ignite/common/ignite_result.h"
+
+#include <vector>
+
+namespace ignite {
+
+/**
+ * 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

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

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

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