This is core patch for WL#6030.

This patch does the following things:
  - changes Field class so that it is possible to set NOT NULL fields
    temporary to NULL;
  - allows NOT NULL fields to accept NULL values temporarily during
    BEFORE-trigger processing;
  - checks that the NOT NULL constraint is enforced after
    BEFORE-trigger processing.
  - adds support for handling NULL fields in trigger BEFORE-insert
    during execution of statement LOAD DATA INFILE.

Details:
  This patch
    - changes the class Field to allow set the field in
      temporary value NULL;
    - extends interface of class Field by method check_constraints
      in order to check NOT NULL constraint for the field;
    - adds static overloaded functions check_record() into sql_base.cc
      that iterates over table's fields and checks NOT NULL constraint.
      These functions are called after trigger processing has been finished
      as part of execution function fill_records_n_invoke_before_triggers;
    - extends interface of class Table_triggers_list by two additional
      methods enable_fields_temporary_nullability() and
      disable_fields_temporary_nullability(). These methods are used in the
      function fill_record_n_invoke_before_triggers() to enable/disable
      temporary nullability for table's fields during execution of
      BEFORE trigger.
    - adds function check_that_all_fields_are_given_values(). This function
      is called after all fields that are used (explicitly or implicitly
      as part BEFORE INSERT trigger) in the statement INSERT have been
      prepared.

      - turns on temporary nullability for table's fields before start loading of
        file records during handling statement LOAD DATA INFILE. After records
        loading has been finished iterates over table's fields and checks
        NOT NULL constraint.

Author: Dmitry Shulga

mysql-test/suite/funcs_1/r/innodb_trig_09.result

@@ -198,7 +198,7 @@ a	NULL	Test 3.5.9.4-trig	00000	999	NULL
 select  @tr_var_b4_118, @tr_var_b4_121, @tr_var_b4_122,
 @tr_var_b4_136, @tr_var_b4_151, @tr_var_b4_163;
 @tr_var_b4_118	@tr_var_b4_121	@tr_var_b4_122	@tr_var_b4_136	@tr_var_b4_151	@tr_var_b4_163
-a	NULL	Test 3.5.9.4-trig	0	999	NULL
+a	NULL	Test 3.5.9.4-trig	NULL	999	NULL
 select  @tr_var_af_118, @tr_var_af_121, @tr_var_af_122,
 @tr_var_af_136, @tr_var_af_151, @tr_var_af_163;
 @tr_var_af_118	@tr_var_af_121	@tr_var_af_122	@tr_var_af_136	@tr_var_af_151	@tr_var_af_163

mysql-test/suite/funcs_1/r/memory_trig_09.result

@@ -199,7 +199,7 @@ a	NULL	Test 3.5.9.4-trig	00000	999	NULL
 select  @tr_var_b4_118, @tr_var_b4_121, @tr_var_b4_122,
 @tr_var_b4_136, @tr_var_b4_151, @tr_var_b4_163;
 @tr_var_b4_118	@tr_var_b4_121	@tr_var_b4_122	@tr_var_b4_136	@tr_var_b4_151	@tr_var_b4_163
-a	NULL	Test 3.5.9.4-trig	0	999	NULL
+a	NULL	Test 3.5.9.4-trig	NULL	999	NULL
 select  @tr_var_af_118, @tr_var_af_121, @tr_var_af_122,
 @tr_var_af_136, @tr_var_af_151, @tr_var_af_163;
 @tr_var_af_118	@tr_var_af_121	@tr_var_af_122	@tr_var_af_136	@tr_var_af_151	@tr_var_af_163

mysql-test/suite/funcs_1/r/myisam_trig_09.result

@@ -199,7 +199,7 @@ a	NULL	Test 3.5.9.4-trig	00000	999	NULL
 select  @tr_var_b4_118, @tr_var_b4_121, @tr_var_b4_122,
 @tr_var_b4_136, @tr_var_b4_151, @tr_var_b4_163;
 @tr_var_b4_118	@tr_var_b4_121	@tr_var_b4_122	@tr_var_b4_136	@tr_var_b4_151	@tr_var_b4_163
-a	NULL	Test 3.5.9.4-trig	0	999	NULL
+a	NULL	Test 3.5.9.4-trig	NULL	999	NULL
 select  @tr_var_af_118, @tr_var_af_121, @tr_var_af_122,
 @tr_var_af_136, @tr_var_af_151, @tr_var_af_163;
 @tr_var_af_118	@tr_var_af_121	@tr_var_af_122	@tr_var_af_136	@tr_var_af_151	@tr_var_af_163

