/*****************************************************************************

Copyright (c) 2007, 2018, Oracle and/or its affiliates. All Rights Reserved.
Copyright (c) 2016, 2022, 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

*****************************************************************************/

/******************************************************************//**
@file fts/fts0opt.cc
Full Text Search optimize thread

Created 2007/03/27 Sunny Bains
Completed 2011/7/10 Sunny and Jimmy Yang

***********************************************************************/

#define MYSQL_SERVER

#include "fts0fts.h"
#include "fts0exec.h"
#include "row0sel.h"
#include "que0types.h"
#include "fts0priv.h"
#include "fts0types.h"
#include "ut0wqueue.h"
#include "srv0start.h"
#include "ut0list.h"
#include "zlib.h"
#include "fts0opt.h"
#include "fts0vlc.h"
#include "log.h"
#include "sql_class.h"

/** The FTS optimize thread's work queue. */
ib_wqueue_t* fts_optimize_wq;
static void fts_optimize_callback(void *);
static void timer_callback(void*);
static tpool::timer* timer;

static tpool::task_group task_group(1);
static tpool::waitable_task task(fts_optimize_callback,0, &task_group);

/** FTS optimize thread, for MDL acquisition */
static THD *fts_opt_thd;

/** The FTS vector to store fts_slot_t */
static ib_vector_t*  fts_slots;

/** Default optimize interval in secs. */
static const ulint FTS_OPTIMIZE_INTERVAL_IN_SECS = 300;

/** Server is shutting down, so does we exiting the optimize thread */
static bool fts_opt_start_shutdown = false;

/** Condition variable for shutting down the optimize thread.
Protected by fts_optimize_wq->mutex. */
static pthread_cond_t fts_opt_shutdown_cond;

/** Last time we did check whether system need a sync */
static time_t last_check_sync_time;

/** FTS optimize thread message types. */
enum fts_msg_type_t {
 FTS_MSG_STOP,   /*!< Stop optimizing and exit thread */

 FTS_MSG_ADD_TABLE,  /*!< Add table to the optimize thread's
work queue */

 FTS_MSG_DEL_TABLE,  /*!< Remove a table from the optimize
threads work queue */
 FTS_MSG_SYNC_TABLE  /*!< Sync fts cache of a table */
};

/** Compressed list of words that have been read from FTS INDEX
that needs to be optimized. */
struct fts_zip_t {
 lint  status;  /*!< Status of (un)/zip operation */

 ulint  n_words; /*!< Number of words compressed */

 ulint  block_sz; /*!< Size of a block in bytes */

 ib_vector_t* blocks;  /*!< Vector of compressed blocks */

 ib_alloc_t* heap_alloc; /*!< Heap to use for allocations */

 ulint  pos;  /*!< Offset into blocks */

 ulint  last_big_block; /*!< Offset of last block in the
     blocks array that is of size
     block_sz. Blocks beyond this offset
are of size FTS_MAX_WORD_LEN */

 z_streamp zp;  /*!< ZLib state */

     /*!< The value of the last word read
     from the FTS INDEX table. This is
used to discard duplicates */

 fts_string_t word;  /*!< UTF-8 string */

 ulint  max_words; /*!< maximum number of words to read
in one pass */
};

/** Prepared statements used during optimize */
struct fts_optimize_graph_t {
     /*!< Delete a word from FTS INDEX */
 que_t*  delete_nodes_graph;
     /*!< Insert a word into FTS INDEX */
 que_t*  write_nodes_graph;
     /*!< COMMIT a transaction */
 que_t*  commit_graph;
     /*!< Read the nodes from FTS_INDEX */
 que_t*  read_nodes_graph;
};

/** Used by fts_optimize() to store state. */
struct fts_optimize_t {
 trx_t*  trx;  /*!< The transaction used for all SQL */

 ib_alloc_t* self_heap; /*!< Heap to use for allocations */

 char*  name_prefix; /*!< FTS table name prefix */

 dict_table_t* table;  /*!< Table that has to be queried */

 dict_index_t* index;  /*!< The FTS index to be optimized */

 fts_doc_ids_t* to_delete; /*!< doc ids to delete, we check against
     this vector and purge the matching
     entries during the optimizing
     process. The vector entries are
sorted on doc id */

 ulint  del_pos; /*!< Offset within to_delete vector,
     this is used to keep track of where
we are up to in the vector */

 ibool  done;  /*!< TRUE when optimize finishes */

 ib_vector_t* words;  /*!< Word + Nodes read from FTS_INDEX,
it contains instances of fts_word_t */

 fts_zip_t* zip;  /*!< Words read from the FTS_INDEX */

 fts_optimize_graph_t  /*!< Prepared statements used during */
   graph;  /*optimize */

 ulint  n_completed; /*!< Number of FTS indexes that have
been optimized */
 ibool  del_list_regenerated;
     /*!< BEING_DELETED list regenerated */
};

/** Used by the optimize, to keep state during compacting nodes. */
struct fts_encode_t {
 doc_id_t src_last_doc_id;/*!< Last doc id read from src node */
 byte*  src_ilist_ptr; /*!< Current ptr within src ilist */
};

/** We use this information to determine when to start the optimize
cycle for a table. */
struct fts_slot_t {
 /** table, or NULL if the slot is unused */
 dict_table_t* table;

 /** whether this slot is being processed */
 bool  running;

 ulint  added;  /*!< Number of doc ids added since the
last time this table was optimized */

 ulint  deleted; /*!< Number of doc ids deleted since the
last time this table was optimized */

 /** time(NULL) of completing fts_optimize_table_bk() */
 time_t  last_run;

 /** time(NULL) of latest successful fts_optimize_table() */
 time_t  completed;
};

/** A table remove message for the FTS optimize thread. */
struct fts_msg_del_t
{
  /** the table to remove */
  dict_table_t *table;
  /** condition variable to signal message consumption */
  pthread_cond_t *cond;
};

/** The FTS optimize message work queue message type. */
struct fts_msg_t {
 fts_msg_type_t type;  /*!< Message type */

 void*  ptr;  /*!< The message contents */

 mem_heap_t* heap;  /*!< The heap used to allocate this
     message, the message consumer will
free the heap. */
};

/** The number of words to read and optimize in a single pass. */
ulong fts_num_word_optimize;

/** ZLib compressed block size.*/
static constexpr ulint FTS_ZIP_BLOCK_SIZE = 1024;

/** The amount of time optimizing in a single pass, in seconds. */
static ulint fts_optimize_time_limit;

/** It's defined in fts0fts.cc  */
extern const char* fts_common_tables[];

/**********************************************************************//**
Initialize fts_zip_t. */
static
void
fts_zip_initialize(
/*===============*/
 fts_zip_t* zip)  /*!< out: zip instance to initialize */
{
 zip->pos = 0;
 zip->n_words = 0;

 zip->status = Z_OK;

 zip->last_big_block = 0;

 zip->word.f_len = 0;
 *zip->word.f_str = 0;

 ib_vector_reset(zip->blocks);

 memset(zip->zp, 0, sizeof(*zip->zp));
}

/**********************************************************************//**
Create an instance of fts_zip_t.
@return a new instance of fts_zip_t */
static
fts_zip_t*
fts_zip_create(
/*===========*/
 mem_heap_t* heap,  /*!< in: heap */
 ulint  block_sz, /*!< in: size of a zip block.*/
 ulint  max_words) /*!< in: max words to read */
{
 fts_zip_t* zip;

 zip = static_cast<fts_zip_t*>(mem_heap_zalloc(heap, sizeof(*zip)));

 zip->word.f_str = static_cast<byte*>(
  mem_heap_zalloc(heap, FTS_MAX_WORD_LEN + 1));

 zip->block_sz = block_sz;

 zip->heap_alloc = ib_heap_allocator_create(heap);

 zip->blocks = ib_vector_create(zip->heap_alloc, sizeof(void*), 128);

 zip->max_words = max_words;

 zip->zp = static_cast<z_stream*>(
  mem_heap_zalloc(heap, sizeof(*zip->zp)));

 return(zip);
}

/**********************************************************************//**
Initialize an instance of fts_zip_t. */
static
void
fts_zip_init(
/*=========*/

 fts_zip_t* zip)  /*!< in: zip instance to init */
{
 memset(zip->zp, 0, sizeof(*zip->zp));

 zip->word.f_len = 0;
 *zip->word.f_str = '\0';
}

dberr_t fts_index_fetch_nodes(FTSQueryExecutor *executor, dict_index_t *index,
                              const fts_string_t *word, void *user_arg,
                              FTSRecordProcessor processor,
                              AuxCompareMode compare_mode) noexcept
{
  dberr_t error= DB_SUCCESS;
  trx_t *trx= executor->trx();
  trx->op_info= "fetching FTS index nodes";
  CHARSET_INFO *cs= fts_index_get_charset(index);
  uint8_t selected= fts_select_index(cs, word->f_str, word->f_len);
  ulint total_memory= 0;
  for (;;)
  {
    AuxRecordReader reader= processor
   ? AuxRecordReader(user_arg, processor, compare_mode)
   : AuxRecordReader(user_arg, &total_memory, compare_mode);
    error= executor->read_aux(selected, word, reader);
    if (UNIV_LIKELY(error == DB_SUCCESS || error == DB_RECORD_NOT_FOUND))
    {
      if (error == DB_RECORD_NOT_FOUND) error = DB_SUCCESS;
      fts_sql_commit(trx);
      break;
    }
    else
    {
      fts_sql_rollback(trx);
      if (error == DB_LOCK_WAIT_TIMEOUT)
      {
        sql_print_warning("InnoDB: Lock wait timeout reading FTS index."
                          "Retrying!");
        trx->error_state= DB_SUCCESS;
      }
      else
      {
        sql_print_error("InnoDB: (%s) while reading FTS index.",
                        ut_strerr(error));
        break;
      }
    }
  }
  return error;
}

/**********************************************************************//**
Read a word */
static
byte*
fts_zip_read_word(
/*==============*/
 fts_zip_t* zip,  /*!< in: Zip state + data */
 fts_string_t* word)  /*!< out: uncompressed word */
{
 short  len = 0;
 void*  null = NULL;
 byte*  ptr = word->f_str;
 int  flush = Z_NO_FLUSH;

 /* Either there was an error or we are at the Z_STREAM_END. */
 if (zip->status != Z_OK) {
  return(NULL);
 }

 zip->zp->next_out = reinterpret_cast<byte*>(&len);
 zip->zp->avail_out = sizeof(len);

 while (zip->status == Z_OK && zip->zp->avail_out > 0) {

  /* Finished decompressing block. */
  if (zip->zp->avail_in == 0) {

   /* Free the block that's been decompressed. */
   if (zip->pos > 0) {
    ulint prev = zip->pos - 1;

    ut_a(zip->pos < ib_vector_size(zip->blocks));

    ut_free(ib_vector_getp(zip->blocks, prev));
    ib_vector_set(zip->blocks, prev, &null);
   }

   /* Any more blocks to decompress. */
   if (zip->pos < ib_vector_size(zip->blocks)) {

    zip->zp->next_in = static_cast<byte*>(
     ib_vector_getp(
      zip->blocks, zip->pos));

    if (zip->pos > zip->last_big_block) {
     zip->zp->avail_in =
      FTS_MAX_WORD_LEN;
    } else {
     zip->zp->avail_in =
      static_cast<uInt>(zip->block_sz);
    }

    ++zip->pos;
   } else {
    flush = Z_FINISH;
   }
  }

  switch (zip->status = inflate(zip->zp, flush)) {
  case Z_OK:
   if (zip->zp->avail_out == 0 && len > 0) {

    ut_a(len <= FTS_MAX_WORD_LEN);
    ptr[len] = 0;

    zip->zp->next_out = ptr;
    zip->zp->avail_out = uInt(len);

    word->f_len = ulint(len);
    len = 0;
   }
   break;

  case Z_BUF_ERROR: /* No progress possible. */
  case Z_STREAM_END:
   inflateEnd(zip->zp);
   break;

  case Z_STREAM_ERROR:
  default:
   ut_error;
  }
 }

 /* All blocks must be freed at end of inflate. */
 if (zip->status != Z_OK) {
  for (ulint i = 0; i < ib_vector_size(zip->blocks); ++i) {
   if (ib_vector_getp(zip->blocks, i)) {
    ut_free(ib_vector_getp(zip->blocks, i));
    ib_vector_set(zip->blocks, i, &null);
   }
  }
 }

 if (ptr != NULL) {
  ut_ad(word->f_len == strlen((char*) ptr));
 }

 return(zip->status == Z_OK || zip->status == Z_STREAM_END ? ptr : NULL);
}

/**********************************************************************//**
Finish Zip deflate. */
static
void
fts_zip_deflate_end(
/*================*/
 fts_zip_t* zip)  /*!< in: instance that should be closed*/
{
 ut_a(zip->zp->avail_in == 0);
 ut_a(zip->zp->next_in == NULL);

 zip->status = deflate(zip->zp, Z_FINISH);

 ut_a(ib_vector_size(zip->blocks) > 0);
 zip->last_big_block = ib_vector_size(zip->blocks) - 1;

 /* Allocate smaller block(s), since this is trailing data. */
 while (zip->status == Z_OK) {
  byte*  block;

  ut_a(zip->zp->avail_out == 0);

  block = static_cast<byte*>(
   ut_malloc_nokey(FTS_MAX_WORD_LEN + 1));

  ib_vector_push(zip->blocks, &block);

  zip->zp->next_out = block;
  zip->zp->avail_out = FTS_MAX_WORD_LEN;

  zip->status = deflate(zip->zp, Z_FINISH);
 }

 ut_a(zip->status == Z_STREAM_END);

 zip->status = deflateEnd(zip->zp);
 ut_a(zip->status == Z_OK);

 /* Reset the ZLib data structure. */
 memset(zip->zp, 0, sizeof(*zip->zp));
}

/** Read the words from the FTS INDEX.
@param executor query executor
@param optim optimize scratch pad
@param word  get words gerater than this
@param n_words max words to read
@return DB_SUCCESS if all OK, DB_TABLE_NOT_FOUND if no more indexes
        to search else error code */
static MY_ATTRIBUTE((nonnull, warn_unused_result))
dberr_t fts_index_fetch_words(FTSQueryExecutor *executor,
                              fts_optimize_t *optim,
                              const fts_string_t *word,
                              ulint n_words) noexcept
{
  dberr_t error= DB_SUCCESS;
  mem_heap_t *heap= static_cast<mem_heap_t*>(optim->self_heap->arg);
  optim->trx->op_info= "fetching FTS index words";

  if (optim->zip == NULL)
    optim->zip = fts_zip_create(heap, FTS_ZIP_BLOCK_SIZE, n_words);
  else fts_zip_initialize(optim->zip);

  CHARSET_INFO *cs= fts_index_get_charset(optim->index);

  /* Create compression processor with state */
  bool compress_inited = false;
  auto compress_processor= [&compress_inited](
    const rec_t *rec, const dict_index_t *index,
    const rec_offs *offsets, void *user_arg) -> dberr_t
  {
    fts_zip_t* zip= static_cast<fts_zip_t*>(user_arg);

    /* Extract word field using rec_get_nth_field() */
    ulint word_len;
    const byte* word_data= rec_get_nth_field(rec, offsets, 0, &word_len);

    if (!word_data || word_len == UNIV_SQL_NULL || word_len > FTS_MAX_WORD_LEN)
      return DB_SUCCESS;

    /* Skip duplicate words */
    if (zip->word.f_len == word_len &&
        !memcmp(zip->word.f_str, word_data, word_len))
      return DB_SUCCESS;

    /* Initialize deflate if not done yet */
    if (!compress_inited)
    {
      int err = deflateInit(zip->zp, 9);
      if (err != Z_OK)
      {
        sql_print_error("InnoDB: ZLib deflateInit() failed: %d", err);
        return DB_ERROR;
      }
      compress_inited = true;
    }

    /* Update current word */
    memcpy(zip->word.f_str, word_data, word_len);
    zip->word.f_len = word_len;
    ut_a(zip->zp->avail_in == 0);
    ut_a(zip->zp->next_in == NULL);

    /* Compress the word with length prefix */
    uint16_t len = static_cast<uint16_t>(word_len);
    zip->zp->next_in = reinterpret_cast<byte*>(&len);
    zip->zp->avail_in = sizeof(len);

    /* Compress the word, create output blocks as necessary */
    while (zip->zp->avail_in > 0)
    {
      /* No space left in output buffer, create a new one */
      if (zip->zp->avail_out == 0)
      {
        byte* block= static_cast<byte*>(ut_malloc_nokey(zip->block_sz));
        ib_vector_push(zip->blocks, &block);
        zip->zp->next_out= block;
        zip->zp->avail_out= static_cast<uInt>(zip->block_sz);
      }

      switch (zip->status = deflate(zip->zp, Z_NO_FLUSH))
      {
      case Z_OK:
        if (zip->zp->avail_in == 0)
        {
          zip->zp->next_in= const_cast<byte*>(word_data);
          zip->zp->avail_in = static_cast<uInt>(len);
          ut_a(len <= FTS_MAX_WORD_LEN);
          len = 0;
        }
        continue;
      case Z_STREAM_END:
      case Z_BUF_ERROR:
      case Z_STREAM_ERROR:
      default:
        ut_error;
      }
    }

    /* All data should have been compressed */
    ut_a(zip->zp->avail_in == 0);
    zip->zp->next_in = NULL;

    ++zip->n_words;

    /* Continue until we reach max words */
    return zip->n_words < zip->max_words ? DB_SUCCESS : DB_SUCCESS_LOCKED_REC;
  };

  for (uint8_t selected= fts_select_index(cs, word->f_str, word->f_len);
       selected < FTS_NUM_AUX_INDEX; selected++)
  {
    for (;;)
    {
      AuxRecordReader aux_reader(optim->zip, compress_processor,
                                 AuxCompareMode::GREATER);

      error= executor->read_aux(selected, word, aux_reader);
      if (UNIV_LIKELY(error == DB_SUCCESS || error == DB_RECORD_NOT_FOUND)) {
        if (error == DB_RECORD_NOT_FOUND) error = DB_SUCCESS;
        break;
      }
      else
      {
        if (error == DB_LOCK_WAIT_TIMEOUT)
        {
          sql_print_warning("InnoDB: Lock wait timeout reading "
                            "words. Retrying!");
          if (compress_inited)
          {
            deflateEnd(optim->zip->zp);
            fts_zip_init(optim->zip);
            compress_inited= false;
          }
          optim->trx->error_state = DB_SUCCESS;
        }
        else
        {
          sql_print_error("InnoDB: (%s) while reading words.",
                          ut_strerr(error));
          break;
        }
      }
    }

    if (optim->zip->n_words >= n_words) break;
  }

  fts_zip_t *zip = optim->zip;
  if (error == DB_SUCCESS && zip->status == Z_OK && zip->n_words > 0)
  {
    /* All data should have been read */
    ut_a(zip->zp->avail_in == 0);
    fts_zip_deflate_end(zip);
  }
  else deflateEnd(zip->zp);

  return error;
}

