Bug #21056 ndb pushdown equal/setValue error on datetime: only pushdown like of string type fields

mysql-test/r/ndb_condition_pushdown.result

@@ -1782,6 +1782,28 @@ select * from t5 where b like '%jo%' order by a;
 a	b
 1	jonas
 3	johan
+set engine_condition_pushdown = off;
+select auto from t1 where date_time like '1902-02-02 %';
+auto
+2
+select auto from t1 where date_time not like '1902-02-02 %';
+auto
+3
+4
+set engine_condition_pushdown = on;
+explain select auto from t1 where date_time like '1902-02-02 %';
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	3	Using where
+select auto from t1 where date_time like '1902-02-02 %';
+auto
+2
+explain select auto from t1 where date_time not like '1902-02-02 %';
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	3	Using where
+select auto from t1 where date_time not like '1902-02-02 %';
+auto
+3
+4
 drop table t1;
 create table t1 (a int, b varchar(3), primary key using hash(a))
 engine=ndb;

mysql-test/t/ndb_condition_pushdown.test

@@ -1649,6 +1649,16 @@ set engine_condition_pushdown = on;
 explain select * from t5 where b like '%jo%';
 select * from t5 where b like '%jo%' order by a;
 
+# bug#21056  	ndb pushdown equal/setValue error on datetime
+set engine_condition_pushdown = off;
+select auto from t1 where date_time like '1902-02-02 %';
+select auto from t1 where date_time not like '1902-02-02 %';
+set engine_condition_pushdown = on;
+explain select auto from t1 where date_time like '1902-02-02 %';
+select auto from t1 where date_time like '1902-02-02 %';
+explain select auto from t1 where date_time not like '1902-02-02 %';
+select auto from t1 where date_time not like '1902-02-02 %';
+
 # bug#17421 -1
 drop table t1;
 create table t1 (a int, b varchar(3), primary key using hash(a))

sql/ha_ndbcluster.cc

@@ -6880,11 +6880,13 @@ void ndb_serialize_cond(const Item *item, void *arg)
             DBUG_PRINT("info", ("FIELD_ITEM"));
             DBUG_PRINT("info", ("table %s", tab->getName()));
             DBUG_PRINT("info", ("column %s", field->field_name));
+            DBUG_PRINT("info", ("type %d", field->type()));
             DBUG_PRINT("info", ("result type %d", field->result_type()));
             
             // Check that we are expecting a field and with the correct
             // result type
             if (context->expecting(Item::FIELD_ITEM) &&
+                context->expecting_field_type(field->type()) &&
                 (context->expecting_field_result(field->result_type()) ||
                  // Date and year can be written as string or int
                  ((type == MYSQL_TYPE_TIME ||
@@ -7104,6 +7106,9 @@ void ndb_serialize_cond(const Item *item, void *arg)
                                               func_item);      
             context->expect(Item::STRING_ITEM);
             context->expect(Item::FIELD_ITEM);
+            context->expect_only_field_type(MYSQL_TYPE_STRING);
+            context->expect_field_type(MYSQL_TYPE_VAR_STRING);
+            context->expect_field_type(MYSQL_TYPE_VARCHAR);
             context->expect_field_result(STRING_RESULT);
             context->expect(Item::FUNC_ITEM);
             break;

sql/ha_ndbcluster.h

@@ -362,8 +362,8 @@ class Ndb_cond_traverse_context
   Ndb_cond_traverse_context(TABLE *tab, void* ndb_tab, Ndb_cond_stack* stack)
     : table(tab), ndb_table(ndb_tab), 
     supported(TRUE), stack_ptr(stack), cond_ptr(NULL),
-    expect_mask(0), expect_field_result_mask(0), skip(0), collation(NULL),
-    rewrite_stack(NULL)
+    expect_mask(0), expect_field_type_mask(0), expect_field_result_mask(0), 
+    skip(0), collation(NULL), rewrite_stack(NULL)
   {
     if (stack)
       cond_ptr= stack->ndb_cond;
@@ -375,6 +375,7 @@ class Ndb_cond_traverse_context
   void expect(Item::Type type)
   {
     expect_mask|= (1 << type);
+    if (type == Item::FIELD_ITEM) expect_all_field_types();
   };
   void dont_expect(Item::Type type)
   {
@@ -394,6 +395,28 @@ class Ndb_cond_traverse_context
     expect(type);
   };
 
+  void expect_field_type(enum_field_types result)
+  {
+    expect_field_type_mask|= (1 << result);
+  };
+  void expect_all_field_types()
+  {
+    expect_field_type_mask= ~0;
+  };
+  bool expecting_field_type(enum_field_types result)
+  {
+    return (expect_field_type_mask & (1 << result));
+  };
+  void expect_no_field_type()
+  {
+    expect_field_type_mask= 0;
+  };
+  void expect_only_field_type(enum_field_types result)
+  {
+    expect_field_type_mask= 0;
+    expect_field_type(result);
+  };
+
   void expect_field_result(Item_result result)
   {
     expect_field_result_mask|= (1 << result);
@@ -429,6 +452,7 @@ class Ndb_cond_traverse_context
   Ndb_cond_stack* stack_ptr;
   Ndb_cond* cond_ptr;
   uint expect_mask;
+  uint expect_field_type_mask;
   uint expect_field_result_mask;
   uint skip;
   CHARSET_INFO* collation;