~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/optimizer/range.h>

#include <drizzled/optimizer/range_param.h>

#include <drizzled/optimizer/sel_arg.h>

#include <drizzled/util/test.h>

namespace drizzled

{

/* Functions to fix up the tree after insert and delete */

static void left_rotate(optimizer::SEL_ARG **root, optimizer::SEL_ARG *leaf)

{

  optimizer::SEL_ARG *y= leaf->right;

  leaf->right= y->left;

  if (y->left != &optimizer::null_element)

    y->left->parent= leaf;

  if (! (y->parent=leaf->parent))

    *root= y;

  else

    *leaf->parent_ptr()= y;

  y->left= leaf;

  leaf->parent= y;

}

static void right_rotate(optimizer::SEL_ARG **root, optimizer::SEL_ARG *leaf)

{

  optimizer::SEL_ARG *y= leaf->left;

  leaf->left= y->right;

  if (y->right != &optimizer::null_element)

    y->right->parent= leaf;

  if (! (y->parent=leaf->parent))

    *root= y;

  else

    *leaf->parent_ptr()= y;

  y->right= leaf;

  leaf->parent= y;

}

/* Get overlapping range */

optimizer::SEL_ARG *optimizer::SEL_ARG::clone_and(optimizer::SEL_ARG *arg)

{

  unsigned char *new_min= NULL;

  unsigned char *new_max= NULL;

  uint8_t flag_min= 0;

  uint8_t flag_max= 0;

  if (cmp_min_to_min(arg) >= 0)

  {

    new_min= min_value;

    flag_min= min_flag;

  }

  else

  {

    new_min=arg->min_value; flag_min=arg->min_flag;

  }

  if (cmp_max_to_max(arg) <= 0)

  {

    new_max= max_value;

    flag_max= max_flag;

  }

  else

  {

    new_max= arg->max_value;

    flag_max= arg->max_flag;

  }

  return (new SEL_ARG(field,

                      part,

                      new_min,

                      new_max,

                      flag_min,

                      flag_max,

                      test(maybe_flag && arg->maybe_flag)));

}

/* min <= X , arg->min */

optimizer::SEL_ARG *optimizer::SEL_ARG::clone_first(optimizer::SEL_ARG *arg)

{

  return (new SEL_ARG(field,part,

                      min_value,

                      arg->min_value,

                      min_flag,

                      arg->min_flag & NEAR_MIN ? 0 : NEAR_MAX,

		                                     maybe_flag | arg->maybe_flag));

}

/* min <= X <= key_max */

optimizer::SEL_ARG *optimizer::SEL_ARG::clone_last(optimizer::SEL_ARG *arg)

{

  return (new SEL_ARG(field,

                      part,

                      min_value,

                      arg->max_value,

                      min_flag,

                      arg->max_flag,

                      maybe_flag | arg->maybe_flag));

}

/* Get overlapping range */

bool optimizer::SEL_ARG::copy_min(optimizer::SEL_ARG *arg)

{

  if (cmp_min_to_min(arg) > 0)

  {

    min_value= arg->min_value;

    min_flag=arg->min_flag;

    if ((max_flag & (NO_MAX_RANGE | NO_MIN_RANGE)) ==

        (NO_MAX_RANGE | NO_MIN_RANGE))

    {

      return 1;	// Full range

    }

  }

  maybe_flag|= arg->maybe_flag;

  return 0;

}

/* Get overlapping range */

bool optimizer::SEL_ARG::copy_max(optimizer::SEL_ARG *arg)

{

  if (cmp_max_to_max(arg) <= 0)

  {

    max_value= arg->max_value;

    max_flag= arg->max_flag;

    if ((max_flag & (NO_MAX_RANGE | NO_MIN_RANGE)) ==

        (NO_MAX_RANGE | NO_MIN_RANGE))

    {

      return 1;	// Full range

    }

  }

  maybe_flag|= arg->maybe_flag;

  return 0;

}

void optimizer::SEL_ARG::copy_min_to_min(optimizer::SEL_ARG *arg)

{

  min_value= arg->min_value;

  min_flag= arg->min_flag;

}

void optimizer::SEL_ARG::copy_min_to_max(optimizer::SEL_ARG *arg)

{

  max_value= arg->min_value;

  max_flag= arg->min_flag & NEAR_MIN ? 0 : NEAR_MAX;

}

void optimizer::SEL_ARG::copy_max_to_min(optimizer::SEL_ARG *arg)

{

  min_value= arg->max_value;

  min_flag= arg->max_flag & NEAR_MAX ? 0 : NEAR_MIN;

}

int optimizer::SEL_ARG::store_min(uint32_t length, unsigned char **min_key, uint32_t min_key_flag)

{

  /* "(kp1 > c1) AND (kp2 OP c2) AND ..." -> (kp1 > c1) */

  if ((! (min_flag & NO_MIN_RANGE) &&

      ! (min_key_flag & (NO_MIN_RANGE | NEAR_MIN))))

  {

    if (maybe_null && *min_value)

    {

      **min_key= 1;

      memset(*min_key+1, 0, length-1);

    }

    else

    {

      memcpy(*min_key,min_value,length);

    }

    (*min_key)+= length;

    return 1;

  }

  return 0;

}

int optimizer::SEL_ARG::store_max(uint32_t length, unsigned char **max_key, uint32_t max_key_flag)

{

  if (! (max_flag & NO_MAX_RANGE) &&

      ! (max_key_flag & (NO_MAX_RANGE | NEAR_MAX)))

  {

    if (maybe_null && *max_value)

    {

      **max_key= 1;

      memset(*max_key + 1, 0, length-1);

    }

    else

    {

      memcpy(*max_key,max_value,length);

    }

    (*max_key)+= length;

    return 1;

  }

  return 0;

}

int optimizer::SEL_ARG::store_min_key(KEY_PART *key, unsigned char **range_key, uint32_t *range_key_flag)

{

  optimizer::SEL_ARG *key_tree= first();

  uint32_t res= key_tree->store_min(key[key_tree->part].store_length,

                                    range_key,

                                    *range_key_flag);

  *range_key_flag|= key_tree->min_flag;

  if (key_tree->next_key_part &&

      key_tree->next_key_part->part == key_tree->part+1 &&

      ! (*range_key_flag & (NO_MIN_RANGE | NEAR_MIN)) &&

      key_tree->next_key_part->type == optimizer::SEL_ARG::KEY_RANGE)

  {

    res+= key_tree->next_key_part->store_min_key(key,

                                                 range_key,

                                                 range_key_flag);

  }

  return res;

}

int optimizer::SEL_ARG::store_max_key(KEY_PART *key, unsigned char **range_key, uint32_t *range_key_flag)

{

  SEL_ARG *key_tree= last();

  uint32_t res= key_tree->store_max(key[key_tree->part].store_length,

                                    range_key,

                                    *range_key_flag);

  (*range_key_flag)|= key_tree->max_flag;

  if (key_tree->next_key_part &&

      key_tree->next_key_part->part == key_tree->part+1 &&

      ! (*range_key_flag & (NO_MAX_RANGE | NEAR_MAX)) &&

      key_tree->next_key_part->type == SEL_ARG::KEY_RANGE)

    res+= key_tree->next_key_part->store_max_key(key,

                                                 range_key,

                                                 range_key_flag);

  return res;

}

optimizer::SEL_ARG::SEL_ARG(optimizer::SEL_ARG &arg)