dberr_t fts_table_fetch_doc_ids(FTSQueryExecutor *executor,
                                const char *tbl_name,
                                fts_doc_ids_t *doc_ids) noexcept
{
  ut_ad(executor != nullptr);
  executor->trx()->op_info = "fetching FTS doc ids";
  /* Append doc_ids straight into the caller's vector rather than
  buffering the whole set in the reader first. */
  CommonTableReader reader(doc_ids->doc_ids);
  dberr_t err= executor->read_all_common(tbl_name, reader);

  if (err == DB_SUCCESS)
    fts_doc_ids_sort(doc_ids->doc_ids);

  return err;
}

/**********************************************************************//**
Do a binary search for a doc id in the array
@return +ve index if found -ve index where it should be inserted
        if not found */
int
fts_bsearch(
/*========*/
 doc_id_t* array, /*!< in: array to sort */
 int  lower, /*!< in: the array lower bound */
 int  upper, /*!< in: the array upper bound */
 doc_id_t doc_id) /*!< in: the doc id to search for */
{
 int orig_size = upper;

 if (upper == 0) {
  /* Nothing to search */
  return(-1);
 } else {
  while (lower < upper) {
   int i = (lower + upper) >> 1;

   if (doc_id > array[i]) {
    lower = i + 1;
   } else if (doc_id < array[i]) {
    upper = i - 1;
   } else {
    return(i); /* Found. */
   }
  }
 }

 if (lower == upper && lower < orig_size) {
  if (doc_id == array[lower]) {
   return(lower);
  } else if (lower == 0) {
   return(-1);
  }
 }

 /* Not found. */
 return( (lower == 0) ? -1 : -(lower));
}

/**********************************************************************//**
Search in the to delete array whether any of the doc ids within
the [first, last] range are to be deleted
@return +ve index if found -ve index where it should be inserted
        if not found */
static
int
fts_optimize_lookup(
/*================*/
 ib_vector_t* doc_ids, /*!< in: array to search */
 ulint  lower,  /*!< in: lower limit of array */
 doc_id_t first_doc_id, /*!< in: doc id to lookup */
 doc_id_t last_doc_id) /*!< in: doc id to lookup */
{
 int  pos;
 int  upper = static_cast<int>(ib_vector_size(doc_ids));
 doc_id_t* array = (doc_id_t*) doc_ids->data;

 pos = fts_bsearch(array, static_cast<int>(lower), upper, first_doc_id);

 ut_a(abs(pos) <= upper + 1);

 if (pos < 0) {

  int i = abs(pos);

  /* If i is 1, it could be first_doc_id is less than
  either the first or second array item, do a
double check */
  if (i == 1 && array[0] <= last_doc_id
      && first_doc_id < array[0]) {
   pos = 0;
  } else if (i < upper && array[i] <= last_doc_id) {

   /* Check if the "next" doc id is within the
first & last doc id of the node. */
   pos = i;
  }
 }

 return(pos);
}

/**********************************************************************//**
Encode the word pos list into the node
@return DB_SUCCESS or error code*/
static MY_ATTRIBUTE((nonnull))
dberr_t
fts_optimize_encode_node(
/*=====================*/
 fts_node_t* node,  /*!< in: node to fill*/
 doc_id_t doc_id,  /*!< in: doc id to encode */
 fts_encode_t* enc)  /*!< in: encoding state.*/
{
 byte*  dst;
 ulint  enc_len;
 ulint  pos_enc_len;
 doc_id_t doc_id_delta;
 dberr_t  error = DB_SUCCESS;
 const byte* src = enc->src_ilist_ptr;

 if (node->first_doc_id == 0) {
  ut_a(node->last_doc_id == 0);

  node->first_doc_id = doc_id;
 }

 /* Calculate the space required to store the ilist. */
 ut_ad(doc_id > node->last_doc_id);
 doc_id_delta = doc_id - node->last_doc_id;
 enc_len = fts_get_encoded_len(static_cast<ulint>(doc_id_delta));

 /* Calculate the size of the encoded pos array. */
 while (*src) {
  fts_decode_vlc(&src);
 }

 /* Skip the 0x00 byte at the end of the word positions list. */
 ++src;

 /* Number of encoded pos bytes to copy. */
 pos_enc_len = ulint(src - enc->src_ilist_ptr);

 /* Total number of bytes required for copy. */
 enc_len += pos_enc_len;

 /* Check we have enough space in the destination buffer for
copying the document word list. */
 if (!node->ilist) {
  ulint new_size;

  ut_a(node->ilist_size == 0);

  new_size = enc_len > FTS_ILIST_MAX_SIZE
   ? enc_len : FTS_ILIST_MAX_SIZE;

  node->ilist = static_cast<byte*>(ut_malloc_nokey(new_size));
  node->ilist_size_alloc = new_size;

 } else if ((node->ilist_size + enc_len) > node->ilist_size_alloc) {
  ulint new_size = node->ilist_size + enc_len;
  byte* ilist = static_cast<byte*>(ut_malloc_nokey(new_size));

  memcpy(ilist, node->ilist, node->ilist_size);

  ut_free(node->ilist);

  node->ilist = ilist;
  node->ilist_size_alloc = new_size;
 }

 src = enc->src_ilist_ptr;
 dst = node->ilist + node->ilist_size;

 /* Encode the doc id. Cast to ulint, the delta should be small and
therefore no loss of precision. */
 dst = fts_encode_int(doc_id_delta, dst);

 /* Copy the encoded pos array. */
 memcpy(dst, src, pos_enc_len);

 node->last_doc_id = doc_id;

 /* Data copied upto here. */
 node->ilist_size += enc_len;
 enc->src_ilist_ptr += pos_enc_len;

 ut_a(node->ilist_size <= node->ilist_size_alloc);

 return(error);
}

/**********************************************************************//**
Optimize the data contained in a node.
@return DB_SUCCESS or error code*/
static MY_ATTRIBUTE((nonnull))
dberr_t
fts_optimize_node(
/*==============*/
 ib_vector_t* del_vec, /*!< in: vector of doc ids to delete*/
 int*  del_pos, /*!< in: offset into above vector */
 fts_node_t* dst_node, /*!< in: node to fill*/
 fts_node_t* src_node, /*!< in: source node for data*/
 fts_encode_t* enc)  /*!< in: encoding state */
{
 ulint  copied;
 dberr_t  error = DB_SUCCESS;
 doc_id_t doc_id = enc->src_last_doc_id;

 if (!enc->src_ilist_ptr) {
  enc->src_ilist_ptr = src_node->ilist;
 }

 copied = ulint(enc->src_ilist_ptr - src_node->ilist);

 /* While there is data in the source node and space to copy
into in the destination node. */
 while (copied < src_node->ilist_size
        && dst_node->ilist_size < FTS_ILIST_MAX_SIZE) {

  doc_id_t delta;
  doc_id_t del_doc_id = FTS_NULL_DOC_ID;

  delta = fts_decode_vlc(
   (const byte**)&enc->src_ilist_ptr);

test_again:
  /* Check whether the doc id is in the delete list, if
  so then we skip the entries but we need to track the
  delta for decoding the entries following this document's
entries. */
  if (*del_pos >= 0 && *del_pos < (int) ib_vector_size(del_vec)) {
   doc_id_t* update;

   update = (doc_id_t*) ib_vector_get(
    del_vec, ulint(*del_pos));

   del_doc_id = *update;
  }

  if (enc->src_ilist_ptr == src_node->ilist && doc_id == 0) {
   ut_a(delta == src_node->first_doc_id);
  }

  doc_id += delta;

  if (del_doc_id > 0 && doc_id == del_doc_id) {

   ++*del_pos;

   /* Skip the entries for this document. */
   while (*enc->src_ilist_ptr) {
    fts_decode_vlc((const byte**)&enc->src_ilist_ptr);
   }

   /* Skip the end of word position marker. */
   ++enc->src_ilist_ptr;

  } else {

   /* DOC ID already becomes larger than
del_doc_id, check the next del_doc_id */
   if (del_doc_id > 0 && doc_id > del_doc_id) {
    del_doc_id = 0;
    ++*del_pos;
    delta = 0;
    goto test_again;
   }

   /* Decode and copy the word positions into
the dest node. */
   fts_optimize_encode_node(dst_node, doc_id, enc);

   ++dst_node->doc_count;

   ut_a(dst_node->last_doc_id == doc_id);
  }

  /* Bytes copied so for from source. */
  copied = ulint(enc->src_ilist_ptr - src_node->ilist);
 }

 if (copied >= src_node->ilist_size) {
  ut_a(doc_id == src_node->last_doc_id);
 }

 enc->src_last_doc_id = doc_id;

 return(error);
}

/**********************************************************************//**
Determine the starting pos within the deleted doc id vector for a word.
@return delete position */
static MY_ATTRIBUTE((nonnull, warn_unused_result))
int
fts_optimize_deleted_pos(
/*=====================*/
 fts_optimize_t* optim,  /*!< in: optimize state data */
 fts_word_t* word)  /*!< in: the word data to check */
{
 int  del_pos;
 ib_vector_t* del_vec = optim->to_delete->doc_ids;

 /* Get the first and last dict ids for the word, we will use
 these values to determine which doc ids need to be removed
 when we coalesce the nodes. This way we can reduce the numer
 of elements that need to be searched in the deleted doc ids
 vector and secondly we can remove the doc ids during the
coalescing phase. */
 if (ib_vector_size(del_vec) > 0) {
  fts_node_t* node;
  doc_id_t last_id;
  doc_id_t first_id;
  ulint  size = ib_vector_size(word->nodes);

  node = (fts_node_t*) ib_vector_get(word->nodes, 0);
  first_id = node->first_doc_id;

  node = (fts_node_t*) ib_vector_get(word->nodes, size - 1);
  last_id = node->last_doc_id;

  ut_a(first_id <= last_id);

  del_pos = fts_optimize_lookup(
   del_vec, optim->del_pos, first_id, last_id);
 } else {

  del_pos = -1; /* Note that there is nothing to delete. */
 }

 return(del_pos);
}

/**********************************************************************//**
Compact the nodes for a word, we also remove any doc ids during the
compaction pass.
@return DB_SUCCESS or error code.*/
static
ib_vector_t*
fts_optimize_word(
/*==============*/
 fts_optimize_t* optim,  /*!< in: optimize state data */
 fts_word_t* word)  /*!< in: the word to optimize */
{
 fts_encode_t enc;
 ib_vector_t* nodes;
 ulint  i = 0;
 int  del_pos;
 fts_node_t* dst_node = NULL;
 ib_vector_t* del_vec = optim->to_delete->doc_ids;
 ulint  size = ib_vector_size(word->nodes);

 del_pos = fts_optimize_deleted_pos(optim, word);
 nodes = ib_vector_create(word->heap_alloc, sizeof(*dst_node), 128);

 enc.src_last_doc_id = 0;
 enc.src_ilist_ptr = NULL;

 while (i < size) {
  ulint  copied;
  fts_node_t* src_node;

  src_node = (fts_node_t*) ib_vector_get(word->nodes, i);

  if (dst_node == NULL
      || dst_node->last_doc_id >= src_node->first_doc_id) {

   dst_node = static_cast<fts_node_t*>(
    ib_vector_push(nodes, NULL));
   memset(dst_node, 0, sizeof(*dst_node));
  }

  /* Copy from the src to the dst node. */
  fts_optimize_node(del_vec, &del_pos, dst_node, src_node, &enc);

  ut_a(enc.src_ilist_ptr != NULL);

  /* Determine the numer of bytes copied to dst_node. */
  copied = ulint(enc.src_ilist_ptr - src_node->ilist);

  /* Can't copy more than whats in the vlc array. */
  ut_a(copied <= src_node->ilist_size);

  /* We are done with this node release the resources. */
  if (copied == src_node->ilist_size) {

   enc.src_last_doc_id = 0;
   enc.src_ilist_ptr = NULL;

   ut_free(src_node->ilist);

   src_node->ilist = NULL;
   src_node->ilist_size = src_node->ilist_size_alloc = 0;

   src_node = NULL;

   ++i; /* Get next source node to OPTIMIZE. */
  }

  if (dst_node->ilist_size >= FTS_ILIST_MAX_SIZE || i >= size) {

   dst_node = NULL;
  }
 }

 /* All dst nodes created should have been added to the vector. */
 ut_a(dst_node == NULL);

 /* Return the OPTIMIZED nodes. */
 return(nodes);
}

/** Write the words and ilist to disk.
@param executor query executor
@param index fulltext index
@param word word to update
@param nodes nodes to update
@return DB_SUCCESS or error code */
static MY_ATTRIBUTE((nonnull, warn_unused_result))
dberr_t fts_optimize_write_word(FTSQueryExecutor *executor, dict_index_t *index,
                                fts_string_t *word, ib_vector_t *nodes) noexcept
{
  CHARSET_INFO *cs= fts_index_get_charset(index);
  uint8_t selected= fts_select_index(cs, word->f_str, word->f_len);
  fts_aux_data_t aux_data((const char*)word->f_str, word->f_len);
  dberr_t err= executor->delete_aux_record(selected, &aux_data);
  if (err != DB_SUCCESS)
  {
    sql_print_error("InnoDB: (%s) during optimize, when "
                    "deleting a word from the FTS index.",
                    ut_strerr(err));
    return err;
  }

  for (ulint i = 0; i < ib_vector_size(nodes); ++i)
  {
    fts_node_t* node = (fts_node_t*) ib_vector_get(nodes, i);
    if (!node->ilist || node->ilist_size == 0) continue;

    fts_aux_data_t insert_data(
      (const char*)word->f_str, word->f_len,
      node->first_doc_id, node->last_doc_id,
      static_cast<uint32_t>(node->doc_count), node->ilist,
      node->ilist_size);

    err = executor->insert_aux_record(selected, &insert_data);
    if (err != DB_SUCCESS)
    {
      sql_print_error("InnoDB: (%s) during optimize, when "
                      "inserting a word to the FTS index.",
                      ut_strerr(err));
      return err;
    }
    ut_free(node->ilist);
    node->ilist= nullptr;
    node->ilist_size= node->ilist_size_alloc= 0;
  }

  return DB_SUCCESS;
}

/**********************************************************************//**
Free fts_optimizer_word_t instanace.*/
void
fts_word_free(
/*==========*/
 fts_word_t* word)  /*!< in: instance to free.*/
{
 mem_heap_t* heap = static_cast<mem_heap_t*>(word->heap_alloc->arg);

#ifdef UNIV_DEBUG
 memset(word, 0, sizeof(*word));
#endif /* UNIV_DEBUG */

 if (heap) {
  mem_heap_free(heap);
 }
}

/**********************************************************************//**
Compact the nodes for a given word, the nodes passed in are
already optimized.
@return status one of RESTART, EXIT, ERROR */
static MY_ATTRIBUTE((nonnull, warn_unused_result))
dberr_t
fts_optimize_compact(
/*=================*/
 FTSQueryExecutor* executor, /*!< in: query executor */
 fts_optimize_t* optim,  /*!< in: optimize state data */
 dict_index_t* index,  /*!< in: current FTS being optimized */
 time_t  start_time) noexcept /*!< in: optimize start time */
{
 ulint  i;
 dberr_t  error = DB_SUCCESS;
 ulint  size = ib_vector_size(optim->words);

 for (i = 0; i < size && error == DB_SUCCESS && !optim->done; ++i) {
  fts_word_t* word;
  ib_vector_t* nodes;

  word = (fts_word_t*) ib_vector_get(optim->words, i);

  /* nodes is allocated from the word heap and will be destroyed
  when the word is freed. We however have to be careful about
the ilist, that needs to be freed explicitly. */
  nodes = fts_optimize_word(optim, word);

  /* Update the data on disk. */
  error = fts_optimize_write_word(
   executor, index, &word->text, nodes);

  if (error == DB_SUCCESS) {
   /* Write the last word optimized to the config table,
we use this value for restarting optimize. */
   error = fts_config_set_index_value(
    executor, index,
    FTS_LAST_OPTIMIZED_WORD, &word->text);
  }

  /* Free the word that was optimized. */
  fts_word_free(word);

  ulint interval = ulint(time(NULL) - start_time);

  if (fts_optimize_time_limit > 0
      && (lint(interval) < 0
   || interval > fts_optimize_time_limit)) {

   optim->done = TRUE;
  }
 }

 return(error);
}

/**********************************************************************//**
Create an instance of fts_optimize_t. Also create a new
background transaction.*/
static
fts_optimize_t*
fts_optimize_create(
/*================*/
 dict_table_t* table)  /*!< in: table with FTS indexes */
{
 fts_optimize_t* optim;
 mem_heap_t* heap = mem_heap_create(128);

 optim = (fts_optimize_t*) mem_heap_zalloc(heap, sizeof(*optim));

 optim->self_heap = ib_heap_allocator_create(heap);

 optim->to_delete = fts_doc_ids_create();

 optim->words = ib_vector_create(
  optim->self_heap, sizeof(fts_word_t), 256);

 optim->table = table;

 optim->trx = trx_create();
 trx_start_internal(optim->trx);

 /* The common prefix for all this parent table's aux tables. */
 char table_id[FTS_AUX_MIN_TABLE_ID_LENGTH];
 const size_t table_id_len = 1
  + size_t(fts_write_object_id(table->id, table_id,
                                             FTS_AUX_MIN_TABLE_ID_LENGTH));
 dict_sys.freeze(SRW_LOCK_CALL);
 /* Include the separator as well. */
 const size_t dbname_len = table->name.dblen() + 1;
 ut_ad(dbname_len > 1);
 const size_t prefix_name_len = dbname_len + 4 + table_id_len;
 char* prefix_name = static_cast<char*>(
  ut_malloc_nokey(prefix_name_len));
 memcpy(prefix_name, table->name.m_name, dbname_len);
 dict_sys.unfreeze();
 memcpy(prefix_name + dbname_len, "FTS_", 4);
 memcpy(prefix_name + dbname_len + 4, table_id, table_id_len);
 optim->name_prefix =prefix_name;

 return(optim);
}

/**********************************************************************//**
Free the optimize prepared statements.*/
static
void
fts_optimize_graph_free(
/*====================*/
 fts_optimize_graph_t* graph) /*!< in/out: The graph instances
to free */
{
 if (graph->commit_graph) {
  que_graph_free(graph->commit_graph);
  graph->commit_graph = NULL;
 }

 if (graph->write_nodes_graph) {
  que_graph_free(graph->write_nodes_graph);
  graph->write_nodes_graph = NULL;
 }

 if (graph->delete_nodes_graph) {
  que_graph_free(graph->delete_nodes_graph);
  graph->delete_nodes_graph = NULL;
 }

 if (graph->read_nodes_graph) {
  que_graph_free(graph->read_nodes_graph);
  graph->read_nodes_graph = NULL;
 }
}

/**********************************************************************//**
Free all optimize resources. */
static
void
fts_optimize_free(
/*==============*/
 fts_optimize_t* optim)  /*!< in: table with on FTS index */
{
 mem_heap_t* heap = static_cast<mem_heap_t*>(optim->self_heap->arg);

 trx_commit_for_mysql(optim->trx);
 optim->trx->clear_and_free();
 optim->trx = NULL;

 fts_doc_ids_free(optim->to_delete);
 fts_optimize_graph_free(&optim->graph);

 ut_free(optim->name_prefix);

 /* This will free the heap from which optim itself was allocated. */
 mem_heap_free(heap);
}

/** Get the max time optimize should run in millisecs.
@param executor query executor
@param table user table to be optimized
@return max optimize time limit in millisecs. */
static
ulint fts_optimize_get_time_limit(FTSQueryExecutor *executor,
                                  const dict_table_t *table) noexcept
{
  ulint time_limit= 0;
  fts_string_t value;
  value.f_len= FTS_MAX_CONFIG_VALUE_LEN;
  value.f_str= static_cast<byte*>(ut_malloc_nokey(value.f_len + 1));
  dberr_t error= fts_config_get_value(executor, table,
                                      FTS_OPTIMIZE_LIMIT_IN_SECS, &value);
  if (error == DB_SUCCESS)
    time_limit= strtoul(reinterpret_cast<char*>(value.f_str), nullptr, 10);
  ut_free(value.f_str);
  /* FIXME: This is returning milliseconds, while the variable
  is being stored and interpreted as seconds! */
  return(time_limit * 1000);
}

/** Run OPTIMIZE on the given table. Note: this can take a very
long time (hours).
@param executor query executor
@param optim optimize instance
@param index current fts being optimized
@param word starting word to optimize */
static
void fts_optimize_words(FTSQueryExecutor *executor, fts_optimize_t *optim, 
                        dict_index_t *index, fts_string_t *word) noexcept
{
  ut_a(!optim->done);
  /* Get the time limit from the config table. */
  fts_optimize_time_limit=
    fts_optimize_get_time_limit(executor, index->table);
  const time_t start_time= time(NULL);

  while (!optim->done)
  {
    trx_t *trx= optim->trx;
    ut_a(ib_vector_size(optim->words) == 0);
    /* Read the index records to optimize. */
    dberr_t error= fts_index_fetch_nodes(
      executor, index, word, optim->words, nullptr, AuxCompareMode::EQUAL);
    if (error == DB_SUCCESS)
    {
      /* There must be some nodes to read. */
      ut_a(ib_vector_size(optim->words) > 0);
      /* Optimize the nodes that were read and write back to DB. */
      error = fts_optimize_compact(executor, optim, index, start_time);
      if (error == DB_SUCCESS) fts_sql_commit(optim->trx);
      else fts_sql_rollback(optim->trx);
    }
    ib_vector_reset(optim->words);

    if (error == DB_SUCCESS)
    {
      if (!optim->done && !fts_zip_read_word(optim->zip, word))
        optim->done= TRUE;
    }
    else if (error == DB_LOCK_WAIT_TIMEOUT)
    {
      sql_print_warning("InnoDB: Lock wait timeout during optimize. "
                        "Retrying!");
      trx->error_state= DB_SUCCESS;
    }
    else if (error == DB_DEADLOCK)
    {
      sql_print_warning("InnoDB: Deadlock during optimize. Retrying!");
      trx->error_state = DB_SUCCESS;
    }
    else optim->done = TRUE;
  }
}

/** Optimize is complete. Set the completion time, and reset the
optimize start string for this FTS index to "".
@param executor  query executor
@param optim  optimize instance
@param index  table with one FTS index
@return DB_SUCCESS if all OK */
static MY_ATTRIBUTE((nonnull, warn_unused_result))
dberr_t
fts_optimize_index_completed(FTSQueryExecutor *executor,
                             fts_optimize_t *optim,
                             dict_index_t *index) noexcept
{
  fts_string_t word;
  dberr_t error= DB_SUCCESS;
  byte  buf[sizeof(ulint)];
  /* If we've reached the end of the index then set the start
  word to the empty string. */
  word.f_len= 0;
  word.f_str= buf;
  *word.f_str= '\0';

  error= fts_config_set_index_value(
    executor, index, FTS_LAST_OPTIMIZED_WORD, &word);

  if (UNIV_UNLIKELY(error != DB_SUCCESS))
    sql_print_error("InnoDB: (%s) while updating last optimized word!",
                    ut_strerr(error));
  return error;
}


/** Read the words that will be optimized in this pass.
@param executor  query executor
@param optim  optimize instance
@param index  table with one FTS index
@param word  buffer to use
@return DB_SUCCESS if all OK */
static MY_ATTRIBUTE((nonnull, warn_unused_result))
dberr_t
fts_optimize_index_read_words(
 FTSQueryExecutor* executor,
 fts_optimize_t*  optim,
 dict_index_t*  index,
 fts_string_t*  word)
{
 dberr_t error = DB_SUCCESS;

 if (optim->del_list_regenerated) {
  word->f_len = 0;
 } else {
  /* Get the last word that was optimized from
the config table. */
  error = fts_config_get_index_value(
   executor, index, FTS_LAST_OPTIMIZED_WORD, word);
 }

 /* If record not found then we start from the top. */
 if (error == DB_RECORD_NOT_FOUND) {
  word->f_len = 0;
  error = DB_SUCCESS;
 }

 optim->index = index;
 while (error == DB_SUCCESS) {

  error = fts_index_fetch_words(
   executor, optim, word, fts_num_word_optimize);

  if (error == DB_SUCCESS) {
   /* Reset the last optimized word to '' if no
more words could be read from the FTS index. */
   if (optim->zip->n_words == 0) {
    word->f_len = 0;
    *word->f_str = 0;
   }

   break;
  }
 }

 return(error);
}

/**********************************************************************//**
Run OPTIMIZE on the given FTS index. Note: this can take a very long
time (hours).
@return DB_SUCCESS if all OK */
static MY_ATTRIBUTE((nonnull, warn_unused_result))
dberr_t
fts_optimize_index(
/*===============*/
 FTSQueryExecutor* executor, /*!< in: query executor */
 fts_optimize_t*  optim,  /*!< in: optimize instance */
 dict_index_t*  index)  /*!< in: table with one FTS index */
{
 fts_string_t word;
 dberr_t  error;
 byte  str[FTS_MAX_WORD_LEN + 1];

 optim->done = FALSE; /* Optimize until !done */

 /* We need to read the last word optimized so that we start from
the next word. */
 word.f_str = str;

 /* We set the length of word to the size of str since we
need to pass the max len info to the fts_get_config_value() function. */
 word.f_len = sizeof(str) - 1;

 memset(word.f_str, 0x0, word.f_len);

 /* Read the words that will be optimized in this pass. */
 error = fts_optimize_index_read_words(executor, optim, index, &word);

 if (error == DB_SUCCESS) {
  int zip_error;

  ut_a(optim->zip->pos == 0);
  ut_a(optim->zip->zp->total_in == 0);
  ut_a(optim->zip->zp->total_out == 0);

  zip_error = inflateInit(optim->zip->zp);
  ut_a(zip_error == Z_OK);

  word.f_len = 0;
  word.f_str = str;

  /* Read the first word to optimize from the Zip buffer. */
  if (!fts_zip_read_word(optim->zip, &word)) {

   optim->done = TRUE;
  } else {
   fts_optimize_words(executor, optim, index, &word);
  }

  /* If we couldn't read any records then optimize is
  complete. Increment the number of indexes that have
  been optimized and set FTS index optimize state to
completed. */
  if (error == DB_SUCCESS && optim->zip->n_words == 0) {

   error = fts_optimize_index_completed(
     executor, optim, index);

   if (error == DB_SUCCESS) {
    ++optim->n_completed;
   }
  }
 }

 return(error);
}

/** Purge the doc ids that are in the snapshot from
the master deleted table.
@param executor  query executor
@param optim  optimize instance
@return DB_SUCCESS if all OK */
static MY_ATTRIBUTE((nonnull, warn_unused_result))
dberr_t fts_optimize_purge_deleted_doc_ids(FTSQueryExecutor *executor,
                                           fts_optimize_t *optim) noexcept
{
  dberr_t error= DB_SUCCESS;
  ut_a(ib_vector_size(optim->to_delete->doc_ids) > 0);
  for (ulint i= 0;
       i < ib_vector_size(optim->to_delete->doc_ids) && error != DB_SUCCESS;
       ++i)
  {
    doc_id_t *update=
      static_cast<doc_id_t*>(ib_vector_get(optim->to_delete->doc_ids, i));
    error= executor->delete_common_record("DELETED", *update);
    if (error == DB_SUCCESS)
      error= executor->delete_common_record("DELETED_CACHE", *update);
  }

  if (error != DB_SUCCESS)
    fts_sql_rollback(optim->trx);
  return error;
}

/** Delete the document ids in the pending delete, and delete tables.
@param executor  query executor
@param optim  optimize instance
@return DB_SUCCESS if all OK */
static MY_ATTRIBUTE((nonnull, warn_unused_result))
dberr_t fts_optimize_purge_deleted_doc_id_snapshot(FTSQueryExecutor *executor,
                                                  fts_optimize_t *optim) noexcept
{
  dberr_t error= executor->delete_all_common_records("BEING_DELETED");
  if (error == DB_SUCCESS)
    error= executor->delete_all_common_records("BEING_DELETED_CACHE");
  return error;
}

/** Check if there are records in BEING_DELETED table
@param executor  query executor
@param optim  optimize fts instance
@param n_rows  number of rows exist in being_deleted table
@return DB_SUCCESS if all OK */
static
dberr_t fts_optimize_being_deleted_count(FTSQueryExecutor *executor,
                                         fts_optimize_t *optim,
                                         ulint *n_rows) noexcept
{
  CommonTableReader reader;
  dberr_t err= executor->read_all_common("BEING_DELETED", reader);
  if (err == DB_SUCCESS) *n_rows= reader.size();
  return err;
}

/** Create a snapshot of deleted document IDs by moving them from
DELETED to BEING_DELETED and from DELETED_CACHE to
BEING_DELETED_CACHE.
@param executor  query executor
@param optim  optimize fts instance
@return DB_SUCCESS or error code */
static MY_ATTRIBUTE((nonnull, warn_unused_result))
dberr_t fts_optimize_create_deleted_doc_id_snapshot(FTSQueryExecutor *executor,
                                                   fts_optimize_t *optim) noexcept
{
  dberr_t err= DB_SUCCESS;
  CommonTableReader reader;

  err= executor->read_all_common("DELETED", reader);
  if (err != DB_SUCCESS && err != DB_RECORD_NOT_FOUND) return err;

  for (ulint i= 0, n= reader.size(); i < n; i++)
  {
    err= executor->insert_common_record("BEING_DELETED", reader.get(i));
    if (err != DB_SUCCESS) return err;
  }

  reader.clear();
  err= executor->read_all_common("DELETED_CACHE", reader);
  if (err != DB_SUCCESS && err != DB_RECORD_NOT_FOUND) return err;

  for (ulint i= 0, n= reader.size(); i < n; i++)
  {
    err= executor->insert_common_record("BEING_DELETED_CACHE", reader.get(i));
    if (err != DB_SUCCESS) return err;
  }

  optim->del_list_regenerated= TRUE;
  return err;
}

/*********************************************************************//**
Read in the document ids that are to be purged during optimize. The
transaction is committed upon successfully read.
@return DB_SUCCESS if all OK */
static MY_ATTRIBUTE((nonnull, warn_unused_result))
dberr_t
fts_optimize_read_deleted_doc_id_snapshot(
/*======================================*/
 FTSQueryExecutor* executor, /*!< in: FTS query executor */
 fts_optimize_t*  optim) noexcept /*!< in: optimize instance */
{
 /* Read the doc_ids to delete. */
 dberr_t error = fts_table_fetch_doc_ids(
     executor, "BEING_DELETED", optim->to_delete);

 if (error == DB_SUCCESS) {

  /* Read additional doc_ids to delete. */
  error = fts_table_fetch_doc_ids(
   executor, "BEING_DELETED_CACHE", optim->to_delete);
 }

 if (error != DB_SUCCESS) {
  fts_doc_ids_free(optim->to_delete);
  optim->to_delete = NULL;
 }
 return(error);
}

/*********************************************************************//**
Optimize the FTS indexes of a table.
@return DB_SUCCESS if all OK */
static MY_ATTRIBUTE((nonnull, warn_unused_result))
dberr_t
fts_optimize_indexes(
/*=================*/
 FTSQueryExecutor* executor, /*!< in: query executor */
 fts_optimize_t*  optim) noexcept /*!< in: optimize instance */
{
 ulint  i;
 dberr_t  error = DB_SUCCESS;
 fts_t*  fts = optim->table->fts;

 /* Optimize the FTS indexes. */
 for (i = 0; i < ib_vector_size(fts->indexes); ++i) {
  dict_index_t* index;

  index = static_cast<dict_index_t*>(
   ib_vector_getp(fts->indexes, i));

  /* Open auxiliary tables for this index */
  error = executor->open_all_aux_tables(index);
  if (error != DB_SUCCESS) {
   break;
  }

  error = fts_optimize_index(executor, optim, index);
  if (error != DB_SUCCESS) {
   break;
  }
 }

 if (error == DB_SUCCESS) {
  fts_sql_commit(optim->trx);
 } else {
  fts_sql_rollback(optim->trx);
 }

 return(error);
}

/*********************************************************************//**
Cleanup the snapshot tables and the master deleted table.
@return DB_SUCCESS if all OK */
static MY_ATTRIBUTE((nonnull, warn_unused_result))
dberr_t
fts_optimize_purge_snapshot(
/*========================*/
 FTSQueryExecutor* executor, /*!< in: query executor */
 fts_optimize_t*  optim) noexcept /*!< in: optimize instance */
{
 dberr_t  error;

 /* Delete the doc ids from the master deleted tables, that were
in the snapshot that was taken at the start of optimize. */
 error = fts_optimize_purge_deleted_doc_ids(executor, optim);

 if (error == DB_SUCCESS) {
  /* Destroy the deleted doc id snapshot. */
  error = fts_optimize_purge_deleted_doc_id_snapshot(
    executor, optim);
 }

 if (error == DB_SUCCESS) {
  fts_sql_commit(optim->trx);
 } else {
  fts_sql_rollback(optim->trx);
 }

 return(error);
}


/*********************************************************************//**
Run OPTIMIZE on the given table by a background thread.
@return DB_SUCCESS if all OK */
static MY_ATTRIBUTE((nonnull))
dberr_t
fts_optimize_table_bk(
/*==================*/
 fts_slot_t* slot) /*!< in: table to optimiza */
{
 const time_t now = time(NULL);
 const ulint interval = ulint(now - slot->last_run);

 /* Avoid optimizing tables that were optimized recently. */
 if (slot->last_run > 0
     && lint(interval) >= 0
     && interval < FTS_OPTIMIZE_INTERVAL_IN_SECS) {

  return(DB_SUCCESS);
 }

 dict_table_t* table = slot->table;
 dberr_t  error;

 if (table->is_accessible()
     && table->fts && table->fts->cache
     && table->fts->cache->deleted >= FTS_OPTIMIZE_THRESHOLD) {
  error = fts_optimize_table(table, fts_opt_thd);

  slot->last_run = time(NULL);

  if (error == DB_SUCCESS) {
   slot->running = false;
   slot->completed = slot->last_run;
  }
 } else {
  /* Note time this run completed. */
  slot->last_run = now;
  error = DB_SUCCESS;
 }

 return(error);
}
/** Run OPTIMIZE on the given table.
@param table table to be optimized
@param thd   thread which executes optimize table
@return DB_SUCCESS if all OK */
dberr_t
fts_optimize_table(dict_table_t *table, THD *thd)
{
 ut_ad(!srv_read_only_mode || recv_sys.rpo);

 if (recv_sys.rpo) {
  return DB_READ_ONLY;
 }

 /* Serialize concurrent fts_optimize_table() on the same table:
 acquire MDL_EXCLUSIVE first so a second caller blocks here, then
 downgrade to MDL_SHARED_UPGRADABLE so other operations can proceed
while optimization is in progress. */
 MDL_ticket* mdl_ticket = nullptr;
 dict_sys.freeze(SRW_LOCK_CALL);
 if (dict_acquire_mdl<false, true>(table, thd, &mdl_ticket)
       != table) {
  dict_sys.unfreeze();
  if (mdl_ticket) {
   thd->mdl_context.release_lock(mdl_ticket);
  }
  return DB_TABLE_NOT_FOUND;
 }
 dict_sys.unfreeze();

 if (mdl_ticket) {
  mdl_ticket->downgrade_lock(MDL_SHARED_UPGRADABLE);
 }

 dberr_t  error = DB_SUCCESS;
 fts_optimize_t* optim = NULL;
 fts_t*  fts = table->fts;

 optim = fts_optimize_create(table);

 optim->trx->mysql_thd = thd;

 /* Create FTSQueryExecutor and open common tables */
 FTSQueryExecutor executor(optim->trx, table);
 error = executor.open_all_deletion_tables();
 if (error != DB_SUCCESS) {
err_exit:
  fts_optimize_free(optim);
  if (mdl_ticket) {
   thd->mdl_context.release_lock(mdl_ticket);
  }
  return error;
 }

 error = executor.open_config_table();
 if (error) { goto err_exit; }

 // FIXME: Call this only at the start of optimize, currently we
 // rely on DB_DUPLICATE_KEY to handle corrupting the snapshot.

 /* Check whether there are still records in BEING_DELETED table */
 ulint n_rows = 0;
 error= fts_optimize_being_deleted_count(&executor, optim, &n_rows);

 if (error == DB_SUCCESS && n_rows == 0) {
  /* Take a snapshot of the deleted document ids, they are copied
to the BEING_ tables. */
  error = fts_optimize_create_deleted_doc_id_snapshot(
     &executor, optim);
 }

 /* A duplicate error is OK, since we don't erase the
 doc ids from the being deleted state until all FTS
indexes have been optimized. */
 if (error == DB_DUPLICATE_KEY) {
  error = DB_SUCCESS;
 }

 if (error == DB_SUCCESS) {

  /* These document ids will be filtered out during the
  index optimization phase. They are in the snapshot that we
took above, at the start of the optimize. */
  error = fts_optimize_read_deleted_doc_id_snapshot(&executor, optim);

  if (error == DB_SUCCESS) {

   /* Commit the read of being deleted
doc ids transaction. */
   fts_sql_commit(optim->trx);

   /* We would do optimization only if there
are deleted records to be cleaned up */
   if (ib_vector_size(optim->to_delete->doc_ids) > 0) {
    error = fts_optimize_indexes(&executor, optim);
   }

  } else {
   ut_a(optim->to_delete == NULL);
  }

  /* Only after all indexes have been optimized can we
  delete the (snapshot) doc ids in the pending delete,
and master deleted tables. */
  if (error == DB_SUCCESS
      && optim->n_completed == ib_vector_size(fts->indexes)) {

   if (ib_vector_size(optim->to_delete->doc_ids) > 0) {

    /* Purge the doc ids that were in the
    snapshot from the snapshot tables and
the master deleted table. */
    error = fts_optimize_purge_snapshot(
      &executor, optim);
   }

  }
 }

 fts_optimize_free(optim);

 if (mdl_ticket) {
  thd->mdl_context.release_lock(mdl_ticket);
 }

 return(error);
}

/********************************************************************//**
Add the table to add to the OPTIMIZER's list.
@return new message instance */
static
fts_msg_t*
fts_optimize_create_msg(
/*====================*/
 fts_msg_type_t type,  /*!< in: type of message */
 void*  ptr)  /*!< in: message payload */
{
 mem_heap_t* heap;
 fts_msg_t* msg;

 heap = mem_heap_create(sizeof(*msg) + sizeof(ib_list_node_t) + 16);
 msg = static_cast<fts_msg_t*>(mem_heap_alloc(heap, sizeof(*msg)));

 msg->ptr = ptr;
 msg->type = type;
 msg->heap = heap;

 return(msg);
}

/** Add message to wqueue, signal thread pool*/
static void add_msg(fts_msg_t *msg)
{
  ib_wqueue_add(fts_optimize_wq, msg, msg->heap, true);
  srv_thread_pool->submit_task(&task);
}

/**
Called by "idle" timer. Submits optimize task, which
will only recalculate is_sync_needed, in case the queue is empty.
*/
static void timer_callback(void*)
{
  srv_thread_pool->submit_task(&task);
}

/** Add the table to add to the OPTIMIZER's list.
@param[in] table table to add */
void fts_optimize_add_table(dict_table_t* table)
{
 fts_msg_t* msg;

 if (!fts_optimize_wq) {
  return;
 }

 /* Make sure table with FTS index cannot be evicted */
 dict_sys.prevent_eviction(table);

 msg = fts_optimize_create_msg(FTS_MSG_ADD_TABLE, table);

 mysql_mutex_lock(&fts_optimize_wq->mutex);

 add_msg(msg);

 table->fts->in_queue = true;

 mysql_mutex_unlock(&fts_optimize_wq->mutex);
}

/**********************************************************************//**
Remove the table from the OPTIMIZER's list. We do wait for
acknowledgement from the consumer of the message. */
void
fts_optimize_remove_table(
/*======================*/
 dict_table_t* table)   /*!< in: table to remove */
{
  if (!fts_optimize_wq)
    return;

  if (fts_opt_start_shutdown)
  {
    sql_print_information("InnoDB: Try to remove table %s after FTS optimize "
                          "thread exiting.", table->name.m_name);
    while (fts_optimize_wq)
      std::this_thread::sleep_for(std::chrono::milliseconds(10));
    return;
  }

  mysql_mutex_lock(&fts_optimize_wq->mutex);

  if (table->fts->in_queue)
  {
    fts_msg_t *msg= fts_optimize_create_msg(FTS_MSG_DEL_TABLE, nullptr);
    pthread_cond_t cond;
    pthread_cond_init(&cond, nullptr);
    msg->ptr= new(mem_heap_alloc(msg->heap, sizeof(fts_msg_del_t)))
      fts_msg_del_t{table, &cond};
    add_msg(msg);
    my_cond_wait(&cond, &fts_optimize_wq->mutex.m_mutex);
    pthread_cond_destroy(&cond);
    ut_ad(!table->fts->in_queue);
  }

  mysql_mutex_unlock(&fts_optimize_wq->mutex);
}

/** Send sync fts cache for the table.
@param[in] table table to sync */
void
fts_optimize_request_sync_table(
 dict_table_t* table)
{
 /* if the optimize system not yet initialized, return */
 if (!fts_optimize_wq) {
  return;
 }

 mysql_mutex_lock(&fts_optimize_wq->mutex);

 /* FTS optimizer thread is already exited */
 if (fts_opt_start_shutdown) {
  sql_print_information("InnoDB: Try to sync table %s "
                                      "after FTS optimize thread exiting.",
          table->name.m_name);
 } else if (table->fts->sync_message) {
  /* If the table already has SYNC message in
fts_optimize_wq queue then ignore it */
 } else {
  add_msg(fts_optimize_create_msg(FTS_MSG_SYNC_TABLE, table));
  table->fts->sync_message = true;
  DBUG_EXECUTE_IF("fts_optimize_wq_count_check",
    DBUG_ASSERT(fts_optimize_wq->length <= 1000););
 }

 mysql_mutex_unlock(&fts_optimize_wq->mutex);
}

/** Add a table to fts_slots if it doesn't already exist. */
static bool fts_optimize_new_table(dict_table_t* table)
{
 ut_ad(table);

 ulint  i;
 fts_slot_t* slot;
 fts_slot_t* empty = NULL;

 /* Search for duplicates, also find a free slot if one exists. */
 for (i = 0; i < ib_vector_size(fts_slots); ++i) {

  slot = static_cast<fts_slot_t*>(ib_vector_get(fts_slots, i));

  if (!slot->table) {
   empty = slot;
  } else if (slot->table == table) {
   /* Already exists in our optimize queue. */
   return false;
  }
 }

 slot = empty ? empty : static_cast<fts_slot_t*>(
  ib_vector_push(fts_slots, NULL));

 memset(slot, 0x0, sizeof(*slot));

 slot->table = table;
 return true;
}

/** Remove a table from fts_slots if it exists.
@param remove table to be removed from fts_slots */
static bool fts_optimize_del_table(fts_msg_del_t *remove)
{
 const dict_table_t* table = remove->table;
 ut_ad(table);
 for (ulint i = 0; i < ib_vector_size(fts_slots); ++i) {
  fts_slot_t* slot;

  slot = static_cast<fts_slot_t*>(ib_vector_get(fts_slots, i));

  if (slot->table == table) {
   mysql_mutex_lock(&fts_optimize_wq->mutex);
   table->fts->in_queue = false;
   pthread_cond_signal(remove->cond);
   mysql_mutex_unlock(&fts_optimize_wq->mutex);
   slot->table = NULL;
   return true;
  }
 }

 mysql_mutex_lock(&fts_optimize_wq->mutex);
 pthread_cond_signal(remove->cond);
 mysql_mutex_unlock(&fts_optimize_wq->mutex);
 return false;
}

/**********************************************************************//**
Calculate how many tables in fts_slots need to be optimized.
@return no. of tables to optimize */
static ulint fts_optimize_how_many()
{
 ulint n_tables = 0;
 const time_t current_time = time(NULL);

 for (ulint i = 0; i < ib_vector_size(fts_slots); ++i) {
  const fts_slot_t* slot = static_cast<const fts_slot_t*>(
   ib_vector_get_const(fts_slots, i));
  if (!slot->table) {
   continue;
  }

  const time_t end = slot->running
   ? slot->last_run : slot->completed;
  ulint interval = ulint(current_time - end);

  if (lint(interval) < 0
      || interval >= FTS_OPTIMIZE_INTERVAL_IN_SECS) {
   ++n_tables;
  }
 }

 return(n_tables);
}

/**********************************************************************//**
Check if the total memory used by all FTS table exceeds the maximum limit.
@return true if a sync is needed, false otherwise */
static bool fts_is_sync_needed()
{
 ulint  total_memory = 0;
 const time_t now = time(NULL);
 double  time_diff = difftime(now, last_check_sync_time);

 if (fts_need_sync || (time_diff >= 0 && time_diff < 5)) {
  return(false);
 }

 last_check_sync_time = now;

 for (ulint i = 0; i < ib_vector_size(fts_slots); ++i) {
  const fts_slot_t* slot = static_cast<const fts_slot_t*>(
   ib_vector_get_const(fts_slots, i));

  if (!slot->table) {
   continue;
  }

  if (slot->table->fts && slot->table->fts->cache) {
   total_memory += slot->table->fts->cache->total_size;
  }

  if (total_memory > fts_max_total_cache_size) {
   return(true);
  }
 }

 return(false);
}

