~drizzle-developers/ubuntu/natty/drizzle/natty

/* -*- mode: c++; c-basic-offset: 2; indent-tabs-mode: nil; -*-
 *  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
 *
 *  Copyright (C) 2008-2009 Sun Microsystems
 *
 *  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 <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];
  memory::init_sql_alloc(&alloc, session->variables.range_alloc_block_size, 0);
  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<optimizer::QuickSelectInterface *,
                       vector<optimizer::QuickSelectInterface *>,
                       optimizer::compare_functor >(optimizer::compare_functor(this));
  if (! (cur_rowid= (unsigned char*) alloc.alloc_root(2 * head->cursor->ref_length)))
  {
    return 0;
  }
  prev_rowid= cur_rowid + head->cursor->ref_length;
  return 0;
}


int optimizer::QuickRorUnionSelect::reset()
{
  int error;
  have_prev_rowid= false;
  if (! scans_inited)
  {
    for (vector<optimizer::QuickSelectInterface *>::iterator it= quick_selects.begin();
         it != quick_selects.end();
         ++it)
    {
      if ((*it)->init_ror_merged_scan(false))
      {
        return 0;
      }
    }
    scans_inited= true;
  }
  while (! queue->empty())
  {
    queue->pop();
  }
  /*
    Initialize scans for merged quick selects and put all merged quick
    selects into the queue.
  */
  for (vector<optimizer::QuickSelectInterface *>::iterator it= quick_selects.begin();
       it != quick_selects.end();
       ++it)
  {
    if ((*it)->reset())
    {
      return 0;
    }
    
    error= (*it)->get_next();
    if (error)
    {
      if (error == HA_ERR_END_OF_FILE)
      {
        continue;
      }
    }
    (*it)->save_last_pos();
    queue->push(*it);
  }

  if (head->cursor->startTableScan(1))
  {
    return 0;
  }

  return 0;
}


bool
optimizer::QuickRorUnionSelect::push_quick_back(QuickSelectInterface *quick_sel_range)
{
  quick_selects.push_back(quick_sel_range);
  return false;
}


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 MyBitmap *fields)
{
  for (vector<optimizer::QuickSelectInterface *>::iterator it= quick_selects.begin();
       it != quick_selects.end();
       ++it)
  {
    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(STRING_WITH_LEN("union("));
  for (vector<optimizer::QuickSelectInterface *>::iterator it= quick_selects.begin();
       it != quick_selects.end();
       ++it)
  {
    if (! first)
      str->append(',');
    else
      first= false;
    (*it)->add_info_string(str);
  }
  str->append(')');
}


void optimizer::QuickRorUnionSelect::add_keys_and_lengths(String *key_names,
                                                          String *used_lengths)
{
  bool first= true;
  for (vector<optimizer::QuickSelectInterface *>::iterator it= quick_selects.begin();
       it != quick_selects.end();
       ++it)
  {
    if (first)
    {
      first= false;
    }
    else
    {
      used_lengths->append(',');
      key_names->append(',');
    }
    (*it)->add_keys_and_lengths(key_names, used_lengths);
  }
}

} /* namespace drizzled */