/* Copyright (C) 2006 MySQL AB & MySQL Finland AB & TCX DataKonsult AB
   Copyright (c) 2020, 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 02110-1335 USA */

/*
  Read and write key blocks

  The basic structure of a key block is as follows:

  LSN  7 (LSN_STORE_SIZE);     Log number for last change;
    Only for transactional pages
  PACK_TRANSID  6 (TRANSID_SIZE);       Relative transid to pack page transid's
    Only for transactional pages
  KEYNR  1 (KEYPAGE_KEYID_SIZE)  Which index this page belongs to
  FLAG          1 (KEYPAGE_FLAG_SIZE)   Flags for page
  PAGE_SIZE 2 (KEYPAGE_USED_SIZE)   How much of the page is used.
       high-byte-first

  The flag is a combination of the following values:

   KEYPAGE_FLAG_ISNOD            Page is a node
   KEYPAGE_FLAG_HAS_TRANSID      There may be a transid on the page.

  After this we store key data, either packed or not packed, directly
  after each other.  If the page is a node flag, there is a pointer to
  the next key page at page start and after each key.

  At end of page the last KEYPAGE_CHECKSUM_SIZE bytes are reserved for a
  page checksum.
*/

#include "maria_def.h"
#include "trnman.h"
#include "ma_key_recover.h"

/**
   Fill MARIA_PAGE structure for usage with _ma_write_keypage
*/

void _ma_page_setup(MARIA_PAGE *page, MARIA_HA *info,
                    const MARIA_KEYDEF *keyinfo, my_off_t pos,
                    uchar *buff)
{
  MARIA_SHARE *share= info->s;

  page->info=    info;
  page->keyinfo= keyinfo;
  page->buff=    buff;
  page->pos=     pos;
  page->size=    _ma_get_page_used(share, buff);
  page->org_size= page->size;
  page->flag=    _ma_get_keypage_flag(share, buff);
  page->node=    ((page->flag & KEYPAGE_FLAG_ISNOD) ?
                  share->base.key_reflength : 0);
}

#ifdef IDENTICAL_PAGES_AFTER_RECOVERY
void page_cleanup(MARIA_SHARE *share, MARIA_PAGE *page)
{
  uint length= page->size;
  DBUG_ASSERT(length <= share->max_index_block_size);
  bzero(page->buff + length, share->block_size - length);
}
#endif


/**
  Fetch a key-page in memory

  @fn _ma_fetch_keypage()
  @param page  Fill this struct with information about read page
  @param info  Maria handler
  @param keyinfo        Key definition for used key
  @param pos  Position for page (in bytes)
  @param lock  Lock type for page
  @param level  Importance of page; Priority for page cache
  @param buff         Buffer to use for page
  @param return_buffer  Set to 1 if we want to force usage of the "buff"

  @return
  @retval 0  ok
  @retval 1  error
*/

