~drizzle-trunk/drizzle/development

/* Copyright (C) 2000-2006 MySQL AB

   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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */

/* Describe, check and repair of MyISAM tables */

/*

  About checksum calculation.

  There are two types of checksums. Table checksum and row checksum.

  Row checksum is an additional byte at the end of dynamic length

  records. It must be calculated if the table is configured for them.

  Otherwise they must not be used. The variable

  MYISAM_SHARE::calc_checksum determines if row checksums are used.

  MI_INFO::checksum is used as temporary storage during row handling.

  For parallel repair we must assure that only one thread can use this

  variable. There is no problem on the write side as this is done by one

  thread only. But when checking a record after read this could go

  wrong. But since all threads read through a common read buffer, it is

  sufficient if only one thread checks it.

  Table checksum is an eight byte value in the header of the index file.

  It can be calculated even if row checksums are not used. The variable

  MI_CHECK::glob_crc is calculated over all records.

  MI_SORT_PARAM::calc_checksum determines if this should be done. This

  variable is not part of MI_CHECK because it must be set per thread for

  parallel repair. The global glob_crc must be changed by one thread

  only. And it is sufficient to calculate the checksum once only.

*/

#include "myisamdef.h"

#include <mystrings/m_string.h>

#include <stdarg.h>

#include <mysys/my_getopt.h>

#ifdef HAVE_SYS_VADVISE_H

#include <sys/vadvise.h>

#endif

#ifdef HAVE_SYS_TYPES

#include <sys/types.h>

#endif

#ifdef HAVE_SYS_MMAN_H

#include <sys/mman.h>

#endif

#include <drizzled/util/test.h>

/* Functions defined in this file */

static int check_k_link(MI_CHECK *param, MI_INFO *info,uint32_t nr);

static int chk_index(MI_CHECK *param, MI_INFO *info,MI_KEYDEF *keyinfo,

		     my_off_t page, unsigned char *buff, ha_rows *keys,

		     ha_checksum *key_checksum, uint32_t level);

static uint32_t isam_key_length(MI_INFO *info,MI_KEYDEF *keyinfo);

static ha_checksum calc_checksum(ha_rows count);

static int writekeys(MI_SORT_PARAM *sort_param);

static int sort_one_index(MI_CHECK *param, MI_INFO *info,MI_KEYDEF *keyinfo,

			  my_off_t pagepos, File new_file);

static int sort_key_read(MI_SORT_PARAM *sort_param,void *key);

static int sort_get_next_record(MI_SORT_PARAM *sort_param);

static int sort_key_cmp(MI_SORT_PARAM *sort_param, const void *a,const void *b);

static int sort_key_write(MI_SORT_PARAM *sort_param, const void *a);

static my_off_t get_record_for_key(MI_INFO *info,MI_KEYDEF *keyinfo,

				unsigned char *key);

static int sort_insert_key(MI_SORT_PARAM  *sort_param,

                           register SORT_KEY_BLOCKS *key_block,

			   unsigned char *key, my_off_t prev_block);

static int sort_delete_record(MI_SORT_PARAM *sort_param);

/*static int flush_pending_blocks(MI_CHECK *param);*/

static SORT_KEY_BLOCKS	*alloc_key_blocks(MI_CHECK *param, uint32_t blocks,

					  uint32_t buffer_length);

static ha_checksum mi_byte_checksum(const unsigned char *buf, uint32_t length);

static void set_data_file_type(SORT_INFO *sort_info, MYISAM_SHARE *share);

void myisamchk_init(MI_CHECK *param)

{

  memset(param, 0, sizeof(*param));

  param->opt_follow_links=1;

  param->keys_in_use= ~(uint64_t) 0;

  param->search_after_block=HA_OFFSET_ERROR;

  param->auto_increment_value= 0;

  param->use_buffers=USE_BUFFER_INIT;

  param->read_buffer_length=READ_BUFFER_INIT;

  param->write_buffer_length=READ_BUFFER_INIT;

  param->sort_buffer_length=SORT_BUFFER_INIT;

  param->sort_key_blocks=BUFFERS_WHEN_SORTING;

  param->tmpfile_createflag=O_RDWR | O_TRUNC | O_EXCL;

  param->myf_rw=MYF(MY_NABP | MY_WME | MY_WAIT_IF_FULL);

  param->start_check_pos=0;

  param->max_record_length= INT64_MAX;

  param->key_cache_block_size= KEY_CACHE_BLOCK_SIZE;

  param->stats_method= MI_STATS_METHOD_NULLS_NOT_EQUAL;

}

	/* Check the status flags for the table */

int chk_status(MI_CHECK *param, register MI_INFO *info)

{

  MYISAM_SHARE *share=info->s;

  if (mi_is_crashed_on_repair(info))

    mi_check_print_warning(param,

			   "Table is marked as crashed and last repair failed");

  else if (mi_is_crashed(info))

    mi_check_print_warning(param,

			   "Table is marked as crashed");

  if (share->state.open_count != (uint) (info->s->global_changed ? 1 : 0))

  {

    /* Don't count this as a real warning, as check can correct this ! */

    uint32_t save=param->warning_printed;

    mi_check_print_warning(param,

			   share->state.open_count==1 ?

			   "%d client is using or hasn't closed the table properly" :

			   "%d clients are using or haven't closed the table properly",

			   share->state.open_count);

    /* If this will be fixed by the check, forget the warning */

    if (param->testflag & T_UPDATE_STATE)

      param->warning_printed=save;

  }

  return 0;

}

	/* Check delete links */

int chk_del(MI_CHECK *param, register MI_INFO *info, uint32_t test_flag)

{

  register ha_rows i;

  uint32_t delete_link_length;

  my_off_t empty, next_link, old_link= 0;

  char buff[22],buff2[22];

  param->record_checksum=0;

  delete_link_length=((info->s->options & HA_OPTION_PACK_RECORD) ? 20 :

		      info->s->rec_reflength+1);

  if (!(test_flag & T_SILENT))

    puts("- check record delete-chain");

  next_link=info->s->state.dellink;

  if (info->state->del == 0)

  {

    if (test_flag & T_VERBOSE)

    {

      puts("No recordlinks");

    }

  }

  else

  {

    if (test_flag & T_VERBOSE)

      printf("Recordlinks:    ");

    empty=0;

    for (i= info->state->del ; i > 0L && next_link != HA_OFFSET_ERROR ; i--)

    {

      if (*killed_ptr(param))

        return(1);

      if (test_flag & T_VERBOSE)

	printf(" %9s",llstr(next_link,buff));

      if (next_link >= info->state->data_file_length)

	goto wrong;

      if (my_pread(info->dfile, (unsigned char*) buff,delete_link_length,

		   next_link,MYF(MY_NABP)))

      {

	if (test_flag & T_VERBOSE) puts("");

	mi_check_print_error(param,"Can't read delete-link at filepos: %s",

		    llstr(next_link,buff));

	return(1);

      }

      if (*buff != '\0')

      {

	if (test_flag & T_VERBOSE) puts("");

	mi_check_print_error(param,"Record at pos: %s is not remove-marked",

		    llstr(next_link,buff));

	goto wrong;

      }

      if (info->s->options & HA_OPTION_PACK_RECORD)

      {

	my_off_t prev_link=mi_sizekorr(buff+12);

	if (empty && prev_link != old_link)

	{

	  if (test_flag & T_VERBOSE) puts("");

	  mi_check_print_error(param,"Deleted block at %s doesn't point back at previous delete link",llstr(next_link,buff2));

	  goto wrong;

	}

	old_link=next_link;

	next_link=mi_sizekorr(buff+4);

	empty+=mi_uint3korr(buff+1);

      }

      else

      {

	param->record_checksum+=(ha_checksum) next_link;

	next_link=_mi_rec_pos(info->s,(unsigned char*) buff+1);

	empty+=info->s->base.pack_reclength;

      }

    }

    if (test_flag & T_VERBOSE)

      puts("\n");

    if (empty != info->state->empty)

    {

      mi_check_print_warning(param,

			     "Found %s deleted space in delete link chain. Should be %s",

			     llstr(empty,buff2),

			     llstr(info->state->empty,buff));

    }

    if (next_link != HA_OFFSET_ERROR)

    {

      mi_check_print_error(param,

			   "Found more than the expected %s deleted rows in delete link chain",

			   llstr(info->state->del, buff));

      goto wrong;

    }

    if (i != 0)

    {

      mi_check_print_error(param,

			   "Found %s deleted rows in delete link chain. Should be %s",

			   llstr(info->state->del - i, buff2),

			   llstr(info->state->del, buff));

      goto wrong;

    }

  }

  return(0);

wrong:

  param->testflag|=T_RETRY_WITHOUT_QUICK;

  if (test_flag & T_VERBOSE) puts("");

  mi_check_print_error(param,"record delete-link-chain corrupted");

  return(1);

} /* chk_del */

	/* Check delete links in index file */

