Moved all code related to engine_condition_pushdown to a new class,

ha_ndbcluster_cond.
Added new files:
sql/ha_ndbcluster_cond.h
sql/ha_ndbcluster_cond.cc


sql/ha_ndbcluster_cond.cc:
  BitKeeper file /windows/Linux_space/MySQL/mysql-5.0-ndb/sql/ha_ndbcluster_cond.cc
sql/ha_ndbcluster_cond.h:
  BitKeeper file /windows/Linux_space/MySQL/mysql-5.0-ndb/sql/ha_ndbcluster_cond.h
sql/Makefile.am:
  Added compilation of new separate files for engine_condition_pushdown
sql/ha_ndbcluster.cc:
  Moved all code related to engine_condition_pushdown to ha_ndbcluster_cond
sql/ha_ndbcluster.h:
  Moved all code related to engine_condition_pushdown to ha_ndbcluster_cond

5 files changed, 1976 insertions, 1857 deletions

diff --git a/sql/Makefile.am b/sql/Makefile.am
index 4f84023724f..9be3bf93746 100644
--- a/sql/Makefile.am
+++ b/sql/Makefile.am
@@ -54,7 +54,8 @@ noinst_HEADERS =	item.h item_func.h item_sum.h item_cmpfunc.h \
 			sql_error.h field.h handler.h mysqld_suffix.h \
 			ha_myisammrg.h\
 			ha_heap.h ha_myisam.h ha_berkeley.h ha_innodb.h \
-			ha_ndbcluster.h opt_range.h protocol.h \
+			ha_ndbcluster.h ha_ndbcluster_cond.h \
+			opt_range.h protocol.h \
 			sql_select.h structs.h table.h sql_udf.h hash_filo.h\
 			lex.h lex_symbol.h sql_acl.h sql_crypt.h  \
 			log_event.h sql_repl.h slave.h \
@@ -88,7 +89,7 @@ mysqld_SOURCES =	sql_lex.cc sql_handler.cc \
 		   	records.cc filesort.cc handler.cc \
 		        ha_heap.cc ha_myisam.cc ha_myisammrg.cc \
 	                ha_berkeley.cc ha_innodb.cc \
-			ha_ndbcluster.cc \
+			ha_ndbcluster.cc ha_ndbcluster_cond.cc \
 			sql_db.cc sql_table.cc sql_rename.cc sql_crypt.cc \
 			sql_load.cc mf_iocache.cc field_conv.cc sql_show.cc \
 			sql_udf.cc sql_analyse.cc sql_analyse.h sql_cache.cc \
diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
index e9ba854300d..dd7aa45f303 100644
--- a/sql/ha_ndbcluster.cc
+++ b/sql/ha_ndbcluster.cc
@@ -29,7 +29,7 @@
 #include <my_dir.h>
 #include "ha_ndbcluster.h"
 #include <ndbapi/NdbApi.hpp>
-#include <ndbapi/NdbScanFilter.hpp>
+#include "ha_ndbcluster_cond.h"
 
 // options from from mysqld.cc
 extern my_bool opt_ndb_optimized_node_selection;
@@ -2114,7 +2114,7 @@ int ha_ndbcluster::ordered_index_scan(const key_range *start_key,
       DBUG_RETURN(res);
   }
 
-  if (!restart && generate_scan_filter(m_cond_stack, op))
+  if (!restart && m_cond && m_cond->generate_scan_filter(op))
     DBUG_RETURN(ndb_err(trans));
   
   if (!restart && (res= define_read_attrs(buf, op)))
@@ -2153,7 +2153,14 @@ int ha_ndbcluster::unique_index_scan(const KEY* key_info,
 		     parallelism))
     ERR_RETURN(trans->getNdbError());
   m_active_cursor= op;
-  if (generate_scan_filter_from_key(op, key_info, key, key_len, buf))
+  if (!m_cond)
+    m_cond= new ha_ndbcluster_cond;
+  if (!m_cond)
+  {
+    my_errno= HA_ERR_OUT_OF_MEM;
+    DBUG_RETURN(my_errno);
+  }       
+  if (m_cond->generate_scan_filter_from_key(op, key_info, key, key_len, buf))
     DBUG_RETURN(ndb_err(trans));
   if ((res= define_read_attrs(buf, op)))
     DBUG_RETURN(res);
@@ -2186,7 +2193,7 @@ int ha_ndbcluster::full_table_scan(byte *buf)
 		     parallelism))
     ERR_RETURN(trans->getNdbError());
   m_active_cursor= op;
-  if (generate_scan_filter(m_cond_stack, op))
+  if (m_cond && m_cond->generate_scan_filter(op))
     DBUG_RETURN(ndb_err(trans));
   if ((res= define_read_attrs(buf, op)))
     DBUG_RETURN(res);
@@ -3449,7 +3456,10 @@ int ha_ndbcluster::extra(enum ha_extra_function operation)
 int ha_ndbcluster::reset()
 {
   DBUG_ENTER("ha_ndbcluster::reset");
-  cond_clear();
+  if (m_cond)
+  {
+    m_cond->cond_clear();
+  }
 
   /* reset flags set by extra calls */
   m_retrieve_all_fields= FALSE;
@@ -4873,7 +4883,7 @@ ha_ndbcluster::ha_ndbcluster(TABLE *table_arg):
   m_force_send(TRUE),
   m_autoincrement_prefetch((ha_rows) 32),
   m_transaction_on(TRUE),
-  m_cond_stack(NULL),
+  m_cond(NULL),
   m_multi_cursor(NULL)
 {
   int i;
@@ -4920,9 +4930,13 @@ ha_ndbcluster::~ha_ndbcluster()
   }
   DBUG_ASSERT(m_active_trans == NULL);
 
-  // Discard the condition stack
-  DBUG_PRINT("info", ("Clearing condition stack"));
-  cond_clear();
+  // Discard any generated condition
+  DBUG_PRINT("info", ("Deleting generated condition"));
+  if (m_cond)
+  {
+    delete m_cond;
+    m_cond= NULL;
+  }
 
   DBUG_VOID_RETURN;
 }