my_bool _ma_fetch_keypage(MARIA_PAGE *page, MARIA_HA *info,
                          const MARIA_KEYDEF *keyinfo,
                          my_off_t pos, enum pagecache_page_lock lock,
                          int level, uchar *buff,
                          my_bool return_buffer __attribute__ ((unused)))
{
  uchar *tmp;
  MARIA_PINNED_PAGE page_link;
  MARIA_SHARE *share= info->s;
  uint block_size= share->block_size;
  DBUG_ENTER("_ma_fetch_keypage");
  DBUG_PRINT("enter",("page: %lu", (ulong) (pos / block_size)));

  tmp= pagecache_read(share->pagecache, &share->kfile,
                      (pgcache_page_no_t) (pos / block_size), level, buff,
                      share->page_type, lock, &page_link.link);

  if (lock != PAGECACHE_LOCK_LEFT_UNLOCKED)
  {
    DBUG_ASSERT(lock == PAGECACHE_LOCK_WRITE || lock == PAGECACHE_LOCK_READ);
    page_link.unlock= (lock == PAGECACHE_LOCK_WRITE ?
                       PAGECACHE_LOCK_WRITE_UNLOCK :
                       PAGECACHE_LOCK_READ_UNLOCK);
    page_link.changed= 0;
    push_dynamic(&info->pinned_pages, (void*) &page_link);
    page->link_offset= (uint)info->pinned_pages.elements-1;
  }

  if (tmp == info->buff)
    info->keyread_buff_used=1;
  else if (!tmp)
  {
    DBUG_PRINT("error",("Got errno: %d from pagecache_read",my_errno));
    info->last_keypage=HA_OFFSET_ERROR;
    _ma_set_fatal_error(info, my_errno);
    DBUG_RETURN(1);
  }
  info->last_keypage= pos;

  /*
    Setup page structure to make pages easy to use
    This is same as page_fill_info, but here inlined as this si used
    so often.
  */
  page->info=    info;
  page->keyinfo= keyinfo;
  page->buff=    tmp;
  page->pos=     pos;
  page->size=    _ma_get_page_used(share, tmp);
  page->org_size= page->size;                    /* For debugging */
  page->flag=    _ma_get_keypage_flag(share, tmp);
  page->node=   ((page->flag & KEYPAGE_FLAG_ISNOD) ?
                 share->base.key_reflength : 0);

#ifdef EXTRA_DEBUG
  {
    uint page_size= page->size;
    if (page_size < 4 || page_size > share->max_index_block_size ||
        _ma_get_keynr(share, tmp) != keyinfo->key_nr)
    {
      DBUG_PRINT("error",("page %lu had wrong page length: %u  page_header: %u  keynr: %u",
                          (ulong) (pos / block_size), page_size,
                          share->keypage_header,
                          _ma_get_keynr(share, tmp)));
      DBUG_DUMP("page", tmp, page_size);
      info->last_keypage = HA_OFFSET_ERROR;
      _ma_set_fatal_error(info, HA_ERR_CRASHED);
      DBUG_RETURN(1);
    }
  }
#endif
  DBUG_RETURN(0);
} /* _ma_fetch_keypage */


/* Write a key-page on disk */

