Fix the bug that some decoding is not protected by alter_lock

dbms/src/Storages/Transaction/ApplySnapshot.cpp

@@ -155,7 +155,7 @@ void KVStore::onSnapshot(const RegionPtrWrap & new_region_wrap, RegionPtr old_re
             try
             {
                 auto & context = tmt.getContext();
-                // Acquire lock so that no other threads can drop storage
+                // Acquire `drop_lock` so that no other threads can drop the storage. `alter_lock` is not required.
                 auto table_lock = storage->lockForShare(getThreadName());
                 auto dm_storage = std::dynamic_pointer_cast<StorageDeltaMerge>(storage);
                 auto key_range = DM::RowKeyRange::fromRegionRange(
@@ -583,13 +583,13 @@ RegionPtr KVStore::handleIngestSSTByDTFile(const RegionPtr & region, const SSTVi
             auto & context = tmt.getContext();
             try
             {
-                // Acquire lock so that no other threads can drop storage
+                // Acquire `drop_lock` so that no other threads can drop the storage. `alter_lock` is not required.
                 auto table_lock = storage->lockForShare(getThreadName());
-                auto dm_storage = std::dynamic_pointer_cast<StorageDeltaMerge>(storage);
                 auto key_range = DM::RowKeyRange::fromRegionRange(
                     region->getRange(), table_id, storage->isCommonHandle(), storage->getRowKeyColumnSize());
                 // Call `ingestFiles` to ingest external DTFiles.
                 // Note that ingest sst won't remove the data in the key range
+                auto dm_storage = std::dynamic_pointer_cast<StorageDeltaMerge>(storage);
                 dm_storage->ingestFiles(key_range, ingest_ids, /*clear_data_in_range=*/false, context.getSettingsRef());
             }
             catch (DB::Exception & e)

dbms/src/Storages/Transaction/KVStore.cpp

@@ -109,7 +109,7 @@ void KVStore::tryFlushRegionCacheInStorage(TMTContext & tmt, const Region & regi
 
         try
         {
-            // Try to get a read lock on `storage` so it won't be dropped during `flushCache`
+            // Acquire `drop_lock` so that no other threads can drop the storage during `flushCache`. `alter_lock` is not required.
             auto storage_lock = storage->lockForShare(getThreadName());
             auto rowkey_range = DM::RowKeyRange::fromRegionRange(
                 region.getRange(), region.getRange()->getMappedTableID(), storage->isCommonHandle(), storage->getRowKeyColumnSize());

dbms/src/Storages/Transaction/PartitionStreams.cpp

@@ -56,15 +56,15 @@ static void writeRegionDataToStorage(
                 return true;
         }
 
-        /// Lock throughout decode and write, during which schema must not change.
+        /// Get a structure read lock throughout decode, during which schema must not change.
         TableStructureLockHolder lock;
         try
         {
             lock = storage->lockStructureForShare(getThreadName());
         }
         catch (DB::Exception & e)
         {
-            // If the storage is physical dropped (but not removed from `ManagedStorages`) when we want to flsuh raft data into it, consider the write done.
+            // If the storage is physical dropped (but not removed from `ManagedStorages`) when we want to write raft data into it, consider the write done.
             if (e.code() == ErrorCodes::TABLE_IS_DROPPED)
                 return true;
             else
@@ -439,7 +439,7 @@ RegionPtrWithBlock::CachePtr GenRegionPreDecodeBlockData(const RegionPtr & regio
                 return true;
         }
 
-        /// Lock throughout decode and write, during which schema must not change.
+        /// Get a structure read lock throughout decode, during which schema must not change.
         TableStructureLockHolder lock;
         try
         {
@@ -502,7 +502,9 @@ AtomicGetStorageSchema(const RegionPtr & region, TMTContext & tmt)
             if (storage == nullptr) // Table must have just been GC-ed
                 return true;
         }
-        auto table_lock = storage->lockForShare(getThreadName());
+        // Get a structure read lock. It will throw exception if the table has been dropped,
+        // the caller should handle this situation.
+        auto table_lock = storage->lockStructureForShare(getThreadName());
         is_common_handle = storage->isCommonHandle();
         if (unlikely(storage->engineType() != ::TiDB::StorageEngine::DT))
         {
@@ -611,7 +613,9 @@ Block GenRegionBlockDatawithSchema(const RegionPtr & region,
 
     {
         Stopwatch watch;
-        auto table_lock = dm_storage->lockForShare(getThreadName());
+        // Get a structure read lock. It will throw exception if the table has been
+        // dropped, the caller should handle this situation.
+        auto table_lock = dm_storage->lockStructureForShare(getThreadName());
         // Compare schema_snap with current schema, throw exception if changed.
         auto store = dm_storage->getStore();
         auto cur_schema_snap = store->getStoreColumns();

dbms/src/Storages/Transaction/RegionTable.cpp

@@ -238,8 +238,7 @@ void removeObsoleteDataInStorage(
 
     try
     {
-        // acquire a read lock so that no other threads can drop the `storage`
-        // if storage is already dropped, this will throw exception
+        // Acquire a `drop_lock` so that no other threads can drop the `storage`
         auto storage_lock = storage->lockForShare(getThreadName());
 
         auto dm_storage = std::dynamic_pointer_cast<StorageDeltaMerge>(storage);
@@ -254,7 +253,7 @@ void removeObsoleteDataInStorage(
     }
     catch (DB::Exception & e)
     {
-        // We can ignore if storage is already dropped.
+        // We can ignore if the storage is already dropped.
         if (e.code() != ErrorCodes::TABLE_IS_DROPPED)
             throw;
     }