Bug #19493 NDB does not ignore duplicate keys when using LOAD DATA LOCAL

- make sure to disable bulk insert when check for duplicate key is needed

mysql-test/r/ndb_loaddatalocal.result

+DROP TABLE IF EXISTS t1;
+create table t1(a int) engine=myisam;
+select * into outfile 'MYSQLTEST_VARDIR/master-data/ndb_loaddatalocal.select_outfile' from t1;
+drop table t1;
+create table t1(a int) engine=ndb;
+load data local infile 'MYSQLTEST_VARDIR/master-data/ndb_loaddatalocal.select_outfile' into table t1;
+select count(*) from t1;
+count(*)
+10000
+drop table t1;
+create table t1(a int) engine=myisam;
+insert into t1 values (1), (2), (2), (3);
+select * into outfile 'MYSQLTEST_VARDIR/master-data/ndb_loaddatalocal.select_outfile' from t1;
+drop table t1;
+create table t1(a int primary key) engine=ndb;
+load data local infile 'MYSQLTEST_VARDIR/master-data/ndb_loaddatalocal.select_outfile' into table t1;
+select * from t1 order by a;
+a
+1
+2
+3
+drop table t1;
+create table t1(a int) engine=myisam;
+insert into t1 values (1), (1), (2), (3);
+select * into outfile 'MYSQLTEST_VARDIR/master-data/ndb_loaddatalocal.select_outfile' from t1;
+drop table t1;
+create table t1(a int primary key) engine=ndb;
+load data local infile 'MYSQLTEST_VARDIR/master-data/ndb_loaddatalocal.select_outfile' into table t1;
+select * from t1 order by a;
+a
+1
+2
+3
+drop table t1;
+create table t1(a int) engine=myisam;
+insert into t1 values (1), (2), (3), (3);
+select * into outfile 'MYSQLTEST_VARDIR/master-data/ndb_loaddatalocal.select_outfile' from t1;
+drop table t1;
+create table t1(a int primary key) engine=ndb;
+load data local infile 'MYSQLTEST_VARDIR/master-data/ndb_loaddatalocal.select_outfile' into table t1;
+select * from t1 order by a;
+a
+1
+2
+3
+drop table t1;

mysql-test/t/ndb_loaddatalocal.test

+-- source include/have_ndb.inc
+-- source include/not_embedded.inc
+
+--disable_warnings
+DROP TABLE IF EXISTS t1;
+--enable_warnings
+
+create table t1(a int) engine=myisam;
+let $1=10000;
+disable_query_log;
+set SQL_LOG_BIN=0;
+while ($1)
+{
+ insert into t1 values(1);
+ dec $1;
+}
+set SQL_LOG_BIN=1;
+enable_query_log;
+--replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+eval select * into outfile '$MYSQLTEST_VARDIR/master-data/ndb_loaddatalocal.select_outfile' from t1;
+#This will generate a 20KB file, now test LOAD DATA LOCAL
+drop table t1;
+
+create table t1(a int) engine=ndb;
+--replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+eval load data local infile '$MYSQLTEST_VARDIR/master-data/ndb_loaddatalocal.select_outfile' into table t1;
+select count(*) from t1;
+system rm $MYSQLTEST_VARDIR/master-data/ndb_loaddatalocal.select_outfile ;
+drop table t1;
+
+create table t1(a int) engine=myisam;
+insert into t1 values (1), (2), (2), (3);
+--replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+eval select * into outfile '$MYSQLTEST_VARDIR/master-data/ndb_loaddatalocal.select_outfile' from t1;
+drop table t1;
+
+create table t1(a int primary key) engine=ndb;
+--replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+eval load data local infile '$MYSQLTEST_VARDIR/master-data/ndb_loaddatalocal.select_outfile' into table t1;
+system rm $MYSQLTEST_VARDIR/master-data/ndb_loaddatalocal.select_outfile;
+select * from t1 order by a;
+drop table t1;
+
+create table t1(a int) engine=myisam;
+insert into t1 values (1), (1), (2), (3);
+--replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+eval select * into outfile '$MYSQLTEST_VARDIR/master-data/ndb_loaddatalocal.select_outfile' from t1;
+drop table t1;
+
+create table t1(a int primary key) engine=ndb;
+--replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+eval load data local infile '$MYSQLTEST_VARDIR/master-data/ndb_loaddatalocal.select_outfile' into table t1;
+system rm $MYSQLTEST_VARDIR/master-data/ndb_loaddatalocal.select_outfile;
+select * from t1 order by a;
+drop table t1;
+
+create table t1(a int) engine=myisam;
+insert into t1 values (1), (2), (3), (3);
+--replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+eval select * into outfile '$MYSQLTEST_VARDIR/master-data/ndb_loaddatalocal.select_outfile' from t1;
+drop table t1;
+
+create table t1(a int primary key) engine=ndb;
+--replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+eval load data local infile '$MYSQLTEST_VARDIR/master-data/ndb_loaddatalocal.select_outfile' into table t1;
+system rm $MYSQLTEST_VARDIR/master-data/ndb_loaddatalocal.select_outfile;
+select * from t1 order by a;
+drop table t1;
+
+# End of 4.1 tests

sql/ha_ndbcluster.cc

@@ -1836,6 +1836,11 @@ int ha_ndbcluster::write_row(byte *record)
 
   if(m_ignore_dup_key && table->primary_key != MAX_KEY)
   {
+    /*
+      compare if expression with that in start_bulk_insert()
+      start_bulk_insert will set parameters to ensure that each
+      write_row is committed individually
+    */
     int peek_res= peek_row(record);
     
     if (!peek_res) 
@@ -2996,6 +3001,19 @@ void ha_ndbcluster::start_bulk_insert(ha_rows rows)
   DBUG_PRINT("enter", ("rows: %d", (int)rows));
   
   m_rows_inserted= (ha_rows) 0;
+  if (!m_use_write && m_ignore_dup_key)
+  {
+    /*
+      compare if expression with that in write_row
+      we have a situation where peek_row() will be called
+      so we cannot batch
+    */
+    DBUG_PRINT("info", ("Batching turned off as duplicate key is "
+                        "ignored by using peek_row"));
+    m_rows_to_insert= 1;
+    m_bulk_insert_rows= 1;
+    DBUG_VOID_RETURN;
+  }
   if (rows == (ha_rows) 0)
   {
     /* We don't know how many will be inserted, guess */