  :

    memory::SqlAlloc()

{

  type= arg.type;

  min_flag= arg.min_flag;

  max_flag= arg.max_flag;

  maybe_flag= arg.maybe_flag;

  maybe_null= arg.maybe_null;

  part= arg.part;

  field= arg.field;

  min_value= arg.min_value;

  max_value= arg.max_value;

  next_key_part= arg.next_key_part;

  use_count=1;

  elements=1;

}

void optimizer::SEL_ARG::make_root()

{

  left= right= &optimizer::null_element;

  color= BLACK;

  next= prev =0;

  use_count= 0;

  elements= 1;

}

optimizer::SEL_ARG::SEL_ARG(Field *f,

                            const unsigned char *min_value_arg,

                            const unsigned char *max_value_arg)

  :

    min_flag(0),

    max_flag(0),

    maybe_flag(0),

    maybe_null(f->real_maybe_null()),

    elements(1),

    use_count(1),

    field(f),

    min_value((unsigned char*) min_value_arg),

    max_value((unsigned char*) max_value_arg),

    next(0),

    prev(0),

    next_key_part(0),

    color(BLACK),

    type(KEY_RANGE)

{

  left= right= &optimizer::null_element;

}

optimizer::SEL_ARG::SEL_ARG(Field *field_,

                            uint8_t part_,

                            unsigned char *min_value_,

                            unsigned char *max_value_,

		            uint8_t min_flag_,

                            uint8_t max_flag_,

                            uint8_t maybe_flag_)

