sql/ha_innodb.cc

    enabled query cache for ndb
    modified engine interface somewhat
sql/ha_innodb.h
    enabled query cache for ndb
    modified engine interface somewhat
sql/ha_ndbcluster.cc
    enabled query cache for ndb
    modified engine interface somewhat
    ndb will only allow caching and retrieval if running autocommit
      - return false, but do not invalidate
    commit count is used as engine data, i.e.
      - store commit count before store of cache
      - allow retrieval if commit count has not changed on a table
      - invalidate if commit count has changed
sql/ha_ndbcluster.h
    enabled query cache for ndb
    modified engine interface somewhat
sql/handler.cc
    enabled query cache for ndb
    modified engine interface somewhat
sql/handler.h
    enabled query cache for ndb
    modified engine interface somewhat
    new virtual handler method cached_table_registration called on each table before alowing store in query cache
      - return TRUE - ok to cache, FALSE - not allowed to cache, invalidate queries if engine_data below has changed
      - sets ulonglong (engine_data) that is stored in query cache for each table
      - sets callback to be called for each table before usage of cached query, callback = 0 -> no check later
sql/mysql_priv.h
    enabled query cache for ndb
    modified engine interface somewhat
    callcack prototype for callback to engine before query cache retrieval
sql/sql_cache.cc
    enabled query cache for ndb
    modified engine interface somewhat
    if callback is set on table in cache, do callback to check if allowed to use cache
    if not allowed to use cache, check if engine_data has changed, if so, invalidate all queries with that table
    + changes to store and pass callback and engine_data around
sql/sql_cache.h
    enabled query cache for ndb
    modified engine interface somewhat
    changes to store callback and engine_data
sql/table.h
    enabled query cache for ndb
    modified engine interface somewhat
    changes to store callback and engine_data

10 files changed, 232 insertions, 35 deletions

diff --git a/sql/ha_innodb.cc b/sql/ha_innodb.cc
index 07d8da63733..b5c94386677 100644
--- a/sql/ha_innodb.cc
+++ b/sql/ha_innodb.cc
@@ -655,8 +655,9 @@ innobase_query_caching_of_table_permitted(
 	char*	full_name,	/* in: concatenation of database name,
 				the null character '\0', and the table
 				name */
-	uint	full_name_len)	/* in: length of the full name, i.e.
+	uint	full_name_len,	/* in: length of the full name, i.e.
 				len(dbname) + len(tablename) + 1 */
+        ulonglong *unused)      /* unused for this engine */
 {
 	ibool	is_autocommit;
 	trx_t*	trx;
diff --git a/sql/ha_innodb.h b/sql/ha_innodb.h
index e76a966c6b9..e3b058d0b42 100644
--- a/sql/ha_innodb.h
+++ b/sql/ha_innodb.h
@@ -33,6 +33,10 @@ typedef struct st_innobase_share {
 } INNOBASE_SHARE;
 
 
+my_bool innobase_query_caching_of_table_permitted(THD* thd, char* full_name,
+                                                  uint full_name_len,
+                                                  ulonglong *unused);
+
 /* The class defining a handle to an Innodb table */
 class ha_innobase: public handler
 {
@@ -168,6 +172,20 @@ class ha_innobase: public handler
 	void init_table_handle_for_HANDLER(); 
 	longlong get_auto_increment();
         uint8 table_cache_type() { return HA_CACHE_TBL_ASKTRANSACT; }
+        /*
+          ask handler about permission to cache table during query registration
+        */
+        my_bool cached_table_registration(THD *thd, char *table_key,
+                                          uint key_length,
+                                          qc_engine_callback *call_back,
+                                          ulonglong *engine_data)
+        {
+          *call_back= innobase_query_caching_of_table_permitted;
+          *engine_data= 0;
+          return innobase_query_caching_of_table_permitted(thd, table_key,
+                                                           key_length,
+                                                           engine_data);
+        }
 
         static char      *get_mysql_bin_log_name();
         static ulonglong get_mysql_bin_log_pos();
@@ -233,8 +251,6 @@ int innobase_close_connection(THD *thd);
 int innobase_drop_database(char *path);
 int innodb_show_status(THD* thd);
 
-my_bool innobase_query_caching_of_table_permitted(THD* thd, char* full_name,
-						uint full_name_len);
 void innobase_release_temporary_latches(void* innobase_tid);
 
 void innobase_store_binlog_offset_and_flush_log(char *binlog_name,longlong offset);
diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
index 8d82f60ae85..8621fb89563 100644
--- a/sql/ha_ndbcluster.cc
+++ b/sql/ha_ndbcluster.cc
@@ -4455,10 +4455,138 @@ const char* ha_ndbcluster::index_type(uint key_number)
 }
 uint8 ha_ndbcluster::table_cache_type()
 {
+  DBUG_ENTER("ha_ndbcluster::table_cache_type");
   if (m_use_local_query_cache)
-    return HA_CACHE_TBL_TRANSACT;
+  {
+    DBUG_PRINT("exit",("HA_CACHE_TBL_ASKTRANSACT"));
+    DBUG_RETURN(HA_CACHE_TBL_ASKTRANSACT); //HA_CACHE_TBL_TRANSACT;
+  }
+  else
+  {
+    DBUG_PRINT("exit",("HA_CACHE_TBL_NOCACHE"));
+    DBUG_RETURN(HA_CACHE_TBL_NOCACHE);
+  }
+}
+
+static
+my_bool
+ndbcluster_cache_retrieval_allowed(
+/*======================================*/
+				/* out: TRUE if permitted, FALSE if not;
+				note that the value FALSE means invalidation
+				of query cache if *engine_data is changed */
+	THD*	thd,		/* in: thd of the user who is trying to
+				store a result to the query cache or
+				retrieve it */
+	char*	full_name,	/* in: concatenation of database name,
+				the null character '\0', and the table
+				name */
+	uint	full_name_len,	/* in: length of the full name, i.e.
+				len(dbname) + len(tablename) + 1 */
+        ulonglong *engine_data) /* in: value set in call to 
+				ha_ndbcluster::cached_table_registration
+				   out: if return FALSE this is used to invalidate 
+				all cached queries with this table*/
+{
+  DBUG_ENTER("ndbcluster_cache_retrieval_allowed");
+  char tabname[128];
+  char *dbname= full_name;
+  my_bool is_autocommit;
+  {
+    int dbname_len= strlen(full_name);
+    int tabname_len= full_name_len-dbname_len-1;
+    memcpy(tabname, full_name+dbname_len+1, tabname_len);
+    tabname[tabname_len]= '\0';
+  }
+  if (thd->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN))
+    is_autocommit = FALSE;
+  else
+    is_autocommit = TRUE;
+  DBUG_PRINT("enter",("dbname=%s, tabname=%s, autocommit=%d",
+		      dbname,tabname,is_autocommit));
+  if (!is_autocommit)
+  {
+    DBUG_PRINT("info",("OPTION_NOT_AUTOCOMMIT=%d OPTION_BEGIN=%d",
+		       thd->options & OPTION_NOT_AUTOCOMMIT,
+		       thd->options & OPTION_BEGIN));
+    // ToDo enable cache inside a transaction
+    // no need to invalidate though so leave *engine_data
+    DBUG_RETURN(FALSE);
+  }
+  {
+    Ndb *ndb;
+    Uint64 commit_count;
+    if (!(ndb= check_ndb_in_thd(thd)))
+    {
+      *engine_data= *engine_data+1; // invalidate
+      DBUG_RETURN(FALSE);
+    }
+    ndb->setDatabaseName(dbname);
+    if (ndb_get_table_statistics(ndb, tabname, 0, &commit_count))
+    {
+      *engine_data= *engine_data+1; // invalidate
+      DBUG_RETURN(FALSE);
+    }
+    if (*engine_data != commit_count)
+    {
+      *engine_data= commit_count; // invalidate
+      DBUG_RETURN(FALSE);
+    }
+  }
+  DBUG_PRINT("exit",("*engine_data=%d ok, use cache",*engine_data));
+  DBUG_RETURN(TRUE);
+}
+
+my_bool
+ha_ndbcluster::cached_table_registration(
+/*======================================*/
+				/* out: TRUE if permitted, FALSE if not;
+				note that the value FALSE means invalidation
+				of query cache if *engine_data is changed */
+	THD*	thd,		/* in: thd of the user who is trying to
+				store a result to the query cache or
+				retrieve it */
+	char*	full_name,	/* in: concatenation of database name,
+				the null character '\0', and the table
+				name */
+	uint	full_name_len,	/* in: length of the full name, i.e.
+				len(dbname) + len(tablename) + 1 */
+	qc_engine_callback 
+	*engine_callback,       /* out: function to be called before using
+				   cache on this table */
+        ulonglong *engine_data) /* out: if return FALSE this is used to 
+				   invalidate all cached queries with this table*/
+{
+  DBUG_ENTER("ha_ndbcluster::cached_table_registration");
+  my_bool is_autocommit;
+  if (thd->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN))
+    is_autocommit = FALSE;
   else
-    return HA_CACHE_TBL_NOCACHE;
+    is_autocommit = TRUE;
+  DBUG_PRINT("enter",("dbname=%s, tabname=%s, is_autocommit=%d",
+		      m_dbname,m_tabname,is_autocommit));
+  if (!is_autocommit)
+  {
+    DBUG_PRINT("info",("OPTION_NOT_AUTOCOMMIT=%d OPTION_BEGIN=%d",
+		       thd->options & OPTION_NOT_AUTOCOMMIT,
+		       thd->options & OPTION_BEGIN));
+    // ToDo enable cache inside a transaction
+    // no need to invalidate though so leave *engine_data
+    DBUG_RETURN(FALSE);
+  }
+  {
+    Uint64 commit_count;
+    m_ndb->setDatabaseName(m_dbname);
+    if (ndb_get_table_statistics(m_ndb, m_tabname, 0, &commit_count))
+    {
+      *engine_data= 0;
+      DBUG_RETURN(FALSE);
+    }
+    *engine_data= commit_count;
+  }
+  *engine_callback= ndbcluster_cache_retrieval_allowed;
+  DBUG_PRINT("exit",("*engine_data=%d", *engine_data));
+  DBUG_RETURN(TRUE);
 }
 
 /*
diff --git a/sql/ha_ndbcluster.h b/sql/ha_ndbcluster.h
index f6c712620c1..7b3b5658175 100644
--- a/sql/ha_ndbcluster.h
+++ b/sql/ha_ndbcluster.h
@@ -146,7 +146,10 @@ class ha_ndbcluster: public handler
   static Thd_ndb* seize_thd_ndb();
   static void release_thd_ndb(Thd_ndb* thd_ndb);
   uint8 table_cache_type();
-    
+  my_bool cached_table_registration(THD *thd, char *table_key,
+				    uint key_length,
+				    qc_engine_callback *engine_callback,
+				    ulonglong *engine_data);
  private:
   int alter_table_name(const char *from, const char *to);
   int drop_table();
diff --git a/sql/handler.cc b/sql/handler.cc
index 7ddd7b80a34..9e781817c02 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -229,15 +229,6 @@ handler *get_new_handler(TABLE *table, enum db_type db_type)
   }
 }
 
-bool ha_caching_allowed(THD* thd, char* table_key,
-                        uint key_length, uint8 cache_type)
-{
-#ifdef HAVE_INNOBASE_DB
-  if (cache_type == HA_CACHE_TBL_ASKTRANSACT)
-    return innobase_query_caching_of_table_permitted(thd, table_key, key_length);
-#endif
-  return 1;
-}
 
 int ha_init()
 {
diff --git a/sql/handler.h b/sql/handler.h
index 252861e5c37..31710ec728c 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -506,10 +506,15 @@ public:
 
   /* Type of table for caching query */
   virtual uint8 table_cache_type() { return HA_CACHE_TBL_NONTRANSACT; }
-  /*
-    Is query with this table cachable (have sense only for ASKTRANSACT
-    tables)
-  */
+  /* ask handler about permission to cache table during query registration */
+  virtual my_bool cached_table_registration(THD *thd, char *table_key,
+                                            uint key_length,
+                                            qc_engine_callback *engine_callback,
+                                            ulonglong *engine_data)
+  {
+    *engine_callback= 0;
+    return 1;
+  }
 };
 
 	/* Some extern variables used with handlers */
@@ -528,8 +533,6 @@ extern TYPELIB tx_isolation_typelib;
                                  T != DB_TYPE_BERKELEY_DB && \
                                  T != DB_TYPE_NDBCLUSTER)
 
-bool ha_caching_allowed(THD* thd, char* table_key,
-                        uint key_length, uint8 cache_type);
 enum db_type ha_resolve_by_name(const char *name, uint namelen);
 const char *ha_get_storage_engine(enum db_type db_type);
 handler *get_new_handler(TABLE *table, enum db_type db_type);
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index 3f55a88b262..77703c2b390 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -340,6 +340,9 @@ inline THD *_current_thd(void)
 }
 #define current_thd _current_thd()
 
