MDEV-10014 Add RETURNING to INSERT

Closes #1384

14 files changed, 304 insertions, 105 deletions

diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 2a7470dc689..1df08eb739e 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -7431,6 +7431,7 @@ SHOW_VAR status_vars[]= {
   {"Feature_dynamic_columns",  (char*) offsetof(STATUS_VAR, feature_dynamic_columns), SHOW_LONG_STATUS},
   {"Feature_fulltext",         (char*) offsetof(STATUS_VAR, feature_fulltext), SHOW_LONG_STATUS},
   {"Feature_gis",              (char*) offsetof(STATUS_VAR, feature_gis), SHOW_LONG_STATUS},
+  {"Feature_insert_returning",   (char*)offsetof(STATUS_VAR, feature_insert_returning), SHOW_LONG_STATUS},
   {"Feature_invisible_columns",   (char*) offsetof(STATUS_VAR, feature_invisible_columns), SHOW_LONG_STATUS},
   {"Feature_json",             (char*) offsetof(STATUS_VAR, feature_json), SHOW_LONG_STATUS},
   {"Feature_locale",           (char*) offsetof(STATUS_VAR, feature_locale), SHOW_LONG_STATUS},
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index feb79990796..15c0e976c26 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -7632,6 +7632,26 @@ bool setup_fields(THD *thd, Ref_ptr_array ref_pointer_array,
 
 
 /*
+  Perform checks like all given fields exists, if exists fill struct with
+  current data and expand all '*' in given fields for LEX::returning.
+
+  SYNOPSIS
+     thd                 Thread handler
+     table_list          Global/local table list
+*/
+
+int setup_returning_fields(THD* thd, TABLE_LIST* table_list)
+{
+  if (!thd->lex->has_returning())
+    return 0;
+  return setup_wild(thd, table_list, thd->lex->returning()->item_list, NULL,
+                    thd->lex->returning())
+      || setup_fields(thd, Ref_ptr_array(), thd->lex->returning()->item_list,
+                      MARK_COLUMNS_READ, NULL, NULL, false);
+}
+
+
+/*
   make list of leaves of join table tree
 
   SYNOPSIS
diff --git a/sql/sql_base.h b/sql/sql_base.h
index 40dd3ed0907..211993c41ac 100644
--- a/sql/sql_base.h
+++ b/sql/sql_base.h
@@ -176,6 +176,7 @@ void make_leaves_list(THD *thd, List<TABLE_LIST> &list, TABLE_LIST *tables,
                       bool full_table_list, TABLE_LIST *boundary);
 int setup_wild(THD *thd, TABLE_LIST *tables, List<Item> &fields,
 	       List<Item> *sum_func_list, SELECT_LEX *sl);
+int setup_returning_fields(THD* thd, TABLE_LIST* table_list);
 bool setup_fields(THD *thd, Ref_ptr_array ref_pointer_array,
                   List<Item> &item, enum_column_usage column_usage,
                   List<Item> *sum_func_list, List<Item> *pre_fix,
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 34bafea8784..a208b627b7e 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -878,6 +878,7 @@ typedef struct system_status_var
   ulong feature_system_versioning;  /* +1 opening a table WITH SYSTEM VERSIONING */
   ulong feature_application_time_periods;
                                     /* +1 opening a table with application-time period */
+  ulong feature_insert_returning;  /* +1 when INSERT...RETURNING is used */
   ulong feature_timezone;	    /* +1 when XPATH is used */
   ulong feature_trigger;	    /* +1 opening a table with triggers */
   ulong feature_xml;		    /* +1 when XPATH is used */
@@ -5517,16 +5518,17 @@ public:
 
 class select_insert :public select_result_interceptor {
  public:
+  select_result *sel_result;
   TABLE_LIST *table_list;
   TABLE *table;
   List<Item> *fields;
   ulonglong autoinc_value_of_last_inserted_row; // autogenerated or not
   COPY_INFO info;
   bool insert_into_view;
-  select_insert(THD *thd_arg, TABLE_LIST *table_list_par,
-		TABLE *table_par, List<Item> *fields_par,
-		List<Item> *update_fields, List<Item> *update_values,
-		enum_duplicates duplic, bool ignore);
+  select_insert(THD *thd_arg, TABLE_LIST *table_list_par, TABLE *table_par,
+                List<Item> *fields_par, List<Item> *update_fields,
+                List<Item> *update_values, enum_duplicates duplic,
+                bool ignore, select_result *sel_ret_list);
   ~select_insert();
   int prepare(List<Item> &list, SELECT_LEX_UNIT *u);
   virtual int prepare2(JOIN *join);
@@ -5562,7 +5564,7 @@ public:
                 List<Item> &select_fields,enum_duplicates duplic, bool ignore,
                 TABLE_LIST *select_tables_arg):
     select_insert(thd_arg, table_arg, NULL, &select_fields, 0, 0, duplic,
-                  ignore),
+                  ignore, NULL),
     create_table(table_arg),
     create_info(create_info_par),
     select_tables(select_tables_arg),
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index 6a58dab9394..8161a761486 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -319,10 +319,10 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
   ORDER *order= (ORDER *) ((order_list && order_list->elements) ?
                            order_list->first : NULL);
   SELECT_LEX   *select_lex= thd->lex->first_select_lex();
+  SELECT_LEX   *returning= thd->lex->has_returning() ? thd->lex->returning() : 0;
   killed_state killed_status= NOT_KILLED;
   THD::enum_binlog_query_type query_type= THD::ROW_QUERY_TYPE;
   bool binlog_is_row;
-  bool with_select= !select_lex->item_list.is_empty();
   Explain_delete *explain;
   Delete_plan query_plan(thd->mem_root);
   Unique * deltempfile= NULL;
@@ -385,16 +385,14 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
   query_plan.select_lex= thd->lex->first_select_lex();
   query_plan.table= table;
 
-  if (mysql_prepare_delete(thd, table_list, select_lex->with_wild,
-                           select_lex->item_list, &conds,
-                           &delete_while_scanning))
+  if (mysql_prepare_delete(thd, table_list, &conds, &delete_while_scanning))
     DBUG_RETURN(TRUE);
 
   if (delete_history)
     table->vers_write= false;
 
-  if (with_select)
-    (void) result->prepare(select_lex->item_list, NULL);
+  if (returning)
+    (void) result->prepare(returning->item_list, NULL);
 
   if (thd->lex->current_select->first_cond_optimization)
   {
@@ -459,9 +457,8 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
 
   has_triggers= table->triggers && table->triggers->has_delete_triggers();
 
-  if (!with_select && !using_limit && const_cond_result &&
-      (!thd->is_current_stmt_binlog_format_row() &&
-       !has_triggers)
+  if (!returning && !using_limit && const_cond_result &&
+      (!thd->is_current_stmt_binlog_format_row() && !has_triggers)
       && !table->versioned(VERS_TIMESTAMP) && !table_list->has_period())
   {
     /* Update the table->file->stats.records number */
@@ -632,7 +629,7 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
   */
 
   if ((table->file->ha_table_flags() & HA_CAN_DIRECT_UPDATE_AND_DELETE) &&
-      !has_triggers && !binlog_is_row && !with_select &&
+      !has_triggers && !binlog_is_row && !returning &&
       !table_list->has_period())
   {
     table->mark_columns_needed_for_delete();
@@ -682,7 +679,7 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
         DELETE ... RETURNING we can't, because the RETURNING part may have
         a subquery in it)
       */
-      if (!with_select)
+      if (!returning)
         free_underlaid_joins(thd, select_lex);
       select= 0;
     }
@@ -717,11 +714,10 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
       !table->prepare_triggers_for_delete_stmt_or_event())
     will_batch= !table->file->start_bulk_delete();
 
-  if (with_select)
+  if (returning)
   {
-    if (unlikely(result->send_result_set_metadata(select_lex->item_list,
-                                                  Protocol::SEND_NUM_ROWS |
-                                                  Protocol::SEND_EOF)))
+    if (result->send_result_set_metadata(returning->item_list,
+                                Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
       goto cleanup;
   }
 
@@ -804,7 +800,7 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
       }
 
       // no LIMIT / OFFSET
-      if (with_select && result->send_data(select_lex->item_list) < 0)
+      if (returning && result->send_data(returning->item_list) < 0)
       {
         error=1;
         break;
@@ -948,7 +944,7 @@ cleanup:
     if (thd->lex->analyze_stmt)
       goto send_nothing_and_leave;
 
-    if (with_select)
+    if (returning)
       result->send_eof();
     else
       my_ok(thd, deleted);
@@ -998,16 +994,13 @@ got_error:
     mysql_prepare_delete()
     thd			- thread handler
     table_list		- global/local table list
-    wild_num            - number of wildcards used in optional SELECT clause 
-    field_list          - list of items in optional SELECT clause
     conds		- conditions
 
   RETURN VALUE
     FALSE OK
     TRUE  error
 */
-int mysql_prepare_delete(THD *thd, TABLE_LIST *table_list,
-                         uint wild_num, List<Item> &field_list, Item **conds,
+int mysql_prepare_delete(THD *thd, TABLE_LIST *table_list, Item **conds,
                          bool *delete_while_scanning)
 {
   Item *fake_conds= 0;
@@ -1017,12 +1010,9 @@ int mysql_prepare_delete(THD *thd, TABLE_LIST *table_list,
 
   *delete_while_scanning= true;
   thd->lex->allow_sum_func.clear_all();
-  if (setup_tables_and_check_access(thd,
-                                    &thd->lex->first_select_lex()->context,
-                                    &thd->lex->first_select_lex()->
-                                      top_join_list,
-                                    table_list, 
-                                    select_lex->leaf_tables, FALSE, 
+  if (setup_tables_and_check_access(thd, &select_lex->context,
+                                    &select_lex->top_join_list, table_list,
+                                    select_lex->leaf_tables, FALSE,
                                     DELETE_ACL, SELECT_ACL, TRUE))
     DBUG_RETURN(TRUE);
   if (table_list->vers_conditions.is_set())
@@ -1049,9 +1039,7 @@ int mysql_prepare_delete(THD *thd, TABLE_LIST *table_list,
       DBUG_RETURN(true);
   }
 
-  if ((wild_num && setup_wild(thd, table_list, field_list, NULL, select_lex)) ||
-      setup_fields(thd, Ref_ptr_array(),
-                   field_list, MARK_COLUMNS_READ, NULL, NULL, 0) ||
+  if (setup_returning_fields(thd, table_list) ||
       setup_conds(thd, table_list, select_lex->leaf_tables, conds) ||
       setup_ftfuncs(select_lex))
     DBUG_RETURN(TRUE);
diff --git a/sql/sql_delete.h b/sql/sql_delete.h
index 7af8564abf9..520524c72cc 100644
--- a/sql/sql_delete.h
+++ b/sql/sql_delete.h
@@ -26,8 +26,7 @@ class select_result;
 typedef class Item COND;
 template <typename T> class SQL_I_List;
 
-int mysql_prepare_delete(THD *thd, TABLE_LIST *table_list,
-                         uint wild_num, List<Item> &field_list, Item **conds,
+int mysql_prepare_delete(THD *thd, TABLE_LIST *table_list, Item **conds,
                          bool *delete_while_scanning);
 bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
                   SQL_I_List<ORDER> *order, ha_rows rows, 
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index e7f746263db..ba7bf4c44c7 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -677,18 +677,19 @@ Field **TABLE::field_to_fill()
 /**
   INSERT statement implementation
 
+  SYNOPSIS
+  mysql_insert()
+  result    NULL if the insert is not outputing results
+            via 'RETURNING' clause.
+
   @note Like implementations of other DDL/DML in MySQL, this function
   relies on the caller to close the thread tables. This is done in the
   end of dispatch_command().
 */
-
-bool mysql_insert(THD *thd,TABLE_LIST *table_list,
-                  List<Item> &fields,
-                  List<List_item> &values_list,
-                  List<Item> &update_fields,
-                  List<Item> &update_values,
-                  enum_duplicates duplic,
-		  bool ignore)
+bool mysql_insert(THD *thd, TABLE_LIST *table_list,
+                  List<Item> &fields, List<List_item> &values_list,
+                  List<Item> &update_fields, List<Item> &update_values,
+                  enum_duplicates duplic, bool ignore, select_result *result)
 {
   bool retval= true;
   int error, res;
@@ -707,6 +708,8 @@ bool mysql_insert(THD *thd,TABLE_LIST *table_list,
   List_item *values;
   Name_resolution_context *context;
   Name_resolution_context_state ctx_state;
+  SELECT_LEX   *returning= thd->lex->has_returning() ? thd->lex->returning() : 0;
+
 #ifndef EMBEDDED_LIBRARY
   char *query= thd->query();
   /*
@@ -765,6 +768,9 @@ bool mysql_insert(THD *thd,TABLE_LIST *table_list,
                            &unused_conds, FALSE))
     goto abort;
 
+  /* Prepares LEX::returing_list if it is not empty */
+  if (returning)
+    result->prepare(returning->item_list, NULL);
   /* mysql_prepare_insert sets table_list->table if it was not set */
   table= table_list->table;
 
@@ -934,6 +940,16 @@ bool mysql_insert(THD *thd,TABLE_LIST *table_list,
       goto values_loop_end;
     }
   }
+  /*
+    If statement returns result set, we need to send the result set metadata
+    to the client so that it knows that it has to expect an EOF or ERROR.
+    At this point we have all the required information to send the result set
+    metadata.
+  */
+  if (returning &&
+      result->send_result_set_metadata(returning->item_list,
+                                Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
+    goto values_loop_end;
 
   THD_STAGE_INFO(thd, stage_update);
   do
@@ -1063,6 +1079,18 @@ bool mysql_insert(THD *thd,TABLE_LIST *table_list,
       error=write_record(thd, table ,&info);
       if (unlikely(error))
         break;
+      /*
+        We send the row after writing it to the table so that the
+        correct values are sent to the client. Otherwise it won't show
+        autoinc values (generated inside the handler::ha_write()) and
+        values updated in ON DUPLICATE KEY UPDATE (handled inside
+        write_record()).
+      */
+      if (returning && result->send_data(returning->item_list) < 0)
+      {
+        error= 1;
+        break;
+      }
       thd->get_stmt_da()->inc_current_row_for_warning();
     }
     its.rewind();
@@ -1218,14 +1246,24 @@ values_loop_end:
     goto abort;
   if (thd->lex->analyze_stmt)
   {
-    retval= thd->lex->explain->send_explain(thd);
+    retval= 0;
     goto abort;
   }
 
   if ((iteration * values_list.elements) == 1 && (!(thd->variables.option_bits & OPTION_WARNINGS) ||
 				    !thd->cuted_fields))
   {
-    my_ok(thd, info.copied + info.deleted +
+    /*
+      Client expects an EOF/OK packet if result set metadata was sent. If
+      LEX::has_returning and the statement returns result set
+      we send EOF which is the indicator of the end of the row stream.
+      Oherwise we send an OK packet i.e when the statement returns only the
+      status information
+    */
+   if (returning)
+      result->send_eof();
+   else
+      my_ok(thd, info.copied + info.deleted +
                ((thd->client_capabilities & CLIENT_FOUND_ROWS) ?
                 info.touched : info.updated), id);
   }
@@ -1237,14 +1275,17 @@ values_loop_end:
 
     if (ignore)
       sprintf(buff, ER_THD(thd, ER_INSERT_INFO), (ulong) info.records,
-	      (lock_type == TL_WRITE_DELAYED) ? (ulong) 0 :
-	      (ulong) (info.records - info.copied),
+              (lock_type == TL_WRITE_DELAYED) ? (ulong) 0 :
+              (ulong) (info.records - info.copied),
               (long) thd->get_stmt_da()->current_statement_warn_count());
     else
       sprintf(buff, ER_THD(thd, ER_INSERT_INFO), (ulong) info.records,
-	      (ulong) (info.deleted + updated),
+              (ulong) (info.deleted + updated),
               (long) thd->get_stmt_da()->current_statement_warn_count());
-    ::my_ok(thd, info.copied + info.deleted + updated, id, buff);
+    if (returning)
+      result->send_eof();
+    else
+      ::my_ok(thd, info.copied + info.deleted + updated, id, buff);
   }
   thd->abort_on_warning= 0;
   if (thd->lex->current_select->first_cond_optimization)
@@ -1528,10 +1569,11 @@ bool mysql_prepare_insert(THD *thd, TABLE_LIST *table_list,
     table_list->next_local= 0;
     context->resolve_in_table_list_only(table_list);
 
-    res= (setup_fields(thd, Ref_ptr_array(),
-                       *values, MARK_COLUMNS_READ, 0, NULL, 0) ||
+    res= setup_returning_fields(thd, table_list) ||
+         setup_fields(thd, Ref_ptr_array(),
+                      *values, MARK_COLUMNS_READ, 0, NULL, 0) ||
           check_insert_fields(thd, context->table_list, fields, *values,
-                              !insert_into_view, 0, &map));
+                              !insert_into_view, 0, &map);
 
     if (!res)
       res= setup_fields(thd, Ref_ptr_array(),
@@ -3527,7 +3569,7 @@ bool Delayed_insert::handle_inserts(void)
     TRUE  Error
 */
 
-bool mysql_insert_select_prepare(THD *thd)
+bool mysql_insert_select_prepare(THD *thd, select_result *sel_res)
 {
   LEX *lex= thd->lex;
   SELECT_LEX *select_lex= lex->first_select_lex();
@@ -3544,6 +3586,13 @@ bool mysql_insert_select_prepare(THD *thd)
                            &select_lex->where, TRUE))
     DBUG_RETURN(TRUE);
 
+  /*
+    If sel_res is not empty, it means we have items in returing_list.
+    So we prepare the list now
+  */
+  if (sel_res)
+    sel_res->prepare(lex->returning()->item_list, NULL);
+
   DBUG_ASSERT(select_lex->leaf_tables.elements != 0);
   List_iterator<TABLE_LIST> ti(select_lex->leaf_tables);
   TABLE_LIST *table;
@@ -3586,8 +3635,10 @@ select_insert::select_insert(THD *thd_arg, TABLE_LIST *table_list_par,
                              List<Item> *update_fields,
                              List<Item> *update_values,
                              enum_duplicates duplic,
-                             bool ignore_check_option_errors):
+                             bool ignore_check_option_errors,
+                             select_result *result):
   select_result_interceptor(thd_arg),
+  sel_result(result),
   table_list(table_list_par), table(table_par), fields(fields_par),
   autoinc_value_of_last_inserted_row(0),
   insert_into_view(table_list_par && table_list_par->view != 0)
@@ -3606,7 +3657,7 @@ int
 select_insert::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
 {
   LEX *lex= thd->lex;
-  int res;
+  int res= 0;
   table_map map= 0;
   SELECT_LEX *lex_current_select_save= lex->current_select;
   DBUG_ENTER("select_insert::prepare");
@@ -3620,10 +3671,10 @@ select_insert::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
   */
   lex->current_select= lex->first_select_lex();
 
-  res= (setup_fields(thd, Ref_ptr_array(),
-                     values, MARK_COLUMNS_READ, 0, NULL, 0) ||
-        check_insert_fields(thd, table_list, *fields, values,
-                            !insert_into_view, 1, &map));
+  res= setup_returning_fields(thd, table_list) ||
+       setup_fields(thd, Ref_ptr_array(), values, MARK_COLUMNS_READ, 0, 0, 0) ||
+       check_insert_fields(thd, table_list, *fields, values,
+                                  !insert_into_view, 1, &map);
 
   if (!res && fields->elements)
   {
@@ -3790,11 +3841,18 @@ select_insert::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
 int select_insert::prepare2(JOIN *)
 {
   DBUG_ENTER("select_insert::prepare2");
+  if (table->validate_default_values_of_unset_fields(thd))
+    DBUG_RETURN(1);
+  if (thd->lex->describe)
+    DBUG_RETURN(0);
   if (thd->lex->current_select->options & OPTION_BUFFER_RESULT &&
-      thd->locked_tables_mode <= LTM_LOCK_TABLES &&
-      !thd->lex->describe)
+      thd->locked_tables_mode <= LTM_LOCK_TABLES)
     table->file->ha_start_bulk_insert((ha_rows) 0);
-  if (table->validate_default_values_of_unset_fields(thd))
+
+  /* Same as the other variants of INSERT */
+  if (sel_result &&
+      sel_result->send_result_set_metadata(thd->lex->returning()->item_list,
+                                Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
     DBUG_RETURN(1);
   DBUG_RETURN(0);
 }
@@ -3809,6 +3867,7 @@ void select_insert::cleanup()
 select_insert::~select_insert()
 {
   DBUG_ENTER("~select_insert");
+  sel_result= NULL;
   if (table && table->is_created())
   {
     table->next_number_field=0;
@@ -3850,6 +3909,15 @@ int select_insert::send_data(List<Item> &values)
   }
 
   error= write_record(thd, table, &info);
+  /*
+    Sending the result set to the cliet after writing record. The reason is
+    same as other variants of insert.
+  */
+  if (sel_result && sel_result->send_data(thd->lex->returning()->item_list) < 0)
+  {
+    error= 1;
+    DBUG_RETURN(1);
+  }
   table->vers_write= table->versioned();
   table->auto_increment_field_not_null= FALSE;
 
@@ -4004,7 +4072,14 @@ bool select_insert::send_ok_packet() {
      thd->first_successful_insert_id_in_prev_stmt :
      (info.copied ? autoinc_value_of_last_inserted_row : 0));
 
-  ::my_ok(thd, row_count, id, message);
+  /*
+    Client expects an EOF/OK packet If LEX::has_returning and if result set
+    meta was sent. See explanation for other variants of INSERT.
+  */
+  if (sel_result)
+    sel_result->send_eof();
+  else
+    ::my_ok(thd, row_count, id, message);
 
   DBUG_RETURN(false);
 }
diff --git a/sql/sql_insert.h b/sql/sql_insert.h
index a37ed1f31e5..5b86f09d13b 100644
--- a/sql/sql_insert.h
+++ b/sql/sql_insert.h
@@ -31,7 +31,7 @@ bool mysql_prepare_insert(THD *thd, TABLE_LIST *table_list, TABLE *table,
 bool mysql_insert(THD *thd,TABLE_LIST *table,List<Item> &fields,
                   List<List_item> &values, List<Item> &update_fields,
                   List<Item> &update_values, enum_duplicates flag,
-                  bool ignore);
+                  bool ignore, select_result* result);
 void upgrade_lock_type_for_insert(THD *thd, thr_lock_type *lock_type,
                                   enum_duplicates duplic,
                                   bool is_multi_insert);
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index ebc66418937..a05264195fb 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -998,14 +998,16 @@ public:
   bool explainable()
   {
     /*
-      Save plans for child subqueries, when
-      (1) they are not parts of eliminated WHERE/ON clauses.
-      (2) they are not merged derived tables
-      (3) they are not hanging CTEs (they are needed for execution)
+      EXPLAIN/ANALYZE unit, when:
+      (1) if it's a subquery - it's not part of eliminated WHERE/ON clause.
+      (2) if it's a CTE - it's not hanging (needed for execution)
+      (3) if it's a derived - it's not merged
+      if it's not 1/2/3 - it's some weird internal thing, ignore it
     */
-    return !(item && item->eliminated) &&
-           !(derived && !derived->is_materialized_derived()) &&
-           !(with_element && (!derived || !derived->derived_result));
+    return item ? !item->eliminated :                           // (1)
+           with_element ? derived && derived->derived_result :  // (2)
+           derived ? derived->is_materialized_derived() :       // (3)
+           false;
   }
 
   void reset_distinct();
@@ -4531,6 +4533,11 @@ public:
   void stmt_purge_to(const LEX_CSTRING &to);
   bool stmt_purge_before(Item *item);
 
+  SELECT_LEX *returning()
+  { return &builtin_select; }
+  bool has_returning()
+  { return !builtin_select.item_list.is_empty(); }
+
 private:
   bool stmt_create_routine_start(const DDL_options_st &options)
   {
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index bbf00e591e3..0493176b264 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -4451,6 +4451,7 @@ mysql_execute_command(THD *thd)
   case SQLCOM_INSERT:
   {
     WSREP_SYNC_WAIT(thd, WSREP_SYNC_WAIT_BEFORE_INSERT_REPLACE);
+    select_result *sel_result= NULL;
     DBUG_ASSERT(first_table == all_tables && first_table != 0);
 
     WSREP_SYNC_WAIT(thd, WSREP_SYNC_WAIT_BEFORE_INSERT_REPLACE);
@@ -4471,9 +4472,41 @@ mysql_execute_command(THD *thd)
       break;
 
     MYSQL_INSERT_START(thd->query());
+    Protocol* save_protocol=NULL;
+
+    if (lex->has_returning())
+    {
+      status_var_increment(thd->status_var.feature_insert_returning);
+
+      /* This is INSERT ... RETURNING. It will return output to the client */
+      if (thd->lex->analyze_stmt)
+      {
+        /*
+          Actually, it is ANALYZE .. INSERT .. RETURNING. We need to produce
+          output and then discard it.
+        */
+        sel_result= new (thd->mem_root) select_send_analyze(thd);
+        save_protocol= thd->protocol;
+        thd->protocol= new Protocol_discard(thd);
+      }
+      else
+      {
+        if (!(sel_result= new (thd->mem_root) select_send(thd)))
+          goto error;
+      }
+    }
+
     res= mysql_insert(thd, all_tables, lex->field_list, lex->many_values,
-		      lex->update_list, lex->value_list,
-                      lex->duplicates, lex->ignore);
+                      lex->update_list, lex->value_list,
+                      lex->duplicates, lex->ignore, sel_result);
+    if (save_protocol)
+    {
+      delete thd->protocol;
+      thd->protocol= save_protocol;
+    }
+    if (!res && thd->lex->analyze_stmt)
+      res= thd->lex->explain->send_explain(thd);
+    delete sel_result;
     MYSQL_INSERT_DONE(res, (ulong) thd->get_row_count_func());
     /*
       If we have inserted into a VIEW, and the base table has
@@ -4505,6 +4538,7 @@ mysql_execute_command(THD *thd)
   {
     WSREP_SYNC_WAIT(thd, WSREP_SYNC_WAIT_BEFORE_INSERT_REPLACE);
     select_insert *sel_result;
+    select_result *result= NULL;
     bool explain= MY_TEST(lex->describe);
     DBUG_ASSERT(first_table == all_tables && first_table != 0);
     WSREP_SYNC_WAIT(thd, WSREP_SYNC_WAIT_BEFORE_UPDATE_DELETE);
@@ -4553,6 +4587,31 @@ mysql_execute_command(THD *thd)
         Only the INSERT table should be merged. Other will be handled by
         select.
       */
+
+      Protocol* save_protocol=NULL;
+
+      if (lex->has_returning())
+      {
+        status_var_increment(thd->status_var.feature_insert_returning);
+
+        /* This is INSERT ... RETURNING. It will return output to the client */
+        if (thd->lex->analyze_stmt)
+        {
+          /*
+            Actually, it is ANALYZE .. INSERT .. RETURNING. We need to produce
+            output and then discard it.
+          */
+          result= new (thd->mem_root) select_send_analyze(thd);
+          save_protocol= thd->protocol;
+          thd->protocol= new Protocol_discard(thd);
+        }
+        else
+        {
+          if (!(result= new (thd->mem_root) select_send(thd)))
+            goto error;
+        }
+      }
+
       /* Skip first table, which is the table we are inserting in */
       TABLE_LIST *second_table= first_table->next_local;
       /*
@@ -4563,17 +4622,19 @@ mysql_execute_command(THD *thd)
         be done properly as well)
       */
       select_lex->table_list.first= second_table;
-      select_lex->context.table_list= 
+      select_lex->context.table_list=
         select_lex->context.first_name_resolution_table= second_table;
-      res= mysql_insert_select_prepare(thd);
-      if (!res && (sel_result= new (thd->mem_root) select_insert(thd,
-                                                             first_table,
-                                                             first_table->table,
-                                                             &lex->field_list,
-                                                             &lex->update_list,
-                                                             &lex->value_list,
-                                                             lex->duplicates,
-                                                             lex->ignore)))
+      res= mysql_insert_select_prepare(thd, result);
+      if (!res &&
+          (sel_result= new (thd->mem_root)
+                       select_insert(thd, first_table,
+                                    first_table->table,
+                                    &lex->field_list,
+                                    &lex->update_list,
+                                    &lex->value_list,
+                                    lex->duplicates,
+                                    lex->ignore,
+                                    result)))
       {
         if (lex->analyze_stmt)
           ((select_result_interceptor*)sel_result)->disable_my_ok_calls();
@@ -4609,7 +4670,12 @@ mysql_execute_command(THD *thd)
         }
         delete sel_result;
       }
-
+      delete result;
+      if (save_protocol)
+      {
+        delete thd->protocol;
+        thd->protocol= save_protocol;
+      }
       if (!res && (explain || lex->analyze_stmt))
         res= thd->lex->explain->send_explain(thd);
 
@@ -4644,7 +4710,7 @@ mysql_execute_command(THD *thd)
     MYSQL_DELETE_START(thd->query());
     Protocol *save_protocol= NULL;
 
-    if (!select_lex->item_list.is_empty())
+    if (lex->has_returning())
     {
       /* This is DELETE ... RETURNING.  It will return output to the client */
       if (thd->lex->analyze_stmt)
diff --git a/sql/sql_parse.h b/sql/sql_parse.h
index 1d25b898ca4..a90628781cc 100644
--- a/sql/sql_parse.h
+++ b/sql/sql_parse.h
@@ -41,7 +41,7 @@ bool multi_update_precheck(THD *thd, TABLE_LIST *tables);
 bool multi_delete_precheck(THD *thd, TABLE_LIST *tables);
 int mysql_multi_update_prepare(THD *thd);
 int mysql_multi_delete_prepare(THD *thd);
-bool mysql_insert_select_prepare(THD *thd);
+bool mysql_insert_select_prepare(THD *thd,select_result *sel_res);
 bool update_precheck(THD *thd, TABLE_LIST *tables);
 bool delete_precheck(THD *thd, TABLE_LIST *tables);
 bool insert_precheck(THD *thd, TABLE_LIST *tables);
diff --git a/sql/sql_prepare.cc b/sql/sql_prepare.cc
index 9a3013e5d47..e5723cb3b78 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -1482,8 +1482,6 @@ static bool mysql_test_delete(Prepared_statement *stmt,
   }
 
   DBUG_RETURN(mysql_prepare_delete(thd, table_list,
-                                   lex->first_select_lex()->with_wild,
-                                   lex->first_select_lex()->item_list,
                                    &lex->first_select_lex()->where,
                                    &delete_while_scanning));
 error:
@@ -2157,7 +2155,7 @@ static int mysql_insert_select_prepare_tester(THD *thd)
     thd->lex->first_select_lex()->context.first_name_resolution_table=
     second_table;
 
-  return mysql_insert_select_prepare(thd);
+  return mysql_insert_select_prepare(thd, NULL);
 }
 
 
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 5477d1f50bf..036c9aaf3e7 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -2080,7 +2080,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
         opt_persistent_stat_clause persistent_stat_spec
         persistent_column_stat_spec persistent_index_stat_spec
         table_column_list table_index_list table_index_name
-        check start checksum
+        check start checksum opt_returning
         field_list field_list_item kill key_def constraint_def
         keycache_list keycache_list_or_parts assign_to_keycache
         assign_to_keycache_parts
@@ -13477,7 +13477,8 @@ insert:
           {
             Select->set_lock_for_tables($4, true);
           }
-          insert_field_spec opt_insert_update stmt_end {}
+          insert_field_spec opt_insert_update opt_returning
+          stmt_end {}
           ;
 
 replace:
@@ -13490,7 +13491,8 @@ replace:
           {
             Select->set_lock_for_tables($4, true);
           }
-          insert_field_spec stmt_end {}
+          insert_field_spec opt_returning
+          stmt_end {}
           ;
 
 insert_start: {
@@ -13890,7 +13892,7 @@ single_multi:
           opt_where_clause
           opt_order_clause
           delete_limit_clause
-          opt_select_expressions 
+          opt_returning
           {
             if ($3)
               Select->order_list= *($3);
@@ -13920,9 +13922,28 @@ single_multi:
           } stmt_end {}
         ;
 
-opt_select_expressions:
-          /* empty */ 
-        | RETURNING_SYM select_item_list 
+opt_returning:
+          /* empty */
+          {
+            DBUG_ASSERT(!Lex->has_returning());
+          }
+        | RETURNING_SYM
+          {
+            DBUG_ASSERT(!Lex->has_returning());
+            if (($<num>$= (Select != Lex->returning())))
+            {
+              SELECT_LEX *sl= Lex->returning();
+              sl->set_master_unit(0);
+              Select->add_slave(Lex->create_unit(sl));
+              sl->include_global((st_select_lex_node**)&Lex->all_selects_list);
+              Lex->push_select(sl);
+            }
+          }
+          select_item_list
+          {
+            if ($<num>2)
+              Lex->pop_select();
+          }
         ;
 
 table_wild_list:
diff --git a/sql/sql_yacc_ora.yy b/sql/sql_yacc_ora.yy
index 976e239beeb..471364e29b7 100644
--- a/sql/sql_yacc_ora.yy
+++ b/sql/sql_yacc_ora.yy
@@ -1555,7 +1555,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
         opt_persistent_stat_clause persistent_stat_spec
         persistent_column_stat_spec persistent_index_stat_spec
         table_column_list table_index_list table_index_name
-        check start checksum
+        check start checksum opt_returning
         field_list field_list_item kill key_def constraint_def
         keycache_list keycache_list_or_parts assign_to_keycache
         assign_to_keycache_parts
@@ -13594,7 +13594,8 @@ insert:
           {
             Select->set_lock_for_tables($4, true);
           }
-          insert_field_spec opt_insert_update stmt_end {}
+          insert_field_spec opt_insert_update opt_returning
+          stmt_end {}
           ;
 
 replace:
@@ -13607,7 +13608,8 @@ replace:
           {
             Select->set_lock_for_tables($4, true);
           }
-          insert_field_spec stmt_end {}
+          insert_field_spec opt_returning
+          stmt_end {}
           ;
 
 insert_start: {
@@ -14007,7 +14009,7 @@ single_multi:
           opt_where_clause
           opt_order_clause
           delete_limit_clause
-          opt_select_expressions 
+          opt_returning
           {
             if ($3)
               Select->order_list= *($3);
@@ -14037,9 +14039,28 @@ single_multi:
           } stmt_end {}
         ;
 
-opt_select_expressions:
-          /* empty */ 
-        | RETURNING_SYM select_item_list 
+opt_returning:
+          /* empty */
+          {
+            DBUG_ASSERT(!Lex->has_returning());
+          }
+        | RETURNING_SYM
+          {
+            DBUG_ASSERT(!Lex->has_returning());
+            if (($<num>$= (Select != Lex->returning())))
+            {
+              SELECT_LEX *sl= Lex->returning();
+              sl->set_master_unit(0);
+              Select->add_slave(Lex->create_unit(sl));
+              sl->include_global((st_select_lex_node**)&Lex->all_selects_list);
+              Lex->push_select(sl);
+            }
+          }
+          select_item_list
+          {
+            if ($<num>2)
+              Lex->pop_select();
+          }
         ;
 
 table_wild_list: