transaction_commands.cpp 7.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244
  1. #include "config.h"
  2. #include "transaction_commands.h"
  3. #include <yt/yt/client/transaction_client/timestamp_provider.h>
  4. #include <yt/yt/client/api/transaction.h>
  5. #include <yt/yt/core/concurrency/scheduler.h>
  6. #include <yt/yt/core/ytree/helpers.h>
  7. #include <yt/yt/core/ytree/fluent.h>
  8. namespace NYT::NDriver {
  9. using namespace NYTree;
  10. using namespace NTransactionClient;
  11. using namespace NCypressClient;
  12. using namespace NObjectClient;
  13. using namespace NConcurrency;
  14. using namespace NYson;
  15. ////////////////////////////////////////////////////////////////////////////////
  16. void TStartTransactionCommand::Register(TRegistrar registrar)
  17. {
  18. registrar.Parameter("type", &TThis::Type)
  19. .Default(NTransactionClient::ETransactionType::Master);
  20. registrar.Parameter("attributes", &TThis::Attributes)
  21. .Default(nullptr);
  22. registrar.ParameterWithUniversalAccessor<bool>(
  23. "sticky",
  24. [] (TThis* command) -> auto& {
  25. return command->Options.Sticky;
  26. })
  27. .Optional(/*init*/ false);
  28. registrar.ParameterWithUniversalAccessor<std::optional<TDuration>>(
  29. "timeout",
  30. [] (TThis* command) -> auto& {
  31. return command->Options.Timeout;
  32. })
  33. .Optional(/*init*/ false);
  34. registrar.ParameterWithUniversalAccessor<TTransactionId>(
  35. "transaction_id_override",
  36. [] (TThis* command) -> auto& {
  37. return command->Options.Id;
  38. })
  39. .Optional(/*init*/ false);
  40. registrar.ParameterWithUniversalAccessor<TTimestamp>(
  41. "start_timestamp_override",
  42. [] (TThis* command) -> auto& {
  43. return command->Options.StartTimestamp;
  44. })
  45. .Optional(/*init*/ false);
  46. registrar.ParameterWithUniversalAccessor<TTransactionId>(
  47. "transaction_id",
  48. [] (TThis* command) -> auto& {
  49. return command->Options.ParentId;
  50. })
  51. .Optional(/*init*/ false);
  52. registrar.ParameterWithUniversalAccessor<bool>(
  53. "ping_ancestor_transactions",
  54. [] (TThis* command) -> auto& {
  55. return command->Options.PingAncestors;
  56. })
  57. .Optional(/*init*/ false);
  58. registrar.ParameterWithUniversalAccessor<std::vector<TTransactionId>>(
  59. "prerequisite_transaction_ids",
  60. [] (TThis* command) -> auto& {
  61. return command->Options.PrerequisiteTransactionIds;
  62. })
  63. .Optional(/*init*/ false);
  64. registrar.ParameterWithUniversalAccessor<std::optional<TInstant>>(
  65. "deadline",
  66. [] (TThis* command) -> auto& {
  67. return command->Options.Deadline;
  68. })
  69. .Optional(/*init*/ false);
  70. registrar.ParameterWithUniversalAccessor<EAtomicity>(
  71. "atomicity",
  72. [] (TThis* command) -> auto& {
  73. return command->Options.Atomicity;
  74. })
  75. .Optional(/*init*/ false);
  76. registrar.ParameterWithUniversalAccessor<EDurability>(
  77. "durability",
  78. [] (TThis* command) -> auto& {
  79. return command->Options.Durability;
  80. })
  81. .Optional(/*init*/ false);
  82. registrar.ParameterWithUniversalAccessor<bool>(
  83. "suppress_start_timestamp_generation",
  84. [] (TThis* command) -> auto& {
  85. return command->Options.SuppressStartTimestampGeneration;
  86. })
  87. .Optional(/*init*/ false);
  88. registrar.ParameterWithUniversalAccessor<TCellTag>(
  89. "coordinator_master_cell_tag",
  90. [] (TThis* command) -> auto& {
  91. return command->Options.CoordinatorMasterCellTag;
  92. })
  93. .Optional(/*init*/ false);
  94. registrar.ParameterWithUniversalAccessor<std::optional<TCellTagList>>(
  95. "replicate_to_master_cell_tags",
  96. [] (TThis* command) -> auto& {
  97. return command->Options.ReplicateToMasterCellTags;
  98. })
  99. .Optional(/*init*/ false);
  100. registrar.ParameterWithUniversalAccessor<bool>(
  101. "start_cypress_transaction",
  102. [] (TThis* command) -> auto& {
  103. return command->Options.StartCypressTransaction;
  104. })
  105. .Optional(/*init*/ false);
  106. }
  107. void TStartTransactionCommand::DoExecute(ICommandContextPtr context)
  108. {
  109. Options.Ping = true;
  110. Options.AutoAbort = false;
  111. if (Attributes) {
  112. Options.Attributes = ConvertToAttributes(Attributes);
  113. }
  114. if (Type != ETransactionType::Master) {
  115. Options.Sticky = true;
  116. }
  117. auto transaction = WaitFor(context->GetClient()->StartTransaction(Type, Options))
  118. .ValueOrThrow();
  119. if (Options.Sticky) {
  120. context->GetDriver()->GetStickyTransactionPool()->RegisterTransaction(transaction);
  121. } else {
  122. transaction->Detach();
  123. }
  124. ProduceSingleOutputValue(context, "transaction_id", transaction->GetId());
  125. }
  126. ////////////////////////////////////////////////////////////////////////////////
  127. void TPingTransactionCommand::DoExecute(ICommandContextPtr context)
  128. {
  129. // Specially for evvers@ :)
  130. if (!Options.TransactionId) {
  131. return;
  132. }
  133. auto transaction = AttachTransaction(context, true);
  134. WaitFor(transaction->Ping())
  135. .ThrowOnError();
  136. ProduceEmptyOutput(context);
  137. }
  138. ////////////////////////////////////////////////////////////////////////////////
  139. void TCommitTransactionCommand::DoExecute(ICommandContextPtr context)
  140. {
  141. auto transaction = AttachTransaction(context, true);
  142. auto transactionCommitResult = WaitFor(transaction->Commit(Options))
  143. .ValueOrThrow();
  144. if (context->GetConfig()->ApiVersion >= ApiVersion4) {
  145. ProduceOutput(
  146. context,
  147. [&] (IYsonConsumer* consumer) {
  148. BuildYsonFluently(consumer)
  149. .BeginMap()
  150. .Item("primary_commit_timestamp").Value(transactionCommitResult.PrimaryCommitTimestamp)
  151. .Item("commit_timestamps").DoMapFor(
  152. transactionCommitResult.CommitTimestamps.Timestamps,
  153. [&] (auto fluent, const auto& pair) {
  154. fluent.Item(ToString(pair.first)).Value(pair.second);
  155. })
  156. .EndMap();
  157. });
  158. } else {
  159. ProduceEmptyOutput(context);
  160. }
  161. }
  162. ////////////////////////////////////////////////////////////////////////////////
  163. void TAbortTransactionCommand::Register(TRegistrar registrar)
  164. {
  165. registrar.ParameterWithUniversalAccessor<bool>(
  166. "force",
  167. [] (TThis* command) -> auto& {
  168. return command->Options.Force;
  169. })
  170. .Optional(/*init*/ false);
  171. }
  172. void TAbortTransactionCommand::DoExecute(ICommandContextPtr context)
  173. {
  174. auto transaction = AttachTransaction(context, true);
  175. WaitFor(transaction->Abort(Options))
  176. .ThrowOnError();
  177. ProduceEmptyOutput(context);
  178. }
  179. ////////////////////////////////////////////////////////////////////////////////
  180. void TGenerateTimestampCommand::Register(TRegistrar registrar)
  181. {
  182. registrar.ParameterWithUniversalAccessor<std::optional<TCellTag>>(
  183. "clock_cluster_tag",
  184. [] (TThis* command) -> auto& {
  185. return command->Options.ClockClusterTag;
  186. })
  187. .Optional(/*init*/ false);
  188. }
  189. void TGenerateTimestampCommand::DoExecute(ICommandContextPtr context)
  190. {
  191. auto timestampProvider = context->GetClient()->GetTimestampProvider();
  192. auto clockClusterTag = Options.ClockClusterTag.value_or(InvalidCellTag);
  193. auto timestamp = WaitFor(timestampProvider->GenerateTimestamps(1, clockClusterTag))
  194. .ValueOrThrow();
  195. ProduceSingleOutputValue(context, "timestamp", timestamp);
  196. }
  197. ////////////////////////////////////////////////////////////////////////////////
  198. } // namespace NYT::NDriver