Added API for copying aria tables in mariabackup

7 files changed, 779 insertions, 6 deletions

diff --git a/.gitignore b/.gitignore
index 2ae0a2332fd..9c2e1d7a8c0 100644
--- a/.gitignore
+++ b/.gitignore
@@ -182,6 +182,7 @@ storage/maria/ma_sp_test
 storage/maria/ma_test1
 storage/maria/ma_test2
 storage/maria/ma_test3
+storage/maria/test_ma_backup
 storage/myisam/mi_test1
 storage/myisam/mi_test2
 storage/myisam/mi_test3
diff --git a/include/aria_backup.h b/include/aria_backup.h
new file mode 100644
index 00000000000..bb2d06b0acd
--- /dev/null
+++ b/include/aria_backup.h
@@ -0,0 +1,34 @@
+/* Copyright (C) 2018 MariaDB corporation
+
+   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 Street, Fifth Floor, Boston, MA 02111-1301 USA */
+
+/* Interfaces for doing backups of Aria tables */
+
+typedef struct st_aria_table_capabilities
+{
+  my_off_t header_size;
+  ulong bitmap_pages_covered;
+  uint block_size;
+  uint keypage_header;
+  my_bool checksum;
+  my_bool transactional;
+  /* This is true if the table can be copied without any locks */
+  my_bool online_backup_safe;
+} ARIA_TABLE_CAPABILITIES;
+
+int aria_get_capabilities(File kfile, ARIA_TABLE_CAPABILITIES *cap);
+int aria_read_index(File kfile, ARIA_TABLE_CAPABILITIES *cap, ulonglong block,
+                    uchar *buffer);
+int aria_read_data(File dfile, ARIA_TABLE_CAPABILITIES *cap, ulonglong block,
+                   uchar *buffer, size_t *bytes_read);
diff --git a/storage/maria/CMakeLists.txt b/storage/maria/CMakeLists.txt
index 77dfb201413..6ed370653d5 100644
--- a/storage/maria/CMakeLists.txt
+++ b/storage/maria/CMakeLists.txt
@@ -47,7 +47,7 @@ SET(ARIA_SOURCES ma_init.c ma_open.c ma_extra.c ma_info.c ma_rkey.c
             ma_checkpoint.c ma_recovery.c ma_commit.c ma_pagecrc.c
             ha_maria.h maria_def.h ma_recovery_util.c ma_servicethread.c
             ma_norec.c
-            ma_crypt.c
+            ma_crypt.c ma_backup.c
 )
 
 IF(APPLE)
@@ -98,7 +98,12 @@ IF(WITH_UNIT_TESTS)
 
   ADD_EXECUTABLE(ma_sp_test ma_sp_test.c)
   TARGET_LINK_LIBRARIES(ma_sp_test aria)
+
+  ADD_EXECUTABLE(test_ma_backup test_ma_backup.c)
+  TARGET_LINK_LIBRARIES(test_ma_backup aria)
+
   ADD_SUBDIRECTORY(unittest)
+
 ENDIF()
   
 IF (MSVC)
