diff --git a/programs/client/CMakeLists.txt b/programs/client/CMakeLists.txt
index 084e1b45911..6888ec141c9 100644
--- a/programs/client/CMakeLists.txt
+++ b/programs/client/CMakeLists.txt
@@ -1,8 +1,6 @@
 set (CLICKHOUSE_CLIENT_SOURCES
     Client.cpp
-    ConnectionParameters.cpp
     QueryFuzzer.cpp
-    Suggest.cpp
 )
 
 set (CLICKHOUSE_CLIENT_LINK
diff --git a/programs/client/Client.cpp b/programs/client/Client.cpp
index 98d932a6044..6cc891f7761 100644
--- a/programs/client/Client.cpp
+++ b/programs/client/Client.cpp
@@ -1,6 +1,4 @@
-#include "ConnectionParameters.h"
 #include "QueryFuzzer.h"
-#include "Suggest.h"
 #include "TestHint.h"
 
 #include <stdlib.h>
@@ -17,7 +15,6 @@
 #include <boost/program_options.hpp>
 #include <boost/algorithm/string/replace.hpp>
 #include <Poco/String.h>
-//#include <Poco/Util/Application.h>
 #if !defined(ARCADIA_BUILD)
 #    include <Common/config_version.h>
 #endif
@@ -80,7 +77,6 @@
 #include <common/argsToConfig.h>
 //#include <Common/TerminalSize.h>
 #include <Common/UTF8Helpers.h>
-#include <Common/ProgressIndication.h>
 #include <filesystem>
 #include <Common/filesystemHelpers.h>
 
@@ -141,8 +137,6 @@ public:
     Client() = default;
 
 private:
-    /// If not empty, queries will be read from these files
-    std::vector<std::string> queries_files;
     /// If not empty, run queries from these files before processing every file from 'queries_files'.
     std::vector<std::string> interleave_queries_files;
 
@@ -161,8 +155,6 @@ private:
     size_t insert_format_max_block_size = 0; /// Max block size when reading INSERT data.
     size_t max_client_network_bandwidth = 0; /// The maximum speed of data exchange over the network for the client in bytes per second.
 
-    bool has_vertical_output_suffix = false; /// Is \G present at the end of the query string?
-
     SharedContextHolder shared_context = Context::createShared();
     ContextMutablePtr context = Context::createGlobal(shared_context.get());
 
@@ -182,15 +174,8 @@ private:
     String server_logs_file;
     BlockOutputStreamPtr logs_out_stream;
 
-    String home_path;
-
     String current_profile;
 
-    String prompt_by_server_display_name;
-
-    /// Path to a file containing command history.
-    String history_file;
-
     /// How many rows have been read or written.
     size_t processed_rows = 0;
 
@@ -209,15 +194,6 @@ private:
 
     UInt64 server_revision = 0;
     String server_version;
-    String server_display_name;
-
-    /// true by default - for interactive mode, might be changed when --progress option is checked for
-    /// non-interactive mode.
-    bool need_render_progress = true;
-
-    bool written_first_block = false;
-
-    ProgressIndication progress_indication;
 
     /// External tables info.
     std::list<ExternalTable> external_tables;
@@ -230,12 +206,12 @@ private:
     QueryFuzzer fuzzer;
     int query_fuzzer_runs = 0;
 
-    std::optional<Suggest> suggest;
-
     /// We will format query_id in interactive mode in various ways, the default is just to print Query id: ...
     std::vector<std::pair<String, String>> query_id_formats;
     QueryProcessingStage::Enum query_processing_stage;
 
+    bool supportPasswordOption() override { return true; }
+
     void initialize(Poco::Util::Application & self) override
     {
         Poco::Util::Application::initialize(self);
@@ -273,55 +249,32 @@ private:
             query_id_formats.emplace_back("Query id:", " {query_id}\n");
     }
 
-
-    int main(const std::vector<std::string> & /*args*/) override
+    void processMainImplException(const Exception & e) override
     {
-        try
+        bool print_stack_trace = config().getBool("stacktrace", false);
+
+        std::string text = e.displayText();
+
+        /** If exception is received from server, then stack trace is embedded in message.
+            * If exception is thrown on client, then stack trace is in separate field.
+            */
+
+        auto embedded_stack_trace_pos = text.find("Stack trace");
+        if (std::string::npos != embedded_stack_trace_pos && !print_stack_trace)
+            text.resize(embedded_stack_trace_pos);
+
+        std::cerr << "Code: " << e.code() << ". " << text << std::endl << std::endl;
+
+        /// Don't print the stack trace on the client if it was logged on the server.
+        /// Also don't print the stack trace in case of network errors.
+        if (print_stack_trace && e.code() != ErrorCodes::NETWORK_ERROR && std::string::npos == embedded_stack_trace_pos)
         {
-            return mainImpl();
-        }
-        catch (const Exception & e)
-        {
-            bool print_stack_trace = config().getBool("stacktrace", false);
-
-            std::string text = e.displayText();
-
-            /** If exception is received from server, then stack trace is embedded in message.
-              * If exception is thrown on client, then stack trace is in separate field.
-              */
-
-            auto embedded_stack_trace_pos = text.find("Stack trace");
-            if (std::string::npos != embedded_stack_trace_pos && !print_stack_trace)
-                text.resize(embedded_stack_trace_pos);
-
-            std::cerr << "Code: " << e.code() << ". " << text << std::endl << std::endl;
-
-            /// Don't print the stack trace on the client if it was logged on the server.
-            /// Also don't print the stack trace in case of network errors.
-            if (print_stack_trace && e.code() != ErrorCodes::NETWORK_ERROR && std::string::npos == embedded_stack_trace_pos)
-            {
-                std::cerr << "Stack trace:" << std::endl << e.getStackTraceString();
-            }
-
-            /// If exception code isn't zero, we should return non-zero return code anyway.
-            return e.code() ? e.code() : -1;
-        }
-        catch (...)
-        {
-            std::cerr << getCurrentExceptionMessage(false) << std::endl;
-            return getCurrentExceptionCode();
+            std::cerr << "Stack trace:" << std::endl << e.getStackTraceString();
         }
     }
 
-
-    int mainImpl()
+    bool isInteractive() override
     {
-        UseSSL use_ssl;
-
-        registerFormats();
-        registerFunctions();
-        registerAggregateFunctions();
-
         /// Batch mode is enabled if one of the following is true:
         /// - -e (--query) command line option is present.
         ///   The value of the option is used as the text of query (or of multiple queries).
@@ -329,168 +282,67 @@ private:
         /// - stdin is not a terminal. In this case queries are read from it.
         /// - -qf (--queries-file) command line option is present.
         ///   The value of the option is used as file with query (or of multiple queries) to execute.
-        if (!stdin_is_a_tty || config().has("query") || !queries_files.empty())
-            is_interactive = false;
 
-        if (config().has("query") && !queries_files.empty())
+        return stdin_is_a_tty && !config().has("query") && queries_files.empty();
+    }
+
+    void loadSuggestionDataIfPossible() override
+    {
+        if (server_revision >= Suggest::MIN_SERVER_REVISION && !config().getBool("disable_suggestion", false))
         {
-            throw Exception("Specify either `query` or `queries-file` option", ErrorCodes::BAD_ARGUMENTS);
+            /// Load suggestion data from the server.
+            suggest->load(connection_parameters, config().getInt("suggestion_limit"));
+        }
+    }
+
+    bool processQueryFromInteractive(const String & input) override
+    {
+        try
+        {
+            return processQueryText(input);
+        }
+        catch (const Exception & e)
+        {
+            // We don't need to handle the test hints in the interactive
+            // mode.
+            std::cerr << std::endl
+                        << "Exception on client:" << std::endl
+                        << "Code: " << e.code() << ". " << e.displayText() << std::endl;
+
+            if (config().getBool("stacktrace", false))
+                std::cerr << "Stack trace:" << std::endl << e.getStackTraceString() << std::endl;
+
+            std::cerr << std::endl;
+
+            client_exception = std::make_unique<Exception>(e);
         }
 
-        std::cout << std::fixed << std::setprecision(3);
-        std::cerr << std::fixed << std::setprecision(3);
-
-        if (is_interactive)
+        if (client_exception)
         {
-            clearTerminal();
-            showClientVersion();
+            /// client_exception may have been set above or elsewhere.
+            /// Client-side exception during query execution can result in the loss of
+            /// sync in the connection protocol.
+            /// So we reconnect and allow to enter the next query.
+            connect();
         }
 
-        is_default_format = !config().has("vertical") && !config().has("format");
-        if (config().has("vertical"))
-            format = config().getString("format", "Vertical");
-        else
-            format = config().getString("format", is_interactive ? "PrettyCompact" : "TabSeparated");
+        /// Continue processing queries.
+        return true;
+    }
 
-        format_max_block_size = config().getInt("format_max_block_size", context->getSettingsRef().max_block_size);
+    int childMainImpl() override
+    {
+        UseSSL use_ssl;
 
-        insert_format = "Values";
-
-        /// Setting value from cmd arg overrides one from config
-        if (context->getSettingsRef().max_insert_block_size.changed)
-            insert_format_max_block_size = context->getSettingsRef().max_insert_block_size;
-        else
-            insert_format_max_block_size = config().getInt("insert_format_max_block_size", context->getSettingsRef().max_insert_block_size);
-
-        if (!is_interactive)
-        {
-            need_render_progress = config().getBool("progress", false);
-            echo_queries = config().getBool("echo", false);
-            ignore_error = config().getBool("ignore-error", false);
-        }
-
-        ClientInfo & client_info = context->getClientInfo();
-        client_info.setInitialQuery();
-        client_info.quota_key = config().getString("quota_key", "");
+        registerFormats();
+        registerFunctions();
+        registerAggregateFunctions();
 
         connect();
 
-        /// Initialize DateLUT here to avoid counting time spent here as query execution time.
-        const auto local_tz = DateLUT::instance().getTimeZone();
-
         if (is_interactive)
         {
-            if (config().has("query_id"))
-                throw Exception("query_id could be specified only in non-interactive mode", ErrorCodes::BAD_ARGUMENTS);
-            if (print_time_to_stderr)
-                throw Exception("time option could be specified only in non-interactive mode", ErrorCodes::BAD_ARGUMENTS);
-
-            suggest.emplace();
-            if (server_revision >= Suggest::MIN_SERVER_REVISION && !config().getBool("disable_suggestion", false))
-            {
-                /// Load suggestion data from the server.
-                suggest->load(connection_parameters, config().getInt("suggestion_limit"));
-            }
-
-            /// Load command history if present.
-            if (config().has("history_file"))
-                history_file = config().getString("history_file");
-            else
-            {
-                auto * history_file_from_env = getenv("CLICKHOUSE_HISTORY_FILE");
-                if (history_file_from_env)
-                    history_file = history_file_from_env;
-                else if (!home_path.empty())
-                    history_file = home_path + "/.clickhouse-client-history";
-            }
-
-            if (!history_file.empty() && !fs::exists(history_file))
-            {
-                /// Avoid TOCTOU issue.
-                try
-                {
-                    FS::createFile(history_file);
-                }
-                catch (const ErrnoException & e)
-                {
-                    if (e.getErrno() != EEXIST)
-                        throw;
-                }
-            }
-
-            LineReader::Patterns query_extenders = {"\\"};
-            LineReader::Patterns query_delimiters = {";", "\\G"};
-
-#if USE_REPLXX
-            replxx::Replxx::highlighter_callback_t highlight_callback{};
-            if (config().getBool("highlight"))
-                highlight_callback = highlight;
-
-            ReplxxLineReader lr(*suggest, history_file, config().has("multiline"), query_extenders, query_delimiters, highlight_callback);
-
-#elif defined(USE_READLINE) && USE_READLINE
-            ReadlineLineReader lr(*suggest, history_file, config().has("multiline"), query_extenders, query_delimiters);
-#else
-            LineReader lr(history_file, config().has("multiline"), query_extenders, query_delimiters);
-#endif
-
-            /// Enable bracketed-paste-mode only when multiquery is enabled and multiline is
-            ///  disabled, so that we are able to paste and execute multiline queries in a whole
-            ///  instead of erroring out, while be less intrusive.
-            if (config().has("multiquery") && !config().has("multiline"))
-                lr.enableBracketedPaste();
-
-            do
-            {
-                auto input = lr.readLine(prompt(), ":-] ");
-                if (input.empty())
-                    break;
-
-                has_vertical_output_suffix = false;
-                if (input.ends_with("\\G"))
-                {
-                    input.resize(input.size() - 2);
-                    has_vertical_output_suffix = true;
-                }
-
-                try
-                {
-                    if (!processQueryText(input))
-                        break;
-                }
-                catch (const Exception & e)
-                {
-                    // We don't need to handle the test hints in the interactive
-                    // mode.
-                    std::cerr << std::endl
-                              << "Exception on client:" << std::endl
-                              << "Code: " << e.code() << ". " << e.displayText() << std::endl;
-
-                    if (config().getBool("stacktrace", false))
-                        std::cerr << "Stack trace:" << std::endl << e.getStackTraceString() << std::endl;
-
-                    std::cerr << std::endl;
-
-                    client_exception = std::make_unique<Exception>(e);
-                }
-
-                if (client_exception)
-                {
-                    /// client_exception may have been set above or elsewhere.
-                    /// Client-side exception during query execution can result in the loss of
-                    /// sync in the connection protocol.
-                    /// So we reconnect and allow to enter the next query.
-                    connect();
-                }
-            } while (true);
-
-            if (isNewYearMode())
-                std::cout << "Happy new year." << std::endl;
-            else if (isChineseNewYearMode(local_tz))
-                std::cout << "Happy Chinese new year. 春节快乐!" << std::endl;
-            else
-                std::cout << "Bye." << std::endl;
-            return 0;
+            runInteractive();
         }
         else
         {
@@ -513,11 +365,10 @@ private:
                 // case so that at least we don't lose an error.
                 return -1;
             }
-
-            return 0;
         }
-    }
 
+        return 0;
+    }
 
     void connect()
     {
@@ -611,12 +462,10 @@ private:
             }
         }
 
-        Strings keys;
-
         prompt_by_server_display_name = config().getRawString("prompt_by_server_display_name.default", "{display_name} :) ");
 
+        Strings keys;
         config().keys("prompt_by_server_display_name", keys);
-
         for (const String & key : keys)
         {
             if (key != "default" && server_display_name.find(key) != std::string::npos)
@@ -647,13 +496,6 @@ private:
             boost::replace_all(prompt_by_server_display_name, "{" + key + "}", value);
     }
 
-
-    inline String prompt() const
-    {
-        return boost::replace_all_copy(prompt_by_server_display_name, "{database}", config().getString("database", "default"));
-    }
-
-
     void nonInteractive()
     {
         String text;
@@ -2192,11 +2034,6 @@ private:
         }
     }
 
-    static void showClientVersion()
-    {
-        std::cout << DBMS_NAME << " client version " << VERSION_STRING << VERSION_OFFICIAL << "." << std::endl;
-    }
-
 public:
     void readArguments(int argc, char ** argv, Arguments & common_arguments, std::vector<Arguments> & external_tables_arguments) override
     {
@@ -2508,6 +2345,26 @@ public:
         {
             context->getClientInfo().client_trace_context.tracestate = options["opentelemetry-tracestate"].as<std::string>();
         }
+
+        is_default_format = !config().has("vertical") && !config().has("format");
+        if (config().has("vertical"))
+            format = config().getString("format", "Vertical");
+        else
+            format = config().getString("format", is_interactive ? "PrettyCompact" : "TabSeparated");
+
+        format_max_block_size = config().getInt("format_max_block_size", context->getSettingsRef().max_block_size);
+
+        insert_format = "Values";
+
+        /// Setting value from cmd arg overrides one from config
+        if (context->getSettingsRef().max_insert_block_size.changed)
+            insert_format_max_block_size = context->getSettingsRef().max_insert_block_size;
+        else
+            insert_format_max_block_size = config().getInt("insert_format_max_block_size", context->getSettingsRef().max_insert_block_size);
+
+        ClientInfo & client_info = context->getClientInfo();
+        client_info.setInitialQuery();
+        client_info.quota_key = config().getString("quota_key", "");
     }
 };
 }
diff --git a/programs/local/LocalServer.cpp b/programs/local/LocalServer.cpp
index f69a993af6c..12d7b869ede 100644
--- a/programs/local/LocalServer.cpp
+++ b/programs/local/LocalServer.cpp
@@ -12,6 +12,7 @@
 #include <Interpreters/executeQuery.h>
 #include <Interpreters/loadMetadata.h>
 #include <Interpreters/DatabaseCatalog.h>
+#include <common/getFQDNOrHostName.h>
 #include <Common/Exception.h>
 #include <Common/Macros.h>
 #include <Common/Config/ConfigProcessor.h>
@@ -191,15 +192,27 @@ static void attachSystemTables(ContextPtr context)
     attachSystemTablesLocal(*system_database);
 }
 
+void LocalServer::processMainImplException(const Exception &)
+{
+    try
+    {
+        cleanup();
+    }
+    catch (...)
+    {
+        tryLogCurrentException(__PRETTY_FUNCTION__);
+    }
 
-int LocalServer::main(const std::vector<std::string> & /*args*/)
-try
+    std::cerr << getCurrentExceptionMessage(config().hasOption("stacktrace")) << '\n';
+}
+
+int LocalServer::childMainImpl()
 {
     Poco::Logger * log = &logger();
     ThreadStatus thread_status;
     UseSSL use_ssl;
 
-    if (!config().has("query") && !config().has("table-structure") && !config().has("queries-file")) /// Nothing to process
+    if (!is_interactive && !config().has("query") && !config().has("table-structure") && !config().has("queries-file")) /// Nothing to process
     {
         if (config().hasOption("verbose"))
             std::cerr << "There are no queries to process." << '\n';
@@ -208,9 +221,19 @@ try
     }
 
     if (config().has("query") && config().has("queries-file"))
-    {
         throw Exception("Specify either `query` or `queries-file` option", ErrorCodes::BAD_ARGUMENTS);
-    }
+
+    echo_queries = config().hasOption("echo") || config().hasOption("verbose");
+
+    prompt_by_server_display_name = config().getRawString("prompt_by_server_display_name.default", "{display_name} :) ");
+    server_display_name = config().getString("display_name", getFQDNOrHostName());
+
+    /// Prompt may contain the following substitutions in a form of {name}.
+    std::map<String, String> prompt_substitutions{{"display_name", server_display_name}};
+
+    /// Quite suboptimal.
+    for (const auto & [key, value] : prompt_substitutions)
+        boost::replace_all(prompt_by_server_display_name, "{" + key + "}", value);
 
     shared_context = Context::createShared();
     global_context = Context::createGlobal(shared_context.get());
@@ -301,7 +324,40 @@ try
         attachSystemTables(global_context);
     }
 
-    processQueries();
+    /// we can't mutate global global_context (can lead to races, as it was already passed to some background threads)
+    /// so we can't reuse it safely as a query context and need a copy here
+    query_context = Context::createCopy(global_context);
+
+    query_context->makeSessionContext();
+    query_context->makeQueryContext();
+
+    query_context->setUser("default", "", Poco::Net::SocketAddress{});
+    query_context->setCurrentQueryId("");
+    applyCmdSettings(query_context);
+
+    /// Use the same query_id (and thread group) for all queries
+    CurrentThread::QueryScope query_scope_holder(query_context);
+
+    if (need_render_progress)
+    {
+        query_context->setProgressCallback([&](const Progress & value)
+        {
+            /// Write progress only if progress was updated
+            if (progress_indication.updateProgress(value))
+                progress_indication.writeProgress();
+        });
+
+        progress_indication.setFileProgressCallback(query_context);
+    }
+
+    if (is_interactive)
+    {
+        runInteractive();
+    }
+    else
+    {
+        processQueries();
+    }
 
     global_context->shutdown();
     global_context.reset();
@@ -311,23 +367,6 @@ try
 
     return Application::EXIT_OK;
 }
-catch (const Exception & e)
-{
-    try
-    {
-        cleanup();
-    }
-    catch (...)
-    {
-        tryLogCurrentException(__PRETTY_FUNCTION__);
-    }
-
-    std::cerr << getCurrentExceptionMessage(config().hasOption("stacktrace")) << '\n';
-
-    /// If exception code isn't zero, we should return non-zero return code anyway.
-    return e.code() ? e.code() : -1;
-}
-
 
 std::string LocalServer::getInitialCreateTableQuery()
 {
@@ -351,6 +390,37 @@ std::string LocalServer::getInitialCreateTableQuery()
     "; ";
 }
 
+void LocalServer::processQuery(const String & query, std::exception_ptr exception)
+{
+    written_first_block = false;
+    progress_indication.resetProgress();
+
+    ReadBufferFromString read_buf(query);
+    WriteBufferFromFileDescriptor write_buf(STDOUT_FILENO);
+
+    if (echo_queries)
+    {
+        writeString(query, write_buf);
+        writeChar('\n', write_buf);
+        write_buf.next();
+    }
+
+    try
+    {
+        std::cerr << "executing query\n";
+        executeQuery(read_buf, write_buf, /* allow_into_outfile = */ true, query_context, {});
+    }
+    catch (...)
+    {
+        if (!config().hasOption("ignore-error"))
+            throw;
+
+        if (!exception)
+            exception = std::current_exception();
+
+        std::cerr << getCurrentExceptionMessage(config().hasOption("stacktrace")) << '\n';
+    }
+}
 
 void LocalServer::processQueries()
 {
@@ -375,69 +445,11 @@ void LocalServer::processQueries()
     if (!parse_res.second)
         throw Exception("Cannot parse and execute the following part of query: " + String(parse_res.first), ErrorCodes::SYNTAX_ERROR);
 
-    /// we can't mutate global global_context (can lead to races, as it was already passed to some background threads)
-    /// so we can't reuse it safely as a query context and need a copy here
-    auto context = Context::createCopy(global_context);
-
-    context->makeSessionContext();
-    context->makeQueryContext();
-
-    context->setUser("default", "", Poco::Net::SocketAddress{});
-    context->setCurrentQueryId("");
-    applyCmdSettings(context);
-
-    /// Use the same query_id (and thread group) for all queries
-    CurrentThread::QueryScope query_scope_holder(context);
-
-    ///Set progress show
-    need_render_progress = config().getBool("progress", false);
-
-    if (need_render_progress)
-    {
-        context->setProgressCallback([&](const Progress & value)
-        {
-            /// Write progress only if progress was updated
-            if (progress_indication.updateProgress(value))
-                progress_indication.writeProgress();
-        });
-    }
-
-    bool echo_queries = config().hasOption("echo") || config().hasOption("verbose");
-
-    if (need_render_progress)
-        progress_indication.setFileProgressCallback(context);
-
     std::exception_ptr exception;
 
     for (const auto & query : queries)
     {
-        written_first_block = false;
-        progress_indication.resetProgress();
-
-        ReadBufferFromString read_buf(query);
-        WriteBufferFromFileDescriptor write_buf(STDOUT_FILENO);
-
-        if (echo_queries)
-        {
-            writeString(query, write_buf);
-            writeChar('\n', write_buf);
-            write_buf.next();
-        }
-
-        try
-        {
-            executeQuery(read_buf, write_buf, /* allow_into_outfile = */ true, context, {});
-        }
-        catch (...)
-        {
-            if (!config().hasOption("ignore-error"))
-                throw;
-
-            if (!exception)
-                exception = std::current_exception();
-
-            std::cerr << getCurrentExceptionMessage(config().hasOption("stacktrace")) << '\n';
-        }
+        processQuery(query, exception);
     }
 
     if (exception)
@@ -464,7 +476,6 @@ static const char * minimal_default_user_xml =
 "    </quotas>"
 "</yandex>";
 
-
 static ConfigurationPtr getConfigurationFromXMLString(const char * xml_data)
 {
     std::stringstream ss{std::string{xml_data}};    // STYLE_CHECK_ALLOW_STD_STRING_STREAM
@@ -472,7 +483,6 @@ static ConfigurationPtr getConfigurationFromXMLString(const char * xml_data)
     return {new Poco::Util::XMLConfiguration{&input_source}};
 }
 
-
 void LocalServer::setupUsers()
 {
     ConfigurationPtr users_config;
@@ -533,6 +543,12 @@ static std::string getHelpFooter()
             " BY mem_total DESC FORMAT PrettyCompact\"";
 }
 
+bool LocalServer::isInteractive()
+{
+    /// Nothing to process
+    return stdin_is_a_tty && !config().has("query") && !config().has("table-structure") && queries_files.empty();
+}
+
 void LocalServer::printHelpMessage(const OptionsDescription & options_description)
 {
     std::cout << getHelpHeader() << "\n";
@@ -569,7 +585,6 @@ void LocalServer::addOptions(OptionsDescription & options_description)
         ("version,V", "print version information and exit")
         ("progress", "print progress of queries execution")
         ;
-
 }
 
 void LocalServer::readArguments(int argc, char ** argv, Arguments & arguments, std::vector<Arguments> &)
@@ -621,6 +636,9 @@ void LocalServer::processOptions(const OptionsDescription &, const CommandLineOp
         config().setBool("ignore-error", true);
     if (options.count("no-system-tables"))
         config().setBool("no-system-tables", true);
+
+    if (options.count("queries-file"))
+        queries_files = options["queries-file"].as<std::vector<std::string>>();
 }
 
 void LocalServer::applyCmdOptions(ContextMutablePtr context)
diff --git a/programs/local/LocalServer.h b/programs/local/LocalServer.h
index bfe1a3a2fdb..d9c9f3f4321 100644
--- a/programs/local/LocalServer.h
+++ b/programs/local/LocalServer.h
@@ -23,8 +23,6 @@ public:
 
     void initialize(Poco::Util::Application & self) override;
 
-    int main(const std::vector<std::string> & args) override;
-
     ~LocalServer() override;
 
 private:
@@ -35,14 +33,35 @@ private:
     std::string getInitialCreateTableQuery();
 
     void tryInitPath();
+
     void applyCmdOptions(ContextMutablePtr context);
+
     void applyCmdSettings(ContextMutablePtr context);
+
     void processQueries();
+
     void setupUsers();
+
     void cleanup();
 
 
 protected:
+    void processMainImplException(const Exception & e) override;
+
+    int childMainImpl() override;
+
+    bool isInteractive() override;
+
+    bool processQueryFromInteractive(const String & input) override
+    {
+        std::exception_ptr e;
+        processQuery(input, e);
+
+        /// For clickhouse local it is ok, to return true here - i.e. interactive
+        /// mode will only be stopped by exit command.
+        return true;
+    }
+
     void printHelpMessage(const OptionsDescription & options_description) override;
 
     void readArguments(int argc, char ** argv, Arguments & common_arguments, std::vector<Arguments> &) override;
@@ -53,16 +72,17 @@ protected:
                         const CommandLineOptions & options,
                         const std::vector<Arguments> &) override;
 
+    bool supportPasswordOption() override { return false; }
+
     SharedContextHolder shared_context;
     ContextMutablePtr global_context;
 
-    bool need_render_progress = false;
-
-    bool written_first_block = false;
-
-    ProgressIndication progress_indication;
-
     std::optional<std::filesystem::path> temporary_directory_to_delete;
+
+private:
+    ContextMutablePtr query_context;
+
+    void processQuery(const String & query, std::exception_ptr exception);
 };
 
 }
diff --git a/programs/client/ConnectionParameters.cpp b/src/Client/ConnectionParameters.cpp
similarity index 98%
rename from programs/client/ConnectionParameters.cpp
rename to src/Client/ConnectionParameters.cpp
index fd0b250e66b..94854fc1b5b 100644
--- a/programs/client/ConnectionParameters.cpp
+++ b/src/Client/ConnectionParameters.cpp
@@ -13,9 +13,10 @@
 #include <common/scope_guard.h>
 
 #if !defined(ARCADIA_BUILD)
-#include <readpassphrase.h> // Y_IGNORE
+#include <readpassphrase/readpassphrase.h>
 #endif
 
+
 namespace DB
 {
 
diff --git a/programs/client/ConnectionParameters.h b/src/Client/ConnectionParameters.h
similarity index 100%
rename from programs/client/ConnectionParameters.h
rename to src/Client/ConnectionParameters.h
diff --git a/src/Client/IClient.cpp b/src/Client/IClient.cpp
index 0b1b15db648..b04a2fa58a0 100644
--- a/src/Client/IClient.cpp
+++ b/src/Client/IClient.cpp
@@ -1,6 +1,7 @@
 #include <Client/IClient.h>
 
 #include <iostream>
+#include <iomanip>
 
 #if !defined(ARCADIA_BUILD)
 #    include <Common/config_version.h>
@@ -12,6 +13,12 @@
 #include <Common/TerminalSize.h>
 #include <common/argsToConfig.h>
 #include <Common/clearPasswordFromCommandLine.h>
+#include <Common/filesystemHelpers.h>
+#include <common/LineReader.h>
+#include <Common/Config/configReadClient.h>
+#include <filesystem>
+
+namespace fs = std::filesystem;
 
 
 namespace DB
@@ -22,6 +29,7 @@ namespace ErrorCodes
     extern const int UNRECOGNIZED_ARGUMENTS;
 }
 
+
 /// Should we celebrate a bit?
 bool IClient::isNewYearMode()
 {
@@ -35,6 +43,7 @@ bool IClient::isNewYearMode()
     return (now.month() == 12 && now.day() >= 20) || (now.month() == 1 && now.day() <= 5);
 }
 
+
 bool IClient::isChineseNewYearMode(const String & local_tz)
 {
     /// Days of Dec. 20 in Chinese calendar starting from year 2019 to year 2105
@@ -104,6 +113,7 @@ bool IClient::isChineseNewYearMode(const String & local_tz)
     return false;
 }
 
+
 #if USE_REPLXX
 void IClient::highlight(const String & query, std::vector<replxx::Replxx::Color> & colors)
 {
@@ -176,6 +186,7 @@ void IClient::highlight(const String & query, std::vector<replxx::Replxx::Color>
 }
 #endif
 
+
 void IClient::clearTerminal()
 {
     /// Clear from cursor until end of screen.
@@ -186,11 +197,159 @@ void IClient::clearTerminal()
                     "\033[?25h";
 }
 
+
 static void showClientVersion()
 {
     std::cout << DBMS_NAME << " client version " << VERSION_STRING << VERSION_OFFICIAL << "." << std::endl;
 }
 
+
+void IClient::runInteractive()
+{
+    if (config().has("query_id"))
+        throw Exception("query_id could be specified only in non-interactive mode", ErrorCodes::BAD_ARGUMENTS);
+    if (print_time_to_stderr)
+        throw Exception("time option could be specified only in non-interactive mode", ErrorCodes::BAD_ARGUMENTS);
+
+    /// Initialize DateLUT here to avoid counting time spent here as query execution time.
+    const auto local_tz = DateLUT::instance().getTimeZone();
+
+    suggest.emplace();
+    loadSuggestionDataIfPossible();
+
+    if (home_path.empty())
+    {
+        const char * home_path_cstr = getenv("HOME");
+        if (home_path_cstr)
+            home_path = home_path_cstr;
+
+        configReadClient(config(), home_path);
+    }
+
+    /// Load command history if present.
+    if (config().has("history_file"))
+        history_file = config().getString("history_file");
+    else
+    {
+        auto * history_file_from_env = getenv("CLICKHOUSE_HISTORY_FILE");
+        if (history_file_from_env)
+            history_file = history_file_from_env;
+        else if (!home_path.empty())
+            history_file = home_path + "/.clickhouse-client-history";
+    }
+
+    if (!history_file.empty() && !fs::exists(history_file))
+    {
+        /// Avoid TOCTOU issue.
+        try
+        {
+            FS::createFile(history_file);
+        }
+        catch (const ErrnoException & e)
+        {
+            if (e.getErrno() != EEXIST)
+                throw;
+        }
+    }
+
+    LineReader::Patterns query_extenders = {"\\"};
+    LineReader::Patterns query_delimiters = {";", "\\G"};
+
+#if USE_REPLXX
+    replxx::Replxx::highlighter_callback_t highlight_callback{};
+    if (config().getBool("highlight", true))
+        highlight_callback = highlight;
+
+    ReplxxLineReader lr(*suggest, history_file, config().has("multiline"), query_extenders, query_delimiters, highlight_callback);
+
+#elif defined(USE_READLINE) && USE_READLINE
+    ReadlineLineReader lr(*suggest, history_file, config().has("multiline"), query_extenders, query_delimiters);
+#else
+    LineReader lr(history_file, config().has("multiline"), query_extenders, query_delimiters);
+#endif
+
+    /// Enable bracketed-paste-mode only when multiquery is enabled and multiline is
+    ///  disabled, so that we are able to paste and execute multiline queries in a whole
+    ///  instead of erroring out, while be less intrusive.
+    if (config().has("multiquery") && !config().has("multiline"))
+        lr.enableBracketedPaste();
+
+    do
+    {
+        auto input = lr.readLine(prompt(), ":-] ");
+        if (input.empty())
+            break;
+
+        has_vertical_output_suffix = false;
+        if (input.ends_with("\\G"))
+        {
+            input.resize(input.size() - 2);
+            has_vertical_output_suffix = true;
+        }
+
+        if (!processQueryFromInteractive(input))
+            break;
+    }
+    while (true);
+
+    if (isNewYearMode())
+        std::cout << "Happy new year." << std::endl;
+    else if (isChineseNewYearMode(local_tz))
+        std::cout << "Happy Chinese new year. 春节快乐!" << std::endl;
+    else
+        std::cout << "Bye." << std::endl;
+}
+
+
+int IClient::mainImpl()
+{
+    if (isInteractive())
+        is_interactive = true;
+
+    if (config().has("query") && !queries_files.empty())
+        throw Exception("Specify either `query` or `queries-file` option", ErrorCodes::BAD_ARGUMENTS);
+
+    std::cout << std::fixed << std::setprecision(3);
+    std::cerr << std::fixed << std::setprecision(3);
+
+    if (is_interactive)
+    {
+
+        clearTerminal();
+        showClientVersion();
+    }
+    else
+    {
+        need_render_progress = config().getBool("progress", false);
+        echo_queries = config().getBool("echo", false);
+        ignore_error = config().getBool("ignore-error", false);
+    }
+
+    return childMainImpl();
+}
+
+
+int IClient::main(const std::vector<std::string> & /*args*/)
+{
+    try
+    {
+        return mainImpl();
+    }
+    catch (const Exception & e)
+    {
+        processMainImplException(e);
+
+        /// If exception code isn't zero, we should return non-zero return code anyway.
+        return e.code() ? e.code() : -1;
+    }
+    catch (...)
+    {
+        std::cerr << getCurrentExceptionMessage(false) << std::endl;
+        return getCurrentExceptionCode();
+    }
+}
+
+
 void IClient::init(int argc, char ** argv)
 {
     namespace po = boost::program_options;
@@ -250,7 +409,8 @@ void IClient::init(int argc, char ** argv)
     processOptions(options_description, options, external_tables_arguments);
 
     argsToConfig(common_arguments, config(), 100);
-    clearPasswordFromCommandLine(argc, argv);
+    if (supportPasswordOption())
+        clearPasswordFromCommandLine(argc, argv);
 }
 
 }
diff --git a/src/Client/IClient.h b/src/Client/IClient.h
index 5fd8ddd8486..4d23b75f08a 100644
--- a/src/Client/IClient.h
+++ b/src/Client/IClient.h
@@ -13,6 +13,8 @@
 #include <Core/Names.h>
 #include <Core/Types.h>
 #include <Interpreters/Context.h>
+#include <Common/ProgressIndication.h>
+#include <Client/Suggest.h>
 
 
 namespace DB
@@ -23,16 +25,26 @@ class IClient : public Poco::Util::Application
 public:
     using Arguments = std::vector<std::string>;
 
+    int main(const std::vector<std::string> & /*args*/) override;
+
+    int mainImpl();
+
     void init(int argc, char ** argv);
 
 protected:
     NameSet exit_strings{"exit", "quit", "logout", "учше", "йгше", "дщпщге", "exit;", "quit;", "logout;", "учшеж",
                          "йгшеж", "дщпщгеж", "q", "й", "\\q", "\\Q", "\\й", "\\Й", ":q", "Жй"};
 
-    bool is_interactive = true; /// Use either interactive line editing interface or batch mode.
+    bool is_interactive = false; /// Use either interactive line editing interface or batch mode.
+
+    bool need_render_progress = true;
+    bool written_first_block = false;
+    ProgressIndication progress_indication;
+
     bool echo_queries = false; /// Print queries before execution in batch mode.
     bool ignore_error = false; /// In case of errors, don't print error message, continue to next query. Only applicable for non-interactive mode.
     bool print_time_to_stderr = false; /// Output execution time to stderr in batch mode.
+
     bool stdin_is_a_tty = false; /// stdin is a terminal.
     bool stdout_is_a_tty = false; /// stdout is a terminal.
 
@@ -41,7 +53,23 @@ protected:
     /// Settings specified via command line args
     Settings cmd_settings;
 
+    /// If not empty, queries will be read from these files
+    std::vector<String> queries_files;
+
+    std::optional<Suggest> suggest;
+
+    /// Path to a file containing command history.
+    String history_file;
+
+    String home_path;
+
+    bool has_vertical_output_suffix = false; /// Is \G present at the end of the query string?
+
+    String prompt_by_server_display_name;
+    String server_display_name;
+
     static bool isNewYearMode();
+
     static bool isChineseNewYearMode(const String & local_tz);
 
 #if USE_REPLXX
@@ -50,6 +78,18 @@ protected:
 
     static void clearTerminal();
 
+    void runInteractive();
+
+    virtual int childMainImpl() = 0;
+
+    virtual void processMainImplException(const Exception & e) = 0;
+
+    virtual bool isInteractive() = 0;
+
+    virtual void loadSuggestionDataIfPossible() {}
+
+    virtual bool processQueryFromInteractive(const String & input) = 0;
+
     virtual void readArguments(int argc, char ** argv,
                                Arguments & common_arguments, std::vector<Arguments> &) = 0;
 
@@ -69,6 +109,15 @@ protected:
     virtual void processOptions(const OptionsDescription & options_description,
                                 const CommandLineOptions & options,
                                 const std::vector<Arguments> & external_tables_arguments) = 0;
+
+    virtual bool supportPasswordOption() = 0;
+
+private:
+    inline String prompt() const
+    {
+        return boost::replace_all_copy(prompt_by_server_display_name, "{database}", config().getString("database", "default"));
+    }
+
 };
 
 }
diff --git a/programs/client/Suggest.cpp b/src/Client/Suggest.cpp
similarity index 100%
rename from programs/client/Suggest.cpp
rename to src/Client/Suggest.cpp
diff --git a/programs/client/Suggest.h b/src/Client/Suggest.h
similarity index 100%
rename from programs/client/Suggest.h
rename to src/Client/Suggest.h