YT-20179: Implement PatchSpec

* Changelog entry
Type: feature
Component: map-reduce

New PatchSpec API call can be used to tweak spec at runtime.
Now it only supports changing max_failed_job_count, but more spec options are on the way!
commit_hash:21ce2961e5d685a66d4139b8f4bcf0e68f5c49cc

Author: coteeq

yt/yt/client/formats/format.cpp

@@ -52,6 +52,11 @@ const IAttributeDictionary& TFormat::Attributes() const
     return *Attributes_;
 }
 
+bool TFormat::operator==(const TFormat& other) const
+{
+    return GetType() == other.GetType() && Attributes() == other.Attributes();
+}
+
 void Serialize(const TFormat& value, IYsonConsumer* consumer)
 {
     BuildYsonFluently(consumer)

yt/yt/client/formats/format.h

@@ -30,6 +30,8 @@ class TFormat
 
     const NYTree::IAttributeDictionary& Attributes() const;
 
+    bool operator==(const TFormat& other) const;
+
 private:
     NYTree::IAttributeDictionaryPtr Attributes_;
 

yt/yt/client/misc/workload.h

@@ -49,6 +49,8 @@ struct TWorkloadDescriptor
     //! Computes the aggregated priority.
     //! Larger is better.
     i64 GetPriority() const;
+
+    bool operator==(const TWorkloadDescriptor& other) const = default;
 };
 
 i64 GetBasicPriority(EWorkloadCategory category);

yt/yt/client/scheduler/public.h

@@ -78,6 +78,7 @@ YT_DEFINE_ERROR_ENUM(
     ((CannotUseBothAclAndAco)                 (221))
     ((GangOperationsAllowedOnlyInFifoPools)   (222))
     ((OperationLaunchedInNonexistentPool)     (223))
+    ((OperationHasNoController)               (224))
 );
 
 DEFINE_ENUM(EUnavailableChunkAction,

yt/yt/client/scheduler/spec_patch.cpp

@@ -28,6 +28,12 @@ void FromProto(TSpecPatchPtr patch, const NProto::TSpecPatch* protoPatch)
     patch->Value = ConvertToNode(TYsonString(protoPatch->value()));
 }
 
+void FromProto(TSpecPatchPtr* patch, const NProto::TSpecPatch& protoPatch)
+{
+    *patch = New<TSpecPatch>();
+    FromProto(*patch, &protoPatch);
+}
+
 void FormatValue(TStringBuilderBase* builder, const TSpecPatchPtr& patch, TStringBuf /*spec*/)
 {
     builder->AppendString(ConvertToYsonString(patch).ToString());

yt/yt/client/scheduler/spec_patch.h

@@ -26,6 +26,7 @@ DEFINE_REFCOUNTED_TYPE(TSpecPatch)
 
 void ToProto(NProto::TSpecPatch* protoPatch, const TSpecPatchPtr& patch);
 void FromProto(TSpecPatchPtr patch, const NProto::TSpecPatch* protoPatch);
+void FromProto(TSpecPatchPtr* patch, const NProto::TSpecPatch& protoPatch);
 
 void FormatValue(TStringBuilderBase* builder, const TSpecPatchPtr& patch, TStringBuf spec);
 

yt/yt/core/misc/config.h

@@ -23,6 +23,8 @@ struct TExponentialBackoffOptions
     TDuration MaxBackoff = DefaultMaxBackoff;
     double BackoffMultiplier = DefaultBackoffMultiplier;
     double BackoffJitter = DefaultBackoffJitter;
+
+    bool operator==(const TExponentialBackoffOptions& other) const = default;
 };
 
 ////////////////////////////////////////////////////////////////////////////////

yt/yt/core/ytree/yson_struct_update-inl.h

@@ -174,7 +174,9 @@ void Update(
     TIntrusivePtr<TStruct> newStruct)
 {
     const auto* meta = oldStruct->GetMeta();
+    YT_VERIFY(meta == newStruct->GetMeta());
     const auto& parameterToFieldRegistrar = registrar.RegisteredFields_->ParameterToFieldRegistrar;
+    YT_VERIFY(registrar.RegisteredFields_->Meta == meta);
     for (const auto& [name, parameter] : meta->GetParameterMap()) {
         if (parameter->CompareParameter(oldStruct.Get(), newStruct.Get())) {
             continue;
@@ -183,8 +185,9 @@ void Update(
         auto fieldDescIter = parameterToFieldRegistrar.find(parameter);
         if (fieldDescIter == parameterToFieldRegistrar.end()) {
             THROW_ERROR_EXCEPTION("Field %Qv is not marked as updatable, but was changed", name);
+        } else {
+            fieldDescIter->second->DoUpdate(parameter, oldStruct.Get(), newStruct.Get());
         }
-        fieldDescIter->second->DoUpdate(parameter, oldStruct.Get(), newStruct.Get());
     }
 }