MDEV-4001: Cassandra: server crashes in ha_cassandra::end_bulk_insert on...

MDEV-4001: Cassandra: server crashes in ha_cassandra::end_bulk_insert on INSERT .. SELECT with a non-existing column
- INSERT ... SELECT may call handler->end_bulk_insert() without having
  called handler->start_bulk_insert(). Let Cassandra SE handle this.

mysql-test/suite/plugins/r/cassandra.result

@@ -627,3 +627,10 @@ rowkey	b
 1	fooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo-bar
 2	qqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqq-baz
 drop table t1;
+#
+# MDEV-4001: Cassandra: server crashes in ha_cassandra::end_bulk_insert on INSERT .. SELECT with a non-existing column
+#
+create table t1 (rowkey int primary key, a int)  ENGINE=cassandra thrift_host='127.0.0.1' keyspace='mariadbtest2' column_family='cf14';
+insert into t1 (a) select b from t1;
+ERROR 42S22: Unknown column 'b' in 'field list'
+drop table t1;

mysql-test/suite/plugins/t/cassandra.test

@@ -93,6 +93,8 @@ create columnfamily cf12 (rowkey varchar primary key, decimal_col decimal);
 
 create columnfamily cf13 (rowkey int primary key, b blob);
 
+create columnfamily cf14 ( rowkey int primary key, a int );
+
 EOF
 --error 0,1,2
 --system cqlsh -3 -f $MYSQLTEST_VARDIR/cassandra_test_init.cql
@@ -727,6 +729,14 @@ insert into t1 values (2, 'qqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqq
 select * from t1;
 drop table t1;
 
+--echo #
+--echo # MDEV-4001: Cassandra: server crashes in ha_cassandra::end_bulk_insert on INSERT .. SELECT with a non-existing column
+--echo #
+create table t1 (rowkey int primary key, a int)  ENGINE=cassandra thrift_host='127.0.0.1' keyspace='mariadbtest2' column_family='cf14';
+--error ER_BAD_FIELD_ERROR
+insert into t1 (a) select b from t1;
+drop table t1;
+
 ############################################################################
 ## Cassandra cleanup
 ############################################################################
@@ -740,6 +750,7 @@ drop columnfamily cf6;
 drop columnfamily cf7;
 
 drop columnfamily cf13;
+drop columnfamily cf14;
 --enable_parsing
 ############################################################################
 ## Cassandra cleanup ends

storage/cassandra/ha_cassandra.cc

@@ -2041,6 +2041,12 @@ void ha_cassandra::start_bulk_insert(ha_rows rows, uint flags)
 int ha_cassandra::end_bulk_insert()
 {
   DBUG_ENTER("ha_cassandra::end_bulk_insert");
+  
+  if (!doing_insert_batch)
+  {
+    /* SQL layer can make end_bulk_insert call without start_bulk_insert call */
+    DBUG_RETURN(0);
+  }
 
   /* Flush out the insert buffer */
   doing_insert_batch= false;