~ubuntu-branches/ubuntu/saucy/drizzle/saucy-proposed

/* -*- 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/optimizer/quick_range.h"
#include "drizzled/optimizer/quick_range_select.h"
#include "drizzled/sql_bitmap.h"
#include "drizzled/internal/m_string.h"
#include <fcntl.h>
#include "drizzled/memory/multi_malloc.h"

using namespace std;

namespace drizzled
{


optimizer::QuickRangeSelect::QuickRangeSelect(Session *session,
                                              Table *table,
                                              uint32_t key_nr,
                                              bool no_alloc,
                                              memory::Root *parent_alloc,
                                              bool *create_error)
  :
    mrr_flags(0),
    alloc(),
    cursor(NULL),
    ranges(),
    in_ror_merged_scan(false),
    column_bitmap(),
    save_read_set(NULL),
    save_write_set(NULL),
    free_file(false),
    cur_range(NULL),
    last_range(NULL),
    qr_traversal_ctx(),
    mrr_buf_size(0),
    mrr_buf_desc(NULL),
    key_parts(NULL),
    dont_free(false)
{
  my_bitmap_map *bitmap= NULL;

  sorted= 0;
  index= key_nr;
  head= table;
  key_part_info= head->key_info[index].key_part;
  my_init_dynamic_array(&ranges, sizeof(optimizer::QuickRange*), 16, 16);

  /* 'session' is not accessible in QuickRangeSelect::reset(). */
  mrr_buf_size= session->variables.read_rnd_buff_size;
  mrr_buf_desc= NULL;

  if (! no_alloc && ! parent_alloc)
  {
    // Allocates everything through the internal memroot
    memory::init_sql_alloc(&alloc, session->variables.range_alloc_block_size, 0);
    session->mem_root= &alloc;
  }
  else
  {
    memset(&alloc, 0, sizeof(alloc));
  }

  cursor= head->cursor;
  record= head->record[0];
  save_read_set= head->read_set;
  save_write_set= head->write_set;

  /* Allocate a bitmap for used columns. Using memory::sql_alloc instead of malloc
     simply as a "fix" to the MySQL 6.0 code that also free()s it at the
     same time we destroy the mem_root.
   */

  bitmap= reinterpret_cast<my_bitmap_map*>(memory::sql_alloc(head->s->column_bitmap_size));
  if (! bitmap)
  {
    column_bitmap.setBitmap(NULL);
    *create_error= 1;
  }
  else
  {
    column_bitmap.init(bitmap, head->s->fields);
  }
}


int optimizer::QuickRangeSelect::init()
{
  if (cursor->inited != Cursor::NONE)
    cursor->ha_index_or_rnd_end();
  return (cursor->ha_index_init(index, 1));
}


void optimizer::QuickRangeSelect::range_end()
{
  if (cursor->inited != Cursor::NONE)
    cursor->ha_index_or_rnd_end();
}


optimizer::QuickRangeSelect::~QuickRangeSelect()
{
  if (! dont_free)
  {
    /* cursor is NULL for CPK scan on covering ROR-intersection */
    if (cursor)
    {
      range_end();
      if (head->key_read)
      {
        head->key_read= 0;
        cursor->extra(HA_EXTRA_NO_KEYREAD);
      }
      if (free_file)
      {
        cursor->ha_external_lock(current_session, F_UNLCK);
        cursor->close();
        delete cursor;
      }
    }
    delete_dynamic(&ranges); /* ranges are allocated in alloc */
    free_root(&alloc,MYF(0));
  }
  head->column_bitmaps_set(save_read_set, save_write_set);
  assert(mrr_buf_desc == NULL);
  if (mrr_buf_desc)
  {
    free(mrr_buf_desc);
  }
}


