added support of view and CHECK OPTION of view to LOAD DATA (BUG#5996)

mysql-test/r/view.result

@@ -1570,3 +1570,61 @@ select * from t1;
 s1
 drop view v2, v1;
 drop table t1;
+create table t1 (a int, b char(10));
+create view v1 as select * from t1 where a != 0 with check option;
+load data infile '../../std_data/loaddata3.dat' into table v1 fields terminated by '' enclosed by '' ignore 1 lines;
+ERROR HY000: CHECK OPTION failed 'test.v1'
+select * from t1;
+a	b
+1	row 1
+2	row 2
+select * from v1;
+a	b
+1	row 1
+2	row 2
+delete from t1;
+load data infile '../../std_data/loaddata3.dat' ignore into table v1 fields terminated by '' enclosed by '' ignore 1 lines;
+Warnings:
+Warning	1264	Out of range value adjusted for column 'a' at row 3
+Error	1369	CHECK OPTION failed 'test.v1'
+Warning	1264	Out of range value adjusted for column 'a' at row 4
+Error	1369	CHECK OPTION failed 'test.v1'
+select * from t1;
+a	b
+1	row 1
+2	row 2
+3	row 3
+select * from v1;
+a	b
+1	row 1
+2	row 2
+3	row 3
+drop view v1;
+drop table t1;
+create table t1 (a text, b text);
+create view v1 as select * from t1 where a <> 'Field A' with check option;
+load data infile '../../std_data/loaddata2.dat' into table v1 fields terminated by ',' enclosed by '''';
+ERROR HY000: CHECK OPTION failed 'test.v1'
+select concat('|',a,'|'), concat('|',b,'|') from t1;
+concat('|',a,'|')	concat('|',b,'|')
+select concat('|',a,'|'), concat('|',b,'|') from v1;
+concat('|',a,'|')	concat('|',b,'|')
+delete from t1;
+load data infile '../../std_data/loaddata2.dat' ignore into table v1 fields terminated by ',' enclosed by '''';
+Warnings:
+Error	1369	CHECK OPTION failed 'test.v1'
+Warning	1261	Row 2 doesn't contain data for all columns
+select concat('|',a,'|'), concat('|',b,'|') from t1;
+concat('|',a,'|')	concat('|',b,'|')
+|Field 1|	|Field 2' 
+Field 3,'Field 4|
+|Field 5' ,'Field 6|	NULL
+|Field 6|	| 'Field 7'|
+select concat('|',a,'|'), concat('|',b,'|') from v1;
+concat('|',a,'|')	concat('|',b,'|')
+|Field 1|	|Field 2' 
+Field 3,'Field 4|
+|Field 5' ,'Field 6|	NULL
+|Field 6|	| 'Field 7'|
+drop view v1;
+drop table t1;

mysql-test/t/view.test

@@ -1523,3 +1523,34 @@ select * from v2;
 select * from t1;
 drop view v2, v1;
 drop table t1;
+
+#
+# LOAD DATA with view and CHECK OPTION
+#
+# fixed length fields
+create table t1 (a int, b char(10));
+create view v1 as select * from t1 where a != 0 with check option;
+-- error 1369
+load data infile '../../std_data/loaddata3.dat' into table v1 fields terminated by '' enclosed by '' ignore 1 lines;
+select * from t1;
+select * from v1;
+delete from t1;
+load data infile '../../std_data/loaddata3.dat' ignore into table v1 fields terminated by '' enclosed by '' ignore 1 lines;
+select * from t1;
+select * from v1;
+drop view v1;
+drop table t1;
+# variable length fields
+create table t1 (a text, b text);
+create view v1 as select * from t1 where a <> 'Field A' with check option;
+-- error 1369
+load data infile '../../std_data/loaddata2.dat' into table v1 fields terminated by ',' enclosed by '''';
+select concat('|',a,'|'), concat('|',b,'|') from t1;
+select concat('|',a,'|'), concat('|',b,'|') from v1;
+delete from t1;
+load data infile '../../std_data/loaddata2.dat' ignore into table v1 fields terminated by ',' enclosed by '''';
+select concat('|',a,'|'), concat('|',b,'|') from t1;
+select concat('|',a,'|'), concat('|',b,'|') from v1;
+drop view v1;
+drop table t1;
+

sql/log_event.cc

@@ -2548,7 +2548,7 @@ int Load_log_event::exec_event(NET* net, struct st_relay_log_info* rli,
 	thd->net.pkt_nr = net->pkt_nr;
       }
       if (mysql_load(thd, &ex, &tables, field_list, handle_dup, net != 0,
-		     TL_WRITE))
+		     TL_WRITE, 0))
 	thd->query_error = 1;
       if (thd->cuted_fields)
       {

sql/mysql_priv.h

@@ -838,9 +838,10 @@ inline TABLE_LIST *find_table_in_local_list(TABLE_LIST *table,
 bool eval_const_cond(COND *cond);
 
 /* sql_load.cc */
-int mysql_load(THD *thd,sql_exchange *ex, TABLE_LIST *table_list,
+int mysql_load(THD *thd, sql_exchange *ex, TABLE_LIST *table_list,
 	       List<Item> &fields, enum enum_duplicates handle_duplicates,
-	       bool local_file,thr_lock_type lock_type);
+	       bool local_file, thr_lock_type lock_type,
+	       bool ignore_check_option_errors);
 int write_record(THD *thd, TABLE *table, COPY_INFO *info);
 
 /* sql_manager.cc */

sql/sql_lex.cc

@@ -1616,6 +1616,7 @@ bool st_lex::can_use_merged()
   case SQLCOM_INSERT_SELECT:
   case SQLCOM_REPLACE:
   case SQLCOM_REPLACE_SELECT:
+  case SQLCOM_LOAD:
     return TRUE;
   default:
     return FALSE;

sql/sql_load.cc

@@ -71,16 +71,19 @@ public:
   void set_io_cache_arg(void* arg) { cache.arg = arg; }
 };
 
-static int read_fixed_length(THD *thd,COPY_INFO &info,TABLE *table,
+static int read_fixed_length(THD *thd, COPY_INFO &info, TABLE_LIST *table_list,
 			     List<Item> &fields, READ_INFO &read_info,
-			     ulong skip_lines);
-static int read_sep_field(THD *thd,COPY_INFO &info,TABLE *table,
+			     ulong skip_lines,
+			     bool ignore_check_option_errors);
+static int read_sep_field(THD *thd, COPY_INFO &info, TABLE_LIST *table_list,
 			  List<Item> &fields, READ_INFO &read_info,
-			  String &enclosed, ulong skip_lines);
+			  String &enclosed, ulong skip_lines,
+			  bool ignore_check_option_errors);
 
 int mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
 	       List<Item> &fields, enum enum_duplicates handle_duplicates,
-	       bool read_file_from_client,thr_lock_type lock_type)
+	       bool read_file_from_client,thr_lock_type lock_type,
+	       bool ignore_check_option_errors)
 {
   char name[FN_REFLEN];
   File file;
@@ -88,6 +91,7 @@ int mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
   int error;
   String *field_term=ex->field_term,*escaped=ex->escaped;
   String *enclosed=ex->enclosed;
+  Item *unused_conds;
   bool is_fifo=0;
 #ifndef EMBEDDED_LIBRARY
   LOAD_FILE_INFO lf_info;
@@ -117,8 +121,9 @@ int mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
   table_list->lock_type= lock_type;
   if ((res= open_and_lock_tables(thd, table_list)))
     DBUG_RETURN(res);
-  /* TODO: add key check when we will support VIEWs in LOAD */
-  if (!table_list->updatable)
+  if (setup_tables(thd, table_list, &unused_conds))
+     DBUG_RETURN(-1);
+  if (!table_list->updatable || check_key_in_view(thd, table_list))
   {
     my_error(ER_NON_UPDATABLE_TABLE, MYF(0), table_list->alias, "LOAD");
     DBUG_RETURN(-1);
@@ -294,11 +299,12 @@ int mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
                               MODE_STRICT_ALL_TABLES)));
 
     if (!field_term->length() && !enclosed->length())
