Merge lp:~barry-leslie/pbxt/pbms into lp:pbxt

[Barry Leslie]

Here is an update for PBMS compatibility. I have added a configure option --with-pbms to define PBMS_ENABLED for building a pbms enabled version.

=== modified file 'configure.in'
--- configure.in	2010-07-08 11:50:48 +0000
+++ configure.in	2010-07-21 22:08:48 +0000
@@ -138,6 +138,21 @@
 	    ENG_PLUGIN_DIR="$libdir/mysql/plugin"
     fi
 fi
+
+# ----- AC_ARG_WITH(pbms-port
+AC_ARG_WITH(pbms,
+    [  --with-pbms       Enable PBMS support],
+    [with_pbms="true"],
+    [with_pbms="false"])
+
+if test "$with_pbms" = "true" 
+then
+	CFLAGS="$CFLAGS -DPBMS_ENABLED"
+	CXXFLAGS="$CXXFLAGS  -DPBMS_ENABLED"
+fi
+
+
+
 AC_SUBST(ENG_PLUGIN_DIR)
 AC_SUBST(ENG_MYSQL_SRC)
 

=== modified file 'src/ha_pbxt.cc'
--- src/ha_pbxt.cc	2010-05-26 10:59:10 +0000
+++ src/ha_pbxt.cc	2010-07-21 22:08:48 +0000
@@ -1048,7 +1048,7 @@
 	 * tables belonging to this engine. This in turn may require some of
 	 * the stuff below (like xt_create_thread() called from pbxt_close_table()! */
 #ifdef PBMS_ENABLED
-	pbms_finalize();
+	pbms_finalize("PBXT");
 #endif
 	pbxt_call_exit(self);
 	xt_exit_threading(self);
@@ -1176,7 +1176,7 @@
 
 #ifdef PBMS_ENABLED
 		PBMSResultRec result;
-		if (!pbms_initialize("PBXT", false, &result)) {
+		if (!pbms_initialize("PBXT", false, true, &result, NULL)) {
 			xt_logf(XT_NT_ERROR, "pbms_initialize() Error: %s", result.mr_message);
 			goto error_2;
 		}
@@ -1345,7 +1345,7 @@
 
 	error_3:
 #ifdef PBMS_ENABLED
-	pbms_finalize();
+	pbms_finalize("PBXT");
 
 	error_2:
 #endif
@@ -2720,7 +2720,8 @@
 
 	done:
 #ifdef PBMS_ENABLED
-	pbms_completed(table, (err == 0));
+	if (result.mr_had_blobs)
+		pbms_completed(table, (err == 0));
 #endif
 	return err;
 }
@@ -2797,16 +2798,11 @@
 #ifdef PBMS_ENABLED
 	PBMSResultRec result;
 
-	err = pbms_delete_row_blobs(table, old_data, &result);
+	err = pbms_update_row_blobs(table, old_data, new_data, &result);
 	if (err) {
-		xt_logf(XT_NT_ERROR, "update_row:pbms_delete_row_blobs() Error: %s", result.mr_message);
+		xt_logf(XT_NT_ERROR, "update_row:pbms_update_row_blobs() Error: %s", result.mr_message);
 		return err;
 	}
-	err = pbms_write_row_blobs(table, new_data, &result);
-	if (err) { 
-		xt_logf(XT_NT_ERROR, "update_row:pbms_write_row_blobs() Error: %s", result.mr_message);
-		goto pbms_done;
-	}
 #endif
 
 	/* GOTCHA: We need to check the auto-increment value on update
@@ -2833,8 +2829,8 @@
 	pb_open_tab->ot_table->tab_locks.xt_remove_temp_lock(pb_open_tab, TRUE);
 	
 #ifdef PBMS_ENABLED
-	pbms_done:
-	pbms_completed(table, (err == 0));
+	if (result.mr_had_blobs)
+		pbms_completed(table, (err == 0));
 #endif
 
 	return err;
@@ -2879,7 +2875,8 @@
 	pb_open_tab->ot_table->tab_locks.xt_remove_temp_lock(pb_open_tab, TRUE);
 
 #ifdef PBMS_ENABLED
-	pbms_completed(table, (err == 0));
+	if (result.mr_had_blobs)
+		pbms_completed(table, (err == 0));
 #endif
 	return err;
 }

=== modified file 'src/pbms.h'
--- src/pbms.h	2009-09-03 06:16:11 +0000
+++ src/pbms.h	2010-07-21 22:08:48 +0000
@@ -26,8 +26,8 @@
  * are streaming enabled.
  *
  */
-#ifndef __streaming_unx_h__
-#define __streaming_unx_h__
+#ifndef __PBMS_H__
+#define __PBMS_H__
 
 #include <stdio.h>
 #include <sys/types.h>
@@ -39,22 +39,17 @@
 #include <signal.h>
 #include <ctype.h>
 #include <errno.h>
-
+#include <inttypes.h>
+#include <stdint.h>
 
 #ifdef USE_PRAGMA_INTERFACE
 #pragma interface			/* gcc class implementation */
 #endif
 
-/*			2	10		1			10			20			10				10			20				20
- * Format: "~*"<db_id><'~' || '_'><tab_id>"-"<blob_id>"-"<auth_code>"-"<server_id>"-"<blob_ref_id>"-"<blob_size>
- */
-//If URL_FMT changes do not forget to update couldBeURL() in this file.
- 
-#define URL_FMT "~*%lu%c%lu-%llu-%lx-%lu-%llu-%llu"
 
 #define MS_SHARED_MEMORY_MAGIC			0x7E9A120C
-#define MS_ENGINE_VERSION				1
-#define MS_CALLBACK_VERSION				4
+#define MS_ENGINE_VERSION				3
+#define MS_CALLBACK_VERSION				6
 #define MS_SHARED_MEMORY_VERSION		2
 #define MS_ENGINE_LIST_SIZE				10
 #define MS_TEMP_FILE_PREFIX				"pbms_temp_"
@@ -80,6 +75,8 @@
 #define MS_ERR_DUPLICATE_DB				14
 #define MS_ERR_DUPLICATE_DB_ID			15
 #define MS_ERR_INVALID_OPERATION		16
+#define MS_ERR_MISSING_CLOUD_REFFERENCE	17
+#define MS_ERR_SYSTAB_VERSION			18
 
 #define MS_LOCK_NONE					0
 #define MS_LOCK_READONLY				1
@@ -94,6 +91,7 @@
 #define MS_RESULT_STACK_SIZE			200
 
 typedef struct PBMSResultRec {
+	uint8_t				mr_had_blobs;							/* A flag to indicate if the statement had any PBMS blobs. */
 	int						mr_code;								/* Engine specific error code. */ 
 	char					mr_message[MS_RESULT_MESSAGE_SIZE];		/* Error message, required if non-zero return code. */
 	char					mr_stack[MS_RESULT_STACK_SIZE];			/* Trace information about where the error occurred. */
@@ -102,15 +100,28 @@
 
 
 typedef struct PBMSBlobID {
-	u_int32_t				bi_db_id;	
-	u_int64_t				bi_blob_size;	
-	u_int64_t				bi_blob_id;				// or repo file offset if type = REPO
-	u_int64_t				bi_blob_ref_id;			
-	u_int32_t				bi_tab_id;				// or repo ID if type = REPO
-	u_int32_t				bi_auth_code;
-	u_int32_t				bi_blob_type;
+	uint32_t				bi_db_id;	
+	uint64_t				bi_blob_size;	
+	uint64_t				bi_blob_id;				// or repo file offset if type = REPO
+	uint64_t				bi_blob_ref_id;			
+	uint32_t				bi_tab_id;				// or repo ID if type = REPO
+	uint32_t				bi_auth_code;
+	uint32_t				bi_blob_type;
 } PBMSBlobIDRec, *PBMSBlobIDPtr;
 
+
+typedef struct MSBlobURL {
+	uint8_t					bu_type;
+	uint32_t				bu_db_id;
+	uint32_t				bu_tab_id;				// or repo ID if type = REPO
+	uint64_t				bu_blob_id;				// or repo file offset if type = REPO
+	uint32_t				bu_auth_code;
+	uint32_t				bu_server_id;
+	uint64_t				bu_blob_size;			
+	uint64_t				bu_blob_ref_id;			// Unique identifier of the blob reference
+} MSBlobURLRec, *MSBlobURLPtr;
+
+
 typedef struct PBMSBlobURL {
 	char					bu_data[PBMS_BLOB_URL_SIZE];
 } PBMSBlobURLRec, *PBMSBlobURLPtr;
@@ -121,8 +132,82 @@
 	int						ms_removing;						/* TRUE (1) if the engine is being removed. */
 	int						ms_internal;						/* TRUE (1) if the engine is supported directly in the mysq/drizzle handler code . */
 	char					ms_engine_name[32];
+	int						ms_has_transactions;				/* TRUE (1) if the engine supports transactions. */
 } PBMSEngineRec, *PBMSEnginePtr;
 
+/*			2	10		1			10			20			10				10			20				20
+ * Format: "~*"<db_id><'~' || '_'><tab_id>"-"<blob_id>"-"<auth_code>"-"<server_id>"-"<blob_ref_id>"-"<blob_size>
+ */
+
+#ifndef  PRIu64
+#define URL_FMT "~*%u%c%u-%llu-%x-%u-%llu-%llu"
+#else
+#define URL_FMT "~*%"PRIu32"%c%"PRIu32"-%"PRIu64"-%x-%"PRIu32"-%"PRIu64"-%"PRIu64""
+#endif
+#define MS_URL_TYPE_BLOB	'~'
+#define MS_URL_TYPE_REPO	'_'
+class PBMSBlobURLTools
+{
+	public:
+	static bool couldBeURL(const char *blob_url, size_t size, MSBlobURLPtr blob)
+	{
+		if (blob_url && (size < PBMS_BLOB_URL_SIZE)) {
+			MSBlobURLRec ignored_blob;
+			char	buffer[PBMS_BLOB_URL_SIZE+1];
+			char	junk[5];
+			int		scanned;
+			
+			if (!blob)
+				blob = &ignored_blob;
+			
+			junk[0] = 0;
+			if (blob_url[size]) { // There is no guarantee that the URL will be null terminated.
+				memcpy(buffer, blob_url, size);
+				buffer[size] = 0;
+				blob_url = buffer;
+			}
+			
+			scanned = sscanf(blob_url, URL_FMT"%4s", 
+				&blob->bu_db_id, 
+				&blob->bu_type, 
+				&blob->bu_tab_id, 
+				&blob->bu_blob_id, 
+				&blob->bu_auth_code, 
+				&blob->bu_server_id, 
+				&blob->bu_blob_ref_id, 
+				&blob->bu_blob_size, 
+				junk);
+				
+			if ((scanned != 8) || (blob->bu_type != MS_URL_TYPE_BLOB && blob->bu_type != MS_URL_TYPE_REPO)) {// If junk is found at the end this will also result in an invalid URL. 
+				//printf("Bad URL \"%s\": scanned = %d, junk: %d, %d, %d, %d\n", blob_url, scanned, junk[0], junk[1], junk[2], junk[3]); 
+				return false;
+			}
+		
+			return true;
+		}
+		
+		return false;
+	}
+	
+	static bool couldBeURL(const char *blob_url, MSBlobURLPtr blob)
+	{
+		return couldBeURL(blob_url, strlen(blob_url), blob);
+	}
+	
+	static void buildBlobURL(MSBlobURLPtr blob, PBMSBlobURLPtr url)
+	{
+		snprintf(url->bu_data, PBMS_BLOB_URL_SIZE, URL_FMT,	blob->bu_db_id, 
+								blob->bu_type, 
+								blob->bu_tab_id, 
+								blob->bu_blob_id, 
+								blob->bu_auth_code, 
+								blob->bu_server_id, 
+								blob->bu_blob_ref_id, 
+								blob->bu_blob_size);
+	}
+};
+
+#ifndef DRIZZLED
 /*
  * This function should never be called directly, it is called
  * by deregisterEngine() below.
@@ -138,7 +223,7 @@
  *
  * The BLOB URL must still be retained or it will automaticly be deleted after a timeout expires.
  */
-typedef int (*ECCreateBlobsFunc)(bool built_in, const char *db_name, const char *tab_name, char *blob, size_t blob_len, char *blob_url, unsigned short col_index, PBMSResultPtr result);
+typedef int (*ECCreateBlobsFunc)(bool built_in, const char *db_name, const char *tab_name, char *blob, size_t blob_len, PBMSBlobURLPtr blob_url, PBMSResultPtr result);
 
 /*
  * Call this function for each BLOB to be retained. When a BLOB is used, the 
@@ -148,7 +233,7 @@
  * The returned URL must be inserted into the row in place of the given
  * URL.
  */
-typedef int (*ECRetainBlobsFunc)(bool built_in, const char *db_name, const char *tab_name, char *ret_blob_url, char *blob_url, unsigned short col_index, PBMSResultPtr result);
+typedef int (*ECRetainBlobsFunc)(bool built_in, const char *db_name, const char *tab_name, PBMSBlobURLPtr ret_blob_url, char *blob_url, unsigned short col_index, PBMSResultPtr result);
 
 /*
  * If a row containing a BLOB is deleted, then the BLOBs in the
@@ -161,7 +246,7 @@
 
 typedef int (*ECDropTable)(bool built_in, const char *db_name, const char *tab_name, PBMSResultPtr result);
 
-typedef int (*ECRenameTable)(bool built_in, const char *db_name, const char *from_table, const char *to_table, PBMSResultPtr result);
+typedef int (*ECRenameTable)(bool built_in, const char *db_name, const char *from_table, const char *to_db, const char *to_table, PBMSResultPtr result);
 
 typedef void (*ECCallCompleted)(bool built_in, bool ok);
 
@@ -245,9 +330,9 @@
 	}
 
 	/*
-	 * Register the engine with the Stream Engine.
+	 * Register the engine with the Stream Daemon.
 	 */
-	int registerEngine(PBMSEnginePtr engine, PBMSResultPtr result) {
+	int registerEngine(PBMSEnginePtr the_engine, PBMSResultPtr result) {
 		int err;
 
 		deleteTempFiles();
@@ -256,8 +341,18 @@
 		if ((err = getSharedMemory(true, result)))
 			return err;
 
+		lock();
 		for (int i=0; i<sharedMemory->sm_list_size; i++) {
 			if (!sharedMemory->sm_engine_list[i]) {
+				PBMSEnginePtr engine;
+				engine = (PBMSEnginePtr) malloc(sizeof(PBMSEngineRec));
+				if (!engine) {
+					strcpy(MS_RESULT_MESSAGE_SIZE, result->mr_message, "Out of memory.");
+					err = MS_ERR_ENGINE;
+					goto done;
+				}
+				memcpy(engine, the_engine, sizeof(PBMSEngineRec));
+				
 				sharedMemory->sm_engine_list[i] = engine;
 				engine->ms_index = i;
 				if (i >= sharedMemory->sm_list_len)
@@ -266,14 +361,20 @@
 					sharedMemory->sm_callbacks->cb_register(engine);
 					
 				built_in = (engine->ms_internal == 1);
-				return MS_OK;
+				err =  MS_OK;
+				goto done;
 			}
 		}
 		
 		result->mr_code = 15010;
 		strcpy(MS_RESULT_MESSAGE_SIZE, result->mr_message, "Too many BLOB streaming engines already registered");
 		*result->mr_stack = 0;
-		return MS_ERR_ENGINE;
+		
+		err = MS_ERR_ENGINE;
+		
+	done:
+		unlock();
+		return err;
 	}
 
 	void lock() {
@@ -292,7 +393,7 @@
 		sharedMemory->sm_shutdown_lock--;
 	}
 
-	void deregisterEngine(PBMSEnginePtr engine) {
+	void deregisterEngine(const char *engine_name) {
 		PBMSResultRec result;
 		int err;
 
@@ -303,10 +404,12 @@
 
 		bool empty = true;
 		for (int i=0; i<sharedMemory->sm_list_len; i++) {
-			if (sharedMemory->sm_engine_list[i]) {
-				if (sharedMemory->sm_engine_list[i] == engine) {
+			PBMSEnginePtr engine = sharedMemory->sm_engine_list[i];
+			if (engine) {				 
+				if (strcmp(engine->ms_engine_name, engine_name) == 0) {
 					if (sharedMemory->sm_callbacks)
 						sharedMemory->sm_callbacks->cb_deregister(engine);
+					free(engine);
 					sharedMemory->sm_engine_list[i] = NULL;
 				}
 				else
@@ -335,51 +438,22 @@
 		sharedMemory = NULL;
 		
 		while (*prefix) {
-			getTempFileName(temp_file, *prefix, getpid());
+			getTempFileName(temp_file, 100, *prefix, getpid());
 			unlink(temp_file);
 			prefix++;
 		}
 	}
 	
-	int couldBeURL(char *blob_url, int size)
+	bool isPBMSLoaded()
 	{
-		if (blob_url && (size < PBMS_BLOB_URL_SIZE)) {
-			char				buffer[PBMS_BLOB_URL_SIZE+1];
-			unsigned long		db_id = 0;
-			unsigned long		tab_id = 0;
-			unsigned long long	blob_id = 0;
-			unsigned long long	blob_ref_id = 0;
-			unsigned long long	blob_size = 0;
-			unsigned long		auth_code = 0;
-			unsigned long		server_id = 0;
-			char				type, junk[5];
-			int					scanned;
-
-			junk[0] = 0;
-			if (blob_url[size]) { // There is no guarantee that the URL will be null terminated.
-				memcpy(buffer, blob_url, size);
-				buffer[size] = 0;
-				blob_url = buffer;
-			}
-			
-			scanned = sscanf(blob_url, URL_FMT"%4s", &db_id, &type, &tab_id, &blob_id, &auth_code, &server_id, &blob_ref_id, &blob_size, junk);
-			if (scanned != 8) {// If junk is found at the end this will also result in an invalid URL. 
-				printf("Bad URL \"%s\": scanned = %d, junk: %d, %d, %d, %d\n", blob_url, scanned, junk[0], junk[1], junk[2], junk[3]); 
-				return 0;
-			}
-			
-			if (junk[0] || (type != '~' && type != '_')) {
-				printf("Bad URL \"%s\": scanned = %d, junk: %d, %d, %d, %d\n", blob_url, scanned, junk[0], junk[1], junk[2], junk[3]); 
-				return 0;
-			}
-		
-			return 1;
-		}
-		
-		return 0;
+		PBMSResultRec result;
+		if (getSharedMemory(false, &result))
+			return false;
+			
+		return (sharedMemory->sm_callbacks != NULL);
 	}
 	
-	int  retainBlob(const char *db_name, const char *tab_name, char *ret_blob_url, char *blob_url, size_t blob_size, unsigned short col_index, PBMSResultPtr result)
+	int  retainBlob(const char *db_name, const char *tab_name, PBMSBlobURLPtr ret_blob_url, char *blob_url, size_t blob_size, unsigned short col_index, PBMSResultPtr result)
 	{
 		int err;
 		char safe_url[PBMS_BLOB_URL_SIZE+1];
@@ -388,17 +462,17 @@
 		if ((err = getSharedMemory(false, result)))
 			return err;
 
-		if (!couldBeURL(blob_url, blob_size)) {
+		if (!PBMSBlobURLTools::couldBeURL(blob_url, blob_size, NULL)) {
 		
 			if (!sharedMemory->sm_callbacks)  {
-				*ret_blob_url = 0;
+				ret_blob_url->bu_data[0] = 0;
 				return MS_OK;
 			}
-			err = sharedMemory->sm_callbacks->cb_create_blob(built_in, db_name, tab_name, blob_url, blob_size, ret_blob_url, col_index, result);
+			err = sharedMemory->sm_callbacks->cb_create_blob(built_in, db_name, tab_name, blob_url, blob_size, ret_blob_url, result);
 			if (err)
 				return err;
 				
-			blob_url = ret_blob_url;
+			blob_url = ret_blob_url->bu_data;
 		} else {
 			// Make sure the url is a C string:
 			if (blob_url[blob_size]) {
@@ -411,7 +485,7 @@
 
 		if (!sharedMemory->sm_callbacks) {
 			result->mr_code = MS_ERR_INCORRECT_URL;
-			strcpy(MS_RESULT_MESSAGE_SIZE, result->mr_message, "BLOB streaming engine (PBMS) not installed");
+			strcpy(MS_RESULT_MESSAGE_SIZE, result->mr_message, "BLOB streaming daemon (PBMS) not installed");
 			*result->mr_stack = 0;
 			return MS_ERR_INCORRECT_URL;
 		}
@@ -430,7 +504,7 @@
 		if (!sharedMemory->sm_callbacks)
 			return MS_OK;
 
-		if (!couldBeURL(blob_url, blob_size))
+		if (!PBMSBlobURLTools::couldBeURL(blob_url, blob_size, NULL))
 			return MS_OK;
 
 		if (blob_url[blob_size]) {
@@ -455,7 +529,7 @@
 		return sharedMemory->sm_callbacks->cb_drop_table(built_in, db_name, tab_name, result);
 	}
 
-	int renameTable(const char *db_name, const char *from_table, const char *to_table, PBMSResultPtr result)
+	int renameTable(const char *db_name, const char *from_table, const char *to_db, const char *to_table, PBMSResultPtr result)
 	{
 		int err;
 
@@ -465,7 +539,7 @@
 		if (!sharedMemory->sm_callbacks)
 			return MS_OK;
 			
-		return sharedMemory->sm_callbacks->cb_rename_table(built_in, db_name, from_table, to_table, result);
+		return sharedMemory->sm_callbacks->cb_rename_table(built_in, db_name, from_table, to_db, to_table, result);
 	}
 
 	void completed(int ok)
@@ -495,7 +569,7 @@
 			return MS_OK;
 
 		while (*prefix) {
-			getTempFileName(temp_file, *prefix, getpid());
+			getTempFileName(temp_file, 100, *prefix, getpid());
 			tmp_f = open(temp_file, O_RDWR | (create ? O_CREAT : 0), SH_MASK);
 			if (tmp_f == -1)
 				return setOSResult(errno, "open", temp_file, result);
@@ -515,7 +589,7 @@
 			}
 
 			buffer[tfer] = 0;
-			sscanf(buffer, "%p", &sharedMemory);
+			sscanf(buffer, "%p", (void**) &sharedMemory);
 			if (!sharedMemory || sharedMemory->sm_magic != MS_SHARED_MEMORY_MAGIC) {
 				if (!create)
 					return MS_OK;
@@ -531,9 +605,9 @@
 					return setOSResult(errno, "fseek", temp_file, result);
 				}
 
-				sprintf(buffer, "%p", sharedMemory);
+				snprintf(buffer, 100, "%p", (void*) sharedMemory);
 				tfer = write(tmp_f, buffer, strlen(buffer));
-				if (tfer != strlen(buffer)) {
+				if (tfer != (ssize_t) strlen(buffer)) {
 					close(tmp_f);
 					return setOSResult(errno, "write", temp_file, result);
 				}
@@ -583,9 +657,9 @@
 
 	void strcat(size_t size, char *to, int val)
 	{
-		char buffer[100];
+		char buffer[20];
 
-		sprintf(buffer, "%d", val);
+		snprintf(buffer, 20, "%d", val);
 		strcat(size, to, buffer);
 	}
 
@@ -634,9 +708,9 @@
 		return MS_ERR_ENGINE;
 	}
 
-	void getTempFileName(char *temp_file, const char * prefix, int pid)
+	void getTempFileName(char *temp_file, int buffer_size, const char * prefix, int pid)
 	{
-		sprintf(temp_file, "/tmp/%s%d", prefix,  pid);
+		snprintf(temp_file, buffer_size, "/tmp/%s%d", prefix,  pid);
 	}
 
 	bool startsWith(const char *cstr, const char *w_cstr)
@@ -652,43 +726,34 @@
 
 	void deleteTempFiles()
 	{
-		struct dirent	*entry;
+		struct dirent	entry;
 		struct dirent	*result;
 		DIR				*odir;
 		int				err;
-		size_t			sz;
 		char			temp_file[100];
 
-#ifdef __sun
-		sz = sizeof(struct dirent) + pathconf("/tmp/", _PC_NAME_MAX); // Solaris, see readdir(3C)
-#else
-		sz = sizeof(struct dirent);
-#endif
-		if (!(entry = (struct dirent *) malloc(sz)))
-			return;
 		if (!(odir = opendir("/tmp/")))
 			return;
-		err = readdir_r(odir, entry, &result);
+		err = readdir_r(odir, &entry, &result);
 		while (!err && result) {
 			const char **prefix = temp_prefix;
 			
 			while (*prefix) {
-				if (startsWith(entry->d_name, *prefix)) {
-					int pid = atoi(entry->d_name + strlen(*prefix));
+				if (startsWith(entry.d_name, *prefix)) {
+					int pid = atoi(entry.d_name + strlen(*prefix));
 					
 					/* If the process does not exist: */
 					if (kill(pid, 0) == -1 && errno == ESRCH) {
-						getTempFileName(temp_file, *prefix, pid);
+						getTempFileName(temp_file, 100, *prefix, pid);
 						unlink(temp_file);
 					}
 				}
 				prefix++;
 			}
 			
-			err = readdir_r(odir, entry, &result);
+			err = readdir_r(odir, &entry, &result);
 		}
 		closedir(odir);
-		free(entry);
 	}
 };
 #endif // PBMS_API
@@ -741,5 +806,7 @@
 * PBMSIDToURL():Convert a blob URL to a blob ID.
 */
 extern bool PBMSURLToID(char *url, PBMSBlobIDPtr blob_id);
-
-#endif
+#endif //DRIZZLED 
+
+
+#endif //__PBMS_H__

=== modified file 'src/pbms_enabled.cc'
--- src/pbms_enabled.cc	2010-05-04 16:23:07 +0000
+++ src/pbms_enabled.cc	2010-07-21 22:08:48 +0000
@@ -22,113 +22,205 @@
  *
  * H&G2JCtL
  *
- * PBMS interface used to enable engines for use with the PBMS engine.
+ * PBMS interface used to enable engines for use with the PBMS daemon.
  *
  * For an example on how to build this into an engine have a look at the PBXT engine
  * in file ha_pbxt.cc. Search for 'PBMS_ENABLED'.
  *
  */
 
-#include "xt_config.h"
-
-#ifdef PBMS_ENABLED
-
-#ifdef DRIZZLED
-#include <sys/stat.h>
-#include <drizzled/common_includes.h>
-#include <drizzled/plugin.h>
+#ifndef DRIZZLED
+#if defined(MSDOS) || defined(__WIN__)
+#include "pbms_enabled.h"
+
+// Windows is not supported yet so just stub out the functions..
+bool pbms_initialize(const char *engine_name __attribute__((unused)), 
+					bool isServer __attribute__((unused)), 
+					bool isTransactional __attribute__((unused)), 
+					PBMSResultPtr result __attribute__((unused)), 
+					IsPBMSFilterFunc is_pbms_blob __attribute__((unused))
+					) { return true;}
+void pbms_finalize() {}
+int pbms_write_row_blobs(const TABLE *table __attribute__((unused)), 
+						unsigned char *buf __attribute__((unused)), 
+						PBMSResultPtr result __attribute__((unused))
+						){ return 0;}
+int pbms_update_row_blobs(const TABLE *table __attribute__((unused)), 
+						const unsigned char *old_row __attribute__((unused)), 
+						unsigned char *new_row __attribute__((unused)), 
+						PBMSResultPtr result __attribute__((unused))
+						){ return 0;}
+int pbms_delete_row_blobs(const TABLE *table __attribute__((unused)), 
+						const unsigned char *buf __attribute__((unused)), 
+						PBMSResultPtr result __attribute__((unused))
+						){ return 0;}
+int pbms_rename_table_with_blobs(const char *old_table_path __attribute__((unused)), 
+								const char *new_table_path __attribute__((unused)), 
+								PBMSResultPtr result __attribute__((unused))
+								){ return 0;}
+int pbms_delete_table_with_blobs(const char *table_path __attribute__((unused)), 
+								PBMSResultPtr result __attribute__((unused))
+								){ return 0;}
+void pbms_completed(TABLE *table __attribute__((unused)), 
+					bool ok __attribute__((unused))
+					){}
 #else
+#define PBMS_API	pbms_enabled_api
+
+#include "pbms_enabled.h"
 #include "mysql_priv.h"
 #include <mysql/plugin.h>
 #define session_alloc(sess, size) thd_alloc(sess, size);
 #define current_session current_thd
-#endif
-
-#define GET_BLOB_FIELD(t, i)	(Field_blob *)(t->field[t->s->blob_field[i]])
-#define DB_NAME(f)				(f->table->s->db.str)
-#define TAB_NAME(f)				(*(f->table_name))
-
-#define PBMS_API	pbms_enabled_api
-
-#include "pbms_enabled.h"
+
+#define GET_BLOB_FIELD(t, i) (Field_blob *)(t->field[t->s->blob_field[i]])
+#define DB_NAME(f) (f->table->s->db.str)
+#define TAB_NAME(f) (*(f->table_name))
 
 static PBMS_API pbms_api;
 
-PBMSEngineRec enabled_engine = {
-	MS_ENGINE_VERSION
-};
+/* 
+ * A callback function to check if the column is a PBMS BLOB. 
+ * Can be NULL if no check is to be done. 
+ */ 
+static IsPBMSFilterFunc	is_pbms_blob = NULL; 
 
 //====================
-bool pbms_initialize(const char *engine_name, bool isServer, PBMSResultPtr result)
+bool pbms_initialize(const char *engine_name, bool isServer, bool isTransactional, PBMSResultPtr result, IsPBMSFilterFunc is_pbms_blob_arg)
 {
-	int						err;
+	int	err;
+	PBMSEngineRec enabled_engine = {
+		MS_ENGINE_VERSION,
+		0,
+		0,
+		0,
+		{0},	
+		0
+	};
 
 	strncpy(enabled_engine.ms_engine_name, engine_name, 32);
 	enabled_engine.ms_internal = isServer;
+	enabled_engine.ms_has_transactions = isTransactional;
 	enabled_engine.ms_engine_name[31] = 0;
 
 	err = pbms_api.registerEngine(&enabled_engine, result);
+	is_pbms_blob = is_pbms_blob_arg;
 
 	return (err == 0);
 }
 
 
 //====================
-void pbms_finalize()
-{
-	pbms_api.deregisterEngine(&enabled_engine);
+void pbms_finalize(const char *engine_name)
+{
+	pbms_api.deregisterEngine(engine_name);
+}
+
+//==================================
+static int insertRecord(Field_blob *field, char *blob,  size_t org_length, unsigned char *blob_rec, size_t packlength, PBMSResultPtr result)
+{
+	int err;
+	size_t length;
+	PBMSBlobURLRec blob_url;
+	
+	err = pbms_api.retainBlob(DB_NAME(field), TAB_NAME(field), &blob_url, blob, org_length, field->field_index, result);
+	if (err)
+		return err;
+		
+	// If the BLOB length changed reset it. 
+	// This will happen if the BLOB data was replaced with a BLOB reference. 
+	length = strlen(blob_url.bu_data)  +1;
+	if ((length != org_length) || memcmp(blob_url.bu_data, blob, length)) {
+		if (length != org_length) {
+			field->store_length(blob_rec, packlength, length);
+		}
+		
+		if (length > org_length) {
+			// This can only happen if the BLOB URL is actually larger than the BLOB itself.
+			blob = (char *) session_alloc(current_session, length);
+			memcpy(blob_rec+packlength, &blob, sizeof(char*));
+		}			
+		memcpy(blob, blob_url.bu_data, length);
+	} 
+		
+	return 0;
 }
 
 //====================
-int pbms_write_row_blobs(TABLE *table, uchar *row_buffer, PBMSResultPtr result)
+int pbms_update_row_blobs(const TABLE *table, const unsigned char *old_row, unsigned char *new_row, PBMSResultPtr result)
 {
 	Field_blob *field;
-	char *blob_rec, *blob;
-	size_t packlength, i, org_length, length;
-	char blob_url_buffer[PBMS_BLOB_URL_SIZE];
+	uint32_t field_offset;
+	const unsigned char *old_blob_rec;
+	unsigned char *new_blob_rec;
+	char *old_blob_url, *new_blob_url;
+	size_t packlength, i, old_length, new_length;
 	int err;
-	String type_name;
+	bool old_null_blob, new_null_blob;
 
+	result->mr_had_blobs = false;
+	
+	if (!pbms_api.isPBMSLoaded())
+		return 0;
+		
 	if (table->s->blob_fields == 0)
 		return 0;
 		
 	for (i= 0; i < table->s->blob_fields; i++) {
 		field = GET_BLOB_FIELD(table, i);
 
-		// Note: field->type() always returns MYSQL_TYPE_BLOB regardless of the type of BLOB
-		field->sql_type(type_name);
-		if (strcasecmp(type_name.c_ptr(), "LongBlob"))
-			continue;
+		old_null_blob = field->is_null_in_record(old_row);
+		new_null_blob = field->is_null_in_record(new_row);
+		if (old_null_blob && new_null_blob)
+			continue;
+
+		{
+			String type_name;
+			// Note: field->type() always returns MYSQL_TYPE_BLOB regardless of the type of BLOB
+			field->sql_type(type_name);
+			if (strcasecmp(type_name.c_ptr(), "LongBlob"))
+				continue;
+		}
+			
+		if( is_pbms_blob && !is_pbms_blob(field) )
+			continue;
+			
 			
 		// Get the blob record:
-		blob_rec = (char *)row_buffer + field->offset(field->table->record[0]);
+		field_offset = field->offset(field->table->record[0]);
 		packlength = field->pack_length() - field->table->s->blob_ptr_size;
 
-		memcpy(&blob, blob_rec +packlength, sizeof(char*));
-		org_length = field->get_length((uchar *)blob_rec);
+		if (new_null_blob) {
+			new_blob_url = NULL;
+		} else {
+			new_blob_rec = new_row + field_offset;
+			new_length = field->get_length(new_blob_rec);
+			memcpy(&new_blob_url, new_blob_rec +packlength, sizeof(char*));
+		}
+		
+		if (old_null_blob) {
+			old_blob_url = NULL;
+		} else {
+			old_blob_rec = old_row + field_offset;
+			old_length = field->get_length(old_blob_rec);
+			memcpy(&old_blob_url, old_blob_rec +packlength, sizeof(char*));
+		}
+		
+		// Check to see if the BLOBs are the same.
+		// I am assuming that if the BLOB pointer is different then teh BLOB has changed.
+		// Zero length BLOBs are a special case because they may have a NULL data pointer,
+		// to catch this and distiguish it from a NULL BLOB I do a check to see if one field was NULL:
+		// (old_null_blob != new_null_blob)
+		if ((old_blob_url != new_blob_url) || (old_null_blob != new_null_blob)) {
+			
+			result->mr_had_blobs = true;
 
-		
-		// Signal PBMS to record a new reference to the BLOB.
-		// If 'blob' is not a BLOB URL then it will be stored in the repositor as a new BLOB
-		// and a reference to it will be created.
-		err = pbms_api.retainBlob(DB_NAME(field), TAB_NAME(field), blob_url_buffer, blob, org_length, field->field_index, result);
-		if (err)
-			return err;
-			
-		// If the BLOB length changed reset it. 
-		// This will happen if the BLOB data was replaced with a BLOB reference. 
-		length = strlen(blob_url_buffer)  +1;
-		if ((length != org_length) || memcmp(blob_url_buffer, blob, length)) {
-			if (length != org_length) {
-				field->store_length((uchar *)blob_rec, packlength, length);
-			}
-			
-			if (length > org_length) {
-				// This can only happen if the BLOB URL is actually larger than the BLOB itself.
-				blob = (char *) session_alloc(current_session, length);
-				memcpy(blob_rec+packlength, &blob, sizeof(char*));
-			}			
-			memcpy(blob, blob_url_buffer, length);
+			// The BLOB was updated so delete the old one and insert the new one.
+			if ((old_null_blob == false) && (err = pbms_api.releaseBlob(DB_NAME(field), TAB_NAME(field), old_blob_url, old_length, result)))
+				return err;
+				
+			if ((new_null_blob == false) && (err = insertRecord(field, new_blob_url, new_length, new_blob_rec, packlength, result)))
+				return err;
 		} 
 	}
 	
@@ -136,33 +228,100 @@
 }
 
 //====================
-int pbms_delete_row_blobs(TABLE *table, const uchar *row_buffer, PBMSResultPtr result)
-{
-	Field_blob *field;
-	const char *blob_rec;
+int pbms_write_row_blobs(const TABLE *table, unsigned char *row_buffer, PBMSResultPtr result)
+{
+
+	Field_blob *field;
+	unsigned char *blob_rec;
+	char *blob_url;
+	size_t packlength, i, length;
+	int err;
+
+	result->mr_had_blobs = false;
+
+	if (!pbms_api.isPBMSLoaded())
+		return 0;
+		
+	if (table->s->blob_fields == 0)
+		return 0;
+		
+	for (i= 0; i <  table->s->blob_fields; i++) {
+		field =  GET_BLOB_FIELD(table, i);
+		
+		if (field->is_null_in_record(row_buffer))
+			continue;
+			
+		{
+			String type_name;
+			// Note: field->type() always returns MYSQL_TYPE_BLOB regardless of the type of BLOB
+			field->sql_type(type_name);
+			if (strcasecmp(type_name.c_ptr(), "LongBlob"))
+				continue;
+		}
+			
+		if( is_pbms_blob && !is_pbms_blob(field) )
+			continue;
+
+		result->mr_had_blobs = true;
+
+		// Get the blob record:
+		packlength = field->pack_length() - field->table->s->blob_ptr_size;
+		blob_rec = row_buffer + field->offset(field->table->record[0]);
+		
+		length = field->get_length(blob_rec);
+		memcpy(&blob_url, blob_rec +packlength, sizeof(char*));
+
+		if ((err = insertRecord(field, blob_url, length, blob_rec, packlength, result)))
+			return err;
+	}
+
+	return 0;
+}
+
+//====================
+int pbms_delete_row_blobs(const TABLE *table, const unsigned char *row_buffer, PBMSResultPtr result)
+{
+	Field_blob *field;
+	const unsigned char *blob_rec;
 	char *blob;
 	size_t packlength, i, length;
+	bool call_failed = false;
 	int err;
-	String type_name;
+	
+	result->mr_had_blobs = false;
 
+	if (!pbms_api.isPBMSLoaded())
+		return 0;
+		
 	if (table->s->blob_fields == 0)
 		return 0;
 		
 	for (i= 0; i < table->s->blob_fields; i++) {
 		field = GET_BLOB_FIELD(table, i);
+		
+		if (field->is_null_in_record(row_buffer))
+			continue;
+			
+		{
+			String type_name;
+			// Note: field->type() always returns MYSQL_TYPE_BLOB regardless of the type of BLOB
+			field->sql_type(type_name);
+			if (strcasecmp(type_name.c_ptr(), "LongBlob"))
+				continue;
+		}
+			
+		if(is_pbms_blob && !is_pbms_blob(field) )
+			continue;
 
-		// Note: field->type() always returns MYSQL_TYPE_BLOB regardless of the type of BLOB
-		field->sql_type(type_name);
-		if (strcasecmp(type_name.c_ptr(), "LongBlob"))
-			continue;
-			
+		result->mr_had_blobs = true;	
+		
 		// Get the blob record:
-		blob_rec = (char *)row_buffer + field->offset(field->table->record[0]);
 		packlength = field->pack_length() - field->table->s->blob_ptr_size;
 
-		length = field->get_length((uchar *)blob_rec);
+		blob_rec = row_buffer + field->offset(field->table->record[0]);
+		length = field->get_length(blob_rec);
 		memcpy(&blob, blob_rec +packlength, sizeof(char*));
-		
+
 		// Signal PBMS to delete the reference to the BLOB.
 		err = pbms_api.releaseBlob(DB_NAME(field), TAB_NAME(field), blob, length, result);
 		if (err)
@@ -208,18 +367,16 @@
 {
 	char o_db_name[MAX_NAME_SIZE], n_db_name[MAX_NAME_SIZE], o_tab_name[MAX_NAME_SIZE], n_tab_name[MAX_NAME_SIZE];
 
+	result->mr_had_blobs = false; 
+	if (!pbms_api.isPBMSLoaded())
+		return 0;
+		
+	result->mr_had_blobs = true; // Assume it has blobs.
+	
 	parse_table_path(old_table_path, o_db_name, o_tab_name);
 	parse_table_path(new_table_path, n_db_name, n_tab_name);
 	
-	if (strcmp(o_db_name, n_db_name)) {
-		result->mr_code = MS_ERR_INVALID_OPERATION;
-		strcpy(result->mr_message, "PBMS does not support renaming tables across databases.");
-		strcpy(result->mr_stack, "pbms_rename_table_with_blobs()");
-		return MS_ERR_INVALID_OPERATION;
-	}
-	
-	
-	 return pbms_api.renameTable(o_db_name, o_tab_name, n_tab_name, result);
+	return pbms_api.renameTable(o_db_name, o_tab_name, n_db_name, n_tab_name, result);
 }
 
 //====================
@@ -227,23 +384,26 @@
 {
 	char db_name[MAX_NAME_SIZE], tab_name[MAX_NAME_SIZE];
 		
+	result->mr_had_blobs = false; 
+	if (!pbms_api.isPBMSLoaded())
+		return 0;
+		
+	result->mr_had_blobs = true; // Assume it has blobs.
 	parse_table_path(table_path, db_name, tab_name);
 
 	return pbms_api.dropTable(db_name, tab_name, result);
 }
 
 //====================
-void pbms_completed(TABLE *table, bool ok)
+void pbms_completed(const TABLE *table, bool ok)
 {
+	if (!pbms_api.isPBMSLoaded())
+		return;
+		
 	if ((!table) || (table->s->blob_fields != 0))
 		pbms_api.completed(ok) ;
 		
 	 return ;
 }
-
-#elif defined(__WIN__)
-
-// Remove linker warning 4221 about empty file
-namespace { char dummy; };
-
-#endif // PBMS_ENABLED
+#endif
+#endif // DRIZZLED

=== modified file 'src/pbms_enabled.h'
--- src/pbms_enabled.h	2009-09-16 06:39:30 +0000
+++ src/pbms_enabled.h	2010-07-21 22:08:48 +0000
@@ -22,7 +22,7 @@
  *
  * H&G2JCtL
  *
- * PBMS interface used to enable engines for use with the PBMS engine.
+ * PBMS interface used to enable engines for use with the PBMS daemon.
  *
  * For an example on how to build this into an engine have a look at the PBXT engine
  * in file ha_pbxt.cc. Search for 'PBMS_ENABLED'.
@@ -35,6 +35,17 @@
 
 #include "pbms.h"
 
+#ifdef DRIZZLED
+#include <drizzled/common.h>
+#define TABLE Table
+#define uchar unsigned char
+#else
+#include <mysql_priv.h>
+#endif
+
+class Field;
+typedef bool (*IsPBMSFilterFunc)(Field *field);
+
 /*
  * pbms_initialize() should be called from the engines plugIn's 'init()' function.
  * The engine_name is the name of your engine, "PBXT" or "InnoDB" for example.
@@ -43,31 +54,38 @@
  * true if this is being built into the server's handler code above the engine level
  * calls. 
  */
-extern bool pbms_initialize(const char *engine_name, bool isServer, PBMSResultPtr result);
+extern bool pbms_initialize(const char *engine_name, bool isServer, bool isTransactional, PBMSResultPtr result, IsPBMSFilterFunc is_pbms_blob);
 
 /*
  * pbms_finalize() should be called from the engines plugIn's 'deinit()' function.
  */
-extern void pbms_finalize();
+extern void pbms_finalize(const char *engine_name);
 
 /*
  * pbms_write_row_blobs() should be called from the engine's 'write_row' function.
  * It can alter the row data so it must be called before any other function using the row data.
- * It should also be called from engine's 'update_row' function for the new row.
- *
- * pbms_completed() must be called after calling pbms_write_row_blobs() and just before
- * returning from write_row() to indicate if the operation completed successfully.
- */
-extern int pbms_write_row_blobs(TABLE *table, uchar *buf, PBMSResultPtr result);
+ *
+ * pbms_completed() must be called after calling pbms_write_row_blobs() and just before
+ * returning from write_row() to indicate if the operation completed successfully.
+ */
+extern int pbms_write_row_blobs(const TABLE *table, unsigned char *buf, PBMSResultPtr result);
+
+/*
+ * pbms_update_row_blobs() should be called from the engine's 'update_row' function.
+ * It can alter the row data so it must be called before any other function using the row data.
+ *
+ * pbms_completed() must be called after calling pbms_write_row_blobs() and just before
+ * returning from write_row() to indicate if the operation completed successfully.
+ */
+extern int pbms_update_row_blobs(const TABLE *table, const unsigned char *old_row, unsigned char *new_row, PBMSResultPtr result);
 
 /*
  * pbms_delete_row_blobs() should be called from the engine's 'delete_row' function.
- * It should also be called from engine's 'update_row' function for the old row.
  *
  * pbms_completed() must be called after calling pbms_delete_row_blobs() and just before
  * returning from delete_row() to indicate if the operation completed successfully.
  */
-extern int pbms_delete_row_blobs(TABLE *table, const uchar *buf, PBMSResultPtr result);
+extern int pbms_delete_row_blobs(const TABLE *table, const unsigned char *buf, PBMSResultPtr result);
 
 /*
  * pbms_rename_table_with_blobs() should be called from the engine's 'rename_table' function.
@@ -98,6 +116,6 @@
  * pbms_completed() has the effect of committing or rolling back the changes made if the session
  * is in 'autocommit' mode.
  */
-extern void pbms_completed(TABLE *table, bool ok);
+extern void pbms_completed(const TABLE *table, bool ok);
 
 #endif