int optimizer::QuickRangeSelect::init_ror_merged_scan(bool reuse_handler)
{
  Cursor *save_file= cursor, *org_file;
  Session *session;

  in_ror_merged_scan= 1;
  if (reuse_handler)
  {
    if (init() || reset())
    {
      return 0;
    }
    head->column_bitmaps_set(&column_bitmap, &column_bitmap);
    goto end;
  }

  /* Create a separate Cursor object for this quick select */
  if (free_file)
  {
    /* already have own 'Cursor' object. */
    return 0;
  }

  session= head->in_use;
  if (! (cursor= head->cursor->clone(session->mem_root)))
  {
    /*
      Manually set the error flag. Note: there seems to be quite a few
      places where a failure could cause the server to "hang" the client by
      sending no response to a query. ATM those are not real errors because
      the storage engine calls in question happen to never fail with the
      existing storage engines.
    */
    my_error(ER_OUT_OF_RESOURCES, MYF(0));
    /* Caller will free the memory */
    goto failure;
  }

  head->column_bitmaps_set(&column_bitmap, &column_bitmap);

  if (cursor->ha_external_lock(session, F_RDLCK))
    goto failure;

  if (init() || reset())
  {
    cursor->ha_external_lock(session, F_UNLCK);
    cursor->close();
    goto failure;
  }
  free_file= true;
  last_rowid= cursor->ref;

end:
  /*
    We are only going to read key fields and call position() on 'cursor'
    The following sets head->tmp_set to only use this key and then updates
    head->read_set and head->write_set to use this bitmap.
    The now bitmap is stored in 'column_bitmap' which is used in ::get_next()
  */
  org_file= head->cursor;
  head->cursor= cursor;
  /* We don't have to set 'head->keyread' here as the 'cursor' is unique */
  if (! head->no_keyread)
  {
    head->key_read= 1;
    head->mark_columns_used_by_index(index);
  }
  head->prepare_for_position();
  head->cursor= org_file;
  column_bitmap= *head->read_set;
  head->column_bitmaps_set(&column_bitmap, &column_bitmap);

  return 0;

failure:
  head->column_bitmaps_set(save_read_set, save_write_set);
  delete cursor;
  cursor= save_file;
  return 0;
}


void optimizer::QuickRangeSelect::save_last_pos()
{
  cursor->position(record);
}


bool optimizer::QuickRangeSelect::unique_key_range() const
{
  if (ranges.elements == 1)
  {
    optimizer::QuickRange *tmp= *((optimizer::QuickRange**)ranges.buffer);
    if ((tmp->flag & (EQ_RANGE | NULL_RANGE)) == EQ_RANGE)
    {
      KEY *key=head->key_info+index;
      return ((key->flags & (HA_NOSAME)) == HA_NOSAME &&
	      key->key_length == tmp->min_length);
    }
  }
  return false;
}


int optimizer::QuickRangeSelect::reset()
{
  uint32_t buf_size= 0;
  unsigned char *mrange_buff= NULL;
  int error= 0;
  HANDLER_BUFFER empty_buf;
  last_range= NULL;
  cur_range= (optimizer::QuickRange**) ranges.buffer;

  if (cursor->inited == Cursor::NONE && (error= cursor->ha_index_init(index, 1)))
  {
    return error;
  }

  /* Allocate buffer if we need one but haven't allocated it yet */
  if (mrr_buf_size && ! mrr_buf_desc)
  {
    buf_size= mrr_buf_size;
    while (buf_size && ! memory::multi_malloc(false,
                                              &mrr_buf_desc,
                                              sizeof(*mrr_buf_desc),
                                              &mrange_buff,
                                              buf_size,
                                              NULL))
    {
      /* Try to shrink the buffers until both are 0. */
      buf_size/= 2;
    }
    if (! mrr_buf_desc)
    {
      return HA_ERR_OUT_OF_MEM;
    }

    /* Initialize the Cursor buffer. */
    mrr_buf_desc->buffer= mrange_buff;
    mrr_buf_desc->buffer_end= mrange_buff + buf_size;
    mrr_buf_desc->end_of_used_area= mrange_buff;
  }

  if (! mrr_buf_desc)
  {
    empty_buf.buffer= NULL;
    empty_buf.buffer_end= NULL;
    empty_buf.end_of_used_area= NULL;
  }

  if (sorted)
  {
     mrr_flags|= HA_MRR_SORTED;
  }
  RANGE_SEQ_IF seq_funcs= {
    optimizer::quick_range_seq_init,
    optimizer::quick_range_seq_next
  };
  error= cursor->multi_range_read_init(&seq_funcs,
                                       (void*) this,
                                       ranges.elements,
                                       mrr_flags,
                                       mrr_buf_desc ? mrr_buf_desc : &empty_buf);
  return error;
}


int optimizer::QuickRangeSelect::get_next()
{
  char *dummy= NULL;
  if (in_ror_merged_scan)
  {
    /*
      We don't need to signal the bitmap change as the bitmap is always the
      same for this head->cursor
    */
    head->column_bitmaps_set(&column_bitmap, &column_bitmap);
  }

  int result= cursor->multi_range_read_next(&dummy);

  if (in_ror_merged_scan)
  {
    /* Restore bitmaps set on entry */
    head->column_bitmaps_set(save_read_set, save_write_set);
  }
  return result;
}


