Background 1:

When binlog_format = row, CREATE ... SELECT is logged in two pieces,
like:

Anonymous_Gtid
query_log_event(CREATE TABLE without SELECT)
Anonymous_Gtid
query_log_event(BEGIN)
...row events...
query_log_event(COMMIT) (or Xid_log_event)

Internally, there is a call to MYSQL_BIN_LOG::commit after the table has
been created and before the rows are selected.

When gtid_next='ANONYMOUS', we must not release anonymous ownership for
the commit occurring in the middle of the statement (since that would
allow a concurrent client to set gtid_mode=on, making it impossible to
commit the rest of the statement). Also, the commit in the middle of the
statement should not decrease the counter of ongoing GTID-violating
transactions, since that would allow a concurrent client to set
ENFORCE_GTID_CONSISTENCY=ON even if there is an ongoing transaction that
violates GTID-consistency.

The logic to skip releasing anonymous ownership and skip decreasing the
counter is as follows. Before calling mysql_bin_log.commit, it sets the
flag thd->is_commit_in_middle_of_statement. Eventually,
mysql_bin_log.commit calls gtid_state->update_on_commit, which calls
gtid_state->update_gtids_impl, which reads the
thd->is_commit_in_middle_of_statement and accordingly decides to skip
releasing anonymous ownership and/or skips decreasing the counter.

Problem 1:

When thd->is_commit_in_middle_of_statement has been set, it is crucial
that there is another call to update_gtids_impl when the transaction
ends (otherwise the session will keep holding anonymous ownership and
will not decrease the counters). Normally, this happens because
mysql_bin_log.commit is always called, and mysql_bin_log.commit normally
invokes ordered_commit, which calls update_gtids_impl. However, in case
the SELECT part of the statement does not find any rows,
mysql_bin_log.commit skips the call to ordered_commit, so
update_gtids_impl does not get called. This is the first problem we fix
in this commit.

Fix 1:

We fix this problem as follows. After calling mysql_bin_log.commit to
log the CREATE part of CREATE...SELECT, the CREATE...SELECT code sets
thd->pending_gtid_state_update=true (this is a new flag that we
introduce in this patch). If the flag is set, update_gtids_impl clears
it. At the end of mysql_bin_log.commit, we check the flag to see if
update_gtids_impl has been called by any function invoked by
mysql_bin_log.commit. If not, i.e., if the flag is still true at the end
of mysql_bin_log.commit, it means we have reached the corner case where
update_gtids_impl was skipped. Thus we call it explicitly from
mysql_bin_log.commit.

Background 2:

GTID-violating DDL (CREATE...SELECT and CREATE TEMPORARY) is detected in
is_ddl_gtid_compatible, called from gtid_pre_statement_checks, which is
called from mysql_parse just before the implicit pre-commit.
is_ddl_gtid_compatible determines whether an error or warning or nothing
is to be generated, and whether to increment the counters of GTID-
violating transactions. In case an error is generated, it is important
that the error happens before the implicit commit, so that the statement
fails before it commits the ongoing transaction.

Problem 2:

In case a warning is to be generated, and there is an ongoing
transaction, the implicit commit will write to the binlog, and thus it
will call gtid_state->update_gtids_impl, which will decrease the
counters of GTID-violating transactions. Thus, the counters will be zero
for the duration of the transaction.

Fix 2:

We call is_ddl_gtid_compatible *both* before the implicit commit and
after the implicit commit. If an error is to be generated, the error is
generated before the commit. If a warning is to be generated and/or the
counter of GTID-violating transactions is to be increased, then this
happens after the commit.

Code changes #1:

@sql/binlog.cc
- Move MYSQL_BIN_LOG::commit to a new function
MYSQL_BIN_LOG::write_binlog_and_commit_engine. Make
MYSQL_BIN_LOG::commit call this function, and after the return check
thd->pending_gtid_state_update to see if another call to
gtid_state->update_on_[commit|rollback] is needed.
- Simplify MYSQL_BIN_LOG::write_bin_log_and_commit_engine; remove
useless local variable 'error' that would never change its value.

@sql/binlog.h:
- Declaration of new function.

