mirror/mariadb
1
0
Fork 0
mirror of https://github.com/MariaDB/server.git synced 2025-01-29 02:05:57 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

IB, SQL: removed VTQ, added TRT on SQL layer [closes #305]

Browse source
This commit is contained in:
Aleksey Midenkov 2017-11-10 17:54:46 +03:00
parent fa79f6ac86
commit 33085349e9
56 changed files with 656 additions and 1605 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
mysql-test/suite/versioning/common.inc
View file

@ -1,5 +1,5 @@
set @@session.time_zone='+00:00';
select ifnull(max(transaction_id), 0) into @start_trx_id from information_schema.innodb_vtq;
select ifnull(max(transaction_id), 0) into @start_trx_id from mysql.transaction_registry;
set @test_start=now(6);
delimiter ~~;
@ -12,11 +12,11 @@ begin
commit_id > transaction_id as B,
begin_timestamp > @test_start as C,
commit_timestamp >= begin_timestamp as D
from information_schema.innodb_vtq
from mysql.transaction_registry
where transaction_id > @start_trx_id;
select ifnull(max(transaction_id), 0)
into @start_trx_id
from information_schema.innodb_vtq;
from mysql.transaction_registry;
end~~
create function if not exists default_engine()

1
mysql-test/suite/versioning/common.opt
View file

@ -14,6 +14,5 @@
--innodb-sys-indexes
--innodb-sys-tables
--innodb-sys-virtual
--innodb-vtq
--versioning-hide=implicit
--plugin-load=versioning

81
mysql-test/suite/versioning/r/alter.result
View file

@ -248,7 +248,7 @@ t CREATE TABLE `t` (
PERIOD FOR SYSTEM_TIME (`sys_trx_start`, `sys_trx_end`)
) ENGINE=MyISAM DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
set @@session.time_zone='+00:00';
select ifnull(max(transaction_id), 0) into @start_trx_id from information_schema.innodb_vtq;
select ifnull(max(transaction_id), 0) into @start_trx_id from mysql.transaction_registry;
set @test_start=now(6);
create procedure if not exists verify_vtq()
begin
@ -259,11 +259,11 @@ transaction_id > 0 as A,
commit_id > transaction_id as B,
begin_timestamp > @test_start as C,
commit_timestamp >= begin_timestamp as D
from information_schema.innodb_vtq
from mysql.transaction_registry
where transaction_id > @start_trx_id;
select ifnull(max(transaction_id), 0)
into @start_trx_id
from information_schema.innodb_vtq;
from mysql.transaction_registry;
end~~
create function if not exists default_engine()
returns varchar(255)
@ -335,11 +335,16 @@ add column trx_end timestamp(6) generated always as row end,
add period for system_time(trx_start, trx_end),
with system versioning;
ERROR HY000: `trx_start` must be of type `BIGINT(20) UNSIGNED` for versioned table `t`
call verify_vtq;
No A B C D
alter table t
add column trx_start bigint(20) unsigned generated always as row start,
add column trx_end bigint(20) unsigned generated always as row end,
add period for system_time(trx_start, trx_end),
with system versioning;
call verify_vtq;
No A B C D
1 1 1 1 1
show create table t;
Table Create Table
t CREATE TABLE `t` (
@ -349,8 +354,15 @@ t CREATE TABLE `t` (
PERIOD FOR SYSTEM_TIME (`trx_start`, `trx_end`)
) ENGINE=InnoDB DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
alter table t drop column trx_start, drop column trx_end;
alter table t without system versioning;
call verify_vtq;
No A B C D
alter table t without system versioning, algorithm=copy;
call verify_vtq;
No A B C D
alter table t with system versioning, algorithm=copy;
call verify_vtq;
No A B C D
1 1 1 1 1
show create table t;
Table Create Table
t CREATE TABLE `t` (
@ -359,11 +371,14 @@ t CREATE TABLE `t` (
`sys_trx_end` bigint(20) unsigned GENERATED ALWAYS AS ROW END,
PERIOD FOR SYSTEM_TIME (`sys_trx_start`, `sys_trx_end`)
) ENGINE=InnoDB DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
update t set a=2;
update t set a= 2;
select * from t for system_time all;
a
2
1
call verify_vtq;
No A B C D
1 1 1 1 1
alter table t add column b int, algorithm=copy;
show create table t;
Table Create Table
@ -377,6 +392,8 @@ t CREATE TABLE `t` (
select * from t;
a b
2 NULL
call verify_vtq;
No A B C D
alter table t drop column b, algorithm=copy;
show create table t;
Table Create Table
@ -390,6 +407,31 @@ select * from t for system_time all;
a
2
1
call verify_vtq;
No A B C D
alter table t without system versioning, algorithm=inplace;
call verify_vtq;
No A B C D
alter table t with system versioning, algorithm=inplace;
call verify_vtq;
No A B C D
1 1 1 1 1
show create table t;
Table Create Table
t CREATE TABLE `t` (
`a` int(11) DEFAULT NULL,
`sys_trx_start` bigint(20) unsigned GENERATED ALWAYS AS ROW START,
`sys_trx_end` bigint(20) unsigned GENERATED ALWAYS AS ROW END,
PERIOD FOR SYSTEM_TIME (`sys_trx_start`, `sys_trx_end`)
) ENGINE=InnoDB DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
update t set a= 1;
select * from t for system_time all;
a
1
2
call verify_vtq;
No A B C D
1 1 1 1 1
alter table t add column b int, algorithm=inplace;
show create table t;
Table Create Table
@ -402,7 +444,9 @@ t CREATE TABLE `t` (
) ENGINE=InnoDB DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
select * from t;
a b
2 NULL
1 NULL
call verify_vtq;
No A B C D
alter table t drop column b, algorithm=inplace;
show create table t;
Table Create Table
@ -414,8 +458,8 @@ t CREATE TABLE `t` (
) ENGINE=InnoDB DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
select * from t for system_time all;
a
2
1
2
alter table t without system versioning, algorithm=copy;
show create table t;
Table Create Table
@ -486,26 +530,6 @@ a b
2 -2
3 1
4 2
create or replace table t (a int) engine innodb;
insert into t values (1);
alter table t with system versioning, algorithm=inplace;
select * from t for system_time all;
a
1
update t set a=2;
select * from t for system_time all;
a
2
1
alter table t add column b int, algorithm=inplace;
select * from t for system_time all;
a b
2 NULL
1 NULL
alter table t without system versioning, algorithm=inplace;
select * from t;
a b
2 NULL
create or replace table t (
a int,
sys_trx_start bigint(20) unsigned generated always as row start,
@ -582,9 +606,6 @@ No A B C D
9 1 1 1 1
10 1 1 1 1
11 1 1 1 1
12 1 1 1 1
13 1 1 1 1
14 1 1 1 1
drop table t;
drop procedure verify_vtq;
drop procedure innodb_verify_vtq;

6
mysql-test/suite/versioning/r/auto_increment.result
View file

@ -1,5 +1,5 @@
set @@session.time_zone='+00:00';
select ifnull(max(transaction_id), 0) into @start_trx_id from information_schema.innodb_vtq;
select ifnull(max(transaction_id), 0) into @start_trx_id from mysql.transaction_registry;
set @test_start=now(6);
create procedure if not exists verify_vtq()
begin
@ -10,11 +10,11 @@ transaction_id > 0 as A,
commit_id > transaction_id as B,
begin_timestamp > @test_start as C,
commit_timestamp >= begin_timestamp as D
from information_schema.innodb_vtq
from mysql.transaction_registry
where transaction_id > @start_trx_id;
select ifnull(max(transaction_id), 0)
into @start_trx_id
from information_schema.innodb_vtq;
from mysql.transaction_registry;
end~~
create function if not exists default_engine()
returns varchar(255)

23
mysql-test/suite/versioning/r/commit_id.result
View file

@ -1,5 +1,5 @@
set @@session.time_zone='+00:00';
select ifnull(max(transaction_id), 0) into @start_trx_id from information_schema.innodb_vtq;
select ifnull(max(transaction_id), 0) into @start_trx_id from mysql.transaction_registry;
set @test_start=now(6);
create procedure if not exists verify_vtq()
begin
@ -10,11 +10,11 @@ transaction_id > 0 as A,
commit_id > transaction_id as B,
begin_timestamp > @test_start as C,
commit_timestamp >= begin_timestamp as D
from information_schema.innodb_vtq
from mysql.transaction_registry
where transaction_id > @start_trx_id;
select ifnull(max(transaction_id), 0)
into @start_trx_id
from information_schema.innodb_vtq;
from mysql.transaction_registry;
end~~
create function if not exists default_engine()
returns varchar(255)
@ -81,19 +81,22 @@ insert into t1 values ();
set @ts0= now(6);
insert into t1 values ();
select sys_trx_start from t1 where id = last_insert_id() into @tx0;
select transaction_id = @tx0 from information_schema.innodb_vtq limit 1;
select transaction_id = @tx0 from mysql.transaction_registry
order by transaction_id desc limit 1;
transaction_id = @tx0
1
set @ts1= now(6);
insert into t1 values ();
select sys_trx_start from t1 where id = last_insert_id() into @tx1;
select transaction_id = @tx1 from information_schema.innodb_vtq limit 1;
select transaction_id = @tx1 from mysql.transaction_registry
order by transaction_id desc limit 1;
transaction_id = @tx1
1
set @ts2= now(6);
insert into t1 values ();
select sys_trx_start from t1 where id = last_insert_id() into @tx2;
select transaction_id = @tx2 from information_schema.innodb_vtq limit 1;
select transaction_id = @tx2 from mysql.transaction_registry
order by transaction_id desc limit 1;
transaction_id = @tx2
1
set @ts3= now(6);
@ -132,25 +135,25 @@ A B C D E F H
set transaction isolation level read uncommitted;
insert into t1 values ();
select sys_trx_start from t1 where id = last_insert_id() into @tx3;
select isolation_level = 'READ-UNCOMMITTED' from information_schema.innodb_vtq where transaction_id = @tx3;
select isolation_level = 'READ-UNCOMMITTED' from mysql.transaction_registry where transaction_id = @tx3;
isolation_level = 'READ-UNCOMMITTED'
1
set transaction isolation level read committed;
insert into t1 values ();
select sys_trx_start from t1 where id = last_insert_id() into @tx4;
select isolation_level = 'READ-COMMITTED' from information_schema.innodb_vtq where transaction_id = @tx4;
select isolation_level = 'READ-COMMITTED' from mysql.transaction_registry where transaction_id = @tx4;
isolation_level = 'READ-COMMITTED'
1
set transaction isolation level serializable;
insert into t1 values ();
select sys_trx_start from t1 where id = last_insert_id() into @tx5;
select isolation_level = 'SERIALIZABLE' from information_schema.innodb_vtq where transaction_id = @tx5;
select isolation_level = 'SERIALIZABLE' from mysql.transaction_registry where transaction_id = @tx5;
isolation_level = 'SERIALIZABLE'
1
set transaction isolation level repeatable read;
insert into t1 values ();
select sys_trx_start from t1 where id = last_insert_id() into @tx6;
select isolation_level = 'REPEATABLE-READ' from information_schema.innodb_vtq where transaction_id = @tx6;
select isolation_level = 'REPEATABLE-READ' from mysql.transaction_registry where transaction_id = @tx6;
isolation_level = 'REPEATABLE-READ'
1
drop table t1;

6
mysql-test/suite/versioning/r/create.result
View file

@ -1,5 +1,5 @@
set @@session.time_zone='+00:00';
select ifnull(max(transaction_id), 0) into @start_trx_id from information_schema.innodb_vtq;
select ifnull(max(transaction_id), 0) into @start_trx_id from mysql.transaction_registry;
set @test_start=now(6);
create procedure if not exists verify_vtq()
begin
@ -10,11 +10,11 @@ transaction_id > 0 as A,
commit_id > transaction_id as B,
begin_timestamp > @test_start as C,
commit_timestamp >= begin_timestamp as D
from information_schema.innodb_vtq
from mysql.transaction_registry
where transaction_id > @start_trx_id;
select ifnull(max(transaction_id), 0)
into @start_trx_id
from information_schema.innodb_vtq;
from mysql.transaction_registry;
end~~
create function if not exists default_engine()
returns varchar(255)

6
mysql-test/suite/versioning/r/ddl.result
View file

@ -1,5 +1,5 @@
set @@session.time_zone='+00:00';
select ifnull(max(transaction_id), 0) into @start_trx_id from information_schema.innodb_vtq;
select ifnull(max(transaction_id), 0) into @start_trx_id from mysql.transaction_registry;
set @test_start=now(6);
create procedure if not exists verify_vtq()
begin
@ -10,11 +10,11 @@ transaction_id > 0 as A,
commit_id > transaction_id as B,
begin_timestamp > @test_start as C,
commit_timestamp >= begin_timestamp as D
from information_schema.innodb_vtq
from mysql.transaction_registry
where transaction_id > @start_trx_id;
select ifnull(max(transaction_id), 0)
into @start_trx_id
from information_schema.innodb_vtq;
from mysql.transaction_registry;
end~~
create function if not exists default_engine()
returns varchar(255)

6
mysql-test/suite/versioning/r/delete.result
View file

@ -1,5 +1,5 @@
set @@session.time_zone='+00:00';
select ifnull(max(transaction_id), 0) into @start_trx_id from information_schema.innodb_vtq;
select ifnull(max(transaction_id), 0) into @start_trx_id from mysql.transaction_registry;
set @test_start=now(6);
create procedure if not exists verify_vtq()
begin
@ -10,11 +10,11 @@ transaction_id > 0 as A,
commit_id > transaction_id as B,
begin_timestamp > @test_start as C,
commit_timestamp >= begin_timestamp as D
from information_schema.innodb_vtq
from mysql.transaction_registry
where transaction_id > @start_trx_id;
select ifnull(max(transaction_id), 0)
into @start_trx_id
from information_schema.innodb_vtq;
from mysql.transaction_registry;
end~~
create function if not exists default_engine()
returns varchar(255)

6
mysql-test/suite/versioning/r/insert.result
View file

@ -1,5 +1,5 @@
set @@session.time_zone='+00:00';
select ifnull(max(transaction_id), 0) into @start_trx_id from information_schema.innodb_vtq;
select ifnull(max(transaction_id), 0) into @start_trx_id from mysql.transaction_registry;
set @test_start=now(6);
create procedure if not exists verify_vtq()
begin
@ -10,11 +10,11 @@ transaction_id > 0 as A,
commit_id > transaction_id as B,
begin_timestamp > @test_start as C,
commit_timestamp >= begin_timestamp as D
from information_schema.innodb_vtq
from mysql.transaction_registry
where transaction_id > @start_trx_id;
select ifnull(max(transaction_id), 0)
into @start_trx_id
from information_schema.innodb_vtq;
from mysql.transaction_registry;
end~~
create function if not exists default_engine()
returns varchar(255)

6
mysql-test/suite/versioning/r/partition.result
View file

@ -1,5 +1,5 @@
set @@session.time_zone='+00:00';
select ifnull(max(transaction_id), 0) into @start_trx_id from information_schema.innodb_vtq;
select ifnull(max(transaction_id), 0) into @start_trx_id from mysql.transaction_registry;
set @test_start=now(6);
create procedure if not exists verify_vtq()
begin
@ -10,11 +10,11 @@ transaction_id > 0 as A,
commit_id > transaction_id as B,
begin_timestamp > @test_start as C,
commit_timestamp >= begin_timestamp as D
from information_schema.innodb_vtq
from mysql.transaction_registry
where transaction_id > @start_trx_id;
select ifnull(max(transaction_id), 0)
into @start_trx_id
from information_schema.innodb_vtq;
from mysql.transaction_registry;
end~~
create function if not exists default_engine()
returns varchar(255)

6
mysql-test/suite/versioning/r/select.result
View file

@ -1,5 +1,5 @@
set @@session.time_zone='+00:00';
select ifnull(max(transaction_id), 0) into @start_trx_id from information_schema.innodb_vtq;
select ifnull(max(transaction_id), 0) into @start_trx_id from mysql.transaction_registry;
set @test_start=now(6);
create procedure if not exists verify_vtq()
begin
@ -10,11 +10,11 @@ transaction_id > 0 as A,
commit_id > transaction_id as B,
begin_timestamp > @test_start as C,
commit_timestamp >= begin_timestamp as D
from information_schema.innodb_vtq
from mysql.transaction_registry
where transaction_id > @start_trx_id;
select ifnull(max(transaction_id), 0)
into @start_trx_id
from information_schema.innodb_vtq;
from mysql.transaction_registry;
end~~
create function if not exists default_engine()
returns varchar(255)

6
mysql-test/suite/versioning/r/select_sp.result
View file

@ -1,5 +1,5 @@
set @@session.time_zone='+00:00';
select ifnull(max(transaction_id), 0) into @start_trx_id from information_schema.innodb_vtq;
select ifnull(max(transaction_id), 0) into @start_trx_id from mysql.transaction_registry;
set @test_start=now(6);
create procedure if not exists verify_vtq()
begin
@ -10,11 +10,11 @@ transaction_id > 0 as A,
commit_id > transaction_id as B,
begin_timestamp > @test_start as C,
commit_timestamp >= begin_timestamp as D
from information_schema.innodb_vtq
from mysql.transaction_registry
where transaction_id > @start_trx_id;
select ifnull(max(transaction_id), 0)
into @start_trx_id
from information_schema.innodb_vtq;
from mysql.transaction_registry;
end~~
create function if not exists default_engine()
returns varchar(255)

6
mysql-test/suite/versioning/r/update.result
View file

@ -1,5 +1,5 @@
set @@session.time_zone='+00:00';
select ifnull(max(transaction_id), 0) into @start_trx_id from information_schema.innodb_vtq;
select ifnull(max(transaction_id), 0) into @start_trx_id from mysql.transaction_registry;
set @test_start=now(6);
create procedure if not exists verify_vtq()
begin
@ -10,11 +10,11 @@ transaction_id > 0 as A,
commit_id > transaction_id as B,
begin_timestamp > @test_start as C,
commit_timestamp >= begin_timestamp as D
from information_schema.innodb_vtq
from mysql.transaction_registry
where transaction_id > @start_trx_id;
select ifnull(max(transaction_id), 0)
into @start_trx_id
from information_schema.innodb_vtq;
from mysql.transaction_registry;
end~~
create function if not exists default_engine()
returns varchar(255)

6
mysql-test/suite/versioning/r/view.result
View file

@ -1,5 +1,5 @@
set @@session.time_zone='+00:00';
select ifnull(max(transaction_id), 0) into @start_trx_id from information_schema.innodb_vtq;
select ifnull(max(transaction_id), 0) into @start_trx_id from mysql.transaction_registry;
set @test_start=now(6);
create procedure if not exists verify_vtq()
begin
@ -10,11 +10,11 @@ transaction_id > 0 as A,
commit_id > transaction_id as B,
begin_timestamp > @test_start as C,
commit_timestamp >= begin_timestamp as D
from information_schema.innodb_vtq
from mysql.transaction_registry
where transaction_id > @start_trx_id;
select ifnull(max(transaction_id), 0)
into @start_trx_id
from information_schema.innodb_vtq;
from mysql.transaction_registry;
end~~
create function if not exists default_engine()
returns varchar(255)

35
mysql-test/suite/versioning/t/alter.test
View file

@ -141,32 +141,52 @@ alter table t
add period for system_time(trx_start, trx_end),
with system versioning;
call verify_vtq;
alter table t
add column trx_start bigint(20) unsigned generated always as row start,
add column trx_end bigint(20) unsigned generated always as row end,
add period for system_time(trx_start, trx_end),
with system versioning;
call verify_vtq;
show create table t;
alter table t drop column trx_start, drop column trx_end;
alter table t without system versioning;
call verify_vtq;
alter table t without system versioning, algorithm=copy;
call verify_vtq;
alter table t with system versioning, algorithm=copy;
call verify_vtq;
show create table t;
update t set a=2;
update t set a= 2;
select * from t for system_time all;
call verify_vtq;
alter table t add column b int, algorithm=copy;
show create table t;
select * from t;
call verify_vtq;
alter table t drop column b, algorithm=copy;
show create table t;
select * from t for system_time all;
call verify_vtq;
alter table t without system versioning, algorithm=inplace;
call verify_vtq;
alter table t with system versioning, algorithm=inplace;
call verify_vtq;
show create table t;
update t set a= 1;
select * from t for system_time all;
call verify_vtq;
alter table t add column b int, algorithm=inplace;
show create table t;
select * from t;
call verify_vtq;
alter table t drop column b, algorithm=inplace;
show create table t;
@ -207,17 +227,6 @@ select * from t for system_time all;
insert into t values (4, NULL);
select * from t for system_time all;
create or replace table t (a int) engine innodb;
insert into t values (1);
alter table t with system versioning, algorithm=inplace;
select * from t for system_time all;
update t set a=2;
select * from t for system_time all;
alter table t add column b int, algorithm=inplace;
select * from t for system_time all;
alter table t without system versioning, algorithm=inplace;
select * from t;
create or replace table t (
a int,
sys_trx_start bigint(20) unsigned generated always as row start,

17
mysql-test/suite/versioning/t/commit_id.test
View file

@ -13,17 +13,20 @@ insert into t1 values ();
set @ts0= now(6);
insert into t1 values ();
select sys_trx_start from t1 where id = last_insert_id() into @tx0;
select transaction_id = @tx0 from information_schema.innodb_vtq limit 1;
select transaction_id = @tx0 from mysql.transaction_registry
order by transaction_id desc limit 1;
set @ts1= now(6);
insert into t1 values ();
select sys_trx_start from t1 where id = last_insert_id() into @tx1;
select transaction_id = @tx1 from information_schema.innodb_vtq limit 1;
select transaction_id = @tx1 from mysql.transaction_registry
order by transaction_id desc limit 1;
set @ts2= now(6);
insert into t1 values ();
select sys_trx_start from t1 where id = last_insert_id() into @tx2;
select transaction_id = @tx2 from information_schema.innodb_vtq limit 1;
select transaction_id = @tx2 from mysql.transaction_registry
order by transaction_id desc limit 1;
set @ts3= now(6);
@ -62,22 +65,22 @@ select
set transaction isolation level read uncommitted;
insert into t1 values ();
select sys_trx_start from t1 where id = last_insert_id() into @tx3;
select isolation_level = 'READ-UNCOMMITTED' from information_schema.innodb_vtq where transaction_id = @tx3;
select isolation_level = 'READ-UNCOMMITTED' from mysql.transaction_registry where transaction_id = @tx3;
set transaction isolation level read committed;
insert into t1 values ();
select sys_trx_start from t1 where id = last_insert_id() into @tx4;
select isolation_level = 'READ-COMMITTED' from information_schema.innodb_vtq where transaction_id = @tx4;
select isolation_level = 'READ-COMMITTED' from mysql.transaction_registry where transaction_id = @tx4;
set transaction isolation level serializable;
insert into t1 values ();
select sys_trx_start from t1 where id = last_insert_id() into @tx5;
select isolation_level = 'SERIALIZABLE' from information_schema.innodb_vtq where transaction_id = @tx5;
select isolation_level = 'SERIALIZABLE' from mysql.transaction_registry where transaction_id = @tx5;
set transaction isolation level repeatable read;
insert into t1 values ();
select sys_trx_start from t1 where id = last_insert_id() into @tx6;
select isolation_level = 'REPEATABLE-READ' from information_schema.innodb_vtq where transaction_id = @tx6;
select isolation_level = 'REPEATABLE-READ' from mysql.transaction_registry where transaction_id = @tx6;
drop table t1;

60
plugin/versioning/versioning.cc
View file

@ -22,14 +22,10 @@
#include "sql_plugin.h" // st_plugin_int
#include "sql_class.h"
#include "item.h"
#include "vtq.h"
#include "vers_utils.h"
plugin_ref innodb_plugin= NULL;
static handlerton* innodb_hton= NULL;
/* System Versioning: VTQ_TRX_ID(), VTQ_COMMIT_ID(), VTQ_BEGIN_TS(), VTQ_COMMIT_TS(), VTQ_ISO_LEVEL() */
template <vtq_field_t VTQ_FIELD>
template <TR_table::field_id_t VTQ_FIELD>
class Create_func_vtq : public Create_native_func
{
public:
@ -42,10 +38,10 @@ protected:
virtual ~Create_func_vtq<VTQ_FIELD>() {}
};
template<vtq_field_t VTQ_FIELD>
template<TR_table::field_id_t VTQ_FIELD>
Create_func_vtq<VTQ_FIELD> Create_func_vtq<VTQ_FIELD>::s_singleton;
template <vtq_field_t VTQ_FIELD>
template <TR_table::field_id_t VTQ_FIELD>
Item*
Create_func_vtq<VTQ_FIELD>::create_native(THD *thd, LEX_CSTRING *name,
List<Item> *item_list)
@ -62,14 +58,14 @@ Create_func_vtq<VTQ_FIELD>::create_native(THD *thd, LEX_CSTRING *name,
Item *param_1= item_list->pop();
switch (VTQ_FIELD)
{
case VTQ_BEGIN_TS:
case VTQ_COMMIT_TS:
func= new (thd->mem_root) Item_func_vtq_ts(thd, innodb_hton, param_1, VTQ_FIELD);
case TR_table::FLD_BEGIN_TS:
case TR_table::FLD_COMMIT_TS:
func= new (thd->mem_root) Item_func_vtq_ts(thd, param_1, VTQ_FIELD);
break;
case VTQ_TRX_ID:
case VTQ_COMMIT_ID:
case VTQ_ISO_LEVEL:
func= new (thd->mem_root) Item_func_vtq_id(thd, innodb_hton, param_1, VTQ_FIELD);
case TR_table::FLD_TRX_ID:
case TR_table::FLD_COMMIT_ID:
case TR_table::FLD_ISO_LEVEL:
func= new (thd->mem_root) Item_func_vtq_id(thd, param_1, VTQ_FIELD);
break;
default:
DBUG_ASSERT(0);
@ -82,9 +78,9 @@ Create_func_vtq<VTQ_FIELD>::create_native(THD *thd, LEX_CSTRING *name,
Item *param_2= item_list->pop();
switch (VTQ_FIELD)
{
case VTQ_TRX_ID:
case VTQ_COMMIT_ID:
func= new (thd->mem_root) Item_func_vtq_id(thd, innodb_hton, param_1, param_2, VTQ_FIELD);
case TR_table::FLD_TRX_ID:
case TR_table::FLD_COMMIT_ID:
func= new (thd->mem_root) Item_func_vtq_id(thd, param_1, param_2, VTQ_FIELD);
break;
default:
goto error;
@ -119,7 +115,7 @@ public:
{
Item *param_1= item_list->pop();
Item *param_2= item_list->pop();
func= new (thd->mem_root) Item_func_vtq_trx_seesX(thd, innodb_hton, param_1, param_2);
func= new (thd->mem_root) Item_func_vtq_trx_seesX(thd, param_1, param_2);
break;
}
default:
@ -144,11 +140,11 @@ Create_func_vtq_trx_sees<X> Create_func_vtq_trx_sees<X>::s_singleton;
static Native_func_registry func_array[] =
{
{ { C_STRING_WITH_LEN("VTQ_BEGIN_TS") }, BUILDER(Create_func_vtq<VTQ_BEGIN_TS>)},
{ { C_STRING_WITH_LEN("VTQ_COMMIT_ID") }, BUILDER(Create_func_vtq<VTQ_COMMIT_ID>)},
{ { C_STRING_WITH_LEN("VTQ_COMMIT_TS") }, BUILDER(Create_func_vtq<VTQ_COMMIT_TS>)},
{ { C_STRING_WITH_LEN("VTQ_ISO_LEVEL") }, BUILDER(Create_func_vtq<VTQ_ISO_LEVEL>)},
{ { C_STRING_WITH_LEN("VTQ_TRX_ID") }, BUILDER(Create_func_vtq<VTQ_TRX_ID>)},
{ { C_STRING_WITH_LEN("VTQ_BEGIN_TS") }, BUILDER(Create_func_vtq<TR_table::FLD_BEGIN_TS>)},
{ { C_STRING_WITH_LEN("VTQ_COMMIT_ID") }, BUILDER(Create_func_vtq<TR_table::FLD_COMMIT_ID>)},
{ { C_STRING_WITH_LEN("VTQ_COMMIT_TS") }, BUILDER(Create_func_vtq<TR_table::FLD_COMMIT_TS>)},
{ { C_STRING_WITH_LEN("VTQ_ISO_LEVEL") }, BUILDER(Create_func_vtq<TR_table::FLD_ISO_LEVEL>)},
{ { C_STRING_WITH_LEN("VTQ_TRX_ID") }, BUILDER(Create_func_vtq<TR_table::FLD_TRX_ID>)},
{ { C_STRING_WITH_LEN("VTQ_TRX_SEES") }, BUILDER(Create_func_vtq_trx_sees<Item_func_vtq_trx_sees>)},
{ { C_STRING_WITH_LEN("VTQ_TRX_SEES_EQ") }, BUILDER(Create_func_vtq_trx_sees<Item_func_vtq_trx_sees_eq>)},
{ {0, 0}, NULL}
@ -164,8 +160,6 @@ static Native_func_registry func_array[] =
static int versioning_plugin_init(void *p __attribute__ ((unused)))
{
static LString InnoDB= "InnoDB";
DBUG_ENTER("versioning_plugin_init");
// No need in locking since we so far single-threaded
int res= item_create_append(func_array);
@ -175,28 +169,12 @@ static int versioning_plugin_init(void *p __attribute__ ((unused)))
DBUG_RETURN(res);
}
innodb_plugin= ha_resolve_by_name(NULL, &InnoDB.lex_cstring(), false);
if (!innodb_plugin)
{
my_error(ER_PLUGIN_IS_NOT_LOADED, MYF(0), InnoDB.ptr());
DBUG_RETURN(1);
}
innodb_hton= plugin_hton(innodb_plugin);
if (!innodb_hton || (innodb_hton->flags & HTON_NOT_USER_SELECTABLE))
{
my_message(ER_PLUGIN_IS_NOT_LOADED, "Can't get handlerton" , MYF(0));
DBUG_RETURN(1);
}
DBUG_RETURN(0);
}
static int versioning_plugin_deinit(void *p __attribute__ ((unused)))
{
DBUG_ENTER("versioning_plugin_deinit");
if (innodb_plugin)
plugin_unlock(NULL, innodb_plugin);
DBUG_RETURN(0);
}

18
scripts/mysql_system_tables.sql
View file

@ -131,6 +131,19 @@ SET @create_innodb_index_stats="CREATE TABLE IF NOT EXISTS innodb_index_stats (
PRIMARY KEY (database_name, table_name, index_name, stat_name)
) ENGINE=INNODB DEFAULT CHARSET=utf8 COLLATE=utf8_bin STATS_PERSISTENT=0";
SET @create_transaction_registry="CREATE TABLE IF NOT EXISTS transaction_registry (
transaction_id BIGINT UNSIGNED NOT NULL,
commit_id BIGINT UNSIGNED NOT NULL,
begin_timestamp TIMESTAMP(6) NOT NULL DEFAULT '0000-00-00 00:00:00.000000',
commit_timestamp TIMESTAMP(6) NOT NULL DEFAULT '0000-00-00 00:00:00.000000',
isolation_level ENUM('READ-UNCOMMITTED', 'READ-COMMITTED',
'REPEATABLE-READ', 'SERIALIZABLE') NOT NULL,
PRIMARY KEY (transaction_id),
UNIQUE KEY (commit_id),
INDEX (begin_timestamp),
INDEX (commit_timestamp, transaction_id)
) ENGINE=INNODB DEFAULT CHARSET=utf8 COLLATE=utf8_bin STATS_PERSISTENT=0";
SET @create_vtmd_template="CREATE OR REPLACE TABLE vtmd_template (
start BIGINT UNSIGNED GENERATED ALWAYS AS ROW START COMMENT 'TRX_ID of table lifetime start',
end BIGINT UNSIGNED GENERATED ALWAYS AS ROW END COMMENT 'TRX_ID of table lifetime end',
@ -152,6 +165,11 @@ PREPARE stmt FROM @str;
EXECUTE stmt;
DROP PREPARE stmt;
SET @str=IF(@have_innodb <> 0, @create_transaction_registry, "SET @dummy = 0");
PREPARE stmt FROM @str;
EXECUTE stmt;
DROP PREPARE stmt;
SET @str=IF(@have_innodb <> 0, @create_vtmd_template, "SET @dummy = 0");
PREPARE stmt FROM @str;
EXECUTE stmt;

17
sql/field.cc
View file

@ -1997,9 +1997,12 @@ bool Field_vers_trx_id::get_date(MYSQL_TIME *ltime, ulonglong fuzzydate, ulonglo
(table->versioned() && table->s->table_category == TABLE_CATEGORY_TEMPORARY));
if (!trx_id)
return true;
THD *thd= get_thd();
DBUG_ASSERT(thd);
if (trx_id == ULONGLONG_MAX)
{
get_thd()->variables.time_zone->gmt_sec_to_TIME(ltime, TIMESTAMP_MAX_VALUE);
thd->variables.time_zone->gmt_sec_to_TIME(ltime, TIMESTAMP_MAX_VALUE);
ltime->second_part= TIME_MAX_SECOND_PART;
return false;
}
@ -2008,16 +2011,20 @@ bool Field_vers_trx_id::get_date(MYSQL_TIME *ltime, ulonglong fuzzydate, ulonglo
*ltime= cache;
return false;
}
handlerton *hton= table->file->partition_ht();
DBUG_ASSERT(hton);
DBUG_ASSERT(hton->vers_query_trx_id);
bool found= hton->vers_query_trx_id(get_thd(), &cache, trx_id, VTQ_COMMIT_TS);
TR_table trt(thd);
bool found= trt.query(trx_id);
if (found)
{
trt[TR_table::FLD_COMMIT_TS]->get_date(&cache, fuzzydate);
*ltime= cache;
cached= trx_id;
return false;
}
push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN,
ER_VERS_NO_TRX_ID, ER_THD(thd, ER_VERS_NO_TRX_ID),
trx_id);
return true;
}

31
sql/handler.cc
View file

@ -1414,6 +1414,27 @@ int ha_commit_trans(THD *thd, bool all)
goto err;
}
if (rw_trans || thd->lex->sql_command == SQLCOM_ALTER_TABLE)
{
for (Ha_trx_info *hi= ha_info; hi; hi= hi->next())
{
handlerton *ht= hi->ht();
if ((ht->flags & HTON_NATIVE_SYS_VERSIONING) &&
thd->lex->sql_command == SQLCOM_ALTER_TABLE ?
hi->is_trx_tmp_read_write() :
hi->is_trx_read_write())
{
TR_table trt(thd, true);
bool updated;
if (trt.update(updated))
goto err;
if (updated && all)
trans_commit_stmt(thd);
break;
}
}
}
if (trans->no_2pc || (rw_ha_count <= 1))
{
error= ha_commit_one_phase(thd, all);
@ -4054,6 +4075,8 @@ void handler::mark_trx_read_write_internal()
*/
if (table_share == NULL || table_share->tmp_table == NO_TMP_TABLE)
ha_info->set_trx_read_write();
else
ha_info->set_trx_tmp_read_write();
}
}
@ -4278,6 +4301,14 @@ bool handler::ha_commit_inplace_alter_table(TABLE *altered_table,
MDL_EXCLUSIVE) ||
!commit);
if (commit)
{
TR_table trt(ha_thd(), true);
bool updated;
if (trt.update(updated))
return true;
}
return commit_inplace_alter_table(altered_table, ha_alter_info, commit);
}

49
sql/handler.h
View file

@ -34,7 +34,6 @@
#include "structs.h" /* SHOW_COMP_OPTION */
#include "sql_array.h" /* Dynamic_array<> */
#include "mdl.h"
#include "vtq.h"
#include "vers_string.h"
#include "sql_analyze_stmt.h" // for Exec_time_tracker
@ -981,6 +980,7 @@ struct handler_iterator {
class handler;
class group_by_handler;
struct Query;
class TR_table;
typedef class st_select_lex SELECT_LEX;
typedef struct st_order ORDER;
@ -1390,36 +1390,11 @@ struct handlerton
/*
System Versioning
*/
/**
Query VTQ by TRX_ID.
@param[in] thd MySQL thread
@param[out] out field value or whole record returned by query (selected by `field`)
@param[in] in_trx_id query parameter TRX_ID
@param[in] field field to get in `out` or VTQ_ALL for whole record (vtq_record_t)
@return TRUE if record is found, FALSE otherwise */
bool (*vers_query_trx_id)(THD* thd, void *out, ulonglong trx_id, vtq_field_t field);
/** Query VTQ by COMMIT_TS.
@param[in] thd MySQL thread
@param[out] out field value or whole record returned by query (selected by `field`)
@param[in] commit_ts query parameter COMMIT_TS
@param[in] field field to get in `out` or VTQ_ALL for whole record (vtq_record_t)
@param[in] backwards direction of VTQ search
@return TRUE if record is found, FALSE otherwise */
bool (*vers_query_commit_ts)(THD* thd, void *out, const MYSQL_TIME &commit_ts,
vtq_field_t field, bool backwards);
/** Check if transaction TX1 sees transaction TX0.
@param[in] thd MySQL thread
@param[out] result true if TX1 sees TX0
@param[in] trx_id1 TX1 TRX_ID
@param[in] trx_id0 TX0 TRX_ID
@param[in] commit_id1 TX1 COMMIT_ID
@param[in] iso_level1 TX1 isolation level
@param[in] commit_id0 TX0 COMMIT_ID
@return FALSE if there is no trx_id1 in VTQ, otherwise TRUE */
bool (*vers_trx_sees)(THD *thd, bool &result, ulonglong trx_id1, ulonglong trx_id0,
ulonglong commit_id1, uchar iso_level1, ulonglong commit_id0);
/** Fill TRT record for update.
@param[out] trt TRT table which record[0] will be filled with
transaction data.
@return TRUE if update of TRT is required, FALSE otherwise */
bool (*vers_get_trt_data)(TR_table &trt);
};
@ -1607,11 +1582,21 @@ public:
DBUG_ASSERT(is_started());
m_flags|= (int) TRX_READ_WRITE;
}
void set_trx_tmp_read_write()
{
DBUG_ASSERT(is_started());
m_flags|= (int) TRX_TMP_READ_WRITE;
}
bool is_trx_read_write() const
{
DBUG_ASSERT(is_started());
return m_flags & (int) TRX_READ_WRITE;
}
bool is_trx_tmp_read_write() const
{
DBUG_ASSERT(is_started());
return m_flags & (int) (TRX_READ_WRITE | TRX_TMP_READ_WRITE);
}
bool is_started() const { return m_ht != NULL; }
/** Mark this transaction read-write if the argument is read-write. */
void coalesce_trx_with(const Ha_trx_info *stmt_trx)
@ -1636,7 +1621,7 @@ public:
return m_ht;
}
private:
enum { TRX_READ_ONLY= 0, TRX_READ_WRITE= 1 };
enum { TRX_READ_ONLY= 0, TRX_READ_WRITE= 1, TRX_TMP_READ_WRITE= 2 };
/** Auxiliary, used for ha_list management */
Ha_trx_info *m_next;
/**

25
sql/item.h
View file

@ -303,6 +303,28 @@ public:
}
};
class Name_resolution_context_backup
{
Name_resolution_context &ctx;
TABLE_LIST &table_list;
table_map save_map;
Name_resolution_context_state ctx_state;
public:
Name_resolution_context_backup(Name_resolution_context &_ctx, TABLE_LIST &_table_list)
: ctx(_ctx), table_list(_table_list), save_map(_table_list.map)
{
ctx_state.save_state(&ctx, &table_list);
ctx.table_list= &table_list;
ctx.first_name_resolution_table= &table_list;
}
~Name_resolution_context_backup()
{
ctx_state.restore_state(&ctx, &table_list);
table_list.map= save_map;
}
};
/*
This enum is used to report information about monotonicity of function
@ -1944,9 +1966,6 @@ public:
{
marker &= ~EXTRACTION_MASK;
}
/* System versioning */
virtual vtq_record_t *vtq_cached_result() { return NULL; }
};
template <class T>

