database/mariadb
1
0
Fork 0
mirror of https://github.com/MariaDB/server.git synced 2025-07-18 23:03:28 +03:00
Code Activity
Files
d52ddae57b5875c34e8971a645cba50732cf2ded
mariadb/sql/ha_sequence.cc
Yuchen Pei d52ddae57b MDEV-22491 Support mariadb-check and CHECK TABLE with SEQUENCE 
The check go through the following steps:

1. Run check on the underlying engine. If not ok, then return.
2. Check that there's only one row in the table, and
   2.1 warn if more than one row
   2.2 return HA_ADMIN_CORRUPT if fewer than one row (i.e. 0 rows)
3. If the sequence is not initialised (e.g. after an ALTER TABLE ...
   SEQUENCE=1), initialise the sequence by reading the sequence
   metadata from the table. This will also flush the next_free_value,
   i.e. set it to the next not cached value (SEQUENCE::reserved_until)
4. Check that the sequence metadata is valid, i.e. nothing out of
   order e.g. minvalue < maxvalue etc. If invalid it reports
   HA_ERR_SEQUENCE_INVALID_DATA
5. Check that the sequence has not been exhausted. It reports
   ER_SEQUENCE_RUN_OUT as a warning if and only if a SELECT NEXTVAL
   would do so

Limitations:

1. The check is independent of flags, so the vanilla check is the same
   as CHECK ... EXTENDED or CHECK ... FOR UPGRADE etc.
2. When the check discovers invalid metadata from the table,
   subsequent SELECT NEXTVAL will carry on (or fail) without this
   piece of knowledge, independent of the CHECK. This is to ensure
   consistency, i.e. CHECK does not modify behaviour of SELECT, and if
   anything it makes more sense that SELECT reports
   HA_ERR_SEQUENCE_INVALID_DATA in this case, regardless of prior
   CHECK
2025-05-05 11:36:55 +10:00

562 lines
16 KiB
C++
Raw Blame History

