~drizzle-trunk/drizzle/development

/* Copyright (C) 2000-2006 MySQL AB

   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; version 2 of the License.

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

/*

  Functions to create a unireg form-file from a FIELD and a fieldname-fieldinfo

  struct.

  In the following functions FIELD * is an ordinary field-structure with

  the following exeptions:

    sc_length,typepos,row,kol,dtype,regnr and field need not to be set.

    str is a (long) to record position where 0 is the first position.

*/

#include <drizzled/server_includes.h>

#include <drizzled/error.h>

#include <drizzled/virtual_column_info.h>

#include <drizzled/session.h>

#include <drizzled/unireg.h>

/* For proto */

#include <string>

#include <fstream>

#include <drizzled/serialize/serialize.h>

using namespace std;

#define FCOMP			17		/* Bytes for a packed field */

static unsigned char * pack_screens(List<Create_field> &create_fields,

                                    uint32_t *info_length, uint32_t *screens,

                                    bool small_file);

static uint32_t pack_keys(unsigned char *keybuff,uint32_t key_count,

                          KEY *key_info, ulong data_offset);

static bool pack_header(unsigned char *forminfo,

                        List<Create_field> &create_fields,

                        uint32_t info_length, uint32_t screens,

                        uint32_t table_options,

                        ulong data_offset, handler *file);

static uint32_t get_interval_id(uint32_t *int_count,

                                List<Create_field> &create_fields,

                                Create_field *last_field);

static bool pack_fields(File file, List<Create_field> &create_fields,

                        ulong data_offset);

static bool make_empty_rec(Session *session, int file,

                           enum legacy_db_type table_type,

                           uint32_t table_options,

                           List<Create_field> &create_fields,

                           uint32_t reclength, ulong data_offset,

                           handler *handler);

/**

  An interceptor to hijack ER_TOO_MANY_FIELDS error from

  pack_screens and retry again without UNIREG screens.

  XXX: what is a UNIREG  screen?

*/

struct Pack_header_error_handler: public Internal_error_handler

{

  virtual bool handle_error(uint32_t sql_errno,

                            const char *message,

                            DRIZZLE_ERROR::enum_warning_level level,

                            Session *session);

  bool is_handled;

  Pack_header_error_handler() :is_handled(false) {}

};

bool

Pack_header_error_handler::

handle_error(uint32_t sql_errno,

             const char * /* message */,

             DRIZZLE_ERROR::enum_warning_level /* level */,

             Session * /* session */)

{

  is_handled= (sql_errno == ER_TOO_MANY_FIELDS);

  return is_handled;

}

/*

  Create a frm (table definition) file

  SYNOPSIS

    mysql_create_frm()

    session			Thread handler

    file_name		Path for file (including database and .frm)

    db                  Name of database

    table               Name of table

    create_info		create info parameters

    create_fields	Fields to create

    keys		number of keys to create

    key_info		Keys to create

    db_file		Handler to use. May be zero, in which case we use

			create_info->db_type

  RETURN

    0  ok

    1  error

*/

bool mysql_create_frm(Session *session, const char *file_name,

                      const char *db, const char *table,

		      HA_CREATE_INFO *create_info,

		      List<Create_field> &create_fields,

		      uint32_t keys, KEY *key_info,

		      handler *db_file)

{

  LEX_STRING str_db_type;

  uint32_t reclength, info_length, screens, key_info_length, maxlength, tmp_len;

  ulong key_buff_length;

  File file;

  ulong filepos, data_offset;

  unsigned char fileinfo[64],forminfo[288],*keybuff;

  TYPELIB formnames;

  unsigned char *screen_buff;

  char buff[128];

  const uint32_t format_section_header_size= 8;

  uint32_t format_section_len;

  Pack_header_error_handler pack_header_error_handler;

  int error;

  assert(*fn_rext((char*)file_name)); // Check .frm extension

  formnames.type_names=0;

  if (!(screen_buff=pack_screens(create_fields,&info_length,&screens,0)))

    return(1);

  assert(db_file != NULL);

 /* If fixed row records, we need one bit to check for deleted rows */

  if (!(create_info->table_options & HA_OPTION_PACK_RECORD))

    create_info->null_bits++;

  data_offset= (create_info->null_bits + 7) / 8;

  session->push_internal_handler(&pack_header_error_handler);

  error= pack_header(forminfo,

                     create_fields,info_length,

                     screens, create_info->table_options,

                     data_offset, db_file);

  session->pop_internal_handler();

  if (error)

  {

    free(screen_buff);

    if (! pack_header_error_handler.is_handled)

      return(1);

    // Try again without UNIREG screens (to get more columns)

    if (!(screen_buff=pack_screens(create_fields,&info_length,&screens,1)))

      return(1);

    if (pack_header(forminfo,

                    create_fields,info_length,

		    screens, create_info->table_options, data_offset, db_file))

    {

      free(screen_buff);

      return(1);

    }

  }

  reclength=uint2korr(forminfo+266);

  /* Calculate extra data segment length */

  str_db_type.str= (char *) ha_resolve_storage_engine_name(create_info->db_type);

  str_db_type.length= strlen(str_db_type.str);

  /* str_db_type */

  create_info->extra_size= (2 + str_db_type.length +

                            2 + create_info->connect_string.length);

  /*

    Partition:

      Length of partition info = 4 byte

      Potential NULL byte at end of partition info string = 1 byte

      Indicator if auto-partitioned table = 1 byte

      => Total 6 byte

  */

  create_info->extra_size+= 6;

  /* Add space for storage type and field format array of fields */

  format_section_len=

    format_section_header_size + 1 + create_fields.elements;

  create_info->extra_size+= format_section_len;

  tmp_len= system_charset_info->cset->charpos(system_charset_info,

                                              create_info->comment.str,

                                              create_info->comment.str +

                                              create_info->comment.length,

                                              TABLE_COMMENT_MAXLEN);

  if (tmp_len < create_info->comment.length)

  {

    my_error(ER_WRONG_STRING_LENGTH, MYF(0),

             create_info->comment.str,"Table COMMENT",

             (uint) TABLE_COMMENT_MAXLEN);

    free(screen_buff);

    return(1);

  }

  //if table comment is larger than 180 bytes, store into extra segment.

  if (create_info->comment.length > 180)

  {

    forminfo[46]=255;

    create_info->extra_size+= 2 + create_info->comment.length;

  }

  else{

    strncpy((char*) forminfo+47, create_info->comment.str ?

            create_info->comment.str : "", create_info->comment.length);

    forminfo[46]=(unsigned char) create_info->comment.length;

#ifdef EXTRA_DEBUG

    /*

      EXTRA_DEBUG causes strmake() to initialize its buffer behind the

      payload with a magic value to detect wrong buffer-sizes. We

      explicitly zero that segment again.

    */

    memset(forminfo+47 + forminfo[46], 0, 61 - forminfo[46]);

#endif

  }

  if ((file=create_frm(session, file_name, db, table, reclength, fileinfo,

		       create_info, keys, key_info)) < 0)

  {

    free(screen_buff);

    return(1);

  }

  key_buff_length= uint4korr(fileinfo+47);

  keybuff=(unsigned char*) malloc(key_buff_length);

  key_info_length= pack_keys(keybuff, keys, key_info, data_offset);

  get_form_pos(file,fileinfo,&formnames);

  if (!(filepos=make_new_entry(file,fileinfo,&formnames,"")))

    goto err;

  maxlength=(uint) next_io_size((ulong) (uint2korr(forminfo)+1000));

  int2store(forminfo+2,maxlength);

  int4store(fileinfo+10,(ulong) (filepos+maxlength));

  fileinfo[26]= (unsigned char) test((create_info->max_rows == 1) &&

			     (create_info->min_rows == 1) && (keys == 0));

  int2store(fileinfo+28,key_info_length);

  int2store(fileinfo+59,db_file->extra_rec_buf_length());

  if (pwrite(file, fileinfo, 64, 0L) == 0 ||

      pwrite(file, keybuff, key_info_length, (ulong) uint2korr(fileinfo+6)) == 0)

    goto err;

  lseek(file, (off_t)(uint2korr(fileinfo+6) + key_buff_length), SEEK_SET);

  if (make_empty_rec(session,file,ha_legacy_type(create_info->db_type),

                     create_info->table_options,

		     create_fields,reclength, data_offset, db_file))

    goto err;

  int2store(buff, create_info->connect_string.length);

  if (my_write(file, (const unsigned char*)buff, 2, MYF(MY_NABP)) ||

      my_write(file, (const unsigned char*)create_info->connect_string.str,

               create_info->connect_string.length, MYF(MY_NABP)))

      goto err;

  int2store(buff, str_db_type.length);

  if (my_write(file, (const unsigned char*)buff, 2, MYF(MY_NABP)) ||

      my_write(file, (const unsigned char*)str_db_type.str,

               str_db_type.length, MYF(MY_NABP)))

    goto err;

  {

    memset(buff, 0, 6);

    if (my_write(file, (unsigned char*) buff, 6, MYF_RW))

      goto err;

  }

  if (forminfo[46] == (unsigned char)255)

  {

    unsigned char comment_length_buff[2];

    int2store(comment_length_buff,create_info->comment.length);

    if (my_write(file, comment_length_buff, 2, MYF(MY_NABP)) ||

        my_write(file, (unsigned char*)create_info->comment.str,

                  create_info->comment.length, MYF(MY_NABP)))

      goto err;

  }

  /* Store storage type and field format array of fields */

  {

    /* prepare header */

    {

      uint32_t flags= 0;

      memset(buff, 0, format_section_header_size);

      /* length of section 2 bytes*/

      int2store(buff+0, format_section_len);

      /* flags of section 4 bytes*/

      int4store(buff+2, flags);

      /* 2 bytes left for future use */

    }

    /* write header */

    if (my_write(file, (const unsigned char*)buff, format_section_header_size, MYF_RW))

      goto err;

    buff[0]= 0;

    if (my_write(file, (const unsigned char*)buff, 1, MYF_RW))

      goto err;

    /* write column info, 1 byte per column */

    {

      List_iterator<Create_field> it(create_fields);

      Create_field *field;

      unsigned char column_format, write_byte;

      while ((field=it++))

      {

        column_format= (unsigned char)field->column_format();

        write_byte= (column_format << COLUMN_FORMAT_SHIFT);

        if (my_write(file, &write_byte, 1, MYF_RW))

          goto err;

      }

    }

  }

  lseek(file,filepos,SEEK_SET);

  if (my_write(file, forminfo, 288, MYF_RW) ||

      my_write(file, screen_buff, info_length, MYF_RW) ||

      pack_fields(file, create_fields, data_offset))

    goto err;

  free(screen_buff);

  free(keybuff);

  if (!(create_info->options & HA_LEX_CREATE_TMP_TABLE) &&

      (my_sync(file, MYF(MY_WME)) ||

       my_sync_dir_by_file(file_name, MYF(MY_WME))))

      goto err2;

  if (my_close(file,MYF(MY_WME)))

    goto err3;

  {

    /*

      Restore all UCS2 intervals.

      HEX representation of them is not needed anymore.

    */

    List_iterator<Create_field> it(create_fields);

    Create_field *field;

    while ((field=it++))

    {

      if (field->save_interval)

      {

        field->interval= field->save_interval;

        field->save_interval= 0;

      }

    }

  }

  return(0);

err:

  free(screen_buff);

  free(keybuff);

err2:

  my_close(file,MYF(MY_WME));

err3:

  my_delete(file_name,MYF(0));

  return(1);

} /* mysql_create_frm */

