MDEV-25080 Fix crash for CREATE TABLE from pushed union

During st_select_lex_unit::prepare() the member select_unit*
st_select_lex_unit::union_result is being assigned to an instance
of one of the following classes:
 - select_unit
 - select_unit_ext
 - select_unit_recursive
 - select_union_direct
Select_union_direct used to pass the result of the query directly to
the receiving select_result without filling a temporary table. This class
wraps a select_result object and is currently used to process UNION ALL
queries. Other select_unit_* classes involve some additional result processing.
Pushed down units are processed on the engine side so the results must be
also passed directly to a select_result object. So in the case when
the unit pushdown is employed st_select_lex_unit::union_result must be
assigned to an instance of select_union_direct.

mysql-test/suite/federated/federatedx_create_handlers.result

@@ -555,14 +555,36 @@ WHERE id=2) dt2) dt
 a	b	a	b	id	name
 1	1	NULL	NULL	NULL	NULL
 2	2	NULL	NULL	NULL	NULL
+DROP TABLES federated.t1, federated.t2, federated.t3, federated.t10,
+federated.t11;
+connection slave;
+DROP TABLES federated.t1, federated.t2, federated.t3, federated.t10,
+federated.t11;
 # MDEV-25080: Allow pushdown of queries involving UNIONs
 #   in outer select to foreign engines
 #
-connection master;
+connection slave;
-TRUNCATE TABLE federated.t1;
+CREATE TABLE federated.t1 (
-TRUNCATE TABLE federated.t2;
+a varchar(10)
+)
+DEFAULT CHARSET=latin1;
+CREATE TABLE federated.t2 (
+a varchar(16) NOT NULL default ''
+)
+DEFAULT CHARSET=latin1;
 INSERT INTO federated.t1 VALUES ('abc'), ('bcd'), ('cde');
 INSERT INTO federated.t2 VALUES ('abc'), ('bcd'), ('cde'), ('def'), ('efg');
+connection master;
+CREATE TABLE federated.t1 (
+a varchar(10)
+)
+ENGINE="FEDERATED" DEFAULT CHARSET=latin1
+CONNECTION='mysql://root@127.0.0.1:SLAVE_PORT/federated/t1';
+CREATE TABLE federated.t2 (
+a varchar(16) NOT NULL default ''
+)
+ENGINE="FEDERATED" DEFAULT CHARSET=latin1
+CONNECTION='mysql://root@127.0.0.1:SLAVE_PORT/federated/t2';
 CREATE TABLE t3 (a varchar(30)) ENGINE=MyISAM;
 CREATE TABLE t4 (a varchar(30)) ENGINE=MyISAM;
 INSERT INTO t3 VALUES ('t3_myisam1'), ('t3_myisam2'), ('t3_myisam3');
@@ -1033,9 +1055,38 @@ common
 123456789000
 t14abcde
 t14xyzzz
+# CREATE TABLE .. AS from a pushed UNION
+CREATE TABLE t5 AS SELECT * FROM federated.t13 UNION
+SELECT * FROM federated.t14;
+SHOW CREATE TABLE t5;
+Table	Create Table
+t5	CREATE TABLE `t5` (
+  `a` varchar(8) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+SELECT * FROM t5;
+a
+common
+t13abc
+t13xx
+t14abcde
+t14xyzzz
+CREATE TABLE t6 AS SELECT a FROM federated.t12 EXCEPT
+SELECT 1 UNION ALL
+SELECT a FROM federated.t11 EXCEPT
+SELECT 0;
+SHOW CREATE TABLE t6;
+Table	Create Table
+t6	CREATE TABLE `t6` (
+  `a` decimal(10,0) NOT NULL DEFAULT 0
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+SELECT * FROM t6;
+a
+-1
+-32678
+32767
 connection master;
-DROP TABLES federated.t1, federated.t2, t3, t4, federated.t11, federated.t12,
+DROP TABLES federated.t1, federated.t2, t3, t4, t5, t6, federated.t11,
-federated.t13, federated.t14;
+federated.t12, federated.t13, federated.t14;
 connection slave;
 DROP TABLES federated.t1, federated.t2, federated.t11, federated.t12,
 federated.t13, federated.t14;

mysql-test/suite/federated/federatedx_create_handlers.test

@@ -375,18 +375,49 @@ SELECT * FROM t10 LEFT JOIN
     WHERE id=2) dt2) dt
   ) ON t10.a=t11.a;
 