static int check_k_link(MI_CHECK *param, register MI_INFO *info, uint32_t nr)

{

  my_off_t next_link;

  uint32_t block_size=(nr+1)*MI_MIN_KEY_BLOCK_LENGTH;

  ha_rows records;

  char llbuff[21], llbuff2[21];

  unsigned char *buff;

  if (param->testflag & T_VERBOSE)

    printf("block_size %4u:", block_size); /* purecov: tested */

  next_link=info->s->state.key_del[nr];

  records= (ha_rows) (info->state->key_file_length / block_size);

  while (next_link != HA_OFFSET_ERROR && records > 0)

  {

    if (*killed_ptr(param))

      return(1);

    if (param->testflag & T_VERBOSE)

      printf("%16s",llstr(next_link,llbuff));

    /* Key blocks must lay within the key file length entirely. */

    if (next_link + block_size > info->state->key_file_length)

    {

      /* purecov: begin tested */

      mi_check_print_error(param, "Invalid key block position: %s  "

                           "key block size: %u  file_length: %s",

                           llstr(next_link, llbuff), block_size,

                           llstr(info->state->key_file_length, llbuff2));

      return(1);

      /* purecov: end */

    }

    /* Key blocks must be aligned at MI_MIN_KEY_BLOCK_LENGTH. */

    if (next_link & (MI_MIN_KEY_BLOCK_LENGTH - 1))

    {

      /* purecov: begin tested */

      mi_check_print_error(param, "Mis-aligned key block: %s  "

                           "minimum key block length: %u",

                           llstr(next_link, llbuff), MI_MIN_KEY_BLOCK_LENGTH);

      return(1);

      /* purecov: end */

    }

    /*

      Read the key block with MI_MIN_KEY_BLOCK_LENGTH to find next link.

      If the key cache block size is smaller than block_size, we can so

      avoid unecessary eviction of cache block.

    */

    if (!(buff=key_cache_read(info->s->key_cache,

                              info->s->kfile, next_link, DFLT_INIT_HITS,

                              (unsigned char*) info->buff, MI_MIN_KEY_BLOCK_LENGTH,

                              MI_MIN_KEY_BLOCK_LENGTH, 1)))

    {

      /* purecov: begin tested */

      mi_check_print_error(param, "key cache read error for block: %s",

			   llstr(next_link,llbuff));

      return(1);

      /* purecov: end */

    }

    next_link=mi_sizekorr(buff);

    records--;

    param->key_file_blocks+=block_size;

  }

  if (param->testflag & T_VERBOSE)

  {

    if (next_link != HA_OFFSET_ERROR)

      printf("%16s\n",llstr(next_link,llbuff));

    else

      puts("");

  }

  return (next_link != HA_OFFSET_ERROR);

} /* check_k_link */

	/* Check sizes of files */

int chk_size(MI_CHECK *param, register MI_INFO *info)