sql/field.cc

@@ -1351,7 +1351,10 @@ String *Field::val_int_as_str(String *val_buffer, my_bool unsigned_val)
 Field::Field(uchar *ptr_arg,uint32 length_arg,uchar *null_ptr_arg,
 	     uchar null_bit_arg,
 	     utype unireg_check_arg, const char *field_name_arg)
-  :ptr(ptr_arg), m_null_ptr(null_ptr_arg),
+  :ptr(ptr_arg),
+   m_null_ptr(null_ptr_arg),
+   m_is_tmp_nullable(false),
+   m_is_tmp_null(false),
    table(0), orig_table(0), table_name(0),
    field_name(field_name_arg),
    unireg_check(unireg_check_arg),
@@ -1365,6 +1368,104 @@ Field::Field(uchar *ptr_arg,uint32 length_arg,uchar *null_ptr_arg,
 }
 
 
+/**
+  Check NOT NULL constraint on the field after temporary nullability is
+  disabled.
+
+  @param warning_no Warning to report.
+
+  @return TYPE_OK if the value is Ok, or corresponding error code from
+  the type_conversion_status enum.
+*/
+type_conversion_status Field::check_constraints(int warning_no) const
+{
+  /*
+    Ensure that Field::check_constraints() is called only when temporary
+    nullability is disabled.
+  */
+
+  DBUG_ASSERT(!is_tmp_nullable());
+
+  if (real_maybe_null())
+    return TYPE_OK; // If the field is nullable, we're Ok.
+
+  if (!m_is_tmp_null)
+    return TYPE_OK; // If the field was not NULL, we're Ok.
+
+  // The field has been set to NULL.
+
+  /*
+    If the field is of AUTO_INCREMENT, and the next number
+    has been assigned to it, we're Ok.
+  */
+
+  if (this == table->next_number_field)
+    return TYPE_OK;
+
+  /*
+    If the field is of TIMESTAMP its default value is CURRENT_TIMESTAMP,
+    so we're OK.
+  */
+
+  if (type() == MYSQL_TYPE_TIMESTAMP)
+    return TYPE_OK;
+
+  switch (m_count_cuted_fields_saved) {
+  case CHECK_FIELD_WARN:
+    set_warning(Sql_condition::SL_WARNING, warning_no, 1);
+    /* fall through */
+  case CHECK_FIELD_IGNORE:
+    return TYPE_OK;
+  case CHECK_FIELD_ERROR_FOR_NULL:
+    if (!table->in_use->no_errors)
+      my_error(ER_BAD_NULL_ERROR, MYF(0), field_name);
+    return TYPE_ERR_NULL_CONSTRAINT_VIOLATION;
+  }
+
+  DBUG_ASSERT(0); // impossible
+  return TYPE_ERR_NULL_CONSTRAINT_VIOLATION;
+}
+
+
+/**
+  Set field to value NULL.
+
+  @param row_offset    This is the offset between the row being updated
+                       and table->record[0]
+*/
+void Field::set_null(my_ptrdiff_t row_offset)
+{
+  if (real_maybe_null())
+  {
+    m_null_ptr[row_offset]|= null_bit;
+  }
+  else if (is_tmp_nullable())
+  {
+    m_is_tmp_null= true;
+    m_count_cuted_fields_saved= table->in_use->count_cuted_fields;
+  }
+}
+
+
+/**
+  Set field to value NOT NULL.
+
+  @param row_offset    This is the offset between the row being updated
+                       and table->record[0]
+*/
+void Field::set_notnull(my_ptrdiff_t row_offset)
+{
+  if (real_maybe_null())
+  {
+    m_null_ptr[row_offset]&= (uchar) ~null_bit;
+  }
+  else if (is_tmp_nullable())
+  {
+    m_is_tmp_null= false;
+  }
+}
+
+
 void Field::hash(ulong *nr, ulong *nr2)
 {
   if (is_null())
@@ -1395,7 +1496,8 @@ void Field::copy_data(my_ptrdiff_t src_record_offset)
     // Set to NULL if the source record is NULL, otherwise set to NOT-NULL.
     m_null_ptr[0]= (m_null_ptr[0]                 & ~null_bit) |
                    (m_null_ptr[src_record_offset] &  null_bit);
-  }
+  } else if (is_tmp_nullable())
+    m_is_tmp_null= false;
 }
 
 