diff --git a/storage/maria/ma_backup.c b/storage/maria/ma_backup.c
new file mode 100644
index 00000000000..8f20209c48a
--- /dev/null
+++ b/storage/maria/ma_backup.c
@@ -0,0 +1,281 @@
+/* Copyright (C) 2018 MariaDB corporation
+
+   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 Street, Fifth Floor, Boston, MA 02111-1301 USA */
+
+/* Code for doing backups of Aria tables */
+
+#include "maria_def.h"
+#include "ma_blockrec.h"                        /* PAGE_SUFFIX_SIZE */
+#include "ma_checkpoint.h"
+#include <aria_backup.h>
+
+static uchar *_ma_base_info_read(uchar *ptr, MARIA_BASE_INFO *base);
+
+/**
+  @brief Get capabilites for an Aria table
+
+  @param kfile   key file (.MAI)
+  @param cap     Capabilities are stored here
+
+  @return 0      ok
+  @return X      errno
+*/
+
+int aria_get_capabilities(File kfile, ARIA_TABLE_CAPABILITIES *cap)
+{
+  MARIA_SHARE share;
+  int error= 0;
+  uint head_length= sizeof(share.state.header), base_pos;
+  uint aligned_bit_blocks;
+  size_t info_length;
+  uchar *disc_cache;
+  DBUG_ENTER("aria_get_capabilities");
+
+  bzero(cap, sizeof(*cap));
+  if (my_pread(kfile,share.state.header.file_version, head_length, 0,
+               MYF(MY_NABP)))
+    DBUG_RETURN(HA_ERR_NOT_A_TABLE);
+
+  if (memcmp(share.state.header.file_version, maria_file_magic, 4))
+    DBUG_RETURN(HA_ERR_NOT_A_TABLE);
+
+  share.options= mi_uint2korr(share.state.header.options);
+
+  info_length= mi_uint2korr(share.state.header.header_length);
+  base_pos=    mi_uint2korr(share.state.header.base_pos);
+
+  /*
+    Allocate space for header information and for data that is too
+    big to keep on stack
+  */
+  if (!(disc_cache= my_malloc(info_length, MYF(MY_WME))))
+    DBUG_RETURN(ENOMEM);
+
+  if (my_pread(kfile, disc_cache, info_length, 0L, MYF(MY_NABP)))
+  {
+    error= my_errno;
+    goto err;
+  }
+  _ma_base_info_read(disc_cache + base_pos, &share.base);
+  cap->transactional= share.base.born_transactional;
+  cap->checksum= MY_TEST(share.options & HA_OPTION_PAGE_CHECKSUM);
+  cap->online_backup_safe= cap->transactional && cap->checksum;
+  cap->header_size= share.base.keystart;
+  cap->keypage_header= ((share.base.born_transactional ?
+                         LSN_STORE_SIZE + TRANSID_SIZE :
+                         0) + KEYPAGE_KEYID_SIZE + KEYPAGE_FLAG_SIZE +
+                        KEYPAGE_USED_SIZE);
+  cap->block_size= share.base.block_size;
+
+  if (share.state.header.data_file_type == BLOCK_RECORD)
+  {
+    /* Calulate how man pages the row bitmap covers. From _ma_bitmap_init() */
+    aligned_bit_blocks= (cap->block_size - PAGE_SUFFIX_SIZE) / 6;
+    /*
+      In each 6 bytes, we have 6*8/3 = 16 pages covered
+      The +1 is to add the bitmap page, as this doesn't have to be covered
+    */
+    cap->bitmap_pages_covered= aligned_bit_blocks * 16 + 1;
+  }
+
+  /* Do a check that that we got things right */
+  if (share.state.header.data_file_type != BLOCK_RECORD &&
+      cap->online_backup_safe)
+    error= HA_ERR_NOT_A_TABLE;
+
+err:
+  my_free(disc_cache);
+  DBUG_RETURN(error);
+} /* maria_get_capabilities */
+
+
+/*
+  This is a copy of my_base_info_read from ma_open().
+  The base information will never change (something may be added
+  last, but not relevant for maria_get_capabilities), so it's safe to
+  copy it here.
+
+  The copy is done to avoid linking in the fill Aria library just
+  because maria_backup uses maria_get_capabilities()
+*/
+
+
+static uchar *_ma_base_info_read(uchar *ptr, MARIA_BASE_INFO *base)
+{
+  bmove(base->uuid, ptr, MY_UUID_SIZE);                 ptr+= MY_UUID_SIZE;
+  base->keystart= mi_sizekorr(ptr);			ptr+= 8;
+  base->max_data_file_length= mi_sizekorr(ptr); 	ptr+= 8;
+  base->max_key_file_length= mi_sizekorr(ptr);		ptr+= 8;
+  base->records=  (ha_rows) mi_sizekorr(ptr);		ptr+= 8;
+  base->reloc= (ha_rows) mi_sizekorr(ptr);		ptr+= 8;
+  base->mean_row_length= mi_uint4korr(ptr);		ptr+= 4;
+  base->reclength= mi_uint4korr(ptr);			ptr+= 4;
+  base->pack_reclength= mi_uint4korr(ptr);		ptr+= 4;
+  base->min_pack_length= mi_uint4korr(ptr);		ptr+= 4;
+  base->max_pack_length= mi_uint4korr(ptr);		ptr+= 4;
+  base->min_block_length= mi_uint4korr(ptr);		ptr+= 4;
+  base->fields= mi_uint2korr(ptr);			ptr+= 2;
+  base->fixed_not_null_fields= mi_uint2korr(ptr);       ptr+= 2;
+  base->fixed_not_null_fields_length= mi_uint2korr(ptr);ptr+= 2;
+  base->max_field_lengths= mi_uint2korr(ptr);	        ptr+= 2;
+  base->pack_fields= mi_uint2korr(ptr);			ptr+= 2;
+  base->extra_options= mi_uint2korr(ptr);		ptr+= 2;
+  base->null_bytes= mi_uint2korr(ptr);			ptr+= 2;
+  base->original_null_bytes= mi_uint2korr(ptr);		ptr+= 2;
+  base->field_offsets= mi_uint2korr(ptr);		ptr+= 2;
+  base->language= mi_uint2korr(ptr);		        ptr+= 2;
+  base->block_size= mi_uint2korr(ptr);			ptr+= 2;
+
+  base->rec_reflength= *ptr++;
+  base->key_reflength= *ptr++;
+  base->keys=	       *ptr++;
+  base->auto_key=      *ptr++;
+  base->born_transactional= *ptr++;
+  ptr++;
+  base->pack_bytes= mi_uint2korr(ptr);			ptr+= 2;
+  base->blobs= mi_uint2korr(ptr);			ptr+= 2;
+  base->max_key_block_length= mi_uint2korr(ptr);	ptr+= 2;
+  base->max_key_length= mi_uint2korr(ptr);		ptr+= 2;
+  base->extra_alloc_bytes= mi_uint2korr(ptr);		ptr+= 2;
+  base->extra_alloc_procent= *ptr++;
+  ptr+= 16;
+  return ptr;
+}
+
+
+/**
+   @brief Copy an index block with re-read if checksum doesn't match
+
+   @param dfile       data file (.MAD)
+   @param cap         aria capabilities from aria_get_capabilities
+   @param block       block number to read (0, 1, 2, 3...)
+   @param buffer      read data to this buffer
+   @param bytes_read  number of bytes actually read (in case of end of file)
+
+   @return 0 ok
+   @return HA_ERR_END_OF_FILE  ; End of file
+   @return # error number
+*/
+
+#define MAX_RETRY 10
+
+int aria_read_index(File kfile, ARIA_TABLE_CAPABILITIES *cap, ulonglong block,
+                    uchar *buffer)
+{
+  MARIA_SHARE share;
+  int retry= 0;
+  DBUG_ENTER("aria_read_index");
+
+  share.keypage_header= cap->keypage_header;
+  share.block_size= cap->block_size;
+  do
+  {
+    int error;
+    size_t length;
+    if ((length= my_pread(kfile, buffer, cap->block_size,
+                          block * cap->block_size, MYF(0))) != cap->block_size)
+    {
+      if (length == 0)
+        DBUG_RETURN(HA_ERR_END_OF_FILE);
+      if (length == (size_t) -1)
+        DBUG_RETURN(my_errno ? my_errno : -1);
+      /* Assume we got a half read; Do a re-read */
+    }
+    /* If not transactional or key file header, there are no checksums */
+    if (!cap->online_backup_safe ||
+        block < cap->header_size/ cap->block_size)
+      DBUG_RETURN(length == cap->block_size ? 0 : HA_ERR_CRASHED);
+
+    if (length == cap->block_size)
+    {
+      length= _ma_get_page_used(&share, buffer);
+      if (length > cap->block_size - CRC_SIZE)
+        DBUG_RETURN(HA_ERR_CRASHED);
+      error= maria_page_crc_check(buffer, block, &share,
+                                  MARIA_NO_CRC_NORMAL_PAGE,
+                                  (int) length);
+      if (error != HA_ERR_WRONG_CRC)
+        DBUG_RETURN(error);
+    }
+    my_sleep(100000);                              /* Sleep 0.1 seconds */
+  } while (retry < MAX_RETRY);
+  DBUG_RETURN(HA_ERR_WRONG_CRC);
+}
+
+
+/**
+   @brief Copy a data block with re-read if checksum doesn't match
+
+   @param dfile       data file (.MAD)
+   @param cap         aria capabilities from aria_get_capabilities
+   @param block       block number to read (0, 1, 2, 3...)
+   @param buffer      read data to this buffer
+   @param bytes_read  number of bytes actually read (in case of end of file)
+
+   @return 0 ok
+   @return HA_ERR_END_OF_FILE  ; End of file
+   @return # error number
+*/
+
+int aria_read_data(File dfile, ARIA_TABLE_CAPABILITIES *cap, ulonglong block,
+                   uchar *buffer, size_t *bytes_read)
+{
+  MARIA_SHARE share;
+  int retry= 0;
+  DBUG_ENTER("aria_read_data");
+
+  share.keypage_header= cap->keypage_header;
+  share.block_size= cap->block_size;
+
+  if (!cap->online_backup_safe)
+  {
+    *bytes_read= my_pread(dfile, buffer, cap->block_size,
+                          block * cap->block_size, MY_WME);
+    if (*bytes_read == 0)
+      DBUG_RETURN(HA_ERR_END_OF_FILE);
+    DBUG_RETURN(*bytes_read > 0 ? 0 : (my_errno ? my_errno : -1));
+  }
+
+  *bytes_read= cap->block_size;
+  do
+  {
+    int error;
+    size_t length;
+    if ((length= my_pread(dfile, buffer, cap->block_size,
+                          block * cap->block_size, MYF(0))) != cap->block_size)
+    {
+      if (length == 0)
+        DBUG_RETURN(HA_ERR_END_OF_FILE);
+      if (length == (size_t) -1)
+        DBUG_RETURN(my_errno ? my_errno : -1);
+    }
+
+    /* If not transactional or key file header, there are no checksums */
+    if (!cap->online_backup_safe)
+      DBUG_RETURN(length == cap->block_size ? 0 : HA_ERR_CRASHED);
+
+    if (length == cap->block_size)
+    {
+      error= maria_page_crc_check(buffer, block, &share,
+                                  ((block % cap->bitmap_pages_covered) == 0 ?
+                                   MARIA_NO_CRC_BITMAP_PAGE :
+                                   MARIA_NO_CRC_NORMAL_PAGE),
+                                  share.block_size - CRC_SIZE);
+      if (error != HA_ERR_WRONG_CRC)
+        DBUG_RETURN(error);
+    }
+    my_sleep(100000);                              /* Sleep 0.1 seconds */
+  } while (retry < MAX_RETRY);
+    DBUG_RETURN(HA_ERR_WRONG_CRC);
+}
diff --git a/storage/maria/ma_pagecrc.c b/storage/maria/ma_pagecrc.c
index 8982c7e5c09..aae11158286 100644
--- a/storage/maria/ma_pagecrc.c
+++ b/storage/maria/ma_pagecrc.c
@@ -54,11 +54,11 @@ static uint32 maria_page_crc(uint32 start, uchar *data, uint length)
   @retval 1 Error
 */
 
