/* 
 * Copyright (c) 2007, 2013, Oracle and/or its affiliates. All rights reserved.
 *
 * 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 "stdafx.h"

#ifndef _WIN32
#include <pcre.h>
#include <stdio.h>
#endif

//XXX should remove this mysql specific code from here --alfredo
#include <mysql.h> // needed for enum_field_types
#include <boost/scoped_array.hpp>

#include "db_object_helpers.h"
#include <grtpp_undo_manager.h>
#include <grtpp_util.h>
#include "grt/common.h"
#include "grt/parse_utils.h"
#include "grts/structs.workbench.physical.h"
#include "base/string_utilities.h"

#ifndef _WIN32
#include <string.h>
#endif

using namespace bec;
using namespace grt;


template<class T>
class auto_array_ptr
{
  T *_ptr;

public:
  auto_array_ptr(T* ptr) : _ptr(ptr) {}
  ~auto_array_ptr() { delete[] _ptr; }

  operator T* () { return _ptr; }
};

// Important: this functions works only for model objects, not live objects!
db_mgmt_RdbmsRef get_rdbms_for_db_object(const ::grt::ValueRef &object)
{
  GrtObjectRef parent= GrtObjectRef::cast_from(object);
  while (parent.is_valid() && !parent.is_instance("workbench.physical.Model"))
    parent= parent->owner();

  // do it the hard way to avoid havig to link to objimpl for model
  //return workbench_physical_ModelRef::cast_from(parent)->rdbms();

  if (parent.is_valid())
    return db_mgmt_RdbmsRef::cast_from(parent.get_member("rdbms"));
  return db_mgmt_RdbmsRef();
}

db_SimpleDatatypeRef CatalogHelper::get_datatype(grt::ListRef<db_SimpleDatatype> types, const std::string &name)
{
  for (size_t c= types.count(), i= 0; i < c; i++)
  {
    if (g_strcasecmp(types[i]->name().c_str(), name.c_str())==0)
      return types[i];
    //for (size_t d= types[i]->synonyms().count(), j= 0; j < d; j++)
    //{
    //  if (g_strcasecmp(types[i]->synonyms().get(j).c_str(), name.c_str())==0)
    //    return types[i];
    //}
  }

  return db_SimpleDatatypeRef();
}

//--------------------------------------------------------------------------------------------------

/**
 * Parses the given version string into its components and returns a GRT version class.
 * Unspecified components are set to -1 to allow for fuzzy comparisons.
 */
GrtVersionRef CatalogHelper::parse_version(grt::GRT *grt, const std::string &target_version)
{
  int major = 0, minor = -1, release = -1, build = -1;
  
  sscanf(target_version.c_str(), "%i.%i.%i.%i", &major, &minor, &release, &build);
  
  GrtVersionRef version(grt);
  version->name("Version");
  version->majorNumber(major);
  version->minorNumber(minor);
  version->releaseNumber(release);
  version->buildNumber(build);
  
  return version;
}

//--------------------------------------------------------------------------------------------------

/**
 * Compares the given version numbers to see if a is equal to b.
 * In order to support wildcards, missing values are interpreted as matching, e.g. 5 is equal to 5.1
 * and 5.1 is equal to 5.1.1.
 */
bool CatalogHelper::version_equal(GrtVersionRef a, GrtVersionRef b)
{
  // Major version number is always there.
  if (a->majorNumber() != b->majorNumber())
    return false;
  
  if (a->minorNumber() == -1 || b->minorNumber() == -1)
    return true;
  
  if (a->minorNumber() != b->minorNumber())
    return false;

  if (a->releaseNumber() == -1 || b->releaseNumber() == -1)
    return true;
  
  if (a->releaseNumber() != b->releaseNumber())
    return false;

  if (a->buildNumber() == -1 || b->buildNumber() == -1)
    return true;

  if (a->buildNumber() != b->buildNumber())
    return false;

  return true;
}

/**
 * This test is similar to the one above (except that it tests for greater-than relations).
 * For this however we have to treat unset values differently to handle cases like
 * 5 > 5.1 (false) or 5.1 > 5 (true) correctly.
 */
bool CatalogHelper::version_greater(GrtVersionRef a, GrtVersionRef b)
{
  if (a->majorNumber() > b->majorNumber()) // Major number should always be set.
    return true;

  if (a->majorNumber() == b->majorNumber())
  {
    if (a->minorNumber() == -1) // An unset value can never be larger than anything else.
      return false;
    
    if (b->minorNumber() == -1) // An unset value is always smaller than any other set value.
        return true;
    
    if (a->minorNumber() > b->minorNumber())
      return true;
    
    // Same approach for release + build numbers.
    if (a->minorNumber() == b->minorNumber())
    {
      if (a->releaseNumber() == -1)
        return false;

      if (b->releaseNumber() == -1)
        return true;

      if (a->releaseNumber() > b->releaseNumber())
        return true;

      if (a->releaseNumber() == b->releaseNumber())
      {
        if (a->buildNumber() == -1)
          return false;

        if (b->buildNumber() == -1)
          return true;

        if (a->buildNumber() > b->buildNumber())
          return true;
      }
      return false;
    }
    return false;
  }

  return false;
}


bool CatalogHelper::is_type_valid_for_version(const db_SimpleDatatypeRef &type, const GrtVersionRef &target_version)
{
  std::string validity = type->validity();
  GrtVersionRef valid_version;
  if (!validity.empty())
  {
    bool match = false;
    switch (validity[0])
    {
      case '<':
        if (validity[1] == '=')
        {
          valid_version = CatalogHelper::parse_version(type.get_grt(), validity.substr(2));
          if (version_equal(target_version, valid_version) || version_greater(valid_version, target_version))
            match = true;
        }
        else
        {
          valid_version = CatalogHelper::parse_version(type.get_grt(), validity.substr(1));
          if (version_greater(valid_version, target_version))
            match = true;
        }
        break;
      case '>':
        if (validity[1] == '=')
        {
          valid_version = CatalogHelper::parse_version(type.get_grt(), validity.substr(2));
          if (version_equal(target_version, valid_version) || version_greater(target_version, valid_version))
            match = true;
        }
        else
        {
          valid_version = CatalogHelper::parse_version(type.get_grt(), validity.substr(1));
          if (version_greater(target_version, valid_version))
            match = true;
        }
        break;
      case '=':
        valid_version = CatalogHelper::parse_version(type.get_grt(), validity.substr(1));
        if (version_equal(target_version, valid_version))
          match = true;
        break;
    }
    return match;
  }
  return true;
}