int optimizer::QuickRangeSelect::get_next_prefix(uint32_t prefix_length,
                                                 key_part_map keypart_map,
                                                 unsigned char *cur_prefix)
{
  for (;;)
  {
    int result;
    key_range start_key, end_key;
    if (last_range)
    {
      /* Read the next record in the same range with prefix after cur_prefix. */
      assert(cur_prefix != 0);
      result= cursor->index_read_map(record,
                                     cur_prefix,
                                     keypart_map,
                                     HA_READ_AFTER_KEY);
      if (result || (cursor->compare_key(cursor->end_range) <= 0))
        return result;
    }

    uint32_t count= ranges.elements - (cur_range - (optimizer::QuickRange**) ranges.buffer);
    if (count == 0)
    {
      /* Ranges have already been used up before. None is left for read. */
      last_range= 0;
      return HA_ERR_END_OF_FILE;
    }
    last_range= *(cur_range++);

    start_key.key= (const unsigned char*) last_range->min_key;
    start_key.length= min(last_range->min_length, (uint16_t)prefix_length);
    start_key.keypart_map= last_range->min_keypart_map & keypart_map;
    start_key.flag= ((last_range->flag & NEAR_MIN) ? HA_READ_AFTER_KEY :
		                                                (last_range->flag & EQ_RANGE) ?
		                                                HA_READ_KEY_EXACT : HA_READ_KEY_OR_NEXT);
    end_key.key= (const unsigned char*) last_range->max_key;
    end_key.length= min(last_range->max_length, (uint16_t)prefix_length);
    end_key.keypart_map= last_range->max_keypart_map & keypart_map;
    /*
      We use READ_AFTER_KEY here because if we are reading on a key
      prefix we want to find all keys with this prefix
    */
    end_key.flag= (last_range->flag & NEAR_MAX ? HA_READ_BEFORE_KEY :
		                                             HA_READ_AFTER_KEY);

    result= cursor->read_range_first(last_range->min_keypart_map ? &start_key : 0,
				                             last_range->max_keypart_map ? &end_key : 0,
                                     test(last_range->flag & EQ_RANGE),
				                             sorted);
    if (last_range->flag == (UNIQUE_RANGE | EQ_RANGE))
      last_range= 0; // Stop searching

    if (result != HA_ERR_END_OF_FILE)
      return result;
    last_range= 0; // No matching rows; go to next range
  }
}


bool optimizer::QuickRangeSelect::row_in_ranges()
{
  optimizer::QuickRange *res= NULL;
  uint32_t min= 0;
  uint32_t max= ranges.elements - 1;
  uint32_t mid= (max + min) / 2;

  while (min != max)
  {
    if (cmp_next(*(optimizer::QuickRange**)dynamic_array_ptr(&ranges, mid)))
    {
      /* current row value > mid->max */
      min= mid + 1;
    }
    else
      max= mid;
    mid= (min + max) / 2;
  }
  res= *(optimizer::QuickRange**)dynamic_array_ptr(&ranges, mid);
  return (! cmp_next(res) && ! cmp_prev(res));
}


int optimizer::QuickRangeSelect::cmp_next(optimizer::QuickRange *range_arg)
{
  if (range_arg->flag & NO_MAX_RANGE)
    return 0;                                   /* key can't be to large */

  KEY_PART *key_part= key_parts;
  uint32_t store_length;

  for (unsigned char *key=range_arg->max_key, *end=key+range_arg->max_length;
       key < end;
       key+= store_length, key_part++)
  {
    int cmp;
    store_length= key_part->store_length;
    if (key_part->null_bit)
    {
      if (*key)
      {
        if (! key_part->field->is_null())
          return 1;
        continue;
      }
      else if (key_part->field->is_null())
        return 0;
      key++;					// Skip null byte
      store_length--;
    }
    if ((cmp= key_part->field->key_cmp(key, key_part->length)) < 0)
      return 0;
    if (cmp > 0)
      return 1;
  }
  return (range_arg->flag & NEAR_MAX) ? 1 : 0;          // Exact match
}


int optimizer::QuickRangeSelect::cmp_prev(optimizer::QuickRange *range_arg)
{
  if (range_arg->flag & NO_MIN_RANGE)
    return 0; /* key can't be to small */

  int cmp= key_cmp(key_part_info,
                   range_arg->min_key,
                   range_arg->min_length);
  if (cmp > 0 || (cmp == 0 && (range_arg->flag & NEAR_MIN) == false))
    return 0;
  return 1; // outside of range
}


void optimizer::QuickRangeSelect::add_info_string(String *str)
{
  KEY *key_info= head->key_info + index;
  str->append(key_info->name);
}


void optimizer::QuickRangeSelect::add_keys_and_lengths(String *key_names,
                                                       String *used_lengths)
{
  char buf[64];
  uint32_t length;
  KEY *key_info= head->key_info + index;
  key_names->append(key_info->name);
  length= internal::int64_t2str(max_used_key_length, buf, 10) - buf;
  used_lengths->append(buf, length);
}


