summaryrefslogtreecommitdiffstats
path: root/yt/cpp/mapreduce/client/operation.cpp
diff options
context:
space:
mode:
authormax42 <[email protected]>2023-06-30 11:13:34 +0300
committermax42 <[email protected]>2023-06-30 11:13:34 +0300
commit3e1899838408bbad47622007aa382bc8a2b01f87 (patch)
tree0f21c1e6add187ddb6c3ccc048a7d640ce03fb87 /yt/cpp/mapreduce/client/operation.cpp
parent5463eb3f5e72a86f858a3d27c886470a724ede34 (diff)
Revert "YT-19324: move YT provider to ydb/library/yql"
This reverts commit ca272f12fdd0e8d5c3e957fc87939148f1caaf72, reversing changes made to 49f8acfc8b0b5c0071b804423bcf53fda26c7c12.
Diffstat (limited to 'yt/cpp/mapreduce/client/operation.cpp')
-rw-r--r--yt/cpp/mapreduce/client/operation.cpp2981
1 files changed, 0 insertions, 2981 deletions
diff --git a/yt/cpp/mapreduce/client/operation.cpp b/yt/cpp/mapreduce/client/operation.cpp
deleted file mode 100644
index fc1600c2402..00000000000
--- a/yt/cpp/mapreduce/client/operation.cpp
+++ /dev/null
@@ -1,2981 +0,0 @@
-#include "operation.h"
-
-#include "abortable_registry.h"
-#include "client.h"
-#include "operation_helpers.h"
-#include "operation_tracker.h"
-#include "transaction.h"
-#include "prepare_operation.h"
-#include "retry_heavy_write_request.h"
-#include "skiff.h"
-#include "structured_table_formats.h"
-#include "yt_poller.h"
-
-#include <yt/cpp/mapreduce/common/helpers.h>
-#include <yt/cpp/mapreduce/common/retry_lib.h>
-#include <yt/cpp/mapreduce/common/wait_proxy.h>
-
-#include <yt/cpp/mapreduce/interface/config.h>
-#include <yt/cpp/mapreduce/interface/errors.h>
-#include <yt/cpp/mapreduce/interface/fluent.h>
-#include <yt/cpp/mapreduce/interface/format.h>
-#include <yt/cpp/mapreduce/interface/job_statistics.h>
-#include <yt/cpp/mapreduce/interface/protobuf_format.h>
-
-#include <yt/cpp/mapreduce/interface/logging/yt_log.h>
-#include <yt/cpp/mapreduce/interface/logging/yt_log.h>
-
-#include <yt/cpp/mapreduce/http/requests.h>
-#include <yt/cpp/mapreduce/http/retry_request.h>
-
-#include <yt/cpp/mapreduce/io/job_reader.h>
-#include <yt/cpp/mapreduce/io/job_writer.h>
-#include <yt/cpp/mapreduce/io/yamr_table_reader.h>
-#include <yt/cpp/mapreduce/io/yamr_table_writer.h>
-#include <yt/cpp/mapreduce/io/node_table_reader.h>
-#include <yt/cpp/mapreduce/io/node_table_writer.h>
-#include <yt/cpp/mapreduce/io/proto_table_reader.h>
-#include <yt/cpp/mapreduce/io/proto_table_writer.h>
-#include <yt/cpp/mapreduce/io/proto_helpers.h>
-#include <yt/cpp/mapreduce/io/skiff_table_reader.h>
-
-#include <yt/cpp/mapreduce/raw_client/raw_batch_request.h>
-#include <yt/cpp/mapreduce/raw_client/raw_requests.h>
-
-#include <library/cpp/yson/node/serialize.h>
-
-#include <util/generic/hash_set.h>
-
-#include <util/string/builder.h>
-#include <util/string/cast.h>
-
-#include <util/system/thread.h>
-
-namespace NYT {
-namespace NDetail {
-
-using namespace NRawClient;
-
-using ::ToString;
-
-////////////////////////////////////////////////////////////////////////////////
-
-static const ui64 DefaultExrtaTmpfsSize = 1024LL * 1024LL;
-
-////////////////////////////////////////////////////////////////////////////////
-
-namespace {
-
-////////////////////////////////////////////////////////////////////////////////
-
-struct TMapReduceOperationIo
-{
- TVector<TRichYPath> Inputs;
- TVector<TRichYPath> MapOutputs;
- TVector<TRichYPath> Outputs;
-
- TMaybe<TFormat> MapperInputFormat;
- TMaybe<TFormat> MapperOutputFormat;
-
- TMaybe<TFormat> ReduceCombinerInputFormat;
- TMaybe<TFormat> ReduceCombinerOutputFormat;
-
- TFormat ReducerInputFormat = TFormat::YsonBinary();
- TFormat ReducerOutputFormat = TFormat::YsonBinary();
-
- TVector<TSmallJobFile> MapperJobFiles;
- TVector<TSmallJobFile> ReduceCombinerJobFiles;
- TVector<TSmallJobFile> ReducerJobFiles;
-};
-
-template <typename T>
-void VerifyHasElements(const TVector<T>& paths, TStringBuf name)
-{
- if (paths.empty()) {
- ythrow TApiUsageError() << "no " << name << " table is specified";
- }
-}
-
-////////////////////////////////////////////////////////////////////////////////
-
-TVector<TSmallJobFile> CreateFormatConfig(
- TMaybe<TSmallJobFile> inputConfig,
- const TMaybe<TSmallJobFile>& outputConfig)
-{
- TVector<TSmallJobFile> result;
- if (inputConfig) {
- result.push_back(std::move(*inputConfig));
- }
- if (outputConfig) {
- result.push_back(std::move(*outputConfig));
- }
- return result;
-}
-
-template <typename T>
-ENodeReaderFormat NodeReaderFormatFromHintAndGlobalConfig(const TUserJobFormatHintsBase<T>& formatHints)
-{
- auto result = TConfig::Get()->NodeReaderFormat;
- if (formatHints.InputFormatHints_ && formatHints.InputFormatHints_->SkipNullValuesForTNode_) {
- Y_ENSURE_EX(
- result != ENodeReaderFormat::Skiff,
- TApiUsageError() << "skiff format doesn't support SkipNullValuesForTNode format hint");
- result = ENodeReaderFormat::Yson;
- }
- return result;
-}
-
-template <class TSpec>
-const TVector<TStructuredTablePath>& GetStructuredInputs(const TSpec& spec)
-{
- if constexpr (std::is_same_v<TSpec, TVanillaTask>) {
- static const TVector<TStructuredTablePath> empty;
- return empty;
- } else {
- return spec.GetStructuredInputs();
- }
-}
-
-template <class TSpec>
-const TVector<TStructuredTablePath>& GetStructuredOutputs(const TSpec& spec)
-{
- return spec.GetStructuredOutputs();
-}
-
-template <class TSpec>
-const TMaybe<TFormatHints>& GetInputFormatHints(const TSpec& spec)
-{
- if constexpr (std::is_same_v<TSpec, TVanillaTask>) {
- static const TMaybe<TFormatHints> empty = Nothing();
- return empty;
- } else {
- return spec.InputFormatHints_;
- }
-}
-
-template <class TSpec>
-const TMaybe<TFormatHints>& GetOutputFormatHints(const TSpec& spec)
-{
- return spec.OutputFormatHints_;
-}
-
-template <class TSpec>
-ENodeReaderFormat GetNodeReaderFormat(const TSpec& spec, bool allowSkiff)
-{
- if constexpr (std::is_same<TSpec, TVanillaTask>::value) {
- return ENodeReaderFormat::Yson;
- } else {
- return allowSkiff
- ? NodeReaderFormatFromHintAndGlobalConfig(spec)
- : ENodeReaderFormat::Yson;
- }
-}
-
-static void SortColumnsToNames(const TSortColumns& sortColumns, THashSet<TString>* result)
-{
- auto names = sortColumns.GetNames();
- result->insert(names.begin(), names.end());
-}
-
-static THashSet<TString> SortColumnsToNames(const TSortColumns& sortColumns)
-{
- THashSet<TString> columnNames;
- SortColumnsToNames(sortColumns, &columnNames);
- return columnNames;
-}
-
-THashSet<TString> GetColumnsUsedInOperation(const TJoinReduceOperationSpec& spec)
-{
- return SortColumnsToNames(spec.JoinBy_);
-}
-
-THashSet<TString> GetColumnsUsedInOperation(const TReduceOperationSpec& spec) {
- auto result = SortColumnsToNames(spec.SortBy_);
- SortColumnsToNames(spec.ReduceBy_, &result);
- if (spec.JoinBy_) {
- SortColumnsToNames(*spec.JoinBy_, &result);
- }
- return result;
-}
-
-THashSet<TString> GetColumnsUsedInOperation(const TMapReduceOperationSpec& spec)
-{
- auto result = SortColumnsToNames(spec.SortBy_);
- SortColumnsToNames(spec.ReduceBy_, &result);
- return result;
-}
-
-THashSet<TString> GetColumnsUsedInOperation(const TMapOperationSpec&)
-{
- return THashSet<TString>();
-}
-
-THashSet<TString> GetColumnsUsedInOperation(const TVanillaTask&)
-{
- return THashSet<TString>();
-}
-
-TStructuredJobTableList ApplyProtobufColumnFilters(
- const TStructuredJobTableList& tableList,
- const TOperationPreparer& preparer,
- const THashSet<TString>& columnsUsedInOperations,
- const TOperationOptions& options)
-{
- bool hasInputQuery = options.Spec_.Defined() && options.Spec_->IsMap() && options.Spec_->HasKey("input_query");
- if (hasInputQuery) {
- return tableList;
- }
-
- auto isDynamic = BatchTransform(
- CreateDefaultRequestRetryPolicy(preparer.GetContext().Config),
- preparer.GetContext(),
- tableList,
- [&] (TRawBatchRequest& batch, const auto& table) {
- return batch.Get(preparer.GetTransactionId(), table.RichYPath->Path_ + "/@dynamic", TGetOptions());
- });
-
- auto newTableList = tableList;
- for (size_t tableIndex = 0; tableIndex < tableList.size(); ++tableIndex) {
- if (isDynamic[tableIndex].AsBool()) {
- continue;
- }
- auto& table = newTableList[tableIndex];
- Y_VERIFY(table.RichYPath);
- if (table.RichYPath->Columns_) {
- continue;
- }
- if (!std::holds_alternative<TProtobufTableStructure>(table.Description)) {
- continue;
- }
- const auto& descriptor = std::get<TProtobufTableStructure>(table.Description).Descriptor;
- if (!descriptor) {
- continue;
- }
- auto fromDescriptor = NDetail::InferColumnFilter(*descriptor);
- if (!fromDescriptor) {
- continue;
- }
- THashSet<TString> columns(fromDescriptor->begin(), fromDescriptor->end());
- columns.insert(columnsUsedInOperations.begin(), columnsUsedInOperations.end());
- table.RichYPath->Columns(TVector<TString>(columns.begin(), columns.end()));
- }
- return newTableList;
-}
-
-template <class TSpec>
-TSimpleOperationIo CreateSimpleOperationIo(
- const IStructuredJob& structuredJob,
- const TOperationPreparer& preparer,
- const TSpec& spec,
- const TOperationOptions& options,
- bool allowSkiff)
-{
- if (!std::holds_alternative<TVoidStructuredRowStream>(structuredJob.GetInputRowStreamDescription())) {
- VerifyHasElements(GetStructuredInputs(spec), "input");
- }
-
- TUserJobFormatHints hints;
- hints.InputFormatHints_ = GetInputFormatHints(spec);
- hints.OutputFormatHints_ = GetOutputFormatHints(spec);
- ENodeReaderFormat nodeReaderFormat = GetNodeReaderFormat(spec, allowSkiff);
-
- return CreateSimpleOperationIoHelper(
- structuredJob,
- preparer,
- options,
- CanonizeStructuredTableList(preparer.GetContext(), GetStructuredInputs(spec)),
- CanonizeStructuredTableList(preparer.GetContext(), GetStructuredOutputs(spec)),
- hints,
- nodeReaderFormat,
- GetColumnsUsedInOperation(spec));
-}
-
-template <class T>
-TSimpleOperationIo CreateSimpleOperationIo(
- const IJob& job,
- const TOperationPreparer& preparer,
- const TSimpleRawOperationIoSpec<T>& spec)
-{
- auto getFormatOrDefault = [&] (const TMaybe<TFormat>& maybeFormat, const char* formatName) {
- if (maybeFormat) {
- return *maybeFormat;
- } else if (spec.Format_) {
- return *spec.Format_;
- } else {
- ythrow TApiUsageError() << "Neither " << formatName << "format nor default format is specified for raw operation";
- }
- };
-
- auto inputs = CanonizeYPaths(/* retryPolicy */ nullptr, preparer.GetContext(), spec.GetInputs());
- auto outputs = CanonizeYPaths(/* retryPolicy */ nullptr, preparer.GetContext(), spec.GetOutputs());
-
- VerifyHasElements(inputs, "input");
- VerifyHasElements(outputs, "output");
-
- TUserJobFormatHints hints;
-
- auto outputSchemas = PrepareOperation(
- job,
- TOperationPreparationContext(
- inputs,
- outputs,
- preparer.GetContext(),
- preparer.GetClientRetryPolicy(),
- preparer.GetTransactionId()),
- &inputs,
- &outputs,
- hints);
-
- Y_VERIFY(outputs.size() == outputSchemas.size());
- for (int i = 0; i < static_cast<int>(outputs.size()); ++i) {
- if (!outputs[i].Schema_ && !outputSchemas[i].Columns().empty()) {
- outputs[i].Schema_ = outputSchemas[i];
- }
- }
-
- return TSimpleOperationIo {
- inputs,
- outputs,
-
- getFormatOrDefault(spec.InputFormat_, "input"),
- getFormatOrDefault(spec.OutputFormat_, "output"),
-
- TVector<TSmallJobFile>{},
- };
-}
-
-////////////////////////////////////////////////////////////////////
-
-TString GetJobStderrWithRetriesAndIgnoreErrors(
- const IRequestRetryPolicyPtr& retryPolicy,
- const TClientContext& context,
- const TOperationId& operationId,
- const TJobId& jobId,
- const size_t stderrTailSize,
- const TGetJobStderrOptions& options = TGetJobStderrOptions())
-{
- TString jobStderr;
- try {
- jobStderr = GetJobStderrWithRetries(
- retryPolicy,
- context,
- operationId,
- jobId,
- options);
- } catch (const TErrorResponse& e) {
- YT_LOG_ERROR("Cannot get job stderr (OperationId: %v, JobId: %v, Error: %v)",
- operationId,
- jobId,
- e.what());
- }
- if (jobStderr.size() > stderrTailSize) {
- jobStderr = jobStderr.substr(jobStderr.size() - stderrTailSize, stderrTailSize);
- }
- return jobStderr;
-}
-
-TVector<TFailedJobInfo> GetFailedJobInfo(
- const IClientRetryPolicyPtr& clientRetryPolicy,
- const TClientContext& context,
- const TOperationId& operationId,
- const TGetFailedJobInfoOptions& options)
-{
- const auto listJobsResult = ListJobs(
- clientRetryPolicy->CreatePolicyForGenericRequest(),
- context,
- operationId,
- TListJobsOptions()
- .State(EJobState::Failed)
- .Limit(options.MaxJobCount_));
-
- const auto stderrTailSize = options.StderrTailSize_;
-
- TVector<TFailedJobInfo> result;
- for (const auto& job : listJobsResult.Jobs) {
- auto& info = result.emplace_back();
- Y_ENSURE(job.Id);
- info.JobId = *job.Id;
- info.Error = job.Error.GetOrElse(TYtError(TString("unknown error")));
- if (job.StderrSize.GetOrElse(0) != 0) {
- // There are cases when due to bad luck we cannot read stderr even if
- // list_jobs reports that stderr_size > 0.
- //
- // Such errors don't have special error code
- // so we ignore all errors and try our luck on other jobs.
- info.Stderr = GetJobStderrWithRetriesAndIgnoreErrors(
- clientRetryPolicy->CreatePolicyForGenericRequest(),
- context,
- operationId,
- *job.Id,
- stderrTailSize);
- }
- }
- return result;
-}
-
-struct TGetJobsStderrOptions
-{
- using TSelf = TGetJobsStderrOptions;
-
- // How many jobs to download. Which jobs will be chosen is undefined.
- FLUENT_FIELD_DEFAULT(ui64, MaxJobCount, 10);
-
- // How much of stderr should be downloaded.
- FLUENT_FIELD_DEFAULT(ui64, StderrTailSize, 64 * 1024);
-};
-
-static TVector<TString> GetJobsStderr(
- const IClientRetryPolicyPtr& clientRetryPolicy,
- const TClientContext& context,
- const TOperationId& operationId,
- const TGetJobsStderrOptions& options = TGetJobsStderrOptions())
-{
- const auto listJobsResult = ListJobs(
- clientRetryPolicy->CreatePolicyForGenericRequest(),
- context,
- operationId,
- TListJobsOptions().Limit(options.MaxJobCount_).WithStderr(true));
- const auto stderrTailSize = options.StderrTailSize_;
- TVector<TString> result;
- for (const auto& job : listJobsResult.Jobs) {
- result.push_back(
- // There are cases when due to bad luck we cannot read stderr even if
- // list_jobs reports that stderr_size > 0.
- //
- // Such errors don't have special error code
- // so we ignore all errors and try our luck on other jobs.
- GetJobStderrWithRetriesAndIgnoreErrors(
- clientRetryPolicy->CreatePolicyForGenericRequest(),
- context,
- operationId,
- *job.Id,
- stderrTailSize)
- );
- }
- return result;
-}
-
-int CountIntermediateTables(const TStructuredJobTableList& tables)
-{
- int result = 0;
- for (const auto& table : tables) {
- if (table.RichYPath) {
- break;
- }
- ++result;
- }
- return result;
-}
-
-////////////////////////////////////////////////////////////////////////////////
-
-} // namespace
-
-////////////////////////////////////////////////////////////////////////////////
-
-TSimpleOperationIo CreateSimpleOperationIoHelper(
- const IStructuredJob& structuredJob,
- const TOperationPreparer& preparer,
- const TOperationOptions& options,
- TStructuredJobTableList structuredInputs,
- TStructuredJobTableList structuredOutputs,
- TUserJobFormatHints hints,
- ENodeReaderFormat nodeReaderFormat,
- const THashSet<TString>& columnsUsedInOperations)
-{
- auto intermediateInputTableCount = CountIntermediateTables(structuredInputs);
- auto intermediateOutputTableCount = CountIntermediateTables(structuredOutputs);
-
- auto jobSchemaInferenceResult = PrepareOperation(
- structuredJob,
- TOperationPreparationContext(
- structuredInputs,
- structuredOutputs,
- preparer.GetContext(),
- preparer.GetClientRetryPolicy(),
- preparer.GetTransactionId()),
- &structuredInputs,
- &structuredOutputs,
- hints);
-
- TVector<TSmallJobFile> formatConfigList;
- TFormatBuilder formatBuilder(preparer.GetClientRetryPolicy(), preparer.GetContext(), preparer.GetTransactionId(), options);
-
- auto [inputFormat, inputFormatConfig] = formatBuilder.CreateFormat(
- structuredJob,
- EIODirection::Input,
- structuredInputs,
- hints.InputFormatHints_,
- nodeReaderFormat,
- /* allowFormatFromTableAttribute = */ true);
-
- auto [outputFormat, outputFormatConfig] = formatBuilder.CreateFormat(
- structuredJob,
- EIODirection::Output,
- structuredOutputs,
- hints.OutputFormatHints_,
- ENodeReaderFormat::Yson,
- /* allowFormatFromTableAttribute = */ false);
-
- const bool inferOutputSchema = options.InferOutputSchema_.GetOrElse(preparer.GetContext().Config->InferTableSchema);
-
- auto outputPaths = GetPathList(
- TStructuredJobTableList(structuredOutputs.begin() + intermediateOutputTableCount, structuredOutputs.end()),
- TVector<TTableSchema>(jobSchemaInferenceResult.begin() + intermediateOutputTableCount, jobSchemaInferenceResult.end()),
- inferOutputSchema);
-
- auto inputPaths = GetPathList(
- ApplyProtobufColumnFilters(
- TStructuredJobTableList(structuredInputs.begin() + intermediateInputTableCount, structuredInputs.end()),
- preparer,
- columnsUsedInOperations,
- options),
- /*schemaInferenceResult*/ Nothing(),
- /*inferSchema*/ false);
-
- return TSimpleOperationIo {
- inputPaths,
- outputPaths,
-
- inputFormat,
- outputFormat,
-
- CreateFormatConfig(inputFormatConfig, outputFormatConfig)
- };
-}
-
-EOperationBriefState CheckOperation(
- const IClientRetryPolicyPtr& clientRetryPolicy,
- const TClientContext& context,
- const TOperationId& operationId)
-{
- auto attributes = GetOperation(
- clientRetryPolicy->CreatePolicyForGenericRequest(),
- context,
- operationId,
- TGetOperationOptions().AttributeFilter(TOperationAttributeFilter()
- .Add(EOperationAttribute::State)
- .Add(EOperationAttribute::Result)));
- Y_VERIFY(attributes.BriefState,
- "get_operation for operation %s has not returned \"state\" field",
- GetGuidAsString(operationId).Data());
- if (*attributes.BriefState == EOperationBriefState::Completed) {
- return EOperationBriefState::Completed;
- } else if (*attributes.BriefState == EOperationBriefState::Aborted || *attributes.BriefState == EOperationBriefState::Failed) {
- YT_LOG_ERROR("Operation %v %v (%v)",
- operationId,
- ToString(*attributes.BriefState),
- ToString(TOperationExecutionTimeTracker::Get()->Finish(operationId)));
-
- auto failedJobInfoList = GetFailedJobInfo(
- clientRetryPolicy,
- context,
- operationId,
- TGetFailedJobInfoOptions());
-
- Y_VERIFY(attributes.Result && attributes.Result->Error);
- ythrow TOperationFailedError(
- *attributes.BriefState == EOperationBriefState::Aborted
- ? TOperationFailedError::Aborted
- : TOperationFailedError::Failed,
- operationId,
- *attributes.Result->Error,
- failedJobInfoList);
- }
- return EOperationBriefState::InProgress;
-}
-
-void WaitForOperation(
- const IClientRetryPolicyPtr& clientRetryPolicy,
- const TClientContext& context,
- const TOperationId& operationId)
-{
- const TDuration checkOperationStateInterval =
- UseLocalModeOptimization(context, clientRetryPolicy)
- ? Min(TDuration::MilliSeconds(100), context.Config->OperationTrackerPollPeriod)
- : context.Config->OperationTrackerPollPeriod;
-
- while (true) {
- auto status = CheckOperation(clientRetryPolicy, context, operationId);
- if (status == EOperationBriefState::Completed) {
- YT_LOG_INFO("Operation %v completed (%v)",
- operationId,
- TOperationExecutionTimeTracker::Get()->Finish(operationId));
- break;
- }
- TWaitProxy::Get()->Sleep(checkOperationStateInterval);
- }
-}
-
-////////////////////////////////////////////////////////////////////////////////
-
-namespace {
-
-TNode BuildAutoMergeSpec(const TAutoMergeSpec& options)
-{
- TNode result;
- if (options.Mode_) {
- result["mode"] = ToString(*options.Mode_);
- }
- if (options.MaxIntermediateChunkCount_) {
- result["max_intermediate_chunk_count"] = *options.MaxIntermediateChunkCount_;
- }
- if (options.ChunkCountPerMergeJob_) {
- result["chunk_count_per_merge_job"] = *options.ChunkCountPerMergeJob_;
- }
- if (options.ChunkSizeThreshold_) {
- result["chunk_size_threshold"] = *options.ChunkSizeThreshold_;
- }
- return result;
-}
-
-TNode BuildJobProfilerSpec(const TJobProfilerSpec& profilerSpec)
-{
- TNode result;
- if (profilerSpec.ProfilingBinary_) {
- result["binary"] = ToString(*profilerSpec.ProfilingBinary_);
- }
- if (profilerSpec.ProfilerType_) {
- result["type"] = ToString(*profilerSpec.ProfilerType_);
- }
- if (profilerSpec.ProfilingProbability_) {
- result["profiling_probability"] = *profilerSpec.ProfilingProbability_;
- }
- if (profilerSpec.SamplingFrequency_) {
- result["sampling_frequency"] = *profilerSpec.SamplingFrequency_;
- }
-
- return result;
-}
-
-// Returns undefined node if resources doesn't contain any meaningful field
-TNode BuildSchedulerResourcesSpec(const TSchedulerResources& resources)
-{
- TNode result;
- if (resources.UserSlots().Defined()) {
- result["user_slots"] = *resources.UserSlots();
- }
- if (resources.Cpu().Defined()) {
- result["cpu"] = *resources.Cpu();
- }
- if (resources.Memory().Defined()) {
- result["memory"] = *resources.Memory();
- }
- return result;
-}
-
-void BuildUserJobFluently(
- const TJobPreparer& preparer,
- const TMaybe<TFormat>& inputFormat,
- const TMaybe<TFormat>& outputFormat,
- TFluentMap fluent)
-{
- const auto& userJobSpec = preparer.GetSpec();
- TMaybe<i64> memoryLimit = userJobSpec.MemoryLimit_;
- TMaybe<double> cpuLimit = userJobSpec.CpuLimit_;
- TMaybe<ui16> portCount = userJobSpec.PortCount_;
-
- // Use 1MB extra tmpfs size by default, it helps to detect job sandbox as tmp directory
- // for standard python libraries. See YTADMINREQ-14505 for more details.
- auto tmpfsSize = preparer.GetSpec().ExtraTmpfsSize_.GetOrElse(DefaultExrtaTmpfsSize);
- if (preparer.ShouldMountSandbox()) {
- tmpfsSize += preparer.GetTotalFileSize();
- if (tmpfsSize == 0) {
- // This can be a case for example when it is local mode and we don't upload binary.
- // NOTE: YT doesn't like zero tmpfs size.
- tmpfsSize = RoundUpFileSize(1);
- }
- memoryLimit = memoryLimit.GetOrElse(512ll << 20) + tmpfsSize;
- }
-
- fluent
- .Item("file_paths").List(preparer.GetFiles())
- .Item("command").Value(preparer.GetCommand())
- .Item("class_name").Value(preparer.GetClassName())
- .DoIf(!userJobSpec.Environment_.empty(), [&] (TFluentMap fluentMap) {
- TNode environment;
- for (const auto& item : userJobSpec.Environment_) {
- environment[item.first] = item.second;
- }
- fluentMap.Item("environment").Value(environment);
- })
- .DoIf(userJobSpec.DiskSpaceLimit_.Defined(), [&] (TFluentMap fluentMap) {
- fluentMap.Item("disk_space_limit").Value(*userJobSpec.DiskSpaceLimit_);
- })
- .DoIf(inputFormat.Defined(), [&] (TFluentMap fluentMap) {
- fluentMap.Item("input_format").Value(inputFormat->Config);
- })
- .DoIf(outputFormat.Defined(), [&] (TFluentMap fluentMap) {
- fluentMap.Item("output_format").Value(outputFormat->Config);
- })
- .DoIf(memoryLimit.Defined(), [&] (TFluentMap fluentMap) {
- fluentMap.Item("memory_limit").Value(*memoryLimit);
- })
- .DoIf(userJobSpec.MemoryReserveFactor_.Defined(), [&] (TFluentMap fluentMap) {
- fluentMap.Item("memory_reserve_factor").Value(*userJobSpec.MemoryReserveFactor_);
- })
- .DoIf(cpuLimit.Defined(), [&] (TFluentMap fluentMap) {
- fluentMap.Item("cpu_limit").Value(*cpuLimit);
- })
- .DoIf(portCount.Defined(), [&] (TFluentMap fluentMap) {
- fluentMap.Item("port_count").Value(*portCount);
- })
- .DoIf(userJobSpec.JobTimeLimit_.Defined(), [&] (TFluentMap fluentMap) {
- fluentMap.Item("job_time_limit").Value(userJobSpec.JobTimeLimit_->MilliSeconds());
- })
- .DoIf(userJobSpec.NetworkProject_.Defined(), [&] (TFluentMap fluentMap) {
- fluentMap.Item("network_project").Value(*userJobSpec.NetworkProject_);
- })
- .DoIf(preparer.ShouldMountSandbox(), [&] (TFluentMap fluentMap) {
- fluentMap.Item("tmpfs_path").Value(".");
- fluentMap.Item("tmpfs_size").Value(tmpfsSize);
- fluentMap.Item("copy_files").Value(true);
- })
- .Item("profilers")
- .BeginList()
- .DoFor(userJobSpec.JobProfilers_, [&] (TFluentList list, const auto& jobProfiler) {
- list.Item().Value(BuildJobProfilerSpec(jobProfiler));
- })
- .EndList();
-}
-
-template <typename T>
-void BuildCommonOperationPart(const TConfigPtr& config, const TOperationSpecBase<T>& baseSpec, const TOperationOptions& options, TFluentMap fluent)
-{
- const TProcessState* properties = TProcessState::Get();
- TString pool = config->Pool;
-
- if (baseSpec.Pool_) {
- pool = *baseSpec.Pool_;
- }
-
- fluent
- .Item("started_by")
- .BeginMap()
- .Item("hostname").Value(properties->FqdnHostName)
- .Item("pid").Value(properties->Pid)
- .Item("user").Value(properties->UserName)
- .Item("command").List(properties->CensoredCommandLine)
- .Item("wrapper_version").Value(properties->ClientVersion)
- .EndMap()
- .DoIf(!pool.empty(), [&] (TFluentMap fluentMap) {
- fluentMap.Item("pool").Value(pool);
- })
- .DoIf(baseSpec.Weight_.Defined(), [&] (TFluentMap fluentMap) {
- fluentMap.Item("weight").Value(*baseSpec.Weight_);
- })
- .DoIf(baseSpec.TimeLimit_.Defined(), [&] (TFluentMap fluentMap) {
- fluentMap.Item("time_limit").Value(baseSpec.TimeLimit_->MilliSeconds());
- })
- .DoIf(baseSpec.PoolTrees().Defined(), [&] (TFluentMap fluentMap) {
- TNode poolTreesSpec = TNode::CreateList();
- for (const auto& tree : *baseSpec.PoolTrees()) {
- poolTreesSpec.Add(tree);
- }
- fluentMap.Item("pool_trees").Value(poolTreesSpec);
- })
- .DoIf(baseSpec.ResourceLimits().Defined(), [&] (TFluentMap fluentMap) {
- auto resourceLimitsSpec = BuildSchedulerResourcesSpec(*baseSpec.ResourceLimits());
- if (!resourceLimitsSpec.IsUndefined()) {
- fluentMap.Item("resource_limits").Value(std::move(resourceLimitsSpec));
- }
- })
- .DoIf(options.SecureVault_.Defined(), [&] (TFluentMap fluentMap) {
- Y_ENSURE(options.SecureVault_->IsMap(),
- "SecureVault must be a map node, got " << options.SecureVault_->GetType());
- fluentMap.Item("secure_vault").Value(*options.SecureVault_);
- })
- .DoIf(baseSpec.Title_.Defined(), [&] (TFluentMap fluentMap) {
- fluentMap.Item("title").Value(*baseSpec.Title_);
- });
-}
-
-template <typename TSpec>
-void BuildCommonUserOperationPart(const TSpec& baseSpec, TNode* spec)
-{
- if (baseSpec.MaxFailedJobCount_.Defined()) {
- (*spec)["max_failed_job_count"] = *baseSpec.MaxFailedJobCount_;
- }
- if (baseSpec.FailOnJobRestart_.Defined()) {
- (*spec)["fail_on_job_restart"] = *baseSpec.FailOnJobRestart_;
- }
- if (baseSpec.StderrTablePath_.Defined()) {
- (*spec)["stderr_table_path"] = *baseSpec.StderrTablePath_;
- }
- if (baseSpec.CoreTablePath_.Defined()) {
- (*spec)["core_table_path"] = *baseSpec.CoreTablePath_;
- }
- if (baseSpec.WaitingJobTimeout_.Defined()) {
- (*spec)["waiting_job_timeout"] = baseSpec.WaitingJobTimeout_->MilliSeconds();
- }
-}
-
-template <typename TSpec>
-void BuildJobCountOperationPart(const TSpec& spec, TNode* nodeSpec)
-{
- if (spec.JobCount_.Defined()) {
- (*nodeSpec)["job_count"] = *spec.JobCount_;
- }
- if (spec.DataSizePerJob_.Defined()) {
- (*nodeSpec)["data_size_per_job"] = *spec.DataSizePerJob_;
- }
-}
-
-template <typename TSpec>
-void BuildPartitionCountOperationPart(const TSpec& spec, TNode* nodeSpec)
-{
- if (spec.PartitionCount_.Defined()) {
- (*nodeSpec)["partition_count"] = *spec.PartitionCount_;
- }
- if (spec.PartitionDataSize_.Defined()) {
- (*nodeSpec)["partition_data_size"] = *spec.PartitionDataSize_;
- }
-}
-
-template <typename TSpec>
-void BuildDataSizePerSortJobPart(const TSpec& spec, TNode* nodeSpec)
-{
- if (spec.DataSizePerSortJob_.Defined()) {
- (*nodeSpec)["data_size_per_sort_job"] = *spec.DataSizePerSortJob_;
- }
-}
-
-template <typename TSpec>
-void BuildPartitionJobCountOperationPart(const TSpec& spec, TNode* nodeSpec)
-{
- if (spec.PartitionJobCount_.Defined()) {
- (*nodeSpec)["partition_job_count"] = *spec.PartitionJobCount_;
- }
- if (spec.DataSizePerPartitionJob_.Defined()) {
- (*nodeSpec)["data_size_per_partition_job"] = *spec.DataSizePerPartitionJob_;
- }
-}
-
-template <typename TSpec>
-void BuildMapJobCountOperationPart(const TSpec& spec, TNode* nodeSpec)
-{
- if (spec.MapJobCount_.Defined()) {
- (*nodeSpec)["map_job_count"] = *spec.MapJobCount_;
- }
- if (spec.DataSizePerMapJob_.Defined()) {
- (*nodeSpec)["data_size_per_map_job"] = *spec.DataSizePerMapJob_;
- }
-}
-
-template <typename TSpec>
-void BuildIntermediateDataPart(const TSpec& spec, TNode* nodeSpec)
-{
- if (spec.IntermediateDataAccount_.Defined()) {
- (*nodeSpec)["intermediate_data_account"] = *spec.IntermediateDataAccount_;
- }
- if (spec.IntermediateDataReplicationFactor_.Defined()) {
- (*nodeSpec)["intermediate_data_replication_factor"] = *spec.IntermediateDataReplicationFactor_;
- }
-}
-
-////////////////////////////////////////////////////////////////////////////////
-
-TNode MergeSpec(TNode dst, TNode spec, const TOperationOptions& options)
-{
- MergeNodes(dst["spec"], spec);
- if (options.Spec_) {
- MergeNodes(dst["spec"], *options.Spec_);
- }
- return dst;
-}
-
-template <typename TSpec>
-void CreateDebugOutputTables(const TSpec& spec, const TOperationPreparer& preparer)
-{
- if (spec.StderrTablePath_.Defined()) {
- NYT::NDetail::Create(
- preparer.GetClientRetryPolicy()->CreatePolicyForGenericRequest(),
- preparer.GetContext(),
- TTransactionId(),
- *spec.StderrTablePath_,
- NT_TABLE,
- TCreateOptions()
- .IgnoreExisting(true)
- .Recursive(true));
- }
- if (spec.CoreTablePath_.Defined()) {
- NYT::NDetail::Create(
- preparer.GetClientRetryPolicy()->CreatePolicyForGenericRequest(),
- preparer.GetContext(),
- TTransactionId(),
- *spec.CoreTablePath_,
- NT_TABLE,
- TCreateOptions()
- .IgnoreExisting(true)
- .Recursive(true));
- }
-}
-
-void CreateOutputTable(
- const TOperationPreparer& preparer,
- const TRichYPath& path)
-{
- Y_ENSURE(path.Path_, "Output table is not set");
- Create(
- preparer.GetClientRetryPolicy()->CreatePolicyForGenericRequest(),
- preparer.GetContext(), preparer.GetTransactionId(), path.Path_, NT_TABLE,
- TCreateOptions()
- .IgnoreExisting(true)
- .Recursive(true));
-}
-
-void CreateOutputTables(
- const TOperationPreparer& preparer,
- const TVector<TRichYPath>& paths)
-{
- for (auto& path : paths) {
- CreateOutputTable(preparer, path);
- }
-}
-
-void CheckInputTablesExist(
- const TOperationPreparer& preparer,
- const TVector<TRichYPath>& paths)
-{
- Y_ENSURE(!paths.empty(), "Input tables are not set");
- for (auto& path : paths) {
- auto curTransactionId = path.TransactionId_.GetOrElse(preparer.GetTransactionId());
- Y_ENSURE_EX(
- Exists(
- preparer.GetClientRetryPolicy()->CreatePolicyForGenericRequest(),
- preparer.GetContext(),
- curTransactionId,
- path.Path_),
- TApiUsageError() << "Input table '" << path.Path_ << "' doesn't exist");
- }
-}
-
-void LogJob(const TOperationId& opId, const IJob* job, const char* type)
-{
- if (job) {
- YT_LOG_INFO("Operation %v; %v = %v",
- opId,
- type,
- TJobFactory::Get()->GetJobName(job));
- }
-}
-
-void LogYPaths(const TOperationId& opId, const TVector<TRichYPath>& paths, const char* type)
-{
- for (size_t i = 0; i < paths.size(); ++i) {
- YT_LOG_INFO("Operation %v; %v[%v] = %v",
- opId,
- type,
- i,
- paths[i].Path_);
- }
-}
-
-void LogYPath(const TOperationId& opId, const TRichYPath& path, const char* type)
-{
- YT_LOG_INFO("Operation %v; %v = %v",
- opId,
- type,
- path.Path_);
-}
-
-TString AddModeToTitleIfDebug(const TString& title) {
-#ifndef NDEBUG
- return title + " (debug build)";
-#else
- return title;
-#endif
-}
-
-} // namespace
-
-////////////////////////////////////////////////////////////////////////////////
-
-template <typename T>
-void DoExecuteMap(
- const TOperationPtr& operation,
- const TOperationPreparerPtr& preparer,
- const TSimpleOperationIo& operationIo,
- TMapOperationSpecBase<T> spec,
- const IJobPtr& mapper,
- const TOperationOptions& options)
-{
- if (options.CreateDebugOutputTables_) {
- CreateDebugOutputTables(spec, *preparer);
- }
- if (options.CreateOutputTables_) {
- CheckInputTablesExist(*preparer, operationIo.Inputs);
- CreateOutputTables(*preparer, operationIo.Outputs);
- }
-
- TJobPreparer map(
- *preparer,
- spec.MapperSpec_,
- *mapper,
- operationIo.Outputs.size(),
- operationIo.JobFiles,
- options);
-
- spec.Title_ = spec.Title_.GetOrElse(AddModeToTitleIfDebug(map.GetClassName()));
-
- TNode specNode = BuildYsonNodeFluently()
- .BeginMap().Item("spec").BeginMap()
- .Item("mapper").DoMap([&] (TFluentMap fluent) {
- BuildUserJobFluently(
- map,
- operationIo.InputFormat,
- operationIo.OutputFormat,
- fluent);
- })
- .DoIf(spec.AutoMerge_.Defined(), [&] (TFluentMap fluent) {
- auto autoMergeSpec = BuildAutoMergeSpec(*spec.AutoMerge_);
- if (!autoMergeSpec.IsUndefined()) {
- fluent.Item("auto_merge").Value(std::move(autoMergeSpec));
- }
- })
- .Item("input_table_paths").List(operationIo.Inputs)
- .Item("output_table_paths").List(operationIo.Outputs)
- .DoIf(spec.Ordered_.Defined(), [&] (TFluentMap fluent) {
- fluent.Item("ordered").Value(spec.Ordered_.GetRef());
- })
- .Do(std::bind(BuildCommonOperationPart<T>, preparer->GetContext().Config, spec, options, std::placeholders::_1))
- .EndMap().EndMap();
-
- specNode["spec"]["job_io"]["control_attributes"]["enable_row_index"] = TNode(true);
- specNode["spec"]["job_io"]["control_attributes"]["enable_range_index"] = TNode(true);
- if (!preparer->GetContext().Config->TableWriter.Empty()) {
- specNode["spec"]["job_io"]["table_writer"] = preparer->GetContext().Config->TableWriter;
- }
-
- BuildCommonUserOperationPart(spec, &specNode["spec"]);
- BuildJobCountOperationPart(spec, &specNode["spec"]);
-
- auto startOperation = [
- operation=operation.Get(),
- spec=MergeSpec(std::move(specNode), preparer->GetContext().Config->Spec, options),
- preparer,
- operationIo,
- mapper
- ] () {
- auto operationId = preparer->StartOperation(operation, "map", spec);
-
- LogJob(operationId, mapper.Get(), "mapper");
- LogYPaths(operationId, operationIo.Inputs, "input");
- LogYPaths(operationId, operationIo.Outputs, "output");
-
- return operationId;
- };
- operation->SetDelayedStartFunction(std::move(startOperation));
-}
-
-void ExecuteMap(
- const TOperationPtr& operation,
- const TOperationPreparerPtr& preparer,
- const TMapOperationSpec& spec,
- const ::TIntrusivePtr<IStructuredJob>& mapper,
- const TOperationOptions& options)
-{
- YT_LOG_DEBUG("Starting map operation (PreparationId: %v)",
- preparer->GetPreparationId());
- auto operationIo = CreateSimpleOperationIo(*mapper, *preparer, spec, options, /* allowSkiff = */ true);
- DoExecuteMap(
- operation,
- preparer,
- operationIo,
- spec,
- mapper,
- options);
-}
-
-void ExecuteRawMap(
- const TOperationPtr& operation,
- const TOperationPreparerPtr& preparer,
- const TRawMapOperationSpec& spec,
- const ::TIntrusivePtr<IRawJob>& mapper,
- const TOperationOptions& options)
-{
- YT_LOG_DEBUG("Starting raw map operation (PreparationId: %v)",
- preparer->GetPreparationId());
- auto operationIo = CreateSimpleOperationIo(*mapper, *preparer, spec);
- DoExecuteMap(
- operation,
- preparer,
- operationIo,
- spec,
- mapper,
- options);
-}
-
-////////////////////////////////////////////////////////////////////////////////
-
-template <typename T>
-void DoExecuteReduce(
- const TOperationPtr& operation,
- const TOperationPreparerPtr& preparer,
- const TSimpleOperationIo& operationIo,
- TReduceOperationSpecBase<T> spec,
- const IJobPtr& reducer,
- const TOperationOptions& options)
-{
- if (options.CreateDebugOutputTables_) {
- CreateDebugOutputTables(spec, *preparer);
- }
- if (options.CreateOutputTables_) {
- CheckInputTablesExist(*preparer, operationIo.Inputs);
- CreateOutputTables(*preparer, operationIo.Outputs);
- }
-
- TJobPreparer reduce(
- *preparer,
- spec.ReducerSpec_,
- *reducer,
- operationIo.Outputs.size(),
- operationIo.JobFiles,
- options);
-
- spec.Title_ = spec.Title_.GetOrElse(AddModeToTitleIfDebug(reduce.GetClassName()));
-
- TNode specNode = BuildYsonNodeFluently()
- .BeginMap().Item("spec").BeginMap()
- .Item("reducer").DoMap([&] (TFluentMap fluent) {
- BuildUserJobFluently(
- reduce,
- operationIo.InputFormat,
- operationIo.OutputFormat,
- fluent);
- })
- .Item("sort_by").Value(spec.SortBy_)
- .Item("reduce_by").Value(spec.ReduceBy_)
- .DoIf(spec.JoinBy_.Defined(), [&] (TFluentMap fluent) {
- fluent.Item("join_by").Value(spec.JoinBy_.GetRef());
- })
- .DoIf(spec.EnableKeyGuarantee_.Defined(), [&] (TFluentMap fluent) {
- fluent.Item("enable_key_guarantee").Value(spec.EnableKeyGuarantee_.GetRef());
- })
- .Item("input_table_paths").List(operationIo.Inputs)
- .Item("output_table_paths").List(operationIo.Outputs)
- .Item("job_io").BeginMap()
- .Item("control_attributes").BeginMap()
- .Item("enable_key_switch").Value(true)
- .Item("enable_row_index").Value(true)
- .Item("enable_range_index").Value(true)
- .EndMap()
- .DoIf(!preparer->GetContext().Config->TableWriter.Empty(), [&] (TFluentMap fluent) {
- fluent.Item("table_writer").Value(preparer->GetContext().Config->TableWriter);
- })
- .EndMap()
- .DoIf(spec.AutoMerge_.Defined(), [&] (TFluentMap fluent) {
- fluent.Item("auto_merge").Value(BuildAutoMergeSpec(*spec.AutoMerge_));
- })
- .Do(std::bind(BuildCommonOperationPart<T>, preparer->GetContext().Config, spec, options, std::placeholders::_1))
- .EndMap().EndMap();
-
- BuildCommonUserOperationPart(spec, &specNode["spec"]);
- BuildJobCountOperationPart(spec, &specNode["spec"]);
-
- auto startOperation = [
- operation=operation.Get(),
- spec=MergeSpec(std::move(specNode), preparer->GetContext().Config->Spec, options),
- preparer,
- operationIo,
- reducer
- ] () {
- auto operationId = preparer->StartOperation(operation, "reduce", spec);
-
- LogJob(operationId, reducer.Get(), "reducer");
- LogYPaths(operationId, operationIo.Inputs, "input");
- LogYPaths(operationId, operationIo.Outputs, "output");
-
- return operationId;
- };
-
- operation->SetDelayedStartFunction(std::move(startOperation));
-}
-
-void ExecuteReduce(
- const TOperationPtr& operation,
- const TOperationPreparerPtr& preparer,
- const TReduceOperationSpec& spec,
- const ::TIntrusivePtr<IStructuredJob>& reducer,
- const TOperationOptions& options)
-{
- YT_LOG_DEBUG("Starting reduce operation (PreparationId: %v)",
- preparer->GetPreparationId());
- auto operationIo = CreateSimpleOperationIo(*reducer, *preparer, spec, options, /* allowSkiff = */ false);
- DoExecuteReduce(
- operation,
- preparer,
- operationIo,
- spec,
- reducer,
- options);
-}
-
-void ExecuteRawReduce(
- const TOperationPtr& operation,
- const TOperationPreparerPtr& preparer,
- const TRawReduceOperationSpec& spec,
- const ::TIntrusivePtr<IRawJob>& reducer,
- const TOperationOptions& options)
-{
- YT_LOG_DEBUG("Starting raw reduce operation (PreparationId: %v)",
- preparer->GetPreparationId());
- auto operationIo = CreateSimpleOperationIo(*reducer, *preparer, spec);
- DoExecuteReduce(
- operation,
- preparer,
- operationIo,
- spec,
- reducer,
- options);
-}
-
-////////////////////////////////////////////////////////////////////////////////
-
-template <typename T>
-void DoExecuteJoinReduce(
- const TOperationPtr& operation,
- const TOperationPreparerPtr& preparer,
- const TSimpleOperationIo& operationIo,
- TJoinReduceOperationSpecBase<T> spec,
- const IJobPtr& reducer,
- const TOperationOptions& options)
-{
- if (options.CreateDebugOutputTables_) {
- CreateDebugOutputTables(spec, *preparer);
- }
- if (options.CreateOutputTables_) {
- CheckInputTablesExist(*preparer, operationIo.Inputs);
- CreateOutputTables(*preparer, operationIo.Outputs);
- }
-
- TJobPreparer reduce(
- *preparer,
- spec.ReducerSpec_,
- *reducer,
- operationIo.Outputs.size(),
- operationIo.JobFiles,
- options);
-
- spec.Title_ = spec.Title_.GetOrElse(AddModeToTitleIfDebug(reduce.GetClassName()));
-
- TNode specNode = BuildYsonNodeFluently()
- .BeginMap().Item("spec").BeginMap()
- .Item("reducer").DoMap([&] (TFluentMap fluent) {
- BuildUserJobFluently(
- reduce,
- operationIo.InputFormat,
- operationIo.OutputFormat,
- fluent);
- })
- .Item("join_by").Value(spec.JoinBy_)
- .Item("input_table_paths").List(operationIo.Inputs)
- .Item("output_table_paths").List(operationIo.Outputs)
- .Item("job_io").BeginMap()
- .Item("control_attributes").BeginMap()
- .Item("enable_key_switch").Value(true)
- .Item("enable_row_index").Value(true)
- .Item("enable_range_index").Value(true)
- .EndMap()
- .DoIf(!preparer->GetContext().Config->TableWriter.Empty(), [&] (TFluentMap fluent) {
- fluent.Item("table_writer").Value(preparer->GetContext().Config->TableWriter);
- })
- .EndMap()
- .Do(std::bind(BuildCommonOperationPart<T>, preparer->GetContext().Config, spec, options, std::placeholders::_1))
- .EndMap().EndMap();
-
- BuildCommonUserOperationPart(spec, &specNode["spec"]);
- BuildJobCountOperationPart(spec, &specNode["spec"]);
-
- auto startOperation = [
- operation=operation.Get(),
- spec=MergeSpec(std::move(specNode), preparer->GetContext().Config->Spec, options),
- preparer,
- reducer,
- operationIo
- ] () {
- auto operationId = preparer->StartOperation(operation, "join_reduce", spec);
-
- LogJob(operationId, reducer.Get(), "reducer");
- LogYPaths(operationId, operationIo.Inputs, "input");
- LogYPaths(operationId, operationIo.Outputs, "output");
-
- return operationId;
- };
-
- operation->SetDelayedStartFunction(std::move(startOperation));
-}
-
-void ExecuteJoinReduce(
- const TOperationPtr& operation,
- const TOperationPreparerPtr& preparer,
- const TJoinReduceOperationSpec& spec,
- const ::TIntrusivePtr<IStructuredJob>& reducer,
- const TOperationOptions& options)
-{
- YT_LOG_DEBUG("Starting join reduce operation (PreparationId: %v)",
- preparer->GetPreparationId());
- auto operationIo = CreateSimpleOperationIo(*reducer, *preparer, spec, options, /* allowSkiff = */ false);
- return DoExecuteJoinReduce(
- operation,
- preparer,
- operationIo,
- spec,
- reducer,
- options);
-}
-
-void ExecuteRawJoinReduce(
- const TOperationPtr& operation,
- const TOperationPreparerPtr& preparer,
- const TRawJoinReduceOperationSpec& spec,
- const ::TIntrusivePtr<IRawJob>& reducer,
- const TOperationOptions& options)
-{
- YT_LOG_DEBUG("Starting raw join reduce operation (PreparationId: %v)",
- preparer->GetPreparationId());
- auto operationIo = CreateSimpleOperationIo(*reducer, *preparer, spec);
- return DoExecuteJoinReduce(
- operation,
- preparer,
- operationIo,
- spec,
- reducer,
- options);
-}
-
-////////////////////////////////////////////////////////////////////////////////
-
-template <typename T>
-void DoExecuteMapReduce(
- const TOperationPtr& operation,
- const TOperationPreparerPtr& preparer,
- const TMapReduceOperationIo& operationIo,
- TMapReduceOperationSpecBase<T> spec,
- const IJobPtr& mapper,
- const IJobPtr& reduceCombiner,
- const IJobPtr& reducer,
- const TOperationOptions& options)
-{
- TVector<TRichYPath> allOutputs;
- allOutputs.insert(allOutputs.end(), operationIo.MapOutputs.begin(), operationIo.MapOutputs.end());
- allOutputs.insert(allOutputs.end(), operationIo.Outputs.begin(), operationIo.Outputs.end());
-
- if (options.CreateDebugOutputTables_) {
- CreateDebugOutputTables(spec, *preparer);
- }
- if (options.CreateOutputTables_) {
- CheckInputTablesExist(*preparer, operationIo.Inputs);
- CreateOutputTables(*preparer, allOutputs);
- }
-
- TSortColumns sortBy = spec.SortBy_;
- TSortColumns reduceBy = spec.ReduceBy_;
-
- if (sortBy.Parts_.empty()) {
- sortBy = reduceBy;
- }
-
- const bool hasMapper = mapper != nullptr;
- const bool hasCombiner = reduceCombiner != nullptr;
-
- TVector<TRichYPath> files;
-
- TJobPreparer reduce(
- *preparer,
- spec.ReducerSpec_,
- *reducer,
- operationIo.Outputs.size(),
- operationIo.ReducerJobFiles,
- options);
-
- TString title;
-
- TNode specNode = BuildYsonNodeFluently()
- .BeginMap().Item("spec").BeginMap()
- .DoIf(hasMapper, [&] (TFluentMap fluent) {
- TJobPreparer map(
- *preparer,
- spec.MapperSpec_,
- *mapper,
- 1 + operationIo.MapOutputs.size(),
- operationIo.MapperJobFiles,
- options);
- fluent.Item("mapper").DoMap([&] (TFluentMap fluent) {
- BuildUserJobFluently(
- std::cref(map),
- *operationIo.MapperInputFormat,
- *operationIo.MapperOutputFormat,
- fluent);
- });
-
- title = "mapper:" + map.GetClassName() + " ";
- })
- .DoIf(hasCombiner, [&] (TFluentMap fluent) {
- TJobPreparer combine(
- *preparer,
- spec.ReduceCombinerSpec_,
- *reduceCombiner,
- size_t(1),
- operationIo.ReduceCombinerJobFiles,
- options);
- fluent.Item("reduce_combiner").DoMap([&] (TFluentMap fluent) {
- BuildUserJobFluently(
- combine,
- *operationIo.ReduceCombinerInputFormat,
- *operationIo.ReduceCombinerOutputFormat,
- fluent);
- });
- title += "combiner:" + combine.GetClassName() + " ";
- })
- .Item("reducer").DoMap([&] (TFluentMap fluent) {
- BuildUserJobFluently(
- reduce,
- operationIo.ReducerInputFormat,
- operationIo.ReducerOutputFormat,
- fluent);
- })
- .Item("sort_by").Value(sortBy)
- .Item("reduce_by").Value(reduceBy)
- .Item("input_table_paths").List(operationIo.Inputs)
- .Item("output_table_paths").List(allOutputs)
- .Item("mapper_output_table_count").Value(operationIo.MapOutputs.size())
- .DoIf(spec.ForceReduceCombiners_.Defined(), [&] (TFluentMap fluent) {
- fluent.Item("force_reduce_combiners").Value(*spec.ForceReduceCombiners_);
- })
- .Item("map_job_io").BeginMap()
- .Item("control_attributes").BeginMap()
- .Item("enable_row_index").Value(true)
- .Item("enable_range_index").Value(true)
- .EndMap()
- .DoIf(!preparer->GetContext().Config->TableWriter.Empty(), [&] (TFluentMap fluent) {
- fluent.Item("table_writer").Value(preparer->GetContext().Config->TableWriter);
- })
- .EndMap()
- .Item("sort_job_io").BeginMap()
- .Item("control_attributes").BeginMap()
- .Item("enable_key_switch").Value(true)
- .EndMap()
- .DoIf(!preparer->GetContext().Config->TableWriter.Empty(), [&] (TFluentMap fluent) {
- fluent.Item("table_writer").Value(preparer->GetContext().Config->TableWriter);
- })
- .EndMap()
- .Item("reduce_job_io").BeginMap()
- .Item("control_attributes").BeginMap()
- .Item("enable_key_switch").Value(true)
- .EndMap()
- .DoIf(!preparer->GetContext().Config->TableWriter.Empty(), [&] (TFluentMap fluent) {
- fluent.Item("table_writer").Value(preparer->GetContext().Config->TableWriter);
- })
- .EndMap()
- .Do([&] (TFluentMap) {
- spec.Title_ = spec.Title_.GetOrElse(AddModeToTitleIfDebug(title + "reducer:" + reduce.GetClassName()));
- })
- .Do(std::bind(BuildCommonOperationPart<T>, preparer->GetContext().Config, spec, options, std::placeholders::_1))
- .EndMap().EndMap();
-
- if (spec.Ordered_) {
- specNode["spec"]["ordered"] = *spec.Ordered_;
- }
-
- BuildCommonUserOperationPart(spec, &specNode["spec"]);
- BuildMapJobCountOperationPart(spec, &specNode["spec"]);
- BuildPartitionCountOperationPart(spec, &specNode["spec"]);
- BuildIntermediateDataPart(spec, &specNode["spec"]);
- BuildDataSizePerSortJobPart(spec, &specNode["spec"]);
-
- auto startOperation = [
- operation=operation.Get(),
- spec=MergeSpec(std::move(specNode), preparer->GetContext().Config->Spec, options),
- preparer,
- mapper,
- reduceCombiner,
- reducer,
- inputs=operationIo.Inputs,
- allOutputs
- ] () {
- auto operationId = preparer->StartOperation(operation, "map_reduce", spec);
-
- LogJob(operationId, mapper.Get(), "mapper");
- LogJob(operationId, reduceCombiner.Get(), "reduce_combiner");
- LogJob(operationId, reducer.Get(), "reducer");
- LogYPaths(operationId, inputs, "input");
- LogYPaths(operationId, allOutputs, "output");
-
- return operationId;
- };
-
- operation->SetDelayedStartFunction(std::move(startOperation));
-}
-
-void ExecuteMapReduce(
- const TOperationPtr& operation,
- const TOperationPreparerPtr& preparer,
- const TMapReduceOperationSpec& spec_,
- const ::TIntrusivePtr<IStructuredJob>& mapper,
- const ::TIntrusivePtr<IStructuredJob>& reduceCombiner,
- const ::TIntrusivePtr<IStructuredJob>& reducer,
- const TOperationOptions& options)
-{
- YT_LOG_DEBUG("Starting map-reduce operation (PreparationId: %v)",
- preparer->GetPreparationId());
- TMapReduceOperationSpec spec = spec_;
-
- TMapReduceOperationIo operationIo;
- auto structuredInputs = CanonizeStructuredTableList(preparer->GetContext(), spec.GetStructuredInputs());
- auto structuredMapOutputs = CanonizeStructuredTableList(preparer->GetContext(), spec.GetStructuredMapOutputs());
- auto structuredOutputs = CanonizeStructuredTableList(preparer->GetContext(), spec.GetStructuredOutputs());
-
- const bool inferOutputSchema = options.InferOutputSchema_.GetOrElse(preparer->GetContext().Config->InferTableSchema);
-
- TVector<TTableSchema> currentInferenceResult;
-
- auto fixSpec = [&] (const TFormat& format) {
- if (format.IsYamredDsv()) {
- spec.SortBy_.Parts_.clear();
- spec.ReduceBy_.Parts_.clear();
-
- const TYamredDsvAttributes attributes = format.GetYamredDsvAttributes();
- for (auto& column : attributes.KeyColumnNames) {
- spec.SortBy_.Parts_.push_back(column);
- spec.ReduceBy_.Parts_.push_back(column);
- }
- for (const auto& column : attributes.SubkeyColumnNames) {
- spec.SortBy_.Parts_.push_back(column);
- }
- }
- };
-
- VerifyHasElements(structuredInputs, "inputs");
-
- TFormatBuilder formatBuilder(
- preparer->GetClientRetryPolicy(),
- preparer->GetContext(),
- preparer->GetTransactionId(),
- options);
-
- if (mapper) {
- auto mapperOutputDescription =
- spec.GetIntermediateMapOutputDescription()
- .GetOrElse(TUnspecifiedTableStructure());
- TStructuredJobTableList mapperOutput = {
- TStructuredJobTable::Intermediate(mapperOutputDescription),
- };
-
- for (const auto& table : structuredMapOutputs) {
- mapperOutput.push_back(TStructuredJobTable{table.Description, table.RichYPath});
- }
-
- auto hints = spec.MapperFormatHints_;
-
- auto mapperInferenceResult = PrepareOperation<TStructuredJobTableList>(
- *mapper,
- TOperationPreparationContext(
- structuredInputs,
- mapperOutput,
- preparer->GetContext(),
- preparer->GetClientRetryPolicy(),
- preparer->GetTransactionId()),
- &structuredInputs,
- /* outputs */ nullptr,
- hints);
-
- auto nodeReaderFormat = NodeReaderFormatFromHintAndGlobalConfig(spec.MapperFormatHints_);
-
- auto [inputFormat, inputFormatConfig] = formatBuilder.CreateFormat(
- *mapper,
- EIODirection::Input,
- structuredInputs,
- hints.InputFormatHints_,
- nodeReaderFormat,
- /* allowFormatFromTableAttribute */ true);
-
- auto [outputFormat, outputFormatConfig] = formatBuilder.CreateFormat(
- *mapper,
- EIODirection::Output,
- mapperOutput,
- hints.OutputFormatHints_,
- ENodeReaderFormat::Yson,
- /* allowFormatFromTableAttribute */ false);
-
- operationIo.MapperJobFiles = CreateFormatConfig(inputFormatConfig, outputFormatConfig);
- operationIo.MapperInputFormat = inputFormat;
- operationIo.MapperOutputFormat = outputFormat;
-
- Y_VERIFY(mapperInferenceResult.size() >= 1);
- currentInferenceResult = TVector<TTableSchema>{mapperInferenceResult[0]};
- // The first output as it corresponds to the intermediate data.
- TVector<TTableSchema> additionalOutputsInferenceResult(mapperInferenceResult.begin() + 1, mapperInferenceResult.end());
-
- operationIo.MapOutputs = GetPathList(
- structuredMapOutputs,
- additionalOutputsInferenceResult,
- inferOutputSchema);
- }
-
- if (reduceCombiner) {
- const bool isFirstStep = !mapper;
- TStructuredJobTableList inputs;
- if (isFirstStep) {
- inputs = structuredInputs;
- } else {
- auto reduceCombinerIntermediateInput =
- spec.GetIntermediateReduceCombinerInputDescription()
- .GetOrElse(TUnspecifiedTableStructure());
- inputs = {
- TStructuredJobTable::Intermediate(reduceCombinerIntermediateInput),
- };
- }
-
- auto reduceCombinerOutputDescription = spec.GetIntermediateReduceCombinerOutputDescription()
- .GetOrElse(TUnspecifiedTableStructure());
-
- TStructuredJobTableList outputs = {
- TStructuredJobTable::Intermediate(reduceCombinerOutputDescription),
- };
-
- auto hints = spec.ReduceCombinerFormatHints_;
-
- if (isFirstStep) {
- currentInferenceResult = PrepareOperation<TStructuredJobTableList>(
- *reduceCombiner,
- TOperationPreparationContext(
- inputs,
- outputs,
- preparer->GetContext(),
- preparer->GetClientRetryPolicy(),
- preparer->GetTransactionId()),
- &inputs,
- /* outputs */ nullptr,
- hints);
- } else {
- currentInferenceResult = PrepareOperation<TStructuredJobTableList>(
- *reduceCombiner,
- TSpeculativeOperationPreparationContext(
- currentInferenceResult,
- inputs,
- outputs),
- /* inputs */ nullptr,
- /* outputs */ nullptr,
- hints);
- }
-
- auto [inputFormat, inputFormatConfig] = formatBuilder.CreateFormat(
- *reduceCombiner,
- EIODirection::Input,
- inputs,
- hints.InputFormatHints_,
- ENodeReaderFormat::Yson,
- /* allowFormatFromTableAttribute = */ isFirstStep);
-
- auto [outputFormat, outputFormatConfig] = formatBuilder.CreateFormat(
- *reduceCombiner,
- EIODirection::Output,
- outputs,
- hints.OutputFormatHints_,
- ENodeReaderFormat::Yson,
- /* allowFormatFromTableAttribute = */ false);
-
- operationIo.ReduceCombinerJobFiles = CreateFormatConfig(inputFormatConfig, outputFormatConfig);
- operationIo.ReduceCombinerInputFormat = inputFormat;
- operationIo.ReduceCombinerOutputFormat = outputFormat;
-
- if (isFirstStep) {
- fixSpec(*operationIo.ReduceCombinerInputFormat);
- }
- }
-
- const bool isFirstStep = (!mapper && !reduceCombiner);
- TStructuredJobTableList reducerInputs;
- if (isFirstStep) {
- reducerInputs = structuredInputs;
- } else {
- auto reducerInputDescription =
- spec.GetIntermediateReducerInputDescription()
- .GetOrElse(TUnspecifiedTableStructure());
- reducerInputs = {
- TStructuredJobTable::Intermediate(reducerInputDescription),
- };
- }
-
- auto hints = spec.ReducerFormatHints_;
-
- TVector<TTableSchema> reducerInferenceResult;
- if (isFirstStep) {
- reducerInferenceResult = PrepareOperation(
- *reducer,
- TOperationPreparationContext(
- structuredInputs,
- structuredOutputs,
- preparer->GetContext(),
- preparer->GetClientRetryPolicy(),
- preparer->GetTransactionId()),
- &structuredInputs,
- &structuredOutputs,
- hints);
- } else {
- reducerInferenceResult = PrepareOperation<TStructuredJobTableList>(
- *reducer,
- TSpeculativeOperationPreparationContext(
- currentInferenceResult,
- reducerInputs,
- structuredOutputs),
- /* inputs */ nullptr,
- &structuredOutputs,
- hints);
- }
-
- auto [inputFormat, inputFormatConfig] = formatBuilder.CreateFormat(
- *reducer,
- EIODirection::Input,
- reducerInputs,
- hints.InputFormatHints_,
- ENodeReaderFormat::Yson,
- /* allowFormatFromTableAttribute = */ isFirstStep);
-
- auto [outputFormat, outputFormatConfig] = formatBuilder.CreateFormat(
- *reducer,
- EIODirection::Output,
- ToStructuredJobTableList(spec.GetStructuredOutputs()),
- hints.OutputFormatHints_,
- ENodeReaderFormat::Yson,
- /* allowFormatFromTableAttribute = */ false);
- operationIo.ReducerJobFiles = CreateFormatConfig(inputFormatConfig, outputFormatConfig);
- operationIo.ReducerInputFormat = inputFormat;
- operationIo.ReducerOutputFormat = outputFormat;
-
- if (isFirstStep) {
- fixSpec(operationIo.ReducerInputFormat);
- }
-
- operationIo.Inputs = GetPathList(
- ApplyProtobufColumnFilters(
- structuredInputs,
- *preparer,
- GetColumnsUsedInOperation(spec),
- options),
- /* jobSchemaInferenceResult */ Nothing(),
- /* inferSchema */ false);
-
- operationIo.Outputs = GetPathList(
- structuredOutputs,
- reducerInferenceResult,
- inferOutputSchema);
-
- VerifyHasElements(operationIo.Outputs, "outputs");
-
- return DoExecuteMapReduce(
- operation,
- preparer,
- operationIo,
- spec,
- mapper,
- reduceCombiner,
- reducer,
- options);
-}
-
-void ExecuteRawMapReduce(
- const TOperationPtr& operation,
- const TOperationPreparerPtr& preparer,
- const TRawMapReduceOperationSpec& spec,
- const ::TIntrusivePtr<IRawJob>& mapper,
- const ::TIntrusivePtr<IRawJob>& reduceCombiner,
- const ::TIntrusivePtr<IRawJob>& reducer,
- const TOperationOptions& options)
-{
- YT_LOG_DEBUG("Starting raw map-reduce operation (PreparationId: %v)",
- preparer->GetPreparationId());
- TMapReduceOperationIo operationIo;
- operationIo.Inputs = CanonizeYPaths(/* retryPolicy */ nullptr, preparer->GetContext(), spec.GetInputs());
- operationIo.MapOutputs = CanonizeYPaths(/* retryPolicy */ nullptr, preparer->GetContext(), spec.GetMapOutputs());
- operationIo.Outputs = CanonizeYPaths(/* retryPolicy */ nullptr, preparer->GetContext(), spec.GetOutputs());
-
- VerifyHasElements(operationIo.Inputs, "inputs");
- VerifyHasElements(operationIo.Outputs, "outputs");
-
- auto getFormatOrDefault = [&] (const TMaybe<TFormat>& maybeFormat, const TMaybe<TFormat> stageDefaultFormat, const char* formatName) {
- if (maybeFormat) {
- return *maybeFormat;
- } else if (stageDefaultFormat) {
- return *stageDefaultFormat;
- } else {
- ythrow TApiUsageError() << "Cannot derive " << formatName;
- }
- };
-
- if (mapper) {
- operationIo.MapperInputFormat = getFormatOrDefault(spec.MapperInputFormat_, spec.MapperFormat_, "mapper input format");
- operationIo.MapperOutputFormat = getFormatOrDefault(spec.MapperOutputFormat_, spec.MapperFormat_, "mapper output format");
- }
-
- if (reduceCombiner) {
- operationIo.ReduceCombinerInputFormat = getFormatOrDefault(spec.ReduceCombinerInputFormat_, spec.ReduceCombinerFormat_, "reduce combiner input format");
- operationIo.ReduceCombinerOutputFormat = getFormatOrDefault(spec.ReduceCombinerOutputFormat_, spec.ReduceCombinerFormat_, "reduce combiner output format");
- }
-
- operationIo.ReducerInputFormat = getFormatOrDefault(spec.ReducerInputFormat_, spec.ReducerFormat_, "reducer input format");
- operationIo.ReducerOutputFormat = getFormatOrDefault(spec.ReducerOutputFormat_, spec.ReducerFormat_, "reducer output format");
-
- return DoExecuteMapReduce(
- operation,
- preparer,
- operationIo,
- spec,
- mapper,
- reduceCombiner,
- reducer,
- options);
-}
-
-void ExecuteSort(
- const TOperationPtr& operation,
- const TOperationPreparerPtr& preparer,
- const TSortOperationSpec& spec,
- const TOperationOptions& options)
-{
- YT_LOG_DEBUG("Starting sort operation (PreparationId: %v)",
- preparer->GetPreparationId());
- auto inputs = CanonizeYPaths(/* retryPolicy */ nullptr, preparer->GetContext(), spec.Inputs_);
- auto output = CanonizeYPath(nullptr, preparer->GetContext(), spec.Output_);
-
- if (options.CreateOutputTables_) {
- CheckInputTablesExist(*preparer, inputs);
- CreateOutputTable(*preparer, output);
- }
-
- TNode specNode = BuildYsonNodeFluently()
- .BeginMap().Item("spec").BeginMap()
- .Item("input_table_paths").List(inputs)
- .Item("output_table_path").Value(output)
- .Item("sort_by").Value(spec.SortBy_)
- .DoIf(spec.SchemaInferenceMode_.Defined(), [&] (TFluentMap fluent) {
- fluent.Item("schema_inference_mode").Value(ToString(*spec.SchemaInferenceMode_));
- })
- .Do(std::bind(BuildCommonOperationPart<TSortOperationSpec>, preparer->GetContext().Config, spec, options, std::placeholders::_1))
- .EndMap().EndMap();
-
- BuildPartitionCountOperationPart(spec, &specNode["spec"]);
- BuildPartitionJobCountOperationPart(spec, &specNode["spec"]);
- BuildIntermediateDataPart(spec, &specNode["spec"]);
-
- auto startOperation = [
- operation=operation.Get(),
- spec=MergeSpec(std::move(specNode), preparer->GetContext().Config->Spec, options),
- preparer,
- inputs,
- output
- ] () {
- auto operationId = preparer->StartOperation(operation, "sort", spec);
-
- LogYPaths(operationId, inputs, "input");
- LogYPath(operationId, output, "output");
-
- return operationId;
- };
-
- operation->SetDelayedStartFunction(std::move(startOperation));
-}
-
-void ExecuteMerge(
- const TOperationPtr& operation,
- const TOperationPreparerPtr& preparer,
- const TMergeOperationSpec& spec,
- const TOperationOptions& options)
-{
- YT_LOG_DEBUG("Starting merge operation (PreparationId: %v)",
- preparer->GetPreparationId());
- auto inputs = CanonizeYPaths(/* retryPolicy */ nullptr, preparer->GetContext(), spec.Inputs_);
- auto output = CanonizeYPath(nullptr, preparer->GetContext(), spec.Output_);
-
- if (options.CreateOutputTables_) {
- CheckInputTablesExist(*preparer, inputs);
- CreateOutputTable(*preparer, output);
- }
-
- TNode specNode = BuildYsonNodeFluently()
- .BeginMap().Item("spec").BeginMap()
- .Item("input_table_paths").List(inputs)
- .Item("output_table_path").Value(output)
- .Item("mode").Value(ToString(spec.Mode_))
- .Item("combine_chunks").Value(spec.CombineChunks_)
- .Item("force_transform").Value(spec.ForceTransform_)
- .Item("merge_by").Value(spec.MergeBy_)
- .DoIf(spec.SchemaInferenceMode_.Defined(), [&] (TFluentMap fluent) {
- fluent.Item("schema_inference_mode").Value(ToString(*spec.SchemaInferenceMode_));
- })
- .Do(std::bind(BuildCommonOperationPart<TMergeOperationSpec>, preparer->GetContext().Config, spec, options, std::placeholders::_1))
- .EndMap().EndMap();
-
- BuildJobCountOperationPart(spec, &specNode["spec"]);
-
- auto startOperation = [
- operation=operation.Get(),
- spec=MergeSpec(std::move(specNode), preparer->GetContext().Config->Spec, options),
- preparer,
- inputs,
- output
- ] () {
- auto operationId = preparer->StartOperation(operation, "merge", spec);
-
- LogYPaths(operationId, inputs, "input");
- LogYPath(operationId, output, "output");
-
- return operationId;
- };
-
- operation->SetDelayedStartFunction(std::move(startOperation));
-}
-
-void ExecuteErase(
- const TOperationPtr& operation,
- const TOperationPreparerPtr& preparer,
- const TEraseOperationSpec& spec,
- const TOperationOptions& options)
-{
- YT_LOG_DEBUG("Starting erase operation (PreparationId: %v)",
- preparer->GetPreparationId());
- auto tablePath = CanonizeYPath(nullptr, preparer->GetContext(), spec.TablePath_);
-
- TNode specNode = BuildYsonNodeFluently()
- .BeginMap().Item("spec").BeginMap()
- .Item("table_path").Value(tablePath)
- .Item("combine_chunks").Value(spec.CombineChunks_)
- .DoIf(spec.SchemaInferenceMode_.Defined(), [&] (TFluentMap fluent) {
- fluent.Item("schema_inference_mode").Value(ToString(*spec.SchemaInferenceMode_));
- })
- .Do(std::bind(BuildCommonOperationPart<TEraseOperationSpec>, preparer->GetContext().Config, spec, options, std::placeholders::_1))
- .EndMap().EndMap();
-
- auto startOperation = [
- operation=operation.Get(),
- spec=MergeSpec(std::move(specNode), preparer->GetContext().Config->Spec, options),
- preparer,
- tablePath
- ] () {
- auto operationId = preparer->StartOperation(operation, "erase", spec);
-
- LogYPath(operationId, tablePath, "table_path");
-
- return operationId;
- };
-
- operation->SetDelayedStartFunction(std::move(startOperation));
-}
-
-void ExecuteRemoteCopy(
- const TOperationPtr& operation,
- const TOperationPreparerPtr& preparer,
- const TRemoteCopyOperationSpec& spec,
- const TOperationOptions& options)
-{
- YT_LOG_DEBUG("Starting remote copy operation (PreparationId: %v)",
- preparer->GetPreparationId());
- auto inputs = CanonizeYPaths(/* retryPolicy */ nullptr, preparer->GetContext(), spec.Inputs_);
- auto output = CanonizeYPath(nullptr, preparer->GetContext(), spec.Output_);
-
- if (options.CreateOutputTables_) {
- CreateOutputTable(*preparer, output);
- }
-
- Y_ENSURE_EX(!spec.ClusterName_.empty(), TApiUsageError() << "ClusterName parameter is required");
-
- TNode specNode = BuildYsonNodeFluently()
- .BeginMap().Item("spec").BeginMap()
- .Item("cluster_name").Value(spec.ClusterName_)
- .Item("input_table_paths").List(inputs)
- .Item("output_table_path").Value(output)
- .DoIf(spec.NetworkName_.Defined(), [&] (TFluentMap fluent) {
- fluent.Item("network_name").Value(*spec.NetworkName_);
- })
- .DoIf(spec.SchemaInferenceMode_.Defined(), [&] (TFluentMap fluent) {
- fluent.Item("schema_inference_mode").Value(ToString(*spec.SchemaInferenceMode_));
- })
- .Item("copy_attributes").Value(spec.CopyAttributes_)
- .DoIf(!spec.AttributeKeys_.empty(), [&] (TFluentMap fluent) {
- Y_ENSURE_EX(spec.CopyAttributes_, TApiUsageError() <<
- "Specifying nonempty AttributeKeys in RemoteCopy "
- "doesn't make sense without CopyAttributes == true");
- fluent.Item("attribute_keys").List(spec.AttributeKeys_);
- })
- .Do(std::bind(BuildCommonOperationPart<TRemoteCopyOperationSpec>, preparer->GetContext().Config, spec, options, std::placeholders::_1))
- .EndMap().EndMap();
-
- auto startOperation = [
- operation=operation.Get(),
- spec=MergeSpec(specNode, preparer->GetContext().Config->Spec, options),
- preparer,
- inputs,
- output
- ] () {
- auto operationId = preparer->StartOperation(operation, "remote_copy", spec);
-
- LogYPaths(operationId, inputs, "input");
- LogYPath(operationId, output, "output");
-
- return operationId;
- };
-
- operation->SetDelayedStartFunction(std::move(startOperation));
-}
-
-void ExecuteVanilla(
- const TOperationPtr& operation,
- const TOperationPreparerPtr& preparer,
- const TVanillaOperationSpec& spec,
- const TOperationOptions& options)
-{
- YT_LOG_DEBUG("Starting vanilla operation (PreparationId: %v)",
- preparer->GetPreparationId());
-
- auto addTask = [&](TFluentMap fluent, const TVanillaTask& task) {
- Y_VERIFY(task.Job_.Get());
- if (std::holds_alternative<TVoidStructuredRowStream>(task.Job_->GetOutputRowStreamDescription())) {
- Y_ENSURE_EX(task.Outputs_.empty(),
- TApiUsageError() << "Vanilla task with void IVanillaJob doesn't expect output tables");
- TJobPreparer jobPreparer(
- *preparer,
- task.Spec_,
- *task.Job_,
- /* outputTableCount */ 0,
- /* smallFileList */ {},
- options);
- fluent
- .Item(task.Name_).BeginMap()
- .Item("job_count").Value(task.JobCount_)
- .DoIf(task.NetworkProject_.Defined(), [&](TFluentMap fluent) {
- fluent.Item("network_project").Value(*task.NetworkProject_);
- })
- .Do([&] (TFluentMap fluent) {
- BuildUserJobFluently(
- std::cref(jobPreparer),
- /* inputFormat */ Nothing(),
- /* outputFormat */ Nothing(),
- fluent);
- })
- .EndMap();
- } else {
- auto operationIo = CreateSimpleOperationIo(
- *task.Job_,
- *preparer,
- task,
- options,
- false);
- Y_ENSURE_EX(operationIo.Outputs.size() > 0,
- TApiUsageError() << "Vanilla task with IVanillaJob that has table writer expects output tables");
- if (options.CreateOutputTables_) {
- CreateOutputTables(*preparer, operationIo.Outputs);
- }
- TJobPreparer jobPreparer(
- *preparer,
- task.Spec_,
- *task.Job_,
- operationIo.Outputs.size(),
- operationIo.JobFiles,
- options);
- fluent
- .Item(task.Name_).BeginMap()
- .Item("job_count").Value(task.JobCount_)
- .DoIf(task.NetworkProject_.Defined(), [&](TFluentMap fluent) {
- fluent.Item("network_project").Value(*task.NetworkProject_);
- })
- .Do([&] (TFluentMap fluent) {
- BuildUserJobFluently(
- std::cref(jobPreparer),
- /* inputFormat */ Nothing(),
- operationIo.OutputFormat,
- fluent);
- })
- .Item("output_table_paths").List(operationIo.Outputs)
- .Item("job_io").BeginMap()
- .DoIf(!preparer->GetContext().Config->TableWriter.Empty(), [&](TFluentMap fluent) {
- fluent.Item("table_writer").Value(preparer->GetContext().Config->TableWriter);
- })
- .Item("control_attributes").BeginMap()
- .Item("enable_row_index").Value(TNode(true))
- .Item("enable_range_index").Value(TNode(true))
- .EndMap()
- .EndMap()
- .EndMap();
- }
- };
-
- if (options.CreateDebugOutputTables_) {
- CreateDebugOutputTables(spec, *preparer);
- }
-
- TNode specNode = BuildYsonNodeFluently()
- .BeginMap().Item("spec").BeginMap()
- .Item("tasks").DoMapFor(spec.Tasks_, addTask)
- .Do(std::bind(BuildCommonOperationPart<TVanillaOperationSpec>, preparer->GetContext().Config, spec, options, std::placeholders::_1))
- .EndMap().EndMap();
-
- BuildCommonUserOperationPart(spec, &specNode["spec"]);
-
- auto startOperation = [operation=operation.Get(), spec=MergeSpec(std::move(specNode), preparer->GetContext().Config->Spec, options), preparer] () {
- auto operationId = preparer->StartOperation(operation, "vanilla", spec, /* useStartOperationRequest */ true);
- return operationId;
- };
-
- operation->SetDelayedStartFunction(std::move(startOperation));
-}
-
-////////////////////////////////////////////////////////////////////////////////
-
-class TOperation::TOperationImpl
- : public TThrRefBase
-{
-public:
- TOperationImpl(
- IClientRetryPolicyPtr clientRetryPolicy,
- TClientContext context,
- const TMaybe<TOperationId>& operationId = {})
- : ClientRetryPolicy_(clientRetryPolicy)
- , Context_(std::move(context))
- , Id_(operationId)
- , PreparedPromise_(::NThreading::NewPromise<void>())
- , StartedPromise_(::NThreading::NewPromise<void>())
- {
- if (Id_) {
- PreparedPromise_.SetValue();
- StartedPromise_.SetValue();
- } else {
- PreparedPromise_.GetFuture().Subscribe([this_=::TIntrusivePtr(this)] (const ::NThreading::TFuture<void>& preparedResult) {
- try {
- preparedResult.GetValue();
- } catch (...) {
- this_->StartedPromise_.SetException(std::current_exception());
- return;
- }
- });
- }
- }
-
- const TOperationId& GetId() const;
- TString GetWebInterfaceUrl() const;
-
- void OnPrepared();
- void SetDelayedStartFunction(std::function<TOperationId()> start);
- void Start();
- bool IsStarted() const;
- void OnPreparationException(std::exception_ptr e);
-
- TString GetStatus();
- void OnStatusUpdated(const TString& newStatus);
-
- ::NThreading::TFuture<void> GetPreparedFuture();
- ::NThreading::TFuture<void> GetStartedFuture();
- ::NThreading::TFuture<void> Watch(TClientPtr client);
-
- EOperationBriefState GetBriefState();
- TMaybe<TYtError> GetError();
- TJobStatistics GetJobStatistics();
- TMaybe<TOperationBriefProgress> GetBriefProgress();
- void AbortOperation();
- void CompleteOperation();
- void SuspendOperation(const TSuspendOperationOptions& options);
- void ResumeOperation(const TResumeOperationOptions& options);
- TOperationAttributes GetAttributes(const TGetOperationOptions& options);
- void UpdateParameters(const TUpdateOperationParametersOptions& options);
- TJobAttributes GetJob(const TJobId& jobId, const TGetJobOptions& options);
- TListJobsResult ListJobs(const TListJobsOptions& options);
-
- void AsyncFinishOperation(TOperationAttributes operationAttributes);
- void FinishWithException(std::exception_ptr exception);
- void UpdateBriefProgress(TMaybe<TOperationBriefProgress> briefProgress);
- void AnalyzeUnrecognizedSpec(TNode unrecognizedSpec);
-
- const TClientContext& GetContext() const;
-
-private:
- void OnStarted(const TOperationId& operationId);
-
- void UpdateAttributesAndCall(bool needJobStatistics, std::function<void(const TOperationAttributes&)> func);
-
- void SyncFinishOperationImpl(const TOperationAttributes&);
- static void* SyncFinishOperationProc(void* );
-
- void ValidateOperationStarted() const;
-
-private:
- IClientRetryPolicyPtr ClientRetryPolicy_;
- const TClientContext Context_;
- TMaybe<TOperationId> Id_;
- TMutex Lock_;
-
- ::NThreading::TPromise<void> PreparedPromise_;
- ::NThreading::TPromise<void> StartedPromise_;
- TMaybe<::NThreading::TPromise<void>> CompletePromise_;
-
- std::function<TOperationId()> DelayedStartFunction_;
- TString Status_;
- TOperationAttributes Attributes_;
-};
-
-////////////////////////////////////////////////////////////////////////////////
-
-class TOperationPollerItem
- : public IYtPollerItem
-{
-public:
- TOperationPollerItem(::TIntrusivePtr<TOperation::TOperationImpl> operationImpl)
- : OperationImpl_(std::move(operationImpl))
- { }
-
- void PrepareRequest(TRawBatchRequest* batchRequest) override
- {
- auto filter = TOperationAttributeFilter()
- .Add(EOperationAttribute::State)
- .Add(EOperationAttribute::BriefProgress)
- .Add(EOperationAttribute::Result);
-
- if (!UnrecognizedSpecAnalyzed_) {
- filter.Add(EOperationAttribute::UnrecognizedSpec);
- }
-
- OperationState_ = batchRequest->GetOperation(
- OperationImpl_->GetId(),
- TGetOperationOptions().AttributeFilter(filter));
- }
-
- EStatus OnRequestExecuted() override
- {
- try {
- const auto& attributes = OperationState_.GetValue();
- if (!UnrecognizedSpecAnalyzed_ && !attributes.UnrecognizedSpec.Empty()) {
- OperationImpl_->AnalyzeUnrecognizedSpec(*attributes.UnrecognizedSpec);
- UnrecognizedSpecAnalyzed_ = true;
- }
- Y_VERIFY(attributes.BriefState,
- "get_operation for operation %s has not returned \"state\" field",
- GetGuidAsString(OperationImpl_->GetId()).Data());
- if (*attributes.BriefState != EOperationBriefState::InProgress) {
- OperationImpl_->AsyncFinishOperation(attributes);
- return PollBreak;
- } else {
- OperationImpl_->UpdateBriefProgress(attributes.BriefProgress);
- }
- } catch (const TErrorResponse& e) {
- if (!IsRetriable(e)) {
- OperationImpl_->FinishWithException(std::current_exception());
- return PollBreak;
- }
- } catch (const std::exception& e) {
- OperationImpl_->FinishWithException(std::current_exception());
- return PollBreak;
- }
- return PollContinue;
- }
-
- void OnItemDiscarded() override {
- OperationImpl_->FinishWithException(std::make_exception_ptr(yexception() << "Operation cancelled"));
- }
-
-private:
- ::TIntrusivePtr<TOperation::TOperationImpl> OperationImpl_;
- ::NThreading::TFuture<TOperationAttributes> OperationState_;
- bool UnrecognizedSpecAnalyzed_ = false;
-};
-
-////////////////////////////////////////////////////////////////////////////////
-
-const TOperationId& TOperation::TOperationImpl::GetId() const
-{
- ValidateOperationStarted();
- return *Id_;
-}
-
-TString TOperation::TOperationImpl::GetWebInterfaceUrl() const
-{
- ValidateOperationStarted();
- return GetOperationWebInterfaceUrl(Context_.ServerName, *Id_);
-}
-
-void TOperation::TOperationImpl::OnPrepared()
-{
- Y_VERIFY(!PreparedPromise_.HasException() && !PreparedPromise_.HasValue());
- PreparedPromise_.SetValue();
-}
-
-void TOperation::TOperationImpl::SetDelayedStartFunction(std::function<TOperationId()> start)
-{
- DelayedStartFunction_ = std::move(start);
-}
-
-void TOperation::TOperationImpl::Start()
-{
- {
- auto guard = Guard(Lock_);
- if (Id_) {
- ythrow TApiUsageError() << "Start() should not be called on running operations";
- }
- }
- GetPreparedFuture().GetValueSync();
-
- std::function<TOperationId()> startStuff;
- {
- auto guard = Guard(Lock_);
- startStuff.swap(DelayedStartFunction_);
- }
- if (!startStuff) {
- ythrow TApiUsageError() << "Seems that Start() was called multiple times. If not, contact yt@";
- }
-
- TOperationId operationId;
- try {
- operationId = startStuff();
- } catch (...) {
- auto exception = std::current_exception();
- StartedPromise_.SetException(exception);
- std::rethrow_exception(exception);
- }
- OnStarted(operationId);
-}
-
-bool TOperation::TOperationImpl::IsStarted() const {
- auto guard = Guard(Lock_);
- return bool(Id_);
-}
-
-void TOperation::TOperationImpl::OnPreparationException(std::exception_ptr e)
-{
- Y_VERIFY(!PreparedPromise_.HasValue() && !PreparedPromise_.HasException());
- PreparedPromise_.SetException(e);
-}
-
-TString TOperation::TOperationImpl::GetStatus()
-{
- {
- auto guard = Guard(Lock_);
- if (!Id_) {
- return Status_;
- }
- }
- TMaybe<TString> state;
- UpdateAttributesAndCall(false, [&] (const TOperationAttributes& attributes) {
- state = attributes.State;
- });
-
- return "On YT cluster: " + state.GetOrElse("undefined state");
-}
-
-void TOperation::TOperationImpl::OnStatusUpdated(const TString& newStatus)
-{
- auto guard = Guard(Lock_);
- Status_ = newStatus;
-}
-
-::NThreading::TFuture<void> TOperation::TOperationImpl::GetPreparedFuture()
-{
- return PreparedPromise_.GetFuture();
-}
-
-::NThreading::TFuture<void> TOperation::TOperationImpl::GetStartedFuture()
-{
- return StartedPromise_.GetFuture();
-}
-
-::NThreading::TFuture<void> TOperation::TOperationImpl::Watch(TClientPtr client)
-{
- {
- auto guard = Guard(Lock_);
- if (CompletePromise_) {
- return *CompletePromise_;
- }
- CompletePromise_ = ::NThreading::NewPromise<void>();
- }
- GetStartedFuture().Subscribe([
- this_=::TIntrusivePtr(this),
- client=std::move(client)
- ] (const ::NThreading::TFuture<void>& startedResult) {
- try {
- startedResult.GetValue();
- } catch (...) {
- this_->CompletePromise_->SetException(std::current_exception());
- return;
- }
- client->GetYtPoller().Watch(::MakeIntrusive<TOperationPollerItem>(this_));
- auto operationId = this_->GetId();
- auto registry = TAbortableRegistry::Get();
- registry->Add(
- operationId,
- ::MakeIntrusive<TOperationAbortable>(this_->ClientRetryPolicy_, this_->Context_, operationId));
- // We have to own an IntrusivePtr to registry to prevent use-after-free
- auto removeOperation = [registry, operationId] (const ::NThreading::TFuture<void>&) {
- registry->Remove(operationId);
- };
- this_->CompletePromise_->GetFuture().Subscribe(removeOperation);
- });
-
- return *CompletePromise_;
-}
-
-EOperationBriefState TOperation::TOperationImpl::GetBriefState()
-{
- ValidateOperationStarted();
- EOperationBriefState result = EOperationBriefState::InProgress;
- UpdateAttributesAndCall(false, [&] (const TOperationAttributes& attributes) {
- Y_VERIFY(attributes.BriefState,
- "get_operation for operation %s has not returned \"state\" field",
- GetGuidAsString(*Id_).Data());
- result = *attributes.BriefState;
- });
- return result;
-}
-
-TMaybe<TYtError> TOperation::TOperationImpl::GetError()
-{
- ValidateOperationStarted();
- TMaybe<TYtError> result;
- UpdateAttributesAndCall(false, [&] (const TOperationAttributes& attributes) {
- Y_VERIFY(attributes.Result);
- result = attributes.Result->Error;
- });
- return result;
-}
-
-TJobStatistics TOperation::TOperationImpl::GetJobStatistics()
-{
- ValidateOperationStarted();
- TJobStatistics result;
- UpdateAttributesAndCall(true, [&] (const TOperationAttributes& attributes) {
- if (attributes.Progress) {
- result = attributes.Progress->JobStatistics;
- }
- });
- return result;
-}
-
-TMaybe<TOperationBriefProgress> TOperation::TOperationImpl::GetBriefProgress()
-{
- ValidateOperationStarted();
- {
- auto g = Guard(Lock_);
- if (CompletePromise_.Defined()) {
- // Poller do this job for us
- return Attributes_.BriefProgress;
- }
- }
- TMaybe<TOperationBriefProgress> result;
- UpdateAttributesAndCall(false, [&] (const TOperationAttributes& attributes) {
- result = attributes.BriefProgress;
- });
- return result;
-}
-
-void TOperation::TOperationImpl::UpdateBriefProgress(TMaybe<TOperationBriefProgress> briefProgress)
-{
- auto g = Guard(Lock_);
- Attributes_.BriefProgress = std::move(briefProgress);
-}
-
-void TOperation::TOperationImpl::AnalyzeUnrecognizedSpec(TNode unrecognizedSpec)
-{
- static const TVector<TVector<TString>> knownUnrecognizedSpecFieldPaths = {
- {"mapper", "class_name"},
- {"reducer", "class_name"},
- {"reduce_combiner", "class_name"},
- };
-
- auto removeByPath = [] (TNode& node, auto pathBegin, auto pathEnd, auto& removeByPath) {
- if (pathBegin == pathEnd) {
- return;
- }
- if (!node.IsMap()) {
- return;
- }
- auto* child = node.AsMap().FindPtr(*pathBegin);
- if (!child) {
- return;
- }
- removeByPath(*child, std::next(pathBegin), pathEnd, removeByPath);
- if (std::next(pathBegin) == pathEnd || (child->IsMap() && child->Empty())) {
- node.AsMap().erase(*pathBegin);
- }
- };
-
- Y_VERIFY(unrecognizedSpec.IsMap());
- for (const auto& knownFieldPath : knownUnrecognizedSpecFieldPaths) {
- Y_VERIFY(!knownFieldPath.empty());
- removeByPath(unrecognizedSpec, knownFieldPath.cbegin(), knownFieldPath.cend(), removeByPath);
- }
-
- if (!unrecognizedSpec.Empty()) {
- YT_LOG_INFO(
- "WARNING! Unrecognized spec for operation %s is not empty "
- "(fields added by the YT API library are excluded): %s",
- GetGuidAsString(*Id_).Data(),
- NodeToYsonString(unrecognizedSpec).Data());
- }
-}
-
-void TOperation::TOperationImpl::OnStarted(const TOperationId& operationId)
-{
- auto guard = Guard(Lock_);
- Y_VERIFY(!Id_,
- "OnStarted() called with operationId = %s for operation with id %s",
- GetGuidAsString(operationId).Data(),
- GetGuidAsString(*Id_).Data());
- Id_ = operationId;
-
- Y_VERIFY(!StartedPromise_.HasValue() && !StartedPromise_.HasException());
- StartedPromise_.SetValue();
-}
-
-void TOperation::TOperationImpl::UpdateAttributesAndCall(bool needJobStatistics, std::function<void(const TOperationAttributes&)> func)
-{
- {
- auto g = Guard(Lock_);
- if (Attributes_.BriefState
- && *Attributes_.BriefState != EOperationBriefState::InProgress
- && (!needJobStatistics || Attributes_.Progress))
- {
- func(Attributes_);
- return;
- }
- }
-
- TOperationAttributes attributes = NDetail::GetOperation(
- ClientRetryPolicy_->CreatePolicyForGenericRequest(),
- Context_,
- *Id_,
- TGetOperationOptions().AttributeFilter(TOperationAttributeFilter()
- .Add(EOperationAttribute::Result)
- .Add(EOperationAttribute::Progress)
- .Add(EOperationAttribute::State)
- .Add(EOperationAttribute::BriefProgress)));
-
- func(attributes);
-
- Y_ENSURE(attributes.BriefState);
- if (*attributes.BriefState != EOperationBriefState::InProgress) {
- auto g = Guard(Lock_);
- Attributes_ = std::move(attributes);
- }
-}
-
-void TOperation::TOperationImpl::FinishWithException(std::exception_ptr e)
-{
- CompletePromise_->SetException(std::move(e));
-}
-
-void TOperation::TOperationImpl::AbortOperation()
-{
- ValidateOperationStarted();
- NYT::NDetail::AbortOperation(ClientRetryPolicy_->CreatePolicyForGenericRequest(), Context_, *Id_);
-}
-
-void TOperation::TOperationImpl::CompleteOperation()
-{
- ValidateOperationStarted();
- NYT::NDetail::CompleteOperation(ClientRetryPolicy_->CreatePolicyForGenericRequest(), Context_, *Id_);
-}
-
-void TOperation::TOperationImpl::SuspendOperation(const TSuspendOperationOptions& options)
-{
- ValidateOperationStarted();
- NYT::NDetail::SuspendOperation(ClientRetryPolicy_->CreatePolicyForGenericRequest(), Context_, *Id_, options);
-}
-
-void TOperation::TOperationImpl::ResumeOperation(const TResumeOperationOptions& options)
-{
- ValidateOperationStarted();
- NYT::NDetail::ResumeOperation(ClientRetryPolicy_->CreatePolicyForGenericRequest(), Context_, *Id_, options);
-}
-
-TOperationAttributes TOperation::TOperationImpl::GetAttributes(const TGetOperationOptions& options)
-{
- ValidateOperationStarted();
- return NYT::NDetail::GetOperation(ClientRetryPolicy_->CreatePolicyForGenericRequest(), Context_, *Id_, options);
-}
-
-void TOperation::TOperationImpl::UpdateParameters(const TUpdateOperationParametersOptions& options)
-{
- ValidateOperationStarted();
- return NYT::NDetail::UpdateOperationParameters(ClientRetryPolicy_->CreatePolicyForGenericRequest(), Context_, *Id_, options);
-}
-
-TJobAttributes TOperation::TOperationImpl::GetJob(const TJobId& jobId, const TGetJobOptions& options)
-{
- ValidateOperationStarted();
- return NYT::NDetail::GetJob(ClientRetryPolicy_->CreatePolicyForGenericRequest(), Context_, *Id_, jobId, options);
-}
-
-TListJobsResult TOperation::TOperationImpl::ListJobs(const TListJobsOptions& options)
-{
- ValidateOperationStarted();
- return NYT::NDetail::ListJobs(ClientRetryPolicy_->CreatePolicyForGenericRequest(), Context_, *Id_, options);
-}
-
-struct TAsyncFinishOperationsArgs
-{
- ::TIntrusivePtr<TOperation::TOperationImpl> OperationImpl;
- TOperationAttributes OperationAttributes;
-};
-
-void TOperation::TOperationImpl::AsyncFinishOperation(TOperationAttributes operationAttributes)
-{
- auto args = new TAsyncFinishOperationsArgs;
- args->OperationImpl = this;
- args->OperationAttributes = std::move(operationAttributes);
-
- TThread thread(TThread::TParams(&TOperation::TOperationImpl::SyncFinishOperationProc, args).SetName("finish operation"));
- thread.Start();
- thread.Detach();
-}
-
-void* TOperation::TOperationImpl::SyncFinishOperationProc(void* pArgs)
-{
- THolder<TAsyncFinishOperationsArgs> args(static_cast<TAsyncFinishOperationsArgs*>(pArgs));
- args->OperationImpl->SyncFinishOperationImpl(args->OperationAttributes);
- return nullptr;
-}
-
-void TOperation::TOperationImpl::SyncFinishOperationImpl(const TOperationAttributes& attributes)
-{
- {
- auto guard = Guard(Lock_);
- Y_VERIFY(Id_);
- }
- Y_VERIFY(attributes.BriefState,
- "get_operation for operation %s has not returned \"state\" field",
- GetGuidAsString(*Id_).Data());
- Y_VERIFY(*attributes.BriefState != EOperationBriefState::InProgress);
-
- {
- try {
- // `attributes' that came from poller don't have JobStatistics
- // so we call `GetJobStatistics' in order to get it from server
- // and cache inside object.
- GetJobStatistics();
- } catch (const TErrorResponse& ) {
- // But if for any reason we failed to get attributes
- // we complete operation using what we have.
- auto g = Guard(Lock_);
- Attributes_ = attributes;
- }
- }
-
- if (*attributes.BriefState == EOperationBriefState::Completed) {
- CompletePromise_->SetValue();
- } else if (*attributes.BriefState == EOperationBriefState::Aborted || *attributes.BriefState == EOperationBriefState::Failed) {
- Y_VERIFY(attributes.Result && attributes.Result->Error);
- const auto& error = *attributes.Result->Error;
- YT_LOG_ERROR("Operation %v is `%v' with error: %v",
- *Id_,
- ToString(*attributes.BriefState),
- error.FullDescription());
-
- TString additionalExceptionText;
- TVector<TFailedJobInfo> failedJobStderrInfo;
- if (*attributes.BriefState == EOperationBriefState::Failed) {
- try {
- failedJobStderrInfo = NYT::NDetail::GetFailedJobInfo(ClientRetryPolicy_, Context_, *Id_, TGetFailedJobInfoOptions());
- } catch (const std::exception& e) {
- additionalExceptionText = "Cannot get job stderrs: ";
- additionalExceptionText += e.what();
- }
- }
- CompletePromise_->SetException(
- std::make_exception_ptr(
- TOperationFailedError(
- *attributes.BriefState == EOperationBriefState::Failed
- ? TOperationFailedError::Failed
- : TOperationFailedError::Aborted,
- *Id_,
- error,
- failedJobStderrInfo) << additionalExceptionText));
- }
-}
-
-void TOperation::TOperationImpl::ValidateOperationStarted() const
-{
- auto guard = Guard(Lock_);
- if (!Id_) {
- ythrow TApiUsageError() << "Operation is not started";
- }
-}
-
-const TClientContext& TOperation::TOperationImpl::GetContext() const
-{
- return Context_;
-}
-
-////////////////////////////////////////////////////////////////////////////////
-
-TOperation::TOperation(TClientPtr client)
- : Client_(std::move(client))
- , Impl_(::MakeIntrusive<TOperationImpl>(Client_->GetRetryPolicy(), Client_->GetContext()))
-{
-}
-
-TOperation::TOperation(TOperationId id, TClientPtr client)
- : Client_(std::move(client))
- , Impl_(::MakeIntrusive<TOperationImpl>(Client_->GetRetryPolicy(), Client_->GetContext(), id))
-{
-}
-
-const TOperationId& TOperation::GetId() const
-{
- return Impl_->GetId();
-}
-
-TString TOperation::GetWebInterfaceUrl() const
-{
- return Impl_->GetWebInterfaceUrl();
-}
-
-void TOperation::OnPrepared()
-{
- Impl_->OnPrepared();
-}
-
-void TOperation::SetDelayedStartFunction(std::function<TOperationId()> start)
-{
- Impl_->SetDelayedStartFunction(std::move(start));
-}
-
-void TOperation::Start()
-{
- Impl_->Start();
-}
-
-bool TOperation::IsStarted() const
-{
- return Impl_->IsStarted();
-}
-
-void TOperation::OnPreparationException(std::exception_ptr e)
-{
- Impl_->OnPreparationException(std::move(e));
-}
-
-TString TOperation::GetStatus() const
-{
- return Impl_->GetStatus();
-}
-
-void TOperation::OnStatusUpdated(const TString& newStatus)
-{
- Impl_->OnStatusUpdated(newStatus);
-}
-
-::NThreading::TFuture<void> TOperation::GetPreparedFuture()
-{
- return Impl_->GetPreparedFuture();
-}
-
-::NThreading::TFuture<void> TOperation::GetStartedFuture()
-{
- return Impl_->GetStartedFuture();
-}
-
-::NThreading::TFuture<void> TOperation::Watch()
-{
- return Impl_->Watch(Client_);
-}
-
-TVector<TFailedJobInfo> TOperation::GetFailedJobInfo(const TGetFailedJobInfoOptions& options)
-{
- return NYT::NDetail::GetFailedJobInfo(Client_->GetRetryPolicy(), Client_->GetContext(), GetId(), options);
-}
-
-EOperationBriefState TOperation::GetBriefState()
-{
- return Impl_->GetBriefState();
-}
-
-TMaybe<TYtError> TOperation::GetError()
-{
- return Impl_->GetError();
-}
-
-TJobStatistics TOperation::GetJobStatistics()
-{
- return Impl_->GetJobStatistics();
-}
-
-TMaybe<TOperationBriefProgress> TOperation::GetBriefProgress()
-{
- return Impl_->GetBriefProgress();
-}
-
-void TOperation::AbortOperation()
-{
- Impl_->AbortOperation();
-}
-
-void TOperation::CompleteOperation()
-{
- Impl_->CompleteOperation();
-}
-
-void TOperation::SuspendOperation(const TSuspendOperationOptions& options)
-{
- Impl_->SuspendOperation(options);
-}
-
-void TOperation::ResumeOperation(const TResumeOperationOptions& options)
-{
- Impl_->ResumeOperation(options);
-}
-
-TOperationAttributes TOperation::GetAttributes(const TGetOperationOptions& options)
-{
- return Impl_->GetAttributes(options);
-}
-
-void TOperation::UpdateParameters(const TUpdateOperationParametersOptions& options)
-{
- Impl_->UpdateParameters(options);
-}
-
-TJobAttributes TOperation::GetJob(const TJobId& jobId, const TGetJobOptions& options)
-{
- return Impl_->GetJob(jobId, options);
-}
-
-TListJobsResult TOperation::ListJobs(const TListJobsOptions& options)
-{
- return Impl_->ListJobs(options);
-}
-
-////////////////////////////////////////////////////////////////////////////////
-
-struct TAsyncPrepareAndStartOperationArgs
-{
- std::function<void()> PrepareAndStart;
-};
-
-void* SyncPrepareAndStartOperation(void* pArgs)
-{
- THolder<TAsyncPrepareAndStartOperationArgs> args(static_cast<TAsyncPrepareAndStartOperationArgs*>(pArgs));
- args->PrepareAndStart();
- return nullptr;
-}
-
-::TIntrusivePtr<TOperation> ProcessOperation(
- NYT::NDetail::TClientPtr client,
- std::function<void()> prepare,
- ::TIntrusivePtr<TOperation> operation,
- const TOperationOptions& options)
-{
- auto prepareAndStart = [prepare = std::move(prepare), operation, mode = options.StartOperationMode_] () {
- try {
- prepare();
- operation->OnPrepared();
- } catch (...) {
- operation->OnPreparationException(std::current_exception());
- }
- if (mode >= TOperationOptions::EStartOperationMode::AsyncStart) {
- try {
- operation->Start();
- } catch (...) { }
- }
- };
- if (options.StartOperationMode_ >= TOperationOptions::EStartOperationMode::SyncStart) {
- prepareAndStart();
- WaitIfRequired(operation, client, options);
- } else {
- auto args = new TAsyncPrepareAndStartOperationArgs;
- args->PrepareAndStart = std::move(prepareAndStart);
-
- TThread thread(TThread::TParams(SyncPrepareAndStartOperation, args).SetName("prepare and start operation"));
- thread.Start();
- thread.Detach();
- }
- return operation;
-}
-
-void WaitIfRequired(const TOperationPtr& operation, const TClientPtr& client, const TOperationOptions& options)
-{
- auto retryPolicy = client->GetRetryPolicy();
- auto context = client->GetContext();
- if (options.StartOperationMode_ >= TOperationOptions::EStartOperationMode::SyncStart) {
- operation->GetStartedFuture().GetValueSync();
- }
- if (options.StartOperationMode_ == TOperationOptions::EStartOperationMode::SyncWait) {
- auto finishedFuture = operation->Watch();
- TWaitProxy::Get()->WaitFuture(finishedFuture);
- finishedFuture.GetValue();
- if (context.Config->WriteStderrSuccessfulJobs) {
- auto stderrs = GetJobsStderr(retryPolicy, context, operation->GetId());
- for (const auto& jobStderr : stderrs) {
- if (!jobStderr.empty()) {
- Cerr << jobStderr << '\n';
- }
- }
- }
- }
-}
-
-////////////////////////////////////////////////////////////////////////////////
-
-void ResetUseClientProtobuf(const char* methodName)
-{
- Cerr << "WARNING! OPTION `TConfig::UseClientProtobuf' IS RESET TO `true'; "
- << "IT CAN DETERIORATE YOUR CODE PERFORMANCE!!! DON'T USE DEPRECATED METHOD `"
- << "TOperationIOSpec::" << methodName << "' TO AVOID THIS RESET" << Endl;
- // Give users some time to contemplate about usage of deprecated functions.
- Cerr << "Sleeping for 5 seconds..." << Endl;
- Sleep(TDuration::Seconds(5));
- TConfig::Get()->UseClientProtobuf = true;
-}
-
-} // namespace NDetail
-
-////////////////////////////////////////////////////////////////////////////////
-
-::TIntrusivePtr<INodeReaderImpl> CreateJobNodeReader(TRawTableReaderPtr rawTableReader)
-{
- if (auto schema = NDetail::GetJobInputSkiffSchema()) {
- return new NDetail::TSkiffTableReader(rawTableReader, schema);
- } else {
- return new TNodeTableReader(rawTableReader);
- }
-}
-
-::TIntrusivePtr<IYaMRReaderImpl> CreateJobYaMRReader(TRawTableReaderPtr rawTableReader)
-{
- return new TYaMRTableReader(rawTableReader);
-}
-
-::TIntrusivePtr<IProtoReaderImpl> CreateJobProtoReader(TRawTableReaderPtr rawTableReader)
-{
- if (TConfig::Get()->UseClientProtobuf) {
- return new TProtoTableReader(
- rawTableReader,
- GetJobInputDescriptors());
- } else {
- return new TLenvalProtoTableReader(
- rawTableReader,
- GetJobInputDescriptors());
- }
-}
-
-::TIntrusivePtr<INodeWriterImpl> CreateJobNodeWriter(THolder<IProxyOutput> rawJobWriter)
-{
- return new TNodeTableWriter(std::move(rawJobWriter));
-}
-
-::TIntrusivePtr<IYaMRWriterImpl> CreateJobYaMRWriter(THolder<IProxyOutput> rawJobWriter)
-{
- return new TYaMRTableWriter(std::move(rawJobWriter));
-}
-
-::TIntrusivePtr<IProtoWriterImpl> CreateJobProtoWriter(THolder<IProxyOutput> rawJobWriter)
-{
- if (TConfig::Get()->UseClientProtobuf) {
- return new TProtoTableWriter(
- std::move(rawJobWriter),
- GetJobOutputDescriptors());
- } else {
- return new TLenvalProtoTableWriter(
- std::move(rawJobWriter),
- GetJobOutputDescriptors());
- }
-}
-
-////////////////////////////////////////////////////////////////////////////////
-
-} // namespace NYT
generated by cgit v1.3 (git 2.53.0) at 2026-06-26 08:40:10 +0000