{

  int error=0;

  register my_off_t skr,size;

  char buff[22],buff2[22];

  if (!(param->testflag & T_SILENT)) puts("- check file-size");

  /* The following is needed if called externally (not from myisamchk) */

  flush_key_blocks(info->s->key_cache,

		   info->s->kfile, FLUSH_FORCE_WRITE);

  size= lseek(info->s->kfile, 0, SEEK_END);

  if ((skr=(my_off_t) info->state->key_file_length) != size)

  {

    /* Don't give error if file generated by myisampack */

    if (skr > size && mi_is_any_key_active(info->s->state.key_map))

    {

      error=1;

      mi_check_print_error(param,

			   "Size of indexfile is: %-8s        Should be: %s",

			   llstr(size,buff), llstr(skr,buff2));

    }

    else

      mi_check_print_warning(param,

			     "Size of indexfile is: %-8s      Should be: %s",

			     llstr(size,buff), llstr(skr,buff2));

  }

  if (!(param->testflag & T_VERY_SILENT) &&

      ! (info->s->options & HA_OPTION_COMPRESS_RECORD) &&

      uint64_t2double(info->state->key_file_length) >

      uint64_t2double(info->s->base.margin_key_file_length)*0.9)

    mi_check_print_warning(param,"Keyfile is almost full, %10s of %10s used",

			   llstr(info->state->key_file_length,buff),

			   llstr(info->s->base.max_key_file_length-1,buff));

  size=lseek(info->dfile,0L,SEEK_END);

  skr=(my_off_t) info->state->data_file_length;

  if (info->s->options & HA_OPTION_COMPRESS_RECORD)

    skr+= MEMMAP_EXTRA_MARGIN;

#ifdef USE_RELOC

  if (info->data_file_type == STATIC_RECORD &&

      skr < (my_off_t) info->s->base.reloc*info->s->base.min_pack_length)

    skr=(my_off_t) info->s->base.reloc*info->s->base.min_pack_length;

#endif

  if (skr != size)

  {

    info->state->data_file_length=size;	/* Skip other errors */

    if (skr > size && skr != size + MEMMAP_EXTRA_MARGIN)

    {

      error=1;

      mi_check_print_error(param,"Size of datafile is: %-9s         Should be: %s",

		    llstr(size,buff), llstr(skr,buff2));

      param->testflag|=T_RETRY_WITHOUT_QUICK;

    }

    else

    {

      mi_check_print_warning(param,

			     "Size of datafile is: %-9s       Should be: %s",

			     llstr(size,buff), llstr(skr,buff2));

    }

  }

  if (!(param->testflag & T_VERY_SILENT) &&

      !(info->s->options & HA_OPTION_COMPRESS_RECORD) &&

      uint64_t2double(info->state->data_file_length) >

      (uint64_t2double(info->s->base.max_data_file_length)*0.9))

    mi_check_print_warning(param, "Datafile is almost full, %10s of %10s used",

			   llstr(info->state->data_file_length,buff),

			   llstr(info->s->base.max_data_file_length-1,buff2));

  return(error);

} /* chk_size */

	/* Check keys */

int chk_key(MI_CHECK *param, register MI_INFO *info)

