New simplified engine interface for PBMS

By making PBMS transactional I have been able to greatly simplify the engine interface making it much easier for engine builders to build in support for PBMS. How much simpler is it? From the time I decided to make InnoDB PBMS enabled to when I started the rebuild of MySQL was less than half an hour!

The same way that I added PBMS support to InnoDB it can be added directly to the MySQL server so that the PBMS engine will be used for BLOB storage for all engines regardless of if they have been enabled or not. PBMS support for drizzle will be provided via a data filter plug-in which I have yet to write but will soon.

To add PBMS support all you need to do is add the file pbms_enabled.cc to your source code and add the following to your handler code. I will use the InnoDB handler code as an example:


File ha_innodb.cc:

#ifdef USE_PRAGMA_IMPLEMENTATION
#pragma implementation    // gcc: Class implementation
#endif
:
:
:
/* Include necessary InnoDB headers */
extern "C" {
#include "../storage/innobase/include/univ.i"
:
:
:
#include "../storage/innobase/include/ha_prototypes.h"
}

#define PBMS_ENABLED

#ifdef PBMS_ENABLED
#include "pbms_enabled.h"
#endif

static const long AUTOINC_OLD_STYLE_LOCKING = 0;
static const long AUTOINC_NEW_STYLE_LOCKING = 1;
:
:
:
innobase_init(
/*==========*/
           /* out: 0 on success, error code on failure */
void    *p) /* in: InnoDB handlerton */
{
:
:
:
err = innobase_start_or_create_for_mysql();

if (err != DB_SUCCESS) {
   my_free(internal_innobase_data_file_path,
                    MYF(MY_ALLOW_ZERO_PTR));
   goto error;
}

#ifdef PBMS_ENABLED
   PBMSResultRec result;
   if (!pbms_initialize("InnoDB", &result)) {
       sql_print_error("pbms_initialize() Error: %s", result.mr_message);
       goto error;
   }
#endif
(void) hash_init(&innobase_open_tables,system_charset_info, 32, 0, 0,
               (hash_get_key) innobase_get_key, 0, 0);
:
:
:
error:
DBUG_RETURN(TRUE);
}
:
:
:
innobase_end(handlerton *hton, ha_panic_function type)
/*==============*/
           /* out: TRUE if error */
{
int err= 0;

DBUG_ENTER("innobase_end");

#ifdef __NETWARE    /* some special cleanup for NetWare */
if (nw_panic) {
   set_panic_flag_for_netware();
}
#endif
if (innodb_inited) {

:
:
:

#ifdef PBMS_ENABLED
       pbms_finalize();
#endif
}

DBUG_RETURN(err);
}

:
:
:
int
ha_innobase::write_row(
/*===================*/
       /* out: error code */
uchar* record) /* in: a row in MySQL format */
{
int      error = 0;
ibool    auto_inc_used= FALSE;
ulint    sql_command;
trx_t*   trx = thd_to_trx(user_thd);

DBUG_ENTER("ha_innobase::write_row");

#ifdef PBMS_ENABLED
   PBMSResultRec result;
   error = pbms_write_row_blobs(table, record, &result);
   if (error) {
       sql_print_error( "pbms_write_row_blobs() Error: %s", result.mr_message);
       DBUG_RETURN(error);
   }
#endif

:
:
:
#ifdef PBMS_ENABLED
   pbms_completed(table, (error == 0));
#endif
DBUG_RETURN(error);
}
:
:
:
ha_innobase::update_row(
/*====================*/
      /* out: error number or 0 */
const uchar*  old_row,  /* in: old row in MySQL format */
uchar*        new_row)  /* in: new row in MySQL format */
{
upd_t* uvect;
int    error = 0;
trx_t* trx = thd_to_trx(user_thd);

DBUG_ENTER("ha_innobase::update_row");

#ifdef PBMS_ENABLED
   PBMSResultRec result;

   error = pbms_delete_row_blobs(table, old_row, &result);
   if (error) {
       sql_print_error( "update_row:pbms_delete_row_blobs() Error: %s",
           result.mr_message);
       DBUG_RETURN(error);
   }
   error = pbms_write_row_blobs(table, new_row, &result);
   if (error) {
       sql_print_error( "update_row:pbms_write_row_blobs() Error: %s",
           result.mr_message);
       goto pbms_done;
   }
#endif

:
:
:
#ifdef PBMS_ENABLED
   pbms_done:
   pbms_completed(table, (error == 0));
#endif
DBUG_RETURN(error);
}
:
:
:
int
ha_innobase::delete_row(
/*====================*/
                       /* out: error number or 0 */
const uchar*    record) /* in: a row in MySQL format */
{
int    error = 0;
trx_t* trx = thd_to_trx(user_thd);

DBUG_ENTER("ha_innobase::delete_row");

#ifdef PBMS_ENABLED
   PBMSResultRec result;

   error = pbms_delete_row_blobs(table, record, &result);
   if (error) {
       sql_print_error( "pbms_delete_row_blobs() Error: %s", result.mr_message);
       DBUG_RETURN(error);
   }
#endif

:
:
:
#ifdef PBMS_ENABLED
   pbms_completed(table, (error == 0));
#endif
DBUG_RETURN(error);
}
:
:
:
int
ha_innobase::delete_table(
/*======================*/
                   /* out: error number */
const char* name)   /* in: table name */
{
:
:
:
#ifdef PBMS_ENABLED
   /* Call pbms_delete_table_with_blobs() last because it cannot be undone. */
   if (!error) {
       PBMSResultRec result;

       if (pbms_delete_table_with_blobs(name, &result)) {
           sql_print_error( "pbms_delete_table_with_blobs() Error: %s",
               result.mr_message);
       }
  
       pbms_completed(NULL, true);
   }
#endif
DBUG_RETURN(error);
}

:
:
:
int
ha_innobase::rename_table(
/*======================*/
                  /* out: 0 or error code */
const char* from,  /* in: old name of the table */
const char* to)    /* in: new name of the table */
{
ulint   name_len1;
ulint   name_len2;
int     error;
trx_t*  parent_trx;
trx_t*  trx;
char    norm_from[1000];
char    norm_to[1000];
THD*    thd = ha_thd();

DBUG_ENTER("ha_innobase::rename_table");

#ifdef PBMS_ENABLED
   PBMSResultRec result;

   error = pbms_rename_table_with_blobs(from, to, &result);
   if (error) {
       sql_print_error( "pbms_rename_table_with_blobs() Error: %s", result.mr_message);
       DBUG_RETURN(error);
   }
#endif
:
:
:
#ifdef PBMS_ENABLED
   pbms_completed(NULL, (error == 0));
#endif
DBUG_RETURN(error);
}

Now you are probably wondering how I spent half an hour just adding that.