std::string CatalogHelper::dbobject_to_dragdata(const db_DatabaseObjectRef &object)
{
  return object.class_name()+":"+object.id();
}


db_DatabaseObjectRef CatalogHelper::dragdata_to_dbobject(const db_CatalogRef &catalog, const std::string &data)
{
  if (data.find(':') != std::string::npos)
  {
    std::string oid= data.substr(data.find(':')+1);
    
    return db_DatabaseObjectRef::cast_from(find_child_object(catalog, oid));
  }
  return db_DatabaseObjectRef();
}


std::string CatalogHelper::dbobject_list_to_dragdata(const std::list<db_DatabaseObjectRef> &objects)
{
  std::string ret;
  for (std::list<db_DatabaseObjectRef>::const_iterator iter= objects.begin(); iter != objects.end(); ++iter)
  {
    if (!ret.empty())
      ret.append("\n");
    ret.append(dbobject_to_dragdata(*iter));
  }
  return ret;
}


std::list<db_DatabaseObjectRef> CatalogHelper::dragdata_to_dbobject_list(const db_CatalogRef &catalog, const std::string &data)
{
  std::list<db_DatabaseObjectRef> dbobjects;
  std::vector<std::string> items= base::split(data, "\n");
  for (std::vector<std::string>::const_iterator item= items.begin(); item != items.end(); ++item)
  {
    db_DatabaseObjectRef object= dragdata_to_dbobject(catalog, *item);
    if (object.is_valid())
      dbobjects.push_back(object);
  }

  return dbobjects;
}

//------------------------------------------------------------------------------------
std::set<std::string> SchemaHelper::get_foreign_key_names(const db_SchemaRef &schema)
{
  std::set<std::string> used_names;
  
  GRTLIST_FOREACH(db_Table, schema->tables(), table)
  {
    GRTLIST_FOREACH(db_ForeignKey, (*table)->foreignKeys(), fk)
    {
      used_names.insert((*fk)->name());
    }
  }

  return used_names;
}

std::string SchemaHelper::get_unique_foreign_key_name(std::set<std::string> &used_names, 
                                                      const std::string &prefix_,
                                                      int maxlength)
{
  std::string prefix;
  std::string the_name= prefix_;
  int index= 0;
  
  // truncate if too long, on byte count, because that's what matters when creating
  // stuff on the server
  if ((int) the_name.size() > maxlength-2)
  {
    const char *start = the_name.c_str();
    const char *end = the_name.c_str()+maxlength-2;
    
    the_name= the_name.substr(0, g_utf8_find_prev_char(start, end)-start);
  }
  
  prefix= the_name;
  
  while (used_names.find(the_name) != used_names.end())
    the_name= base::strfmt("%s%i", prefix.c_str(), index++);
  
  if (the_name != prefix)
    used_names.insert(the_name);
  
  return the_name;  
}

std::string SchemaHelper::get_unique_foreign_key_name(const db_SchemaRef &schema, 
                                                      const std::string &prefix_,
                                                      int maxlength)
{
  std::set<std::string> used_names;
  std::string prefix;
  std::string the_name= prefix_;
  int index= 0;
  
  // truncate if too long, on byte count, because that's what matters when creating
  // stuff on the server
  if ((int) the_name.size() > maxlength-2)
  {
    const char *start = the_name.c_str();
    const char *end = the_name.c_str()+maxlength-2;
      
    the_name= the_name.substr(0, g_utf8_find_prev_char(start, end)-start);
  }
  
  prefix= the_name;

  GRTLIST_FOREACH(db_Table, schema->tables(), table)
  {
    GRTLIST_FOREACH(db_ForeignKey, (*table)->foreignKeys(), fk)
    {
      used_names.insert((*fk)->name());
      if (the_name == prefix && index == 0)
        index++;
    }
  }
  
  // prefix is duplicated
  if (index > 0)
  {
    do 
    {
      the_name= base::strfmt("%s%i", prefix.c_str(), index++);
    } while (used_names.find(the_name) != used_names.end());
  }
  
  return the_name;
}

//------------------------------------------------------------------------------------

db_TableRef TableHelper::create_associative_table(const db_SchemaRef &schema, 
                                               const db_TableRef &table1, const db_TableRef &table2, 
                                               bool mandatory1, bool mandatory2,
                                                  const db_mgmt_RdbmsRef &rdbms,
                                               const grt::DictRef &global_options,
                                               const grt::DictRef &options)
{
  db_TableRef atable;
  std::string name;

  AutoUndo undo(schema->get_grt());

  name= options.get_string("AuxTableTemplate", global_options.get_string("AuxTableTemplate", "%stable%_%dtable%"));

  name= replace_variable(name, "%stable%", table1->name().c_str());
  name= replace_variable(name, "%dtable%", table2->name().c_str());

  atable= schema.get_grt()->create_object<db_Table>(table1.get_metaclass()->name());
  atable->owner(schema);
  atable->name(get_name_suggestion_for_list_object(schema->tables(), name, false));
  atable->oldName(atable->name());
  
  if (atable.has_member("tableEngine"))
    atable->set_member("tableEngine", table1.get_member("tableEngine"));
  if (atable.has_member("defaultCharacterSetName"))
    atable->set_member("defaultCharacterSetName", table1.get_member("defaultCharacterSetName"));
  if (atable.has_member("defaultCollationName"))
    atable->set_member("defaultCollationName", table1.get_member("defaultCollationName"));

  db_ForeignKeyRef first_fk = create_foreign_key_to_table(atable, table1, true, mandatory1, true, true, rdbms, global_options, options);
  schema->tables().insert(atable);//Insert put newly added FK into schema to avoid name collision
  create_foreign_key_to_table(atable, table2, true, mandatory2, true, true, rdbms, global_options, options);
  
  // when the 1st FK is created, the PK and FK will have the same columns, so the index will be reused
  // when the 2nd FK is created, the PK will have both columns so it can't be used by the 1st FK, so we have
  // to create a new one for it
  db_IndexRef index = create_index_for_fk(schema->get_grt(), first_fk, rdbms->maximumIdentifierLength());
  first_fk->index(index);
  atable->indices().insert(index);
  
  undo.end("Create Associative Table");

  return atable;
}

//
//static std::string format_ident_with_table(const std::string &fmt, const db_TableRef &table)
//{
//  return replace_variable(fmt, "%table%", table->name().c_str());
//}

