restore cropped names in JOIN ON section for distributed queries

dbms/src/Interpreters/TranslateQualifiedNamesVisitor.cpp

@@ -267,4 +267,14 @@ void TranslateQualifiedNamesMatcher::extractJoinUsingColumns(const ASTPtr ast, D
     }
 }
 
+void RestoreQualifiedNamesData::visit(ASTIdentifier & identifier, ASTPtr & ast)
+{
+    if (IdentifierSemantic::getColumnName(identifier) &&
+        IdentifierSemantic::getMembership(identifier))
+    {
+        ast = identifier.clone();
+        ast->as<ASTIdentifier>()->restoreCompoundName();
+    }
+}
+
 }

dbms/src/Interpreters/TranslateQualifiedNamesVisitor.h

@@ -66,4 +66,15 @@ private:
 /// It finds columns and translate their names to the normal form. Expand asterisks and qualified asterisks with column names.
 using TranslateQualifiedNamesVisitor = TranslateQualifiedNamesMatcher::Visitor;
 
+/// Restore ASTIdentifiers to long form
+struct RestoreQualifiedNamesData
+{
+    using TypeToVisit = ASTIdentifier;
+
+    void visit(ASTIdentifier & identifier, ASTPtr & ast);
+};
+
+using RestoreQualifiedNamesMatcher = OneTypeMatcher<RestoreQualifiedNamesData>;
+using RestoreQualifiedNamesVisitor = InDepthNodeVisitor<RestoreQualifiedNamesMatcher, true>;
+
 }

dbms/src/Parsers/ASTIdentifier.cpp

@@ -43,6 +43,15 @@ void ASTIdentifier::setShortName(const String & new_name)
     semantic->special = special;
 }
 
+void ASTIdentifier::restoreCompoundName()
+{
+    if (name_parts.empty())
+        return;
+    name = name_parts[0];
+    for (size_t i = 1; i < name_parts.size(); ++i)
+        name += '.' + name_parts[i];
+}
+
 void ASTIdentifier::formatImplWithoutAlias(const FormatSettings & settings, FormatState &, FormatStateStacked) const
 {
     auto format_element = [&](const String & elem_name)

dbms/src/Parsers/ASTIdentifier.h

@@ -38,6 +38,7 @@ public:
     bool isShort() const { return name_parts.empty() || name == name_parts.back(); }
 
     void setShortName(const String & new_name);
+    void restoreCompoundName();
 
     const String & shortName() const
     {

dbms/src/Storages/StorageDistributed.cpp

@@ -34,6 +34,7 @@
 #include <Interpreters/InterpreterAlterQuery.h>
 #include <Interpreters/InterpreterDescribeQuery.h>
 #include <Interpreters/InterpreterSelectQuery.h>
+#include <Interpreters/TranslateQualifiedNamesVisitor.h>
 #include <Interpreters/SyntaxAnalyzer.h>
 #include <Interpreters/createBlockSelector.h>
 #include <Interpreters/evaluateConstantExpression.h>
@@ -78,10 +79,20 @@ namespace
 ASTPtr rewriteSelectQuery(const ASTPtr & query, const std::string & database, const std::string & table, ASTPtr table_function_ptr = nullptr)
 {
     auto modified_query_ast = query->clone();
+
+    ASTSelectQuery & select_query = modified_query_ast->as<ASTSelectQuery &>();
+
+    /// restore long column names in JOIN ON expressions
+    if (auto tables = select_query.tables())
+    {
+        RestoreQualifiedNamesVisitor::Data data;
+        RestoreQualifiedNamesVisitor(data).visit(tables);
+    }
+
     if (table_function_ptr)
-        modified_query_ast->as<ASTSelectQuery &>().addTableFunction(table_function_ptr);
+        select_query.addTableFunction(table_function_ptr);
     else
-        modified_query_ast->as<ASTSelectQuery &>().replaceDatabaseAndTable(database, table);
+        select_query.replaceDatabaseAndTable(database, table);
     return modified_query_ast;
 }
 

dbms/tests/queries/0_stateless/00974_distributed_join_on.reference

@@ -0,0 +1,8 @@
+1
+1
+1
+1
+1
+1
+42	42
+42	42

dbms/tests/queries/0_stateless/00974_distributed_join_on.sql

@@ -0,0 +1,34 @@
+DROP TABLE IF EXISTS source_table1;
+DROP TABLE IF EXISTS source_table2;
+DROP TABLE IF EXISTS distributed_table1;
+DROP TABLE IF EXISTS distributed_table2;
+
+CREATE TABLE source_table1 (a Int64, b String) ENGINE = Memory;
+CREATE TABLE source_table2 (c Int64, d String) ENGINE = Memory;
+
+INSERT INTO source_table1 VALUES (42, 'qwe');
+INSERT INTO source_table2 VALUES (42, 'qwe');
+
+CREATE TABLE distributed_table1 AS source_table1
+ENGINE = Distributed('test_shard_localhost', currentDatabase(), source_table1);
+
+CREATE TABLE distributed_table2 AS source_table2
+ENGINE = Distributed('test_shard_localhost', currentDatabase(), source_table2);
+
+SET prefer_localhost_replica = 1;
+SELECT 1 FROM distributed_table1 AS t1 GLOBAL JOIN distributed_table2 AS t2 ON t1.a = t2.c LIMIT 1;
+SELECT 1 FROM distributed_table1 AS t1 GLOBAL JOIN distributed_table2 AS t2 ON t2.c = t1.a LIMIT 1;
+SELECT 1 FROM distributed_table1 AS t1 GLOBAL JOIN distributed_table1 AS t2 ON t1.a = t2.a LIMIT 1;
+
+SET prefer_localhost_replica = 0;
+SELECT 1 FROM distributed_table1 AS t1 GLOBAL JOIN distributed_table2 AS t2 ON t1.a = t2.c LIMIT 1;
+SELECT 1 FROM distributed_table1 AS t1 GLOBAL JOIN distributed_table2 AS t2 ON t2.c = t1.a LIMIT 1;
+SELECT 1 FROM distributed_table1 AS t1 GLOBAL JOIN distributed_table1 AS t2 ON t1.a = t2.a LIMIT 1;
+
+SELECT t1.a as t1_a, t2.a as t2_a FROM source_table1 AS t1 JOIN source_table1 AS t2 ON t1_a = t2_a LIMIT 1;
+SELECT t1.a as t1_a, t2.a as t2_a FROM distributed_table1 AS t1 GLOBAL JOIN distributed_table1 AS t2 ON t1_a = t2_a LIMIT 1;
+
+DROP TABLE source_table1;
+DROP TABLE source_table2;
+DROP TABLE distributed_table1;
+DROP TABLE distributed_table2;