+DROP TABLES federated.t1, federated.t2, federated.t3, federated.t10,
+   federated.t11;
+connection slave;
+DROP TABLES federated.t1, federated.t2, federated.t3, federated.t10,
+   federated.t11;
+
 --echo # MDEV-25080: Allow pushdown of queries involving UNIONs
 --echo #   in outer select to foreign engines
 --echo #
 
-connection master;
+connection slave;
 
-TRUNCATE TABLE federated.t1;
+CREATE TABLE federated.t1 (
-TRUNCATE TABLE federated.t2;
+  a varchar(10)
+)
+DEFAULT CHARSET=latin1;
+
+CREATE TABLE federated.t2 (
+  a varchar(16) NOT NULL default ''
+)
+DEFAULT CHARSET=latin1;
 
 INSERT INTO federated.t1 VALUES ('abc'), ('bcd'), ('cde');
 INSERT INTO federated.t2 VALUES ('abc'), ('bcd'), ('cde'), ('def'), ('efg');
 
+connection master;
+
+--replace_result $SLAVE_MYPORT SLAVE_PORT
+eval
+CREATE TABLE federated.t1 (
+  a varchar(10)
+)
+ENGINE="FEDERATED" DEFAULT CHARSET=latin1
+CONNECTION='mysql://root@127.0.0.1:$SLAVE_MYPORT/federated/t1';
+
+--replace_result $SLAVE_MYPORT SLAVE_PORT
+eval
+CREATE TABLE federated.t2 (
+  a varchar(16) NOT NULL default ''
+)
+ENGINE="FEDERATED" DEFAULT CHARSET=latin1
+CONNECTION='mysql://root@127.0.0.1:$SLAVE_MYPORT/federated/t2';
+
 CREATE TABLE t3 (a varchar(30)) ENGINE=MyISAM;
 CREATE TABLE t4 (a varchar(30)) ENGINE=MyISAM;
 
@@ -467,7 +498,6 @@ EXPLAIN
 --echo # at the server side
 
 --disable_warnings
-
 SELECT count(*) FROM federated.t1 UNION
   SELECT count(*) FROM federated.t1 EXCEPT
   SELECT count(*)+1 FROM federated.t1
@@ -482,7 +512,6 @@ EXPLAIN FORMAT=JSON SELECT count(*) FROM federated.t1 UNION
   SELECT count(*) FROM federated.t2 EXCEPT
   SELECT count(*)+2 FROM federated.t2
   INTO @var;
-
 --enable_warnings
 
 --echo # Prepared statements
@@ -549,6 +578,7 @@ EXPLAIN (SELECT * FROM federated.t1 UNION SELECT * FROM federated.t2)
   
 --echo # Union of tables containing different INT data types
 connection slave;
+
 CREATE TABLE federated.t11 (a smallint(6) NOT NULL);
 INSERT INTO federated.t11 VALUES (-32678), (-1), (0);
 
@@ -625,10 +655,25 @@ SELECT * FROM federated.t13 UNION
   SELECT '123456789000' UNION
   SELECT * FROM federated.t14;
 
+--echo # CREATE TABLE .. AS from a pushed UNION
+CREATE TABLE t5 AS SELECT * FROM federated.t13 UNION
+  SELECT * FROM federated.t14;
+SHOW CREATE TABLE t5;
+--sorted_result
+SELECT * FROM t5;
+
+CREATE TABLE t6 AS SELECT a FROM federated.t12 EXCEPT
+  SELECT 1 UNION ALL
+  SELECT a FROM federated.t11 EXCEPT
+  SELECT 0;
+SHOW CREATE TABLE t6;
+--sorted_result
+SELECT * FROM t6;
+
 # Cleanup
 connection master;
