Rework fix: scope dedup to items with different non-empty aliases

Revert the `identity()` wrapper approach in `StorageDistributed.cpp`
(it cannot handle 3+ duplicates or existing `identity()` projections).

Instead, fix in `PlannerActionsVisitorImpl::visit` with a precise
condition: add a DAG ALIAS node only when a later list item resolves to
the same DAG node as an earlier item AND the two items carry different
non-empty query-tree aliases.  This targets the distributed ALIAS column
case (where `ReplaseAliasColumnsVisitor` inlines identical expressions
under different aliases like "a1" and "a2") without affecting:

- `arrayJoin` duplicates (second item has no alias → skipped)
- `SELECT x, x` (column refs have no alias on the tree node → skipped)
- `SELECT * REPLACE(a AS b)` producing same-alias duplicates (same
  alias → condition false → skipped)
- JOIN projections with duplicate columns (same alias → skipped)

The generated alias name incorporates the query-tree alias for
readability (e.g. `toString(__table1.x)_a2`) and loops with a suffix
until fresh in the scope to avoid collisions.

Co-authored-by: Yakov Olkhovskiy <[email protected]>

src/Planner/PlannerActionsVisitor.cpp

@@ -752,10 +752,37 @@ std::pair<ActionsDAG::NodeRawConstPtrs, CorrelatedSubtrees> PlannerActionsVisito
 
     if (auto * expression_list_node = expression_node->as<ListNode>())
     {
-        for (auto & node : expression_list_node->getNodes())
+        auto & list_nodes = expression_list_node->getNodes();
+
+        for (size_t i = 0; i < list_nodes.size(); ++i)
         {
-            auto [node_name, _] = visitImpl(node);
-            result.push_back(actions_stack.front().getNodeOrThrow(node_name));
+            auto [node_name, _] = visitImpl(list_nodes[i]);
+            const auto * dag_node = actions_stack.front().getNodeOrThrow(node_name);
+
+            /// When two list items with different aliases resolve to the same DAG node
+            /// (e.g. two ALIAS columns expanding to the same expression in a Distributed
+            /// query), downstream code deduplicates outputs by name, shrinking the column
+            /// count.  Add an ALIAS node with a fresh name for the later occurrence so
+            /// that each item produces a distinct output.  Only do this when both items
+            /// carry different non-empty aliases — this avoids interfering with arrayJoin
+            /// or intentional duplicates like SELECT x, x.
+            const auto & current_alias = list_nodes[i]->getAlias();
+            if (!current_alias.empty())
+            {
+                for (size_t j = 0; j < i; ++j)
+                {
+                    if (result[j] == dag_node && list_nodes[j]->getAlias() != current_alias)
+                    {
+                        String unique_name = node_name + "_" + current_alias;
+                        while (actions_stack.front().containsNode(unique_name))
+                            unique_name += "_";
+                        dag_node = actions_stack.front().addAliasIfNecessary(unique_name, dag_node);
+                        break;
+                    }
+                }
+            }
+
+            result.push_back(dag_node);
         }
     }
     else

src/Storages/StorageDistributed.cpp

@@ -52,7 +52,6 @@
 
 #include <Analyzer/ColumnNode.h>
 #include <Analyzer/FunctionNode.h>
-#include <Functions/FunctionFactory.h>
 #include <Analyzer/TableNode.h>
 #include <Analyzer/TableFunctionNode.h>
 #include <Analyzer/QueryNode.h>
@@ -936,36 +935,6 @@ QueryTreeNodePtr buildQueryTreeDistributed(SelectQueryInfo & query_info,
     ReplaseAliasColumnsVisitor replace_alias_columns_visitor;
     replace_alias_columns_visitor.visit(query_tree_to_modify);
 
-    /// After inlining ALIAS columns, the projection may contain structurally identical
-    /// expressions (e.g. two ALIAS columns both defined as `toString(x)`).  The planner
-    /// assigns action-node names by expression structure, so duplicates get the same name
-    /// and are later deduplicated — shrinking the column count and causing
-    /// NUMBER_OF_COLUMNS_DOESNT_MATCH on the initiator.  Wrap subsequent duplicates in
-    /// identity() to make their action-node names unique.
-    auto * query_node_ptr = query_tree_to_modify->as<QueryNode>();
-    if (query_node_ptr)
-    {
-        auto & projection_nodes = query_node_ptr->getProjection().getNodes();
-        struct TreeHashHash { size_t operator()(const IQueryTreeNode::Hash & h) const { return h.low64 ^ h.high64; } };
-        std::unordered_set<IQueryTreeNode::Hash, TreeHashHash> seen_hashes;
-        auto identity_resolver = FunctionFactory::instance().get("identity", query_context);
-
-        for (auto & proj_node : projection_nodes)
-        {
-            auto tree_hash = proj_node->getTreeHash({.compare_aliases = false});
-            while (!seen_hashes.emplace(tree_hash).second)
-            {
-                auto alias = proj_node->getAlias();
-                auto wrapper = std::make_shared<FunctionNode>("identity");
-                wrapper->getArguments().getNodes().push_back(std::move(proj_node));
-                wrapper->resolveAsFunction(identity_resolver);
-                wrapper->setAlias(alias);
-                proj_node = std::move(wrapper);
-                tree_hash = proj_node->getTreeHash({.compare_aliases = false});
-            }
-        }
-    }
-
     const auto & settings = query_context->getSettingsRef();
 
     if (settings[Setting::prefer_global_in_and_join])