my_bool _ma_write_keypage(MARIA_PAGE *page, enum pagecache_page_lock lock,
                          int level)
{
  MARIA_SHARE *share= page->info->s;
  uint block_size= share->block_size;
  uchar *buff= page->buff;
  my_bool res;
  MARIA_PINNED_PAGE page_link;
  DBUG_ENTER("_ma_write_keypage");

  /*
    The following ensures that for transactional tables we have logged
    all changes that changes the page size (as the logging code sets
    page->org_size)
  */
  DBUG_ASSERT(!share->now_transactional || page->size == page->org_size);

#ifdef EXTRA_DEBUG    /* Safety check */
  {
    uint page_length, nod_flag;
    page_length= _ma_get_page_used(share, buff);
    nod_flag=    _ma_test_if_nod(share, buff);

    DBUG_ASSERT(page->size == page_length);
    DBUG_ASSERT(page->size <= share->max_index_block_size);
    DBUG_ASSERT(page->flag == _ma_get_keypage_flag(share, buff));

    if (page->pos < share->base.keystart ||
        page->pos+block_size > share->state.state.key_file_length ||
        (page->pos & (maria_block_size-1)))
    {
      DBUG_PRINT("error",("Trying to write inside key status region: "
                          "key_start: %lu  length: %lu  page_pos: %lu",
                          (long) share->base.keystart,
                          (long) share->state.state.key_file_length,
                          (long) page->pos));
      my_errno=EINVAL;
      DBUG_ASSERT(0);
      DBUG_RETURN(1);
    }
    DBUG_PRINT("page",("write page at: %lu",(ulong) (page->pos / block_size)));
    DBUG_DUMP("buff", buff, page_length);
    DBUG_ASSERT(page_length >= share->keypage_header + nod_flag +
                page->keyinfo->minlength || maria_in_recovery);
  }
#endif

  /* Verify that keynr is correct */
  DBUG_ASSERT(_ma_get_keynr(share, buff) == page->keyinfo->key_nr);

#if defined(EXTRA_DEBUG) && defined(HAVE_valgrind) && defined(WHEN_DEBUGGING)
  MEM_CHECK_DEFINED(buff, block_size);
#endif

  page_cleanup(share, page);
  {
    PAGECACHE_BLOCK_LINK **link;
    enum pagecache_page_pin pin;
    if (lock == PAGECACHE_LOCK_LEFT_WRITELOCKED)
    {
      pin= PAGECACHE_PIN_LEFT_PINNED;
      link= &page_link.link;
    }
    else if (lock == PAGECACHE_LOCK_WRITE_UNLOCK)
    {
      pin= PAGECACHE_UNPIN;
      /*
        We  unlock this page so link should be 0 to prevent it usage
        even accidentally
      */
      link= NULL;
    }
    else
    {
      pin= PAGECACHE_PIN;
      link= &page_link.link;
    }
    res= pagecache_write(share->pagecache,
                         &share->kfile,
                         (pgcache_page_no_t) (page->pos / block_size),
                         level, buff, share->page_type,
                         lock, pin, PAGECACHE_WRITE_DELAY, link,
                         LSN_IMPOSSIBLE);
  }

  if (lock == PAGECACHE_LOCK_WRITE)
  {
    /* It was not locked before, we have to unlock it when we unpin pages */
    page_link.unlock= PAGECACHE_LOCK_WRITE_UNLOCK;
    page_link.changed= 1;
    push_dynamic(&page->info->pinned_pages, (void*) &page_link);
  }
  DBUG_RETURN(res);
}


/**
  @brief Put page in free list

  @fn    _ma_dispose()
  @param info  Maria handle
  @param pos   Address to page
  @param page_not_read  1 if page has not yet been read

  @note
    The page at 'pos' must have been read with a write lock.
    This function does logging (unlike _ma_new()).

  @return
  @retval  0    ok
  @retval  1    error

*/

