Merge remote-tracking branch 'origin/YQ-RD-Add-YQL-conclusion-status' into YQ-3738-pass-UV-from-filter-to-read-actor

Author: GrigoriyPA

ydb/core/kqp/gateway/behaviour/tablestore/operations/upsert_opt.cpp

@@ -10,7 +10,12 @@ TConclusionStatus TUpsertOptionsOperation::DoDeserialize(NYql::TObjectSettingsIm
         return TConclusionStatus::Fail("Incorrect value for SCHEME_NEED_ACTUALIZATION: cannot parse as boolean");
     }
     SchemeNeedActualization = *value;
-    ExternalGuaranteeExclusivePK = features.Extract<bool>("EXTERNAL_GUARANTEE_EXCLUSIVE_PK");
+    ScanReaderPolicyName = features.Extract<TString>("SCAN_READER_POLICY_NAME");
+    if (ScanReaderPolicyName) {
+        if (*ScanReaderPolicyName != "PLAIN" && *ScanReaderPolicyName != "SIMPLE") {
+            return TConclusionStatus::Fail("SCAN_READER_POLICY_NAME have to be in ['PLAIN', 'SIMPLE']");
+        }
+    }
     if (const auto className = features.Extract<TString>("COMPACTION_PLANNER.CLASS_NAME")) {
         if (!CompactionPlannerConstructor.Initialize(*className)) {
             return TConclusionStatus::Fail("incorrect class name for compaction planner:" + *className);
@@ -52,8 +57,8 @@ TConclusionStatus TUpsertOptionsOperation::DoDeserialize(NYql::TObjectSettingsIm
 
 void TUpsertOptionsOperation::DoSerializeScheme(NKikimrSchemeOp::TAlterColumnTableSchema& schemaData) const {
     schemaData.MutableOptions()->SetSchemeNeedActualization(SchemeNeedActualization);
-    if (ExternalGuaranteeExclusivePK) {
-        schemaData.MutableOptions()->SetExternalGuaranteeExclusivePK(*ExternalGuaranteeExclusivePK);
+    if (ScanReaderPolicyName) {
+        schemaData.MutableOptions()->SetScanReaderPolicyName(*ScanReaderPolicyName);
     }
     if (CompactionPlannerConstructor.HasObject()) {
         CompactionPlannerConstructor.SerializeToProto(*schemaData.MutableOptions()->MutableCompactionPlannerConstructor());

ydb/core/kqp/gateway/behaviour/tablestore/operations/upsert_opt.h

@@ -13,7 +13,7 @@ class TUpsertOptionsOperation: public ITableStoreOperation {
     static inline const auto Registrator = TFactory::TRegistrator<TUpsertOptionsOperation>(GetTypeName());
 private:
     bool SchemeNeedActualization = false;
-    std::optional<bool> ExternalGuaranteeExclusivePK;
+    std::optional<TString> ScanReaderPolicyName;
     NOlap::NStorageOptimizer::TOptimizerPlannerConstructorContainer CompactionPlannerConstructor;
     NOlap::NDataAccessorControl::TMetadataManagerConstructorContainer MetadataManagerConstructor;
 public:

ydb/core/kqp/provider/yql_kikimr_datasink.cpp

@@ -391,6 +391,8 @@ class TKiSinkIntentDeterminationTransformer: public TKiSinkVisitorTransformer {
                 return TStatus::Ok;
             case TKikimrKey::Type::BackupCollection:
                 return TStatus::Ok;
+            case TKikimrKey::Type::Sequence:
+                return TStatus::Ok;
         }
 
         ctx.AddError(TIssue(ctx.GetPosition(node.Pos()), "Invalid table key type."));
@@ -845,10 +847,21 @@ class TKikimrDataSink : public TDataProviderBase
             ? settings.ValueType.Cast()
             : Build<TCoAtom>(ctx, node->Pos()).Value("Null").Done();
 
+        TString sequence;
+
+        if (key.GetKeyType() == TKikimrKey::Type::Sequence) {
+            sequence = key.GetSequencePath();
+        } else if (key.GetKeyType() == TKikimrKey::Type::PGObject) {
+            sequence = key.GetPGObjectId();
+        } else {
+            YQL_ENSURE(false, "Invalid key type for sequence");
+        }
+        
+
         return Build<TKiAlterSequence>(ctx, node->Pos())
             .World(node->Child(0))
             .DataSink(node->Child(1))
-            .Sequence().Build(key.GetPGObjectId())
+            .Sequence().Build(sequence)
             .ValueType(valueType)
             .SequenceSettings(settings.SequenceSettings.Cast())
             .Settings(settings.Other)
@@ -1267,6 +1280,21 @@ class TKikimrDataSink : public TDataProviderBase
                 }
                 break;
             }
+            case TKikimrKey::Type::Sequence: {
+                NCommon::TWriteSequenceSettings settings = NCommon::ParseSequenceSettings(TExprList(node->Child(4)), ctx);
+
+                YQL_ENSURE(settings.Mode);
+                auto mode = settings.Mode.Cast();
+
+                if (mode == "alter" || mode == "alter_if_exists") {
+                    return MakeAlterSequence(node, settings, key, ctx);
+                } else {
+                    ctx.AddError(TIssue(ctx.GetPosition(node->Pos()), "Unknown operation type for sequence"));
+                    return nullptr;
+                }
+
+                break;
+            }
             case TKikimrKey::Type::Object:
             {
                 NCommon::TWriteObjectSettings settings = NCommon::ParseWriteObjectSettings(TExprList(node->Child(4)), ctx);

ydb/core/kqp/provider/yql_kikimr_datasource.cpp

@@ -166,6 +166,8 @@ class TKiSourceIntentDeterminationTransformer: public TKiSourceVisitorTransforme
                 return TStatus::Ok;
             case TKikimrKey::Type::BackupCollection:
                 return TStatus::Ok;
+            case TKikimrKey::Type::Sequence:
+                return TStatus::Ok;
         }
 
         return TStatus::Error;

ydb/core/kqp/provider/yql_kikimr_provider.cpp

@@ -440,6 +440,9 @@ bool TKikimrKey::Extract(const TExprNode& key) {
     } else if(tagName == "permission") {
         KeyType = Type::Permission;
         Target = key.Child(0)->Child(1)->Child(0)->Content();
+    } else if (tagName == "sequence") {
+        KeyType = Type::Sequence;
+        Target = key.Child(0)->Child(1)->Child(0)->Content();
     } else if (tagName == "pgObject") {
         KeyType = Type::PGObject;
         Target = key.Child(0)->Child(1)->Child(0)->Content();

ydb/core/kqp/provider/yql_kikimr_provider_impl.h

@@ -97,6 +97,7 @@ class TKikimrKey {
         PGObject,
         Replication,
         BackupCollection,
+        Sequence,
     };
 
     struct TViewDescription {
@@ -130,6 +131,12 @@ class TKikimrKey {
         return Target;
     }
 
+    TString GetSequencePath() const {
+        Y_DEBUG_ABORT_UNLESS(KeyType.Defined());
+        Y_DEBUG_ABORT_UNLESS(KeyType == Type::Sequence);
+        return Target;
+    }
+
     TString GetReplicationPath() const {
         Y_DEBUG_ABORT_UNLESS(KeyType.Defined());
         Y_DEBUG_ABORT_UNLESS(KeyType == Type::Replication);

ydb/core/kqp/provider/yql_kikimr_type_ann.cpp

@@ -227,6 +227,10 @@ class TKiSourceTypeAnnotationTransformer : public TKiSourceVisitorTransformer {
             {
                 return TStatus::Ok;
             }
+            case TKikimrKey::Type::Sequence:
+            {
+                return TStatus::Ok;
+            }
         }
 
         return TStatus::Error;

ydb/core/kqp/ut/olap/aggregations_ut.cpp

@@ -75,6 +75,33 @@ Y_UNIT_TEST_SUITE(KqpOlapAggregations) {
             Cout << result << Endl;
             CompareYson(result, R"([[23000u;]])");
         }
+
+        {
+            auto alterQuery = TStringBuilder() <<
+                R"(
+                ALTER OBJECT `/Root/olapStore` (TYPE TABLESTORE) SET (ACTION=UPSERT_OPTIONS, `SCAN_READER_POLICY_NAME`=`SIMPLE`)
+                )";
+            auto session = tableClient.CreateSession().GetValueSync().GetSession();
+            auto alterResult = session.ExecuteSchemeQuery(alterQuery).GetValueSync();
+            UNIT_ASSERT_VALUES_EQUAL_C(alterResult.GetStatus(), NYdb::EStatus::SUCCESS, alterResult.GetIssues().ToString());
+        }
+
+        {
+            auto it = tableClient
+                          .StreamExecuteScanQuery(R"(
+                --!syntax_v1
+
+                SELECT
+                    COUNT(*)
+                FROM `/Root/olapStore/olapTable`
+            )")
+                          .GetValueSync();
+
+            UNIT_ASSERT_C(it.IsSuccess(), it.GetIssues().ToString());
+            TString result = StreamResultToYson(it);
+            Cout << result << Endl;
+            CompareYson(result, R"([[23000u;]])");
+        }
     }
 
     Y_UNIT_TEST(AggregationCountPushdown) {