~drizzle-trunk/drizzle/development

/* -*- mode: c++; c-basic-offset: 2; indent-tabs-mode: nil; -*-

 *  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:

 *

 *  Copyright (C) 2008-2009 Sun Microsystems, Inc.

 *

 *  This program is free software; you can redistribute it and/or modify

 *  it under the terms of the GNU General Public License as published by

 *  the Free Software Foundation; version 2 of the License.

 *

 *  This program is distributed in the hope that it will be useful,

 *  but WITHOUT ANY WARRANTY; without even the implied warranty of

 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

 *  GNU General Public License for more details.

 *

 *  You should have received a copy of the GNU General Public License

 *  along with this program; if not, write to the Free Software

 *  Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA

 */

#include <config.h>

#include <drizzled/session.h>

#include <drizzled/records.h>

#include <drizzled/util/functors.h>

#include <drizzled/optimizer/quick_range_select.h>

#include <drizzled/optimizer/quick_index_merge_select.h>

#include <drizzled/internal/m_string.h>

#include <drizzled/unique.h>

#include <drizzled/key.h>

#include <drizzled/table.h>

#include <drizzled/system_variables.h>

#include <vector>

using namespace std;

namespace drizzled {

static int refpos_order_cmp(void *arg, const void *a, const void *b)

{

  Cursor *cursor= (Cursor*)arg;

  return cursor->cmp_ref((const unsigned char *) a, (const unsigned char *) b);

}

optimizer::QuickIndexMergeSelect::QuickIndexMergeSelect(Session *session_param,

                                                        Table *table)

  :

    pk_quick_select(NULL),

    session(session_param)

{

  index= MAX_KEY;

  head= table;

  memset(&read_record, 0, sizeof(read_record));

  alloc.init(session->variables.range_alloc_block_size);

}

int optimizer::QuickIndexMergeSelect::init()

{

  return 0;

}

int optimizer::QuickIndexMergeSelect::reset()

{

  return (read_keys_and_merge());

}

void optimizer::QuickIndexMergeSelect::push_quick_back(optimizer::QuickRangeSelect *quick_sel_range)

{

  /*

    Save quick_select that does scan on clustered primary key as it will be

    processed separately.

  */

  if (head->cursor->primary_key_is_clustered() &&

      quick_sel_range->index == head->getShare()->getPrimaryKey())

  {

    pk_quick_select= quick_sel_range;

  }

  else

  {

    quick_selects.push_back(quick_sel_range);

  }

}

optimizer::QuickIndexMergeSelect::~QuickIndexMergeSelect()

{

  BOOST_FOREACH(QuickRangeSelect* it, quick_selects)

    it->cursor= NULL;

  for_each(quick_selects.begin(), quick_selects.end(), DeletePtr());

  quick_selects.clear();

  delete pk_quick_select;

  alloc.free_root(MYF(0));

}

int optimizer::QuickIndexMergeSelect::read_keys_and_merge()

{

  vector<optimizer::QuickRangeSelect *>::iterator it= quick_selects.begin();

  optimizer::QuickRangeSelect *cur_quick= NULL;

  int result;

  Unique *unique= NULL;

  Cursor *cursor= head->cursor;

  cursor->extra(HA_EXTRA_KEYREAD);

  head->prepare_for_position();

  cur_quick= *it;

  ++it;

  assert(cur_quick != 0);

  /*

    We reuse the same instance of Cursor so we need to call both init and

    reset here.

  */

  if (cur_quick->init() || cur_quick->reset())

    return 0;

  unique= new Unique(refpos_order_cmp,

                     (void *) cursor,

                     cursor->ref_length,

                     session->variables.sortbuff_size);

  if (! unique)

    return 0;

  for (;;)

  {

    while ((result= cur_quick->get_next()) == HA_ERR_END_OF_FILE)

    {

      cur_quick->range_end();

      if (it == quick_selects.end())

      {

        break;

      }

      cur_quick= *it;

      ++it;

      if (! cur_quick)

        break;

      if (cur_quick->cursor->inited != Cursor::NONE)

        cur_quick->cursor->endIndexScan();

      if (cur_quick->init() || cur_quick->reset())

        return 0;

    }

    if (result)

    {

      if (result != HA_ERR_END_OF_FILE)

      {

        cur_quick->range_end();

        return result;

      }

      break;

    }

    if (session->getKilled())

      return 0;

    /* skip row if it will be retrieved by clustered PK scan */

    if (pk_quick_select && pk_quick_select->row_in_ranges())

      continue;

    cur_quick->cursor->position(cur_quick->record);

    result= unique->unique_add((char*) cur_quick->cursor->ref);

    if (result)

      return 0;

  }

  /* ok, all row ids are in Unique */

  result= unique->get(head);

  delete unique;

  doing_pk_scan= false;

  /* index_merge currently doesn't support "using index" at all */

  cursor->extra(HA_EXTRA_NO_KEYREAD);

  /* start table scan */

  if ((result= read_record.init_read_record(session, head, (optimizer::SqlSelect*) 0, 1, 1)))

  {

    head->print_error(result, MYF(0));

    return 0;

  }

  return result;

}

int optimizer::QuickIndexMergeSelect::get_next()

{

  int result;

  if (doing_pk_scan)

    return(pk_quick_select->get_next());

  if ((result= read_record.read_record(&read_record)) == -1)

  {

    result= HA_ERR_END_OF_FILE;

    read_record.end_read_record();

    /* All rows from Unique have been retrieved, do a clustered PK scan */

    if (pk_quick_select)

    {

      doing_pk_scan= true;

      if ((result= pk_quick_select->init()) ||

          (result= pk_quick_select->reset()))

        return result;

      return(pk_quick_select->get_next());

    }

  }

  return result;

}

bool optimizer::QuickIndexMergeSelect::is_keys_used(const boost::dynamic_bitset<>& fields)

{

  BOOST_FOREACH(QuickRangeSelect* it, quick_selects)

  {

    if (is_key_used(head, it->index, fields))

      return 1;

  }

  return 0;

}

void optimizer::QuickIndexMergeSelect::add_info_string(string *str)

{

  bool first= true;

  str->append("sort_union(");

  BOOST_FOREACH(QuickRangeSelect* it, quick_selects)

  {

    if (! first)

      str->append(",");

    else

      first= false;

    it->add_info_string(str);

  }

  if (pk_quick_select)

  {

    str->append(",");

    pk_quick_select->add_info_string(str);

  }

  str->append(")");

}

void optimizer::QuickIndexMergeSelect::add_keys_and_lengths(string *key_names,

                                                            string *used_lengths)

{

  char buf[64];

  uint32_t length= 0;

  bool first= true;

  BOOST_FOREACH(QuickRangeSelect* it, quick_selects)

  {

    if (first)

      first= false;

    else

    {

      key_names->append(",");

      used_lengths->append(",");

    }

    KeyInfo *key_info= head->key_info + it->index;

    key_names->append(key_info->name);

    length= internal::int64_t2str(it->max_used_key_length, buf, 10) - buf;

    used_lengths->append(buf, length);

  }

  if (pk_quick_select)

  {

    KeyInfo *key_info= head->key_info + pk_quick_select->index;

    key_names->append(",");

    key_names->append(key_info->name);

    length= internal::int64_t2str(pk_quick_select->max_used_key_length, buf, 10) - buf;

    used_lengths->append(",");

    used_lengths->append(buf, length);

  }

}

} /* namespace drizzled */