Merge pull request ClickHouse#78828 from ClickHouse/fix-bad-cast

Fix bad cast in `StorageDistributed`

Author: thevar1able

src/Parsers/ASTSelectQuery.cpp

@@ -448,7 +448,7 @@ void ASTSelectQuery::replaceDatabaseAndTable(const StorageID & table_id)
 }
 
 
-void ASTSelectQuery::addTableFunction(ASTPtr & table_function_ptr)
+void ASTSelectQuery::addTableFunction(const ASTPtr & table_function_ptr)
 {
     ASTTableExpression * table_expression = getFirstTableExpression(*this);
 

src/Parsers/ASTSelectQuery.h

@@ -153,7 +153,7 @@ class ASTSelectQuery : public IAST
     bool withFill() const;
     void replaceDatabaseAndTable(const String & database_name, const String & table_name);
     void replaceDatabaseAndTable(const StorageID & table_id);
-    void addTableFunction(ASTPtr & table_function_ptr);
+    void addTableFunction(const ASTPtr & table_function_ptr);
     void updateTreeHashImpl(SipHash & hash_state, bool ignore_aliases) const override;
 
     void setFinal();

src/Storages/StorageDistributed.cpp

@@ -37,6 +37,7 @@
 #include <Common/threadPoolCallbackRunner.h>
 #include <Common/typeid_cast.h>
 
+#include <Parsers/ASTAsterisk.h>
 #include <Parsers/ASTExpressionList.h>
 #include <Parsers/ASTFunction.h>
 #include <Parsers/ASTIdentifier.h>
@@ -970,14 +971,30 @@ std::optional<QueryPipeline> StorageDistributed::distributedWriteBetweenDistribu
     const auto & settings = local_context->getSettingsRef();
     auto new_query = std::dynamic_pointer_cast<ASTInsertQuery>(query.clone());
 
-    /// Unwrap view() function.
     if (src_distributed.remote_table_function_ptr)
     {
         const TableFunctionPtr src_table_function =
             TableFunctionFactory::instance().get(src_distributed.remote_table_function_ptr, local_context);
-        const TableFunctionView * view_function =
-            assert_cast<const TableFunctionView *>(src_table_function.get());
-        new_query->select = view_function->getSelectQuery().clone();
+        if (const TableFunctionView * view_function = typeid_cast<const TableFunctionView *>(src_table_function.get()))
+        {
+            new_query->select = view_function->getSelectQuery().clone();
+        }
+        else
+        {
+            const auto select_with_union_query = std::make_shared<ASTSelectWithUnionQuery>();
+            select_with_union_query->list_of_selects = std::make_shared<ASTExpressionList>();
+
+            const auto select = std::make_shared<ASTSelectQuery>();
+
+            auto expression_list = std::make_shared<ASTExpressionList>();
+            expression_list->children.push_back(std::make_shared<ASTAsterisk>());
+            select->setExpression(ASTSelectQuery::Expression::SELECT, expression_list->clone());
+            select->addTableFunction(src_distributed.remote_table_function_ptr);
+
+            select_with_union_query->list_of_selects->children.push_back(select->clone());
+
+            new_query->select = select_with_union_query;
+        }
     }
     else
     {

tests/queries/0_stateless/01099_parallel_distributed_insert_select.reference

@@ -28,6 +28,11 @@ test_shard_localhost
 0
 1
 2
+0
+1
+2
+3
+4
 test_cluster_two_shards_localhost
 0	2
 1	2

tests/queries/0_stateless/01099_parallel_distributed_insert_select.sql

@@ -169,6 +169,17 @@ DROP TABLE local_01099_b;
 DROP TABLE distributed_01099_a;
 DROP TABLE distributed_01099_b;
 
+--- https://github.com/ClickHouse/ClickHouse/issues/78464
+CREATE TABLE local_01099_c (n UInt64) ENGINE = Log;
+CREATE TABLE distributed_01099_c AS local_01099_c ENGINE = Distributed('test_shard_localhost', currentDatabase(), local_01099_c, rand());
+
+INSERT INTO TABLE FUNCTION clusterAllReplicas('test_shard_localhost', currentDatabase(), 'distributed_01099_c') (n) SELECT number FROM remote('localhost', numbers(5)) tx;
+
+SELECT * FROM distributed_01099_c;
+
+DROP TABLE local_01099_c;
+DROP TABLE distributed_01099_c;
+
 --
 -- test_cluster_two_shards_localhost
 --