int _ma_dispose(register MARIA_HA *info, my_off_t pos, my_bool page_not_read)
{
  my_off_t old_link;
  uchar buff[MAX_KEYPAGE_HEADER_SIZE+ 8 + 2];
  ulonglong page_no;
  MARIA_SHARE *share= info->s;
  MARIA_PINNED_PAGE page_link;
  uint block_size= share->block_size;
  int result= 0;
  enum pagecache_page_lock lock_method;
  enum pagecache_page_pin pin_method;
  DBUG_ENTER("_ma_dispose");
  DBUG_PRINT("enter",("page: %lu", (ulong) (pos / block_size)));
  DBUG_ASSERT(pos % block_size == 0);

  (void) _ma_lock_key_del(info, 0);

  old_link= share->key_del_current;
  share->key_del_current= pos;
  page_no= pos / block_size;
  bzero(buff, share->keypage_header);
  _ma_store_keynr(share, buff, (uchar) MARIA_DELETE_KEY_NR);
  _ma_store_page_used(share, buff, share->keypage_header + 8);
  mi_sizestore(buff + share->keypage_header, old_link);
  share->state.changed|= STATE_NOT_SORTED_PAGES;

  if (share->now_transactional)
  {
    LSN lsn;
    uchar log_data[FILEID_STORE_SIZE + PAGE_STORE_SIZE * 2];
    LEX_CUSTRING log_array[TRANSLOG_INTERNAL_PARTS + 1];
    my_off_t page;

    /* Store address of deleted page */
    page_store(log_data + FILEID_STORE_SIZE, page_no);

    /* Store link to next unused page (the link that is written to page) */
    page= (old_link == HA_OFFSET_ERROR ? IMPOSSIBLE_PAGE_NO :
           old_link / block_size);
    page_store(log_data + FILEID_STORE_SIZE + PAGE_STORE_SIZE, page);

    log_array[TRANSLOG_INTERNAL_PARTS + 0].str=    log_data;
    log_array[TRANSLOG_INTERNAL_PARTS + 0].length= sizeof(log_data);

    if (translog_write_record(&lsn, LOGREC_REDO_INDEX_FREE_PAGE,
                              info->trn, info,
                              (translog_size_t) sizeof(log_data),
                              TRANSLOG_INTERNAL_PARTS + 1, log_array,
                              log_data, NULL))
      result= 1;
  }

  if (page_not_read)
  {
    lock_method= PAGECACHE_LOCK_WRITE;
    pin_method= PAGECACHE_PIN;
  }
  else
  {
    lock_method= PAGECACHE_LOCK_LEFT_WRITELOCKED;
    pin_method= PAGECACHE_PIN_LEFT_PINNED;
  }

  if (pagecache_write_part(share->pagecache,
                           &share->kfile, (pgcache_page_no_t) page_no,
                           PAGECACHE_PRIORITY_LOW, buff,
                           share->page_type,
                           lock_method, pin_method,
                           PAGECACHE_WRITE_DELAY, &page_link.link,
      LSN_IMPOSSIBLE,
                           0, share->keypage_header + 8))
    result= 1;

#ifdef IDENTICAL_PAGES_AFTER_RECOVERY
  {
    uchar *page_buff= pagecache_block_link_to_buffer(page_link.link);
    bzero(page_buff + share->keypage_header + 8,
          block_size - share->keypage_header - 8 - KEYPAGE_CHECKSUM_SIZE);
  }
#endif

  if (page_not_read)
  {
    /* It was not locked before, we have to unlock it when we unpin pages */
    page_link.unlock= PAGECACHE_LOCK_WRITE_UNLOCK;
    page_link.changed= 1;
    push_dynamic(&info->pinned_pages, (void*) &page_link);
  }

  DBUG_RETURN(result);
} /* _ma_dispose */


/**
  @brief Get address for free page to use

  @fn     _ma_new()
  @param  info  Maria handle
  @param  level         Type of key block (caching priority for pagecache)
  @param  page_link Pointer to page in page cache if read. One can
                        check if this is used by checking if
                        page_link->changed != 0

  @note Logging of this is left to the caller (so that the "new"ing and the
  first changes done to this new page can be logged as one single entry - one
  single _ma_log_new()) call).

  @return
    HA_OFFSET_ERROR     File is full or page read error or tmp space full
    #          Page address to use
*/

my_off_t _ma_new(register MARIA_HA *info, int level,
                 MARIA_PINNED_PAGE **page_link)