static void fill_table_proto(drizzle::Table *table_proto,

                             const char *table_name,

                             List<Create_field> &create_fields,

                             HA_CREATE_INFO *create_info,

                             uint32_t keys,

                             KEY *key_info)

{

  Create_field *field_arg;

  List_iterator<Create_field> it(create_fields);

  drizzle::Table::StorageEngine *engine= table_proto->mutable_engine();

  drizzle::Table::TableOptions *table_options= table_proto->mutable_options();

  engine->set_name(create_info->db_type->name);

  table_proto->set_name(table_name);

  table_proto->set_type(drizzle::Table::STANDARD);

  while ((field_arg= it++))

  {

    drizzle::Table::Field *attribute;

    //drizzle::Table::Field::FieldConstraints *constraints;

    attribute= table_proto->add_field();

    attribute->set_name(field_arg->field_name);

    switch (field_arg->sql_type) {

    case DRIZZLE_TYPE_TINY:

      attribute->set_type(drizzle::Table::Field::TINYINT);

      break;

    case DRIZZLE_TYPE_LONG:

      attribute->set_type(drizzle::Table::Field::INTEGER);

      break;

    case DRIZZLE_TYPE_DOUBLE:

      attribute->set_type(drizzle::Table::Field::DOUBLE);

      break;

    case DRIZZLE_TYPE_NULL  :

      assert(1); /* Not a user definable type */

    case DRIZZLE_TYPE_TIMESTAMP:

      attribute->set_type(drizzle::Table::Field::TIMESTAMP);

      break;

    case DRIZZLE_TYPE_LONGLONG:

      attribute->set_type(drizzle::Table::Field::BIGINT);

      break;

    case DRIZZLE_TYPE_TIME:

      attribute->set_type(drizzle::Table::Field::TIME);

      break;

    case DRIZZLE_TYPE_DATETIME:

      attribute->set_type(drizzle::Table::Field::DATETIME);

      break;

    case DRIZZLE_TYPE_DATE:

      attribute->set_type(drizzle::Table::Field::DATE);

      break;

    case DRIZZLE_TYPE_VARCHAR:

      {

        drizzle::Table::Field::StringFieldOptions *string_field_options;

        string_field_options= attribute->mutable_string_options();

        attribute->set_type(drizzle::Table::Field::VARCHAR);

        string_field_options->set_length(field_arg->char_length);

        string_field_options->set_collation_id(field_arg->charset->number);

        string_field_options->set_collation(field_arg->charset->name);

        break;

      }

    case DRIZZLE_TYPE_NEWDECIMAL:

      {

        drizzle::Table::Field::NumericFieldOptions *numeric_field_options;

        attribute->set_type(drizzle::Table::Field::DECIMAL);

        numeric_field_options= attribute->mutable_numeric_options();

        /* This is magic, I hate magic numbers -Brian */

        numeric_field_options->set_precision(field_arg->length + ( field_arg->decimals ? -2 : -1));

        numeric_field_options->set_scale(field_arg->decimals);

        break;

      }

    case DRIZZLE_TYPE_ENUM:

      {

        drizzle::Table::Field::SetFieldOptions *set_field_options;

        assert(field_arg->interval);

        attribute->set_type(drizzle::Table::Field::ENUM);

        set_field_options= attribute->mutable_set_options();

        for (uint32_t pos= 0; pos < field_arg->interval->count; pos++)

        {

          const char *src= field_arg->interval->type_names[pos];

          set_field_options->add_value(src);

        }

        set_field_options->set_count_elements(set_field_options->value_size());

        break;

      }

    case DRIZZLE_TYPE_BLOB:

      attribute->set_type(drizzle::Table::Field::BLOB);

      break;

    default:

      assert(1);

    }

#ifdef NOTDONE

    field_constraints= attribute->mutable_constraints();

    constraints->set_is_nullable(field_arg->def->null_value);

#endif

    /* Set the comment */

    if (field_arg->comment.length)

      attribute->set_comment(field_arg->comment.str);

  }

  if (create_info->comment.length)

    table_options->set_comment(create_info->comment.str);

  if (create_info->table_charset)

  {

    table_options->set_collation_id(field_arg->charset->number);

    table_options->set_collation(field_arg->charset->name);

  }

  if (create_info->connect_string.length)

    table_options->set_connect_string(create_info->connect_string.str);

  if (create_info->data_file_name)

    table_options->set_data_file_name(create_info->data_file_name);

  if (create_info->index_file_name)

    table_options->set_index_file_name(create_info->index_file_name);

  if (create_info->max_rows)

    table_options->set_max_rows(create_info->max_rows);

  if (create_info->min_rows)

    table_options->set_min_rows(create_info->min_rows);

  if (create_info->auto_increment_value)

    table_options->set_auto_increment_value(create_info->auto_increment_value);

  if (create_info->avg_row_length)

    table_options->set_avg_row_length(create_info->avg_row_length);

  if (create_info->key_block_size)

    table_options->set_key_block_size(create_info->key_block_size);

  if (create_info->block_size)

    table_options->set_block_size(create_info->block_size);

  for (unsigned int i= 0; i < keys; i++)

  {

    drizzle::Table::Index *idx;

    idx= table_proto->add_index();

    assert(test(key_info[i].flags & HA_USES_COMMENT) ==

           (key_info[i].comment.length > 0));

    idx->set_name(key_info[i].name);

    if(is_primary_key_name(key_info[i].name))

      idx->set_is_primary(true);

    else

      idx->set_is_primary(false);

    switch(key_info[i].algorithm)

    {

    case HA_KEY_ALG_HASH:

      idx->set_type(drizzle::Table::Index::HASH);

      break;

    case HA_KEY_ALG_BTREE:

      idx->set_type(drizzle::Table::Index::BTREE);

      break;

    case HA_KEY_ALG_RTREE:

    case HA_KEY_ALG_FULLTEXT:

    case HA_KEY_ALG_UNDEF:

      idx->set_type(drizzle::Table::Index::UNKNOWN_INDEX);

      break;

    default:

      abort(); /* Somebody's brain broke. haven't added index type to proto */

      break;

    }

    if (key_info[i].flags & HA_NOSAME)

      idx->set_is_unique(true);

    else

      idx->set_is_unique(false);

    /* FIXME: block_size ? */

    for(unsigned int j=0; j< key_info[i].key_parts; j++)

    {

      drizzle::Table::Index::IndexPart *idxpart;

      drizzle::Table::Field *field;

      idxpart= idx->add_index_part();

      field= idxpart->mutable_field();

      *field= table_proto->field(key_info[i].key_part[j].fieldnr);

      idxpart->set_compare_length(key_info[i].key_part[j].length);

    }

    if (key_info[i].flags & HA_USES_COMMENT)

      idx->set_comment(key_info[i].comment.str);

  }

}

