~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 */

/* Pack MyISAM file */

#ifndef USE_MY_FUNC

#define USE_MY_FUNC			/* We need at least my_malloc */

#endif

#include "myisamdef.h"

#include <queues.h>

#include <my_tree.h>

#include "mysys_err.h"

#ifndef __GNU_LIBRARY__

#define __GNU_LIBRARY__			/* Skip warnings in getopt.h */

#endif

#include <my_getopt.h>

#include <assert.h>

#if SIZEOF_LONG_LONG > 4

#define BITS_SAVED 64

#else

#define BITS_SAVED 32

#endif

#define IS_OFFSET ((uint) 32768)	/* Bit if offset or char in tree */

#define HEAD_LENGTH	32

#define ALLOWED_JOIN_DIFF	256	/* Diff allowed to join trees */

#define DATA_TMP_EXT		".TMD"

#define OLD_EXT			".OLD"

#define WRITE_COUNT		MY_HOW_OFTEN_TO_WRITE

struct st_file_buffer {

  File file;

  uchar *buffer,*pos,*end;

  my_off_t pos_in_file;

  int bits;

  uint64_t bitbucket;

};

struct st_huff_tree;

struct st_huff_element;

typedef struct st_huff_counts {

  uint	field_length,max_zero_fill;

  uint	pack_type;

  uint	max_end_space,max_pre_space,length_bits,min_space;

  ulong max_length;

  enum en_fieldtype field_type;

  struct st_huff_tree *tree;		/* Tree for field */

  my_off_t counts[256];

  my_off_t end_space[8];

  my_off_t pre_space[8];

  my_off_t tot_end_space,tot_pre_space,zero_fields,empty_fields,bytes_packed;

  TREE int_tree;        /* Tree for detecting distinct column values. */

  uchar *tree_buff;      /* Column values, 'field_length' each. */

  uchar *tree_pos;       /* Points to end of column values in 'tree_buff'. */

} HUFF_COUNTS;

typedef struct st_huff_element HUFF_ELEMENT;

/*

  WARNING: It is crucial for the optimizations in calc_packed_length()

  that 'count' is the first element of 'HUFF_ELEMENT'.

*/

struct st_huff_element {

  my_off_t count;

  union un_element {

    struct st_nod {

      HUFF_ELEMENT *left,*right;

    } nod;

    struct st_leaf {

      HUFF_ELEMENT *null;

      uint	element_nr;		/* Number of element */

    } leaf;

  } a;

};

typedef struct st_huff_tree {

  HUFF_ELEMENT *root,*element_buffer;

  HUFF_COUNTS *counts;

  uint tree_number;

  uint elements;

  my_off_t bytes_packed;

  uint tree_pack_length;

  uint min_chr,max_chr,char_bits,offset_bits,max_offset,height;

  uint64_t *code;

  uchar *code_len;

} HUFF_TREE;

typedef struct st_isam_mrg {

  MI_INFO **file,**current,**end;

  uint free_file;

  uint count;

  uint	min_pack_length;		/* Theese is used by packed data */

  uint	max_pack_length;

  uint	ref_length;

  uint	max_blob_length;

  my_off_t records;

  /* true if at least one source file has at least one disabled index */

  my_bool src_file_has_indexes_disabled;

} PACK_MRG_INFO;

extern int main(int argc,char * *argv);

static void get_options(int *argc,char ***argv);

static MI_INFO *open_isam_file(char *name,int mode);

static my_bool open_isam_files(PACK_MRG_INFO *mrg,char **names,uint count);

static int compress(PACK_MRG_INFO *file,char *join_name);

static HUFF_COUNTS *init_huff_count(MI_INFO *info,my_off_t records);

static void free_counts_and_tree_and_queue(HUFF_TREE *huff_trees,

					   uint trees,

					   HUFF_COUNTS *huff_counts,

					   uint fields);

static int compare_tree(void* cmp_arg __attribute__((unused)),

			const uchar *s,const uchar *t);

static int get_statistic(PACK_MRG_INFO *mrg,HUFF_COUNTS *huff_counts);

static void check_counts(HUFF_COUNTS *huff_counts,uint trees,

			 my_off_t records);

static int test_space_compress(HUFF_COUNTS *huff_counts,my_off_t records,

			       uint max_space_length,my_off_t *space_counts,

			       my_off_t tot_space_count,

			       enum en_fieldtype field_type);

static HUFF_TREE* make_huff_trees(HUFF_COUNTS *huff_counts,uint trees);

static int make_huff_tree(HUFF_TREE *tree,HUFF_COUNTS *huff_counts);

static int compare_huff_elements(void *not_used, uchar *a,uchar *b);

static int save_counts_in_queue(uchar *key,element_count count,

				    HUFF_TREE *tree);

static my_off_t calc_packed_length(HUFF_COUNTS *huff_counts,uint flag);

static uint join_same_trees(HUFF_COUNTS *huff_counts,uint trees);

static int make_huff_decode_table(HUFF_TREE *huff_tree,uint trees);

static void make_traverse_code_tree(HUFF_TREE *huff_tree,

				    HUFF_ELEMENT *element,uint size,

				    uint64_t code);

static int write_header(PACK_MRG_INFO *isam_file, uint header_length,uint trees,

			my_off_t tot_elements,my_off_t filelength);

static void write_field_info(HUFF_COUNTS *counts, uint fields,uint trees);

static my_off_t write_huff_tree(HUFF_TREE *huff_tree,uint trees);

static uint *make_offset_code_tree(HUFF_TREE *huff_tree,

				       HUFF_ELEMENT *element,

				       uint *offset);

static uint max_bit(uint value);

static int compress_isam_file(PACK_MRG_INFO *file,HUFF_COUNTS *huff_counts);

static char *make_new_name(char *new_name,char *old_name);

static char *make_old_name(char *new_name,char *old_name);

static void init_file_buffer(File file,bool read_buffer);

static int flush_buffer(ulong neaded_length);

static void end_file_buffer(void);

static void write_bits(uint64_t value, uint bits);

static void flush_bits(void);

static int save_state(MI_INFO *isam_file,PACK_MRG_INFO *mrg,my_off_t new_length,

		      ha_checksum crc);

static int save_state_mrg(File file,PACK_MRG_INFO *isam_file,my_off_t new_length,

			  ha_checksum crc);

static int mrg_close(PACK_MRG_INFO *mrg);

static int mrg_rrnd(PACK_MRG_INFO *info,uchar *buf);

static void mrg_reset(PACK_MRG_INFO *mrg);

static int error_on_write=0,test_only=0,verbose=0,silent=0,

	   write_loop=0,force_pack=0, isamchk_neaded=0;

static int tmpfile_createflag=O_RDWR | O_TRUNC | O_EXCL;

static my_bool backup, opt_wait;

/*

  tree_buff_length is somewhat arbitrary. The bigger it is the better

  the chance to win in terms of compression factor. On the other hand,

  this table becomes part of the compressed file header. And its length

  is coded with 16 bits in the header. Hence the limit is 2**16 - 1.

*/

static uint tree_buff_length= 65536 - MALLOC_OVERHEAD;

static char tmp_dir[FN_REFLEN]={0},*join_table;

static my_off_t intervall_length;

static ha_checksum glob_crc;

static struct st_file_buffer file_buffer;

static QUEUE queue;

static HUFF_COUNTS *global_count;

static char zero_string[]={0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0};

static const char *load_default_groups[]= { "myisampack",0 };

	/* The main program */

int main(int argc, char **argv)

