Merge mskold@bk-internal.mysql.com:/home/bk/mysql-5.1

into  mysql.com:/windows/Linux_space/MySQL/mysql-5.1


sql/ha_ndbcluster.cc:
  Auto merged

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 %' order by auto;
+auto
+2
+select auto from t1 where date_time not like '1902-02-02 %' order by auto;
+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 %' order by auto;
+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 %' order by auto;
+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 %' order by auto;
+select auto from t1 where date_time not like '1902-02-02 %' order by auto;
+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 %' order by auto;
+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 %' order by auto;
+
 # bug#17421 -1
 drop table t1;
 create table t1 (a int, b varchar(3), primary key using hash(a))

sql/ha_ndbcluster.cc

@@ -8592,11 +8592,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 ||
@@ -8816,6 +8818,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

@@ -461,8 +461,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;
@@ -474,6 +474,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)
   {
@@ -493,6 +494,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);
@@ -528,6 +551,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;

sql/ha_ndbcluster_binlog.cc

@@ -1579,10 +1579,12 @@ ndb_handle_schema_change(THD *thd, Ndb *ndb, NdbEventOperation *pOp,
                         dbname, tabname));
     build_table_filename(key, FN_LEN-1, dbname, tabname, NullS, 0);
     /*
-      If the frm of the altered table is different than the one on
-      disk then overwrite it with the new table definition
+      If the there is no local table shadowing the altered table and 
+      it has an frm that is different than the one on disk then 
+      overwrite it with the new table definition
     */
-    if (readfrm(key, &data, &length) == 0 &&
+    if (!ndbcluster_check_if_local_table(dbname, tabname) &&
+	readfrm(key, &data, &length) == 0 &&
         packfrm(data, length, &pack_data, &pack_length) == 0 &&
         cmp_frm(altered_table, pack_data, pack_length))
     {
@@ -1799,7 +1801,16 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb,
         // fall through
         case SOT_CREATE_TABLE:
           pthread_mutex_lock(&LOCK_open);
-          if (ndb_create_table_from_engine(thd, schema->db, schema->name))
+	  if (ndbcluster_check_if_local_table(schema->db, schema->name))
+	  {
+	    DBUG_PRINT("info", ("NDB binlog: Skipping locally defined table '%s.%s'",
+				schema->db, schema->name, schema->query));
+            sql_print_error("NDB binlog: Skipping locally defined table '%s.%s' from "
+                            "binlog schema event '%s' from node %d. ",
+                            schema->db, schema->name, schema->query,
+                            schema->node_id);
+	  }
+          else if (ndb_create_table_from_engine(thd, schema->db, schema->name))
           {
             sql_print_error("NDB binlog: Could not discover table '%s.%s' from "
                             "binlog schema event '%s' from node %d. "
@@ -2050,7 +2061,16 @@ ndb_binlog_thread_handle_schema_event_post_epoch(THD *thd,
             share= 0;
           }
           pthread_mutex_lock(&LOCK_open);
-          if (ndb_create_table_from_engine(thd, schema->db, schema->name))
+	  if (ndbcluster_check_if_local_table(schema->db, schema->name))
+	  {
+	    DBUG_PRINT("info", ("NDB binlog: Skipping locally defined table '%s.%s'",
+				schema->db, schema->name, schema->query));
+            sql_print_error("NDB binlog: Skipping locally defined table '%s.%s' from "
+                            "binlog schema event '%s' from node %d. ",
+                            schema->db, schema->name, schema->query,
+                            schema->node_id);
+	  }
+          else if (ndb_create_table_from_engine(thd, schema->db, schema->name))
 	  {
 	    sql_print_error("NDB binlog: Could not discover table '%s.%s' from "
                             "binlog schema event '%s' from node %d. my_errno: %d",
@@ -2290,6 +2310,28 @@ ndb_rep_event_name(String *event_name,const char *db, const char *tbl)
   }
 }
 
+bool
+ndbcluster_check_if_local_table(const char *dbname, const char *tabname)
+{
+  char key[FN_REFLEN];
+  char ndb_file[FN_REFLEN];
+
+  DBUG_ENTER("ndbcluster_check_if_local_table");
+  build_table_filename(key, FN_LEN-1, dbname, tabname, reg_ext, 0);
+  build_table_filename(ndb_file, FN_LEN-1, dbname, tabname, ha_ndb_ext, 0);
+  /* Check that any defined table is an ndb table */
+  DBUG_PRINT("info", ("Looking for file %s and %s", key, ndb_file));
+  if ((! my_access(key, F_OK)) && my_access(ndb_file, F_OK))
+  {
+    DBUG_PRINT("info", ("table file %s not on disk, local table", ndb_file));   
+  
+  
+    DBUG_RETURN(true);
+  }
+
+  DBUG_RETURN(false);
+}
+
 /*
   Common function for setting up everything for logging a table at
   create/discover.

sql/ha_ndbcluster_binlog.h

@@ -122,6 +122,8 @@ void ndbcluster_binlog_init_handlerton();
 */
 void ndbcluster_binlog_init_share(NDB_SHARE *share, TABLE *table);
 
+bool ndbcluster_check_if_local_table(const char *dbname, const char *tabname);
+
 int ndbcluster_create_binlog_setup(Ndb *ndb, const char *key,
                                    uint key_len,
                                    const char *db,