Merge pull request ClickHouse#79963 from ClickHouse/revert-79898-revert-pushing-to-views

Revert "Revert "rework pushing to views ClickHouse#77309""

Author: CheSema

src/Client/LocalConnection.cpp

@@ -466,6 +466,23 @@ bool LocalConnection::poll(size_t)
         return true;
     }
 
+    // pushing executors have to be finished before the final stats are sent
+    if (state->is_finished)
+    {
+        if (state->executor)
+        {
+            // no op
+        }
+        else if (state->pushing_async_executor)
+        {
+            state->pushing_async_executor->finish();
+        }
+        else if (state->pushing_executor)
+        {
+            state->pushing_executor->finish();
+        }
+    }
+
     if (state->is_finished && !state->sent_totals)
     {
         state->sent_totals = true;
@@ -514,7 +531,7 @@ bool LocalConnection::poll(size_t)
     {
         state->sent_profile_events = true;
 
-        if (send_profile_events && state->executor)
+        if (send_profile_events && (state->executor || state->pushing_async_executor || state->pushing_executor))
         {
             sendProfileEvents();
             return true;

src/Common/ErrorCodes.cpp

@@ -625,6 +625,7 @@
     M(743, ICEBERG_SPECIFICATION_VIOLATION) \
     M(744, SESSION_ID_EMPTY) \
     M(745, SERVER_OVERLOADED) \
+    M(746, DEPENDENCIES_NOT_FOUND) \
 \
     M(900, DISTRIBUTED_CACHE_ERROR) \
     M(901, CANNOT_USE_DISTRIBUTED_CACHE) \

src/Common/Exception.cpp

@@ -593,15 +593,15 @@ void tryLogException(std::exception_ptr e, const char * log_name, const std::str
     }
 }
 
-void tryLogException(std::exception_ptr e, LoggerPtr logger, const std::string & start_of_message)
+void tryLogException(std::exception_ptr e, LoggerPtr logger, const std::string & start_of_message, LogsLevel level)
 {
     try
     {
         std::rethrow_exception(std::move(e)); // NOLINT
     }
     catch (...)
     {
-        tryLogCurrentException(logger, start_of_message);
+        tryLogCurrentException(logger, start_of_message, level);
     }
 }
 

src/Common/Exception.h

@@ -326,7 +326,7 @@ struct ExecutionStatus
 
 /// TODO: Logger leak constexpr overload
 void tryLogException(std::exception_ptr e, const char * log_name, const std::string & start_of_message = "");
-void tryLogException(std::exception_ptr e, LoggerPtr logger, const std::string & start_of_message = "");
+void tryLogException(std::exception_ptr e, LoggerPtr logger, const std::string & start_of_message = "", LogsLevel level = LogsLevel::error);
 void tryLogException(std::exception_ptr e, const AtomicLogger & logger, const std::string & start_of_message = "");
 
 std::string getExceptionMessage(const Exception & e, bool with_stacktrace, bool check_embedded_stacktrace = false);

src/Common/Logger.h

@@ -14,7 +14,7 @@ class Logger;
 using LoggerPtr = std::shared_ptr<Logger>;
 }
 
-using LoggerPtr = std::shared_ptr<Poco::Logger>;
+using LoggerPtr = Poco::LoggerPtr;
 using LoggerRawPtr = Poco::Logger *;
 
 /** RAII wrappers around Poco/Logger.h.

src/Common/MemorySpillScheduler.h

@@ -1,6 +1,7 @@
 #pragma once
 
 #include <atomic>
+#include <memory>
 #include <mutex>
 #include <unordered_map>
 #include <base/types.h>
@@ -21,6 +22,8 @@ struct ProcessorMemoryStats
 class MemorySpillScheduler
 {
 public:
+    using Ptr = std::shared_ptr<MemorySpillScheduler>;
+
     explicit MemorySpillScheduler(bool enable_ = false) : enable(enable_) {}
     ~MemorySpillScheduler() = default;
 

src/Common/ThreadStatus.cpp

@@ -107,7 +107,7 @@ static thread_local bool has_alt_stack = false;
 
 ThreadGroup::ThreadGroup()
     : master_thread_id(CurrentThread::get().thread_id)
-    , memory_spill_scheduler(false)
+    , memory_spill_scheduler(std::make_shared<MemorySpillScheduler>(false))
 {}
 
 ThreadStatus::ThreadStatus(bool check_current_thread_on_destruction_)

src/Common/ThreadStatus.h

@@ -69,6 +69,7 @@ class ThreadGroup
     ThreadGroup();
     using FatalErrorCallback = std::function<void()>;
     explicit ThreadGroup(ContextPtr query_context_, FatalErrorCallback fatal_error_callback_ = {});
+    explicit ThreadGroup(ThreadGroupPtr parent);
 
     /// The first thread created this thread group
     const UInt64 master_thread_id;
@@ -79,7 +80,7 @@ class ThreadGroup
 
     const FatalErrorCallback fatal_error_callback;
 
-    MemorySpillScheduler memory_spill_scheduler;
+    MemorySpillScheduler::Ptr memory_spill_scheduler;
     ProfileEvents::Counters performance_counters{VariableContext::Process};
     MemoryTracker memory_tracker{VariableContext::Process};
 
@@ -118,11 +119,14 @@ class ThreadGroup
 
     static ThreadGroupPtr createForBackgroundProcess(ContextPtr storage_context);
 
+    static ThreadGroupPtr createForMaterializedView();
+
     std::vector<UInt64> getInvolvedThreadIds() const;
     size_t getPeakThreadsUsage() const;
+    UInt64 getThreadsTotalElapsedMs() const;
 
     void linkThread(UInt64 thread_id);
-    void unlinkThread();
+    void unlinkThread(UInt64 elapsed_thread_counter_ms);
 
 private:
     mutable std::mutex mutex;
@@ -138,6 +142,8 @@ class ThreadGroup
 
     /// Peak threads count in the group
     size_t peak_threads_usage TSA_GUARDED_BY(mutex) = 0;
+
+    UInt64 elapsed_total_threads_counter_ms TSA_GUARDED_BY(mutex) = 0;
 };
 
 /**
@@ -240,10 +246,13 @@ class ThreadStatus : public boost::noncopyable
         UInt64 microseconds() const;
         UInt64 seconds() const;
 
+        UInt64 elapsedMilliseconds() const;
+        UInt64 elapsedMilliseconds(const TimePoint & current) const;
+
         std::chrono::time_point<std::chrono::system_clock> point;
     };
 
-    TimePoint query_start_time{};
+    TimePoint thread_attach_time{};
 
     // CPU and Real time query profilers
     std::unique_ptr<QueryProfilerReal> query_profiler_real;
@@ -255,9 +264,6 @@ class ThreadStatus : public boost::noncopyable
     Stopwatch stopwatch{CLOCK_MONOTONIC_COARSE};
     UInt64 last_performance_counters_update_time = 0;
 
-    /// See setInternalThread()
-    bool internal_thread = false;
-
     /// This is helpful for cut linking dependencies for clickhouse_common_io
     using Deleter = std::function<void()>;
     Deleter deleter;
@@ -277,22 +283,6 @@ class ThreadStatus : public boost::noncopyable
     ContextPtr getQueryContext() const;
     ContextPtr getGlobalContext() const;
 
-    /// "Internal" ThreadStatus is used for materialized views for separate
-    /// tracking into system.query_views_log
-    ///
-    /// You can have multiple internal threads, but only one non-internal with
-    /// the same thread_id.
-    ///
-    /// "Internal" thread:
-    /// - cannot have query profiler
-    ///   since the running (main query) thread should already have one
-    /// - should not try to obtain latest counter on detach
-    ///   because detaching of such threads will be done from a different
-    ///   thread_id, and some counters are not available (i.e. getrusage()),
-    ///   but anyway they are accounted correctly in the main ThreadStatus of a
-    ///   query.
-    void setInternalThread();
-
     /// Attaches slave thread to existing thread group
     void attachToGroup(const ThreadGroupPtr & thread_group_, bool check_detached = true);
 

src/Interpreters/CancellationChecker.cpp

@@ -62,7 +62,7 @@ bool CancellationChecker::removeQueryFromSet(std::shared_ptr<QueryStatus> query)
 
     if (it != querySet.end())
     {
-        LOG_TEST(log, "Removing query {} from done tasks", query->getInfo().query);
+        LOG_TEST(log, "Removing query {} from done tasks", query->getClientInfo().current_query_id);
         querySet.erase(it);
         return true;
     }
@@ -74,7 +74,7 @@ void CancellationChecker::appendTask(const std::shared_ptr<QueryStatus> & query,
 {
     if (timeout <= 0) // Avoid cases when the timeout is less or equal zero
     {
-        LOG_TEST(log, "Did not add the task because the timeout is 0. Query: {}", query->getInfo().query);
+        LOG_TEST(log, "Did not add the task because the timeout is 0, query_id: {}", query->getClientInfo().current_query_id);
         return;
     }
     std::unique_lock<std::mutex> lock(m);
@@ -122,10 +122,11 @@ void CancellationChecker::workerFunction()
 
             if ((end_time_ms <= now_ms && duration_milliseconds.count() != 0))
             {
-                LOG_TRACE(log, "Cancelling the task because of the timeout: {} ms, query: {}", duration, next_task.query->getInfo().query);
+                LOG_DEBUG(log, "Cancelling the task because of the timeout: {} ms, query_id: {}",
+                    duration, next_task.query->getClientInfo().current_query_id);
+
                 cancelTask(next_task);
                 querySet.erase(next_task);
-
                 continue;
             }
         }

src/Interpreters/Context.cpp

@@ -112,6 +112,7 @@
 #include <Common/logger_useful.h>
 #include <Common/RemoteHostFilter.h>
 #include <Common/HTTPHeaderFilter.h>
+#include <Interpreters/StorageID.h>
 #include <Interpreters/SystemLog.h>
 #include <Interpreters/InterpreterSelectQueryAnalyzer.h>
 #include <Interpreters/AsynchronousInsertQueue.h>
@@ -2172,6 +2173,12 @@ bool Context::hasScalar(const String & name) const
     return scalars.contains(name);
 }
 
+void Context::addQueryAccessInfo(
+    const StorageID & table_id,
+    const Names & column_names)
+{
+    addQueryAccessInfo(backQuoteIfNeed(table_id.getDatabaseName()), table_id.getFullTableName(), column_names);
+}
 
 void Context::addQueryAccessInfo(
     const String & quoted_database_name,