Condition pushdown to storage engine

16 files changed, 1093 insertions, 29 deletions

diff --git a/include/mysql_com.h b/include/mysql_com.h
index 6a03fe90eb5..c55cc997648 100644
--- a/include/mysql_com.h
+++ b/include/mysql_com.h
@@ -339,7 +339,7 @@ struct rand_struct {
 
   /* The following is for user defined functions */
 
-enum Item_result {STRING_RESULT, REAL_RESULT, INT_RESULT, ROW_RESULT};
+enum Item_result {STRING_RESULT= 0, REAL_RESULT, INT_RESULT, ROW_RESULT};
 
 typedef struct st_udf_args
 {
diff --git a/mysql-test/r/ndb_condition_pushdown.result b/mysql-test/r/ndb_condition_pushdown.result
new file mode 100644
index 00000000000..168ebdc0cfa
--- /dev/null
+++ b/mysql-test/r/ndb_condition_pushdown.result
@@ -0,0 +1,26 @@
+DROP TABLE IF EXISTS t1,t2;
+CREATE TABLE t1 (pk1 int unsigned NOT NULL PRIMARY KEY,   attr1 int unsigned NOT NULL,   attr2 int unsigned,   attr3 VARCHAR(10) ) ENGINE=ndbcluster;
+insert into t1 values (0,0,0, "a"),(1,1,1,"b"),(2,2,NULL,NULL),(3,3,3,"d"),(4,4,4,"e"),(5,5,5,"f");
+CREATE TABLE  t2 (pk1 int unsigned NOT NULL PRIMARY KEY,   attr1 int unsigned NOT NULL,   attr2 int unsigned,   attr3 VARCHAR(10) ) ENGINE=ndbcluster;
+insert into t2 values (0,0,0, "a"),(1,1,1,"b"),(2,2,2,"c"),(3,3,3,"d"),(4,4,4,"e"),(5,5,5,"f");
+set @old_ndbcpd = @@session.ndb_condition_pushdown;
+set ndb_condition_pushdown = off;
+select * from t1 where attr3 is null or attr1 > 2 and pk1= 3 order by pk1;
+pk1	attr1	attr2	attr3
+2	2	NULL	NULL
+3	3	3	d
+select * from t1,t2 where t1.attr1 > 1 and t1.attr2 = t2.attr2 and t2.attr1 < 5 order by t1.pk1;
+pk1	attr1	attr2	attr3	pk1	attr1	attr2	attr3
+3	3	3	d	3	3	3	d
+4	4	4	e	4	4	4	e
+set ndb_condition_pushdown = on;
+select * from t1 where attr3 is null or attr1 > 2 and pk1= 3 order by pk1;
+pk1	attr1	attr2	attr3
+2	2	NULL	NULL
+3	3	3	d
+select * from t1,t2 where t1.attr1 > 1 and t1.attr2 = t2.attr2 and t2.attr1 < 5 order by t1.pk1;
+pk1	attr1	attr2	attr3	pk1	attr1	attr2	attr3
+3	3	3	d	3	3	3	d
+4	4	4	e	4	4	4	e
+set ndb_condition_pushdown = @old_ndbcpd;
+DROP TABLE t1,t2;
diff --git a/mysql-test/t/ndb_condition_pushdown.test b/mysql-test/t/ndb_condition_pushdown.test
new file mode 100644
index 00000000000..16e366b8dfb
--- /dev/null
+++ b/mysql-test/t/ndb_condition_pushdown.test
@@ -0,0 +1,26 @@
+-- source include/have_ndb.inc
+
+--disable_warnings
+DROP TABLE IF EXISTS t1,t2;
+--enable_warnings
+
+#
+# Test of condition pushdown to storage engine
+#
+CREATE TABLE t1 (pk1 int unsigned NOT NULL PRIMARY KEY,   attr1 int unsigned NOT NULL,   attr2 int unsigned,   attr3 VARCHAR(10) ) ENGINE=ndbcluster;
+
+insert into t1 values (0,0,0, "a"),(1,1,1,"b"),(2,2,NULL,NULL),(3,3,3,"d"),(4,4,4,"e"),(5,5,5,"f");
+
+CREATE TABLE  t2 (pk1 int unsigned NOT NULL PRIMARY KEY,   attr1 int unsigned NOT NULL,   attr2 int unsigned,   attr3 VARCHAR(10) ) ENGINE=ndbcluster;
+
+insert into t2 values (0,0,0, "a"),(1,1,1,"b"),(2,2,2,"c"),(3,3,3,"d"),(4,4,4,"e"),(5,5,5,"f");
+
+set @old_ndbcpd = @@session.ndb_condition_pushdown;
+set ndb_condition_pushdown = off;
+select * from t1 where attr3 is null or attr1 > 2 and pk1= 3 order by pk1;
+select * from t1,t2 where t1.attr1 > 1 and t1.attr2 = t2.attr2 and t2.attr1 < 5 order by t1.pk1;
+set ndb_condition_pushdown = on;
+select * from t1 where attr3 is null or attr1 > 2 and pk1= 3 order by pk1;
+select * from t1,t2 where t1.attr1 > 1 and t1.attr2 = t2.attr2 and t2.attr1 < 5 order by t1.pk1;
+set ndb_condition_pushdown = @old_ndbcpd;
+DROP TABLE t1,t2;
diff --git a/ndb/src/ndbapi/NdbDictionaryImpl.hpp b/ndb/src/ndbapi/NdbDictionaryImpl.hpp
index 3ea20e2af16..5b52c57ff97 100644
--- a/ndb/src/ndbapi/NdbDictionaryImpl.hpp
+++ b/ndb/src/ndbapi/NdbDictionaryImpl.hpp
@@ -531,7 +531,7 @@ NdbTableImpl::getColumn(const char * name){
     do {
       if(hashValue == (tmp & 0xFFFE)){
 	NdbColumnImpl* col = cols[tmp >> 16];
-	if(strcmp(name, col->m_name.c_str()) == 0){
+	if(strncmp(name, col->m_name.c_str(), NDB_MAX_ATTR_NAME_SIZE-1) == 0){
 	  return col;
 	}
       }
@@ -549,7 +549,7 @@ NdbTableImpl::getColumn(const char * name){
   } else {
     for(Uint32 i = 0; i<sz; i++){
       NdbColumnImpl* col = * cols++;
-      if(col != 0 && strcmp(name, col->m_name.c_str()) == 0)
+      if(col != 0 && strncmp(name, col->m_name.c_str(), NDB_MAX_ATTR_NAME_SIZE-1) == 0)
 	return col;
     }
   }
diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
index 2e395758304..f1e01369498 100644
--- a/sql/ha_ndbcluster.cc
+++ b/sql/ha_ndbcluster.cc
@@ -197,9 +197,211 @@ int execute_no_commit_ie(ha_ndbcluster *h, NdbConnection *trans)
 }
 
 /*
-  Place holder for ha_ndbcluster thread specific data
+  CPDH condition storage support
 */
+Ndb_item::Ndb_item(NDB_ITEM_TYPE item_type, 
+		   NDB_ITEM_QUALIFICATION item_qualification,
+		   const Item *item_value) 
+  : type(item_type), qualification(item_qualification)
+{ 
+  switch(item_type) {
+  case(NDB_VALUE):
+  {
+    switch(item_qualification.value_type) {
+    case(Item::STRING_ITEM): {
+      Ndb_item_string_value *string_value = new Ndb_item_string_value();
+      Item_string *string_item= (Item_string *)item_value;
+      string_value->s= string_item->str_value;
+      string_value->c= string_item->collation.collation;
+      value.string_value= string_value;
+      break;
+    }
+    case(Item::INT_ITEM): {
+      value.int_value= ((Item_int *)item_value)->val_int();
+      break;
+    }
+    case(Item::REAL_ITEM): {
+      value.real_value= ((Item_real *)item_value)->val_real();
+      break;
+    }
+    case(Item::NULL_ITEM):
+      break;
+    case(Item::VARBIN_ITEM): {
+      Ndb_item_string_value *string_value = new Ndb_item_string_value();
+      Item_varbinary *varbin_item= (Item_varbinary *)item_value;
+      string_value->s= varbin_item->str_value;
+      string_value->c= varbin_item->collation.collation;
+      value.string_value= string_value;
+      break;
+    }
+    default:
+      break;
+    }
+  }
+  break;
+  case(NDB_FIELD): {
+    NDB_ITEM_FIELD_VALUE *field_value= new NDB_ITEM_FIELD_VALUE();
+    Item_field *field_item= (Item_field *) item_value;
+    field_value->field= field_item->field;
+    field_value->column_no= -1; // Will be fetched at scan filter generation
+    value.field_value= field_value;
+    break;
+  }
+  case(NDB_FUNCTION):
+    break;
+  }
+}
+ 
+Ndb_item::Ndb_item(longlong int_value) : type(NDB_VALUE)
+{
+  qualification.value_type= Item::INT_ITEM;
+  value.int_value= int_value; 
+}
+
+Ndb_item::Ndb_item(double real_value) : type(NDB_VALUE)
+{
+  qualification.value_type= Item::REAL_ITEM;
+  value.real_value= real_value; 
+}
+
+Ndb_item::Ndb_item(): type(NDB_VALUE)
+{
+  qualification.value_type= Item::NULL_ITEM;
+}
+
+Ndb_item::Ndb_item(Field *field, int column_no) : type(NDB_FIELD)
+{
+    NDB_ITEM_FIELD_VALUE *field_value= new NDB_ITEM_FIELD_VALUE();
+    qualification.field_type= field->type();
+    field_value->field= field;
+    field_value->column_no= column_no;
+    value.field_value= field_value;
+}
+
+Ndb_item::Ndb_item(Item_func::Functype func_type) : type(NDB_FUNCTION)
+{
+  qualification.function_type= func_type;
+}
+
+Ndb_item::~Ndb_item() 
+{ 
+  if (type == NDB_VALUE && 
+      (qualification.value_type == Item::STRING_ITEM ||
+       qualification.value_type == Item::VARBIN_ITEM))
+  {
+    delete value.string_value;
+    value.string_value= NULL;
+  }
+  else if (type == NDB_FIELD)
+  {
+    delete value.field_value;
+    value.field_value= NULL;
+  }
+}
+
+void Ndb_item::print(String* str)
+{
+  switch(type) {
+  case(NDB_VALUE):
+    str->append("[#NDB_VALUE ");
+    switch(qualification.value_type) {
+    case (Item::INT_ITEM): {
+      String tmp;
+      tmp.set(value.int_value, &my_charset_bin);
+      str->append(tmp);
+      break;
+    }
+    case (Item::REAL_ITEM): {
+      String tmp;
+      tmp.set(value.real_value, 4 , &my_charset_bin);
+      str->append(tmp);
+      break;
+    }
+    case (Item::STRING_ITEM): {
+      str->append(value.string_value->s.ptr());
+      break;
+    }
+    case (Item::VARBIN_ITEM): {
+      str->append(value.string_value->s.ptr());
+      break;
+    }
+    case (Item::NULL_ITEM):
+      str->append("NULL");
+      break;
+    default:
+      str->append("ILLEGAL VALUE");
+    }
+    str->append("]");
+    break;
+  case(NDB_FIELD):  
+    str->append("[#NDB_FIELD ");
+    str->append(value.field_value->field->field_name);
+    str->append("]");
+    break;
+  case(NDB_FUNCTION):
+    str->append("[#NDB_FUNCTION ");
+    switch(qualification.function_type) {
+    case(Item_func::UNKNOWN_FUNC): {
+      str->append("UNKNOWN]");
+      break;
+    }
+    case(Item_func::EQ_FUNC): {
+      str->append("=]");
+      break;
+    }
+    case(Item_func::NE_FUNC): {
+      str->append("!=]");
+      break;
+    }
+    case(Item_func::LT_FUNC): {
+      str->append("<]");
+      break;
+    }
+    case(Item_func::LE_FUNC): {
+      str->append("<=]");
+      break;
+    }
+    case(Item_func::GE_FUNC): {
+      str->append(">=]");
+      break;
+    }
+    case(Item_func::GT_FUNC): {
+      str->append(">]");
+      break;
+    }
+    case(Item_func::LIKE_FUNC): {
+      str->append("like]");
+      break;
+    }
+    case(Item_func::NOTLIKE_FUNC): {
+      str->append("notlike]");
+      break;
+    }
+    case(Item_func::ISNULL_FUNC): {
+      str->append("isnull]");
+      break;
+    }
+    case(Item_func::ISNOTNULL_FUNC): {
+      str->append("isnotnull]");
+      break;
+    }
+    case(Item_func::COND_AND_FUNC): {
+      str->append("and]");
+      break;
+    }
+    case(Item_func::COND_OR_FUNC): {
+      str->append("or]");
+      break;
+    }
+    default:
+      str->append("UNSUPPORTED]");
+    }
+  }
+}
 
+/*
+  Place holder for ha_ndbcluster thread specific data
+*/
 Thd_ndb::Thd_ndb()
 {
   ndb= new Ndb(g_ndb_cluster_connection, "");
@@ -1701,7 +1903,7 @@ int ha_ndbcluster::full_table_scan(byte *buf)
       op->readTuples(lm, 0, parallelism))
     ERR_RETURN(trans->getNdbError());
   m_active_cursor= op;
-  
+  generate_scan_filter(m_cond_stack, op);
   if((res= define_read_attrs(buf, op)))
     DBUG_RETURN(res);
 
@@ -2723,6 +2925,8 @@ int ha_ndbcluster::extra(enum ha_extra_function operation)
     break;
   case HA_EXTRA_RESET:                 /* Reset database to after open */
     DBUG_PRINT("info", ("HA_EXTRA_RESET"));
+    DBUG_PRINT("info", ("Clearing condition stack"));
+    cond_clear();
     break;
   case HA_EXTRA_CACHE:                 /* Cash record in HA_rrnd() */
     DBUG_PRINT("info", ("HA_EXTRA_CACHE"));
@@ -2916,14 +3120,6 @@ int ha_ndbcluster::extra_opt(enum ha_extra_function operation, ulong cache_size)
 }
 
 
-int ha_ndbcluster::reset()
-{
-  DBUG_ENTER("reset");
-  // Reset what?
-  DBUG_RETURN(1);
-}
-
-
 const char **ha_ndbcluster::bas_ext() const
 { static const char *ext[]= { ha_ndb_ext, NullS }; return ext; }
 
@@ -3832,7 +4028,8 @@ ha_ndbcluster::ha_ndbcluster(TABLE *table_arg):
   m_force_send(TRUE),
   m_autoincrement_prefetch(32),
   m_transaction_on(TRUE),
-  m_use_local_query_cache(FALSE)
+  m_use_local_query_cache(FALSE),
+  m_cond_stack(NULL)
 { 
   int i;
   
@@ -3877,6 +4074,10 @@ ha_ndbcluster::~ha_ndbcluster()
   }
   DBUG_ASSERT(m_active_trans == NULL);
 
+  // Discard the condition stack
+  DBUG_PRINT("info", ("Clearing condition stack"));
+  cond_clear();
+
   DBUG_VOID_RETURN;
 }
 
@@ -5141,4 +5342,583 @@ ha_ndbcluster::setup_recattr(const NdbRecAttr* curr)
 }
 #endif
 
+/*
+  Condition pushdown
+*/
+const 
+COND* 
+ha_ndbcluster::cond_push(const COND *cond) 
+{ 
+  THD *thd= current_thd;
+  Ndb_cond_stack *ndb_cond = new Ndb_cond_stack();
+  DBUG_ENTER("cond_push");
+
+  if (thd->variables.ndb_condition_pushdown)
+  {
+    DBUG_EXECUTE("where",print_where((COND *)cond, m_tabname););
+    if (m_cond_stack)
+      ndb_cond->next= m_cond_stack;
+    else
+      ndb_cond->next= NULL;
+    m_cond_stack= ndb_cond;
+    
+    if (serialize_cond(cond, ndb_cond))
+    {
+      DBUG_RETURN(NULL);
+    }
+    else
+    {
+      cond_pop();
+    }
+  }
+  DBUG_RETURN(cond); 
+}
+
+inline
+void 
+ha_ndbcluster::cond_pop() 
+{ 
+  Ndb_cond_stack *ndb_cond_stack= m_cond_stack;  
+  if (ndb_cond_stack)
+  {
+    m_cond_stack= ndb_cond_stack->next;
+    delete ndb_cond_stack;
+  }
+};
+
+void
+ha_ndbcluster::cond_clear()
+{
+  DBUG_ENTER("cond_clear");
+  while (m_cond_stack)
+    cond_pop();
+
+  DBUG_VOID_RETURN;
+}
+
+void ndb_serialize_cond(const Item *item, void *arg)
+{
+  Ndb_cond_traverse_context *context= (Ndb_cond_traverse_context *) arg;
+  DBUG_ENTER("ndb_serialize_cond");  
+
+  if (*context->supported_ptr)
+  {
+    Ndb_cond_stack *ndb_stack= context->stack_ptr;
+    Ndb_cond *prev_cond= context->cond_ptr;
+    Ndb_cond *curr_cond= context->cond_ptr= new Ndb_cond();
+    if (!ndb_stack->ndb_cond)
+      ndb_stack->ndb_cond= curr_cond;
+    curr_cond->prev= prev_cond;
+    if (prev_cond) prev_cond->next= curr_cond;
+    
+    switch(item->type()) {
+    case(Item::FIELD_ITEM): {
+      Item_field *field_item= (Item_field *) item;
+      Field *field= field_item->field;
+      /*
+	Check that the field is part of the table of the handler
+	instance and that we expect a field with of this result type.
+      */
+      if (context->table == field->table)
+      {	
+	const NDBTAB *tab= (const NDBTAB *) context->ndb_table;
+	const NDBCOL *col= tab->getColumn(field->field_name);
+	DBUG_ASSERT(col);
+	DBUG_PRINT("info", ("FIELD_ITEM"));
+	DBUG_PRINT("info", ("table %s", tab->getName()));
+	DBUG_PRINT("info", ("column %s", field->field_name));
+	
+	if(context->expecting(Item::FIELD_ITEM) &&
+	   context->expecting_field_result(field->result_type()))
+	{
+	  // Currently only support for unsigned int
+	  if (field->result_type() == INT_RESULT  &&
+	      (field->type() != MYSQL_TYPE_LONG ||
+	       !(field->flags & UNSIGNED_FLAG)))
+	    *context->supported_ptr= FALSE;  
+	  else
+	  {
+	    curr_cond->ndb_item= new Ndb_item(field, col->getColumnNo());
+	    context->dont_expect(Item::FIELD_ITEM);
+	    context->expect_no_field_result();
+	    break;
+	  }
+	}
+      }
+      *context->supported_ptr= FALSE;
+      break;
+    }
+    case(Item::FUNC_ITEM): {
+      Item_func *func_item= (Item_func *) item;
+
+      context->expect_nothing();
+      switch(func_item->functype()) {
+      case(Item_func::UNKNOWN_FUNC): {
+	DBUG_PRINT("info", ("UNKNOWN_FUNC"));      
+	DBUG_PRINT("info", ("value %d", func_item->val_int()));
+	break;
+      }
+      case(Item_func::EQ_FUNC): {
+	DBUG_PRINT("info", ("EQ_FUNC"));      
+	curr_cond->ndb_item= new Ndb_item(func_item->functype());      
+	context->expect(Item::FIELD_ITEM);
+	context->expect_field_result(INT_RESULT);
+	context->expect(Item::INT_ITEM);
+	break;
+      }
+      case(Item_func::NE_FUNC): {
+	DBUG_PRINT("info", ("NE_FUNC"));      
+	curr_cond->ndb_item= new Ndb_item(func_item->functype());      
+	context->expect(Item::FIELD_ITEM);
+	context->expect_field_result(INT_RESULT);
+	context->expect(Item::INT_ITEM);
+	break;
+      }
+      case(Item_func::LT_FUNC): {
+	DBUG_PRINT("info", ("LT_FUNC"));      
+	curr_cond->ndb_item= new Ndb_item(func_item->functype());      
+	context->expect(Item::FIELD_ITEM);
+	context->expect_field_result(INT_RESULT);
+	context->expect(Item::INT_ITEM);
+	break;
+      }
+      case(Item_func::LE_FUNC): {
+	DBUG_PRINT("info", ("LE_FUNC"));      
+	curr_cond->ndb_item= new Ndb_item(func_item->functype());      
+	context->expect(Item::FIELD_ITEM);
+	context->expect_field_result(INT_RESULT);
+	context->expect(Item::INT_ITEM);
+	break;
+      }
+      case(Item_func::GE_FUNC): {
+	DBUG_PRINT("info", ("GE_FUNC"));      
+	curr_cond->ndb_item= new Ndb_item(func_item->functype());      
+	context->expect(Item::FIELD_ITEM);
+	context->expect_field_result(INT_RESULT);
+	context->expect(Item::INT_ITEM);
+	break;
+      }
+      case(Item_func::GT_FUNC): {
+	DBUG_PRINT("info", ("GT_FUNC"));      
+	curr_cond->ndb_item= new Ndb_item(func_item->functype());      
+	context->expect(Item::FIELD_ITEM);
+	context->expect_field_result(INT_RESULT);
+	context->expect(Item::INT_ITEM);
+	break;
+      }
+      case(Item_func::LIKE_FUNC): {
+	DBUG_PRINT("info", ("LIKE_FUNC"));      
+	curr_cond->ndb_item= new Ndb_item(func_item->functype());      
+	context->expect(Item::STRING_ITEM);
+	*context->supported_ptr= FALSE; // Currently not supported
+	break;
+      }
+      case(Item_func::NOTLIKE_FUNC): {
+	DBUG_PRINT("info", ("NOTLIKE_FUNC"));      
+	curr_cond->ndb_item= new Ndb_item(func_item->functype());      
+	context->expect(Item::STRING_ITEM);
+	*context->supported_ptr= FALSE; // Currently not supported
+	break;
+      }
+      case(Item_func::ISNULL_FUNC): {
+	DBUG_PRINT("info", ("ISNULL_FUNC"));      
+	curr_cond->ndb_item= new Ndb_item(func_item->functype());      
+	context->expect(Item::FIELD_ITEM);
+	context->expect_field_result(STRING_RESULT);
+	context->expect_field_result(REAL_RESULT);
+	context->expect_field_result(INT_RESULT);
+	break;
+      }
+      case(Item_func::ISNOTNULL_FUNC): {
+	DBUG_PRINT("info", ("ISNOTNULL_FUNC"));      
+	curr_cond->ndb_item= new Ndb_item(func_item->functype());     
+	context->expect(Item::FIELD_ITEM);
+	context->expect_field_result(STRING_RESULT);
+	context->expect_field_result(REAL_RESULT);
+	context->expect_field_result(INT_RESULT);
+	break;
+      }
+      default: {
+	DBUG_PRINT("info", ("Found func_item of type %d", 
+			    func_item->functype()));
+	*context->supported_ptr= FALSE;
+      }
+      }
+      break;
+    }
+    case(Item::STRING_ITEM):
+      if (context->expecting(Item::STRING_ITEM)) 
+      {
+	char buff[256];
+	String str(buff,(uint32) sizeof(buff), system_charset_info);
+	str.length(0);
+	Item_string *string_item= (Item_string *) item;      
+	DBUG_PRINT("info", ("STRING_ITEM")); 
+	DBUG_PRINT("info", ("value \"%s\"", 
+			    string_item->val_str(&str)->ptr()));
+	NDB_ITEM_QUALIFICATION q;
+	q.value_type= Item::STRING_ITEM;
+	curr_cond->ndb_item= new Ndb_item(NDB_VALUE, q, item);      
+	context->dont_expect(Item::STRING_ITEM);
+      }
+      else
+	*context->supported_ptr= FALSE;
+      break;
+    case(Item::INT_ITEM): 
+      if (context->expecting(Item::INT_ITEM)) 
+      {
+	Item_int *int_item= (Item_int *) item;      
+	DBUG_PRINT("info", ("INT_ITEM"));
+	DBUG_PRINT("info", ("value %d", int_item->value));
+	curr_cond->ndb_item= new Ndb_item(int_item->value);
+	context->dont_expect(Item::INT_ITEM);
+      }
+      else
+	*context->supported_ptr= FALSE;
+      break;
+    case(Item::REAL_ITEM):
+      if (context->expecting(Item::REAL_ITEM)) 
+      {
+	Item_real *real_item= (Item_real *) item;      
+	DBUG_PRINT("info", ("REAL_ITEM %s"));
+	DBUG_PRINT("info", ("value %f", real_item->value));
+	curr_cond->ndb_item= new Ndb_item(real_item->value);
+	context->dont_expect(Item::REAL_ITEM);
+	*context->supported_ptr= FALSE; // Currently not supported
+      }
+      else
+	*context->supported_ptr= FALSE;
+      break;
+    case(Item::COND_ITEM): {
+      Item_cond *cond_item= (Item_cond *) item;
+      switch(cond_item->functype()) {
+      case(Item_func::COND_AND_FUNC):
+	DBUG_PRINT("info", ("COND_AND_FUNC"));
+	curr_cond->ndb_item= new Ndb_item(cond_item->functype());      
+	break;
+      case(Item_func::COND_OR_FUNC):
+	DBUG_PRINT("info", ("COND_OR_FUNC"));
+	curr_cond->ndb_item= new Ndb_item(cond_item->functype());      
+	break;
+      default:
+	DBUG_PRINT("info", ("COND_ITEM %d", cond_item->functype()));
+	*context->supported_ptr= FALSE;
+	break;
+      }
+      break;
+    }
+    default: {
+      DBUG_PRINT("info", ("Found item of type %d", item->type()));
+      *context->supported_ptr= FALSE;
+    }
+    }
+  }
+
+  DBUG_VOID_RETURN;
+}
+
+bool
+ha_ndbcluster::serialize_cond(const COND *cond, Ndb_cond_stack *ndb_cond)
+{
+  DBUG_ENTER("serialize_cond");
+  Item *item= (Item *) cond;
+  bool supported= TRUE;
+  Ndb_cond_traverse_context context(table, (void *)m_table, 
+				    &supported, ndb_cond);
+  item->traverse_cond(&ndb_serialize_cond, (void *) &context, Item::PREFIX);
+  DBUG_PRINT("info", ("The pushed condition is %ssupported", (supported)?"":"not "));
+
+  DBUG_RETURN(supported);
+}
+
+Ndb_cond *
+ha_ndbcluster::build_scan_filter_predicate(Ndb_cond *cond, 
+					   NdbScanFilter *filter)
+{
+  DBUG_ENTER("build_scan_filter_predicate");  
+  switch(cond->ndb_item->type) {
+  case(NDB_FUNCTION): {
+    if (!cond->next)
+      break;
+    Ndb_item *a= cond->next->ndb_item;
+    switch(cond->ndb_item->qualification.function_type) {
+    case(Item_func::EQ_FUNC): {
+      if (!cond->next->next)
+	break;
+      Ndb_item *b= cond->next->next->ndb_item;
+      Ndb_item *value= 
+	(a->type == NDB_VALUE)? a
+	: (b->type == NDB_VALUE)? b
+	: NULL;
+      Ndb_item *field= 
+	(a->type == NDB_FIELD)? a
+	: (b->type == NDB_FIELD)? b
+	: NULL;
+      if (!value || !field) break;
+      DBUG_PRINT("info", ("Generating EQ filter"));
+      filter->eq(field->getFieldNo(),
+		 (Uint32) value->getIntValue());
+      DBUG_RETURN(cond->next->next->next);
+    }
+    case(Item_func::NE_FUNC): {
+      if (!cond->next->next)
+	break;
+      Ndb_item *b= cond->next->next->ndb_item;
+      Ndb_item *value= 
+	(a->type == NDB_VALUE)? a
+	: (b->type == NDB_VALUE)? b
+	: NULL;
+      Ndb_item *field= 
+	(a->type == NDB_FIELD)? a
+	: (b->type == NDB_FIELD)? b
+	: NULL;
+      if (!value || !field) break;
+      DBUG_PRINT("info", ("Generating NE filter"));
+      filter->ne(field->getFieldNo(),
+		 (Uint32) value->getIntValue());
+      DBUG_RETURN(cond->next->next->next);
+    }
+    case(Item_func::LT_FUNC): {
+      if (!cond->next->next)
+	break;
+      Ndb_item *b= cond->next->next->ndb_item;
+      Ndb_item *value= 
+	(a->type == NDB_VALUE)? a
+	: (b->type == NDB_VALUE)? b
+	: NULL;
+      Ndb_item *field= 
+	(a->type == NDB_FIELD)? a
+	: (b->type == NDB_FIELD)? b
+	: NULL;
+      if (!value || !field) break;
+      DBUG_PRINT("info", ("Generating LT filter"));
+      if (a == field)
+	filter->lt(field->getFieldNo(),
+		   (Uint32) value->getIntValue());
+      else
+	filter->gt(field->getFieldNo(),
+		   (Uint32) value->getIntValue());
+      DBUG_RETURN(cond->next->next->next);
+    }
+    case(Item_func::LE_FUNC): {
+      if (!cond->next->next)
+	break;
+      Ndb_item *b= cond->next->next->ndb_item;
+      Ndb_item *value= 
+	(a->type == NDB_VALUE)? a
+	: (b->type == NDB_VALUE)? b
+	: NULL;
+      Ndb_item *field= 
+	(a->type == NDB_FIELD)? a
+	: (b->type == NDB_FIELD)? b
+	: NULL;
+      if (!value || !field) break;
+      DBUG_PRINT("info", ("Generating LE filter"));
+      if (a == field)
+	filter->le(field->getFieldNo(),
+		   (Uint32) value->getIntValue());
+      else
+	filter->ge(field->getFieldNo(),
+		   (Uint32) value->getIntValue());
+      DBUG_RETURN(cond->next->next->next);
+    }
+    case(Item_func::GE_FUNC): {
+      if (!cond->next->next)
+	break;
+      Ndb_item *b= cond->next->next->ndb_item;
+      Ndb_item *value= 
+	(a->type == NDB_VALUE)? a
+	: (b->type == NDB_VALUE)? b
+	: NULL;
+      Ndb_item *field= 
+	(a->type == NDB_FIELD)? a
+	: (b->type == NDB_FIELD)? b
+	: NULL;
+      if (!value || !field) break;
+      DBUG_PRINT("info", ("Generating GE filter"));
+      if (a == field)
+	filter->ge(field->getFieldNo(),
+		   (Uint32) value->getIntValue());
+      else
+	filter->le(field->getFieldNo(),
+		   (Uint32) value->getIntValue());
+      DBUG_RETURN(cond->next->next->next);
+    }
+    case(Item_func::GT_FUNC): {
+      if (!cond->next->next)
+	break;
+      Ndb_item *b= cond->next->next->ndb_item;
+      Ndb_item *value= 
+	(a->type == NDB_VALUE)? a
+	: (b->type == NDB_VALUE)? b
+	: NULL;
+      Ndb_item *field= 
+	(a->type == NDB_FIELD)? a
+	: (b->type == NDB_FIELD)? b
+	: NULL;
+      if (!value || !field) break;
+      DBUG_PRINT("info", ("Generating GT filter"));
+      if (a == field)
+	filter->gt(field->getFieldNo(),
+		   (Uint32) value->getIntValue());
+      else
+	filter->lt(field->getFieldNo(),
+		   (Uint32) value->getIntValue());
+      DBUG_RETURN(cond->next->next->next);
+    }
+    case(Item_func::LIKE_FUNC): {
+      if (!cond->next->next)
+	break;
+      Ndb_item *b= cond->next->next->ndb_item;
+      Ndb_item *value= 
+	(a->type == NDB_VALUE)? a
+	: (b->type == NDB_VALUE)? b
+	: NULL;
+      Ndb_item *field= 
+	(a->type == NDB_FIELD)? a
+	: (b->type == NDB_FIELD)? b
+	: NULL;
+      if (!value || !field) break;
+      if (value->qualification.value_type != Item::STRING_ITEM) break;
+      String *str= value->getStringValue();
+      DBUG_PRINT("info", ("Generating LIKE filter: like(%d,%s,%d)", field->getFieldNo(), str->ptr(), str->length()));
+      filter->like(field->getFieldNo(),
+		   str->ptr(), str->length(), TRUE);
+      DBUG_RETURN(cond->next->next->next);
+    }
+    case(Item_func::NOTLIKE_FUNC): {
+      if (!cond->next->next)
+	break;
+      Ndb_item *b= cond->next->next->ndb_item;
+      Ndb_item *value= 
+	(a->type == NDB_VALUE)? a
+	: (b->type == NDB_VALUE)? b
+	: NULL;
+      Ndb_item *field= 
+	(a->type == NDB_FIELD)? a
+	: (b->type == NDB_FIELD)? b
+	: NULL;
+      if (!value || !field) break;
+      if (value->qualification.value_type != Item::STRING_ITEM) break;
+      String *str= value->getStringValue();
+      DBUG_PRINT("info", ("Generating NOTLIKE filter: notlike(%d,%s,%d)", field->getFieldNo(), str->ptr(), str->length()));
+      filter->notlike(field->getFieldNo(),
+		      str->ptr(), str->length());
+      DBUG_RETURN(cond->next->next->next);
+    }
+    case(Item_func::ISNULL_FUNC):
+      if (a->type == NDB_FIELD) {
+	DBUG_PRINT("info", ("Generating ISNULL filter"));
+	filter->isnull(a->getFieldNo());
+      }
+      DBUG_RETURN(cond->next->next);
+    case(Item_func::ISNOTNULL_FUNC): {
+      if (a->type == NDB_FIELD) {
+	DBUG_PRINT("info", ("Generating ISNOTNULL filter"));
+	filter->isnotnull(a->getFieldNo());
+      }
+      DBUG_RETURN(cond->next->next);
+    }
+    default:
+      break;
+    }
+    break;
+  }
+  default:
+    break;
+  }
+  DBUG_PRINT("info", ("Found illegal condition"));
+  DBUG_RETURN(NULL);
+}
+
+Ndb_cond *
+ha_ndbcluster::build_scan_filter_group(Ndb_cond *cond, NdbScanFilter *filter)
+{
+  DBUG_ENTER("build_scan_filter_group");
+  switch(cond->ndb_item->type) {
+  case(NDB_FUNCTION):
+    switch(cond->ndb_item->qualification.function_type) {
+    case(Item_func::COND_AND_FUNC): {
+      DBUG_PRINT("info", ("Generating AND group"));
+      filter->begin(NdbScanFilter::AND);
+      cond= cond->next;
+      cond= build_scan_filter_group(cond, filter);
+      cond= build_scan_filter_group(cond, filter);
+      filter->end();
+      break;
+    }
+    case(Item_func::COND_OR_FUNC): {
+      DBUG_PRINT("info", ("Generating OR group"));
+      filter->begin(NdbScanFilter::OR);
+      cond= cond->next;
+      cond= build_scan_filter_group(cond, filter);
+      cond= build_scan_filter_group(cond, filter);
+      filter->end();
+      break;
+    }
+    default:
+      cond= build_scan_filter_predicate(cond, filter);     
+    }
+    break;
+  default: {
+    DBUG_PRINT("info", ("Illegal scan filter"));
+  }
+  }
+  
+  DBUG_RETURN(cond);
+}
+
+void 
+ha_ndbcluster::build_scan_filter(Ndb_cond *cond, NdbScanFilter *filter)
+{
+  bool simple_cond= TRUE;
+  DBUG_ENTER("build_scan_filter");  
+
+  switch(cond->ndb_item->type) {
+  case(Item_func::COND_AND_FUNC):
+    simple_cond= FALSE;
+    break;
+  case(Item_func::COND_OR_FUNC):
+    simple_cond= FALSE;
+    break;
+  default:
+    break;
+  }
+  if (simple_cond) filter->begin();
+  build_scan_filter_group(cond, filter);
+  if (simple_cond) filter->end();
+
+  DBUG_VOID_RETURN;
+}
+
+void
+ha_ndbcluster::generate_scan_filter(Ndb_cond_stack *ndb_cond_stack,
+				    NdbScanOperation *op)
+{
+  DBUG_ENTER("generate_scan_filter");
+  if (ndb_cond_stack)
+  {
+    NdbScanFilter filter(op);
+    bool multiple_cond= FALSE;
+    // Wrap an AND group around multiple conditions
+    if (ndb_cond_stack->next) {
+      multiple_cond= TRUE;
+      filter.begin();
+    }
+    for (Ndb_cond_stack *stack= ndb_cond_stack; 
+	 (stack); 
+	 stack= stack->next)
+      {
+	build_scan_filter(stack->ndb_cond, &filter);
+      }
+    if (multiple_cond) filter.end();
+  }
+  else
+  {  
+    DBUG_PRINT("info", ("Empty stack"));
+  }
+
+  DBUG_VOID_RETURN;
+}
+
 #endif /* HAVE_NDBCLUSTER_DB */
diff --git a/sql/ha_ndbcluster.h b/sql/ha_ndbcluster.h
index 3aeeb28d8bd..e3c356869bc 100644
--- a/sql/ha_ndbcluster.h
+++ b/sql/ha_ndbcluster.h
@@ -32,6 +32,7 @@ class NdbOperation;    // Forward declaration
 class NdbConnection;   // Forward declaration
 class NdbRecAttr;      // Forward declaration
 class NdbScanOperation; 
+class NdbScanFilter; 
 class NdbIndexScanOperation; 
 class NdbBlob;
 
@@ -60,6 +61,145 @@ typedef struct st_ndbcluster_share {
   uint table_name_length,use_count;
 } NDB_SHARE;
 
+typedef enum ndb_item_type {
+  NDB_VALUE = 0,   // Qualified more with Item::Type
+  NDB_FIELD = 1,   // Qualified from table definition
+  NDB_FUNCTION = 2 // Qualified from Item_func::Functype
+} NDB_ITEM_TYPE;
+
+typedef union ndb_item_qualification {
+  Item::Type value_type; 
+  enum_field_types field_type;       // Instead of Item::FIELD_ITEM
+  Item_func::Functype function_type; // Instead of Item::FUNC_ITEM
+} NDB_ITEM_QUALIFICATION;
+
+class Ndb_item_string_value {
+ public:
+  String s;
+  CHARSET_INFO *c;
+};
+
+typedef struct ndb_item_field_value {
+  Field* field;
+  int column_no;
+} NDB_ITEM_FIELD_VALUE;
+
+typedef union ndb_item_value {
+  longlong int_value;
+  double real_value;
+  Ndb_item_string_value *string_value;
+  NDB_ITEM_FIELD_VALUE *field_value;
+} NDB_ITEM_VALUE;
+
+class Ndb_item {
+ public:
+  Ndb_item(NDB_ITEM_TYPE item_type, 
+	   NDB_ITEM_QUALIFICATION item_qualification,
+	   const Item *item_value);
+  Ndb_item(longlong int_value);
+  Ndb_item(double real_value);
+  Ndb_item();
+  Ndb_item(Field *field, int column_no);
+  Ndb_item(Item_func::Functype func_type);
+  ~Ndb_item();
+  void print(String *str);
+  // Getters and Setters
+  longlong getIntValue() { return value.int_value; };
+  double getRealValue() { return value.real_value; };
+  String * getStringValue() { return &value.string_value->s; };
+  CHARSET_INFO * getStringCharset() { return value.string_value->c; };
+  Field * getField() { return value.field_value->field; };
+  int getFieldNo() { return value.field_value->column_no; };
+
+ public:
+  NDB_ITEM_TYPE type;
+  NDB_ITEM_QUALIFICATION qualification;
+
+
+ private:
+  NDB_ITEM_VALUE value;
+
+};
+
+class Ndb_cond {
+ public:
+  Ndb_cond() : ndb_item(NULL), next(NULL), prev(NULL) {};
+  ~Ndb_cond() 
+  { 
+    if (ndb_item) delete ndb_item; 
+    ndb_item= NULL; 
+    if (next) delete next;
+    next= prev= NULL; 
+  };
+  Ndb_item *ndb_item;
+  Ndb_cond *next;
+  Ndb_cond *prev;
+};
+
+class Ndb_cond_stack {
+ public:
+  Ndb_cond_stack() : ndb_cond(NULL), next(NULL) {};
+  ~Ndb_cond_stack() 
+  { 
+    if (ndb_cond) delete ndb_cond; 
+    ndb_cond= NULL; 
+    next= NULL; 
+  };
+  Ndb_cond *ndb_cond;
+  Ndb_cond_stack *next;
+};
+
+class Ndb_cond_traverse_context {
+ public:
+  Ndb_cond_traverse_context(TABLE *tab, void* ndb_tab, 
+			    bool *supported, Ndb_cond_stack* stack)
+    : table(tab), ndb_table(ndb_tab), 
+    supported_ptr(supported), stack_ptr(stack), cond_ptr(NULL),
+    expect_mask(0), expect_field_result_mask(0)
+  {
+    if (stack)
+      cond_ptr= stack->ndb_cond;
+  };
+  void expect(Item::Type type)
+  {
+    expect_mask|= (1 << type);
+  };
+  void dont_expect(Item::Type type)
+  {
+    expect_mask&= ~(1 << type);
+  };
+  bool expecting(Item::Type type)
+  {
+    return (expect_mask & (1 << type));
+  };
+  void expect_nothing()
+  {
+    expect_mask= 0;
+  };
+
+  void expect_field_result(Item_result result)
+  {
+    expect_field_result_mask|= (1 << result);
+  };
+  bool expecting_field_result(Item_result result)
+  {
+    return (expect_field_result_mask & (1 << result));
+  };
+  void expect_no_field_result()
+  {
+    expect_field_result_mask= 0;
+  };
+
+  TABLE* table;
+  void* ndb_table;
+  bool *supported_ptr;
+  Ndb_cond_stack* stack_ptr;
+  Ndb_cond* cond_ptr;
+  private:
+  uint expect_mask;
+  uint expect_field_result_mask;
+};
+
 /*
   Place holder for ha_ndbcluster thread specific data
 */
@@ -122,7 +262,6 @@ class ha_ndbcluster: public handler
   void info(uint);
   int extra(enum ha_extra_function operation);
   int extra_opt(enum ha_extra_function operation, ulong cache_size);
-  int reset();
   int external_lock(THD *thd, int lock_type);
   int start_stmt(THD *thd);
   const char * table_type() const;
@@ -152,6 +291,13 @@ class ha_ndbcluster: public handler
 
   static Thd_ndb* seize_thd_ndb();
   static void release_thd_ndb(Thd_ndb* thd_ndb);
+ 
+  /*
+    Condition pushdown
+  */
+  const COND *cond_push(const COND *cond);
+  void cond_pop();
+
   uint8 table_cache_type();
     
  private:
@@ -214,9 +360,32 @@ class ha_ndbcluster: public handler
 
   int write_ndb_file();
 
- private:
   int check_ndb_connection();
 
+  void set_rec_per_key();
+  void records_update();
+  void no_uncommitted_rows_execute_failure();
+  void no_uncommitted_rows_update(int);
+  void no_uncommitted_rows_init(THD *);
+  void no_uncommitted_rows_reset(THD *);
+
+  /*
+    Condition Pushdown to Handler (CPDH), private methods
+  */
+  void cond_clear();
+  bool serialize_cond(const COND *cond, Ndb_cond_stack *ndb_cond);
+  Ndb_cond * build_scan_filter_predicate(Ndb_cond* cond, 
+					 NdbScanFilter* filter);
+  Ndb_cond * build_scan_filter_group(Ndb_cond* cond, 
+				     NdbScanFilter* filter);
+  void build_scan_filter(Ndb_cond* cond, NdbScanFilter* filter);
+  void generate_scan_filter(Ndb_cond_stack* cond_stack, 
+			    NdbScanOperation* op);
+
+  friend int execute_commit(ha_ndbcluster*, NdbConnection*);
+  friend int execute_no_commit(ha_ndbcluster*, NdbConnection*);
+  friend int execute_no_commit_ie(ha_ndbcluster*, NdbConnection*);
+
   NdbConnection *m_active_trans;
   NdbScanOperation *m_active_cursor;
   Ndb *m_ndb;
@@ -254,7 +423,7 @@ class ha_ndbcluster: public handler
   ha_rows m_autoincrement_prefetch;
   bool m_transaction_on;
   bool m_use_local_query_cache;
-
+  Ndb_cond_stack *m_cond_stack;
   bool m_disable_multi_read;
   byte *m_multi_range_result_ptr;
   uint m_multi_range_defined_count;
@@ -263,16 +432,6 @@ class ha_ndbcluster: public handler
   byte *m_multi_range_cursor_result_ptr;
   int setup_recattr(const NdbRecAttr*);
   
-  void set_rec_per_key();
-  void records_update();
-  void no_uncommitted_rows_execute_failure();
-  void no_uncommitted_rows_update(int);
-  void no_uncommitted_rows_init(THD *);
-  void no_uncommitted_rows_reset(THD *);
-
-  friend int execute_no_commit(ha_ndbcluster*, NdbConnection*);
-  friend int execute_commit(ha_ndbcluster*, NdbConnection*);
-  friend int execute_no_commit_ie(ha_ndbcluster*, NdbConnection*);
 };
 
 bool ndbcluster_init(void);