{

  int error,ok;

  PACK_MRG_INFO merge;

  char **default_argv;

  MY_INIT(argv[0]);

  load_defaults("my",load_default_groups,&argc,&argv);

  default_argv= argv;

  get_options(&argc,&argv);

  error=ok=isamchk_neaded=0;

  if (join_table)

  {						/* Join files into one */

    if (open_isam_files(&merge,argv,(uint) argc) ||

	compress(&merge,join_table))

      error=1;

  }

  else while (argc--)

  {

    MI_INFO *isam_file;

    if (!(isam_file=open_isam_file(*argv++,O_RDWR)))

      error=1;

    else

    {

      merge.file= &isam_file;

      merge.current=0;

      merge.free_file=0;

      merge.count=1;

      if (compress(&merge,0))

	error=1;

      else

	ok=1;

    }

  }

  if (ok && isamchk_neaded && !silent)

    puts("Remember to run myisamchk -rq on compressed tables");

  VOID(fflush(stdout));

  VOID(fflush(stderr));

  free_defaults(default_argv);

  my_end(verbose ? MY_CHECK_ERROR | MY_GIVE_INFO : MY_CHECK_ERROR);

  exit(error ? 2 : 0);

#ifndef _lint

  return 0;					/* No compiler warning */

#endif

}

enum options_mp {OPT_CHARSETS_DIR_MP=256, OPT_AUTO_CLOSE};

static struct my_option my_long_options[] =

{

  {"backup", 'b', "Make a backup of the table as table_name.OLD.",

   (char**) &backup, (char**) &backup, 0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},

  {"character-sets-dir", OPT_CHARSETS_DIR_MP,

   "Directory where character sets are.", (char**) &charsets_dir,

   (char**) &charsets_dir, 0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},

  {"debug", '#', "Output debug log. Often this is 'd:t:o,filename'.",

   0, 0, 0, GET_STR, OPT_ARG, 0, 0, 0, 0, 0, 0},

  {"force", 'f',

   "Force packing of table even if it gets bigger or if tempfile exists.",

   0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},

  {"join", 'j',

   "Join all given tables into 'new_table_name'. All tables MUST have identical layouts.",

   (char**) &join_table, (char**) &join_table, 0, GET_STR, REQUIRED_ARG, 0, 0, 0,

   0, 0, 0},

  {"help", '?', "Display this help and exit.",

   0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},

  {"silent", 's', "Be more silent.",

   0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},

  {"tmpdir", 'T', "Use temporary directory to store temporary table.",

   0, 0, 0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},

  {"test", 't', "Don't pack table, only test packing it.",

   0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},

  {"verbose", 'v', "Write info about progress and packing result. Use many -v for more verbosity!",

   0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},

  {"version", 'V', "Output version information and exit.",

   0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},

  {"wait", 'w', "Wait and retry if table is in use.", (char**) &opt_wait,

   (char**) &opt_wait, 0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},

  { 0, 0, 0, 0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0}

};

static void print_version(void)

{

  VOID(printf("%s Ver 1.23 for %s on %s\n",

              my_progname, SYSTEM_TYPE, MACHINE_TYPE));

}

static void usage(void)

{

  print_version();

  puts("Copyright (C) 2002 MySQL AB");

  puts("This software comes with ABSOLUTELY NO WARRANTY. This is free software,");

  puts("and you are welcome to modify and redistribute it under the GPL license\n");

  puts("Pack a MyISAM-table to take much less space.");

  puts("Keys are not updated, you must run myisamchk -rq on the datafile");

  puts("afterwards to update the keys.");

  puts("You should give the .MYI file as the filename argument.");

  VOID(printf("\nUsage: %s [OPTIONS] filename...\n", my_progname));

  my_print_help(my_long_options);

  print_defaults("my", load_default_groups);

  my_print_variables(my_long_options);

}

static bool

get_one_option(int optid, const struct my_option *opt __attribute__((unused)),

	       char *argument)

