quick_ror_union_select.cc

/* -*- 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/util/functors.h>
#include <drizzled/optimizer/range.h>
#include <drizzled/optimizer/quick_range_select.h>
#include <drizzled/optimizer/quick_ror_union_select.h>
#include <drizzled/table.h>
#include <drizzled/system_variables.h>

#include <vector>
#include <algorithm>

using namespace std;

namespace drizzled {

optimizer::QuickRorUnionSelect::QuickRorUnionSelect(Session *session_param,
                                                    Table *table)
  :
    session(session_param),
    scans_inited(false)
{
  index= MAX_KEY;
  head= table;
  rowid_length= table->cursor->ref_length;
  record= head->record[0];
  alloc.init(session->variables.range_alloc_block_size);
  session_param->mem_root= &alloc;
}

/*
 * Function object that is used as the comparison function
 * for the priority queue in the QuickRorUnionSelect
 * class.
 */
class optimizer::compare_functor
{
  optimizer::QuickRorUnionSelect *self;
  public:
  compare_functor(optimizer::QuickRorUnionSelect *in_arg)
    : self(in_arg) { }
  inline bool operator()(const optimizer::QuickSelectInterface *i, const optimizer::QuickSelectInterface *j) const
  {
    int val= self->head->cursor->cmp_ref(i->last_rowid,
                                         j->last_rowid);
    return (val >= 0);
  }
};


int optimizer::QuickRorUnionSelect::init()
{
  queue= new priority_queue<QuickSelectInterface*, vector<QuickSelectInterface*>, compare_functor>(compare_functor(this));
  cur_rowid= alloc.alloc(2 * head->cursor->ref_length);
  prev_rowid= cur_rowid + head->cursor->ref_length;
  return 0;
}


int optimizer::QuickRorUnionSelect::reset()
{
  have_prev_rowid= false;
  if (not scans_inited)
  {
    BOOST_FOREACH(QuickSelectInterface* it, quick_selects)
    {
      if (it->init_ror_merged_scan(false))
      {
        return 0;
      }
    }
    scans_inited= true;
  }
  while (not queue->empty())
  {
    queue->pop();
  }
  /*
    Initialize scans for merged quick selects and put all merged quick
    selects into the queue.
  */
  BOOST_FOREACH(QuickSelectInterface* it, quick_selects)
  {
    if (it->reset())
      return 0;
    if (it->get_next() == HA_ERR_END_OF_FILE)
      continue;
    it->save_last_pos();
    queue->push(it);
  }
  if (head->cursor->startTableScan(1))
    return 0;
  return 0;
}


void optimizer::QuickRorUnionSelect::push_quick_back(QuickSelectInterface *quick_sel_range)
{
  quick_selects.push_back(quick_sel_range);
}


optimizer::QuickRorUnionSelect::~QuickRorUnionSelect()
{
  while (! queue->empty())
  {
    queue->pop();
  }
  delete queue;
  for_each(quick_selects.begin(),
           quick_selects.end(),
           DeletePtr());
  quick_selects.clear();
  if (head->cursor->inited != Cursor::NONE)
  {
    head->cursor->endTableScan();
  }
  alloc.free_root(MYF(0));
}


bool optimizer::QuickRorUnionSelect::is_keys_used(const boost::dynamic_bitset<>& fields)
{
  BOOST_FOREACH(QuickSelectInterface* it, quick_selects)
  {
    if (it->is_keys_used(fields))
      return true;
  }
  return false;
}


int optimizer::QuickRorUnionSelect::get_next()
{
  int error;
  int dup_row;
  optimizer::QuickSelectInterface *quick= NULL;
  unsigned char *tmp= NULL;

  do
  {
    do
    {
      if (queue->empty())
        return HA_ERR_END_OF_FILE;
      /* Ok, we have a queue with >= 1 scans */

      quick= queue->top();
      memcpy(cur_rowid, quick->last_rowid, rowid_length);

      /* put into queue rowid from the same stream as top element */
      if ((error= quick->get_next()))
      {
        if (error != HA_ERR_END_OF_FILE)
          return(error);
        queue->pop();
      }
      else
      {
        quick->save_last_pos();
        queue->pop();
        queue->push(quick);
      }

      if (!have_prev_rowid)
      {
        /* No rows have been returned yet */
        dup_row= false;
        have_prev_rowid= true;
      }
      else
        dup_row= ! head->cursor->cmp_ref(cur_rowid, prev_rowid);
    } while (dup_row);

    tmp= cur_rowid;
    cur_rowid= prev_rowid;
    prev_rowid= tmp;

    error= head->cursor->rnd_pos(quick->record, prev_rowid);
  } while (error == HA_ERR_RECORD_DELETED);
  return error;
}


void optimizer::QuickRorUnionSelect::add_info_string(string *str)
{
  bool first= true;
  str->append("union(");
  BOOST_FOREACH(QuickSelectInterface* it, quick_selects)
  {
    if (first)
      first= false;
    else
      str->append(",");
    it->add_info_string(str);
  }
  str->append(")");
}


void optimizer::QuickRorUnionSelect::add_keys_and_lengths(string *key_names, string *used_lengths)
{
  bool first= true;
  BOOST_FOREACH(QuickSelectInterface* it, quick_selects)
  {
    if (first)
      first= false;
    else
    {
      used_lengths->append(",");
      key_names->append(",");
    }
    it->add_keys_and_lengths(key_names, used_lengths);
  }
}

} /* namespace drizzled */