+typedef my_bool (*qc_engine_callback)(THD *thd, char *table_key,
+                                      uint key_length,
+                                      ulonglong *engine_data);
 #include "sql_string.h"
 #include "sql_list.h"
 #include "sql_map.h"
diff --git a/sql/sql_cache.cc b/sql/sql_cache.cc
index 1bf8d179770..5c2698bcef2 100644
--- a/sql/sql_cache.cc
+++ b/sql/sql_cache.cc
@@ -912,12 +912,12 @@ end:
 int
 Query_cache::send_result_to_client(THD *thd, char *sql, uint query_length)
 {
+  ulonglong engine_data;
   Query_cache_query *query;
   Query_cache_block *first_result_block, *result_block;
   Query_cache_block_table *block_table, *block_table_end;
   ulong tot_length;
   Query_cache_query_flags flags;
-  bool check_tables;
   DBUG_ENTER("Query_cache::send_result_to_client");
 
   if (query_cache_size == 0 || thd->variables.query_cache_type == 0)
@@ -1018,7 +1018,6 @@ Query_cache::send_result_to_client(THD *thd, char *sql, uint query_length)
     goto err_unlock;
   }
       
-  check_tables= query->tables_type() & HA_CACHE_TBL_ASKTRANSACT;
   // Check access;
   block_table= query_block->table(0);
   block_table_end= block_table+query_block->n_tables;
@@ -1079,19 +1078,29 @@ Query_cache::send_result_to_client(THD *thd, char *sql, uint query_length)
       goto err_unlock;				// Parse query
     }
 #endif /*!NO_EMBEDDED_ACCESS_CHECKS*/
-    if (check_tables && !ha_caching_allowed(thd, table->db(), 
-                                         table->key_length(),
-                                         table->type()))
+    engine_data= table->engine_data();
+    if (table->callback() &&
+        !(*table->callback())(thd, table->db(),
+                              table->key_length(),
+                              &engine_data))
     {
       DBUG_PRINT("qcache", ("Handler does not allow caching for %s.%s",
 			    table_list.db, table_list.alias));
       BLOCK_UNLOCK_RD(query_block);
       thd->lex->safe_to_cache_query= 0;          // Don't try to cache this
+      if (engine_data != table->engine_data())
+      {
+        DBUG_PRINT("qcache",
+                   ("Handler require invalidation queries of %s.%s %lld-%lld",
+                              table_list.db, table_list.alias,
+                              engine_data, table->engine_data()));
+        invalidate_table(table->db(), table->key_length());
+      }
       goto err_unlock;				// Parse query
     }
     else
-      DBUG_PRINT("qcache", ("handler allow caching (%d) %s,%s",
-			    check_tables, table_list.db, table_list.alias));
+      DBUG_PRINT("qcache", ("handler allow caching %s,%s",
+			    table_list.db, table_list.alias));
   }
   move_to_query_list_end(query_block);
   hits++;
@@ -2116,7 +2125,9 @@ my_bool Query_cache::register_all_tables(Query_cache_block *block,
     if (!insert_table(tables_used->table->key_length,
 		      tables_used->table->table_cache_key, block_table,
 		      tables_used->db_length,
-		      tables_used->table->file->table_cache_type()))
+		      tables_used->table->file->table_cache_type(),
+                      tables_used->callback_func,
+                      tables_used->engine_data))
       break;
 
     if (tables_used->table->db_type == DB_TYPE_MRG_MYISAM)
@@ -2132,9 +2143,13 @@ my_bool Query_cache::register_all_tables(Query_cache_block *block,
 	uint key_length= filename_2_table_key(key, table->table->filename,
 					      &db_length);
 	(++block_table)->n= ++n;
+        /*
+          There are not callback function for for MyISAM, and engine data
+        */
 	if (!insert_table(key_length, key, block_table,
 			  db_length,
-			  tables_used->table->file->table_cache_type()))
+			  tables_used->table->file->table_cache_type(),
+                          0, 0))
 	  goto err;
       }
     }
