diff --git a/src/Storages/StorageMerge.cpp b/src/Storages/StorageMerge.cpp
index d485e3283637..8b9ce51bfd61 100644
--- a/src/Storages/StorageMerge.cpp
+++ b/src/Storages/StorageMerge.cpp
@@ -1,5 +1,6 @@
 #include <functional>
 #include <iterator>
+#include <Access/ContextAccess.h>
 #include <Analyzer/ConstantNode.h>
 #include <Analyzer/ColumnNode.h>
 #include <Analyzer/FunctionNode.h>
@@ -72,6 +73,18 @@ namespace Setting
     extern const SettingsUInt64 merge_table_max_tables_to_look_for_schema_inference;
 }
 
+namespace ErrorCodes
+{
+extern const int LOGICAL_ERROR;
+extern const int BAD_ARGUMENTS;
+extern const int NOT_IMPLEMENTED;
+extern const int NUMBER_OF_ARGUMENTS_DOESNT_MATCH;
+extern const int SAMPLING_NOT_SUPPORTED;
+extern const int ALTER_OF_COLUMN_IS_FORBIDDEN;
+extern const int CANNOT_EXTRACT_TABLE_STRUCTURE;
+extern const int STORAGE_REQUIRES_PARAMETER;
+}
+
 namespace
 {
 
@@ -109,18 +122,6 @@ void rewriteEntityInAst(ASTPtr ast, const String & column_name, const Field & va
 
 }
 
-namespace ErrorCodes
-{
-    extern const int LOGICAL_ERROR;
-    extern const int BAD_ARGUMENTS;
-    extern const int NOT_IMPLEMENTED;
-    extern const int NUMBER_OF_ARGUMENTS_DOESNT_MATCH;
-    extern const int SAMPLING_NOT_SUPPORTED;
-    extern const int ALTER_OF_COLUMN_IS_FORBIDDEN;
-    extern const int CANNOT_EXTRACT_TABLE_STRUCTURE;
-    extern const int STORAGE_REQUIRES_PARAMETER;
-}
-
 StorageMerge::DatabaseNameOrRegexp::DatabaseNameOrRegexp(
     const String & source_database_name_or_regexp_,
     bool database_is_regexp_,
@@ -142,6 +143,8 @@ StorageMerge::StorageMerge(
     const String & source_database_name_or_regexp_,
     bool database_is_regexp_,
     const DBToTableSetMap & source_databases_and_tables_,
+    const std::optional<String> & table_to_write_,
+    bool table_to_write_auto_,
     ContextPtr context_)
     : IStorage(table_id_)
     , WithContext(context_->getGlobalContext())
@@ -150,14 +153,17 @@ StorageMerge::StorageMerge(
         database_is_regexp_,
         source_database_name_or_regexp_, {},
         source_databases_and_tables_)
+    , table_to_write_auto(table_to_write_auto_)
 {
     StorageInMemoryMetadata storage_metadata;
     storage_metadata.setColumns(columns_.empty()
-        ? getColumnsDescriptionFromSourceTables(context_->getSettingsRef()[Setting::merge_table_max_tables_to_look_for_schema_inference])
+        ? getColumnsDescriptionFromSourceTables(context_)
         : columns_);
     storage_metadata.setComment(comment);
     setInMemoryMetadata(storage_metadata);
     setVirtuals(createVirtuals());
+    if (!table_to_write_auto)
+        setTableToWrite(table_to_write_, source_database_name_or_regexp_, database_is_regexp_);
 }
 
 StorageMerge::StorageMerge(
@@ -167,6 +173,8 @@ StorageMerge::StorageMerge(
     const String & source_database_name_or_regexp_,
     bool database_is_regexp_,
     const String & source_table_regexp_,
+    const std::optional<String> & table_to_write_,
+    bool table_to_write_auto_,
     ContextPtr context_)
     : IStorage(table_id_)
     , WithContext(context_->getGlobalContext())
@@ -175,14 +183,17 @@ StorageMerge::StorageMerge(
         database_is_regexp_,
         source_database_name_or_regexp_,
         source_table_regexp_, {})
+    , table_to_write_auto(table_to_write_auto_)
 {
     StorageInMemoryMetadata storage_metadata;
     storage_metadata.setColumns(columns_.empty()
-        ? getColumnsDescriptionFromSourceTables(context_->getSettingsRef()[Setting::merge_table_max_tables_to_look_for_schema_inference])
+        ? getColumnsDescriptionFromSourceTables(context_)
         : columns_);
     storage_metadata.setComment(comment);
     setInMemoryMetadata(storage_metadata);
     setVirtuals(createVirtuals());
+    if (!table_to_write_auto)
+        setTableToWrite(table_to_write_, source_database_name_or_regexp_, database_is_regexp_);
 }
 
 StorageMerge::DatabaseTablesIterators StorageMerge::getDatabaseIterators(ContextPtr context_) const
@@ -190,12 +201,44 @@ StorageMerge::DatabaseTablesIterators StorageMerge::getDatabaseIterators(Context
     return database_name_or_regexp.getDatabaseIterators(context_);
 }
 
-ColumnsDescription StorageMerge::unifyColumnsDescription(std::function<void(std::function<void(const StoragePtr &)>)> for_each_table)
+ColumnsDescription StorageMerge::getColumnsDescriptionFromSourceTables(
+    const ContextPtr & query_context,
+    const String & source_database_name_or_regexp,
+    bool database_is_regexp,
+    const String & source_table_regexp,
+    size_t max_tables_to_look)
+{
+    DatabaseNameOrRegexp database_name_or_regexp(source_database_name_or_regexp, database_is_regexp, source_database_name_or_regexp, source_table_regexp, {});
+    return getColumnsDescriptionFromSourceTablesImpl(query_context, database_name_or_regexp, max_tables_to_look, nullptr);
+}
+
+ColumnsDescription StorageMerge::getColumnsDescriptionFromSourceTables(const ContextPtr & query_context) const
+{
+    auto max_tables_to_look = query_context->getSettingsRef()[Setting::merge_table_max_tables_to_look_for_schema_inference];
+    auto res = getColumnsDescriptionFromSourceTablesImpl(query_context, database_name_or_regexp, max_tables_to_look, this);
+    if (res.empty())
+        throw Exception{DB::ErrorCodes::CANNOT_EXTRACT_TABLE_STRUCTURE, "There are no tables satisfied provided regexp, you must specify table structure manually"};
+    return res;
+}
+
+ColumnsDescription StorageMerge::getColumnsDescriptionFromSourceTablesImpl(
+    const ContextPtr & query_context,
+    const DatabaseNameOrRegexp & database_name_or_regexp,
+    size_t max_tables_to_look,
+    const IStorage * ignore_self)
 {
+    auto access = query_context->getAccess();
+    size_t table_num = 0;
     ColumnsDescription res;
 
-    for_each_table([&res](auto && t)
+    traverseTablesUntilImpl(query_context, ignore_self, database_name_or_regexp, [&table_num, &access, &res, max_tables_to_look](auto && t)
     {
+        if (!t)
+            return false;
+
+        if (auto id = t->getStorageID(); !access->isGranted(AccessType::SHOW_TABLES, id.database_name, id.table_name))
+            return false;
+
         auto structure = t->getInMemoryMetadataPtr()->getColumns();
         String prev_column_name;
         for (const ColumnDescription & column : structure)
@@ -215,44 +258,31 @@ ColumnsDescription StorageMerge::unifyColumnsDescription(std::function<void(std:
             }
             prev_column_name = column.name;
         }
-    });
 
-    if (res.empty())
-        throw Exception{ErrorCodes::CANNOT_EXTRACT_TABLE_STRUCTURE, "There are no tables satisfied provided regexp, you must specify table structure manually"};
+        ++table_num;
+        return table_num >= max_tables_to_look;
+    });
 
     return res;
 }
 
-ColumnsDescription StorageMerge::getColumnsDescriptionFromSourceTables(size_t max_tables_to_look) const
+template <typename F>
+StoragePtr StorageMerge::traverseTablesUntil(F && predicate) const
 {
-    size_t table_num = 0;
-
-    return unifyColumnsDescription([&table_num, max_tables_to_look, this](std::function<void(const StoragePtr &)> callback)
-    {
-        getFirstTable([&table_num, &callback, max_tables_to_look](auto && t)
-        {
-            if (!t)
-                return false;
-
-            callback(t);
-
-            ++table_num;
-            return table_num >= max_tables_to_look;
-        });
-    });
+    return traverseTablesUntilImpl(getContext(), this, database_name_or_regexp, std::forward<F>(predicate));
 }
 
 template <typename F>
-StoragePtr StorageMerge::getFirstTable(F && predicate) const
+StoragePtr StorageMerge::traverseTablesUntilImpl(const ContextPtr & query_context, const IStorage * ignore_self, const DatabaseNameOrRegexp & database_name_or_regexp, F && predicate)
 {
-    auto database_table_iterators = database_name_or_regexp.getDatabaseIterators(getContext());
+    auto database_table_iterators = database_name_or_regexp.getDatabaseIterators(query_context);
 
     for (auto & iterator : database_table_iterators)
     {
         while (iterator->isValid())
         {
             const auto & table = iterator->table();
-            if (table.get() != this && predicate(table))
+            if (table.get() != ignore_self && predicate(table))
                 return table;
 
             iterator->next();
@@ -265,7 +295,7 @@ StoragePtr StorageMerge::getFirstTable(F && predicate) const
 template <typename F>
 void StorageMerge::forEachTable(F && func) const
 {
-    getFirstTable([&func](const auto & table)
+    traverseTablesUntil([&func](const auto & table)
     {
         func(table);
         /// Always continue to the next table.
@@ -273,15 +303,38 @@ void StorageMerge::forEachTable(F && func) const
     });
 }
 
+template <typename F>
+void StorageMerge::forEachTableName(F && func) const
+{
+    auto database_table_iterators = database_name_or_regexp.getDatabaseIterators(getContext());
+
+    for (auto & iterator : database_table_iterators)
+    {
+        while (iterator->isValid())
+        {
+            const auto & table = iterator->table();
+            if (table.get() != this)
+            {
+                QualifiedTableName table_name;
+                table_name.database = iterator->databaseName();
+                table_name.table = iterator->name();
+                func(table_name);
+            }
+
+            iterator->next();
+        }
+    }
+}
+
 bool StorageMerge::isRemote() const
 {
-    auto first_remote_table = getFirstTable([](const StoragePtr & table) { return table && table->isRemote(); });
+    auto first_remote_table = traverseTablesUntil([](const StoragePtr & table) { return table && table->isRemote(); });
     return first_remote_table != nullptr;
 }
 
 bool StorageMerge::supportsPrewhere() const
 {
-    return getFirstTable([](const auto & table) { return !table->supportsPrewhere(); }) == nullptr;
+    return traverseTablesUntil([](const auto & table) { return !table->supportsPrewhere(); }) == nullptr;
 }
 
 bool StorageMerge::canMoveConditionsToPrewhere() const
@@ -296,7 +349,7 @@ bool StorageMerge::canMoveConditionsToPrewhere() const
     /// NOTE: Type can be different, and in this case, PREWHERE cannot be
     /// applied for those columns, but there a separate method to return
     /// supported columns for PREWHERE - supportedPrewhereColumns().
-    return getFirstTable([](const auto & table) { return !table->canMoveConditionsToPrewhere(); }) == nullptr;
+    return traverseTablesUntil([](const auto & table) { return !table->canMoveConditionsToPrewhere(); }) == nullptr;
 }
 
 std::optional<NameSet> StorageMerge::supportedPrewhereColumns() const
@@ -395,12 +448,20 @@ VirtualColumnsDescription StorageMerge::createVirtuals()
     return desc;
 }
 
-StorageSnapshotPtr StorageMerge::getStorageSnapshot(const StorageMetadataPtr & metadata_snapshot, ContextPtr) const
+StorageSnapshotPtr StorageMerge::getStorageSnapshot(const StorageMetadataPtr & metadata_snapshot, ContextPtr query_context) const
 {
     static const auto common_virtuals = createVirtuals();
+    const auto & access = query_context->getAccess();
 
     auto virtuals = common_virtuals;
-    if (auto first_table = getFirstTable([](auto && table) { return table; }))
+    if (auto first_table = traverseTablesUntil([access](auto && table)
+    {
+        if (!table)
+            return false;
+
+        auto id = table->getStorageID();
+        return access->isGranted(AccessType::SHOW_TABLES, id.database_name, id.table_name);
+    }))
     {
         auto table_virtuals = first_table->getVirtualsPtr();
         for (const auto & column : *table_virtuals)
@@ -1290,22 +1351,41 @@ StorageMerge::StorageListWithLocks ReadFromMerge::getSelectedTables(
     StorageListWithLocks res;
     DatabaseTablesIterators database_table_iterators = assert_cast<StorageMerge &>(*storage_merge).getDatabaseIterators(query_context);
 
-    MutableColumnPtr database_name_virtual_column;
-    MutableColumnPtr table_name_virtual_column;
-    if (filter_by_database_virtual_column)
+    std::function<bool(const String&,const String&)> table_filter;
+    if (filter_actions_dag && (filter_by_database_virtual_column || filter_by_table_virtual_column))
     {
-        database_name_virtual_column = ColumnString::create();
-    }
-
-    if (filter_by_table_virtual_column)
-    {
-        table_name_virtual_column = ColumnString::create();
+        Block sample_block = {
+            ColumnWithTypeAndName(std::make_shared<DataTypeString>(), "_database"),
+            ColumnWithTypeAndName(std::make_shared<DataTypeString>(), "_table")
+        };
+        // Extract predicate part, that could be evaluated only with _database and _table columns
+        auto table_filter_dag = VirtualColumnUtils::splitFilterDagForAllowedInputs(filter_actions_dag->getOutputs().at(0), &sample_block);
+        if (table_filter_dag)
+        {
+            auto filter_expression = VirtualColumnUtils::buildFilterExpression(std::move(*table_filter_dag), query_context);
+            auto filter_column_name = filter_expression->getActionsDAG().getOutputs().at(0)->result_name;
+            table_filter = [filter=std::move(filter_expression), column_name=std::move(filter_column_name)] (const auto& database_name, const auto& table_name)
+            {
+                MutableColumnPtr database_column = ColumnString::create();
+                MutableColumnPtr table_column = ColumnString::create();
+                database_column->insert(database_name);
+                table_column->insert(table_name);
+                Block block{
+                    ColumnWithTypeAndName(std::move(database_column), std::make_shared<DataTypeString>(), "_database"),
+                    ColumnWithTypeAndName(std::move(table_column), std::make_shared<DataTypeString>(), "_table")
+                };
+                filter->execute(block);
+                // Valid only when block has exactly one row.
+                return block.getByName(column_name).column->getBool(0);
+            };
+        }
     }
 
+    auto access = query_context->getAccess();
     for (const auto & iterator : database_table_iterators)
     {
-        if (filter_by_database_virtual_column)
-            database_name_virtual_column->insert(iterator->databaseName());
+        auto granted_show_on_all_tables = access->isGranted(AccessType::SHOW_TABLES, iterator->databaseName());
+        auto granted_select_on_all_tables = access->isGranted(AccessType::SELECT, iterator->databaseName());
         while (iterator->isValid())
         {
             StoragePtr storage = iterator->table();
@@ -1313,48 +1393,20 @@ StorageMerge::StorageListWithLocks ReadFromMerge::getSelectedTables(
                 continue;
 
             if (storage.get() != storage_merge.get())
-            {
-                auto table_lock = storage->lockForShare(query_context->getCurrentQueryId(), settings[Setting::lock_acquire_timeout]);
-                res.emplace_back(iterator->databaseName(), storage, std::move(table_lock), iterator->name());
-                if (filter_by_table_virtual_column)
-                    table_name_virtual_column->insert(iterator->name());
-            }
+                if (!table_filter || table_filter(iterator->databaseName(), iterator->name()))
+                    if (granted_show_on_all_tables || access->isGranted(AccessType::SHOW_TABLES, iterator->databaseName(), iterator->name()))
+                    {
+
+                        if  (!granted_select_on_all_tables)
+                            access->checkAccess(AccessType::SELECT, iterator->databaseName(), iterator->name(), column_names);
 
+                        auto table_lock = storage->lockForShare(query_context->getCurrentQueryId(), settings[Setting::lock_acquire_timeout]);
+                        res.emplace_back(iterator->databaseName(), storage, std::move(table_lock), iterator->name());
+                    }
             iterator->next();
         }
     }
 
-    if (!filter_by_database_virtual_column && !filter_by_table_virtual_column)
-        return res;
-
-    if (!filter_actions_dag)
-        return res;
-
-    const auto * predicate = filter_actions_dag->getOutputs().at(0);
-
-    if (filter_by_database_virtual_column)
-    {
-        /// Filter names of selected tables if there is a condition on "_database" virtual column in WHERE clause
-        Block virtual_columns_block
-            = Block{ColumnWithTypeAndName(std::move(database_name_virtual_column), std::make_shared<DataTypeString>(), "_database")};
-        VirtualColumnUtils::filterBlockWithPredicate(predicate, virtual_columns_block, query_context);
-        auto values = VirtualColumnUtils::extractSingleValueFromBlock<String>(virtual_columns_block, "_database");
-
-        /// Remove unused databases from the list
-        res.remove_if([&](const auto & elem) { return values.find(std::get<0>(elem)) == values.end(); });
-    }
-
-    if (filter_by_table_virtual_column)
-    {
-        /// Filter names of selected tables if there is a condition on "_table" virtual column in WHERE clause
-        Block virtual_columns_block = Block{ColumnWithTypeAndName(std::move(table_name_virtual_column), std::make_shared<DataTypeString>(), "_table")};
-        VirtualColumnUtils::filterBlockWithPredicate(predicate, virtual_columns_block, query_context);
-        auto values = VirtualColumnUtils::extractSingleValueFromBlock<String>(virtual_columns_block, "_table");
-
-        /// Remove unused tables from the list
-        res.remove_if([&](const auto & elem) { return values.find(std::get<3>(elem)) == values.end(); });
-    }
-
     return res;
 }
 
@@ -1653,7 +1705,7 @@ bool StorageMerge::supportsTrivialCountOptimization(const StorageSnapshotPtr &,
 {
     /// Here we actually need storage snapshot of all nested tables.
     /// But to avoid complexity pass nullptr to make more lightweight check in MergeTreeData.
-    return getFirstTable([&](const auto & table) { return !table->supportsTrivialCountOptimization(nullptr, ctx); }) == nullptr;
+    return traverseTablesUntil([&](const auto & table) { return !table->supportsTrivialCountOptimization(nullptr, ctx); }) == nullptr;
 }
 
 std::optional<UInt64> StorageMerge::totalRows(ContextPtr query_context) const
@@ -1670,7 +1722,7 @@ template <typename F>
 std::optional<UInt64> StorageMerge::totalRowsOrBytes(F && func) const
 {
     UInt64 total_rows_or_bytes = 0;
-    auto first_table = getFirstTable([&](const auto & table)
+    auto first_table = traverseTablesUntil([&](const auto & table)
     {
         if (auto rows_or_bytes = func(table))
         {
@@ -1683,6 +1735,58 @@ std::optional<UInt64> StorageMerge::totalRowsOrBytes(F && func) const
     return first_table ? std::nullopt : std::make_optional(total_rows_or_bytes);
 }
 
+void StorageMerge::setTableToWrite(
+    const std::optional<String> & table_to_write_,
+    const String & source_database_name_or_regexp_,
+    bool database_is_regexp_)
+{
+    if (!table_to_write_.has_value())
+    {
+        table_to_write = std::nullopt;
+        return;
+    }
+
+    auto qualified_name = QualifiedTableName::parseFromString(*table_to_write_);
+
+    if (qualified_name.database.empty())
+    {
+        if (database_is_regexp_)
+            throw Exception(ErrorCodes::BAD_ARGUMENTS, "Argument 'table_to_write' must contain database if 'db_name' is regular expression.");
+
+        qualified_name.database = source_database_name_or_regexp_;
+    }
+
+    table_to_write = qualified_name;
+}
+
+SinkToStoragePtr StorageMerge::write(
+    const ASTPtr & query,
+    const StorageMetadataPtr & metadata_snapshot,
+    ContextPtr context_,
+    bool async_insert)
+{
+    if (table_to_write_auto)
+    {
+        table_to_write = std::nullopt;
+        forEachTableName([&](const auto & table_name)
+        {
+            if (!table_to_write.has_value())
+                table_to_write = table_name;
+            else if (table_to_write->getFullName() < table_name.getFullName())
+                table_to_write = table_name;
+        });
+    }
+
+    if (!table_to_write.has_value())
+        throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Method write is not allowed in storage {} without described table to write.", getName());
+
+    context_->getAccess()->checkAccess(AccessType::INSERT, table_to_write->database, table_to_write->table);
+
+    auto database = DatabaseCatalog::instance().getDatabase(table_to_write->database);
+    auto table = database->getTable(table_to_write->table, context_);
+    return table->write(query, metadata_snapshot, context_, async_insert);
+}
+
 void registerStorageMerge(StorageFactory & factory)
 {
     factory.registerStorage("Merge", [](const StorageFactory::Arguments & args)
@@ -1693,10 +1797,12 @@ void registerStorageMerge(StorageFactory & factory)
 
         ASTs & engine_args = args.engine_args;
 
-        if (engine_args.size() != 2)
+        size_t size = engine_args.size();
+
+        if (size < 2 || size > 3)
             throw Exception(ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH,
-                            "Storage Merge requires exactly 2 parameters - name "
-                            "of source database and regexp for table names.");
+                            "Storage Merge requires 2 or 3 parameters - name "
+                            "of source database, regexp for table names, and optional table name for writing.");
 
         auto [is_regexp, database_ast] = StorageMerge::evaluateDatabaseName(engine_args[0], args.getLocalContext());
 
@@ -1708,8 +1814,24 @@ void registerStorageMerge(StorageFactory & factory)
         engine_args[1] = evaluateConstantExpressionAsLiteral(engine_args[1], args.getLocalContext());
         String table_name_regexp = checkAndGetLiteralArgument<String>(engine_args[1], "table_name_regexp");
 
+        std::optional<String> table_to_write = std::nullopt;
+        bool table_to_write_auto = false;
+        if (size == 3)
+        {
+            bool is_identifier = engine_args[2]->as<ASTIdentifier>();
+            engine_args[2] = evaluateConstantExpressionOrIdentifierAsLiteral(engine_args[2], args.getLocalContext());
+            table_to_write = checkAndGetLiteralArgument<String>(engine_args[2], "table_to_write");
+            if (is_identifier && table_to_write == "auto")
+            {
+                if (is_regexp)
+                    throw Exception(ErrorCodes::BAD_ARGUMENTS, "RegExp for database with auto table_to_write is forbidden.");
+                table_to_write_auto = true;
+            }
+        }
+
         return std::make_shared<StorageMerge>(
-            args.table_id, args.columns, args.comment, source_database_name_or_regexp, is_regexp, table_name_regexp, args.getContext());
+            args.table_id, args.columns, args.comment, source_database_name_or_regexp, is_regexp,
+            table_name_regexp, table_to_write, table_to_write_auto, args.getLocalContext());
     },
     {
         .supports_schema_inference = true
diff --git a/src/Storages/StorageMerge.h b/src/Storages/StorageMerge.h
index 562b3e981e2b..d9761b21cf85 100644
--- a/src/Storages/StorageMerge.h
+++ b/src/Storages/StorageMerge.h
@@ -30,6 +30,8 @@ class StorageMerge final : public IStorage, WithContext
         const String & source_database_name_or_regexp_,
         bool database_is_regexp_,
         const DBToTableSetMap & source_databases_and_tables_,
+        const std::optional<String> & table_to_write_,
+        bool table_to_write_auto_,
         ContextPtr context_);
 
     StorageMerge(
@@ -39,6 +41,8 @@ class StorageMerge final : public IStorage, WithContext
         const String & source_database_name_or_regexp_,
         bool database_is_regexp_,
         const String & source_table_regexp_,
+        const std::optional<String> & table_to_write_,
+        bool table_to_write_auto_,
         ContextPtr context_);
 
     std::string getName() const override { return "Merge"; }
@@ -70,6 +74,12 @@ class StorageMerge final : public IStorage, WithContext
         size_t max_block_size,
         size_t num_streams) override;
 
+    SinkToStoragePtr write(
+        const ASTPtr & query,
+        const StorageMetadataPtr & metadata_snapshot,
+        ContextPtr context,
+        bool async_insert) override;
+
     void checkAlterIsPossible(const AlterCommands & commands, ContextPtr context) const override;
 
     /// you need to add and remove columns in the sub-tables manually
@@ -87,9 +97,12 @@ class StorageMerge final : public IStorage, WithContext
     using DatabaseTablesIterators = std::vector<DatabaseTablesIteratorPtr>;
     DatabaseTablesIterators getDatabaseIterators(ContextPtr context) const;
 
-    /// Returns a unified column structure among multiple tables.
-    /// Takes a function that invokes a callback for every table. NOTE: This is quite inconvenient.
-    static ColumnsDescription unifyColumnsDescription(std::function<void(std::function<void(const StoragePtr &)>)> for_each_table);
+    static ColumnsDescription getColumnsDescriptionFromSourceTables(
+        const ContextPtr & query_context,
+        const String & source_database_name_or_regexp,
+        bool database_is_regexp,
+        const String & source_table_regexp,
+        size_t max_tables_to_look);
 
 private:
     /// (Database, Table, Lock, TableName)
@@ -119,15 +132,31 @@ class StorageMerge final : public IStorage, WithContext
 
     DatabaseNameOrRegexp database_name_or_regexp;
 
+    std::optional<QualifiedTableName> table_to_write;
+    bool table_to_write_auto = false;
+
     template <typename F>
-    StoragePtr getFirstTable(F && predicate) const;
+    StoragePtr traverseTablesUntil(F && predicate) const;
 
     template <typename F>
     void forEachTable(F && func) const;
 
+    template <typename F>
+    void forEachTableName(F && func) const;
+
+    template <typename F>
+    static StoragePtr traverseTablesUntilImpl(const ContextPtr & query_context, const IStorage * ignore_self, const DatabaseNameOrRegexp & database_name_or_regexp, F && predicate);
+
+    /// Returns a unified column structure among multiple tables.
+    static ColumnsDescription getColumnsDescriptionFromSourceTablesImpl(
+        const ContextPtr & context,
+        const DatabaseNameOrRegexp & database_name_or_regexp,
+        size_t max_tables_to_look,
+        const IStorage * ignore_self);
+
     ColumnSizeByName getColumnSizes() const override;
 
-    ColumnsDescription getColumnsDescriptionFromSourceTables(size_t max_tables_to_look) const;
+    ColumnsDescription getColumnsDescriptionFromSourceTables(const ContextPtr & context) const;
 
     static VirtualColumnsDescription createVirtuals();
 
@@ -136,6 +165,11 @@ class StorageMerge final : public IStorage, WithContext
     template <typename F>
     std::optional<UInt64> totalRowsOrBytes(F && func) const;
 
+    void setTableToWrite(
+        const std::optional<String> & table_to_write_,
+        const String & source_database_name_or_regexp_,
+        bool database_is_regexp_);
+
     friend class ReadFromMerge;
 };
 
diff --git a/src/TableFunctions/TableFunctionMerge.cpp b/src/TableFunctions/TableFunctionMerge.cpp
index e6ada9e5fcaf..443a31693ff5 100644
--- a/src/TableFunctions/TableFunctionMerge.cpp
+++ b/src/TableFunctions/TableFunctionMerge.cpp
@@ -57,18 +57,13 @@ class TableFunctionMerge : public ITableFunction
     StoragePtr executeImpl(const ASTPtr & ast_function, ContextPtr context, const std::string & table_name, ColumnsDescription cached_columns, bool is_insert_query) const override;
     const char * getStorageTypeName() const override { return "Merge"; }
 
-    using TableSet = std::set<String>;
-    using DBToTableSetMap = std::map<String, TableSet>;
-    const DBToTableSetMap & getSourceDatabasesAndTables(ContextPtr context) const;
     ColumnsDescription getActualTableStructure(ContextPtr context, bool is_insert_query) const override;
     std::vector<size_t> skipAnalysisForArguments(const QueryTreeNodePtr & query_node_table_function, ContextPtr context) const override;
     void parseArguments(const ASTPtr & ast_function, ContextPtr context) override;
-    static TableSet getMatchedTablesWithAccess(const String & database_name, const String & table_regexp, const ContextPtr & context);
 
     String source_database_name_or_regexp;
     String source_table_regexp;
     bool database_is_regexp = false;
-    mutable std::optional<DBToTableSetMap> source_databases_and_tables;
 };
 
 std::vector<size_t> TableFunctionMerge::skipAnalysisForArguments(const QueryTreeNodePtr & query_node_table_function, ContextPtr) const
@@ -129,111 +124,39 @@ void TableFunctionMerge::parseArguments(const ASTPtr & ast_function, ContextPtr
     }
 }
 
-
-const TableFunctionMerge::DBToTableSetMap & TableFunctionMerge::getSourceDatabasesAndTables(ContextPtr context) const
-{
-    if (source_databases_and_tables)
-        return *source_databases_and_tables;
-
-    source_databases_and_tables.emplace();
-
-    /// database_name is not a regexp
-    if (!database_is_regexp)
-    {
-        auto source_tables = getMatchedTablesWithAccess(source_database_name_or_regexp, source_table_regexp, context);
-        if (source_tables.empty())
-            throwNoTablesMatchRegexp(source_database_name_or_regexp, source_table_regexp);
-        (*source_databases_and_tables)[source_database_name_or_regexp] = source_tables;
-    }
-
-    /// database_name is a regexp
-    else
-    {
-        OptimizedRegularExpression database_re(source_database_name_or_regexp);
-        auto databases = DatabaseCatalog::instance().getDatabases();
-
-        for (const auto & db : databases)
-            if (database_re.match(db.first))
-                (*source_databases_and_tables)[db.first] = getMatchedTablesWithAccess(db.first, source_table_regexp, context);
-
-        if (source_databases_and_tables->empty())
-            throwNoTablesMatchRegexp(source_database_name_or_regexp, source_table_regexp);
-    }
-
-    return *source_databases_and_tables;
-}
-
 ColumnsDescription TableFunctionMerge::getActualTableStructure(ContextPtr context, bool /*is_insert_query*/) const
 {
-    size_t table_num = 0;
-    size_t max_tables_to_look = context->getSettingsRef()[Setting::merge_table_max_tables_to_look_for_schema_inference];
+    auto res = StorageMerge::getColumnsDescriptionFromSourceTables(
+        context,
+        source_database_name_or_regexp,
+        database_is_regexp,
+        source_table_regexp,
+        context->getSettingsRef()[Setting::merge_table_max_tables_to_look_for_schema_inference]);
+    if (res.empty())
+        throwNoTablesMatchRegexp(source_database_name_or_regexp, source_table_regexp);
 
-    return StorageMerge::unifyColumnsDescription([&table_num, &context, max_tables_to_look, this](std::function<void(const StoragePtr &)> callback)
-    {
-        for (const auto & db_with_tables : getSourceDatabasesAndTables(context))
-        {
-            for (const auto & table : db_with_tables.second)
-            {
-                if (table_num >= max_tables_to_look)
-                    return;
-
-                auto storage = DatabaseCatalog::instance().tryGetTable(StorageID{db_with_tables.first, table}, context);
-                if (storage)
-                {
-                    ++table_num;
-                    callback(storage);
-                }
-            }
-        }
-    });
+    return res;
 }
 
 
 StoragePtr TableFunctionMerge::executeImpl(const ASTPtr & /*ast_function*/, ContextPtr context, const std::string & table_name, ColumnsDescription /*cached_columns*/, bool /*is_insert_query*/) const
 {
+    std::optional<std::string> table_to_write = std::nullopt;
     auto res = std::make_shared<StorageMerge>(
         StorageID(getDatabaseName(), table_name),
         ColumnsDescription{},
         String{},
         source_database_name_or_regexp,
         database_is_regexp,
-        getSourceDatabasesAndTables(context),
+        source_table_regexp,
+        table_to_write,
+        false,
         context);
 
     res->startup();
     return res;
 }
 
-TableFunctionMerge::TableSet
-TableFunctionMerge::getMatchedTablesWithAccess(const String & database_name, const String & table_regexp, const ContextPtr & context)
-{
-    OptimizedRegularExpression table_re(table_regexp);
-
-    auto table_name_match = [&](const String & table_name) { return table_re.match(table_name); };
-
-    auto access = context->getAccess();
-
-    auto database = DatabaseCatalog::instance().getDatabase(database_name);
-
-    bool granted_show_on_all_tables = access->isGranted(AccessType::SHOW_TABLES, database_name);
-    bool granted_select_on_all_tables = access->isGranted(AccessType::SELECT, database_name);
-
-    TableSet tables;
-
-    for (auto it = database->getTablesIterator(context, table_name_match); it->isValid(); it->next())
-    {
-        if (!it->table())
-            continue;
-        bool granted_show = granted_show_on_all_tables || access->isGranted(AccessType::SHOW_TABLES, database_name, it->name());
-        if (!granted_show)
-            continue;
-        if (!granted_select_on_all_tables)
-            access->checkAccess(AccessType::SELECT, database_name, it->name());
-        tables.emplace(it->name());
-    }
-    return tables;
-}
-
 }
 
 void registerTableFunctionMerge(TableFunctionFactory & factory)
diff --git a/tests/integration/test_table_functions_access_rights/test.py b/tests/integration/test_table_functions_access_rights/test.py
index a4c8017af631..8f646bf43b3e 100644
--- a/tests/integration/test_table_functions_access_rights/test.py
+++ b/tests/integration/test_table_functions_access_rights/test.py
@@ -46,7 +46,7 @@ def test_merge():
     )
 
     instance.query("GRANT CREATE TEMPORARY TABLE ON *.* TO A")
-    assert "no tables in the database matches" in instance.query_and_get_error(
+    assert "no tables satisfied provided regexp" in instance.query_and_get_error(
         select_query, user="A"
     )
 
@@ -63,7 +63,7 @@ def test_merge():
     instance.query("GRANT SELECT ON default.table1 TO A")
     instance.query("GRANT INSERT ON default.table2 TO A")
     assert (
-        "it's necessary to have the grant SELECT ON default.table2"
+        "it's necessary to have the grant SELECT(x) ON default.table2"
         in instance.query_and_get_error(select_query, user="A")
     )
 
diff --git a/tests/queries/0_stateless/01902_table_function_merge_db_params.sql b/tests/queries/0_stateless/01902_table_function_merge_db_params.sql
index 3d97cf2b0c64..caa783211796 100644
--- a/tests/queries/0_stateless/01902_table_function_merge_db_params.sql
+++ b/tests/queries/0_stateless/01902_table_function_merge_db_params.sql
@@ -1,3 +1,5 @@
+-- Tags: no-parallel
+
 DROP DATABASE IF EXISTS 01902_db_params;
 CREATE DATABASE 01902_db_params;
 CREATE TABLE 01902_db_params.t(n Int8) ENGINE=MergeTree ORDER BY n;
@@ -5,7 +7,7 @@ INSERT INTO 01902_db_params.t SELECT * FROM numbers(3);
 SELECT _database, _table, n FROM merge(REGEXP('^01902_db_params'), '^t') ORDER BY _database, _table, n;
 
 SELECT _database, _table, n FROM merge() ORDER BY _database, _table, n; -- {serverError NUMBER_OF_ARGUMENTS_DOESNT_MATCH}
-SELECT _database, _table, n FROM merge('^t') ORDER BY _database, _table, n; -- {serverError BAD_ARGUMENTS}
+SELECT _database, _table, n FROM merge('^t') ORDER BY _database, _table, n; -- {serverError CANNOT_EXTRACT_TABLE_STRUCTURE}
 
 USE 01902_db_params;
 SELECT _database, _table, n FROM merge('^t') ORDER BY _database, _table, n;
diff --git a/tests/queries/0_stateless/01902_table_function_merge_db_repr.reference b/tests/queries/0_stateless/01902_table_function_merge_db_repr.reference
index 4fd27ceec776..fdbb80a73715 100644
--- a/tests/queries/0_stateless/01902_table_function_merge_db_repr.reference
+++ b/tests/queries/0_stateless/01902_table_function_merge_db_repr.reference
@@ -1,195 +1,195 @@
-CREATE TABLE t_merge as 01902_db.t ENGINE=Merge(REGEXP(^01902_db), ^t)
-SELECT _database, _table, n FROM 01902_db.t_merge ORDER BY _database, _table, n
-01902_db	t	0
-01902_db	t	1
-01902_db	t	2
-01902_db	t	3
-01902_db	t	4
-01902_db	t	5
-01902_db	t	6
-01902_db	t	7
-01902_db	t	8
-01902_db	t	9
-01902_db1	t1	0
-01902_db1	t1	1
-01902_db1	t1	2
-01902_db1	t1	3
-01902_db1	t1	4
-01902_db1	t1	5
-01902_db1	t1	6
-01902_db1	t1	7
-01902_db1	t1	8
-01902_db1	t1	9
-01902_db2	t2	0
-01902_db2	t2	1
-01902_db2	t2	2
-01902_db2	t2	3
-01902_db2	t2	4
-01902_db2	t2	5
-01902_db2	t2	6
-01902_db2	t2	7
-01902_db2	t2	8
-01902_db2	t2	9
-01902_db3	t3	0
-01902_db3	t3	1
-01902_db3	t3	2
-01902_db3	t3	3
-01902_db3	t3	4
-01902_db3	t3	5
-01902_db3	t3	6
-01902_db3	t3	7
-01902_db3	t3	8
-01902_db3	t3	9
-SHOW CREATE TABLE 01902_db.t_merge
-CREATE TABLE `01902_db`.t_merge\n(\n    `n` Int8\n)\nENGINE = Merge(REGEXP(\'^01902_db\'), \'^t\')
-SELECT _database, _table, n FROM merge(REGEXP(^01902_db), ^t) ORDER BY _database, _table, n
-01902_db	t	0
-01902_db	t	1
-01902_db	t	2
-01902_db	t	3
-01902_db	t	4
-01902_db	t	5
-01902_db	t	6
-01902_db	t	7
-01902_db	t	8
-01902_db	t	9
-01902_db	t_merge	0
-01902_db	t_merge	0
-01902_db	t_merge	0
-01902_db	t_merge	0
-01902_db	t_merge	1
-01902_db	t_merge	1
-01902_db	t_merge	1
-01902_db	t_merge	1
-01902_db	t_merge	2
-01902_db	t_merge	2
-01902_db	t_merge	2
-01902_db	t_merge	2
-01902_db	t_merge	3
-01902_db	t_merge	3
-01902_db	t_merge	3
-01902_db	t_merge	3
-01902_db	t_merge	4
-01902_db	t_merge	4
-01902_db	t_merge	4
-01902_db	t_merge	4
-01902_db	t_merge	5
-01902_db	t_merge	5
-01902_db	t_merge	5
-01902_db	t_merge	5
-01902_db	t_merge	6
-01902_db	t_merge	6
-01902_db	t_merge	6
-01902_db	t_merge	6
-01902_db	t_merge	7
-01902_db	t_merge	7
-01902_db	t_merge	7
-01902_db	t_merge	7
-01902_db	t_merge	8
-01902_db	t_merge	8
-01902_db	t_merge	8
-01902_db	t_merge	8
-01902_db	t_merge	9
-01902_db	t_merge	9
-01902_db	t_merge	9
-01902_db	t_merge	9
-01902_db1	t1	0
-01902_db1	t1	1
-01902_db1	t1	2
-01902_db1	t1	3
-01902_db1	t1	4
-01902_db1	t1	5
-01902_db1	t1	6
-01902_db1	t1	7
-01902_db1	t1	8
-01902_db1	t1	9
-01902_db2	t2	0
-01902_db2	t2	1
-01902_db2	t2	2
-01902_db2	t2	3
-01902_db2	t2	4
-01902_db2	t2	5
-01902_db2	t2	6
-01902_db2	t2	7
-01902_db2	t2	8
-01902_db2	t2	9
-01902_db3	t3	0
-01902_db3	t3	1
-01902_db3	t3	2
-01902_db3	t3	3
-01902_db3	t3	4
-01902_db3	t3	5
-01902_db3	t3	6
-01902_db3	t3	7
-01902_db3	t3	8
-01902_db3	t3	9
-SELECT _database, _table, n FROM 01902_db.t_merge WHERE _database = 01902_db1 ORDER BY _database, _table, n
-01902_db1	t1	0
-01902_db1	t1	1
-01902_db1	t1	2
-01902_db1	t1	3
-01902_db1	t1	4
-01902_db1	t1	5
-01902_db1	t1	6
-01902_db1	t1	7
-01902_db1	t1	8
-01902_db1	t1	9
-SELECT _database, _table, n FROM 01902_db.t_merge WHERE _table = t1 ORDER BY _database, _table, n
-01902_db1	t1	0
-01902_db1	t1	1
-01902_db1	t1	2
-01902_db1	t1	3
-01902_db1	t1	4
-01902_db1	t1	5
-01902_db1	t1	6
-01902_db1	t1	7
-01902_db1	t1	8
-01902_db1	t1	9
-CREATE TABLE t_merge1 as 01902_db.t ENGINE=Merge(01902_db, ^t$)
-SELECT _database, _table, n FROM 01902_db.t_merge1 ORDER BY _database, _table, n
-01902_db	t	0
-01902_db	t	1
-01902_db	t	2
-01902_db	t	3
-01902_db	t	4
-01902_db	t	5
-01902_db	t	6
-01902_db	t	7
-01902_db	t	8
-01902_db	t	9
-SELECT _database, _table, n FROM merge(01902_db, ^t$) ORDER BY _database, _table, n
-01902_db	t	0
-01902_db	t	1
-01902_db	t	2
-01902_db	t	3
-01902_db	t	4
-01902_db	t	5
-01902_db	t	6
-01902_db	t	7
-01902_db	t	8
-01902_db	t	9
-CREATE TABLE t_merge_1 as 01902_db.t ENGINE=Merge(currentDatabase(), ^t)
-SELECT _database, _table, n FROM 01902_db.t_merge_1 ORDER BY _database, _table, n
-01902_db1	t1	0
-01902_db1	t1	1
-01902_db1	t1	2
-01902_db1	t1	3
-01902_db1	t1	4
-01902_db1	t1	5
-01902_db1	t1	6
-01902_db1	t1	7
-01902_db1	t1	8
-01902_db1	t1	9
-SHOW CREATE TABLE 01902_db.t_merge_1
-CREATE TABLE `01902_db`.t_merge_1\n(\n    `n` Int8\n)\nENGINE = Merge(\'01902_db1\', \'^t\')
+CREATE TABLE t_merge as 01902_db_repr.t ENGINE=Merge(REGEXP(^01902_db_repr), ^t)
+SELECT _database, _table, n FROM 01902_db_repr.t_merge ORDER BY _database, _table, n
+01902_db_repr	t	0
+01902_db_repr	t	1
+01902_db_repr	t	2
+01902_db_repr	t	3
+01902_db_repr	t	4
+01902_db_repr	t	5
+01902_db_repr	t	6
+01902_db_repr	t	7
+01902_db_repr	t	8
+01902_db_repr	t	9
+01902_db_repr1	t1	0
+01902_db_repr1	t1	1
+01902_db_repr1	t1	2
+01902_db_repr1	t1	3
+01902_db_repr1	t1	4
+01902_db_repr1	t1	5
+01902_db_repr1	t1	6
+01902_db_repr1	t1	7
+01902_db_repr1	t1	8
+01902_db_repr1	t1	9
+01902_db_repr2	t2	0
+01902_db_repr2	t2	1
+01902_db_repr2	t2	2
+01902_db_repr2	t2	3
+01902_db_repr2	t2	4
+01902_db_repr2	t2	5
+01902_db_repr2	t2	6
+01902_db_repr2	t2	7
+01902_db_repr2	t2	8
+01902_db_repr2	t2	9
+01902_db_repr3	t3	0
+01902_db_repr3	t3	1
+01902_db_repr3	t3	2
+01902_db_repr3	t3	3
+01902_db_repr3	t3	4
+01902_db_repr3	t3	5
+01902_db_repr3	t3	6
+01902_db_repr3	t3	7
+01902_db_repr3	t3	8
+01902_db_repr3	t3	9
+SHOW CREATE TABLE 01902_db_repr.t_merge
+CREATE TABLE `01902_db_repr`.t_merge\n(\n    `n` Int8\n)\nENGINE = Merge(REGEXP(\'^01902_db_repr\'), \'^t\')
+SELECT _database, _table, n FROM merge(REGEXP(^01902_db_repr), ^t) ORDER BY _database, _table, n
+01902_db_repr	t	0
+01902_db_repr	t	1
+01902_db_repr	t	2
+01902_db_repr	t	3
+01902_db_repr	t	4
+01902_db_repr	t	5
+01902_db_repr	t	6
+01902_db_repr	t	7
+01902_db_repr	t	8
+01902_db_repr	t	9
+01902_db_repr	t_merge	0
+01902_db_repr	t_merge	0
+01902_db_repr	t_merge	0
+01902_db_repr	t_merge	0
+01902_db_repr	t_merge	1
+01902_db_repr	t_merge	1
+01902_db_repr	t_merge	1
+01902_db_repr	t_merge	1
+01902_db_repr	t_merge	2
+01902_db_repr	t_merge	2
+01902_db_repr	t_merge	2
+01902_db_repr	t_merge	2
+01902_db_repr	t_merge	3
+01902_db_repr	t_merge	3
+01902_db_repr	t_merge	3
+01902_db_repr	t_merge	3
+01902_db_repr	t_merge	4
+01902_db_repr	t_merge	4
+01902_db_repr	t_merge	4
+01902_db_repr	t_merge	4
+01902_db_repr	t_merge	5
+01902_db_repr	t_merge	5
+01902_db_repr	t_merge	5
+01902_db_repr	t_merge	5
+01902_db_repr	t_merge	6
+01902_db_repr	t_merge	6
+01902_db_repr	t_merge	6
+01902_db_repr	t_merge	6
+01902_db_repr	t_merge	7
+01902_db_repr	t_merge	7
+01902_db_repr	t_merge	7
+01902_db_repr	t_merge	7
+01902_db_repr	t_merge	8
+01902_db_repr	t_merge	8
+01902_db_repr	t_merge	8
+01902_db_repr	t_merge	8
+01902_db_repr	t_merge	9
+01902_db_repr	t_merge	9
+01902_db_repr	t_merge	9
+01902_db_repr	t_merge	9
+01902_db_repr1	t1	0
+01902_db_repr1	t1	1
+01902_db_repr1	t1	2
+01902_db_repr1	t1	3
+01902_db_repr1	t1	4
+01902_db_repr1	t1	5
+01902_db_repr1	t1	6
+01902_db_repr1	t1	7
+01902_db_repr1	t1	8
+01902_db_repr1	t1	9
+01902_db_repr2	t2	0
+01902_db_repr2	t2	1
+01902_db_repr2	t2	2
+01902_db_repr2	t2	3
+01902_db_repr2	t2	4
+01902_db_repr2	t2	5
+01902_db_repr2	t2	6
+01902_db_repr2	t2	7
+01902_db_repr2	t2	8
+01902_db_repr2	t2	9
+01902_db_repr3	t3	0
+01902_db_repr3	t3	1
+01902_db_repr3	t3	2
+01902_db_repr3	t3	3
+01902_db_repr3	t3	4
+01902_db_repr3	t3	5
+01902_db_repr3	t3	6
+01902_db_repr3	t3	7
+01902_db_repr3	t3	8
+01902_db_repr3	t3	9
+SELECT _database, _table, n FROM 01902_db_repr.t_merge WHERE _database = 01902_db_repr1 ORDER BY _database, _table, n
+01902_db_repr1	t1	0
+01902_db_repr1	t1	1
+01902_db_repr1	t1	2
+01902_db_repr1	t1	3
+01902_db_repr1	t1	4
+01902_db_repr1	t1	5
+01902_db_repr1	t1	6
+01902_db_repr1	t1	7
+01902_db_repr1	t1	8
+01902_db_repr1	t1	9
+SELECT _database, _table, n FROM 01902_db_repr.t_merge WHERE _table = t1 ORDER BY _database, _table, n
+01902_db_repr1	t1	0
+01902_db_repr1	t1	1
+01902_db_repr1	t1	2
+01902_db_repr1	t1	3
+01902_db_repr1	t1	4
+01902_db_repr1	t1	5
+01902_db_repr1	t1	6
+01902_db_repr1	t1	7
+01902_db_repr1	t1	8
+01902_db_repr1	t1	9
+CREATE TABLE t_merge1 as 01902_db_repr.t ENGINE=Merge(01902_db_repr, ^t$)
+SELECT _database, _table, n FROM 01902_db_repr.t_merge1 ORDER BY _database, _table, n
+01902_db_repr	t	0
+01902_db_repr	t	1
+01902_db_repr	t	2
+01902_db_repr	t	3
+01902_db_repr	t	4
+01902_db_repr	t	5
+01902_db_repr	t	6
+01902_db_repr	t	7
+01902_db_repr	t	8
+01902_db_repr	t	9
+SELECT _database, _table, n FROM merge(01902_db_repr, ^t$) ORDER BY _database, _table, n
+01902_db_repr	t	0
+01902_db_repr	t	1
+01902_db_repr	t	2
+01902_db_repr	t	3
+01902_db_repr	t	4
+01902_db_repr	t	5
+01902_db_repr	t	6
+01902_db_repr	t	7
+01902_db_repr	t	8
+01902_db_repr	t	9
+CREATE TABLE t_merge_1 as 01902_db_repr.t ENGINE=Merge(currentDatabase(), ^t)
+SELECT _database, _table, n FROM 01902_db_repr.t_merge_1 ORDER BY _database, _table, n
+01902_db_repr1	t1	0
+01902_db_repr1	t1	1
+01902_db_repr1	t1	2
+01902_db_repr1	t1	3
+01902_db_repr1	t1	4
+01902_db_repr1	t1	5
+01902_db_repr1	t1	6
+01902_db_repr1	t1	7
+01902_db_repr1	t1	8
+01902_db_repr1	t1	9
+SHOW CREATE TABLE 01902_db_repr.t_merge_1
+CREATE TABLE `01902_db_repr`.t_merge_1\n(\n    `n` Int8\n)\nENGINE = Merge(\'01902_db_repr1\', \'^t\')
 SELECT _database, _table, n FROM merge(currentDatabase(), ^t) ORDER BY _database, _table, n
-01902_db1	t1	0
-01902_db1	t1	1
-01902_db1	t1	2
-01902_db1	t1	3
-01902_db1	t1	4
-01902_db1	t1	5
-01902_db1	t1	6
-01902_db1	t1	7
-01902_db1	t1	8
-01902_db1	t1	9
+01902_db_repr1	t1	0
+01902_db_repr1	t1	1
+01902_db_repr1	t1	2
+01902_db_repr1	t1	3
+01902_db_repr1	t1	4
+01902_db_repr1	t1	5
+01902_db_repr1	t1	6
+01902_db_repr1	t1	7
+01902_db_repr1	t1	8
+01902_db_repr1	t1	9
diff --git a/tests/queries/0_stateless/01902_table_function_merge_db_repr.sql b/tests/queries/0_stateless/01902_table_function_merge_db_repr.sql
index ee6f052d6940..660082f325f7 100644
--- a/tests/queries/0_stateless/01902_table_function_merge_db_repr.sql
+++ b/tests/queries/0_stateless/01902_table_function_merge_db_repr.sql
@@ -1,73 +1,73 @@
 -- Tags: no-parallel
 
-DROP DATABASE IF EXISTS 01902_db;
-DROP DATABASE IF EXISTS 01902_db1;
-DROP DATABASE IF EXISTS 01902_db2;
-DROP DATABASE IF EXISTS 01902_db3;
+DROP DATABASE IF EXISTS 01902_db_repr;
+DROP DATABASE IF EXISTS 01902_db_repr1;
+DROP DATABASE IF EXISTS 01902_db_repr2;
+DROP DATABASE IF EXISTS 01902_db_repr3;
 
-CREATE DATABASE 01902_db;
-CREATE DATABASE 01902_db1;
-CREATE DATABASE 01902_db2;
-CREATE DATABASE 01902_db3;
+CREATE DATABASE 01902_db_repr;
+CREATE DATABASE 01902_db_repr1;
+CREATE DATABASE 01902_db_repr2;
+CREATE DATABASE 01902_db_repr3;
 
-CREATE TABLE 01902_db.t   (n Int8) ENGINE=MergeTree ORDER BY n;
-CREATE TABLE 01902_db1.t1 (n Int8) ENGINE=MergeTree ORDER BY n;
-CREATE TABLE 01902_db2.t2 (n Int8) ENGINE=MergeTree ORDER BY n;
-CREATE TABLE 01902_db3.t3 (n Int8) ENGINE=MergeTree ORDER BY n;
+CREATE TABLE 01902_db_repr.t   (n Int8) ENGINE=MergeTree ORDER BY n;
+CREATE TABLE 01902_db_repr1.t1 (n Int8) ENGINE=MergeTree ORDER BY n;
+CREATE TABLE 01902_db_repr2.t2 (n Int8) ENGINE=MergeTree ORDER BY n;
+CREATE TABLE 01902_db_repr3.t3 (n Int8) ENGINE=MergeTree ORDER BY n;
 
-INSERT INTO 01902_db.t   SELECT * FROM numbers(10);
-INSERT INTO 01902_db1.t1 SELECT * FROM numbers(10);
-INSERT INTO 01902_db2.t2 SELECT * FROM numbers(10);
-INSERT INTO 01902_db3.t3 SELECT * FROM numbers(10);
+INSERT INTO 01902_db_repr.t   SELECT * FROM numbers(10);
+INSERT INTO 01902_db_repr1.t1 SELECT * FROM numbers(10);
+INSERT INTO 01902_db_repr2.t2 SELECT * FROM numbers(10);
+INSERT INTO 01902_db_repr3.t3 SELECT * FROM numbers(10);
 
-SELECT 'CREATE TABLE t_merge as 01902_db.t ENGINE=Merge(REGEXP(^01902_db), ^t)';
-CREATE TABLE 01902_db.t_merge as 01902_db.t ENGINE=Merge(REGEXP('^01902_db'), '^t');
+SELECT 'CREATE TABLE t_merge as 01902_db_repr.t ENGINE=Merge(REGEXP(^01902_db_repr), ^t)';
+CREATE TABLE 01902_db_repr.t_merge as 01902_db_repr.t ENGINE=Merge(REGEXP('^01902_db_repr'), '^t');
 
-SELECT 'SELECT _database, _table, n FROM 01902_db.t_merge ORDER BY _database, _table, n';
-SELECT _database, _table, n FROM 01902_db.t_merge ORDER BY _database, _table, n;
+SELECT 'SELECT _database, _table, n FROM 01902_db_repr.t_merge ORDER BY _database, _table, n';
+SELECT _database, _table, n FROM 01902_db_repr.t_merge ORDER BY _database, _table, n;
 
-SELECT 'SHOW CREATE TABLE 01902_db.t_merge';
-SHOW CREATE  TABLE 01902_db.t_merge;
+SELECT 'SHOW CREATE TABLE 01902_db_repr.t_merge';
+SHOW CREATE  TABLE 01902_db_repr.t_merge;
 
-SELECT 'SELECT _database, _table, n FROM merge(REGEXP(^01902_db), ^t) ORDER BY _database, _table, n';
-SELECT _database, _table, n FROM merge(REGEXP('^01902_db'), '^t') ORDER BY _database, _table, n;
+SELECT 'SELECT _database, _table, n FROM merge(REGEXP(^01902_db_repr), ^t) ORDER BY _database, _table, n';
+SELECT _database, _table, n FROM merge(REGEXP('^01902_db_repr'), '^t') ORDER BY _database, _table, n;
 
-SELECT 'SELECT _database, _table, n FROM 01902_db.t_merge WHERE _database = 01902_db1 ORDER BY _database, _table, n';
-SELECT _database, _table, n FROM 01902_db.t_merge WHERE _database = '01902_db1' ORDER BY _database, _table, n;
+SELECT 'SELECT _database, _table, n FROM 01902_db_repr.t_merge WHERE _database = 01902_db_repr1 ORDER BY _database, _table, n';
+SELECT _database, _table, n FROM 01902_db_repr.t_merge WHERE _database = '01902_db_repr1' ORDER BY _database, _table, n;
 
-SELECT 'SELECT _database, _table, n FROM 01902_db.t_merge WHERE _table = t1 ORDER BY _database, _table, n';
-SELECT _database, _table, n FROM 01902_db.t_merge WHERE _table = 't1' ORDER BY _database, _table, n;
+SELECT 'SELECT _database, _table, n FROM 01902_db_repr.t_merge WHERE _table = t1 ORDER BY _database, _table, n';
+SELECT _database, _table, n FROM 01902_db_repr.t_merge WHERE _table = 't1' ORDER BY _database, _table, n;
 
 -- not regexp
-SELECT 'CREATE TABLE t_merge1 as 01902_db.t ENGINE=Merge(01902_db, ^t$)';
-CREATE TABLE 01902_db.t_merge1 as 01902_db.t ENGINE=Merge('01902_db', '^t$');
+SELECT 'CREATE TABLE t_merge1 as 01902_db_repr.t ENGINE=Merge(01902_db_repr, ^t$)';
+CREATE TABLE 01902_db_repr.t_merge1 as 01902_db_repr.t ENGINE=Merge('01902_db_repr', '^t$');
 
-SELECT 'SELECT _database, _table, n FROM 01902_db.t_merge1 ORDER BY _database, _table, n';
-SELECT _database, _table, n FROM 01902_db.t_merge1 ORDER BY _database, _table, n;
+SELECT 'SELECT _database, _table, n FROM 01902_db_repr.t_merge1 ORDER BY _database, _table, n';
+SELECT _database, _table, n FROM 01902_db_repr.t_merge1 ORDER BY _database, _table, n;
 
-SELECT 'SELECT _database, _table, n FROM merge(01902_db, ^t$) ORDER BY _database, _table, n';
-SELECT _database, _table, n FROM merge('01902_db', '^t$') ORDER BY _database, _table, n;
+SELECT 'SELECT _database, _table, n FROM merge(01902_db_repr, ^t$) ORDER BY _database, _table, n';
+SELECT _database, _table, n FROM merge('01902_db_repr', '^t$') ORDER BY _database, _table, n;
 
-USE 01902_db1;
+USE 01902_db_repr1;
 
-SELECT 'CREATE TABLE t_merge_1 as 01902_db.t ENGINE=Merge(currentDatabase(), ^t)';
-CREATE TABLE 01902_db.t_merge_1 as 01902_db.t ENGINE=Merge(currentDatabase(), '^t');
+SELECT 'CREATE TABLE t_merge_1 as 01902_db_repr.t ENGINE=Merge(currentDatabase(), ^t)';
+CREATE TABLE 01902_db_repr.t_merge_1 as 01902_db_repr.t ENGINE=Merge(currentDatabase(), '^t');
 
-SELECT 'SELECT _database, _table, n FROM 01902_db.t_merge_1 ORDER BY _database, _table, n';
-SELECT _database, _table, n FROM 01902_db.t_merge_1 ORDER BY _database, _table, n;
+SELECT 'SELECT _database, _table, n FROM 01902_db_repr.t_merge_1 ORDER BY _database, _table, n';
+SELECT _database, _table, n FROM 01902_db_repr.t_merge_1 ORDER BY _database, _table, n;
 
-SELECT 'SHOW CREATE TABLE 01902_db.t_merge_1';
-SHOW CREATE TABLE 01902_db.t_merge_1;
+SELECT 'SHOW CREATE TABLE 01902_db_repr.t_merge_1';
+SHOW CREATE TABLE 01902_db_repr.t_merge_1;
 
 SELECT 'SELECT _database, _table, n FROM merge(currentDatabase(), ^t) ORDER BY _database, _table, n';
 SELECT _database, _table, n FROM merge(currentDatabase(), '^t') ORDER BY _database, _table, n;
 
 --fuzzed LOGICAL_ERROR
-CREATE TABLE 01902_db.t4 (n Date) ENGINE=MergeTree ORDER BY n;
-INSERT INTO 01902_db.t4   SELECT * FROM numbers(10);
-SELECT NULL FROM 01902_db.t_merge WHERE n ORDER BY _table DESC; -- {serverError ILLEGAL_TYPE_OF_COLUMN_FOR_FILTER}
-
-DROP DATABASE 01902_db;
-DROP DATABASE 01902_db1;
-DROP DATABASE 01902_db2;
-DROP DATABASE 01902_db3;
+CREATE TABLE 01902_db_repr.t4 (n Date) ENGINE=MergeTree ORDER BY n;
+INSERT INTO 01902_db_repr.t4   SELECT * FROM numbers(10);
+SELECT NULL FROM 01902_db_repr.t_merge WHERE n ORDER BY _table DESC; -- {serverError ILLEGAL_TYPE_OF_COLUMN_FOR_FILTER}
+
+DROP DATABASE 01902_db_repr;
+DROP DATABASE 01902_db_repr1;
+DROP DATABASE 01902_db_repr2;
+DROP DATABASE 01902_db_repr3;
diff --git a/tests/queries/0_stateless/03302_merge_table_structure_unification.reference b/tests/queries/0_stateless/03302_merge_table_structure_unification.reference
index 15565cb3091d..7eb6877201ae 100644
--- a/tests/queries/0_stateless/03302_merge_table_structure_unification.reference
+++ b/tests/queries/0_stateless/03302_merge_table_structure_unification.reference
@@ -1,5 +1,6 @@
 -1		['Goodbye']	2025-01-01 02:03:04
 1	Hello	['World']	1970-01-01 00:00:00
+--- table function
 a	Int32					
 b	String					
 c	Array(Nullable(String))					
@@ -8,8 +9,10 @@ d	DateTime(\'UTC\')	DEFAULT	now()
 -1		['Goodbye']	2025-01-01 02:03:04
 1	Hello	['World']	1970-01-01 00:00:00
 1	Hello	['World']	1970-01-01 00:00:00
--1		['Goodbye']	2025-01-01 02:03:04
-1	Hello	['World']	1970-01-01 00:00:00
+--- merge_table_max_tables_to_look_for_schema_inference = 1
+1	Hello	['World']
+255		['Goodbye']
+--- table function
 a	UInt8					
 b	String					
 c	Array(String)					
diff --git a/tests/queries/0_stateless/03302_merge_table_structure_unification.sql b/tests/queries/0_stateless/03302_merge_table_structure_unification.sql
index b6548b21527c..41bfa8dc2bef 100644
--- a/tests/queries/0_stateless/03302_merge_table_structure_unification.sql
+++ b/tests/queries/0_stateless/03302_merge_table_structure_unification.sql
@@ -26,6 +26,7 @@ CREATE TABLE test_merge ENGINE = Merge(currentDatabase(), '^test_');
 -- TODO: defaults are not calculated
 SELECT * FROM test_merge ORDER BY a;
 
+SELECT '--- table function';
 DESCRIBE merge('^test_');
 
 -- Note that this will also pick up the test_merge table, duplicating the results
@@ -37,8 +38,10 @@ SET merge_table_max_tables_to_look_for_schema_inference = 1;
 
 CREATE TABLE test_merge ENGINE = Merge(currentDatabase(), '^test_');
 
+SELECT '--- merge_table_max_tables_to_look_for_schema_inference = 1';
 SELECT * FROM test_merge ORDER BY a;
 
+SELECT '--- table function';
 DESCRIBE merge('^test_');
 
 SELECT * FROM merge('^test_') ORDER BY a;
diff --git a/tests/queries/0_stateless/03373_write_to_merge_table.reference b/tests/queries/0_stateless/03373_write_to_merge_table.reference
new file mode 100644
index 000000000000..0ecbe836033f
--- /dev/null
+++ b/tests/queries/0_stateless/03373_write_to_merge_table.reference
@@ -0,0 +1,44 @@
+2	1
+2	2
+2	3
+1	1
+2	1
+2	2
+2	3
+1	1
+2	1
+2	2
+2	3
+1	1
+2	1
+2	2
+2	3
+1	1
+2	1
+2	2
+2	3
+1	1
+2	1
+2	2
+2	3
+4
+2	1
+2	2
+2	3
+3	1
+4
+2	1
+2	2
+2	3
+3	1
+1
+3	2
+4
+2	1
+2	2
+2	3
+3	1
+0
+2
+3	2
+3	3
diff --git a/tests/queries/0_stateless/03373_write_to_merge_table.sql b/tests/queries/0_stateless/03373_write_to_merge_table.sql
new file mode 100644
index 000000000000..1bbbf2203a22
--- /dev/null
+++ b/tests/queries/0_stateless/03373_write_to_merge_table.sql
@@ -0,0 +1,74 @@
+-- Tags: no-parallel
+
+DROP TABLE IF EXISTS test03373_db.test03373_table_1;
+DROP TABLE IF EXISTS test03373_db.test03373_table_2;
+DROP TABLE IF EXISTS test03373_db.test03373_table_3;
+DROP TABLE IF EXISTS test03373_db.test03373_table_4;
+DROP TABLE IF EXISTS test03373_db.test03373_merge_ro;
+DROP TABLE IF EXISTS test03373_db.test03373_merge_wr_1;
+DROP TABLE IF EXISTS test03373_db.test03373_merge_wr_2;
+DROP TABLE IF EXISTS test03373_db.test03373_merge_wr_3;
+DROP TABLE IF EXISTS test03373_db.test03373_merge_wr_auto;
+DROP DATABASE IF EXISTS test03373_db;
+
+CREATE DATABASE test03373_db;
+
+CREATE TABLE test03373_db.test03373_table_1 (key UInt32, value UInt32) ENGINE=MergeTree() ORDER BY key;
+CREATE TABLE test03373_db.test03373_table_2 (key UInt32, value UInt32) ENGINE=MergeTree() ORDER BY key;
+
+CREATE TABLE test03373_db.test03373_merge_ro (key UInt32, value UInt32) ENGINE=Merge(test03373_db, 'test03373_table_\d+');
+
+CREATE TABLE test03373_db.test03373_merge_wr_1 (key UInt32, value UInt32) ENGINE=Merge(test03373_db, 'test03373_table_\d+', test03373_table_2);
+CREATE TABLE test03373_db.test03373_merge_wr_2 (key UInt32, value UInt32) ENGINE=Merge(test03373_db, 'test03373_table_\d+', test03373_db.test03373_table_2);
+CREATE TABLE test03373_db.test03373_merge_wr_3 (key UInt32, value UInt32) ENGINE=Merge(REGEXP('test03373_.*'), 'test03373_table_\d+', test03373_db.test03373_table_2);
+
+CREATE TABLE test03373_db.test03373_merge_wr_auto (key UInt32, value UInt32) ENGINE=Merge(test03373_db, 'test03373_table_\d+', auto);
+
+INSERT INTO test03373_db.test03373_table_1 VALUES (1,1);
+
+INSERT INTO test03373_db.test03373_merge_wr_1 VALUES (2,1);
+INSERT INTO test03373_db.test03373_merge_wr_2 VALUES (2,2);
+INSERT INTO test03373_db.test03373_merge_wr_3 VALUES (2,3);
+
+SELECT * FROM test03373_db.test03373_table_2 ORDER BY key, value;
+
+SELECT * FROM test03373_db.test03373_merge_ro ORDER BY key, value;
+SELECT * FROM test03373_db.test03373_merge_wr_1 ORDER BY key, value;
+SELECT * FROM test03373_db.test03373_merge_wr_2 ORDER BY key, value;
+SELECT * FROM test03373_db.test03373_merge_wr_3 ORDER BY key, value;
+
+SELECT * FROM test03373_db.test03373_merge_wr_auto ORDER BY key, value;
+
+-- insert into test03373_table_2
+INSERT INTO test03373_db.test03373_merge_wr_auto VALUES (3,1);
+SELECT count() FROM test03373_db.test03373_table_2;
+SELECT * FROM test03373_db.test03373_table_2 ORDER BY key, value;
+
+CREATE TABLE test03373_db.test03373_table_4 (key UInt32, value UInt32) ENGINE=MergeTree() ORDER BY key;
+-- insert into test03373_table_4
+INSERT INTO test03373_db.test03373_merge_wr_auto VALUES (3,2);
+SELECT count() FROM test03373_db.test03373_table_2;
+SELECT * FROM test03373_db.test03373_table_2 ORDER BY key, value;
+SELECT count() FROM test03373_db.test03373_table_4;
+SELECT * FROM test03373_db.test03373_table_4 ORDER BY key, value;
+
+CREATE TABLE test03373_db.test03373_table_3 (key UInt32, value UInt32) ENGINE=MergeTree() ORDER BY key;
+-- insert into test03373_table_4
+INSERT INTO test03373_db.test03373_merge_wr_auto VALUES (3,3);
+SELECT count() FROM test03373_db.test03373_table_2;
+SELECT * FROM test03373_db.test03373_table_2 ORDER BY key, value;
+SELECT count() FROM test03373_db.test03373_table_3;
+SELECT * FROM test03373_db.test03373_table_3 ORDER BY key, value;
+SELECT count() FROM test03373_db.test03373_table_4;
+SELECT * FROM test03373_db.test03373_table_4 ORDER BY key, value;
+
+DROP TABLE IF EXISTS test03373_db.test03373_table_1;
+DROP TABLE IF EXISTS test03373_db.test03373_table_2;
+DROP TABLE IF EXISTS test03373_db.test03373_table_3;
+DROP TABLE IF EXISTS test03373_db.test03373_table_4;
+DROP TABLE IF EXISTS test03373_db.test03373_merge_ro;
+DROP TABLE IF EXISTS test03373_db.test03373_merge_wr_1;
+DROP TABLE IF EXISTS test03373_db.test03373_merge_wr_2;
+DROP TABLE IF EXISTS test03373_db.test03373_merge_wr_3;
+DROP TABLE IF EXISTS test03373_db.test03373_merge_wr_auto;
+DROP DATABASE IF EXISTS test03373_db;
diff --git a/tests/queries/0_stateless/03400_merge_storage_underlying_tables_access_validation.reference b/tests/queries/0_stateless/03400_merge_storage_underlying_tables_access_validation.reference
new file mode 100644
index 000000000000..1a74a5721120
--- /dev/null
+++ b/tests/queries/0_stateless/03400_merge_storage_underlying_tables_access_validation.reference
@@ -0,0 +1,142 @@
+allowed	0	0	0
+allowed	1	1	0
+allowed	2	2	0
+allowed	3	3	0
+allowed	4	4	0
+allowed	5	5	0
+allowed	6	6	0
+allowed	7	7	0
+allowed	8	8	0
+allowed	9	9	0
+partial_allowed	10	0	0
+partial_allowed	11	1	0
+partial_allowed	12	2	0
+partial_allowed	13	3	0
+partial_allowed	14	4	0
+partial_allowed	15	5	0
+partial_allowed	16	6	0
+partial_allowed	17	7	0
+partial_allowed	18	8	0
+partial_allowed	19	9	0
+not_allowed	20	0	0
+not_allowed	21	1	0
+not_allowed	22	2	0
+not_allowed	23	3	0
+not_allowed	24	4	0
+not_allowed	25	5	0
+not_allowed	26	6	0
+not_allowed	27	7	0
+not_allowed	28	8	0
+not_allowed	29	9	0
+no_show_allowed	30	0	0
+no_show_allowed	31	1	1
+no_show_allowed	32	2	2
+no_show_allowed	33	3	3
+no_show_allowed	34	4	4
+no_show_allowed	35	5	5
+no_show_allowed	36	6	6
+no_show_allowed	37	7	7
+no_show_allowed	38	8	8
+no_show_allowed	39	9	9
+----Table engine
+----select allowed columns and databases
+allowed	0
+allowed	1
+allowed	2
+allowed	3
+allowed	4
+allowed	5
+allowed	6
+allowed	7
+allowed	8
+allowed	9
+partial_allowed	10
+partial_allowed	11
+partial_allowed	12
+partial_allowed	13
+partial_allowed	14
+partial_allowed	15
+partial_allowed	16
+partial_allowed	17
+partial_allowed	18
+partial_allowed	19
+----
+allowed	0	0
+allowed	1	1
+allowed	2	2
+allowed	3	3
+allowed	4	4
+allowed	5	5
+allowed	6	6
+allowed	7	7
+allowed	8	8
+allowed	9	9
+----
+allowed	0
+allowed	1
+allowed	2
+allowed	3
+allowed	4
+allowed	5
+allowed	6
+allowed	7
+allowed	8
+allowed	9
+----create without show all
+a
+b
+----select user created table
+allowed	0	0
+allowed	1	1
+allowed	2	2
+allowed	3	3
+allowed	4	4
+allowed	5	5
+allowed	6	6
+allowed	7	7
+allowed	8	8
+allowed	9	9
+----Table function
+----select allowed columns and databases
+allowed	0
+allowed	1
+allowed	2
+allowed	3
+allowed	4
+allowed	5
+allowed	6
+allowed	7
+allowed	8
+allowed	9
+partial_allowed	10
+partial_allowed	11
+partial_allowed	12
+partial_allowed	13
+partial_allowed	14
+partial_allowed	15
+partial_allowed	16
+partial_allowed	17
+partial_allowed	18
+partial_allowed	19
+----
+allowed	0	0
+allowed	1	1
+allowed	2	2
+allowed	3	3
+allowed	4	4
+allowed	5	5
+allowed	6	6
+allowed	7	7
+allowed	8	8
+allowed	9	9
+----
+allowed	0
+allowed	1
+allowed	2
+allowed	3
+allowed	4
+allowed	5
+allowed	6
+allowed	7
+allowed	8
+allowed	9
diff --git a/tests/queries/0_stateless/03400_merge_storage_underlying_tables_access_validation.sh b/tests/queries/0_stateless/03400_merge_storage_underlying_tables_access_validation.sh
new file mode 100755
index 000000000000..f165c923719f
--- /dev/null
+++ b/tests/queries/0_stateless/03400_merge_storage_underlying_tables_access_validation.sh
@@ -0,0 +1,72 @@
+#!/usr/bin/env bash
+# Tags: no-parallel
+# Tag no-parallel: create user
+
+CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+# shellcheck source=../shell_config.sh
+. "$CURDIR"/../shell_config.sh
+
+${CLICKHOUSE_CLIENT} --multiline -q """
+DROP DATABASE IF EXISTS database_03400;
+CREATE DATABASE database_03400;
+CREATE TABLE database_03400.allowed (a Int64, b Int64) Engine=MergeTree ORDER BY a;
+CREATE TABLE database_03400.partial_allowed (a Int64, b Int64) Engine=MergeTree ORDER BY a;
+CREATE TABLE database_03400.not_allowed (a Int64, b Int64) Engine=MergeTree ORDER BY a;
+CREATE TABLE database_03400.no_show_allowed (a Int64, b Int64, c Int64) Engine=MergeTree ORDER BY a;
+
+INSERT INTO database_03400.allowed SELECT number, number FROM numbers(10);
+INSERT INTO database_03400.partial_allowed SELECT number + 10, number FROM numbers(10);
+INSERT INTO database_03400.not_allowed SELECT number + 20, number FROM numbers(10);
+INSERT INTO database_03400.no_show_allowed SELECT number + 30, number, number FROM numbers(10);
+
+CREATE TABLE database_03400.merge Engine=Merge(database_03400, '.*allowed');
+SELECT _table, * FROM database_03400.merge ORDER BY a SETTINGS enable_analyzer = 1;
+
+DROP USER IF EXISTS user_test_03400;
+CREATE USER user_test_03400 IDENTIFIED WITH plaintext_password BY 'user_test_03400';
+
+GRANT TABLE ENGINE ON Merge TO 'user_test_03400';
+GRANT CREATE TABLE ON database_03400.* TO 'user_test_03400';
+GRANT SHOW ON database_03400.* TO 'user_test_03400';
+REVOKE ALL ON database_03400.no_show_allowed FROM 'user_test_03400';
+GRANT SELECT ON database_03400.allowed TO 'user_test_03400';
+GRANT SELECT(a) ON database_03400.partial_allowed TO 'user_test_03400';
+GRANT SELECT ON database_03400.merge* TO 'user_test_03400';
+"""
+
+echo "----Table engine"
+# access from the Merge table
+$CLICKHOUSE_CLIENT --multiline --user user_test_03400 --password user_test_03400 -q """
+SELECT * FROM database_03400.merge; -- { serverError ACCESS_DENIED }
+SELECT a FROM database_03400.merge; -- { serverError ACCESS_DENIED }
+SELECT '----select allowed columns and databases';
+SELECT _table, a FROM database_03400.merge WHERE _database='database_03400' AND _table IN ('allowed', 'partial_allowed')  ORDER BY a;
+SELECT '----';
+SELECT _table, a, b FROM database_03400.merge WHERE _database='database_03400' AND _table = 'allowed'  ORDER BY a;
+SELECT '----';
+SELECT _table, a FROM database_03400.merge WHERE _database='database_03400' AND _table IN ('allowed', 'no_show_allowed')  ORDER BY a;
+SELECT '----create without show all';
+CREATE TABLE database_03400.merge_user Engine=Merge(database_03400, '.*allowed');
+SELECT name FROM system.columns WHERE database='database_03400' AND table='merge_user' ORDER BY name;
+SELECT '----select user created table';
+SELECT _table, * FROM database_03400.merge_user WHERE _table = 'allowed' ORDER BY a;
+CREATE TABLE database_03400.merge_user_fail Engine=Merge(database_03400, 'no_show_allowed'); -- { serverError CANNOT_EXTRACT_TABLE_STRUCTURE }
+"""
+
+echo "----Table function"
+# access from the Merge table function
+$CLICKHOUSE_CLIENT --multiline --user user_test_03400 --password user_test_03400 -q """
+SELECT * FROM merge(database_03400, '.*allowed'); -- { serverError ACCESS_DENIED }
+SELECT a FROM merge(database_03400, '.*allowed'); -- { serverError ACCESS_DENIED }
+SELECT '----select allowed columns and databases';
+SELECT _table, a FROM merge(database_03400, '.*allowed') WHERE _table IN ('allowed', 'partial_allowed')  ORDER BY a;
+SELECT '----';
+SELECT _table, a, b FROM merge(database_03400, '.*allowed') WHERE _database='database_03400' AND _table = 'allowed'  ORDER BY a;
+SELECT '----';
+SELECT _table, a FROM merge(database_03400, '.*allowed') WHERE _table IN ('allowed', 'no_show_allowed')  ORDER BY a;
+"""
+
+${CLICKHOUSE_CLIENT} --multiline -q """
+DROP DATABASE IF EXISTS database_03400;
+DROP USER IF EXISTS user_test_03400;
+"""