@@ -6510,7 +6524,7 @@ ha_ndbcluster::read_multi_range_first(KEY_MULTI_RANGE **found_range_p,
         else if ((scanOp= m_active_trans->getNdbIndexScanOperation(idx, tab)) 
                  &&!scanOp->readTuples(lm, 0, parallelism, sorted, 
 				       FALSE, TRUE, need_pk, TRUE)
-                 &&!generate_scan_filter(m_cond_stack, scanOp)
+                 &&!(m_cond && m_cond->generate_scan_filter(scanOp))
                  &&!define_read_attrs(end_of_buffer-reclength, scanOp))
         {
           m_multi_cursor= scanOp;
@@ -6969,1412 +6983,6 @@ next:
   DBUG_RETURN(NULL);
 }
 
-/*
-  Condition pushdown
-*/
-/*
-  Push a condition to ndbcluster storage engine for evaluation 
-  during table   and index scans. The conditions will be stored on a stack
-  for possibly storing several conditions. The stack can be popped
-  by calling cond_pop, handler::extra(HA_EXTRA_RESET) (handler::reset())
-  will clear the stack.
-  The current implementation supports arbitrary AND/OR nested conditions
-  with comparisons between columns and constants (including constant
-  expressions and function calls) and the following comparison operators:
-  =, !=, >, >=, <, <=, "is null", and "is not null".
-  
-  RETURN
-    NULL The condition was supported and will be evaluated for each 
-    row found during the scan
-    cond The condition was not supported and all rows will be returned from
-         the scan for evaluation (and thus not saved on stack)
-*/
-const 
-COND* 
-ha_ndbcluster::cond_push(const COND *cond) 
-{ 
-  DBUG_ENTER("cond_push");
-  Ndb_cond_stack *ndb_cond = new Ndb_cond_stack();
-  if (ndb_cond == NULL)
-  {
-    my_errno= HA_ERR_OUT_OF_MEM;
-    DBUG_RETURN(NULL);
-  }
-  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); 
-}
-
-/*
-  Pop the top condition from the condition stack of the handler instance.
-*/
-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;
-  }
-}
-
-/*
-  Clear the condition stack
-*/
-void
-ha_ndbcluster::cond_clear()
-{
-  DBUG_ENTER("cond_clear");
-  while (m_cond_stack)
-    cond_pop();
-
-  DBUG_VOID_RETURN;
-}
-
-/*
-  Serialize the item tree into a linked list represented by Ndb_cond
-  for fast generation of NbdScanFilter. Adds information such as
-  position of fields that is not directly available in the Item tree.
-  Also checks if condition is supported.
-*/
-void ndb_serialize_cond(const Item *item, void *arg)
-{
-  Ndb_cond_traverse_context *context= (Ndb_cond_traverse_context *) arg;
-  DBUG_ENTER("ndb_serialize_cond");  
-
-  // Check if we are skipping arguments to a function to be evaluated
-  if (context->skip)
-  {
-    DBUG_PRINT("info", ("Skiping argument %d", context->skip));
-    context->skip--;
-    switch (item->type()) {
-    case Item::FUNC_ITEM:
-    {
-      Item_func *func_item= (Item_func *) item;
-      context->skip+= func_item->argument_count();
-      break;
-    }
-    case Item::INT_ITEM:
-    case Item::REAL_ITEM:
-    case Item::STRING_ITEM:
-    case Item::VARBIN_ITEM:
-    case Item::DECIMAL_ITEM:
-      break;
-    default:
-      context->supported= FALSE;
-      break;
-    }
-    
-    DBUG_VOID_RETURN;
-  }
-  
-  if (context->supported)
-  {
-    Ndb_rewrite_context *rewrite_context2= context->rewrite_stack;
-    const Item_func *rewrite_func_item;
-    // Check if we are rewriting some unsupported function call
-    if (rewrite_context2 &&
-        (rewrite_func_item= rewrite_context2->func_item) &&
-        rewrite_context2->count++ == 0)
-    {
-      switch (rewrite_func_item->functype()) {
-      case Item_func::BETWEEN:
-        /*
-          Rewrite 
-          <field>|<const> BETWEEN <const1>|<field1> AND <const2>|<field2>
-          to <field>|<const> > <const1>|<field1> AND 
-          <field>|<const> < <const2>|<field2>
-          or actually in prefix format
-          BEGIN(AND) GT(<field>|<const>, <const1>|<field1>), 
-          LT(<field>|<const>, <const2>|<field2>), END()
-        */
-      case Item_func::IN_FUNC:
-      {
-        /*
-          Rewrite <field>|<const> IN(<const1>|<field1>, <const2>|<field2>,..)
-          to <field>|<const> = <const1>|<field1> OR 
-          <field> = <const2>|<field2> ...
-          or actually in prefix format
-          BEGIN(OR) EQ(<field>|<const>, <const1><field1>), 
-          EQ(<field>|<const>, <const2>|<field2>), ... END()
-          Each part of the disjunction is added for each call
-          to ndb_serialize_cond and end of rewrite statement 
-          is wrapped in end of ndb_serialize_cond
-        */
-        if (context->expecting(item->type()))
-        {
-          // This is the <field>|<const> item, save it in the rewrite context
-          rewrite_context2->left_hand_item= item;
-          if (item->type() == Item::FUNC_ITEM)
-          {
-            Item_func *func_item= (Item_func *) item;
-            if (func_item->functype() == Item_func::UNKNOWN_FUNC &&
-                func_item->const_item())
-            {
-              // Skip any arguments since we will evaluate function instead
-              DBUG_PRINT("info", ("Skip until end of arguments marker"));
-              context->skip= func_item->argument_count();
-            }
-            else
-            {
-              DBUG_PRINT("info", ("Found unsupported functional expression in BETWEEN|IN"));
-              context->supported= FALSE;
-              DBUG_VOID_RETURN;
-              
-            }
-          }
-        }
-        else
-        {
-          // Non-supported BETWEEN|IN expression
-          DBUG_PRINT("info", ("Found unexpected item of type %u in BETWEEN|IN",
-                              item->type()));
-          context->supported= FALSE;
-          DBUG_VOID_RETURN;
-        }
-        break;
-      }
-      default:
-        context->supported= FALSE;
-        break;
-      }
-      DBUG_VOID_RETURN;
-    }
-    else
-    {
-      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;
-    // Check if we are rewriting some unsupported function call
-      if (context->rewrite_stack)
-      {
-        Ndb_rewrite_context *rewrite_context= context->rewrite_stack;
-        const Item_func *func_item= rewrite_context->func_item;
-        switch (func_item->functype()) {
-        case Item_func::BETWEEN:
-        {
-          /*
-            Rewrite 
-            <field>|<const> BETWEEN <const1>|<field1> AND <const2>|<field2>
-            to <field>|<const> > <const1>|<field1> AND 
-            <field>|<const> < <const2>|<field2>
-            or actually in prefix format
-            BEGIN(AND) GT(<field>|<const>, <const1>|<field1>), 
-            LT(<field>|<const>, <const2>|<field2>), END()
-          */
-          if (rewrite_context->count == 2)
-          {
-            // Lower limit of BETWEEN
-            DBUG_PRINT("info", ("GE_FUNC"));      
-            curr_cond->ndb_item= new Ndb_item(Item_func::GE_FUNC, 2);
-          }
-          else if (rewrite_context->count == 3)
-          {
-            // Upper limit of BETWEEN
-            DBUG_PRINT("info", ("LE_FUNC"));      
-            curr_cond->ndb_item= new Ndb_item(Item_func::LE_FUNC, 2);
-          }
-          else
-          {
-            // Illegal BETWEEN expression
-            DBUG_PRINT("info", ("Illegal BETWEEN expression"));
-            context->supported= FALSE;
-            DBUG_VOID_RETURN;
-          }
-          break;
-        }
-        case Item_func::IN_FUNC:
-        {
-          /*
-            Rewrite <field>|<const> IN(<const1>|<field1>, <const2>|<field2>,..)
-            to <field>|<const> = <const1>|<field1> OR 
-            <field> = <const2>|<field2> ...
-            or actually in prefix format
-            BEGIN(OR) EQ(<field>|<const>, <const1><field1>), 
-            EQ(<field>|<const>, <const2>|<field2>), ... END()
-            Each part of the disjunction is added for each call
-            to ndb_serialize_cond and end of rewrite statement 
-            is wrapped in end of ndb_serialize_cond
-          */
-          DBUG_PRINT("info", ("EQ_FUNC"));      
-          curr_cond->ndb_item= new Ndb_item(Item_func::EQ_FUNC, 2);
-          break;
-        }
-        default:
-          context->supported= FALSE;
-        }
-        // Handle left hand <field>|<const>
-        context->rewrite_stack= NULL; // Disable rewrite mode
-        context->expect_only(Item::FIELD_ITEM);
-        context->expect_field_result(STRING_RESULT);
-        context->expect_field_result(REAL_RESULT);
-        context->expect_field_result(INT_RESULT);
-        context->expect_field_result(DECIMAL_RESULT);
-        context->expect(Item::INT_ITEM);
-        context->expect(Item::STRING_ITEM);
-        context->expect(Item::VARBIN_ITEM);
-        context->expect(Item::FUNC_ITEM);
-        ndb_serialize_cond(rewrite_context->left_hand_item, arg);
-        context->skip= 0; // Any FUNC_ITEM expression has already been parsed
-        context->rewrite_stack= rewrite_context; // Enable rewrite mode
-        if (!context->supported)
-          DBUG_VOID_RETURN;
-
-        prev_cond= context->cond_ptr;
-        curr_cond= context->cond_ptr= new Ndb_cond();
-        prev_cond->next= curr_cond;
-      }
-      
-      // Check for end of AND/OR expression
-      if (!item)
-      {
-        // End marker for condition group
-        DBUG_PRINT("info", ("End of condition group"));
-        curr_cond->ndb_item= new Ndb_item(NDB_END_COND);
-      }
-      else
-      {
-        switch (item->type()) {
-        case Item::FIELD_ITEM:
-        {
-          Item_field *field_item= (Item_field *) item;
-          Field *field= field_item->field;
-          enum_field_types type= field->type();
-          /*
-            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->s == field->table->s)
-          {       
-            const NDBTAB *tab= (const NDBTAB *) context->ndb_table;
-            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 ||
-                   type == MYSQL_TYPE_DATE || 
-                   type == MYSQL_TYPE_YEAR ||
-                   type == MYSQL_TYPE_DATETIME)
-                  ? (context->expecting_field_result(STRING_RESULT) ||
-                     context->expecting_field_result(INT_RESULT))
-                  : true)) &&
-                // Bit fields no yet supported in scan filter
-                type != MYSQL_TYPE_BIT &&
-                // No BLOB support in scan filter
-                type != MYSQL_TYPE_TINY_BLOB &&
-                type != MYSQL_TYPE_MEDIUM_BLOB &&
-                type != MYSQL_TYPE_LONG_BLOB &&
-                type != MYSQL_TYPE_BLOB)
-            {
-              const NDBCOL *col= tab->getColumn(field->field_name);
-              DBUG_ASSERT(col);
-              curr_cond->ndb_item= new Ndb_item(field, col->getColumnNo());
-              context->dont_expect(Item::FIELD_ITEM);
-              context->expect_no_field_result();
-              if (! context->expecting_nothing())
-              {
-                // We have not seen second argument yet
-                if (type == MYSQL_TYPE_TIME ||
-                    type == MYSQL_TYPE_DATE || 
-                    type == MYSQL_TYPE_YEAR ||
-                    type == MYSQL_TYPE_DATETIME)
-                {
-                  context->expect_only(Item::STRING_ITEM);
-                  context->expect(Item::INT_ITEM);
-                }
-                else
-                  switch (field->result_type()) {
-                  case STRING_RESULT:
-                    // Expect char string or binary string
-                    context->expect_only(Item::STRING_ITEM);
-                    context->expect(Item::VARBIN_ITEM);
-                    context->expect_collation(field_item->collation.collation);
-                    break;
-                  case REAL_RESULT:
-                    context->expect_only(Item::REAL_ITEM);
-                    context->expect(Item::DECIMAL_ITEM);
-                    context->expect(Item::INT_ITEM);
-                    break;
-                  case INT_RESULT:
-                    context->expect_only(Item::INT_ITEM);
-                    context->expect(Item::VARBIN_ITEM);
-                    break;
-                  case DECIMAL_RESULT:
-                    context->expect_only(Item::DECIMAL_ITEM);
-                    context->expect(Item::REAL_ITEM);
-                    context->expect(Item::INT_ITEM);
-                    break;
-                  default:
-                    break;
-                  }    
-              }
-              else
-              {
-                // Expect another logical expression
-                context->expect_only(Item::FUNC_ITEM);
-                context->expect(Item::COND_ITEM);
-                // Check that field and string constant collations are the same
-                if ((field->result_type() == STRING_RESULT) &&
-                    !context->expecting_collation(item->collation.collation)
-                    && type != MYSQL_TYPE_TIME
-                    && type != MYSQL_TYPE_DATE
-                    && type != MYSQL_TYPE_YEAR
-                    && type != MYSQL_TYPE_DATETIME)
-                {
-                  DBUG_PRINT("info", ("Found non-matching collation %s",  
-                                      item->collation.collation->name)); 
-                  context->supported= FALSE;                
-                }
-              }
-              break;
-            }
-            else
-            {
-              DBUG_PRINT("info", ("Was not expecting field of type %u(%u)",
-                                  field->result_type(), type));
-              context->supported= FALSE;
-            }
-          }
-          else
-          {
-            DBUG_PRINT("info", ("Was not expecting field from table %s(%s)",
-                                context->table->s->table_name, 
-                                field->table->s->table_name));
-            context->supported= FALSE;
-          }
-          break;
-        }
-        case Item::FUNC_ITEM:
-        {
-          Item_func *func_item= (Item_func *) item;
-          // Check that we expect a function or functional expression here
-          if (context->expecting(Item::FUNC_ITEM) || 
-              func_item->functype() == Item_func::UNKNOWN_FUNC)
-            context->expect_nothing();
-          else
-          {
-            // Did not expect function here
-            context->supported= FALSE;
-            break;
-          }
-          
-          switch (func_item->functype()) {
-          case Item_func::EQ_FUNC:
-          {
-            DBUG_PRINT("info", ("EQ_FUNC"));      
-            curr_cond->ndb_item= new Ndb_item(func_item->functype(), 
-                                              func_item);      
-            context->expect(Item::STRING_ITEM);
-            context->expect(Item::INT_ITEM);
-            context->expect(Item::REAL_ITEM);
-            context->expect(Item::DECIMAL_ITEM);
-            context->expect(Item::VARBIN_ITEM);
-            context->expect(Item::FIELD_ITEM);
-            context->expect_field_result(STRING_RESULT);
-            context->expect_field_result(REAL_RESULT);
-            context->expect_field_result(INT_RESULT);
-            context->expect_field_result(DECIMAL_RESULT);
-            break;
-          }
-          case Item_func::NE_FUNC:
-          {
-            DBUG_PRINT("info", ("NE_FUNC"));      
-            curr_cond->ndb_item= new Ndb_item(func_item->functype(),
-                                              func_item);      
-            context->expect(Item::STRING_ITEM);
-            context->expect(Item::INT_ITEM);
-            context->expect(Item::REAL_ITEM);
-            context->expect(Item::DECIMAL_ITEM);
-            context->expect(Item::VARBIN_ITEM);
-            context->expect(Item::FIELD_ITEM);
-            context->expect_field_result(STRING_RESULT);
-            context->expect_field_result(REAL_RESULT);
-            context->expect_field_result(INT_RESULT);
-            context->expect_field_result(DECIMAL_RESULT);
-            break;
-          }
-          case Item_func::LT_FUNC:
-          {
-            DBUG_PRINT("info", ("LT_FUNC"));      
-            curr_cond->ndb_item= new Ndb_item(func_item->functype(),
-                                              func_item);      
-            context->expect(Item::STRING_ITEM);
-            context->expect(Item::INT_ITEM);
-            context->expect(Item::REAL_ITEM);
-            context->expect(Item::DECIMAL_ITEM);
-            context->expect(Item::VARBIN_ITEM);
-            context->expect(Item::FIELD_ITEM);
-            context->expect_field_result(STRING_RESULT);
-            context->expect_field_result(REAL_RESULT);
-            context->expect_field_result(INT_RESULT);
-            context->expect_field_result(DECIMAL_RESULT);
-            break;
-          }
-          case Item_func::LE_FUNC:
-          {
-            DBUG_PRINT("info", ("LE_FUNC"));      
-            curr_cond->ndb_item= new Ndb_item(func_item->functype(),
-                                              func_item);      
-            context->expect(Item::STRING_ITEM);
-            context->expect(Item::INT_ITEM);
-            context->expect(Item::REAL_ITEM);
-            context->expect(Item::DECIMAL_ITEM);
-            context->expect(Item::VARBIN_ITEM);
-            context->expect(Item::FIELD_ITEM);
-            context->expect_field_result(STRING_RESULT);
-            context->expect_field_result(REAL_RESULT);
-            context->expect_field_result(INT_RESULT);
-            context->expect_field_result(DECIMAL_RESULT);
-            break;
-          }
-          case Item_func::GE_FUNC:
-          {
-            DBUG_PRINT("info", ("GE_FUNC"));      
-            curr_cond->ndb_item= new Ndb_item(func_item->functype(),
-                                              func_item);      
-            context->expect(Item::STRING_ITEM);
-            context->expect(Item::INT_ITEM);
-            context->expect(Item::REAL_ITEM);
-            context->expect(Item::DECIMAL_ITEM);
-            context->expect(Item::VARBIN_ITEM);
-            context->expect(Item::FIELD_ITEM);
-            context->expect_field_result(STRING_RESULT);
-            context->expect_field_result(REAL_RESULT);
-            context->expect_field_result(INT_RESULT);
-            context->expect_field_result(DECIMAL_RESULT);
-            break;
-          }
-          case Item_func::GT_FUNC:
-          {
-            DBUG_PRINT("info", ("GT_FUNC"));      
-            curr_cond->ndb_item= new Ndb_item(func_item->functype(),
-                                              func_item);      
-            context->expect(Item::STRING_ITEM);
-            context->expect(Item::REAL_ITEM);
-            context->expect(Item::DECIMAL_ITEM);
-            context->expect(Item::INT_ITEM);
-            context->expect(Item::VARBIN_ITEM);
-            context->expect(Item::FIELD_ITEM);
-            context->expect_field_result(STRING_RESULT);
-            context->expect_field_result(REAL_RESULT);
-            context->expect_field_result(INT_RESULT);
-            context->expect_field_result(DECIMAL_RESULT);
-            break;
-          }
-          case Item_func::LIKE_FUNC:
-          {
-            DBUG_PRINT("info", ("LIKE_FUNC"));      
-            curr_cond->ndb_item= new Ndb_item(func_item->functype(),
-                                              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;
-          }
-          case Item_func::ISNULL_FUNC:
-          {
-            DBUG_PRINT("info", ("ISNULL_FUNC"));      
-            curr_cond->ndb_item= new Ndb_item(func_item->functype(),
-                                              func_item);      
-            context->expect(Item::FIELD_ITEM);
-            context->expect_field_result(STRING_RESULT);
-            context->expect_field_result(REAL_RESULT);
-            context->expect_field_result(INT_RESULT);
-            context->expect_field_result(DECIMAL_RESULT);
-            break;
-          }
-          case Item_func::ISNOTNULL_FUNC:
-          {
-            DBUG_PRINT("info", ("ISNOTNULL_FUNC"));      
-            curr_cond->ndb_item= new Ndb_item(func_item->functype(),
-                                              func_item);     
-            context->expect(Item::FIELD_ITEM);
-            context->expect_field_result(STRING_RESULT);
-            context->expect_field_result(REAL_RESULT);
-            context->expect_field_result(INT_RESULT);
-            context->expect_field_result(DECIMAL_RESULT);
-            break;
-          }
-          case Item_func::NOT_FUNC:
-          {
-            DBUG_PRINT("info", ("NOT_FUNC"));      
-            curr_cond->ndb_item= new Ndb_item(func_item->functype(),
-                                              func_item);     
-            context->expect(Item::FUNC_ITEM);
-            context->expect(Item::COND_ITEM);
-            break;
-          }
-          case Item_func::BETWEEN:
-          {
-            DBUG_PRINT("info", ("BETWEEN, rewriting using AND"));
-            Item_func_between *between_func= (Item_func_between *) func_item;
-            Ndb_rewrite_context *rewrite_context= 
-              new Ndb_rewrite_context(func_item);
-            rewrite_context->next= context->rewrite_stack;
-            context->rewrite_stack= rewrite_context;
-            if (between_func->negated)
-            {
-              DBUG_PRINT("info", ("NOT_FUNC"));
-              curr_cond->ndb_item= new Ndb_item(Item_func::NOT_FUNC, 1);
-              prev_cond= curr_cond;
-              curr_cond= context->cond_ptr= new Ndb_cond();
-              curr_cond->prev= prev_cond;
-              prev_cond->next= curr_cond;
-            }
-            DBUG_PRINT("info", ("COND_AND_FUNC"));
-            curr_cond->ndb_item= 
-              new Ndb_item(Item_func::COND_AND_FUNC, 
-                           func_item->argument_count() - 1);
-            context->expect_only(Item::FIELD_ITEM);
-            context->expect(Item::INT_ITEM);
-            context->expect(Item::STRING_ITEM);
-            context->expect(Item::VARBIN_ITEM);
-            context->expect(Item::FUNC_ITEM);
-            break;
-          }
-          case Item_func::IN_FUNC:
-          {
-            DBUG_PRINT("info", ("IN_FUNC, rewriting using OR"));
-            Item_func_in *in_func= (Item_func_in *) func_item;
-            Ndb_rewrite_context *rewrite_context= 
-              new Ndb_rewrite_context(func_item);
-            rewrite_context->next= context->rewrite_stack;
-            context->rewrite_stack= rewrite_context;
-            if (in_func->negated)
-            {
-              DBUG_PRINT("info", ("NOT_FUNC"));
-              curr_cond->ndb_item= new Ndb_item(Item_func::NOT_FUNC, 1);
-              prev_cond= curr_cond;
-              curr_cond= context->cond_ptr= new Ndb_cond();
-              curr_cond->prev= prev_cond;
-              prev_cond->next= curr_cond;
-            }
-            DBUG_PRINT("info", ("COND_OR_FUNC"));
-            curr_cond->ndb_item= new Ndb_item(Item_func::COND_OR_FUNC, 
-                                              func_item->argument_count() - 1);
-            context->expect_only(Item::FIELD_ITEM);
-            context->expect(Item::INT_ITEM);
-            context->expect(Item::STRING_ITEM);
-            context->expect(Item::VARBIN_ITEM);
-            context->expect(Item::FUNC_ITEM);
-            break;
-          }
-          case Item_func::UNKNOWN_FUNC:
-          {
-            DBUG_PRINT("info", ("UNKNOWN_FUNC %s", 
-                                func_item->const_item()?"const":""));  
-            DBUG_PRINT("info", ("result type %d", func_item->result_type()));
-            if (func_item->const_item())
-            {
-              switch (func_item->result_type()) {
-              case STRING_RESULT:
-              {
-                NDB_ITEM_QUALIFICATION q;
-                q.value_type= Item::STRING_ITEM;
-                curr_cond->ndb_item= new Ndb_item(NDB_VALUE, q, item); 
-                if (! context->expecting_no_field_result())
-                {
-                  // We have not seen the field argument yet
-                  context->expect_only(Item::FIELD_ITEM);
-                  context->expect_only_field_result(STRING_RESULT);
-                  context->expect_collation(func_item->collation.collation);
-                }
-                else
-                {
-                  // Expect another logical expression
-                  context->expect_only(Item::FUNC_ITEM);
-                  context->expect(Item::COND_ITEM);
-                  // Check that string result have correct collation
-                  if (!context->expecting_collation(item->collation.collation))
-                  {
-                    DBUG_PRINT("info", ("Found non-matching collation %s",  
-                                        item->collation.collation->name));
-                    context->supported= FALSE;
-                  }
-                }
-                // Skip any arguments since we will evaluate function instead
-                DBUG_PRINT("info", ("Skip until end of arguments marker"));
-                context->skip= func_item->argument_count();
-                break;
-              }
-              case REAL_RESULT:
-              {
-                NDB_ITEM_QUALIFICATION q;
-                q.value_type= Item::REAL_ITEM;
-                curr_cond->ndb_item= new Ndb_item(NDB_VALUE, q, item);
-                if (! context->expecting_no_field_result()) 
-                {
-                  // We have not seen the field argument yet
-                  context->expect_only(Item::FIELD_ITEM);
-                  context->expect_only_field_result(REAL_RESULT);
-                }
-                else
-                {
-                  // Expect another logical expression
-                  context->expect_only(Item::FUNC_ITEM);
-                  context->expect(Item::COND_ITEM);
-                }
-                
-                // Skip any arguments since we will evaluate function instead
-                DBUG_PRINT("info", ("Skip until end of arguments marker"));
-                context->skip= func_item->argument_count();
-                break;
-              }
-              case INT_RESULT:
-              {
-                NDB_ITEM_QUALIFICATION q;
-                q.value_type= Item::INT_ITEM;
-                curr_cond->ndb_item= new Ndb_item(NDB_VALUE, q, item);
-                if (! context->expecting_no_field_result()) 
-                {
-                  // We have not seen the field argument yet
-                  context->expect_only(Item::FIELD_ITEM);
-                  context->expect_only_field_result(INT_RESULT);
-                }
-                else
-                {
-                  // Expect another logical expression
-                  context->expect_only(Item::FUNC_ITEM);
-                  context->expect(Item::COND_ITEM);
-                }
-                
-                // Skip any arguments since we will evaluate function instead
-                DBUG_PRINT("info", ("Skip until end of arguments marker"));
-                context->skip= func_item->argument_count();
-                break;
-              }
-              case DECIMAL_RESULT:
-              {
-                NDB_ITEM_QUALIFICATION q;
-                q.value_type= Item::DECIMAL_ITEM;
-                curr_cond->ndb_item= new Ndb_item(NDB_VALUE, q, item);
-                if (! context->expecting_no_field_result()) 
-                {
-                  // We have not seen the field argument yet
-                  context->expect_only(Item::FIELD_ITEM);
-                  context->expect_only_field_result(DECIMAL_RESULT);
-                }
-                else
-                {
-                  // Expect another logical expression
-                  context->expect_only(Item::FUNC_ITEM);
-                  context->expect(Item::COND_ITEM);
-                }
-                // Skip any arguments since we will evaluate function instead
-                DBUG_PRINT("info", ("Skip until end of arguments marker"));
-                context->skip= func_item->argument_count();
-                break;
-              }
-              default:
-                break;
-              }
-            }
-            else
-              // Function does not return constant expression
-              context->supported= FALSE;
-            break;
-          }
-          default:
-          {
-            DBUG_PRINT("info", ("Found func_item of type %d", 
-                                func_item->functype()));
-            context->supported= FALSE;
-          }
-          }
-          break;
-        }
-        case Item::STRING_ITEM:
-          DBUG_PRINT("info", ("STRING_ITEM")); 
-          if (context->expecting(Item::STRING_ITEM)) 
-          {
-#ifndef DBUG_OFF
-            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", ("value \"%s\"", 
-                                string_item->val_str(&str)->ptr()));
-#endif
-            NDB_ITEM_QUALIFICATION q;
-            q.value_type= Item::STRING_ITEM;
-            curr_cond->ndb_item= new Ndb_item(NDB_VALUE, q, item);      
-            if (! context->expecting_no_field_result())
-            {
-              // We have not seen the field argument yet
-              context->expect_only(Item::FIELD_ITEM);
-              context->expect_only_field_result(STRING_RESULT);
-              context->expect_collation(item->collation.collation);
-            }
-            else 
-            {
-              // Expect another logical expression
-              context->expect_only(Item::FUNC_ITEM);
-              context->expect(Item::COND_ITEM);
-              // Check that we are comparing with a field with same collation
-              if (!context->expecting_collation(item->collation.collation))
-              {
-                DBUG_PRINT("info", ("Found non-matching collation %s",  
-                                    item->collation.collation->name));
-                context->supported= FALSE;
-              }
-            }
-          }
-          else
-            context->supported= FALSE;
-          break;
-        case Item::INT_ITEM:
-          DBUG_PRINT("info", ("INT_ITEM"));
-          if (context->expecting(Item::INT_ITEM)) 
-          {
-            DBUG_PRINT("info", ("value %ld",
-                                (long) ((Item_int*) item)->value));
-            NDB_ITEM_QUALIFICATION q;
-            q.value_type= Item::INT_ITEM;
-            curr_cond->ndb_item= new Ndb_item(NDB_VALUE, q, item);
-            if (! context->expecting_no_field_result()) 
-            {
-              // We have not seen the field argument yet
-              context->expect_only(Item::FIELD_ITEM);
-              context->expect_only_field_result(INT_RESULT);
-              context->expect_field_result(REAL_RESULT);
-              context->expect_field_result(DECIMAL_RESULT);
-            }
-            else
-            {
-              // Expect another logical expression
-              context->expect_only(Item::FUNC_ITEM);
-              context->expect(Item::COND_ITEM);
-            }
-          }
-          else
-            context->supported= FALSE;
-          break;
-        case Item::REAL_ITEM:
-          DBUG_PRINT("info", ("REAL_ITEM"));
-          if (context->expecting(Item::REAL_ITEM)) 
-          {
-            DBUG_PRINT("info", ("value %f", ((Item_float *) item)->value));
-            NDB_ITEM_QUALIFICATION q;
-            q.value_type= Item::REAL_ITEM;
-            curr_cond->ndb_item= new Ndb_item(NDB_VALUE, q, item);
-            if (! context->expecting_no_field_result()) 
-            {
-              // We have not seen the field argument yet
-              context->expect_only(Item::FIELD_ITEM);
-              context->expect_only_field_result(REAL_RESULT);
-            }
-            else
-            {
-              // Expect another logical expression
-              context->expect_only(Item::FUNC_ITEM);
-              context->expect(Item::COND_ITEM);
-            }
-          }
-          else
-            context->supported= FALSE;
-          break;
-        case Item::VARBIN_ITEM:
-          DBUG_PRINT("info", ("VARBIN_ITEM"));
-          if (context->expecting(Item::VARBIN_ITEM)) 
-          {
-            NDB_ITEM_QUALIFICATION q;
-            q.value_type= Item::VARBIN_ITEM;
-            curr_cond->ndb_item= new Ndb_item(NDB_VALUE, q, item);      
-            if (! context->expecting_no_field_result())
-            {
-              // We have not seen the field argument yet
-              context->expect_only(Item::FIELD_ITEM);
-              context->expect_only_field_result(STRING_RESULT);
-            }
-            else
-            {
-              // Expect another logical expression
-              context->expect_only(Item::FUNC_ITEM);
-              context->expect(Item::COND_ITEM);
-            }
-          }
-          else
-            context->supported= FALSE;
-          break;
-        case Item::DECIMAL_ITEM:
-          DBUG_PRINT("info", ("DECIMAL_ITEM"));
-          if (context->expecting(Item::DECIMAL_ITEM)) 
-          {
-            DBUG_PRINT("info", ("value %f",
-                                ((Item_decimal*) item)->val_real()));
-            NDB_ITEM_QUALIFICATION q;
-            q.value_type= Item::DECIMAL_ITEM;
-            curr_cond->ndb_item= new Ndb_item(NDB_VALUE, q, item);
-            if (! context->expecting_no_field_result()) 
-            {
-              // We have not seen the field argument yet
-              context->expect_only(Item::FIELD_ITEM);
-              context->expect_only_field_result(REAL_RESULT);
-              context->expect_field_result(DECIMAL_RESULT);
-            }
-            else
-            {
-              // Expect another logical expression
-              context->expect_only(Item::FUNC_ITEM);
-              context->expect(Item::COND_ITEM);
-            }
-          }
-          else
-            context->supported= FALSE;
-          break;
-        case Item::COND_ITEM:
-        {
-          Item_cond *cond_item= (Item_cond *) item;
-          
-          if (context->expecting(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(),
-                                                cond_item);      
-              break;
-            case Item_func::COND_OR_FUNC:
-              DBUG_PRINT("info", ("COND_OR_FUNC"));
-              curr_cond->ndb_item= new Ndb_item(cond_item->functype(),
-                                                cond_item);      
-              break;
-            default:
-              DBUG_PRINT("info", ("COND_ITEM %d", cond_item->functype()));
-              context->supported= FALSE;
-              break;
-            }
-          }
-          else
-          {
-            /* Did not expect condition */
-            context->supported= FALSE;          
-          }
-          break;
-        }
-        default:
-        {
-          DBUG_PRINT("info", ("Found item of type %d", item->type()));
-          context->supported= FALSE;
-        }
-        }
-      }
-      if (context->supported && context->rewrite_stack)
-      {
-        Ndb_rewrite_context *rewrite_context= context->rewrite_stack;
-        if (rewrite_context->count == 
-            rewrite_context->func_item->argument_count())
-        {
-          // Rewrite is done, wrap an END() at the en
-          DBUG_PRINT("info", ("End of condition group"));
-          prev_cond= curr_cond;
-          curr_cond= context->cond_ptr= new Ndb_cond();
-          curr_cond->prev= prev_cond;
-          prev_cond->next= curr_cond;
-          curr_cond->ndb_item= new Ndb_item(NDB_END_COND);
-          // Pop rewrite stack
-          context->rewrite_stack=  rewrite_context->next;
-          rewrite_context->next= NULL;
-          delete(rewrite_context);
-        }
-      }
-    }
-  }
- 
-  DBUG_VOID_RETURN;
-}
-
-bool
-ha_ndbcluster::serialize_cond(const COND *cond, Ndb_cond_stack *ndb_cond)
-{
-  DBUG_ENTER("serialize_cond");
-  Item *item= (Item *) cond;
-  Ndb_cond_traverse_context context(table, (void *)m_table, ndb_cond);
-  // Expect a logical expression
-  context.expect(Item::FUNC_ITEM);
-  context.expect(Item::COND_ITEM);
-  item->traverse_cond(&ndb_serialize_cond, (void *) &context, Item::PREFIX);
-  DBUG_PRINT("info", ("The pushed condition is %ssupported", (context.supported)?"":"not "));
-
-  DBUG_RETURN(context.supported);
-}
-
-int
-ha_ndbcluster::build_scan_filter_predicate(Ndb_cond * &cond, 
-                                           NdbScanFilter *filter,
-                                           bool negated)
-{
-  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;
-    Ndb_item *b, *field, *value= NULL;
-    LINT_INIT(field);
-
-    switch (cond->ndb_item->argument_count()) {
-    case 1:
-      field= 
-        (a->type == NDB_FIELD)? a : NULL;
-      break;
-    case 2:
-      if (!cond->next->next)
-        break;
-      b= cond->next->next->ndb_item;
-      value= 
-        (a->type == NDB_VALUE)? a
-        : (b->type == NDB_VALUE)? b
-        : NULL;
-      field= 
-        (a->type == NDB_FIELD)? a
-        : (b->type == NDB_FIELD)? b
-        : NULL;
-      break;
-    default:
-      break;
-    }
-    switch ((negated) ? 
-            Ndb_item::negate(cond->ndb_item->qualification.function_type)
-            : cond->ndb_item->qualification.function_type) {
-    case NDB_EQ_FUNC:
-    {
-      if (!value || !field) break;
-      // Save value in right format for the field type
-      value->save_in_field(field);
-      DBUG_PRINT("info", ("Generating EQ filter"));
-      if (filter->cmp(NdbScanFilter::COND_EQ, 
-                      field->get_field_no(),
-                      field->get_val(),
-                      field->pack_length()) == -1)
-        DBUG_RETURN(1);
-      cond= cond->next->next->next;
-      DBUG_RETURN(0);
-    }
-    case NDB_NE_FUNC:
-    {
-      if (!value || !field) break;
-      // Save value in right format for the field type
-      value->save_in_field(field);
-      DBUG_PRINT("info", ("Generating NE filter"));
-      if (filter->cmp(NdbScanFilter::COND_NE, 
-                      field->get_field_no(),
-                      field->get_val(),
-                      field->pack_length()) == -1)
-        DBUG_RETURN(1);
-      cond= cond->next->next->next;
-      DBUG_RETURN(0);
-    }
-    case NDB_LT_FUNC:
-    {
-      if (!value || !field) break;
-      // Save value in right format for the field type
-      value->save_in_field(field);
-      if (a == field)
-      {
-        DBUG_PRINT("info", ("Generating LT filter")); 
-        if (filter->cmp(NdbScanFilter::COND_LT, 
-                        field->get_field_no(),
-                        field->get_val(),
-                        field->pack_length()) == -1)
-          DBUG_RETURN(1);
-      }
-      else
-      {
-        DBUG_PRINT("info", ("Generating GT filter")); 
-        if (filter->cmp(NdbScanFilter::COND_GT, 
-                        field->get_field_no(),
-                        field->get_val(),
-                        field->pack_length()) == -1)
-          DBUG_RETURN(1);
-      }
-      cond= cond->next->next->next;
-      DBUG_RETURN(0);
-    }
-    case NDB_LE_FUNC:
-    {
-      if (!value || !field) break;
-      // Save value in right format for the field type
-      value->save_in_field(field);
-      if (a == field)
-      {
-        DBUG_PRINT("info", ("Generating LE filter")); 
-        if (filter->cmp(NdbScanFilter::COND_LE, 
-                        field->get_field_no(),
-                        field->get_val(),
-                        field->pack_length()) == -1)
-          DBUG_RETURN(1);       
-      }
-      else
-      {
-        DBUG_PRINT("info", ("Generating GE filter")); 
-        if (filter->cmp(NdbScanFilter::COND_GE, 
-                        field->get_field_no(),
-                        field->get_val(),
-                        field->pack_length()) == -1)
-          DBUG_RETURN(1);
-      }
-      cond= cond->next->next->next;
-      DBUG_RETURN(0);
-    }
-    case NDB_GE_FUNC:
-    {
-      if (!value || !field) break;
-      // Save value in right format for the field type
-      value->save_in_field(field);
-      if (a == field)
-      {
-        DBUG_PRINT("info", ("Generating GE filter")); 
-        if (filter->cmp(NdbScanFilter::COND_GE, 
-                        field->get_field_no(),
-                        field->get_val(),
-                        field->pack_length()) == -1)
-          DBUG_RETURN(1);
-      }
-      else
-      {
-        DBUG_PRINT("info", ("Generating LE filter")); 
-        if (filter->cmp(NdbScanFilter::COND_LE, 
-                        field->get_field_no(),
-                        field->get_val(),
-                        field->pack_length()) == -1)
-          DBUG_RETURN(1);
-      }
-      cond= cond->next->next->next;
-      DBUG_RETURN(0);
-    }
-    case NDB_GT_FUNC:
-    {
-      if (!value || !field) break;
-      // Save value in right format for the field type
-      value->save_in_field(field);
-      if (a == field)
-      {
-        DBUG_PRINT("info", ("Generating GT filter"));
-        if (filter->cmp(NdbScanFilter::COND_GT, 
-                        field->get_field_no(),
-                        field->get_val(),
-                        field->pack_length()) == -1)
-          DBUG_RETURN(1);
-      }
-      else
-      {
-        DBUG_PRINT("info", ("Generating LT filter"));
-        if (filter->cmp(NdbScanFilter::COND_LT, 
-                        field->get_field_no(),
-                        field->get_val(),
-                        field->pack_length()) == -1)
-          DBUG_RETURN(1);
-      }
-      cond= cond->next->next->next;
-      DBUG_RETURN(0);
-    }
-    case NDB_LIKE_FUNC:
-    {
-      if (!value || !field) break;
-      if ((value->qualification.value_type != Item::STRING_ITEM) &&
-          (value->qualification.value_type != Item::VARBIN_ITEM))
-          break;
-      // Save value in right format for the field type
-      value->save_in_field(field);
-      DBUG_PRINT("info", ("Generating LIKE filter: like(%d,%s,%d)", 
-                          field->get_field_no(), value->get_val(), 
-                          value->pack_length()));
-      if (filter->cmp(NdbScanFilter::COND_LIKE, 
-                      field->get_field_no(),
-                      value->get_val(),
-                      value->pack_length()) == -1)
-        DBUG_RETURN(1);
-      cond= cond->next->next->next;
-      DBUG_RETURN(0);
-    }
-    case NDB_NOTLIKE_FUNC:
-    {
-      if (!value || !field) break;
-      if ((value->qualification.value_type != Item::STRING_ITEM) &&
-          (value->qualification.value_type != Item::VARBIN_ITEM))
-          break;
-      // Save value in right format for the field type
-      value->save_in_field(field);
-      DBUG_PRINT("info", ("Generating NOTLIKE filter: notlike(%d,%s,%d)", 
-                          field->get_field_no(), value->get_val(), 
-                          value->pack_length()));
-      if (filter->cmp(NdbScanFilter::COND_NOT_LIKE, 
-                      field->get_field_no(),
-                      value->get_val(),
-                      value->pack_length()) == -1)
-        DBUG_RETURN(1);
-      cond= cond->next->next->next;
-      DBUG_RETURN(0);
-    }
-    case NDB_ISNULL_FUNC:
-      if (!field)
-        break;
-      DBUG_PRINT("info", ("Generating ISNULL filter"));
-      if (filter->isnull(field->get_field_no()) == -1)
-        DBUG_RETURN(1);
-      cond= cond->next->next;
-      DBUG_RETURN(0);
-    case NDB_ISNOTNULL_FUNC:
-    {
-      if (!field)
-        break;
-      DBUG_PRINT("info", ("Generating ISNOTNULL filter"));
-      if (filter->isnotnull(field->get_field_no()) == -1)
-        DBUG_RETURN(1);         
-      cond= cond->next->next;
-      DBUG_RETURN(0);
-    }
-    default:
-      break;
-    }
-    break;
-  }
-  default:
-    break;
-  }
-  DBUG_PRINT("info", ("Found illegal condition"));
-  DBUG_RETURN(1);
-}
-
-
-int
-ha_ndbcluster::build_scan_filter_group(Ndb_cond* &cond, NdbScanFilter *filter)
-{
-  uint level=0;
-  bool negated= FALSE;
-  DBUG_ENTER("build_scan_filter_group");
-
-  do
-  {
-    if (!cond)
-      DBUG_RETURN(1);
-    switch (cond->ndb_item->type) {
-    case NDB_FUNCTION:
-    {
-      switch (cond->ndb_item->qualification.function_type) {
-      case NDB_COND_AND_FUNC:
-      {
-        level++;
-        DBUG_PRINT("info", ("Generating %s group %u", (negated)?"NAND":"AND",
-                            level));
-        if ((negated) ? filter->begin(NdbScanFilter::NAND)
-            : filter->begin(NdbScanFilter::AND) == -1)
-          DBUG_RETURN(1);
-        negated= FALSE;
-        cond= cond->next;
-        break;
-      }
-      case NDB_COND_OR_FUNC:
-      {
-        level++;
-        DBUG_PRINT("info", ("Generating %s group %u", (negated)?"NOR":"OR",
-                            level));
-        if ((negated) ? filter->begin(NdbScanFilter::NOR)
-            : filter->begin(NdbScanFilter::OR) == -1)
-          DBUG_RETURN(1);
-        negated= FALSE;
-        cond= cond->next;
-        break;
-      }
-      case NDB_NOT_FUNC:
-      {
-        DBUG_PRINT("info", ("Generating negated query"));
-        cond= cond->next;
-        negated= TRUE;
-        break;
-      }
-      default:
-        if (build_scan_filter_predicate(cond, filter, negated))
-          DBUG_RETURN(1);
-        negated= FALSE;
-        break;
-      }
-      break;
-    }
-    case NDB_END_COND:
-      DBUG_PRINT("info", ("End of group %u", level));
-      level--;
-      if (cond) cond= cond->next;
-      if (filter->end() == -1)
-        DBUG_RETURN(1);
-      if (!negated)
-        break;
-      // else fall through (NOT END is an illegal condition)
-    default:
-    {
-      DBUG_PRINT("info", ("Illegal scan filter"));
-    }
-    }
-  }  while (level > 0 || negated);
-  
-  DBUG_RETURN(0);
-}
-
-
-int
-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 NDB_FUNCTION:
-      switch (cond->ndb_item->qualification.function_type) {
-      case NDB_COND_AND_FUNC:
-      case NDB_COND_OR_FUNC:
-        simple_cond= FALSE;
-        break;
-      default:
-        break;
-      }
-      break;
-    default:
-      break;
-    }
-  if (simple_cond && filter->begin() == -1)
-    DBUG_RETURN(1);
-  if (build_scan_filter_group(cond, filter))
-    DBUG_RETURN(1);
-  if (simple_cond && filter->end() == -1)
-    DBUG_RETURN(1);
-
-  DBUG_RETURN(0);
-}
-
-int
-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);
-    
-    DBUG_RETURN(generate_scan_filter_from_cond(ndb_cond_stack, filter));
-  }
-  else
-  {  
-    DBUG_PRINT("info", ("Empty stack"));
-  }
-
-  DBUG_RETURN(0);
-}
-
-
-int
-ha_ndbcluster::generate_scan_filter_from_cond(Ndb_cond_stack *ndb_cond_stack,
-					      NdbScanFilter& filter)
-{
-  bool multiple_cond= FALSE;
-  DBUG_ENTER("generate_scan_filter_from_cond");
-
-  // Wrap an AND group around multiple conditions
-  if (ndb_cond_stack->next) 
-  {
-    multiple_cond= TRUE;
-    if (filter.begin() == -1)
-      DBUG_RETURN(1); 
-  }
-  for (Ndb_cond_stack *stack= ndb_cond_stack; 
-       (stack); 
-       stack= stack->next)
-  {
-    Ndb_cond *cond= stack->ndb_cond;
-    
-    if (build_scan_filter(cond, &filter))
-    {
-      DBUG_PRINT("info", ("build_scan_filter failed"));
-      DBUG_RETURN(1);
-    }
-  }
-  if (multiple_cond && filter.end() == -1)
-    DBUG_RETURN(1);
-
-  DBUG_RETURN(0);
-}
-
-
-int ha_ndbcluster::generate_scan_filter_from_key(NdbScanOperation *op,
-						 const KEY* key_info, 
-						 const byte *key, 
-						 uint key_len,
-						 byte *buf)
-{
-  KEY_PART_INFO* key_part= key_info->key_part;
-  KEY_PART_INFO* end= key_part+key_info->key_parts;
-  NdbScanFilter filter(op);
-  int res;
-  DBUG_ENTER("generate_scan_filter_from_key");
-
-  filter.begin(NdbScanFilter::AND);
-  for (; key_part != end; key_part++) 
-  {
-    Field* field= key_part->field;
-    uint32 pack_len= field->pack_length();
-    const byte* ptr= key;
-    DBUG_PRINT("info", ("Filtering value for %s", field->field_name));
-    DBUG_DUMP("key", (char*)ptr, pack_len);
-    if (key_part->null_bit)
-    {
-      DBUG_PRINT("info", ("Generating ISNULL filter"));
-      if (filter.isnull(key_part->fieldnr-1) == -1)
-	DBUG_RETURN(1);
-    }
-    else
-    {
-      DBUG_PRINT("info", ("Generating EQ filter"));
-      if (filter.cmp(NdbScanFilter::COND_EQ, 
-		     key_part->fieldnr-1,
-		     ptr,
-		     pack_len) == -1)
-	DBUG_RETURN(1);
-    }
-    key += key_part->store_length;
-  }      
-  // Add any pushed condition
-  if (m_cond_stack &&
-      (res= generate_scan_filter_from_cond(m_cond_stack, filter)))
-    DBUG_RETURN(res);
-    
-  if (filter.end() == -1)
-    DBUG_RETURN(1);
-
-  DBUG_RETURN(0);
-}
-
 int
 ndbcluster_show_status(THD* thd)
 {
@@ -8422,4 +7030,50 @@ ndbcluster_show_status(THD* thd)
   DBUG_RETURN(FALSE);
 }
 