143
sql/item_vers.cc
View file

@ -23,32 +23,15 @@
#include "tztime.h"
#include "item.h"
Item_func_vtq_ts::Item_func_vtq_ts(
THD *thd,
handlerton* hton,
Item* a,
vtq_field_t _vtq_field) :
VTQ_common<Item_datetimefunc>(thd, hton, a),
Item_func_vtq_ts::Item_func_vtq_ts(THD *thd, Item* a, TR_table::field_id_t _vtq_field) :
Item_datetimefunc(thd, a),
vtq_field(_vtq_field)
{
decimals= 6;
null_value= true;
DBUG_ASSERT(arg_count == 1 && args[0]);
check_hton();
}
template <class Item_func_X>
void
VTQ_common<Item_func_X>::check_hton()
{
DBUG_ASSERT(hton);
if (!(hton->flags & HTON_NATIVE_SYS_VERSIONING) && hton->db_type != DB_TYPE_HEAP)
{
my_error(ER_VERS_ENGINE_UNSUPPORTED, MYF(0), this->Item::name.str ?
this->Item::name.str : this->func_name());
hton= NULL;
}
}
bool
Item_func_vtq_ts::get_date(MYSQL_TIME *res, ulonglong fuzzy_date)
@ -72,58 +55,46 @@ Item_func_vtq_ts::get_date(MYSQL_TIME *res, ulonglong fuzzy_date)
return false;
}
DBUG_ASSERT(hton && hton->vers_query_trx_id);
null_value= !hton->vers_query_trx_id(thd, res, trx_id, vtq_field);
TR_table trt(thd);
null_value= !trt.query(trx_id);
if (null_value)
{
my_error(ER_VERS_NO_TRX_ID, MYF(0), trx_id);
return true;
}
return null_value;
return trt[vtq_field]->get_date(res, fuzzy_date);
}
Item_func_vtq_id::Item_func_vtq_id(
THD *thd,
handlerton *hton,
Item* a,
vtq_field_t _vtq_field,
bool _backwards) :
VTQ_common<Item_longlong_func>(thd, hton, a),
Item_func_vtq_id::Item_func_vtq_id(THD *thd, Item* a, TR_table::field_id_t _vtq_field,
bool _backwards) :
Item_longlong_func(thd, a),
vtq_field(_vtq_field),
backwards(_backwards)
{
memset(&cached_result, 0, sizeof(cached_result));
decimals= 0;
unsigned_flag= 1;
null_value= true;
DBUG_ASSERT(arg_count == 1 && args[0]);
check_hton();
}
Item_func_vtq_id::Item_func_vtq_id(
THD *thd,
handlerton *hton,
Item* a,
Item* b,
vtq_field_t _vtq_field) :
VTQ_common<Item_longlong_func>(thd, hton, a, b),
Item_func_vtq_id::Item_func_vtq_id(THD *thd, Item* a, Item* b, TR_table::field_id_t _vtq_field) :
Item_longlong_func(thd, a, b),
vtq_field(_vtq_field),
backwards(false)
{
memset(&cached_result, 0, sizeof(cached_result));
decimals= 0;
unsigned_flag= 1;
null_value= true;
DBUG_ASSERT(arg_count == 2 && args[0] && args[1]);
check_hton();
}
longlong
Item_func_vtq_id::get_by_trx_id(ulonglong trx_id)
{
ulonglong res;
THD *thd= current_thd; // can it differ from constructor's?
THD *thd= current_thd;
DBUG_ASSERT(thd);
if (trx_id == ULONGLONG_MAX)
@ -132,52 +103,34 @@ Item_func_vtq_id::get_by_trx_id(ulonglong trx_id)
return 0;
}
DBUG_ASSERT(hton->vers_query_trx_id);
null_value= !hton->vers_query_trx_id(thd, &res, trx_id, vtq_field);
return res;
TR_table trt(thd);
null_value= !trt.query(trx_id);
if (null_value)
return 0;
return trt[vtq_field]->val_int();
}
longlong
Item_func_vtq_id::get_by_commit_ts(MYSQL_TIME &commit_ts, bool backwards)
{
THD *thd= current_thd; // can it differ from constructor's?
THD *thd= current_thd;
DBUG_ASSERT(thd);
DBUG_ASSERT(hton->vers_query_commit_ts);
null_value= !hton->vers_query_commit_ts(thd, &cached_result, commit_ts, VTQ_ALL, backwards);
TR_table trt(thd);
null_value= !trt.query(commit_ts, backwards);
if (null_value)
{
return 0;
}
switch (vtq_field)
{
case VTQ_COMMIT_ID:
return cached_result.commit_id;
case VTQ_ISO_LEVEL:
return cached_result.iso_level;
case VTQ_TRX_ID:
return cached_result.trx_id;
default:
DBUG_ASSERT(0);
null_value= true;
}
return 0;
return trt[vtq_field]->val_int();
}
longlong
Item_func_vtq_id::val_int()
{
if (!hton)
{
null_value= true;
return 0;
}
if (args[0]->is_null())
{
if (arg_count < 2 || vtq_field == VTQ_TRX_ID)
if (arg_count < 2 || vtq_field == TR_table::FLD_TRX_ID)
{
null_value= true;
return 0;
@ -201,12 +154,8 @@ Item_func_vtq_id::val_int()
}
}
Item_func_vtq_trx_sees::Item_func_vtq_trx_sees(
THD *thd,
handlerton *hton,
Item* a,
Item* b) :
VTQ_common<Item_bool_func>(thd, hton, a, b),
Item_func_vtq_trx_sees::Item_func_vtq_trx_sees(THD *thd, Item* a, Item* b) :
Item_bool_func(thd, a, b),
accept_eq(false)
{
null_value= true;
@ -219,42 +168,12 @@ Item_func_vtq_trx_sees::val_int()
THD *thd= current_thd;
DBUG_ASSERT(thd);
if (!hton)
{
null_value= true;
return 0;
}
ulonglong trx_id1, trx_id0;
ulonglong commit_id1= 0;
ulonglong commit_id0= 0;
uchar iso_level1= 0;
DBUG_ASSERT(arg_count > 1);
trx_id1= args[0]->val_uint();
trx_id0= args[1]->val_uint();
ulonglong trx_id1= args[0]->val_uint();
ulonglong trx_id0= args[1]->val_uint();
bool result= accept_eq;
vtq_record_t *cached= args[0]->vtq_cached_result();
if (cached && cached->commit_id)
{
commit_id1= cached->commit_id;
iso_level1= cached->iso_level;
}
cached= args[1]->vtq_cached_result();
if (cached && cached->commit_id)
{
commit_id0= cached->commit_id;
}
if (accept_eq && trx_id1 && trx_id1 == trx_id0)
{
null_value= false;
return true;
}
DBUG_ASSERT(hton->vers_trx_sees);
bool result= false;
null_value= !hton->vers_trx_sees(thd, result, trx_id1, trx_id0, commit_id1, iso_level1, commit_id0);
TR_table trt(thd);
null_value= trt.query_sees(result, trx_id1, trx_id0);
return result;
}

53
sql/item_vers.h
View file

@ -18,36 +18,18 @@
/* System Versioning items */
#include "vtq.h"
#ifdef USE_PRAGMA_INTERFACE
#pragma interface /* gcc class implementation */
#endif
template <class Item_func_X>
class VTQ_common : public Item_func_X
class Item_func_vtq_ts: public Item_datetimefunc
{
protected:
handlerton *hton;
void check_hton();
TR_table::field_id_t vtq_field;
public:
VTQ_common(THD *thd, handlerton* _hton, Item* a, Item* b) :
Item_func_X(thd, a, b),
hton(_hton) {}
VTQ_common(THD *thd, handlerton* _hton, Item* a) :
Item_func_X(thd, a),
hton(_hton) {}
};
class Item_func_vtq_ts :
public VTQ_common<Item_datetimefunc>
{
vtq_field_t vtq_field;
public:
Item_func_vtq_ts(THD *thd, handlerton *hton, Item* a, vtq_field_t _vtq_field);
Item_func_vtq_ts(THD *thd, Item* a, TR_table::field_id_t _vtq_field);
const char *func_name() const
{
if (vtq_field == VTQ_BEGIN_TS)
if (vtq_field == TR_table::FLD_BEGIN_TS)
{
return "vtq_begin_ts";
}
@ -59,31 +41,27 @@ public:
void fix_length_and_dec() { fix_attributes_datetime(decimals); }
};
class Item_func_vtq_id :
public VTQ_common<Item_longlong_func>
class Item_func_vtq_id : public Item_longlong_func
{
vtq_field_t vtq_field;
vtq_record_t cached_result;
TR_table::field_id_t vtq_field;
bool backwards;
longlong get_by_trx_id(ulonglong trx_id);
longlong get_by_commit_ts(MYSQL_TIME &commit_ts, bool backwards);
public:
Item_func_vtq_id(THD *thd, handlerton *hton, Item* a, vtq_field_t _vtq_field, bool _backwards= false);
Item_func_vtq_id(THD *thd, handlerton *hton, Item* a, Item* b, vtq_field_t _vtq_field);
vtq_record_t *vtq_cached_result() { return &cached_result; }
Item_func_vtq_id(THD *thd, Item* a, TR_table::field_id_t _vtq_field, bool _backwards= false);
Item_func_vtq_id(THD *thd, Item* a, Item* b, TR_table::field_id_t _vtq_field);
const char *func_name() const
{
switch (vtq_field)
{
case VTQ_TRX_ID:
case TR_table::FLD_TRX_ID:
return "vtq_trx_id";
case VTQ_COMMIT_ID:
case TR_table::FLD_COMMIT_ID:
return "vtq_commit_id";
case VTQ_ISO_LEVEL:
case TR_table::FLD_ISO_LEVEL:
return "vtq_iso_level";
default:
DBUG_ASSERT(0);
@ -102,14 +80,13 @@ public:
{ return get_item_copy<Item_func_vtq_id>(thd, mem_root, this); }
};
class Item_func_vtq_trx_sees :
public VTQ_common<Item_bool_func>
class Item_func_vtq_trx_sees : public Item_bool_func
{
protected:
bool accept_eq;
public:
Item_func_vtq_trx_sees(THD *thd, handlerton *hton, Item* a, Item* b);
Item_func_vtq_trx_sees(THD *thd, Item* a, Item* b);
const char *func_name() const
{
return "vtq_trx_sees";
@ -123,8 +100,8 @@ class Item_func_vtq_trx_sees_eq :
public Item_func_vtq_trx_sees
{
public:
Item_func_vtq_trx_sees_eq(THD *thd, handlerton *hton, Item* a, Item* b) :
Item_func_vtq_trx_sees(thd, hton, a, b)
Item_func_vtq_trx_sees_eq(THD *thd, Item* a, Item* b) :
Item_func_vtq_trx_sees(thd, a, b)
{
accept_eq= true;
}

32
sql/opt_range.h
View file

@ -1651,6 +1651,38 @@ class SQL_SELECT :public Sql_alloc {
};
class SQL_SELECT_auto
{
SQL_SELECT *select;
public:
SQL_SELECT_auto(): select(NULL)
{}
~SQL_SELECT_auto()
{
delete select;
}
SQL_SELECT_auto&
operator= (SQL_SELECT *_select)
{
select= _select;
return *this;
}
operator SQL_SELECT * () const
{
return select;
}
SQL_SELECT *
operator-> () const
{
return select;
}
operator bool () const
{
return select;
}
};
class FT_SELECT: public QUICK_RANGE_SELECT
{
public:

6
sql/partition_info.cc
View file

@ -3387,8 +3387,8 @@ bool partition_info::vers_trx_id_to_ts(THD* thd, Field* in_trx_id, Field_timesta
handlerton *hton= plugin_hton(table->s->db_plugin);
DBUG_ASSERT(hton);
ulonglong trx_id= in_trx_id->val_int();
MYSQL_TIME ts;
bool found= hton->vers_query_trx_id(thd, &ts, trx_id, VTQ_COMMIT_TS);
TR_table trt(thd);
bool found= trt.query(trx_id);
if (!found)
{
push_warning_printf(thd,
@ -3398,6 +3398,8 @@ bool partition_info::vers_trx_id_to_ts(THD* thd, Field* in_trx_id, Field_timesta
trx_id);
return true;
}
MYSQL_TIME ts;
trt[TR_table::FLD_COMMIT_TS]->get_date(&ts, 0);
out_ts.store_time_dec(&ts, 6);
return false;
}

2
sql/share/errmsg-utf8.txt
View file

@ -7894,7 +7894,7 @@ ER_VERS_HISTORY_LOCK
eng "Versioned SELECT write-locking of history rows"
ER_VERS_NO_TRX_ID
eng "TRX_ID %lu not found in VTQ"
eng "TRX_ID %lu not found in `mysql.transaction_registry`"
ER_WRONG_TABLESPACE_NAME 42000
eng "Incorrect tablespace name `%-.192s`"

1
sql/sql_lex.cc
View file

@ -7298,6 +7298,7 @@ Query_tables_backup::Query_tables_backup(THD* _thd) :
thd(_thd)
{
thd->lex->reset_n_backup_query_tables_list(&backup);
thd->lex->sql_command= backup.sql_command;
}

20
sql/sql_select.cc
View file

@ -924,8 +924,6 @@ int vers_setup_select(THD *thd, TABLE_LIST *tables, COND **where_expr,
else
{
DBUG_ASSERT(table->table->s && table->table->s->db_plugin);
handlerton *hton= plugin_hton(table->table->s->db_plugin);
DBUG_ASSERT(hton);
Item *trx_id0, *trx_id1;
@ -937,27 +935,27 @@ int vers_setup_select(THD *thd, TABLE_LIST *tables, COND **where_expr,
break;
case FOR_SYSTEM_TIME_AS_OF:
trx_id0= vers_conditions.unit_start == UNIT_TIMESTAMP ?
newx Item_func_vtq_id(thd, hton, vers_conditions.start, VTQ_TRX_ID) :
newx Item_func_vtq_id(thd, vers_conditions.start, TR_table::FLD_TRX_ID) :
vers_conditions.start;
cond1= newx Item_func_vtq_trx_sees_eq(thd, hton, trx_id0, row_start);
cond2= newx Item_func_vtq_trx_sees(thd, hton, row_end, trx_id0);
cond1= newx Item_func_vtq_trx_sees_eq(thd, trx_id0, row_start);
cond2= newx Item_func_vtq_trx_sees(thd, row_end, trx_id0);
break;
case FOR_SYSTEM_TIME_FROM_TO:
case FOR_SYSTEM_TIME_BETWEEN:
trx_id0= vers_conditions.unit_start == UNIT_TIMESTAMP ?
newx Item_func_vtq_id(thd, hton, vers_conditions.start, VTQ_TRX_ID, true) :
newx Item_func_vtq_id(thd, vers_conditions.start, TR_table::FLD_TRX_ID, true) :
vers_conditions.start;
trx_id1= vers_conditions.unit_end == UNIT_TIMESTAMP ?
newx Item_func_vtq_id(thd, hton, vers_conditions.end, VTQ_TRX_ID, false) :
newx Item_func_vtq_id(thd, vers_conditions.end, TR_table::FLD_TRX_ID, false) :
vers_conditions.end;
cond1= vers_conditions.type == FOR_SYSTEM_TIME_FROM_TO ?
newx Item_func_vtq_trx_sees(thd, hton, trx_id1, row_start) :
newx Item_func_vtq_trx_sees_eq(thd, hton, trx_id1, row_start);
cond2= newx Item_func_vtq_trx_sees_eq(thd, hton, row_end, trx_id0);
newx Item_func_vtq_trx_sees(thd, trx_id1, row_start) :
newx Item_func_vtq_trx_sees_eq(thd, trx_id1, row_start);
cond2= newx Item_func_vtq_trx_sees_eq(thd, row_end, trx_id0);
break;
case FOR_SYSTEM_TIME_BEFORE:
trx_id0= vers_conditions.unit_start == UNIT_TIMESTAMP ?
newx Item_func_vtq_id(thd, hton, vers_conditions.start, VTQ_TRX_ID) :
newx Item_func_vtq_id(thd, vers_conditions.start, TR_table::FLD_TRX_ID) :
vers_conditions.start;
cond1= newx Item_func_lt(thd, row_end, trx_id0);
break;

205
sql/table.cc
View file

@ -67,6 +67,8 @@ LEX_CSTRING GENERAL_LOG_NAME= {STRING_WITH_LEN("general_log")};
/* SLOW_LOG name */
LEX_CSTRING SLOW_LOG_NAME= {STRING_WITH_LEN("slow_log")};
LEX_CSTRING TRANSACTION_REG_NAME= {STRING_WITH_LEN("transaction_registry")};
/*
Keyword added as a prefix when parsing the defining expression for a
virtual column read from the column definition saved in the frm file
@ -264,6 +266,9 @@ TABLE_CATEGORY get_table_category(const LEX_CSTRING *db,
if (lex_string_eq(&SLOW_LOG_NAME, name) == 0)
return TABLE_CATEGORY_LOG;
if (lex_string_eq(&TRANSACTION_REG_NAME, name) == 0)
return TABLE_CATEGORY_LOG;
}
return TABLE_CATEGORY_USER;
@ -8479,6 +8484,206 @@ LEX_CSTRING *fk_option_name(enum_fk_option opt)
return names + opt;
}
TR_table::TR_table(THD* _thd, bool rw) : thd(_thd)
{
init_one_table(LEX_STRING_WITH_LEN(MYSQL_SCHEMA_NAME),
LEX_STRING_WITH_LEN(TRANSACTION_REG_NAME),
TRANSACTION_REG_NAME.str, rw ? TL_WRITE : TL_READ);
open_tables_backup= new Open_tables_backup;
if (open_tables_backup)
open_log_table(thd, this, open_tables_backup);
else
my_error(ER_OUT_OF_RESOURCES, MYF(0));
}
TR_table::~TR_table()
{
if (table)
close_log_table(thd, open_tables_backup);
delete open_tables_backup;
}
void TR_table::store(uint field_id, ulonglong val)
{
table->field[field_id]->store(val, true);
table->field[field_id]->set_notnull();
}
void TR_table::store(uint field_id, timeval ts)
{
table->field[field_id]->store_timestamp(ts.tv_sec, ts.tv_usec);
table->field[field_id]->set_notnull();
}
void TR_table::store_data(ulonglong trx_id, ulonglong commit_id, timeval commit_ts)
{
timeval start_time= {thd->start_time, thd->start_time_sec_part};
store(FLD_TRX_ID, trx_id);
store(FLD_COMMIT_ID, commit_id);
store(FLD_COMMIT_TS, commit_ts);
store(FLD_BEGIN_TS, start_time);
store_iso_level(thd->tx_isolation);
}
enum_tx_isolation TR_table::iso_level() const
{
enum_tx_isolation res= (enum_tx_isolation) ((*this)[FLD_ISO_LEVEL]->val_int() - 1);
DBUG_ASSERT(res <= ISO_SERIALIZABLE);
return res;
}
bool TR_table::update(bool &updated)
{
if (!table)
return true;
DBUG_ASSERT(table->s);
handlerton *hton= table->s->db_type();
DBUG_ASSERT(hton);
DBUG_ASSERT(hton->flags & HTON_NATIVE_SYS_VERSIONING);
if ((updated= hton->vers_get_trt_data(*this)))
{
int error= table->file->ha_write_row(table->record[0]);
if (error)
{
table->file->print_error(error, MYF(0));
}
return error;
}
return false;
}
#define newx new (thd->mem_root)
bool TR_table::query(ulonglong trx_id)
{
if (!table)
return false;
SQL_SELECT_auto select;
READ_RECORD info;
int error;
List<TABLE_LIST> dummy;
SELECT_LEX &slex= thd->lex->select_lex;
Name_resolution_context_backup backup(slex.context, *this);
Item *field= newx Item_field(thd, &slex.context, (*this)[FLD_TRX_ID]);
Item *value= newx Item_int(thd, trx_id);
COND *conds= newx Item_func_eq(thd, field, value);
if ((error= setup_conds(thd, this, dummy, &conds)))
return false;
select= make_select(table, 0, 0, conds, NULL, 0, &error);
if (error || !select)
return false;
// FIXME: (performance) force index 'transaction_id'
error= init_read_record(&info, thd, table, select, NULL,
1 /* use_record_cache */, true /* print_error */,
false /* disable_rr_cache */);
while (!(error= info.read_record()) && !thd->killed && !thd->is_error())
{
if (select->skip_record(thd) > 0)
return true;
}
return false;
}
bool TR_table::query(MYSQL_TIME &commit_time, bool backwards)
{
if (!table)
return false;
SQL_SELECT_auto select;
READ_RECORD info;
int error;
List<TABLE_LIST> dummy;
SELECT_LEX &slex= thd->lex->select_lex;
Name_resolution_context_backup backup(slex.context, *this);
Item *field= newx Item_field(thd, &slex.context, (*this)[FLD_COMMIT_TS]);
Item *value= newx Item_datetime_literal(thd, &commit_time, 6);
COND *conds;
if (backwards)
conds= newx Item_func_ge(thd, field, value);
else
conds= newx Item_func_le(thd, field, value);
if ((error= setup_conds(thd, this, dummy, &conds)))
return false;
// FIXME: (performance) force index 'commit_timestamp'
select= make_select(table, 0, 0, conds, NULL, 0, &error);
if (error || !select)
return false;
error= init_read_record(&info, thd, table, select, NULL,
1 /* use_record_cache */, true /* print_error */,
false /* disable_rr_cache */);
// With PK by transaction_id the records are ordered by PK
bool found= false;
while (!(error= info.read_record()) && !thd->killed && !thd->is_error())
{
if (select->skip_record(thd) > 0)
{
if (backwards)
return true;
found= true;
// TODO: (performance) make ORDER DESC and break after first found.
// Otherwise it is O(n) scan (+copy)!
store_record(table, record[1]);
}
else
{
if (found)
restore_record(table, record[1]);
if (!backwards)
break;
}
}
return found;
}
#undef newx
bool TR_table::query_sees(bool &result, ulonglong trx_id1, ulonglong trx_id0,
ulonglong commit_id1, enum_tx_isolation iso_level1,
ulonglong commit_id0)
{
if (trx_id1 == trx_id0)
{
return false;
}
if (trx_id1 == ULONGLONG_MAX || trx_id0 == 0)
{
result= true;
return false;
}
if (!commit_id1)
{
if (!query(trx_id1))
return true;
commit_id1= (*this)[FLD_COMMIT_ID]->val_int();
iso_level1= iso_level();
}
if (!commit_id0)
{
if (!query(trx_id0))
return true;
commit_id0= (*this)[FLD_COMMIT_ID]->val_int();
}
// Trivial case: TX1 started after TX0 committed
if (trx_id1 > commit_id0
// Concurrent transactions: TX1 committed after TX0 and TX1 is read (un)committed
|| (commit_id1 > commit_id0 && iso_level1 < ISO_REPEATABLE_READ))
{
result= true;
}
else // All other cases: TX1 does not see TX0
{
result= false;
}
return false;
}
void vers_select_conds_t::resolve_units(bool timestamps_only)
{
DBUG_ASSERT(type != FOR_SYSTEM_TIME_UNSPECIFIED);

59
sql/table.h
View file

@ -2894,6 +2894,7 @@ extern LEX_CSTRING PERFORMANCE_SCHEMA_DB_NAME;
extern LEX_CSTRING GENERAL_LOG_NAME;
extern LEX_CSTRING SLOW_LOG_NAME;
extern LEX_CSTRING TRANSACTION_REG_NAME;
/* information schema */
extern LEX_CSTRING INFORMATION_SCHEMA_NAME;
@ -2922,6 +2923,64 @@ inline void mark_as_null_row(TABLE *table)
bool is_simple_order(ORDER *order);
/** Transaction Registry Table (TRT)
This table holds transaction IDs, their corresponding times and other
transaction-related data which is used for transaction order resolution.
When versioned table marks its records lifetime with transaction IDs,
TRT is used to get their actual timestamps. */
class Open_tables_backup;
class TR_table: public TABLE_LIST
{
THD *thd;
Open_tables_backup *open_tables_backup;
public:
enum field_id_t {
FLD_TRX_ID= 0,
FLD_COMMIT_ID,
FLD_BEGIN_TS,
FLD_COMMIT_TS,
FLD_ISO_LEVEL,
FIELD_COUNT
};
TR_table(THD *_thd, bool rw= false);
~TR_table();
THD *get_thd() const { return thd; }
void store(uint field_id, ulonglong val);
void store(uint field_id, timeval ts);
void store_data(ulonglong trx_id, ulonglong commit_id, timeval commit_ts);
bool update(bool &updated);
// return true if found; false if not found or error
bool query(ulonglong trx_id);
bool query(MYSQL_TIME &commit_time, bool backwards);
// return true if error
bool query_sees(bool &result, ulonglong trx_id1, ulonglong trx_id0,
ulonglong commit_id1= 0, enum_tx_isolation iso_level1= ISO_READ_UNCOMMITTED,
ulonglong commit_id0= 0);
TABLE * operator-> () const
{
return table;
}
Field * operator[] (uint field_id) const
{
DBUG_ASSERT(field_id < FIELD_COUNT);
return table->field[field_id];
}
operator bool () const
{
return table;
}
enum_tx_isolation iso_level() const;
void store_iso_level(enum_tx_isolation iso_level)
{
DBUG_ASSERT(iso_level <= ISO_SERIALIZABLE);
store(FLD_ISO_LEVEL, iso_level + 1);
}
};
#endif /* MYSQL_CLIENT */
#endif /* TABLE_INCLUDED */

7
sql/vtmd.cc
View file

@ -34,7 +34,6 @@ VTMD_table::create(THD *thd)
TL_READ);
Query_tables_backup backup(thd);
thd->lex->sql_command= backup.get().sql_command;
thd->lex->add_to_query_tables(&src_table);
MDL_auto_lock mdl_lock(thd, table);
@ -256,6 +255,12 @@ err:
}
quit:
if (!result)
{
TR_table trt(thd, true);
result= trt.update(result);
}
close_log_table(thd, &open_tables_backup);
open_error:

47
sql/vtq.h
View file

@ -1,47 +0,0 @@
#ifndef VTQ_INCLUDED
#define VTQ_INCLUDED
/* Copyright (c) 2016, 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 Street, Fifth Floor, Boston, MA 02110-1301, USA */
/**
VTQ stands for 'versioning transaction query': InnoDB system table that holds
transaction IDs, their corresponding times and other transaction-related
data which is used for transaction order resolution. When versioned table
marks its records lifetime with transaction IDs, VTQ is used to get their
actual timestamps. */
enum vtq_field_t
{
VTQ_ALL = 0,
VTQ_TRX_ID,
VTQ_COMMIT_ID,
VTQ_BEGIN_TS,
VTQ_COMMIT_TS,
VTQ_ISO_LEVEL
};
struct vtq_record_t
{
ulonglong trx_id;
ulonglong commit_id;
timeval begin_ts;
timeval commit_ts;
uchar iso_level;
};
#endif /* VTQ_INCLUDED */

5
storage/heap/ha_heap.cc
View file

@ -26,9 +26,6 @@
#include "ha_heap.h"
#include "sql_base.h" // enum_tdc_remove_table_type
bool vtq_query_trx_id(THD *thd, void *out, ulonglong in_trx_id,
vtq_field_t field);
static handler *heap_create_handler(handlerton *hton,
TABLE_SHARE *table,
MEM_ROOT *mem_root);
@ -58,8 +55,6 @@ int heap_init(void *p)
heap_hton->panic= heap_panic;
heap_hton->flags= HTON_CAN_RECREATE;
heap_hton->vers_query_trx_id = vtq_query_trx_id;
return 0;
}

3
storage/innobase/CMakeLists.txt
View file

@ -147,8 +147,7 @@ SET(INNOBASE_SOURCES
ut/ut0ut.cc
ut/ut0vec.cc
ut/ut0wqueue.cc
ut/ut0timer.cc
vers/vers0vtq.cc)
ut/ut0timer.cc)
MYSQL_ADD_PLUGIN(innobase ${INNOBASE_SOURCES} STORAGE_ENGINE
DEFAULT RECOMPILE_FOR_EMBEDDED

124
storage/innobase/dict/dict0crea.cc
View file

@ -1877,130 +1877,6 @@ dict_create_or_check_sys_virtual()
return(err);
}
UNIV_INTERN
dberr_t
dict_create_or_check_vtq_table(void)
/*================================================*/
{
trx_t* trx;
my_bool srv_file_per_table_backup;
dberr_t err;
dberr_t sys_vtq_err;
dict_index_t* index;
ut_a(srv_get_active_thread_type() == SRV_NONE);
/* Note: The master thread has not been started at this point. */
static const int vtq_num_indexes = 4;
sys_vtq_err = dict_check_if_system_table_exists(
"SYS_VTQ", DICT_NUM_FIELDS__SYS_VTQ + 1, vtq_num_indexes);
if (sys_vtq_err == DB_SUCCESS) {
err = DB_SUCCESS;
goto assign_and_exit;
}
trx = trx_allocate_for_mysql();
trx_set_dict_operation(trx, TRX_DICT_OP_TABLE);
trx->op_info = "creating VTQ sys table";
row_mysql_lock_data_dictionary(trx);
/* Check which incomplete table definition to drop. */
if (sys_vtq_err == DB_CORRUPTION) {
ib::warn() <<
"Dropping incompletely created "
"SYS_VTQ table.";
row_drop_table_for_mysql("SYS_VTQ", trx, false, TRUE);
}
ib::info() <<
"Creating VTQ system table.";
srv_file_per_table_backup = srv_file_per_table;
/* We always want SYSTEM tables to be created inside the system
tablespace. */
srv_file_per_table = 0;
err = que_eval_sql(
NULL,
"PROCEDURE CREATE_VTQ_SYS_TABLE_PROC () IS\n"
"BEGIN\n"
"CREATE TABLE\n"
"SYS_VTQ("
"TRX_ID BIGINT UNSIGNED, "
"COMMIT_ID BIGINT UNSIGNED, "
"BEGIN_TS BIGINT UNSIGNED, "
"COMMIT_TS BIGINT UNSIGNED, "
"ISO_LEVEL CHAR(1));\n"
"CREATE UNIQUE CLUSTERED INDEX TRX_ID_IND"
" ON SYS_VTQ (TRX_ID);\n"
"CREATE INDEX COMMIT_ID_IND"
" ON SYS_VTQ (COMMIT_ID);\n"
"CREATE INDEX BEGIN_TS_IND"
" ON SYS_VTQ (BEGIN_TS);\n"
"CREATE INDEX COMMIT_TS_IND"
" ON SYS_VTQ (COMMIT_TS);\n"
"END;\n",
FALSE, trx);
if (err != DB_SUCCESS) {
ib::error() << "Creation of SYS_VTQ"
" failed: " << ut_strerr(err) << ". Tablespace is"
" full or too many transactions."
" Dropping incompletely created tables.";
ut_ad(err == DB_OUT_OF_FILE_SPACE
|| err == DB_TOO_MANY_CONCURRENT_TRXS);
row_drop_table_for_mysql("SYS_VTQ", trx, false, TRUE);
if (err == DB_OUT_OF_FILE_SPACE) {
err = DB_MUST_GET_MORE_FILE_SPACE;
}
}
trx_commit_for_mysql(trx);
row_mysql_unlock_data_dictionary(trx);
trx_free_for_mysql(trx);
srv_file_per_table = srv_file_per_table_backup;
if (err == DB_SUCCESS) {
ib::info() <<
"VTQ system table created";
}
/* Note: The master thread has not been started at this point. */
/* Confirm and move to the non-LRU part of the table LRU list. */
sys_vtq_err = dict_check_if_system_table_exists(
"SYS_VTQ", DICT_NUM_FIELDS__SYS_VTQ + 1, vtq_num_indexes);
ut_a(sys_vtq_err == DB_SUCCESS);
assign_and_exit:
mutex_enter(&dict_sys->mutex);
dict_sys->sys_vtq = dict_table_get_low("SYS_VTQ");
ut_ad(dict_sys->sys_vtq);
index = dict_table_get_first_index(dict_sys->sys_vtq);
for (int i = 0; i < 3; ++i) {
index = dict_table_get_next_index(index);
ut_ad(index);
}
ut_ad(strcmp(index->name, "COMMIT_TS_IND") == 0);
dict_sys->vtq_commit_ts_ind = index;
mutex_exit(&dict_sys->mutex);
return(err);
}
/****************************************************************//**
Evaluate the given foreign key SQL statement.
@return error code or DB_SUCCESS */

69
storage/innobase/dict/dict0load.cc
View file

@ -824,75 +824,6 @@ const char* dict_print_error(mem_heap_t* heap, ulint col, ulint len, ulint expec
col, len, expected);
}
/********************************************************************//**
This function parses a SYS_VTQ record, extracts necessary
information from the record and returns it to the caller.
@return error message, or NULL on success */
UNIV_INTERN
const char*
dict_process_sys_vtq(
/*=======================*/
mem_heap_t* heap, /*!< in/out: heap memory */
const rec_t* rec, /*!< in: current rec */
vtq_record_t& out /*!< out: field values */
)
{
ulint len, nfld;
const byte *field;
if (rec_get_deleted_flag(rec, 0)) {
return("delete-marked record in SYS_VTQ");
}
if (rec_get_n_fields_old(rec) != DICT_NUM_FIELDS__SYS_VTQ) {
return("wrong number of columns in SYS_VTQ record");
}
/* TRX_ID */
field = rec_get_nth_field_old(
rec, (nfld = DICT_FLD__SYS_VTQ__TRX_ID), &len);
if (len != sizeof(trx_id_t))
return dict_print_error(heap, nfld, len, sizeof(trx_id_t));
out.trx_id = mach_read_from_8(field);
/* COMMIT_ID */
field = rec_get_nth_field_old(
rec, (nfld = DICT_FLD__SYS_VTQ__COMMIT_ID), &len);
if (len != sizeof(trx_id_t))
return dict_print_error(heap, nfld, len, sizeof(trx_id_t));
out.commit_id = mach_read_from_8(field);
/* BEGIN_TS */
field = rec_get_nth_field_old(
rec, (nfld = DICT_FLD__SYS_VTQ__BEGIN_TS), &len);
if (len != sizeof(uint64_t))
return dict_print_error(heap, nfld, len, sizeof(uint64_t));
out.begin_ts.tv_sec = mach_read_from_4(field);
out.begin_ts.tv_usec = mach_read_from_4(field + 4);
/* COMMIT_TS */
field = rec_get_nth_field_old(
rec, (nfld = DICT_FLD__SYS_VTQ__COMMIT_TS), &len);
if (len != sizeof(uint64_t))
return dict_print_error(heap, nfld, len, sizeof(uint64_t));
out.commit_ts.tv_sec = mach_read_from_4(field);
out.commit_ts.tv_usec = mach_read_from_4(field + 4);
/* ISOLATION_LEVEL */
field = rec_get_nth_field_old(
rec, (nfld = DICT_FLD__SYS_VTQ__ISOLATION_LEVEL), &len);
if (len != sizeof(byte))
return dict_print_error(heap, nfld, len, sizeof(byte));
out.iso_level = *field;
return(NULL);
}
/** Get the first filepath from SYS_DATAFILES for a given space_id.
@param[in] space_id Tablespace ID
@return First filepath (caller must invoke ut_free() on it)

45
storage/innobase/handler/ha_innodb.cc
View file

@ -117,7 +117,6 @@ this program; if not, write to the Free Software Foundation, Inc.,
#include "trx0xa.h"
#include "ut0mem.h"
#include "row0ext.h"
#include "vers0vtq.h"
#define thd_get_trx_isolation(X) ((enum_tx_isolation)thd_tx_isolation(X))
@ -3626,6 +3625,26 @@ static const char* ha_innobase_exts[] = {
NullS
};
bool innodb_get_trt_data(TR_table &trt)
{
THD *thd = trt.get_thd();
trx_t *trx = thd_to_trx(thd);
ut_a(trx);
if (trx->vers_update_trt)
{
timeval commit_ts;
mutex_enter(&trx_sys->mutex);
trx_id_t commit_id = trx_sys_get_new_trx_id();
ut_usectime((ulint *)&commit_ts.tv_sec, (ulint *)&commit_ts.tv_usec);
mutex_exit(&trx_sys->mutex);
trt.store_data(trx->id, commit_id, commit_ts);
trx->vers_update_trt = false;
return true;
}
return false;
}
/*********************************************************************//**
Opens an InnoDB database.
@return 0 on success, 1 on failure */
@ -3694,9 +3713,7 @@ innobase_init(
innobase_hton->table_options = innodb_table_option_list;
/* System Versioning */
innobase_hton->vers_query_trx_id = vtq_query_trx_id;
innobase_hton->vers_query_commit_ts = vtq_query_commit_ts;
innobase_hton->vers_trx_sees = vtq_trx_sees;
innobase_hton->vers_get_trt_data = innodb_get_trt_data;
innodb_remember_check_sysvar_funcs();
@ -4428,14 +4445,6 @@ innobase_commit_ordered_2(
/* Don't do write + flush right now. For group commit
to work we want to do the flush later. */
trx->flush_log_later = true;
/* Notify VTQ on System Versioned tables update */
if (trx->vtq_notify_on_commit) {
vers_notify_vtq(trx);
trx->vtq_notify_on_commit = false;
}
} else {
DBUG_ASSERT(!trx->vtq_notify_on_commit);
}
innobase_commit_low(trx);
@ -4559,11 +4568,6 @@ innobase_commit(
if (commit_trx
|| (!thd_test_options(thd, OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN))) {
/* Notify VTQ on System Versioned tables update */
if (trx->vtq_notify_on_commit) {
vers_notify_vtq(trx);
trx->vtq_notify_on_commit = false;
}
DBUG_EXECUTE_IF("crash_innodb_before_commit",
DBUG_SUICIDE(););
@ -8870,7 +8874,7 @@ calc_row_difference(
n_changed++;
if (!prebuilt->upd_node->versioned &&
prebuilt->table->with_versioning() &&
prebuilt->table->versioned() &&
!(field->flags & VERS_OPTIMIZED_UPDATE_FLAG)) {
prebuilt->upd_node->versioned = true;
}
@ -8981,7 +8985,7 @@ calc_row_difference(
++n_changed;
if (!prebuilt->upd_node->versioned &&
prebuilt->table->with_versioning() &&
prebuilt->table->versioned() &&
!(field->flags & VERS_OPTIMIZED_UPDATE_FLAG)) {
prebuilt->upd_node->versioned = true;
}
@ -21291,8 +21295,7 @@ i_s_innodb_sys_virtual,
i_s_innodb_mutexes,
i_s_innodb_sys_semaphore_waits,
i_s_innodb_tablespaces_encryption,
i_s_innodb_tablespaces_scrubbing,
i_s_innodb_vtq
i_s_innodb_tablespaces_scrubbing
maria_declare_plugin_end;
/** @brief Initialize the default value of innodb_commit_concurrency.

267
storage/innobase/handler/i_s.cc
View file

@ -9628,270 +9628,3 @@ UNIV_INTERN struct st_maria_plugin i_s_innodb_sys_semaphore_waits =
STRUCT_FLD(version_info, INNODB_VERSION_STR),
STRUCT_FLD(maturity, MariaDB_PLUGIN_MATURITY_STABLE),
};
/* Fields of the dynamic table INFORMATION_SCHEMA.innodb_vtq */
static ST_FIELD_INFO innodb_vtq_fields_info[] =
{
#define SYS_VTQ_TRX_ID 0
{ STRUCT_FLD(field_name, "transaction_id"),
STRUCT_FLD(field_length, MY_INT64_NUM_DECIMAL_DIGITS),
STRUCT_FLD(field_type, MYSQL_TYPE_LONGLONG),
STRUCT_FLD(value, 0),
STRUCT_FLD(field_flags, MY_I_S_UNSIGNED),
STRUCT_FLD(old_name, ""),
STRUCT_FLD(open_method, SKIP_OPEN_TABLE) },
#define SYS_VTQ_COMMIT_ID 1
{ STRUCT_FLD(field_name, "commit_id"),
STRUCT_FLD(field_length, MY_INT64_NUM_DECIMAL_DIGITS),
STRUCT_FLD(field_type, MYSQL_TYPE_LONGLONG),
STRUCT_FLD(value, 0),
STRUCT_FLD(field_flags, MY_I_S_UNSIGNED),
STRUCT_FLD(old_name, ""),
STRUCT_FLD(open_method, SKIP_OPEN_TABLE) },
#define SYS_VTQ_BEGIN_TS 2
{ STRUCT_FLD(field_name, "begin_timestamp"),
STRUCT_FLD(field_length, 6),
STRUCT_FLD(field_type, MYSQL_TYPE_TIMESTAMP),
STRUCT_FLD(value, 0),
STRUCT_FLD(field_flags, 0),
STRUCT_FLD(old_name, ""),
STRUCT_FLD(open_method, SKIP_OPEN_TABLE) },
#define SYS_VTQ_COMMIT_TS 3
{ STRUCT_FLD(field_name, "commit_timestamp"),
STRUCT_FLD(field_length, 6),
STRUCT_FLD(field_type, MYSQL_TYPE_TIMESTAMP),
STRUCT_FLD(value, 0),
STRUCT_FLD(field_flags, 0),
STRUCT_FLD(old_name, ""),
STRUCT_FLD(open_method, SKIP_OPEN_TABLE) },
#define SYS_VTQ_ISO_LEVEL 4
{ STRUCT_FLD(field_name, "isolation_level"),
STRUCT_FLD(field_length, 16),
STRUCT_FLD(field_type, MYSQL_TYPE_STRING),
STRUCT_FLD(value, 0),
STRUCT_FLD(field_flags, 0),
STRUCT_FLD(old_name, ""),
STRUCT_FLD(open_method, SKIP_OPEN_TABLE) },
END_OF_ST_FIELD_INFO
};
/******************************************************************//**
Maps a InnoDB trx isolation level code to the MySQL isolation level name
@return MySQL isolation level name */
static inline
const char*
i_s_isolation_name(
/*=========================*/
ulint iso) /*!< in: InnoDB isolation level code */
{
enum_tx_isolation mysql_iso;
switch (iso) {
case TRX_ISO_REPEATABLE_READ:
mysql_iso = ISO_REPEATABLE_READ;
break;
case TRX_ISO_READ_COMMITTED:
mysql_iso = ISO_READ_COMMITTED;
break;
case TRX_ISO_SERIALIZABLE:
mysql_iso = ISO_SERIALIZABLE;
break;
case TRX_ISO_READ_UNCOMMITTED:
mysql_iso = ISO_READ_UNCOMMITTED;
break;
default:
ut_error;
return NULL;
}
return tx_isolation_names[mysql_iso];
}
/**********************************************************************//**
Function to fill INFORMATION_SCHEMA.INNODB_SYS_VTQ with information
collected by scanning SYS_VTQ table.
@return 0 on success */
static
int
i_s_dict_fill_vtq(
/*========================*/
THD* thd, /*!< in: thread */
vtq_record_t& vtq, /*!< in: table fields */
TABLE* table_to_fill) /*!< in/out: fill this table */
{
Field** fields;
const char* iso_level;
DBUG_ENTER("i_s_dict_fill_vtq");
fields = table_to_fill->field;
iso_level = i_s_isolation_name(vtq.iso_level);
OK(field_store_uint64_t(fields[SYS_VTQ_TRX_ID], vtq.trx_id));
OK(field_store_uint64_t(fields[SYS_VTQ_COMMIT_ID], vtq.commit_id));
OK(field_store_timeval(fields[SYS_VTQ_BEGIN_TS], vtq.begin_ts, thd));
OK(field_store_timeval(fields[SYS_VTQ_COMMIT_TS], vtq.commit_ts, thd));
OK(field_store_string(fields[SYS_VTQ_ISO_LEVEL], iso_level));
OK(schema_table_store_record(thd, table_to_fill));
DBUG_RETURN(0);
}
/*******************************************************************//**
Function to populate INFORMATION_SCHEMA.INNODB_SYS_VTQ table.
Loop through each record in SYS_VTQ, and extract the column
information and fill the INFORMATION_SCHEMA.INNODB_SYS_VTQ table.
@return 0 on success */
static const int I_S_SYS_VTQ_LIMIT = 10000; // maximum number of records in I_S.INNODB_SYS_VTQ
static
int
i_s_sys_vtq_fill_table(
/*=========================*/
THD* thd, /*!< in: thread */
TABLE_LIST* tables, /*!< in/out: tables to fill */
Item* ) /*!< in: condition (not used) */
{
btr_pcur_t pcur;
const rec_t* rec;
mem_heap_t* heap;
mtr_t mtr;
int err = 0;
DBUG_ENTER("i_s_sys_vtq_fill_table");
RETURN_IF_INNODB_NOT_STARTED(tables->schema_table_name);
/* deny access to user without PROCESS_ACL privilege */
if (check_global_access(thd, PROCESS_ACL)) {
DBUG_RETURN(0);
}
heap = mem_heap_create(1000);
mutex_enter(&dict_sys->mutex);
mtr_start(&mtr);
rec = dict_startscan_system(&pcur, &mtr, SYS_VTQ, false);
for (int i = 0; rec && i < I_S_SYS_VTQ_LIMIT; ++i) {
const char* err_msg;
vtq_record_t fields;
/* Extract necessary information from SYS_VTQ row */
err_msg = dict_process_sys_vtq(
heap,
rec,
fields);
mtr_commit(&mtr);
mutex_exit(&dict_sys->mutex);
if (!err_msg) {
err = i_s_dict_fill_vtq(
thd,
fields,
tables->table);
} else {
push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN,
ER_CANT_FIND_SYSTEM_REC, "%s",
err_msg);
err = 1;
}
if (err)
break;
mem_heap_empty(heap);
/* Get the next record */
mutex_enter(&dict_sys->mutex);
mtr_start(&mtr);
rec = dict_getnext_system(&pcur, &mtr);
}
if (!err) {
mtr_commit(&mtr);
mutex_exit(&dict_sys->mutex);
}
mem_heap_free(heap);
DBUG_RETURN(err);
}
/*******************************************************************//**
Bind the dynamic table INFORMATION_SCHEMA.innodb_vtq
@return 0 on success */
static
int
innodb_vtq_init(
/*===================*/
void* p) /*!< in/out: table schema object */
{
ST_SCHEMA_TABLE* schema;
DBUG_ENTER("innodb_vtq_init");
schema = (ST_SCHEMA_TABLE*) p;
schema->fields_info = innodb_vtq_fields_info;
schema->fill_table = i_s_sys_vtq_fill_table;
DBUG_RETURN(0);
}
UNIV_INTERN struct st_maria_plugin i_s_innodb_vtq =
{
/* the plugin type (a MYSQL_XXX_PLUGIN value) */
/* int */
STRUCT_FLD(type, MYSQL_INFORMATION_SCHEMA_PLUGIN),
/* pointer to type-specific plugin descriptor */
/* void* */
STRUCT_FLD(info, &i_s_info),
/* plugin name */
/* const char* */
STRUCT_FLD(name, "INNODB_VTQ"),
/* plugin author (for SHOW PLUGINS) */
/* const char* */
STRUCT_FLD(author, plugin_author),
/* general descriptive text (for SHOW PLUGINS) */
/* const char* */
STRUCT_FLD(descr, "InnoDB Versioning Transaction Query table"),
/* the plugin license (PLUGIN_LICENSE_XXX) */
/* int */
STRUCT_FLD(license, PLUGIN_LICENSE_GPL),
/* the function to invoke when plugin is loaded */
/* int (*)(void*); */
STRUCT_FLD(init, innodb_vtq_init),
/* the function to invoke when plugin is unloaded */
/* int (*)(void*); */
STRUCT_FLD(deinit, i_s_common_deinit),
/* plugin version (for SHOW PLUGINS) */
/* unsigned int */
STRUCT_FLD(version, INNODB_VERSION_SHORT),
/* struct st_mysql_show_var* */
STRUCT_FLD(status_vars, NULL),
/* struct st_mysql_sys_var** */
STRUCT_FLD(system_vars, NULL),
/* Maria extension */
STRUCT_FLD(version_info, INNODB_VERSION_STR),
STRUCT_FLD(maturity, MariaDB_PLUGIN_MATURITY_GAMMA),
};

1
storage/innobase/handler/i_s.h
View file

@ -64,7 +64,6 @@ extern struct st_maria_plugin i_s_innodb_sys_virtual;
extern struct st_maria_plugin i_s_innodb_tablespaces_encryption;
extern struct st_maria_plugin i_s_innodb_tablespaces_scrubbing;
extern struct st_maria_plugin i_s_innodb_sys_semaphore_waits;
extern struct st_maria_plugin i_s_innodb_vtq;
/** maximum number of buffer page info we would cache. */
#define MAX_BUF_INFO_CACHED 10000

24
storage/innobase/include/dict0boot.h
View file

@ -325,30 +325,6 @@ enum dict_fld_sys_datafiles_enum {
DICT_FLD__SYS_DATAFILES__PATH = 3,
DICT_NUM_FIELDS__SYS_DATAFILES = 4
};
/* The columns in SYS_VTQ */
enum dict_col_sys_vtq_enum
{
DICT_COL__SYS_VTQ__TRX_ID = 0,
DICT_COL__SYS_VTQ__COMMIT_ID = 1,
DICT_COL__SYS_VTQ__BEGIN_TS = 2,
DICT_COL__SYS_VTQ__COMMIT_TS = 3,
DICT_COL__SYS_VTQ__ISOLATION_LEVEL = 4,
DICT_NUM_COLS__SYS_VTQ = 5
};
/* The field numbers in the SYS_VTQ clustered index */
enum dict_fld_sys_vtq_enum
{
DICT_FLD__SYS_VTQ__TRX_ID = 0,
DICT_FLD__SYS_VTQ__DB_TRX_ID = 1,
DICT_FLD__SYS_VTQ__DB_ROLL_PTR = 2,
DICT_FLD__SYS_VTQ__COMMIT_ID = 3,
DICT_FLD__SYS_VTQ__BEGIN_TS = 4,
DICT_FLD__SYS_VTQ__COMMIT_TS = 5,
DICT_FLD__SYS_VTQ__ISOLATION_LEVEL = 6,
DICT_NUM_FIELDS__SYS_VTQ = 7
};
/* The columns in SYS_VIRTUAL */
enum dict_col_sys_virtual_enum {
DICT_COL__SYS_VIRTUAL__TABLE_ID = 0,

4
storage/innobase/include/dict0crea.h
View file

@ -311,10 +311,6 @@ struct tab_node_t{
storage */
};
UNIV_INTERN
dberr_t
dict_create_or_check_vtq_table(void);
/* Table create node states */
#define TABLE_BUILD_TABLE_DEF 1
#define TABLE_BUILD_COL_DEF 2

2
storage/innobase/include/dict0dict.h
View file

@ -1708,8 +1708,6 @@ struct dict_sys_t{
dict_table_t* sys_columns; /*!< SYS_COLUMNS table */
dict_table_t* sys_indexes; /*!< SYS_INDEXES table */
dict_table_t* sys_fields; /*!< SYS_FIELDS table */
dict_table_t* sys_vtq; /*!< SYS_VTQ table */
dict_index_t* vtq_commit_ts_ind;
dict_table_t* sys_virtual; /*!< SYS_VIRTUAL table */
/*=============================*/

21
storage/innobase/include/dict0load.h
View file

@ -38,8 +38,6 @@ Created 4/24/1996 Heikki Tuuri
#include <deque>
struct vtq_record_t;
/** A stack of table names related through foreign key constraints */
typedef std::deque<const char*, ut_allocator<const char*> > dict_names_t;
@ -319,20 +317,6 @@ dict_process_sys_datafiles(
ulint* space, /*!< out: pace id */
const char** path); /*!< out: datafile path */
/** This function parses a SYS_VTQ record, extracts necessary
information from the record and returns it to the caller.
@param[in,out] heap Heap memory
@param[in] rec Current record
@param[out] fields Field values
@return error message, or NULL on success */
UNIV_INTERN
const char*
dict_process_sys_vtq(
mem_heap_t* heap, /*!< in/out: heap memory */
const rec_t* rec, /*!< in: current rec */
vtq_record_t& fields /*!< out: field values */
);
/** Update the record for space_id in SYS_TABLESPACES to this filepath.
@param[in] space_id Tablespace ID
@param[in] filepath Tablespace filepath
@ -356,9 +340,4 @@ dict_replace_tablespace_and_filepath(
const char* filepath,
ulint fsp_flags);
UNIV_INTERN
dict_table_t*
get_vtq_table();
#endif

2
storage/innobase/include/dict0mem.h
View file

@ -1489,7 +1489,7 @@ struct dict_table_t {
/** Add the table definition to the data dictionary cache */
void add_to_cache();
bool with_versioning() const { return vers_start || vers_end; }
bool versioned() const { return vers_start || vers_end; }
/** Id of the table. */
table_id_t id;

5
storage/innobase/include/row0ins.h
View file

@ -182,11 +182,6 @@ row_ins_step(
/*=========*/
que_thr_t* thr); /*!< in: query thread */
/***********************************************************//**
Inserts a row to SYS_VTQ table.
@return error state */
void vers_notify_vtq(trx_t* trx);
/* Insert node structure */
struct ins_node_t{

33
storage/innobase/include/trx0trx.h
View file

@ -883,33 +883,6 @@ struct TrxVersion {
ulint m_version;
};
/** Class which is used to query VTQ and also serves as a cache to VTQ of size 1
*/
class vtq_query_t
{
public:
/** VTQ used to translate timestamps to nearest trx_id and this is
a timestamp for a row we're caching now */
timeval prev_query;
/** We search for nearest trx_id on the left or on the right and
we search forwards or backwards */
bool backwards;
/** Cached row from VTQ */
vtq_record_t result;
/** Parses record and stores its value in a result field
but disables a cache */
const char * cache_result(mem_heap_t* heap, const rec_t* rec);
/** Parses record and stores its value in a result field and enables
cache (prev_query, backward) */
const char * cache_result(
mem_heap_t* heap,
const rec_t* rec,
const timeval &_ts_query,
bool _backwards);
};
typedef std::list<TrxVersion, ut_allocator<TrxVersion> > hit_list_t;
struct trx_t {
@ -1296,10 +1269,8 @@ struct trx_t {
#endif /* WITH_WSREP */
/* System Versioning */
bool vtq_notify_on_commit;
/*!< Notify VTQ for System Versioned update */
vtq_query_t vtq_query; /*!< Structure to query VTQ and store
one row result */
bool vers_update_trt;
/*!< Notify TRT on System Versioned write */
ulint magic_n;
/** @return whether any persistent undo log has been generated */

2
storage/innobase/include/trx0types.h
View file

@ -146,7 +146,7 @@ typedef ib_id_t undo_no_t;
/** Transaction savepoint */
struct trx_savept_t{
undo_no_t least_undo_no; /*!< least undo number to undo */
bool vtq_notify_on_commit; /*!< Notify VTQ for System Versioned update */
bool vers_update_trt; /*!< Notify TRT for System Versioned write */
};
/** File objects */

52
storage/innobase/include/vers0vtq.h
View file

@ -1,52 +0,0 @@
/* Copyright (c) 2017, 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 Street, Fifth Floor, Boston, MA 02110-1301, USA */
/** Query VTQ by TRX_ID.
@param[in] thd MySQL thread
@param[out] out field value or whole record returned by query (selected by `field`)
@param[in] in_trx_id query parameter TRX_ID
@param[in] field field to get in `out` or VTQ_ALL for whole record (vtq_record_t)
@return TRUE if record is found, FALSE otherwise */
bool
vtq_query_trx_id(THD* thd, void *out, ulonglong in_trx_id, vtq_field_t field);
/** Query VTQ by COMMIT_TS.
@param[in] thd MySQL thread
@param[out] out field value or whole record returned by query (selected by `field`)
@param[in] commit_ts query parameter COMMIT_TS
@param[in] field field to get in `out` or VTQ_ALL for whole record (vtq_record_t)
@param[in] backwards direction of VTQ search
@return TRUE if record is found, FALSE otherwise */
bool
vtq_query_commit_ts(THD* thd, void *out, const MYSQL_TIME &commit_ts, vtq_field_t field, bool backwards);
/** Check if transaction TX1 sees transaction TX0.
@param[in] thd MySQL thread
@param[out] result true if TX1 sees TX0
@param[in] trx_id1 TX1 TRX_ID
@param[in] trx_id0 TX0 TRX_ID
@param[in] commit_id1 TX1 COMMIT_ID
@param[in] iso_level1 TX1 isolation level
@param[in] commit_id0 TX0 COMMIT_ID
@return FALSE if there is no trx_id1 in VTQ, otherwise TRUE */
bool
vtq_trx_sees(
THD *thd,
bool &result,
ulonglong trx_id1,
ulonglong trx_id0,
ulonglong commit_id1,
uchar iso_level1,
ulonglong commit_id0);

95
storage/innobase/row/row0ins.cc
View file

@ -1705,7 +1705,7 @@ row_ins_check_foreign_constraint(
}
/* System Versioning: if sys_trx_end != Inf, we
suppress the foreign key check */
if (table->with_versioning() &&
if (table->versioned() &&
dfield_get_type(field)->prtype & DATA_VERS_END) {
byte* data = static_cast<byte*>(dfield_get_data(field));
ut_ad(data);
@ -1841,7 +1841,7 @@ row_ins_check_foreign_constraint(
cmp = cmp_dtuple_rec(entry, rec, offsets);
if (cmp == 0) {
if (check_table->with_versioning()) {
if (check_table->versioned()) {
trx_id_t end_trx_id = 0;
if (dict_index_is_clust(check_index)) {
@ -3987,94 +3987,3 @@ error_handling:
return(thr);
}
/***********************************************************//**
Inserts a row to SYS_VTQ, low level.
@return DB_SUCCESS if operation successfully completed, else error
code */
static __attribute__((nonnull, warn_unused_result))
dberr_t
vers_row_ins_vtq_low(trx_t* trx, mem_heap_t* heap, dtuple_t* tuple)
{
dberr_t err;
dtuple_t* entry;
ulint n_index = 0;
dict_index_t* index = dict_table_get_first_index(dict_sys->sys_vtq);
static const ulint flags
= (BTR_KEEP_SYS_FLAG
| BTR_NO_LOCKING_FLAG
| BTR_NO_UNDO_LOG_FLAG);
entry = row_build_index_entry(tuple, NULL, index, heap);
dfield_t* dfield = dtuple_get_nth_field(entry, DATA_TRX_ID);
ut_ad(dfield->type.len == DATA_TRX_ID_LEN);
dfield_set_data(dfield, mem_heap_alloc(heap, DATA_TRX_ID_LEN), DATA_TRX_ID_LEN);
row_upd_index_entry_sys_field(entry, index, DATA_TRX_ID, trx->id);
err = row_ins_clust_index_entry_low(
flags, BTR_MODIFY_TREE, index, index->n_uniq, entry, 0, NULL, false, trx);
switch (err) {
case DB_SUCCESS:
break;
case DB_SUCCESS_LOCKED_REC:
/* The row had already been copied to the table. */
ib::info() << "InnoDB: duplicate VTQ record!";
return DB_SUCCESS;
default:
return err;
}
mem_heap_t* offsets_heap = mem_heap_create(1024);
do {
if (!(index = dict_table_get_next_index(index))) {
break;
}
n_index++;
entry = row_build_index_entry(tuple, NULL, index, heap);
err = row_ins_sec_index_entry_low(
flags, BTR_MODIFY_TREE,
index, offsets_heap, heap, entry, trx->id, NULL, false, trx);
} while (err == DB_SUCCESS);
ut_ad(n_index == 3 || err != DB_SUCCESS);
mem_heap_free(offsets_heap);
return err;
}
/***********************************************************//**
Inserts a row to SYS_VTQ table.
@return error state */
void vers_notify_vtq(trx_t* trx)
{
dberr_t err;
mem_heap_t* heap = mem_heap_create(1024);
dtuple_t* tuple = dtuple_create(heap, dict_table_get_n_cols(dict_sys->sys_vtq));
timeval begin_ts, commit_ts;
begin_ts.tv_sec = static_cast<my_time_t>(trx->start_time);
begin_ts.tv_usec = trx->start_time_micro % 1000000;
mutex_enter(&trx_sys->mutex);
trx_id_t commit_id = trx_sys_get_new_trx_id();
ut_usectime((ulint *)&commit_ts.tv_sec, (ulint *)&commit_ts.tv_usec);
mutex_exit(&trx_sys->mutex);
dict_table_copy_types(tuple, dict_sys->sys_vtq);
row_ins_set_tuple_col_8(tuple, DICT_COL__SYS_VTQ__TRX_ID, trx->id, heap);
row_ins_set_tuple_col_8(tuple, DICT_COL__SYS_VTQ__COMMIT_ID, commit_id, heap);
row_ins_set_tuple_col_8(tuple, DICT_COL__SYS_VTQ__BEGIN_TS, begin_ts, heap);
row_ins_set_tuple_col_8(tuple, DICT_COL__SYS_VTQ__COMMIT_TS, commit_ts, heap);
ut_ad(trx->isolation_level < 256);
row_ins_set_tuple_col_1(tuple, DICT_COL__SYS_VTQ__ISOLATION_LEVEL, trx->isolation_level, heap);
err = vers_row_ins_vtq_low(trx, heap, tuple);
if (DB_SUCCESS != err)
ib::error()
<< "failed to insert VTQ record (error " << err << ")";
mem_heap_free(heap);
}

10
storage/innobase/row/row0merge.cc
View file

@ -2241,7 +2241,7 @@ end_of_index:
< dict_table_get_n_user_cols(new_table));
bool historical_row = false;
if (new_table->with_versioning()) {
if (new_table->versioned()) {
const dfield_t *dfield = dtuple_get_nth_field(
row, new_table->vers_end);
const byte *data = static_cast<const byte *>(
@ -2303,8 +2303,8 @@ end_of_index:
}
}
if (old_table->with_versioning()) {
if (new_table->with_versioning() && !drop_historical) {
if (old_table->versioned()) {
if (new_table->versioned() && !drop_historical) {
dfield_t *end = dtuple_get_nth_field(
row, new_table->vers_end);
byte *data = static_cast<byte *>(
@ -2316,6 +2316,7 @@ end_of_index:
void *data = dfield_get_data(start);
ut_ad(data);
mach_write_to_8(data, trx->id);
trx->vers_update_trt= true;
}
} else {
const dict_col_t *col =
@ -2330,7 +2331,7 @@ end_of_index:
if (mach_read_from_8(sys_trx_end) != TRX_ID_MAX)
continue;
}
} else if (new_table->with_versioning()) {
} else if (new_table->versioned()) {
void *sys_trx_start = mem_heap_alloc(row_heap, 8);
void *sys_trx_end = mem_heap_alloc(row_heap, 8);
mach_write_to_8(sys_trx_start, trx->id);
@ -2341,6 +2342,7 @@ end_of_index:
row, new_table->vers_end);
dfield_set_data(start, sys_trx_start, 8);
dfield_set_data(end, sys_trx_end, 8);
trx->vers_update_trt= true;
}
write_buffers:

10
storage/innobase/row/row0mysql.cc
View file

@ -1569,8 +1569,8 @@ error_exit:
node->duplicate = NULL;
if (node->table->with_versioning()) {
trx->vtq_notify_on_commit = true;
if (node->table->versioned() && ins_mode != ROW_INS_NORMAL) {
trx->vers_update_trt = true;
}
if (dict_table_has_fts_index(table)) {
@ -2129,7 +2129,7 @@ run_again:
node->cascade_upd_nodes = cascade_upd_nodes;
cascade_upd_nodes->pop_front();
thr->fk_cascade_depth++;
vers_set_fields = node->table->with_versioning() &&
vers_set_fields = node->table->versioned() &&
(node->is_delete || node->versioned);
goto run_again;
@ -2210,12 +2210,12 @@ run_again:
prebuilt->table->stat_modified_counter++;
}
if (node->table->with_versioning() &&
if (node->table->versioned() &&
(node->versioned || node->vers_delete ||
// TODO: imrove this check (check if we touch only
// unversioned fields in foreigh table)
node->foreign)) {
trx->vtq_notify_on_commit = true;
trx->vers_update_trt = true;
}
trx->op_info = "";

4
storage/innobase/srv/srv0start.cc
View file

@ -2588,10 +2588,6 @@ files_checked:
err = dict_create_or_check_sys_tablespace();
if (err == DB_SUCCESS) {
err = dict_create_or_check_sys_virtual();
if (err == DB_SUCCESS) {
/* Create the SYS_VTQ system table */
err = dict_create_or_check_vtq_table();
}
}
}
switch (err) {

4
storage/innobase/trx/trx0roll.cc
View file

@ -128,7 +128,7 @@ trx_rollback_to_savepoint_low(
} else {
trx->lock.que_state = TRX_QUE_RUNNING;
MONITOR_INC(MONITOR_TRX_ROLLBACK_SAVEPOINT);
trx->vtq_notify_on_commit = savept->vtq_notify_on_commit;
trx->vers_update_trt = savept->vers_update_trt;
}
ut_a(trx->error_state == DB_SUCCESS);
@ -618,7 +618,7 @@ trx_savept_take(
trx_savept_t savept;
savept.least_undo_no = trx->undo_no;
savept.vtq_notify_on_commit = trx->vtq_notify_on_commit;
savept.vers_update_trt = trx->vers_update_trt;
return(savept);
}

2
storage/innobase/trx/trx0trx.cc
View file

@ -145,7 +145,7 @@ trx_init(
trx->check_unique_secondary = true;
trx->vtq_notify_on_commit = false;
trx->vers_update_trt = false;
trx->lock.n_rec_locks = 0;

479
storage/innobase/vers/vers0vtq.cc
View file

@ -1,479 +0,0 @@
/* Copyright (c) 2017, 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 Street, Fifth Floor, Boston, MA 02110-1301, USA */
#include <sql_class.h>
#include <tztime.h>
#include "btr0pcur.h"
#include "dict0load.h"
#include "ha_innodb.h"
#include "row0ins.h"
#include "row0row.h"
#include "trx0trx.h"
#include "trx0types.h"
/** Field or record selector.
@param[in] thd MySQL thread
@param[in] q VTQ record to get values from
@param[out] out field value or whole record returned
@param[in] field field to get in `out` or VTQ_ALL for whole record (vtq_record_t copied) */
static
inline
void
vtq_result(THD* thd, vtq_record_t& q, void *out, vtq_field_t field)
{
ut_ad(field == VTQ_ALL || out);
switch (field) {
case VTQ_ALL:
if (out) {
*reinterpret_cast<vtq_record_t *>(out) = q;
}
break;
case VTQ_TRX_ID:
*reinterpret_cast<trx_id_t *>(out) = q.trx_id;
break;
case VTQ_COMMIT_ID:
*reinterpret_cast<trx_id_t *>(out) = q.commit_id;
break;
case VTQ_BEGIN_TS: {
MYSQL_TIME* out_ts = reinterpret_cast<MYSQL_TIME *>(out);
thd_get_timezone(thd)->gmt_sec_to_TIME(out_ts, q.begin_ts.tv_sec);
out_ts->second_part = q.begin_ts.tv_usec;
break;
}
case VTQ_COMMIT_TS: {
MYSQL_TIME* out_ts = reinterpret_cast<MYSQL_TIME *>(out);
thd_get_timezone(thd)->gmt_sec_to_TIME(out_ts, q.commit_ts.tv_sec);
out_ts->second_part = q.commit_ts.tv_usec;
break;
}
case VTQ_ISO_LEVEL:
*reinterpret_cast<uint *>(out) = q.iso_level;
break;
default:
ut_error;
}
}
inline
const char *
vtq_query_t::cache_result(mem_heap_t* heap, const rec_t* rec)
{
prev_query.tv_sec = 0;
return dict_process_sys_vtq(heap, rec, result);
}
/** Query VTQ by TRX_ID.
@param[in] thd MySQL thread
@param[out] out field value or whole record returned by query (selected by `field`)
@param[in] in_trx_id query parameter TRX_ID
@param[in] field field to get in `out` or VTQ_ALL for whole record (vtq_record_t)
@return TRUE if record is found, FALSE otherwise */
UNIV_INTERN
bool
vtq_query_trx_id(THD* thd, void *out, ulonglong _in_trx_id, vtq_field_t field)
{
trx_t* trx;
dict_index_t* index;
btr_pcur_t pcur;
dtuple_t* tuple;
dfield_t* dfield;
trx_id_t trx_id_net;
mtr_t mtr;
mem_heap_t* heap;
rec_t* rec;
bool found = false;
DBUG_ENTER("vtq_query_trx_id");
if (_in_trx_id == 0) {
DBUG_RETURN(false);
}
ut_ad(sizeof(_in_trx_id) == sizeof(trx_id_t));
trx_id_t in_trx_id = static_cast<trx_id_t>(_in_trx_id);
trx = thd_to_trx(thd);
ut_a(trx);
vtq_record_t &cached = trx->vtq_query.result;
if (cached.trx_id == in_trx_id) {
vtq_result(thd, cached, out, field);
DBUG_RETURN(true);
}
index = dict_table_get_first_index(dict_sys->sys_vtq);
heap = mem_heap_create(0);
ut_ad(index);
ut_ad(dict_index_is_clust(index));
mach_write_to_8(
reinterpret_cast<byte*>(&trx_id_net),
in_trx_id);
tuple = dtuple_create(heap, 1);
dfield = dtuple_get_nth_field(tuple, DICT_FLD__SYS_VTQ__TRX_ID);
dfield_set_data(dfield, &trx_id_net, 8);
dict_index_copy_types(tuple, index, 1);
mtr_start_trx(&mtr, trx);
btr_pcur_open_on_user_rec(index, tuple, PAGE_CUR_GE,
BTR_SEARCH_LEAF, &pcur, &mtr);
if (!btr_pcur_is_on_user_rec(&pcur))
goto not_found;
rec = btr_pcur_get_rec(&pcur);
{
const char *err = trx->vtq_query.cache_result(heap, rec);
if (err) {
ib::error()
<< "vtq_query_trx_id: get VTQ field failed: "
<< err;
ut_ad(false && "get VTQ field failed");
goto not_found;
}
}
if (cached.trx_id != in_trx_id)
goto not_found;
vtq_result(thd, cached, out, field);
found = true;
not_found:
btr_pcur_close(&pcur);
mtr_commit(&mtr);
mem_heap_free(heap);
DBUG_RETURN(found);
}
static
inline
void rec_get_timeval(const rec_t* rec, ulint nfield, timeval& out)
{
ulint len;
const byte* field;
field = rec_get_nth_field_old(
rec, nfield, &len);
ut_ad(len == sizeof(uint64_t));
out.tv_sec = mach_read_from_4(field);
out.tv_usec = mach_read_from_4(field + 4);
}
inline
const char *
vtq_query_t::cache_result(
mem_heap_t* heap,
const rec_t* rec,
const timeval& _ts_query,
bool _backwards)
{
prev_query = _ts_query;
backwards = _backwards;
return dict_process_sys_vtq(heap, rec, result);
}
static
inline
bool
operator== (const timeval &a, const timeval &b)
{
return a.tv_sec == b.tv_sec && a.tv_usec == b.tv_usec;
}
static
inline
bool
operator!= (const timeval &a, const timeval &b)
{
return !(a == b);
}
static
inline
bool
operator> (const timeval &a, const timeval &b)
{
return a.tv_sec > b.tv_sec || (a.tv_sec == b.tv_sec && a.tv_usec > b.tv_usec);
}
static
inline
bool
operator< (const timeval &a, const timeval &b)
{
return b > a;
}
static
trx_id_t
read_trx_id(const rec_t *rec) {
ulint len = 0;
const rec_t *field = rec_get_nth_field_old(rec, 1, &len);
DBUG_ASSERT(len == 8);
return mach_read_from_8(field);
}
/** Find a row with given commit_ts but MAX()/MIN() trx_id
@param[in] mtr mini-transaction handler
@param[in, out] pcur btree cursor which may be changed by this function
@param[in] backwards search direction
@param[in] commit_ts target timestamp for records
@param[in] rec record buffer for pcur
*/
static
void
find_best_match(
mtr_t &mtr,
btr_pcur_t &pcur,
bool backwards,
timeval commit_ts,
const rec_t *rec)
{
btr_pcur_t best;
btr_pcur_init(&best);
btr_pcur_copy_stored_position(&best, &pcur);
trx_id_t best_trx_id = read_trx_id(rec);
while (true) {
if (backwards ? !btr_pcur_move_to_prev_user_rec(&pcur, &mtr)
: !btr_pcur_move_to_next_user_rec(&pcur, &mtr))
break;
timeval tv;
rec = btr_pcur_get_rec(&pcur);
rec_get_timeval(rec, 0, tv);
if (tv != commit_ts)
break;
trx_id_t trx_id = read_trx_id(rec);
if (backwards ? trx_id < best_trx_id : trx_id > best_trx_id) {
best_trx_id = trx_id;
btr_pcur_copy_stored_position(&best, &pcur);
}
}
btr_pcur_copy_stored_position(&pcur, &best);
btr_pcur_free(&best);
}
/** Query VTQ by COMMIT_TS.
@param[in] thd MySQL thread
@param[out] out field value or whole record returned by query (selected by `field`)
@param[in] commit_ts query parameter COMMIT_TS
@param[in] field field to get in `out` or VTQ_ALL for whole record (vtq_record_t)
@param[in] backwards direction of VTQ search
@return TRUE if record is found, FALSE otherwise */
UNIV_INTERN
bool
vtq_query_commit_ts(
THD* thd,
void *out,
const MYSQL_TIME &_commit_ts,
vtq_field_t field,
bool backwards)
{
trx_t* trx;
btr_pcur_t pcur;
dtuple_t* tuple;
page_cur_mode_t mode;
mtr_t mtr;
mem_heap_t* heap;
uint err;
timeval commit_ts;
timeval rec_ts = { 0, 0 };
const rec_t *rec, *clust_rec;
dict_index_t* index = dict_sys->vtq_commit_ts_ind;
dict_index_t* clust_index;
bool found = false;
DBUG_ENTER("vtq_query_commit_ts");
mode = backwards ? PAGE_CUR_LE : PAGE_CUR_GE;
trx = thd_to_trx(thd);
ut_a(trx);
vtq_record_t &cached = trx->vtq_query.result;
timeval &prev_query = trx->vtq_query.prev_query;
bool prev_bwds = trx->vtq_query.backwards;
commit_ts.tv_usec = _commit_ts.second_part;
commit_ts.tv_sec = thd_get_timezone(thd)->TIME_to_gmt_sec(&_commit_ts, &err);
if (err) {
if (err == ER_WARN_DATA_OUT_OF_RANGE) {
if (_commit_ts.year <= TIMESTAMP_MIN_YEAR) {
commit_ts.tv_usec = 0;
commit_ts.tv_sec = 1;
} else {
ut_ad(_commit_ts.year >= TIMESTAMP_MAX_YEAR);
commit_ts.tv_usec = TIME_MAX_SECOND_PART;
commit_ts.tv_sec = MY_TIME_T_MAX;
}
} else {
DBUG_RETURN(false);
}
} else if (cached.commit_ts == commit_ts ||
(prev_query.tv_sec && prev_bwds == backwards && (
(!backwards && (commit_ts < prev_query) && commit_ts > cached.commit_ts) ||
(backwards && (commit_ts > prev_query) && commit_ts < cached.commit_ts))))
{
vtq_result(thd, cached, out, field);
DBUG_RETURN(true);
}
heap = mem_heap_create(0);
tuple = dtuple_create(heap, 1);
dict_index_copy_types(tuple, index, 1);
dtuple_get_nth_field(tuple, 0)->len = UNIV_SQL_NULL;
row_ins_set_tuple_col_8(tuple, 0, commit_ts, heap);
mtr_start_trx(&mtr, trx);
btr_pcur_open_on_user_rec(index, tuple, mode,
BTR_SEARCH_LEAF, &pcur, &mtr);
if (btr_pcur_is_on_user_rec(&pcur)) {
rec = btr_pcur_get_rec(&pcur);
rec_get_timeval(rec, 0, rec_ts);
if (rec_ts == commit_ts) {
find_best_match(mtr, pcur, backwards, commit_ts, rec);
goto found;
}
} else {
rec_ts = commit_ts;
}
if (mode == PAGE_CUR_GE) {
btr_pcur_move_to_prev_user_rec(&pcur, &mtr);
} else {
btr_pcur_move_to_next_user_rec(&pcur, &mtr);
}
if (!btr_pcur_is_on_user_rec(&pcur))
goto not_found;
rec = btr_pcur_get_rec(&pcur);
found:
clust_rec = row_get_clust_rec(BTR_SEARCH_LEAF, rec, index, &clust_index, &mtr);
if (!clust_rec) {
ib::error() << "vtq_query_commit_ts: secondary index is out of "
"sync";
ut_ad(false && "secondary index is out of sync");
goto not_found;
}
{
const char *err =
trx->vtq_query.cache_result(
heap,
clust_rec,
rec_ts,
backwards);
if (err) {
ib::error()
<< "vtq_query_commit_ts: get VTQ field failed: "
<< err;
ut_ad(false && "get VTQ field failed");
goto not_found;
}
}
vtq_result(thd, cached, out, field);
found = true;
not_found:
btr_pcur_close(&pcur);
mtr_commit(&mtr);
mem_heap_free(heap);
DBUG_RETURN(found);
}
/** Check if transaction TX1 sees transaction TX0.
@param[in] thd MySQL thread
@param[out] result true if TX1 sees TX0
@param[in] trx_id1 TX1 TRX_ID
@param[in] trx_id0 TX0 TRX_ID
@param[in] commit_id1 TX1 COMMIT_ID
@param[in] iso_level1 TX1 isolation level
@param[in] commit_id0 TX0 COMMIT_ID
@return FALSE if there is no trx_id1 in VTQ, otherwise TRUE */
bool
vtq_trx_sees(
THD *thd,
bool &result,
ulonglong trx_id1,
ulonglong trx_id0,
ulonglong commit_id1,
uchar iso_level1,
ulonglong commit_id0)
{
DBUG_ENTER("vtq_trx_sees");
if (trx_id1 == trx_id0) {
result = false;
DBUG_RETURN(true);
}
if (trx_id1 == ULONGLONG_MAX || trx_id0 == 0) {
result = true;
DBUG_RETURN(true);
}
if (!commit_id1) {
if (!vtq_query_trx_id(thd, NULL, trx_id1, VTQ_ALL)) {
ib::info() << "vtq_trx_sees: can't find COMMIT_ID0 by "
"TRX_ID: "
<< trx_id1;
DBUG_RETURN(false);
}
trx_t* trx = thd_to_trx(thd);
ut_ad(trx);
commit_id1 = trx->vtq_query.result.commit_id;
iso_level1 = trx->vtq_query.result.iso_level;
}
if (!commit_id0) {
if (!vtq_query_trx_id(thd, &commit_id0, trx_id0, VTQ_COMMIT_ID)) {
ib::info() << "vtq_trx_sees: can't find COMMIT_ID1 by "
"TRX_ID: "
<< trx_id0;
DBUG_RETURN(false);
}
}
// Trivial case: TX1 started after TX0 committed
if (trx_id1 > commit_id0
// Concurrent transactions: TX1 committed after TX0 and TX1 is read (un)committed
|| (commit_id1 > commit_id0 && iso_level1 < TRX_ISO_REPEATABLE_READ))
{
result = true;
} else {
// All other cases: TX1 does not see TX0
result = false;
}
DBUG_RETURN(true);
}