{
  my_off_t pos;
  MARIA_SHARE *share= info->s;
  uint block_size= share->block_size;
  DBUG_ENTER("_ma_new");

  if (_ma_lock_key_del(info, 1))
  {
    mysql_mutex_lock(&share->intern_lock);
    pos= share->state.state.key_file_length;
    if (pos >= share->base.max_key_file_length - block_size)
    {
      my_errno=HA_ERR_INDEX_FILE_FULL;
      mysql_mutex_unlock(&share->intern_lock);
      _ma_unlock_key_del(info);
      DBUG_RETURN(HA_OFFSET_ERROR);
    }
    share->state.state.key_file_length+= block_size;
    if (info->s->tracked &&
        _ma_update_tmp_file_size(&share->track_index,
                                 share->state.state.key_file_length))
    {
      mysql_mutex_unlock(&share->intern_lock);
      _ma_unlock_key_del(info);
      DBUG_RETURN(HA_OFFSET_ERROR);
    }
    /* Following is for not transactional tables */
    info->state->key_file_length= share->state.state.key_file_length;
    mysql_mutex_unlock(&share->intern_lock);
    (*page_link)->changed= 0;
    (*page_link)->write_lock= PAGECACHE_LOCK_WRITE;
  }
  else
  {
    uchar *buff;
    pos= share->key_del_current;                /* Protected */
    DBUG_ASSERT(share->pagecache->block_size == block_size);
    if (!(buff= pagecache_read(share->pagecache,
                               &share->kfile,
                               (pgcache_page_no_t) (pos / block_size), level,
                               0, share->page_type,
                               PAGECACHE_LOCK_WRITE, &(*page_link)->link)))
    {
      pos= HA_OFFSET_ERROR;
      _ma_set_fatal_error(info, my_errno);
    }
    else
    {
      /*
        Next deleted page's number is in the header of the present page
        (single linked list):
      */
#ifdef DBUG_ASSERT_EXISTS
      my_off_t key_del_current;
#endif
      share->key_del_current= mi_sizekorr(buff+share->keypage_header);
#ifdef DBUG_ASSERT_EXISTS
      key_del_current= share->key_del_current;
      DBUG_ASSERT((key_del_current != 0) &&
                  ((key_del_current == HA_OFFSET_ERROR) ||
                   (key_del_current <=
                    (share->state.state.key_file_length - block_size))));
#endif
    }

    (*page_link)->unlock=     PAGECACHE_LOCK_WRITE_UNLOCK;
    (*page_link)->write_lock= PAGECACHE_LOCK_WRITE;
    /*
      We have to mark it changed as _ma_flush_pending_blocks() uses
      'changed' to know if we used the page cache or not
    */
    (*page_link)->changed= 1;
    push_dynamic(&info->pinned_pages, (void*) *page_link);
    *page_link= dynamic_element(&info->pinned_pages,
                                info->pinned_pages.elements-1,
                                MARIA_PINNED_PAGE *);
  }
  share->state.changed|= STATE_NOT_SORTED_PAGES;
  DBUG_PRINT("exit",("Pos: %ld",(long) pos));
  DBUG_RETURN(pos);
} /* _ma_new */


/**
   Log compactation of a index page
*/

static my_bool _ma_log_compact_keypage(MARIA_PAGE *ma_page,
                                       TrID min_read_from)
{
  LSN lsn;
  uchar log_data[FILEID_STORE_SIZE + PAGE_STORE_SIZE + 1 + 7 + TRANSID_SIZE];
  uchar *log_pos;
  LEX_CUSTRING log_array[TRANSLOG_INTERNAL_PARTS + 1];
  MARIA_HA *info= ma_page->info;
  MARIA_SHARE *share= info->s;
  uint translog_parts, extra_length;
  my_off_t page= ma_page->pos;
  DBUG_ENTER("_ma_log_compact_keypage");
  DBUG_PRINT("enter", ("page: %lu", (ulong) (page / share->block_size)));

  /* Store address of new root page */
  page/= share->block_size;
  page_store(log_data + FILEID_STORE_SIZE, page);

  log_pos= log_data + FILEID_STORE_SIZE + PAGE_STORE_SIZE;

  log_pos[0]= KEY_OP_COMPACT_PAGE;
  transid_store(log_pos + 1, min_read_from);
  log_pos+= 1 + TRANSID_SIZE;

  log_array[TRANSLOG_INTERNAL_PARTS + 0].str=    log_data;
  log_array[TRANSLOG_INTERNAL_PARTS + 0].length= (uint) (log_pos -
                                                         log_data);
  translog_parts= 1;
  extra_length= 0;

  _ma_log_key_changes(ma_page,
                      log_array + TRANSLOG_INTERNAL_PARTS + translog_parts,
                      log_pos, &extra_length, &translog_parts);
  /* Remember new page length for future log entires for same page */
  ma_page->org_size= ma_page->size;

  if (translog_write_record(&lsn, LOGREC_REDO_INDEX,
                            info->trn, info,
                            (translog_size_t)(log_array[TRANSLOG_INTERNAL_PARTS +
                                      0].length + extra_length),
                            TRANSLOG_INTERNAL_PARTS + translog_parts,
                            log_array, log_data, NULL))
    DBUG_RETURN(1);
  DBUG_RETURN(0);
}


