Merge poseidon.ndb.mysql.com:/home/tomas/mysql-4.1

into poseidon.ndb.mysql.com:/home/tomas/mysql-4.1-qc


sql/ha_ndbcluster.cc:
  Auto merged
sql/ha_ndbcluster.h:
  Auto merged

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;

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);

sql/ha_ndbcluster.cc

@@ -4457,19 +4457,145 @@ const char* ha_ndbcluster::index_type(uint key_number)
 }
 uint8 ha_ndbcluster::table_cache_type()
 {
+  DBUG_ENTER("ha_ndbcluster::table_cache_type");
   switch (m_query_cache_type)
   {
   case 0:
-    return HA_CACHE_TBL_NOCACHE;
+    DBUG_PRINT("exit",("HA_CACHE_TBL_NOCACHE"));
+    DBUG_RETURN(HA_CACHE_TBL_NOCACHE);
   case 1:
-    return HA_CACHE_TBL_ASKTRANSACT;
+    DBUG_PRINT("exit",("HA_CACHE_TBL_ASKTRANSACT"));
+    DBUG_RETURN(HA_CACHE_TBL_ASKTRANSACT);
   case 2:
-    return HA_CACHE_TBL_TRANSACT;
+    DBUG_PRINT("exit",("HA_CACHE_TBL_TRANSACT"));
+    DBUG_RETURN(HA_CACHE_TBL_TRANSACT);
   default:
-    return HA_CACHE_TBL_NOCACHE;
+    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
+    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);
+}
+
 /*
   Handling the shared NDB_SHARE structure that is needed to 
   provide table locking.

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();

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()
 {

sql/handler.h

@@ -506,10 +506,15 @@ class handler :public Sql_alloc
 
   /* 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);

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"

sql/sql_cache.cc

@@ -918,12 +918,12 @@ void Query_cache::store_query(THD *thd, TABLE_LIST *tables_used)
 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)
@@ -1024,7 +1024,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;
@@ -1085,19 +1084,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(), 
+    engine_data= table->engine_data();
+    if (table->callback() &&
+        !(*table->callback())(thd, table->db(),
                               table->key_length(),
-                                         table->type()))
+                              &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++;
@@ -2122,7 +2131,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)
@@ -2138,9 +2149,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;
       }
     }
@@ -2167,7 +2182,9 @@ my_bool Query_cache::register_all_tables(Query_cache_block *block,
 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",
@@ -2177,6 +2194,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)",
@@ -2207,6 +2241,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);
@@ -2727,9 +2763,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,
+    handler *handler= table->file;
+    if (!handler->cached_table_registration(thd, table->table_cache_key,
                                             table->key_length,
-                         table->file->table_cache_type()))
+                                            &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));

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 @@ class Query_cache
 			      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);

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 */