Merge pull request ClickHouse#77578 from azat/fix-applying-async_insert-from-server

Fix applying async_insert from server (via apply_settings_from_server)

Author: azat

programs/client/Client.cpp

@@ -71,7 +71,7 @@ Client::Client()
 
 Client::~Client() = default;
 
-void Client::processError(const String & query) const
+void Client::processError(std::string_view query) const
 {
     if (server_exception)
     {

programs/client/Client.h

@@ -27,13 +27,13 @@ class Client : public ClientApplicationBase
 protected:
     Poco::Util::LayeredConfiguration & getClientConfiguration() override;
 
-    bool processWithFuzzing(const String & full_query) override;
+    bool processWithFuzzing(std::string_view full_query) override;
     bool buzzHouse() override;
     std::optional<bool> processFuzzingStep(const String & query_to_execute, const ASTPtr & parsed_query, bool permissive);
 
     void connect() override;
 
-    void processError(const String & query) const override;
+    void processError(std::string_view query) const override;
 
     String getName() const override { return "client"; }
 

programs/client/FuzzLoop.cpp

@@ -47,7 +47,8 @@ extern const int BUZZHOUSE;
 
 std::optional<bool> Client::processFuzzingStep(const String & query_to_execute, const ASTPtr & parsed_query, const bool permissive)
 {
-    processParsedSingleQuery(query_to_execute, query_to_execute, parsed_query);
+    bool async_insert = false;
+    processParsedSingleQuery(query_to_execute, parsed_query, async_insert);
 
     const auto * exception = server_exception ? server_exception.get() : client_exception.get();
     // Sometimes you may get TOO_DEEP_RECURSION from the server,
@@ -105,7 +106,7 @@ std::optional<bool> Client::processFuzzingStep(const String & query_to_execute,
 }
 
 /// Returns false when server is not available.
-bool Client::processWithFuzzing(const String & full_query)
+bool Client::processWithFuzzing(std::string_view full_query)
 {
     ASTPtr orig_ast;
 

programs/local/LocalServer.cpp

@@ -150,7 +150,7 @@ Poco::Util::LayeredConfiguration & LocalServer::getClientConfiguration()
     return config();
 }
 
-void LocalServer::processError(const String &) const
+void LocalServer::processError(std::string_view) const
 {
     if (ignore_error)
         return;

programs/local/LocalServer.h

@@ -34,7 +34,7 @@ class LocalServer : public ClientApplicationBase, public Loggers
 
     void connect() override;
 
-    void processError(const String & query) const override;
+    void processError(std::string_view query) const override;
 
     String getName() const override { return "local"; }
 

programs/server/users.xml

@@ -5,6 +5,7 @@
     <profiles>
         <!-- Default settings. -->
         <default>
+            <!-- <async_insert>1</async_insert> -->
         </default>
 
         <!-- Profile that allows only read queries. -->

src/Client/ClientBase.cpp

@@ -1091,18 +1091,6 @@ void ClientBase::updateSuggest(const ASTPtr & ast)
         suggest->addWords(std::move(new_words));
 }
 
-bool ClientBase::isSyncInsertWithData(const ASTInsertQuery & insert_query, const ContextPtr & context)
-{
-    if (!insert_query.data)
-        return false;
-
-    auto settings = context->getSettingsCopy();
-    if (insert_query.settings_ast)
-        settings.applyChanges(insert_query.settings_ast->as<ASTSetQuery>()->changes);
-
-    return !settings[Setting::async_insert];
-}
-
 bool ClientBase::processTextAsSingleQuery(const String & full_query)
 {
     /// Some parts of a query (result output and formatting) are executed
@@ -1115,8 +1103,6 @@ bool ClientBase::processTextAsSingleQuery(const String & full_query)
     if (!parsed_query)
         return false;
 
-    String query_to_execute;
-
     /// Query will be parsed before checking the result because error does not
     /// always means a problem, i.e. if table already exists, and it is no a
     /// huge problem if suggestion will be added even on error, since this is
@@ -1127,19 +1113,10 @@ bool ClientBase::processTextAsSingleQuery(const String & full_query)
     if (suggest)
         updateSuggest(parsed_query);
 
-    /// An INSERT query may have the data that follows query text.
-    /// Send part of the query without data, because data will be sent separately.
-    /// But for asynchronous inserts we don't extract data, because it's needed
-    /// to be done on server side in that case (for coalescing the data from multiple inserts on server side).
-    const auto * insert = parsed_query->as<ASTInsertQuery>();
-    if (insert && isSyncInsertWithData(*insert, client_context))
-        query_to_execute = full_query.substr(0, insert->data - full_query.data());
-    else
-        query_to_execute = full_query;
-
     try
     {
-        processParsedSingleQuery(full_query, query_to_execute, parsed_query, echo_queries);
+        bool is_async_insert_with_inlined_data = false;
+        processParsedSingleQuery(full_query, parsed_query, is_async_insert_with_inlined_data);
     }
     catch (Exception & e)
     {
@@ -1155,10 +1132,8 @@ bool ClientBase::processTextAsSingleQuery(const String & full_query)
     return !have_error;
 }
 
-void ClientBase::processOrdinaryQuery(const String & query_to_execute, ASTPtr parsed_query)
+void ClientBase::processOrdinaryQuery(String query, ASTPtr parsed_query)
 {
-    auto query = query_to_execute;
-
     /// Rewrite query only when we have query parameters.
     /// Note that if query is rewritten, comments in query are lost.
     /// But the user often wants to see comments in server logs, query log, processlist, etc.
@@ -1760,9 +1735,8 @@ bool isStdinNotEmptyAndValid(ReadBuffer & std_in)
 }
 
 
-void ClientBase::processInsertQuery(const String & query_to_execute, ASTPtr parsed_query)
+void ClientBase::processInsertQuery(String query, ASTPtr parsed_query)
 {
-    auto query = query_to_execute;
     if (!query_parameters.empty()
         && connection->getServerRevision(connection_parameters.timeouts) < DBMS_MIN_PROTOCOL_VERSION_WITH_PARAMETERS)
     {
@@ -2173,8 +2147,11 @@ void ClientBase::cancelQuery()
     cancelled = true;
 }
 
-void ClientBase::processParsedSingleQuery(const String & full_query, const String & query_to_execute,
-        ASTPtr parsed_query, std::optional<bool> echo_query_, bool report_error)
+void ClientBase::processParsedSingleQuery(
+    std::string_view query_,
+    ASTPtr parsed_query,
+    bool & is_async_insert_with_inlined_data,
+    size_t insert_query_without_data_length)
 {
     resetOutput();
     have_error = false;
@@ -2183,13 +2160,6 @@ void ClientBase::processParsedSingleQuery(const String & full_query, const Strin
     client_exception.reset();
     server_exception.reset();
 
-    if (echo_query_ && *echo_query_)
-    {
-        writeString(full_query, *std_out);
-        writeChar('\n', *std_out);
-        std_out->next();
-    }
-
     if (is_interactive)
     {
         client_context->setCurrentQueryId("");
@@ -2258,7 +2228,8 @@ void ClientBase::processParsedSingleQuery(const String & full_query, const Strin
         if (insert && insert->select)
             insert->tryFindInputFunction(input_function);
 
-        bool is_async_insert_with_inlined_data = client_context->getSettingsRef()[Setting::async_insert] && insert && insert->hasInlinedData();
+        /// Update async_insert after applying settings from server
+        is_async_insert_with_inlined_data = client_context->getSettingsRef()[Setting::async_insert] && insert && insert->hasInlinedData();
 
         if (is_async_insert_with_inlined_data)
         {
@@ -2270,16 +2241,26 @@ void ClientBase::processParsedSingleQuery(const String & full_query, const Strin
                     "Processing async inserts with both inlined and external data (from stdin or infile) is not supported");
         }
 
+        String query;
+        /// An INSERT query may have the data that follows query text.
+        /// Send part of the query without data, because data will be sent separately.
+        /// But for asynchronous inserts we don't extract data, because it's needed
+        /// to be done on server side in that case (for coalescing the data from multiple inserts on server side).
+        if (insert && insert->data && !is_async_insert_with_inlined_data && insert_query_without_data_length)
+            query = query_.substr(0, insert_query_without_data_length);
+        else
+            query = query_;
+
         /// INSERT query for which data transfer is needed (not an INSERT SELECT or input()) is processed separately.
         if (insert && (!insert->select || input_function) && (!is_async_insert_with_inlined_data || input_function))
         {
             if (input_function && insert->format.empty())
                 throw Exception(ErrorCodes::INVALID_USAGE_OF_INPUT, "FORMAT must be specified for function input()");
 
-            processInsertQuery(query_to_execute, parsed_query);
+            processInsertQuery(query, parsed_query);
         }
         else
-            processOrdinaryQuery(query_to_execute, parsed_query);
+            processOrdinaryQuery(query, parsed_query);
     }
 
     /// Do not change context (current DB, settings) in case of an exception.
@@ -2371,15 +2352,12 @@ void ClientBase::processParsedSingleQuery(const String & full_query, const Strin
     {
         output_stream << "Processed rows: " << processed_rows << "\n";
     }
-
-    if (have_error && report_error)
-        processError(full_query);
 }
 
 
 MultiQueryProcessingStage ClientBase::analyzeMultiQueryText(
     const char *& this_query_begin, const char *& this_query_end, const char * all_queries_end,
-    String & query_to_execute, ASTPtr & parsed_query, const String & all_queries_text,
+    ASTPtr & parsed_query,
     std::unique_ptr<Exception> & current_exception)
 {
     if (!is_interactive && cancelled)
@@ -2458,7 +2436,6 @@ MultiQueryProcessingStage ClientBase::analyzeMultiQueryText(
             insert_ast = explain_ast->getExplainedQuery()->as<ASTInsertQuery>();
         }
     }
-    const char * query_to_execute_end = this_query_end;
     if (insert_ast && insert_ast->data)
     {
         if (insert_ast->format == "Values")
@@ -2495,18 +2472,8 @@ MultiQueryProcessingStage ClientBase::analyzeMultiQueryText(
                 this_query_end = all_queries_end;
         }
         insert_ast->end = this_query_end;
-        query_to_execute_end = isSyncInsertWithData(*insert_ast, client_context) ? insert_ast->data : this_query_end;
     }
 
-    query_to_execute = all_queries_text.substr(this_query_begin - all_queries_text.data(), query_to_execute_end - this_query_begin);
-
-    // Try to include the trailing comment with test hints. It is just
-    // a guess for now, because we don't yet know where the query ends
-    // if it is an INSERT query with inline data. We will do it again
-    // after we have processed the query. But even this guess is
-    // beneficial so that we see proper trailing comments in "echo" and
-    // server log.
-    adjustQueryEnd(this_query_end, all_queries_end, max_parser_depth, max_parser_backtracks);
     return MultiQueryProcessingStage::EXECUTE_QUERY;
 }
 
@@ -2546,8 +2513,6 @@ bool ClientBase::executeMultiQuery(const String & all_queries_text)
     UInt32 script_query_number = 0;
     UInt32 script_line_number = 0;
 
-    String full_query; // full_query is the query + inline INSERT data + trailing comments (the latter is our best guess for now).
-    String query_to_execute;
     ASTPtr parsed_query;
     std::unique_ptr<Exception> current_exception;
     size_t retries_count = 0;
@@ -2556,7 +2521,7 @@ bool ClientBase::executeMultiQuery(const String & all_queries_text)
     while (true)
     {
         auto stage = analyzeMultiQueryText(this_query_begin, this_query_end, all_queries_end,
-                                           query_to_execute, parsed_query, all_queries_text, current_exception);
+                                           parsed_query, current_exception);
         switch (stage)
         {
             case MultiQueryProcessingStage::QUERIES_END:
@@ -2611,7 +2576,26 @@ bool ClientBase::executeMultiQuery(const String & all_queries_text)
             case MultiQueryProcessingStage::EXECUTE_QUERY:
             {
                 is_first = false;
-                full_query = all_queries_text.substr(this_query_begin - all_queries_text.data(), this_query_end - this_query_begin);
+
+                // Save query without trailing comment
+                auto query_to_execute = std::string_view(all_queries_text).substr(this_query_begin - all_queries_text.data(), this_query_end - this_query_begin);
+                size_t insert_query_without_data_length = 0;
+                if (const auto * insert = parsed_query->as<ASTInsertQuery>())
+                    insert_query_without_data_length = insert->data - query_to_execute.data();
+
+                // Try to include the trailing comment with test hints. It is just
+                // a guess for now, because we don't yet know where the query ends
+                // if it is an INSERT query with inline data. We will do it again
+                // after we have processed the query. But even this guess is
+                // beneficial so that we see proper trailing comments in "echo" and
+                // server log.
+                {
+                    unsigned max_parser_depth = static_cast<unsigned>(client_context->getSettingsRef()[Setting::max_parser_depth]);
+                    unsigned max_parser_backtracks = static_cast<unsigned>(client_context->getSettingsRef()[Setting::max_parser_backtracks]);
+                    adjustQueryEnd(this_query_end, all_queries_end, max_parser_depth, max_parser_backtracks);
+                }
+                // query + inline INSERT data + trailing comments (the latter is our best guess for now).
+                auto full_query = std::string_view(all_queries_text).substr(this_query_begin - all_queries_text.data(), this_query_end - this_query_begin);
 
                 ++script_query_number;
                 script_line_number += std::count(prev_query_begin, this_query_begin, '\n');
@@ -2637,10 +2621,21 @@ bool ClientBase::executeMultiQuery(const String & all_queries_text)
 
                 // Echo all queries if asked; makes for a more readable reference file.
                 echo_query = test_hint.echoQueries().value_or(echo_query);
+                bool is_async_insert_with_inlined_data = false;
+
+                if (echo_query)
+                {
+                    writeString(full_query, *std_out);
+                    writeChar('\n', *std_out);
+                    std_out->next();
+                }
 
                 try
                 {
-                    processParsedSingleQuery(full_query, query_to_execute, parsed_query, echo_query, false);
+                    processParsedSingleQuery(query_to_execute,
+                        parsed_query,
+                        is_async_insert_with_inlined_data,
+                        insert_query_without_data_length);
                 }
                 catch (...)
                 {
@@ -2773,7 +2768,7 @@ bool ClientBase::executeMultiQuery(const String & all_queries_text)
                 // , where the inline data is delimited by semicolon and not by a
                 // newline.
                 auto * insert_ast = parsed_query->as<ASTInsertQuery>();
-                if (insert_ast && isSyncInsertWithData(*insert_ast, client_context))
+                if (insert_ast && insert_ast->data && !is_async_insert_with_inlined_data)
                 {
                     this_query_end = insert_ast->end;
                     adjustQueryEnd(
@@ -2891,6 +2886,9 @@ bool ClientBase::addMergeTreeSettings(ASTCreateQuery & ast_create)
 void ClientBase::applySettingsFromServerIfNeeded()
 {
     const Settings & settings = client_context->getSettingsRef();
+    if (!settings[Setting::apply_settings_from_server])
+        return;
+
     SettingsChanges changes_to_apply;
     for (const SettingChange & change : settings_from_server)
     {
@@ -2900,8 +2898,7 @@ void ClientBase::applySettingsFromServerIfNeeded()
             changes_to_apply.push_back(change);
     }
 
-    if (settings[Setting::apply_settings_from_server])
-        global_context->applySettingsChanges(changes_to_apply);
+    global_context->applySettingsChanges(changes_to_apply);
 }
 
 void ClientBase::startKeystrokeInterceptorIfExists()

src/Client/ClientBase.h

@@ -115,7 +115,7 @@ class ClientBase
     /// This is the analogue of Poco::Application::config()
     virtual Poco::Util::LayeredConfiguration & getClientConfiguration() = 0;
 
-    virtual bool processWithFuzzing(const String &)
+    virtual bool processWithFuzzing(std::string_view)
     {
         throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Query processing with fuzzing is not implemented");
     }
@@ -126,14 +126,18 @@ class ClientBase
     }
 
     virtual void connect() = 0;
-    virtual void processError(const String & query) const = 0;
+    virtual void processError(std::string_view query) const = 0;
     virtual String getName() const = 0;
 
-    void processOrdinaryQuery(const String & query_to_execute, ASTPtr parsed_query);
-    void processInsertQuery(const String & query_to_execute, ASTPtr parsed_query);
+    void processOrdinaryQuery(String query, ASTPtr parsed_query);
+    void processInsertQuery(String query, ASTPtr parsed_query);
 
-    void processParsedSingleQuery(const String & full_query, const String & query_to_execute,
-        ASTPtr parsed_query, std::optional<bool> echo_query_ = {}, bool report_error = false);
+    void processParsedSingleQuery(
+        std::string_view query_,
+        ASTPtr parsed_query,
+        bool & is_async_insert_with_inlined_data,
+        // to handle INSERT w/o async_insert
+        size_t insert_query_without_data_length = 0);
 
     static void adjustQueryEnd(const char *& this_query_end, const char * all_queries_end, uint32_t max_parser_depth, uint32_t max_parser_backtracks);
     virtual void setupSignalHandler() = 0;
@@ -143,7 +147,7 @@ class ClientBase
     bool executeMultiQuery(const String & all_queries_text);
     MultiQueryProcessingStage analyzeMultiQueryText(
         const char *& this_query_begin, const char *& this_query_end, const char * all_queries_end,
-        String & query_to_execute, ASTPtr & parsed_query, const String & all_queries_text,
+        ASTPtr & parsed_query,
         std::unique_ptr<Exception> & current_exception);
 
     void clearTerminal();