+/*
+  Condition pushdown
+*/
+/*
+  Push a condition to ndbcluster storage engine for evaluation 
+  during table   and index scans. The conditions will be stored on a stack
+  for possibly storing several conditions. The stack can be popped
+  by calling cond_pop, handler::extra(HA_EXTRA_RESET) (handler::reset())
+  will clear the stack.
+  The current implementation supports arbitrary AND/OR nested conditions
+  with comparisons between columns and constants (including constant
+  expressions and function calls) and the following comparison operators:
+  =, !=, >, >=, <, <=, "is null", and "is not null".
+  
+  RETURN
+    NULL The condition was supported and will be evaluated for each 
+    row found during the scan
+    cond The condition was not supported and all rows will be returned from
+         the scan for evaluation (and thus not saved on stack)
+*/
+const 
+COND* 
+ha_ndbcluster::cond_push(const COND *cond) 
+{ 
+  DBUG_ENTER("cond_push");
+  if (!m_cond) 
+    m_cond= new ha_ndbcluster_cond;
+  if (!m_cond)
+  {
+    my_errno= HA_ERR_OUT_OF_MEM;
+    DBUG_RETURN(NULL);
+  }
+  DBUG_EXECUTE("where",print_where((COND *)cond, m_tabname););
+  DBUG_RETURN(m_cond->cond_push(cond, table, (NDBTAB *)m_table));
+}
+
+/*
+  Pop the top condition from the condition stack of the handler instance.
+*/
+void 
+ha_ndbcluster::cond_pop() 
+{ 
+  if (m_cond)
+    m_cond->cond_pop();
+}
+
 #endif /* HAVE_NDBCLUSTER_DB */
