MDEV-39227: Schema Versioning - Server-side

sql/handler.cc

@@ -7098,12 +7098,14 @@ bool ha_table_exists(THD *thd, const LEX_CSTRING *db,
   {
     if (!hton)
       hton= &dummy;
-    *hton= element->share->db_type();
-    *is_sequence= element->share->table_type == TABLE_TYPE_SEQUENCE;
-    if (*hton != view_pseudo_hton && element->share->tabledef_version.length &&
+    *hton= element->current()->share->db_type();
+    *is_sequence= element->current()->share->table_type == TABLE_TYPE_SEQUENCE;
+    if (*hton != view_pseudo_hton &&
+        element->current()->share->tabledef_version.length &&
         table_id &&
         (table_id->str= (uchar*)
-         thd->memdup(element->share->tabledef_version.str, MY_UUID_SIZE)))
+         thd->memdup(element->current()->share->tabledef_version.str,
+                     MY_UUID_SIZE)))
       table_id->length= MY_UUID_SIZE;
     tdc_unlock_share(element);
     DBUG_RETURN(TRUE);

sql/mdl.cc

@@ -3719,6 +3719,14 @@ void MDL_context::release_transactional_locks(THD *thd)
   DBUG_ASSERT(!(thd->server_status &
                 (SERVER_STATUS_IN_TRANS | SERVER_STATUS_IN_TRANS_READONLY)));
   release_locks_stored_before(MDL_STATEMENT, NULL);
+  /*
+    Release schema bindings before MDL_TRANSACTION tickets. Each binding
+    holds a ref_count pin on its TABLE_SHARE_VERSION; dropping the pin
+    here may trigger GC of OLDER versions. Keeping the MDL_TRANSACTION
+    ticket held during this step prevents concurrent X-locking DDL from
+    racing with the share's tdc->LOCK_table_share manipulations.
+  */
+  thd->release_transaction_schema_bindings();
   release_locks_stored_before(MDL_TRANSACTION, NULL);
   DBUG_VOID_RETURN;
 }

sql/sql_base.cc

@@ -289,11 +289,14 @@ static my_bool list_open_tables_callback(void *el, void *a)
   (*arg->start_list)->in_use= 0;
 
   mysql_mutex_lock(&element->LOCK_table_share);
-  All_share_tables_list::Iterator it(element->all_tables);
   TABLE *table;
-  while ((table= it++))
-    if (table->in_use)
-      ++(*arg->start_list)->in_use;
+  for (TABLE_SHARE_VERSION *v= element->versions_head; v; v= v->next)
+  {
+    All_share_tables_list::Iterator it(v->all_tables);
+    while ((table= it++))
+      if (table->in_use)
+        ++(*arg->start_list)->in_use;
+  }
   mysql_mutex_unlock(&element->LOCK_table_share);
   (*arg->start_list)->locked= 0;                   /* Obsolete. */
   arg->start_list= &(*arg->start_list)->next;
@@ -476,28 +479,51 @@ static my_bool tc_collect_used_shares(void *el, void *a)
 
   DYNAMIC_ARRAY *shares= &arg->shares;
   mysql_mutex_lock(&element->LOCK_table_share);