{

  uint32_t key,found_keys=0,full_text_keys=0,result=0;

  ha_rows keys;

  ha_checksum old_record_checksum,init_checksum;

  my_off_t all_keydata,all_totaldata,key_totlength,length;

  ulong   *rec_per_key_part;

  MYISAM_SHARE *share=info->s;

  MI_KEYDEF *keyinfo;

  char buff[22],buff2[22];

  if (!(param->testflag & T_SILENT))

    puts("- check key delete-chain");

  param->key_file_blocks=info->s->base.keystart;

  for (key=0 ; key < info->s->state.header.max_block_size_index ; key++)

    if (check_k_link(param,info,key))

    {

      if (param->testflag & T_VERBOSE) puts("");

      mi_check_print_error(param,"key delete-link-chain corrupted");

      return(-1);

    }

  if (!(param->testflag & T_SILENT)) puts("- check index reference");

  all_keydata=all_totaldata=key_totlength=0;

  old_record_checksum=0;

  init_checksum=param->record_checksum;

  if (!(share->options &

	(HA_OPTION_PACK_RECORD | HA_OPTION_COMPRESS_RECORD)))

    old_record_checksum=calc_checksum(info->state->records+info->state->del-1)*

      share->base.pack_reclength;

  rec_per_key_part= param->rec_per_key_part;

  for (key= 0,keyinfo= &share->keyinfo[0]; key < share->base.keys ;

       rec_per_key_part+=keyinfo->keysegs, key++, keyinfo++)

  {

    param->key_crc[key]=0;

    if (! mi_is_key_active(share->state.key_map, key))

    {

      /* Remember old statistics for key */

      assert(rec_per_key_part >= param->rec_per_key_part);

      memcpy(rec_per_key_part,

	     (share->state.rec_per_key_part +

              (rec_per_key_part - param->rec_per_key_part)),

	     keyinfo->keysegs*sizeof(*rec_per_key_part));

      continue;

    }

    found_keys++;

    param->record_checksum=init_checksum;

    memset(&param->unique_count, 0, sizeof(param->unique_count));

    memset(&param->notnull_count, 0, sizeof(param->notnull_count));

    if ((!(param->testflag & T_SILENT)))

      printf ("- check data record references index: %d\n",key+1);

    if (share->state.key_root[key] == HA_OFFSET_ERROR && (info->state->records == 0))

      goto do_stat;

    if (!_mi_fetch_keypage(info,keyinfo,share->state.key_root[key],

                           DFLT_INIT_HITS,info->buff,0))

    {

      mi_check_print_error(param,"Can't read indexpage from filepos: %s",

		  llstr(share->state.key_root[key],buff));

      if (!(param->testflag & T_INFO))

	return(-1);

      result= -1;

      continue;

    }

    param->key_file_blocks+=keyinfo->block_length;

    keys=0;

    param->keydata=param->totaldata=0;

    param->key_blocks=0;

    param->max_level=0;

    if (chk_index(param,info,keyinfo,share->state.key_root[key],info->buff,

		  &keys, param->key_crc+key,1))

      return(-1);

    if(!(0))

    {

      if (keys != info->state->records)

      {

	mi_check_print_error(param,"Found %s keys of %s",llstr(keys,buff),

		    llstr(info->state->records,buff2));

	if (!(param->testflag & T_INFO))

	return(-1);

	result= -1;

	continue;

      }

      if (found_keys - full_text_keys == 1 &&

	  ((share->options &

	    (HA_OPTION_PACK_RECORD | HA_OPTION_COMPRESS_RECORD)) ||

	   (param->testflag & T_DONT_CHECK_CHECKSUM)))

	old_record_checksum=param->record_checksum;

      else if (old_record_checksum != param->record_checksum)

      {

	if (key)

	  mi_check_print_error(param,"Key %u doesn't point at same records that key 1",

		      key+1);

	else

	  mi_check_print_error(param,"Key 1 doesn't point at all records");

	if (!(param->testflag & T_INFO))

	  return(-1);

	result= -1;

	continue;

      }

    }

    if ((uint) share->base.auto_key -1 == key)

    {

      /* Check that auto_increment key is bigger than max key value */

      uint64_t auto_increment;

      info->lastinx=key;

      _mi_read_key_record(info, 0L, info->rec_buff);

      auto_increment= retrieve_auto_increment(info, info->rec_buff);

      if (auto_increment > info->s->state.auto_increment)

      {

        mi_check_print_warning(param, "Auto-increment value: %s is smaller "

                               "than max used value: %s",

                               llstr(info->s->state.auto_increment,buff2),

                               llstr(auto_increment, buff));

      }

      if (param->testflag & T_AUTO_INC)

      {

        set_if_bigger(info->s->state.auto_increment,

                      auto_increment);

        set_if_bigger(info->s->state.auto_increment,

                      param->auto_increment_value);

      }

      /* Check that there isn't a row with auto_increment = 0 in the table */

      mi_extra(info,HA_EXTRA_KEYREAD,0);

      memset(info->lastkey, 0, keyinfo->seg->length);

      if (!mi_rkey(info, info->rec_buff, key, (const unsigned char*) info->lastkey,

		   (key_part_map)1, HA_READ_KEY_EXACT))

      {

	/* Don't count this as a real warning, as myisamchk can't correct it */

	uint32_t save=param->warning_printed;

        mi_check_print_warning(param, "Found row where the auto_increment "

                               "column has the value 0");

	param->warning_printed=save;

      }

      mi_extra(info,HA_EXTRA_NO_KEYREAD,0);

    }

    length=(my_off_t) isam_key_length(info,keyinfo)*keys + param->key_blocks*2;

    if (param->testflag & T_INFO && param->totaldata != 0L && keys != 0L)

      printf("Key: %2d:  Keyblocks used: %3d%%  Packed: %4d%%  Max levels: %2d\n",

	     key+1,

	     (int) (my_off_t2double(param->keydata)*100.0/my_off_t2double(param->totaldata)),

	     (int) ((my_off_t2double(length) - my_off_t2double(param->keydata))*100.0/

		    my_off_t2double(length)),

	     param->max_level);

    all_keydata+=param->keydata; all_totaldata+=param->totaldata; key_totlength+=length;

do_stat:

    if (param->testflag & T_STATISTICS)

      update_key_parts(keyinfo, rec_per_key_part, param->unique_count,

                       param->stats_method == MI_STATS_METHOD_IGNORE_NULLS?

                       param->notnull_count: NULL,

                       (uint64_t)info->state->records);

  }

  if (param->testflag & T_INFO)

  {

    if (all_totaldata != 0L && found_keys > 0)

      printf("Total:    Keyblocks used: %3d%%  Packed: %4d%%\n\n",

	     (int) (my_off_t2double(all_keydata)*100.0/

		    my_off_t2double(all_totaldata)),

	     (int) ((my_off_t2double(key_totlength) -

		     my_off_t2double(all_keydata))*100.0/

		     my_off_t2double(key_totlength)));

    else if (all_totaldata != 0L && mi_is_any_key_active(share->state.key_map))

      puts("");

  }

  if (param->key_file_blocks != info->state->key_file_length &&

      param->keys_in_use != ~(uint64_t) 0)

    mi_check_print_warning(param, "Some data are unreferenced in keyfile");

  if (found_keys != full_text_keys)

    param->record_checksum=old_record_checksum-init_checksum;	/* Remove delete links */

  else

    param->record_checksum=0;

  return(result);

} /* chk_key */

