fix table function dedup (#883)

Author: yl-lisen

src/Processors/QueryPlan/Streaming/DedupTransformStep.cpp

@@ -35,6 +35,8 @@ DedupTransformStep::DedupTransformStep(
 
 void DedupTransformStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings & /* settings */)
 {
+    /// Use one dedup transform for all input streams
+    pipeline.resize(1);
     pipeline.addSimpleTransform([&](const Block & header) { /// STYLE_CHECK_ALLOW_BRACE_SAME_LINE_LAMBDA
         return std::make_shared<DedupTransform>(header, getOutputStream().header, dedup_func_desc);
     });

src/TableFunctions/TableFunctionProxyBase.cpp

@@ -76,8 +76,8 @@ StoragePtr TableFunctionProxyBase::calculateColumnDescriptions(ContextPtr contex
     }
     else if (subquery)
     {
-        auto interpreter_subquery
-            = std::make_unique<InterpreterSelectWithUnionQuery>(subquery->children[0], context, SelectQueryOptions().subquery().analyze());
+        auto interpreter_subquery = std::make_unique<InterpreterSelectWithUnionQuery>(
+            subquery->children[0], context, SelectQueryOptions().subquery().analyze().removeDuplicates());
 
         auto source_header = interpreter_subquery->getSampleBlock();
         columns = ColumnsDescription(source_header.getNamesAndTypesList());

tests/queries_ported/0_stateless/99010_dedup_bug.reference

@@ -0,0 +1,4 @@
+3
+2024-01-01 00:00:00.100Z	1
+2024-01-01 00:00:00.200Z	2
+2024-01-01 00:00:00.300Z	3

tests/queries_ported/0_stateless/99010_dedup_bug.sql

@@ -0,0 +1,16 @@
+DROP STREAM IF EXISTS 99010_test_stream;
+CREATE STREAM 99010_test_stream(id int);
+
+SELECT sleep(2) FORMAT Null;
+
+INSERT INTO 99010_test_stream(id, _tp_time) VALUES (1, '2024-01-01 00:00:00.1');
+INSERT INTO 99010_test_stream(id, _tp_time) VALUES (2, '2024-01-01 00:00:00.2');
+INSERT INTO 99010_test_stream(id, _tp_time) VALUES (3, '2024-01-01 00:00:00.3');
+INSERT INTO 99010_test_stream(id, _tp_time) VALUES (1, '2024-01-01 00:00:00.4');
+
+SELECT sleep(3) FORMAT Null;
+
+with cte as (select _tp_time, * from table(99010_test_stream)) select count() from dedup(cte, id);
+with cte as (select _tp_time, * from table(99010_test_stream) order by _tp_time asc) select * from dedup(cte, id);
+
+DROP STREAM 99010_test_stream;