sql/field.h

@@ -489,6 +489,27 @@ class Field
   */
   uchar *m_null_ptr;
 
+  /**
+    Flag: if the NOT-NULL field can be temporary NULL.
+  */
+  bool m_is_tmp_nullable;
+
+  /**
+    This is a flag with the following semantics:
+      - it can be changed only when m_is_tmp_nullable is true;
+      - it specifies if this field in the first current record
+        (TABLE::record[0]) was set to NULL (temporary NULL).
+
+    This flag is used for trigger handling.
+  */
+  bool m_is_tmp_null;
+
+  /**
+    The value of THD::count_cuted_fields at the moment of setting
+    m_is_tmp_null attribute.
+  */
+  enum_check_fields m_count_cuted_fields_saved;
+
 protected:
   const uchar *get_null_ptr() const
   { return m_null_ptr; }
@@ -551,6 +572,34 @@ class Field
         const char *field_name_arg);
   virtual ~Field() {}
 
+  /**
+    Turn on/off temporary nullability for the field.
+
+    @param is_tmp_nullable    NULLability flag:
+                                  true - turn on temporary NULLability
+                                  false - turn off temporary NULLability
+  */
+  void set_tmp_nullable(bool is_tmp_nullable)
+  { m_is_tmp_nullable= is_tmp_nullable; }
+
+  /**
+    Return temporary NULLability flag.
+
+    @return    true - if NULL can be assigned temporary to the Field
+               false - if NULL can not be assigned temporary to the Field
+  */
+  bool is_tmp_nullable() const
+  { return m_is_tmp_nullable; }
+
+  /**
+    Check whether Field has temporary value NULL.
+
+    @return    true - if the Field has temporary value NULL
+               false - if the Field's value is NOT NULL
+  */
+  bool is_tmp_null() const
+  { return is_tmp_nullable() && m_is_tmp_null; }
+
   /* Store functions returns 1 on overflow and -1 on fatal error */
   virtual type_conversion_status store(const char *to, uint length,
                                        const CHARSET_INFO *cs)=0;
@@ -869,6 +918,12 @@ class Field
   bool is_temporal_with_date_and_time() const
   { return is_temporal_type_with_date_and_time(type()); }
 
+  /**
+    Check whether the full table's row is NULL or the Field has value NULL.
+
+    @return    true if the full table's row is NULL or the Field has value NULL
+               false if neither table's row nor the Field has value NULL
+  */
   bool is_null(my_ptrdiff_t row_offset= 0) const
   {
     /*
@@ -883,27 +938,62 @@ class Field
       pointer, and its NULLity is recorded in the "null_bit" bit of
       m_null_ptr[row_offset].
     */
-    return table->null_row ? true : is_real_null(row_offset);
+    return table->null_row ?
+           true :
+           is_real_null(row_offset);
   }
 
-  bool is_real_null(my_ptrdiff_t row_offset= 0) const
-  { return real_maybe_null() ? test(m_null_ptr[row_offset] & null_bit) : false; }
-
-  bool is_null_in_record(const uchar *record) const
-  { return real_maybe_null() ? test(record[null_offset()] & null_bit) : false; }
+  /**
+    Check whether the Field has value NULL (temporary or actual).
 
-  void set_null(my_ptrdiff_t row_offset= 0)
+    @return   true if the Field has value NULL (temporary or actual)
+              false if the Field has value NOT NULL.
+  */
+  bool is_real_null(my_ptrdiff_t row_offset= 0) const
   {
     if (real_maybe_null())
-      m_null_ptr[row_offset]|= null_bit;
+      return test(m_null_ptr[row_offset] & null_bit);
+
+    if (is_tmp_nullable())
+      return m_is_tmp_null;
+
+    return false;
   }
 
-  void set_notnull(my_ptrdiff_t row_offset= 0)
+  /**
+    Check if the Field has value NULL or the record specified by argument
+    has value NULL for this Field.
+
+    @return    true if the Field has value NULL or the record has value NULL
+               for thois Field.
+  */
+  bool is_null_in_record(const uchar *record) const
   {
     if (real_maybe_null())
-      m_null_ptr[row_offset]&= (uchar) ~null_bit;
+      return test(record[null_offset()] & null_bit);
+
+    if (is_tmp_nullable())
+      return m_is_tmp_null;
+
+    return false;
   }
 
+  void set_null(my_ptrdiff_t row_offset= 0);
+
+  void set_notnull(my_ptrdiff_t row_offset= 0);
+
+  type_conversion_status check_constraints(int warning_no) const;
+
+  /**
+    Remember the value of THD::count_cuted_fields to handle possible
+    NOT-NULL constraint errors after BEFORE-trigger execution is finished.
+    We should save the value of THD::count_cuted_fields before starting
+    BEFORE-trigger processing since during triggers execution the
+    value of THD::count_cuted_fields could be changed.
+  */
+  void set_count_cuted_fields(enum_check_fields count_cuted_fields)
+  { m_count_cuted_fields_saved= count_cuted_fields; }
+
   bool maybe_null(void) const
   { return real_maybe_null() || table->maybe_null; }
 
@@ -1017,7 +1107,7 @@ class Field
   virtual void move_field_offset(my_ptrdiff_t ptr_diff)
   {
     ptr= ADD_TO_PTR(ptr, ptr_diff, uchar*);
-    if (m_null_ptr)
+    if (real_maybe_null())
       m_null_ptr= ADD_TO_PTR(m_null_ptr, ptr_diff, uchar*);
   }
 

sql/field_conv.cc

@@ -100,7 +100,7 @@ static void do_field_to_null_str(Copy_field *copy)
 
 type_conversion_status set_field_to_null(Field *field)
 {
-  if (field->real_maybe_null())
+  if (field->real_maybe_null() || field->is_tmp_nullable())
   {
     field->set_null();
     field->reset();
@@ -149,8 +149,20 @@ set_field_to_null_with_conversions(Field *field, bool no_conversions)
     field->reset();
     return TYPE_OK;
   }
+
   if (no_conversions)
-    return TYPE_ERR_NULL_CONSTRAINT_VIOLATION;
+  {
+    if (field->is_tmp_nullable())
+    {
+      field->set_null();
+      field->reset();
+      return TYPE_OK;
+    }
+    else
+    {
+      return TYPE_ERR_NULL_CONSTRAINT_VIOLATION;
+    }
+  }
 
   /*
     Check if this is a special type, which will get a special walue
@@ -166,7 +178,7 @@ set_field_to_null_with_conversions(Field *field, bool no_conversions)
     Item_func_now_local::store_in(field);
     return TYPE_OK;			// Ok to set time to NULL
   }
-  
+
   // Note: we ignore any potential failure of reset() here.
   field->reset();
 
@@ -175,6 +187,14 @@ set_field_to_null_with_conversions(Field *field, bool no_conversions)
     field->table->auto_increment_field_not_null= FALSE;
     return TYPE_OK;		        // field is set in fill_record()
   }
+
+  if (field->is_tmp_nullable())
+  {
+    field->set_null();
+    field->reset();
+    return TYPE_OK;
+  }
+
   switch (field->table->in_use->count_cuted_fields) {
   case CHECK_FIELD_WARN:
     field->set_warning(Sql_condition::SL_WARNING, ER_BAD_NULL_ERROR, 1);

sql/item.cc

@@ -2534,7 +2534,7 @@ adjust_max_effective_column_length(Field *field_par, uint32 max_length)
 void Item_field::set_field(Field *field_par)
 {
   field=result_field=field_par;			// for easy coding with fields
-  maybe_null=field->maybe_null();
+  maybe_null= field->maybe_null() || field->is_tmp_nullable();
   decimals= field->decimals();
   table_name= *field_par->table_name;
   field_name= field_par->field_name;

sql/item_sum.cc

@@ -971,7 +971,7 @@ bool Aggregator_distinct::add()
       return TRUE;
 
     for (Field **field=table->field ; *field ; field++)
-      if ((*field)->is_real_null(0))
+      if ((*field)->is_real_null())
         return 0;					// Don't count NULL
 
     if (tree)
@@ -3253,7 +3253,7 @@ bool Item_func_group_concat::add()
     if (!show_item->const_item())
     {
       Field *f= show_item->get_tmp_table_field();
-      if (f->is_real_null())
+      if (f->is_null_in_record((const uchar*) table->record[0]))
         return 0;                               // Skip row if it contains null
     }
   }