#cache ttl per table cache

src/Access/AeolusAccessUtil.h

@@ -7,7 +7,7 @@
 namespace DB
 {
 
-bool aeolusCheck(const Context & context, const String & full_table_name)
+inline bool aeolusCheck(const Context & context, const String & full_table_name)
 {
     String access_table_names = context.getSettingsRef().access_table_names;
 

src/Analyzers/SubstituteSelectItemToAnyFunction.cpp

@@ -149,7 +149,9 @@ void SubstituteSelectItemToAnyFunction::visit(ASTSelectQuery * select_query)
     // process having and order by
     if (!processed_identifier_qualified_names.empty())
     {
-        SubstituteIdentifierToAnyFunction::Data expression_data{{}, processed_identifier_qualified_names, {}, context, false, false};
+        QualifiedNames empty_qualified_names;
+        NameSet empty_aliases;
+        SubstituteIdentifierToAnyFunction::Data expression_data{empty_qualified_names, processed_identifier_qualified_names, empty_aliases, context, false, false};
         SubstituteIdentifierToAnyFunction expression_visitor(expression_data);
         if (select_query->having())
             expression_visitor.visit(select_query->refHaving());

src/Catalog/Catalog.cpp

@@ -976,9 +976,13 @@ namespace Catalog
             }
 
             StoragePtr storage;
-            if (auto query_context = CurrentThread::getGroup()->query_context.lock())
-                storage = tryGetTableByUUID(*query_context, UUIDHelpers::UUIDToString(uuid), TxnTimestamp::maxTS());
-            else
+            auto thread_group = CurrentThread::getGroup();
+            if (thread_group)
+            {
+                if (auto query_context = thread_group->query_context.lock())
+                    storage = tryGetTableByUUID(*query_context, UUIDHelpers::UUIDToString(uuid), TxnTimestamp::maxTS());
+            }
+            if (!storage)
                 storage = tryGetTableByUUID(context, UUIDHelpers::UUIDToString(uuid), TxnTimestamp::maxTS());
 
             if (auto pcm = context.getPartCacheManager(); pcm && storage)
@@ -4262,7 +4266,7 @@ namespace Catalog
                             return;
 
                         start_key.clear();
-                        auto it = meta_proxy->getAllTransactionRecord(name_space, start_key, max_result_number);
+                        it = meta_proxy->getAllTransactionRecord(name_space, start_key, max_result_number);
                         if (!it->next())
                             return;
                     }

src/CloudServices/CnchMergeMutateThread.cpp

@@ -221,16 +221,22 @@ FutureManipulationTask::~FutureManipulationTask()
 /// Add source parts (include invisible parts) to merging_mutating_parts.
 FutureManipulationTask & FutureManipulationTask::tagSourceParts(ServerDataPartsVector && parts_)
 {
-    auto check_and_add = [&](const auto & part_name) {
-        if (parent.currently_merging_mutating_parts.count(part_name))
-            throw Exception("Part '" + part_name + "' was already in other Task, cancel merge.", ErrorCodes::ABORTED);
-        parent.currently_merging_mutating_parts.emplace(part_name);
-    };
-
     if (!record->try_execute)
     {
         std::lock_guard lock(parent.currently_merging_mutating_parts_mutex);
 
+        std::vector<String> added;
+        auto check_and_add = [&](const auto & part_name) {
+            if (parent.currently_merging_mutating_parts.count(part_name))
+            {
+                for (const auto & n : added)
+                    parent.currently_merging_mutating_parts.erase(n);
+                throw Exception("Part '" + part_name + "' was already in other Task, cancel merge.", ErrorCodes::ABORTED);
+            }
+            parent.currently_merging_mutating_parts.emplace(part_name);
+            added.push_back(part_name);
+        };
+
         for (const auto & p : parts_)
         {
             check_and_add(p->name());

src/CloudServices/CnchServerClient.cpp

@@ -1034,7 +1034,10 @@ brpc::CallId CnchServerClient::submitPreloadTask(const MergeTreeMetaBase & stora
     auto * cntl = new brpc::Controller();
     auto call_id = cntl->call_id();
     if (parts.empty())
+    {
+        delete cntl;
         return call_id;
+    }
 
     Protos::SubmitPreloadTaskReq request;
     request.set_ts(time(nullptr));

src/CloudServices/CnchServerServiceImpl.cpp

@@ -751,49 +751,17 @@ void CnchServerServiceImpl::fetchPartitions(
             ASTPtr query_ptr = deserializeAST(rb);
             /// We should to add `database` into AST before calling `buildSelectQueryInfoForQuery`.
             {
-                StoragePtr storage = gc->getCnchCatalog()->getTable(*gc, request->database(), request->table(), TxnTimestamp::maxTS());
-
-                auto calculated_host
-                    = gc->getCnchTopologyMaster()
-                          ->getTargetServer(UUIDHelpers::UUIDToString(storage->getStorageUUID()), storage->getServerVwName(), true)
-                          .getRPCAddress();
-
-                if (request->remote_host() != calculated_host)
-                    throw Exception(
-                        "Fetch partitions failed because of inconsistent view of topology in remote server, remote_host: "
-                            + request->remote_host() + ", calculated_host: " + calculated_host,
-                        ErrorCodes::LOGICAL_ERROR);
-
-                Names column_names;
-                for (const auto & name : request->column_name_filter())
-                    column_names.push_back(name);
-                auto session_context = Context::createCopy(gc);
-                session_context->setCurrentDatabase(request->database());
-                ReadBufferFromString rb(request->predicate());
-                ASTPtr query_ptr = deserializeAST(rb);
-                /// We should to add `database` into AST before calling `buildSelectQueryInfoForQuery`.
-                {
-                    ASTSelectQuery * select_query = query_ptr->as<ASTSelectQuery>();
-                    if (!select_query)
-                        throw Exception(ErrorCodes::LOGICAL_ERROR, "Unexpected AST type found in buildSelectQueryInfoForQuery");
-                    select_query->replaceDatabaseAndTable(request->database(), request->table());
-                }
-                SelectQueryInfo query_info = buildSelectQueryInfoForQuery(query_ptr, session_context);
-
-                session_context->setTemporaryTransaction(
-                    TxnTimestamp(request->has_txnid() ? request->txnid() : session_context->getTimestamp()), 0, false);
-                auto required_partitions = gc->getCnchCatalog()->getPartitionsByPredicate(
-                    session_context, storage, query_info, column_names, request->has_ignore_ttl() && request->ignore_ttl());
-
-                response->set_total_size(required_partitions.total_partition_number);
-                auto & mutable_partitions = *response->mutable_partitions();
-                for (auto & partition : required_partitions.partitions)
-                    *mutable_partitions.Add() = std::move(partition);
+                ASTSelectQuery * select_query = query_ptr->as<ASTSelectQuery>();
+                if (!select_query)
+                    throw Exception(ErrorCodes::LOGICAL_ERROR, "Unexpected AST type found in buildSelectQueryInfoForQuery");
+                select_query->replaceDatabaseAndTable(request->database(), request->table());
             }
             SelectQueryInfo query_info = buildSelectQueryInfoForQuery(query_ptr, session_context);
 
-            session_context->setTemporaryTransaction(TxnTimestamp(request->has_txnid() ? request->txnid() : session_context->getTimestamp()), 0, false);
-            auto required_partitions = gc->getCnchCatalog()->getPartitionsByPredicate(session_context, storage, query_info, column_names, request->has_ignore_ttl() && request->ignore_ttl());
+            session_context->setTemporaryTransaction(
+                TxnTimestamp(request->has_txnid() ? request->txnid() : session_context->getTimestamp()), 0, false);
+            auto required_partitions = gc->getCnchCatalog()->getPartitionsByPredicate(
+                session_context, storage, query_info, column_names, request->has_ignore_ttl() && request->ignore_ttl());
 
             response->set_total_size(required_partitions.total_partition_number);
             auto & mutable_partitions = *response->mutable_partitions();

src/CloudServices/CnchWorkerClient.cpp

@@ -175,6 +175,60 @@ std::vector<ManipulationInfo> CnchWorkerClient::getManipulationTasksStatus()
     return res;
 }
 
+std::vector<Protos::TTLCacheTableStats> CnchWorkerClient::getTTLCacheStats()
+{
+    brpc::Controller cntl;
+    Protos::GetTTLCacheStatsReq request;
+    Protos::GetTTLCacheStatsResp response;
+
+    stub->getTTLCacheStats(&cntl, &request, &response, nullptr);
+
+    assertController(cntl);
+    RPCHelpers::checkResponse(response);
+
+    std::vector<Protos::TTLCacheTableStats> res;
+    res.reserve(response.tables_size());
+    for (const auto & t : response.tables())
+        res.push_back(t);
+    return res;
+}
+
+std::vector<Protos::TTLCachePartitionStats> CnchWorkerClient::getTTLCachePartitionStats()
+{
+    brpc::Controller cntl;
+    Protos::GetTTLCachePartitionStatsReq request;
+    Protos::GetTTLCachePartitionStatsResp response;
+
+    stub->getTTLCachePartitionStats(&cntl, &request, &response, nullptr);
+
+    assertController(cntl);
+    RPCHelpers::checkResponse(response);
+
+    std::vector<Protos::TTLCachePartitionStats> res;
+    res.reserve(response.partitions_size());
+    for (const auto & p : response.partitions())
+        res.push_back(p);
+    return res;
+}
+
+std::vector<Protos::PreloadPartitionStats> CnchWorkerClient::getPreloadStats()
+{
+    brpc::Controller cntl;
+    Protos::GetPreloadStatsReq request;
+    Protos::GetPreloadStatsResp response;
+
+    stub->getPreloadStats(&cntl, &request, &response, nullptr);
+
+    assertController(cntl);
+    RPCHelpers::checkResponse(response);
+
+    std::vector<Protos::PreloadPartitionStats> res;
+    res.reserve(response.partitions_size());
+    for (const auto & p : response.partitions())
+        res.push_back(p);
+    return res;
+}
+
 void CnchWorkerClient::submitMvRefreshTask(
     const StorageMaterializedView & , const ManipulationTaskParams & params, TxnTimestamp txn_id)
 {

src/CloudServices/CnchWorkerClient.h

@@ -43,6 +43,9 @@ namespace DB
 namespace Protos
 {
     class CnchWorkerService_Stub;
+    class TTLCacheTableStats;
+    class TTLCachePartitionStats;
+    class PreloadPartitionStats;
 }
 
 namespace IngestColumnCnch
@@ -81,6 +84,9 @@ class CnchWorkerClient : public RpcClientBase
     void shutdownManipulationTasks(const UUID & table_uuid, const Strings & task_ids = Strings{});
     std::unordered_set<String> touchManipulationTasks(const UUID & table_uuid, const Strings & tasks_id);
     std::vector<ManipulationInfo> getManipulationTasksStatus();
+    std::vector<Protos::TTLCacheTableStats> getTTLCacheStats();
+    std::vector<Protos::TTLCachePartitionStats> getTTLCachePartitionStats();
+    std::vector<Protos::PreloadPartitionStats> getPreloadStats();
 
     void submitMvRefreshTask(
         const StorageMaterializedView & storage, const ManipulationTaskParams & params, TxnTimestamp txn_id);

src/CloudServices/CnchWorkerServiceImpl.cpp

@@ -34,6 +34,9 @@
 #include <Protos/DataModelHelpers.h>
 #include <Protos/RPCHelpers.h>
 #include <Storages/DiskCache/IDiskCache.h>
+#include <Storages/DiskCache/DiskCacheFactory.h>
+#include <Storages/DiskCache/DiskCacheTTL.h>
+#include <Storages/DiskCache/PreloadRegistry.h>
 #include <Storages/MergeTree/CnchMergeTreeMutationEntry.h>
 #include <Storages/MergeTree/IMergeTreeDataPart_fwd.h>
 #include <Storages/MergeTree/MergeTreeDataPartCNCH.h>
@@ -605,10 +608,23 @@ void CnchWorkerServiceImpl::preloadDataParts(
         }
         else
         {
+            // Group parts by partition and register with PreloadRegistry before scheduling
+            // so in-flight counts are visible immediately.
+            auto & registry = PreloadRegistry::instance();
+            String table_name = cloud_merge_tree.getStorageID().getFullNameNotQuoted();
+            String table_uuid_str = toString(cloud_merge_tree.getStorageUUID());
+            std::unordered_map<String, size_t> partition_counts;
+            for (const auto & part : data_parts)
+                partition_counts[part->info.partition_id]++;
+            for (const auto & [pid, cnt] : partition_counts)
+                registry.registerParts(table_name, table_uuid_str, pid, cnt, preload_level);
+
             ThreadPool * preload_thread_pool = &(IDiskCache::getPreloadPool());
             for (const auto & part : data_parts)
             {
-                preload_thread_pool->scheduleOrThrowOnError([part, preload_level, submit_ts, read_injection, storage] {
+                String pid = part->info.partition_id;
+                preload_thread_pool->scheduleOrThrowOnError([part, preload_level, submit_ts, read_injection, storage, table_uuid_str, pid, &registry] {
+                    SCOPE_EXIT({ registry.partFinished(table_uuid_str, pid); });
                     part->remote_fs_read_failed_injection = read_injection;
                     part->disk_cache_mode = DiskCacheMode::SKIP_DISK_CACHE;// avoid getCheckum & getIndex re-cache
                     part->preload(preload_level, submit_ts);
@@ -1317,6 +1333,104 @@ void CnchWorkerServiceImpl::getCloudMergeTreeStatus(
 {
 }
 
+void CnchWorkerServiceImpl::getTTLCacheStats(
+    google::protobuf::RpcController *,
+    const Protos::GetTTLCacheStatsReq *,
+    Protos::GetTTLCacheStatsResp * response,
+    google::protobuf::Closure * done)
+{
+    SUBMIT_THREADPOOL({
+        auto ttl_caches = DiskCacheFactory::instance().getAllTableTTLCaches();
+        LOG_INFO(log, "getTTLCacheStats: {} TTL cache(s) in registry", ttl_caches.size());
+        for (const auto & [uuid, cache_ptr] : ttl_caches)
+        {
+            auto * ttl_cache = dynamic_cast<DiskCacheTTL *>(cache_ptr.get());
+            if (!ttl_cache)
+                continue;
+
+            auto stats = ttl_cache->getStats();
+            LOG_INFO(log, "getTTLCacheStats: returning stats for table={} uuid={}", ttl_cache->getName(), stats.table_uuid);
+            auto * t = response->add_tables();
+            t->set_table_name(ttl_cache->getName());
+            t->set_table_uuid(stats.table_uuid);
+            t->set_ttl_minutes(ttl_cache->getTTLMinutes());
+            t->set_max_size_bytes(ttl_cache->getMaxSizeBytes());
+            t->set_last_eviction_run(stats.last_eviction_run);
+            t->set_evicted_expired(stats.evicted_expired);
+            t->set_evicted_size_limit(stats.evicted_size_limit);
+            t->set_async_triggered_evicted(stats.async_eviction_triggered);
+            t->set_async_skipped_rate_limit_evicted(stats.async_eviction_skipped_rate_limit);
+            t->set_rejected_non_time_partition(stats.rejected_non_time_partition);
+            t->set_rejected_too_old(stats.rejected_too_old);
+            t->set_count_preload(stats.cached_from_preload);
+            t->set_count_query(stats.cached_from_query);
+            t->set_bytes_preload(stats.cached_bytes_preload);
+            t->set_bytes_query(stats.cached_bytes_query);
+            t->set_count_restored(stats.cached_from_restored);
+            t->set_bytes_restored(stats.cached_bytes_restored);
+            t->set_idx_count_preload(stats.cached_idx_from_preload);
+            t->set_idx_bytes_preload(stats.cached_idx_bytes_preload);
+            t->set_idx_count_query(stats.cached_idx_from_query);
+            t->set_idx_bytes_query(stats.cached_idx_bytes_query);
+            t->set_data_hits(stats.data_hits);
+            t->set_data_misses(stats.data_misses);
+            t->set_idx_hits(stats.idx_hits);
+            t->set_idx_misses(stats.idx_misses);
+        }
+    })
+}
+
+void CnchWorkerServiceImpl::getTTLCachePartitionStats(
+    google::protobuf::RpcController *,
+    const Protos::GetTTLCachePartitionStatsReq *,
+    Protos::GetTTLCachePartitionStatsResp * response,
+    google::protobuf::Closure * done)
+{
+    SUBMIT_THREADPOOL({
+        auto ttl_caches = DiskCacheFactory::instance().getAllTableTTLCaches();
+        LOG_DEBUG(log, "getTTLCachePartitionStats: {} TTL cache(s) in registry", ttl_caches.size());
+        for (const auto & [uuid, cache_ptr] : ttl_caches)
+        {
+            auto * ttl_cache = dynamic_cast<DiskCacheTTL *>(cache_ptr.get());
+            if (!ttl_cache)
+                continue;
+
+            auto table_stats = ttl_cache->getStats();
+            LOG_DEBUG(log, "getTTLCachePartitionStats: returning partition stats for table={} uuid={}", ttl_cache->getName(), table_stats.table_uuid);
+            for (const auto & ps : ttl_cache->getPartitionStats())
+            {
+                auto * p = response->add_partitions();
+                p->set_table_name(ttl_cache->getName());
+                p->set_table_uuid(table_stats.table_uuid);
+                p->set_partition(ps.partition_id);
+                p->set_entry_count(ps.entry_count);
+                p->set_bytes(ps.total_bytes);
+            }
+        }
+    })
+}
+
+void CnchWorkerServiceImpl::getPreloadStats(
+    google::protobuf::RpcController *,
+    const Protos::GetPreloadStatsReq *,
+    Protos::GetPreloadStatsResp * response,
+    google::protobuf::Closure * done)
+{
+    SUBMIT_THREADPOOL({
+        for (const auto & snap : PreloadRegistry::instance().getSnapshot())
+        {
+            auto * p = response->add_partitions();
+            p->set_table_name(snap.table_name);
+            p->set_table_uuid(snap.table_uuid);
+            p->set_partition_id(snap.partition_id);
+            p->set_parts_in_flight(snap.parts_in_flight);
+            p->set_parts_submitted(snap.parts_submitted);
+            p->set_elapsed_ms(snap.elapsed_ms);
+            p->set_preload_level(snap.preload_level);
+        }
+    })
+}
+
 #if defined(__clang__)
 #    pragma clang diagnostic pop
 #else

src/CloudServices/CnchWorkerServiceImpl.h

@@ -67,6 +67,24 @@ class CnchWorkerServiceImpl : protected WithMutableContext, public DB::Protos::C
         Protos::GetManipulationTasksStatusResp * response,
         google::protobuf::Closure * done) override;
 
+    void getTTLCacheStats(
+        google::protobuf::RpcController * cntl,
+        const Protos::GetTTLCacheStatsReq * request,
+        Protos::GetTTLCacheStatsResp * response,
+        google::protobuf::Closure * done) override;
+
+    void getTTLCachePartitionStats(
+        google::protobuf::RpcController * cntl,
+        const Protos::GetTTLCachePartitionStatsReq * request,
+        Protos::GetTTLCachePartitionStatsResp * response,
+        google::protobuf::Closure * done) override;
+
+    void getPreloadStats(
+        google::protobuf::RpcController * cntl,
+        const Protos::GetPreloadStatsReq * request,
+        Protos::GetPreloadStatsResp * response,
+        google::protobuf::Closure * done) override;
+
     void GetPreallocatedStatus(
         google::protobuf::RpcController *,
         const Protos::GetPreallocatedStatusReq * request,