@@ -2161,7 +2176,9 @@ err:
 my_bool
 Query_cache::insert_table(uint key_len, char *key,
 			  Query_cache_block_table *node,
-			  uint32 db_length, uint8 cache_type)
+			  uint32 db_length, uint8 cache_type,
+                          qc_engine_callback callback,
+                          ulonglong engine_data)
 {
   DBUG_ENTER("Query_cache::insert_table");
   DBUG_PRINT("qcache", ("insert table node 0x%lx, len %d",
@@ -2171,6 +2188,23 @@ Query_cache::insert_table(uint key_len, char *key,
 				    hash_search(&tables, (byte*) key,
 						key_len));
 
+  if (table_block &&
+      table_block->table()->engine_data() != engine_data)
+  {
+    DBUG_PRINT("qcache",
+               ("Handler require invalidation queries of %s.%s %lld-%lld",
+                table_block->table()->db(),
+                table_block->table()->table(),
+                engine_data,
+                table_block->table()->engine_data()));
+    /*
+      as far as we delete all queries with this table, table block will be
+      deleted, too
+    */
+    invalidate_table(table_block);
+    table_block= 0;
+  }
+
   if (table_block == 0)
   {
     DBUG_PRINT("qcache", ("new table block from 0x%lx (%u)",
@@ -2201,6 +2235,8 @@ Query_cache::insert_table(uint key_len, char *key,
     header->table(db + db_length + 1);
     header->key_length(key_len);
     header->type(cache_type);
+    header->callback(callback);
+    header->engine_data(engine_data);
   }
 
   Query_cache_block_table *list_root = table_block->table(0);
@@ -2721,9 +2757,11 @@ my_bool Query_cache::ask_handler_allowance(THD *thd,
   for (; tables_used; tables_used= tables_used->next)
   {
     TABLE *table= tables_used->table;
-    if (!ha_caching_allowed(thd, table->table_cache_key,
-                         table->key_length,
-                         table->file->table_cache_type()))
+    handler *handler= table->file;
+    if (!handler->cached_table_registration(thd, table->table_cache_key,
+                                            table->key_length,
+                                            &tables_used->callback_func,
+                                            &tables_used->engine_data))
     {
       DBUG_PRINT("qcache", ("Handler does not allow caching for %s.%s",
 			    tables_used->db, tables_used->alias));
diff --git a/sql/sql_cache.h b/sql/sql_cache.h
index c933a2349af..7595bfbbd54 100644
--- a/sql/sql_cache.h
+++ b/sql/sql_cache.h
@@ -145,6 +145,10 @@ struct Query_cache_table
   char *tbl;
   uint32 key_len;
   uint8 table_type;
+  /* unique for every engine reference */
+  qc_engine_callback callback_func;
+  /* data need by some engines */
+  ulonglong engine_data_buff;
 
   inline char *db()			     { return (char *) data(); }
   inline char *table()			     { return tbl; }
@@ -153,6 +157,10 @@ struct Query_cache_table
   inline void key_length(uint32 len)         { key_len= len; }
   inline uint8 type()                        { return table_type; }
   inline void type(uint8 t)                  { table_type= t; }
+  inline qc_engine_callback callback()       { return callback_func; }
+  inline void callback(qc_engine_callback fn){ callback_func= fn; }
+  inline ulonglong engine_data()             { return engine_data_buff; }
+  inline void engine_data(ulonglong data)    { engine_data_buff= data; }
   inline gptr data()
   {
     return (gptr)(((byte*)this)+
@@ -281,7 +289,9 @@ protected:
 			      TABLE_COUNTER_TYPE tables);
   my_bool insert_table(uint key_len, char *key,
 		       Query_cache_block_table *node,
-		       uint32 db_length, uint8 cache_type);
+		       uint32 db_length, uint8 cache_type,
+		       qc_engine_callback callback,
+		       ulonglong engine_data);
   void unlink_table(Query_cache_block_table *node);
   Query_cache_block *get_free_block (ulong len, my_bool not_less,
 				      ulong min);
diff --git a/sql/table.h b/sql/table.h
index 2eb854f553d..c3945ac5d79 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -207,6 +207,10 @@ typedef struct st_table_list
   TABLE          *table;      /* opened table */
   st_table_list  *table_list; /* pointer to node of list of all tables */
   class st_select_lex_unit *derived;	/* SELECT_LEX_UNIT of derived table */
+  /* data need by some engines in query cache*/
+  ulonglong     engine_data;
+  /* call back function for asking handler about caching in query cache */
+  qc_engine_callback callback_func;
  GRANT_INFO	grant;
   thr_lock_type lock_type;
   uint		outer_join;		/* Which join type */