int rename_table_proto_file(const char *from, const char* to)

{

  string from_path(from);

  string to_path(to);

  string file_ext = ".tabledefinition";

  from_path.append(file_ext);

  to_path.append(file_ext);

  return my_rename(from_path.c_str(),to_path.c_str(),MYF(MY_WME));

}

int delete_table_proto_file(char *file_name)

{

  string new_path(file_name);

  string file_ext = ".tabledefinition";

  new_path.append(file_ext);

  return my_delete(new_path.c_str(), MYF(0));

}

int create_table_proto_file(char *file_name,

                            const char *table_name,

                            HA_CREATE_INFO *create_info,

                            List<Create_field> &create_fields,

                            uint32_t keys,

                            KEY *key_info)

{

  (void)key_info;

  drizzle::Table table_proto;

  string new_path(file_name);

  string file_ext = ".tabledefinition";

  fill_table_proto(&table_proto, table_name, create_fields, create_info,

                   keys, key_info);

  new_path.replace(new_path.find(".frm"), file_ext.length(), file_ext );

  fstream output(new_path.c_str(), ios::out | ios::trunc | ios::binary);

  if (!table_proto.SerializeToOstream(&output))

  {

    printf("Failed to write schema.\n");

    fprintf(stderr, "Failed to write schema.\n");

    return -1;

  }

  return 0;

}

/*

  Create a frm (table definition) file and the tables

  SYNOPSIS

    rea_create_table()

    session			Thread handler

    path		Name of file (including database, without .frm)

    db			Data base name

    table_name		Table name

    create_info		create info parameters

    create_fields	Fields to create

    keys		number of keys to create

    key_info		Keys to create

    file		Handler to use

  RETURN

    0  ok

    1  error

*/

int rea_create_table(Session *session, const char *path,

                     const char *db, const char *table_name,

                     HA_CREATE_INFO *create_info,

                     List<Create_field> &create_fields,

                     uint32_t keys, KEY *key_info, handler *file)

{

  char frm_name[FN_REFLEN];

  strxmov(frm_name, path, reg_ext, NULL);

  if (mysql_create_frm(session, frm_name, db, table_name, create_info,

                       create_fields, keys, key_info, file))

    return(1);

  if (create_table_proto_file(frm_name, table_name, create_info,

                              create_fields, keys, key_info) != 0)

    return 1;

  // Make sure mysql_create_frm din't remove extension

  assert(*fn_rext(frm_name));

  if (session->variables.keep_files_on_create)

    create_info->options|= HA_CREATE_KEEP_FILES;

  if (file->ha_create_handler_files(path, NULL, CHF_CREATE_FLAG, create_info))

    goto err_handler;

  if ( ha_create_table(session, path, db, table_name,

                                                create_info,0))

    goto err_handler;

  return(0);

err_handler: