1berry/MariaDB-server
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

WL#3303 (RBR: Engine-controlled logging format):

Browse Source 
Moving code to check storage engine capabilities to after tables
are locked.  Moving code to cache table flags so that table flags
are read from the storage engine at the beginning of the statement
in addition to when the storage engine is opened.

To handle CREATE-SELECT, the decision function is called after the
table is created and it is called with all tables that are in the select
part of the statement as well as the newly created table.


sql/handler.cc:
  Changing code to cache table flags on a per-statement basis. The table
  flags are now retrieved inside ha_external_lock().
sql/handler.h:
  Extending TABLEOP_HOOKS with postlock() member.
sql/mysql_priv.h:
  Adding prototype declaration of decide_logging_format() function.
sql/sql_base.cc:
  Factoring out code to check capabilities into decide_logging_format().
  Moving code to check engine capabilities to after the tables are locked.
  Correcting a bug causing row-based to not be set when the engines
  were not statement-logging capable.
sql/sql_class.h:
  Adding selected tables as select_create::select_tables member variable.
sql/sql_insert.cc:
  Introducing logic to handle post-locking hook.
  select_create::prepare now uses post-lock hook instead of pre-lock hook.
  Deciding on logging format especially for CREATE-SELECT by calling
  decide_logging_format() in the post-lock hook.
sql/sql_parse.cc:
  Adding selected tables as argument to select_create constructor.
This commit is contained in:
unknown 2007-05-29 17:13:17 +02:00
parent ff3a6b5587
commit 48a3194420
7 changed files with 173 additions and 99 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
sql/handler.cc
View File

@ -3607,7 +3607,15 @@ int handler::ha_external_lock(THD *thd, int lock_type)
taken a table lock), ha_release_auto_increment() was too.
*/
DBUG_ASSERT(next_insert_id == 0);
DBUG_RETURN(external_lock(thd, lock_type));
/*
We cache the table flags if the locking succeeded. Otherwise, we
keep them as they were when they were fetched in ha_open().
*/
int error= external_lock(thd, lock_type);
if (error == 0)
cached_table_flags= table_flags();
DBUG_RETURN(error);
}

22
sql/handler.h
View File

@ -804,19 +804,37 @@ typedef struct st_key_create_information
class TABLEOP_HOOKS
{
public:
TABLEOP_HOOKS() {}
virtual ~TABLEOP_HOOKS() {}
inline void prelock(TABLE **tables, uint count)
{
do_prelock(tables, count);
}
virtual ~TABLEOP_HOOKS() {}
TABLEOP_HOOKS() {}
inline int postlock(TABLE **tables, uint count)
{
return do_postlock(tables, count);
}
private:
/* Function primitive that is called prior to locking tables */
virtual void do_prelock(TABLE **tables, uint count)
{
/* Default is to do nothing */
}
/**
Primitive called after tables are locked.
If an error is returned, the tables will be unlocked and error
handling start.
@return Error code or zero.
*/
virtual int do_postlock(TABLE **tables, uint count)
{
return 0; /* Default is to do nothing */
}
};
typedef struct st_savepoint SAVEPOINT;

1
sql/mysql_priv.h
View File

@ -1270,6 +1270,7 @@ int simple_open_n_lock_tables(THD *thd,TABLE_LIST *tables);
bool open_and_lock_tables(THD *thd,TABLE_LIST *tables);
bool open_normal_and_derived_tables(THD *thd, TABLE_LIST *tables, uint flags);
int lock_tables(THD *thd, TABLE_LIST *tables, uint counter, bool *need_reopen);
int decide_logging_format(THD *thd, TABLE_LIST *tables);
TABLE *open_temporary_table(THD *thd, const char *path, const char *db,
const char *table_name, bool link_in_list);
bool rm_temporary_table(handlerton *base, char *path);

202
sql/sql_base.cc
View File

@ -3547,6 +3547,116 @@ static void mark_real_tables_as_free_for_reuse(TABLE_LIST *table)
}
/**
Decide on logging format to use for the statement.
Compute the capabilities vector for the involved storage engines
and mask out the flags for the binary log. Right now, the binlog
flags only include the capabilities of the storage engines, so this
is safe.
We now have three alternatives that prevent the statement from
being loggable:
1. If there are no capabilities left (all flags are clear) it is
not possible to log the statement at all, so we roll back the
statement and report an error.
2. Statement mode is set, but the capabilities indicate that
statement format is not possible.
3. Row mode is set, but the capabilities indicate that row
format is not possible.
4. Statement is unsafe, but the capabilities indicate that row
format is not possible.
If we are in MIXED mode, we then decide what logging format to use:
1. If the statement is unsafe, row-based logging is used.
2. If statement-based logging is not possible, row-based logging is
used.
3. Otherwise, statement-based logging is used.
@param thd Client thread
@param tables Tables involved in the query
*/
int decide_logging_format(THD *thd, TABLE_LIST *tables)
{
if (mysql_bin_log.is_open() && (thd->options & OPTION_BIN_LOG))
{
handler::Table_flags binlog_flags= ~handler::Table_flags();
for (TABLE_LIST *table= tables; table; table= table->next_global)
if (!table->placeholder() && table->lock_type >= TL_WRITE_ALLOW_WRITE)
{
#define FLAGSTR(S,F) ((S) & (F) ? #F " " : "")
#ifndef DBUG_OFF
ulonglong flags= table->table->file->ha_table_flags();
DBUG_PRINT("info", ("table: %s; ha_table_flags: %s%s",
table->table_name,
FLAGSTR(flags, HA_BINLOG_STMT_CAPABLE),
FLAGSTR(flags, HA_BINLOG_ROW_CAPABLE)));
#endif
binlog_flags &= table->table->file->ha_table_flags();
}
binlog_flags&= HA_BINLOG_FLAGS;
DBUG_PRINT("info", ("binlog_flags: %s%s",
FLAGSTR(binlog_flags, HA_BINLOG_STMT_CAPABLE),
FLAGSTR(binlog_flags, HA_BINLOG_ROW_CAPABLE)));
DBUG_PRINT("info", ("thd->variables.binlog_format: %ld",
thd->variables.binlog_format));
int error= 0;
if (binlog_flags == 0)
{
error= ER_BINLOG_ENGINES_INCOMPATIBLE;
}
else if (thd->variables.binlog_format == BINLOG_FORMAT_STMT &&
(binlog_flags & HA_BINLOG_STMT_CAPABLE) == 0)
{
error= ER_BINLOG_STMT_FORMAT_FORBIDDEN;
}
else if ((thd->variables.binlog_format == BINLOG_FORMAT_ROW ||
thd->lex->is_stmt_unsafe()) &&
(binlog_flags & HA_BINLOG_ROW_CAPABLE) == 0)
{
error= ER_BINLOG_ROW_FORMAT_FORBIDDEN;
}
DBUG_PRINT("info", ("error: %d", error));
if (error)
{
ha_rollback_stmt(thd);
my_error(error, MYF(0));
return -1;
}
/*
We switch to row-based format if we are in mixed mode and one of
the following are true:
1. If the statement is unsafe
2. If statement format cannot be used
Observe that point to cannot be decided before the tables
involved in a statement has been checked, i.e., we cannot put
this code in reset_current_stmt_binlog_row_based(), it has to be
here.
*/
if (thd->lex->is_stmt_unsafe() ||
(binlog_flags & HA_BINLOG_STMT_CAPABLE) == 0)
{
thd->set_current_stmt_binlog_row_based_if_mixed();
}
}
return 0;
}
/*
Lock all tables in list
@ -3587,94 +3697,8 @@ int lock_tables(THD *thd, TABLE_LIST *tables, uint count, bool *need_reopen)
DBUG_ASSERT(!thd->prelocked_mode || !thd->lex->requires_prelocking());
*need_reopen= FALSE;
if (mysql_bin_log.is_open() && (thd->options | OPTION_BIN_LOG))
{
/*
Compute the capabilities vector for the involved storage engines
and mask out the flags for the binary log. Right now, the binlog
flags only include the capabilities of the storage engines, so
this is safe.
*/
handler::Table_flags binlog_flags= ~handler::Table_flags();
DBUG_PRINT("info", ("HA_BINLOG_FLAGS: 0x%0llx",
(ulonglong) HA_BINLOG_FLAGS));
for (table= tables; table; table= table->next_global)
if (!table->placeholder() && table->lock_type >= TL_WRITE_ALLOW_WRITE)
{
DBUG_PRINT("info", ("table: %s; ha_table_flags: 0x%0llx",
table->table_name,
(ulonglong) table->table->file->ha_table_flags()));
binlog_flags &= table->table->file->ha_table_flags();
}
binlog_flags&= HA_BINLOG_FLAGS;
DBUG_PRINT("info", ("binlog_flags: 0x%0llx", (ulonglong) binlog_flags));
DBUG_PRINT("info", ("thd->variables.binlog_format: %ld",
thd->variables.binlog_format));
/*
We have three alternatives that prevent the statement from being
loggable:
1. If there are no capabilities left (all flags are clear) it is
not possible to log the statement at all, so we roll back the
statement and report an error.
2. Statement mode is set, but the capabilities indicate that
statement format is not possible.
3. Row mode is set, but the capabilities indicate that row
format is not possible.
4. Statement is unsafe, but the capabilities indicate that row
format is not possible.
*/
int error= 0;
if (binlog_flags == 0)
{
error= ER_BINLOG_ENGINES_INCOMPATIBLE;
}
else if (thd->variables.binlog_format == BINLOG_FORMAT_STMT &&
(binlog_flags & HA_BINLOG_STMT_CAPABLE) == 0)
{
error= ER_BINLOG_STMT_FORMAT_FORBIDDEN;
}
else if ((thd->variables.binlog_format == BINLOG_FORMAT_ROW ||
thd->lex->is_stmt_unsafe()) &&
(binlog_flags & HA_BINLOG_ROW_CAPABLE) == 0)
{
error= ER_BINLOG_ROW_FORMAT_FORBIDDEN;
}
DBUG_PRINT("info", ("error: %d", error));
if (error)
{
ha_rollback_stmt(thd);
my_error(error, MYF(0));
DBUG_RETURN(-1);
}
/*
We switch to row-based format if we are in mixed mode and one of
the following are true:
1. If the statement is unsafe
2. If statement format cannot be used
Observe that point to cannot be decided before the tables
involved in a statement has been checked, i.e., we cannot put
this code in reset_current_stmt_binlog_row_based(), it has to be
here.
*/
if (thd->lex->is_stmt_unsafe() ||
(binlog_flags | HA_BINLOG_STMT_CAPABLE) == 0)
{
thd->set_current_stmt_binlog_row_based_if_mixed();
}
}
if (!tables && !thd->lex->requires_prelocking())
DBUG_RETURN(0);
DBUG_RETURN(decide_logging_format(thd, tables));
/*
We need this extra check for thd->prelocked_mode because we want to avoid
@ -3727,6 +3751,7 @@ int lock_tables(THD *thd, TABLE_LIST *tables, uint count, bool *need_reopen)
}
DBUG_RETURN(-1);
}
if (thd->lex->requires_prelocking() &&
thd->lex->sql_command != SQLCOM_LOCK_TABLES)
{
@ -3793,7 +3818,8 @@ int lock_tables(THD *thd, TABLE_LIST *tables, uint count, bool *need_reopen)
thd->prelocked_mode= PRELOCKED_UNDER_LOCK_TABLES;
}
}
DBUG_RETURN(0);
DBUG_RETURN(decide_logging_format(thd, tables));
}

6
sql/sql_class.h
View File

@ -1921,6 +1921,7 @@ class select_insert :public select_result_interceptor {
class select_create: public select_insert {
ORDER *group;
TABLE_LIST *create_table;
TABLE_LIST *select_tables;
List<create_field> *extra_fields;
List<Key> *keys;
HA_CREATE_INFO *create_info;
@ -1930,10 +1931,11 @@ public:
HA_CREATE_INFO *create_info_par,
List<create_field> &fields_par,
List<Key> &keys_par,
List<Item> &select_fields,enum_duplicates duplic, bool ignore)
List<Item> &select_fields,enum_duplicates duplic, bool ignore,
TABLE_LIST *select_tables_arg)
:select_insert (NULL, NULL, &select_fields, 0, 0, duplic, ignore),
create_table(table_arg), extra_fields(&fields_par),keys(&keys_par),
create_info(create_info_par)
create_info(create_info_par), select_tables(select_tables_arg)
{}
int prepare(List<Item> &list, SELECT_LEX_UNIT *u);

28
sql/sql_insert.cc
View File

@ -3106,8 +3106,15 @@ static TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
table->reginfo.lock_type=TL_WRITE;
hooks->prelock(&table, 1); // Call prelock hooks
if (! ((*lock)= mysql_lock_tables(thd, &table, 1,
MYSQL_LOCK_IGNORE_FLUSH, &not_used)))
MYSQL_LOCK_IGNORE_FLUSH, &not_used)) ||
hooks->postlock(&table, 1))
{
if (*lock)
{
mysql_unlock_tables(thd, *lock);
*lock= 0;
}
VOID(pthread_mutex_lock(&LOCK_open));
hash_delete(&open_cache,(byte*) table);
VOID(pthread_mutex_unlock(&LOCK_open));
@ -3146,24 +3153,35 @@ select_create::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
*/
class MY_HOOKS : public TABLEOP_HOOKS {
public:
MY_HOOKS(select_create *x) : ptr(x) { }
MY_HOOKS(select_create *x, TABLE_LIST *create_table,
TABLE_LIST *select_tables)
: ptr(x), all_tables(*create_table)
{
all_tables.next_global= select_tables;
}
private:
virtual void do_prelock(TABLE **tables, uint count)
virtual int do_postlock(TABLE **tables, uint count)
{
THD *thd= const_cast<THD*>(ptr->get_thd());
if (int error= decide_logging_format(thd, &all_tables))
return error;
TABLE const *const table = *tables;
if (ptr->get_thd()->current_stmt_binlog_row_based &&
if (thd->current_stmt_binlog_row_based &&
!table->s->tmp_table &&
!ptr->get_create_info()->table_existed)
{
ptr->binlog_show_create_table(tables, count);
}
return 0;
}
select_create *ptr;
TABLE_LIST all_tables;
};
MY_HOOKS hooks(this);
MY_HOOKS hooks(this, create_table, select_tables);
hook_ptr= &hooks;
unit= u;

3
sql/sql_parse.cc
View File

@ -2193,7 +2193,8 @@ mysql_execute_command(THD *thd)
lex->key_list,
select_lex->item_list,
lex->duplicates,
lex->ignore)))
lex->ignore,
select_tables)))
{
/*
CREATE from SELECT give its SELECT_LEX for SELECT,