operation.h

Go to the documentation of this file.

#pragma once

 
#include "client_method_options.h"
#include "errors.h"
#include "io.h"
#include "job_statistics.h"
#include "job_counters.h"
 
#include <library/cpp/threading/future/future.h>
#include <library/cpp/type_info/type_info.h>
 
#include <util/datetime/base.h>
#include <util/generic/vector.h>
#include <util/generic/maybe.h>
#include <util/system/file.h>
#include <util/system/types.h>
 
#include <variant>
 
namespace NYT {


struct TUnspecifiedTableStructure
{ };

struct TProtobufTableStructure
{
    const ::google::protobuf::Descriptor* Descriptor = nullptr;
};
 

using TTableStructure = std::variant<
    TUnspecifiedTableStructure,
    TProtobufTableStructure
>;
 
bool operator==(const TUnspecifiedTableStructure&, const TUnspecifiedTableStructure&);
bool operator==(const TProtobufTableStructure& lhs, const TProtobufTableStructure& rhs);

struct TStructuredTablePath
{
    TStructuredTablePath(TRichYPath richYPath = TRichYPath(), TTableStructure description = TUnspecifiedTableStructure())
        : RichYPath(std::move(richYPath))
        , Description(std::move(description))
    { }
 
    TStructuredTablePath(TRichYPath richYPath, const ::google::protobuf::Descriptor* descriptor)
        : RichYPath(std::move(richYPath))
        , Description(TProtobufTableStructure({descriptor}))
    { }
 
    TStructuredTablePath(TYPath path)
        : RichYPath(std::move(path))
        , Description(TUnspecifiedTableStructure())
    { }
 
    TStructuredTablePath(const char* path)
        : RichYPath(path)
        , Description(TUnspecifiedTableStructure())
    { }
 
    TRichYPath RichYPath;
    TTableStructure Description;
};

template <typename TRow>
TStructuredTablePath Structured(TRichYPath richYPath);

template <typename TRow>
TTableStructure StructuredTableDescription();


struct TVoidStructuredRowStream
{ };

struct TTNodeStructuredRowStream
{ };

struct TTYaMRRowStructuredRowStream
{ };

struct TProtobufStructuredRowStream
{
    const ::google::protobuf::Descriptor* Descriptor = nullptr;
};

using TStructuredRowStreamDescription = std::variant<
    TVoidStructuredRowStream,
    TTNodeStructuredRowStream,
    TTYaMRRowStructuredRowStream,
    TProtobufStructuredRowStream
>;


struct TJobBinaryDefault
{ };

struct TJobBinaryLocalPath
{
    TString Path;
    TMaybe<TString> MD5CheckSum;
};

struct TJobBinaryCypressPath
{
    TYPath Path;
    TMaybe<TTransactionId> TransactionId;
};

 

namespace NDetail {
    extern i64 OutputTableCount;
} // namespace NDetail


enum class EAutoMergeMode
{
    Disabled   /* "disabled" */,

    Relaxed    /* "relaxed" */,

    Economy    /* "economy" */,

    Manual     /* "manual" */,
};

class TAutoMergeSpec
{
public:
    using TSelf = TAutoMergeSpec;

    FLUENT_FIELD_OPTION(EAutoMergeMode, Mode);

    FLUENT_FIELD_OPTION(i64, MaxIntermediateChunkCount);

    FLUENT_FIELD_OPTION(i64, ChunkCountPerMergeJob);

    FLUENT_FIELD_OPTION(i64, ChunkSizeThreshold);
};

template <class TDerived>
class TWithAutoMergeSpec
{
public:
    using TSelf = TDerived;

    FLUENT_FIELD_OPTION(TAutoMergeSpec, AutoMerge);
};

class TSchedulerResources
{
public:
    using TSelf = TSchedulerResources;

    FLUENT_FIELD_OPTION_ENCAPSULATED(i64, UserSlots);

    FLUENT_FIELD_OPTION_ENCAPSULATED(i64, Cpu);

    FLUENT_FIELD_OPTION_ENCAPSULATED(i64, Memory);
};

template <class TDerived>
class TUserJobInputFormatHintsBase
{
public:
    using TSelf = TDerived;

    FLUENT_FIELD_OPTION(TFormatHints, InputFormatHints);
};

template <class TDerived>
class TUserJobOutputFormatHintsBase
{
public:
    using TSelf = TDerived;

    FLUENT_FIELD_OPTION(TFormatHints, OutputFormatHints);
};

template <class TDerived>
class TUserJobFormatHintsBase
    : public TUserJobInputFormatHintsBase<TDerived>
    , public TUserJobOutputFormatHintsBase<TDerived>
{
public:
    using TSelf = TDerived;
};

class TUserJobFormatHints
    : public TUserJobFormatHintsBase<TUserJobFormatHints>
{ };

template <class TDerived>
class TRawOperationIoTableSpec
{
public:
    using TSelf = TDerived;

    TDerived& AddInput(const TRichYPath& path);

    TDerived& SetInput(size_t tableIndex, const TRichYPath& path);

    TDerived& AddOutput(const TRichYPath& path);

    TDerived& SetOutput(size_t tableIndex, const TRichYPath& path);

    const TVector<TRichYPath>& GetInputs() const;

    const TVector<TRichYPath>& GetOutputs() const;
 
private:
    TVector<TRichYPath> Inputs_;
    TVector<TRichYPath> Outputs_;
};

template <class TDerived>
struct TSimpleRawOperationIoSpec
    : public TRawOperationIoTableSpec<TDerived>
{
    using TSelf = TDerived;

    FLUENT_FIELD_OPTION(TFormat, Format);

    FLUENT_FIELD_OPTION(TFormat, InputFormat);

    FLUENT_FIELD_OPTION(TFormat, OutputFormat);
};

template <class TDerived>
class TRawMapReduceOperationIoSpec
    : public TRawOperationIoTableSpec<TDerived>
{
public:
    using TSelf = TDerived;

    FLUENT_FIELD_OPTION(TFormat, MapperFormat);

    FLUENT_FIELD_OPTION(TFormat, MapperInputFormat);

    FLUENT_FIELD_OPTION(TFormat, MapperOutputFormat);

    FLUENT_FIELD_OPTION(TFormat, ReduceCombinerFormat);

    FLUENT_FIELD_OPTION(TFormat, ReduceCombinerInputFormat);

    FLUENT_FIELD_OPTION(TFormat, ReduceCombinerOutputFormat);

    FLUENT_FIELD_OPTION(TFormat, ReducerFormat);

    FLUENT_FIELD_OPTION(TFormat, ReducerInputFormat);

    FLUENT_FIELD_OPTION(TFormat, ReducerOutputFormat);

    TDerived& AddMapOutput(const TRichYPath& path);

    TDerived& SetMapOutput(size_t tableIndex, const TRichYPath& path);

    const TVector<TRichYPath>& GetMapOutputs() const;
 
private:
    TVector<TRichYPath> MapOutputs_;
};

class TOperationInputSpecBase
{
public:
    template <class T, class = void>
    struct TFormatAdder;

    template <class T>
    void AddInput(const TRichYPath& path);

    void AddStructuredInput(TStructuredTablePath path);

    template <class T>
    void SetInput(size_t tableIndex, const TRichYPath& path);

    TVector<TRichYPath> Inputs_;

    const TVector<TStructuredTablePath>& GetStructuredInputs() const;
 
private:
    TVector<TStructuredTablePath> StructuredInputs_;
    friend struct TOperationIOSpecBase;
    template <class T>
    friend struct TOperationIOSpec;
};

class TOperationOutputSpecBase
{
public:
    template <class T, class = void>
    struct TFormatAdder;

