/*
 * Copyright (c) 2007, 2015, 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
 */

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

#include "base/string_utilities.h"
#include "base/util_functions.h"
#include "base/log.h"

#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 "grtdb/db_helpers.h"

#include "mysql-parser.h"
#include "MySQLParser.h" // For token types.

DEFAULT_LOG_DOMAIN("dbhelpers");

#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 (base::string_compare(types[i]->name(), name, false)==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();
}

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

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 = 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 = parse_version(type.get_grt(), validity.substr(1));
          if (version_greater(valid_version, target_version))
            match = true;
        }
        break;
      case '>':
        if (validity[1] == '=')
        {
          valid_version = 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 = parse_version(type.get_grt(), validity.substr(1));
          if (version_greater(target_version, valid_version))
            match = true;
        }
        break;
      case '=':
        valid_version = 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);

  atable->createDate(grt::StringRef(base::fmttime(0, DATETIME_FMT)));
  atable->lastChangeDate(atable->createDate());
  
  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;
}


void TableHelper::reorder_foreign_key_for_index(const db_ForeignKeyRef &fk, const db_IndexRef &index)
{
  // make the order of the columns in the FK match that of the index
  size_t column_count = fk->columns().count();

  if (fk->columns().count() != fk->referencedColumns().count())
  {
    log_error("Internal consistency error: number of items in fk->columns and fk->referencedColumns() for %s.%s.%s do not match\n",
              fk->owner()->owner()->name().c_str(), fk->owner()->name().c_str(), fk->name().c_str());
    return;
  }
  if (column_count > index->columns().count())
  {
    log_error("Internal consistency error: number of items in index for FK is less than columns in FK %s.%s.%s\n",
             fk->owner()->owner()->name().c_str(), fk->owner()->name().c_str(), fk->name().c_str());
    return;
  }

  for (size_t j= 0; j < column_count; j++)
  {
    if (index->columns()[j]->referencedColumn() != fk->columns()[j])
    {
      // if the column in the index is not the expected one, reorder the FK

      // find the position of the column in the FK
      for (size_t k = j+1; k < column_count; k++)
      {
        if (index->columns()[j]->referencedColumn() == fk->columns()[k])
        {
          // reorder the columns in the FK
          fk->columns().reorder(k, j);
          fk->referencedColumns().reorder(k, j);
          break;
        }
      }
      break;
    }
  }
}


db_IndexRef TableHelper::find_index_usable_by_fk(const db_ForeignKeyRef &fk, const db_IndexRef &other_than, bool allow_any_order)
{
  size_t column_count = fk->columns().count();
  db_TableRef table(db_TableRef::cast_from(fk->owner()));

  if (column_count == 0)
    return db_IndexRef();
  
  for (size_t c= table->indices().count(), i= 0; i < c; i++)
  {
    db_IndexRef index(table->indices()[i]);

    if (index == other_than)
      continue;

    // smaller indexes than the FK won't work
    size_t index_column_count = index->columns().count();
    if (index_column_count >= column_count)
    {
      bool index_usable;
      if (allow_any_order)
      {
        index_usable = true;
        for (size_t j= 0; j < column_count; j++)
        {
          bool ok= false;
          // check whether this FK column is in the index
          for (size_t k= 0; k < index_column_count; k++)
          {
            if (index->columns()[k]->referencedColumn() == fk->columns()[j])
            {
              ok= true;
              break;
            }
          }
          if (!ok)
          {
            index_usable= false;
            break;
          }
        }

        if (index_usable)
        {
          size_t c = 0;
          // now check whether only columns from the prefix of the index are in use in the FK..
          // example: FK (a, b) and INDEX (a, b, c) is OK
          //          FK (b, c) and INDEX (a, b, c) is not ok
          for (size_t j= 0; j < column_count && c < column_count; j++)
          {
            if (fk->columns().get_index(index->columns()[j]->referencedColumn()) != grt::BaseListRef::npos)
              c++; // jth column is in the FK, so it's OK so far
            else
            {
              index_usable = false;
              break; // jth column of the index is not in the FK, so this index can't be used
            }
          }
        }
      }
      else
      {
        index_usable = false;
        // make sure that the FK form prefix of the columns in the index
        for (size_t j= 0; j < column_count; j++)
        {
          if (index->columns()[j]->referencedColumn() != fk->columns()[j])
          {
            index_usable = false;
            break;
          }
          else
            index_usable = true; // there's at least one match
        }
      }
      if (index_usable)
        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())
  {
    db_IndexRef other;
    if ((other = find_index_usable_by_fk(fk, index, true)).is_valid())
    {
      // the index from this FK is redundant, so remove it
      fk->index(db_IndexRef());
      table->indices().remove_value(index);
      reorder_foreign_key_for_index(fk, other);
      return;
    }

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

    // remove columns that are gone from the FK
    for (ssize_t 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
    create_index_for_fk_if_needed(fk);
}


bool TableHelper::create_index_for_fk_if_needed(db_ForeignKeyRef fk)
{
  db_IndexRef index = find_index_usable_by_fk(fk, db_IndexRef(), true);
  if (index.is_valid())
    reorder_foreign_key_for_index(fk, index);
  else if (fk->columns().count() > 0)
  {
    index = create_index_for_fk(fk.get_grt(), fk);
    fk->index(index);
    fk->owner()->indices().insert(index);
    return true;
  }
  return false;
}


bool TableHelper::create_missing_indexes_for_foreign_keys(const db_TableRef &table)
{
  bool created= false;
  GRTLIST_FOREACH(db_ForeignKey, table->foreignKeys(), fk)
  {
    // check if an index already exists for the FK columns and if not, create one
    if (!(*fk)->index().is_valid())
      created = created || create_index_for_fk_if_needed(*fk);
    else
      reorder_foreign_key_for_index(*fk, (*fk)->index());
  }
  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;
  size_t 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, 
                                                         (int)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;
  size_t 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,
                                                         (int)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 empty line (NOT a 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.
 */
static void split_comment(const std::string &comment, size_t db_comment_len,
                          std::string *comment_ret, std::string *leftover_ret)
{
  size_t res;
  // XXX: check for Unicode line breaks! especially asian languages may not use the ANSI new line.
  const gchar* pointer_to_linebreak = NULL;

  { // find 1st occurrence of 2 consecutive newlines
    std::string::size_type pos = comment.find("\n\n");
    if (pos == std::string::npos)
      pos = comment.find("\r\n\r\n");

    if (pos != std::string::npos)
      pointer_to_linebreak = comment.c_str() + pos;
  }

  // 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 = comment.size(); // it's wrong to use g_utf8_strlen here because the comment length must be measured in bytes, not unichars

  // Don't break in the middle of a utf8 sequence
  if (res > db_comment_len)
  {
    if (g_utf8_get_char_validated(comment.c_str() + db_comment_len, (gssize)(res - db_comment_len)) == (gunichar)-1)
      res = g_utf8_pointer_to_offset(comment.c_str(), g_utf8_find_prev_char(comment.c_str(), comment.c_str() + db_comment_len));
    else
      res = db_comment_len;
  }
  if (comment_ret)
    *comment_ret = comment.substr(0, res);
  if (leftover_ret)
  {
    if (pointer_to_linebreak != NULL)
      *leftover_ret = comment.substr(res+1);
    else
      *leftover_ret = comment.substr(res);
  }
}

std::string TableHelper::get_sync_comment(const std::string &comment, const size_t max_len)
{
  std::string ret;
  if (comment.size() > max_len)
    split_comment(comment, max_len, &ret, NULL);
  else
    ret = comment;
  return ret;
};


std::string TableHelper::normalize_table_name_list(const std::string &schema, const std::string &table_name_list)
{
  std::vector<std::string> names = base::split(table_name_list, ",");

  for (std::vector<std::string>::iterator it = names.begin(); it != names.end(); ++it)
  {
    std::vector<std::string> tokens(base::split_qualified_identifier(base::trim(*it)));
    if (tokens.size() == 1)
      tokens.insert(tokens.begin(), schema);

    for (std::vector<std::string>::iterator t = tokens.begin(); t != tokens.end(); ++t)
      *t = base::quote_identifier(base::unquote_identifier(*t), '`');

    *it = base::join(tokens, ".");
  }

  return base::join(names, ",");
}

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(base::string_compare(value, "NULL", false) ? 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.
}

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

void CatalogHelper::apply_defaults(db_mysql_ColumnRef column)
{

  // for numeric types only
  static std::map<std::string, int> def_precision_map;
  if (def_precision_map.empty())
  {
    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;

  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))
  {
    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(defaultCollationForCharset(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());
      }
      else
      {
        if(strlen(table->defaultCollationName().c_str()) == 0)
          table->defaultCollationName(defaultCollationForCharset(table->defaultCharacterSetName()));
      }
      
      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));
      }
    }
  }
}

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

/**
 *	Compares the given typename with what is in the type list, including the synonyms and returns
 *	the type whose name or synonym matches.
 */
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)
  {
    bool type_found = base::same_string(types[i]->name(), name, false);
    if (!type_found)
    {
      // Type has not the default name, but maybe one of the synonyms.
      for (ListRef<internal::String>::const_iterator synonym = types[i]->synonyms().begin(); synonym != types[i]->synonyms().end(); ++synonym)
      {
        if (base::same_string(*synonym, name, false))
        {
          type_found = true;
          break;
        }
      }
    }

    if (type_found)
    {
      if (!target_version.is_valid() || CatalogHelper::is_type_valid_for_version(types[i], target_version))
        return types[i];
    }
  }
  return db_SimpleDatatypeRef();
}

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

static bool parse_type(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)
{
  // No char sets necessary for parsing data types as there's no repertoire necessary/allowed in any
  // data type part. Neither do we need an sql mode (string lists in enum defs only allow
  // single quoted text).
  // 
  // Note: we parse here more than the pure data type name + precision/scale (namely additional parameters
  // like charsets etc.). That doesn't affect the main task here, however. Additionally stuff
  // is simply ignored for now (but it must be a valid definition).
  MySQLRecognizer recognizer(bec::version_to_int(targetVersion), "", std::set<std::string>());
  recognizer.parse(type.c_str(), type.size(), true, PuDataType);
  if (!recognizer.error_info().empty())
    return false;

  MySQLRecognizerTreeWalker walker = recognizer.tree_walker();

  // A type name can consist of up to 3 parts (e.g. "national char varying").
  std::string type_name = walker.token_text();
  
  switch (walker.token_type())
  {
  case DOUBLE_SYMBOL:
    walker.next();
    if (walker.token_type() == PRECISION_SYMBOL)
      walker.next(); // Simply ignore syntactic sugar.
    break;

  case NATIONAL_SYMBOL:
    walker.next();
    type_name += " " + walker.token_text();
    walker.next();
    if (walker.token_type() == VARYING_SYMBOL)
    {
      type_name += " " + walker.token_text();
      walker.next();
    }
    break;

  case NCHAR_SYMBOL:
    walker.next();
    if (walker.token_type() == VARCHAR_SYMBOL || walker.token_type() == VARYING_SYMBOL)
    {
      type_name += " " + walker.token_text();
      walker.next();
    }
    break;

  case CHAR_SYMBOL:
    walker.next();
    if (walker.token_type() == VARYING_SYMBOL)
    {
      type_name += " " + walker.token_text();
      walker.next();
    }
    break;

  case LONG_SYMBOL:
    walker.next();
    switch (walker.token_type())
    {
    case CHAR_SYMBOL: // LONG CHAR VARYING
      if (walker.look_ahead(1) == VARYING_SYMBOL) // Otherwise we may get e.g. LONG CHAR SET...
      {
        type_name += " " + walker.token_text();
        walker.next();
        type_name += " " + walker.token_text();
        walker.next();
      }
      break;

    case VARBINARY_SYMBOL:
    case VARCHAR_SYMBOL:
      type_name += " " + walker.token_text();
      walker.next();
    }
    break;

  default:
    walker.next();
  }

  simpleType = findType(typeList, targetVersion, type_name);

  if (!simpleType.is_valid()) // Should always be valid at this point or we have messed up our type list.
    return false;

  if (!walker.is(OPEN_PAR_SYMBOL))
    return true;

  walker.next();

  // We use the simple type properties for char length to learn if we have a length here or a precision.
  // We could indicate that in the grammar instead, however the handling in WB is a bit different
  // than what the server grammar would suggest (e.g. the length is also used for integer types, in the grammar).
  if (simpleType->characterMaximumLength() != bec::EMPTY_TYPE_MAXIMUM_LENGTH
    || simpleType->characterOctetLength() != bec::EMPTY_TYPE_OCTET_LENGTH)
  {
    if (walker.token_type() != INT_NUMBER)
      return false;
    length = base::atoi<int>(walker.token_text().c_str());
    return true;
  }

  if (!walker.is(INT_NUMBER))
  {
    // ENUM or SET.
    do 
    {
      if (!explicitParams.empty())
        explicitParams += ", ";
      explicitParams += walker.token_text(true);
      walker.next();
      walker.skip_if(COMMA_SYMBOL); // We normalize the whitespace around the commas.
    } while (!walker.is(CLOSE_PAR_SYMBOL));
    explicitParams = "(" + explicitParams + ")";

    return true;
  }

  // Finally all cases with either precision, scale or both.
  precision = base::atoi<int>(walker.token_text().c_str());
  walker.next();
  if (walker.token_type() != COMMA_SYMBOL)
    return true;
  walker.next();
  scale = base::atoi<int>(walker.token_text().c_str());

  return true;
}

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

bool bec::parse_type_definition(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 (base::string_compare(utype->name(), typeName, false) == 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) // Are there 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 specified args and append to the specification.
      finalType.append(type.substr(tp));

      overriden_args = true;
    }

    // Parse user type definition.
    if (!parse_type(finalType, targetVersion, typeList.is_valid() ? typeList : default_type_list,
      simpleType, precision, scale, length, datatypeExplicitParams))
      return false;

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

    userType = db_UserDatatypeRef();
  }
  return true;
}

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

std::string bec::get_default_collation_for_charset(const db_SchemaRef &schema, const std::string &character_set)
{
  if (schema->owner().is_valid())
  {
    db_CatalogRef catalog(db_CatalogRef::cast_from(schema->owner()));
    db_CharacterSetRef charset = find_named_object_in_list(catalog->characterSets(), character_set);
    if (charset.is_valid())
    {
      return charset->defaultCollation();
    }
  }
  else
    log_warning("While checking diff, catalog ref was found to be invalid\n");


  return "";
}

std::string bec::get_default_collation_for_charset(const db_TableRef &table, const std::string &character_set)
{
  if (table->owner().is_valid())
    return bec::get_default_collation_for_charset(db_SchemaRef::cast_from(table->owner()), character_set);
  else
    log_warning("While checking diff, table ref was found to be invalid\n");
  return "";
}


std::string bec::TableHelper::generate_comment_text(const std::string& comment_text, size_t comment_lenght)
{
  if (comment_text.size() > comment_lenght)
  {
    std::string comment, leftover;

    split_comment(comment_text, comment_lenght, &comment, &leftover);

    if (!comment.empty())
      comment = "'"+base::escape_sql_string(comment)+"'";
    if (!leftover.empty())
    {
      base::replace(leftover, "*/", "*\\/");
      comment.append(" /* comment truncated */ /*")
      .append(leftover)
      .append("*/");
    }
    return comment;
  }
  else if (!comment_text.empty())
    return "'"+base::escape_sql_string(comment_text)+"'";

  return "";
}