merge

include/mysqld_error.h

@@ -383,4 +383,6 @@
 #define ER_NO_DEFAULT_FOR_FIELD 1364
 #define ER_DIVISION_BY_ZERO 1365
 #define ER_TRUNCATED_WRONG_VALUE_FOR_FIELD 1366
-#define ER_ERROR_MESSAGES 367
+#define ER_VIEW_NONUPD_CHECK 1367
+#define ER_VIEW_CHECK_FAILED 1368
+#define ER_ERROR_MESSAGES 369

mysql-test/r/view.result

@@ -475,8 +475,10 @@ drop view v1;
 drop table t1;
 create table t1 (a int);
 create view v1 as select distinct a from t1 WITH CHECK OPTION;
-create view v2 as select distinct a from t1 WITH CASCADED CHECK OPTION;
-create view v3 as select distinct a from t1 WITH LOCAL CHECK OPTION;
+ERROR HY000: CHECK OPTION on non-updatable view 'test.v1'
+create view v1 as select a from t1 WITH CHECK OPTION;
+create view v2 as select a from t1 WITH CASCADED CHECK OPTION;
+create view v3 as select a from t1 WITH LOCAL CHECK OPTION;
 drop view v3 RESTRICT;
 drop view v2 CASCADE;
 drop view v1;
@@ -1338,3 +1340,87 @@ select * from t2;
 ERROR HY000: Table 't2' was not locked with LOCK TABLES
 drop view v1;
 drop table t1, t2;
+create table t1 (a int);
+create view v1 as select * from t1 where a < 2 with check option;
+insert into v1 values(1);
+insert into v1 values(3);
+ERROR HY000: CHECK OPTION failed 'test.v1'
+insert ignore into v1 values (2),(3),(0);
+Warnings:
+Error	1365	CHECK OPTION failed 'test.v1'
+Error	1365	CHECK OPTION failed 'test.v1'
+select * from t1;
+a
+1
+0
+delete from t1;
+insert into v1 SELECT 1;
+insert into v1 SELECT 3;
+ERROR HY000: CHECK OPTION failed 'test.v1'
+create table t2 (a int);
+insert into t2 values (2),(3),(0);
+insert ignore into v1 SELECT a from t2;
+Warnings:
+Error	1365	CHECK OPTION failed 'test.v1'
+Error	1365	CHECK OPTION failed 'test.v1'
+select * from t1;
+a
+1
+0
+update v1 set a=-1 where a=0;
+update v1 set a=2 where a=1;
+ERROR HY000: CHECK OPTION failed 'test.v1'
+select * from t1;
+a
+1
+-1
+update v1 set a=0 where a=0;
+insert into t2 values (1);
+update v1,t2 set v1.a=v1.a-1 where v1.a=t2.a;
+select * from t1;
+a
+0
+-1
+update v1 set a=a+1;
+update ignore v1,t2 set v1.a=v1.a+1 where v1.a=t2.a;
+Warnings:
+Error	1365	CHECK OPTION failed 'test.v1'
+select * from t1;
+a
+1
+1
+drop view v1;
+drop table t1, t2;
+create table t1 (a int);
+create view v1 as select * from t1 where a < 2 with check option;
+create view v2 as select * from v1 where a > 0 with local check option;
+create view v3 as select * from v1 where a > 0 with cascaded check option;
+insert into v2 values (1);
+insert into v3 values (1);
+insert into v2 values (0);
+ERROR HY000: CHECK OPTION failed 'test.v2'
+insert into v3 values (0);
+ERROR HY000: CHECK OPTION failed 'test.v3'
+insert into v2 values (2);
+insert into v3 values (2);
+ERROR HY000: CHECK OPTION failed 'test.v3'
+select * from t1;
+a
+1
+1
+2
+drop view v3,v2,v1;
+drop table t1;
+create table t1 (a int, primary key (a));
+create view v1 as select * from t1 where a < 2 with check option;
+insert into v1 values (1) on duplicate key update a=2;
+insert into v1 values (1) on duplicate key update a=2;
+ERROR HY000: CHECK OPTION failed 'test.v1'
+insert ignore into v1 values (1) on duplicate key update a=2;
+Warnings:
+Error	1365	CHECK OPTION failed 'test.v1'
+select * from t1;
+a
+1
+drop view v1;
+drop table t1;

mysql-test/t/view.test

@@ -394,9 +394,11 @@ drop table t1;
 # syntax compatibility
 #
 create table t1 (a int);
+-- error 1364
 create view v1 as select distinct a from t1 WITH CHECK OPTION;
-create view v2 as select distinct a from t1 WITH CASCADED CHECK OPTION;
-create view v3 as select distinct a from t1 WITH LOCAL CHECK OPTION;
+create view v1 as select a from t1 WITH CHECK OPTION;
+create view v2 as select a from t1 WITH CASCADED CHECK OPTION;
+create view v3 as select a from t1 WITH LOCAL CHECK OPTION;
 drop view v3 RESTRICT;
 drop view v2 CASCADE;
 drop view v1;