-      error=read_fixed_length(thd,info,table,fields,read_info,
-			      skip_lines);
+      error= read_fixed_length(thd, info, table_list, fields,read_info,
+			       skip_lines, ignore_check_option_errors);
     else
-      error=read_sep_field(thd,info,table,fields,read_info,*enclosed,
-			   skip_lines);
+      error= read_sep_field(thd, info, table_list, fields, read_info,
+			    *enclosed, skip_lines,
+			    ignore_check_option_errors);
     if (table->file->end_bulk_insert())
       error=1;					/* purecov: inspected */
     table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
@@ -401,11 +407,13 @@ err:
 ****************************************************************************/
 
 static int
-read_fixed_length(THD *thd,COPY_INFO &info,TABLE *table,List<Item> &fields,
-		  READ_INFO &read_info, ulong skip_lines)
+read_fixed_length(THD *thd, COPY_INFO &info, TABLE_LIST *table_list,
+		  List<Item> &fields, READ_INFO &read_info, ulong skip_lines,
+		  bool ignore_check_option_errors)
 {
   List_iterator_fast<Item> it(fields);
   Item_field *sql_field;
+  TABLE *table= table_list->table;
   ulonglong id;
   bool no_trans_update;
   DBUG_ENTER("read_fixed_length");
@@ -472,6 +480,17 @@ read_fixed_length(THD *thd,COPY_INFO &info,TABLE *table,List<Item> &fields,
                           ER_WARN_TOO_MANY_RECORDS, 
                           ER(ER_WARN_TOO_MANY_RECORDS), thd->row_count); 
     }