{

  uint length;

  switch(optid) {

#ifdef __NETWARE__

  case OPT_AUTO_CLOSE:

    setscreenmode(SCR_AUTOCLOSE_ON_EXIT);

    break;

#endif

  case 'f':

    force_pack= 1;

    tmpfile_createflag= O_RDWR | O_TRUNC;

    break;

  case 's':

    write_loop= verbose= 0;

    silent= 1;

    break;

  case 't':

    test_only= 1;

    /* Avoid to reset 'verbose' if it was already set > 1. */

    if (! verbose)

      verbose= 1;

    break;

  case 'T':

    length= (uint) (strmov(tmp_dir, argument) - tmp_dir);

    if (length != dirname_length(tmp_dir))

    {

      tmp_dir[length]=FN_LIBCHAR;

      tmp_dir[length+1]=0;

    }

    break;

  case 'v':

    verbose++; /* Allow for selecting the level of verbosity. */

    silent= 0;

    break;

  case 'V':

    print_version();

    exit(0);

  case 'I':

  case '?':

    usage();

    exit(0);

  }

  return 0;

}

	/* reads options */

	/* Initiates DEBUG - but no debugging here ! */

static void get_options(int *argc,char ***argv)

{

  int ho_error;

  my_progname= argv[0][0];

  if (isatty(fileno(stdout)))

    write_loop=1;

  if ((ho_error=handle_options(argc, argv, my_long_options, get_one_option)))

    exit(ho_error);

  if (!*argc)

  {

    usage();

    exit(1);

  }

  if (join_table)

  {

    backup=0;					/* Not needed */

    tmp_dir[0]=0;

  }

  return;

}

static MI_INFO *open_isam_file(char *name,int mode)

{

  MI_INFO *isam_file;

  MYISAM_SHARE *share;

  if (!(isam_file=mi_open(name,mode,

			  (opt_wait ? HA_OPEN_WAIT_IF_LOCKED :

			   HA_OPEN_ABORT_IF_LOCKED))))

  {

    VOID(fprintf(stderr, "%s gave error %d on open\n", name, my_errno));

    return(0);

  }

  share=isam_file->s;

  if (share->options & HA_OPTION_COMPRESS_RECORD && !join_table)

  {

    if (!force_pack)

    {

      VOID(fprintf(stderr, "%s is already compressed\n", name));

      VOID(mi_close(isam_file));

      return(0);

    }

    if (verbose)

      puts("Recompressing already compressed table");

    share->options&= ~HA_OPTION_READ_ONLY_DATA; /* We are modifing it */

  }

  if (! force_pack && share->state.state.records != 0 &&

      (share->state.state.records <= 1 ||

       share->state.state.data_file_length < 1024))

  {

    VOID(fprintf(stderr, "%s is too small to compress\n", name));

    VOID(mi_close(isam_file));

    return(0);

  }

  VOID(mi_lock_database(isam_file,F_WRLCK));

  return(isam_file);

}

static my_bool open_isam_files(PACK_MRG_INFO *mrg, char **names, uint count)

{

  uint i,j;

  mrg->count=0;

  mrg->current=0;

  mrg->file=(MI_INFO**) my_malloc(sizeof(MI_INFO*)*count,MYF(MY_FAE));

  mrg->free_file=1;

  mrg->src_file_has_indexes_disabled= 0;

  for (i=0; i < count ; i++)

  {

    if (!(mrg->file[i]=open_isam_file(names[i],O_RDONLY)))

      goto error;

    mrg->src_file_has_indexes_disabled|=

      ! mi_is_all_keys_active(mrg->file[i]->s->state.key_map,

                              mrg->file[i]->s->base.keys);

  }

  /* Check that files are identical */

  for (j=0 ; j < count-1 ; j++)

  {

    MI_COLUMNDEF *m1,*m2,*end;

    if (mrg->file[j]->s->base.reclength != mrg->file[j+1]->s->base.reclength ||

	mrg->file[j]->s->base.fields != mrg->file[j+1]->s->base.fields)

      goto diff_file;

    m1=mrg->file[j]->s->rec;

    end=m1+mrg->file[j]->s->base.fields;

    m2=mrg->file[j+1]->s->rec;

    for ( ; m1 != end ; m1++,m2++)

    {

      if (m1->type != m2->type || m1->length != m2->length)

	goto diff_file;

    }

  }

  mrg->count=count;

  return 0;

 diff_file:

  VOID(fprintf(stderr, "%s: Tables '%s' and '%s' are not identical\n",

               my_progname, names[j], names[j+1]));

 error:

  while (i--)

    mi_close(mrg->file[i]);

  my_free((uchar*) mrg->file,MYF(0));

  return 1;

}