@sql/rpl_gtid_state.cc:
- Set thd->pending_gtid_state_update to false at the end of
update_gtids_impl.

Code changes #2:
@sql/binlog.cc:
- Add two parameters to handle_gtid_consistency and is_ddl_compatible:
handle_error is true in the call to is_ddl_gtid_compatible that happens
*before* the implicit commit and fals in the call to
is_ddl_gtid_compatible that happens *after* the implicit commit. It
tells the function to generate the error, if an error is to be
generated. The other parameter, handle_nonerror, is true in the call to
is_ddl_gtid_compatible that happens *after* the implicit commit and
false in the call that happens *before* the implicit commit. It tells
the function to generate the warnings and increment the counter, if that
needs to be done.

@sql/rpl_gtid_execution.cc:
- Call is_ddl_gtid_compatible after the implicit commit. Pass the two
new parameters to the function.

@sql/sql_class.h:
- Update prototype for is_ddl_gtid_compatible.

@sql/sql_insert.cc:
- Set thd->pending_gtid_state_update = true after committing the CREATE
part of a CREATE...SELECT.

Misc changes:

@sql/binlog.cc
- Add DEBUG_SYNC symbol end_decide_logging_format used in tests.

@sql/rpl_gtid_state.cc:
- For modularity, move out parts of update_gtids_impl to a new function,
end_gtid_violating_transaction.
- Move the lock/unlock of global_sid_lock into update_gtids_impl.
- Make update_gtids_impl release global_sid_lock before the call to
end_gtid_violating_transaction, so as to hold it as short as possible.

@sql/rpl_gtid.h
- Because we release the locks earlier in update_gtids_impl in
rpl_gtid_state.cc, we need to acquire the lock again in
end_[anonymous|automatic]_gtid_violating_transaction, in order to do
some debug assertions.
- Add DBUG_PRINT for the counters.

Test changes:
- Split binlog_enforce_gtid_consistency into six tests, depending on the
type of scenarios it tests:
Three classes of GTID-violation:
    *_create_select_*: test CREATE ... SELECT.
    *_tmp_*: test CREATE TEMPORARY/DROP TEMPORARY inside a transaction.
    *_trx_nontrx_*: test combinations of transactional and
    non-transactional updates in the same statement or in the same
    transaction.
For each class of GTID-violation, one positive and one negative test:
    *_consistent.test: Cases which are *not* GTID-violating
    *_violation.test: Cases which *are* GTID-violating.
- The general logic of these test is:
- extra/binlog_tests/enforce_gtid_consistency.test iterates over all
    values of GTID_MODE, ENFORCE_GTID_CONSISTENCY, and GTID_NEXT. For
    each case, it sources file; the name of the sourced file is
    specified by the file that sources
    extra/binlog_tests/enforce_gtid_consistency.test
- The top-level file in suite/binlog/t invokes
    extra/binlog_tests/enforce_gtid_consistency.test, specifying one
    of the filenames
    extra/binlog_tests/enforce_gtid_consistency_[create_select|tmp|
    trx_nontrx]_[consistent|violation].test.
- Each of the files
    extra/binlog_tests/enforce_gtid_consistency_[create_select|tmp|
    trx_nontrx]_[consistent|violation].test
    sets up a number of test scenarios. Each test scenario is executed
    by sourcing
    extra/binlog_tests/enforce_gtid_consistency_statement.inc.
- extra/binlog_tests/enforce_gtid_consistency_statement.inc executes
    the specified statement, checks that warnings are generated and
    counters incremented/decremented as specified by the caller.
- Since the tests set GTID_MODE explicitly, it doesn't make sense to run
the test in both combinations GTID_MODE=ON/OFF. However, for the
*_trx_nontrx_* cases, it is important to test that it works both with
binlog_direct_non_transactional_updates=on and off. The suite is never
run with those combinations. To leverage from the combinations
GTID_MODE=ON/OFF, we run the test with
binlog_direct_non_transactional_updates=on if GTID_MODE=ON, and we run
the test with binlog_direct_non_transactional_updates=off if
GTID_MODE=OFF.