-static my_bool maria_page_crc_check(uchar *page,
-                                    pgcache_page_no_t page_no,
-                                    MARIA_SHARE *share,
-                                    uint32 no_crc_val,
-                                    int data_length)
+my_bool maria_page_crc_check(uchar *page,
+                             pgcache_page_no_t page_no,
+                             MARIA_SHARE *share,
+                             uint32 no_crc_val,
+                             int data_length)
 {
   uint32 crc= uint4korr(page + share->block_size - CRC_SIZE), new_crc;
   my_bool res;
diff --git a/storage/maria/maria_def.h b/storage/maria/maria_def.h
index d5f2b1e80fd..7d051faeac9 100644
--- a/storage/maria/maria_def.h
+++ b/storage/maria/maria_def.h
@@ -1417,6 +1417,9 @@ extern my_bool maria_page_crc_check_bitmap(int, PAGECACHE_IO_HOOK_ARGS *args);
 extern my_bool maria_page_crc_check_data(int, PAGECACHE_IO_HOOK_ARGS *args);
 extern my_bool maria_page_crc_check_index(int, PAGECACHE_IO_HOOK_ARGS *args);
 extern my_bool maria_page_crc_check_none(int, PAGECACHE_IO_HOOK_ARGS *args);
+extern my_bool maria_page_crc_check(uchar *page, pgcache_page_no_t page_no,
+                                    MARIA_SHARE *share, uint32 no_crc_val,
+                                    int data_length);
 extern my_bool maria_page_filler_set_bitmap(PAGECACHE_IO_HOOK_ARGS *args);
 extern my_bool maria_page_filler_set_normal(PAGECACHE_IO_HOOK_ARGS *args);
 extern my_bool maria_page_filler_set_none(PAGECACHE_IO_HOOK_ARGS *args);
diff --git a/storage/maria/test_ma_backup.c b/storage/maria/test_ma_backup.c
new file mode 100644
index 00000000000..439ea7fd665
--- /dev/null
+++ b/storage/maria/test_ma_backup.c
@@ -0,0 +1,449 @@
+/* Copyright (C) 2018 MariaDB corporation
+
+   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 Street, Fifth Floor, Boston, MA 02111-1301 USA */
+
+/* Code for doing backups of Aria tables */
+
+/******************************************************************************
+  Testing ma_backup interface
+  Table creation code is taken from ma_test1
+******************************************************************************/
+
+#define ROWS_IN_TEST 100000
+
+#include "maria_def.h"
+#include "ma_blockrec.h"                        /* PAGE_SUFFIX_SIZE */
+#include "ma_checkpoint.h"
+#include <aria_backup.h>
+
+static int silent;
+static int create_test_table(const char *table_name, int stage);
+static int copy_table(const char *table_name, int stage);
+static void create_record(uchar *record,uint rownr);
+
+int main(int argc,char *argv[])
+{
+  int error= 1;
+  int i;
+  char buff[FN_REFLEN];
+#ifdef SAFE_MUTEX
+  safe_mutex_deadlock_detector= 1;
+#endif
+  MY_INIT(argv[0]);
+  maria_data_root= (char *)".";
+
+  /* Maria requires that we always have a page cache */
+  if (maria_init() ||
+      (init_pagecache(maria_pagecache, maria_block_size * 2000, 0, 0,
+                      maria_block_size, 0, MY_WME) == 0) ||
+      ma_control_file_open(TRUE, TRUE) ||
+      (init_pagecache(maria_log_pagecache,
+                      TRANSLOG_PAGECACHE_SIZE, 0, 0,
+                      TRANSLOG_PAGE_SIZE, 0, MY_WME) == 0) ||
+      translog_init(maria_data_root, TRANSLOG_FILE_SIZE,
+                    0, 0, maria_log_pagecache,
+                    TRANSLOG_DEFAULT_FLAGS, 0) ||
+      (trnman_init(0) || ma_checkpoint_init(0)))
+  {
+    fprintf(stderr, "Error in initialization\n");
+    exit(1);
+  }
+  init_thr_lock();
+
+  fn_format(buff, "test_copy", maria_data_root, "", MYF(0));
+
+  for (i= 0; i < 5 ; i++)
+  {
+    printf("Stage: %d\n", i);
+    fflush(stdout);
+    if (create_test_table(buff, i))
+      goto err;
+    if (copy_table(buff, i))
+      goto err;
+  }
+  error= 0;
+  printf("test ok\n");
+err:
+  if (error)
+    fprintf(stderr, "Test %i failed\n", i);
+  maria_end();
+  my_uuid_end();
+  my_end(MY_CHECK_ERROR);
+  exit(error);
+}
+
+
+/**
+   Example of how to read an Aria table
+*/
+
+static int copy_table(const char *table_name, int stage)
+{
+  char old_name[FN_REFLEN];
+  uchar *copy_buffer= 0;
+  ARIA_TABLE_CAPABILITIES cap;
+  ulonglong block;
+  File org_file= -1;
+  int error= 1;
+
+  strxmov(old_name, table_name, ".MAI", NullS);
+
+  if ((org_file= my_open(old_name,
+                         O_RDONLY | O_SHARE | O_NOFOLLOW | O_CLOEXEC,
+                         MYF(MY_WME))) < 0)
+    goto err;
+  if ((error= aria_get_capabilities(org_file, &cap)))
+  {
+    fprintf(stderr, "aria_get_capabilities failed:  %d\n", error);
+    goto err;
+  }
+
+  printf("- Capabilities read. oneline_backup_safe: %d\n",
+         cap.online_backup_safe);
+  printf("- Copying index file\n");
+
+  copy_buffer= my_malloc(cap.block_size, MYF(0));
+  for (block= 0 ; ; block++)
+  {
+    if ((error= aria_read_index(org_file, &cap, block, copy_buffer) ==
+         HA_ERR_END_OF_FILE))
+      break;
+    if (error)
+    {
+      fprintf(stderr, "aria_read_index failed:  %d\n", error);
+      goto err;
+    }
+  }
+  my_close(org_file, MYF(MY_WME));
+
+
+  printf("- Copying data file\n");
+  strxmov(old_name, table_name, ".MAD", NullS);
+  if ((org_file= my_open(old_name, O_RDONLY | O_SHARE | O_NOFOLLOW | O_CLOEXEC,
+                         MYF(MY_WME))) < 0)
+    goto err;
+
+  for (block= 0 ; ; block++)
+  {
+    size_t length;
+    if ((error= aria_read_data(org_file, &cap, block, copy_buffer,
+                               &length) == HA_ERR_END_OF_FILE))
+      break;
+    if (error)
+    {
+      fprintf(stderr, "aria_read_index failed:  %d\n", error);
+      goto err;
+    }
+  }
+  error= 0;
+
+err:
+  my_free(copy_buffer);
+  if (org_file >= 0)
+    my_close(org_file, MYF(MY_WME));
+  if (error)
+    fprintf(stderr, "Failed in copy_table stage: %d\n", stage);
+  return error;
+}
+
+
+/* Code extracted from ma_test1.c */
+#define MAX_REC_LENGTH 1024
+
+static MARIA_COLUMNDEF recinfo[4];
+static MARIA_KEYDEF keyinfo[10];
+static HA_KEYSEG keyseg[10];
+static HA_KEYSEG uniqueseg[10];
+
+
+/**
+   Create a test table and fill it with some data
+*/
+
+static int create_test_table(const char *table_name, int type_of_table)
+{
+  MARIA_HA *file;
+  int i,error,uniques=0;
+  int key_field=FIELD_SKIP_PRESPACE,extra_field=FIELD_SKIP_ENDSPACE;
+  int key_type=HA_KEYTYPE_NUM;
+  int create_flag=0;
+  uint offset_to_key;
+  uint pack_seg=0, pack_keys= 0;
+  uint key_length;
+  uchar record[MAX_REC_LENGTH];
+  MARIA_UNIQUEDEF uniquedef;
+  MARIA_CREATE_INFO create_info;
+  enum data_file_type record_type= DYNAMIC_RECORD;
+  my_bool null_fields= 0, unique_key= 0;
+  my_bool opt_unique= 0;
+  my_bool transactional= 0;
+
+  key_length= 12;
+  switch (type_of_table) {
+  case 0:
+    break;
+  case 1:
+    create_flag|= HA_CREATE_CHECKSUM | HA_CREATE_PAGE_CHECKSUM;
+    break;
+  case 2:                                       /* transactional */
+    create_flag|= HA_CREATE_CHECKSUM | HA_CREATE_PAGE_CHECKSUM;
+    record_type= BLOCK_RECORD;
+    transactional= 1;
+    break;
+  case 3:                                       /* transactional */
+    create_flag|= HA_CREATE_CHECKSUM | HA_CREATE_PAGE_CHECKSUM;
+    record_type= BLOCK_RECORD;
+    transactional= 1;
+    key_field=FIELD_VARCHAR;			/* varchar keys */
+    extra_field= FIELD_VARCHAR;
+    key_type= HA_KEYTYPE_VARTEXT1;
+    pack_seg|= HA_VAR_LENGTH_PART;
+    null_fields= 1;
+    break;
+  case 4:                                       /* transactional */
+    create_flag|= HA_CREATE_CHECKSUM | HA_CREATE_PAGE_CHECKSUM;
+    record_type= BLOCK_RECORD;
+    transactional= 1;
+    key_field=FIELD_BLOB;			/* blob key */
+    extra_field= FIELD_BLOB;
+    pack_seg|= HA_BLOB_PART;
+    key_type= HA_KEYTYPE_VARTEXT1;
+    break;
+  }
+
+
+  bzero((char*) recinfo,sizeof(recinfo));
+  bzero((char*) &create_info,sizeof(create_info));
+
+  /* First define 2 columns */
+  create_info.null_bytes= 1;
+  recinfo[0].type= key_field;
+  recinfo[0].length= (key_field == FIELD_BLOB ? 4+portable_sizeof_char_ptr :
+		      key_length);
+  if (key_field == FIELD_VARCHAR)
+    recinfo[0].length+= HA_VARCHAR_PACKLENGTH(key_length);
+  recinfo[1].type=extra_field;
+  recinfo[1].length= (extra_field == FIELD_BLOB ? 4 + portable_sizeof_char_ptr : 24);
+  if (extra_field == FIELD_VARCHAR)
+    recinfo[1].length+= HA_VARCHAR_PACKLENGTH(recinfo[1].length);
+  recinfo[1].null_bit= null_fields ? 2 : 0;
+
+  if (opt_unique)
+  {
+    recinfo[2].type=FIELD_CHECK;
+    recinfo[2].length=MARIA_UNIQUE_HASH_LENGTH;
+  }
+
+  if (key_type == HA_KEYTYPE_VARTEXT1 &&
+      key_length > 255)
+    key_type= HA_KEYTYPE_VARTEXT2;
+
+  /* Define a key over the first column */
+  keyinfo[0].seg=keyseg;
+  keyinfo[0].keysegs=1;
+  keyinfo[0].block_length= 0;                   /* Default block length */
+  keyinfo[0].key_alg=HA_KEY_ALG_BTREE;
+  keyinfo[0].seg[0].type= key_type;
+  keyinfo[0].seg[0].flag= pack_seg;
+  keyinfo[0].seg[0].start=1;
+  keyinfo[0].seg[0].length=key_length;
+  keyinfo[0].seg[0].null_bit= null_fields ? 2 : 0;
+  keyinfo[0].seg[0].null_pos=0;
+  keyinfo[0].seg[0].language= default_charset_info->number;
+  if (pack_seg & HA_BLOB_PART)
+  {
+    keyinfo[0].seg[0].bit_start=4;		/* Length of blob length */
+  }
+  keyinfo[0].flag = (uint8) (pack_keys | unique_key);
+
+  if (opt_unique)
+  {
+    uint start;
+    uniques=1;
+    bzero((char*) &uniquedef,sizeof(uniquedef));
+    bzero((char*) uniqueseg,sizeof(uniqueseg));
+    uniquedef.seg=uniqueseg;
+    uniquedef.keysegs=2;
+
+    /* Make a unique over all columns (except first NULL fields) */
+    for (i=0, start=1 ; i < 2 ; i++)
+    {
+      uniqueseg[i].start=start;
+      start+=recinfo[i].length;
+      uniqueseg[i].length=recinfo[i].length;
+      uniqueseg[i].language= default_charset_info->number;
+    }
+    uniqueseg[0].type= key_type;
+    uniqueseg[0].null_bit= null_fields ? 2 : 0;
+    uniqueseg[1].type= HA_KEYTYPE_TEXT;
+    if (extra_field == FIELD_BLOB)
+    {
+      uniqueseg[1].length=0;			/* The whole blob */
+      uniqueseg[1].bit_start=4;			/* long blob */
+      uniqueseg[1].flag|= HA_BLOB_PART;
+    }
+    else if (extra_field == FIELD_VARCHAR)
+    {
+      uniqueseg[1].flag|= HA_VAR_LENGTH_PART;
+      uniqueseg[1].type= (HA_VARCHAR_PACKLENGTH(recinfo[1].length-1) == 1 ?
+                          HA_KEYTYPE_VARTEXT1 : HA_KEYTYPE_VARTEXT2);
+    }
+  }
+  else
+    uniques=0;
+
+  offset_to_key= MY_TEST(null_fields);
+  if (key_field == FIELD_BLOB || key_field == FIELD_VARCHAR)
+    offset_to_key+= 2;
+
+  if (!silent)
+    printf("- Creating Aria file\n");
+  create_info.max_rows= 0;
+  create_info.transactional= transactional;
+  if (maria_create(table_name, record_type, 1, keyinfo,2+opt_unique,recinfo,
+		uniques, &uniquedef, &create_info,
+		create_flag))
+    goto err;
+  if (!(file=maria_open(table_name,2,HA_OPEN_ABORT_IF_LOCKED)))
+    goto err;
+  if (!silent)
+    printf("- Writing key:s\n");
+
+  if (maria_begin(file))
+    goto err;
+  my_errno=0;
+  for (i= 0 ; i < ROWS_IN_TEST ; i++)
+  {
+    create_record(record,i);
+    if ((error=maria_write(file,record)))
+      goto err;
+  }
+
+  if (maria_commit(file) | maria_close(file))
+    goto err;
+  printf("- Data copied\n");
+  return 0;
+
+err:
+  printf("got error: %3d when using maria-database\n",my_errno);
+  return 1;			/* skip warning */
+}
+
+
+static void create_key_part(uchar *key,uint rownr)
+{
+  if (keyinfo[0].seg[0].type == HA_KEYTYPE_NUM)
+  {
+    sprintf((char*) key,"%*d",keyinfo[0].seg[0].length,rownr);
+  }
+  else if (keyinfo[0].seg[0].type == HA_KEYTYPE_VARTEXT1 ||
+           keyinfo[0].seg[0].type == HA_KEYTYPE_VARTEXT2)
+  {						/* Alpha record */
+    /* Create a key that may be easily packed */
+    bfill(key,keyinfo[0].seg[0].length,rownr < 10 ? 'A' : 'B');
+    sprintf((char*) key+keyinfo[0].seg[0].length-2,"%-2d",rownr % 100);
+    if ((rownr & 7) == 0)
+    {
+      /* Change the key to force a unpack of the next key */
+      bfill(key+3,keyinfo[0].seg[0].length-5,rownr < 10 ? 'a' : 'b');
+    }
+  }
+  else
+  {						/* Alpha record */
+    if (keyinfo[0].seg[0].flag & HA_SPACE_PACK)
+      sprintf((char*) key,"%-*d",keyinfo[0].seg[0].length,rownr);
+    else
+    {
+      /* Create a key that may be easily packed */
+      bfill(key,keyinfo[0].seg[0].length,rownr < 10 ? 'A' : 'B');
+      sprintf((char*) key+keyinfo[0].seg[0].length-2,"%-2d",rownr % 100);
+      if ((rownr & 7) == 0)
+      {
+	/* Change the key to force a unpack of the next key */
+	key[1]= (rownr < 10 ? 'a' : 'b');
+      }
+    }
+  }
+}
+
+
+static uchar blob_key[MAX_REC_LENGTH];
+static uchar blob_record[MAX_REC_LENGTH+20*20];
+
+
+static void create_record(uchar *record,uint rownr)
+{
+  uchar *pos;
+  bzero((char*) record,MAX_REC_LENGTH);
+  record[0]=1;					/* delete marker */
+  if (rownr == 0 && keyinfo[0].seg[0].null_bit)
+    record[0]|=keyinfo[0].seg[0].null_bit;	/* Null key */
+
+  pos=record+1;
+  if (recinfo[0].type == FIELD_BLOB)
+  {
+    size_t tmp;
+    uchar *ptr;
+    create_key_part(blob_key,rownr);
+    tmp=strlen((char*) blob_key);
+    int4store(pos,tmp);
+    ptr=blob_key;
+    memcpy(pos+4,&ptr,sizeof(char*));
+    pos+=recinfo[0].length;
+  }
+  else if (recinfo[0].type == FIELD_VARCHAR)
+  {
+    size_t tmp, pack_length= HA_VARCHAR_PACKLENGTH(recinfo[0].length-1);
+    create_key_part(pos+pack_length,rownr);
+    tmp= strlen((char*) pos+pack_length);
+    if (pack_length == 1)
+      *(uchar*) pos= (uchar) tmp;
+    else
+      int2store(pos,tmp);
+    pos+= recinfo[0].length;
+  }
+  else
+  {
+    create_key_part(pos,rownr);
+    pos+=recinfo[0].length;
+  }
+  if (recinfo[1].type == FIELD_BLOB)
+  {
+    size_t tmp;
+    uchar *ptr;;
+    sprintf((char*) blob_record,"... row: %d", rownr);
+    strappend((char*) blob_record, rownr % MAX_REC_LENGTH,'x');
+    tmp=strlen((char*) blob_record);
+    int4store(pos,tmp);
+    ptr=blob_record;
+    memcpy(pos+4,&ptr,sizeof(char*));
+  }
+  else if (recinfo[1].type == FIELD_VARCHAR)
+  {
+    size_t tmp, pack_length= HA_VARCHAR_PACKLENGTH(recinfo[1].length-1);
+    sprintf((char*) pos+pack_length, "... row: %d", rownr);
+    tmp= strlen((char*) pos+pack_length);
+    if (pack_length == 1)
+      *pos= (uchar) tmp;
+    else
+      int2store(pos,tmp);
+  }
+  else
+  {
+    sprintf((char*) pos,"... row: %d", rownr);
+    strappend((char*) pos,recinfo[1].length,' ');
+  }
+}
+
+#include "ma_check_standalone.h"