~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., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA */

#include <config.h>

#include <drizzled/error.h>

#include <drizzled/session.h>

#include <drizzled/sql_table.h>

#include <drizzled/message/statement_transform.h>

#include <drizzled/plugin/storage_engine.h>

#include <drizzled/internal/my_sys.h>

#include <drizzled/typelib.h>

#include <drizzled/util/test.h>

/* For proto */

#include <string>

#include <fstream>

#include <fcntl.h>

#include <drizzled/message/schema.h>

#include <drizzled/message/table.h>

#include <google/protobuf/io/zero_copy_stream.h>

#include <google/protobuf/io/zero_copy_stream_impl.h>

#include <google/protobuf/message.h>

#include <drizzled/table_proto.h>

#include <drizzled/charset.h>

#include <drizzled/create_field.h>

#include <drizzled/function/time/typecast.h>

using namespace std;

namespace drizzled {

static

bool fill_table_proto(const identifier::Table& identifier,

                      message::Table &table_proto,

                      List<CreateField> &create_fields,

                      HA_CREATE_INFO *create_info,

                      uint32_t keys,

                      KeyInfo *key_info)

{

  CreateField *field_arg;

  List<CreateField>::iterator it(create_fields.begin());

  message::Table::TableOptions *table_options= table_proto.mutable_options();

  if (create_fields.size() > MAX_FIELDS)

  {

    my_error(ER_TOO_MANY_FIELDS, MYF(0), ER(ER_TOO_MANY_FIELDS));

    return true;

  }

  assert(strcmp(table_proto.engine().name().c_str(),

		create_info->db_type->getName().c_str())==0);

  message::schema::shared_ptr schema_message= plugin::StorageEngine::getSchemaDefinition(identifier);

  if (schema_message and not message::is_replicated(*schema_message))

  {

    message::set_is_replicated(table_proto, false);

  }

  int field_number= 0;

  bool use_existing_fields= table_proto.field_size() > 0;

  while ((field_arg= it++))

  {

    message::Table::Field *attribute;

    /* some (one) code path for CREATE TABLE fills the proto

       out more than the others, so we already have partially

       filled out Field messages */

    if (use_existing_fields)

    {

      attribute= table_proto.mutable_field(field_number++);

    }

    else

    {

      /* Other code paths still have to fill out the proto */

      attribute= table_proto.add_field();

      if (field_arg->flags & NOT_NULL_FLAG)

      {

        attribute->mutable_constraints()->set_is_notnull(true);

      }

      if (field_arg->flags & UNSIGNED_FLAG and 

          (field_arg->sql_type == DRIZZLE_TYPE_LONGLONG or field_arg->sql_type == DRIZZLE_TYPE_LONG))

      {

        field_arg->sql_type= DRIZZLE_TYPE_LONGLONG;

        attribute->mutable_constraints()->set_is_unsigned(true);

      }

      attribute->set_name(field_arg->field_name);

    }

    assert(((field_arg->flags & NOT_NULL_FLAG)) == attribute->constraints().is_notnull());

    assert(strcmp(attribute->name().c_str(), field_arg->field_name)==0);

    message::Table::Field::FieldType parser_type= attribute->type();

    if (field_arg->sql_type == DRIZZLE_TYPE_NULL)

    {

      my_error(ER_CANT_CREATE_TABLE, MYF(ME_BELL+ME_WAITTANG), table_proto.name().c_str(), -1);

      return true;

    }

    if (field_arg->flags & UNSIGNED_FLAG and 

       (field_arg->sql_type == DRIZZLE_TYPE_LONGLONG or field_arg->sql_type == DRIZZLE_TYPE_LONG))

    {

      message::Table::Field::FieldConstraints *constraints= attribute->mutable_constraints();

      field_arg->sql_type= DRIZZLE_TYPE_LONGLONG;

      constraints->set_is_unsigned(true);

    }

    attribute->set_type(message::internalFieldTypeToFieldProtoType(field_arg->sql_type));

    switch (attribute->type()) {

    case message::Table::Field::BIGINT:

    case message::Table::Field::INTEGER:

    case message::Table::Field::DATE:

    case message::Table::Field::DATETIME:

    case message::Table::Field::UUID:

    case message::Table::Field::TIME:

    case message::Table::Field::BOOLEAN:

      break;

    case message::Table::Field::DOUBLE:

      {

        /*

         * For DOUBLE, we only add a specific scale and precision iff

         * the fixed decimal point has been specified...

         */

        if (field_arg->decimals != NOT_FIXED_DEC)

        {

          message::Table::Field::NumericFieldOptions *numeric_field_options;

          numeric_field_options= attribute->mutable_numeric_options();

          numeric_field_options->set_precision(field_arg->length);

          numeric_field_options->set_scale(field_arg->decimals);

        }

      }

      break;

    case message::Table::Field::VARCHAR:

      {

        message::Table::Field::StringFieldOptions *string_field_options;

        string_field_options= attribute->mutable_string_options();

        if (! use_existing_fields || string_field_options->length()==0)

          string_field_options->set_length(field_arg->length

                                           / field_arg->charset->mbmaxlen);

        else

          assert((uint32_t)string_field_options->length() == (uint32_t)(field_arg->length / field_arg->charset->mbmaxlen));

        if (! string_field_options->has_collation())

        {

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

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

        }

        break;

      }

    case message::Table::Field::DECIMAL:

      {

        message::Table::Field::NumericFieldOptions *numeric_field_options;

        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 message::Table::Field::ENUM:

      {

        message::Table::Field::EnumerationValues *enumeration_options;

        assert(field_arg->interval);

        enumeration_options= attribute->mutable_enumeration_values();

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

        {

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

          enumeration_options->add_field_value(src);

        }

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

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

        break;

      }

    case message::Table::Field::BLOB:

      {

        message::Table::Field::StringFieldOptions *string_field_options;

        string_field_options= attribute->mutable_string_options();

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

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

      }

      break;

    case message::Table::Field::EPOCH:

      {

        if (field_arg->sql_type == DRIZZLE_TYPE_MICROTIME)

          attribute->mutable_time_options()->set_microseconds(true);

      }

      break;

    }

    assert (!use_existing_fields || parser_type == attribute->type());

#ifdef NOTDONE

    field_constraints= attribute->mutable_constraints();

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

#endif

    if (field_arg->comment.length)

    {

      uint32_t tmp_len;

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

						  field_arg->comment.str,

						  field_arg->comment.str +

						  field_arg->comment.length,

						  COLUMN_COMMENT_MAXLEN);

      if (tmp_len < field_arg->comment.length)

      {

	my_error(ER_WRONG_STRING_LENGTH, MYF(0),

		 field_arg->comment.str,"COLUMN COMMENT",

		 (uint32_t) COLUMN_COMMENT_MAXLEN);

	return true;

      }

      if (! use_existing_fields)

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

      assert(strcmp(attribute->comment().c_str(), field_arg->comment.str)==0);

    }

    if (field_arg->unireg_check == Field::NEXT_NUMBER)

    {

      message::Table::Field::NumericFieldOptions *field_options;

      field_options= attribute->mutable_numeric_options();

      field_options->set_is_autoincrement(true);

    }

    if (field_arg->unireg_check == Field::TIMESTAMP_DN_FIELD

       || field_arg->unireg_check == Field::TIMESTAMP_DNUN_FIELD)

    {

      message::Table::Field::FieldOptions *field_options;

      field_options= attribute->mutable_options();

      field_options->set_default_expression("CURRENT_TIMESTAMP");

    }

    if (field_arg->unireg_check == Field::TIMESTAMP_UN_FIELD

       || field_arg->unireg_check == Field::TIMESTAMP_DNUN_FIELD)

    {

      message::Table::Field::FieldOptions *field_options;

      field_options= attribute->mutable_options();

      field_options->set_update_expression("CURRENT_TIMESTAMP");

    }

    if (field_arg->def == NULL  && not attribute->constraints().is_notnull())

    {

      message::Table::Field::FieldOptions *field_options;

      field_options= attribute->mutable_options();

      field_options->set_default_null(true);

    }

    if (field_arg->def)

    {

      message::Table::Field::FieldOptions *field_options;

      field_options= attribute->mutable_options();

      if (field_arg->def->is_null())

      {

	field_options->set_default_null(true);

      }

      else

      {

	String d;

	String *default_value= field_arg->def->val_str(&d);

	assert(default_value);

	if ((field_arg->sql_type==DRIZZLE_TYPE_VARCHAR

	   || field_arg->sql_type==DRIZZLE_TYPE_BLOB)

	   && ((field_arg->length / field_arg->charset->mbmaxlen)

	   < default_value->length()))

	{

	  my_error(ER_INVALID_DEFAULT, MYF(0), field_arg->field_name);

	  return true;

	}

        if (field::isDateTime(field_arg->sql_type))

        {

          type::Time ltime;

          if (field_arg->def->get_date(ltime, TIME_FUZZY_DATE))

          {

            my_error(ER_INVALID_DATETIME_VALUE, MYF(ME_FATALERROR),

                     default_value->c_str());

            return true;

          }

          /* We now do the casting down to the appropriate type.

             Yes, this implicit casting is balls.

             It was previously done on reading the proto back in,

             but we really shouldn't store the bogus things in the proto,

             and instead do the casting behaviour here.

             the timestamp errors are taken care of elsewhere.

          */

          if (field_arg->sql_type == DRIZZLE_TYPE_DATETIME)

          {

            Item *typecast= new Item_datetime_typecast(field_arg->def);

            typecast->quick_fix_field();

            typecast->val_str(default_value);

          }

          else if (field_arg->sql_type == DRIZZLE_TYPE_DATE)

          {

            Item *typecast= new Item_date_typecast(field_arg->def);

            typecast->quick_fix_field();

            typecast->val_str(default_value);

          }

        }

	if ((field_arg->sql_type == DRIZZLE_TYPE_VARCHAR && field_arg->charset == &my_charset_bin)

	   || (field_arg->sql_type == DRIZZLE_TYPE_BLOB && field_arg->charset == &my_charset_bin))

	{

	  field_options->set_default_bin_value(string(default_value->c_ptr(), default_value->length()));

	}

	else

	{

	  field_options->set_default_value(default_value->c_ptr());

	}

      }

    }

    assert(field_arg->unireg_check == Field::NONE

	   || field_arg->unireg_check == Field::NEXT_NUMBER

	   || field_arg->unireg_check == Field::TIMESTAMP_DN_FIELD

	   || field_arg->unireg_check == Field::TIMESTAMP_UN_FIELD

	   || field_arg->unireg_check == Field::TIMESTAMP_DNUN_FIELD);

  }

  assert(! use_existing_fields || (field_number == table_proto.field_size()));

  if (create_info->table_options & HA_OPTION_PACK_RECORD)

    table_options->set_pack_record(true);

  if (table_options->has_comment() && table_options->comment().length() == 0)

    table_options->clear_comment();

  if (table_options->has_comment())

  {

    uint32_t tmp_len;

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

                                                table_options->comment().c_str(),

                                                table_options->comment().c_str() +

                                                table_options->comment().length(),

                                                TABLE_COMMENT_MAXLEN);

    if (tmp_len < table_options->comment().length())

    {

      my_error(ER_WRONG_STRING_LENGTH, MYF(0),

               table_options->comment().c_str(),"Table COMMENT",

               (uint32_t) TABLE_COMMENT_MAXLEN);

      return true;

    }

  }

  if (create_info->default_table_charset)

  {

    table_options->set_collation_id(create_info->default_table_charset->number);

    table_options->set_collation(create_info->default_table_charset->name);

  }

  if (create_info->used_fields & HA_CREATE_USED_AUTO)

    table_options->set_has_user_set_auto_increment_value(true);

  else

    table_options->set_has_user_set_auto_increment_value(false);

  if (create_info->auto_increment_value)

    table_options->set_auto_increment_value(create_info->auto_increment_value);

  for (uint32_t i= 0; i < keys; i++)

  {

    message::Table::Index *idx;

    idx= table_proto.add_indexes();

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

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

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

    idx->set_key_length(key_info[i].key_length);

    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(message::Table::Index::HASH);

      break;

    case HA_KEY_ALG_BTREE:

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

      break;

    case HA_KEY_ALG_UNDEF:

      {

        idx->set_type(create_info->db_type->default_index_type());

        break;

      }

    }

    if (key_info[i].flags & HA_NOSAME)

      idx->set_is_unique(true);

    else

      idx->set_is_unique(false);

    message::Table::Index::Options *index_options= idx->mutable_options();

    if (key_info[i].flags & HA_USES_BLOCK_SIZE)

      index_options->set_key_block_size(key_info[i].block_size);

    if (key_info[i].flags & HA_PACK_KEY)

      index_options->set_pack_key(true);

    if (key_info[i].flags & HA_BINARY_PACK_KEY)

      index_options->set_binary_pack_key(true);

    if (key_info[i].flags & HA_VAR_LENGTH_PART)

      index_options->set_var_length_key(true);

    if (key_info[i].flags & HA_NULL_PART_KEY)

      index_options->set_null_part_key(true);

    if (key_info[i].flags & HA_KEY_HAS_PART_KEY_SEG)

      index_options->set_has_partial_segments(true);

    if (key_info[i].flags & HA_GENERATED_KEY)

      index_options->set_auto_generated_key(true);

    if (key_info[i].flags & HA_USES_COMMENT)

    {

      uint32_t tmp_len;

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

						  key_info[i].comment.str,

						  key_info[i].comment.str +

						  key_info[i].comment.length,

						  TABLE_COMMENT_MAXLEN);

      if (tmp_len < key_info[i].comment.length)

      {

	my_error(ER_WRONG_STRING_LENGTH, MYF(0),

		 key_info[i].comment.str,"Index COMMENT",

		 (uint32_t) TABLE_COMMENT_MAXLEN);

	return true;

      }

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

    }

    static const uint64_t unknown_index_flag= (HA_NOSAME | HA_PACK_KEY |

                                               HA_USES_BLOCK_SIZE | 

                                               HA_BINARY_PACK_KEY |

                                               HA_VAR_LENGTH_PART |

                                               HA_NULL_PART_KEY | 

                                               HA_KEY_HAS_PART_KEY_SEG |

                                               HA_GENERATED_KEY |

                                               HA_USES_COMMENT);

    if (key_info[i].flags & ~unknown_index_flag)

      abort(); // Invalid (unknown) index flag.

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

    {

      message::Table::Index::IndexPart *idxpart;

      const int fieldnr= key_info[i].key_part[j].fieldnr;

      int mbmaxlen= 1;

      idxpart= idx->add_index_part();

      idxpart->set_fieldnr(fieldnr);

      if (table_proto.field(fieldnr).type() == message::Table::Field::VARCHAR

          || table_proto.field(fieldnr).type() == message::Table::Field::BLOB)

      {

        uint32_t collation_id;

        if (table_proto.field(fieldnr).string_options().has_collation_id())

          collation_id= table_proto.field(fieldnr).string_options().collation_id();

        else

          collation_id= table_proto.options().collation_id();

        const charset_info_st *cs= get_charset(collation_id);

        mbmaxlen= cs->mbmaxlen;

      }

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

    }

  }

  if (not table_proto.IsInitialized())

  {

    my_error(ER_CORRUPT_TABLE_DEFINITION, MYF(0),

             table_proto.name().c_str(),

             table_proto.InitializationErrorString().c_str());

    return true;

  }

  /*

    Here we test to see if we can validate the Table Message before we continue. 

    We do this by serializing the protobuffer.

  */

  {

    string tmp_string;

    try {

      table_proto.SerializeToString(&tmp_string);

    }

    catch (...)

    {

      my_error(ER_CORRUPT_TABLE_DEFINITION, MYF(0),

               table_proto.name().c_str(),

               table_proto.InitializationErrorString().c_str());

      return true;

    }

  }

  return false;

}

/*

  Create a table definition proto 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

  RETURN

    0  ok

    1  error

*/

bool rea_create_table(Session *session,

                      const identifier::Table &identifier,

                      message::Table &table_proto,

                      HA_CREATE_INFO *create_info,

                      List<CreateField> &create_fields,

                      uint32_t keys, KeyInfo *key_info)

{

  assert(table_proto.has_name());

  if (fill_table_proto(identifier,

                       table_proto, create_fields, create_info,

                       keys, key_info))

  {

    return false;

  }

  assert(table_proto.name() == identifier.getTableName());

  if (not plugin::StorageEngine::createTable(*session,

                                             identifier,

                                             table_proto))

  {

    return false;

  }

  return true;

} /* rea_create_table */

} /* namespace drizzled */