    template <class T>
    void AddOutput(const TRichYPath& path);

    void AddStructuredOutput(TStructuredTablePath path);

    template <class T>
    void SetOutput(size_t tableIndex, const TRichYPath& path);

    TVector<TRichYPath> Outputs_;

    const TVector<TStructuredTablePath>& GetStructuredOutputs() const;
 
private:
    TVector<TStructuredTablePath> StructuredOutputs_;
    friend struct TOperationIOSpecBase;
    template <class T>
    friend struct TOperationIOSpec;
};

struct TOperationIOSpecBase
    : public TOperationInputSpecBase
    , public TOperationOutputSpecBase
{ };

template <class TDerived>
struct TOperationIOSpec
    : public TOperationIOSpecBase
{
    using TSelf = TDerived;
 
    template <class T>
    TDerived& AddInput(const TRichYPath& path);
 
    TDerived& AddStructuredInput(TStructuredTablePath path);
 
    template <class T>
    TDerived& SetInput(size_t tableIndex, const TRichYPath& path);
 
    template <class T>
    TDerived& AddOutput(const TRichYPath& path);
 
    TDerived& AddStructuredOutput(TStructuredTablePath path);
 
    template <class T>
    TDerived& SetOutput(size_t tableIndex, const TRichYPath& path);
 
 
    // DON'T USE THESE METHODS! They are left solely for backward compatibility.
    // These methods are the only way to do equivalent of (Add/Set)(Input/Output)<Message>
    // but please consider using (Add/Set)(Input/Output)<TConcreteMessage>
    // (where TConcreteMessage is some descendant of Message)
    // because they are faster and better (see https://st.yandex-team.ru/YT-6967)
    TDerived& AddProtobufInput_VerySlow_Deprecated(const TRichYPath& path);
    TDerived& AddProtobufOutput_VerySlow_Deprecated(const TRichYPath& path);
};

template <class TDerived>
struct TOperationSpecBase
{
    using TSelf = TDerived;

    FLUENT_FIELD_OPTION(TDuration, TimeLimit);

    FLUENT_FIELD_OPTION(TString, Alias);

    FLUENT_FIELD_OPTION(TString, Title);

    FLUENT_FIELD_OPTION(TString, Pool);

    FLUENT_FIELD_OPTION(double, Weight);

    FLUENT_OPTIONAL_VECTOR_FIELD_ENCAPSULATED(TString, PoolTree);

    FLUENT_FIELD_OPTION_ENCAPSULATED(TSchedulerResources, ResourceLimits);

    FLUENT_FIELD_OPTION(ui64, MaxFailedJobCount);
 
    // Arbitrary structured information related to the operation.
    FLUENT_FIELD_OPTION(TNode, Annotations);
 
    // Similar to Annotations, shown on the operation page. Recommends concise, human-readable entries to prevent clutter.
    FLUENT_FIELD_OPTION(TNode, Description);
};

template <class TDerived>
struct TUserOperationSpecBase
    : TOperationSpecBase<TDerived>
{
    using TSelf = TDerived;

    FLUENT_FIELD_OPTION(bool, FailOnJobRestart);

    FLUENT_FIELD_OPTION(TYPath, StderrTablePath);

    FLUENT_FIELD_OPTION(TYPath, CoreTablePath);

    FLUENT_FIELD_OPTION(TDuration, WaitingJobTimeout);
};

template <class TDerived>
struct TIntermediateTablesHintSpec
{
    template <class T>
    TDerived& HintMapOutput();

    template <class T>
    TDerived& HintReduceCombinerInput();

    template <class T>
    TDerived& HintReduceCombinerOutput();

    template <class T>
    TDerived& HintReduceInput();

    template <class T>
    TDerived& AddMapOutput(const TRichYPath& path);
 
    TVector<TRichYPath> MapOutputs_;
 
    const TVector<TStructuredTablePath>& GetStructuredMapOutputs() const;
    const TMaybe<TTableStructure>& GetIntermediateMapOutputDescription() const;
    const TMaybe<TTableStructure>& GetIntermediateReduceCombinerInputDescription() const;
    const TMaybe<TTableStructure>& GetIntermediateReduceCombinerOutputDescription() const;
    const TMaybe<TTableStructure>& GetIntermediateReducerInputDescription() const;
 
private:
    TVector<TStructuredTablePath> StructuredMapOutputs_;
    TMaybe<TTableStructure> IntermediateMapOutputDescription_;
    TMaybe<TTableStructure> IntermediateReduceCombinerInputDescription_;
    TMaybe<TTableStructure> IntermediateReduceCombinerOutputDescription_;
    TMaybe<TTableStructure> IntermediateReducerInputDescription_;
};

 
struct TAddLocalFileOptions
{
    using TSelf = TAddLocalFileOptions;

    FLUENT_FIELD_OPTION(TString, PathInJob);

    FLUENT_FIELD_OPTION(TString, MD5CheckSum);

    FLUENT_FIELD_OPTION(bool, BypassArtifactCache);
};


enum class EProfilingBinary
{
    JobProxy       /* "job_proxy" */,

    UserJob        /* "user_job" */,
};

enum class EProfilerType
{
    Cpu            /* "cpu" */,

    Memory         /* "memory" */,

    PeakMemory     /* "peak_memory" */,
};

struct TJobProfilerSpec
{
    using TSelf = TJobProfilerSpec;

    FLUENT_FIELD_OPTION(EProfilingBinary, ProfilingBinary);

    FLUENT_FIELD_OPTION(EProfilerType, ProfilerType);

    FLUENT_FIELD_OPTION(double, ProfilingProbability);

    FLUENT_FIELD_OPTION(int, SamplingFrequency);
};


struct TDiskRequest
{
    using TSelf = TDiskRequest;

    FLUENT_FIELD_OPTION(i64, DiskSpace);

    FLUENT_FIELD_OPTION(i64, InodeCount);

    FLUENT_FIELD_OPTION(TString, Account);

    FLUENT_FIELD_OPTION(TString, MediumName);
};


struct TUserJobSpec
{
    using TSelf = TUserJobSpec;

    TSelf& AddLocalFile(const TLocalFilePath& path, const TAddLocalFileOptions& options = TAddLocalFileOptions());

    TVector<std::tuple<TLocalFilePath, TAddLocalFileOptions>> GetLocalFiles() const;

    FLUENT_VECTOR_FIELD(TRichYPath, File);

    FLUENT_VECTOR_FIELD(TYPath, Layer);

    FLUENT_FIELD_OPTION(TString, DockerImage);

    FLUENT_FIELD_OPTION(i64, MemoryLimit);

    FLUENT_FIELD_OPTION(i64, ExtraTmpfsSize);

    FLUENT_FIELD_OPTION(double, CpuLimit);

    FLUENT_FIELD_OPTION(double, MemoryReserveFactor);

    TUserJobSpec& JobBinaryLocalPath(TString path, TMaybe<TString> md5 = Nothing());

    TUserJobSpec& JobBinaryCypressPath(TString path, TMaybe<TTransactionId> transactionId = Nothing());

    FLUENT_FIELD(TString, JobCommandPrefix);

    FLUENT_FIELD(TString, JobCommandSuffix);

    FLUENT_MAP_FIELD(TString, TString, Environment);

    FLUENT_FIELD_OPTION(ui64, DiskSpaceLimit);

    FLUENT_FIELD_OPTION(ui16, PortCount);

    FLUENT_FIELD_OPTION(TString, NetworkProject);

    FLUENT_FIELD_OPTION(TDuration, JobTimeLimit);

    const TJobBinaryConfig& GetJobBinary() const;

    FLUENT_VECTOR_FIELD(TJobProfilerSpec, JobProfiler);

    FLUENT_FIELD_OPTION(TDiskRequest, DiskRequest);

    FLUENT_FIELD_DEFAULT(bool, EnableRpcProxyInJobProxy, false);
 
private:
    TVector<std::tuple<TLocalFilePath, TAddLocalFileOptions>> LocalFiles_;
    TJobBinaryConfig JobBinary_;
};


template <typename TDerived>
struct TMapOperationSpecBase
    : public TUserOperationSpecBase<TDerived>
    , public TWithAutoMergeSpec<TDerived>
{
    using TSelf = TDerived;

    FLUENT_FIELD(TUserJobSpec, MapperSpec);

    FLUENT_FIELD_OPTION(bool, Ordered);

    FLUENT_FIELD_OPTION(ui32, JobCount);

    FLUENT_FIELD_OPTION(ui64, DataSizePerJob);
};

struct TMapOperationSpec
    : public TMapOperationSpecBase<TMapOperationSpec>
    , public TOperationIOSpec<TMapOperationSpec>
    , public TUserJobFormatHintsBase<TMapOperationSpec>
{ };

struct TRawMapOperationSpec
    : public TMapOperationSpecBase<TRawMapOperationSpec>
    , public TSimpleRawOperationIoSpec<TRawMapOperationSpec>
{ };


template <typename TDerived>
struct TReduceOperationSpecBase
    : public TUserOperationSpecBase<TDerived>
    , public TWithAutoMergeSpec<TDerived>
{
    using TSelf = TDerived;

    FLUENT_FIELD(TUserJobSpec, ReducerSpec);

    FLUENT_FIELD(TSortColumns, SortBy);

    FLUENT_FIELD(TSortColumns, ReduceBy);

    FLUENT_FIELD_OPTION(TSortColumns, JoinBy);

    FLUENT_FIELD_OPTION(bool, EnableKeyGuarantee);

    FLUENT_FIELD_OPTION(ui32, JobCount);

    FLUENT_FIELD_OPTION(ui64, DataSizePerJob);
};

struct TReduceOperationSpec
    : public TReduceOperationSpecBase<TReduceOperationSpec>
    , public TOperationIOSpec<TReduceOperationSpec>
    , public TUserJobFormatHintsBase<TReduceOperationSpec>
{ };

struct TRawReduceOperationSpec
    : public TReduceOperationSpecBase<TRawReduceOperationSpec>
    , public TSimpleRawOperationIoSpec<TRawReduceOperationSpec>
{ };


template <typename TDerived>
struct TJoinReduceOperationSpecBase
    : public TUserOperationSpecBase<TDerived>
{
    using TSelf = TDerived;

    FLUENT_FIELD(TUserJobSpec, ReducerSpec);

    FLUENT_FIELD(TSortColumns, JoinBy);

    FLUENT_FIELD_OPTION(ui32, JobCount);

    FLUENT_FIELD_OPTION(ui64, DataSizePerJob);
};

struct TJoinReduceOperationSpec
    : public TJoinReduceOperationSpecBase<TJoinReduceOperationSpec>
    , public TOperationIOSpec<TJoinReduceOperationSpec>
    , public TUserJobFormatHintsBase<TJoinReduceOperationSpec>
{ };

struct TRawJoinReduceOperationSpec
    : public TJoinReduceOperationSpecBase<TRawJoinReduceOperationSpec>
    , public TSimpleRawOperationIoSpec<TRawJoinReduceOperationSpec>
{ };


template <typename TDerived>
struct TMapReduceOperationSpecBase
    : public TUserOperationSpecBase<TDerived>
{
    using TSelf = TDerived;

    FLUENT_FIELD(TUserJobSpec, MapperSpec);

    FLUENT_FIELD(TUserJobSpec, ReducerSpec);

    FLUENT_FIELD(TUserJobSpec, ReduceCombinerSpec);

    FLUENT_FIELD(TSortColumns, SortBy);

    FLUENT_FIELD(TSortColumns, ReduceBy);

    FLUENT_FIELD_OPTION(ui32, MapJobCount);

    FLUENT_FIELD_OPTION(ui64, DataSizePerMapJob);

    FLUENT_FIELD_OPTION(ui64, PartitionCount);

    FLUENT_FIELD_OPTION(ui64, PartitionDataSize);

    FLUENT_FIELD_OPTION(TString, IntermediateDataAccount);

    FLUENT_FIELD_OPTION(ui64,  IntermediateDataReplicationFactor);

    FLUENT_FIELD_OPTION(ui64, DataSizePerSortJob);

    FLUENT_FIELD_OPTION(bool, Ordered);

    FLUENT_FIELD_OPTION(bool, ForceReduceCombiners);
};

struct TMapReduceOperationSpec
    : public TMapReduceOperationSpecBase<TMapReduceOperationSpec>
    , public TOperationIOSpec<TMapReduceOperationSpec>
    , public TIntermediateTablesHintSpec<TMapReduceOperationSpec>
{
    using TSelf = TMapReduceOperationSpec;

    FLUENT_FIELD_DEFAULT(TUserJobFormatHints, MapperFormatHints, TUserJobFormatHints());

    FLUENT_FIELD_DEFAULT(TUserJobFormatHints, ReducerFormatHints, TUserJobFormatHints());

    FLUENT_FIELD_DEFAULT(TUserJobFormatHints, ReduceCombinerFormatHints, TUserJobFormatHints());
};

struct TRawMapReduceOperationSpec
    : public TMapReduceOperationSpecBase<TRawMapReduceOperationSpec>
    , public TRawMapReduceOperationIoSpec<TRawMapReduceOperationSpec>
{ };


enum class ESchemaInferenceMode : int
{
    FromInput   /* "from_input" */,
    FromOutput  /* "from_output" */,
    Auto        /* "auto" */,
};

struct TSortOperationSpec
    : TOperationSpecBase<TSortOperationSpec>
{
    using TSelf = TSortOperationSpec;

    FLUENT_VECTOR_FIELD(TRichYPath, Input);

    FLUENT_FIELD(TRichYPath, Output);

    FLUENT_FIELD(TSortColumns, SortBy);

    FLUENT_FIELD_OPTION(ui64, PartitionCount);

    FLUENT_FIELD_OPTION(ui64, PartitionDataSize);

    FLUENT_FIELD_OPTION(ui64, PartitionJobCount);

    FLUENT_FIELD_OPTION(ui64, DataSizePerPartitionJob);

    FLUENT_FIELD_OPTION(ESchemaInferenceMode, SchemaInferenceMode);

    FLUENT_FIELD_OPTION(TString, IntermediateDataAccount);

    FLUENT_FIELD_OPTION(ui64, IntermediateDataReplicationFactor);
};
 

enum EMergeMode : int
{
    MM_UNORDERED    /* "unordered" */,
    MM_ORDERED      /* "ordered" */,
    MM_SORTED       /* "sorted" */,
};

struct TMergeOperationSpec
    : TOperationSpecBase<TMergeOperationSpec>
{
    using TSelf = TMergeOperationSpec;

    FLUENT_VECTOR_FIELD(TRichYPath, Input);

    FLUENT_FIELD(TRichYPath, Output);

    FLUENT_FIELD(TSortColumns, MergeBy);

    FLUENT_FIELD_DEFAULT(EMergeMode, Mode, MM_UNORDERED);

    FLUENT_FIELD_DEFAULT(bool, CombineChunks, false);

    FLUENT_FIELD_DEFAULT(bool, ForceTransform, false);

    FLUENT_FIELD_OPTION(ui32, JobCount);

    FLUENT_FIELD_OPTION(ui64, DataSizePerJob);

    FLUENT_FIELD_OPTION(ESchemaInferenceMode, SchemaInferenceMode);
};

struct TEraseOperationSpec
    : TOperationSpecBase<TEraseOperationSpec>
{
    using TSelf = TEraseOperationSpec;

    FLUENT_FIELD(TRichYPath, TablePath);

    FLUENT_FIELD_DEFAULT(bool, CombineChunks, false);

    FLUENT_FIELD_OPTION(ESchemaInferenceMode, SchemaInferenceMode);
};

struct TRemoteCopyOperationSpec
    : TOperationSpecBase<TRemoteCopyOperationSpec>
{
    using TSelf = TRemoteCopyOperationSpec;

    FLUENT_FIELD(TString, ClusterName);

    FLUENT_FIELD_OPTION(TString, NetworkName);

    FLUENT_VECTOR_FIELD(TRichYPath, Input);

    FLUENT_FIELD(TRichYPath, Output);

    FLUENT_FIELD_OPTION(ESchemaInferenceMode, SchemaInferenceMode);

    FLUENT_FIELD_DEFAULT(bool, CopyAttributes, false);

    FLUENT_VECTOR_FIELD(TString, AttributeKey);
 
private:

    FLUENT_FIELD_OPTION(TNode, ClusterConnection);
};
 
class IVanillaJobBase;

struct TVanillaTask
    : public TOperationOutputSpecBase
    , public TUserJobOutputFormatHintsBase<TVanillaTask>
{
    using TSelf = TVanillaTask;

    template <class T>
    TSelf& AddOutput(const TRichYPath& path);

    TSelf& AddStructuredOutput(TStructuredTablePath path);

    template <class T>
    TSelf& SetOutput(size_t tableIndex, const TRichYPath& path);

    FLUENT_FIELD(TString, Name);

    FLUENT_FIELD(::TIntrusivePtr<IVanillaJobBase>, Job);

    FLUENT_FIELD(TUserJobSpec, Spec);

    FLUENT_FIELD(ui64, JobCount);

    FLUENT_FIELD(TMaybe<TString>, NetworkProject);
 
};

struct TVanillaOperationSpec
    : TUserOperationSpecBase<TVanillaOperationSpec>
{
    using TSelf = TVanillaOperationSpec;

    FLUENT_VECTOR_FIELD(TVanillaTask, Task);
};


struct TOperationOptions
{
    using TSelf = TOperationOptions;

    FLUENT_FIELD_OPTION(TNode, Spec);

    enum class EStartOperationMode : int
    {
        AsyncPrepare,

        AsyncStart,

        SyncStart,

        SyncWait,
    };

    FLUENT_FIELD_DEFAULT(EStartOperationMode, StartOperationMode, EStartOperationMode::SyncWait);

    TSelf& Wait(bool value) {
        StartOperationMode_ = value ? EStartOperationMode::SyncWait : EStartOperationMode::SyncStart;
        return static_cast<TSelf&>(*this);
    }

    FLUENT_FIELD_DEFAULT(bool, UseTableFormats, false);

    FLUENT_FIELD(TString, JobCommandPrefix);

    FLUENT_FIELD(TString, JobCommandSuffix);

    FLUENT_FIELD_DEFAULT(bool, MountSandboxInTmpfs, false);

    FLUENT_FIELD_OPTION(TString, FileStorage);

    FLUENT_FIELD_OPTION(TDuration, FileExpirationTimeout);

    FLUENT_FIELD_OPTION(TNode, SecureVault);

    enum class EFileCacheMode : int
    {
        ApiCommandBased,

        CachelessRandomPathUpload,
    };

    FLUENT_FIELD_DEFAULT(EFileCacheMode, FileCacheMode, EFileCacheMode::ApiCommandBased);

    FLUENT_FIELD(TTransactionId, FileStorageTransactionId);

    FLUENT_FIELD_DEFAULT(bool, CreateDebugOutputTables, true);

    FLUENT_FIELD_DEFAULT(bool, CreateOutputTables, true);

    FLUENT_FIELD_OPTION(bool, InferOutputSchema);

    FLUENT_FIELD_DEFAULT(i64, MinJobStateSizeToPassViaFile, 0);
};


const TNode& GetJobSecureVault();


class TRawJobContext
{
public:
    explicit TRawJobContext(size_t outputTableCount);

    const TFile& GetInputFile() const;

    const TVector<TFile>& GetOutputFileList() const;
 
private:
    TFile InputFile_;
    TVector<TFile> OutputFileList_;
};


class ISerializableForJob
{
public:
    virtual ~ISerializableForJob() = default;

    virtual void Save(IOutputStream& stream) const = 0;

    virtual void Load(IInputStream& stream) = 0;
};


class IOperationPreparationContext
{
public:
    virtual ~IOperationPreparationContext() = default;

    virtual int GetInputCount() const = 0;

    virtual int GetOutputCount() const = 0;

    virtual const TTableSchema& GetInputSchema(int index) const = 0;

    virtual const TVector<TTableSchema>& GetInputSchemas() const = 0;

    virtual TMaybe<TYPath> GetInputPath(int index) const = 0;

    virtual TMaybe<TYPath> GetOutputPath(int index) const = 0;
};

class TJobOperationPreparer
{
public:

    class TInputGroup
    {
    public:
        TInputGroup(TJobOperationPreparer& preparer, TVector<int> indices);

        template <typename TRow>
        TInputGroup& Description();

        TInputGroup& ColumnRenaming(const THashMap<TString, TString>& renaming);

        TInputGroup& ColumnFilter(const TVector<TString>& columns);

        TJobOperationPreparer& EndInputGroup();
 
    private:
        TJobOperationPreparer& Preparer_;
        TVector<int> Indices_;
    };

    class TOutputGroup
    {
    public:
        TOutputGroup(TJobOperationPreparer& preparer, TVector<int> indices);

        template <typename TRow>
        TOutputGroup& Description(bool inferSchema = true);

        TOutputGroup& Schema(const TTableSchema& schema);

        TOutputGroup& NoSchema();

        TJobOperationPreparer& EndOutputGroup();
 
    private:
        TJobOperationPreparer& Preparer_;
        TVector<int> Indices_;
    };
 
public:
    explicit TJobOperationPreparer(const IOperationPreparationContext& context);

    TInputGroup BeginInputGroup(int begin, int end);

    template <typename TCont>
    TInputGroup BeginInputGroup(const TCont& indices);

    TOutputGroup BeginOutputGroup(int begin, int end);

    template <typename TCont>
    TOutputGroup BeginOutputGroup(const TCont& indices);

    TJobOperationPreparer& OutputSchema(int tableIndex, TTableSchema schema);

    TJobOperationPreparer& NoOutputSchema(int tableIndex);

    TJobOperationPreparer& InputColumnRenaming(int tableIndex, const THashMap<TString, TString>& renaming);

    TJobOperationPreparer& InputColumnFilter(int tableIndex, const TVector<TString>& columns);

    template <typename TRow>
    TJobOperationPreparer& InputDescription(int tableIndex);

    template <typename TRow>
    TJobOperationPreparer& OutputDescription(int tableIndex, bool inferSchema = true);

    TJobOperationPreparer& NodeOutput(int tableIndex);

    TJobOperationPreparer& InputFormatHints(TFormatHints hints);

    TJobOperationPreparer& OutputFormatHints(TFormatHints hints);

    TJobOperationPreparer& FormatHints(TUserJobFormatHints newFormatHints);


    void Finish();

    TVector<TTableSchema> GetOutputSchemas();

    const TVector<THashMap<TString, TString>>& GetInputColumnRenamings() const;

    const TVector<TMaybe<TVector<TString>>>& GetInputColumnFilters() const;

    const TVector<TMaybe<TTableStructure>>& GetInputDescriptions() const;

    const TVector<TMaybe<TTableStructure>>& GetOutputDescriptions() const;

    const TUserJobFormatHints& GetFormatHints() const;

private:

    void ValidateMissingOutputSchema(int tableIndex) const;

    void ValidateMissingInputDescription(int tableIndex) const;

    void ValidateMissingOutputDescription(int tableIndex) const;

    void ValidateInputTableIndex(int tableIndex, TStringBuf message) const;

    void ValidateOutputTableIndex(int tableIndex, TStringBuf message) const;

    void FinallyValidate() const;
 
    static TTableSchema EmptyNonstrictSchema();
 
private:
    const IOperationPreparationContext& Context_;
 
    TVector<TMaybe<TTableSchema>> OutputSchemas_;
    TVector<THashMap<TString, TString>> InputColumnRenamings_;
    TVector<TMaybe<TVector<TString>>> InputColumnFilters_;
    TVector<TMaybe<TTableStructure>> InputTableDescriptions_;
    TVector<TMaybe<TTableStructure>> OutputTableDescriptions_;
    TUserJobFormatHints FormatHints_ = {};
};


class IJob
    : public TThrRefBase
{
public:

    enum EType
    {
        Mapper,
        Reducer,
        ReducerAggregator,
        RawJob,
        VanillaJob,
    };

    virtual void Save(IOutputStream& stream) const
    {
        Y_UNUSED(stream);
    }

    virtual void Load(IInputStream& stream)
    {
        Y_UNUSED(stream);
    }

    const TNode& SecureVault() const
    {
        return GetJobSecureVault();
    }

    i64 GetOutputTableCount() const
    {
        Y_ABORT_UNLESS(NDetail::OutputTableCount > 0);
 
        return NDetail::OutputTableCount;
    }

    virtual void PrepareOperation(const IOperationPreparationContext& context, TJobOperationPreparer& preparer) const;
};

#define Y_SAVELOAD_JOB(...) \
    virtual void Save(IOutputStream& stream) const override { Save(&stream); } \
    virtual void Load(IInputStream& stream) override { Load(&stream); } \
    Y_PASS_VA_ARGS(Y_SAVELOAD_DEFINE(__VA_ARGS__))

#define Y_SAVELOAD_JOB_DERIVED(Base, ...) \
    virtual void Save(IOutputStream& stream) const override { \
        Base::Save(stream); \
        Save(&stream); \
    } \
    virtual void Load(IInputStream& stream) override { \
        Base::Load(stream); \
        Load(&stream); \
    } \
    Y_PASS_VA_ARGS(Y_SAVELOAD_DEFINE(__VA_ARGS__))


class IStructuredJob
    : public IJob
{
public:
    virtual TRawTableReaderPtr CreateCustomRawJobReader(int fd) const;
    virtual THolder<IProxyOutput> CreateCustomRawJobWriter(size_t outputTableCount) const;
 
    virtual TStructuredRowStreamDescription GetInputRowStreamDescription() const = 0;
    virtual TStructuredRowStreamDescription GetOutputRowStreamDescription() const = 0;
};


TRawTableReaderPtr CreateRawJobReader(int fd = 0);

THolder<IProxyOutput> CreateRawJobWriter(size_t outputTableCount);


class IMapperBase
    : public IStructuredJob
{ };

template <class TR, class TW>
class IMapper
    : public IMapperBase
{
public:
    using TReader = TR;
    using TWriter = TW;
 
public:
    static constexpr EType JobType = EType::Mapper;

    virtual void Start(TWriter* writer)
    {
        Y_UNUSED(writer);
    }

    virtual void Do(TReader* reader, TWriter* writer) = 0;

    virtual void Finish(TWriter* writer)
    {
        Y_UNUSED(writer);
    }
 
    virtual TStructuredRowStreamDescription GetInputRowStreamDescription() const override;
    virtual TStructuredRowStreamDescription GetOutputRowStreamDescription() const override;
};


class IReducerBase
    : public IStructuredJob
{ };

template <class TR, class TW>
class IReducer
    : public IReducerBase
{
public:
    using TReader = TR;
    using TWriter = TW;
 
public:
    static constexpr EType JobType = EType::Reducer;
 
public:

    virtual void Start(TWriter* writer)
    {
        Y_UNUSED(writer);
    }

    virtual void Do(TReader* reader, TWriter* writer) = 0;

    virtual void Finish(TWriter* writer)
    {
        Y_UNUSED(writer);
    }

    void Break();
 
    virtual TStructuredRowStreamDescription GetInputRowStreamDescription() const override;
    virtual TStructuredRowStreamDescription GetOutputRowStreamDescription() const override;
};


template <class TR, class TW>
class IAggregatorReducer
    : public IReducerBase
{
public:
    using TReader = TR;
    using TWriter = TW;
 
public:
    static constexpr EType JobType = EType::ReducerAggregator;
 
public:
    virtual void Start(TWriter* writer)
    {
        Y_UNUSED(writer);
    }

    virtual void Do(TReader* reader, TWriter* writer) = 0;

    virtual void Finish(TWriter* writer)
    {
        Y_UNUSED(writer);
    }
 
    virtual TStructuredRowStreamDescription GetInputRowStreamDescription() const override;
    virtual TStructuredRowStreamDescription GetOutputRowStreamDescription() const override;
};


class IRawJob
    : public IJob
{
public:
    static constexpr EType JobType = EType::RawJob;

    virtual void Do(const TRawJobContext& jobContext) = 0;
};

class ICommandJob
    : public IJob
{
public:
    virtual const TString& GetCommand() const = 0;
};

class TCommandRawJob
    : public IRawJob
    , public ICommandJob
{
public:
    explicit TCommandRawJob(TStringBuf command = {});
 
    const TString& GetCommand() const override;
    void Do(const TRawJobContext& jobContext) override;
 
private:
    TString Command_;
};


class IVanillaJobBase
    : public virtual IStructuredJob
{
public:
    static constexpr EType JobType = EType::VanillaJob;
};
 
template <class TW = void>
class IVanillaJob;

template <>
class IVanillaJob<void>
    : public IVanillaJobBase
{
public:
    virtual void Do() = 0;
 
    virtual TStructuredRowStreamDescription GetInputRowStreamDescription() const override;
    virtual TStructuredRowStreamDescription GetOutputRowStreamDescription() const override;
};

class TCommandVanillaJob
    : public IVanillaJob<>
    , public ICommandJob
{
public:
    explicit TCommandVanillaJob(TStringBuf command = {});
 
    const TString& GetCommand() const override;
    void Do() override;
 
private:
    TString Command_;
};

template <class TW>
class IVanillaJob
    : public IVanillaJobBase
{
public:
    using TWriter = TW;
 
public:
    virtual void Start(TWriter* /* writer */)
    { }

    virtual void Do(TWriter* writer) = 0;

    virtual void Finish(TWriter* /* writer */)
    { }
 
    virtual TStructuredRowStreamDescription GetInputRowStreamDescription() const override;
    virtual TStructuredRowStreamDescription GetOutputRowStreamDescription() const override;
};


enum class EOperationAttribute : int
{
    Id                /* "id" */,
    Type              /* "type" */,
    State             /* "state" */,
    AuthenticatedUser /* "authenticated_user" */,
    StartTime         /* "start_time" */,
    FinishTime        /* "finish_time" */,
    BriefProgress     /* "brief_progress" */,
    BriefSpec         /* "brief_spec" */,
    Suspended         /* "suspended" */,
    Result            /* "result" */,
    Progress          /* "progress" */,
    Events            /* "events" */,
    Spec              /* "spec" */,
    FullSpec          /* "full_spec" */,
    UnrecognizedSpec  /* "unrecognized_spec" */,
    Alerts            /* "alerts" */,
};

struct TOperationAttributeFilter
{
    using TSelf = TOperationAttributeFilter;
 