static std::string format_ident_with_stable_dtable(const std::string &fmt, const db_TableRef &stable, const db_TableRef &dtable)
{
  return replace_variable(replace_variable(fmt, "%stable%", stable->name().c_str()), "%dtable%", dtable->name().c_str());
}

static std::string format_ident_with_column(const std::string &fmt, const db_ColumnRef &column)
{
  return replace_variable(replace_variable(fmt, "%table%", db_TableRef::cast_from(column->owner())->name().c_str()),
                        "%column%", column->name().c_str());
}


db_IndexRef TableHelper::create_index_for_fk(grt::GRT *grt, const db_ForeignKeyRef &fk, const size_t max_len)
{
  std::string index_name(fk->name().c_str());
  if (index_name.length() > (max_len-5))
      index_name.resize(max_len-5);
  index_name.append("_idx");
  index_name = grt::get_name_suggestion_for_list_object(fk->owner()->indices(),index_name,false);

  // add the corresponding index
  db_IndexRef index= grt->create_object<db_Index>(db_TableRef::cast_from(fk->owner()).get_metaclass()->get_member_type("indices").content.object_class.c_str());
  
  index->owner(fk->owner());
  index->name(index_name);
  index->oldName(fk->oldName());
  index->indexType("INDEX");

  for(size_t i= 0, sz= fk->columns().count(); i < sz; i++)
  {
    db_ColumnRef col(fk->columns().get(i));

    db_IndexColumnRef index_col(grt->create_object<db_IndexColumn>(index.get_metaclass()->get_member_type("columns").content.object_class));
    index_col->owner(index);
//    "name", col->name().valueptr(),
    index_col->descend(grt::IntegerRef(0));
    //"columnLength", grt::IntegerRef(col->length().valueptr()).valueptr(),
    index_col->columnLength(grt::IntegerRef(0));
    index_col->referencedColumn(col);

    index->columns().insert(index_col);
  }

  return index;
}


db_IndexRef TableHelper::find_index_usable_by_fk(const db_ForeignKeyRef &fk, const db_IndexRef &other_than)
{
  size_t column_count = fk->columns().count();
  db_TableRef table(db_TableRef::cast_from(fk->owner()));
  
  for (size_t c= table->indices().count(), i= 0; i < c; i++)
  {
    db_IndexRef index(table->indices()[i]);

    if (index == other_than)
      continue;
    
    if (index->columns().count() == column_count)
    {
      bool all_columns_match= true;
      for (size_t j= 0; j < column_count; j++)
      {
        bool ok= false;
        for (size_t k= 0; k < column_count; k++)
        {
          if (index->columns()[k]->referencedColumn() == fk->columns()[j])
          {
            ok= true;
            break;
          }
        }
        if (!ok)
        {
          all_columns_match= false;
          break;
        }
      }
      if (all_columns_match)
        return index;
    }
  }
  return db_IndexRef();
}


void TableHelper::update_foreign_keys_from_column_notnull(const db_TableRef &table, const db_ColumnRef &column)
{
  AutoUndo undo(table->get_grt());

  // go through all foreign keys and update the ones that have this column
  grt::ListRef<db_ForeignKey> fklist(table->foreignKeys());

  for (size_t c= fklist.count(), i= 0; i < c; i++)
  {
    db_ForeignKeyRef fk(fklist[i]);
    size_t notnull= 0;
    bool flag= false;

    for (size_t d= fk->columns().count(), j= 0; j < d; j++)
    {
      db_ColumnRef col(fk->columns()[j]);

      if (*col->isNotNull())
        notnull++;

      if (col == column)
        flag= true;
    }
    if (flag)
    {
      if (notnull == fk->columns().count())
      {
        // if the FK is optional, then it means the ref table is optional
        fk->referencedMandatory(1);
      }
      else if (notnull == 0)
      {
        fk->referencedMandatory(0);
      }
    }
  }

  undo.end("Update FK Mandatory Flag");
}


std::string TableHelper::generate_foreign_key_name()
{
  return std::string("fk_") + grt::get_guid();
}


db_ForeignKeyRef TableHelper::create_empty_foreign_key(grt::GRT *grt, const db_TableRef &table, const std::string &name)
{
  db_ForeignKeyRef fk;

  // create a new FK
  fk= grt->create_object<db_ForeignKey>(table.get_metaclass()->get_member_type("foreignKeys").content.object_class);
  fk->owner(table);
  fk->name(name.empty()?generate_foreign_key_name():name);

  grt::AutoUndo undo(grt);

  table->foreignKeys().insert(fk);

  /* don't always create an index for the FK, it should only be created if there are no other indexes that match it
   the check for indexes should be done every time the FK columns are edited
  db_IndexRef index(create_index_for_fk(grt, fk));
  fk->index(index);
  table->indices().insert(index);
  */
  undo.end(_("Create Foreign Key"));

  return fk;
}


void TableHelper::update_foreign_key_index(const db_ForeignKeyRef &fk)
{
  //! todo: add undo group

  db_TableRef table(fk->owner());
  db_IndexRef index(fk->index());

  if (index.is_valid())
  {
    if (find_index_usable_by_fk(fk, index).is_valid())
    {
      // the index from this FK is redundant, so remove it
      fk->index(db_IndexRef());
      table->indices().remove_value(index);
      return;
    }

    // make sure that the columns in the index match the ones in the FK

    // remove columns that are gone from the FK
    for (int i = index->columns().count()-1; i >= 0; --i)
    {
      if (fk->columns().get_index(index->columns()[i]) == grt::BaseListRef::npos)
        index->columns().remove(i);
    }

    // recreate index columns
    while (index->columns().count() > 0)
      index->columns().remove(0);

    ListRef<db_Column> fk_columns(fk->columns());
    for (size_t n= 0, count= fk_columns.count(); n < count; ++n)
    {
      db_ColumnRef column(fk_columns.get(n));
      db_IndexColumnRef index_column(fk->get_grt()->create_object<db_IndexColumn>(index.get_metaclass()->get_member_type("columns").content.object_class));
      index_column->owner(index);
      //"name", column->name().valueptr(),
      index_column->referencedColumn(column);
      index->columns().insert(index_column);
    }

    if (index->columns().count() == 0)
    {
      // the index is empty, remove it
      fk->index(db_IndexRef());
      table->indices().remove_value(index);
      return;
    }
  }
  else
  {
    // check if there's a suitable index
    if (find_index_usable_by_fk(fk).is_valid())
      return; // suitable index exists, no action needed

    // create a new index for the FK if it's valid at all
    if (fk->columns().count() > 0)
    {
      index = create_index_for_fk(table->get_grt(), fk);
      fk->index(index);
      table->indices().insert(index);
    }
  }
}


bool TableHelper::create_missing_indexes_for_foreign_keys(const db_TableRef &table)
{
  bool created= false;
  GRTLIST_FOREACH(db_ForeignKey, table->foreignKeys(), fk)
  {
    db_IndexRef index;
    // check if an index already exists for the FK columns and if not, create one
    if (!(*fk)->index().is_valid() && !find_index_usable_by_fk(*fk).is_valid())
    {
      index = create_index_for_fk(table->get_grt(), *fk);
      (*fk)->index(index);
      table->indices().insert(index);
      created= true;
    }
  }
  return created;
}


/** 
 ****************************************************************************
 * @brief Creates a foreign key in table, refering to ref_table
 *
 * 
 * Options:
 * FKNameTemplate, FKColumnNameTemplate, db_ForeignKey:deleteRule, updateRule
 * 
 * @param table the table to have the FK added to
 * @param ref_table the table that the FK refers to
 * @param mandatory if the relationship is mandatory in table
 * @param ref_mandatory if the relationship is mandatory in referenced table
 * @param many cardinality of the table in the relationship
 * @param identifying whether the rel is identifying (FK is also made PK)
 * @param options the options dictionary (see above)
 *
 * @return the created foreign key
 ****************************************************************************
 */
db_ForeignKeyRef TableHelper::create_foreign_key_to_table(const db_TableRef &table, 
                                                          const db_TableRef &ref_table,
                                                          bool mandatory, bool ref_mandatory,
                                                          bool many, bool identifying,
                                                          const db_mgmt_RdbmsRef &rdbms,
                                                          const grt::DictRef &global_options,
                                                          const grt::DictRef &options)
{
  db_ForeignKeyRef new_fk;
  db_IndexRef pk= ref_table->primaryKey();
  grt::GRT *grt= table.get_grt();
  std::string name_format;
  std::string column_name_format;
  std::string scolumn_name;
  std::string dcolumn_name;
  int max_identifier_length= rdbms->maximumIdentifierLength();

  // check if there is a PK
  if (!pk.is_valid() || pk->columns().count() == 0)
    return new_fk;

  grt::AutoUndo undo(grt, !table->is_global());

  name_format= options.get_string("FKNameTemplate", global_options.get_string("FKNameTemplate","FK%table%"));
  column_name_format= options.get_string("FKColumnNameTemplate", global_options.get_string("FKColumnNameTemplate", "FK%table%%column%"));

  name_format= format_ident_with_stable_dtable(name_format, table, ref_table);
  name_format= format_ident_with_column(name_format, pk->columns().get(0)->referencedColumn());

  column_name_format= format_ident_with_stable_dtable(column_name_format, table, ref_table);

  // create a new FK
  new_fk= grt->create_object<db_ForeignKey>(table.get_metaclass()->get_member_type("foreignKeys").content.object_class);
  new_fk->oldName(new_fk->name());

  new_fk->deleteRule(options.get_string("db.ForeignKey:deleteRule", global_options.get_string("db.ForeignKey:deleteRule", "NO ACTION")));
  new_fk->updateRule(options.get_string("db.ForeignKey:updateRule", global_options.get_string("db.ForeignKey:updateRule", "NO ACTION")));

  new_fk->mandatory(mandatory?1:0);
  new_fk->referencedMandatory(ref_mandatory?1:0);
  new_fk->many(many?1:0);

  new_fk->referencedTable(ref_table);

  for (size_t c= pk->columns().count(), i= 0; i < c; i++)
  {
    db_IndexColumnRef pk_column= pk->columns().get(i);
    db_ColumnRef column= pk_column->referencedColumn();
    db_ColumnRef new_fk_column;

    // create the column that will be the FK in the other table
    new_fk_column= grt->create_object<db_Column>(column.class_name());
    new_fk_column->owner(table);
    new_fk_column->name(get_name_suggestion_for_list_object(table->columns(), format_ident_with_column(column_name_format, column), false));
    new_fk_column->oldName(new_fk_column->name());
    new_fk_column->isNotNull(ref_mandatory?1:0);

    ColumnHelper::copy_column(column, new_fk_column);

    table->columns().insert(new_fk_column);

    // add the new column to the FK
    new_fk->columns().insert(new_fk_column);
    new_fk->referencedColumns().insert(column);

    if (identifying)
      table->addPrimaryKeyColumn(new_fk_column);
    
    if (scolumn_name.empty())
    {
      scolumn_name= column->name();
      dcolumn_name= new_fk_column->name();
    }
  }
  
  // substitute scolumn/dcolumn now that we know the created column name
  name_format= replace_variable(replace_variable(name_format, "%scolumn%", scolumn_name), "%dcolumn%", dcolumn_name);
  
  new_fk->name(SchemaHelper::get_unique_foreign_key_name(db_SchemaRef::cast_from(table->owner()), 
                                                         name_format, 
                                                         max_identifier_length));
  new_fk->oldName(new_fk->name());
  
  new_fk->owner(table); // set the owner last

  db_IndexRef index;
  // check if an index already exists for the FK columns and if not, create one
  if (!find_index_usable_by_fk(new_fk).is_valid())
  {
    index = create_index_for_fk(grt, new_fk, rdbms->maximumIdentifierLength());
    new_fk->index(index);
    // index is inserted later down
  }

  // add the FK to the source table
  table->foreignKeys().insert(new_fk);
  undo.set_description_for_last_action("Add Foreign Key to Table");

  if (index.is_valid())
  {
    table->indices().insert(index);
    undo.set_description_for_last_action("Add Index for FK to Table");
  }
  undo.end(_("Add Foreign Key"));
  
  return new_fk;
}



db_ForeignKeyRef TableHelper::create_foreign_key_to_table(const db_TableRef &table, const std::vector<db_ColumnRef> &columns,
                                                     const db_TableRef &ref_table, const std::vector<db_ColumnRef> &refcolumns,
                                                     bool mandatory, 
                                                     bool many,
                                                          const db_mgmt_RdbmsRef &rdbms,
                                                     const grt::DictRef &global_options,
                                                     const grt::DictRef &options)
{
  db_ForeignKeyRef new_fk;
  grt::GRT *grt= table.get_grt();
  std::string name_format;
  std::string scolumn_name;
  std::string dcolumn_name;
  int max_identifier_length= rdbms->maximumIdentifierLength();
  bool ref_mandatory;

  grt::AutoUndo undo(grt, table->is_global());
  
  name_format= options.get_string("FKNameTemplate", global_options.get_string("FKNameTemplate","FK%table%"));

  name_format= format_ident_with_stable_dtable(name_format, table, ref_table);
  name_format= format_ident_with_column(name_format, refcolumns[0]);

  // create a new FK
  new_fk= grt->create_object<db_ForeignKey>(table.get_metaclass()->get_member_type("foreignKeys").content.object_class);

  new_fk->deleteRule(options.get_string("db.ForeignKey:deleteRule", global_options.get_string("db.ForeignKey:deleteRule", "NO ACTION")));
  new_fk->updateRule(options.get_string("db.ForeignKey:updateRule", global_options.get_string("db.ForeignKey:updateRule", "NO ACTION")));

  new_fk->referencedTable(ref_table);
  
  new_fk->mandatory(mandatory?1:0);
  //new_fk->referencedMandatory(ref_mandatory?1:0); done later
  new_fk->many(many?1:0);

  ref_mandatory= false;
  for (size_t c= refcolumns.size(), i= 0; i < c; i++)
  {
    db_ColumnRef column(refcolumns[i]);
    db_ColumnRef fk_column(columns[i]);

    if (column->isNotNull())
      ref_mandatory= true;
    
    // add the new column to the FK
    new_fk->columns().insert(fk_column);
    new_fk->referencedColumns().insert(column);
    if (scolumn_name.empty())
    {
      scolumn_name= fk_column->name();
      dcolumn_name= column->name();
    }
  }
  
  // substitute scolumn/dcolumn now that we know the created column name
  name_format= replace_variable(replace_variable(name_format, "%scolumn%", scolumn_name), "%dcolumn%", dcolumn_name);  
  
  new_fk->name(SchemaHelper::get_unique_foreign_key_name(db_SchemaRef::cast_from(table->owner()), 
                                                         name_format,
                                                         max_identifier_length));
  new_fk->oldName(new_fk->name());
  
  new_fk->referencedMandatory(ref_mandatory?1:0);

  new_fk->owner(table);

  // check if an index already exists for the FK columns and if not, create one
  if (!find_index_usable_by_fk(new_fk).is_valid())
  {
    db_IndexRef index(create_index_for_fk(grt, new_fk,rdbms->maximumIdentifierLength()));
    new_fk->index(index);
    
    // add the FK to the source table
    table->foreignKeys().insert(new_fk);
    table->indices().insert(index);    
  }
  else // add the FK to the source table
    table->foreignKeys().insert(new_fk);

  undo.end(_("Add Foreign Key"));
  
  return new_fk;
}


bool TableHelper::rename_foreign_key(const db_TableRef &table, db_ForeignKeyRef &fk, const std::string &new_name)
{
  std::string old_name;

  // check if the name is already taken
  if (find_named_object_in_list(table->foreignKeys(), new_name).is_valid())
    return false;

  old_name= fk->name();

  grt::AutoUndo undo(table->get_grt());

  fk->name(new_name);

  // only rename the index if the old names match
  if (fk->index().is_valid())
  {
    if (old_name == *fk->index()->name())
        fk->index()->name(new_name);
//  else
//    g_warning("ForeignKey %s has no attached index", fk->name().c_str());
  }
  undo.end(_("Rename Foreign Key"));

  return true;
}


bool TableHelper::is_identifying_foreign_key(const db_TableRef &table, const db_ForeignKeyRef &fk)
{
  // check if the fk is part of the PK
  if (table->primaryKey().is_valid())
  {
    for (size_t c= fk->columns().count(), i= 0; i < c; i++)
    {
      if (!table->isPrimaryKeyColumn(fk->columns().get(i)))
        return false;
    }
    return true;
  }
  return false;
}


db_mysql_StorageEngineRef TableHelper::get_engine_by_name(grt::GRT *grt, const std::string &name)
{
  grt::ListRef<db_mysql_StorageEngine> engines;
  Module *module= grt->get_module("DbMySQL");
  
  if (!module)
    throw std::logic_error("module DbMySQL not found");
  
  grt::BaseListRef args(grt);
  engines= grt::ListRef<db_mysql_StorageEngine>::cast_from(module->call_function("getKnownEngines", args));

  if (engines.is_valid())
  {
    for (grt::ListRef<db_mysql_StorageEngine>::const_iterator iter= engines.begin();
         iter != engines.end(); ++iter)
    {
      if ((*iter)->name() == name)
        return *iter;
    }
  }
  return db_mysql_StorageEngineRef();
}

/**
 * Returns a pointer to the position of the first line break or the first character after the
 * maximum count, whichever comes first. If no line break was found and the string is shorter
 * than the maximum length the result points to the terminating 0.
 */
gchar* find_comment_split(const std::string &comment,const size_t &db_comment_len)
{
  size_t res;
  // XXX: check for Unicode line breaks! especially asian languages don't use the ANSI new line.
  gchar* pointer_to_linebreak = g_utf8_strrchr(comment.c_str(), 0, '\n');

  // We need the number of characters which the string part includes, so convert to a char count.
  if (pointer_to_linebreak != NULL)
    res = g_utf8_pointer_to_offset(comment.c_str(), pointer_to_linebreak);
  else
    res = g_utf8_strlen(comment.c_str(), -1);

  if (res > db_comment_len)
    res = db_comment_len;

  // Finally convert the offset back into a pointer.
  return g_utf8_offset_to_pointer(comment.c_str(), res);
};

std::string TableHelper::get_sync_comment(const std::string &comment, const int max_len)
{
  gchar* split_position = find_comment_split(comment, max_len);
  return std::string(comment.c_str(), split_position - comment.c_str());
};

std::string TableHelper::get_document_comment(const std::string &comment, const int max_len)
{
  return std::string(find_comment_split(comment, max_len));
};

void ColumnHelper::copy_column(const db_ColumnRef &from, db_ColumnRef &to)
{
  to->userType(from->userType());
  to->precision(from->precision());
  to->scale(from->scale());
  to->length(from->length());
  to->characterSetName(from->characterSetName());
  to->collationName(from->collationName());
  while (to->flags().count() > 0)
    to->flags().remove(0);
  for (size_t c= from->flags().count(), i= 0; i < c; i++)
    to->flags().insert(from->flags().get(i));
  to->simpleType(from->simpleType());
  to->structuredType(from->structuredType());
  to->datatypeExplicitParams(from->datatypeExplicitParams());
}


ColumnTypeCompareResult ColumnHelper::compare_column_types(const db_ColumnRef &from, const db_ColumnRef &to)
{
  // not to be used for foreign key column matching as the rules are different and DB dependant
  std::string sfrom= from->formattedType();
  std::string sto= to->formattedType();
  
  if (sfrom != sto)
    return COLUMNS_TYPES_DIFFER;

  if (to->characterSetName() != from->characterSetName() )
    return COLUMNS_CHARSETS_DIFFER;
    
  if ( to->collationName() != from->collationName() )
    return COLUMNS_COLLATIONS_DIFFER;

  if (to->flags().count() != from->flags().count())
    return COLUMNS_FLAGS_DIFFER;

  for (size_t c= from->flags().count(), i= 0; i < c; i++)
    if (to->flags().get_index(from->flags().get(i)) == BaseListRef::npos)
      return COLUMNS_FLAGS_DIFFER;

  //XXX compare db specific attribs

  return COLUMNS_TYPES_EQUAL;
}


void ColumnHelper::set_default_value(db_ColumnRef column, const std::string &value)
{
  column->defaultValueIsNull(g_strcasecmp(value.c_str(), "NULL") ? 0 : 1);
  column->defaultValue(value.c_str());

  // If a default value of NULL was set then the column can no longer be a not-null column.
  if (column->defaultValueIsNull())
    column->isNotNull(false);

  // Handling auto-increment columns is MySQL specific and done elsewhere.
}

//------------------------------------------------------------------------------------
//
//
////XXX the following code must be moved to db.mysql module
//
//static enum_field_types get_mysql_type_from_string(const char *type_name)
//{
//  if(g_strcasecmp(type_name, "INT") == 0)
//    return MYSQL_TYPE_LONG;
//  if(g_strcasecmp(type_name, "TINYINT") == 0)
//    return MYSQL_TYPE_TINY;
//  if(g_strcasecmp(type_name, "SMALLINT") == 0)
//    return MYSQL_TYPE_SHORT;
//  if(g_strcasecmp(type_name, "BIGINT") == 0)
//    return MYSQL_TYPE_LONGLONG;
//  if(g_strcasecmp(type_name, "FLOAT") == 0)
//    return MYSQL_TYPE_FLOAT;
//  if(g_strcasecmp(type_name, "DOUBLE") == 0)
//    return MYSQL_TYPE_DOUBLE;
//  if(g_strcasecmp(type_name, "TIME") == 0)
//    return MYSQL_TYPE_TIME;
//  if(g_strcasecmp(type_name, "DATE") == 0)
//    return MYSQL_TYPE_DATE;
//  if(g_strcasecmp(type_name, "DATETIME") == 0)
//    return MYSQL_TYPE_DATETIME;
//  if(g_strcasecmp(type_name, "TIMESTAMP") == 0)
//    return MYSQL_TYPE_TIMESTAMP;
//  if(g_strcasecmp(type_name, "TEXT") == 0)
//    return MYSQL_TYPE_STRING;
//  if(g_strcasecmp(type_name, "CHAR") == 0)
//    return MYSQL_TYPE_STRING;
//  if(g_strcasecmp(type_name, "VARCHAR") == 0)
//    return MYSQL_TYPE_STRING;
//  if(g_strcasecmp(type_name, "BLOB") == 0)
//    return MYSQL_TYPE_BLOB;
//  if(g_strcasecmp(type_name, "BINARY") == 0)
//    return MYSQL_TYPE_BLOB;
//  if(g_strcasecmp(type_name, "VARBINARY") == 0)
//    return MYSQL_TYPE_BLOB;
//
//  return MYSQL_TYPE_NULL;
//}

void CatalogHelper::apply_defaults(db_mysql_ColumnRef column)
{

  // for numeric types only
  std::map<std::string, int> def_precision_map;
  def_precision_map["INT"]= 11;
  def_precision_map["TINYINT"]= 4;
  def_precision_map["SMALLINT"]= 6;
  def_precision_map["MEDIUMINT"]= 9;
  def_precision_map["BIGINT"]= 20;
  def_precision_map["FLOAT"]= 11;
  def_precision_map["DOUBLE"]= 11;
  def_precision_map["BIT"]= 1;
  def_precision_map["CHAR"]= 1;

  bool default_default_value= !column->isNotNull() && (strlen(column->defaultValue().c_str()) == 0);

  if (!column->simpleType().is_valid())
    return;

//  if (strcmp(column->simpleType()->name().c_str(), "DECIMAL") == 0)
//    column->scale(0);

  //enum_field_types ft= get_mysql_type_from_string(column->simpleType()->name().c_str());

  std::map<std::string, int>::const_iterator prec_map_it= 
    def_precision_map.find(column->simpleType()->name().c_str());
  if(prec_map_it != def_precision_map.end())
  {
    if((strcmp(column->simpleType()->name().c_str(), "CHAR") != 0))//length should be used for chars
      column->length(grt::IntegerRef(-1));
    if(column->precision() == bec::EMPTY_COLUMN_PRECISION)
    {
      if(column->flags().get_index("UNSIGNED") != grt::BaseListRef::npos)
        column->precision(grt::IntegerRef(prec_map_it->second)-1);
      else 
        column->precision(grt::IntegerRef(prec_map_it->second));
    }
    if (default_default_value)
      bec::ColumnHelper::set_default_value(column, "NULL");
  }
  else if((strcmp(column->simpleType()->name().c_str(), "VARCHAR") == 0)
    || (strcmp(column->simpleType()->name().c_str(), "CHAR") == 0)
    || (strcmp(column->simpleType()->name().c_str(), "DECIMAL") == 0)
    || (strcmp(column->simpleType()->name().c_str(), "BOOLEAN") == 0)
    || (strcmp(column->simpleType()->name().c_str(), "BINARY") == 0)
    || (strcmp(column->simpleType()->name().c_str(), "TINYBLOB") == 0)
    || (strcmp(column->simpleType()->name().c_str(), "MEDIUMBLOB") == 0)
    || (strcmp(column->simpleType()->name().c_str(), "BLOB") == 0)
    || (strcmp(column->simpleType()->name().c_str(), "LONGBLOB") == 0)
    || (strcmp(column->simpleType()->name().c_str(), "TINYTEXT") == 0)
    || (strcmp(column->simpleType()->name().c_str(), "TEXT") == 0)
    || (strcmp(column->simpleType()->name().c_str(), "MEDIUMTEXT") == 0)
    || (strcmp(column->simpleType()->name().c_str(), "LONGTEXT") == 0)
    || (strcmp(column->simpleType()->name().c_str(), "ENUM") == 0)
    || (strcmp(column->simpleType()->name().c_str(), "SET") == 0))
  {
    if (default_default_value)
      bec::ColumnHelper::set_default_value(column, "NULL");
  }
  else if((strcmp(column->simpleType()->name().c_str(), "DATETIME") == 0)
    || (strcmp(column->simpleType()->name().c_str(), "DATE") == 0)
    || (strcmp(column->simpleType()->name().c_str(), "TIME") == 0)
    || (strcmp(column->simpleType()->name().c_str(), "TIMESTAMP") == 0)
    || (strcmp(column->simpleType()->name().c_str(), "YEAR") == 0)
    || (strcmp(column->simpleType()->name().c_str(), "BIT") == 0))
  {
    column->length(grt::IntegerRef(-1));
    if (default_default_value)
      bec::ColumnHelper::set_default_value(column, "NULL");
  }

};

void CatalogHelper::apply_defaults(db_mysql_CatalogRef cat, std::string default_engine)
{
  cat->defaultCharacterSetName("utf8");
  cat->defaultCollationName("utf8_general_ci");

  for(size_t i= 0, schemata_count= cat->schemata().count(); i < schemata_count; i++)
  {
    db_mysql_SchemaRef schema= cat->schemata().get(i);

    if(strlen(schema->defaultCharacterSetName().c_str()) == 0)
      schema->defaultCharacterSetName(cat->defaultCharacterSetName());
    if(strlen(schema->defaultCollationName().c_str()) == 0)
      schema->defaultCollationName(get_cs_def_collation(schema->defaultCharacterSetName().c_str()));

    for(size_t j= 0, tables_count= schema->tables().count(); j < tables_count; j++)
    {
      db_mysql_TableRef table= schema->tables().get(j);

      if(strlen(table->defaultCharacterSetName().c_str()) == 0)
        table->defaultCharacterSetName(schema->defaultCharacterSetName());

      if(table->defaultCharacterSetName() == schema->defaultCharacterSetName())
      {
      if(strlen(table->defaultCollationName().c_str()) == 0)
        table->defaultCollationName(schema->defaultCollationName());
      }
      
      if(strlen(table->tableEngine().c_str()) == 0)
        table->tableEngine(default_engine.empty()?"InnoDB":default_engine);

          for (size_t c= table->foreignKeys().count(), i= 0; i < c; i++)
          {
            db_ForeignKeyRef fk(table->foreignKeys()[i]);
            if (fk->referencedTable().is_valid())
            {
              for (size_t d= fk->referencedColumns().count(), j= 0; j < d; j++)
              {
                db_mysql_ColumnRef col = db_mysql_ColumnRef::cast_from(fk->referencedColumns().get(j));
                apply_defaults(col);
              }
            }
          }

      for(size_t k= 0, col_count= table->columns().count(); k < col_count; k++)
      {
        apply_defaults(table->columns().get(k));
      }
    }
  }
}

static int pcre_compile_exec(const char *pattern, const char *str, int *patres, int patresnum)
{
  const char *errptr;
  int erroffs;
  int c;
  pcre *patre= pcre_compile(pattern, 0, &errptr, &erroffs, NULL);
  if (!patre)
    throw std::logic_error("error compiling regex "+std::string(errptr));

  c= pcre_exec(patre, NULL, str, strlen(str), 0, 0, patres, patresnum);
  pcre_free(patre);

  return c;
}

static db_SimpleDatatypeRef findType(const grt::ListRef<db_SimpleDatatype> &types, const GrtVersionRef &target_version, const std::string &name)
{
  for (size_t c= types.count(), i= 0; i < c; i++)
  {
    if (g_strcasecmp(types[i]->name().c_str(), name.c_str())==0)
    {
      if (!target_version.is_valid() || CatalogHelper::is_type_valid_for_version(types[i], target_version))
        return types[i];
    }
  }
  return db_SimpleDatatypeRef();
}

static bool parseTypeDefinition(const std::string &type,
                                const GrtVersionRef &targetVersion,
                                const grt::ListRef<db_SimpleDatatype> &typeList,
                                db_SimpleDatatypeRef &simpleType,
                                int &precision,
                                int &scale,
                                int &length,
                                std::string &explicitParams)
{
  auto_array_ptr<char> buffer(new char[type.size()+1]);
  int patres[21];
  int c;
    
  /* split the datatype name from the parameters
   * (\w+)\s*(\(.*\))?
   *
   * 1.. TYPE
   * 2.. params (optional)
   */
  
  c= pcre_compile_exec("^(\\w+)\\s*(\\(.*\\))?\\s*$", type.c_str(), patres, sizeof(patres)/sizeof(int));
  
  if (c > 0 &&
      pcre_copy_substring(type.c_str(), patres, c, 1, buffer, static_cast<int>(type.size()+1)) > 0)
  {
    // find the simple datatype with this type_name
    simpleType= findType(typeList, targetVersion, std::string(buffer));
    
    if (simpleType.is_valid())
    {
      std::string params;
      std::string param1;
      std::string param2;
      
      // split the parameter into its components
      
      if (pcre_copy_substring(type.c_str(), patres, c, 2, buffer, static_cast<int>(type.size()+1)) > 0)
        params= buffer;
      
      switch (*simpleType->parameterFormatType())
      {
        case 0: // no params
          if (!params.empty())
            return false;
          break;
          
        case 1: // (n)
          c= pcre_compile_exec("^\\((\\d+)\\)$", params.c_str(), patres, sizeof(patres)/sizeof(int));
          if (c < 0)
            return false;
          if (pcre_copy_substring(params.c_str(), patres, c, 1, buffer, static_cast<int>(params.size()+1)) > 0)
          {
            param1= buffer;
          }
          else
            return false;
          break;
          
        case 2: // [(n)]
          c= pcre_compile_exec("^(\\((\\d+)\\))?$", params.c_str(), patres, sizeof(patres)/sizeof(int));
          if (c < 0)
            return false;
          if (pcre_copy_substring(params.c_str(), patres, c, 2, buffer, static_cast<int>(params.size()+1)) > 0)
          {
            param1= buffer;
          }
          break;
          
        case 3: // (m, n)
          c= pcre_compile_exec("^\\((\\d+),\\s*(\\d+)\\)$", params.c_str(), patres, sizeof(patres)/sizeof(int));
          if (c < 0)
            return false;
          if (pcre_copy_substring(params.c_str(), patres, c, 1, buffer, static_cast<int>(params.size()+1)) > 0)
          {
            param1= buffer;
            
            if (pcre_copy_substring(params.c_str(), patres, c, 2, buffer, static_cast<int>(params.size()+1)) > 0)
              param2= buffer;
            else
              return false;
          }
          else
            return false;
          break;
          
        case 4: // (m[,n])
          c= pcre_compile_exec("^\\((\\d+)(,\\s*(\\d+))?\\)$", params.c_str(), patres, sizeof(patres)/sizeof(int));
          if (c < 0)
            return false;
          if (pcre_copy_substring(params.c_str(), patres, c, 1, buffer, static_cast<int>(params.size()+1)) > 0)
          {
            param1= buffer;
            if (pcre_copy_substring(params.c_str(), patres, c, 3, buffer, static_cast<int>(params.size()+1)) > 0)
              param2= buffer;
          }
          else
            return false;
          break;
          
        case 5: // [(m,n)]
          c= pcre_compile_exec("^(\\((\\d+)(,\\s*(\\d+))\\))?$", params.c_str(), patres, sizeof(patres)/sizeof(int));
          if (c < 0)
            return false;
          if (pcre_copy_substring(params.c_str(), patres, c, 2, buffer, static_cast<int>(params.size()+1)) > 0)
          {
            param1= buffer;
            if (pcre_copy_substring(params.c_str(), patres, c, 4, buffer, static_cast<int>(params.size()+1)) > 0)
              param2= buffer;
            else
              return false;
          }
          break;
          
        case 6: // [(m[,n])]
          c= pcre_compile_exec("^(\\((\\d+)(,\\s*(\\d+))?\\))?$", params.c_str(), patres, sizeof(patres)/sizeof(int));
          if (c < 0)
            return false;
          if (pcre_copy_substring(params.c_str(), patres, c, 2, buffer, static_cast<int>(params.size()+1)) > 0)
          {
            param1= buffer;
            if (pcre_copy_substring(params.c_str(), patres, c, 4, buffer, static_cast<int>(params.size()+1)) > 0)
              param2= buffer;
          }
          break;
          
        case 10: // ('a','b','c')
          if (params.empty())
            return false;
                
          c= pcre_compile_exec("^\\((.*?)\\)?$", params.c_str(), patres, sizeof(patres)/sizeof(int));
          if (c < 0)
            return false;
          // For now we just keep the parameter list as a whole. If necessary this can be split later for other uses.
          param1= buffer;
          // validate format
          if (pcre_copy_substring(params.c_str(), patres, c, 1, buffer, static_cast<int>(params.size()+1)) > 0)
          {
            std::list<std::string> tokens;
            if (!bec::tokenize_string_list(std::string(buffer), '\'', true, tokens))
              return false;
          }
          else
            return false;
          break;
      }
      
      // get precision
      if (*simpleType->numericPrecision() != bec::EMPTY_TYPE_PRECISION)
      {
        if (!param1.empty())
          precision= atoi(param1.c_str());
        
        // get scale
        if (*simpleType->numericScale() != bec::EMPTY_TYPE_SCALE)
        {
          if (!param2.empty())
            scale= atoi(param2.c_str());
        }
      }
      // get length
      else if (simpleType->characterMaximumLength() != bec::EMPTY_TYPE_MAXIMUM_LENGTH 
               || simpleType->characterOctetLength()   != bec::EMPTY_TYPE_OCTET_LENGTH)
      {
        if (!param1.empty())
          length= atoi(param1.c_str());
      }
      else if (*simpleType->parameterFormatType() == 10)
        explicitParams= param1;
    }
    else
      return false; // unknown type
  }
  else
    return false; // bad format
  
  return true;
}

bool bec::parseType(const std::string &type,
    const GrtVersionRef &targetVersion,
    const grt::ListRef<db_SimpleDatatype> &typeList,
    const grt::ListRef<db_UserDatatype>& user_types,
    const grt::ListRef<db_SimpleDatatype>& default_type_list,
    db_SimpleDatatypeRef &simpleType,
    db_UserDatatypeRef& userType,
    int &precision,
    int &scale,
    int &length,
    std::string &datatypeExplicitParams)
{
    if (user_types.is_valid())
    {
        std::string::size_type argp= type.find('(');
        std::string typeName= type;

        if (argp != std::string::npos)
            typeName= type.substr(0, argp);

        // 1st check if this is a user defined type
        for (size_t c= user_types.count(), i= 0; i < c; i++)
        {
            db_UserDatatypeRef utype(user_types[i]);

            if (g_strcasecmp(utype->name().c_str(), typeName.c_str()) == 0)
            {
                userType= utype;
                break;
            }
        }
    }

    if (userType.is_valid())
    {
        // if the type spec has an argument, we replace the arguments from the type definition
        // with the one provided by the user
        std::string finalType= userType->sqlDefinition();
        std::string::size_type tp;
        bool overriden_args= false;

        if ((tp= type.find('(')) != std::string::npos) // has user specified args
        {
            std::string::size_type p= finalType.find('('); 
            if (p != std::string::npos) // strip the original args
                finalType= finalType.substr(0, p);

            // extract the user spcified args and append to the specification
            finalType.append(type.substr(tp));

            overriden_args= true;
        }

        // parse usertype definition    
        if (!parseTypeDefinition(finalType, targetVersion, typeList.is_valid() ? typeList : default_type_list, simpleType,
            precision, scale, length, datatypeExplicitParams))
            return false;

        simpleType = db_SimpleDatatypeRef();
        userType = userType;
        if (!overriden_args)
        {
            precision = bec::EMPTY_COLUMN_PRECISION;
            scale = bec::EMPTY_COLUMN_SCALE;
            length = bec::EMPTY_COLUMN_LENGTH;
            datatypeExplicitParams = "";
        }
    }
    else
    {
        if (!parseTypeDefinition(type, targetVersion, typeList.is_valid() ? typeList : default_type_list, simpleType,
            precision, scale, length, datatypeExplicitParams))
            return false;

        userType = db_UserDatatypeRef();
    }
    return true;
}
