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:51:12 UTC
[ignite-3] 15/19: IGNITE-17607 Execute colocated stub
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: