Support join on syntax

dbms/src/Core/Names.h

@@ -12,5 +12,6 @@ namespace DB
 using Names = std::vector<std::string>;
 using NameSet = std::unordered_set<std::string>;
 using NameToNameMap = std::unordered_map<std::string, std::string>;
+using NameToNameSetMap = std::unordered_map<std::string, NameSet>;
 
 }

dbms/src/DataStreams/CreatingSetsBlockInputStream.cpp

@@ -125,6 +125,21 @@ void CreatingSetsBlockInputStream::createOne(SubqueryForSet & subquery)
 
         if (!done_with_join)
         {
+            if (subquery.joined_block_actions)
+                subquery.joined_block_actions->execute(block);
+
+            for (const auto & name_with_alias : subquery.joined_block_aliases)
+            {
+                if (block.has(name_with_alias.first))
+                {
+                    auto pos = block.getPositionByName(name_with_alias.first);
+                    auto column = block.getByPosition(pos);
+                    block.erase(pos);
+                    column.name = name_with_alias.second;
+                    block.insert(std::move(column));
+                }
+            }
+
             if (!subquery.join->insertFromBlock(block))
                 done_with_join = true;
         }

dbms/src/Interpreters/ExpressionActions.cpp

@@ -41,6 +41,8 @@ Names ExpressionAction::getNeededColumns() const
 
     res.insert(res.end(), array_joined_columns.begin(), array_joined_columns.end());
 
+    res.insert(res.end(), join_key_names_left.begin(), join_key_names_left.end());
+
     for (const auto & column : projection)
         res.push_back(column.first);
 
@@ -146,11 +148,14 @@ ExpressionAction ExpressionAction::arrayJoin(const NameSet & array_joined_column
     return a;
 }
 
-ExpressionAction ExpressionAction::ordinaryJoin(std::shared_ptr<const Join> join_, const NamesAndTypesList & columns_added_by_join_)
+ExpressionAction ExpressionAction::ordinaryJoin(std::shared_ptr<const Join> join_,
+                                                const Names & join_key_names_left,
+                                                const NamesAndTypesList & columns_added_by_join_)
 {
     ExpressionAction a;
     a.type = JOIN;
-    a.join = join_;
+    a.join = std::move(join_);
+    a.join_key_names_left = join_key_names_left;
     a.columns_added_by_join = columns_added_by_join_;
     return a;
 }

dbms/src/Interpreters/ExpressionActions.h

@@ -34,11 +34,14 @@ using DataTypePtr = std::shared_ptr<const IDataType>;
 class IBlockInputStream;
 using BlockInputStreamPtr = std::shared_ptr<IBlockInputStream>;
 
+class ExpressionActions;
 
 /** Action on the block.
   */
 struct ExpressionAction
 {
+private:
+    using ExpressionActionsPtr = std::shared_ptr<ExpressionActions>;
 public:
     enum Type
     {
@@ -85,6 +88,7 @@ struct ExpressionAction
 
     /// For JOIN
     std::shared_ptr<const Join> join;
+    Names join_key_names_left;
     NamesAndTypesList columns_added_by_join;
 
     /// For PROJECT.
@@ -103,7 +107,8 @@ struct ExpressionAction
     static ExpressionAction project(const NamesWithAliases & projected_columns_);
     static ExpressionAction project(const Names & projected_columns_);
     static ExpressionAction arrayJoin(const NameSet & array_joined_columns, bool array_join_is_left, const Context & context);
-    static ExpressionAction ordinaryJoin(std::shared_ptr<const Join> join_, const NamesAndTypesList & columns_added_by_join_);
+    static ExpressionAction ordinaryJoin(std::shared_ptr<const Join> join_, const Names & join_key_names_left,
+                                         const NamesAndTypesList & columns_added_by_join_);
 
     /// Which columns necessary to perform this action.
     Names getNeededColumns() const;