/**
   Remove all transaction id's less than given one from a key page

   @fn    _ma_compact_keypage()
   @param keyinfo        Key handler
   @param page_pos       Page position on disk
   @param page           Buffer for page
   @param min_read_from  Remove all trids from page less than this

   @retval 0             Ok
   ®retval 1             Error;  my_errno contains the error
*/

my_bool _ma_compact_keypage(MARIA_PAGE *ma_page, TrID min_read_from)
{
  MARIA_HA *info= ma_page->info;
  MARIA_SHARE *share= info->s;
  MARIA_KEY key;
  uchar *page, *endpos, *start_of_empty_space;
  uint page_flag, nod_flag, saved_space;
  my_bool page_has_transid;
  DBUG_ENTER("_ma_compact_keypage");

  page_flag= ma_page->flag;
  if (!(page_flag & KEYPAGE_FLAG_HAS_TRANSID))
    DBUG_RETURN(0);                    /* No transaction id on page */

  nod_flag= ma_page->node;
  page=    ma_page->buff;
  endpos= page + ma_page->size;
  key.data= info->lastkey_buff;
  key.keyinfo= (MARIA_KEYDEF*) ma_page->keyinfo;

  page_has_transid= 0;
  page+= share->keypage_header + nod_flag;
  key.data[0]= 0;                             /* safety */
  start_of_empty_space= 0;
  saved_space= 0;
  do
  {
    if (!(page= (*ma_page->keyinfo->skip_key)(&key, 0, 0, page)))
    {
      DBUG_PRINT("error",("Couldn't find last key:  page_pos: %p",
                          page));
      _ma_set_fatal_error(info, HA_ERR_CRASHED);
      DBUG_RETURN(1);
    }
    if (key_has_transid(page-1))
    {
      uint transid_length;
      transid_length= transid_packed_length(page);

      if (min_read_from == ~(TrID) 0 ||
          min_read_from < transid_get_packed(share, page))
      {
        page[-1]&= 254;                           /* Remove transid marker */
        transid_length= transid_packed_length(page);
        if (start_of_empty_space)
        {
          /* Move block before the transid up in page */
          uint copy_length= (uint) (page - start_of_empty_space) - saved_space;
          memmove(start_of_empty_space, start_of_empty_space + saved_space,
                  copy_length);
          start_of_empty_space+= copy_length;
        }
        else
          start_of_empty_space= page;
        saved_space+= transid_length;
      }
      else
        page_has_transid= 1;                /* At least one id left */
      page+= transid_length;
    }
    page+= nod_flag;
  } while (page < endpos);

  DBUG_ASSERT(page == endpos);

  if (start_of_empty_space)
  {
    /*
      Move last block down
      This is always true if any transid was removed
    */
    uint copy_length= (uint) (endpos - start_of_empty_space) - saved_space;

    if (copy_length)
      memmove(start_of_empty_space, start_of_empty_space + saved_space,
              copy_length);
    ma_page->size= (uint) (start_of_empty_space + copy_length - ma_page->buff);
    page_store_size(share, ma_page);
  }

  if (!page_has_transid)
  {
    ma_page->flag&= ~KEYPAGE_FLAG_HAS_TRANSID;
    _ma_store_keypage_flag(share, ma_page->buff, ma_page->flag);
    /* Clear packed transid (in case of zerofill) */
    bzero(ma_page->buff + LSN_STORE_SIZE, TRANSID_SIZE);
  }

  if (share->now_transactional)
  {
    if (_ma_log_compact_keypage(ma_page, min_read_from))
      DBUG_RETURN(1);
  }
  DBUG_RETURN(0);
}