    TVector<EOperationAttribute> Attributes_;

    TSelf& Add(EOperationAttribute attribute)
    {
        Attributes_.push_back(attribute);
        return *this;
    }
};

struct TGetOperationOptions
{
    using TSelf = TGetOperationOptions;

    FLUENT_FIELD_OPTION(TOperationAttributeFilter, AttributeFilter);
 
    FLUENT_FIELD_OPTION(bool, IncludeRuntime);
};

enum class EOperationBriefState : int
{
    InProgress    /* "in_progress" */,
    Completed     /* "completed" */,
    Aborted       /* "aborted" */,

    Failed        /* "failed" */,
};
 

enum class EOperationState : int
{
    None                /* "none" */,
    Starting            /* "starting" */,
    Orphaned            /* "orphaned" */,
    WaitingForAgent     /* "waiting_for_agent" */,
    Initializing        /* "initializing" */,
    Preparing           /* "preparing" */,
    Materializing       /* "orphaned" */,
    ReviveInitializing  /* "revive_initializing" */,
    Reviving            /* "reviving" */,
    RevivingJobs        /* "reviving_jobs" */,
    Pending             /* "pending" */,
    Running             /* "running" */,
    Completing          /* "completing" */,
    Completed           /* "completed" */,
    Aborting            /* "aborting" */,
    Aborted             /* "aborted" */,
    Failing             /* "failing" */,
    Failed              /* "failed" */,
};

enum class EOperationType : int
{
    Map         /* "map" */,
    Merge       /* "merge" */,
    Erase       /* "erase" */,
    Sort        /* "sort" */,
    Reduce      /* "reduce" */,
    MapReduce   /* "map_reduce" */,
    RemoteCopy  /* "remote_copy" */,
    JoinReduce  /* "join_reduce" */,
    Vanilla     /* "vanilla" */,
};

struct TOperationProgress
{
    TJobStatistics JobStatistics;

    TJobCounters JobCounters;

    TMaybe<TInstant> BuildTime;
};

struct TOperationBriefProgress
{
    i64 Aborted = 0;
    i64 Completed = 0;
    i64 Failed = 0;
    i64 Lost = 0;
    i64 Pending = 0;
    i64 Running = 0;
    i64 Total = 0;
};

struct TOperationResult
{
    TMaybe<TYtError> Error;
};

struct TOperationEvent
{
    TString State;

    TInstant Time;
};

struct TOperationAttributes
{
    TMaybe<TOperationId> Id;

    TMaybe<EOperationType> Type;

    TMaybe<TString> State;

    TMaybe<EOperationBriefState> BriefState;

    TMaybe<TString> AuthenticatedUser;

    TMaybe<TInstant> StartTime;

    TMaybe<TInstant> FinishTime;

    TMaybe<TOperationBriefProgress> BriefProgress;

    TMaybe<TNode> BriefSpec;

    TMaybe<TNode> Spec;

    TMaybe<TNode> FullSpec;

    TMaybe<TNode> UnrecognizedSpec;

    TMaybe<bool> Suspended;

    TMaybe<TOperationResult> Result;

    TMaybe<TOperationProgress> Progress;

    TMaybe<TVector<TOperationEvent>> Events;

    TMaybe<THashMap<TString, TYtError>> Alerts;
};

enum class ECursorDirection
{
    Past /* "past" */,
    Future /* "future" */,
};

struct TListOperationsOptions
{
    using TSelf = TListOperationsOptions;


    FLUENT_FIELD_OPTION(TInstant, FromTime);

    FLUENT_FIELD_OPTION(TInstant, ToTime);

    FLUENT_FIELD_OPTION(TInstant, CursorTime);

    FLUENT_FIELD_OPTION(ECursorDirection, CursorDirection);



    FLUENT_FIELD_OPTION(TString, Filter);

    FLUENT_FIELD_OPTION(TString, Pool);

    FLUENT_FIELD_OPTION(TString, User);

    FLUENT_FIELD_OPTION(EOperationState, State);

    FLUENT_FIELD_OPTION(EOperationType, Type);

    FLUENT_FIELD_OPTION(bool, WithFailedJobs);


    FLUENT_FIELD_OPTION(bool, IncludeArchive);

    FLUENT_FIELD_OPTION(bool, IncludeCounters);

    FLUENT_FIELD_OPTION(i64, Limit);
};

struct TListOperationsResult
{
    TVector<TOperationAttributes> Operations;


    TMaybe<THashMap<TString, i64>> PoolCounts;

    TMaybe<THashMap<TString, i64>> UserCounts;

    TMaybe<THashMap<TString, i64>> StateCounts;

    TMaybe<THashMap<EOperationType, i64>> TypeCounts;

    TMaybe<i64> WithFailedJobsCount;


    bool Incomplete;
};


enum class EListJobsDataSource : int
{
    Runtime  /* "runtime" */,
    Archive  /* "archive" */,
    Auto     /* "auto" */,
    Manual   /* "manual" */,
};

enum class EJobType : int
{
    Map               /* "map" */,
    PartitionMap      /* "partition_map" */,
    SortedMerge       /* "sorted_merge" */,
    OrderedMerge      /* "ordered_merge" */,
    UnorderedMerge    /* "unordered_merge" */,
    Partition         /* "partition" */,
    SimpleSort        /* "simple_sort" */,
    FinalSort         /* "final_sort" */,
    SortedReduce      /* "sorted_reduce" */,
    PartitionReduce   /* "partition_reduce" */,
    ReduceCombiner    /* "reduce_combiner" */,
    RemoteCopy        /* "remote_copy" */,
    IntermediateSort  /* "intermediate_sort" */,
    OrderedMap        /* "ordered_map" */,
    JoinReduce        /* "join_reduce" */,
    Vanilla           /* "vanilla" */,
    SchedulerUnknown  /* "scheduler_unknown" */,
    ReplicateChunk    /* "replicate_chunk" */,
    RemoveChunk       /* "remove_chunk" */,
    RepairChunk       /* "repair_chunk" */,
    SealChunk         /* "seal_chunk" */,
    ShallowMerge      /* "shallow_merge" */,
    MergeChunks       /* "merge_chunks" */,
    AutotomizeChunk   /* "autotomize_chunk" */,
    ReincarnateChunk  /* "reincarnate_chunk" */,
};

enum class ETaskName : int
{
    Map               /* "map" */,
    PartitionMap0     /* "partition_map(0)" */,
    SortedMerge       /* "sorted_merge" */,
    OrderedMerge      /* "ordered_merge" */,
    UnorderedMerge    /* "unordered_merge" */,
    Partition0        /* "partition(0)" */,
    Partition1        /* "partition(1)" */,
    Partition2        /* "partition(2)" */,
    SimpleSort        /* "simple_sort" */,
    FinalSort         /* "final_sort" */,
    SortedReduce      /* "sorted_reduce" */,
    PartitionReduce   /* "partition_reduce" */,
    ReduceCombiner    /* "reduce_combiner" */,
    RemoteCopy        /* "remote_copy" */,
    IntermediateSort  /* "intermediate_sort" */,
    OrderedMap        /* "ordered_map" */,
    JoinReduce        /* "join_reduce" */,
};

class TTaskName
{
public:
 
