fix indexes usage, option for exclusive pk case (#3731)

ivanmorozov333 · web-flow · commit 26b8ef392963 · 2024-04-15T17:10:57.000+03:00
diff --git a/ydb/core/kqp/gateway/behaviour/tablestore/operations/upsert_opt.cpp b/ydb/core/kqp/gateway/behaviour/tablestore/operations/upsert_opt.cpp
@@ -10,11 +10,16 @@ TConclusionStatus TUpsertOptionsOperation::DoDeserialize(NYql::TObjectSettingsIm
         TConclusionStatus::Fail("Incorrect value for SCHEME_NEED_ACTUALIZATION: cannot parse as boolean");
     }
     SchemeNeedActualization = *value;
+    ExternalGuaranteeExclusivePK = features.Extract<bool>("EXTERNAL_GUARANTEE_EXCLUSIVE_PK");
     return TConclusionStatus::Success();
 }
 
 void TUpsertOptionsOperation::DoSerializeScheme(NKikimrSchemeOp::TAlterColumnTableSchema& schemaData) const {
     schemaData.MutableOptions()->SetSchemeNeedActualization(SchemeNeedActualization);
+    if (ExternalGuaranteeExclusivePK) {
+        schemaData.MutableOptions()->SetExternalGuaranteeExclusivePK(*ExternalGuaranteeExclusivePK);
+    }
+    
 }
 
 }
diff --git a/ydb/core/kqp/gateway/behaviour/tablestore/operations/upsert_opt.h b/ydb/core/kqp/gateway/behaviour/tablestore/operations/upsert_opt.h
@@ -12,6 +12,7 @@ class TUpsertOptionsOperation : public ITableStoreOperation {
     static inline const auto Registrator = TFactory::TRegistrator<TUpsertOptionsOperation>(GetTypeName());
 private:
     bool SchemeNeedActualization = false;
+    std::optional<bool> ExternalGuaranteeExclusivePK;
 public:
     TConclusionStatus DoDeserialize(NYql::TObjectSettingsImpl::TFeaturesExtractor& features) override;
 
diff --git a/ydb/core/kqp/ut/olap/indexes_ut.cpp b/ydb/core/kqp/ut/olap/indexes_ut.cpp
@@ -19,6 +19,7 @@ Y_UNIT_TEST_SUITE(KqpOlapIndexes) {
         auto csController = NYDBTest::TControllers::RegisterCSControllerGuard<NYDBTest::NColumnShard::TController>();
         csController->SetPeriodicWakeupActivationPeriod(TDuration::Seconds(1));
         csController->SetLagForCompactionBeforeTierings(TDuration::Seconds(1));
+        csController->SetOverrideReduceMemoryIntervalLimit(1LLU << 30);
 
         TLocalHelper(kikimr).CreateTestOlapTable();
         auto tableClient = kikimr.GetTableClient();
@@ -77,7 +78,7 @@ Y_UNIT_TEST_SUITE(KqpOlapIndexes) {
 
         {
             auto alterQuery = TStringBuilder() <<
-                "ALTER OBJECT `/Root/olapStore` (TYPE TABLESTORE) SET (ACTION=UPSERT_OPTIONS, SCHEME_NEED_ACTUALIZATION=`true`);";
+                "ALTER OBJECT `/Root/olapStore` (TYPE TABLESTORE) SET (ACTION=UPSERT_OPTIONS, SCHEME_NEED_ACTUALIZATION=`true`, EXTERNAL_GUARANTEE_EXCLUSIVE_PK=`true`);";
             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());
@@ -104,7 +105,7 @@ Y_UNIT_TEST_SUITE(KqpOlapIndexes) {
         }
     }
 
-    Y_UNIT_TEST(IndexesActualizationRebuildScheme) {
+    Y_UNIT_TEST(SchemeActualizationOnceOnStart) {
         auto settings = TKikimrSettings()
             .SetWithSampleTables(false);
         TKikimrRunner kikimr(settings);
@@ -179,6 +180,7 @@ Y_UNIT_TEST_SUITE(KqpOlapIndexes) {
 //        Tests::NCommon::TLoggerInit(kikimr).Initialize();
 
         auto csController = NYDBTest::TControllers::RegisterCSControllerGuard<NYDBTest::NColumnShard::TController>();
+        csController->SetOverrideReduceMemoryIntervalLimit(1LLU << 30);
 
         {
             auto alterQuery = TStringBuilder() <<
@@ -198,6 +200,13 @@ Y_UNIT_TEST_SUITE(KqpOlapIndexes) {
             auto alterResult = session.ExecuteSchemeQuery(alterQuery).GetValueSync();
             UNIT_ASSERT_VALUES_EQUAL_C(alterResult.GetStatus(), NYdb::EStatus::SUCCESS, alterResult.GetIssues().ToString());
         }
+        {
+            auto alterQuery = TStringBuilder() <<
+                "ALTER OBJECT `/Root/olapStore` (TYPE TABLESTORE) SET (ACTION=UPSERT_OPTIONS, EXTERNAL_GUARANTEE_EXCLUSIVE_PK=`true`);";
+            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());
+        }
 
         std::vector<TString> uids;
         std::vector<TString> resourceIds;
diff --git a/ydb/core/protos/flat_scheme_op.proto b/ydb/core/protos/flat_scheme_op.proto
@@ -495,6 +495,7 @@ message TStorageTierConfig {
 
 message TColumnTableSchemeOptions {
     optional bool SchemeNeedActualization = 1 [default = false];
+    optional bool ExternalGuaranteeExclusivePK = 2 [default = false];
 }
 
 message TColumnTableSchema {
@@ -523,6 +524,7 @@ message TColumnTableSchema {
 
 message TColumnTableRequestedOptions {
     optional bool SchemeNeedActualization = 1 [default = false];
+    optional bool ExternalGuaranteeExclusivePK = 2;
 }
 
 message TAlterColumnTableSchema {
diff --git a/ydb/core/tx/columnshard/engines/reader/abstract/read_metadata.h b/ydb/core/tx/columnshard/engines/reader/abstract/read_metadata.h
@@ -62,6 +62,10 @@ struct TReadMetadataBase {
         return ResultIndexSchema;
     }
 
+    bool HasGuaranteeExclusivePK() const {
+        return GetIndexInfo().GetExternalGuaranteeExclusivePK();
+    }
+
     ISnapshotSchema::TPtr GetLoadSchemaVerified(const TPortionInfo& porition) const;
 
     std::shared_ptr<arrow::Schema> GetBlobSchema(const ui64 version) const {
diff --git a/ydb/core/tx/columnshard/engines/reader/plain_reader/iterator/context.cpp b/ydb/core/tx/columnshard/engines/reader/plain_reader/iterator/context.cpp
@@ -34,7 +34,9 @@ std::shared_ptr<TFetchingScript> TSpecialReadContext::GetColumnsFetchingPlan(con
     const bool needSnapshots = !source->GetExclusiveIntervalOnly() || ReadMetadata->GetRequestSnapshot() < source->GetRecordSnapshotMax() || !source->IsSourceInMemory();
     const bool partialUsageByPK = ReadMetadata->GetPKRangesFilter().IsPortionInPartialUsage(source->GetStartReplaceKey(), source->GetFinishReplaceKey(), ReadMetadata->GetIndexInfo());
     const bool useIndexes = (IndexChecker ? source->HasIndexes(IndexChecker->GetIndexIds()) : false);
-    if (auto result = CacheFetchingScripts[needSnapshots ? 1 : 0][(source->GetExclusiveIntervalOnly() && source->IsSourceInMemory()) ? 1 : 0][partialUsageByPK ? 1 : 0][useIndexes ? 1 : 0]) {
+    const bool isWholeExclusiveSource = source->GetExclusiveIntervalOnly() && source->IsSourceInMemory();
+    if (auto result = CacheFetchingScripts[needSnapshots ? 1 : 0][isWholeExclusiveSource ? 1 : 0][partialUsageByPK ? 1 : 0][useIndexes ? 1 : 0]) {
+//        AFL_WARN(NKikimrServices::TX_COLUMNSHARD_SCAN)("SS", needSnapshots)("PK", partialUsageByPK)("IDX", useIndexes)("EXCL", source->GetExclusiveIntervalOnly())("MEM", source->IsSourceInMemory())("result", result->DebugString());
         return result;
     }
     {
@@ -49,7 +51,7 @@ std::shared_ptr<TFetchingScript> TSpecialReadContext::BuildColumnsFetchingPlan(c
     const bool partialUsageByPredicateExt, const bool useIndexes) const {
     std::shared_ptr<TFetchingScript> result = std::make_shared<TFetchingScript>();
     const bool partialUsageByPredicate = partialUsageByPredicateExt && PredicateColumns->GetColumnsCount();
-    if (!!IndexChecker && useIndexes) {
+    if (!!IndexChecker && useIndexes && exclusiveSource) {
         result->AddStep(std::make_shared<TIndexBlobsFetchingStep>(std::make_shared<TIndexesSet>(IndexChecker->GetIndexIds())));
         result->AddStep(std::make_shared<TApplyIndexStep>(IndexChecker));
     }
diff --git a/ydb/core/tx/columnshard/engines/reader/plain_reader/iterator/scanner.cpp b/ydb/core/tx/columnshard/engines/reader/plain_reader/iterator/scanner.cpp
@@ -56,12 +56,13 @@ void TScanHead::OnIntervalResult(const std::optional<NArrow::TShardedRecordBatch
 }
 
 TConclusionStatus TScanHead::Start() {
+    const bool guaranteeExclusivePK = Context->GetCommonContext()->GetReadMetadata()->HasGuaranteeExclusivePK();
     TScanContext context;
     for (auto itPoint = BorderPoints.begin(); itPoint != BorderPoints.end(); ++itPoint) {
         auto& point = itPoint->second;
         context.OnStartPoint(point);
         if (context.GetIsSpecialPoint()) {
-            auto detectorResult = DetectSourcesFeatureInContextIntervalScan(context.GetCurrentSources(), false);
+            auto detectorResult = DetectSourcesFeatureInContextIntervalScan(context.GetCurrentSources(), guaranteeExclusivePK);
             for (auto&& i : context.GetCurrentSources()) {
                 i.second->IncIntervalsCount();
             }
@@ -82,7 +83,7 @@ TConclusionStatus TScanHead::Start() {
             for (auto&& i : context.GetCurrentSources()) {
                 i.second->IncIntervalsCount();
             }
-            auto detectorResult = DetectSourcesFeatureInContextIntervalScan(context.GetCurrentSources(), context.GetIsExclusiveInterval());
+            auto detectorResult = DetectSourcesFeatureInContextIntervalScan(context.GetCurrentSources(), guaranteeExclusivePK || context.GetIsExclusiveInterval());
             if (!detectorResult) {
                 AFL_ERROR(NKikimrServices::TX_COLUMNSHARD_SCAN)("event", "scanner_initializer_aborted")("reason", detectorResult.GetErrorMessage());
                 Abort();
diff --git a/ydb/core/tx/columnshard/engines/scheme/index_info.cpp b/ydb/core/tx/columnshard/engines/scheme/index_info.cpp
@@ -320,6 +320,7 @@ bool TIndexInfo::DeserializeFromProto(const NKikimrSchemeOp::TColumnTableSchema&
 
     {
         SchemeNeedActualization = schema.GetOptions().GetSchemeNeedActualization();
+        ExternalGuaranteeExclusivePK = schema.GetOptions().GetExternalGuaranteeExclusivePK();
     }
 
     if (schema.HasDefaultCompression()) {
diff --git a/ydb/core/tx/columnshard/engines/scheme/index_info.h b/ydb/core/tx/columnshard/engines/scheme/index_info.h
@@ -42,12 +42,17 @@ struct TIndexInfo : public NTable::TScheme::TTableSchema, public IIndexInfo {
     std::map<TString, NStatistics::TOperatorContainer> StatisticsByName;
     TIndexInfo(const TString& name);
     bool SchemeNeedActualization = false;
+    bool ExternalGuaranteeExclusivePK = false;
     bool DeserializeFromProto(const NKikimrSchemeOp::TColumnTableSchema& schema, const std::shared_ptr<IStoragesManager>& operators);
     TColumnFeatures& GetOrCreateColumnFeatures(const ui32 columnId) const;
     void BuildSchemaWithSpecials();
     void BuildArrowSchema();
     void InitializeCaches(const std::shared_ptr<IStoragesManager>& operators);
 public:
+    bool GetExternalGuaranteeExclusivePK() const {
+        return ExternalGuaranteeExclusivePK;
+    }
+
     const TColumnFeatures& GetColumnFeaturesVerified(const ui32 columnId) const {
         auto it = ColumnFeatures.find(columnId);
         AFL_VERIFY(it != ColumnFeatures.end());
diff --git a/ydb/core/tx/schemeshard/olap/options/schema.cpp b/ydb/core/tx/schemeshard/olap/options/schema.cpp
@@ -4,15 +4,24 @@ namespace NKikimr::NSchemeShard {
 
 bool TOlapOptionsDescription::ApplyUpdate(const TOlapOptionsUpdate& schemaUpdate, IErrorCollector& /*errors*/) {
     SchemeNeedActualization = schemaUpdate.GetSchemeNeedActualization();
+    if (!!schemaUpdate.GetExternalGuaranteeExclusivePK()) {
+        ExternalGuaranteeExclusivePK = *schemaUpdate.GetExternalGuaranteeExclusivePK();
+    }
     return true;
 }
 
 void TOlapOptionsDescription::Parse(const NKikimrSchemeOp::TColumnTableSchema& tableSchema) {
     SchemeNeedActualization = tableSchema.GetOptions().GetSchemeNeedActualization();
+    if (tableSchema.GetOptions().HasExternalGuaranteeExclusivePK()) {
+        ExternalGuaranteeExclusivePK = tableSchema.GetOptions().GetExternalGuaranteeExclusivePK();
+    }
 }
 
 void TOlapOptionsDescription::Serialize(NKikimrSchemeOp::TColumnTableSchema& tableSchema) const {
     tableSchema.MutableOptions()->SetSchemeNeedActualization(SchemeNeedActualization);
+    if (ExternalGuaranteeExclusivePK) {
+        tableSchema.MutableOptions()->SetExternalGuaranteeExclusivePK(ExternalGuaranteeExclusivePK);
+    }
 }
 
 bool TOlapOptionsDescription::Validate(const NKikimrSchemeOp::TColumnTableSchema& /*opSchema*/, IErrorCollector& /*errors*/) const {
diff --git a/ydb/core/tx/schemeshard/olap/options/schema.h b/ydb/core/tx/schemeshard/olap/options/schema.h
@@ -8,6 +8,7 @@ class TOlapSchema;
 class TOlapOptionsDescription {
 private:
     YDB_READONLY(bool, SchemeNeedActualization, false);
+    YDB_READONLY(bool, ExternalGuaranteeExclusivePK, false);
 public:
     bool ApplyUpdate(const TOlapOptionsUpdate& schemaUpdate, IErrorCollector& errors);
 
diff --git a/ydb/core/tx/schemeshard/olap/options/update.h b/ydb/core/tx/schemeshard/olap/options/update.h
@@ -10,13 +10,21 @@ namespace NKikimr::NSchemeShard {
     class TOlapOptionsUpdate {
     private:
         YDB_ACCESSOR(bool, SchemeNeedActualization, false);
+        YDB_ACCESSOR_DEF(std::optional<bool>, ExternalGuaranteeExclusivePK);
     public:
         bool Parse(const NKikimrSchemeOp::TAlterColumnTableSchema& alterRequest, IErrorCollector& /*errors*/) {
             SchemeNeedActualization = alterRequest.GetOptions().GetSchemeNeedActualization();
+            if (alterRequest.GetOptions().HasExternalGuaranteeExclusivePK()) {
+                ExternalGuaranteeExclusivePK = alterRequest.GetOptions().GetExternalGuaranteeExclusivePK();
+            }
             return true;
         }
         void SerializeToProto(NKikimrSchemeOp::TAlterColumnTableSchema& alterRequest) const {
             alterRequest.MutableOptions()->SetSchemeNeedActualization(SchemeNeedActualization);
+            if (ExternalGuaranteeExclusivePK) {
+                alterRequest.MutableOptions()->SetExternalGuaranteeExclusivePK(*ExternalGuaranteeExclusivePK);
+            }
+            
         }
     };
 }

Original file line number	Diff line number	Diff line change
`@@ -10,11 +10,16 @@ TConclusionStatus TUpsertOptionsOperation::DoDeserialize(NYql::TObjectSettingsIm`
`10`	`10`	`TConclusionStatus::Fail("Incorrect value for SCHEME_NEED_ACTUALIZATION: cannot parse as boolean");`
`11`	`11`	`}`
`12`	`12`	`SchemeNeedActualization = *value;`
	`13`	`+ ExternalGuaranteeExclusivePK = features.Extract<bool>("EXTERNAL_GUARANTEE_EXCLUSIVE_PK");`
`13`	`14`	`return TConclusionStatus::Success();`
`14`	`15`	`}`
`15`	`16`
`16`	`17`	`void TUpsertOptionsOperation::DoSerializeScheme(NKikimrSchemeOp::TAlterColumnTableSchema& schemaData) const {`
`17`	`18`	`schemaData.MutableOptions()->SetSchemeNeedActualization(SchemeNeedActualization);`
	`19`	`+ if (ExternalGuaranteeExclusivePK) {`
	`20`	`+ schemaData.MutableOptions()->SetExternalGuaranteeExclusivePK(*ExternalGuaranteeExclusivePK);`
	`21`	`+ }`
	`22`	`+`
`18`	`23`	`}`
`19`	`24`
`20`	`25`	`}`
Original file line number	Diff line number	Diff line change
`@@ -495,6 +495,7 @@ message TStorageTierConfig {`
`495`	`495`
`496`	`496`	`message TColumnTableSchemeOptions {`
`497`	`497`	`optional bool SchemeNeedActualization = 1 [default = false];`
	`498`	`+ optional bool ExternalGuaranteeExclusivePK = 2 [default = false];`
`498`	`499`	`}`
`499`	`500`
`500`	`501`	`message TColumnTableSchema {`
`@@ -523,6 +524,7 @@ message TColumnTableSchema {`
`523`	`524`
`524`	`525`	`message TColumnTableRequestedOptions {`
`525`	`526`	`optional bool SchemeNeedActualization = 1 [default = false];`
	`527`	`+ optional bool ExternalGuaranteeExclusivePK = 2;`
`526`	`528`	`}`
`527`	`529`
`528`	`530`	`message TAlterColumnTableSchema {`
Original file line number	Diff line number	Diff line change
`@@ -62,6 +62,10 @@ struct TReadMetadataBase {`
`62`	`62`	`return ResultIndexSchema;`
`63`	`63`	`}`
`64`	`64`
	`65`	`+ bool HasGuaranteeExclusivePK() const {`
	`66`	`+ return GetIndexInfo().GetExternalGuaranteeExclusivePK();`
	`67`	`+ }`
	`68`	`+`
`65`	`69`	`ISnapshotSchema::TPtr GetLoadSchemaVerified(const TPortionInfo& porition) const;`
`66`	`70`
`67`	`71`	`std::shared_ptr<arrow::Schema> GetBlobSchema(const ui64 version) const {`
Original file line number	Diff line number	Diff line change
`@@ -34,7 +34,9 @@ std::shared_ptr<TFetchingScript> TSpecialReadContext::GetColumnsFetchingPlan(con`
`34`	`34`	`const bool needSnapshots = !source->GetExclusiveIntervalOnly() \|\| ReadMetadata->GetRequestSnapshot() < source->GetRecordSnapshotMax() \|\| !source->IsSourceInMemory();`
`35`	`35`	`const bool partialUsageByPK = ReadMetadata->GetPKRangesFilter().IsPortionInPartialUsage(source->GetStartReplaceKey(), source->GetFinishReplaceKey(), ReadMetadata->GetIndexInfo());`
`36`	`36`	`const bool useIndexes = (IndexChecker ? source->HasIndexes(IndexChecker->GetIndexIds()) : false);`
`37`		`- if (auto result = CacheFetchingScripts[needSnapshots ? 1 : 0][(source->GetExclusiveIntervalOnly() && source->IsSourceInMemory()) ? 1 : 0][partialUsageByPK ? 1 : 0][useIndexes ? 1 : 0]) {`
	`37`	`+ const bool isWholeExclusiveSource = source->GetExclusiveIntervalOnly() && source->IsSourceInMemory();`
	`38`	`+ if (auto result = CacheFetchingScripts[needSnapshots ? 1 : 0][isWholeExclusiveSource ? 1 : 0][partialUsageByPK ? 1 : 0][useIndexes ? 1 : 0]) {`
	`39`	`+// AFL_WARN(NKikimrServices::TX_COLUMNSHARD_SCAN)("SS", needSnapshots)("PK", partialUsageByPK)("IDX", useIndexes)("EXCL", source->GetExclusiveIntervalOnly())("MEM", source->IsSourceInMemory())("result", result->DebugString());`
`38`	`40`	`return result;`
`39`	`41`	`}`
`40`	`42`	`{`
`@@ -49,7 +51,7 @@ std::shared_ptr<TFetchingScript> TSpecialReadContext::BuildColumnsFetchingPlan(c`
`49`	`51`	`const bool partialUsageByPredicateExt, const bool useIndexes) const {`
`50`	`52`	`std::shared_ptr<TFetchingScript> result = std::make_shared<TFetchingScript>();`
`51`	`53`	`const bool partialUsageByPredicate = partialUsageByPredicateExt && PredicateColumns->GetColumnsCount();`
`52`		`- if (!!IndexChecker && useIndexes) {`
	`54`	`+ if (!!IndexChecker && useIndexes && exclusiveSource) {`
`53`	`55`	`result->AddStep(std::make_shared<TIndexBlobsFetchingStep>(std::make_shared<TIndexesSet>(IndexChecker->GetIndexIds())));`
`54`	`56`	`result->AddStep(std::make_shared<TApplyIndexStep>(IndexChecker));`
`55`	`57`	`}`
Original file line number	Diff line number	Diff line change
`@@ -320,6 +320,7 @@ bool TIndexInfo::DeserializeFromProto(const NKikimrSchemeOp::TColumnTableSchema&`
`320`	`320`
`321`	`321`	`{`
`322`	`322`	`SchemeNeedActualization = schema.GetOptions().GetSchemeNeedActualization();`
	`323`	`+ ExternalGuaranteeExclusivePK = schema.GetOptions().GetExternalGuaranteeExclusivePK();`
`323`	`324`	`}`
`324`	`325`
`325`	`326`	`if (schema.HasDefaultCompression()) {`