Bug #17154 load data infile of char values into a table of char(PK) hangs

Bug #17158  	load data infile of char values into table of char with no (PK) fails to load
Bug #17081  	Doing "LOAD DATA INFILE" directly after delete can cause missing data


mysql-test/r/ndb_load.result:
  New BitKeeper file ``mysql-test/r/ndb_load.result''
mysql-test/t/ndb_load.test:
  New BitKeeper file ``mysql-test/t/ndb_load.test''

4 files changed, 127 insertions, 6 deletions

diff --git a/mysql-test/r/ndb_load.result b/mysql-test/r/ndb_load.result
new file mode 100644
index 00000000000..76da5b2a215
--- /dev/null
+++ b/mysql-test/r/ndb_load.result
@@ -0,0 +1,80 @@
+DROP TABLE IF EXISTS t1;
+CREATE TABLE t1 (word CHAR(20) NOT NULL PRIMARY KEY) ENGINE=NDB;
+LOAD DATA INFILE '../../std_data/words.dat' INTO TABLE t1 ;
+ERROR 23000: Can't write; duplicate key in table 't1'
+DROP TABLE t1;
+CREATE TABLE t1 (word CHAR(20) NOT NULL) ENGINE=NDB;
+LOAD DATA INFILE '../../std_data/words.dat' INTO TABLE t1 ;
+SELECT * FROM t1 ORDER BY word;
+word
+Aarhus
+Aarhus
+Aaron
+Aaron
+Ababa
+Ababa
+aback
+aback
+abaft
+abaft
+abandon
+abandon
+abandoned
+abandoned
+abandoning
+abandoning
+abandonment
+abandonment
+abandons
+abandons
+abase
+abased
+abasement
+abasements
+abases
+abash
+abashed
+abashes
+abashing
+abasing
+abate
+abated
+abatement
+abatements
+abater
+abates
+abating
+Abba
+abbe
+abbey
+abbeys
+abbot
+abbots
+Abbott
+abbreviate
+abbreviated
+abbreviates
+abbreviating
+abbreviation
+abbreviations
+Abby
+abdomen
+abdomens
+abdominal
+abduct
+abducted
+abduction
+abductions
+abductor
+abductors
+abducts
+Abe
+abed
+Abel
+Abelian
+Abelson
+Aberdeen
+Abernathy
+aberrant
+aberration
+DROP TABLE t1;
diff --git a/mysql-test/t/ndb_load.test b/mysql-test/t/ndb_load.test
new file mode 100644
index 00000000000..72a5b53eaad
--- /dev/null
+++ b/mysql-test/t/ndb_load.test
@@ -0,0 +1,24 @@
+-- source include/have_ndb.inc
+-- source include/not_embedded.inc
+
+--disable_warnings
+DROP TABLE IF EXISTS t1;
+--enable_warnings
+
+#
+# Basic test for different types of loading data
+#
+
+# should give duplicate key
+CREATE TABLE t1 (word CHAR(20) NOT NULL PRIMARY KEY) ENGINE=NDB;
+--error 1022
+LOAD DATA INFILE '../../std_data/words.dat' INTO TABLE t1 ;
+DROP TABLE t1;
+
+# now without a primary key we should be ok
+CREATE TABLE t1 (word CHAR(20) NOT NULL) ENGINE=NDB;
+LOAD DATA INFILE '../../std_data/words.dat' INTO TABLE t1 ;
+SELECT * FROM t1 ORDER BY word;
+DROP TABLE t1;
+
+# End of 4.1 tests
diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
index 594551b918a..f5c0f3dc173 100644
--- a/sql/ha_ndbcluster.cc
+++ b/sql/ha_ndbcluster.cc
@@ -3046,9 +3046,23 @@ int ha_ndbcluster::end_bulk_insert()
                         "rows_inserted:%d, bulk_insert_rows: %d", 
                         (int) m_rows_inserted, (int) m_bulk_insert_rows)); 
     m_bulk_insert_not_flushed= FALSE;
-    if (execute_no_commit(this,trans) != 0) {
-      no_uncommitted_rows_execute_failure();
-      my_errno= error= ndb_err(trans);
+    if (m_transaction_on)
+    {
+      if (execute_no_commit(this, trans) != 0)
+      {
+        no_uncommitted_rows_execute_failure();
+        my_errno= error= ndb_err(trans);
+      }
+    }
+    else
+    {
+      if (execute_commit(this, trans) != 0)
+      {
+        no_uncommitted_rows_execute_failure();
+        my_errno= error= ndb_err(trans);
+      }
+      int res= trans->restart();
+      DBUG_ASSERT(res == 0);
     }
   }
 
@@ -4867,7 +4881,7 @@ bool ha_ndbcluster::low_byte_first() const
 }
 bool ha_ndbcluster::has_transactions()
 {
-  return m_transaction_on;
+  return TRUE;
 }
 const char* ha_ndbcluster::index_type(uint key_number)
 {
diff --git a/sql/sql_load.cc b/sql/sql_load.cc
index aa4ea3e6c8c..4e6c458cc43 100644
--- a/sql/sql_load.cc
+++ b/sql/sql_load.cc
@@ -285,8 +285,11 @@ int mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
     else
       error=read_sep_field(thd,info,table,fields,read_info,*enclosed,
 			   skip_lines);
-    if (table->file->end_bulk_insert())
-      error=1;					/* purecov: inspected */
+    if (table->file->end_bulk_insert() && !error)
+    {
+      table->file->print_error(my_errno, MYF(0));
+      error= 1;
+    }
     table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
     table->next_number_field=0;
   }