static int chk_index_down(MI_CHECK *param, MI_INFO *info, MI_KEYDEF *keyinfo,

                     my_off_t page, unsigned char *buff, ha_rows *keys,

                     ha_checksum *key_checksum, uint32_t level)

{

  char llbuff[22],llbuff2[22];

  /* Key blocks must lay within the key file length entirely. */

  if (page + keyinfo->block_length > info->state->key_file_length)

  {

    /* purecov: begin tested */

    /* Give it a chance to fit in the real file size. */

    my_off_t max_length= lseek(info->s->kfile, 0, SEEK_END);

    mi_check_print_error(param, "Invalid key block position: %s  "

                         "key block size: %u  file_length: %s",

                         llstr(page, llbuff), keyinfo->block_length,

                         llstr(info->state->key_file_length, llbuff2));

    if (page + keyinfo->block_length > max_length)

      goto err;

    /* Fix the remebered key file length. */

    info->state->key_file_length= (max_length &

                                   ~ (my_off_t) (keyinfo->block_length - 1));

    /* purecov: end */

  }

  /* Key blocks must be aligned at MI_MIN_KEY_BLOCK_LENGTH. */

  if (page & (MI_MIN_KEY_BLOCK_LENGTH - 1))

  {

    /* purecov: begin tested */

    mi_check_print_error(param, "Mis-aligned key block: %s  "

                         "minimum key block length: %u",

                         llstr(page, llbuff), MI_MIN_KEY_BLOCK_LENGTH);

    goto err;

    /* purecov: end */

  }

  if (!_mi_fetch_keypage(info,keyinfo,page, DFLT_INIT_HITS,buff,0))

  {

    mi_check_print_error(param,"Can't read key from filepos: %s",

        llstr(page,llbuff));

    goto err;

  }

  param->key_file_blocks+=keyinfo->block_length;

  if (chk_index(param,info,keyinfo,page,buff,keys,key_checksum,level))

    goto err;

  return(0);

  /* purecov: begin tested */

err:

  return(1);

  /* purecov: end */

}

/*

  "Ignore NULLs" statistics collection method: process first index tuple.

  SYNOPSIS

    mi_collect_stats_nonulls_first()

      keyseg   IN     Array of key part descriptions

      notnull  INOUT  Array, notnull[i] = (number of {keypart1...keypart_i}

                                           tuples that don't contain NULLs)

      key      IN     Key values tuple

  DESCRIPTION

    Process the first index tuple - find out which prefix tuples don't

    contain NULLs, and update the array of notnull counters accordingly.

*/

static

void mi_collect_stats_nonulls_first(HA_KEYSEG *keyseg, uint64_t *notnull,

                                    unsigned char *key)

{

  uint32_t first_null, kp;

  first_null= ha_find_null(keyseg, key) - keyseg;

  /*

    All prefix tuples that don't include keypart_{first_null} are not-null

    tuples (and all others aren't), increment counters for them.

  */

  for (kp= 0; kp < first_null; kp++)

    notnull[kp]++;

}