fix case indexation removed table (#8713)

Author: ivanmorozov333

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

@@ -47,6 +47,27 @@ Y_UNIT_TEST_SUITE(KqpOlapWrite) {
         AFL_VERIFY(!Singleton<NKikimr::NWrappers::NExternalStorage::TFakeExternalStorage>()->GetSize());
     }
 
+    Y_UNIT_TEST(TestRemoveTableBeforeIndexation) {
+        auto csController = NKikimr::NYDBTest::TControllers::RegisterCSControllerGuard<NKikimr::NYDBTest::NColumnShard::TController>();
+        csController->SetIndexWriteControllerEnabled(false);
+        csController->SetOverridePeriodicWakeupActivationPeriod(TDuration::Seconds(1));
+        csController->DisableBackground(NKikimr::NYDBTest::ICSController::EBackground::Indexation);
+        csController->DisableBackground(NKikimr::NYDBTest::ICSController::EBackground::Compaction);
+
+        auto settings = TKikimrSettings().SetWithSampleTables(false);
+        TKikimrRunner kikimr(settings);
+        TLocalHelper(kikimr).CreateTestOlapTable();
+        Tests::NCommon::TLoggerInit(kikimr).SetComponents({ NKikimrServices::TX_COLUMNSHARD }, "CS").SetPriority(NActors::NLog::PRI_DEBUG).Initialize();
+        auto tableClient = kikimr.GetTableClient();
+
+        WriteTestData(kikimr, "/Root/olapStore/olapTable", 30000, 1000000, 11000);
+        TTypedLocalHelper("Utf8", kikimr).ExecuteSchemeQuery("DROP TABLE `/Root/olapStore/olapTable`;");
+        csController->EnableBackground(NKikimr::NYDBTest::ICSController::EBackground::Indexation);
+        csController->EnableBackground(NKikimr::NYDBTest::ICSController::EBackground::Compaction);
+        csController->WaitIndexation(TDuration::Seconds(5));
+        csController->WaitCompactions(TDuration::Seconds(5));
+    }
+
     Y_UNIT_TEST(TierDraftsGCWithRestart) {
         auto csController = NKikimr::NYDBTest::TControllers::RegisterCSControllerGuard<NKikimr::NYDBTest::NColumnShard::TController>();
         csController->SetIndexWriteControllerEnabled(false);

ydb/core/tx/columnshard/columnshard_impl.cpp

@@ -638,6 +638,9 @@ void TColumnShard::StartIndexTask(std::vector<const NOlap::TInsertedData*>&& dat
     data.reserve(dataToIndex.size());
     for (auto& ptr : dataToIndex) {
         data.push_back(*ptr);
+        if (!TablesManager.HasTable(data.back().PathId)) {
+            data.back().SetRemove();
+        }
     }
 
     Y_ABORT_UNLESS(data.size());

ydb/core/tx/columnshard/engines/changes/indexation.cpp

@@ -220,13 +220,20 @@ TConclusionStatus TInsertColumnEngineChanges::DoConstructBlobs(TConstructionCont
 
     TPathesData pathBatches(resultSchema);
     for (auto& inserted : DataToIndex) {
+        if (inserted.GetRemove()) {
+            continue;
+        }
         pathBatches.AddChunkInfo(inserted, context);
     }
 
     pathBatches.FinishChunksInfo();
 
     for (auto& inserted : DataToIndex) {
         const TBlobRange& blobRange = inserted.GetBlobRange();
+        if (inserted.GetRemove()) {
+            Blobs.Extract(IStoragesManager::DefaultStorageId, blobRange);
+            continue;
+        }
         auto blobSchema = context.SchemaVersions.GetSchemaVerified(inserted.GetSchemaVersion());
 
         std::shared_ptr<NArrow::TGeneralContainer> batch;
@@ -261,7 +268,7 @@ TConclusionStatus TInsertColumnEngineChanges::DoConstructBlobs(TConstructionCont
         filters.resize(batches.size());
 
         auto itGranule = PathToGranule.find(pathId);
-        AFL_VERIFY(itGranule != PathToGranule.end());
+        AFL_VERIFY(itGranule != PathToGranule.end())("path_id", pathId);
         NCompaction::TMerger merger(context, SaverContext, std::move(batches), std::move(filters));
         merger.SetOptimizationWritingPackMode(true);
         auto localAppended = merger.Execute(stats, itGranule->second, filteredSnapshot, pathId, shardingVersion);

ydb/core/tx/columnshard/engines/column_engine_logs.cpp

@@ -285,7 +285,10 @@ std::shared_ptr<TInsertColumnEngineChanges> TColumnEngineForLogs::StartInsert(st
         if (changes->PathToGranule.contains(pathId)) {
             continue;
         }
-        AFL_VERIFY(changes->PathToGranule.emplace(pathId, GetGranulePtrVerified(pathId)->GetBucketPositions()).second);
+        if (!data.GetRemove()) {
+            AFL_VERIFY(changes->PathToGranule.emplace(pathId, GetGranulePtrVerified(pathId)->GetBucketPositions()).second);
+        }
+        
     }
 
     return changes;

ydb/core/tx/columnshard/engines/insert_table/data.h

@@ -24,6 +24,7 @@ struct TInsertedData {
     };
 
     std::shared_ptr<TBlobStorageGuard> BlobDataGuard;
+    YDB_READONLY(bool, Remove, false);
 
 public:
     ui64 PlanStep = 0;
@@ -36,6 +37,11 @@ struct TInsertedData {
     YDB_READONLY_FLAG(NotAbortable, false);
 
 public:
+    void SetRemove() {
+        AFL_VERIFY(!Remove);
+        Remove = true;
+    }
+
     void MarkAsNotAbortable() {
         NotAbortableFlag = true;
     }