Condition pushdown to storage engine

include/mysql_com.h

@@ -339,7 +339,7 @@ struct rand_struct {
 
   /* The following is for user defined functions */
 
-enum Item_result {STRING_RESULT, REAL_RESULT, INT_RESULT, ROW_RESULT};
+enum Item_result {STRING_RESULT= 0, REAL_RESULT, INT_RESULT, ROW_RESULT};
 
 typedef struct st_udf_args
 {

mysql-test/r/ndb_condition_pushdown.result

+DROP TABLE IF EXISTS t1,t2;
+CREATE TABLE t1 (pk1 int unsigned NOT NULL PRIMARY KEY,   attr1 int unsigned NOT NULL,   attr2 int unsigned,   attr3 VARCHAR(10) ) ENGINE=ndbcluster;
+insert into t1 values (0,0,0, "a"),(1,1,1,"b"),(2,2,NULL,NULL),(3,3,3,"d"),(4,4,4,"e"),(5,5,5,"f");
+CREATE TABLE  t2 (pk1 int unsigned NOT NULL PRIMARY KEY,   attr1 int unsigned NOT NULL,   attr2 int unsigned,   attr3 VARCHAR(10) ) ENGINE=ndbcluster;
+insert into t2 values (0,0,0, "a"),(1,1,1,"b"),(2,2,2,"c"),(3,3,3,"d"),(4,4,4,"e"),(5,5,5,"f");
+set @old_ndbcpd = @@session.ndb_condition_pushdown;
+set ndb_condition_pushdown = off;
+select * from t1 where attr3 is null or attr1 > 2 and pk1= 3 order by pk1;
+pk1	attr1	attr2	attr3
+2	2	NULL	NULL
+3	3	3	d
+select * from t1,t2 where t1.attr1 > 1 and t1.attr2 = t2.attr2 and t2.attr1 < 5 order by t1.pk1;
+pk1	attr1	attr2	attr3	pk1	attr1	attr2	attr3
+3	3	3	d	3	3	3	d
+4	4	4	e	4	4	4	e
+set ndb_condition_pushdown = on;
+select * from t1 where attr3 is null or attr1 > 2 and pk1= 3 order by pk1;
+pk1	attr1	attr2	attr3
+2	2	NULL	NULL
+3	3	3	d
+select * from t1,t2 where t1.attr1 > 1 and t1.attr2 = t2.attr2 and t2.attr1 < 5 order by t1.pk1;
+pk1	attr1	attr2	attr3	pk1	attr1	attr2	attr3
+3	3	3	d	3	3	3	d
+4	4	4	e	4	4	4	e
+set ndb_condition_pushdown = @old_ndbcpd;
+DROP TABLE t1,t2;

mysql-test/t/ndb_condition_pushdown.test

+-- source include/have_ndb.inc
+
+--disable_warnings
+DROP TABLE IF EXISTS t1,t2;
+--enable_warnings
+
+#
+# Test of condition pushdown to storage engine
+#
+CREATE TABLE t1 (pk1 int unsigned NOT NULL PRIMARY KEY,   attr1 int unsigned NOT NULL,   attr2 int unsigned,   attr3 VARCHAR(10) ) ENGINE=ndbcluster;
+
+insert into t1 values (0,0,0, "a"),(1,1,1,"b"),(2,2,NULL,NULL),(3,3,3,"d"),(4,4,4,"e"),(5,5,5,"f");
+
+CREATE TABLE  t2 (pk1 int unsigned NOT NULL PRIMARY KEY,   attr1 int unsigned NOT NULL,   attr2 int unsigned,   attr3 VARCHAR(10) ) ENGINE=ndbcluster;
+
+insert into t2 values (0,0,0, "a"),(1,1,1,"b"),(2,2,2,"c"),(3,3,3,"d"),(4,4,4,"e"),(5,5,5,"f");
+
+set @old_ndbcpd = @@session.ndb_condition_pushdown;
+set ndb_condition_pushdown = off;
+select * from t1 where attr3 is null or attr1 > 2 and pk1= 3 order by pk1;
+select * from t1,t2 where t1.attr1 > 1 and t1.attr2 = t2.attr2 and t2.attr1 < 5 order by t1.pk1;
+set ndb_condition_pushdown = on;
+select * from t1 where attr3 is null or attr1 > 2 and pk1= 3 order by pk1;
+select * from t1,t2 where t1.attr1 > 1 and t1.attr2 = t2.attr2 and t2.attr1 < 5 order by t1.pk1;
+set ndb_condition_pushdown = @old_ndbcpd;
+DROP TABLE t1,t2;

ndb/src/ndbapi/NdbDictionaryImpl.hpp

@@ -531,7 +531,7 @@ NdbTableImpl::getColumn(const char * name){
     do {
       if(hashValue == (tmp & 0xFFFE)){
 	NdbColumnImpl* col = cols[tmp >> 16];
-	if(strcmp(name, col->m_name.c_str()) == 0){
+	if(strncmp(name, col->m_name.c_str(), NDB_MAX_ATTR_NAME_SIZE-1) == 0){
 	  return col;
 	}
       }
@@ -549,7 +549,7 @@ NdbTableImpl::getColumn(const char * name){
   } else {
     for(Uint32 i = 0; i<sz; i++){
       NdbColumnImpl* col = * cols++;
-      if(col != 0 && strcmp(name, col->m_name.c_str()) == 0)
+      if(col != 0 && strncmp(name, col->m_name.c_str(), NDB_MAX_ATTR_NAME_SIZE-1) == 0)
 	return col;
     }
   }

sql/ha_ndbcluster.h

@@ -32,6 +32,7 @@ class NdbOperation;    // Forward declaration
 class NdbConnection;   // Forward declaration
 class NdbRecAttr;      // Forward declaration
 class NdbScanOperation; 
+class NdbScanFilter; 
 class NdbIndexScanOperation; 
 class NdbBlob;
 
@@ -60,6 +61,145 @@ typedef struct st_ndbcluster_share {
   uint table_name_length,use_count;
 } NDB_SHARE;
 
+typedef enum ndb_item_type {
+  NDB_VALUE = 0,   // Qualified more with Item::Type
+  NDB_FIELD = 1,   // Qualified from table definition
+  NDB_FUNCTION = 2 // Qualified from Item_func::Functype
+} NDB_ITEM_TYPE;
+
+typedef union ndb_item_qualification {
+  Item::Type value_type; 
+  enum_field_types field_type;       // Instead of Item::FIELD_ITEM
+  Item_func::Functype function_type; // Instead of Item::FUNC_ITEM
+} NDB_ITEM_QUALIFICATION;
+
+class Ndb_item_string_value {
+ public:
+  String s;
+  CHARSET_INFO *c;
+};
+
+typedef struct ndb_item_field_value {
+  Field* field;
+  int column_no;
+} NDB_ITEM_FIELD_VALUE;
+
+typedef union ndb_item_value {
+  longlong int_value;
+  double real_value;
+  Ndb_item_string_value *string_value;
+  NDB_ITEM_FIELD_VALUE *field_value;
+} NDB_ITEM_VALUE;
+
+class Ndb_item {
+ public:
+  Ndb_item(NDB_ITEM_TYPE item_type, 
+	   NDB_ITEM_QUALIFICATION item_qualification,
+	   const Item *item_value);
+  Ndb_item(longlong int_value);
+  Ndb_item(double real_value);
+  Ndb_item();
+  Ndb_item(Field *field, int column_no);
+  Ndb_item(Item_func::Functype func_type);
+  ~Ndb_item();
+  void print(String *str);
+  // Getters and Setters
+  longlong getIntValue() { return value.int_value; };
+  double getRealValue() { return value.real_value; };
+  String * getStringValue() { return &value.string_value->s; };
+  CHARSET_INFO * getStringCharset() { return value.string_value->c; };
+  Field * getField() { return value.field_value->field; };
+  int getFieldNo() { return value.field_value->column_no; };
+
+ public:
+  NDB_ITEM_TYPE type;
+  NDB_ITEM_QUALIFICATION qualification;
+
+
+ private:
+  NDB_ITEM_VALUE value;
+
+};
+
+class Ndb_cond {
+ public:
+  Ndb_cond() : ndb_item(NULL), next(NULL), prev(NULL) {};
+  ~Ndb_cond() 
+  { 
+    if (ndb_item) delete ndb_item; 
+    ndb_item= NULL; 
+    if (next) delete next;
+    next= prev= NULL; 
+  };
+  Ndb_item *ndb_item;
+  Ndb_cond *next;
+  Ndb_cond *prev;
+};
+
+class Ndb_cond_stack {
+ public:
+  Ndb_cond_stack() : ndb_cond(NULL), next(NULL) {};
+  ~Ndb_cond_stack() 
+  { 
+    if (ndb_cond) delete ndb_cond; 
+    ndb_cond= NULL; 
+    next= NULL; 
+  };
+  Ndb_cond *ndb_cond;
+  Ndb_cond_stack *next;
+};
+
+class Ndb_cond_traverse_context {
+ public:
+  Ndb_cond_traverse_context(TABLE *tab, void* ndb_tab, 
+			    bool *supported, Ndb_cond_stack* stack)
+    : table(tab), ndb_table(ndb_tab), 
+    supported_ptr(supported), stack_ptr(stack), cond_ptr(NULL),
+    expect_mask(0), expect_field_result_mask(0)
+  {
+    if (stack)
+      cond_ptr= stack->ndb_cond;
+  };
+  void expect(Item::Type type)
+  {
+    expect_mask|= (1 << type);
+  };
+  void dont_expect(Item::Type type)
+  {
+    expect_mask&= ~(1 << type);
+  };
+  bool expecting(Item::Type type)
+  {
+    return (expect_mask & (1 << type));
+  };
+  void expect_nothing()
+  {
+    expect_mask= 0;
+  };
+
+  void expect_field_result(Item_result result)
+  {
+    expect_field_result_mask|= (1 << result);
+  };
+  bool expecting_field_result(Item_result result)
+  {
+    return (expect_field_result_mask & (1 << result));
+  };
+  void expect_no_field_result()
+  {
+    expect_field_result_mask= 0;
+  };
+
+  TABLE* table;
+  void* ndb_table;
+  bool *supported_ptr;
+  Ndb_cond_stack* stack_ptr;
+  Ndb_cond* cond_ptr;
+  private:
+  uint expect_mask;
+  uint expect_field_result_mask;
+};
+
 /*
   Place holder for ha_ndbcluster thread specific data
 */
@@ -122,7 +262,6 @@ class ha_ndbcluster: public handler
   void info(uint);
   int extra(enum ha_extra_function operation);
   int extra_opt(enum ha_extra_function operation, ulong cache_size);
-  int reset();
   int external_lock(THD *thd, int lock_type);
   int start_stmt(THD *thd);
   const char * table_type() const;
@@ -152,6 +291,13 @@ class ha_ndbcluster: public handler
 
   static Thd_ndb* seize_thd_ndb();
   static void release_thd_ndb(Thd_ndb* thd_ndb);
+ 
+  /*
+    Condition pushdown
+  */
+  const COND *cond_push(const COND *cond);
+  void cond_pop();
+
   uint8 table_cache_type();
     
  private:
@@ -214,9 +360,32 @@ class ha_ndbcluster: public handler
 
   int write_ndb_file();
 
- private:
   int check_ndb_connection();
 
+  void set_rec_per_key();
+  void records_update();
+  void no_uncommitted_rows_execute_failure();
+  void no_uncommitted_rows_update(int);
+  void no_uncommitted_rows_init(THD *);
+  void no_uncommitted_rows_reset(THD *);
+
+  /*
+    Condition Pushdown to Handler (CPDH), private methods
+  */
+  void cond_clear();
+  bool serialize_cond(const COND *cond, Ndb_cond_stack *ndb_cond);
+  Ndb_cond * build_scan_filter_predicate(Ndb_cond* cond, 
+					 NdbScanFilter* filter);
+  Ndb_cond * build_scan_filter_group(Ndb_cond* cond, 
+				     NdbScanFilter* filter);
+  void build_scan_filter(Ndb_cond* cond, NdbScanFilter* filter);
+  void generate_scan_filter(Ndb_cond_stack* cond_stack, 
+			    NdbScanOperation* op);
+
+  friend int execute_commit(ha_ndbcluster*, NdbConnection*);
+  friend int execute_no_commit(ha_ndbcluster*, NdbConnection*);
+  friend int execute_no_commit_ie(ha_ndbcluster*, NdbConnection*);
+
   NdbConnection *m_active_trans;
   NdbScanOperation *m_active_cursor;
   Ndb *m_ndb;
@@ -254,7 +423,7 @@ class ha_ndbcluster: public handler
   ha_rows m_autoincrement_prefetch;
   bool m_transaction_on;
   bool m_use_local_query_cache;
-
+  Ndb_cond_stack *m_cond_stack;
   bool m_disable_multi_read;
   byte *m_multi_range_result_ptr;
   uint m_multi_range_defined_count;
@@ -263,16 +432,6 @@ class ha_ndbcluster: public handler
   byte *m_multi_range_cursor_result_ptr;
   int setup_recattr(const NdbRecAttr*);
   
-  void set_rec_per_key();
-  void records_update();
-  void no_uncommitted_rows_execute_failure();
-  void no_uncommitted_rows_update(int);
-  void no_uncommitted_rows_init(THD *);
-  void no_uncommitted_rows_reset(THD *);
-
-  friend int execute_no_commit(ha_ndbcluster*, NdbConnection*);
-  friend int execute_commit(ha_ndbcluster*, NdbConnection*);
-  friend int execute_no_commit_ie(ha_ndbcluster*, NdbConnection*);
 };
 
 bool ndbcluster_init(void);

sql/handler.h

@@ -264,6 +264,9 @@ typedef struct st_table TABLE;
 struct st_foreign_key_info;
 typedef struct st_foreign_key_info FOREIGN_KEY_INFO;
 
+/* Forward declaration for Condition Pushdown to Handler (CPDH) */
+typedef struct Item COND;
+
 typedef struct st_ha_check_opt
 {
   ulong sort_buffer_size;
@@ -563,7 +566,7 @@ class handler :public Sql_alloc
 
   /* Type of table for caching query */
   virtual uint8 table_cache_type() { return HA_CACHE_TBL_NONTRANSACT; }
- 
+
  /*
   RETURN
     true  Primary key (if there is one) is clustered key covering all fields
@@ -575,6 +578,12 @@ class handler :public Sql_alloc
  {
    return memcmp(ref1, ref2, ref_length);
  }
+ 
+ /*
+   Condition pushdown to storage engines
+ */
+ virtual const COND *cond_push(const COND *cond) { return cond; };
+ virtual void cond_pop() { return; };
 };
 
 	/* Some extern variables used with handlers */

sql/item.h

@@ -113,6 +113,8 @@ class DTCollation {
 typedef bool (Item::*Item_processor)(byte *arg);
 typedef Item* (Item::*Item_transformer) (byte *arg);
 
+typedef void (*Item_cond_traverser) (const Item *item, void *arg);
+
 class Item {
   Item(const Item &);			/* Prevent use of these */
   void operator=(Item &);
@@ -124,7 +126,7 @@ class Item {
   static void operator delete(void *ptr,size_t size, MEM_ROOT *mem_root)
   { TRASH(ptr, size); }
 
-  enum Type {FIELD_ITEM, FUNC_ITEM, SUM_FUNC_ITEM, STRING_ITEM,
+  enum Type {FIELD_ITEM= 0, FUNC_ITEM, SUM_FUNC_ITEM, STRING_ITEM,
 	     INT_ITEM, REAL_ITEM, NULL_ITEM, VARBIN_ITEM,
 	     COPY_STR_ITEM, FIELD_AVG_ITEM, DEFAULT_VALUE_ITEM,
 	     PROC_ITEM,COND_ITEM, REF_ITEM, FIELD_STD_ITEM,
@@ -133,6 +135,8 @@ class Item {
              PARAM_ITEM, TRIGGER_FIELD_ITEM};
 
   enum cond_result { COND_UNDEF,COND_OK,COND_TRUE,COND_FALSE };
+
+  enum traverse_order { POSTFIX, PREFIX };
   
   /*
     str_values's main purpose is to be used to cache the value in
@@ -314,6 +318,13 @@ class Item {
   {
     return (this->*transformer)(arg);
   }
+
+   virtual void traverse_cond(Item_cond_traverser traverser, 
+			      void *arg,
+			      traverse_order order = POSTFIX)
+   {
+     (*traverser)(this, arg);
+   }
  
   virtual bool remove_dependence_processor(byte * arg) { return 0; }
   virtual bool remove_fixed(byte * arg) { fixed= 0; return 0; }

sql/item_cmpfunc.cc

@@ -2037,6 +2037,20 @@ Item *Item_cond::transform(Item_transformer transformer, byte *arg)
   return Item_func::transform(transformer, arg);
 }
 
+void Item_cond::traverse_cond(Item_cond_traverser traverser, 
+			      void *arg,
+			      traverse_order order)
+{
+  List_iterator<Item> li(list);
+  Item *item;
+
+  if (order == PREFIX) (*traverser)(this, arg);
+  while ((item= li++))
+  {
+    item->traverse_cond(traverser, arg, order);
+  }
+  if (order == POSTFIX) (*traverser)(this, arg);
+}
 
 void Item_cond::split_sum_func(THD *thd, Item **ref_pointer_array,
                                List<Item> &fields)

sql/item_cmpfunc.h

@@ -965,6 +965,9 @@ class Item_cond :public Item_bool_func
   void copy_andor_arguments(THD *thd, Item_cond *item);
   bool walk(Item_processor processor, byte *arg);
   Item *transform(Item_transformer transformer, byte *arg);
+  void traverse_cond(Item_cond_traverser, 
+		     void *arg,
+		     traverse_order order = POSTFIX);
   void neg_arguments(THD *thd);
 };
 

sql/item_func.cc

@@ -342,6 +342,22 @@ bool Item_func::walk (Item_processor processor, byte *argument)
   return (this->*processor)(argument);
 }
 
+void Item_func::traverse_cond(Item_cond_traverser traverser, 
+			      void *argument,
+			      traverse_order order)
+{
+  if (arg_count)
+  {
+    Item **arg,**arg_end;
+    if (order == PREFIX) (traverser)(this, argument);
+    for (arg= args, arg_end= args+arg_count; arg != arg_end; arg++)
+    {
+      (*arg)->traverse_cond(traverser, argument, order);
+    }
+  }
+  if (order == POSTFIX) (traverser)(this, argument);
+}
+
 
 /*
   Transform an Item_func object with a transformer callback function

sql/item_func.h

@@ -154,6 +154,9 @@ class Item_func :public Item_result_field
                         uint flags= 0);
   bool walk(Item_processor processor, byte *arg);
   Item *transform(Item_transformer transformer, byte *arg);
+  void traverse_cond(Item_cond_traverser traverser, 
+		     void * arg,
+		     traverse_order order = POSTFIX);
 };
 
 

sql/mysqld.cc

@@ -4093,6 +4093,7 @@ enum options_mysqld
   OPT_INNODB, OPT_ISAM,
   OPT_NDBCLUSTER, OPT_NDB_CONNECTSTRING, OPT_NDB_USE_EXACT_COUNT,
   OPT_NDB_FORCE_SEND, OPT_NDB_AUTOINCREMENT_PREFETCH_SZ,
+  OPT_NDB_CONDITION_PUSHDOWN,
   OPT_SKIP_SAFEMALLOC,
   OPT_TEMP_POOL, OPT_TX_ISOLATION,
   OPT_SKIP_STACK_TRACE, OPT_SKIP_SYMLINKS,
@@ -4547,6 +4548,12 @@ Disable with --skip-ndbcluster (will save memory).",
    (gptr*) &opt_ndbcluster, (gptr*) &opt_ndbcluster, 0, GET_BOOL, NO_ARG,
    OPT_NDBCLUSTER_DEFAULT, 0, 0, 0, 0, 0},
 #ifdef HAVE_NDBCLUSTER_DB
+  {"ndb-condition-pushdown", 
+   OPT_NDB_CONDITION_PUSHDOWN,
+   "Push supported query conditions to the ndbcluster storage engine.",
+   (gptr*) &global_system_variables.ndb_condition_pushdown,
+   (gptr*) &global_system_variables.ndb_condition_pushdown,
+   0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
   {"ndb-connectstring", OPT_NDB_CONNECTSTRING,
    "Connect string for ndbcluster.",
    (gptr*) &ndbcluster_connectstring, (gptr*) &ndbcluster_connectstring,

sql/set_var.cc

@@ -388,6 +388,9 @@ sys_ndb_use_exact_count("ndb_use_exact_count",
 sys_var_thd_bool
 sys_ndb_use_transactions("ndb_use_transactions",
 			 &SV::ndb_use_transactions);
+sys_var_thd_bool
+sys_ndb_condition_pushdown("ndb_condition_pushdown",
+			 &SV::ndb_condition_pushdown);
 // ndb server global variable settings
 // none
 #endif
@@ -654,6 +657,7 @@ sys_var *sys_variables[]=
   &sys_ndb_force_send,
   &sys_ndb_use_exact_count,
   &sys_ndb_use_transactions,
+  &sys_ndb_condition_pushdown,
 #endif
   &sys_unique_checks,
   &sys_updatable_views_with_limit,
@@ -824,6 +828,8 @@ struct show_var_st init_vars[]= {
   {sys_ndb_force_send.name,   (char*) &sys_ndb_force_send,          SHOW_SYS},
   {sys_ndb_use_exact_count.name,(char*) &sys_ndb_use_exact_count,   SHOW_SYS},
   {sys_ndb_use_transactions.name,(char*) &sys_ndb_use_transactions, SHOW_SYS},
+  {sys_ndb_condition_pushdown.name, (char*) &sys_ndb_condition_pushdown,      
+   SHOW_SYS},
 #endif
   {sys_net_buffer_length.name,(char*) &sys_net_buffer_length,       SHOW_SYS},
   {sys_net_read_timeout.name, (char*) &sys_net_read_timeout,        SHOW_SYS},

sql/sql_class.h

@@ -436,6 +436,7 @@ struct system_variables
   my_bool ndb_force_send;
   my_bool ndb_use_exact_count;
   my_bool ndb_use_transactions;
+  my_bool ndb_condition_pushdown;
 #endif /* HAVE_NDBCLUSTER_DB */
   my_bool old_passwords;
   

sql/sql_select.cc

@@ -5269,6 +5269,7 @@ make_join_select(JOIN *join,SQL_SELECT *select,COND *cond)
           if (!(tmp= add_found_match_trig_cond(first_inner_tab, tmp, 0)))
             DBUG_RETURN(1);
           tab->select_cond=sel->cond=tmp;
+	  tab->table->file->cond_push(tmp); // Push condition to handler
         }
         else
           tab->select_cond= sel->cond= NULL;
@@ -5390,6 +5391,8 @@ make_join_select(JOIN *join,SQL_SELECT *select,COND *cond)
 		join->thd->memdup((gptr) sel, sizeof(SQL_SELECT));
 	      tab->cache.select->cond=tmp;
 	      tab->cache.select->read_tables=join->const_table_map;
+	      if (tmp != tab->select_cond)
+		tab->table->file->cond_push(tmp); // Push condition to handler
 	    }
 	  }
 	}