+
+    switch(table_list->view_check_option(thd,
+					 ignore_check_option_errors))
+    {
+    case VIEW_CHECK_SKIP:
+      read_info.next_line();
+      goto continue_loop;
+    case VIEW_CHECK_ERROR:
+      DBUG_RETURN(-1);
+    }
+
     if (thd->killed || write_record(thd,table,&info))
       DBUG_RETURN(1);
     thd->no_trans_update= no_trans_update;
@@ -496,6 +515,7 @@ read_fixed_length(THD *thd,COPY_INFO &info,TABLE *table,List<Item> &fields,
                           ER(ER_WARN_TOO_MANY_RECORDS), thd->row_count); 
     }
     thd->row_count++;
+continue_loop:;
   }
   if (id && !read_info.error)
     thd->insert_id(id);			// For binary/update log
@@ -505,12 +525,14 @@ read_fixed_length(THD *thd,COPY_INFO &info,TABLE *table,List<Item> &fields,
 
 
 static int
-read_sep_field(THD *thd,COPY_INFO &info,TABLE *table,
+read_sep_field(THD *thd, COPY_INFO &info, TABLE_LIST *table_list,
 	       List<Item> &fields, READ_INFO &read_info,
-	       String &enclosed, ulong skip_lines)
+	       String &enclosed, ulong skip_lines,
+	       bool ignore_check_option_errors)
 {
   List_iterator_fast<Item> it(fields);
   Item_field *sql_field;
+  TABLE *table= table_list->table;
   uint enclosed_length;
   ulonglong id;
   bool no_trans_update;
@@ -580,6 +602,18 @@ read_sep_field(THD *thd,COPY_INFO &info,TABLE *table,
                 	    ER(ER_WARN_TOO_FEW_RECORDS), thd->row_count);
       }
     }
+
+    switch(table_list->view_check_option(thd,
+					 ignore_check_option_errors))
+    {
+    case VIEW_CHECK_SKIP:
+      read_info.next_line();
+      goto continue_loop;
+    case VIEW_CHECK_ERROR:
+      DBUG_RETURN(-1);
+    }
+
+
     if (thd->killed || write_record(thd, table, &info))
       DBUG_RETURN(1);
     /*
@@ -605,6 +639,7 @@ read_sep_field(THD *thd,COPY_INFO &info,TABLE *table,
         DBUG_RETURN(1);
     }
     thd->row_count++;
+continue_loop:;
   }
   if (id && !read_info.error)
     thd->insert_id(id);			// For binary/update log

sql/sql_parse.cc

@@ -3135,7 +3135,8 @@ unsent_create_error:
 	goto error;
     }
     res= mysql_load(thd, lex->exchange, first_table, lex->field_list,
-                    lex->duplicates, (bool) lex->local_file, lex->lock_option);
+                    lex->duplicates, (bool) lex->local_file,
+		    lex->lock_option, lex->duplicates == DUP_IGNORE);
     break;
   }