@@ -1308,3 +1310,80 @@ select * from v1;
 select * from t2;
 drop view v1;
 drop table t1, t2;
+
+#
+# WITH CHECK OPTION insert/update test
+#
+create table t1 (a int);
+create view v1 as select * from t1 where a < 2 with check option;
+# simple insert
+insert into v1 values(1);
+-- error 1365
+insert into v1 values(3);
+# simple insert with ignore
+insert ignore into v1 values (2),(3),(0);
+select * from t1;
+# prepare data for next check
+delete from t1;
+# INSERT SELECT test
+insert into v1 SELECT 1;
+-- error 1365
+insert into v1 SELECT 3;
+# prepare data for next check
+create table t2 (a int);
+insert into t2 values (2),(3),(0);
+# INSERT SELECT with ignore test
+insert ignore into v1 SELECT a from t2;
+select * from t1;
+#simple UPDATE test
+update v1 set a=-1 where a=0;
+-- error 1365
+update v1 set a=2 where a=1;
+select * from t1;
+# prepare data for next check
+update v1 set a=0 where a=0;
+insert into t2 values (1);
+# multiupdate test
+update v1,t2 set v1.a=v1.a-1 where v1.a=t2.a;
+select * from t1;
+# prepare data for next check
+update v1 set a=a+1;
+# multiupdate with ignore test
+update ignore v1,t2 set v1.a=v1.a+1 where v1.a=t2.a;
+select * from t1;
+
+drop view v1;
+drop table t1, t2;
+
+#
+# CASCADED/LOCAL CHECK OPTION test
+#
+create table t1 (a int);
+create view v1 as select * from t1 where a < 2 with check option;
+create view v2 as select * from v1 where a > 0 with local check option;
+create view v3 as select * from v1 where a > 0 with cascaded check option;
+insert into v2 values (1);
+insert into v3 values (1);
+-- error 1365
+insert into v2 values (0);
+-- error 1365
+insert into v3 values (0);
+insert into v2 values (2);
+-- error 1365
+insert into v3 values (2);
+select * from t1;
+drop view v3,v2,v1;
+drop table t1;
+
+#
+# CHECK OPTION with INSERT ... ON DUPLICATE KEY UPDATE
+#
+create table t1 (a int, primary key (a));
+create view v1 as select * from t1 where a < 2 with check option;
+insert into v1 values (1) on duplicate key update a=2;
+-- error 1365
+insert into v1 values (1) on duplicate key update a=2;
+insert ignore into v1 values (1) on duplicate key update a=2;
+select * from t1;
+drop view v1;
+drop table t1;

sql/share/czech/errmsg.txt

@@ -395,3 +395,5 @@ character-set=latin2
 "Field '%-.64s' doesn't have a default value",
 "Division by 0",
 "Incorrect %-.32s value: '%-.128s' for column '%.64s' at row %ld",
+"CHECK OPTION on non-updatable view '%-.64s.%-.64s'"
+"CHECK OPTION failed '%-.64s.%-.64s'"

sql/share/danish/errmsg.txt

@@ -386,3 +386,5 @@ character-set=latin1
 "Field '%-.64s' doesn't have a default value",
 "Division by 0",
 "Incorrect %-.32s value: '%-.128s' for column '%.64s' at row %ld",
+"CHECK OPTION on non-updatable view '%-.64s.%-.64s'"
+"CHECK OPTION failed '%-.64s.%-.64s'"

sql/share/dutch/errmsg.txt

@@ -395,3 +395,5 @@ character-set=latin1
 "Field '%-.64s' doesn't have a default value",
 "Division by 0",
 "Incorrect %-.32s value: '%-.128s' for column '%.64s' at row %ld",
+"CHECK OPTION on non-updatable view '%-.64s.%-.64s'"
+"CHECK OPTION failed '%-.64s.%-.64s'"

sql/share/english/errmsg.txt

@@ -383,3 +383,5 @@ character-set=latin1
 "Field '%-.64s' doesn't have a default value",
 "Division by 0",
 "Incorrect %-.32s value: '%-.128s' for column '%.64s' at row %ld",
+"CHECK OPTION on non-updatable view '%-.64s.%-.64s'"
+"CHECK OPTION failed '%-.64s.%-.64s'"

sql/share/estonian/errmsg.txt

@@ -388,3 +388,5 @@ character-set=latin7
 "Field '%-.64s' doesn't have a default value",
 "Division by 0",
 "Incorrect %-.32s value: '%-.128s' for column '%.64s' at row %ld",
+"CHECK OPTION on non-updatable view '%-.64s.%-.64s'"
+"CHECK OPTION failed '%-.64s.%-.64s'"

