Intermediate changes

commit_hash:9accadbf8068e1d496f6fc9ea6b878476f55f89b

Author: robot-piglet

yt/yql/providers/yt/fmr/coordinator/impl/yql_yt_coordinator_impl.cpp

@@ -290,6 +290,37 @@ class TFmrCoordinator: public IFmrCoordinator {
         };
     }
 
+    std::vector<TTaskTableRef> TaskInputTablesFromOperationInputTables(const std::vector<TOperationTableRef>& operationTables) {
+        std::vector<TTaskTableRef> taskInputTables;
+        for (auto& elem: operationTables) {
+            if (const TYtTableRef* ytTableRef = std::get_if<TYtTableRef>(&elem)) {
+                taskInputTables.emplace_back(*ytTableRef);
+            } else {
+                TFmrTableRef fmrTableRef = std::get<TFmrTableRef>(elem);
+                TString inputTableId = fmrTableRef.FmrTableId.Id;
+                TFmrTableInputRef tableInput{
+                    .TableId = inputTableId,
+                    .TableRanges = {GetTableRangeFromId(inputTableId)}
+                };
+                taskInputTables.emplace_back(tableInput);
+            }
+        }
+        return taskInputTables;
+    }
+
+    std::vector<TFmrTableOutputRef> TaskOutputTablesFromOperationOutputTables(const std::vector<TFmrTableRef>& operationTables) {
+        std::vector<TFmrTableOutputRef> taskOutputTables;
+        for (auto& fmrTableRef: operationTables) {
+                TString outputTableId = fmrTableRef.FmrTableId.Id;
+                TFmrTableOutputRef tableOutput{
+                    .TableId = outputTableId,
+                    .PartId = GetTableRangeFromId(outputTableId).PartId
+                };
+                taskOutputTables.emplace_back(tableOutput);
+            }
+        return taskOutputTables;
+    }
+
     TTaskParams MakeDefaultTaskParamsFromOperation(const TOperationParams& operationParams) {
         if (const TUploadOperationParams* uploadOperationParams = std::get_if<TUploadOperationParams>(&operationParams)) {
             TUploadTaskParams uploadTaskParams{};
@@ -311,27 +342,20 @@ class TFmrCoordinator: public IFmrCoordinator {
             };
             downloadTaskParams.Output = fmrTableOutput;
             return downloadTaskParams;
-        } else {
-            TMergeOperationParams mergeOperationParams = std::get<TMergeOperationParams>(operationParams);
+        } else if (const TMergeOperationParams* mergeOperationParams = std::get_if<TMergeOperationParams>(&operationParams)) {
             TMergeTaskParams mergeTaskParams;
-            std::vector<TTaskTableRef> mergeInputTasks;
-            for (auto& elem: mergeOperationParams.Input) {
-                if (const TYtTableRef* ytTableRef = std::get_if<TYtTableRef>(&elem)) {
-                    mergeInputTasks.emplace_back(*ytTableRef);
-                } else {
-                    TFmrTableRef fmrTableRef = std::get<TFmrTableRef>(elem);
-                    TString inputTableId = fmrTableRef.FmrTableId.Id;
-                    TFmrTableInputRef tableInput{
-                        .TableId = inputTableId,
-                        .TableRanges = {GetTableRangeFromId(inputTableId)}
-                    };
-                    mergeInputTasks.emplace_back(tableInput);
-                }
-            }
-            mergeTaskParams.Input = mergeInputTasks;
+            mergeTaskParams.Input = TaskInputTablesFromOperationInputTables(mergeOperationParams->Input);
             TFmrTableOutputRef outputTable;
-            mergeTaskParams.Output = TFmrTableOutputRef{.TableId = mergeOperationParams.Output.FmrTableId.Id};
+            mergeTaskParams.Output = TFmrTableOutputRef{.TableId = mergeOperationParams->Output.FmrTableId.Id};
             return mergeTaskParams;
+        } else if (const TMapOperationParams* mapOperationParams = std::get_if<TMapOperationParams>(&operationParams)) {
+            TMapTaskParams mapTaskParams;
+            mapTaskParams.Input = TaskInputTablesFromOperationInputTables(mapOperationParams->Input);
+            mapTaskParams.Output = TaskOutputTablesFromOperationOutputTables(mapOperationParams->Output);
+            mapTaskParams.Executable = mapOperationParams->Executable;
+            return mapTaskParams;
+        } else {
+            ythrow yexception() << "Unknown operation params";
         }
     }
 

