Commit 0d55b2ea authored by unknown's avatar unknown
Browse files

Bug #18947 CRBR: order in binlog of create table and insert (on different table) not determ 

- move clear of slock until after completion of data events on the binlog
parent 9dea1ccc

sql/ha_ndbcluster_binlog.cc

+31 −22
Original line number Diff line number Diff line
@@ -1720,37 +1720,26 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb, 
      bitmap_init(&slock, schema->slock, 8*SCHEMA_SLOCK_SIZE, false);

      uint node_id= g_ndb_cluster_connection->node_id();

      ndbcluster_get_schema(tmp_share, schema);

      enum SCHEMA_OP_TYPE schema_type= (enum SCHEMA_OP_TYPE)schema->type;

      DBUG_PRINT("info",

                 ("%s.%s: log query_length: %d  query: '%s'  type: %d",

                  schema->db, schema->name,

                  schema->query_length, schema->query,

                  schema->type));

      char key[FN_REFLEN];

      build_table_filename(key, sizeof(key), schema->db, schema->name, "");

      if ((enum SCHEMA_OP_TYPE)schema->type == SOT_CLEAR_SLOCK)

      {

        pthread_mutex_lock(&ndbcluster_mutex);

        NDB_SCHEMA_OBJECT *ndb_schema_object=

          (NDB_SCHEMA_OBJECT*) hash_search(&ndb_schema_objects,

                                           (byte*) key, strlen(key));

        if (ndb_schema_object)

                  schema_type));

      if (schema_type == SOT_CLEAR_SLOCK)

      {

          pthread_mutex_lock(&ndb_schema_object->mutex);

          memcpy(ndb_schema_object->slock, schema->slock,

                 sizeof(ndb_schema_object->slock));

          DBUG_DUMP("ndb_schema_object->slock_bitmap.bitmap",

                    (char*)ndb_schema_object->slock_bitmap.bitmap,

                    no_bytes_in_map(&ndb_schema_object->slock_bitmap));

          pthread_mutex_unlock(&ndb_schema_object->mutex);

          pthread_cond_signal(&injector_cond);

        }

        pthread_mutex_unlock(&ndbcluster_mutex);

        /*

          handle slock after epoch is completed to ensure that

          schema events get inserted in the binlog after any data

          events

        */

        post_epoch_log_list->push_back(schema, mem_root);

        DBUG_RETURN(0);

      }

      if (schema->node_id != node_id)

      {

        int log_query= 0, post_epoch_unlock= 0;

        switch ((enum SCHEMA_OP_TYPE)schema->type)

        switch (schema_type)

        {

        case SOT_DROP_TABLE:

          // fall through
@@ -1944,10 +1933,30 @@ ndb_binlog_thread_handle_schema_event_post_epoch(THD *thd, 
                schema->type));

    int log_query= 0;

    {

      enum SCHEMA_OP_TYPE schema_type= (enum SCHEMA_OP_TYPE)schema->type;

      char key[FN_REFLEN];

      build_table_filename(key, sizeof(key), schema->db, schema->name, "");

      if (schema_type == SOT_CLEAR_SLOCK)

      {

        pthread_mutex_lock(&ndbcluster_mutex);

        NDB_SCHEMA_OBJECT *ndb_schema_object=

          (NDB_SCHEMA_OBJECT*) hash_search(&ndb_schema_objects,

                                           (byte*) key, strlen(key));

        if (ndb_schema_object)

        {

          pthread_mutex_lock(&ndb_schema_object->mutex);

          memcpy(ndb_schema_object->slock, schema->slock,

                 sizeof(ndb_schema_object->slock));

          DBUG_DUMP("ndb_schema_object->slock_bitmap.bitmap",

                    (char*)ndb_schema_object->slock_bitmap.bitmap,

                    no_bytes_in_map(&ndb_schema_object->slock_bitmap));

          pthread_mutex_unlock(&ndb_schema_object->mutex);

          pthread_cond_signal(&injector_cond);

        }

        pthread_mutex_unlock(&ndbcluster_mutex);

        continue;

      }

      NDB_SHARE *share= get_share(key, 0, false, false);

      enum SCHEMA_OP_TYPE schema_type= (enum SCHEMA_OP_TYPE)schema->type;

      switch (schema_type)

      {

      case SOT_DROP_DB: