open_table

/*
  Open a table.

  SYNOPSIS
    open_table()
    thd                 Thread context.
    table_list          Open first table in list.
    action       INOUT  Pointer to variable of enum_open_table_action type
                        which will be set according to action which is
                        required to remedy problem appeared during attempt
                        to open table.
    flags               Bitmap of flags to modify how open works:
                          MYSQL_OPEN_IGNORE_FLUSH - Open table even if
                          someone has done a flush or there is a pending
                          exclusive metadata lock requests against it
                          (i.e. request high priority metadata lock).
                          No version number checking is done.
                          MYSQL_OPEN_TEMPORARY_ONLY - Open only temporary
                          table not the base table or view.
                          MYSQL_OPEN_TAKE_UPGRADABLE_MDL - Obtain upgradable
                          metadata lock for tables on which we are going to
                          take some kind of write table-level lock.

  IMPLEMENTATION
    Uses a cache of open tables to find a table not in use.

    If TABLE_LIST::open_strategy is set to OPEN_IF_EXISTS, the table is opened
    only if it exists. If the open strategy is OPEN_STUB, the underlying table
    is never opened. In both cases, metadata locks are always taken according
    to the lock strategy.

  RETURN
    TRUE  Open failed. "action" parameter may contain type of action
          needed to remedy problem before retrying again.
    FALSE Success. Members of TABLE_LIST structure are filled properly (e.g.
          TABLE_LIST::table is set for real tables and TABLE_LIST::view is
          set for views).
*/

bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
                Open_table_context *ot_ctx)
{
  reg1    TABLE *table;
  char    key[MAX_DBKEY_LENGTH];
  uint    key_length;
  char    *alias= table_list->alias;
  uint flags= ot_ctx->get_flags();
  MDL_ticket *mdl_ticket;
  int error;
  TABLE_SHARE *share;
  my_hash_value_type hash_value;
  DBUG_ENTER("open_table");

  /* an open table operation needs a lot of the stack space */
  if (check_stack_overrun(thd, STACK_MIN_SIZE_FOR_OPEN, (uchar *)&alias))
    DBUG_RETURN(TRUE);

  if (thd->killed)
    DBUG_RETURN(TRUE);

  key_length= (create_table_def_key(thd, key, table_list, 1) -
               TMP_TABLE_KEY_EXTRA);

  /*
    Unless requested otherwise, try to resolve this table in the list
    of temporary tables of this thread. In MySQL temporary tables
    are always thread-local and "shadow" possible base tables with the
    same name. This block implements the behaviour.
    TODO: move this block into a separate function.
  */
  if (table_list->open_type != OT_BASE_ONLY &&
      ! (flags & MYSQL_OPEN_SKIP_TEMPORARY))
  {
    for (table= thd->temporary_tables; table ; table=table->next)
    {
      if (table->s->table_cache_key.length == key_length +
          TMP_TABLE_KEY_EXTRA &&
      !memcmp(table->s->table_cache_key.str, key,
          key_length + TMP_TABLE_KEY_EXTRA))
      {
        /*
          We‘re trying to use the same temporary table twice in a query.
          Right now we don‘t support this because a temporary table
          is always represented by only one TABLE object in THD, and
          it can not be cloned. Emit an error for an unsupported behaviour.
        */
    if (table->query_id)
    {
          DBUG_PRINT("error",
                     ("query_id: %lu  server_id: %u  pseudo_thread_id: %lu",
                      (ulong) table->query_id, (uint) thd->server_id,
                      (ulong) thd->variables.pseudo_thread_id));
      my_error(ER_CANT_REOPEN_TABLE, MYF(0), table->alias);
      DBUG_RETURN(TRUE);
    }
    table->query_id= thd->query_id;
    thd->thread_specific_used= TRUE;
        DBUG_PRINT("info",("Using temporary table"));
        goto reset;
      }
    }
  }

  if (table_list->open_type == OT_TEMPORARY_ONLY ||
      (flags & MYSQL_OPEN_TEMPORARY_ONLY))
  {
    if (table_list->open_strategy == TABLE_LIST::OPEN_NORMAL)
    {
      my_error(ER_NO_SUCH_TABLE, MYF(0), table_list->db, table_list->table_name);
      DBUG_RETURN(TRUE);
    }
    else
      DBUG_RETURN(FALSE);
  }

  /*
    The table is not temporary - if we‘re in pre-locked or LOCK TABLES
    mode, let‘s try to find the requested table in the list of pre-opened
    and locked tables. If the table is not there, return an error - we can‘t
    open not pre-opened tables in pre-locked/LOCK TABLES mode.
    TODO: move this block into a separate function.
  */
  if (thd->locked_tables_mode &&
      ! (flags & MYSQL_OPEN_GET_NEW_TABLE))
  {                        // Using table locks
    TABLE *best_table= 0;
    int best_distance= INT_MIN;
    for (table=thd->open_tables; table ; table=table->next)
    {
      if (table->s->table_cache_key.length == key_length &&
      !memcmp(table->s->table_cache_key.str, key, key_length))
      {
        if (!my_strcasecmp(system_charset_info, table->alias, alias) &&
            table->query_id != thd->query_id && /* skip tables already used */
            (thd->locked_tables_mode == LTM_LOCK_TABLES ||
             table->query_id == 0))
        {
          int distance= ((int) table->reginfo.lock_type -
                         (int) table_list->lock_type);

          /*
            Find a table that either has the exact lock type requested,
            or has the best suitable lock. In case there is no locked
            table that has an equal or higher lock than requested,
            we us the closest matching lock to be able to produce an error
            message about wrong lock mode on the table. The best_table
            is changed if bd < 0 <= d or bd < d < 0 or 0 <= d < bd.

            distance <  0 - No suitable lock found
            distance >  0 - we have lock mode higher then we require
            distance == 0 - we have lock mode exactly which we need
          */
          if ((best_distance < 0 && distance > best_distance) ||
              (distance >= 0 && distance < best_distance))
          {
            best_distance= distance;
            best_table= table;
            if (best_distance == 0)
            {
              /*
                We have found a perfect match and can finish iterating
                through open tables list. Check for table use conflict
                between calling statement and SP/trigger is done in
                lock_tables().
              */
              break;
            }
          }
        }
      }
    }
    if (best_table)
    {
      table= best_table;
      table->query_id= thd->query_id;
      DBUG_PRINT("info",("Using locked table"));
      goto reset;
    }
    /*
      Is this table a view and not a base table?
      (it is work around to allow to open view with locked tables,
      real fix will be made after definition cache will be made)

      Since opening of view which was not explicitly locked by LOCK
      TABLES breaks metadata locking protocol (potentially can lead
      to deadlocks) it should be disallowed.
    */
    if (thd->mdl_context.is_lock_owner(MDL_key::TABLE,
                                       table_list->db,
                                       table_list->table_name,
                                       MDL_SHARED))
    {
      char path[FN_REFLEN + 1];
      enum legacy_db_type not_used;
      build_table_filename(path, sizeof(path) - 1,
                           table_list->db, table_list->table_name, reg_ext, 0);
      /*
        Note that we can‘t be 100% sure that it is a view since it‘s
        possible that we either simply have not found unused TABLE
        instance in THD::open_tables list or were unable to open table
        during prelocking process (in this case in theory we still
        should hold shared metadata lock on it).
      */
      if (dd_frm_type(thd, path, &not_used) == FRMTYPE_VIEW)
      {
        if (!tdc_open_view(thd, table_list, alias, key, key_length,
                           mem_root, 0))
        {
          DBUG_ASSERT(table_list->view != 0);
          DBUG_RETURN(FALSE); // VIEW
        }
      }
    }
    /*
      No table in the locked tables list. In case of explicit LOCK TABLES
      this can happen if a user did not include the able into the list.
      In case of pre-locked mode locked tables list is generated automatically,
      so we may only end up here if the table did not exist when
      locked tables list was created.
    */
    if (thd->locked_tables_mode == LTM_PRELOCKED)
      my_error(ER_NO_SUCH_TABLE, MYF(0), table_list->db, table_list->alias);
    else
      my_error(ER_TABLE_NOT_LOCKED, MYF(0), alias);
    DBUG_RETURN(TRUE);
  }

  /*
    Non pre-locked/LOCK TABLES mode, and the table is not temporary.
    This is the normal use case.
  */

  if (! (flags & MYSQL_OPEN_HAS_MDL_LOCK))
  {
    /*
      We are not under LOCK TABLES and going to acquire write-lock/
      modify the base table. We need to acquire protection against
      global read lock until end of this statement in order to have
      this statement blocked by active FLUSH TABLES WITH READ LOCK.

      We don‘t block acquire this protection under LOCK TABLES as
      such protection already acquired at LOCK TABLES time and
      not released until UNLOCK TABLES.

      We don‘t block statements which modify only temporary tables
      as these tables are not preserved by backup by any form of
      backup which uses FLUSH TABLES WITH READ LOCK.

      TODO: The fact that we sometimes acquire protection against
            GRL only when we encounter table to be write-locked
            slightly increases probability of deadlock.
            This problem will be solved once Alik pushes his
            temporary table refactoring patch and we can start
            pre-acquiring metadata locks at the beggining of
            open_tables() call.
    */
    if (table_list->mdl_request.type >= MDL_SHARED_WRITE &&
        ! (flags & (MYSQL_OPEN_IGNORE_GLOBAL_READ_LOCK |
                    MYSQL_OPEN_FORCE_SHARED_MDL |
                    MYSQL_OPEN_FORCE_SHARED_HIGH_PRIO_MDL |
                    MYSQL_OPEN_SKIP_SCOPED_MDL_LOCK)) &&
        ! ot_ctx->has_protection_against_grl())
    {
      MDL_request protection_request;
      MDL_deadlock_handler mdl_deadlock_handler(ot_ctx);

      if (thd->global_read_lock.can_acquire_protection())
        DBUG_RETURN(TRUE);

      protection_request.init(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE,
                              MDL_STATEMENT);

      /*
        Install error handler which if possible will convert deadlock error
        into request to back-off and restart process of opening tables.
      */
      thd->push_internal_handler(&mdl_deadlock_handler);
      bool result= thd->mdl_context.acquire_lock(&protection_request,
                                                 ot_ctx->get_timeout());
      thd->pop_internal_handler();

      if (result)
        DBUG_RETURN(TRUE);

      ot_ctx->set_has_protection_against_grl();
    }

    if (open_table_get_mdl_lock(thd, ot_ctx, &table_list->mdl_request,
                                flags, &mdl_ticket) ||
        mdl_ticket == NULL)
    {
      DEBUG_SYNC(thd, "before_open_table_wait_refresh");
      DBUG_RETURN(TRUE);
    }
    DEBUG_SYNC(thd, "after_open_table_mdl_shared");
  }
  else
  {
    /*
      Grab reference to the MDL lock ticket that was acquired
      by the caller.
    */
    mdl_ticket= table_list->mdl_request.ticket;
  }

  hash_value= my_calc_hash(&table_def_cache, (uchar*) key, key_length);

  if (table_list->open_strategy == TABLE_LIST::OPEN_IF_EXISTS)
  {
    bool exists;

    if (check_if_table_exists(thd, table_list, &exists))
      DBUG_RETURN(TRUE);

    if (!exists)
      DBUG_RETURN(FALSE);

    /* Table exists. Let us try to open it. */
  }
  else if (table_list->open_strategy == TABLE_LIST::OPEN_STUB)
    DBUG_RETURN(FALSE);

retry_share:

  mysql_mutex_lock(&LOCK_open);

  if (!(share= get_table_share_with_discover(thd, table_list, key,
                                             key_length, OPEN_VIEW,
                                             &error,
                                             hash_value)))
  {
    mysql_mutex_unlock(&LOCK_open);
    /*
      If thd->is_error() is not set, we either need discover
      (error == 7), or the error was silenced by the prelocking
      handler (error == 0), in which case we should skip this
      table.
    */
    if (error == 7 && !thd->is_error())
    {
      (void) ot_ctx->request_backoff_action(Open_table_context::OT_DISCOVER,
                                            table_list);
    }
    DBUG_RETURN(TRUE);
  }

  if (share->is_view)
  {
    /*
      If parent_l of the table_list is non null then a merge table
      has this view as child table, which is not supported.
    */
    if (table_list->parent_l)
    {
      my_error(ER_WRONG_MRG_TABLE, MYF(0));
      goto err_unlock;
    }

    /*
      This table is a view. Validate its metadata version: in particular,
      that it was a view when the statement was prepared.
    */
    if (check_and_update_table_version(thd, table_list, share))
      goto err_unlock;
    if (table_list->i_s_requested_object & OPEN_TABLE_ONLY)
    {
      my_error(ER_NO_SUCH_TABLE, MYF(0), table_list->db,
               table_list->table_name);
      goto err_unlock;
    }

    /* Open view */
    if (open_new_frm(thd, share, alias,
                     (uint) (HA_OPEN_KEYFILE | HA_OPEN_RNDFILE |
                             HA_GET_INDEX | HA_TRY_READ_ONLY),
                     READ_KEYINFO | COMPUTE_TYPES | EXTRA_RECORD,
                     thd->open_options,
                     0, table_list, mem_root))
      goto err_unlock;

    /* TODO: Don‘t free this */
    release_table_share(share);

    DBUG_ASSERT(table_list->view);

    mysql_mutex_unlock(&LOCK_open);
    DBUG_RETURN(FALSE);
  }

  /*
    Note that situation when we are trying to open a table for what
    was a view during previous execution of PS will be handled in by
    the caller. Here we should simply open our table even if
    TABLE_LIST::view is true.
  */

  if (table_list->i_s_requested_object &  OPEN_VIEW_ONLY)
  {
    my_error(ER_NO_SUCH_TABLE, MYF(0), table_list->db,
             table_list->table_name);
    goto err_unlock;
  }

  if (!(flags & MYSQL_OPEN_IGNORE_FLUSH))
  {
    if (share->has_old_version())
    {
      /*
        We already have an MDL lock. But we have encountered an old
        version of table in the table definition cache which is possible
        when someone changes the table version directly in the cache
        without acquiring a metadata lock (e.g. this can happen during
        "rolling" FLUSH TABLE(S)).
        Release our reference to share, wait until old version of
        share goes away and then try to get new version of table share.
      */
      MDL_deadlock_handler mdl_deadlock_handler(ot_ctx);
      bool wait_result;

      release_table_share(share);
      mysql_mutex_unlock(&LOCK_open);

      thd->push_internal_handler(&mdl_deadlock_handler);
      wait_result= tdc_wait_for_old_version(thd, table_list->db,
                                            table_list->table_name,
                                            ot_ctx->get_timeout(),
                                            mdl_ticket->get_deadlock_weight());
      thd->pop_internal_handler();

      if (wait_result)
        DBUG_RETURN(TRUE);

      goto retry_share;
    }

    if (thd->open_tables && thd->open_tables->s->version != share->version)
    {
      /*
        If the version changes while we‘re opening the tables,
        we have to back off, close all the tables opened-so-far,
        and try to reopen them. Note: refresh_version is currently
        changed only during FLUSH TABLES.
      */
      release_table_share(share);
      mysql_mutex_unlock(&LOCK_open);
      (void)ot_ctx->request_backoff_action(Open_table_context::OT_REOPEN_TABLES,
                                           NULL);
      DBUG_RETURN(TRUE);
    }
  }

  if (!share->free_tables.is_empty())
  {
    table= share->free_tables.front();
    table_def_use_table(thd, table);
    /* We need to release share as we have EXTRA reference to it in our hands. */
    release_table_share(share);
  }
  else
  {
    /* We have too many TABLE instances around let us try to get rid of them. */
    while (table_cache_count > table_cache_size && unused_tables)
      free_cache_entry(unused_tables);

    mysql_mutex_unlock(&LOCK_open);

    /* make a new table */
    if (!(table=(TABLE*) my_malloc(sizeof(*table),MYF(MY_WME))))
      goto err_lock;

    error= open_table_from_share(thd, share, alias,
                                 (uint) (HA_OPEN_KEYFILE |
                                         HA_OPEN_RNDFILE |
                                         HA_GET_INDEX |
                                         HA_TRY_READ_ONLY),
                                 (READ_KEYINFO | COMPUTE_TYPES |
                                  EXTRA_RECORD),
                                 thd->open_options, table, FALSE);

    if (error)
    {
      my_free(table);

      if (error == 7)
        (void) ot_ctx->request_backoff_action(Open_table_context::OT_DISCOVER,
                                              table_list);
      else if (share->crashed)
        (void) ot_ctx->request_backoff_action(Open_table_context::OT_REPAIR,
                                              table_list);

      goto err_lock;
    }

    if (open_table_entry_fini(thd, share, table))
    {
      closefrm(table, 0);
      my_free(table);
      goto err_lock;
    }

    mysql_mutex_lock(&LOCK_open);
    /* Add table to the share‘s used tables list. */
    table_def_add_used_table(thd, table);
  }

  mysql_mutex_unlock(&LOCK_open);

  table->mdl_ticket= mdl_ticket;

  table->next= thd->open_tables;        /* Link into simple list */
  thd->set_open_tables(table);

  table->reginfo.lock_type=TL_READ;        /* Assume read */

 reset:
  /*
    Check that there is no reference to a condtion from an earlier query
    (cf. Bug#58553).
  */
  DBUG_ASSERT(table->file->pushed_cond == NULL);
  table_list->updatable= 1; // It is not derived table nor non-updatable VIEW
  table_list->table= table;

  table->init(thd, table_list);

  DBUG_RETURN(FALSE);

err_lock:
  mysql_mutex_lock(&LOCK_open);
err_unlock:
  release_table_share(share);
  mysql_mutex_unlock(&LOCK_open);

  DBUG_RETURN(TRUE);
}

/*
  Open a table based on a TABLE_SHARE

  SYNOPSIS
    open_table_from_share()
    thd            Thread handler
    share        Table definition
    alias           Alias for table
    db_stat        open flags (for example HA_OPEN_KEYFILE|
                HA_OPEN_RNDFILE..) can be 0 (example in
                        ha_example_table)
    prgflag           READ_ALL etc..
    ha_open_flags    HA_OPEN_ABORT_IF_LOCKED etc..
    outparam           result table

  RETURN VALUES
   0    ok
   1    Error (see open_table_error)
   2    Error (see open_table_error)
   3    Wrong data in .frm file
   4    Error (see open_table_error)
   5    Error (see open_table_error: charset unavailable)
   7    Table definition has changed in engine
*/

int open_table_from_share(THD *thd, TABLE_SHARE *share, const char *alias,
                          uint db_stat, uint prgflag, uint ha_open_flags,
                          TABLE *outparam, bool is_create_table)
{
  int error;
  uint records, i, bitmap_size;
  bool error_reported= FALSE;
  uchar *record, *bitmaps;
  Field **field_ptr;
  DBUG_ENTER("open_table_from_share");
  DBUG_PRINT("enter",("name: ‘%s.%s‘  form: 0x%lx", share->db.str,
                      share->table_name.str, (long) outparam));

  error= 1;
  bzero((char*) outparam, sizeof(*outparam));
  outparam->in_use= thd;
  outparam->s= share;
  outparam->db_stat= db_stat;
  outparam->write_row_record= NULL;

  init_sql_alloc(&outparam->mem_root, TABLE_ALLOC_BLOCK_SIZE, 0);

  if (!(outparam->alias= my_strdup(alias, MYF(MY_WME))))
    goto err;
  outparam->quick_keys.init();
  outparam->covering_keys.init();
  outparam->merge_keys.init();
  outparam->keys_in_use_for_query.init();

  /* Allocate handler */
  outparam->file= 0;
  if (!(prgflag & OPEN_FRM_FILE_ONLY))
  {
    if (!(outparam->file= get_new_handler(share, &outparam->mem_root,
                                          share->db_type())))
      goto err;
  }
  else
  {
    DBUG_ASSERT(!db_stat);
  }

  error= 4;
  outparam->reginfo.lock_type= TL_UNLOCK;
  outparam->current_lock= F_UNLCK;
  records=0;
  if ((db_stat & HA_OPEN_KEYFILE) || (prgflag & DELAYED_OPEN))
    records=1;
  if (prgflag & (READ_ALL+EXTRA_RECORD))
    records++;

  if (!(record= (uchar*) alloc_root(&outparam->mem_root,
                                   share->rec_buff_length * records)))
    goto err;                                   /* purecov: inspected */

  if (records == 0)
  {
    /* We are probably in hard repair, and the buffers should not be used */
    outparam->record[0]= outparam->record[1]= share->default_values;
  }
  else
  {
    outparam->record[0]= record;
    if (records > 1)
      outparam->record[1]= record+ share->rec_buff_length;
    else
      outparam->record[1]= outparam->record[0];   // Safety
  }

#ifdef HAVE_purify
  /*
    We need this because when we read var-length rows, we are not updating
    bytes after end of varchar
  */
  if (records > 1)
  {
    memcpy(outparam->record[0], share->default_values, share->rec_buff_length);
    memcpy(outparam->record[1], share->default_values, share->null_bytes);
    if (records > 2)
      memcpy(outparam->record[1], share->default_values,
             share->rec_buff_length);
  }
#endif

  if (!(field_ptr = (Field **) alloc_root(&outparam->mem_root,
                                          (uint) ((share->fields+1)*
                                                  sizeof(Field*)))))
    goto err;                                   /* purecov: inspected */

  outparam->field= field_ptr;

  record= (uchar*) outparam->record[0]-1;    /* Fieldstart = 1 */
  if (share->null_field_first)
    outparam->null_flags= (uchar*) record+1;
  else
    outparam->null_flags= (uchar*) (record+ 1+ share->reclength -
                                    share->null_bytes);

  /* Setup copy of fields from share, but use the right alias and record */
  for (i=0 ; i < share->fields; i++, field_ptr++)
  {
    if (!((*field_ptr)= share->field[i]->clone(&outparam->mem_root, outparam)))
      goto err;
  }
  (*field_ptr)= 0;                              // End marker

  if (share->found_next_number_field)
    outparam->found_next_number_field=
      outparam->field[(uint) (share->found_next_number_field - share->field)];
  if (share->timestamp_field)
    outparam->timestamp_field= (Field_timestamp*) outparam->field[share->timestamp_field_offset];

  /* Fix key->name and key_part->field */
  if (share->key_parts)
  {
    KEY    *key_info, *key_info_end;
    KEY_PART_INFO *key_part;
    uint n_length;
    n_length= share->keys*sizeof(KEY) + share->key_parts*sizeof(KEY_PART_INFO);
    if (!(key_info= (KEY*) alloc_root(&outparam->mem_root, n_length)))
      goto err;
    outparam->key_info= key_info;
    key_part= (reinterpret_cast<KEY_PART_INFO*>(key_info+share->keys));

    memcpy(key_info, share->key_info, sizeof(*key_info)*share->keys);
    memcpy(key_part, share->key_info[0].key_part, (sizeof(*key_part) *
                                                   share->key_parts));

    for (key_info_end= key_info + share->keys ;
         key_info < key_info_end ;
         key_info++)
    {
      KEY_PART_INFO *key_part_end;

      key_info->table= outparam;
      key_info->key_part= key_part;

      for (key_part_end= key_part+ key_info->key_parts ;
           key_part < key_part_end ;
           key_part++)
      {
        Field *field= key_part->field= outparam->field[key_part->fieldnr-1];

        if (field->key_length() != key_part->length &&
            !(field->flags & BLOB_FLAG))
        {
          /*
            We are using only a prefix of the column as a key:
            Create a new field for the key part that matches the index
          */
          field= key_part->field=field->new_field(&outparam->mem_root,
                                                  outparam, 0);
          field->field_length= key_part->length;
        }
      }
    }
  }

#ifdef WITH_PARTITION_STORAGE_ENGINE
  if (share->partition_info_str_len && outparam->file)
  {
  /*
    In this execution we must avoid calling thd->change_item_tree since
    we might release memory before statement is completed. We do this
    by changing to a new statement arena. As part of this arena we also
    set the memory root to be the memory root of the table since we
    call the parser and fix_fields which both can allocate memory for
    item objects. We keep the arena to ensure that we can release the
    free_list when closing the table object.
    SEE Bug #21658
  */

    Query_arena *backup_stmt_arena_ptr= thd->stmt_arena;
    Query_arena backup_arena;
    Query_arena part_func_arena(&outparam->mem_root,
                                Query_arena::STMT_INITIALIZED);
    thd->set_n_backup_active_arena(&part_func_arena, &backup_arena);
    thd->stmt_arena= &part_func_arena;
    bool tmp;
    bool work_part_info_used;

    tmp= mysql_unpack_partition(thd, share->partition_info_str,
                                share->partition_info_str_len,
                                outparam, is_create_table,
                                share->default_part_db_type,
                                &work_part_info_used);
    if (tmp)
    {
      thd->stmt_arena= backup_stmt_arena_ptr;
      thd->restore_active_arena(&part_func_arena, &backup_arena);
      goto partititon_err;
    }
    outparam->part_info->is_auto_partitioned= share->auto_partitioned;
    DBUG_PRINT("info", ("autopartitioned: %u", share->auto_partitioned));
    /* we should perform the fix_partition_func in either local or
       caller‘s arena depending on work_part_info_used value
    */
    if (!work_part_info_used)
      tmp= fix_partition_func(thd, outparam, is_create_table);
    thd->stmt_arena= backup_stmt_arena_ptr;
    thd->restore_active_arena(&part_func_arena, &backup_arena);
    if (!tmp)
    {
      if (work_part_info_used)
        tmp= fix_partition_func(thd, outparam, is_create_table);
    }
    outparam->part_info->item_free_list= part_func_arena.free_list;
partititon_err:
    if (tmp)
    {
      if (is_create_table)
      {
        /*
          During CREATE/ALTER TABLE it is ok to receive errors here.
          It is not ok if it happens during the opening of an frm
          file as part of a normal query.
        */
        error_reported= TRUE;
      }
      goto err;
    }
  }
#endif

  /* Allocate bitmaps */

  bitmap_size= share->column_bitmap_size;
  if (!(bitmaps= (uchar*) alloc_root(&outparam->mem_root, bitmap_size*3)))
    goto err;
  bitmap_init(&outparam->def_read_set,
              (my_bitmap_map*) bitmaps, share->fields, FALSE);
  bitmap_init(&outparam->def_write_set,
              (my_bitmap_map*) (bitmaps+bitmap_size), share->fields, FALSE);
  bitmap_init(&outparam->tmp_set,
              (my_bitmap_map*) (bitmaps+bitmap_size*2), share->fields, FALSE);
  outparam->default_column_bitmaps();

  /* The table struct is now initialized;  Open the table */
  error= 2;
  if (db_stat)
  {
    int ha_err;
    if ((ha_err= (outparam->file->
                  ha_open(outparam, share->normalized_path.str,
                          (db_stat & HA_READ_ONLY ? O_RDONLY : O_RDWR),
                          (db_stat & HA_OPEN_TEMPORARY ? HA_OPEN_TMP_TABLE :
                           ((db_stat & HA_WAIT_IF_LOCKED) ||
                            (specialflag & SPECIAL_WAIT_IF_LOCKED)) ?
                           HA_OPEN_WAIT_IF_LOCKED :
                           (db_stat & (HA_ABORT_IF_LOCKED | HA_GET_INFO)) ?
                          HA_OPEN_ABORT_IF_LOCKED :
                           HA_OPEN_IGNORE_IF_LOCKED) | ha_open_flags))))
    {
      /* Set a flag if the table is crashed and it can be auto. repaired */
      share->crashed= ((ha_err == HA_ERR_CRASHED_ON_USAGE) &&
                       outparam->file->auto_repair() &&
                       !(ha_open_flags & HA_OPEN_FOR_REPAIR));

      switch (ha_err)
      {
        case HA_ERR_NO_SUCH_TABLE:
      /*
            The table did not exists in storage engine, use same error message
            as if the .frm file didn‘t exist
          */
      error= 1;
      my_errno= ENOENT;
          break;
        case EMFILE:
      /*
            Too many files opened, use same error message as if the .frm
            file can‘t open
           */
          DBUG_PRINT("error", ("open file: %s failed, too many files opened (errno: %d)",
          share->normalized_path.str, ha_err));
      error= 1;
      my_errno= EMFILE;
          break;
        default:
          outparam->file->print_error(ha_err, MYF(0));
          error_reported= TRUE;
          if (ha_err == HA_ERR_TABLE_DEF_CHANGED)
            error= 7;
          break;
      }
      goto err;                                 /* purecov: inspected */
    }
  }

#if defined(HAVE_purify) && !defined(DBUG_OFF)
  bzero((char*) bitmaps, bitmap_size*3);
#endif

  outparam->no_replicate= outparam->file &&
                          test(outparam->file->ha_table_flags() &
                               HA_HAS_OWN_BINLOGGING);
  /* Increment the opened_tables counter, only when open flags set. */
  if (db_stat)
    thd->status_var.opened_tables++;

  DBUG_RETURN (0);

 err:
  if (! error_reported)
    open_table_error(share, error, my_errno, 0);
  delete outparam->file;
#ifdef WITH_PARTITION_STORAGE_ENGINE
  if (outparam->part_info)
    free_items(outparam->part_info->item_free_list);
#endif
  outparam->file= 0;                // For easier error checking
  outparam->db_stat=0;
  free_root(&outparam->mem_root, MYF(0));       // Safe to call on bzero‘d root
  my_free((void *) outparam->alias);
  DBUG_RETURN (error);
}