  :

    min_flag(min_flag_),

    max_flag(max_flag_),

    maybe_flag(maybe_flag_),

    part(part_),

    maybe_null(field_->real_maybe_null()),

    elements(1),

    use_count(1),

    field(field_),

    min_value(min_value_),

    max_value(max_value_),

    next(0),

    prev(0),

    next_key_part(0),

    color(BLACK),

    type(KEY_RANGE)

{

  left= right= &optimizer::null_element;

}

optimizer::SEL_ARG *optimizer::SEL_ARG::clone(RangeParameter *param,

                                              optimizer::SEL_ARG *new_parent,

                                              optimizer::SEL_ARG **next_arg)

{

  optimizer::SEL_ARG *tmp= NULL;

  /* Bail out if we have already generated too many SEL_ARGs */

  if (++param->alloced_sel_args > MAX_SEL_ARGS)

    return 0;

  if (type != KEY_RANGE)

  {

    if (! (tmp= new (param->mem_root) optimizer::SEL_ARG(type)))

      return 0;	// out of memory

    tmp->prev= *next_arg; // Link into next/prev chain

    (*next_arg)->next= tmp;

    (*next_arg)= tmp;

  }

  else

  {

    if (! (tmp= new (param->mem_root) optimizer::SEL_ARG(field,

                                                         part,

                                                         min_value,

                                                         max_value,

                                                         min_flag,

                                                         max_flag,

                                                         maybe_flag)))

      return 0; // OOM

    tmp->parent= new_parent;

    tmp->next_key_part= next_key_part;

    if (left != &optimizer::null_element)

      if (! (tmp->left= left->clone(param, tmp, next_arg)))

	return 0; // OOM

    tmp->prev= *next_arg; // Link into next/prev chain

    (*next_arg)->next= tmp;

    (*next_arg)= tmp;

    if (right != &optimizer::null_element)

      if (! (tmp->right= right->clone(param, tmp, next_arg)))

	return 0; // OOM

  }

  increment_use_count(1);

  tmp->color= color;

  tmp->elements= this->elements;

  return tmp;

}

optimizer::SEL_ARG *optimizer::SEL_ARG::first()

{

  optimizer::SEL_ARG *next_arg= this;

  if (! next_arg->left)

    return 0; // MAYBE_KEY

  while (next_arg->left != &optimizer::null_element)

    next_arg= next_arg->left;

  return next_arg;

}

optimizer::SEL_ARG *optimizer::SEL_ARG::last()

{

  SEL_ARG *next_arg= this;

  if (! next_arg->right)

    return 0; // MAYBE_KEY

  while (next_arg->right != &optimizer::null_element)

    next_arg=next_arg->right;

  return next_arg;

}

optimizer::SEL_ARG *optimizer::SEL_ARG::clone_tree(RangeParameter *param)

{

  optimizer::SEL_ARG tmp_link;

  optimizer::SEL_ARG*next_arg= NULL;

  optimizer::SEL_ARG *root= NULL;

  next_arg= &tmp_link;

  if (! (root= clone(param, (SEL_ARG *) 0, &next_arg)))

    return 0;

  next_arg->next= 0; // Fix last link

  tmp_link.next->prev= 0; // Fix first link

  if (root) // If not OOM

    root->use_count= 0;

  return root;

}

optimizer::SEL_ARG *

optimizer::SEL_ARG::insert(optimizer::SEL_ARG *key)

{

  optimizer::SEL_ARG *element= NULL;

  optimizer::SEL_ARG **par= NULL;

  optimizer::SEL_ARG *last_element= NULL;

  for (element= this; element != &optimizer::null_element; )

  {

    last_element= element;

    if (key->cmp_min_to_min(element) > 0)

    {

      par= &element->right; element= element->right;

    }

    else

    {

      par= &element->left; element= element->left;

    }

  }

  *par= key;

  key->parent= last_element;

	/* Link in list */

  if (par == &last_element->left)

  {

    key->next= last_element;

    if ((key->prev=last_element->prev))

      key->prev->next= key;

    last_element->prev= key;

  }

  else

  {

    if ((key->next= last_element->next))

      key->next->prev= key;

    key->prev= last_element;

    last_element->next= key;

  }

  key->left= key->right= &optimizer::null_element;

  optimizer::SEL_ARG *root= rb_insert(key); // rebalance tree

  root->use_count= this->use_count;		// copy root info

  root->elements= this->elements+1;

  root->maybe_flag= this->maybe_flag;

  return root;

}

/*

** Find best key with min <= given key

** Because the call context this should never return 0 to get_range

*/

optimizer::SEL_ARG *

optimizer::SEL_ARG::find_range(optimizer::SEL_ARG *key)

{

  optimizer::SEL_ARG *element= this;

  optimizer::SEL_ARG *found= NULL;

  for (;;)

  {

    if (element == &optimizer::null_element)

      return found;

    int cmp= element->cmp_min_to_min(key);

    if (cmp == 0)

      return element;

    if (cmp < 0)

    {

      found= element;

      element= element->right;

    }

    else

      element= element->left;

  }

}

/*

  Remove a element from the tree

  SYNOPSIS

    tree_delete()

    key		Key that is to be deleted from tree (this)

  NOTE

    This also frees all sub trees that is used by the element

  RETURN

    root of new tree (with key deleted)

*/

optimizer::SEL_ARG *

optimizer::SEL_ARG::tree_delete(optimizer::SEL_ARG *key)

{

  enum leaf_color remove_color;

  optimizer::SEL_ARG *root= NULL;

  optimizer::SEL_ARG *nod= NULL;

  optimizer::SEL_ARG **par= NULL;

  optimizer::SEL_ARG *fix_par= NULL;

  root= this;

  this->parent= 0;

  /* Unlink from list */

  if (key->prev)

    key->prev->next= key->next;

  if (key->next)

    key->next->prev= key->prev;

  key->increment_use_count(-1);

  if (! key->parent)

    par= &root;

  else

    par= key->parent_ptr();

  if (key->left == &optimizer::null_element)

  {

    *par= nod= key->right;

    fix_par= key->parent;

    if (nod != &optimizer::null_element)

      nod->parent= fix_par;

    remove_color= key->color;

  }

  else if (key->right == &optimizer::null_element)

  {

    *par= nod= key->left;

    nod->parent= fix_par= key->parent;

    remove_color= key->color;

  }

  else

  {

    optimizer::SEL_ARG *tmp= key->next; // next bigger key (exist!)

    nod= *tmp->parent_ptr()= tmp->right;	// unlink tmp from tree

    fix_par= tmp->parent;

    if (nod != &optimizer::null_element)

      nod->parent= fix_par;

    remove_color= tmp->color;

    tmp->parent= key->parent;			// Move node in place of key

    (tmp->left= key->left)->parent= tmp;

    if ((tmp->right=key->right) != &optimizer::null_element)

      tmp->right->parent= tmp;

    tmp->color= key->color;

    *par= tmp;

    if (fix_par == key)				// key->right == key->next

      fix_par= tmp;				// new parent of nod

  }

  if (root == &optimizer::null_element)

    return 0;				// Maybe root later

  if (remove_color == BLACK)

    root= rb_delete_fixup(root, nod, fix_par);

  root->use_count= this->use_count;		// Fix root counters

  root->elements= this->elements-1;

  root->maybe_flag= this->maybe_flag;

  return root;

}

optimizer::SEL_ARG *

optimizer::SEL_ARG::rb_insert(optimizer::SEL_ARG *leaf)

{

  optimizer::SEL_ARG *y= NULL;

  optimizer::SEL_ARG *par= NULL;

  optimizer::SEL_ARG *par2= NULL;

  optimizer::SEL_ARG *root= NULL;

  root= this;

  root->parent= 0;

  leaf->color= RED;

  while (leaf != root && (par= leaf->parent)->color == RED)

  {					// This can't be root or 1 level under

    if (par == (par2= leaf->parent->parent)->left)

    {

      y= par2->right;

      if (y->color == RED)

      {

        par->color= BLACK;

        y->color= BLACK;

        leaf= par2;

        leaf->color= RED;		/* And the loop continues */

      }

      else

      {

        if (leaf == par->right)

        {

          left_rotate(&root,leaf->parent);

          par= leaf; /* leaf is now parent to old leaf */

        }

        par->color= BLACK;

        par2->color= RED;

        right_rotate(&root, par2);

        break;

      }

    }

    else

    {

      y= par2->left;

      if (y->color == RED)

      {

        par->color= BLACK;

        y->color= BLACK;

        leaf= par2;

        leaf->color= RED;		/* And the loop continues */

      }

      else

      {

        if (leaf == par->left)

        {

          right_rotate(&root,par);

          par= leaf;

        }

        par->color= BLACK;

        par2->color= RED;

        left_rotate(&root, par2);

        break;

      }

    }

  }

  root->color= BLACK;

  return root;

}

optimizer::SEL_ARG *optimizer::rb_delete_fixup(optimizer::SEL_ARG *root,

                                               optimizer::SEL_ARG *key,

                                               optimizer::SEL_ARG *par)

{

  optimizer::SEL_ARG *x= NULL;

  optimizer::SEL_ARG *w= NULL;

  root->parent= 0;

  x= key;

  while (x != root && x->color == optimizer::SEL_ARG::BLACK)

  {

    if (x == par->left)

    {

      w= par->right;

      if (w->color == optimizer::SEL_ARG::RED)

      {

        w->color= optimizer::SEL_ARG::BLACK;

        par->color= optimizer::SEL_ARG::RED;

        left_rotate(&root, par);

        w= par->right;

      }

      if (w->left->color == optimizer::SEL_ARG::BLACK &&

          w->right->color == optimizer::SEL_ARG::BLACK)

      {

        w->color= optimizer::SEL_ARG::RED;

        x= par;

      }

      else

      {

        if (w->right->color == optimizer::SEL_ARG::BLACK)

        {

          w->left->color= optimizer::SEL_ARG::BLACK;

          w->color= optimizer::SEL_ARG::RED;

          right_rotate(&root, w);

          w= par->right;

        }

        w->color= par->color;

        par->color= optimizer::SEL_ARG::BLACK;

        w->right->color= optimizer::SEL_ARG::BLACK;

        left_rotate(&root, par);

        x= root;

        break;

      }

    }

    else

    {

      w= par->left;

      if (w->color == optimizer::SEL_ARG::RED)

      {

        w->color= optimizer::SEL_ARG::BLACK;

        par->color= optimizer::SEL_ARG::RED;

        right_rotate(&root, par);

        w= par->left;

      }

      if (w->right->color == optimizer::SEL_ARG::BLACK &&

          w->left->color == optimizer::SEL_ARG::BLACK)

      {

        w->color= optimizer::SEL_ARG::RED;

        x= par;

      }

      else

      {

        if (w->left->color == SEL_ARG::BLACK)

        {

          w->right->color= optimizer::SEL_ARG::BLACK;

          w->color= optimizer::SEL_ARG::RED;

          left_rotate(&root, w);

          w= par->left;

        }

        w->color= par->color;

        par->color= optimizer::SEL_ARG::BLACK;

        w->left->color= optimizer::SEL_ARG::BLACK;

        right_rotate(&root, par);

        x= root;