branch-4.1: [refactor](scan) remove SchemaCache to fix concurrent crash in OlapScanner::prepare #62327 (#62414)

Cherry-picked from #62327

Co-authored-by: Chenyang Sun <sunchenyang@selectdb.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: 3 people

be/src/common/config.cpp

@@ -1276,8 +1276,6 @@ DEFINE_mInt64(hdfs_write_batch_buffer_size_mb, "1"); // 1MB
 
 //disable shrink memory by default
 DEFINE_mBool(enable_shrink_memory, "false");
-DEFINE_mInt32(schema_cache_capacity, "1024");
-DEFINE_mInt32(schema_cache_sweep_time_sec, "100");
 
 // max number of segment cache, default -1 for backward compatibility fd_number*2/5
 DEFINE_Int32(segment_cache_capacity, "-1");

be/src/common/config.h

@@ -1320,9 +1320,6 @@ DECLARE_mInt64(file_cache_max_file_reader_cache_size);
 DECLARE_mInt64(hdfs_write_batch_buffer_size_mb);
 //enable shrink memory
 DECLARE_mBool(enable_shrink_memory);
-// enable cache for high concurrent point query work load
-DECLARE_mInt32(schema_cache_capacity);
-DECLARE_mInt32(schema_cache_sweep_time_sec);
 
 // max number of segment cache
 DECLARE_Int32(segment_cache_capacity);

be/src/exec/scan/olap_scanner.cpp

@@ -51,7 +51,6 @@
 #include "runtime/runtime_profile.h"
 #include "runtime/runtime_state.h"
 #include "service/backend_options.h"
-#include "storage/cache/schema_cache.h"
 #include "storage/id_manager.h"
 #include "storage/index/inverted/inverted_index_profile.h"
 #include "storage/iterator/block_reader.h"
@@ -164,68 +163,24 @@ Status OlapScanner::prepare() {
     // value (e.g. select a from t where a .. and b ... limit 1),
     // it will be very slow when reading data in segment iterator
     _tablet_reader->set_batch_size(_state->batch_size());
-    TabletSchemaSPtr cached_schema;
-    std::string schema_key;
     {
         TOlapScanNode& olap_scan_node = local_state->olap_scan_node();
 
-        const auto check_can_use_cache = [&]() {
-            if (!(olap_scan_node.__isset.schema_version && olap_scan_node.__isset.columns_desc &&
-                  !olap_scan_node.columns_desc.empty() &&
-                  olap_scan_node.columns_desc[0].col_unique_id >= 0 && // Why check first column?
-                  tablet->tablet_schema()->num_variant_columns() == 0 &&
-                  tablet->tablet_schema()->num_virtual_columns() == 0)) {
-                return false;
+        // Each scanner builds its own TabletSchema to avoid concurrent modification.
+        tablet_schema = std::make_shared<TabletSchema>();
+        tablet_schema->copy_from(*tablet->tablet_schema());
+        if (olap_scan_node.__isset.columns_desc && !olap_scan_node.columns_desc.empty() &&
+            olap_scan_node.columns_desc[0].col_unique_id >= 0) {
+            tablet_schema->clear_columns();
+            for (const auto& column_desc : olap_scan_node.columns_desc) {
+                tablet_schema->append_column(TabletColumn(column_desc));
             }
-
-            // If `delete_predicates` is not empty, will merge the columns in delete predicate into current tablet schema
-            if (!_tablet_reader_params.delete_predicates.empty()) {
-                return false;
+            if (olap_scan_node.__isset.schema_version) {
+                tablet_schema->set_schema_version(olap_scan_node.schema_version);
             }
-
-            const bool has_pruned_column =
-                    std::ranges::any_of(_output_tuple_desc->slots(), [](const auto& slot) {
-                        if ((slot->type()->get_primitive_type() == PrimitiveType::TYPE_STRUCT ||
-                             slot->type()->get_primitive_type() == PrimitiveType::TYPE_MAP ||
-                             slot->type()->get_primitive_type() == PrimitiveType::TYPE_ARRAY) &&
-                            !slot->all_access_paths().empty()) {
-                            return true;
-                        }
-                        return false;
-                    });
-            return !has_pruned_column;
-        }();
-
-        if (check_can_use_cache) {
-            schema_key =
-                    SchemaCache::get_schema_key(tablet->tablet_id(), olap_scan_node.columns_desc,
-                                                olap_scan_node.schema_version);
-            cached_schema = SchemaCache::instance()->get_schema(schema_key);
         }
-        if (cached_schema && cached_schema->num_virtual_columns() == 0) {
-            tablet_schema = cached_schema;
-        } else {
-            // If schema is not cached or cached schema has virtual columns,
-            // we need to create a new TabletSchema.
-            tablet_schema = std::make_shared<TabletSchema>();
-            tablet_schema->copy_from(*tablet->tablet_schema());
-            if (olap_scan_node.__isset.columns_desc && !olap_scan_node.columns_desc.empty() &&
-                olap_scan_node.columns_desc[0].col_unique_id >= 0) {
-                // Originally scanner get TabletSchema from tablet object in BE.
-                // To support lightweight schema change for adding / dropping columns,
-                // tabletschema is bounded to rowset and tablet's schema maybe outdated,
-                //  so we have to use schema from a query plan witch FE puts it in query plans.
-                tablet_schema->clear_columns();
-                for (const auto& column_desc : olap_scan_node.columns_desc) {
-                    tablet_schema->append_column(TabletColumn(column_desc));
-                }
-                if (olap_scan_node.__isset.schema_version) {
-                    tablet_schema->set_schema_version(olap_scan_node.schema_version);
-                }
-            }
-            if (olap_scan_node.__isset.indexes_desc) {
-                tablet_schema->update_indexes_from_thrift(olap_scan_node.indexes_desc);
-            }
+        if (olap_scan_node.__isset.indexes_desc) {
+            tablet_schema->update_indexes_from_thrift(olap_scan_node.indexes_desc);
         }
 
         if (_tablet_reader_params.rs_splits.empty()) {
@@ -283,12 +238,6 @@ Status OlapScanner::prepare() {
                                   read_columns_to_string(tablet_schema, _return_columns));
     }
 
-    // Add newly created tablet schema to schema cache if it does not have virtual columns.
-    if (cached_schema == nullptr && !schema_key.empty() &&
-        tablet_schema->num_virtual_columns() == 0 && !tablet_schema->has_pruned_columns()) {
-        SchemaCache::instance()->insert_schema(schema_key, tablet_schema);
-    }
-
     if (_tablet_reader_params.score_runtime) {
         SCOPED_TIMER(local_state->_statistics_collect_timer);
         _tablet_reader_params.collection_statistics = std::make_shared<CollectionStatistics>();

be/src/runtime/exec_env.h

@@ -119,7 +119,6 @@ class CdcClientMgr;
 class TabletSchemaCache;
 class TabletColumnObjectPool;
 class UserFunctionCache;
-class SchemaCache;
 class StoragePageCache;
 class AnnIndexIVFListCache;
 class SegmentLoader;
@@ -379,7 +378,6 @@ class ExecEnv {
 
     TabletSchemaCache* get_tablet_schema_cache() { return _tablet_schema_cache; }
     TabletColumnObjectPool* get_tablet_column_object_pool() { return _tablet_column_object_pool; }
-    SchemaCache* schema_cache() { return _schema_cache; }
     StoragePageCache* get_storage_page_cache() { return _storage_page_cache; }
     AnnIndexIVFListCache* get_ann_index_ivf_list_cache() { return _ann_index_ivf_list_cache; }
     SegmentLoader* segment_loader() { return _segment_loader; }
@@ -539,7 +537,6 @@ class ExecEnv {
     TabletSchemaCache* _tablet_schema_cache = nullptr;
     TabletColumnObjectPool* _tablet_column_object_pool = nullptr;
     std::unique_ptr<BaseStorageEngine> _storage_engine;
-    SchemaCache* _schema_cache = nullptr;
     StoragePageCache* _storage_page_cache = nullptr;
     AnnIndexIVFListCache* _ann_index_ivf_list_cache = nullptr;
     SegmentLoader* _segment_loader = nullptr;

be/src/runtime/exec_env_init.cpp

@@ -99,7 +99,6 @@
 #include "service/point_query_executor.h"
 #include "storage/cache/ann_index_ivf_list_cache.h"
 #include "storage/cache/page_cache.h"
-#include "storage/cache/schema_cache.h"
 #include "storage/id_manager.h"
 #include "storage/index/inverted/inverted_index_cache.h"
 #include "storage/olap_define.h"
@@ -641,8 +640,6 @@ Status ExecEnv::init_mem_env() {
               << " segment_cache_capacity: " << segment_cache_capacity
               << " min_segment_cache_mem_limit " << segment_cache_mem_limit;
 
-    _schema_cache = new SchemaCache(config::schema_cache_capacity);
-
     size_t block_file_cache_fd_cache_size =
             std::min((uint64_t)config::file_cache_max_file_reader_cache_size, fd_number / 3);
     LOG(INFO) << "max file reader cache size is: " << block_file_cache_fd_cache_size
@@ -725,7 +722,7 @@ void ExecEnv::init_mem_tracker() {
     _segcompaction_mem_tracker =
             MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::COMPACTION, "SegCompaction");
     _tablets_no_cache_mem_tracker = MemTrackerLimiter::create_shared(
-            MemTrackerLimiter::Type::METADATA, "Tablets(not in SchemaCache, TabletSchemaCache)");
+            MemTrackerLimiter::Type::METADATA, "Tablets(not in TabletSchemaCache)");
     _segments_no_cache_mem_tracker = MemTrackerLimiter::create_shared(
             MemTrackerLimiter::Type::METADATA, "Segments(not in SegmentCache)");
     _rowsets_no_cache_mem_tracker =
@@ -885,7 +882,6 @@ void ExecEnv::destroy() {
     SAFE_DELETE(_encoding_info_resolver);
     SAFE_DELETE(_condition_cache);
     SAFE_DELETE(_lookup_connection_cache);
-    SAFE_DELETE(_schema_cache);
     SAFE_DELETE(_segment_loader);
     SAFE_DELETE(_row_cache);
     SAFE_DELETE(_query_cache);

be/src/runtime/memory/cache_policy.h

@@ -149,8 +149,8 @@ class CachePolicy {
         }
     }
 
-    inline static std::vector<CacheType> MetadataCache {
-            CacheType::SEGMENT_CACHE, CacheType::SCHEMA_CACHE, CacheType::TABLET_SCHEMA_CACHE};
+    inline static std::vector<CacheType> MetadataCache {CacheType::SEGMENT_CACHE,
+                                                        CacheType::TABLET_SCHEMA_CACHE};
 
     CachePolicy(CacheType type, size_t capacity, uint32_t stale_sweep_time_s, bool enable_prune);
     virtual ~CachePolicy();

be/src/runtime/memory/memory_profile.cpp

@@ -23,8 +23,8 @@
 #include "runtime/memory/jemalloc_control.h"
 #include "runtime/memory/mem_tracker_limiter.h"
 #include "runtime/runtime_profile.h"
-#include "storage/cache/schema_cache.h"
 #include "storage/metadata_adder.h"
+#include "storage/segment/segment_loader.h"
 #include "storage/tablet/tablet_schema_cache.h"
 #include "util/mem_info.h"
 
@@ -162,8 +162,7 @@ void MemoryProfile::refresh_memory_overview_profile() {
     // 2 refresh metadata memory tracker
     ExecEnv::GetInstance()->tablets_no_cache_mem_tracker()->set_consumption(
             MetadataAdder<TabletMeta>::get_all_tablets_size() -
-            TabletSchemaCache::instance()->value_mem_consumption() -
-            SchemaCache::instance()->value_mem_consumption());
+            TabletSchemaCache::instance()->value_mem_consumption());
     ExecEnv::GetInstance()->rowsets_no_cache_mem_tracker()->set_consumption(
             MetadataAdder<RowsetMeta>::get_all_rowsets_size());
     ExecEnv::GetInstance()->segments_no_cache_mem_tracker()->set_consumption(

be/src/storage/cache/schema_cache.cpp

@@ -26,7 +26,6 @@
 #include "core/block/column_with_type_and_name.h"
 #include "core/column/column.h"
 #include "core/data_type/data_type.h"
-#include "storage/cache/schema_cache.h"
 #include "storage/field.h"
 #include "storage/iterators.h"
 #include "storage/olap_common.h"