-  if (element->ref_count > 0 && !element->share->is_view)
+  /*
+    "Table is in use" per FLUSH TABLES semantics = any version of this
+    name has ref_count > 0. After a lock-free DDL the chain may briefly
+    hold an OLDER version pinned by an in-flight transaction together
+    with a CURRENT version that no one has touched yet — we still need
+    to treat the table as in use.
+
+    The do_flush properties (online_backup, table_category) and the
+    share we push for downstream processing are taken from current(),
+    which is what FLUSH TABLES will ultimately drain.
+  */
+  bool any_in_use= false;
+  if (element->current())
+  {
+    for (TABLE_SHARE_VERSION *v= element->versions_head; v; v= v->next)
+    {
+      if (v->ref_count > 0)
+      {
+        any_in_use= true;
+        break;
+      }
+    }
+  }
+  if (any_in_use && !element->current()->share->is_view)
   {
-    DBUG_ASSERT(element->share);
+    DBUG_ASSERT(element->current()->share);
     bool do_flush= 0;
     switch (arg->flush_type) {
     case FLUSH_ALL:
       do_flush= 1;
       break;
     case FLUSH_NON_TRANS_TABLES:
-      if (!element->share->online_backup &&
-          element->share->table_category == TABLE_CATEGORY_USER)
+      if (!element->current()->share->online_backup &&
+          element->current()->share->table_category == TABLE_CATEGORY_USER)
         do_flush= 1;
       break;
     case FLUSH_SYS_TABLES:
-      if (!element->share->online_backup &&
-          element->share->table_category != TABLE_CATEGORY_USER)
+      if (!element->current()->share->online_backup &&
+          element->current()->share->table_category != TABLE_CATEGORY_USER)
         do_flush= 1;
     }
     if (do_flush)
     {
-      element->ref_count++;                       // Protect against delete
-      if (push_dynamic(shares, (uchar*) &element->share))
+      element->current()->ref_count++;            // Protect against delete
+      if (push_dynamic(shares, (uchar*) &element->current()->share))
         result= TRUE;
     }
   }
@@ -600,7 +626,7 @@ bool flush_tables(THD *thd, flush_tables_type flag)
   {
     TABLE_SHARE *share= *dynamic_element(&collect_arg.shares, i,
                                          TABLE_SHARE**);
-    TABLE *table= tc_acquire_table(thd, share->tdc);
+    TABLE *table= tc_acquire_table(thd, share->tdc->current());
     if (table)
     {
       (void) table->file->extra(HA_EXTRA_FLUSH);
@@ -745,7 +771,7 @@ close_all_tables_for_name(THD *thd, TABLE_SHARE *share,
                           TABLE *skip_table)
 {
   DBUG_ASSERT(!share->tmp_table);
-  DBUG_ASSERT(share->tdc->flushed);
+  DBUG_ASSERT(share->version->flushed);
 
   char key[MAX_DBKEY_LENGTH];
   size_t key_length= share->table_cache_key.length;
@@ -2245,7 +2271,7 @@ bool open_table(THD *thd, TABLE_LIST *table_list, Open_table_context *ot_ctx)
   if (!(flags & MYSQL_OPEN_IGNORE_FLUSH))
 #endif
   {
-    if (share->tdc->flushed)
+    if (share->version->flushed)
     {
       /*
         We already have an MDL lock. But we have encountered an old
@@ -2277,7 +2303,7 @@ bool open_table(THD *thd, TABLE_LIST *table_list, Open_table_context *ot_ctx)
       goto retry_share;
     }
 
-    if (thd->open_tables && thd->open_tables->s->tdc->flushed)
+    if (thd->open_tables && thd->open_tables->s->version->flushed)
     {
       /*
         If the version changes while we're opening the tables,

sql/sql_class.cc

@@ -863,6 +863,7 @@ THD::THD(my_thread_id id, bool is_wsrep_applier)
   is_slave_error= FALSE;
   my_hash_clear(&handler_tables_hash);
   my_hash_clear(&ull_hash);
+  my_hash_clear(&transaction_schema_bindings);
   tmp_table=0;
   cuted_fields= 0L;
   limit_found_rows= 0;
@@ -1753,6 +1754,8 @@ void THD::cleanup(void)
 
   my_hash_free(&user_vars);
   my_hash_free(&sequences);
+  if (my_hash_inited(&transaction_schema_bindings))
+    my_hash_free(&transaction_schema_bindings);
   sp_caches_clear();
   statement_rcontext_reinit();
   auto_inc_intervals_forced.empty();
@@ -6642,6 +6645,125 @@ void THD::leave_locked_tables_mode()
   locked_tables_mode= LTM_NONE;
 }
 
+
+/*
+  Per-transaction schema bindings.
+
+  See declarations in sql_class.h. Bindings are populated by
+  tdc_acquire_share() on first open of a name within a transaction and
+  cleared by MDL_context::release_transactional_locks() at end of
+  transaction. Each binding pins its TABLE_SHARE_VERSION (ref_count +1)
+  so that the version stays alive across statement boundaries — letting
+  the transaction continue using its snapshot even after a concurrent
+  DDL has installed a newer version.
+*/
+
+namespace {
+struct Schema_binding
+{
+  uchar key[NAME_LEN + 1 + NAME_LEN + 1];
+  uint key_length;
+  TABLE_SHARE_VERSION *version;
+};
+
+extern "C" const uchar *schema_binding_key(const void *binding, size_t *length,
+                                           my_bool)
+{
+  const Schema_binding *b= static_cast<const Schema_binding*>(binding);
+  *length= b->key_length;
+  return b->key;
+}
+} // anonymous namespace
+
+
+TABLE_SHARE_VERSION *THD::lookup_schema_binding(const uchar *key,
+                                                uint key_length)
+{
+  if (!my_hash_inited(&transaction_schema_bindings))
+    return NULL;
+  Schema_binding *b= reinterpret_cast<Schema_binding*>(
+    my_hash_search(&transaction_schema_bindings, key, key_length));
+  return b ? b->version : NULL;
+}
+
+
+bool THD::add_schema_binding(const uchar *key, uint key_length,
+                             TABLE_SHARE_VERSION *version)
+{
+  if (!my_hash_inited(&transaction_schema_bindings))
+  {
+    if (my_hash_init(PSI_INSTRUMENT_ME, &transaction_schema_bindings,
+                     &my_charset_bin, 0, 0, 0, schema_binding_key,
+                     my_free, 0))
+      return true;
+  }
+  Schema_binding *b= static_cast<Schema_binding*>(
+    my_malloc(PSI_INSTRUMENT_ME, sizeof(*b), MYF(MY_WME)));
+  if (!b)
+    return true;
+  DBUG_ASSERT(key_length <= sizeof(b->key));
+  memcpy(b->key, key, key_length);
+  b->key_length= key_length;
+  b->version= version;
+
+  /*
+    The binding pins its version by holding a +1 on ref_count. The pin is
+    dropped by release_transaction_schema_bindings() at end of transaction,
+    or earlier by remove_schema_binding() when the share is being torn
+    down.
+  */
+  mysql_mutex_lock(&version->share->tdc->LOCK_table_share);
+  version->ref_count++;
+  mysql_mutex_unlock(&version->share->tdc->LOCK_table_share);
+
+  if (my_hash_insert(&transaction_schema_bindings, reinterpret_cast<uchar*>(b)))
+  {
+    /* OOM after the ref bump — undo the pin and free. */
+    mysql_mutex_lock(&version->share->tdc->LOCK_table_share);
+    --version->ref_count;
+    mysql_mutex_unlock(&version->share->tdc->LOCK_table_share);
+    my_free(b);
+    return true;
+  }
+  return false;
+}
+
+
+void THD::remove_schema_binding(const uchar *key, uint key_length)
+{
+  if (!my_hash_inited(&transaction_schema_bindings))
+    return;
+  Schema_binding *b= reinterpret_cast<Schema_binding*>(
+    my_hash_search(&transaction_schema_bindings, key, key_length));
+  if (!b)
+    return;
+  TABLE_SHARE *share= b->version->share;
+  /* my_hash_delete frees the Schema_binding (via the registered my_free). */
+  my_hash_delete(&transaction_schema_bindings, reinterpret_cast<uchar*>(b));
+  /* Drop the binding's ref. tdc_release_share dispatches CURRENT vs OLDER. */
+  tdc_release_share(share);
+}
+
+
+void THD::release_transaction_schema_bindings()
+{
+  if (!my_hash_inited(&transaction_schema_bindings) ||
+      transaction_schema_bindings.records == 0)
+    return;
+  /*
+    Walk every binding by index and release its version's ref. Calling
+    tdc_release_share may GC the version (if OLDER and ref reaches 0) but
+    does not touch our Schema_binding entry; my_hash_reset below frees those.
+  */
+  for (ulong i= 0; i < transaction_schema_bindings.records; i++)
+  {
+    Schema_binding *b= reinterpret_cast<Schema_binding*>(
+      my_hash_element(&transaction_schema_bindings, i));
+    tdc_release_share(b->version->share);
+  }
+  my_hash_reset(&transaction_schema_bindings);
+}
+
 void THD::get_definer(LEX_USER *definer, bool role)
 {
   binlog_invoker(role);

sql/sql_class.h

@@ -113,6 +113,7 @@ class rpl_io_thread_info;
 class rpl_sql_thread_info;
 #ifdef HAVE_REPLICATION
 struct Slave_info;
+struct TABLE_SHARE_VERSION;
 #endif
 
 enum enum_ha_read_modes { RFIRST, RNEXT, RPREV, RLAST, RKEY, RNEXT_SAME };
@@ -3407,6 +3408,20 @@ class THD: public THD_count, /* this must be first */
     chapter 'Miscellaneous functions', for functions GET_LOCK, RELEASE_LOCK.
   */
   HASH ull_hash;
+  /*
+    Per-transaction bindings of (db, table_name) to the TABLE_SHARE_VERSION
+    the transaction first opened. Populated by tdc_acquire_share() on first
+    open of a name within the transaction; released at end of transaction
+    by release_transaction_schema_bindings().
+
+    Each binding holds a +1 on its version's ref_count for the lifetime of
+    the transaction. This pin keeps the version alive (and thus visible to
+    bound-path lookups in tdc_acquire_share) even after a concurrent DDL
+    has installed a newer version as the chain tail.
+
+    Initialized lazily on first use.
+  */
+  HASH transaction_schema_bindings;
   /* Hash of used sequences (for PREVIOUS value) */
   HASH sequences;
 #ifdef DBUG_ASSERT_EXISTS
@@ -5716,6 +5731,25 @@ class THD: public THD_count, /* this must be first */
     if (!in_active_multi_stmt_transaction())
       mdl_context.release_transactional_locks(this);
   }
+
+  /*
+    Per-transaction schema binding API:
+    - lookup_schema_binding() returns the TABLE_SHARE_VERSION the txn is bound
+      to for this name, or NULL if not bound yet.
+    - add_schema_binding() records a binding and bumps version->ref_count;
+      returns true on OOM.
+    - remove_schema_binding() drops a single binding (and its ref) — used
+      when the share is being torn down out-of-band.
+    - release_transaction_schema_bindings() clears all bindings; called at
+      end of transaction from MDL_context::release_transactional_locks().
+  */
+  TABLE_SHARE_VERSION *lookup_schema_binding(const uchar *key,
+                                             uint key_length);
+  bool add_schema_binding(const uchar *key, uint key_length,
+                          TABLE_SHARE_VERSION *version);
+  void remove_schema_binding(const uchar *key, uint key_length);
+  void release_transaction_schema_bindings();
+
   int decide_logging_format(TABLE_LIST *tables);
 
   /*

sql/sql_handler.cc

@@ -1191,7 +1191,7 @@ void mysql_ha_flush(THD *thd)
         ((hash_tables->table->mdl_ticket &&
          hash_tables->table->mdl_ticket->has_pending_conflicting_lock()) ||
          (!hash_tables->table->s->tmp_table &&
-          hash_tables->table->s->tdc->flushed)))
+          hash_tables->table->s->version->flushed)))
       mysql_ha_close_table(hash_tables);
   }
 

sql/sql_insert.cc

@@ -3888,7 +3888,7 @@ bool Delayed_insert::handle_inserts(void)
 
   THD_STAGE_INFO(&thd, stage_insert);
   max_rows= delayed_insert_limit;
-  if (thd.killed || table->s->tdc->flushed)
+  if (thd.killed || table->s->version->flushed)
   {
     thd.set_killed(KILL_SYSTEM_THREAD);
     max_rows= ULONG_MAX;                     // Do as much as possible