/*
  This is a hack: we inherit from QUICK_SELECT so that we can use the
  get_next() interface, but we have to hold a pointer to the original
  QUICK_SELECT because its data are used all over the place.  What
  should be done is to factor out the data that is needed into a base
  class (QUICK_SELECT), and then have two subclasses (_ASC and _DESC)
  which handle the ranges and implement the get_next() function.  But
  for now, this seems to work right at least.
 */
optimizer::QuickSelectDescending::QuickSelectDescending(optimizer::QuickRangeSelect *q, uint32_t, bool *)
  :
    optimizer::QuickRangeSelect(*q)
{
  optimizer::QuickRange **pr= (optimizer::QuickRange**) ranges.buffer;
  optimizer::QuickRange **end_range= pr + ranges.elements;
  for (; pr != end_range; pr++)
  {
    rev_ranges.push_back(*pr);
  }
  rev_it= rev_ranges.begin();

  /* Remove EQ_RANGE flag for keys that are not using the full key */
  for (vector<optimizer::QuickRange *>::iterator it= rev_ranges.begin();
       it != rev_ranges.end();
       ++it)
  {
    optimizer::QuickRange *r= *it;
    if ((r->flag & EQ_RANGE) &&
        head->key_info[index].key_length != r->max_length)
    {
      r->flag&= ~EQ_RANGE;
    }
  }
  q->dont_free= 1; // Don't free shared mem
  delete q;
}


int optimizer::QuickSelectDescending::get_next()
{
  /* The max key is handled as follows:
   *   - if there is NO_MAX_RANGE, start at the end and move backwards
   *   - if it is an EQ_RANGE, which means that max key covers the entire
   *     key, go directly to the key and read through it (sorting backwards is
   *     same as sorting forwards)
   *   - if it is NEAR_MAX, go to the key or next, step back once, and
   *     move backwards
   *   - otherwise (not NEAR_MAX == include the key), go after the key,
   *     step back once, and move backwards
   */
  for (;;)
  {
    int result;
    if (last_range)
    {						// Already read through key
      result= ((last_range->flag & EQ_RANGE) ?
		           cursor->index_next_same(record, last_range->min_key,
					                             last_range->min_length) :
		           cursor->index_prev(record));
      if (! result)
      {
          if (cmp_prev(*(rev_it - 1)) == 0)
            return 0;
      }
      else if (result != HA_ERR_END_OF_FILE)
        return result;
    }

    if (rev_it == rev_ranges.end())
    {
      return HA_ERR_END_OF_FILE; // All ranges used
    }
    last_range= *rev_it;
    ++rev_it;

    if (last_range->flag & NO_MAX_RANGE)        // Read last record
    {
      int local_error;
      if ((local_error= cursor->index_last(record)))
        return local_error;	// Empty table
      if (cmp_prev(last_range) == 0)
        return 0;
      last_range= 0; // No match; go to next range
      continue;
    }

    if (last_range->flag & EQ_RANGE)
    {
      result = cursor->index_read_map(record,
                                      last_range->max_key,
                                      last_range->max_keypart_map,
                                      HA_READ_KEY_EXACT);
    }
    else
    {
      assert(last_range->flag & NEAR_MAX ||
             range_reads_after_key(last_range));
      result= cursor->index_read_map(record,
                                     last_range->max_key,
                                     last_range->max_keypart_map,
                                     ((last_range->flag & NEAR_MAX) ?
                                      HA_READ_BEFORE_KEY :
                                      HA_READ_PREFIX_LAST_OR_PREV));
    }
    if (result)
    {
      if (result != HA_ERR_KEY_NOT_FOUND && result != HA_ERR_END_OF_FILE)
        return result;
      last_range= 0;                            // Not found, to next range
      continue;
    }
    if (cmp_prev(last_range) == 0)
    {
      if (last_range->flag == (UNIQUE_RANGE | EQ_RANGE))
        last_range= 0;				// Stop searching
      return 0;				// Found key is in range
    }
    last_range= 0;                              // To next range
  }
}


/*
 * true if this range will require using HA_READ_AFTER_KEY
   See comment in get_next() about this
 */
bool optimizer::QuickSelectDescending::range_reads_after_key(optimizer::QuickRange *range_arg)
{
  return ((range_arg->flag & (NO_MAX_RANGE | NEAR_MAX)) ||
	        ! (range_arg->flag & EQ_RANGE) ||
	        head->key_info[index].key_length != range_arg->max_length) ? 1 : 0;
}


} /* namespace drizzled */