Add support for Gemini table handler, Monty has checked and approved

Fix bug when read return error


acconfig.h:
  Add Gemini to configure
acinclude.m4:
  Add Gemini to configure
include/my_base.h:
  Add error codes for tables handlers
mysql-test/t/select.test:
  Force temporary tables to MyISAM
sql-bench/server-cfg.sh:
  Allow Gemini to run sql-bench
sql/Makefile.am:
  Add Gemini to configure
sql/handler.cc:
  Add support for Gemini table handler
sql/handler.h:
  Add support for Gemini table handler
sql/lex.h:
  Add support for Gemini table handler
sql/mysqld.cc:
  Add support for Gemini table handler
sql/opt_range.cc:
  Fix bug when read return error
sql/records.cc:
  Fix bug when read return error
sql/sql_class.cc:
  Add support for Gemini table handler
sql/sql_class.h:
  Add support for Gemini table handler
sql/sql_lex.h:
  Add support for Gemini table handler
sql/sql_rename.cc:
  Add commit for table rename
sql/sql_table.cc:
  Add commit for table rename
BitKeeper/etc/logging_ok:
  Logging to logging@openlogging.org accepted

1 files changed, 2685 insertions, 0 deletions

diff --git a/sql/ha_gemini.cc b/sql/ha_gemini.cc
new file mode 100644
index 00000000000..863a0d6931c
--- /dev/null
+++ b/sql/ha_gemini.cc
@@ -0,0 +1,2685 @@
+/* Copyright (C) 2000 NuSphere Corporation
+   
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; either version 2 of the License, or
+   (at your option) any later version.
+   
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+   
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
+
+
+#ifdef __GNUC__
+#pragma implementation				// gcc: Class implementation
+#endif
+
+#include <string.h>
+
+#include "mysql_priv.h"
+#include "my_pthread.h"
+
+#ifdef HAVE_GEMINI_DB
+#include "dbconfig.h"
+#include "dsmpub.h"
+#include "recpub.h"
+#include "vststat.h"
+
+#include <m_ctype.h>
+#include <myisampack.h>
+#include <assert.h>
+#include <hash.h>
+#include <stdarg.h>
+#include "geminikey.h"
+#include "ha_gemini.h"
+
+const char *ha_gemini_ext=".gmd";
+const char *ha_gemini_idx_ext=".gmi";
+
+bool gemini_skip=0;
+long gemini_options = 0;
+long gemini_buffer_cache;
+long gemini_io_threads;
+long gemini_log_cluster_size;
+long gemini_locktablesize;
+long gemini_lock_wait_timeout;
+long gemini_spin_retries;
+long gemini_connection_limit;
+
+const char gemini_dbname[] = "gemini";
+dsmContext_t *pfirstContext = NULL;
+
+ulong gemini_recovery_options = GEMINI_RECOVERY_FULL;
+/* bits in gemini_recovery_options */
+const char *gemini_recovery_names[] =
+{ "FULL", "NONE", "FORCE" };
+TYPELIB gemini_recovery_typelib= {array_elements(gemini_recovery_names),"",
+				 gemini_recovery_names};
+
+const int start_of_name = 2;  /* Name passed as ./<db>/<table-name>
+                               and we're not interested in the ./ */
+static const int keyBufSize = MYMAXKEYSIZE * 2;
+
+static int gemini_tx_begin(THD *thd);
+static void print_msg(THD *thd, const char *table_name, const char *op_name,
+                      const char *msg_type, const char *fmt, ...);
+
+static int gemini_helper_threads(dsmContext_t *pContext);
+pthread_handler_decl(gemini_watchdog,arg );
+pthread_handler_decl(gemini_rl_writer,arg );
+pthread_handler_decl(gemini_apw,arg);
+
+/* General functions */
+
+bool gemini_init(void)
+{
+    dsmStatus_t     rc = 0;
+    char            pmsgsfile[MAXPATHN];
+
+    DBUG_ENTER("gemini_init");
+
+    /* If datadir isn't set, bail out */
+    if (*mysql_real_data_home == '\0')
+    {
+        goto badret;
+    }
+
+    /* Gotta connect to the database regardless of the operation */
+    rc = dsmContextCreate(&pfirstContext);
+    if( rc != 0 )
+    {
+        printf("dsmContextCreate failed %ld\n",rc);
+        goto badret;
+    }
+    rc = dsmContextSetString(pfirstContext, DSM_TAGDB_DBNAME,
+                             strlen(gemini_dbname), (TEXT *)gemini_dbname);
+    if( rc != 0 )
+    {
+        printf("Dbname tag failed %ld\n", rc);
+        goto badret;
+    }
+
+    fn_format(pmsgsfile, GEM_MSGS_FILE, language, ".db", 2 | 4);
+    rc = dsmContextSetString(pfirstContext, DSM_TAGDB_MSGS_FILE,
+                             strlen(pmsgsfile), (TEXT *)pmsgsfile);
+    if( rc != 0 )
+    {
+        printf("MSGS_DIR tag failed %ld\n", rc);
+        goto badret;
+    }
+
+    rc = dsmContextSetLong(pfirstContext,DSM_TAGDB_ACCESS_TYPE,DSM_ACCESS_STARTUP);
+    if ( rc != 0 )
+    {
+        printf("ACCESS TAG set failed %ld\n",rc);
+        goto badret;
+    }
+    rc = dsmContextSetLong(pfirstContext,DSM_TAGDB_ACCESS_ENV, DSM_SQL_ENGINE);
+    if( rc != 0 )
+    {
+        printf("ACCESS_ENV set failed %ld",rc);
+        goto badret;
+    }
+
+    rc = dsmContextSetString(pfirstContext, DSM_TAGDB_DATADIR,
+                             strlen(mysql_real_data_home),
+                             (TEXT *)mysql_real_data_home);
+    if( rc != 0 )
+    {
+        printf("Datadir tag failed %ld\n", rc);
+        goto badret;
+    }
+
+    rc = dsmContextSetLong(pfirstContext, DSM_TAGDB_MAX_USERS,
+                           gemini_connection_limit);
+    if(rc != 0)
+    {
+        printf("MAX_USERS tag set failed %ld",rc);
+        goto badret;
+    }
+
+    rc = dsmContextSetLong(pfirstContext, DSM_TAGDB_DEFAULT_LOCK_TIMEOUT,
+                           gemini_lock_wait_timeout);
+    if(rc != 0)
+    {
+        printf("MAX_LOCK_ENTRIES tag set failed %ld",rc);
+        goto badret;
+    }
+
+    rc = dsmContextSetLong(pfirstContext, DSM_TAGDB_MAX_LOCK_ENTRIES,
+                           gemini_locktablesize);
+    if(rc != 0)
+    {
+        printf("MAX_LOCK_ENTRIES tag set failed %ld",rc);
+        goto badret;
+    }
+
+    rc = dsmContextSetLong(pfirstContext, DSM_TAGDB_SPIN_AMOUNT,
+                           gemini_spin_retries);
+    if(rc != 0)
+    {
+        printf("SPIN_AMOUNT tag set failed %ld",rc);
+        goto badret;
+    }
+
+    /* blocksize is hardcoded to 8K.  Buffer cache is in bytes
+       need to convert this to 8K blocks */
+    gemini_buffer_cache = gemini_buffer_cache / 8192;
+    rc = dsmContextSetLong(pfirstContext, DSM_TAGDB_DB_BUFFERS,
+                           gemini_buffer_cache);
+    if(rc != 0)
+    {
+        printf("DB_BUFFERS tag set failed %ld",rc);
+        goto badret;
+    }
+
+    rc = dsmContextSetLong(pfirstContext, DSM_TAGDB_FLUSH_AT_COMMIT,
+                           ((gemini_options & GEMOPT_FLUSH_LOG) ? 1 : 0));
+    if(rc != 0)
+    {
+        printf("FLush_Log_At_Commit tag set failed %ld",rc);
+        goto badret;
+    }
+    rc = dsmContextSetLong(pfirstContext, DSM_TAGDB_DIRECT_IO,
+                           ((gemini_options & GEMOPT_UNBUFFERED_IO) ? 1 : 0));
+    if(rc != 0)
+    {
+        printf("DIRECT_IO tag set failed %ld",rc);
+        goto badret;
+    }
+
+    rc = dsmContextSetLong(pfirstContext, DSM_TAGDB_CRASH_PROTECTION,
+                      ((gemini_recovery_options & GEMINI_RECOVERY_FULL) ? 1 : 0));
+    if(rc != 0)
+    {
+        printf("CRASH_PROTECTION tag set failed %ld",rc);
+        goto badret;
+    }
+
+    /* cluster size will come in bytes, need to convert it to 
+       16 K units. */
+    gemini_log_cluster_size = (gemini_log_cluster_size + 16383) / 16384;
+    rc = dsmContextSetLong(pfirstContext, DSM_TAGDB_BI_CLUSTER_SIZE,
+                           gemini_log_cluster_size);
+                      
+    if(rc != 0)
+    {
+        printf("CRASH_PROTECTION tag set failed %ld",rc);
+        goto badret;
+    }
+
+    rc = dsmUserConnect(pfirstContext,(TEXT *)"Multi-user",
+                        DSM_DB_OPENDB | DSM_DB_OPENFILE);
+    if( rc != 0 )
+    {
+        printf("dsmUserConnect failed rc = %ld\n",rc);
+        goto badret;
+    }
+    /* Set access to shared for subsequent user connects    */
+    rc = dsmContextSetLong(pfirstContext,DSM_TAGDB_ACCESS_TYPE,DSM_ACCESS_SHARED);
+    rc = gemini_helper_threads(pfirstContext);
+    DBUG_RETURN(0);
+
+badret:
+    gemini_skip = 1;
+    DBUG_RETURN(0);
+}
+
+static int gemini_helper_threads(dsmContext_t *pContext)
+{
+  int   rc = 0;
+  pthread_t hThread;
+  DBUG_ENTER("gemini_helper_threads");
+  rc = pthread_create (&hThread, 0, gemini_watchdog, (void *)pContext);
+  if (rc)
+  {
+    printf("Can't create gemini watchdog thread");
+    goto done;
+  }
+  if(!gemini_io_threads)
+    goto done;
+
+  rc = pthread_create(&hThread, 0, gemini_rl_writer, (void *)pContext);
+  if(rc)
+  {
+    printf("Can't create gemini recovery log writer thread");
+    goto done;
+  }
+
+  for( int i = gemini_io_threads - 1;i;i--)
+  {
+    rc = pthread_create(&hThread, 0, gemini_apw, (void *)pContext);
+    if(rc)
+    {
+      printf("Can't create gemini page writer thread");
+      goto done;
+    }
+  }
+done:
+
+  DBUG_RETURN(rc);
+}
+
+pthread_handler_decl(gemini_watchdog,arg )
+{
+  int  rc = 0;
+  dsmContext_t *pcontext = (dsmContext_t *)arg;
+  dsmContext_t *pmyContext = NULL;
+
+
+  rc = dsmContextCopy(pcontext,&pmyContext, DSMCONTEXTDB);
+  if( rc != 0 )
+  {
+      printf("dsmContextCopy failed for watchdog %d\n",rc);
+
+      return 0;
+  }
+  rc = dsmUserConnect(pmyContext,NULL,0);
+
+  if( rc != 0 )
+  {
+      printf("dsmUserConnect failed for watchdog %d\n",rc);
+
+      return 0;
+  }
+ 
+  my_thread_init();
+  pthread_detach_this_thread();
+
+  while(rc == 0)
+  {
+    rc = dsmDatabaseProcessEvents(pmyContext);
+    if(!rc)
+      rc = dsmWatchdog(pmyContext);
+    sleep(1);
+  } 
+  rc = dsmUserDisconnect(pmyContext,0); 
+  my_thread_end();
+  return 0;
+}
+
+pthread_handler_decl(gemini_rl_writer,arg )
+{
+  int  rc = 0;
+  dsmContext_t *pcontext = (dsmContext_t *)arg;
+  dsmContext_t *pmyContext = NULL;
+
+
+  rc = dsmContextCopy(pcontext,&pmyContext, DSMCONTEXTDB);
+  if( rc != 0 )
+  {
+      printf("dsmContextCopy failed for recovery log writer %d\n",rc);
+
+      return 0;
+  }
+  rc = dsmUserConnect(pmyContext,NULL,0);
+
+  if( rc != 0 )
+  {
+      printf("dsmUserConnect failed for recovery log writer  %d\n",rc);
+
+      return 0;
+  }
+
+  my_thread_init();
+  pthread_detach_this_thread();
+
+  while(rc == 0)
+  {
+    rc = dsmRLwriter(pmyContext);
+  }
+  rc = dsmUserDisconnect(pmyContext,0);
+  my_thread_end();
+  return 0;
+}
+
+pthread_handler_decl(gemini_apw,arg )
+{
+  int  rc = 0;
+  dsmContext_t *pcontext = (dsmContext_t *)arg;
+  dsmContext_t *pmyContext = NULL;
+
+  my_thread_init();
+  pthread_detach_this_thread();
+
+  rc = dsmContextCopy(pcontext,&pmyContext, DSMCONTEXTDB);
+  if( rc != 0 )
+  {
+      printf("dsmContextCopy failed for gemini page writer %d\n",rc);
+      my_thread_end();
+      return 0;
+  }
+  rc = dsmUserConnect(pmyContext,NULL,0);
+
+  if( rc != 0 )
+  {
+      printf("dsmUserConnect failed for gemini page writer  %d\n",rc);
+      my_thread_end();
+      return 0;
+  }
+
+  while(rc == 0)
+  {
+    rc = dsmAPW(pmyContext);
+  }
+  rc = dsmUserDisconnect(pmyContext,0);
+  my_thread_end();
+  return 0;
+}
+
+int gemini_set_option_long(int optid, long optval)
+{
+  dsmStatus_t rc = 0;
+
+  switch (optid)
+  {
+  case GEM_OPTID_SPIN_RETRIES:
+    /* If we don't have a context yet, skip the set and just save the
+    ** value in gemini_spin_retries for a later gemini_init().  This
+    ** may not ever happen, but we're covered if it does.
+    */
+    if (pfirstContext)
+    {
+      rc = dsmContextSetLong(pfirstContext, DSM_TAGDB_SPIN_AMOUNT,
+                             optval);
+    }
+    if (rc)
+    {
+      printf("SPIN_AMOUNT tag set failed %ld",rc);
+    }
+    else
+    {
+      gemini_spin_retries = optval;
+    }
+    break;
+  }
+
+  return rc;
+}
+
+static int gemini_connect(THD *thd)
+{
+  DBUG_ENTER("gemini_connect");
+
+  dsmStatus_t   rc;
+
+  rc = dsmContextCopy(pfirstContext,(dsmContext_t **)&thd->gemini.context,
+                      DSMCONTEXTDB);
+  if( rc != 0 )
+  {
+      printf("dsmContextCopy failed %ld\n",rc);
+
+      return(rc);
+  }
+  rc = dsmUserConnect((dsmContext_t *)thd->gemini.context,NULL,0);
+
+  if( rc != 0 )
+  {
+      printf("dsmUserConnect failed %ld\n",rc);
+
+      return(rc);
+  }
+  rc = (dsmStatus_t)gemini_tx_begin(thd);
+
+ DBUG_RETURN(rc);
+}
+
+void gemini_disconnect(THD *thd)
+{
+  dsmStatus_t  rc;
+
+  if(thd->gemini.context)
+  {
+    rc = dsmUserDisconnect((dsmContext_t *)thd->gemini.context,0);
+  }
+  return;
+}
+
+bool gemini_end(void)
+{
+  dsmStatus_t rc;
+  THD  *thd;
+
+  DBUG_ENTER("gemini_end");
+  if(pfirstContext)
+  {
+    rc = dsmShutdownSet(pfirstContext, DSM_SHUTDOWN_NORMAL);
+    sleep(2);
+    rc = dsmContextSetLong(pfirstContext,DSM_TAGDB_ACCESS_TYPE,DSM_ACCESS_STARTUP);
+    rc = dsmShutdown(pfirstContext, DSMNICEBIT,DSMNICEBIT);
+  }
+  DBUG_RETURN(0);
+}
+
+bool gemini_flush_logs()
+{
+  DBUG_ENTER("gemini_flush_logs");
+
+  DBUG_RETURN(0);
+}
+
+static int gemini_tx_begin(THD *thd)
+{
+  dsmStatus_t  rc;
+  DBUG_ENTER("gemini_tx_begin");
+  
+  thd->gemini.savepoint = 1;
+
+  rc = dsmTransaction((dsmContext_t *)thd->gemini.context,
+                       &thd->gemini.savepoint,DSMTXN_START,0,NULL);
+  if(!rc)
+    thd->gemini.needSavepoint = 1; 
+
+  thd->gemini.tx_isolation = thd->tx_isolation;
+
+  DBUG_PRINT("trans",("beginning transaction"));
+  DBUG_RETURN(rc);
+}
+
+int gemini_commit(THD *thd)
+{
+  dsmStatus_t  rc;
+  LONG         txNumber = 0;
+
+  DBUG_ENTER("gemini_commit");
+
+  if(!thd->gemini.context)
+    DBUG_RETURN(0);
+ 
+  rc = dsmTransaction((dsmContext_t *)thd->gemini.context,
+			0,DSMTXN_COMMIT,0,NULL);
+  if(!rc)
+    rc = gemini_tx_begin(thd);
+
+  thd->gemini.lock_count = 0;
+  
+  DBUG_PRINT("trans",("ending transaction"));
+  DBUG_RETURN(rc);
+}
+
+int gemini_rollback(THD *thd)
+{
+  dsmStatus_t rc;
+  LONG        txNumber;
+
+  DBUG_ENTER("gemini_rollback");
+  DBUG_PRINT("trans",("aborting transaction"));
+
+  if(!thd->gemini.context)
+    DBUG_RETURN(0);
+
+  thd->gemini.savepoint = 0;
+  rc = dsmTransaction((dsmContext_t *)thd->gemini.context,
+                        &thd->gemini.savepoint,DSMTXN_ABORT,0,NULL);
+  if(!rc)
+    rc = gemini_tx_begin(thd);
+
+  thd->gemini.lock_count = 0;
+
+  DBUG_RETURN(rc);
+}
+
+int gemini_rollback_to_savepoint(THD *thd)
+{
+  dsmStatus_t   rc = 0;
+  DBUG_ENTER("gemini_rollback_to_savepoint");
+  if(thd->gemini.savepoint > 1)
+  {
+    rc = dsmTransaction((dsmContext_t *)thd->gemini.context,
+                        &thd->gemini.savepoint,DSMTXN_UNSAVE,0,NULL);
+  }
+  DBUG_RETURN(rc);
+}
+
+/* gemDataType - translates from mysql data type constant to gemini
+                 key services data type contstant                     */
+int gemDataType ( int mysqlType )
+{
+  switch (mysqlType)
+  {
+    case FIELD_TYPE_LONG:
+    case FIELD_TYPE_TINY:
+    case FIELD_TYPE_SHORT:
+    case FIELD_TYPE_TIMESTAMP:
+    case FIELD_TYPE_LONGLONG:
+    case FIELD_TYPE_INT24:
+    case FIELD_TYPE_DATE:
+    case FIELD_TYPE_TIME:
+    case FIELD_TYPE_DATETIME:
+    case FIELD_TYPE_YEAR:
+    case FIELD_TYPE_NEWDATE:
+    case FIELD_TYPE_ENUM:
+    case FIELD_TYPE_SET:
+      return GEM_INT;
+    case FIELD_TYPE_DECIMAL:
+      return GEM_DECIMAL;
+    case FIELD_TYPE_FLOAT:
+      return GEM_FLOAT;
+    case FIELD_TYPE_DOUBLE:
+      return GEM_DOUBLE;
+    case FIELD_TYPE_TINY_BLOB:
+      return GEM_TINYBLOB;
+    case FIELD_TYPE_MEDIUM_BLOB:
+      return GEM_MEDIUMBLOB;
+    case FIELD_TYPE_LONG_BLOB:
+      return GEM_LONGBLOB;
+    case FIELD_TYPE_BLOB:
+      return GEM_BLOB;
+    case FIELD_TYPE_VAR_STRING:
+    case FIELD_TYPE_STRING:
+      return GEM_CHAR;
+  }
+  return -1;
+}
+
+/*****************************************************************************
+** Gemini tables
+*****************************************************************************/
+
+const char **ha_gemini::bas_ext() const
+{ static const char *ext[]= { ha_gemini_ext, ha_gemini_idx_ext, NullS };
+  return ext;
+}
+
+
+int ha_gemini::open(const char *name, int mode, uint test_if_locked)
+{
+  dsmObject_t  tableId = 0;
+  THD          *thd;
+  char name_buff[FN_REFLEN];
+  char tabname_buff[FN_REFLEN];
+  char dbname_buff[FN_REFLEN];
+  unsigned     i,nameLen;
+  LONG         txNumber;  
+  dsmStatus_t  rc;
+
+  DBUG_ENTER("ha_gemini::open");
+
+  thd = current_thd;
+  thr_lock_init(&alock);
+  thr_lock_data_init(&alock,&lock,(void*)0);
+  ref_length = sizeof(dsmRecid_t);
+
+  if(thd->gemini.context == NULL)
+  {
+    /* Need to get this thread a connection into the database */
+    rc = gemini_connect(thd);
+    if(rc)
+      return rc;
+  }
+  if (!(rec_buff=my_malloc(table->rec_buff_length,
+                           MYF(MY_WME))))
+  {
+    DBUG_RETURN(1);
+  }
+
+  /* separate out the name of the table and the database (a VST must be
+  ** created in the mysql database)
+  */
+  rc = gemini_parse_table_name(name, dbname_buff, tabname_buff);
+  if (rc == 0)
+  {
+    if (strcmp(dbname_buff, "mysql") == 0)
+    {
+      tableId = gemini_is_vst(tabname_buff);
+    }
+  }
+  sprintf(name_buff, "%s.%s", dbname_buff, tabname_buff);
+
+  /* if it's not a VST, get the table number the regular way */
+  if (!tableId)
+  {
+    rc = dsmObjectNameToNum((dsmContext_t *)thd->gemini.context, 
+                            (dsmText_t *)name_buff,
+                            &tableId);
+  }
+  tableNumber = tableId;
+  
+  if(!rc)
+    rc = index_open(name_buff);
+ 
+  fixed_length_row=!(table->db_create_options & HA_OPTION_PACK_RECORD);
+  key_read = 0;
+  using_ignore = 0;
+
+  /* Get the gemini table status -- we want to know if the table
+     crashed while being in the midst of a repair operation */
+  rc = dsmTableStatus((dsmContext_t *)thd->gemini.context,
+                       tableNumber,&tableStatus);
+  if(tableStatus)
+    tableStatus = HA_ERR_CRASHED;
+  DBUG_RETURN (rc);
+}
+
+/* Look up and store the object numbers for the indexes on this table */
+int ha_gemini::index_open(char *tableName)
+{
+  dsmStatus_t  rc = 0;
+  int          nameLen;
+
+  DBUG_ENTER("ha_gemini::index_open");
+  if(table->keys)
+  {
+    THD   *thd = current_thd;
+    dsmObject_t  objectNumber; 
+    if (!(pindexNumbers=(dsmIndex_t *)my_malloc(table->keys*sizeof(dsmIndex_t),
+                           MYF(MY_WME))))
+    {
+      DBUG_RETURN(1);
+    }
+    nameLen = strlen(tableName);
+    tableName[nameLen] = '.';
+    nameLen++;
+    
+    for( uint i = 0; i < table->keys && !rc; i++)
+    {
+      strcpy(&tableName[nameLen],table->key_info[i].name);
+      rc = dsmObjectNameToNum((dsmContext_t *)thd->gemini.context,
+                          (dsmText_t *)tableName,
+                          &objectNumber); 
+      pindexNumbers[i] = objectNumber;
+    }
+  }
+  else
+    pindexNumbers = 0;
+     
+  DBUG_RETURN(rc);
+}
+
+int ha_gemini::close(void)
+{
+  DBUG_ENTER("ha_gemini::close");
+  thr_lock_delete(&alock);
+  my_free(rec_buff,MYF(MY_ALLOW_ZERO_PTR));
+  rec_buff = 0;
+  my_free((char *)pindexNumbers,MYF(MY_ALLOW_ZERO_PTR));
+  pindexNumbers = 0;
+  DBUG_RETURN(0);
+}
+
+
+int ha_gemini::write_row(byte * record)
+{
+  int error = 0;
+  dsmRecord_t dsmRecord;
+  THD         *thd;
+
+  DBUG_ENTER("write_row");
+
+  if(tableStatus)
+    DBUG_RETURN(tableStatus);
+
+  thd = current_thd;
+
+  statistic_increment(ha_write_count,&LOCK_status);
+  if (table->time_stamp)
+    update_timestamp(record+table->time_stamp-1);
+
+  if(thd->gemini.needSavepoint || using_ignore)
+  {
+    thd->gemini.savepoint++;
+    error = dsmTransaction((dsmContext_t *)thd->gemini.context,
+                           &thd->gemini.savepoint,
+                           DSMTXN_SAVE, 0, 0);
+    if (error)
+       DBUG_RETURN(error);
+    thd->gemini.needSavepoint = 0;
+  }
+
+  if (table->next_number_field && record == table->record[0])
+  {
+    if(thd->next_insert_id)
+    {
+      ULONG64  nr;
+       /* A set insert-id statement so set the auto-increment value if this
+          value is higher than it's current value      */
+       error = dsmTableAutoIncrement((dsmContext_t *)thd->gemini.context,
+                                  tableNumber, (ULONG64 *)&nr);
+       if(thd->next_insert_id > nr)
+       {
+         error = dsmTableAutoIncrementSet((dsmContext_t *)thd->gemini.context,tableNumber,
+                                        (ULONG64)thd->next_insert_id);
+       }
+     }
+       
+     update_auto_increment();
+  }
+
+  dsmRecord.table = tableNumber;
+  dsmRecord.maxLength = table->reclength;
+
+  if ((error=pack_row((byte **)&dsmRecord.pbuffer, (int *)&dsmRecord.recLength,
+                      record)))
+    DBUG_RETURN(error);
+
+  error = dsmRecordCreate((dsmContext_t *)thd->gemini.context,
+                          &dsmRecord,0);
+
+  if(!error)
+  {
+    error = handleIndexEntries(record, dsmRecord.recid,KEY_CREATE);
+    if(error == HA_ERR_FOUND_DUPP_KEY && using_ignore)
+    {
+       dsmStatus_t rc;
+       rc = dsmTransaction((dsmContext_t *)thd->gemini.context,
+                    &thd->gemini.savepoint,DSMTXN_UNSAVE,0,NULL);
+       thd->gemini.needSavepoint = 1;
+    }
+  }
+
+  DBUG_RETURN(error);
+}
+
+longlong ha_gemini::get_auto_increment()
+{
+  longlong nr;
+  int      error;
+  THD      *thd=current_thd;
+
+  error = dsmTableAutoIncrement((dsmContext_t *)thd->gemini.context,
+                                  tableNumber, (ULONG64 *)&nr);
+  return nr;
+}
+
+/* Put or delete index entries for a row                          */
+int ha_gemini::handleIndexEntries(const byte * record, dsmRecid_t recid,
+                                  enum_key_string_options option)
+{
+  dsmStatus_t rc = 0;
+
+  DBUG_ENTER("handleIndexEntries");
+
+  for (uint i = 0; i < table->keys && rc == 0; i++)
+  {
+    rc = handleIndexEntry(record, recid,option, i);
+  }
+  DBUG_RETURN(rc);
+}
+
+int ha_gemini::handleIndexEntry(const byte * record, dsmRecid_t recid,
+                                  enum_key_string_options option,uint keynr)
+{
+  dsmStatus_t rc = 0;
+  KEY  *key_info;
+  int   keyStringLen;
+  bool  thereIsAnull;
+  THD   *thd;
+
+  AUTOKEY(theKey,keyBufSize);
+
+  DBUG_ENTER("handleIndexEntry");
+
+  thd = current_thd;
+  key_info=table->key_info+keynr;
+  thereIsAnull = false;
+  rc = createKeyString(record, key_info, theKey.akey.keystr,
+                          sizeof(theKey.apad),&keyStringLen,
+                          (short)pindexNumbers[keynr],
+                           &thereIsAnull);
+  if(!rc)
+  {
+    theKey.akey.index = pindexNumbers[keynr];
+    theKey.akey.keycomps = (COUNT)key_info->key_parts;
+
+    /* We have to subtract three here since cxKeyPrepare
+       expects that the three lead bytes of the header are
+       not counted in this length -- But cxKeyPrepare also
+       expects that these three bytes are present in the keystr */
+    theKey.akey.keyLen = (COUNT)keyStringLen - 3;
+    theKey.akey.unknown_comp = thereIsAnull;
+    theKey.akey.word_index = 0;
+    theKey.akey.descending_key =0;
+    if(option == KEY_CREATE)
+    {
+      rc = dsmKeyCreate((dsmContext_t *)thd->gemini.context, &theKey.akey,
+                        (dsmTable_t)tableNumber, recid, NULL);
+      if(rc == DSM_S_IXDUPKEY)
+      {
+        last_dup_key=keynr;
+        rc = HA_ERR_FOUND_DUPP_KEY;
+      }
+    }
+    else if(option == KEY_DELETE)
+    {
+      rc = dsmKeyDelete((dsmContext_t *)thd->gemini.context, &theKey.akey,
+                        (dsmTable_t)tableNumber, recid, 0, NULL);
+    }
+    else
+    {
+      /* KEY_CHECK  */
+      dsmCursid_t  aCursorId;
+      int          error;
+
+      rc = dsmCursorCreate((dsmContext_t *)thd->gemini.context,
+			   (dsmTable_t)tableNumber,
+			   (dsmIndex_t)pindexNumbers[keynr],
+			   &aCursorId,NULL);
+                           
+      rc = dsmCursorFind((dsmContext_t *)thd->gemini.context,
+			 &aCursorId,&theKey.akey,NULL,DSMDBKEY,
+			 DSMFINDFIRST,DSM_LK_SHARE,0,
+			 &lastRowid,0);
+      error = dsmCursorDelete((dsmContext_t *)thd->gemini.context,
+                          &aCursorId, 0);
+
+    }
+  }
+  DBUG_RETURN(rc);
+}
+
+int ha_gemini::createKeyString(const byte * record, KEY *pkeyinfo,
+                               unsigned char *pkeyBuf, int bufSize,
+                               int  *pkeyStringLen, 
+                               short geminiIndexNumber, 
+                               bool  *thereIsAnull)
+{
+  dsmStatus_t  rc = 0;
+  int          componentLen;
+  int          fieldType;
+  int          isNull;
+
+  KEY_PART_INFO *key_part;
+
+  DBUG_ENTER("createKeyString");
+  
+  rc = gemKeyInit(pkeyBuf,pkeyStringLen, geminiIndexNumber);
+
+  for(uint i = 0; i < pkeyinfo->key_parts && rc == 0; i++)
+  {
+    unsigned char *pos;
+
+    key_part = pkeyinfo->key_part + i;
+    fieldType = gemDataType(key_part->field->type());
+    if(fieldType == GEM_CHAR)
+    {
+      /* Save the current ptr to the field in case we're building a key
+         to remove an old key value when an indexed character column 
+         gets updated.                                                 */
+      char *ptr = key_part->field->ptr;
+      key_part->field->ptr = (char *)record + key_part->offset;
+      key_part->field->sort_string(rec_buff, key_part->length);
+      key_part->field->ptr = ptr;
+      pos = (unsigned char *)rec_buff;
+    }
+    else
+    {
+      pos = (unsigned char *)record + key_part->offset;
+    }
+
+    isNull = record[key_part->null_offset] & key_part->null_bit;
+    if(isNull)
+      *thereIsAnull = true;
+
+    rc = gemFieldToIdxComponent(pos,
+                                (unsigned long) key_part->length,
+                                fieldType,
+                                isNull ,
+                                key_part->field->flags & UNSIGNED_FLAG,
+                                pkeyBuf + *pkeyStringLen,
+                                bufSize,
+                                &componentLen);
+    *pkeyStringLen += componentLen;
+  }  
+  DBUG_RETURN(rc);
+} 
+
+
+int ha_gemini::update_row(const byte * old_record, byte * new_record)
+{
+  int error = 0;
+  dsmRecord_t dsmRecord;
+  unsigned long savepoint;
+  THD         *thd = current_thd;
+  DBUG_ENTER("update_row");
+
+  statistic_increment(ha_update_count,&LOCK_status);
+  if (table->time_stamp)
+    update_timestamp(new_record+table->time_stamp-1);
+
+  if(thd->gemini.needSavepoint || using_ignore)
+  {
+    thd->gemini.savepoint++;
+    error = dsmTransaction((dsmContext_t *)thd->gemini.context,
+                           &thd->gemini.savepoint,
+                           DSMTXN_SAVE, 0, 0);
+    if (error)
+       DBUG_RETURN(error);
+    thd->gemini.needSavepoint = 0;
+  }
+  for (uint keynr=0 ; keynr < table->keys ; keynr++)
+  {
+    if(key_cmp(keynr,old_record, new_record))
+    {
+      error = handleIndexEntry(old_record,lastRowid,KEY_DELETE,keynr);
+      if(error)
+        DBUG_RETURN(error);
+      error = handleIndexEntry(new_record, lastRowid, KEY_CREATE, keynr);
+      if(error)
+      {
+        if (using_ignore && error == HA_ERR_FOUND_DUPP_KEY)
+        {
+           dsmStatus_t rc;
+           rc = dsmTransaction((dsmContext_t *)thd->gemini.context,
+                        &thd->gemini.savepoint,DSMTXN_UNSAVE,0,NULL);
+           thd->gemini.needSavepoint = 1;
+        }
+        DBUG_RETURN(error);
+      }
+    }
+  }
+
+  dsmRecord.table = tableNumber;
+  dsmRecord.recid = lastRowid;
+  dsmRecord.maxLength = table->reclength;
+
+  if ((error=pack_row((byte **)&dsmRecord.pbuffer, (int *)&dsmRecord.recLength,
+                      new_record)))
+  {
+    DBUG_RETURN(error);
+  }
+  error = dsmRecordUpdate((dsmContext_t *)thd->gemini.context,
+                          &dsmRecord, 0, NULL);
+                          
+  DBUG_RETURN(error);
+}
+
+
+int ha_gemini::delete_row(const byte * record)
+{
+  int error = 0;
+  dsmRecord_t  dsmRecord;
+  THD         *thd = current_thd;
+  DBUG_ENTER("delete_row");
+
+  statistic_increment(ha_delete_count,&LOCK_status);
+
+  if(thd->gemini.needSavepoint)
+  {
+    thd->gemini.savepoint++;
+    error = dsmTransaction((dsmContext_t *)thd->gemini.context,
+                           &thd->gemini.savepoint,
+                           DSMTXN_SAVE, 0, 0);
+    if (error)
+       DBUG_RETURN(error);
+    thd->gemini.needSavepoint = 0;
+  }
+
+  dsmRecord.table = tableNumber;
+  dsmRecord.recid = lastRowid;
+
+  error = handleIndexEntries(record, dsmRecord.recid,KEY_DELETE);
+  if(!error)
+  {
+    error = dsmRecordDelete((dsmContext_t *)thd->gemini.context,
+                          &dsmRecord, 0, NULL);
+  }
+
+  DBUG_RETURN(error);
+}  
+
+int ha_gemini::index_init(uint keynr)
+{
+  int error = 0;
+  int keyStringLen;
+  THD *thd;
+  DBUG_ENTER("index_init");
+  thd = current_thd;
+
+  lastRowid = 0;
+  active_index=keynr;
+  error = dsmCursorCreate((dsmContext_t *)thd->gemini.context,
+                           (dsmTable_t)tableNumber, 
+                           (dsmIndex_t)pindexNumbers[keynr],
+                           &cursorId,NULL); 
+  pbracketBase = (dsmKey_t *)my_malloc(sizeof(dsmKey_t) + keyBufSize,
+                           MYF(MY_WME));
+  if(!pbracketBase)
+    DBUG_RETURN(1);
+  pbracketLimit = (dsmKey_t *)my_malloc(sizeof(dsmKey_t) + keyBufSize,MYF(MY_WME));
+  if(!pbracketLimit)
+  {
+    my_free((char *)pbracketLimit,MYF(0));
+    DBUG_RETURN(1);
+  }
+  pbracketBase->index = 0;
+  pbracketLimit->index = (dsmIndex_t)pindexNumbers[keynr];
+  pbracketLimit->keycomps = 1;
+  keyStringLen = 0;
+  error = gemKeyHigh(pbracketLimit->keystr, &keyStringLen,
+                     pbracketLimit->index);
+  
+  /* We have to subtract three here since cxKeyPrepare
+     expects that the three lead bytes of the header are
+     not counted in this length -- But cxKeyPrepare also
+     expects that these three bytes are present in the keystr */
+  pbracketLimit->keyLen = (COUNT)keyStringLen - 3;
+
+  pbracketBase->descending_key = pbracketLimit->descending_key = 0;
+  pbracketBase->ksubstr = pbracketLimit->ksubstr = 0;
+
+  pfoundKey = (dsmKey_t *)my_malloc(sizeof(dsmKey_t) + keyBufSize,MYF(MY_WME));
+  if(!pfoundKey)
+  {
+    my_free((char *)pbracketLimit,MYF(0));
+    my_free((char *)pbracketBase,MYF(0));
+    DBUG_RETURN(1);
+  }
+
+  DBUG_RETURN(error);
+}
+
+int ha_gemini::index_end()
+{
+  int error = 0;
+  THD *thd;
+  DBUG_ENTER("index_end");
+  thd = current_thd;
+  error = dsmCursorDelete((dsmContext_t *)thd->gemini.context,
+                          &cursorId, 0);
+  if(pbracketLimit)
+    my_free((char *)pbracketLimit,MYF(0));
+  if(pbracketBase)
+    my_free((char *)pbracketBase,MYF(0));
+  if(pfoundKey)
+    my_free((char *)pfoundKey,MYF(0));
+
+  pbracketLimit = 0;
+  pbracketBase = 0;
+  pfoundKey = 0;
+  DBUG_RETURN(error);
+}
+
+/* This is only used to read whole keys */
+
+int ha_gemini::index_read_idx(byte * buf, uint keynr, const byte * key,
+                              uint key_len, enum ha_rkey_function find_flag)
+{
+  int error = 0;
+  DBUG_ENTER("index_read_idx");
+  statistic_increment(ha_read_key_count,&LOCK_status);
+
+  error = index_init(keynr);
+  if (!error)
+    error = index_read(buf,key,key_len,find_flag);
+
+  if(error == HA_ERR_END_OF_FILE)
+    error = HA_ERR_KEY_NOT_FOUND;
+
+  table->status = error ? STATUS_NOT_FOUND : 0;
+  DBUG_RETURN(error);
+}
+
+int ha_gemini::pack_key( uint keynr, dsmKey_t *pkey, 
+                           const byte *key_ptr, uint key_length)
+{
+  KEY *key_info=table->key_info+keynr;
+  KEY_PART_INFO *key_part=key_info->key_part;
+  KEY_PART_INFO *end=key_part+key_info->key_parts;
+  int rc;
+  int componentLen;
+  DBUG_ENTER("pack_key");
+
+  rc = gemKeyInit(pkey->keystr,&componentLen,
+             (short)pindexNumbers[active_index]);
+  pkey->keyLen = componentLen;
+
+  for (; key_part != end && (int) key_length > 0 && !rc; key_part++)
+  {
+    uint offset=0;
+    unsigned char *pos;
+
+    int fieldType; 
+    if (key_part->null_bit)
+    {
+      offset=1;
+      if (*key_ptr != 0)         // Store 0 if NULL
+      {
+        key_length-= key_part->store_length;
+        key_ptr+=   key_part->store_length;
+        rc = gemFieldToIdxComponent(
+                 (unsigned char *)key_ptr + offset,
+                                 (unsigned long) key_part->length,
+                                  0,
+                                  1 ,  /* Tells it to build a null component */
+                                  key_part->field->flags & UNSIGNED_FLAG,
+                                  pkey->keystr + pkey->keyLen,
+                                  keyBufSize,
+                                  &componentLen);
+        pkey->keyLen += componentLen;        
+        continue;
+      }
+    }
+    fieldType = gemDataType(key_part->field->type());
+    if(fieldType == GEM_CHAR)
+    {
+      key_part->field->store(key_ptr + offset, key_part->length); 
+      key_part->field->sort_string(rec_buff, key_part->length);
+      pos = (unsigned char *)rec_buff;
+    }
+    else
+    {
+      pos = (unsigned char *)key_ptr + offset;
+    }
+
+    rc = gemFieldToIdxComponent(
+                                 pos,
+                                 (unsigned long) key_part->length,
+                                  fieldType,
+                                  0 ,
+                                  key_part->field->flags & UNSIGNED_FLAG,
+                                  pkey->keystr + pkey->keyLen,
+                                  keyBufSize,
+                                  &componentLen);
+
+    key_ptr+=key_part->store_length;
+    key_length-=key_part->store_length;
+    pkey->keyLen += componentLen;
+  }
+  DBUG_RETURN(rc);
+}
+
+void ha_gemini::unpack_key(char *record, dsmKey_t *key, uint index)
+{
+  KEY *key_info=table->key_info+index;
+  KEY_PART_INFO *key_part= key_info->key_part,
+                *end=key_part+key_info->key_parts;
+  int fieldIsNull, fieldType;
+  int rc = 0;
+
+  char unsigned *pos= &key->keystr[7];
+
+  for ( ; key_part != end; key_part++)
+  {
+    fieldType = gemDataType(key_part->field->type());
+    if(fieldType == GEM_CHAR)
+    {
+      /* Can't get data from character indexes since the sort weights
+         are in the index and not the characters.   */
+      key_read = 0;
+    }
+    rc = gemIdxComponentToField(pos, fieldType,
+                                (unsigned char *)record + key_part->field->offset(),
+                                key_part->field->field_length,
+                                key_part->field->decimals(),
+                                &fieldIsNull);
+    if(fieldIsNull)
+    {
+      record[key_part->null_offset] |= key_part->null_bit;
+    }
+    else if (key_part->null_bit)
+    {
+      record[key_part->null_offset]&= ~key_part->null_bit;
+    }
+    while(*pos++);       /* Advance to next field in key by finding */
+                         /*   a null byte                             */
+  }
+}
+
+int ha_gemini::index_read(byte * buf, const byte * key,
+			    uint key_len, enum ha_rkey_function find_flag)
+{
+  int error = 0;
+  THD   *thd;
+  int componentLen;
+ 
+  DBUG_ENTER("index_read");
+  statistic_increment(ha_read_key_count,&LOCK_status);
+
+
+  pbracketBase->index = (short)pindexNumbers[active_index];
+  pbracketBase->keycomps =  1;
+  
+
+  /* Its a greater than operation so create a base bracket
+       from the input key data.                              */
+  error = pack_key(active_index, pbracketBase, key, key_len);
+  if(error)
+    goto errorReturn;
+
+  if(find_flag == HA_READ_AFTER_KEY)
+  {
+    /* A greater than operation      */
+    error = gemKeyAddLow(pbracketBase->keystr + pbracketBase->keyLen,
+                         &componentLen);
+    pbracketBase->keyLen += componentLen;
+  }
+  if(find_flag == HA_READ_KEY_EXACT)
+  {
+    /* Need to set up a high bracket for an equality operator 
+       Which is a copy of the base bracket plus a hi lim term */
+    bmove(pbracketLimit,pbracketBase,(size_t)pbracketBase->keyLen + sizeof(dsmKey_t));
+    error = gemKeyAddHigh(pbracketLimit->keystr + pbracketLimit->keyLen,
+                          &componentLen);
+    if(error)
+      goto errorReturn;
+    pbracketLimit->keyLen += componentLen;
+  }
+  else
+  {
+    /* Always add a high range -- except for HA_READ_KEY_EXACT this
+       is all we need for the upper index bracket    */
+    error = gemKeyHigh(pbracketLimit->keystr, &componentLen,
+		       pbracketLimit->index);
+  
+    pbracketLimit->keyLen = componentLen;
+  }
+  /* We have to subtract three here since cxKeyPrepare
+     expects that the three lead bytes of the header are
+     not counted in this length -- But cxKeyPrepare also
+     expects that these three bytes are present in the keystr */
+  pbracketBase->keyLen -= 3;
+  pbracketLimit->keyLen -= 3;
+
+  thd = current_thd;
+
+  error = findRow(thd, DSMFINDFIRST, buf);
+
+errorReturn:
+  if (error == DSM_S_ENDLOOP)
+    error = HA_ERR_KEY_NOT_FOUND;
+ 
+  table->status = error ? STATUS_NOT_FOUND : 0;
+  DBUG_RETURN(error);
+}
+
+
+int ha_gemini::index_next(byte * buf)
+{
+  THD  *thd;
+  int error = 1;
+  int keyStringLen=0;
+  dsmMask_t   findMode;
+  DBUG_ENTER("index_next");
+
+  if(tableStatus)
+    DBUG_RETURN(tableStatus);
+
+  thd = current_thd;
+  
+  if(pbracketBase->index == 0)
+  {
+    error = gemKeyLow(pbracketBase->keystr, &keyStringLen, 
+                      pbracketLimit->index);
+    
+    pbracketBase->keyLen = (COUNT)keyStringLen - 3;
+    pbracketBase->index = pbracketLimit->index;
+    pbracketBase->keycomps =  1;
+    findMode = DSMFINDFIRST;
+  }    
+  else
+    findMode = DSMFINDNEXT;
+
+  error = findRow(thd,findMode,buf);
+
+  if (error == DSM_S_ENDLOOP)
+    error = HA_ERR_END_OF_FILE;
+
+  table->status = error ? STATUS_NOT_FOUND : 0;
+  DBUG_RETURN(error);
+}
+
+int ha_gemini::index_next_same(byte * buf, const byte *key, uint keylen)
+{
+  int error = 0;
+  DBUG_ENTER("index_next_same");
+  statistic_increment(ha_read_next_count,&LOCK_status);
+  DBUG_RETURN(index_next(buf));
+}
+
+
+int ha_gemini::index_prev(byte * buf)
+{
+  int error = 0;
+  THD *thd = current_thd;
+
+  DBUG_ENTER("index_prev");
+  statistic_increment(ha_read_prev_count,&LOCK_status);
+
+  error = findRow(thd, DSMFINDPREV, buf);
+
+  if (error == DSM_S_ENDLOOP)
+    error = HA_ERR_END_OF_FILE;
+
+
+  table->status = error ? STATUS_NOT_FOUND : 0;
+  DBUG_RETURN(error);
+}
+  
+
+int ha_gemini::index_first(byte * buf)
+{
+  DBUG_ENTER("index_first");
+  statistic_increment(ha_read_first_count,&LOCK_status);
+  DBUG_RETURN(index_next(buf));
+}
+
+int ha_gemini::index_last(byte * buf)
+{
+  int error = 0;
+  THD  *thd;
+  int keyStringLen;
+  dsmMask_t   findMode;
+  thd = current_thd;
+ 
+  DBUG_ENTER("index_last");
+  statistic_increment(ha_read_last_count,&LOCK_status);
+
+  error = gemKeyLow(pbracketBase->keystr, &keyStringLen,
+                      pbracketLimit->index);
+  if(error)
+    goto errorReturn;
+
+  pbracketBase->keyLen = (COUNT)keyStringLen - 3;
+  pbracketBase->index = pbracketLimit->index;
+  pbracketBase->keycomps =  1;
+
+  error = findRow(thd,DSMFINDLAST,buf);
+
+errorReturn:
+  if (error == DSM_S_ENDLOOP)
+    error = HA_ERR_END_OF_FILE;
+
+  table->status = error ? STATUS_NOT_FOUND : 0;
+  DBUG_RETURN(error);
+  
+  table->status = error ? STATUS_NOT_FOUND : 0;
+  DBUG_RETURN(error);
+}
+
+int ha_gemini::rnd_init(bool scan)
+{
+  THD  *thd = current_thd;
+
+  lastRowid = 0;
+
+  return 0;
+}
+
+int ha_gemini::rnd_end()
+{
+/*
+  return gem_scan_end();
+*/
+  return 0;
+}
+
+int ha_gemini::rnd_next(byte *buf)
+{
+  int error = 0;
+  dsmRecord_t  dsmRecord;
+  THD  *thd;
+
+  DBUG_ENTER("rnd_next");
+
+  if(tableStatus)
+    DBUG_RETURN(tableStatus);
+
+  thd = current_thd;
+  if(thd->gemini.tx_isolation == ISO_READ_COMMITTED && !(lockMode & DSM_LK_EXCL)
+     && lastRowid)
+    error = dsmObjectUnlock((dsmContext_t *)thd->gemini.context,
+			    tableNumber, DSMOBJECT_RECORD, lastRowid, 
+			    lockMode | DSM_UNLK_FREE, 0);
+
+  statistic_increment(ha_read_rnd_next_count,&LOCK_status);
+  dsmRecord.table = tableNumber;
+  dsmRecord.recid = lastRowid;
+  dsmRecord.pbuffer = (dsmBuffer_t *)rec_buff;
+  dsmRecord.recLength = table->reclength;
+  dsmRecord.maxLength = table->reclength;
+
+  error = dsmTableScan((dsmContext_t *)thd->gemini.context,
+                       &dsmRecord, DSMFINDNEXT, lockMode, 0);
+
+  if(!error)
+  {
+     lastRowid = dsmRecord.recid;
+     unpack_row((char *)buf,(char *)dsmRecord.pbuffer);
+  }     
+  if(!error)
+    ;
+  else if (error == DSM_S_ENDLOOP)
+    error = HA_ERR_END_OF_FILE;
+  else if (error ==  DSM_S_RQSTREJ)
+    error = HA_ERR_LOCK_WAIT_TIMEOUT;
+  else if (error == DSM_S_LKTBFULL)
+    error = HA_ERR_LOCK_TABLE_FULL;
+
+  table->status = error ? STATUS_NOT_FOUND : 0;
+  DBUG_RETURN(error);
+}
+
+
+int ha_gemini::rnd_pos(byte * buf, byte *pos)
+{
+  int error;
+  int rc;
+
+  THD *thd;
+
+  statistic_increment(ha_read_rnd_count,&LOCK_status);
+  thd = current_thd;
+  memcpy((void *)&lastRowid,pos,ref_length);
+  if(thd->gemini.tx_isolation == ISO_READ_COMMITTED && !(lockMode & DSM_LK_EXCL))
+  {
+    /* Lock the row      */
+
+    error = dsmObjectLock((dsmContext_t *)thd->gemini.context,
+                       (dsmObject_t)tableNumber,DSMOBJECT_RECORD,lastRowid,
+                       lockMode, 1, 0);
+    if ( error ) 
+      goto errorReturn;
+  }
+  error = fetch_row(thd->gemini.context, buf);  
+  if(thd->gemini.tx_isolation == ISO_READ_COMMITTED && !(lockMode & DSM_LK_EXCL))
+  {
+    /* Unlock the row      */
+
+    rc = dsmObjectUnlock((dsmContext_t *)thd->gemini.context,
+                       (dsmObject_t)tableNumber,DSMOBJECT_RECORD,lastRowid,
+                       lockMode | DSM_UNLK_FREE , 0);
+  }
+  if(error == DSM_S_RMNOTFND)
+    error = HA_ERR_RECORD_DELETED;
+
+ errorReturn:
+  table->status = error ? STATUS_NOT_FOUND : 0;
+  return error;
+}
+
+int ha_gemini::fetch_row(void *gemini_context,const byte *buf)
+{
+  dsmStatus_t  rc = 0;
+  dsmRecord_t  dsmRecord;
+
+  DBUG_ENTER("fetch_row");
+  dsmRecord.table = tableNumber;
+  dsmRecord.recid = lastRowid;
+  dsmRecord.pbuffer = (dsmBuffer_t *)rec_buff;
+  dsmRecord.recLength = table->reclength;
+  dsmRecord.maxLength = table->reclength;
+
+  rc = dsmRecordGet((dsmContext_t *)gemini_context,
+                       &dsmRecord, 0);
+
+  if(!rc)
+  {
+     unpack_row((char *)buf,(char *)dsmRecord.pbuffer);
+  }
+
+   DBUG_RETURN(rc);
+}
+int ha_gemini::findRow(THD *thd, dsmMask_t findMode, byte *buf)
+{
+  dsmStatus_t rc;
+  dsmKey_t    *pkey;
+
+  DBUG_ENTER("findRow"); 
+
+  if(thd->gemini.tx_isolation == ISO_READ_COMMITTED && !(lockMode & DSM_LK_EXCL)
+     && lastRowid)
+    rc = dsmObjectUnlock((dsmContext_t *)thd->gemini.context,
+			    tableNumber, DSMOBJECT_RECORD, lastRowid, 
+			    lockMode | DSM_UNLK_FREE, 0);
+  if( key_read )
+    pkey = pfoundKey;
+  else
+    pkey = 0;
+ 
+  rc = dsmCursorFind((dsmContext_t *)thd->gemini.context,
+                        &cursorId,
+                        pbracketBase,
+                        pbracketLimit,
+                        DSMPARTIAL,
+                        findMode,
+                        lockMode,
+                        NULL,
+                        &lastRowid,
+                        pkey);
+  if( rc )
+    goto errorReturn;
+
+  if(key_read)
+  {
+    unpack_key(buf, pkey, active_index);
+  }
+  if(!key_read)  /* unpack_key may have turned off key_read  */
+  {
+    rc = fetch_row((dsmContext_t *)thd->gemini.context,buf);
+  }
+
+errorReturn:
+  if(!rc)
+    ;
+  else if(rc ==  DSM_S_RQSTREJ)
+    rc = HA_ERR_LOCK_WAIT_TIMEOUT;
+  else if (rc == DSM_S_LKTBFULL)
+    rc = HA_ERR_LOCK_TABLE_FULL;
+
+ DBUG_RETURN(rc);
+}
+
+void ha_gemini::position(const byte *record)
+{
+  memcpy(ref,&lastRowid,ref_length);
+}
+
+
+void ha_gemini::info(uint flag)
+{
+  DBUG_ENTER("info");
+
+  if ((flag & HA_STATUS_VARIABLE))
+  {
+    THD  *thd = current_thd;
+    dsmStatus_t error;
+    ULONG64     rows;
+
+    error = dsmRowCount((dsmContext_t *)thd->gemini.context,tableNumber,&rows);
+    records = (ha_rows)rows;
+    deleted = 0;
+  }
+  else if ((flag & HA_STATUS_CONST))
+  {
+    ;
+  }
+  else if ((flag & HA_STATUS_ERRKEY))
+  {
+    errkey=last_dup_key;
+  }
+  else if ((flag & HA_STATUS_TIME))
+  {
+    ;
+  }
+  else if ((flag & HA_STATUS_AUTO))
+  {
+    ;
+  }
+
+  DBUG_VOID_RETURN;
+}
+
+
+int ha_gemini::extra(enum ha_extra_function operation)
+{
+  switch (operation)
+  {
+  case HA_EXTRA_RESET:
+  case HA_EXTRA_RESET_STATE:
+    key_read=0;
+    using_ignore=0;
+    break;
+  case HA_EXTRA_KEYREAD:
+    key_read=1;					// Query satisfied with key
+    break;
+  case HA_EXTRA_NO_KEYREAD:
+    key_read=0;
+    break;
+  case HA_EXTRA_IGNORE_DUP_KEY:
+    using_ignore=1;
+    break;
+  case HA_EXTRA_NO_IGNORE_DUP_KEY:
+    using_ignore=0;
+    break;
+
+  default:
+    break;
+  }
+  return 0;
+}
+
+
+int ha_gemini::reset(void)
+{
+  key_read=0;					// Reset to state after open
+  return 0;
+}
+
+
+/*
+  As MySQL will execute an external lock for every new table it uses
+  we can use this to start the transactions.
+*/
+
+int ha_gemini::external_lock(THD *thd, int lock_type)
+{
+  dsmStatus_t rc = 0;
+  LONG        txNumber;
+
+  DBUG_ENTER("ha_gemini::external_lock");
+
+  if (lock_type != F_UNLCK)
+  {
+    if (!thd->gemini.lock_count)
+    {
+      thd->gemini.lock_count = 1;
+      thd->gemini.tx_isolation = thd->tx_isolation;
+    }
+    
+    if(thd->gemini.context == NULL)
+    {
+      /* Need to get this thread a connection into the database */
+      rc = gemini_connect(thd);
+      if(rc)
+        return rc;
+    }
+    /* Set need savepoint flag */
+    thd->gemini.needSavepoint = 1;
+
+    if(rc)
+      DBUG_RETURN(rc);
+
+
+    if( thd->in_lock_tables || thd->gemini.tx_isolation == ISO_SERIALIZABLE )
+    {
+      rc = dsmObjectLock((dsmContext_t *)thd->gemini.context,
+			 (dsmObject_t)tableNumber,DSMOBJECT_TABLE,0,
+			 lockMode, 1, 0);
+    } 
+  }
+  else /* lock_type == F_UNLK  */
+  {
+    /* Commit the tx if we're in auto-commit mode */
+    if (!(thd->options & OPTION_NOT_AUTO_COMMIT)&&
+	!(thd->options & OPTION_BEGIN))
+      gemini_commit(thd);
+  }
+
+  DBUG_RETURN(rc);
+}  
+
+
+THR_LOCK_DATA **ha_gemini::store_lock(THD *thd, THR_LOCK_DATA **to,
+					enum thr_lock_type lock_type)
+{
+  if (lock_type != TL_IGNORE && lock.type == TL_UNLOCK)
+  {
+    /* If we are not doing a LOCK TABLE, then allow multiple writers */
+    if ((lock_type >= TL_WRITE_CONCURRENT_INSERT &&
+	 lock_type <= TL_WRITE) &&
+	!thd->in_lock_tables)
+      lock_type = TL_WRITE_ALLOW_WRITE;
+    lock.type=lock_type;
+
+    if(thd->gemini.tx_isolation == ISO_READ_UNCOMMITTED)
+      lockMode = DSM_LK_NOLOCK;
+    else if(table->reginfo.lock_type > TL_WRITE_ALLOW_READ)
+       lockMode = DSM_LK_EXCL;
+    else
+       lockMode = DSM_LK_SHARE;
+  }
+  *to++= &lock;
+  return to;
+}
+
+
+int ha_gemini::create(const char *name, register TABLE *form,
+                      HA_CREATE_INFO *create_info)
+{
+  THD *thd;
+  char name_buff[FN_REFLEN];
+  char dbname_buff[FN_REFLEN];
+  DBUG_ENTER("ha_gemini::create");
+  dsmContext_t  *pcontext;
+  dsmStatus_t   rc;
+  dsmArea_t     areaNumber;
+  dsmObject_t   tableNumber = 0;
+  dsmDbkey_t    dummy = 0;
+  unsigned      i;
+  int           baseNameLen;
+  dsmObject_t   indexNumber;
+ 
+  /* separate out the name of the table and the database (a VST must be
+  ** created in the mysql database)
+  */
+  rc = gemini_parse_table_name(name, dbname_buff, name_buff);
+  if (rc == 0)
+  {
+    /* If the table is a VST, don't create areas or extents */
+    if (strcmp(dbname_buff, "mysql") == 0)
+    {
+      tableNumber = gemini_is_vst(name_buff);
+      if (tableNumber)
+      {
+        return 0;
+      }
+    }
+  }
+
+  thd = current_thd;
+  if(thd->gemini.context == NULL)
+  {
+    /* Need to get this thread a connection into the database */
+    rc = gemini_connect(thd);
+    if(rc)
+      return rc;
+  }
+  pcontext = (dsmContext_t *)thd->gemini.context;
+
+  if(thd->gemini.needSavepoint || using_ignore)
+  {
+    thd->gemini.savepoint++;
+    rc = dsmTransaction((dsmContext_t *)thd->gemini.context,
+                           &thd->gemini.savepoint,
+                           DSMTXN_SAVE, 0, 0);
+    if (rc)
+       DBUG_RETURN(rc);
+    thd->gemini.needSavepoint = 0;
+  }
+
+  fn_format(name_buff, name, "", ha_gemini_ext, 2 | 4);
+  /* Create a storage area                               */
+  rc = dsmAreaNew(pcontext,gemini_blocksize,DSMAREA_TYPE_DATA,
+                  &areaNumber, gemini_recbits,
+		(dsmText_t *)"gemini_data_area");
+  if( rc != 0 )
+  {
+      printf("dsmAreaNew failed %ld\n",rc);
+      return(rc);
+  }
+  
+  /* Create an extent                                    */
+  /* Don't pass in leading ./ in name_buff                  */
+  rc = dsmExtentCreate(pcontext,areaNumber,1,15,5,
+                       (dsmText_t *)&name_buff[start_of_name]);
+  if( rc != 0 )
+  {
+      printf("dsmExtentCreate failed %ld\n",rc);
+      return(rc);
+  }
+
+  /* Create the table storage object                     */
+  /* Change slashes in the name to periods               */
+  for(  i = 0; i < strlen(name_buff); i++)
+    if(name_buff[i] == '/' || name_buff[i] == '\\')
+      name_buff[i] = '.';
+ 
+  /* Get rid of .gmd suffix      */
+  name_buff[strlen(name_buff) - 4] = '\0';
+ 
+  rc = dsmObjectCreate(pcontext, areaNumber, &tableNumber, 
+                       DSMOBJECT_MIXTABLE,0,0,0,
+		       (dsmText_t *)&name_buff[start_of_name],
+                       &dummy,&dummy);
+
+  if(rc == 0 && form->keys)
+  {
+    fn_format(name_buff, name, "", ha_gemini_idx_ext, 2 | 4); 
+       /* Create a storage area                               */
+    rc = dsmAreaNew(pcontext,gemini_blocksize,DSMAREA_TYPE_DATA,
+                  &areaNumber, gemini_recbits,
+                (dsmText_t *)"gemini_index_area");
+    if( rc != 0 )
+    {
+        printf("dsmAreaNew failed %ld\n",rc);
+        return(rc);
+    } 
+    /* Create an extent                                    */
+    /* Don't pass in leading ./ in name_buff                  */
+    rc = dsmExtentCreate(pcontext,areaNumber,1,15,5,
+                       (dsmText_t *)&name_buff[start_of_name]);
+    if( rc != 0 )
+    {
+        printf("dsmExtentCreate failed %ld\n",rc);
+        return(rc);
+    }
+    
+    /* Change slashes in the name to periods               */
+    for(  i = 0; i < strlen(name_buff); i++)
+        if(name_buff[i] == '/' || name_buff[i] == '\\')
+          name_buff[i] = '.';
+
+    /* Get rid of .gmi suffix      */
+    name_buff[strlen(name_buff) - 4] = '\0';
+
+    baseNameLen = strlen(name_buff);
+    name_buff[baseNameLen] = '.';
+    baseNameLen++;
+    for( i = 0; i < form->keys; i++)
+    {
+       dsmObjectAttr_t  indexUnique;
+
+       indexNumber = DSMINDEX_INVALID;    
+       /* Create a storage object record for each index */
+       /* Add the index name so the object name is in the form
+          <db>.<table>.<index_name>                     */
+       strcpy(&name_buff[baseNameLen],table->key_info[i].name);
+       if(table->key_info[i].flags & HA_NOSAME)
+         indexUnique = 1;
+       else
+         indexUnique = 0;
+       rc = dsmObjectCreate(pcontext, areaNumber, &indexNumber,
+                       DSMOBJECT_MIXINDEX,indexUnique,tableNumber,
+                       DSMOBJECT_MIXTABLE,
+                       (dsmText_t *)&name_buff[start_of_name],
+                       &dummy,&dummy);
+           
+    }
+  }
+  rc = dsmTableAutoIncrementSet(pcontext,tableNumber,
+                                create_info->auto_increment_value);
+
+
+  
+  /* Get a table lock on this table in case this table is being
+     created as part of an alter table statement.  We don't want
+     the alter table statement to abort because of a lock table overflow
+  */
+  if (thd->lex.sql_command == SQLCOM_CREATE_INDEX ||
+      thd->lex.sql_command == SQLCOM_ALTER_TABLE ||
+      thd->lex.sql_command == SQLCOM_DROP_INDEX)
+  {
+    rc = dsmObjectLock(pcontext,
+                       (dsmObject_t)tableNumber,DSMOBJECT_TABLE,0,
+                       DSM_LK_EXCL, 1, 0);
+    /* and don't commit so we won't release the table on the table number
+       of the table being altered */
+  }
+  else
+  {
+    if(!rc)
+      rc = gemini_commit(thd);
+  }
+
+  DBUG_RETURN(rc);
+}
+
+int ha_gemini::delete_table(const char *pname)
+{
+    THD *thd;
+    dsmStatus_t     rc;
+    dsmContext_t   *pcontext;
+    unsigned        i,nameLen;
+    dsmArea_t       indexArea = 0;
+    dsmArea_t       tableArea = 0;
+    dsmObjectAttr_t objectAttr;
+    dsmObject_t     associate;
+    dsmObjectType_t associateType;
+    dsmDbkey_t      block, root;
+    int             need_txn = 0;
+    dsmObject_t     tableNum = 0;
+    char            name_buff[FN_REFLEN];
+    char            dbname_buff[FN_REFLEN];
+    DBUG_ENTER("ha_gemini::delete_table");
+ 
+    /* separate out the name of the table and the database (a VST must be
+    ** located in the mysql database)
+    */
+    rc = gemini_parse_table_name(pname, dbname_buff, name_buff);
+    if (rc == 0)
+    {
+      /* If the table is a VST, there are no areas or extents to delete */
+      if (strcmp(dbname_buff, "mysql") == 0)
+      { 
+        tableNum = gemini_is_vst(name_buff);
+        if (tableNum)
+        {
+          return 0;
+        }
+      }
+    }
+
+    thd = current_thd;
+    if(thd->gemini.context == NULL)
+    {
+        /* Need to get this thread a connection into the database */
+        rc = gemini_connect(thd);
+        if(rc)
+        {
+            DBUG_RETURN(rc);
+        }
+    }
+    pcontext = (dsmContext_t *)thd->gemini.context;
+
+
+    bzero(name_buff, FN_REFLEN);
+
+    nameLen = strlen(pname);
+    for( i = start_of_name; i < nameLen; i++)
+    {
+        if(pname[i] == '/' || pname[i] == '\\')
+            name_buff[i-start_of_name] = '.';
+        else
+            name_buff[i-start_of_name] = pname[i];
+    }
+
+    rc = dsmObjectNameToNum(pcontext, (dsmText_t *)name_buff, 
+               (dsmObject_t *)&tableNum);
+    if (rc)
+    {
+        printf("Cound not find table number for %s with string %s, %ld\n",
+                    pname,name_buff,rc);
+        rc = gemini_rollback(thd);
+        if (rc)
+        {
+            printf("Error in rollback %ld\n",rc);
+        }
+        DBUG_RETURN(rc);
+    }
+
+    rc = dsmObjectInfo(pcontext, tableNum, DSMOBJECT_MIXTABLE, &tableArea,
+                &objectAttr, &associate, &associateType, &block, &root);
+    if (rc)
+    {
+        printf("Failed to get area number for table %d, %s, return %ld\n",
+                 tableNum, pname, rc);
+        rc = gemini_rollback(thd);
+        if (rc)
+        {
+            printf("Error in rollback %ld\n",rc);
+        }
+    }
+                        
+    indexArea = DSMAREA_INVALID;
+
+    /* Delete the indexes and tables storage objects for with the table */
+    rc = dsmObjectDeleteAssociate(pcontext, tableNum, &indexArea);
+    if (rc)
+    {
+        printf("Error deleting storage objects for table number %d, return %ld\n", 
+                       (int)tableNum, rc);
+
+        /* roll back txn and return */
+        rc = gemini_rollback(thd);
+        if (rc)
+        {
+            printf("Error in rollback %ld\n",rc);
+        }
+        DBUG_RETURN(rc);
+    }
+
+    if (indexArea != DSMAREA_INVALID)
+    {
+        /* Delete the extents for both Index and Table */
+        rc = dsmExtentDelete(pcontext, indexArea, 0);
+        rc = dsmAreaDelete(pcontext, indexArea);
+        if (rc)
+        {
+            printf("Error deleting Index Area %ld, return %ld\n", indexArea, rc);
+
+            /* roll back txn and return */
+            rc = gemini_rollback(thd);
+            if (rc)
+            {
+                printf("Error in rollback %ld\n",rc);
+            }
+            DBUG_RETURN(rc);
+        }
+    }
+
+    rc = dsmExtentDelete(pcontext, tableArea, 0);
+    rc = dsmAreaDelete(pcontext, tableArea);
+    if (rc)
+    {
+        printf("Error deleting table Area %ld, name %s, return %ld\n",
+                    tableArea, pname, rc);
+        /* roll back txn and return */
+        rc = gemini_rollback(thd);
+        if (rc)
+        {
+            printf("Error in rollback %ld\n",rc);
+        }
+        DBUG_RETURN(rc);
+    }
+
+
+    /* Commit the transaction */
+    rc = gemini_commit(thd);
+    if (rc)
+    {
+        printf("Failed to commit transaction %ld\n",rc);
+    }
+
+  
+    /* now remove all the files that need to be removed and
+       cause a checkpoint so recovery will work */
+    rc = dsmExtentUnlink(pcontext);
+
+    DBUG_RETURN(0);
+}
+
+
+int ha_gemini::rename_table(const char *pfrom, const char *pto)
+{
+    THD            *thd;
+    dsmContext_t   *pcontext;
+    dsmStatus_t     rc;
+    char            tabname_buff[FN_REFLEN];
+    char            dbname_buff[FN_REFLEN];
+    char            name_buff[FN_REFLEN];
+    char            newname_buff[FN_REFLEN];
+    char            newextname_buff[FN_REFLEN];
+    char            newidxextname_buff[FN_REFLEN];
+    unsigned        i, nameLen;
+    dsmObject_t     tableNum;
+    dsmArea_t       indexArea = 0;
+
+    DBUG_ENTER("ha_gemini::rename_table");
+
+    /* don't allow rename of VSTs */
+    rc = gemini_parse_table_name(pfrom, dbname_buff, name_buff);
+    if (rc == 0)
+    {
+      /* If the table is a VST, don't create areas or extents */
+      if (strcmp(dbname_buff, "mysql") == 0)
+      {
+        if (gemini_is_vst(name_buff))
+        {
+          return 0;
+        }
+      }
+    }
+
+    thd = current_thd;
+    if (thd->gemini.context == NULL)
+    {
+      /* Need to get this thread a connection into the database */
+      rc = gemini_connect(thd);
+      if (rc)
+      {
+        DBUG_RETURN(rc);
+      }
+    }
+
+    pcontext = (dsmContext_t *)thd->gemini.context;
+
+    /* change the slashes to dots in the old and new names */
+    nameLen = strlen(pfrom);
+    for( i = start_of_name; i < nameLen; i++)
+    {
+        if(pfrom[i] == '/' || pfrom[i] == '\\')
+            name_buff[i-start_of_name] = '.';
+        else
+            name_buff[i-start_of_name] = pfrom[i];
+    }
+    name_buff[i-start_of_name] = '\0';
+
+    nameLen = strlen(pto);
+    for( i = start_of_name; i < nameLen; i++)
+    {
+        if(pto[i] == '/' || pto[i] == '\\')
+            newname_buff[i-start_of_name] = '.';
+        else
+            newname_buff[i-start_of_name] = pto[i];
+    }
+    newname_buff[i-start_of_name] = '\0';
+
+    /* generate new extent names (for table and index extents) */
+    fn_format(newextname_buff, pto, "", ha_gemini_ext, 2 | 4);
+    fn_format(newidxextname_buff, pto, "", ha_gemini_idx_ext, 2 | 4);
+
+    rc = dsmObjectNameToNum(pcontext, (dsmText_t *)name_buff, &tableNum);
+    if (rc)
+        goto errorReturn;
+
+    rc = dsmObjectRename(pcontext, tableNum,
+                         (dsmText_t *)newname_buff,
+                         (dsmText_t *)&newidxextname_buff[start_of_name],
+                         (dsmText_t *)&newextname_buff[start_of_name],
+                         &indexArea);
+    if (rc)
+        goto errorReturn;
+
+    /* rename the physical table and index files (if necessary) */
+    rc = rename_file_ext(pfrom, pto, ha_gemini_ext);
+    if (!rc && indexArea)
+    {
+        rc = rename_file_ext(pfrom, pto, ha_gemini_idx_ext);
+    }
+
+errorReturn:
+    DBUG_RETURN(rc);
+}
+
+
+/*
+  How many seeks it will take to read through the table
+  This is to be comparable to the number returned by records_in_range so
+  that we can decide if we should scan the table or use keys.
+*/
+
+double ha_gemini::scan_time()
+{
+  return records / (gemini_blocksize / table->reclength);
+}
+
+int ha_gemini::check(THD* thd, HA_CHECK_OPT* check_opt)
+{
+  int error;
+  int checkStatus = HA_ADMIN_OK;
+  ha_rows indexCount;
+  byte         *buf = 0, *indexBuf = 0;
+  int		errorCount = 0;
+
+  /* Get a shared table lock                  */
+  if(thd->gemini.needSavepoint)
+  {
+    /* We don't really need a savepoint here but do it anyway
+       just to keep the savepoint number correct.             */
+    thd->gemini.savepoint++;
+    error = dsmTransaction((dsmContext_t *)thd->gemini.context,
+                           &thd->gemini.savepoint,
+                           DSMTXN_SAVE, 0, 0);
+    if (error)
+       return(error);
+    thd->gemini.needSavepoint = 0;
+  }
+  buf = my_malloc(table->rec_buff_length,MYF(MY_WME));
+  indexBuf = my_malloc(table->rec_buff_length,MYF(MY_WME));
+  /* Lock the table    */
+  error = dsmObjectLock((dsmContext_t *)thd->gemini.context,
+                        (dsmObject_t)tableNumber,
+                         DSMOBJECT_TABLE,0,
+                         DSM_LK_SHARE, 1, 0);
+  if(error)
+    return error;
+
+  info(HA_STATUS_VARIABLE);
+
+  /* If quick option just scan along index converting and counting entries */
+  for (uint i = 0; i < table->keys; i++)
+  {
+    key_read = 1;
+    indexCount = 0;
+    error = index_init(i);
+    error = index_first(indexBuf);
+    while(!error)
+    {
+      indexCount++;
+      if(!check_opt->quick)
+      {  
+	/* Fetch row and compare to data produced from key */
+	error = fetch_row(thd->gemini.context,buf);
+        if(!error)
+	{
+	  if(key_cmp(i,buf,indexBuf))
+	  {
+	    print_msg(thd,table->real_name,"check","error",
+                    "Key does not match row for rowid %d for index %s",
+		    lastRowid,table->key_info[i].name);
+	    checkStatus = HA_ADMIN_CORRUPT;
+	    errorCount++;
+	    if(errorCount > 1000)
+	      goto error_return;
+	  }
+	  else if(error == DSM_S_RMNOTFND)
+	  {
+	    errorCount++;
+	    checkStatus = HA_ADMIN_CORRUPT;
+            print_msg(thd,table->real_name,"check","error",
+		      "Key does not have a valid row pointer %d for index %s",
+		      lastRowid,table->key_info[i].name);
+            if(errorCount > 1000)
+	      goto error_return;
+	    error = 0;
+	  }
+	}
+      }
+      if(!error)
+	error = index_next(indexBuf);
+    }
+
+    if(error == HA_ERR_END_OF_FILE)
+    {
+      /* Check count of rows   */
+
+      if(records != indexCount)
+      {
+        /* Number of index entries does not agree with the number of
+           rows in the index.                            */
+        checkStatus = HA_ADMIN_CORRUPT;
+        print_msg(thd,table->real_name,"check","error",
+                "Total rows %d does not match total index entries %d for %s",
+                   records, indexCount,
+                    table->key_info[i].name);
+      }
+    }
+    else
+    {
+      checkStatus = HA_ADMIN_FAILED;
+      goto error_return;
+    }
+    index_end();
+  }
+  if(!check_opt->quick)
+  {
+    /* Now scan the table and for each row generate the keys
+       and find them in the index                               */
+    error = fullCheck(thd, buf);\
+    if(error)
+      checkStatus = error;
+  }
+
+error_return:
+  my_free(buf,MYF(MY_ALLOW_ZERO_PTR));
+  index_end();
+  key_read = 0;
+  error = dsmObjectUnlock((dsmContext_t *)thd->gemini.context,
+                        (dsmObject_t)tableNumber,
+                         DSMOBJECT_TABLE,0,
+                         DSM_LK_SHARE,0);
+
+  return checkStatus;
+}
+
+int ha_gemini::fullCheck(THD *thd,byte *buf)
+{
+  int error;
+  int errorCount = 0;
+  int checkStatus = 0;
+
+  lastRowid = 0;
+
+  while(((error = rnd_next( buf)) != HA_ERR_END_OF_FILE) && errorCount <= 1000)
+  {
+    if(!error)
+    {
+      error = handleIndexEntries(buf,lastRowid,KEY_CHECK);
+      if(error)
+	{
+	/* Error finding an index entry for a row.   */
+	print_msg(thd,table->real_name,"check","error",
+		  "Unable to find all index entries for row %d",
+		  lastRowid);
+        errorCount++;
+	checkStatus = HA_ADMIN_CORRUPT;
+	error = 0;
+      }
+    }
+    else
+    {
+      /* Error reading a row          */
+      print_msg(thd,table->real_name,"check","error",
+		  "Error reading row %d status = %d",
+		  lastRowid,error);
+      errorCount++;
+      checkStatus = HA_ADMIN_CORRUPT;
+      error = 0;
+    }
+  }
+
+  return checkStatus;
+}
+
+int ha_gemini::repair(THD* thd,  HA_CHECK_OPT* check_opt)
+{
+  int error;
+  dsmRecord_t  dsmRecord;
+  byte         *buf;
+
+  if(thd->gemini.needSavepoint) 
+  {
+    /* We don't really need a savepoint here but do it anyway
+       just to keep the savepoint number correct.             */
+    thd->gemini.savepoint++;
+    error = dsmTransaction((dsmContext_t *)thd->gemini.context,
+                           &thd->gemini.savepoint,
+                           DSMTXN_SAVE, 0, 0);
+    if (error)
+       return(error);
+    thd->gemini.needSavepoint = 0;
+  }
+
+
+  /* Lock the table    */
+  error = dsmObjectLock((dsmContext_t *)thd->gemini.context,
+                        (dsmObject_t)tableNumber,
+                         DSMOBJECT_TABLE,0,
+                         DSM_LK_EXCL, 1, 0);
+  if(error)
+    return error;
+  
+  error = dsmContextSetLong((dsmContext_t *)thd->gemini.context,
+                             DSM_TAGCONTEXT_NO_LOGGING,1);
+
+  error = dsmTableReset((dsmContext_t *)thd->gemini.context,
+                        (dsmTable_t)tableNumber, table->keys,
+                        pindexNumbers);
+
+  buf = my_malloc(table->rec_buff_length,MYF(MY_WME));
+  dsmRecord.table = tableNumber;
+  dsmRecord.recid = 0;
+  dsmRecord.pbuffer = (dsmBuffer_t *)rec_buff;
+  dsmRecord.recLength = table->reclength;
+  dsmRecord.maxLength = table->reclength;
+  while(!error)
+  {
+    error = dsmTableScan((dsmContext_t *)thd->gemini.context,
+                       &dsmRecord, DSMFINDNEXT, DSM_LK_NOLOCK,
+                        1);
+    if(!error)
+    {
+      unpack_row((char *)buf,(char *)dsmRecord.pbuffer);
+      error = handleIndexEntries(buf,dsmRecord.recid,KEY_CREATE);
+      if(error == HA_ERR_FOUND_DUPP_KEY)
+      {
+	/* We don't want to stop on duplicate keys -- we're repairing 
+	   here so let's get as much repaired as possible.   */
+	error = 0;
+      }
+    }
+  } 
+  error = dsmObjectUnlock((dsmContext_t *)thd->gemini.context,
+                        (dsmObject_t)tableNumber,
+                         DSMOBJECT_TABLE,0,
+                         DSM_LK_EXCL,0);
+  my_free(buf,MYF(MY_ALLOW_ZERO_PTR));  
+ 
+  error = dsmContextSetLong((dsmContext_t *)thd->gemini.context,
+                             DSM_TAGCONTEXT_NO_LOGGING,0);
+    
+  return error;
+}
+
+ha_rows ha_gemini::records_in_range(int keynr,
+				      const byte *start_key,uint start_key_len,
+				      enum ha_rkey_function start_search_flag,
+				      const byte *end_key,uint end_key_len,
+				      enum ha_rkey_function end_search_flag)
+{
+  int   error;
+  int   componentLen;
+  float pctInrange;
+  ha_rows rows = 5;
+
+  DBUG_ENTER("records_in_range");
+
+  error = index_init(keynr);
+  if(error)
+    DBUG_RETURN(rows);
+
+  pbracketBase->index = (short)pindexNumbers[keynr];
+  pbracketBase->keycomps =  1;
+
+  if(start_key)
+  {
+    error = pack_key(keynr, pbracketBase, start_key, start_key_len);  
+    if(start_search_flag == HA_READ_AFTER_KEY)
+    {
+      /* A greater than operation      */
+      error = gemKeyAddLow(pbracketBase->keystr + pbracketBase->keyLen,
+                         &componentLen);
+      pbracketBase->keyLen += componentLen;
+    }
+  }
+  else
+  {
+    error = gemKeyLow(pbracketBase->keystr, &componentLen,
+                      pbracketBase->index);
+    pbracketBase->keyLen = componentLen;
+
+  }
+  pbracketBase->keyLen -= 3;
+
+  if(end_key)
+  {
+    error = pack_key(keynr, pbracketLimit, end_key, end_key_len);
+    if(!error && end_search_flag == HA_READ_AFTER_KEY)
+    {
+      error = gemKeyAddHigh(pbracketLimit->keystr + pbracketLimit->keyLen,
+                            &componentLen);
+      pbracketLimit->keyLen += componentLen; 
+    }
+  }
+  else
+  {
+    error = gemKeyHigh(pbracketLimit->keystr,&componentLen,
+                     pbracketLimit->index);
+    pbracketLimit->keyLen = componentLen;
+  }
+  
+  pbracketLimit->keyLen -= 3;
+  error = dsmIndexRowsInRange((dsmContext_t *)current_thd->gemini.context,
+                              pbracketBase,pbracketLimit,
+                              &pctInrange);
+  if(pctInrange >= 1)
+    rows = (ha_rows)pctInrange;
+  else
+  {
+     rows = (ha_rows)(records * pctInrange);
+     if(!rows && pctInrange > 0)
+        rows = 1;
+  }
+  index_end(); 
+
+  DBUG_RETURN(rows);
+}
+
+
+/*
+  Pack a row for storage.  If the row is of fixed length, just store the
+  row 'as is'.
+  If not, we will generate a packed row suitable for storage.
+  This will only fail if we don't have enough memory to pack the row, which;
+  may only happen in rows with blobs,  as the default row length is
+  pre-allocated.
+*/
+int ha_gemini::pack_row(byte **pprow, int *ppackedLength, const byte *record)
+{
+  if (fixed_length_row)
+  {
+    *pprow = (byte *)record;
+    *ppackedLength=(int)table->reclength;
+    return 0;
+  }
+  if (table->blob_fields)
+  {
+    return HA_ERR_WRONG_COMMAND;
+  }
+  /* Copy null bits */
+  memcpy(rec_buff, record, table->null_bytes);
+  byte *ptr=rec_buff + table->null_bytes;
+
+  for (Field **field=table->field ; *field ; field++)
+    ptr=(byte*) (*field)->pack((char*) ptr,record + (*field)->offset());
+
+  *pprow=rec_buff;
+  *ppackedLength=  (ptr - rec_buff);
+  return 0;
+}
+
+void ha_gemini::unpack_row(char *record, char *prow)
+{
+  if (fixed_length_row)
+  {
+    /* If the table is a VST, the row is in Gemini internal format.
+    ** Convert the fields to MySQL format.
+    */
+    if (RM_IS_VST(tableNumber))
+    {
+      int i = 2; /* VST fields are numbered sequentially starting at 2 */
+      long longValue;
+      char *fld;
+      unsigned long unknown;
+
+      for (Field **field = table->field; *field; field++, i++)
+      {
+        switch ((*field)->type())
+        {
+          case FIELD_TYPE_LONG:
+          case FIELD_TYPE_TINY:
+          case FIELD_TYPE_SHORT:
+          case FIELD_TYPE_TIMESTAMP:
+          case FIELD_TYPE_LONGLONG:
+          case FIELD_TYPE_INT24:
+          case FIELD_TYPE_DATE:
+          case FIELD_TYPE_TIME:
+          case FIELD_TYPE_DATETIME:
+          case FIELD_TYPE_YEAR:
+          case FIELD_TYPE_NEWDATE:
+          case FIELD_TYPE_ENUM:
+          case FIELD_TYPE_SET:
+            recGetLONG((dsmText_t *)prow, i, 0, &longValue, &unknown);
+            if (unknown)
+            {
+              (*field)->set_null();
+            }
+            else
+            {
+              (*field)->set_notnull();
+              (*field)->store((longlong)longValue);
+            }
+            break;
+
+          case FIELD_TYPE_DECIMAL:
+          case FIELD_TYPE_DOUBLE:
+          case FIELD_TYPE_TINY_BLOB:
+          case FIELD_TYPE_MEDIUM_BLOB:
+          case FIELD_TYPE_LONG_BLOB:
+          case FIELD_TYPE_BLOB:
+          case FIELD_TYPE_VAR_STRING:
+            break;
+
+          case FIELD_TYPE_STRING:
+            svcByteString_t stringFld;
+
+            fld = (char *)my_malloc((*field)->field_length, MYF(MY_WME));
+            stringFld.pbyte = (TEXT *)fld;
+            stringFld.size = (*field)->field_length;
+            recGetBYTES((dsmText_t *)prow, i, 0, &stringFld, &unknown);
+            if (unknown)
+            {
+              (*field)->set_null();
+            }
+            else
+            {
+              (*field)->set_notnull();
+              (*field)->store(fld, (*field)->field_length);
+            }
+            my_free(fld, MYF(MY_ALLOW_ZERO_PTR));
+            break;
+
+          default:
+            break;
+        }
+      }
+    }
+    else
+    {
+      memcpy(record,(char*) prow,table->reclength);
+    }
+  }
+  else
+  {
+    /* Copy null bits */
+    const char *ptr= (const char*) prow;
+    memcpy(record, ptr, table->null_bytes);
+    ptr+=table->null_bytes;
+    for (Field **field=table->field ; *field ; field++)
+      ptr= (*field)->unpack(record + (*field)->offset(), ptr);
+  }
+}
+
+int ha_gemini::key_cmp(uint keynr, const byte * old_row,
+                         const byte * new_row)
+{
+  KEY_PART_INFO *key_part=table->key_info[keynr].key_part;
+  KEY_PART_INFO *end=key_part+table->key_info[keynr].key_parts;
+
+  for ( ; key_part != end ; key_part++)
+  {
+    if (key_part->null_bit)
+    {
+      if ((old_row[key_part->null_offset] & key_part->null_bit) !=
+          (new_row[key_part->null_offset] & key_part->null_bit))
+        return 1;
+    }
+    if (key_part->key_part_flag & (HA_BLOB_PART | HA_VAR_LENGTH))
+    {
+
+      if (key_part->field->cmp_binary(old_row + key_part->offset,
+                                      new_row + key_part->offset,
+                                      (ulong) key_part->length))
+        return 1;
+    }
+    else
+    {
+      if (memcmp(old_row+key_part->offset, new_row+key_part->offset,
+                 key_part->length))
+        return 1;
+    }
+  }
+  return 0;
+}
+
+int gemini_parse_table_name(const char *fullname, char *dbname, char *tabname)
+{
+  char *namestart;
+  char *nameend;
+
+  /* separate out the name of the table and the database
+  */
+  namestart = strchr(fullname + start_of_name, '/');
+  if (!namestart)
+  {
+    /* if on Windows, slashes go the other way */
+    namestart = strchr(fullname + start_of_name, '\\');
+  }
+  nameend = strchr(fullname + start_of_name, '.');
+  /* sometimes fullname has an extension, sometimes it doesn't */
+  if (!nameend)
+  {
+    nameend = (char *)fullname + strlen(fullname);
+  }
+  strncpy(dbname, fullname + start_of_name,
+          (namestart - fullname) - start_of_name);
+  dbname[(namestart - fullname) - start_of_name] = '\0';
+  strncpy(tabname, namestart + 1, (nameend - namestart) - 1);
+  tabname[nameend - namestart - 1] = '\0';
+
+  return 0;
+}
+
+/* PROGRAM: gemini_is_vst - if the name is the name of a VST, return
+ *                          its number
+ *
+ * RETURNS:   Table number if a match is found
+ *            0 if not a VST
+ */
+int
+gemini_is_vst(const char *pname) /* IN the name */
+{
+    int tablenum = 0;
+
+    for (int i = 0; i < vstnumfils; i++)
+    {
+        if (strcmp(pname, vstfil[i].filename) == 0)
+        {
+            tablenum = vstfil[i].filnum;
+            break;
+        }
+    }
+
+    return tablenum;
+}
+
+static void print_msg(THD *thd, const char *table_name, const char *op_name,
+                      const char *msg_type, const char *fmt, ...)
+{
+  String* packet = &thd->packet;
+  packet->length(0);
+  char msgbuf[256];
+  msgbuf[0] = 0;
+  va_list args;
+  va_start(args,fmt);
+
+  my_vsnprintf(msgbuf, sizeof(msgbuf), fmt, args);
+  msgbuf[sizeof(msgbuf) - 1] = 0; // healthy paranoia
+
+  DBUG_PRINT(msg_type,("message: %s",msgbuf));
+
+  net_store_data(packet, table_name);
+  net_store_data(packet, op_name);
+  net_store_data(packet, msg_type);
+  net_store_data(packet, msgbuf);
+  if (my_net_write(&thd->net, (char*)thd->packet.ptr(),
+                   thd->packet.length()))
+    thd->killed=1;
+}
+
+#endif /* HAVE_GEMINI_DB */