diff --git a/sql/handler.h b/sql/handler.h
index c9adaefa888..3c7bda70de2 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -264,6 +264,9 @@ typedef struct st_table TABLE;
 struct st_foreign_key_info;
 typedef struct st_foreign_key_info FOREIGN_KEY_INFO;
 
+/* Forward declaration for Condition Pushdown to Handler (CPDH) */
+typedef struct Item COND;
+
 typedef struct st_ha_check_opt
 {
   ulong sort_buffer_size;
@@ -563,7 +566,7 @@ public:
 
   /* Type of table for caching query */
   virtual uint8 table_cache_type() { return HA_CACHE_TBL_NONTRANSACT; }
- 
+
  /*
   RETURN
     true  Primary key (if there is one) is clustered key covering all fields
@@ -575,6 +578,12 @@ public:
  {
    return memcmp(ref1, ref2, ref_length);
  }
+ 
+ /*
+   Condition pushdown to storage engines
+ */
+ virtual const COND *cond_push(const COND *cond) { return cond; };
+ virtual void cond_pop() { return; };
 };
 
 	/* Some extern variables used with handlers */
diff --git a/sql/item.h b/sql/item.h
index cf3dc8896a5..06614faecc6 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -113,6 +113,8 @@ public:
 typedef bool (Item::*Item_processor)(byte *arg);
 typedef Item* (Item::*Item_transformer) (byte *arg);
 
+typedef void (*Item_cond_traverser) (const Item *item, void *arg);
+
 class Item {
   Item(const Item &);			/* Prevent use of these */
   void operator=(Item &);
@@ -124,7 +126,7 @@ public:
   static void operator delete(void *ptr,size_t size, MEM_ROOT *mem_root)
   { TRASH(ptr, size); }
 
-  enum Type {FIELD_ITEM, FUNC_ITEM, SUM_FUNC_ITEM, STRING_ITEM,
+  enum Type {FIELD_ITEM= 0, FUNC_ITEM, SUM_FUNC_ITEM, STRING_ITEM,
 	     INT_ITEM, REAL_ITEM, NULL_ITEM, VARBIN_ITEM,
 	     COPY_STR_ITEM, FIELD_AVG_ITEM, DEFAULT_VALUE_ITEM,
 	     PROC_ITEM,COND_ITEM, REF_ITEM, FIELD_STD_ITEM,
@@ -133,6 +135,8 @@ public:
              PARAM_ITEM, TRIGGER_FIELD_ITEM};
 
   enum cond_result { COND_UNDEF,COND_OK,COND_TRUE,COND_FALSE };
+
+  enum traverse_order { POSTFIX, PREFIX };
   
   /*
     str_values's main purpose is to be used to cache the value in
@@ -314,6 +318,13 @@ public:
   {
     return (this->*transformer)(arg);
   }
+
+   virtual void traverse_cond(Item_cond_traverser traverser, 
+			      void *arg,
+			      traverse_order order = POSTFIX)
+   {
+     (*traverser)(this, arg);
+   }
  
   virtual bool remove_dependence_processor(byte * arg) { return 0; }
   virtual bool remove_fixed(byte * arg) { fixed= 0; return 0; }
diff --git a/sql/item_cmpfunc.cc b/sql/item_cmpfunc.cc
index 90eee9e76d1..597ed7a39d5 100644
--- a/sql/item_cmpfunc.cc
+++ b/sql/item_cmpfunc.cc
@@ -2037,6 +2037,20 @@ Item *Item_cond::transform(Item_transformer transformer, byte *arg)
   return Item_func::transform(transformer, arg);
 }
 
+void Item_cond::traverse_cond(Item_cond_traverser traverser, 
+			      void *arg,
+			      traverse_order order)
+{
+  List_iterator<Item> li(list);
+  Item *item;
+
+  if (order == PREFIX) (*traverser)(this, arg);
+  while ((item= li++))
+  {
+    item->traverse_cond(traverser, arg, order);
+  }
+  if (order == POSTFIX) (*traverser)(this, arg);
+}
 
 void Item_cond::split_sum_func(THD *thd, Item **ref_pointer_array,
                                List<Item> &fields)
diff --git a/sql/item_cmpfunc.h b/sql/item_cmpfunc.h
index 6a7e037bed1..d9196eeb5eb 100644
--- a/sql/item_cmpfunc.h
+++ b/sql/item_cmpfunc.h
@@ -965,6 +965,9 @@ public:
   void copy_andor_arguments(THD *thd, Item_cond *item);
   bool walk(Item_processor processor, byte *arg);
   Item *transform(Item_transformer transformer, byte *arg);
+  void traverse_cond(Item_cond_traverser, 
+		     void *arg,
+		     traverse_order order = POSTFIX);
   void neg_arguments(THD *thd);
 };
 
diff --git a/sql/item_func.cc b/sql/item_func.cc
index aba53b9b397..9e9d4193f84 100644
--- a/sql/item_func.cc
+++ b/sql/item_func.cc
@@ -342,6 +342,22 @@ bool Item_func::walk (Item_processor processor, byte *argument)
   return (this->*processor)(argument);
 }
 
+void Item_func::traverse_cond(Item_cond_traverser traverser, 
+			      void *argument,
+			      traverse_order order)
+{
+  if (arg_count)
+  {
+    Item **arg,**arg_end;
+    if (order == PREFIX) (traverser)(this, argument);
+    for (arg= args, arg_end= args+arg_count; arg != arg_end; arg++)
+    {
+      (*arg)->traverse_cond(traverser, argument, order);
+    }
+  }
+  if (order == POSTFIX) (traverser)(this, argument);
+}
+
 
 /*
   Transform an Item_func object with a transformer callback function
diff --git a/sql/item_func.h b/sql/item_func.h
index 4657ee81dfb..7c7c397bce0 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -154,6 +154,9 @@ public:
                         uint flags= 0);
   bool walk(Item_processor processor, byte *arg);
   Item *transform(Item_transformer transformer, byte *arg);
+  void traverse_cond(Item_cond_traverser traverser, 
+		     void * arg,
+		     traverse_order order = POSTFIX);
 };
 
 
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 42ae6982eb0..d250dd48a6c 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -4093,6 +4093,7 @@ enum options_mysqld
   OPT_INNODB, OPT_ISAM,
   OPT_NDBCLUSTER, OPT_NDB_CONNECTSTRING, OPT_NDB_USE_EXACT_COUNT,
   OPT_NDB_FORCE_SEND, OPT_NDB_AUTOINCREMENT_PREFETCH_SZ,
+  OPT_NDB_CONDITION_PUSHDOWN,
   OPT_SKIP_SAFEMALLOC,
   OPT_TEMP_POOL, OPT_TX_ISOLATION,
   OPT_SKIP_STACK_TRACE, OPT_SKIP_SYMLINKS,
@@ -4547,6 +4548,12 @@ Disable with --skip-ndbcluster (will save memory).",
    (gptr*) &opt_ndbcluster, (gptr*) &opt_ndbcluster, 0, GET_BOOL, NO_ARG,
    OPT_NDBCLUSTER_DEFAULT, 0, 0, 0, 0, 0},
 #ifdef HAVE_NDBCLUSTER_DB
+  {"ndb-condition-pushdown", 
+   OPT_NDB_CONDITION_PUSHDOWN,
+   "Push supported query conditions to the ndbcluster storage engine.",
+   (gptr*) &global_system_variables.ndb_condition_pushdown,
+   (gptr*) &global_system_variables.ndb_condition_pushdown,
+   0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
   {"ndb-connectstring", OPT_NDB_CONNECTSTRING,
    "Connect string for ndbcluster.",
    (gptr*) &ndbcluster_connectstring, (gptr*) &ndbcluster_connectstring,
diff --git a/sql/set_var.cc b/sql/set_var.cc
index 234ec6617c3..1eb7d54845a 100644
--- a/sql/set_var.cc
+++ b/sql/set_var.cc
@@ -388,6 +388,9 @@ sys_ndb_use_exact_count("ndb_use_exact_count",
 sys_var_thd_bool
 sys_ndb_use_transactions("ndb_use_transactions",
 			 &SV::ndb_use_transactions);
+sys_var_thd_bool
+sys_ndb_condition_pushdown("ndb_condition_pushdown",
+			 &SV::ndb_condition_pushdown);
 // ndb server global variable settings
 // none
 #endif
@@ -654,6 +657,7 @@ sys_var *sys_variables[]=
   &sys_ndb_force_send,
   &sys_ndb_use_exact_count,
   &sys_ndb_use_transactions,
+  &sys_ndb_condition_pushdown,
 #endif
   &sys_unique_checks,
   &sys_updatable_views_with_limit,
@@ -824,6 +828,8 @@ struct show_var_st init_vars[]= {
   {sys_ndb_force_send.name,   (char*) &sys_ndb_force_send,          SHOW_SYS},
   {sys_ndb_use_exact_count.name,(char*) &sys_ndb_use_exact_count,   SHOW_SYS},
   {sys_ndb_use_transactions.name,(char*) &sys_ndb_use_transactions, SHOW_SYS},
+  {sys_ndb_condition_pushdown.name, (char*) &sys_ndb_condition_pushdown,      
+   SHOW_SYS},
 #endif
   {sys_net_buffer_length.name,(char*) &sys_net_buffer_length,       SHOW_SYS},
   {sys_net_read_timeout.name, (char*) &sys_net_read_timeout,        SHOW_SYS},
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 6cec2c2c787..2e9440378d5 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -436,6 +436,7 @@ struct system_variables
   my_bool ndb_force_send;
   my_bool ndb_use_exact_count;
   my_bool ndb_use_transactions;
+  my_bool ndb_condition_pushdown;
 #endif /* HAVE_NDBCLUSTER_DB */
   my_bool old_passwords;
   
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index ed3606856a0..4b8e1578f37 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -5269,6 +5269,7 @@ make_join_select(JOIN *join,SQL_SELECT *select,COND *cond)
           if (!(tmp= add_found_match_trig_cond(first_inner_tab, tmp, 0)))
             DBUG_RETURN(1);
           tab->select_cond=sel->cond=tmp;
+	  tab->table->file->cond_push(tmp); // Push condition to handler
         }
         else
           tab->select_cond= sel->cond= NULL;
@@ -5390,6 +5391,8 @@ make_join_select(JOIN *join,SQL_SELECT *select,COND *cond)
 		join->thd->memdup((gptr) sel, sizeof(SQL_SELECT));
 	      tab->cache.select->cond=tmp;
 	      tab->cache.select->read_tables=join->const_table_map;
+	      if (tmp != tab->select_cond)
+		tab->table->file->cond_push(tmp); // Push condition to handler
 	    }
 	  }
 	}