YT-23616: Make BatchTransform implementation being common

commit_hash:4191c9aa7cde449475eddf88d8c04e1ebf0b8ad9

Author: hiddenpath

yt/cpp/mapreduce/client/operation.cpp

@@ -225,11 +225,10 @@ TStructuredJobTableList ApplyProtobufColumnFilters(
     }
 
     auto isDynamic = NRawClient::BatchTransform(
-        CreateDefaultRequestRetryPolicy(preparer.GetContext().Config),
-        preparer.GetContext(),
+        preparer.GetClient()->GetRawClient(),
         tableList,
-        [&] (IRawBatchRequest& batch, const auto& table) {
-            return batch.Get(preparer.GetTransactionId(), table.RichYPath->Path_ + "/@dynamic", TGetOptions());
+        [&] (IRawBatchRequestPtr batch, const auto& table) {
+            return batch->Get(preparer.GetTransactionId(), table.RichYPath->Path_ + "/@dynamic", TGetOptions());
         });
 
     auto newTableList = tableList;
@@ -642,7 +641,7 @@ TNode BuildAutoMergeSpec(const TAutoMergeSpec& options)
     return result;
 }
 
-TNode BuildJobProfilerSpec(const TJobProfilerSpec& profilerSpec)
+[[maybe_unused]] TNode BuildJobProfilerSpec(const TJobProfilerSpec& profilerSpec)
 {
     TNode result;
     if (profilerSpec.ProfilingBinary_) {

yt/cpp/mapreduce/client/skiff.cpp

@@ -279,6 +279,7 @@ TFormat CreateSkiffFormat(const NSkiff::TSkiffSchemaPtr& schema) {
 }
 
 NSkiff::TSkiffSchemaPtr CreateSkiffSchemaIfNecessary(
+    const IRawClientPtr& rawClient,
     const TClientContext& context,
     const IClientRetryPolicyPtr& clientRetryPolicy,
     const TTransactionId& transactionId,
@@ -304,18 +305,17 @@ NSkiff::TSkiffSchemaPtr CreateSkiffSchemaIfNecessary(
     }
 
     auto nodes = NRawClient::BatchTransform(
-        clientRetryPolicy->CreatePolicyForGenericRequest(),
-        context,
+        rawClient,
         NRawClient::CanonizeYPaths(clientRetryPolicy->CreatePolicyForGenericRequest(), context, tablePaths),
-        [&] (IRawBatchRequest& batch, const TRichYPath& path) {
+        [&] (IRawBatchRequestPtr batch, const TRichYPath& path) {
             auto getOptions = TGetOptions()
                 .AttributeFilter(
                     TAttributeFilter()
                         .AddAttribute("schema")
                         .AddAttribute("dynamic")
                         .AddAttribute("type")
                 );
-            return batch.Get(transactionId, path.Path_, getOptions);
+            return batch->Get(transactionId, path.Path_, getOptions);
         });
 
     TVector<NSkiff::TSkiffSchemaPtr> schemas;

yt/cpp/mapreduce/client/skiff.h

@@ -59,6 +59,7 @@ void Deserialize(NSkiff::TSkiffSchemaPtr& schema, const TNode& node);
 TFormat CreateSkiffFormat(const NSkiff::TSkiffSchemaPtr& schema);
 
 NSkiff::TSkiffSchemaPtr CreateSkiffSchemaIfNecessary(
+    const IRawClientPtr& rawClient,
     const TClientContext& context,
     const IClientRetryPolicyPtr& clientRetryPolicy,
     const TTransactionId& transactionId,

yt/cpp/mapreduce/client/structured_table_formats.cpp

@@ -121,6 +121,7 @@ namespace NDetail {
 ////////////////////////////////////////////////////////////////////////////////
 
 NSkiff::TSkiffSchemaPtr TryCreateSkiffSchema(
+    const IRawClientPtr& rawClient,
     const TClientContext& context,
     const IClientRetryPolicyPtr& clientRetryPolicy,
     const TTransactionId& transactionId,
@@ -135,6 +136,7 @@ NSkiff::TSkiffSchemaPtr TryCreateSkiffSchema(
         return nullptr;
     }
     return CreateSkiffSchemaIfNecessary(
+        rawClient,
         context,
         clientRetryPolicy,
         transactionId,
@@ -434,6 +436,7 @@ std::pair<TFormat, TMaybe<TSmallJobFile>> TFormatBuilder::CreateNodeFormat(
             tableList.emplace_back(*table.RichYPath);
         }
         skiffSchema = TryCreateSkiffSchema(
+            RawClient_,
             Context_,
             ClientRetryPolicy_,
             TransactionId_,

yt/cpp/mapreduce/raw_client/raw_requests.h

@@ -9,6 +9,7 @@
 #include <yt/cpp/mapreduce/interface/client.h>
 #include <yt/cpp/mapreduce/interface/client_method_options.h>
 #include <yt/cpp/mapreduce/interface/operation.h>
+#include <yt/cpp/mapreduce/interface/raw_client.h>
 
 namespace NYT {
 
@@ -53,19 +54,18 @@ TAuthorizationInfo WhoAmI(const TClientContext& context);
 
 template<typename TSrc, typename TBatchAdder>
 auto BatchTransform(
-    const IRequestRetryPolicyPtr& retryPolicy,
-    const TClientContext& context,
+    const IRawClientPtr& rawClient,
     const TSrc& src,
     TBatchAdder batchAdder,
     const TExecuteBatchOptions& executeBatchOptions = {})
 {
-    THttpRawBatchRequest batch(context, retryPolicy);
+    auto batch = rawClient->CreateRawBatchRequest();
     using TFuture = decltype(batchAdder(batch, *std::begin(src)));
     TVector<TFuture> futures;
     for (const auto& el : src) {
         futures.push_back(batchAdder(batch, el));
     }
-    batch.ExecuteBatch(executeBatchOptions);
+    batch->ExecuteBatch(executeBatchOptions);
     using TDst = decltype(futures[0].ExtractValueSync());
     TVector<TDst> result;
     result.reserve(std::size(src));