    // Constructors are implicit by design.

    TTaskName(TString taskName);

    TTaskName(const char* taskName);

    TTaskName(ETaskName taskName);
 
    const TString& Get() const;
 
private:
    TString TaskName_;
};

enum class EJobState : int
{
    None       /* "none" */,
    Waiting    /* "waiting" */,
    Running    /* "running" */,
    Aborting   /* "aborting" */,
    Completed  /* "completed" */,
    Failed     /* "failed" */,
    Aborted    /* "aborted" */,
    Lost       /* "lost" */,
};

enum class EJobSortField : int
{
    Type       /* "type" */,
    State      /* "state" */,
    StartTime  /* "start_time" */,
    FinishTime /* "finish_time" */,
    Address    /* "address" */,
    Duration   /* "duration" */,
    Progress   /* "progress" */,
    Id         /* "id" */,
    TaskName   /* "task_name" */,
};

enum class EJobSortDirection : int
{
    Ascending /* "ascending" */,
    Descending /* "descending" */,
};

enum class EJobAttribute : int
{
    Id                /* "id" */,
    Type              /* "type" */,
    State             /* "state" */,
    Address           /* "address" */,
    TaskName          /* "task_name" */,
    StartTime         /* "start_time" */,
    FinishTime        /* "finish_time" */,
    Progress          /* "progress" */,
    StderrSize        /* "stderr_size" */,
    Error             /* "error" */,
    Result            /* "result" */,
    BriefStatistics   /* "brief_statistics" */,
    InputPaths        /* "input_paths" */,
    CoreInfos         /* "core_infos" */,
};

struct TJobAttributeFilter
{
    using TSelf = TJobAttributeFilter;
 
    THashSet<EJobAttribute> Attributes_;

    TSelf& Add(EJobAttribute attribute)
    {
        Attributes_.insert(attribute);
        return *this;
    }
};

struct TListJobsOptions
{
    using TSelf = TListJobsOptions;


    FLUENT_FIELD_OPTION(EJobType, Type);

    FLUENT_FIELD_OPTION(EJobState, State);

    FLUENT_FIELD_OPTION(TString, Address);

    FLUENT_FIELD_OPTION(bool, WithStderr);

    FLUENT_FIELD_OPTION(bool, WithSpec);

    FLUENT_FIELD_OPTION(bool, WithFailContext);

    FLUENT_FIELD_OPTION(bool, WithMonitoringDescriptor);

    FLUENT_FIELD_OPTION(bool, WithInterruptionInfo);

    FLUENT_FIELD_OPTION(TString, OperationIncarnation);

    FLUENT_FIELD_OPTION(TString, MonitoringDescriptor);

    FLUENT_FIELD_OPTION(TInstant, FromTime);

    FLUENT_FIELD_OPTION(TInstant, ToTime);

    FLUENT_FIELD_OPTION(TString, ContinuationToken);

    FLUENT_FIELD_OPTION(TJobAttributeFilter, AttributeFilter);



    FLUENT_FIELD_OPTION(EJobSortField, SortField);

    FLUENT_FIELD_OPTION(ESortOrder, SortOrder);


    FLUENT_FIELD_OPTION(EListJobsDataSource, DataSource);

    FLUENT_FIELD_OPTION(bool, IncludeCypress);

    FLUENT_FIELD_OPTION(bool, IncludeControllerAgent);

    FLUENT_FIELD_OPTION(bool, IncludeArchive);

    FLUENT_FIELD_OPTION(i64, Limit);

    FLUENT_FIELD_OPTION(i64, Offset);
};

struct TCoreInfo
{
    i64 ProcessId;
    TString ExecutableName;
    TMaybe<ui64> Size;
    TMaybe<TYtError> Error;
};

struct TJobAttributes
{
    TMaybe<TJobId> Id;

    TMaybe<EJobType> Type;

    TMaybe<EJobState> State;

    TMaybe<TString> Address;

    TMaybe<TString> TaskName;

    TMaybe<TInstant> StartTime;

    TMaybe<TInstant> FinishTime;

    TMaybe<double> Progress;

    TMaybe<i64> StderrSize;

    TMaybe<TYtError> Error;

    TMaybe<TNode> BriefStatistics;

    TMaybe<TVector<TRichYPath>> InputPaths;

    TMaybe<TVector<TCoreInfo>> CoreInfos;

    TMaybe<TNode> ExecAttributes;

    TMaybe<ui64> Cookie;
};

struct TListJobsResult
{
    TVector<TJobAttributes> Jobs;

    TMaybe<i64> CypressJobCount;

    TMaybe<i64> ControllerAgentJobCount;

    TMaybe<i64> ArchiveJobCount;
};


struct TGetJobOptions
{
    using TSelf = TGetJobOptions;

    FLUENT_FIELD_OPTION(TJobAttributeFilter, AttributeFilter);
};

struct TGetJobInputOptions
{
    using TSelf = TGetJobInputOptions;
};

struct TGetJobFailContextOptions
{
    using TSelf = TGetJobFailContextOptions;
};

struct TGetJobStderrOptions
{
    using TSelf = TGetJobStderrOptions;
};


struct TGetFailedJobInfoOptions
{
    using TSelf = TGetFailedJobInfoOptions;

    FLUENT_FIELD_DEFAULT(ui64, MaxJobCount, 10);

    FLUENT_FIELD_DEFAULT(ui64, StderrTailSize, 64 * 1024);
};


struct TGetJobTraceOptions
{
    using TSelf = TGetJobTraceOptions;

    FLUENT_FIELD_OPTION(TJobId, JobId);

    FLUENT_FIELD_OPTION(TJobTraceId, TraceId);

    FLUENT_FIELD_OPTION(TInstant, FromTime);

    FLUENT_FIELD_OPTION(TInstant, ToTime);
};

struct TJobTraceEvent
{
    TOperationId OperationId;

    TJobId JobId;

    TJobTraceId TraceId;

    i64 EventIndex = 0;

    std::string Event;

