Mark mutable_id for YT writes (ydb-platform#1744)

* init * fix * canonize * canonize
shnikd · Feb 10, 2024 · 590c21c · 590c21c
1 parent b1f681e
commit 590c21c
Show file tree

Hide file tree

Showing 69 changed files with 3,121 additions and 3,076 deletions.
diff --git a/ydb/library/yql/providers/yt/provider/yql_yt_datasink.cpp b/ydb/library/yql/providers/yt/provider/yql_yt_datasink.cpp
@@ -243,6 +243,11 @@ class TYtDataSink : public TDataProviderBase {
                     ctx);
                 res = ctx.ChangeChild(*res, TYtWriteTable::idx_Settings, std::move(settings));
             }
+            auto mutationId = ++NextMutationId_;
+            res = ctx.ChangeChild(*res, TYtWriteTable::idx_Settings, 
+                NYql::AddSetting(*res->Child(TYtWriteTable::idx_Settings),
+                    EYtSettingType::MutationId,
+                    ctx.NewAtom(res->Child(TYtWriteTable::idx_Settings)->Pos(), ToString(mutationId)), ctx));
             if (State_->Configuration->UseSystemColumns.Get().GetOrElse(DEFAULT_USE_SYS_COLUMNS)) {
                 res = ctx.ChangeChild(*res, TYtWriteTable::idx_Content,
                     ctx.Builder(node->Pos())
@@ -268,6 +273,7 @@ class TYtDataSink : public TDataProviderBase {
     void Reset() final {
         TDataProviderBase::Reset();
         State_->Reset();
+        NextMutationId_ = 0;
     }
 
     bool CanExecute(const TExprNode& node) override {
@@ -567,6 +573,7 @@ class TYtDataSink : public TDataProviderBase {
     }
 
 private:
+    ui32 NextMutationId_ = 0;
     TYtState::TPtr State_;
     TLazyInitHolder<IGraphTransformer> IntentDeterminationTransformer_;
     TLazyInitHolder<TVisitorTransformerBase> TypeAnnotationTransformer_;

diff --git a/ydb/library/yql/providers/yt/provider/yql_yt_datasink_type_ann.cpp b/ydb/library/yql/providers/yt/provider/yql_yt_datasink_type_ann.cpp
@@ -1435,6 +1435,7 @@ class TYtDataSinkTypeAnnotationTransformer : public TVisitorTransformerBase {
             | EYtSettingType::PrimaryMedium
             | EYtSettingType::Expiration
             | EYtSettingType::MonotonicKeys
+            | EYtSettingType::MutationId
             , ctx))
         {
             return TStatus::Error;
@@ -1712,6 +1713,7 @@ class TYtDataSinkTypeAnnotationTransformer : public TVisitorTransformerBase {
             | EYtSettingType::PrimaryMedium
             | EYtSettingType::Expiration
             | EYtSettingType::MonotonicKeys
+            | EYtSettingType::MutationId
             , ctx))
         {
             return TStatus::Error;

diff --git a/ydb/library/yql/providers/yt/provider/yql_yt_op_settings.cpp b/ydb/library/yql/providers/yt/provider/yql_yt_op_settings.cpp
@@ -795,6 +795,20 @@ bool ValidateSettings(const TExprNode& settingsNode, EYtSettingTypes accepted, T
                 return false;
             }
             break;
+        case EYtSettingType::MutationId:
+            if (!EnsureTupleSize(*setting, 2, ctx)) {
+                return false;
+            }
+            if (!EnsureAtom(*setting->Child(1), ctx)) {
+                return false;
+            }
+            ui32 mutationId;
+            if (!TryFromString(setting->Child(1)->Content(), mutationId)) {
+                ctx.AddError(TIssue(ctx.GetPosition(setting->Child(1)->Pos()), TStringBuilder()
+                    << "Expected a number, but got: " << TString{setting->Child(1)->Content()}.Quote()));
+                return false;
+            }
+            break;
         }
     }
 

diff --git a/ydb/library/yql/providers/yt/provider/yql_yt_op_settings.h b/ydb/library/yql/providers/yt/provider/yql_yt_op_settings.h
@@ -112,6 +112,7 @@ enum class EYtSettingType: ui64 {
     PrimaryMedium               = 1ull << 59 /* "primary_medium", "primarymedium" */,
     KeepMeta                    = 1ull << 60 /* "keep_meta", "keepmeta" */,
     MonotonicKeys               = 1ull << 61 /* "monotonic_keys", "monotonickeys" */,
+    MutationId                  = 1ull << 62 /* "mutationid", "mutation_id" */,
 };
 
 Y_DECLARE_FLAGS(EYtSettingTypes, EYtSettingType);

diff --git a/ydb/library/yql/sql/v1/sql_query.cpp b/ydb/library/yql/sql/v1/sql_query.cpp
@@ -299,6 +299,11 @@ bool TSqlQuery::Statement(TVector<TNodePtr>& blocks, const TRule_sql_stmt_core&
             break;
         }
         case TRule_sql_stmt_core::kAltSqlStmtCore8: {
+            if (Ctx.ParallelModeCount > 0) {
+                Error() << humanStatementName << " statement is not supported in parallel mode";
+                return false;
+            }
+
             Ctx.BodyPart();
             const auto& rule = core.GetAlt_sql_stmt_core8().GetRule_commit_stmt1();
             Token(rule.GetToken1());
@@ -324,6 +329,11 @@ bool TSqlQuery::Statement(TVector<TNodePtr>& blocks, const TRule_sql_stmt_core&
             break;
         }
         case TRule_sql_stmt_core::kAltSqlStmtCore11: {
+            if (Ctx.ParallelModeCount > 0) {
+                Error() << humanStatementName << " statement is not supported in parallel mode";
+                return false;
+            }
+
             Ctx.BodyPart();
             const auto& rule = core.GetAlt_sql_stmt_core11().GetRule_rollback_stmt1();
             Token(rule.GetToken1());

diff --git a/ydb/library/yql/tests/s-expressions/yt_native_file/part0/canondata/result.json b/ydb/library/yql/tests/s-expressions/yt_native_file/part0/canondata/result.json