/*
Copyright (c) 2017, Aliyun and/or its affiliates.
Copyright (c) 2017, 2020, 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 St, Fifth Floor, Boston, MA 02110-1301 USA
*/
#include "mariadb.h"
#include "sql_list.h"
#include "table.h"
#include "sql_table.h"
#include "sql_sequence.h"
#include "ha_sequence.h"
#include "sql_plugin.h"
#include "mysql/plugin.h"
#include "sql_priv.h"
#include "sql_parse.h"
#include "sql_update.h"
#include "sql_base.h"
#include "log_event.h"
#ifdef WITH_WSREP
#include "wsrep_trans_observer.h" /* wsrep_start_transaction() */
#endif
/*
Table flags we should inherit and disable from the original engine.
We add HA_STATS_RECORDS_IS_EXACT as ha_sequence::info() will ensure
that records is always 1
*/
#define SEQUENCE_ENABLED_TABLE_FLAGS (HA_STATS_RECORDS_IS_EXACT | \
HA_PERSISTENT_TABLE)
#define SEQUENCE_DISABLED_TABLE_FLAGS (HA_CAN_SQL_HANDLER | \
HA_CAN_INSERT_DELAYED | \
HA_BINLOG_STMT_CAPABLE)
handlerton *sql_sequence_hton;
/*
Create a sequence handler
*/
ha_sequence::ha_sequence(handlerton *hton, TABLE_SHARE *share)
:handler(hton, share), write_locked(0)
{
sequence= share->sequence;
DBUG_ASSERT(share->sequence);
}
/**
Destructor method must remove the underlying handler
*/
ha_sequence::~ha_sequence()
{
delete file;
}
/**
Sequence table open method
@param name Path to file (dbname and tablename)
@param mode mode
@param flags Flags how to open file
RETURN VALUES
@retval 0 Success
@retval != 0 Failure
*/
int ha_sequence::open(const char *name, int mode, uint flags)
{
int error;
DBUG_ENTER("ha_sequence::open");
DBUG_ASSERT(table->s == table_share && file);
file->table= table;
if (likely(!(error= file->open(name, mode, flags))))
{
/*
Allocate ref in table's mem_root. We can't use table's ref
as it's allocated by ha_ caller that allocates this.
*/
ref_length= file->ref_length;
if (!(ref= (uchar*) alloc_root(&table->mem_root,ALIGN_SIZE(ref_length)*2)))
{
file->ha_close();
error=HA_ERR_OUT_OF_MEM;
DBUG_RETURN(error);
}
file->ref= ref;
file->dup_ref= dup_ref= ref+ALIGN_SIZE(file->ref_length);
/*
ha_open() sets the following for us. We have to set this for the
underlying handler
*/
file->cached_table_flags= (file->table_flags() | HA_REUSES_FILE_NAMES);
file->reset_statistics();
internal_tmp_table= file->internal_tmp_table=
MY_TEST(flags & HA_OPEN_INTERNAL_TABLE);
reset_statistics();
/*
Don't try to read the initial row if the call is part of CREATE, REPAIR
or FLUSH
*/
if (!(flags & (HA_OPEN_FOR_CREATE | HA_OPEN_FOR_REPAIR |
HA_OPEN_FOR_FLUSH)))
{
if (unlikely((error= table->s->sequence->read_initial_values(table))))
file->ha_close();
}
else if (!table->s->tmp_table)
table->internal_set_needs_reopen(true);
/*
The following is needed to fix comparison of rows in
ha_update_first_row() for InnoDB
*/
if (!error)
memcpy(table->record[1], table->s->default_values, table->s->reclength);
}
DBUG_RETURN(error);
}
/*
Clone the sequence. Needed if table is used by range optimization
(Very, very unlikely)
*/
handler *ha_sequence::clone(const char *name, MEM_ROOT *mem_root)
{
ha_sequence *new_handler;
DBUG_ENTER("ha_sequence::clone");
if (!(new_handler= new (mem_root) ha_sequence(ht, table_share)))
DBUG_RETURN(NULL);
/*
Allocate new_handler->ref here because otherwise ha_open will allocate it
on this->table->mem_root and we will not be able to reclaim that memory
when the clone handler object is destroyed.
*/
if (!(new_handler->ref= (uchar*) alloc_root(mem_root,
ALIGN_SIZE(ref_length)*2)))
goto err;
if (new_handler->ha_open(table, name,
table->db_stat,
HA_OPEN_IGNORE_IF_LOCKED | HA_OPEN_NO_PSI_CALL))
goto err;
/* Reuse original storage engine data for duplicate key reference */
new_handler->ref= file->ref;
new_handler->ref_length= file->ref_length;
new_handler->dup_ref= file->dup_ref;
DBUG_RETURN((handler*) new_handler);
err:
delete new_handler;
DBUG_RETURN(NULL);
}
/*
Map the create table to the original storage engine
*/
int ha_sequence::create(const char *name, TABLE *form,
HA_CREATE_INFO *create_info)
{
DBUG_ASSERT(create_info->sequence);
/* Sequence tables has one and only one row */
create_info->max_rows= create_info->min_rows= 1;
return (file->create(name, form, create_info));
}
/**
Sequence write row method.
A sequence table has only one row. Any inserts in the table
will update this row.
@retval 0 Success
@retval != 0 Failure
NOTES:
write_locked is set if we are called from SEQUENCE::next_value
In this case the mutex is already locked and we should not update
the sequence with 'buf' as the sequence object is already up to date.
*/
int ha_sequence::write_row(const uchar *buf)
{
int error;
sequence_definition tmp_seq;
bool sequence_locked;
THD *thd= table->in_use;
DBUG_ENTER("ha_sequence::write_row");
DBUG_ASSERT(table->record[0] == buf);
/*
Log to binary log even if this function has been called before
(The function ends by setting row_logging to 0)
*/
row_logging= row_logging_init;
if (unlikely(sequence->initialized == SEQUENCE::SEQ_IN_PREPARE))
{
/* This calls is from ha_open() as part of create table */
DBUG_RETURN(file->write_row(buf));
}
if (unlikely(sequence->initialized == SEQUENCE::SEQ_IN_ALTER))
{
int error= 0;
/* This is called from alter table */
tmp_seq.read_fields(table);
if (tmp_seq.check_and_adjust(thd, 0))
DBUG_RETURN(HA_ERR_SEQUENCE_INVALID_DATA);
sequence->copy(&tmp_seq);
if (likely(!(error= file->write_row(buf))))
sequence->initialized= SEQUENCE::SEQ_READY_TO_USE;
row_logging= 0;
DBUG_RETURN(error);
}
if (unlikely(sequence->initialized != SEQUENCE::SEQ_READY_TO_USE))
DBUG_RETURN(HA_ERR_WRONG_COMMAND);
sequence_locked= write_locked;
if (!write_locked) // If not from next_value()
{
/*
User tries to write a full row directly to the sequence table with
INSERT or LOAD DATA.
- Get an exclusive lock for the table. This is needed to ensure that
we excute all full inserts (same as ALTER SEQUENCE) in same order
on master and slaves
- Check that the new row is an accurate SEQUENCE object
*/
/* mark a full binlog image insert to force non-parallel slave */
thd->transaction->stmt.mark_trans_did_ddl();
if (table->s->tmp_table == NO_TMP_TABLE &&
thd->mdl_context.upgrade_shared_lock(table->mdl_ticket,
MDL_EXCLUSIVE,
thd->variables.
lock_wait_timeout))
DBUG_RETURN(ER_LOCK_WAIT_TIMEOUT);
tmp_seq.read_fields(table);
if (tmp_seq.check_and_adjust(thd, 0))
DBUG_RETURN(HA_ERR_SEQUENCE_INVALID_DATA);
/*
Lock sequence to ensure that no one can come in between
while sequence, table and binary log are updated.
*/
sequence->write_lock(table);
}
#ifdef WITH_WSREP
if (WSREP_ON && WSREP(thd) && wsrep_thd_is_local(thd))
{
if (sequence_locked &&
(wsrep_thd_is_SR(thd) || wsrep_streaming_enabled(thd)))
{
my_error(ER_NOT_SUPPORTED_YET, MYF(0),
"SEQUENCEs with streaming replication in Galera cluster");
DBUG_RETURN(HA_ERR_UNSUPPORTED);
}
/*
We need to start Galera transaction for select NEXT VALUE FOR
sequence if it is not yet started. Note that ALTER is handled
as TOI.
*/
if (!thd->wsrep_trx().active())
{
wsrep_start_transaction(thd, thd->wsrep_next_trx_id());
}
}
#endif
if (likely(!(error= file->update_first_row(buf))))
{
Log_func *log_func= Write_rows_log_event::binlog_row_logging_function;
if (!sequence_locked)
sequence->copy(&tmp_seq);
rows_stats.updated++;
/* We have to do the logging while we hold the sequence mutex */
error= binlog_log_row(0, buf, log_func);
}
/* Row is already logged, don't log it again in ha_write_row() */
row_logging= 0;
sequence->all_values_used= 0;
if (!sequence_locked)
sequence->write_unlock(table);
DBUG_RETURN(error);
}
/*
Inherit the sequence base table flags.
*/
handler::Table_flags ha_sequence::table_flags() const
{
DBUG_ENTER("ha_sequence::table_flags");
DBUG_RETURN((file->table_flags() & ~SEQUENCE_DISABLED_TABLE_FLAGS) |
SEQUENCE_ENABLED_TABLE_FLAGS);
}
int ha_sequence::info(uint flag)
{
DBUG_ENTER("ha_sequence::info");
file->info(flag);
/* Inform optimizer that we have always only one record */
stats= file->stats;
stats.records= 1;
DBUG_RETURN(false);
}
int ha_sequence::extra(enum ha_extra_function operation)
{
if (operation == HA_EXTRA_PREPARE_FOR_ALTER_TABLE)
{
/* In case of ALTER TABLE allow ::write_row() to copy rows */
sequence->initialized= SEQUENCE::SEQ_IN_ALTER;
}
return file->extra(operation);
}
bool ha_sequence::check_if_incompatible_data(HA_CREATE_INFO *create_info,
uint table_changes)
{
/* Table definition is locked for SEQUENCE tables */
return(COMPATIBLE_DATA_YES);
}
enum_alter_inplace_result
ha_sequence::check_if_supported_inplace_alter(TABLE *altered_table,
Alter_inplace_info *ai)
{
return file->check_if_supported_inplace_alter(altered_table, ai);
}
int ha_sequence::external_lock(THD *thd, int lock_type)
{
int error= file->external_lock(thd, lock_type);
/*
Copy lock flag to satisfy DBUG_ASSERT checks in ha_* functions in
handler.cc when we later call it with file->ha_..()
*/
if (!error)
file->m_lock_type= lock_type;
return error;
}
int ha_sequence::discard_or_import_tablespace(my_bool discard)
{
int error= file->discard_or_import_tablespace(discard);
if (!error && !discard)
{
/* Doing import table space. Read the imported values */
if (!(error= table->s->sequence->read_stored_values(table)))
{
table->s->sequence->initialized= SEQUENCE::SEQ_READY_TO_USE;
memcpy(table->record[1], table->s->default_values, table->s->reclength);
}
}
return error;
}
/*
Sequence engine error deal method
*/
void ha_sequence::print_error(int error, myf errflag)
{
const char *sequence_db= table_share->db.str;
const char *sequence_name= table_share->table_name.str;
DBUG_ENTER("ha_sequence::print_error");
switch (error) {
case HA_ERR_SEQUENCE_INVALID_DATA:
{
my_error(ER_SEQUENCE_INVALID_DATA, MYF(errflag), sequence_db,
sequence_name);
DBUG_VOID_RETURN;
}
case HA_ERR_SEQUENCE_RUN_OUT:
{
my_error(ER_SEQUENCE_RUN_OUT, MYF(errflag), sequence_db, sequence_name);
DBUG_VOID_RETURN;
}
case HA_ERR_WRONG_COMMAND:
my_error(ER_ILLEGAL_HA, MYF(0), "SEQUENCE", sequence_db, sequence_name);
DBUG_VOID_RETURN;
case ER_WRONG_INSERT_INTO_SEQUENCE:
my_error(error, MYF(0));
DBUG_VOID_RETURN;
}
file->print_error(error, errflag);
DBUG_VOID_RETURN;
}
int ha_sequence::check(THD* thd, HA_CHECK_OPT* check_opt)
{
DBUG_ENTER("ha_sequence::check");
/* Check the underlying engine */
if (int ret= file->check(thd, check_opt))
DBUG_RETURN(ret);
/* Check number of rows */
if ((file->table_flags() & HA_STATS_RECORDS_IS_EXACT))
{
if (file->stats.records > 1)
push_warning(thd, Sql_condition::WARN_LEVEL_WARN,
ER_SEQUENCE_TABLE_HAS_TOO_MANY_ROWS,
ER_THD(thd, ER_SEQUENCE_TABLE_HAS_TOO_MANY_ROWS));
else if (file->stats.records == 0)
{
my_error(ER_SEQUENCE_TABLE_HAS_TOO_FEW_ROWS, MYF(0));
DBUG_RETURN(HA_ADMIN_CORRUPT);
}
}
/*
Initialise the sequence from the table if needed.
*/
if (sequence->initialized == SEQUENCE::SEQ_UNINTIALIZED)
{
if (sequence->read_stored_values(table))
DBUG_RETURN(HA_ADMIN_FAILED);
else
sequence->initialized= SEQUENCE::SEQ_READY_TO_USE;
}
DBUG_ASSERT(sequence->initialized == SEQUENCE::SEQ_READY_TO_USE);
/* Check and adjust sequence state */
if (sequence->check_and_adjust(thd, false, /*adjust_next=*/false))
{
print_error(HA_ERR_SEQUENCE_INVALID_DATA, MYF(0));
DBUG_RETURN(HA_ADMIN_CORRUPT);
}
/* Check value not exhausted */
if (sequence->has_run_out())
push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN,
ER_SEQUENCE_RUN_OUT, ER_THD(thd, ER_SEQUENCE_RUN_OUT),
table->s->db.str, table->s->table_name.str);
DBUG_RETURN(0);
}
/*****************************************************************************
Sequence plugin interface
*****************************************************************************/
/*
Create an new handler
*/
static handler *sequence_create_handler(handlerton *hton,
TABLE_SHARE *share,
MEM_ROOT *mem_root)
{
DBUG_ENTER("sequence_create_handler");
if (unlikely(!share))
{
/*
This can happen if we call get_new_handler with a non existing share
*/
DBUG_RETURN(0);
}
DBUG_RETURN(new (mem_root) ha_sequence(hton, share));
}
/*
Sequence engine end.
SYNOPSIS
sequence_end()
p handlerton.
type panic type.
RETURN VALUES
0 Success
!=0 Failure
*/
static int sequence_end(handlerton* hton,
ha_panic_function type __attribute__((unused)))
{
DBUG_ENTER("sequence_end");
DBUG_RETURN(0);
}
/*
Sequence engine init.
SYNOPSIS
sequence_initialize()
@param p handlerton.
retval 0 Success
retval !=0 Failure
*/
static int sequence_initialize(void *p)
{
handlerton *local_sequence_hton= (handlerton *)p;
DBUG_ENTER("sequence_initialize");
local_sequence_hton->db_type= DB_TYPE_SEQUENCE;
local_sequence_hton->create= sequence_create_handler;
local_sequence_hton->panic= sequence_end;
local_sequence_hton->flags= (HTON_NOT_USER_SELECTABLE |
HTON_HIDDEN |
HTON_TEMPORARY_NOT_SUPPORTED |
HTON_ALTER_NOT_SUPPORTED |
#ifdef WITH_WSREP
HTON_WSREP_REPLICATION |
#endif
HTON_NO_PARTITION);
DBUG_RETURN(0);
}
static struct st_mysql_storage_engine sequence_storage_engine=
{ MYSQL_HANDLERTON_INTERFACE_VERSION };
maria_declare_plugin(sql_sequence)
{
MYSQL_STORAGE_ENGINE_PLUGIN,
&sequence_storage_engine,
"SQL_SEQUENCE",
"jianwei.zhao @ Aliyun & Monty @ MariaDB corp",
"Sequence Storage Engine for CREATE SEQUENCE",
PLUGIN_LICENSE_GPL,
sequence_initialize, /* Plugin Init */
NULL, /* Plugin Deinit */
0x0100, /* 1.0 */
NULL, /* status variables */
NULL, /* system variables */
"1.0", /* string version */
MariaDB_PLUGIN_MATURITY_STABLE /* maturity */
}
maria_declare_plugin_end;
View Git Blame Copy Permalink