yt/yql/providers/yt/fmr/job/impl/yql_yt_job_impl.cpp

@@ -107,6 +107,16 @@ class TFmrJob: public IFmrJob {
         }
     }
 
+    virtual std::variant<TError, TStatistics> Map(
+        const TMapTaskParams& /* params */,
+        const std::unordered_map<TFmrTableId, TClusterConnection>& /* clusterConnections */,
+        std::shared_ptr<std::atomic<bool>> /* cancelFlag */
+    ) override {
+        Cerr << "MAP NOT IMPLEMENTED" << Endl;
+        YQL_CLOG(ERROR, FastMapReduce) << "MAP NOT IMPLEMENTED";
+        ythrow yexception() << "Not implemented";
+    }
+
 private:
     NYT::TRawTableReaderPtr GetTableInputStream(const TTaskTableRef& tableRef, const std::unordered_map<TFmrTableId, TClusterConnection>& clusterConnections) const {
         auto ytTable = std::get_if<TYtTableRef>(&tableRef);
@@ -154,8 +164,10 @@ TJobResult RunJob(
             return job->Download(taskParams, task->ClusterConnections, cancelFlag);
         } else if constexpr (std::is_same_v<T, TMergeTaskParams>) {
             return job->Merge(taskParams, task->ClusterConnections, cancelFlag);
+        } else if constexpr (std::is_same_v<T, TMapTaskParams>) {
+            return job->Map(taskParams, task->ClusterConnections, cancelFlag);;
         } else {
-            throw std::runtime_error{"Unsupported task type"};
+            ythrow yexception() << "Unsupported task type";
         }
     };
 

yt/yql/providers/yt/fmr/job/interface/yql_yt_job.h

@@ -15,6 +15,8 @@ class IFmrJob: public TThrRefBase {
     virtual std::variant<TError, TStatistics> Upload(const TUploadTaskParams& params, const std::unordered_map<TFmrTableId, TClusterConnection>& clusterConnections = {}, std::shared_ptr<std::atomic<bool>> cancelFlag = nullptr) = 0;
 
     virtual std::variant<TError, TStatistics> Merge(const TMergeTaskParams& params, const std::unordered_map<TFmrTableId, TClusterConnection>& clusterConnections = {}, std::shared_ptr<std::atomic<bool>> cancelFlag = nullptr) = 0;
+
+    virtual std::variant<TError, TStatistics> Map(const TMapTaskParams& params, const std::unordered_map<TFmrTableId, TClusterConnection>& clusterConnections = {}, std::shared_ptr<std::atomic<bool>> cancelFlag = nullptr) = 0;
 };
 
 } // namespace NYql

yt/yql/providers/yt/fmr/request_options/yql_yt_request_options.h

@@ -31,7 +31,8 @@ enum class ETaskType {
     Unknown,
     Download,
     Upload,
-    Merge
+    Merge,
+    Map
 };
 
 enum class EFmrComponent {
@@ -183,9 +184,21 @@ struct TMergeTaskParams {
     TFmrTableOutputRef Output;
 };
 
-using TOperationParams = std::variant<TUploadOperationParams, TDownloadOperationParams, TMergeOperationParams>;
+struct TMapOperationParams {
+    std::vector<TOperationTableRef> Input;
+    std::vector<TFmrTableRef> Output;
+    TString Executable;
+};
+
+struct TMapTaskParams {
+    std::vector<TTaskTableRef> Input;
+    std::vector<TFmrTableOutputRef> Output;
+    TString Executable;
+};
+
+using TOperationParams = std::variant<TUploadOperationParams, TDownloadOperationParams, TMergeOperationParams, TMapOperationParams>;
 
-using TTaskParams = std::variant<TUploadTaskParams, TDownloadTaskParams, TMergeTaskParams>;
+using TTaskParams = std::variant<TUploadTaskParams, TDownloadTaskParams, TMergeTaskParams, TMapTaskParams>;
 
 struct TClusterConnection {
     TString TransactionId;