diff --git a/sql/ha_ndbcluster.h b/sql/ha_ndbcluster.h
index 3495f35e10f..3904d3da416 100644
--- a/sql/ha_ndbcluster.h
+++ b/sql/ha_ndbcluster.h
@@ -27,15 +27,15 @@
 #include <ndbapi_limits.h>
 
 #define NDB_HIDDEN_PRIMARY_KEY_LENGTH 8
-
-class Ndb;             // Forward declaration
-class NdbOperation;    // Forward declaration
-class NdbTransaction;  // Forward declaration
-class NdbRecAttr;      // Forward declaration
+/* Forward declarations */
+class Ndb;             
+class NdbOperation;    
+class NdbTransaction;  
+class NdbRecAttr;      
 class NdbScanOperation; 
-class NdbScanFilter; 
 class NdbIndexScanOperation; 
 class NdbBlob;
+class ha_ndbcluster_cond;
 
 // connectstring to cluster if given by mysqld
 extern const char *ndbcluster_connectstring;
@@ -67,417 +67,6 @@ typedef struct st_ndbcluster_share {
   ulonglong commit_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_END_COND = 3 // End marker for condition group
-} NDB_ITEM_TYPE;
-
-typedef enum ndb_func_type {
-  NDB_EQ_FUNC = 0,
-  NDB_NE_FUNC = 1,
-  NDB_LT_FUNC = 2,
-  NDB_LE_FUNC = 3,
-  NDB_GT_FUNC = 4,
-  NDB_GE_FUNC = 5,
-  NDB_ISNULL_FUNC = 6,
-  NDB_ISNOTNULL_FUNC = 7,
-  NDB_LIKE_FUNC = 8,
-  NDB_NOTLIKE_FUNC = 9,
-  NDB_NOT_FUNC = 10,
-  NDB_UNKNOWN_FUNC = 11,
-  NDB_COND_AND_FUNC = 12,
-  NDB_COND_OR_FUNC = 13,
-  NDB_UNSUPPORTED_FUNC = 14
-} NDB_FUNC_TYPE;
-
-typedef union ndb_item_qualification {
-  Item::Type value_type; 
-  enum_field_types field_type; // Instead of Item::FIELD_ITEM
-  NDB_FUNC_TYPE function_type; // Instead of Item::FUNC_ITEM
-} NDB_ITEM_QUALIFICATION;
-
-typedef struct ndb_item_field_value {
-  Field* field;
-  int column_no;
-} NDB_ITEM_FIELD_VALUE;
-
-typedef union ndb_item_value {
-  const Item *item;
-  NDB_ITEM_FIELD_VALUE *field_value;
-  uint arg_count;
-} NDB_ITEM_VALUE;
-
-struct negated_function_mapping
-{
-  NDB_FUNC_TYPE pos_fun;
-  NDB_FUNC_TYPE neg_fun;
-};
-
-/*
-  Define what functions can be negated in condition pushdown.
-  Note, these HAVE to be in the same order as in definition enum
-*/
-static const negated_function_mapping neg_map[]= 
-{
-  {NDB_EQ_FUNC, NDB_NE_FUNC},
-  {NDB_NE_FUNC, NDB_EQ_FUNC},
-  {NDB_LT_FUNC, NDB_GE_FUNC},
-  {NDB_LE_FUNC, NDB_GT_FUNC},
-  {NDB_GT_FUNC, NDB_LE_FUNC},
-  {NDB_GE_FUNC, NDB_LT_FUNC},
-  {NDB_ISNULL_FUNC, NDB_ISNOTNULL_FUNC},
-  {NDB_ISNOTNULL_FUNC, NDB_ISNULL_FUNC},
-  {NDB_LIKE_FUNC, NDB_NOTLIKE_FUNC},
-  {NDB_NOTLIKE_FUNC, NDB_LIKE_FUNC},
-  {NDB_NOT_FUNC, NDB_UNSUPPORTED_FUNC},
-  {NDB_UNKNOWN_FUNC, NDB_UNSUPPORTED_FUNC},
-  {NDB_COND_AND_FUNC, NDB_UNSUPPORTED_FUNC},
-  {NDB_COND_OR_FUNC, NDB_UNSUPPORTED_FUNC},
-  {NDB_UNSUPPORTED_FUNC, NDB_UNSUPPORTED_FUNC}
-};
-  
-/*
-  This class is the construction element for serialization of Item tree 
-  in condition pushdown.
-  An instance of Ndb_Item represents a constant, table field reference,
-  unary or binary comparison predicate, and start/end of AND/OR.
-  Instances of Ndb_Item are stored in a linked list implemented by Ndb_cond
-  class.
-  The order of elements produced by Ndb_cond::next corresponds to
-  breadth-first traversal of the Item (i.e. expression) tree in prefix order.
-  AND and OR have arbitrary arity, so the end of AND/OR group is marked with  
-  Ndb_item with type == NDB_END_COND.
-  NOT items represent negated conditions and generate NAND/NOR groups.
-*/
-class Ndb_item {
- public:
-  Ndb_item(NDB_ITEM_TYPE item_type) : type(item_type) {};
-  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):
-      value.item= item_value;
-      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):
-      value.item= item_value;
-      value.arg_count= ((Item_func *) item_value)->argument_count();
-      break;
-    case(NDB_END_COND):
-      break;
-    }
-  };
-  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(Item_func::Functype func_type, const Item *item_value) 
-    : type(NDB_FUNCTION)
-  {
-    qualification.function_type= item_func_to_ndb_func(func_type);
-    value.item= item_value;
-    value.arg_count= ((Item_func *) item_value)->argument_count();
-  };
-  Ndb_item(Item_func::Functype func_type, uint no_args) 
-    : type(NDB_FUNCTION)
-  {
-    qualification.function_type= item_func_to_ndb_func(func_type);
-    value.arg_count= no_args;
-  };
-  ~Ndb_item()
-  { 
-    if (type == NDB_FIELD)
-      {
-        delete value.field_value;
-        value.field_value= NULL;
-      }
-  };
-
-  uint32 pack_length() 
-  { 
-    switch(type) {
-    case(NDB_VALUE):
-      if(qualification.value_type == Item::STRING_ITEM)
-        return value.item->str_value.length();
-      break;
-    case(NDB_FIELD):
-      return value.field_value->field->pack_length(); 
-    default:
-      break;
-    }
-    
-    return 0;
-  };
-
-  Field * get_field() { return value.field_value->field; };
-
-  int get_field_no() { return value.field_value->column_no; };
-
-  int argument_count() 
-  { 
-    return value.arg_count;
-  };
-
-  const char* get_val() 
-  {  
-    switch(type) {
-    case(NDB_VALUE):
-      if(qualification.value_type == Item::STRING_ITEM)
-        return value.item->str_value.ptr();
-      break;
-    case(NDB_FIELD):
-      return value.field_value->field->ptr; 
-    default:
-      break;
-    }
-    
-    return NULL;
-  };
-
-  void save_in_field(Ndb_item *field_item)
-  {
-    Field *field = field_item->value.field_value->field;
-    const Item *item= value.item;
-
-    if (item && field)
-      ((Item *)item)->save_in_field(field, false);
-  };
-
-  static NDB_FUNC_TYPE item_func_to_ndb_func(Item_func::Functype fun)
-  {
-    switch (fun) {
-    case (Item_func::EQ_FUNC): { return NDB_EQ_FUNC; }
-    case (Item_func::NE_FUNC): { return NDB_NE_FUNC; }
-    case (Item_func::LT_FUNC): { return NDB_LT_FUNC; }
-    case (Item_func::LE_FUNC): { return NDB_LE_FUNC; }
-    case (Item_func::GT_FUNC): { return NDB_GT_FUNC; }
-    case (Item_func::GE_FUNC): { return NDB_GE_FUNC; }
-    case (Item_func::ISNULL_FUNC): { return NDB_ISNULL_FUNC; }
-    case (Item_func::ISNOTNULL_FUNC): { return NDB_ISNOTNULL_FUNC; }
-    case (Item_func::LIKE_FUNC): { return NDB_LIKE_FUNC; }
-    case (Item_func::NOT_FUNC): { return NDB_NOT_FUNC; }
-    case (Item_func::UNKNOWN_FUNC): { return NDB_UNKNOWN_FUNC; }
-    case (Item_func::COND_AND_FUNC): { return NDB_COND_AND_FUNC; }
-    case (Item_func::COND_OR_FUNC): { return NDB_COND_OR_FUNC; }
-    default: { return NDB_UNSUPPORTED_FUNC; }
-    }
-  };
-
-  static NDB_FUNC_TYPE negate(NDB_FUNC_TYPE fun)
-  {
-    uint i= (uint) fun;
-    DBUG_ASSERT(fun == neg_map[i].pos_fun);
-    return  neg_map[i].neg_fun;
-  };
-
-  NDB_ITEM_TYPE type;
-  NDB_ITEM_QUALIFICATION qualification;
- private:
-  NDB_ITEM_VALUE value;
-};
-
-/*
-  This class implements a linked list used for storing a
-  serialization of the Item tree for condition pushdown.
- */
-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;
-};
-
-/*
-  This class implements a stack for storing several conditions
-  for pushdown (represented as serialized Item trees using Ndb_cond).
-  The current implementation only pushes one condition, but is
-  prepared for handling several (C1 AND C2 ...) if the logic for 
-  pushing conditions is extended in sql_select.
-*/
-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; 
-    if (next) delete next;
-    next= NULL; 
-  };
-  Ndb_cond *ndb_cond;
-  Ndb_cond_stack *next;
-};
-
-class Ndb_rewrite_context
-{
-public:
-  Ndb_rewrite_context(Item_func *func) 
-    : func_item(func), left_hand_item(NULL), count(0) {};
-  ~Ndb_rewrite_context()
-  {
-    if (next) delete next;
-  }
-  const Item_func *func_item;
-  const Item *left_hand_item;
-  uint count;
-  Ndb_rewrite_context *next;
-};
-
-/*
-  This class is used for storing the context when traversing
-  the Item tree. It stores a reference to the table the condition
-  is defined on, the serialized representation being generated, 
-  if the condition found is supported, and information what is
-  expected next in the tree inorder for the condition to be supported.
-*/
-class Ndb_cond_traverse_context 
-{
- public:
-  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),
-    skip(0), collation(NULL), rewrite_stack(NULL)
-  {
-    // Allocate type checking bitmaps   
-    bitmap_init(&expect_mask, 0, 512, FALSE);
-    bitmap_init(&expect_field_type_mask, 0, 512, FALSE);
-    bitmap_init(&expect_field_result_mask, 0, 512, FALSE);
-
-    if (stack)
-      cond_ptr= stack->ndb_cond;
-  };
-  ~Ndb_cond_traverse_context()
-  {
-    bitmap_free(&expect_mask);
-    bitmap_free(&expect_field_type_mask);
-    bitmap_free(&expect_field_result_mask);
-    if (rewrite_stack) delete rewrite_stack;
-  }
-  void expect(Item::Type type)
-  {
-    bitmap_set_bit(&expect_mask, (uint) type);
-    if (type == Item::FIELD_ITEM) expect_all_field_types();
-  };
-  void dont_expect(Item::Type type)
-  {
-    bitmap_clear_bit(&expect_mask, (uint) type);
-  };
-  bool expecting(Item::Type type)
-  {
-    return bitmap_is_set(&expect_mask, (uint) type);
-  };
-  void expect_nothing()
-  {
-    bitmap_clear_all(&expect_mask);
-  };
-  bool expecting_nothing()
-  {
-    return bitmap_is_clear_all(&expect_mask);
-  }
-  void expect_only(Item::Type type)
-  {
-    expect_nothing();
-    expect(type);
-  };
-
-  void expect_field_type(enum_field_types type)
-  {
-    bitmap_set_bit(&expect_field_type_mask, (uint) type);
-  };
-  void expect_all_field_types()
-  {
-    bitmap_set_all(&expect_field_type_mask);
-  };
-  bool expecting_field_type(enum_field_types type)
-  {
-    return bitmap_is_set(&expect_field_type_mask, (uint) type);
-  };
-  void expect_no_field_type()
-  {
-    bitmap_clear_all(&expect_field_type_mask);
-  };
-  bool expecting_no_field_type()
-  {
-    return bitmap_is_clear_all(&expect_field_type_mask);
-  }
-  void expect_only_field_type(enum_field_types result)
-  {
-    expect_no_field_type();
-    expect_field_type(result);
-  };
-
-  void expect_field_result(Item_result result)
-  {
-    bitmap_set_bit(&expect_field_result_mask, (uint) result);
-  };
-  bool expecting_field_result(Item_result result)
-  {
-    return bitmap_is_set(&expect_field_result_mask, (uint) result);
-  };
-  void expect_no_field_result()
-  {
-    bitmap_clear_all(&expect_field_result_mask);
-  };
-  bool expecting_no_field_result()
-  {
-    return bitmap_is_clear_all(&expect_field_result_mask);
-  }
-  void expect_only_field_result(Item_result result)
-  {
-    expect_no_field_result();
-    expect_field_result(result);
-  };
-  void expect_collation(CHARSET_INFO* col)
-  {
-    collation= col;
-  };
-  bool expecting_collation(CHARSET_INFO* col)
-  {
-    bool matching= (!collation) ? true : (collation == col);
-    collation= NULL;
-
-    return matching;
-  };
-
-  TABLE* table;
-  void* ndb_table;
-  bool supported;
-  Ndb_cond_stack* stack_ptr;
-  Ndb_cond* cond_ptr;
-  MY_BITMAP expect_mask;
-  MY_BITMAP expect_field_type_mask;
-  MY_BITMAP expect_field_result_mask;
-  uint skip;
-  CHARSET_INFO* collation;
-  Ndb_rewrite_context *rewrite_stack;
-};
-
 typedef enum ndb_query_state_bits {
   NDB_QUERY_NORMAL = 0,
   NDB_QUERY_MULTI_READ_RANGE = 1
@@ -721,27 +310,6 @@ bool uses_blob_value(bool all_fields);
   void no_uncommitted_rows_init(THD *);
   void no_uncommitted_rows_reset(THD *);
 
-  /*
-    Condition pushdown
-  */
-  void cond_clear();
-  bool serialize_cond(const COND *cond, Ndb_cond_stack *ndb_cond);
-  int build_scan_filter_predicate(Ndb_cond* &cond, 
-                                  NdbScanFilter* filter,
-                                  bool negated= false);
-  int build_scan_filter_group(Ndb_cond* &cond, 
-                              NdbScanFilter* filter);
-  int build_scan_filter(Ndb_cond* &cond, NdbScanFilter* filter);
-  int generate_scan_filter(Ndb_cond_stack* cond_stack, 
-                           NdbScanOperation* op);
-  int generate_scan_filter_from_cond(Ndb_cond_stack* cond_stack, 
-				     NdbScanFilter& filter);
-  int generate_scan_filter_from_key(NdbScanOperation* op,
-                                   const KEY* key_info, 
-                                   const byte *key, 
-                                   uint key_len,
-                                   byte *buf);
-
   friend int execute_commit(ha_ndbcluster*, NdbTransaction*);
   friend int execute_no_commit(ha_ndbcluster*, NdbTransaction*, bool);
   friend int execute_no_commit_ie(ha_ndbcluster*, NdbTransaction*, bool);
@@ -791,7 +359,7 @@ bool uses_blob_value(bool all_fields);
   bool m_transaction_on;
   void release_completed_operations(NdbTransaction*, bool);
 
-  Ndb_cond_stack *m_cond_stack;
+  ha_ndbcluster_cond *m_cond;
   bool m_disable_multi_read;
   byte *m_multi_range_result_ptr;
   KEY_MULTI_RANGE *m_multi_ranges;
diff --git a/sql/ha_ndbcluster_cond.cc b/sql/ha_ndbcluster_cond.cc
new file mode 100644
index 00000000000..a284bc13308
--- /dev/null
+++ b/sql/ha_ndbcluster_cond.cc
@@ -0,0 +1,1427 @@
+/* Copyright (C) 2000-2003 MySQL AB
+
+  This program is free software; you can redistribute it and/or modify
+  it under the terms of the GNU General Public License as published by
+  the Free Software Foundation; version 2 of the License.
+
+  This program is distributed in the hope that it will be useful,
+  but WITHOUT ANY WARRANTY; without even the implied warranty of
+  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+  GNU General Public License for more details.
+
+  You should have received a copy of the GNU General Public License
+  along with this program; if not, write to the Free Software
+  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+*/
+
+/*
+  This file defines the NDB Cluster handler: the interface between MySQL and
+  NDB Cluster
+*/
+
+#ifdef USE_PRAGMA_IMPLEMENTATION
+#pragma implementation				// gcc: Class implementation
+#endif
+
+#include "mysql_priv.h"
+
+#ifdef HAVE_NDBCLUSTER_DB
+#include <ndbapi/NdbApi.hpp>
+#include "ha_ndbcluster_cond.h"
+
+// Typedefs for long names 
+typedef NdbDictionary::Column NDBCOL;
+typedef NdbDictionary::Table NDBTAB;
+
+/*
+  Serialize the item tree into a linked list represented by Ndb_cond
+  for fast generation of NbdScanFilter. Adds information such as
+  position of fields that is not directly available in the Item tree.
+  Also checks if condition is supported.
+*/
+void ndb_serialize_cond(const Item *item, void *arg)
+{
+  Ndb_cond_traverse_context *context= (Ndb_cond_traverse_context *) arg;
+  DBUG_ENTER("ndb_serialize_cond");  
+
+  // Check if we are skipping arguments to a function to be evaluated
+  if (context->skip)
+  {
+    DBUG_PRINT("info", ("Skiping argument %d", context->skip));
+    context->skip--;
+    switch (item->type()) {
+    case Item::FUNC_ITEM:
+    {
+      Item_func *func_item= (Item_func *) item;
+      context->skip+= func_item->argument_count();
+      break;
+    }
+    case Item::INT_ITEM:
+    case Item::REAL_ITEM:
+    case Item::STRING_ITEM:
+    case Item::VARBIN_ITEM:
+    case Item::DECIMAL_ITEM:
+      break;
+    default:
+      context->supported= FALSE;
+      break;
+    }
+    
+    DBUG_VOID_RETURN;
+  }
+  
+  if (context->supported)
+  {
+    Ndb_rewrite_context *rewrite_context2= context->rewrite_stack;
+    const Item_func *rewrite_func_item;
+    // Check if we are rewriting some unsupported function call
+    if (rewrite_context2 &&
+        (rewrite_func_item= rewrite_context2->func_item) &&
+        rewrite_context2->count++ == 0)
+    {
+      switch (rewrite_func_item->functype()) {
+      case Item_func::BETWEEN:
+        /*
+          Rewrite 
+          <field>|<const> BETWEEN <const1>|<field1> AND <const2>|<field2>
+          to <field>|<const> > <const1>|<field1> AND 
+          <field>|<const> < <const2>|<field2>
+          or actually in prefix format
+          BEGIN(AND) GT(<field>|<const>, <const1>|<field1>), 
+          LT(<field>|<const>, <const2>|<field2>), END()
+        */
+      case Item_func::IN_FUNC:
+      {
+        /*
+          Rewrite <field>|<const> IN(<const1>|<field1>, <const2>|<field2>,..)
+          to <field>|<const> = <const1>|<field1> OR 
+          <field> = <const2>|<field2> ...
+          or actually in prefix format
+          BEGIN(OR) EQ(<field>|<const>, <const1><field1>), 
+          EQ(<field>|<const>, <const2>|<field2>), ... END()
+          Each part of the disjunction is added for each call
+          to ndb_serialize_cond and end of rewrite statement 
+          is wrapped in end of ndb_serialize_cond
+        */
+        if (context->expecting(item->type()))
+        {
+          // This is the <field>|<const> item, save it in the rewrite context
+          rewrite_context2->left_hand_item= item;
+          if (item->type() == Item::FUNC_ITEM)
+          {
+            Item_func *func_item= (Item_func *) item;
+            if (func_item->functype() == Item_func::UNKNOWN_FUNC &&
+                func_item->const_item())
+            {
+              // Skip any arguments since we will evaluate function instead
+              DBUG_PRINT("info", ("Skip until end of arguments marker"));
+              context->skip= func_item->argument_count();
+            }
+            else
+            {
+              DBUG_PRINT("info", ("Found unsupported functional expression in BETWEEN|IN"));
+              context->supported= FALSE;
+              DBUG_VOID_RETURN;
+              
+            }
+          }
+        }
+        else
+        {
+          // Non-supported BETWEEN|IN expression
+          DBUG_PRINT("info", ("Found unexpected item of type %u in BETWEEN|IN",
+                              item->type()));
+          context->supported= FALSE;
+          DBUG_VOID_RETURN;
+        }
+        break;
+      }
+      default:
+        context->supported= FALSE;
+        break;
+      }
+      DBUG_VOID_RETURN;
+    }
+    else
+    {
+      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;
+    // Check if we are rewriting some unsupported function call
+      if (context->rewrite_stack)
+      {
+        Ndb_rewrite_context *rewrite_context= context->rewrite_stack;
+        const Item_func *func_item= rewrite_context->func_item;
+        switch (func_item->functype()) {
+        case Item_func::BETWEEN:
+        {
+          /*
+            Rewrite 
+            <field>|<const> BETWEEN <const1>|<field1> AND <const2>|<field2>
+            to <field>|<const> > <const1>|<field1> AND 
+            <field>|<const> < <const2>|<field2>
+            or actually in prefix format
+            BEGIN(AND) GT(<field>|<const>, <const1>|<field1>), 
+            LT(<field>|<const>, <const2>|<field2>), END()
+          */
+          if (rewrite_context->count == 2)
+          {
+            // Lower limit of BETWEEN
+            DBUG_PRINT("info", ("GE_FUNC"));      
+            curr_cond->ndb_item= new Ndb_item(Item_func::GE_FUNC, 2);
+          }
+          else if (rewrite_context->count == 3)
+          {
+            // Upper limit of BETWEEN
+            DBUG_PRINT("info", ("LE_FUNC"));      
+            curr_cond->ndb_item= new Ndb_item(Item_func::LE_FUNC, 2);
+          }
+          else
+          {
+            // Illegal BETWEEN expression
+            DBUG_PRINT("info", ("Illegal BETWEEN expression"));
+            context->supported= FALSE;
+            DBUG_VOID_RETURN;
+          }
+          break;
+        }
+        case Item_func::IN_FUNC:
+        {
+          /*
+            Rewrite <field>|<const> IN(<const1>|<field1>, <const2>|<field2>,..)
+            to <field>|<const> = <const1>|<field1> OR 
+            <field> = <const2>|<field2> ...
+            or actually in prefix format
+            BEGIN(OR) EQ(<field>|<const>, <const1><field1>), 
+            EQ(<field>|<const>, <const2>|<field2>), ... END()
+            Each part of the disjunction is added for each call
+            to ndb_serialize_cond and end of rewrite statement 
+            is wrapped in end of ndb_serialize_cond
+          */
+          DBUG_PRINT("info", ("EQ_FUNC"));      
+          curr_cond->ndb_item= new Ndb_item(Item_func::EQ_FUNC, 2);
+          break;
+        }
+        default:
+          context->supported= FALSE;
+        }
+        // Handle left hand <field>|<const>
+        context->rewrite_stack= NULL; // Disable rewrite mode
+        context->expect_only(Item::FIELD_ITEM);
+        context->expect_field_result(STRING_RESULT);
+        context->expect_field_result(REAL_RESULT);
+        context->expect_field_result(INT_RESULT);
+        context->expect_field_result(DECIMAL_RESULT);
+        context->expect(Item::INT_ITEM);
+        context->expect(Item::STRING_ITEM);
+        context->expect(Item::VARBIN_ITEM);
+        context->expect(Item::FUNC_ITEM);
+        ndb_serialize_cond(rewrite_context->left_hand_item, arg);
+        context->skip= 0; // Any FUNC_ITEM expression has already been parsed
+        context->rewrite_stack= rewrite_context; // Enable rewrite mode
+        if (!context->supported)
+          DBUG_VOID_RETURN;
+
+        prev_cond= context->cond_ptr;
+        curr_cond= context->cond_ptr= new Ndb_cond();
+        prev_cond->next= curr_cond;
+      }
+      
+      // Check for end of AND/OR expression
+      if (!item)
+      {
+        // End marker for condition group
+        DBUG_PRINT("info", ("End of condition group"));
+        curr_cond->ndb_item= new Ndb_item(NDB_END_COND);
+      }
+      else
+      {
+        switch (item->type()) {
+        case Item::FIELD_ITEM:
+        {
+          Item_field *field_item= (Item_field *) item;
+          Field *field= field_item->field;
+          enum_field_types type= field->type();
+          /*
+            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->s == field->table->s)
+          {       
+            const NDBTAB *tab= (const NDBTAB *) context->ndb_table;
+            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 ||
+                   type == MYSQL_TYPE_DATE || 
+                   type == MYSQL_TYPE_YEAR ||
+                   type == MYSQL_TYPE_DATETIME)
+                  ? (context->expecting_field_result(STRING_RESULT) ||
+                     context->expecting_field_result(INT_RESULT))
+                  : true)) &&
+                // Bit fields no yet supported in scan filter
+                type != MYSQL_TYPE_BIT &&
+                // No BLOB support in scan filter
+                type != MYSQL_TYPE_TINY_BLOB &&
+                type != MYSQL_TYPE_MEDIUM_BLOB &&
+                type != MYSQL_TYPE_LONG_BLOB &&
+                type != MYSQL_TYPE_BLOB)
+            {
+              const NDBCOL *col= tab->getColumn(field->field_name);
+              DBUG_ASSERT(col);
+              curr_cond->ndb_item= new Ndb_item(field, col->getColumnNo());
+              context->dont_expect(Item::FIELD_ITEM);
+              context->expect_no_field_result();
+              if (! context->expecting_nothing())
+              {
+                // We have not seen second argument yet
+                if (type == MYSQL_TYPE_TIME ||
+                    type == MYSQL_TYPE_DATE || 
+                    type == MYSQL_TYPE_YEAR ||
+                    type == MYSQL_TYPE_DATETIME)
+                {
+                  context->expect_only(Item::STRING_ITEM);
+                  context->expect(Item::INT_ITEM);
+                }
+                else
+                  switch (field->result_type()) {
+                  case STRING_RESULT:
+                    // Expect char string or binary string
+                    context->expect_only(Item::STRING_ITEM);
+                    context->expect(Item::VARBIN_ITEM);
+                    context->expect_collation(field_item->collation.collation);
+                    break;
+                  case REAL_RESULT:
+                    context->expect_only(Item::REAL_ITEM);
+                    context->expect(Item::DECIMAL_ITEM);
+                    context->expect(Item::INT_ITEM);
+                    break;
+                  case INT_RESULT:
+                    context->expect_only(Item::INT_ITEM);
+                    context->expect(Item::VARBIN_ITEM);
+                    break;
+                  case DECIMAL_RESULT:
+                    context->expect_only(Item::DECIMAL_ITEM);
+                    context->expect(Item::REAL_ITEM);
+                    context->expect(Item::INT_ITEM);
+                    break;
+                  default:
+                    break;
+                  }    
+              }
+              else
+              {
+                // Expect another logical expression
+                context->expect_only(Item::FUNC_ITEM);
+                context->expect(Item::COND_ITEM);
+                // Check that field and string constant collations are the same
+                if ((field->result_type() == STRING_RESULT) &&
+                    !context->expecting_collation(item->collation.collation)
+                    && type != MYSQL_TYPE_TIME
+                    && type != MYSQL_TYPE_DATE
+                    && type != MYSQL_TYPE_YEAR
+                    && type != MYSQL_TYPE_DATETIME)
+                {
+                  DBUG_PRINT("info", ("Found non-matching collation %s",  
+                                      item->collation.collation->name)); 
+                  context->supported= FALSE;                
+                }
+              }
+              break;
+            }
+            else
+            {
+              DBUG_PRINT("info", ("Was not expecting field of type %u(%u)",
+                                  field->result_type(), type));
+              context->supported= FALSE;
+            }
+          }
+          else
+          {
+            DBUG_PRINT("info", ("Was not expecting field from table %s(%s)",
+                                context->table->s->table_name, 
+                                field->table->s->table_name));
+            context->supported= FALSE;
+          }
+          break;
+        }
+        case Item::FUNC_ITEM:
+        {
+          Item_func *func_item= (Item_func *) item;
+          // Check that we expect a function or functional expression here
+          if (context->expecting(Item::FUNC_ITEM) || 
+              func_item->functype() == Item_func::UNKNOWN_FUNC)
+            context->expect_nothing();
+          else
+          {
+            // Did not expect function here
+            context->supported= FALSE;
+            break;
+          }
+          
+          switch (func_item->functype()) {
+          case Item_func::EQ_FUNC:
+          {
+            DBUG_PRINT("info", ("EQ_FUNC"));      
+            curr_cond->ndb_item= new Ndb_item(func_item->functype(), 
+                                              func_item);      
+            context->expect(Item::STRING_ITEM);
+            context->expect(Item::INT_ITEM);
+            context->expect(Item::REAL_ITEM);
+            context->expect(Item::DECIMAL_ITEM);
+            context->expect(Item::VARBIN_ITEM);
+            context->expect(Item::FIELD_ITEM);
+            context->expect_field_result(STRING_RESULT);
+            context->expect_field_result(REAL_RESULT);
+            context->expect_field_result(INT_RESULT);
+            context->expect_field_result(DECIMAL_RESULT);
+            break;
+          }
+          case Item_func::NE_FUNC:
+          {
+            DBUG_PRINT("info", ("NE_FUNC"));      
+            curr_cond->ndb_item= new Ndb_item(func_item->functype(),
+                                              func_item);      
+            context->expect(Item::STRING_ITEM);
+            context->expect(Item::INT_ITEM);
+            context->expect(Item::REAL_ITEM);
+            context->expect(Item::DECIMAL_ITEM);
+            context->expect(Item::VARBIN_ITEM);
+            context->expect(Item::FIELD_ITEM);
+            context->expect_field_result(STRING_RESULT);
+            context->expect_field_result(REAL_RESULT);
+            context->expect_field_result(INT_RESULT);
+            context->expect_field_result(DECIMAL_RESULT);
+            break;
+          }
+          case Item_func::LT_FUNC:
+          {
+            DBUG_PRINT("info", ("LT_FUNC"));      
+            curr_cond->ndb_item= new Ndb_item(func_item->functype(),
+                                              func_item);      
+            context->expect(Item::STRING_ITEM);
+            context->expect(Item::INT_ITEM);
+            context->expect(Item::REAL_ITEM);
+            context->expect(Item::DECIMAL_ITEM);
+            context->expect(Item::VARBIN_ITEM);
+            context->expect(Item::FIELD_ITEM);
+            context->expect_field_result(STRING_RESULT);
+            context->expect_field_result(REAL_RESULT);
+            context->expect_field_result(INT_RESULT);
+            context->expect_field_result(DECIMAL_RESULT);
+            break;
+          }
+          case Item_func::LE_FUNC:
+          {
+            DBUG_PRINT("info", ("LE_FUNC"));      
+            curr_cond->ndb_item= new Ndb_item(func_item->functype(),
+                                              func_item);      
+            context->expect(Item::STRING_ITEM);
+            context->expect(Item::INT_ITEM);
+            context->expect(Item::REAL_ITEM);
+            context->expect(Item::DECIMAL_ITEM);
+            context->expect(Item::VARBIN_ITEM);
+            context->expect(Item::FIELD_ITEM);
+            context->expect_field_result(STRING_RESULT);
+            context->expect_field_result(REAL_RESULT);
+            context->expect_field_result(INT_RESULT);
+            context->expect_field_result(DECIMAL_RESULT);
+            break;
+          }
+          case Item_func::GE_FUNC:
+          {
+            DBUG_PRINT("info", ("GE_FUNC"));      
+            curr_cond->ndb_item= new Ndb_item(func_item->functype(),
+                                              func_item);      
+            context->expect(Item::STRING_ITEM);
+            context->expect(Item::INT_ITEM);
+            context->expect(Item::REAL_ITEM);
+            context->expect(Item::DECIMAL_ITEM);
+            context->expect(Item::VARBIN_ITEM);
+            context->expect(Item::FIELD_ITEM);
+            context->expect_field_result(STRING_RESULT);
+            context->expect_field_result(REAL_RESULT);
+            context->expect_field_result(INT_RESULT);
+            context->expect_field_result(DECIMAL_RESULT);
+            break;
+          }
+          case Item_func::GT_FUNC:
+          {
+            DBUG_PRINT("info", ("GT_FUNC"));      
+            curr_cond->ndb_item= new Ndb_item(func_item->functype(),
+                                              func_item);      
+            context->expect(Item::STRING_ITEM);
+            context->expect(Item::REAL_ITEM);
+            context->expect(Item::DECIMAL_ITEM);
+            context->expect(Item::INT_ITEM);
+            context->expect(Item::VARBIN_ITEM);
+            context->expect(Item::FIELD_ITEM);
+            context->expect_field_result(STRING_RESULT);
+            context->expect_field_result(REAL_RESULT);
+            context->expect_field_result(INT_RESULT);
+            context->expect_field_result(DECIMAL_RESULT);
+            break;
+          }
+          case Item_func::LIKE_FUNC:
+          {
+            DBUG_PRINT("info", ("LIKE_FUNC"));      
+            curr_cond->ndb_item= new Ndb_item(func_item->functype(),
+                                              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;
+          }
+          case Item_func::ISNULL_FUNC:
+          {
+            DBUG_PRINT("info", ("ISNULL_FUNC"));      
+            curr_cond->ndb_item= new Ndb_item(func_item->functype(),
+                                              func_item);      
+            context->expect(Item::FIELD_ITEM);
+            context->expect_field_result(STRING_RESULT);
+            context->expect_field_result(REAL_RESULT);
+            context->expect_field_result(INT_RESULT);
+            context->expect_field_result(DECIMAL_RESULT);
+            break;
+          }
+          case Item_func::ISNOTNULL_FUNC:
+          {
+            DBUG_PRINT("info", ("ISNOTNULL_FUNC"));      
+            curr_cond->ndb_item= new Ndb_item(func_item->functype(),
+                                              func_item);     
+            context->expect(Item::FIELD_ITEM);
+            context->expect_field_result(STRING_RESULT);
+            context->expect_field_result(REAL_RESULT);
+            context->expect_field_result(INT_RESULT);
+            context->expect_field_result(DECIMAL_RESULT);
+            break;
+          }
+          case Item_func::NOT_FUNC:
+          {
+            DBUG_PRINT("info", ("NOT_FUNC"));      
+            curr_cond->ndb_item= new Ndb_item(func_item->functype(),
+                                              func_item);     
+            context->expect(Item::FUNC_ITEM);
+            context->expect(Item::COND_ITEM);
+            break;
+          }
+          case Item_func::BETWEEN:
+          {
+            DBUG_PRINT("info", ("BETWEEN, rewriting using AND"));
+            Item_func_between *between_func= (Item_func_between *) func_item;
+            Ndb_rewrite_context *rewrite_context= 
+              new Ndb_rewrite_context(func_item);
+            rewrite_context->next= context->rewrite_stack;
+            context->rewrite_stack= rewrite_context;
+            if (between_func->negated)
+            {
+              DBUG_PRINT("info", ("NOT_FUNC"));
+              curr_cond->ndb_item= new Ndb_item(Item_func::NOT_FUNC, 1);
+              prev_cond= curr_cond;
+              curr_cond= context->cond_ptr= new Ndb_cond();
+              curr_cond->prev= prev_cond;
+              prev_cond->next= curr_cond;
+            }
+            DBUG_PRINT("info", ("COND_AND_FUNC"));
+            curr_cond->ndb_item= 
+              new Ndb_item(Item_func::COND_AND_FUNC, 
+                           func_item->argument_count() - 1);
+            context->expect_only(Item::FIELD_ITEM);
+            context->expect(Item::INT_ITEM);
+            context->expect(Item::STRING_ITEM);
+            context->expect(Item::VARBIN_ITEM);
+            context->expect(Item::FUNC_ITEM);
+            break;
+          }
+          case Item_func::IN_FUNC:
+          {
+            DBUG_PRINT("info", ("IN_FUNC, rewriting using OR"));
+            Item_func_in *in_func= (Item_func_in *) func_item;
+            Ndb_rewrite_context *rewrite_context= 
+              new Ndb_rewrite_context(func_item);
+            rewrite_context->next= context->rewrite_stack;
+            context->rewrite_stack= rewrite_context;
+            if (in_func->negated)
+            {
+              DBUG_PRINT("info", ("NOT_FUNC"));
+              curr_cond->ndb_item= new Ndb_item(Item_func::NOT_FUNC, 1);
+              prev_cond= curr_cond;
+              curr_cond= context->cond_ptr= new Ndb_cond();
+              curr_cond->prev= prev_cond;
+              prev_cond->next= curr_cond;
+            }
+            DBUG_PRINT("info", ("COND_OR_FUNC"));
+            curr_cond->ndb_item= new Ndb_item(Item_func::COND_OR_FUNC, 
+                                              func_item->argument_count() - 1);
+            context->expect_only(Item::FIELD_ITEM);
+            context->expect(Item::INT_ITEM);
+            context->expect(Item::STRING_ITEM);
+            context->expect(Item::VARBIN_ITEM);
+            context->expect(Item::FUNC_ITEM);
+            break;
+          }
+          case Item_func::UNKNOWN_FUNC:
+          {
+            DBUG_PRINT("info", ("UNKNOWN_FUNC %s", 
+                                func_item->const_item()?"const":""));  
+            DBUG_PRINT("info", ("result type %d", func_item->result_type()));
+            if (func_item->const_item())
+            {
+              switch (func_item->result_type()) {
+              case STRING_RESULT:
+              {
+                NDB_ITEM_QUALIFICATION q;
+                q.value_type= Item::STRING_ITEM;
+                curr_cond->ndb_item= new Ndb_item(NDB_VALUE, q, item); 
+                if (! context->expecting_no_field_result())
+                {
+                  // We have not seen the field argument yet
+                  context->expect_only(Item::FIELD_ITEM);
+                  context->expect_only_field_result(STRING_RESULT);
+                  context->expect_collation(func_item->collation.collation);
+                }
+                else
+                {
+                  // Expect another logical expression
+                  context->expect_only(Item::FUNC_ITEM);
+                  context->expect(Item::COND_ITEM);
+                  // Check that string result have correct collation
+                  if (!context->expecting_collation(item->collation.collation))
+                  {
+                    DBUG_PRINT("info", ("Found non-matching collation %s",  
+                                        item->collation.collation->name));
+                    context->supported= FALSE;
+                  }
+                }
+                // Skip any arguments since we will evaluate function instead
+                DBUG_PRINT("info", ("Skip until end of arguments marker"));
+                context->skip= func_item->argument_count();
+                break;
+              }
+              case REAL_RESULT:
+              {
+                NDB_ITEM_QUALIFICATION q;
+                q.value_type= Item::REAL_ITEM;
+                curr_cond->ndb_item= new Ndb_item(NDB_VALUE, q, item);
+                if (! context->expecting_no_field_result()) 
+                {
+                  // We have not seen the field argument yet
+                  context->expect_only(Item::FIELD_ITEM);
+                  context->expect_only_field_result(REAL_RESULT);
+                }
+                else
+                {
+                  // Expect another logical expression
+                  context->expect_only(Item::FUNC_ITEM);
+                  context->expect(Item::COND_ITEM);
+                }
+                
+                // Skip any arguments since we will evaluate function instead
+                DBUG_PRINT("info", ("Skip until end of arguments marker"));
+                context->skip= func_item->argument_count();
+                break;
+              }
+              case INT_RESULT:
+              {
+                NDB_ITEM_QUALIFICATION q;
+                q.value_type= Item::INT_ITEM;
+                curr_cond->ndb_item= new Ndb_item(NDB_VALUE, q, item);
+                if (! context->expecting_no_field_result()) 
+                {
+                  // We have not seen the field argument yet
+                  context->expect_only(Item::FIELD_ITEM);
+                  context->expect_only_field_result(INT_RESULT);
+                }
+                else
+                {
+                  // Expect another logical expression
+                  context->expect_only(Item::FUNC_ITEM);
+                  context->expect(Item::COND_ITEM);
+                }
+                
+                // Skip any arguments since we will evaluate function instead
+                DBUG_PRINT("info", ("Skip until end of arguments marker"));
+                context->skip= func_item->argument_count();
+                break;
+              }
+              case DECIMAL_RESULT:
+              {
+                NDB_ITEM_QUALIFICATION q;
+                q.value_type= Item::DECIMAL_ITEM;
+                curr_cond->ndb_item= new Ndb_item(NDB_VALUE, q, item);
+                if (! context->expecting_no_field_result()) 
+                {
+                  // We have not seen the field argument yet
+                  context->expect_only(Item::FIELD_ITEM);
+                  context->expect_only_field_result(DECIMAL_RESULT);
+                }
+                else
+                {
+                  // Expect another logical expression
+                  context->expect_only(Item::FUNC_ITEM);
+                  context->expect(Item::COND_ITEM);
+                }
+                // Skip any arguments since we will evaluate function instead
+                DBUG_PRINT("info", ("Skip until end of arguments marker"));
+                context->skip= func_item->argument_count();
+                break;
+              }
+              default:
+                break;
+              }
+            }
+            else
+              // Function does not return constant expression
+              context->supported= FALSE;
+            break;
+          }
+          default:
+          {
+            DBUG_PRINT("info", ("Found func_item of type %d", 
+                                func_item->functype()));
+            context->supported= FALSE;
+          }
+          }
+          break;
+        }
+        case Item::STRING_ITEM:
+          DBUG_PRINT("info", ("STRING_ITEM")); 
+          if (context->expecting(Item::STRING_ITEM)) 
+          {
+#ifndef DBUG_OFF
+            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", ("value \"%s\"", 
+                                string_item->val_str(&str)->ptr()));
+#endif
+            NDB_ITEM_QUALIFICATION q;
+            q.value_type= Item::STRING_ITEM;
+            curr_cond->ndb_item= new Ndb_item(NDB_VALUE, q, item);      
+            if (! context->expecting_no_field_result())
+            {
+              // We have not seen the field argument yet
+              context->expect_only(Item::FIELD_ITEM);
+              context->expect_only_field_result(STRING_RESULT);
+              context->expect_collation(item->collation.collation);
+            }
+            else 
+            {
+              // Expect another logical expression
+              context->expect_only(Item::FUNC_ITEM);
+              context->expect(Item::COND_ITEM);
+              // Check that we are comparing with a field with same collation
+              if (!context->expecting_collation(item->collation.collation))
+              {
+                DBUG_PRINT("info", ("Found non-matching collation %s",  
+                                    item->collation.collation->name));
+                context->supported= FALSE;
+              }
+            }
+          }
+          else
+            context->supported= FALSE;
+          break;
+        case Item::INT_ITEM:
+          DBUG_PRINT("info", ("INT_ITEM"));
+          if (context->expecting(Item::INT_ITEM)) 
+          {
+            DBUG_PRINT("info", ("value %ld",
+                                (long) ((Item_int*) item)->value));
+            NDB_ITEM_QUALIFICATION q;
+            q.value_type= Item::INT_ITEM;
+            curr_cond->ndb_item= new Ndb_item(NDB_VALUE, q, item);
+            if (! context->expecting_no_field_result()) 
+            {
+              // We have not seen the field argument yet
+              context->expect_only(Item::FIELD_ITEM);
+              context->expect_only_field_result(INT_RESULT);
+              context->expect_field_result(REAL_RESULT);
+              context->expect_field_result(DECIMAL_RESULT);
+            }
+            else
+            {
+              // Expect another logical expression
+              context->expect_only(Item::FUNC_ITEM);
+              context->expect(Item::COND_ITEM);
+            }
+          }
+          else
+            context->supported= FALSE;
+          break;
+        case Item::REAL_ITEM:
+          DBUG_PRINT("info", ("REAL_ITEM"));
+          if (context->expecting(Item::REAL_ITEM)) 
+          {
+            DBUG_PRINT("info", ("value %f", ((Item_float *) item)->value));
+            NDB_ITEM_QUALIFICATION q;
+            q.value_type= Item::REAL_ITEM;
+            curr_cond->ndb_item= new Ndb_item(NDB_VALUE, q, item);
+            if (! context->expecting_no_field_result()) 
+            {
+              // We have not seen the field argument yet
+              context->expect_only(Item::FIELD_ITEM);
+              context->expect_only_field_result(REAL_RESULT);
+            }
+            else
+            {
+              // Expect another logical expression
+              context->expect_only(Item::FUNC_ITEM);
+              context->expect(Item::COND_ITEM);
+            }
+          }
+          else
+            context->supported= FALSE;
+          break;
+        case Item::VARBIN_ITEM:
+          DBUG_PRINT("info", ("VARBIN_ITEM"));
+          if (context->expecting(Item::VARBIN_ITEM)) 
+          {
+            NDB_ITEM_QUALIFICATION q;
+            q.value_type= Item::VARBIN_ITEM;
+            curr_cond->ndb_item= new Ndb_item(NDB_VALUE, q, item);      
+            if (! context->expecting_no_field_result())
+            {
+              // We have not seen the field argument yet
+              context->expect_only(Item::FIELD_ITEM);
+              context->expect_only_field_result(STRING_RESULT);
+            }
+            else
+            {
+              // Expect another logical expression
+              context->expect_only(Item::FUNC_ITEM);
+              context->expect(Item::COND_ITEM);
+            }
+          }
+          else
+            context->supported= FALSE;
+          break;
+        case Item::DECIMAL_ITEM:
+          DBUG_PRINT("info", ("DECIMAL_ITEM"));
+          if (context->expecting(Item::DECIMAL_ITEM)) 
+          {
+            DBUG_PRINT("info", ("value %f",
+                                ((Item_decimal*) item)->val_real()));
+            NDB_ITEM_QUALIFICATION q;
+            q.value_type= Item::DECIMAL_ITEM;
+            curr_cond->ndb_item= new Ndb_item(NDB_VALUE, q, item);
+            if (! context->expecting_no_field_result()) 
+            {
+              // We have not seen the field argument yet
+              context->expect_only(Item::FIELD_ITEM);
+              context->expect_only_field_result(REAL_RESULT);
+              context->expect_field_result(DECIMAL_RESULT);
+            }
+            else
+            {
+              // Expect another logical expression
+              context->expect_only(Item::FUNC_ITEM);
+              context->expect(Item::COND_ITEM);
+            }
+          }
+          else
+            context->supported= FALSE;
+          break;
+        case Item::COND_ITEM:
+        {
+          Item_cond *cond_item= (Item_cond *) item;
+          
+          if (context->expecting(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(),
+                                                cond_item);      
+              break;
+            case Item_func::COND_OR_FUNC:
+              DBUG_PRINT("info", ("COND_OR_FUNC"));
+              curr_cond->ndb_item= new Ndb_item(cond_item->functype(),
+                                                cond_item);      
+              break;
+            default:
+              DBUG_PRINT("info", ("COND_ITEM %d", cond_item->functype()));
+              context->supported= FALSE;
+              break;
+            }
+          }
+          else
+          {
+            /* Did not expect condition */
+            context->supported= FALSE;          
+          }
+          break;
+        }
+        default:
+        {
+          DBUG_PRINT("info", ("Found item of type %d", item->type()));
+          context->supported= FALSE;
+        }
+        }
+      }
+      if (context->supported && context->rewrite_stack)
+      {
+        Ndb_rewrite_context *rewrite_context= context->rewrite_stack;
+        if (rewrite_context->count == 
+            rewrite_context->func_item->argument_count())
+        {
+          // Rewrite is done, wrap an END() at the en
+          DBUG_PRINT("info", ("End of condition group"));
+          prev_cond= curr_cond;
+          curr_cond= context->cond_ptr= new Ndb_cond();
+          curr_cond->prev= prev_cond;
+          prev_cond->next= curr_cond;
+          curr_cond->ndb_item= new Ndb_item(NDB_END_COND);
+          // Pop rewrite stack
+          context->rewrite_stack=  rewrite_context->next;
+          rewrite_context->next= NULL;
+          delete(rewrite_context);
+        }
+      }
+    }
+  }
+ 
+  DBUG_VOID_RETURN;
+}
+
+/*
+  Push a condition
+ */
+const 
+COND* 
+ha_ndbcluster_cond::cond_push(const COND *cond, 
+                              TABLE *table, NDBTAB *ndb_table)
+{ 
+  DBUG_ENTER("cond_push");
+  Ndb_cond_stack *ndb_cond = new Ndb_cond_stack();
+  if (ndb_cond == NULL)
+  {
+    my_errno= HA_ERR_OUT_OF_MEM;
+    DBUG_RETURN(NULL);
+  }
+  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, table, ndb_table))
+  {
+    DBUG_RETURN(NULL);
+  }
+  else
+  {
+    cond_pop();
+  }
+  DBUG_RETURN(cond); 
+}
+
+/*
+  Pop the top condition from the condition stack
+*/
+void 
+ha_ndbcluster_cond::cond_pop() 
+{ 
+  Ndb_cond_stack *ndb_cond_stack= m_cond_stack;  
+  if (ndb_cond_stack)
+  {
+    m_cond_stack= ndb_cond_stack->next;
+    ndb_cond_stack->next= NULL;
+    delete ndb_cond_stack;
+  }
+}
+
+/*
+  Clear the condition stack
+*/
+void
+ha_ndbcluster_cond::cond_clear()
+{
+  DBUG_ENTER("cond_clear");
+  while (m_cond_stack)
+    cond_pop();
+
+  DBUG_VOID_RETURN;
+}
+
+bool
+ha_ndbcluster_cond::serialize_cond(const COND *cond, Ndb_cond_stack *ndb_cond,
+                                   TABLE *table, NDBTAB *ndb_table)
+{
+  DBUG_ENTER("serialize_cond");
+  Item *item= (Item *) cond;
+  Ndb_cond_traverse_context context(table, (void *)ndb_table, 
+				    ndb_cond);
+  // Expect a logical expression
+  context.expect(Item::FUNC_ITEM);
+  context.expect(Item::COND_ITEM);
+  item->traverse_cond(&ndb_serialize_cond, (void *) &context, Item::PREFIX);
+  DBUG_PRINT("info", ("The pushed condition is %ssupported", (context.supported)?"":"not "));
+
+  DBUG_RETURN(context.supported);
+}
+
+int
+ha_ndbcluster_cond::build_scan_filter_predicate(Ndb_cond * &cond, 
+                                                NdbScanFilter *filter,
+                                                bool negated)
+{
+  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;
+    Ndb_item *b, *field, *value= NULL;
+    LINT_INIT(field);
+
+    switch (cond->ndb_item->argument_count()) {
+    case 1:
+      field= 
+        (a->type == NDB_FIELD)? a : NULL;
+      break;
+    case 2:
+      if (!cond->next->next)
+        break;
+      b= cond->next->next->ndb_item;
+      value= 
+        (a->type == NDB_VALUE)? a
+        : (b->type == NDB_VALUE)? b
+        : NULL;
+      field= 
+        (a->type == NDB_FIELD)? a
+        : (b->type == NDB_FIELD)? b
+        : NULL;
+      break;
+    default:
+      break;
+    }
+    switch ((negated) ? 
+            Ndb_item::negate(cond->ndb_item->qualification.function_type)
+            : cond->ndb_item->qualification.function_type) {
+    case NDB_EQ_FUNC:
+    {
+      if (!value || !field) break;
+      // Save value in right format for the field type
+      value->save_in_field(field);
+      DBUG_PRINT("info", ("Generating EQ filter"));
+      if (filter->cmp(NdbScanFilter::COND_EQ, 
+                      field->get_field_no(),
+                      field->get_val(),
+                      field->pack_length()) == -1)
+        DBUG_RETURN(1);
+      cond= cond->next->next->next;
+      DBUG_RETURN(0);
+    }
+    case NDB_NE_FUNC:
+    {
+      if (!value || !field) break;
+      // Save value in right format for the field type
+      value->save_in_field(field);
+      DBUG_PRINT("info", ("Generating NE filter"));
+      if (filter->cmp(NdbScanFilter::COND_NE, 
+                      field->get_field_no(),
+                      field->get_val(),
+                      field->pack_length()) == -1)
+        DBUG_RETURN(1);
+      cond= cond->next->next->next;
+      DBUG_RETURN(0);
+    }
+    case NDB_LT_FUNC:
+    {
+      if (!value || !field) break;
+      // Save value in right format for the field type
+      value->save_in_field(field);
+      if (a == field)
+      {
+        DBUG_PRINT("info", ("Generating LT filter")); 
+        if (filter->cmp(NdbScanFilter::COND_LT, 
+                        field->get_field_no(),
+                        field->get_val(),
+                        field->pack_length()) == -1)
+          DBUG_RETURN(1);
+      }
+      else
+      {
+        DBUG_PRINT("info", ("Generating GT filter")); 
+        if (filter->cmp(NdbScanFilter::COND_GT, 
+                        field->get_field_no(),
+                        field->get_val(),
+                        field->pack_length()) == -1)
+          DBUG_RETURN(1);
+      }
+      cond= cond->next->next->next;
+      DBUG_RETURN(0);
+    }
+    case NDB_LE_FUNC:
+    {
+      if (!value || !field) break;
+      // Save value in right format for the field type
+      value->save_in_field(field);
+      if (a == field)
+      {
+        DBUG_PRINT("info", ("Generating LE filter")); 
+        if (filter->cmp(NdbScanFilter::COND_LE, 
+                        field->get_field_no(),
+                        field->get_val(),
+                        field->pack_length()) == -1)
+          DBUG_RETURN(1);       
+      }
+      else
+      {
+        DBUG_PRINT("info", ("Generating GE filter")); 
+        if (filter->cmp(NdbScanFilter::COND_GE, 
+                        field->get_field_no(),
+                        field->get_val(),
+                        field->pack_length()) == -1)
+          DBUG_RETURN(1);
+      }
+      cond= cond->next->next->next;
+      DBUG_RETURN(0);
+    }
+    case NDB_GE_FUNC:
+    {
+      if (!value || !field) break;
+      // Save value in right format for the field type
+      value->save_in_field(field);
+      if (a == field)
+      {
+        DBUG_PRINT("info", ("Generating GE filter")); 
+        if (filter->cmp(NdbScanFilter::COND_GE, 
+                        field->get_field_no(),
+                        field->get_val(),
+                        field->pack_length()) == -1)
+          DBUG_RETURN(1);
+      }
+      else
+      {
+        DBUG_PRINT("info", ("Generating LE filter")); 
+        if (filter->cmp(NdbScanFilter::COND_LE, 
+                        field->get_field_no(),
+                        field->get_val(),
+                        field->pack_length()) == -1)
+          DBUG_RETURN(1);
+      }
+      cond= cond->next->next->next;
+      DBUG_RETURN(0);
+    }
+    case NDB_GT_FUNC:
+    {
+      if (!value || !field) break;
+      // Save value in right format for the field type
+      value->save_in_field(field);
+      if (a == field)
+      {
+        DBUG_PRINT("info", ("Generating GT filter"));
+        if (filter->cmp(NdbScanFilter::COND_GT, 
+                        field->get_field_no(),
+                        field->get_val(),
+                        field->pack_length()) == -1)
+          DBUG_RETURN(1);
+      }
+      else
+      {
+        DBUG_PRINT("info", ("Generating LT filter"));
+        if (filter->cmp(NdbScanFilter::COND_LT, 
+                        field->get_field_no(),
+                        field->get_val(),
+                        field->pack_length()) == -1)
+          DBUG_RETURN(1);
+      }
+      cond= cond->next->next->next;
+      DBUG_RETURN(0);
+    }
+    case NDB_LIKE_FUNC:
+    {
+      if (!value || !field) break;
+      if ((value->qualification.value_type != Item::STRING_ITEM) &&
+          (value->qualification.value_type != Item::VARBIN_ITEM))
+          break;
+      // Save value in right format for the field type
+      value->save_in_field(field);
+      DBUG_PRINT("info", ("Generating LIKE filter: like(%d,%s,%d)", 
+                          field->get_field_no(), value->get_val(), 
+                          value->pack_length()));
+      if (filter->cmp(NdbScanFilter::COND_LIKE, 
+                      field->get_field_no(),
+                      value->get_val(),
+                      value->pack_length()) == -1)
+        DBUG_RETURN(1);
+      cond= cond->next->next->next;
+      DBUG_RETURN(0);
+    }
+    case NDB_NOTLIKE_FUNC:
+    {
+      if (!value || !field) break;
+      if ((value->qualification.value_type != Item::STRING_ITEM) &&
+          (value->qualification.value_type != Item::VARBIN_ITEM))
+          break;
+      // Save value in right format for the field type
+      value->save_in_field(field);
+      DBUG_PRINT("info", ("Generating NOTLIKE filter: notlike(%d,%s,%d)", 
+                          field->get_field_no(), value->get_val(), 
+                          value->pack_length()));
+      if (filter->cmp(NdbScanFilter::COND_NOT_LIKE, 
+                      field->get_field_no(),
+                      value->get_val(),
+                      value->pack_length()) == -1)
+        DBUG_RETURN(1);
+      cond= cond->next->next->next;
+      DBUG_RETURN(0);
+    }
+    case NDB_ISNULL_FUNC:
+      if (!field)
+        break;
+      DBUG_PRINT("info", ("Generating ISNULL filter"));
+      if (filter->isnull(field->get_field_no()) == -1)
+        DBUG_RETURN(1);
+      cond= cond->next->next;
+      DBUG_RETURN(0);
+    case NDB_ISNOTNULL_FUNC:
+    {
+      if (!field)
+        break;
+      DBUG_PRINT("info", ("Generating ISNOTNULL filter"));
+      if (filter->isnotnull(field->get_field_no()) == -1)
+        DBUG_RETURN(1);         
+      cond= cond->next->next;
+      DBUG_RETURN(0);
+    }
+    default:
+      break;
+    }
+    break;
+  }
+  default:
+    break;
+  }
+  DBUG_PRINT("info", ("Found illegal condition"));
+  DBUG_RETURN(1);
+}
+
+
+int
+ha_ndbcluster_cond::build_scan_filter_group(Ndb_cond* &cond, 
+                                            NdbScanFilter *filter)
+{
+  uint level=0;
+  bool negated= FALSE;
+  DBUG_ENTER("build_scan_filter_group");
+
+  do
+  {
+    if (!cond)
+      DBUG_RETURN(1);
+    switch (cond->ndb_item->type) {
+    case NDB_FUNCTION:
+    {
+      switch (cond->ndb_item->qualification.function_type) {
+      case NDB_COND_AND_FUNC:
+      {
+        level++;
+        DBUG_PRINT("info", ("Generating %s group %u", (negated)?"NAND":"AND",
+                            level));
+        if ((negated) ? filter->begin(NdbScanFilter::NAND)
+            : filter->begin(NdbScanFilter::AND) == -1)
+          DBUG_RETURN(1);
+        negated= FALSE;
+        cond= cond->next;
+        break;
+      }
+      case NDB_COND_OR_FUNC:
+      {
+        level++;
+        DBUG_PRINT("info", ("Generating %s group %u", (negated)?"NOR":"OR",
+                            level));
+        if ((negated) ? filter->begin(NdbScanFilter::NOR)
+            : filter->begin(NdbScanFilter::OR) == -1)
+          DBUG_RETURN(1);
+        negated= FALSE;
+        cond= cond->next;
+        break;
+      }
+      case NDB_NOT_FUNC:
+      {
+        DBUG_PRINT("info", ("Generating negated query"));
+        cond= cond->next;
+        negated= TRUE;
+        break;
+      }
+      default:
+        if (build_scan_filter_predicate(cond, filter, negated))
+          DBUG_RETURN(1);
+        negated= FALSE;
+        break;
+      }
+      break;
+    }
+    case NDB_END_COND:
+      DBUG_PRINT("info", ("End of group %u", level));
+      level--;
+      if (cond) cond= cond->next;
+      if (filter->end() == -1)
+        DBUG_RETURN(1);
+      if (!negated)
+        break;
+      // else fall through (NOT END is an illegal condition)
+    default:
+    {
+      DBUG_PRINT("info", ("Illegal scan filter"));
+    }
+    }
+  }  while (level > 0 || negated);
+  
+  DBUG_RETURN(0);
+}
+
+
+int
+ha_ndbcluster_cond::build_scan_filter(Ndb_cond * &cond, NdbScanFilter *filter)
+{
+  bool simple_cond= TRUE;
+  DBUG_ENTER("build_scan_filter");  
+
+    switch (cond->ndb_item->type) {
+    case NDB_FUNCTION:
+      switch (cond->ndb_item->qualification.function_type) {
+      case NDB_COND_AND_FUNC:
+      case NDB_COND_OR_FUNC:
+        simple_cond= FALSE;
+        break;
+      default:
+        break;
+      }
+      break;
+    default:
+      break;
+    }
+  if (simple_cond && filter->begin() == -1)
+    DBUG_RETURN(1);
+  if (build_scan_filter_group(cond, filter))
+    DBUG_RETURN(1);
+  if (simple_cond && filter->end() == -1)
+    DBUG_RETURN(1);
+
+  DBUG_RETURN(0);
+}
+
+int
+ha_ndbcluster_cond::generate_scan_filter(NdbScanOperation *op)
+{
+  DBUG_ENTER("generate_scan_filter");
+
+  if (m_cond_stack)
+  {
+    NdbScanFilter filter(op);
+    
+    DBUG_RETURN(generate_scan_filter_from_cond(filter));
+  }
+  else
+  {  
+    DBUG_PRINT("info", ("Empty stack"));
+  }
+
+  DBUG_RETURN(0);
+}
+
+
+int
+ha_ndbcluster_cond::generate_scan_filter_from_cond(NdbScanFilter& filter)
+{
+  bool multiple_cond= FALSE;
+  DBUG_ENTER("generate_scan_filter_from_cond");
+
+  // Wrap an AND group around multiple conditions
+  if (m_cond_stack->next) 
+  {
+    multiple_cond= TRUE;
+    if (filter.begin() == -1)
+      DBUG_RETURN(1); 
+  }
+  for (Ndb_cond_stack *stack= m_cond_stack; 
+       (stack); 
+       stack= stack->next)
+  {
+    Ndb_cond *cond= stack->ndb_cond;
+    
+    if (build_scan_filter(cond, &filter))
+    {
+      DBUG_PRINT("info", ("build_scan_filter failed"));
+      DBUG_RETURN(1);
+    }
+  }
+  if (multiple_cond && filter.end() == -1)
+    DBUG_RETURN(1);
+
+  DBUG_RETURN(0);
+}
+
+
+int ha_ndbcluster_cond::generate_scan_filter_from_key(NdbScanOperation *op,
+                                                      const KEY* key_info, 
+                                                      const byte *key, 
+                                                      uint key_len,
+                                                      byte *buf)
+{
+  KEY_PART_INFO* key_part= key_info->key_part;
+  KEY_PART_INFO* end= key_part+key_info->key_parts;
+  NdbScanFilter filter(op);
+  int res;
+  DBUG_ENTER("generate_scan_filter_from_key");
+
+  filter.begin(NdbScanFilter::AND);
+  for (; key_part != end; key_part++) 
+  {
+    Field* field= key_part->field;
+    uint32 pack_len= field->pack_length();
+    const byte* ptr= key;
+    DBUG_PRINT("info", ("Filtering value for %s", field->field_name));
+    DBUG_DUMP("key", (char*)ptr, pack_len);
+    if (key_part->null_bit)
+    {
+      DBUG_PRINT("info", ("Generating ISNULL filter"));
+      if (filter.isnull(key_part->fieldnr-1) == -1)
+	DBUG_RETURN(1);
+    }
+    else
+    {
+      DBUG_PRINT("info", ("Generating EQ filter"));
+      if (filter.cmp(NdbScanFilter::COND_EQ, 
+		     key_part->fieldnr-1,
+		     ptr,
+		     pack_len) == -1)
+	DBUG_RETURN(1);
+    }
+    key += key_part->store_length;
+  }      
+  // Add any pushed condition
+  if (m_cond_stack &&
+      (res= generate_scan_filter_from_cond(filter)))
+    DBUG_RETURN(res);
+    
+  if (filter.end() == -1)
+    DBUG_RETURN(1);
+
+  DBUG_RETURN(0);
+}
+
+#endif /* HAVE_NDBCLUSTER_DB */
diff --git a/sql/ha_ndbcluster_cond.h b/sql/ha_ndbcluster_cond.h
new file mode 100644
index 00000000000..5afa817fade
--- /dev/null
+++ b/sql/ha_ndbcluster_cond.h
@@ -0,0 +1,469 @@
+/* Copyright (C) 2000-2007 MySQL AB
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; version 2 of the License.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
+
+/*
+  This file defines the data structures used by engine condition pushdown in
+  the NDB Cluster handler
+*/
+
+#ifdef USE_PRAGMA_INTERFACE
+#pragma interface                       /* gcc class implementation */
+#endif
+
+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_END_COND = 3 // End marker for condition group
+} NDB_ITEM_TYPE;
+
+typedef enum ndb_func_type {
+  NDB_EQ_FUNC = 0,
+  NDB_NE_FUNC = 1,
+  NDB_LT_FUNC = 2,
+  NDB_LE_FUNC = 3,
+  NDB_GT_FUNC = 4,
+  NDB_GE_FUNC = 5,
+  NDB_ISNULL_FUNC = 6,
+  NDB_ISNOTNULL_FUNC = 7,
+  NDB_LIKE_FUNC = 8,
+  NDB_NOTLIKE_FUNC = 9,
+  NDB_NOT_FUNC = 10,
+  NDB_UNKNOWN_FUNC = 11,
+  NDB_COND_AND_FUNC = 12,
+  NDB_COND_OR_FUNC = 13,
+  NDB_UNSUPPORTED_FUNC = 14
+} NDB_FUNC_TYPE;
+
+typedef union ndb_item_qualification {
+  Item::Type value_type; 
+  enum_field_types field_type; // Instead of Item::FIELD_ITEM
+  NDB_FUNC_TYPE function_type; // Instead of Item::FUNC_ITEM
+} NDB_ITEM_QUALIFICATION;
+
+typedef struct ndb_item_field_value {
+  Field* field;
+  int column_no;
+} NDB_ITEM_FIELD_VALUE;
+
+typedef union ndb_item_value {
+  const Item *item;
+  NDB_ITEM_FIELD_VALUE *field_value;
+  uint arg_count;
+} NDB_ITEM_VALUE;
+
+struct negated_function_mapping
+{
+  NDB_FUNC_TYPE pos_fun;
+  NDB_FUNC_TYPE neg_fun;
+};
+
+/*
+  Define what functions can be negated in condition pushdown.
+  Note, these HAVE to be in the same order as in definition enum
+*/
+static const negated_function_mapping neg_map[]= 
+{
+  {NDB_EQ_FUNC, NDB_NE_FUNC},
+  {NDB_NE_FUNC, NDB_EQ_FUNC},
+  {NDB_LT_FUNC, NDB_GE_FUNC},
+  {NDB_LE_FUNC, NDB_GT_FUNC},
+  {NDB_GT_FUNC, NDB_LE_FUNC},
+  {NDB_GE_FUNC, NDB_LT_FUNC},
+  {NDB_ISNULL_FUNC, NDB_ISNOTNULL_FUNC},
+  {NDB_ISNOTNULL_FUNC, NDB_ISNULL_FUNC},
+  {NDB_LIKE_FUNC, NDB_NOTLIKE_FUNC},
+  {NDB_NOTLIKE_FUNC, NDB_LIKE_FUNC},
+  {NDB_NOT_FUNC, NDB_UNSUPPORTED_FUNC},
+  {NDB_UNKNOWN_FUNC, NDB_UNSUPPORTED_FUNC},
+  {NDB_COND_AND_FUNC, NDB_UNSUPPORTED_FUNC},
+  {NDB_COND_OR_FUNC, NDB_UNSUPPORTED_FUNC},
+  {NDB_UNSUPPORTED_FUNC, NDB_UNSUPPORTED_FUNC}
+};
+  
+/*
+  This class is the construction element for serialization of Item tree 
+  in condition pushdown.
+  An instance of Ndb_Item represents a constant, table field reference,
+  unary or binary comparison predicate, and start/end of AND/OR.
+  Instances of Ndb_Item are stored in a linked list implemented by Ndb_cond
+  class.
+  The order of elements produced by Ndb_cond::next corresponds to
+  breadth-first traversal of the Item (i.e. expression) tree in prefix order.
+  AND and OR have arbitrary arity, so the end of AND/OR group is marked with  
+  Ndb_item with type == NDB_END_COND.
+  NOT items represent negated conditions and generate NAND/NOR groups.
+*/
+class Ndb_item : public Sql_alloc
+{
+public:
+  Ndb_item(NDB_ITEM_TYPE item_type) : type(item_type) {};
+  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):
+      value.item= item_value;
+      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):
+      value.item= item_value;
+      value.arg_count= ((Item_func *) item_value)->argument_count();
+      break;
+    case(NDB_END_COND):
+      break;
+    }
+  };
+  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(Item_func::Functype func_type, const Item *item_value) 
+    : type(NDB_FUNCTION)
+  {
+    qualification.function_type= item_func_to_ndb_func(func_type);
+    value.item= item_value;
+    value.arg_count= ((Item_func *) item_value)->argument_count();
+  };
+  Ndb_item(Item_func::Functype func_type, uint no_args) 
+    : type(NDB_FUNCTION)
+  {
+    qualification.function_type= item_func_to_ndb_func(func_type);
+    value.arg_count= no_args;
+  };
+  ~Ndb_item()
+  { 
+    if (type == NDB_FIELD)
+      {
+        delete value.field_value;
+        value.field_value= NULL;
+      }
+  };
+
+  uint32 pack_length() 
+  { 
+    switch(type) {
+    case(NDB_VALUE):
+      if(qualification.value_type == Item::STRING_ITEM)
+        return value.item->str_value.length();
+      break;
+    case(NDB_FIELD):
+      return value.field_value->field->pack_length(); 
+    default:
+      break;
+    }
+    
+    return 0;
+  };
+
+  Field * get_field() { return value.field_value->field; };
+
+  int get_field_no() { return value.field_value->column_no; };
+
+  int argument_count() 
+  { 
+    return value.arg_count;
+  };
+
+  const char* get_val() 
+  {  
+    switch(type) {
+    case(NDB_VALUE):
+      if(qualification.value_type == Item::STRING_ITEM)
+        return value.item->str_value.ptr();
+      break;
+    case(NDB_FIELD):
+      return value.field_value->field->ptr; 
+    default:
+      break;
+    }
+    
+    return NULL;
+  };
+
+  void save_in_field(Ndb_item *field_item)
+  {
+    Field *field = field_item->value.field_value->field;
+    const Item *item= value.item;
+
+    if (item && field)
+      ((Item *)item)->save_in_field(field, false);
+  };
+
+  static NDB_FUNC_TYPE item_func_to_ndb_func(Item_func::Functype fun)
+  {
+    switch (fun) {
+    case (Item_func::EQ_FUNC): { return NDB_EQ_FUNC; }
+    case (Item_func::NE_FUNC): { return NDB_NE_FUNC; }
+    case (Item_func::LT_FUNC): { return NDB_LT_FUNC; }
+    case (Item_func::LE_FUNC): { return NDB_LE_FUNC; }
+    case (Item_func::GT_FUNC): { return NDB_GT_FUNC; }
+    case (Item_func::GE_FUNC): { return NDB_GE_FUNC; }
+    case (Item_func::ISNULL_FUNC): { return NDB_ISNULL_FUNC; }
+    case (Item_func::ISNOTNULL_FUNC): { return NDB_ISNOTNULL_FUNC; }
+    case (Item_func::LIKE_FUNC): { return NDB_LIKE_FUNC; }
+    case (Item_func::NOT_FUNC): { return NDB_NOT_FUNC; }
+    case (Item_func::UNKNOWN_FUNC): { return NDB_UNKNOWN_FUNC; }
+    case (Item_func::COND_AND_FUNC): { return NDB_COND_AND_FUNC; }
+    case (Item_func::COND_OR_FUNC): { return NDB_COND_OR_FUNC; }
+    default: { return NDB_UNSUPPORTED_FUNC; }
+    }
+  };
+
+  static NDB_FUNC_TYPE negate(NDB_FUNC_TYPE fun)
+  {
+    uint i= (uint) fun;
+    DBUG_ASSERT(fun == neg_map[i].pos_fun);
+    return  neg_map[i].neg_fun;
+  };
+
+  NDB_ITEM_TYPE type;
+  NDB_ITEM_QUALIFICATION qualification;
+ private:
+  NDB_ITEM_VALUE value;
+};
+
+/*
+  This class implements a linked list used for storing a
+  serialization of the Item tree for condition pushdown.
+ */
+class Ndb_cond : public Sql_alloc
+{
+ 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;
+};
+
+/*
+  This class implements a stack for storing several conditions
+  for pushdown (represented as serialized Item trees using Ndb_cond).
+  The current implementation only pushes one condition, but is
+  prepared for handling several (C1 AND C2 ...) if the logic for 
+  pushing conditions is extended in sql_select.
+*/
+class Ndb_cond_stack : public Sql_alloc
+{
+ public:
+  Ndb_cond_stack() : ndb_cond(NULL), next(NULL) {};
+  ~Ndb_cond_stack() 
+  { 
+    if (ndb_cond) delete ndb_cond; 
+    ndb_cond= NULL; 
+    if (next) delete next;
+    next= NULL; 
+  };
+  Ndb_cond *ndb_cond;
+  Ndb_cond_stack *next;
+};
+
+class Ndb_rewrite_context : public Sql_alloc
+{
+public:
+  Ndb_rewrite_context(Item_func *func) 
+    : func_item(func), left_hand_item(NULL), count(0) {};
+  ~Ndb_rewrite_context()
+  {
+    if (next) delete next;
+  }
+  const Item_func *func_item;
+  const Item *left_hand_item;
+  uint count;
+  Ndb_rewrite_context *next;
+};
+
+/*
+  This class is used for storing the context when traversing
+  the Item tree. It stores a reference to the table the condition
+  is defined on, the serialized representation being generated, 
+  if the condition found is supported, and information what is
+  expected next in the tree inorder for the condition to be supported.
+*/
+class Ndb_cond_traverse_context : public Sql_alloc
+{
+ public:
+  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),
+    skip(0), collation(NULL), rewrite_stack(NULL)
+  {
+    // Allocate type checking bitmaps   
+    bitmap_init(&expect_mask, 0, 512, FALSE);
+    bitmap_init(&expect_field_type_mask, 0, 512, FALSE);
+    bitmap_init(&expect_field_result_mask, 0, 512, FALSE);
+
+    if (stack)
+      cond_ptr= stack->ndb_cond;
+  };
+  ~Ndb_cond_traverse_context()
+  {
+    bitmap_free(&expect_mask);
+    bitmap_free(&expect_field_type_mask);
+    bitmap_free(&expect_field_result_mask);
+    if (rewrite_stack) delete rewrite_stack;
+  }
+  void expect(Item::Type type)
+  {
+    bitmap_set_bit(&expect_mask, (uint) type);
+    if (type == Item::FIELD_ITEM) expect_all_field_types();
+  };
+  void dont_expect(Item::Type type)
+  {
+    bitmap_clear_bit(&expect_mask, (uint) type);
+  };
+  bool expecting(Item::Type type)
+  {
+    return bitmap_is_set(&expect_mask, (uint) type);
+  };
+  void expect_nothing()
+  {
+    bitmap_clear_all(&expect_mask);
+  };
+  bool expecting_nothing()
+  {
+    return bitmap_is_clear_all(&expect_mask);
+  }
+  void expect_only(Item::Type type)
+  {
+    expect_nothing();
+    expect(type);
+  };
+
+  void expect_field_type(enum_field_types type)
+  {
+    bitmap_set_bit(&expect_field_type_mask, (uint) type);
+  };
+  void expect_all_field_types()
+  {
+    bitmap_set_all(&expect_field_type_mask);
+  };
+  bool expecting_field_type(enum_field_types type)
+  {
+    return bitmap_is_set(&expect_field_type_mask, (uint) type);
+  };
+  void expect_no_field_type()
+  {
+    bitmap_clear_all(&expect_field_type_mask);
+  };
+  bool expecting_no_field_type()
+  {
+    return bitmap_is_clear_all(&expect_field_type_mask);
+  }
+  void expect_only_field_type(enum_field_types result)
+  {
+    expect_no_field_type();
+    expect_field_type(result);
+  };
+
+  void expect_field_result(Item_result result)
+  {
+    bitmap_set_bit(&expect_field_result_mask, (uint) result);
+  };
+  bool expecting_field_result(Item_result result)
+  {
+    return bitmap_is_set(&expect_field_result_mask, (uint) result);
+  };
+  void expect_no_field_result()
+  {
+    bitmap_clear_all(&expect_field_result_mask);
+  };
+  bool expecting_no_field_result()
+  {
+    return bitmap_is_clear_all(&expect_field_result_mask);
+  }
+  void expect_only_field_result(Item_result result)
+  {
+    expect_no_field_result();
+    expect_field_result(result);
+  };
+  void expect_collation(CHARSET_INFO* col)
+  {
+    collation= col;
+  };
+  bool expecting_collation(CHARSET_INFO* col)
+  {
+    bool matching= (!collation) ? true : (collation == col);
+    collation= NULL;
+
+    return matching;
+  };
+
+  TABLE* table;
+  void* ndb_table;
+  bool supported;
+  Ndb_cond_stack* stack_ptr;
+  Ndb_cond* cond_ptr;
+  MY_BITMAP expect_mask;
+  MY_BITMAP expect_field_type_mask;
+  MY_BITMAP expect_field_result_mask;
+  uint skip;
+  CHARSET_INFO* collation;
+  Ndb_rewrite_context *rewrite_stack;
+};
+
+class ha_ndbcluster;
+
+class ha_ndbcluster_cond
+{
+public:
+  ha_ndbcluster_cond() 
+  : m_cond_stack(NULL)
+  {}
+  ~ha_ndbcluster_cond() 
+  { if (m_cond_stack) delete m_cond_stack; }
+  const COND *cond_push(const COND *cond, 
+                        TABLE *table, NdbDictionary::Table *ndb_table);
+  void cond_pop();
+  void cond_clear();
+  int generate_scan_filter(NdbScanOperation* op);
+  int generate_scan_filter_from_cond(NdbScanFilter& filter);
+  int generate_scan_filter_from_key(NdbScanOperation* op,
+                                    const KEY* key_info, 
+                                    const byte *key, 
+                                    uint key_len,
+                                    byte *buf);
+private:
+  bool serialize_cond(const COND *cond, Ndb_cond_stack *ndb_cond,
+		      TABLE *table, NdbDictionary::Table *ndb_table);
+  int build_scan_filter_predicate(Ndb_cond* &cond, 
+                                  NdbScanFilter* filter,
+                                  bool negated= false);
+  int build_scan_filter_group(Ndb_cond* &cond, 
+                              NdbScanFilter* filter);
+  int build_scan_filter(Ndb_cond* &cond, NdbScanFilter* filter);
+
+  Ndb_cond_stack *m_cond_stack;
+};