cherry pick PR#3283 about sending schemas in exchangeSender (#4633)

close #3146

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;

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 |
++------+