-DROP TABLES federated.t1, federated.t2, t3, t4, federated.t11, federated.t12,
+DROP TABLES federated.t1, federated.t2, t3, t4, t5, t6, federated.t11,
-  federated.t13, federated.t14;
+  federated.t12, federated.t13, federated.t14;
 
 connection slave;
 DROP TABLES federated.t1, federated.t2, federated.t11, federated.t12,

sql/sql_lex.h

@@ -1056,6 +1056,7 @@ public:
 private:
   bool exec_inner();
   bool is_derived_eliminated() const;
+  bool set_direct_union_result(select_result *sel_result);
 };
 
 typedef class st_select_lex_unit SELECT_LEX_UNIT;

sql/sql_union.cc

@@ -1388,6 +1388,7 @@ bool st_select_lex_unit::prepare(TABLE_LIST *derived_arg,
   bool have_except= false, have_intersect= false,
     have_except_all_or_intersect_all= false;
   bool instantiate_tmp_table= false;
+  bool use_direct_union_result= false;
   bool single_tvc= !first_sl->next_select() && first_sl->tvc;
   bool single_tvc_wo_order= single_tvc && !first_sl->order_list.elements;
   bool distinct_key= 0;
@@ -1511,23 +1512,18 @@ bool st_select_lex_unit::prepare(TABLE_LIST *derived_arg,
     }
   }
 
-  /* Global option */
-
   if (is_union_select || is_recursive)
   {
     if ((single_tvc_wo_order && !fake_select_lex) ||
         (is_unit_op() && !union_needs_tmp_table() &&
-	 !have_except && !have_intersect && !single_tvc))
+         !have_except && !have_intersect && !single_tvc))
     {
-      SELECT_LEX *last= first_select();
+      if (unlikely(set_direct_union_result(sel_result)))
-      while (last->next_select())
+        goto err;
-        last= last->next_select();
+      tmp_result= union_result;
-      if (!(tmp_result= union_result=
-              new (thd->mem_root) select_union_direct(thd, sel_result,
-                                                          last)))
-        goto err; /* purecov: inspected */
       fake_select_lex= NULL;
       instantiate_tmp_table= false;
+      use_direct_union_result= true;
     }
     else
     {
@@ -1763,6 +1759,24 @@ bool st_select_lex_unit::prepare(TABLE_LIST *derived_arg,
     goto err;
 
 cont:
+  pushdown_unit= find_unit_handler(thd, this);
+  if (pushdown_unit)
+  {
+    if (unlikely(pushdown_unit->prepare()))
+      goto err;
+    /*
+      Always use select_union_direct result for pushed down units, overwrite
+      the previous union_result unless select_union_direct is already used
+    */
+    if (!use_direct_union_result)
+    {
+      if (unlikely(set_direct_union_result(sel_result)))
+        goto err;
+      fake_select_lex= NULL;
+      instantiate_tmp_table= false;
+      use_direct_union_result= true;
+    }
+  }
   /*
     If the query is using select_union_direct, we have postponed
     preparation of the underlying select_result until column types
@@ -1965,13 +1979,6 @@ cont:
     }
   }
 
-  pushdown_unit= find_unit_handler(thd, this);
-  if (pushdown_unit)
-  {
-    if (unlikely(pushdown_unit->prepare()))
-      DBUG_RETURN(TRUE);
-  }
-
   thd->lex->current_select= lex_select_save;
 
   DBUG_RETURN(saved_error || thd->is_fatal_error);
@@ -1982,6 +1989,15 @@ err:
   DBUG_RETURN(TRUE);
 }
 
+bool st_select_lex_unit::set_direct_union_result(select_result *sel_result)
+{
+  SELECT_LEX *last= first_select();
+  while (last->next_select())
+    last= last->next_select();
+  union_result= new (thd->mem_root) select_union_direct(thd, sel_result,
+                                                      last);
+  return (union_result == nullptr);
+}
 
 /**
   @brief