/** Sync fts cache of a table
@param[in,out]  table           table to be synced
@param[in]      process_message processing messages from fts_optimize_wq */
static void fts_optimize_sync_table(dict_table_t *table,
                                    bool process_message= false)
{
  MDL_ticket* mdl_ticket= nullptr;
  dict_table_t *sync_table= dict_acquire_mdl<true>(table, fts_opt_thd,
                                                   &mdl_ticket);

  if (!sync_table)
    return;

  if (sync_table->fts && sync_table->fts->cache && sync_table->is_accessible())
  {
    /* fts_opt_thd is the background optimizer's THD; binding it
    gives the internal sync trx a real wait-timeout policy and
    well-defined per-THD wait-time accounting. */
    fts_sync_table(sync_table, false, fts_opt_thd);
    if (process_message)
    {
      mysql_mutex_lock(&fts_optimize_wq->mutex);
      sync_table->fts->sync_message = false;
      mysql_mutex_unlock(&fts_optimize_wq->mutex);
    }
  }

  DBUG_EXECUTE_IF("ib_optimize_wq_hang",
    std::this_thread::sleep_for(std::chrono::seconds(6)););

  if (mdl_ticket)
    dict_table_close(sync_table, fts_opt_thd, mdl_ticket);
}

/**********************************************************************//**
Optimize all FTS tables.
@return Dummy return */
static void fts_optimize_callback(void *)
{
 ut_ad(!srv_read_only_mode);
 ut_ad(!recv_sys.rpo);

 static ulint current;
 static bool done;
 static ulint n_optimize;

 if (!fts_optimize_wq || done) {
  /* Possibly timer initiated callback, can come after FTS_MSG_STOP.*/
  return;
 }

 static ulint  n_tables = ib_vector_size(fts_slots);

 while (!done && srv_shutdown_state <= SRV_SHUTDOWN_INITIATED) {
#ifdef WITH_WSREP
  ut_d(extern Atomic_relaxed<bool> wsrep_sst_disable_writes);
  ut_ad(!wsrep_sst_disable_writes);
#endif
  /* If there is no message in the queue and we have tables
to optimize then optimize the tables. */

  if (!done
      && ib_wqueue_is_empty(fts_optimize_wq)
      && n_tables > 0
      && n_optimize > 0) {

   /* The queue is empty but we have tables
to optimize. */
   fts_slot_t* slot = static_cast<fts_slot_t*>(
    ib_vector_get(fts_slots, current));

   /* Handle the case of empty slots. */
   if (slot->table) {
    slot->running = true;
    fts_optimize_table_bk(slot);
   }

   /* Wrap around the counter. */
   if (++current >= ib_vector_size(fts_slots)) {
    n_optimize = fts_optimize_how_many();
    current = 0;
   }
  } else if (n_optimize == 0
      || !ib_wqueue_is_empty(fts_optimize_wq)) {
   fts_msg_t* msg = static_cast<fts_msg_t*>
    (ib_wqueue_nowait(fts_optimize_wq));
   /* Timeout ? */
   if (!msg) {
    if (fts_is_sync_needed()) {
     fts_need_sync = true;
    }
    if (n_tables) {
     timer->set_time(5000, 0);
    }
    return;
   }

   switch (msg->type) {
   case FTS_MSG_STOP:
    done = true;
    break;

   case FTS_MSG_ADD_TABLE:
    ut_a(!done);
    if (fts_optimize_new_table(
         static_cast<dict_table_t*>(
          msg->ptr))) {
     ++n_tables;
    }
    break;

   case FTS_MSG_DEL_TABLE:
    if (fts_optimize_del_table(
         static_cast<fts_msg_del_t*>(
          msg->ptr))) {
     --n_tables;
    }
    break;

   case FTS_MSG_SYNC_TABLE:
    DBUG_EXECUTE_IF(
     "fts_instrument_msg_sync_sleep",
     std::this_thread::sleep_for(
      std::chrono::milliseconds(
       300)););

    fts_optimize_sync_table(
     static_cast<dict_table_t*>(msg->ptr),
     true);
    break;

   default:
    ut_error;
   }

   mem_heap_free(msg->heap);
   n_optimize = done ? 0 : fts_optimize_how_many();
  }
 }

 /* Server is being shutdown, sync the data from FTS cache to disk
if needed */
 if (n_tables > 0) {
  for (ulint i = 0; i < ib_vector_size(fts_slots); i++) {
   fts_slot_t* slot = static_cast<fts_slot_t*>(
    ib_vector_get(fts_slots, i));

   if (slot->table) {
    fts_optimize_sync_table(slot->table);
   }
  }
 }

 ib_vector_free(fts_slots);
 mysql_mutex_lock(&fts_optimize_wq->mutex);
 fts_slots = NULL;
 pthread_cond_broadcast(&fts_opt_shutdown_cond);
 mysql_mutex_unlock(&fts_optimize_wq->mutex);

 sql_print_information("InnoDB: FTS optimize thread exiting.");
}

/** Startup the optimize task and create the work queue. */
void fts_optimize_init()
{
 mem_heap_t* heap;
 ib_alloc_t*     heap_alloc;

 ut_ad(!srv_read_only_mode);
 ut_ad(!recv_sys.rpo);

 /* For now we only support one optimize thread. */
 ut_a(!fts_optimize_wq);

 /* Create FTS optimize work queue */
 fts_optimize_wq = ib_wqueue_create();
 timer = srv_thread_pool->create_timer(timer_callback);

 /* Create FTS vector to store fts_slot_t */
 heap = mem_heap_create(sizeof(dict_table_t*) * 64);
 heap_alloc = ib_heap_allocator_create(heap);
 fts_slots = ib_vector_create(heap_alloc, sizeof(fts_slot_t), 4);

 fts_opt_thd = innobase_create_background_thd("InnoDB FTS optimizer");
 /* Add fts tables to fts_slots which could be skipped
 during dict_load_table_one() because fts_optimize_thread
wasn't even started. */
 dict_sys.freeze(SRW_LOCK_CALL);
 for (dict_table_t* table = UT_LIST_GET_FIRST(dict_sys.table_LRU);
      table != NULL;
      table = UT_LIST_GET_NEXT(table_LRU, table)) {
  if (!table->fts || !dict_table_has_fts_index(table)) {
   continue;
  }

  /* fts_optimize_thread is not started yet. So there is no
  need to acquire fts_optimize_wq->mutex for adding the fts
table to the fts slots. */
  ut_ad(!table->can_be_evicted);
  fts_optimize_new_table(table);
  table->fts->in_queue = true;
 }
 dict_sys.unfreeze();

 pthread_cond_init(&fts_opt_shutdown_cond, nullptr);
 last_check_sync_time = time(NULL);
}

/** Shut down the fts optimize thread. */
void fts_optimize_shutdown()
{
 ut_ad(!srv_read_only_mode);
 ut_ad(!recv_sys.rpo);

 /* If there is an ongoing activity on dictionary, such as
srv_master_evict_from_table_cache(), wait for it */
 dict_sys.freeze(SRW_LOCK_CALL);
 mysql_mutex_lock(&fts_optimize_wq->mutex);
 /* Tells FTS optimizer system that we are exiting from
 optimizer thread, messages sent thereafter will not be
processed */
 fts_opt_start_shutdown = true;
 dict_sys.unfreeze();

 /* We tell the OPTIMIZE thread to switch to state done, we
 can't delete the work queue here because the add thread needs
deregister the FTS tables. */
 timer->disarm();
 task_group.cancel_pending(&task);

 add_msg(fts_optimize_create_msg(FTS_MSG_STOP, nullptr));

 while (fts_slots) {
  my_cond_wait(&fts_opt_shutdown_cond,
        &fts_optimize_wq->mutex.m_mutex);
 }

 destroy_background_thd(fts_opt_thd);
 fts_opt_thd = NULL;
 pthread_cond_destroy(&fts_opt_shutdown_cond);
 mysql_mutex_unlock(&fts_optimize_wq->mutex);

 ib_wqueue_free(fts_optimize_wq);
 fts_optimize_wq = NULL;

 delete timer;
 timer = NULL;
}

#ifdef WITH_WSREP
/** Pause the optimize subsystem. */
void fts_optimize_pause()
{
  ut_ad(!srv_read_only_mode);
  /* Prevent fts_optimize_callback() from being scheduled. */
  timer->disarm();
  /* Wait for any current fts_optimize_callback() to finish. */
  task.wait();
}

/** Resume after fts_optimize_stop() */
void fts_optimize_resume()
{
  /* Schedule fts_optimize_callback() immediately.
  It will reschedule itself via the timer when needed. */
  srv_thread_pool->submit_task(&task);
}
#endif

/** Sync the table during commit phase
@param[in] table table to be synced */
void fts_sync_during_ddl(dict_table_t* table, THD* thd)
{
  if (!fts_optimize_wq)
    return;
  mysql_mutex_lock(&fts_optimize_wq->mutex);
  const auto sync_message= table->fts->sync_message;
  mysql_mutex_unlock(&fts_optimize_wq->mutex);
  if (!sync_message)
    return;

  fts_sync_table(table, false, thd);

  mysql_mutex_lock(&fts_optimize_wq->mutex);
  table->fts->sync_message = false;
  mysql_mutex_unlock(&fts_optimize_wq->mutex);
}