From: Date: August 3 2008 12:13pm Subject: bzr commit into mysql-6.1-fk branch (dlenev:2680) WL#148 List-Archive: http://lists.mysql.com/commits/50845 Message-Id: <20080803101312.4F8952040D9@mockturtle.local> MIME-Version: 1.0 Content-Type: text/plain; charset="us-ascii" Content-Transfer-Encoding: 7bit #At file:///home/dlenev/src/bzr/mysql-6.1-mil4-2/ 2680 Dmitry Lenev 2008-08-03 Patch for the 4th milestone of WL#148 "Foreign keys" ("Implement meta-data locking for foreign key definitions and make prelocking algorithm aware of foreign keys"). Extended algortihm which calculates list of prelocked tables for the statement to take into account foreign keys. (Basically after opening each table we analyze foreign keys in which this table participates and depending on operation which is going to be carried on this table add child or parent tables of foreign key to the statement table list to be opened and locked with appropriate lock type. If we are going to INSERT into or UPDATE this table then for all foreign keys for which this table serves as child we add parent tables with request for read lock. If we are going to UPDATE or DELETE from this table we add child tables of foreign keys that reference this table and determine type of lock depending on type ON UPDATE/ON DELETE option for this foreign key. We avoid processing (and thus adding tables) for the same foreign key in the same role more than once by keeping set of already processed pairs.) As a pre-requisite changed code responsible for table creation to store information about foreign keys on table being created also in .FRMs of parent tables (with appropriate meta-data locking). modified: mysql-test/r/create.result mysql-test/r/foreign_key_acceptance.result mysql-test/r/foreign_key_all_engines_2.result mysql-test/r/lock_multi.result mysql-test/t/create.test mysql-test/t/foreign_key_acceptance.test mysql-test/t/foreign_key_all_engines_2.test mysql-test/t/lock_multi.test sql/fk_dd.cc sql/fk_dd.h sql/item.cc sql/item.h sql/mysql_priv.h sql/sp.cc sql/sp.h sql/sp_head.h sql/sql_base.cc sql/sql_class.cc sql/sql_class.h sql/sql_lex.cc sql/sql_lex.h sql/sql_list.h sql/sql_parse.cc sql/sql_show.cc sql/sql_table.cc sql/sql_trigger.cc sql/sql_trigger.h sql/sql_yacc.yy sql/table.cc sql/table.h sql/unireg.cc per-file messages: mysql-test/r/create.result Added coverage for CREATE TABLE behavior under LOCK TABLES. mysql-test/r/foreign_key_acceptance.result Adjusted test case for one of earlier milestones to avoid error about missing parent table which started to be emitted on this milestone. mysql-test/r/foreign_key_all_engines_2.result Added basic coverage for 4th milestone "Implement meta-data locking for foreign key definitions and make prelocking algorithm aware of foreign keys". mysql-test/r/lock_multi.result Adjusted test case to the fact that we no longer allow create non-temporary tables under LOCK TABLES. mysql-test/t/create.test Added coverage for CREATE TABLE behavior under LOCK TABLES. mysql-test/t/foreign_key_acceptance.test Adjusted test case for one of earlier milestones to avoid error about missing parent table which started to be emitted on this milestone. mysql-test/t/foreign_key_all_engines_2.test Added basic coverage for 4th milestone "Implement meta-data locking for foreign key definitions and make prelocking algorithm aware of foreign keys". mysql-test/t/lock_multi.test Adjusted test case to the fact that we no longer allow create non-temporary tables under LOCK TABLES. sql/fk_dd.cc Now we save information about foreign keys also in .FRMs of parent tables. In order to do this we have introduced Foreign_key_parent class which represents description of foreign key for its parent table. Also now we use Foreign_key_child_share/parent_share classes instead of Foreign_key class to represent foreign keys in shares of their child and parent tables. Therefore code responsible for saving/ restoring information about foreign keys in/from .FRM files was reorganized. Code common for both child and parent cases was left in save_to_frm() and restore_from_frm() methods while code which handles specifics was moved into virtual methods. Implemented Foreign_key_child/parent_share::make_foreig_key() methods which allows to get Foreign_key/Foreign_key_parent objects from foreign key description in the table share, to be used in ALTER TABLE implementation. Introduced fk_init_column_sets() function for initialization of set of used columns in Foreign_key_share objects for particular table share. Added fk_add_tables_for_fks() function to be used by prelocking algorithm for adding tables required for checks/actions for foreign keys which are associated with particular table (also added auxiliary functions used by above function). sql/fk_dd.h fk_get_frm_section_length() and fk_save_to_frm_section() now take list of objects describing foreign keys for which this table serves as parent and information about which should be saved into .FRM. fk_check_constraint_added() no longer needs "db" argument since the same information now can be obtained from Foreign_key object. Introduced fk_init_column_sets() function for initialization of sets of used columns in Foreign_key_share objects for particular table share. Added fk_add_tables_for_fks() function to be used by prelocking algorithm for adding tables required for checks/actions for foreign keys which are associated with particular table. sql/item.cc Now Item_trigger_field::setup_field() is responsible for marking found column as used in the sets of used columns which are passed to it as parameter. Also simplified its code by using new find_field_in_table_def() function. sql/item.h Now Item_trigger_field::setup_field() is responsible for marking found column as used in the sets of used columns which are passed to it as parameter. sql/mysql_priv.h Introduced new flag for st_select_lex::add_table_to_list() method which tells this method not to add table to table list if it is already in the table list and return pointer to the existing table list element instead. mysql_create_table() now takes table list element for the table being created and list of tables referenced by table's foreign keys as its arguments. Introduced find_field_in_table_def() function for looking up field by its name in the table's share. rea_create_table() and mysql_create_frm() now take list of foreign keys for which table being created is parent as one of parameters. sql/sp.cc Renamed add_used_routine() to sp_add_used_routine() and made it available outside of sp.cc. sql/sp.h Renamed add_used_routine() to sp_add_used_routine() and made it available outside of sp.cc. sql/sp_head.h Introduced TYPE_ENUM_FOREIGN_KEY type of "routine" so prelocking algorithm can distinguish foreign keys that it has processed from other kinds of objects. sql/sql_base.cc open_tables(): Extended algorithm which calculates list of prelocked tables to handle foreign keys. find_field_in_table_def(): Introduced new function for looking up field by its name in the table's share. sql/sql_class.cc Introduced Foreign_key_parent class which represents information about foreign key to be saved in .FRM of its parent table (and also serves as ancestor for Foreign_key class). sql/sql_class.h Moved fk_option and fk_match_opt enums out of Foreign_key class to simplify its usage outside of this class. Introduced Foreign_key_parent class which represents information about foreign key to be saved in .FRM of its parent table (and also serves as ancestor for Foreign_key class). Reorganized code responsible for saving information about foreign keys in .FRM by keeping common code in Foreign_key_parent::save_to_frm() method and introducing virtual methods which handle differences between Foreign_key_parent and Foreign_key classes. Added Foreign_key_share/child_share/parent_share classes which represent foreign keys in shares of its child and parent tables. Moved code that restores description of foreign key from .FRM to methods of this class organized in similar fashion to methods of Foreign_key_parent/Foreign_key classes responsible for saving data in .FRM. sql/sql_lex.cc Alter_info: In order to be able to save information about foreign keys that reference this table in its .FRM added member for storing list of objects representing foreign keys for which table being created is parent. LEX: Added create_foreign_key_tables member for storing list of tables referenced by foreign keys in the table being created. Changed unlink_all_tables_except_first() method to properly set TABLE_LIST::prev_global pointer for tables being unlinked. sql/sql_lex.h Alter_info: In order to be able to save information about foreign keys that reference this table in its .FRM added member for storing list of objects representing foreign keys for which table being created is parent. LEX: fk_match_opt and fk_option enums were moved out of Foreign_key class to simplify their usage outside of it. Added create_foreign_key_tables member for storing list of tables referenced by foreign keys in the table being created. Changed unlink_all_tables_except_first() method to properly set TABLE_LIST::prev_global pointer for tables being unlinked. sql/sql_list.h Fixed List::swap() method to properly handle empty lists. sql/sql_parse.cc mysql_create_table() now takes table list element for the table being created and list of tables referenced by table's foreign keys as its arguments. st_select_lex::add_table_to_list() method now has new flag (to be passed in table_options argument) which tells this method not to add table to table list if it is already in the table list and return pointer to the existing table list element instead. sql/sql_show.cc Adjusted code after introducing Foreign_key_child_share - new class that represents foreign key in the share of table which participates in this foreign key as child. sql/sql_table.cc mysql_create_table(): Now when we create table with foreign key constraints in --foreign-key-all-engines mode we also add information about these foreign keys to .FRM of parent tables. Indeed to do this safely we have to acquire exclusive metadata locks on these tables. Since according to our metadata protocol acquiring exclusive lock while holding other metadata locks can lead to deadlock we can't allow creation of table with foreign keys under LOCK TABLES. Since creation of non-temporary table under LOCK TABLES does not makes much sense anyway we simply prohibit it in both --foreign-key-all-engines=0|1 modes. This also allowed to merge metadata locking code for both these cases. To implement changing .FRMs of parent tables added several auxiliary functions. mysql_prepare_alter_table(): Take table list element for the source table instead of TABLE object for it as a parameter. Since now we save information about foreign keys also in .FRMs of parent tables we have two lists with objects describing foreign keys in Alter_info structure - one for foreign keys in which this table is child and another for those in which this table is parent. We have similar arrangement for lists of objects describing foreign keys in table's share. Changed code accordingly. We no longer need to pass information about database where we are going to create our table to mysql_prepare_create_table() and other similar function since now this information, which is needed for proper handling of foreign keys, is available directly from Foreign_key objects. sql/sql_trigger.cc Changed approach to calculating set of table columns used by trigger on this table. Instead of calculating it each time when Table_triggers_list::mark_fields_used() is called we pre-calculate these sets for each kind of trigger event during loading trigger and then use them in mark_fields_used(). In order to do this added Table_triggers_list::read/write_set members and changed prepare_record1_accessors() method to prepare_for_trigger_fields(). As nice side-effect of this change now we can use write_set for UPDATE triggers in optimization in part of prelocking algorithm which handles foreign keys. Also got rid of unused Table_triggers_list::set_table() method. sql/sql_trigger.h Changed approach to calculating set of table columns used by trigger on this table. Instead of calculating it each time when Table_triggers_list::mark_fields_used() is called we pre-calculate these sets for each kind of trigger event during loading trigger and then use them in mark_fields_used(). In order to do this added Table_triggers_list::read/write_set members and changed prepare_record1_accessors() method to prepare_for_trigger_fields(). As nice side-effect of this change now we can use write_set for UPDATE triggers in optimization in part of prelocking algorithm which handles foreign keys. Also got rid of unused Table_triggers_list::set_table() method. sql/sql_yacc.yy Moved fk_option and fk_match_opt enums out of Foreign_key class to simplify its usage in other classes. Store list of tables referenced by foreign keys in the table being created in LEX::create_foreign_key_tables member. When adding parent tables for a foreign key to the table list avoid adding same table several times. Changed places where we use Foreign_key constructor to pass table list element for child table and appropriate values for its other new parameters. Finally now we call Foreign_key::set_missing_options() from the parser in order to be able easily create corresponding Foreign_key_parent objects early in the process of table creation. sql/table.cc Adjusted code to the fact that now in TABLE_SHARE we have two lists for foreign keys in which this table participates (one for those in which this table is child and one for those in which this table is parent). Also now at the final stage of loading information from .FRM into TABLE_SHARE we initialize sets of used columns in Foreign_key_share objects for this share. sql/table.h TABLE_SHARE: Now we have two separate lists for objects representing foreign keys in which this table particates -- one for those in which this table is child and another for those in which this table is parent. We also use different classes for representing such foreign keys (Foreign_key_child_share and Foreign_key_parent_share). TABLE_LIST: Added "cols_to_be_updated" member for storing pointer to the list of columns to be affected by update on the table in cases when such list is known. sql/unireg.cc rea_create_table()/mysql_create_frm(): Since now we also save information about foreign keys in .FRMs of parent tables, for each table we create we have to accept list of foreign keys in which this table participates as child as argument and pass it to fk_get_frm_section_length() and fk_save_to_frm_section(). === modified file 'mysql-test/r/create.result' --- a/mysql-test/r/create.result 2008-05-23 13:54:03 +0000 +++ b/mysql-test/r/create.result 2008-08-03 10:13:01 +0000 @@ -99,6 +99,14 @@ create table t1 (`` int); ERROR 42000: Incorrect column name '' create table t1 (i int, index `` (i)); ERROR 42000: Incorrect index name '' +create table t1 (i int); +lock tables t1 read; +create table t2 (j int); +ERROR HY000: Can't execute the given command because you have active locked tables or an active transaction +create temporary table t2 (j int); +drop temporary table t2; +unlock tables; +drop table t1; create table t1 (a int auto_increment not null primary key, B CHAR(20)); insert into t1 (b) values ("hello"),("my"),("world"); create table t2 (key (b)) select * from t1; === modified file 'mysql-test/r/foreign_key_acceptance.result' --- a/mysql-test/r/foreign_key_acceptance.result 2008-07-03 05:41:57 +0000 +++ b/mysql-test/r/foreign_key_acceptance.result 2008-08-03 10:13:01 +0000 @@ -16,6 +16,7 @@ Warnings: Warning 1674 Foreign key warning: Constraint 'y': Syntax 'FOREIGN KEY [id]' is old style, changing to CONSTRAINT [id] DROP TABLE IF EXISTS t4,t3,t2,t1; DROP TABLE IF EXISTS t2,t1; +CREATE TABLE t1 (s1 INT PRIMARY KEY); CREATE TABLE t2 ( s1 INT, CONSTRAINT `PRIMARY` FOREIGN KEY (s1) REFERENCES t1 (s1)); === modified file 'mysql-test/r/foreign_key_all_engines_2.result' --- a/mysql-test/r/foreign_key_all_engines_2.result 2008-06-23 07:14:00 +0000 +++ b/mysql-test/r/foreign_key_all_engines_2.result 2008-08-03 10:13:01 +0000 @@ -1,3 +1,93 @@ +drop tables if exists t1, t2, t3; +create table t1 (pk int primary key) engine=myisam; +insert into t1 values (1), (2), (3); +lock tables t1 read; +create table t2 (fk int references t1 (pk)) engine=myisam; +unlock tables; +lock tables t1 write; +insert into t2 values (1); +unlock tables; +lock table t1 read; +insert into t2 values (2); +unlock tables; +lock table t2 write; +update t1 set pk= 4 where pk=3; +unlock tables; +lock table t2 read; +update t1 set pk= 5 where pk=4; +unlock tables; +drop tables t1, t2; +create table t1 (pk int primary key) engine=myisam; +insert into t1 values (1), (2), (3); +create table t2 (fk int references t1 (pk) on delete cascade) engine=myisam; +lock tables t2 read; +delete from t1 where pk=3; +unlock tables; +drop tables t1, t2; +create table t1 (pk int primary key) engine=myisam; +insert into t1 values (1); +create table t2 (pk int primary key) engine=myisam; +insert into t2 values (1); +create table t3 (fk1 int references t1 (pk) on delete set null, +fk2 int references t2 (pk)) engine=myisam; +insert into t3 values (1, 1); +lock table t1 write; +insert into t2 values (2); +unlock tables; +drop tables t1, t2, t3; +create table t1 (pk int primary key) engine=myisam; +insert into t1 values (1); +create table t2 (pk1 int, pk2 int, primary key(pk1, pk2)) engine=myisam; +insert into t2 values (1, 1); +create table t3 (fk1 int references t1 (pk) on delete set null, +fk2 int, +foreign key (fk1, fk2) references t2 (pk1, pk2)) engine=myisam; +insert into t3 values (1, 1); +lock table t1 write; +insert into t2 values (2, 2); +unlock tables; +drop tables t1, t2, t3; +create table t1 (pk int primary key) engine=myisam; +insert into t1 values (1); +create table t2 (pk int primary key, +fk int references t1 (pk) on delete set null) engine=myisam; +insert into t2 values (1, 1); +create table t3 (fk int references t2 (pk) on delete cascade) engine=myisam; +insert into t3 values (1); +lock table t3 read; +delete from t1 where pk= 1; +unlock tables; +drop tables t1, t2, t3; +create table t1 (pk int primary key) engine=myisam; +insert into t1 values (1); +create table t2 (pk1 int, +pk2_fk int references t1 (pk) on update cascade) engine=myisam; +insert into t2 values (1, 1); +create table t3 (fk1 int, fk2 int, +foreign key (fk1, fk2) references t2 (pk1, pk2_fk) +on update set null) engine=myisam; +insert into t3 values (1, 1); +lock table t3 read; +update t1 set pk= 2; +unlock tables; +drop tables t1, t2, t3; +create table t1 (pk int primary key) engine=myisam; +insert into t1 values (1), (2); +create table t2 (pk int primary key, +fk int references t1 (pk) on delete set null) engine=myisam; +insert into t2 values (1, 1), (2, 2); +create table t3 (fk int references t2 (pk) on update set null) engine=myisam; +insert into t3 values (1), (2); +create trigger t2_bu before update on t2 for each row set @a:= 1; +lock table t3 read; +delete from t1 where pk=1; +unlock tables; +drop trigger t2_bu; +create trigger t2_bu before update on t2 for each row set new.pk=3; +lock table t3 read; +delete from t1 where pk=2; +unlock tables; +drop tables t1, t2, t3; drop tables if exists t1, t2; create table t1 (s1 int primary key) engine=innodb; create table t2 (s1 int, foreign key (s1) references t1 (s1)) engine=innodb; === modified file 'mysql-test/r/lock_multi.result' --- a/mysql-test/r/lock_multi.result 2008-02-19 14:09:52 +0000 +++ b/mysql-test/r/lock_multi.result 2008-08-03 10:13:01 +0000 @@ -72,9 +72,10 @@ CREATE TABLE t1 (c1 int); LOCK TABLE t1 WRITE; FLUSH TABLES WITH READ LOCK; CREATE TABLE t2 (c1 int); +ERROR HY000: Can't execute the given command because you have active locked tables or an active transaction UNLOCK TABLES; UNLOCK TABLES; -DROP TABLE t1, t2; +DROP TABLE t1; CREATE TABLE t1 (c1 int); LOCK TABLE t1 WRITE; FLUSH TABLES WITH READ LOCK; === modified file 'mysql-test/t/create.test' --- a/mysql-test/t/create.test 2008-05-23 13:54:03 +0000 +++ b/mysql-test/t/create.test 2008-08-03 10:13:01 +0000 @@ -101,6 +101,24 @@ create table t1 (`` int); --error 1280 create table t1 (i int, index `` (i)); + +# +# CREATE TABLE under LOCK TABLES +# +# We don't allow creation of non-temporary tables under LOCK TABLES +# as following meta-data locking protocol in this case can lead to +# deadlock. +create table t1 (i int); +lock tables t1 read; +--error ER_LOCK_OR_ACTIVE_TRANSACTION +create table t2 (j int); +# OTOH creating of temporary table should be OK +create temporary table t2 (j int); +drop temporary table t2; +unlock tables; +drop table t1; + + # # Test of CREATE ... SELECT with indexes # === modified file 'mysql-test/t/foreign_key_acceptance.test' --- a/mysql-test/t/foreign_key_acceptance.test 2008-07-03 05:41:57 +0000 +++ b/mysql-test/t/foreign_key_acceptance.test 2008-08-03 10:13:01 +0000 @@ -47,6 +47,7 @@ DROP TABLE IF EXISTS t4,t3,t2,t1; --disable_warnings DROP TABLE IF EXISTS t2,t1; --enable_warnings +CREATE TABLE t1 (s1 INT PRIMARY KEY); --error ER_FK_CONSTRAINT_NAME_ILLEGAL CREATE TABLE t2 ( s1 INT, === modified file 'mysql-test/t/foreign_key_all_engines_2.test' --- a/mysql-test/t/foreign_key_all_engines_2.test 2008-06-23 07:14:00 +0000 +++ b/mysql-test/t/foreign_key_all_engines_2.test 2008-08-03 10:13:01 +0000 @@ -5,6 +5,227 @@ # --source include/have_innodb.inc --source include/have_partition.inc +--source include/not_embedded.inc + +# +# Some simple tests for 4th milestone of WL#148 ("Implement meta-data +# locking for foreign key definitions and make prelocking algorithm +# aware of foreign keys"). +# +--disable_warnings +drop tables if exists t1, t2, t3; +--enable_warnings +connect (blocker, localhost, root,,); +connection default; +create table t1 (pk int primary key) engine=myisam; +insert into t1 values (1), (2), (3); +connection blocker; +# Take table-level lock and thus shared metadata lock on parent table +lock tables t1 read; +connection default; +# This statement takes exclusive metadata lock on parent so it +# it should be blocked +--send create table t2 (fk int references t1 (pk)) engine=myisam +connection blocker; +let $wait_condition= + select count(*) = 1 from information_schema.processlist + where state = "Waiting for table" and + info = "create table t2 (fk int references t1 (pk)) engine=myisam"; +--source include/wait_condition.inc +unlock tables; +connection default; +--reap +connection blocker; +# Take table-level write lock on parent table +lock tables t1 write; +connection default; +# This statement should take read lock on parent table and +# therefore should be blocked. +--send insert into t2 values (1) +connection blocker; +let $wait_condition= + select count(*) = 1 from information_schema.processlist + where state = "Table lock" and info = "insert into t2 values (1)"; +--source include/wait_condition.inc +unlock tables; +connection default; +--reap +connection blocker; +# Let us check that our check for presence of row in the parent is +# compatible with table-level read lock. +lock table t1 read; +connection default; +insert into t2 values (2); +connection blocker; +unlock tables; +lock table t2 write; +connection default; +# Should be blocked since it has to obtain read lock on t2 +--send update t1 set pk= 4 where pk=3 +connection blocker; +let $wait_condition= + select count(*) = 1 from information_schema.processlist + where state = "Table lock" and info = "update t1 set pk= 4 where pk=3"; +--source include/wait_condition.inc +unlock tables; +connection default; +--reap +connection blocker; +lock table t2 read; +connection default; +# Should not be blocked +update t1 set pk= 5 where pk=4; +connection blocker; +unlock tables; +connection default; +drop tables t1, t2; +# This part of test might start failing once we fully implement all checks +# in CREATE TABLE +create table t1 (pk int primary key) engine=myisam; +insert into t1 values (1), (2), (3); +create table t2 (fk int references t1 (pk) on delete cascade) engine=myisam; +connection blocker; +lock tables t2 read; +connection default; +# Should be blocked since it has to obtain write lock on t1 +--send delete from t1 where pk=3 +connection blocker; +let $wait_condition= + select count(*) = 1 from information_schema.processlist + where state = "Table lock" and info = "delete from t1 where pk=3"; +--source include/wait_condition.inc +unlock tables; +connection default; +--reap +drop tables t1, t2; +# +# Test that column-based locking optimizations work +# +# Tests where column-based optimization works for updates on child table. +create table t1 (pk int primary key) engine=myisam; +insert into t1 values (1); +create table t2 (pk int primary key) engine=myisam; +insert into t2 values (1); +create table t3 (fk1 int references t1 (pk) on delete set null, + fk2 int references t2 (pk)) engine=myisam; +insert into t3 values (1, 1); +connection blocker; +# This will implicitly lock t3 for update, but should not +# lock t2 as cascading change to t3 won't affect t3.fk2. +lock table t1 write; +connection default; +# Should not be blocked +insert into t2 values (2); +connection blocker; +unlock tables; +connection default; +drop tables t1, t2, t3; +create table t1 (pk int primary key) engine=myisam; +insert into t1 values (1); +create table t2 (pk1 int, pk2 int, primary key(pk1, pk2)) engine=myisam; +insert into t2 values (1, 1); +create table t3 (fk1 int references t1 (pk) on delete set null, + fk2 int, + foreign key (fk1, fk2) references t2 (pk1, pk2)) engine=myisam; +insert into t3 values (1, 1); +connection blocker; +# This will implicitly lock t3 for update and t2 for read, +# since cascading change to t3 affects t3.fk2 and thus new +# value of foreign key should be checked. +lock table t1 write; +connection default; +--send insert into t2 values (2, 2) +connection blocker; +let $wait_condition= + select count(*) = 1 from information_schema.processlist + where state = "Table lock" and info = "insert into t2 values (2, 2)"; +--source include/wait_condition.inc +unlock tables; +connection default; +--reap +drop tables t1, t2, t3; +# +# Tests where column-based optimization works for updates on parent table. +# +create table t1 (pk int primary key) engine=myisam; +insert into t1 values (1); +create table t2 (pk int primary key, + fk int references t1 (pk) on delete set null) engine=myisam; +insert into t2 values (1, 1); +create table t3 (fk int references t2 (pk) on delete cascade) engine=myisam; +insert into t3 values (1); +connection blocker; +lock table t3 read; +connection default; +# Should not be blocked since we are not updating t2.pk and +# thus should not do any checks/actions to t3. +delete from t1 where pk= 1; +connection blocker; +unlock tables; +connection default; +drop tables t1, t2, t3; +create table t1 (pk int primary key) engine=myisam; +insert into t1 values (1); +create table t2 (pk1 int, + pk2_fk int references t1 (pk) on update cascade) engine=myisam; +insert into t2 values (1, 1); +create table t3 (fk1 int, fk2 int, + foreign key (fk1, fk2) references t2 (pk1, pk2_fk) + on update set null) engine=myisam; +insert into t3 values (1, 1); +connection blocker; +lock table t3 read; +connection default; +# Should be blocked since we implicitly change primary +# key in t2 and thus should update t3. +--send update t1 set pk= 2 +connection blocker; +let $wait_condition= + select count(*) = 1 from information_schema.processlist + where state = "Table lock" and info = "update t1 set pk= 2"; +--source include/wait_condition.inc +unlock tables; +connection default; +--reap +drop tables t1, t2, t3; +# +# Test that column based optimization properly works with triggers. +# +create table t1 (pk int primary key) engine=myisam; +insert into t1 values (1), (2); +create table t2 (pk int primary key, + fk int references t1 (pk) on delete set null) engine=myisam; +insert into t2 values (1, 1), (2, 2); +create table t3 (fk int references t2 (pk) on update set null) engine=myisam; +insert into t3 values (1), (2); +create trigger t2_bu before update on t2 for each row set @a:= 1; +connection blocker; +lock table t3 read; +connection default; +# Should not be blocked since trigger on t2 doesn't change its +# primary key and therefore t3 won't be updated. +delete from t1 where pk=1; +connection blocker; +unlock tables; +connection default; +# Now scenario in which we should block +drop trigger t2_bu; +create trigger t2_bu before update on t2 for each row set new.pk=3; +connection blocker; +lock table t3 read; +connection default; +# Should be blocked since trigger on t2 will change its primary key +# and therefore t3 should be updated. +--send delete from t1 where pk=2 +connection blocker; +let $wait_condition= + select count(*) = 1 from information_schema.processlist + where state = "Table lock" and info = "delete from t1 where pk=2"; +--source include/wait_condition.inc +unlock tables; +connection default; +--reap +drop tables t1, t2, t3; # === modified file 'mysql-test/t/lock_multi.test' --- a/mysql-test/t/lock_multi.test 2008-05-23 13:54:03 +0000 +++ b/mysql-test/t/lock_multi.test 2008-08-03 10:13:01 +0000 @@ -184,6 +184,7 @@ let $wait_condition= where state = "Waiting for table" and info = "FLUSH TABLES WITH READ LOCK"; --source include/wait_condition.inc # This must not block. +--error ER_LOCK_OR_ACTIVE_TRANSACTION CREATE TABLE t2 (c1 int); UNLOCK TABLES; # @@ -193,7 +194,7 @@ reap; UNLOCK TABLES; # connection default; -DROP TABLE t1, t2; +DROP TABLE t1; # # Test if CREATE TABLE SELECT with LOCK TABLE deadlocks. # === modified file 'sql/fk_dd.cc' --- a/sql/fk_dd.cc 2008-06-25 20:11:52 +0000 +++ b/sql/fk_dd.cc 2008-08-03 10:13:01 +0000 @@ -14,7 +14,10 @@ Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */ #include "mysql_priv.h" +#include "sp.h" #include "fk_dd.h" +#include "sql_trigger.h" +#include "sp_head.h" /* Sizes of various fields in .FRM section describing all foreign keys. */ @@ -30,18 +33,6 @@ const uint FK_FRM_NUM_COLUMNS_SIZE= 2; const uint FK_FRM_NAMES_LENGTH_SIZE= 2; const uint FK_FRM_NAMES_SEP_SIZE= 1; -/* Offsets of fields in description of individual FK. */ -const uint FK_FRM_FLAGS_OFFSET= FK_FRM_FK_DESCR_LENGTH_SIZE; -const uint FK_FRM_MATCH_OPT_OFFSET= FK_FRM_FLAGS_OFFSET + - FK_FRM_FLAGS_SIZE; -const uint FK_FRM_ON_UPDATE_OPT_OFFSET= FK_FRM_MATCH_OPT_OFFSET + - FK_FRM_MATCH_OPT_SIZE; -const uint FK_FRM_ON_DELETE_OPT_OFFSET= FK_FRM_ON_UPDATE_OPT_OFFSET + - FK_FRM_ON_UPDATE_OPT_SIZE; - -const uint FK_FRM_NAMES_LENGTH_OFFSET= FK_FRM_ON_DELETE_OPT_OFFSET + - FK_FRM_ON_DELETE_OPT_SIZE + - FK_FRM_NUM_COLUMNS_SIZE; /** Calculate length of foreign key description in .FRM file. @@ -49,7 +40,7 @@ const uint FK_FRM_NAMES_LENGTH_OFFSET= F @return Length of foreign key description in .FRM file in bytes. */ -uint Foreign_key::get_frm_description_length() +uint Foreign_key_parent::get_frm_description_length() { List_iterator col_it1(columns); List_iterator col_it2(ref_columns); @@ -60,10 +51,11 @@ uint Foreign_key::get_frm_description_le length+= FK_FRM_FK_DESCR_LENGTH_SIZE; /* Space for flags. */ length+= FK_FRM_FLAGS_SIZE; - /* Space for MATCH, ON UPDATE, ON DELETE clauses */ - length+= FK_FRM_MATCH_OPT_SIZE + - FK_FRM_ON_UPDATE_OPT_SIZE + + /* Space for ON UPDATE, ON DELETE clauses */ + length+= FK_FRM_ON_UPDATE_OPT_SIZE + FK_FRM_ON_DELETE_OPT_SIZE; + /* For child table we reserve space for MATCH option. */ + length+= get_additional_fixed_part_length(); /* Space for number of columns in FK. */ length+= FK_FRM_NUM_COLUMNS_SIZE; @@ -72,39 +64,73 @@ uint Foreign_key::get_frm_description_le /* Space for names and columns section separator. */ length+= FK_FRM_NAMES_SEP_SIZE; length+= name.length + FK_FRM_NAMES_SEP_SIZE; + /* + In .FRM of child table we save parent's database and name, in case + of parent table we save database and name of child. If child and parent tables belong to the same database we don't store name of database in .FRM (actually we store empty string instead). */ if (!is_within_one_db) - length+= ref_table->db_length; + length+= get_table_for_frm()->db_length; length+= FK_FRM_NAMES_SEP_SIZE; - length+= ref_table->table_name_length + FK_FRM_NAMES_SEP_SIZE; + length+= get_table_for_frm()->table_name_length + FK_FRM_NAMES_SEP_SIZE; + while ((col= col_it1++)) length+= col->field_name.length + FK_FRM_NAMES_SEP_SIZE; while ((col= col_it2++)) length+= col->field_name.length + FK_FRM_NAMES_SEP_SIZE; + return length; } /** + Get length of fixed part of foreign key description in .FRM + which is specific for parent table. +*/ + +uint Foreign_key_parent::get_additional_fixed_part_length() +{ + return 0; +} + + +/** + Get length of fixed part of foreign key description in .FRM + which is specific for child table. +*/ + +uint Foreign_key::get_additional_fixed_part_length() +{ + return FK_FRM_MATCH_OPT_SIZE; +} + + +/** Calculate total length of .FRM section describing foreign keys. - @param fkey_list List of foreign keys to be save in .FRM + @param fkey_list List of foreign keys for which this table is child. + @param parent_fkey_list List of foreign keys for which this table is parent. @return Length of foreign key section in .FRM file in bytes. */ -uint fk_get_frm_section_length(List &fkey_list) +uint fk_get_frm_section_length(List &fkey_list, + List &parent_fkey_list) { List_iterator fkey_iterator(fkey_list); + List_iterator fkey_p_iterator(parent_fkey_list); Foreign_key *fkey; - uint length= FK_FRM_FKS_SECTION_LENGTH_SIZE; + Foreign_key_parent *fkey_p; + uint length= 2*FK_FRM_FKS_SECTION_LENGTH_SIZE; while ((fkey= fkey_iterator++)) length+= fkey->get_frm_description_length(); + while ((fkey_p= fkey_p_iterator++)) + length+= fkey_p->get_frm_description_length(); + return length; } @@ -125,35 +151,39 @@ static const char FK_FRM_FLAG_WITHIN_ONE bytes of buffer space. */ -void Foreign_key::save_to_frm(char **buff_p) +void Foreign_key_parent::save_to_frm(char **buff_p) { List_iterator col_it1(columns); List_iterator col_it2(ref_columns); const Key_part_spec *col; char *fk_len_pos; char *cur_pos= *buff_p; + char *names_len_pos; /* Here we will store length of FK description a bit later. */ fk_len_pos= cur_pos; cur_pos+= FK_FRM_FK_DESCR_LENGTH_SIZE; /* - If child and parent belong to the same database we don't store - its name. + If child and parent belong to the same database we don't store its name. */ - *cur_pos= (is_table_constraint ? FK_FRM_FLAG_TABLE_CONSTRAINT : 0) | + *cur_pos= get_additional_flags() | (is_within_one_db ? FK_FRM_FLAG_WITHIN_ONE_DATABASE : 0); cur_pos+= FK_FRM_FLAGS_SIZE; - *cur_pos= match_opt; - cur_pos+= FK_FRM_MATCH_OPT_SIZE; + + /* Ensure that we are not saving non-fully-initialized object into .FRM. */ + DBUG_ASSERT(update_opt != FK_OPTION_UNDEF && delete_opt != FK_OPTION_UNDEF); + *cur_pos= update_opt; cur_pos+= FK_FRM_ON_UPDATE_OPT_SIZE; *cur_pos= delete_opt; cur_pos+= FK_FRM_ON_DELETE_OPT_SIZE; + save_additional_fixed_part(&cur_pos); int2store(cur_pos, columns.elements); cur_pos+= FK_FRM_NUM_COLUMNS_SIZE; - /* Again we will store length of names & columns section later. */ + /* Here we will store length of names & columns section later. */ + names_len_pos= cur_pos; cur_pos+= FK_FRM_NAMES_LENGTH_SIZE; /* Names & columns section starts with separator character. */ @@ -163,15 +193,18 @@ void Foreign_key::save_to_frm(char **buf *cur_pos= NAMES_SEP_CHAR; cur_pos+= FK_FRM_NAMES_SEP_SIZE; /* + In .FRM of child table we save parent's database and name, + in case of parent table we save database and name of child. + By storing lowercased version of db and table name when lowercase_table_names option is enabled we follow InnoDB implementation. */ if (!is_within_one_db) - cur_pos= strmov(cur_pos, ref_table->db); + cur_pos= strmov(cur_pos, get_table_for_frm()->db); *cur_pos= NAMES_SEP_CHAR; cur_pos+= FK_FRM_NAMES_SEP_SIZE; - cur_pos= strmov(cur_pos, ref_table->table_name); + cur_pos= strmov(cur_pos, get_table_for_frm()->table_name); *cur_pos= NAMES_SEP_CHAR; cur_pos+= FK_FRM_NAMES_SEP_SIZE; while ((col= col_it1++)) @@ -187,10 +220,10 @@ void Foreign_key::save_to_frm(char **buf cur_pos+= FK_FRM_NAMES_SEP_SIZE; } /* Names & columns part ends. Write its length. */ - int2store(fk_len_pos + FK_FRM_NAMES_LENGTH_OFFSET, - cur_pos - (fk_len_pos + FK_FRM_NAMES_LENGTH_OFFSET + - FK_FRM_NAMES_LENGTH_SIZE)); + int2store(names_len_pos, cur_pos - (names_len_pos + FK_FRM_NAMES_LENGTH_SIZE)); + /* This is space reserved for future extensions. */ + /* This is the end of the whole description for this FK so we can store its length. @@ -201,39 +234,126 @@ void Foreign_key::save_to_frm(char **buf /** + Get flags for foreign key description in .FRM which are specific for + description in parent table. +*/ + +char Foreign_key_parent::get_additional_flags() +{ + return 0; +} + + +/** + Save fixed part which is specific for child table to the + description of foreign key in .FRM. +*/ + +void Foreign_key_parent::save_additional_fixed_part(char **buff_p) +{ +} + + +/** + Get table database and name of which should be saved in the description + of foreign key in parent table. +*/ + +TABLE_LIST* Foreign_key_parent::get_table_for_frm() +{ + return table; +} + + +/** + Get flags for foreign key description in .FRM which are specific for + description in child table. +*/ + +char Foreign_key::get_additional_flags() +{ + return is_table_constraint ? FK_FRM_FLAG_TABLE_CONSTRAINT : 0; +} + + +/** + Save fixed part which is specific for child table to the + description of foreign key in .FRM. +*/ + +void Foreign_key::save_additional_fixed_part(char **buff_p) +{ + /* Ensure that we are not saving non-fully-initialized object into .FRM. */ + DBUG_ASSERT(match_opt != FK_MATCH_UNDEF); + **buff_p= match_opt; + *buff_p+= FK_FRM_MATCH_OPT_SIZE; +} + + +/** + Get table database and name of which should be saved in the description + of foreign key in child table. +*/ + +TABLE_LIST* Foreign_key::get_table_for_frm() +{ + return ref_table; +} + + +/** Save descriptions of foreign keys in the .FRM file. @param thd Thread context. @param file Descriptor of open .FRM file. - @param fkey_list List of foreign keys. - @param predicted_length Length of the description which was predicted + @param fkey_list List of foreign keys for which this table is child. + @param parent_fkey_list List of foreign keys for which this table is parent. + @param predicted_length Length of the descriptions which was predicted by fk_get_frm_section_length() function. @retval TRUE in case of error. @retval FALSE otherwise. */ -bool fk_save_to_frm_section(THD *thd, File file, List &fkey_list, +bool fk_save_to_frm_section(THD *thd, File file, + List &fkey_list, + List &parent_fkey_list, uint predicted_length) { List_iterator fkey_iterator(fkey_list); + List_iterator fkey_p_iterator(parent_fkey_list); char *buff= (char *)thd->alloc(predicted_length); char *cur_pos= buff; + char *sec_pos; Foreign_key *fkey; + Foreign_key_parent *fkey_p; /* Check for OOM. */ if (!buff) return TRUE; - /* Store length of the whole section describing FKs. */ - int2store(cur_pos, predicted_length - FK_FRM_FKS_SECTION_LENGTH_SIZE); + /* Save descriptions of foreign keys in which this table is child. */ + sec_pos= cur_pos; cur_pos+= FK_FRM_FKS_SECTION_LENGTH_SIZE; - while ((fkey= fkey_iterator++)) { DBUG_ASSERT(cur_pos < buff + predicted_length); fkey->save_to_frm(&cur_pos); } + int2store(sec_pos, cur_pos - (sec_pos + FK_FRM_FKS_SECTION_LENGTH_SIZE)); + + /* Save descriptions of foreign keys in which this table is parent. */ + sec_pos= cur_pos; + cur_pos+= FK_FRM_FKS_SECTION_LENGTH_SIZE; + while ((fkey_p= fkey_p_iterator++)) + { + DBUG_ASSERT(cur_pos < buff + predicted_length); + fkey_p->save_to_frm(&cur_pos); + } + int2store(sec_pos, cur_pos - (sec_pos + FK_FRM_FKS_SECTION_LENGTH_SIZE)); + + DBUG_ASSERT(cur_pos == buff + predicted_length); + return my_write(file, (const uchar*)buff, predicted_length, MYF(MY_NABP)); } @@ -242,32 +362,28 @@ bool fk_save_to_frm_section(THD *thd, Fi Restore foreign key object from its description in .FRM file. @param thd [in] Thread context. - @param mem_root [in] MEM_ROOT on which object representing FK - should be allocated. - @param db [in] Database in which with which this FK - is associated belongs. + @param share [in] Share of the table which .FRM is being read and + which memory root will be used for allocating + auxiliary objects for foreign key. @param buff_p [in/out] On entering routine should point to the beginning of FK description in the buffer, on leaving it will point to the position right after this description. - @retval non-NULL Pointer to Foreign_key object constructed from - description provided. - @retval NULL In case of error. + @retval FALSE Success. + @retval TRUE Failure. */ -Foreign_key *Foreign_key::restore_from_frm(THD *thd, MEM_ROOT *mem_root, - const LEX_STRING &db, - const char **buff_p) +bool Foreign_key_share::restore_from_frm(THD *thd, TABLE_SHARE *share, + const char **buff_p) { TYPELIB dummy_type; - uint columns, fkey_descr_length, names_cols_len, names_num; + uint col_number, fkey_descr_length, names_cols_len, names_num; char *names_cols_buff; const char **names_cols_arr, **names_cols_ptr; unsigned int *names_cols_lengths, *names_cols_lengths_ptr; - TABLE_LIST *table; - List cols, ref_cols; + LEX_STRING *col_str_arr; + LEX_STRING db, table_name; uint i; - bool is_within_one_db; const char *cur_pos= *buff_p; const char *fkey_descr_start= cur_pos; @@ -276,86 +392,150 @@ Foreign_key *Foreign_key::restore_from_f cur_pos+= FK_FRM_FK_DESCR_LENGTH_SIZE; is_within_one_db= (*cur_pos) & FK_FRM_FLAG_WITHIN_ONE_DATABASE; + restore_additional_flags(*cur_pos); cur_pos+= FK_FRM_FLAGS_SIZE; + update_opt= (fk_option)*cur_pos; + cur_pos+= FK_FRM_ON_UPDATE_OPT_SIZE; + delete_opt= (fk_option)*cur_pos; + cur_pos+= FK_FRM_ON_DELETE_OPT_SIZE; - /* Skip space where info about MATCH and ON UPDATE/DELETE are stored. */ - cur_pos+= FK_FRM_MATCH_OPT_SIZE + FK_FRM_ON_UPDATE_OPT_SIZE + - FK_FRM_ON_DELETE_OPT_SIZE; + /* Read part of fixed header which is specific for child or parent. */ + restore_from_additional_fixed_part(&cur_pos); - columns= uint2korr(cur_pos); + col_number= uint2korr(cur_pos); cur_pos+= FK_FRM_NUM_COLUMNS_SIZE; /* - Load names of constraint, parent table and columns into the + Load names of constraint, parent or child table and columns into the strings allocated on share's memory root. */ names_cols_len= uint2korr(cur_pos); cur_pos+= FK_FRM_NAMES_LENGTH_SIZE; - names_cols_buff= strmake_root(mem_root, cur_pos, names_cols_len); - names_num= 3 + columns * 2 + 1; - if (!(names_cols_arr= (const char **)alloc_root(mem_root, - sizeof(char*) * names_num + - sizeof(unsigned int) * - names_num))) - return NULL; + names_cols_buff= strmake_root(&share->mem_root, cur_pos, names_cols_len); + names_num= 3 + col_number * 2 + 1; + if (!(multi_alloc_root(&share->mem_root, + &names_cols_arr, sizeof(char*) * names_num, + &names_cols_lengths, sizeof(unsigned int) * names_num, + &col_str_arr, sizeof(LEX_STRING) * 2 * 2 * col_number, + NULL, 0))) + return TRUE; names_cols_ptr= names_cols_arr; - names_cols_lengths= (unsigned int *)(names_cols_arr + names_num); names_cols_lengths_ptr= names_cols_lengths; build_typelibs(&names_cols_ptr, &names_cols_lengths_ptr, &dummy_type, 1, &names_cols_buff); DBUG_ASSERT(dummy_type.count == names_num - 1); - for (i= 0; i< columns; i++) - cols.push_back(new (mem_root) Key_part_spec(names_cols_arr[3+i], - names_cols_lengths[3+i], - 0), - mem_root); - for (i= 0; i< columns; i++) - ref_cols.push_back(new (mem_root) Key_part_spec( - names_cols_arr[3+columns+i], - names_cols_lengths[3+columns+i], - 0), - mem_root); + name.str= (char*)names_cols_arr[0]; + name.length= names_cols_lengths[0]; + + for (i= 0; i< col_number; i++) + { + col_str_arr[i].str= (char *)names_cols_arr[3+i]; + col_str_arr[i].length= names_cols_lengths[3+i]; + columns.push_back(&col_str_arr[i], &share->mem_root); + } + for (i= 0; i< col_number; i++) + { + col_str_arr[col_number+i].str= (char *)names_cols_arr[3+col_number+i]; + col_str_arr[col_number+i].length= names_cols_lengths[3+col_number+i]; + ref_columns.push_back(&col_str_arr[col_number+i], &share->mem_root); + } /* Check for OOM .*/ if (thd->is_fatal_error) - return NULL; - - if (!(table= (TABLE_LIST *)alloc_root(mem_root, sizeof(TABLE_LIST)))) - return NULL; - bzero(table, sizeof(TABLE_LIST)); + return TRUE; if (is_within_one_db) - { - table->db= db.str; - table->db_length= db.length; - } + db= share->db; else { - table->db= (char *)names_cols_arr[1]; - table->db_length= names_cols_lengths[1]; + db.str= (char *)names_cols_arr[1]; + db.length= names_cols_lengths[1]; } - table->table_name= (char *)names_cols_arr[2]; - table->table_name_length= names_cols_lengths[2]; + table_name.str= (char *)names_cols_arr[2]; + table_name.length= names_cols_lengths[2]; + set_table_from_frm(db, table_name); + + /* Skip part of description which comes from later versions of server. */ *buff_p= fkey_descr_start + FK_FRM_FK_DESCR_LENGTH_SIZE + fkey_descr_length; - return new (mem_root) Foreign_key(names_cols_arr[0], - names_cols_lengths[0], - (fkey_descr_start[FK_FRM_FLAGS_OFFSET] & - FK_FRM_FLAG_TABLE_CONSTRAINT), - is_within_one_db, cols, table, ref_cols, - fkey_descr_start[FK_FRM_ON_DELETE_OPT_OFFSET], - fkey_descr_start[FK_FRM_ON_UPDATE_OPT_OFFSET], - fkey_descr_start[FK_FRM_MATCH_OPT_OFFSET], - TRUE); + return FALSE; +} + + +/** + Restore information about foreign key which was saved in .FRM as flags + specific for foreign key description in child table. +*/ +void Foreign_key_child_share::restore_additional_flags(char flags) +{ + is_table_constraint= flags & FK_FRM_FLAG_TABLE_CONSTRAINT; } /** - Restore list of foreign keys for the table from its description in the .FRM. + Restore information about foreign key which was saved in fixed part of foreign + key description and which is specific for description in child table. +*/ + +void +Foreign_key_child_share::restore_from_additional_fixed_part(const char **buff_p) +{ + match_opt= (fk_match_opt) **buff_p; + *buff_p+= FK_FRM_MATCH_OPT_SIZE; +} + + +/** + Set name and database of another (i.e. parent) table participating in + foreign key constraint from its description in child table. +*/ + +void Foreign_key_child_share::set_table_from_frm(LEX_STRING &db, LEX_STRING &table_name) +{ + parent_table_db= db; + parent_table_name= table_name; +} + + +/** + Restore information about foreign key which was saved in .FRM as flags + specific for foreign key description in parent table. +*/ + +void Foreign_key_parent_share::restore_additional_flags(char flags) +{ +} + + +/** + Restore information about foreign key which was saved in fixed part of foreign + key description and which is specific for description in parent table. +*/ + +void +Foreign_key_parent_share::restore_from_additional_fixed_part(const char **buff_p) +{ +} + + +/** + Set name and database of another (i.e. child) table participating in + foreign key constraint from its description in parent table. +*/ + +void Foreign_key_parent_share::set_table_from_frm(LEX_STRING &db, LEX_STRING &table_name) +{ + child_table_db= db; + child_table_name= table_name; +} + + +/** + Restore foreign keys for the table their description in the .FRM. @param thd [in] Thread context. @param share [in] Share of the table FKs for which should be restored. @@ -371,29 +551,43 @@ Foreign_key *Foreign_key::restore_from_f bool fk_restore_from_frm_section(THD *thd, TABLE_SHARE *share, const char **buff_p, const char *buff_end) { - Foreign_key *fkey; uint section_length; - if (*buff_p + FK_FRM_FKS_SECTION_LENGTH_SIZE >= buff_end) + if (*buff_p + 2 * FK_FRM_FKS_SECTION_LENGTH_SIZE >= buff_end) return FALSE; + /* Restore foreign keys for which this table is child. */ + section_length= uint2korr(*buff_p); + (*buff_p)+= FK_FRM_FKS_SECTION_LENGTH_SIZE; + + buff_end= *buff_p + section_length; + + while ((*buff_p) < buff_end) + { + Foreign_key_child_share *fkc= new (&share->mem_root) Foreign_key_child_share(); + if (fkc->restore_from_frm(thd, share, buff_p)) + return TRUE; + share->fkeys_child.push_back(fkc, &share->mem_root); + } + + DBUG_ASSERT(*buff_p == buff_end); + + /* Restore foreign keys for which this table is parent. */ section_length= uint2korr(*buff_p); (*buff_p)+= FK_FRM_FKS_SECTION_LENGTH_SIZE; - /* - TODO: Investigate if we should do more checks for corruption. - E.g. that new_buff_end <= buff_end ? - */ buff_end= *buff_p + section_length; while ((*buff_p) < buff_end) { - if (!(fkey= Foreign_key::restore_from_frm(thd, &share->mem_root, - share->db, buff_p)) || - share->fkeys.push_back(fkey, &share->mem_root)) + Foreign_key_parent_share *fkp= new (&share->mem_root) Foreign_key_parent_share(); + if (fkp->restore_from_frm(thd, share, buff_p)) return TRUE; + share->fkeys_parent.push_back(fkp, &share->mem_root); } + DBUG_ASSERT(*buff_p == buff_end); + return FALSE; } @@ -429,15 +623,12 @@ const LEX_STRING fk_match_type_names[]= @param thd [in] Thread context. @param str [in/out] String to which clause should be appended. - @param db [in] Name of database where the table resides. */ -void Foreign_key::get_create_statement_clause(THD *thd, String *str, - const char *db) +void Foreign_key_child_share::get_create_statement_clause(THD *thd, String *str) { - List_iterator col_it1(columns); - List_iterator col_it2(ref_columns); - Key_part_spec *col; + List_iterator col_it1(columns), col_it2(ref_columns); + LEX_STRING *col; bool first; if (is_table_constraint) @@ -456,7 +647,7 @@ void Foreign_key::get_create_statement_c str->append(STRING_WITH_LEN(", ")); else first= FALSE; - append_identifier(thd, str, col->field_name.str, col->field_name.length); + append_identifier(thd, str, col->str, col->length); } str->append(')'); } @@ -469,11 +660,10 @@ void Foreign_key::get_create_statement_c */ if (!is_within_one_db) { - append_identifier(thd, str, ref_table->db, ref_table->db_length); + append_identifier(thd, str, parent_table_db.str, parent_table_db.length); str->append('.'); } - append_identifier(thd, str, ref_table->table_name, - ref_table->table_name_length); + append_identifier(thd, str, parent_table_name.str, parent_table_name.length); str->append(STRING_WITH_LEN(" (")); first= TRUE; while ((col= col_it2++)) @@ -482,25 +672,25 @@ void Foreign_key::get_create_statement_c str->append(STRING_WITH_LEN(", ")); else first= FALSE; - append_identifier(thd, str, col->field_name.str, col->field_name.length); + append_identifier(thd, str, col->str, col->length); } str->append(')'); - if (match_opt != Foreign_key::FK_MATCH_SIMPLE) + if (match_opt != FK_MATCH_SIMPLE) { str->append(STRING_WITH_LEN(" MATCH ")); str->append(fk_match_type_names[match_opt].str, fk_match_type_names[match_opt].length); } - if (delete_opt != Foreign_key::FK_OPTION_NO_ACTION) + if (delete_opt != FK_OPTION_NO_ACTION) { str->append(STRING_WITH_LEN(" ON DELETE ")); str->append(fk_ref_action_names[delete_opt].str, fk_ref_action_names[delete_opt].length); } - if (update_opt != Foreign_key::FK_OPTION_NO_ACTION) + if (update_opt != FK_OPTION_NO_ACTION) { str->append(STRING_WITH_LEN(" ON UPDATE ")); str->append(fk_ref_action_names[update_opt].str, @@ -522,15 +712,15 @@ void Foreign_key::get_create_statement_c void fk_get_column_constraint_for_create(THD *thd, String *str, TABLE *table, const char *field_name) { - List_iterator fkey_it(table->s->fkeys); - Foreign_key *fkey; + List_iterator fkey_it(table->s->fkeys_child); + Foreign_key_child_share *fkey; while ((fkey= fkey_it++)) { if (!fkey->is_table_constraint && !my_strcasecmp(system_charset_info, field_name, - fkey->columns.head()->field_name.str)) + fkey->columns.head()->str)) { - fkey->get_create_statement_clause(thd, str, table->s->db.str); + fkey->get_create_statement_clause(thd, str); continue; } } @@ -548,11 +738,11 @@ void fk_get_column_constraint_for_create void fk_get_table_constraints_for_create(THD *thd, String *str, TABLE *table) { - List_iterator fkey_it(table->s->fkeys); - Foreign_key *fkey; + List_iterator fkey_it(table->s->fkeys_child); + Foreign_key_child_share *fkey; while ((fkey= fkey_it++)) if (fkey->is_table_constraint) - fkey->get_create_statement_clause(thd, str, table->s->db.str); + fkey->get_create_statement_clause(thd, str); } @@ -748,12 +938,22 @@ static bool fk_create_constraint_name(co bool fk_create_constraint_names(List &fkey_list, const char *db) { List_iterator fkey_iterator(fkey_list); - Foreign_key *fkey; + Foreign_key *fkey, *fkey_err; while ((fkey= fkey_iterator++)) - if ((!fkey->exists) && fk_create_constraint_name(db, fkey->name.str)) - return TRUE; + { + DBUG_ASSERT(!fkey->exists); + if (fk_create_constraint_name(db, fkey->name.str)) + goto err; + } return FALSE; + +err: + fkey_iterator.rewind(); + fkey_err= fkey; + while ((fkey= fkey_iterator++) && fkey != fkey_err) + (void) fk_drop_constraint_name(db, fkey->name.str); + return TRUE; } @@ -816,8 +1016,8 @@ bool fk_drop_all_constraint_names_for_ta return TRUE; } - List_iterator fkey_iterator(share.fkeys); - Foreign_key *fkey; + List_iterator fkey_iterator(share.fkeys_child); + Foreign_key_child_share *fkey; while ((fkey= fkey_iterator++)) if (fk_drop_constraint_name(table->db, fkey->name.str)) @@ -837,15 +1037,13 @@ bool fk_drop_all_constraint_names_for_ta @param thd Thread context @param fkey Foreign key to be created. @param fkey_list List of all foreign keys for the table. - @param db Database of child table for this FK. @retval FALSE Success, all checks passed. @retval TRUE Some error discovered. */ bool fk_check_constraint_added(THD *thd, Foreign_key *fkey, - List &fkey_list, - const char *db) + List &fkey_list) { List_iterator fkey_it2(fkey_list); Foreign_key *fkey2; @@ -860,7 +1058,7 @@ bool fk_check_constraint_added(THD *thd, if (fkey->exists) DBUG_RETURN(FALSE); - if (fkey->match_opt == Foreign_key::FK_MATCH_PARTIAL) + if (fkey->match_opt == FK_MATCH_PARTIAL) { my_error(ER_FK_MATCH_PARTIAL, MYF(0), fkey->name.str); DBUG_RETURN(TRUE); @@ -914,7 +1112,7 @@ bool fk_check_constraint_added(THD *thd, my_error(ER_FK_CONSTRAINT_NAME_ILLEGAL, MYF(0), fkey->name.str); DBUG_RETURN(TRUE); } - if (fk_check_if_constraint_name_exists(db, fkey->name.str)) + if (fk_check_if_constraint_name_exists(fkey->table->db, fkey->name.str)) { my_error(ER_FK_CONSTRAINT_NAME_DUPLICATE, MYF(0), fkey->name.str); DBUG_RETURN(TRUE); @@ -934,4 +1132,485 @@ bool fk_check_constraint_added(THD *thd, } DBUG_RETURN(FALSE); +} + + +/** + Convert list of columns names to a list of Key_part_spec objects + which can be used for creation of foreign key using this columns. + + @param colnames [in] List of column names to be converted + @param cols [in/out] Initially empty list of Key_part_spec objects + to be filled according to colnames list + @param mem_root [in] Memory root on which objects shoudl be allocated + + @note Result list will contain references to the strings from original + list and thus lifetime of the new list is limited by lifetime of + original one. + + @retval TRUE An error occured (OOM). + @retval FALSE Success. +*/ + +static bool fk_colnames_to_key_part_spec(List &colnames, + List *cols, + MEM_ROOT *mem_root) +{ + List_iterator it(colnames); + LEX_STRING *col; + Key_part_spec *kps; + + while ((col= it++)) + { + if (!(kps= new (mem_root) Key_part_spec(*col, 0)) || + cols->push_back(kps, mem_root)) + return TRUE; + } + return FALSE; +} + + +/** + Given object representing foreign key in the share of parent table + create Foreign_key_parent instance describing it, which can be used + for saving information about this foreign key into .FRM (of the + parent table) during ALTER TABLE. + + @param mem_root Memory root to be used for allocating new object. + + @note Since created object will contain references to the memory and + members of original object its life-time is limited by lifetime + of the original object. + + @retval non-NULL Pointer to new Foreign_key_parent instance. + @retval NULL OOM occurred. +*/ + +Foreign_key_parent* +Foreign_key_parent_share::make_foreign_key(MEM_ROOT *mem_root) +{ + List cols, ref_cols; + TABLE_LIST *table; + + if (fk_colnames_to_key_part_spec(columns, &cols, mem_root) || + fk_colnames_to_key_part_spec(ref_columns, &ref_cols, mem_root)) + return NULL; + + if (!(table= (TABLE_LIST*)alloc_root(mem_root, sizeof(TABLE_LIST)))) + return NULL; + + bzero(table, sizeof(TABLE_LIST)); + table->db= child_table_db.str; + table->db_length= child_table_db.length; + table->table_name= child_table_name.str; + table->table_name_length= child_table_name.length; + + return new (mem_root) Foreign_key_parent(name, table, cols, ref_cols, + delete_opt, update_opt, + is_within_one_db); +} + + +/** + Given object representing foreign key in the share of child table + create Foreign_key instance describing it, which can be used for + saving information about this foreign key into .FRM during ALTER + TABLE. + + @param mem_root Memory root to be used for allocating new object. + + @note Since created object will contain references to the memory and + members of original object its life-time is limited by lifetime + of the original object. + + @retval non-NULL Pointer to new Foreign_key instance. + @retval NULL OOM occurred. +*/ + +Foreign_key* +Foreign_key_child_share::make_foreign_key(MEM_ROOT *mem_root, TABLE_LIST *table) +{ + List cols, ref_cols; + TABLE_LIST *ref_table; + + if (fk_colnames_to_key_part_spec(columns, &cols, mem_root) || + fk_colnames_to_key_part_spec(ref_columns, &ref_cols, mem_root)) + return NULL; + + if (!(ref_table= (TABLE_LIST*)alloc_root(mem_root, sizeof(TABLE_LIST)))) + return NULL; + + bzero(ref_table, sizeof(TABLE_LIST)); + ref_table->db= parent_table_db.str; + ref_table->db_length= parent_table_db.length; + ref_table->table_name= parent_table_name.str; + ref_table->table_name_length= parent_table_name.length; + + return new (mem_root) Foreign_key(name, null_lex_str, is_table_constraint, + table, cols, ref_table, ref_cols, + delete_opt, update_opt, match_opt, + is_within_one_db, TRUE); +} + + +/** + Mark columns from the list as used in the set of used table columns. + + @param columns List of column names + @param share Table's share + @param bitmap Set of used columns +*/ + +static void fk_colnames_to_set(List &columns, TABLE_SHARE *share, + MY_BITMAP *bitmap) +{ + List_iterator col_it(columns); + LEX_STRING *col; + Field *field; + while((col= col_it++)) + { + if ((field= find_field_in_table_def(share, col->str, col->length))) + bitmap_set_bit(bitmap, field->field_index); + } +} + + +/** + Initialize set of columns of the table participating in the foreign key. + + @note This set is used by optimization in fk_add_tables_for_fks() + function for determining if foreign key is affected by the + update operation on the table. + @note This method is invoked at the end of the process of TABLE_SHARE + initialization (when we have properly set-up structures for + looking up fields by names). + + @param share Share of the table for which this Foreign_key_share object + represents a foreign key. + + @retval FALSE Success. + @retval TRUE Failure (OOM). +*/ + +bool Foreign_key_share::init_column_set(TABLE_SHARE *share) +{ + my_bitmap_map *bitmap_buff; + if (!(bitmap_buff= (my_bitmap_map*) alloc_root(&share->mem_root, + share->column_bitmap_size))) + return TRUE; + bitmap_init(&column_set, bitmap_buff, share->fields, FALSE); + bitmap_clear_all(&column_set); + fk_colnames_to_set(get_table_columns(), share, &column_set); + return FALSE; +} + + +/** + For the object respesenting foreign key in a table share get list + of columns in this table which participate in this foreign key. + (in case of Foreign_key_child_share it is list of referencing + columns). +*/ + +List& Foreign_key_child_share::get_table_columns() +{ + return columns; +} + + +/** + For the object respesenting foreign key in a table share get list + of columns in this table which participate in this foreign key. + (in case of Foreign_key_parent_share it is list of referenced + columns). +*/ + +List& Foreign_key_parent_share::get_table_columns() +{ + return ref_columns; +} + + +/** + Initialize sets of the table columns participating in foreign keys + for all table's foreign keys. + + @sa Foreign_key_share::column_set/init_column_set(). + + @param share Table's share. + + @retval FALSE Success. + @retval TRUE Failure (OOM). +*/ + +bool fk_init_column_sets(TABLE_SHARE *share) +{ + List_iterator it(share->fkeys_child); + List_iterator it2(share->fkeys_parent); + Foreign_key_child_share *c_fkey; + Foreign_key_parent_share *p_fkey; + + while ((c_fkey= it++)) + if (c_fkey->init_column_set(share)) + return TRUE; + while ((p_fkey= it2++)) + if (p_fkey->init_column_set(share)) + return TRUE; + return FALSE; +} + + +/** + The role in which this foreign key is used in the current statement. + + When generating the set of tables to prelock, we also construct a + unique set of all foreign keys invoked by the statement, directly + or indirectly. + A foreign key may be present several times in this set, but not more + than 3: one for each role enumerated below (so actually it is set + of pairs). + Addition of a new combination happens only if it is not + yet present in the set. When adding a combination to the set, we also + add TABLE_LIST elements for involved tables to the list of tables to + prelock. + In case of FK_ROLE_CHILD_INSERT, the parent table is added with READ + lock. In case of FK_ROLE_PARENT_UPDATE or FK_ROLE_PARENT_DELETE, the + child table with WRITE lock (to perform ON UPDATE/DELETE CASCADE/SET + NULL/SET DEFAULT). +*/ + +enum fk_role {FK_ROLE_CHILD_INSERT= 0, FK_ROLE_PARENT_UPDATE, + FK_ROLE_PARENT_DELETE}; + + +/** + Add foreign key constraint + role combination to the set of "stored + routines" used by statement. + + @note By doing this we ensure that each foreign key/role pair is + handled by prelocking algorithm only once for each statement. + + @param thd Thread context. + @param lex LEX of this statement + @param fk Foreign key + @param role Role of the foreign key to be added + + @retval TRUE - constraint/role pair was added. + @retval FALSE - constraint/role pair was not added since + it is already present in the set. +*/ + +static bool fk_add_used_fk(THD *thd, LEX *lex, const char *db, + const char *name, fk_role role) +{ + LEX_STRING key; + char key_buff[1+1+2*NAME_LEN+1+1]; + key_buff[0]= TYPE_ENUM_FOREIGN_KEY; + key_buff[1]= role; + key.str= key_buff; + key.length= strxnmov(key_buff+2, sizeof(key_buff)-3, db, ".", + name, NULL) - key_buff; + return sp_add_used_routine(lex, thd->stmt_arena, &key, 0); +} + + +/** + Add table needed for foreign key constraint check or action to the + table list of statement. + + @param thd Thread context + @param lex LEX for the statement + @param tab TABLE_LIST for the table from the foreign key + description + @param lock_type Type of lock + @param trg_event_map Bitmap with operations to be performed on table + @param belong_to_view Uppermost view which uses this foreign key + (0 if foreign key is not used by view). + @param columns Pointer to the list of columns which are going + to be updated by the action (0 - if check/action + won't involve update). + + @retval FALSE Success. + @retval TRUE Failure (OOM). +*/ + +static bool fk_add_table_to_table_list(THD *thd, LEX *lex, + const LEX_STRING &db, + const LEX_STRING &table_name, + thr_lock_type lock_type, + uint8 trg_event_map, + TABLE_LIST *belong_to_view, + List *columns) +{ + TABLE_LIST *table; + char *db_buff, *table_name_buff; + Query_arena *arena, backup; + + arena= thd->activate_stmt_arena_if_needed(&backup); + + if (!multi_alloc_root(thd->mem_root, &table, sizeof(TABLE_LIST), + &db_buff, db.length + 1, + &table_name_buff, table_name.length + 1, + NULL, 0)) + return TRUE; + + bzero(table, sizeof(TABLE_LIST)); + strmov(db_buff, db.str); + table->db= db_buff; + table->db_length= db.length; + strmov(table_name_buff, table_name.str); + table->table_name= table_name_buff; + table->table_name_length= table_name.length; + table->alias= table->table_name; + table->lock_type= lock_type; + table->lock_timeout= -1; + table->lock_transactional= FALSE; + table->cacheable_table= 1; + table->prelocking_placeholder= 1; + table->belong_to_view= belong_to_view; + table->trg_event_map= trg_event_map; + table->cols_to_be_updated= columns; + table->mdl_lock_data= mdl_alloc_lock(0, table->db, table->table_name, + thd->locked_tables_root ? + thd->locked_tables_root : + thd->mem_root); + + if (!table->mdl_lock_data) + return TRUE; + + /* Everything else should be zeroed. */ + if (arena) + thd->restore_active_arena(arena, &backup); + + lex->add_to_query_tables(table); + + return FALSE; +} + + +/** + Process all foreign key constraints which are going to be involved + in operation performed on the table and if needed add appropriate + tables to the table list. + + @param thd Thread context + @param lex LEX of the statement + @param table Element of table list to be processed + + @retval FALSE - Success + @retval TRUE - Failure (OOM) +*/ + +bool fk_add_tables_for_fks(THD *thd, LEX *lex, TABLE_LIST *table) +{ + MY_BITMAP *cols_to_be_updated_set; + bool is_insert= (table->trg_event_map & + static_cast(1 << static_cast(TRG_EVENT_INSERT))); + bool is_update= (table->trg_event_map & + static_cast(1 << static_cast(TRG_EVENT_UPDATE))); + bool is_delete= (table->trg_event_map & + static_cast(1 << static_cast(TRG_EVENT_DELETE))); + + if (table->cols_to_be_updated) + { + /* + For this table we know list of columns which are going + to be updated. Let us convert it to set representation. + */ + cols_to_be_updated_set= &table->table->tmp_set; + bitmap_clear_all(cols_to_be_updated_set); + fk_colnames_to_set(*table->cols_to_be_updated, table->table->s, + cols_to_be_updated_set); + /* Add set of columns which can be changed by ON UPDATE trigger. */ + if (table->table->triggers) + { + bitmap_union(cols_to_be_updated_set, + table->table->triggers->get_update_triggers_write_set()); + } + } + + if (is_insert || is_update) + { + List_iterator it(table->table->s->fkeys_child); + Foreign_key_child_share *fk; + while ((fk= it++)) + { + if ((is_insert || + !table->cols_to_be_updated || + bitmap_is_overlapping(&fk->column_set, cols_to_be_updated_set)) && + fk_add_used_fk(thd, lex, table->db, fk->name.str, + FK_ROLE_CHILD_INSERT)) + { + if (fk_add_table_to_table_list(thd, lex, + fk->parent_table_db, + fk->parent_table_name, + TL_READ_NO_INSERT, 0, + table->belong_to_view, 0)) + return TRUE; + } + } + } + if (is_update || is_delete) + { + List_iterator it(table->table->s->fkeys_parent); + Foreign_key_parent_share *fk; + while ((fk= it++)) + { + if (is_update && + (!table->cols_to_be_updated || + bitmap_is_overlapping(&fk->column_set, cols_to_be_updated_set)) && + fk_add_used_fk(thd, lex, fk->child_table_db.str, fk->name.str, + FK_ROLE_PARENT_UPDATE)) + { + if (fk->update_opt != FK_OPTION_NO_ACTION && + fk->update_opt != FK_OPTION_RESTRICT) + { + if (fk_add_table_to_table_list(thd, lex, + fk->child_table_db, fk->child_table_name, TL_WRITE, + static_cast(1 << static_cast(TRG_EVENT_UPDATE)), + table->belong_to_view, &fk->columns)) + return TRUE; + } + else + { + if (fk_add_table_to_table_list(thd, lex, fk->child_table_db, + fk->child_table_name, + TL_READ_NO_INSERT, 0, + table->belong_to_view, 0)) + return TRUE; + } + } + if (is_delete && + fk_add_used_fk(thd, lex, fk->child_table_db.str, fk->name.str, + FK_ROLE_PARENT_DELETE)) + { + if (fk->delete_opt == FK_OPTION_NO_ACTION || + fk->delete_opt == FK_OPTION_RESTRICT) + { + if (fk_add_table_to_table_list(thd, lex, fk->child_table_db, + fk->child_table_name, + TL_READ_NO_INSERT, 0, + table->belong_to_view, 0)) + return TRUE; + } + else if (fk->delete_opt == FK_OPTION_CASCADE) + { + if (fk_add_table_to_table_list(thd, lex, + fk->child_table_db, fk->child_table_name, TL_WRITE, + static_cast(1 << static_cast(TRG_EVENT_DELETE)), + table->belong_to_view, 0)) + return TRUE; + } + else + { + if (fk_add_table_to_table_list(thd, lex, + fk->child_table_db, fk->child_table_name, TL_WRITE, + static_cast(1 << static_cast(TRG_EVENT_UPDATE)), + table->belong_to_view, &fk->columns)) + return TRUE; + } + } + } + } + return FALSE; } === modified file 'sql/fk_dd.h' --- a/sql/fk_dd.h 2008-06-25 19:52:25 +0000 +++ b/sql/fk_dd.h 2008-08-03 10:13:01 +0000 @@ -16,8 +16,10 @@ #ifndef SQL_FK_H #define SQL_FK_H -uint fk_get_frm_section_length(List &fkey_list); +uint fk_get_frm_section_length(List &fkey_list, + List &parent_fkey_list); bool fk_save_to_frm_section(THD *thd, File file, List &fkey_list, + List &parent_fkey_list, uint predicted_length); bool fk_restore_from_frm_section(THD *thd, TABLE_SHARE *share, const char **buff_p, const char *buff_end); @@ -39,7 +41,10 @@ bool fk_drop_constraint_name(const char bool fk_drop_all_constraint_names_for_table(THD *thd, TABLE_LIST *table); bool fk_check_constraint_added(THD *thd, Foreign_key *fkey, - List &fkey_list, - const char *db); + List &fkey_list); + +bool fk_init_column_sets(TABLE_SHARE *share); + +bool fk_add_tables_for_fks(THD *thd, LEX *lex, TABLE_LIST *table); #endif === modified file 'sql/item.cc' --- a/sql/item.cc 2008-05-27 17:53:05 +0000 +++ b/sql/item.cc 2008-08-03 10:13:01 +0000 @@ -6333,44 +6333,57 @@ void Item_insert_value::print(String *st /** - Find index of Field object which will be appropriate for item - representing field of row being changed in trigger. + Find index of Field object corresponding to the item representing + field of row being changed in trigger, mark this field as used in + appropriate set. @param thd current thread context @param table table of trigger (and where we looking for fields) @param table_grant_info GRANT_INFO of the subject table + @param read_set set of fields read by triggers + @param write_set set of fields written by triggers @note - This function does almost the same as fix_fields() for Item_field - but is invoked right after trigger definition parsing. Since at - this stage we can't say exactly what Field object (corresponding - to TABLE::record[0] or TABLE::record[1]) should be bound to this - Item, we only find out index of the Field and then select concrete - Field object in fix_fields() (by that time Table_trigger_list::old_field/ - new_field should point to proper array of Fields). - It also binds Item_trigger_field to Table_triggers_list object for - table of trigger which uses this item. + This function does almost the same thing as fix_fields() for + Item_field but is invoked right after trigger definition parsing. + Since at this stage we can't say exactly what Field object + (corresponding to TABLE::record[0] or TABLE::record[1]) should be + bound to this Item, we only find out index of the Field and then + select concrete Field object in fix_fields() (by that time + Table_trigger_list::old_field/new_field should point to proper + array of Fields). We also mark this field as read/written in + sets of fields used by triggers for particular event type. + Finally we bind Item_trigger_field to Table_triggers_list object + for table of trigger which uses this item. */ void Item_trigger_field::setup_field(THD *thd, TABLE *table, - GRANT_INFO *table_grant_info) + GRANT_INFO *table_grant_info, + MY_BITMAP *read_set, + MY_BITMAP *write_set) { - /* - It is too early to mark fields used here, because before execution - of statement that will invoke trigger other statements may use same - TABLE object, so all such mark-up will be wiped out. - So instead we do it in Table_triggers_list::mark_fields_used() - method which is called during execution of these statements. - */ - enum_mark_columns save_mark_used_columns= thd->mark_used_columns; - thd->mark_used_columns= MARK_COLUMNS_NONE; - /* - Try to find field by its name and if it will be found - set field_idx properly. - */ - (void)find_field_in_table(thd, table, field_name, (uint) strlen(field_name), - 0, &field_idx); - thd->mark_used_columns= save_mark_used_columns; + Field *field_in_share; + + if ((field_in_share= find_field_in_table_def(table->s, field_name, + (uint)strlen(field_name)))) + { + field_idx= field_in_share->field_index; + if (read_set) + { + /* + Mark fields as used in sets of read/written fields for particular + event type. It is too early to mark them as used in TABLE::write_set + and read_set, because before execution of statement that will invoke + trigger other statements may use same TABLE object, so all such + mark-up will be wiped out. So instead we use sets in Table_triggers_list + and then update TABLE::write/read_set in mark_fields_used() method + which is called during execution of statements invoking triggers. + */ + bitmap_set_bit(read_set, field_idx); + if (!read_only) + bitmap_set_bit(write_set, field_idx); + } + } triggers= table->triggers; table_grants= table_grant_info; } @@ -6416,7 +6429,7 @@ bool Item_trigger_field::fix_fields(THD /* Set field. */ - if (field_idx != (uint)-1) + if (field_idx != UINT_MAX) { #ifndef NO_EMBEDDED_ACCESS_CHECKS /* === modified file 'sql/item.h' --- a/sql/item.h 2008-05-22 18:40:15 +0000 +++ b/sql/item.h 2008-08-03 10:13:01 +0000 @@ -2730,10 +2730,11 @@ public: ulong priv, const bool ro) :Item_field(context_arg, (const char *)NULL, (const char *)NULL, field_name_arg), - row_version(row_ver_arg), field_idx((uint)-1), original_privilege(priv), + row_version(row_ver_arg), field_idx(UINT_MAX), original_privilege(priv), want_privilege(priv), table_grants(NULL), read_only (ro) {} - void setup_field(THD *thd, TABLE *table, GRANT_INFO *table_grant_info); + void setup_field(THD *thd, TABLE *table, GRANT_INFO *table_grant_info, + MY_BITMAP *read_set, MY_BITMAP *write_set); enum Type type() const { return TRIGGER_FIELD_ITEM; } bool eq(const Item *item, bool binary_cmp) const; bool fix_fields(THD *, Item **); === modified file 'sql/mysql_priv.h' --- a/sql/mysql_priv.h 2008-06-23 13:26:17 +0000 +++ b/sql/mysql_priv.h 2008-08-03 10:13:01 +0000 @@ -631,6 +631,7 @@ void view_store_options(THD *thd, TABLE_ #define TL_OPTION_FORCE_INDEX 2 #define TL_OPTION_IGNORE_LEAVES 4 #define TL_OPTION_ALIAS 8 +#define TL_OPTION_DONT_ADD_DUPS 16 /* Some portable defines */ @@ -1256,7 +1257,8 @@ int prepare_create_field(Create_field *s uint *blob_columns, int *timestamps, int *timestamps_with_niladic, longlong table_flags); -bool mysql_create_table(THD *thd,const char *db, const char *table_name, +bool mysql_create_table(THD *thd, TABLE_LIST *create_table, + TABLE_LIST *fkey_tables, HA_CREATE_INFO *create_info, Alter_info *alter_info, bool tmp_table, uint select_field_count); @@ -1359,6 +1361,8 @@ find_field_in_table(THD *thd, TABLE *tab bool allow_rowid, uint *cached_field_index_ptr); Field * find_field_in_table_sef(TABLE *table, const char *name); +Field * +find_field_in_table_def(TABLE_SHARE *share, const char *name, uint length); #endif /* MYSQL_SERVER */ @@ -2194,6 +2198,7 @@ bool mysql_create_frm(THD *thd, const ch List &create_field, uint key_count,KEY *key_info, List &fkey_list, + List &parent_fkey_list, handler *db_type); int rea_create_table(THD *thd, const char *path, const char *db, const char *table_name, @@ -2201,6 +2206,7 @@ int rea_create_table(THD *thd, const cha List &create_field, uint key_count,KEY *key_info, List &fkey_list, + List &parent_fkey_list, handler *file); int format_number(uint inputflag,uint max_length,char * pos,uint length, char * *errpos); === modified file 'sql/sp.cc' --- a/sql/sp.cc 2008-06-04 11:18:52 +0000 +++ b/sql/sp.cc 2008-08-03 10:13:01 +0000 @@ -1486,9 +1486,9 @@ void sp_get_prelocking_info(THD *thd, bo the set). */ -static bool add_used_routine(LEX *lex, Query_arena *arena, - const LEX_STRING *key, - TABLE_LIST *belong_to_view) +bool sp_add_used_routine(LEX *lex, Query_arena *arena, + const LEX_STRING *key, + TABLE_LIST *belong_to_view) { hash_init_opt(&lex->sroutines, system_charset_info, Query_tables_list::START_SROUTINES_HASH_SIZE, @@ -1535,7 +1535,7 @@ void sp_add_used_routine(LEX *lex, Query sp_name *rt, char rt_type) { rt->set_routine_type(rt_type); - (void)add_used_routine(lex, arena, &rt->m_sroutines_key, 0); + (void)sp_add_used_routine(lex, arena, &rt->m_sroutines_key, 0); lex->sroutines_list_own_last= lex->sroutines_list.next; lex->sroutines_list_own_elements= lex->sroutines_list.elements; } @@ -1616,7 +1616,7 @@ sp_update_stmt_used_routines(THD *thd, L for (uint i=0 ; i < src->records ; i++) { Sroutine_hash_entry *rt= (Sroutine_hash_entry *)hash_element(src, i); - (void)add_used_routine(lex, thd->stmt_arena, &rt->key, belong_to_view); + (void)sp_add_used_routine(lex, thd->stmt_arena, &rt->key, belong_to_view); } } @@ -1640,7 +1640,7 @@ static void sp_update_stmt_used_routines { for (Sroutine_hash_entry *rt= (Sroutine_hash_entry *)src->first; rt; rt= rt->next) - (void)add_used_routine(lex, thd->stmt_arena, &rt->key, belong_to_view); + (void)sp_add_used_routine(lex, thd->stmt_arena, &rt->key, belong_to_view); } @@ -1845,8 +1845,9 @@ sp_cache_routines_and_add_tables_for_tri /* We can have only one trigger per action type currently */ sp_head *trigger= table->table->triggers->bodies[i][j]; if (trigger && - add_used_routine(lex, thd->stmt_arena, &trigger->m_sroutines_key, - table->belong_to_view)) + sp_add_used_routine(lex, thd->stmt_arena, + &trigger->m_sroutines_key, + table->belong_to_view)) { int ret; /* Sic: excludes the trigger key from processing */ === modified file 'sql/sp.h' --- a/sql/sp.h 2008-04-08 16:36:01 +0000 +++ b/sql/sp.h 2008-08-03 10:13:01 +0000 @@ -93,6 +93,9 @@ void sp_get_prelocking_info(THD *thd, bo bool *first_no_prelocking); void sp_add_used_routine(LEX *lex, Query_arena *arena, sp_name *rt, char rt_type); +bool sp_add_used_routine(LEX *lex, Query_arena *arena, + const LEX_STRING *key, + TABLE_LIST *belong_to_view); void sp_remove_not_own_routines(LEX *lex); void sp_update_sp_used_routines(HASH *dst, HASH *src); int sp_cache_routines_and_add_tables(THD *thd, LEX *lex, === modified file 'sql/sp_head.h' --- a/sql/sp_head.h 2008-05-16 13:55:54 +0000 +++ b/sql/sp_head.h 2008-08-03 10:13:01 +0000 @@ -33,6 +33,7 @@ #define TYPE_ENUM_FUNCTION 1 #define TYPE_ENUM_PROCEDURE 2 #define TYPE_ENUM_TRIGGER 3 +#define TYPE_ENUM_FOREIGN_KEY 4 Item_result sp_map_result_type(enum enum_field_types type); === modified file 'sql/sql_base.cc' --- a/sql/sql_base.cc 2008-06-25 16:40:20 +0000 +++ b/sql/sql_base.cc 2008-08-03 10:13:01 +0000 @@ -21,6 +21,7 @@ #include "sp_head.h" #include "sp.h" #include "sql_trigger.h" +#include "fk_dd.h" #include #include #include @@ -4098,24 +4099,44 @@ int open_tables(THD *thd, TABLE_LIST **s prelocking list. If we lock table for reading we won't update it so there is no need to process its triggers since they never will be activated. + + The same applies to foreign keys. If we are going to update table + participating in a foreign key relationship we need to add tables which + are going to be involved in foreign key checks/actions to the table list. */ if (thd->locked_tables_mode <= LTM_LOCK_TABLES && !thd->lex->requires_prelocking() && - tables->trg_event_map && tables->table->triggers && + tables->trg_event_map && tables->lock_type >= TL_WRITE_ALLOW_WRITE) { - if (!query_tables_last_own) - query_tables_last_own= thd->lex->query_tables_last; - if (sp_cache_routines_and_add_tables_for_triggers(thd, thd->lex, - tables)) + if (tables->table->triggers) { - /* - Serious error during reading stored routines from mysql.proc table. - Something's wrong with the table or its contents, and an error has - been emitted; we must abort. - */ - result= -1; - goto err; + if (!query_tables_last_own) + query_tables_last_own= thd->lex->query_tables_last; + if (sp_cache_routines_and_add_tables_for_triggers(thd, thd->lex, + tables)) + { + /* + Serious error during reading stored routines from mysql.proc table. + Something's wrong with the table or its contents, and an error has + been emitted; we must abort. + */ + result= -1; + goto err; + } + } + if (!tables->table->s->fkeys_child.is_empty() || + !tables->table->s->fkeys_parent.is_empty()) + { + DBUG_ASSERT(opt_fk_all_engines); + + if (!query_tables_last_own) + query_tables_last_own= thd->lex->query_tables_last; + if (fk_add_tables_for_fks(thd, thd->lex, tables)) + { + result= -1; + goto err; + } } } @@ -5314,6 +5335,46 @@ find_field_in_table(THD *thd, TABLE *tab update_field_dependencies(thd, field, table); DBUG_RETURN(field); +} + + +/** + Find field by field name in the share of the base table. + + @param share Table's share. + @param name Field name. + @param length Length of field name. + + @note Unlike find_field_in_table() requires properly initialized + TABLE_SHARE::field array and therefore does not work for + various temporary tables. Also unlike find_field_in_table() + does not have any side-effects. + + @retval non-0 - Pointer to Field object if field was found. + @retval 0 - If field was not found. +*/ + +Field* find_field_in_table_def(TABLE_SHARE *share, const char *name, + uint length) +{ + Field **field_ptr; + DBUG_ENTER("find_field_in_table_def"); + + if (share->name_hash.records) + { + if ((field_ptr= (Field**) hash_search(&share->name_hash, (uchar*)name, + length))) + DBUG_RETURN(*field_ptr); + } + else + { + for (field_ptr= share->field; *field_ptr; ++field_ptr) + { + if (!my_strcasecmp(system_charset_info, (*field_ptr)->field_name, name)) + DBUG_RETURN(*field_ptr); + } + } + DBUG_RETURN(0); } === modified file 'sql/sql_class.cc' --- a/sql/sql_class.cc 2008-06-25 19:52:25 +0000 +++ b/sql/sql_class.cc 2008-08-03 10:13:01 +0000 @@ -112,6 +112,29 @@ Key::Key(const Key &rhs, MEM_ROOT *mem_r list_copy_and_replace_each_value(columns, mem_root); } + +/** + Construct an (almost) deep copy of Foreign_key_parent object. + Only those elements that are known to never change are not copied. + If out of memory, a partial copy is returned and an error is set + in THD. +*/ + +Foreign_key_parent::Foreign_key_parent(const Foreign_key_parent &rhs, MEM_ROOT *mem_root) + : + name(rhs.name), + columns(rhs.columns, mem_root), + table(rhs.table), + ref_columns(rhs.ref_columns, mem_root), + delete_opt(rhs.delete_opt), + update_opt(rhs.update_opt), + is_within_one_db(rhs.is_within_one_db) +{ + list_copy_and_replace_each_value(columns, mem_root); + list_copy_and_replace_each_value(ref_columns, mem_root); +} + + /** Construct an (almost) deep copy of this foreign key. Only those elements that are known to never change are not copied. @@ -121,22 +144,16 @@ Key::Key(const Key &rhs, MEM_ROOT *mem_r Foreign_key::Foreign_key(const Foreign_key &rhs, MEM_ROOT *mem_root) : - name(rhs.name), - columns(rhs.columns), + Foreign_key_parent(rhs, mem_root), ref_table(rhs.ref_table), - ref_columns(rhs.ref_columns), - delete_opt(rhs.delete_opt), - update_opt(rhs.update_opt), match_opt(rhs.match_opt), fkey_clause_name(rhs.fkey_clause_name), is_table_constraint(rhs.is_table_constraint), is_name_generated(rhs.is_name_generated), - is_within_one_db(rhs.is_within_one_db), exists(rhs.exists) { - list_copy_and_replace_each_value(columns, mem_root); - list_copy_and_replace_each_value(ref_columns, mem_root); } + /* Test if a foreign key (= generated key) is a prefix of the given key === modified file 'sql/sql_class.h' --- a/sql/sql_class.h 2008-06-25 19:52:25 +0000 +++ b/sql/sql_class.h 2008-08-03 10:13:01 +0000 @@ -237,18 +237,74 @@ public: }; -class Foreign_key: public Sql_alloc { -public: - enum fk_match_opt { FK_MATCH_UNDEF, FK_MATCH_FULL, - FK_MATCH_PARTIAL, FK_MATCH_SIMPLE}; - enum fk_option { FK_OPTION_UNDEF, FK_OPTION_RESTRICT, FK_OPTION_CASCADE, - FK_OPTION_SET_NULL, FK_OPTION_NO_ACTION, FK_OPTION_DEFAULT}; +enum fk_match_opt { FK_MATCH_UNDEF, FK_MATCH_FULL, + FK_MATCH_PARTIAL, FK_MATCH_SIMPLE}; +enum fk_option { FK_OPTION_UNDEF, FK_OPTION_RESTRICT, FK_OPTION_CASCADE, + FK_OPTION_SET_NULL, FK_OPTION_NO_ACTION, FK_OPTION_DEFAULT}; + + +/** + Class that represents foreign key in the structures describing its + parent table in CREATE TABLE/ALTER TABLE statements. Contains all + information about foreign key which should be saved in .FRM of its + parent table. +*/ +class Foreign_key_parent: public Sql_alloc { +public: LEX_STRING name; List columns; - TABLE_LIST *ref_table; + TABLE_LIST *table; List ref_columns; - uint delete_opt, update_opt, match_opt; + fk_option delete_opt, update_opt; + /** + Indicates whether child and parent table of foreign key are in + the same database. + */ + bool is_within_one_db; + + Foreign_key_parent(const LEX_STRING &name_arg, + TABLE_LIST *tab, const List &cols, + const List &ref_cols, + fk_option delete_opt_arg, + fk_option update_opt_arg, + bool within_one_db_arg) + : name(name_arg), columns(cols), table(tab), ref_columns(ref_cols), + delete_opt(delete_opt_arg), update_opt(update_opt_arg), + is_within_one_db(within_one_db_arg) + {} + + Foreign_key_parent(const Foreign_key_parent &rhs, MEM_ROOT *mem_root); + /** + Used to make a clone of this object for ALTER/CREATE TABLE + @sa comment for Key_part_spec::clone + */ + Foreign_key_parent *clone(MEM_ROOT *mem_root) const + { return new (mem_root) Foreign_key_parent(*this, mem_root); } + + virtual ~Foreign_key_parent() {}; + + uint get_frm_description_length(); + void save_to_frm(char **buff_p); +protected: + virtual uint get_additional_fixed_part_length(); + virtual char get_additional_flags(); + virtual void save_additional_fixed_part(char **buff_p); + virtual TABLE_LIST* get_table_for_frm(); +}; + + +/** + Class that represents foreign key in CREATE TABLE/ALTER TABLE + statements. Contains all information about foreign key which + is needed for its creation (and for saving its description + in .FRM file of its child table). +*/ + +class Foreign_key: public Foreign_key_parent { +public: + TABLE_LIST *ref_table; + fk_match_opt match_opt; /** Name of foreign key which was specified in non-standard way in FOREIGN KEY clause. @@ -266,59 +322,37 @@ public: */ bool is_name_generated; /** - Indicates whether child and parent table of foreign key are in - the same database. - */ - bool is_within_one_db; - /** Indicates that this object represents already existing constraint so we don't need to perform some of checks and actions for it when saving its description to .FRM file. */ bool exists; + Foreign_key(const LEX_STRING &name_arg, const LEX_STRING &fkey_clause_name_arg, - bool table_con_arg, List &cols, - TABLE_LIST *table, List &ref_cols, - uint delete_opt_arg, uint update_opt_arg, uint match_opt_arg) - : name(name_arg), columns(cols), - ref_table(table), ref_columns(ref_cols), - delete_opt(delete_opt_arg), update_opt(update_opt_arg), - match_opt(match_opt_arg), fkey_clause_name(fkey_clause_name_arg), - is_table_constraint(table_con_arg), is_name_generated(FALSE), - is_within_one_db(FALSE), exists(FALSE) + bool table_con_arg, + TABLE_LIST *tab, const List &cols, + TABLE_LIST *ref_tab, const List &ref_cols, + fk_option delete_opt_arg, fk_option update_opt_arg, + fk_match_opt match_opt_arg, bool within_one_db_arg, + bool exists_arg) + : Foreign_key_parent(name_arg, tab, cols, ref_cols, delete_opt_arg, + update_opt_arg, within_one_db_arg), + ref_table(ref_tab), match_opt(match_opt_arg), + fkey_clause_name(fkey_clause_name_arg), + is_table_constraint(table_con_arg), is_name_generated(FALSE), + exists(exists_arg) {} - Foreign_key(const char *name_arg, size_t name_len_arg, bool table_con_arg, - bool within_one_db_arg, List &cols, - TABLE_LIST *table, List &ref_cols, - uint delete_opt_arg, uint update_opt_arg, - uint match_opt_arg, bool exists_arg) - : columns(cols), - ref_table(table), ref_columns(ref_cols), - delete_opt(delete_opt_arg), update_opt(update_opt_arg), - match_opt(match_opt_arg), - is_table_constraint(table_con_arg), is_name_generated(FALSE), - is_within_one_db(within_one_db_arg), exists(exists_arg) - { - name.str= (char *)name_arg; - name.length= name_len_arg; - fkey_clause_name.str= NULL; - fkey_clause_name.length= 0; - } Foreign_key(const Foreign_key &rhs, MEM_ROOT *mem_root); - /** - Used to make a clone of this object for ALTER/CREATE TABLE - @sa comment for Key_part_spec::clone - */ Foreign_key *clone(MEM_ROOT *mem_root) const { return new (mem_root) Foreign_key(*this, mem_root); } /** If needed set values for MATCH and ON UPDATE/DELETE clauses to their default value, also set value of is_within_one_db - attribute.. + attribute. */ - void set_missing_options(const char *db) + void set_missing_options() { if (match_opt == FK_MATCH_UNDEF) match_opt= FK_MATCH_SIMPLE; @@ -327,20 +361,101 @@ public: if (delete_opt == FK_OPTION_UNDEF) delete_opt= FK_OPTION_NO_ACTION; - if (!strcmp(db, ref_table->db)) + if (!strcmp(table->db, ref_table->db)) is_within_one_db= TRUE; } - uint get_frm_description_length(); - void save_to_frm(char **buff_p); - static Foreign_key* restore_from_frm(THD *thd, MEM_ROOT *mem_root, - const LEX_STRING &db, - const char **buff_p); - - void get_create_statement_clause(THD *thd, String *str, - const char *db); void generate_name_if_needed(THD *thd, const char *table_name); + +protected: + virtual uint get_additional_fixed_part_length(); + virtual char get_additional_flags(); + virtual void save_additional_fixed_part(char **buff_p); + virtual TABLE_LIST* get_table_for_frm(); +}; + + +/** + Base class for classes representing foreign keys in table's share. +*/ + +class Foreign_key_share: public Sql_alloc +{ +public: + LEX_STRING name; + List columns; + List ref_columns; + fk_option delete_opt, update_opt; + bool is_within_one_db; + /** + Set of columns in the table participating in foreign key. + Note that it is always set of columns in the table for + which this Foreign_key_share object represents a constraint. + I.e. for Foreign_key_child_share it is set of referencing + columns and for Foreign_key_parent_share it is set of columns + being referenced. + */ + MY_BITMAP column_set; + + virtual ~Foreign_key_share() {}; + + bool restore_from_frm(THD *thd, TABLE_SHARE *share, const char **buff_p); + + bool init_column_set(TABLE_SHARE *share); + +protected: + virtual void restore_additional_flags(char flags) = 0; + virtual void restore_from_additional_fixed_part(const char **buff_p) = 0; + virtual void set_table_from_frm(LEX_STRING &db, LEX_STRING &table_name) = 0; + virtual List& get_table_columns() = 0; }; + + +/** + Class that represents foreign key in the share of table which + participates in this foreign key as child. +*/ + +class Foreign_key_child_share: public Foreign_key_share +{ +public: + LEX_STRING parent_table_db; + LEX_STRING parent_table_name; + fk_match_opt match_opt; + bool is_table_constraint; + + void get_create_statement_clause(THD *thd, String *str); + + Foreign_key *make_foreign_key(MEM_ROOT *mem_root, TABLE_LIST *table); + +protected: + virtual void restore_additional_flags(char flags); + virtual void restore_from_additional_fixed_part(const char **buff_p); + virtual void set_table_from_frm(LEX_STRING &db, LEX_STRING &table_name); + virtual List& get_table_columns(); +}; + + +/** + Class that represents foreign key in the share of table which + participates in this foreign key as parent. +*/ + +class Foreign_key_parent_share : public Foreign_key_share +{ +public: + LEX_STRING child_table_db; + LEX_STRING child_table_name; + + Foreign_key_parent *make_foreign_key(MEM_ROOT *mem_root); + +protected: + virtual void restore_additional_flags(char flags); + virtual void restore_from_additional_fixed_part(const char **buff_p); + virtual void set_table_from_frm(LEX_STRING &db, LEX_STRING &table_name); + virtual List& get_table_columns(); +}; + typedef struct st_mysql_lock { === modified file 'sql/sql_lex.cc' --- a/sql/sql_lex.cc 2008-06-27 05:11:56 +0000 +++ b/sql/sql_lex.cc 2008-08-03 10:13:01 +0000 @@ -1496,6 +1496,7 @@ Alter_info::Alter_info(const Alter_info alter_list(rhs.alter_list, mem_root), key_list(rhs.key_list, mem_root), foreign_key_list(rhs.foreign_key_list, mem_root), + parent_foreign_key_list(rhs.parent_foreign_key_list, mem_root), create_list(rhs.create_list, mem_root), flags(rhs.flags), keys_onoff(rhs.keys_onoff), @@ -1519,6 +1520,7 @@ Alter_info::Alter_info(const Alter_info list_copy_and_replace_each_value(alter_list, mem_root); list_copy_and_replace_each_value(key_list, mem_root); list_copy_and_replace_each_value(foreign_key_list, mem_root); + list_copy_and_replace_each_value(parent_foreign_key_list, mem_root); list_copy_and_replace_each_value(create_list, mem_root); /* partition_names are not deeply copied currently */ } @@ -2775,10 +2777,11 @@ void st_lex::link_first_table_back(TABLE @note We assume that the first table in outer select and global list exist and are the same. - @return Head of the list with unlinked tables. + @param link Pointer to the head of the list to which unlinked + tables should be attached. */ -TABLE_LIST *st_lex::unlink_all_tables_except_first() +void st_lex::unlink_all_tables_except_first(TABLE_LIST **link) { TABLE_LIST *first_table= select_lex.get_table_list(); TABLE_LIST *rest_tables= first_table->next_local; @@ -2790,7 +2793,10 @@ TABLE_LIST *st_lex::unlink_all_tables_ex DBUG_ASSERT(first_table == query_tables); first_table->next_global= 0; query_tables_last= &(first_table->next_global); - return rest_tables; + /* Attach removed tables to the specified list. */ + if (rest_tables) + rest_tables->prev_global= link; + *link= rest_tables; } === modified file 'sql/sql_lex.h' --- a/sql/sql_lex.h 2008-06-27 05:11:56 +0000 +++ b/sql/sql_lex.h 2008-08-03 10:13:01 +0000 @@ -883,6 +883,7 @@ public: List alter_list; List key_list; List foreign_key_list; + List parent_foreign_key_list; List create_list; uint flags; enum enum_enable_or_disable keys_onoff; @@ -910,6 +911,7 @@ public: alter_list.empty(); key_list.empty(); foreign_key_list.empty(); + parent_foreign_key_list.empty(); create_list.empty(); flags= 0; keys_onoff= LEAVE_AS_IS; @@ -1643,9 +1645,9 @@ typedef struct st_lex : public Query_tab enum ha_storage_media storage_type; enum column_format_type column_format; uint grant, grant_tot_col, which_columns; - enum Foreign_key::fk_match_opt fk_match_option; - enum Foreign_key::fk_option fk_update_opt; - enum Foreign_key::fk_option fk_delete_opt; + enum fk_match_opt fk_match_option; + enum fk_option fk_update_opt; + enum fk_option fk_delete_opt; uint slave_thd_opt, start_transaction_opt; int nest_level; /* @@ -1680,6 +1682,8 @@ typedef struct st_lex : public Query_tab bool subqueries, ignore; st_parsing_options parsing_options; Alter_info alter_info; + /* List of tables referenced by foreign keys in the table being created. */ + TABLE_LIST *create_foreign_key_tables; /* Prepared statements SQL syntax:*/ LEX_STRING prepared_stmt_name; /* Statement name (in all queries) */ /* @@ -1793,7 +1797,7 @@ typedef struct st_lex : public Query_tab TABLE_LIST *unlink_first_table(bool *link_to_local); void link_first_table_back(TABLE_LIST *first, bool link_to_local); void first_lists_tables_same(); - TABLE_LIST *unlink_all_tables_except_first(); + void unlink_all_tables_except_first(TABLE_LIST **link); bool can_be_merged(); bool can_use_merged(); === modified file 'sql/sql_list.h' --- a/sql/sql_list.h 2008-03-28 17:47:53 +0000 +++ b/sql/sql_list.h 2008-08-03 10:13:01 +0000 @@ -214,6 +214,10 @@ public: swap_variables(list_node *, first, rhs.first); swap_variables(list_node **, last, rhs.last); swap_variables(uint, elements, rhs.elements); + if (is_empty()) + last= &first; + if (rhs.is_empty()) + rhs.last= &rhs.first; } inline list_node* last_node() { return *last; } inline list_node* first_node() { return first;} === modified file 'sql/sql_parse.cc' --- a/sql/sql_parse.cc 2008-06-23 13:26:17 +0000 +++ b/sql/sql_parse.cc 2008-08-03 10:13:01 +0000 @@ -2437,9 +2437,9 @@ mysql_execute_command(THD *thd) &create_info); else { - res= mysql_create_table(thd, create_table->db, - create_table->table_name, &create_info, - &alter_info, 0, 0); + res= mysql_create_table(thd, create_table, + lex->create_foreign_key_tables, + &create_info, &alter_info, 0, 0); } if (!res) my_ok(thd); @@ -6009,6 +6009,9 @@ bool add_to_list(THD *thd, SQL_LIST &lis - TL_OPTION_UPDATING : Table will be updated - TL_OPTION_FORCE_INDEX : Force usage of index - TL_OPTION_ALIAS : an alias in multi table DELETE + - TL_OPTION_DONT_ADD_DUPS : don't add table if it is + already present in the table list, return pointer + to already existing table list element. @param lock_type How table should be locked @param use_index List of indexed used in USE INDEX @param ignore_index List of indexed used in IGNORE INDEX @@ -6029,7 +6032,9 @@ TABLE_LIST *st_select_lex::add_table_to_ { register TABLE_LIST *ptr; TABLE_LIST *previous_table_ref; /* The table preceding the current one. */ - char *alias_str; + char *alias_str, *db; + size_t db_length; + bool is_fqtn; LEX *lex= thd->lex; DBUG_ENTER("add_table_to_list"); LINT_INIT(previous_table_ref); @@ -6062,6 +6067,37 @@ TABLE_LIST *st_select_lex::add_table_to_ if (!(alias_str= (char*) thd->memdup(alias_str,table->table.length+1))) DBUG_RETURN(0); } + + if (lower_case_table_names && table->table.length) + table->table.length= my_casedn_str(files_charset_info, table->table.str); + + if (table->db.str) + { + is_fqtn= TRUE; + db= table->db.str; + db_length= table->db.length; + } + else if (lex->copy_db_to(&db, &db_length)) + DBUG_RETURN(0); + else + is_fqtn= FALSE; + + if (test(table_options & TL_OPTION_DONT_ADD_DUPS)) + { + /* + Once we start using this option with lock types other than TL_IGNORE + we might want to adjust lock types for elements already present in + the table list. + */ + DBUG_ASSERT(lock_type == TL_IGNORE); + for(ptr= lex->query_tables; ptr; ptr= ptr->next_global) + { + if (!strcmp(db, ptr->db) && + !strcmp(table->table.str, ptr->table_name)) + DBUG_RETURN(ptr); + } + } + if (!(ptr = (TABLE_LIST *) thd->calloc(sizeof(TABLE_LIST)))) DBUG_RETURN(0); /* purecov: inspected */ if (table->db.str) @@ -6075,10 +6111,11 @@ TABLE_LIST *st_select_lex::add_table_to_ else ptr->is_fqtn= FALSE; + ptr->db= db; + ptr->db_length= db_length; + ptr->is_fqtn= is_fqtn; ptr->alias= alias_str; ptr->is_alias= alias ? TRUE : FALSE; - if (lower_case_table_names && table->table.length) - table->table.length= my_casedn_str(files_charset_info, table->table.str); ptr->table_name=table->table.str; ptr->table_name_length=table->table.length; ptr->lock_type= lock_type; === modified file 'sql/sql_show.cc' --- a/sql/sql_show.cc 2008-06-23 13:26:17 +0000 +++ b/sql/sql_show.cc 2008-08-03 10:13:01 +0000 @@ -4784,9 +4784,9 @@ static int get_schema_constraints_record if (opt_fk_all_engines) { - List_iterator fkey_it(show_table->s->fkeys); - Foreign_key *fkey; - while ((fkey= fkey_it++)) + List_iterator fk_it(show_table->s->fkeys_child); + Foreign_key_child_share *fkey; + while ((fkey= fk_it++)) { if (store_constraints(thd, table, db_name, table_name, fkey->name.str, fkey->name.length, @@ -4986,13 +4986,12 @@ static int get_schema_key_column_usage_r if (opt_fk_all_engines) { - List_iterator fkey_it(show_table->s->fkeys); - Foreign_key *fkey; - while ((fkey= fkey_it++)) - { - List_iterator it(fkey->columns); - List_iterator it1(fkey->ref_columns); - Key_part_spec *f_col, *r_col; + List_iterator fk_it(show_table->s->fkeys_child); + Foreign_key_child_share *fkey; + while ((fkey= fk_it++)) + { + List_iterator it(fkey->columns), it1(fkey->ref_columns); + LEX_STRING *f_col, *r_col; uint f_idx= 0; while ((f_col= it++)) @@ -5002,20 +5001,20 @@ static int get_schema_key_column_usage_r restore_record(table, s->default_values); store_key_column_usage(table, db_name, table_name, fkey->name.str, fkey->name.length, - f_col->field_name.str, f_col->field_name.length, + f_col->str, f_col->length, (longlong) f_idx); table->field[8]->store((longlong) f_idx, TRUE); table->field[8]->set_notnull(); - table->field[9]->store(fkey->ref_table->db, - fkey->ref_table->db_length, + table->field[9]->store(fkey->parent_table_db.str, + fkey->parent_table_db.length, system_charset_info); table->field[9]->set_notnull(); - table->field[10]->store(fkey->ref_table->table_name, - fkey->ref_table->table_name_length, + table->field[10]->store(fkey->parent_table_name.str, + fkey->parent_table_name.length, system_charset_info); table->field[10]->set_notnull(); - table->field[11]->store(r_col->field_name.str, - r_col->field_name.length, + table->field[11]->store(r_col->str, + r_col->length, system_charset_info); table->field[11]->set_notnull(); if (schema_table_store_record(thd, table)) @@ -5732,15 +5731,15 @@ get_referential_constraints_record(THD * TABLE *show_table= tables->table; if (opt_fk_all_engines) { - List_iterator fkey_it(show_table->s->fkeys); - Foreign_key *fkey; - while ((fkey= fkey_it++)) + List_iterator fk_it(show_table->s->fkeys_child); + Foreign_key_child_share *fkey; + while ((fkey= fk_it++)) { restore_record(table, s->default_values); table->field[1]->store(db_name->str, db_name->length, cs); table->field[2]->store(fkey->name.str, fkey->name.length, cs); - table->field[4]->store(fkey->ref_table->db, - fkey->ref_table->db_length, cs); + table->field[4]->store(fkey->parent_table_db.str, + fkey->parent_table_db.length, cs); #ifdef WILL_BE_FILLED_ONCE_WE_WILL_REACH_LATER_MILESTONES_OF_WL148 table->field[5]->store(referenced_key_name.str, referenced_key_name.length, cs); @@ -5754,8 +5753,8 @@ get_referential_constraints_record(THD * table->field[8]->store(fk_ref_action_names[fkey->delete_opt].str, fk_ref_action_names[fkey->delete_opt].length, cs); table->field[9]->store(table_name->str, table_name->length, cs); - table->field[10]->store(fkey->ref_table->table_name, - fkey->ref_table->table_name_length, cs); + table->field[10]->store(fkey->parent_table_name.str, + fkey->parent_table_name.length, cs); if (schema_table_store_record(thd, table)) DBUG_RETURN(1); } === modified file 'sql/sql_table.cc' --- a/sql/sql_table.cc 2008-06-25 19:52:25 +0000 +++ b/sql/sql_table.cc 2008-08-03 10:13:01 +0000 @@ -44,15 +44,14 @@ static int copy_data_between_tables(TABL static bool prepare_blob_field(THD *thd, Create_field *sql_field); static bool check_engine(THD *, const char *, HA_CREATE_INFO *); static int -mysql_prepare_create_table(THD *thd, const char *db, - HA_CREATE_INFO *create_info, +mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info, Alter_info *alter_info, bool tmp_table, uint *db_options, handler *file, KEY **key_info_buffer, uint *key_count, int select_field_count); static bool -mysql_prepare_alter_table(THD *thd, TABLE *table, +mysql_prepare_alter_table(THD *thd, TABLE_LIST *table_list, HA_CREATE_INFO *create_info, Alter_info *alter_info); @@ -1272,7 +1271,7 @@ bool mysql_write_frm(ALTER_PARTITION_PAR strxmov(shadow_frm_name, shadow_path, reg_ext, NullS); if (flags & WFRM_WRITE_SHADOW) { - if (mysql_prepare_create_table(lpt->thd, lpt->db, lpt->create_info, + if (mysql_prepare_create_table(lpt->thd, lpt->create_info, lpt->alter_info, /*tmp_table*/ 1, &lpt->db_options, @@ -1306,6 +1305,7 @@ bool mysql_write_frm(ALTER_PARTITION_PAR lpt->alter_info->create_list, lpt->key_count, lpt->key_info_buffer, lpt->alter_info->foreign_key_list, + lpt->alter_info->parent_foreign_key_list, lpt->table->file)) || lpt->table->file->ha_create_handler_files(shadow_path, NULL, CHF_CREATE_FLAG, @@ -2238,7 +2238,6 @@ int prepare_create_field(Create_field *s SYNOPSIS mysql_prepare_create_table() thd Thread object. - db Name of database for table to being created. create_info Create information (like MAX_ROWS). alter_info List of columns and indexes to create tmp_table If a temporary table is to be created. @@ -2260,7 +2259,7 @@ int prepare_create_field(Create_field *s */ static int -mysql_prepare_create_table(THD *thd, const char *db, +mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info, Alter_info *alter_info, bool tmp_table, @@ -2624,12 +2623,9 @@ mysql_prepare_create_table(THD *thd, con fkey->name.str : "(none)")); fk_key_count++; - fkey->set_missing_options(db); - if (opt_fk_all_engines) { - if (fk_check_constraint_added(thd, fkey, alter_info->foreign_key_list, - db)) + if (fk_check_constraint_added(thd, fkey, alter_info->foreign_key_list)) DBUG_RETURN(TRUE); } else @@ -3554,7 +3550,7 @@ bool mysql_create_table_no_lock(THD *thd set_table_default_charset(thd, create_info, (char*) db); - if (mysql_prepare_create_table(thd, db, create_info, alter_info, + if (mysql_prepare_create_table(thd, create_info, alter_info, internal_tmp_table, &db_options, file, &key_info_buffer, &key_count, @@ -3697,7 +3693,9 @@ bool mysql_create_table_no_lock(THD *thd if (rea_create_table(thd, path, db, table_name, create_info, alter_info->create_list, key_count, key_info_buffer, - alter_info->foreign_key_list, file)) + alter_info->foreign_key_list, + alter_info->parent_foreign_key_list, + file)) goto unlock_and_end; if (create_info->options & HA_LEX_CREATE_TMP_TABLE) @@ -3784,24 +3782,274 @@ static bool lock_table_name_if_not_cache } -/* - Database locking aware wrapper for mysql_create_table_no_lock(), +/** + Auxiliary function which creates new version of .FRM for table + which is referenced via foreign key from table being created. + + @param thd Thread context + @param table Table for which we need new .FRM + @param dst_path Name for the new .FRM + @param fk_list List of foreign keys some of which reference + this table. + + @retval FALSE Success + @retval TRUE Failure */ -bool mysql_create_table(THD *thd, const char *db, const char *table_name, +static bool create_frm_with_ref_fks(THD *thd, TABLE_LIST *table, + char *dst_path, + List &fk_list) +{ + HA_CREATE_INFO create_info; + Alter_info alter_info; + KEY *key_info_buffer; + uint db_options= 0, key_count; + DBUG_ENTER("create_frm_with_ref_fks"); + + bzero((char*) &create_info, sizeof(create_info)); + create_info.db_type= table->table->s->db_type(); + create_info.row_type= table->table->s->row_type; + alter_info.flags= (ALTER_CHANGE_COLUMN | ALTER_RECREATE); + table->table->use_all_columns(); + if (mysql_prepare_alter_table(thd, table, &create_info, &alter_info)) + DBUG_RETURN(TRUE); + + List_iterator fkey_it(fk_list); + Foreign_key *fkey; + Foreign_key_parent *fkey_p; + + /* + Select foreign keys which reference this table and add their + descriptions to the list foreign keys for this table. + */ + while ((fkey= fkey_it++)) + { + if (fkey->ref_table == table) + { + if (!(fkey_p= new (thd->mem_root) Foreign_key_parent(*fkey, thd->mem_root))) + DBUG_RETURN(TRUE); + if (alter_info.parent_foreign_key_list.push_back(fkey_p, thd->mem_root)) + DBUG_RETURN(TRUE); + } + } + + if (mysql_prepare_create_table(thd, &create_info, + &alter_info, + FALSE, &db_options, + table->table->file, + &key_info_buffer, + &key_count, 0)) + DBUG_RETURN(TRUE); + + if (mysql_create_frm(thd, dst_path, NullS, NullS, + &create_info, alter_info.create_list, + key_count, key_info_buffer, + alter_info.foreign_key_list, + alter_info.parent_foreign_key_list, + table->table->file)) + DBUG_RETURN(TRUE); + DBUG_RETURN(FALSE); +} + + +/** + Prepare for creation of table with foreign keys by opening referenced + tables and adjusting definition of this table if has self-references. + + @param thd Thread context + @param create_table Table being created + @param fkey_tables List of referenced tables + @param alter_info Alter_info for the table being created + + @retval FALSE Success + @retval TRUE Failure +*/ + +static bool prepare_create_table_with_fks(THD *thd, + TABLE_LIST *create_table, + TABLE_LIST *fkey_tables, + Alter_info *alter_info) +{ + List_iterator fkey_it(alter_info->foreign_key_list); + TABLE_LIST *tab; + Foreign_key *fkey; + Foreign_key_parent *fkey_p; + + /* + Open referenced tables so they are ready for checking if we can + create foreign key referencing them and also for creation of new + version of .FRM file. + */ + for (tab= fkey_tables; tab; tab= tab->next_global) + { + enum enum_open_table_action ot_action_unused; + if (open_table(thd, tab, thd->mem_root, &ot_action_unused, + MYSQL_OPEN_REOPEN)) + return TRUE; + } + + /* + Foreign keys for which table being created serves as both child and + parent should present in table description twice. First as foreign + key on this child table and second as foreign key which references + to this parent table. + */ + while ((fkey= fkey_it++)) + { + if (fkey->ref_table == create_table) + { + if (!(fkey_p= new (thd->mem_root) Foreign_key_parent(*fkey, + thd->mem_root)) || + alter_info->parent_foreign_key_list.push_back(fkey_p, thd->mem_root)) + return TRUE; + } + } + + return FALSE; +} + + +/** + Finalize creation of table with foreign keys by updating .FRM files + of referenced tables and creating constraint names. + + @param thd Thread context + @param create_table Table being created + @param fkey_tables List of referenced tables + @param alter_info Alter_info for the table being created + + @retval FALSE Success + @retval TRUE Failure +*/ + +static bool finish_create_table_with_fks(THD *thd, + TABLE_LIST *create_table, + TABLE_LIST *fkey_tables, + Alter_info *alter_info) +{ + /* + We construct temporary name for new version of .FRM of each parent + table from fixed '#sql---' part and variable + '' part. So we put fixed part in the buffer for + name only once and then keep changing variable part of the name. + We use same approach for names of old versions of .FRMs. + */ + char new_ver_name[NAME_LEN+1], // Buffer for temporary names for new .FRMs + old_ver_name[NAME_LEN+1], // Buffer for temporary names for old .FRMs + dst_path[FN_REFLEN]; + uint new_ver_var_part, // Offset of variable part in names for new .FRMs + old_ver_var_part; // Offset of variable part in names for old .FRMs + TABLE_LIST *tab, *err_tab; + ulong i; + + new_ver_var_part= my_snprintf(new_ver_name, sizeof(new_ver_name), + "%s-%lx_%lx_", tmp_file_prefix, + current_pid, thd->thread_id); + old_ver_var_part= my_snprintf(old_ver_name, sizeof(new_ver_name), + "%s2-%lx_%lx_", tmp_file_prefix, + current_pid, thd->thread_id); + + if (lower_case_table_names) + new_ver_var_part= my_casedn_str(files_charset_info, new_ver_name); + if (lower_case_table_names) + old_ver_var_part= my_casedn_str(files_charset_info, old_ver_name); + + for (tab= fkey_tables, i= 0; tab; tab= tab->next_global, i++) + { + int10_to_str(i, new_ver_name + new_ver_var_part, 10); + build_table_filename(dst_path, sizeof(dst_path), tab->db, + new_ver_name, reg_ext, FN_IS_TMP); + if (create_frm_with_ref_fks(thd, tab, dst_path, + alter_info->foreign_key_list)) + { + err_tab= tab; + goto error_remove_new_frm; + } + } + + for (tab= fkey_tables, i= 0; tab; tab= tab->next_global, i++) + { + close_all_tables_for_name(thd, tab->table->s, FALSE); + pthread_mutex_lock(&LOCK_open); + tdc_remove_table(thd, TDC_RT_REMOVE_ALL, tab->db, tab->table_name); + pthread_mutex_unlock(&LOCK_open); + + int10_to_str(i, new_ver_name + new_ver_var_part, 10); + int10_to_str(i, old_ver_name + old_ver_var_part, 10); + + if (mysql_rename_table(NULL, tab->db, tab->table_name, tab->db, + old_ver_name, FN_TO_IS_TMP)) + { + err_tab= tab; + goto error_revert_renames; + } + if (mysql_rename_table(NULL, tab->db, new_ver_name, tab->db, + tab->table_name, FN_FROM_IS_TMP)) + { + (void) mysql_rename_table(NULL, tab->db, old_ver_name, tab->db, + tab->table_name, FN_FROM_IS_TMP); + err_tab= tab; + goto error_revert_renames; + } + } + + if (fk_create_constraint_names(alter_info->foreign_key_list, + create_table->db)) + { + err_tab= 0; + goto error_revert_renames; + } + + for (tab= fkey_tables, i= 0; tab; tab= tab->next_global, i++) + { + int10_to_str(i, old_ver_name + old_ver_var_part, 10); + (void) quick_rm_table(NULL, tab->db, old_ver_name, FN_IS_TMP); + } + return FALSE; + +error_revert_renames: + for (tab= fkey_tables, i= 0; tab != err_tab; tab= tab->next_global, i++) + { + int10_to_str(i, new_ver_name + new_ver_var_part, 10); + int10_to_str(i, old_ver_name + old_ver_var_part, 10); + (void) mysql_rename_table(NULL, tab->db, tab->table_name, tab->db, + new_ver_name, FN_TO_IS_TMP); + (void) mysql_rename_table(NULL, tab->db, old_ver_name, tab->db, + tab->table_name, FN_FROM_IS_TMP); + } + err_tab= 0; + +error_remove_new_frm: + for (tab= fkey_tables, i= 0; tab != err_tab; tab= tab->next_global, i++) + { + int10_to_str(i, new_ver_name + new_ver_var_part, 10); + (void) quick_rm_table(NULL, tab->db, new_ver_name, FN_IS_TMP); + } + return TRUE; +} + + +/** + Database locking and foreign keys aware wrapper for + mysql_create_table_no_lock(). +*/ + +bool mysql_create_table(THD *thd, TABLE_LIST *create_table, + TABLE_LIST *fkey_tables, HA_CREATE_INFO *create_info, Alter_info *alter_info, bool internal_tmp_table, uint select_field_count) { - MDL_LOCK_DATA *target_lock_data= 0; + TABLE_LIST *tab; bool result; DBUG_ENTER("mysql_create_table"); /* Wait for any database locks */ pthread_mutex_lock(&LOCK_lock_db); while (!thd->killed && - hash_search(&lock_db_cache,(uchar*) db, strlen(db))) + hash_search(&lock_db_cache, (uchar*)create_table->db, + create_table->db_length)) { wait_for_condition(thd, &LOCK_lock_db, &COND_refresh); pthread_mutex_lock(&LOCK_lock_db); @@ -3817,47 +4065,78 @@ bool mysql_create_table(THD *thd, const if (!(create_info->options & HA_LEX_CREATE_TMP_TABLE)) { - if (lock_table_name_if_not_cached(thd, db, table_name, &target_lock_data)) + if (thd->locked_tables_mode) { + /* + Since under LOCK TABLES attempt to acquire exclusive metadata lock + on the table to be created (and thus not locked) can lead to deadlock + we prohibit creation of non-temporary tables under LOCK TABLES. + */ + my_error(ER_LOCK_OR_ACTIVE_TRANSACTION, MYF(0)); result= TRUE; - goto unlock; + goto unlock_db; } - if (!target_lock_data) + + if (opt_fk_all_engines) + fk_generate_constraint_names(thd, create_table->table_name, + alter_info->foreign_key_list); + + mdl_set_lock_type(create_table->mdl_lock_data, MDL_EXCLUSIVE); + mdl_add_lock(&thd->mdl_context, create_table->mdl_lock_data); + + if (opt_fk_all_engines) { - if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS) + for (tab= fkey_tables; tab; tab= tab->next_global) { - push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_NOTE, - ER_TABLE_EXISTS_ERROR, ER(ER_TABLE_EXISTS_ERROR), - table_name); - create_info->table_existed= 1; - result= FALSE; + mdl_set_lock_type(tab->mdl_lock_data, MDL_EXCLUSIVE); + mdl_add_lock(&thd->mdl_context, tab->mdl_lock_data); } - else - { - my_error(ER_TABLE_EXISTS_ERROR,MYF(0),table_name); - result= TRUE; - } - goto unlock; } - } - if (opt_fk_all_engines) - fk_generate_constraint_names(thd, table_name, alter_info->foreign_key_list); + if (mdl_acquire_exclusive_locks(&thd->mdl_context)) + { + mdl_remove_all_locks(&thd->mdl_context); + result= TRUE; + goto unlock_db; + } + + if (opt_fk_all_engines && + prepare_create_table_with_fks(thd, create_table, fkey_tables, + alter_info)) + { + result= TRUE; + goto close_and_unlock; + } + } - result= mysql_create_table_no_lock(thd, db, table_name, create_info, + result= mysql_create_table_no_lock(thd, create_table->db, + create_table->table_name, + create_info, alter_info, internal_tmp_table, select_field_count); - if (!result && opt_fk_all_engines) - result= fk_create_constraint_names(alter_info->foreign_key_list, db); + if (opt_fk_all_engines && + !(create_info->options & HA_LEX_CREATE_TMP_TABLE) && + !result && !create_info->table_existed && + finish_create_table_with_fks(thd, create_table, fkey_tables, alter_info)) + { + (void) quick_rm_table(create_info->db_type, create_table->db, + create_table->table_name, 0); + result= TRUE; + } -unlock: - if (target_lock_data) +close_and_unlock: + if (!(create_info->options & HA_LEX_CREATE_TMP_TABLE)) { - mdl_release_lock(&thd->mdl_context, target_lock_data); - mdl_remove_lock(&thd->mdl_context, target_lock_data); + /* + close_thread_tables() takes care about both closing open tables (which + might be still around in case of error) and releasing metadata locks. + */ + close_thread_tables(thd); } + +unlock_db: pthread_mutex_lock(&LOCK_lock_db); if (!--creating_table && creating_database) pthread_cond_signal(&COND_refresh); @@ -4758,7 +5037,6 @@ bool mysql_preload_keys(THD* thd, TABLE_ @param[in] thd thread handler @param[in] schema_table I_S table - @param[in] db database for table being created @param[in] dst_path path where frm should be created @param[in] create_info Create info @@ -4770,7 +5048,6 @@ bool mysql_preload_keys(THD* thd, TABLE_ bool mysql_create_like_schema_frm(THD* thd, TABLE_LIST* schema_table, - const char *db, char *dst_path, HA_CREATE_INFO *create_info) { HA_CREATE_INFO local_create_info; @@ -4786,10 +5063,10 @@ bool mysql_create_like_schema_frm(THD* t local_create_info.default_table_charset=default_charset_info; alter_info.flags= (ALTER_CHANGE_COLUMN | ALTER_RECREATE); schema_table->table->use_all_columns(); - if (mysql_prepare_alter_table(thd, schema_table->table, - &local_create_info, &alter_info)) + if (mysql_prepare_alter_table(thd, schema_table, &local_create_info, + &alter_info)) DBUG_RETURN(1); - if (mysql_prepare_create_table(thd, db, &local_create_info, &alter_info, + if (mysql_prepare_create_table(thd, &local_create_info, &alter_info, tmp_table, &db_options, schema_table->table->file, &schema_table->table->s->key_info, &keys, 0)) @@ -4799,6 +5076,7 @@ bool mysql_create_like_schema_frm(THD* t &local_create_info, alter_info.create_list, keys, schema_table->table->s->key_info, alter_info.foreign_key_list, + alter_info.parent_foreign_key_list, schema_table->table->file)) DBUG_RETURN(1); DBUG_RETURN(0); @@ -4897,7 +5175,7 @@ bool mysql_create_like_table(THD* thd, T pthread_mutex_lock(&LOCK_open); if (src_table->schema_table) { - if (mysql_create_like_schema_frm(thd, src_table, db, dst_path, + if (mysql_create_like_schema_frm(thd, src_table, dst_path, create_info)) { pthread_mutex_unlock(&LOCK_open); @@ -5188,9 +5466,6 @@ void setup_ha_alter_flags(Alter_info *al /** @param thd Thread @param table The original table. - @param db Database for new table. - QQ: Do we really need this argument or we - can use some dummy value ? @param alter_info Alter options, fields and keys for the new table. @param create_info Create options for the new table. @@ -5223,7 +5498,6 @@ static bool compare_tables(THD *thd, TABLE *table, - const char *db, Alter_info *alter_info, HA_CREATE_INFO *create_info, uint order_num, @@ -5264,7 +5538,7 @@ compare_tables(THD *thd, THD *thd= table->in_use; uint db_options= 0; /* not used */ /* Create the prepared information. */ - if (mysql_prepare_create_table(thd, db, create_info, + if (mysql_prepare_create_table(thd, create_info, &tmp_alter_info, (table->s->tmp_table != NO_TMP_TABLE), &db_options, @@ -5932,7 +6206,8 @@ mysql_fast_or_online_alter_table(THD *th @param[in,out] thd thread handle. Used as a memory pool and source of environment information. - @param[in] table the source table, open and locked + @param[in] table_list table_list element for the source table, + which should be open and locked. Used as an interface to the storage engine to acquire additional information about the original table. @@ -5957,16 +6232,18 @@ mysql_fast_or_online_alter_table(THD *th */ static bool -mysql_prepare_alter_table(THD *thd, TABLE *table, +mysql_prepare_alter_table(THD *thd, TABLE_LIST *table_list, HA_CREATE_INFO *create_info, Alter_info *alter_info) { + TABLE *table= table_list->table; /* New column definitions are added here */ List new_create_list; /* New key definitions are added here */ List new_key_list; /* FKs definitions for new version of table are added here. */ List new_foreign_key_list; + List new_parent_foreign_key_list; List_iterator drop_it(alter_info->drop_list); List_iterator def_it(alter_info->create_list); List_iterator alter_it(alter_info->alter_list); @@ -6166,8 +6443,24 @@ mysql_prepare_alter_table(THD *thd, TABL is implemented we simply copy foreign keys from old definition of table without adding or dropping anything. */ - new_foreign_key_list= table->s->fkeys; - list_copy_and_replace_each_value(new_foreign_key_list, thd->mem_root); + List_iterator fk_c_it(table->s->fkeys_child); + List_iterator fk_p_it(table->s->fkeys_parent); + Foreign_key_child_share *fk_c_s; + Foreign_key_parent_share *fk_p_s; + Foreign_key *fk; + Foreign_key_parent *fk_p; + while ((fk_c_s= fk_c_it++)) + { + if (!(fk= fk_c_s->make_foreign_key(thd->mem_root, table_list)) || + new_foreign_key_list.push_back(fk, thd->mem_root)) + goto err; + } + while ((fk_p_s= fk_p_it++)) + { + if (!(fk_p= fk_p_s->make_foreign_key(thd->mem_root)) || + new_parent_foreign_key_list.push_back(fk_p, thd->mem_root)) + goto err; + } } else { @@ -6177,6 +6470,7 @@ mysql_prepare_alter_table(THD *thd, TABL create supporting indexes. */ new_foreign_key_list.swap(alter_info->foreign_key_list); + new_parent_foreign_key_list.swap(alter_info->parent_foreign_key_list); } /* @@ -6352,6 +6646,7 @@ mysql_prepare_alter_table(THD *thd, TABL alter_info->create_list.swap(new_create_list); alter_info->key_list.swap(new_key_list); alter_info->foreign_key_list.swap(new_foreign_key_list); + alter_info->parent_foreign_key_list.swap(new_parent_foreign_key_list); err: DBUG_RETURN(rc); } @@ -6825,7 +7120,7 @@ view_err: */ new_db_type= create_info->db_type; - if (mysql_prepare_alter_table(thd, table, create_info, alter_info)) + if (mysql_prepare_alter_table(thd, table_list, create_info, alter_info)) goto err; set_table_default_charset(thd, create_info, db); @@ -6864,7 +7159,7 @@ view_err: uint table_changes= IS_EQUAL_YES; bool need_copy_table= TRUE; /* Check how much the tables differ. */ - if (compare_tables(thd, table, new_db, alter_info, + if (compare_tables(thd, table, alter_info, create_info, order_num, &ha_alter_flags, &ha_alter_info, === modified file 'sql/sql_trigger.cc' --- a/sql/sql_trigger.cc 2008-06-11 11:49:58 +0000 +++ b/sql/sql_trigger.cc 2008-08-03 10:13:01 +0000 @@ -651,7 +651,7 @@ bool Table_triggers_list::create_trigger NOTE: now we do not check privileges at CREATE TRIGGER time. This will be changed in the future. */ - trg_field->setup_field(thd, table, NULL); + trg_field->setup_field(thd, table, NULL, NULL, NULL); if (!trg_field->fixed && trg_field->fix_fields(thd, (Item **)0)) @@ -986,9 +986,15 @@ Table_triggers_list::~Table_triggers_lis /** - Prepare array of Field objects referencing to TABLE::record[1] instead - of record[0] (they will represent OLD.* row values in ON UPDATE trigger - and in ON DELETE trigger which will be called during REPLACE execution). + Prepare structures needed for executing triggers with references to + OLD.* and NEW.* pseudo-variables. + + @note Prepares array of Field objects referencing to TABLE::record[1] + instead of record[0] (they will represent OLD.* row values in + ON UPDATE trigger and in ON DELETE trigger which will be called + during REPLACE execution). + @note Allocates memory and provides initial values for sets of table + columns to be read and updated by triggers via pseudo-variables. @param table pointer to TABLE object for which we are creating fields. @@ -997,9 +1003,11 @@ Table_triggers_list::~Table_triggers_lis @retval True error */ -bool Table_triggers_list::prepare_record1_accessors(TABLE *table) +bool Table_triggers_list::prepare_for_trigger_fields(TABLE *table) { + uchar *bitmap; Field **fld, **old_fld; + int i; if (!(record1_field= (Field **)alloc_root(&table->mem_root, (table->s->fields + 1) * @@ -1020,24 +1028,28 @@ bool Table_triggers_list::prepare_record } *old_fld= 0; - return 0; -} - - -/** - Adjust Table_triggers_list with new TABLE pointer. - - @param new_table new pointer to TABLE instance -*/ + /* + Allocate memory and provide initial state for sets of subject + table columns to be read and updated by triggers via OLD/NEW.* + pseudo-variables. Initialization of these sets will be finished + in Item_trigger_field::setup_field(). + */ + if (!(bitmap= (uchar*) alloc_root(&table->mem_root, + (table->s->column_bitmap_size * 2 * + TRG_EVENT_MAX)))) + return 1; -void Table_triggers_list::set_table(TABLE *new_table) -{ - trigger_table= new_table; - for (Field **field= new_table->triggers->record1_field ; *field ; field++) + for (i= 0; i < (int)TRG_EVENT_MAX; i++) { - (*field)->table= (*field)->orig_table= new_table; - (*field)->table_name= &new_table->alias; + bitmap_init(&read_set[i], (my_bitmap_map*)bitmap, table->s->fields, FALSE); + bitmap_clear_all(&read_set[i]); + bitmap+= table->s->column_bitmap_size; + bitmap_init(&write_set[i], (my_bitmap_map*)bitmap, table->s->fields, FALSE); + bitmap_clear_all(&write_set[i]); + bitmap+= table->s->column_bitmap_size; } + + return 0; } @@ -1264,7 +1276,7 @@ bool Table_triggers_list::check_n_load(T TODO: This could be avoided if there is no triggers for UPDATE and DELETE. */ - if (!names_only && triggers->prepare_record1_accessors(table)) + if (!names_only && triggers->prepare_for_trigger_fields(table)) DBUG_RETURN(1); List_iterator_fast itm(triggers->definition_modes_list); @@ -1408,7 +1420,9 @@ bool Table_triggers_list::check_n_load(T { trg_field->setup_field(thd, table, &triggers->subject_table_grants[lex.trg_chistics.event] - [lex.trg_chistics.action_time]); + [lex.trg_chistics.action_time], + &triggers->read_set[lex.trg_chistics.event], + &triggers->write_set[lex.trg_chistics.event]); } lex_end(&lex); @@ -1989,23 +2003,8 @@ bool Table_triggers_list::process_trigge void Table_triggers_list::mark_fields_used(trg_event_type event) { - int action_time; - Item_trigger_field *trg_field; - - for (action_time= 0; action_time < (int)TRG_ACTION_MAX; action_time++) - { - for (trg_field= trigger_fields[event][action_time]; trg_field; - trg_field= trg_field->next_trg_field) - { - /* We cannot mark fields which does not present in table. */ - if (trg_field->field_idx != (uint)-1) - { - bitmap_set_bit(trigger_table->read_set, trg_field->field_idx); - if (trg_field->get_settable_routine_parameter()) - bitmap_set_bit(trigger_table->write_set, trg_field->field_idx); - } - } - } + bitmap_union(trigger_table->read_set, &read_set[event]); + bitmap_union(trigger_table->write_set, &write_set[event]); trigger_table->file->column_bitmaps_signal(); } === modified file 'sql/sql_trigger.h' --- a/sql/sql_trigger.h 2007-10-16 20:11:50 +0000 +++ b/sql/sql_trigger.h 2008-08-03 10:13:01 +0000 @@ -38,10 +38,16 @@ class Table_triggers_list: public Sql_al /** During execution of trigger new_field and old_field should point to the array of fields representing new or old version of row correspondingly - (so it can point to TABLE::field or to Tale_triggers_list::record1_field) + (so it can point to TABLE::field or to Table_triggers_list::record1_field) */ Field **new_field; Field **old_field; + /** + Sets of subject table columns read and updated by triggers for + respective event types. + */ + MY_BITMAP read_set[TRG_EVENT_MAX]; + MY_BITMAP write_set[TRG_EVENT_MAX]; /* TABLE instance for which this triggers list object was created */ TABLE *trigger_table; @@ -136,16 +142,19 @@ public: bodies[TRG_EVENT_DELETE][TRG_ACTION_AFTER]); } - void set_table(TABLE *new_table); - void mark_fields_used(trg_event_type event); + const MY_BITMAP *get_update_triggers_write_set() + { + return &write_set[TRG_EVENT_UPDATE]; + } + friend class Item_trigger_field; friend int sp_cache_routines_and_add_tables_for_triggers(THD *thd, LEX *lex, TABLE_LIST *table); private: - bool prepare_record1_accessors(TABLE *table); + bool prepare_for_trigger_fields(TABLE *table); LEX_STRING* change_table_name_in_trignames(const char *db_name, LEX_STRING *new_table_name, LEX_STRING *stopper); === modified file 'sql/sql_yacc.yy' --- a/sql/sql_yacc.yy 2008-06-27 05:11:56 +0000 +++ b/sql/sql_yacc.yy 2008-08-03 10:13:01 +0000 @@ -580,7 +580,7 @@ bool setup_select_in_parentheses(LEX *le enum index_hint_type index_hint; enum enum_filetype filetype; enum ha_build_method build_method; - enum Foreign_key::fk_option m_fk_option; + enum fk_option m_fk_option; } %{ @@ -1825,6 +1825,7 @@ create: lex->create_info.default_table_charset= NULL; lex->name.str= 0; lex->name.length= 0; + lex->create_foreign_key_tables= 0; } create2 { @@ -4716,11 +4717,10 @@ create_field_list: field_list { /* - Until one of the later milestones of WL#148 "Foreign key" is - implemented we should parent tables for foreign key constraints - from the table lists. + Move parent tables for foreign key constraints to the separate + table list. */ - Lex->unlink_all_tables_except_first(); + Lex->unlink_all_tables_except_first(&Lex->create_foreign_key_tables); } ; @@ -4757,16 +4757,20 @@ column_ref_constraint_def: lex->col_list.push_back(new (thd->mem_root) Key_part_spec(lex->ident, 0)); if (!(table= lex->select_lex.add_table_to_list(YYTHD, $2, NULL, - TL_OPTION_UPDATING, - TL_IGNORE))) + (TL_OPTION_UPDATING | + TL_OPTION_DONT_ADD_DUPS), + TL_IGNORE))) YYABORT; key= new (thd->mem_root) Foreign_key($1, null_lex_str, FALSE, + lex->query_tables, lex->col_list, table, lex->ref_list, lex->fk_delete_opt, lex->fk_update_opt, - lex->fk_match_option); + lex->fk_match_option, + FALSE, FALSE); + key->set_missing_options(); lex->alter_info.foreign_key_list.push_back(key); /* Only used for ALTER TABLE. Ignored otherwise. */ lex->alter_info.flags|= ALTER_FOREIGN_KEY; @@ -4804,16 +4808,21 @@ key_def: Foreign_key *key; TABLE_LIST *table; if (!(table= lex->select_lex.add_table_to_list(YYTHD, $8, NULL, - TL_OPTION_UPDATING, - TL_IGNORE))) + (TL_OPTION_UPDATING | + TL_OPTION_DONT_ADD_DUPS), + TL_IGNORE))) YYABORT; key= new (thd->mem_root) Foreign_key($1.str ? $1 : $4, $4, - TRUE, lex->col_list, + TRUE, + lex->query_tables, + lex->col_list, table, lex->ref_list, lex->fk_delete_opt, lex->fk_update_opt, - lex->fk_match_option); + lex->fk_match_option, + FALSE, FALSE); + key->set_missing_options(); lex->alter_info.foreign_key_list.push_back(key); lex->col_list.empty(); /* Alloced by sql_alloc */ /* Only used for ALTER TABLE. Ignored otherwise. */ @@ -5439,32 +5448,32 @@ ref_list: opt_match_clause: /* empty */ - { Lex->fk_match_option= Foreign_key::FK_MATCH_UNDEF; } + { Lex->fk_match_option= FK_MATCH_UNDEF; } | MATCH FULL - { Lex->fk_match_option= Foreign_key::FK_MATCH_FULL; } + { Lex->fk_match_option= FK_MATCH_FULL; } | MATCH PARTIAL - { Lex->fk_match_option= Foreign_key::FK_MATCH_PARTIAL; } + { Lex->fk_match_option= FK_MATCH_PARTIAL; } | MATCH SIMPLE_SYM - { Lex->fk_match_option= Foreign_key::FK_MATCH_SIMPLE; } + { Lex->fk_match_option= FK_MATCH_SIMPLE; } ; opt_on_update_delete: /* empty */ { LEX *lex= Lex; - lex->fk_update_opt= Foreign_key::FK_OPTION_UNDEF; - lex->fk_delete_opt= Foreign_key::FK_OPTION_UNDEF; + lex->fk_update_opt= FK_OPTION_UNDEF; + lex->fk_delete_opt= FK_OPTION_UNDEF; } | ON UPDATE_SYM delete_option { LEX *lex= Lex; lex->fk_update_opt= $3; - lex->fk_delete_opt= Foreign_key::FK_OPTION_UNDEF; + lex->fk_delete_opt= FK_OPTION_UNDEF; } | ON DELETE_SYM delete_option { LEX *lex= Lex; - lex->fk_update_opt= Foreign_key::FK_OPTION_UNDEF; + lex->fk_update_opt= FK_OPTION_UNDEF; lex->fk_delete_opt= $3; } | ON UPDATE_SYM delete_option @@ -5484,11 +5493,11 @@ opt_on_update_delete: ; delete_option: - RESTRICT { $$= Foreign_key::FK_OPTION_RESTRICT; } - | CASCADE { $$= Foreign_key::FK_OPTION_CASCADE; } - | SET NULL_SYM { $$= Foreign_key::FK_OPTION_SET_NULL; } - | NO_SYM ACTION { $$= Foreign_key::FK_OPTION_NO_ACTION; } - | SET DEFAULT { $$= Foreign_key::FK_OPTION_DEFAULT; } + RESTRICT { $$= FK_OPTION_RESTRICT; } + | CASCADE { $$= FK_OPTION_CASCADE; } + | SET NULL_SYM { $$= FK_OPTION_SET_NULL; } + | NO_SYM ACTION { $$= FK_OPTION_NO_ACTION; } + | SET DEFAULT { $$= FK_OPTION_DEFAULT; } ; key_type: === modified file 'sql/table.cc' --- a/sql/table.cc 2008-06-23 13:26:17 +0000 +++ b/sql/table.cc 2008-08-03 10:13:01 +0000 @@ -347,7 +347,8 @@ TABLE_SHARE *alloc_table_share(TABLE_LIS memcpy((char*) &share->mem_root, (char*) &mem_root, sizeof(mem_root)); pthread_mutex_init(&share->LOCK_ha_data, MY_MUTEX_INIT_FAST); - share->fkeys.empty(); + share->fkeys_child.empty(); + share->fkeys_parent.empty(); } DBUG_RETURN(share); } @@ -414,7 +415,8 @@ void init_tmp_table_share(THD *thd, TABL share->used_tables.empty(); share->free_tables.empty(); - share->fkeys.empty(); + share->fkeys_child.empty(); + share->fkeys_parent.empty(); DBUG_VOID_RETURN; } @@ -1662,6 +1664,9 @@ static int open_binary_frm(THD *thd, TAB goto err; bitmap_init(&share->all_set, bitmaps, share->fields, FALSE); bitmap_set_all(&share->all_set); + + if (fk_init_column_sets(share)) + goto err; delete handler_file; #ifndef DBUG_OFF === modified file 'sql/table.h' --- a/sql/table.h 2008-06-23 13:26:17 +0000 +++ b/sql/table.h 2008-08-03 10:13:01 +0000 @@ -26,7 +26,8 @@ class st_select_lex; class partition_info; class COND_EQUAL; class Security_context; -class Foreign_key; +class Foreign_key_child_share; +class Foreign_key_parent_share; struct MDL_LOCK_DATA; /*************************************************************************/ @@ -383,8 +384,10 @@ typedef struct st_table_share /** place to store storage engine specific data */ void *ha_data; - /** List of foreign keys for this table. */ - List fkeys; + /** List of foreign keys in which this table participates as child. */ + List fkeys_child; + /** List of foreign keys in which this table participates as parent. */ + List fkeys_parent; /* @@ -1326,6 +1329,14 @@ struct TABLE_LIST the parsed tree is created. */ uint8 trg_event_map; + + /** + Pointer to the list of columns which are going to be updated if + this table list element represents table to be updated and such + list is known. 0 - otherwise (i.e. list of columns is not known + or table list element does not correspond to table to be updated). + */ + List *cols_to_be_updated; uint i_s_requested_object; bool has_db_lookup_value; === modified file 'sql/unireg.cc' --- a/sql/unireg.cc 2008-06-23 07:14:00 +0000 +++ b/sql/unireg.cc 2008-08-03 10:13:01 +0000 @@ -90,7 +90,8 @@ handle_error(uint sql_errno, create_fields Fields to create keys number of keys to create key_info Keys to create - fkey_list List of foreign keys to be created + fkey_list List of foreign keys in which this table is child + parent_fkey_list List of foreign keys in which this table is parent db_file Handler to use. May be zero, in which case we use create_info->db_type RETURN @@ -104,6 +105,7 @@ bool mysql_create_frm(THD *thd, const ch List &create_fields, uint keys, KEY *key_info, List &fkey_list, + List &parent_fkey_list, handler *db_file) { LEX_STRING str_db_type; @@ -246,7 +248,7 @@ bool mysql_create_frm(THD *thd, const ch if (opt_fk_all_engines) { - fk_sec_length= fk_get_frm_section_length(fkey_list); + fk_sec_length= fk_get_frm_section_length(fkey_list, parent_fkey_list); create_info->extra_size+= fk_sec_length; } @@ -384,7 +386,8 @@ bool mysql_create_frm(THD *thd, const ch if (opt_fk_all_engines) { - if (fk_save_to_frm_section(thd, file, fkey_list, fk_sec_length)) + if (fk_save_to_frm_section(thd, file, fkey_list, parent_fkey_list, + fk_sec_length)) goto err; } @@ -468,7 +471,8 @@ err3: create_info create info parameters create_fields Fields to create keys number of keys to create - fkey_list List of foreign keys to create + fkey_list List of foreign keys in which this table is child + parent_fkey_list List of foreign keys in which this table is parent key_info Keys to create file Handler to use @@ -483,6 +487,7 @@ int rea_create_table(THD *thd, const cha List &create_fields, uint keys, KEY *key_info, List &fkey_list, + List &parent_fkey_list, handler *file) { DBUG_ENTER("rea_create_table"); @@ -490,7 +495,8 @@ int rea_create_table(THD *thd, const cha char frm_name[FN_REFLEN]; strxmov(frm_name, path, reg_ext, NullS); if (mysql_create_frm(thd, frm_name, db, table_name, create_info, - create_fields, keys, key_info, fkey_list, file)) + create_fields, keys, key_info, fkey_list, + parent_fkey_list, file)) DBUG_RETURN(1);