static int compress(PACK_MRG_INFO *mrg,char *result_table)

{

  int error;

  File new_file,join_isam_file;

  MI_INFO *isam_file;

  MYISAM_SHARE *share;

  char org_name[FN_REFLEN],new_name[FN_REFLEN],temp_name[FN_REFLEN];

  uint i,header_length,fields,trees,used_trees;

  my_off_t old_length,new_length,tot_elements;

  HUFF_COUNTS *huff_counts;

  HUFF_TREE *huff_trees;

  isam_file=mrg->file[0];			/* Take this as an example */

  share=isam_file->s;

  new_file=join_isam_file= -1;

  trees=fields=0;

  huff_trees=0;

  huff_counts=0;

  /* Create temporary or join file */

  if (backup)

    VOID(fn_format(org_name,isam_file->filename,"",MI_NAME_DEXT,2));

  else

    VOID(fn_format(org_name,isam_file->filename,"",MI_NAME_DEXT,2+4+16));

  if (!test_only && result_table)

  {

    /* Make a new indexfile based on first file in list */

    uint length;

    uchar *buff;

    strmov(org_name,result_table);		/* Fix error messages */

    VOID(fn_format(new_name,result_table,"",MI_NAME_IEXT,2));

    if ((join_isam_file=my_create(new_name,0,tmpfile_createflag,MYF(MY_WME)))

	< 0)

      goto err;

    length=(uint) share->base.keystart;

    if (!(buff= (uchar*) my_malloc(length,MYF(MY_WME))))

      goto err;

    if (my_pread(share->kfile,buff,length,0L,MYF(MY_WME | MY_NABP)) ||

	my_write(join_isam_file,buff,length,

		 MYF(MY_WME | MY_NABP | MY_WAIT_IF_FULL)))

    {

      my_free(buff,MYF(0));

      goto err;

    }

    my_free(buff,MYF(0));

    VOID(fn_format(new_name,result_table,"",MI_NAME_DEXT,2));

  }

  else if (!tmp_dir[0])

    VOID(make_new_name(new_name,org_name));

  else

    VOID(fn_format(new_name,org_name,tmp_dir,DATA_TMP_EXT,1+2+4));

  if (!test_only &&

      (new_file=my_create(new_name,0,tmpfile_createflag,MYF(MY_WME))) < 0)

    goto err;

  /* Start calculating statistics */

  mrg->records=0;

  for (i=0 ; i < mrg->count ; i++)

    mrg->records+=mrg->file[i]->s->state.state.records;

  if (write_loop || verbose)

  {

    VOID(printf("Compressing %s: (%lu records)\n",

                result_table ? new_name : org_name, (ulong) mrg->records));

  }

  trees=fields=share->base.fields;

  huff_counts=init_huff_count(isam_file,mrg->records);

  QUICK_SAFEMALLOC;

  /*

    Read the whole data file(s) for statistics.

  */

  if (write_loop || verbose)

    VOID(printf("- Calculating statistics\n"));

  if (get_statistic(mrg,huff_counts))

    goto err;

  NORMAL_SAFEMALLOC;

  old_length=0;

  for (i=0; i < mrg->count ; i++)

    old_length+= (mrg->file[i]->s->state.state.data_file_length -

		  mrg->file[i]->s->state.state.empty);

  /*

    Create a global priority queue in preparation for making 

    temporary Huffman trees.

  */

  if (init_queue(&queue,256,0,0,compare_huff_elements,0))

    goto err;

  /*

    Check each column if we should use pre-space-compress, end-space-

    compress, empty-field-compress or zero-field-compress.

  */

  check_counts(huff_counts,fields,mrg->records);

  /*

    Build a Huffman tree for each column.

  */

  huff_trees=make_huff_trees(huff_counts,trees);

  /*

    If the packed lengths of combined columns is less then the sum of

    the non-combined columns, then create common Huffman trees for them.

    We do this only for byte compressed columns, not for distinct values

    compressed columns.

  */

  if ((int) (used_trees=join_same_trees(huff_counts,trees)) < 0)

    goto err;

  /*

    Assign codes to all byte or column values.

  */

  if (make_huff_decode_table(huff_trees,fields))

    goto err;

  /* Prepare a file buffer. */

  init_file_buffer(new_file,0);

  /*

    Reserve space in the target file for the fixed compressed file header.

  */

  file_buffer.pos_in_file=HEAD_LENGTH;

  if (! test_only)

    VOID(my_seek(new_file,file_buffer.pos_in_file,MY_SEEK_SET,MYF(0)));

  /*

    Write field infos: field type, pack type, length bits, tree number.

  */

  write_field_info(huff_counts,fields,used_trees);

  /*

    Write decode trees.

  */

  if (!(tot_elements=write_huff_tree(huff_trees,trees)))

    goto err;

  /*

    Calculate the total length of the compression info header.

    This includes the fixed compressed file header, the column compression

    type descriptions, and the decode trees.

  */

  header_length=(uint) file_buffer.pos_in_file+

    (uint) (file_buffer.pos-file_buffer.buffer);

  /*

    Compress the source file into the target file.

  */

  if (write_loop || verbose)

    VOID(printf("- Compressing file\n"));

  error=compress_isam_file(mrg,huff_counts);

  new_length=file_buffer.pos_in_file;

  if (!error && !test_only)

  {

    uchar buff[MEMMAP_EXTRA_MARGIN];		/* End marginal for memmap */

    memset(buff, 0, sizeof(buff));

    error=my_write(file_buffer.file,buff,sizeof(buff),

		   MYF(MY_WME | MY_NABP | MY_WAIT_IF_FULL)) != 0;

  }

  /*

    Write the fixed compressed file header.

  */

  if (!error)

    error=write_header(mrg,header_length,used_trees,tot_elements,

		       new_length);

  /* Flush the file buffer. */

  end_file_buffer();

  /* Display statistics. */

  if (verbose && mrg->records)

    VOID(printf("Min record length: %6d   Max length: %6d   "

                "Mean total length: %6ld\n", mrg->min_pack_length,

                mrg->max_pack_length, (ulong) (new_length/mrg->records)));

  /* Close source and target file. */

  if (!test_only)

  {

    error|=my_close(new_file,MYF(MY_WME));

    if (!result_table)

    {

      error|=my_close(isam_file->dfile,MYF(MY_WME));

      isam_file->dfile= -1;		/* Tell mi_close file is closed */

    }

  }

  /* Cleanup. */

  free_counts_and_tree_and_queue(huff_trees,trees,huff_counts,fields);

  if (! test_only && ! error)

  {

    if (result_table)

    {

      error=save_state_mrg(join_isam_file,mrg,new_length,glob_crc);

    }

    else

    {

      if (backup)

      {

	if (my_rename(org_name,make_old_name(temp_name,isam_file->filename),

		      MYF(MY_WME)))

	  error=1;

	else

	{

	  if (tmp_dir[0])

	    error=my_copy(new_name,org_name,MYF(MY_WME));

	  else

	    error=my_rename(new_name,org_name,MYF(MY_WME));

	  if (!error)

          {

	    VOID(my_copystat(temp_name,org_name,MYF(MY_COPYTIME)));

            if (tmp_dir[0])

              VOID(my_delete(new_name,MYF(MY_WME)));

          }

	}

      }

      else

      {

	if (tmp_dir[0])

        {

	  error=my_copy(new_name,org_name,

			MYF(MY_WME | MY_HOLD_ORIGINAL_MODES | MY_COPYTIME));

          if (!error)

            VOID(my_delete(new_name,MYF(MY_WME)));

        }

	else

	  error=my_redel(org_name,new_name,MYF(MY_WME | MY_COPYTIME));

      }

      if (! error)

	error=save_state(isam_file,mrg,new_length,glob_crc);

    }