    TInstant EventTime;
};


struct IOperation
    : public TThrRefBase
{
    ~IOperation() override = default;

    virtual const TOperationId& GetId() const = 0;

    virtual TString GetWebInterfaceUrl() const = 0;

    virtual TString GetStatus() const = 0;

    virtual ::NThreading::TFuture<void> GetPreparedFuture() = 0;

    virtual void Start() = 0;

    virtual bool IsStarted() const = 0;

    virtual ::NThreading::TFuture<void> GetStartedFuture() = 0;

    virtual ::NThreading::TFuture<void> Watch() = 0;

    virtual TVector<TFailedJobInfo> GetFailedJobInfo(const TGetFailedJobInfoOptions& options = TGetFailedJobInfoOptions()) = 0;

    virtual EOperationBriefState GetBriefState() = 0;

    virtual TMaybe<TYtError> GetError() = 0;

    virtual TJobStatistics GetJobStatistics() = 0;

    virtual TMaybe<TOperationBriefProgress> GetBriefProgress() = 0;

    virtual TMaybe<THashMap<TString, TYtError>> GetAlerts() = 0;

    virtual void AbortOperation() = 0;

    virtual void CompleteOperation() = 0;

    virtual void SuspendOperation(
        const TSuspendOperationOptions& options = TSuspendOperationOptions()) = 0;

    virtual void ResumeOperation(
        const TResumeOperationOptions& options = TResumeOperationOptions()) = 0;

    virtual TOperationAttributes GetAttributes(
        const TGetOperationOptions& options = TGetOperationOptions()) = 0;

    virtual void UpdateParameters(
        const TUpdateOperationParametersOptions& options = TUpdateOperationParametersOptions()) = 0;

    virtual TJobAttributes GetJob(
        const TJobId& jobId,
        const TGetJobOptions& options = TGetJobOptions()) = 0;

    virtual TListJobsResult ListJobs(
        const TListJobsOptions& options = TListJobsOptions()) = 0;
};

struct IOperationClient
{
    IOperationPtr Map(
        const TMapOperationSpec& spec,
        ::TIntrusivePtr<IMapperBase> mapper,
        const TOperationOptions& options = TOperationOptions());

    IOperationPtr Map(
        ::TIntrusivePtr<IMapperBase> mapper,
        const TOneOrMany<TStructuredTablePath>& input,
        const TOneOrMany<TStructuredTablePath>& output,
        const TMapOperationSpec& spec = TMapOperationSpec(),
        const TOperationOptions& options = TOperationOptions());

    virtual IOperationPtr RawMap(
        const TRawMapOperationSpec& spec,
        ::TIntrusivePtr<IRawJob> rawJob,
        const TOperationOptions& options = TOperationOptions()) = 0;

    IOperationPtr Reduce(
        const TReduceOperationSpec& spec,
        ::TIntrusivePtr<IReducerBase> reducer,
        const TOperationOptions& options = TOperationOptions());

    IOperationPtr Reduce(
        ::TIntrusivePtr<IReducerBase> reducer,
        const TOneOrMany<TStructuredTablePath>& input,
        const TOneOrMany<TStructuredTablePath>& output,
        const TSortColumns& reduceBy,
        const TReduceOperationSpec& spec = TReduceOperationSpec(),
        const TOperationOptions& options = TOperationOptions());

    virtual IOperationPtr RawReduce(
        const TRawReduceOperationSpec& spec,
        ::TIntrusivePtr<IRawJob> rawJob,
        const TOperationOptions& options = TOperationOptions()) = 0;

    IOperationPtr JoinReduce(
        const TJoinReduceOperationSpec& spec,
        ::TIntrusivePtr<IReducerBase> reducer,
        const TOperationOptions& options = TOperationOptions());

    virtual IOperationPtr RawJoinReduce(
        const TRawJoinReduceOperationSpec& spec,
        ::TIntrusivePtr<IRawJob> rawJob,
        const TOperationOptions& options = TOperationOptions()) = 0;

    IOperationPtr MapReduce(
        const TMapReduceOperationSpec& spec,
        ::TIntrusivePtr<IMapperBase> mapper,
        ::TIntrusivePtr<IReducerBase> reducer,
        const TOperationOptions& options = TOperationOptions());

    IOperationPtr MapReduce(
        const TMapReduceOperationSpec& spec,
        ::TIntrusivePtr<IMapperBase> mapper,
        ::TIntrusivePtr<IReducerBase> reduceCombiner,
        ::TIntrusivePtr<IReducerBase> reducer,
        const TOperationOptions& options = TOperationOptions());

    IOperationPtr MapReduce(
        ::TIntrusivePtr<IMapperBase> mapper,
        ::TIntrusivePtr<IReducerBase> reducer,
        const TOneOrMany<TStructuredTablePath>& input,
        const TOneOrMany<TStructuredTablePath>& output,
        const TSortColumns& reduceBy,
        TMapReduceOperationSpec spec = TMapReduceOperationSpec(),
        const TOperationOptions& options = TOperationOptions());

    IOperationPtr MapReduce(
        ::TIntrusivePtr<IMapperBase> mapper,
        ::TIntrusivePtr<IReducerBase> reduceCombiner,
        ::TIntrusivePtr<IReducerBase> reducer,
        const TOneOrMany<TStructuredTablePath>& input,
        const TOneOrMany<TStructuredTablePath>& output,
        const TSortColumns& reduceBy,
        TMapReduceOperationSpec spec = TMapReduceOperationSpec(),
        const TOperationOptions& options = TOperationOptions());

    virtual IOperationPtr RawMapReduce(
        const TRawMapReduceOperationSpec& spec,
        ::TIntrusivePtr<IRawJob> mapper,
        ::TIntrusivePtr<IRawJob> reduceCombiner,
        ::TIntrusivePtr<IRawJob> reducer,
        const TOperationOptions& options = TOperationOptions()) = 0;

    virtual IOperationPtr Sort(
        const TSortOperationSpec& spec,
        const TOperationOptions& options = TOperationOptions()) = 0;

    IOperationPtr Sort(
        const TOneOrMany<TRichYPath>& input,
        const TRichYPath& output,
        const TSortColumns& sortBy,
        const TSortOperationSpec& spec = TSortOperationSpec(),
        const TOperationOptions& options = TOperationOptions());

    virtual IOperationPtr Merge(
        const TMergeOperationSpec& spec,
        const TOperationOptions& options = TOperationOptions()) = 0;

    virtual IOperationPtr Erase(
        const TEraseOperationSpec& spec,
        const TOperationOptions& options = TOperationOptions()) = 0;

    virtual IOperationPtr RemoteCopy(
        const TRemoteCopyOperationSpec& spec,
        const TOperationOptions& options = TOperationOptions()) = 0;

    virtual IOperationPtr RunVanilla(
        const TVanillaOperationSpec& spec,
        const TOperationOptions& options = TOperationOptions()) = 0;

    virtual void AbortOperation(
        const TOperationId& operationId) = 0;

    virtual void CompleteOperation(
        const TOperationId& operationId) = 0;

    virtual void WaitForOperation(
        const TOperationId& operationId) = 0;

    virtual EOperationBriefState CheckOperation(
        const TOperationId& operationId) = 0;

    virtual IOperationPtr AttachOperation(const TOperationId& operationId) = 0;
 
private:
    virtual IOperationPtr DoMap(
        const TMapOperationSpec& spec,
        ::TIntrusivePtr<IStructuredJob> mapper,
        const TOperationOptions& options) = 0;
 
    virtual IOperationPtr DoReduce(
        const TReduceOperationSpec& spec,
        ::TIntrusivePtr<IStructuredJob> reducer,
        const TOperationOptions& options) = 0;
 
    virtual IOperationPtr DoJoinReduce(
        const TJoinReduceOperationSpec& spec,
        ::TIntrusivePtr<IStructuredJob> reducer,
        const TOperationOptions& options) = 0;
 
    virtual IOperationPtr DoMapReduce(
        const TMapReduceOperationSpec& spec,
        ::TIntrusivePtr<IStructuredJob> mapper,
        ::TIntrusivePtr<IStructuredJob> reduceCombiner,
        ::TIntrusivePtr<IStructuredJob> reducer,
        const TOperationOptions& options) = 0;
};

 
} // namespace NYT
 
#define OPERATION_INL_H_
#include "operation-inl.h"
#undef OPERATION_INL_H_