sql/share/french/errmsg.txt

@@ -383,3 +383,5 @@ character-set=latin1
 "Field '%-.64s' doesn't have a default value",
 "Division by 0",
 "Incorrect %-.32s value: '%-.128s' for column '%.64s' at row %ld",
+"CHECK OPTION on non-updatable view '%-.64s.%-.64s'"
+"CHECK OPTION failed '%-.64s.%-.64s'"

sql/share/german/errmsg.txt

@@ -396,3 +396,5 @@ character-set=latin1
 "Field '%-.64s' doesn't have a default value",
 "Division by 0",
 "Incorrect %-.32s value: '%-.128s' for column '%.64s' at row %ld",
+"CHECK OPTION on non-updatable view '%-.64s.%-.64s'"
+"CHECK OPTION failed '%-.64s.%-.64s'"

sql/share/greek/errmsg.txt

@@ -383,3 +383,5 @@ character-set=greek
 "Field '%-.64s' doesn't have a default value",
 "Division by 0",
 "Incorrect %-.32s value: '%-.128s' for column '%.64s' at row %ld",
+"CHECK OPTION on non-updatable view '%-.64s.%-.64s'"
+"CHECK OPTION failed '%-.64s.%-.64s'"

sql/share/hungarian/errmsg.txt

@@ -388,3 +388,5 @@ character-set=latin2
 "Field '%-.64s' doesn't have a default value",
 "Division by 0",
 "Incorrect %-.32s value: '%-.128s' for column '%.64s' at row %ld",
+"CHECK OPTION on non-updatable view '%-.64s.%-.64s'"
+"CHECK OPTION failed '%-.64s.%-.64s'"

sql/share/italian/errmsg.txt

@@ -383,3 +383,5 @@ character-set=latin1
 "Field '%-.64s' doesn't have a default value",
 "Division by 0",
 "Incorrect %-.32s value: '%-.128s' for column '%.64s' at row %ld",
+"CHECK OPTION on non-updatable view '%-.64s.%-.64s'"
+"CHECK OPTION failed '%-.64s.%-.64s'"

sql/share/japanese/errmsg.txt

@@ -387,3 +387,5 @@ character-set=ujis
 "Field '%-.64s' doesn't have a default value",
 "Division by 0",
 "Incorrect %-.32s value: '%-.128s' for column '%.64s' at row %ld",
+"CHECK OPTION on non-updatable view '%-.64s.%-.64s'"
+"CHECK OPTION failed '%-.64s.%-.64s'"

sql/share/korean/errmsg.txt

@@ -383,3 +383,5 @@ character-set=euckr
 "Field '%-.64s' doesn't have a default value",
 "Division by 0",
 "Incorrect %-.32s value: '%-.128s' for column '%.64s' at row %ld",
+"CHECK OPTION on non-updatable view '%-.64s.%-.64s'"
+"CHECK OPTION failed '%-.64s.%-.64s'"

sql/share/norwegian-ny/errmsg.txt

@@ -385,3 +385,5 @@ character-set=latin1
 "Field '%-.64s' doesn't have a default value",
 "Division by 0",
 "Incorrect %-.32s value: '%-.128s' for column '%.64s' at row %ld",
+"CHECK OPTION on non-updatable view '%-.64s.%-.64s'"
+"CHECK OPTION failed '%-.64s.%-.64s'"

sql/share/norwegian/errmsg.txt

@@ -385,3 +385,5 @@ character-set=latin1
 "Field '%-.64s' doesn't have a default value",
 "Division by 0",
 "Incorrect %-.32s value: '%-.128s' for column '%.64s' at row %ld",
+"CHECK OPTION on non-updatable view '%-.64s.%-.64s'"
+"CHECK OPTION failed '%-.64s.%-.64s'"

sql/share/polish/errmsg.txt

@@ -388,3 +388,5 @@ character-set=latin2
 "Field '%-.64s' doesn't have a default value",
 "Division by 0",
 "Incorrect %-.32s value: '%-.128s' for column '%.64s' at row %ld",
+"CHECK OPTION on non-updatable view '%-.64s.%-.64s'"
+"CHECK OPTION failed '%-.64s.%-.64s'"

sql/share/portuguese/errmsg.txt

@@ -385,3 +385,5 @@ character-set=latin1
 "Field '%-.64s' doesn't have a default value",
 "Division by 0",
 "Incorrect %-.32s value: '%-.128s' for column '%.64s' at row %ld",
+"CHECK OPTION on non-updatable view '%-.64s.%-.64s'"
+"CHECK OPTION failed '%-.64s.%-.64s'"

sql/share/romanian/errmsg.txt

@@ -388,3 +388,5 @@ character-set=latin2
 "Field '%-.64s' doesn't have a default value",
 "Division by 0",
 "Incorrect %-.32s value: '%-.128s' for column '%.64s' at row %ld",
