Revert "Merge pull request ClickHouse#77309 from ClickHouse/chesema-deduplicate-sync-inserts"

This reverts commit 120768e, reversing
changes made to 0d42a5c.

Author: nikitamikhaylov

src/Client/LocalConnection.cpp

@@ -466,23 +466,6 @@ 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;
@@ -531,7 +514,7 @@ bool LocalConnection::poll(size_t)
     {
         state->sent_profile_events = true;
 
-        if (send_profile_events && (state->executor || state->pushing_async_executor || state->pushing_executor))
+        if (send_profile_events && state->executor)
         {
             sendProfileEvents();
             return true;

src/Common/ErrorCodes.cpp

@@ -625,7 +625,6 @@
     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, LogsLevel level)
+void tryLogException(std::exception_ptr e, LoggerPtr logger, const std::string & start_of_message)
 {
     try
     {
         std::rethrow_exception(std::move(e)); // NOLINT
     }
     catch (...)
     {
-        tryLogCurrentException(logger, start_of_message, level);
+        tryLogCurrentException(logger, start_of_message);
     }
 }
 

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 = "", LogsLevel level = LogsLevel::error);
+void tryLogException(std::exception_ptr e, LoggerPtr logger, const std::string & start_of_message = "");
 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 = Poco::LoggerPtr;
+using LoggerPtr = std::shared_ptr<Poco::Logger>;
 using LoggerRawPtr = Poco::Logger *;
 
 /** RAII wrappers around Poco/Logger.h.

src/Common/MemorySpillScheduler.h

@@ -1,7 +1,6 @@
 #pragma once
 
 #include <atomic>
-#include <memory>
 #include <mutex>
 #include <unordered_map>
 #include <base/types.h>
@@ -22,8 +21,6 @@ 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(std::make_shared<MemorySpillScheduler>(false))
+    , memory_spill_scheduler(false)
 {}
 
 ThreadStatus::ThreadStatus(bool check_current_thread_on_destruction_)

src/Common/ThreadStatus.h

@@ -69,7 +69,6 @@ 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;
@@ -80,7 +79,7 @@ class ThreadGroup
 
     const FatalErrorCallback fatal_error_callback;
 
-    MemorySpillScheduler::Ptr memory_spill_scheduler;
+    MemorySpillScheduler memory_spill_scheduler;
     ProfileEvents::Counters performance_counters{VariableContext::Process};
     MemoryTracker memory_tracker{VariableContext::Process};
 
@@ -119,14 +118,11 @@ 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(UInt64 elapsed_thread_counter_ms);
+    void unlinkThread();
 
 private:
     mutable std::mutex mutex;
@@ -142,8 +138,6 @@ 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;
 };
 
 /**
@@ -246,13 +240,10 @@ 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 thread_attach_time{};
+    TimePoint query_start_time{};
 
     // CPU and Real time query profilers
     std::unique_ptr<QueryProfilerReal> query_profiler_real;
@@ -264,6 +255,9 @@ 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;
@@ -283,6 +277,22 @@ 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->getClientInfo().current_query_id);
+        LOG_TEST(log, "Removing query {} from done tasks", query->getInfo().query);
         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_id: {}", query->getClientInfo().current_query_id);
+        LOG_TEST(log, "Did not add the task because the timeout is 0. Query: {}", query->getInfo().query);
         return;
     }
     std::unique_lock<std::mutex> lock(m);
@@ -122,11 +122,10 @@ void CancellationChecker::workerFunction()
 
             if ((end_time_ms <= now_ms && duration_milliseconds.count() != 0))
             {
-                LOG_DEBUG(log, "Cancelling the task because of the timeout: {} ms, query_id: {}",
-                    duration, next_task.query->getClientInfo().current_query_id);
-
+                LOG_TRACE(log, "Cancelling the task because of the timeout: {} ms, query: {}", duration, next_task.query->getInfo().query);
                 cancelTask(next_task);
                 querySet.erase(next_task);
+
                 continue;
             }
         }

src/Interpreters/Context.cpp

@@ -112,7 +112,6 @@
 #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>
@@ -2173,12 +2172,6 @@ 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,