From 594a0c2df9d1c190fca153103e49e517b2d53cb8 Mon Sep 17 00:00:00 2001
From: Zhuhe Fang <fzhedu@gmail.com>
Date: Tue, 19 Apr 2022 20:30:03 +0800
Subject: [PATCH] cherry pick PR#3283 about sending schemas in exchangeSender
 (#4633)

close pingcap/tiflash#3146
---
 dbms/src/Flash/Coprocessor/DAGQueryBlock.cpp  | 12 ++-
 .../fullstack-test-dt/union_push_down.test    | 96 +++++++++++++++++++
 2 files changed, 107 insertions(+), 1 deletion(-)
 create mode 100644 tests/tidb-ci/fullstack-test-dt/union_push_down.test

diff --git a/dbms/src/Flash/Coprocessor/DAGQueryBlock.cpp b/dbms/src/Flash/Coprocessor/DAGQueryBlock.cpp
index 16dfc1696bd..92d60ad8ec8 100644
--- a/dbms/src/Flash/Coprocessor/DAGQueryBlock.cpp
+++ b/dbms/src/Flash/Coprocessor/DAGQueryBlock.cpp
@@ -10,7 +10,6 @@
 
 namespace DB
 {
-
 namespace ErrorCodes
 {
 extern const int NOT_IMPLEMENTED;
@@ -222,6 +221,17 @@ DAGQueryBlock::DAGQueryBlock(UInt32 id_, const ::google::protobuf::RepeatedPtrFi
 
 void DAGQueryBlock::fillOutputFieldTypes()
 {
+    /// the top block has exchangeSender, which decides the output fields, keeping the same with exchangeReceiver
+    if (exchangeSender != nullptr && exchangeSender->has_exchange_sender() && !exchangeSender->exchange_sender().all_field_types().empty())
+    {
+        output_field_types.clear();
+        for (auto & field_type : exchangeSender->exchange_sender().all_field_types())
+        {
+            output_field_types.push_back(field_type);
+        }
+        return;
+    }
+    /// the non-top block
     if (!output_field_types.empty())
     {
         return;
diff --git a/tests/tidb-ci/fullstack-test-dt/union_push_down.test b/tests/tidb-ci/fullstack-test-dt/union_push_down.test
new file mode 100644
index 00000000000..30ea7576c7e
--- /dev/null
+++ b/tests/tidb-ci/fullstack-test-dt/union_push_down.test
@@ -0,0 +1,96 @@
+# the null flag should be aligned for each sub select clause under union, this test is related to the tidb PR https://github.com/pingcap/tidb/pull/28990
+
+mysql> drop table if exists test.t;
+mysql> drop table if exists test.tt;
+mysql> create table test.t (id int, d double, nd double not null);
+mysql> alter table test.t set tiflash replica 1;
+mysql> insert into test.t values(0,0,0),(1,1,1),(2,null,2);
+mysql> analyze table test.t;
+mysql> create table test.tt like test.t;
+mysql> alter table test.tt set tiflash replica 1;
+mysql> insert into test.tt  select * from test.t;
+mysql> insert into test.tt  select * from test.t;
+mysql> insert into test.tt  select * from test.t;
+mysql> insert into test.tt  select * from test.t;
+mysql> analyze table test.tt;
+
+func> wait_table test t
+func> wait_table test tt
+
+mysql> use test;  set @@tidb_allow_mpp=1; set @@tidb_enforce_mpp=1;  Select DD,NDD,IDD from tt join  ( (select d as DD, nd as NDD, id as IDD from t) union all (select d as DD, 0 as NDD, id as IDD from t) union all (select d as DD, nd as NDD, 0 as IDD from t)  ) u on tt.id = u.IDD;
++------+------+------+
+| DD   | NDD  | IDD  |
++------+------+------+
+|    0 |    0 |    0 |
+|    0 |    0 |    0 |
+|    0 |    0 |    0 |
+| NULL |    2 |    0 |
+|    1 |    1 |    0 |
+|    1 |    1 |    1 |
+|    1 |    0 |    1 |
+| NULL |    2 |    2 |
+| NULL |    0 |    2 |
+|    0 |    0 |    0 |
+|    0 |    0 |    0 |
+|    0 |    0 |    0 |
+| NULL |    2 |    0 |
+|    1 |    1 |    0 |
+|    1 |    1 |    1 |
+|    1 |    0 |    1 |
+| NULL |    2 |    2 |
+| NULL |    0 |    2 |
+|    0 |    0 |    0 |
+|    0 |    0 |    0 |
+|    0 |    0 |    0 |
+| NULL |    2 |    0 |
+|    1 |    1 |    0 |
+|    1 |    1 |    1 |
+|    1 |    0 |    1 |
+| NULL |    2 |    2 |
+| NULL |    0 |    2 |
+|    0 |    0 |    0 |
+|    0 |    0 |    0 |
+|    0 |    0 |    0 |
+| NULL |    2 |    0 |
+|    1 |    1 |    0 |
+|    1 |    1 |    1 |
+|    1 |    0 |    1 |
+| NULL |    2 |    2 |
+| NULL |    0 |    2 |
++------+------+------+
+
+mysql> use test;  set @@tidb_allow_mpp=1; set @@tidb_enforce_mpp=1; Select IDD from tt join  ( (select 127 as IDD from t) union all (select id as IDD from t)  ) u on tt.id = u.IDD;
++------+
+| IDD  |
++------+
+|    0 |
+|    1 |
+|    2 |
+|    0 |
+|    1 |
+|    2 |
+|    0 |
+|    1 |
+|    2 |
+|    0 |
+|    1 |
+|    2 |
++------+
+
+mysql> use test;  set @@tidb_allow_mpp=1; set @@tidb_enforce_mpp=1; Select IDD from tt join  ( (select 127 as IDD from t) union all (select 1 as IDD from t)  ) u on tt.id = u.IDD;
++------+
+| IDD  |
++------+
+|    1 |
+|    1 |
+|    1 |
+|    1 |
+|    1 |
+|    1 |
+|    1 |
+|    1 |
+|    1 |
+|    1 |
+|    1 |
+|    1 |
++------+