+"CHECK OPTION on non-updatable view '%-.64s.%-.64s'"
+"CHECK OPTION failed '%-.64s.%-.64s'"

sql/share/russian/errmsg.txt

@@ -388,3 +388,5 @@ character-set=koi8r
 "Field '%-.64s' doesn't have a default value",
 "Division by 0",
 "Incorrect %-.32s value: '%-.128s' for column '%.64s' at row %ld",
+"CHECK OPTION   VIEW '%-.64s.%-.64s'"
+" CHECK OPTION  VIEW '%-.64s.%-.64s' "

sql/share/serbian/errmsg.txt

@@ -376,3 +376,5 @@ character-set=cp1250
 "Field '%-.64s' doesn't have a default value",
 "Division by 0",
 "Incorrect %-.32s value: '%-.128s' for column '%.64s' at row %ld",
+"CHECK OPTION on non-updatable view '%-.64s.%-.64s'"
+"CHECK OPTION failed '%-.64s.%-.64s'"

sql/share/slovak/errmsg.txt

@@ -391,3 +391,5 @@ character-set=latin2
 "Field '%-.64s' doesn't have a default value",
 "Division by 0",
 "Incorrect %-.32s value: '%-.128s' for column '%.64s' at row %ld",
+"CHECK OPTION on non-updatable view '%-.64s.%-.64s'"
+"CHECK OPTION failed '%-.64s.%-.64s'"

sql/share/spanish/errmsg.txt

@@ -387,3 +387,5 @@ character-set=latin1
 "Field '%-.64s' doesn't have a default value",
 "Division by 0",
 "Incorrect %-.32s value: '%-.128s' for column '%.64s' at row %ld",
+"CHECK OPTION on non-updatable view '%-.64s.%-.64s'"
+"CHECK OPTION failed '%-.64s.%-.64s'"

sql/share/swedish/errmsg.txt

@@ -383,3 +383,5 @@ character-set=latin1
 "Field '%-.64s' doesn't have a default value",
 "Division by 0",
 "Incorrect %-.32s value: '%-.128s' for column '%.64s' at row %ld",
+"CHECK OPTION on non-updatable view '%-.64s.%-.64s'"
+"CHECK OPTION failed '%-.64s.%-.64s'"

sql/share/ukrainian/errmsg.txt

@@ -389,3 +389,5 @@ character-set=koi8u
 "Field '%-.64s' doesn't have a default value",
 "Division by 0",
 "Incorrect %-.32s value: '%-.128s' for column '%.64s' at row %ld",
+"CHECK OPTION  VIEW '%-.64s.%-.64s'     "
+"צ CHECK OPTION  VIEW '%-.64s.%-.64s'  "

sql/sql_class.h

@@ -227,6 +227,9 @@ typedef struct st_copy_info {
 /* for INSERT ... UPDATE */
   List<Item> *update_fields;
   List<Item> *update_values;
+/* for VIEW ... WITH CHECK OPTION */
+  TABLE_LIST *view;
+  bool ignore;
 } COPY_INFO;
 
 
@@ -1278,14 +1281,8 @@ class select_insert :public select_result {
   bool insert_into_view;
 
   select_insert(TABLE_LIST *table_list_par, TABLE *table_par,
-                List<Item> *fields_par, enum_duplicates duplic)
-    :table_list(table_list_par), table(table_par), fields(fields_par),
-     last_insert_id(0),
-     insert_into_view(table_list_par && table_list_par->view != 0)
-  {
-    bzero((char*) &info,sizeof(info));
-    info.handle_duplicates=duplic;
-  }
+                List<Item> *fields_par, enum_duplicates duplic,
+                bool ignore_check_option_errors);
   ~select_insert();
   int prepare(List<Item> &list, SELECT_LEX_UNIT *u);
   bool send_fields(List<Item> &list, uint flags) { return 0; }
@@ -1311,7 +1308,7 @@ class select_create: public select_insert {
 		 List<create_field> &fields_par,
 		 List<Key> &keys_par,
 		 List<Item> &select_fields,enum_duplicates duplic)
-    :select_insert (NULL, NULL, &select_fields, duplic), create_table(table),
+    :select_insert (NULL, NULL, &select_fields, duplic, 0), create_table(table),
     extra_fields(&fields_par),keys(&keys_par), create_info(create_info_par),
     lock(0)
     {}

sql/sql_insert.cc

@@ -130,6 +130,7 @@ int mysql_insert(THD *thd,TABLE_LIST *table_list,
   */
   bool log_on= (thd->options & OPTION_BIN_LOG) || (!(thd->master_access & SUPER_ACL));
   bool transactional_table, log_delayed;
+  bool ignore_err= (thd->lex->duplicates == DUP_IGNORE);
   uint value_count;
   ulong counter = 1;
   ulonglong id;
@@ -243,6 +244,8 @@ int mysql_insert(THD *thd,TABLE_LIST *table_list,
   info.handle_duplicates=duplic;
   info.update_fields=&update_fields;
   info.update_values=&update_values;
+  info.view= (table_list->view ? table_list : 0);
+  info.ignore= ignore_err;
   /*
     Count warnings for all inserts.
     For single line insert, generate an error if try to set a NOT NULL field
@@ -315,6 +318,14 @@ int mysql_insert(THD *thd,TABLE_LIST *table_list,
 	break;
       }
     }
+    if ((res= table_list->view_check_option(thd, ignore_err)) ==
+        VIEW_CHECK_SKIP)
+      continue;
+    else if (res == VIEW_CHECK_ERROR)
+    {
+      error= 1;
+      break;
+    }
 
     /*
       FIXME: Actually we should do this before
@@ -716,6 +727,7 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
       }
       if (info->handle_duplicates == DUP_UPDATE)
       {
+        int res= 0;
         /* we don't check for other UNIQUE keys - the first row
            that matches, is updated. If update causes a conflict again,
            an error is returned
@@ -724,6 +736,15 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
         restore_record(table,record[1]);
         if (fill_record(*info->update_fields, *info->update_values, 0))
           goto err;
+
+        /* CHECK OPTION for VIEW ... ON DUPLICATE KEY UPDATE ... */
+        if (info->view &&
+            (res= info->view->view_check_option(current_thd, info->ignore)) ==
+            VIEW_CHECK_SKIP)
+          break;
+        else if (res == VIEW_CHECK_ERROR)
+          goto err;
+
         if ((error=table->file->update_row(table->record[1],table->record[0])))
           goto err;
         info->updated++;
@@ -1635,6 +1656,11 @@ int mysql_insert_select_prepare(THD *thd)
 {
   LEX *lex= thd->lex;
   DBUG_ENTER("mysql_insert_select_prepare");
+  /*
+    SELECT_LEX do not belong to INSERT statement, so we can't add WHERE
+    clasue if table is VIEW
+  */
+  lex->query_tables->no_where_clause= 1;
   if (mysql_prepare_insert_check_table(thd, lex->query_tables,
                                        lex->field_list,
                                        &lex->select_lex.where))
@@ -1643,6 +1669,21 @@ int mysql_insert_select_prepare(THD *thd)
 }
 
 
+select_insert::select_insert(TABLE_LIST *table_list_par, TABLE *table_par,
+                             List<Item> *fields_par, enum_duplicates duplic,
+                             bool ignore_check_option_errors)
+  :table_list(table_list_par), table(table_par), fields(fields_par),
+   last_insert_id(0),
+   insert_into_view(table_list_par && table_list_par->view != 0)
+{
+  bzero((char*) &info,sizeof(info));
+  info.handle_duplicates=duplic;
+  if (table_list_par)
+    info.view= (table_list_par->view ? table_list_par : 0);
+  info.ignore= ignore_check_option_errors;
+}
+
+
 int
 select_insert::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
 {
@@ -1702,6 +1743,14 @@ bool select_insert::send_data(List<Item> &values)
     fill_record(*fields, values, 1);
   else
     fill_record(table->field, values, 1);
+  switch (table_list->view_check_option(thd,
+                                        thd->lex->duplicates == DUP_IGNORE))
+  {
+  case VIEW_CHECK_SKIP:
+    DBUG_RETURN(0);
+  case VIEW_CHECK_ERROR:
+    DBUG_RETURN(1);
+  }
   if (thd->net.report_error || write_record(thd, table, &info))
     DBUG_RETURN(1);
   if (table->next_number_field)		// Clear for next record

sql/sql_lex.h

@@ -691,6 +691,7 @@ typedef struct st_lex
   uint8 describe;
   uint8 derived_tables;
   uint8 create_view_algorithm;
+  uint8 create_view_check;
   bool drop_if_exists, drop_temporary, local_file, one_shot_set;
   bool in_comment, ignore_space, verbose, no_write_to_binlog;
   /* special JOIN::prepare mode: changing of query is prohibited */

sql/sql_parse.cc

@@ -2818,7 +2818,8 @@ mysql_execute_command(THD *thd)
       if ((res= mysql_insert_select_prepare(thd)))
         break;
       if ((result= new select_insert(first_table, first_table->table,
-                                     &lex->field_list, lex->duplicates)))
+                                     &lex->field_list, lex->duplicates,
+                                     lex->duplicates == DUP_IGNORE)))
       {
 	/* Skip first table, which is the table we are inserting in */
 	lex->select_lex.table_list.first= (byte*) first_table->next_local;

sql/sql_update.cc

@@ -97,10 +97,12 @@ int mysql_update(THD *thd,
 		 ha_rows limit,
 		 enum enum_duplicates handle_duplicates)
 {
-  bool 		using_limit=limit != HA_POS_ERROR;
+  bool		using_limit= limit != HA_POS_ERROR;
   bool		safe_update= thd->options & OPTION_SAFE_UPDATES;
   bool		used_key_is_modified, transactional_table, log_delayed;
+  bool          ignore_err= (thd->lex->duplicates == DUP_IGNORE);
   int		error=0;
+  int           res;
   uint		used_index;
 #ifndef NO_EMBEDDED_ACCESS_CHECKS
   uint		want_privilege;
@@ -152,7 +154,7 @@ int mysql_update(THD *thd,
 #endif
   {
     thd->lex->select_lex.no_wrap_view_item= 1;
-    int res= setup_fields(thd, 0, table_list, fields, 1, 0, 0);
+    res= setup_fields(thd, 0, table_list, fields, 1, 0, 0);
     thd->lex->select_lex.no_wrap_view_item= 0;
     if (res)
       DBUG_RETURN(-1);				/* purecov: inspected */
@@ -369,6 +371,18 @@ int mysql_update(THD *thd,
 
       if (compare_record(table, query_id))
       {
+        if ((res= table_list->view_check_option(thd, ignore_err)) !=
+            VIEW_CHECK_OK)
+        {
+          found--;
+          if (res == VIEW_CHECK_SKIP)
+            continue;
+          else if (res == VIEW_CHECK_ERROR)
+          {
+            error= 1;
+            break;
+          }
+        }
 	if (!(error=table->file->update_row((byte*) table->record[1],
 					    (byte*) table->record[0])))
 	{
@@ -979,6 +993,7 @@ multi_update::~multi_update()
 bool multi_update::send_data(List<Item> &not_used_values)
 {
   TABLE_LIST *cur_table;
+  bool ignore_err= (thd->lex->duplicates == DUP_IGNORE);
   DBUG_ENTER("multi_update::send_data");
 
   for (cur_table= update_tables; cur_table; cur_table= cur_table->next_local)
@@ -1011,6 +1026,15 @@ bool multi_update::send_data(List<Item> &not_used_values)
       if (compare_record(table, thd->query_id))
       {
 	int error;
+        if ((error= cur_table->view_check_option(thd, ignore_err)) !=
+            VIEW_CHECK_OK)
+        {
+          found--;
+          if (error == VIEW_CHECK_SKIP)
+            continue;
+          else if (error == VIEW_CHECK_ERROR)
+            DBUG_RETURN(1);
+        }
 	if (!updated++)
 	{
 	  /*

sql/sql_view.cc

@@ -302,7 +302,6 @@ static const int required_view_parameters= 7;
   Note that one should NOT change the order for this, as it's used by
   parse()
 */
-
 static File_option view_parameters[]=
 {{{(char*) "query", 5},		offsetof(TABLE_LIST, query),
   FILE_OPTIONS_STRING},
@@ -312,6 +311,8 @@ static File_option view_parameters[]=
   FILE_OPTIONS_ULONGLONG},
  {{(char*) "algorithm", 9},	offsetof(TABLE_LIST, algorithm),
   FILE_OPTIONS_ULONGLONG},
+ {{"with_check_option", 17},    offsetof(TABLE_LIST, with_check),
+   FILE_OPTIONS_ULONGLONG},
  {{(char*) "revision", 8},	offsetof(TABLE_LIST, revision),
   FILE_OPTIONS_REV},
  {{(char*) "timestamp", 9},	offsetof(TABLE_LIST, timestamp),
@@ -383,7 +384,7 @@ static int mysql_register_view(THD *thd, TABLE_LIST *view,
     char path_buff[FN_REFLEN];
     LEX_STRING path;
     File_parser *parser;
- 
+
     path.str= path_buff;
     fn_format(path_buff, file.str, dir.str, 0, MY_UNPACK_FILENAME);
     path.length= strlen(path_buff);
@@ -393,7 +394,7 @@ static int mysql_register_view(THD *thd, TABLE_LIST *view,
       if (mode == VIEW_CREATE_NEW)
       {
 	my_error(ER_TABLE_EXISTS_ERROR, MYF(0), view->alias);
-	DBUG_RETURN(1);
+	DBUG_RETURN(-1);
       }
 
       if (!(parser= sql_parse_prepare(&path, &thd->mem_root, 0)))
@@ -404,7 +405,7 @@ static int mysql_register_view(THD *thd, TABLE_LIST *view,
       {
         my_error(ER_WRONG_OBJECT, MYF(0), (view->db ? view->db : thd->db),
                  view->real_name, "VIEW");
-        DBUG_RETURN(1);
+        DBUG_RETURN(-1);
       }
 
       /*
@@ -416,7 +417,7 @@ static int mysql_register_view(THD *thd, TABLE_LIST *view,
       if (parser->parse((gptr)view, &thd->mem_root,
                         view_parameters + revision_number_position, 1))
       {
-        DBUG_RETURN(1);
+        DBUG_RETURN(thd->net.report_error? -1 : 0);
       }
     }
     else
@@ -424,7 +425,7 @@ static int mysql_register_view(THD *thd, TABLE_LIST *view,
       if (mode == VIEW_ALTER)
       {
 	my_error(ER_NO_SUCH_TABLE, MYF(0), view->db, view->alias);
-	DBUG_RETURN(1);
+	DBUG_RETURN(-1);
       }
     }
   }
@@ -446,6 +447,7 @@ static int mysql_register_view(THD *thd, TABLE_LIST *view,
     thd->lex->create_view_algorithm= VIEW_ALGORITHM_UNDEFINED;
   }
   view->algorithm= thd->lex->create_view_algorithm;
+  view->with_check= thd->lex->create_view_check;
   if ((view->updatable_view= (can_be_merged &&
                               view->algorithm != VIEW_ALGORITHM_TMPTABLE)))
   {
@@ -461,10 +463,18 @@ static int mysql_register_view(THD *thd, TABLE_LIST *view,
       }
     }
   }
+
+  if (view->with_check != VIEW_CHECK_NONE &&
+      !view->updatable_view)
+  {
+    my_error(ER_VIEW_NONUPD_CHECK, MYF(0), view->db, view->real_name);
+    DBUG_RETURN(-1);
+  }
+
   if (sql_create_definition_file(&dir, &file, view_file_type,
 				 (gptr)view, view_parameters, 3))
   {
-    DBUG_RETURN(1);
+    DBUG_RETURN(thd->net.report_error? -1 : 1);
   }
   DBUG_RETURN(0);
 }
@@ -720,6 +730,7 @@ mysql_make_view(File_parser *parser, TABLE_LIST *table)
     DBUG_PRINT("info", ("algorithm: TEMPORARY TABLE"));
     lex->select_lex.linkage= DERIVED_TABLE_TYPE;
     table->updatable= 0;
+    table->with_check= VIEW_CHECK_NONE;
 
     /* SELECT tree link */
     lex->unit.include_down(table->select_lex);

sql/sql_yacc.yy

@@ -7898,9 +7898,13 @@ algorithm:
 	  { Lex->create_view_algorithm= VIEW_ALGORITHM_TMPTABLE; }
 	;
 check_option:
-        /* empty */ {}
-        | WITH CHECK_SYM OPTION {}
-        | WITH CASCADED CHECK_SYM OPTION {}
-        | WITH LOCAL_SYM CHECK_SYM OPTION {}
+        /* empty */
+          { Lex->create_view_check= VIEW_CHECK_NONE; }
+        | WITH CHECK_SYM OPTION
+          { Lex->create_view_check= VIEW_CHECK_LOCAL; }
+        | WITH CASCADED CHECK_SYM OPTION
+          { Lex->create_view_check= VIEW_CHECK_CASCADED; }
+        | WITH LOCAL_SYM CHECK_SYM OPTION
+          { Lex->create_view_check= VIEW_CHECK_LOCAL; }
         ;
 

sql/table.cc

@@ -1595,33 +1595,69 @@ bool st_table_list::setup_ancestor(THD *thd, Item **conds)
     if (arena)
       thd->set_n_backup_item_arena(arena, &backup);
 
-    /* Go up to join tree and try to find left join */
-    for (; tbl; tbl= tbl->embedding)
+    if (with_check)
     {
-      if (tbl->outer_join)
+      check_option= where->copy_andor_structure(thd);
+      if (with_check == VIEW_CHECK_CASCADED)
       {
-        /*
-          Store WHERE condition to ON expression for outer join, because we
-          can't use WHERE to correctly execute jeft joins on VIEWs and this
-          expression will not be moved to WHERE condition (i.e. will be clean
-          correctly for PS/SP)
-        */
-        tbl->on_expr= and_conds(tbl->on_expr, where);
-        break;
+        check_option= and_conds(check_option, ancestor->check_option);
       }
     }
-    if (tbl == 0)
+
+    /*
+      check that it is not VIEW in which we insert with INSERT SELECT
+      (in this case we can't add view WHERE condition to main SELECT_LEX)
+    */
+    if (!no_where_clause)
     {
-      /*
-        It is conds of JOIN, but it will be stored in st_select_lex::prep_where
-        for next reexecution
-      */
-      *conds= and_conds(*conds, where);
+      /* Go up to join tree and try to find left join */
+      for (; tbl; tbl= tbl->embedding)
+      {
+        if (tbl->outer_join)
+        {
+          /*
+            Store WHERE condition to ON expression for outer join, because
+            we can't use WHERE to correctly execute jeft joins on VIEWs and
+            this expression will not be moved to WHERE condition (i.e. will
+            be clean correctly for PS/SP)
+          */
+          tbl->on_expr= and_conds(tbl->on_expr, where);
+          break;
+        }
+      }
+      if (tbl == 0)
+      {
+        if (outer_join)
+        {
+          /*
+            Store WHERE condition to ON expression for outer join, because
+            we can't use WHERE to correctly execute jeft joins on VIEWs and
+            this expression will not be moved to WHERE condition (i.e. will
+            be clean correctly for PS/SP)
+          */
+          on_expr= and_conds(on_expr, where);
+        }
+        else
+        {
+          /*
+            It is conds of JOIN, but it will be stored in
+            st_select_lex::prep_where for next reexecution
+          */
+          *conds= and_conds(*conds, where);
+        }
+      }
     }
 
     if (arena)
       thd->restore_backup_item_arena(arena, &backup);
   }
+  /*
+    fix_fields do not need tables, because new are only AND operation and we
+    just need recollect statistics
+  */
+  if (check_option && !check_option->fixed &&
+      check_option->fix_fields(thd, 0, &check_option))
+    goto err;
 
   /* full text function moving to current select */
   if (view->select_lex.ftfunc_list->elements)
@@ -1655,6 +1691,40 @@ bool st_table_list::setup_ancestor(THD *thd, Item **conds)
 }
 
 
+/*
+  check CHECK OPTION condition
+
+  SYNOPSIS
+    check_option()
+    ignore_failure ignore check option fail
+
+  RETURN
+    VIEW_CHECK_OK     OK
+    VIEW_CHECK_ERROR  FAILED
+    VIEW_CHECK_SKIP   FAILED, but continue
+*/
+
+int st_table_list::view_check_option(THD *thd, bool ignore_failure)
+{
+  if (check_option && check_option->val_int() == 0)
+  {
+    if (ignore_failure)
+    {
+      push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_ERROR,
+                          ER_VIEW_CHECK_FAILED, ER(ER_VIEW_CHECK_FAILED),
+                          view_db.str, view_name.str);
+      return(VIEW_CHECK_SKIP);
+    }
+    else
+    {
+      my_error(ER_VIEW_CHECK_FAILED, MYF(0), view_db.str, view_name.str);
+      return(VIEW_CHECK_ERROR);
+    }
+  }
+  return(VIEW_CHECK_OK);
+}
+
+
 void Field_iterator_view::set(TABLE_LIST *table)
 {
   ptr= table->field_translation;

sql/table.h

@@ -188,6 +188,16 @@ struct st_table {
 #define VIEW_ALGORITHM_TMPTABLE	1
 #define VIEW_ALGORITHM_MERGE		2
 
+/* view WITH CHECK OPTION parameter options */
+#define VIEW_CHECK_NONE       0
+#define VIEW_CHECK_LOCAL      1
+#define VIEW_CHECK_CASCADED   2
+
+/* result of view WITH CHECK OPTION parameter check */
+#define VIEW_CHECK_OK         0
+#define VIEW_CHECK_ERROR      1
+#define VIEW_CHECK_SKIP       2
+
 struct st_lex;
 
 typedef struct st_table_list
@@ -223,6 +233,7 @@ typedef struct st_table_list
   /* next_global before adding VIEW tables */
   st_table_list	*old_next;
   Item          *where;                 /* VIEW WHERE clause condition */
+  Item          *check_option;          /* WITH CHECK OPTION condition */
   LEX_STRING	query;			/* text of (CRETE/SELECT) statement */
   LEX_STRING	md5;			/* md5 of query tesxt */
   LEX_STRING	source;			/* source of CREATE VIEW */
@@ -233,6 +244,7 @@ typedef struct st_table_list
   ulonglong     updatable_view;         /* VIEW can be updated */
   ulonglong	revision;		/* revision control number */
   ulonglong	algorithm;		/* 0 any, 1 tmp tables , 2 merging */
+  ulonglong     with_check;             /* WITH CHECK OPTION */
   uint          effective_algorithm;    /* which algorithm was really used */
   GRANT_INFO	grant;
   thr_lock_type lock_type;
@@ -244,6 +256,7 @@ typedef struct st_table_list
   bool          updating;               /* for replicate-do/ignore table */
   bool		force_index;		/* prefer index over table scan */
   bool          ignore_leaves;          /* preload only non-leaf nodes */
+  bool          no_where_clause;        /* do not attach WHERE to SELECT */
   table_map     dep_tables;             /* tables the table depends on      */
   table_map     on_expr_dep_tables;     /* tables on expression depends on  */
   struct st_nested_join *nested_join;   /* if the element is a nested join  */
@@ -262,6 +275,7 @@ typedef struct st_table_list
 
   void calc_md5(char *buffer);
   void set_ancestor();
+  int view_check_option(THD *thd, bool ignore_failure);
   bool setup_ancestor(THD *thd, Item **conds);
   bool placeholder() {return derived || view; }
   void print(THD *thd, String *str);